Coverage for async_durable_execution/_operation/flow.py: 99%
1114 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-30 23:43 +0000
1"""Declarative acyclic durable workflow composition."""
3from __future__ import annotations
5import asyncio
6import builtins
7import functools
8import heapq
9import inspect
10from collections.abc import Awaitable, Callable, Iterable, Iterator, Mapping, Set
11from contextvars import ContextVar
12from dataclasses import dataclass, field, fields as dataclass_fields, is_dataclass
13from enum import Enum
14from typing import TYPE_CHECKING, Any, Generic, NoReturn, ParamSpec, TypeVar, cast
16from .._core import (
17 CallableRuntimeError,
18 DurableContext,
19 DurableExecutionsError,
20 ErrorObject,
21 ExecutionError,
22 ExtendedTypeSerDes,
23 InvalidStateError,
24 InvocationError,
25 OperationSubType,
26 SerDes,
27 SerDesError,
28 SuspendExecution,
29 TimedSuspendExecution,
30 ValidationError,
31 MappingModel,
32 _restore_sdk_control_error,
33 bind_current_context,
34 bind_durable_definition,
35 create_eager_task,
36 ensure_durable_operations_allowed,
37 get_current_context,
38 get_durable_context,
39)
40from ..extension import get_extension_context
41from .parallel import _BatchResultSerDes
43if TYPE_CHECKING:
44 from .child import SummaryGenerator
47T = TypeVar("T")
48Params = ParamSpec("Params")
49_BASE_EXCEPTION_GROUP_TYPE = getattr(builtins, "BaseExceptionGroup", None)
52def run_in_child_context(
53 func: Callable[[], Awaitable[T]],
54 *,
55 name: str | None = None,
56 serdes: SerDes | None = None,
57 summary_generator: SummaryGenerator | None = None,
58 is_virtual: bool = False,
59) -> asyncio.Task[T]:
60 """Run an SDK-owned flow scope through the stable operation SPI."""
61 return (
62 get_extension_context()
63 ._reserve_sdk_operation(name) # noqa: SLF001
64 ._run_in_child_context( # noqa: SLF001
65 func,
66 sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT,
67 serdes=serdes,
68 summary_generator=summary_generator,
69 is_virtual=is_virtual,
70 )
71 )
74class FlowDefinitionError(ValidationError):
75 """Raised when a declarative flow definition is invalid."""
78class FlowExecutionError(DurableExecutionsError):
79 """Raised after a flow checkpoints a result with unhandled node failures."""
81 def __init__(self, message: str, result: Any) -> None:
82 super().__init__(message)
83 self.result = result
86class FlowNodeStatus(Enum):
87 """Logical status of a node in a completed flow."""
89 SUCCEEDED = "SUCCEEDED"
90 FAILED = "FAILED"
91 SKIPPED = "SKIPPED"
94@dataclass(frozen=True)
95class FlowNodeResult(MappingModel, Generic[T]):
96 """Logical result of one flow node."""
98 status: FlowNodeStatus
99 outcome: T | None = field(default=None, metadata={"omit_if_none": False})
100 error: ErrorObject | None = field(default=None, metadata={"omit_if_none": False})
102 @classmethod
103 def succeeded(cls, outcome: T) -> FlowNodeResult[T]:
104 return cls(status=FlowNodeStatus.SUCCEEDED, outcome=outcome)
106 @classmethod
107 def failed(cls, error: ErrorObject) -> FlowNodeResult[T]:
108 return cls(status=FlowNodeStatus.FAILED, error=error)
110 @classmethod
111 def skipped(cls) -> FlowNodeResult[T]:
112 return cls(status=FlowNodeStatus.SKIPPED)
115@dataclass(frozen=True)
116class FlowResult:
117 """Complete logical result of a flow."""
119 results: dict[str, FlowNodeResult[Any]]
120 outputs: tuple[Any, ...] = ()
121 unhandled_failures: tuple[str, ...] = ()
122 unavailable_outputs: tuple[str, ...] = ()
123 _output_kinds: tuple[_FlowNodeInputKind, ...] = field(
124 default=(),
125 repr=False,
126 )
128 def __post_init__(self) -> None:
129 if not self._output_kinds and self.outputs:
130 object.__setattr__(
131 self,
132 "_output_kinds",
133 (_FlowNodeInputKind.RESULT,) * len(self.outputs),
134 )
135 if len(self._output_kinds) != len(self.outputs):
136 msg = "Flow output values and projections must have the same length."
137 raise InvalidStateError(msg)
139 @property
140 def output(self) -> Any:
141 """Return selected output while preserving the definition's arity."""
142 if not self.outputs:
143 return None
144 if len(self.outputs) == 1:
145 return self.outputs[0]
146 return self.outputs
148 @property
149 def has_unhandled_failures(self) -> bool:
150 return bool(self.unhandled_failures)
152 @property
153 def has_unavailable_outputs(self) -> bool:
154 return bool(self.unavailable_outputs)
156 def get_result(self, name: str) -> FlowNodeResult[Any]:
157 """Return a node result by its declared name."""
158 try:
159 return self.results[name]
160 except KeyError:
161 msg = f"Flow has no node named {name!r}."
162 raise KeyError(msg) from None
164 def to_dict(self) -> dict[str, Any]:
165 """Convert the flow result to a serialization-friendly mapping."""
166 return {
167 "results": {
168 name: node_result.to_dict()
169 for name, node_result in self.results.items()
170 },
171 "outputs": [
172 _flow_output_to_dict(output, kind)
173 for output, kind in zip(
174 self.outputs,
175 self._output_kinds,
176 strict=True,
177 )
178 ],
179 "outputProjections": [kind.value for kind in self._output_kinds],
180 "unhandledFailures": list(self.unhandled_failures),
181 "unavailableOutputs": list(self.unavailable_outputs),
182 }
184 @classmethod
185 def from_dict(cls, data: Mapping[str, Any]) -> FlowResult:
186 raw_outputs = tuple(data.get("outputs", ()))
187 raw_kinds = data.get("outputProjections")
188 try:
189 output_kinds = (
190 tuple(_FlowNodeInputKind(str(kind)) for kind in raw_kinds)
191 if raw_kinds is not None
192 else (_FlowNodeInputKind.RESULT,) * len(raw_outputs)
193 )
194 except (TypeError, ValueError) as error:
195 msg = "Serialized flow result contains an invalid output projection."
196 raise SerDesError(msg) from error
197 if len(output_kinds) != len(raw_outputs):
198 msg = (
199 "Serialized flow output values and projections have different lengths."
200 )
201 raise SerDesError(msg)
202 return cls(
203 results={
204 str(name): FlowNodeResult.from_dict(node_result)
205 for name, node_result in data["results"].items()
206 },
207 outputs=tuple(
208 _flow_output_from_dict(output, kind)
209 for output, kind in zip(
210 raw_outputs,
211 output_kinds,
212 strict=True,
213 )
214 ),
215 unhandled_failures=tuple(
216 str(name) for name in data.get("unhandledFailures", ())
217 ),
218 unavailable_outputs=tuple(
219 str(name) for name in data.get("unavailableOutputs", ())
220 ),
221 _output_kinds=output_kinds,
222 )
225class _DependencyCondition(Enum):
226 SUCCEEDED = "SUCCEEDED"
227 FAILED = "FAILED"
228 COMPLETED = "COMPLETED"
230 def matches(self, status: FlowNodeStatus) -> bool:
231 if self is _DependencyCondition.COMPLETED:
232 return True
233 if self is _DependencyCondition.SUCCEEDED:
234 return status is FlowNodeStatus.SUCCEEDED
235 return status is FlowNodeStatus.FAILED
238class _DependencyMode(Enum):
239 ALL = "ALL"
240 ANY = "ANY"
243class _FlowNodeInputKind(Enum):
244 OUTCOME = "OUTCOME"
245 ERROR = "ERROR"
246 RESULT = "RESULT"
249def _flow_output_to_dict(value: Any, kind: _FlowNodeInputKind) -> Any:
250 if kind is _FlowNodeInputKind.ERROR:
251 return value.to_dict() if value is not None else None
252 if kind is _FlowNodeInputKind.RESULT:
253 return cast("FlowNodeResult[Any]", value).to_dict()
254 return value
257def _flow_output_from_dict(value: Any, kind: _FlowNodeInputKind) -> Any:
258 if kind is _FlowNodeInputKind.ERROR:
259 return ErrorObject.from_dict(value) if value is not None else None
260 if kind is _FlowNodeInputKind.RESULT:
261 return FlowNodeResult.from_dict(value)
262 return value
265def _flow_node_result_to_checkpoint_dict(
266 value: FlowNodeResult[Any],
267) -> dict[str, Any]:
268 """Preserve outcomes for type-aware encoding by ExtendedTypeSerDes."""
269 return {
270 "status": value.status.value,
271 "outcome": value.outcome,
272 "error": value.error.to_dict() if value.error is not None else None,
273 }
276def _flow_result_to_checkpoint_dict(value: FlowResult) -> dict[str, Any]:
277 outputs = [
278 (
279 _flow_node_result_to_checkpoint_dict(cast("FlowNodeResult[Any]", output))
280 if kind is _FlowNodeInputKind.RESULT
281 else _flow_output_to_dict(output, kind)
282 )
283 for output, kind in zip(
284 value.outputs,
285 value._output_kinds,
286 strict=True,
287 )
288 ]
289 return {
290 "results": {
291 name: _flow_node_result_to_checkpoint_dict(node_result)
292 for name, node_result in value.results.items()
293 },
294 "outputs": outputs,
295 "outputProjections": [kind.value for kind in value._output_kinds],
296 "unhandledFailures": list(value.unhandled_failures),
297 "unavailableOutputs": list(value.unavailable_outputs),
298 }
301_FLOW_VALUE_VERSION_KEY = "__async_durable_execution_flow_value__"
302_FLOW_VALUE_VERSION = 1
303_FLOW_VALUE_KIND_KEY = "kind"
304_FLOW_VALUE_PAYLOAD_KEY = "value"
307class _FlowValueKind(Enum):
308 ESCAPED_DICT = "ESCAPED_DICT"
309 ERROR_OBJECT = "ERROR_OBJECT"
310 FLOW_NODE_RESULT = "FLOW_NODE_RESULT"
311 FLOW_RESULT = "FLOW_RESULT"
314def _encode_flow_value(
315 value: Any,
316 *,
317 active_objects: set[int] | None = None,
318) -> Any:
319 recursive = isinstance(
320 value,
321 (FlowResult, FlowNodeResult, ErrorObject, list, tuple, dict),
322 )
323 if active_objects is None:
324 active_objects = set()
325 object_id = id(value)
326 if recursive and object_id in active_objects:
327 msg = "Circular references are not supported in flow values."
328 raise SerDesError(msg)
329 if recursive:
330 active_objects.add(object_id)
332 try:
333 if isinstance(value, FlowResult):
334 kind = _FlowValueKind.FLOW_RESULT
335 payload = _encode_flow_value(
336 _flow_result_to_checkpoint_dict(value),
337 active_objects=active_objects,
338 )
339 elif isinstance(value, FlowNodeResult):
340 kind = _FlowValueKind.FLOW_NODE_RESULT
341 payload = _encode_flow_value(
342 _flow_node_result_to_checkpoint_dict(value),
343 active_objects=active_objects,
344 )
345 elif isinstance(value, ErrorObject):
346 kind = _FlowValueKind.ERROR_OBJECT
347 payload = _encode_flow_value(
348 value.to_dict(),
349 active_objects=active_objects,
350 )
351 elif isinstance(value, list):
352 return [
353 _encode_flow_value(item, active_objects=active_objects)
354 for item in value
355 ]
356 elif isinstance(value, tuple):
357 return tuple(
358 _encode_flow_value(item, active_objects=active_objects)
359 for item in value
360 )
361 elif isinstance(value, dict):
362 payload = {
363 key: _encode_flow_value(item, active_objects=active_objects)
364 for key, item in value.items()
365 }
366 if _FLOW_VALUE_VERSION_KEY not in value:
367 return payload
368 kind = _FlowValueKind.ESCAPED_DICT
369 else:
370 return value
371 return {
372 _FLOW_VALUE_VERSION_KEY: _FLOW_VALUE_VERSION,
373 _FLOW_VALUE_KIND_KEY: kind.value,
374 _FLOW_VALUE_PAYLOAD_KEY: payload,
375 }
376 finally:
377 if recursive:
378 active_objects.remove(object_id)
381def _decode_flow_value(value: Any) -> Any:
382 if isinstance(value, list):
383 return [_decode_flow_value(item) for item in value]
384 if isinstance(value, tuple):
385 return tuple(_decode_flow_value(item) for item in value)
386 if not isinstance(value, Mapping):
387 return value
388 if _FLOW_VALUE_VERSION_KEY not in value:
389 return {key: _decode_flow_value(item) for key, item in value.items()}
390 if (
391 type(value.get(_FLOW_VALUE_VERSION_KEY)) is not int
392 or value[_FLOW_VALUE_VERSION_KEY] != _FLOW_VALUE_VERSION
393 ):
394 msg = "Serialized flow value has an invalid envelope."
395 raise SerDesError(msg)
396 try:
397 kind = _FlowValueKind(value[_FLOW_VALUE_KIND_KEY])
398 payload = value[_FLOW_VALUE_PAYLOAD_KEY]
399 except (KeyError, TypeError, ValueError) as error:
400 msg = "Serialized flow value has an invalid kind or payload."
401 raise SerDesError(msg) from error
403 if kind is _FlowValueKind.ESCAPED_DICT:
404 if not isinstance(payload, Mapping):
405 msg = "Serialized escaped dict flow value must be a mapping."
406 raise SerDesError(msg)
407 return {key: _decode_flow_value(item) for key, item in payload.items()}
409 decoded = _decode_flow_value(payload)
410 if not isinstance(decoded, Mapping):
411 msg = f"Serialized {kind.value.lower()} flow value must contain a mapping."
412 raise SerDesError(msg)
413 if kind is _FlowValueKind.ERROR_OBJECT:
414 return ErrorObject.from_dict(decoded)
415 if kind is _FlowValueKind.FLOW_NODE_RESULT:
416 return FlowNodeResult.from_dict(decoded)
417 return FlowResult.from_dict(decoded)
420class _EvaluationStatus(Enum):
421 PENDING = "PENDING"
422 MATCHED = "MATCHED"
423 UNMATCHED = "UNMATCHED"
426@dataclass(frozen=True)
427class _Evaluation:
428 status: _EvaluationStatus
429 handled_failures: tuple[FlowNode[Any], ...] = ()
432class _DependencyExpression:
433 """Internal immutable dependency expression."""
435 builder: _FlowBuilder
437 def leaves(self) -> tuple[_DependencyLeaf, ...]:
438 raise NotImplementedError
440 def evaluate(
441 self,
442 results: Mapping[FlowNode[Any], FlowNodeResult[Any]],
443 any_winners: dict[int, int] | None = None,
444 ) -> _Evaluation:
445 raise NotImplementedError
447 def __and__(
448 self, other: FlowNode[Any] | _DependencyExpression
449 ) -> _DependencyExpression:
450 return self._combine(other, _DependencyMode.ALL)
452 def __or__(
453 self, other: FlowNode[Any] | _DependencyExpression
454 ) -> _DependencyExpression:
455 return self._combine(other, _DependencyMode.ANY)
457 def _combine(
458 self,
459 other: FlowNode[Any] | _DependencyExpression,
460 mode: _DependencyMode,
461 ) -> _DependencyExpression:
462 _require_active_builder(self.builder)
463 other_expression = _coerce_expression(other)
464 if other_expression.builder is not self.builder:
465 msg = "Dependency expressions from different flow definitions cannot be mixed."
466 raise InvalidStateError(msg)
468 children: list[_DependencyExpression] = []
469 for expression in (self, other_expression):
470 if (
471 isinstance(expression, _CompositeDependencyExpression)
472 and expression.mode is mode
473 ):
474 children.extend(expression.children)
475 else:
476 children.append(expression)
477 return _CompositeDependencyExpression(
478 builder=self.builder,
479 mode=mode,
480 children=tuple(children),
481 )
483 def __rshift__(
484 self, target: FlowNode[Any] | tuple[FlowNode[Any], ...]
485 ) -> FlowNode[Any] | tuple[FlowNode[Any], ...]:
486 _require_active_builder(self.builder)
487 targets = target if isinstance(target, tuple) else (target,)
488 if not targets:
489 msg = "A dependency expression must target at least one node."
490 raise FlowDefinitionError(msg)
492 for flow_node in targets:
493 if not isinstance(flow_node, FlowNode):
494 msg = "Dependency targets must be FlowNode instances."
495 raise FlowDefinitionError(msg)
496 if flow_node._builder is not self.builder:
497 msg = "Nodes from different flow definitions cannot be mixed."
498 raise InvalidStateError(msg)
499 self.builder.add_dependency(flow_node, self)
500 return target
503@dataclass(frozen=True)
504class _FlowNodeInput(Generic[T]):
505 """Deferred node projection used while a flow definition is evaluated."""
507 node: FlowNode[Any]
508 kind: _FlowNodeInputKind
510 @property
511 def condition(self) -> _DependencyCondition:
512 if self.kind is _FlowNodeInputKind.OUTCOME:
513 return _DependencyCondition.SUCCEEDED
514 if self.kind is _FlowNodeInputKind.ERROR:
515 return _DependencyCondition.FAILED
516 return _DependencyCondition.COMPLETED
518 def resolve(
519 self,
520 results: Mapping[FlowNode[Any], FlowNodeResult[Any]],
521 ) -> T:
522 result = results.get(self.node)
523 if result is None:
524 msg = f"Result for required input {self.node.name!r} is not available."
525 raise InvalidStateError(msg)
526 if not self.condition.matches(result.status):
527 msg = (
528 f"Result for required input {self.node.name!r} has status "
529 f"{result.status.value}, expected {self.condition.value}."
530 )
531 raise InvalidStateError(msg)
532 if self.kind is _FlowNodeInputKind.OUTCOME:
533 return cast("T", result.outcome)
534 if self.kind is _FlowNodeInputKind.ERROR and result.error is None:
535 msg = f"Failed input {self.node.name!r} has no error."
536 raise InvalidStateError(msg)
537 if self.kind is _FlowNodeInputKind.ERROR:
538 return cast("T", result.error)
539 return cast("T", result)
541 def project(self, result: FlowNodeResult[Any]) -> T:
542 """Project a settled node result without imposing an input condition."""
543 if self.kind is _FlowNodeInputKind.OUTCOME:
544 return cast("T", result.outcome)
545 if self.kind is _FlowNodeInputKind.ERROR:
546 return cast("T", result.error)
547 return cast("T", result)
550@dataclass(frozen=True)
551class _DependencyLeaf(_DependencyExpression):
552 builder: _FlowBuilder
553 node: FlowNode[Any]
554 condition: _DependencyCondition
556 def leaves(self) -> tuple[_DependencyLeaf, ...]:
557 return (self,)
559 def evaluate(
560 self,
561 results: Mapping[FlowNode[Any], FlowNodeResult[Any]],
562 any_winners: dict[int, int] | None = None,
563 ) -> _Evaluation:
564 node_result = results.get(self.node)
565 if node_result is None:
566 return _Evaluation(_EvaluationStatus.PENDING)
567 if not self.condition.matches(node_result.status):
568 return _Evaluation(_EvaluationStatus.UNMATCHED)
570 handled = (
571 (self.node,)
572 if self.condition is _DependencyCondition.FAILED
573 and node_result.status is FlowNodeStatus.FAILED
574 else ()
575 )
576 return _Evaluation(_EvaluationStatus.MATCHED, handled)
579@dataclass(frozen=True)
580class _CompositeDependencyExpression(_DependencyExpression):
581 builder: _FlowBuilder
582 mode: _DependencyMode
583 children: tuple[_DependencyExpression, ...]
585 def leaves(self) -> tuple[_DependencyLeaf, ...]:
586 return tuple(leaf for child in self.children for leaf in child.leaves())
588 def evaluate(
589 self,
590 results: Mapping[FlowNode[Any], FlowNodeResult[Any]],
591 any_winners: dict[int, int] | None = None,
592 ) -> _Evaluation:
593 if any_winners is None:
594 any_winners = {}
595 if self.mode is _DependencyMode.ANY and id(self) in any_winners:
596 winner = any_winners[id(self)]
597 return self.children[winner].evaluate(results, any_winners)
599 evaluations = [child.evaluate(results, any_winners) for child in self.children]
600 if self.mode is _DependencyMode.ALL:
601 if any(
602 evaluation.status is _EvaluationStatus.PENDING
603 for evaluation in evaluations
604 ):
605 return _Evaluation(_EvaluationStatus.PENDING)
606 if any(
607 evaluation.status is _EvaluationStatus.UNMATCHED
608 for evaluation in evaluations
609 ):
610 return _Evaluation(_EvaluationStatus.UNMATCHED)
611 return _Evaluation(
612 _EvaluationStatus.MATCHED,
613 tuple(
614 node
615 for evaluation in evaluations
616 for node in evaluation.handled_failures
617 ),
618 )
620 for index, evaluation in enumerate(evaluations):
621 if evaluation.status is _EvaluationStatus.MATCHED:
622 any_winners[id(self)] = index
623 return evaluation
624 if any(
625 evaluation.status is _EvaluationStatus.PENDING for evaluation in evaluations
626 ):
627 return _Evaluation(_EvaluationStatus.PENDING)
628 return _Evaluation(_EvaluationStatus.UNMATCHED)
631class FlowNode(Generic[T]):
632 """Typed handle for a node declared inside a durable DAG definition."""
634 def __init__(
635 self,
636 builder: _FlowBuilder,
637 index: int,
638 func: Callable[[], Awaitable[T]],
639 name: str,
640 ) -> None:
641 self._builder = builder
642 self._index = index
643 self._func = func
644 self.name = name
645 self._dependency: _DependencyExpression | None = None
647 def __repr__(self) -> str:
648 return f"FlowNode(name={self.name!r})"
650 @property
651 def result(self) -> FlowNodeResult[T]:
652 """Reference a completed output or return its result in a running node."""
653 builder = _current_flow_builder.get()
654 if builder is not None and not builder.frozen:
655 return cast(
656 "FlowNodeResult[T]",
657 _FlowNodeInput[FlowNodeResult[T]](
658 self,
659 _FlowNodeInputKind.RESULT,
660 ),
661 )
662 try:
663 context = get_current_context()
664 except RuntimeError as error:
665 msg = "Flow node results are only available while a flow node is executing."
666 raise InvalidStateError(msg) from error
667 if not isinstance(context, FlowNodeContext):
668 msg = "Flow node results are only available while a flow node is executing."
669 raise InvalidStateError(msg)
670 return context.result(self)
672 @property
673 def status(self) -> FlowNodeStatus:
674 """Return this direct dependency's logical status."""
675 return self.result.status
677 @property
678 def outcome(self) -> T:
679 """Reference a required successful input or return its resolved outcome."""
680 builder = _current_flow_builder.get()
681 if builder is not None and not builder.frozen:
682 return cast(
683 "T",
684 _FlowNodeInput[Any](self, _FlowNodeInputKind.OUTCOME),
685 )
686 result = self.result
687 if result.status is not FlowNodeStatus.SUCCEEDED:
688 msg = (
689 f"Flow node {self.name!r} did not succeed "
690 f"(status {result.status.value}); inspect status, error, or result."
691 )
692 raise InvalidStateError(msg)
693 return cast("T", result.outcome)
695 @property
696 def error(self) -> ErrorObject | None:
697 """Reference a required failed input or return its captured error."""
698 builder = _current_flow_builder.get()
699 if builder is not None and not builder.frozen:
700 return cast(
701 "ErrorObject",
702 _FlowNodeInput[ErrorObject](self, _FlowNodeInputKind.ERROR),
703 )
704 return self.result.error
706 @property
707 def succeeded(self) -> _DependencyExpression:
708 return _DependencyLeaf(self._builder, self, _DependencyCondition.SUCCEEDED)
710 @property
711 def failed(self) -> _DependencyExpression:
712 return _DependencyLeaf(self._builder, self, _DependencyCondition.FAILED)
714 @property
715 def completed(self) -> _DependencyExpression:
716 return _DependencyLeaf(self._builder, self, _DependencyCondition.COMPLETED)
718 def __and__(
719 self, other: FlowNode[Any] | _DependencyExpression
720 ) -> _DependencyExpression:
721 return self.succeeded & other
723 def __or__(
724 self, other: FlowNode[Any] | _DependencyExpression
725 ) -> _DependencyExpression:
726 return self.succeeded | other
728 def __rshift__(
729 self, target: FlowNode[Any] | tuple[FlowNode[Any], ...]
730 ) -> FlowNode[Any] | tuple[FlowNode[Any], ...]:
731 return self.succeeded >> target
734@dataclass(frozen=True)
735class FlowNodeContext(DurableContext):
736 """Durable context exposed by get_node_context() inside a flow node."""
738 _direct_dependencies: frozenset[FlowNode[Any]] = field(default_factory=frozenset)
739 _dependency_results: Mapping[FlowNode[Any], FlowNodeResult[Any]] = field(
740 default_factory=dict
741 )
743 def result(self, dependency: FlowNode[T]) -> FlowNodeResult[T]:
744 """Return the settled result of a declared direct dependency."""
745 if dependency not in self._direct_dependencies:
746 msg = (
747 f"Node {dependency.name!r} is not a direct dependency of the "
748 "current flow node."
749 )
750 raise InvalidStateError(msg)
751 if dependency not in self._dependency_results:
752 msg = (
753 f"Result for dependency {dependency.name!r} is not available. "
754 "An ANY dependency may start before its other branches settle."
755 )
756 raise InvalidStateError(msg)
757 return cast("FlowNodeResult[T]", self._dependency_results[dependency])
759 @property
760 def dependency_results(self) -> Mapping[str, FlowNodeResult[Any]]:
761 """Return settled direct dependency results captured for this node."""
762 return {
763 dependency.name: result
764 for dependency, result in self._dependency_results.items()
765 }
767 def get_dependency_result(self, name: str) -> FlowNodeResult[Any] | None:
768 """Return an available direct dependency result by stable node name."""
769 dependency = self._dependency_by_name(name)
770 return self._dependency_results.get(dependency)
772 def require_dependency_result(self, name: str) -> FlowNodeResult[Any]:
773 """Return an available direct dependency result or raise."""
774 result = self.get_dependency_result(name)
775 if result is None:
776 msg = (
777 f"Result for dependency {name!r} is not available. "
778 "An ANY dependency may start before its other branches settle."
779 )
780 raise InvalidStateError(msg)
781 return result
783 def _dependency_by_name(self, name: str) -> FlowNode[Any]:
784 for dependency in self._direct_dependencies:
785 if dependency.name == name:
786 return dependency
787 msg = f"Node {name!r} is not a direct dependency of the current flow node."
788 raise InvalidStateError(msg)
791def get_node_context() -> FlowNodeContext:
792 """Return the active `FlowNodeContext`."""
793 current_context = get_current_context()
794 if not isinstance(current_context, FlowNodeContext):
795 msg = "get_node_context() can only be used while a flow node is executing."
796 raise RuntimeError(msg)
797 return current_context
800@dataclass(frozen=True)
801class _FrozenFlow:
802 nodes: tuple[FlowNode[Any], ...]
803 topological_nodes: tuple[FlowNode[Any], ...]
804 execution_nodes: tuple[FlowNode[Any], ...]
805 outputs: tuple[_FlowNodeInput[Any], ...]
808class _FlowBuilder:
809 def __init__(self) -> None:
810 self.nodes: list[FlowNode[Any]] = []
811 self.frozen = False
813 def add_node(
814 self,
815 func: Callable[[], Awaitable[T]],
816 name: str,
817 ) -> FlowNode[T]:
818 if self.frozen:
819 msg = "Cannot add nodes after a flow definition has been frozen."
820 raise InvalidStateError(msg)
821 flow_node = FlowNode(self, len(self.nodes), func, name)
822 self.nodes.append(flow_node)
823 return flow_node
825 def add_dependency(
826 self, target: FlowNode[Any], expression: _DependencyExpression
827 ) -> None:
828 if self.frozen:
829 msg = "Cannot add dependencies after a flow definition has been frozen."
830 raise InvalidStateError(msg)
831 if target._dependency is not None:
832 msg = (
833 f"Node {target.name!r} already has a dependency expression. "
834 "Use an explicit '&' or '|' expression for multiple dependencies."
835 )
836 raise FlowDefinitionError(msg)
837 target._dependency = expression
839 def freeze(self, output: Any) -> _FrozenFlow:
840 self.frozen = True
841 outputs = self._validate_outputs(output)
842 self._validate_nodes()
843 topological_nodes = self._topological_sort()
844 execution_nodes = self._execution_nodes(topological_nodes, outputs)
845 return _FrozenFlow(
846 nodes=tuple(self.nodes),
847 topological_nodes=topological_nodes,
848 execution_nodes=execution_nodes,
849 outputs=outputs,
850 )
852 def _validate_outputs(self, output: Any) -> tuple[_FlowNodeInput[Any], ...]:
853 if output is None:
854 return ()
855 outputs = output if isinstance(output, tuple) else (output,)
856 for output_reference in outputs:
857 if isinstance(output_reference, FlowNode):
858 msg = (
859 "A durable DAG definition cannot return a FlowNode directly; "
860 "return node.outcome, node.error, node.result, or a tuple "
861 "of those projections."
862 )
863 raise FlowDefinitionError(msg)
864 if not isinstance(output_reference, _FlowNodeInput):
865 msg = (
866 "A durable DAG definition must return node.outcome, node.error, "
867 "node.result, a tuple of those projections, or None."
868 )
869 raise FlowDefinitionError(msg)
870 if (
871 output_reference.node._builder is not self
872 or output_reference.node not in self.nodes
873 ):
874 msg = "Flow outputs must reference nodes from the current definition."
875 raise InvalidStateError(msg)
876 return outputs
878 def _validate_nodes(self) -> None:
879 names: set[str] = set()
880 known_nodes = set(self.nodes)
881 for flow_node in self.nodes:
882 if not isinstance(flow_node.name, str) or not flow_node.name.strip():
883 msg = "Flow node names must be non-empty strings."
884 raise FlowDefinitionError(msg)
885 if flow_node.name in names:
886 msg = f"Flow node name {flow_node.name!r} is duplicated."
887 raise FlowDefinitionError(msg)
888 names.add(flow_node.name)
889 if not callable(flow_node._func) or not getattr(
890 flow_node._func, "_durable_node_callable", False
891 ):
892 msg = (
893 f"Flow node {flow_node.name!r} must use a bound callable "
894 "produced by @durable_node."
895 )
896 raise FlowDefinitionError(msg)
898 dependency = flow_node._dependency
899 if dependency is None:
900 continue
901 if dependency.builder is not self:
902 msg = "Dependency expressions must belong to the current definition."
903 raise InvalidStateError(msg)
905 dependencies: set[FlowNode[Any]] = set()
906 for leaf in dependency.leaves():
907 if leaf.builder is not self or leaf.node not in known_nodes:
908 msg = f"Node {flow_node.name!r} references an unknown dependency."
909 raise FlowDefinitionError(msg)
910 if leaf.node is flow_node:
911 msg = f"Flow node {flow_node.name!r} cannot depend on itself."
912 raise FlowDefinitionError(msg)
913 if leaf.node in dependencies:
914 msg = (
915 f"Node {flow_node.name!r} contains duplicate dependency "
916 f"{leaf.node.name!r}."
917 )
918 raise FlowDefinitionError(msg)
919 dependencies.add(leaf.node)
921 def _topological_sort(self) -> tuple[FlowNode[Any], ...]:
922 adjacency: dict[FlowNode[Any], list[FlowNode[Any]]] = {
923 flow_node: [] for flow_node in self.nodes
924 }
925 indegree = {flow_node: 0 for flow_node in self.nodes}
926 for target in self.nodes:
927 if target._dependency is None:
928 continue
929 for leaf in target._dependency.leaves():
930 adjacency[leaf.node].append(target)
931 indegree[target] += 1
933 for targets in adjacency.values():
934 targets.sort(key=lambda flow_node: flow_node._index)
936 ready = [
937 flow_node._index for flow_node in self.nodes if indegree[flow_node] == 0
938 ]
939 heapq.heapify(ready)
940 ordered: list[FlowNode[Any]] = []
941 while ready:
942 index = heapq.heappop(ready)
943 flow_node = self.nodes[index]
944 ordered.append(flow_node)
945 for target in adjacency[flow_node]:
946 indegree[target] -= 1
947 if indegree[target] == 0:
948 heapq.heappush(ready, target._index)
950 if len(ordered) != len(self.nodes):
951 remaining = {
952 flow_node for flow_node in self.nodes if indegree[flow_node] > 0
953 }
954 cycle = self._find_cycle(adjacency, remaining)
955 cycle_path = " -> ".join(flow_node.name for flow_node in cycle)
956 msg = f"Flow contains a cycle: {cycle_path}."
957 raise FlowDefinitionError(msg)
958 return tuple(ordered)
960 def _execution_nodes(
961 self,
962 topological_nodes: tuple[FlowNode[Any], ...],
963 outputs: tuple[_FlowNodeInput[Any], ...],
964 ) -> tuple[FlowNode[Any], ...]:
965 reachable = {output.node for output in outputs}
966 pending = list(reachable)
967 while pending:
968 flow_node = pending.pop()
969 expression = flow_node._dependency
970 if expression is None:
971 continue
972 for leaf in expression.leaves():
973 if leaf.node not in reachable:
974 reachable.add(leaf.node)
975 pending.append(leaf.node)
976 return tuple(
977 flow_node for flow_node in topological_nodes if flow_node in reachable
978 )
980 def _find_cycle(
981 self,
982 adjacency: Mapping[FlowNode[Any], list[FlowNode[Any]]],
983 remaining: set[FlowNode[Any]],
984 ) -> tuple[FlowNode[Any], ...]:
985 state: dict[FlowNode[Any], int] = {}
986 stack: list[FlowNode[Any]] = []
987 stack_positions: dict[FlowNode[Any], int] = {}
989 def visit(flow_node: FlowNode[Any]) -> tuple[FlowNode[Any], ...] | None:
990 state[flow_node] = 1
991 stack_positions[flow_node] = len(stack)
992 stack.append(flow_node)
993 for target in adjacency[flow_node]:
994 if target not in remaining:
995 continue
996 if state.get(target, 0) == 0:
997 cycle = visit(target)
998 if cycle is not None:
999 return cycle
1000 elif state[target] == 1:
1001 start = stack_positions[target]
1002 return tuple((*stack[start:], target))
1003 stack.pop()
1004 stack_positions.pop(flow_node)
1005 state[flow_node] = 2
1006 return None
1008 for flow_node in self.nodes:
1009 if flow_node in remaining and state.get(flow_node, 0) == 0:
1010 cycle = visit(flow_node)
1011 if cycle is not None:
1012 return cycle
1014 msg = "Flow cycle detection failed to identify a concrete cycle."
1015 raise FlowDefinitionError(msg)
1018_current_flow_builder: ContextVar[_FlowBuilder | None] = ContextVar(
1019 "async_durable_execution.current_flow_builder",
1020 default=None,
1021)
1024def _require_active_builder(builder: _FlowBuilder) -> None:
1025 if _current_flow_builder.get() is not builder or builder.frozen:
1026 msg = "Flow nodes and dependency expressions may only be used in their definition."
1027 raise InvalidStateError(msg)
1030def _coerce_expression(
1031 value: FlowNode[Any] | _DependencyExpression,
1032) -> _DependencyExpression:
1033 if isinstance(value, FlowNode):
1034 return value.succeeded
1035 if isinstance(value, _DependencyExpression):
1036 return value
1037 msg = "Dependencies must be FlowNode handles or dependency expressions."
1038 raise FlowDefinitionError(msg)
1041def _flow_node_inputs(
1042 value: Any,
1043 *,
1044 active_containers: set[int] | None = None,
1045) -> tuple[_FlowNodeInput[Any], ...]:
1046 if isinstance(value, _FlowNodeInput):
1047 return (value,)
1048 if not isinstance(value, (list, tuple, dict)):
1049 hidden_references = _unsupported_flow_node_inputs(value)
1050 if hidden_references:
1051 container_type = type(value).__qualname__
1052 msg = (
1053 "Flow node projections nested in unsupported container type "
1054 f"{container_type!r} cannot be resolved. Use a list, tuple, or dict."
1055 )
1056 raise FlowDefinitionError(msg)
1057 return ()
1059 if active_containers is None:
1060 active_containers = set()
1061 container_id = id(value)
1062 if container_id in active_containers:
1063 msg = "Flow node arguments cannot contain recursive containers."
1064 raise FlowDefinitionError(msg)
1066 active_containers.add(container_id)
1067 try:
1068 values = (
1069 (*value.keys(), *value.values())
1070 if isinstance(value, dict)
1071 else tuple(value)
1072 )
1073 return tuple(
1074 reference
1075 for item in values
1076 for reference in _flow_node_inputs(
1077 item,
1078 active_containers=active_containers,
1079 )
1080 )
1081 finally:
1082 active_containers.remove(container_id)
1085def _unsupported_flow_node_inputs(
1086 value: Any,
1087 *,
1088 active_objects: set[int] | None = None,
1089) -> tuple[_FlowNodeInput[Any], ...]:
1090 if isinstance(value, _FlowNodeInput):
1091 return (value,)
1092 if isinstance(
1093 value,
1094 (
1095 str,
1096 bytes,
1097 bytearray,
1098 memoryview,
1099 range,
1100 FlowNode,
1101 _DependencyExpression,
1102 ),
1103 ) or callable(value):
1104 return ()
1105 if isinstance(value, Iterator):
1106 msg = (
1107 "Flow node arguments cannot use iterators because they cannot be "
1108 "inspected safely. Materialize the iterator as a list or tuple."
1109 )
1110 raise FlowDefinitionError(msg)
1112 if active_objects is None:
1113 active_objects = set()
1114 object_id = id(value)
1115 if object_id in active_objects:
1116 return ()
1118 active_objects.add(object_id)
1119 try:
1120 values: tuple[Any, ...] | None = None
1121 if isinstance(value, Mapping):
1122 values = (*value.keys(), *value.values())
1123 elif isinstance(value, (list, tuple)):
1124 values = tuple(value)
1125 elif isinstance(value, Set):
1126 values = tuple(value)
1127 elif is_dataclass(value) and not isinstance(value, type):
1128 values = tuple(
1129 getattr(value, dataclass_field.name)
1130 for dataclass_field in dataclass_fields(value)
1131 )
1132 else:
1133 attrs_fields = getattr(type(value), "__attrs_attrs__", None)
1134 if attrs_fields is not None:
1135 values = tuple(
1136 getattr(value, attribute.name) for attribute in attrs_fields
1137 )
1138 elif isinstance(value, Iterable):
1139 container_type = type(value).__qualname__
1140 msg = (
1141 "Flow node arguments cannot use unsupported container type "
1142 f"{container_type!r}. Use a list, tuple, or dict."
1143 )
1144 raise FlowDefinitionError(msg)
1145 elif hasattr(value, "__dict__"):
1146 values = tuple(vars(value).values())
1147 else:
1148 slots = getattr(type(value), "__slots__", ())
1149 if isinstance(slots, str):
1150 slots = (slots,)
1151 slot_values = tuple(
1152 getattr(value, slot)
1153 for slot in slots
1154 if slot not in {"__dict__", "__weakref__"} and hasattr(value, slot)
1155 )
1156 if slot_values:
1157 values = slot_values
1159 if values is None:
1160 return ()
1161 return tuple(
1162 reference
1163 for item in values
1164 for reference in _unsupported_flow_node_inputs(
1165 item,
1166 active_objects=active_objects,
1167 )
1168 )
1169 finally:
1170 active_objects.remove(object_id)
1173def _bound_flow_node_inputs(
1174 func: Callable[[], Awaitable[Any]],
1175) -> tuple[_FlowNodeInput[Any], ...]:
1176 args = cast("tuple[Any, ...]", getattr(func, "_durable_node_args"))
1177 kwargs = cast("Mapping[str, Any]", getattr(func, "_durable_node_kwargs"))
1178 return tuple(
1179 reference
1180 for value in (*args, *kwargs.values())
1181 for reference in _flow_node_inputs(value)
1182 )
1185def _input_dependency_expression(
1186 builder: _FlowBuilder,
1187 func: Callable[[], Awaitable[Any]],
1188) -> _DependencyExpression | None:
1189 references: dict[FlowNode[Any], _FlowNodeInput[Any]] = {}
1190 for reference in _bound_flow_node_inputs(func):
1191 if reference.node._builder is not builder:
1192 msg = "Flow node inputs must come from the current flow definition."
1193 raise InvalidStateError(msg)
1194 existing = references.get(reference.node)
1195 if existing is not None and existing.kind is not reference.kind:
1196 msg = (
1197 f"Flow node input {reference.node.name!r} cannot require multiple "
1198 "projections."
1199 )
1200 raise FlowDefinitionError(msg)
1201 references[reference.node] = reference
1203 leaves = tuple(
1204 _DependencyLeaf(builder, reference.node, reference.condition)
1205 for reference in sorted(
1206 references.values(),
1207 key=lambda item: item.node._index,
1208 )
1209 )
1210 if not leaves:
1211 return None
1212 if len(leaves) == 1:
1213 return leaves[0]
1214 return _CompositeDependencyExpression(
1215 builder=builder,
1216 mode=_DependencyMode.ALL,
1217 children=leaves,
1218 )
1221def node(
1222 func: Callable[[], Awaitable[T]],
1223 *,
1224 name: str | None = None,
1225 dependency: FlowNode[Any] | _DependencyExpression | None = None,
1226) -> FlowNode[T]:
1227 """Declare a node and derive required dependencies from its bound inputs."""
1228 builder = _current_flow_builder.get()
1229 if builder is None or builder.frozen:
1230 msg = "node() can only be used while a @durable_dag definition is evaluating."
1231 raise InvalidStateError(msg)
1232 required_metadata = (
1233 "_durable_node_function",
1234 "_durable_node_args",
1235 "_durable_node_kwargs",
1236 )
1237 if (
1238 not callable(func)
1239 or not getattr(func, "_durable_node_callable", False)
1240 or any(not hasattr(func, attribute) for attribute in required_metadata)
1241 ):
1242 msg = "node() requires a bound callable produced by @durable_node."
1243 raise FlowDefinitionError(msg)
1245 node_name = name if name is not None else getattr(func, "__name__", None)
1246 input_expression = _input_dependency_expression(builder, func)
1247 explicit_expression = (
1248 _coerce_expression(dependency) if dependency is not None else None
1249 )
1251 expression: _DependencyExpression | None
1252 if input_expression is not None and explicit_expression is not None:
1253 input_nodes = {leaf.node for leaf in input_expression.leaves()}
1254 duplicated = next(
1255 (
1256 leaf.node
1257 for leaf in explicit_expression.leaves()
1258 if leaf.node in input_nodes
1259 ),
1260 None,
1261 )
1262 if duplicated is not None:
1263 msg = (
1264 f"Node {node_name!r} declares {duplicated.name!r} as both "
1265 "a required input and an explicit dependency."
1266 )
1267 raise FlowDefinitionError(msg)
1268 expression = input_expression & explicit_expression
1269 else:
1270 expression = input_expression or explicit_expression
1272 flow_node = builder.add_node(func, cast("str", node_name))
1273 if expression is not None:
1274 builder.add_dependency(flow_node, expression)
1275 return flow_node
1278def durable_node(
1279 func: Callable[Params, Awaitable[T]],
1280) -> Callable[Params, Callable[[], Awaitable[T]]]:
1281 """Bind arguments to an async function used as a durable flow node."""
1282 if isinstance(func, classmethod):
1283 return classmethod(durable_node(func.__func__))
1284 if isinstance(func, staticmethod):
1285 return staticmethod(durable_node(func.__func__))
1286 if not inspect.iscoroutinefunction(func):
1287 msg = "@durable_node can only decorate an async node function."
1288 raise FlowDefinitionError(msg)
1290 @functools.wraps(func)
1291 def wrapper(
1292 *args: Params.args, **kwargs: Params.kwargs
1293 ) -> Callable[[], Awaitable[T]]:
1294 inspect.signature(func).bind(*args, **kwargs)
1295 bound = functools.partial(func, *args, **kwargs)
1296 setattr(bound, "__name__", func.__name__)
1297 setattr(bound, "_durable_node_callable", True)
1298 setattr(bound, "_durable_node_function", func)
1299 setattr(bound, "_durable_node_args", args)
1300 setattr(bound, "_durable_node_kwargs", kwargs)
1301 return bound
1303 setattr(wrapper, "_durable_node", True)
1304 return wrapper
1307def durable_dag(
1308 func: Callable[Params, Any],
1309) -> Callable[Params, Callable[[], Any]]:
1310 """Bind arguments to a synchronous declarative flow definition."""
1311 if isinstance(func, classmethod):
1312 return classmethod(durable_dag(func.__func__))
1313 if isinstance(func, staticmethod):
1314 return staticmethod(durable_dag(func.__func__))
1315 if inspect.iscoroutinefunction(func):
1316 msg = "@durable_dag can only decorate a synchronous definition function."
1317 raise FlowDefinitionError(msg)
1319 @functools.wraps(func)
1320 def wrapper(*args: Params.args, **kwargs: Params.kwargs) -> Callable[[], Any]:
1321 bound = functools.partial(func, *args, **kwargs)
1322 setattr(bound, "__name__", func.__name__)
1323 setattr(bound, "_durable_dag_definition", True)
1324 return bound
1326 setattr(wrapper, "_durable_dag", True)
1327 return wrapper
1330@dataclass(frozen=True)
1331class _NodeExecution:
1332 result: FlowNodeResult[Any]
1333 handled_failures: tuple[str, ...] = ()
1335 def to_dict(self) -> dict[str, Any]:
1336 return {
1337 "result": _flow_node_result_to_checkpoint_dict(self.result),
1338 "handledFailures": list(self.handled_failures),
1339 }
1341 @classmethod
1342 def from_dict(cls, data: Mapping[str, Any]) -> _NodeExecution:
1343 return cls(
1344 result=FlowNodeResult.from_dict(data["result"]),
1345 handled_failures=tuple(
1346 str(name) for name in data.get("handledFailures", ())
1347 ),
1348 )
1351class _FlowValueSerDes(SerDes[Any]):
1352 def __init__(self) -> None:
1353 self.delegate: SerDes[Any] = _BatchResultSerDes()
1355 async def serialize(self, value: Any) -> str:
1356 return await self.delegate.serialize(_encode_flow_value(value))
1358 async def deserialize(self, data: str) -> Any:
1359 return _decode_flow_value(await self.delegate.deserialize(data))
1362class _NodeExecutionSerDes(SerDes[_NodeExecution]):
1363 def __init__(self) -> None:
1364 self.delegate = _FlowValueSerDes()
1366 async def serialize(self, value: _NodeExecution) -> str:
1367 return await self.delegate.serialize(value.to_dict())
1369 async def deserialize(self, data: str) -> _NodeExecution:
1370 decoded = await self.delegate.deserialize(data)
1371 if not isinstance(decoded, Mapping):
1372 msg = "Serialized flow node result must be a mapping."
1373 raise SerDesError(msg)
1374 return _NodeExecution.from_dict(decoded)
1377@dataclass(frozen=True)
1378class _PersistedDependencyResolution:
1379 matched: bool
1380 selected_nodes: tuple[str, ...]
1381 handled_failures: tuple[str, ...] = ()
1383 def to_dict(self) -> dict[str, Any]:
1384 return {
1385 "matched": self.matched,
1386 "selectedNodes": list(self.selected_nodes),
1387 "handledFailures": list(self.handled_failures),
1388 }
1390 @classmethod
1391 def from_dict(cls, data: Mapping[str, Any]) -> _PersistedDependencyResolution:
1392 return cls(
1393 matched=bool(data["matched"]),
1394 selected_nodes=tuple(str(name) for name in data["selectedNodes"]),
1395 handled_failures=tuple(
1396 str(name) for name in data.get("handledFailures", ())
1397 ),
1398 )
1401class _PersistedDependencyResolutionSerDes(SerDes[_PersistedDependencyResolution]):
1402 def __init__(self) -> None:
1403 self.delegate: ExtendedTypeSerDes[Any] = ExtendedTypeSerDes()
1405 async def serialize(self, value: _PersistedDependencyResolution) -> str:
1406 return await self.delegate.serialize(value.to_dict())
1408 async def deserialize(self, data: str) -> _PersistedDependencyResolution:
1409 decoded = await self.delegate.deserialize(data)
1410 if not isinstance(decoded, Mapping):
1411 msg = "Serialized flow dependency resolution must be a mapping."
1412 raise SerDesError(msg)
1413 return _PersistedDependencyResolution.from_dict(decoded)
1416class _FlowResultSerDes(SerDes[FlowResult]):
1417 def __init__(self) -> None:
1418 self.delegate = _FlowValueSerDes()
1420 async def serialize(self, value: FlowResult) -> str:
1421 return await self.delegate.serialize(_flow_result_to_checkpoint_dict(value))
1423 async def deserialize(self, data: str) -> FlowResult:
1424 decoded = await self.delegate.deserialize(data)
1425 if not isinstance(decoded, Mapping):
1426 msg = "Serialized flow result must be a mapping."
1427 raise SerDesError(msg)
1428 return FlowResult.from_dict(decoded)
1431_NODE_EXECUTION_SERDES = _NodeExecutionSerDes()
1432_DEPENDENCY_RESOLUTION_SERDES = _PersistedDependencyResolutionSerDes()
1433_FLOW_RESULT_SERDES = _FlowResultSerDes()
1434_FLOW_VALUE_SERDES = _FlowValueSerDes()
1437async def _clone_flow_value(value: Any) -> Any:
1438 """Clone a value using the same representation as flow checkpoints."""
1439 return await _FLOW_VALUE_SERDES.deserialize(
1440 await _FLOW_VALUE_SERDES.serialize(value)
1441 )
1444async def _clone_dependency_results(
1445 results: Mapping[FlowNode[Any], FlowNodeResult[Any]],
1446) -> dict[FlowNode[Any], FlowNodeResult[Any]]:
1447 """Clone settled results once for one consumer using checkpoint semantics."""
1448 cloned: dict[FlowNode[Any], FlowNodeResult[Any]] = {}
1449 for flow_node, result in results.items():
1450 value = await _clone_flow_value(result)
1451 if not isinstance(value, FlowNodeResult):
1452 msg = "Cloned flow dependency result has an invalid type."
1453 raise SerDesError(msg)
1454 cloned[flow_node] = value
1455 return cloned
1458class _FlowControlSignal(BaseException):
1459 """Carry SDK control failures through child executors without checkpointing."""
1461 def __init__(self, error: Exception) -> None:
1462 super().__init__(str(error))
1463 self.error = error
1466@dataclass(frozen=True)
1467class _DependencyResolution:
1468 matched: bool
1469 results: Mapping[FlowNode[Any], FlowNodeResult[Any]]
1470 handled_failures: tuple[FlowNode[Any], ...] = ()
1473def _callable_error_object(error: Exception) -> ErrorObject:
1474 if isinstance(error, CallableRuntimeError):
1475 return ErrorObject(
1476 message=error.message,
1477 type=error.error_type,
1478 data=error.data,
1479 stack_trace=error.stack_trace,
1480 )
1481 return ErrorObject.from_exception(error)
1484def _find_control_error(error: Exception) -> Exception | None:
1485 pending: list[BaseException] = [error]
1486 seen: set[int] = set()
1487 while pending:
1488 current = pending.pop()
1489 if not isinstance(current, Exception) or id(current) in seen:
1490 continue
1491 seen.add(id(current))
1492 if isinstance(current, (ExecutionError, InvocationError, SerDesError)):
1493 return current
1494 if isinstance(current, CallableRuntimeError):
1495 error_type = current.error_type or ""
1496 message = current.message or str(current)
1497 control_error = _restore_sdk_control_error(
1498 message,
1499 current.error_type,
1500 current.data,
1501 )
1502 if control_error is not None:
1503 return control_error
1504 related: list[BaseException] = []
1505 if _BASE_EXCEPTION_GROUP_TYPE is not None and isinstance(
1506 current,
1507 _BASE_EXCEPTION_GROUP_TYPE,
1508 ):
1509 related.extend(
1510 nested
1511 for nested in getattr(current, "exceptions", ())
1512 if isinstance(nested, BaseException)
1513 )
1514 cause = current.__cause__ or current.__context__
1515 if cause is not None:
1516 related.append(cause)
1517 pending.extend(reversed(related))
1518 return None
1521async def _await_node_execution(
1522 task: asyncio.Task[_NodeExecution],
1523) -> _NodeExecution:
1524 try:
1525 return await asyncio.shield(task)
1526 except _FlowControlSignal:
1527 raise
1528 except Exception as error:
1529 control_error = _find_control_error(error)
1530 if control_error is not None:
1531 raise _FlowControlSignal(control_error) from error
1532 raise
1535async def _await_resolver_resolution(
1536 task: asyncio.Task[_PersistedDependencyResolution],
1537 expression: _DependencyExpression,
1538 tasks: Mapping[FlowNode[Any], asyncio.Task[_NodeExecution]],
1539) -> _DependencyResolution:
1540 try:
1541 persisted = await asyncio.shield(task)
1542 except _FlowControlSignal:
1543 raise
1544 except Exception as error:
1545 control_error = _find_control_error(error)
1546 if control_error is not None:
1547 raise _FlowControlSignal(control_error) from error
1548 raise
1550 nodes_by_name = {leaf.node.name: leaf.node for leaf in expression.leaves()}
1551 selected: dict[FlowNode[Any], FlowNodeResult[Any]] = {}
1552 for node_name in persisted.selected_nodes:
1553 flow_node = nodes_by_name[node_name]
1554 execution = await _await_node_execution(tasks[flow_node])
1555 selected[flow_node] = execution.result
1557 return _DependencyResolution(
1558 matched=persisted.matched,
1559 results=selected,
1560 handled_failures=tuple(
1561 nodes_by_name[node_name] for node_name in persisted.handled_failures
1562 ),
1563 )
1566def _persisted_resolution(
1567 resolution: _DependencyResolution,
1568) -> _PersistedDependencyResolution:
1569 return _PersistedDependencyResolution(
1570 matched=resolution.matched,
1571 selected_nodes=tuple(node.name for node in resolution.results),
1572 handled_failures=tuple(node.name for node in resolution.handled_failures),
1573 )
1576def _raise_task_error(value: BaseException) -> NoReturn:
1577 if isinstance(value, _FlowControlSignal):
1578 raise value
1579 if isinstance(value, Exception):
1580 control_error = _find_control_error(value)
1581 if control_error is not None:
1582 raise _FlowControlSignal(control_error) from value
1583 raise value
1584 raise value
1587def _raise_collected_task_errors(values: Iterable[object]) -> None:
1588 suspensions: list[SuspendExecution] = []
1589 for value in values:
1590 if isinstance(value, SuspendExecution):
1591 suspensions.append(value)
1592 elif isinstance(value, _FlowControlSignal):
1593 raise value
1594 elif isinstance(value, Exception):
1595 control_error = _find_control_error(value)
1596 if control_error is not None:
1597 raise _FlowControlSignal(control_error) from value
1598 raise _FlowControlSignal(value) from value
1599 elif isinstance(value, BaseException):
1600 raise value
1602 timed_suspensions = [
1603 suspension
1604 for suspension in suspensions
1605 if isinstance(suspension, TimedSuspendExecution)
1606 ]
1607 if timed_suspensions:
1608 raise min(
1609 timed_suspensions,
1610 key=lambda suspension: suspension.scheduled_timestamp,
1611 )
1612 if suspensions:
1613 raise suspensions[0]
1616async def _resolve_dependency_expression(
1617 target: FlowNode[Any],
1618 expression: _DependencyExpression,
1619 tasks: Mapping[FlowNode[Any], asyncio.Task[_NodeExecution]],
1620 resolver_tasks: Mapping[
1621 tuple[FlowNode[Any], int],
1622 asyncio.Task[_PersistedDependencyResolution],
1623 ],
1624) -> _DependencyResolution:
1625 if isinstance(expression, _DependencyLeaf):
1626 execution = await _await_node_execution(tasks[expression.node])
1627 result = execution.result
1628 evaluation = expression.evaluate({expression.node: result})
1629 return _DependencyResolution(
1630 matched=evaluation.status is _EvaluationStatus.MATCHED,
1631 results={expression.node: result},
1632 handled_failures=evaluation.handled_failures,
1633 )
1635 composite = cast(_CompositeDependencyExpression, expression)
1636 if composite.mode is _DependencyMode.ANY:
1637 return await _await_resolver_resolution(
1638 resolver_tasks[(target, id(composite))],
1639 composite,
1640 tasks,
1641 )
1643 child_tasks: list[asyncio.Task[_DependencyResolution]] = []
1644 for child_expression in composite.children:
1646 async def resolve_child(
1647 current_child: _DependencyExpression = child_expression,
1648 ) -> _DependencyResolution:
1649 return await _resolve_dependency_expression(
1650 target,
1651 current_child,
1652 tasks,
1653 resolver_tasks,
1654 )
1656 child_tasks.append(create_eager_task(resolve_child))
1658 values = await asyncio.gather(*child_tasks, return_exceptions=True)
1659 children: list[_DependencyResolution] = []
1660 for value in values:
1661 if isinstance(value, BaseException):
1662 _raise_task_error(value)
1663 children.append(value)
1665 selected: dict[FlowNode[Any], FlowNodeResult[Any]] = {}
1666 for child_resolution in children:
1667 selected.update(child_resolution.results)
1668 if any(not child_resolution.matched for child_resolution in children):
1669 return _DependencyResolution(matched=False, results=selected)
1670 return _DependencyResolution(
1671 matched=True,
1672 results=selected,
1673 handled_failures=tuple(
1674 dependency
1675 for child_resolution in children
1676 for dependency in child_resolution.handled_failures
1677 ),
1678 )
1681async def _resolve_any_expression(
1682 target: FlowNode[Any],
1683 expression: _CompositeDependencyExpression,
1684 tasks: Mapping[FlowNode[Any], asyncio.Task[_NodeExecution]],
1685 resolver_tasks: Mapping[
1686 tuple[FlowNode[Any], int],
1687 asyncio.Task[_PersistedDependencyResolution],
1688 ],
1689) -> _PersistedDependencyResolution:
1690 children: list[asyncio.Task[_DependencyResolution]] = []
1691 completed: asyncio.Queue[int] = asyncio.Queue()
1692 for index, child_expression in enumerate(expression.children):
1694 async def resolve_child(
1695 current_child: _DependencyExpression = child_expression,
1696 current_index: int = index,
1697 ) -> _DependencyResolution:
1698 try:
1699 return await _resolve_dependency_expression(
1700 target,
1701 current_child,
1702 tasks,
1703 resolver_tasks,
1704 )
1705 finally:
1706 completed.put_nowait(current_index)
1708 children.append(create_eager_task(resolve_child))
1710 pending = set(range(len(children)))
1711 unmatched: dict[FlowNode[Any], FlowNodeResult[Any]] = {}
1712 suspensions: list[SuspendExecution] = []
1713 try:
1714 while pending:
1715 index = await completed.get()
1716 pending.remove(index)
1717 child_task = children[index]
1718 try:
1719 resolution = child_task.result()
1720 except SuspendExecution as error:
1721 suspensions.append(error)
1722 continue
1723 except BaseException as error:
1724 _raise_task_error(error)
1725 if resolution.matched:
1726 return _persisted_resolution(
1727 _DependencyResolution(
1728 matched=True,
1729 results={**unmatched, **resolution.results},
1730 handled_failures=resolution.handled_failures,
1731 )
1732 )
1733 unmatched.update(resolution.results)
1734 if suspensions:
1735 timed_suspensions = [
1736 error
1737 for error in suspensions
1738 if isinstance(error, TimedSuspendExecution)
1739 ]
1740 if timed_suspensions:
1741 raise min(
1742 timed_suspensions,
1743 key=lambda error: error.scheduled_timestamp,
1744 )
1745 raise suspensions[0]
1746 return _persisted_resolution(
1747 _DependencyResolution(matched=False, results=unmatched)
1748 )
1749 finally:
1750 for index in pending:
1751 children[index].cancel()
1752 if pending:
1753 await asyncio.gather(
1754 *(children[index] for index in pending),
1755 return_exceptions=True,
1756 )
1759def _any_expressions(
1760 expression: _DependencyExpression,
1761) -> tuple[_CompositeDependencyExpression, ...]:
1762 if isinstance(expression, _DependencyLeaf):
1763 return ()
1764 composite = cast(_CompositeDependencyExpression, expression)
1765 nested = tuple(
1766 nested_expression
1767 for child in composite.children
1768 for nested_expression in _any_expressions(child)
1769 )
1770 if composite.mode is _DependencyMode.ANY:
1771 return (*nested, composite)
1772 return nested
1775def _flow_node_context(
1776 context: DurableContext,
1777 direct_dependencies: frozenset[FlowNode[Any]],
1778 dependency_results: Mapping[FlowNode[Any], FlowNodeResult[Any]],
1779) -> FlowNodeContext:
1780 flow_context = FlowNodeContext(
1781 execution_state=context.execution_state,
1782 operation_identifier=context.operation_identifier,
1783 step_id_prefix=context.step_id_prefix,
1784 replaying=context.is_replaying(),
1785 _direct_dependencies=direct_dependencies,
1786 _dependency_results=dependency_results,
1787 )
1788 if "step_counter" in context.__dict__:
1789 flow_context.__dict__["step_counter"] = context.__dict__["step_counter"]
1790 return flow_context
1793def _resolve_flow_node_inputs(
1794 value: Any,
1795 results: Mapping[FlowNode[Any], FlowNodeResult[Any]],
1796) -> Any:
1797 if isinstance(value, _FlowNodeInput):
1798 return value.resolve(results)
1799 if isinstance(value, list):
1800 resolved_list = [_resolve_flow_node_inputs(item, results) for item in value]
1801 return (
1802 value
1803 if all(a is b for a, b in zip(resolved_list, value, strict=True))
1804 else resolved_list
1805 )
1806 if isinstance(value, tuple):
1807 resolved_tuple = tuple(
1808 _resolve_flow_node_inputs(item, results) for item in value
1809 )
1810 if all(a is b for a, b in zip(resolved_tuple, value, strict=True)):
1811 return value
1812 if hasattr(value, "_fields"):
1813 return type(value)(*resolved_tuple)
1814 return resolved_tuple
1815 if isinstance(value, dict):
1816 resolved_items = tuple(
1817 (
1818 _resolve_flow_node_inputs(key, results),
1819 _resolve_flow_node_inputs(item, results),
1820 )
1821 for key, item in value.items()
1822 )
1823 if all(
1824 resolved_key is key and resolved_value is item
1825 for (resolved_key, resolved_value), (key, item) in zip(
1826 resolved_items,
1827 value.items(),
1828 strict=True,
1829 )
1830 ):
1831 return value
1832 return dict(resolved_items)
1833 return value
1836async def _clone_flow_node_arguments(
1837 args: tuple[Any, ...],
1838 kwargs: Mapping[str, Any],
1839 results: Mapping[FlowNode[Any], FlowNodeResult[Any]],
1840) -> tuple[tuple[Any, ...], dict[str, Any]]:
1841 """Clone one node's complete argument graph at its execution boundary."""
1842 resolved_args = tuple(_resolve_flow_node_inputs(value, results) for value in args)
1843 resolved_kwargs = {
1844 key: _resolve_flow_node_inputs(value, results) for key, value in kwargs.items()
1845 }
1846 cloned = await _clone_flow_value((resolved_args, resolved_kwargs))
1847 if not isinstance(cloned, tuple) or len(cloned) != 2:
1848 msg = "Cloned flow node arguments have an invalid structure."
1849 raise SerDesError(msg)
1850 cloned_args, cloned_kwargs = cloned
1851 if not isinstance(cloned_args, tuple) or not isinstance(cloned_kwargs, dict):
1852 msg = "Cloned flow node arguments have an invalid structure."
1853 raise SerDesError(msg)
1854 restored_args = cast(
1855 "tuple[Any, ...]",
1856 _restore_flow_node_input_references(args, cloned_args, results),
1857 )
1858 restored_kwargs = cast(
1859 "dict[str, Any]",
1860 _restore_flow_node_input_references(dict(kwargs), cloned_kwargs, results),
1861 )
1862 return restored_args, restored_kwargs
1865def _restore_flow_node_input_references(
1866 template: Any,
1867 cloned: Any,
1868 results: Mapping[FlowNode[Any], FlowNodeResult[Any]],
1869) -> Any:
1870 """Rebind projections after cloning while retaining cloned container values."""
1871 if isinstance(template, _FlowNodeInput):
1872 return template.resolve(results)
1873 if isinstance(template, list) and isinstance(cloned, list):
1874 return [
1875 _restore_flow_node_input_references(source, value, results)
1876 for source, value in zip(template, cloned, strict=True)
1877 ]
1878 if isinstance(template, tuple) and isinstance(cloned, tuple):
1879 restored = tuple(
1880 _restore_flow_node_input_references(source, value, results)
1881 for source, value in zip(template, cloned, strict=True)
1882 )
1883 if hasattr(template, "_fields"):
1884 return type(template)(*restored)
1885 return restored
1886 if isinstance(template, dict) and isinstance(cloned, dict):
1887 return dict(
1888 (
1889 _restore_flow_node_input_references(
1890 source_key,
1891 cloned_key,
1892 results,
1893 ),
1894 _restore_flow_node_input_references(
1895 source_value,
1896 cloned_value,
1897 results,
1898 ),
1899 )
1900 for (source_key, source_value), (cloned_key, cloned_value) in zip(
1901 template.items(),
1902 cloned.items(),
1903 strict=True,
1904 )
1905 )
1906 return cloned
1909async def _invoke_flow_node(
1910 flow_node: FlowNode[Any],
1911 results: Mapping[FlowNode[Any], FlowNodeResult[Any]],
1912) -> Any:
1913 func = cast(
1914 "Callable[..., Awaitable[Any]]",
1915 getattr(flow_node._func, "_durable_node_function"),
1916 )
1917 args = cast("tuple[Any, ...]", getattr(flow_node._func, "_durable_node_args"))
1918 kwargs = cast(
1919 "Mapping[str, Any]",
1920 getattr(flow_node._func, "_durable_node_kwargs"),
1921 )
1922 cloned_args, cloned_kwargs = await _clone_flow_node_arguments(
1923 args,
1924 kwargs,
1925 results,
1926 )
1927 return await func(*cloned_args, **cloned_kwargs)
1930async def _execute_node(
1931 flow_node: FlowNode[Any],
1932 tasks: Mapping[FlowNode[Any], asyncio.Task[_NodeExecution]],
1933 resolver_tasks: Mapping[
1934 tuple[FlowNode[Any], int],
1935 asyncio.Task[_PersistedDependencyResolution],
1936 ],
1937 resolver_ready: asyncio.Event,
1938) -> _NodeExecution:
1939 expression = flow_node._dependency
1940 if expression is None:
1941 resolution = _DependencyResolution(matched=True, results={})
1942 direct_dependencies: frozenset[FlowNode[Any]] = frozenset()
1943 else:
1944 await resolver_ready.wait()
1945 resolution = await _resolve_dependency_expression(
1946 flow_node,
1947 expression,
1948 tasks,
1949 resolver_tasks,
1950 )
1951 direct_dependencies = frozenset(leaf.node for leaf in expression.leaves())
1953 handled_failures = tuple(
1954 dependency.name for dependency in resolution.handled_failures
1955 )
1956 if not resolution.matched:
1957 return _NodeExecution(
1958 result=FlowNodeResult.skipped(),
1959 handled_failures=handled_failures,
1960 )
1962 consumer_results = await _clone_dependency_results(resolution.results)
1963 context = get_durable_context()
1964 flow_context = _flow_node_context(
1965 context,
1966 direct_dependencies,
1967 consumer_results,
1968 )
1969 try:
1970 with bind_current_context(flow_context):
1971 outcome = await _invoke_flow_node(flow_node, consumer_results)
1972 return _NodeExecution(
1973 result=FlowNodeResult.succeeded(outcome),
1974 handled_failures=handled_failures,
1975 )
1976 except Exception as error:
1977 control_error = _find_control_error(error)
1978 if control_error is not None:
1979 raise _FlowControlSignal(control_error) from error
1980 return _NodeExecution(
1981 result=FlowNodeResult.failed(_callable_error_object(error)),
1982 handled_failures=handled_failures,
1983 )
1986async def _execute_flow(frozen_flow: _FrozenFlow) -> FlowResult:
1987 tasks: dict[FlowNode[Any], asyncio.Task[_NodeExecution]] = {}
1988 resolver_tasks: dict[
1989 tuple[FlowNode[Any], int],
1990 asyncio.Task[_PersistedDependencyResolution],
1991 ] = {}
1992 resolver_ready = asyncio.Event()
1993 for flow_node in frozen_flow.execution_nodes:
1995 async def run_node(current_node: FlowNode[Any] = flow_node) -> _NodeExecution:
1996 return await _execute_node(
1997 current_node,
1998 tasks,
1999 resolver_tasks,
2000 resolver_ready,
2001 )
2003 tasks[flow_node] = run_in_child_context(
2004 run_node,
2005 name=flow_node.name,
2006 serdes=_NODE_EXECUTION_SERDES,
2007 )
2009 resolver_task_order: list[asyncio.Task[_PersistedDependencyResolution]] = []
2010 for flow_node in frozen_flow.execution_nodes:
2011 expression = flow_node._dependency
2012 if expression is None:
2013 continue
2014 for ordinal, any_expression in enumerate(
2015 _any_expressions(expression),
2016 start=1,
2017 ):
2019 async def run_resolver(
2020 target: FlowNode[Any] = flow_node,
2021 current_expression: _CompositeDependencyExpression = any_expression,
2022 ) -> _PersistedDependencyResolution:
2023 await resolver_ready.wait()
2024 return await _resolve_any_expression(
2025 target,
2026 current_expression,
2027 tasks,
2028 resolver_tasks,
2029 )
2031 resolver_task = run_in_child_context(
2032 run_resolver,
2033 name=f"flow-any-resolution-{flow_node.name}-{ordinal}",
2034 serdes=_DEPENDENCY_RESOLUTION_SERDES,
2035 )
2036 resolver_tasks[(flow_node, id(any_expression))] = resolver_task
2037 resolver_task_order.append(resolver_task)
2039 resolver_ready.set()
2040 values = await asyncio.gather(
2041 *tasks.values(),
2042 *resolver_task_order,
2043 return_exceptions=True,
2044 )
2045 _raise_collected_task_errors(values)
2046 node_values = values[: len(tasks)]
2047 execution_nodes = set(frozen_flow.execution_nodes)
2048 executions = {
2049 flow_node: _NodeExecution(result=FlowNodeResult.skipped())
2050 for flow_node in frozen_flow.nodes
2051 if flow_node not in execution_nodes
2052 }
2053 for flow_node, value in zip(
2054 frozen_flow.execution_nodes,
2055 node_values,
2056 strict=True,
2057 ):
2058 executions[flow_node] = cast("_NodeExecution", value)
2060 results = {
2061 flow_node.name: executions[flow_node].result for flow_node in frozen_flow.nodes
2062 }
2063 handled_failures = {
2064 name for execution in executions.values() for name in execution.handled_failures
2065 }
2066 handled_failures.update(
2067 output.node.name
2068 for output in frozen_flow.outputs
2069 if output.kind in {_FlowNodeInputKind.ERROR, _FlowNodeInputKind.RESULT}
2070 and executions[output.node].result.status is FlowNodeStatus.FAILED
2071 )
2072 unhandled_failures = tuple(
2073 flow_node.name
2074 for flow_node in frozen_flow.nodes
2075 if executions[flow_node].result.status is FlowNodeStatus.FAILED
2076 and flow_node.name not in handled_failures
2077 )
2078 unavailable_outputs = tuple(
2079 dict.fromkeys(
2080 output.node.name
2081 for output in frozen_flow.outputs
2082 if output.kind is _FlowNodeInputKind.OUTCOME
2083 and executions[output.node].result.status is not FlowNodeStatus.SUCCEEDED
2084 )
2085 )
2086 return FlowResult(
2087 results=results,
2088 outputs=tuple(
2089 output.project(executions[output.node].result)
2090 for output in frozen_flow.outputs
2091 ),
2092 unhandled_failures=unhandled_failures,
2093 unavailable_outputs=unavailable_outputs,
2094 _output_kinds=tuple(output.kind for output in frozen_flow.outputs),
2095 )
2098def _evaluate_definition(definition: Callable[[], Any]) -> _FrozenFlow:
2099 if not getattr(definition, "_durable_dag_definition", False):
2100 msg = "flow() requires a bound callable produced by @durable_dag."
2101 raise FlowDefinitionError(msg)
2103 builder = _FlowBuilder()
2104 token = _current_flow_builder.set(builder)
2105 try:
2106 with bind_durable_definition("flow"):
2107 output = definition()
2108 finally:
2109 _current_flow_builder.reset(token)
2111 if inspect.isawaitable(output):
2112 if inspect.iscoroutine(output):
2113 output.close()
2114 msg = "A durable DAG definition must execute synchronously."
2115 raise FlowDefinitionError(msg)
2116 return builder.freeze(output)
2119def flow(
2120 definition: Callable[[], Any],
2121 *,
2122 name: str | None = None,
2123) -> asyncio.Task[FlowResult]:
2124 """Validate and start a declarative acyclic durable workflow."""
2125 ensure_durable_operations_allowed("flow")
2126 get_durable_context()
2127 frozen_flow = _evaluate_definition(definition)
2128 flow_name = name or getattr(definition, "__name__", None) or "flow"
2129 child_task = run_in_child_context(
2130 functools.partial(_execute_flow, frozen_flow),
2131 name=flow_name,
2132 serdes=_FLOW_RESULT_SERDES,
2133 )
2135 async def finish_flow() -> FlowResult:
2136 try:
2137 result = await child_task
2138 except _FlowControlSignal as signal:
2139 raise signal.error
2140 except Exception as error:
2141 control_error = _find_control_error(error)
2142 if control_error is not None:
2143 raise control_error from error
2144 raise
2146 problems: list[str] = []
2147 if result.has_unhandled_failures:
2148 failures = ", ".join(result.unhandled_failures)
2149 problems.append(f"unhandled node failures: {failures}")
2150 if result.has_unavailable_outputs:
2151 outputs = ", ".join(result.unavailable_outputs)
2152 problems.append(
2153 f"unavailable outcome outputs: {outputs}; return node.result "
2154 "for conditional outputs"
2155 )
2156 if problems:
2157 msg = f"Flow has {'; '.join(problems)}."
2158 raise FlowExecutionError(msg, result)
2159 return result
2161 return create_eager_task(finish_flow)