feat: ADR 0032 M3 Slice A — plan gate interceptor in shadow mode (parity-proven)

Per §13.3, lands the plan-gate-as-interceptor migration as the first of two
commits. PlanGateInterceptor (in teaagent/runner/_plan_validator.py) evaluates
the plan gate on TOOL_CALL_REQUESTED. Here it is registered in SHADOW mode
(raise_on_deny=False) and the inline evaluate_write_gate stays authoritative —
so behavior is unchanged. The runner now emits TOOL_CALL_REQUESTED (a new audit
line; frozen contract updated accordingly).

Parity is proven before any enforcement flip: test_plan_gate_interceptor_parity
asserts the interceptor decision equals the inline decision (reason codes are
the oracle), plus a shadow-mode no-veto test and interceptor unit tests.

Slice B (separate commit) flips to enforce and removes the inline gate.

Constraint: shadow only — inline gate authoritative; behavior unchanged except audit gains tool_call_requested; reason codes are the parity oracle
Tested: lifecycle (incl. parity + shadow tests), adversarial + first-hour 35, smoke 200, acceptance 646/646, full mypy clean 1009 files, ruff clean
Confidence: high
Roadmap-Status: unchanged
Allow-test-weakening: frozen audit-contract test updated to include the new tool_call_requested emit; net assertions stronger (parity + interceptor coverage added)

Author: johnteee

teaagent/runner/_core.py

@@ -49,7 +49,7 @@
 from ._approval_manager import RunnerApprovalCoordinator  # noqa: E402
 from ._auto_mode_manager import AutoModeManager  # noqa: E402
 from ._events import EventSpine, RunEventType, register_audit_consumer  # noqa: E402
-from ._plan_validator import PlanValidator  # noqa: E402
+from ._plan_validator import PlanGateInterceptor, PlanValidator  # noqa: E402
 from ._types import (  # noqa: E402
     ApprovalHandler,
     BudgetPromptHandler,
@@ -171,6 +171,22 @@ def __init__(
             require_plan=require_plan,
             skip_plan_check=skip_plan_check,
         )
+
+        # M3-T002 Slice A: register the plan gate interceptor in SHADOW mode.
+        # It computes its decision on TOOL_CALL_REQUESTED but does NOT veto;
+        # the inline evaluate_write_gate below stays authoritative. The parity
+        # test asserts interceptor decision == inline decision per reason code.
+        # Slice B (separate commit) flips this to enforce and removes the inline
+        # gate.
+        self._plan_gate_interceptor = PlanGateInterceptor(
+            self.plan_validator,
+            raise_on_deny=False,
+        )
+        self.event_spine.register_interceptor(
+            self._plan_gate_interceptor,
+            name='plan_gate_shadow',
+        )
+
         self.auto_mode_manager = AutoModeManager(
             auto_mode_config=auto_mode_config,
         )
@@ -624,13 +640,33 @@ def _execute_tool_decision(  # noqa: C901
                 tool_name=decision.tool_name,
                 arguments=decision.arguments,
             )
+
+        # M3-T002 Slice A: emit TOOL_CALL_REQUESTED so the SHADOW plan-gate
+        # interceptor records its decision (it does not veto). The inline gate
+        # below remains authoritative until Slice B.
+        tcr_payload: dict[str, Any] = {
+            'tool_name': decision.tool_name,
+        }
+        if decision.arguments is not None:
+            tcr_payload['arguments'] = decision.arguments
+        plan_contract = context.get('plan_contract')
+        if plan_contract is not None:
+            tcr_payload['plan_contract'] = plan_contract
+        self.event_spine.emit(
+            RunEventType.TOOL_CALL_REQUESTED,
+            run_id,
+            tcr_payload,
+        )
+
+        # M3-T002 Slice A: inline plan gate stays authoritative (removed in B).
         gate_error = self.plan_validator.evaluate_write_gate(
             tool_name=decision.tool_name,
             context=cast(dict[str, Any], context),
             tool_arguments=decision.arguments,
         )
         if gate_error:
             raise ToolPermissionError(gate_error)
+
         # Auto mode: block disallowed tools, auto-approve allowed ones
         self.auto_mode_manager.validate_tool_allowed(decision.tool_name)
         auto_approve_policy = self.auto_mode_manager.get_auto_approve_policy()

teaagent/runner/_plan_validator.py

@@ -6,6 +6,7 @@
 
 from teaagent.governance.plan_gate import assert_write_allowed, assert_write_scope
 from teaagent.policy import ApprovalPolicy, PermissionMode
+from teaagent.runner._events import RunEvent, RunEventType
 from teaagent.spec_exemption import ExemptionDetector, SpecExemptionReceipt
 
 
@@ -147,3 +148,66 @@ def check_read_only_lint_errors(self) -> Optional[str]:
     def get_plan_contract(self) -> Any:
         """Get the current plan contract."""
         return self._plan_contract
+
+
+class PlanGateInterceptor:
+    """EventSpine interceptor that enforces the plan gate on TOOL_CALL_REQUESTED.
+
+    Evaluates ``PlanValidator.evaluate_write_gate()`` and raises
+    ``ToolPermissionError`` if the write is blocked by plan policy, read-only
+    lint, or scope drift.
+
+    In shadow mode (``raise_on_deny=False``) the interceptor records its
+    decision without vetoing — used for parity testing before enforcement
+    is enabled.
+    """
+
+    def __init__(
+        self,
+        plan_validator: PlanValidator,
+        *,
+        raise_on_deny: bool = True,
+    ) -> None:
+        self._pv = plan_validator
+        self._raise_on_deny = raise_on_deny
+        # Exposed for parity assertions: set after each invocation.
+        self.last_decision: str | None = None
+
+    def __call__(self, event: RunEvent) -> None:
+        """Evaluate the plan gate for TOOL_CALL_REQUESTED events.
+
+        Args:
+            event: The run event.
+
+        Raises:
+            ToolPermissionError: If the gate blocks and raise_on_deny is True.
+        """
+        if event.type != RunEventType.TOOL_CALL_REQUESTED:
+            return
+
+        tool_name = event.payload['tool_name']
+        arguments = event.payload.get('arguments')
+        # Reconstruct minimal context from event payload for plan_contract check.
+        context: dict[str, Any] = {}
+        plan_contract = event.payload.get('plan_contract')
+        if plan_contract is not None:
+            context['plan_contract'] = plan_contract
+
+        from teaagent.errors import ToolPermissionError
+
+        try:
+            gate_error = self._pv.evaluate_write_gate(
+                tool_name=tool_name,
+                context=context,
+                tool_arguments=arguments,
+            )
+        except ToolPermissionError as exc:
+            # evaluate_write_gate can raise directly from assert_write_allowed.
+            self.last_decision = str(exc)
+            if self._raise_on_deny:
+                raise
+            return
+
+        self.last_decision = gate_error
+        if gate_error and self._raise_on_deny:
+            raise ToolPermissionError(gate_error)