agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Support plaintext agent messages (#27830)

Browse Source 
## 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`
This commit is contained in:
jif
2026-06-12 21:50:04 +01:00
committed by GitHub
Unverified
parent 3e2ee1da3f
commit 8f2d6416ce
44 changed files with 716 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/json/ClientRequest.json
Generated
+20
View File
@@ -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": {
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
Generated
+20
View File
@@ -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": {
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
Generated
+20
View File
@@ -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": {
codex-rs/app-server-protocol/schema/json/v2/RawResponseItemCompletedNotification.json
Generated
+20
View File
@@ -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": {
codex-rs/app-server-protocol/schema/json/v2/ThreadResumeParams.json
Generated
+20
View File
@@ -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": {
codex-rs/app-server-protocol/schema/typescript/AgentMessageInputContent.ts
Generated
+1 -1
View File
@@ -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, };
codex-rs/app-server-protocol/src/protocol/thread_history.rs
+3 -1
View File
@@ -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(_) => {}
}
}
codex-rs/core/src/agent/control/spawn.rs
+1
View File
@@ -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.
codex-rs/core/src/client_common.rs
+1 -21
View File
@@ -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<ResponseItem> {
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<ResponseItem> {
let mut input = self.get_formatted_input();
let mut input = self.input.clone();
if use_responses_lite {
strip_image_details(&mut input);
}
codex-rs/core/src/guardian/prompt.rs
+17
View File
@@ -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::<Vec<_>>()
.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(),
codex-rs/core/src/hook_runtime.rs
+8
View File
@@ -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;
}
codex-rs/core/src/prompt_debug.rs
+1 -1
View File
@@ -98,5 +98,5 @@ pub(crate) async fn build_prompt_input_from_session(
base_instructions,
);
Ok(prompt.get_formatted_input())
Ok(prompt.input)
}
codex-rs/core/src/realtime_context.rs
+18
View File
@@ -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<String> {
}
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::<Vec<_>>()
.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}"));
}
_ => {}
}
}
codex-rs/core/src/session/handlers.rs
+1 -1
View File
@@ -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<Session>,
codex-rs/core/src/session/input_queue.rs
+7 -10
View File
@@ -16,6 +16,7 @@ pub(crate) enum TurnInput {
client_id: Option<String>,
},
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<ResponseItem> {
pub(crate) async fn drain_mailbox_input_items(&self) -> Vec<TurnInput> {
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);
codex-rs/core/src/session/mod.rs
+20
View File
@@ -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<Self>,
turn_context: &Arc<TurnContext>,
codex-rs/core/src/session/rollout_reconstruction.rs
+12
View File
@@ -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)
codex-rs/core/src/session/rollout_reconstruction_tests.rs
+25
View File
@@ -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()
{
codex-rs/core/src/session/tests.rs
+4 -8
View File
@@ -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)],
);
}
codex-rs/core/src/session/turn.rs
+2 -2
View File
@@ -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| {
codex-rs/core/src/tasks/review.rs
+1 -1
View File
@@ -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(_) => {}
}
}
codex-rs/core/src/thread_rollout_truncation.rs
+9 -1
View File
@@ -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<us
///
/// A fork-turn boundary is either:
/// - a real user message boundary, or
/// - an assistant inter-agent envelope whose parsed `trigger_turn` is `true`.
/// - an inter-agent communication whose `trigger_turn` is `true`, or
/// - a legacy assistant inter-agent envelope with the same flag.
///
/// Like `user_message_positions_in_rollout`, this applies `ThreadRolledBack` markers so indexing
/// reflects the effective post-rollback history. Rollback counts instruction turns, so a rollback
@@ -77,6 +79,12 @@ pub(crate) fn fork_turn_positions_in_rollout(items: &[RolloutItem]) -> Vec<usize
fork_turn_positions.push(idx);
}
}
RolloutItem::InterAgentCommunication(communication) => {
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 {
codex-rs/core/src/thread_rollout_truncation_tests.rs
+24
View File
@@ -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![
codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_commentary.snap
+1 -1
View File
@@ -14,4 +14,4 @@ Scenario: /responses POST bodies (input only, redacted like other suite snapshot
01:message/user:<ENVIRONMENT_CONTEXT:cwd=<CWD>>
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
codex-rs/core/tests/suite/snapshots/all__suite__pending_input__pending_input_queued_mail_after_reasoning.snap
+1 -1
View File
@@ -14,4 +14,4 @@ Scenario: /responses POST bodies (input only, redacted like other suite snapshot
01:message/user:<ENVIRONMENT_CONTEXT:cwd=<CWD>>
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
codex-rs/core/tests/suite/subagent_notifications.rs
+108
View File
@@ -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, "<subagent_notification>")
},
sse(vec![
ev_response_created("resp-parent-2"),
ev_assistant_message("msg-parent-2", "parent done"),
ev_completed("resp-parent-2"),
]),
)
.await;
let notification = "<subagent_notification>\n{\"agent_path\":\"/root/worker\",\"status\":{\"completed\":\"child done\"}}\n</subagent_notification>";
// 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, "<subagent_notification>")
},
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, "<subagent_notification>")
},
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(()));
codex-rs/ext/web-search/src/history.rs
+23
View File
@@ -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<ResponseItem>, 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::<Vec<_>>()
.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,
codex-rs/memories/write/src/phase1.rs
+41 -5
View File
@@ -405,12 +405,15 @@ mod job {
) -> codex_protocol::error::Result<String> {
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::<Vec<_>>();
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<ResponseItem> = 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![
codex-rs/protocol/src/models.rs
+1
View File
@@ -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 },
}
codex-rs/protocol/src/protocol.rs
+15 -9
View File
@@ -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),
codex-rs/rollout-trace/src/model/conversation.rs
+13
View File
@@ -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<ConversationChannel>,
pub kind: ConversationItemKind,
/// Routing metadata carried by a Responses `agent_message` item.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub agent_message: Option<AgentMessageMetadata>,
pub body: ConversationBody,
/// Protocol/model `call_id` for function/custom tool call and output items.
pub call_id: Option<ModelVisibleCallId>,
@@ -39,6 +43,15 @@ pub struct ConversationItem {
pub produced_by: Vec<ProducerRef>,
}
/// 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")]
codex-rs/rollout-trace/src/reducer/conversation.rs
+3
View File
@@ -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
}
codex-rs/rollout-trace/src/reducer/conversation/normalize.rs
+58
View File
@@ -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<ConversationChannel>,
pub(super) kind: ConversationItemKind,
pub(super) agent_message: Option<AgentMessageMetadata>,
pub(super) body: ConversationBody,
pub(super) call_id: Option<String>,
}
@@ -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<NormalizedConversationItem> {
let raw_payload_id = &raw_payload.raw_payload_id;
let response_item =
serde_json::from_value::<ResponseItem>(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::<Vec<_>>();
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,
})
codex-rs/rollout-trace/src/reducer/conversation_tests.rs
+87
View File
@@ -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::<Vec<_>>();
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()?;
codex-rs/rollout-trace/src/reducer/tool/agents.rs
+59 -11
View File
@@ -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<String>,
@@ -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<i64>,
) -> 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<String> {
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<String> {
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<String>, 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::<InterAgentCommunication>(text).ok()?;
Some((
communication.author.to_string(),
communication.recipient.to_string(),
communication
.encrypted_content
codex-rs/rollout-trace/src/reducer/tool/agents_tests.rs
+21 -11
View File
@@ -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#"<subagent_notification>{"agent_path":"/root/child","status":{"failed":"boom"}}</subagent_notification>"#;
let carried_payload = writer.write_json_payload(
codex-rs/rollout/src/list.rs
+8 -4
View File
@@ -1117,11 +1117,10 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result<HeadTai
summary.saw_session_meta = true;
}
}
RolloutItem::ResponseItem(_) => {
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<Vec<serde_json::Va
head.push(value);
}
}
RolloutItem::InterAgentCommunication(communication) => {
if let Ok(value) = serde_json::to_value(communication.to_model_input_item()) {
head.push(value);
}
}
RolloutItem::Compacted(_)
| RolloutItem::TurnContext(_)
| RolloutItem::EventMsg(_) => {}
codex-rs/rollout/src/metadata.rs
+2
View File
@@ -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,
codex-rs/rollout/src/policy.rs
+1
View File
@@ -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(_) => {
codex-rs/rollout/src/recorder.rs
+11 -21
View File
@@ -868,28 +868,17 @@ impl RolloutRecorder {
// Parse the rollout line structure
match serde_json::from_value::<RolloutLine>(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,
})
codex-rs/rollout/src/search.rs
+1
View File
@@ -283,6 +283,7 @@ fn conversation_text_from_item(item: &RolloutItem) -> Option<String> {
| RolloutItem::TurnContext(_)
| RolloutItem::EventMsg(_)
| RolloutItem::ResponseItem(_)
| RolloutItem::InterAgentCommunication(_)
| RolloutItem::Compacted(_) => None,
}
}
codex-rs/state/src/extract.rs
+5 -3
View File
@@ -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,
}
}
codex-rs/state/src/runtime/threads.rs
+1
View File
@@ -1078,6 +1078,7 @@ pub(super) fn extract_memory_mode(items: &[RolloutItem]) -> Option<String> {
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,
codex-rs/thread-store/src/thread_metadata_sync.rs
+1
View File
@@ -272,6 +272,7 @@ impl ThreadMetadataSync {
RolloutItem::SessionMeta(_)
| RolloutItem::EventMsg(_)
| RolloutItem::ResponseItem(_)
| RolloutItem::InterAgentCommunication(_)
| RolloutItem::Compacted(_) => {}
}
}