feat(mcp): add verifier latency observer

docs/mcp.md

@@ -251,7 +251,7 @@ Use `MCPVerifier` directly when you're not using FastMCP — works with the raw
 
 ```python
 from tenuo import Authorizer, PublicKey, CompiledMcpConfig, McpConfig
-from tenuo.mcp import MCPVerifier
+from tenuo.mcp import MCPVerificationResult, MCPVerifier
 
 authorizer = Authorizer(trusted_roots=[PublicKey.from_bytes(root_pub)])
 config = CompiledMcpConfig.compile(McpConfig.from_file("mcp-config.yaml"))
@@ -266,6 +266,35 @@ execute_tool(result.clean_arguments)
 clean = verifier.verify_or_raise("read_file", {"path": path}, meta=request_meta)
 ```
 
+To record end-to-end verification latency in your own metrics pipeline, pass a
+`latency_observer`. It receives a result snapshot and elapsed time in
+microseconds on every allow and denial path. Observer failures and snapshot
+mutations never change the authorization decision.
+
+```python
+from prometheus_client import Histogram
+
+verification_latency = Histogram(
+    "tenuo_mcp_verification_latency_seconds",
+    "End-to-end MCP warrant verification latency",
+    ["tool", "allowed"],
+)
+
+
+def observe_latency(result: MCPVerificationResult, latency_us: int) -> None:
+    verification_latency.labels(
+        tool=result.tool,
+        allowed=str(result.allowed).lower(),
+    ).observe(latency_us / 1_000_000)
+
+
+verifier = MCPVerifier(
+    authorizer=authorizer,
+    config=config,
+    latency_observer=observe_latency,
+)
+```
+
 ### Pattern 4: Securing LangChain MCP Adapters
 
 If you're already using `langchain-mcp-adapters`, wrap its tools with `guard_tools()`:

tenuo-python/tenuo/mcp/server.py

@@ -104,8 +104,10 @@ async def call_tool(name: str, arguments: dict) -> list:
 
 import base64
 import logging
+import time
+from copy import deepcopy
 from dataclasses import dataclass, field
-from typing import Any, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional
 
 from .._pop_canonicalize import strip_none_values
 from ..exceptions import (
@@ -375,6 +377,9 @@ def __init__(
         require_warrant: bool = True,
         control_plane: Optional[Any] = None,
         nonce_store: Optional[Any] = None,
+        latency_observer: Callable[
+            [MCPVerificationResult, int], None
+        ] | None = None,
     ) -> None:
         """
         Args:
@@ -399,6 +404,12 @@ def __init__(
                 ``enable_default_nonce_store()`` at startup or pass an explicit
                 ``NonceStore(backend=RedisNonceBackend(...))`` for distributed
                 deployments.
+            latency_observer: Optional callback invoked after every verification
+                with a snapshot of the final :class:`MCPVerificationResult` and
+                end-to-end elapsed time in microseconds. Observer failures and
+                snapshot mutations never change the authorization result. Use
+                this to feed local Prometheus, OpenTelemetry, or structured-log
+                instrumentation.
         """
         from tenuo._extension import require_extension
         require_extension("MCPVerifier")
@@ -411,6 +422,7 @@ def __init__(
             control_plane = get_or_create()
         self._control_plane = control_plane
         self._nonce_store = nonce_store
+        self._latency_observer = latency_observer
 
     def verify(
         self,
@@ -441,27 +453,38 @@ def verify(
             or ``allowed=False`` with ``denial_reason`` and
             ``jsonrpc_error_code`` on failure.
         """
-        args: Dict[str, Any] = arguments or {}
-        # PoP bytes cover the wire-args view. Both client and server apply
-        # strip_none_values to that view so optional arguments with None
-        # defaults don't crash the Rust canonicalizer and don't silently
-        # diverge the signed-bytes shape between sides.
-        pop_args: Dict[str, Any] = strip_none_values(args)
+        start_ns = time.perf_counter_ns()
 
         def _emit_and_return(
             result: MCPVerificationResult,
             chain_result: Any = None,
-            latency_us: int = 0,
         ) -> MCPVerificationResult:
+            latency_us = (time.perf_counter_ns() - start_ns) // 1000
             if self._control_plane:
                 try:
                     self._control_plane.emit_for_enforcement(
                         result, chain_result=chain_result, latency_us=latency_us
                     )
                 except Exception:
                     logger.warning("Control plane emission failed for '%s'; audit event lost", result.tool, exc_info=True)
+            if self._latency_observer is not None:
+                try:
+                    self._latency_observer(deepcopy(result), latency_us)
+                except Exception:
+                    logger.warning(
+                        "Latency observer failed for '%s'",
+                        result.tool,
+                        exc_info=True,
+                    )
             return result
 
+        args: Dict[str, Any] = arguments or {}
+        # PoP bytes cover the wire-args view. Both client and server apply
+        # strip_none_values to that view so optional arguments with None
+        # defaults don't crash the Rust canonicalizer and don't silently
+        # diverge the signed-bytes shape between sides.
+        pop_args: Dict[str, Any] = strip_none_values(args)
+
         # ------------------------------------------------------------------
         # Step 1: extract Tenuo envelope from params._meta
         # ------------------------------------------------------------------
@@ -676,8 +699,6 @@ def _emit_and_return(
         # ------------------------------------------------------------------
         # Step 6: authorize
         # ------------------------------------------------------------------
-        import time
-        start_ns = time.perf_counter_ns()
         chain_result = None
         result: MCPVerificationResult
 
@@ -730,7 +751,6 @@ def _emit_and_return(
                         ),
                         jsonrpc_error_code=-32001,
                     ),
-                    latency_us=(time.perf_counter_ns() - start_ns) // 1000,
                 )
 
             logger.debug("MCP call authorized for '%s' (warrant=%s)", tool_name, warrant_id)
@@ -861,8 +881,7 @@ def _emit_and_return(
                 jsonrpc_error_code=-32001,
             )
 
-        latency_us = (time.perf_counter_ns() - start_ns) // 1000
-        return _emit_and_return(result, chain_result=chain_result, latency_us=latency_us)
+        return _emit_and_return(result, chain_result=chain_result)
 
     def verify_or_raise(
         self,

tenuo-python/tests/adapters/test_mcp_server.py

@@ -945,6 +945,116 @@ def test_default_nonce_store_used_when_enabled(
         assert "replay" in (result2.denial_reason or "").lower()
 
 
+# ---------------------------------------------------------------------------
+# MCPVerifier latency observer
+# ---------------------------------------------------------------------------
+
+
+class TestLatencyObserver:
+    def test_early_denial_observes_end_to_end_latency(
+        self, authorizer: Authorizer
+    ) -> None:
+        from unittest.mock import MagicMock, patch
+
+        observed: list[tuple[MCPVerificationResult, int]] = []
+        mock_cp = MagicMock()
+        verifier = MCPVerifier(
+            authorizer=authorizer,
+            control_plane=mock_cp,
+            latency_observer=lambda result, latency_us: observed.append(
+                (result, latency_us)
+            ),
+        )
+
+        with patch(
+            "tenuo.mcp.server.time.perf_counter_ns",
+            side_effect=[1_000_000, 6_000_000],
+        ):
+            result = verifier.verify("read_file", {"path": "/x"})
+
+        assert not result.allowed
+        assert observed == [(result, 5_000)]
+        assert mock_cp.emit_for_enforcement.call_args.kwargs["latency_us"] == 5_000
+
+    def test_authorized_call_observes_latency(
+        self,
+        authorizer: Authorizer,
+        simple_warrant: Warrant,
+        agent_key: SigningKey,
+    ) -> None:
+        from unittest.mock import MagicMock
+
+        observer = MagicMock()
+        verifier = MCPVerifier(
+            authorizer=authorizer,
+            control_plane=None,
+            latency_observer=observer,
+        )
+        tool_args = {"path": "/data/f.txt"}
+        arguments, meta = _make_arguments(
+            simple_warrant, agent_key, "read_file", tool_args
+        )
+
+        result = verifier.verify("read_file", arguments, meta=meta)
+
+        assert result.allowed
+        observer.assert_called_once()
+        observed_result, latency_us = observer.call_args.args
+        assert observed_result == result
+        assert observed_result is not result
+        assert latency_us >= 0
+
+    def test_observer_failure_does_not_change_result_or_audit(
+        self,
+        authorizer: Authorizer,
+        caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        from unittest.mock import MagicMock
+
+        def mutate_and_raise(result: MCPVerificationResult, latency_us: int) -> None:
+            result.allowed = True
+            result.clean_arguments["path"] = "/mutated"
+            raise RuntimeError("metrics unavailable")
+
+        observer = MagicMock(side_effect=mutate_and_raise)
+        mock_cp = MagicMock()
+        verifier = MCPVerifier(
+            authorizer=authorizer,
+            control_plane=mock_cp,
+            latency_observer=observer,
+        )
+
+        result = verifier.verify("read_file", {"path": "/x"})
+
+        assert not result.allowed
+        assert result.clean_arguments == {"path": "/x"}
+        observer.assert_called_once()
+        mock_cp.emit_for_enforcement.assert_called_once()
+        assert "Latency observer failed for 'read_file'" in caplog.text
+
+    def test_falsey_callable_observer_is_invoked(
+        self, authorizer: Authorizer
+    ) -> None:
+        class FalseyObserver:
+            def __init__(self) -> None:
+                self.calls = 0
+
+            def __bool__(self) -> bool:
+                return False
+
+            def __call__(
+                self, result: MCPVerificationResult, latency_us: int
+            ) -> None:
+                self.calls += 1
+
+        observer = FalseyObserver()
+        verifier = MCPVerifier(authorizer=authorizer, latency_observer=observer)
+
+        verifier.verify("read_file", {"path": "/x"})
+
+        assert observer.calls == 1
+
+
 # ---------------------------------------------------------------------------
 # Control plane emission coverage tests
 # ---------------------------------------------------------------------------