Coverage for src / lexigram / contracts / workflow / protocols.py: 100%
66 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-19 05:41 +0800
1"""Workflow orchestration protocols for Lexigram Framework.
3Defines protocols for pipeline execution, bulk processing, and saga
4coordination. These are the contracts that lexigram-workflow implements.
5"""
7from __future__ import annotations
9from enum import StrEnum
10from typing import Any, Protocol, TypeVar, runtime_checkable
12from lexigram.contracts.workflow.types import (
13 StateTransitionRecord as StateTransitionRecord,
14)
16StepResult = TypeVar("StepResult")
19@runtime_checkable
20class WorkflowGraphProtocol(Protocol):
21 """Protocol for workflow DAG/graph representations."""
23 def add_node(self, node_id: str, node: Any) -> None: ...
24 def add_edge(self, from_id: str, to_id: str) -> None: ...
25 def get_node(self, node_id: str) -> Any | None: ...
26 def topological_order(self) -> list[str]: ...
29@runtime_checkable
30class WorkflowNodeProtocol(Protocol):
31 """Protocol for individual workflow step nodes.
33 A workflow node represents a single unit of work in a directed graph.
34 Nodes receive the current shared state, perform their work, and
35 return a state update dict that is merged into the global state.
36 """
38 @property
39 def name(self) -> str:
40 """Unique node identifier within the workflow."""
41 ...
43 async def execute(self, state: dict[str, Any]) -> dict[str, Any]:
44 """Execute this node with the current workflow state.
46 Args:
47 state: Current shared workflow state.
49 Returns:
50 Dict of state updates to merge into the shared state.
51 """
52 ...
55@runtime_checkable
56class ApprovalProtocol(Protocol):
57 """Protocol for human-in-the-loop approval gates."""
59 async def request_approval(
60 self,
61 workflow_id: str,
62 step_id: str,
63 context: Any,
64 ) -> bool: ...
66 async def cancel_approval(self, approval_id: str) -> None: ...
69@runtime_checkable
70class ExecutionProtocol(Protocol):
71 """Protocol for workflow execution engines."""
73 async def execute(self, workflow_id: str, context: Any) -> Any: ...
74 async def resume(self, execution_id: str, result: Any) -> Any: ...
75 async def cancel(self, execution_id: str) -> None: ...
78__all__ = [
79 "ApprovalProtocol",
80 "BulkProcessorProtocol",
81 "ExecutionProtocol",
82 "PipelineContextProtocol",
83 "PipelineProtocol",
84 "PipelineStepProtocol",
85 "SagaManagerProtocol",
86 "SagaProtocol",
87 "SagaState",
88 "SagaStoreProtocol",
89 "StateMachineProtocol",
90 "StatePersistenceProtocol",
91 "StateTransitionRecord",
92 "StepResult",
93 "WorkflowGraphProtocol",
94 "WorkflowNodeProtocol",
95]
98class SagaState(StrEnum):
99 """Lifecycle states for an orchestration saga.
101 Transitions:
102 PENDING → RUNNING → COMPLETED
103 PENDING → RUNNING → COMPENSATING → FAILED
104 """
106 PENDING = "pending"
107 RUNNING = "running"
108 COMPENSATING = "compensating"
109 COMPLETED = "completed"
110 FAILED = "failed"
113@runtime_checkable
114class SagaStoreProtocol(Protocol):
115 """Protocol for durable saga state persistence.
117 Implementations back the store with a database or cache so that
118 saga state survives process restarts and multi-worker deployments.
119 """
121 async def save(
122 self,
123 saga_id: str,
124 state: SagaState,
125 data: dict[str, Any],
126 ) -> None:
127 """Persist the current saga state and its associated data.
129 Args:
130 saga_id: Stable identifier for the saga instance.
131 state: Current lifecycle state.
132 data: Arbitrary saga-specific payload (must be JSON-serialisable).
133 """
134 ...
136 async def load(
137 self,
138 saga_id: str,
139 ) -> tuple[SagaState, dict[str, Any]] | None:
140 """Load persisted saga state.
142 Args:
143 saga_id: Stable identifier for the saga instance.
145 Returns:
146 A ``(state, data)`` tuple, or ``None`` if no record exists.
147 """
148 ...
150 async def delete(self, saga_id: str) -> None:
151 """Remove a completed or failed saga record.
153 Args:
154 saga_id: Stable identifier for the saga instance.
155 """
156 ...
159@runtime_checkable
160class PipelineContextProtocol(Protocol):
161 """Protocol for step-to-step data passing within a pipeline.
163 Provides typed access to intermediate results and shared metadata
164 accumulated as the pipeline executes each step.
165 """
167 def get_result(
168 self, step_name: str
169 ) -> Any: # Intentional: step results are heterogeneous by design
170 """Return the result stored for the named step.
172 Args:
173 step_name: Unique name of the step whose result to retrieve.
175 Returns:
176 The stored result value, or ``None`` if the step has not run.
177 """
178 ...
180 def set_result(
181 self, step_name: str, value: Any
182 ) -> None: # Intentional: stores any step output
183 """Store a result value for the named step.
185 Args:
186 step_name: Unique name of the step producing the result.
187 value: Result value to store.
188 """
189 ...
191 @property
192 def metadata(self) -> dict[str, Any]:
193 """Arbitrary metadata shared across all steps in the pipeline."""
194 ...
197@runtime_checkable
198class PipelineStepProtocol(Protocol):
199 """Protocol for a single executable step within a pipeline.
201 Each step receives the shared context, performs its work, and stores
202 its output via ``context.set_result``.
203 """
205 @property
206 def name(self) -> str:
207 """Unique name identifying this step within its pipeline."""
208 ...
210 async def execute(self, context: PipelineContextProtocol) -> StepResult:
211 """Execute the step using the shared pipeline context.
213 Args:
214 context: Shared context supplying previous step results.
216 Returns:
217 The step's output value.
218 """
219 ...
222@runtime_checkable
223class PipelineProtocol(Protocol):
224 """Protocol for a composable, sequential pipeline of steps.
226 A pipeline collects named steps and executes them in registration
227 order, passing a shared context between them.
228 """
230 def add_step(self, step: PipelineStepProtocol) -> None:
231 """Register a step to be executed as part of this pipeline.
233 Args:
234 step: Step to append to the execution sequence.
235 """
236 ...
238 async def execute(
239 self, initial_context: dict[str, Any] | None = None
240 ) -> StepResult:
241 """Execute all registered steps in order.
243 Args:
244 initial_context: Optional seed data for the pipeline context.
246 Returns:
247 The result produced by the final step.
248 """
249 ...
252@runtime_checkable
253class BulkProcessorProtocol(Protocol):
254 """Protocol for processing a batch of items through a pipeline.
256 Implementations receive a homogeneous list of items and return a
257 list of processed results in the same order.
258 """
260 async def process_batch(self, items: list[Any]) -> list[Any]:
261 """Process a batch of items.
263 Args:
264 items: Input items to process.
266 Returns:
267 Processed results, one entry per input item.
268 """
269 ...
272@runtime_checkable
273class SagaProtocol(Protocol):
274 """Protocol for saga / process-manager implementations.
276 Sagas coordinate long-running business processes that span multiple
277 aggregates by reacting to domain events and dispatching commands.
278 """
280 async def handle(
281 self, event: Any
282 ) -> list[Any]: # Intentional: domain events and commands are heterogeneous
283 """Handle a domain event and produce commands.
285 Args:
286 event: Domain event to handle.
288 Returns:
289 List of commands to dispatch to the command bus.
290 """
291 ...
294@runtime_checkable
295class SagaManagerProtocol(Protocol):
296 """Protocol for saga lifecycle management.
298 The manager routes incoming events to all registered sagas and
299 coordinates their execution.
300 """
302 async def process(
303 self, event: Any
304 ) -> None: # Intentional: domain events are heterogeneous
305 """Process an event through all relevant sagas.
307 Args:
308 event: Domain event to route and process.
309 """
310 ...
313@runtime_checkable
314class StateMachineProtocol(Protocol):
315 """Protocol for finite state machine implementations.
317 A state machine manages transitions between named states according to
318 a defined set of allowed transitions. Callers check
319 :meth:`can_transition` before calling :meth:`transition` to avoid
320 raising :exc:`~lexigram.contracts.workflow.StateError`.
321 """
323 @property
324 def current_state(self) -> str:
325 """The name of the current state.
327 Returns:
328 String name of the active state.
329 """
330 ...
332 @property
333 def version(self) -> int:
334 """Current state version for optimistic concurrency control.
336 Returns:
337 Monotonic transition version starting at ``0``.
338 """
339 ...
341 def can_transition(self, event: str) -> bool:
342 """Return whether *event* is a valid trigger from the current state.
344 Args:
345 event: The event name to check.
347 Returns:
348 ``True`` when the transition is permitted; ``False`` otherwise.
349 """
350 ...
352 async def transition(self, event: str) -> str:
353 """Trigger *event*, advancing to the next state.
355 Args:
356 event: The event name that triggers the transition.
358 Returns:
359 The name of the new current state.
361 Raises:
362 StateError: When *event* is not permitted from the current state.
363 """
364 ...
366 async def recover(self, machine_id: str | None = None) -> str:
367 """Recover current state from persisted transitions.
369 Args:
370 machine_id: Optional machine identifier override. Implementations
371 may use the configured default when omitted.
373 Returns:
374 Recovered current state name.
375 """
376 ...
379@runtime_checkable
380class StatePersistenceProtocol(Protocol):
381 """Protocol for durable state transition persistence.
383 Implementations persist each transition as an append-only event stream so
384 a state machine can recover after process restarts and enforce optimistic
385 concurrency via transition versions.
386 """
388 async def append_transition(
389 self,
390 machine_id: str,
391 from_state: str,
392 event: str,
393 to_state: str,
394 expected_version: int,
395 ) -> int:
396 """Persist a transition with optimistic version check.
398 Args:
399 machine_id: Stable state machine identifier.
400 from_state: Previous state.
401 event: Transition trigger.
402 to_state: New state.
403 expected_version: Version caller believes is current.
405 Returns:
406 Newly persisted version.
408 Raises:
409 RuntimeError: When optimistic locking fails.
410 """
411 ...
413 async def load_transitions(self, machine_id: str) -> list[StateTransitionRecord]:
414 """Load persisted transitions ordered by ascending version.
416 Args:
417 machine_id: Stable state machine identifier.
419 Returns:
420 List of transition records in order.
421 """
422 ...
424 async def get_current_version(self, machine_id: str) -> int:
425 """Return current persisted version for *machine_id*.
427 Args:
428 machine_id: Stable state machine identifier.
430 Returns:
431 Latest transition version, or ``0`` when no transitions exist.
432 """
433 ...