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.

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	Skip steps at index < resume_from (for resume-after-partial). Steps are 0-indexed.	None
on_step	StepHook	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	Skip steps at index < resume_from.	None
on_step	AsyncStepHook	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.

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).