feat(hooks): Loop B — opt-in detached whole-repo report refresh (T2b / C5, hardened)

src/roam/commands/cmd_hooks.py

@@ -525,7 +525,10 @@ def status(ctx):
 # copies. The stamp is a bare comment line inserted after the shebang:
 # compile-code's mode-override surgery rewrites only the roam INVOCATION lines,
 # never this, so a healed body keeps its version marker.
-_HOOK_BODY_VERSION = 2
+# v3 (2026-07-16): Stop hook gained the opt-in Loop-B whole-repo report refresh
+# (ROAM_HOOK_REPORT_REFRESH). Deployed v2 bodies heal to v3 on `hooks claude
+# --write`, which is how this new body reaches already-wired installs.
+_HOOK_BODY_VERSION = 3
 _HOOK_VERSION_MARKER = "# roam-hook-version:"
 
 _CLAUDE_UPS_HOOK_FILENAME = "roam-compile-ups.py"
@@ -665,6 +668,10 @@ def _claude_hook_dir(user_level: bool) -> Path:
   ROAM_HOOK_VIBE     (default 0) -- attach `roam vibe-check`'s AI-rot
       score when it crosses ROAM_HOOK_VIBE_THRESHOLD (default 50).
       Repo-scoped (not diff-scoped), so advisory only. Closes F23.
+  ROAM_HOOK_REPORT_REFRESH (default 0) -- Loop B: on an edit-stop, spawn a
+      DETACHED, THROTTLED (>= 6h) whole-repo `verify --report --persist` so the
+      next compile's known_findings is fresh. Never blocks the stop; opt-in
+      because it runs a background whole-repo verify.
 """
 import json
 import os
@@ -717,6 +724,79 @@ def _env_on(name, default):
     return (os.environ.get(name, default) or "").strip().lower() not in ("", "0", "false", "no", "off")
 
 
+_REPORT_REFRESH_HOURS = 6.0
+_REFRESH_CLAIM_MINUTES = 30.0  # single-flight window: at most one spawn per claim age
+
+
+def _hook_repo_root():
+    """Nearest ancestor (cwd included) containing .git; cwd when none found.
+
+    The spawned verify persists at the PROJECT ROOT (find_project_root walks
+    up to .git) — the throttle and claim must anchor to the same directory,
+    or a hook run from a repo subdir respawns on every stop forever.
+    """
+    d = os.getcwd()
+    while True:
+        if os.path.exists(os.path.join(d, ".git")):
+            return d
+        parent = os.path.dirname(d)
+        if parent == d:
+            return os.getcwd()
+        d = parent
+
+
+def _maybe_refresh_whole_repo_report():
+    """Loop B: keep .roam/verify-report.json fresh so `compile`'s known_findings
+    probe can embed the repo's OPEN findings for the edited file.
+
+    DETACHED + THROTTLED + SINGLE-FLIGHT + fail-open, by hard requirement:
+      - detached: a whole-repo `verify --report` is ~minutes; running it inline
+        would blow the Stop-hook budget (it is the exact stall the diff-only
+        scope avoids). We fire-and-forget and never read its result here.
+      - never --diff-only: that persists a diff-SCOPED view into the whole-repo
+        report path, poisoning known_findings. This is a SEPARATE whole-repo run.
+      - throttled: skip if the report is younger than _REPORT_REFRESH_HOURS.
+      - single-flight: a claim marker bounds spawning to one verify per
+        _REFRESH_CLAIM_MINUTES per repo. The report mtime alone cannot close
+        the in-flight window (it only moves AFTER the ~minutes-long verify
+        persists), and it never moves at all when the persist cannot land
+        (empty targets, missing DB, crash) — without the claim every edit-stop
+        would respawn a whole-repo verify, each re-running ensure_index().
+    Opt-in (ROAM_HOOK_REPORT_REFRESH, default OFF): it spawns a background
+    whole-repo verify, so enabling it is the user's call; enabling closes Loop B.
+    """
+    if not _env_on("ROAM_HOOK_REPORT_REFRESH", "0"):
+        return
+    try:
+        root = _hook_repo_root()
+        report = os.path.join(root, ".roam", "verify-report.json")
+        try:
+            if (time.time() - os.path.getmtime(report)) / 3600.0 < _REPORT_REFRESH_HOURS:
+                return  # fresh enough — don't respawn every edit-stop
+        except OSError:
+            pass  # missing/unreadable -> refresh
+        claim = os.path.join(root, ".roam", "verify-refresh-claim")
+        try:
+            if (time.time() - os.path.getmtime(claim)) / 60.0 < _REFRESH_CLAIM_MINUTES:
+                return  # a refresh is (or was recently) in flight
+        except OSError:
+            pass  # no claim yet
+        os.makedirs(os.path.join(root, ".roam"), exist_ok=True)
+        with open(claim, "w") as fh:
+            fh.write("pid=%d time=%d" % (os.getpid(), int(time.time())))  # observability crumb
+        kwargs = dict(stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        if os.name == "nt":
+            kwargs["creationflags"] = 0x00000008 | 0x00000200  # DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP
+        else:
+            kwargs["start_new_session"] = True
+        # whole-repo (NO --diff-only) --report --persist -> <root>/.roam/verify-report.json
+        # cwd=root pins the spawned verify's project-root resolution to the SAME
+        # directory the throttle checked.
+        subprocess.Popen(["roam", "verify", "--auto", "--report", "--persist"], cwd=root, **kwargs)
+    except Exception:
+        return  # a refresh we could not spawn is never the user's problem
+
+
 def _run_roam(args, timeout, env=None):
     """Run `roam --json <args>`; return the parsed envelope or None. Fail-open."""
     try:
@@ -1003,6 +1083,13 @@ def main():
             d = _run_roam(["verify", "--auto", "--diff-only"], _VERIFY_TIMEOUT_S,
                           env={**os.environ, **_ADVISORY_ENV})
             verify_ms = int((time.perf_counter() - verify_t0) * 1000)
+            # Loop B (opt-in, detached, throttled): the edits just landed, so the
+            # persisted whole-repo report is now conceptually stale — kick off a
+            # background refresh so the NEXT compile's known_findings is current.
+            # Fire AFTER the blocking diff-only verify: a whole-repo verify
+            # competing for CPU/DB locks can push the gate past its 90s budget,
+            # and a timed-out gate silently allows — the worst possible trade.
+            _maybe_refresh_whole_repo_report()
         summary = (d or {}).get("summary") or {}
         verdict = str(summary.get("verdict") or "")
         verify_failed = bool(d) and bool(verdict) and not verdict.upper().startswith("PASS")
@@ -1126,6 +1213,9 @@ def _with_version_stamp(body: str) -> str:
 # defect that shipped in #77).
 _KNOWN_HOOK_BODY_SHAS: frozenset[str] = frozenset(
     {
+        "b28bcb7a414f92e1694ecbeb54ff1d5e69b8a4c46d4ee035e6b88975712e0805",  # stop v3 pristine (2026-07-16 loop-b)
+        "fa76db1e06bb44a947084ed10f94b553aa68289d60511cb246b67f5f85acfd44",  # stop v3 surgered (2026-07-16 loop-b)
+        "18e19f503c957e09850ec4173fc451b078c7a0356eb6c964d6406b9e5a8300a5",  # ups v3 pristine (2026-07-16 loop-b)
         "0313b8d53749fa9d188c9e6554b37826ff677cdd166627ab5b613538bb4b4573",  # stop v2 pristine (2026-07-16 18326816)
         "2c81e646c1102ebd010b6f470d6a153d8b47d68921584e55806de9052da13fa7",  # stop v2 surgered (2026-07-16 18326816)
         "fd8a7522fe488b6429f159146523524dcab6465ddbdc09aa91a3515a89bf58a2",  # stop pre-stamp (2026-06-10 ffa51bb1)

tests/test_hooks_claude_setup.py

@@ -197,10 +197,11 @@ def test_stop_hook_fails_open_and_respects_loop_guard(self, in_tmp):
 # module shim: argv ["roam", ...] is redirected to the Python stub above;
 # everything else (git ...) passes through to the real subprocess module.
 _STOP_HOOK_DRIVER_PY = """\
-import sys, types, subprocess as _real_subprocess
+import os, sys, types, subprocess as _real_subprocess
 
 _STUB = sys.argv[1]
 _SCRIPT = sys.argv[2]
+_POPEN_LOG = os.path.join(os.path.dirname(_STUB), "popen-called.txt")
 
 shim = types.ModuleType("subprocess")
 
@@ -211,7 +212,25 @@ def _run(args, **kwargs):
     return _real_subprocess.run(args, **kwargs)
 
 
+def _popen(args, **kwargs):
+    # record roam detached spawns (Loop-B report refresh) WITHOUT actually
+    # launching a background process — keeps the test deterministic.
+    if args and args[0] == "roam":
+        with open(_POPEN_LOG, "a", encoding="utf-8") as fh:
+            fh.write(" ".join(args) + chr(10))
+
+        class _Dummy:
+            pass
+
+        return _Dummy()
+    return _real_subprocess.Popen(args, **kwargs)
+
+
 shim.run = _run
+shim.Popen = _popen
+shim.DEVNULL = _real_subprocess.DEVNULL
+shim.PIPE = _real_subprocess.PIPE
+shim.TimeoutExpired = _real_subprocess.TimeoutExpired
 sys.modules["subprocess"] = shim
 with open(_SCRIPT, encoding="utf-8") as fh:
     code = fh.read()
@@ -367,6 +386,15 @@ def _register(monkeypatch, *bodies: str):
         extra = {hashlib.sha256(b.encode("utf-8")).hexdigest() for b in bodies}
         monkeypatch.setattr(cmd_hooks, "_KNOWN_HOOK_BODY_SHAS", cmd_hooks._KNOWN_HOOK_BODY_SHAS | extra)
 
+    def test_hook_bodies_compile(self):
+        """A syntax error in a body constant makes every absence-asserting hook
+        test pass vacuously (the broken hook spawns nothing) — guard the whole
+        class of failures with an explicit compile check."""
+        from roam.commands.cmd_hooks import _CLAUDE_STOP_HOOK_SCRIPT, _CLAUDE_UPS_HOOK_SCRIPT
+
+        compile(_CLAUDE_UPS_HOOK_SCRIPT, "roam-compile-ups.py", "exec")
+        compile(_CLAUDE_STOP_HOOK_SCRIPT, "roam-verify-stop.py", "exec")
+
     def test_registry_is_seeded(self):
         """F1 guard: an empty/garbled registry silently disables pre-stamp healing."""
         from roam.commands.cmd_hooks import _KNOWN_HOOK_BODY_SHAS
@@ -375,15 +403,17 @@ def test_registry_is_seeded(self):
         assert all(len(s) == 64 and set(s) <= set("0123456789abcdef") for s in _KNOWN_HOOK_BODY_SHAS)
 
     def test_heal_state_classification(self, monkeypatch):
-        from roam.commands.cmd_hooks import _CLAUDE_UPS_HOOK_SCRIPT, _hook_heal_state
+        from roam.commands.cmd_hooks import _CLAUDE_UPS_HOOK_SCRIPT, _HOOK_BODY_VERSION, _hook_heal_state
 
         canonical = _CLAUDE_UPS_HOOK_SCRIPT
+        cur = f"# roam-hook-version: {_HOOK_BODY_VERSION}"
+        old = f"# roam-hook-version: {_HOOK_BODY_VERSION - 1}"
         assert _hook_heal_state(canonical, canonical) == "current"
         # an older stamp alone proves NOTHING: unknown content -> modified
-        older_unknown = canonical.replace("# roam-hook-version: 2", "# roam-hook-version: 1") + "# my edit\n"
+        older_unknown = canonical.replace(cur, old) + "# my edit\n"
         assert _hook_heal_state(older_unknown, canonical) == "modified"
         # a REGISTERED older body (roam provably shipped it) -> healable
-        older_known = canonical.replace("# roam-hook-version: 2", "# roam-hook-version: 1")
+        older_known = canonical.replace(cur, old)
         self._register(monkeypatch, older_known)
         from roam.commands.cmd_hooks import _hook_heal_state as heal_state
 
@@ -413,8 +443,12 @@ def test_prestamp_registered_body_is_healed(self, in_tmp, monkeypatch):
     def test_older_stamped_modified_body_not_healed(self, in_tmp):
         """F3: a stamped-but-edited body is never silently overwritten."""
         hook = self._install(in_tmp)
+        from roam.commands.cmd_hooks import _HOOK_BODY_VERSION
+
         edited = (
-            hook.read_text(encoding="utf-8").replace("# roam-hook-version: 2", "# roam-hook-version: 1")
+            hook.read_text(encoding="utf-8").replace(
+                f"# roam-hook-version: {_HOOK_BODY_VERSION}", f"# roam-hook-version: {_HOOK_BODY_VERSION - 1}"
+            )
             + "# my customization\n"
         )
         hook.write_text(edited, encoding="utf-8")
@@ -554,3 +588,137 @@ def test_doctor_reports_hook_body_state(self, in_tmp, monkeypatch):
         result = _check_claude_hook_bodies()
         assert result["passed"] is False
         assert "missing" in result["detail"]
+
+
+class TestLoopBReportRefresh:
+    """C5 / T2b: the Stop hook (opt-in) spawns a DETACHED, THROTTLED whole-repo
+    `verify --report --persist` on edit-stops so the next compile's
+    known_findings is fresh — never blocking, never --diff-only into the
+    whole-repo report path."""
+
+    @staticmethod
+    def _git_repo(tmp_path):
+        import subprocess
+
+        repo = tmp_path / "repo"
+        repo.mkdir()
+        (repo / "tracked.txt").write_text("tracked\n", encoding="utf-8")
+        for args in (
+            ["git", "init", "-q"],
+            ["git", "config", "user.email", "t@t"],
+            ["git", "config", "user.name", "t"],
+            ["git", "add", "-A"],
+            ["git", "commit", "-q", "-m", "init"],
+        ):
+            subprocess.run(args, cwd=repo, check=True, capture_output=True)
+        (repo / ".roam").mkdir()
+        return repo
+
+    def _run(self, repo, stub_dir, env_extra=None, cwd=None):
+        import os
+        import subprocess
+        import sys
+
+        from roam.commands.cmd_hooks import _CLAUDE_STOP_HOOK_SCRIPT
+
+        hook = repo.parent / "stop-hook.py"
+        hook.write_text(_CLAUDE_STOP_HOOK_SCRIPT, encoding="utf-8")
+        driver = repo.parent / "stop-hook-driver.py"
+        driver.write_text(_STOP_HOOK_DRIVER_PY, encoding="utf-8")
+        env = {**os.environ, **(env_extra or {})}
+        subprocess.run(
+            [sys.executable, str(driver), str(stub_dir / "roam-stub.py"), str(hook)],
+            input="{}",
+            capture_output=True,
+            text=True,
+            timeout=60,
+            cwd=str(cwd or repo),
+            env=env,
+        )
+        log = stub_dir / "popen-called.txt"
+        return log.read_text(encoding="utf-8") if log.exists() else ""
+
+    def test_default_off_no_refresh_spawn(self, tmp_path):
+        repo = self._git_repo(tmp_path)
+        (repo / "tracked.txt").write_text("changed\n", encoding="utf-8")  # edit-stop
+        stub_dir, _ = _install_roam_verify_stub(tmp_path, _PASS_ENVELOPE)
+        popen_log = self._run(repo, stub_dir)  # no opt-in env
+        assert "verify" not in popen_log  # default OFF -> no background refresh
+
+    def test_optin_spawns_whole_repo_report_refresh(self, tmp_path):
+        repo = self._git_repo(tmp_path)
+        (repo / "tracked.txt").write_text("changed\n", encoding="utf-8")  # edit-stop
+        stub_dir, _ = _install_roam_verify_stub(tmp_path, _PASS_ENVELOPE)
+        popen_log = self._run(repo, stub_dir, {"ROAM_HOOK_REPORT_REFRESH": "1"})
+        # whole-repo (report+persist), NOT --diff-only (that would poison the report)
+        assert "roam verify --auto --report --persist" in popen_log
+        assert "--diff-only" not in popen_log
+
+    def test_throttled_when_report_fresh(self, tmp_path):
+        import time
+
+        repo = self._git_repo(tmp_path)
+        (repo / "tracked.txt").write_text("changed\n", encoding="utf-8")
+        # a FRESH report exists -> refresh must be throttled (not respawned)
+        report = repo / ".roam" / "verify-report.json"
+        report.write_text("{}", encoding="utf-8")
+        os_utime_now = time.time()
+        import os
+
+        os.utime(report, (os_utime_now, os_utime_now))
+        stub_dir, _ = _install_roam_verify_stub(tmp_path, _PASS_ENVELOPE)
+        popen_log = self._run(repo, stub_dir, {"ROAM_HOOK_REPORT_REFRESH": "1"})
+        assert "verify" not in popen_log  # fresh -> throttled
+
+    def test_no_refresh_on_clean_tree(self, tmp_path):
+        repo = self._git_repo(tmp_path)  # no edits -> fast-exit, no refresh
+        stub_dir, _ = _install_roam_verify_stub(tmp_path, _PASS_ENVELOPE)
+        popen_log = self._run(repo, stub_dir, {"ROAM_HOOK_REPORT_REFRESH": "1"})
+        assert "verify" not in popen_log
+
+    def test_claim_marker_is_single_flight(self, tmp_path):
+        """Review MAJOR-1/2: repeated edit-stops during the in-flight window (or
+        with a persist that never lands) must spawn exactly ONE verify."""
+        repo = self._git_repo(tmp_path)
+        (repo / "tracked.txt").write_text("changed\n", encoding="utf-8")
+        stub_dir, _ = _install_roam_verify_stub(tmp_path, _PASS_ENVELOPE)
+        first = self._run(repo, stub_dir, {"ROAM_HOOK_REPORT_REFRESH": "1"})
+        assert first.count("roam verify --auto --report --persist") == 1
+        assert (repo / ".roam" / "verify-refresh-claim").exists()  # claim taken
+        # second stop, report still absent (persist "never landed") -> no respawn
+        (repo / "tracked.txt").write_text("changed again\n", encoding="utf-8")
+        second = self._run(repo, stub_dir, {"ROAM_HOOK_REPORT_REFRESH": "1"})
+        assert second.count("roam verify --auto --report --persist") == 1  # log unchanged
+
+    def test_stale_claim_respawns(self, tmp_path):
+        import os as _os
+        import time as _time
+
+        repo = self._git_repo(tmp_path)
+        (repo / "tracked.txt").write_text("changed\n", encoding="utf-8")
+        (repo / ".roam").mkdir(exist_ok=True)
+        claim = repo / ".roam" / "verify-refresh-claim"
+        claim.write_text("pid=0 time=0\n", encoding="utf-8")
+        old = _time.time() - 7200  # 2h-old claim: well past the 30-min window
+        _os.utime(claim, (old, old))
+        stub_dir, _ = _install_roam_verify_stub(tmp_path, _PASS_ENVELOPE)
+        popen_log = self._run(repo, stub_dir, {"ROAM_HOOK_REPORT_REFRESH": "1"})
+        assert "roam verify --auto --report --persist" in popen_log
+
+    def test_throttle_anchored_at_repo_root(self, tmp_path):
+        """Review MAJOR-3: a hook running in a repo SUBDIR must see the root
+        report (where the spawned verify persists), not a cwd-relative path."""
+        import os as _os
+        import time as _time
+
+        repo = self._git_repo(tmp_path)
+        sub = repo / "pkg"
+        sub.mkdir()
+        (repo / "tracked.txt").write_text("changed\n", encoding="utf-8")
+        report = repo / ".roam" / "verify-report.json"
+        report.write_text("{}", encoding="utf-8")
+        now = _time.time()
+        _os.utime(report, (now, now))  # fresh AT THE ROOT
+        stub_dir, _ = _install_roam_verify_stub(tmp_path, _PASS_ENVELOPE)
+        popen_log = self._run(repo, stub_dir, {"ROAM_HOOK_REPORT_REFRESH": "1"}, cwd=sub)
+        assert "verify --auto --report --persist" not in popen_log  # root throttle honored