fix(discovery): exhaust paginated tools/list before comparison (#633)

* fix(discovery): exhaust upstream tool catalog pagination

Signed-off-by: Noah Ingwers <98993329+noah-ing@users.noreply.github.com>

* test(discovery): write cancellation handshake as exact bytes

Signed-off-by: Noah Ingwers <98993329+noah-ing@users.noreply.github.com>

* test(stdio): cover later-page transport failure

Signed-off-by: Noah Ingwers <98993329+noah-ing@users.noreply.github.com>

* fix(discovery): share first-contact acquisition across checks

Signed-off-by: Noah Ingwers <98993329+noah-ing@users.noreply.github.com>

* test(discovery): tolerate cancelled child socket reset on Windows

Signed-off-by: Noah Ingwers <98993329+noah-ing@users.noreply.github.com>

---------

Signed-off-by: Noah Ingwers <98993329+noah-ing@users.noreply.github.com>

Author: noah-ing

CHANGELOG.md

@@ -161,6 +161,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **Exhaust upstream `tools/list` pagination before drift and provenance
+  comparisons** (#631). HTTP and stdio share bounded acquisition: later-page
+  failures, malformed or ambiguous listings, and cursor cycles are unchecked,
+  never a comparison against a partial catalog. Existing drift policy and
+  unchecked-call behavior are unchanged. Observed by solloek369-arch on #566
+  and confirmed in #631 by Imran Siddique.
+  Drift and provenance now share one completed discovery acquisition per
+  server/authority per session, including unchecked outcomes, avoiding a second
+  full pagination walk on cold calls with provenance configured. Concurrent
+  readers wait for completion; cancelled reads are not cached.
+  Session rebinding resets acquisition and comparison caches together. The
+  duplicate-fetch cost was identified by qubeena07 during review of #633.
+
 - TLS pinning test fixtures set `minimum_version = TLSv1_2`; the server was built
   with `PROTOCOL_TLS_SERVER` and no floor, leaving TLSv1 and TLSv1.1 reachable in
   the test that asserts the gateway's transport rules.

LIMITATIONS.md

@@ -25,6 +25,10 @@ cMCP compares what each upstream server advertises against the approved catalog
 
 Separately, the approved description rather than the live one is what the gateway serves to the agent on `tools/list`, so a mutated description does not reach the model through cMCP even in the windows above. That is a structural property of proxying an approved catalog, not a detection result, and it does not extend to the tool's behaviour once called.
 
+HTTP and stdio discovery exhaust `tools/list` pagination before comparing either drift or provenance. A later-page failure, malformed discovery shape, duplicate tool name, repeated/cyclic cursor, or continuation beyond 1,000 pages makes the entire acquisition unchecked; no partial list is compared. Cursors are passed back unchanged, including an empty string. This bound limits page count, not total elapsed time or response bytes, and pagination does not establish an atomic snapshot of a changing server. This is acquisition validation, not full MCP schema validation or a new approval/hash policy. The unchecked-call behavior above is unchanged.
+
+Drift and provenance share the completed first-contact acquisition for the same server and publisher authority within a session, rather than independently walking all pages. An unchecked acquisition is also cached until the next session. Cancellation leaves no cached acquisition; existing stdio child-close behavior is unchanged, not an automatic child restart. Close/reset drains admitted calls before cleanup invalidates both comparisons and the shared acquisition, even if resource cleanup subsequently fails. This avoids duplicate discovery work; it does not add continuous monitoring or make the listing an atomic snapshot.
+
 **Phase 2 completeness: server-side attestation**
 Phase 1 attests the gateway boundary. It does not attest what happens on the other side of that boundary. The `tool_transcript.hash` field in the TRACE Claim records a hash of the audit chain tip, but the tool transcript binding that ties a specific tool execution to a specific response is Phase 2 work. Phase 1 partially addresses P1.4 (transitive trust into upstream dependencies) and P4.1 (typosquatted packages added to catalog) -- both are fully closed by Phase 2. Any compliance claim that relies on server-side proof must wait for Phase 2.
 

src/cmcp_runtime/mcp/discovery.py

@@ -0,0 +1,68 @@
+"""Acquire a complete tools/list before drift or provenance comparison.
+
+This is acquisition validation, not approval or a catalog hash construction.
+No partial result escapes: every page must be attributable and well-shaped,
+and pagination must terminate within the local page budget.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+# A page bound also terminates servers that issue endlessly distinct cursors.
+# It is not a wall-clock deadline or a guarantee of an atomic server snapshot.
+MAX_DISCOVERY_PAGES = 1000
+
+PageFetcher = Callable[[str, dict[str, Any]], Awaitable[dict[str, Any]]]
+
+
+class DiscoveryError(ValueError):
+    """A bounded local reason; never embed upstream payloads or cursor values."""
+
+
+async def collect_tools(fetch_page: PageFetcher) -> list[dict[str, Any]]:
+    """Return only an exhausted, unambiguous listing; otherwise raise.
+
+    Transport failures and cancellation propagate to the owning transport.
+    Cursors are opaque strings, including the empty string. Only absence of
+    nextCursor terminates a listing; an empty page alone does not.
+    """
+    tools: list[dict[str, Any]] = []
+    names: set[str] = set()
+    cursors: set[str] = set()
+    params: dict[str, Any] = {}
+    for page in range(MAX_DISCOVERY_PAGES):
+        request_id = f"provenance-tools-list-{page}"
+        body = await fetch_page(request_id, params)
+        if (
+            not isinstance(body, dict)
+            or body.get("jsonrpc") != "2.0"
+            or body.get("id") != request_id
+            or "error" in body
+        ):
+            raise DiscoveryError("invalid response envelope")
+        result = body.get("result")
+        if not isinstance(result, dict) or not isinstance(result.get("tools"), list):
+            raise DiscoveryError("invalid tools page")
+        for tool in result["tools"]:
+            if (
+                not isinstance(tool, dict)
+                or not isinstance(tool.get("name"), str)
+                or not tool["name"]
+            ):
+                raise DiscoveryError("invalid tool name")
+            if tool["name"] in names:
+                raise DiscoveryError("duplicate tool name")
+            names.add(tool["name"])
+            tools.append(tool)
+        if "nextCursor" not in result:
+            return tools
+        cursor = result["nextCursor"]
+        if not isinstance(cursor, str):
+            raise DiscoveryError("invalid continuation cursor")
+        if cursor in cursors:
+            raise DiscoveryError("repeated continuation cursor")
+        cursors.add(cursor)
+        params = {"cursor": cursor}
+    raise DiscoveryError("discovery page limit exceeded")

src/cmcp_runtime/mcp/proxy.py

@@ -42,6 +42,7 @@
 )
 from cmcp_runtime.execution import valid_execution_id
 from cmcp_runtime.mcp import tls_pinning
+from cmcp_runtime.mcp.discovery import DiscoveryError, collect_tools
 from cmcp_runtime.mcp.stdio import StdioServer
 from cmcp_runtime.mcp.streamable_http import (
     build_request,
@@ -291,18 +292,9 @@ def __init__(
         # in memory would carry it from one agent's session into the next, and
         # the audit chain cannot see that happen (docs/spec/stdio-transport.md).
         self._stdio_servers: dict[tuple[str, ...], StdioServer] = {}
-        # Provenance outcome per server, decided once per session on first use.
-        # Cached because the answer cannot change within a session without the
-        # server being replaced underneath us, and re-listing tools on every call
-        # would make the check expensive enough to be turned off.
-        self._provenance: dict[tuple[str, ...], ProvenanceResult] = {}
+        self._reset_upstream_checks()
         # Servers already warned about unenforceable pinning (warn once each).
         self._tls_pin_warned: set[str] = set()
-        # #521: servers whose advertised tool definitions have been compared against
-        # the catalog. Cached per server for the same reason provenance is: one
-        # tools/list round trip per server per session is affordable, one per call
-        # is not, and a check expensive enough to hurt is a check that gets disabled.
-        self._drift_checked: set[tuple[str, ...]] = set()
         self._catalog_scanner = catalog_scanner
         # #625: serialises first-use stdio spawns so two calls racing on the
         # same server's first use cannot both spawn a child - see `_stdio_for`.
@@ -328,6 +320,17 @@ def __init__(
         self._failed_terminal_call: str | None = None
         self._shutting_down = False
 
+    def _reset_upstream_checks(self) -> None:
+        # Drift and provenance share one completed paginated acquisition per
+        # server/authority per session, including an unchecked (None) outcome.
+        # These are first-contact observations, not continuous monitoring.
+        # Replace, rather than clear: an in-flight acquisition retains its old
+        # cache identity and must retry before returning into a new session.
+        self._advertised: dict[tuple[str, ...], list[dict[str, Any]] | None] = {}
+        self._discovery_locks: dict[tuple[str, ...], asyncio.Lock] = {}
+        self._provenance: dict[tuple[str, ...], ProvenanceResult] = {}
+        self._drift_checked: set[tuple[str, ...]] = set()
+
     def _ensure_running(self) -> None:
         if self._shutting_down:
             raise UpstreamUnavailable("gateway is shutting down")
@@ -718,8 +721,7 @@ async def aclose(self) -> None:
         """
         stdio_items = tuple(self._stdio_servers.items())
         http_items = tuple(self._http_clients.items())
-        self._provenance.clear()
-        self._drift_checked.clear()
+        self._reset_upstream_checks()
 
         if not stdio_items and not http_items:
             self._cleanup_incomplete = False
@@ -766,30 +768,54 @@ async def aclose(self) -> None:
         self._cleanup_incomplete = False
 
     async def _advertised_tools(self, entry: CatalogEntry) -> list[dict[str, Any]] | None:
-        """What the server offers *this gateway*, for the provenance comparison.
+        """One first-contact acquisition shared by drift and provenance checks.
 
         Returns ``None`` when the server will not say, which the caller records as
         ``unchecked`` rather than as a pass. Never falls back to the catalog's own
         approved definitions: comparing a record against our approval instead of
         against the server is the substitution that turns the check into theatre.
         """
+        key = _server_provenance_key(entry)
+        while True:
+            cache = self._advertised
+            lock = self._discovery_locks.setdefault(key, asyncio.Lock())
+            async with lock:
+                if cache is not self._advertised:
+                    continue
+                if key in cache:
+                    return cache[key]
+                advertised = await self._discover_tools(entry)
+                if cache is not self._advertised:
+                    continue
+                # Only completed acquisition outcomes are cached. In particular,
+                # cancellation propagates without storing partial data or None.
+                cache[key] = advertised
+                return advertised
+
+    async def _discover_tools(self, entry: CatalogEntry) -> list[dict[str, Any]] | None:
+        """Acquire the entire listing, or None on an ordinary acquisition failure."""
         if entry.server.is_stdio:
             return await (await self._stdio_for(entry)).list_tools()
-        try:
+
+        async def fetch_page(request_id: str, params: dict[str, Any]) -> dict[str, Any]:
+            payload, headers = build_request(request_id, "tools/list", params)
             client = self._client_for_upstream(entry)
-            payload, headers = build_request("provenance-tools-list", "tools/list", {})
             resp = await client.post(
                 entry.server.url,
                 json=payload,
                 headers=headers,
             )
             resp.raise_for_status()
-            result = parse_response(resp, "provenance-tools-list").get("result")
-        except Exception as exc:  # noqa: BLE001 - any failure means "could not check"
-            logger.warning("could not list tools for provenance check: %s", exc)
-            return None
-        tools = result.get("tools") if isinstance(result, dict) else None
-        return tools if isinstance(tools, list) else None
+            return parse_response(resp, request_id)
+
+        try:
+            return await collect_tools(fetch_page)
+        except DiscoveryError as exc:
+            logger.warning("tools discovery incomplete: %s", exc)
+        except Exception:  # noqa: BLE001 - any acquisition failure means "could not check"
+            # Upstream exceptions may contain response bodies or opaque cursors.
+            logger.warning("tools discovery incomplete: upstream request failed")
+        return None
 
     async def _check_upstream_drift(self, entry: CatalogEntry) -> bool:
         """Compare what a server advertises against what we approved (P4.2).
@@ -810,14 +836,17 @@ async def _check_upstream_drift(self, entry: CatalogEntry) -> bool:
         key = _server_provenance_key(entry)
         if key in self._drift_checked:
             return self._session.catalog_drift
-        self._drift_checked.add(key)
-
         advertised = await self._advertised_tools(entry)
+        # Another caller may have completed the comparison while this one was
+        # waiting for discovery. An in-flight check is never marked completed.
+        if key in self._drift_checked:
+            return self._session.catalog_drift
         if advertised is None:
             logger.info(
                 "upstream drift: server=%s outcome=unchecked (server would not list tools)",
                 key,
             )
+            self._drift_checked.add(key)
             return self._session.catalog_drift
 
         by_name = {
@@ -840,6 +869,7 @@ async def _check_upstream_drift(self, entry: CatalogEntry) -> bool:
 
         if not drifted:
             logger.info("upstream drift: server=%s outcome=match", key)
+            self._drift_checked.add(key)
             return self._session.catalog_drift
 
         fail_closed = self._config.catalog.drift_policy is DriftPolicy.FAIL_CLOSED
@@ -872,6 +902,7 @@ async def _check_upstream_drift(self, entry: CatalogEntry) -> bool:
 
         if fail_closed:
             self._session.catalog_drift = True
+        self._drift_checked.add(key)
         return self._session.catalog_drift
 
     async def _check_provenance(self, entry: CatalogEntry) -> ProvenanceResult:

src/cmcp_runtime/mcp/stdio.py

@@ -47,6 +47,7 @@ class rather than folded into hardware attestation.
 from typing import Any
 
 from cmcp_runtime.errors import ConfigError, UpstreamToolError, UpstreamUnavailable
+from cmcp_runtime.mcp.discovery import DiscoveryError, collect_tools
 
 logger = logging.getLogger(__name__)
 
@@ -244,14 +245,21 @@ async def list_tools(self) -> list[dict[str, Any]] | None:
         is one whose provenance could not be checked, and the two are recorded
         differently.
         """
+        async def fetch_page(request_id: str, params: dict[str, Any]) -> dict[str, Any]:
+            return await self._request(request_id, "tools/list", params)
+
         try:
-            body = await self._request("provenance-tools-list", "tools/list", {})
-        except (UpstreamUnavailable, UpstreamToolError) as exc:
-            logger.warning("could not list tools for provenance check: %s", exc)
-            return None
-        result = body.get("result")
-        tools = result.get("tools") if isinstance(result, dict) else None
-        return tools if isinstance(tools, list) else None
+            return await collect_tools(fetch_page)
+        except asyncio.CancelledError:
+            # An interrupted read can leave a late reply in the pipe. Do not
+            # let a later tool call consume it as that call's response.
+            await self.close()
+            raise
+        except DiscoveryError as exc:
+            logger.warning("tools discovery incomplete: %s", exc)
+        except (UpstreamUnavailable, UpstreamToolError):
+            logger.warning("tools discovery incomplete: upstream request failed")
+        return None
 
     async def call(self, call_id: str, tool_name: str, arguments: dict[str, Any]) -> str:
         """One JSON-RPC ``tools/call`` over the child's stdin/stdout."""