From 1f3c7a5588e63ec8c701bba2c999a08588784993 Mon Sep 17 00:00:00 2001 From: easong-openai Date: Tue, 19 Aug 2025 20:08:38 -0700 Subject: [PATCH] parseable event --- codex-rs/core/src/codex.rs | 39 +++++++++++-------- .../src/event_processor_with_human_output.rs | 16 ++++++++ codex-rs/mcp-server/src/codex_tool_runner.rs | 1 + codex-rs/protocol/src/protocol.rs | 11 ++++++ codex-rs/tui/src/chatwidget.rs | 12 ++++++ codex-rs/tui/src/history_cell.rs | 12 ------ 6 files changed, 62 insertions(+), 29 deletions(-) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 397246a7de..f8e0bd96cf 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -94,6 +94,7 @@ use crate::protocol::PatchApplyEndEvent; use crate::protocol::ReviewDecision; use crate::protocol::SandboxPolicy; use crate::protocol::SessionConfiguredEvent; +use crate::protocol::StreamRetryEvent; use crate::protocol::Submission; use crate::protocol::TaskCompleteEvent; use crate::protocol::TurnDiffEvent; @@ -1501,16 +1502,17 @@ async fn run_turn( "stream disconnected - retrying turn ({retries}/{max_retries} in {delay:?})...", ); - // Surface retry information to any UI/front‑end so the - // user understands what is happening instead of staring - // at a seemingly frozen screen. - sess.notify_background_event( - &sub_id, - format!( - "stream error: {e}; retrying {retries}/{max_retries} in {delay:?}…" - ), - ) - .await; + // Surface retry information to any UI/front‑end in a structured way. + let event = Event { + id: sub_id.to_string(), + msg: EventMsg::StreamRetry(StreamRetryEvent { + attempt: retries as u32, + max_attempts: max_retries as u32, + delay, + cause: e.to_string(), + }), + }; + let _ = sess.tx_event.send(event).await; tokio::time::sleep(delay).await; } else { @@ -1739,13 +1741,16 @@ async fn run_compact_task( if retries < max_retries { retries += 1; let delay = backoff(retries); - sess.notify_background_event( - &sub_id, - format!( - "stream error: {e}; retrying {retries}/{max_retries} in {delay:?}…" - ), - ) - .await; + let event = Event { + id: sub_id.clone(), + msg: EventMsg::StreamRetry(StreamRetryEvent { + attempt: retries as u32, + max_attempts: max_retries as u32, + delay, + cause: e.to_string(), + }), + }; + sess.send_event(event).await; tokio::time::sleep(delay).await; continue; } else { 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 98d7a1fb6a..689ff9f7eb 100644 --- a/codex-rs/exec/src/event_processor_with_human_output.rs +++ b/codex-rs/exec/src/event_processor_with_human_output.rs @@ -20,6 +20,7 @@ use codex_core::protocol::McpToolCallEndEvent; use codex_core::protocol::PatchApplyBeginEvent; use codex_core::protocol::PatchApplyEndEvent; use codex_core::protocol::SessionConfiguredEvent; +use codex_core::protocol::StreamRetryEvent; use codex_core::protocol::TaskCompleteEvent; use codex_core::protocol::TurnAbortReason; use codex_core::protocol::TurnDiffEvent; @@ -174,6 +175,21 @@ impl EventProcessor for EventProcessorWithHumanOutput { EventMsg::BackgroundEvent(BackgroundEventEvent { message }) => { ts_println!(self, "{}", message.style(self.dimmed)); } + EventMsg::StreamRetry(StreamRetryEvent { + attempt, + max_attempts, + delay, + cause, + }) => { + ts_println!( + self, + "{}", + format!( + "stream error: {cause}; retrying {attempt}/{max_attempts} in {delay:?}…" + ) + .style(self.dimmed) + ); + } EventMsg::TaskStarted => { // Ignore. } diff --git a/codex-rs/mcp-server/src/codex_tool_runner.rs b/codex-rs/mcp-server/src/codex_tool_runner.rs index c0d14ecedc..c0808d38d5 100644 --- a/codex-rs/mcp-server/src/codex_tool_runner.rs +++ b/codex-rs/mcp-server/src/codex_tool_runner.rs @@ -267,6 +267,7 @@ async fn run_codex_tool_session_inner( | EventMsg::ExecCommandOutputDelta(_) | EventMsg::ExecCommandEnd(_) | EventMsg::BackgroundEvent(_) + | EventMsg::StreamRetry(_) | EventMsg::PatchApplyBegin(_) | EventMsg::PatchApplyEnd(_) | EventMsg::TurnDiff(_) diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 23ef4668f1..115591dac5 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -439,6 +439,9 @@ pub enum EventMsg { ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent), + /// Notification that a streaming response failed transiently and will be retried. + StreamRetry(StreamRetryEvent), + BackgroundEvent(BackgroundEventEvent), /// Notification that the agent is about to apply a code patch. Mirrors @@ -681,6 +684,14 @@ pub struct ApplyPatchApprovalRequestEvent { pub grant_root: Option, } +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct StreamRetryEvent { + pub attempt: u32, + pub max_attempts: u32, + pub delay: Duration, + pub cause: String, +} + #[derive(Debug, Clone, Deserialize, Serialize)] pub struct BackgroundEventEvent { pub message: String, diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 270c2f1176..543fc6e41f 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -22,6 +22,7 @@ use codex_core::protocol::McpToolCallBeginEvent; use codex_core::protocol::McpToolCallEndEvent; use codex_core::protocol::Op; use codex_core::protocol::PatchApplyBeginEvent; +use codex_core::protocol::StreamRetryEvent; use codex_core::protocol::TaskCompleteEvent; use codex_core::protocol::TokenUsage; use codex_core::protocol::TurnDiffEvent; @@ -653,6 +654,17 @@ impl ChatWidget<'_> { EventMsg::BackgroundEvent(BackgroundEventEvent { message }) => { self.on_background_event(message) } + EventMsg::StreamRetry(StreamRetryEvent { + attempt, + max_attempts, + delay, + cause, + }) => { + let text = format!( + "stream error: {cause}; retrying {attempt}/{max_attempts} in {delay:?}…" + ); + self.on_background_event(text) + } } // Coalesce redraws: issue at most one after handling the event if self.needs_redraw { diff --git a/codex-rs/tui/src/history_cell.rs b/codex-rs/tui/src/history_cell.rs index 1ebf85b326..555226dddd 100644 --- a/codex-rs/tui/src/history_cell.rs +++ b/codex-rs/tui/src/history_cell.rs @@ -639,18 +639,6 @@ pub(crate) fn new_error_event(message: String) -> PlainHistoryCell { PlainHistoryCell { lines } } -pub(crate) fn new_background_event(message: String) -> PlainHistoryCell { - let lines: Vec> = vec![ - Line::from(vec![ - "ℹ".magenta(), - " ".into(), - Span::styled(message, Style::default().add_modifier(Modifier::DIM)), - ]), - Line::from(""), - ]; - PlainHistoryCell { lines } -} - /// Render a user‑friendly plan update styled like a checkbox todo list. pub(crate) fn new_plan_update(update: UpdatePlanArgs) -> PlainHistoryCell { let UpdatePlanArgs { explanation, plan } = update;