diff --git a/README.md b/README.md index 38ab2be6..d4fcf849 100644 --- a/README.md +++ b/README.md @@ -62,9 +62,18 @@ curl -fsSL https://raw.githubusercontent.com/XTraceAI/agent-plugins/main/install This installs MemHub into each of Claude Code, Codex and Cursor it finds on `PATH` (`sh -s -- --agents claude,codex` for a subset), disables any other -enabled MemHub copy in Claude Code so sessions are not captured twice, installs -the Codex hooks bridge, and signs the machine in once — every agent's capture -hooks read the same credential. It is safe to re-run. It cannot do the steps +enabled MemHub copy in Claude Code so sessions are not captured twice, turns on +Claude Code's marketplace auto-update and its fork agent (`--no-auto-update` / +`--no-fork-agent` leave either as it is), installs the Codex hooks bridge, and +signs the machine in once — every agent's capture hooks read the same +credential. On a first install it asks two things when a terminal is attached: +where MemHub is active (only repositories you add later with +`/memhub:repos add .`, every repository, or a list now) and whether to show the +companion animal (off by default; a one-line status strip shows instead). +`--selected-repos`, `--all-repos`, `--repos org/a,org/b`, `--companion` and +`--no-companion` answer from the command line (after `sh -s --`); unattended, it +picks selected repositories (none yet) and no animal. It is safe to re-run. It +cannot do the steps only the host UI can: restarting, trusting the Codex hooks in `/hooks`, and selecting **Add** in Cursor; it prints those at the end. The per-host steps below are the same install done by hand. diff --git a/tests/claude_hook_guard_test.py b/tests/claude_hook_guard_test.py index 10a5292f..6c0e16d7 100644 --- a/tests/claude_hook_guard_test.py +++ b/tests/claude_hook_guard_test.py @@ -80,11 +80,16 @@ def test_every_claude_handler_is_guarded_and_only_boundaries_capture(): for group in groups: for handler in group["hooks"]: commands.append((event, handler["command"])) - assert len(commands) == 21 # - the three retired directive-recall + assert len(commands) == 22 # + PostToolUseFailure (rulebook_hook.py post — + # the failure lane, ENG-1236), + # + UserPromptExpansion (rulebook_hook.py + # expansion — a typed command noted, ENG-1235), + # - the three retired directive-recall # handlers (PreToolUse ×2, PostToolUse), - # + PreToolUse (add_memory_gate.py, and - # create_rule_origin.py — the session a rule - # is filed from), + # - the retired add_memory_gate.py (the + # server removed add_memory), + # + PreToolUse (create_rule_origin.py — the + # session a rule is filed from), # + UserPromptSubmit (brain_brief.py prompt, # and rulebook_hook.py prompt — the lane that # arms a prompt-armed obligation), diff --git a/tests/codex_rulebook_context_test.py b/tests/codex_rulebook_context_test.py index 058d5fb9..8f327611 100644 --- a/tests/codex_rulebook_context_test.py +++ b/tests/codex_rulebook_context_test.py @@ -50,10 +50,14 @@ def test_normalized_desktop_payload_and_explicit_context(): assert 'fixture rule fired' in output.getvalue(), output.getvalue() rows = [json.loads(x) for x in (base/'state/ledger/fires.jsonl').read_text().splitlines()] assert len(rows) == 1 and rows[0]['host'] == 'codex' - for cmd in ['echo x; '+command, 'cd '+shlex.quote(str(repo))+'; echo rulebook-probe', - 'cd "$TARGET" && echo rulebook-probe']: + # The call's repo is where its FIRST command runs: a command before the + # cd runs in the (non-repo) cwd, and an unset variable names nothing. + for cmd in ['echo x; '+command, 'cd "$TARGET_UNSET_RB" && echo rulebook-probe']: payload['tool_input']['command'] = cmd assert rb.repo_of_call(payload)[0] == '' + # A `cd X;` chain runs the next command in X, like `cd X &&`. + payload['tool_input']['command'] = 'cd '+shlex.quote(str(repo))+'; echo rulebook-probe' + assert rb.repo_of_call(payload)[1] == str(repo) # The raw exec_command shape remains supported. payload.update(tool_name='exec_command', tool_input={'cmd':'echo probe','workdir':str(repo)}) normalized=json.loads(bridge._rulebook_payload(json.dumps(payload).encode())) @@ -83,9 +87,7 @@ def test_bundled_route_defers_only_for_installed_matching_user_handler(): assert not bridge._user_bridge_handles('PreToolUse',b'{"tool_name":"Bash"}') -def test_the_package_selects_codex_hooks_and_gives_a_projectless_notice(): - # Was asserted against plugins/memhub-staging until the staging build moved - # to agent-plugins-internal; the shipped package makes the same promise. +def test_staging_selects_codex_hooks_and_projectless_notice(): manifest=json.loads((ROOT/'plugins/memhub/.codex-plugin/plugin.json').read_text()) assert manifest['hooks']=='./hooks/codex-hooks.json' assert manifest['name']=='memhub' diff --git a/tests/cursor_capture_test.py b/tests/cursor_capture_test.py index ff671c5a..e5205b5c 100644 --- a/tests/cursor_capture_test.py +++ b/tests/cursor_capture_test.py @@ -118,8 +118,13 @@ def test_cursor_manifest_uses_one_portable_launcher_per_event(): .read_text(encoding="utf-8")) assert set(document["hooks"]) == { "beforeShellExecution", "afterShellExecution", "afterFileEdit", - "afterAgentResponse", "stop", "beforeSubmitPrompt", "sessionEnd" + "afterAgentResponse", "stop", "beforeSubmitPrompt", "sessionEnd", + # The rulebook lane (cursor_rulebook.py). + "sessionStart", "preToolUse", "postToolUse", "beforeReadFile", } + # Only the Read/Task calls the rulebook lane tracks, not every tool call. + assert document["hooks"]["preToolUse"][0]["matcher"] == "Read|Task" + assert document["hooks"]["postToolUse"][0]["matcher"] == "Task" for event, handlers in document["hooks"].items(): assert len(handlers) == 1, event command = handlers[0]["command"] @@ -515,6 +520,64 @@ def _res(structured=None, texts=(), is_error=False): print("PASS test_import_verdicts_and_dormancy") +def test_live_store_falls_back_to_the_transcript(): + """A store that exists but cannot be READ must not win the source choice. + + cursor-agent holds its chat store open for the life of the session, so at + sessionEnd — the only hook that captures a CLI session — sqlite refuses the + read-only open the reader uses. locate() names the file regardless, so the + flusher used to pick "store", fail to open it, and skip: no cursor-agent + session ever reached MemHub, with a complete transcript sitting beside it. + """ + import sqlite3 + from readers import cursor as cursor_reader + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + uuid = "019c6e48-b66c-7881-9301-99c87fc66cf6" + store = root / "store.db" + con = sqlite3.connect(store) + con.execute("create table blobs (id text)") + con.commit() + con.close() + assert cursor_reader.readable(store) # closed: readable + transcript = root / uuid / f"{uuid}.jsonl" + transcript.parent.mkdir() + transcript.write_text("{}\n", encoding="utf-8") + + located, readable = [store, ""], [True] + original_locate = cursor_reader.locate + original_readable = cursor_reader.readable + original_valid = cursor_flush._valid_transcript_path + cursor_reader.locate = lambda ref: (located[0], located[1]) + cursor_reader.readable = lambda path: readable[0] + cursor_flush._valid_transcript_path = lambda raw, u, root=None: ( + (transcript, "") if raw else (None, "no transcript_path")) + try: + payload = {"transcript_path": str(transcript)} + # readable store: unchanged behavior, the richer source still wins + kind, path, err = cursor_flush._source_for(uuid, payload, {}) + assert (kind, path) == ("store", store), (kind, path, err) + # live store: the validated transcript is used instead of skipping + readable[0] = False + kind, path, err = cursor_flush._source_for(uuid, payload, {}) + assert (kind, path) == ("transcript", transcript), (kind, path, err) + # neither readable nor a transcript: still an error, never a guess + kind, path, err = cursor_flush._source_for(uuid, {}, {}) + assert kind is None and path is None and err + # a STICKY store stays sticky: switching sources mid-session would + # renumber every record uuid, which is why only the first choice + # consults readability. + kind, path, err = cursor_flush._source_for( + uuid, payload, {"source_kind": "store"}) + assert kind == "store" and path == store + finally: + cursor_reader.locate = original_locate + cursor_reader.readable = original_readable + cursor_flush._valid_transcript_path = original_valid + print("PASS test_live_store_falls_back_to_the_transcript") + + if __name__ == "__main__": for name, fn in sorted(globals().items()): if name.startswith("test_") and callable(fn): diff --git a/tests/documentation_test.py b/tests/documentation_test.py index 7ce5e8c7..959bb5cc 100644 --- a/tests/documentation_test.py +++ b/tests/documentation_test.py @@ -105,14 +105,14 @@ def test_the_finder_collects_facts_from_the_pr_it_was_given() -> None: """`gh pr view ` resolves the number against the CURRENT checkout, so a URL for another repo would have ranked sessions against a different pull request and then linked them to the one the user named (Codex, #182).""" - base = ROOT / "plugins" / "memhub" / "skills" / "find-contributing-sessions" + base = ROOT / "plugins" / "memhub" / "skills" / "link-pr" skill = (base / "SKILL.md").read_text(encoding="utf-8") scanner = (base / "scripts" / "find_sessions.py").read_text(encoding="utf-8") flat = " ".join(skill.split()) # Since ENG-1128 the scanner collects the facts itself (batch mode), so the # property moved into the script: step 1 still resolves a bare NUMBER to a # URL, and the facts are read from that URL. - facts = skill[skill.index("## 2. Collect the facts"):skill.index("## 3. One consolidated")] + facts = skill[skill.index("### 3.2 Collect the facts"):skill.index("### 3.3 One consolidated")] check("the skill hands the scanner URLs, not numbers", "--prs-from" in facts and "**by URL**" in facts) check("…and it says why", "resolves the number against the CURRENT checkout" in flat) @@ -124,16 +124,17 @@ def test_the_finder_collects_facts_from_the_pr_it_was_given() -> None: def test_the_finder_maps_my_unlinked_prs() -> None: """ENG-1128: no argument maps the caller's own unlinked PRs through the - backend's list tool, `--pr` names specific ones, and without a linked + backend's list tool, `--find ...` names specific ones, and without a linked GitHub identity the user's own `gh` is the fallback.""" - skill = (ROOT / "plugins" / "memhub" / "skills" / "find-contributing-sessions" + skill = (ROOT / "plugins" / "memhub" / "skills" / "link-pr" / "SKILL.md").read_text(encoding="utf-8") flat = " ".join(skill.split()) - for text in ("list_my_unlinked_prs(limit=25)", "--pr ", + for text in ("list_my_unlinked_prs(limit=25)", "--find [pr-or-url ...]", "gh search prs --author=@me", "`github_identity_linked: false`", "`github_connected: false`", "legacy_branch_sessions", - "Link nothing without an explicit yes", "Never send `pr_type`"): - check(f"find-contributing-sessions says {text!r}", text in flat) + "Link nothing without an explicit yes", "Never send `pr_type` from `--find`", + 'link_source="session_found"'): + check(f"link-pr --find says {text!r}", text in flat) frontmatter = skill.split("---", 2)[1] for spelling in ("mcp__plugin_memhub_memhub__list_my_unlinked_prs", "mcp__plugin_memhub_memhub__list_my_unlinked_prs"): @@ -250,16 +251,21 @@ def test_create_rule_skill_keeps_its_authoring_gates() -> None: # cp/git/python3/rm can still fire on the parent's shell. "must be the sub-agent's rather than the parent's", # ENG-1195: one rulebook per scope. Step 0 decides who the - # rule applies to without asking: org-wide unless the person - # says "just me" or names a workspace, and the workspace - # scope is never offered unprompted. + # rule applies to without asking, and the workspace scope is + # never offered unprompted. ENG-1243: personal unless the + # person puts it on everyone or the agent can say in one + # sentence why anyone working in the repo or org needs it. "## 0. Who it applies to", - "**Otherwise use `org`.**", + "| `personal` (**the default**) |", + "**Otherwise use `personal`**", + "why\n the rule is right for **anyone**", + "with that one\nsentence", "never offer the workspace scope", "`scope` from step 0", - # the harness draft path files org-wide and never looks up - # or makes a book - 'always files with `scope: "org"`', + # the harness draft path files personal too, says its org + # reason in `why`, and never looks up or makes a book + 'a harness draft files with\n `scope: "personal"`', + "(`Org-wide because …`)", # the report names who the rule reaches, never a stored # book name "never a\nrulebook's stored name"): @@ -307,7 +313,7 @@ def part(text: str, start: str, end: str) -> str: "inherits what it does not name" in file_ and "re-state `when`" in file_) harness = (CREATE_RULE / "references" / "harness.md").read_text(encoding="utf-8") check("the harness draft's call carries `when`, `do`, `why`", - '`scope: "org"`, `title`, `statement`, `when`, `do`, `why`' in harness) + '`scope` (step 0 above), `title`, `statement`, `when`, `do`, `why`' in harness) check("…written without asking", "`when`, `do` and `why` are written here without asking" in harness) start = (skills / "start-rulebook" / "SKILL.md").read_text(encoding="utf-8") @@ -333,13 +339,14 @@ def part(text: str, start: str, end: str) -> str: # shipped file that still names a removed tool (or the old `memory_type` # parameter) teaches the agent a call the server refuses. ENG-1195 retired # `create_rulebook` (rulebooks are made per scope by the server; the backend -# keeps a deprecated shim for older plugins), so the plugin's surface is 33. +# keeps a deprecated shim for older plugins), so the plugin's surface is 33; +# the backend then removed `add_memory`, so it is 32. _MCP_SURFACE = { "list_agent_brains", "get_brain_overview", "search_memory", "read_memory", "get_artifact_lineage", "list_tags", "save_artifact", "ingest_document_from_url", "tag_memory", "copy_to_agent_brain", "delete_memory", "create_agent_brain", "share_agent_brain", "list_agent_brain_access", "list_brain_folders", - "move_brain_to_folder", "list_sessions", "add_memory", "import_conversation", + "move_brain_to_folder", "list_sessions", "import_conversation", "list_rulebooks", "list_rules", "create_rule", "link_pr", "unlink_pr", "list_my_unlinked_prs", "get_spec_workflow", "list_skills", "get_skill", "create_skill", "list_orgs", "list_workspaces", "create_workspace", @@ -354,8 +361,8 @@ def part(text: str, start: str, end: str) -> str: ) -def test_mcp_surface_is_33_tools() -> None: - check("the surface this test pins is 33 tools", len(_MCP_SURFACE) == 33) +def test_mcp_surface_is_32_tools() -> None: + check("the surface this test pins is 32 tools", len(_MCP_SURFACE) == 32) def test_shipped_files_name_no_removed_mcp_tool() -> None: diff --git a/tests/find_sessions_test.py b/tests/find_sessions_test.py index 8e5e4f10..aa339e82 100644 --- a/tests/find_sessions_test.py +++ b/tests/find_sessions_test.py @@ -25,7 +25,7 @@ from pathlib import Path ROOT = Path(__file__).resolve().parents[1] -SCRIPT = (ROOT / "plugins" / "memhub" / "skills" / "find-contributing-sessions" +SCRIPT = (ROOT / "plugins" / "memhub" / "skills" / "link-pr" / "scripts" / "find_sessions.py") SCRIPTS = ROOT / "plugins" / "memhub" / "scripts" @@ -495,6 +495,390 @@ def test_git_add_and_commit_are_found_by_tokens_not_at_command_position(): find_sessions._git_path_arguments("git log --oneline") == []) +# --- batch mode (ENG-1128): many PRs, facts read with gh, one pass --------- +# +# A fake `gh` on PATH answers from a JSON file and logs every call, so the +# suite needs neither the real CLI nor the network. + +_FAKE_GH = '''\ +import json, os, sys +args = sys.argv[1:] +with open(os.environ["FAKE_GH_LOG"], "a", encoding="utf-8") as log: + log.write(json.dumps(args) + "\\n") +db = json.load(open(os.environ["FAKE_GH_DB"], encoding="utf-8")) +if args[:2] == ["pr", "view"]: + reply = db.get("view", {}).get(args[2]) +elif args[:1] == ["api"]: + reply = db.get("api", {}).get(args[1]) +else: + reply = None +if reply is None or reply == "FAIL": + sys.stderr.write("gh: secret-ish diagnostic that must not be echoed\\n") + sys.exit(1) +sys.stdout.write(reply if isinstance(reply, str) else json.dumps(reply)) +''' + + +def _pr_view(url: str, files: list[str], head: str, base: str = "main", + shas: list[str] = (), created: str = "2026-09-10T00:00:00Z", + title: str = "a pull request") -> dict: + return {"url": url, "title": title, "headRefName": head, "baseRefName": base, + "createdAt": created, "files": [{"path": f} for f in files], + "commits": [{"oid": s} for s in shas]} + + +def _fake_gh(td: str, db: dict) -> tuple[dict, Path]: + """PATH with a fake `gh` first, plus the env it reads; returns (env, log).""" + bin_dir = Path(td) / "bin" + bin_dir.mkdir(exist_ok=True) + gh = bin_dir / "gh" + gh.write_text(f"#!{sys.executable}\n" + _FAKE_GH, encoding="utf-8") + gh.chmod(0o755) + db_path = Path(td) / "gh-db.json" + db_path.write_text(json.dumps(db), encoding="utf-8") + log = Path(td) / "gh-log.jsonl" + log.write_text("", encoding="utf-8") + env = {"PATH": f"{bin_dir}{os.pathsep}{os.environ.get('PATH', '')}", + "FAKE_GH_DB": str(db_path), "FAKE_GH_LOG": str(log)} + return env, log + + +def run_batch(home: Path, env_extra: dict, *args: str, + prs_file_lines: list[str] | None = None) -> tuple[int, str, str]: + extra: list[str] = [] + listing = None + if prs_file_lines is not None: + with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False, + encoding="utf-8") as handle: + handle.write("\n".join(prs_file_lines) + "\n") + listing = handle.name + extra = ["--prs-from", listing] + try: + proc = subprocess.run( + [sys.executable, str(SCRIPT), *extra, *args], + capture_output=True, text=True, encoding="utf-8", timeout=120, + env={**os.environ, "HOME": str(home), "USERPROFILE": str(home), + "PYTHONUTF8": "1", "MEMHUB_PLUGIN_SCRIPTS": str(SCRIPTS), + **env_extra}) + return proc.returncode, proc.stdout, proc.stderr + finally: + if listing: + os.unlink(listing) + + +def _gh_calls(log: Path) -> list[list[str]]: + return [json.loads(line) for line in log.read_text(encoding="utf-8").splitlines() + if line.strip()] + + +PR_A = "https://github.com/acme/repo/pull/7" +PR_B = "https://github.com/acme/repo/pull/8" + + +def test_batch_ranks_each_session_under_the_pr_it_wrote(): + with tempfile.TemporaryDirectory() as td: + home = Path(td) / "home" + wt = str(Path(td) / "checkouts" / "repo") + claude_session(home, "writer-a", wt, branch="feat/a", + edits=[f"{wt}/app/a1.py", f"{wt}/app/a2.py"]) + claude_session(home, "writer-b", wt, branch="feat/b", + edits=[f"{wt}/app/b1.py", f"{wt}/app/b2.py"]) + # Edited one file of each PR on neither branch: shows under BOTH. + claude_session(home, "both", wt, branch="scratch", + edits=[f"{wt}/app/a1.py", f"{wt}/app/b1.py", f"{wt}/app/shared.py"]) + env, log = _fake_gh(td, {"view": { + PR_A: _pr_view(PR_A, ["app/a1.py", "app/a2.py", "app/shared.py"], "feat/a", + title="feat: the a side"), + PR_B: _pr_view(PR_B, ["app/b1.py", "app/b2.py"], "feat/b")}}) + rc, out, err = run_batch(home, env, "--pr-url", PR_A, "--pr-url", PR_B, + "--host", "claude") + check("batch exits 0 with a JSON object", rc == 0 and out.strip().startswith("{"), + out + err) + doc = json.loads(out) if rc == 0 else {"prs": []} + by_url = {p["pr_url"]: p for p in doc["prs"]} + check("both PRs come back, in input order", + [p["pr_url"] for p in doc["prs"]] == [PR_A, PR_B], out[:300]) + a_ids = [c["conversation_id"] for c in by_url.get(PR_A, {}).get("candidates", [])] + b_ids = [c["conversation_id"] for c in by_url.get(PR_B, {}).get("candidates", [])] + check("PR A's writer leads PR A", a_ids[:1] == ["writer-a"], repr(a_ids)) + check("PR B's writer leads PR B", b_ids[:1] == ["writer-b"], repr(b_ids)) + check("PR A's writer is not a candidate for PR B", "writer-a" not in b_ids, + repr(b_ids)) + check("a session that touched both PRs is a candidate for both", + "both" in a_ids and "both" in b_ids, repr((a_ids, b_ids))) + check("the PR's own metadata rides along", + by_url.get(PR_A, {}).get("title") == "feat: the a side" + and by_url[PR_A]["repo"] == "acme/repo" and by_url[PR_A]["number"] == 7 + and by_url[PR_A]["head_ref"] == "feat/a", out[:400]) + check("candidate rows keep the single-mode shape", + by_url.get(PR_A, {}).get("candidates") + and set(by_url[PR_A]["candidates"][0]) == { + "conversation_id", "session_id", "host", "cwd", "mtime", "score", + "evidence"}, out[:400]) + check("no transcript prose in batch output", PROSE not in out) + check("sessions_scanned counts parsed sessions", doc.get("sessions_scanned") == 3, + repr(doc.get("sessions_scanned"))) + views = [c for c in _gh_calls(log) if c[:2] == ["pr", "view"]] + check("gh pr view is given the URL, never a bare number", + sorted(c[2] for c in views) == sorted([PR_A, PR_B]), repr(views)) + + +def test_batch_parses_each_transcript_once(): + """The point of batching: 25 PRs over 200 sessions must not read every + transcript 25 times.""" + with tempfile.TemporaryDirectory() as td: + home = Path(td) / "home" + wt = str(Path(td) / "repo") + for index in range(3): + claude_session(home, f"s{index}", wt, branch="feat/a", + edits=[f"{wt}/app/a1.py"]) + env, _log = _fake_gh(td, {"view": { + PR_A: _pr_view(PR_A, ["app/a1.py"], "feat/a"), + PR_B: _pr_view(PR_B, ["app/a1.py"], "feat/b")}}) + # A child process, not this one: the readers fix their store paths from + # $HOME at IMPORT, so only a process started with the fake HOME is + # guaranteed never to read the developer's real transcripts. + counter = ( + "import importlib.util, json, sys\n" + f"spec = importlib.util.spec_from_file_location('fs', {str(SCRIPT)!r})\n" + "fs = importlib.util.module_from_spec(spec); spec.loader.exec_module(fs)\n" + "reader = fs.readers.reader_for('claude'); original = reader.to_canonical\n" + "calls = []\n" + "def counting(path, *a, **k):\n" + " calls.append(str(path)); return original(path, *a, **k)\n" + "reader.to_canonical = counting\n" + "rc = fs.main(sys.argv[1:])\n" + "sys.stderr.write('CALLS=' + json.dumps(calls) + '\\n')\n" + "sys.exit(rc)\n") + proc = subprocess.run( + [sys.executable, "-c", counter, "--pr-url", PR_A, "--pr-url", PR_B, + "--host", "claude"], + capture_output=True, text=True, encoding="utf-8", timeout=120, + env={**os.environ, "HOME": str(home), "USERPROFILE": str(home), + "PYTHONUTF8": "1", "MEMHUB_PLUGIN_SCRIPTS": str(SCRIPTS), **env}) + line = [ln for ln in proc.stderr.splitlines() if ln.startswith("CALLS=")] + calls = json.loads(line[0][len("CALLS="):]) if line else [] + check("counted batch run exits 0", proc.returncode == 0, proc.stderr[-300:]) + check("every parse was inside the fake HOME", + calls and all(c.startswith(str(home)) for c in calls), repr(calls[:3])) + check("three sessions, two PRs → three parses, not six", + len(calls) == 3 and len(set(calls)) == 3, repr(calls)) + + +def test_batch_skips_a_bad_pr_and_keeps_going(): + with tempfile.TemporaryDirectory() as td: + home = Path(td) / "home" + wt = str(Path(td) / "repo") + claude_session(home, "writer-a", wt, branch="feat/a", edits=[f"{wt}/app/a1.py"]) + wrong = "https://github.com/acme/repo/pull/9" + failing = "https://github.com/acme/repo/pull/10" + empty = "https://github.com/acme/repo/pull/11" + env, _log = _fake_gh(td, {"view": { + PR_A: _pr_view(PR_A, ["app/a1.py"], "feat/a"), + # gh answered for a DIFFERENT pull request than the one asked for. + wrong: _pr_view("https://github.com/other/thing/pull/99", ["x.py"], "x"), + failing: "FAIL", + empty: _pr_view(empty, [], "feat/empty")}}) + rc, out, err = run_batch( + home, env, "--host", "claude", + prs_file_lines=["# a comment", "", PR_A, wrong, failing, empty, + "not a url", PR_A + "/", PR_A + "#discussion"]) + doc = json.loads(out) if rc == 0 else {"prs": [], "skipped_prs": []} + reasons = {s["pr_url"]: s["reason"] for s in doc["skipped_prs"]} + check("exit 0 although most PRs failed", rc == 0, err) + check("the good PR is still scanned", + [p["pr_url"] for p in doc["prs"]] == [PR_A], out[:300]) + check("a mismatched gh answer → url_mismatch", reasons.get(wrong) == "url_mismatch", + repr(reasons)) + check("a failing gh call → gh_failed", reasons.get(failing) == "gh_failed", + repr(reasons)) + check("a PR with no files → no_files", reasons.get(empty) == "no_files", + repr(reasons)) + check("an unparseable line → invalid_url", reasons.get("not a url") == "invalid_url", + repr(reasons)) + check("the same PR spelled three ways is scanned once", + len(doc["prs"]) == 1 and PR_A not in reasons, repr(doc["skipped_prs"])) + check("gh's own stderr is never echoed", "secret-ish" not in out + err, err) + + +def test_batch_url_spellings_and_redirects(): + """A PR tab (`/files`), `http://` and `www.github.com` all name the PR the + user meant; a renamed repo is the same PR under another name, reported as + such rather than lumped in with a gh answer for a different PR.""" + with tempfile.TemporaryDirectory() as td: + home = Path(td) / "home" + wt = str(Path(td) / "repo") + claude_session(home, "writer-a", wt, branch="feat/a", edits=[f"{wt}/app/a1.py"]) + renamed = "https://github.com/acme/old-name/pull/12" + env, log = _fake_gh(td, {"view": { + PR_A: _pr_view(PR_A, ["app/a1.py"], "feat/a", + title="fix: Ёжик Álvaro — non-ASCII survives"), + renamed: _pr_view("https://github.com/acme/new-name/pull/12", ["x.py"], "x")}}) + rc, out, err = run_batch( + home, env, "--host", "claude", + prs_file_lines=[PR_A + "/files", "http://github.com/acme/repo/pull/7", + "https://www.github.com/acme/repo/pull/7/commits/abc1234", + renamed]) + doc = json.loads(out) if rc == 0 else {"prs": [], "skipped_prs": []} + check("three spellings of PR A collapse to one scanned PR", + [p["pr_url"] for p in doc["prs"]] == [PR_A], out[:300] + err) + check("gh is asked about the normalised URL", + [c[2] for c in _gh_calls(log) if c[:2] == ["pr", "view"]][:1] == [PR_A], + repr(_gh_calls(log))) + check("a non-ASCII title comes through intact", + doc["prs"] and doc["prs"][0]["title"] == "fix: Ёжик Álvaro — non-ASCII survives", + out[:300]) + check("a renamed repo → url_redirected with the canonical URL", + {"pr_url": renamed, "reason": "url_redirected", + "canonical_url": "https://github.com/acme/new-name/pull/12"} + in doc["skipped_prs"], repr(doc["skipped_prs"])) + + +def test_batch_matches_single_mode_evidence(): + """Batch scoring reuses one feature pass per session; its evidence must be + exactly what single mode reports for the same PR.""" + with tempfile.TemporaryDirectory() as td: + home = Path(td) / "home" + wt = str(Path(td) / "repo") + sha = "abcdef1234567890abcdef1234567890abcdef12" + # The helper pairs every result with the FIRST call (`t1`), so the + # commit must be that call for its output to count as proof. + claude_session(home, "committer", wt, branch="feat/a", + commands=["git commit -m 'x' app/a1.py app/a2.py"], + results=[f"[feat/a {sha[:7]}] x"]) + env, _log = _fake_gh(td, {"view": { + PR_A: _pr_view(PR_A, ["app/a1.py", "app/a2.py"], "feat/a", shas=[sha], + created="2026-09-10T00:00:00Z")}}) + rc, out, err = run_batch(home, env, "--pr-url", PR_A, "--host", "claude") + batch = json.loads(out)["prs"][0]["candidates"] if rc == 0 else [] + rc2, out2, err2 = run(home, ["app/a1.py", "app/a2.py"], "--branch", "feat/a", + "--base", "main", "--sha", sha, + "--created-at", "2026-09-10T00:00:00Z", + "--repo", "repo", "--host", "claude") + single = json.loads(out2) if rc2 == 0 else None + check("batch and single mode agree row for row", batch and batch == single, + f"{batch!r}\n{single!r}") + check("…including the sha proof", batch and batch[0]["evidence"]["shas"] == [sha[:7]], + repr(batch)) + + +def test_batch_caps_the_number_of_prs(): + with tempfile.TemporaryDirectory() as td: + home = Path(td) / "home" + home.mkdir() + env, log = _fake_gh(td, {"view": { + PR_A: _pr_view(PR_A, ["a.py"], "feat/a"), + PR_B: _pr_view(PR_B, ["b.py"], "feat/b")}}) + rc, out, err = run_batch(home, env, "--pr-url", PR_A, "--pr-url", PR_B, + "--max-prs", "1", "--host", "claude") + doc = json.loads(out) if rc == 0 else {"prs": [], "skipped_prs": []} + check("only the first PR is scanned", + [p["pr_url"] for p in doc["prs"]] == [PR_A], out[:300]) + check("the second is reported over_cap", + doc["skipped_prs"] == [{"pr_url": PR_B, "reason": "over_cap"}], + repr(doc["skipped_prs"])) + check("…and gh was never asked about it", + all(PR_B not in c for c in _gh_calls(log)), repr(_gh_calls(log))) + + +def test_batch_pages_a_large_pr_s_files_through_the_api(): + """`gh pr view --json files` stops at 100. A PR past that must be read + through the REST API.""" + with tempfile.TemporaryDirectory() as td: + home = Path(td) / "home" + wt = str(Path(td) / "repo") + claude_session(home, "late-file", wt, branch="elsewhere", + edits=[f"{wt}/app/f150.py", f"{wt}/app/f151.py"]) + first_page = [f"app/f{i}.py" for i in range(100)] + everything = "\n".join(f"app/f{i}.py" for i in range(160)) + "\n" + env, log = _fake_gh(td, { + "view": {PR_A: _pr_view(PR_A, first_page, "feat/big")}, + "api": {"repos/acme/repo/pulls/7/files": everything}}) + rc, out, err = run_batch(home, env, "--pr-url", PR_A, "--host", "claude") + doc = json.loads(out) if rc == 0 else {"prs": []} + api = [c for c in _gh_calls(log) if c[:1] == ["api"]] + check("the files API was paged", len(api) == 1 and "--paginate" in api[0], + repr(api)) + cands = doc["prs"][0]["candidates"] if doc["prs"] else [] + check("a file past the first 100 still counts as evidence", + cands and "app/f150.py" in cands[0]["evidence"]["files"], out[:300]) + + +def test_batch_skips_an_enterprise_host(): + """MemHub's link_pr takes github.com PRs only, so an enterprise PR is + reported as unsupported_host — never scanned, never asked of gh, and it + does not use up the PR cap.""" + ghe = "https://ghe.corp/acme/repo/pull/7" + with tempfile.TemporaryDirectory() as td: + home = Path(td) / "home" + home.mkdir() + env, log = _fake_gh(td, {"view": { + ghe: _pr_view(ghe, ["a.py"], "feat/a"), + PR_A: _pr_view(PR_A, ["a.py"], "feat/a")}}) + rc, out, err = run_batch(home, env, "--pr-url", ghe, "--pr-url", ghe + "/files", + "--pr-url", PR_A, "--max-prs", "1", "--host", "claude") + doc = json.loads(out) if rc == 0 else {"prs": [], "skipped_prs": []} + check("the enterprise PR is skipped once as unsupported_host", + doc["skipped_prs"] == [{"pr_url": ghe, "reason": "unsupported_host"}], + repr(doc["skipped_prs"]) + err) + check("…the github.com PR still fits under the cap", + [p["pr_url"] for p in doc["prs"]] == [PR_A], out[:300]) + check("…and gh was never asked about it", + all(not any("ghe.corp" in a for a in c) for c in _gh_calls(log)), + repr(_gh_calls(log))) + + +def test_batch_repo_filter_is_per_pr(): + with tempfile.TemporaryDirectory() as td: + home = Path(td) / "home" + # No git remote in these directories, so the repo name falls back to + # the directory's basename — `repo` and `other-project`. + in_repo = str(Path(td) / "repo") + elsewhere = str(Path(td) / "other-project") + claude_session(home, "in-repo", in_repo, branch="x", edits=[f"{in_repo}/README.md"], + commands=["git switch -c feat/a"]) + claude_session(home, "elsewhere", elsewhere, branch="x", + edits=[f"{elsewhere}/README.md"], commands=["git switch -c feat/a"]) + # GitHub names are case-insensitive: a checkout spelled `REPO` is the PR's repo. + shouted = str(Path(td) / "upper" / "REPO") + claude_session(home, "shouted", shouted, branch="x", + edits=[f"{shouted}/README.md"], commands=["git switch -c feat/a"]) + env, _log = _fake_gh(td, {"view": {PR_A: _pr_view(PR_A, ["README.md"], "feat/a")}}) + rc, out, err = run_batch(home, env, "--pr-url", PR_A, "--host", "claude") + doc = json.loads(out) if rc == 0 else {"prs": [{"candidates": []}]} + ids = [c["conversation_id"] for c in doc["prs"][0]["candidates"]] + check("the session in the PR's repo stays", "in-repo" in ids, repr(ids)) + check("…and so does one whose checkout spells it in another case", + "shouted" in ids, repr(ids)) + check("an unrelated project's README.md does not", "elsewhere" not in ids, repr(ids)) + + +def test_batch_argument_errors_are_clean(): + with tempfile.TemporaryDirectory() as td: + home = Path(td) / "home" + home.mkdir() + env, _log = _fake_gh(td, {"view": {}}) + listing = Path(td) / "files.txt" + listing.write_text("a.py\n", encoding="utf-8") + for label, args in ( + ("both modes", ["--files-from", str(listing), "--pr-url", PR_A]), + ("neither mode", []), + ("a single-mode flag in a batch", ["--pr-url", PR_A, "--branch", "x"]), + ("--repo in a batch", ["--pr-url", PR_A, "--repo", "repo"]), + ("--max-prs 0", ["--pr-url", PR_A, "--max-prs", "0"])): + rc, _out, err = run_batch(home, env, *args) + check(f"{label} → exit 2, no traceback", + rc == 2 and "ERROR" in err and "Traceback" not in err, err) + # gh absent: PATH holds only python's own directory. + bare = {"PATH": str(Path(sys.executable).parent)} + if (Path(sys.executable).parent / "gh").exists(): + print(" skip gh-absent check: gh sits next to python") + else: + rc, _out, err = run_batch(home, bare, "--pr-url", PR_A) + check("gh not on PATH → exit 2 naming gh", + rc == 2 and "gh not found" in err and "Traceback" not in err, err) + + if __name__ == "__main__": print("find_sessions") for name, fn in sorted(globals().items()): diff --git a/tests/harness_stop_test.py b/tests/harness_stop_test.py index 7199f21b..4c50cef1 100644 --- a/tests/harness_stop_test.py +++ b/tests/harness_stop_test.py @@ -105,7 +105,7 @@ def _classifier(reply): """Substitute the one server call; yields the windows it was sent.""" sent = [] real = hx.server_classify - hx.server_classify = lambda window, repo="", timeout=0: (sent.append(window), (reply, 0.1))[1] + hx.server_classify = lambda window, repo="", timeout=0, **_kw: (sent.append(window), (reply, 0.1))[1] try: yield sent finally: @@ -367,6 +367,73 @@ def test_the_handoff_command_prints_the_fork_instruction(): print("PASS test_the_handoff_command_prints_the_fork_instruction") +def test_a_cost_verdict_hands_a_repair_once_and_handoff_prints_the_repair_prompt(): + """Live repair (harness-self-improvement-spec §4). The flush recorded a + `cost` verdict for one of this session's fires; the next Stop says one + line for the person naming the rule and the turn the fire happened on, + and the `handoff --repair` command prints a fork prompt that repairs that + rule as a proposed supersede. The same cost is never handed twice, and a + verdict below the bar hands nothing.""" + import fire_effect as fe + import rulebook_hook as rh + with _Env() as env: + tp = env.base / "sess.jsonl" + _transcript(tp, [("tidy the notes", "done", [("Edit", {"file_path": "n.md"}, "ok", False)]), + ("thanks", "ok", []), + ("next", "ok", [])]) + ld = rh._ledger_dir() + os.makedirs(ld, exist_ok=True) + with open(os.path.join(ld, "fires.jsonl"), "a", encoding="utf-8") as f: + f.write(json.dumps({"fire_id": "f-cost", "session_id": "sess", "rule_id": "R-1", + "title": "Resolve bot review threads", "tool_use_id": "t1-0", + "mode": "advise", "hook_phase": "pre", + "fired_at": "2026-10-13T10:00:00+00:00"}) + "\n") + f.write(json.dumps({"fire_id": "f-low", "session_id": "sess", "rule_id": "R-2", + "title": "Low cost", "tool_use_id": "t1-0", "mode": "advise", + "hook_phase": "pre", "fired_at": "2026-10-13T10:00:01+00:00"}) + "\n") + fe.record_verdicts(ld, [{"fire_id": "f-low", "effect": "cost", "effect_helped": 0.2, "effect_cost": 0.5}]) + with _classifier({"signal": False, "reason": "quiet"}): + out = _stop(env, tp) + assert out is None, f"a cost below the bar hands nothing: {out}" + fe.record_verdicts(ld, [{"fire_id": "f-cost", "effect": "cost", "effect_helped": 0.1, "effect_cost": 0.9}]) + with _classifier({"signal": False, "reason": "quiet"}): + out = _stop(env, tp) + ctx = (out or {}).get("hookSpecificOutput", {}).get("additionalContext", "") + assert ctx.startswith(hs.REPAIR_PREFIX) and "Resolve bot review threads" in ctx and "at turn 1" in ctx, ctx + assert hs.FORK_MARK not in ctx, "the fork prompt never reaches the person's screen" + log = hx.log_path("stop.log").read_text() + assert "repair sess t1: fork requested for rule R-1" in log, log + with _classifier({"signal": False, "reason": "quiet"}): + again = _stop(env, tp) + assert again is None, f"the same cost is handed once: {again}" + run = lambda *args: subprocess.run( # noqa: E731 + [sys.executable, str(SCRIPTS / "harness_stop.py"), "handoff", *args], + capture_output=True, text=True, timeout=30, + env=dict(os.environ, MEMHUB_HARNESS_EXTRACT="0", CLAUDE_CONFIG_DIR=str(env.base / "claude"))) + proj = env.base / "claude" / "projects" / "-w" + proj.mkdir(parents=True) + shutil.copy(tp, proj / "sess.jsonl") + ok = run("--moment", "sess#1", "--repair") + assert ok.returncode == 0, ok.stderr + reason = ok.stdout + assert reason.startswith(hs.BLOCK_PREFIX) and "a rule cost the work at turn 1" in reason, reason + assert f"{hs.FORK_MARK}, moment sess#1" in reason and "to REPAIR one team rule" in reason, reason + assert 'Rule "Resolve bot review threads" (rule_id R-1)' in reason and "p_cost 0.90" in reason, reason + assert "Harness-repair path" in reason and "supersedes_rule_id" in reason, reason + assert f'stamp --transcript "{proj / "sess.jsonl"}" --turn 1' in reason, reason + assert f'"{hs.FORK_MARK}: repaired "' in reason and 'says "filed" or "repaired"' in reason, reason + bad = run("--moment", "sess#2", "--repair") + assert bad.returncode == 2 and "no repair noted" in bad.stderr, bad + print("PASS test_a_cost_verdict_hands_a_repair_once_and_handoff_prints_the_repair_prompt") + + +def test_the_standing_rule_names_the_repair_line_and_its_handoff(): + rule = hs.session_rule("sess") + assert hs.REPAIR_PREFIX in rule and "handoff --moment sess#N --repair" in rule, rule + assert "files or repairs a rule" in rule, rule + print("PASS test_the_standing_rule_names_the_repair_line_and_its_handoff") + + def test_session_start_carries_the_standing_rule(): """The rule that keeps the agent from narrating a launch goes out at SessionStart, a channel the person is not shown; with the flag off, or for diff --git a/tests/mine_sessions_test.py b/tests/mine_sessions_test.py index 320194fd..f83a33dd 100644 --- a/tests/mine_sessions_test.py +++ b/tests/mine_sessions_test.py @@ -10,6 +10,7 @@ sys.path.insert(0, str(ROOT / "plugins" / "memhub" / "scripts")) import rulebook_paths # noqa: E402 +import rulebook_hook # noqa: E402 def _book_dir(home) -> Path: @@ -18,8 +19,9 @@ def _book_dir(home) -> Path: return Path(home) / ".config" / "memhub-plugin" / "rulebook" / rulebook_paths.backend_key() / "book" -def _run(*extra, home): - env = {k: v for k, v in os.environ.items() if k not in ("CLAUDE_PLUGIN_ROOT", "MEMHUB_PLUGIN_SCRIPTS")} +def _run(*extra, home, **more_env): + env = {k: v for k, v in os.environ.items() if k not in ("CLAUDE_PLUGIN_ROOT", "MEMHUB_PLUGIN_SCRIPTS", "CLAUDE_CONFIG_DIR")} + env.update(more_env) env["HOME"] = home env["USERPROFILE"] = home return subprocess.run([sys.executable, str(SCRIPT), *extra], capture_output=True, text=True, env=env, cwd=home, timeout=120) @@ -54,9 +56,9 @@ def main() -> int: {"title": "empty-origin-probe", "matcher": {"event": "bash", "command_rx": "never-happens-abc\\b"}, "claude_md": {}}, "not-a-dict"])) book = _book_dir(home); book.mkdir(parents=True) - (book / "x.json").write_text(json.dumps({"rules": [{"delivery": "agent_hook", "matcher": {"event": "bash", "command_rx": "x"}}, # no title: must be skipped, not fatal + (book / os.path.basename(rulebook_hook.book_path("probe-repo"))).write_text(json.dumps({"rules": [{"delivery": "agent_hook", "matcher": {"event": "bash", "command_rx": "x"}}, # no title: must be skipped, not fatal {"title": "ok-rule", "rule_id": "r1", "delivery": "agent_hook", "mode": "advise", "version": "not-a-number", "matcher": {"event": "bash", "command_rx": "git\\s+push\\b", "warn_once_per": "session"}, "scope_repos": [], "scope_paths": [], "scope_exclude_paths": []}]})) - p = _run("--out", str(out), "--rule-file", str(cand), "--rule-file", str(partial), "--rule-file", str(anchor), "--rule-file", str(sess_ord), "--rule-file", str(prompt_ord), "--rule-file", str(bare_prompt), "--candidates", str(cands), "--claude-md", str(md), "--facets", str(facets), home=home) + p = _run("--out", str(out), "--repo", "probe-repo", "--rule-file", str(cand), "--rule-file", str(partial), "--rule-file", str(anchor), "--rule-file", str(sess_ord), "--rule-file", str(prompt_ord), "--rule-file", str(bare_prompt), "--candidates", str(cands), "--claude-md", str(md), "--facets", str(facets), home=home) ok = p.returncode == 0 and "sessions read" in p.stdout and "probe-rule" in p.stdout and "WHAT CLAUDE.MD DECLARES" in p.stdout and "PROPOSED RULES" in p.stdout and "unknown friction category ['bogus_label']" in p.stderr and "=== WHAT WENT WRONG" not in p.stdout and (out / "digests").is_dir() print(("ok " if ok else "FAIL"), "empty HOME: runs, backtests --rule-file, seeds from --claude-md, warns on --facets vocab but reports no facet for a session not in the corpus, writes digests/"); fails += not ok if not ok: print(p.stdout[-800:], p.stderr[-800:]) @@ -235,5 +237,95 @@ def _offered(): return sorted(os.path.basename(x) for batch in json.load(open(ou if not ok: print(stamps, _offered()) return 1 if fails else 0 + +def _rec(kind, content, cwd, ts, branch="fm-feat/probe-only-branch"): + return json.dumps({"type": kind, "cwd": cwd, "gitBranch": branch, "timestamp": ts, "message": {"content": content}}) + + +def _use(i, name, inp): + return {"type": "tool_use", "id": f"tu{i}", "name": name, "input": inp} + + +def counting() -> int: + """What a session counts as: its subagents' calls, a repo even after its worktree is deleted, an edit read the + way the hook reads it, one facet per session, and only the books of the repos the sessions are in.""" + fails = 0 + with tempfile.TemporaryDirectory() as home: + repo = Path(home) / "src" / "probe-repo"; repo.mkdir(parents=True) + subprocess.run(["git", "init", "-q", str(repo)], check=True) + subprocess.run(["git", "-C", str(repo), "remote", "add", "origin", "https://example.com/acme/probe-repo.git"], check=True) + gone = repo / ".claude" / "worktrees" / "deleted-one" # never created: a worktree removed after the session + proj = Path(home) / ".claude" / "projects" / "-probe"; proj.mkdir(parents=True) + sid = "cccccccc-1111-2222-3333-444444444444" + (proj / f"{sid}.jsonl").write_text("\n".join([ + _rec("user", "fix the parser", str(gone), "2026-10-10T10:00:00Z"), + _rec("assistant", [_use(1, "Edit", {"file_path": "/x/a.py", "old_string": "# TODO keep\n", "new_string": "# TODO keep\nx = 1\n"})], str(gone), "2026-10-10T10:01:00Z"), + _rec("assistant", [_use(2, "MultiEdit", {"file_path": "/x/b.py", "edits": [{"old_string": "a", "new_string": "print('dbg')"}]})], str(gone), "2026-10-10T10:02:00Z"), + _rec("assistant", [_use(3, "Edit", {"file_path": "/x/c.py", "old_string": "y", "new_string": "import os\n"})], str(gone), "2026-10-10T10:03:00Z"), + ])) + sub = proj / sid / "subagents"; sub.mkdir(parents=True) + (sub / "agent-a1.jsonl").write_text("\n".join([ + _rec("user", "no, push it now", str(gone), "2026-10-10T10:01:30Z"), # the parent's prompt: never a correction by the person + _rec("assistant", [_use(9, "Bash", {"command": "git push origin HEAD"})], str(gone), "2026-10-10T10:01:40Z"), + ])) + rules = { + "push-probe": {"title": "push-probe", "matcher": {"event": "bash", "command_rx": "git\\s+push\\b"}}, + "todo-probe": {"title": "todo-probe", "matcher": {"event": "edit", "content_rx": "TODO"}}, + "print-probe": {"title": "print-probe", "matcher": {"event": "edit", "content_rx": "print\\("}}, + } + files = [] + for name, body in rules.items(): + f = Path(home) / f"{name}.json"; f.write_text(json.dumps(body)); files += ["--rule-file", str(f)] + book = _book_dir(home); book.mkdir(parents=True) + (book / os.path.basename(rulebook_hook.book_path("probe-repo"))).write_text(json.dumps({"rules": [ + {"title": "fresh-rule", "rule_id": "r1", "delivery": "agent_hook", "mode": "advise", "version": 1, "matcher": {"event": "bash", "command_rx": "git\\s+push\\b"}}]})) + (book / os.path.basename(rulebook_hook.book_path("some-fixture"))).write_text(json.dumps({"rules": [ + {"title": "stale-fixture-rule", "rule_id": "r9", "delivery": "agent_hook", "mode": "advise", "version": 1, "matcher": {"event": "bash", "command_rx": "git\\s+push\\b"}}]})) + out = Path(home) / "out" + p = _run("--out", str(out), "--repo", str(repo), "--digest-top", "0", *files, home=home) + rows = {r["title"]: r for r in json.load(open(out / "proposals.json")) if isinstance(r, dict) and r.get("title")} if (out / "proposals.json").exists() else {} + fired = {k: (rows.get(k) or {}).get("fired_n") for k in rules} + ok = p.returncode == 0 and "--repo " + str(repo) + " -> probe-repo" in p.stdout and "(M=1)" in p.stdout and "parent 1" in p.stdout + print(("ok " if ok else "FAIL"), "--repo given as a path names its repo, and a session whose worktree was deleted is placed by the repo it sat in"); fails += not ok + if not ok: print(p.stdout[:800], p.stderr[-600:]) + ok = fired["push-probe"] == 1 and "1 subagent transcripts folded" in p.stdout + print(("ok " if ok else "FAIL"), "a command run inside a subagent counts for its session"); fails += not ok + ok = fired["todo-probe"] == 0 and fired["print-probe"] == 1 + print(("ok " if ok else "FAIL"), "an edit is read as the hook reads it: a line it kept does not fire, a MultiEdit's edits do"); fails += not ok + if not ok: print(fired) + ok = "fresh-rule" in p.stdout and "stale-fixture-rule" not in p.stdout + print(("ok " if ok else "FAIL"), "rules already on come from this repo's book, not every book in the cache"); fails += not ok + corpus = json.load(open(out / "corpus.json")) if (out / "corpus.json").exists() else [{}] + ok = corpus[0].get("users") == ["fix the parser"] and corpus[0].get("stamp") == "1u3c0r" and "main_calls" not in corpus[0] + print(("ok " if ok else "FAIL"), "a subagent's prompt is not the person's turn, and the facet stamp counts the session's own calls"); fails += not ok + if not ok: print({k: corpus[0].get(k) for k in ("users", "stamp")}) + + # one facet per session: yours wins over an /insights facet for the same session + facets = Path(home) / "facets.json" + facets.write_text(json.dumps([{"session_id": sid, "outcome": "mostly", "friction": [{"category": "wrong_source", "detail": "mine"}]}])) + ins = Path(home) / ".claude" / "usage-data" / "facets"; ins.mkdir(parents=True) + (ins / f"{sid}.json").write_text(json.dumps({"session_id": sid, "friction_counts": {"buggy_code": 2}, "friction_detail": "insights"})) + p = _run("--out", str(out), "--repo", "probe-repo", "--digest-top", "0", "--facets", str(facets), home=home) + ok = p.returncode == 0 and "1 /insights facets left out" in p.stdout and "buggy_code" not in p.stdout and "wrong_source" in p.stdout + print(("ok " if ok else "FAIL"), "an /insights facet for a session your facets already read is left out, not counted twice"); fails += not ok + if not ok: print(p.stdout[-800:]) + + # Claude Code's own config dir, when set, is where its transcripts are + moved = Path(home) / "cfg"; shutil.move(str(Path(home) / ".claude"), str(moved)) + p = _run("--out", str(out), "--repo", "probe-repo", "--digest-top", "0", home=home, CLAUDE_CONFIG_DIR=str(moved)) + ok = p.returncode == 0 and "(M=1)" in p.stdout + print(("ok " if ok else "FAIL"), "transcripts are read from CLAUDE_CONFIG_DIR when it is set"); fails += not ok + if not ok: print(p.stdout[:400]) + shutil.move(str(moved), str(Path(home) / ".claude")) + + # the repo is remembered: once even the parent is gone, the cache still places the session + shutil.rmtree(repo) + p = _run("--out", str(out), "--repo", "probe-repo", "--digest-top", "0", home=home) + ok = p.returncode == 0 and "(M=1)" in p.stdout + print(("ok " if ok else "FAIL"), "a repo resolved once is kept, so the session stays placed after its whole checkout is gone"); fails += not ok + if not ok: print(p.stdout[:600]) + return fails + + if __name__ == "__main__": - sys.exit(main()) + sys.exit(1 if (main() + counting()) else 0) diff --git a/tests/pr_link_test.py b/tests/pr_link_test.py index 9b270bc1..80b9b3f1 100644 --- a/tests/pr_link_test.py +++ b/tests/pr_link_test.py @@ -35,6 +35,7 @@ sys.path.insert(0, str(SCRIPTS)) import mcp_http # noqa: E402 import pr_babysit_trigger # noqa: E402 +from _memhub_auth import skill_command # noqa: E402 import pr_link # noqa: E402 failures: list[str] = [] @@ -465,7 +466,7 @@ def test_github_mcp_tools_are_recognised_by_their_server_segment(): # …but a server whose NAME contains underscores is still GitHub. # `[^_]*` stopped at the first one and rejected every tool from # `github_enterprise`, or from any plugin-provided server — this repo's - # own tools arrive as `mcp__plugin_memhub-staging_memhub__…` + # own tools arrive as `mcp__plugin_memhub_memhub__…` # (Codex review, PR #182). ("mcp__github_enterprise__create_pull_request", True, True), ("mcp__plugin_github_github__create_pull_request", True, True), @@ -1131,7 +1132,12 @@ def test_the_contexts_say_the_right_thing_for_each_case(): check("B2 ends with the already-linked escape that keeps a babysit loop quiet", in_play.rstrip().endswith("say nothing at all."), in_play) check("B2 offers the finder and forbids running it unasked", - "/memhub:find-contributing-sessions" in in_play + # NOTE: public hardcodes the production prefix here and pins the + # rendered bytes in pr_work_type_test's GOLDEN_PRE_CLASSIFICATION, so + # W1.2's "skill prefix from the manifest" is NOT applied to this + # string. On a staging install this tells the user to run a command + # that does not exist there. Tracked as a W1.2 follow-up. + f"/memhub:link-pr --find {url}" in in_play and "without the user saying yes" in in_play) disconnected = pr_link.context_for( diff --git a/tests/pr_work_type_test.py b/tests/pr_work_type_test.py index 7d6c3cec..044bebf9 100644 --- a/tests/pr_work_type_test.py +++ b/tests/pr_work_type_test.py @@ -35,8 +35,11 @@ # classification sentence. A pull request that already carries a type must # render exactly this again — the ask disappears rather than degrading into # something new. Change these deliberately, never to make the suite green. -GOLDEN_PRE_CLASSIFICATION = {"CREATED": "0feb75e89d6c385f", - "IN_PLAY": "c79dc68b09546397"} +# Re-pinned once, deliberately: the finder the templates offer is now spelled +# `/memhub:link-pr --find {pr_url}` (the separate finder skill was merged into +# link-pr); no other byte changed. +GOLDEN_PRE_CLASSIFICATION = {"CREATED": "8ee3f30a78f924a2", + "IN_PLAY": "8711d86ee07c070c"} failures: list[str] = [] diff --git a/tests/rulebook_hook_test.py b/tests/rulebook_hook_test.py index 3a72f38f..92fb673c 100644 --- a/tests/rulebook_hook_test.py +++ b/tests/rulebook_hook_test.py @@ -104,6 +104,9 @@ def portability_check() -> None: os.mkdir(incomplete) incomplete_hook = os.path.join(incomplete, "rulebook_hook.py") shutil.copy2(HOOK, incomplete_hook) + # The lock shim is what is missing here; `rulebook_shell` is part of the + # hook itself (split out for the release tree's per-file size limit). + shutil.copy2(os.path.join(os.path.dirname(HOOK), "rulebook_shell.py"), incomplete) incomplete_repo = os.path.join(td, "incomplete-repo") os.makedirs(os.path.join(incomplete_repo, ".git")) with open( @@ -692,9 +695,7 @@ def bash(cmd, **kw): denied, c, _, _ = bash(cmd) check(f"read via bash: silent — {cmd.replace(repo, '<repo>')[:50]}", not denied and "[big-read]" not in c, c) - # the measured shape: a relative path after a cd (the session cwd stays the - # trust boundary — a cd OUT of the session's repo is not followed, as for - # every other lane) + # the measured shape: a relative path after a cd for cmd in ("cd src && cat big.py", f"cd {repo}/src && cat big.py", "cd src; cat ./big.py"): denied, c, _, _ = bash(cmd) check(f"read via bash: a relative path resolves against the cd — {cmd[:40]}", @@ -775,7 +776,9 @@ def replace_file(path, old, new): f.write(text.replace(old, new)) def bash(command, *, session="b1", run_it=True, pre=True, resp=None, mutate=None): - """pre → apply the command's file effect → post, the way a session does. + """pre → apply the command's file effect → post, the way a session does; + the advice of both lanes (a write the command line shows fires in the + pre lane, the disk scan in the post lane catches the rest). The hook receives the real shell syntax under test. Applying its expected file effect in Python keeps this fixture independent of @@ -785,8 +788,9 @@ def bash(command, *, session="b1", run_it=True, pre=True, resp=None, mutate=None tid = f"tu{n[0]}" ev = {"cwd": repo, "session_id": session, "tool_name": "Bash", "tool_use_id": tid, "tool_input": {"command": command}} + said = "" if pre: - run("pre", ev, env) + said = ctx(run("pre", ev, env)[1]) if run_it: if mutate is not None: mutate() @@ -794,7 +798,8 @@ def bash(command, *, session="b1", run_it=True, pre=True, resp=None, mutate=None else: r = subprocess.run(command, shell=True, cwd=repo, capture_output=True, text=True) resp = resp or {"stdout": r.stdout, "stderr": r.stderr, "exit_code": r.returncode} - return ctx(run("post", dict(ev, tool_response=resp or {"stdout": "", "exit_code": 0}), env)[1]) + return (said + "\n" + ctx(run("post", dict(ev, tool_response=resp or {"stdout": "", "exit_code": 0}), + env)[1])).strip() mig = "alembic/versions/20260902_user_logins.py" c = bash(f"cat > {mig} <<'EOF'\ndef upgrade():\n op.create_table('user_logins')\nEOF", @@ -1071,6 +1076,66 @@ def lines(command): H.command_root(td, "cd /nope/nowhere && gh pr create") == "") +def spec_pending_probe_checks() -> None: + """The post lane takes the branch diff (~20 git calls) only to answer a + pending `spec_untouched` entry for this checkout; with none it must not + run git for it, and with one the conversion still lands.""" + if shutil.which("git") is None: + check("spec pending probe (git unavailable here — skipped)", True) + return + with tempfile.TemporaryDirectory() as td: + td = os.path.realpath(td) + repo, base, shim = os.path.join(td, "repo"), os.path.join(td, "base"), os.path.join(td, "bin") + os.makedirs(repo) + os.makedirs(shim) + _git(repo, "init", "-q", "-b", "main") + with open(os.path.join(repo, "README.md"), "w") as f: + f.write("r\n") + _git(repo, "add", "-A") + _git(repo, "commit", "-qm", "init") + _git(repo, "checkout", "-qb", "feature") + os.makedirs(os.path.join(repo, "docs", "specs")) + spec = "docs/specs/a.md" + with open(os.path.join(repo, spec), "w") as f: + f.write("---\nspec: a\nowns:\n - app/x.py\n---\n# A\n") + seed_book(base, "repo", [_row("r1", {"event": "read", "path_rx": r"nothing-matches-this"})]) + log = os.path.join(td, "git.log") + real_git = shutil.which("git") + with open(os.path.join(shim, "git"), "w") as f: + f.write(f'#!/bin/sh\necho "$@" >> "{log}"\nexec "{real_git}" "$@"\n') + os.chmod(os.path.join(shim, "git"), 0o755) + env = {"MEMHUB_RULEBOOK_BASE": base, "MEMHUB_RULEBOOK_FETCH": "0", + "MEMHUB_RULEBOOK_RECALL": "0", "MEMHUB_RULEBOOK_JUDGE": "0", + "PATH": shim + os.pathsep + os.environ.get("PATH", "")} + + def post(session): + open(log, "w").close() + run("post", {"session_id": session, "cwd": repo, "hook_event_name": "PostToolUse", + "tool_name": "Read", "tool_input": {"file_path": os.path.join(repo, "README.md")}, + "tool_response": {"type": "text"}, "tool_use_id": "t1"}, env) + with open(log) as f: + return f.read() + + calls = post("none-pending") + check("spec pending probe: no pending entry, no branch-diff git calls", + "merge-base" not in calls and "diff --name-only" not in calls, calls) + + os.makedirs(os.path.join(base, "state"), exist_ok=True) + with open(os.path.join(base, "state", "one-pending.json"), "w") as f: + json.dump({"spec_pending": {json.dumps(["r1", repo, "feature"]): { + "root": repo, "branch": "feature", "spec_dir": "docs/specs", "paths": [spec]}}}, f) + calls = post("one-pending") + check("spec pending probe: a pending entry for this checkout takes the branch diff", + "merge-base" in calls and "diff --name-only" in calls, calls) + events = [] + ev_path = os.path.join(base, "ledger", "events.jsonl") + if os.path.exists(ev_path): + with open(ev_path, encoding="utf-8") as f: + events = [json.loads(line) for line in f if line.strip()] + check("spec pending probe: the pending entry still converts", + any(e.get("kind") == "converted" and e.get("rule_id") == "r1" for e in events), str(events)) + + def apply_patch_checks() -> None: """Codex edits with `apply_patch`: the whole patch in `tool_input.command`, no `file_path`, a path absolute or relative to cwd. Payloads are the shape @@ -1143,8 +1208,215 @@ def pre(ev): "[tests-before-push]" in c, c) +# ── ENG-1201: one rulebook state dir per backend ───────────────────────── + +def _install(root, url): + """A copy of the plugin's scripts as one install: `<root>/scripts/…` beside + a `.mcp.json` naming `url` — all that tells a staging install from the + prod one promote_export.py makes.""" + shutil.copytree(os.path.dirname(os.path.abspath(HOOK)), os.path.join(root, "scripts"), + ignore=shutil.ignore_patterns("__pycache__")) + if url is not None: + with open(os.path.join(root, ".mcp.json"), "w", encoding="utf-8") as f: + json.dump({"mcpServers": {"memhub": {"type": "http", "url": url}}}, f) + return os.path.join(root, "scripts", "rulebook_hook.py") + + +def _paths_cli(hook): + """The install's `rulebook_paths.py`, which prints its base when run.""" + return os.path.join(os.path.dirname(hook), "rulebook_paths.py") + + +def _hook_at(hook, home, args, payload=None, **env_extra): + """Run one install's hook with HOME = `home` and none of the developer's + MEMHUB_* overrides, so the default (keyed) base is what is exercised and + the real ~/.config is never reached.""" + env = {k: v for k, v in os.environ.items() + if not k.startswith("MEMHUB_") and k != "CLAUDE_PLUGIN_ROOT"} + env.update(HOME=home, MEMHUB_RULEBOOK_FETCH="0", **env_extra) + argv = [_paths_cli(hook)] if args == ["base"] else [hook, *args] + p = subprocess.run([sys.executable, *argv], input=json.dumps(payload or {}), + capture_output=True, text=True, env=env, timeout=30) + return p.returncode, p.stdout.strip() + + +def env_keyed_base_checks() -> None: + """Two installs on one machine (staging + the prod export) keep separate + books, ledgers and session state — observed 2026-10-05: prod's refresh + replaced the staging book (89 rules) with its own (1 rule) mid-session, + because both lived at <base>/book/<repo>-<hash>.json. And the pre-keying + ledger is still drained, never dropped.""" + with tempfile.TemporaryDirectory() as td: + home = os.path.join(td, "home") + os.makedirs(home) + staging = _install(os.path.join(td, "staging"), "https://api.staging.example.test/mcp-server/mcp") + prod = _install(os.path.join(td, "prod"), "https://api.example.test/mcp-server/mcp") + broken = _install(os.path.join(td, "broken"), None) + shared = os.path.join(home, ".config", "memhub-plugin", "rulebook") + + _, s_base = _hook_at(staging, home, ["base"]) + _, p_base = _hook_at(prod, home, ["base"]) + _, s_book = _hook_at(staging, home, ["book-path", "xmem"]) + _, p_book = _hook_at(prod, home, ["book-path", "xmem"]) + check("env base: each install keys its state by its own backend host", + s_base == os.path.join(shared, "api.staging.example.test") + and p_base == os.path.join(shared, "api.example.test"), f"{s_base} | {p_base}") + check("env base: two installs never share a repo's book", + s_book != p_book and s_book.startswith(s_base + os.sep) + and p_book.startswith(p_base + os.sep), f"{s_book} | {p_book}") + _, b_base = _hook_at(broken, home, ["base"]) + check("env base: an install whose .mcp.json is unreadable keeps the old unkeyed base", + b_base == shared, b_base) + override = os.path.join(td, "override") + _, o_base = _hook_at(staging, home, ["base"], MEMHUB_RULEBOOK_BASE=override) + _, o_book = _hook_at(prod, home, ["book-path", "xmem"], MEMHUB_RULEBOOK_BASE=override) + check("env base: MEMHUB_RULEBOOK_BASE still names the base verbatim, unkeyed", + o_base == override and o_book.startswith(os.path.join(override, "book") + os.sep), + f"{o_base} | {o_book}") + _, x_base = _hook_at(staging, home, ["base"], MEMHUB_MCP_BASE_URL="http://127.0.0.1:9") + check("env base: a per-process MEMHUB_MCP_BASE_URL (a test proxy) does not re-key the install", + x_base == s_base, x_base) + + # The live failure, end to end: the staging book gates, the prod book + # (cached for the same repo) has nothing — each install reads its own, + # and the staging fire lands in the staging ledger only. + repo = os.path.join(td, "xmem") + os.makedirs(os.path.join(repo, ".git")) + with open(os.path.join(repo, ".git", "HEAD"), "w", encoding="utf-8") as f: + f.write("ref: refs/heads/main\n") + gate = {"id": "staging-gate", "on": "bash", "rx": r"git\s+push", "mode": "gate", + "fire_scope": "call", "repo_scope": "any", "status": "active", + "text": "Do not push yet", "why": "staging rule"} + seed_book(s_base, "xmem", [gate]) + seed_book(p_base, "xmem", []) + call = {"cwd": repo, "session_id": "env-s1", "tool_name": "Bash", + "tool_input": {"command": "git push origin main"}} + + def denied(out): + try: + return json.loads(out)["hookSpecificOutput"].get("permissionDecision") == "deny" + except Exception: + return False + _, s_out = _hook_at(staging, home, ["pre"], call) + _, p_out = _hook_at(prod, home, ["pre"], dict(call, session_id="env-p1")) + check("env base: the staging install enforces the staging book; the prod install its own", + denied(s_out) and not denied(p_out), f"{s_out!r} | {p_out!r}") + check("env base: the fire lands in that install's ledger, never the other's or the shared one", + os.path.isfile(os.path.join(s_base, "ledger", "fires.jsonl")) + and not os.path.exists(os.path.join(p_base, "ledger", "fires.jsonl")) + and not os.path.exists(os.path.join(shared, "ledger", "fires.jsonl"))) + check("env base: session state is per install too", + os.path.isdir(os.path.join(s_base, "state")) + and not os.path.exists(os.path.join(shared, "state"))) + + # The pre-keying ledger: rows recorded before the upgrade and never flushed + # are still shipped, behind their own watermark, until it is drained. + sys.path.insert(0, os.path.dirname(os.path.abspath(HOOK))) + import rulebook_hook as H + import rulebook_paths + + def flush_lane(): # what `rulebook_hook.py flush` runs + H.flush_fires(final=True) + rulebook_paths.drain_legacy(vars(H), final=True) # the hook passes globals() + saved = (H.BASE, H.LEGACY_BASE, H._api, H._ACTIVE_BASE) + with tempfile.TemporaryDirectory() as td: + keyed, legacy = os.path.join(td, "rulebook", "api.example.test"), os.path.join(td, "rulebook") + for base, ids in ((keyed, ("new-1",)), (legacy, ("old-1", "old-2"))): + os.makedirs(os.path.join(base, "ledger"), exist_ok=True) + with open(os.path.join(base, "ledger", "fires.jsonl"), "w", encoding="utf-8") as f: + for i in ids: + f.write(json.dumps({"fire_id": i, "rule_id": "r", "fired_at": "t"}) + "\n") + posted, fail = [], {"on": True} + + class _R: + def __init__(self, n): + self.status = 500 if fail["on"] else 202 + self.data = None if fail["on"] else {"accepted": n, "rejected": 0} + + class _Http: + @staticmethod + def rest(url, bearer, method, body=None, headers=None, timeout=None): + rows = (body or {}).get("fires") or [] + if not fail["on"]: + posted.extend(r["fire_id"] for r in rows) + return _R(len(rows)) + try: + H.BASE, H.LEGACY_BASE, H._ACTIVE_BASE = keyed, legacy, "" + H._api = lambda: ("http://x", "t", _Http) + legacy_sent = os.path.join(legacy, "ledger", ".sent") + flush_lane() # server down + check("legacy ledger: a failed flush leaves the old rows behind their watermark", + H.load_sent().get("fires_offset", 0) == 0 and not os.path.exists(legacy_sent) + and posted == [], str(posted)) + fail["on"] = False + flush_lane() + check("legacy ledger: the pre-upgrade rows are flushed after this install's own", + posted == ["new-1", "old-1", "old-2"], str(posted)) + with open(legacy_sent, encoding="utf-8") as f: + lw = json.load(f) + check("legacy ledger: …and its own watermark reaches the end of the old file", + lw.get("fires_offset") == os.path.getsize(os.path.join(legacy, "ledger", "fires.jsonl")), str(lw)) + check("legacy ledger: the drain leaves this process on its own base", + H._base() == keyed and H._ledger_dir() == os.path.join(keyed, "ledger")) + real_flush, legacy_runs = H.flush_fires, [] + + def counting_flush(final=False): + if H._ACTIVE_BASE == legacy: + legacy_runs.append(1) + return real_flush(final=final) + H.flush_fires = counting_flush + try: + flush_lane() + finally: + H.flush_fires = real_flush + check("legacy ledger: a drained old ledger sends nothing again", + posted == ["new-1", "old-1", "old-2"], str(posted)) + check("legacy ledger: …and once drained it is not flushed at all (no per-flush cost)", + legacy_runs == [], str(legacy_runs)) + H.LEGACY_BASE = "" # MEMHUB_RULEBOOK_BASE in force + with open(os.path.join(legacy, "ledger", "fires.jsonl"), "a", encoding="utf-8") as f: + f.write(json.dumps({"fire_id": "old-3", "rule_id": "r", "fired_at": "t"}) + "\n") + flush_lane() + check("legacy ledger: with an override there is no old ledger to drain", + "old-3" not in posted, str(posted)) + # A send failure during the drain is loud where capture_health looks: + # the flush lane breadcrumbs under the base it is flushing (the old + # one), so the drain re-raises it in THIS install's .last_error. + H.LEGACY_BASE = legacy + with open(os.path.join(legacy, "ledger", "fires.jsonl"), "a", encoding="utf-8") as f: + f.write(json.dumps({"fire_id": "old-4", "rule_id": "r", "fired_at": "t"}) + "\n") + before = json.load(open(legacy_sent, encoding="utf-8"))["fires_offset"] + + class _Down: + @staticmethod + def rest(url, bearer, method, body=None, headers=None, timeout=None): + raise OSError("connection refused (harness)") + H._api = lambda: ("http://x", "t", _Down) + own_crumb = os.path.join(keyed, "ledger", ".last_error") + if os.path.exists(own_crumb): + os.unlink(own_crumb) + flush_lane() + try: + crumb = json.load(open(own_crumb, encoding="utf-8")) + except Exception: + crumb = {} + check("legacy ledger: a send failure while draining lands in this install's own .last_error", + crumb.get("what") == "flush" and "old shared ledger" in str(crumb.get("error")) + and "connection refused" in str(crumb.get("error")), str(crumb)) + check("legacy ledger: …and the old rows stay behind their watermark", + json.load(open(legacy_sent, encoding="utf-8"))["fires_offset"] == before) + with open(HOOK, encoding="utf-8") as f: + src = f.read() + lane = src[src.index('if mode == "flush":'):] + check("legacy ledger: the hook's flush lane runs the drain after its own flush", + lane.find("_rbp.drain_legacy(") > lane.find("flush_fires(final=final)") >= 0) + finally: + H.BASE, H.LEGACY_BASE, H._api, H._ACTIVE_BASE = saved + + def main() -> int: portability_check() + env_keyed_base_checks() # ENG-1201: one state dir per backend with tempfile.TemporaryDirectory() as td: repo = os.path.join(td, "xmem") # fake git repo named xmem os.makedirs(os.path.join(repo, ".git")) @@ -1320,9 +1592,11 @@ def main() -> int: tool_response={"stdout": "ok"}), cenv) with open(os.path.join(td, "ledger", "fires.jsonl"), encoding="utf-8") as f: rows = [json.loads(l) for l in f if l.strip() and '"cap-' in l] - modes = {r["rule_id"]: r["mode"] for r in rows} + modes = [(r["rule_id"], r["mode"]) for r in rows] check("cap: the cut rule is logged mode=suppressed, never silently dropped", - modes == {"cap-0": "advise", "cap-1": "advise", "cap-2": "suppressed"}, str(modes)) + modes[:3] == [("cap-0", "advise"), ("cap-1", "advise"), ("cap-2", "suppressed")], str(modes)) + check("cap: a rule the cap cut is not spent — it fires on the next matching call", + modes[3:] == [("cap-2", "advise")], str(modes)) check("cap: raw_matches_before_fire recorded", all(r["raw_matches_before_fire"] == 1 for r in rows)) with open(os.path.join(td, "ledger", "events.jsonl"), encoding="utf-8") as f: evs = [json.loads(l) for l in f if l.strip()] @@ -1575,10 +1849,77 @@ def _windows(end): "gives up quickly rather than reading it all", got is None and (time.time() - t0) < 1.0) - # --- what the recall lane sends --------------------------------------- - # /recall is the one lane that carries content: the relevance judge needs - # the call itself. A command line is also where credentials live, and they - # are worth nothing to the judge. + # --- anchor rules: matched locally, as whole identifiers (ENG-1203) ------ + anc = {"id": "a", "on": "anchor", "anchors": ["agent-plugins", "SKILL.md", + "app/core/config.py", "docs/specs/", "/"]} + for label, text, want in [ + ("a longer name containing the anchor does not match", + "cd ~/xtrace/agent-plugins-internal && ls", []), + ("the anchor itself matches", "gh repo view XTraceAI/agent-plugins", ["agent-plugins"]), + ("a file name matches after a directory", "cat skills/x/SKILL.md", ["SKILL.md"]), + ("a relative path matches the tail of an absolute one", + "/Users/f/MemHub-Backend/app/core/config.py", ["app/core/config.py"]), + ("…but not a longer file name", "app/core/config.pyc", []), + ("…nor a different directory prefix", "myapp/core/config.py", []), + ("a trailing-slash anchor matches the directory named whole", "ls docs/specs", ["docs/specs/"]), + ("…and any path under it", "cat docs/specs/x.md", ["docs/specs/"]), + ("…but not a longer directory name", "ls docs/specs-old/x.md", []), + ("case-sensitive", "cat skill.md", []), + ("a bare `/` anchor never matches", "ls /", []), + ]: + check("anchor_hits: %s" % label, rb.anchor_hits(anc, text) == want, str(rb.anchor_hits(anc, text))) + check("the /recall client is gone", not hasattr(rb, "recall_anchor_rules")) + + with tempfile.TemporaryDirectory() as td: + repo = os.path.join(td, "anchrepo") + os.makedirs(os.path.join(repo, ".git")) + with open(os.path.join(repo, ".git", "HEAD"), "w", encoding="utf-8") as f: + f.write("ref: refs/heads/main\n") + seed_book(td, "anchrepo", [ + _row("anc-cfg", delivery="anchor_recall", anchors=["config.py"], statement="CFG ANCHOR"), + _row("anc-scoped", delivery="anchor_recall", anchors=["models.py"], + statement="SCOPED ANCHOR", scope_paths=["app/**"]), + _row("anc-draft", delivery="anchor_recall", anchors=["config.py"], + statement="DRAFT ANCHOR", status="draft"), + _row("anc-other", delivery="anchor_recall", anchors=["config.py"], + statement="OTHER REPO ANCHOR", scope_repos=["some-other-repo"]), + ] + [_row("anc-cap%d" % i, delivery="anchor_recall", anchors=["cap.py"], + statement="CAP ANCHOR %d" % i) for i in (1, 2, 3)]) + # No token and no server: the lane must not need either. + aenv = {"MEMHUB_RULEBOOK_BASE": td, "MEMHUB_RULEBOOK_FETCH": "0", + "MEMHUB_RULEBOOK_JUDGE": "0", "MEMHUB_TOKEN": "", + "MEMHUB_MCP_BASE_URL": "http://127.0.0.1:9"} + + def edit(sid, path): + return run("pre", {"cwd": repo, "session_id": sid, "tool_name": "Edit", + "tool_input": {"file_path": os.path.join(repo, path), + "old_string": "a", "new_string": "b"}}, aenv)[1] + out = edit("an1", "src/config.py") + check("anchor lane: an edit naming the anchor fires it locally, with no server", + "CFG ANCHOR" in ctx(out), out) + check("anchor lane: a draft or other-repo anchor rule does not fire", + "DRAFT ANCHOR" not in ctx(out) and "OTHER REPO ANCHOR" not in ctx(out), out) + check("anchor lane: path scope — outside `app/**` the scoped rule stays quiet", + "SCOPED ANCHOR" not in ctx(edit("an2", "src/models.py"))) + check("anchor lane: path scope — inside `app/**` it fires", + "SCOPED ANCHOR" in ctx(edit("an3", "app/models.py"))) + out = run("pre", {"cwd": repo, "session_id": "an4", "tool_name": "Bash", + "tool_input": {"command": "python -m pytest app/models.py"}}, aenv)[1] + check("anchor lane: a Bash call carries no path, so an include-scoped rule stays quiet", + "SCOPED ANCHOR" not in ctx(out), out) + out1, out2 = ctx(edit("an6", "cap.py")), ctx(edit("an6", "cap.py")) + check("anchor lane: at most MAX_ADVISE anchors per call, first by book order; " + "the one left over is still unspent and fires on the next call", + "CAP ANCHOR 1" in out1 and "CAP ANCHOR 2" in out1 and "CAP ANCHOR 3" not in out1 + and "CAP ANCHOR 3" in out2 and "CAP ANCHOR 1" not in out2, out1 + " || " + out2) + out = run("pre", {"cwd": repo, "session_id": "an5", "tool_name": "Bash", + "tool_input": {"command": "python config.py"}}, + dict(aenv, MEMHUB_RULEBOOK_RECALL="0"))[1] + check("anchor lane: MEMHUB_RULEBOOK_RECALL=0 turns it off", "CFG ANCHOR" not in ctx(out), out) + + # --- credentials out of command text --------------------------------- + # The judge sees the call itself, and a command line is where credentials + # live; they are worth nothing to the judge. for label, cmd in [ ("Authorization header", 'curl -H "Authorization: Bearer sk-live-abcdefghij1234567890" https://x'), ("credentials in a URL", 'psql "postgres://admin:hunter2@db.internal/prod"'), @@ -1590,7 +1931,7 @@ def _windows(end): ("our own access key", 'curl -H "x: mhk_AbCdEf123456789" https://x'), ]: out = rb.redact_secrets(cmd) - check("recall redacts: %s" % label, + check("redact_secrets: %s" % label, "<redacted>" in out and "hunter2" not in out and "p4ssw0rd" not in out and "super-secret-value" not in out and "ghp_AAAABBBBCCCCDDDDEEEEFFFF1111" not in out and "AKIAIOSFODNN7EXAMPLE" not in out, out) @@ -1605,14 +1946,14 @@ def _windows(end): ('curl --user user:"pass"', 'curl --user admin:"s3cr3t pw" https://x.com', "s3cr3t pw"), ]: out = rb.redact_secrets(cmd) - check("recall redacts a quoted value: %s" % label, + check("redact_secrets: a quoted value: %s" % label, "<redacted>" in out and secret not in out, out) # Over-redaction is not free: it costs the judge the verb of the command. for cmd in ["git push --force origin main", "gh auth login", "kubectl get secrets", "npm run build -- --token-budget 500", "pytest tests/ -k rulebook", "ls -la ~/.ssh"]: - check("recall leaves an innocent command intact: %s" % cmd[:34], + check("redact_secrets leaves an innocent command intact: %s" % cmd[:34], rb.redact_secrets(cmd) == cmd, rb.redact_secrets(cmd)) check("redaction never raises on empty or None", @@ -2044,10 +2385,17 @@ def denied(cmd): apply_patch_checks() read_lane_checks() diff_base_checks() + spec_pending_probe_checks() armed_lane_checks() min_hook_version_checks() prompt_lane_checks() + edit_added_checks() + server_shape_edge_checks() spec_untouched_text_checks() + run_cache_checks() + fresh_book_fetch_checks() + first_fetch_checks() + ledger_rotation_checks() print() if FAILURES: @@ -2550,7 +2898,9 @@ def post(session, command, exit_code=0): ("Approach this assignment about staging", False), ("<system-reminder>staging</system-reminder>", True), ("Base directory for this skill: /x staging", True), - ("This session is being continued about staging", True)): + ("This session is being continued about staging", True), + ('Another Claude session sent a message:\n<agent-message from="a">' + '\n[Subagent hand-back] staging is broken', True)): check(f"harness_prompt({text[:34]!r}...) is {want}", rb_mod.harness_prompt(text) is want, str(rb_mod.harness_prompt(text))) @@ -2651,6 +3001,546 @@ def post(session, command, exit_code=0): rb_mod.to_hook_rule(dict(probe_rule)) is not None) +def fresh_book_fetch_checks() -> None: + """SessionStart and a prompt start NO fetch child when the cached book is + younger than REFRESH_AFTER_S: one was spawned on every SessionStart, a + fresh interpreter re-fetching a book a minute old. A stale book is still + fetched synchronously, bounded, and still starts no child.""" + import sys as _sys + _sys.path.insert(0, os.path.join(os.path.dirname(HOOK))) + import rulebook_hook as H + + calls = [] + saved = (H.spawn_fetch, H.fetch_book, H.load_rules, os.environ.get("MEMHUB_RULEBOOK_FETCH")) + H.spawn_fetch = lambda repo: calls.append(("spawn", repo)) + H.fetch_book = lambda repo, timeout=None: calls.append(("fetch", repo, timeout)) + H.load_rules = lambda repo: (["new"], "v", H._now(), {"s": 2}) + os.environ["MEMHUB_RULEBOOK_FETCH"] = "1" + try: + fresh = H._now() + out = H.refresh_if_stale("r", ["old"], fresh, {"s": 1}) + check("fresh book: no detached fetch child, no fetch, the cache as it is", + calls == [] and out == (["old"], fresh, {"s": 1}), str((calls, out))) + stale = (datetime.now(timezone.utc) - timedelta(seconds=H.REFRESH_AFTER_S + 5)).isoformat() + out = H.refresh_if_stale("r", ["old"], stale, {"s": 1}) + check("stale book: one bounded synchronous fetch, then the reloaded book, no child", + calls == [("fetch", "r", H.SESSION_FETCH_TIMEOUT_S)] and out[0] == ["new"], + str((calls, out))) + finally: + H.spawn_fetch, H.fetch_book, H.load_rules = saved[:3] + if saved[3] is None: + os.environ.pop("MEMHUB_RULEBOOK_FETCH", None) + else: + os.environ["MEMHUB_RULEBOOK_FETCH"] = saved[3] + + +def first_fetch_checks() -> None: + """A call in a repo with NO cached book gets one bounded synchronous fetch + before it is judged (ENG-1249: a session started above a bare `.git` and + its worktrees had no SessionStart fetch, and the detached refresh landed + after the first edit — which fired nothing). Once a minute per repo through + the `.refresh` stamp, so a dead server costs one timeout a minute; a cached + book, stale or not, is still maybe_refresh's detached business.""" + import sys as _sys + _sys.path.insert(0, os.path.join(os.path.dirname(HOOK))) + import rulebook_hook as H + + calls = [] + saved = (H.spawn_fetch, H.fetch_book, H.load_rules, H.upgrade_status, H.BASE, + os.environ.get("MEMHUB_RULEBOOK_FETCH")) + H.spawn_fetch = lambda repo: calls.append(("spawn", repo)) + H.fetch_book = lambda repo, timeout=None: calls.append(("fetch", repo, timeout)) + H.load_rules = lambda repo: (["new"], "v", H._now(), {"s": 2}) + H.upgrade_status = lambda repo: None + os.environ["MEMHUB_RULEBOOK_FETCH"] = "1" + try: + with tempfile.TemporaryDirectory() as td: + H.BASE = td + out = H.first_fetch("r", [], None, {}) + check("no cached book: one bounded synchronous fetch, then the reloaded book, no child", + calls == [("fetch", "r", H.SESSION_FETCH_TIMEOUT_S)] and out == (["new"], out[1], {"s": 2}) + and out[1] is not None, str((calls, out))) + calls.clear() + out = H.first_fetch("r", [], None, {}) + check("still no book a moment later (server down): no second fetch within the minute", + calls == [] and out == ([], None, {}), str((calls, out))) + H.maybe_refresh("r", None) + check("…and maybe_refresh starts no detached child either: the stamp is shared", + calls == [], str(calls)) + os.remove(H.book_path("r") + ".refresh") + H.maybe_refresh("r", None) + check("stamp gone: maybe_refresh's detached child again, as before", + calls == [("spawn", "r")], str(calls)) + calls.clear() + stale = (datetime.now(timezone.utc) - timedelta(seconds=H.REFRESH_AFTER_S + 5)).isoformat() + out = H.first_fetch("s", ["old"], stale, {"s": 1}) + check("a cached book, however stale, is not first_fetch's: untouched, no fetch", + calls == [] and out == (["old"], stale, {"s": 1}), str((calls, out))) + out = H.first_fetch("", [], None, {}) + # A projectless Claude call's book (repo="": rules binding every + # repo) is fetched like any other; Codex never reaches here. + check("no repo named: the projectless book is fetched like any other", + calls == [("fetch", "", H.SESSION_FETCH_TIMEOUT_S)] and out[0] == ["new"], str((calls, out))) + calls.clear() + H.upgrade_status = lambda repo: {"scope": "org", "minimum_version": "9.9.9"} + out = H.first_fetch("u", [], None, {}) + check("an upgrade notice suspends the book: no fetch", calls == [], str(calls)) + H.upgrade_status = lambda repo: None + os.environ["MEMHUB_RULEBOOK_FETCH"] = "0" + out = H.first_fetch("f", [], None, {}) + check("MEMHUB_RULEBOOK_FETCH=0: nothing", calls == [], str(calls)) + finally: + H.spawn_fetch, H.fetch_book, H.load_rules, H.upgrade_status, H.BASE = saved[:5] + if saved[5] is None: + os.environ.pop("MEMHUB_RULEBOOK_FETCH", None) + else: + os.environ["MEMHUB_RULEBOOK_FETCH"] = saved[5] + + +_ROTATION_CRASH = r""" +import os, sys +sys.path.insert(0, sys.argv[1]) +import rulebook_hook as H +H.BASE = sys.argv[2] +when, real = sys.argv[3], os.replace +def replace(src, dst): + if when == "before-intent" and dst.endswith(".sent") and "rotating" in open(src).read(): + os._exit(9) + if when == "before-swap" and dst.endswith("fires.jsonl"): + os._exit(9) + real(src, dst) + if when == "after-swap" and dst.endswith("fires.jsonl"): + os._exit(9) +H.os.replace = replace +H.rulebook_ledger.rotate(os.path.join(H.BASE, "ledger"), [x[:2] for x in H.LEDGERS], + H._LEDGER_TAIL_BYTES) +os._exit(0) +""" + + +def ledger_rotation_checks() -> None: + """fires.jsonl / events.jsonl are rotated (ENG-1201) — and rotation may + only ever drop rows the server already confirmed: rows below the `.sent` + watermark. Every check here compares what the NEXT flush would post with + exactly the rows that were unsent before, so a lost row and a re-sent row + both fail it. Crashes are real (`os._exit` in a child) at each step.""" + import sys as _sys + import threading + _sys.path.insert(0, os.path.join(os.path.dirname(HOOK))) + import rulebook_hook as H + L = H.rulebook_ledger + + def rotate(ldir): # what flush_fires runs first, under .flush.lock + L.rotate(ldir, [x[:2] for x in H.LEDGERS], H._LEDGER_TAIL_BYTES) + + saved_base = H.BASE + + def setup(td, n=3000, sent_rows=2000, pad=480): + H.BASE = td + ldir = H._ledger_dir() + rows = [{"fire_id": f"f{i:05d}", "rule_id": "r", "session_id": f"s{i % 7}", + "fired_at": f"2026-10-05T00:{i // 60 % 60:02d}:{i % 60:02d}+00:00", + "excerpt": "x" * pad} for i in range(n)] + data = b"".join((json.dumps(r) + "\n").encode() for r in rows) + with open(os.path.join(ldir, "fires.jsonl"), "wb") as f: + f.write(data) + wm = sum(len((json.dumps(r) + "\n").encode()) for r in rows[:sent_rows]) + H._atomic_json(H._sent_path(), {"fires_offset": wm, "events_offset": 0, + "last_flush_at": None}) + return ldir, [r["fire_id"] for r in rows], wm + + def next_flush_ids(): + batches, _ = H.pending_batches(H.load_sent()) + return [r["fire_id"] for b, _ in batches for r in b] + + def on_disk_ids(ldir): + return [json.loads(x)["fire_id"] for x in + open(os.path.join(ldir, "fires.jsonl"), "rb").read().splitlines() if x] + + try: + # ── the happy path ─────────────────────────────────────────────── + with tempfile.TemporaryDirectory() as td: + ldir, ids, wm = setup(td) + size0 = os.path.getsize(os.path.join(ldir, "fires.jsonl")) + tails = {f"s{k}": H._earliest_fire_after(f"s{k}", "2026-10-05T00:00:00+00:00") + for k in range(7)} + rotate(ldir) + kept = on_disk_ids(ldir) + size1 = os.path.getsize(os.path.join(ldir, "fires.jsonl")) + check("rotation: a ledger past 1 MB shrinks", size0 > L.ROTATE_AT and size1 < size0, + f"{size0} -> {size1}") + check("rotation: what stays is a whole-line suffix of the ledger (every unsent row kept)", + kept == ids[len(ids) - len(kept):] and len(kept) >= 1000, str(len(kept))) + check("rotation: the next flush posts exactly the rows that were unsent", + next_flush_ids() == ids[2000:], str(next_flush_ids()[:3])) + check("rotation: the newest 256 KB stay, so the turn_end tail reader sees the same fires", + size1 >= H._LEDGER_TAIL_BYTES - 1024 and tails == { + k: H._earliest_fire_after(k, "2026-10-05T00:00:00+00:00") for k in tails}, + str(size1)) + sent = H.load_sent() + check("rotation: no intent and no half-written file left behind", + "rotating" not in sent and not [n for n in os.listdir(ldir) if n.startswith(".rot-")], + str((sent, os.listdir(ldir)))) + rotate(ldir) + check("rotation: a second pass with nothing confirmed to drop changes nothing", + on_disk_ids(ldir) == kept and next_flush_ids() == ids[2000:]) + + # Mostly confirmed: the cut is set by the 256 KB tail, not the + # watermark, so confirmed rows stay above the cut and the watermark + # must move down by exactly the cut. + with tempfile.TemporaryDirectory() as td: + ldir, ids, wm = setup(td, sent_rows=2900) + tails = {f"s{k}": H._earliest_fire_after(f"s{k}", "2026-10-05T00:00:00+00:00") + for k in range(7)} + rotate(ldir) + kept = on_disk_ids(ldir) + check("rotation (tail-bound cut): keeps a whole-line suffix past the unsent rows", + kept == ids[len(ids) - len(kept):] and 100 < len(kept) < 1000, str(len(kept))) + check("rotation (tail-bound cut): the next flush posts exactly the unsent rows", + next_flush_ids() == ids[2900:], str(next_flush_ids()[:3])) + check("rotation (tail-bound cut): the tail reader sees the same fires", + os.path.getsize(os.path.join(ldir, "fires.jsonl")) >= H._LEDGER_TAIL_BYTES - 1024 + and tails == {k: H._earliest_fire_after(k, "2026-10-05T00:00:00+00:00") + for k in tails}) + + # flush_fires runs it, under its lock, before (and whatever) it posts + with tempfile.TemporaryDirectory() as td: + ldir, ids, wm = setup(td) + real_api, H._api = H._api, lambda: None # no credential: nothing posts + try: + H.flush_fires(final=True) + finally: + H._api = real_api + check("rotation: flush_fires rotates an oversized ledger, exactly", + len(on_disk_ids(ldir)) < len(ids) and next_flush_ids() == ids[2000:]) + + # ── when it must not rotate ────────────────────────────────────── + with tempfile.TemporaryDirectory() as td: + ldir, ids, wm = setup(td, n=1500) # < 1 MB + rotate(ldir) + check("rotation: a ledger under 1 MB is left alone", on_disk_ids(ldir) == ids) + with tempfile.TemporaryDirectory() as td: + ldir, ids, wm = setup(td, sent_rows=200) # mostly unsent + rotate(ldir) + check("rotation: little confirmed → nothing to gain, nothing cut", + on_disk_ids(ldir) == ids and next_flush_ids() == ids[200:]) + with tempfile.TemporaryDirectory() as td: + ldir, ids, wm = setup(td) + with open(os.path.join(ldir, "conversions.jsonl"), "w") as f: + f.write(json.dumps({"fire_id": "f00001", "converted": True}) + "\n") + rotate(ldir) + check("rotation: fires stay whole while pre-v0.59 verdicts wait to drain " + "(the drain reads fire rows from the start)", on_disk_ids(ldir) == ids) + + # ── a crash at every step ──────────────────────────────────────── + for when, n_sent in [(w, n) for w in ("before-intent", "before-swap", "after-swap") + for n in (2000, 2900)]: + with tempfile.TemporaryDirectory() as td: + ldir, ids, wm = setup(td, sent_rows=n_sent) + p = subprocess.run([sys.executable, "-c", _ROTATION_CRASH, + os.path.dirname(HOOK), td, when], + capture_output=True, text=True, timeout=60) + crashed = p.returncode == 9 + H.BASE = td + pending = H.load_sent().get("rotating") + L.finish(ldir) # what the next flush does first + check(f"rotation crash {when}/{n_sent}: the next flush posts exactly the unsent rows", + crashed and next_flush_ids() == ids[n_sent:], + f"rc={p.returncode} {p.stderr[-300:]} intent={pending} " + f"first={next_flush_ids()[:2]} n={len(next_flush_ids())}") + check(f"rotation crash {when}/{n_sent}: the intent and any half-written file are gone", + "rotating" not in H.load_sent() + and not [n for n in os.listdir(ldir) if n.startswith(".rot-")]) + rotate(ldir) # and a later rotation still works + check(f"rotation crash {when}/{n_sent}: a later rotation completes, still exact", + next_flush_ids() == ids[n_sent:] and len(on_disk_ids(ldir)) < len(ids)) + + # ── an append waits for a rotation in progress ────────────────── + with tempfile.TemporaryDirectory() as td: + ldir, ids, wm = setup(td, n=10) + lock = L.append_lock(ldir) + done = threading.Event() + t = threading.Thread(target=lambda: (H.log_event({"session": "s"}, "turn_end"), done.set())) + t.start() + waited = not done.wait(0.5) + L.release(lock) + t.join(10) + ev = os.path.join(ldir, "events.jsonl") + check("rotation: a ledger append waits while a rotation holds the append lock", + waited and done.is_set() and os.path.exists(ev) and len(open(ev).readlines()) == 1) + finally: + H.BASE = saved_base + + +def server_shape_edge_checks() -> None: + """Two rule shapes the server accepts that this hook used to drop silently: + a regex past 400 characters (the server allows 2000, floored at 0.88.0) and + an edit rule with content_rx but no path_rx (validation takes either).""" + import sys as _sys + _sys.path.insert(0, os.path.join(os.path.dirname(HOOK))) + import rulebook_hook as H + + long_rx = "longpattern-hit|" + "|".join("alt%04d" % i for i in range(70)) + r = H.to_hook_rule(_row("long-rx", {"event": "bash", "command_rx": long_rx}, + min_hook_version="0.88.0")) + check("edge: a %d-char regex the server accepted loads" % len(long_rx), + len(long_rx) > 400 and r is not None, str(r)[:200]) + check("edge: …and fires on its command", + r is not None and bool(H.evaluate(r, hook_phase="pre", tool="Bash", + cmd="echo longpattern-hit"))) + check("edge: a regex past the server's 2000 cap is still dropped", + H.to_hook_rule(_row("too-long", {"event": "bash", "command_rx": "a" * 2001})) is None) + + r = H.to_hook_rule(_row("content-only", {"event": "edit", "content_rx": "TODO"})) + check("edge: a content-only edit rule loads", r is not None, str(r)) + check("edge: …and fires on any path whose body matches", + r is not None and bool(H.evaluate(r, hook_phase="pre", tool="Edit", + file_path="/x/a.py", body="# TODO later"))) + check("edge: …and stays silent when the body does not match", + r is not None and not H.evaluate(r, hook_phase="pre", tool="Edit", + file_path="/x/a.py", body="# done")) + + +def run_cache_checks() -> None: + """The hook process keeps two caches so a tool call pays for neither the + per-row lint of an unchanged book nor a `git` subprocess to name an + unchanged checkout. Both must be invisible: a hit returns exactly what a + cold load returns, any change to what the answer depends on misses, and a + corrupt cache is the cold path. In-process, `H._rc` is set by hand (only + `__main__` imports rulebook_cache) and BASE points at a temp dir.""" + import sys as _sys + _sys.path.insert(0, os.path.join(os.path.dirname(HOOK))) + import rulebook_cache as RC + import rulebook_hook as H + RI = H.repo_identity # the hook's own pinned copy, not a second import + + rows = [ + _row("rc-bash", {"event": "bash", "command_rx": r"\bgit\s+push\b", + "command_not_rx": "--dry-run", "warn_once_per": "turn"}, mode="gate"), + _row("rc-given", {"event": "bash", "command_rx": r"\bpytest\b", + "given": {"repo": {"branch_rx": "^main$"}}}), + _row("rc-output", {"event": "output", "command_rx": "pytest", "content_rx": "FAILED"}), + _row("rc-prompt", {"event": "prompt", "prompt_rx": "ship it"}), + _row("rc-session", delivery="session_context"), + _row("rc-future", {"event": "bash", "command_rx": "deploy"}, mode="gate", + min_hook_version="99.0.0"), + _row("rc-bad-rx", {"event": "bash", "command_rx": "(a+)+"}), + _row("rc-bash", {"event": "bash", "command_rx": "dup"}), # duplicate id: first wins + {"id": "rc-pilot", "on": "edit", "path_rx": r"\.py$", "text": "pilot", "why": "w"}, + ] + old_base, old_flag = H.BASE, H._rc + with tempfile.TemporaryDirectory() as td: + H.BASE = td + try: + seed_book(td, "cache-repo", rows) + book = H.book_path("cache-repo") + norm = book + ".norm" + H._rc = None + fresh = H.load_rules("cache-repo") + check("cache: an import never writes the normalised book", not os.path.exists(norm)) + H._rc = RC + cold = H.load_rules("cache-repo") + check("cache: the hook process writes it beside the book", os.path.exists(norm)) + + calls = [] + real = H.to_hook_rule + H.to_hook_rule = lambda row: calls.append(row) or real(row) + try: + hit = H.load_rules("cache-repo") + check("cache: a hit re-lints no row", calls == [], str(len(calls))) + check("cache: a hit returns exactly the cold load (rules, fetched_at, sources)", + hit == cold == fresh and [r["id"] for r in hit[0]][0] == "rc-bash" + and "rc-bad-rx" not in hit[3] and len(hit[0]) == 7, str(hit)[:300]) + + # Same length, same file (in place, so the inode stays): only the bytes differ. + with open(book, encoding="utf-8") as f: + text = f.read() + with open(book, "w", encoding="utf-8") as f: + f.write(text.replace("rc-session text", "rc-sessXon text")) + calls.clear() + edited = H.load_rules("cache-repo") + check("cache: an edited book misses and is re-linted", + len(calls) == len(rows) + and any(r.get("text") == "rc-sessXon text" for r in edited[0]), + str([r.get("text") for r in edited[0]])) + + H._HOOK_VERSION[:] = [(99, 0, 0)] + calls.clear() + newer = H.load_rules("cache-repo") + future = [r for r in newer[0] if r["id"] == "rc-future"] + check("cache: a different hook version misses (min_hook_version re-judged)", + len(calls) == len(rows) and future and future[0].get("mode") == "gate" + and "_degraded" not in future[0], str(future)) + H._HOOK_VERSION[:] = [] + + # the last one carries the RIGHT key: only the row check refuses it + real_key = [H.hashlib.sha1(open(book, "rb").read()).hexdigest(), + RC.hook_identity(H.__file__, H.hook_version(), H._ACTIVE_BASE)] + for garbage in ("{not json", json.dumps({"key": "x", "rules": []}), + json.dumps({"key": real_key, "rules": [{"id": ""}]})): + with open(norm, "w", encoding="utf-8") as f: + f.write(garbage) + calls.clear() + H._HOOK_VERSION[:] = [] + got = H.load_rules("cache-repo") + H._rc = None + want = H.load_rules("cache-repo") + H._rc = RC + check("cache: a corrupt or foreign cache is the cold path (%s…)" % garbage[:12], + got == want and len(calls) >= len(rows), str(got)[:200]) + finally: + H.to_hook_rule = real + H._HOOK_VERSION[:] = [] + + # End to end, through the real process: the second call is served + # from the cache and fires the same gate with the same text. + repo = os.path.join(td, "cache-repo") + os.makedirs(os.path.join(repo, ".git")) + with open(os.path.join(repo, ".git", "HEAD"), "w", encoding="utf-8") as f: + f.write("ref: refs/heads/feature\n") + os.remove(norm) + env = {"MEMHUB_RULEBOOK_BASE": td, "MEMHUB_RULEBOOK_FETCH": "0"} + outs = [] + for i in range(2): + _, out = run("pre", {"session_id": f"rc-{i}", "cwd": repo, "tool_name": "Bash", + "tool_input": {"command": "git push origin feature"}}, env) + outs.append(out) + check("cache: the hook process writes the cache on a cold call", os.path.exists(norm)) + check("cache: a cached call denies with the same text as the cold one", + '"deny"' in outs[0] and outs[0] == outs[1].replace("rc-1", "rc-0"), outs[1][:300]) + with open(norm, "w", encoding="utf-8") as f: + f.write("\0garbage") + _, out = run("pre", {"session_id": "rc-0", "cwd": repo, "tool_name": "Bash", + "tool_input": {"command": "git push origin feature"}}, env) + check("cache: a corrupt cache still gates, silently", out == outs[0], out[:300]) + finally: + H.BASE, H._rc = old_base, old_flag + + # The repo name: no `git` on a hit, a new name the moment the remote moves. + with tempfile.TemporaryDirectory() as td: + H.BASE = td + H._rc = RC + real_remote = RI._remote_basename + try: + main = os.path.join(td, "dir-name") + os.makedirs(main) + _git(main, "init", "-q") + _git(main, "remote", "add", "origin", "https://github.com/Org/first.git") + gitdir = os.path.join(main, ".git") + names = os.path.join(td, RC.REPO_NAMES_FILE) + + def name(): + RC._REPO_NAME_MEMO.clear() + RI._CACHE.clear() + return H.repo_info(main)[0] + + check("repo cache: a cold call names the repo from its remote", name() == "first") + check("repo cache: …and stores it under BASE", os.path.exists(names)) + spawned = [] + RI._remote_basename = lambda d: spawned.append(d) or real_remote(d) + check("repo cache: a hit spawns no git", name() == "first" and spawned == [], str(spawned)) + _git(main, "remote", "set-url", "origin", "https://github.com/Org/second.git") + check("repo cache: a changed remote URL misses and re-resolves", + name() == "second" and spawned, str(spawned)) + + # A linked worktree reads the MAIN checkout's config. + wt = os.path.join(td, "wt-dir") + _git(main, "commit", "-q", "--allow-empty", "-m", "x") + _git(main, "worktree", "add", "-q", wt) + RC._REPO_NAME_MEMO.clear() + first = H.repo_info(wt)[0] + _git(main, "remote", "set-url", "origin", "https://github.com/Org/third.git") + RC._REPO_NAME_MEMO.clear() + RI._CACHE.clear() + check("repo cache: a worktree follows its main checkout's remote", + first == "second" and H.repo_info(wt)[0] == "third", f"{first} {H.repo_info(wt)[0]}") + + with open(names, "w", encoding="utf-8") as f: + f.write("[corrupt") + check("repo cache: a corrupt map re-resolves", name() == "third") + + # git could not answer (missing / past its half second): the + # fallback name is served but never stored. + os.remove(names) + RI._remote_basename = lambda d: None + check("repo cache: an unsettled lookup falls back to the directory", + name() == "dir-name") + check("repo cache: …and is not stored", not os.path.exists(names)) + RI._remote_basename = real_remote + check("repo cache: the next call asks git again", name() == "third") + + # A git that fails for any reason but "no such remote" (exit 128, + # say) did not answer: its fallback is served, never stored. + os.remove(names) + fake_bin = os.path.join(td, "fake-bin") + os.makedirs(fake_bin) + with open(os.path.join(fake_bin, "git"), "w", encoding="utf-8") as f: + f.write("#!/bin/sh\necho 'fatal: unexpected failure' >&2\nexit 128\n") + os.chmod(os.path.join(fake_bin, "git"), 0o755) + old_path = os.environ["PATH"] + os.environ["PATH"] = fake_bin + os.pathsep + old_path + try: + check("repo cache: a git failing with exit 128 falls back to the directory", + name() == "dir-name") + finally: + os.environ["PATH"] = old_path + check("repo cache: …and that guess is not stored", not os.path.exists(names)) + # "No such remote" IS an answer: a remote-less checkout's name is stored. + bare = os.path.join(td, "no-origin") + os.makedirs(bare) + _git(bare, "init", "-q") + RC._REPO_NAME_MEMO.clear() + RI._CACHE.clear() + check("repo cache: a checkout with no origin is named by its directory and stored", + H.repo_info(bare)[0] == "no-origin" and os.path.exists(names) + and any(k.startswith(bare + "\0") for k in json.load(open(names))), + str(os.path.exists(names))) + finally: + RI._remote_basename = real_remote + RC._REPO_NAME_MEMO.clear() + RI._CACHE.clear() + H.BASE, H._rc = old_base, old_flag + + +def edit_added_checks() -> None: + """An edit rule's `content_rx` reads what the edit ADDED, not the lines its + new_string kept: a gate on `create_task(` blocked an Edit that only kept an + existing call (24 such files in MemHub-Backend app/), and comment rules + fired on comments the edit merely re-wrapped. `content_not_rx` still reads + the whole new text — the complied-with form may sit on a kept line.""" + with tempfile.TemporaryDirectory() as td: + repo = os.path.join(td, "editrepo") + os.makedirs(repo) + assert _git(repo, "-c", "init.defaultBranch=main", "init", "-q").returncode == 0 + env = {"MEMHUB_RULEBOOK_BASE": td, "MEMHUB_RULEBOOK_FETCH": "0", "MEMHUB_RULEBOOK_JUDGE": "0"} + seed_book(td, "editrepo", [_row("task-gate", {"event": "edit", "path_rx": r"\.py$", + "content_rx": r"create_task\(", + "content_not_rx": "# task-ok"}, + mode="gate", statement="No bare create_task")]) + kept = "x = 1\nasyncio.create_task(job())" + + def edit(sid, old, new, **inp): + ti = dict({"file_path": os.path.join(repo, "a.py"), "old_string": old, "new_string": new}, **inp) + return run("pre", {"cwd": repo, "session_id": sid, "tool_name": "Edit", "tool_input": ti}, env)[1] + + out = edit("e1", kept, kept + "\ny = 2") + check("edit: an Edit that only KEEPS a create_task line is not blocked", "task-gate" not in out, out) + out = edit("e2", "x = 1", kept) + check("edit: an Edit that ADDS a create_task line is", "task-gate" in out, out) + out = edit("e3", "x = 1", kept + " # task-ok") + check("edit: content_not_rx still exempts the added line", "task-gate" not in out, out) + out = edit("e4", "x = 1\nasyncio.create_task(job()) # task-ok", "x = 2\nasyncio.create_task(job()) # task-ok") + check("edit: a kept, complied-with line beside a changed one stays quiet", "task-gate" not in out, out) + out = run("pre", {"cwd": repo, "session_id": "e5", "tool_name": "MultiEdit", "tool_input": { + "file_path": os.path.join(repo, "a.py"), + "edits": [{"old_string": kept, "new_string": kept + "\nz = 3"}, + {"old_string": "q = 0", "new_string": "q = 0\nloop.create_task(other())"}]}}, env)[1] + check("edit: MultiEdit — the one edit that adds a create_task line fires", "task-gate" in out, out) + out = run("pre", {"cwd": repo, "session_id": "e6", "tool_name": "Write", "tool_input": { + "file_path": os.path.join(repo, "b.py"), "content": kept}}, env)[1] + check("edit: a Write is still read whole (its old file is not read)", "task-gate" in out, out) + + def prompt_lane_checks() -> None: """`event: prompt` (ENG-1184): a rule that speaks when a wake-up arrives. A `/loop` tick or a `<task-notification>` reaches the hook only as a @@ -2770,6 +3660,24 @@ def fires(rid): "tool_input": {"command": "echo '<task-notification>'"}}, env)[1]) check("prompt: a prompt rule never fires on a tool call", c == "", c) + # Another agent's message — a subagent's hand-back — is nobody's prompt: + # a rule written for what the person asked must not fire on a report + # that merely mentions it. A wake-up still does (the lane's purpose). + spec = _row("spec-ask", {"event": "prompt", "prompt_rx": r"\bspec\b", + "warn_once_per": "turn"}) + seed_book(td, "promptrepo", [spec, every]) + out = prompt("h1", "find the spec behind this work") + check("prompt: a typed prompt fires a prompt rule", "spec-ask" in out, out) + handback = ('Another Claude session sent a message:\n<agent-message from="a1">\n' + "[Subagent hand-back] I read the spec and the code.\n" + wake + "\n</agent-message>") + n_spec, n_every = len(fires("spec-ask")), len(fires("every-tick")) + out = prompt("h1", handback) + check("prompt: a subagent hand-back fires no prompt rule, even one it quotes a wake-up for", + out.strip() == "" and len(fires("spec-ask")) == n_spec + and len(fires("every-tick")) == n_every, out) + out = prompt("h1", wake) + check("prompt: …a real wake-up still fires", "every-tick" in out, out) + def min_hook_version_checks() -> None: """A rule can be newer than the hook reading it, and that used to be @@ -3479,5 +4387,6 @@ def decision(out): H.BASE = old_base + if __name__ == "__main__": sys.exit(main()) diff --git a/tests/rulebook_repo_of_call_test.py b/tests/rulebook_repo_of_call_test.py index 8d081e8f..25d67c94 100644 --- a/tests/rulebook_repo_of_call_test.py +++ b/tests/rulebook_repo_of_call_test.py @@ -16,11 +16,14 @@ back to the MAIN checkout, so a rule scoped to the repo matches from any of its worktrees — the machinery was always correct, only the starting point was wrong; -* a Bash call carries no path, so a non-git cwd stays silent exactly as - before (the fail-open property the hook is built on); -* the session cwd is the trust boundary: a path outside it is ignored, a - symlink under it cannot smuggle the lookup out, and a relative path binds - to the SESSION's cwd rather than this process's; +* a Bash call that names no directory keeps the cwd, so a non-git cwd stays + silent exactly as before (the fail-open property the hook is built on); +* what the call TOUCHES decides, wherever it is: a file in another checkout, + or a Bash command's `cd` chain / `git -C` (`~`, `$HOME`, `$VAR`, relative + paths resolved), so the target repo's book judges it — not the session's + (2026-10 audit: 28 of 56 explained misses were a call judged by the + session repo's book); a target that is no checkout keeps the cwd answer; + a relative path binds to the SESSION's cwd rather than this process's; * `root` comes back in the session's own path space, because it keys OrderingEngine state (`{rid}@{root}:{branch}`) and one worktree reached two ways would split into two keys and silently re-arm its rules; @@ -125,7 +128,7 @@ def mkworktree(parent: str, main_repo: str, name: str, branch: str) -> str: def unit_checks() -> None: - """`repo_of_call` and `_acted_on_dir` are pure and importable — replayed + """`repo_of_call` and `_call_seed` are pure and importable — replayed here, so what is tested is what runs.""" with tempfile.TemporaryDirectory() as td: container = os.path.join(td, "container") # holds checkouts, is not one @@ -163,7 +166,8 @@ def call(cwd, **inp): # --- no path to act on: unchanged, still silent ---------------------- check("a Bash call from a container cwd still resolves nothing", - call(container, command="git push") == ("", "", "", "")) + rb.repo_of_call({"cwd": container, "tool_name": "Bash", + "tool_input": {"command": "git push"}}) == ("", "", "", "")) check("no tool_input at all (SessionStart) still resolves nothing", rb.repo_of_call({"cwd": container}) == ("", "", "", "")) # Assert the PAIR: main and the worktree now share a repo NAME, so a @@ -172,33 +176,40 @@ def call(cwd, **inp): check("a cwd inside a checkout is unaffected when no file is named", rb.repo_of_call({"cwd": wt})[:2] == ("MainRepo", wt)) - # --- the trust boundary --------------------------------------------- - check("an absolute path OUTSIDE the session cwd is ignored", - call(container, file_path=os.path.join(outside, "app", "svc.py")) - == ("", "", "", "")) - - link = os.path.join(container, "escape") - try: - os.symlink(outside, link) - check("a symlink under cwd cannot smuggle the lookup outside it", - call(container, file_path=os.path.join(link, "app", "svc.py")) - == ("", "", "", "")) - except (OSError, NotImplementedError): # unprivileged Windows - print("SKIP symlink containment (symlinks unavailable)") - - # The DANGEROUS direction, and the one a resolved-only check misses: a - # link OUTSIDE cwd whose target is inside it. Containment on the real - # path passes, while the lexical parents still lead to another - # checkout — which is the one `repo_info` would walk up into. - inward = os.path.join(td, "outside-link") - try: - os.symlink(os.path.join(container, "MainRepo-feature"), inward) - check("a symlink outside cwd pointing INTO it does not escape the boundary", - call(container, file_path=os.path.join(inward, "app", "svc.py")) - == ("", "", "", ""), - str(call(container, file_path=os.path.join(inward, "app", "svc.py")))) - except (OSError, NotImplementedError): - print("SKIP inward-symlink containment (symlinks unavailable)") + # --- the call's own target decides, wherever it is ------------------ + def bash(cwd, command, home=None): + saved = os.environ.get("HOME") + if home is not None: + os.environ["HOME"] = home + try: + return rb.repo_of_call({"cwd": cwd, "tool_name": "Bash", + "tool_input": {"command": command}}) + finally: + if home is not None: + os.environ["HOME"] = saved if saved is not None else "" + + check("an Edit of a file in ANOTHER checkout resolves that checkout", + call(wt, file_path=os.path.join(outside, "app", "svc.py"))[:2] + == ("OutsideRepo", outside)) + check("`cd <other repo> && git push` from a checkout resolves the other repo", + bash(wt, f"cd {outside} && git push")[:2] == ("OutsideRepo", outside)) + check("`cd ~/<other repo> && git push` expands ~ and resolves it", + bash(wt, "cd ~/OutsideRepo && git push", home=td)[:2] == ("OutsideRepo", outside)) + check("`cd $HOME/…` and `${HOME}` expand too", + bash(wt, "cd $HOME/OutsideRepo && git push", home=td)[:2] == ("OutsideRepo", outside) + and bash(wt, 'cd "${HOME}/OutsideRepo"; git push', home=td)[:2] == ("OutsideRepo", outside)) + check("a relative `cd ../..` chain resolves against the cwd", + bash(wt, "cd .. && cd ../OutsideRepo && git status")[:2] == ("OutsideRepo", outside)) + check("`git -C <other repo>` resolves the other repo", + bash(wt, f"git -C {outside} push")[:2] == ("OutsideRepo", outside)) + check("a `cd` into a subdirectory of the same checkout keeps it", + bash(wt, "cd app && ls")[:2] == ("MainRepo", wt)) + check("a `cd` to a directory that is no checkout keeps the cwd's", + bash(wt, f"cd {td} && ls")[:2] == ("MainRepo", wt)) + check("an unresolvable `cd \"$(…)\"` keeps the cwd's", + bash(wt, f'cd "$(echo {outside})" && git push')[:2] == ("MainRepo", wt)) + check("an unset variable keeps the cwd's", + bash(wt, "cd $NO_SUCH_VAR_RB && git push")[:2] == ("MainRepo", wt)) # --- one path space per session (the OrderingEngine key) ------------ # An absolute path spelled in a different-but-equivalent space @@ -266,12 +277,11 @@ def call(cwd, **inp): call(container, notebook_path=os.path.join(wt, "app", "n.ipynb"))[:2] == ("MainRepo", wt)) check("a junk file_path does not shadow a good notebook_path", - rb._acted_on_dir(container, {"file_path": 123, - "notebook_path": os.path.join(wt, "app", "n.ipynb")}) - == os.path.join(wt, "app")) + call(container, file_path=123, notebook_path=os.path.join(wt, "app", "n.ipynb"))[:2] + == ("MainRepo", wt)) check("a file that resolves nowhere falls back to the cwd's checkout", rb.repo_of_call({"cwd": wt, "tool_input": - {"file_path": os.path.join(outside, "a.py")}})[0] + {"file_path": os.path.join(td, "loose", "a.py")}})[0] == "MainRepo") @@ -321,15 +331,68 @@ def hook_checks() -> None: check("Bash from a container cwd is still silent, exit 0", rc == 0 and out.strip() == "", out) - # A file outside the session cwd must not reach across into another - # checkout's book, even when one is cached for it. + # A file in ANOTHER checkout is judged by that checkout's book: its + # scoped rule fires, and the session repo's rule does not. other = mkmain(td, "MainRepo2") - seed_book(base, "MainRepo2", [dict(RULE, _scope_repos=["MainRepo2"])]) - rc, out = run("pre", {"session_id": "o1", "cwd": container, "tool_name": "Write", + seed_book(base, "MainRepo2", [dict(RULE, id="metered-llm-2", _scope_repos=["MainRepo2"]), + {"id": "no-push-2", "on": "bash", "rx": r"\bgit\s+push\b", + "_scope_repos": ["MainRepo2"], "text": "MainRepo2 pushes go via CI", + "why": "w"}]) + rc, out = run("pre", {"session_id": "o1", "cwd": wt, "tool_name": "Write", "tool_input": {"file_path": os.path.join(other, "app", "x.py"), "content": BODY}}, env) - check("a file outside the session cwd loads no book and stays silent", - rc == 0 and out.strip() == "", out) + check("an Edit of a file in another repo is judged by that repo's book", + rc == 0 and "[metered-llm-2]" in ctx(out) and "[metered-llm]" not in ctx(out), out) + rc, out = run("pre", {"session_id": "o2", "cwd": wt, "tool_name": "Bash", + "tool_input": {"command": "cd ~/MainRepo2 && git push"}}, + dict(env, HOME=td)) + check("`cd ~/<other repo> && git push` fires the other repo's scoped rule", + rc == 0 and "[no-push-2]" in ctx(out), out) + rc, out = run("pre", {"session_id": "o3", "cwd": wt, "tool_name": "Bash", + "tool_input": {"command": "git push"}}, dict(env, HOME=td)) + check("…and the same push in the session's own repo does not", + rc == 0 and "[no-push-2]" not in ctx(out), out) + + # A Bash command that WRITES or READS a file in another repo's checkout, + # with no `cd` (the call stays the session repo's), has that file + # judged by the other repo's book — the live repro of PR #176's review. + repo_a = mkmain(td, "RepoA") # the session's repo, no rules + seed_book(base, "MainRepo2", [ + {"id": "no-data-files", "on": "edit", "path_rx": r"(^|/)[^/]*\.(csv|tsv)$", + "text": "Never commit data files to the repo", "why": "w"}, + {"id": "meter", "on": "edit", "path_rx": r"^(?:.*/)?app/.*\.py$", + "content_rx": r"(AsyncOpenAI|AsyncAnthropic)\(\s*api_key", "text": "Every model call reaches the meter", + "why": "w"}, + {"id": "no-meter-gate", "on": "edit", "mode": "gate", "path_rx": r"app/gated/.*\.py$", + "content_rx": r"AsyncOpenAI\(", "text": "Gated meter", "why": "w"}, + {"id": "readme-read", "on": "read", "path_rx": r"README\.md$", "text": "Read the README via the brain", + "why": "w"}, + {"id": "no-push-2", "on": "bash", "rx": r"\bgit\s+push\b", "_scope_repos": ["MainRepo2"], + "text": "MainRepo2 pushes go via CI", "why": "w"}]) + henv = dict(env, HOME=td) + for n, cmd, want in [ + (1, "echo a,b > ../MainRepo2/live_probe.csv", "[no-data-files]"), + (2, "tee $HOME/MainRepo2/app/services/live_probe_llm.py >/dev/null <<'PY'\n" + "client = AsyncOpenAI(api_key=key)\nPY", "[meter]"), + (3, "cat ~/MainRepo2/README.md", "[readme-read]"), + (4, f"printf 'x,y' > {other}/data/t.tsv", "[no-data-files]")]: + rc, out = run("pre", {"session_id": f"x{n}", "cwd": repo_a, "tool_name": "Bash", + "tool_use_id": f"x{n}", "tool_input": {"command": cmd}}, henv) + check(f"a Bash write/read into another repo is judged by that repo's book ({want})", + rc == 0 and want in ctx(out), out) + rc, out = run("pre", {"session_id": "x5", "cwd": repo_a, "tool_name": "Bash", + "tool_input": {"command": "cat > ~/MainRepo2/app/gated/m.py <<'EOF'\n" + "c = AsyncOpenAI()\nEOF"}}, henv) + check("…and that repo's gate-mode edit rule blocks the command", + rc == 0 and '"permissionDecision": "deny"' in out and "[no-meter-gate]" in out, out) + rc, out = run("pre", {"session_id": "x6", "cwd": repo_a, "tool_name": "Bash", + "tool_input": {"command": "echo a,b > ../MainRepo2/x.csv && git push"}}, henv) + check("…while the command's own rules stay the session repo's (no MainRepo2 push rule)", + "[no-data-files]" in ctx(out) and "[no-push-2]" not in ctx(out), out) + with open(os.path.join(base, "ledger", "fires.jsonl"), encoding="utf-8") as f: + rows = [json.loads(line) for line in f if '"x1"' in line] + check("…and the fire is recorded against the repo whose book fired it", + [r["repo"] for r in rows] == ["MainRepo2"], str(rows)) # An edit that does not match still says nothing — the fix widens where # rules are LOOKED for, never what counts as a match. diff --git a/tests/starter_rulebook_test.py b/tests/starter_rulebook_test.py index 888445f6..6fb3259a 100644 --- a/tests/starter_rulebook_test.py +++ b/tests/starter_rulebook_test.py @@ -297,10 +297,12 @@ def test_the_skill_asks_before_it_reads_and_warns_before_it_waits() -> None: check("--all is only ever the person's ask", "only when they ask for all of" in skill) check("one scope for the whole batch, decided before anything is filed", "one scope for the\nwhole batch" in skill) - check("…org-wide unless the person says otherwise, without asking", - "**Otherwise `org`** — do not ask" in skill) + check("…just the person unless the batch is plainly for everyone, without asking", + "**Otherwise `personal`** —" in skill and "Do not ask," in skill) + check("…and an explicit 'just me' is never escalated", + '"just me" / "only me" / "personal" → `personal`' in skill) check("…and the workspace scope only when they ask for it", - "never offer the\nworkspace scope unprompted" in skill) + "never offer the workspace scope unprompted" in skill) check("the skill never makes a rulebook", "create_rulebook" not in skill) @@ -710,6 +712,126 @@ def _verify(mutate): p.returncode == 0 and "why" not in body and body["when"] and body["do"]) +def _run_args(repo: Path, out: Path, *extra: str): + env = {k: v for k, v in os.environ.items() if k != "CLAUDE_PLUGIN_ROOT"} + p = subprocess.run([sys.executable, str(SCRIPT), "all", "--repo", str(repo), "--out", str(out), *extra], + capture_output=True, text=True, env=env, timeout=300) + load = lambda n: json.loads((out / n).read_text()) + return p, load("signals.json"), load("candidates.json") + + +def _verify_body(body: dict, fires: list, silent: list) -> tuple[bool, list]: + sys.path.insert(0, str(ROOT / "plugins" / "memhub" / "scripts")) + import rulebook_verify as V + ok, _, lines = V.verify(body, fires, silent) + return ok, lines + + +def test_a_test_run_launched_the_usual_ways_clears_the_push_gate() -> None: + """The receipt knew only bare `pytest`; on one repo 2,082 of 2,290 runs were + `uv run python -m pytest`, so the gate never cleared and people learned to override it.""" + with tempfile.TemporaryDirectory() as tmp: + repo = _make(PYTHON_REPO, Path(tmp) / "svc") + p, signals, cands = _run_args(repo, Path(tmp) / "out") + body = {c["id"]: c["body"] for c in cands}["suite-before-push"] + ran = ["uv run python -m pytest tests/", "uv run --extra dev pytest -q", ".venv/bin/pytest -q", + ".venv/bin/python3.12 -m pytest", "poetry run pytest", "$PY -m pytest tests/ -q", "uvx pytest"] + ok, lines = _verify_body(body, fires=["edit:src/a.py >> ok:uv run python -m pytest --version >> gate:git push", + "edit:src/a.py >> ok:grep -rn pytest docs/ >> gate:git push"], + silent=["edit:src/a.py >> ok:%s >> gate:git push" % r for r in ran]) + check("every launcher form clears suite-before-push; --version and a grep do not", ok, "\n".join(lines)) + + +def test_a_linter_launched_the_usual_ways_clears_the_lint_gate() -> None: + with tempfile.TemporaryDirectory() as tmp: + repo = _make(PYTHON_REPO, Path(tmp) / "svc") + p, signals, cands = _run_args(repo, Path(tmp) / "out") + body = {c["id"]: c["body"] for c in cands}["lint-before-push"] + ok, lines = _verify_body(body, fires=["edit:src/a.py >> ok:uv run ruff --version >> gate:git push"], + silent=["edit:src/a.py >> ok:%s >> gate:git push" % r for r in + ("uv run ruff check .", "uvx ruff check .", ".venv/bin/mypy app", + "uv run python -m mypy app", "ruff format --check .")]) + check("every launcher form clears lint-before-push", ok, "\n".join(lines)) + + +def test_the_cap_is_the_hooks_own() -> None: + """The starter cuts patterns to the hook's load cap, read from the hook — never a copy that can + go stale (one sat at 400 while the hook loaded 2000).""" + sys.path.insert(0, str(SCRIPT.parent)) + sys.path.insert(0, str(SKILL.parents[1] / "scripts")) + import rulebook_hook as H + import starter_rulebook as S + check("starter RX_MAX is the hook's _RX_MAX", S.RX_MAX == H._RX_MAX, "%s vs %s" % (S.RX_MAX, H._RX_MAX)) + + +def test_every_toolchain_mix_keeps_its_receipts_loadable() -> None: + """Python + two Node runners + Go + Rust: both receipts keep the full Python launcher and still + load.""" + files = {**PYTHON_REPO, **NODE_REPO, "go.mod": "module x\n", "a_test.go": "package x\n", + "Cargo.toml": "[package]\nname = 'x'\n", "src/lib.rs": "\n", "Makefile": "test:\n\tpytest\n"} + with tempfile.TemporaryDirectory() as tmp: + p, signals, cands = _run_args(_make(files, Path(tmp) / "poly"), Path(tmp) / "out") + slots = signals["slots"] + check("exit 0, every rule verifies", p.returncode == 0, p.stdout[-400:]) + sys.path.insert(0, str(SCRIPT.parent)) + import starter_rulebook as S + check("both receipts fit the hook's cap with CMD in front", + all(len(slots["CMD"]) + len(slots[k]) <= S.RX_MAX for k in ("test_receipt_rx", "lint_receipt_rx"))) + check("both receipts carry the full launcher", + all(slots[k].count(S.PY_LAUNCH) == 1 for k in ("test_receipt_rx", "lint_receipt_rx")), + str([slots[k] for k in ("test_receipt_rx", "lint_receipt_rx")])) + + +def test_a_linter_only_the_ci_runs_is_still_evidence() -> None: + files = {"pyproject.toml": '[project]\nname = "svc"\n', "app/a.py": "x = 1\n", + ".github/workflows/lint.yml": "jobs:\n l:\n steps:\n - run: uvx ruff check .\n"} + with tempfile.TemporaryDirectory() as tmp: + p, signals, _ = _run_args(_make(files, Path(tmp) / "ci"), Path(tmp) / "out") + check("ruff run by CI is found", "ruff check ." in (signals["signals"]["lint"].get("value") or []), + signals["signals"]["lint"]["evidence"]) + + +def _commits(repo: Path, branch: str, subjects: list, start: str = "trunk") -> None: + env = dict(os.environ, GIT_CONFIG_GLOBAL=os.devnull, GIT_CONFIG_SYSTEM=os.devnull) + g = lambda *a: subprocess.run(["git", "-C", str(repo), "-c", "user.email=t@t", "-c", "user.name=t", *a], + check=True, capture_output=True, env=env) + g("checkout", "-q", "-b", branch, start) + for s in subjects: + g("commit", "-q", "--allow-empty", "-m", s) + g("update-ref", "refs/remotes/origin/" + branch, "HEAD") + g("checkout", "-q", "trunk") + + +def test_the_branch_pull_requests_land_on_is_named_and_one_flag_guards_it() -> None: + """origin/HEAD is `trunk`, but the team merges every PR into `staging`. The scan names staging + (a hint, not a slot) and does NOT name a feature branch whose tip a merge bot versioned.""" + with tempfile.TemporaryDirectory() as tmp: + repo = _make(BARE_REPO, Path(tmp) / "flow") + _commits(repo, "staging", ["feat: a (#1)", "fix: b (#2)", "Merge pull request #3 from acme/c", "feat: d (#4)"]) + _commits(repo, "fm-feat", ["wip", "more work", "address review", "chore: version 1.2.0 (#9)"]) + p, signals, cands = _run_args(repo, Path(tmp) / "out") + ev = signals["signals"]["branch"] + check("the hint names staging and not the bot-versioned feature branch", + ev.get("also_landing") == ["staging"], str(ev)) + check("…and only names it: the guarded branch is still origin/HEAD alone", + signals["slots"]["default_branch_rx"] == "^(trunk)$") + check("…and says the flag to pass", "--default-branch trunk --default-branch staging" in ev["evidence"]) + + p, signals, cands = _run_args(repo, Path(tmp) / "out2", "--default-branch", "trunk", + "--default-branch", "staging") + check("exit 0 with two branches", p.returncode == 0, p.stdout[-400:]) + check("one flag fills every branch slot", signals["slots"]["default_branch_rx"] == "^(trunk|staging)$" + and signals["slots"]["default_branch"] == "trunk") + body = {c["id"]: c["body"] for c in cands} + rx = body["push-main-refspec"]["matcher"]["command_rx"] + for cmd in ("git push origin HEAD:staging", "git push origin trunk"): + check(f"refspec rule fires on `{cmd}`", bool(re.search(rx, cmd))) + check("…and not on a feature branch", not re.search(rx, "git push origin HEAD:fm-feat")) + ok, lines = _verify_body(body["push-main"], fires=[{"case": "git push", "branch": "staging"}], + silent=[{"case": "git push", "branch": "fm-feat"}]) + check("push-main guards a push made from staging", ok, "\n".join(lines)) + + if __name__ == "__main__": for name, fn in sorted(globals().items()): if name.startswith("test_") and callable(fn):