feat(kora): KR-MCP-AUDIT-ON-DENIAL — emit audit at cap-gate denial

kora_cli/alerts/aggregator.py

@@ -50,17 +50,17 @@
     ``heartbeat_probes/types.py:28``: {healthy, degraded,
     unhealthy, unknown}.
 
-# Forward-compat note: capability_denied
-
-The ``capability_denied_24h`` rule is forward-looking: today the
-``mcp.tool_called`` audit emit at
-``kora_cli/listeners/mcp_tools.py:714`` is reached AFTER the
-capability gate (``listeners/mcp.py:181``), so denial responses
-are NOT currently audit-logged. This rule's matcher uses
-``details.result == "capability_denied"`` so when a follow-on
-bucket adds audit-on-denial the rule activates automatically; in
-the meantime it emits zero alerts (no false negatives — the data
-genuinely isn't there).
+# capability_denied rule — live as of KR-MCP-AUDIT-ON-DENIAL
+
+The ``capability_denied_24h`` rule consumes JSONL audit rows that
+``listeners/mcp.py:_emit_capability_denied_audit`` writes BEFORE
+returning the -32001 envelope at the cap gate. Discriminator is
+``details.result == "capability_denied"``. The companion path —
+``actor_id_required_for_stop`` from the ``kora__request_stop``
+tool's extra gate — writes
+``details.result == "actor_id_required"`` so it's queryable
+separately (future ``actor_id_required_24h`` rule can grep it
+without conflating two distinct operator-fix surfaces).
 """
 
 from __future__ import annotations
@@ -338,14 +338,16 @@ def _rules_from_webhook_audit() -> List[Alert]:
 
 
 def _rules_from_capability_denied_audit() -> List[Alert]:
-    """Forward-looking — see module docstring's forward-compat note.
-
-    Today the audit log at ``mcp.tool_called`` only records successful
-    invocations (capability gate at ``listeners/mcp.py:181`` returns
-    BEFORE the audit emit at ``listeners/mcp_tools.py:714``). This
-    rule's matcher is forward-compatible: when a follow-on bucket
-    adds audit-on-denial it'll start firing without an aggregator
-    edit.
+    """Surface a warning when too many cap-gate denials in 24h.
+
+    KR-MCP-AUDIT-ON-DENIAL wired the cap-gate at
+    ``listeners/mcp.py`` to call
+    ``_emit_capability_denied_audit`` BEFORE returning the
+    -32001 envelope, so denial responses now land in the audit
+    JSONL with ``details.result == "capability_denied"``. This
+    rule fires when more than
+    :data:`CAPABILITY_DENIED_24H_THRESHOLD` denials accumulate
+    in the trailing 24h.
     """
     since = datetime.now(timezone.utc) - timedelta(hours=24)
     count = _count_audit_entries_in_window(

kora_cli/listeners/mcp.py

@@ -203,6 +203,119 @@ def _check_cap_gate(
     }
 
 
+# ---------------------------------------------------------------------------
+# KR-MCP-AUDIT-ON-DENIAL — JSONL audit emit for the two denial paths
+# ---------------------------------------------------------------------------
+#
+# Both helpers write to the same ``mcp.tool_called`` seam used by the
+# success-path audit in ``mcp_tools._emit_audit``. They run BEFORE the
+# JSON-RPC error envelope is returned so a denied call leaves a JSONL
+# row even though no executor ran. The ``details.result`` field is the
+# alert-rule discriminator — values match the patterns CC#1's
+# ``capability_denied_24h`` rule (+ future actor_id_required rules)
+# already grep for.
+#
+# Why not call into ``mcp_tools._emit_audit``: that helper takes a
+# ``result`` STRING the executor builds from its own state-change
+# vocabulary (e.g. ``"active->paused"``) and embeds ``args_keys`` from
+# the executor's input. Denial paths don't have either — the request
+# never reached the executor. Keeping a separate helper here avoids
+# pretzeling _emit_audit's contract for two callers with different
+# inputs.
+
+
+def _emit_capability_denied_audit(
+    *, tool_name: Optional[str], caller: Caller
+) -> None:
+    """Write a JSONL audit row for a cap-gate denial.
+
+    Reason text NEVER in the audit (caller-supplied; might leak
+    sensitive context — same posture as the stop-tool's audit
+    omission). Captures only operator-actionable fields: which tool,
+    which caller actor_kind, which capability was required, and the
+    discriminator literal CC#1's alert rule matches on.
+    """
+    # Best-effort emit — never raise from this path. An audit-sink
+    # failure must NOT mask the denial response to the caller.
+    try:
+        from kora_cli.audit import emit_audit
+
+        emit_audit(
+            seam="mcp.tool_called",
+            details={
+                "tool_name": tool_name or "<unknown>",
+                "tool_kind": "mutating",
+                "caller_actor_kind": caller.actor_kind,
+                "caller_actor_id": caller.actor_id,
+                "required_capability": tool_name or "<unknown>",
+                "duration_ms": 0,
+                "tool_status": "not_allowed",
+                "result": "capability_denied",
+            },
+            source="mcp_http",
+        )
+        # Also emit the structured-log line for operator grep — mirrors
+        # the dual-write pattern in mcp_tools._emit_audit.
+        logger.info(
+            "[kora.mcp.tool_denied] tool=%s caller_actor_kind=%s "
+            "result=capability_denied",
+            tool_name,
+            caller.actor_kind,
+        )
+    except Exception:  # pragma: no cover — sink failure must not mask denial
+        logger.exception(
+            "[kora.mcp.tool_denied] emit_audit failed for "
+            "tool=%s caller_actor_kind=%s — denial response still "
+            "returned, but JSONL row missing",
+            tool_name,
+            caller.actor_kind,
+        )
+
+
+def _emit_actor_id_required_audit(
+    *, tool_name: Optional[str], caller: Caller
+) -> None:
+    """Write a JSONL audit row for an actor_id-required denial.
+
+    Distinct ``result`` literal so CC#1's ``capability_denied_24h``
+    alert rule's ``detail_match`` doesn't conflate this with a cap
+    denial — different operator-fix path (the cap IS granted; the
+    actor_id field is missing in mcp_callers.yaml). A future
+    ``actor_id_required_24h`` alert rule can grep this separately.
+    """
+    try:
+        from kora_cli.audit import emit_audit
+
+        emit_audit(
+            seam="mcp.tool_called",
+            details={
+                "tool_name": tool_name or "<unknown>",
+                "tool_kind": "mutating",
+                "caller_actor_kind": caller.actor_kind,
+                "caller_actor_id": caller.actor_id,
+                "required_capability": tool_name or "<unknown>",
+                "duration_ms": 0,
+                "tool_status": "not_allowed",
+                "result": "actor_id_required",
+            },
+            source="mcp_http",
+        )
+        logger.info(
+            "[kora.mcp.tool_denied] tool=%s caller_actor_kind=%s "
+            "result=actor_id_required",
+            tool_name,
+            caller.actor_kind,
+        )
+    except Exception:  # pragma: no cover
+        logger.exception(
+            "[kora.mcp.tool_denied] emit_audit failed for "
+            "tool=%s caller_actor_kind=%s — denial response still "
+            "returned, but JSONL row missing",
+            tool_name,
+            caller.actor_kind,
+        )
+
+
 def _execute_daemon_status() -> Dict[str, Any]:
     """Body for ``kora__daemon_status``. Returns the JSON dict."""
     coord = current_coordinator()
@@ -311,6 +424,15 @@ async def post_jsonrpc(
         # caller's allowed_caps must include the tool name.
         gate_err = _check_cap_gate(req_id, tool_name, caller)
         if gate_err is not None:
+            # KR-MCP-AUDIT-ON-DENIAL — emit a JSONL audit row for the
+            # denial BEFORE returning. The cap-gate is the only gate
+            # that needs this explicit emit; the success-path audit
+            # at mcp_tools._emit_audit runs from inside each executor
+            # AFTER dispatch, so it never sees denied calls. CC#1's
+            # capability_denied_24h alert rule consumes these rows.
+            _emit_capability_denied_audit(
+                tool_name=tool_name, caller=caller
+            )
             return gate_err
 
         if tool_name == "kora__daemon_status":
@@ -376,6 +498,18 @@ async def post_jsonrpc(
                 # caller still lacks the actor_id field needed for
                 # substrate attribution. Operator-fix path is in the
                 # error message.
+                #
+                # KR-MCP-AUDIT-ON-DENIAL — emit symmetrically with the
+                # cap-gate denial so the alerts panel can surface
+                # actor_id-required misconfigurations the same way.
+                # Same seam + same tool_kind; result discriminator is
+                # "actor_id_required" (distinct from
+                # "capability_denied" so the alert rule's
+                # detail_match doesn't conflate the two — they're
+                # different operator-fix paths).
+                _emit_actor_id_required_audit(
+                    tool_name=tool_name, caller=caller
+                )
                 return _jsonrpc_error(
                     req_id,
                     -32001,

tests/kora_cli/alerts/test_aggregator.py

@@ -330,19 +330,46 @@ def fake_read(seam=None, since=None):
     assert str(CAPABILITY_DENIED_24H_THRESHOLD + 1) in matching[0].title
 
 
-def test_capability_denied_today_no_alert_since_audit_doesnt_emit_denials(
+def test_capability_denied_today_emits_alert_when_threshold_exceeded(
     patch_sources,
 ):
-    """Forward-compat: today the audit emit at mcp_tools.py:714 runs
-    AFTER the cap-gate, so denial responses aren't logged. Rule
-    emits zero alerts in the current state — documented in the
-    aggregator module docstring."""
+    """KR-MCP-AUDIT-ON-DENIAL — the cap-gate's denial path now writes
+    ``mcp.tool_called`` rows with ``details.result ==
+    "capability_denied"``. With more than
+    :data:`CAPABILITY_DENIED_24H_THRESHOLD` such rows in the
+    trailing 24h, the alert rule fires exactly one warning so the
+    panel can prompt the operator to review mcp_callers.yaml.
+
+    Previously this test pinned the OPPOSITE behavior (forward-compat
+    "no alert because audit doesn't emit denials yet") — that
+    contract closed when audit-on-denial landed. The test is kept
+    under a renamed identity to lock in the new behavior in the
+    same slot so a future regression can't quietly drop the alert."""
     sources, _ = patch_sources
-    # Only "ok" entries — no capability_denied results.
-    entries = [
-        _make_audit_entry("mcp.tool_called", details={"result": "ok"})
-        for _ in range(CAPABILITY_DENIED_24H_THRESHOLD + 50)
-    ]
+    # 11 denials + a mix of OK + actor_id_required entries that should
+    # NOT count against the capability_denied threshold.
+    entries = (
+        [
+            _make_audit_entry(
+                "mcp.tool_called",
+                details={"result": "capability_denied"},
+            )
+            for _ in range(CAPABILITY_DENIED_24H_THRESHOLD + 1)
+        ]
+        + [
+            _make_audit_entry(
+                "mcp.tool_called", details={"result": "ok"}
+            )
+            for _ in range(7)
+        ]
+        + [
+            _make_audit_entry(
+                "mcp.tool_called",
+                details={"result": "actor_id_required"},
+            )
+            for _ in range(5)
+        ]
+    )
 
     def fake_read(seam=None, since=None):
         return entries if seam == "mcp.tool_called" else []
@@ -352,8 +379,15 @@ def fake_read(seam=None, since=None):
         side_effect=fake_read,
     ):
         alerts = compute_active_alerts()
-    ids = [a.id for a in alerts]
-    assert "capability_denied_24h" not in ids
+    matching = [a for a in alerts if a.id == "capability_denied_24h"]
+    assert len(matching) == 1, (
+        "expected exactly one capability_denied_24h alert when "
+        "11 denials exceed the threshold of 10"
+    )
+    assert matching[0].severity == "info"
+    # Title surfaces the matched-only count (11 — not the mixed-bag
+    # total of 11 + 7 + 5 = 23).
+    assert str(CAPABILITY_DENIED_24H_THRESHOLD + 1) in matching[0].title
 
 
 def test_reasoning_errors_fires_when_over_threshold(patch_sources):