From 1b24ba912ac4c56ef936364deb1c3e294b0ef9fa Mon Sep 17 00:00:00 2001 From: jif Date: Tue, 16 Jun 2026 13:34:54 +0100 Subject: [PATCH] core: surface terminal subagent errors to parent agents (#28375) ## Why When a subagent exhausts its retries, it emits an `Error`, but the generic task lifecycle then emits `TurnComplete(None)`. That completion used to overwrite the subagent's `Errored` status with `Completed(None)`, so the parent received an empty completion notification. This made a failed child look indistinguishable from a child that completed without an answer. In unattended or long-running multi-agent work, the root could silently continue without knowing that delegated work failed or how to restart it. ## Behavior Before, a terminal stream failure was reduced to an empty completion: ```text {"agent_path":"/root/worker","status":{"completed":null}} ``` Now the parent receives the actual terminal error, bounded to 1,000 tokens, together with an actionable recovery hint: ```text { "agent_path": "/root/worker", "status": { "errored": "stream disconnected before completion: stream closed before response.completed" }, "next_action": "This agent's turn failed. If you still need this agent, use `followup_task` to give it another task." } ``` The notification remains queue-only: it does not wake the root or replay the failed request. The root sees it at the next sampling boundary and can use `followup_task` to start a new turn for that agent. ## What changed - Added terminal-error precedence to the [agent status reducer](https://github.com/openai/codex/blob/e95fcfe2bb6a02f1a75650afa20048859f556511/codex-rs/core/src/agent/status.rs#L23-L34), so a closing `TurnComplete` cannot erase an immediately preceding `Errored` status. - Made MultiAgentV2 completion forwarding use the retained session status instead of re-deriving `Completed(None)` from the final event. - Extended the [subagent notification fragment](https://github.com/openai/codex/blob/e95fcfe2bb6a02f1a75650afa20048859f556511/codex-rs/core/src/context/subagent_notification.rs#L6-L60) with a `next_action` for terminal errors and a hard cap on model-visible error text. - Kept successful completions and interrupted turns unchanged. ## Verification - Added a status-reducer test proving that `Errored` survives the trailing `TurnComplete`. - Added an integration test that exhausts a subagent's stream retries and verifies the exact `agent_message` delivered to the parent, including the error and `followup_task` guidance. - Re-ran the existing successful-completion and interrupted-turn notification tests. --- codex-rs/core/src/hook_runtime.rs | 3 +- codex-rs/core/src/session/mod.rs | 26 +++++++++- codex-rs/core/src/session/review.rs | 1 + codex-rs/core/src/session/turn_context.rs | 3 ++ codex-rs/core/src/session_prefix.rs | 17 ++++++- codex-rs/core/src/session_prefix_tests.rs | 20 ++++++++ .../tests/suite/subagent_notifications.rs | 47 +++++++++++++++---- 7 files changed, 104 insertions(+), 13 deletions(-) create mode 100644 codex-rs/core/src/session_prefix_tests.rs diff --git a/codex-rs/core/src/hook_runtime.rs b/codex-rs/core/src/hook_runtime.rs index 325fd17c9..96ae6c990 100644 --- a/codex-rs/core/src/hook_runtime.rs +++ b/codex-rs/core/src/hook_runtime.rs @@ -25,6 +25,7 @@ use codex_protocol::items::TurnItem; use codex_protocol::items::UserMessageItem; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::HookCompletedEvent; use codex_protocol::protocol::HookEventName; @@ -490,7 +491,7 @@ pub(crate) async fn run_legacy_after_agent_hook( }; let event = EventMsg::Error(codex_protocol::protocol::ErrorEvent { message, - codex_error_info: None, + codex_error_info: Some(CodexErrorInfo::Other), }); sess.send_event(turn_context, event).await; true diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 02809a134..242e35f37 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1654,6 +1654,18 @@ 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 + .codex_error_info + .as_ref() + .is_some_and(CodexErrorInfo::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); @@ -1705,8 +1717,18 @@ impl Session { return; }; - let Some(status) = agent_status_from_event(msg) else { - return; + let status = match turn_context.terminal_error.lock().await.take() { + Some(error) => { + let status = AgentStatus::Errored(error); + self.agent_status.send_replace(status.clone()); + status + } + None => { + let Some(status) = agent_status_from_event(msg) else { + return; + }; + status + } }; if !is_final(&status) { return; diff --git a/codex-rs/core/src/session/review.rs b/codex-rs/core/src/session/review.rs index 9a7dfc829..746265971 100644 --- a/codex-rs/core/src/session/review.rs +++ b/codex-rs/core/src/session/review.rs @@ -152,6 +152,7 @@ pub(super) async fn spawn_review_thread( extension_data, 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)), server_model_warning_emitted: AtomicBool::new(false), model_verification_emitted: AtomicBool::new(false), }; diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index acb126d7e..042eca892 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -167,6 +167,7 @@ pub struct TurnContext { pub(crate) extension_data: Arc, pub(crate) turn_skills: TurnSkillsContext, pub(crate) turn_timing_state: Arc, + pub(crate) terminal_error: Arc>>, pub(crate) server_model_warning_emitted: AtomicBool, pub(crate) model_verification_emitted: AtomicBool, } @@ -336,6 +337,7 @@ impl TurnContext { extension_data: Arc::clone(&self.extension_data), turn_skills: self.turn_skills.clone(), turn_timing_state: Arc::clone(&self.turn_timing_state), + terminal_error: Arc::clone(&self.terminal_error), server_model_warning_emitted: AtomicBool::new( self.server_model_warning_emitted.load(Ordering::Relaxed), ), @@ -639,6 +641,7 @@ impl Session { extension_data, turn_skills: TurnSkillsContext::new(skills_outcome), turn_timing_state: Arc::new(TurnTimingState::default()), + terminal_error: Arc::new(Mutex::new(None)), server_model_warning_emitted: AtomicBool::new(false), model_verification_emitted: AtomicBool::new(false), } diff --git a/codex-rs/core/src/session_prefix.rs b/codex-rs/core/src/session_prefix.rs index 4e6b9da94..7d4a74a02 100644 --- a/codex-rs/core/src/session_prefix.rs +++ b/codex-rs/core/src/session_prefix.rs @@ -1,10 +1,18 @@ use codex_protocol::AgentPath; use codex_protocol::protocol::AgentStatus; +use codex_utils_output_truncation::TruncationPolicy; +use codex_utils_output_truncation::truncate_text; use crate::context::ContextualUserFragment; use crate::context::InterAgentCompletionMessage; use crate::context::SubagentNotification; +const COMPLETION_MESSAGE_MAX_TOKENS: usize = 1_000; +const COMPLETION_MESSAGE_ENVELOPE_TOKEN_RESERVE: usize = 100; +const ERROR_MAX_TOKENS: usize = + COMPLETION_MESSAGE_MAX_TOKENS - COMPLETION_MESSAGE_ENVELOPE_TOKEN_RESERVE; +const ERROR_NEXT_ACTION: &str = "This agent's turn failed. If you still need this agent, use the available collaboration tools to give it another task."; + // Helpers for model-visible session state markers that are stored in user-role // messages but are not user intent. @@ -24,7 +32,10 @@ pub(crate) fn format_inter_agent_completion_message( let payload = match status { AgentStatus::Completed(Some(message)) => message.clone(), AgentStatus::Completed(None) => String::new(), - AgentStatus::Errored(error) => format!("Agent errored: {error}"), + AgentStatus::Errored(error) => { + let error = truncate_text(error, TruncationPolicy::Tokens(ERROR_MAX_TOKENS)); + format!("Agent errored: {error}\n\n{ERROR_NEXT_ACTION}") + } AgentStatus::Shutdown => "Agent shut down.".to_string(), AgentStatus::NotFound => "Agent was not found.".to_string(), AgentStatus::PendingInit | AgentStatus::Running | AgentStatus::Interrupted => return None, @@ -32,6 +43,10 @@ pub(crate) fn format_inter_agent_completion_message( Some(InterAgentCompletionMessage::new(task_name, sender, payload).render()) } +#[cfg(test)] +#[path = "session_prefix_tests.rs"] +mod tests; + pub(crate) fn format_subagent_context_line( agent_reference: &str, agent_nickname: Option<&str>, diff --git a/codex-rs/core/src/session_prefix_tests.rs b/codex-rs/core/src/session_prefix_tests.rs new file mode 100644 index 000000000..6a658ce85 --- /dev/null +++ b/codex-rs/core/src/session_prefix_tests.rs @@ -0,0 +1,20 @@ +use codex_protocol::AgentPath; +use codex_protocol::protocol::AgentStatus; +use codex_utils_output_truncation::approx_token_count; + +use super::COMPLETION_MESSAGE_MAX_TOKENS; +use super::ERROR_NEXT_ACTION; +use super::format_inter_agent_completion_message; + +#[test] +fn error_completion_message_stays_below_manual_review_threshold() { + let message = format_inter_agent_completion_message( + AgentPath::root(), + AgentPath::try_from("/root/worker").expect("valid agent path"), + &AgentStatus::Errored("stream disconnected ".repeat(1_000)), + ) + .expect("error status should produce a completion message"); + + assert!(approx_token_count(&message) < COMPLETION_MESSAGE_MAX_TOKENS); + assert!(message.contains(ERROR_NEXT_ACTION)); +} diff --git a/codex-rs/core/tests/suite/subagent_notifications.rs b/codex-rs/core/tests/suite/subagent_notifications.rs index 8f1e35681..764380be7 100644 --- a/codex-rs/core/tests/suite/subagent_notifications.rs +++ b/codex-rs/core/tests/suite/subagent_notifications.rs @@ -40,6 +40,7 @@ use serde_json::json; use std::fs; use std::path::Path; use std::time::Duration; +use test_case::test_case; use tokio::time::Instant; use tokio::time::sleep; use wiremock::MockServer; @@ -1103,8 +1104,18 @@ async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result Ok(()) } +#[derive(Clone, Copy)] +enum CompletionScenario { + Completed, + TerminalError, +} + +#[test_case(CompletionScenario::Completed ; "completed")] +#[test_case(CompletionScenario::TerminalError ; "terminal_error")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn plaintext_multi_agent_v2_completion_sends_agent_message() -> Result<()> { +async fn plaintext_multi_agent_v2_completion_sends_agent_message( + scenario: CompletionScenario, +) -> Result<()> { let server = start_mock_server().await; let spawn_args = serde_json::to_string(&json!({ "message": "opaque-encrypted-message", @@ -1120,15 +1131,18 @@ async fn plaintext_multi_agent_v2_completion_sends_agent_message() -> Result<()> ]), ) .await; - let child_request = mount_response_once_match( - &server, - |req: &wiremock::Request| body_contains(req, "\"type\":\"agent_message\""), - sse_response(sse(vec![ + let child_events = match scenario { + CompletionScenario::Completed => vec![ ev_response_created("resp-child-1"), ev_assistant_message("msg-child-1", "child done"), ev_completed("resp-child-1"), - ])) - .set_delay(Duration::from_secs(1)), + ], + CompletionScenario::TerminalError => vec![ev_response_created("resp-child-1")], + }; + let child_request = mount_response_once_match( + &server, + |req: &wiremock::Request| body_contains(req, "\"type\":\"agent_message\""), + sse_response(sse(child_events)).set_delay(Duration::from_secs(1)), ) .await; mount_sse_once_match( @@ -1143,8 +1157,19 @@ async fn plaintext_multi_agent_v2_completion_sends_agent_message() -> Result<()> ]), ) .await; - let notification = - "Message Type: FINAL_ANSWER\nTask name: /root\nSender: /root/worker\nPayload:\nchild done"; + let error = "stream disconnected before completion: stream closed before response.completed"; + let (payload, expected_text) = match scenario { + CompletionScenario::Completed => ("child done".to_string(), "child done"), + CompletionScenario::TerminalError => ( + format!( + "Agent errored: {error}\n\nThis agent's turn failed. If you still need this agent, use the available collaboration tools to give it another task." + ), + error, + ), + }; + let notification = format!( + "Message Type: FINAL_ANSWER\nTask name: /root\nSender: /root/worker\nPayload:\n{payload}" + ); // If the child is still running when the parent turn starts, wait_agent blocks // until mailbox delivery. The follow-up request must then contain that delivery. mount_sse_once_match( @@ -1165,6 +1190,7 @@ async fn plaintext_multi_agent_v2_completion_sends_agent_message() -> Result<()> |req: &wiremock::Request| { body_contains(req, TURN_2_NO_WAIT_PROMPT) && body_contains(req, "Message Type: FINAL_ANSWER") + && body_contains(req, expected_text) }, sse(vec![ ev_response_created("resp-parent-4"), @@ -1184,6 +1210,9 @@ async fn plaintext_multi_agent_v2_completion_sends_agent_message() -> Result<()> .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; }) .build(&server) .await?;