From 27251b597ddf8e9806d2822aa532dd50b9b3f2fe Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Sun, 2 Nov 2025 10:03:21 -0800 Subject: [PATCH] thread/archive --- .../app-server/src/codex_message_processor.rs | 176 +++++++++++++++++- .../app-server/tests/common/mcp_process.rs | 10 + .../tests/suite/v2/thread_archive.rs | 96 ++++++++++ 3 files changed, 276 insertions(+), 6 deletions(-) create mode 100644 codex-rs/app-server/tests/suite/v2/thread_archive.rs diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 786359447c..c24570c049 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -60,6 +60,8 @@ use codex_app_server_protocol::SetDefaultModelParams; use codex_app_server_protocol::SetDefaultModelResponse; use codex_app_server_protocol::SortOrder; use codex_app_server_protocol::Thread; +use codex_app_server_protocol::ThreadArchiveParams; +use codex_app_server_protocol::ThreadArchiveResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStartedNotification; @@ -193,12 +195,8 @@ impl CodexMessageProcessor { self.send_unimplemented_error(request_id, "thread/resume") .await; } - ClientRequest::ThreadArchive { - request_id, - params: _, - } => { - self.send_unimplemented_error(request_id, "thread/archive") - .await; + ClientRequest::ThreadArchive { request_id, params } => { + self.thread_archive(request_id, params).await; } ClientRequest::ThreadList { request_id, @@ -947,6 +945,172 @@ impl CodexMessageProcessor { } } + async fn thread_archive(&self, request_id: RequestId, params: ThreadArchiveParams) { + // Resolve conversation id from v2 thread id string. + let conversation_id = match ConversationId::from_string(¶ms.thread.id) { + Ok(id) => id, + Err(err) => { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("invalid thread id: {err}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + // Locate rollout path from conversation id. + let rollout_path = match find_conversation_path_by_id_str( + &self.config.codex_home, + &conversation_id.to_string(), + ) + .await + { + Ok(Some(p)) => p, + Ok(None) => { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("no rollout found for conversation id {conversation_id}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + Err(err) => { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!( + "failed to locate conversation id {conversation_id}: {err}" + ), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + // The remaining logic mirrors archive_conversation(), but returns + // ThreadArchiveResponse for v2. + + let rollout_folder = self.config.codex_home.join(codex_core::SESSIONS_SUBDIR); + let canonical_rollout_path = tokio::fs::canonicalize(&rollout_path).await; + let canonical_rollout_path = if let Ok(path) = canonical_rollout_path + && path.starts_with(&rollout_folder) + { + path + } else { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!( + "rollout path `{}` must be in sessions directory", + rollout_path.display() + ), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + }; + + let required_suffix = format!("{conversation_id}.jsonl"); + let Some(file_name) = canonical_rollout_path.file_name().map(OsStr::to_owned) else { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!( + "rollout path `{}` missing file name", + rollout_path.display() + ), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + }; + + if !file_name + .to_string_lossy() + .ends_with(required_suffix.as_str()) + { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!( + "rollout path `{}` does not match conversation id {conversation_id}", + rollout_path.display() + ), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + + let removed_conversation = self + .conversation_manager + .remove_conversation(&conversation_id) + .await; + if let Some(conversation) = removed_conversation { + info!("conversation {conversation_id} was active; shutting down"); + let conversation_clone = conversation.clone(); + let notify = Arc::new(tokio::sync::Notify::new()); + let notify_clone = notify.clone(); + + let is_shutdown = tokio::spawn(async move { + loop { + select! { + _ = notify_clone.notified() => { + break; + } + event = conversation_clone.next_event() => { + if let Ok(event) = event && matches!(event.msg, EventMsg::ShutdownComplete) { + break; + } + } + } + } + }); + + match conversation.submit(Op::Shutdown).await { + Ok(_) => { + select! { + _ = is_shutdown => {} + _ = tokio::time::sleep(Duration::from_secs(10)) => { + warn!("conversation {conversation_id} shutdown timed out; proceeding with archive"); + notify.notify_one(); + } + } + } + Err(err) => { + error!("failed to submit Shutdown to conversation {conversation_id}: {err}"); + notify.notify_one(); + } + } + } + + let result: std::io::Result<()> = async { + let archive_folder = self + .config + .codex_home + .join(codex_core::ARCHIVED_SESSIONS_SUBDIR); + tokio::fs::create_dir_all(&archive_folder).await?; + tokio::fs::rename(&canonical_rollout_path, &archive_folder.join(&file_name)).await?; + Ok(()) + } + .await; + + match result { + Ok(()) => { + let response = ThreadArchiveResponse {}; + self.outgoing.send_response(request_id, response).await; + } + Err(err) => { + let error = JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + message: format!("failed to archive conversation: {err}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + } + } + } + async fn get_conversation_summary( &self, request_id: RequestId, diff --git a/codex-rs/app-server/tests/common/mcp_process.rs b/codex-rs/app-server/tests/common/mcp_process.rs index 67c70ad07d..8806dc1809 100644 --- a/codex-rs/app-server/tests/common/mcp_process.rs +++ b/codex-rs/app-server/tests/common/mcp_process.rs @@ -31,6 +31,7 @@ use codex_app_server_protocol::SendUserMessageParams; use codex_app_server_protocol::SendUserTurnParams; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::SetDefaultModelParams; +use codex_app_server_protocol::ThreadArchiveParams; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::JSONRPCError; @@ -317,6 +318,15 @@ impl McpProcess { self.send_request("loginChatGpt", None).await } + /// Send a `thread/archive` JSON-RPC request (v2). + pub async fn send_thread_archive_request( + &mut self, + params: ThreadArchiveParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("thread/archive", params).await + } + /// Send a `cancelLoginChatGpt` JSON-RPC request. pub async fn send_cancel_login_chat_gpt_request( &mut self, diff --git a/codex-rs/app-server/tests/suite/v2/thread_archive.rs b/codex-rs/app-server/tests/suite/v2/thread_archive.rs new file mode 100644 index 0000000000..8fad8f47fa --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/thread_archive.rs @@ -0,0 +1,96 @@ +use anyhow::Result; +use app_test_support::McpProcess; +use app_test_support::to_response; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadArchiveParams; +use codex_app_server_protocol::ThreadArchiveResponse; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_core::ARCHIVED_SESSIONS_SUBDIR; +use codex_core::SESSIONS_SUBDIR; +use std::path::Path; +use tempfile::TempDir; +use tokio::time::timeout; + +const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn thread_archive_moves_rollout_into_archived_directory() -> Result<()> { + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + // Start a thread. + let start_id = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + model_provider: None, + profile: None, + cwd: None, + approval_policy: None, + sandbox: None, + config: None, + base_instructions: None, + developer_instructions: None, + compact_prompt: None, + include_apply_patch_tool: None, + }) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .await??; + let ThreadStartResponse { thread } = to_response::(start_resp)?; + assert!(!thread.id.is_empty()); + + // Construct the rollout path derived from the id we got back. + let rollout_path = codex_home + .path() + .join(SESSIONS_SUBDIR) + .join(format!("{}.jsonl", thread.id)); + assert!(rollout_path.exists(), "expected {} to exist", rollout_path.display()); + + // Archive the thread. + let archive_id = mcp + .send_thread_archive_request(ThreadArchiveParams { thread: thread.clone() }) + .await?; + let archive_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(archive_id)), + ) + .await??; + let _: ThreadArchiveResponse = to_response::(archive_resp)?; + + // Verify file moved. + let archived_directory = codex_home.path().join(ARCHIVED_SESSIONS_SUBDIR); + let archived_rollout_path = archived_directory.join(format!("{}.jsonl", thread.id)); + assert!( + !rollout_path.exists(), + "expected rollout path {} to be moved", + rollout_path.display() + ); + assert!( + archived_rollout_path.exists(), + "expected archived rollout path {} to exist", + archived_rollout_path.display() + ); + + Ok(()) +} + +fn create_config_toml(codex_home: &Path) -> std::io::Result<()> { + let config_toml = codex_home.join("config.toml"); + std::fs::write(config_toml, config_contents()) +} + +fn config_contents() -> &'static str { + r#"model = "mock-model" +approval_policy = "never" +sandbox_mode = "read-only" +"# +} +