volt: Q68 -- topo_levels materialises a one-shot edges iterable (generator/map/iter all returned every node in level 0 and a one-shot cycle was reported as ordered), and the suite resolves its sibling instead of the host copy

edges = list(edges) above validation. The redundant ready-queue pre-sort is deleted (level is re-sorted one line later; 3000 random DAGs, 0 level-set diffs with it removed). The flattened-invariant assert is a raise, so -O cannot strip it. test_topo_levels resolves topo_levels from Path(__file__).parent, raises ImportError if the import resolves elsewhere, and its 17 cells are TestCase methods so unittest discover collects them (module-level functions were invisible to the gate). test_rendezvous_hash no longer sys.exit()s at import; its trials are a TestCase.

Measured at be363f5+worktree: bash wizard-work/beam/CHECK rc 0, Ran 136 tests, ALL PASS (242 checks) -- at be363f5 the same gate is rc 1 with zero cells run (ImportError: test_topo_sort imported from /root/.hermes/wizard-outputs/beam). Mutant grading and control cells: wizard-work/volt/q68-2026-10-25.md.

wizard-work/beam/test_rendezvous_hash.py

@@ -103,5 +103,16 @@ def gen():
 ok &= check("generator", rendezvous("key", gen()) in nodes)
 ok &= check("dedup", rendezvous("key", ["a","a","b"]) in ("a","b"))
 
-print("ALL" if ok else "SOME FAILED")
+import unittest
-sys.exit(0 if ok else 1)
+
+
+class Trials(unittest.TestCase):
+    def test_all_trials(self):
+        # runs under `unittest discover`, where a module-level sys.exit() at
+        # import time would turn the whole gate leg into an ImportError.
+        self.assertTrue(ok, "rendezvous trials failed")
+
+
+if __name__ == "__main__":
+    print("ALL" if ok else "SOME FAILED")
+    sys.exit(0 if ok else 1)

wizard-work/beam/test_topo_levels.py

@@ -1,63 +1,145 @@
+"""Cells for `topo_levels`, collectable by the lane's `CHECK`.
+
+`python3 -m unittest discover -p 'test_*.py'` collects this file only because
+the cells are `unittest.TestCase` methods; the previous module-level functions
+were invisible to the gate (Q68's third fault, the `retry_jitter` shape).
+
+The module under test is resolved from this file's directory and the import is
+verified to land on the sibling file: a copy under
+`/root/.hermes/wizard-outputs/beam` is never imported (Q62's host-pin class).
+The check is a `raise`, not an `assert`, so it keeps grading under `-O`.
+"""
+from __future__ import annotations
+
 import sys
-sys.path.insert(0, "/root/.hermes/wizard-outputs/beam")
+import unittest
-from topo_levels import topo_levels, flatten, CycleError
+from pathlib import Path
 
-def test_basic():
+HERE = Path(__file__).resolve().parent
-    nodes = ["a","b","c","d"]
+if str(HERE) not in sys.path:
-    edges = [("a","c"),("b","c"),("c","d")]
+    sys.path.insert(0, str(HERE))
-    levels = topo_levels(nodes, edges)
-    assert levels == [["a","b"],["c"],["d"]], levels
 
-def test_disconnected():
+import topo_levels  # noqa: E402
-    nodes = [1,2,3]
+from topo_levels import CycleError, flatten  # noqa: E402
-    assert topo_levels(nodes, []) == [[1,2,3]]
 
-def test_cycle():
+EXPECTED_MODULE = HERE / "topo_levels.py"
-    nodes = ["a","b","c"]
+_RESOLVED = Path(topo_levels.__file__).resolve()
-    edges = [("a","b"),("b","c"),("c","a")]
+if _RESOLVED != EXPECTED_MODULE:
-    try:
+    raise ImportError(
-        topo_levels(nodes, edges)
+        f"topo_levels resolved to {str(_RESOLVED)!r}, expected {str(EXPECTED_MODULE)!r}"
-    except CycleError as e:
+    )
-        assert set(e.remaining) == {"a","b","c"}
-    else:
-        raise AssertionError("expected CycleError")
 
-def test_unknown_node():
-    try:
-        topo_levels(["a"], [("a","z")])
-    except KeyError:
-        pass
-    else:
-        raise AssertionError("expected KeyError")
 
-def test_single_node_no_edges():
+class TestTopoLevels(unittest.TestCase):
-    assert topo_levels(["x"], []) == [["x"]]
+    # ---- the import itself is graded, not assumed -------------------------
+    def test_import_is_sibling(self):
+        self.assertEqual(Path(topo_levels.__file__).resolve(), EXPECTED_MODULE)
 
-def test_diamond():
+    # ---- level shape ------------------------------------------------------
-    nodes = ["a","b","c","d"]
+    def test_basic(self):
-    edges = [("a","b"),("a","c"),("b","d"),("c","d")]
+        levels = topo_levels.topo_levels(
-    levels = topo_levels(nodes, edges)
+            ["a", "b", "c", "d"], [("a", "c"), ("b", "c"), ("c", "d")]
-    assert levels == [["a"],["b","c"],["d"]], levels
+        )
+        self.assertEqual(levels, [["a", "b"], ["c"], ["d"]])
 
-def test_determinism():
+    def test_disconnected(self):
-    nodes = list("zyxwv")
+        self.assertEqual(topo_levels.topo_levels([1, 2, 3], []), [[1, 2, 3]])
-    edges = [("z","w"),("y","w")]
+
-    r1 = topo_levels(nodes, edges)
+    def test_single_node_no_edges(self):
-    r2 = topo_levels(list(reversed(nodes)), edges)
+        self.assertEqual(topo_levels.topo_levels(["x"], []), [["x"]])
-    assert r1 == r2, (r1, r2)
+
+    def test_diamond(self):
+        levels = topo_levels.topo_levels(
+            ["a", "b", "c", "d"], [("a", "b"), ("a", "c"), ("b", "d"), ("c", "d")]
+        )
+        self.assertEqual(levels, [["a"], ["b", "c"], ["d"]])
+
+    def test_determinism_under_input_permutation(self):
+        nodes = list("zyxwv")
+        edges = [("z", "w"), ("y", "w")]
+        self.assertEqual(
+            topo_levels.topo_levels(nodes, edges),
+            topo_levels.topo_levels(list(reversed(nodes)), edges),
+        )
+
+    def test_level_interior_is_sorted_by_key(self):
+        # adjacency arrival order ('c' before 'a') differs from sorted order;
+        # a level emitted in arrival order ([['p'], ['c', 'a']]) must fail.
+        self.assertEqual(
+            topo_levels.topo_levels(["p", "c", "a"], [("p", "c"), ("p", "a")]),
+            [["p"], ["a", "c"]],
+        )
+
+    def test_duplicate_nodes_deduped(self):
+        # without dedupe this returns [['a', 'a', 'b']].
+        self.assertEqual(
+            topo_levels.topo_levels(["a", "a", "b"], [("a", "b")]), [["a"], ["b"]]
+        )
+
+    # ---- one-shot iterables (edges is annotated Iterable) ------------------
+    def test_generator_edges_materialised(self):
+        edges = [("a", "c"), ("b", "c")]
+        self.assertEqual(
+            topo_levels.topo_levels(["a", "b", "c"], (e for e in edges)),
+            [["a", "b"], ["c"]],
+        )
+
+    def test_iter_edges_materialised(self):
+        edges = [("a", "c"), ("b", "c")]
+        self.assertEqual(
+            topo_levels.topo_levels(["a", "b", "c"], iter(edges)), [["a", "b"], ["c"]]
+        )
+
+    def test_map_edges_materialised(self):
+        edges = [("a", "c"), ("b", "c")]
+        self.assertEqual(
+            topo_levels.topo_levels(["a", "b", "c"], map(tuple, edges)),
+            [["a", "b"], ["c"]],
+        )
+
+    def test_generator_cycle_still_raises(self):
+        cycle = [("a", "b"), ("b", "c"), ("c", "a")]
+        with self.assertRaises(CycleError):
+            topo_levels.topo_levels(["a", "b", "c"], (e for e in cycle))
+
+    # ---- cycles -----------------------------------------------------------
+    def test_cycle_remaining_is_exact(self):
+        # 'd' is acyclic: `remaining = list(node_set)` would add it here.
+        with self.assertRaises(CycleError) as ctx:
+            topo_levels.topo_levels(
+                ["a", "b", "c", "d"], [("a", "b"), ("b", "c"), ("c", "a")]
+            )
+        self.assertEqual(ctx.exception.remaining, ["a", "b", "c"])
+
+    # ---- unknown nodes: the message, not just the exception type -----------
+    def test_unknown_prerequisite_message(self):
+        with self.assertRaises(KeyError) as ctx:
+            topo_levels.topo_levels(["a"], [("z", "a")])
+        self.assertEqual(ctx.exception.args[0], "edge references unknown node: 'z'")
+
+    def test_unknown_dependent_message(self):
+        with self.assertRaises(KeyError) as ctx:
+            topo_levels.topo_levels(["a"], [("a", "z")])
+        self.assertEqual(ctx.exception.args[0], "edge references unknown node: 'z'")
+
+    # ---- flatten ----------------------------------------------------------
+    def test_flatten_roundtrip(self):
+        nodes = ["a", "b", "c", "d", "e"]
+        edges = [("a", "c"), ("b", "c"), ("c", "d"), ("c", "e")]
+        levels = topo_levels.topo_levels(nodes, edges)
+        flat = flatten(levels)
+        self.assertEqual(sorted(flat), sorted(nodes))
+        pos = {n: i for i, n in enumerate(flat)}
+        for u, v in edges:
+            self.assertLess(pos[u], pos[v])
+
+    def test_flatten_keeps_level_order(self):
+        # reversing a level inside flatten gives ['b', 'a', 'c'].
+        self.assertEqual(
+            flatten(topo_levels.topo_levels(["a", "b"], [])),
+            ["a", "b"],
+        )
 
-def test_flatten_roundtrip():
-    nodes = ["a","b","c","d","e"]
-    edges = [("a","c"),("b","c"),("c","d"),("c","e")]
-    levels = topo_levels(nodes, edges)
-    flat = flatten(levels)
-    assert sorted(flat) == sorted(nodes)
-    # every edge prerequisite before dependent
-    pos = {n:i for i,n in enumerate(flat)}
-    for u,v in edges:
-        assert pos[u] < pos[v]
 
 if __name__ == "__main__":
-    for name in [n for n in globals() if n.startswith("test_")]:
+    unittest.main(verbosity=2)
-        globals()[name]()
-    print("ALL TESTS PASSED")

wizard-work/beam/topo_levels.py

@@ -29,6 +29,7 @@ def topo_levels(
     """
     node_set = list(dict.fromkeys(nodes))  # preserve insertion order, dedupe
     index = {n: i for i, n in enumerate(node_set)}
+    edges = list(edges)  # materialise once: `edges` may be a one-shot iterable
 
     # validate edges reference known nodes
     for a, b in edges:
@@ -43,8 +44,9 @@ def topo_levels(
         adj[a].append(b)
         indeg[b] += 1
 
-    # deterministic: process ready queue sorted by key
+    # deterministic: the only ordering point is `level` below; pre-sorting the
-    ready = sorted([n for n in node_set if indeg[n] == 0], key=_key)
+    # ready queue here would re-sort at the next line and change nothing.
+    ready = [n for n in node_set if indeg[n] == 0]
     levels: List[List[Any]] = []
     processed = 0
     # Kahn in waves
@@ -66,11 +68,16 @@ def topo_levels(
         remaining = [n for n in node_set if indeg[n] > 0]
         raise CycleError(sorted(remaining, key=_key))
 
-    # flatten check: every node appears exactly once
+    # flatten check: every node appears exactly once (a raise, not an assert,
+    # so the invariant is not stripped under -O)
     seen = []
     for lvl in levels:
         seen.extend(lvl)
-    assert len(seen) == len(node_set) == len(set(seen)), "internal invariant violated"
+    if not (len(seen) == len(node_set) == len(set(seen))):
+        raise RuntimeError(
+            f"internal invariant violated: {len(seen)} emitted, "
+            f"{len(node_set)} unique nodes, {len(set(seen))} distinct"
+        )
     return levels