Merge branch 'wt/review-002-con17-mcp-pagination'

src/adapters/from_mcp/mod.rs

@@ -20,6 +20,15 @@
 //! Repeated imports accumulate sessions — reconnecting importers should
 //! import once per process (or wait for a teardown handle in a later
 //! version).
+//!
+//! Tool discovery is a bounded pagination loop over `tools/list` — never
+//! rmcp's unbounded `Peer::list_all_tools`: at most `MCP_MAX_TOOLS_LIST_PAGES`
+//! pages within `MCP_TOOLS_LIST_DEADLINE` overall. Trip either budget and
+//! the import fails loudly with `AdapterError::DiscoveryFailed` naming the
+//! budget (pages fetched, tools accumulated) — no silent truncation, no
+//! partial registration (review-002 CON-14).
+
+use std::time::{Duration, Instant};
 
 use alkcall::client::{AdapterError, OperationAdapter};
 use alkcall::core::types::{Capabilities, Secret};
@@ -33,7 +42,7 @@ use alkcall::registry::spec::{
 };
 use rmcp::model::{
     CallToolRequestParams, CallToolResult, ClientCapabilities, ClientInfo, Content, Implementation,
-    JsonObject, Tool,
+    JsonObject, PaginatedRequestParams, Tool,
 };
 use rmcp::service::RoleClient;
 use rmcp::transport::{
@@ -45,6 +54,12 @@ use serde_json::{Map, Value};
 
 const MCP_CAPABILITY_KEY: &str = "mcp";
 
+/// Upper bound on `tools/list` pages a single `import()` will fetch.
+const MCP_MAX_TOOLS_LIST_PAGES: u32 = 100;
+
+/// Overall time budget for the full `tools/list` pagination walk.
+const MCP_TOOLS_LIST_DEADLINE: Duration = Duration::from_secs(60);
+
 /// Declared error for transport-level `tools/call` failures (review-001
 /// CON-11): the remote is unreachable, the transport closed, or the call
 /// timed out. Payload is the array-of-content-blocks shape; retryable.
@@ -111,12 +126,7 @@ impl OperationAdapter for FromMCP {
             .map_err(|e| classify_init_error(&e))?;
         let peer: Peer<RoleClient> = running.peer().clone();
 
-        let tools = peer
-            .list_all_tools()
-            .await
-            .map_err(|e| AdapterError::DiscoveryFailed {
-                message: format!("tools/list failed: {e}"),
-            })?;
+        let tools = list_all_tools_bounded(&peer).await?;
 
         let bundles = tools
             .into_iter()
@@ -128,6 +138,54 @@ impl OperationAdapter for FromMCP {
     }
 }
 
+/// Bounded `tools/list` pagination walk (review-002 CON-14): the rmcp
+/// equivalent loops `while cursor.is_some()` with no page cap and no
+/// deadline, so a hostile/buggy server that never clears `next_cursor`
+/// hangs `import()` while memory grows without bound. Stops after
+/// [`MCP_MAX_TOOLS_LIST_PAGES`] pages or [`MCP_TOOLS_LIST_DEADLINE`]
+/// elapses and fails loudly — naming pages fetched and tools accumulated —
+/// rather than silently truncating.
+async fn list_all_tools_bounded(peer: &Peer<RoleClient>) -> Result<Vec<Tool>, AdapterError> {
+    let started = Instant::now();
+    let mut tools = Vec::new();
+    let mut cursor = None;
+    for pages_fetched in 1..=MCP_MAX_TOOLS_LIST_PAGES {
+        let remaining = MCP_TOOLS_LIST_DEADLINE
+            .checked_sub(started.elapsed())
+            .ok_or_else(|| pagination_budget_error(pages_fetched, tools.len()))?;
+        let page = match tokio::time::timeout(
+            remaining,
+            peer.list_tools(Some(PaginatedRequestParams::default().with_cursor(cursor))),
+        )
+        .await
+        {
+            Ok(result) => result.map_err(|e| AdapterError::DiscoveryFailed {
+                message: format!("tools/list failed: {e}"),
+            })?,
+            Err(_) => return Err(pagination_budget_error(pages_fetched, tools.len())),
+        };
+        tools.extend(page.tools);
+        cursor = page.next_cursor;
+        if cursor.is_none() {
+            return Ok(tools);
+        }
+    }
+    Err(pagination_budget_error(
+        MCP_MAX_TOOLS_LIST_PAGES,
+        tools.len(),
+    ))
+}
+
+fn pagination_budget_error(pages_fetched: u32, tools_accumulated: usize) -> AdapterError {
+    AdapterError::DiscoveryFailed {
+        message: format!(
+            "tools/list pagination exceeded budget (max {MCP_MAX_TOOLS_LIST_PAGES} pages or \
+             {MCP_TOOLS_LIST_DEADLINE:?} overall) after {pages_fetched} page(s) with \
+             {tools_accumulated} tool(s) accumulated; import failed without partial registration"
+        ),
+    }
+}
+
 fn classify_init_error(e: &rmcp::service::ClientInitializeError) -> AdapterError {
     use rmcp::service::ClientInitializeError as E;
     match e {

tests/from_mcp_integration.rs

@@ -168,6 +168,37 @@ impl ServerHandler for PagingServer {
     }
 }
 
+/// A hostile paging `tools/list` server (review-002 CON-14): always
+/// responds with the same single tool and `next_cursor: Some("a")` — the
+/// cursor cycles and never clears. The importer must bound the walk and
+/// fail with a clean `DiscoveryFailed` rather than hang forever.
+struct CyclingCursorServer;
+
+impl ServerHandler for CyclingCursorServer {
+    fn list_tools(
+        &self,
+        _request: Option<PaginatedRequestParams>,
+        _context: RequestContext<RoleServer>,
+    ) -> impl std::future::Future<Output = Result<ListToolsResult, rmcp::ErrorData>>
+           + rmcp::service::MaybeSendFuture
+           + '_ {
+        let tools = vec![Tool::new_with_raw(
+            "hostile",
+            Some("served on every page, forever".into()),
+            Arc::new(serde_json::Map::new()),
+        )];
+        std::future::ready(Ok(ListToolsResult {
+            meta: None,
+            next_cursor: Some("a".to_string()),
+            tools,
+        }))
+    }
+
+    fn get_info(&self) -> rmcp::model::ServerInfo {
+        rmcp::model::ServerInfo::default()
+    }
+}
+
 async fn spawn_server_for<S: ServerHandler + 'static>(
     server: impl Fn() -> Result<S, std::io::Error> + Send + Sync + 'static,
 ) -> (String, tokio::task::JoinHandle<()>) {
@@ -316,6 +347,39 @@ async fn import_follows_tools_list_pagination() {
     );
 }
 
+#[tokio::test]
+async fn import_cycling_cursor_fails_bounded_with_discovery_failed() {
+    let (endpoint, _handle) = spawn_server_for(|| Ok(CyclingCursorServer)).await;
+    let adapter = FromMCP::new(endpoint, "x");
+    let started = Instant::now();
+    let result = tokio::time::timeout(Duration::from_secs(10), adapter.import()).await;
+    let elapsed = started.elapsed();
+
+    let outcome = result.expect("import must terminate, not hang (CON-14)");
+    match outcome {
+        Ok(_) => panic!("expected Err for a cycling-cursor server"),
+        Err(alkcall::client::AdapterError::DiscoveryFailed { message }) => {
+            assert!(
+                message.contains("pagination exceeded budget"),
+                "error must name the budget, got: {message}"
+            );
+            assert!(
+                message.contains("page"),
+                "error must name pages fetched, got: {message}"
+            );
+            assert!(
+                message.contains("tool"),
+                "error must name tools accumulated, got: {message}"
+            );
+        }
+        Err(other) => panic!("expected DiscoveryFailed, got {other}"),
+    }
+    assert!(
+        elapsed < Duration::from_secs(10),
+        "bounded walk must trip the page cap well inside the outer guard, took {elapsed:?}"
+    );
+}
+
 #[tokio::test]
 async fn import_refuses_tool_name_containing_slash() {
     struct SlashToolServer;