feat(proxy): record advertised but unadmitted tools as audit observat…

src/cmcp_runtime/audit/chain.py

@@ -30,6 +30,7 @@
     "suspicious_call_sequence",
     "attestation_stale",
     "catalog_drift",
+    "tool_observed_unadmitted",
     "break_glass_used",
 ]
 

src/cmcp_runtime/mcp/proxy.py

@@ -61,6 +61,10 @@
 
 logger = logging.getLogger(__name__)
 
+# Per-comparison audit recording bounds, not discovery or admission limits.
+NAME_OBSERVATION_CAP = 64
+TOOL_NAME_RECORD_MAX_LENGTH = 256
+
 
 _EXTERNAL_EVIDENCE_FIELDS: frozenset[str] = frozenset(
     {
@@ -927,15 +931,19 @@ async def _check_upstream_drift(self, entry: CatalogEntry) -> bool:
             self._drift_checked.add(key)
             return self._session.catalog_drift
 
-        by_name = {
-            t.get("name"): t
+        by_name: dict[str, dict[str, Any]] = {
+            t["name"]: t
             for t in advertised
             if isinstance(t, dict) and isinstance(t.get("name"), str)
         }
         drifted: list[tuple[str, str]] = []
+        active_admitted_names: set[str] = set()
+        active_exception_count = 0
         for tool_name, catalog_entry in self._catalog.entries.items():
             if _server_provenance_key(catalog_entry) != key:
                 continue
+            active_admitted_names.add(tool_name)
+            active_exception_count += int(catalog_entry.catalog_exception)
             offered = by_name.get(tool_name)
             if offered is None:
                 drifted.append((tool_name, "withdrawn"))
@@ -945,6 +953,54 @@ async def _check_upstream_drift(self, entry: CatalogEntry) -> bool:
             ):
                 drifted.append((tool_name, "definition_changed"))
 
+        # Extra names are evidence, not drift. The active basis includes runtime
+        # exceptions; catalog_hash still identifies the original measured catalog.
+        unadmitted_names = sorted(by_name.keys() - active_admitted_names)
+        observation_context: dict[str, str | int | float] = {
+            "source": "upstream",
+            "measured_catalog_hash": self._catalog.catalog_hash,
+            "admission_basis": "active_catalog_entries",
+            "active_admitted_count": len(active_admitted_names),
+            "active_exception_count": active_exception_count,
+        }
+        for tool_name in unadmitted_names[:NAME_OBSERVATION_CAP]:
+            truncated = len(tool_name) > TOOL_NAME_RECORD_MAX_LENGTH
+            detail: dict[str, str | int | float] = {
+                **observation_context,
+                "status": "observed_unadmitted",
+                "recorded_name_truncated": truncated,
+            }
+            if truncated:
+                # A recorded prefix is not a tool identity. Bind the original
+                # name without persisting it; surrogatepass handles every str
+                # accepted by the existing full-name comparison.
+                detail["tool_name_original_length"] = len(tool_name)
+                detail["tool_name_sha256"] = hashlib.sha256(
+                    tool_name.encode("utf-8", errors="surrogatepass")
+                ).hexdigest()
+            self._audit.append(
+                "tool_observed_unadmitted",
+                tool_name=tool_name[:TOOL_NAME_RECORD_MAX_LENGTH],
+                server_identity=entry.server.url,
+                detail=detail,
+                session_sensitivity_before=self._session.max_sensitivity,
+                session_sensitivity_after=self._session.max_sensitivity,
+            )
+        omitted_name_count = len(unadmitted_names) - NAME_OBSERVATION_CAP
+        if omitted_name_count > 0:
+            self._audit.append(
+                "tool_observed_unadmitted",
+                tool_name=None,
+                server_identity=entry.server.url,
+                detail={
+                    **observation_context,
+                    "status": "observed_unadmitted_summary",
+                    "omitted_name_count": omitted_name_count,
+                },
+                session_sensitivity_before=self._session.max_sensitivity,
+                session_sensitivity_after=self._session.max_sensitivity,
+            )
+
         if not drifted:
             logger.info("upstream drift: server=%s outcome=match", key)
             self._drift_checked.add(key)