Skip to content

flow

Chain flow construction and validation.

A ChainFlow is the immutable routing plan for a unit of work: which stages exist, what they require, and how they connect. Build one with ChainFlowBuilder; the build step validates the topology once so consumers can rely on the flow's shape without re-checking.

ChainFlowValidationError

Bases: ValueError

Raised when a chain flow fails structural validation at build time.

Source code in horde_sdk/worker/chaining/flow.py 
17
18
class ChainFlowValidationError(ValueError):
    """Raised when a chain flow fails structural validation at build time."""

ChainFlow

An immutable, validated directed acyclic graph of stage nodes.

Instances are created by ChainFlowBuilder.build; the constructor assumes already-validated inputs.

Source code in horde_sdk/worker/chaining/flow.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class ChainFlow:
    """An immutable, validated directed acyclic graph of stage nodes.

    Instances are created by `ChainFlowBuilder.build`; the constructor assumes already-validated inputs.
    """

    def __init__(
        self,
        *,
        graph: TypedChainGraph[ChainNodeHandle],
        nodes: dict[ChainNodeHandle, ChainStageNode],
        edges: dict[tuple[ChainNodeHandle, ChainNodeHandle], ChainEdge],
    ) -> None:
        """Initialize the flow from builder-validated parts.

        Args:
            graph (TypedChainGraph[ChainNodeHandle]): The validated topology.
            nodes (dict[ChainNodeHandle, ChainStageNode]): The node payloads keyed by handle.
            edges (dict[tuple[ChainNodeHandle, ChainNodeHandle], ChainEdge]): The edge payloads keyed by
                (source, target).
        """
        self._graph = graph
        self._nodes = dict(nodes)
        self._edges = dict(edges)
        self._topological_order = graph.topological_order()
        self._entry_progress_index = {node.entry_progress: node.handle for node in nodes.values()}
        self._completion_progress_index = {node.completion_progress: node.handle for node in nodes.values()}

    @property
    def handles(self) -> tuple[ChainNodeHandle, ...]:
        """The node handles in topological order."""
        return self._topological_order

    @property
    def nodes(self) -> tuple[ChainStageNode, ...]:
        """The stage nodes in topological order."""
        return tuple(self._nodes[handle] for handle in self._topological_order)

    @property
    def edges(self) -> tuple[ChainEdge, ...]:
        """All edges in the flow."""
        return tuple(self._edges.values())

    @property
    def source_handle(self) -> ChainNodeHandle:
        """The handle of the flow's single starting node."""
        return self._graph.sources()[0]

    def get_node(self, handle: ChainNodeHandle) -> ChainStageNode:
        """Return the stage node for a handle.

        Args:
            handle (ChainNodeHandle): The handle to resolve.

        Raises:
            KeyError: If the handle is not part of this flow.
        """
        return self._nodes[handle]

    def get_node_handle(self, name: str) -> ChainNodeHandle:
        """Return the handle for a node name.

        Args:
            name (str): The node name.

        Raises:
            KeyError: If no node has the given name.
        """
        handle = ChainNodeHandle(name=name)
        if handle not in self._nodes:
            raise KeyError(f"No node named '{name}' in this flow")
        return handle

    def get_edge(self, source: ChainNodeHandle, target: ChainNodeHandle) -> ChainEdge:
        """Return the edge between two handles.

        Args:
            source (ChainNodeHandle): The edge origin.
            target (ChainNodeHandle): The edge destination.

        Raises:
            KeyError: If there is no such edge.
        """
        return self._edges[(source, target)]

    def successors(self, handle: ChainNodeHandle) -> tuple[ChainNodeHandle, ...]:
        """Return the direct successors of a node."""
        return self._graph.successors(handle)

    def predecessors(self, handle: ChainNodeHandle) -> tuple[ChainNodeHandle, ...]:
        """Return the direct predecessors of a node."""
        return self._graph.predecessors(handle)

    def descendants(self, handle: ChainNodeHandle) -> set[ChainNodeHandle]:
        """Return all nodes reachable from a node, excluding the node itself."""
        return self._graph.descendants(handle)

    def node_for_entry_progress(self, progress: GENERATION_PROGRESS) -> ChainNodeHandle | None:
        """Return the node whose stage begins at the given generation progress, if any."""
        return self._entry_progress_index.get(progress)

    def node_for_completion_progress(self, progress: GENERATION_PROGRESS) -> ChainNodeHandle | None:
        """Return the node whose stage completes at the given generation progress, if any."""
        return self._completion_progress_index.get(progress)

handles property 

handles: tuple[ChainNodeHandle, ...]

The node handles in topological order.

nodes property 

nodes: tuple[ChainStageNode, ...]

The stage nodes in topological order.

edges property 

edges: tuple[ChainEdge, ...]

All edges in the flow.

source_handle property 

source_handle: ChainNodeHandle

The handle of the flow's single starting node.

__init__ 

__init__(
    *,
    graph: TypedChainGraph[ChainNodeHandle],
    nodes: dict[ChainNodeHandle, ChainStageNode],
    edges: dict[
        tuple[ChainNodeHandle, ChainNodeHandle], ChainEdge
    ]
) -> None

Initialize the flow from builder-validated parts.

Parameters:

Source code in horde_sdk/worker/chaining/flow.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    *,
    graph: TypedChainGraph[ChainNodeHandle],
    nodes: dict[ChainNodeHandle, ChainStageNode],
    edges: dict[tuple[ChainNodeHandle, ChainNodeHandle], ChainEdge],
) -> None:
    """Initialize the flow from builder-validated parts.

    Args:
        graph (TypedChainGraph[ChainNodeHandle]): The validated topology.
        nodes (dict[ChainNodeHandle, ChainStageNode]): The node payloads keyed by handle.
        edges (dict[tuple[ChainNodeHandle, ChainNodeHandle], ChainEdge]): The edge payloads keyed by
            (source, target).
    """
    self._graph = graph
    self._nodes = dict(nodes)
    self._edges = dict(edges)
    self._topological_order = graph.topological_order()
    self._entry_progress_index = {node.entry_progress: node.handle for node in nodes.values()}
    self._completion_progress_index = {node.completion_progress: node.handle for node in nodes.values()}

get_node 

get_node(handle: ChainNodeHandle) -> ChainStageNode

Return the stage node for a handle.

Parameters:

Raises:

  • KeyError –

    If the handle is not part of this flow.

Source code in horde_sdk/worker/chaining/flow.py 
69
70
71
72
73
74
75
76
77
78
def get_node(self, handle: ChainNodeHandle) -> ChainStageNode:
    """Return the stage node for a handle.

    Args:
        handle (ChainNodeHandle): The handle to resolve.

    Raises:
        KeyError: If the handle is not part of this flow.
    """
    return self._nodes[handle]

get_node_handle 

get_node_handle(name: str) -> ChainNodeHandle

Return the handle for a node name.

Parameters:

  • name (str) –

    The node name.

Raises:

  • KeyError –

    If no node has the given name.

Source code in horde_sdk/worker/chaining/flow.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
def get_node_handle(self, name: str) -> ChainNodeHandle:
    """Return the handle for a node name.

    Args:
        name (str): The node name.

    Raises:
        KeyError: If no node has the given name.
    """
    handle = ChainNodeHandle(name=name)
    if handle not in self._nodes:
        raise KeyError(f"No node named '{name}' in this flow")
    return handle

get_edge 

get_edge(
    source: ChainNodeHandle, target: ChainNodeHandle
) -> ChainEdge

Return the edge between two handles.

Parameters:

Raises:

  • KeyError –

    If there is no such edge.

Source code in horde_sdk/worker/chaining/flow.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def get_edge(self, source: ChainNodeHandle, target: ChainNodeHandle) -> ChainEdge:
    """Return the edge between two handles.

    Args:
        source (ChainNodeHandle): The edge origin.
        target (ChainNodeHandle): The edge destination.

    Raises:
        KeyError: If there is no such edge.
    """
    return self._edges[(source, target)]

successors 

successors(
    handle: ChainNodeHandle,
) -> tuple[ChainNodeHandle, ...]

Return the direct successors of a node.

Source code in horde_sdk/worker/chaining/flow.py 
106
107
108
def successors(self, handle: ChainNodeHandle) -> tuple[ChainNodeHandle, ...]:
    """Return the direct successors of a node."""
    return self._graph.successors(handle)

predecessors 

predecessors(
    handle: ChainNodeHandle,
) -> tuple[ChainNodeHandle, ...]

Return the direct predecessors of a node.

Source code in horde_sdk/worker/chaining/flow.py 
110
111
112
def predecessors(self, handle: ChainNodeHandle) -> tuple[ChainNodeHandle, ...]:
    """Return the direct predecessors of a node."""
    return self._graph.predecessors(handle)

descendants 

descendants(
    handle: ChainNodeHandle,
) -> set[ChainNodeHandle]

Return all nodes reachable from a node, excluding the node itself.

Source code in horde_sdk/worker/chaining/flow.py 
114
115
116
def descendants(self, handle: ChainNodeHandle) -> set[ChainNodeHandle]:
    """Return all nodes reachable from a node, excluding the node itself."""
    return self._graph.descendants(handle)

node_for_entry_progress 

node_for_entry_progress(
    progress: GENERATION_PROGRESS,
) -> ChainNodeHandle | None

Return the node whose stage begins at the given generation progress, if any.

Source code in horde_sdk/worker/chaining/flow.py 
118
119
120
def node_for_entry_progress(self, progress: GENERATION_PROGRESS) -> ChainNodeHandle | None:
    """Return the node whose stage begins at the given generation progress, if any."""
    return self._entry_progress_index.get(progress)

node_for_completion_progress 

node_for_completion_progress(
    progress: GENERATION_PROGRESS,
) -> ChainNodeHandle | None

Return the node whose stage completes at the given generation progress, if any.

Source code in horde_sdk/worker/chaining/flow.py 
122
123
124
def node_for_completion_progress(self, progress: GENERATION_PROGRESS) -> ChainNodeHandle | None:
    """Return the node whose stage completes at the given generation progress, if any."""
    return self._completion_progress_index.get(progress)

ChainFlowBuilder

Accumulates stage nodes and edges, then validates and freezes them into a ChainFlow.

Source code in horde_sdk/worker/chaining/flow.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
class ChainFlowBuilder:
    """Accumulates stage nodes and edges, then validates and freezes them into a `ChainFlow`."""

    def __init__(self) -> None:
        """Initialize an empty builder."""
        self._graph: TypedChainGraph[ChainNodeHandle] = TypedChainGraph()
        self._nodes: dict[ChainNodeHandle, ChainStageNode] = {}
        self._edges: dict[tuple[ChainNodeHandle, ChainNodeHandle], ChainEdge] = {}

    def add_stage(self, node: ChainStageNode) -> ChainNodeHandle:
        """Add a stage node to the flow under construction.

        Args:
            node (ChainStageNode): The stage to add.

        Returns:
            ChainNodeHandle: The handle referring to the added node.

        Raises:
            ChainFlowValidationError: If a node with the same name already exists.
        """
        handle = node.handle
        if handle in self._nodes:
            raise ChainFlowValidationError(f"A node named '{node.name}' already exists in this flow")
        self._graph.add_node(handle)
        self._nodes[handle] = node
        return handle

    def connect(
        self,
        source: ChainNodeHandle,
        target: ChainNodeHandle,
        kind: CHAIN_EDGE_KIND = CHAIN_EDGE_KIND.SAME_GENERATION,
        *,
        edge: ChainEdge | None = None,
    ) -> ChainEdge:
        """Connect two stages with a directed edge.

        Args:
            source (ChainNodeHandle): The stage whose completion feeds the edge.
            target (ChainNodeHandle): The stage that becomes eligible when the edge fires.
            kind (CHAIN_EDGE_KIND, optional): The edge kind for a default edge.
                Defaults to CHAIN_EDGE_KIND.SAME_GENERATION.
            edge (ChainEdge | None, optional): A pre-built edge to use instead of constructing a default one.
                Its source and target must match the given handles. Defaults to None.

        Returns:
            ChainEdge: The edge that was added.

        Raises:
            ChainFlowValidationError: If either handle is unknown or the edge endpoints do not match.
        """
        if source not in self._nodes:
            raise ChainFlowValidationError(f"Unknown source node '{source}'")
        if target not in self._nodes:
            raise ChainFlowValidationError(f"Unknown target node '{target}'")

        if edge is None:
            edge = ChainEdge(kind=kind, source=source, target=target)
        elif edge.source != source or edge.target != target:
            raise ChainFlowValidationError(
                f"Edge endpoints ({edge.source} -> {edge.target}) do not match connect() arguments "
                f"({source} -> {target})",
            )

        self._graph.add_edge(source, target)
        self._edges[(source, target)] = edge
        return edge

    def build(self) -> ChainFlow:
        """Validate the accumulated topology and freeze it into a `ChainFlow`.

        Returns:
            ChainFlow: The validated flow.

        Raises:
            ChainFlowValidationError: If the flow is empty, cyclic, disconnected, has multiple starting nodes,
                or two stages share an entry or completion progress state.
        """
        if not self._nodes:
            raise ChainFlowValidationError("A chain flow requires at least one stage")

        if not self._graph.is_directed_acyclic():
            raise ChainFlowValidationError("A chain flow must be acyclic")

        if not self._graph.is_connected():
            raise ChainFlowValidationError("A chain flow must be connected; found unreachable stages")

        flow_sources = self._graph.sources()
        if len(flow_sources) != 1:
            source_names = ", ".join(str(handle) for handle in flow_sources)
            raise ChainFlowValidationError(f"A chain flow requires exactly one starting stage; found: {source_names}")

        entry_progress_seen: dict[GENERATION_PROGRESS, ChainNodeHandle] = {}
        completion_progress_seen: dict[GENERATION_PROGRESS, ChainNodeHandle] = {}
        for handle, node in self._nodes.items():
            if node.entry_progress in entry_progress_seen:
                raise ChainFlowValidationError(
                    f"Stages '{entry_progress_seen[node.entry_progress]}' and '{handle}' share entry progress "
                    f"{node.entry_progress}; progress-driven routing requires unique stage boundaries",
                )
            if node.completion_progress in completion_progress_seen:
                raise ChainFlowValidationError(
                    f"Stages '{completion_progress_seen[node.completion_progress]}' and '{handle}' share "
                    f"completion progress {node.completion_progress}; progress-driven routing requires unique "
                    "stage boundaries",
                )
            entry_progress_seen[node.entry_progress] = handle
            completion_progress_seen[node.completion_progress] = handle

        return ChainFlow(graph=self._graph, nodes=self._nodes, edges=self._edges)

__init__ 

__init__() -> None

Initialize an empty builder.

Source code in horde_sdk/worker/chaining/flow.py 
130
131
132
133
134
def __init__(self) -> None:
    """Initialize an empty builder."""
    self._graph: TypedChainGraph[ChainNodeHandle] = TypedChainGraph()
    self._nodes: dict[ChainNodeHandle, ChainStageNode] = {}
    self._edges: dict[tuple[ChainNodeHandle, ChainNodeHandle], ChainEdge] = {}

add_stage 

add_stage(node: ChainStageNode) -> ChainNodeHandle

Add a stage node to the flow under construction.

Parameters:

Returns:

  • ChainNodeHandle ( ChainNodeHandle ) –

    The handle referring to the added node.

Raises:

Source code in horde_sdk/worker/chaining/flow.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def add_stage(self, node: ChainStageNode) -> ChainNodeHandle:
    """Add a stage node to the flow under construction.

    Args:
        node (ChainStageNode): The stage to add.

    Returns:
        ChainNodeHandle: The handle referring to the added node.

    Raises:
        ChainFlowValidationError: If a node with the same name already exists.
    """
    handle = node.handle
    if handle in self._nodes:
        raise ChainFlowValidationError(f"A node named '{node.name}' already exists in this flow")
    self._graph.add_node(handle)
    self._nodes[handle] = node
    return handle

connect 

connect(
    source: ChainNodeHandle,
    target: ChainNodeHandle,
    kind: CHAIN_EDGE_KIND = CHAIN_EDGE_KIND.SAME_GENERATION,
    *,
    edge: ChainEdge | None = None
) -> ChainEdge

Connect two stages with a directed edge.

Parameters:

  • source (ChainNodeHandle) –

    The stage whose completion feeds the edge.

  • target (ChainNodeHandle) –

    The stage that becomes eligible when the edge fires.

  • kind (CHAIN_EDGE_KIND, default: SAME_GENERATION ) –

    The edge kind for a default edge. Defaults to CHAIN_EDGE_KIND.SAME_GENERATION.

  • edge (ChainEdge | None, default: None ) –

    A pre-built edge to use instead of constructing a default one. Its source and target must match the given handles. Defaults to None.

Returns:

  • ChainEdge ( ChainEdge ) –

    The edge that was added.

Raises:

Source code in horde_sdk/worker/chaining/flow.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def connect(
    self,
    source: ChainNodeHandle,
    target: ChainNodeHandle,
    kind: CHAIN_EDGE_KIND = CHAIN_EDGE_KIND.SAME_GENERATION,
    *,
    edge: ChainEdge | None = None,
) -> ChainEdge:
    """Connect two stages with a directed edge.

    Args:
        source (ChainNodeHandle): The stage whose completion feeds the edge.
        target (ChainNodeHandle): The stage that becomes eligible when the edge fires.
        kind (CHAIN_EDGE_KIND, optional): The edge kind for a default edge.
            Defaults to CHAIN_EDGE_KIND.SAME_GENERATION.
        edge (ChainEdge | None, optional): A pre-built edge to use instead of constructing a default one.
            Its source and target must match the given handles. Defaults to None.

    Returns:
        ChainEdge: The edge that was added.

    Raises:
        ChainFlowValidationError: If either handle is unknown or the edge endpoints do not match.
    """
    if source not in self._nodes:
        raise ChainFlowValidationError(f"Unknown source node '{source}'")
    if target not in self._nodes:
        raise ChainFlowValidationError(f"Unknown target node '{target}'")

    if edge is None:
        edge = ChainEdge(kind=kind, source=source, target=target)
    elif edge.source != source or edge.target != target:
        raise ChainFlowValidationError(
            f"Edge endpoints ({edge.source} -> {edge.target}) do not match connect() arguments "
            f"({source} -> {target})",
        )

    self._graph.add_edge(source, target)
    self._edges[(source, target)] = edge
    return edge

build 

build() -> ChainFlow

Validate the accumulated topology and freeze it into a ChainFlow.

Returns:

  • ChainFlow ( ChainFlow ) –

    The validated flow.

Raises:

  • ChainFlowValidationError –

    If the flow is empty, cyclic, disconnected, has multiple starting nodes, or two stages share an entry or completion progress state.

Source code in horde_sdk/worker/chaining/flow.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def build(self) -> ChainFlow:
    """Validate the accumulated topology and freeze it into a `ChainFlow`.

    Returns:
        ChainFlow: The validated flow.

    Raises:
        ChainFlowValidationError: If the flow is empty, cyclic, disconnected, has multiple starting nodes,
            or two stages share an entry or completion progress state.
    """
    if not self._nodes:
        raise ChainFlowValidationError("A chain flow requires at least one stage")

    if not self._graph.is_directed_acyclic():
        raise ChainFlowValidationError("A chain flow must be acyclic")

    if not self._graph.is_connected():
        raise ChainFlowValidationError("A chain flow must be connected; found unreachable stages")

    flow_sources = self._graph.sources()
    if len(flow_sources) != 1:
        source_names = ", ".join(str(handle) for handle in flow_sources)
        raise ChainFlowValidationError(f"A chain flow requires exactly one starting stage; found: {source_names}")

    entry_progress_seen: dict[GENERATION_PROGRESS, ChainNodeHandle] = {}
    completion_progress_seen: dict[GENERATION_PROGRESS, ChainNodeHandle] = {}
    for handle, node in self._nodes.items():
        if node.entry_progress in entry_progress_seen:
            raise ChainFlowValidationError(
                f"Stages '{entry_progress_seen[node.entry_progress]}' and '{handle}' share entry progress "
                f"{node.entry_progress}; progress-driven routing requires unique stage boundaries",
            )
        if node.completion_progress in completion_progress_seen:
            raise ChainFlowValidationError(
                f"Stages '{completion_progress_seen[node.completion_progress]}' and '{handle}' share "
                f"completion progress {node.completion_progress}; progress-driven routing requires unique "
                "stage boundaries",
            )
        entry_progress_seen[node.entry_progress] = handle
        completion_progress_seen[node.completion_progress] = handle

    return ChainFlow(graph=self._graph, nodes=self._nodes, edges=self._edges)