fix(core): preserve terminal subagent errors

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
 

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;

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<AgentStatus>,
     ) {
         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<AgentStatus>,
+    ) {
         // 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<AgentStatus>,
+    ) {
         // 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 {

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),
     };

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;

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<TurnMetadataState>,
     pub(crate) turn_skills: TurnSkillsContext,
     pub(crate) turn_timing_state: Arc<TurnTimingState>,
+    pub(crate) terminal_error: Arc<Mutex<Option<String>>>,
+    pub(crate) terminal_error_parent_notification: Arc<OnceCell<()>>,
     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),
         }