openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-25 18:28:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Log multi-agent communication lifecycle

Browse Source
This commit is contained in:
Michael Bolin
2026-07-01 15:31:35 -07:00
parent f6dcc0e68b
commit 425acfd752
21 changed files with 594 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control.rs
+81 -10
View File
@@ -4,6 +4,8 @@ use crate::agent::registry::AgentRegistry;
use crate::agent::role::DEFAULT_ROLE_NAME;
use crate::agent::role::resolve_role_config;
use crate::agent::status::is_final;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::codex_thread::ThreadConfigSnapshot;
use crate::config::Config;
use crate::config::RolloutBudgetConfig;
@@ -61,6 +63,21 @@ pub(crate) enum SpawnAgentForkMode {
LastNTurns(usize),
}
#[derive(Clone, Debug)]
pub(crate) enum AgentSubmission {
Op(Box<Op>),
InterAgentCommunication {
communication: InterAgentCommunication,
context: AgentCommunicationContext,
},
}
impl AgentSubmission {
pub(crate) fn op(op: Op) -> Self {
Self::Op(Box::new(op))
}
}
#[derive(Clone, Debug, Default)]
pub(crate) struct SpawnAgentOptions {
pub(crate) fork_parent_spawn_call_id: Option<String>,
@@ -142,9 +159,10 @@ impl AgentControl {
initial_operation: Op,
) -> CodexResult<String> {
let state = self.upgrade()?;
self.ensure_execution_capacity_for_op(agent_id, &initial_operation)
let submission = AgentSubmission::op(initial_operation);
self.ensure_execution_capacity_for_submission(agent_id, &submission)
.await?;
self.send_input_after_capacity_check(agent_id, &state, initial_operation)
self.send_input_after_capacity_check(agent_id, &state, submission)
.await
}
@@ -152,13 +170,19 @@ impl AgentControl {
&self,
agent_id: ThreadId,
state: &Arc<ThreadManagerState>,
initial_operation: Op,
submission: AgentSubmission,
) -> CodexResult<String> {
if let Op::InterAgentCommunication { communication } = initial_operation {
return self
.submit_inter_agent_communication(agent_id, state, communication)
.await;
}
let initial_operation = match submission {
AgentSubmission::InterAgentCommunication {
communication,
context,
} => {
return self
.submit_inter_agent_communication(agent_id, state, communication, context)
.await;
}
AgentSubmission::Op(initial_operation) => *initial_operation,
};
let last_task_message = non_empty_task_message(render_input_preview(&initial_operation));
let result = self
@@ -183,8 +207,16 @@ impl AgentControl {
&self,
agent_id: ThreadId,
communication: InterAgentCommunication,
agent_communication_context: AgentCommunicationContext,
) -> CodexResult<String> {
self.send_input(agent_id, Op::InterAgentCommunication { communication })
let state = self.upgrade()?;
let submission = AgentSubmission::InterAgentCommunication {
communication,
context: agent_communication_context,
};
self.ensure_execution_capacity_for_submission(agent_id, &submission)
.await?;
self.send_input_after_capacity_check(agent_id, &state, submission)
.await
}
@@ -193,8 +225,27 @@ impl AgentControl {
agent_id: ThreadId,
state: &Arc<ThreadManagerState>,
communication: InterAgentCommunication,
context: AgentCommunicationContext,
) -> CodexResult<String> {
let last_task_message = last_task_message_from_communication(&communication);
let send_log = if crate::agent_communication::logging_enabled() {
let clock_thread = match state.get_thread(context.sender_thread_id).await {
Ok(thread) => Some(thread),
Err(_) => state.get_thread(agent_id).await.ok(),
};
clock_thread.map(|thread| {
let services = &thread.codex.session.services;
let runtime_handle = services.runtime_handle.clone();
let simclock_time =
runtime_handle.spawn(crate::agent_communication::read_simclock_time(
Arc::clone(&services.simclock_time_provider),
context.sender_thread_id,
));
(runtime_handle, simclock_time, communication.clone())
})
} else {
None
};
let result = self
.handle_thread_request_result(
agent_id,
@@ -204,6 +255,24 @@ impl AgentControl {
.await,
)
.await;
if let Some((runtime_handle, simclock_time, communication)) = send_log {
match result.as_ref() {
Ok(communication_id) => {
let communication_id = communication_id.clone();
runtime_handle.spawn(async move {
let simclock_time = simclock_time.await.ok().flatten();
crate::agent_communication::emit_agent_communication_send(
&communication_id,
&context,
&communication,
agent_id,
simclock_time,
);
});
}
Err(_) => simclock_time.abort(),
}
}
if result.is_ok() {
match last_task_message {
Some(last_task_message) => self
@@ -494,8 +563,10 @@ impl AgentControl {
message,
/*trigger_turn*/ false,
);
let context =
AgentCommunicationContext::new(AgentCommunicationKind::Result, child_thread_id);
let _ = control
.send_inter_agent_communication(parent_thread_id, communication)
.send_inter_agent_communication(parent_thread_id, communication, context)
.await;
return;
}
codex-rs/core/src/agent/control/execution.rs
+29 -1
View File
@@ -1,4 +1,5 @@
use super::AgentControl;
use super::AgentSubmission;
use codex_protocol::ThreadId;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
@@ -32,7 +33,25 @@ impl AgentControl {
thread_id: ThreadId,
op: &Op,
) -> CodexResult<()> {
if !op_starts_turn(op) {
self.ensure_execution_capacity_for_turn_start(thread_id, op_starts_turn(op))
.await
}
pub(crate) async fn ensure_execution_capacity_for_submission(
&self,
thread_id: ThreadId,
submission: &AgentSubmission,
) -> CodexResult<()> {
self.ensure_execution_capacity_for_turn_start(thread_id, submission_starts_turn(submission))
.await
}
async fn ensure_execution_capacity_for_turn_start(
&self,
thread_id: ThreadId,
starts_turn: bool,
) -> CodexResult<()> {
if !starts_turn {
return Ok(());
}
let state = self.upgrade()?;
@@ -92,6 +111,15 @@ impl AgentExecutionLimiter {
}
}
fn submission_starts_turn(submission: &AgentSubmission) -> bool {
match submission {
AgentSubmission::Op(op) => op_starts_turn(op),
AgentSubmission::InterAgentCommunication { communication, .. } => {
communication.trigger_turn
}
}
}
fn op_starts_turn(op: &Op) -> bool {
matches!(op, Op::UserInput { .. })
|| matches!(op, Op::InterAgentCommunication { communication } if communication.trigger_turn)
codex-rs/core/src/agent/control/spawn.rs
+5 -5
View File
@@ -94,7 +94,7 @@ impl AgentControl {
) -> CodexResult<ThreadId> {
let spawned_agent = Box::pin(self.spawn_agent_internal(
config,
initial_operation,
AgentSubmission::op(initial_operation),
session_source,
SpawnAgentOptions::default(),
))
@@ -106,11 +106,11 @@ impl AgentControl {
pub(crate) async fn spawn_agent_with_metadata(
&self,
config: Config,
initial_operation: Op,
initial_submission: AgentSubmission,
session_source: Option<SessionSource>,
options: SpawnAgentOptions, // TODO(jif) drop with new fork.
) -> CodexResult<LiveAgent> {
Box::pin(self.spawn_agent_internal(config, initial_operation, session_source, options))
Box::pin(self.spawn_agent_internal(config, initial_submission, session_source, options))
.await
}
@@ -197,7 +197,7 @@ impl AgentControl {
async fn spawn_agent_internal(
&self,
config: Config,
initial_operation: Op,
initial_submission: AgentSubmission,
session_source: Option<SessionSource>,
options: SpawnAgentOptions,
) -> CodexResult<LiveAgent> {
@@ -356,7 +356,7 @@ impl AgentControl {
)
.await;
self.send_input_after_capacity_check(new_thread.thread_id, &state, initial_operation)
self.send_input_after_capacity_check(new_thread.thread_id, &state, initial_submission)
.await?;
if multi_agent_version != MultiAgentVersion::V2 {
let child_reference = agent_metadata
codex-rs/core/src/agent/control_tests.rs
+27 -13
View File
@@ -3,6 +3,8 @@ use crate::CodexThread;
use crate::StateDbHandle;
use crate::ThreadManager;
use crate::agent::agent_status_from_event;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::config::AgentRoleConfig;
use crate::config::Config;
use crate::config::ConfigBuilder;
@@ -531,7 +533,11 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig
let submission_id = harness
.control
.send_inter_agent_communication(thread_id, communication.clone())
.send_inter_agent_communication(
thread_id,
communication.clone(),
AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()),
)
.await
.expect("send_inter_agent_communication should succeed");
assert!(!submission_id.is_empty());
@@ -591,7 +597,7 @@ async fn ensure_v2_agent_loaded_reloads_registered_unloaded_agent() {
.control
.spawn_agent_with_metadata(
harness.config.clone(),
text_input("hello child"),
AgentSubmission::op(text_input("hello child")),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
@@ -656,7 +662,11 @@ async fn ensure_v2_agent_loaded_reloads_registered_unloaded_agent() {
);
harness
.control
.send_inter_agent_communication(spawned_agent.thread_id, communication.clone())
.send_inter_agent_communication(
spawned_agent.thread_id,
communication.clone(),
AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()),
)
.await
.expect("send_inter_agent_communication should succeed after reload");
let expected = (
@@ -770,7 +780,7 @@ async fn encrypted_inter_agent_communication_clears_existing_last_task_message()
.control
.spawn_agent_with_metadata(
harness.config.clone(),
text_input("old plaintext task"),
AgentSubmission::op(text_input("old plaintext task")),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
@@ -803,7 +813,11 @@ async fn encrypted_inter_agent_communication_clears_existing_last_task_message()
);
harness
.control
.send_inter_agent_communication(spawned_agent.thread_id, communication)
.send_inter_agent_communication(
spawned_agent.thread_id,
communication,
AgentCommunicationContext::new(AgentCommunicationKind::Followup, ThreadId::new()),
)
.await
.expect("send_inter_agent_communication should succeed");
@@ -996,7 +1010,7 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
.control
.spawn_agent_with_metadata(
child_config,
text_input("child task"),
AgentSubmission::op(text_input("child task")),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
@@ -1057,7 +1071,7 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
.control
.spawn_agent_with_metadata(
no_hint_child_config,
text_input("child task without hints"),
AgentSubmission::op(text_input("child task without hints")),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
@@ -1197,7 +1211,7 @@ async fn spawn_agent_fork_strips_parent_usage_hints_from_compacted_history() {
.control
.spawn_agent_with_metadata(
child_config,
text_input("child task"),
AgentSubmission::op(text_input("child task")),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
@@ -1267,7 +1281,7 @@ async fn spawn_agent_fork_flushes_parent_rollout_before_loading_history() {
.control
.spawn_agent_with_metadata(
harness.config.clone(),
text_input("child task"),
AgentSubmission::op(text_input("child task")),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
@@ -1384,7 +1398,7 @@ async fn spawn_agent_fork_last_n_turns_keeps_only_recent_turns() {
.control
.spawn_agent_with_metadata(
harness.config.clone(),
text_input("child task"),
AgentSubmission::op(text_input("child task")),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
@@ -1524,7 +1538,7 @@ async fn spawn_agent_fork_last_n_turns_drops_parent_startup_prefix_when_under_li
.control
.spawn_agent_with_metadata(
harness.config.clone(),
text_input("child task"),
AgentSubmission::op(text_input("child task")),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
@@ -1643,7 +1657,7 @@ async fn spawn_agent_fork_last_n_turns_strips_parent_usage_hints() {
.control
.spawn_agent_with_metadata(
child_config,
text_input("child task"),
AgentSubmission::op(text_input("child task")),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
@@ -2297,7 +2311,7 @@ async fn spawn_thread_subagents_persist_parent_originator_across_new_and_truncat
.control
.spawn_agent_with_metadata(
harness.config.clone(),
text_input("hello forked child"),
AgentSubmission::op(text_input("hello forked child")),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: parent.thread_id,
depth: 1,
codex-rs/core/src/agent_communication.rs
+96
View File
@@ -0,0 +1,96 @@
use codex_protocol::ThreadId;
use codex_protocol::protocol::InterAgentCommunication;
use std::sync::Arc;
use crate::current_time::TimeProvider;
const AGENT_COMMUNICATION_TARGET: &str = "codex_core::agent_communication";
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum AgentCommunicationKind {
Spawn,
Message,
Followup,
Result,
}
impl AgentCommunicationKind {
fn as_str(self) -> &'static str {
match self {
Self::Spawn => "spawn",
Self::Message => "message",
Self::Followup => "followup",
Self::Result => "result",
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct AgentCommunicationContext {
kind: AgentCommunicationKind,
pub(crate) sender_thread_id: ThreadId,
}
impl AgentCommunicationContext {
pub(crate) fn new(kind: AgentCommunicationKind, sender_thread_id: ThreadId) -> Self {
Self {
kind,
sender_thread_id,
}
}
}
pub(crate) fn logging_enabled() -> bool {
tracing::enabled!(target: AGENT_COMMUNICATION_TARGET, tracing::Level::TRACE)
}
pub(crate) fn emit_agent_communication_send(
communication_id: &str,
context: &AgentCommunicationContext,
communication: &InterAgentCommunication,
receiver_thread_id: ThreadId,
simclock_time: Option<i64>,
) {
tracing::trace!(
target: AGENT_COMMUNICATION_TARGET,
{
event.name = "codex.agent_communication",
communication_id,
kind = context.kind.as_str(),
state = "send",
sender_thread_id = %context.sender_thread_id,
receiver_thread_id = %receiver_thread_id,
content = if communication.content.is_empty() {
communication.encrypted_content.as_deref().unwrap_or_default()
} else {
communication.content.as_str()
},
simclock_time,
},
"agent communication"
);
}
pub(crate) async fn read_simclock_time(
time_provider: Arc<dyn TimeProvider>,
thread_id: ThreadId,
) -> Option<i64> {
time_provider
.current_time(thread_id)
.await
.ok()
.map(|time| time.timestamp())
}
pub(crate) fn emit_agent_communication_receive(communication_id: &str, simclock_time: Option<i64>) {
tracing::trace!(
target: AGENT_COMMUNICATION_TARGET,
{
event.name = "codex.agent_communication",
communication_id,
state = "receive",
simclock_time,
},
"agent communication"
);
}
codex-rs/core/src/lib.rs
+1
View File
@@ -31,6 +31,7 @@ pub use codex_thread::TryStartTurnIfIdleError;
pub use codex_thread::TryStartTurnIfIdleRejectionReason;
pub use session::turn_context::TurnContext;
mod agent;
mod agent_communication;
mod attestation;
mod codex_delegate;
mod command_canonicalization;
codex-rs/core/src/session/handlers.rs
+13
View File
@@ -285,6 +285,19 @@ pub async fn inter_agent_communication(
sess.input_queue
.enqueue_mailbox_communication(communication)
.await;
if crate::agent_communication::logging_enabled() {
let communication_id = sub_id.clone();
let thread_id = sess.thread_id;
let time_provider = Arc::clone(&sess.services.simclock_time_provider);
sess.services.runtime_handle.spawn(async move {
let simclock_time =
crate::agent_communication::read_simclock_time(time_provider, thread_id).await;
crate::agent_communication::emit_agent_communication_receive(
&communication_id,
simclock_time,
);
});
}
if trigger_turn {
sess.maybe_start_turn_for_pending_work_with_sub_id(sub_id)
.await;
codex-rs/core/src/session/mod.rs
+5 -1
View File
@@ -14,6 +14,8 @@ use crate::agent::AgentControl;
use crate::agent::AgentStatus;
use crate::agent::agent_status_from_event;
use crate::agent::status::is_final;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::attestation::AttestationProvider;
use crate::build_available_skills;
use crate::compact;
@@ -1867,10 +1869,12 @@ impl Session {
message,
/*trigger_turn*/ false,
);
let context =
AgentCommunicationContext::new(AgentCommunicationKind::Result, self.thread_id);
if let Err(err) = self
.services
.agent_control
.send_inter_agent_communication(parent_thread_id, communication)
.send_inter_agent_communication(parent_thread_id, communication, context)
.await
{
debug!("failed to notify parent thread {parent_thread_id}: {err}");
codex-rs/core/src/session/session.rs
+4
View File
@@ -557,6 +557,9 @@ impl Session {
.effective_agent_max_threads(MultiAgentVersion::V2)
.unwrap_or(usize::MAX),
);
let simclock_time_provider: Arc<dyn TimeProvider> = external_time_provider
.clone()
.unwrap_or_else(|| Arc::new(crate::current_time::SystemTimeProvider));
let time_provider = crate::current_time::resolve_time_provider(
config.current_time_reminder.as_ref(),
external_time_provider,
@@ -1094,6 +1097,7 @@ impl Session {
thread_store: Arc::clone(&thread_store),
attestation_provider: attestation_provider.clone(),
time_provider,
simclock_time_provider,
model_client: ModelClient::new(
Some(Arc::clone(&auth_manager)),
if config.features.enabled(Feature::UseAgentIdentity) {
codex-rs/core/src/session/tests.rs
+2
View File
@@ -5431,6 +5431,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
)),
attestation_provider: None,
time_provider: Arc::new(crate::current_time::SystemTimeProvider),
simclock_time_provider: Arc::new(crate::current_time::SystemTimeProvider),
model_client: ModelClient::new(
Some(auth_manager.clone()),
AgentIdentityAuthPolicy::JwtOnly,
@@ -7557,6 +7558,7 @@ where
)),
attestation_provider: None,
time_provider: Arc::new(crate::current_time::SystemTimeProvider),
simclock_time_provider: Arc::new(crate::current_time::SystemTimeProvider),
model_client: ModelClient::new(
Some(Arc::clone(&auth_manager)),
AgentIdentityAuthPolicy::JwtOnly,
codex-rs/core/src/state/service.rs
+1
View File
@@ -94,6 +94,7 @@ pub(crate) struct SessionServices {
pub(crate) thread_store: Arc<dyn ThreadStore>,
pub(crate) attestation_provider: Option<Arc<dyn AttestationProvider>>,
pub(crate) time_provider: Arc<dyn TimeProvider>,
pub(crate) simclock_time_provider: Arc<dyn TimeProvider>,
/// Session-scoped model client shared across turns.
pub(crate) model_client: ModelClient,
pub(crate) code_mode_service: CodeModeService,
codex-rs/core/src/tools/handlers/agent_jobs.rs
+2 -1
View File
@@ -1,3 +1,4 @@
use crate::agent::control::AgentSubmission;
use crate::agent::control::SpawnAgentOptions;
use crate::agent::status::is_final;
use crate::config::Config;
@@ -203,7 +204,7 @@ async fn run_agent_job_loop(
.agent_control
.spawn_agent_with_metadata(
options.spawn_config.clone(),
items.into(),
AgentSubmission::op(items.into()),
Some(SessionSource::SubAgent(SubAgentSource::Other(format!(
"agent_job:{job_id}"
)))),
codex-rs/core/src/tools/handlers/multi_agents/spawn.rs
+2 -1
View File
@@ -1,4 +1,5 @@
use super::*;
use crate::agent::control::AgentSubmission;
use crate::agent::control::SpawnAgentForkMode;
use crate::agent::control::SpawnAgentOptions;
use crate::agent::control::render_input_preview;
@@ -119,7 +120,7 @@ async fn handle_spawn_agent(
let result = Box::pin(session.services.agent_control.spawn_agent_with_metadata(
config,
input_items,
AgentSubmission::op(input_items),
Some(thread_spawn_source(
session.thread_id,
&turn.session_source,
codex-rs/core/src/tools/handlers/multi_agents_tests.rs
+43 -30
View File
@@ -1,5 +1,6 @@
use super::*;
use crate::ThreadManager;
use crate::agent::control::AgentSubmission;
use crate::config::AgentRoleConfig;
use crate::config::DEFAULT_AGENT_MAX_DEPTH;
use crate::function_tool::FunctionCallError;
@@ -1374,11 +1375,13 @@ async fn multi_agent_v2_send_message_accepts_root_target_from_child() {
.agent_control
.spawn_agent_with_metadata(
(*turn.config).clone(),
vec![UserInput::Text {
text: "inspect this repo".to_string(),
text_elements: Vec::new(),
}]
.into(),
AgentSubmission::op(
vec![UserInput::Text {
text: "inspect this repo".to_string(),
text_elements: Vec::new(),
}]
.into(),
),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
@@ -1451,11 +1454,13 @@ async fn multi_agent_v2_followup_task_rejects_root_target_from_child() {
.agent_control
.spawn_agent_with_metadata(
(*turn.config).clone(),
vec![UserInput::Text {
text: "inspect this repo".to_string(),
text_elements: Vec::new(),
}]
.into(),
AgentSubmission::op(
vec![UserInput::Text {
text: "inspect this repo".to_string(),
text_elements: Vec::new(),
}]
.into(),
),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
@@ -1623,11 +1628,13 @@ async fn multi_agent_v2_list_agents_filters_by_relative_path_prefix() {
.agent_control
.spawn_agent_with_metadata(
config.clone(),
vec![UserInput::Text {
text: "research".to_string(),
text_elements: Vec::new(),
}]
.into(),
AgentSubmission::op(
vec![UserInput::Text {
text: "research".to_string(),
text_elements: Vec::new(),
}]
.into(),
),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
@@ -1644,11 +1651,13 @@ async fn multi_agent_v2_list_agents_filters_by_relative_path_prefix() {
.agent_control
.spawn_agent_with_metadata(
config,
vec![UserInput::Text {
text: "build".to_string(),
text_elements: Vec::new(),
}]
.into(),
AgentSubmission::op(
vec![UserInput::Text {
text: "build".to_string(),
text_elements: Vec::new(),
}]
.into(),
),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 2,
@@ -4094,11 +4103,13 @@ async fn multi_agent_v2_interrupt_agent_rejects_self_target_by_id() {
.agent_control
.spawn_agent_with_metadata(
(*turn.config).clone(),
vec![UserInput::Text {
text: "inspect this repo".to_string(),
text_elements: Vec::new(),
}]
.into(),
AgentSubmission::op(
vec![UserInput::Text {
text: "inspect this repo".to_string(),
text_elements: Vec::new(),
}]
.into(),
),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
@@ -4162,11 +4173,13 @@ async fn multi_agent_v2_interrupt_agent_rejects_self_target_by_task_name() {
.agent_control
.spawn_agent_with_metadata(
(*turn.config).clone(),
vec![UserInput::Text {
text: "inspect this repo".to_string(),
text_elements: Vec::new(),
}]
.into(),
AgentSubmission::op(
vec![UserInput::Text {
text: "inspect this repo".to_string(),
text_elements: Vec::new(),
}]
.into(),
),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs
+8 -1
View File
@@ -4,6 +4,8 @@
//! resulting `InterAgentCommunication` should wake the target immediately.
use super::*;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::tools::context::FunctionToolOutput;
use crate::turn_timing::now_unix_timestamp_ms;
use codex_protocol::protocol::InterAgentCommunication;
@@ -101,10 +103,15 @@ pub(crate) async fn handle_message_string_tool(
.unwrap_or_else(AgentPath::root);
let communication =
communication_from_tool_message(author, receiver_agent_path.clone(), message);
let kind = match mode {
MessageDeliveryMode::QueueOnly => AgentCommunicationKind::Message,
MessageDeliveryMode::TriggerTurn => AgentCommunicationKind::Followup,
};
let context = AgentCommunicationContext::new(kind, session.thread_id);
let result = session
.services
.agent_control
.send_inter_agent_communication(receiver_thread_id, mode.apply(communication))
.send_inter_agent_communication(receiver_thread_id, mode.apply(communication), context)
.await
.map_err(|err| collab_agent_error(receiver_thread_id, err));
result?;
codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs
+35 -28
View File
@@ -1,9 +1,12 @@
use super::*;
use crate::agent::control::AgentSubmission;
use crate::agent::control::SpawnAgentForkMode;
use crate::agent::control::SpawnAgentOptions;
use crate::agent::next_thread_spawn_depth;
use crate::agent::role::DEFAULT_ROLE_NAME;
use crate::agent::role::apply_role_to_config;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::tools::handlers::multi_agents_spec::SpawnAgentToolOptions;
use crate::tools::handlers::multi_agents_spec::create_spawn_agent_tool_v2;
use crate::turn_timing::now_unix_timestamp_ms;
@@ -104,34 +107,38 @@ async fn handle_spawn_agent(
"spawned agent is missing a canonical task name".to_string(),
)
})?;
let spawned_agent = Box::pin(
session.services.agent_control.spawn_agent_with_metadata(
config,
match initial_operation {
Op::UserInput { items, .. }
if items
.iter()
.all(|item| matches!(item, UserInput::Text { .. })) =>
{
let author = turn
.session_source
.get_agent_path()
.unwrap_or_else(AgentPath::root);
let communication =
communication_from_tool_message(author, new_agent_path.clone(), message);
Op::InterAgentCommunication { communication }
}
initial_operation => initial_operation,
},
Some(spawn_source),
SpawnAgentOptions {
fork_parent_spawn_call_id: fork_mode.as_ref().map(|_| call_id.clone()),
fork_mode,
parent_thread_id: Some(session.thread_id),
environments: Some(turn.environments.to_selections()),
},
),
)
let initial_submission = match initial_operation {
Op::UserInput { items, .. }
if items
.iter()
.all(|item| matches!(item, UserInput::Text { .. })) =>
{
let author = turn
.session_source
.get_agent_path()
.unwrap_or_else(AgentPath::root);
let communication =
communication_from_tool_message(author, new_agent_path.clone(), message);
let context =
AgentCommunicationContext::new(AgentCommunicationKind::Spawn, session.thread_id);
AgentSubmission::InterAgentCommunication {
communication,
context,
}
}
initial_operation => AgentSubmission::op(initial_operation),
};
let spawned_agent = Box::pin(session.services.agent_control.spawn_agent_with_metadata(
config,
initial_submission,
Some(spawn_source),
SpawnAgentOptions {
fork_parent_spawn_call_id: fork_mode.as_ref().map(|_| call_id.clone()),
fork_mode,
parent_thread_id: Some(session.thread_id),
environments: Some(turn.environments.to_selections()),
},
))
.await
.map_err(collab_spawn_error)?;
let new_thread_id = spawned_agent.thread_id;
codex-rs/core/tests/suite/agent_communication.rs
+204
View File
@@ -0,0 +1,204 @@
use anyhow::Result;
use chrono::DateTime;
use chrono::Utc;
use codex_core::SleepFuture;
use codex_core::TimeFuture;
use codex_core::TimeProvider;
use codex_features::Feature;
use codex_protocol::ThreadId;
use core_test_support::responses::ev_assistant_message;
use core_test_support::responses::ev_completed;
use core_test_support::responses::ev_function_call_with_namespace;
use core_test_support::responses::ev_response_created;
use core_test_support::responses::mount_sse_once_match;
use core_test_support::responses::sse;
use core_test_support::responses::start_mock_server;
use core_test_support::test_codex::test_codex;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use tracing::Level;
use tracing_test::internal::MockWriter;
use wiremock::Request;
const ROOT_SIMCLOCK_TIME: i64 = 1_000;
const CHILD_SIMCLOCK_TIME: i64 = 2_000;
#[derive(Default)]
struct ThreadSimClock {
root_thread_id: Mutex<Option<ThreadId>>,
}
impl ThreadSimClock {
fn set_root_thread_id(&self, thread_id: ThreadId) {
*self.root_thread_id.lock().expect("simclock lock") = Some(thread_id);
}
}
impl TimeProvider for ThreadSimClock {
fn current_time(&self, thread_id: ThreadId) -> TimeFuture<'_> {
let root_thread_id = self
.root_thread_id
.lock()
.expect("simclock lock")
.expect("root thread ID should be installed before reading the clock");
let timestamp = if thread_id == root_thread_id {
ROOT_SIMCLOCK_TIME
} else {
CHILD_SIMCLOCK_TIME
};
Box::pin(async move {
Ok(DateTime::<Utc>::from_timestamp(timestamp, 0)
.expect("test simclock timestamp should be valid"))
})
}
fn sleep(&self, _thread_id: ThreadId, _duration: Duration) -> SleepFuture<'_> {
Box::pin(async { Ok(()) })
}
}
#[tokio::test]
async fn v2_spawn_logs_correlated_send_and_receive_with_simclock_time() -> Result<()> {
let output: &'static Mutex<Vec<u8>> = Box::leak(Box::new(Mutex::new(Vec::new())));
let subscriber = tracing_subscriber::fmt()
.with_ansi(false)
.with_max_level(Level::TRACE)
.with_writer(MockWriter::new(output))
.finish();
let _guard = tracing::subscriber::set_default(subscriber);
let server = start_mock_server().await;
let spawn_message = "inspect the repository";
let spawn_args = serde_json::to_string(&serde_json::json!({
"message": spawn_message,
"task_name": "worker",
}))?;
mount_sse_once_match(
&server,
|request: &Request| body_contains(request, "start"),
sse(vec![
ev_response_created("resp-parent-1"),
ev_function_call_with_namespace(
"spawn-call-1",
"collaboration",
"spawn_agent",
&spawn_args,
),
ev_completed("resp-parent-1"),
]),
)
.await;
let child_request = mount_sse_once_match(
&server,
|request: &Request| request_has_input_type(request, "agent_message"),
sse(vec![
ev_response_created("resp-child-1"),
ev_completed("resp-child-1"),
]),
)
.await;
mount_sse_once_match(
&server,
|request: &Request| {
body_contains(request, "spawn-call-1")
&& !request_has_input_type(request, "agent_message")
},
sse(vec![
ev_response_created("resp-parent-2"),
ev_assistant_message("msg-parent-2", "done"),
ev_completed("resp-parent-2"),
]),
)
.await;
let simclock = Arc::new(ThreadSimClock::default());
let mut builder = test_codex()
.with_external_time_provider(simclock.clone())
.with_config(|config| {
config
.features
.enable(Feature::Collab)
.expect("test config should allow feature update");
config
.features
.enable(Feature::MultiAgentV2)
.expect("test config should allow feature update");
config
.features
.disable(Feature::EnableRequestCompression)
.expect("test config should allow feature update");
});
let test = builder.build(&server).await?;
let root_thread_id = test.session_configured.thread_id;
simclock.set_root_thread_id(root_thread_id);
test.submit_turn("start").await?;
assert!(!child_request.requests().is_empty());
let child_thread_id = test
.thread_manager
.list_thread_ids()
.await
.into_iter()
.find(|thread_id| *thread_id != root_thread_id)
.expect("child thread ID");
let logs = tokio::time::timeout(Duration::from_secs(5), async {
loop {
let logs = String::from_utf8(output.lock().expect("buffer lock").clone())
.expect("logs should be UTF-8");
if logs.contains("kind=\"spawn\"") && logs.contains("state=\"receive\"") {
break logs;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("spawn communication logs should be emitted");
let send = logs
.lines()
.find(|line| line.contains("kind=\"spawn\"") && line.contains("state=\"send\""))
.expect("spawn send event");
assert!(send.contains(&format!("sender_thread_id={root_thread_id}")));
assert!(send.contains(&format!("receiver_thread_id={child_thread_id}")));
assert!(send.contains(&format!("content=\"{spawn_message}\"")));
assert!(send.contains(&format!("simclock_time={ROOT_SIMCLOCK_TIME}")));
let communication_id = log_field(send, "communication_id").expect("communication ID");
let receive = logs
.lines()
.find(|line| {
line.contains("state=\"receive\"")
&& log_field(line, "communication_id") == Some(communication_id)
})
.expect("correlated receive event");
assert!(receive.contains(&format!("simclock_time={CHILD_SIMCLOCK_TIME}")));
Ok(())
}
fn body_contains(request: &Request, text: &str) -> bool {
String::from_utf8_lossy(&request.body).contains(text)
}
fn request_has_input_type(request: &Request, ty: &str) -> bool {
serde_json::from_slice::<serde_json::Value>(&request.body)
.ok()
.and_then(|body| {
body.get("input")
.and_then(serde_json::Value::as_array)
.cloned()
})
.is_some_and(|items| {
items
.iter()
.any(|item| item.get("type").and_then(serde_json::Value::as_str) == Some(ty))
})
}
fn log_field<'a>(line: &'a str, name: &str) -> Option<&'a str> {
let prefix = format!("{name}=");
line.split_ascii_whitespace()
.find_map(|field| field.strip_prefix(&prefix))
.map(|value| value.trim_matches('"'))
}
codex-rs/core/tests/suite/mod.rs
+1
View File
@@ -30,6 +30,7 @@ pub static CODEX_ALIASES_TEMP_DIR: Option<TestBinaryDispatchGuard> = {
#[cfg(not(target_os = "windows"))]
mod abort_tasks;
mod additional_context;
mod agent_communication;
mod agent_execution;
mod agent_jobs;
mod agent_websocket;
codex-rs/feedback/src/lib.rs
+28 -1
View File
@@ -18,6 +18,7 @@ use tracing::Event;
use tracing::Level;
use tracing::field::Visit;
use tracing_subscriber::Layer;
use tracing_subscriber::filter::LevelFilter;
use tracing_subscriber::filter::Targets;
use tracing_subscriber::fmt::writer::MakeWriter;
use tracing_subscriber::registry::LookupSpan;
@@ -204,7 +205,11 @@ impl CodexFeedback {
.with_target(false)
// Capture everything, regardless of the caller's `RUST_LOG`, so feedback includes the
// full trace when the user uploads a report.
.with_filter(Targets::new().with_default(Level::TRACE))
.with_filter(
Targets::new()
.with_default(Level::TRACE)
.with_target("codex_core::agent_communication", LevelFilter::OFF),
)
}
/// Returns a [`tracing_subscriber`] layer that collects structured metadata for feedback.
@@ -723,6 +728,28 @@ mod tests {
pretty_assertions::assert_eq!(snap.tags.get("cached").map(String::as_str), Some("true"));
}
#[test]
fn logger_layer_excludes_agent_communication_content() {
let fb = CodexFeedback::new();
let _guard = tracing_subscriber::registry()
.with(fb.logger_layer())
.set_default();
tracing::info!(target: "codex_core", "retained-log");
tracing::trace!(
target: "codex_core::agent_communication",
content = "secret",
"dropped-log"
);
let logs = std::str::from_utf8(fb.snapshot(/*session_id*/ None).as_bytes())
.expect("feedback logs should be UTF-8")
.to_string();
assert!(logs.contains("retained-log"));
assert!(!logs.contains("secret"));
assert!(!logs.contains("dropped-log"));
}
#[test]
fn feedback_attachments_gate_connectivity_diagnostics() {
let extra_filename = format!("codex-feedback-extra-{}.jsonl", ThreadId::new());
codex-rs/state/src/log_db.rs
+1
View File
@@ -54,6 +54,7 @@ pub fn default_filter() -> Targets {
Targets::new()
.with_default(LevelFilter::TRACE)
.with_target("log", LevelFilter::OFF)
.with_target("codex_core::agent_communication", LevelFilter::OFF)
.with_target("codex_otel.log_only", LevelFilter::OFF)
.with_target("codex_otel.trace_safe", LevelFilter::OFF)
}
codex-rs/state/src/log_db_filter_tests.rs
+6 -6
View File
@@ -1,5 +1,4 @@
use pretty_assertions::assert_eq;
use tracing_subscriber::filter::Targets;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;
use uuid::Uuid;
@@ -16,17 +15,18 @@ async fn sqlite_sink_drops_low_level_opentelemetry_sdk_logs() {
let layer = start(runtime.clone());
let guard = tracing_subscriber::registry()
.with(
layer
.clone()
.with_filter(Targets::new().with_default(tracing::Level::TRACE)),
)
.with(layer.clone().with_filter(default_filter()))
.set_default();
tracing::trace!(target: "opentelemetry_sdk", "dropped-trace");
tracing::debug!(target: "opentelemetry_sdk", "dropped-debug");
tracing::info!(target: "opentelemetry_sdk", "retained-info");
tracing::trace!(target: "codex_state", "retained-trace");
tracing::trace!(
target: "codex_core::agent_communication",
content = "secret",
"dropped-agent-communication"
);
layer.flush().await;
drop(guard);