From 2cfee7de25d98b20c55ced2ed82736d4c452f8a0 Mon Sep 17 00:00:00 2001 From: Alex Daley Date: Sat, 5 Sep 2026 15:20:07 +0000 Subject: [PATCH] Refresh live thread tools through `app/installed` (#43039) ## Why Calling `app/installed` with `threadId` and `forceRefresh: true` refreshed a separate runtime snapshot without updating the thread's tools for subsequent turns. ## What changed Use the thread's current configuration and refresh its live app tools when a thread is specified. Account for the refreshed snapshot's model-visible tools when reporting `callable`. Requests without `threadId` continue to use a separate runtime for refreshes. ## Testing Add integration coverage verifying that refreshed tools replace previous tools in subsequent model requests and can be called, and that a failed refresh preserves the last working tools. Extend thread configuration coverage to include forced refreshes. GitOrigin-RevId: fc13e28a59da00af470cf25c0cde356504e1575b --- codex-rs/app-server/README.md | 2 +- .../apps_processor/installed.rs | 59 ++++-- .../tests/suite/v2/app_installed.rs | 175 +++++++++++++++++- 3 files changed, 208 insertions(+), 28 deletions(-) diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 13468680e7..4fc9588ab2 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -2395,7 +2395,7 @@ Use `app/installed` to read installed apps and whether each app is currently ena `id` is the app's connector ID, and `runtimeName` is the nullable name reported by the runtime. `enabled` reflects effective app configuration and workspace policy. `callable` is true when the app is enabled and has at least one model-visible tool allowed by app and tool policy. -When `threadId` is provided, the response uses that thread's effective configuration; otherwise it uses the current global configuration. `forceRefresh` defaults to `false`. Set it to `true` to refresh the hosted connector runtime tool snapshot before reading the response. When Apps are disabled by global or workspace policy, previously observed apps may still be returned with `enabled` and `callable` set to `false`. +When `threadId` is provided, the response uses that thread's current configuration; otherwise it uses the current global configuration. `forceRefresh` defaults to `false`. Set it to `true` to refresh app tools before returning the response. With `threadId`, subsequent turns can use the refreshed tools. When Apps are disabled by the effective configuration or workspace policy, previously observed apps may still be returned with `enabled` and `callable` set to `false`. Use `app/list` to fetch available apps (connectors). Each entry includes metadata like the app `id`, display `name`, `installUrl`, legacy logo URLs, structured light and dark icon assets, `branding`, `appMetadata`, `labels`, whether it is currently accessible, and whether it is enabled in config. diff --git a/codex-rs/app-server/src/request_processors/apps_processor/installed.rs b/codex-rs/app-server/src/request_processors/apps_processor/installed.rs index 8b2ccbcd4d..00f4f86c49 100644 --- a/codex-rs/app-server/src/request_processors/apps_processor/installed.rs +++ b/codex-rs/app-server/src/request_processors/apps_processor/installed.rs @@ -47,25 +47,40 @@ impl AppsRequestProcessor { let mut snapshot_age = None; let mut snapshot_tool_count = 0; let result = async { - let config = self - .load_apps_config(params.thread_id.as_deref()) - .await?; + let (config, thread) = match params.thread_id.as_deref() { + Some(thread_id) => { + let (_, thread) = self.load_thread(thread_id).await?; + let config = thread.config().await; + (config.as_ref().clone(), Some(thread)) + } + None => ( + self.load_latest_config(/*fallback_cwd*/ None).await?, + None, + ), + }; let auth = self.auth_manager.auth().await; let runtime_enabled = config .features .apps_enabled_for_auth(auth.as_ref().is_some_and(CodexAuth::uses_codex_backend)); let mcp_manager = self.thread_manager.mcp_manager(); - let mcp_config = mcp_manager.runtime_config(&config).await; - let mut mcp_servers = effective_mcp_servers(&mcp_config, auth.as_ref()); - mcp_servers.retain(|name, _| name == CODEX_APPS_MCP_SERVER_NAME); - let mcp_config = Arc::new(mcp_config.for_threadless_operations(&mcp_servers)); let cache_key = connector_runtime_context_key(auth.as_ref()); let previous_snapshot = mcp_manager .codex_apps_tools_cache() .current_snapshot(config.codex_home.to_path_buf(), cache_key.clone()); - let snapshot = if force_refresh && runtime_enabled { + let mut model_visible_tool_names = None; + let tools = if force_refresh && runtime_enabled { let refresh_result = async { + if let Some(thread) = thread { + let snapshot = thread.refresh_codex_apps_tools().await?; + model_visible_tool_names = Some(snapshot.model_visible_tool_names); + snapshot_age = Some(Duration::ZERO); + return Ok(snapshot.tools); + } + let mcp_config = mcp_manager.runtime_config(&config).await; + let mut mcp_servers = effective_mcp_servers(&mcp_config, auth.as_ref()); + mcp_servers.retain(|name, _| name == CODEX_APPS_MCP_SERVER_NAME); + let mcp_config = Arc::new(mcp_config.for_threadless_operations(&mcp_servers)); anyhow::ensure!( !mcp_servers.is_empty(), "host-owned MCP server '{CODEX_APPS_MCP_SERVER_NAME}' is not enabled" @@ -125,14 +140,17 @@ impl AppsRequestProcessor { }; cancellation_token.cancel(); runtime.shutdown().await; - result + result.map(|snapshot| { + snapshot_age = Some(snapshot.age()); + snapshot.tools().to_vec() + }) } .await; match refresh_result { - Ok(snapshot) => { + Ok(tools) => { refresh_disposition = "success"; - Some(snapshot) + tools } Err(err) => { refresh_disposition = "error"; @@ -147,17 +165,22 @@ impl AppsRequestProcessor { refresh_disposition = "skipped_apps_disabled"; retained_previous_snapshot = previous_snapshot.is_some(); } - previous_snapshot - }; - let Some(snapshot) = snapshot else { - return Ok(AppsInstalledResponse { apps: Vec::new() }); + previous_snapshot.map_or_else(Vec::new, |snapshot| { + snapshot_age = Some(snapshot.age()); + snapshot.tools().to_vec() + }) }; - snapshot_age = Some(snapshot.age()); - snapshot_tool_count = snapshot.tools().len(); + snapshot_tool_count = tools.len(); let apps = installed_connector_runtime( &config.config_layer_stack, - snapshot.tools().iter().map(connector_runtime_tool), + tools.iter().map(|tool| { + let mut runtime_tool = connector_runtime_tool(tool); + if let Some(names) = &model_visible_tool_names { + runtime_tool.model_visible &= names.contains(tool.tool.name.as_ref()); + } + runtime_tool + }), ) .into_iter() .map(|app| InstalledApp { diff --git a/codex-rs/app-server/tests/suite/v2/app_installed.rs b/codex-rs/app-server/tests/suite/v2/app_installed.rs index 4c473fefa6..098c50d79b 100644 --- a/codex-rs/app-server/tests/suite/v2/app_installed.rs +++ b/codex-rs/app-server/tests/suite/v2/app_installed.rs @@ -9,6 +9,7 @@ use std::time::Duration; use anyhow::Result; use app_test_support::ChatGptAuthFixture; +use app_test_support::MockResponsesConfig; use app_test_support::TestAppServer; use app_test_support::write_chatgpt_auth; use axum::Json; @@ -22,9 +23,17 @@ use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStatus; +use codex_app_server_protocol::UserInput; use codex_config::types::AuthCredentialsStoreMode; +use codex_features::Feature; +use core_test_support::responses; use pretty_assertions::assert_eq; use rmcp::handler::server::ServerHandler; +use rmcp::model::CallToolRequestParams; +use rmcp::model::CallToolResult; +use rmcp::model::ContentBlock; use rmcp::model::ListToolsResult; use rmcp::model::ServerCapabilities; use rmcp::model::ServerInfo; @@ -141,14 +150,6 @@ async fn installed_apps_thread_id_uses_effective_thread_config() -> Result<()> { let ThreadStartResponse { thread, .. } = timeout(DEFAULT_TIMEOUT, app_server.read_response(request_id)).await??; - let request_id = app_server - .send_apps_installed_request(AppsInstalledParams { - thread_id: Some(thread.id), - force_refresh: false, - }) - .await?; - let response: AppsInstalledResponse = - timeout(DEFAULT_TIMEOUT, app_server.read_response(request_id)).await??; let alpha = expected .apps .iter_mut() @@ -156,8 +157,153 @@ async fn installed_apps_thread_id_uses_effective_thread_config() -> Result<()> { .expect("alpha app should be installed"); alpha.enabled = false; alpha.callable = false; - assert_eq!(response, expected); + for force_refresh in [false, true] { + let request_id = app_server + .send_apps_installed_request(AppsInstalledParams { + thread_id: Some(thread.id.clone()), + force_refresh, + }) + .await?; + let response: AppsInstalledResponse = + timeout(DEFAULT_TIMEOUT, app_server.read_response(request_id)).await??; + assert_eq!(response, expected); + } + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn installed_apps_thread_refresh_updates_live_tools_and_retains_them_on_failure() -> Result<()> +{ + let fixture = InstalledAppsFixture::start().await?; + fixture.set_tools(vec![connector_tool("alpha", "Alpha")?]); + let responses_server = responses::start_mock_server().await; + let codex_home = configured_codex_home(fixture.base_url())?; + MockResponsesConfig::new(&responses_server.uri()) + .with_root_config(&format!( + "chatgpt_base_url = {:?}\nmcp_oauth_credentials_store = \"file\"", + fixture.base_url() + )) + .enable_feature(Feature::Apps) + .write(codex_home.path())?; + let mut app_server = start_app_server(codex_home.path()).await?; + let ThreadStartResponse { thread, .. } = app_server + .start_thread(ThreadStartParams::default()) + .await?; + + let initial_model_request = responses::mount_sse_once( + &responses_server, + responses::sse(vec![responses::ev_completed("before-refresh")]), + ) + .await; + let completed = timeout( + DEFAULT_TIMEOUT, + app_server.start_turn_and_wait_for_completion(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "Which tools are available?".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }), + ) + .await??; + assert_eq!(completed.turn.status, TurnStatus::Completed); + assert!( + initial_model_request + .single_request() + .tool_by_name("mcp__codex_apps__alpha", "connector_alpha") + .is_some() + ); + + fixture.set_tools(vec![connector_tool("beta", "Beta")?]); + let expected = AppsInstalledResponse { + apps: vec![InstalledApp { + id: "beta".to_string(), + runtime_name: Some("Beta".to_string()), + enabled: true, + callable: true, + }], + }; + + for (call_id, refresh_fails) in [("after-refresh", false), ("after-failed-refresh", true)] { + if refresh_fails { + fixture.fail_next_list_tools(); + } + let list_tools_calls = fixture.list_tools_calls(); + let mut request_id = app_server + .send_apps_installed_request(AppsInstalledParams { + thread_id: Some(thread.id.clone()), + force_refresh: true, + }) + .await?; + if refresh_fails { + let error = timeout( + DEFAULT_TIMEOUT, + app_server.read_stream_until_error_message(RequestId::Integer(request_id)), + ) + .await??; + assert_eq!(error.error.code, -32603); + request_id = app_server + .send_apps_installed_request(AppsInstalledParams { + thread_id: Some(thread.id.clone()), + force_refresh: false, + }) + .await?; + } + let installed: AppsInstalledResponse = + timeout(DEFAULT_TIMEOUT, app_server.read_response(request_id)).await??; + assert_eq!(installed, expected); + assert_eq!(fixture.list_tools_calls(), list_tools_calls + 1); + + let model_requests = responses::mount_sse_sequence( + &responses_server, + vec![ + responses::sse(vec![ + responses::ev_function_call_with_namespace( + call_id, + "mcp__codex_apps__beta", + "connector_beta", + "{}", + ), + responses::ev_completed(call_id), + ]), + responses::sse(vec![responses::ev_completed("done")]), + ], + ) + .await; + let completed = timeout( + DEFAULT_TIMEOUT, + app_server.start_turn_and_wait_for_completion(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "Call Beta.".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }), + ) + .await??; + assert_eq!(completed.turn.status, TurnStatus::Completed); + let requests = model_requests.requests(); + assert_eq!(requests.len(), 2); + assert!( + requests[0] + .tool_by_name("mcp__codex_apps__beta", "connector_beta") + .is_some() + ); + assert!( + requests[0] + .tool_by_name("mcp__codex_apps__alpha", "connector_alpha") + .is_none() + ); + assert_eq!( + requests[1].function_call_output(call_id)["output"][1], + json!({"type": "input_text", "text": "called connector_beta"}) + ); + assert_eq!(fixture.list_tools_calls(), list_tools_calls + 1); + } Ok(()) } @@ -251,6 +397,17 @@ impl ServerHandler for InstalledAppsMcpServer { ServerInfo::new(ServerCapabilities::builder().enable_tools().build()) } + async fn call_tool( + &self, + request: CallToolRequestParams, + _context: rmcp::service::RequestContext, + ) -> Result { + Ok( + CallToolResult::success(vec![ContentBlock::text(format!("called {}", request.name))]) + .into(), + ) + } + fn list_tools( &self, _request: Option,