diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index 4435a2eb630c..ecbabb2641d8 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -157,7 +157,7 @@ serial_test = { workspace = true } tempfile = { workspace = true } test-log = { workspace = true } tracing-opentelemetry = { workspace = true } -tracing-subscriber = { workspace = true } +tracing-subscriber = { workspace = true, features = ["json"] } tracing-test = { workspace = true, features = ["no-env-filter"] } walkdir = { workspace = true } wiremock = { workspace = true } diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 49bbab0f87a7..ef71341db492 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)] @@ -76,6 +79,11 @@ pub(crate) struct LiveAgent { pub(crate) status: AgentStatus, } +enum SubmissionId<'a> { + Generated, + Provided(&'a str), +} + #[derive(Clone, Debug, Serialize, PartialEq, Eq)] pub(crate) struct ListedAgent { pub(crate) agent_name: String, @@ -144,8 +152,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, + SubmissionId::Generated, + ) + .await } async fn send_input_after_capacity_check( @@ -153,6 +166,7 @@ impl AgentControl { agent_id: ThreadId, state: &Arc, initial_operation: Op, + submission_id: SubmissionId<'_>, ) -> CodexResult { let last_task_message = match &initial_operation { Op::InterAgentCommunication { communication } => { @@ -160,12 +174,16 @@ impl AgentControl { } _ => non_empty_task_message(render_input_preview(&initial_operation)), }; + let send_result = match submission_id { + SubmissionId::Generated => state.send_op(agent_id, initial_operation).await, + SubmissionId::Provided(id) => { + state + .send_op_with_id(agent_id, id.to_string(), initial_operation) + .await + } + }; let result = self - .handle_thread_request_result( - agent_id, - state, - state.send_op(agent_id, initial_operation).await, - ) + .handle_thread_request_result(agent_id, state, send_result) .await; if result.is_ok() { match last_task_message { @@ -182,23 +200,23 @@ impl AgentControl { &self, agent_id: ThreadId, communication: InterAgentCommunication, + context: AgentCommunicationContext, ) -> CodexResult { - let last_task_message = last_task_message_from_communication(&communication); + crate::agent_communication::emit_agent_communication_created( + &context, + &communication, + agent_id, + ); let state = self.upgrade()?; let op = Op::InterAgentCommunication { communication }; self.ensure_execution_capacity_for_op(agent_id, &op).await?; - let result = self - .handle_thread_request_result(agent_id, &state, state.send_op(agent_id, op).await) - .await; - if result.is_ok() { - match last_task_message { - Some(last_task_message) => self - .state - .update_last_task_message(agent_id, last_task_message), - None => self.state.clear_last_task_message(agent_id), - } - } - result + self.send_input_after_capacity_check( + agent_id, + &state, + op, + SubmissionId::Provided(context.id()), + ) + .await } /// Interrupt the current task for an existing agent thread. @@ -480,8 +498,16 @@ impl AgentControl { message, /*trigger_turn*/ false, ); + let communication_context = AgentCommunicationContext::without_source_call( + AgentCommunicationKind::Result, + child_thread_id, + ); let _ = control - .send_inter_agent_communication(parent_thread_id, communication) + .send_inter_agent_communication( + parent_thread_id, + communication, + 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 c4c40577ca4f..96658be44e62 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -356,8 +356,26 @@ impl AgentControl { ) .await; - self.send_input_after_capacity_check(new_thread.thread_id, &state, initial_operation) - .await?; + if let (Some(context), Op::InterAgentCommunication { communication }) = ( + options.agent_communication_context.as_ref(), + &initial_operation, + ) { + crate::agent_communication::emit_agent_communication_created( + context, + communication, + new_thread.thread_id, + ); + } + self.send_input_after_capacity_check( + new_thread.thread_id, + &state, + initial_operation, + match options.agent_communication_context.as_ref() { + Some(context) => SubmissionId::Provided(context.id()), + None => SubmissionId::Generated, + }, + ) + .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 9863bfb7e24c..a84f21c204b7 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -3,6 +3,7 @@ use crate::CodexThread; use crate::StateDbHandle; use crate::ThreadManager; use crate::agent::agent_status_from_event; +use crate::agent_communication::AgentCommunicationKind; use crate::config::AgentRoleConfig; use crate::config::Config; use crate::config::ConfigBuilder; @@ -431,6 +432,63 @@ async fn send_input_errors_when_thread_missing() { assert_matches!(err, CodexErr::ThreadNotFound(id) if id == thread_id); } +#[tokio::test] +async fn failed_communication_send_emits_created_without_enqueued() { + use tracing_subscriber::filter::LevelFilter; + use tracing_subscriber::filter::Targets; + use tracing_subscriber::layer::SubscriberExt; + use tracing_test::internal::MockWriter; + + let output: &'static std::sync::Mutex> = + Box::leak(Box::new(std::sync::Mutex::new(Vec::new()))); + let subscriber = tracing_subscriber::registry() + .with( + Targets::new() + .with_default(LevelFilter::OFF) + .with_target("codex_core::agent_communication", LevelFilter::TRACE), + ) + .with( + tracing_subscriber::fmt::layer() + .json() + .with_current_span(false) + .with_span_list(false) + .with_writer(MockWriter::new(output)), + ); + let _guard = tracing::subscriber::set_default(subscriber); + + let harness = AgentControlHarness::new().await; + let sender_thread_id = ThreadId::new(); + let receiver_thread_id = ThreadId::new(); + let context = crate::agent_communication::AgentCommunicationContext::from_tool_call( + AgentCommunicationKind::Message, + sender_thread_id, + "call-1", + ); + let id = context.id().to_string(); + let communication = InterAgentCommunication::new( + AgentPath::root(), + AgentPath::try_from("/root/missing").expect("agent path"), + Vec::new(), + "mail".to_string(), + /*trigger_turn*/ false, + ); + let err = harness + .control + .send_inter_agent_communication(receiver_thread_id, communication, context) + .await + .expect_err("send should fail for missing receiver"); + assert_matches!(err, CodexErr::ThreadNotFound(id) if id == receiver_thread_id); + + let events = String::from_utf8(output.lock().expect("buffer lock").clone()) + .expect("JSON logs should be UTF-8") + .lines() + .map(|line| serde_json::from_str::(line).expect("valid JSON log event")) + .collect::>(); + assert_eq!(events.len(), 1); + assert_eq!(events[0]["fields"]["communication_id"], id); + assert_eq!(events[0]["fields"]["state"], "created"); +} + #[tokio::test] async fn get_status_returns_not_found_for_missing_thread() { let harness = AgentControlHarness::new().await; @@ -529,12 +587,17 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig /*trigger_turn*/ false, ); + let context = crate::agent_communication::AgentCommunicationContext::without_source_call( + AgentCommunicationKind::Message, + ThreadId::new(), + ); + let communication_id = context.id().to_string(); let submission_id = harness .control - .send_inter_agent_communication(thread_id, communication.clone()) + .send_inter_agent_communication(thread_id, communication.clone(), context) .await .expect("send_inter_agent_communication should succeed"); - assert!(!submission_id.is_empty()); + assert_eq!(submission_id, communication_id); let expected = ( thread_id, @@ -656,7 +719,14 @@ 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(), + crate::agent_communication::AgentCommunicationContext::without_source_call( + AgentCommunicationKind::Message, + ThreadId::new(), + ), + ) .await .expect("send_inter_agent_communication should succeed after reload"); let expected = ( @@ -803,7 +873,14 @@ 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, + crate::agent_communication::AgentCommunicationContext::without_source_call( + AgentCommunicationKind::Followup, + ThreadId::new(), + ), + ) .await .expect("send_inter_agent_communication should succeed"); @@ -2112,28 +2189,26 @@ async fn multi_agent_v2_completion_queues_message_for_direct_parent() { &AgentStatus::Completed(Some("done".to_string())), ) .expect("completed status should render"); - let expected = ( - worker_thread_id, - Op::InterAgentCommunication { - communication: InterAgentCommunication::new( - tester_path.clone(), - worker_path.clone(), - Vec::new(), - expected_message.clone(), - /*trigger_turn*/ false, - ), - }, + let expected = InterAgentCommunication::new( + tester_path.clone(), + worker_path.clone(), + Vec::new(), + expected_message.clone(), + /*trigger_turn*/ false, ); timeout(Duration::from_secs(5), async { loop { - let captured = harness - .manager - .captured_ops() - .into_iter() - .find(|entry| *entry == expected); - if captured == Some(expected.clone()) { - break; + for (thread_id, op) in harness.manager.captured_ops() { + if thread_id != worker_thread_id { + continue; + } + let Op::InterAgentCommunication { communication } = op else { + continue; + }; + if communication == expected { + return; + } } sleep(Duration::from_millis(10)).await; } diff --git a/codex-rs/core/src/agent_communication.rs b/codex-rs/core/src/agent_communication.rs new file mode 100644 index 000000000000..571af936e219 --- /dev/null +++ b/codex-rs/core/src/agent_communication.rs @@ -0,0 +1,110 @@ +use codex_protocol::ThreadId; +use codex_protocol::protocol::InterAgentCommunication; +use uuid::Uuid; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum AgentCommunicationKind { + InitialTask, + Message, + Followup, + Result, +} + +impl AgentCommunicationKind { + fn as_str(self) -> &'static str { + match self { + Self::InitialTask => "initialTask", + Self::Message => "message", + Self::Followup => "followup", + Self::Result => "result", + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct AgentCommunicationContext { + id: String, + kind: AgentCommunicationKind, + sender_thread_id: ThreadId, + source_call_id: Option, +} + +impl AgentCommunicationContext { + pub(crate) fn from_tool_call( + kind: AgentCommunicationKind, + sender_thread_id: ThreadId, + source_call_id: &str, + ) -> Self { + Self::new(kind, sender_thread_id, Some(source_call_id.to_string())) + } + + pub(crate) fn without_source_call( + kind: AgentCommunicationKind, + sender_thread_id: ThreadId, + ) -> Self { + Self::new(kind, sender_thread_id, None) + } + + fn new( + kind: AgentCommunicationKind, + sender_thread_id: ThreadId, + source_call_id: Option, + ) -> Self { + Self { + id: Uuid::now_v7().to_string(), + kind, + sender_thread_id, + source_call_id, + } + } + + pub(crate) fn id(&self) -> &str { + &self.id + } +} + +pub(crate) fn emit_agent_communication_created( + context: &AgentCommunicationContext, + communication: &InterAgentCommunication, + receiver_thread_id: ThreadId, +) { + tracing::trace!( + { + event.name = "codex.agent_communication", + communication_id = %context.id, + kind = context.kind.as_str(), + state = "created", + sender_thread_id = %context.sender_thread_id, + receiver_thread_id = %receiver_thread_id, + content = communication_content(communication), + source_call_id = context.source_call_id.as_deref(), + }, + "agent communication updated" + ); +} + +pub(crate) fn emit_agent_communication_enqueued(id: &str) { + tracing::trace!( + { + event.name = "codex.agent_communication", + communication_id = id, + state = "enqueued", + }, + "agent communication updated" + ); +} + +pub(crate) fn communication_content(communication: &InterAgentCommunication) -> &str { + if communication.content.is_empty() { + communication + .encrypted_content + .as_deref() + .unwrap_or_default() + } else { + &communication.content + } +} + +#[cfg(test)] +#[path = "agent_communication_tests.rs"] +mod tests; diff --git a/codex-rs/core/src/agent_communication_tests.rs b/codex-rs/core/src/agent_communication_tests.rs new file mode 100644 index 000000000000..ddd6e0ab35eb --- /dev/null +++ b/codex-rs/core/src/agent_communication_tests.rs @@ -0,0 +1,120 @@ +use super::*; +use codex_protocol::AgentPath; +use pretty_assertions::assert_eq; +use serde_json::Value; +use serde_json::json; +use tracing_subscriber::filter::LevelFilter; +use tracing_subscriber::filter::Targets; +use tracing_subscriber::layer::SubscriberExt; +use tracing_test::internal::MockWriter; + +#[test] +fn emits_structured_lifecycle_events_at_trace() { + let output: &'static std::sync::Mutex> = + Box::leak(Box::new(std::sync::Mutex::new(Vec::new()))); + let filter = Targets::new() + .with_default(LevelFilter::WARN) + .with_target("codex_core::agent_communication", LevelFilter::TRACE); + let subscriber = tracing_subscriber::registry().with(filter).with( + tracing_subscriber::fmt::layer() + .json() + .with_current_span(false) + .with_span_list(false) + .with_writer(MockWriter::new(output)), + ); + let _guard = tracing::subscriber::set_default(subscriber); + + let sender_thread_id = ThreadId::new(); + let receiver_thread_id = ThreadId::new(); + let communication = InterAgentCommunication::new( + AgentPath::root(), + AgentPath::root(), + Vec::new(), + "hello".to_string(), + /*trigger_turn*/ false, + ); + let context = AgentCommunicationContext::from_tool_call( + AgentCommunicationKind::Message, + sender_thread_id, + "call-1", + ); + emit_agent_communication_created(&context, &communication, receiver_thread_id); + emit_agent_communication_enqueued(context.id()); + + let result_context = AgentCommunicationContext::without_source_call( + AgentCommunicationKind::Result, + receiver_thread_id, + ); + emit_agent_communication_created(&result_context, &communication, sender_thread_id); + + let events = String::from_utf8(output.lock().expect("buffer lock").clone()) + .expect("JSON logs should be UTF-8") + .lines() + .map(|line| serde_json::from_str::(line).expect("valid JSON log event")) + .collect::>(); + assert_eq!(events.len(), 3); + assert_eq!(events[0]["level"], "TRACE"); + assert_eq!(events[0]["target"], "codex_core::agent_communication"); + assert_eq!( + events[0]["fields"], + json!({ + "message": "agent communication updated", + "event.name": "codex.agent_communication", + "communication_id": context.id(), + "kind": "message", + "state": "created", + "sender_thread_id": sender_thread_id.to_string(), + "receiver_thread_id": receiver_thread_id.to_string(), + "content": "hello", + "source_call_id": "call-1", + }) + ); + assert_eq!(events[1]["level"], "TRACE"); + assert_eq!(events[1]["target"], "codex_core::agent_communication"); + assert_eq!( + events[1]["fields"], + json!({ + "message": "agent communication updated", + "event.name": "codex.agent_communication", + "communication_id": context.id(), + "state": "enqueued", + }) + ); + assert_eq!( + events[2]["fields"], + json!({ + "message": "agent communication updated", + "event.name": "codex.agent_communication", + "communication_id": result_context.id(), + "kind": "result", + "state": "created", + "sender_thread_id": receiver_thread_id.to_string(), + "receiver_thread_id": sender_thread_id.to_string(), + "content": "hello", + }) + ); +} + +#[test] +fn content_prefers_plaintext_and_falls_back_to_encrypted_content() { + let plaintext = InterAgentCommunication { + encrypted_content: Some("encrypted".to_string()), + ..InterAgentCommunication::new( + AgentPath::root(), + AgentPath::root(), + Vec::new(), + "plain".to_string(), + /*trigger_turn*/ false, + ) + }; + let encrypted = InterAgentCommunication::new_encrypted( + AgentPath::root(), + AgentPath::root(), + Vec::new(), + "encrypted".to_string(), + /*trigger_turn*/ false, + ); + + let contents = [&plaintext, &encrypted].map(communication_content); + assert_eq!(contents, ["plain", "encrypted"]); +} diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index e9b749fea949..df8753a38641 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 a483ce5e0f75..f4b2c72f761f 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -285,6 +285,7 @@ pub async fn inter_agent_communication( sess.input_queue .enqueue_mailbox_communication(communication) .await; + crate::agent_communication::emit_agent_communication_enqueued(&sub_id); 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 f45616eed824..938698723128 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -14,6 +14,7 @@ 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::AgentCommunicationKind; use crate::attestation::AttestationProvider; use crate::build_available_skills; use crate::compact; @@ -892,7 +893,7 @@ impl Codex { /// /// Some use cases take advantage of the fact that these are UUID7 which /// encodes a timestamp, so think carefully before changing this. -fn new_submission_id() -> String { +pub(crate) fn new_submission_id() -> String { Uuid::now_v7().to_string() } @@ -1867,10 +1868,15 @@ impl Session { message, /*trigger_turn*/ false, ); + let communication_context = + crate::agent_communication::AgentCommunicationContext::without_source_call( + 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, communication_context) .await { debug!("failed to notify parent thread {parent_thread_id}: {err}"); diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index ece9d446b57e..aa36863ec507 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -9781,6 +9781,58 @@ async fn task_finish_emits_thread_idle_lifecycle_after_active_turn_clears() { assert!(session.active_turn.lock().await.is_none()); } +#[tokio::test] +async fn mailbox_enqueue_emits_enqueued_communication_with_same_id() { + use tracing_subscriber::filter::LevelFilter; + use tracing_subscriber::filter::Targets; + use tracing_subscriber::layer::SubscriberExt; + use tracing_test::internal::MockWriter; + + let output: &'static std::sync::Mutex> = + Box::leak(Box::new(std::sync::Mutex::new(Vec::new()))); + let subscriber = tracing_subscriber::registry() + .with( + Targets::new() + .with_default(LevelFilter::OFF) + .with_target("codex_core::agent_communication", LevelFilter::TRACE), + ) + .with( + tracing_subscriber::fmt::layer() + .json() + .with_current_span(false) + .with_span_list(false) + .with_writer(MockWriter::new(output)), + ); + let _guard = tracing::subscriber::set_default(subscriber); + + let (session, _) = make_session_and_context().await; + let id = Uuid::new_v4().to_string(); + let communication = InterAgentCommunication::new( + AgentPath::root(), + AgentPath::root(), + Vec::new(), + "mail".to_string(), + /*trigger_turn*/ false, + ); + super::handlers::inter_agent_communication(&Arc::new(session), id.clone(), communication).await; + + let events = String::from_utf8(output.lock().expect("buffer lock").clone()) + .expect("JSON logs should be UTF-8") + .lines() + .map(|line| serde_json::from_str::(line).expect("valid JSON log event")) + .collect::>(); + assert_eq!(events.len(), 1); + assert_eq!( + events[0]["fields"], + serde_json::json!({ + "message": "agent communication updated", + "event.name": "codex.agent_communication", + "communication_id": id, + "state": "enqueued", + }) + ); +} + #[tokio::test] async fn thread_idle_lifecycle_waits_for_trigger_turn_mailbox_work() { struct ThreadIdleRecorder { diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 4ea86cb33d00..4f2e4c261bb4 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -13,6 +13,7 @@ use crate::session::Codex; use crate::session::CodexSpawnArgs; use crate::session::CodexSpawnOk; use crate::session::INITIAL_SUBMIT_ID; +use crate::session::new_submission_id; use crate::session::resolve_multi_agent_version; use crate::tasks::InterruptedTurnHistoryMarker; use crate::tasks::interrupted_turn_history_marker; @@ -56,6 +57,7 @@ use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::SessionConfiguredEvent; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; +use codex_protocol::protocol::Submission; use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadSource; use codex_protocol::protocol::TurnAbortReason; @@ -1146,13 +1148,31 @@ impl ThreadManagerState { /// Send an operation to a thread by ID. pub(crate) async fn send_op(&self, thread_id: ThreadId, op: Op) -> CodexResult { + self.send_op_with_id(thread_id, new_submission_id(), op) + .await + } + + pub(crate) async fn send_op_with_id( + &self, + thread_id: ThreadId, + id: String, + op: Op, + ) -> CodexResult { let thread = self.get_thread(thread_id).await?; if let Some(ops_log) = &self.ops_log && let Ok(mut log) = ops_log.lock() { log.push((thread_id, op.clone())); } - thread.submit(op).await + thread + .submit_with_id(Submission { + id: id.clone(), + op, + client_user_message_id: None, + trace: None, + }) + .await?; + Ok(id) } /// Remove a thread from the manager by ID, returning it when present. 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 204385934e2b..9dc8bc208e53 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 7fdb80e8bcb2..092a0abfaeaa 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,7 @@ //! resulting `InterAgentCommunication` should wake the target immediately. use super::*; +use crate::agent_communication::AgentCommunicationKind; use crate::tools::context::FunctionToolOutput; use crate::turn_timing::now_unix_timestamp_ms; use codex_protocol::protocol::InterAgentCommunication; @@ -88,6 +89,23 @@ pub(crate) async fn handle_message_string_tool( let receiver_agent_path = receiver_agent.agent_path.clone().ok_or_else(|| { FunctionCallError::RespondToModel("target agent is missing an agent_path".to_string()) })?; + let author = turn + .session_source + .get_agent_path() + .unwrap_or_else(AgentPath::root); + let communication = + communication_from_tool_message(author, receiver_agent_path.clone(), message); + let communication_kind = match mode { + MessageDeliveryMode::QueueOnly => AgentCommunicationKind::Message, + MessageDeliveryMode::TriggerTurn => AgentCommunicationKind::Followup, + }; + let communication_context = + crate::agent_communication::AgentCommunicationContext::from_tool_call( + communication_kind, + session.thread_id, + call_id.as_str(), + ); + let resume_config = build_agent_resume_config(turn.as_ref())?; session .services @@ -95,16 +113,14 @@ pub(crate) async fn handle_message_string_tool( .ensure_v2_agent_loaded(resume_config, receiver_thread_id) .await .map_err(|err| collab_agent_error(receiver_thread_id, err))?; - let author = turn - .session_source - .get_agent_path() - .unwrap_or_else(AgentPath::root); - let communication = - communication_from_tool_message(author, receiver_agent_path.clone(), message); 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), + 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 2186a4a6d605..a34be0076989 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,7 @@ 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::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 +105,39 @@ 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 = crate::agent_communication::AgentCommunicationContext::from_tool_call( + AgentCommunicationKind::InitialTask, + session.thread_id, + call_id.as_str(), + ); + (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/feedback/src/lib.rs b/codex-rs/feedback/src/lib.rs index 7c27d2b3b0b3..c12a06a8f2f3 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,24 @@ 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 2cf0f5d9af42..55ef26eb7601 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 36ee27aba8a1..aef1ee97c226 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,14 @@ 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);