From 91621372af40425b81fdc8ed13e33177f76e5533 Mon Sep 17 00:00:00 2001 From: Joe Savona Date: Fri, 13 Nov 2026 03:56:54 +0000 Subject: [PATCH] Add MCP inventory support for statusline providers (#61248) ## What changed Add a TUI module for reading MCP server identities and tool catalogs from the connected app server's global inventory. Support paginated discovery and targeted server lookup through `mcpServerStatus/list`. Retain plugin identities and tool errors, filter entries to those with an explicit null `httpOrigin`, and reject repeated pagination cursors or mismatched targeted responses. This adds inventory infrastructure without enabling provider discovery yet. ## Testing Add tests for global inventory independence from startup project overrides, preservation of tool catalogs and errors across pages, transport filtering, and rejection of a targeted response naming another server. GitOrigin-RevId: f429991deed38eb106641b1780f24c701626b4e1 --- codex-rs/tui/src/lib.rs | 1 + codex-rs/tui/src/statusline_provider.rs | 21 ++ .../tui/src/statusline_provider/inventory.rs | 148 ++++++++++++++ .../statusline_provider/inventory_tests.rs | 188 ++++++++++++++++++ .../src/statusline_provider/test_support.rs | 38 ++++ 5 files changed, 396 insertions(+) create mode 100644 codex-rs/tui/src/statusline_provider.rs create mode 100644 codex-rs/tui/src/statusline_provider/inventory.rs create mode 100644 codex-rs/tui/src/statusline_provider/inventory_tests.rs create mode 100644 codex-rs/tui/src/statusline_provider/test_support.rs diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index c3abb00935..c8ab4b69aa 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -229,6 +229,7 @@ mod startup_presentation; mod startup_recovery; mod status; mod status_indicator_widget; +mod statusline_provider; mod streaming; mod style; mod subscription; diff --git a/codex-rs/tui/src/statusline_provider.rs b/codex-rs/tui/src/statusline_provider.rs new file mode 100644 index 0000000000..1af7b9634f --- /dev/null +++ b/codex-rs/tui/src/statusline_provider.rs @@ -0,0 +1,21 @@ +//! MCP server identity and inventory for statusline providers. + +#[cfg_attr(not(test), expect(dead_code, reason = "staged provider discovery"))] +mod inventory; +#[cfg(test)] +mod test_support; + +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub(crate) struct McpServerName(String); + +impl McpServerName { + pub(super) fn as_str(&self) -> &str { + &self.0 + } +} + +impl From<&str> for McpServerName { + fn from(name: &str) -> Self { + Self(name.to_owned()) + } +} diff --git a/codex-rs/tui/src/statusline_provider/inventory.rs b/codex-rs/tui/src/statusline_provider/inventory.rs new file mode 100644 index 0000000000..498ed9d251 --- /dev/null +++ b/codex-rs/tui/src/statusline_provider/inventory.rs @@ -0,0 +1,148 @@ +//! Read MCP tool catalogs from the connected app server’s global inventory. + +use super::McpServerName; +use codex_app_server_client::AppServerRequestHandle; +use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::ListMcpServerStatusParams; +use codex_app_server_protocol::McpServerConnectionStatus; +use codex_app_server_protocol::McpServerStatus; +use codex_app_server_protocol::McpServerStatusDetail; +use codex_app_server_protocol::RequestId; +use codex_plugin::PluginId; +use codex_plugin::PluginIdError; +use codex_protocol::mcp::Tool; +use color_eyre::Result; +use color_eyre::eyre::bail; +use futures::Stream; +use futures::stream; +use serde::Deserialize; +use serde_json::Value; +use std::collections::HashMap; +use std::collections::HashSet; +use uuid::Uuid; + +#[derive(Debug, PartialEq)] +pub(super) struct StatuslineServer { + pub(super) name: McpServerName, + pub(super) runtime_status: Option, + pub(super) plugin_id: Option, + pub(super) tools: Result, String>, +} + +impl TryFrom for StatuslineServer { + type Error = PluginIdError; + + fn try_from(row: McpServerStatus) -> Result { + let tools = match row.tools_error { + Some(error) => Err(error), + None if row.runtime_status == Some(McpServerConnectionStatus::Failed) => { + Err(format!("MCP server {} is not connected", row.name)) + } + None => Ok(row.tools), + }; + Ok(Self { + name: McpServerName(row.name), + runtime_status: row.runtime_status, + plugin_id: row.plugin_id.as_deref().map(PluginId::parse).transpose()?, + tools, + }) + } +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct InventoryPage { + data: Vec, + next_cursor: Option, +} + +async fn read_page( + client: &AppServerRequestHandle, + params: ListMcpServerStatusParams, +) -> Result { + Ok(client + .request_typed(ClientRequest::McpServerStatusList { + request_id: RequestId::String(Uuid::new_v4().to_string()), + params, + }) + .await?) +} + +/// Request the whole inventory. Some hosts paginate by offset into a newly rebuilt catalog. +pub(super) fn global_pages( + client: &AppServerRequestHandle, +) -> impl Stream>> + '_ { + stream::try_unfold(Some((None, HashSet::new())), move |next| async move { + let Some((cursor, mut seen)) = next else { + return Ok(None); + }; + if let Some(cursor) = &cursor + && !seen.insert(String::clone(cursor)) + { + bail!("MCP inventory repeated a cursor"); + } + let page = read_page( + client, + ListMcpServerStatusParams { + thread_id: None, + server_name: None, + detail: Some(McpServerStatusDetail::ToolsAndAuthOnly), + cursor, + limit: None, + }, + ) + .await?; + let mut servers = Vec::new(); + for value in page.data { + if let Some(row) = local_server(value)? { + servers.push(row); + } + } + Ok(Some(( + servers, + page.next_cursor.map(|next| (Some(next), seen)), + ))) + }) +} + +fn local_server(value: Value) -> Result> { + // Stdio servers are eligible. The app server sets `httpOrigin` to null for stdio + // and for invalid or unsupported HTTP URLs. Skip entries from older hosts that + // omit the field because their transport is unknown. + if value.get("httpOrigin") != Some(&Value::Null) { + return Ok(None); + } + let status = serde_json::from_value::(value)?; + Ok(StatuslineServer::try_from(status).ok()) +} + +pub(super) async fn server( + client: &AppServerRequestHandle, + name: &McpServerName, +) -> Result> { + let mut page = read_page( + client, + ListMcpServerStatusParams { + thread_id: None, + server_name: Some(name.0.clone()), + detail: Some(McpServerStatusDetail::ToolsAndAuthOnly), + cursor: None, + limit: None, + }, + ) + .await?; + if page.next_cursor.is_some() || page.data.len() > 1 { + bail!("targeted MCP inventory returned more than one server") + } + let Some(value) = page.data.pop() else { + return Ok(None); + }; + if value["name"] != name.as_str() { + bail!("targeted MCP inventory did not identify the requested server") + } + local_server(value) +} + +#[cfg(test)] +#[path = "inventory_tests.rs"] +mod tests; diff --git a/codex-rs/tui/src/statusline_provider/inventory_tests.rs b/codex-rs/tui/src/statusline_provider/inventory_tests.rs new file mode 100644 index 0000000000..a94ecc5268 --- /dev/null +++ b/codex-rs/tui/src/statusline_provider/inventory_tests.rs @@ -0,0 +1,188 @@ +#![cfg(test)] +//! Verify global inventory against the connected app server and its configuration. + +use super::super::test_support::reply; +use super::super::test_support::row; +use super::super::test_support::test_client; +use super::*; +use crate::AppServerTarget; +use crate::backend_id::BackendId; +use crate::legacy_core::config::ConfigBuilder; +use crate::legacy_core::config::ConfigOverrides; +use codex_config::LoaderOverrides; +use codex_exec_server::EnvironmentManager; +use futures::TryStreamExt; +use pretty_assertions::assert_eq; +use serde_json::json; +use std::sync::Arc; + +#[tokio::test] +async fn global_inventory_ignores_the_startup_projects_overrides() -> Result<()> { + let temp = tempfile::tempdir()?; + let home = temp.path().join("home"); + let project = temp.path().join("project"); + std::fs::create_dir_all(&home)?; + std::fs::create_dir_all(project.join(".git"))?; + std::fs::write(project.join(".git/HEAD"), "ref: refs/heads/main\n")?; + std::fs::create_dir(project.join(".codex"))?; + let package = home.join("plugins/cache/local/daily/local"); + std::fs::create_dir_all(package.join(".codex-plugin"))?; + std::fs::write( + package.join(".codex-plugin/plugin.json"), + r#"{"name":"daily"}"#, + )?; + let missing = temp.path().join("missing-mcp-fixture"); + std::fs::write( + package.join(".mcp.json"), + json!({"mcpServers":{ + "implicit-calendar":{"command":missing} + }}) + .to_string(), + )?; + let command = toml::Value::String(missing.to_string_lossy().into_owned()); + let replacement = toml::Value::String( + temp.path() + .join("missing-project-fixture") + .to_string_lossy() + .into_owned(), + ); + let trusted = toml::Value::String(project.to_string_lossy().into_owned()); + std::fs::write( + home.join("config.toml"), + format!( + r#" +[projects.{trusted}] +trust_level = "trusted" +[features] +plugins = true +remote_plugin = false +[plugins."daily@local"] +enabled = true +[mcp_servers.calendar] +command = {command} +"# + ), + )?; + std::fs::write( + project.join(".codex/config.toml"), + format!( + r#" +[mcp_servers.calendar] +command = {replacement} +[mcp_servers.project_only] +command = {replacement} +[mcp_servers.implicit-calendar] +command = {replacement} +"# + ), + )?; + let config = ConfigBuilder::default() + .codex_home(home.clone()) + .harness_overrides(ConfigOverrides { + cwd: Some(project.clone()), + ..Default::default() + }) + .build() + .await?; + assert!(config.mcp_servers.get().contains_key("project_only")); + let session = Box::pin(crate::start_app_server_for_picker( + BackendId::new(), + &config, + &AppServerTarget::Embedded, + vec![("mcp_servers.cli-weather.command".into(), command)], + LoaderOverrides::without_managed_config_for_tests(), + /*state_db*/ None, + Arc::new(EnvironmentManager::default_for_tests()), + )) + .await?; + let client = session.request_handle(); + let servers = global_pages(&client).try_concat().await?; + let daily = PluginId::parse("daily@local")?; + let identities = servers + .iter() + .map(|s| (s.name.as_str(), s.plugin_id.as_ref())) + .collect::>(); + assert_eq!( + identities, + [ + ("calendar", None), + ("cli-weather", None), + ("implicit-calendar", Some(&daily)) + ] + ); + session.shutdown().await?; + Ok(()) +} + +#[tokio::test] +async fn global_pages_preserve_catalogs_and_skip_hosted_or_unidentified_transports() -> Result<()> { + let (client, mut commands) = test_client(); + let host = async { + let tool = json!({"name":"forecast","inputSchema":{}}); + let direct = row("cli-weather", vec![tool.clone()]); + let mut plugin = row("planner", vec![tool]); + plugin["pluginId"] = json!("work@local"); + let mut hosted = row("hosted", vec![]); + hosted["httpOrigin"] = json!("https://example.test"); + let mut older = row("missing-origin", vec![]); + older.as_object_mut().unwrap().remove("httpOrigin"); + let mut failed = row("failed", vec![]); + failed["runtimeStatus"] = Value::Null; + failed["toolsError"] = json!("connection refused"); + for (cursor, rows, next) in [ + (None, vec![direct, hosted, older], Some("next")), + (Some("next"), vec![plugin, failed], None), + ] { + reply( + &mut commands, + "mcpServerStatus/list", + json!({"threadId":null,"serverName":null,"detail":"toolsAndAuthOnly","cursor":cursor,"limit":null}), + json!({"data":rows,"nextCursor":next}), + ).await; + } + }; + let (servers, ()) = tokio::join!(global_pages(&client).try_concat(), host); + let catalogs = servers? + .into_iter() + .map(|s| { + ( + s.name, + s.plugin_id, + s.tools.map(|tools| tools.into_keys().collect::>()), + ) + }) + .collect::>(); + assert_eq!( + catalogs, + [ + ("cli-weather".into(), None, Ok(vec!["forecast".to_owned()])), + ( + "planner".into(), + Some(PluginId::parse("work@local")?), + Ok(vec!["forecast".to_owned()]) + ), + ("failed".into(), None, Err("connection refused".to_owned())), + ] + ); + Ok(()) +} + +#[tokio::test] +async fn targeted_inventory_rejects_another_server() { + let (client, mut commands) = test_client(); + let name = "calendar".into(); + let other = row("other", vec![json!({"name":"footer","inputSchema":{}})]); + let (result, ()) = tokio::join!( + server(&client, &name), + reply( + &mut commands, + "mcpServerStatus/list", + json!({"threadId":null,"serverName":"calendar","detail":"toolsAndAuthOnly","cursor":null,"limit":null}), + json!({"data":[other],"nextCursor":null}), + ) + ); + assert_eq!( + result.unwrap_err().to_string(), + "targeted MCP inventory did not identify the requested server" + ); +} diff --git a/codex-rs/tui/src/statusline_provider/test_support.rs b/codex-rs/tui/src/statusline_provider/test_support.rs new file mode 100644 index 0000000000..4a4c28a882 --- /dev/null +++ b/codex-rs/tui/src/statusline_provider/test_support.rs @@ -0,0 +1,38 @@ +#![cfg(test)] +//! Drive statusline RPCs through the same request handle a remote TUI uses. + +use codex_app_server_client::AppServerClient; +use codex_app_server_client::AppServerRequestHandle; +use codex_app_server_client::CloudClient; +use codex_app_server_client::CloudCommand; +use codex_app_server_client::CloudOperation; +use pretty_assertions::assert_eq; +use serde_json::Value; +use serde_json::json; +use tokio::sync::mpsc; + +pub(super) type Requests = mpsc::Receiver; + +pub(super) fn test_client() -> (AppServerRequestHandle, Requests) { + let (cloud, commands, _events) = CloudClient::channel(/*capacity*/ 8); + (AppServerClient::Cloud(cloud).request_handle(), commands) +} + +pub(super) async fn reply(commands: &mut Requests, method: &str, params: Value, result: Value) { + let command = commands.recv().await.unwrap(); + let CloudOperation::Request(request) = command.operation else { + panic!("expected {method}"); + }; + let wire = serde_json::to_value(request).unwrap(); + assert_eq!(wire["method"], method); + assert_eq!(wire["params"], params); + command.response.send(Ok(Ok(result))).unwrap(); +} + +pub(super) fn row(name: &str, tools: Vec) -> Value { + let tools = tools + .into_iter() + .map(|tool| (tool["name"].as_str().unwrap().to_owned(), tool)) + .collect::>(); + json!({"name":name,"runtimeStatus":"connected","httpOrigin":null,"tools":tools,"authStatus":"unsupported","resources":[],"resourceTemplates":[]}) +}