beam: content-based duplicate file detector with tests

wizard-work/beam/_run_tests.py

@@ -0,0 +1,6 @@
+import test_factor as t
+for name in sorted(dir(t)):
+    if name.startswith('test_'):
+        getattr(t, name)()
+        print('PASS', name)
+print('ALL PASS')

wizard-work/beam/dedup.py

@@ -0,0 +1,100 @@
+#!/usr/bin/env python3
+"""Content-based duplicate file detector.
+
+Groups files by size first (cheap), then hashes only size-collisions (SHA-256),
+so identical content is detected regardless of name. Handles symlinks (skips
+by default to avoid loops), empty files (grouped together as one duplicate set),
+and large files (streamed in 1 MiB chunks).
+
+Usage:
+    dedup.py <dir> [--min-size N] [--follow-symlinks] [--json]
+"""
+import argparse
+import hashlib
+import json
+import os
+import sys
+from collections import defaultdict
+
+CHUNK = 1024 * 1024
+
+
+def sha256_file(path):
+    h = hashlib.sha256()
+    with open(path, "rb") as f:
+        while True:
+            b = f.read(CHUNK)
+            if not b:
+                break
+            h.update(b)
+    return h.hexdigest()
+
+
+def collect(root, follow_symlinks, min_size):
+    by_size = defaultdict(list)
+    for dirpath, dirnames, filenames in os.walk(root, followlinks=follow_symlinks):
+        # prune symlinked dirs unless following
+        if not follow_symlinks:
+            dirnames[:] = [d for d in dirnames
+                           if not os.path.islink(os.path.join(dirpath, d))]
+        for name in filenames:
+            p = os.path.join(dirpath, name)
+            if os.path.islink(p) and not follow_symlinks:
+                continue
+            try:
+                st = os.stat(p, follow_symlinks=follow_symlinks)
+            except OSError:
+                continue
+            if not os.path.isfile(p) if False else False:
+                pass
+            if st.st_size < min_size:
+                continue
+            by_size[st.st_size].append(p)
+    return by_size
+
+
+def find_duplicates(root, follow_symlinks=False, min_size=1):
+    by_size = collect(root, follow_symlinks, min_size)
+    groups = []
+    for size, paths in by_size.items():
+        if len(paths) < 2:
+            continue
+        by_hash = defaultdict(list)
+        for p in paths:
+            try:
+                by_hash[sha256_file(p)].append(p)
+            except OSError:
+                continue
+        for h, ps in by_hash.items():
+            if len(ps) >= 2:
+                groups.append({"hash": h, "size": size, "files": sorted(ps)})
+    groups.sort(key=lambda g: (-g["size"], g["hash"]))
+    return groups
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("root")
+    ap.add_argument("--min-size", type=int, default=0)
+    ap.add_argument("--follow-symlinks", action="store_true")
+    ap.add_argument("--json", action="store_true")
+    args = ap.parse_args()
+    if not os.path.isdir(args.root):
+        print("error: not a directory: " + args.root, file=sys.stderr)
+        return 2
+    groups = find_duplicates(args.root, args.follow_symlinks, args.min_size)
+    if args.json:
+        print(json.dumps(groups, indent=2))
+        return 0
+    if not groups:
+        print("no duplicates found")
+        return 0
+    for i, g in enumerate(groups, 1):
+        print("group %d: %d files, %d bytes, sha256=%s" % (i, len(g["files"]), g["size"], g["hash"]))
+        for f in g["files"]:
+            print("  " + f)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

wizard-work/beam/test_counterparty_sig.py

@@ -0,0 +1,111 @@
+"""Cells for counterparty_sig.py (Q85): authorship-aware repository signature.
+
+The module had no collected cell, so the property it exists for -- a signature
+that names only counterparty-authored bytes -- was ungraded.  Each cell builds
+a throwaway git repository with two identities and reads the signature back, so
+the authorship rule is measured against real commits rather than described.
+"""
+import hashlib
+import os
+import subprocess
+import tempfile
+import unittest
+
+import counterparty_sig as S
+
+SELF = ("Volt", "volt@troupe.local")
+THEM = ("Merlin", "merlin@troupe.local")
+
+
+def git(repo, *args, identity=None):
+    cmd = ["git", "-C", repo, "-c", "commit.gpgsign=false", "-c", "init.defaultBranch=main"]
+    if identity:
+        cmd += ["-c", "user.name=" + identity[0], "-c", "user.email=" + identity[1]]
+    return subprocess.run(cmd + list(args), check=True, capture_output=True, text=True)
+
+
+def commit(repo, path, body, identity, message):
+    full = os.path.join(repo, path)
+    os.makedirs(os.path.dirname(full), exist_ok=True)
+    with open(full, "w", encoding="utf-8") as fh:
+        fh.write(body)
+    git(repo, "add", path)
+    git(repo, "commit", "-q", "-m", message, identity=identity)
+    return git(repo, "rev-parse", "HEAD").stdout.strip()
+
+
+class SignatureFixture(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix="counterparty-sig.")
+        self.repo = self.tmp.name
+        git(self.repo, "init", "-q")
+        self.mine = commit(self.repo, "inbox/mine.txt", "my note\n", SELF, "volt: note")
+        self.theirs = commit(self.repo, "inbox/theirs.md", "their note\n", THEM, "merlin: note")
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+
+class TestCounterpartySignature(SignatureFixture):
+    def test_head_is_the_newest_counterparty_commit(self):
+        text, digest = S.signature(self.repo, SELF[0], ["inbox"])
+        self.assertEqual(text.splitlines()[0], "head %s" % self.theirs[:12])
+        self.assertNotIn(self.mine[:12], text)
+
+    def test_only_counterparty_authored_paths_are_listed(self):
+        text, _ = S.signature(self.repo, SELF[0], ["inbox"])
+        self.assertIn("inbox/theirs.md", text)
+        self.assertNotIn("inbox/mine.txt", text)
+
+    def test_blob_ids_are_the_twelve_character_prefixes_at_head(self):
+        text, _ = S.signature(self.repo, SELF[0], ["inbox"])
+        blob = git(self.repo, "rev-parse", "%s:inbox/theirs.md" % self.theirs).stdout.strip()
+        self.assertIn("inbox/theirs.md %s" % blob[:12], text)
+
+    def test_digest_is_the_hash_of_the_text_and_is_stable(self):
+        text1, digest1 = S.signature(self.repo, SELF[0], ["inbox"])
+        text2, digest2 = S.signature(self.repo, SELF[0], ["inbox"])
+        self.assertEqual(text1, text2)
+        self.assertEqual(digest1, digest2)
+        self.assertEqual(digest1, hashlib.sha256(text1.encode()).hexdigest())
+        self.assertEqual(len(digest1), 64)
+
+    def test_self_authored_later_edit_drops_the_path(self):
+        # The rule is authorship of the newest commit touching the path, not of
+        # the commit the file is read from: a self edit at head removes it.
+        commit(self.repo, "inbox/theirs.md", "their note\nplus mine\n", SELF, "volt: edit their note")
+        text, _ = S.signature(self.repo, SELF[0], ["inbox"])
+        self.assertNotIn("inbox/theirs.md", text)
+        self.assertEqual(len(text.splitlines()), 1, text)
+
+    def test_a_repository_with_no_counterparty_commit_raises(self):
+        with tempfile.TemporaryDirectory(prefix="counterparty-solo.") as solo:
+            git(solo, "init", "-q")
+            commit(solo, "a.txt", "only me\n", SELF, "volt: only")
+            with self.assertRaises(RuntimeError):
+                S.signature(solo, SELF[0], ["."])
+
+    def test_git_failure_is_raised_with_the_command_named(self):
+        with self.assertRaises(RuntimeError) as ctx:
+            S.run(os.path.join(self.tmp.name, "no-such-repo"), "log")
+        self.assertIn("git", str(ctx.exception))
+
+    def test_commits_reads_author_triples(self):
+        rows = S.commits(self.repo)
+        self.assertEqual([row[1] for row in rows], ["Merlin", "Volt"])
+        self.assertEqual([row[2] for row in rows], [THEM[1], SELF[1]])
+        for sha, _name, _email in rows:
+            self.assertEqual(len(sha), 40)
+
+    def test_newest_commit_for_path_is_cached(self):
+        cache = {}
+        first = S.newest_commit_for_path(self.repo, "inbox/theirs.md", SELF[0], cache)
+        self.assertEqual(first[0], self.theirs)
+        self.assertTrue(first[3], "a counterparty commit must be marked counterparty")
+        before = dict(cache)
+        S.newest_commit_for_path(self.repo, "inbox/theirs.md", SELF[0], cache)
+        self.assertEqual(cache, before, "the second call must not re-enter git")
+
+
+if __name__ == "__main__":
+    unittest.main()

wizard-work/beam/test_dedup.py

@@ -0,0 +1,81 @@
+import json, os, subprocess, sys, tempfile, unittest
+
+ROOT = os.path.dirname(os.path.abspath(__file__))
+SCRIPT = os.path.join(ROOT, "dedup.py")
+PY = sys.executable
+
+
+def make_tree(tmp):
+    d = os.path.join(tmp, "t")
+    os.makedirs(d)
+    a = os.path.join(d, "a.txt"); b = os.path.join(d, "b.txt")
+    with open(a, "w") as f: f.write("hello world\n" * 100)
+    with open(b, "w") as f: f.write("hello world\n" * 100)
+    with open(os.path.join(d, "c.txt"), "w") as f: f.write("unique content")
+    open(os.path.join(d, "e1"), "w").close()
+    open(os.path.join(d, "e2"), "w").close()
+    sub = os.path.join(d, "sub"); os.makedirs(sub)
+    with open(os.path.join(sub, "dup.txt"), "w") as f: f.write("hello world\n" * 100)
+    return d
+
+
+def run(args):
+    return subprocess.run([PY, SCRIPT] + args, capture_output=True, text=True)
+
+
+class Tests(unittest.TestCase):
+    def test_finds_duplicates(self):
+        with tempfile.TemporaryDirectory() as td:
+            d = make_tree(td)
+            r = run([d, "--json"])
+            self.assertEqual(r.returncode, 0, r.stderr)
+            groups = json.loads(r.stdout)
+            sizes = sorted(len(g["files"]) for g in groups)
+            self.assertEqual(sizes, [2, 3], sizes)
+            for g in groups:
+                self.assertEqual(len(g["hash"]), 64)
+
+    def test_no_duplicates(self):
+        with tempfile.TemporaryDirectory() as td:
+            d = os.path.join(td, "t2"); os.makedirs(d)
+            with open(os.path.join(d, "x"), "w") as f: f.write("x")
+            r = run([d])
+            self.assertEqual(r.returncode, 0)
+            self.assertIn("no duplicates", r.stdout)
+
+    def test_min_size_filters(self):
+        with tempfile.TemporaryDirectory() as td:
+            d = make_tree(td)
+            r = run([d, "--min-size", "100000", "--json"])
+            self.assertEqual(r.returncode, 0)
+            self.assertEqual(json.loads(r.stdout), [])
+
+    def test_missing_dir(self):
+        r = run(["/no/such/dir/xyz"])
+        self.assertEqual(r.returncode, 2)
+        self.assertIn("not a directory", r.stderr)
+
+    def test_symlink_skipped_by_default(self):
+        with tempfile.TemporaryDirectory() as td:
+            d = make_tree(td)
+            link = os.path.join(d, "link.txt")
+            os.symlink(os.path.join(d, "a.txt"), link)
+            r = run([d, "--json"])
+            groups = json.loads(r.stdout)
+            content_group = [g for g in groups if g["size"] > 0][0]
+            self.assertEqual(len(content_group["files"]), 3)
+            self.assertNotIn(link, content_group["files"])
+
+    def test_follow_symlinks(self):
+        with tempfile.TemporaryDirectory() as td:
+            d = make_tree(td)
+            link = os.path.join(d, "link.txt")
+            os.symlink(os.path.join(d, "a.txt"), link)
+            r = run([d, "--follow-symlinks", "--json"])
+            groups = json.loads(r.stdout)
+            content_group = [g for g in groups if g["size"] > 0][0]
+            self.assertEqual(len(content_group["files"]), 4)
+
+
+if __name__ == "__main__":
+    unittest.main()

wizard-work/beam/test_handle_resolve.py

@@ -0,0 +1,66 @@
+"""Cells for handle_resolve.py (Q85): longest-roster-prefix handle resolution.
+
+The module carried a `selftest()` of ten cases and no collected cell, so a
+regression in `resolve_handle` moved nothing the gate reads.  These cells assert
+the ten cases through the entry point, then pin the two rules a reader has to
+know: the boundary is a hyphen (not a string prefix), and the LONGEST match
+wins when the roster nests one handle inside another.
+"""
+import unittest
+
+import handle_resolve
+from handle_resolve import resolve_handle
+
+
+class TestResolveHandle(unittest.TestCase):
+    def test_documented_cases(self):
+        cases = [
+            (("volt", {"volt"}), "volt"),
+            (("volt-02", {"volt"}), "volt"),
+            (("voltmeter", {"volt"}), None),
+            (("volt", {"volt", "vol"}), "volt"),
+            (("cassian-q5-census", {"cassian"}), "cassian"),
+            (("gandalf", {"volt"}), None),
+            (("", {"volt"}), None),
+            (("volt-02", set()), None),
+            (("VOLT-02", {"volt"}), "volt"),
+            (("zephyrus-02", {"zephyrus", "zephyrus-0"}), "zephyrus"),
+        ]
+        for args, expected in cases:
+            self.assertEqual(resolve_handle(*args), expected, args)
+
+    def test_boundary_is_a_hyphen_not_a_string_prefix(self):
+        for name in ("voltmeter", "volt2", "voltaire", "volts"):
+            self.assertIsNone(resolve_handle(name, {"volt"}), name)
+        self.assertEqual(resolve_handle("volt-2", {"volt"}), "volt")
+
+    def test_longest_handle_wins(self):
+        self.assertEqual(resolve_handle("zephyrus-0-02", {"zephyrus", "zephyrus-0"}), "zephyrus-0")
+        self.assertEqual(resolve_handle("a-b-c", {"a", "a-b", "a-b-c"}), "a-b-c")
+
+    def test_exact_match_beats_a_longer_non_match(self):
+        self.assertEqual(resolve_handle("volt-02", {"volt", "volt-03"}), "volt")
+
+    def test_case_insensitive_both_ways(self):
+        self.assertEqual(resolve_handle("Volt-02", {"VOLT"}), "volt")
+        self.assertEqual(resolve_handle("volt-02", {"VoLt"}), "volt")
+
+    def test_roster_noise_is_ignored(self):
+        # Empty handles and None rosters are tolerated, not crashed on.
+        self.assertEqual(resolve_handle("volt-02", {"", "volt"}), "volt")
+        self.assertIsNone(resolve_handle("volt", None))
+        self.assertIsNone(resolve_handle(None, {"volt"}))
+
+    def test_returns_the_roster_spelling_lowercased(self):
+        # The result is a roster handle, so it is a set member -- lowercased.
+        roster = {"Volt", "volt-0"}
+        self.assertEqual(resolve_handle("VOLT-0-02", roster), "volt-0")
+        self.assertIn(resolve_handle("VOLT-02", roster), {h.lower() for h in roster})
+
+    def test_module_selftest_exits_zero(self):
+        # The file's own __main__ contract, which nothing collected could see.
+        self.assertEqual(handle_resolve.selftest(), 0)
+
+
+if __name__ == "__main__":
+    unittest.main()

wizard-work/beam/test_largest_rect_hist.py

@@ -1,6 +1,15 @@
-import random, sys
+import pathlib, random, sys
-sys.path.insert(0, "/root/.hermes/wizard-outputs/beam")
+
+# Bind the module from THIS file's directory, never from a host path (Q155:
+# this line read "/root/.hermes/wizard-outputs/beam", so the committed
+# largest_rect_hist.py was never the module the cells graded).
+HERE = pathlib.Path(__file__).resolve().parent
+sys.path.insert(0, str(HERE))
 from largest_rect_hist import largest_rectangle_area, _brute
+_MODULE_FILE = sys.modules["largest_rect_hist"].__file__
+assert _MODULE_FILE is not None, "largest_rect_hist imported without a __file__"
+MODULE_PATH = pathlib.Path(_MODULE_FILE).resolve()
+assert MODULE_PATH == HERE / "largest_rect_hist.py", MODULE_PATH
 
 def test_random():
     rng = random.Random(2024)

wizard-work/beam/test_majority.py

@@ -0,0 +1,109 @@
+"""Cells for majority.py (Q85): the generalized Boyer-Moore vote.
+
+No collected test imported the module, so its `__main__` corpus was the only
+thing that ever ran it.  Every cell here compares against a Counter, which
+shares no code with the module, so the vote's bookkeeping is graded rather
+than echoed.
+"""
+import random
+import unittest
+from collections import Counter
+
+import majority as M
+
+
+def brute_gt_k(data, k):
+    n = len(data)
+    return sorted(x for x, c in Counter(data).items() if c > n // k)
+
+
+def brute_majority(data):
+    if not data:
+        return None
+    hits = [x for x, c in Counter(data).items() if c > len(data) // 2]
+    return hits[0] if hits else None
+
+
+class TestMajority(unittest.TestCase):
+    def test_empty_and_small(self):
+        self.assertIsNone(M.majority_element([]))
+        self.assertEqual(M.majority_element([1]), 1)
+        self.assertIsNone(M.majority_element([1, 2]))
+        self.assertEqual(M.majority_element([1, 1, 2]), 1)
+        self.assertIsNone(M.majority_element([1, 2, 3]))
+
+    def test_odd_and_even_near_misses(self):
+        # Exactly half is not a majority; one more than half is.
+        self.assertIsNone(M.majority_element([1, 1, 2, 2]))
+        self.assertEqual(M.majority_element([1, 1, 1, 2]), 1)
+        self.assertIsNone(M.majority_element([1, 1, 2, 3]))
+
+    def test_aliases_agree(self):
+        data = [4, 4, 4, 1, 2]
+        self.assertEqual(M.majority(data), M.majority_element_v2(data))
+        self.assertEqual(M.majority(data), 4)
+
+    def test_counter_resistant_pattern(self):
+        # The order that defeats a naive "cancel adjacent" reading: no prefix
+        # has more candidates than cancellations until the very end.
+        data = [2, 1, 2, 1, 2]
+        self.assertEqual(M.majority_element(data), 2)
+        data = [1, 2, 1, 2, 1, 2, 2]
+        self.assertEqual(M.majority_element(data), 2)
+
+    def test_random_majority_against_counter(self):
+        rng = random.Random(42)
+        for _ in range(1500):
+            n = rng.randint(0, 40)
+            if n and rng.random() < 0.5:
+                maj = rng.randint(0, 5)
+                data = [maj] * (n // 2 + 1) + [rng.randint(0, 5) for _ in range(n - n // 2 - 1)]
+                rng.shuffle(data)
+            else:
+                data = [rng.randint(0, 5) for _ in range(n)]
+            self.assertEqual(M.majority_element(data), brute_majority(data), data)
+
+    def test_gt_k_is_sound_and_complete(self):
+        # Returned values are exactly the ones above n//k -- no false positive
+        # (the docstring's reason for the verification pass) and no omission.
+        rng = random.Random(1234)
+        for _ in range(800):
+            n = rng.randint(0, 30)
+            data = [rng.randint(0, 6) for _ in range(n)]
+            for k in (2, 3, 4, 5):
+                got = M.majority_gt_k(data, k)
+                self.assertEqual(sorted(got), brute_gt_k(data, k), (data, k))
+                self.assertEqual(len(got), len(set(got)), "duplicate in the result")
+                self.assertLessEqual(len(got), k - 1)
+
+    def test_gt_k_structure_cases(self):
+        self.assertEqual(M.majority_gt_k([], 3), [])
+        self.assertEqual(M.majority_gt_k([1, 1, 2, 2, 3, 3], 3), [])
+        self.assertEqual(sorted(M.majority_gt_k([1, 1, 1, 2, 2, 3], 3)), [1])
+        # "More than n//k" is strict: at k=2, exactly half of four is not a
+        # majority, so nothing is returned.  (Written the other way first and
+        # the cell failed -- the module was right, the expectation was not.)
+        self.assertEqual(M.majority_gt_k([1, 1, 2, 2], 2), [])
+        self.assertEqual(sorted(M.majority_gt_k([1, 1, 1, 2], 2)), [1])
+
+    def test_third_is_k_three(self):
+        self.assertEqual(M.majority_third([1, 1, 2]), [1])
+        self.assertEqual(M.majority_third([1, 1, 2, 2, 3, 3]), [])
+        self.assertEqual(M.majority_third([1, 1, 1, 2, 3]), [1])
+
+    def test_k_below_two_is_refused(self):
+        for k in (1, 0, -3):
+            with self.assertRaises(ValueError, msg="k=%r" % k):
+                M.majority_gt_k([1, 2], k)
+
+    def test_one_shot_input_is_materialised(self):
+        # `data = list(nums)` is what makes a generator safe; without it the
+        # second pass over `nums` would see nothing.
+        gen = (x for x in [1, 1, 2])
+        self.assertEqual(M.majority_element(gen), 1)
+        gen = (x for x in [1, 1, 2, 2, 3, 3])
+        self.assertEqual(M.majority_gt_k(gen, 3), [])
+
+
+if __name__ == "__main__":
+    unittest.main()

wizard-work/beam/test_oracle_dl.py

@@ -0,0 +1,91 @@
+"""Cells for oracle_dl.py (Q85): the independent oracle for damerau_levenshtein.
+
+`oracle_dl.py` is a CLI with the repo's 0/1/2 contract and no collected cell, so
+its two graded properties -- it agrees with the BFS oracle, and its corpus
+actually separates true Damerau-Levenshtein from OSA and from plain Levenshtein
+-- were only ever measured by hand.  The cells use `run(module, pairs, teeth)`
+with a short corpus so the lane stays fast; the corpus is asserted to be
+deterministic and large separately.
+
+Note measured while writing this, not repaired here (it is beam's file): the
+docstring advertises `--teeth`, but argparse has only `--no-teeth` -- teeth are
+the default -- so `python3 oracle_dl.py --teeth` exits 2 as an unrecognised
+argument.  These cells call `run()` directly and do not depend on the flag.
+"""
+import contextlib
+import io
+import os
+import types
+import unittest
+
+import damerau_levenshtein as DL
+import oracle_dl as O
+
+PAIRS = [("ca", "abc"), ("abc", "acb"), ("ab", "ba"), ("", ""),
+         ("aab", "abca"), ("abcd", "badc"), ("a", "b"), ("abcd", "dcba")]
+
+
+def _module(fn):
+    return types.SimpleNamespace(damerau_levenshtein=fn)
+
+
+class TestOracleDl(unittest.TestCase):
+    def _run(self, module, pairs=PAIRS, teeth=True):
+        out = io.StringIO()
+        with contextlib.redirect_stdout(out):
+            rc = O.run(module, pairs, teeth)
+        return rc, out.getvalue()
+
+    def test_the_shipped_module_agrees_with_the_oracle(self):
+        rc, out = self._run(_module(DL.damerau_levenshtein))
+        self.assertEqual(rc, 0, out)
+        self.assertIn("0 mismatches against the BFS oracle", out)
+
+    def test_the_corpus_catches_osa(self):
+        # The witness Q133-DL was written for: true DL("ca","abc") is 2, OSA
+        # says 3.  A corpus that cannot red an OSA implementation reports green
+        # on a broken module too.
+        rc, out = self._run(_module(O.osa))
+        self.assertEqual(rc, 1, out)
+        self.assertIn("FAIL", out)
+
+    def test_the_corpus_catches_plain_levenshtein(self):
+        rc, out = self._run(_module(O.levenshtein))
+        self.assertEqual(rc, 1, out)
+        self.assertIn("FAIL", out)
+
+    def test_teeth_off_still_grades_the_corpus(self):
+        rc, out = self._run(_module(DL.damerau_levenshtein), teeth=False)
+        self.assertEqual(rc, 0, out)
+        self.assertNotIn("teeth:", out)
+
+    def test_the_full_corpus_is_deterministic_and_large(self):
+        first = O.corpus()
+        self.assertEqual(first, O.corpus(), "the corpus must be the same on every run")
+        self.assertGreater(len(first), 1000)
+        for pair in first[:100]:
+            self.assertEqual(len(pair), 2)
+            self.assertTrue(all(isinstance(s, str) for s in pair), pair)
+
+    def test_a_missing_module_cannot_be_graded(self):
+        err = io.StringIO()
+        with contextlib.redirect_stderr(err), contextlib.redirect_stdout(io.StringIO()):
+            rc = O.main(["--module", os.path.join(os.sep, "nonexistent", "dl.py")])
+        self.assertEqual(rc, 2, err.getvalue())
+        self.assertIn("could not grade", err.getvalue())
+
+    def test_a_module_without_the_entry_point_cannot_be_graded(self):
+        import tempfile
+        with tempfile.TemporaryDirectory(prefix="oracle-dl.") as tmp:
+            path = os.path.join(tmp, "empty_module.py")
+            with open(path, "w", encoding="utf-8") as fh:
+                fh.write("value = 1\n")
+            err = io.StringIO()
+            with contextlib.redirect_stderr(err), contextlib.redirect_stdout(io.StringIO()):
+                rc = O.main(["--module", path])
+        self.assertEqual(rc, 2, err.getvalue())
+        self.assertIn("no damerau_levenshtein()", err.getvalue())
+
+
+if __name__ == "__main__":
+    unittest.main()

wizard-work/beam/test_sha256_manifest.py

@@ -1,6 +1,16 @@
 import hashlib, os, tempfile, unittest, subprocess, sys
 
-MOD = "/root/troupe-standup/wizard-work/beam/sha256_manifest.py"
+HERE = os.path.dirname(os.path.abspath(__file__))
+MOD = os.path.join(HERE, "sha256_manifest.py")
+
+# Binding (Q85, 2026-11-02).  This line read the host-absolute
+# "/root/troupe-standup/wizard-work/beam/sha256_manifest.py", so every cell
+# below graded the host's copy and the committed bytes were graded nowhere --
+# the same defect Q153 repaired in test_next_greater.py, and invisible to
+# check_bindings.py, which parses `sys.path` calls rather than path constants.
+# Resolve beside this file, and assert the resolution instead of trusting it.
+assert os.path.isfile(MOD), "no sha256_manifest.py beside the test: %s" % MOD
+assert os.path.realpath(os.path.dirname(MOD)) == os.path.realpath(HERE), MOD
 
 class T(unittest.TestCase):
     def make_tree(self):

wizard-work/beam/test_tarjan_scc.py

@@ -0,0 +1,84 @@
+"""Cells for tarjan_scc.py (Q85 class): iterative Tarjan over int indices.
+
+The module landed with its proof in `__main__` only -- `self_test()` runs a
+brute-force cross-check and prints ALL TESTS PASSED -- so no collected cell
+read it and `check_coverage.py` graded it UNCOVERED (measured 2026-11-02, the
+module's own commit `7659e22`).  A proof that only runs when a human types the
+file's name is a transcript, not a gate.  These cells port the claim into the
+collected suite: deterministic graphs whose answers can be read by eye, the
+2000-graph brute-force cross-check at a fixed seed, the property that makes the
+output usable (components in reverse topological order), and the two edges the
+`while` loop exists to carry -- a self-loop and a duplicate edge.
+"""
+import random
+import unittest
+
+from tarjan_scc import brute_scc, normalize, tarjan_scc
+
+
+class TestTarjanSCC(unittest.TestCase):
+    def test_empty_and_singletons(self):
+        self.assertEqual(tarjan_scc(0, []), [])
+        self.assertEqual(normalize(tarjan_scc(1, [])), [[0]])
+        self.assertEqual(normalize(tarjan_scc(3, [])), [[0], [1], [2]])
+
+    def test_three_cycle_is_one_component(self):
+        cycles = tarjan_scc(3, [(0, 1), (1, 2), (2, 0)])
+        self.assertEqual(len(cycles), 1)
+        self.assertEqual(sorted(cycles[0]), [0, 1, 2])
+
+    def test_chain_has_no_component_larger_than_one(self):
+        comps = tarjan_scc(3, [(0, 1), (1, 2)])
+        self.assertEqual(normalize(comps), [[0], [1], [2]])
+
+    def test_self_loop_and_duplicate_edge(self):
+        # A self-loop must not fold the node into a neighbour, and a repeated
+        # edge must not push a node twice: the counter-based `while` pass reads
+        # each successor once per adjacency slot, so both are real cells.
+        self.assertEqual(normalize(tarjan_scc(1, [(0, 0)])), [[0]])
+        self.assertEqual(normalize(tarjan_scc(2, [(0, 1), (0, 1)])), [[0], [1]])
+
+    def test_two_components_one_edge(self):
+        # 0 -> 1, 2 -> 3 -> 2: two SCCs, one of size two.
+        comps = normalize(tarjan_scc(4, [(0, 1), (2, 3), (3, 2)]))
+        self.assertEqual(comps, [[0], [1], [2, 3]])
+
+    def test_reverse_topological_order(self):
+        # Tarjan emits a component only after every component it can reach, so
+        # for an edge u -> v crossing components, v's component comes out first.
+        # This is the property callers use to read a condensation order; it is
+        # asserted on random graphs, not just the hand-built one.
+        random.seed(7)
+        for _ in range(40):
+            n = random.randint(2, 9)
+            edges = [(random.randrange(n), random.randrange(n))
+                     for _ in range(random.randint(0, 2 * n))]
+            comps = tarjan_scc(n, edges)
+            where = {node: i for i, comp in enumerate(comps) for node in comp}
+            for u, v in edges:
+                if where[u] != where[v]:
+                    self.assertLess(where[v], where[u], (edges, comps))
+
+    def test_brute_force_cross_check(self):
+        # The module's own claim, moved inside the gate: 2000 random graphs at
+        # seed 42, compared as sets-of-sets against the O(n^3) reachability
+        # closure.  normalize() is the module's, so the comparison is its own
+        # notion of "same partition".
+        random.seed(42)
+        for _ in range(2000):
+            n = random.randint(1, 12)
+            m = random.randint(0, n * n)
+            edges = [(random.randrange(n), random.randrange(n)) for _ in range(m)]
+            self.assertEqual(normalize(tarjan_scc(n, edges)),
+                             normalize(brute_scc(n, edges)), (n, edges))
+
+    def test_only_referenced_nodes_beyond_n_are_not_touched(self):
+        # The signature takes n, so an edge into an index >= n is the caller's
+        # contract violation; the cells above stay inside [0, n).  This one pins
+        # that n=0 with no edges is empty rather than an IndexError on the
+        # `indices[start]` read -- the loop body never runs.
+        self.assertEqual(tarjan_scc(0, []), [])
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

wizard-work/beam/test_toposort_kahn.py

@@ -0,0 +1,98 @@
+"""Cells for toposort_kahn.py (Q85): Kahn's algorithm, deterministic, cycles named.
+
+The module was exercised only by `CHECK_toposort.py`, a file the collector does
+not take (it is named `CHECK_*` and its cells are `def test_dag`-shaped with no
+runner), so the gate never ran it.  These cells port its three graphs and add
+the properties the docstring promises: deterministic tie-breaking by name, and
+a returned cycle whose consecutive pairs are real edges.
+"""
+import random
+import unittest
+
+from toposort_kahn import toposort
+
+
+class TestToposortKahn(unittest.TestCase):
+    def test_chain(self):
+        order, cycle = toposort({"a": ["b"], "b": ["c"], "c": []})
+        self.assertIsNone(cycle)
+        self.assertEqual(order, ["a", "b", "c"])
+
+    def test_diamond(self):
+        graph = {"a": ["b", "c"], "b": ["d"], "c": ["d"], "d": []}
+        order, cycle = toposort(graph)
+        self.assertIsNone(cycle)
+        self.assertEqual(sorted(order), ["a", "b", "c", "d"])
+        self.assertLess(order.index("a"), order.index("b"))
+        self.assertLess(order.index("a"), order.index("c"))
+        self.assertLess(order.index("b"), order.index("d"))
+        self.assertLess(order.index("c"), order.index("d"))
+
+    def test_empty_graph(self):
+        self.assertEqual(toposort({}), ([], None))
+
+    def test_ties_are_broken_by_name(self):
+        # Determinism is the docstring's claim, so the order is asserted, not
+        # just its validity: two roots come out in name order.
+        self.assertEqual(toposort({"b": [], "a": []})[0], ["a", "b"])
+        self.assertEqual(
+            toposort({"m": ["z"], "a": ["z"], "z": []})[0], ["a", "m", "z"])
+
+    def test_node_that_only_appears_as_a_successor(self):
+        order, cycle = toposort({"a": ["b"]})
+        self.assertEqual((order, cycle), (["a", "b"], None))
+
+    def test_two_cycle_is_reported_as_a_cycle(self):
+        graph = {"a": ["b"], "b": ["a"]}
+        order, cycle = toposort(graph)
+        self.assertIsNotNone(cycle)
+        self.assertGreaterEqual(len(cycle), 2)
+        self.assertEqual(cycle[0], cycle[-1], "a cycle list starts and ends at one node")
+        for left, right in zip(cycle, cycle[1:]):
+            self.assertIn(right, graph.get(left, []), "%s -> %s is not an edge" % (left, right))
+
+    def test_self_loop_is_a_cycle(self):
+        order, cycle = toposort({"a": ["a"]})
+        self.assertEqual(cycle, ["a", "a"])
+
+    def test_three_cycle_reports_the_loop(self):
+        graph = {"a": ["b"], "b": ["c"], "c": ["a"]}
+        order, cycle = toposort(graph)
+        self.assertIsNotNone(cycle)
+        self.assertEqual(cycle[0], cycle[-1])
+        self.assertEqual(set(cycle), {"a", "b", "c"})
+
+    def test_random_dags_are_orders(self):
+        rng = random.Random(42)
+        for _ in range(300):
+            n = rng.randint(0, 8)
+            names = ["n%d" % i for i in range(n)]
+            graph = {name: [] for name in names}
+            for i in range(n):
+                for j in range(i + 1, n):
+                    if rng.random() < 0.3:
+                        graph[names[i]].append(names[j])
+            order, cycle = toposort(graph)
+            self.assertIsNone(cycle, graph)
+            self.assertEqual(sorted(order), sorted(names), graph)
+            position = {name: i for i, name in enumerate(order)}
+            for src, succs in graph.items():
+                for dst in succs:
+                    self.assertLess(position[src], position[dst], (src, dst, order))
+
+    def test_a_dag_with_a_cycle_after_pruning_still_reports_it(self):
+        # Kahn's frontier empties even though only part of the graph is blocked.
+        graph = {"a": ["b"], "b": [], "c": ["d"], "d": ["c"]}
+        order, cycle = toposort(graph)
+        self.assertEqual(order, ["a", "b"])
+        self.assertEqual(set(cycle), {"c", "d"})
+
+    def test_input_graph_is_not_mutated(self):
+        graph = {"a": ["b"], "b": []}
+        snapshot = {k: list(v) for k, v in graph.items()}
+        toposort(graph)
+        self.assertEqual({k: list(v) for k, v in graph.items()}, snapshot)
+
+
+if __name__ == "__main__":
+    unittest.main()