tests/test_handoff.py

#!/usr/bin/env python3 """Prove the handoff path does what it claims, including the parts that are easy to fake. Four claims are tested, and each test is checked by BREAKING the thing it tests, so a test that could never fail does not get counted as a passing one. 1. A brief is dramatically smaller than its source and keeps every user instruction. 2. A handoff is consumed once. Twice would silently re-inject stale context forever. 3. wipe removes handoff files. They are plain-text conversation and must not survive a wipe that claims to delete everything. 4. Nothing in either file reaches the network. """ from __future__ import annotations import json import os import shutil import subprocess import sys import tempfile from pathlib import Path BIN = Path(__file__).resolve().parent.parent / "bin" FAILED: list[str] = [] def check(name: str, ok: bool, detail: str = "") -> bool: print(f" {'ok ' if ok else 'FAIL'} {name}" + (f" {detail}" if detail else "")) if not ok: FAILED.append(name) return ok def fake_transcript(path: Path, asks: list[str], noise_kb: int = 400) -> None: """A transcript shaped like a real one: a little intent, a lot of machinery.""" lines = [] for i, a in enumerate(asks): lines.append(json.dumps({"type": "user", "uuid": f"u{i}", "message": {"role": "user", "content": a}})) lines.append(json.dumps({"type": "assistant", "uuid": f"a{i}", "message": { "role": "assistant", "content": [ {"type": "thinking", "thinking": "", "signature": "S" * 1600}, {"type": "text", "text": f"Reported conclusion number {i}, stated plainly."}, {"type": "tool_use", "id": f"t{i}", "name": "Bash", "input": {"command": f"grep -rn thing /some/path/{i}"}}]}})) lines.append(json.dumps({"type": "user", "uuid": f"r{i}", "toolUseResult": {"stdout": "X" * (noise_kb * 1024 // len(asks))}, "message": {"role": "user", "content": [ {"type": "tool_result", "tool_use_id": f"t{i}", "content": "X" * (noise_kb * 1024 // len(asks))}]}})) path.write_text("\n".join(lines) + "\n", encoding="utf-8") def run(args, env=None, inp=None): e = dict(os.environ) e.update(env or {}) e["PYTHONPATH"] = str(BIN) return subprocess.run([sys.executable] + args, capture_output=True, text=True, env=e, input=inp, timeout=180) def main() -> int: print("HANDOFF / BRIEF TEST") tmp = Path(tempfile.mkdtemp(prefix="recall-handoff-test-")) try: home = tmp / "store" env = {"RECALL_HOME": str(home), "FRIDAY_RECALL_ROOT": str(home), "RECALL_PROJECT": "unittest"} t = tmp / "t.jsonl" asks = ["remove the thing that calls home", "make sure it works on windows 10/11 and linux", "never list anyone by name in the readme"] fake_transcript(t, asks) src = t.stat().st_size # --- 1. shrink, without losing intent ------------------------------------- r = run([str(BIN / "brief.py"), str(t), "--json"], env) d = json.loads(r.stdout) ratio = d["src_bytes"] / max(1, d["out_bytes"]) check("brief is at least 10x smaller", ratio >= 10, f"{ratio:.0f}x") kept = all(a[:40] in d["brief"] for a in asks) check("every user instruction survives verbatim", kept) check("opaque signatures are dropped", "SSSSSSSS" not in d["brief"]) check("bulk tool output is dropped", "XXXXXXXX" not in d["brief"]) # can this test fail? drop an instruction and it must notice. broken = d["brief"].replace(asks[2][:40], "") check(" (self-check) missing instruction IS detected", not all(a[:40] in broken for a in asks)) # --- 2. one-shot ---------------------------------------------------------- run([str(BIN / "handoff.py"), "write", str(t)], env) first = run([str(BIN / "handoff.py"), "take"], env).stdout second = run([str(BIN / "handoff.py"), "take"], env).stdout check("handoff is delivered once", len(first) > 500, f"{len(first)}B") check("handoff is NOT delivered twice", second.strip() == "", f"{len(second)}B second time") check("handoff states its own age", "written" in first and "ago" in first) # --- 3. wipe takes it ----------------------------------------------------- run([str(BIN / "handoff.py"), "write", str(t)], env) hd = home / "handoff" before = list(hd.glob("*")) if hd.exists() else [] check("handoff files exist before wipe", len(before) > 0, f"{len(before)} file(s)") # Check that wipe RAN. The first version of this test passed an option wipe # does not accept, argparse rejected it, wipe never executed - and the test # then inspected the aftermath of a command that had not happened. Asserting # on the end state without asserting the action succeeded tests nothing. w = run([str(BIN / "wipe.py"), "--yes"], env) check("wipe actually ran", w.returncode == 0, (w.stderr or w.stdout).strip().splitlines()[-1][:90] if w.returncode else "") after = [p for p in (hd.glob("*") if hd.exists() else [])] check("wipe removes every handoff file", not after, "left: " + ", ".join(p.name for p in after) if after else "") # --- 4. the core survives a brutal budget --------------------------------- # The point of a budget is that something gets cut. What must NEVER be cut is # the part that is still binding, so this squeezes hard and checks the rules # are still there. rules_src = ["never publish anything to an outside server", "make sure it works on windows and linux"] t2 = tmp / "t2.jsonl" fake_transcript(t2, rules_src + ["do some unrelated task " + "z" * 50], noise_kb=600) tiny = json.loads(run([str(BIN / "brief.py"), str(t2), "--budget", "1500", "--json"], env).stdout)["brief"] kept = [r for r in rules_src if r[:28] in tiny] check("standing rules survive a 1500-char budget", len(kept) == len(rules_src), f"{len(kept)}/{len(rules_src)} kept, brief is {len(tiny)}B") # Force a real cut: budget well under the full brief. Asserting "NOT INCLUDED # appears" without making sure anything was actually dropped tests nothing - # the first version of this check passed a budget the brief already fit inside. full = json.loads(run([str(BIN / "brief.py"), str(t2), "--json"], env).stdout)["brief"] squeezed = json.loads(run([str(BIN / "brief.py"), str(t2), "--budget", str(max(300, len(full) // 3)), "--json"], env).stdout)["brief"] check("the squeeze really removed something", len(squeezed) < len(full), f"{len(full)}B -> {len(squeezed)}B") check("what got cut is declared, not silent", "NOT INCLUDED" in squeezed) # --- 5. two modes --------------------------------------------------------- run([str(BIN / "handoff.py"), "write", str(t2)], env) run([str(BIN / "handoff.py"), "mode", "seed"], env) seed = run([str(BIN / "handoff.py"), "take", "--peek"], env).stdout run([str(BIN / "handoff.py"), "mode", "pointer"], env) ptr = run([str(BIN / "handoff.py"), "take", "--peek"], env).stdout check("pointer mode is much smaller than seed", len(ptr) < len(seed) * 0.5, f"seed {len(seed)}B vs pointer {len(ptr)}B") still_bound = [r for r in rules_src if r[:28] in ptr] check("pointer mode STILL carries the standing rules", len(still_bound) == len(rules_src), f"{len(still_bound)}/{len(rules_src)}") check("pointer mode says how to get the rest", "handoff show" in ptr) check("mode choice is remembered", run([str(BIN / "handoff.py"), "mode"], env).stdout.strip().endswith("pointer")) # --- 6. queued instructions are not lost ---------------------------------- # Messages typed while the assistant is working are stored as their own record # type with the text at top level. Reading only "message" misses them, which # for anyone who works by queueing is most of what they said. t3 = tmp / "t3.jsonl" lines = [json.dumps({"type": "user", "uuid": "u", "message": {"role": "user", "content": "start the job"}}), json.dumps({"type": "queue-operation", "operation": "enqueue", "content": "never send anything off this machine"}), json.dumps({"type": "queue-operation", "operation": "enqueue", "content": "<task-notification>ignore me</task-notification>"})] t3.write_text("\n".join(lines) + "\n", encoding="utf-8") b3 = json.loads(run([str(BIN / "brief.py"), str(t3), "--json"], env).stdout)["brief"] check("queued instruction is captured", "never send anything off this machine" in b3) check("queued task-notification is ignored", "ignore me" not in b3) # --- 7. it never adds anything it did not see ----------------------------- # The whole value of a locally extracted brief is that it is EXTRACTION, not # authorship. No model writes it, so nothing can be inferred, smoothed over or # invented. This checks that literally: every bullet in the brief must appear # in the source transcript. If it is not in there, it does not go in. corpus = " ".join(t.read_text(encoding="utf-8").split()) full2 = json.loads(run([str(BIN / "brief.py"), str(t), "--json"], env).stdout)["brief"] # One section is DERIVED rather than quoted: command shapes are normalised # (quoted arguments, long paths and numbers are replaced) so that many similar # runs collapse into one line with a count. Those strings are deliberately not # verbatim, and the brief now says so in the heading. Everything else must # trace back to the transcript word for word. invented, derived_section = [], False for line in full2.splitlines(): if line.startswith("## "): derived_section = line.startswith("## COMMANDS, COLLAPSED BY SHAPE") continue if derived_section or not line.startswith("- "): continue claim = line[2:].split(" [x")[0].split(" -> ")[0].strip() claim = claim.split(" ...")[0].strip() if len(claim) < 25 or claim.startswith(("W ", "(read only")): continue probe = claim[:45] if probe not in corpus: invented.append(probe) check("nothing in the brief was invented", not invented, f"not found in source: {invented[:2]}" if invented else "every bullet traced back to the transcript") # can this check fail? plant a line that is not in the source. planted = full2 + "\n- this sentence was never in any transcript anywhere at all" found_planted = [l[2:] for l in planted.splitlines() if l.startswith("- ") and l[2:47] not in corpus and len(l) > 27] check(" (self-check) an invented line WOULD be caught", bool(found_planted)) # --- 8. one project's handoff never lands in another ---------------------- # Identity came from the directory's BASE NAME at first, so two unrelated # projects each containing a "src" folder shared one handoff - meaning a # conversation about one job could open inside a different job. A handoff is # real conversation text, so that is a leak, not a cosmetic clash. import subprocess as _sp a_dir = tmp / "projA" / "src" b_dir = tmp / "projB" / "src" a_dir.mkdir(parents=True); b_dir.mkdir(parents=True) def project_in(d): e = dict(os.environ); e.update(env); e.pop("RECALL_PROJECT", None) e["PYTHONPATH"] = str(BIN) return _sp.run([sys.executable, "-c", "import sys;sys.path.insert(0,r'%s');import handoff;" "print(handoff._project())" % BIN], capture_output=True, text=True, cwd=str(d), env=e).stdout.strip() pa, pb = project_in(a_dir), project_in(b_dir) check("same-named folders get DIFFERENT handoff identities", pa != pb, f"{pa} vs {pb}") check("the same folder is stable across runs", project_in(a_dir) == pa) check("the identity is still human-readable", pa.startswith("src-"), pa) # --- 9. no network -------------------------------------------------------- src_text = ((BIN / "brief.py").read_text(encoding="utf-8") + (BIN / "handoff.py").read_text(encoding="utf-8")) # Match how these are USED, not the bare words. Matching the words alone made # this fire on an ordinary English comment - the word "requests" meaning the # things a person asked for - and a check that goes red over prose is a check # people learn to wave through. It still catches every real import or call. import re as _re _net_pat = _re.compile( r"(?:^|\n)\s*(?:import|from)\s+(?:socket|urllib|http\.client|requests)\b" r"|\bsocket\.socket\s*\(" r"|\burllib\.request\b" r"|\brequests\.(?:get|post|put|delete|head|patch|Session)\s*\(" r"|\burlopen\s*\(") net = sorted({m.group(0).strip() for m in _net_pat.finditer(src_text)}) check("no network machinery in brief/handoff", not net, ", ".join(net)) finally: shutil.rmtree(tmp, ignore_errors=True) print() if FAILED: print("FAILED: " + ", ".join(FAILED)) return 1 print("all handoff checks passed") return 0 if __name__ == "__main__": raise SystemExit(main())