mirror of
https://github.com/pchuan98/codex.git
synced 2026-07-01 00:31:56 +08:00
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
<subagent_notification>
{"agent_path":"/root/worker","status":{"completed":null}}
</subagent_notification>
```
Now the parent receives the actual terminal error, bounded to 1,000
tokens, together with an actionable recovery hint:
```text
<subagent_notification>
{
"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."
}
</subagent_notification>
```
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.
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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),
|
||||
};
|
||||
|
||||
@@ -167,6 +167,7 @@ pub struct TurnContext {
|
||||
pub(crate) extension_data: Arc<codex_extension_api::ExtensionData>,
|
||||
pub(crate) turn_skills: TurnSkillsContext,
|
||||
pub(crate) turn_timing_state: Arc<TurnTimingState>,
|
||||
pub(crate) terminal_error: Arc<Mutex<Option<String>>>,
|
||||
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),
|
||||
}
|
||||
|
||||
@@ -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>,
|
||||
|
||||
@@ -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));
|
||||
}
|
||||
@@ -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?;
|
||||
|
||||
Reference in New Issue
Block a user