feat(guard): injection-phrase FP instrumentation (Refs #12)

CHANGELOG.md

@@ -31,6 +31,23 @@ CI. The v0.3 `guard` proxy adds deterministic runtime *result* inspection
 
 ### Added
 
+- **Injection-phrase FP-instrumentation (Refs #12).** Shippable, non-default-changing
+  groundwork for eventually promoting `WRD-RES-INJECT-PHRASE` to default-block once field
+  false-positive (FP) data justifies it. **No default posture changed** — the fuzzy tier
+  stays monitor-only, and `WRD-RES-INJECT-PHRASE` keeps its tier, its default action, and its
+  place in the error-replacement set. Added: (1) a discrete **`matched_phrases`** array on the
+  finding record (JSONL) + **`matchedPhrases`** SARIF property, so per-phrase aggregation reads
+  a structured field instead of parsing `message`; (2) a **`run-summary`** JSONL record + SARIF
+  run property carrying **`frames_inspected`** — the base-rate denominator for a per-phrase FP
+  rate — plus `inject_phrase_findings`; (3) **`--block-inject-phrase-only <file>`**, a
+  default-off, per-phrase opt-in that blocks ONLY the named exact phrases while every other
+  curated phrase stays monitor (narrower than `--block-inject-phrase`; the future
+  deterministic-subset promotion mechanism); (4) an operator **record → inspect → label**
+  FP-collection workflow in `docs/RESULT_INSPECTION.md` §10. **Security:** the telemetry
+  surface emits only the curated denylist phrase, rule id, metadata, action, and counts —
+  **never raw result content** (which can carry secrets/PII); all aggregation is local-only,
+  no phone-home. Issue #12 stays **open** (the default-block flip remains gated on the FP data
+  this instrumentation collects).
 - **CI coverage / lint / CVE gates.** The `CI` workflow now enforces three new
   standing gates: (1) a **coverage floor** — `pytest --cov=mcp_warden
   --cov-fail-under=80` (whole-project coverage is ~86%; the floor is pinned below

docs/RESULT_INSPECTION.md

@@ -402,3 +402,71 @@ artifact. There is no way to relax a check at runtime without a lock edit.
    kill the session).
 10. **SARIF `ruleId` == the `WRD-RES-*` id verbatim; `level` per §8.** Severity→level
     mapping matches `CHECKS.md` §2 (critical/high→error, medium→warning, low→note).
+
+---
+
+## 10. Operator false-positive collection workflow (issue #12)
+
+`WRD-RES-INJECT-PHRASE` is **monitor-only by default and stays that way** (§4, §7). Promoting
+it to default-block is gated on **field false-positive (FP) data** that does not exist yet.
+This section is the record → inspect → label loop an operator runs to *produce* that data
+**with zero blocking risk**. It changes no default.
+
+**Security invariant (non-negotiable).** The FP-collection surface emits **only** the curated
+matched phrase (from our own `SEED_INJECT_PHRASES` / org denylist), the rule id, tool/server
+metadata, the action, and counts. **Raw result content is NEVER transmitted or written** — a
+tool result can carry secrets/PII (that is exactly what `WRD-RES-SECRET-ECHO` catches). All
+aggregation is **local-only** and there is **no phone-home**.
+
+### 10.1 The loop
+
+1. **Record** a live session with the fuzzy tier in pure monitor mode (the default — do NOT
+   pass `--block-inject-phrase`):
+
+   ```
+   mcp-warden guard --record trace.jsonl --json findings.jsonl <server-cmd...>
+   ```
+
+   `--record` captures the frames; `--json` writes the finding stream. Injection-phrase hits
+   are logged and **forwarded unblocked** (`action: shadowed`).
+
+2. **Inspect** the recorded trace offline (identical catalog, no live process, no blocking):
+
+   ```
+   mcp-warden inspect trace.jsonl --json inspected.jsonl
+   ```
+
+3. **Review** the `WRD-RES-INJECT-PHRASE` records. Each finding record carries a discrete
+   **`matched_phrases`** array (the curated phrase(s) that matched — never the surrounding
+   text), so you aggregate **per phrase** without parsing prose. The final **`run-summary`**
+   record (`kind: "run-summary"`) carries **`frames_inspected`** — the base-rate denominator —
+   and `inject_phrase_findings`. FP rate for a phrase = FP-labeled hits ÷ `frames_inspected`.
+
+4. **Label** each hit TP or FP locally (a hit is an FP when the phrase appeared in benign
+   content — documentation, a quoted example, a changelog — and was not an actual injection).
+   Because `matched_phrases` is structured, a one-line `jq` group-by yields per-phrase counts:
+
+   ```
+   jq -r 'select(.kind=="result-finding" and .rule_id=="WRD-RES-INJECT-PHRASE")
+          | .matched_phrases[]' inspected.jsonl | sort | uniq -c
+   ```
+
+### 10.2 Graduated promotion (per-phrase opt-in block)
+
+Once a specific phrase proves **deterministic-enough** (FP rate ≈ 0 across enough
+`frames_inspected`), a cautious operator can block **only that phrase** while every other
+curated phrase stays monitor-only:
+
+```
+mcp-warden guard --block-inject-phrase-only phrases.txt <server-cmd...>
+```
+
+`phrases.txt` is one exact phrase per line (`#` comments allowed). This is **narrower** than
+`--block-inject-phrase` (which promotes the whole fuzzy tier) and is **default-off**. It does
+**not** change the rule's tier, its default action, or its position in the error-replacement
+set — it is a runtime, opt-in narrowing keyed on the safe `matched_phrases` field. It is also
+the intended delivery mechanism for a future "deterministic-enough subset" promotion.
+
+> **Issue #12 remains the gate.** The default-block flip for `WRD-RES-INJECT-PHRASE` stays
+> **BLOCKED** until this field FP data justifies it. The instrumentation above ships the
+> evidence-collection path; it does **not** flip any default.

src/mcp_warden/cli_guard.py

@@ -15,7 +15,12 @@
 from rich.table import Table
 
 from . import res_rules
-from .emit_res import build_result_sarif, result_findings_to_jsonl, result_sarif_to_json
+from .emit_res import (
+    build_result_sarif,
+    result_findings_to_jsonl,
+    result_sarif_to_json,
+    run_summary_to_dict,
+)
 from .guard import run_guard
 from .guard_banner import render_posture_banner
 from .guard_lifecycle import (
@@ -90,6 +95,17 @@ def guard(
         no_block_policy: bool = typer.Option(False, "--no-block-policy", help="Demote argument-policy deny to shadow"),
         no_block_deterministic: bool = typer.Option(False, "--no-block-deterministic", help="Demote the WHOLE deterministic tier + both gates"),
         block_inject_phrase: bool = typer.Option(False, "--block-inject-phrase", help="Opt-in block for WRD-RES-INJECT-PHRASE (fuzzy)"),
+        block_inject_phrase_only: Optional[Path] = typer.Option(
+            None,
+            "--block-inject-phrase-only",
+            help=(
+                "Block ONLY these exact injection phrases (one per line; '#' comments); "
+                "every other curated phrase stays monitor-only. Narrower than "
+                "--block-inject-phrase (which promotes the whole fuzzy tier). Does NOT "
+                "change the rule's tier or default. If --block-inject-phrase is also set, "
+                "block-all wins."
+            ),
+        ),
         block_ansi: bool = typer.Option(False, "--block-ansi", help="DEPRECATED no-op (now default-on)", hidden=True),
         block_secret_echo: bool = typer.Option(False, "--block-secret-echo", help="DEPRECATED no-op", hidden=True),
         block_exfil_domain: bool = typer.Option(False, "--block-exfil-domain", help="DEPRECATED no-op", hidden=True),
@@ -182,6 +198,9 @@ def guard(
             no_block_list_changed=no_block_list_changed or no_block_deterministic,
             no_block_policy=no_block_policy or no_block_deterministic,
             block_inject_phrase=block_inject_phrase,
+            block_inject_phrases_subset=frozenset(
+                res_rules.normalize_phrase_text(p) for p in _load_line_list(block_inject_phrase_only)
+            ),
             armed_list_changed=lock is not None,
             armed_policy=policy_file is not None,
             redact_secret_echo=redact_secret_echo,
@@ -216,6 +235,10 @@ def guard(
 
         findings_sink: list = []
         record_lines: list[str] = []
+        summary_box = {"frames_inspected": 0}
+
+        def _on_summary(n: int) -> None:
+            summary_box["frames_inspected"] = n
 
         def _on_finding(f) -> None:
             findings_sink.append(f)
@@ -241,14 +264,23 @@ def _record(direction: str, frame: dict) -> None:
             inject_phrases=phrases,
             on_finding=_on_finding,
             record=_record if record is not None else None,
+            on_summary=_on_summary,
         )
 
+        frames_inspected = summary_box["frames_inspected"]
+        inject_findings = sum(1 for f in findings_sink if f.rule_id == "WRD-RES-INJECT-PHRASE")
+        summary = run_summary_to_dict(
+            frames_inspected=frames_inspected, inject_phrase_findings=inject_findings
+        )
         if record is not None:
             record.write_text("\n".join(record_lines) + ("\n" if record_lines else ""), encoding="utf-8")
         if sarif is not None:
-            sarif.write_text(result_sarif_to_json(build_result_sarif(findings_sink)), encoding="utf-8")
+            sarif.write_text(
+                result_sarif_to_json(build_result_sarif(findings_sink, frames_inspected=frames_inspected)),
+                encoding="utf-8",
+            )
         if json_out is not None:
-            json_out.write_text(result_findings_to_jsonl(findings_sink), encoding="utf-8")
+            json_out.write_text(result_findings_to_jsonl(findings_sink, summary=summary), encoding="utf-8")
 
         raise typer.Exit(code=code)
 
@@ -274,16 +306,27 @@ def inspect(
         exfil = res_rules.SEED_EXFIL_DENYLIST + _load_line_list(exfil_denylist)
         phrases = res_rules.SEED_INJECT_PHRASES + _load_line_list(inject_phrases)
 
+        stats: dict = {}
         try:
-            findings = analyze_trace(trace, lock=lock_doc, exfil_denylist=exfil, inject_phrases=phrases)
+            findings = analyze_trace(
+                trace, lock=lock_doc, exfil_denylist=exfil, inject_phrases=phrases, stats=stats
+            )
         except TraceError as exc:
             err_console.print(f"[red]error:[/red] {exc}")
             raise typer.Exit(code=2) from exc
 
+        frames_inspected = stats.get("frames_inspected", 0)
+        inject_findings = sum(1 for f in findings if f.rule_id == "WRD-RES-INJECT-PHRASE")
+        summary = run_summary_to_dict(
+            frames_inspected=frames_inspected, inject_phrase_findings=inject_findings
+        )
         if sarif is not None:
-            sarif.write_text(result_sarif_to_json(build_result_sarif(findings)), encoding="utf-8")
+            sarif.write_text(
+                result_sarif_to_json(build_result_sarif(findings, frames_inspected=frames_inspected)),
+                encoding="utf-8",
+            )
         if json_out is not None:
-            json_out.write_text(result_findings_to_jsonl(findings), encoding="utf-8")
+            json_out.write_text(result_findings_to_jsonl(findings, summary=summary), encoding="utf-8")
         else:
             _print_result_findings(console, findings)
 

src/mcp_warden/emit_res.py

@@ -38,37 +38,46 @@ def _sarif_result(f: ResultFinding) -> dict[str, Any]:
             "tool": f.tool,
             "contentBlockIndex": f.block_index,
             "subRule": f.sub_rule,
+            # Curated denylist phrases only (never raw result content) — issue #12.
+            "matchedPhrases": list(f.matched_phrases),
         },
     }
 
 
-def build_result_sarif(findings: list[ResultFinding]) -> dict[str, Any]:
+def build_result_sarif(
+    findings: list[ResultFinding], *, frames_inspected: int | None = None
+) -> dict[str, Any]:
     """Build a SARIF 2.1.0 log from result-inspection findings.
 
     Args:
         findings: The stamped :class:`ResultFinding` list.
+        frames_inspected: Optional count of ``tools/call`` result frames inspected
+            this run. When supplied it is attached as a run-level
+            ``properties.framesInspected`` so a per-phrase FP rate has a base-rate
+            denominator (issue #12). It is a plain count — no result content.
 
     Returns:
         A SARIF ``dict`` ready for ``json.dumps``.
     """
     rule_ids = sorted({f.rule_id for f in findings})
     rules = [{"id": rid, "name": rid} for rid in rule_ids]
+    run: dict[str, Any] = {
+        "tool": {
+            "driver": {
+                "name": TOOL_NAME,
+                "version": __version__,
+                "informationUri": INFO_URI,
+                "rules": rules,
+            }
+        },
+        "results": [_sarif_result(f) for f in findings],
+    }
+    if frames_inspected is not None:
+        run["properties"] = {"framesInspected": frames_inspected}
     return {
         "version": SARIF_VERSION,
         "$schema": SARIF_SCHEMA,
-        "runs": [
-            {
-                "tool": {
-                    "driver": {
-                        "name": TOOL_NAME,
-                        "version": __version__,
-                        "informationUri": INFO_URI,
-                        "rules": rules,
-                    }
-                },
-                "results": [_sarif_result(f) for f in findings],
-            }
-        ],
+        "runs": [run],
     }
 
 
@@ -78,7 +87,13 @@ def result_sarif_to_json(sarif: dict[str, Any]) -> str:
 
 
 def result_finding_to_dict(f: ResultFinding) -> dict[str, Any]:
-    """JSON-serializable record for one finding (one JSONL line)."""
+    """JSON-serializable record for one finding (one JSONL line).
+
+    ``matched_phrases`` carries the discrete curated denylist phrases for a
+    ``WRD-RES-INJECT-PHRASE`` finding (empty for every other rule). It is sourced
+    from our own denylist — NEVER from raw result content — so per-phrase FP
+    aggregation reads a structured field instead of parsing ``message`` (#12).
+    """
     return {
         "kind": "result-finding",
         "rule_id": f.rule_id,
@@ -93,10 +108,48 @@ def result_finding_to_dict(f: ResultFinding) -> dict[str, Any]:
         "block_index": f.block_index,
         "message": f.message,
         "snippet": f.snippet,
+        "matched_phrases": list(f.matched_phrases),
+    }
+
+
+def run_summary_to_dict(*, frames_inspected: int, inject_phrase_findings: int = 0) -> dict[str, Any]:
+    """Build the one-line ``run-summary`` telemetry record (issue #12).
+
+    The summary gives a per-phrase FP rate its **base rate**: ``frames_inspected``
+    is the denominator (every ``tools/call`` result frame the catalog inspected
+    this run) and ``inject_phrase_findings`` is a convenience numerator (how many
+    ``WRD-RES-INJECT-PHRASE`` findings fired). It carries ONLY counts — never any
+    result content, phrase text, tool arguments, or secrets.
+
+    Args:
+        frames_inspected: Count of inspected ``tools/call`` result frames.
+        inject_phrase_findings: Count of ``WRD-RES-INJECT-PHRASE`` findings.
+
+    Returns:
+        A JSON-serializable ``run-summary`` record.
+    """
+    return {
+        "kind": "run-summary",
+        "frames_inspected": frames_inspected,
+        "inject_phrase_findings": inject_phrase_findings,
     }
 
 
-def result_findings_to_jsonl(findings: list[ResultFinding]) -> str:
-    """Serialize result findings as newline-delimited JSON (one record per line)."""
+def result_findings_to_jsonl(
+    findings: list[ResultFinding], *, summary: dict[str, Any] | None = None
+) -> str:
+    """Serialize result findings as newline-delimited JSON (one record per line).
+
+    Args:
+        findings: The stamped findings (one ``result-finding`` record per line).
+        summary: Optional ``run-summary`` record (see :func:`run_summary_to_dict`)
+            appended as a final, distinctly-``kind``ed line so a consumer reads the
+            base-rate denominator from the same stream (issue #12).
+
+    Returns:
+        Newline-delimited JSON; empty string when there is nothing to emit.
+    """
     lines = [json.dumps(result_finding_to_dict(f), ensure_ascii=False) for f in findings]
+    if summary is not None:
+        lines.append(json.dumps(summary, ensure_ascii=False))
     return "\n".join(lines) + ("\n" if lines else "")

src/mcp_warden/guard.py

@@ -386,6 +386,7 @@ def run_guard(
     inject_phrases: tuple[str, ...] | None = None,
     on_finding: Callable | None = None,
     record: Callable | None = None,
+    on_summary: Callable[[int], None] | None = None,
 ) -> int:
     """Synchronous entry point for the CLI: build state and run the loop.
 
@@ -395,6 +396,9 @@ def run_guard(
         lock/policy: Optional loaded lock + policy.
         exfil_denylist/inject_phrases: Merged seed+org lists (defaults to seed).
         on_finding/record: Optional sinks.
+        on_summary: Optional callback invoked once, after the loop finishes, with
+            the count of inspected ``tools/call`` result frames (issue #12 base-rate
+            denominator). A plain integer — never any result content.
 
     Returns:
         The child's exit code.
@@ -410,4 +414,7 @@ def run_guard(
         on_finding=on_finding,
         record=record,
     )
-    return anyio.run(run_guard_async, command, args, state)
+    code = anyio.run(run_guard_async, command, args, state)
+    if on_summary is not None:
+        on_summary(state.frames_inspected)
+    return code

src/mcp_warden/guard_banner.py

@@ -137,6 +137,14 @@ def _blocking_lines(cfg: GuardConfig) -> list[str]:
     if cfg.category_enabled("WRD-RES-INJECT-PHRASE"):
         # Opt-in only (--block-inject-phrase); when on it is a real BLOCK tier.
         active.append("  - injection-phrase block (WRD-RES-INJECT-PHRASE, opt-in)")
+    elif cfg.block_inject_phrases_subset:
+        # Per-phrase opt-in (--block-inject-phrase-only): only NAMED phrases block;
+        # all other fuzzy matches stay monitor-only. Report it so posture is honest.
+        n = len(cfg.block_inject_phrases_subset)
+        active.append(
+            f"  - injection-phrase block for {n} named phrase(s) "
+            "(WRD-RES-INJECT-PHRASE, --block-inject-phrase-only)"
+        )
     if cfg.list_changed_enabled():
         active.append("  - tools/list_changed drift gate (MCP-DRIFT, armed by --lock)")
     if cfg.policy_block_enabled():
@@ -155,11 +163,19 @@ def _monitor_lines(cfg: GuardConfig) -> list[str]:
     """The MONITOR-ONLY bucket — detect + log, never block."""
     lines = ["MONITOR-ONLY (detect + log, no block):"]
     if not cfg.category_enabled("WRD-RES-INJECT-PHRASE"):
-        # When NOT opted-in (or under audit-only) the fuzzy tier only logs.
-        lines.append(
-            "  - injection-phrase tier (WRD-RES-INJECT-PHRASE) — fuzzy; "
-            "logs matches, does NOT block (enable: --block-inject-phrase)"
-        )
+        if cfg.block_inject_phrases_subset and not cfg.audit_only:
+            # Per-phrase opt-in: the NAMED phrases block (listed above); every OTHER
+            # curated phrase in this tier still only logs.
+            lines.append(
+                "  - injection-phrase tier (WRD-RES-INJECT-PHRASE) — fuzzy; only the "
+                "named phrases block (--block-inject-phrase-only), all others log only"
+            )
+        else:
+            # When NOT opted-in (or under audit-only) the fuzzy tier only logs.
+            lines.append(
+                "  - injection-phrase tier (WRD-RES-INJECT-PHRASE) — fuzzy; "
+                "logs matches, does NOT block (enable: --block-inject-phrase)"
+            )
     lines.append(
         "  - uninspectable / non-tools/call / uncorrelated-id frames pass through "
         "uninspected (GUARD_PROXY.md §4.4)"