diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 8de6443402..c80e54fad0 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; @@ -67,6 +69,7 @@ pub(crate) struct SpawnAgentOptions { pub(crate) fork_mode: Option, pub(crate) parent_thread_id: Option, pub(crate) environments: Option>, + pub(crate) agent_communication_context: Option, } #[derive(Clone, Debug)] @@ -144,8 +147,13 @@ impl AgentControl { let state = self.upgrade()?; self.ensure_execution_capacity_for_op(agent_id, &initial_operation) .await?; - self.send_input_after_capacity_check(agent_id, &state, initial_operation) - .await + self.send_input_after_capacity_check( + agent_id, + &state, + initial_operation, + /*agent_communication_context*/ None, + ) + .await } async fn send_input_after_capacity_check( @@ -153,10 +161,19 @@ impl AgentControl { agent_id: ThreadId, state: &Arc, initial_operation: Op, + agent_communication_context: Option, ) -> CodexResult { if let Op::InterAgentCommunication { communication } = initial_operation { + let agent_communication_context = agent_communication_context.ok_or_else(|| { + CodexErr::Fatal("inter-agent communication is missing logging context".to_string()) + })?; return self - .submit_inter_agent_communication(agent_id, state, communication) + .submit_inter_agent_communication( + agent_id, + state, + communication, + agent_communication_context, + ) .await; } @@ -183,9 +200,18 @@ impl AgentControl { &self, agent_id: ThreadId, communication: InterAgentCommunication, + agent_communication_context: AgentCommunicationContext, ) -> CodexResult { - self.send_input(agent_id, Op::InterAgentCommunication { communication }) - .await + let state = self.upgrade()?; + let op = Op::InterAgentCommunication { communication }; + self.ensure_execution_capacity_for_op(agent_id, &op).await?; + self.send_input_after_capacity_check( + agent_id, + &state, + op, + Some(agent_communication_context), + ) + .await } async fn submit_inter_agent_communication( @@ -193,8 +219,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 +249,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 +557,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/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index c4c40577ca..9344dacbb1 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -356,8 +356,13 @@ impl AgentControl { ) .await; - self.send_input_after_capacity_check(new_thread.thread_id, &state, initial_operation) - .await?; + self.send_input_after_capacity_check( + new_thread.thread_id, + &state, + initial_operation, + options.agent_communication_context.clone(), + ) + .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/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs index 204385934e..9dc8bc208e 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs @@ -132,6 +132,7 @@ async fn handle_spawn_agent( fork_mode: args.fork_context.then_some(SpawnAgentForkMode::FullHistory), parent_thread_id: Some(session.thread_id), environments: Some(turn.environments.to_selections()), + agent_communication_context: None, }, )) .await 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..431ccbccd5 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; @@ -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..e1799230cd 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,6 +4,8 @@ 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 +106,36 @@ 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_operation, agent_communication_context) = 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); + (Op::InterAgentCommunication { communication }, Some(context)) + } + initial_operation => (initial_operation, None), + }; + let spawned_agent = Box::pin(session.services.agent_control.spawn_agent_with_metadata( + config, + 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()), + agent_communication_context, + }, + )) .await .map_err(collab_spawn_error)?; let new_thread_id = spawned_agent.thread_id; diff --git a/codex-rs/core/tests/suite/agent_communication.rs b/codex-rs/core/tests/suite/agent_communication.rs new file mode 100644 index 0000000000..9ad17f56f5 --- /dev/null +++ b/codex-rs/core/tests/suite/agent_communication.rs @@ -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>, +} + +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(()) }) + } +} + +#[tokio::test] +async fn v2_spawn_logs_correlated_send_and_receive_with_simclock_time() -> 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 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::(&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('"')) +} diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index 3e27463370..42fc86b18b 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -30,6 +30,7 @@ pub static CODEX_ALIASES_TEMP_DIR: Option = { #[cfg(not(target_os = "windows"))] mod abort_tasks; mod additional_context; +mod agent_communication; mod agent_execution; mod agent_jobs; mod agent_websocket; 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);