diff --git a/src/openhuman/mcp_registry/registries/mcp_official.rs b/src/openhuman/mcp_registry/registries/mcp_official.rs index 1393ad179..0b5d9b0f4 100644 --- a/src/openhuman/mcp_registry/registries/mcp_official.rs +++ b/src/openhuman/mcp_registry/registries/mcp_official.rs @@ -8,10 +8,35 @@ //! - `GET /v0/servers/{name}` — full detail for one server (or a fallback //! path that searches by exact name when the direct endpoint 404s) //! -//! The official registry uses cursor pagination. We map our 1-indexed `page` -//! parameter onto it by treating `page == 1` as "no cursor" and refusing -//! deeper pagination for now — the caller gets back the first page plus a -//! `total_pages` hint of `1`. Cursor-aware pagination is a follow-up. +//! ## Pagination model +//! +//! The official registry uses cursor pagination: each list response carries +//! an opaque `metadata.nextCursor` token (or no token at all when the result +//! set ends). The OpenHuman trait, however, talks in 1-indexed `page` +//! numbers — Smithery's native shape — so this adapter maps page → cursor by +//! caching the cursor that produced each page in a per-process `HashMap` +//! keyed by `(query, page_size, page)`. +//! +//! On a `page > 1` request: +//! - The adapter looks up the cursor that produced `page - 1` in the cache. +//! - **Cache hit**: one HTTP fetch with that cursor. +//! - **Cache miss** (typical after a process restart, or a deep-link to page +//! N without having walked 1..N-1): the adapter walks `page = 1` forward +//! sequentially, caching each cursor as it goes, until it has fetched the +//! requested page. Walks beyond `MAX_CURSOR_WALK_PAGES` bail rather than +//! risk a DoS — UIs that need deep deep-links should switch to a paging +//! surface that follows the cursor explicitly. +//! +//! `total_pages` is reported as `page + 1` when the response includes a +//! `nextCursor`, else `page`. Matches the trait doc: "best-effort upper +//! bound — registries that can't compute it report the current page number." +//! +//! ## Response shape +//! +//! The list endpoint wraps each server as `{ "server": { ... }, "_meta": ... }`. +//! The previous DTO assumed a flat shape and silently produced empty +//! summaries when `serde` filled the missing top-level fields with defaults. +//! [`OfficialServerEnvelope`] now matches the real wire shape. //! //! Auth: optional `MCP_OFFICIAL_REGISTRY_TOKEN` env var sent as bearer. //! @@ -19,9 +44,12 @@ use anyhow::{Context, Result}; use async_trait::async_trait; +use parking_lot::Mutex; use reqwest::Client; use serde::Deserialize; use serde_json::Value; +use std::collections::HashMap; +use std::sync::OnceLock; use crate::openhuman::config::Config; @@ -31,6 +59,45 @@ use super::{Registry, SOURCE_MCP_OFFICIAL}; const DEFAULT_BASE: &str = "https://registry.modelcontextprotocol.io"; +/// Cap on the sequential cursor walk for deep-page cache misses. +/// +/// At `page_size = 50` this allows the UI to deep-link up to the 2500th +/// result without a primed cursor cache. Walks past this point bail rather +/// than fan a single user request into hundreds of upstream requests — +/// pagination UIs that need to go deeper should call sequentially so the +/// cache builds up naturally. +const MAX_CURSOR_WALK_PAGES: u32 = 50; + +/// Per-process cache mapping `(query, page_size, page)` → cursor that +/// produced *that* page. Cursor for `page = 1` is the empty string (no +/// cursor sent), so we only insert entries for `page >= 2`. +/// +/// `parking_lot::Mutex` matches the rest of the memory subsystem and keeps +/// the critical section synchronous — every access is a `HashMap` op, no +/// `.await` while the lock is held. +fn cursor_cache() -> &'static Mutex> { + static CACHE: OnceLock>> = OnceLock::new(); + CACHE.get_or_init(|| Mutex::new(HashMap::new())) +} + +fn cursor_cache_get(query: &str, page_size: u32, page: u32) -> Option { + cursor_cache() + .lock() + .get(&(query.to_string(), page_size, page)) + .cloned() +} + +fn cursor_cache_set(query: &str, page_size: u32, page: u32, cursor: String) { + cursor_cache() + .lock() + .insert((query.to_string(), page_size, page), cursor); +} + +#[cfg(test)] +fn cursor_cache_clear() { + cursor_cache().lock().clear(); +} + pub struct McpOfficialRegistry; #[async_trait] @@ -48,57 +115,64 @@ impl Registry for McpOfficialRegistry { ) -> Result<(Vec, u32)> { let q = query.unwrap_or("").trim(); let limit = page_size.max(1); + let page = page.max(1); let cache_key = format!("mcp_official:search:{q}:{page}:{limit}"); if let Ok(Some(cached_body)) = store::get_cached(config, &cache_key) { - tracing::debug!("[mcp-official] search cache hit key={cache_key}"); + tracing::debug!( + "[mcp-official] search cache hit has_query={} q_len={} page={page} limit={limit}", + !q.is_empty(), + q.len() + ); if let Ok(parsed) = serde_json::from_str::(&cached_body) { - return Ok((parsed.into_summaries(), 1)); + let total_pages = total_pages_hint(page, parsed.next_cursor().is_some()); + if let Some(cursor) = parsed.next_cursor() { + cursor_cache_set(q, limit, page, cursor.to_string()); + } + return Ok((parsed.into_summaries(), total_pages)); } } - if page > 1 { - // Cursor pagination not yet wired up — return empty for page > 1 - // so the UI doesn't loop fetching nonexistent pages. - tracing::debug!( - "[mcp-official] search returning empty for page>1 \ - (cursor pagination not implemented)" - ); - return Ok((Vec::new(), 1)); - } - - tracing::debug!("[mcp-official] search fetching q={q:?} limit={limit}"); - - let client = http_client()?; - let url = format!("{}/v0/servers", base_url()); - let mut req = client.get(&url).header("Accept", "application/json"); - if !q.is_empty() { - req = req.query(&[("search", q)]); - } - req = req.query(&[("limit", &limit.to_string())]); - req = apply_auth(req); - - let resp = req.send().await.context("MCP official search failed")?; - let status = resp.status(); - let body = resp.text().await.context("MCP official read failed")?; - - if !status.is_success() { - tracing::warn!("[mcp-official] search HTTP {status} for key={cache_key}"); - anyhow::bail!( - "MCP official registry returned HTTP {status}: {}", - &body[..body.len().min(200)] - ); - } + let cursor_for_request = if page == 1 { + None + } else if let Some(cached) = cursor_cache_get(q, limit, page - 1) { + // Cache hit: we have the cursor that produced page-1, so one HTTP + // call gets us page. + Some(cached) + } else { + // Cache miss: walk forward from page 1 until we have a cursor for + // (page - 1). The walk also primes the cache so subsequent + // page+1/+2/... requests stay single-hop. + match walk_cursor_for_page(config, q, limit, page).await? { + Some(c) => Some(c), + None => { + // The walk ran out of results before reaching `page`. + // Return empty + report `page` so the UI stops paging. + tracing::debug!( + "[mcp-official] walk exhausted has_query={} target_page={page} limit={limit}", + !q.is_empty() + ); + return Ok((Vec::new(), page)); + } + } + }; + let body = fetch_page(q, limit, cursor_for_request.as_deref()).await?; let parsed: OfficialListResponse = serde_json::from_str(&body) .with_context(|| format!("Failed to parse MCP official response: {body}"))?; + let next_cursor = parsed.next_cursor().map(str::to_string); let summaries = parsed.into_summaries(); + + if let Some(ref c) = next_cursor { + cursor_cache_set(q, limit, page, c.clone()); + } let _ = store::set_cached(config, &cache_key, &body); tracing::debug!( - "[mcp-official] search ok servers={} (cursor pagination not wired)", - summaries.len() + "[mcp-official] search ok page={page} servers={} has_next={}", + summaries.len(), + next_cursor.is_some() ); - Ok((summaries, 1)) + Ok((summaries, total_pages_hint(page, next_cursor.is_some()))) } async fn get(&self, config: &Config, qualified_name: &str) -> Result { @@ -137,6 +211,147 @@ impl Registry for McpOfficialRegistry { } } +/// Fetch one page from the registry, optionally with a cursor. Returns the +/// raw response body so callers can both parse it and write it to the SQLite +/// response cache. +async fn fetch_page(q: &str, limit: u32, cursor: Option<&str>) -> Result { + // `q` is user-typed search input — log presence + length only so the + // diagnostic doesn't leak query text into log aggregators. + tracing::debug!( + "[mcp-official] fetch has_query={} q_len={} limit={limit} has_cursor={}", + !q.is_empty(), + q.len(), + cursor.is_some() + ); + + let client = http_client()?; + let url = format!("{}/v0/servers", base_url()); + let mut req = client.get(&url).header("Accept", "application/json"); + if !q.is_empty() { + req = req.query(&[("search", q)]); + } + req = req.query(&[("limit", &limit.to_string())]); + if let Some(c) = cursor { + req = req.query(&[("cursor", c)]); + } + req = apply_auth(req); + + let resp = req.send().await.context("MCP official search failed")?; + let status = resp.status(); + let body = resp.text().await.context("MCP official read failed")?; + + if !status.is_success() { + tracing::warn!("[mcp-official] search HTTP {status}"); + anyhow::bail!( + "MCP official registry returned HTTP {status}: {}", + &body[..body.len().min(200)] + ); + } + Ok(body) +} + +/// Walk the cursor chain forward starting from page 1 until we have the +/// cursor that, when sent with the next request, produces `target_page`. +/// +/// Returns `Some(cursor)` to feed into the request for `target_page`, or +/// `None` if the cursor chain ran out before reaching `target_page`. +/// +/// Bails after [`MAX_CURSOR_WALK_PAGES`] iterations to keep a single user +/// request from fanning into hundreds of upstream calls. +async fn walk_cursor_for_page( + config: &Config, + q: &str, + limit: u32, + target_page: u32, +) -> Result> { + if target_page <= 1 { + return Ok(None); + } + if target_page > MAX_CURSOR_WALK_PAGES { + tracing::warn!( + "[mcp-official] walk refused has_query={} target_page={target_page} max={MAX_CURSOR_WALK_PAGES}", + !q.is_empty() + ); + anyhow::bail!( + "MCP official deep-page walk refused: page={target_page} > MAX_CURSOR_WALK_PAGES={MAX_CURSOR_WALK_PAGES}" + ); + } + + tracing::debug!( + "[mcp-official] walk start has_query={} q_len={} target_page={target_page} limit={limit}", + !q.is_empty(), + q.len() + ); + + let mut cursor: Option = None; + let mut net_fetches = 0u32; + let mut cache_fetches = 0u32; + // We need the cursor that produces `target_page`, which is the cursor + // returned by the response for `target_page - 1`. + for page in 1..target_page { + let cache_key = format!("mcp_official:search:{q}:{page}:{limit}"); + + // Try the persisted SQLite response cache first. After a process + // restart the in-memory cursor map is empty, but page bodies from a + // previous run may still be on disk — using them shaves up to N-1 + // HTTP calls off a deep-link walk that has nothing to do with the + // network's current state. + let body = match store::get_cached(config, &cache_key) { + Ok(Some(body)) => { + cache_fetches += 1; + body + } + _ => { + let body = fetch_page(q, limit, cursor.as_deref()).await?; + let _ = store::set_cached(config, &cache_key, &body); + net_fetches += 1; + body + } + }; + + let parsed: OfficialListResponse = serde_json::from_str(&body) + .with_context(|| format!("Failed to parse MCP official response: {body}"))?; + let next = parsed.next_cursor().map(str::to_string); + + // Prime the in-memory cursor map as we go so a subsequent direct + // lookup for `page` doesn't have to re-walk. + if let Some(ref c) = next { + cursor_cache_set(q, limit, page, c.clone()); + } + + match next { + Some(c) => cursor = Some(c), + None => { + // Cursor chain exhausted before we reached target_page. + tracing::debug!( + "[mcp-official] walk done (exhausted) page={page} net={net_fetches} cache={cache_fetches}" + ); + return Ok(None); + } + } + } + tracing::debug!( + "[mcp-official] walk done (cursor ready) target_page={target_page} net={net_fetches} cache={cache_fetches}" + ); + Ok(cursor) +} + +/// `total_pages` reporting for the trait contract. +/// +/// `has_next` is the boolean derived from `metadata.nextCursor.is_some()` on +/// the current page's response. We can't know the *true* total without +/// walking the entire cursor chain (which is what the bug was originally +/// trying to avoid), so we report `page + 1` when more results exist — +/// matches the trait's "best-effort upper bound" contract and lets the UI +/// render a "next" affordance without overcommitting to a fixed total. +fn total_pages_hint(page: u32, has_next: bool) -> u32 { + if has_next { + page.saturating_add(1) + } else { + page + } +} + fn http_client() -> Result { Client::builder() .timeout(std::time::Duration::from_secs(15)) @@ -185,23 +400,73 @@ fn urlencoding_encode(s: &str) -> String { // // The official registry OpenAPI evolves; these are deliberately permissive // (every nested field is optional) so a schema bump doesn't break parsing. +// +// The real list response wraps each server as +// `{ "server": { ...inner... }, "_meta": { ... } }`. An earlier version of +// this adapter parsed the inner shape at the top level and so silently +// produced empty `OfficialServer` defaults at runtime — the test fixtures +// passed because they were built against the wrong shape too. The envelope +// here matches the actual wire payload (verified against +// `/v0/servers?limit=2` on `registry.modelcontextprotocol.io`). #[derive(Debug, Clone, Deserialize)] struct OfficialListResponse { #[serde(default)] - servers: Vec, + servers: Vec, #[serde(default)] - #[allow(dead_code)] - metadata: Option, + metadata: Option, } impl OfficialListResponse { fn into_summaries(self) -> Vec { - self.servers.into_iter().map(|s| s.into_summary()).collect() + self.servers + .into_iter() + .map(|env| env.server.into_summary()) + .collect() + } + + /// Cursor for the *next* page, if the registry indicates there's more. + /// `None` means the result set ends here. + fn next_cursor(&self) -> Option<&str> { + self.metadata + .as_ref() + .and_then(|m| m.next_cursor.as_deref()) + .filter(|s| !s.is_empty()) } } #[derive(Debug, Clone, Deserialize)] +struct OfficialMetadata { + #[serde(default, rename = "nextCursor")] + next_cursor: Option, + /// Server-reported count for the *current* page (not the total). Kept + /// for debug/observability; we don't use it to compute `total_pages`. + #[serde(default)] + #[allow(dead_code)] + count: Option, +} + +/// `{ "server": OfficialServer, "_meta": ... }` envelope. +/// +/// `server` is intentionally **not** `#[serde(default)]` — that's exactly +/// the failure mode the wrapper fix is closing out. If upstream ever +/// renames or omits the `server` key, deserialisation must surface as a +/// parse error so the broken wire shape is loud rather than silently +/// producing blank summary cards (the bug this PR was opened to fix). +/// +/// `_meta` carries registry-side fields (`status`, `publishedAt`, +/// `isLatest`); we don't need them for summary/detail rendering today, but +/// capturing the whole `Value` keeps the door open without another DTO +/// bump. +#[derive(Debug, Clone, Deserialize)] +struct OfficialServerEnvelope { + server: OfficialServer, + #[serde(default, rename = "_meta")] + #[allow(dead_code)] + meta: Option, +} + +#[derive(Debug, Clone, Default, Deserialize)] struct OfficialServer { /// Reverse-DNS-style identifier, e.g. `io.github.foo/server-bar`. #[serde(default)] @@ -300,5 +565,155 @@ mod tests { let raw = json!({ "servers": [] }); let parsed: OfficialListResponse = serde_json::from_value(raw).unwrap(); assert!(parsed.servers.is_empty()); + assert_eq!(parsed.next_cursor(), None); + } + + /// The earlier DTO parsed the *inner* shape at the top level, so a real + /// `{ "server": { ... } }` envelope deserialised into a default-empty + /// `OfficialServer` and silently produced blank summary cards in the UI. + /// This regression test pins the wrapper to the real wire shape. + #[test] + fn envelope_parses_wrapped_server_payload() { + let raw = json!({ + "servers": [ + { + "server": { + "name": "io.github.example/wrapped", + "description": "Wrapped server", + "remotes": [{ "url": "https://example.com/mcp" }], + }, + "_meta": { + "io.modelcontextprotocol.registry/official": { + "status": "active", + "isLatest": true + } + } + } + ], + "metadata": { "nextCursor": "tok-xyz", "count": 1 } + }); + let parsed: OfficialListResponse = serde_json::from_value(raw).unwrap(); + let summaries = parsed.into_summaries(); + assert_eq!(summaries.len(), 1); + assert_eq!(summaries[0].qualified_name, "io.github.example/wrapped"); + assert_eq!(summaries[0].description.as_deref(), Some("Wrapped server")); + // `_meta` is preserved as a raw `Value` — no panic on unknown keys. + } + + /// `metadata.nextCursor` drives both the cursor cache and the + /// `total_pages` hint. An empty string is treated as "no cursor" so a + /// future schema bump that stops omitting the field doesn't fool us + /// into walking forever. + #[test] + fn next_cursor_extraction_handles_missing_and_empty() { + let with_cursor: OfficialListResponse = + serde_json::from_value(json!({ "servers": [], "metadata": { "nextCursor": "abc" }})) + .unwrap(); + assert_eq!(with_cursor.next_cursor(), Some("abc")); + + let empty_cursor: OfficialListResponse = + serde_json::from_value(json!({ "servers": [], "metadata": { "nextCursor": "" }})) + .unwrap(); + assert_eq!(empty_cursor.next_cursor(), None); + + let no_meta: OfficialListResponse = + serde_json::from_value(json!({ "servers": [] })).unwrap(); + assert_eq!(no_meta.next_cursor(), None); + } + + /// Pins the trait-doc contract: report `page + 1` when more pages exist + /// so the UI renders "next", report `page` when the cursor chain ends + /// so the UI stops paging. Saturating_add guards against the (silly but + /// real) `page = u32::MAX` overflow case. + #[test] + fn total_pages_hint_reports_best_effort_upper_bound() { + assert_eq!(total_pages_hint(1, false), 1); + assert_eq!(total_pages_hint(1, true), 2); + assert_eq!(total_pages_hint(7, true), 8); + assert_eq!(total_pages_hint(7, false), 7); + assert_eq!(total_pages_hint(u32::MAX, true), u32::MAX); + } + + /// The cursor cache is a process-level singleton keyed by + /// `(query, page_size, page)`. Confirms reads see what writes wrote, + /// across queries / page_size partitions, and that the test-only + /// `cursor_cache_clear` actually drops entries. + #[test] + fn cursor_cache_round_trips_and_partitions_by_key() { + cursor_cache_clear(); + + cursor_cache_set("rust", 50, 1, "cur-rust-1".to_string()); + cursor_cache_set("rust", 50, 2, "cur-rust-2".to_string()); + cursor_cache_set("python", 50, 1, "cur-python-1".to_string()); + cursor_cache_set("rust", 25, 1, "cur-rust-25".to_string()); // different page_size + + assert_eq!( + cursor_cache_get("rust", 50, 1).as_deref(), + Some("cur-rust-1") + ); + assert_eq!( + cursor_cache_get("rust", 50, 2).as_deref(), + Some("cur-rust-2") + ); + assert_eq!( + cursor_cache_get("python", 50, 1).as_deref(), + Some("cur-python-1") + ); + assert_eq!( + cursor_cache_get("rust", 25, 1).as_deref(), + Some("cur-rust-25") + ); + // Unrelated key is empty. + assert_eq!(cursor_cache_get("rust", 50, 99), None); + assert_eq!(cursor_cache_get("ruby", 50, 1), None); + + cursor_cache_clear(); + assert_eq!(cursor_cache_get("rust", 50, 1), None); + } + + /// Bare-minimum DoS guard: the deep-page walk refuses to fan one user + /// request into hundreds of upstream calls. + #[tokio::test] + async fn walk_cursor_refuses_above_max_walk_pages() { + use crate::openhuman::config::Config; + let config = Config::default(); + let res = walk_cursor_for_page(&config, "anything", 50, MAX_CURSOR_WALK_PAGES + 1).await; + assert!(res.is_err(), "expected refusal above MAX_CURSOR_WALK_PAGES"); + let msg = format!("{:#}", res.unwrap_err()); + assert!( + msg.contains("MAX_CURSOR_WALK_PAGES"), + "error should name the limit: {msg}" + ); + } + + /// `server` is now required on the envelope. A payload that omits or + /// renames the `server` key must surface as a parse error — the exact + /// silent-empty-summary failure mode this whole PR was opened to fix. + /// Without this regression test, dropping `#[serde(default)]` on + /// `server` could quietly come back in a future "make it more + /// permissive" change. + #[test] + fn envelope_rejects_payload_missing_server_key() { + // The wrapper has `_meta` but no `server`. + let raw = json!({ + "servers": [ + { "_meta": { "io.modelcontextprotocol.registry/official": { "status": "active" } } } + ] + }); + let parsed = serde_json::from_value::(raw); + assert!( + parsed.is_err(), + "missing `server` key must be a parse error, not a silent default" + ); + + // And a renamed key ("srv") also fails — defends against an upstream + // schema rename quietly producing blank cards. + let renamed = json!({ + "servers": [{ "srv": { "name": "io.github.example/foo" } }] + }); + assert!( + serde_json::from_value::(renamed).is_err(), + "renamed `server` field must surface as parse error" + ); } }