[Feature]: first-class acyclic DAG workflow composition

[zhongkechen]

What would you like?

Add a first-class, SDK-level API for authoring static acyclic workflows with
declarative dependencies and conditional success or failure routes.

The public API should use a Python-native definition DSL:

result = await flow(my_flow(param))

This proposal intentionally excludes loops and state-machine execution because
the current backend cannot compact or replace checkpoint history. A cyclic
workflow can run with existing primitives, but each iteration increases stored
history and replay cost.

Related requests:

• First-class DAG workflow authoring across Java, TypeScript, and Python aws/aws-durable-execution-docs#224
• First-class state machine workflow authoring across Java, TypeScript, and Python aws/aws-durable-execution-docs#225

Proposed API

@durable_dag decorates a synchronous graph-definition function. Like
@durable_callable, calling the decorated function binds its arguments and
returns a zero-argument callable without executing the function body. flow()
evaluates that callable synchronously inside an isolated definition context,
validates the complete graph, and only then creates a durable child context and
executes the graph.

from async_durable_execution import durable_dag, flow, node


@durable_dag
def my_flow(param):
    a = node(run_a, name="A")
    b = node(run_b, name="B")
    c = node(run_c, name="C")
    d = node(run_d, name="D")

    a >> b
    a.failed >> c
    (b.succeeded | c.succeeded) >> d

    return d


result = await flow(my_flow(param), name="conditional-workflow")

If A succeeds, B executes and C is skipped. If A fails, C executes and B is
skipped. D runs after whichever route succeeds.

The definition function must be synchronous. Decorating an async def function
with @durable_dag raises a definition error. Definition code and bound
parameters that affect graph construction must be deterministic and replay
stable.

Dependency operators

Operators only construct the graph; they never start durable execution:

• a >> b: shorthand for a successful dependency.
• a.succeeded >> b: explicitly run B after A succeeds.
• a.failed >> b: run B after A fails.
• a.completed >> b: run B after any terminal logical status.
• (a & b) >> c: require all matching dependencies.
• (a | b) >> c: require any matching dependency.
• a >> (b, c): fan out to B and C.

Examples should require parentheses around & and | expressions because
Python operator precedence can otherwise be surprising.

Each node may have only one root dependency expression. Repeated statements
targeting the same node, such as a >> c followed by b >> c, are rejected.
Authors must express the intended relationship explicitly as (a & b) >> c or
(a | b) >> c. Duplicate dependencies inside one expression are also rejected.

node() returns a typed FlowNode[T]. A node callable receives a
FlowNodeContext, allowing direct dependency results to be accessed by node
handle rather than string:

@durable_dag
def result_access_example():
    a = node(run_a, name="A")

    async def run_b(ctx: FlowNodeContext) -> Output:
        a_result = ctx.result(a)
        if a_result.status is FlowNodeStatus.FAILED:
            handle_error(a_result.error)
        return await transform(a_result.outcome)

    b = node(run_b, name="B")
    a.completed >> b
    return b

FlowNodeResult[T] contains:

• status: SUCCEEDED, FAILED, or SKIPPED.
• outcome: the node outcome when available.
• error: the captured user error when available.

A result may contain an outcome, an error, or both. A skipped result normally
contains neither. FlowNodeContext.result() is limited to declared direct
dependencies.

The definition may return one node or a tuple of nodes to select flow outputs.
Returning None produces only the complete per-node result. Selected outputs
are returned as FlowNodeResult values, preserving status, outcome, and error
rather than collapsing skipped or failed outputs to None.

FlowResult contains every node result plus the selected outputs.
FlowExecutionError contains the complete FlowResult when the flow has
unhandled failures.

Definition phase

Definition and execution are separate:

1. flow() receives the bound zero-argument callable produced by
   @durable_dag.
2. It binds a fresh graph builder and a definition guard through ContextVars.
3. It invokes the synchronous definition callable.
4. Starting step(), wait(), invoke(), parallel(), map(), a child
   context, flow(), or another durable operation while defining a graph raises
   InvalidStateError.
5. flow() freezes and validates the graph.
6. Only after validation succeeds does flow() create its named durable child
   context and durable node tasks.

The definition guard is checked centrally by durable-operation context lookup,
so calling a durable operation without awaiting it is also rejected.

node() outside a definition context raises InvalidStateError. Nodes and
dependency expressions are owned by their builder and cannot be mixed across
graph definitions.

Validation and cycle detection

Validation must finish before any checkpoint:

1. Require unique, non-empty node names.
2. Reject unknown dependencies, repeated target expressions, duplicate
   dependencies, and self-edges.
3. Run a stable Kahn topological sort using declaration order as the tie-breaker.
4. If fewer than all nodes are consumed, run DFS over the remaining nodes to
   report a cycle such as A -> B -> C -> A.

All conditional edges participate in cycle detection even when their conditions
are mutually exclusive at runtime. The graph must be structurally acyclic.
Validation and sorting should run in O(V + E) time.

Execution model

The runtime can be a composite over existing primitives:

1. Evaluate, freeze, and validate the graph definition without checkpoints.
2. Create one named run_in_child_context() operation for the flow itself.
3. Inside the flow child, register every node as a named child-context task in
   stable topological order before awaiting any result.
4. Inside each node child, await only that node's direct dependency tasks.
5. Evaluate its dependency expression and either invoke the node callable or
   return a logical SKIPPED result.
6. Collect node tasks without allowing one user failure to cancel or orphan
   independent durable work.
7. Return and checkpoint the complete FlowResult from the flow child.
8. Outside the flow child boundary, return FlowResult or raise
   FlowExecutionError containing that checkpointed result.

Stable registration allocates deterministic operation IDs before completion
timing can affect scheduling. Each node has its own operation-ID scope, so
concurrent completion order cannot alter child IDs during replay.

Only user-code failures are logical node failures. Suspension, task
cancellation, retryable invocation failures, checkpoint failures, serialization
failures, and other SDK or infrastructure control errors propagate out of the
flow child and do not activate .failed routes or produce a terminal
FlowResult.

No new backend operation type, runner processor, or protocol change is needed.
Existing child checkpoints provide replay, result and error persistence, SerDes,
and per-node operation-tree visibility.

Durable operations inside nodes

The definition-phase restriction applies only while the decorated function
builds the graph. Once the graph is frozen and a node becomes eligible, its
callable runs inside that node's named durable child context and may use the full
existing SDK.

FlowNodeContext extends DurableContext, following the existing
MapItemContext pattern. It adds dependency-result access while preserving the
execution state, operation-ID prefix, replay state, and active context binding
required by top-level durable helpers.

This permits step(), wait(), callbacks, invoke(), parallel(), map(),
with_retry(), nested flows, and nested child contexts inside nodes. Their
operation IDs are scoped beneath the node child context and remain deterministic
regardless of sibling completion order.

Dependency matching

• ALL: wait for every dependency, then run only if every condition matches.
• ANY: race accepted dependency conditions and resolve on the first matching
  terminal result, analogous to
  DurableFuture.anyOf.
  A terminal dependency whose condition does not match does not win the race;
  skip only after all dependencies settle without a match.
• Nodes without dependencies start immediately.

When a condition cannot match, the node callable is not invoked. Because the
backend has no SKIPPED status, the child checkpoints a successful serialized
logical result with status=SKIPPED.

.completed matches all terminal logical statuses, including SKIPPED. A
.completed transition from a failed node does not by itself handle that
failure; only an activated .failed transition does.

Failure semantics

• Only user-code failures become logical FAILED node results.
• Node failures are durably checkpointed and included in FlowResult.
• Independent branches continue running after a user failure.
• A matching .failed dependency activates its failure-handler node.
• A source failure is handled only when that source's own .failed condition
  participates in the dependency match that activates a handler. For ANY,
  another condition winning the race does not retroactively handle a later
  source failure.
• Handler failures are evaluated independently.
• A failure with no activated failure transition remains unhandled.
• Any unhandled failure causes FlowExecutionError after all runnable nodes
  settle and the complete FlowResult is checkpointed.
• Existing step retries and with_retry() remain the retry mechanism before a
  node reaches terminal failure.

Explicit non-goals

• Dynamic or cyclic graphs, back edges, and repeated node activation.
• Asynchronous graph construction or graph construction that performs I/O.
• State-machine snapshots or terminal-state semantics.
• History compaction or continue_as_new.
• Native edge visualization in the Lambda service; the current protocol does
  not persist graph edges.
• Using recurse() as history compaction. It creates a chained invocation and
  is subject to Lambda recursion protection.

Acceptance criteria

• @durable_dag binds parameters without executing the definition body.
• Decorating an asynchronous definition is rejected.
• await flow(my_flow(param)) synchronously evaluates and validates the complete
  definition before creating any checkpoint.
• The flow itself is a named child context and its complete result is
  checkpointed.
• Definition-time durable operations are rejected, including un-awaited calls.
• Nodes and expressions cannot be mixed across graph definitions.
• Linear, fan-out, fan-in, diamond, disconnected, empty, and single-node flows.
• Success and failure routing, including A-to-B on success and A-to-C on failure.
• Correct ALL and first-matching ANY behavior.
• Repeated target expressions are rejected; explicit & or | is required.
• A downstream node starts once its own dependencies match without waiting for
  unrelated nodes.
• Independent work settles after another node has a user failure.
• SDK control and infrastructure errors propagate and never activate failure
  routes.
• Skipped node callables are never invoked.
• Results and selected outputs preserve SUCCEEDED, FAILED, and SKIPPED
  status plus outcome and error.
• Eligible node callables can use every existing durable operation.
• Durable operations in different nodes receive isolated deterministic
  operation-ID scopes.
• Stable operation IDs under different completion orders.
• Partial replay reuses completed checkpoints and runs only unfinished work.
• Duplicate names, invalid references, repeated target expressions, self-edges,
  and cycles fail before any checkpoint.
• Cycle errors include a concrete cycle path.
• An unhandled failure raises FlowExecutionError with the complete checkpointed
  FlowResult.

Compatibility

This is non-breaking. A future state-machine API may share outcome and
conditional-edge types, but discussion #225 requires backend support for
bounded snapshots, history compaction, or continue_as_new.

Is this a breaking change?

No.

Does this require an RFC?

Yes.

[zhongkechen]

Clarification: durable operations inside nodes

The definition-phase restriction applies only while my_flow() builds the graph. Once the graph is frozen and a node becomes eligible, its callable runs inside that node's named durable child context and may use the full existing SDK.

from datetime import timedelta

from async_durable_execution import parallel, step, wait


async def run_b(ctx: FlowNodeContext) -> Output:
    a_value = ctx.result(a)
    prepared = await step(prepare(a_value), name="prepare")
    await wait(timedelta(seconds=5), name="settle")
    branches = await parallel(
        branches=[left_branch, right_branch],
        name="process",
    )
    return combine(prepared, branches)

FlowNodeContext should extend DurableContext, following the existing MapItemContext pattern. It adds dependency-outcome access while preserving the execution state, operation-ID prefix, replay state, and active context binding required by top-level durable helpers.

This permits step(), wait(), callbacks, invoke(), parallel(), map(), with_retry(), and nested child contexts inside nodes. Their operation IDs are scoped beneath the node child context and remain deterministic regardless of sibling completion order. Existing replay, retry, SerDes, and failure behavior applies unchanged.

Acceptance criteria should include:

• Eligible node callables can use every existing durable operation.
• Durable operations in different nodes receive isolated deterministic operation-ID scopes.
• Durable operations remain prohibited only during flow definition, before graph validation.