tests/test_live_lifecycle.py

#!/usr/bin/env python3 """The whole thing, once, in the order a real day happens. Every other test here checks one part in isolation. This one runs the actual sequence a session goes through - start, work, compact, come back, end, hand over - and asks one question at the end that no unit test can ask: did the instruction the person typed at the beginning survive all the way into the session that replaces this one? That is the product. If that sentence gets lost, nothing else being green matters. It also refuses to be reassured by exit codes. A hook that prints the wrong SHAPE exits 0 and looks perfect from here, which is exactly how a broken PreCompact hook survived in this package for days, so each step is checked against the shape the harness actually consumes. """ from __future__ import annotations import json import os import shutil import subprocess import sys import tempfile from pathlib import Path ROOT = Path(__file__).resolve().parent.parent BIN = ROOT / "bin" HOOK = BIN / "hook.py" # The sentence that has to survive the whole journey. RULE = "never send anything to an outside server" 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, ok, detail=""): global passed, failed if ok: passed += 1 print(f" ok {label}") else: failed += 1 print(f" FAIL {label}") for line in str(detail).splitlines()[:6]: print(f" {line}") def make_transcript(path: Path) -> Path: """A transcript shaped like a real one: prose, tool traffic, thinking blocks.""" recs = [ {"type": "user", "message": {"role": "user", "content": [ {"type": "text", "text": f"build the exporter, and {RULE}"}]}}, {"type": "queue-operation", "operation": "enqueue", "content": "also make sure it works on windows 10/11 and linux"}, ] for i in range(80): recs.append({"type": "assistant", "message": {"role": "assistant", "content": [ {"type": "thinking", "thinking": "", "signature": "S" * 1200}, {"type": "text", "text": f"Working on step {i}. " + "detail " * 25}, {"type": "tool_use", "id": f"t{i}", "name": "Bash", "input": {"command": f"ls -la /some/path/{i}"}}]}}) recs.append({"type": "user", "message": {"role": "user", "content": [ {"type": "tool_result", "tool_use_id": f"t{i}", "content": "output line\n" * 60}]}}) path.write_text("\n".join(json.dumps(r) for r in recs), encoding="utf-8") return path def hook(event: str, payload: dict, env: dict, cwd: Path): 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 valid_session_start(out: str) -> tuple[bool, str]: """SessionStart is a JSON hook. PreCompact is not. Getting these two mixed up is the whole reason this file exists.""" try: d = json.loads(out) except Exception as e: return False, f"not JSON at all: {e}" h = d.get("hookSpecificOutput") if not isinstance(h, dict): return False, "no hookSpecificOutput object" if h.get("hookEventName") != "SessionStart": return False, f"hookEventName is {h.get('hookEventName')!r}" if not isinstance(h.get("additionalContext"), str): return False, "additionalContext is not a string" return True, "" def main() -> int: tmp = Path(tempfile.mkdtemp(prefix="recall-live-")) home, work = tmp / "home", tmp / "work" work.mkdir(parents=True) (home / ".claude").mkdir(parents=True) env = isolated(home) t = make_transcript(tmp / "session-one.jsonl") before = {p for p in tmp.rglob("*")} OLD, NEW = "1111aaaa", "2222bbbb" # ── the day starts ──────────────────────────────────────────────────────── r = hook("session-start", {"session_id": OLD, "source": "startup"}, env, work) check("1. SessionStart(startup) exits 0 on a store with nothing in it", r.returncode == 0, r.stderr[:300]) if r.stdout.strip(): ok, why = valid_session_start(r.stdout) check(" and whatever it emits is the shape SessionStart consumes", ok, why) # ── work happens; the Stop lane captures it ─────────────────────────────── r = hook("stop", {"session_id": OLD, "transcript_path": str(t)}, env, work) check("2. Stop lane exits 0", r.returncode == 0, r.stderr[:300]) # ── the window fills ────────────────────────────────────────────────────── r = hook("pre-compact", {"session_id": OLD, "transcript_path": str(t), "trigger": "auto"}, env, work) steer = r.stdout.strip() check("3. PreCompact exits 0 and says something", r.returncode == 0 and bool(steer), r.stderr[:300]) is_json_envelope = False try: is_json_envelope = "hookSpecificOutput" in json.loads(steer) except Exception: pass check(" PreCompact emits PLAIN TEXT, not the JSON envelope", not is_json_envelope, "the harness rejects the envelope and discards the brief without a word") check(" the steer repeats the standing instruction verbatim", RULE in steer, steer[:300]) check(" the steer is short enough to be echoed at a person", len(steer) <= 2600, f"{len(steer)}B goes to the terminal on every compaction") # ── back from compaction ────────────────────────────────────────────────── r = hook("session-start", {"session_id": OLD, "source": "compact"}, env, work) ok, why = valid_session_start(r.stdout) check("4. SessionStart(compact) emits a valid SessionStart payload", ok, why) ctx = json.loads(r.stdout)["hookSpecificOutput"]["additionalContext"] if ok else "" check(" the full local extract comes back after the compaction", RULE in ctx, ctx[:300]) check(" and it carries more than the steer did", len(ctx) > len(steer), f"ctx={len(ctx)}B steer={len(steer)}B") # ── the session ends; the next one has to inherit ───────────────────────── r = hook("session-end", {"session_id": OLD, "transcript_path": str(t), "reason": "exit"}, env, work) check("5. SessionEnd exits 0", r.returncode == 0, r.stderr[:300]) r = hook("session-start", {"session_id": NEW, "source": "startup"}, env, work) ok, why = valid_session_start(r.stdout) check("6. the NEW session gets a valid SessionStart payload", ok, why) ctx = json.loads(r.stdout)["hookSpecificOutput"]["additionalContext"] if ok else "" # ── the one question that matters ───────────────────────────────────────── check("7. THE INSTRUCTION SURVIVED INTO THE NEXT SESSION", RULE in ctx, "this is the product. Everything else being green is irrelevant if this is red.\n" + ctx[:400]) check(" it is labelled as prior context, not as a fresh order", "PREVIOUS SESSION" in ctx.upper(), ctx[:200]) check(" the queued instruction survived too", "windows 10/11" in ctx.lower(), ctx[:400]) r2 = hook("session-start", {"session_id": "3333cccc", "source": "startup"}, env, work) ctx2 = "" if r2.stdout.strip(): try: ctx2 = json.loads(r2.stdout)["hookSpecificOutput"]["additionalContext"] except Exception: ctx2 = r2.stdout # Check for the HANDOFF BANNER, not for the rule text. The rule legitimately # reappears in the ordinary recent-context block, because the Stop lane captured # it into the store - that is the memory feature working, not a replay. Asserting # on the rule here would have failed for a reason that has nothing to do with # what this line is about, and "fixing" that would have meant weakening a feature. check("8. the handoff is delivered ONCE, not replayed to every later session", "HANDOFF FROM A PREVIOUS SESSION" not in ctx2.upper(), ctx2[:200]) # ── it must not have written anywhere it was not invited ────────────────── allowed = (home / "store", home / ".claude", home / "xdg", tmp / "session-one.jsonl") strays = [p for p in tmp.rglob("*") if p not in before and p.is_file() and not any(str(p).startswith(str(a)) for a in allowed)] check("9. nothing was written outside the store", not strays, [str(x) for x in strays[:5]]) size = t.stat().st_size print(f"\n transcript {size:,}B -> carried {len(ctx):,}B " f"({size / max(len(ctx), 1):.0f}x smaller, 0 model calls, 0 network)") 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())