diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index 48fa56d687..6d77e52c77 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -44,6 +44,7 @@ use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::ThreadRolledBackEvent; use codex_protocol::protocol::TurnAbortedEvent; use codex_protocol::protocol::TurnCompleteEvent; +use codex_protocol::protocol::TurnOutcome; use codex_protocol::protocol::TurnStartedEvent; use codex_protocol::protocol::UserMessageEvent; use codex_protocol::protocol::ViewImageToolCallEvent; @@ -920,9 +921,19 @@ impl ThreadHistoryBuilder { } fn handle_turn_complete(&mut self, payload: &TurnCompleteEvent) { - let mark_completed = |status: &mut TurnStatus| { - if matches!(*status, TurnStatus::Completed | TurnStatus::InProgress) { - *status = TurnStatus::Completed; + let mark_finished = |turn: &mut PendingTurn| match &payload.outcome { + TurnOutcome::Succeeded { .. } => { + if matches!(turn.status, TurnStatus::Completed | TurnStatus::InProgress) { + turn.status = TurnStatus::Completed; + } + } + TurnOutcome::Failed { error } => { + turn.status = TurnStatus::Failed; + turn.error = Some(V2TurnError { + message: error.message.clone(), + codex_error_info: error.codex_error_info.clone().map(Into::into), + additional_details: None, + }); } }; @@ -932,7 +943,7 @@ impl ThreadHistoryBuilder { .as_mut() .filter(|turn| turn.id == payload.turn_id) { - mark_completed(&mut current_turn.status); + mark_finished(current_turn); self.finish_current_turn(); return; } @@ -942,13 +953,27 @@ impl ThreadHistoryBuilder { .iter_mut() .find(|turn| turn.id == payload.turn_id) { - mark_completed(&mut turn.status); + match &payload.outcome { + TurnOutcome::Succeeded { .. } => { + if matches!(turn.status, TurnStatus::Completed | TurnStatus::InProgress) { + turn.status = TurnStatus::Completed; + } + } + TurnOutcome::Failed { error } => { + turn.status = TurnStatus::Failed; + turn.error = Some(V2TurnError { + message: error.message.clone(), + codex_error_info: error.codex_error_info.clone().map(Into::into), + additional_details: None, + }); + } + } return; } // If the completion event cannot be matched, apply it to the active turn. if let Some(current_turn) = self.current_turn.as_mut() { - mark_completed(&mut current_turn.status); + mark_finished(current_turn); self.finish_current_turn(); } } diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 9a850c0b1e..db317c3e09 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -128,6 +128,7 @@ use codex_protocol::protocol::RealtimeEvent; use codex_protocol::protocol::ReviewDecision; use codex_protocol::protocol::ReviewOutputEvent; use codex_protocol::protocol::TokenCountEvent; +use codex_protocol::protocol::TurnCompleteEvent; use codex_protocol::protocol::TurnDiffEvent; use codex_protocol::request_permissions::PermissionGrantScope as CorePermissionGrantScope; use codex_protocol::request_permissions::RequestPermissionProfile as CoreRequestPermissionProfile; @@ -297,11 +298,22 @@ pub(crate) async fn apply_bespoke_event_handling( EventMsg::TurnComplete(_ev) => { // All per-thread requests are bound to a turn, so abort them. outgoing.abort_pending_server_requests().await; - let turn_failed = thread_state.lock().await.turn_summary.last_error.is_some(); + let turn_failed = thread_state.lock().await.turn_summary.last_error.is_some() + || matches!( + _ev.error(), + Some(error) if error.affects_turn_status() + ); thread_watch_manager .note_turn_completed(&conversation_id.to_string(), turn_failed) .await; - handle_turn_complete(conversation_id, event_turn_id, &outgoing, &thread_state).await; + handle_turn_complete( + conversation_id, + event_turn_id, + &_ev, + &outgoing, + &thread_state, + ) + .await; } EventMsg::SkillsUpdateAvailable => { if let ApiVersion::V2 = api_version { @@ -2100,6 +2112,7 @@ async fn find_and_remove_turn_summary( async fn handle_turn_complete( conversation_id: ThreadId, event_turn_id: String, + event: &TurnCompleteEvent, outgoing: &ThreadScopedOutgoingMessageSender, thread_state: &Arc>, ) { @@ -2107,7 +2120,17 @@ async fn handle_turn_complete( let (status, error) = match turn_summary.last_error { Some(error) => (TurnStatus::Failed, Some(error)), - None => (TurnStatus::Completed, None), + None => match event.error() { + Some(error) if error.affects_turn_status() => ( + TurnStatus::Failed, + Some(TurnError { + message: error.message.clone(), + codex_error_info: error.codex_error_info.clone().map(Into::into), + additional_details: None, + }), + ), + _ => (TurnStatus::Completed, None), + }, }; emit_turn_completed_with_status(conversation_id, event_turn_id, status, error, outgoing).await; diff --git a/codex-rs/core/src/agent/status.rs b/codex-rs/core/src/agent/status.rs index c343e19503..2c1063d70a 100644 --- a/codex-rs/core/src/agent/status.rs +++ b/codex-rs/core/src/agent/status.rs @@ -1,19 +1,25 @@ use codex_protocol::protocol::AgentStatus; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::TurnOutcome; /// Derive the next agent status from a single emitted event. /// Returns `None` when the event does not affect status tracking. pub(crate) fn agent_status_from_event(msg: &EventMsg) -> Option { match msg { EventMsg::TurnStarted(_) => Some(AgentStatus::Running), - EventMsg::TurnComplete(ev) => Some(AgentStatus::Completed(ev.last_agent_message.clone())), + EventMsg::TurnComplete(ev) => Some(match &ev.outcome { + TurnOutcome::Succeeded { last_agent_message } => { + AgentStatus::Completed(last_agent_message.clone()) + } + TurnOutcome::Failed { error } => AgentStatus::Errored(error.message.clone()), + }), EventMsg::TurnAborted(ev) => match ev.reason { codex_protocol::protocol::TurnAbortReason::Interrupted => { Some(AgentStatus::Interrupted) } _ => Some(AgentStatus::Errored(format!("{:?}", ev.reason))), }, - EventMsg::Error(ev) => Some(AgentStatus::Errored(ev.message.clone())), + EventMsg::Error(_) => None, EventMsg::ShutdownComplete => Some(AgentStatus::Shutdown), _ => None, } diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 17993a9743..45d5bc0d4d 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -18,6 +18,7 @@ use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::Op; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::SubAgentSource; +use codex_protocol::protocol::TurnOutcome; use codex_protocol::user_input::UserInput; use serde_json::Value; use tokio::sync::Mutex; @@ -578,8 +579,6 @@ async fn wait_for_guardian_review( ) -> (GuardianReviewSessionOutcome, bool) { let timeout = tokio::time::sleep_until(deadline); tokio::pin!(timeout); - let mut last_error_message: Option = None; - loop { tokio::select! { _ = &mut timeout => { @@ -600,21 +599,16 @@ async fn wait_for_guardian_review( match event { Ok(event) => match event.msg { EventMsg::TurnComplete(turn_complete) => { - if turn_complete.last_agent_message.is_none() - && let Some(error_message) = last_error_message - { - return ( - GuardianReviewSessionOutcome::Completed(Err(anyhow!(error_message))), + return match turn_complete.outcome { + TurnOutcome::Succeeded { last_agent_message } => ( + GuardianReviewSessionOutcome::Completed(Ok(last_agent_message)), true, - ); - } - return ( - GuardianReviewSessionOutcome::Completed(Ok(turn_complete.last_agent_message)), - true, - ); - } - EventMsg::Error(error) => { - last_error_message = Some(error.message); + ), + TurnOutcome::Failed { error } => ( + GuardianReviewSessionOutcome::Completed(Err(anyhow!(error.message))), + true, + ), + }; } EventMsg::TurnAborted(_) => { return (GuardianReviewSessionOutcome::Aborted, true); diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 4108c6255d..e32d9c0d2e 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -438,13 +438,13 @@ impl Session { ); } let event = match completion { - TaskCompletion::Completed(last_agent_message) => { - EventMsg::TurnComplete(TurnCompleteEvent { - turn_id: turn_context.sub_id.clone(), - last_agent_message, - }) - } - TaskCompletion::Failed(error) => EventMsg::Error(error), + TaskCompletion::Completed(last_agent_message) => EventMsg::TurnComplete( + TurnCompleteEvent::succeeded(turn_context.sub_id.clone(), last_agent_message), + ), + TaskCompletion::Failed(error) => EventMsg::TurnComplete(TurnCompleteEvent::failed( + turn_context.sub_id.clone(), + error, + )), }; self.send_event(turn_context.as_ref(), event).await; } diff --git a/codex-rs/core/src/tasks/review.rs b/codex-rs/core/src/tasks/review.rs index 8a72391731..bb357cfd02 100644 --- a/codex-rs/core/src/tasks/review.rs +++ b/codex-rs/core/src/tasks/review.rs @@ -15,6 +15,7 @@ use codex_protocol::protocol::ExitedReviewModeEvent; use codex_protocol::protocol::ItemCompletedEvent; use codex_protocol::protocol::ReviewOutputEvent; use codex_protocol::protocol::SubAgentSource; +use codex_protocol::protocol::TurnOutcome; use codex_utils_template::Template; use tokio_util::sync::CancellationToken; @@ -167,12 +168,18 @@ async fn process_review_events( | EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { .. }) | EventMsg::AgentMessageContentDelta(AgentMessageContentDeltaEvent { .. }) => {} EventMsg::TurnComplete(task_complete) => { - // Parse review output from the last agent message (if present). - let out = task_complete - .last_agent_message - .as_deref() - .map(parse_review_output_event); - return out; + return match task_complete.outcome { + TurnOutcome::Succeeded { last_agent_message } => { + last_agent_message.as_deref().map(parse_review_output_event) + } + TurnOutcome::Failed { error } => { + session + .clone_session() + .send_event(ctx.as_ref(), EventMsg::Error(error)) + .await; + None + } + }; } EventMsg::TurnAborted(_) => { // Cancellation or abort: consumer will finalize with None. diff --git a/codex-rs/mcp-server/src/codex_tool_runner.rs b/codex-rs/mcp-server/src/codex_tool_runner.rs index 64d553bbff..5fc6e55705 100644 --- a/codex-rs/mcp-server/src/codex_tool_runner.rs +++ b/codex-rs/mcp-server/src/codex_tool_runner.rs @@ -22,6 +22,7 @@ use codex_protocol::protocol::ExecApprovalRequestEvent; use codex_protocol::protocol::Op; use codex_protocol::protocol::Submission; use codex_protocol::protocol::TurnCompleteEvent; +use codex_protocol::protocol::TurnOutcome; use codex_protocol::user_input::UserInput; use rmcp::model::CallToolResult; use rmcp::model::Content; @@ -291,16 +292,15 @@ async fn run_codex_tool_session_inner( .await; continue; } - EventMsg::TurnComplete(TurnCompleteEvent { - last_agent_message, .. - }) => { - let text = match last_agent_message { - Some(msg) => msg, - None => "".to_string(), + EventMsg::TurnComplete(TurnCompleteEvent { outcome, .. }) => { + let (text, is_error) = match outcome { + TurnOutcome::Succeeded { last_agent_message } => { + (last_agent_message.unwrap_or_default(), None) + } + TurnOutcome::Failed { error } => (error.message, Some(true)), }; - let result = create_call_tool_result_with_thread_id( - thread_id, text, /*is_error*/ None, - ); + let result = + create_call_tool_result_with_thread_id(thread_id, text, is_error); outgoing.send_response(request_id.clone(), result).await; // unregister the id so we don't keep it in the map running_requests_id_to_codex_uuid diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index aa9f319ffd..01582acf80 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -1858,10 +1858,48 @@ pub struct ModelRerouteEvent { #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] pub struct ContextCompactedEvent; +#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] +#[serde(untagged)] +pub enum TurnOutcome { + Succeeded { last_agent_message: Option }, + Failed { error: ErrorEvent }, +} + #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] pub struct TurnCompleteEvent { pub turn_id: String, - pub last_agent_message: Option, + #[serde(flatten)] + pub outcome: TurnOutcome, +} + +impl TurnCompleteEvent { + pub fn succeeded(turn_id: String, last_agent_message: Option) -> Self { + Self { + turn_id, + outcome: TurnOutcome::Succeeded { last_agent_message }, + } + } + + pub fn failed(turn_id: String, error: ErrorEvent) -> Self { + Self { + turn_id, + outcome: TurnOutcome::Failed { error }, + } + } + + pub fn last_agent_message(&self) -> Option<&str> { + match &self.outcome { + TurnOutcome::Succeeded { last_agent_message } => last_agent_message.as_deref(), + TurnOutcome::Failed { .. } => None, + } + } + + pub fn error(&self) -> Option<&ErrorEvent> { + match &self.outcome { + TurnOutcome::Succeeded { .. } => None, + TurnOutcome::Failed { error } => Some(error), + } + } } #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] diff --git a/codex-rs/tui/src/app/app_server_adapter.rs b/codex-rs/tui/src/app/app_server_adapter.rs index 12b1c669b6..1f9be3ef0b 100644 --- a/codex-rs/tui/src/app/app_server_adapter.rs +++ b/codex-rs/tui/src/app/app_server_adapter.rs @@ -731,17 +731,17 @@ fn turn_snapshot_events( /// /// - `Completed` → `TurnComplete` /// - `Interrupted` → `TurnAborted { reason: Interrupted }` -/// - `Failed` → `Error` (if present) then `TurnComplete` +/// - `Failed` → `TurnComplete` with failed outcome /// - `InProgress` → no events (the turn is still running) #[cfg(test)] fn append_terminal_turn_events(events: &mut Vec, turn: &Turn, include_failed_error: bool) { match turn.status { TurnStatus::Completed => events.push(Event { id: String::new(), - msg: EventMsg::TurnComplete(TurnCompleteEvent { - turn_id: turn.id.clone(), - last_agent_message: None, - }), + msg: EventMsg::TurnComplete(TurnCompleteEvent::succeeded( + turn.id.clone(), + /*last_agent_message*/ None, + )), }), TurnStatus::Interrupted => events.push(Event { id: String::new(), @@ -751,24 +751,25 @@ fn append_terminal_turn_events(events: &mut Vec, turn: &Turn, include_fai }), }), TurnStatus::Failed => { - if include_failed_error && let Some(error) = &turn.error { - events.push(Event { - id: String::new(), - msg: EventMsg::Error(ErrorEvent { - message: error.message.clone(), - codex_error_info: error - .codex_error_info - .clone() - .and_then(app_server_codex_error_info_to_core), - }), - }); - } + let _ = include_failed_error; events.push(Event { id: String::new(), - msg: EventMsg::TurnComplete(TurnCompleteEvent { - turn_id: turn.id.clone(), - last_agent_message: None, - }), + msg: EventMsg::TurnComplete(TurnCompleteEvent::failed( + turn.id.clone(), + ErrorEvent { + message: turn + .error + .as_ref() + .map(|error| error.message.clone()) + .unwrap_or_default(), + codex_error_info: turn.error.as_ref().and_then(|error| { + error + .codex_error_info + .clone() + .and_then(app_server_codex_error_info_to_core) + }), + }, + )), }); } TurnStatus::InProgress => { diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 1267e6b732..44aa83fa76 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -193,10 +193,10 @@ use codex_protocol::protocol::TerminalInteractionEvent; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TokenUsageInfo; use codex_protocol::protocol::TurnAbortReason; -#[cfg(test)] use codex_protocol::protocol::TurnCompleteEvent; #[cfg(test)] use codex_protocol::protocol::TurnDiffEvent; +use codex_protocol::protocol::TurnOutcome; #[cfg(test)] use codex_protocol::protocol::UndoCompletedEvent; #[cfg(test)] @@ -6927,11 +6927,14 @@ impl ChatWidget { self.on_task_started(); } } - EventMsg::TurnComplete(TurnCompleteEvent { - last_agent_message, .. - }) => { - self.on_task_complete(last_agent_message, from_replay); - } + EventMsg::TurnComplete(TurnCompleteEvent { outcome, .. }) => match outcome { + TurnOutcome::Succeeded { last_agent_message } => { + self.on_task_complete(last_agent_message, from_replay); + } + TurnOutcome::Failed { error } => { + self.on_error(error.message); + } + }, EventMsg::TokenCount(ev) => { self.set_token_info(ev.info); self.on_rate_limit_snapshot(ev.rate_limits);