diff --git a/CHANGELOG.md b/CHANGELOG.md index 8030d0bc31..268d899b17 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,7 @@ ## [Unreleased] -- (none) +- Agents: preserve terminal subagent stream failures through trailing turn completion so child status and parent notifications report the real error instead of an empty success. ## [0.6.116] - 2026-06-04 diff --git a/code-rs/core/src/agent/control_tests.rs b/code-rs/core/src/agent/control_tests.rs index b95aad4489..c784d0297f 100644 --- a/code-rs/core/src/agent/control_tests.rs +++ b/code-rs/core/src/agent/control_tests.rs @@ -1,6 +1,7 @@ use super::*; use crate::CodexThread; use crate::StateDbHandle; +use crate::StartThreadOptions; use crate::ThreadManager; use crate::agent::agent_status_from_event; use crate::config::AgentRoleConfig; @@ -17,8 +18,10 @@ use codex_protocol::config_types::ModeKind; use codex_protocol::models::ContentItem; use codex_protocol::models::MessagePhase; use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; @@ -1407,6 +1410,145 @@ async fn multi_agent_v2_completion_queues_message_for_direct_parent() { )); } +#[tokio::test] +async fn multi_agent_v2_terminal_error_queues_message_for_direct_parent() { + for (case, codex_error_info) in [ + ( + "typed", + Some(CodexErrorInfo::ResponseTooManyFailedAttempts { + http_status_code: None, + }), + ), + ("untyped", None), + ] { + let harness = AgentControlHarness::new().await; + let (parent_thread_id, _parent_thread) = harness.start_thread().await; + let worker_path = AgentPath::root().join("worker").expect("worker path"); + let mut child_config = harness.config.clone(); + let _ = child_config.features.enable(Feature::MultiAgentV2); + let child = harness + .manager + .start_thread_with_options(StartThreadOptions { + config: child_config, + initial_history: InitialHistory::New, + session_source: Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_path: Some(worker_path.clone()), + agent_nickname: None, + agent_role: Some("explorer".to_string()), + })), + thread_source: None, + dynamic_tools: Vec::new(), + persist_extended_history: false, + metrics_service_name: None, + parent_trace: None, + environments: Vec::new(), + }) + .await + .expect("child thread should start"); + let turn = child.thread.codex.session.new_default_turn().await; + let error = format!("{case} stream disconnected before completion"); + child + .thread + .codex + .session + .send_event( + turn.as_ref(), + EventMsg::Error(ErrorEvent { + message: error.clone(), + codex_error_info, + }), + ) + .await; + child + .thread + .codex + .session + .send_event( + turn.as_ref(), + EventMsg::TurnComplete(TurnCompleteEvent { + turn_id: turn.sub_id.clone(), + last_agent_message: None, + completed_at: None, + duration_ms: None, + time_to_first_token_ms: None, + }), + ) + .await; + + assert_eq!( + child.thread.agent_status().await, + AgentStatus::Errored(error.clone()), + "{case} terminal error should remain the final child status" + ); + + let expected_message = crate::session_prefix::format_subagent_notification_message( + worker_path.as_str(), + &AgentStatus::Errored(error), + ); + let expected = ( + parent_thread_id, + Op::InterAgentCommunication { + communication: InterAgentCommunication::new( + worker_path, + AgentPath::root(), + Vec::new(), + expected_message, + /*trigger_turn*/ false, + ), + }, + ); + + timeout(Duration::from_secs(5), async { + loop { + if harness + .manager + .captured_ops() + .into_iter() + .any(|entry| entry == expected) + { + break; + } + sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap_or_else(|_| panic!("{case} terminal error should be queued for the direct parent")); + + child + .thread + .codex + .session + .send_event( + turn.as_ref(), + EventMsg::TurnAborted(TurnAbortedEvent { + turn_id: Some(turn.sub_id.clone()), + reason: TurnAbortReason::Interrupted, + completed_at: None, + duration_ms: None, + }), + ) + .await; + + assert_eq!( + child.thread.agent_status().await, + AgentStatus::Errored(format!("{case} stream disconnected before completion")), + "{case} terminal error should survive a trailing abort" + ); + assert_eq!( + harness + .manager + .captured_ops() + .into_iter() + .filter(|entry| entry == &expected) + .count(), + 1, + "{case} terminal error should notify the direct parent exactly once" + ); + } +} + #[tokio::test] async fn completion_watcher_notifies_parent_when_child_is_missing() { let harness = AgentControlHarness::new().await; diff --git a/code-rs/core/src/session/mod.rs b/code-rs/core/src/session/mod.rs index ee97f62e19..bbbae9ea0e 100644 --- a/code-rs/core/src/session/mod.rs +++ b/code-rs/core/src/session/mod.rs @@ -1517,6 +1517,15 @@ impl Session { /// Persist the event to rollout and send it to clients. pub(crate) async fn send_event(&self, turn_context: &TurnContext, msg: EventMsg) { let legacy_source = msg.clone(); + if let EventMsg::Error(error) = &legacy_source + && error.affects_turn_status() + { + turn_context + .terminal_error + .lock() + .await + .replace(error.message.clone()); + } self.services .rollout_thread_trace .record_codex_turn_event(&turn_context.sub_id, &legacy_source); @@ -1527,9 +1536,27 @@ impl Session { id: turn_context.sub_id.clone(), msg, }; - self.send_event_raw(event).await; - self.maybe_notify_parent_of_terminal_turn(turn_context, &legacy_source) + let status_override = if matches!( + legacy_source, + EventMsg::TurnComplete(_) | EventMsg::TurnAborted(_) + ) { + turn_context + .terminal_error + .lock() + .await + .clone() + .map(AgentStatus::Errored) + } else { + None + }; + self.send_event_raw_with_status(event, status_override.clone()) .await; + self.maybe_notify_parent_of_terminal_turn( + turn_context, + &legacy_source, + status_override, + ) + .await; self.maybe_mirror_event_text_to_realtime(&legacy_source) .await; self.maybe_clear_realtime_handoff_for_event(&legacy_source) @@ -1550,6 +1577,7 @@ impl Session { &self, turn_context: &TurnContext, msg: &EventMsg, + status_override: Option, ) { if !self.enabled(Feature::MultiAgentV2) { return; @@ -1568,20 +1596,42 @@ impl Session { return; }; - let Some(status) = agent_status_from_event(msg) else { - return; + let uses_terminal_error = status_override.is_some(); + let status = match status_override { + Some(status) => status, + None => { + let Some(status) = agent_status_from_event(msg) else { + return; + }; + status + } }; if !is_final(&status) { return; } - self.forward_child_completion_to_parent( - turn_context, - *parent_thread_id, - child_agent_path, - status, - ) - .await; + if uses_terminal_error { + turn_context + .terminal_error_parent_notification + .get_or_init(|| async { + self.forward_child_completion_to_parent( + turn_context, + *parent_thread_id, + child_agent_path, + status, + ) + .await; + }) + .await; + } else { + self.forward_child_completion_to_parent( + turn_context, + *parent_thread_id, + child_agent_path, + status, + ) + .await; + } } /// Sends the standard completion envelope from a spawned MultiAgentV2 child to its parent. @@ -1664,18 +1714,37 @@ impl Session { } pub(crate) async fn send_event_raw(&self, event: Event) { + self.send_event_raw_with_status(event, /*status_override*/ None) + .await; + } + + async fn send_event_raw_with_status( + &self, + event: Event, + status_override: Option, + ) { // Persist the event into rollout storage (the store filters as needed). let rollout_items = vec![RolloutItem::EventMsg(event.msg.clone())]; self.persist_rollout_items(&rollout_items).await; self.services .rollout_thread_trace .record_protocol_event(&event.msg); - self.deliver_event_raw(event).await; + self.deliver_event_raw_with_status(event, status_override) + .await; } async fn deliver_event_raw(&self, event: Event) { + self.deliver_event_raw_with_status(event, /*status_override*/ None) + .await; + } + + async fn deliver_event_raw_with_status( + &self, + event: Event, + status_override: Option, + ) { // Record the last known agent status. - if let Some(status) = agent_status_from_event(&event.msg) { + if let Some(status) = status_override.or_else(|| agent_status_from_event(&event.msg)) { self.agent_status.send_replace(status); } if let Err(e) = self.tx_event.send(event).await { diff --git a/code-rs/core/src/session/review.rs b/code-rs/core/src/session/review.rs index 7d4b1b736a..28678ad7b7 100644 --- a/code-rs/core/src/session/review.rs +++ b/code-rs/core/src/session/review.rs @@ -1,6 +1,7 @@ use super::turn_context::image_generation_tool_auth_allowed; use super::*; use std::sync::atomic::AtomicBool; +use tokio::sync::OnceCell; /// Spawn a review thread using the given prompt. pub(super) async fn spawn_review_thread( @@ -152,6 +153,8 @@ pub(super) async fn spawn_review_thread( turn_metadata_state, turn_skills: TurnSkillsContext::new(parent_turn_context.turn_skills.outcome.clone()), turn_timing_state: Arc::new(TurnTimingState::default()), + terminal_error: Arc::new(Mutex::new(None)), + terminal_error_parent_notification: Arc::new(OnceCell::new()), server_model_warning_emitted: AtomicBool::new(false), model_verification_emitted: AtomicBool::new(false), }; diff --git a/code-rs/core/src/session/turn.rs b/code-rs/core/src/session/turn.rs index 1723904cff..41fc2c7461 100644 --- a/code-rs/core/src/session/turn.rs +++ b/code-rs/core/src/session/turn.rs @@ -621,7 +621,7 @@ pub(crate) async fn run_turn( &turn_context, EventMsg::Error(ErrorEvent { message, - codex_error_info: None, + codex_error_info: Some(CodexErrorInfo::Other), }), ) .await; diff --git a/code-rs/core/src/session/turn_context.rs b/code-rs/core/src/session/turn_context.rs index d4fe30063f..dda13de8e9 100644 --- a/code-rs/core/src/session/turn_context.rs +++ b/code-rs/core/src/session/turn_context.rs @@ -13,6 +13,7 @@ use codex_sandboxing::policy_transforms::effective_file_system_sandbox_policy; use codex_sandboxing::policy_transforms::effective_network_sandbox_policy; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; +use tokio::sync::OnceCell; pub(super) fn image_generation_tool_auth_allowed(auth_manager: Option<&AuthManager>) -> bool { auth_manager.is_some_and(AuthManager::current_auth_uses_codex_backend) @@ -95,6 +96,8 @@ pub(crate) struct TurnContext { pub(crate) turn_metadata_state: Arc, pub(crate) turn_skills: TurnSkillsContext, pub(crate) turn_timing_state: Arc, + pub(crate) terminal_error: Arc>>, + pub(crate) terminal_error_parent_notification: Arc>, pub(crate) server_model_warning_emitted: AtomicBool, pub(crate) model_verification_emitted: AtomicBool, } @@ -278,6 +281,10 @@ impl TurnContext { turn_metadata_state: self.turn_metadata_state.clone(), turn_skills: self.turn_skills.clone(), turn_timing_state: Arc::clone(&self.turn_timing_state), + terminal_error: Arc::clone(&self.terminal_error), + terminal_error_parent_notification: Arc::clone( + &self.terminal_error_parent_notification, + ), server_model_warning_emitted: AtomicBool::new( self.server_model_warning_emitted.load(Ordering::Relaxed), ), @@ -573,6 +580,8 @@ impl Session { turn_metadata_state, turn_skills: TurnSkillsContext::new(skills_outcome), turn_timing_state: Arc::new(TurnTimingState::default()), + terminal_error: Arc::new(Mutex::new(None)), + terminal_error_parent_notification: Arc::new(OnceCell::new()), server_model_warning_emitted: AtomicBool::new(false), model_verification_emitted: AtomicBool::new(false), } diff --git a/code-rs/core/tests/suite/subagent_notifications.rs b/code-rs/core/tests/suite/subagent_notifications.rs index 3a0c37acc7..310933850d 100644 --- a/code-rs/core/tests/suite/subagent_notifications.rs +++ b/code-rs/core/tests/suite/subagent_notifications.rs @@ -2,6 +2,8 @@ use anyhow::Result; use codex_core::ThreadConfigSnapshot; use codex_core::config::AgentRoleConfig; use codex_features::Feature; +use codex_protocol::protocol::AgentStatus; +use codex_protocol::protocol::EventMsg; use codex_protocol::ThreadId; use codex_protocol::openai_models::ReasoningEffort; use core_test_support::responses::ResponsesRequest; @@ -24,6 +26,7 @@ use std::path::Path; use std::time::Duration; use tokio::time::Instant; use tokio::time::sleep; +use tokio::time::timeout; use wiremock::MockServer; const SPAWN_CALL_ID: &str = "spawn-call-1"; @@ -38,6 +41,12 @@ const REQUESTED_REASONING_EFFORT: ReasoningEffort = ReasoningEffort::Low; const ROLE_MODEL: &str = "gpt-5.4"; const ROLE_REASONING_EFFORT: ReasoningEffort = ReasoningEffort::High; +#[derive(Clone, Copy)] +enum ChildCompletionScenario { + Completed, + TerminalError, +} + fn body_contains(req: &wiremock::Request, text: &str) -> bool { let is_zstd = req .headers @@ -154,6 +163,7 @@ async fn setup_turn_one_with_spawned_child( }), child_response_delay, /*wait_for_parent_notification*/ true, + ChildCompletionScenario::Completed, |builder| builder, ) .await @@ -164,6 +174,7 @@ async fn setup_turn_one_with_custom_spawned_child( spawn_args: serde_json::Value, child_response_delay: Option, wait_for_parent_notification: bool, + child_completion_scenario: ChildCompletionScenario, configure_test: impl FnOnce( core_test_support::test_codex::TestCodexBuilder, ) -> core_test_support::test_codex::TestCodexBuilder, @@ -181,11 +192,17 @@ async fn setup_turn_one_with_custom_spawned_child( ) .await; - let child_sse = sse(vec![ - ev_response_created("resp-child-1"), - ev_assistant_message("msg-child-1", "child done"), - ev_completed("resp-child-1"), - ]); + let child_events = match child_completion_scenario { + ChildCompletionScenario::Completed => vec![ + ev_response_created("resp-child-1"), + ev_assistant_message("msg-child-1", "child done"), + ev_completed("resp-child-1"), + ], + ChildCompletionScenario::TerminalError => { + vec![ev_response_created("resp-child-1")] + } + }; + let child_sse = sse(child_events); let child_request_log = if let Some(delay) = child_response_delay { mount_response_once_match( server, @@ -267,6 +284,7 @@ async fn spawn_child_and_capture_snapshot( spawn_args, /*child_response_delay*/ None, /*wait_for_parent_notification*/ false, + ChildCompletionScenario::Completed, configure_test, ) .await?; @@ -305,6 +323,53 @@ async fn subagent_notification_is_included_without_wait() -> Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn multi_agent_v2_terminal_stream_error_preserves_errored_status() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let (test, spawned_id) = setup_turn_one_with_custom_spawned_child( + &server, + json!({ + "message": CHILD_PROMPT, + "task_name": "worker", + }), + /*child_response_delay*/ None, + /*wait_for_parent_notification*/ false, + ChildCompletionScenario::TerminalError, + |builder| { + builder.with_config(|config| { + config + .features + .enable(Feature::MultiAgentV2) + .expect("test config should allow feature update"); + config.model_provider.request_max_retries = Some(0); + config.model_provider.stream_max_retries = Some(0); + config.model_provider.supports_websockets = false; + }) + }, + ) + .await?; + + let child_thread_id = ThreadId::from_string(&spawned_id)?; + let child_thread = test.thread_manager.get_thread(child_thread_id).await?; + loop { + let event = timeout(Duration::from_secs(6), child_thread.next_event()).await??; + if matches!(event.msg, EventMsg::TurnComplete(_)) { + break; + } + } + + let expected_error = + "stream disconnected before completion: stream closed before response.completed"; + assert_eq!( + child_thread.agent_status().await, + AgentStatus::Errored(expected_error.to_string()) + ); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn spawned_child_receives_forked_parent_context() -> Result<()> { skip_if_no_network!(Ok(()));