tests/test_hook_contract.py

#!/usr/bin/env python3 """What the HARNESS accepts - not what we hoped it accepts. Every check here exists because something shipped that looked right from the inside. The PreCompact hook ran on every compaction for days, produced a correct brief, and had 100% of it thrown away, because it printed the wrong SHAPE. Nothing raised. Nothing logged. The only symptom was a red block on the user's screen that this code could not see. So these tests assert the contract with the outside world, and each one is written so it can fail. Where a test could pass for the wrong reason, it proves the wrong reason is absent first. """ from __future__ import annotations import json import os import shutil import subprocess import sys import tempfile import time from pathlib import Path # Runs against whichever tree it is pointed at, so every build is held to the same # contract rather than to several drifting copies of it. BIN = Path(os.environ.get("RECALL_BIN") or (Path(__file__).resolve().parent.parent / "bin")) HOOK = BIN / "hook.py" HANDOFF = BIN / "handoff.py" def isolated(home: Path, store: Path | None = None) -> dict: """An environment where "the user's home directory" is a temporary one. ⚠️ SETTING HOME IS NOT ENOUGH, AND THE DIFFERENCE IS NOT COSMETIC. On Windows, Path.home() reads USERPROFILE first and ignores HOME entirely. A test that sets only HOME therefore runs the installer against the REAL profile - so running this suite would quietly rewrite the tester's own live ~/.claude/settings.json while the assertions read an empty temp directory and reported failures. A test that edits the machine it is checking is worse than a test that fails. Every variable any platform uses to answer "where is home" is redirected here, plus the two application-data roots the store can land in. """ store = store or (home / "store") return dict( os.environ, HOME=str(home), USERPROFILE=str(home), HOMEDRIVE="", HOMEPATH=str(home), LOCALAPPDATA=str(home / "AppData" / "Local"), APPDATA=str(home / "AppData" / "Roaming"), XDG_DATA_HOME=str(home / "xdg"), RECALL_HOME=str(store), FRIDAY_RECALL_ROOT=str(store), ) passed = failed = 0 def check(label: str, ok: bool, detail: str = "") -> None: global passed, failed if ok: passed += 1 print(f" ok {label}") else: failed += 1 print(f" FAIL {label}") if detail: for line in str(detail).splitlines()[:6]: print(f" {line}") def transcript(path: Path, lines: int = 60) -> Path: """A transcript with enough bulk to be worth briefing, and a rule inside it.""" recs = [{"type": "user", "message": {"role": "user", "content": [ {"type": "text", "text": "no outside servers"}]}}, {"type": "user", "message": {"role": "user", "content": [ {"type": "text", "text": "make sure it works windows 10/11 linux"}]}}] for i in range(lines): recs.append({"type": "assistant", "message": {"role": "assistant", "content": [ {"type": "text", "text": f"step {i} " + "padding " * 40}]}}) path.write_text("\n".join(json.dumps(r) for r in recs), encoding="utf-8") return path def run_hook(event: str, payload: dict, home: Path, cwd: Path) -> subprocess.CompletedProcess: # One test, several builds. Some read RECALL_HOME; this one routes everything # through paths.py and reads FRIDAY_RECALL_ROOT. Setting only one of them # silently tests the developer's real store instead of a temporary directory. env = isolated(home, home) return subprocess.run([sys.executable, str(HOOK), event], input=json.dumps(payload), capture_output=True, text=True, cwd=str(cwd), timeout=180, env=env) def run_handoff(args: list[str], home: Path, cwd: Path) -> subprocess.CompletedProcess: # One test, several builds. Some read RECALL_HOME; this one routes everything # through paths.py and reads FRIDAY_RECALL_ROOT. Setting only one of them # silently tests the developer's real store instead of a temporary directory. env = isolated(home, home) return subprocess.run([sys.executable, str(HANDOFF)] + args, capture_output=True, text=True, cwd=str(cwd), timeout=180, env=env) def main() -> int: tmp = Path(tempfile.mkdtemp(prefix="recall-contract-")) home = tmp / "home" work = tmp / "work" work.mkdir(parents=True) t = transcript(tmp / "t.jsonl") # ── 1. THE SHAPE PreCompact ACTUALLY READS ──────────────────────────────── r = run_hook("pre-compact", {"session_id": "aaaa1111", "transcript_path": str(t), "trigger": "manual"}, home, work) out = r.stdout.strip() check("pre-compact exits 0", r.returncode == 0, r.stderr) check("pre-compact produced output at all", bool(out), "silence here is indistinguishable from a hook that never ran") # The defect, stated as a test. PreCompact has no additionalContext; emitting the # JSON hook envelope makes the harness reject the WHOLE object. is_envelope = False try: d = json.loads(out) is_envelope = isinstance(d, dict) and "hookSpecificOutput" in d except Exception: pass check("pre-compact does NOT emit the JSON hook envelope", not is_envelope, "stdout parsed as {'hookSpecificOutput': ...} - the harness rejects this " "and discards the brief silently") check("pre-compact steer is short enough to be echoed to a terminal", len(out) <= 2600, f"{len(out)}B would wallpaper the screen every compaction") check("pre-compact carries the standing instructions forward", "no outside servers" in out, out[:200]) carry = home / "handoff" / "carry--aaaa1111.md" check("pre-compact wrote the full brief to the carry slot", carry.exists(), f"looked for {carry}") if carry.exists(): # The real property is not "bigger" - a short conversation can brief shorter # than the steer that describes it. It is that the carry is the WHOLE extract, # byte for byte, because the steer is deliberately lossy and something has to # hold the rest. full = subprocess.run([sys.executable, str(BIN / "brief.py"), str(t), "--budget", "12000"], capture_output=True, text=True) check("the carry holds the complete brief, not a trimmed copy", carry.read_text(encoding="utf-8").strip() == full.stdout.strip(), f"carry={len(carry.read_text(encoding='utf-8'))}B brief={len(full.stdout)}B") # ── 2. SessionStart source gating - the cross-session leak ──────────────── # A handoff left by SOMEBODY ELSE, sitting in this folder. hd = home / "handoff" hd.mkdir(parents=True, exist_ok=True) stranger_body = "<!-- recall-handoff project=x session=bbbb2222 written=now -->\nSTRANGER CONTENT" def plant() -> Path: # named for THIS cwd so it is a genuine candidate, not an unrelated file proj = run_handoff(["status"], home, work).stdout import hashlib, re as _re tag = hashlib.sha256(str(work.resolve()).encode()).hexdigest()[:8] base = _re.sub(r"[^A-Za-z0-9._-]", "_", work.name)[:40] f = hd / f"{base}-{tag}--bbbb2222.md" f.write_text(stranger_body, encoding="utf-8") return f f = plant() r = run_hook("session-start", {"session_id": "aaaa1111", "source": "compact"}, home, work) check("compact SessionStart does NOT consume another session's handoff", f.exists(), "this is the 2026-08-21 leak: a session that merely compacted ate a handoff " "left for somebody else, and it can never be given back") check("compact SessionStart does not print the stranger's conversation", "STRANGER CONTENT" not in r.stdout, r.stdout[:300]) check("compact SessionStart DOES inject the carry it is entitled to", "no outside servers" in r.stdout, r.stdout[:300]) check("the carry is consumed once", not carry.exists(), "a carry left in place would replay into every later session") for src in ("resume", "clear", "fork"): run_hook("session-start", {"session_id": "cccc3333", "source": src}, home, work) check(f"{src} SessionStart leaves the handoff alone", f.exists()) # ── 3. startup IS entitled - but must not guess ─────────────────────────── r = run_hook("session-start", {"session_id": "cccc3333", "source": "startup"}, home, work) check("startup SessionStart delivers the single waiting handoff", "STRANGER CONTENT" in r.stdout, r.stdout[:300]) check("and consumes it", not f.exists()) f1 = plant() f2 = hd / f1.name.replace("bbbb2222", "dddd4444") f2.write_text(stranger_body.replace("bbbb2222", "dddd4444").replace( "STRANGER CONTENT", "SECOND STRANGER"), encoding="utf-8") r = run_handoff(["take", "--session", "eeee5555"], home, work) check("two candidates: refuses to guess", "NOT DELIVERED" in r.stdout, r.stdout[:300]) check("two candidates: consumes NEITHER", f1.exists() and f2.exists(), "picking the newest would silently open somebody else's conversation") check("two candidates: names them so a person can choose", "bbbb2222" in r.stdout and "dddd4444" in r.stdout, r.stdout[:300]) # ── 4. a session is never handed back its own notes ─────────────────────── r = run_handoff(["take", "--session", "dddd4444"], home, work) check("its own handoff is not offered back to it, so only one candidate remains", "SECOND STRANGER" not in r.stdout and "STRANGER CONTENT" in r.stdout, r.stdout[:300]) # ── 5. review shows without consuming ───────────────────────────────────── f3 = plant() r = run_handoff(["review", str(t)], home, work) check("review runs", r.returncode == 0, r.stderr[:300]) check("review consumes nothing", f3.exists()) check("review reports the saving in numbers", "smaller" in r.stdout, r.stdout[:300]) check("review shows what would be carried", "no outside servers" in r.stdout, r.stdout[:400]) # ── 6. stale is not injected, and not destroyed either ──────────────────── # Clear the field first. With another fresh candidate still pending this would # take THAT one and the stale path would never be reached - the test would pass # green having measured nothing, which is the failure mode this suite exists for. for leftover in hd.glob("*.md"): if leftover != f3: leftover.unlink() check("stale case is set up with exactly one candidate", len(list(hd.glob("*.md"))) == 1, [x.name for x in hd.glob("*.md")]) old = time.time() - 40 * 3600 os.utime(f3, (old, old)) r = run_handoff(["take", "--session", "ffff6666"], home, work) check("a stale handoff is not injected as if current", "STRANGER CONTENT" not in r.stdout, r.stdout[:300]) check("a stale handoff says so instead of going quiet", "older than" in r.stdout, r.stdout[:300]) check("a stale handoff is left on disk, not deleted", f3.exists(), "an uncollected handoff is evidence that a session ended mid-flight") # ── 6b. a rule must be something the PERSON said ────────────────────────── # A carried summary quotes code and other agents in exactly the same punctuation # it uses for the user. Promoting all of it put a docstring line and another # crew's board post into STANDING INSTRUCTIONS, presented as the user's orders. carried = ( "This session is being continued from a previous conversation.\n" "## 3. Files and Code Sections\n" 'The helper is documented as "Scrub first, THEN shorten. Never the other ' 'way round." and that is deliberate.\n' 'A crew member posted "we must never add a responder to this build" today.\n' "## 6. All user messages:\n" '- "make sure you never call anyone out by name in the README"\n' ) tt = tmp / "carried.jsonl" recs = [{"type": "user", "message": {"role": "user", "content": [ {"type": "text", "text": carried}]}}] for i in range(40): recs.append({"type": "assistant", "message": {"role": "assistant", "content": [ {"type": "text", "text": f"work {i} " + "filler " * 30}]}}) tt.write_text("\n".join(json.dumps(r) for r in recs), encoding="utf-8") b = subprocess.run([sys.executable, str(BIN / "brief.py"), str(tt), "--budget", "12000"], capture_output=True, text=True).stdout standing = b.split("## STANDING INSTRUCTIONS", 1)[-1].split("\n## ", 1)[0] check("a quoted DOCSTRING is not promoted to a standing instruction", "Scrub first" not in standing, standing[:400]) check("another crew's quoted words are not promoted either", "never add a responder" not in standing, standing[:400]) check("the user's own quoted words ARE still promoted", "call anyone out by name" in standing, standing[:400]) # ── 6c. no undefined names anywhere in the package ──────────────────────── # This is here because porting these fixes to the second tree dropped two module # constants and left the function that used them referencing nothing. It would # have raised NameError inside a broad except that logs and carries on - the auto # handoff would simply never have happened, on the exact runs it exists for, and # nothing would have said so. A syntax check does not catch that. This does. lint = subprocess.run([sys.executable, "-m", "pyflakes"] + [str(f) for f in sorted(BIN.glob("*.py"))], capture_output=True, text=True) undefined = [l for l in (lint.stdout + lint.stderr).splitlines() if "undefined name" in l] if "No module named" in lint.stderr: # NOT a failure, and NOT silent either. # # This package promises that Python is its only dependency, so a person # running these tests will usually not have pyflakes. Failing their suite for # a missing developer tool would tell them the product is broken when it is # not, and the very first thing a new user does is run the checks. # Printing nothing would be the other mistake: a check that quietly stops # checking is how the bug it guards against gets back in. print(" -- pyflakes not installed, so UNDEFINED NAMES WERE NOT CHECKED here.") print(" Not a failure - it is a developer tool and this package needs no") print(" dependencies. If you are changing this code: pip install pyflakes") else: check("no undefined names in the package", not undefined, "\n".join(undefined[:5])) # ── 6d. "delete everything" must reach the NEW files too ────────────────── # A carry file is verbatim conversation text. Handoffs were already missed once # by wipe, which left the most readable copy of a session on disk after the user # had been told nothing remained. Every new file that holds words from a # conversation gets added to this check on the day it is written. wipe = BIN / "wipe.py" if not wipe.exists(): print(" -- wipe.py not in this tree, wipe coverage NOT checked here") else: hd.mkdir(parents=True, exist_ok=True) planted = [hd / "carry--zzzz9999.md", hd / ".auto--zzzz9999", hd / "wipeme--zzzz9999.md"] for f in planted: f.write_text("conversation text", encoding="utf-8") w = subprocess.run([sys.executable, str(wipe), "--yes"], capture_output=True, text=True, env=isolated(home, home)) # Assert the command RAN. A rejected argument would leave every file in place # and the check below would be inspecting the aftermath of nothing. check("wipe actually ran", w.returncode == 0, (w.stderr or w.stdout)[:300]) left = [f.name for f in planted if f.exists()] check("wipe removes carry files, auto markers and handoffs", not left, left) # ── 6e. the brief must not hand back the window it just freed ───────────── # Both of these were measured on a real 22 MB session, not imagined. The request # log was treated as "intent, never trimmed" alongside the standing rules, so it # grew to 527 verbatim requests - 107,485 of the brief's 119,031 bytes - and all # of it was injected straight back after the compaction that was supposed to free # the window. The context came down from nearly full to 81% and stopped there. many = tmp / "many.jsonl" recs = [] for i in range(200): recs.append({"type": "user", "message": {"role": "user", "content": [ {"type": "text", "text": f"please do task number {i} for me today"}]}}) recs.append({"type": "user", "message": {"role": "user", "content": [ {"type": "text", "text": "no outside servers"}]}}) # A task-notification ARRIVES AS A USER MESSAGE. Untreated, its fragments became # standing orders the user never gave. recs.append({"type": "user", "message": {"role": "user", "content": [ {"type": "text", "text": "<task-notification><summary>build skills\" finished" "</summary> <note>never stop this agent</note>" "</task-notification>"}]}}) for i in range(40): recs.append({"type": "assistant", "message": {"role": "assistant", "content": [ {"type": "text", "text": f"doing {i} " + "x " * 30}]}}) many.write_text("\n".join(json.dumps(r) for r in recs), encoding="utf-8") b = subprocess.run([sys.executable, str(BIN / "brief.py"), str(many), "--budget", "12000"], capture_output=True, text=True).stdout asked = b.split("## WHAT WAS ASKED", 1)[-1].split("\n## ", 1)[0] n_asks = len([l for l in asked.splitlines() if l.startswith("- ") and "not listed" not in l]) check("the request log is capped instead of handing the window back", n_asks <= 70, f"{n_asks} requests carried") check("and what was dropped is COUNTED, never silently cut", "earlier requests not listed" in asked, asked[:200]) standing = b.split("## STANDING INSTRUCTIONS", 1)[-1].split("\n## ", 1)[0] check("a task-notification does not become a standing order", "never stop this agent" not in standing and "</note>" not in standing, standing[:300]) check("a THREE-word absolute still survives - losing a real rule is the worse error", "no outside servers" in standing, standing[:300]) # ── 6f. text I/O must name its encoding ─────────────────────────────────── # Python on Windows does not default to UTF-8; it uses the machine's legacy code # page. A file written as UTF-8 then fails to read back, and a file that is read # and rewritten can be written back MANGLED. The worst instance was in wipe.py, # which rewrites the user's settings.json - on a Windows machine with an accented # path or a non-English name in that file, the command whose job is to leave the # machine clean would have corrupted their editor configuration. import re as _re bad = [] for f in sorted(BIN.glob("*.py")): for i, line in enumerate(f.read_text(encoding="utf-8").splitlines(), 1): if "encoding=" in line or "sqlite3" in line: continue if _re.search(r"\.(?:read_text|write_text)\(", line) or \ _re.search(r"(?<![\w.])open\(", line): if _re.search(r"['\"][rwab+]*b[rwab+]*['\"]", line): continue # binary mode names no encoding, correctly bad.append(f"{f.name}:{i} {line.strip()[:70]}") check("all text file I/O names its encoding (Windows does not default to UTF-8)", not bad, "\n".join(bad[:6])) # ── 6g. the build must be the same on every machine, before AND after ───── # Two separate ways this was broken, both found by building on a second OS: # · sorted() on Path objects uses the platform's order, and Windows folds case - # so README.md sorted first on one machine and last on the other. Same files, # same bytes, different archive, different checksum. # · the mutation harness read and rewrote source as TEXT, which converts every # line ending on Windows - so merely RUNNING the tests permanently changed four # files and the package checksum with them. # Either one turns "check the sha yourself" into a lie for half our users. bp = BIN.parent / "build.py" if bp.exists(): bsrc = bp.read_text(encoding="utf-8") check("the packaged file order is decided by text, not by platform path rules", "as_posix()" in bsrc and "out.sort(key=" in bsrc, "sorted() on Path objects is case-folded on Windows") mp = Path(__file__).resolve().parent / "mutation_proof.py" if mp.exists(): msrc = mp.read_text(encoding="utf-8") check("the mutation harness restores files BYTE for byte", "read_bytes()" in msrc and "write_bytes(src)" in msrc, "text mode rewrites line endings and changes the package checksum") # ── 7. nothing here reaches the network ─────────────────────────────────── src_all = (HOOK.read_text(encoding="utf-8") + HANDOFF.read_text(encoding="utf-8")) for bad in ("urllib.request", "http.client", "requests", "socket.socket"): check(f"no network: {bad} absent", bad not in src_all) shutil.rmtree(tmp, ignore_errors=True) print(f"\n ---- {passed} passed, {failed} failed ----") return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(main())