From e6c206d19d4e9e2735e96d05b2fd3e1b92b36d44 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Tue, 13 May 2025 19:22:16 -0700 Subject: [PATCH 1/2] fix: tighten up some logic around session timestamps and ids (#922) * update `SessionConfigured` event to include the UUID for the session * show the UUID in the Rust TUI * use local timestamps in log files instead of UTC * include timestamps in log file names for easier discovery --- codex-rs/Cargo.lock | 13 ++++ codex-rs/core/Cargo.toml | 4 +- codex-rs/core/src/codex.rs | 20 +++-- codex-rs/core/src/protocol.rs | 15 +++- codex-rs/core/src/rollout.rs | 21 ++--- codex-rs/tui/Cargo.toml | 1 + codex-rs/tui/src/chatwidget.rs | 12 +-- .../tui/src/conversation_history_widget.rs | 14 ++-- codex-rs/tui/src/history_cell.rs | 78 ++++++++++--------- 9 files changed, 101 insertions(+), 77 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 15a6298385..d67a2df70a 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -639,6 +639,7 @@ dependencies = [ "tui-input", "tui-markdown", "tui-textarea", + "uuid", ] [[package]] @@ -2275,6 +2276,15 @@ dependencies = [ "libc", ] +[[package]] +name = "num_threads" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c7398b9c8b70908f6371f47ed36737907c87c52af34c268fed0bf0ceb92ead9" +dependencies = [ + "libc", +] + [[package]] name = "object" version = "0.32.2" @@ -3686,7 +3696,9 @@ checksum = "8a7619e19bc266e0f9c5e6686659d394bc57973859340060a69221e57dbc0c40" dependencies = [ "deranged", "itoa", + "libc", "num-conv", + "num_threads", "powerfmt", "serde", "time-core", @@ -4097,6 +4109,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "458f7a779bf54acc9f347480ac654f68407d3aab21269a6e3c9f922acd9e2da9" dependencies = [ "getrandom 0.3.2", + "serde", ] [[package]] diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index 6154d91d0c..e7a93d3dea 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -31,7 +31,7 @@ reqwest = { version = "0.12", features = ["json", "stream"] } serde = { version = "1", features = ["derive"] } serde_json = "1" thiserror = "2.0.12" -time = { version = "0.3", features = ["formatting", "macros"] } +time = { version = "0.3", features = ["formatting", "local-offset", "macros"] } tokio = { version = "1", features = [ "io-std", "macros", @@ -44,7 +44,7 @@ toml = "0.8.20" tracing = { version = "0.1.41", features = ["log"] } tree-sitter = "0.25.3" tree-sitter-bash = "0.23.3" -uuid = { version = "1", features = ["v4"] } +uuid = { version = "1", features = ["serde", "v4"] } [target.'cfg(target_os = "linux")'.dependencies] libc = "0.2.172" diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 82296ccb5d..26e1f665bf 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -30,6 +30,7 @@ use tracing::error; use tracing::info; use tracing::trace; use tracing::warn; +use uuid::Uuid; use crate::WireApi; use crate::client::ModelClient; @@ -62,6 +63,7 @@ use crate::protocol::InputItem; use crate::protocol::Op; use crate::protocol::ReviewDecision; use crate::protocol::SandboxPolicy; +use crate::protocol::SessionConfiguredEvent; use crate::protocol::Submission; use crate::rollout::RolloutRecorder; use crate::safety::SafetyCheck; @@ -596,13 +598,15 @@ async fn submission_loop( // Attempt to create a RolloutRecorder *before* moving the // `instructions` value into the Session struct. - let rollout_recorder = match RolloutRecorder::new(instructions.clone()).await { - Ok(r) => Some(r), - Err(e) => { - tracing::warn!("failed to initialise rollout recorder: {e}"); - None - } - }; + let session_id = Uuid::new_v4(); + let rollout_recorder = + match RolloutRecorder::new(session_id, instructions.clone()).await { + Ok(r) => Some(r), + Err(e) => { + tracing::warn!("failed to initialise rollout recorder: {e}"); + None + } + }; sess = Some(Arc::new(Session { client, @@ -622,7 +626,7 @@ async fn submission_loop( // ack let events = std::iter::once(Event { id: sub.id.clone(), - msg: EventMsg::SessionConfigured { model }, + msg: EventMsg::SessionConfigured(SessionConfiguredEvent { session_id, model }), }) .chain(mcp_connection_errors.into_iter()); for event in events { diff --git a/codex-rs/core/src/protocol.rs b/codex-rs/core/src/protocol.rs index 1069a90499..e4b8382635 100644 --- a/codex-rs/core/src/protocol.rs +++ b/codex-rs/core/src/protocol.rs @@ -10,6 +10,7 @@ use std::path::PathBuf; use mcp_types::CallToolResult; use serde::Deserialize; use serde::Serialize; +use uuid::Uuid; use crate::model_provider_info::ModelProviderInfo; @@ -323,10 +324,7 @@ pub enum EventMsg { }, /// Ack the client's configure message. - SessionConfigured { - /// Tell the client what model is being queried. - model: String, - }, + SessionConfigured(SessionConfiguredEvent), McpToolCallBegin { /// Identifier so this can be paired with the McpToolCallEnd event. @@ -429,6 +427,15 @@ pub enum EventMsg { }, } +#[derive(Debug, Default, Clone, Deserialize, Serialize)] +pub struct SessionConfiguredEvent { + /// Unique id for this session. + pub session_id: Uuid, + + /// Tell the client what model is being queried. + pub model: String, +} + /// User's decision in response to an ExecApprovalRequest. #[derive(Debug, Default, Clone, Copy, Deserialize, Serialize)] #[serde(rename_all = "snake_case")] diff --git a/codex-rs/core/src/rollout.rs b/codex-rs/core/src/rollout.rs index 2a45222a4e..7a014f401c 100644 --- a/codex-rs/core/src/rollout.rs +++ b/codex-rs/core/src/rollout.rs @@ -37,8 +37,8 @@ struct SessionMeta { /// Rollouts are recorded as JSONL and can be inspected with tools such as: /// /// ```ignore -/// $ jq -C . ~/.codex/sessions/rollout-2025-05-07-5973b6c0-94b8-487b-a530-2aeb6098ae0e.jsonl -/// $ fx ~/.codex/sessions/rollout-2025-05-07-5973b6c0-94b8-487b-a530-2aeb6098ae0e.jsonl +/// $ jq -C . ~/.codex/sessions/rollout-2025-05-07T17-24-21-5973b6c0-94b8-487b-a530-2aeb6098ae0e.jsonl +/// $ fx ~/.codex/sessions/rollout-2025-05-07T17-24-21-5973b6c0-94b8-487b-a530-2aeb6098ae0e.jsonl /// ``` #[derive(Clone)] pub(crate) struct RolloutRecorder { @@ -49,12 +49,12 @@ impl RolloutRecorder { /// Attempt to create a new [`RolloutRecorder`]. If the sessions directory /// 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(instructions: Option) -> std::io::Result { + pub async fn new(uuid: Uuid, instructions: Option) -> std::io::Result { let LogFileInfo { file, session_id, timestamp, - } = create_log_file()?; + } = create_log_file(uuid)?; // Build the static session metadata JSON first. let timestamp_format: &[FormatItem] = format_description!( @@ -154,18 +154,19 @@ struct LogFileInfo { timestamp: OffsetDateTime, } -fn create_log_file() -> std::io::Result { +fn create_log_file(session_id: Uuid) -> std::io::Result { // Resolve ~/.codex/sessions and create it if missing. let mut dir = codex_dir()?; dir.push(SESSIONS_SUBDIR); fs::create_dir_all(&dir)?; - // Generate a v4 UUID – matches the JS CLI implementation. - let session_id = Uuid::new_v4(); - let timestamp = OffsetDateTime::now_utc(); + let timestamp = OffsetDateTime::now_local() + .map_err(|e| IoError::new(ErrorKind::Other, format!("failed to get local time: {e}")))?; - // Custom format for YYYY-MM-DD. - let format: &[FormatItem] = format_description!("[year]-[month]-[day]"); + // Custom format for YYYY-MM-DDThh-mm-ss. Use `-` instead of `:` for + // compatibility with filesystems that do not allow colons in filenames. + let format: &[FormatItem] = + format_description!("[year]-[month]-[day]T[hour]-[minute]-[second]"); let date_str = timestamp .format(format) .map_err(|e| IoError::new(ErrorKind::Other, format!("failed to format timestamp: {e}")))?; diff --git a/codex-rs/tui/Cargo.toml b/codex-rs/tui/Cargo.toml index 230cbd2b17..4bd23015e9 100644 --- a/codex-rs/tui/Cargo.toml +++ b/codex-rs/tui/Cargo.toml @@ -42,3 +42,4 @@ tracing-subscriber = { version = "0.3.19", features = ["env-filter"] } tui-input = "0.11.1" tui-markdown = "0.3.3" tui-textarea = "0.7.0" +uuid = { version = "1" } diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index c9a04b7b0a..accb73053c 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -102,8 +102,6 @@ impl ChatWidget<'_> { config, }; - let _ = chat_widget.submit_welcome_message(); - if initial_prompt.is_some() || !initial_images.is_empty() { let text = initial_prompt.unwrap_or_default(); let _ = chat_widget.submit_user_message_with_images(text, initial_images); @@ -161,12 +159,6 @@ impl ChatWidget<'_> { } } - fn submit_welcome_message(&mut self) -> std::result::Result<(), SendError> { - self.conversation_history.add_welcome_message(&self.config); - self.request_redraw()?; - Ok(()) - } - fn submit_user_message( &mut self, text: String, @@ -215,10 +207,10 @@ impl ChatWidget<'_> { ) -> std::result::Result<(), SendError> { let Event { id, msg } = event; match msg { - EventMsg::SessionConfigured { model } => { + EventMsg::SessionConfigured(event) => { // Record session information at the top of the conversation. self.conversation_history - .add_session_info(&self.config, model); + .add_session_info(&self.config, event); self.request_redraw()?; } EventMsg::AgentMessage { message } => { diff --git a/codex-rs/tui/src/conversation_history_widget.rs b/codex-rs/tui/src/conversation_history_widget.rs index 70e7b6c46e..f7a9405954 100644 --- a/codex-rs/tui/src/conversation_history_widget.rs +++ b/codex-rs/tui/src/conversation_history_widget.rs @@ -3,6 +3,7 @@ use crate::history_cell::HistoryCell; use crate::history_cell::PatchEventType; use codex_core::config::Config; use codex_core::protocol::FileChange; +use codex_core::protocol::SessionConfiguredEvent; use crossterm::event::KeyCode; use crossterm::event::KeyEvent; use ratatui::prelude::*; @@ -162,8 +163,11 @@ impl ConversationHistoryWidget { self.scroll_position = usize::MAX; } - pub fn add_welcome_message(&mut self, config: &Config) { - self.add_to_history(HistoryCell::new_welcome_message(config)); + /// Note `model` could differ from `config.model` if the agent decided to + /// use a different model than the one requested by the user. + pub fn add_session_info(&mut self, config: &Config, event: SessionConfiguredEvent) { + let is_first_event = self.history.is_empty(); + self.add_to_history(HistoryCell::new_session_info(config, event, is_first_event)); } pub fn add_user_message(&mut self, message: String) { @@ -195,12 +199,6 @@ impl ConversationHistoryWidget { self.add_to_history(HistoryCell::new_patch_event(event_type, changes)); } - /// Note `model` could differ from `config.model` if the agent decided to - /// use a different model than the one requested by the user. - pub fn add_session_info(&mut self, config: &Config, model: String) { - self.add_to_history(HistoryCell::new_session_info(config, model)); - } - pub fn add_active_exec_command(&mut self, call_id: String, command: Vec) { self.add_to_history(HistoryCell::new_active_exec_command(call_id, command)); } diff --git a/codex-rs/tui/src/history_cell.rs b/codex-rs/tui/src/history_cell.rs index 4f4259aaa6..23ce66679b 100644 --- a/codex-rs/tui/src/history_cell.rs +++ b/codex-rs/tui/src/history_cell.rs @@ -2,6 +2,7 @@ use codex_ansi_escape::ansi_escape_line; use codex_common::elapsed::format_duration; use codex_core::config::Config; use codex_core::protocol::FileChange; +use codex_core::protocol::SessionConfiguredEvent; use ratatui::prelude::*; use ratatui::style::Color; use ratatui::style::Modifier; @@ -94,29 +95,50 @@ pub(crate) enum HistoryCell { const TOOL_CALL_MAX_LINES: usize = 5; impl HistoryCell { - pub(crate) fn new_welcome_message(config: &Config) -> Self { - let mut lines: Vec> = vec![ - Line::from(vec![ - "OpenAI ".into(), - "Codex".bold(), - " (research preview)".dim(), - ]), - Line::from(""), - Line::from("codex session:".magenta().bold()), - ]; + pub(crate) fn new_session_info( + config: &Config, + event: SessionConfiguredEvent, + is_first_event: bool, + ) -> Self { + let SessionConfiguredEvent { model, session_id } = event; + if is_first_event { + let mut lines: Vec> = vec![ + Line::from(vec![ + "OpenAI ".into(), + "Codex".bold(), + " (research preview)".dim(), + ]), + Line::from(""), + Line::from(vec![ + "codex session".magenta().bold(), + " ".into(), + session_id.to_string().dim(), + ]), + ]; - let entries = vec![ - ("workdir", config.cwd.display().to_string()), - ("model", config.model.clone()), - ("provider", config.model_provider_id.clone()), - ("approval", format!("{:?}", config.approval_policy)), - ("sandbox", format!("{:?}", config.sandbox_policy)), - ]; - for (key, value) in entries { - lines.push(Line::from(vec![format!("{key}: ").bold(), value.into()])); + let entries = vec![ + ("workdir", config.cwd.display().to_string()), + ("model", config.model.clone()), + ("provider", config.model_provider_id.clone()), + ("approval", format!("{:?}", config.approval_policy)), + ("sandbox", format!("{:?}", config.sandbox_policy)), + ]; + for (key, value) in entries { + lines.push(Line::from(vec![format!("{key}: ").bold(), value.into()])); + } + lines.push(Line::from("")); + HistoryCell::WelcomeMessage { lines } + } else if config.model == model { + HistoryCell::SessionInfo { lines: vec![] } + } else { + let lines = vec![ + Line::from("model changed:".magenta().bold()), + Line::from(format!("requested: {}", config.model)), + Line::from(format!("used: {}", model)), + Line::from(""), + ]; + HistoryCell::SessionInfo { lines } } - lines.push(Line::from("")); - HistoryCell::WelcomeMessage { lines } } pub(crate) fn new_user_prompt(message: String) -> Self { @@ -296,20 +318,6 @@ impl HistoryCell { HistoryCell::ErrorEvent { lines } } - pub(crate) fn new_session_info(config: &Config, model: String) -> Self { - if config.model == model { - HistoryCell::SessionInfo { lines: vec![] } - } else { - let lines = vec![ - Line::from("model changed:".magenta().bold()), - Line::from(format!("requested: {}", config.model)), - Line::from(format!("used: {}", model)), - Line::from(""), - ]; - HistoryCell::SessionInfo { lines } - } - } - /// Create a new `PendingPatch` cell that lists the file‑level summary of /// a proposed patch. The summary lines should already be formatted (e.g. /// "A path/to/file.rs"). From 001a24460db8960907eb6bb125a94ccf89a906af Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Tue, 13 May 2025 20:08:43 -0700 Subject: [PATCH 2/2] fix: change EventMsg enum so every variant takes a single struct --- codex-rs/core/src/codex.rs | 62 +++--- codex-rs/core/src/codex_wrapper.rs | 2 +- codex-rs/core/src/mcp_tool_call.rs | 18 +- codex-rs/core/src/protocol.rs | 218 +++++++++++-------- codex-rs/core/tests/live_agent.rs | 33 ++- codex-rs/core/tests/previous_response_id.rs | 2 +- codex-rs/exec/src/event_processor.rs | 38 ++-- codex-rs/mcp-server/src/codex_tool_runner.rs | 8 +- codex-rs/tui/src/chatwidget.rs | 47 ++-- 9 files changed, 244 insertions(+), 184 deletions(-) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 26e1f665bf..dfc9c1dce9 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -55,12 +55,24 @@ use crate::models::ResponseInputItem; use crate::models::ResponseItem; use crate::models::ShellToolCallParams; use crate::project_doc::create_full_instructions; +use crate::protocol::AgentMessageEvent; +use crate::protocol::AgentReasoningEvent; +use crate::protocol::ApplyPatchApprovalRequestEvent; use crate::protocol::AskForApproval; +use crate::protocol::BackgroundEventEvent; +use crate::protocol::ErrorEvent; use crate::protocol::Event; use crate::protocol::EventMsg; +use crate::protocol::ExecApprovalRequestEvent; +use crate::protocol::ExecCommandBeginEvent; +use crate::protocol::ExecCommandEndEvent; use crate::protocol::FileChange; use crate::protocol::InputItem; +use crate::protocol::McpToolCallBeginEvent; +use crate::protocol::McpToolCallEndEvent; use crate::protocol::Op; +use crate::protocol::PatchApplyBeginEvent; +use crate::protocol::PatchApplyEndEvent; use crate::protocol::ReviewDecision; use crate::protocol::SandboxPolicy; use crate::protocol::SessionConfiguredEvent; @@ -227,11 +239,11 @@ impl Session { let (tx_approve, rx_approve) = oneshot::channel(); let event = Event { id: sub_id.clone(), - msg: EventMsg::ExecApprovalRequest { + msg: EventMsg::ExecApprovalRequest(ExecApprovalRequestEvent { command, cwd, reason, - }, + }), }; let _ = self.tx_event.send(event).await; { @@ -251,11 +263,11 @@ impl Session { let (tx_approve, rx_approve) = oneshot::channel(); let event = Event { id: sub_id.clone(), - msg: EventMsg::ApplyPatchApprovalRequest { + msg: EventMsg::ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent { changes: convert_apply_patch_to_protocol(action), reason, grant_root, - }, + }), }; let _ = self.tx_event.send(event).await; { @@ -297,11 +309,11 @@ impl Session { async fn notify_exec_command_begin(&self, sub_id: &str, call_id: &str, params: &ExecParams) { let event = Event { id: sub_id.to_string(), - msg: EventMsg::ExecCommandBegin { + msg: EventMsg::ExecCommandBegin(ExecCommandBeginEvent { call_id: call_id.to_string(), command: params.command.clone(), cwd: params.cwd.clone(), - }, + }), }; let _ = self.tx_event.send(event).await; } @@ -319,12 +331,12 @@ impl Session { id: sub_id.to_string(), // Because stdout and stderr could each be up to 100 KiB, we send // truncated versions. - msg: EventMsg::ExecCommandEnd { + msg: EventMsg::ExecCommandEnd(ExecCommandEndEvent { call_id: call_id.to_string(), stdout: stdout.chars().take(MAX_STREAM_OUTPUT).collect(), stderr: stderr.chars().take(MAX_STREAM_OUTPUT).collect(), exit_code, - }, + }), }; let _ = self.tx_event.send(event).await; } @@ -335,9 +347,9 @@ impl Session { async fn notify_background_event(&self, sub_id: &str, message: impl Into) { let event = Event { id: sub_id.to_string(), - msg: EventMsg::BackgroundEvent { + msg: EventMsg::BackgroundEvent(BackgroundEventEvent { message: message.into(), - }, + }), }; let _ = self.tx_event.send(event).await; } @@ -460,9 +472,9 @@ impl AgentTask { self.handle.abort(); let event = Event { id: self.sub_id, - msg: EventMsg::Error { + msg: EventMsg::Error(ErrorEvent { message: "Turn interrupted".to_string(), - }, + }), }; let tx_event = self.sess.tx_event.clone(); tokio::spawn(async move { @@ -483,10 +495,10 @@ async fn submission_loop( let send_no_session_event = |sub_id: String| async { let event = Event { id: sub_id, - msg: EventMsg::Error { + msg: EventMsg::Error(ErrorEvent { message: "No session initialized, expected 'ConfigureSession' as first Op" .to_string(), - }, + }), }; tx_event.send(event).await.ok(); }; @@ -534,7 +546,7 @@ async fn submission_loop( error!(message); let event = Event { id: sub.id, - msg: EventMsg::Error { message }, + msg: EventMsg::Error(ErrorEvent { message }), }; if let Err(e) = tx_event.send(event).await { error!("failed to send error message: {e:?}"); @@ -577,7 +589,7 @@ async fn submission_loop( error!("{message}"); mcp_connection_errors.push(Event { id: sub.id.clone(), - msg: EventMsg::Error { message }, + msg: EventMsg::Error(ErrorEvent { message }), }); (McpConnectionManager::default(), Default::default()) } @@ -591,7 +603,7 @@ async fn submission_loop( error!("{message}"); mcp_connection_errors.push(Event { id: sub.id.clone(), - msg: EventMsg::Error { message }, + msg: EventMsg::Error(ErrorEvent { message }), }); } } @@ -792,9 +804,9 @@ async fn run_task(sess: Arc, sub_id: String, input: Vec) { info!("Turn error: {e:#}"); let event = Event { id: sub_id.clone(), - msg: EventMsg::Error { + msg: EventMsg::Error(ErrorEvent { message: e.to_string(), - }, + }), }; sess.tx_event.send(event).await.ok(); return; @@ -933,7 +945,7 @@ async fn handle_response_item( if let ContentItem::OutputText { text } = item { let event = Event { id: sub_id.to_string(), - msg: EventMsg::AgentMessage { message: text }, + msg: EventMsg::AgentMessage(AgentMessageEvent { message: text }), }; sess.tx_event.send(event).await.ok(); } @@ -946,7 +958,7 @@ async fn handle_response_item( }; let event = Event { id: sub_id.to_string(), - msg: EventMsg::AgentReasoning { text }, + msg: EventMsg::AgentReasoning(AgentReasoningEvent { text }), }; sess.tx_event.send(event).await.ok(); } @@ -1346,11 +1358,11 @@ async fn apply_patch( .tx_event .send(Event { id: sub_id.clone(), - msg: EventMsg::PatchApplyBegin { + msg: EventMsg::PatchApplyBegin(PatchApplyBeginEvent { call_id: call_id.clone(), auto_approved, changes: convert_apply_patch_to_protocol(&action), - }, + }), }) .await; @@ -1435,12 +1447,12 @@ async fn apply_patch( .tx_event .send(Event { id: sub_id.clone(), - msg: EventMsg::PatchApplyEnd { + msg: EventMsg::PatchApplyEnd(PatchApplyEndEvent { call_id: call_id.clone(), stdout: String::from_utf8_lossy(&stdout).to_string(), stderr: String::from_utf8_lossy(&stderr).to_string(), success: success_flag, - }, + }), }) .await; diff --git a/codex-rs/core/src/codex_wrapper.rs b/codex-rs/core/src/codex_wrapper.rs index 431b580c96..f2ece22da7 100644 --- a/codex-rs/core/src/codex_wrapper.rs +++ b/codex-rs/core/src/codex_wrapper.rs @@ -24,7 +24,7 @@ pub async fn init_codex(config: Config) -> anyhow::Result<(Codex, Event, Arc ( - EventMsg::McpToolCallEnd { + EventMsg::McpToolCallEnd(McpToolCallEndEvent { call_id, success: !result.is_error.unwrap_or(false), result: Some(result), - }, + }), None, ), Err(e) => ( - EventMsg::McpToolCallEnd { + EventMsg::McpToolCallEnd(McpToolCallEndEvent { call_id, success: false, result: None, - }, + }), Some(e), ), }; notify_mcp_tool_call_event(sess, sub_id, tool_call_end_event.clone()).await; - let EventMsg::McpToolCallEnd { + let EventMsg::McpToolCallEnd(McpToolCallEndEvent { call_id, success, result, - } = tool_call_end_event + }) = tool_call_end_event else { unimplemented!("unexpected event type"); }; diff --git a/codex-rs/core/src/protocol.rs b/codex-rs/core/src/protocol.rs index e4b8382635..d097ca77de 100644 --- a/codex-rs/core/src/protocol.rs +++ b/codex-rs/core/src/protocol.rs @@ -303,9 +303,7 @@ pub struct Event { #[serde(tag = "type", rename_all = "snake_case")] pub enum EventMsg { /// Error while executing a submission - Error { - message: String, - }, + Error(ErrorEvent), /// Agent has started a task TaskStarted, @@ -314,117 +312,145 @@ pub enum EventMsg { TaskComplete, /// Agent text output message - AgentMessage { - message: String, - }, + AgentMessage(AgentMessageEvent), /// Reasoning event from agent. - AgentReasoning { - text: String, - }, + AgentReasoning(AgentReasoningEvent), /// Ack the client's configure message. SessionConfigured(SessionConfiguredEvent), - McpToolCallBegin { - /// Identifier so this can be paired with the McpToolCallEnd event. - call_id: String, + McpToolCallBegin(McpToolCallBeginEvent), - /// Name of the MCP server as defined in the config. - server: String, - - /// Name of the tool as given by the MCP server. - tool: String, - - /// Arguments to the tool call. - arguments: Option, - }, - - McpToolCallEnd { - /// Identifier for the McpToolCallBegin that finished. - call_id: String, - - /// Whether the tool call was successful. If `false`, `result` might - /// not be present. - success: bool, - - /// Result of the tool call. Note this could be an error. - result: Option, - }, + McpToolCallEnd(McpToolCallEndEvent), /// Notification that the server is about to execute a command. - ExecCommandBegin { - /// Identifier so this can be paired with the ExecCommandEnd event. - call_id: String, - /// The command to be executed. - command: Vec, - /// The command's working directory if not the default cwd for the - /// agent. - cwd: PathBuf, - }, + ExecCommandBegin(ExecCommandBeginEvent), - ExecCommandEnd { - /// Identifier for the ExecCommandBegin that finished. - call_id: String, - /// Captured stdout - stdout: String, - /// Captured stderr - stderr: String, - /// The command's exit code. - exit_code: i32, - }, + ExecCommandEnd(ExecCommandEndEvent), - ExecApprovalRequest { - /// The command to be executed. - command: Vec, - /// The command's working directory. - cwd: PathBuf, - /// Optional human‑readable reason for the approval (e.g. retry without - /// sandbox). - #[serde(skip_serializing_if = "Option::is_none")] - reason: Option, - }, + ExecApprovalRequest(ExecApprovalRequestEvent), - ApplyPatchApprovalRequest { - changes: HashMap, - /// Optional explanatory reason (e.g. request for extra write access). - #[serde(skip_serializing_if = "Option::is_none")] - reason: Option, + ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent), - /// When set, the agent is asking the user to allow writes under this - /// root for the remainder of the session. - #[serde(skip_serializing_if = "Option::is_none")] - grant_root: Option, - }, - - BackgroundEvent { - message: String, - }, + BackgroundEvent(BackgroundEventEvent), /// Notification that the agent is about to apply a code patch. Mirrors /// `ExecCommandBegin` so front‑ends can show progress indicators. - PatchApplyBegin { - /// Identifier so this can be paired with the PatchApplyEnd event. - call_id: String, - - /// If true, there was no ApplyPatchApprovalRequest for this patch. - auto_approved: bool, - - /// The changes to be applied. - changes: HashMap, - }, + PatchApplyBegin(PatchApplyBeginEvent), /// Notification that a patch application has finished. - PatchApplyEnd { - /// Identifier for the PatchApplyBegin that finished. - call_id: String, - /// Captured stdout (summary printed by apply_patch). - stdout: String, - /// Captured stderr (parser errors, IO failures, etc.). - stderr: String, - /// Whether the patch was applied successfully. - success: bool, - }, + PatchApplyEnd(PatchApplyEndEvent), +} + +// Individual event payload types matching each `EventMsg` variant. + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct ErrorEvent { + pub message: String, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct AgentMessageEvent { + pub message: String, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct AgentReasoningEvent { + pub text: String, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct McpToolCallBeginEvent { + /// Identifier so this can be paired with the McpToolCallEnd event. + pub call_id: String, + /// Name of the MCP server as defined in the config. + pub server: String, + /// Name of the tool as given by the MCP server. + pub tool: String, + /// Arguments to the tool call. + pub arguments: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct McpToolCallEndEvent { + /// Identifier for the corresponding McpToolCallBegin that finished. + pub call_id: String, + /// Whether the tool call was successful. If `false`, `result` might not be present. + pub success: bool, + /// Result of the tool call. Note this could be an error. + pub result: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct ExecCommandBeginEvent { + /// Identifier so this can be paired with the ExecCommandEnd event. + pub call_id: String, + /// The command to be executed. + pub command: Vec, + /// The command's working directory if not the default cwd for the agent. + pub cwd: PathBuf, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct ExecCommandEndEvent { + /// Identifier for the ExecCommandBegin that finished. + pub call_id: String, + /// Captured stdout + pub stdout: String, + /// Captured stderr + pub stderr: String, + /// The command's exit code. + pub exit_code: i32, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct ExecApprovalRequestEvent { + /// The command to be executed. + pub command: Vec, + /// The command's working directory. + pub cwd: PathBuf, + /// Optional human-readable reason for the approval (e.g. retry without sandbox). + #[serde(skip_serializing_if = "Option::is_none")] + pub reason: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct ApplyPatchApprovalRequestEvent { + pub changes: HashMap, + /// Optional explanatory reason (e.g. request for extra write access). + #[serde(skip_serializing_if = "Option::is_none")] + pub reason: Option, + /// When set, the agent is asking the user to allow writes under this root for the remainder of the session. + #[serde(skip_serializing_if = "Option::is_none")] + pub grant_root: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct BackgroundEventEvent { + pub message: String, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct PatchApplyBeginEvent { + /// Identifier so this can be paired with the PatchApplyEnd event. + pub call_id: String, + /// If true, there was no ApplyPatchApprovalRequest for this patch. + pub auto_approved: bool, + /// The changes to be applied. + pub changes: HashMap, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct PatchApplyEndEvent { + /// Identifier for the PatchApplyBegin that finished. + pub call_id: String, + /// Captured stdout (summary printed by apply_patch). + pub stdout: String, + /// Captured stderr (parser errors, IO failures, etc.). + pub stderr: String, + /// Whether the patch was applied successfully. + pub success: bool, } #[derive(Debug, Default, Clone, Deserialize, Serialize)] diff --git a/codex-rs/core/tests/live_agent.rs b/codex-rs/core/tests/live_agent.rs index c43c5c193d..75ee023486 100644 --- a/codex-rs/core/tests/live_agent.rs +++ b/codex-rs/core/tests/live_agent.rs @@ -92,9 +92,11 @@ async fn live_streaming_and_prev_id_reset() { .expect("agent closed"); match ev.msg { - EventMsg::AgentMessage { .. } => saw_message_before_complete = true, + EventMsg::AgentMessage(_) => saw_message_before_complete = true, EventMsg::TaskComplete => break, - EventMsg::Error { message } => panic!("agent reported error in task1: {message}"), + EventMsg::Error(codex_core::protocol::ErrorEvent { message }) => { + panic!("agent reported error in task1: {message}") + } _ => (), } } @@ -122,11 +124,15 @@ async fn live_streaming_and_prev_id_reset() { .expect("agent closed"); match &ev.msg { - EventMsg::AgentMessage { message } if message.contains("second turn succeeded") => { + EventMsg::AgentMessage(codex_core::protocol::AgentMessageEvent { message }) + if message.contains("second turn succeeded") => + { got_expected = true; } EventMsg::TaskComplete => break, - EventMsg::Error { message } => panic!("agent reported error in task2: {message}"), + EventMsg::Error(codex_core::protocol::ErrorEvent { message }) => { + panic!("agent reported error in task2: {message}") + } _ => (), } } @@ -171,19 +177,28 @@ async fn live_shell_function_call() { .expect("agent closed"); match ev.msg { - EventMsg::ExecCommandBegin { command, .. } => { + EventMsg::ExecCommandBegin(codex_core::protocol::ExecCommandBeginEvent { + command, + call_id: _, + cwd: _, + }) => { assert_eq!(command, vec!["echo", MARKER]); saw_begin = true; } - EventMsg::ExecCommandEnd { - stdout, exit_code, .. - } => { + EventMsg::ExecCommandEnd(codex_core::protocol::ExecCommandEndEvent { + stdout, + exit_code, + call_id: _, + stderr: _, + }) => { assert_eq!(exit_code, 0, "echo returned non‑zero exit code"); assert!(stdout.contains(MARKER)); saw_end_with_output = true; } EventMsg::TaskComplete => break, - EventMsg::Error { message } => panic!("agent error during shell test: {message}"), + EventMsg::Error(codex_core::protocol::ErrorEvent { message }) => { + panic!("agent error during shell test: {message}") + } _ => (), } } diff --git a/codex-rs/core/tests/previous_response_id.rs b/codex-rs/core/tests/previous_response_id.rs index 2c899df0e9..7b5256aaef 100644 --- a/codex-rs/core/tests/previous_response_id.rs +++ b/codex-rs/core/tests/previous_response_id.rs @@ -150,7 +150,7 @@ async fn keeps_previous_response_id_between_tasks() { .unwrap(); match ev.msg { codex_core::protocol::EventMsg::TaskComplete => break, - codex_core::protocol::EventMsg::Error { message } => { + codex_core::protocol::EventMsg::Error(codex_core::protocol::ErrorEvent { message }) => { panic!("unexpected error: {message}") } _ => (), diff --git a/codex-rs/exec/src/event_processor.rs b/codex-rs/exec/src/event_processor.rs index d43f9d593c..a0cc77a95a 100644 --- a/codex-rs/exec/src/event_processor.rs +++ b/codex-rs/exec/src/event_processor.rs @@ -95,11 +95,11 @@ impl EventProcessor { pub(crate) fn process_event(&mut self, event: Event) { let Event { id, msg } = event; match msg { - EventMsg::Error { message } => { + EventMsg::Error(codex_core::protocol::ErrorEvent { message }) => { let prefix = "ERROR:".style(self.red); ts_println!("{prefix} {message}"); } - EventMsg::BackgroundEvent { message } => { + EventMsg::BackgroundEvent(codex_core::protocol::BackgroundEventEvent { message }) => { ts_println!("{}", message.style(self.dimmed)); } EventMsg::TaskStarted => { @@ -110,15 +110,15 @@ impl EventProcessor { let msg = format!("Task complete: {id}"); ts_println!("{}", msg.style(self.bold)); } - EventMsg::AgentMessage { message } => { + EventMsg::AgentMessage(codex_core::protocol::AgentMessageEvent { message }) => { let prefix = "Agent message:".style(self.bold); ts_println!("{prefix} {message}"); } - EventMsg::ExecCommandBegin { + EventMsg::ExecCommandBegin(codex_core::protocol::ExecCommandBeginEvent { call_id, command, cwd, - } => { + }) => { self.call_id_to_command.insert( call_id.clone(), ExecCommandBegin { @@ -133,12 +133,12 @@ impl EventProcessor { cwd.to_string_lossy(), ); } - EventMsg::ExecCommandEnd { + EventMsg::ExecCommandEnd(codex_core::protocol::ExecCommandEndEvent { call_id, stdout, stderr, exit_code, - } => { + }) => { let exec_command = self.call_id_to_command.remove(&call_id); let (duration, call) = if let Some(ExecCommandBegin { command, @@ -173,19 +173,21 @@ impl EventProcessor { } // Handle MCP tool calls (e.g. calling external functions via MCP). - EventMsg::McpToolCallBegin { + EventMsg::McpToolCallBegin(codex_core::protocol::McpToolCallBeginEvent { call_id, server, tool, arguments, - } => { + }) => { // Build fully-qualified tool name: server.tool let fq_tool_name = format!("{server}.{tool}"); // Format arguments as compact JSON so they fit on one line. let args_str = arguments .as_ref() - .map(|v| serde_json::to_string(v).unwrap_or_else(|_| v.to_string())) + .map(|v: &serde_json::Value| { + serde_json::to_string(v).unwrap_or_else(|_| v.to_string()) + }) .unwrap_or_default(); let invocation = if args_str.is_empty() { @@ -208,11 +210,11 @@ impl EventProcessor { invocation.style(self.bold), ); } - EventMsg::McpToolCallEnd { + EventMsg::McpToolCallEnd(codex_core::protocol::McpToolCallEndEvent { call_id, success, result, - } => { + }) => { // Retrieve start time and invocation for duration calculation and labeling. let info = self.call_id_to_tool_call.remove(&call_id); @@ -243,11 +245,11 @@ impl EventProcessor { } } } - EventMsg::PatchApplyBegin { + EventMsg::PatchApplyBegin(codex_core::protocol::PatchApplyBeginEvent { call_id, auto_approved, changes, - } => { + }) => { // Store metadata so we can calculate duration later when we // receive the corresponding PatchApplyEnd event. self.call_id_to_patch.insert( @@ -321,12 +323,12 @@ impl EventProcessor { } } } - EventMsg::PatchApplyEnd { + EventMsg::PatchApplyEnd(codex_core::protocol::PatchApplyEndEvent { call_id, stdout, stderr, success, - } => { + }) => { let patch_begin = self.call_id_to_patch.remove(&call_id); // Compute duration and summary label similar to exec commands. @@ -355,10 +357,10 @@ impl EventProcessor { println!("{}", line.style(self.dimmed)); } } - EventMsg::ExecApprovalRequest { .. } => { + EventMsg::ExecApprovalRequest(_) => { // Should we exit? } - EventMsg::ApplyPatchApprovalRequest { .. } => { + EventMsg::ApplyPatchApprovalRequest(_) => { // Should we exit? } _ => { diff --git a/codex-rs/mcp-server/src/codex_tool_runner.rs b/codex-rs/mcp-server/src/codex_tool_runner.rs index 2f8a1a34ae..fa03da99f4 100644 --- a/codex-rs/mcp-server/src/codex_tool_runner.rs +++ b/codex-rs/mcp-server/src/codex_tool_runner.rs @@ -85,10 +85,10 @@ pub async fn run_codex_tool_session( let _ = outgoing.send(codex_event_to_notification(&event)).await; match &event.msg { - EventMsg::AgentMessage { message } => { + EventMsg::AgentMessage(codex_core::protocol::AgentMessageEvent { message }) => { last_agent_message = Some(message.clone()); } - EventMsg::ExecApprovalRequest { .. } => { + EventMsg::ExecApprovalRequest(_) => { let result = CallToolResult { content: vec![CallToolResultContent::TextContent(TextContent { r#type: "text".to_string(), @@ -106,7 +106,7 @@ pub async fn run_codex_tool_session( .await; break; } - EventMsg::ApplyPatchApprovalRequest { .. } => { + EventMsg::ApplyPatchApprovalRequest(_) => { let result = CallToolResult { content: vec![CallToolResultContent::TextContent(TextContent { r#type: "text".to_string(), @@ -153,7 +153,7 @@ pub async fn run_codex_tool_session( .await; break; } - EventMsg::SessionConfigured { .. } => { + EventMsg::SessionConfigured(_) => { tracing::error!("unexpected SessionConfigured event"); } _ => {} diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index accb73053c..a3dcdc338a 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -213,11 +213,11 @@ impl ChatWidget<'_> { .add_session_info(&self.config, event); self.request_redraw()?; } - EventMsg::AgentMessage { message } => { + EventMsg::AgentMessage(codex_core::protocol::AgentMessageEvent { message }) => { self.conversation_history.add_agent_message(message); self.request_redraw()?; } - EventMsg::AgentReasoning { text } => { + EventMsg::AgentReasoning(codex_core::protocol::AgentReasoningEvent { text }) => { self.conversation_history.add_agent_reasoning(text); self.request_redraw()?; } @@ -229,15 +229,15 @@ impl ChatWidget<'_> { self.bottom_pane.set_task_running(false)?; self.request_redraw()?; } - EventMsg::Error { message } => { + EventMsg::Error(codex_core::protocol::ErrorEvent { message }) => { self.conversation_history.add_error(message); self.bottom_pane.set_task_running(false)?; } - EventMsg::ExecApprovalRequest { + EventMsg::ExecApprovalRequest(codex_core::protocol::ExecApprovalRequestEvent { command, cwd, reason, - } => { + }) => { let request = ApprovalRequest::Exec { id, command, @@ -246,11 +246,13 @@ impl ChatWidget<'_> { }; self.bottom_pane.push_approval_request(request)?; } - EventMsg::ApplyPatchApprovalRequest { - changes, - reason, - grant_root, - } => { + EventMsg::ApplyPatchApprovalRequest( + codex_core::protocol::ApplyPatchApprovalRequestEvent { + changes, + reason, + grant_root, + }, + ) => { // ------------------------------------------------------------------ // Before we even prompt the user for approval we surface the patch // summary in the main conversation so that the dialog appears in a @@ -276,18 +278,20 @@ impl ChatWidget<'_> { self.bottom_pane.push_approval_request(request)?; self.request_redraw()?; } - EventMsg::ExecCommandBegin { - call_id, command, .. - } => { + EventMsg::ExecCommandBegin(codex_core::protocol::ExecCommandBeginEvent { + call_id, + command, + cwd: _, + }) => { self.conversation_history .add_active_exec_command(call_id, command); self.request_redraw()?; } - EventMsg::PatchApplyBegin { + EventMsg::PatchApplyBegin(codex_core::protocol::PatchApplyBeginEvent { call_id: _, auto_approved, changes, - } => { + }) => { // Even when a patch is auto‑approved we still display the // summary so the user can follow along. self.conversation_history @@ -297,32 +301,31 @@ impl ChatWidget<'_> { } self.request_redraw()?; } - EventMsg::ExecCommandEnd { + EventMsg::ExecCommandEnd(codex_core::protocol::ExecCommandEndEvent { call_id, exit_code, stdout, stderr, - .. - } => { + }) => { self.conversation_history .record_completed_exec_command(call_id, stdout, stderr, exit_code); self.request_redraw()?; } - EventMsg::McpToolCallBegin { + EventMsg::McpToolCallBegin(codex_core::protocol::McpToolCallBeginEvent { call_id, server, tool, arguments, - } => { + }) => { self.conversation_history .add_active_mcp_tool_call(call_id, server, tool, arguments); self.request_redraw()?; } - EventMsg::McpToolCallEnd { + EventMsg::McpToolCallEnd(codex_core::protocol::McpToolCallEndEvent { call_id, success, result, - } => { + }) => { self.conversation_history .record_completed_mcp_tool_call(call_id, success, result); self.request_redraw()?;