diff --git a/src/brigade/claude_hooks/runtime.py b/src/brigade/claude_hooks/runtime.py index 48762a95..77a730c5 100644 --- a/src/brigade/claude_hooks/runtime.py +++ b/src/brigade/claude_hooks/runtime.py @@ -1614,11 +1614,97 @@ def _session_fingerprint(session_id: str) -> str: return localio.stable_hash({"claude_session_id": session_id}) -def _verify_replacement(target: Path, command: str, session_fingerprint: str) -> str: +def _skill_id_from_skill_path(target: Path, raw_path: object) -> str | None: + """Return an installed target skill id when ``raw_path`` names its ``SKILL.md``.""" + if not isinstance(raw_path, str) or not raw_path: + return None + path = Path(raw_path).expanduser() + if not path.is_absolute(): + path = target / path + parts = path.parts + for index, part in enumerate(parts): + if part == "skills" and index + 2 < len(parts) and parts[index + 2] == "SKILL.md": + skill_id = parts[index + 1] + if skill_id and skill_id not in {".", ".."}: + from .. import outcome_cmd + + if outcome_cmd._artifact_known(target, skill_id, "skill"): + return skill_id + return None + + +def _parse_capture_flag(command: object) -> str | None: + """Extract ``--capture `` from a verify command string when present.""" + if not isinstance(command, str) or not command.strip(): + return None + try: + tokens = shlex.split(command, posix=os.name != "nt") + except ValueError: + return None + for index, token in enumerate(tokens): + if token == "--capture" and index + 1 < len(tokens): + value = tokens[index + 1] + if value and not value.startswith("-"): + return value + if token.startswith("--capture="): + value = token.split("=", 1)[1] + if value: + return value + return None + + +def _record_exercised_artifact(state: dict[str, Any], artifact_id: str | None, *, kind: str = "skill") -> bool: + """Persist the most specific non-generic exercised artifact on session state.""" + from .. import outcome_cmd + + if not isinstance(artifact_id, str): + return False + trimmed = artifact_id.strip() + if not trimmed: + return False + current = state.get("exercised_artifact_id") + if ( + trimmed == outcome_cmd.DEFAULT_CAPTURE_ARTIFACT_ID + and isinstance(current, str) + and current.strip() + and current.strip() != outcome_cmd.DEFAULT_CAPTURE_ARTIFACT_ID + ): + return False + if current == trimmed and state.get("exercised_artifact_kind") == kind: + return False + state["exercised_artifact_id"] = trimmed + state["exercised_artifact_kind"] = kind + return True + + +def exercised_artifact_for_fingerprint(target: Path, session_fingerprint: str) -> str | None: + """Look up an exercised artifact id for a Claude session fingerprint.""" + for state in iter_session_states(target, limit=MAX_RECENT_SESSION_STATES): + if state.get("session_fingerprint") != session_fingerprint: + continue + artifact_id = state.get("exercised_artifact_id") + if isinstance(artifact_id, str) and artifact_id.strip(): + return artifact_id.strip() + return None + + +def _verify_replacement( + target: Path, + command: str, + session_fingerprint: str, + *, + capture_artifact_id: str | None = None, +) -> str: + from .. import outcome_cmd + + artifact_id = outcome_cmd.resolve_capture_artifact_id( + capture_artifact_id, + exercised_artifact_for_fingerprint(target, session_fingerprint), + ) return ( f"{CLAUDE_SESSION_ENV}={shlex.quote(session_fingerprint)} " f"brigade work verify run --target {shlex.quote(str(target))} " - f"--command {shlex.quote(command)} --capture brigade-work" + f"--command {shlex.quote(command)} --capture {shlex.quote(artifact_id)}" ) @@ -1684,6 +1770,11 @@ def _normalize_state(target: Path, session_id: str, payload: dict[str, Any] | No session_repos = payload.get("session_repos") if isinstance(session_repos, list) and all(isinstance(item, str) for item in session_repos): normalized["session_repos"] = list(session_repos) + exercised = payload.get("exercised_artifact_id") + if isinstance(exercised, str) and exercised.strip(): + normalized["exercised_artifact_id"] = exercised.strip() + kind = payload.get("exercised_artifact_kind") + normalized["exercised_artifact_kind"] = kind if isinstance(kind, str) and kind.strip() else "skill" return normalized @@ -1769,17 +1860,24 @@ def handle_payload(event: str, payload: dict[str, Any]) -> dict[str, Any] | None return None state["verify_denied_count"] = int(state.get("verify_denied_count") or 0) + 1 write_session_state(target, session_id, state) + capture_artifact_id = state.get("exercised_artifact_id") + if not isinstance(capture_artifact_id, str): + capture_artifact_id = None if _has_unsupported_verifier_structure(str(command)): + from .. import outcome_cmd + + capture_id = outcome_cmd.resolve_capture_artifact_id(capture_artifact_id) reason = ( "Route verification through Brigade so failed, rejected, and passing results create receipts.\n" "Split shell grouping, command substitution, pipelines, redirection, or complex directory changes " - "from the verifier, then run that verifier with `brigade work verify run --capture brigade-work`." + f"from the verifier, then run that verifier with `brigade work verify run --capture {capture_id}`." ) else: replacement = _verify_replacement( target, _first_verifier_command(str(command)), str(state["session_fingerprint"]), + capture_artifact_id=capture_artifact_id, ) reason = ( "Route verification through Brigade so failed, rejected, and passing results create receipts.\n" @@ -1798,7 +1896,14 @@ def handle_payload(event: str, payload: dict[str, Any]) -> dict[str, Any] | None raw_post_tool_input = payload.get("tool_input") post_tool_input: dict[str, Any] = raw_post_tool_input if isinstance(raw_post_tool_input, dict) else {} command = post_tool_input.get("command") + if tool_name == "Read": + skill_id = _skill_id_from_skill_path(target, post_tool_input.get("file_path")) + if _record_exercised_artifact(state, skill_id): + write_session_state(target, session_id, state) + return None if tool_name == "Bash" and (_is_routed_verify(command) or _is_brigade_run(command)): + if _is_routed_verify(command): + _record_exercised_artifact(state, _parse_capture_flag(command)) state.pop("pending_bash_fingerprint", None) state.pop("pending_bash_started_at", None) write_session_state(target, session_id, state) @@ -1841,9 +1946,16 @@ def handle_payload(event: str, payload: dict[str, Any]) -> dict[str, Any] | None tool_input = raw_tool_input if isinstance(raw_tool_input, dict) else {} command = tool_input.get("command") if payload.get("tool_name") == "Bash" and (is_raw_verification(command) or _is_routed_verify(command)): + from .. import outcome_cmd + + capture_id = outcome_cmd.resolve_capture_artifact_id( + _parse_capture_flag(command), + state.get("exercised_artifact_id") if isinstance(state.get("exercised_artifact_id"), str) else None, + ) return _additional_context( "PostToolUseFailure", - "The failed or rejected verification must remain recorded in Brigade before retrying. Inspect the receipt, fix the cause, then rerun through `brigade work verify run --capture brigade-work`.", + "The failed or rejected verification must remain recorded in Brigade before retrying. Inspect the receipt, fix the cause, then rerun through " + f"`brigade work verify run --capture {capture_id}`.", ) return None @@ -1868,7 +1980,13 @@ def handle_payload(event: str, payload: dict[str, Any]) -> dict[str, Any] | None or stop_state.get("started_at") ) if not _receipt_since(stop_target, receipt_threshold, session_fingerprint=fingerprint): - replacement = _verify_replacement(stop_target, "", fingerprint) + exercised = stop_state.get("exercised_artifact_id") + replacement = _verify_replacement( + stop_target, + "", + fingerprint, + capture_artifact_id=exercised if isinstance(exercised, str) else None, + ) blocking_failures.append(f"{stop_target}: run `{replacement}`") elif not _handoff_since(stop_target, stop_state.get("started_at")): handoff_target = stop_target diff --git a/src/brigade/outcome_cmd.py b/src/brigade/outcome_cmd.py index e6d9f2ee..46f1d5d0 100644 --- a/src/brigade/outcome_cmd.py +++ b/src/brigade/outcome_cmd.py @@ -26,11 +26,25 @@ _LOCK_WAIT_SECONDS = 30.0 _LOCK_SENTINEL_BYTE = b"\0" +# Generic fallback when no exercised skill/card is known. Prefer a real artifact +# id at capture time so distinct skills do not collapse into one rank bucket. +DEFAULT_CAPTURE_ARTIFACT_ID = "brigade-work" + class OutcomeLedgerError(RuntimeError): """Outcome ledger persistence failed; callers must not assume the append succeeded.""" +def resolve_capture_artifact_id(*candidates: str | None) -> str: + """Return the first non-empty capture id, else the generic brigade-work fallback.""" + for candidate in candidates: + if isinstance(candidate, str): + trimmed = candidate.strip() + if trimmed: + return trimmed + return DEFAULT_CAPTURE_ARTIFACT_ID + + def _records_lock_path(target: Path) -> Path: return _records_path(target).parent / ".records.lock" diff --git a/src/brigade/work_cmd/verification.py b/src/brigade/work_cmd/verification.py index 001ffe3e..8059ad5e 100644 --- a/src/brigade/work_cmd/verification.py +++ b/src/brigade/work_cmd/verification.py @@ -789,18 +789,53 @@ def _find_uncaptured_failed_verify_receipt(target: Path, planned_identity: list[ return None -def _capture_before_retry_message(run_id: str) -> str: - return f"brigade outcome capture brigade-work --run-id {run_id}" +def _receipt_capture_artifact_id(receipt: dict[str, Any] | None) -> str | None: + """Return the artifact id stamped on a verify receipt for outcome capture, if any.""" + if not isinstance(receipt, dict): + return None + capture = receipt.get("outcome_capture") + if not isinstance(capture, dict): + return None + artifact_id = capture.get("artifact_id") + if isinstance(artifact_id, str) and artifact_id.strip(): + return artifact_id.strip() + return None -def _enforce_capture_before_retry(target: Path, planned_identity: list[str], *, mode: str) -> int | None: +def _stamp_outcome_capture(receipt: dict[str, Any], capture: str | None, capture_kind: str) -> None: + """Record the intended outcome artifact on the receipt before digests are sealed.""" + if not capture: + return + receipt["outcome_capture"] = { + "artifact_id": capture, + "artifact_kind": capture_kind, + } + + +def _capture_before_retry_message(run_id: str, *, artifact_id: str) -> str: + return f"brigade outcome capture {artifact_id} --run-id {run_id}" + + +def _enforce_capture_before_retry( + target: Path, + planned_identity: list[str], + *, + mode: str, + capture_artifact_id: str | None = None, +) -> int | None: """Block or warn when the latest matching failed receipt has no outcome capture.""" if mode == "off": return None failed = _find_uncaptured_failed_verify_receipt(target, planned_identity) if failed is None: return None - message = _capture_before_retry_message(str(failed.get("run_id") or "")) + from .. import outcome_cmd + + artifact_id = outcome_cmd.resolve_capture_artifact_id( + _receipt_capture_artifact_id(failed), + capture_artifact_id, + ) + message = _capture_before_retry_message(str(failed.get("run_id") or ""), artifact_id=artifact_id) if mode == "block": print(f"error: {message}", file=sys.stderr) return 1 @@ -1116,6 +1151,8 @@ def _run_verify_commands( *, graphtrail_timeout: float, manifest: verify_manifest.VerifyManifest | None = None, + capture: str | None = None, + capture_kind: str = "skill", ) -> tuple[dict[str, Any], int]: started = helpers._now() run_id = f"{started.strftime('%Y%m%d-%H%M%S')}-work-verify-{uuid4().hex[:6]}" @@ -1141,6 +1178,7 @@ def _run_verify_commands( } receipt.update(identity) _stamp_harness_session(receipt) + _stamp_outcome_capture(receipt, capture, capture_kind) try: graph_delta_before = graphtrail_delta.capture_before(target, run_dir, timeout=graphtrail_timeout) except KeyboardInterrupt: @@ -1274,6 +1312,9 @@ def _write_reused_receipt( latest: dict[str, Any], planned_display: list[str], timeout: int, + *, + capture: str | None = None, + capture_kind: str = "skill", ) -> tuple[dict[str, Any], int]: """Write a fresh receipt dir that records a reused passing run (no commands executed).""" started = helpers._now() @@ -1296,6 +1337,7 @@ def _write_reused_receipt( } receipt.update(identity) _stamp_harness_session(receipt) + _stamp_outcome_capture(receipt, capture, capture_kind) reused_from = latest.get("run_id") if isinstance(reused_from, str) and reused_from: receipt["reused_from"] = reused_from @@ -1559,7 +1601,12 @@ def verify_run( except ValueError as exc: print(f"error: {exc}", file=sys.stderr) return 2 - blocked_rc = _enforce_capture_before_retry(target, planned_identity, mode=capture_before_retry) + blocked_rc = _enforce_capture_before_retry( + target, + planned_identity, + mode=capture_before_retry, + capture_artifact_id=capture, + ) if blocked_rc is not None: return blocked_rc try: @@ -1574,7 +1621,14 @@ def verify_run( and latest.get("tree_fingerprint") == fingerprint and latest.get("planned_commands") == planned_display ): - receipt, rc = _write_reused_receipt(target, latest, planned_display, timeout) + receipt, rc = _write_reused_receipt( + target, + latest, + planned_display, + timeout, + capture=capture, + capture_kind=capture_kind, + ) if receipt is None: receipt, rc = _run_verify_commands( target, @@ -1582,6 +1636,8 @@ def verify_run( timeout, graphtrail_timeout=effective_graphtrail_timeout, manifest=manifest, + capture=capture, + capture_kind=capture_kind, ) except KeyboardInterrupt: print("error: verification canceled by user", file=sys.stderr) diff --git a/tests/test_claude_hooks_runtime.py b/tests/test_claude_hooks_runtime.py index c983a202..149d3c85 100644 --- a/tests/test_claude_hooks_runtime.py +++ b/tests/test_claude_hooks_runtime.py @@ -1796,3 +1796,88 @@ def test_heredoc_scanner_is_quote_aware_and_queues_delimiters(tmp_path: Path): stripped_bs = runtime._strip_heredoc_bodies(backslash) assert stripped_bs == "cat <<\\END\necho done" assert "pytest" not in stripped_bs + + +def test_posttooluse_read_skill_sets_exercised_artifact(tmp_path: Path): + target = _wired_claude(tmp_path) + skill = target / ".claude" / "skills" / "taste" / "SKILL.md" + skill.parent.mkdir(parents=True, exist_ok=True) + skill.write_text("# taste\n") + runtime.handle_payload( + "PostToolUse", + _payload( + target, + "PostToolUse", + tool_name="Read", + tool_input={"file_path": str(skill)}, + ), + ) + state = runtime.read_session_state(target, "session-1") + assert state is not None + assert state["exercised_artifact_id"] == "taste" + + +def test_posttooluse_read_external_skill_does_not_set_exercised_artifact(tmp_path: Path): + target = _wired_claude(tmp_path) + reference_skill = tmp_path / "other-repo" / ".claude" / "skills" / "reference-only" / "SKILL.md" + reference_skill.parent.mkdir(parents=True) + reference_skill.write_text("# reference-only\n") + + runtime.handle_payload( + "PostToolUse", + _payload( + target, + "PostToolUse", + tool_name="Read", + tool_input={"file_path": str(reference_skill)}, + ), + ) + + state = runtime.read_session_state(target, "session-1") + assert state is not None + assert "exercised_artifact_id" not in state + + +def test_verify_replacement_uses_session_exercised_artifact(tmp_path: Path): + target = _wired_claude(tmp_path) + skill = target / ".claude" / "skills" / "taste" / "SKILL.md" + skill.parent.mkdir(parents=True, exist_ok=True) + skill.write_text("# taste\n") + runtime.handle_payload( + "PostToolUse", + _payload( + target, + "PostToolUse", + tool_name="Read", + tool_input={"file_path": str(skill)}, + ), + ) + result = runtime.handle_payload( + "PreToolUse", + _payload(target, "PreToolUse", tool_name="Bash", tool_input={"command": "python -m pytest -q"}), + ) + reason = result["hookSpecificOutput"]["permissionDecisionReason"] + assert "--capture taste" in reason + assert "--capture brigade-work" not in reason + + +def test_verify_replacement_falls_back_to_brigade_work(tmp_path: Path): + target = _wired_claude(tmp_path) + result = runtime.handle_payload( + "PreToolUse", + _payload(target, "PreToolUse", tool_name="Bash", tool_input={"command": "python -m pytest -q"}), + ) + reason = result["hookSpecificOutput"]["permissionDecisionReason"] + assert "--capture brigade-work" in reason + + +def test_posttooluse_routed_verify_records_capture_artifact(tmp_path: Path): + target = _wired_claude(tmp_path) + command = 'brigade work verify run --target . --command "true" --capture ultra-work-scout' + runtime.handle_payload( + "PostToolUse", + _payload(target, "PostToolUse", tool_name="Bash", tool_input={"command": command}), + ) + state = runtime.read_session_state(target, "session-1") + assert state is not None + assert state["exercised_artifact_id"] == "ultra-work-scout" diff --git a/tests/test_work_cmd_verification.py b/tests/test_work_cmd_verification.py index 4d2433a7..1d455c0d 100644 --- a/tests/test_work_cmd_verification.py +++ b/tests/test_work_cmd_verification.py @@ -2701,3 +2701,102 @@ def test_archive_verify_run_strips_recovery_checkpoint_bodies(tmp_path): assert '"task"' not in archived_cp.read_text(encoding="utf-8") # Source verify-run is deleted after archive; the privacy rule is on the export. assert not (root / "20260101-000001-a").exists() + + +def test_capture_before_retry_uses_receipt_stamped_artifact(tmp_target, monkeypatch, capsys): + from brigade.work_cmd import verification + + _init_verify_target_with_head(tmp_target) + monkeypatch.setenv("GRAPHTRAIL_BIN", str(tmp_target / "missing-graphtrail")) + skill_dir = tmp_target / ".claude" / "skills" / "taste" + skill_dir.mkdir(parents=True) + (skill_dir / "SKILL.md").write_text("# taste\n") + + verification.verify_run(target=tmp_target, commands=["false"], timeout=60, capture="taste") + failed = verification._verify_receipts(tmp_target)[0] + assert failed["outcome_capture"]["artifact_id"] == "taste" + # Drop the auto-captured ledger row so retry enforcement still fires, while + # the receipt keeps the stamped capture intent. + records = tmp_target / "memory" / "outcome" / "records.jsonl" + if records.is_file(): + records.write_text("") + capsys.readouterr() + rc = verification.verify_run(target=tmp_target, commands=["false"], timeout=60) + assert rc != 0 + err = capsys.readouterr().err + assert f"warning: brigade outcome capture taste --run-id {failed['run_id']}" in err + assert "capture brigade-work" not in err + + +def test_capture_before_retry_prefers_receipt_stamp_over_current_capture(tmp_target, monkeypatch, capsys): + from brigade.work_cmd import verification + + _init_verify_target_with_head(tmp_target) + monkeypatch.setenv("GRAPHTRAIL_BIN", str(tmp_target / "missing-graphtrail")) + skill_dir = tmp_target / ".claude" / "skills" / "taste" + skill_dir.mkdir(parents=True) + (skill_dir / "SKILL.md").write_text("# taste\n") + + verification.verify_run(target=tmp_target, commands=["false"], timeout=60, capture="taste") + failed = verification._verify_receipts(tmp_target)[0] + records = tmp_target / "memory" / "outcome" / "records.jsonl" + if records.is_file(): + records.write_text("") + capsys.readouterr() + + rc = verification.verify_run(target=tmp_target, commands=["false"], timeout=60, capture="refire") + + assert rc != 0 + err = capsys.readouterr().err + assert f"warning: brigade outcome capture taste --run-id {failed['run_id']}" in err + assert f"warning: brigade outcome capture refire --run-id {failed['run_id']}" not in err + + +def test_capture_before_retry_uses_current_capture_when_receipt_unstamped(tmp_target, monkeypatch, capsys): + from brigade.work_cmd import verification + + _init_verify_target_with_head(tmp_target) + monkeypatch.setenv("GRAPHTRAIL_BIN", str(tmp_target / "missing-graphtrail")) + + verification.verify_run(target=tmp_target, commands=["false"], timeout=60) + failed = verification._verify_receipts(tmp_target)[0] + rc = verification.verify_run(target=tmp_target, commands=["false"], timeout=60, capture="refire") + assert rc != 0 + err = capsys.readouterr().err + assert f"warning: brigade outcome capture refire --run-id {failed['run_id']}" in err + + +def test_capture_before_retry_falls_back_to_brigade_work(tmp_target, monkeypatch, capsys): + from brigade.work_cmd import verification + + _init_verify_target_with_head(tmp_target) + monkeypatch.setenv("GRAPHTRAIL_BIN", str(tmp_target / "missing-graphtrail")) + + verification.verify_run(target=tmp_target, commands=["false"], timeout=60) + failed = verification._verify_receipts(tmp_target)[0] + verification.verify_run(target=tmp_target, commands=["false"], timeout=60) + err = capsys.readouterr().err + assert f"warning: brigade outcome capture brigade-work --run-id {failed['run_id']}" in err + + +def test_distinct_skill_captures_accumulate_separate_rank_signals(tmp_target, monkeypatch, capsys): + from brigade import outcome_cmd + from brigade.work_cmd import verification + + _init_verify_target_with_head(tmp_target) + monkeypatch.setenv("GRAPHTRAIL_BIN", str(tmp_target / "missing-graphtrail")) + for skill in ("taste", "refire"): + skill_dir = tmp_target / ".claude" / "skills" / skill + skill_dir.mkdir(parents=True) + (skill_dir / "SKILL.md").write_text(f"# {skill}\n") + + assert verification.verify_run(target=tmp_target, commands=["true"], timeout=60, capture="taste") == 0 + (tmp_target / "touch.txt").write_text("change\n") + assert verification.verify_run(target=tmp_target, commands=["true"], timeout=60, capture="refire", reuse=False) == 0 + capsys.readouterr() + assert outcome_cmd.rank(target=tmp_target, json_output=True) == 0 + payload = json.loads(capsys.readouterr().out) + by_id = {entry["artifact_id"]: entry for entry in payload["ranking"]} + assert by_id["taste"]["helped"] == 1 + assert by_id["refire"]["helped"] == 1 + assert "brigade-work" not in by_id or by_id.get("brigade-work", {}).get("helped", 0) == 0