From 8f2d6416ce41be54551185c640f83e22f061eccd Mon Sep 17 00:00:00 2001 From: jif Date: Fri, 12 Jun 2026 21:50:04 +0100 Subject: [PATCH] Support plaintext agent messages (#27830) ## Why Multi-agent v2 `send_message` deliveries already reach the receiving model as typed `agent_message` items with encrypted content. Child-completion notifications are generated by Codex itself, so their content is plaintext and previously fell back to a serialized JSON envelope inside an assistant message. With plaintext `input_text` supported for `agent_message`, both delivery paths can use the same model-visible type while preserving explicit author and recipient metadata. ## What changed - add plaintext `input_text` support to `AgentMessageInputContent` and regenerate the affected app-server schemas - preserve `InterAgentCommunication` as structured mailbox input instead of converting it to assistant text - record delivered communications as typed `agent_message` history items - persist a dedicated rollout item so local delivery metadata such as `trigger_turn` remains available without leaking into the Responses request - reconstruct typed agent messages on resume and preserve fork-turn truncation behavior - remove request-time assistant-content parsing - preserve plaintext and encrypted inter-agent deliveries in stage-one memory inputs - normalize and link plaintext and encrypted agent messages in rollout traces without treating inbound messages as child results - cover the real MultiAgent V2 child-completion path end to end with deterministic mailbox synchronization ## Verification - `just test -p codex-core plaintext_multi_agent_v2_completion_sends_agent_message` - `just test -p codex-core input_queue_drains_mailbox_in_delivery_order record_initial_history_reconstructs_typed_inter_agent_message fork_turn_positions_use_inter_agent_delivery_metadata` - `just test -p codex-memories-write serializes_inter_agent_communications_for_memory` - `just test -p codex-rollout-trace agent_messages_preserve_routing_and_content sub_agent_started_activity_creates_spawn_edge` - `just test -p codex-rollout-trace agent_result_edge_falls_back_to_child_thread_without_result_message` - `just test -p codex-protocol -p codex-rollout -p codex-app-server-protocol` --- .../schema/json/ClientRequest.json | 20 ++++ .../codex_app_server_protocol.schemas.json | 20 ++++ .../codex_app_server_protocol.v2.schemas.json | 20 ++++ .../RawResponseItemCompletedNotification.json | 20 ++++ .../schema/json/v2/ThreadResumeParams.json | 20 ++++ .../typescript/AgentMessageInputContent.ts | 2 +- .../src/protocol/thread_history.rs | 4 +- codex-rs/core/src/agent/control/spawn.rs | 1 + codex-rs/core/src/client_common.rs | 22 +--- codex-rs/core/src/guardian/prompt.rs | 17 +++ codex-rs/core/src/hook_runtime.rs | 8 ++ codex-rs/core/src/prompt_debug.rs | 2 +- codex-rs/core/src/realtime_context.rs | 18 +++ codex-rs/core/src/session/handlers.rs | 2 +- codex-rs/core/src/session/input_queue.rs | 17 ++- codex-rs/core/src/session/mod.rs | 20 ++++ .../src/session/rollout_reconstruction.rs | 12 ++ .../session/rollout_reconstruction_tests.rs | 25 ++++ codex-rs/core/src/session/tests.rs | 12 +- codex-rs/core/src/session/turn.rs | 4 +- codex-rs/core/src/tasks/review.rs | 2 +- .../core/src/thread_rollout_truncation.rs | 10 +- .../src/thread_rollout_truncation_tests.rs | 24 ++++ ...ng_input_queued_mail_after_commentary.snap | 2 +- ...ing_input_queued_mail_after_reasoning.snap | 2 +- .../tests/suite/subagent_notifications.rs | 108 ++++++++++++++++++ codex-rs/ext/web-search/src/history.rs | 23 ++++ codex-rs/memories/write/src/phase1.rs | 46 +++++++- codex-rs/protocol/src/models.rs | 1 + codex-rs/protocol/src/protocol.rs | 24 ++-- .../rollout-trace/src/model/conversation.rs | 13 +++ .../rollout-trace/src/reducer/conversation.rs | 3 + .../src/reducer/conversation/normalize.rs | 58 ++++++++++ .../src/reducer/conversation_tests.rs | 87 ++++++++++++++ .../rollout-trace/src/reducer/tool/agents.rs | 70 ++++++++++-- .../src/reducer/tool/agents_tests.rs | 32 ++++-- codex-rs/rollout/src/list.rs | 12 +- codex-rs/rollout/src/metadata.rs | 2 + codex-rs/rollout/src/policy.rs | 1 + codex-rs/rollout/src/recorder.rs | 32 ++---- codex-rs/rollout/src/search.rs | 1 + codex-rs/state/src/extract.rs | 8 +- codex-rs/state/src/runtime/threads.rs | 1 + .../thread-store/src/thread_metadata_sync.rs | 1 + 44 files changed, 716 insertions(+), 113 deletions(-) diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 1c7d3d0df..0829bcb1a 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -36,6 +36,26 @@ }, "AgentMessageInputContent": { "oneOf": [ + { + "properties": { + "text": { + "type": "string" + }, + "type": { + "enum": [ + "input_text" + ], + "title": "InputTextAgentMessageInputContentType", + "type": "string" + } + }, + "required": [ + "text", + "type" + ], + "title": "InputTextAgentMessageInputContent", + "type": "object" + }, { "properties": { "encrypted_content": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index f8ddf9329..41040e9fd 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -6054,6 +6054,26 @@ }, "AgentMessageInputContent": { "oneOf": [ + { + "properties": { + "text": { + "type": "string" + }, + "type": { + "enum": [ + "input_text" + ], + "title": "InputTextAgentMessageInputContentType", + "type": "string" + } + }, + "required": [ + "text", + "type" + ], + "title": "InputTextAgentMessageInputContent", + "type": "object" + }, { "properties": { "encrypted_content": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index e04aca08f..da1370be3 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -323,6 +323,26 @@ }, "AgentMessageInputContent": { "oneOf": [ + { + "properties": { + "text": { + "type": "string" + }, + "type": { + "enum": [ + "input_text" + ], + "title": "InputTextAgentMessageInputContentType", + "type": "string" + } + }, + "required": [ + "text", + "type" + ], + "title": "InputTextAgentMessageInputContent", + "type": "object" + }, { "properties": { "encrypted_content": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/RawResponseItemCompletedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/RawResponseItemCompletedNotification.json index c18f9aeb1..2d954061f 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/RawResponseItemCompletedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/RawResponseItemCompletedNotification.json @@ -3,6 +3,26 @@ "definitions": { "AgentMessageInputContent": { "oneOf": [ + { + "properties": { + "text": { + "type": "string" + }, + "type": { + "enum": [ + "input_text" + ], + "title": "InputTextAgentMessageInputContentType", + "type": "string" + } + }, + "required": [ + "text", + "type" + ], + "title": "InputTextAgentMessageInputContent", + "type": "object" + }, { "properties": { "encrypted_content": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeParams.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeParams.json index dc5ef5cda..42505ac24 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeParams.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeParams.json @@ -7,6 +7,26 @@ }, "AgentMessageInputContent": { "oneOf": [ + { + "properties": { + "text": { + "type": "string" + }, + "type": { + "enum": [ + "input_text" + ], + "title": "InputTextAgentMessageInputContentType", + "type": "string" + } + }, + "required": [ + "text", + "type" + ], + "title": "InputTextAgentMessageInputContent", + "type": "object" + }, { "properties": { "encrypted_content": { diff --git a/codex-rs/app-server-protocol/schema/typescript/AgentMessageInputContent.ts b/codex-rs/app-server-protocol/schema/typescript/AgentMessageInputContent.ts index a3bb64559..3fd526d83 100644 --- a/codex-rs/app-server-protocol/schema/typescript/AgentMessageInputContent.ts +++ b/codex-rs/app-server-protocol/schema/typescript/AgentMessageInputContent.ts @@ -2,4 +2,4 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. -export type AgentMessageInputContent = { "type": "encrypted_content", encrypted_content: string, }; +export type AgentMessageInputContent = { "type": "input_text", text: string, } | { "type": "encrypted_content", encrypted_content: string, }; diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index f0b900f08..f499e568f 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -241,7 +241,9 @@ impl ThreadHistoryBuilder { RolloutItem::EventMsg(event) => self.handle_event(event), RolloutItem::Compacted(payload) => self.handle_compacted(payload), RolloutItem::ResponseItem(item) => self.handle_response_item(item), - RolloutItem::TurnContext(_) | RolloutItem::SessionMeta(_) => {} + RolloutItem::InterAgentCommunication(_) + | RolloutItem::TurnContext(_) + | RolloutItem::SessionMeta(_) => {} } } diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index a0d52ea4d..dcfa1b83a 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -55,6 +55,7 @@ fn keep_forked_rollout_item(item: &RolloutItem, preserve_reference_context_item: | ResponseItem::ContextCompaction { .. } | ResponseItem::Other, ) => false, + RolloutItem::InterAgentCommunication(_) => false, // Full-history forks preserve the cached prompt prefix and can keep diffing // from the parent's durable baseline. Truncated forks drop part of that prompt, // so they must rebuild context on their first child turn. diff --git a/codex-rs/core/src/client_common.rs b/codex-rs/core/src/client_common.rs index 4d7729d11..71d102598 100644 --- a/codex-rs/core/src/client_common.rs +++ b/codex-rs/core/src/client_common.rs @@ -5,7 +5,6 @@ use codex_protocol::models::BaseInstructions; use codex_protocol::models::ContentItem; use codex_protocol::models::FunctionCallOutputContentItem; use codex_protocol::models::ResponseItem; -use codex_protocol::protocol::InterAgentCommunication; use codex_tools::ToolSpec; use futures::Stream; use serde_json::Value; @@ -55,30 +54,11 @@ impl Default for Prompt { } impl Prompt { - pub(crate) fn get_formatted_input(&self) -> Vec { - self.input - .iter() - .cloned() - .map(|item| { - let ResponseItem::Message { role, content, .. } = &item else { - return item; - }; - if role != "assistant" { - return item; - } - InterAgentCommunication::from_message_content(content) - .filter(|communication| communication.encrypted_content.is_some()) - .map(|communication| communication.to_model_input_item()) - .unwrap_or(item) - }) - .collect() - } - pub(crate) fn get_formatted_input_for_request( &self, use_responses_lite: bool, ) -> Vec { - let mut input = self.get_formatted_input(); + let mut input = self.input.clone(); if use_responses_lite { strip_image_details(&mut input); } diff --git a/codex-rs/core/src/guardian/prompt.rs b/codex-rs/core/src/guardian/prompt.rs index 008e97957..c40a03bf5 100644 --- a/codex-rs/core/src/guardian/prompt.rs +++ b/codex-rs/core/src/guardian/prompt.rs @@ -1,5 +1,6 @@ use std::collections::HashMap; +use codex_protocol::models::AgentMessageInputContent; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::GuardianRiskLevel; use codex_protocol::protocol::GuardianUserAuthorization; @@ -452,6 +453,22 @@ pub(crate) fn collect_guardian_transcript_entries( ResponseItem::Message { role, content, .. } if role == "assistant" => { content_entry(GuardianTranscriptEntryKind::Assistant, content) } + ResponseItem::AgentMessage { + author, content, .. + } => { + let text = content + .iter() + .filter_map(|content| match content { + AgentMessageInputContent::InputText { text } => Some(text.as_str()), + AgentMessageInputContent::EncryptedContent { .. } => None, + }) + .collect::>() + .join("\n"); + (!text.trim().is_empty()).then(|| GuardianTranscriptEntry { + kind: GuardianTranscriptEntryKind::Assistant, + text: format!("Agent message from {author}:\n{text}"), + }) + } ResponseItem::LocalShellCall { action, .. } => serialized_entry( GuardianTranscriptEntryKind::Tool("tool shell call".to_string()), serde_json::to_string(action).ok(), diff --git a/codex-rs/core/src/hook_runtime.rs b/codex-rs/core/src/hook_runtime.rs index 1097f7bc7..325fd17c9 100644 --- a/codex-rs/core/src/hook_runtime.rs +++ b/codex-rs/core/src/hook_runtime.rs @@ -528,6 +528,10 @@ pub(crate) async fn inspect_pending_input( should_stop: false, additional_contexts: Vec::new(), }, + TurnInput::InterAgentCommunication(_) => HookRuntimeOutcome { + should_stop: false, + additional_contexts: Vec::new(), + }, } } @@ -550,6 +554,10 @@ pub(crate) async fn record_pending_input( sess.record_conversation_items(turn_context, std::slice::from_ref(&item)) .await; } + TurnInput::InterAgentCommunication(communication) => { + sess.record_inter_agent_communication(turn_context, communication) + .await; + } } record_additional_contexts(sess, turn_context, additional_contexts).await; } diff --git a/codex-rs/core/src/prompt_debug.rs b/codex-rs/core/src/prompt_debug.rs index 74fe19f38..0da42d8a4 100644 --- a/codex-rs/core/src/prompt_debug.rs +++ b/codex-rs/core/src/prompt_debug.rs @@ -98,5 +98,5 @@ pub(crate) async fn build_prompt_input_from_session( base_instructions, ); - Ok(prompt.get_formatted_input()) + Ok(prompt.input) } diff --git a/codex-rs/core/src/realtime_context.rs b/codex-rs/core/src/realtime_context.rs index f9bb49a42..6aaf90149 100644 --- a/codex-rs/core/src/realtime_context.rs +++ b/codex-rs/core/src/realtime_context.rs @@ -4,6 +4,7 @@ use crate::session::session::Session; use chrono::Utc; use codex_exec_server::LOCAL_FS; use codex_git_utils::resolve_root_git_project_for_trust; +use codex_protocol::models::AgentMessageInputContent; use codex_protocol::models::ResponseItem; use codex_thread_store::ListThreadsParams; use codex_thread_store::SortDirection; @@ -238,6 +239,23 @@ fn build_current_thread_section(items: &[ResponseItem]) -> Option { } current_assistant.push(text); } + ResponseItem::AgentMessage { + author, content, .. + } => { + let text = content + .iter() + .filter_map(|content| match content { + AgentMessageInputContent::InputText { text } => Some(text.as_str()), + AgentMessageInputContent::EncryptedContent { .. } => None, + }) + .collect::>() + .join("\n"); + if text.trim().is_empty() || current_user.is_empty() && current_assistant.is_empty() + { + continue; + } + current_assistant.push(format!("Agent message from {author}:\n{text}")); + } _ => {} } } diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index bfb87a0ff..d4525cf45 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -273,7 +273,7 @@ pub(super) async fn user_input_or_turn_inner( } } -/// Records an inter-agent assistant envelope, then lets the shared pending-work scheduler +/// Queues an inter-agent message, then lets the shared pending-work scheduler /// decide whether an idle session should start a regular turn. pub async fn inter_agent_communication( sess: &Arc, diff --git a/codex-rs/core/src/session/input_queue.rs b/codex-rs/core/src/session/input_queue.rs index b62383e09..199ffd004 100644 --- a/codex-rs/core/src/session/input_queue.rs +++ b/codex-rs/core/src/session/input_queue.rs @@ -16,6 +16,7 @@ pub(crate) enum TurnInput { client_id: Option, }, ResponseItem(ResponseItem), + InterAgentCommunication(InterAgentCommunication), } /// Turn-local pending input storage owned by the input queue flow. @@ -70,12 +71,12 @@ impl InputQueue { .any(|mail| mail.trigger_turn) } - pub(crate) async fn drain_mailbox_input_items(&self) -> Vec { + pub(crate) async fn drain_mailbox_input_items(&self) -> Vec { self.mailbox_pending_mails .lock() .await .drain(..) - .map(|mail| ResponseItem::from(mail.to_response_input_item())) + .map(TurnInput::InterAgentCommunication) .collect() } @@ -189,11 +190,7 @@ impl InputQueue { if !accepts_mailbox_delivery { return pending_input; } - let mailbox_items = self - .drain_mailbox_input_items() - .await - .into_iter() - .map(TurnInput::ResponseItem); + let mailbox_items = self.drain_mailbox_input_items().await.into_iter(); if pending_input.is_empty() { mailbox_items.collect() } else { @@ -290,7 +287,7 @@ mod tests { AgentPath::try_from("/root/worker").expect("agent path"), AgentPath::root(), "two", - /*trigger_turn*/ false, + /*trigger_turn*/ true, ); input_queue @@ -303,8 +300,8 @@ mod tests { assert_eq!( input_queue.drain_mailbox_input_items().await, vec![ - ResponseItem::from(mail_one.to_response_input_item()), - ResponseItem::from(mail_two.to_response_input_item()) + TurnInput::InterAgentCommunication(mail_one), + TurnInput::InterAgentCommunication(mail_two) ] ); assert!(!input_queue.has_pending_mailbox_items().await); diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index b6241408b..b1013965d 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -2659,6 +2659,26 @@ impl Session { self.send_raw_response_items(turn_context, items).await; } + pub(crate) async fn record_inter_agent_communication( + &self, + turn_context: &TurnContext, + communication: InterAgentCommunication, + ) { + let response_item = communication.to_model_input_item(); + let items = self.prepare_conversation_items_for_history( + turn_context, + std::slice::from_ref(&response_item), + ); + let items = items.as_ref(); + { + let mut state = self.state.lock().await; + state.record_items(items.iter(), turn_context.truncation_policy); + } + self.persist_rollout_items(&[RolloutItem::InterAgentCommunication(communication)]) + .await; + self.send_raw_response_items(turn_context, items).await; + } + async fn maybe_warn_on_server_model_mismatch( self: &Arc, turn_context: &Arc, diff --git a/codex-rs/core/src/session/rollout_reconstruction.rs b/codex-rs/core/src/session/rollout_reconstruction.rs index e88664851..c844a630f 100644 --- a/codex-rs/core/src/session/rollout_reconstruction.rs +++ b/codex-rs/core/src/session/rollout_reconstruction.rs @@ -220,6 +220,11 @@ impl Session { active_segment.get_or_insert_with(ActiveReplaySegment::default); active_segment.counts_as_user_turn |= is_user_turn_boundary(response_item); } + RolloutItem::InterAgentCommunication(_) => { + let active_segment = + active_segment.get_or_insert_with(ActiveReplaySegment::default); + active_segment.counts_as_user_turn = true; + } RolloutItem::EventMsg(_) | RolloutItem::SessionMeta(_) => {} } @@ -269,6 +274,13 @@ impl Session { turn_context.truncation_policy, ); } + RolloutItem::InterAgentCommunication(communication) => { + let response_item = communication.to_model_input_item(); + history.record_items( + std::iter::once(&response_item), + turn_context.truncation_policy, + ); + } RolloutItem::Compacted(compacted) => { if let Some(replacement_history) = &compacted.replacement_history { // This should actually never happen, because the reverse loop above (to build rollout_suffix) diff --git a/codex-rs/core/src/session/rollout_reconstruction_tests.rs b/codex-rs/core/src/session/rollout_reconstruction_tests.rs index ca8494168..d24394676 100644 --- a/codex-rs/core/src/session/rollout_reconstruction_tests.rs +++ b/codex-rs/core/src/session/rollout_reconstruction_tests.rs @@ -52,6 +52,31 @@ fn inter_agent_assistant_message(text: &str) -> ResponseItem { } } +#[tokio::test] +async fn record_initial_history_reconstructs_typed_inter_agent_message() { + let (session, _turn_context) = make_session_and_context().await; + let communication = InterAgentCommunication::new( + AgentPath::root().join("worker").expect("worker path"), + AgentPath::root(), + Vec::new(), + "child done".to_string(), + /*trigger_turn*/ false, + ); + + session + .record_initial_history(InitialHistory::Resumed(ResumedHistory { + conversation_id: ThreadId::default(), + history: vec![RolloutItem::InterAgentCommunication(communication.clone())], + rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")), + })) + .await; + + assert_eq!( + session.state.lock().await.clone_history().raw_items(), + &[communication.to_model_input_item()] + ); +} + #[tokio::test] async fn record_initial_history_resumed_bare_turn_context_does_not_hydrate_previous_turn_settings() { diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 8068524d9..a04290c5d 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -9222,9 +9222,7 @@ async fn queue_only_mailbox_mail_waits_for_next_turn_after_answer_boundary() { assert_eq!( sess.input_queue.get_pending_input(&sess.active_turn).await, - vec![TurnInput::ResponseItem(ResponseItem::from( - communication.to_response_input_item() - ))], + vec![TurnInput::InterAgentCommunication(communication)], ); } @@ -9313,7 +9311,7 @@ async fn steered_input_reopens_mailbox_delivery_for_current_turn() { }], client_id: None }, - TurnInput::ResponseItem(ResponseItem::from(communication.to_response_input_item())), + TurnInput::InterAgentCommunication(communication), ], ); } @@ -9371,7 +9369,7 @@ async fn stale_defer_mailbox_delivery_does_not_override_steered_input() { }], client_id: None }, - TurnInput::ResponseItem(ResponseItem::from(communication.to_response_input_item())), + TurnInput::InterAgentCommunication(communication), ], ); } @@ -9426,9 +9424,7 @@ async fn tool_calls_reopen_mailbox_delivery_for_current_turn() { assert!(output.tool_future.is_some()); assert_eq!( sess.input_queue.get_pending_input(&sess.active_turn).await, - vec![TurnInput::ResponseItem(ResponseItem::from( - communication.to_response_input_item() - ))], + vec![TurnInput::InterAgentCommunication(communication)], ); } diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index e30e93064..6244274a4 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -466,7 +466,7 @@ async fn build_skills_and_plugins( .iter() .filter_map(|item| match item { TurnInput::UserInput { content, .. } => Some(content.as_slice()), - TurnInput::ResponseItem(_) => None, + TurnInput::ResponseItem(_) | TurnInput::InterAgentCommunication(_) => None, }) .flatten() .cloned() @@ -685,7 +685,7 @@ async fn track_turn_resolved_config_analytics( .iter() .filter_map(|item| match item { TurnInput::UserInput { content, .. } => Some(content.as_slice()), - TurnInput::ResponseItem(_) => None, + TurnInput::ResponseItem(_) | TurnInput::InterAgentCommunication(_) => None, }) .flatten() .filter(|item| { diff --git a/codex-rs/core/src/tasks/review.rs b/codex-rs/core/src/tasks/review.rs index 9f7eba957..0d9e3521d 100644 --- a/codex-rs/core/src/tasks/review.rs +++ b/codex-rs/core/src/tasks/review.rs @@ -65,7 +65,7 @@ impl SessionTask for ReviewTask { for item in input { match item { TurnInput::UserInput { mut content, .. } => user_input.append(&mut content), - TurnInput::ResponseItem(_) => {} + TurnInput::ResponseItem(_) | TurnInput::InterAgentCommunication(_) => {} } } diff --git a/codex-rs/core/src/thread_rollout_truncation.rs b/codex-rs/core/src/thread_rollout_truncation.rs index 69e615c4f..407586715 100644 --- a/codex-rs/core/src/thread_rollout_truncation.rs +++ b/codex-rs/core/src/thread_rollout_truncation.rs @@ -19,6 +19,7 @@ pub(crate) fn initial_history_has_prior_user_turns(conversation_history: &Initia fn rollout_item_is_user_turn_boundary(item: &RolloutItem) -> bool { match item { RolloutItem::ResponseItem(item) => is_user_turn_boundary(item), + RolloutItem::InterAgentCommunication(_) => true, _ => false, } } @@ -58,7 +59,8 @@ pub(crate) fn user_message_positions_in_rollout(items: &[RolloutItem]) -> Vec Vec { + rollback_turn_positions.push(idx); + if communication.trigger_turn { + fork_turn_positions.push(idx); + } + } RolloutItem::EventMsg(EventMsg::ThreadRolledBack(rollback)) => { let num_turns = usize::try_from(rollback.num_turns).unwrap_or(usize::MAX); if num_turns == 0 { diff --git a/codex-rs/core/src/thread_rollout_truncation_tests.rs b/codex-rs/core/src/thread_rollout_truncation_tests.rs index e0a883716..652682776 100644 --- a/codex-rs/core/src/thread_rollout_truncation_tests.rs +++ b/codex-rs/core/src/thread_rollout_truncation_tests.rs @@ -51,6 +51,16 @@ fn inter_agent_msg(text: &str, trigger_turn: bool) -> ResponseItem { communication.to_response_input_item().into() } +fn inter_agent_communication(text: &str, trigger_turn: bool) -> RolloutItem { + RolloutItem::InterAgentCommunication(InterAgentCommunication::new( + AgentPath::root(), + AgentPath::try_from("/root/worker").expect("agent path"), + Vec::new(), + text.to_string(), + trigger_turn, + )) +} + #[test] fn truncates_rollout_from_start_before_nth_user_only() { let items = [ @@ -208,6 +218,20 @@ fn truncates_rollout_to_last_n_fork_turns_counts_trigger_turn_messages() { ); } +#[test] +fn fork_turn_positions_use_inter_agent_delivery_metadata() { + let rollout = vec![ + RolloutItem::ResponseItem(user_msg("user task")), + inter_agent_communication("queued during user turn", /*trigger_turn*/ false), + RolloutItem::ResponseItem(assistant_msg("first answer")), + inter_agent_communication("follow-up task", /*trigger_turn*/ true), + RolloutItem::ResponseItem(assistant_msg("second answer")), + RolloutItem::ResponseItem(user_msg("next user task")), + ]; + + assert_eq!(fork_turn_positions_in_rollout(&rollout), vec![0, 3, 5]); +} + #[test] fn truncates_rollout_to_last_n_fork_turns_drops_startup_prefix_even_when_under_limit() { let rollout = vec![ diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_commentary.snap b/codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_commentary.snap index e65a5f34f..d05b893ed 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_commentary.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_commentary.snap @@ -14,4 +14,4 @@ Scenario: /responses POST bodies (input only, redacted like other suite snapshot 01:message/user:> 02:message/user:first prompt 03:message/assistant:first answer -04:message/assistant:{"author":"/root/worker","recipient":"/root","other_recipients":[],"content":"queued child update","trigger_turn":false} +04:agent_message diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_reasoning.snap b/codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_reasoning.snap index 004196f97..9d220817b 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_reasoning.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_reasoning.snap @@ -14,4 +14,4 @@ Scenario: /responses POST bodies (input only, redacted like other suite snapshot 01:message/user:> 02:message/user:first prompt 03:reasoning:summary=thinking:encrypted=true -04:message/assistant:{"author":"/root/worker","recipient":"/root","other_recipients":[],"content":"queued child update","trigger_turn":false} +04:agent_message diff --git a/codex-rs/core/tests/suite/subagent_notifications.rs b/codex-rs/core/tests/suite/subagent_notifications.rs index c3468e465..40ac2c797 100644 --- a/codex-rs/core/tests/suite/subagent_notifications.rs +++ b/codex-rs/core/tests/suite/subagent_notifications.rs @@ -1102,6 +1102,114 @@ async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn plaintext_multi_agent_v2_completion_sends_agent_message() -> Result<()> { + let server = start_mock_server().await; + let spawn_args = serde_json::to_string(&json!({ + "message": "opaque-encrypted-message", + "task_name": "worker", + }))?; + mount_sse_once_match( + &server, + |req: &wiremock::Request| body_contains(req, TURN_1_PROMPT), + sse(vec![ + ev_response_created("resp-parent-1"), + ev_function_call(SPAWN_CALL_ID, "spawn_agent", &spawn_args), + ev_completed("resp-parent-1"), + ]), + ) + .await; + let child_request = mount_response_once_match( + &server, + |req: &wiremock::Request| body_contains(req, "\"type\":\"agent_message\""), + sse_response(sse(vec![ + ev_response_created("resp-child-1"), + ev_assistant_message("msg-child-1", "child done"), + ev_completed("resp-child-1"), + ])) + .set_delay(Duration::from_secs(1)), + ) + .await; + mount_sse_once_match( + &server, + |req: &wiremock::Request| { + body_contains(req, SPAWN_CALL_ID) && !body_contains(req, "") + }, + sse(vec![ + ev_response_created("resp-parent-2"), + ev_assistant_message("msg-parent-2", "parent done"), + ev_completed("resp-parent-2"), + ]), + ) + .await; + let notification = "\n{\"agent_path\":\"/root/worker\",\"status\":{\"completed\":\"child done\"}}\n"; + // If the child is still running when the parent turn starts, wait_agent blocks + // until mailbox delivery. The follow-up request must then contain that delivery. + mount_sse_once_match( + &server, + |req: &wiremock::Request| { + body_contains(req, TURN_2_NO_WAIT_PROMPT) + && !body_contains(req, "") + }, + sse(vec![ + ev_response_created("resp-parent-3"), + ev_function_call("wait-agent-call", "wait_agent", "{}"), + ev_completed("resp-parent-3"), + ]), + ) + .await; + let agent_request = mount_sse_once_match( + &server, + |req: &wiremock::Request| { + body_contains(req, TURN_2_NO_WAIT_PROMPT) + && body_contains(req, "") + }, + sse(vec![ + ev_response_created("resp-parent-4"), + ev_assistant_message("msg-parent-4", "done"), + ev_completed("resp-parent-4"), + ]), + ) + .await; + let test = 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"); + }) + .build(&server) + .await?; + + test.submit_turn(TURN_1_PROMPT).await?; + let _ = wait_for_requests(&child_request).await?; + test.submit_turn(TURN_2_NO_WAIT_PROMPT).await?; + + let request = wait_for_requests(&agent_request) + .await? + .pop() + .expect("agent message request"); + assert_eq!( + request.inputs_of_type("agent_message"), + vec![json!({ + "type": "agent_message", + "author": "/root/worker", + "recipient": "/root", + "content": [{ + "type": "input_text", + "text": notification, + }], + })] + ); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn skills_toggle_skips_instructions_for_parent_and_spawned_child() -> Result<()> { skip_if_no_network!(Ok(())); diff --git a/codex-rs/ext/web-search/src/history.rs b/codex-rs/ext/web-search/src/history.rs index b7a26d32c..94f9f3977 100644 --- a/codex-rs/ext/web-search/src/history.rs +++ b/codex-rs/ext/web-search/src/history.rs @@ -1,6 +1,7 @@ use codex_api::SearchInput; use codex_core::parse_turn_item; use codex_protocol::items::TurnItem; +use codex_protocol::models::AgentMessageInputContent; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; use codex_tools::retain_tail_from_last_n_user_messages; @@ -30,6 +31,28 @@ fn push_visible_message(messages: &mut Vec, item: &ResponseItem) { ResponseItem::Message { role, .. } if role == ASSISTANT_ROLE => { messages.push(item.clone()); } + ResponseItem::AgentMessage { + author, content, .. + } => { + let text = content + .iter() + .filter_map(|content| match content { + AgentMessageInputContent::InputText { text } => Some(text.as_str()), + AgentMessageInputContent::EncryptedContent { .. } => None, + }) + .collect::>() + .join("\n"); + if !text.trim().is_empty() { + messages.push(ResponseItem::Message { + id: None, + role: ASSISTANT_ROLE.to_string(), + content: vec![ContentItem::OutputText { + text: format!("Agent message from {author}:\n{text}"), + }], + phase: None, + }); + } + } ResponseItem::Message { id, role, diff --git a/codex-rs/memories/write/src/phase1.rs b/codex-rs/memories/write/src/phase1.rs index 326490619..3edad7ff7 100644 --- a/codex-rs/memories/write/src/phase1.rs +++ b/codex-rs/memories/write/src/phase1.rs @@ -405,12 +405,15 @@ mod job { ) -> codex_protocol::error::Result { let filtered = items .iter() - .filter_map(|item| { - if let RolloutItem::ResponseItem(item) = item { - sanitize_response_item_for_memories(item) - } else { - None + .filter_map(|item| match item { + RolloutItem::ResponseItem(item) => sanitize_response_item_for_memories(item), + RolloutItem::InterAgentCommunication(communication) => { + Some(communication.to_model_input_item()) } + RolloutItem::SessionMeta(_) + | RolloutItem::Compacted(_) + | RolloutItem::TurnContext(_) + | RolloutItem::EventMsg(_) => None, }) .collect::>(); let serialized = serde_json::to_string(&filtered).map_err(|err| { @@ -656,6 +659,8 @@ fn emit_metrics(context: &StageOneRequestContext, counts: &Stats) { #[cfg(test)] mod tests { use super::*; + use codex_protocol::AgentPath; + use codex_protocol::protocol::InterAgentCommunication; use pretty_assertions::assert_eq; #[test] @@ -745,6 +750,37 @@ mod tests { assert!(serialized.contains("[REDACTED_SECRET]")); } + #[test] + fn serializes_inter_agent_communications_for_memory() { + let plaintext = InterAgentCommunication::new( + AgentPath::root().join("worker").expect("worker path"), + AgentPath::root(), + Vec::new(), + "child done".to_string(), + /*trigger_turn*/ false, + ); + let encrypted = InterAgentCommunication::new_encrypted( + AgentPath::root(), + AgentPath::root().join("worker").expect("worker path"), + Vec::new(), + "encrypted payload".to_string(), + /*trigger_turn*/ true, + ); + let expected = vec![ + plaintext.to_model_input_item(), + encrypted.to_model_input_item(), + ]; + + let serialized = job::serialize_filtered_rollout_response_items(&[ + RolloutItem::InterAgentCommunication(plaintext), + RolloutItem::InterAgentCommunication(encrypted), + ]) + .expect("serialize"); + let parsed: Vec = serde_json::from_str(&serialized).expect("parse"); + + assert_eq!(parsed, expected); + } + #[test] fn count_outcomes_sums_token_usage_across_all_jobs() { let counts = aggregate_stats(vec![ diff --git a/codex-rs/protocol/src/models.rs b/codex-rs/protocol/src/models.rs index 5df6d1e77..9d4647b30 100644 --- a/codex-rs/protocol/src/models.rs +++ b/codex-rs/protocol/src/models.rs @@ -719,6 +719,7 @@ pub enum ContentItem { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, JsonSchema, TS)] #[serde(tag = "type", rename_all = "snake_case")] pub enum AgentMessageInputContent { + InputText { text: String }, EncryptedContent { encrypted_content: String }, } diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 07ebd3026..f3b9a8ff3 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -528,7 +528,7 @@ pub enum Op { thread_settings: ThreadSettingsOverrides, }, - /// Inter-agent communication that should be recorded as assistant history + /// Inter-agent communication that should be recorded as agent-message history /// while still using the normal thread submission lifecycle. InterAgentCommunication { communication: InterAgentCommunication, @@ -714,15 +714,18 @@ impl InterAgentCommunication { } pub fn to_model_input_item(&self) -> ResponseItem { - match &self.encrypted_content { - Some(encrypted_content) => ResponseItem::AgentMessage { - author: self.author.to_string(), - recipient: self.recipient.to_string(), - content: vec![AgentMessageInputContent::EncryptedContent { - encrypted_content: encrypted_content.clone(), - }], + let content = match &self.encrypted_content { + Some(encrypted_content) => AgentMessageInputContent::EncryptedContent { + encrypted_content: encrypted_content.clone(), }, - None => self.to_response_input_item().into(), + None => AgentMessageInputContent::InputText { + text: self.content.clone(), + }, + }; + ResponseItem::AgentMessage { + author: self.author.to_string(), + recipient: self.recipient.to_string(), + content: vec![content], } } @@ -2800,6 +2803,7 @@ fn multi_agent_version_from_items( RolloutItem::TurnContext(turn_context) => turn_context.multi_agent_version, RolloutItem::SessionMeta(_) | RolloutItem::ResponseItem(_) + | RolloutItem::InterAgentCommunication(_) | RolloutItem::Compacted(_) | RolloutItem::EventMsg(_) => None, }) @@ -2895,6 +2899,8 @@ pub struct SessionMetaLine { pub enum RolloutItem { SessionMeta(SessionMetaLine), ResponseItem(ResponseItem), + /// Durable delivery metadata reconstructed as a model-visible `agent_message`. + InterAgentCommunication(InterAgentCommunication), Compacted(CompactedItem), TurnContext(TurnContextItem), EventMsg(EventMsg), diff --git a/codex-rs/rollout-trace/src/model/conversation.rs b/codex-rs/rollout-trace/src/model/conversation.rs index 1a8af0270..62f60ebae 100644 --- a/codex-rs/rollout-trace/src/model/conversation.rs +++ b/codex-rs/rollout-trace/src/model/conversation.rs @@ -3,6 +3,7 @@ use serde::Serialize; use crate::payload::RawPayloadId; +use super::AgentPath; use super::AgentThreadId; use super::CodeCellId; use super::CodexTurnId; @@ -32,6 +33,9 @@ pub struct ConversationItem { /// Codex channel for assistant/tool content, when the item is channel-specific. pub channel: Option, pub kind: ConversationItemKind, + /// Routing metadata carried by a Responses `agent_message` item. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub agent_message: Option, pub body: ConversationBody, /// Protocol/model `call_id` for function/custom tool call and output items. pub call_id: Option, @@ -39,6 +43,15 @@ pub struct ConversationItem { pub produced_by: Vec, } +/// Sender and destination identities attached to a model-visible agent message. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct AgentMessageMetadata { + /// Agent path that authored the message. + pub author: AgentPath, + /// Agent path that received the message. + pub recipient: AgentPath, +} + /// Model-visible role assigned to a conversation item. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] diff --git a/codex-rs/rollout-trace/src/reducer/conversation.rs b/codex-rs/rollout-trace/src/reducer/conversation.rs index db9752d24..f105ffebd 100644 --- a/codex-rs/rollout-trace/src/reducer/conversation.rs +++ b/codex-rs/rollout-trace/src/reducer/conversation.rs @@ -322,6 +322,7 @@ impl TraceReducer { role: ConversationRole::Assistant, channel: None, kind: ConversationItemKind::CompactionMarker, + agent_message: None, // The summary is a separate model/provider-visible item. Keep the marker body // empty so transcript renderers cannot mistake the boundary for prompt content. body: ConversationBody { parts: Vec::new() }, @@ -419,6 +420,7 @@ impl TraceReducer { role: item.role, channel: item.channel, kind: item.kind, + agent_message: item.agent_message, body: item.body, call_id: item.call_id, produced_by, @@ -579,6 +581,7 @@ fn conversation_item_matches( item.role == normalized.role && item.channel == normalized.channel && item.kind == normalized.kind + && item.agent_message == normalized.agent_message && body_matches && item.call_id == normalized.call_id } diff --git a/codex-rs/rollout-trace/src/reducer/conversation/normalize.rs b/codex-rs/rollout-trace/src/reducer/conversation/normalize.rs index 92136a90f..e67673b96 100644 --- a/codex-rs/rollout-trace/src/reducer/conversation/normalize.rs +++ b/codex-rs/rollout-trace/src/reducer/conversation/normalize.rs @@ -1,9 +1,13 @@ //! Normalization from Responses-shaped JSON items into conversation item data. +use anyhow::Context; use anyhow::Result; use anyhow::bail; +use codex_protocol::models::AgentMessageInputContent; +use codex_protocol::models::ResponseItem; use serde_json::Value; +use crate::model::AgentMessageMetadata; use crate::model::ConversationBody; use crate::model::ConversationChannel; use crate::model::ConversationItemKind; @@ -22,6 +26,7 @@ pub(super) struct NormalizedConversationItem { pub(super) role: ConversationRole, pub(super) channel: Option, pub(super) kind: ConversationItemKind, + pub(super) agent_message: Option, pub(super) body: ConversationBody, pub(super) call_id: Option, } @@ -58,11 +63,13 @@ fn normalize_model_item( }; match item_type { "message" => normalize_message_item(item, raw_payload), + "agent_message" => normalize_agent_message_item(item, raw_payload), "reasoning" => normalize_reasoning_item(item, raw_payload), "function_call" => Ok(NormalizedConversationItem { role: ConversationRole::Assistant, channel: Some(ConversationChannel::Commentary), kind: ConversationItemKind::FunctionCall, + agent_message: None, body: raw_text_or_json_body(item.get("arguments"), raw_payload), call_id: item .get("call_id") @@ -73,6 +80,7 @@ fn normalize_model_item( role: ConversationRole::Tool, channel: Some(ConversationChannel::Commentary), kind: ConversationItemKind::FunctionCallOutput, + agent_message: None, body: tool_output_body(item.get("output"), raw_payload), call_id: item .get("call_id") @@ -83,6 +91,7 @@ fn normalize_model_item( role: ConversationRole::Assistant, channel: Some(ConversationChannel::Commentary), kind: ConversationItemKind::CustomToolCall, + agent_message: None, body: custom_tool_call_body(item, raw_payload), call_id: item .get("call_id") @@ -93,6 +102,7 @@ fn normalize_model_item( role: ConversationRole::Tool, channel: Some(ConversationChannel::Commentary), kind: ConversationItemKind::CustomToolCallOutput, + agent_message: None, body: tool_output_body(item.get("output"), raw_payload), call_id: item .get("call_id") @@ -104,6 +114,7 @@ fn normalize_model_item( role: ConversationRole::Assistant, channel: Some(ConversationChannel::Commentary), kind: ConversationItemKind::FunctionCall, + agent_message: None, body: json_body(item, raw_payload), call_id: item .get("call_id") @@ -115,6 +126,7 @@ fn normalize_model_item( role: ConversationRole::Tool, channel: Some(ConversationChannel::Commentary), kind: ConversationItemKind::FunctionCallOutput, + agent_message: None, body: json_body(item, raw_payload), call_id: item .get("call_id") @@ -126,6 +138,7 @@ fn normalize_model_item( role: ConversationRole::Assistant, channel: Some(ConversationChannel::Summary), kind: ConversationItemKind::Message, + agent_message: None, body: compaction_body(item, raw_payload)?, call_id: None, }) @@ -160,6 +173,7 @@ fn normalize_message_item( .and_then(Value::as_str) .and_then(channel_from_phase), kind: ConversationItemKind::Message, + agent_message: None, body: ConversationBody { parts: content_parts(item.get("content"), raw_payload), }, @@ -167,6 +181,49 @@ fn normalize_message_item( }) } +fn normalize_agent_message_item( + item: &Value, + raw_payload: &RawPayloadRef, +) -> Result { + let raw_payload_id = &raw_payload.raw_payload_id; + let response_item = + serde_json::from_value::(item.clone()).with_context(|| { + format!("failed to parse agent_message item in payload {raw_payload_id}") + })?; + let ResponseItem::AgentMessage { + author, + recipient, + content, + } = response_item + else { + bail!("item in payload {raw_payload_id} was not an agent_message"); + }; + let parts = content + .into_iter() + .map(|content| match content { + AgentMessageInputContent::InputText { text } => ConversationPart::Text { text }, + AgentMessageInputContent::EncryptedContent { encrypted_content } => { + ConversationPart::Encoded { + label: "encrypted_content".to_string(), + value: encrypted_content, + } + } + }) + .collect::>(); + if parts.is_empty() { + bail!("agent_message item in payload {raw_payload_id} contained no content"); + } + + Ok(NormalizedConversationItem { + role: ConversationRole::Assistant, + channel: Some(ConversationChannel::Analysis), + kind: ConversationItemKind::Message, + agent_message: Some(AgentMessageMetadata { author, recipient }), + body: ConversationBody { parts }, + call_id: None, + }) +} + fn normalize_reasoning_item( item: &Value, raw_payload: &RawPayloadRef, @@ -217,6 +274,7 @@ fn normalize_reasoning_item( role: ConversationRole::Assistant, channel: Some(ConversationChannel::Analysis), kind: ConversationItemKind::Reasoning, + agent_message: None, body: ConversationBody { parts }, call_id: None, }) diff --git a/codex-rs/rollout-trace/src/reducer/conversation_tests.rs b/codex-rs/rollout-trace/src/reducer/conversation_tests.rs index a7acd3682..1cf13b1fa 100644 --- a/codex-rs/rollout-trace/src/reducer/conversation_tests.rs +++ b/codex-rs/rollout-trace/src/reducer/conversation_tests.rs @@ -2,9 +2,12 @@ use pretty_assertions::assert_eq; use serde_json::json; use tempfile::TempDir; +use crate::model::AgentMessageMetadata; +use crate::model::ConversationBody; use crate::model::ConversationChannel; use crate::model::ConversationItemKind; use crate::model::ConversationPart; +use crate::model::ConversationRole; use crate::model::ExecutionStatus; use crate::model::ProducerRef; use crate::model::ToolCallKind; @@ -107,6 +110,90 @@ fn response_outputs_enter_thread_conversation_on_completion() -> anyhow::Result< Ok(()) } +#[test] +fn agent_messages_preserve_routing_and_content() -> anyhow::Result<()> { + let temp = TempDir::new()?; + let writer = create_started_writer(&temp)?; + start_turn(&writer, "turn-1")?; + + let request = writer.write_json_payload( + RawPayloadKind::InferenceRequest, + &json!({ + "input": [ + { + "type": "agent_message", + "author": "/root/worker", + "recipient": "/root", + "content": [{"type": "input_text", "text": "done"}] + }, + { + "type": "agent_message", + "author": "/root", + "recipient": "/root/worker", + "content": [{ + "type": "encrypted_content", + "encrypted_content": "encrypted-task" + }] + } + ] + }), + )?; + append_inference_start(&writer, "inference-1", "turn-1", request)?; + + let rollout = replay_bundle(temp.path())?; + let actual = rollout.inference_calls["inference-1"] + .request_item_ids + .iter() + .map(|item_id| { + let item = &rollout.conversation_items[item_id]; + ( + item.role.clone(), + item.channel.clone(), + item.kind.clone(), + item.agent_message.clone(), + item.body.clone(), + ) + }) + .collect::>(); + + assert_eq!( + actual, + vec![ + ( + ConversationRole::Assistant, + Some(ConversationChannel::Analysis), + ConversationItemKind::Message, + Some(AgentMessageMetadata { + author: "/root/worker".to_string(), + recipient: "/root".to_string(), + }), + ConversationBody { + parts: vec![ConversationPart::Text { + text: "done".to_string(), + }], + }, + ), + ( + ConversationRole::Assistant, + Some(ConversationChannel::Analysis), + ConversationItemKind::Message, + Some(AgentMessageMetadata { + author: "/root".to_string(), + recipient: "/root/worker".to_string(), + }), + ConversationBody { + parts: vec![ConversationPart::Encoded { + label: "encrypted_content".to_string(), + value: "encrypted-task".to_string(), + }], + }, + ), + ] + ); + + Ok(()) +} + #[test] fn later_full_request_reuses_prior_json_tool_call_by_position() -> anyhow::Result<()> { let temp = TempDir::new()?; diff --git a/codex-rs/rollout-trace/src/reducer/tool/agents.rs b/codex-rs/rollout-trace/src/reducer/tool/agents.rs index 656fccaf7..381c475b6 100644 --- a/codex-rs/rollout-trace/src/reducer/tool/agents.rs +++ b/codex-rs/rollout-trace/src/reducer/tool/agents.rs @@ -33,6 +33,7 @@ pub(in crate::reducer) struct PendingAgentInteractionEdge { pub(in crate::reducer) kind: InteractionEdgeKind, pub(in crate::reducer) source: TraceAnchor, pub(in crate::reducer) target_thread_id: String, + pub(in crate::reducer) message_author: String, pub(in crate::reducer) message_content: String, /// Spawn-only fallback for children that fail before their task message is model-visible. pub(in crate::reducer) unresolved_spawn_thread_id: Option, @@ -254,6 +255,7 @@ impl TraceReducer { format!("agent activity referenced unknown tool call {tool_call_id}") })?; let started_at_unix_ms = tool_call.execution.started_at_unix_ms; + let message_author = self.agent_path_for_thread(&tool_call.thread_id)?; let message_content = self.agent_message_content_from_invocation(tool_call_id)?; let carried_raw_payload_ids = self.agent_tool_payload_ids(tool_call_id)?; self.queue_or_resolve_agent_interaction_edge(PendingAgentInteractionEdge { @@ -263,6 +265,7 @@ impl TraceReducer { tool_call_id: tool_call_id.to_string(), }, target_thread_id, + message_author, message_content, unresolved_spawn_thread_id, started_at_unix_ms, @@ -354,6 +357,7 @@ impl TraceReducer { let tool_call = &self.rollout.tool_calls[tool_call_id]; let child_thread_id = child_thread_id.to_string(); let edge_id = spawn_edge_id(&payload.sender_thread_id.to_string(), &child_thread_id); + let message_author = self.agent_path_for_thread(&tool_call.thread_id)?; self.queue_or_resolve_agent_interaction_edge(PendingAgentInteractionEdge { edge_id, @@ -362,6 +366,7 @@ impl TraceReducer { tool_call_id: tool_call_id.to_string(), }, target_thread_id: child_thread_id.clone(), + message_author, message_content: payload.prompt.clone(), unresolved_spawn_thread_id: Some(child_thread_id), started_at_unix_ms: tool_call.execution.started_at_unix_ms, @@ -395,6 +400,7 @@ impl TraceReducer { ended_at_unix_ms: Option, ) -> Result<()> { let tool_call = &self.rollout.tool_calls[tool_call_id]; + let message_author = self.agent_path_for_thread(&tool_call.thread_id)?; self.queue_or_resolve_agent_interaction_edge(PendingAgentInteractionEdge { edge_id: tool_edge_id(tool_call_id), kind, @@ -402,6 +408,7 @@ impl TraceReducer { tool_call_id: tool_call_id.to_string(), }, target_thread_id, + message_author, message_content, unresolved_spawn_thread_id: None, started_at_unix_ms: tool_call.execution.started_at_unix_ms, @@ -469,6 +476,7 @@ impl TraceReducer { &mut self, observed: ObservedAgentResultEdge, ) -> Result<()> { + let message_author = self.agent_path_for_thread(&observed.child_thread_id)?; let source = if let Some(source_item_id) = self.latest_assistant_message_item_for_turn( &observed.child_thread_id, &observed.child_codex_turn_id, @@ -492,6 +500,7 @@ impl TraceReducer { kind: InteractionEdgeKind::AgentResult, source, target_thread_id: observed.parent_thread_id, + message_author, message_content: observed.message, unresolved_spawn_thread_id: None, started_at_unix_ms: observed.wall_time_unix_ms, @@ -508,14 +517,21 @@ impl TraceReducer { &mut self, item_id: &str, ) -> Result<()> { - let Some((thread_id, message_content)) = self.inter_agent_message_item(item_id) else { + if self.is_interaction_edge_target_item(item_id) { + return Ok(()); + } + let Some((thread_id, message_author, message_content)) = + self.inter_agent_message_item(item_id) + else { return Ok(()); }; let Some(pending_index) = self .pending_agent_interaction_edges .iter() .position(|pending| { - pending.target_thread_id == thread_id && pending.message_content == message_content + pending.target_thread_id == thread_id + && pending.message_author == message_author + && pending.message_content == message_content }) else { return Ok(()); @@ -530,6 +546,7 @@ impl TraceReducer { ) -> Result<()> { if let Some(item_id) = self.find_unlinked_inter_agent_message_item( &pending.target_thread_id, + &pending.message_author, &pending.message_content, ) { return self.upsert_agent_interaction_edge_for_item(pending, item_id); @@ -543,6 +560,7 @@ impl TraceReducer { if existing.kind != pending.kind || existing.source != pending.source || existing.target_thread_id != pending.target_thread_id + || existing.message_author != pending.message_author || existing.message_content != pending.message_content || existing.unresolved_spawn_thread_id != pending.unresolved_spawn_thread_id { @@ -661,6 +679,7 @@ impl TraceReducer { fn find_unlinked_inter_agent_message_item( &self, thread_id: &str, + message_author: &str, message_content: &str, ) -> Option { self.rollout @@ -672,19 +691,30 @@ impl TraceReducer { !self.is_interaction_edge_target_item(item_id) && self .inter_agent_message_item(item_id) - .is_some_and(|(_, content)| content == message_content) + .is_some_and(|(_, author, content)| { + author == message_author && content == message_content + }) }) .cloned() } - fn inter_agent_message_item(&self, item_id: &str) -> Option<(String, String)> { + fn inter_agent_message_item(&self, item_id: &str) -> Option<(String, String, String)> { let item = self.rollout.conversation_items.get(item_id)?; - let (recipient_agent_path, message_content) = inter_agent_message_fields(item)?; + let (author_agent_path, recipient_agent_path, message_content) = + inter_agent_message_fields(item)?; let thread = self.rollout.threads.get(&item.thread_id)?; if recipient_agent_path != thread.agent_path { return None; } - Some((item.thread_id.clone(), message_content)) + Some((item.thread_id.clone(), author_agent_path, message_content)) + } + + fn agent_path_for_thread(&self, thread_id: &str) -> Result { + self.rollout + .threads + .get(thread_id) + .map(|thread| thread.agent_path.clone()) + .with_context(|| format!("agent edge referenced unknown thread {thread_id}")) } fn is_interaction_edge_target_item(&self, item_id: &str) -> bool { @@ -707,6 +737,7 @@ impl TraceReducer { && item.codex_turn_id.as_deref() == Some(codex_turn_id) && item.role == ConversationRole::Assistant && item.kind == ConversationItemKind::Message + && item.agent_message.is_none() }) .max_by_key(|item| item.first_seen_at_unix_ms) .map(|item| item.item_id.clone()) @@ -735,19 +766,36 @@ fn push_unique(items: &mut Vec, item: &str) { } } -fn inter_agent_message_fields(item: &ConversationItem) -> Option<(String, String)> { - // Multi-agent v2 injects mailbox deliveries as assistant messages whose - // text is serialized `InterAgentCommunication`. Treat only that exact - // transport shape as an edge target; ordinary assistant JSON must not be - // mistaken for cross-thread delivery. +fn inter_agent_message_fields(item: &ConversationItem) -> Option<(String, String, String)> { if item.role != ConversationRole::Assistant || item.kind != ConversationItemKind::Message { return None; } + if let Some(agent_message) = &item.agent_message { + let [content] = item.body.parts.as_slice() else { + return None; + }; + let message_content = match content { + ConversationPart::Text { text } => text, + ConversationPart::Encoded { label, value } if label == "encrypted_content" => value, + _ => return None, + }; + return Some(( + agent_message.author.clone(), + agent_message.recipient.clone(), + message_content.clone(), + )); + } + + // Older traces store multi-agent v2 deliveries as assistant messages whose + // text is serialized `InterAgentCommunication`. Treat only that exact + // transport shape as an edge target; ordinary assistant JSON must not be + // mistaken for cross-thread delivery. let [ConversationPart::Text { text }] = item.body.parts.as_slice() else { return None; }; let communication = serde_json::from_str::(text).ok()?; Some(( + communication.author.to_string(), communication.recipient.to_string(), communication .encrypted_content diff --git a/codex-rs/rollout-trace/src/reducer/tool/agents_tests.rs b/codex-rs/rollout-trace/src/reducer/tool/agents_tests.rs index b009a1543..db8fcc007 100644 --- a/codex-rs/rollout-trace/src/reducer/tool/agents_tests.rs +++ b/codex-rs/rollout-trace/src/reducer/tool/agents_tests.rs @@ -246,18 +246,17 @@ fn sub_agent_started_activity_creates_spawn_edge() -> anyhow::Result<()> { )?; start_thread(&writer, child_thread_id, "/root/reviewer")?; start_turn_for_thread(&writer, child_thread_id, "turn-child-1")?; - let delivered = inter_agent_message( - "/root", - "/root/reviewer", - "review this", - /*trigger_turn*/ true, - ); append_inference_request( &writer, child_thread_id, "turn-child-1", "inference-child-1", - vec![message("assistant", &delivered)], + vec![json!({ + "type": "agent_message", + "author": "/root", + "recipient": "/root/reviewer", + "content": [{"type": "input_text", "text": "review this"}] + })], )?; let replayed = replay_bundle(temp.path())?; @@ -773,10 +772,9 @@ fn agent_result_edge_falls_back_to_child_thread_without_result_message() -> anyh let temp = TempDir::new()?; let writer = create_started_agent_writer(&temp)?; - // The child thread and turn exist, but there is intentionally no completed - // assistant message for this turn. Failed child tasks can still notify the - // parent through AgentStatus, so the result edge must not require a final - // transcript item from the child. + // The child received its task but produced no assistant output. Failed + // child tasks can still notify the parent through AgentStatus, so the + // inbound task must not be mistaken for the child's result. start_thread( &writer, "019d0000-0000-7000-8000-000000000002", @@ -787,6 +785,18 @@ fn agent_result_edge_falls_back_to_child_thread_without_result_message() -> anyh "019d0000-0000-7000-8000-000000000002", "turn-child-1", )?; + append_inference_request( + &writer, + "019d0000-0000-7000-8000-000000000002", + "turn-child-1", + "inference-child-1", + vec![json!({ + "type": "agent_message", + "author": "/root", + "recipient": "/root/child", + "content": [{"type": "input_text", "text": "do the task"}] + })], + )?; let notification = r#"{"agent_path":"/root/child","status":{"failed":"boom"}}"#; let carried_payload = writer.write_json_payload( diff --git a/codex-rs/rollout/src/list.rs b/codex-rs/rollout/src/list.rs index 8cad5d546..845145aae 100644 --- a/codex-rs/rollout/src/list.rs +++ b/codex-rs/rollout/src/list.rs @@ -1117,11 +1117,10 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result { - summary.created_at = summary + RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) => { + summary .created_at - .clone() - .or_else(|| Some(rollout_line.timestamp.clone())); + .get_or_insert_with(|| rollout_line.timestamp.clone()); } RolloutItem::TurnContext(_) => { // Not included in `head`; skip. @@ -1180,6 +1179,11 @@ pub async fn read_head_for_summary(path: &Path) -> io::Result { + if let Ok(value) = serde_json::to_value(communication.to_model_input_item()) { + head.push(value); + } + } RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) => {} diff --git a/codex-rs/rollout/src/metadata.rs b/codex-rs/rollout/src/metadata.rs index 97aa415db..19e9a9320 100644 --- a/codex-rs/rollout/src/metadata.rs +++ b/codex-rs/rollout/src/metadata.rs @@ -68,6 +68,7 @@ pub fn builder_from_items( if let Some(session_meta) = items.iter().find_map(|item| match item { RolloutItem::SessionMeta(meta_line) => Some(meta_line), RolloutItem::ResponseItem(_) + | RolloutItem::InterAgentCommunication(_) | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) => None, @@ -120,6 +121,7 @@ pub async fn extract_metadata_from_rollout( memory_mode: items.iter().rev().find_map(|item| match item { RolloutItem::SessionMeta(meta_line) => meta_line.meta.memory_mode.clone(), RolloutItem::ResponseItem(_) + | RolloutItem::InterAgentCommunication(_) | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) => None, diff --git a/codex-rs/rollout/src/policy.rs b/codex-rs/rollout/src/policy.rs index 9760191bd..40ca77c97 100644 --- a/codex-rs/rollout/src/policy.rs +++ b/codex-rs/rollout/src/policy.rs @@ -6,6 +6,7 @@ use codex_protocol::models::ResponseItem; pub fn is_persisted_rollout_item(item: &RolloutItem) -> bool { match item { RolloutItem::ResponseItem(item) => should_persist_response_item(item), + RolloutItem::InterAgentCommunication(_) => true, RolloutItem::EventMsg(ev) => should_persist_event_msg(ev), // Persist Codex executive markers so we can analyze flows (e.g., compaction, API turns). RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::SessionMeta(_) => { diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index d102d5c54..9ade47489 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -868,28 +868,17 @@ impl RolloutRecorder { // Parse the rollout line structure match serde_json::from_value::(v.clone()) { - Ok(rollout_line) => match rollout_line.item { - RolloutItem::SessionMeta(session_meta_line) => { - // Use the FIRST SessionMeta encountered in the file as the canonical - // thread id and main session information. Keep all items intact. - if thread_id.is_none() { - thread_id = Some(session_meta_line.meta.id); - } - items.push(RolloutItem::SessionMeta(session_meta_line)); + Ok(rollout_line) => { + let item = rollout_line.item; + // Use the FIRST SessionMeta encountered in the file as the canonical + // thread id and main session information. Keep all items intact. + if thread_id.is_none() + && let RolloutItem::SessionMeta(session_meta_line) = &item + { + thread_id = Some(session_meta_line.meta.id); } - RolloutItem::ResponseItem(item) => { - items.push(RolloutItem::ResponseItem(item)); - } - RolloutItem::Compacted(item) => { - items.push(RolloutItem::Compacted(item)); - } - RolloutItem::TurnContext(item) => { - items.push(RolloutItem::TurnContext(item)); - } - RolloutItem::EventMsg(_ev) => { - items.push(RolloutItem::EventMsg(_ev)); - } - }, + items.push(item); + } Err(e) => { trace!("failed to parse rollout line: {e}"); parse_errors = parse_errors.saturating_add(1); @@ -1775,6 +1764,7 @@ async fn resume_candidate_matches_cwd( RolloutItem::TurnContext(turn_context) => Some(turn_context.cwd.as_path()), RolloutItem::SessionMeta(_) | RolloutItem::ResponseItem(_) + | RolloutItem::InterAgentCommunication(_) | RolloutItem::Compacted(_) | RolloutItem::EventMsg(_) => None, }) diff --git a/codex-rs/rollout/src/search.rs b/codex-rs/rollout/src/search.rs index 622fac8b0..799e92717 100644 --- a/codex-rs/rollout/src/search.rs +++ b/codex-rs/rollout/src/search.rs @@ -283,6 +283,7 @@ fn conversation_text_from_item(item: &RolloutItem) -> Option { | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) | RolloutItem::ResponseItem(_) + | RolloutItem::InterAgentCommunication(_) | RolloutItem::Compacted(_) => None, } } diff --git a/codex-rs/state/src/extract.rs b/codex-rs/state/src/extract.rs index c1c2e56c9..fbb63dcea 100644 --- a/codex-rs/state/src/extract.rs +++ b/codex-rs/state/src/extract.rs @@ -22,6 +22,7 @@ pub fn apply_rollout_item( RolloutItem::TurnContext(turn_ctx) => apply_turn_context(metadata, turn_ctx), RolloutItem::EventMsg(event) => apply_event_msg(metadata, event), RolloutItem::ResponseItem(item) => apply_response_item(metadata, item), + RolloutItem::InterAgentCommunication(_) => {} RolloutItem::Compacted(_) => {} } if metadata.model_provider.is_empty() { @@ -36,9 +37,10 @@ pub fn rollout_item_affects_thread_metadata(item: &RolloutItem) -> bool { RolloutItem::EventMsg( EventMsg::TokenCount(_) | EventMsg::UserMessage(_) | EventMsg::ThreadGoalUpdated(_), ) => true, - RolloutItem::EventMsg(_) | RolloutItem::ResponseItem(_) | RolloutItem::Compacted(_) => { - false - } + RolloutItem::EventMsg(_) + | RolloutItem::ResponseItem(_) + | RolloutItem::InterAgentCommunication(_) + | RolloutItem::Compacted(_) => false, } } diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index 98eff16c5..40267f046 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -1078,6 +1078,7 @@ pub(super) fn extract_memory_mode(items: &[RolloutItem]) -> Option { items.iter().rev().find_map(|item| match item { RolloutItem::SessionMeta(meta_line) => meta_line.meta.memory_mode.clone(), RolloutItem::ResponseItem(_) + | RolloutItem::InterAgentCommunication(_) | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) => None, diff --git a/codex-rs/thread-store/src/thread_metadata_sync.rs b/codex-rs/thread-store/src/thread_metadata_sync.rs index 3a379a481..8eb23e363 100644 --- a/codex-rs/thread-store/src/thread_metadata_sync.rs +++ b/codex-rs/thread-store/src/thread_metadata_sync.rs @@ -272,6 +272,7 @@ impl ThreadMetadataSync { RolloutItem::SessionMeta(_) | RolloutItem::EventMsg(_) | RolloutItem::ResponseItem(_) + | RolloutItem::InterAgentCommunication(_) | RolloutItem::Compacted(_) => {} } }