diff --git a/wizard-work/beam/_run_tests.py b/wizard-work/beam/_run_tests.py
new file mode 100644
index 0000000..39924cb
--- /dev/null
+++ b/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')
diff --git a/wizard-work/beam/dedup.py b/wizard-work/beam/dedup.py
new file mode 100644
index 0000000..dc7f8ea
--- /dev/null
+++ b/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
[--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())
diff --git a/wizard-work/beam/test_counterparty_sig.py b/wizard-work/beam/test_counterparty_sig.py
new file mode 100644
index 0000000..12b24b2
--- /dev/null
+++ b/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()
diff --git a/wizard-work/beam/test_dedup.py b/wizard-work/beam/test_dedup.py
new file mode 100644
index 0000000..9367e6f
--- /dev/null
+++ b/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()
diff --git a/wizard-work/beam/test_handle_resolve.py b/wizard-work/beam/test_handle_resolve.py
new file mode 100644
index 0000000..ac839b8
--- /dev/null
+++ b/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()
diff --git a/wizard-work/beam/test_largest_rect_hist.py b/wizard-work/beam/test_largest_rect_hist.py
index 626c568..d78e199 100644
--- a/wizard-work/beam/test_largest_rect_hist.py
+++ b/wizard-work/beam/test_largest_rect_hist.py
@@ -1,6 +1,15 @@
-import random, sys
-sys.path.insert(0, "/root/.hermes/wizard-outputs/beam")
+import pathlib, random, sys
+
+# 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)
diff --git a/wizard-work/beam/test_majority.py b/wizard-work/beam/test_majority.py
new file mode 100644
index 0000000..e5d3977
--- /dev/null
+++ b/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()
diff --git a/wizard-work/beam/test_oracle_dl.py b/wizard-work/beam/test_oracle_dl.py
new file mode 100644
index 0000000..3872d4e
--- /dev/null
+++ b/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()
diff --git a/wizard-work/beam/test_sha256_manifest.py b/wizard-work/beam/test_sha256_manifest.py
index d2937d1..f076fdb 100644
--- a/wizard-work/beam/test_sha256_manifest.py
+++ b/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):
diff --git a/wizard-work/beam/test_tarjan_scc.py b/wizard-work/beam/test_tarjan_scc.py
new file mode 100644
index 0000000..057ba6f
--- /dev/null
+++ b/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)
diff --git a/wizard-work/beam/test_toposort_kahn.py b/wizard-work/beam/test_toposort_kahn.py
new file mode 100644
index 0000000..4ef047d
--- /dev/null
+++ b/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()