From 81c79f11b8f170c093223fdb3bd176cff81d3e3f Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Wed, 1 Jul 2026 15:08:50 -0700 Subject: [PATCH] Log multi-agent communication lifecycle --- codex-rs/core/src/agent/control.rs | 71 ++++++++-- codex-rs/core/src/agent/control/execution.rs | 11 +- codex-rs/core/src/agent/control/spawn.rs | 65 ++++++++- codex-rs/core/src/agent/control_tests.rs | 20 ++- codex-rs/core/src/agent_communication.rs | 96 +++++++++++++ codex-rs/core/src/lib.rs | 1 + codex-rs/core/src/session/handlers.rs | 13 ++ codex-rs/core/src/session/mod.rs | 6 +- codex-rs/core/src/session/session.rs | 4 + codex-rs/core/src/session/tests.rs | 2 + codex-rs/core/src/state/service.rs | 1 + .../src/tools/handlers/multi_agents_v2.rs | 1 - .../handlers/multi_agents_v2/message_tool.rs | 11 +- .../tools/handlers/multi_agents_v2/spawn.rs | 54 ++++---- .../tests/suite/subagent_notifications.rs | 126 ++++++++++++++++-- codex-rs/feedback/src/lib.rs | 29 +++- codex-rs/state/src/log_db.rs | 1 + codex-rs/state/src/log_db_filter_tests.rs | 12 +- 18 files changed, 455 insertions(+), 69 deletions(-) create mode 100644 codex-rs/core/src/agent_communication.rs diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 8de6443402..07d3624052 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -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; @@ -154,12 +156,6 @@ impl AgentControl { state: &Arc, initial_operation: Op, ) -> CodexResult { - if let Op::InterAgentCommunication { communication } = initial_operation { - return self - .submit_inter_agent_communication(agent_id, state, communication) - .await; - } - let last_task_message = non_empty_task_message(render_input_preview(&initial_operation)); let result = self .handle_thread_request_result( @@ -183,8 +179,28 @@ impl AgentControl { &self, agent_id: ThreadId, communication: InterAgentCommunication, + agent_communication_context: AgentCommunicationContext, ) -> CodexResult { - self.send_input(agent_id, Op::InterAgentCommunication { communication }) + let state = self.upgrade()?; + self.ensure_execution_capacity_for_turn_start(agent_id, communication.trigger_turn) + .await?; + self.send_inter_agent_communication_after_capacity_check( + agent_id, + &state, + communication, + agent_communication_context, + ) + .await + } + + async fn send_inter_agent_communication_after_capacity_check( + &self, + agent_id: ThreadId, + state: &Arc, + communication: InterAgentCommunication, + context: AgentCommunicationContext, + ) -> CodexResult { + self.submit_inter_agent_communication(agent_id, state, communication, context) .await } @@ -193,8 +209,27 @@ impl AgentControl { agent_id: ThreadId, state: &Arc, communication: InterAgentCommunication, + context: AgentCommunicationContext, ) -> CodexResult { 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 +239,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 +547,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; } diff --git a/codex-rs/core/src/agent/control/execution.rs b/codex-rs/core/src/agent/control/execution.rs index 42aaa50e02..d8559c1499 100644 --- a/codex-rs/core/src/agent/control/execution.rs +++ b/codex-rs/core/src/agent/control/execution.rs @@ -32,7 +32,16 @@ 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(super) 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()?; diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index c4c40577ca..20fb61b601 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -9,6 +9,14 @@ struct SpawnAgentThreadInheritance { exec_policy: Option>, } +enum SpawnInitialInput { + Op(Box), + InterAgentCommunication { + communication: InterAgentCommunication, + context: AgentCommunicationContext, + }, +} + fn default_agent_nickname_list() -> Vec<&'static str> { AGENT_NAMES .lines() @@ -94,7 +102,7 @@ impl AgentControl { ) -> CodexResult { let spawned_agent = Box::pin(self.spawn_agent_internal( config, - initial_operation, + SpawnInitialInput::Op(Box::new(initial_operation)), session_source, SpawnAgentOptions::default(), )) @@ -110,8 +118,33 @@ impl AgentControl { session_source: Option, options: SpawnAgentOptions, // TODO(jif) drop with new fork. ) -> CodexResult { - Box::pin(self.spawn_agent_internal(config, initial_operation, session_source, options)) - .await + Box::pin(self.spawn_agent_internal( + config, + SpawnInitialInput::Op(Box::new(initial_operation)), + session_source, + options, + )) + .await + } + + pub(crate) async fn spawn_agent_with_communication( + &self, + config: Config, + communication: InterAgentCommunication, + context: AgentCommunicationContext, + session_source: Option, + options: SpawnAgentOptions, + ) -> CodexResult { + Box::pin(self.spawn_agent_internal( + config, + SpawnInitialInput::InterAgentCommunication { + communication, + context, + }, + session_source, + options, + )) + .await } pub(crate) async fn ensure_v2_agent_loaded( @@ -197,7 +230,7 @@ impl AgentControl { async fn spawn_agent_internal( &self, config: Config, - initial_operation: Op, + initial_input: SpawnInitialInput, session_source: Option, options: SpawnAgentOptions, ) -> CodexResult { @@ -356,8 +389,28 @@ impl AgentControl { ) .await; - self.send_input_after_capacity_check(new_thread.thread_id, &state, initial_operation) - .await?; + match initial_input { + SpawnInitialInput::Op(initial_operation) => { + self.send_input_after_capacity_check( + new_thread.thread_id, + &state, + *initial_operation, + ) + .await?; + } + SpawnInitialInput::InterAgentCommunication { + communication, + context, + } => { + self.send_inter_agent_communication_after_capacity_check( + new_thread.thread_id, + &state, + communication, + context, + ) + .await?; + } + } if multi_agent_version != MultiAgentVersion::V2 { let child_reference = agent_metadata .agent_path diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 9863bfb7e2..78362bd432 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -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()); @@ -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 = ( @@ -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"); diff --git a/codex-rs/core/src/agent_communication.rs b/codex-rs/core/src/agent_communication.rs new file mode 100644 index 0000000000..89de549f5b --- /dev/null +++ b/codex-rs/core/src/agent_communication.rs @@ -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, +) { + 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, + thread_id: ThreadId, +) -> Option { + 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) { + tracing::trace!( + target: AGENT_COMMUNICATION_TARGET, + { + event.name = "codex.agent_communication", + communication_id, + state = "receive", + simclock_time, + }, + "agent communication" + ); +} diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index e9b749fea9..df8753a386 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -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; diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index a483ce5e0f..e599de8a24 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -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; diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index d43724e5db..809fa92015 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -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}"); diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 5e7046b148..f304865923 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -557,6 +557,9 @@ impl Session { .effective_agent_max_threads(MultiAgentVersion::V2) .unwrap_or(usize::MAX), ); + let simclock_time_provider: Arc = 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) { diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index ece9d446b5..77c95a586e 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -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, diff --git a/codex-rs/core/src/state/service.rs b/codex-rs/core/src/state/service.rs index a611b94887..7a89a8fd93 100644 --- a/codex-rs/core/src/state/service.rs +++ b/codex-rs/core/src/state/service.rs @@ -94,6 +94,7 @@ pub(crate) struct SessionServices { pub(crate) thread_store: Arc, pub(crate) attestation_provider: Option>, pub(crate) time_provider: Arc, + pub(crate) simclock_time_provider: Arc, /// Session-scoped model client shared across turns. pub(crate) model_client: ModelClient, pub(crate) code_mode_service: CodeModeService, diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2.rs index cae3497a13..8e7f73a762 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2.rs @@ -19,7 +19,6 @@ use codex_protocol::protocol::CollabWaitingEndEvent; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::SubAgentActivityEvent; use codex_protocol::protocol::SubAgentActivityKind; -use codex_protocol::user_input::UserInput; use codex_tools::ToolName; use serde::Deserialize; use serde::Serialize; diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs index 7fdb80e8bc..9501b4355a 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs @@ -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; @@ -46,7 +48,7 @@ pub(crate) struct FollowupTaskArgs { pub(crate) message: String, } -fn message_content(message: String) -> Result { +pub(super) fn message_content(message: String) -> Result { if message.trim().is_empty() { return Err(FunctionCallError::RespondToModel( "Empty message can't be sent to an agent".to_string(), @@ -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?; diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs index 2186a4a6d6..8eeb6ddd1b 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs @@ -4,11 +4,13 @@ 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::tools::handlers::multi_agents_v2::message_tool::message_content; use crate::turn_timing::now_unix_timestamp_ms; use codex_protocol::AgentPath; -use codex_protocol::protocol::Op; use codex_tools::ToolSpec; #[derive(Default)] @@ -55,8 +57,7 @@ async fn handle_spawn_agent( .map(str::trim) .filter(|role| !role.is_empty()); - let message = args.message.clone(); - let initial_operation = parse_collab_input(Some(args.message), /*items*/ None)?; + let message = message_content(args.message)?; let session_source = turn.session_source.clone(); let child_depth = next_thread_spawn_depth(&session_source); let mut config = @@ -104,33 +105,28 @@ async fn handle_spawn_agent( "spawned agent is missing a canonical task name".to_string(), ) })?; + 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); 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()), - }, - ), + session + .services + .agent_control + .spawn_agent_with_communication( + config, + communication, + context, + 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)?; diff --git a/codex-rs/core/tests/suite/subagent_notifications.rs b/codex-rs/core/tests/suite/subagent_notifications.rs index 951fb92f17..f8417abd63 100644 --- a/codex-rs/core/tests/suite/subagent_notifications.rs +++ b/codex-rs/core/tests/suite/subagent_notifications.rs @@ -1,6 +1,11 @@ use anyhow::Result; +use chrono::DateTime; +use chrono::Utc; +use codex_core::SleepFuture; use codex_core::StartThreadOptions; use codex_core::ThreadConfigSnapshot; +use codex_core::TimeFuture; +use codex_core::TimeProvider; use codex_core::config::AgentRoleConfig; use codex_features::Feature; use codex_protocol::ThreadId; @@ -39,10 +44,14 @@ use serde_json::Value; use serde_json::json; use std::fs; use std::path::Path; +use std::sync::Arc; +use std::sync::Mutex; use std::time::Duration; use test_case::test_case; use tokio::time::Instant; use tokio::time::sleep; +use tracing::Level; +use tracing_test::internal::MockWriter; use wiremock::MockServer; const SPAWN_CALL_ID: &str = "spawn-call-1"; @@ -61,6 +70,42 @@ const ROLE_REASONING_EFFORT: ReasoningEffort = ReasoningEffort::High; const SUBAGENT_START_CONTEXT: &str = "subagent start context reaches child"; const SUBAGENT_STOP_CONTINUATION: &str = "continue only the child"; const INTERNAL_SUBAGENT_PROMPT: &str = "internal subagent: review"; +const ROOT_SIMCLOCK_TIME: i64 = 1_000; +const CHILD_SIMCLOCK_TIME: i64 = 2_000; + +#[derive(Default)] +struct ThreadSimClock { + root_thread_id: Mutex>, +} + +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::::from_timestamp(timestamp, 0) + .expect("test simclock timestamp should be valid")) + }) + } + + fn sleep(&self, _thread_id: ThreadId, _duration: Duration) -> SleepFuture<'_> { + Box::pin(async { Ok(()) }) + } +} fn body_contains(req: &wiremock::Request, text: &str) -> bool { decoded_body(req) @@ -96,6 +141,13 @@ fn decoded_body(req: &wiremock::Request) -> Option> { } } +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('"')) +} + fn has_subagent_notification(req: &ResponsesRequest) -> bool { req.message_input_texts("user") .iter() @@ -1042,8 +1094,16 @@ async fn spawned_multi_agent_v2_child_inherits_parent_developer_context() -> Res Ok(()) } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[tokio::test] async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result<()> { + let output: &'static Mutex> = 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 encrypted_message = "opaque-encrypted-message"; let spawn_args = serde_json::to_string(&json!({ @@ -1087,17 +1147,23 @@ async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result ) .await; - let mut builder = test_codex().with_model("koffing").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"); - }); + let simclock = Arc::new(ThreadSimClock::default()); + let mut builder = test_codex() + .with_model("koffing") + .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"); + }); 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(TURN_1_PROMPT).await?; @@ -1124,6 +1190,44 @@ async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result })]) ); + 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=\"{encrypted_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(()) } diff --git a/codex-rs/feedback/src/lib.rs b/codex-rs/feedback/src/lib.rs index 7c27d2b3b0..a34f9e6ced 100644 --- a/codex-rs/feedback/src/lib.rs +++ b/codex-rs/feedback/src/lib.rs @@ -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()); diff --git a/codex-rs/state/src/log_db.rs b/codex-rs/state/src/log_db.rs index 2cf0f5d9af..55ef26eb76 100644 --- a/codex-rs/state/src/log_db.rs +++ b/codex-rs/state/src/log_db.rs @@ -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) } diff --git a/codex-rs/state/src/log_db_filter_tests.rs b/codex-rs/state/src/log_db_filter_tests.rs index 36ee27aba8..4939e057cc 100644 --- a/codex-rs/state/src/log_db_filter_tests.rs +++ b/codex-rs/state/src/log_db_filter_tests.rs @@ -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);