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

@@ -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.
 

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 {

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<rmcp::service::RoleServer>,
+    ) -> Result<rmcp::model::CallToolResponse, rmcp::ErrorData> {
+        Ok(
+            CallToolResult::success(vec![ContentBlock::text(format!("called {}", request.name))])
+                .into(),
+        )
+    }
+
     fn list_tools(
         &self,
         _request: Option<rmcp::model::PaginatedRequestParams>,