Merge e1f966b3ba into sapling-pr-archive-bolinfest

codex-rs/app-server-protocol/schema/json/ClientRequest.json

@@ -2292,12 +2292,6 @@
               },
               "type": "array"
             },
-            "end_turn": {
-              "type": [
-                "boolean",
-                "null"
-              ]
-            },
             "id": {
               "type": [
                 "string",
@@ -6130,4 +6124,4 @@
     }
   ],
   "title": "ClientRequest"
-}
+}

codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json

@@ -12739,12 +12739,6 @@
                 },
                 "type": "array"
               },
-              "end_turn": {
-                "type": [
-                  "boolean",
-                  "null"
-                ]
-              },
               "id": {
                 "type": [
                   "string",
@@ -17758,4 +17752,4 @@
   },
   "title": "CodexAppServerProtocol",
   "type": "object"
-}
+}

codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json

@@ -9413,12 +9413,6 @@
               },
               "type": "array"
             },
-            "end_turn": {
-              "type": [
-                "boolean",
-                "null"
-              ]
-            },
             "id": {
               "type": [
                 "string",
@@ -15643,4 +15637,4 @@
   },
   "title": "CodexAppServerProtocolV2",
   "type": "object"
-}
+}

codex-rs/app-server-protocol/schema/json/v2/RawResponseItemCompletedNotification.json

@@ -345,12 +345,6 @@
               },
               "type": "array"
             },
-            "end_turn": {
-              "type": [
-                "boolean",
-                "null"
-              ]
-            },
             "id": {
               "type": [
                 "string",
@@ -928,4 +922,4 @@
   ],
   "title": "RawResponseItemCompletedNotification",
   "type": "object"
-}
+}

codex-rs/app-server-protocol/schema/json/v2/ThreadResumeParams.json

@@ -756,12 +756,6 @@
               },
               "type": "array"
             },
-            "end_turn": {
-              "type": [
-                "boolean",
-                "null"
-              ]
-            },
             "id": {
               "type": [
                 "string",
@@ -1458,4 +1452,4 @@
   ],
   "title": "ThreadResumeParams",
   "type": "object"
-}
+}

codex-rs/app-server-protocol/schema/typescript/ResponseItem.ts

@@ -11,7 +11,7 @@ import type { ReasoningItemContent } from "./ReasoningItemContent";
 import type { ReasoningItemReasoningSummary } from "./ReasoningItemReasoningSummary";
 import type { WebSearchAction } from "./WebSearchAction";
 
-export type ResponseItem = { "type": "message", role: string, content: Array<ContentItem>, end_turn?: boolean, phase?: MessagePhase, } | { "type": "reasoning", summary: Array<ReasoningItemReasoningSummary>, content?: Array<ReasoningItemContent>, encrypted_content: string | null, } | { "type": "local_shell_call",
+export type ResponseItem = { "type": "message", role: string, content: Array<ContentItem>, phase?: MessagePhase, } | { "type": "reasoning", summary: Array<ReasoningItemReasoningSummary>, content?: Array<ReasoningItemContent>, encrypted_content: string | null, } | { "type": "local_shell_call",
 /**
  * Set when using the Responses API.
  */

codex-rs/app-server-protocol/src/protocol/thread_history.rs

@@ -3096,7 +3096,6 @@ mod tests {
                 content: vec![codex_protocol::models::ContentItem::InputText {
                     text: "plain text".into(),
                 }],
-                end_turn: None,
                 phase: None,
             }),
             RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {

codex-rs/app-server/tests/suite/v2/compaction.rs

@@ -134,7 +134,6 @@ async fn auto_compaction_remote_emits_started_and_completed_items() -> Result<()
             content: vec![ContentItem::OutputText {
                 text: "REMOTE_COMPACT_SUMMARY".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Compaction {

codex-rs/app-server/tests/suite/v2/thread_inject_items.rs

@@ -59,7 +59,6 @@ async fn thread_inject_items_adds_raw_response_items_to_thread_history() -> Resu
         content: vec![ContentItem::OutputText {
             text: injected_text.to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
 
@@ -195,7 +194,6 @@ async fn thread_inject_items_adds_raw_response_items_after_a_turn() -> Result<()
         content: vec![ContentItem::OutputText {
             text: "Injected after first turn".to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
     let injected_value = serde_json::to_value(&injected_item)?;

codex-rs/app-server/tests/suite/v2/thread_resume.rs

@@ -1660,7 +1660,6 @@ async fn thread_resume_rejects_history_when_thread_is_running() -> Result<()> {
                 content: vec![ContentItem::InputText {
                     text: "history override".to_string(),
                 }],
-                end_turn: None,
                 phase: None,
             }]),
             ..Default::default()
@@ -2616,7 +2615,6 @@ async fn thread_resume_supports_history_and_overrides() -> Result<()> {
         content: vec![ContentItem::InputText {
             text: history_text.to_string(),
         }],
-        end_turn: None,
         phase: None,
     }];
 

codex-rs/codex-api/tests/clients.rs

@@ -423,7 +423,6 @@ async fn azure_default_store_attaches_ids_and_headers() -> Result<()> {
             id: Some("msg_1".into()),
             role: "user".into(),
             content: vec![ContentItem::InputText { text: "hi".into() }],
-            end_turn: None,
             phase: None,
         }],
         tools: Vec::new(),

codex-rs/codex-mcp/src/codex_apps.rs

@@ -0,0 +1,258 @@
+//! Codex Apps support for the built-in apps MCP server.
+//!
+//! This module owns the pieces that are unique to ChatGPT-hosted app
+//! connectors: cache scoping by authenticated user, disk cache reads/writes,
+//! connector allow-list filtering, and the normalization that turns app
+//! connector/tool metadata into model-visible MCP callable names.
+
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::time::Instant;
+
+use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
+use crate::runtime::emit_duration;
+use crate::tools::MCP_TOOLS_CACHE_WRITE_DURATION_METRIC;
+use crate::tools::ToolInfo;
+use codex_login::CodexAuth;
+use codex_utils_plugins::mcp_connector::is_connector_id_allowed;
+use codex_utils_plugins::mcp_connector::sanitize_name;
+use serde::Deserialize;
+use serde::Serialize;
+use sha1::Digest;
+use sha1::Sha1;
+
+pub(crate) const CODEX_APPS_TOOLS_CACHE_SCHEMA_VERSION: u8 = 2;
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CodexAppsToolsCacheKey {
+    pub(crate) account_id: Option<String>,
+    pub(crate) chatgpt_user_id: Option<String>,
+    pub(crate) is_workspace_account: bool,
+}
+
+pub fn codex_apps_tools_cache_key(auth: Option<&CodexAuth>) -> CodexAppsToolsCacheKey {
+    CodexAppsToolsCacheKey {
+        account_id: auth.and_then(CodexAuth::get_account_id),
+        chatgpt_user_id: auth.and_then(CodexAuth::get_chatgpt_user_id),
+        is_workspace_account: auth.is_some_and(CodexAuth::is_workspace_account),
+    }
+}
+
+pub fn filter_non_codex_apps_mcp_tools_only(
+    mcp_tools: &HashMap<String, ToolInfo>,
+) -> HashMap<String, ToolInfo> {
+    mcp_tools
+        .iter()
+        .filter(|(_, tool)| tool.server_name != CODEX_APPS_MCP_SERVER_NAME)
+        .map(|(name, tool)| (name.clone(), tool.clone()))
+        .collect()
+}
+
+#[derive(Clone)]
+pub(crate) struct CodexAppsToolsCacheContext {
+    pub(crate) codex_home: PathBuf,
+    pub(crate) user_key: CodexAppsToolsCacheKey,
+}
+
+impl CodexAppsToolsCacheContext {
+    pub(crate) fn cache_path(&self) -> PathBuf {
+        let user_key_json = serde_json::to_string(&self.user_key).unwrap_or_default();
+        let user_key_hash = sha1_hex(&user_key_json);
+        self.codex_home
+            .join(CODEX_APPS_TOOLS_CACHE_DIR)
+            .join(format!("{user_key_hash}.json"))
+    }
+}
+
+pub(crate) enum CachedCodexAppsToolsLoad {
+    Hit(Vec<ToolInfo>),
+    Missing,
+    Invalid,
+}
+
+pub(crate) fn normalize_codex_apps_tool_title(
+    server_name: &str,
+    connector_name: Option<&str>,
+    value: &str,
+) -> String {
+    if server_name != CODEX_APPS_MCP_SERVER_NAME {
+        return value.to_string();
+    }
+
+    let Some(connector_name) = connector_name
+        .map(str::trim)
+        .filter(|name| !name.is_empty())
+    else {
+        return value.to_string();
+    };
+
+    let prefix = format!("{connector_name}_");
+    if let Some(stripped) = value.strip_prefix(&prefix)
+        && !stripped.is_empty()
+    {
+        return stripped.to_string();
+    }
+
+    value.to_string()
+}
+
+pub(crate) fn normalize_codex_apps_callable_name(
+    server_name: &str,
+    tool_name: &str,
+    connector_id: Option<&str>,
+    connector_name: Option<&str>,
+) -> String {
+    if server_name != CODEX_APPS_MCP_SERVER_NAME {
+        return tool_name.to_string();
+    }
+
+    let tool_name = sanitize_name(tool_name);
+
+    if let Some(connector_name) = connector_name
+        .map(str::trim)
+        .map(sanitize_name)
+        .filter(|name| !name.is_empty())
+        && let Some(stripped) = tool_name.strip_prefix(&connector_name)
+        && !stripped.is_empty()
+    {
+        return stripped.to_string();
+    }
+
+    if let Some(connector_id) = connector_id
+        .map(str::trim)
+        .map(sanitize_name)
+        .filter(|name| !name.is_empty())
+        && let Some(stripped) = tool_name.strip_prefix(&connector_id)
+        && !stripped.is_empty()
+    {
+        return stripped.to_string();
+    }
+
+    tool_name
+}
+
+pub(crate) fn normalize_codex_apps_callable_namespace(
+    server_name: &str,
+    connector_name: Option<&str>,
+) -> String {
+    if server_name == CODEX_APPS_MCP_SERVER_NAME
+        && let Some(connector_name) = connector_name
+    {
+        format!("mcp__{}__{}", server_name, sanitize_name(connector_name))
+    } else {
+        format!("mcp__{server_name}__")
+    }
+}
+
+pub(crate) fn write_cached_codex_apps_tools_if_needed(
+    server_name: &str,
+    cache_context: Option<&CodexAppsToolsCacheContext>,
+    tools: &[ToolInfo],
+) {
+    if server_name != CODEX_APPS_MCP_SERVER_NAME {
+        return;
+    }
+
+    if let Some(cache_context) = cache_context {
+        let cache_write_start = Instant::now();
+        write_cached_codex_apps_tools(cache_context, tools);
+        emit_duration(
+            MCP_TOOLS_CACHE_WRITE_DURATION_METRIC,
+            cache_write_start.elapsed(),
+            &[],
+        );
+    }
+}
+
+pub(crate) fn load_startup_cached_codex_apps_tools_snapshot(
+    server_name: &str,
+    cache_context: Option<&CodexAppsToolsCacheContext>,
+) -> Option<Vec<ToolInfo>> {
+    if server_name != CODEX_APPS_MCP_SERVER_NAME {
+        return None;
+    }
+
+    let cache_context = cache_context?;
+
+    match load_cached_codex_apps_tools(cache_context) {
+        CachedCodexAppsToolsLoad::Hit(tools) => Some(tools),
+        CachedCodexAppsToolsLoad::Missing | CachedCodexAppsToolsLoad::Invalid => None,
+    }
+}
+
+#[cfg(test)]
+pub(crate) fn read_cached_codex_apps_tools(
+    cache_context: &CodexAppsToolsCacheContext,
+) -> Option<Vec<ToolInfo>> {
+    match load_cached_codex_apps_tools(cache_context) {
+        CachedCodexAppsToolsLoad::Hit(tools) => Some(tools),
+        CachedCodexAppsToolsLoad::Missing | CachedCodexAppsToolsLoad::Invalid => None,
+    }
+}
+
+pub(crate) fn load_cached_codex_apps_tools(
+    cache_context: &CodexAppsToolsCacheContext,
+) -> CachedCodexAppsToolsLoad {
+    let cache_path = cache_context.cache_path();
+    let bytes = match std::fs::read(cache_path) {
+        Ok(bytes) => bytes,
+        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
+            return CachedCodexAppsToolsLoad::Missing;
+        }
+        Err(_) => return CachedCodexAppsToolsLoad::Invalid,
+    };
+    let cache: CodexAppsToolsDiskCache = match serde_json::from_slice(&bytes) {
+        Ok(cache) => cache,
+        Err(_) => return CachedCodexAppsToolsLoad::Invalid,
+    };
+    if cache.schema_version != CODEX_APPS_TOOLS_CACHE_SCHEMA_VERSION {
+        return CachedCodexAppsToolsLoad::Invalid;
+    }
+    CachedCodexAppsToolsLoad::Hit(filter_disallowed_codex_apps_tools(cache.tools))
+}
+
+pub(crate) fn write_cached_codex_apps_tools(
+    cache_context: &CodexAppsToolsCacheContext,
+    tools: &[ToolInfo],
+) {
+    let cache_path = cache_context.cache_path();
+    if let Some(parent) = cache_path.parent()
+        && std::fs::create_dir_all(parent).is_err()
+    {
+        return;
+    }
+    let tools = filter_disallowed_codex_apps_tools(tools.to_vec());
+    let Ok(bytes) = serde_json::to_vec_pretty(&CodexAppsToolsDiskCache {
+        schema_version: CODEX_APPS_TOOLS_CACHE_SCHEMA_VERSION,
+        tools,
+    }) else {
+        return;
+    };
+    let _ = std::fs::write(cache_path, bytes);
+}
+
+pub(crate) fn filter_disallowed_codex_apps_tools(tools: Vec<ToolInfo>) -> Vec<ToolInfo> {
+    tools
+        .into_iter()
+        .filter(|tool| {
+            tool.connector_id
+                .as_deref()
+                .is_none_or(is_connector_id_allowed)
+        })
+        .collect()
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CodexAppsToolsDiskCache {
+    schema_version: u8,
+    tools: Vec<ToolInfo>,
+}
+
+const CODEX_APPS_TOOLS_CACHE_DIR: &str = "cache/codex_apps_tools";
+
+fn sha1_hex(s: &str) -> String {
+    let mut hasher = Sha1::new();
+    hasher.update(s.as_bytes());
+    let sha1 = hasher.finalize();
+    format!("{sha1:x}")
+}

codex-rs/codex-mcp/src/connection_manager.rs

@@ -0,0 +1,700 @@
+//! Aggregates MCP server connections for Codex.
+//!
+//! [`McpConnectionManager`] owns the set of running async RMCP clients keyed by
+//! MCP server name. It coordinates startup status events, keeps server origin
+//! metadata, aggregates tools/resources/templates across servers, routes tool
+//! calls to the right client, and exposes the public manager API used by
+//! `codex-core`.
+
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Duration;
+use std::time::Instant;
+
+use crate::McpAuthStatusEntry;
+use crate::codex_apps::CodexAppsToolsCacheContext;
+use crate::codex_apps::CodexAppsToolsCacheKey;
+use crate::codex_apps::write_cached_codex_apps_tools_if_needed;
+use crate::elicitation::ElicitationRequestManager;
+use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
+use crate::mcp::ToolPluginProvenance;
+use crate::rmcp_client::AsyncManagedClient;
+use crate::rmcp_client::DEFAULT_STARTUP_TIMEOUT;
+use crate::rmcp_client::MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC;
+use crate::rmcp_client::MCP_TOOLS_LIST_DURATION_METRIC;
+use crate::rmcp_client::ManagedClient;
+use crate::rmcp_client::StartupOutcomeError;
+use crate::rmcp_client::list_tools_for_client_uncached;
+use crate::runtime::McpRuntimeEnvironment;
+use crate::runtime::emit_duration;
+use crate::tools::ToolInfo;
+use crate::tools::filter_tools;
+use crate::tools::qualify_tools;
+use crate::tools::tool_with_model_visible_input_schema;
+use anyhow::Context;
+use anyhow::Result;
+use anyhow::anyhow;
+use async_channel::Sender;
+use codex_config::Constrained;
+use codex_config::McpServerConfig;
+use codex_config::McpServerTransportConfig;
+use codex_config::types::OAuthCredentialsStoreMode;
+use codex_login::CodexAuth;
+use codex_protocol::ToolName;
+use codex_protocol::mcp::CallToolResult;
+use codex_protocol::models::PermissionProfile;
+use codex_protocol::protocol::AskForApproval;
+use codex_protocol::protocol::Event;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::McpStartupCompleteEvent;
+use codex_protocol::protocol::McpStartupFailure;
+use codex_protocol::protocol::McpStartupStatus;
+use codex_protocol::protocol::McpStartupUpdateEvent;
+use codex_rmcp_client::ElicitationResponse;
+use rmcp::model::ListResourceTemplatesResult;
+use rmcp::model::ListResourcesResult;
+use rmcp::model::PaginatedRequestParams;
+use rmcp::model::ReadResourceRequestParams;
+use rmcp::model::ReadResourceResult;
+use rmcp::model::RequestId;
+use rmcp::model::Resource;
+use rmcp::model::ResourceTemplate;
+use tokio::task::JoinSet;
+use tokio_util::sync::CancellationToken;
+use tracing::instrument;
+use tracing::warn;
+use url::Url;
+
+/// A thin wrapper around a set of running [`RmcpClient`] instances.
+pub struct McpConnectionManager {
+    clients: HashMap<String, AsyncManagedClient>,
+    server_origins: HashMap<String, String>,
+    elicitation_requests: ElicitationRequestManager,
+}
+
+impl McpConnectionManager {
+    pub fn new_uninitialized(
+        approval_policy: &Constrained<AskForApproval>,
+        permission_profile: &Constrained<PermissionProfile>,
+    ) -> Self {
+        Self {
+            clients: HashMap::new(),
+            server_origins: HashMap::new(),
+            elicitation_requests: ElicitationRequestManager::new(
+                approval_policy.value(),
+                permission_profile.get().clone(),
+            ),
+        }
+    }
+
+    pub fn has_servers(&self) -> bool {
+        !self.clients.is_empty()
+    }
+
+    pub fn server_origin(&self, server_name: &str) -> Option<&str> {
+        self.server_origins.get(server_name).map(String::as_str)
+    }
+
+    pub fn set_approval_policy(&self, approval_policy: &Constrained<AskForApproval>) {
+        if let Ok(mut policy) = self.elicitation_requests.approval_policy.lock() {
+            *policy = approval_policy.value();
+        }
+    }
+
+    pub fn set_permission_profile(&self, permission_profile: PermissionProfile) {
+        if let Ok(mut profile) = self.elicitation_requests.permission_profile.lock() {
+            *profile = permission_profile;
+        }
+    }
+
+    #[allow(clippy::new_ret_no_self, clippy::too_many_arguments)]
+    pub async fn new(
+        mcp_servers: &HashMap<String, McpServerConfig>,
+        store_mode: OAuthCredentialsStoreMode,
+        auth_entries: HashMap<String, McpAuthStatusEntry>,
+        approval_policy: &Constrained<AskForApproval>,
+        submit_id: String,
+        tx_event: Sender<Event>,
+        initial_permission_profile: PermissionProfile,
+        runtime_environment: McpRuntimeEnvironment,
+        codex_home: PathBuf,
+        codex_apps_tools_cache_key: CodexAppsToolsCacheKey,
+        tool_plugin_provenance: ToolPluginProvenance,
+        auth: Option<&CodexAuth>,
+    ) -> (Self, CancellationToken) {
+        let cancel_token = CancellationToken::new();
+        let mut clients = HashMap::new();
+        let mut server_origins = HashMap::new();
+        let mut join_set = JoinSet::new();
+        let elicitation_requests =
+            ElicitationRequestManager::new(approval_policy.value(), initial_permission_profile);
+        let tool_plugin_provenance = Arc::new(tool_plugin_provenance);
+        let startup_submit_id = submit_id.clone();
+        let codex_apps_auth_provider = auth
+            .filter(|auth| auth.uses_codex_backend())
+            .map(codex_model_provider::auth_provider_from_auth);
+        let mcp_servers = mcp_servers.clone();
+        for (server_name, cfg) in mcp_servers.into_iter().filter(|(_, cfg)| cfg.enabled) {
+            if let Some(origin) = transport_origin(&cfg.transport) {
+                server_origins.insert(server_name.clone(), origin);
+            }
+            let cancel_token = cancel_token.child_token();
+            let _ = emit_update(
+                startup_submit_id.as_str(),
+                &tx_event,
+                McpStartupUpdateEvent {
+                    server: server_name.clone(),
+                    status: McpStartupStatus::Starting,
+                },
+            )
+            .await;
+            let codex_apps_tools_cache_context = if server_name == CODEX_APPS_MCP_SERVER_NAME {
+                Some(CodexAppsToolsCacheContext {
+                    codex_home: codex_home.clone(),
+                    user_key: codex_apps_tools_cache_key.clone(),
+                })
+            } else {
+                None
+            };
+            let uses_env_bearer_token = match &cfg.transport {
+                McpServerTransportConfig::StreamableHttp {
+                    bearer_token_env_var,
+                    ..
+                } => bearer_token_env_var.is_some(),
+                McpServerTransportConfig::Stdio { .. } => false,
+            };
+            let runtime_auth_provider =
+                if server_name == CODEX_APPS_MCP_SERVER_NAME && !uses_env_bearer_token {
+                    codex_apps_auth_provider.clone()
+                } else {
+                    None
+                };
+            let async_managed_client = AsyncManagedClient::new(
+                server_name.clone(),
+                cfg,
+                store_mode,
+                cancel_token.clone(),
+                tx_event.clone(),
+                elicitation_requests.clone(),
+                codex_apps_tools_cache_context,
+                Arc::clone(&tool_plugin_provenance),
+                runtime_environment.clone(),
+                runtime_auth_provider,
+            );
+            clients.insert(server_name.clone(), async_managed_client.clone());
+            let tx_event = tx_event.clone();
+            let submit_id = startup_submit_id.clone();
+            let auth_entry = auth_entries.get(&server_name).cloned();
+            join_set.spawn(async move {
+                let mut outcome = async_managed_client.client().await;
+                if cancel_token.is_cancelled() {
+                    outcome = Err(StartupOutcomeError::Cancelled);
+                }
+                let status = match &outcome {
+                    Ok(_) => McpStartupStatus::Ready,
+                    Err(StartupOutcomeError::Cancelled) => McpStartupStatus::Cancelled,
+                    Err(error) => {
+                        let error_str = mcp_init_error_display(
+                            server_name.as_str(),
+                            auth_entry.as_ref(),
+                            error,
+                        );
+                        McpStartupStatus::Failed { error: error_str }
+                    }
+                };
+
+                let _ = emit_update(
+                    submit_id.as_str(),
+                    &tx_event,
+                    McpStartupUpdateEvent {
+                        server: server_name.clone(),
+                        status,
+                    },
+                )
+                .await;
+
+                (server_name, outcome)
+            });
+        }
+        let manager = Self {
+            clients,
+            server_origins,
+            elicitation_requests: elicitation_requests.clone(),
+        };
+        tokio::spawn(async move {
+            let outcomes = join_set.join_all().await;
+            let mut summary = McpStartupCompleteEvent::default();
+            for (server_name, outcome) in outcomes {
+                match outcome {
+                    Ok(_) => summary.ready.push(server_name),
+                    Err(StartupOutcomeError::Cancelled) => summary.cancelled.push(server_name),
+                    Err(StartupOutcomeError::Failed { error }) => {
+                        summary.failed.push(McpStartupFailure {
+                            server: server_name,
+                            error,
+                        })
+                    }
+                }
+            }
+            let _ = tx_event
+                .send(Event {
+                    id: startup_submit_id,
+                    msg: EventMsg::McpStartupComplete(summary),
+                })
+                .await;
+        });
+        (manager, cancel_token)
+    }
+
+    pub async fn resolve_elicitation(
+        &self,
+        server_name: String,
+        id: RequestId,
+        response: ElicitationResponse,
+    ) -> Result<()> {
+        self.elicitation_requests
+            .resolve(server_name, id, response)
+            .await
+    }
+
+    pub async fn wait_for_server_ready(&self, server_name: &str, timeout: Duration) -> bool {
+        let Some(async_managed_client) = self.clients.get(server_name) else {
+            return false;
+        };
+
+        match tokio::time::timeout(timeout, async_managed_client.client()).await {
+            Ok(Ok(_)) => true,
+            Ok(Err(_)) | Err(_) => false,
+        }
+    }
+
+    pub async fn required_startup_failures(
+        &self,
+        required_servers: &[String],
+    ) -> Vec<McpStartupFailure> {
+        let mut failures = Vec::new();
+        for server_name in required_servers {
+            let Some(async_managed_client) = self.clients.get(server_name).cloned() else {
+                failures.push(McpStartupFailure {
+                    server: server_name.clone(),
+                    error: format!("required MCP server `{server_name}` was not initialized"),
+                });
+                continue;
+            };
+
+            match async_managed_client.client().await {
+                Ok(_) => {}
+                Err(error) => failures.push(McpStartupFailure {
+                    server: server_name.clone(),
+                    error: startup_outcome_error_message(error),
+                }),
+            }
+        }
+        failures
+    }
+
+    /// Returns a single map that contains all tools. Each key is the
+    /// fully-qualified name for the tool.
+    #[instrument(level = "trace", skip_all)]
+    pub async fn list_all_tools(&self) -> HashMap<String, ToolInfo> {
+        let mut tools = Vec::new();
+        for managed_client in self.clients.values() {
+            let Some(server_tools) = managed_client.listed_tools().await else {
+                continue;
+            };
+            tools.extend(server_tools);
+        }
+        qualify_tools(tools)
+    }
+
+    /// Force-refresh codex apps tools by bypassing the in-process cache.
+    ///
+    /// On success, the refreshed tools replace the cache contents and the
+    /// latest filtered tool map is returned directly to the caller. On
+    /// failure, the existing cache remains unchanged.
+    pub async fn hard_refresh_codex_apps_tools_cache(&self) -> Result<HashMap<String, ToolInfo>> {
+        let managed_client = self
+            .clients
+            .get(CODEX_APPS_MCP_SERVER_NAME)
+            .ok_or_else(|| anyhow!("unknown MCP server '{CODEX_APPS_MCP_SERVER_NAME}'"))?
+            .client()
+            .await
+            .context("failed to get client")?;
+
+        let list_start = Instant::now();
+        let fetch_start = Instant::now();
+        let tools = list_tools_for_client_uncached(
+            CODEX_APPS_MCP_SERVER_NAME,
+            &managed_client.client,
+            managed_client.tool_timeout,
+            managed_client.server_instructions.as_deref(),
+        )
+        .await
+        .with_context(|| {
+            format!("failed to refresh tools for MCP server '{CODEX_APPS_MCP_SERVER_NAME}'")
+        })?;
+        emit_duration(
+            MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC,
+            fetch_start.elapsed(),
+            &[],
+        );
+
+        write_cached_codex_apps_tools_if_needed(
+            CODEX_APPS_MCP_SERVER_NAME,
+            managed_client.codex_apps_tools_cache_context.as_ref(),
+            &tools,
+        );
+        emit_duration(
+            MCP_TOOLS_LIST_DURATION_METRIC,
+            list_start.elapsed(),
+            &[("cache", "miss")],
+        );
+        let tools = filter_tools(tools, &managed_client.tool_filter)
+            .into_iter()
+            .map(|mut tool| {
+                tool.tool = tool_with_model_visible_input_schema(&tool.tool);
+                tool
+            });
+        Ok(qualify_tools(tools))
+    }
+
+    /// Returns a single map that contains all resources. Each key is the
+    /// server name and the value is a vector of resources.
+    pub async fn list_all_resources(&self) -> HashMap<String, Vec<Resource>> {
+        let mut join_set = JoinSet::new();
+
+        let clients_snapshot = &self.clients;
+
+        for (server_name, async_managed_client) in clients_snapshot {
+            let server_name = server_name.clone();
+            let Ok(managed_client) = async_managed_client.client().await else {
+                continue;
+            };
+            let timeout = managed_client.tool_timeout;
+            let client = managed_client.client.clone();
+
+            join_set.spawn(async move {
+                let mut collected: Vec<Resource> = Vec::new();
+                let mut cursor: Option<String> = None;
+
+                loop {
+                    let params = cursor.as_ref().map(|next| PaginatedRequestParams {
+                        meta: None,
+                        cursor: Some(next.clone()),
+                    });
+                    let response = match client.list_resources(params, timeout).await {
+                        Ok(result) => result,
+                        Err(err) => return (server_name, Err(err)),
+                    };
+
+                    collected.extend(response.resources);
+
+                    match response.next_cursor {
+                        Some(next) => {
+                            if cursor.as_ref() == Some(&next) {
+                                return (
+                                    server_name,
+                                    Err(anyhow!("resources/list returned duplicate cursor")),
+                                );
+                            }
+                            cursor = Some(next);
+                        }
+                        None => return (server_name, Ok(collected)),
+                    }
+                }
+            });
+        }
+
+        let mut aggregated: HashMap<String, Vec<Resource>> = HashMap::new();
+
+        while let Some(join_res) = join_set.join_next().await {
+            match join_res {
+                Ok((server_name, Ok(resources))) => {
+                    aggregated.insert(server_name, resources);
+                }
+                Ok((server_name, Err(err))) => {
+                    warn!("Failed to list resources for MCP server '{server_name}': {err:#}");
+                }
+                Err(err) => {
+                    warn!("Task panic when listing resources for MCP server: {err:#}");
+                }
+            }
+        }
+
+        aggregated
+    }
+
+    /// Returns a single map that contains all resource templates. Each key is the
+    /// server name and the value is a vector of resource templates.
+    pub async fn list_all_resource_templates(&self) -> HashMap<String, Vec<ResourceTemplate>> {
+        let mut join_set = JoinSet::new();
+
+        let clients_snapshot = &self.clients;
+
+        for (server_name, async_managed_client) in clients_snapshot {
+            let server_name_cloned = server_name.clone();
+            let Ok(managed_client) = async_managed_client.client().await else {
+                continue;
+            };
+            let client = managed_client.client.clone();
+            let timeout = managed_client.tool_timeout;
+
+            join_set.spawn(async move {
+                let mut collected: Vec<ResourceTemplate> = Vec::new();
+                let mut cursor: Option<String> = None;
+
+                loop {
+                    let params = cursor.as_ref().map(|next| PaginatedRequestParams {
+                        meta: None,
+                        cursor: Some(next.clone()),
+                    });
+                    let response = match client.list_resource_templates(params, timeout).await {
+                        Ok(result) => result,
+                        Err(err) => return (server_name_cloned, Err(err)),
+                    };
+
+                    collected.extend(response.resource_templates);
+
+                    match response.next_cursor {
+                        Some(next) => {
+                            if cursor.as_ref() == Some(&next) {
+                                return (
+                                    server_name_cloned,
+                                    Err(anyhow!(
+                                        "resources/templates/list returned duplicate cursor"
+                                    )),
+                                );
+                            }
+                            cursor = Some(next);
+                        }
+                        None => return (server_name_cloned, Ok(collected)),
+                    }
+                }
+            });
+        }
+
+        let mut aggregated: HashMap<String, Vec<ResourceTemplate>> = HashMap::new();
+
+        while let Some(join_res) = join_set.join_next().await {
+            match join_res {
+                Ok((server_name, Ok(templates))) => {
+                    aggregated.insert(server_name, templates);
+                }
+                Ok((server_name, Err(err))) => {
+                    warn!(
+                        "Failed to list resource templates for MCP server '{server_name}': {err:#}"
+                    );
+                }
+                Err(err) => {
+                    warn!("Task panic when listing resource templates for MCP server: {err:#}");
+                }
+            }
+        }
+
+        aggregated
+    }
+
+    /// Invoke the tool indicated by the (server, tool) pair.
+    pub async fn call_tool(
+        &self,
+        server: &str,
+        tool: &str,
+        arguments: Option<serde_json::Value>,
+        meta: Option<serde_json::Value>,
+    ) -> Result<CallToolResult> {
+        let client = self.client_by_name(server).await?;
+        if !client.tool_filter.allows(tool) {
+            return Err(anyhow!(
+                "tool '{tool}' is disabled for MCP server '{server}'"
+            ));
+        }
+
+        let result: rmcp::model::CallToolResult = client
+            .client
+            .call_tool(tool.to_string(), arguments, meta, client.tool_timeout)
+            .await
+            .with_context(|| format!("tool call failed for `{server}/{tool}`"))?;
+
+        let content = result
+            .content
+            .into_iter()
+            .map(|content| {
+                serde_json::to_value(content)
+                    .unwrap_or_else(|_| serde_json::Value::String("<content>".to_string()))
+            })
+            .collect();
+
+        Ok(CallToolResult {
+            content,
+            structured_content: result.structured_content,
+            is_error: result.is_error,
+            meta: result.meta.and_then(|meta| serde_json::to_value(meta).ok()),
+        })
+    }
+
+    pub async fn server_supports_sandbox_state_meta_capability(
+        &self,
+        server: &str,
+    ) -> Result<bool> {
+        Ok(self
+            .client_by_name(server)
+            .await?
+            .server_supports_sandbox_state_meta_capability)
+    }
+
+    /// List resources from the specified server.
+    pub async fn list_resources(
+        &self,
+        server: &str,
+        params: Option<PaginatedRequestParams>,
+    ) -> Result<ListResourcesResult> {
+        let managed = self.client_by_name(server).await?;
+        let timeout = managed.tool_timeout;
+
+        managed
+            .client
+            .list_resources(params, timeout)
+            .await
+            .with_context(|| format!("resources/list failed for `{server}`"))
+    }
+
+    /// List resource templates from the specified server.
+    pub async fn list_resource_templates(
+        &self,
+        server: &str,
+        params: Option<PaginatedRequestParams>,
+    ) -> Result<ListResourceTemplatesResult> {
+        let managed = self.client_by_name(server).await?;
+        let client = managed.client.clone();
+        let timeout = managed.tool_timeout;
+
+        client
+            .list_resource_templates(params, timeout)
+            .await
+            .with_context(|| format!("resources/templates/list failed for `{server}`"))
+    }
+
+    /// Read a resource from the specified server.
+    pub async fn read_resource(
+        &self,
+        server: &str,
+        params: ReadResourceRequestParams,
+    ) -> Result<ReadResourceResult> {
+        let managed = self.client_by_name(server).await?;
+        let client = managed.client.clone();
+        let timeout = managed.tool_timeout;
+        let uri = params.uri.clone();
+
+        client
+            .read_resource(params, timeout)
+            .await
+            .with_context(|| format!("resources/read failed for `{server}` ({uri})"))
+    }
+
+    pub async fn resolve_tool_info(&self, tool_name: &ToolName) -> Option<ToolInfo> {
+        let all_tools = self.list_all_tools().await;
+        all_tools
+            .into_values()
+            .find(|tool| tool.canonical_tool_name() == *tool_name)
+    }
+
+    async fn client_by_name(&self, name: &str) -> Result<ManagedClient> {
+        self.clients
+            .get(name)
+            .ok_or_else(|| anyhow!("unknown MCP server '{name}'"))?
+            .client()
+            .await
+            .context("failed to get client")
+    }
+}
+
+async fn emit_update(
+    submit_id: &str,
+    tx_event: &Sender<Event>,
+    update: McpStartupUpdateEvent,
+) -> Result<(), async_channel::SendError<Event>> {
+    tx_event
+        .send(Event {
+            id: submit_id.to_string(),
+            msg: EventMsg::McpStartupUpdate(update),
+        })
+        .await
+}
+
+fn transport_origin(transport: &McpServerTransportConfig) -> Option<String> {
+    match transport {
+        McpServerTransportConfig::StreamableHttp { url, .. } => {
+            let parsed = Url::parse(url).ok()?;
+            Some(parsed.origin().ascii_serialization())
+        }
+        McpServerTransportConfig::Stdio { .. } => Some("stdio".to_string()),
+    }
+}
+
+fn mcp_init_error_display(
+    server_name: &str,
+    entry: Option<&McpAuthStatusEntry>,
+    err: &StartupOutcomeError,
+) -> String {
+    if let Some(McpServerTransportConfig::StreamableHttp {
+        url,
+        bearer_token_env_var,
+        http_headers,
+        ..
+    }) = &entry.map(|entry| &entry.config.transport)
+        && url == "https://api.githubcopilot.com/mcp/"
+        && bearer_token_env_var.is_none()
+        && http_headers.as_ref().map(HashMap::is_empty).unwrap_or(true)
+    {
+        format!(
+            "GitHub MCP does not support OAuth. Log in by adding a personal access token (https://github.com/settings/personal-access-tokens) to your environment and config.toml:\n[mcp_servers.{server_name}]\nbearer_token_env_var = CODEX_GITHUB_PERSONAL_ACCESS_TOKEN"
+        )
+    } else if is_mcp_client_auth_required_error(err) {
+        format!(
+            "The {server_name} MCP server is not logged in. Run `codex mcp login {server_name}`."
+        )
+    } else if is_mcp_client_startup_timeout_error(err) {
+        let startup_timeout_secs = match entry {
+            Some(entry) => match entry.config.startup_timeout_sec {
+                Some(timeout) => timeout,
+                None => DEFAULT_STARTUP_TIMEOUT,
+            },
+            None => DEFAULT_STARTUP_TIMEOUT,
+        }
+        .as_secs();
+        format!(
+            "MCP client for `{server_name}` timed out after {startup_timeout_secs} seconds. Add or adjust `startup_timeout_sec` in your config.toml:\n[mcp_servers.{server_name}]\nstartup_timeout_sec = XX"
+        )
+    } else {
+        format!("MCP client for `{server_name}` failed to start: {err:#}")
+    }
+}
+
+fn startup_outcome_error_message(error: StartupOutcomeError) -> String {
+    match error {
+        StartupOutcomeError::Cancelled => "MCP startup cancelled".to_string(),
+        StartupOutcomeError::Failed { error } => error,
+    }
+}
+
+fn is_mcp_client_auth_required_error(error: &StartupOutcomeError) -> bool {
+    match error {
+        StartupOutcomeError::Failed { error } => error.contains("Auth required"),
+        _ => false,
+    }
+}
+
+fn is_mcp_client_startup_timeout_error(error: &StartupOutcomeError) -> bool {
+    match error {
+        StartupOutcomeError::Failed { error } => {
+            error.contains("request timed out")
+                || error.contains("timed out handshaking with MCP server")
+        }
+        _ => false,
+    }
+}
+
+#[cfg(test)]
+#[path = "connection_manager_tests.rs"]
+mod tests;

codex-rs/codex-mcp/src/connection_manager_tests.rs

@@ -1,12 +1,36 @@
 use super::*;
+use crate::codex_apps::CODEX_APPS_TOOLS_CACHE_SCHEMA_VERSION;
+use crate::codex_apps::CodexAppsToolsCacheContext;
+use crate::codex_apps::load_startup_cached_codex_apps_tools_snapshot;
+use crate::codex_apps::read_cached_codex_apps_tools;
+use crate::codex_apps::write_cached_codex_apps_tools;
+use crate::declared_openai_file_input_param_names;
+use crate::elicitation::ElicitationRequestManager;
+use crate::elicitation::elicitation_is_rejected_by_policy;
+use crate::rmcp_client::AsyncManagedClient;
+use crate::rmcp_client::ManagedClient;
+use crate::rmcp_client::StartupOutcomeError;
+use crate::rmcp_client::elicitation_capability_for_server;
+use crate::tools::ToolFilter;
+use crate::tools::ToolInfo;
+use crate::tools::filter_tools;
+use crate::tools::qualify_tools;
+use crate::tools::tool_with_model_visible_input_schema;
+use codex_config::Constrained;
 use codex_protocol::ToolName;
 use codex_protocol::models::PermissionProfile;
 use codex_protocol::protocol::GranularApprovalConfig;
 use codex_protocol::protocol::McpAuthStatus;
+use futures::FutureExt;
 use pretty_assertions::assert_eq;
+use rmcp::model::CreateElicitationRequestParams;
+use rmcp::model::ElicitationAction;
+use rmcp::model::ElicitationCapability;
+use rmcp::model::FormElicitationCapability;
 use rmcp::model::JsonObject;
 use rmcp::model::Meta;
 use rmcp::model::NumberOrString;
+use rmcp::model::Tool;
 use std::collections::HashSet;
 use std::sync::Arc;
 use tempfile::tempdir;

codex-rs/codex-mcp/src/elicitation.rs

@@ -0,0 +1,190 @@
+//! MCP elicitation request tracking and policy handling.
+//!
+//! RMCP clients call into this module when a server asks Codex to elicit data
+//! from the user. It decides whether the request can be automatically accepted,
+//! must be declined by policy, or should be surfaced as a Codex protocol event
+//! and later resolved through the stored responder.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::sync::Mutex as StdMutex;
+
+use crate::mcp::mcp_permission_prompt_is_auto_approved;
+use anyhow::Context;
+use anyhow::Result;
+use anyhow::anyhow;
+use async_channel::Sender;
+use codex_protocol::approvals::ElicitationRequest;
+use codex_protocol::approvals::ElicitationRequestEvent;
+use codex_protocol::mcp::RequestId as ProtocolRequestId;
+use codex_protocol::models::PermissionProfile;
+use codex_protocol::protocol::AskForApproval;
+use codex_protocol::protocol::Event;
+use codex_protocol::protocol::EventMsg;
+use codex_rmcp_client::ElicitationResponse;
+use codex_rmcp_client::SendElicitation;
+use futures::future::FutureExt;
+use rmcp::model::CreateElicitationRequestParams;
+use rmcp::model::ElicitationAction;
+use rmcp::model::RequestId;
+use tokio::sync::Mutex;
+use tokio::sync::oneshot;
+
+#[derive(Clone)]
+pub(crate) struct ElicitationRequestManager {
+    requests: Arc<Mutex<ResponderMap>>,
+    pub(crate) approval_policy: Arc<StdMutex<AskForApproval>>,
+    pub(crate) permission_profile: Arc<StdMutex<PermissionProfile>>,
+}
+
+impl ElicitationRequestManager {
+    pub(crate) fn new(
+        approval_policy: AskForApproval,
+        permission_profile: PermissionProfile,
+    ) -> Self {
+        Self {
+            requests: Arc::new(Mutex::new(HashMap::new())),
+            approval_policy: Arc::new(StdMutex::new(approval_policy)),
+            permission_profile: Arc::new(StdMutex::new(permission_profile)),
+        }
+    }
+
+    pub(crate) async fn resolve(
+        &self,
+        server_name: String,
+        id: RequestId,
+        response: ElicitationResponse,
+    ) -> Result<()> {
+        self.requests
+            .lock()
+            .await
+            .remove(&(server_name, id))
+            .ok_or_else(|| anyhow!("elicitation request not found"))?
+            .send(response)
+            .map_err(|e| anyhow!("failed to send elicitation response: {e:?}"))
+    }
+
+    pub(crate) fn make_sender(
+        &self,
+        server_name: String,
+        tx_event: Sender<Event>,
+    ) -> SendElicitation {
+        let elicitation_requests = self.requests.clone();
+        let approval_policy = self.approval_policy.clone();
+        let permission_profile = self.permission_profile.clone();
+        Box::new(move |id, elicitation| {
+            let elicitation_requests = elicitation_requests.clone();
+            let tx_event = tx_event.clone();
+            let server_name = server_name.clone();
+            let approval_policy = approval_policy.clone();
+            let permission_profile = permission_profile.clone();
+            async move {
+                let approval_policy = approval_policy
+                    .lock()
+                    .map(|policy| *policy)
+                    .unwrap_or(AskForApproval::Never);
+                let permission_profile = permission_profile
+                    .lock()
+                    .map(|profile| profile.clone())
+                    .unwrap_or_default();
+                if mcp_permission_prompt_is_auto_approved(approval_policy, &permission_profile)
+                    && can_auto_accept_elicitation(&elicitation)
+                {
+                    return Ok(ElicitationResponse {
+                        action: ElicitationAction::Accept,
+                        content: Some(serde_json::json!({})),
+                        meta: None,
+                    });
+                }
+
+                if elicitation_is_rejected_by_policy(approval_policy) {
+                    return Ok(ElicitationResponse {
+                        action: ElicitationAction::Decline,
+                        content: None,
+                        meta: None,
+                    });
+                }
+
+                let request = match elicitation {
+                    CreateElicitationRequestParams::FormElicitationParams {
+                        meta,
+                        message,
+                        requested_schema,
+                    } => ElicitationRequest::Form {
+                        meta: meta
+                            .map(serde_json::to_value)
+                            .transpose()
+                            .context("failed to serialize MCP elicitation metadata")?,
+                        message,
+                        requested_schema: serde_json::to_value(requested_schema)
+                            .context("failed to serialize MCP elicitation schema")?,
+                    },
+                    CreateElicitationRequestParams::UrlElicitationParams {
+                        meta,
+                        message,
+                        url,
+                        elicitation_id,
+                    } => ElicitationRequest::Url {
+                        meta: meta
+                            .map(serde_json::to_value)
+                            .transpose()
+                            .context("failed to serialize MCP elicitation metadata")?,
+                        message,
+                        url,
+                        elicitation_id,
+                    },
+                };
+                let (tx, rx) = oneshot::channel();
+                {
+                    let mut lock = elicitation_requests.lock().await;
+                    lock.insert((server_name.clone(), id.clone()), tx);
+                }
+                let _ = tx_event
+                    .send(Event {
+                        id: "mcp_elicitation_request".to_string(),
+                        msg: EventMsg::ElicitationRequest(ElicitationRequestEvent {
+                            turn_id: None,
+                            server_name,
+                            id: match id.clone() {
+                                rmcp::model::NumberOrString::String(value) => {
+                                    ProtocolRequestId::String(value.to_string())
+                                }
+                                rmcp::model::NumberOrString::Number(value) => {
+                                    ProtocolRequestId::Integer(value)
+                                }
+                            },
+                            request,
+                        }),
+                    })
+                    .await;
+                rx.await
+                    .context("elicitation request channel closed unexpectedly")
+            }
+            .boxed()
+        })
+    }
+}
+
+pub(crate) fn elicitation_is_rejected_by_policy(approval_policy: AskForApproval) -> bool {
+    match approval_policy {
+        AskForApproval::Never => true,
+        AskForApproval::OnFailure => false,
+        AskForApproval::OnRequest => false,
+        AskForApproval::UnlessTrusted => false,
+        AskForApproval::Granular(granular_config) => !granular_config.allows_mcp_elicitations(),
+    }
+}
+
+type ResponderMap = HashMap<(String, RequestId), oneshot::Sender<ElicitationResponse>>;
+
+fn can_auto_accept_elicitation(elicitation: &CreateElicitationRequestParams) -> bool {
+    match elicitation {
+        CreateElicitationRequestParams::FormElicitationParams {
+            requested_schema, ..
+        } => {
+            // Auto-accept confirm/approval elicitations without schema requirements.
+            requested_schema.properties.is_empty()
+        }
+        CreateElicitationRequestParams::UrlElicitationParams { .. } => false,
+    }
+}

codex-rs/codex-mcp/src/lib.rs

@@ -1,15 +1,15 @@
-pub use mcp_connection_manager::MCP_SANDBOX_STATE_META_CAPABILITY;
-pub use mcp_connection_manager::McpConnectionManager;
-pub use mcp_connection_manager::McpRuntimeEnvironment;
-pub use mcp_connection_manager::SandboxState;
-pub use mcp_connection_manager::ToolInfo;
+pub use connection_manager::McpConnectionManager;
+pub use rmcp_client::MCP_SANDBOX_STATE_META_CAPABILITY;
+pub use runtime::McpRuntimeEnvironment;
+pub use runtime::SandboxState;
+pub use tools::ToolInfo;
 
 pub use mcp::CODEX_APPS_MCP_SERVER_NAME;
 pub use mcp::McpConfig;
 pub use mcp::ToolPluginProvenance;
 
-pub use mcp_connection_manager::CodexAppsToolsCacheKey;
-pub use mcp_connection_manager::codex_apps_tools_cache_key;
+pub use codex_apps::CodexAppsToolsCacheKey;
+pub use codex_apps::codex_apps_tools_cache_key;
 
 pub use mcp::configured_mcp_servers;
 pub use mcp::effective_mcp_servers;
@@ -33,11 +33,15 @@ pub use mcp::oauth_login_support;
 pub use mcp::resolve_oauth_scopes;
 pub use mcp::should_retry_without_scopes;
 
+pub use codex_apps::filter_non_codex_apps_mcp_tools_only;
 pub use mcp::mcp_permission_prompt_is_auto_approved;
 pub use mcp::qualified_mcp_tool_name_prefix;
-pub use mcp_connection_manager::declared_openai_file_input_param_names;
-pub use mcp_connection_manager::filter_non_codex_apps_mcp_tools_only;
+pub use tools::declared_openai_file_input_param_names;
 
+pub(crate) mod codex_apps;
+pub(crate) mod connection_manager;
+pub(crate) mod elicitation;
 pub(crate) mod mcp;
-pub(crate) mod mcp_connection_manager;
-pub(crate) mod mcp_tool_names;
+pub(crate) mod rmcp_client;
+pub(crate) mod runtime;
+pub(crate) mod tools;

codex-rs/codex-mcp/src/mcp/mod.rs

@@ -34,9 +34,9 @@ use rmcp::model::ReadResourceRequestParams;
 use rmcp::model::ReadResourceResult;
 use serde_json::Value;
 
-use crate::mcp_connection_manager::McpConnectionManager;
-use crate::mcp_connection_manager::McpRuntimeEnvironment;
-use crate::mcp_connection_manager::codex_apps_tools_cache_key;
+use crate::codex_apps::codex_apps_tools_cache_key;
+use crate::connection_manager::McpConnectionManager;
+use crate::runtime::McpRuntimeEnvironment;
 
 pub const CODEX_APPS_MCP_SERVER_NAME: &str = "codex_apps";
 const MCP_TOOL_NAME_PREFIX: &str = "mcp";

codex-rs/codex-mcp/src/mcp/mod_tests.rs

@@ -64,6 +64,7 @@ fn mcp_prompt_auto_approval_honors_unrestricted_managed_profiles() {
         },
     ));
 }
+
 #[test]
 fn tool_plugin_provenance_collects_app_and_mcp_sources() {
     let provenance = ToolPluginProvenance::from_capability_summaries(&[

codex-rs/codex-mcp/src/rmcp_client.rs

@@ -0,0 +1,591 @@
+//! RMCP client lifecycle for MCP server connections.
+//!
+//! This module owns startup of individual RMCP clients: building the transport,
+//! initializing the server, listing raw tools, applying per-server tool filters,
+//! and exposing cached startup snapshots while a client is still connecting.
+//! Higher-level aggregation and resource/tool APIs live in
+//! [`crate::connection_manager`].
+
+use std::borrow::Cow;
+use std::collections::HashMap;
+use std::env;
+use std::ffi::OsString;
+use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+use std::time::Instant;
+
+use crate::codex_apps::CachedCodexAppsToolsLoad;
+use crate::codex_apps::CodexAppsToolsCacheContext;
+use crate::codex_apps::filter_disallowed_codex_apps_tools;
+use crate::codex_apps::load_cached_codex_apps_tools;
+use crate::codex_apps::load_startup_cached_codex_apps_tools_snapshot;
+use crate::codex_apps::normalize_codex_apps_callable_name;
+use crate::codex_apps::normalize_codex_apps_callable_namespace;
+use crate::codex_apps::normalize_codex_apps_tool_title;
+use crate::codex_apps::write_cached_codex_apps_tools_if_needed;
+use crate::elicitation::ElicitationRequestManager;
+use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
+use crate::mcp::ToolPluginProvenance;
+use crate::runtime::McpRuntimeEnvironment;
+use crate::runtime::emit_duration;
+use crate::tools::ToolFilter;
+use crate::tools::ToolInfo;
+use crate::tools::filter_tools;
+use crate::tools::tool_with_model_visible_input_schema;
+use anyhow::Result;
+use anyhow::anyhow;
+use async_channel::Sender;
+use codex_api::SharedAuthProvider;
+use codex_async_utils::CancelErr;
+use codex_async_utils::OrCancelExt;
+use codex_config::McpServerConfig;
+use codex_config::McpServerTransportConfig;
+use codex_config::types::OAuthCredentialsStoreMode;
+use codex_exec_server::HttpClient;
+use codex_exec_server::ReqwestHttpClient;
+use codex_protocol::protocol::Event;
+use codex_rmcp_client::ExecutorStdioServerLauncher;
+use codex_rmcp_client::LocalStdioServerLauncher;
+use codex_rmcp_client::RmcpClient;
+use codex_rmcp_client::StdioServerLauncher;
+use futures::future::BoxFuture;
+use futures::future::FutureExt;
+use futures::future::Shared;
+use rmcp::model::ClientCapabilities;
+use rmcp::model::ElicitationCapability;
+use rmcp::model::FormElicitationCapability;
+use rmcp::model::Implementation;
+use rmcp::model::InitializeRequestParams;
+use rmcp::model::ProtocolVersion;
+use tokio_util::sync::CancellationToken;
+
+/// MCP server capability indicating that Codex should include [`SandboxState`]
+/// in tool-call request `_meta` under this key.
+pub const MCP_SANDBOX_STATE_META_CAPABILITY: &str = "codex/sandbox-state-meta";
+
+pub(crate) const MCP_TOOLS_LIST_DURATION_METRIC: &str = "codex.mcp.tools.list.duration_ms";
+pub(crate) const MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC: &str =
+    "codex.mcp.tools.fetch_uncached.duration_ms";
+pub(crate) const DEFAULT_STARTUP_TIMEOUT: Duration = Duration::from_secs(30);
+pub(crate) const DEFAULT_TOOL_TIMEOUT: Duration = Duration::from_secs(120);
+
+#[derive(Clone)]
+pub(crate) struct ManagedClient {
+    pub(crate) client: Arc<RmcpClient>,
+    pub(crate) tools: Vec<ToolInfo>,
+    pub(crate) tool_filter: ToolFilter,
+    pub(crate) tool_timeout: Option<Duration>,
+    pub(crate) server_instructions: Option<String>,
+    pub(crate) server_supports_sandbox_state_meta_capability: bool,
+    pub(crate) codex_apps_tools_cache_context: Option<CodexAppsToolsCacheContext>,
+}
+
+impl ManagedClient {
+    fn listed_tools(&self) -> Vec<ToolInfo> {
+        let total_start = Instant::now();
+        if let Some(cache_context) = self.codex_apps_tools_cache_context.as_ref()
+            && let CachedCodexAppsToolsLoad::Hit(tools) =
+                load_cached_codex_apps_tools(cache_context)
+        {
+            emit_duration(
+                MCP_TOOLS_LIST_DURATION_METRIC,
+                total_start.elapsed(),
+                &[("cache", "hit")],
+            );
+            return filter_tools(tools, &self.tool_filter);
+        }
+
+        if self.codex_apps_tools_cache_context.is_some() {
+            emit_duration(
+                MCP_TOOLS_LIST_DURATION_METRIC,
+                total_start.elapsed(),
+                &[("cache", "miss")],
+            );
+        }
+
+        self.tools.clone()
+    }
+}
+
+#[derive(Clone)]
+pub(crate) struct AsyncManagedClient {
+    pub(crate) client: Shared<BoxFuture<'static, Result<ManagedClient, StartupOutcomeError>>>,
+    pub(crate) startup_snapshot: Option<Vec<ToolInfo>>,
+    pub(crate) startup_complete: Arc<AtomicBool>,
+    pub(crate) tool_plugin_provenance: Arc<ToolPluginProvenance>,
+}
+
+impl AsyncManagedClient {
+    // Keep this constructor flat so the startup inputs remain readable at the
+    // single call site instead of introducing a one-off params wrapper.
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) fn new(
+        server_name: String,
+        config: McpServerConfig,
+        store_mode: OAuthCredentialsStoreMode,
+        cancel_token: CancellationToken,
+        tx_event: Sender<Event>,
+        elicitation_requests: ElicitationRequestManager,
+        codex_apps_tools_cache_context: Option<CodexAppsToolsCacheContext>,
+        tool_plugin_provenance: Arc<ToolPluginProvenance>,
+        runtime_environment: McpRuntimeEnvironment,
+        runtime_auth_provider: Option<SharedAuthProvider>,
+    ) -> Self {
+        let tool_filter = ToolFilter::from_config(&config);
+        let startup_snapshot = load_startup_cached_codex_apps_tools_snapshot(
+            &server_name,
+            codex_apps_tools_cache_context.as_ref(),
+        )
+        .map(|tools| filter_tools(tools, &tool_filter));
+        let startup_tool_filter = tool_filter;
+        let startup_complete = Arc::new(AtomicBool::new(false));
+        let startup_complete_for_fut = Arc::clone(&startup_complete);
+        let fut = async move {
+            let outcome = async {
+                if let Err(error) = validate_mcp_server_name(&server_name) {
+                    return Err(error.into());
+                }
+
+                let client = Arc::new(
+                    make_rmcp_client(
+                        &server_name,
+                        config.clone(),
+                        store_mode,
+                        runtime_environment,
+                        runtime_auth_provider,
+                    )
+                    .await?,
+                );
+                match start_server_task(
+                    server_name,
+                    client,
+                    StartServerTaskParams {
+                        startup_timeout: config
+                            .startup_timeout_sec
+                            .or(Some(DEFAULT_STARTUP_TIMEOUT)),
+                        tool_timeout: config.tool_timeout_sec.unwrap_or(DEFAULT_TOOL_TIMEOUT),
+                        tool_filter: startup_tool_filter,
+                        tx_event,
+                        elicitation_requests,
+                        codex_apps_tools_cache_context,
+                    },
+                )
+                .or_cancel(&cancel_token)
+                .await
+                {
+                    Ok(result) => result,
+                    Err(CancelErr::Cancelled) => Err(StartupOutcomeError::Cancelled),
+                }
+            }
+            .await;
+
+            startup_complete_for_fut.store(true, Ordering::Release);
+            outcome
+        };
+        let client = fut.boxed().shared();
+        if startup_snapshot.is_some() {
+            let startup_task = client.clone();
+            tokio::spawn(async move {
+                let _ = startup_task.await;
+            });
+        }
+
+        Self {
+            client,
+            startup_snapshot,
+            startup_complete,
+            tool_plugin_provenance,
+        }
+    }
+
+    pub(crate) async fn client(&self) -> Result<ManagedClient, StartupOutcomeError> {
+        self.client.clone().await
+    }
+
+    fn startup_snapshot_while_initializing(&self) -> Option<Vec<ToolInfo>> {
+        if !self.startup_complete.load(Ordering::Acquire) {
+            return self.startup_snapshot.clone();
+        }
+        None
+    }
+
+    pub(crate) async fn listed_tools(&self) -> Option<Vec<ToolInfo>> {
+        let annotate_tools = |tools: Vec<ToolInfo>| {
+            let mut tools = tools;
+            for tool in &mut tools {
+                if tool.server_name == CODEX_APPS_MCP_SERVER_NAME {
+                    tool.tool = tool_with_model_visible_input_schema(&tool.tool);
+                }
+
+                let plugin_names = match tool.connector_id.as_deref() {
+                    Some(connector_id) => self
+                        .tool_plugin_provenance
+                        .plugin_display_names_for_connector_id(connector_id),
+                    None => self
+                        .tool_plugin_provenance
+                        .plugin_display_names_for_mcp_server_name(tool.server_name.as_str()),
+                };
+                tool.plugin_display_names = plugin_names.to_vec();
+
+                if plugin_names.is_empty() {
+                    continue;
+                }
+
+                let plugin_source_note = if plugin_names.len() == 1 {
+                    format!("This tool is part of plugin `{}`.", plugin_names[0])
+                } else {
+                    format!(
+                        "This tool is part of plugins {}.",
+                        plugin_names
+                            .iter()
+                            .map(|plugin_name| format!("`{plugin_name}`"))
+                            .collect::<Vec<_>>()
+                            .join(", ")
+                    )
+                };
+                let description = tool
+                    .tool
+                    .description
+                    .as_deref()
+                    .map(str::trim)
+                    .unwrap_or("");
+                let annotated_description = if description.is_empty() {
+                    plugin_source_note
+                } else if matches!(description.chars().last(), Some('.' | '!' | '?')) {
+                    format!("{description} {plugin_source_note}")
+                } else {
+                    format!("{description}. {plugin_source_note}")
+                };
+                tool.tool.description = Some(Cow::Owned(annotated_description));
+            }
+            tools
+        };
+
+        // Keep cache payloads raw; plugin provenance is resolved per-session at read time.
+        let tools = if let Some(startup_tools) = self.startup_snapshot_while_initializing() {
+            Some(startup_tools)
+        } else {
+            match self.client().await {
+                Ok(client) => Some(client.listed_tools()),
+                Err(_) => self.startup_snapshot.clone(),
+            }
+        };
+        tools.map(annotate_tools)
+    }
+}
+
+#[derive(Debug, Clone, thiserror::Error)]
+pub(crate) enum StartupOutcomeError {
+    #[error("MCP startup cancelled")]
+    Cancelled,
+    // We can't store the original error here because anyhow::Error doesn't implement
+    // `Clone`.
+    #[error("MCP startup failed: {error}")]
+    Failed { error: String },
+}
+
+impl From<anyhow::Error> for StartupOutcomeError {
+    fn from(error: anyhow::Error) -> Self {
+        Self::Failed {
+            error: error.to_string(),
+        }
+    }
+}
+
+pub(crate) fn elicitation_capability_for_server(
+    _server_name: &str,
+) -> Option<ElicitationCapability> {
+    // https://modelcontextprotocol.io/specification/2025-06-18/client/elicitation#capabilities
+    // indicates this should be an empty object.
+    Some(ElicitationCapability {
+        form: Some(FormElicitationCapability {
+            schema_validation: None,
+        }),
+        url: None,
+    })
+}
+
+pub(crate) async fn list_tools_for_client_uncached(
+    server_name: &str,
+    client: &Arc<RmcpClient>,
+    timeout: Option<Duration>,
+    server_instructions: Option<&str>,
+) -> Result<Vec<ToolInfo>> {
+    let resp = client
+        .list_tools_with_connector_ids(/*params*/ None, timeout)
+        .await?;
+    let tools = resp
+        .tools
+        .into_iter()
+        .map(|tool| {
+            let callable_name = normalize_codex_apps_callable_name(
+                server_name,
+                &tool.tool.name,
+                tool.connector_id.as_deref(),
+                tool.connector_name.as_deref(),
+            );
+            let callable_namespace = normalize_codex_apps_callable_namespace(
+                server_name,
+                tool.connector_name.as_deref(),
+            );
+            let connector_name = tool.connector_name;
+            let connector_description = tool.connector_description;
+            let mut tool_def = tool.tool;
+            if let Some(title) = tool_def.title.as_deref() {
+                let normalized_title =
+                    normalize_codex_apps_tool_title(server_name, connector_name.as_deref(), title);
+                if tool_def.title.as_deref() != Some(normalized_title.as_str()) {
+                    tool_def.title = Some(normalized_title);
+                }
+            }
+            ToolInfo {
+                server_name: server_name.to_owned(),
+                callable_name,
+                callable_namespace,
+                server_instructions: server_instructions.map(str::to_string),
+                tool: tool_def,
+                connector_id: tool.connector_id,
+                connector_name,
+                plugin_display_names: Vec::new(),
+                connector_description,
+            }
+        })
+        .collect();
+    if server_name == CODEX_APPS_MCP_SERVER_NAME {
+        return Ok(filter_disallowed_codex_apps_tools(tools));
+    }
+    Ok(tools)
+}
+
+fn resolve_bearer_token(
+    server_name: &str,
+    bearer_token_env_var: Option<&str>,
+) -> Result<Option<String>> {
+    let Some(env_var) = bearer_token_env_var else {
+        return Ok(None);
+    };
+
+    match env::var(env_var) {
+        Ok(value) => {
+            if value.is_empty() {
+                Err(anyhow!(
+                    "Environment variable {env_var} for MCP server '{server_name}' is empty"
+                ))
+            } else {
+                Ok(Some(value))
+            }
+        }
+        Err(env::VarError::NotPresent) => Err(anyhow!(
+            "Environment variable {env_var} for MCP server '{server_name}' is not set"
+        )),
+        Err(env::VarError::NotUnicode(_)) => Err(anyhow!(
+            "Environment variable {env_var} for MCP server '{server_name}' contains invalid Unicode"
+        )),
+    }
+}
+
+fn validate_mcp_server_name(server_name: &str) -> Result<()> {
+    let re = regex_lite::Regex::new(r"^[a-zA-Z0-9_-]+$")?;
+    if !re.is_match(server_name) {
+        return Err(anyhow!(
+            "Invalid MCP server name '{server_name}': must match pattern {pattern}",
+            pattern = re.as_str()
+        ));
+    }
+    Ok(())
+}
+
+async fn start_server_task(
+    server_name: String,
+    client: Arc<RmcpClient>,
+    params: StartServerTaskParams,
+) -> Result<ManagedClient, StartupOutcomeError> {
+    let StartServerTaskParams {
+        startup_timeout,
+        tool_timeout,
+        tool_filter,
+        tx_event,
+        elicitation_requests,
+        codex_apps_tools_cache_context,
+    } = params;
+    let elicitation = elicitation_capability_for_server(&server_name);
+    let params = InitializeRequestParams {
+        meta: None,
+        capabilities: ClientCapabilities {
+            experimental: None,
+            extensions: None,
+            roots: None,
+            sampling: None,
+            elicitation,
+            tasks: None,
+        },
+        client_info: Implementation {
+            name: "codex-mcp-client".to_owned(),
+            version: env!("CARGO_PKG_VERSION").to_owned(),
+            title: Some("Codex".into()),
+            description: None,
+            icons: None,
+            website_url: None,
+        },
+        protocol_version: ProtocolVersion::V_2025_06_18,
+    };
+
+    let send_elicitation = elicitation_requests.make_sender(server_name.clone(), tx_event);
+
+    let initialize_result = client
+        .initialize(params, startup_timeout, send_elicitation)
+        .await
+        .map_err(StartupOutcomeError::from)?;
+
+    let server_supports_sandbox_state_meta_capability = initialize_result
+        .capabilities
+        .experimental
+        .as_ref()
+        .and_then(|exp| exp.get(MCP_SANDBOX_STATE_META_CAPABILITY))
+        .is_some();
+    let list_start = Instant::now();
+    let fetch_start = Instant::now();
+    let tools = list_tools_for_client_uncached(
+        &server_name,
+        &client,
+        startup_timeout,
+        initialize_result.instructions.as_deref(),
+    )
+    .await
+    .map_err(StartupOutcomeError::from)?;
+    emit_duration(
+        MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC,
+        fetch_start.elapsed(),
+        &[],
+    );
+    write_cached_codex_apps_tools_if_needed(
+        &server_name,
+        codex_apps_tools_cache_context.as_ref(),
+        &tools,
+    );
+    if server_name == CODEX_APPS_MCP_SERVER_NAME {
+        emit_duration(
+            MCP_TOOLS_LIST_DURATION_METRIC,
+            list_start.elapsed(),
+            &[("cache", "miss")],
+        );
+    }
+    let tools = filter_tools(tools, &tool_filter);
+
+    let managed = ManagedClient {
+        client: Arc::clone(&client),
+        tools,
+        tool_timeout: Some(tool_timeout),
+        tool_filter,
+        server_instructions: initialize_result.instructions,
+        server_supports_sandbox_state_meta_capability,
+        codex_apps_tools_cache_context,
+    };
+
+    Ok(managed)
+}
+
+struct StartServerTaskParams {
+    startup_timeout: Option<Duration>, // TODO: cancel_token should handle this.
+    tool_timeout: Duration,
+    tool_filter: ToolFilter,
+    tx_event: Sender<Event>,
+    elicitation_requests: ElicitationRequestManager,
+    codex_apps_tools_cache_context: Option<CodexAppsToolsCacheContext>,
+}
+
+async fn make_rmcp_client(
+    server_name: &str,
+    config: McpServerConfig,
+    store_mode: OAuthCredentialsStoreMode,
+    runtime_environment: McpRuntimeEnvironment,
+    runtime_auth_provider: Option<SharedAuthProvider>,
+) -> Result<RmcpClient, StartupOutcomeError> {
+    let McpServerConfig {
+        transport,
+        experimental_environment,
+        ..
+    } = config;
+    let remote_environment = match experimental_environment.as_deref() {
+        None | Some("local") => false,
+        Some("remote") => {
+            if !runtime_environment.environment().is_remote() {
+                return Err(StartupOutcomeError::from(anyhow!(
+                    "remote MCP server `{server_name}` requires a remote environment"
+                )));
+            }
+            true
+        }
+        Some(environment) => {
+            return Err(StartupOutcomeError::from(anyhow!(
+                "unsupported experimental_environment `{environment}` for MCP server `{server_name}`"
+            )));
+        }
+    };
+
+    match transport {
+        McpServerTransportConfig::Stdio {
+            command,
+            args,
+            env,
+            env_vars,
+            cwd,
+        } => {
+            let command_os: OsString = command.into();
+            let args_os: Vec<OsString> = args.into_iter().map(Into::into).collect();
+            let env_os = env.map(|env| {
+                env.into_iter()
+                    .map(|(key, value)| (key.into(), value.into()))
+                    .collect::<HashMap<_, _>>()
+            });
+            let launcher = if remote_environment {
+                Arc::new(ExecutorStdioServerLauncher::new(
+                    runtime_environment.environment().get_exec_backend(),
+                    runtime_environment.fallback_cwd(),
+                ))
+            } else {
+                Arc::new(LocalStdioServerLauncher::new(
+                    runtime_environment.fallback_cwd(),
+                )) as Arc<dyn StdioServerLauncher>
+            };
+
+            // `RmcpClient` always sees a launched MCP stdio server. The
+            // launcher hides whether that means a local child process or an
+            // executor process whose stdin/stdout bytes cross the process API.
+            RmcpClient::new_stdio_client(command_os, args_os, env_os, &env_vars, cwd, launcher)
+                .await
+                .map_err(|err| StartupOutcomeError::from(anyhow!(err)))
+        }
+        McpServerTransportConfig::StreamableHttp {
+            url,
+            http_headers,
+            env_http_headers,
+            bearer_token_env_var,
+        } => {
+            let http_client: Arc<dyn HttpClient> = if remote_environment {
+                runtime_environment.environment().get_http_client()
+            } else {
+                Arc::new(ReqwestHttpClient)
+            };
+            let resolved_bearer_token =
+                match resolve_bearer_token(server_name, bearer_token_env_var.as_deref()) {
+                    Ok(token) => token,
+                    Err(error) => return Err(error.into()),
+                };
+            RmcpClient::new_streamable_http_client(
+                server_name,
+                &url,
+                resolved_bearer_token,
+                http_headers,
+                env_http_headers,
+                store_mode,
+                http_client,
+                runtime_auth_provider,
+            )
+            .await
+            .map_err(StartupOutcomeError::from)
+        }
+    }
+}

codex-rs/codex-mcp/src/runtime.rs

@@ -0,0 +1,66 @@
+//! Runtime support for Model Context Protocol (MCP) servers.
+//!
+//! This module contains data that describes the runtime environment in which MCP
+//! servers execute, plus the sandbox state payload sent to capable servers and a
+//! tiny shared metrics helper. Transport startup and orchestration live in
+//! [`crate::rmcp_client`] and [`crate::connection_manager`].
+
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Duration;
+
+use codex_exec_server::Environment;
+use codex_protocol::models::PermissionProfile;
+use codex_protocol::protocol::SandboxPolicy;
+
+use serde::Deserialize;
+use serde::Serialize;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SandboxState {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub permission_profile: Option<PermissionProfile>,
+    pub sandbox_policy: SandboxPolicy,
+    pub codex_linux_sandbox_exe: Option<PathBuf>,
+    pub sandbox_cwd: PathBuf,
+    #[serde(default)]
+    pub use_legacy_landlock: bool,
+}
+
+/// Runtime placement information used when starting MCP server transports.
+///
+/// `McpConfig` describes what servers exist. This value describes where those
+/// servers should run for the current caller. Keep it explicit at manager
+/// construction time so status/snapshot paths and real sessions make the same
+/// local-vs-remote decision. `fallback_cwd` is not a per-server override; it is
+/// used when a stdio server omits `cwd` and the launcher needs a concrete
+/// process working directory.
+#[derive(Clone)]
+pub struct McpRuntimeEnvironment {
+    environment: Arc<Environment>,
+    fallback_cwd: PathBuf,
+}
+
+impl McpRuntimeEnvironment {
+    pub fn new(environment: Arc<Environment>, fallback_cwd: PathBuf) -> Self {
+        Self {
+            environment,
+            fallback_cwd,
+        }
+    }
+
+    pub(crate) fn environment(&self) -> Arc<Environment> {
+        Arc::clone(&self.environment)
+    }
+
+    pub(crate) fn fallback_cwd(&self) -> PathBuf {
+        self.fallback_cwd.clone()
+    }
+}
+
+pub(crate) fn emit_duration(metric: &str, duration: Duration, tags: &[(&str, &str)]) {
+    if let Some(metrics) = codex_otel::global() {
+        let _ = metrics.record_duration(metric, duration, tags);
+    }
+}

codex-rs/codex-mcp/src/tools.rs

@@ -1,18 +1,134 @@
-//! Allocates model-visible MCP tool names while preserving raw MCP identities.
+//! MCP tool metadata, filtering, schema shaping, and name qualification.
+//!
+//! Raw MCP tool identities must be preserved for protocol calls, while
+//! model-visible tool names must be sanitized, deduplicated, and kept within API
+//! limits. This module owns that translation as well as the shared [`ToolInfo`]
+//! type and helpers that adjust tool schemas before exposing them to the model.
 
 use std::collections::HashMap;
 use std::collections::HashSet;
+use std::sync::Arc;
 
+use codex_config::McpServerConfig;
+use codex_protocol::ToolName;
+use rmcp::model::Tool;
+use serde::Deserialize;
+use serde::Serialize;
+use serde_json::Map;
+use serde_json::Value as JsonValue;
 use sha1::Digest;
 use sha1::Sha1;
 use tracing::warn;
 
 use crate::mcp::sanitize_responses_api_tool_name;
-use crate::mcp_connection_manager::ToolInfo;
 
-const MCP_TOOL_NAME_DELIMITER: &str = "__";
-const MAX_TOOL_NAME_LENGTH: usize = 64;
-const CALLABLE_NAME_HASH_LEN: usize = 12;
+pub(crate) const MCP_TOOLS_CACHE_WRITE_DURATION_METRIC: &str =
+    "codex.mcp.tools.cache_write.duration_ms";
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ToolInfo {
+    /// Raw MCP server name used for routing the tool call.
+    pub server_name: String,
+    /// Model-visible tool name used in Responses API tool declarations.
+    #[serde(rename = "tool_name", alias = "callable_name")]
+    pub callable_name: String,
+    /// Model-visible namespace used for deferred tool loading.
+    #[serde(rename = "tool_namespace", alias = "callable_namespace")]
+    pub callable_namespace: String,
+    /// Instructions from the MCP server initialize result.
+    #[serde(default)]
+    pub server_instructions: Option<String>,
+    /// Raw MCP tool definition; `tool.name` is sent back to the MCP server.
+    pub tool: Tool,
+    pub connector_id: Option<String>,
+    pub connector_name: Option<String>,
+    #[serde(default)]
+    pub plugin_display_names: Vec<String>,
+    pub connector_description: Option<String>,
+}
+
+impl ToolInfo {
+    pub fn canonical_tool_name(&self) -> ToolName {
+        ToolName::namespaced(self.callable_namespace.clone(), self.callable_name.clone())
+    }
+}
+
+pub fn declared_openai_file_input_param_names(
+    meta: Option<&Map<String, JsonValue>>,
+) -> Vec<String> {
+    let Some(meta) = meta else {
+        return Vec::new();
+    };
+
+    meta.get(META_OPENAI_FILE_PARAMS)
+        .and_then(JsonValue::as_array)
+        .into_iter()
+        .flatten()
+        .filter_map(JsonValue::as_str)
+        .filter(|value| !value.is_empty())
+        .map(str::to_string)
+        .collect()
+}
+
+/// A tool is allowed to be used if both are true:
+/// 1. enabled is None (no allowlist is set) or the tool is explicitly enabled.
+/// 2. The tool is not explicitly disabled.
+#[derive(Default, Clone)]
+pub(crate) struct ToolFilter {
+    pub(crate) enabled: Option<HashSet<String>>,
+    pub(crate) disabled: HashSet<String>,
+}
+
+impl ToolFilter {
+    pub(crate) fn from_config(cfg: &McpServerConfig) -> Self {
+        let enabled = cfg
+            .enabled_tools
+            .as_ref()
+            .map(|tools| tools.iter().cloned().collect::<HashSet<_>>());
+        let disabled = cfg
+            .disabled_tools
+            .as_ref()
+            .map(|tools| tools.iter().cloned().collect::<HashSet<_>>())
+            .unwrap_or_default();
+
+        Self { enabled, disabled }
+    }
+
+    pub(crate) fn allows(&self, tool_name: &str) -> bool {
+        if let Some(enabled) = &self.enabled
+            && !enabled.contains(tool_name)
+        {
+            return false;
+        }
+
+        !self.disabled.contains(tool_name)
+    }
+}
+
+/// Returns the model-visible view of a tool while preserving the raw metadata
+/// used by execution. Keep cache entries raw and call this at manager return
+/// boundaries.
+pub(crate) fn tool_with_model_visible_input_schema(tool: &Tool) -> Tool {
+    let file_params = declared_openai_file_input_param_names(tool.meta.as_deref());
+    if file_params.is_empty() {
+        return tool.clone();
+    }
+
+    let mut tool = tool.clone();
+    let mut input_schema = JsonValue::Object(tool.input_schema.as_ref().clone());
+    mask_input_schema_for_file_path_params(&mut input_schema, &file_params);
+    if let JsonValue::Object(input_schema) = input_schema {
+        tool.input_schema = Arc::new(input_schema);
+    }
+    tool
+}
+
+pub(crate) fn filter_tools(tools: Vec<ToolInfo>, filter: &ToolFilter) -> Vec<ToolInfo> {
+    tools
+        .into_iter()
+        .filter(|tool| filter.allows(&tool.tool.name))
+        .collect()
+}
 
 /// Returns a qualified-name lookup for MCP tools.
 ///
@@ -121,6 +237,57 @@ struct CallableToolCandidate {
     callable_name: String,
 }
 
+const MCP_TOOL_NAME_DELIMITER: &str = "__";
+const MAX_TOOL_NAME_LENGTH: usize = 64;
+const CALLABLE_NAME_HASH_LEN: usize = 12;
+const META_OPENAI_FILE_PARAMS: &str = "openai/fileParams";
+
+fn mask_input_schema_for_file_path_params(input_schema: &mut JsonValue, file_params: &[String]) {
+    let Some(properties) = input_schema
+        .as_object_mut()
+        .and_then(|schema| schema.get_mut("properties"))
+        .and_then(JsonValue::as_object_mut)
+    else {
+        return;
+    };
+
+    for field_name in file_params {
+        let Some(property_schema) = properties.get_mut(field_name) else {
+            continue;
+        };
+        mask_input_property_schema(property_schema);
+    }
+}
+
+fn mask_input_property_schema(schema: &mut JsonValue) {
+    let Some(object) = schema.as_object_mut() else {
+        return;
+    };
+
+    let mut description = object
+        .get("description")
+        .and_then(JsonValue::as_str)
+        .map(str::to_string)
+        .unwrap_or_default();
+    let guidance = "This parameter expects an absolute local file path. If you want to upload a file, provide the absolute path to that file here.";
+    if description.is_empty() {
+        description = guidance.to_string();
+    } else if !description.contains(guidance) {
+        description = format!("{description} {guidance}");
+    }
+
+    let is_array = object.get("type").and_then(JsonValue::as_str) == Some("array")
+        || object.get("items").is_some();
+    object.clear();
+    object.insert("description".to_string(), JsonValue::String(description));
+    if is_array {
+        object.insert("type".to_string(), JsonValue::String("array".to_string()));
+        object.insert("items".to_string(), serde_json::json!({ "type": "string" }));
+    } else {
+        object.insert("type".to_string(), JsonValue::String("string".to_string()));
+    }
+}
+
 fn sha1_hex(s: &str) -> String {
     let mut hasher = Sha1::new();
     hasher.update(s.as_bytes());

codex-rs/core/BUILD.bazel

@@ -46,7 +46,7 @@ codex_rust_crate(
         "//:AGENTS.md",
     ],
     test_shard_counts = {
-        "core-all-test": 8,
+        "core-all-test": 16,
         "core-unit-tests": 8,
     },
     test_tags = ["no-sandbox"],

codex-rs/core/src/agent/control_tests.rs

@@ -66,7 +66,6 @@ fn assistant_message(text: &str, phase: Option<MessagePhase>) -> ResponseItem {
         content: vec![ContentItem::OutputText {
             text: text.to_string(),
         }],
-        end_turn: None,
         phase,
     }
 }
@@ -519,7 +518,6 @@ async fn append_message_records_assistant_message() {
                 content: vec![ContentItem::InputText {
                     text: message.to_string(),
                 }],
-                end_turn: None,
                 phase: None,
             },
         )
@@ -678,7 +676,6 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
             content: vec![ContentItem::InputText {
                 text: "parent seed context".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         assistant_message("parent final answer", Some(MessagePhase::FinalAnswer)),

codex-rs/core/src/arc_monitor_tests.rs

@@ -65,7 +65,6 @@ async fn build_arc_monitor_request_includes_relevant_history_and_null_policies()
                 content: vec![ContentItem::InputText {
                     text: "first request".to_string(),
                 }],
-                end_turn: None,
                 phase: None,
             }],
             &turn_context,
@@ -94,7 +93,6 @@ async fn build_arc_monitor_request_includes_relevant_history_and_null_policies()
                 content: vec![ContentItem::OutputText {
                     text: "commentary".to_string(),
                 }],
-                end_turn: None,
                 phase: Some(MessagePhase::Commentary),
             }],
             &turn_context,
@@ -108,7 +106,6 @@ async fn build_arc_monitor_request_includes_relevant_history_and_null_policies()
                 content: vec![ContentItem::OutputText {
                     text: "final response".to_string(),
                 }],
-                end_turn: None,
                 phase: Some(MessagePhase::FinalAnswer),
             }],
             &turn_context,
@@ -122,7 +119,6 @@ async fn build_arc_monitor_request_includes_relevant_history_and_null_policies()
                 content: vec![ContentItem::InputText {
                     text: "latest request".to_string(),
                 }],
-                end_turn: None,
                 phase: None,
             }],
             &turn_context,
@@ -277,7 +273,6 @@ async fn monitor_action_posts_expected_arc_request() {
                 content: vec![ContentItem::InputText {
                     text: "please run the tool".to_string(),
                 }],
-                end_turn: None,
                 phase: None,
             }],
             &turn_context,
@@ -358,7 +353,6 @@ async fn monitor_action_uses_env_url_and_token_overrides() {
                 content: vec![ContentItem::InputText {
                     text: "please run the tool".to_string(),
                 }],
-                end_turn: None,
                 phase: None,
             }],
             &turn_context,
@@ -428,7 +422,6 @@ async fn monitor_action_rejects_legacy_response_fields() {
                 content: vec![ContentItem::InputText {
                     text: "please run the tool".to_string(),
                 }],
-                end_turn: None,
                 phase: None,
             }],
             &turn_context,

codex-rs/core/src/codex_thread.rs

@@ -278,7 +278,6 @@ impl CodexThread {
             id: None,
             role: "user".to_string(),
             content: vec![ContentItem::InputText { text: message }],
-            end_turn: None,
             phase: None,
         };
         let pending_item = match pending_message_input_item(&message) {

codex-rs/core/src/compact.rs

@@ -509,7 +509,6 @@ fn build_compacted_history_with_limit(
             content: vec![ContentItem::InputText {
                 text: message.clone(),
             }],
-            end_turn: None,
             phase: None,
         });
     }
@@ -524,7 +523,6 @@ fn build_compacted_history_with_limit(
         id: None,
         role: "user".to_string(),
         content: vec![ContentItem::InputText { text: summary_text }],
-        end_turn: None,
         phase: None,
     });
 

codex-rs/core/src/compact_tests.rs

@@ -63,7 +63,6 @@ fn collect_user_messages_extracts_user_text_only() {
             content: vec![ContentItem::OutputText {
                 text: "ignored".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -72,7 +71,6 @@ fn collect_user_messages_extracts_user_text_only() {
             content: vec![ContentItem::InputText {
                 text: "first".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Other,
@@ -97,7 +95,6 @@ do things
 </INSTRUCTIONS>"#
                     .to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -106,7 +103,6 @@ do things
             content: vec![ContentItem::InputText {
                 text: "<ENVIRONMENT_CONTEXT>cwd=/tmp</ENVIRONMENT_CONTEXT>".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -115,7 +111,6 @@ do things
             content: vec![ContentItem::InputText {
                 text: "real user message".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
     ];
@@ -223,7 +218,6 @@ async fn process_compacted_history_replaces_developer_messages() {
             content: vec![ContentItem::InputText {
                 text: "stale permissions".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -232,7 +226,6 @@ async fn process_compacted_history_replaces_developer_messages() {
             content: vec![ContentItem::InputText {
                 text: "summary".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -241,7 +234,6 @@ async fn process_compacted_history_replaces_developer_messages() {
             content: vec![ContentItem::InputText {
                 text: "stale personality".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
     ];
@@ -256,7 +248,6 @@ async fn process_compacted_history_replaces_developer_messages() {
         content: vec![ContentItem::InputText {
             text: "summary".to_string(),
         }],
-        end_turn: None,
         phase: None,
     });
     assert_eq!(refreshed, expected);
@@ -270,7 +261,6 @@ async fn process_compacted_history_reinjects_full_initial_context() {
         content: vec![ContentItem::InputText {
             text: "summary".to_string(),
         }],
-        end_turn: None,
         phase: None,
     }];
     let (refreshed, mut expected) = process_compacted_history_with_test_session(
@@ -284,7 +274,6 @@ async fn process_compacted_history_reinjects_full_initial_context() {
         content: vec![ContentItem::InputText {
             text: "summary".to_string(),
         }],
-        end_turn: None,
         phase: None,
     });
     assert_eq!(refreshed, expected);
@@ -304,7 +293,6 @@ keep me updated
 </INSTRUCTIONS>"#
                     .to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -317,7 +305,6 @@ keep me updated
 </environment_context>"#
                     .to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -330,7 +317,6 @@ keep me updated
 </turn_aborted>"#
                     .to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -339,7 +325,6 @@ keep me updated
             content: vec![ContentItem::InputText {
                 text: "summary".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -348,7 +333,6 @@ keep me updated
             content: vec![ContentItem::InputText {
                 text: "stale developer instructions".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
     ];
@@ -363,7 +347,6 @@ keep me updated
         content: vec![ContentItem::InputText {
             text: "summary".to_string(),
         }],
-        end_turn: None,
         phase: None,
     });
     assert_eq!(refreshed, expected);
@@ -378,7 +361,6 @@ async fn process_compacted_history_inserts_context_before_last_real_user_message
             content: vec![ContentItem::InputText {
                 text: "older user".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -387,7 +369,6 @@ async fn process_compacted_history_inserts_context_before_last_real_user_message
             content: vec![ContentItem::InputText {
                 text: format!("{SUMMARY_PREFIX}\nsummary text"),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -396,7 +377,6 @@ async fn process_compacted_history_inserts_context_before_last_real_user_message
             content: vec![ContentItem::InputText {
                 text: "latest user".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
     ];
@@ -413,7 +393,6 @@ async fn process_compacted_history_inserts_context_before_last_real_user_message
             content: vec![ContentItem::InputText {
                 text: "older user".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -422,7 +401,6 @@ async fn process_compacted_history_inserts_context_before_last_real_user_message
             content: vec![ContentItem::InputText {
                 text: format!("{SUMMARY_PREFIX}\nsummary text"),
             }],
-            end_turn: None,
             phase: None,
         },
     ];
@@ -433,7 +411,6 @@ async fn process_compacted_history_inserts_context_before_last_real_user_message
         content: vec![ContentItem::InputText {
             text: "latest user".to_string(),
         }],
-        end_turn: None,
         phase: None,
     });
     assert_eq!(refreshed, expected);
@@ -447,7 +424,6 @@ async fn process_compacted_history_reinjects_model_switch_message() {
         content: vec![ContentItem::InputText {
             text: "summary".to_string(),
         }],
-        end_turn: None,
         phase: None,
     }];
     let previous_turn_settings = PreviousTurnSettings {
@@ -477,7 +453,6 @@ async fn process_compacted_history_reinjects_model_switch_message() {
         content: vec![ContentItem::InputText {
             text: "summary".to_string(),
         }],
-        end_turn: None,
         phase: None,
     });
     assert_eq!(refreshed, expected);
@@ -492,7 +467,6 @@ fn insert_initial_context_before_last_real_user_or_summary_keeps_summary_last()
             content: vec![ContentItem::InputText {
                 text: "older user".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -501,7 +475,6 @@ fn insert_initial_context_before_last_real_user_or_summary_keeps_summary_last()
             content: vec![ContentItem::InputText {
                 text: "latest user".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -510,7 +483,6 @@ fn insert_initial_context_before_last_real_user_or_summary_keeps_summary_last()
             content: vec![ContentItem::InputText {
                 text: format!("{SUMMARY_PREFIX}\nsummary text"),
             }],
-            end_turn: None,
             phase: None,
         },
     ];
@@ -520,7 +492,6 @@ fn insert_initial_context_before_last_real_user_or_summary_keeps_summary_last()
         content: vec![ContentItem::InputText {
             text: "fresh permissions".to_string(),
         }],
-        end_turn: None,
         phase: None,
     }];
 
@@ -533,7 +504,6 @@ fn insert_initial_context_before_last_real_user_or_summary_keeps_summary_last()
             content: vec![ContentItem::InputText {
                 text: "older user".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -542,7 +512,6 @@ fn insert_initial_context_before_last_real_user_or_summary_keeps_summary_last()
             content: vec![ContentItem::InputText {
                 text: "fresh permissions".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -551,7 +520,6 @@ fn insert_initial_context_before_last_real_user_or_summary_keeps_summary_last()
             content: vec![ContentItem::InputText {
                 text: "latest user".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -560,7 +528,6 @@ fn insert_initial_context_before_last_real_user_or_summary_keeps_summary_last()
             content: vec![ContentItem::InputText {
                 text: format!("{SUMMARY_PREFIX}\nsummary text"),
             }],
-            end_turn: None,
             phase: None,
         },
     ];
@@ -578,7 +545,6 @@ fn insert_initial_context_before_last_real_user_or_summary_keeps_compaction_last
         content: vec![ContentItem::InputText {
             text: "fresh permissions".to_string(),
         }],
-        end_turn: None,
         phase: None,
     }];
 
@@ -591,7 +557,6 @@ fn insert_initial_context_before_last_real_user_or_summary_keeps_compaction_last
             content: vec![ContentItem::InputText {
                 text: "fresh permissions".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Compaction {

codex-rs/core/src/context/fragment.rs

@@ -81,7 +81,6 @@ pub trait ContextualUserFragment {
             content: vec![ContentItem::InputText {
                 text: self.render(),
             }],
-            end_turn: None,
             phase: None,
         }
     }

codex-rs/core/src/context_manager/history_tests.rs

@@ -41,7 +41,6 @@ fn assistant_msg(text: &str) -> ResponseItem {
         content: vec![ContentItem::OutputText {
             text: text.to_string(),
         }],
-        end_turn: None,
         phase: None,
     }
 }
@@ -60,7 +59,6 @@ fn inter_agent_assistant_msg(text: &str) -> ResponseItem {
         content: vec![ContentItem::OutputText {
             text: serde_json::to_string(&communication).unwrap(),
         }],
-        end_turn: None,
         phase: None,
     }
 }
@@ -80,7 +78,6 @@ fn user_msg(text: &str) -> ResponseItem {
         content: vec![ContentItem::OutputText {
             text: text.to_string(),
         }],
-        end_turn: None,
         phase: None,
     }
 }
@@ -92,7 +89,6 @@ fn user_input_text_msg(text: &str) -> ResponseItem {
         content: vec![ContentItem::InputText {
             text: text.to_string(),
         }],
-        end_turn: None,
         phase: None,
     }
 }
@@ -104,7 +100,6 @@ fn developer_msg(text: &str) -> ResponseItem {
         content: vec![ContentItem::InputText {
             text: text.to_string(),
         }],
-        end_turn: None,
         phase: None,
     }
 }
@@ -119,7 +114,6 @@ fn developer_msg_with_fragments(texts: &[&str]) -> ResponseItem {
                 text: (*text).to_string(),
             })
             .collect(),
-        end_turn: None,
         phase: None,
     }
 }
@@ -200,7 +194,6 @@ fn filters_non_api_messages() {
         content: vec![ContentItem::OutputText {
             text: "ignored".to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
     let reasoning = reasoning_msg("thinking...");
@@ -231,7 +224,6 @@ fn filters_non_api_messages() {
                 content: vec![ContentItem::OutputText {
                     text: "hi".to_string()
                 }],
-                end_turn: None,
                 phase: None,
             },
             ResponseItem::Message {
@@ -240,7 +232,6 @@ fn filters_non_api_messages() {
                 content: vec![ContentItem::OutputText {
                     text: "hello".to_string()
                 }],
-                end_turn: None,
                 phase: None,
             }
         ]
@@ -390,7 +381,6 @@ fn for_prompt_strips_images_when_model_does_not_support_images() {
                     text: "caption".to_string(),
                 },
             ],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::FunctionCall {
@@ -453,7 +443,6 @@ fn for_prompt_strips_images_when_model_does_not_support_images() {
                     text: "caption".to_string(),
                 },
             ],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::FunctionCall {
@@ -512,7 +501,6 @@ fn for_prompt_strips_images_when_model_does_not_support_images() {
                 detail: Some(DEFAULT_IMAGE_DETAIL),
             },
         ],
-        end_turn: None,
         phase: None,
     }]);
     let preserved = with_images.for_prompt(&modalities);
@@ -540,7 +528,6 @@ fn for_prompt_preserves_image_generation_calls_when_images_are_supported() {
             content: vec![ContentItem::InputText {
                 text: "hi".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
     ]);
@@ -560,7 +547,6 @@ fn for_prompt_preserves_image_generation_calls_when_images_are_supported() {
                 content: vec![ContentItem::InputText {
                     text: "hi".to_string(),
                 }],
-                end_turn: None,
                 phase: None,
             }
         ]
@@ -576,7 +562,6 @@ fn for_prompt_clears_image_generation_result_when_images_are_unsupported() {
             content: vec![ContentItem::InputText {
                 text: "generate a lobster".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::ImageGenerationCall {
@@ -596,7 +581,6 @@ fn for_prompt_clears_image_generation_result_when_images_are_unsupported() {
                 content: vec![ContentItem::InputText {
                     text: "generate a lobster".to_string(),
                 }],
-                end_turn: None,
                 phase: None,
             },
             ResponseItem::ImageGenerationCall {
@@ -758,7 +742,6 @@ fn replace_last_turn_images_does_not_touch_user_images() {
             image_url: "data:image/png;base64,AAA".to_string(),
             detail: Some(DEFAULT_IMAGE_DETAIL),
         }],
-        end_turn: None,
         phase: None,
     }];
     let mut history = create_history_with_items(items.clone());
@@ -1690,7 +1673,6 @@ fn image_data_url_payload_does_not_dominate_message_estimate() {
                 detail: Some(DEFAULT_IMAGE_DETAIL),
             },
         ],
-        end_turn: None,
         phase: None,
     };
     let text_only_item = ResponseItem::Message {
@@ -1699,7 +1681,6 @@ fn image_data_url_payload_does_not_dominate_message_estimate() {
         content: vec![ContentItem::InputText {
             text: "Here is the screenshot".to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
 
@@ -1773,7 +1754,6 @@ fn non_base64_image_urls_are_unchanged() {
             image_url: "https://example.com/foo.png".to_string(),
             detail: Some(DEFAULT_IMAGE_DETAIL),
         }],
-        end_turn: None,
         phase: None,
     };
     let function_output_item = ResponseItem::FunctionCallOutput {
@@ -1805,7 +1785,6 @@ fn data_url_without_base64_marker_is_unchanged() {
             image_url: "data:image/svg+xml,<svg xmlns='http://www.w3.org/2000/svg'/>".to_string(),
             detail: Some(DEFAULT_IMAGE_DETAIL),
         }],
-        end_turn: None,
         phase: None,
     };
 
@@ -1846,7 +1825,6 @@ fn mixed_case_data_url_markers_are_adjusted() {
             image_url,
             detail: Some(DEFAULT_IMAGE_DETAIL),
         }],
-        end_turn: None,
         phase: None,
     };
 
@@ -1879,7 +1857,6 @@ fn multiple_inline_images_apply_multiple_fixed_costs() {
                 detail: Some(DEFAULT_IMAGE_DETAIL),
             },
         ],
-        end_turn: None,
         phase: None,
     };
 
@@ -1962,7 +1939,6 @@ fn text_only_items_unchanged() {
         content: vec![ContentItem::OutputText {
             text: "Hello world, this is a response.".to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
 

codex-rs/core/src/context_manager/updates.rs

@@ -197,7 +197,6 @@ fn build_text_message(role: &str, text_sections: Vec<String>) -> Option<Response
         id: None,
         role: role.to_string(),
         content,
-        end_turn: None,
         phase: None,
     })
 }

codex-rs/core/src/event_mapping_tests.rs

@@ -34,7 +34,6 @@ fn parses_user_message_with_text_and_two_images() {
                 detail: Some(DEFAULT_IMAGE_DETAIL),
             },
         ],
-        end_turn: None,
         phase: None,
     };
 
@@ -78,7 +77,6 @@ fn skips_local_image_label_text() {
                 text: user_text.clone(),
             },
         ],
-        end_turn: None,
         phase: None,
     };
 
@@ -108,7 +106,6 @@ fn parses_assistant_message_input_text_for_backward_compatibility() {
             text: "author: /root\nrecipient: /root/worker\nother_recipients: []\nContent: continue"
                 .to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
 
@@ -158,7 +155,6 @@ fn skips_unnamed_image_label_text() {
                 text: user_text.clone(),
             },
         ],
-        end_turn: None,
         phase: None,
     };
 
@@ -188,7 +184,6 @@ fn skips_user_instructions_and_env() {
                 content: vec![ContentItem::InputText {
                     text: "# AGENTS.md instructions for test_directory\n\n<INSTRUCTIONS>\ntest_text\n</INSTRUCTIONS>".to_string(),
                 }],
-                end_turn: None,
             phase: None,
             },
             ResponseItem::Message {
@@ -197,7 +192,6 @@ fn skips_user_instructions_and_env() {
                 content: vec![ContentItem::InputText {
                     text: "<environment_context>test_text</environment_context>".to_string(),
                 }],
-                end_turn: None,
             phase: None,
             },
             ResponseItem::Message {
@@ -206,7 +200,6 @@ fn skips_user_instructions_and_env() {
                 content: vec![ContentItem::InputText {
                     text: "# AGENTS.md instructions for test_directory\n\n<INSTRUCTIONS>\ntest_text\n</INSTRUCTIONS>".to_string(),
                 }],
-                end_turn: None,
             phase: None,
             },
             ResponseItem::Message {
@@ -216,7 +209,6 @@ fn skips_user_instructions_and_env() {
                     text: "<skill>\n<name>demo</name>\n<path>skills/demo/SKILL.md</path>\nbody\n</skill>"
                         .to_string(),
                 }],
-                end_turn: None,
             phase: None,
             },
             ResponseItem::Message {
@@ -225,7 +217,6 @@ fn skips_user_instructions_and_env() {
                 content: vec![ContentItem::InputText {
                     text: "<user_shell_command>echo 42</user_shell_command>".to_string(),
                 }],
-                end_turn: None,
             phase: None,
             },
             ResponseItem::Message {
@@ -241,7 +232,6 @@ fn skips_user_instructions_and_env() {
                                 .to_string(),
                     },
                 ],
-                end_turn: None,
                 phase: None,
             },
         ];
@@ -292,7 +282,6 @@ fn parses_hook_prompt_and_hides_other_contextual_fragments() {
                         .to_string(),
             },
         ],
-        end_turn: None,
         phase: None,
     };
 
@@ -321,7 +310,6 @@ fn parses_agent_message() {
         content: vec![ContentItem::OutputText {
             text: "Hello from Codex".to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
 

codex-rs/core/src/guardian/tests.rs

@@ -176,7 +176,6 @@ async fn seed_guardian_parent_history(session: &Arc<Session>, turn: &Arc<TurnCon
                         text: "Please check the repo visibility and push the docs fix if needed."
                             .to_string(),
                     }],
-                    end_turn: None,
                     phase: None,
                 },
                 ResponseItem::FunctionCall {
@@ -199,7 +198,6 @@ async fn seed_guardian_parent_history(session: &Arc<Session>, turn: &Arc<TurnCon
                         text: "The repo is public; I now need approval to push the docs fix."
                             .to_string(),
                     }],
-                    end_turn: None,
                     phase: None,
                 },
             ],
@@ -343,7 +341,6 @@ async fn build_guardian_prompt_delta_mode_preserves_original_numbering() -> anyh
                     content: vec![ContentItem::InputText {
                         text: "Please also push the second docs fix.".to_string(),
                     }],
-                    end_turn: None,
                     phase: None,
                 },
                 ResponseItem::Message {
@@ -352,7 +349,6 @@ async fn build_guardian_prompt_delta_mode_preserves_original_numbering() -> anyh
                     content: vec![ContentItem::OutputText {
                         text: "I need approval for the second push.".to_string(),
                     }],
-                    end_turn: None,
                     phase: None,
                 },
             ],
@@ -476,7 +472,6 @@ async fn build_guardian_prompt_stale_delta_version_falls_back_to_full_prompt() -
                     content: vec![ContentItem::InputText {
                         text: "Compacted retained user request.".to_string(),
                     }],
-                    end_turn: None,
                     phase: None,
                 },
                 ResponseItem::Message {
@@ -485,7 +480,6 @@ async fn build_guardian_prompt_stale_delta_version_falls_back_to_full_prompt() -
                     content: vec![ContentItem::OutputText {
                         text: "Compacted summary of earlier guardian context.".to_string(),
                     }],
-                    end_turn: None,
                     phase: None,
                 },
             ],
@@ -501,7 +495,6 @@ async fn build_guardian_prompt_stale_delta_version_falls_back_to_full_prompt() -
                     content: vec![ContentItem::InputText {
                         text: "Please push after the compaction.".to_string(),
                     }],
-                    end_turn: None,
                     phase: None,
                 },
                 ResponseItem::Message {
@@ -510,7 +503,6 @@ async fn build_guardian_prompt_stale_delta_version_falls_back_to_full_prompt() -
                     content: vec![ContentItem::OutputText {
                         text: "I need approval for the post-compaction push.".to_string(),
                     }],
-                    end_turn: None,
                     phase: None,
                 },
             ],
@@ -559,7 +551,6 @@ fn collect_guardian_transcript_entries_skips_contextual_user_messages() {
             content: vec![ContentItem::InputText {
                 text: "<environment_context>\n<cwd>/tmp</cwd>\n</environment_context>".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Message {
@@ -568,7 +559,6 @@ fn collect_guardian_transcript_entries_skips_contextual_user_messages() {
             content: vec![ContentItem::OutputText {
                 text: "hello".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
     ];
@@ -594,7 +584,6 @@ fn collect_guardian_transcript_entries_includes_recent_tool_calls_and_output() {
             content: vec![ContentItem::InputText {
                 text: "check the repo".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::FunctionCall {
@@ -616,7 +605,6 @@ fn collect_guardian_transcript_entries_includes_recent_tool_calls_and_output() {
             content: vec![ContentItem::OutputText {
                 text: "I need to push a fix".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
     ];
@@ -1357,7 +1345,6 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow:
                     content: vec![ContentItem::InputText {
                         text: "Please push the second docs fix too.".to_string(),
                     }],
-                    end_turn: None,
                     phase: None,
                 },
                 ResponseItem::Message {
@@ -1366,7 +1353,6 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow:
                     content: vec![ContentItem::OutputText {
                         text: "I need approval for the second docs fix.".to_string(),
                     }],
-                    end_turn: None,
                     phase: None,
                 },
             ],
@@ -1403,7 +1389,6 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow:
                     content: vec![ContentItem::InputText {
                         text: "Please push the third docs fix too.".to_string(),
                     }],
-                    end_turn: None,
                     phase: None,
                 },
                 ResponseItem::Message {
@@ -1412,7 +1397,6 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow:
                     content: vec![ContentItem::OutputText {
                         text: "I need approval for the third docs fix.".to_string(),
                     }],
-                    end_turn: None,
                     phase: None,
                 },
             ],
@@ -1790,7 +1774,6 @@ async fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> a
                         content: vec![ContentItem::InputText {
                             text: "Please inspect pending changes before pushing.".to_string(),
                         }],
-                        end_turn: None,
                         phase: None,
                     },
                     ResponseItem::Message {
@@ -1799,7 +1782,6 @@ async fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> a
                         content: vec![ContentItem::OutputText {
                             text: "I need approval to run git diff.".to_string(),
                         }],
-                        end_turn: None,
                         phase: None,
                     },
                 ],
@@ -1859,7 +1841,6 @@ async fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> a
                         content: vec![ContentItem::InputText {
                             text: "Now inspect whether pushing is safe.".to_string(),
                         }],
-                        end_turn: None,
                         phase: None,
                     },
                     ResponseItem::Message {
@@ -1868,7 +1849,6 @@ async fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> a
                         content: vec![ContentItem::OutputText {
                             text: "I need approval to push after the diff check.".to_string(),
                         }],
-                        end_turn: None,
                         phase: None,
                     },
                 ],

codex-rs/core/src/memories/phase1.rs

@@ -332,7 +332,6 @@ mod job {
                         &rollout_contents,
                     )?,
                 }],
-                end_turn: None,
                 phase: None,
             }],
             tools: Vec::new(),
@@ -491,7 +490,6 @@ mod job {
             id,
             role,
             content,
-            end_turn,
             phase,
         } = item
         else {
@@ -519,7 +517,6 @@ mod job {
             id: id.clone(),
             role: role.clone(),
             content,
-            end_turn: *end_turn,
             phase: phase.clone(),
         })
     }

codex-rs/core/src/memories/phase1_tests.rs

@@ -24,7 +24,6 @@ fn serializes_memory_rollout_with_agents_removed_but_environment_kept() {
                 text: "<environment_context>\n<cwd>/tmp</cwd>\n</environment_context>".to_string(),
             },
         ],
-        end_turn: None,
         phase: None,
     };
     let skill_message = ResponseItem::Message {
@@ -34,7 +33,6 @@ fn serializes_memory_rollout_with_agents_removed_but_environment_kept() {
             text: "<skill>\n<name>demo</name>\n<path>skills/demo/SKILL.md</path>\nbody\n</skill>"
                 .to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
     let subagent_message = ResponseItem::Message {
@@ -44,7 +42,6 @@ fn serializes_memory_rollout_with_agents_removed_but_environment_kept() {
             text: "<subagent_notification>{\"agent_id\":\"a\",\"status\":\"completed\"}</subagent_notification>"
                 .to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
 
@@ -66,7 +63,6 @@ fn serializes_memory_rollout_with_agents_removed_but_environment_kept() {
                     text: "<environment_context>\n<cwd>/tmp</cwd>\n</environment_context>"
                         .to_string(),
                 }],
-                end_turn: None,
                 phase: None,
             },
             subagent_message,

codex-rs/core/src/realtime_context_tests.rs

@@ -70,7 +70,6 @@ fn message(role: &str, content: ContentItem) -> ResponseItem {
         id: None,
         role: role.to_string(),
         content: vec![content],
-        end_turn: None,
         phase: None,
     }
 }

codex-rs/core/src/session/mod.rs

@@ -2364,7 +2364,6 @@ impl Session {
             content: vec![ContentItem::InputText {
                 text: format!("Warning: {}", message.into()),
             }],
-            end_turn: None,
             phase: None,
         };
 

codex-rs/core/src/session/rollout_reconstruction_tests.rs

@@ -19,7 +19,6 @@ fn user_message(text: &str) -> ResponseItem {
         content: vec![ContentItem::InputText {
             text: text.to_string(),
         }],
-        end_turn: None,
         phase: None,
     }
 }
@@ -31,7 +30,6 @@ fn assistant_message(text: &str) -> ResponseItem {
         content: vec![ContentItem::OutputText {
             text: text.to_string(),
         }],
-        end_turn: None,
         phase: None,
     }
 }
@@ -50,7 +48,6 @@ fn inter_agent_assistant_message(text: &str) -> ResponseItem {
         content: vec![ContentItem::OutputText {
             text: serde_json::to_string(&communication).unwrap(),
         }],
-        end_turn: None,
         phase: None,
     }
 }

codex-rs/core/src/session/tests.rs

@@ -175,7 +175,6 @@ fn user_message(text: &str) -> ResponseItem {
         content: vec![ContentItem::InputText {
             text: text.to_string(),
         }],
-        end_turn: None,
         phase: None,
     }
 }
@@ -187,7 +186,6 @@ fn assistant_message(text: &str) -> ResponseItem {
         content: vec![ContentItem::OutputText {
             text: text.to_string(),
         }],
-        end_turn: None,
         phase: None,
     }
 }
@@ -247,7 +245,6 @@ fn skill_message(text: &str) -> ResponseItem {
         content: vec![ContentItem::InputText {
             text: text.to_string(),
         }],
-        end_turn: None,
         phase: None,
     }
 }
@@ -1270,7 +1267,6 @@ async fn reconstruct_history_uses_replacement_history_verbatim() {
         content: vec![ContentItem::InputText {
             text: "summary".to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
     let replacement_history = vec![
@@ -1281,7 +1277,6 @@ async fn reconstruct_history_uses_replacement_history_verbatim() {
             content: vec![ContentItem::InputText {
                 text: "stale developer instructions".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
     ];
@@ -5733,7 +5728,6 @@ async fn record_context_updates_and_set_reference_context_item_reinjects_full_co
         content: vec![ContentItem::InputText {
             text: format!("{}\nsummary", crate::compact::SUMMARY_PREFIX),
         }],
-        end_turn: None,
         phase: None,
     };
     session
@@ -6270,7 +6264,6 @@ async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input()
         content: vec![ContentItem::InputText {
             text: "late pending input".to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
     assert!(
@@ -7465,7 +7458,6 @@ async fn sample_rollout(
         content: vec![ContentItem::InputText {
             text: "first user".to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
     live_history.record_items(
@@ -7480,7 +7472,6 @@ async fn sample_rollout(
         content: vec![ContentItem::OutputText {
             text: "assistant reply one".to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
     live_history.record_items(
@@ -7507,7 +7498,6 @@ async fn sample_rollout(
         content: vec![ContentItem::InputText {
             text: "second user".to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
     live_history.record_items(
@@ -7522,7 +7512,6 @@ async fn sample_rollout(
         content: vec![ContentItem::OutputText {
             text: "assistant reply two".to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
     live_history.record_items(
@@ -7549,7 +7538,6 @@ async fn sample_rollout(
         content: vec![ContentItem::InputText {
             text: "third user".to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
     live_history.record_items(
@@ -7564,7 +7552,6 @@ async fn sample_rollout(
         content: vec![ContentItem::OutputText {
             text: "assistant reply three".to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
     live_history.record_items(

codex-rs/core/src/session/tests/guardian_tests.rs

@@ -552,7 +552,6 @@ async fn process_compacted_history_preserves_separate_guardian_developer_message
                 content: vec![ContentItem::InputText {
                     text: "stale developer message".to_string(),
                 }],
-                end_turn: None,
                 phase: None,
             },
             ResponseItem::Message {
@@ -561,7 +560,6 @@ async fn process_compacted_history_preserves_separate_guardian_developer_message
                 content: vec![ContentItem::InputText {
                     text: "summary".to_string(),
                 }],
-                end_turn: None,
                 phase: None,
             },
         ],

codex-rs/core/src/stream_events_utils_tests.rs

@@ -28,7 +28,6 @@ fn assistant_output_text_with_phase(text: &str, phase: Option<MessagePhase>) ->
         content: vec![ContentItem::OutputText {
             text: text.to_string(),
         }],
-        end_turn: Some(true),
         phase,
     }
 }

codex-rs/core/src/tasks/mod.rs

@@ -104,7 +104,6 @@ pub(crate) fn interrupted_turn_history_marker(
                 content: vec![ContentItem::InputText {
                     text: marker.render(),
                 }],
-                end_turn: None,
                 phase: None,
             })
         }

codex-rs/core/src/tasks/review.rs

@@ -249,7 +249,6 @@ pub(crate) async fn exit_review_mode(
                 id: Some(REVIEW_USER_MESSAGE_ID.to_string()),
                 role: "user".to_string(),
                 content: vec![ContentItem::InputText { text: user_message }],
-                end_turn: None,
                 phase: None,
             }],
         )
@@ -270,7 +269,6 @@ pub(crate) async fn exit_review_mode(
                 content: vec![ContentItem::OutputText {
                     text: assistant_message,
                 }],
-                end_turn: None,
                 phase: None,
             },
         )

codex-rs/core/src/thread_manager_tests.rs

@@ -30,7 +30,6 @@ fn user_msg(text: &str) -> ResponseItem {
         content: vec![ContentItem::OutputText {
             text: text.to_string(),
         }],
-        end_turn: None,
         phase: None,
     }
 }
@@ -41,7 +40,6 @@ fn assistant_msg(text: &str) -> ResponseItem {
         content: vec![ContentItem::OutputText {
             text: text.to_string(),
         }],
-        end_turn: None,
         phase: None,
     }
 }

codex-rs/core/src/thread_rollout_truncation_tests.rs

@@ -14,7 +14,6 @@ fn user_msg(text: &str) -> ResponseItem {
         content: vec![ContentItem::OutputText {
             text: text.to_string(),
         }],
-        end_turn: None,
         phase: None,
     }
 }
@@ -26,7 +25,6 @@ fn assistant_msg(text: &str) -> ResponseItem {
         content: vec![ContentItem::OutputText {
             text: text.to_string(),
         }],
-        end_turn: None,
         phase: None,
     }
 }

codex-rs/core/src/tools/handlers/multi_agents_tests.rs

@@ -2503,7 +2503,6 @@ async fn resume_agent_restores_closed_agent_and_accepts_send_input() {
                 content: vec![ContentItem::InputText {
                     text: "materialized".to_string(),
                 }],
-                end_turn: None,
                 phase: None,
             })]),
             AuthManager::from_auth_for_testing(CodexAuth::from_api_key("dummy")),

codex-rs/core/src/turn_timing_tests.rs

@@ -102,7 +102,6 @@ fn response_item_records_turn_ttft_for_first_output_signals() {
         content: vec![ContentItem::OutputText {
             text: "hello".to_string(),
         }],
-        end_turn: None,
         phase: None,
     }));
 }
@@ -115,7 +114,6 @@ fn response_item_records_turn_ttft_ignores_empty_non_output_items() {
         content: vec![ContentItem::OutputText {
             text: String::new(),
         }],
-        end_turn: None,
         phase: None,
     }));
     assert!(!response_item_records_turn_ttft(

codex-rs/core/tests/common/responses.rs

@@ -686,7 +686,6 @@ pub fn user_message_item(text: &str) -> ResponseItem {
         content: vec![ContentItem::InputText {
             text: text.to_string(),
         }],
-        end_turn: None,
         phase: None,
     }
 }

codex-rs/core/tests/responses_headers.rs

@@ -118,7 +118,6 @@ async fn responses_stream_includes_subagent_header_on_review() {
         content: vec![ContentItem::InputText {
             text: "hello".into(),
         }],
-        end_turn: None,
         phase: None,
     }];
 
@@ -245,7 +244,6 @@ async fn responses_stream_includes_subagent_header_on_other() {
         content: vec![ContentItem::InputText {
             text: "hello".into(),
         }],
-        end_turn: None,
         phase: None,
     }];
 
@@ -361,7 +359,6 @@ async fn responses_respects_model_info_overrides_from_config() {
         content: vec![ContentItem::InputText {
             text: "hello".into(),
         }],
-        end_turn: None,
         phase: None,
     }];
 

codex-rs/core/tests/suite/client.rs

@@ -290,7 +290,6 @@ async fn resume_includes_initial_messages_and_sends_prior_items() {
         content: vec![codex_protocol::models::ContentItem::InputText {
             text: "resumed user message".to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
     let prior_user_json = serde_json::to_value(&prior_user).unwrap();
@@ -312,7 +311,6 @@ async fn resume_includes_initial_messages_and_sends_prior_items() {
         content: vec![codex_protocol::models::ContentItem::OutputText {
             text: "resumed system instruction".to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
     let prior_system_json = serde_json::to_value(&prior_system).unwrap();
@@ -334,7 +332,6 @@ async fn resume_includes_initial_messages_and_sends_prior_items() {
         content: vec![codex_protocol::models::ContentItem::OutputText {
             text: "resumed assistant message".to_string(),
         }],
-        end_turn: None,
         phase: Some(MessagePhase::Commentary),
     };
     let prior_item_json = serde_json::to_value(&prior_item).unwrap();
@@ -517,7 +514,6 @@ async fn resume_replays_legacy_js_repl_image_rollout_shapes() {
                     image_url: legacy_image_url.to_string(),
                     detail: Some(DEFAULT_IMAGE_DETAIL),
                 }],
-                end_turn: None,
                 phase: None,
             }),
         },
@@ -903,7 +899,6 @@ async fn send_provider_auth_request(server: &MockServer, auth: ModelProviderAuth
         content: vec![ContentItem::InputText {
             text: "hello".to_string(),
         }],
-        end_turn: None,
         phase: None,
     });
 
@@ -2318,7 +2313,6 @@ async fn azure_responses_request_includes_store_and_reasoning_ids() {
         content: vec![ContentItem::OutputText {
             text: "message".into(),
         }],
-        end_turn: None,
         phase: None,
     });
     prompt.input.push(ResponseItem::WebSearchCall {

codex-rs/core/tests/suite/client_websockets.rs

@@ -1700,7 +1700,6 @@ fn message_item(text: &str) -> ResponseItem {
         id: None,
         role: "user".into(),
         content: vec![ContentItem::InputText { text: text.into() }],
-        end_turn: None,
         phase: None,
     }
 }
@@ -1710,7 +1709,6 @@ fn assistant_message_item(id: &str, text: &str) -> ResponseItem {
         id: Some(id.to_string()),
         role: "assistant".into(),
         content: vec![ContentItem::OutputText { text: text.into() }],
-        end_turn: None,
         phase: None,
     }
 }

codex-rs/core/tests/suite/compact.rs

@@ -1615,7 +1615,6 @@ async fn auto_compact_runs_after_resume_when_token_usage_is_over_limit() {
             content: vec![codex_protocol::models::ContentItem::OutputText {
                 text: remote_summary.to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         codex_protocol::models::ResponseItem::Compaction {
@@ -2861,7 +2860,6 @@ async fn auto_compact_counts_encrypted_reasoning_before_last_user() {
             content: vec![codex_protocol::models::ContentItem::OutputText {
                 text: "REMOTE_COMPACT_SUMMARY".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         codex_protocol::models::ResponseItem::Compaction {
@@ -2985,7 +2983,6 @@ async fn auto_compact_runs_when_reasoning_header_clears_between_turns() {
             content: vec![codex_protocol::models::ContentItem::OutputText {
                 text: "REMOTE_COMPACT_SUMMARY".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         codex_protocol::models::ResponseItem::Compaction {

codex-rs/core/tests/suite/compact_remote.rs

@@ -1181,7 +1181,6 @@ async fn remote_compact_persists_replacement_history_in_rollout() -> Result<()>
             content: vec![ContentItem::OutputText {
                 text: "COMPACTED_ASSISTANT_NOTE".to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
     ];
@@ -1320,7 +1319,6 @@ async fn remote_compact_and_resume_refresh_stale_developer_instructions() -> Res
             content: vec![ContentItem::InputText {
                 text: stale_developer_message.to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Compaction {
@@ -1458,7 +1456,6 @@ async fn remote_compact_refreshes_stale_developer_instructions_without_resume()
             content: vec![ContentItem::InputText {
                 text: stale_developer_message.to_string(),
             }],
-            end_turn: None,
             phase: None,
         },
         ResponseItem::Compaction {

codex-rs/core/tests/suite/image_rollout.rs

@@ -164,7 +164,6 @@ async fn copy_paste_local_image_persists_rollout_request_shape() -> anyhow::Resu
                 text: "pasted image".to_string(),
             },
         ],
-        end_turn: None,
         phase: None,
     };
 
@@ -253,7 +252,6 @@ async fn drag_drop_image_persists_rollout_request_shape() -> anyhow::Result<()>
                 text: "dropped image".to_string(),
             },
         ],
-        end_turn: None,
         phase: None,
     };
 

codex-rs/core/tests/suite/prompt_debug_tests.rs

@@ -38,7 +38,6 @@ async fn build_prompt_input_includes_context_and_user_message() -> Result<()> {
         content: vec![ContentItem::InputText {
             text: "hello from debug prompt".to_string(),
         }],
-        end_turn: None,
         phase: None,
     };
     assert_eq!(input.last(), Some(&expected_user_message));

codex-rs/core/tests/suite/realtime_conversation.rs

@@ -1605,7 +1605,6 @@ async fn conversation_startup_context_current_thread_selects_many_turns_by_budge
                     id: None,
                     role: "user".to_string(),
                     content: vec![ContentItem::InputText { text: user_turn }],
-                    end_turn: None,
                     phase: None,
                 }),
                 RolloutItem::ResponseItem(ResponseItem::Message {
@@ -1614,7 +1613,6 @@ async fn conversation_startup_context_current_thread_selects_many_turns_by_budge
                     content: vec![ContentItem::OutputText {
                         text: assistant_turn,
                     }],
-                    end_turn: None,
                     phase: None,
                 }),
             ]

codex-rs/core/tests/suite/review.rs

@@ -538,7 +538,6 @@ async fn review_input_isolated_from_parent_history() {
             content: vec![codex_protocol::models::ContentItem::InputText {
                 text: "parent: earlier user message".to_string(),
             }],
-            end_turn: None,
             phase: None,
         };
         let user_json = serde_json::to_value(&user).unwrap();
@@ -558,7 +557,6 @@ async fn review_input_isolated_from_parent_history() {
             content: vec![codex_protocol::models::ContentItem::OutputText {
                 text: "parent: assistant reply".to_string(),
             }],
-            end_turn: None,
             phase: None,
         };
         let assistant_json = serde_json::to_value(&assistant).unwrap();

codex-rs/protocol/src/items.rs

@@ -266,7 +266,6 @@ pub fn build_hook_prompt_message(fragments: &[HookPromptFragment]) -> Option<Res
         id: Some(uuid::Uuid::new_v4().to_string()),
         role: "user".to_string(),
         content,
-        end_turn: None,
         phase: None,
     })
 }

codex-rs/protocol/src/models.rs

@@ -757,10 +757,6 @@ pub enum ResponseItem {
         id: Option<String>,
         role: String,
         content: Vec<ContentItem>,
-        // Do not use directly, no available consistently across all providers.
-        #[serde(default, skip_serializing_if = "Option::is_none")]
-        #[ts(optional)]
-        end_turn: Option<bool>,
         // Optional output-message phase (for example: "commentary", "final_answer").
         // Availability varies by provider/model, so downstream consumers must
         // preserve fallback behavior when this is absent.
@@ -1114,7 +1110,6 @@ impl From<ResponseInputItem> for ResponseItem {
                 role,
                 content,
                 id: None,
-                end_turn: None,
                 phase: None,
             },
             ResponseInputItem::FunctionCallOutput { call_id, output } => {

codex-rs/protocol/src/protocol.rs

@@ -2867,7 +2867,6 @@ impl From<CompactedItem> for ResponseItem {
             content: vec![ContentItem::OutputText {
                 text: value.message,
             }],
-            end_turn: None,
             phase: None,
         }
     }

codex-rs/rollout/src/tests.rs

@@ -1179,7 +1179,6 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> {
                 content: vec![ContentItem::OutputText {
                     text: format!("reply-{idx}"),
                 }],
-                end_turn: None,
                 phase: None,
             }),
         };

codex-rs/state/src/extract.rs

@@ -177,7 +177,6 @@ mod tests {
             content: vec![ContentItem::InputText {
                 text: "hello from response item".to_string(),
             }],
-            end_turn: None,
             phase: None,
         });
 

codex-rs/tui/src/app/side.rs

@@ -440,7 +440,6 @@ impl App {
             content: vec![ContentItem::InputText {
                 text: SIDE_BOUNDARY_PROMPT.to_string(),
             }],
-            end_turn: None,
             phase: None,
         }
     }