github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-29 08:07:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Log multi-agent communication lifecycle (#30872)

Browse Source 
## Why

[#30867](https://github.com/openai/codex/pull/30867) makes
`submit_inter_agent_communication` the common outbound sink for
multi-agent v2 communications. This follow-up uses that single point to
log every communication lifecycle without requiring new hooks as spawn,
messaging, follow-up, or result paths evolve.

For each communication, the logs need to identify its type, sender and
receiver threads, and content, while correlating the successful send
with receipt by the destination mailbox. The logging path must not query
externally supplied time providers because those calls can be expensive
for app-server clients.

## What changed

- Added structured `INFO` events on the OpenTelemetry-exported
`codex_otel.agent_communication` target for `spawn`, `message`,
`followup`, and `result` communications.
- Logged successful sends from `submit_inter_agent_communication` with
the communication kind, sender and receiver thread IDs, content, and
submission ID.
- Logged receives after the communication has been enqueued in the
receiver mailbox, using the same submission ID.
- Avoided time-provider calls and other asynchronous work in the logging
path.
- Narrowed ordinary spawn and send-input APIs to `Vec<UserInput>` so
`Op::InterAgentCommunication` cannot bypass the context-bearing
centralized path.

The refactor does not change submission IDs, capacity checks, last-task
bookkeeping, mailbox ordering, protocol types, rollout data, or
model-visible context.

## Event shape

Illustrative JSON representation of the two independently emitted
records:

```json
[
  {
    "event.name": "codex.agent_communication",
    "communication_id": "019f20e1-40d1-7890-a123-456789abcdef",
    "kind": "spawn",
    "state": "send",
    "sender_thread_id": "019f20df-fbe1-7890-a123-456789abcdef",
    "receiver_thread_id": "019f20e1-3f79-7890-a123-456789abcdef",
    "content": "inspect the repository"
  },
  {
    "event.name": "codex.agent_communication",
    "communication_id": "019f20e1-40d1-7890-a123-456789abcdef",
    "state": "receive"
  }
]
```

Consumers join the receive record to the send record by
`communication_id` for the immutable communication metadata.

## Testing

- Extended the existing end-to-end multi-agent v2 spawn test to verify
content, both thread IDs, and a correlated send/receive submission ID.
- Re-ran focused control and handler coverage for direct messages,
follow-up tasks, and completion results.
This commit is contained in:
Michael Bolin
2026-07-01 18:11:09 -07:00
committed by GitHub
parent a98a21798c
commit 129ea2aaf5
15 changed files with 332 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control.rs
+62 -35
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;
@@ -139,12 +141,12 @@ impl AgentControl {
pub(crate) async fn send_input(
&self,
agent_id: ThreadId,
initial_operation: Op,
input: Vec<UserInput>,
) -> CodexResult<String> {
let state = self.upgrade()?;
self.ensure_execution_capacity_for_op(agent_id, &initial_operation)
self.ensure_execution_capacity_for_turn_start(agent_id, /*starts_turn*/ true)
.await?;
self.send_input_after_capacity_check(agent_id, &state, initial_operation)
self.send_input_after_capacity_check(agent_id, &state, input)
.await
}
@@ -152,20 +154,14 @@ impl AgentControl {
&self,
agent_id: ThreadId,
state: &Arc<ThreadManagerState>,
initial_operation: Op,
input: Vec<UserInput>,
) -> 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 last_task_message = non_empty_task_message(render_input_preview(&input));
let result = self
.handle_thread_request_result(
agent_id,
state,
state.send_op(agent_id, initial_operation).await,
state.send_op(agent_id, input.into()).await,
)
.await;
if result.is_ok() {
@@ -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,11 @@ 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 communication_for_log =
crate::agent_communication::logging_enabled().then(|| communication.clone());
let result = self
.handle_thread_request_result(
agent_id,
@@ -204,6 +223,16 @@ impl AgentControl {
.await,
)
.await;
if let (Some(communication), Ok(communication_id)) =
(communication_for_log, result.as_ref())
{
crate::agent_communication::emit_agent_communication_send(
communication_id,
&context,
&communication,
agent_id,
);
}
if result.is_ok() {
match last_task_message {
Some(last_task_message) => self
@@ -494,8 +523,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;
}
@@ -720,27 +751,23 @@ fn agent_matches_prefix(agent_path: Option<&AgentPath>, prefix: &AgentPath) -> b
})
}
pub(crate) fn render_input_preview(initial_operation: &Op) -> String {
match initial_operation {
Op::UserInput { items, .. } => items
.iter()
.map(|item| match item {
UserInput::Text { text, .. } => text.clone(),
UserInput::Image { .. } => "[image]".to_string(),
UserInput::LocalImage { path, .. } => {
format!("[local_image:{}]", path.display())
}
UserInput::Skill { name, path, .. } => {
format!("[skill:${name}]({})", path.display())
}
UserInput::Mention { name, path, .. } => format!("[mention:${name}]({path})"),
_ => "[input]".to_string(),
})
.collect::<Vec<_>>()
.join("\n"),
Op::InterAgentCommunication { communication } => communication.content.clone(),
_ => String::new(),
}
pub(crate) fn render_input_preview(input: &[UserInput]) -> String {
input
.iter()
.map(|item| match item {
UserInput::Text { text, .. } => text.clone(),
UserInput::Image { .. } => "[image]".to_string(),
UserInput::LocalImage { path, .. } => {
format!("[local_image:{}]", path.display())
}
UserInput::Skill { name, path, .. } => {
format!("[skill:${name}]({})", path.display())
}
UserInput::Mention { name, path, .. } => format!("[mention:${name}]({path})"),
_ => "[input]".to_string(),
})
.collect::<Vec<_>>()
.join("\n")
}
fn last_task_message_from_communication(communication: &InterAgentCommunication) -> Option<String> {
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
+54 -8
View File
@@ -9,6 +9,17 @@ struct SpawnAgentThreadInheritance {
exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
}
/// Initial input delivered after a spawned agent acquires execution capacity.
///
/// V2 communication spawns keep the communication and its context paired so centralized
/// submission and lifecycle logging cannot receive one without the other. Other spawn sources
/// provide user input directly, making an uncontextualized inter-agent communication
/// unrepresentable.
enum SpawnInitialInput {
UserInput(Vec<UserInput>),
InterAgentCommunication(InterAgentCommunication, AgentCommunicationContext),
}
fn default_agent_nickname_list() -> Vec<&'static str> {
AGENT_NAMES
.lines()
@@ -89,12 +100,12 @@ impl AgentControl {
pub(crate) async fn spawn_agent(
&self,
config: Config,
initial_operation: Op,
initial_input: Vec<UserInput>,
session_source: Option<SessionSource>,
) -> CodexResult<ThreadId> {
let spawned_agent = Box::pin(self.spawn_agent_internal(
config,
initial_operation,
SpawnInitialInput::UserInput(initial_input),
session_source,
SpawnAgentOptions::default(),
))
@@ -106,12 +117,34 @@ impl AgentControl {
pub(crate) async fn spawn_agent_with_metadata(
&self,
config: Config,
initial_operation: Op,
initial_input: Vec<UserInput>,
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::UserInput(initial_input),
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,21 @@ impl AgentControl {
)
.await;
self.send_input_after_capacity_check(new_thread.thread_id, &state, initial_operation)
.await?;
match initial_input {
SpawnInitialInput::UserInput(input) => {
self.send_input_after_capacity_check(new_thread.thread_id, &state, input)
.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
+21 -11
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;
@@ -65,12 +67,11 @@ async fn test_config() -> (TempDir, Config) {
test_config_with_cli_overrides(Vec::new()).await
}
fn text_input(text: &str) -> Op {
fn text_input(text: &str) -> Vec<UserInput> {
vec![UserInput::Text {
text: text.to_string(),
text_elements: Vec::new(),
}]
.into()
}
fn assistant_message(text: &str, phase: Option<MessagePhase>) -> ResponseItem {
@@ -311,8 +312,7 @@ async fn send_input_errors_when_manager_dropped() {
vec![UserInput::Text {
text: "hello".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
)
.await
.expect_err("send_input should fail without a manager");
@@ -423,8 +423,7 @@ async fn send_input_errors_when_thread_missing() {
vec![UserInput::Text {
text: "hello".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
)
.await
.expect_err("send_input should fail for missing thread");
@@ -490,8 +489,7 @@ async fn send_input_submits_user_message() {
vec![UserInput::Text {
text: "hello from tests".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
)
.await
.expect("send_input should succeed");
@@ -531,7 +529,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 +658,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 +809,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");