From 2a76a08a9e048a8f2798e36b8cb2160ad59cbc2e Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Tue, 9 Sep 2025 00:11:48 -0700 Subject: [PATCH 1/2] fix: include rollout_path in NewConversationResponse (#3352) Adding the `rollout_path` to the `NewConversationResponse` makes it so a client can perform subsequent operations on a `(ConversationId, PathBuf)` pair. #3353 will introduce support for `ArchiveConversation`. --- [//]: # (BEGIN SAPLING FOOTER) Stack created with [Sapling](https://sapling-scm.com). Best reviewed with [ReviewStack](https://reviewstack.dev/openai/codex/pull/3352). * #3353 * __->__ #3352 --- codex-rs/Cargo.lock | 1 + codex-rs/core/src/codex.rs | 2 ++ codex-rs/core/src/rollout/recorder.rs | 14 ++++++++--- .../src/event_processor_with_human_output.rs | 1 + .../mcp-server/src/codex_message_processor.rs | 1 + codex-rs/mcp-server/src/outgoing_message.rs | 6 +++++ .../suite/codex_message_processor_flow.rs | 1 + .../tests/suite/create_conversation.rs | 1 + codex-rs/protocol/Cargo.toml | 1 + codex-rs/protocol/src/mcp_protocol.rs | 1 + codex-rs/protocol/src/protocol.rs | 24 +++++++++++++++---- codex-rs/tui/src/chatwidget/tests.rs | 4 +++- codex-rs/tui/src/history_cell.rs | 1 + .../tui/tests/fixtures/binary-size-log.jsonl | 4 ++-- 14 files changed, 51 insertions(+), 11 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index e76ea6bb89..cda2e81d17 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -830,6 +830,7 @@ dependencies = [ "strum 0.27.2", "strum_macros 0.27.2", "sys-locale", + "tempfile", "tracing", "ts-rs", "uuid", diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 42960b27aa..cc8768d79b 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -416,6 +416,7 @@ impl Session { error!("failed to initialize rollout recorder: {e:#}"); anyhow::anyhow!("failed to initialize rollout recorder: {e:#}") })?; + let rollout_path = rollout_recorder.rollout_path.clone(); // Create the mutable state for the Session. let state = State { history: ConversationHistory::new(), @@ -509,6 +510,7 @@ impl Session { history_log_id, history_entry_count, initial_messages, + rollout_path, }), }) .chain(post_session_configured_error_events.into_iter()); diff --git a/codex-rs/core/src/rollout/recorder.rs b/codex-rs/core/src/rollout/recorder.rs index 907583353c..0e0af8990b 100644 --- a/codex-rs/core/src/rollout/recorder.rs +++ b/codex-rs/core/src/rollout/recorder.rs @@ -72,6 +72,7 @@ pub struct SavedSession { #[derive(Clone)] pub struct RolloutRecorder { tx: Sender, + pub(crate) rollout_path: PathBuf, } #[derive(Clone)] @@ -119,13 +120,14 @@ impl RolloutRecorder { /// cannot be created or the rollout file cannot be opened we return the /// error so the caller can decide whether to disable persistence. pub async fn new(config: &Config, params: RolloutRecorderParams) -> std::io::Result { - let (file, meta) = match params { + let (file, rollout_path, meta) = match params { RolloutRecorderParams::Create { conversation_id, instructions, } => { let LogFileInfo { file, + path, conversation_id: session_id, timestamp, } = create_log_file(config, conversation_id)?; @@ -140,6 +142,7 @@ impl RolloutRecorder { ( tokio::fs::File::from_std(file), + path, Some(SessionMeta { timestamp, id: session_id, @@ -150,8 +153,9 @@ impl RolloutRecorder { RolloutRecorderParams::Resume { path } => ( tokio::fs::OpenOptions::new() .append(true) - .open(path) + .open(&path) .await?, + path, None, ), }; @@ -169,7 +173,7 @@ impl RolloutRecorder { // driver instead of blocking the runtime. tokio::task::spawn(rollout_writer(file, rx, meta, cwd)); - Ok(Self { tx }) + Ok(Self { tx, rollout_path }) } pub(crate) async fn record_items(&self, items: &[ResponseItem]) -> std::io::Result<()> { @@ -289,6 +293,9 @@ struct LogFileInfo { /// Opened file handle to the rollout file. file: File, + /// Full path to the rollout file. + path: PathBuf, + /// Session ID (also embedded in filename). conversation_id: ConversationId, @@ -328,6 +335,7 @@ fn create_log_file( Ok(LogFileInfo { file, + path, conversation_id, timestamp, }) diff --git a/codex-rs/exec/src/event_processor_with_human_output.rs b/codex-rs/exec/src/event_processor_with_human_output.rs index 7c5e6b147c..ae4708976f 100644 --- a/codex-rs/exec/src/event_processor_with_human_output.rs +++ b/codex-rs/exec/src/event_processor_with_human_output.rs @@ -523,6 +523,7 @@ impl EventProcessor for EventProcessorWithHumanOutput { history_log_id: _, history_entry_count: _, initial_messages: _, + rollout_path: _, } = session_configured_event; ts_println!( diff --git a/codex-rs/mcp-server/src/codex_message_processor.rs b/codex-rs/mcp-server/src/codex_message_processor.rs index e5f44b5687..2169a8f27f 100644 --- a/codex-rs/mcp-server/src/codex_message_processor.rs +++ b/codex-rs/mcp-server/src/codex_message_processor.rs @@ -528,6 +528,7 @@ impl CodexMessageProcessor { let response = NewConversationResponse { conversation_id, model: session_configured.model, + rollout_path: session_configured.rollout_path, }; self.outgoing.send_response(request_id, response).await; } diff --git a/codex-rs/mcp-server/src/outgoing_message.rs b/codex-rs/mcp-server/src/outgoing_message.rs index 537d29db62..5ce2e99423 100644 --- a/codex-rs/mcp-server/src/outgoing_message.rs +++ b/codex-rs/mcp-server/src/outgoing_message.rs @@ -262,6 +262,7 @@ mod tests { use codex_protocol::mcp_protocol::LoginChatGptCompleteNotification; use pretty_assertions::assert_eq; use serde_json::json; + use tempfile::NamedTempFile; use uuid::Uuid; use super::*; @@ -272,6 +273,7 @@ mod tests { let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx); let conversation_id = ConversationId::new(); + let rollout_file = NamedTempFile::new().unwrap(); let event = Event { id: "1".to_string(), msg: EventMsg::SessionConfigured(SessionConfiguredEvent { @@ -280,6 +282,7 @@ mod tests { history_log_id: 1, history_entry_count: 1000, initial_messages: None, + rollout_path: rollout_file.path().to_path_buf(), }), }; @@ -305,12 +308,14 @@ mod tests { let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx); let conversation_id = ConversationId::new(); + let rollout_file = NamedTempFile::new().unwrap(); let session_configured_event = SessionConfiguredEvent { session_id: conversation_id, model: "gpt-4o".to_string(), history_log_id: 1, history_entry_count: 1000, initial_messages: None, + rollout_path: rollout_file.path().to_path_buf(), }; let event = Event { id: "1".to_string(), @@ -340,6 +345,7 @@ mod tests { "history_log_id": session_configured_event.history_log_id, "history_entry_count": session_configured_event.history_entry_count, "type": "session_configured", + "rollout_path": rollout_file.path().to_path_buf(), } }); assert_eq!(params.unwrap(), expected_params); diff --git a/codex-rs/mcp-server/tests/suite/codex_message_processor_flow.rs b/codex-rs/mcp-server/tests/suite/codex_message_processor_flow.rs index 092b135291..7737c49af3 100644 --- a/codex-rs/mcp-server/tests/suite/codex_message_processor_flow.rs +++ b/codex-rs/mcp-server/tests/suite/codex_message_processor_flow.rs @@ -90,6 +90,7 @@ async fn test_codex_jsonrpc_conversation_flow() { let NewConversationResponse { conversation_id, model, + rollout_path: _, } = new_conv_resp; assert_eq!(model, "mock-model"); diff --git a/codex-rs/mcp-server/tests/suite/create_conversation.rs b/codex-rs/mcp-server/tests/suite/create_conversation.rs index 81cc1f3ee7..5071868c85 100644 --- a/codex-rs/mcp-server/tests/suite/create_conversation.rs +++ b/codex-rs/mcp-server/tests/suite/create_conversation.rs @@ -59,6 +59,7 @@ async fn test_conversation_create_and_send_message_ok() { let NewConversationResponse { conversation_id, model, + rollout_path: _, } = to_response::(new_conv_resp) .expect("deserialize newConversation response"); assert_eq!(model, "o3"); diff --git a/codex-rs/protocol/Cargo.toml b/codex-rs/protocol/Cargo.toml index d77da7fc19..f88297a07b 100644 --- a/codex-rs/protocol/Cargo.toml +++ b/codex-rs/protocol/Cargo.toml @@ -28,6 +28,7 @@ uuid = { version = "1", features = ["serde", "v4"] } [dev-dependencies] pretty_assertions = "1.4.1" +tempfile = "3" [package.metadata.cargo-shear] # Required because the not imported as strum_macros in non-nightly builds. diff --git a/codex-rs/protocol/src/mcp_protocol.rs b/codex-rs/protocol/src/mcp_protocol.rs index 2af1a95166..70e334036f 100644 --- a/codex-rs/protocol/src/mcp_protocol.rs +++ b/codex-rs/protocol/src/mcp_protocol.rs @@ -203,6 +203,7 @@ pub struct NewConversationParams { pub struct NewConversationResponse { pub conversation_id: ConversationId, pub model: String, + pub rollout_path: PathBuf, } #[derive(Serialize, Deserialize, Debug, Clone, TS)] diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index eeaa72cd93..2f1c46364a 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -958,6 +958,8 @@ pub struct SessionConfiguredEvent { /// When present, UIs can use these to seed the history. #[serde(skip_serializing_if = "Option::is_none")] pub initial_messages: Option>, + + pub rollout_path: PathBuf, } /// User's decision in response to an ExecApprovalRequest. @@ -1020,12 +1022,15 @@ pub enum TurnAbortReason { #[cfg(test)] mod tests { use super::*; + use serde_json::json; + use tempfile::NamedTempFile; /// Serialize Event to verify that its JSON representation has the expected /// amount of nesting. #[test] fn serialize_event() { let conversation_id = ConversationId(uuid::uuid!("67e55044-10b1-426f-9247-bb680e5fe0c8")); + let rollout_file = NamedTempFile::new().unwrap(); let event = Event { id: "1234".to_string(), msg: EventMsg::SessionConfigured(SessionConfiguredEvent { @@ -1034,13 +1039,22 @@ mod tests { history_log_id: 0, history_entry_count: 0, initial_messages: None, + rollout_path: rollout_file.path().to_path_buf(), }), }; - let serialized = serde_json::to_string(&event).unwrap(); - assert_eq!( - serialized, - r#"{"id":"1234","msg":{"type":"session_configured","session_id":"67e55044-10b1-426f-9247-bb680e5fe0c8","model":"codex-mini-latest","history_log_id":0,"history_entry_count":0}}"# - ); + + let expected = json!({ + "id": "1234", + "msg": { + "type": "session_configured", + "session_id": "67e55044-10b1-426f-9247-bb680e5fe0c8", + "model": "codex-mini-latest", + "history_log_id": 0, + "history_entry_count": 0, + "rollout_path": format!("{}", rollout_file.path().display()), + } + }); + assert_eq!(expected, serde_json::to_value(&event).unwrap()); } #[test] diff --git a/codex-rs/tui/src/chatwidget/tests.rs b/codex-rs/tui/src/chatwidget/tests.rs index 3b61203399..113864dba7 100644 --- a/codex-rs/tui/src/chatwidget/tests.rs +++ b/codex-rs/tui/src/chatwidget/tests.rs @@ -35,6 +35,7 @@ use std::fs::File; use std::io::BufRead; use std::io::BufReader; use std::path::PathBuf; +use tempfile::NamedTempFile; use tokio::sync::mpsc::unbounded_channel; fn test_config() -> Config { @@ -133,7 +134,7 @@ fn resumed_initial_messages_render_history() { let (mut chat, mut rx, _ops) = make_chatwidget_manual(); let conversation_id = ConversationId::new(); - + let rollout_file = NamedTempFile::new().unwrap(); let configured = codex_core::protocol::SessionConfiguredEvent { session_id: conversation_id, model: "test-model".to_string(), @@ -148,6 +149,7 @@ fn resumed_initial_messages_render_history() { message: "assistant reply".to_string(), }), ]), + rollout_path: rollout_file.path().to_path_buf(), }; chat.handle_codex_event(Event { diff --git a/codex-rs/tui/src/history_cell.rs b/codex-rs/tui/src/history_cell.rs index 92bb076225..a7be1d6560 100644 --- a/codex-rs/tui/src/history_cell.rs +++ b/codex-rs/tui/src/history_cell.rs @@ -605,6 +605,7 @@ pub(crate) fn new_session_info( history_log_id: _, history_entry_count: _, initial_messages: _, + rollout_path: _, } = event; if is_first_event { let cwd_str = match relativize_to_home(&config.cwd) { diff --git a/codex-rs/tui/tests/fixtures/binary-size-log.jsonl b/codex-rs/tui/tests/fixtures/binary-size-log.jsonl index a02dd02f9a..98cac7c0ae 100644 --- a/codex-rs/tui/tests/fixtures/binary-size-log.jsonl +++ b/codex-rs/tui/tests/fixtures/binary-size-log.jsonl @@ -34,7 +34,7 @@ {"ts":"2025-08-09T15:51:04.829Z","dir":"to_tui","kind":"app_event","variant":"RequestRedraw"} {"ts":"2025-08-09T15:51:04.829Z","dir":"to_tui","kind":"log_line","line":"[INFO codex_core::codex] resume_path: None"} {"ts":"2025-08-09T15:51:04.830Z","dir":"to_tui","kind":"app_event","variant":"Redraw"} -{"ts":"2025-08-09T15:51:04.856Z","dir":"to_tui","kind":"codex_event","payload":{"id":"0","msg":{"type":"session_configured","session_id":"d126e3d0-80ed-480a-be8c-09d97ff602cf","model":"gpt-5","history_log_id":2532619,"history_entry_count":339}}} +{"ts":"2025-08-09T15:51:04.856Z","dir":"to_tui","kind":"codex_event","payload":{"id":"0","msg":{"type":"session_configured","session_id":"d126e3d0-80ed-480a-be8c-09d97ff602cf","model":"gpt-5","history_log_id":2532619,"history_entry_count":339,"rollout_path":"/tmp/codex-test-rollout.jsonl"}}} {"ts":"2025-08-09T15:51:04.856Z","dir":"to_tui","kind":"insert_history","lines":9} {"ts":"2025-08-09T15:51:04.857Z","dir":"to_tui","kind":"app_event","variant":"RequestRedraw"} {"ts":"2025-08-09T15:51:04.857Z","dir":"to_tui","kind":"app_event","variant":"RequestRedraw"} @@ -16447,7 +16447,7 @@ {"ts":"2025-08-09T16:06:58.083Z","dir":"to_tui","kind":"app_event","variant":"RequestRedraw"} {"ts":"2025-08-09T16:06:58.085Z","dir":"to_tui","kind":"app_event","variant":"Redraw"} {"ts":"2025-08-09T16:06:58.085Z","dir":"to_tui","kind":"log_line","line":"[INFO codex_core::codex] resume_path: None"} -{"ts":"2025-08-09T16:06:58.136Z","dir":"to_tui","kind":"codex_event","payload":{"id":"0","msg":{"type":"session_configured","session_id":"c7df96da-daec-4fe9-aed9-3cd19b7a6192","model":"gpt-5","history_log_id":2532619,"history_entry_count":342}}} +{"ts":"2025-08-09T16:06:58.136Z","dir":"to_tui","kind":"codex_event","payload":{"id":"0","msg":{"type":"session_configured","session_id":"c7df96da-daec-4fe9-aed9-3cd19b7a6192","model":"gpt-5","history_log_id":2532619,"history_entry_count":342,"rollout_path":"/tmp/codex-test-rollout.jsonl"}}} {"ts":"2025-08-09T16:06:58.136Z","dir":"to_tui","kind":"insert_history","lines":9} {"ts":"2025-08-09T16:06:58.136Z","dir":"to_tui","kind":"app_event","variant":"RequestRedraw"} {"ts":"2025-08-09T16:06:58.136Z","dir":"to_tui","kind":"app_event","variant":"RequestRedraw"} From af2d8e14c76818fcd4b3c219447a3d52e790a4d5 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Tue, 9 Sep 2025 00:12:02 -0700 Subject: [PATCH 2/2] feat: add ArchiveConversation to ClientRequest --- codex-rs/core/src/conversation_manager.rs | 11 +- codex-rs/core/src/lib.rs | 2 + codex-rs/core/src/rollout/mod.rs | 3 +- .../mcp-server/src/codex_message_processor.rs | 144 ++++++++++++++++++ .../mcp-server/tests/common/mcp_process.rs | 10 ++ .../tests/suite/archive_conversation.rs | 105 +++++++++++++ codex-rs/mcp-server/tests/suite/mod.rs | 1 + codex-rs/protocol/src/mcp_protocol.rs | 17 +++ 8 files changed, 290 insertions(+), 3 deletions(-) create mode 100644 codex-rs/mcp-server/tests/suite/archive_conversation.rs diff --git a/codex-rs/core/src/conversation_manager.rs b/codex-rs/core/src/conversation_manager.rs index 6fac42d59f..ec6e5a97c8 100644 --- a/codex-rs/core/src/conversation_manager.rs +++ b/codex-rs/core/src/conversation_manager.rs @@ -145,8 +145,15 @@ impl ConversationManager { self.finalize_spawn(codex, conversation_id).await } - pub async fn remove_conversation(&self, conversation_id: ConversationId) { - self.conversations.write().await.remove(&conversation_id); + /// Removes the conversation from the manager's internal map, though the + /// conversation is stored as `Arc`, it is possible that + /// other references to it exist elsewhere. Returns `true` if the + /// conversation was found and removed; otherwise, `false`. + pub async fn remove_conversation( + &self, + conversation_id: &ConversationId, + ) -> Option> { + self.conversations.write().await.remove(conversation_id) } /// Fork an existing conversation by dropping the last `drop_last_messages` diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 3b47830c50..b8cd7bc55a 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -61,7 +61,9 @@ pub mod spawn; pub mod terminal; mod tool_apply_patch; pub mod turn_diff_tracker; +pub use rollout::ARCHIVED_SESSIONS_SUBDIR; pub use rollout::RolloutRecorder; +pub use rollout::SESSIONS_SUBDIR; pub use rollout::SessionMeta; pub use rollout::list::ConversationItem; pub use rollout::list::ConversationsPage; diff --git a/codex-rs/core/src/rollout/mod.rs b/codex-rs/core/src/rollout/mod.rs index 4883517c1f..a78d760959 100644 --- a/codex-rs/core/src/rollout/mod.rs +++ b/codex-rs/core/src/rollout/mod.rs @@ -1,6 +1,7 @@ //! Rollout module: persistence and discovery of session rollout files. -pub(crate) const SESSIONS_SUBDIR: &str = "sessions"; +pub const SESSIONS_SUBDIR: &str = "sessions"; +pub const ARCHIVED_SESSIONS_SUBDIR: &str = "archived_sessions"; pub mod list; pub(crate) mod policy; diff --git a/codex-rs/mcp-server/src/codex_message_processor.rs b/codex-rs/mcp-server/src/codex_message_processor.rs index 2169a8f27f..e6133e174b 100644 --- a/codex-rs/mcp-server/src/codex_message_processor.rs +++ b/codex-rs/mcp-server/src/codex_message_processor.rs @@ -35,6 +35,8 @@ use codex_protocol::mcp_protocol::AddConversationListenerParams; use codex_protocol::mcp_protocol::AddConversationSubscriptionResponse; use codex_protocol::mcp_protocol::ApplyPatchApprovalParams; use codex_protocol::mcp_protocol::ApplyPatchApprovalResponse; +use codex_protocol::mcp_protocol::ArchiveConversationParams; +use codex_protocol::mcp_protocol::ArchiveConversationResponse; use codex_protocol::mcp_protocol::AuthMode; use codex_protocol::mcp_protocol::AuthStatusChangeNotification; use codex_protocol::mcp_protocol::ClientRequest; @@ -73,12 +75,16 @@ use codex_protocol::protocol::USER_MESSAGE_BEGIN; use mcp_types::JSONRPCErrorError; use mcp_types::RequestId; use std::collections::HashMap; +use std::ffi::OsStr; use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; +use tokio::select; use tokio::sync::Mutex; use tokio::sync::oneshot; use tracing::error; +use tracing::info; +use tracing::warn; use uuid::Uuid; // Duration before a ChatGPT login attempt is abandoned. @@ -142,6 +148,9 @@ impl CodexMessageProcessor { ClientRequest::ResumeConversation { request_id, params } => { self.handle_resume_conversation(request_id, params).await; } + ClientRequest::ArchiveConversation { request_id, params } => { + self.archive_conversation(request_id, params).await; + } ClientRequest::SendUserMessage { request_id, params } => { self.send_user_message(request_id, params).await; } @@ -670,6 +679,141 @@ impl CodexMessageProcessor { } } + async fn archive_conversation(&self, request_id: RequestId, params: ArchiveConversationParams) { + let ArchiveConversationParams { + conversation_id, + rollout_path, + } = params; + + // Verify that the rollout path is in the sessions directory or else + // a malicious client could specify an arbitrary path. + 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!("{}.jsonl", conversation_id.0); + 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(); + + // Establish the listener for ShutdownComplete before submitting + // Shutdown so it is not missed. + 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; + } + } + } + } + }); + + // Request shutdown. + match conversation.submit(Op::Shutdown).await { + Ok(_) => { + // Successfully submitted Shutdown; wait before proceeding. + select! { + _ = is_shutdown => { + // Normal shutdown: proceed with archive. + } + _ = 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(); + // Perhaps we lost a shutdown race, so let's continue to + // clean up the .jsonl file. + } + } + } + + // Move the .jsonl file to the archived sessions subdir. + 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 = ArchiveConversationResponse {}; + 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 send_user_message(&self, request_id: RequestId, params: SendUserMessageParams) { let SendUserMessageParams { conversation_id, diff --git a/codex-rs/mcp-server/tests/common/mcp_process.rs b/codex-rs/mcp-server/tests/common/mcp_process.rs index cebc332a3c..64f2cc3852 100644 --- a/codex-rs/mcp-server/tests/common/mcp_process.rs +++ b/codex-rs/mcp-server/tests/common/mcp_process.rs @@ -13,6 +13,7 @@ use anyhow::Context; use assert_cmd::prelude::*; use codex_mcp_server::CodexToolCallParam; use codex_protocol::mcp_protocol::AddConversationListenerParams; +use codex_protocol::mcp_protocol::ArchiveConversationParams; use codex_protocol::mcp_protocol::CancelLoginChatGptParams; use codex_protocol::mcp_protocol::GetAuthStatusParams; use codex_protocol::mcp_protocol::InterruptConversationParams; @@ -186,6 +187,15 @@ impl McpProcess { self.send_request("newConversation", params).await } + /// Send an `archiveConversation` JSON-RPC request. + pub async fn send_archive_conversation_request( + &mut self, + params: ArchiveConversationParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("archiveConversation", params).await + } + /// Send an `addConversationListener` JSON-RPC request. pub async fn send_add_conversation_listener_request( &mut self, diff --git a/codex-rs/mcp-server/tests/suite/archive_conversation.rs b/codex-rs/mcp-server/tests/suite/archive_conversation.rs new file mode 100644 index 0000000000..e54a99896c --- /dev/null +++ b/codex-rs/mcp-server/tests/suite/archive_conversation.rs @@ -0,0 +1,105 @@ +use std::path::Path; + +use codex_core::ARCHIVED_SESSIONS_SUBDIR; +use codex_protocol::mcp_protocol::ArchiveConversationParams; +use codex_protocol::mcp_protocol::ArchiveConversationResponse; +use codex_protocol::mcp_protocol::NewConversationParams; +use codex_protocol::mcp_protocol::NewConversationResponse; +use mcp_test_support::McpProcess; +use mcp_test_support::to_response; +use mcp_types::JSONRPCResponse; +use mcp_types::RequestId; +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 archive_conversation_moves_rollout_into_archived_directory() { + let codex_home = TempDir::new().expect("create temp dir"); + create_config_toml(codex_home.path()).expect("write config.toml"); + + let mut mcp = McpProcess::new(codex_home.path()) + .await + .expect("spawn mcp process"); + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()) + .await + .expect("initialize timeout") + .expect("initialize request"); + + let new_request_id = mcp + .send_new_conversation_request(NewConversationParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await + .expect("send newConversation"); + let new_response: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(new_request_id)), + ) + .await + .expect("newConversation timeout") + .expect("newConversation response"); + + let NewConversationResponse { + conversation_id, + rollout_path, + .. + } = to_response::(new_response) + .expect("deserialize newConversation response"); + + assert!( + rollout_path.exists(), + "expected rollout path {} to exist", + rollout_path.display() + ); + + let archive_request_id = mcp + .send_archive_conversation_request(ArchiveConversationParams { + conversation_id, + rollout_path: rollout_path.clone(), + }) + .await + .expect("send archiveConversation"); + let archive_response: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(archive_request_id)), + ) + .await + .expect("archiveConversation timeout") + .expect("archiveConversation response"); + + let _: ArchiveConversationResponse = + to_response::(archive_response) + .expect("deserialize archiveConversation response"); + + let archived_directory = codex_home.path().join(ARCHIVED_SESSIONS_SUBDIR); + let archived_rollout_path = + archived_directory.join(rollout_path.file_name().unwrap_or_else(|| { + panic!("rollout path {} missing file name", rollout_path.display()) + })); + + 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() + ); +} + +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" +"# +} diff --git a/codex-rs/mcp-server/tests/suite/mod.rs b/codex-rs/mcp-server/tests/suite/mod.rs index 4a9220da91..7f89cc3301 100644 --- a/codex-rs/mcp-server/tests/suite/mod.rs +++ b/codex-rs/mcp-server/tests/suite/mod.rs @@ -1,4 +1,5 @@ // Aggregates all former standalone integration tests as modules. +mod archive_conversation; mod auth; mod codex_message_processor_flow; mod codex_tool; diff --git a/codex-rs/protocol/src/mcp_protocol.rs b/codex-rs/protocol/src/mcp_protocol.rs index 70e334036f..00391c7d01 100644 --- a/codex-rs/protocol/src/mcp_protocol.rs +++ b/codex-rs/protocol/src/mcp_protocol.rs @@ -91,6 +91,11 @@ pub enum ClientRequest { request_id: RequestId, params: ResumeConversationParams, }, + ArchiveConversation { + #[serde(rename = "id")] + request_id: RequestId, + params: ArchiveConversationParams, + }, SendUserMessage { #[serde(rename = "id")] request_id: RequestId, @@ -263,6 +268,18 @@ pub struct AddConversationSubscriptionResponse { pub subscription_id: Uuid, } +/// The [`ConversationId`] must match the `rollout_path`. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, TS)] +#[serde(rename_all = "camelCase")] +pub struct ArchiveConversationParams { + pub conversation_id: ConversationId, + pub rollout_path: PathBuf, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, TS)] +#[serde(rename_all = "camelCase")] +pub struct ArchiveConversationResponse {} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, TS)] #[serde(rename_all = "camelCase")] pub struct RemoveConversationSubscriptionResponse {}