beam: toposort_kahn deterministic Kahn topological sort with cycle reporting + tests

wizard-work/beam/CHECK_toposort.py

@@ -0,0 +1,41 @@
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from toposort_kahn import toposort
+
+def test_dag():
+    g = {"a":["b","c"],"b":["d"],"c":["d"],"d":[]}
+    o,c = toposort(g)
+    assert c is None
+    # a before b,c; d last
+    assert o.index("a") < o.index("b")
+    assert o.index("a") < o.index("c")
+    assert o.index("b") < o.index("d")
+    assert o.index("c") < o.index("d")
+    print("dag ok", o)
+
+def test_cycle():
+    g = {"a":["b"],"b":["c"],"c":["a"]}
+    o,c = toposort(g)
+    assert c is not None
+    assert set(c[:-1]) <= {"a","b","c"}
+    print("cycle ok", c)
+
+def test_empty():
+    o,c = toposort({})
+    assert o == [] and c is None
+    print("empty ok")
+
+def test_single():
+    o,c = toposort({"x":[]})
+    assert o == ["x"] and c is None
+    print("single ok")
+
+def test_disconnected():
+    g = {"a":[],"b":[]}
+    o,c = toposort(g)
+    assert c is None and o == ["a","b"]
+    print("disconnected ok", o)
+
+if __name__ == "__main__":
+    test_dag(); test_cycle(); test_empty(); test_single(); test_disconnected()
+    print("ALL PASS")

wizard-work/beam/toposort_kahn.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+"""toposort_kahn.py -- deterministic topological sort with cycle reporting.
+
+Implements Kahn's algorithm. Returns (order, cycle) where cycle is None on
+success and a list of nodes forming a cycle otherwise. Ties broken by node
+name for reproducibility.
+"""
+from __future__ import annotations
+import heapq
+from typing import Dict, List, Tuple
+
+
+def toposort(adj: Dict[str, List[str]]) -> Tuple[List[str], List[str] | None]:
+    # adj: node -> list of successors (edges node -> succ)
+    nodes = set(adj)
+    for succs in adj.values():
+        for s in succs:
+            nodes.add(s)
+    indeg = {n: 0 for n in nodes}
+    for n in adj:
+        for s in adj.get(n, []):
+            indeg[s] += 1
+    # min-heap for deterministic order
+    heap = [n for n in nodes if indeg[n] == 0]
+    heapq.heapify(heap)
+    order: List[str] = []
+    while heap:
+        n = heapq.heappop(heap)
+        order.append(n)
+        for s in sorted(adj.get(n, [])):
+            indeg[s] -= 1
+            if indeg[s] == 0:
+                heapq.heappush(heap, s)
+    if len(order) != len(nodes):
+        # find a cycle via remaining nodes
+        remaining = [n for n in nodes if n not in set(order)]
+        # build cycle: follow successors among remaining
+        remset = set(remaining)
+        # pick start
+        start = min(remaining)
+        # walk
+        path = [start]
+        seen = {start}
+        cur = start
+        while True:
+            succs = [s for s in adj.get(cur, []) if s in remset]
+            if not succs:
+                # shouldn't happen if indegree>0; fallback
+                break
+            nxt = min(succs)
+            if nxt in seen:
+                idx = path.index(nxt)
+                return order, path[idx:] + [nxt]
+            path.append(nxt)
+            seen.add(nxt)
+            cur = nxt
+        return order, remaining
+    return order, None
+
+
+if __name__ == "__main__":
+    # self-test
+    g = {"a": ["b"], "b": ["c"], "c": []}
+    o, c = toposort(g)
+    assert c is None and o == ["a", "b", "c"], (o, c)
+    g2 = {"a": ["b"], "b": ["a"]}
+    o2, c2 = toposort(g2)
+    assert c2 is not None, (o2, c2)
+    print("OK", o, c2)