fix(outcome): attribute capture to the exercised skill, not brigade-work

Hard-coded brigade-work capture ids in hook replacements and
capture-before-retry messages collapsed distinct skill signals into one
rank bucket. Stamp capture intent on verify receipts, track the exercised
skill in Claude session state, and fall back to brigade-work only when
unknown.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: solomonneas

src/brigade/claude_hooks/runtime.py

@@ -1614,11 +1614,94 @@ 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 a skill id when ``raw_path`` points at ``*/skills/<id>/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 {".", ".."}:
+                return skill_id
+    return None
+
+
+def _parse_capture_flag(command: object) -> str | None:
+    """Extract ``--capture <id>`` 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 +1767,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 +1857,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 +1893,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 +1943,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 +1977,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, "<test>", fingerprint)
+                exercised = stop_state.get("exercised_artifact_id")
+                replacement = _verify_replacement(
+                    stop_target,
+                    "<test>",
+                    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

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"
 

src/brigade/work_cmd/verification.py

@@ -718,18 +718,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(
+        capture_artifact_id,
+        _receipt_capture_artifact_id(failed),
+    )
+    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
@@ -1045,6 +1080,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]}"
@@ -1070,6 +1107,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:
@@ -1203,6 +1241,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()
@@ -1225,6 +1266,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
@@ -1488,7 +1530,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:
@@ -1503,14 +1550,23 @@ 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,
                 planned,
                 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)