openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-22 17:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Log multi-agent communication lifecycle

Browse Source
This commit is contained in:
Michael Bolin
2026-07-01 15:52:28 -07:00
parent f6dcc0e68b
commit 81c79f11b8
18 changed files with 455 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control.rs
+63 -8
View File
@@ -4,6 +4,8 @@ use crate::agent::registry::AgentRegistry;
use crate::agent::role::DEFAULT_ROLE_NAME;
use crate::agent::role::resolve_role_config;
use crate::agent::status::is_final;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::codex_thread::ThreadConfigSnapshot;
use crate::config::Config;
use crate::config::RolloutBudgetConfig;
@@ -154,12 +156,6 @@ impl AgentControl {
state: &Arc<ThreadManagerState>,
initial_operation: Op,
) -> CodexResult<String> {
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<String> {
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<ThreadManagerState>,
communication: InterAgentCommunication,
context: AgentCommunicationContext,
) -> CodexResult<String> {
self.submit_inter_agent_communication(agent_id, state, communication, context)
.await
}
@@ -193,8 +209,27 @@ impl AgentControl {
agent_id: ThreadId,
state: &Arc<ThreadManagerState>,
communication: InterAgentCommunication,
context: AgentCommunicationContext,
) -> CodexResult<String> {
let last_task_message = last_task_message_from_communication(&communication);
let send_log = if crate::agent_communication::logging_enabled() {
let clock_thread = match state.get_thread(context.sender_thread_id).await {
Ok(thread) => Some(thread),
Err(_) => state.get_thread(agent_id).await.ok(),
};
clock_thread.map(|thread| {
let services = &thread.codex.session.services;
let runtime_handle = services.runtime_handle.clone();
let simclock_time =
runtime_handle.spawn(crate::agent_communication::read_simclock_time(
Arc::clone(&services.simclock_time_provider),
context.sender_thread_id,
));
(runtime_handle, simclock_time, communication.clone())
})
} else {
None
};
let result = self
.handle_thread_request_result(
agent_id,
@@ -204,6 +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;
}
codex-rs/core/src/agent/control/execution.rs
+10 -1
View File
@@ -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()?;
codex-rs/core/src/agent/control/spawn.rs
+59 -6
View File
@@ -9,6 +9,14 @@ struct SpawnAgentThreadInheritance {
exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
}
enum SpawnInitialInput {
Op(Box<Op>),
InterAgentCommunication {
communication: InterAgentCommunication,
context: AgentCommunicationContext,
},
}
fn default_agent_nickname_list() -> Vec<&'static str> {
AGENT_NAMES
.lines()
@@ -94,7 +102,7 @@ impl AgentControl {
) -> CodexResult<ThreadId> {
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<SessionSource>,
options: SpawnAgentOptions, // TODO(jif) drop with new fork.
) -> CodexResult<LiveAgent> {
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<SessionSource>,
options: SpawnAgentOptions,
) -> CodexResult<LiveAgent> {
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<SessionSource>,
options: SpawnAgentOptions,
) -> CodexResult<LiveAgent> {
@@ -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
codex-rs/core/src/agent/control_tests.rs
+17 -3
View File
@@ -3,6 +3,8 @@ use crate::CodexThread;
use crate::StateDbHandle;
use crate::ThreadManager;
use crate::agent::agent_status_from_event;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::config::AgentRoleConfig;
use crate::config::Config;
use crate::config::ConfigBuilder;
@@ -531,7 +533,11 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig
let submission_id = harness
.control
.send_inter_agent_communication(thread_id, communication.clone())
.send_inter_agent_communication(
thread_id,
communication.clone(),
AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()),
)
.await
.expect("send_inter_agent_communication should succeed");
assert!(!submission_id.is_empty());
@@ -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");
codex-rs/core/src/agent_communication.rs
+96
View File
@@ -0,0 +1,96 @@
use codex_protocol::ThreadId;
use codex_protocol::protocol::InterAgentCommunication;
use std::sync::Arc;
use crate::current_time::TimeProvider;
const AGENT_COMMUNICATION_TARGET: &str = "codex_core::agent_communication";
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum AgentCommunicationKind {
Spawn,
Message,
Followup,
Result,
}
impl AgentCommunicationKind {
fn as_str(self) -> &'static str {
match self {
Self::Spawn => "spawn",
Self::Message => "message",
Self::Followup => "followup",
Self::Result => "result",
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct AgentCommunicationContext {
kind: AgentCommunicationKind,
pub(crate) sender_thread_id: ThreadId,
}
impl AgentCommunicationContext {
pub(crate) fn new(kind: AgentCommunicationKind, sender_thread_id: ThreadId) -> Self {
Self {
kind,
sender_thread_id,
}
}
}
pub(crate) fn logging_enabled() -> bool {
tracing::enabled!(target: AGENT_COMMUNICATION_TARGET, tracing::Level::TRACE)
}
pub(crate) fn emit_agent_communication_send(
communication_id: &str,
context: &AgentCommunicationContext,
communication: &InterAgentCommunication,
receiver_thread_id: ThreadId,
simclock_time: Option<i64>,
) {
tracing::trace!(
target: AGENT_COMMUNICATION_TARGET,
{
event.name = "codex.agent_communication",
communication_id,
kind = context.kind.as_str(),
state = "send",
sender_thread_id = %context.sender_thread_id,
receiver_thread_id = %receiver_thread_id,
content = if communication.content.is_empty() {
communication.encrypted_content.as_deref().unwrap_or_default()
} else {
communication.content.as_str()
},
simclock_time,
},
"agent communication"
);
}
pub(crate) async fn read_simclock_time(
time_provider: Arc<dyn TimeProvider>,
thread_id: ThreadId,
) -> Option<i64> {
time_provider
.current_time(thread_id)
.await
.ok()
.map(|time| time.timestamp())
}
pub(crate) fn emit_agent_communication_receive(communication_id: &str, simclock_time: Option<i64>) {
tracing::trace!(
target: AGENT_COMMUNICATION_TARGET,
{
event.name = "codex.agent_communication",
communication_id,
state = "receive",
simclock_time,
},
"agent communication"
);
}
codex-rs/core/src/lib.rs
+1
View File
@@ -31,6 +31,7 @@ pub use codex_thread::TryStartTurnIfIdleError;
pub use codex_thread::TryStartTurnIfIdleRejectionReason;
pub use session::turn_context::TurnContext;
mod agent;
mod agent_communication;
mod attestation;
mod codex_delegate;
mod command_canonicalization;
codex-rs/core/src/session/handlers.rs
+13
View File
@@ -285,6 +285,19 @@ pub async fn inter_agent_communication(
sess.input_queue
.enqueue_mailbox_communication(communication)
.await;
if crate::agent_communication::logging_enabled() {
let communication_id = sub_id.clone();
let thread_id = sess.thread_id;
let time_provider = Arc::clone(&sess.services.simclock_time_provider);
sess.services.runtime_handle.spawn(async move {
let simclock_time =
crate::agent_communication::read_simclock_time(time_provider, thread_id).await;
crate::agent_communication::emit_agent_communication_receive(
&communication_id,
simclock_time,
);
});
}
if trigger_turn {
sess.maybe_start_turn_for_pending_work_with_sub_id(sub_id)
.await;
codex-rs/core/src/session/mod.rs
+5 -1
View File
@@ -14,6 +14,8 @@ use crate::agent::AgentControl;
use crate::agent::AgentStatus;
use crate::agent::agent_status_from_event;
use crate::agent::status::is_final;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::attestation::AttestationProvider;
use crate::build_available_skills;
use crate::compact;
@@ -1867,10 +1869,12 @@ impl Session {
message,
/*trigger_turn*/ false,
);
let context =
AgentCommunicationContext::new(AgentCommunicationKind::Result, self.thread_id);
if let Err(err) = self
.services
.agent_control
.send_inter_agent_communication(parent_thread_id, communication)
.send_inter_agent_communication(parent_thread_id, communication, context)
.await
{
debug!("failed to notify parent thread {parent_thread_id}: {err}");
codex-rs/core/src/session/session.rs
+4
View File
@@ -557,6 +557,9 @@ impl Session {
.effective_agent_max_threads(MultiAgentVersion::V2)
.unwrap_or(usize::MAX),
);
let simclock_time_provider: Arc<dyn TimeProvider> = external_time_provider
.clone()
.unwrap_or_else(|| Arc::new(crate::current_time::SystemTimeProvider));
let time_provider = crate::current_time::resolve_time_provider(
config.current_time_reminder.as_ref(),
external_time_provider,
@@ -1094,6 +1097,7 @@ impl Session {
thread_store: Arc::clone(&thread_store),
attestation_provider: attestation_provider.clone(),
time_provider,
simclock_time_provider,
model_client: ModelClient::new(
Some(Arc::clone(&auth_manager)),
if config.features.enabled(Feature::UseAgentIdentity) {
codex-rs/core/src/session/tests.rs
+2
View File
@@ -5431,6 +5431,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
)),
attestation_provider: None,
time_provider: Arc::new(crate::current_time::SystemTimeProvider),
simclock_time_provider: Arc::new(crate::current_time::SystemTimeProvider),
model_client: ModelClient::new(
Some(auth_manager.clone()),
AgentIdentityAuthPolicy::JwtOnly,
@@ -7557,6 +7558,7 @@ where
)),
attestation_provider: None,
time_provider: Arc::new(crate::current_time::SystemTimeProvider),
simclock_time_provider: Arc::new(crate::current_time::SystemTimeProvider),
model_client: ModelClient::new(
Some(Arc::clone(&auth_manager)),
AgentIdentityAuthPolicy::JwtOnly,
codex-rs/core/src/state/service.rs
+1
View File
@@ -94,6 +94,7 @@ pub(crate) struct SessionServices {
pub(crate) thread_store: Arc<dyn ThreadStore>,
pub(crate) attestation_provider: Option<Arc<dyn AttestationProvider>>,
pub(crate) time_provider: Arc<dyn TimeProvider>,
pub(crate) simclock_time_provider: Arc<dyn TimeProvider>,
/// Session-scoped model client shared across turns.
pub(crate) model_client: ModelClient,
pub(crate) code_mode_service: CodeModeService,
codex-rs/core/src/tools/handlers/multi_agents_v2.rs
-1
View File
@@ -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;
codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs
+9 -2
View File
@@ -4,6 +4,8 @@
//! resulting `InterAgentCommunication` should wake the target immediately.
use super::*;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::tools::context::FunctionToolOutput;
use crate::turn_timing::now_unix_timestamp_ms;
use codex_protocol::protocol::InterAgentCommunication;
@@ -46,7 +48,7 @@ pub(crate) struct FollowupTaskArgs {
pub(crate) message: String,
}
fn message_content(message: String) -> Result<String, FunctionCallError> {
pub(super) fn message_content(message: String) -> Result<String, FunctionCallError> {
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?;
codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs
+25 -29
View File
@@ -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)?;
codex-rs/core/tests/suite/subagent_notifications.rs
+115 -11
View File
@@ -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<Option<ThreadId>>,
}
impl ThreadSimClock {
fn set_root_thread_id(&self, thread_id: ThreadId) {
*self.root_thread_id.lock().expect("simclock lock") = Some(thread_id);
}
}
impl TimeProvider for ThreadSimClock {
fn current_time(&self, thread_id: ThreadId) -> TimeFuture<'_> {
let root_thread_id = self
.root_thread_id
.lock()
.expect("simclock lock")
.expect("root thread ID should be installed before reading the clock");
let timestamp = if thread_id == root_thread_id {
ROOT_SIMCLOCK_TIME
} else {
CHILD_SIMCLOCK_TIME
};
Box::pin(async move {
Ok(DateTime::<Utc>::from_timestamp(timestamp, 0)
.expect("test simclock timestamp should be valid"))
})
}
fn sleep(&self, _thread_id: ThreadId, _duration: Duration) -> SleepFuture<'_> {
Box::pin(async { Ok(()) })
}
}
fn body_contains(req: &wiremock::Request, text: &str) -> bool {
decoded_body(req)
@@ -96,6 +141,13 @@ fn decoded_body(req: &wiremock::Request) -> Option<Vec<u8>> {
}
}
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<Vec<u8>> = Box::leak(Box::new(Mutex::new(Vec::new())));
let subscriber = tracing_subscriber::fmt()
.with_ansi(false)
.with_max_level(Level::TRACE)
.with_writer(MockWriter::new(output))
.finish();
let _guard = tracing::subscriber::set_default(subscriber);
let server = start_mock_server().await;
let 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(())
}
codex-rs/feedback/src/lib.rs
+28 -1
View File
@@ -18,6 +18,7 @@ use tracing::Event;
use tracing::Level;
use tracing::field::Visit;
use tracing_subscriber::Layer;
use tracing_subscriber::filter::LevelFilter;
use tracing_subscriber::filter::Targets;
use tracing_subscriber::fmt::writer::MakeWriter;
use tracing_subscriber::registry::LookupSpan;
@@ -204,7 +205,11 @@ impl CodexFeedback {
.with_target(false)
// Capture everything, regardless of the caller's `RUST_LOG`, so feedback includes the
// full trace when the user uploads a report.
.with_filter(Targets::new().with_default(Level::TRACE))
.with_filter(
Targets::new()
.with_default(Level::TRACE)
.with_target("codex_core::agent_communication", LevelFilter::OFF),
)
}
/// Returns a [`tracing_subscriber`] layer that collects structured metadata for feedback.
@@ -723,6 +728,28 @@ mod tests {
pretty_assertions::assert_eq!(snap.tags.get("cached").map(String::as_str), Some("true"));
}
#[test]
fn logger_layer_excludes_agent_communication_content() {
let fb = CodexFeedback::new();
let _guard = tracing_subscriber::registry()
.with(fb.logger_layer())
.set_default();
tracing::info!(target: "codex_core", "retained-log");
tracing::trace!(
target: "codex_core::agent_communication",
content = "secret",
"dropped-log"
);
let logs = std::str::from_utf8(fb.snapshot(/*session_id*/ None).as_bytes())
.expect("feedback logs should be UTF-8")
.to_string();
assert!(logs.contains("retained-log"));
assert!(!logs.contains("secret"));
assert!(!logs.contains("dropped-log"));
}
#[test]
fn feedback_attachments_gate_connectivity_diagnostics() {
let extra_filename = format!("codex-feedback-extra-{}.jsonl", ThreadId::new());
codex-rs/state/src/log_db.rs
+1
View File
@@ -54,6 +54,7 @@ pub fn default_filter() -> Targets {
Targets::new()
.with_default(LevelFilter::TRACE)
.with_target("log", LevelFilter::OFF)
.with_target("codex_core::agent_communication", LevelFilter::OFF)
.with_target("codex_otel.log_only", LevelFilter::OFF)
.with_target("codex_otel.trace_safe", LevelFilter::OFF)
}
codex-rs/state/src/log_db_filter_tests.rs
+6 -6
View File
@@ -1,5 +1,4 @@
use pretty_assertions::assert_eq;
use tracing_subscriber::filter::Targets;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;
use uuid::Uuid;
@@ -16,17 +15,18 @@ async fn sqlite_sink_drops_low_level_opentelemetry_sdk_logs() {
let layer = start(runtime.clone());
let guard = tracing_subscriber::registry()
.with(
layer
.clone()
.with_filter(Targets::new().with_default(tracing::Level::TRACE)),
)
.with(layer.clone().with_filter(default_filter()))
.set_default();
tracing::trace!(target: "opentelemetry_sdk", "dropped-trace");
tracing::debug!(target: "opentelemetry_sdk", "dropped-debug");
tracing::info!(target: "opentelemetry_sdk", "retained-info");
tracing::trace!(target: "codex_state", "retained-trace");
tracing::trace!(
target: "codex_core::agent_communication",
content = "secret",
"dropped-agent-communication"
);
layer.flush().await;
drop(guard);