5. Pipeline Stage Topological Order

Problem

Fin runs a sequential, staged pipeline: refine/filter the query, retrieve + generate via RAG, validate the response, with safety checks gating every stage. A stage cannot run before its dependencies are satisfied.

Given a pipeline's stage names and a list of "must run before" dependency edges, compute one valid execution order, or detect that the dependency graph is broken (a cycle, which would mean the pipeline can never run). Use Kahn's algorithm (BFS with in-degree tracking) — when multiple stages are simultaneously eligible, break ties by processing them in the order they first appear in stages, so output is deterministic.

Source: src/5_pipeline_stage_topo_order.py

def topo_order(stages: list[str], deps: list[tuple[str, str]]) -> list[str]:
    ...

Examples:
>>> topo_order(
...     ["retrieve", "refine", "generate", "validate"],
...     [("refine", "retrieve"), ("retrieve", "generate"), ("generate", "validate")])
['refine', 'retrieve', 'generate', 'validate']

>>> topo_order(["a", "b", "c"], [("a", "b"), ("b", "c"), ("c", "a")])
ValueError: cycle detected

Step-by-Step Approach

1. Model stages as nodes and each (a, b) dependency as a directed edge a → b, meaning a must finish before b.
2. Build an adjacency list and an in-degree count for every node (the count of unfinished prerequisites it's waiting on).
3. Seed a frontier with every node that currently has in-degree 0 (no unresolved prerequisites) — these are immediately runnable.
4. To get deterministic output when multiple stages are simultaneously eligible, keep the frontier as a min-heap of each node's original index in the input list (not the node itself), so ties break by first-appearance order.
5. Repeatedly pop the smallest-index eligible node from the frontier, append it to the result, then decrement the in-degree of each of its outgoing neighbors; any neighbor whose in-degree just hit 0 becomes newly eligible and gets pushed onto the frontier.
6. After the loop, if the result contains every stage, it's a valid topological order; if it's short some stages, those remaining stages are stuck in a cycle (their in-degree never reached 0), so raise ValueError("cycle detected").

The key insight is that a node only becomes eligible once every one of its prerequisites has actually been processed and removed, so in-degree hitting zero is exactly the signal that a stage is unblocked — this is what makes Kahn's algorithm naturally BFS-like and gives cycle detection for free, since nodes trapped in a cycle can never reach in-degree 0.

Reference solution

import heapq


def topo_order(stages: list[str], deps: list[tuple[str, str]]) -> list[str]:
    # Kahn's algorithm, frontier kept as a min-heap of stage indices for
    # deterministic first-appearance tie-break, O((n + e) log n) time, O(n + e) space
    # O(1) lookup of first-appearance index
    index = {s: i for i, s in enumerate(stages)}
    graph: dict[str, list[str]] = {s: [] for s in stages}
    in_degree: dict[str, int] = {s: 0 for s in stages}
    for a, b in deps:
        graph[a].append(b)
        in_degree[b] += 1

    # seed frontier with indices, not names
    heap = [index[s] for s in stages if in_degree[s] == 0]
    # O(n) in-place, cheaper than n calls to heappush
    heapq.heapify(heap)
    result: list[str] = []
    while heap:
        # smallest index = earliest first-appearance wins ties
        i = heapq.heappop(heap)
        node = stages[i]
        result.append(node)
        for neighbor in graph[node]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                # newly eligible once in-degree hits 0
                heapq.heappush(heap, index[neighbor])

    # cycle nodes never reach in-degree 0, so they're missing here
    if len(result) != len(stages):
        raise ValueError("cycle detected")
    return result

Key Functions & Tricks

• {s: i for i, s in enumerate(stages)} — dict comprehension for O(1) index lookup vs O(n) list.index()
• {s: [] for s in stages} / {s: 0 for s in stages} — pre-init adjacency list and in-degree counts
• heapq.heapify(heap) — O(n) in-place heapify vs O(n log n) via repeated heappush
• heapq.heappush(heap, index[neighbor]) / heapq.heappop(heap) — O(log n) frontier push/pop
• Heap of indices instead of stage names — smallest index wins ties, giving first-appearance order
• Kahn's algorithm / in-degree tracking — in-degree 0 means all prerequisites already processed
• len(result) != len(stages) — cycle detection for free; stuck nodes never reach in-degree 0

How to Recognize This Pattern