From 78ce61c78eaf1eb596b5b78031a47443a4706e44 Mon Sep 17 00:00:00 2001 From: Shijie Rao Date: Wed, 15 Apr 2026 14:14:43 -0400 Subject: [PATCH 1/6] Fix empty tool descriptions (#17946) ## Summary - Ensure direct namespaced MCP tool groups are emitted with a non-empty namespace description even when namespace metadata is missing or blank. - Add regression coverage for missing MCP namespace descriptions. ## Cause Latest `main` can serialize a direct namespaced MCP tool group with an empty top-level `description`. The namespace description path used `unwrap_or_default()` when `tool_namespaces` did not include metadata for that namespace, so the outbound Responses API payload could contain a tool like `{"type":"namespace","description":""}`. The Responses API rejects that because namespace tool descriptions must be a non-empty string. ## Fix - Add a fallback namespace description: `Tools in the namespace.` - Preserve provided namespace descriptions after trimming, but treat blank descriptions as missing. ### Issue I am seeing This is what I am seeing on the local build. Screenshot 2026-04-15 at 10 55 55 AM --------- Co-authored-by: Sayan Sisodiya --- codex-rs/tools/src/lib.rs | 1 + codex-rs/tools/src/responses_api.rs | 4 ++ codex-rs/tools/src/tool_discovery.rs | 12 +++--- codex-rs/tools/src/tool_discovery_tests.rs | 6 +-- codex-rs/tools/src/tool_registry_plan.rs | 18 +++++++-- .../tools/src/tool_registry_plan_tests.rs | 40 +++++++++++++++++++ 6 files changed, 67 insertions(+), 14 deletions(-) diff --git a/codex-rs/tools/src/lib.rs b/codex-rs/tools/src/lib.rs index b30e0bffc2..4a3ccc01ae 100644 --- a/codex-rs/tools/src/lib.rs +++ b/codex-rs/tools/src/lib.rs @@ -85,6 +85,7 @@ pub use responses_api::ResponsesApiNamespace; pub use responses_api::ResponsesApiNamespaceTool; pub use responses_api::ResponsesApiTool; pub use responses_api::ToolSearchOutputTool; +pub(crate) use responses_api::default_namespace_description; pub use responses_api::dynamic_tool_to_responses_api_tool; pub use responses_api::mcp_tool_to_deferred_responses_api_tool; pub use responses_api::mcp_tool_to_responses_api_tool; diff --git a/codex-rs/tools/src/responses_api.rs b/codex-rs/tools/src/responses_api.rs index 205c075105..5317a082f1 100644 --- a/codex-rs/tools/src/responses_api.rs +++ b/codex-rs/tools/src/responses_api.rs @@ -55,6 +55,10 @@ pub struct ResponsesApiNamespace { pub tools: Vec, } +pub(crate) fn default_namespace_description(namespace_name: &str) -> String { + format!("Tools in the {namespace_name} namespace.") +} + #[derive(Debug, Clone, Serialize, PartialEq)] #[serde(tag = "type")] pub enum ResponsesApiNamespaceTool { diff --git a/codex-rs/tools/src/tool_discovery.rs b/codex-rs/tools/src/tool_discovery.rs index e35da5735c..b773623edf 100644 --- a/codex-rs/tools/src/tool_discovery.rs +++ b/codex-rs/tools/src/tool_discovery.rs @@ -5,6 +5,7 @@ use crate::ResponsesApiTool; use crate::ToolName; use crate::ToolSearchOutputTool; use crate::ToolSpec; +use crate::default_namespace_description; use crate::mcp_tool_to_deferred_responses_api_tool; use codex_app_server_protocol::AppInfo; use serde::Deserialize; @@ -225,6 +226,8 @@ pub fn collect_tool_search_output_tools<'a>( let description = first_tool .connector_description + .map(str::trim) + .filter(|description| !description.is_empty()) .map(str::to_string) .or_else(|| { first_tool @@ -232,12 +235,6 @@ pub fn collect_tool_search_output_tools<'a>( .map(str::trim) .filter(|connector_name| !connector_name.is_empty()) .map(|connector_name| format!("Tools for working with {connector_name}.")) - }) - .or_else(|| { - Some(format!( - "Tools from the {} MCP server.", - first_tool.server_name - )) }); let tools = tools @@ -251,7 +248,8 @@ pub fn collect_tool_search_output_tools<'a>( results.push(ToolSearchOutputTool::Namespace(ResponsesApiNamespace { name: tool_namespace.to_string(), - description: description.unwrap_or_default(), + description: description + .unwrap_or_else(|| default_namespace_description(tool_namespace)), tools, })); } diff --git a/codex-rs/tools/src/tool_discovery_tests.rs b/codex-rs/tools/src/tool_discovery_tests.rs index 10e108fe04..b745f756cc 100644 --- a/codex-rs/tools/src/tool_discovery_tests.rs +++ b/codex-rs/tools/src/tool_discovery_tests.rs @@ -253,7 +253,7 @@ fn collect_tool_search_output_tools_preserves_search_order_while_grouping_by_nam }), ToolSearchOutputTool::Namespace(ResponsesApiNamespace { name: "mcp__docs__".to_string(), - description: "Tools from the docs MCP server.".to_string(), + description: "Tools in the mcp__docs__ namespace.".to_string(), tools: vec![ResponsesApiNamespaceTool::Function(ResponsesApiTool { name: "search".to_string(), description: "Search docs.".to_string(), @@ -272,7 +272,7 @@ fn collect_tool_search_output_tools_preserves_search_order_while_grouping_by_nam } #[test] -fn collect_tool_search_output_tools_falls_back_to_connector_name_description() { +fn collect_tool_search_output_tools_ignores_blank_connector_description() { let gmail_batch_read_email = mcp_tool("gmail-batch-read-email", "Read multiple emails."); let tools = collect_tool_search_output_tools([ToolSearchResultSource { @@ -281,7 +281,7 @@ fn collect_tool_search_output_tools_falls_back_to_connector_name_description() { tool_name: "_batch_read_email", tool: &gmail_batch_read_email, connector_name: Some("Gmail"), - connector_description: None, + connector_description: Some(" "), }]) .expect("collect tool search output tools"); diff --git a/codex-rs/tools/src/tool_registry_plan.rs b/codex-rs/tools/src/tool_registry_plan.rs index cec938962c..b5da9afb25 100644 --- a/codex-rs/tools/src/tool_registry_plan.rs +++ b/codex-rs/tools/src/tool_registry_plan.rs @@ -55,6 +55,7 @@ use crate::create_wait_agent_tool_v2; use crate::create_wait_tool; use crate::create_web_search_tool; use crate::create_write_stdin_tool; +use crate::default_namespace_description; use crate::dynamic_tool_to_responses_api_tool; use crate::mcp_tool_to_responses_api_tool; use crate::request_permissions_tool_description; @@ -485,11 +486,20 @@ pub fn build_tool_registry_plan( for (namespace, mut entries) in namespace_entries { entries.sort_by_key(|tool| tool.name.name.clone()); - let description = params + let tool_namespace = params .tool_namespaces - .and_then(|namespaces| namespaces.get(&namespace)) - .and_then(|namespace| namespace.description.clone()) - .unwrap_or_default(); + .and_then(|namespaces| namespaces.get(&namespace)); + let description = tool_namespace + .and_then(|namespace| namespace.description.as_deref()) + .map(str::trim) + .filter(|description| !description.is_empty()) + .map(str::to_string) + .unwrap_or_else(|| { + let namespace_name = tool_namespace + .map(|namespace| namespace.name.as_str()) + .unwrap_or(namespace.as_str()); + default_namespace_description(namespace_name) + }); let mut tools = Vec::new(); for tool in entries { match mcp_tool_to_responses_api_tool(&tool.name, tool.tool) { diff --git a/codex-rs/tools/src/tool_registry_plan_tests.rs b/codex-rs/tools/src/tool_registry_plan_tests.rs index 7b9e8ed9ac..a643543ee7 100644 --- a/codex-rs/tools/src/tool_registry_plan_tests.rs +++ b/codex-rs/tools/src/tool_registry_plan_tests.rs @@ -1151,6 +1151,46 @@ fn test_build_specs_mcp_tools_converted() { ); } +#[test] +fn test_build_specs_mcp_namespace_description_falls_back_when_missing() { + let model_info = model_info(); + let mut features = Features::with_defaults(); + features.enable(Feature::UnifiedExec); + let available_models = Vec::new(); + let tools_config = ToolsConfig::new(&ToolsConfigParams { + model_info: &model_info, + available_models: &available_models, + features: &features, + image_generation_tool_auth_allowed: true, + web_search_mode: Some(WebSearchMode::Cached), + session_source: SessionSource::Cli, + sandbox_policy: &SandboxPolicy::DangerFullAccess, + windows_sandbox_level: WindowsSandboxLevel::Disabled, + }); + let (tools, _) = build_specs( + &tools_config, + Some(HashMap::from([( + ToolName::namespaced("test_server/", "do_something_cool"), + mcp_tool( + "do_something_cool", + "Do something cool", + serde_json::json!({"type": "object"}), + ), + )])), + /*deferred_mcp_tools*/ None, + &[], + ); + + let namespace_tool = find_tool(&tools, "test_server/"); + let ToolSpec::Namespace(namespace) = &namespace_tool.spec else { + panic!("expected namespace tool"); + }; + assert_eq!( + namespace.description, + "Tools in the test_server/ namespace." + ); +} + #[test] fn test_build_specs_mcp_tools_sorted_by_name() { let model_info = model_info(); From cdfcd2ca9268adbd4bd3ba5af527c97fa8a82805 Mon Sep 17 00:00:00 2001 From: Tom Date: Wed, 15 Apr 2026 11:34:27 -0700 Subject: [PATCH 2/6] [codex] Add local thread store listing (#17824) Builds on top of #17659 Move the filesystem + sqlite thread listing-related operations inside of a local ThreadStore implementation and call ThreadStore from the places that used to perform these filesystem/sqlite operations. This is the first of a series of PRs that will implement the rest of the local ThreadStore. Testing: - added unit tests for the thread store implementation - adjusted some unit tests in the realtime + personality packages whose callsites changed. Specifically I'm trying to hide ThreadMetadata inside of the local implementation and make ThreadMetadata a sqlite implementation detail concern rather than a public interface, preferring the more generate StoredThread interface instead - added a corner case test for the personality migration package that wasn't covered by the existing test suite - adjust the behavior of searched thread listing to run the existing local rollout repair/backfill pass _before_ querying SQLite results, so callers using ThreadStore::list_threads do not miss matches after a partial metadata warm-up --- codex-rs/Cargo.lock | 10 + codex-rs/Cargo.toml | 1 + codex-rs/app-server/Cargo.toml | 1 + .../app-server/src/codex_message_processor.rs | 187 +++--- .../app-server/tests/suite/v2/thread_list.rs | 25 +- codex-rs/core/Cargo.toml | 1 + codex-rs/core/src/codex.rs | 3 + codex-rs/core/src/codex_tests.rs | 6 + codex-rs/core/src/personality_migration.rs | 86 +-- .../core/src/personality_migration_tests.rs | 28 + codex-rs/core/src/realtime_context.rs | 49 +- codex-rs/core/src/realtime_context_tests.rs | 56 +- codex-rs/core/src/rollout.rs | 4 - codex-rs/core/src/state/service.rs | 2 + .../core/tests/suite/realtime_conversation.rs | 10 +- codex-rs/rollout/src/recorder.rs | 4 + codex-rs/thread-store/Cargo.toml | 10 + codex-rs/thread-store/src/lib.rs | 2 + codex-rs/thread-store/src/local/mod.rs | 565 ++++++++++++++++++ codex-rs/thread-store/src/types.rs | 4 + 20 files changed, 821 insertions(+), 233 deletions(-) create mode 100644 codex-rs/thread-store/src/local/mod.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 5e54b772c1..68e0ec092f 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1461,6 +1461,7 @@ dependencies = [ "codex-sandboxing", "codex-shell-command", "codex-state", + "codex-thread-store", "codex-tools", "codex-utils-absolute-path", "codex-utils-cargo-bin", @@ -1938,6 +1939,7 @@ dependencies = [ "codex-state", "codex-terminal-detection", "codex-test-binary-support", + "codex-thread-store", "codex-tools", "codex-utils-absolute-path", "codex-utils-cache", @@ -2841,9 +2843,17 @@ version = "0.0.0" dependencies = [ "async-trait", "chrono", + "codex-git-utils", "codex-protocol", + "codex-rollout", + "codex-state", + "pretty_assertions", "serde", + "serde_json", + "tempfile", "thiserror 2.0.18", + "tokio", + "uuid", ] [[package]] diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index b57a464d5e..4bb8837211 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -166,6 +166,7 @@ codex-state = { path = "state" } codex-stdio-to-uds = { path = "stdio-to-uds" } codex-terminal-detection = { path = "terminal-detection" } codex-test-binary-support = { path = "test-binary-support" } +codex-thread-store = { path = "thread-store" } codex-tools = { path = "tools" } codex-tui = { path = "tui" } codex-utils-absolute-path = { path = "utils/absolute-path" } diff --git a/codex-rs/app-server/Cargo.toml b/codex-rs/app-server/Cargo.toml index 5dc3a31485..408f39fafe 100644 --- a/codex-rs/app-server/Cargo.toml +++ b/codex-rs/app-server/Cargo.toml @@ -54,6 +54,7 @@ codex-rmcp-client = { workspace = true } codex-rollout = { workspace = true } codex-sandboxing = { workspace = true } codex-state = { workspace = true } +codex-thread-store = { workspace = true } codex-tools = { workspace = true } codex-utils-absolute-path = { workspace = true } codex-utils-json-to-toml = { workspace = true } diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 513837c7ce..05ce2d7974 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -203,7 +203,6 @@ use codex_chatgpt::connectors; use codex_cloud_requirements::cloud_requirements_loader; use codex_config::types::McpServerTransportConfig; use codex_core::CodexThread; -use codex_core::Cursor as RolloutCursor; use codex_core::ForkSnapshot; use codex_core::NewThread; use codex_core::RolloutRecorder; @@ -211,7 +210,6 @@ use codex_core::SessionMeta; use codex_core::SteerInputError; use codex_core::ThreadConfigSnapshot; use codex_core::ThreadManager; -use codex_core::ThreadSortKey as CoreThreadSortKey; use codex_core::append_thread_name; use codex_core::config::Config; use codex_core::config::ConfigOverrides; @@ -232,7 +230,6 @@ use codex_core::find_archived_thread_path_by_id_str; use codex_core::find_thread_name_by_id; use codex_core::find_thread_names_by_ids; use codex_core::find_thread_path_by_id_str; -use codex_core::parse_cursor; use codex_core::path_utils; use codex_core::plugins::MarketplaceAddError; use codex_core::plugins::MarketplaceError; @@ -322,6 +319,12 @@ use codex_state::StateRuntime; use codex_state::ThreadMetadata; use codex_state::ThreadMetadataBuilder; use codex_state::log_db::LogDbLayer; +use codex_thread_store::ListThreadsParams as StoreListThreadsParams; +use codex_thread_store::LocalThreadStore; +use codex_thread_store::StoredThread; +use codex_thread_store::ThreadSortKey as StoreThreadSortKey; +use codex_thread_store::ThreadStore; +use codex_thread_store::ThreadStoreError; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_json_to_toml::json_to_toml; use codex_utils_pty::DEFAULT_OUTPUT_BYTES_CAP; @@ -3819,15 +3822,15 @@ impl CodexMessageProcessor { .map(|value| value as usize) .unwrap_or(THREAD_LIST_DEFAULT_LIMIT) .clamp(1, THREAD_LIST_MAX_LIMIT); - let core_sort_key = match sort_key.unwrap_or(ThreadSortKey::CreatedAt) { - ThreadSortKey::CreatedAt => CoreThreadSortKey::CreatedAt, - ThreadSortKey::UpdatedAt => CoreThreadSortKey::UpdatedAt, + let store_sort_key = match sort_key.unwrap_or(ThreadSortKey::CreatedAt) { + ThreadSortKey::CreatedAt => StoreThreadSortKey::CreatedAt, + ThreadSortKey::UpdatedAt => StoreThreadSortKey::UpdatedAt, }; let (summaries, next_cursor) = match self .list_threads_common( requested_page_size, cursor, - core_sort_key, + store_sort_key, ThreadListFilters { model_providers, source_kinds, @@ -5071,7 +5074,7 @@ impl CodexMessageProcessor { &self, requested_page_size: usize, cursor: Option, - sort_key: CoreThreadSortKey, + sort_key: StoreThreadSortKey, filters: ThreadListFilters, ) -> Result<(Vec, Option), JSONRPCErrorError> { let ThreadListFilters { @@ -5081,16 +5084,7 @@ impl CodexMessageProcessor { cwd, search_term, } = filters; - let mut cursor_obj: Option = match cursor.as_ref() { - Some(cursor_str) => { - Some(parse_cursor(cursor_str).ok_or_else(|| JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("invalid cursor: {cursor_str}"), - data: None, - })?) - } - None => None, - }; + let mut cursor_obj = cursor; let mut last_cursor = cursor_obj.clone(); let mut remaining = requested_page_size; let mut items = Vec::with_capacity(requested_page_size); @@ -5109,54 +5103,26 @@ impl CodexMessageProcessor { let fallback_provider = self.config.model_provider_id.clone(); let (allowed_sources_vec, source_kind_filter) = compute_source_filters(source_kinds); let allowed_sources = allowed_sources_vec.as_slice(); - let state_db_ctx = get_state_db(&self.config).await; + let store = LocalThreadStore::new(codex_rollout::RolloutConfig::from_view(&self.config)); while remaining > 0 { let page_size = remaining.min(THREAD_LIST_MAX_LIMIT); - let page = if archived { - RolloutRecorder::list_archived_threads( - &self.config, + let page = store + .list_threads(StoreListThreadsParams { page_size, - cursor_obj.as_ref(), + cursor: cursor_obj.clone(), sort_key, - allowed_sources, - model_provider_filter.as_deref(), - fallback_provider.as_str(), - search_term.as_deref(), - ) + allowed_sources: allowed_sources.to_vec(), + model_providers: model_provider_filter.clone(), + archived, + search_term: search_term.clone(), + }) .await - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to list threads: {err}"), - data: None, - })? - } else { - RolloutRecorder::list_threads( - &self.config, - page_size, - cursor_obj.as_ref(), - sort_key, - allowed_sources, - model_provider_filter.as_deref(), - fallback_provider.as_str(), - search_term.as_deref(), - ) - .await - .map_err(|err| JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to list threads: {err}"), - data: None, - })? - }; + .map_err(thread_store_list_error)?; let mut filtered = Vec::with_capacity(page.items.len()); for it in page.items { - let Some(summary) = summary_from_thread_list_item( - it, - fallback_provider.as_str(), - state_db_ctx.as_ref(), - ) - .await + let Some(summary) = summary_from_stored_thread(it, fallback_provider.as_str()) else { continue; }; @@ -5176,12 +5142,8 @@ impl CodexMessageProcessor { items.extend(filtered); remaining = requested_page_size.saturating_sub(items.len()); - // Encode RolloutCursor into the JSON-RPC string form returned to clients. let next_cursor_value = page.next_cursor.clone(); - next_cursor = next_cursor_value - .as_ref() - .and_then(|cursor| serde_json::to_value(cursor).ok()) - .and_then(|value| value.as_str().map(str::to_owned)); + next_cursor = next_cursor_value.clone(); if remaining == 0 { break; } @@ -9421,67 +9383,52 @@ fn set_thread_name_from_title(thread: &mut Thread, title: String) { thread.name = Some(title); } -async fn summary_from_thread_list_item( - it: codex_core::ThreadItem, - fallback_provider: &str, - state_db_ctx: Option<&StateDbHandle>, -) -> Option { - if let Some(thread_id) = it.thread_id { - let timestamp = it.created_at.clone(); - let updated_at = it.updated_at.clone().or_else(|| timestamp.clone()); - let model_provider = it - .model_provider - .clone() - .unwrap_or_else(|| fallback_provider.to_string()); - let cwd = it.cwd?; - let cli_version = it.cli_version.unwrap_or_default(); - let source = with_thread_spawn_agent_metadata( - it.source - .unwrap_or(codex_protocol::protocol::SessionSource::Unknown), - it.agent_nickname.clone(), - it.agent_role.clone(), - ); - return Some(ConversationSummary { - conversation_id: thread_id, - path: it.path, - preview: it.first_user_message.unwrap_or_default(), - timestamp, - updated_at, - model_provider, - cwd, - cli_version, - source, - git_info: if it.git_sha.is_none() - && it.git_branch.is_none() - && it.git_origin_url.is_none() - { - None - } else { - Some(ConversationGitInfo { - sha: it.git_sha, - branch: it.git_branch, - origin_url: it.git_origin_url, - }) - }, - }); +fn thread_store_list_error(err: ThreadStoreError) -> JSONRPCErrorError { + match err { + ThreadStoreError::InvalidRequest { message } => JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message, + data: None, + }, + err => JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + message: format!("failed to list threads: {err}"), + data: None, + }, } - if let Some(thread_id) = thread_id_from_rollout_path(it.path.as_path()) { - return read_summary_from_state_db_context_by_thread_id(state_db_ctx, thread_id).await; - } - None } -fn thread_id_from_rollout_path(path: &Path) -> Option { - let file_name = path.file_name()?.to_str()?; - let stem = file_name.strip_suffix(".jsonl")?; - if stem.len() < 37 { - return None; - } - let uuid_start = stem.len().saturating_sub(36); - if !stem[..uuid_start].ends_with('-') { - return None; - } - ThreadId::from_string(&stem[uuid_start..]).ok() +fn summary_from_stored_thread( + thread: StoredThread, + fallback_provider: &str, +) -> Option { + let path = thread.rollout_path?; + let source = with_thread_spawn_agent_metadata( + thread.source, + thread.agent_nickname.clone(), + thread.agent_role.clone(), + ); + let git_info = thread.git_info.map(|git| ConversationGitInfo { + sha: git.commit_hash.map(|sha| sha.0), + branch: git.branch, + origin_url: git.repository_url, + }); + Some(ConversationSummary { + conversation_id: thread.thread_id, + path, + preview: thread.first_user_message.unwrap_or(thread.preview), + timestamp: Some(thread.created_at.to_rfc3339_opts(SecondsFormat::Secs, true)), + updated_at: Some(thread.updated_at.to_rfc3339_opts(SecondsFormat::Secs, true)), + model_provider: if thread.model_provider.is_empty() { + fallback_provider.to_string() + } else { + thread.model_provider + }, + cwd: thread.cwd, + cli_version: thread.cli_version, + source, + git_info, + }) } #[allow(clippy::too_many_arguments)] diff --git a/codex-rs/app-server/tests/suite/v2/thread_list.rs b/codex-rs/app-server/tests/suite/v2/thread_list.rs index cf1ac4ff11..8fd4e47303 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_list.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_list.rs @@ -563,15 +563,34 @@ sqlite = true /*git_info*/ None, )?; - // `thread/list` only applies `search_term` on the sqlite path. In this test we - // create rollouts manually, so we must also create the sqlite DB and mark backfill - // complete; otherwise app-server will permanently use filesystem fallback. + // `thread/list` applies `search_term` on the sqlite fast path. This test creates + // rollouts manually, so mark the DB backfill complete and then run an unsearched + // list large enough to repair every rollout the searched list should find. let state_db = codex_state::StateRuntime::init(codex_home.path().to_path_buf(), "mock_provider".into()) .await?; state_db .mark_backfill_complete(/*last_watermark*/ None) .await?; + let rollout_config = codex_rollout::RolloutConfig { + codex_home: codex_home.path().to_path_buf(), + sqlite_home: codex_home.path().to_path_buf(), + cwd: codex_home.path().to_path_buf(), + model_provider_id: "mock_provider".to_string(), + generate_memories: false, + }; + let repaired_page = codex_core::RolloutRecorder::list_threads( + &rollout_config, + /*page_size*/ 10, + /*cursor*/ None, + codex_core::ThreadSortKey::CreatedAt, + &[], + /*model_providers*/ None, + "mock_provider", + /*search_term*/ None, + ) + .await?; + assert_eq!(repaired_page.items.len(), 3); let mut mcp = init_mcp(codex_home.path()).await?; let request_id = mcp diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index a371966079..78dfd3c070 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -57,6 +57,7 @@ codex-rmcp-client = { workspace = true } codex-sandboxing = { workspace = true } codex-state = { workspace = true } codex-terminal-detection = { workspace = true } +codex-thread-store = { workspace = true } codex-tools = { workspace = true } codex-utils-absolute-path = { workspace = true } codex-utils-cache = { workspace = true } diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index a3d7594813..bc7cd07e4c 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -136,9 +136,11 @@ use codex_protocol::request_permissions::RequestPermissionsResponse; use codex_protocol::request_user_input::RequestUserInputArgs; use codex_protocol::request_user_input::RequestUserInputResponse; use codex_rmcp_client::ElicitationResponse; +use codex_rollout::RolloutConfig; use codex_rollout::state_db; use codex_shell_command::parse_command::parse_command; use codex_terminal_detection::user_agent; +use codex_thread_store::LocalThreadStore; use codex_tools::filter_tool_suggest_discoverable_tools_for_client; use codex_utils_output_truncation::TruncationPolicy; use codex_utils_stream_parser::AssistantTextChunk; @@ -2127,6 +2129,7 @@ impl Session { network_proxy, network_approval: Arc::clone(&network_approval), state_db: state_db_ctx.clone(), + thread_store: LocalThreadStore::new(RolloutConfig::from_view(config.as_ref())), model_client: ModelClient::new( Some(Arc::clone(&auth_manager)), conversation_id, diff --git a/codex-rs/core/src/codex_tests.rs b/codex-rs/core/src/codex_tests.rs index 46d4cda57e..fb7bbaadba 100644 --- a/codex-rs/core/src/codex_tests.rs +++ b/codex-rs/core/src/codex_tests.rs @@ -2836,6 +2836,9 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { network_proxy: None, network_approval: Arc::clone(&network_approval), state_db: None, + thread_store: codex_thread_store::LocalThreadStore::new( + codex_rollout::RolloutConfig::from_view(config.as_ref()), + ), model_client: ModelClient::new( Some(auth_manager.clone()), conversation_id, @@ -3693,6 +3696,9 @@ pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx( network_proxy: None, network_approval: Arc::clone(&network_approval), state_db: None, + thread_store: codex_thread_store::LocalThreadStore::new( + codex_rollout::RolloutConfig::from_view(config.as_ref()), + ), model_client: ModelClient::new( Some(Arc::clone(&auth_manager)), conversation_id, diff --git a/codex-rs/core/src/personality_migration.rs b/codex-rs/core/src/personality_migration.rs index 52cabf55de..4b53a2629e 100644 --- a/codex-rs/core/src/personality_migration.rs +++ b/codex-rs/core/src/personality_migration.rs @@ -1,14 +1,10 @@ use crate::config::edit::ConfigEditsBuilder; -use crate::rollout::ARCHIVED_SESSIONS_SUBDIR; -use crate::rollout::SESSIONS_SUBDIR; -use crate::rollout::list::ThreadListConfig; -use crate::rollout::list::ThreadListLayout; -use crate::rollout::list::ThreadSortKey; -use crate::rollout::list::get_threads_in_root; use codex_config::config_toml::ConfigToml; use codex_protocol::config_types::Personality; -use codex_protocol::protocol::SessionSource; -use codex_rollout::state_db; +use codex_thread_store::ListThreadsParams; +use codex_thread_store::LocalThreadStore; +use codex_thread_store::ThreadSortKey; +use codex_thread_store::ThreadStore; use std::io; use std::path::Path; use tokio::fs::OpenOptions; @@ -64,57 +60,33 @@ pub async fn maybe_migrate_personality( } async fn has_recorded_sessions(codex_home: &Path, default_provider: &str) -> io::Result { - let allowed_sources: &[SessionSource] = &[]; + let store = LocalThreadStore::new(codex_rollout::RolloutConfig { + codex_home: codex_home.to_path_buf(), + sqlite_home: codex_home.to_path_buf(), + cwd: codex_home.to_path_buf(), + model_provider_id: default_provider.to_string(), + generate_memories: false, + }); + if has_threads(&store, /*archived*/ false).await? { + return Ok(true); + } + has_threads(&store, /*archived*/ true).await +} - if let Some(state_db_ctx) = state_db::open_if_present(codex_home, default_provider).await - && let Some(ids) = state_db::list_thread_ids_db( - Some(state_db_ctx.as_ref()), - codex_home, - /*page_size*/ 1, - /*cursor*/ None, - ThreadSortKey::CreatedAt, - allowed_sources, - /*model_providers*/ None, - /*archived_only*/ false, - "personality_migration", - ) +async fn has_threads(store: &LocalThreadStore, archived: bool) -> io::Result { + store + .list_threads(ListThreadsParams { + page_size: 1, + cursor: None, + sort_key: ThreadSortKey::CreatedAt, + allowed_sources: Vec::new(), + model_providers: None, + archived, + search_term: None, + }) .await - && !ids.is_empty() - { - return Ok(true); - } - - let sessions = get_threads_in_root( - codex_home.join(SESSIONS_SUBDIR), - /*page_size*/ 1, - /*cursor*/ None, - ThreadSortKey::CreatedAt, - ThreadListConfig { - allowed_sources, - model_providers: None, - default_provider, - layout: ThreadListLayout::NestedByDate, - }, - ) - .await?; - if !sessions.items.is_empty() { - return Ok(true); - } - - let archived_sessions = get_threads_in_root( - codex_home.join(ARCHIVED_SESSIONS_SUBDIR), - /*page_size*/ 1, - /*cursor*/ None, - ThreadSortKey::CreatedAt, - ThreadListConfig { - allowed_sources, - model_providers: None, - default_provider, - layout: ThreadListLayout::Flat, - }, - ) - .await?; - Ok(!archived_sessions.items.is_empty()) + .map(|page| !page.items.is_empty()) + .map_err(io::Error::other) } async fn create_marker(marker_path: &Path) -> io::Result<()> { diff --git a/codex-rs/core/src/personality_migration_tests.rs b/codex-rs/core/src/personality_migration_tests.rs index de1070ad34..4aef53a5c4 100644 --- a/codex-rs/core/src/personality_migration_tests.rs +++ b/codex-rs/core/src/personality_migration_tests.rs @@ -7,6 +7,8 @@ use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::UserMessageEvent; +use codex_rollout::ARCHIVED_SESSIONS_SUBDIR; +use codex_rollout::SESSIONS_SUBDIR; use pretty_assertions::assert_eq; use tempfile::TempDir; use tokio::io::AsyncWriteExt; @@ -25,6 +27,16 @@ async fn write_session_with_user_event(codex_home: &Path) -> io::Result<()> { .join("2025") .join("01") .join("01"); + write_rollout_with_user_event(&dir, thread_id).await +} + +async fn write_archived_session_with_user_event(codex_home: &Path) -> io::Result<()> { + let thread_id = ThreadId::new(); + let dir = codex_home.join(ARCHIVED_SESSIONS_SUBDIR); + write_rollout_with_user_event(&dir, thread_id).await +} + +async fn write_rollout_with_user_event(dir: &Path, thread_id: ThreadId) -> io::Result<()> { tokio::fs::create_dir_all(&dir).await?; let file_path = dir.join(format!("rollout-{TEST_TIMESTAMP}-{thread_id}.jsonl")); let mut file = tokio::fs::File::create(&file_path).await?; @@ -85,6 +97,22 @@ async fn applies_when_sessions_exist_and_no_personality() -> io::Result<()> { Ok(()) } +#[tokio::test] +async fn applies_when_only_archived_sessions_exist_and_no_personality() -> io::Result<()> { + let temp = TempDir::new()?; + write_archived_session_with_user_event(temp.path()).await?; + + let config_toml = ConfigToml::default(); + let status = maybe_migrate_personality(temp.path(), &config_toml).await?; + + assert_eq!(status, PersonalityMigrationStatus::Applied); + assert!(temp.path().join(PERSONALITY_MIGRATION_FILENAME).exists()); + + let persisted = read_config_toml(temp.path()).await?; + assert_eq!(persisted.personality, Some(Personality::Pragmatic)); + Ok(()) +} + #[tokio::test] async fn skips_when_marker_exists() -> io::Result<()> { let temp = TempDir::new()?; diff --git a/codex-rs/core/src/realtime_context.rs b/codex-rs/core/src/realtime_context.rs index 6e9815f562..30cbd6f636 100644 --- a/codex-rs/core/src/realtime_context.rs +++ b/codex-rs/core/src/realtime_context.rs @@ -4,8 +4,10 @@ use crate::event_mapping::is_contextual_user_message_content; use chrono::Utc; use codex_git_utils::resolve_root_git_project_for_trust; use codex_protocol::models::ResponseItem; -use codex_state::SortKey; -use codex_state::ThreadMetadata; +use codex_thread_store::ListThreadsParams; +use codex_thread_store::StoredThread; +use codex_thread_store::ThreadSortKey; +use codex_thread_store::ThreadStore; use codex_utils_output_truncation::TruncationPolicy; use codex_utils_output_truncation::truncate_text; use dirs::home_dir; @@ -98,7 +100,7 @@ pub(crate) async fn build_realtime_startup_context( } if let Some(section) = format_section( "Notes", - Some("Built at realtime startup from the current thread history, persisted thread metadata in the state DB, and a bounded local workspace scan. This excludes repo memory instructions, AGENTS files, project-doc prompt blends, and memory summaries.".to_string()), + Some("Built at realtime startup from the current thread history, local thread metadata, and a bounded local workspace scan. This excludes repo memory instructions, AGENTS files, project-doc prompt blends, and memory summaries.".to_string()), NOTES_SECTION_TOKEN_BUDGET, ) { parts.push(section); @@ -117,33 +119,31 @@ pub(crate) async fn build_realtime_startup_context( Some(context) } -async fn load_recent_threads(sess: &Session) -> Vec { - let Some(state_db) = sess.services.state_db.as_ref() else { - return Vec::new(); - }; - - match state_db - .list_threads( - MAX_RECENT_THREADS, - /*anchor*/ None, - SortKey::UpdatedAt, - &[], - /*model_providers*/ None, - /*archived_only*/ false, - /*search_term*/ None, - ) +async fn load_recent_threads(sess: &Session) -> Vec { + match sess + .services + .thread_store + .list_threads(ListThreadsParams { + page_size: MAX_RECENT_THREADS, + cursor: None, + sort_key: ThreadSortKey::UpdatedAt, + allowed_sources: Vec::new(), + model_providers: None, + archived: false, + search_term: None, + }) .await { Ok(page) => page.items, Err(err) => { - warn!("failed to load realtime startup threads from state db: {err}"); + warn!("failed to load realtime startup threads from thread store: {err}"); Vec::new() } } } -fn build_recent_work_section(cwd: &Path, recent_threads: &[ThreadMetadata]) -> Option { - let mut groups: HashMap> = HashMap::new(); +fn build_recent_work_section(cwd: &Path, recent_threads: &[StoredThread]) -> Option { + let mut groups: HashMap> = HashMap::new(); for entry in recent_threads { let group = resolve_root_git_project_for_trust(&entry.cwd).unwrap_or_else(|| entry.cwd.clone()); @@ -446,7 +446,7 @@ fn format_section(title: &str, body: Option, budget_tokens: usize) -> Op fn format_thread_group( current_group: &Path, group: &Path, - entries: Vec<&ThreadMetadata>, + entries: Vec<&StoredThread>, ) -> Option { let latest = entries.first()?; let group_label = if resolve_root_git_project_for_trust(latest.cwd.as_path()).is_some() { @@ -461,8 +461,9 @@ fn format_thread_group( ]; if let Some(git_branch) = latest - .git_branch - .as_deref() + .git_info + .as_ref() + .and_then(|git| git.branch.as_deref()) .filter(|git_branch| !git_branch.is_empty()) { lines.push(format!("Latest branch: {git_branch}")); diff --git a/codex-rs/core/src/realtime_context_tests.rs b/codex-rs/core/src/realtime_context_tests.rs index 7495161d46..7e1e82476f 100644 --- a/codex-rs/core/src/realtime_context_tests.rs +++ b/codex-rs/core/src/realtime_context_tests.rs @@ -3,20 +3,31 @@ use super::build_recent_work_section; use super::build_workspace_section_with_user_root; use chrono::TimeZone; use chrono::Utc; +use codex_git_utils::GitSha; use codex_protocol::ThreadId; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; -use codex_state::ThreadMetadata; +use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::GitInfo; +use codex_protocol::protocol::SandboxPolicy; +use codex_protocol::protocol::SessionSource; +use codex_thread_store::StoredThread; use pretty_assertions::assert_eq; use std::fs; use std::path::PathBuf; use std::process::Command; use tempfile::TempDir; -fn thread_metadata(cwd: &str, title: &str, first_user_message: &str) -> ThreadMetadata { - ThreadMetadata { - id: ThreadId::new(), - rollout_path: PathBuf::from("/tmp/rollout.jsonl"), +fn stored_thread(cwd: &str, title: &str, first_user_message: &str) -> StoredThread { + StoredThread { + thread_id: ThreadId::new(), + rollout_path: Some(PathBuf::from("/tmp/rollout.jsonl")), + forked_from_id: None, + preview: first_user_message.to_string(), + name: (!title.is_empty()).then(|| title.to_string()), + model_provider: "test-provider".to_string(), + model: Some("gpt-5".to_string()), + reasoning_effort: None, created_at: Utc .timestamp_opt(1_709_251_100, 0) .single() @@ -25,24 +36,23 @@ fn thread_metadata(cwd: &str, title: &str, first_user_message: &str) -> ThreadMe .timestamp_opt(1_709_251_200, 0) .single() .expect("valid timestamp"), - source: "cli".to_string(), - agent_path: None, - agent_nickname: None, - agent_role: None, - model_provider: "test-provider".to_string(), - model: Some("gpt-5".to_string()), - reasoning_effort: None, + archived_at: None, cwd: PathBuf::from(cwd), cli_version: "test".to_string(), - title: title.to_string(), - sandbox_policy: "workspace-write".to_string(), - approval_mode: "never".to_string(), - tokens_used: 0, + source: SessionSource::Cli, + agent_nickname: None, + agent_role: None, + agent_path: None, + git_info: Some(GitInfo { + commit_hash: Some(GitSha::new("abcdef")), + branch: Some("main".to_string()), + repository_url: None, + }), + approval_mode: AskForApproval::Never, + sandbox_policy: SandboxPolicy::new_read_only_policy(), + token_usage: None, first_user_message: Some(first_user_message.to_string()), - archived_at: None, - git_sha: None, - git_branch: Some("main".to_string()), - git_origin_url: None, + history: None, } } @@ -224,17 +234,17 @@ fn recent_work_section_groups_threads_by_cwd() { fs::create_dir_all(&outside).expect("create outside dir"); let recent_threads = vec![ - thread_metadata( + stored_thread( workspace_a.to_string_lossy().as_ref(), "Investigate realtime startup context", "Log the startup context before sending it", ), - thread_metadata( + stored_thread( workspace_b.to_string_lossy().as_ref(), "Trim websocket startup payload", "Remove memories from the realtime startup context", ), - thread_metadata(outside.to_string_lossy().as_ref(), "", "Inspect flaky test"), + stored_thread(outside.to_string_lossy().as_ref(), "", "Inspect flaky test"), ]; let current_cwd = workspace_a; let repo = fs::canonicalize(repo).expect("canonicalize repo"); diff --git a/codex-rs/core/src/rollout.rs b/codex-rs/core/src/rollout.rs index 26e3a84d5c..4913f957ea 100644 --- a/codex-rs/core/src/rollout.rs +++ b/codex-rs/core/src/rollout.rs @@ -46,11 +46,7 @@ impl codex_rollout::RolloutConfigView for Config { } pub(crate) mod list { - pub use codex_rollout::ThreadListConfig; - pub use codex_rollout::ThreadListLayout; - pub use codex_rollout::ThreadSortKey; pub use codex_rollout::find_thread_path_by_id_str; - pub use codex_rollout::get_threads_in_root; } pub(crate) mod metadata { diff --git a/codex-rs/core/src/state/service.rs b/codex-rs/core/src/state/service.rs index 4b6c5b30f5..5db38f7b72 100644 --- a/codex-rs/core/src/state/service.rs +++ b/codex-rs/core/src/state/service.rs @@ -24,6 +24,7 @@ use codex_mcp::McpConnectionManager; use codex_models_manager::manager::ModelsManager; use codex_otel::SessionTelemetry; use codex_rollout::state_db::StateDbHandle; +use codex_thread_store::LocalThreadStore; use std::path::PathBuf; use tokio::sync::Mutex; use tokio::sync::RwLock; @@ -59,6 +60,7 @@ pub(crate) struct SessionServices { pub(crate) network_proxy: Option, pub(crate) network_approval: Arc, pub(crate) state_db: Option, + pub(crate) thread_store: LocalThreadStore, /// Session-scoped model client shared across turns. pub(crate) model_client: ModelClient, pub(crate) code_mode_service: CodeModeService, diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index add4065f7b..ecaaeb8cba 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -194,10 +194,16 @@ async fn seed_recent_thread( let db = test.codex.state_db().context("state db enabled")?; let thread_id = ThreadId::new(); let updated_at = Utc::now(); + let rollout_path = test + .codex_home_path() + .join(format!("rollout-{thread_id}.jsonl")); + // This helper seeds SQLite metadata directly. Local listing drops stale metadata rows whose + // rollout path no longer exists, so create the placeholder path that the test metadata points + // at without exercising rollout writing in this realtime-context test. + std::fs::write(&rollout_path, "")?; let mut metadata_builder = codex_state::ThreadMetadataBuilder::new( thread_id, - test.codex_home_path() - .join(format!("rollout-{thread_id}.jsonl")), + rollout_path, updated_at, SessionSource::Cli, ); diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index 4f88ff1a21..ab1ae0a5bf 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -279,6 +279,10 @@ impl RolloutRecorder { let codex_home = config.codex_home(); let state_db_ctx = state_db::get_state_db(config).await; + // Search is the SQLite-optimized path and assumes a DB marked backfill-complete is + // actually populated enough to answer the query. If unmigrated rollout files still exist + // on disk, the repair path below may or may not run and catch them depending on whether + // SQLite already has another matching search hit. if search_term.is_some() && let Some(db_page) = state_db::list_threads_db( state_db_ctx.as_deref(), diff --git a/codex-rs/thread-store/Cargo.toml b/codex-rs/thread-store/Cargo.toml index a8f22fbfe5..328d9a4a7f 100644 --- a/codex-rs/thread-store/Cargo.toml +++ b/codex-rs/thread-store/Cargo.toml @@ -14,6 +14,16 @@ workspace = true [dependencies] async-trait = { workspace = true } chrono = { workspace = true, features = ["serde"] } +codex-git-utils = { workspace = true } codex-protocol = { workspace = true } +codex-rollout = { workspace = true } serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } thiserror = { workspace = true } + +[dev-dependencies] +codex-state = { workspace = true } +pretty_assertions = { workspace = true } +tempfile = { workspace = true } +tokio = { workspace = true, features = ["macros"] } +uuid = { workspace = true } diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index 5525db1394..a64bf023be 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/codex-rs/thread-store/src/lib.rs @@ -5,12 +5,14 @@ //! any other backing store. mod error; +mod local; mod recorder; mod store; mod types; pub use error::ThreadStoreError; pub use error::ThreadStoreResult; +pub use local::LocalThreadStore; pub use recorder::ThreadRecorder; pub use store::ThreadStore; pub use types::AppendThreadItemsParams; diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs new file mode 100644 index 0000000000..cc15033cd9 --- /dev/null +++ b/codex-rs/thread-store/src/local/mod.rs @@ -0,0 +1,565 @@ +use async_trait::async_trait; +use chrono::DateTime; +use chrono::Utc; +use codex_git_utils::GitSha; +use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::GitInfo; +use codex_protocol::protocol::SandboxPolicy; +use codex_protocol::protocol::SessionSource; +use codex_rollout::RolloutConfig; +use codex_rollout::RolloutRecorder; +use codex_rollout::ThreadItem; +use codex_rollout::parse_cursor; + +use crate::AppendThreadItemsParams; +use crate::ArchiveThreadParams; +use crate::CreateThreadParams; +use crate::ListThreadsParams; +use crate::LoadThreadHistoryParams; +use crate::ReadThreadParams; +use crate::ResumeThreadRecorderParams; +use crate::SetThreadNameParams; +use crate::StoredThread; +use crate::StoredThreadHistory; +use crate::ThreadPage; +use crate::ThreadRecorder; +use crate::ThreadSortKey; +use crate::ThreadStore; +use crate::ThreadStoreError; +use crate::ThreadStoreResult; +use crate::UpdateThreadMetadataParams; + +/// Local filesystem/SQLite-backed implementation of [`ThreadStore`]. +#[derive(Clone, Debug)] +pub struct LocalThreadStore { + config: RolloutConfig, +} + +impl LocalThreadStore { + /// Create a local store from the rollout configuration used by existing local persistence. + pub fn new(config: RolloutConfig) -> Self { + Self { config } + } +} + +#[async_trait] +impl ThreadStore for LocalThreadStore { + async fn create_thread( + &self, + _params: CreateThreadParams, + ) -> ThreadStoreResult> { + unsupported("create_thread") + } + + async fn resume_thread_recorder( + &self, + _params: ResumeThreadRecorderParams, + ) -> ThreadStoreResult> { + unsupported("resume_thread_recorder") + } + + async fn append_items(&self, _params: AppendThreadItemsParams) -> ThreadStoreResult<()> { + unsupported("append_items") + } + + async fn load_history( + &self, + _params: LoadThreadHistoryParams, + ) -> ThreadStoreResult { + unsupported("load_history") + } + + async fn read_thread(&self, _params: ReadThreadParams) -> ThreadStoreResult { + unsupported("read_thread") + } + + async fn list_threads(&self, params: ListThreadsParams) -> ThreadStoreResult { + let cursor = params + .cursor + .as_deref() + .map(|cursor| { + parse_cursor(cursor).ok_or_else(|| ThreadStoreError::InvalidRequest { + message: format!("invalid cursor: {cursor}"), + }) + }) + .transpose()?; + let sort_key = match params.sort_key { + ThreadSortKey::CreatedAt => codex_rollout::ThreadSortKey::CreatedAt, + ThreadSortKey::UpdatedAt => codex_rollout::ThreadSortKey::UpdatedAt, + }; + let page = list_rollout_threads(&self.config, ¶ms, cursor.as_ref(), sort_key).await?; + + let next_cursor = page + .next_cursor + .as_ref() + .and_then(|cursor| serde_json::to_value(cursor).ok()) + .and_then(|value| value.as_str().map(str::to_owned)); + let items = page + .items + .into_iter() + .filter_map(|item| { + stored_thread_from_rollout_item( + item, + params.archived, + self.config.model_provider_id.as_str(), + ) + }) + .collect::>(); + + Ok(ThreadPage { items, next_cursor }) + } + + async fn set_thread_name(&self, _params: SetThreadNameParams) -> ThreadStoreResult<()> { + unsupported("set_thread_name") + } + + async fn update_thread_metadata( + &self, + _params: UpdateThreadMetadataParams, + ) -> ThreadStoreResult { + unsupported("update_thread_metadata") + } + + async fn archive_thread(&self, _params: ArchiveThreadParams) -> ThreadStoreResult<()> { + unsupported("archive_thread") + } + + async fn unarchive_thread( + &self, + _params: ArchiveThreadParams, + ) -> ThreadStoreResult { + unsupported("unarchive_thread") + } +} + +fn unsupported(operation: &str) -> ThreadStoreResult { + Err(ThreadStoreError::Internal { + message: format!("local thread store does not implement {operation} in this slice"), + }) +} + +async fn list_rollout_threads( + config: &RolloutConfig, + params: &ListThreadsParams, + cursor: Option<&codex_rollout::Cursor>, + sort_key: codex_rollout::ThreadSortKey, +) -> ThreadStoreResult { + let page = if params.archived { + RolloutRecorder::list_archived_threads( + config, + params.page_size, + cursor, + sort_key, + params.allowed_sources.as_slice(), + params.model_providers.as_deref(), + config.model_provider_id.as_str(), + params.search_term.as_deref(), + ) + .await + } else { + RolloutRecorder::list_threads( + config, + params.page_size, + cursor, + sort_key, + params.allowed_sources.as_slice(), + params.model_providers.as_deref(), + config.model_provider_id.as_str(), + params.search_term.as_deref(), + ) + .await + }; + page.map_err(|err| ThreadStoreError::Internal { + message: format!("failed to list threads: {err}"), + }) +} + +fn stored_thread_from_rollout_item( + item: ThreadItem, + archived: bool, + default_provider: &str, +) -> Option { + let thread_id = item + .thread_id + .or_else(|| thread_id_from_rollout_path(item.path.as_path()))?; + let created_at = parse_rfc3339(item.created_at.as_deref()).unwrap_or_else(Utc::now); + let updated_at = parse_rfc3339(item.updated_at.as_deref()).unwrap_or(created_at); + let archived_at = archived.then_some(updated_at); + let git_info = git_info_from_parts( + item.git_sha.clone(), + item.git_branch.clone(), + item.git_origin_url.clone(), + ); + let source = item.source.unwrap_or(SessionSource::Unknown); + let preview = item.first_user_message.clone().unwrap_or_default(); + + Some(StoredThread { + thread_id, + rollout_path: Some(item.path), + forked_from_id: None, + preview, + name: None, + model_provider: item + .model_provider + .filter(|provider| !provider.is_empty()) + .unwrap_or_else(|| default_provider.to_string()), + model: None, + reasoning_effort: None, + created_at, + updated_at, + archived_at, + cwd: item.cwd.unwrap_or_default(), + cli_version: item.cli_version.unwrap_or_default(), + source, + agent_nickname: item.agent_nickname, + agent_role: item.agent_role, + agent_path: None, + git_info, + approval_mode: AskForApproval::OnRequest, + sandbox_policy: SandboxPolicy::new_read_only_policy(), + token_usage: None, + first_user_message: item.first_user_message, + history: None, + }) +} + +fn parse_rfc3339(value: Option<&str>) -> Option> { + DateTime::parse_from_rfc3339(value?) + .ok() + .map(|dt| dt.with_timezone(&Utc)) +} + +fn git_info_from_parts( + sha: Option, + branch: Option, + origin_url: Option, +) -> Option { + if sha.is_none() && branch.is_none() && origin_url.is_none() { + return None; + } + Some(GitInfo { + commit_hash: sha.as_deref().map(GitSha::new), + branch, + repository_url: origin_url, + }) +} + +fn thread_id_from_rollout_path(path: &std::path::Path) -> Option { + let file_name = path.file_name()?.to_str()?; + let stem = file_name.strip_suffix(".jsonl")?; + if stem.len() < 37 { + return None; + } + let uuid_start = stem.len().saturating_sub(36); + if !stem[..uuid_start].ends_with('-') { + return None; + } + codex_protocol::ThreadId::from_string(&stem[uuid_start..]).ok() +} + +#[cfg(test)] +mod tests { + use std::fs; + use std::io::Write; + use std::path::Path; + use std::path::PathBuf; + + use codex_protocol::ThreadId; + use codex_protocol::protocol::SessionSource; + use codex_rollout::ARCHIVED_SESSIONS_SUBDIR; + use pretty_assertions::assert_eq; + use tempfile::TempDir; + use uuid::Uuid; + + use super::*; + + fn test_config(codex_home: &Path) -> RolloutConfig { + RolloutConfig { + codex_home: codex_home.to_path_buf(), + sqlite_home: codex_home.to_path_buf(), + cwd: codex_home.to_path_buf(), + model_provider_id: "test-provider".to_string(), + generate_memories: true, + } + } + + fn write_session_file(root: &Path, ts: &str, uuid: Uuid) -> std::io::Result { + write_session_file_with( + root, + root.join("sessions/2025/01/03"), + ts, + uuid, + "Hello from user", + Some("test-provider"), + ) + } + + fn write_archived_session_file(root: &Path, ts: &str, uuid: Uuid) -> std::io::Result { + write_session_file_with( + root, + root.join(ARCHIVED_SESSIONS_SUBDIR), + ts, + uuid, + "Archived user message", + Some("test-provider"), + ) + } + + fn write_session_file_with( + root: &Path, + day_dir: PathBuf, + ts: &str, + uuid: Uuid, + first_user_message: &str, + model_provider: Option<&str>, + ) -> std::io::Result { + fs::create_dir_all(&day_dir)?; + let path = day_dir.join(format!("rollout-{ts}-{uuid}.jsonl")); + let mut file = fs::File::create(&path)?; + let meta = serde_json::json!({ + "timestamp": ts, + "type": "session_meta", + "payload": { + "id": uuid, + "timestamp": ts, + "cwd": root, + "originator": "test_originator", + "cli_version": "test_version", + "source": "cli", + "model_provider": model_provider, + "git": { + "commit_hash": "abcdef", + "branch": "main", + "repository_url": "https://example.com/repo.git" + } + }, + }); + writeln!(file, "{meta}")?; + let user_event = serde_json::json!({ + "timestamp": ts, + "type": "event_msg", + "payload": { + "type": "user_message", + "message": first_user_message, + "kind": "plain", + }, + }); + writeln!(file, "{user_event}")?; + Ok(path) + } + + #[tokio::test] + async fn list_threads_uses_default_provider_when_rollout_omits_provider() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path())); + write_session_file_with( + home.path(), + home.path().join("sessions/2025/01/03"), + "2025-01-03T12-00-00", + Uuid::from_u128(102), + "Hello from user", + /*model_provider*/ None, + ) + .expect("session file"); + + let page = store + .list_threads(ListThreadsParams { + page_size: 10, + cursor: None, + sort_key: ThreadSortKey::CreatedAt, + allowed_sources: Vec::new(), + model_providers: None, + archived: false, + search_term: None, + }) + .await + .expect("thread listing"); + + assert_eq!(page.items.len(), 1); + assert_eq!(page.items[0].model_provider, "test-provider"); + } + + #[tokio::test] + async fn list_threads_preserves_sqlite_title_search_results() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let store = LocalThreadStore::new(config.clone()); + let uuid = Uuid::from_u128(103); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let rollout_path = home.path().join("rollout-title-search.jsonl"); + fs::write(&rollout_path, "").expect("placeholder rollout file"); + + let runtime = codex_state::StateRuntime::init( + home.path().to_path_buf(), + config.model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + runtime + .mark_backfill_complete(/*last_watermark*/ None) + .await + .expect("backfill should be complete"); + let created_at = Utc::now(); + let mut builder = codex_state::ThreadMetadataBuilder::new( + thread_id, + rollout_path, + created_at, + SessionSource::Cli, + ); + builder.model_provider = Some(config.model_provider_id.clone()); + builder.cwd = home.path().to_path_buf(); + builder.cli_version = Some("test_version".to_string()); + let mut metadata = builder.build(config.model_provider_id.as_str()); + metadata.title = "needle title".to_string(); + metadata.first_user_message = Some("plain preview".to_string()); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + + let page = store + .list_threads(ListThreadsParams { + page_size: 10, + cursor: None, + sort_key: ThreadSortKey::CreatedAt, + allowed_sources: Vec::new(), + model_providers: None, + archived: false, + search_term: Some("needle".to_string()), + }) + .await + .expect("thread listing"); + + let ids = page + .items + .iter() + .map(|item| item.thread_id) + .collect::>(); + assert_eq!(ids, vec![thread_id]); + assert_eq!( + page.items[0].first_user_message.as_deref(), + Some("plain preview") + ); + } + + #[tokio::test] + async fn list_threads_selects_active_or_archived_collection() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path())); + let active_uuid = Uuid::from_u128(105); + let archived_uuid = Uuid::from_u128(106); + write_session_file(home.path(), "2025-01-03T12-00-00", active_uuid) + .expect("active session file"); + write_archived_session_file(home.path(), "2025-01-03T13-00-00", archived_uuid) + .expect("archived session file"); + + let active = store + .list_threads(ListThreadsParams { + page_size: 10, + cursor: None, + sort_key: ThreadSortKey::CreatedAt, + allowed_sources: Vec::new(), + model_providers: None, + archived: false, + search_term: None, + }) + .await + .expect("active listing"); + let archived = store + .list_threads(ListThreadsParams { + page_size: 10, + cursor: None, + sort_key: ThreadSortKey::CreatedAt, + allowed_sources: Vec::new(), + model_providers: None, + archived: true, + search_term: None, + }) + .await + .expect("archived listing"); + + let active_id = ThreadId::from_string(&active_uuid.to_string()).expect("valid thread id"); + let archived_id = + ThreadId::from_string(&archived_uuid.to_string()).expect("valid thread id"); + assert_eq!( + active + .items + .iter() + .map(|item| item.thread_id) + .collect::>(), + vec![active_id] + ); + assert_eq!( + archived + .items + .iter() + .map(|item| item.thread_id) + .collect::>(), + vec![archived_id] + ); + assert_eq!(active.items[0].archived_at, None); + assert_eq!( + archived.items[0].archived_at, + Some(archived.items[0].updated_at) + ); + } + + #[tokio::test] + async fn list_threads_returns_local_rollout_summary() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let store = LocalThreadStore::new(config); + let uuid = Uuid::from_u128(101); + let path = + write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); + + let page = store + .list_threads(ListThreadsParams { + page_size: 10, + cursor: None, + sort_key: ThreadSortKey::CreatedAt, + allowed_sources: vec![SessionSource::Cli], + model_providers: Some(vec!["test-provider".to_string()]), + archived: false, + search_term: None, + }) + .await + .expect("thread listing"); + + let thread_id = + codex_protocol::ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + assert_eq!(page.next_cursor, None); + assert_eq!(page.items.len(), 1); + assert_eq!(page.items[0].thread_id, thread_id); + assert_eq!(page.items[0].rollout_path, Some(path)); + assert_eq!(page.items[0].preview, "Hello from user"); + assert_eq!( + page.items[0].first_user_message.as_deref(), + Some("Hello from user") + ); + assert_eq!(page.items[0].model_provider, "test-provider"); + assert_eq!(page.items[0].cli_version, "test_version"); + assert_eq!(page.items[0].source, SessionSource::Cli); + } + + #[tokio::test] + async fn list_threads_rejects_invalid_cursor() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path())); + + let err = store + .list_threads(ListThreadsParams { + page_size: 10, + cursor: Some("not-a-cursor".to_string()), + sort_key: ThreadSortKey::CreatedAt, + allowed_sources: Vec::new(), + model_providers: None, + archived: false, + search_term: None, + }) + .await + .expect_err("invalid cursor should fail"); + + assert!(matches!(err, ThreadStoreError::InvalidRequest { .. })); + } +} diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 73f983a466..8d489ea6cb 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -135,6 +135,8 @@ pub struct ThreadPage { pub struct StoredThread { /// Thread id. pub thread_id: ThreadId, + /// Local rollout path when the backing store is filesystem-based. + pub rollout_path: Option, /// Source thread id when this thread was forked from another thread. pub forked_from_id: Option, /// Best available user-facing preview, usually the first user message. @@ -155,6 +157,8 @@ pub struct StoredThread { pub archived_at: Option>, /// Working directory captured for the thread. pub cwd: PathBuf, + /// CLI version captured for the thread. + pub cli_version: String, /// Runtime source for the thread. pub source: SessionSource, /// Optional random nickname for thread-spawn sub-agents. From f53210d332f9f8a8433009bea9d1122f1e86a57f Mon Sep 17 00:00:00 2001 From: Shijie Rao Date: Wed, 15 Apr 2026 14:39:06 -0400 Subject: [PATCH 3/6] Add CLI update announcement (#17942) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary - Keep the existing local-build test announcement as the first announcement entry - Add the CLI update reminder for versions below `0.120.0` - Remove expired onboarding and gpt-5.3-codex announcement entries Screenshot 2026-04-15 at 1 32 53 PM --- announcement_tip.toml | 18 ++++++------------ 1 file changed, 6 insertions(+), 12 deletions(-) diff --git a/announcement_tip.toml b/announcement_tip.toml index 2473258368..79c0daea76 100644 --- a/announcement_tip.toml +++ b/announcement_tip.toml @@ -4,20 +4,14 @@ # version_regex matches against the CLI version (env!("CARGO_PKG_VERSION")); omit to apply to all versions. # target_app specify which app should display the announcement (cli, vsce, ...). -[[announcements]] -content = "Welcome to Codex! Check out the new onboarding flow." -from_date = "2024-10-01" -to_date = "2024-10-15" -target_app = "cli" - -# Test announcement only for local build version until 2026-01-10 excluded (past) +# Test announcement only for local build version until 2027-05-10 excluded [[announcements]] content = "This is a test announcement" version_regex = "^0\\.0\\.0$" -to_date = "2026-05-10" +to_date = "2027-05-10" [[announcements]] -content = "**BREAKING NEWS**: `gpt-5.3-codex` is out! Upgrade to `0.98.0` for a faster, smarter, more steerable agent." -from_date = "2026-02-01" -to_date = "2026-02-16" -version_regex = "^0\\.(?:[0-9]|[1-8][0-9]|9[0-7])\\." +content = "Update Required - This version will no longer be supported starting May 8th. Please upgrade to the latest version (https://github.com/openai/codex/releases/latest) using your preferred package manager." +# Matches 0.x.y versions from 0.0.y through 0.119.y; excludes 0.120.0 and newer. +version_regex = "^0\\.(?:[0-9]|[1-9][0-9]|1[01][0-9])\\." +to_date = "2026-05-08" From f5e8eac2ae15114c0a6309ae0114dd109b0a5d58 Mon Sep 17 00:00:00 2001 From: pakrym-oai Date: Wed, 15 Apr 2026 11:52:51 -0700 Subject: [PATCH 4/6] Refactor auth providers to mutate request headers (#17866) ## Summary - Move auth header construction into the `AuthProvider::add_auth_headers` contract. - Inline `CoreAuthProvider` header mutation in its provider impl and remove the shared header-map helper. - Update HTTP, websocket, file upload, sideband websocket, and test auth callsites to use the provider method. - Add direct coverage for `CoreAuthProvider` auth header mutation. ## Testing - `just fmt` - `cargo test -p codex-api` - `cargo test -p codex-core client::tests::auth_request_telemetry_context_tracks_attached_auth_and_retry_phase` - `cargo test -p codex-core` failed on unrelated/reproducible `tools::handlers::multi_agents::tests::multi_agent_v2_followup_task_interrupts_busy_child_without_losing_message` --------- Co-authored-by: Celia Chen --- codex-rs/codex-api/src/api_bridge.rs | 18 ++++++++----- codex-rs/codex-api/src/api_bridge_tests.rs | 21 +++++++++++++++ codex-rs/codex-api/src/auth.rs | 27 ++----------------- codex-rs/codex-api/src/endpoint/compact.rs | 4 +-- codex-rs/codex-api/src/endpoint/memories.rs | 4 +-- codex-rs/codex-api/src/endpoint/models.rs | 4 +-- .../codex-api/src/endpoint/realtime_call.rs | 7 +++-- .../src/endpoint/responses_websocket.rs | 3 +-- codex-rs/codex-api/src/endpoint/session.rs | 4 +-- codex-rs/codex-api/src/files.rs | 15 +++++------ codex-rs/codex-api/tests/clients.rs | 18 ++++++------- .../codex-api/tests/models_integration.rs | 4 +-- codex-rs/codex-api/tests/sse_end_to_end.rs | 4 +-- codex-rs/core/src/client.rs | 13 ++------- 14 files changed, 65 insertions(+), 81 deletions(-) diff --git a/codex-rs/codex-api/src/api_bridge.rs b/codex-rs/codex-api/src/api_bridge.rs index 0ad2b13979..7c36c67fdd 100644 --- a/codex-rs/codex-api/src/api_bridge.rs +++ b/codex-rs/codex-api/src/api_bridge.rs @@ -12,6 +12,7 @@ use codex_protocol::error::RetryLimitReachedError; use codex_protocol::error::UnexpectedResponseError; use codex_protocol::error::UsageLimitReachedError; use http::HeaderMap; +use http::HeaderValue; use serde::Deserialize; use serde_json::Value; @@ -200,11 +201,16 @@ impl CoreAuthProvider { } impl ApiAuthProvider for CoreAuthProvider { - fn bearer_token(&self) -> Option { - self.token.clone() - } - - fn account_id(&self) -> Option { - self.account_id.clone() + fn add_auth_headers(&self, headers: &mut HeaderMap) { + if let Some(token) = self.token.as_ref() + && let Ok(header) = HeaderValue::from_str(&format!("Bearer {token}")) + { + let _ = headers.insert(http::header::AUTHORIZATION, header); + } + if let Some(account_id) = self.account_id.as_ref() + && let Ok(header) = HeaderValue::from_str(account_id) + { + let _ = headers.insert("ChatGPT-Account-ID", header); + } } } diff --git a/codex-rs/codex-api/src/api_bridge_tests.rs b/codex-rs/codex-api/src/api_bridge_tests.rs index 71d3889915..50247c1314 100644 --- a/codex-rs/codex-api/src/api_bridge_tests.rs +++ b/codex-rs/codex-api/src/api_bridge_tests.rs @@ -141,3 +141,24 @@ fn core_auth_provider_reports_when_auth_header_will_attach() { assert!(auth.auth_header_attached()); assert_eq!(auth.auth_header_name(), Some("authorization")); } + +#[test] +fn core_auth_provider_adds_auth_headers() { + let auth = CoreAuthProvider::for_test(Some("access-token"), Some("workspace-123")); + let mut headers = HeaderMap::new(); + + crate::AuthProvider::add_auth_headers(&auth, &mut headers); + + assert_eq!( + headers + .get(http::header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()), + Some("Bearer access-token") + ); + assert_eq!( + headers + .get("ChatGPT-Account-ID") + .and_then(|value| value.to_str().ok()), + Some("workspace-123") + ); +} diff --git a/codex-rs/codex-api/src/auth.rs b/codex-rs/codex-api/src/auth.rs index f649062db1..a7b1e69d14 100644 --- a/codex-rs/codex-api/src/auth.rs +++ b/codex-rs/codex-api/src/auth.rs @@ -1,33 +1,10 @@ -use codex_client::Request; use http::HeaderMap; -use http::HeaderValue; -/// Provides bearer and account identity information for API requests. +/// Adds authentication headers to API requests. /// /// Implementations should be cheap and non-blocking; any asynchronous /// refresh or I/O should be handled by higher layers before requests /// reach this interface. pub trait AuthProvider: Send + Sync { - fn bearer_token(&self) -> Option; - fn account_id(&self) -> Option { - None - } -} - -pub(crate) fn add_auth_headers_to_header_map(auth: &A, headers: &mut HeaderMap) { - if let Some(token) = auth.bearer_token() - && let Ok(header) = HeaderValue::from_str(&format!("Bearer {token}")) - { - let _ = headers.insert(http::header::AUTHORIZATION, header); - } - if let Some(account_id) = auth.account_id() - && let Ok(header) = HeaderValue::from_str(&account_id) - { - let _ = headers.insert("ChatGPT-Account-ID", header); - } -} - -pub(crate) fn add_auth_headers(auth: &A, mut req: Request) -> Request { - add_auth_headers_to_header_map(auth, &mut req.headers); - req + fn add_auth_headers(&self, headers: &mut HeaderMap); } diff --git a/codex-rs/codex-api/src/endpoint/compact.rs b/codex-rs/codex-api/src/endpoint/compact.rs index 44a56a11a7..748ac35558 100644 --- a/codex-rs/codex-api/src/endpoint/compact.rs +++ b/codex-rs/codex-api/src/endpoint/compact.rs @@ -90,9 +90,7 @@ mod tests { struct DummyAuth; impl AuthProvider for DummyAuth { - fn bearer_token(&self) -> Option { - None - } + fn add_auth_headers(&self, _headers: &mut HeaderMap) {} } #[test] diff --git a/codex-rs/codex-api/src/endpoint/memories.rs b/codex-rs/codex-api/src/endpoint/memories.rs index 5cb2a65b1f..3047c859db 100644 --- a/codex-rs/codex-api/src/endpoint/memories.rs +++ b/codex-rs/codex-api/src/endpoint/memories.rs @@ -103,9 +103,7 @@ mod tests { struct DummyAuth; impl AuthProvider for DummyAuth { - fn bearer_token(&self) -> Option { - None - } + fn add_auth_headers(&self, _headers: &mut HeaderMap) {} } #[derive(Clone)] diff --git a/codex-rs/codex-api/src/endpoint/models.rs b/codex-rs/codex-api/src/endpoint/models.rs index 97781ac419..17342d6f99 100644 --- a/codex-rs/codex-api/src/endpoint/models.rs +++ b/codex-rs/codex-api/src/endpoint/models.rs @@ -132,9 +132,7 @@ mod tests { struct DummyAuth; impl AuthProvider for DummyAuth { - fn bearer_token(&self) -> Option { - None - } + fn add_auth_headers(&self, _headers: &mut HeaderMap) {} } fn provider(base_url: &str) -> Provider { diff --git a/codex-rs/codex-api/src/endpoint/realtime_call.rs b/codex-rs/codex-api/src/endpoint/realtime_call.rs index 4d0bcfa7bf..a9a8b963cf 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_call.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_call.rs @@ -284,8 +284,11 @@ mod tests { struct DummyAuth; impl AuthProvider for DummyAuth { - fn bearer_token(&self) -> Option { - Some("test-token".to_string()) + fn add_auth_headers(&self, headers: &mut HeaderMap) { + headers.insert( + http::header::AUTHORIZATION, + HeaderValue::from_static("Bearer test-token"), + ); } } diff --git a/codex-rs/codex-api/src/endpoint/responses_websocket.rs b/codex-rs/codex-api/src/endpoint/responses_websocket.rs index a60d188bda..d2b775cdd5 100644 --- a/codex-rs/codex-api/src/endpoint/responses_websocket.rs +++ b/codex-rs/codex-api/src/endpoint/responses_websocket.rs @@ -1,5 +1,4 @@ use crate::auth::AuthProvider; -use crate::auth::add_auth_headers_to_header_map; use crate::common::ResponseEvent; use crate::common::ResponseStream; use crate::common::ResponsesWsRequest; @@ -310,7 +309,7 @@ impl ResponsesWebsocketClient { let mut headers = merge_request_headers(&self.provider.headers, extra_headers, default_headers); - add_auth_headers_to_header_map(&self.auth, &mut headers); + self.auth.add_auth_headers(&mut headers); let (stream, server_reasoning_included, models_etag, server_model) = connect_websocket(ws_url, headers, turn_state.clone()).await?; diff --git a/codex-rs/codex-api/src/endpoint/session.rs b/codex-rs/codex-api/src/endpoint/session.rs index 00919a0c54..e4a470ceee 100644 --- a/codex-rs/codex-api/src/endpoint/session.rs +++ b/codex-rs/codex-api/src/endpoint/session.rs @@ -1,5 +1,4 @@ use crate::auth::AuthProvider; -use crate::auth::add_auth_headers; use crate::error::ApiError; use crate::provider::Provider; use crate::telemetry::run_with_request_telemetry; @@ -56,7 +55,8 @@ impl EndpointSession { if let Some(body) = body { req.body = Some(RequestBody::Json(body.clone())); } - add_auth_headers(&self.auth, req) + self.auth.add_auth_headers(&mut req.headers); + req } pub(crate) async fn execute( diff --git a/codex-rs/codex-api/src/files.rs b/codex-rs/codex-api/src/files.rs index 6fad5b62f5..ebe35af5ba 100644 --- a/codex-rs/codex-api/src/files.rs +++ b/codex-rs/codex-api/src/files.rs @@ -256,17 +256,14 @@ fn authorized_request( method: reqwest::Method, url: &str, ) -> reqwest::RequestBuilder { + let mut headers = http::HeaderMap::new(); + auth.add_auth_headers(&mut headers); + let client = build_reqwest_client(); - let mut request = client + client .request(method, url) - .timeout(OPENAI_FILE_REQUEST_TIMEOUT); - if let Some(token) = auth.bearer_token() { - request = request.bearer_auth(token); - } - if let Some(account_id) = auth.account_id() { - request = request.header("chatgpt-account-id", account_id); - } - request + .timeout(OPENAI_FILE_REQUEST_TIMEOUT) + .headers(headers) } fn build_reqwest_client() -> reqwest::Client { diff --git a/codex-rs/codex-api/tests/clients.rs b/codex-rs/codex-api/tests/clients.rs index b11c6f9d6b..d82fcc14ce 100644 --- a/codex-rs/codex-api/tests/clients.rs +++ b/codex-rs/codex-api/tests/clients.rs @@ -91,9 +91,7 @@ impl HttpTransport for RecordingTransport { struct NoAuth; impl AuthProvider for NoAuth { - fn bearer_token(&self) -> Option { - None - } + fn add_auth_headers(&self, _headers: &mut HeaderMap) {} } #[derive(Clone)] @@ -112,12 +110,14 @@ impl StaticAuth { } impl AuthProvider for StaticAuth { - fn bearer_token(&self) -> Option { - Some(self.token.clone()) - } - - fn account_id(&self) -> Option { - Some(self.account_id.clone()) + fn add_auth_headers(&self, headers: &mut HeaderMap) { + let token = &self.token; + if let Ok(header) = HeaderValue::from_str(&format!("Bearer {token}")) { + headers.insert(http::header::AUTHORIZATION, header); + } + if let Ok(header) = HeaderValue::from_str(&self.account_id) { + headers.insert("ChatGPT-Account-ID", header); + } } } diff --git a/codex-rs/codex-api/tests/models_integration.rs b/codex-rs/codex-api/tests/models_integration.rs index 91a8477a92..fab135e0ab 100644 --- a/codex-rs/codex-api/tests/models_integration.rs +++ b/codex-rs/codex-api/tests/models_integration.rs @@ -24,9 +24,7 @@ use wiremock::matchers::path; struct DummyAuth; impl AuthProvider for DummyAuth { - fn bearer_token(&self) -> Option { - None - } + fn add_auth_headers(&self, _headers: &mut HeaderMap) {} } fn provider(base_url: &str) -> Provider { diff --git a/codex-rs/codex-api/tests/sse_end_to_end.rs b/codex-rs/codex-api/tests/sse_end_to_end.rs index b15de296a8..4d32e82242 100644 --- a/codex-rs/codex-api/tests/sse_end_to_end.rs +++ b/codex-rs/codex-api/tests/sse_end_to_end.rs @@ -53,9 +53,7 @@ impl HttpTransport for FixtureSseTransport { struct NoAuth; impl AuthProvider for NoAuth { - fn bearer_token(&self) -> Option { - None - } + fn add_auth_headers(&self, _headers: &mut HeaderMap) {} } fn provider(name: &str) -> Provider { diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 16f743943a..bd83c81f0d 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -32,6 +32,7 @@ use std::sync::atomic::AtomicU64; use std::sync::atomic::Ordering; use codex_api::ApiError; +use codex_api::AuthProvider; use codex_api::CompactClient as ApiCompactClient; use codex_api::CompactionInput as ApiCompactionInput; use codex_api::Compression; @@ -83,7 +84,6 @@ use futures::StreamExt; use http::HeaderMap as ApiHeaderMap; use http::HeaderValue; use http::StatusCode as HttpStatusCode; -use http::header::AUTHORIZATION; use reqwest::StatusCode; use std::time::Duration; use std::time::Instant; @@ -277,16 +277,7 @@ pub(crate) struct RealtimeWebrtcCallStart { /// `api.openai.com` sideband path. fn sideband_websocket_auth_headers(api_auth: &CoreAuthProvider) -> ApiHeaderMap { let mut headers = ApiHeaderMap::new(); - if let Some(token) = api_auth.token.as_ref() - && let Ok(value) = HeaderValue::from_str(&format!("Bearer {token}")) - { - headers.insert(AUTHORIZATION, value); - } - if let Some(account_id) = api_auth.account_id.as_ref() - && let Ok(value) = HeaderValue::from_str(account_id) - { - headers.insert("ChatGPT-Account-ID", value); - } + api_auth.add_auth_headers(&mut headers); headers } From 83abf67d20e3294ae9e8be3bc99e92b89f2c9844 Mon Sep 17 00:00:00 2001 From: Ruslan Nigmatullin Date: Wed, 15 Apr 2026 11:52:53 -0700 Subject: [PATCH 5/6] app-server: track remote-control seq IDs per stream (#17902) ## Summary - Track outbound remote-control sequence IDs independently for each client stream. - Retain unacked outbound messages per stream using FIFO buffers. - Require stream-scoped acks and update tests for contiguous per-stream sequencing. ## Why The remote-control peer uses outbound sequence gaps to detect lost messages and re-initialize. A single global outbound sequence counter can create apparent gaps on an individual stream when another stream receives an interleaved message. ## Validation - `just fmt` - `cargo test -p codex-app-server remote_control` - `just fix -p codex-app-server` - `git diff --check` --- .../src/transport/remote_control/protocol.rs | 7 +- .../src/transport/remote_control/tests.rs | 36 ++- .../src/transport/remote_control/websocket.rs | 271 ++++++++++++------ 3 files changed, 216 insertions(+), 98 deletions(-) diff --git a/codex-rs/app-server/src/transport/remote_control/protocol.rs b/codex-rs/app-server/src/transport/remote_control/protocol.rs index 857855f2a0..dcb79e5c35 100644 --- a/codex-rs/app-server/src/transport/remote_control/protocol.rs +++ b/codex-rs/app-server/src/transport/remote_control/protocol.rs @@ -48,9 +48,8 @@ pub enum ClientEvent { message: JSONRPCMessage, }, /// Backend-generated acknowledgement for all server envelopes addressed to - /// `client_id` whose envelope `seq_id` is less than or equal to this ack's - /// `seq_id`. This cursor is client-scoped, not stream-scoped, so receivers - /// must not use `stream_id` to partition acks. + /// `client_id` and `stream_id` whose envelope `seq_id` is less than or equal + /// to this ack's `seq_id`. This cursor is stream-scoped. Ack, Ping, ClientClosed, @@ -65,7 +64,7 @@ pub(crate) struct ClientEnvelope { pub(crate) client_id: ClientId, #[serde(rename = "stream_id", skip_serializing_if = "Option::is_none")] pub(crate) stream_id: Option, - /// For `Ack`, this is the backend-generated per-client cursor over + /// For `Ack`, this is the backend-generated per-stream cursor over /// `ServerEnvelope.seq_id`. #[serde(rename = "seq_id", skip_serializing_if = "Option::is_none")] pub(crate) seq_id: Option, diff --git a/codex-rs/app-server/src/transport/remote_control/tests.rs b/codex-rs/app-server/src/transport/remote_control/tests.rs index 6dd6978fa5..2dd687a3d1 100644 --- a/codex-rs/app-server/src/transport/remote_control/tests.rs +++ b/codex-rs/app-server/src/transport/remote_control/tests.rs @@ -5,6 +5,7 @@ use super::enroll::update_persisted_remote_control_enrollment; use super::protocol::ClientEnvelope; use super::protocol::ClientEvent; use super::protocol::ClientId; +use super::protocol::StreamId; use super::protocol::normalize_remote_control_url; use super::websocket::REMOTE_CONTROL_PROTOCOL_VERSION; use super::*; @@ -163,7 +164,7 @@ async fn remote_control_transport_manages_virtual_clients_and_routes_messages() json!({ "type": "pong", "client_id": "client-1", - "seq_id": 0, + "seq_id": 1, "status": "unknown", }) ); @@ -369,7 +370,7 @@ async fn remote_control_transport_manages_virtual_clients_and_routes_messages() json!({ "type": "pong", "client_id": "client-1", - "seq_id": 3, + "seq_id": 1, "status": "unknown", }) ); @@ -618,12 +619,13 @@ async fn remote_control_transport_clears_outgoing_buffer_when_backend_acks() { )) .await .expect("remote writer should accept outgoing message"); + let (server_event, stream_id) = read_server_event_with_stream_id(&mut first_websocket).await; assert_eq!( - read_server_event(&mut first_websocket).await, + server_event, json!({ "type": "server_message", "client_id": "client-1", - "seq_id": 0, + "seq_id": 1, "message": { "method": "configWarning", "params": { @@ -639,8 +641,8 @@ async fn remote_control_transport_clears_outgoing_buffer_when_backend_acks() { ClientEnvelope { event: ClientEvent::Ack, client_id: client_id.clone(), - stream_id: None, - seq_id: Some(0), + stream_id: Some(stream_id), + seq_id: Some(1), cursor: None, }, ) @@ -854,7 +856,7 @@ async fn remote_control_http_mode_enrolls_before_connecting() { json!({ "type": "server_message", "client_id": backend_client_id.0.clone(), - "seq_id": 0, + "seq_id": 1, "message": { "id": 11, "result": { @@ -882,7 +884,7 @@ async fn remote_control_http_mode_enrolls_before_connecting() { json!({ "type": "server_message", "client_id": backend_client_id.0.clone(), - "seq_id": 1, + "seq_id": 2, "message": { "method": "configWarning", "params": { @@ -1356,6 +1358,12 @@ async fn send_client_event( } async fn read_server_event(websocket: &mut WebSocketStream) -> serde_json::Value { + read_server_event_with_stream_id(websocket).await.0 +} + +async fn read_server_event_with_stream_id( + websocket: &mut WebSocketStream, +) -> (serde_json::Value, StreamId) { loop { let frame = timeout(Duration::from_secs(5), websocket.next()) .await @@ -1366,13 +1374,15 @@ async fn read_server_event(websocket: &mut WebSocketStream) -> serde_ tungstenite::Message::Text(text) => { let mut event: serde_json::Value = serde_json::from_str(text.as_ref()).expect("server event should deserialize"); - if let Some(stream_id) = event + let stream_id = event .as_object_mut() .and_then(|event| event.remove("stream_id")) - { - assert!(stream_id.is_string(), "stream_id should be a string"); - } - return event; + .expect("stream_id should be present"); + let stream_id = stream_id + .as_str() + .expect("stream_id should be a string") + .to_string(); + return (event, StreamId(stream_id)); } tungstenite::Message::Ping(payload) => { websocket diff --git a/codex-rs/app-server/src/transport/remote_control/websocket.rs b/codex-rs/app-server/src/transport/remote_control/websocket.rs index d168ec0e18..82e5f35b33 100644 --- a/codex-rs/app-server/src/transport/remote_control/websocket.rs +++ b/codex-rs/app-server/src/transport/remote_control/websocket.rs @@ -14,6 +14,7 @@ use super::protocol::ClientEvent; use super::protocol::ClientId; use super::protocol::RemoteControlTarget; use super::protocol::ServerEnvelope; +use super::protocol::StreamId; use axum::http::HeaderValue; use base64::Engine; use codex_core::util::backoff; @@ -25,8 +26,8 @@ use futures::SinkExt; use futures::StreamExt; use futures::stream::SplitSink; use futures::stream::SplitStream; -use std::collections::BTreeMap; use std::collections::HashMap; +use std::collections::VecDeque; use std::io; use std::io::ErrorKind; use std::sync::Arc; @@ -57,10 +58,7 @@ const REMOTE_CONTROL_ACCOUNT_ID_RETRY_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1); struct BoundedOutboundBuffer { - // Remote-control acks are generated by the backend at client scope, so - // retransmit retention is keyed by client_id only. stream_id stays on each - // envelope for routing, but it is not part of the ack cursor. - buffer_by_client: HashMap>, + buffer_by_stream: HashMap<(ClientId, StreamId), VecDeque>, used_tx: watch::Sender, } @@ -68,46 +66,50 @@ impl BoundedOutboundBuffer { fn new() -> (Self, watch::Receiver) { let (used_tx, used_rx) = watch::channel(0); let buffer = Self { - buffer_by_client: HashMap::new(), + buffer_by_stream: HashMap::new(), used_tx, }; (buffer, used_rx) } fn insert(&mut self, server_envelope: &ServerEnvelope) { - self.buffer_by_client - .entry(server_envelope.client_id.clone()) + self.buffer_by_stream + .entry(( + server_envelope.client_id.clone(), + server_envelope.stream_id.clone(), + )) .or_default() - .insert(server_envelope.seq_id, server_envelope.clone()); + .push_back(server_envelope.clone()); self.used_tx.send_modify(|used| *used += 1); } - fn ack(&mut self, client_id: &ClientId, acked_seq_id: u64) { - let Some(buffer) = self.buffer_by_client.get_mut(client_id) else { + fn ack(&mut self, client_id: &ClientId, stream_id: &StreamId, acked_seq_id: u64) { + let key = (client_id.clone(), stream_id.clone()); + let Some(buffer) = self.buffer_by_stream.get_mut(&key) else { return; }; - while let Some(seq_id) = buffer.first_key_value().map(|(seq_id, _)| seq_id) - && *seq_id <= acked_seq_id + while let Some(server_envelope) = buffer.front() + && server_envelope.seq_id <= acked_seq_id { - buffer.pop_first(); + buffer.pop_front(); self.used_tx.send_modify(|used| *used -= 1); } if buffer.is_empty() { - self.buffer_by_client.remove(client_id); + self.buffer_by_stream.remove(&key); } } fn server_envelopes(&self) -> impl Iterator { - self.buffer_by_client + self.buffer_by_stream .values() - .flat_map(|buffer| buffer.values()) + .flat_map(|buffer| buffer.iter()) } } struct WebsocketState { outbound_buffer: BoundedOutboundBuffer, subscribe_cursor: Option, - next_seq_id: u64, + next_seq_id_by_stream: HashMap<(ClientId, StreamId), u64>, } pub(crate) struct RemoteControlWebsocket { @@ -162,7 +164,7 @@ impl RemoteControlWebsocket { state: Arc::new(Mutex::new(WebsocketState { outbound_buffer, subscribe_cursor: None, - next_seq_id: 0, + next_seq_id_by_stream: HashMap::new(), })), server_event_rx: Arc::new(Mutex::new(server_event_rx)), used_rx, @@ -445,10 +447,16 @@ impl RemoteControlWebsocket { } } }; - let (server_envelope, write_complete_tx) = { + let (payload, write_complete_tx) = { let mut state = state.lock().await; - let seq_id = state.next_seq_id; - state.next_seq_id = state.next_seq_id.saturating_add(1); + let seq_key = ( + queued_server_envelope.client_id.clone(), + queued_server_envelope.stream_id.clone(), + ); + let seq_id = *state + .next_seq_id_by_stream + .entry(seq_key.clone()) + .or_insert(1); let server_envelope = ServerEnvelope { event: queued_server_envelope.event, @@ -456,17 +464,19 @@ impl RemoteControlWebsocket { seq_id, stream_id: queued_server_envelope.stream_id, }; + let payload = match serde_json::to_string(&server_envelope) { + Ok(payload) => payload, + Err(err) => { + error!("failed to serialize remote-control server event: {err}"); + continue; + } + }; + state + .next_seq_id_by_stream + .insert(seq_key, seq_id.saturating_add(1)); state.outbound_buffer.insert(&server_envelope); - (server_envelope, queued_server_envelope.write_complete_tx) - }; - - let payload = match serde_json::to_string(&server_envelope) { - Ok(payload) => payload, - Err(err) => { - error!("failed to serialize remote-control server event: {err}"); - continue; - } + (payload, queued_server_envelope.write_complete_tx) }; tokio::select! { @@ -590,10 +600,13 @@ impl RemoteControlWebsocket { } if let ClientEvent::Ack = &client_envelope.event && let Some(acked_seq_id) = client_envelope.seq_id + && let Some(stream_id) = client_envelope.stream_id.as_ref() { - websocket_state - .outbound_buffer - .ack(&client_envelope.client_id, acked_seq_id); + websocket_state.outbound_buffer.ack( + &client_envelope.client_id, + stream_id, + acked_seq_id, + ); } drop(websocket_state); @@ -1220,7 +1233,7 @@ mod tests { let state = Arc::new(Mutex::new(WebsocketState { outbound_buffer, subscribe_cursor: None, - next_seq_id: 0, + next_seq_id_by_stream: HashMap::new(), })); let (_server_event_tx, server_event_rx) = mpsc::channel(super::super::CHANNEL_CAPACITY); let server_event_rx = Arc::new(Mutex::new(server_event_rx)); @@ -1248,6 +1261,83 @@ mod tests { .expect("writer should stop cleanly"); } + #[tokio::test] + async fn run_server_writer_inner_assigns_contiguous_seq_ids_per_stream() { + let (client_stream, mut server_stream) = connected_websocket_pair().await; + let (websocket_writer, _websocket_reader) = client_stream.split(); + let (outbound_buffer, used_rx) = BoundedOutboundBuffer::new(); + let state = Arc::new(Mutex::new(WebsocketState { + outbound_buffer, + subscribe_cursor: None, + next_seq_id_by_stream: HashMap::new(), + })); + let (server_event_tx, server_event_rx) = mpsc::channel(super::super::CHANNEL_CAPACITY); + let server_event_rx = Arc::new(Mutex::new(server_event_rx)); + let shutdown_token = CancellationToken::new(); + let writer_task = tokio::spawn(RemoteControlWebsocket::run_server_writer_inner( + state, + server_event_rx, + used_rx, + websocket_writer, + Duration::from_secs(60), + shutdown_token.clone(), + )); + + let client_id = ClientId("client-1".to_string()); + let first_stream = StreamId("stream-1".to_string()); + let second_stream = StreamId("stream-2".to_string()); + for stream_id in [&first_stream, &second_stream, &first_stream] { + server_event_tx + .send(super::super::QueuedServerEnvelope { + event: ServerEvent::Pong { + status: crate::transport::remote_control::protocol::PongStatus::Active, + }, + client_id: client_id.clone(), + stream_id: stream_id.clone(), + write_complete_tx: None, + }) + .await + .expect("server event should queue"); + } + + assert_eq!( + read_server_text_event(&mut server_stream).await, + serde_json::json!({ + "type": "pong", + "client_id": "client-1", + "stream_id": "stream-1", + "seq_id": 1, + "status": "active", + }) + ); + assert_eq!( + read_server_text_event(&mut server_stream).await, + serde_json::json!({ + "type": "pong", + "client_id": "client-1", + "stream_id": "stream-2", + "seq_id": 1, + "status": "active", + }) + ); + assert_eq!( + read_server_text_event(&mut server_stream).await, + serde_json::json!({ + "type": "pong", + "client_id": "client-1", + "stream_id": "stream-1", + "seq_id": 2, + "status": "active", + }) + ); + + shutdown_token.cancel(); + writer_task + .await + .expect("writer task should join") + .expect("writer should stop cleanly"); + } + #[tokio::test] async fn run_websocket_reader_inner_times_out_without_pong_frames() { let (client_stream, _server_stream) = connected_websocket_pair().await; @@ -1256,7 +1346,7 @@ mod tests { let state = Arc::new(Mutex::new(WebsocketState { outbound_buffer, subscribe_cursor: None, - next_seq_id: 0, + next_seq_id_by_stream: HashMap::new(), })); let (server_event_tx, _server_event_rx) = mpsc::channel(super::super::CHANNEL_CAPACITY); let (transport_event_tx, _transport_event_rx) = @@ -1287,70 +1377,32 @@ mod tests { } #[test] - fn outbound_buffer_acks_by_client_id_across_stream_ids() { + fn outbound_buffer_acks_by_stream_id() { let (mut outbound_buffer, used_rx) = BoundedOutboundBuffer::new(); let client_1 = ClientId("client-1".to_string()); let client_2 = ClientId("client-2".to_string()); + let stream_1 = StreamId("stream-1".to_string()); outbound_buffer.insert(&server_envelope( &client_1, "stream-1", - /*seq_id*/ 0, + /*seq_id*/ 1, "first-client-old-stream", )); outbound_buffer.insert(&server_envelope( &client_2, "stream-1", - /*seq_id*/ 1, + /*seq_id*/ 2, "second-client", )); outbound_buffer.insert(&server_envelope( &client_1, "stream-2", - /*seq_id*/ 2, + /*seq_id*/ 3, "first-client-new-stream", )); - outbound_buffer.ack(&client_1, /*acked_seq_id*/ 2); - - let mut retained = outbound_buffer - .server_envelopes() - .map(|server_envelope| { - ( - server_envelope.client_id.0.as_str(), - server_envelope.stream_id.0.as_str(), - server_envelope.seq_id, - ) - }) - .collect::>(); - retained.sort_unstable(); - assert_eq!(retained, vec![("client-2", "stream-1", 1)]); - assert_eq!(*used_rx.borrow(), 1); - } - - #[test] - fn outbound_buffer_retains_unacked_messages_until_ack_advances() { - let (mut outbound_buffer, used_rx) = BoundedOutboundBuffer::new(); - let client_1 = ClientId("client-1".to_string()); - let client_2 = ClientId("client-2".to_string()); - - outbound_buffer.insert(&server_envelope( - &client_1, - "stream-1", - /*seq_id*/ 0, - "first-old", - )); - outbound_buffer.insert(&server_envelope( - &client_1, - "stream-2", - /*seq_id*/ 1, - "first-new", - )); - outbound_buffer.insert(&server_envelope( - &client_2, "stream-1", /*seq_id*/ 2, "second", - )); - - outbound_buffer.ack(&client_1, /*acked_seq_id*/ 0); + outbound_buffer.ack(&client_1, &stream_1, /*acked_seq_id*/ 3); let mut retained = outbound_buffer .server_envelopes() @@ -1365,7 +1417,50 @@ mod tests { retained.sort_unstable(); assert_eq!( retained, - vec![("client-1", "stream-2", 1), ("client-2", "stream-1", 2)] + vec![("client-1", "stream-2", 3), ("client-2", "stream-1", 2)] + ); + assert_eq!(*used_rx.borrow(), 2); + } + + #[test] + fn outbound_buffer_retains_unacked_messages_until_ack_advances() { + let (mut outbound_buffer, used_rx) = BoundedOutboundBuffer::new(); + let client_1 = ClientId("client-1".to_string()); + let client_2 = ClientId("client-2".to_string()); + let stream_1 = StreamId("stream-1".to_string()); + + outbound_buffer.insert(&server_envelope( + &client_1, + "stream-1", + /*seq_id*/ 1, + "first-old", + )); + outbound_buffer.insert(&server_envelope( + &client_1, + "stream-2", + /*seq_id*/ 2, + "first-new", + )); + outbound_buffer.insert(&server_envelope( + &client_2, "stream-1", /*seq_id*/ 3, "second", + )); + + outbound_buffer.ack(&client_1, &stream_1, /*acked_seq_id*/ 1); + + let mut retained = outbound_buffer + .server_envelopes() + .map(|server_envelope| { + ( + server_envelope.client_id.0.as_str(), + server_envelope.stream_id.0.as_str(), + server_envelope.seq_id, + ) + }) + .collect::>(); + retained.sort_unstable(); + assert_eq!( + retained, + vec![("client-1", "stream-2", 2), ("client-2", "stream-1", 3)] ); assert_eq!(*used_rx.borrow(), 2); } @@ -1450,6 +1545,20 @@ mod tests { (client_stream, server_stream) } + async fn read_server_text_event( + server_stream: &mut WebSocketStream, + ) -> serde_json::Value { + let message = timeout(Duration::from_secs(5), server_stream.next()) + .await + .expect("server event should arrive in time") + .expect("server websocket should stay open") + .expect("server event should read"); + let tungstenite::Message::Text(text) = message else { + panic!("expected text event, got {message:?}"); + }; + serde_json::from_str(text.as_ref()).expect("server event should deserialize") + } + async fn respond_with_status_and_headers( mut stream: TcpStream, status: &str, From 66533ddc61e2795ea58f23cf3b88515d6085ee67 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Wed, 15 Apr 2026 12:02:40 -0700 Subject: [PATCH 6/6] mcp: remove codex/sandbox-state custom request support (#17957) ## Why #17763 moved sandbox-state delivery for MCP tool calls to request `_meta` via the `codex/sandbox-state-meta` experimental capability. Keeping the older `codex/sandbox-state` capability meant Codex still maintained a second transport that pushed updates with the custom `codex/sandbox-state/update` request at server startup and when the session sandbox policy changed. That duplicate MCP path is redundant with the per-tool-call metadata path and makes the sandbox-state contract larger than needed. The existing managed network proxy refresh on sandbox-policy changes is still needed, so this keeps that behavior separate from the removed MCP notification. ## What Changed - Removed the exported `MCP_SANDBOX_STATE_CAPABILITY` and `MCP_SANDBOX_STATE_METHOD` constants. - Removed detection of `codex/sandbox-state` during MCP initialization and stopped sending `codex/sandbox-state/update` at server startup. - Removed the `McpConnectionManager::notify_sandbox_state_change` plumbing while preserving the managed network proxy refresh when a user turn changes sandbox policy. - Slimmed `McpConnectionManager::new` so startup paths pass only the initial `SandboxPolicy` needed for MCP elicitation state. - Kept `codex/sandbox-state-meta` support intact; servers that opt in still receive the current `SandboxState` on tool-call request `_meta` ([remaining call path](https://github.com/openai/codex/blob/ff2d3c1e72ff08ce13743b99605d19d338edd51c/codex-rs/core/src/mcp_tool_call.rs#L487-L526)). - Added regression coverage for refreshing the live managed network proxy on a per-turn sandbox-policy change. ## Verification - `cargo test -p codex-core new_turn_refreshes_managed_network_proxy_for_sandbox_change` - `cargo test -p codex-mcp` --- codex-rs/codex-mcp/src/lib.rs | 2 - codex-rs/codex-mcp/src/mcp/mod.rs | 20 +---- .../codex-mcp/src/mcp_connection_manager.rs | 85 +------------------ codex-rs/core/src/codex.rs | 48 ++--------- codex-rs/core/src/codex_tests.rs | 83 ++++++++++++++++++ codex-rs/core/src/connectors.rs | 12 +-- codex-rs/core/src/lib.rs | 2 - 7 files changed, 97 insertions(+), 155 deletions(-) diff --git a/codex-rs/codex-mcp/src/lib.rs b/codex-rs/codex-mcp/src/lib.rs index 36007e6c92..ed0d9b4122 100644 --- a/codex-rs/codex-mcp/src/lib.rs +++ b/codex-rs/codex-mcp/src/lib.rs @@ -36,9 +36,7 @@ pub use mcp::tool_plugin_provenance; pub use mcp::with_codex_apps_mcp; pub use mcp_connection_manager::CodexAppsToolsCacheKey; pub use mcp_connection_manager::DEFAULT_STARTUP_TIMEOUT; -pub use mcp_connection_manager::MCP_SANDBOX_STATE_CAPABILITY; pub use mcp_connection_manager::MCP_SANDBOX_STATE_META_CAPABILITY; -pub use mcp_connection_manager::MCP_SANDBOX_STATE_METHOD; pub use mcp_connection_manager::McpConnectionManager; pub use mcp_connection_manager::SandboxState; pub use mcp_connection_manager::ToolInfo; diff --git a/codex-rs/codex-mcp/src/mcp/mod.rs b/codex-rs/codex-mcp/src/mcp/mod.rs index d53532b951..421ed14c36 100644 --- a/codex-rs/codex-mcp/src/mcp/mod.rs +++ b/codex-rs/codex-mcp/src/mcp/mod.rs @@ -35,7 +35,6 @@ use codex_protocol::protocol::SandboxPolicy; use serde_json::Value; use crate::mcp_connection_manager::McpConnectionManager; -use crate::mcp_connection_manager::SandboxState; use crate::mcp_connection_manager::codex_apps_tools_cache_key; pub type McpManager = McpConnectionManager; @@ -347,14 +346,6 @@ pub async fn collect_mcp_snapshot_with_detail( let (tx_event, rx_event) = unbounded(); drop(rx_event); - // Use ReadOnly sandbox policy for MCP snapshot collection (safest default) - let sandbox_state = SandboxState { - sandbox_policy: SandboxPolicy::new_read_only_policy(), - codex_linux_sandbox_exe: config.codex_linux_sandbox_exe.clone(), - sandbox_cwd: env::current_dir().unwrap_or_else(|_| PathBuf::from("/")), - use_legacy_landlock: config.use_legacy_landlock, - }; - let (mcp_connection_manager, cancel_token) = McpConnectionManager::new( &mcp_servers, config.mcp_oauth_credentials_store_mode, @@ -362,7 +353,7 @@ pub async fn collect_mcp_snapshot_with_detail( &config.approval_policy, submit_id, tx_event, - sandbox_state, + SandboxPolicy::new_read_only_policy(), config.codex_home.clone(), codex_apps_tools_cache_key(auth), tool_plugin_provenance, @@ -421,13 +412,6 @@ pub async fn collect_mcp_server_status_snapshot_with_detail( let (tx_event, rx_event) = unbounded(); drop(rx_event); - let sandbox_state = SandboxState { - sandbox_policy: SandboxPolicy::new_read_only_policy(), - codex_linux_sandbox_exe: config.codex_linux_sandbox_exe.clone(), - sandbox_cwd: env::current_dir().unwrap_or_else(|_| PathBuf::from("/")), - use_legacy_landlock: config.use_legacy_landlock, - }; - let (mcp_connection_manager, cancel_token) = McpConnectionManager::new( &mcp_servers, config.mcp_oauth_credentials_store_mode, @@ -435,7 +419,7 @@ pub async fn collect_mcp_server_status_snapshot_with_detail( &config.approval_policy, submit_id, tx_event, - sandbox_state, + SandboxPolicy::new_read_only_policy(), config.codex_home.clone(), codex_apps_tools_cache_key(auth), tool_plugin_provenance, diff --git a/codex-rs/codex-mcp/src/mcp_connection_manager.rs b/codex-rs/codex-mcp/src/mcp_connection_manager.rs index fadb07d581..1542c7b30b 100644 --- a/codex-rs/codex-mcp/src/mcp_connection_manager.rs +++ b/codex-rs/codex-mcp/src/mcp_connection_manager.rs @@ -439,7 +439,6 @@ struct ManagedClient { tool_filter: ToolFilter, tool_timeout: Option, server_instructions: Option, - server_supports_sandbox_state_capability: bool, server_supports_sandbox_state_meta_capability: bool, codex_apps_tools_cache_context: Option, } @@ -469,22 +468,6 @@ impl ManagedClient { self.tools.clone() } - - /// Returns once the server has ack'd the sandbox state update. - async fn notify_sandbox_state_change(&self, sandbox_state: &SandboxState) -> Result<()> { - if !self.server_supports_sandbox_state_capability { - return Ok(()); - } - - let _response = self - .client - .send_custom_request( - MCP_SANDBOX_STATE_METHOD, - Some(serde_json::to_value(sandbox_state)?), - ) - .await?; - Ok(()) - } } #[derive(Clone)] @@ -642,19 +625,8 @@ impl AsyncManagedClient { }; tools.map(annotate_tools) } - - async fn notify_sandbox_state_change(&self, sandbox_state: &SandboxState) -> Result<()> { - let managed = self.client().await?; - managed.notify_sandbox_state_change(sandbox_state).await - } } -pub const MCP_SANDBOX_STATE_CAPABILITY: &str = "codex/sandbox-state"; - -/// Custom MCP request to push sandbox state updates. -/// When used, the `params` field of the notification is [`SandboxState`]. -pub const MCP_SANDBOX_STATE_METHOD: &str = "codex/sandbox-state/update"; - /// 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"; @@ -735,7 +707,7 @@ impl McpConnectionManager { approval_policy: &Constrained, submit_id: String, tx_event: Sender, - initial_sandbox_state: SandboxState, + initial_sandbox_policy: SandboxPolicy, codex_home: PathBuf, codex_apps_tools_cache_key: CodexAppsToolsCacheKey, tool_plugin_provenance: ToolPluginProvenance, @@ -744,10 +716,8 @@ impl McpConnectionManager { 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_sandbox_state.sandbox_policy.clone(), - ); + let elicitation_requests = + ElicitationRequestManager::new(approval_policy.value(), initial_sandbox_policy); let tool_plugin_provenance = Arc::new(tool_plugin_provenance); let startup_submit_id = submit_id.clone(); let mcp_servers = mcp_servers.clone(); @@ -787,25 +757,13 @@ impl McpConnectionManager { let tx_event = tx_event.clone(); let submit_id = startup_submit_id.clone(); let auth_entry = auth_entries.get(&server_name).cloned(); - let sandbox_state = initial_sandbox_state.clone(); join_set.spawn(async move { let outcome = async_managed_client.client().await; if cancel_token.is_cancelled() { return (server_name, Err(StartupOutcomeError::Cancelled)); } let status = match &outcome { - Ok(_) => { - // Send sandbox state notification immediately after Ready - if let Err(e) = async_managed_client - .notify_sandbox_state_change(&sandbox_state) - .await - { - warn!( - "Failed to notify sandbox state to MCP server {server_name}: {e:#}", - ); - } - McpStartupStatus::Ready - } + Ok(_) => McpStartupStatus::Ready, Err(error) => { let error_str = mcp_init_error_display( server_name.as_str(), @@ -1219,34 +1177,6 @@ impl McpConnectionManager { .into_values() .find(|tool| tool.canonical_tool_name() == *tool_name) } - - pub async fn notify_sandbox_state_change(&self, sandbox_state: &SandboxState) -> Result<()> { - let mut join_set = JoinSet::new(); - - for async_managed_client in self.clients.values() { - let sandbox_state = sandbox_state.clone(); - let async_managed_client = async_managed_client.clone(); - join_set.spawn(async move { - async_managed_client - .notify_sandbox_state_change(&sandbox_state) - .await - }); - } - - while let Some(join_res) = join_set.join_next().await { - match join_res { - Ok(Ok(())) => {} - Ok(Err(err)) => { - warn!("Failed to notify sandbox state change to MCP server: {err:#}"); - } - Err(err) => { - warn!("Task panic when notifying sandbox state change to MCP server: {err:#}"); - } - } - } - - Ok(()) - } } async fn emit_update( @@ -1492,12 +1422,6 @@ async fn start_server_task( .await .map_err(StartupOutcomeError::from)?; - let server_supports_sandbox_state_capability = initialize_result - .capabilities - .experimental - .as_ref() - .and_then(|exp| exp.get(MCP_SANDBOX_STATE_CAPABILITY)) - .is_some(); let server_supports_sandbox_state_meta_capability = initialize_result .capabilities .experimental @@ -1539,7 +1463,6 @@ async fn start_server_task( tool_timeout: Some(tool_timeout), tool_filter, server_instructions: initialize_result.instructions, - server_supports_sandbox_state_capability, server_supports_sandbox_state_meta_capability, codex_apps_tools_cache_context, }; diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index bc7cd07e4c..cc314a66e9 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -78,7 +78,6 @@ use codex_login::CodexAuth; use codex_login::auth_env_telemetry::collect_auth_env_telemetry; use codex_login::default_client::originator; use codex_mcp::McpConnectionManager; -use codex_mcp::SandboxState; use codex_mcp::ToolInfo; use codex_mcp::codex_apps_tools_cache_key; #[cfg(test)] @@ -2212,14 +2211,6 @@ impl Session { // Start the watcher after SessionConfigured so it cannot emit earlier events. sess.start_skills_watcher_listener(); sess.start_agent_identity_registration(); - // Construct sandbox_state before MCP startup so it can be sent to each - // MCP server immediately after it becomes ready (avoiding blocking). - let sandbox_state = SandboxState { - sandbox_policy: session_configuration.sandbox_policy.get().clone(), - codex_linux_sandbox_exe: config.codex_linux_sandbox_exe.clone(), - sandbox_cwd: session_configuration.cwd.to_path_buf(), - use_legacy_landlock: config.features.use_legacy_landlock(), - }; let mut required_mcp_servers: Vec = mcp_servers .iter() .filter(|(_, server)| server.enabled && server.required) @@ -2241,7 +2232,7 @@ impl Session { &session_configuration.approval_policy, INITIAL_SUBMIT_ID.to_owned(), tx_event.clone(), - sandbox_state, + session_configuration.sandbox_policy.get().clone(), config.codex_home.to_path_buf(), codex_apps_tools_cache_key(auth), tool_plugin_provenance, @@ -2673,12 +2664,16 @@ impl Session { &session_source, ); + if sandbox_policy_changed { + self.refresh_managed_network_proxy_for_current_sandbox_policy() + .await; + } + Ok(self .new_turn_from_configuration( sub_id, session_configuration, updates.final_output_json_schema, - sandbox_policy_changed, ) .await) } @@ -2688,7 +2683,6 @@ impl Session { sub_id: String, session_configuration: SessionConfiguration, final_output_json_schema: Option>, - sandbox_policy_changed: bool, ) -> Arc { let per_turn_config = Self::build_per_turn_config(&session_configuration); { @@ -2698,27 +2692,6 @@ impl Session { .set_sandbox_policy(per_turn_config.permissions.sandbox_policy.get()); } - if sandbox_policy_changed { - self.refresh_managed_network_proxy_for_current_sandbox_policy() - .await; - let sandbox_state = SandboxState { - sandbox_policy: per_turn_config.permissions.sandbox_policy.get().clone(), - codex_linux_sandbox_exe: per_turn_config.codex_linux_sandbox_exe.clone(), - sandbox_cwd: per_turn_config.cwd.to_path_buf(), - use_legacy_landlock: per_turn_config.features.use_legacy_landlock(), - }; - if let Err(e) = self - .services - .mcp_connection_manager - .read() - .await - .notify_sandbox_state_change(&sandbox_state) - .await - { - warn!("Failed to notify sandbox state change to MCP servers: {e:#}"); - } - } - let model_info = self .services .models_manager @@ -2867,7 +2840,6 @@ impl Session { sub_id, session_configuration, /*final_output_json_schema*/ None, - /*sandbox_policy_changed*/ false, ) .await } @@ -4579,12 +4551,6 @@ impl Session { .await; let mcp_servers = with_codex_apps_mcp(mcp_servers, auth.as_ref(), &mcp_config); let auth_statuses = compute_auth_statuses(mcp_servers.iter(), store_mode).await; - let sandbox_state = SandboxState { - sandbox_policy: turn_context.sandbox_policy.get().clone(), - codex_linux_sandbox_exe: turn_context.codex_linux_sandbox_exe.clone(), - sandbox_cwd: turn_context.cwd.to_path_buf(), - use_legacy_landlock: turn_context.features.use_legacy_landlock(), - }; { let mut guard = self.services.mcp_startup_cancellation_token.lock().await; guard.cancel(); @@ -4597,7 +4563,7 @@ impl Session { &turn_context.config.permissions.approval_policy, turn_context.sub_id.clone(), self.get_tx_event(), - sandbox_state, + turn_context.sandbox_policy.get().clone(), config.codex_home.to_path_buf(), codex_apps_tools_cache_key(auth.as_ref()), tool_plugin_provenance, diff --git a/codex-rs/core/src/codex_tests.rs b/codex-rs/core/src/codex_tests.rs index fb7bbaadba..eaa2099098 100644 --- a/codex-rs/core/src/codex_tests.rs +++ b/codex-rs/core/src/codex_tests.rs @@ -656,6 +656,89 @@ async fn managed_network_proxy_decider_survives_full_access_start() -> anyhow::R Ok(()) } +#[tokio::test] +async fn new_turn_refreshes_managed_network_proxy_for_sandbox_change() -> anyhow::Result<()> { + let (mut session, _turn_context) = make_session_and_context().await; + let initial_policy = SandboxPolicy::new_workspace_write_policy(); + + let mut network_config = NetworkProxyConfig::default(); + network_config + .network + .set_allowed_domains(vec!["evil.com".to_string()]); + let requirements = NetworkConstraints { + domains: Some(NetworkDomainPermissionsToml { + entries: std::collections::BTreeMap::from([( + "*.example.com".to_string(), + NetworkDomainPermissionToml::Allow, + )]), + }), + ..Default::default() + }; + let spec = crate::config::NetworkProxySpec::from_config_and_constraints( + network_config, + Some(requirements), + &initial_policy, + )?; + let (started_proxy, _) = Session::start_managed_network_proxy( + &spec, + &Policy::empty(), + &initial_policy, + /*network_policy_decider*/ None, + /*blocked_request_observer*/ None, + /*managed_network_requirements_enabled*/ false, + crate::config::NetworkProxyAuditMetadata::default(), + ) + .await?; + assert_eq!( + started_proxy + .proxy() + .current_cfg() + .await? + .network + .allowed_domains(), + Some(vec!["*.example.com".to_string(), "evil.com".to_string()]) + ); + + { + let mut state = session.state.lock().await; + let mut config = (*state.session_configuration.original_config_do_not_use).clone(); + config.permissions.network = Some(spec); + config.permissions.sandbox_policy = + codex_config::Constrained::allow_any(initial_policy.clone()); + state.session_configuration.original_config_do_not_use = Arc::new(config); + state.session_configuration.sandbox_policy = + codex_config::Constrained::allow_any(initial_policy); + } + session.services.network_proxy = Some(started_proxy); + + session + .new_turn_with_sub_id( + "sandbox-policy-change".to_string(), + SessionSettingsUpdate { + sandbox_policy: Some(SandboxPolicy::DangerFullAccess), + ..Default::default() + }, + ) + .await?; + + let started_proxy = session + .services + .network_proxy + .as_ref() + .expect("managed network proxy should be present"); + assert_eq!( + started_proxy + .proxy() + .current_cfg() + .await? + .network + .allowed_domains(), + Some(vec!["*.example.com".to_string()]) + ); + + Ok(()) +} + #[tokio::test] async fn get_base_instructions_no_user_content() { let prompt_with_apply_patch_instructions = diff --git a/codex-rs/core/src/connectors.rs b/codex-rs/core/src/connectors.rs index eef6d5747b..77603adcf1 100644 --- a/codex-rs/core/src/connectors.rs +++ b/codex-rs/core/src/connectors.rs @@ -1,8 +1,6 @@ use std::collections::BTreeSet; use std::collections::HashMap; use std::collections::HashSet; -use std::env; -use std::path::PathBuf; use std::sync::Arc; use std::sync::LazyLock; use std::sync::Mutex as StdMutex; @@ -42,7 +40,6 @@ use codex_login::default_client::is_first_party_chat_originator; use codex_login::default_client::originator; use codex_mcp::CODEX_APPS_MCP_SERVER_NAME; use codex_mcp::McpConnectionManager; -use codex_mcp::SandboxState; use codex_mcp::ToolInfo; use codex_mcp::ToolPluginProvenance; use codex_mcp::codex_apps_tools_cache_key; @@ -228,13 +225,6 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_options_and_status( let (tx_event, rx_event) = unbounded(); drop(rx_event); - let sandbox_state = SandboxState { - sandbox_policy: SandboxPolicy::new_read_only_policy(), - codex_linux_sandbox_exe: config.codex_linux_sandbox_exe.clone(), - sandbox_cwd: env::current_dir().unwrap_or_else(|_| PathBuf::from("/")), - use_legacy_landlock: config.features.use_legacy_landlock(), - }; - let (mcp_connection_manager, cancel_token) = McpConnectionManager::new( &mcp_servers, config.mcp_oauth_credentials_store_mode, @@ -242,7 +232,7 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_options_and_status( &config.permissions.approval_policy, INITIAL_SUBMIT_ID.to_owned(), tx_event, - sandbox_state, + SandboxPolicy::new_read_only_policy(), config.codex_home.to_path_buf(), codex_apps_tools_cache_key(auth.as_ref()), ToolPluginProvenance::default(), diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 37b5dcf5d4..8d463f3b51 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -55,8 +55,6 @@ pub use network_proxy_loader::MtimeConfigReloader; pub use network_proxy_loader::build_network_proxy_state; pub use network_proxy_loader::build_network_proxy_state_and_reloader; mod original_image_detail; -pub use codex_mcp::MCP_SANDBOX_STATE_CAPABILITY; -pub use codex_mcp::MCP_SANDBOX_STATE_METHOD; pub use codex_mcp::SandboxState; mod mcp_openai_file; mod mcp_tool_call;