MODULE REFERENCE

graph

anomx.execution.work.graph

Portable work graphs; the host decides how and where each step executes.

WorkGraph

WorkGraph(steps: list[WorkStep], *, output_node: str | None = None)

A validated, deterministic DAG with explicit predecessor outputs.

The executor receives a step and its inputs. previous is the sole direct predecessor's result (or a mapping for a join), results contains every completed step keyed by ID, and original inputs remain available. Execution is sequential in topological order; dispatch and isolation belong to the host.

Signature parameters
Parameter	Type	Default
`steps`	`list[WorkStep]`	`required`
`output_node`keyword only	`str \| None`	`None`

Members

executemethod

# WorkGraph.execute Source ↗

execute(executor: Callable[[WorkStep, dict[str, Any]], Any], *, inputs: Mapping[str, Any] | None = None) -> object

Execute every step once; propagate errors immediately to the host.

Signature parameters
Parameter	Type	Default
`executor`	`Callable[[WorkStep, dict[str, Any]], Any]`	`required`
`inputs`keyword only	`Mapping[str, Any] \| None`	`None`

Returns object

from_dictclassmethod

# WorkGraph.from_dict Source ↗

from_dict(graph: Mapping[str, Any]) -> WorkGraph

Read the editor's start/code/inference/end representation.

Legacy compute, log, notification and detection nodes remain readable. Start and end are structural nodes and cannot contain executable code.

Signature parameters
Parameter	Type	Default
`graph`	`Mapping[str, Any]`	`required`

Returns WorkGraph

classView source ↗

WorkStep

WorkStep(id: str, kind: Literal['python', 'inference'], code: str, runtime: str = 'task', dependencies: tuple[str, ...] = (), label: str = '', metadata: Mapping[str, Any] = <factory: dict>)

One Python or inference step with an explicit runtime selection.

task executes on the orchestrator, default uses the work's compute target, and other strings identify host-defined resources. The host must authorize those resources. Inference remains a distinct operation so a host can later reuse model sessions without changing the graph contract.

Signature parameters
Parameter	Type	Default
`id`	`str`	`required`
`kind`	`Literal['python', 'inference']`	`required`
`code`	`str`	`required`
`runtime`	`str`	`'task'`
`dependencies`	`tuple[str, ...]`	`()`
`label`	`str`	`''`
`metadata`	`Mapping[str, Any]`	`<factory: dict>`

Members

idattribute

# WorkStep.id Source ↗

id: str

kindattribute

# WorkStep.kind Source ↗

kind: Literal['python', 'inference']

codeattribute

# WorkStep.code Source ↗

code: str

runtimeattribute

# WorkStep.runtime Source ↗

runtime: str = 'task'

dependenciesattribute

# WorkStep.dependencies Source ↗

dependencies: tuple[str, ...] = ()

labelattribute

# WorkStep.label Source ↗

label: str = ''

metadataattribute

# WorkStep.metadata Source ↗

metadata: Mapping[str, Any] = field(default_factory=dict)