diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 9bbfdf4f2..795b5d4c3 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -13,10 +13,8 @@ use crate::outgoing_message::RequestContext; use crate::outgoing_message::ThreadScopedOutgoingMessageSender; use crate::thread_status::ThreadWatchManager; use crate::thread_status::resolve_thread_status; -use chrono::DateTime; use chrono::Duration as ChronoDuration; use chrono::SecondsFormat; -use chrono::Utc; use codex_analytics::AnalyticsEventsClient; use codex_analytics::AnalyticsJsonRpcError; use codex_analytics::InputError; @@ -495,6 +493,7 @@ pub(crate) use self::thread_lifecycle::populate_thread_turns_from_history; pub(crate) use self::thread_processor::thread_from_stored_thread; #[cfg(test)] pub(crate) use self::thread_summary::read_summary_from_rollout; +#[cfg(test)] pub(crate) use self::thread_summary::summary_to_thread; pub(crate) fn build_api_turns_from_rollout_items(items: &[RolloutItem]) -> Vec { diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index cb1b9d31b..d83f5e631 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -1605,11 +1605,8 @@ impl ThreadRequestProcessor { .unarchive_thread(StoreArchiveThreadParams { thread_id }) .await .map_err(|err| thread_store_archive_error("unarchive", err))?; - let summary = summary_from_stored_thread(stored_thread, fallback_provider.as_str()) - .ok_or_else(|| { - internal_error(format!("failed to read unarchived thread {thread_id}")) - })?; - let mut thread = summary_to_thread(summary, &self.config.cwd); + let (mut thread, _) = + thread_from_stored_thread(stored_thread, fallback_provider.as_str(), &self.config.cwd); thread.status = resolve_thread_status( self.thread_watch_manager @@ -3215,12 +3212,7 @@ impl ThreadRequestProcessor { }; let stored_thread = read_result?; - let summary = - summary_from_stored_thread(stored_thread, fallback_provider).ok_or_else(|| { - internal_error( - "failed to load conversation summary: thread is missing rollout path", - ) - })?; + let summary = summary_from_stored_thread(stored_thread, fallback_provider); Ok(GetConversationSummaryResponse { summary }) } @@ -3697,8 +3689,8 @@ pub(crate) fn thread_from_stored_thread( fn summary_from_stored_thread( thread: StoredThread, fallback_provider: &str, -) -> Option { - let path = thread.rollout_path?; +) -> ConversationSummary { + let path = thread.rollout_path.unwrap_or_default(); let source = with_thread_spawn_agent_metadata( thread.source, thread.agent_nickname.clone(), @@ -3709,7 +3701,7 @@ fn summary_from_stored_thread( branch: git.branch, origin_url: git.repository_url, }); - Some(ConversationSummary { + ConversationSummary { conversation_id: thread.thread_id, path, preview: thread.first_user_message.unwrap_or(thread.preview), @@ -3734,7 +3726,7 @@ fn summary_from_stored_thread( cli_version: thread.cli_version, source, git_info, - }) + } } #[allow(clippy::too_many_arguments)] diff --git a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs index 5a18c09db..5642dbbe8 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs @@ -409,8 +409,7 @@ mod thread_processor_behavior_tests { history: None, }; - let summary = - summary_from_stored_thread(stored_thread, "fallback").expect("summary should exist"); + let summary = summary_from_stored_thread(stored_thread, "fallback"); assert_eq!( summary.timestamp.as_deref(), diff --git a/codex-rs/app-server/src/request_processors/thread_summary.rs b/codex-rs/app-server/src/request_processors/thread_summary.rs index f2de590a9..875bd3dea 100644 --- a/codex-rs/app-server/src/request_processors/thread_summary.rs +++ b/codex-rs/app-server/src/request_processors/thread_summary.rs @@ -1,5 +1,10 @@ use super::*; +#[cfg(test)] +use chrono::DateTime; +#[cfg(test)] +use chrono::Utc; + #[cfg(test)] pub(crate) async fn read_summary_from_rollout( path: &Path, @@ -203,6 +208,7 @@ pub(super) fn thread_response_sandbox_policy( sandbox_policy.into() } +#[cfg(test)] fn parse_datetime(timestamp: Option<&str>) -> Option> { timestamp.and_then(|ts| { chrono::DateTime::parse_from_rfc3339(ts) @@ -229,6 +235,7 @@ pub(super) fn thread_started_notification(mut thread: Thread) -> ThreadStartedNo ThreadStartedNotification { thread } } +#[cfg(test)] pub(crate) fn summary_to_thread( summary: ConversationSummary, fallback_cwd: &AbsolutePathBuf, @@ -257,6 +264,7 @@ pub(crate) fn summary_to_thread( AbsolutePathBuf::relative_to_current_dir(path_utils::normalize_for_native_workdir(cwd)) .unwrap_or_else(|err| { warn!( + conversation_id = %conversation_id, path = %path.display(), "failed to normalize thread cwd while summarizing thread: {err}" ); @@ -274,7 +282,7 @@ pub(crate) fn summary_to_thread( created_at: created_at.map(|dt| dt.timestamp()).unwrap_or(0), updated_at: updated_at.map(|dt| dt.timestamp()).unwrap_or(0), status: ThreadStatus::NotLoaded, - path: Some(path), + path: (!path.as_os_str().is_empty()).then_some(path), cwd, cli_version, agent_nickname: source.get_nickname(), diff --git a/codex-rs/app-server/tests/suite/conversation_summary.rs b/codex-rs/app-server/tests/suite/conversation_summary.rs index bb938d9ae..c06ff8599 100644 --- a/codex-rs/app-server/tests/suite/conversation_summary.rs +++ b/codex-rs/app-server/tests/suite/conversation_summary.rs @@ -3,20 +3,41 @@ use app_test_support::McpProcess; use app_test_support::create_fake_rollout; use app_test_support::rollout_path; use app_test_support::to_response; +use codex_app_server::in_process; +use codex_app_server::in_process::InProcessStartArgs; +use codex_app_server_protocol::ClientInfo; +use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ConversationSummary; use codex_app_server_protocol::GetConversationSummaryParams; use codex_app_server_protocol::GetConversationSummaryResponse; +use codex_app_server_protocol::InitializeCapabilities; +use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::RequestId; +use codex_arg0::Arg0DispatchPaths; +use codex_config::CloudRequirementsLoader; +use codex_config::LoaderOverrides; +use codex_core::config::ConfigBuilder; +use codex_exec_server::EnvironmentManager; +use codex_feedback::CodexFeedback; use codex_protocol::ThreadId; +use codex_protocol::models::BaseInstructions; use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::ThreadMemoryMode; +use codex_thread_store::CreateThreadParams; +use codex_thread_store::InMemoryThreadStore; +use codex_thread_store::ThreadEventPersistenceMode; +use codex_thread_store::ThreadPersistenceMetadata; +use codex_thread_store::ThreadStore; use codex_utils_absolute_path::AbsolutePathBuf; use pretty_assertions::assert_eq; use std::path::Path; use std::path::PathBuf; +use std::sync::Arc; use tempfile::TempDir; use tokio::time::timeout; +use uuid::Uuid; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); const FILENAME_TS: &str = "2025-01-02T12-00-00"; @@ -47,7 +68,9 @@ fn normalized_canonical_path(path: impl AsRef) -> Result { } fn normalized_summary_path(mut summary: ConversationSummary) -> Result { - summary.path = normalized_canonical_path(&summary.path)?; + if !summary.path.as_os_str().is_empty() { + summary.path = normalized_canonical_path(summary.path)?; + } Ok(summary) } @@ -122,6 +145,87 @@ async fn get_conversation_summary_by_rollout_path_rejects_remote_thread_store() Ok(()) } +#[tokio::test] +async fn get_conversation_summary_by_thread_id_reads_pathless_store_thread() -> Result<()> { + let codex_home = TempDir::new()?; + let store_id = Uuid::new_v4().to_string(); + create_config_toml_with_in_memory_thread_store(codex_home.path(), &store_id)?; + let store = InMemoryThreadStore::for_id(store_id.clone()); + let _in_memory_store = InMemoryThreadStoreId { store_id }; + let thread_id = ThreadId::from_string("00000000-0000-4000-8000-000000000125")?; + store + .create_thread(CreateThreadParams { + thread_id, + forked_from_id: None, + source: SessionSource::Cli, + thread_source: None, + base_instructions: BaseInstructions::default(), + dynamic_tools: Vec::new(), + metadata: ThreadPersistenceMetadata { + cwd: None, + model_provider: "test-provider".to_string(), + memory_mode: ThreadMemoryMode::Disabled, + }, + event_persistence_mode: ThreadEventPersistenceMode::default(), + }) + .await?; + + let loader_overrides = LoaderOverrides::without_managed_config_for_tests(); + let config = ConfigBuilder::default() + .codex_home(codex_home.path().to_path_buf()) + .fallback_cwd(Some(codex_home.path().to_path_buf())) + .loader_overrides(loader_overrides.clone()) + .build() + .await?; + let client = in_process::start(InProcessStartArgs { + arg0_paths: Arg0DispatchPaths::default(), + config: Arc::new(config), + cli_overrides: Vec::new(), + loader_overrides, + cloud_requirements: CloudRequirementsLoader::default(), + thread_config_loader: Arc::new(codex_config::NoopThreadConfigLoader), + feedback: CodexFeedback::new(), + log_db: None, + state_db: None, + environment_manager: Arc::new(EnvironmentManager::default_for_tests()), + config_warnings: Vec::new(), + session_source: SessionSource::Cli, + enable_codex_api_key_env: false, + initialize: InitializeParams { + client_info: ClientInfo { + name: "codex-app-server-tests".to_string(), + title: None, + version: "0.1.0".to_string(), + }, + capabilities: Some(InitializeCapabilities { + experimental_api: true, + ..Default::default() + }), + }, + channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY, + }) + .await?; + + let result = client + .request(ClientRequest::GetConversationSummary { + request_id: RequestId::Integer(1), + params: GetConversationSummaryParams::ThreadId { + conversation_id: thread_id, + }, + }) + .await? + .expect("getConversationSummary should succeed"); + let GetConversationSummaryResponse { summary } = serde_json::from_value(result)?; + + assert_eq!(summary.conversation_id, thread_id); + assert_eq!(summary.path, PathBuf::new()); + assert_eq!(summary.cwd, PathBuf::new()); + assert_eq!(summary.model_provider, "test"); + + client.shutdown().await?; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn get_conversation_summary_by_relative_rollout_path_resolves_from_codex_home() -> Result<()> { @@ -157,3 +261,39 @@ async fn get_conversation_summary_by_relative_rollout_path_resolves_from_codex_h assert_eq!(normalized_summary_path(received.summary)?, expected); Ok(()) } + +struct InMemoryThreadStoreId { + store_id: String, +} + +impl Drop for InMemoryThreadStoreId { + fn drop(&mut self) { + InMemoryThreadStore::remove_id(&self.store_id); + } +} + +fn create_config_toml_with_in_memory_thread_store( + codex_home: &Path, + store_id: &str, +) -> std::io::Result<()> { + std::fs::write( + codex_home.join("config.toml"), + format!( + r#" +model = "mock-model" +approval_policy = "never" +sandbox_mode = "read-only" +experimental_thread_store = {{ type = "in_memory", id = "{store_id}" }} + +model_provider = "mock_provider" + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "http://127.0.0.1:1/v1" +wire_api = "responses" +request_max_retries = 0 +stream_max_retries = 0 +"# + ), + ) +} diff --git a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs index 588764edb..5b421dcec 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs @@ -2,6 +2,12 @@ use anyhow::Result; use app_test_support::McpProcess; use app_test_support::create_mock_responses_server_repeating_assistant; use app_test_support::to_response; +use codex_app_server::in_process; +use codex_app_server::in_process::InProcessStartArgs; +use codex_app_server_protocol::ClientInfo; +use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::InitializeCapabilities; +use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ThreadArchiveParams; @@ -15,17 +21,36 @@ use codex_app_server_protocol::ThreadUnarchivedNotification; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::UserInput; +use codex_arg0::Arg0DispatchPaths; +use codex_config::CloudRequirementsLoader; +use codex_config::LoaderOverrides; +use codex_core::config::ConfigBuilder; use codex_core::find_archived_thread_path_by_id_str; use codex_core::find_thread_path_by_id_str; +use codex_exec_server::EnvironmentManager; +use codex_feedback::CodexFeedback; +use codex_protocol::ThreadId; +use codex_protocol::models::BaseInstructions; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::ThreadMemoryMode; +use codex_thread_store::CreateThreadParams; +use codex_thread_store::InMemoryThreadStore; +use codex_thread_store::ThreadEventPersistenceMode; +use codex_thread_store::ThreadMetadataPatch; +use codex_thread_store::ThreadPersistenceMetadata; +use codex_thread_store::ThreadStore; +use codex_thread_store::UpdateThreadMetadataParams; use pretty_assertions::assert_eq; use serde_json::Value; use std::fs::FileTimes; use std::fs::OpenOptions; use std::path::Path; +use std::sync::Arc; use std::time::Duration; use std::time::SystemTime; use tempfile::TempDir; use tokio::time::timeout; +use uuid::Uuid; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); @@ -172,11 +197,139 @@ async fn thread_unarchive_moves_rollout_back_into_sessions_directory() -> Result Ok(()) } +#[tokio::test] +async fn thread_unarchive_preserves_pathless_store_metadata() -> Result<()> { + let codex_home = TempDir::new()?; + let store_id = Uuid::new_v4().to_string(); + create_config_toml_with_in_memory_thread_store(codex_home.path(), &store_id)?; + let store = InMemoryThreadStore::for_id(store_id.clone()); + let _in_memory_store = InMemoryThreadStoreId { store_id }; + let thread_id = ThreadId::from_string("00000000-0000-4000-8000-000000000126")?; + let parent_thread_id = ThreadId::from_string("00000000-0000-4000-8000-000000000127")?; + store + .create_thread(CreateThreadParams { + thread_id, + forked_from_id: Some(parent_thread_id), + source: SessionSource::Cli, + thread_source: None, + base_instructions: BaseInstructions::default(), + dynamic_tools: Vec::new(), + metadata: ThreadPersistenceMetadata { + cwd: None, + model_provider: "test-provider".to_string(), + memory_mode: ThreadMemoryMode::Disabled, + }, + event_persistence_mode: ThreadEventPersistenceMode::default(), + }) + .await?; + store + .update_thread_metadata(UpdateThreadMetadataParams { + thread_id, + patch: ThreadMetadataPatch { + name: Some("named pathless thread".to_string()), + ..Default::default() + }, + include_archived: true, + }) + .await?; + + let loader_overrides = LoaderOverrides::without_managed_config_for_tests(); + let config = ConfigBuilder::default() + .codex_home(codex_home.path().to_path_buf()) + .fallback_cwd(Some(codex_home.path().to_path_buf())) + .loader_overrides(loader_overrides.clone()) + .build() + .await?; + let client = in_process::start(InProcessStartArgs { + arg0_paths: Arg0DispatchPaths::default(), + config: Arc::new(config), + cli_overrides: Vec::new(), + loader_overrides, + cloud_requirements: CloudRequirementsLoader::default(), + thread_config_loader: Arc::new(codex_config::NoopThreadConfigLoader), + feedback: CodexFeedback::new(), + log_db: None, + state_db: None, + environment_manager: Arc::new(EnvironmentManager::default_for_tests()), + config_warnings: Vec::new(), + session_source: SessionSource::Cli, + enable_codex_api_key_env: false, + initialize: InitializeParams { + client_info: ClientInfo { + name: "codex-app-server-tests".to_string(), + title: None, + version: "0.1.0".to_string(), + }, + capabilities: Some(InitializeCapabilities { + experimental_api: true, + ..Default::default() + }), + }, + channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY, + }) + .await?; + + let result = client + .request(ClientRequest::ThreadUnarchive { + request_id: RequestId::Integer(1), + params: ThreadUnarchiveParams { + thread_id: thread_id.to_string(), + }, + }) + .await? + .expect("thread/unarchive should succeed"); + let ThreadUnarchiveResponse { thread } = serde_json::from_value(result)?; + + assert_eq!(thread.id, thread_id.to_string()); + assert_eq!(thread.path, None); + assert_eq!(thread.forked_from_id, Some(parent_thread_id.to_string())); + assert_eq!(thread.name, Some("named pathless thread".to_string())); + + client.shutdown().await?; + Ok(()) +} + fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> { let config_toml = codex_home.join("config.toml"); std::fs::write(config_toml, config_contents(server_uri)) } +struct InMemoryThreadStoreId { + store_id: String, +} + +impl Drop for InMemoryThreadStoreId { + fn drop(&mut self) { + InMemoryThreadStore::remove_id(&self.store_id); + } +} + +fn create_config_toml_with_in_memory_thread_store( + codex_home: &Path, + store_id: &str, +) -> std::io::Result<()> { + std::fs::write( + codex_home.join("config.toml"), + format!( + r#" +model = "mock-model" +approval_policy = "never" +sandbox_mode = "read-only" +experimental_thread_store = {{ type = "in_memory", id = "{store_id}" }} + +model_provider = "mock_provider" + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "http://127.0.0.1:1/v1" +wire_api = "responses" +request_max_retries = 0 +stream_max_retries = 0 +"# + ), + ) +} + fn config_contents(server_uri: &str) -> String { format!( r#"model = "mock-model" diff --git a/codex-rs/docs/codex_mcp_interface.md b/codex-rs/docs/codex_mcp_interface.md index 7e3d4c684..bf2def240 100644 --- a/codex-rs/docs/codex_mcp_interface.md +++ b/codex-rs/docs/codex_mcp_interface.md @@ -52,7 +52,7 @@ Use the separate `codex mcp` subcommand to manage configured MCP server launcher Use the v2 thread and turn APIs for all new integrations. `thread/start` creates a thread, `turn/start` submits user input, `turn/interrupt` stops an in-flight turn, and `thread/list` / `thread/read` expose persisted history. -`getConversationSummary` remains as a compatibility helper for clients that still need a summary lookup by `conversationId` or `rolloutPath`. +`getConversationSummary` remains as a compatibility helper for clients that still need a summary lookup by `conversationId` or `rolloutPath`. Lookups by `conversationId` are preferred; lookups by `rolloutPath` won't work with non-local thread stores. For complete request and response shapes, see the app-server README and the protocol definitions in `app-server-protocol/src/protocol/v2.rs`.