Skip to content

Move child completion routing into the agent controller #46552

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
copyberry merged 1 commit into from
Sep 19, 2026
Merged

Move child completion routing into the agent controller #46552

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions codex-rs/core/src/agent/control.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@ pub(crate) use self::interrupt::AgentInterruptError;
pub(crate) use self::interrupt::AgentInterruptOutcome;
use self::residency::V2Residency;

mod completion;
mod delivery;
mod execution;
mod interrupt;
Expand Down
131 changes: 131 additions & 0 deletions codex-rs/core/src/agent/control/completion.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,131 @@
//! Delivers terminal child results and completion activity to the agent tree.
//!
//! Sessions capture terminal state; the controller owns routing and queue-only delivery.
//! Delivery remains best effort, with tracing recorded only after the parent accepts it.

use super::LocalAgentControl;
use crate::TurnStartOptions;
use crate::agent::api::AgentTurnOutcome;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::session_prefix::format_inter_agent_completion_message;
use codex_protocol::AgentPath;
use codex_protocol::items::SubAgentActivityItem;
use codex_protocol::protocol::AgentStatus;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentActivityKind;
use codex_protocol::protocol::SubAgentSource;
use codex_rollout_trace::AgentResultTracePayload;
use codex_rollout_trace::ThreadTraceContext;
use tracing::debug;

impl LocalAgentControl {
/// Routes a captured terminal outcome without retaining the child's live turn context.
pub(crate) async fn notify_parent_of_terminal_turn(
&self,
outcome: AgentTurnOutcome,
trace: &ThreadTraceContext,
) {
let SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
agent_path: Some(child_agent_path),
..
}) = &outcome.source
else {
return;
};
let parent_thread_id = *parent_thread_id;
let status = outcome.status;
let Some(parent_agent_path) = child_agent_path
.as_str()
.rsplit_once('/')
.and_then(|(parent, _)| AgentPath::try_from(parent).ok())
else {
return;
};

if matches!(status, AgentStatus::Completed(_))
&& let Some(parent_turn_id) = outcome.parent_turn_id
{
let initiating_thread_id = match outcome.initiating_agent_path.as_ref() {
Some(initiating_agent_path) if initiating_agent_path != &parent_agent_path => {
self.resolve_agent_reference(
outcome.thread_id,
&outcome.source,
initiating_agent_path.as_str(),
)
.await
.inspect_err(|err| {
debug!(
"failed to resolve completed activity initiator {initiating_agent_path}: {err}"
);
})
.ok()
}
_ => Some(parent_thread_id),
};
if let Some(initiating_thread_id) = initiating_thread_id
&& let Err(err) = self
.emit_sub_agent_activity(
initiating_thread_id,
parent_turn_id,
SubAgentActivityItem {
id: format!("subagent-completed-{}", outcome.turn_id),
kind: SubAgentActivityKind::Completed,
agent_thread_id: outcome.thread_id,
agent_path: child_agent_path.clone(),
},
)
.await
{
debug!(
"failed to emit completed activity to initiating thread {initiating_thread_id}: {err}"
);
}
}

let Some(message) = format_inter_agent_completion_message(
parent_agent_path.clone(),
child_agent_path.clone(),
&status,
) else {
return;
};
// `communication` owns the message. Keep a second copy only when the
// recorder will actually need it after parent delivery succeeds.
let trace_message = trace.is_enabled().then(|| message.clone());
let communication = InterAgentCommunication::new(
child_agent_path.clone(),
parent_agent_path,
Vec::new(),
message,
/*trigger_turn*/ false,
);
let context =
AgentCommunicationContext::new(AgentCommunicationKind::Result, outcome.thread_id);
if let Err(err) = self
.send_inter_agent_communication(
parent_thread_id,
communication,
context,
TurnStartOptions::default(),
)
.await
{
debug!("failed to notify parent thread {parent_thread_id}: {err}");
return;
}
if let Some(message) = trace_message {
trace.record_agent_result_interaction(
outcome.turn_id.as_str(),
parent_thread_id,
&AgentResultTracePayload {
child_agent_path: child_agent_path.as_str(),
message: &message,
status: &status,
},
);
}
}
}
144 changes: 17 additions & 127 deletions codex-rs/core/src/session/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,9 +13,8 @@ use std::time::UNIX_EPOCH;
use crate::agent::AgentStatus;
use crate::agent::LocalAgentControl;
use crate::agent::agent_status_from_event;
use crate::agent::api::AgentTurnOutcome;
use crate::agent::status::is_final;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::agents_md_manager::SessionInstructions;
use crate::attestation::AttestationProvider;
use crate::compact;
Expand Down Expand Up @@ -47,7 +46,6 @@ use crate::session::step_context::StepContext;
use crate::session::step_settings::ResolvedStepSettings;
use crate::session::step_settings::StepSettings;
use crate::session::turn_context::TurnEnvironment;
use crate::session_prefix::format_inter_agent_completion_message;
use crate::shell_snapshot::SnapshotCredentialBrokerState;
use crate::skills_load_input_from_config;
use crate::stream_events_utils::mark_thread_memory_mode_polluted_if_external_context;
Expand Down Expand Up @@ -113,7 +111,6 @@ use codex_protocol::dynamic_tools::DynamicToolResponse;
use codex_protocol::dynamic_tools::DynamicToolSpec;
use codex_protocol::items::EnteredReviewModeItem;
use codex_protocol::items::ModelInvocationContext;
use codex_protocol::items::SubAgentActivityItem;
use codex_protocol::items::TurnItem;
use codex_protocol::items::UserMessageItem;
use codex_protocol::models::ActivePermissionProfile;
Expand Down Expand Up @@ -141,7 +138,6 @@ use codex_protocol::protocol::MULTI_AGENT_MODE_OPEN_TAG;
use codex_protocol::protocol::MultiAgentVersion;
use codex_protocol::protocol::RawResponseItemEvent;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentActivityKind;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::ThreadSettingsOverrides;
Expand All @@ -163,7 +159,6 @@ use codex_protocol::request_user_input::RequestUserInputArgs;
use codex_protocol::request_user_input::RequestUserInputResponse;
use codex_rmcp_client::ElicitationResponse;
use codex_rollout::state_db;
use codex_rollout_trace::AgentResultTracePayload;
use codex_rollout_trace::ThreadStartedTraceMetadata;
use codex_rollout_trace::ThreadTraceContext;
use codex_sandboxing::SandboxType;
Expand Down Expand Up @@ -2399,8 +2394,7 @@ impl Session {
}

let SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
agent_path: Some(child_agent_path),
agent_path: Some(_),
..
}) = &turn_context.session_source
else {
Expand All @@ -2424,127 +2418,23 @@ impl Session {
return;
}

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.
async fn forward_child_completion_to_parent(
&self,
turn_context: &TurnContext,
parent_thread_id: ThreadId,
child_agent_path: &codex_protocol::AgentPath,
status: AgentStatus,
) {
let Some(parent_agent_path) = child_agent_path
.as_str()
.rsplit_once('/')
.and_then(|(parent, _)| codex_protocol::AgentPath::try_from(parent).ok())
else {
return;
};

if matches!(status, AgentStatus::Completed(_))
&& let Some(parent_turn_id) = turn_context.turn_metadata_state.parent_turn_id()
{
let initiating_thread_id = match turn_context
.turn_metadata_state
.initiating_agent_path()
{
Some(initiating_agent_path) if initiating_agent_path != &parent_agent_path => self
.services
.agent_control
.resolve_agent_reference(
self.thread_id,
&turn_context.session_source,
initiating_agent_path.as_str(),
)
.await
.inspect_err(|err| {
debug!(
"failed to resolve completed activity initiator {initiating_agent_path}: {err}"
);
})
.ok(),
_ => Some(parent_thread_id),
};
if let Some(initiating_thread_id) = initiating_thread_id
&& let Err(err) = self
.services
.agent_control
.emit_sub_agent_activity(
initiating_thread_id,
parent_turn_id,
SubAgentActivityItem {
id: format!("subagent-completed-{}", turn_context.sub_id),
kind: SubAgentActivityKind::Completed,
agent_thread_id: self.thread_id,
agent_path: child_agent_path.clone(),
},
)
.await
{
debug!(
"failed to emit completed activity to initiating thread {initiating_thread_id}: {err}"
);
}
}

let Some(message) = format_inter_agent_completion_message(
parent_agent_path.clone(),
child_agent_path.clone(),
&status,
) else {
return;
};
// `communication` owns the message. Keep a second copy only when the
// recorder will actually need it after parent delivery succeeds.
let trace_message = self
.services
.rollout_thread_trace
.is_enabled()
.then(|| message.clone());
let communication = InterAgentCommunication::new(
child_agent_path.clone(),
parent_agent_path,
Vec::new(),
message,
/*trigger_turn*/ false,
);
let context =
AgentCommunicationContext::new(AgentCommunicationKind::Result, self.thread_id);
if let Err(err) = self
.services
self.services
.agent_control
.send_inter_agent_communication(
parent_thread_id,
communication,
context,
TurnStartOptions::default(),
.notify_parent_of_terminal_turn(
AgentTurnOutcome {
thread_id: self.thread_id,
turn_id: turn_context.sub_id.clone(),
source: turn_context.session_source.clone(),
parent_turn_id: turn_context.turn_metadata_state.parent_turn_id(),
initiating_agent_path: turn_context
.turn_metadata_state
.initiating_agent_path()
.cloned(),
status,
},
&self.services.rollout_thread_trace,
)
.await
{
debug!("failed to notify parent thread {parent_thread_id}: {err}");
return;
}
if let Some(message) = trace_message {
self.services
.rollout_thread_trace
.record_agent_result_interaction(
turn_context.sub_id.as_str(),
parent_thread_id,
&AgentResultTracePayload {
child_agent_path: child_agent_path.as_str(),
message: &message,
status: &status,
},
);
}
.await;
}

async fn maybe_mirror_event_text_to_realtime(&self, msg: &EventMsg) {
Expand Down
Loading