16. Topological Sort for a Multi-Step Agent Pipeline

Problem

A lab's agent orchestrator runs a pipeline of named steps (e.g. "fetch", "validate", "summarize", "respond") where some steps must finish before others can start, expressed as dependency edges. Before the orchestrator executes anything, it needs a valid execution order that respects every dependency — and it needs to detect a misconfigured pipeline (a dependency cycle) up front rather than deadlocking at runtime.

Given the step names and a list of "must run before" dependency pairs, return one valid execution order, breaking ties deterministically by picking the lexicographically smallest eligible step. Raise on a cycle.

Source: src/16_agent_pipeline_topo_sort.py

def pipeline_execution_order(steps: list[str], dependencies: list[tuple[str, str]]) -> list[str]:

>>> pipeline_execution_order(
...     ["fetch", "validate", "summarize", "respond"],
...     [("fetch", "validate"), ("validate", "summarize"), ("summarize", "respond")],
... )
['fetch', 'validate', 'summarize', 'respond']
>>> pipeline_execution_order(["a", "b", "c"], [("a", "c"), ("b", "c")])
['a', 'b', 'c']

Step-by-Step Approach

1. Build an indegree count for every step (how many dependencies point into it) and an adjacency list (which steps each step unlocks).
2. Seed a "frontier" with every step that has indegree 0 — these have no unmet dependency and are immediately eligible to run first.
3. Use a min-heap (not a plain queue) for the frontier, so that whenever multiple steps are simultaneously eligible, the lexicographically smallest one is always chosen next, making the output deterministic.
4. Repeatedly pop the smallest eligible step, append it to the output order, and for each step it unlocks, decrement that step's indegree; if a successor's indegree drops to 0, push it onto the frontier.
5. Continue until the frontier is empty, then compare how many steps were emitted to the total step count.
6. If fewer steps were emitted than exist, some steps never reached indegree 0 — they're stuck in a cycle waiting on each other — raise ValueError instead of silently returning a partial (invalid) order.

The key insight is that Kahn's algorithm turns "does this DAG have a valid order" and "what is that order" into the same computation: a step only gets emitted once every one of its dependencies has already been emitted, so if the algorithm terminates having emitted every step, that emission order is automatically a valid topological order — and if it can't emit everything, that's proof of a cycle, with no separate cycle-detection pass needed.

Reference solution

import heapq


def pipeline_execution_order(steps: list[str], dependencies: list[tuple[str, str]]) -> list[str]:
    # Kahn's algorithm with a min-heap frontier instead of a plain queue, so
    # ties among simultaneously-eligible steps resolve deterministically.
    # O((n + e) log n) time, O(n + e) space.
    indegree = {step: 0 for step in steps}
    adjacency = {step: [] for step in steps}
    for u, v in dependencies:
        adjacency[u].append(v)
        indegree[v] += 1

    frontier = [step for step in steps if indegree[step] == 0]
    heapq.heapify(frontier)
    order = []
    while frontier:
        step = heapq.heappop(frontier)
        order.append(step)
        for successor in adjacency[step]:
            indegree[successor] -= 1
            if indegree[successor] == 0:
                heapq.heappush(frontier, successor)

    if len(order) != len(steps):
        # some steps never reached indegree 0 -- they're stuck in a cycle
        raise ValueError("cycle detected in pipeline dependencies")
    return order

Key Functions & Tricks

• indegree dict — tracks how many unmet dependencies each step still has, the core state Kahn's algorithm reduces to zero over time.
• heapq.heapify / heapq.heappush / heapq.heappop — a min-heap frontier makes tie-breaking among simultaneously-eligible steps deterministic (lexicographically smallest first).
• adjacency list — lets each emitted step cheaply notify exactly the steps it unlocks, rather than rescanning all dependencies.
• len(order) != len(steps) — the cycle-detection check: a DAG always fully drains the frontier; a cycle leaves steps permanently stuck at indegree > 0.
• dict comprehension for indegree/adjacency init — guarantees every step (including ones with no dependencies at all) has an entry, avoiding KeyErrors during the main loop.

How to Recognize This Pattern