DAG executor with explicit dependencies and topological sort

[FernandoCelmer]

Labels: enhancement, discovery, RFC

Context

The current execution model uses sibling-style grouping:

• task.add(step=..., group_name="g1") groups tasks.
• Sequential.run iterates the queue linearly
  (workflow.py:235-261).
• SequentialGroup runs each group in a separate process
  sequentially.
• Parallel.run spawns one process per task with no inter-task
  ordering, and each subprocess receives an empty previous_context (workflow.py:459-476, related issue Enhancement: Pass previous_context between task groups #248).

This means:

• There is no explicit dependency declaration; correct ordering
  relies on the order of task.add calls.
• Cycle detection is impossible because there is no graph.
• Fan-out / fan-in cannot be expressed: a task that consumes
  outputs from two upstream tasks has no way to receive them both.

Concept

Introduce a real DAG model:

1. @action accepts an optional depends_on=[...] of upstream
   actions.
2. Manager builds a dict[step, set[step]] adjacency list at
   start.
3. A topological_sort (Kahn's algorithm) produces a level
   ordering. Within a level, all tasks are independent and may run in parallel; between levels, the engine waits for joins.
4. A CycleError is raised at build time if the graph is invalid.

API sketch

@action
def extract():
    return {"users": 150}

@action(depends_on=[extract])
def transform_a(ctx):
    return {"active": ctx.storage["users"] * 0.8}

@action(depends_on=[extract])
def transform_b(ctx):
    return {"churned": ctx.storage["users"] * 0.2}

@action(depends_on=[transform_a, transform_b])
def load(ctx):
    print(ctx.storage)

workflow = DotFlow()
workflow.task.add(step=extract)
workflow.task.add(step=transform_a)
workflow.task.add(step=transform_b)
workflow.task.add(step=load)

workflow.start()
# extract (level 0)
# transform_a, transform_b run in parallel (level 1)
# load (level 2) receives merged context from level 1

Topological sort reference (Kahn)

def topological_sort(graph: dict) -> list[list]:
    """Returns levels: each inner list is independent."""
    indegree = {n: 0 for n in graph}
    for node, deps in graph.items():
        for _ in deps:
            indegree[node] += 1

    levels = []
    ready = {n for n, d in indegree.items() if d == 0}
    visited = set()
    while ready:
        levels.append(sorted(ready, key=lambda n: id(n)))
        next_ready = set()
        for node in ready:
            visited.add(node)
            for other, deps in graph.items():
                if node in deps and other not in visited:
                    indegree[other] -= 1
                    if indegree[other] == 0:
                        next_ready.add(other)
        ready = next_ready

    if len(visited) != len(graph):
        raise CycleError(graph)
    return levels

Acceptance criteria

• @action(depends_on=[...]) accepted and stored on the
  Action instance
• Manager builds the adjacency list from registered tasks
• topological_sort returns levels and raises CycleError
• A new DAG executor consumes levels and runs each level via
  the existing concurrency primitives
• The four current modes (sequential, sequential_group,
  background, parallel) become DAG executor configurations
• Backward-compat: workflows with no depends_on keep current
  behavior
• Tests: linear chain, fan-out, fan-in, diamond, cycle (raises)

Why this is in scope

Issue #248 (pass previous_context between groups) is a symptom of the missing DAG model. Solving it properly closes #248, makes the four execution modes converge into a single executor, and unlocks features like deterministic replay and dry-run.

---