dagpipe

dagpipe

Summary

Directed acyclic graph execution framework for deterministic state propagation.

dagpipe executes pipelines composed of nodes connected in a directed acyclic graph (DAG). Each node receives an immutable State and optionally produces derived states for downstream nodes.

Installation

Install using pip:

pip install dagpipe

---

Quick Start

from dagpipe import State, Payload, Schema, Graph, Engine
from dagpipe.node import Node

class HelloNode(Node):
    id = "hello"
    def resolve(self, state):
        yield self.fork(state, payload_update={"msg": "hello"})

# Build and run
graph = Graph()
graph.add_root(HelloNode())
engine = Engine(graph)

class MyState(State):
    schema = Schema({})

results = engine.run(MyState(payload=Payload({})))

---

Public API

This package re-exports the core pipeline components. Consumers should import from this namespace for standard usage.

Execution Core

• Engine: Responsible for orchestrating node execution and state propagation.
• Graph: Defines the execution topology and node relationships.
• Node: Base class for defining execution logic and transformations.

State & Data

• State: Represents an immutable execution snapshot at a point in time.
• Payload: Immutable hierarchical container for execution data.
• Schema: Defines and validates the allowed structure of payloads.
• SchemaError: Raised when data violates the declared schema.

Declarative Pipelines

• Pipeline: High-level wrapper for an engine, state type, and initial payload.
• load_pipeline: Factory function to create a pipeline from YAML.

---

Classes

AsyncNode

Bases: Node

Base class for nodes whose execution is asynchronous.

Subclasses implement resolve_async (an async generator yielding derived State objects). The engine dispatches to resolve_async when running an async traversal (see Engine.run_async).

Sync-only engines (and the base Node.run) treat an AsyncNode as a no-op consumer: calling run on an AsyncNode returns no states, signalling that an async engine is required.

Functions

__hash__

__hash__() -> int

Return stable hash based on node ID.

Returns

int

__repr__

__repr__() -> str

Return debug representation.

Returns

str

__str__

__str__() -> str

Return display representation.

Returns

str

clean_id_and_name classmethod

clean_id_and_name() -> None

Normalize and validate node ID and display name.

Raises:

Type	Description
TypeError	If ID is not a string.
ValueError	If ID format is invalid.

Notes

Guarantees:

- Generates ID from module and class name if missing.
- Validates ID format.
- Generates human-readable name if missing.

fork

fork(
    state: State,
    *,
    payload_update: Any = None,
    confidence_delta: float = 0.0,
    metadata_update: Any = None
) -> State

Create a child State attributed to this node.

Parameters:

Name	Type	Description	Default
state	State	Parent execution state.	required
payload_update	Any	Dot-path payload updates.	None
confidence_delta	float	Confidence adjustment.	0.0
metadata_update	Any	Metadata updates.	None

Returns:

Name	Type	Description
State	State	New child execution state.

Notes

Responsibilities:

- Convenience wrapper around `State.fork()` that automatically
  records this node's ID in state history.

is_async

is_async() -> bool

Return whether this node executes asynchronously.

node_id_to_name staticmethod

node_id_to_name(node_id: str) -> str

Convert a dotted snake_case node ID into a human-readable name.

Parameters:

Name	Type	Description	Default
node_id	str	Unique node identifier (e.g., 'entity.resolve.numeric_merchant').	required

Returns:

Name	Type	Description
str	str	Human-readable display name (e.g., 'Entity › Resolve › Numeric Merchant').

resolve_async async

resolve_async(state: State) -> Iterable[State]

Execute node logic asynchronously.

Parameters:

Name	Type	Description	Default
state	State	Input execution state.	required

Returns:

Type	Description
Iterable[State]	Iterable[State]: Derived execution state(s).

Notes

Subclasses implement this. Must not mutate the input state. Should use fork() to create child states.

run

run(state: State) -> tuple[State, ...]

Execute this node on a State.

Parameters:

Name	Type	Description	Default
state	State	Input execution state.	required

Returns:

Type	Description
tuple[State, ...]	tuple[State, ...]: Derived execution states.

Raises:

Type	Description
TypeError	If resolve() yields a non-State object.

run_async async

run_async(state: State) -> tuple[State, ...]

Execute this node asynchronously on a state, validating outputs.

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

Graph

Graph()

Directed Acyclic Graph defining execution topology of Node objects.

Notes

Responsibilities:

- Stores node connectivity and validates that the topology remains acyclic.
- Structure determines how `State` flows between nodes during execution.

Guarantees:

- Topology is acyclic. Node relationships remain consistent.
- Thread-safe for concurrent reads after construction.

Create an empty Graph.

Initializes node registry and edge mappings.

Functions

__repr__

__repr__() -> str

Return debug representation.

Returns

str

add_edge

add_edge(src: Node, dst: Node) -> None

Add a directed edge from src to dst.

Parameters:

Name	Type	Description	Default
src	Node	Source node.	required
dst	Node	Destination node.	required

Raises:

Type	Description
TypeError	If src or dst is not a Node.
ValueError	If the edge would create a cycle or if src and dst are common.

Notes

• Validates node types.
• Prevents cycles.
• Registers nodes if not present.
• Updates parent and child mappings.

add_root

add_root(node: Node) -> None

Add a root node with no parents.

Parameters:

Name	Type	Description	Default
node	Node	Node to add as a root.	required

Raises:

Type	Description
TypeError	If node is not a Node instance.

children

children(node: Node) -> tuple[Node, ...]

Return child nodes of a node.

Parameters:

Name	Type	Description	Default
node	Node	Node to query.	required

Returns:

Type	Description
tuple[Node, ...]	tuple[Node, ...]: Outgoing neighbors.

nodes

nodes() -> tuple[Node, ...]

Return all nodes in the graph.

Returns:

Type	Description
tuple[Node, ...]	tuple[Node, ...]: All registered nodes.

parents

parents(node: Node) -> tuple[Node, ...]

Return parent nodes of a node.

Parameters:

Name	Type	Description	Default
node	Node	Node to query.	required

Returns:

Type	Description
tuple[Node, ...]	tuple[Node, ...]: Incoming neighbors.

roots

roots() -> tuple[Node, ...]

Return root nodes (nodes with no incoming edges).

Returns:

Type	Description
tuple[Node, ...]	tuple[Node, ...]: Entry point nodes.

Node

Bases: ABC

Base class for all dagpipe execution nodes.

Attributes:

Name	Type	Description
id	str	Unique identifier of the node (snake_case dotted format).
name	str	Human-readable display name.

Notes

Responsibilities:

- Represents a deterministic unit of execution in the pipeline graph.
- Consumes one `State` and produces zero, one, or many derived states.
- Defines execution logic and enables branching, filtering, and transformation.

Guarantees:

- Nodes must never mutate the input `State`.
- Instances are singletons per subclass and reused across executions.

Functions

__hash__

__hash__() -> int

Return stable hash based on node ID.

Returns

int

__new__

__new__(*args: Any, **kwargs: Any) -> Node

Create or reuse a Node instance.

Stateless subclasses (no parameterized __init__) share one singleton instance per class — matching the original dagpipe behaviour underpinning set_registry-style configuration. Subclasses that declare an __init__ requiring instance-state arguments get a fresh instance per construction so pipeline builders can inject per-run dependencies.

__repr__

__repr__() -> str

Return debug representation.

Returns

str

__str__

__str__() -> str

Return display representation.

Returns

str

clean_id_and_name classmethod

clean_id_and_name() -> None

Normalize and validate node ID and display name.

Raises:

Type	Description
TypeError	If ID is not a string.
ValueError	If ID format is invalid.

Notes

Guarantees:

- Generates ID from module and class name if missing.
- Validates ID format.
- Generates human-readable name if missing.

fork

fork(
    state: State,
    *,
    payload_update: Any = None,
    confidence_delta: float = 0.0,
    metadata_update: Any = None
) -> State

Create a child State attributed to this node.

Parameters:

Name	Type	Description	Default
state	State	Parent execution state.	required
payload_update	Any	Dot-path payload updates.	None
confidence_delta	float	Confidence adjustment.	0.0
metadata_update	Any	Metadata updates.	None

Returns:

Name	Type	Description
State	State	New child execution state.

Notes

Responsibilities:

- Convenience wrapper around `State.fork()` that automatically
  records this node's ID in state history.

is_async

is_async() -> bool

Return whether this node executes asynchronously.

node_id_to_name staticmethod

node_id_to_name(node_id: str) -> str

Convert a dotted snake_case node ID into a human-readable name.

Parameters:

Name	Type	Description	Default
node_id	str	Unique node identifier (e.g., 'entity.resolve.numeric_merchant').	required

Returns:

Name	Type	Description
str	str	Human-readable display name (e.g., 'Entity › Resolve › Numeric Merchant').

resolve abstractmethod

resolve(state: State) -> Iterable[State]

Execute node logic.

Parameters:

Name	Type	Description	Default
state	State	Input execution state.	required

Returns:

Type	Description
Iterable[State]	Iterable[State]: Derived execution state(s).

Notes

Responsibilities:

- Subclasses implement specific resolution behavior.
- Must not mutate input state.
- Should use `fork()` to create child states.
- May yield zero states to terminate a branch.

run

run(state: State) -> tuple[State, ...]

Execute this node on a State.

Parameters:

Name	Type	Description	Default
state	State	Input execution state.	required

Returns:

Type	Description
tuple[State, ...]	tuple[State, ...]: Derived execution states.

Raises:

Type	Description
TypeError	If resolve() yields a non-State object.

Payload dataclass

Payload(_data: Mapping[str, Any])

Immutable hierarchical container with dot-path access.

Attributes:

Name	Type	Description
_data	Mapping[str, Any]	Immutable hierarchical data structure.

Notes

Responsibilities:

- Stores execution data used by `State`.
- Supports efficient atomic updates without modifying existing instances.
- `Payload` instances are fully thread-safe due to immutability.

Functions

as_dict

as_dict() -> Mapping[str, Any]

Return underlying mapping.

Returns:

Type	Description
Mapping[str, Any]	Mapping[str, Any]: Read-only view of the underlying data.

get

get(path: str, default: Any = None) -> Any

Retrieve value using dot-path.

Parameters:

Name	Type	Description	Default
path	str	Dot-separated path to the value.	required
default	Any	Default value if path doesn't exist.	None

Returns:

Name	Type	Description
Any	Any	The retrieved value or default.

has

has(path: str) -> bool

Return True if path exists.

Parameters:

Name	Type	Description	Default
path	str	Dot-separated path to check.	required

Returns:

Name	Type	Description
bool	bool	Existence of the path.

iter_paths classmethod

iter_paths(
    data: Mapping[str, Any], prefix: str = ""
) -> Iterable[str]

Recursively yield dot-paths for all leaf nodes.

Parameters:

Name	Type	Description	Default
data	Mapping[str, Any]	The mapping to iterate over.	required
prefix	str	Current path prefix.	''

Yields:

Name	Type	Description
str	Iterable[str]	Dot-path for each leaf node.

keys

keys() -> Iterable[str]

Return top-level keys.

Returns:

Type	Description
Iterable[str]	Iterable[str]: Iterator over top-level keys.

update

update(updates: Mapping[str, Any]) -> Payload

Create a new Payload with dot-path updates applied.

Parameters:

Name	Type	Description	Default
updates	Mapping[str, Any]	Dot-path to value mapping.	required

Returns:

Name	Type	Description
Payload	Payload	New immutable payload instance with updates.

Notes

Guarantees:

- Preserves existing data by copying only modified branches.
- Returns a new immutable `Payload`.

Pipeline dataclass

Pipeline(
    engine: Engine,
    state_cls: type[State],
    initial_payload: Payload,
)

Executable pipeline created from YAML configuration.

Attributes:

Name	Type	Description
engine	Engine	Execution engine responsible for running the pipeline.
state_cls	Type[State]	Dynamically created State subclass with configured schema.
initial_payload	Payload	Default payload used when execution begins.

Notes

Responsibilities:

- Encapsulates engine, state type, and initial payload.
- Provides a simplified interface for executing configured pipelines.
- Safe for concurrent execution if underlying nodes are thread-safe.

Functions

run

run(
    payload_override: Mapping[str, Any] | None = None,
) -> list[State]

Execute the pipeline.

Parameters:

Name	Type	Description	Default
payload_override	Mapping[str, Any] | None	Payload values overriding initial payload.	None

Returns:

Type	Description
list[State]	list[State]: Terminal execution states.

Notes

Responsibilities:

- Merges override payload with initial payload.
- Creates root `State` and executes engine.

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.

Schema dataclass

Schema(tree: Mapping[str, SchemaNode])

Immutable hierarchical schema defining allowed payload structure.

Attributes:

Name	Type	Description
tree	Mapping[str, SchemaNode]	Hierarchical schema definition.

Notes

Responsibilities:

- Validates `State` payloads and updates.
- Reusable across all `State` instances.
- Fully thread-safe due to immutability.

Functions

validate_payload

validate_payload(payload: Payload) -> None

Validate complete payload structure.

Parameters:

Name	Type	Description	Default
payload	Payload	Payload to validate.	required

Raises:

Type	Description
SchemaError	If payload violates schema.

validate_update

validate_update(updates: Mapping[str, Any]) -> None

Validate payload update paths.

Parameters:

Name	Type	Description	Default
updates	Mapping[str, Any]	Dot-path updates to validate.	required

Raises:

Type	Description
SchemaError	If any path is invalid according to the schema.

SchemaError

Bases: Exception

Raised when payload data violates the declared schema.

Indicates invalid structure, invalid path, or invalid type.

State dataclass

State(
    payload: Payload,
    confidence: float = 1.0,
    parent: State | None = None,
    depth: int = 0,
    history: tuple[str, ...] = tuple(),
    metadata: dict[str, Any] = dict(),
)

Immutable execution state propagated through dagpipe pipeline.

Attributes:

Name	Type	Description
payload	Payload	Execution data container.
schema	ClassVar[Schema]	Payload validation schema.
confidence	float	Execution confidence score.
parent	Optional[State]	Parent state reference.
depth	int	Execution depth.
history	Tuple[str, ...]	Ordered node execution lineage.
metadata	Dict[str, Any]	Execution metadata.

Notes

Responsibilities:

- Represents a complete execution snapshot at a specific point in
  pipeline traversal.
- Fundamental unit of execution in `dagpipe`.
- Fully thread-safe due to immutability.

Functions

__repr__

__repr__() -> str

Concise debug representation.

Avoids printing full data for large states.

fork

fork(
    *,
    payload_update: Mapping[str, Any] | None = None,
    confidence_delta: float = 0.0,
    node_id: str | None = None,
    metadata_update: Mapping[str, Any] | None = None
) -> State

Create a new child State derived from this state.

Parameters:

Name	Type	Description	Default
payload_update	Mapping[str, Any] | None	Dot-path updates applied to the payload.	None
confidence_delta	float	Adjustment applied to current confidence.	0.0
node_id	str | None	Identifier of the node creating this state.	None
metadata_update	Mapping[str, Any] | None	Updates merged into state metadata.	None

Returns:

Name	Type	Description
State	State	A new immutable State instance.

Notes

Guarantees:

- This is the only supported mechanism for modifying execution data.
- Validates payload updates, preserves lineage, increments depth,
  and appends to history.

get

get(key: str, default: Any = None) -> Any

Retrieve payload value.

Parameters:

Name	Type	Description	Default
key	str	Dot-path key.	required
default	Any	Fallback value.	None

Returns:

Name	Type	Description
Any	Any	Stored value or default.

has

has(key: str) -> bool

Check whether payload contains key.

Parameters:

Name	Type	Description	Default
key	str	Dot-path key.	required

Returns:

Name	Type	Description
bool	bool	Existence of the key.

lineage

lineage() -> tuple[State, ...]

Return lineage from root to this State.

Returns:

Type	Description
tuple[State, ...]	tuple[State, ...]: Ordered execution lineage (root first).

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

Functions

Functions

load_pipeline

load_pipeline(path: str) -> Pipeline

Load pipeline from YAML file.

Parameters:

Name	Type	Description	Default
path	str	Path to YAML configuration file.	required

Returns:

Name	Type	Description
Pipeline	Pipeline	Executable pipeline instance.

Notes

Responsibilities:

- Loads YAML configuration and builds schema.
- Creates `State` subclass and loads `Node` instances.
- Builds `Graph` topology and initializes `Engine`.

• Dagpipe