Find bottleneck dependencies in a parallel build

The build is fully determined by the graph: each dependency starts as soon as
its last prerequisite finishes, every build takes one unit, and there is no
limit on parallel builders. So the schedule is fixed, and the question is just
"which time steps contain exactly one dependency?"

Brute force. Simulate the clock. At each step, scan every unbuilt
dependency and check whether all its prerequisites are done; build those, and
if only one was ready, record it. Each step costs O(V + E) and there can be
V steps (a chain), so this is O(V * (V + E)).

Optimal: Kahn's algorithm, one wave at a time. A dependency's build step
is 1 + max(step of its prerequisites), which is its longest-path depth from
a source. Kahn's topological sort already produces exactly these groups if you
process it in waves instead of one node at a time:

1. Count in-degrees. Every node with in-degree 0 is ready at step 1.
2. The current wave is everything ready now. Build the whole wave, then
   decrement the in-degree of each child; children that hit 0 form the next
   wave.
3. A wave with exactly one node is a bottleneck.

Why a child lands in the right wave: its in-degree reaches 0 only when its
last prerequisite is processed, and that prerequisite is in the latest wave
among its parents. So the child joins the wave right after its slowest parent,
which is precisely when it would start building.

Invariants. Every node in wave k has all prerequisites in waves < k,
and at least one in wave k - 1. Every node is enqueued exactly once. If
fewer than n nodes were emitted, the remainder sit on a cycle, so raise
instead of returning a wrong answer.

A trap worth naming: a bottleneck here is about the schedule, not about
graph structure. A node that every path runs through (a dominator or
articulation point) need not be alone in its step, because an unrelated
branch can be building at the same time. Two independent chains 0 -> 1 and
2 -> 3 -> 4 have no shared node, yet 4 is a bottleneck because it builds
alone at step 3.

from collections import deque
from typing import Sequence


def build_levels(n: int, edges: Sequence[tuple[int, int]]) -> list[list[int]]:
    """Group nodes by the time step they build in (Kahn's algorithm, level by level).

    A node builds at step = 1 + the latest step of any prerequisite,
    so each BFS "wave" of zero-indegree nodes is exactly one time step.
    """
    children: list[list[int]] = [[] for _ in range(n)]
    indegree = [0] * n
    for u, v in edges:
        children[u].append(v)
        indegree[v] += 1

    wave = [node for node in range(n) if indegree[node] == 0]
    levels: list[list[int]] = []
    built = 0
    while wave:
        levels.append(wave)
        built += len(wave)
        next_wave: list[int] = []
        for u in wave:
            for v in children[u]:
                indegree[v] -= 1
                if indegree[v] == 0:
                    next_wave.append(v)
        wave = next_wave

    if built != n:
        raise ValueError("dependency graph has a cycle")
    return levels


def find_bottlenecks(n: int, edges: Sequence[tuple[int, int]]) -> list[int]:
    """Nodes that are the only one building during their time step, in build order."""
    return [level[0] for level in build_levels(n, edges) if len(level) == 1]


def find_bottlenecks_weighted(
    durations: Sequence[int], edges: Sequence[tuple[int, int]]
) -> list[int]:
    """Follow-up: node i takes durations[i] > 0 units; unlimited parallel builders.

    A node is a bottleneck if, for some interval of positive length,
    it is the only node building.
    """
    n = len(durations)
    preds: list[list[int]] = [[] for _ in range(n)]
    for u, v in edges:
        preds[v].append(u)

    start = [0] * n
    for level in build_levels(n, edges):  # topological order
        for v in level:
            start[v] = max((start[u] + durations[u] for u in preds[v]), default=0)

    # Sweep: at equal times, process finishes (kind 0) before starts (kind 1).
    events = sorted(
        [(start[i] + durations[i], 0, i) for i in range(n)]
        + [(start[i], 1, i) for i in range(n)]
    )
    active: set[int] = set()
    result: set[int] = set()
    for idx, (time, kind, node) in enumerate(events):
        if kind == 0:
            active.discard(node)
        else:
            active.add(node)
        next_time = events[idx + 1][0] if idx + 1 < len(events) else time
        if next_time > time and len(active) == 1:
            result.add(next(iter(active)))
    return sorted(result, key=lambda i: (start[i], i))


if __name__ == "__main__":
    assert find_bottlenecks(4, [(0, 1), (0, 2), (1, 3), (2, 3)]) == [0, 3]
    assert find_bottlenecks(3, [(0, 1), (1, 2)]) == [0, 1, 2]
    assert find_bottlenecks(5, [(0, 1), (2, 3), (3, 4)]) == [4]
    assert find_bottlenecks_weighted([1, 5, 1], [(0, 1), (0, 2)]) == [0, 1]

find_bottlenecks_weighted is the usual second part (see Follow-ups); the
first part needs only build_levels and find_bottlenecks.

• build_levels / find_bottlenecks: O(V + E) time. Each node enters one
  wave and each edge is relaxed once. O(V + E) space for the adjacency list,
  in-degrees and the waves.
• Brute-force simulation, for comparison: O(V * (V + E)) time.
• find_bottlenecks_weighted: O(V + E) for start times, plus
  O(V log V) to sort the 2V start and finish events. O(V + E) space.

• n = 4, edges = [(0,1), (0,2), (1,3), (2,3)] (diamond) -> [0, 3]. Steps
  are {0}, {1, 2}, {3}.
• n = 3, edges = [(0,1), (1,2)] (chain) -> [0, 1, 2].
• n = 3, edges = [] -> []. All three build together at step 1.
• n = 1, edges = [] -> [0].
• n = 0 -> [].
• n = 5, edges = [(0,1), (1,2), (2,4), (0,3), (3,4)] -> [0, 2, 4]. Node 3
  shares step 2 with node 1; node 2 builds alone at step 3 because node 3 is
  already done, and node 4 waits for both.
• n = 5, edges = [(0,1), (2,3), (3,4)] (two disconnected chains) -> [4].
• n = 2, edges = [(0,1), (1,0)] -> raises ValueError (cycle).
• Duplicate edge [(0,1), (0,1)]: in-degree counts it twice and both copies
  are relaxed, so the answer is still [0, 1]. Mention it, or dedupe up front.
• Weighted, durations = [1, 5, 1], edges = [(0,1), (0,2)] -> [0, 1]. Node
  2 runs in [1, 2) alongside node 1, but node 1 runs alone in [2, 6).
• Weighted, durations = [2, 3], edges = [(0,1)] -> [0, 1]. A handoff at the
  same instant is not overlap.

• Different build durations. Compute each node's start time as the max
  finish time of its prerequisites (topological order), giving intervals
  [start, start + d). Sweep the 2V endpoints sorted by time, processing
  finishes before starts at equal times, and keep the active set. Any gap of
  positive length with exactly one active node marks that node. This is
  find_bottlenecks_weighted above.
• Only k builders. The schedule is no longer forced; you need a policy
  (for example, a ready queue ordered by longest remaining path, the classic
  list-scheduling heuristic). Simulate with a min-heap of finish times and a
  ready priority queue, then report steps where only one builder is busy.
  Note that optimal k-processor scheduling is NP-hard in general, so say
  which policy you picked.
• Total build time and the critical path. The makespan is the number of
  waves (or the max finish time when weighted). Keep a parent pointer to the
  prerequisite that set each node's start time and walk back from the last
  finisher to print the critical path. Every bottleneck lies on every
  critical path, which is a good sanity check.
• Incremental edges. If edges arrive over time and the answer must stay
  current, recomputing is O(V + E) per update. A cheaper version updates
  depths only downstream of the changed edge (a BFS from v that stops when
  a depth does not change), then adjusts per-level counts.
• Huge graphs. If the graph does not fit in memory, compute depths with
  repeated passes over an edge list on disk, or in a distributed job: each
  round propagates depth[v] = max(depth[u] + 1) and the number of rounds is
  the graph depth. Then a group-by on depth gives the level sizes.

Mid-level (L4)

• Recognizes the problem as a topological sort and processes Kahn's
  algorithm level by level.
• Builds adjacency lists and in-degrees correctly and returns the single-node
  levels.
• Walks through a diamond and a chain by hand and handles empty input.

Senior (L5)

• Explains why a wave equals a time step (the child joins after its last
  parent), and contrasts it with the O(V * (V + E)) simulation.
• Detects cycles from the emitted count instead of looping forever.
• Calls out disconnected components and duplicate edges without prompting.
• Extends to weighted durations with a correct sweep, including the
  tie rule at equal timestamps.

Staff (L6+)

• Separates "alone in the schedule" from structural notions like dominators
  and explains why they differ.
• Discusses the k-builder variant: the schedule now depends on policy,
  optimal scheduling is NP-hard, and critical-path-first is a sound
  heuristic.
• Connects bottlenecks to the critical path and to where a real build or
  pipeline system (Bazel, a Databricks job DAG) would spend effort: caching,
  splitting the bottleneck task, or reordering work.
• Sketches incremental or distributed recomputation when the graph is large
  or changing.