Skip to content

context

Per-unit-of-work chain execution state.

A ChainExecutionContext tracks each stage node's execution state for one unit of work. It is descriptive: the owning orchestration performs the actual work and reports progress here, either by marking nodes directly or by feeding generation-progress values through advance_for_progress.

ChainConsistencyError

Bases: RuntimeError

Raised when a reported progression contradicts the chain's routing plan.

Source code in horde_sdk/worker/chaining/context.py 
33
34
class ChainConsistencyError(RuntimeError):
    """Raised when a reported progression contradicts the chain's routing plan."""

ChainExecutionContext

Thread-safe execution state for one traversal of a ChainFlow.

Source code in horde_sdk/worker/chaining/context.py 
 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
125
126
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
238
239
240
241
242
243
244
245
246
247
248
249
250
251
class ChainExecutionContext:
    """Thread-safe execution state for one traversal of a `ChainFlow`."""

    def __init__(self, flow: ChainFlow) -> None:
        """Initialize all nodes as pending.

        Args:
            flow (ChainFlow): The flow being traversed.
        """
        self._flow = flow
        self._lock = threading.RLock()
        self._states: dict[ChainNodeHandle, CHAIN_NODE_STATE] = dict.fromkeys(
            flow.handles,
            CHAIN_NODE_STATE.PENDING,
        )
        self._errors: dict[ChainNodeHandle, list[str]] = {handle: [] for handle in flow.handles}
        self._started_at: dict[ChainNodeHandle, float] = {}
        self._finished_at: dict[ChainNodeHandle, float] = {}

    @property
    def flow(self) -> ChainFlow:
        """The flow this context tracks."""
        return self._flow

    def node_state(self, handle: ChainNodeHandle) -> CHAIN_NODE_STATE:
        """Return the current state of a node."""
        with self._lock:
            return self._states[handle]

    def node_errors(self, handle: ChainNodeHandle) -> tuple[str, ...]:
        """Return the errors recorded against a node."""
        with self._lock:
            return tuple(self._errors[handle])

    def node_duration(self, handle: ChainNodeHandle) -> float | None:
        """Seconds a node spent executing, or None when it has not both started and finished."""
        with self._lock:
            started = self._started_at.get(handle)
            finished = self._finished_at.get(handle)
            if started is None or finished is None:
                return None
            return finished - started

    def snapshot(self) -> dict[str, CHAIN_NODE_STATE]:
        """Return a point-in-time view of every node's state, keyed by node name."""
        with self._lock:
            return {handle.name: state for handle, state in self._states.items()}

    def _is_dependency_satisfied(self, handle: ChainNodeHandle) -> bool:
        """Whether all predecessors of a node have finished without failing."""
        return all(
            self._states[predecessor] in {CHAIN_NODE_STATE.COMPLETED, CHAIN_NODE_STATE.SKIPPED}
            for predecessor in self._flow.predecessors(handle)
        )

    def ready_nodes(self) -> tuple[ChainNodeHandle, ...]:
        """Return the nodes eligible to start: pending or ready, with all dependencies satisfied."""
        with self._lock:
            return tuple(
                handle
                for handle in self._flow.handles
                if self._states[handle] in {CHAIN_NODE_STATE.PENDING, CHAIN_NODE_STATE.READY}
                and self._is_dependency_satisfied(handle)
            )

    @property
    def is_finished(self) -> bool:
        """Whether every node has reached a terminal state."""
        with self._lock:
            return all(state in _TERMINAL_NODE_STATES for state in self._states.values())

    @property
    def has_failed(self) -> bool:
        """Whether any node failed."""
        with self._lock:
            return any(state == CHAIN_NODE_STATE.FAILED for state in self._states.values())

    def mark_executing(self, handle: ChainNodeHandle) -> None:
        """Mark a node as executing.

        Optional pending predecessors are skipped implicitly; a required predecessor that has not finished is a
        routing contradiction.

        Args:
            handle (ChainNodeHandle): The node that started executing.

        Raises:
            ChainConsistencyError: If the node is already terminal or a required dependency has not finished.
        """
        with self._lock:
            current = self._states[handle]
            if current in _TERMINAL_NODE_STATES or current == CHAIN_NODE_STATE.EXECUTING:
                raise ChainConsistencyError(f"Node '{handle}' cannot start executing from state {current}")

            for predecessor in self._flow.predecessors(handle):
                predecessor_state = self._states[predecessor]
                if predecessor_state in {CHAIN_NODE_STATE.COMPLETED, CHAIN_NODE_STATE.SKIPPED}:
                    continue
                if self._flow.get_node(predecessor).optional:
                    self._skip_subtree_rooted_at(predecessor, reason=f"optional stage bypassed by '{handle}'")
                    continue
                raise ChainConsistencyError(
                    f"Node '{handle}' cannot start: required predecessor '{predecessor}' is {predecessor_state}",
                )

            self._states[handle] = CHAIN_NODE_STATE.EXECUTING
            self._started_at[handle] = time.monotonic()

    def _skip_subtree_rooted_at(self, handle: ChainNodeHandle, *, reason: str) -> None:
        """Skip a non-terminal node; used for optional stages that the traversal bypassed."""
        if self._states[handle] in _TERMINAL_NODE_STATES:
            return
        self._states[handle] = CHAIN_NODE_STATE.SKIPPED
        self._errors[handle].append(reason)

    def mark_completed(self, handle: ChainNodeHandle) -> tuple[ChainNodeHandle, ...]:
        """Mark a node as completed and return the nodes that became eligible as a result.

        Args:
            handle (ChainNodeHandle): The node that finished.

        Returns:
            tuple[ChainNodeHandle, ...]: The successors that are now eligible to start.

        Raises:
            ChainConsistencyError: If the node is already terminal.
        """
        with self._lock:
            current = self._states[handle]
            if current in _TERMINAL_NODE_STATES:
                raise ChainConsistencyError(f"Node '{handle}' cannot complete from state {current}")

            self._states[handle] = CHAIN_NODE_STATE.COMPLETED
            self._finished_at[handle] = time.monotonic()
            if handle not in self._started_at:
                self._started_at[handle] = self._finished_at[handle]

            return tuple(
                successor
                for successor in self._flow.successors(handle)
                if self._states[successor] == CHAIN_NODE_STATE.PENDING and self._is_dependency_satisfied(successor)
            )

    def mark_failed(self, handle: ChainNodeHandle, *, error: str | None = None) -> None:
        """Mark a node as failed and skip everything downstream of it.

        Args:
            handle (ChainNodeHandle): The node that failed.
            error (str | None, optional): A description of the failure. Defaults to None.
        """
        with self._lock:
            self._states[handle] = CHAIN_NODE_STATE.FAILED
            self._finished_at[handle] = time.monotonic()
            if error is not None:
                self._errors[handle].append(error)

            for descendant in self._flow.descendants(handle):
                if self._states[descendant] not in _TERMINAL_NODE_STATES:
                    self._states[descendant] = CHAIN_NODE_STATE.SKIPPED
                    self._errors[descendant].append(f"upstream stage '{handle}' failed")

    def mark_skipped(self, handle: ChainNodeHandle, *, reason: str | None = None) -> None:
        """Mark a node as skipped.

        Args:
            handle (ChainNodeHandle): The node to skip.
            reason (str | None, optional): Why the node was skipped. Defaults to None.

        Raises:
            ChainConsistencyError: If the node is not optional or is already terminal.
        """
        with self._lock:
            if not self._flow.get_node(handle).optional:
                raise ChainConsistencyError(f"Node '{handle}' is required and cannot be skipped explicitly")
            if self._states[handle] in _TERMINAL_NODE_STATES:
                raise ChainConsistencyError(f"Node '{handle}' is already {self._states[handle]}")
            self._states[handle] = CHAIN_NODE_STATE.SKIPPED
            if reason is not None:
                self._errors[handle].append(reason)

    def advance_for_progress(self, progress: GENERATION_PROGRESS) -> tuple[ChainNodeHandle, ...]:
        """Advance the chain state from a generation-progress observation.

        This is the bridge that lets an orchestrator drive the chain purely from the generation's state machine:
        a stage's `entry_progress` marks its node executing, a stage's `completion_progress` marks its node
        completed, and a terminal failure progress fails whichever node is executing.

        Args:
            progress (GENERATION_PROGRESS): The progress value the generation just reached.

        Returns:
            tuple[ChainNodeHandle, ...]: The nodes that became eligible to start due to this observation.
        """
        with self._lock:
            if progress in _TERMINAL_FAILURE_PROGRESS:
                for handle in self._flow.handles:
                    if self._states[handle] == CHAIN_NODE_STATE.EXECUTING:
                        self.mark_failed(handle, error=f"generation reached {progress}")
                return ()

            entering = self._flow.node_for_entry_progress(progress)
            if entering is not None and self._states[entering] in {
                CHAIN_NODE_STATE.PENDING,
                CHAIN_NODE_STATE.READY,
            }:
                self.mark_executing(entering)

            completing = self._flow.node_for_completion_progress(progress)
            if completing is not None and self._states[completing] not in _TERMINAL_NODE_STATES:
                newly_ready = self.mark_completed(completing)
                for handle in newly_ready:
                    self._states[handle] = CHAIN_NODE_STATE.READY
                return newly_ready

            return ()

flow property 

flow: ChainFlow

The flow this context tracks.

is_finished property 

is_finished: bool

Whether every node has reached a terminal state.

has_failed property 

has_failed: bool

Whether any node failed.

__init__ 

__init__(flow: ChainFlow) -> None

Initialize all nodes as pending.

Parameters:

  • flow (ChainFlow) –

    The flow being traversed.

Source code in horde_sdk/worker/chaining/context.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init__(self, flow: ChainFlow) -> None:
    """Initialize all nodes as pending.

    Args:
        flow (ChainFlow): The flow being traversed.
    """
    self._flow = flow
    self._lock = threading.RLock()
    self._states: dict[ChainNodeHandle, CHAIN_NODE_STATE] = dict.fromkeys(
        flow.handles,
        CHAIN_NODE_STATE.PENDING,
    )
    self._errors: dict[ChainNodeHandle, list[str]] = {handle: [] for handle in flow.handles}
    self._started_at: dict[ChainNodeHandle, float] = {}
    self._finished_at: dict[ChainNodeHandle, float] = {}

node_state 

node_state(handle: ChainNodeHandle) -> CHAIN_NODE_STATE

Return the current state of a node.

Source code in horde_sdk/worker/chaining/context.py 
61
62
63
64
def node_state(self, handle: ChainNodeHandle) -> CHAIN_NODE_STATE:
    """Return the current state of a node."""
    with self._lock:
        return self._states[handle]

node_errors 

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

Return the errors recorded against a node.

Source code in horde_sdk/worker/chaining/context.py 
66
67
68
69
def node_errors(self, handle: ChainNodeHandle) -> tuple[str, ...]:
    """Return the errors recorded against a node."""
    with self._lock:
        return tuple(self._errors[handle])

node_duration 

node_duration(handle: ChainNodeHandle) -> float | None

Seconds a node spent executing, or None when it has not both started and finished.

Source code in horde_sdk/worker/chaining/context.py 
71
72
73
74
75
76
77
78
def node_duration(self, handle: ChainNodeHandle) -> float | None:
    """Seconds a node spent executing, or None when it has not both started and finished."""
    with self._lock:
        started = self._started_at.get(handle)
        finished = self._finished_at.get(handle)
        if started is None or finished is None:
            return None
        return finished - started

snapshot 

snapshot() -> dict[str, CHAIN_NODE_STATE]

Return a point-in-time view of every node's state, keyed by node name.

Source code in horde_sdk/worker/chaining/context.py 
80
81
82
83
def snapshot(self) -> dict[str, CHAIN_NODE_STATE]:
    """Return a point-in-time view of every node's state, keyed by node name."""
    with self._lock:
        return {handle.name: state for handle, state in self._states.items()}

ready_nodes 

ready_nodes() -> tuple[ChainNodeHandle, ...]

Return the nodes eligible to start: pending or ready, with all dependencies satisfied.

Source code in horde_sdk/worker/chaining/context.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
def ready_nodes(self) -> tuple[ChainNodeHandle, ...]:
    """Return the nodes eligible to start: pending or ready, with all dependencies satisfied."""
    with self._lock:
        return tuple(
            handle
            for handle in self._flow.handles
            if self._states[handle] in {CHAIN_NODE_STATE.PENDING, CHAIN_NODE_STATE.READY}
            and self._is_dependency_satisfied(handle)
        )

mark_executing 

mark_executing(handle: ChainNodeHandle) -> None

Mark a node as executing.

Optional pending predecessors are skipped implicitly; a required predecessor that has not finished is a routing contradiction.

Parameters:

Raises:

Source code in horde_sdk/worker/chaining/context.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def mark_executing(self, handle: ChainNodeHandle) -> None:
    """Mark a node as executing.

    Optional pending predecessors are skipped implicitly; a required predecessor that has not finished is a
    routing contradiction.

    Args:
        handle (ChainNodeHandle): The node that started executing.

    Raises:
        ChainConsistencyError: If the node is already terminal or a required dependency has not finished.
    """
    with self._lock:
        current = self._states[handle]
        if current in _TERMINAL_NODE_STATES or current == CHAIN_NODE_STATE.EXECUTING:
            raise ChainConsistencyError(f"Node '{handle}' cannot start executing from state {current}")

        for predecessor in self._flow.predecessors(handle):
            predecessor_state = self._states[predecessor]
            if predecessor_state in {CHAIN_NODE_STATE.COMPLETED, CHAIN_NODE_STATE.SKIPPED}:
                continue
            if self._flow.get_node(predecessor).optional:
                self._skip_subtree_rooted_at(predecessor, reason=f"optional stage bypassed by '{handle}'")
                continue
            raise ChainConsistencyError(
                f"Node '{handle}' cannot start: required predecessor '{predecessor}' is {predecessor_state}",
            )

        self._states[handle] = CHAIN_NODE_STATE.EXECUTING
        self._started_at[handle] = time.monotonic()

mark_completed 

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

Mark a node as completed and return the nodes that became eligible as a result.

Parameters:

Returns:

  • tuple[ChainNodeHandle, ...] –

    tuple[ChainNodeHandle, ...]: The successors that are now eligible to start.

Raises:

Source code in horde_sdk/worker/chaining/context.py 
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
def mark_completed(self, handle: ChainNodeHandle) -> tuple[ChainNodeHandle, ...]:
    """Mark a node as completed and return the nodes that became eligible as a result.

    Args:
        handle (ChainNodeHandle): The node that finished.

    Returns:
        tuple[ChainNodeHandle, ...]: The successors that are now eligible to start.

    Raises:
        ChainConsistencyError: If the node is already terminal.
    """
    with self._lock:
        current = self._states[handle]
        if current in _TERMINAL_NODE_STATES:
            raise ChainConsistencyError(f"Node '{handle}' cannot complete from state {current}")

        self._states[handle] = CHAIN_NODE_STATE.COMPLETED
        self._finished_at[handle] = time.monotonic()
        if handle not in self._started_at:
            self._started_at[handle] = self._finished_at[handle]

        return tuple(
            successor
            for successor in self._flow.successors(handle)
            if self._states[successor] == CHAIN_NODE_STATE.PENDING and self._is_dependency_satisfied(successor)
        )

mark_failed 

mark_failed(
    handle: ChainNodeHandle, *, error: str | None = None
) -> None

Mark a node as failed and skip everything downstream of it.

Parameters:

  • handle (ChainNodeHandle) –

    The node that failed.

  • error (str | None, default: None ) –

    A description of the failure. Defaults to None.

Source code in horde_sdk/worker/chaining/context.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def mark_failed(self, handle: ChainNodeHandle, *, error: str | None = None) -> None:
    """Mark a node as failed and skip everything downstream of it.

    Args:
        handle (ChainNodeHandle): The node that failed.
        error (str | None, optional): A description of the failure. Defaults to None.
    """
    with self._lock:
        self._states[handle] = CHAIN_NODE_STATE.FAILED
        self._finished_at[handle] = time.monotonic()
        if error is not None:
            self._errors[handle].append(error)

        for descendant in self._flow.descendants(handle):
            if self._states[descendant] not in _TERMINAL_NODE_STATES:
                self._states[descendant] = CHAIN_NODE_STATE.SKIPPED
                self._errors[descendant].append(f"upstream stage '{handle}' failed")

mark_skipped 

mark_skipped(
    handle: ChainNodeHandle, *, reason: str | None = None
) -> None

Mark a node as skipped.

Parameters:

  • handle (ChainNodeHandle) –

    The node to skip.

  • reason (str | None, default: None ) –

    Why the node was skipped. Defaults to None.

Raises:

Source code in horde_sdk/worker/chaining/context.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def mark_skipped(self, handle: ChainNodeHandle, *, reason: str | None = None) -> None:
    """Mark a node as skipped.

    Args:
        handle (ChainNodeHandle): The node to skip.
        reason (str | None, optional): Why the node was skipped. Defaults to None.

    Raises:
        ChainConsistencyError: If the node is not optional or is already terminal.
    """
    with self._lock:
        if not self._flow.get_node(handle).optional:
            raise ChainConsistencyError(f"Node '{handle}' is required and cannot be skipped explicitly")
        if self._states[handle] in _TERMINAL_NODE_STATES:
            raise ChainConsistencyError(f"Node '{handle}' is already {self._states[handle]}")
        self._states[handle] = CHAIN_NODE_STATE.SKIPPED
        if reason is not None:
            self._errors[handle].append(reason)

advance_for_progress 

advance_for_progress(
    progress: GENERATION_PROGRESS,
) -> tuple[ChainNodeHandle, ...]

Advance the chain state from a generation-progress observation.

This is the bridge that lets an orchestrator drive the chain purely from the generation's state machine: a stage's entry_progress marks its node executing, a stage's completion_progress marks its node completed, and a terminal failure progress fails whichever node is executing.

Parameters:

Returns:

  • tuple[ChainNodeHandle, ...] –

    tuple[ChainNodeHandle, ...]: The nodes that became eligible to start due to this observation.

Source code in horde_sdk/worker/chaining/context.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def advance_for_progress(self, progress: GENERATION_PROGRESS) -> tuple[ChainNodeHandle, ...]:
    """Advance the chain state from a generation-progress observation.

    This is the bridge that lets an orchestrator drive the chain purely from the generation's state machine:
    a stage's `entry_progress` marks its node executing, a stage's `completion_progress` marks its node
    completed, and a terminal failure progress fails whichever node is executing.

    Args:
        progress (GENERATION_PROGRESS): The progress value the generation just reached.

    Returns:
        tuple[ChainNodeHandle, ...]: The nodes that became eligible to start due to this observation.
    """
    with self._lock:
        if progress in _TERMINAL_FAILURE_PROGRESS:
            for handle in self._flow.handles:
                if self._states[handle] == CHAIN_NODE_STATE.EXECUTING:
                    self.mark_failed(handle, error=f"generation reached {progress}")
            return ()

        entering = self._flow.node_for_entry_progress(progress)
        if entering is not None and self._states[entering] in {
            CHAIN_NODE_STATE.PENDING,
            CHAIN_NODE_STATE.READY,
        }:
            self.mark_executing(entering)

        completing = self._flow.node_for_completion_progress(progress)
        if completing is not None and self._states[completing] not in _TERMINAL_NODE_STATES:
            newly_ready = self.mark_completed(completing)
            for handle in newly_ready:
                self._states[handle] = CHAIN_NODE_STATE.READY
            return newly_ready

        return ()