Engine

dagpipe.engine

Summary

Execution engine responsible for running pipelines and graphs.

The Engine executes Node objects and propagates immutable State instances through either a linear sequence or a directed acyclic graph (Graph). It orchestrates execution order, branching, and state propagation.

Guarantees

Deterministic execution and consistent state lineage.
Orchestrates execution without modifying Graph, Node, or State objects.

Classes

Engine

Engine(
    nodes_or_graph: Sequence[Node] | Graph,
    *,
    on_step: StepHook | None = None
)

Execution engine responsible for running pipeline logic.

Notes

Responsibilities:

- Accepts either a linear sequence of `Node` objects or a `Graph`
  defining execution topology.
- Propagates immutable `State` objects through `Node` objects and
  collects terminal states.
- Supports synchronous (`run`) and asynchronous (`run_async`)
  execution, dispatching per-node.
- Supports step-wise / resumable execution and progress hooks.

Guarantees:

- Never mutates `State`, `Node`, or `Graph` instances.
- `State` objects are never modified in place; each branch produces
  independent instances.
- Execution order is deterministic and follows graph or pipeline topology.
- Thread-safe for concurrent execution.

Create an engine from a node sequence or a graph.

Parameters:

Name	Type	Description	Default
`nodes_or_graph`	`Sequence[Node] \| Graph`	Either an ordered sequence of `Node` instances (linear mode) or a `Graph` defining the execution topology (graph mode).	required
`on_step`	`StepHook \| None`	Default per-step callback `(step, status, message)` used when a step run does not supply its own hook.	`None`

Raises:

Type	Description
`TypeError`	If a sequence element is not a `Node`, or if `nodes_or_graph` is neither a `Sequence[Node]` nor a `Graph`.

Attributes

nodes `property`

nodes: tuple[Node, ...]

Return nodes managed by this engine.

Returns:

Type	Description
`tuple[Node, ...]`	tuple[Node, ...]: Ordered sequence in linear mode or all nodes in graph mode.

Functions

repr

__repr__() -> str

Return the canonical string representation of the object.

Returns:

Name	Type	Description
`str`	`str`	Representation that uniquely identifies the object and its configuration.

run

run(root: State) -> list[State]

Execute the pipeline starting from a root State.

Parameters:

Name	Type	Description	Default
`root`	`State`	Initial execution state.	required

Returns:

Type	Description
`list[State]`	list[State]: Terminal execution states produced by the pipeline.

Raises:

Type	Description
`TypeError`	If `root` is not a `State` instance.
`RuntimeError`	If the engine execution mode is invalid.

Notes

Responsibilities:

- Selects execution mode, propagates state through nodes, creates
  new instances for branches, and collects terminal states.

run_async `async`

run_async(root: State) -> list[State]

Execute the pipeline starting from root, dispatching sync vs async nodes.

Parameters:

Name	Type	Description	Default
`root`	`State`	Initial execution state.	required

Returns:

Type	Description
`list[State]`	list[State]: Terminal execution states produced by the pipeline.

Notes

Each node is executed with Node.run when synchronous and AsyncNode.run_async when asynchronous. Linear and graph topologies are both supported.

run_steps

run_steps(
    root: State,
    *,
    resume_from: int | None = None,
    on_step: StepHook | None = None
) -> Iterator[StepResult]

Execute the pipeline step-by-step, yielding one StepResult per step.

Parameters:

Name	Type	Description	Default
`root`	`State`	Initial execution state.	required
`resume_from`	`int \| None`	Skip steps at index < `resume_from` (for resume-after-partial). Steps are 0-indexed.	`None`
`on_step`	`StepHook \| None`	Callback `(step, status, message)` invoked per step; falls back to the engine-level hook when unset.	`None`

Yields:

Name	Type	Description
`StepResult`	`StepResult`	One per executed node/step, carrying the produced states.

Notes

This is a synchronous, generator-based checkpoint interface compatible with the imperative resume-by-step behaviour of the legacy orchestrator. Use run_steps_async for async nodes.

run_steps_async `async`

run_steps_async(
    root: State,
    *,
    resume_from: int | None = None,
    on_step: AsyncStepHook | None = None
) -> AsyncIterator[StepResult]

Async variant of run_steps supporting AsyncNode execution.

Parameters:

Name	Type	Description	Default
`root`	`State`	Initial execution state.	required
`resume_from`	`int \| None`	Skip steps at index < `resume_from`.	`None`
`on_step`	`AsyncStepHook \| None`	Async callback `(step, status, message)` invoked per step.	`None`

Yields:

Name	Type	Description
`StepResult`	`AsyncIterator[StepResult]`	One per executed node/step.

ProgressMessage

ProgressMessage(
    *,
    lines: int | None = None,
    blocks: int | None = None,
    count: int | None = None,
    unit: str | None = None,
    raw_ocr_line: str | None = None,
    error: str | None = None,
    step: str = "",
    status: str = ""
)

Lightweight progress payload emitted by engine step hooks.

Mirrors the imperative ProgressMessage used by the legacy orchestrator so callers can surface counts/lines/errors without coupling the engine to pydantic.

Attributes:

Name	Type	Description
`lines`	`int \| None`	Optional count of processed lines.
`blocks`	`int \| None`	Optional count of processed blocks.
`count`	`int \| None`	Optional generic item count.
`unit`	`str \| None`	Optional unit for the count (e.g., 'pages').
`raw_ocr_line`	`str \| None`	Optional raw OCR line payload.
`error`	`str \| None`	Optional error description.
`step`	`str`	Identifier of the step that emitted the message.
`status`	`str`	Status label associated with the step.

Notes

Guarantees:

- Immutable after construction (attributes are never reassigned).
- Independent of pydantic; safe to construct in the engine core.

Create a progress message.

Parameters:

Name	Type	Description	Default
`lines`	`int \| None`	Optional count of processed lines.	`None`
`blocks`	`int \| None`	Optional count of processed blocks.	`None`
`count`	`int \| None`	Optional generic item count.	`None`
`unit`	`str \| None`	Optional unit for the count (e.g., 'pages').	`None`
`raw_ocr_line`	`str \| None`	Optional raw OCR line payload.	`None`
`error`	`str \| None`	Optional error description.	`None`
`step`	`str`	Identifier of the step that emitted the message.	`''`
`status`	`str`	Status label associated with the step.	`''`

Functions

as_dict

as_dict() -> dict[str, Any]

Return the message as a plain dictionary.

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: All attribute values keyed by their attribute name.

StepResult

StepResult(
    index: int,
    node_id: str,
    states: tuple[State, ...],
    completed: bool,
)

A single checkpointed step within an async/resumable engine run.

Attributes:

Name	Type	Description
`index`	`int`	Ordinal index of the step.
`node_id`	`str`	Identifier of the node associated with this step.
`states`	`tuple[State, ...]`	States produced by running this step.
`completed`	`bool`	Whether this step succeeded (vs. paused/interrupted).

Initialise StepResult.

Parameters:

Name	Type	Description	Default
`index`	`int`	Ordinal index of the step.	required
`node_id`	`str`	Identifier of the node associated with this step.	required
`states`	`tuple[State, ...]`	States produced by running this step.	required
`completed`	`bool`	Whether this step succeeded (vs. paused/interrupted).	required

Engine

dagpipe.engine

Summary

Guarantees

Classes

Engine

Attributes

nodes property

Functions

__repr__

run

run_async async

run_steps

run_steps_async async

ProgressMessage

Functions

as_dict

StepResult

Functions

nodes `property`

repr

run_async `async`

run_steps_async `async`