Merge remote-tracking branch 'origin/main' into feat/582-provenance-envelope

Author: solomonneas

src/brigade/claude_hooks/runtime.py

@@ -38,6 +38,7 @@
     "node_modules",
 }
 _SNAPSHOT_GIT_TIMEOUT_SECONDS = 3
+_UNAVAILABLE_FINGERPRINT = "unavailable"
 _BASH_WRITE_COMMANDS = {
     "apply_patch",
     "cp",
@@ -1588,8 +1589,12 @@ def _bash_write_detected(
 ) -> bool:
     if not isinstance(baseline, str) or not baseline:
         return False
+    if baseline == _UNAVAILABLE_FINGERPRINT:
+        return True
     current = repo_worktree_fingerprint(target)
-    if current is None or current == baseline:
+    if current is None:
+        return True
+    if current == baseline:
         return False
     started = localio.parse_iso_datetime(started_at)
     if started is None:
@@ -1757,10 +1762,9 @@ def handle_payload(event: str, payload: dict[str, Any]) -> dict[str, Any] | None
         tool_input: dict[str, Any] = raw_tool_input if isinstance(raw_tool_input, dict) else {}
         command = tool_input.get("command")
         baseline = repo_worktree_fingerprint(target)
-        if baseline is not None:
-            state["pending_bash_fingerprint"] = baseline
-            state["pending_bash_started_at"] = localio.utc_now_iso()
-            write_session_state(target, session_id, state)
+        state["pending_bash_fingerprint"] = baseline or _UNAVAILABLE_FINGERPRINT
+        state["pending_bash_started_at"] = localio.utc_now_iso()
+        write_session_state(target, session_id, state)
         if not is_raw_verification(command):
             return None
         state["verify_denied_count"] = int(state.get("verify_denied_count") or 0) + 1

src/brigade/model_inventory.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import re
+import subprocess
+import tempfile
 from dataclasses import dataclass
 from typing import Literal
 
@@ -15,6 +17,8 @@
     "grok": ["grok", "models"],
     "ollama": ["ollama", "list"],
 }
+_INVENTORY_TIMEOUT_SECONDS = 15.0
+_CURSOR_COMPLETION_PREFIX = "Tip: use --model <id>"
 _EFFORT_SUFFIX = re.compile(r"-(?:none|low|medium|high|xhigh|extra-high|max)$")
 _OLLAMA_OPERATIONAL_ERRORS = (
     "authentication",
@@ -107,12 +111,17 @@ def _inventory(self, cli_ref: str) -> _HarnessInventory:
         cached = self._inventories.get(cli_ref)
         if cached is not None:
             return cached
-        result = proc.run(_LIST_COMMANDS[cli_ref], timeout=15.0)
+        result = (
+            _run_cursor_inventory()
+            if cli_ref == "cursor"
+            else proc.run(_LIST_COMMANDS[cli_ref], timeout=_INVENTORY_TIMEOUT_SECONDS)
+        )
         if result.code != 0:
             diagnostic = result.stderr.strip() or result.stdout.strip() or f"exit {result.code}"
             inventory = _HarnessInventory(error=diagnostic[:160])
         else:
-            recognized, models = _parse_model_list(cli_ref, f"{result.stdout}\n{result.stderr}")
+            output = result.stdout if cli_ref == "cursor" else f"{result.stdout}\n{result.stderr}"
+            recognized, models = _parse_model_list(cli_ref, output)
             if not recognized:
                 inventory = _HarnessInventory(error="command returned an unrecognized inventory shape")
             elif not models and cli_ref != "ollama":
@@ -161,17 +170,61 @@ def _inspect_ollama(self, requested: str) -> ModelInventoryResult:
         return ModelInventoryResult(state, requested, (), diagnostic[:200])
 
 
+def _run_cursor_inventory() -> proc.Result:
+    """Capture Cursor inventory through a file because its pipe output truncates at 8 KiB."""
+    with tempfile.TemporaryFile() as stdout:
+        try:
+            completed = subprocess.run(
+                _LIST_COMMANDS["cursor"],
+                stdout=stdout,
+                stderr=subprocess.PIPE,
+                stdin=subprocess.DEVNULL,
+                timeout=_INVENTORY_TIMEOUT_SECONDS,
+                check=False,
+            )
+        except subprocess.TimeoutExpired as exc:
+            stdout.seek(0)
+            output = stdout.read().decode("utf-8", errors="replace")
+            diagnostic = _decode_subprocess_output(exc.stderr)
+            if diagnostic and not diagnostic.endswith("\n"):
+                diagnostic += "\n"
+            return proc.Result(
+                124,
+                output,
+                f"{diagnostic}timeout after {_INVENTORY_TIMEOUT_SECONDS}s",
+            )
+        except OSError as exc:
+            return proc.Result(127 if isinstance(exc, FileNotFoundError) else 126, "", str(exc))
+        stdout.seek(0)
+        output = stdout.read().decode("utf-8", errors="replace")
+    return proc.Result(completed.returncode, output, _decode_subprocess_output(completed.stderr))
+
+
+def _decode_subprocess_output(value: str | bytes | None) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, str):
+        return value
+    return value.decode("utf-8", errors="replace")
+
+
 def _parse_model_list(cli_ref: str, output: str) -> tuple[bool, tuple[str, ...]]:
     lines = output.splitlines()
     header_index = _inventory_header_index(cli_ref, lines)
     if header_index is None:
         return False, ()
     models: set[str] = set()
+    complete = cli_ref != "cursor"
     for line in lines[header_index + 1 :]:
         if cli_ref == "cursor":
-            if line.startswith("Tip:"):
+            if not line.strip():
+                continue
+            if line.startswith(_CURSOR_COMPLETION_PREFIX):
+                complete = True
                 break
             match = re.match(r"^([a-z0-9][a-z0-9._:/\[\],=-]*)\s+-\s+.+$", line)
+            if match is None:
+                return False, ()
         elif cli_ref == "grok":
             match = re.match(r"^\s*\*\s+([^\s(]+)", line)
         else:
@@ -184,7 +237,7 @@ def _parse_model_list(cli_ref: str, output: str) -> tuple[bool, tuple[str, ...]]
             continue
         if match is not None:
             models.add(match.group(1))
-    return True, tuple(sorted(models))
+    return complete, tuple(sorted(models))
 
 
 def _inventory_header_index(cli_ref: str, lines: list[str]) -> int | None:

tests/test_claude_hooks_runtime.py

@@ -412,7 +412,7 @@ def test_posttooluse_ignores_concurrent_session_write_during_read_only_bash(tmp_
     )
 
 
-def test_posttooluse_snapshot_fails_open_when_state_cannot_be_inspected(tmp_path: Path, monkeypatch):
+def test_posttooluse_snapshot_fails_closed_when_state_cannot_be_inspected(tmp_path: Path, monkeypatch):
     target = _wired_claude(tmp_path)
     session_id = "snapshot-unavailable"
     monkeypatch.setattr(runtime, "repo_worktree_fingerprint", lambda repo: None)
@@ -424,11 +424,61 @@ def test_posttooluse_snapshot_fails_open_when_state_cannot_be_inspected(tmp_path
         tool_input={"command": f"{sys.executable} -c \"print('noop')\""},
     )
     assert runtime.handle_payload("PreToolUse", pretool) is None
-    assert runtime.read_session_state(target, session_id).get("pending_bash_fingerprint") is None
+    assert runtime.read_session_state(target, session_id).get("pending_bash_fingerprint") == "unavailable"
 
     succeeded = {**pretool, "hook_event_name": "PostToolUse"}
     assert runtime.handle_payload("PostToolUse", succeeded) is None
-    assert runtime.read_session_state(target, session_id)["write_observed"] is False
+    assert runtime.read_session_state(target, session_id)["write_observed"] is True
+    blocked = runtime.handle_payload("Stop", _payload(target, "Stop", session_id=session_id, stop_hook_active=False))
+    assert blocked["decision"] == "block"
+
+
+def test_posttooluse_fails_closed_when_post_command_snapshot_is_unavailable(tmp_path: Path, monkeypatch):
+    target = _wired_claude(tmp_path)
+    session_id = "post-snapshot-unavailable"
+    fingerprint_calls = 0
+
+    def sequenced_fingerprint(repo: Path) -> str | None:
+        nonlocal fingerprint_calls
+        fingerprint_calls += 1
+        return "baseline" if fingerprint_calls < 4 else None
+
+    monkeypatch.setattr(runtime, "repo_worktree_fingerprint", sequenced_fingerprint)
+    pretool = _payload(
+        target,
+        "PreToolUse",
+        session_id=session_id,
+        tool_name="Bash",
+        tool_input={"command": f"{sys.executable} -c \"print('noop')\""},
+    )
+
+    assert runtime.handle_payload("PreToolUse", pretool) is None
+    assert runtime.read_session_state(target, session_id)["pending_bash_fingerprint"] == "baseline"
+    assert runtime.handle_payload("PostToolUse", {**pretool, "hook_event_name": "PostToolUse"}) is None
+
+    assert runtime.read_session_state(target, session_id)["write_observed"] is True
+    blocked = runtime.handle_payload("Stop", _payload(target, "Stop", session_id=session_id, stop_hook_active=False))
+    assert blocked["decision"] == "block"
+
+
+@pytest.mark.parametrize("command", ["gh --version", "jq --version", "rg --version"])
+def test_posttooluse_unlisted_read_only_command_does_not_observe_write(tmp_path: Path, command: str):
+    target = _git_wired_claude(tmp_path)
+    session_id = f"read-only-{command.split()[0]}"
+    pretool = _payload(
+        target,
+        "PreToolUse",
+        session_id=session_id,
+        tool_name="Bash",
+        tool_input={"command": command},
+    )
+
+    assert runtime.handle_payload("PreToolUse", pretool) is None
+    assert runtime.handle_payload("PostToolUse", {**pretool, "hook_event_name": "PostToolUse"}) is None
+
+    state = runtime.read_session_state(target, session_id)
+    assert state["write_observed"] is False
+    assert "pending_bash_fingerprint" not in state
     assert (
         runtime.handle_payload("Stop", _payload(target, "Stop", session_id=session_id, stop_hook_active=False)) is None
     )
@@ -840,7 +890,7 @@ def fake_run(repo: Path, *git_args: str):
     assert runtime.repo_worktree_fingerprint(target) is None
 
 
-def test_posttooluse_does_not_record_bash_write_when_hash_object_fails_for_untracked(tmp_path: Path, monkeypatch):
+def test_posttooluse_fails_closed_when_untracked_state_check_fails(tmp_path: Path, monkeypatch):
     target = _git_wired_claude(tmp_path)
     session_id = "hash-object-fail"
     out_file = target / "new.txt"
@@ -864,15 +914,14 @@ def fake_run(repo: Path, *git_args: str):
     assert runtime.handle_payload("PreToolUse", pretool) is None
     state = runtime.read_session_state(target, session_id)
     assert state["write_observed"] is False
-    assert "pending_bash_fingerprint" not in state
+    assert state["pending_bash_fingerprint"] == "unavailable"
 
     out_file.write_text("after")
     succeeded = {**pretool, "hook_event_name": "PostToolUse"}
     assert runtime.handle_payload("PostToolUse", succeeded) is None
-    assert runtime.read_session_state(target, session_id)["write_observed"] is False
-    assert (
-        runtime.handle_payload("Stop", _payload(target, "Stop", session_id=session_id, stop_hook_active=False)) is None
-    )
+    assert runtime.read_session_state(target, session_id)["write_observed"] is True
+    blocked = runtime.handle_payload("Stop", _payload(target, "Stop", session_id=session_id, stop_hook_active=False))
+    assert blocked["decision"] == "block"
 
 
 def test_posttooluse_records_bash_write_on_dirty_tracked_same_size_rewrite(tmp_path: Path):