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

Track parent turns for nested Codex requests (#35835)

Browse Source 
## What changed

- Propagate the initiating turn ID through agent spawns, follow-up tasks, reviews, and delegated Codex sessions.
- Add `parent_turn_id` to Responses client and turn metadata while keeping it out of external MCP metadata.
- Preserve parent-turn provenance across queued agent messages when their triggering parent is unambiguous.

## Testing

- Cover spawned, resumed, nested, reviewed, delegated, and WebSocket request metadata.
- Verify queued messages do not claim ambiguous or queue-only parent turns.

GitOrigin-RevId: 481fdebbe7df2031880fe259509273cce50b20a8
This commit is contained in:
Eddie Chen
2026-07-28 21:58:33 +00:00
committed by copyberry
parent 12b3e88028
commit 1def0a8925
34 changed files with 635 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control.rs
+32 -7
View File
@@ -70,6 +70,7 @@ pub(crate) struct SpawnAgentOptions {
pub(crate) fork_parent_spawn_call_id: Option<String>,
pub(crate) fork_mode: Option<SpawnAgentForkMode>,
pub(crate) parent_thread_id: Option<ThreadId>,
pub(crate) parent_turn_id: Option<String>,
pub(crate) environments: Option<Vec<TurnEnvironmentSelection>>,
}
@@ -143,11 +144,12 @@ impl AgentControl {
&self,
agent_id: ThreadId,
input: Vec<UserInput>,
parent_turn_id: Option<String>,
) -> CodexResult<String> {
let state = self.upgrade()?;
self.ensure_execution_capacity_for_turn_start(agent_id, /*starts_turn*/ true)
.await?;
self.send_input_after_capacity_check(agent_id, &state, input)
self.send_input_after_capacity_check(agent_id, &state, input, parent_turn_id)
.await
}
@@ -156,11 +158,12 @@ impl AgentControl {
agent_id: ThreadId,
state: &Arc<ThreadManagerState>,
input: Vec<UserInput>,
parent_turn_id: Option<String>,
) -> CodexResult<String> {
self.handle_thread_request_result(
agent_id,
state,
state.send_op(agent_id, input.into()).await,
state.send_op(agent_id, input.into(), parent_turn_id).await,
)
.await
}
@@ -170,6 +173,7 @@ impl AgentControl {
agent_id: ThreadId,
communication: InterAgentCommunication,
agent_communication_context: AgentCommunicationContext,
parent_turn_id: Option<String>,
) -> CodexResult<String> {
let state = self.upgrade()?;
self.ensure_execution_capacity_for_turn_start(agent_id, communication.trigger_turn)
@@ -179,6 +183,7 @@ impl AgentControl {
&state,
communication,
agent_communication_context,
parent_turn_id,
)
.await
}
@@ -189,9 +194,16 @@ impl AgentControl {
state: &Arc<ThreadManagerState>,
communication: InterAgentCommunication,
context: AgentCommunicationContext,
parent_turn_id: Option<String>,
) -> CodexResult<String> {
self.submit_inter_agent_communication(agent_id, state, communication, context)
.await
self.submit_inter_agent_communication(
agent_id,
state,
communication,
context,
parent_turn_id,
)
.await
}
async fn submit_inter_agent_communication(
@@ -200,15 +212,21 @@ impl AgentControl {
state: &Arc<ThreadManagerState>,
communication: InterAgentCommunication,
context: AgentCommunicationContext,
parent_turn_id: Option<String>,
) -> CodexResult<String> {
let communication_for_log =
crate::agent_communication::logging_enabled().then(|| communication.clone());
let parent_turn_id = parent_turn_id.filter(|_| communication.trigger_turn);
let result = self
.handle_thread_request_result(
agent_id,
state,
state
.send_op(agent_id, Op::InterAgentCommunication { communication })
.send_op(
agent_id,
Op::InterAgentCommunication { communication },
parent_turn_id,
)
.await,
)
.await;
@@ -231,7 +249,9 @@ impl AgentControl {
self.handle_thread_request_result(
agent_id,
&state,
state.send_op(agent_id, Op::Interrupt).await,
state
.send_op(agent_id, Op::Interrupt, /*parent_turn_id*/ None)
.await,
)
.await
}
@@ -507,7 +527,12 @@ impl AgentControl {
let context =
AgentCommunicationContext::new(AgentCommunicationKind::Result, child_thread_id);
let _ = control
.send_inter_agent_communication(parent_thread_id, communication, context)
.send_inter_agent_communication(
parent_thread_id,
communication,
context,
/*parent_turn_id*/ None,
)
.await;
return;
}
codex-rs/core/src/agent/control/legacy.rs
+6 -2
View File
@@ -12,12 +12,16 @@ impl AgentControl {
let result = if matches!(thread.agent_status().await, AgentStatus::Shutdown) {
Ok(String::new())
} else {
state.send_op(agent_id, Op::Shutdown {}).await
state
.send_op(agent_id, Op::Shutdown {}, /*parent_turn_id*/ None)
.await
};
thread.wait_until_terminated().await;
result
} else {
state.send_op(agent_id, Op::Shutdown {}).await
state
.send_op(agent_id, Op::Shutdown {}, /*parent_turn_id*/ None)
.await
};
let _ = state.remove_thread(&agent_id).await;
self.forget_v2_residency(agent_id);
codex-rs/core/src/agent/control/spawn.rs
+8 -2
View File
@@ -527,8 +527,13 @@ impl AgentControl {
match initial_input {
SpawnInitialInput::UserInput(input) => {
self.send_input_after_capacity_check(new_thread.thread_id, &state, input)
.await?;
self.send_input_after_capacity_check(
new_thread.thread_id,
&state,
input,
options.parent_turn_id,
)
.await?;
}
SpawnInitialInput::InterAgentCommunication(communication, context) => {
self.send_inter_agent_communication_after_capacity_check(
@@ -536,6 +541,7 @@ impl AgentControl {
&state,
communication,
context,
options.parent_turn_id,
)
.await?;
}
codex-rs/core/src/agent/control_tests.rs
+5
View File
@@ -369,6 +369,7 @@ async fn send_input_errors_when_manager_dropped() {
text: "hello".to_string(),
text_elements: Vec::new(),
}],
/*parent_turn_id*/ None,
)
.await
.expect_err("send_input should fail without a manager");
@@ -483,6 +484,7 @@ async fn send_input_errors_when_thread_missing() {
text: "hello".to_string(),
text_elements: Vec::new(),
}],
/*parent_turn_id*/ None,
)
.await
.expect_err("send_input should fail for missing thread");
@@ -555,6 +557,7 @@ async fn send_input_submits_user_message() {
text: "hello from tests".to_string(),
text_elements: Vec::new(),
}],
/*parent_turn_id*/ None,
)
.await
.expect("send_input should succeed");
@@ -598,6 +601,7 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig
thread_id,
communication.clone(),
AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()),
/*parent_turn_id*/ None,
)
.await
.expect("send_inter_agent_communication should succeed");
@@ -729,6 +733,7 @@ async fn ensure_v2_agent_loaded_reloads_registered_unloaded_agent() {
spawned_agent.thread_id,
communication.clone(),
AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()),
/*parent_turn_id*/ None,
)
.await
.expect("send_inter_agent_communication should succeed after reload");
codex-rs/core/src/codex_delegate.rs
+13 -7
View File
@@ -218,6 +218,7 @@ pub(crate) async fn run_codex_thread_one_shot(
// Use a child token so we can stop the delegate after completion without
// requiring the caller to cancel the parent token.
let child_cancel = cancel_token.child_token();
let parent_turn_id = parent_ctx.sub_id.clone();
let (session, io) = Box::pin(run_codex_thread_interactive(
config,
auth_manager,
@@ -233,13 +234,17 @@ pub(crate) async fn run_codex_thread_one_shot(
.await?;
// Send the initial input to kick off the one-shot turn.
io.submit(Op::UserInput {
items: input,
final_output_json_schema,
responsesapi_client_metadata: None,
additional_context: Default::default(),
thread_settings: Default::default(),
})
io.submit_with_trace(
Op::UserInput {
items: input,
final_output_json_schema,
responsesapi_client_metadata: None,
additional_context: Default::default(),
thread_settings: Default::default(),
},
/*trace*/ None,
Some(parent_turn_id),
)
.await?;
// Bridge events so we can observe completion and shut down automatically.
@@ -262,6 +267,7 @@ pub(crate) async fn run_codex_thread_one_shot(
op: Op::Shutdown {},
client_user_message_id: None,
trace: None,
parent_turn_id: None,
})
.await;
child_cancel.cancel();
codex-rs/core/src/codex_delegate_tests.rs
+2
View File
@@ -175,6 +175,7 @@ async fn forward_ops_preserves_submission_trace_context() {
),
tracestate: Some("vendor=state".to_string()),
}),
parent_turn_id: Some("parent-turn".to_string()),
};
tx_ops.send(submission.clone()).await.unwrap();
drop(tx_ops);
@@ -186,6 +187,7 @@ async fn forward_ops_preserves_submission_trace_context() {
assert_eq!(submission.id, forwarded.id);
assert_eq!(submission.op, forwarded.op);
assert_eq!(submission.trace, forwarded.trace);
assert_eq!(submission.parent_turn_id, forwarded.parent_turn_id);
timeout(Duration::from_secs(1), forward)
.await
codex-rs/core/src/codex_thread.rs
+5 -2
View File
@@ -275,7 +275,9 @@ impl CodexThread {
op: Op,
trace: Option<W3cTraceContext>,
) -> CodexResult<String> {
self.io.submit_with_trace(op, trace).await
self.io
.submit_with_trace(op, trace, /*parent_turn_id*/ None)
.await
}
pub async fn submit_user_input_with_client_user_message_id(
@@ -427,7 +429,8 @@ impl CodexThread {
}
/// Use sparingly: this is intended to be removed soon.
pub async fn submit_with_id(&self, sub: Submission) -> CodexResult<()> {
pub async fn submit_with_id(&self, mut sub: Submission) -> CodexResult<()> {
sub.parent_turn_id = None;
self.io.submit_with_id(sub).await
}
codex-rs/core/src/guardian/review_session.rs
+10 -5
View File
@@ -806,10 +806,8 @@ async fn run_review_on_session(
.and_then(|environment| environment.cwd().to_abs_path().ok())
.unwrap_or_else(|| params.parent_turn.config.cwd.clone());
let submit_result = run_before_review_deadline(
deadline,
params.external_cancel.as_ref(),
Box::pin(review_session.io.submit(Op::UserInput {
let submission = review_session.io.submit_with_trace(
Op::UserInput {
items: prompt_items.items,
final_output_json_schema: Some(params.schema.clone()),
responsesapi_client_metadata: None,
@@ -834,7 +832,14 @@ async fn run_review_on_session(
}),
..Default::default()
},
})),
},
/*trace*/ None,
Some(params.parent_turn.sub_id.clone()),
);
let submit_result = run_before_review_deadline(
deadline,
params.external_cancel.as_ref(),
Box::pin(submission),
)
.await;
let child_turn_id = match submit_result {
codex-rs/core/src/guardian/tests.rs
+4 -2
View File
@@ -1476,6 +1476,7 @@ async fn guardian_request_model_for_auto_review(
.auto_review_model_override = auto_review_model_override;
let parent_model = turn.model_info.slug.clone();
let preferred_model = turn.provider.approval_review_preferred_model().to_string();
let parent_turn_id = turn.sub_id.clone();
seed_guardian_parent_history(&session, &turn).await;
let (outcome, analytics_result) = run_guardian_review_session_for_test(
@@ -1500,8 +1501,9 @@ async fn guardian_request_model_for_auto_review(
};
let request = request_log.single_request();
let request_model = request
.body_json()
let request_body = request.body_json();
core_test_support::responses::assert_parent_turn(&request_body, Some(parent_turn_id.as_str()))?;
let request_model = request_body
.get("model")
.and_then(|value| value.as_str())
.expect("guardian request should include a model")
codex-rs/core/src/responses_metadata.rs
+10
View File
@@ -36,6 +36,7 @@ pub(crate) const TURN_STARTED_AT_UNIX_MS_KEY: &str = "turn_started_at_unix_ms";
pub(crate) const FORKED_FROM_THREAD_ID_KEY: &str = "forked_from_thread_id";
pub(crate) const PARENT_THREAD_ID_KEY: &str = "parent_thread_id";
pub(crate) const PARENT_TURN_ID_KEY: &str = "parent_turn_id";
pub(crate) const SUBAGENT_KIND_KEY: &str = "subagent_kind";
pub(crate) const THREAD_SOURCE_KEY: &str = "thread_source";
pub(crate) const SANDBOX_KEY: &str = "sandbox";
@@ -60,6 +61,7 @@ const RESERVED_METADATA_KEYS: &[&str] = &[
TURN_STARTED_AT_UNIX_MS_KEY,
FORKED_FROM_THREAD_ID_KEY,
PARENT_THREAD_ID_KEY,
PARENT_TURN_ID_KEY,
SUBAGENT_KIND_KEY,
THREAD_SOURCE_KEY,
SANDBOX_KEY,
@@ -163,6 +165,7 @@ pub struct CodexResponsesMetadata {
pub(crate) request_kind: Option<CodexResponsesRequestKind>,
pub(crate) forked_from_thread_id: Option<ThreadId>,
pub(crate) parent_thread_id: Option<ThreadId>,
pub(crate) parent_turn_id: Option<String>,
pub(crate) subagent_header: Option<String>,
pub(crate) subagent_kind: Option<String>,
pub(crate) thread_source: Option<ThreadSource>,
@@ -189,6 +192,7 @@ impl CodexResponsesMetadata {
request_kind: None,
forked_from_thread_id: None,
parent_thread_id: None,
parent_turn_id: None,
subagent_header: None,
subagent_kind: None,
thread_source: None,
@@ -237,6 +241,9 @@ impl CodexResponsesMetadata {
parent_thread_id.to_string(),
);
}
if let Some(parent_turn_id) = &self.parent_turn_id {
client_metadata.insert(PARENT_TURN_ID_KEY.to_string(), parent_turn_id.clone());
}
if self.has_turn_metadata()
&& let Some(turn_metadata_json) = self.turn_metadata_json()
{
@@ -296,6 +303,7 @@ impl CodexResponsesMetadata {
request_kind: request_kind_value,
forked_from_thread_id: self.forked_from_thread_id,
parent_thread_id: self.parent_thread_id,
parent_turn_id: self.parent_turn_id.as_deref(),
subagent_kind: self.subagent_kind.as_deref(),
thread_source: self.thread_source.as_ref(),
sandbox: self.sandbox.as_deref(),
@@ -383,6 +391,8 @@ struct CodexTurnMetadataPayload<'a> {
#[serde(default, skip_serializing_if = "Option::is_none")]
parent_thread_id: Option<ThreadId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
parent_turn_id: Option<&'a str>,
#[serde(default, skip_serializing_if = "Option::is_none")]
subagent_kind: Option<&'a str>,
#[serde(default, skip_serializing_if = "Option::is_none")]
thread_source: Option<&'a ThreadSource>,
codex-rs/core/src/session/handlers.rs
+23 -5
View File
@@ -83,8 +83,9 @@ pub async fn user_input_or_turn(
sub_id: String,
op: Op,
client_user_message_id: Option<String>,
parent_turn_id: Option<String>,
) {
user_input_or_turn_inner(sess, sub_id, op, client_user_message_id).await;
user_input_or_turn_inner(sess, sub_id, op, client_user_message_id, parent_turn_id).await;
}
pub async fn update_thread_settings(
@@ -178,6 +179,7 @@ pub(super) async fn user_input_or_turn_inner(
sub_id: String,
op: Op,
client_user_message_id: Option<String>,
parent_turn_id: Option<String>,
) {
let Op::UserInput {
items,
@@ -224,6 +226,9 @@ pub(super) async fn user_input_or_turn_inner(
current_context.session_telemetry.user_prompt(&items);
}
Err(SteerInputError::NoActiveTurn(items)) => {
if let Some(id) = parent_turn_id {
current_context.turn_metadata_state.set_parent_turn_id(id);
}
if let Some(responsesapi_client_metadata) = responsesapi_client_metadata {
current_context
.turn_metadata_state
@@ -268,10 +273,11 @@ pub async fn inter_agent_communication(
sess: &Arc<Session>,
sub_id: String,
communication: InterAgentCommunication,
parent_turn_id: Option<String>,
) {
let trigger_turn = communication.trigger_turn;
sess.input_queue
.enqueue_mailbox_communication(communication)
.enqueue_mailbox_communication(communication, parent_turn_id.filter(|_| trigger_turn))
.await;
crate::agent_communication::emit_agent_communication_receive(&sub_id);
if trigger_turn || sess.has_outstanding_durable_sleep() {
@@ -746,8 +752,14 @@ pub(super) async fn submission_loop(
false
}
Op::UserInput { .. } => {
user_input_or_turn(&sess, sub.id.clone(), sub.op, sub.client_user_message_id)
.await;
user_input_or_turn(
&sess,
sub.id.clone(),
sub.op,
sub.client_user_message_id,
sub.parent_turn_id,
)
.await;
false
}
Op::ThreadSettings { thread_settings } => {
@@ -755,7 +767,13 @@ pub(super) async fn submission_loop(
false
}
Op::InterAgentCommunication { communication } => {
inter_agent_communication(&sess, sub.id.clone(), communication).await;
inter_agent_communication(
&sess,
sub.id.clone(),
communication,
sub.parent_turn_id,
)
.await;
false
}
Op::ExecApproval {
codex-rs/core/src/session/inject.rs
+8 -2
View File
@@ -5,6 +5,7 @@ use crate::codex_thread::TryStartTurnIfIdleError;
use crate::codex_thread::TryStartTurnIfIdleRejectionReason;
use crate::state::ActiveTurn;
use crate::state::TurnState;
use crate::tasks::MailboxParentProvenance;
use crate::tasks::RegularTask;
use codex_protocol::config_types::ModeKind;
use codex_protocol::models::ResponseItem;
@@ -124,8 +125,13 @@ impl Session {
input.into_iter().map(TurnInput::ResponseItem).collect(),
)
.await;
self.start_task(turn_context, Vec::new(), RegularTask::new())
.await;
self.start_task(
turn_context,
Vec::new(),
RegularTask::new(),
MailboxParentProvenance::Ignore,
)
.await;
Ok(())
}
codex-rs/core/src/session/input_queue.rs
+90 -39
View File
@@ -34,7 +34,12 @@ pub(crate) struct TurnInputQueue {
/// Session-scoped pending input storage and active-turn mailbox delivery coordination.
pub(crate) struct InputQueue {
activity_tx: watch::Sender<InputQueueActivity>,
mailbox_pending_mails: Mutex<VecDeque<InterAgentCommunication>>,
mailbox_pending_mails: Mutex<VecDeque<PendingMailboxCommunication>>,
}
struct PendingMailboxCommunication {
communication: InterAgentCommunication,
parent_turn_id: Option<String>,
}
impl InputQueue {
@@ -72,11 +77,15 @@ impl InputQueue {
pub(crate) async fn enqueue_mailbox_communication(
&self,
communication: InterAgentCommunication,
parent_turn_id: Option<String>,
) {
self.mailbox_pending_mails
.lock()
.await
.push_back(communication);
.push_back(PendingMailboxCommunication {
communication,
parent_turn_id,
});
self.activity_tx.send_replace(InputQueueActivity::Mailbox);
}
@@ -89,16 +98,27 @@ impl InputQueue {
.lock()
.await
.iter()
.any(|mail| mail.trigger_turn)
.any(|mail| mail.communication.trigger_turn)
}
pub(crate) async fn drain_mailbox_input_items(&self) -> Vec<TurnInput> {
self.mailbox_pending_mails
pub(crate) async fn drain_mailbox_input_items(&self) -> (Vec<TurnInput>, Option<String>) {
let pending_mails = self
.mailbox_pending_mails
.lock()
.await
.drain(..)
.map(TurnInput::InterAgentCommunication)
.collect()
.collect::<Vec<_>>();
let parent_turn_id = pending_mails
.iter()
.filter(|mail| mail.communication.trigger_turn)
.map(|mail| mail.parent_turn_id.as_deref())
.reduce(|expected, candidate| expected.filter(|id| candidate == Some(*id)))
.and_then(|id| id.filter(|id| !id.trim().is_empty()).map(str::to_string));
let items = pending_mails
.into_iter()
.map(|mail| TurnInput::InterAgentCommunication(mail.communication))
.collect();
(items, parent_turn_id)
}
pub(crate) async fn turn_state_for_sub_id(
@@ -204,7 +224,7 @@ impl InputQueue {
pub(crate) async fn get_pending_input(
&self,
active_turn: &Mutex<Option<ActiveTurn>>,
) -> Vec<TurnInput> {
) -> (Vec<TurnInput>, Option<String>) {
let (pending_input, accepts_mailbox_delivery) = {
let mut active = active_turn.lock().await;
match active.as_mut() {
@@ -223,15 +243,15 @@ impl InputQueue {
}
};
if !accepts_mailbox_delivery {
return pending_input;
return (pending_input, None);
}
let mailbox_items = self.drain_mailbox_input_items().await.into_iter();
let (mailbox_items, parent_turn_id) = self.drain_mailbox_input_items().await;
if pending_input.is_empty() {
mailbox_items.collect()
(mailbox_items, parent_turn_id)
} else {
let mut pending_input = pending_input;
pending_input.extend(mailbox_items);
pending_input
(pending_input, parent_turn_id)
}
}
@@ -299,21 +319,23 @@ mod tests {
input_queue.subscribe_activity(/*turn_state*/ None).await;
assert_eq!(pending_activity, None);
let mail_one = make_mail(
AgentPath::root(),
AgentPath::try_from("/root/worker").expect("agent path"),
"one",
/*trigger_turn*/ false,
);
input_queue
.enqueue_mailbox_communication(make_mail(
AgentPath::root(),
AgentPath::try_from("/root/worker").expect("agent path"),
"one",
/*trigger_turn*/ false,
))
.enqueue_mailbox_communication(mail_one, /*parent_turn_id*/ None)
.await;
let mail_two = make_mail(
AgentPath::root(),
AgentPath::try_from("/root/worker").expect("agent path"),
"two",
/*trigger_turn*/ false,
);
input_queue
.enqueue_mailbox_communication(make_mail(
AgentPath::root(),
AgentPath::try_from("/root/worker").expect("agent path"),
"two",
/*trigger_turn*/ false,
))
.enqueue_mailbox_communication(mail_two, /*parent_turn_id*/ None)
.await;
activity_rx.changed().await.expect("mailbox update");
@@ -388,14 +410,14 @@ mod tests {
);
input_queue
.enqueue_mailbox_communication(mail_one.clone())
.enqueue_mailbox_communication(mail_one.clone(), /*parent_turn_id*/ None)
.await;
input_queue
.enqueue_mailbox_communication(mail_two.clone())
.enqueue_mailbox_communication(mail_two.clone(), /*parent_turn_id*/ None)
.await;
assert_eq!(
input_queue.drain_mailbox_input_items().await,
input_queue.drain_mailbox_input_items().await.0,
vec![
TurnInput::InterAgentCommunication(mail_one),
TurnInput::InterAgentCommunication(mail_two)
@@ -404,27 +426,56 @@ mod tests {
assert!(!input_queue.has_pending_mailbox_items().await);
}
#[tokio::test]
async fn input_queue_requires_one_unambiguous_trigger_parent() {
for (pending_mails, expected_parent_turn_id) in [
(Vec::new(), None),
(vec![(false, Some("q"))], None),
(vec![(true, Some(""))], None),
(vec![(true, Some(" "))], None),
(vec![(true, None)], None),
(vec![(true, Some("a")), (true, Some("b"))], None),
(vec![(true, Some("a")), (true, None)], None),
(vec![(true, Some("a")), (true, Some("a"))], Some("a")),
(vec![(false, Some("q")), (true, Some("a"))], Some("a")),
] {
let input_queue = InputQueue::new();
for (trigger_turn, parent_turn_id) in pending_mails {
input_queue
.enqueue_mailbox_communication(
make_mail(AgentPath::root(), AgentPath::root(), "task", trigger_turn),
parent_turn_id.map(str::to_string),
)
.await;
}
let (_, parent_turn_id) = input_queue.drain_mailbox_input_items().await;
assert_eq!(parent_turn_id.as_deref(), expected_parent_turn_id);
}
}
#[tokio::test]
async fn input_queue_tracks_pending_trigger_turn_mail() {
let input_queue = InputQueue::new();
let queued_mail = make_mail(
AgentPath::root(),
AgentPath::try_from("/root/worker").expect("agent path"),
"queued",
/*trigger_turn*/ false,
);
input_queue
.enqueue_mailbox_communication(make_mail(
AgentPath::root(),
AgentPath::try_from("/root/worker").expect("agent path"),
"queued",
/*trigger_turn*/ false,
))
.enqueue_mailbox_communication(queued_mail, /*parent_turn_id*/ None)
.await;
assert!(!input_queue.has_trigger_turn_mailbox_items().await);
let trigger_mail = make_mail(
AgentPath::root(),
AgentPath::try_from("/root/worker").expect("agent path"),
"wake",
/*trigger_turn*/ true,
);
input_queue
.enqueue_mailbox_communication(make_mail(
AgentPath::root(),
AgentPath::try_from("/root/worker").expect("agent path"),
"wake",
/*trigger_turn*/ true,
))
.enqueue_mailbox_communication(trigger_mail, /*parent_turn_id*/ None)
.await;
assert!(input_queue.has_trigger_turn_mailbox_items().await);
}
codex-rs/core/src/session/mod.rs
+12 -2
View File
@@ -791,13 +791,15 @@ impl Session {
impl SessionIo {
/// Submit the `op` wrapped in a `Submission` with a unique ID.
pub(crate) async fn submit(&self, op: Op) -> CodexResult<String> {
self.submit_with_trace(op, /*trace*/ None).await
self.submit_with_trace(op, /*trace*/ None, /*parent_turn_id*/ None)
.await
}
pub(crate) async fn submit_with_trace(
&self,
op: Op,
trace: Option<W3cTraceContext>,
parent_turn_id: Option<String>,
) -> CodexResult<String> {
let id = new_submission_id();
let sub = Submission {
@@ -805,6 +807,7 @@ impl SessionIo {
op,
client_user_message_id: None,
trace,
parent_turn_id,
};
self.submit_with_id(sub).await?;
Ok(id)
@@ -823,6 +826,7 @@ impl SessionIo {
op,
client_user_message_id,
trace,
parent_turn_id: None,
};
self.submit_with_id(sub).await?;
Ok(id)
@@ -1211,6 +1215,7 @@ impl Session {
thread_settings: Default::default(),
},
/*client_user_message_id*/ None,
/*parent_turn_id*/ None,
)
.await;
}
@@ -1961,7 +1966,12 @@ impl Session {
if let Err(err) = self
.services
.agent_control
.send_inter_agent_communication(parent_thread_id, communication, context)
.send_inter_agent_communication(
parent_thread_id,
communication,
context,
/*parent_turn_id*/ None,
)
.await
{
debug!("failed to notify parent thread {parent_thread_id}: {err}");
codex-rs/core/src/session/tests.rs
+47 -33
View File
@@ -6805,6 +6805,7 @@ async fn submit_with_id_captures_current_span_trace_context() {
id: "sub-1".into(),
op: Op::Interrupt,
client_user_message_id: None,
parent_turn_id: None,
trace: None,
})
.await
@@ -6877,6 +6878,7 @@ fn submission_dispatch_span_prefers_submission_trace_context() {
id: "sub-1".into(),
op: Op::Interrupt,
client_user_message_id: None,
parent_turn_id: None,
trace: Some(submission_trace),
})
});
@@ -6904,6 +6906,7 @@ fn submission_dispatch_span_uses_debug_for_realtime_audio() {
},
}),
client_user_message_id: None,
parent_turn_id: None,
trace: None,
});
@@ -6969,6 +6972,7 @@ async fn user_turn_updates_approvals_reviewer() {
},
},
/*client_user_message_id*/ None,
/*parent_turn_id*/ None,
)
.await;
@@ -7253,6 +7257,7 @@ async fn spawn_task_turn_span_inherits_dispatch_trace_context() {
id: "sub-1".into(),
op: Op::Interrupt,
client_user_message_id: None,
parent_turn_id: None,
trace: Some(submission_trace.clone()),
});
let dispatch_span_id = dispatch_span.context().span().span_context().span_id();
@@ -10422,13 +10427,16 @@ async fn thread_idle_lifecycle_waits_for_trigger_turn_mailbox_work() {
session.services.extensions = Arc::new(builder.build());
session
.input_queue
.enqueue_mailbox_communication(InterAgentCommunication::new(
AgentPath::root(),
AgentPath::root(),
Vec::new(),
"pending trigger".to_string(),
/*trigger_turn*/ true,
))
.enqueue_mailbox_communication(
InterAgentCommunication::new(
AgentPath::root(),
AgentPath::root(),
Vec::new(),
"pending trigger".to_string(),
/*trigger_turn*/ true,
),
/*parent_turn_id*/ None,
)
.await;
session.emit_thread_idle_lifecycle_if_idle().await;
@@ -10458,7 +10466,7 @@ async fn try_start_turn_if_idle_rejects_active_turn_without_injecting() {
assert_eq!(TryStartTurnIfIdleRejectionReason::Busy, err.reason());
assert_eq!(vec![item], err.into_input());
assert_eq!(
Vec::<TurnInput>::new(),
(Vec::<TurnInput>::new(), None),
sess.input_queue.get_pending_input(&sess.active_turn).await
);
@@ -10485,7 +10493,7 @@ async fn try_start_turn_if_idle_rejects_plan_mode_without_injecting() {
assert_eq!(vec![item], err.into_input());
assert!(sess.active_turn.lock().await.is_none());
assert_eq!(
Vec::<TurnInput>::new(),
(Vec::<TurnInput>::new(), None),
sess.input_queue.get_pending_input(&sess.active_turn).await
);
}
@@ -10494,13 +10502,16 @@ async fn try_start_turn_if_idle_rejects_plan_mode_without_injecting() {
async fn try_start_turn_if_idle_rejects_pending_trigger_turn_without_injecting() {
let (sess, _tc, _rx) = make_session_and_context_with_rx().await;
sess.input_queue
.enqueue_mailbox_communication(InterAgentCommunication::new(
AgentPath::root(),
AgentPath::root(),
Vec::new(),
"pending trigger".to_string(),
/*trigger_turn*/ true,
))
.enqueue_mailbox_communication(
InterAgentCommunication::new(
AgentPath::root(),
AgentPath::root(),
Vec::new(),
"pending trigger".to_string(),
/*trigger_turn*/ true,
),
/*parent_turn_id*/ None,
)
.await;
let item = user_message("synthetic idle input");
@@ -10540,7 +10551,7 @@ async fn try_start_turn_if_idle_rejects_active_review_turn_without_injecting() {
assert_eq!(TryStartTurnIfIdleRejectionReason::Busy, err.reason());
assert_eq!(vec![item], err.into_input());
assert_eq!(
Vec::<TurnInput>::new(),
(Vec::<TurnInput>::new(), None),
sess.input_queue.get_pending_input(&sess.active_turn).await
);
@@ -10768,7 +10779,7 @@ async fn queue_only_mailbox_mail_waits_for_next_turn_after_answer_boundary() {
.defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
.await;
sess.input_queue
.enqueue_mailbox_communication(communication.clone())
.enqueue_mailbox_communication(communication.clone(), /*parent_turn_id*/ None)
.await;
assert!(
@@ -10777,13 +10788,13 @@ 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::new()
(Vec::new(), None)
);
sess.abort_all_tasks(TurnAbortReason::Replaced).await;
assert_eq!(
sess.input_queue.get_pending_input(&sess.active_turn).await,
(sess.input_queue.get_pending_input(&sess.active_turn).await).0,
vec![TurnInput::InterAgentCommunication(communication)],
);
}
@@ -10805,13 +10816,16 @@ async fn trigger_turn_mailbox_mail_waits_for_next_turn_after_answer_boundary() {
.defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
.await;
sess.input_queue
.enqueue_mailbox_communication(InterAgentCommunication::new(
AgentPath::try_from("/root/worker").expect("worker path should parse"),
AgentPath::root(),
Vec::new(),
"late trigger update".to_string(),
/*trigger_turn*/ true,
))
.enqueue_mailbox_communication(
InterAgentCommunication::new(
AgentPath::try_from("/root/worker").expect("worker path should parse"),
AgentPath::root(),
Vec::new(),
"late trigger update".to_string(),
/*trigger_turn*/ true,
),
/*parent_turn_id*/ None,
)
.await;
assert!(
@@ -10848,7 +10862,7 @@ async fn steered_input_reopens_mailbox_delivery_for_current_turn() {
.defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
.await;
sess.input_queue
.enqueue_mailbox_communication(communication.clone())
.enqueue_mailbox_communication(communication.clone(), /*parent_turn_id*/ None)
.await;
sess.steer_input(
vec![UserInput::Text {
@@ -10864,7 +10878,7 @@ async fn steered_input_reopens_mailbox_delivery_for_current_turn() {
.expect("steered input should be accepted");
assert_eq!(
sess.input_queue.get_pending_input(&sess.active_turn).await,
(sess.input_queue.get_pending_input(&sess.active_turn).await).0,
vec![
TurnInput::UserInput {
content: vec![UserInput::Text {
@@ -10902,7 +10916,7 @@ async fn stale_defer_mailbox_delivery_does_not_override_steered_input() {
.defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
.await;
sess.input_queue
.enqueue_mailbox_communication(communication.clone())
.enqueue_mailbox_communication(communication.clone(), /*parent_turn_id*/ None)
.await;
sess.steer_input(
vec![UserInput::Text {
@@ -10922,7 +10936,7 @@ async fn stale_defer_mailbox_delivery_does_not_override_steered_input() {
.await;
assert_eq!(
sess.input_queue.get_pending_input(&sess.active_turn).await,
(sess.input_queue.get_pending_input(&sess.active_turn).await).0,
vec![
TurnInput::UserInput {
content: vec![UserInput::Text {
@@ -10960,7 +10974,7 @@ async fn tool_calls_reopen_mailbox_delivery_for_current_turn() {
.defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
.await;
sess.input_queue
.enqueue_mailbox_communication(communication.clone())
.enqueue_mailbox_communication(communication.clone(), /*parent_turn_id*/ None)
.await;
let item = ResponseItem::FunctionCall {
@@ -10986,7 +11000,7 @@ async fn tool_calls_reopen_mailbox_delivery_for_current_turn() {
assert!(output.needs_follow_up);
assert!(output.tool_future.is_some());
assert_eq!(
sess.input_queue.get_pending_input(&sess.active_turn).await,
(sess.input_queue.get_pending_input(&sess.active_turn).await).0,
vec![TurnInput::InterAgentCommunication(communication)],
);
}
codex-rs/core/src/session/turn.rs
+4 -1
View File
@@ -274,7 +274,10 @@ pub(crate) async fn run_turn(
// submitted through the UI while the model was running. Though the UI
// may support this, the model might not.
let pending_input = if can_drain_pending_input {
sess.input_queue.get_pending_input(&sess.active_turn).await
sess.input_queue
.get_pending_input(&sess.active_turn)
.await
.0
} else {
Vec::new()
};
codex-rs/core/src/tasks/mod.rs
+22 -4
View File
@@ -69,6 +69,11 @@ const TASK_COMPACT_METRIC: &str = "codex.task.compact";
pub(crate) type SessionTaskResult = CodexResult<Option<String>>;
pub(crate) enum MailboxParentProvenance {
Ignore,
Attribute,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum InterruptedTurnHistoryMarker {
Disabled,
@@ -319,7 +324,8 @@ impl Session {
) {
self.abort_all_tasks(TurnAbortReason::Replaced).await;
self.clear_connector_selection().await;
self.start_task(turn_context, input, task).await;
self.start_task(turn_context, input, task, MailboxParentProvenance::Ignore)
.await;
}
pub(crate) async fn start_task<T: SessionTask>(
@@ -327,6 +333,7 @@ impl Session {
turn_context: Arc<TurnContext>,
input: Vec<TurnInput>,
task: T,
mailbox_parent_provenance: MailboxParentProvenance,
) {
let task: Arc<dyn AnySessionTask> = Arc::new(task);
let task_kind = task.kind();
@@ -350,7 +357,13 @@ impl Session {
.await
.clear_turn(&turn_context.sub_id);
let pending_items = self.input_queue.get_pending_input(&self.active_turn).await;
let (pending_items, parent_turn_id) =
self.input_queue.get_pending_input(&self.active_turn).await;
if let (MailboxParentProvenance::Attribute, Some(id)) =
(mailbox_parent_provenance, parent_turn_id)
{
turn_context.turn_metadata_state.set_parent_turn_id(id);
}
let turn_state = {
let mut active = self.active_turn.lock().await;
let turn = active.get_or_insert_with(ActiveTurn::default);
@@ -502,8 +515,13 @@ impl Session {
let turn_context = self.new_default_turn_with_sub_id(sub_id).await;
self.maybe_emit_model_warnings_for_turn(turn_context.as_ref())
.await;
self.start_task(turn_context, Vec::new(), RegularTask::new())
.await;
self.start_task(
turn_context,
Vec::new(),
RegularTask::new(),
MailboxParentProvenance::Attribute,
)
.await;
}
pub async fn abort_all_tasks(self: &Arc<Self>, reason: TurnAbortReason) {
codex-rs/core/src/test_support.rs
+5
View File
@@ -195,6 +195,11 @@ pub fn responses_metadata(
}
}
pub fn with_parent_turn(mut metadata: CodexResponsesMetadata, id: &str) -> CodexResponsesMetadata {
metadata.parent_turn_id = Some(id.to_string());
metadata
}
pub fn all_model_presets() -> &'static Vec<ModelPreset> {
&TEST_MODEL_PRESETS
}
codex-rs/core/src/thread_manager.rs
+10 -2
View File
@@ -1330,14 +1330,22 @@ impl ThreadManagerState {
}
/// Send an operation to a thread by ID.
pub(crate) async fn send_op(&self, thread_id: ThreadId, op: Op) -> CodexResult<String> {
pub(crate) async fn send_op(
&self,
thread_id: ThreadId,
op: Op,
parent_turn_id: Option<String>,
) -> CodexResult<String> {
let thread = self.get_thread(thread_id).await?;
if let Some(ops_log) = &self.ops_log
&& let Ok(mut log) = ops_log.lock()
{
log.push((thread_id, op.clone()));
}
thread.submit(op).await
thread
.io
.submit_with_trace(op, /*trace*/ None, parent_turn_id)
.await
}
/// Remove a thread from the manager by ID, returning it when present.
codex-rs/core/src/tools/handlers/multi_agents/send_input.rs
+1 -1
View File
@@ -84,7 +84,7 @@ impl Handler {
.await;
let agent_control = session.services.agent_control.clone();
let result = agent_control
.send_input(receiver_thread_id, input_items)
.send_input(receiver_thread_id, input_items, Some(turn.sub_id.clone()))
.await
.map_err(|err| collab_agent_error(receiver_thread_id, err));
let status = session
codex-rs/core/src/tools/handlers/multi_agents/spawn.rs
+1
View File
@@ -127,6 +127,7 @@ async fn handle_spawn_agent(
fork_parent_spawn_call_id: args.fork_context.then(|| call_id.clone()),
fork_mode: args.fork_context.then_some(SpawnAgentForkMode::FullHistory),
parent_thread_id: Some(session.thread_id),
parent_turn_id: Some(turn.sub_id.clone()),
environments: Some(turn.environments.to_selections()),
},
))
codex-rs/core/src/tools/handlers/multi_agents_tests.rs
+50 -35
View File
@@ -2977,13 +2977,16 @@ async fn multi_agent_v2_wait_agent_accepts_timeout_only_argument() {
session
.input_queue
.enqueue_mailbox_communication(InterAgentCommunication::new(
worker_path,
AgentPath::root(),
Vec::new(),
"hello from worker".to_string(),
/*trigger_turn*/ false,
))
.enqueue_mailbox_communication(
InterAgentCommunication::new(
worker_path,
AgentPath::root(),
Vec::new(),
"hello from worker".to_string(),
/*trigger_turn*/ false,
),
/*parent_turn_id*/ None,
)
.await;
let output = wait_task
@@ -3461,13 +3464,16 @@ async fn multi_agent_v2_wait_agent_returns_summary_for_mailbox_activity() {
session
.input_queue
.enqueue_mailbox_communication(InterAgentCommunication::new(
worker_path,
AgentPath::root(),
Vec::new(),
"completed".to_string(),
/*trigger_turn*/ false,
))
.enqueue_mailbox_communication(
InterAgentCommunication::new(
worker_path,
AgentPath::root(),
Vec::new(),
"completed".to_string(),
/*trigger_turn*/ false,
),
/*parent_turn_id*/ None,
)
.await;
let wait_output = wait_task
@@ -3534,13 +3540,16 @@ async fn multi_agent_v2_wait_agent_returns_for_already_queued_mail() {
session
.input_queue
.enqueue_mailbox_communication(InterAgentCommunication::new(
worker_path,
AgentPath::root(),
Vec::new(),
"already queued".to_string(),
/*trigger_turn*/ false,
))
.enqueue_mailbox_communication(
InterAgentCommunication::new(
worker_path,
AgentPath::root(),
Vec::new(),
"already queued".to_string(),
/*trigger_turn*/ false,
),
/*parent_turn_id*/ None,
)
.await;
let output = timeout(
@@ -3633,13 +3642,16 @@ async fn multi_agent_v2_wait_agent_wakes_on_any_mailbox_notification() {
session
.input_queue
.enqueue_mailbox_communication(InterAgentCommunication::new(
worker_b_path,
AgentPath::root(),
Vec::new(),
"from worker b".to_string(),
/*trigger_turn*/ false,
))
.enqueue_mailbox_communication(
InterAgentCommunication::new(
worker_b_path,
AgentPath::root(),
Vec::new(),
"from worker b".to_string(),
/*trigger_turn*/ false,
),
/*parent_turn_id*/ None,
)
.await;
let output = wait_task
@@ -3721,13 +3733,16 @@ async fn multi_agent_v2_wait_agent_does_not_return_completed_content() {
session
.input_queue
.enqueue_mailbox_communication(InterAgentCommunication::new(
worker_path,
AgentPath::root(),
Vec::new(),
"sensitive child output".to_string(),
/*trigger_turn*/ false,
))
.enqueue_mailbox_communication(
InterAgentCommunication::new(
worker_path,
AgentPath::root(),
Vec::new(),
"sensitive child output".to_string(),
/*trigger_turn*/ false,
),
/*parent_turn_id*/ None,
)
.await;
let output = wait_task
codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs
+8 -1
View File
@@ -107,10 +107,17 @@ pub(crate) async fn handle_message_string_tool(
MessageDeliveryMode::TriggerTurn => AgentCommunicationKind::Followup,
};
let context = AgentCommunicationContext::new(kind, session.thread_id);
let parent_turn_id =
matches!(mode, MessageDeliveryMode::TriggerTurn).then(|| turn.sub_id.clone());
let result = session
.services
.agent_control
.send_inter_agent_communication(receiver_thread_id, mode.apply(communication), context)
.send_inter_agent_communication(
receiver_thread_id,
mode.apply(communication),
context,
parent_turn_id,
)
.await
.map_err(|err| collab_agent_error(receiver_thread_id, err));
result?;
codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs
+1
View File
@@ -118,6 +118,7 @@ async fn handle_spawn_agent(
fork_parent_spawn_call_id: fork_mode.as_ref().map(|_| call_id.clone()),
fork_mode,
parent_thread_id: Some(session.thread_id),
parent_turn_id: Some(turn.sub_id.clone()),
environments: Some(turn.environments.to_selections()),
},
),
codex-rs/core/src/turn_metadata.rs
+13
View File
@@ -2,6 +2,7 @@ use std::collections::BTreeMap;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::OnceLock;
use std::sync::RwLock;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
@@ -12,6 +13,7 @@ use tokio::task::JoinHandle;
use crate::responses_metadata::CODE_MODE_TOOL_NAMES_KEY;
use crate::responses_metadata::CodexResponsesMetadata;
use crate::responses_metadata::CodexResponsesRequestKind;
use crate::responses_metadata::PARENT_TURN_ID_KEY;
use crate::responses_metadata::TurnMetadataWorkspace;
use crate::responses_metadata::filter_extra_metadata;
use crate::responses_metadata::subagent_header_value;
@@ -92,6 +94,7 @@ pub(crate) struct TurnMetadataState {
thread_id: String,
forked_from_thread_id: Option<ThreadId>,
parent_thread_id: Option<ThreadId>,
parent_turn_id: OnceLock<String>,
subagent_header: Option<String>,
subagent_kind: Option<String>,
thread_source: Option<ThreadSource>,
@@ -136,6 +139,7 @@ impl TurnMetadataState {
thread_id,
forked_from_thread_id,
parent_thread_id,
parent_turn_id: OnceLock::new(),
subagent_header: subagent_header_value(session_source),
subagent_kind: subagent_metadata_kind(session_source),
thread_source,
@@ -160,6 +164,7 @@ impl TurnMetadataState {
return None;
};
metadata.remove(CODE_MODE_TOOL_NAMES_KEY); // Precaution: avoid exposing tool data to external MCPs.
metadata.remove(PARENT_TURN_ID_KEY);
metadata.insert(
MODEL_KEY.to_string(),
Value::String(context.model.to_string()),
@@ -219,6 +224,13 @@ impl TurnMetadataState {
(!code_mode_tool_names.is_empty()).then_some(code_mode_tool_names);
}
pub(crate) fn set_parent_turn_id(&self, parent_turn_id: String) {
if parent_turn_id.trim().is_empty() {
return;
}
let _ = self.parent_turn_id.set(parent_turn_id);
}
pub(crate) fn set_responsesapi_client_metadata(
&self,
responsesapi_client_metadata: HashMap<String, String>,
@@ -243,6 +255,7 @@ impl TurnMetadataState {
turn_id: Some(self.turn_id.clone()),
forked_from_thread_id: self.forked_from_thread_id,
parent_thread_id: self.parent_thread_id,
parent_turn_id: self.parent_turn_id.get().cloned(),
subagent_header: self.subagent_header.clone(),
subagent_kind: self.subagent_kind.clone(),
thread_source: self.thread_source.clone(),
codex-rs/core/src/turn_metadata_tests.rs
+7
View File
@@ -4,6 +4,7 @@ use crate::responses_metadata::CODE_MODE_TOOL_NAMES_KEY;
use crate::responses_metadata::CodexResponsesRequestKind;
use crate::responses_metadata::CompactionTurnMetadata;
use crate::responses_metadata::INSTALLATION_ID_KEY;
use crate::responses_metadata::PARENT_TURN_ID_KEY;
use crate::responses_metadata::WINDOW_ID_KEY;
use crate::sandbox_tags::permission_profile_sandbox_tag;
use codex_analytics::CompactionImplementation;
@@ -560,6 +561,7 @@ fn turn_metadata_state_ignores_client_reserved_metadata_before_start() {
"parent_thread_id".to_string(),
"client-supplied".to_string(),
),
("parent_turn_id".to_string(), "client-supplied".to_string()),
("subagent_kind".to_string(), "client-supplied".to_string()),
]));
@@ -570,6 +572,7 @@ fn turn_metadata_state_ignores_client_reserved_metadata_before_start() {
assert!(json.get("turn_started_at_unix_ms").is_none());
assert!(json.get("forked_from_thread_id").is_none());
assert!(json.get("parent_thread_id").is_none());
assert!(json.get("parent_turn_id").is_none());
assert!(json.get("subagent_kind").is_none());
}
@@ -602,6 +605,7 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields(
WindowsSandboxLevel::Disabled,
/*enforce_managed_network*/ false,
);
state.set_parent_turn_id("parent-turn-a".to_string());
state.set_responsesapi_client_metadata(HashMap::from([
("fiber_run_id".to_string(), "fiber-123".to_string()),
("origin".to_string(), "東京".to_string()),
@@ -634,6 +638,7 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields(
"parent_thread_id".to_string(),
"client-supplied".to_string(),
),
("parent_turn_id".to_string(), "client-supplied".to_string()),
("subagent_kind".to_string(), "client-supplied".to_string()),
(
CODE_MODE_TOOL_NAMES_KEY.to_string(),
@@ -694,6 +699,7 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields(
json["parent_thread_id"].as_str(),
Some("55555555-5555-4555-8555-555555555555")
);
assert_eq!(json["parent_turn_id"].as_str(), Some("parent-turn-a"));
assert_eq!(json["subagent_kind"].as_str(), Some("thread_spawn"));
assert_eq!(json["thread_source"].as_str(), Some("automation"));
assert_eq!(json["turn_id"].as_str(), Some("turn-a"));
@@ -727,6 +733,7 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields(
assert_eq!(meta["model"].as_str(), Some("gpt-5.4"));
assert_eq!(meta["reasoning_effort"].as_str(), Some("high"));
assert!(meta.get(CODE_MODE_TOOL_NAMES_KEY).is_none());
assert!(meta.get(PARENT_TURN_ID_KEY).is_none());
assert!(meta.get(WINDOW_ID_KEY).is_none());
assert_eq!(state.workspace_kind().as_deref(), Some("projectless"));
}
codex-rs/core/tests/common/responses.rs
+13
View File
@@ -80,6 +80,19 @@ impl ResponseMock {
}
}
pub fn assert_parent_turn(body: &Value, expected: Option<&str>) -> Result<()> {
let metadata = &body["client_metadata"];
let payload = metadata["x-codex-turn-metadata"]
.as_str()
.expect("canonical turn metadata");
let canonical: Value = serde_json::from_str(payload)?;
let expected = expected.map(Value::from);
let key = "parent_turn_id";
let actual = (metadata.get(key), canonical.get(key));
assert_eq!(actual, (expected.as_ref(), expected.as_ref()));
Ok(())
}
#[derive(Debug, Clone)]
pub struct ResponsesRequest(wiremock::Request);
codex-rs/core/tests/suite/client_websockets.rs
+9 -2
View File
@@ -7,6 +7,7 @@ use codex_core::ModelClientSession;
use codex_core::Prompt;
use codex_core::ResponseEvent;
use codex_core::X_RESPONSESAPI_INCLUDE_TIMING_METRICS_HEADER;
use codex_core::test_support::with_parent_turn;
use codex_features::Feature;
use codex_http_client::OutboundProxyPolicy;
use codex_login::CodexAuth;
@@ -1754,8 +1755,10 @@ async fn responses_websocket_forwards_turn_metadata_on_initial_and_incremental_c
prior_assistant_output,
message_item("second"),
]);
let first_responses_metadata = turn_metadata(&harness, Some("turn-123"));
let second_responses_metadata = turn_metadata(&harness, Some("turn-456"));
let first_responses_metadata =
with_parent_turn(turn_metadata(&harness, Some("turn-123")), "parent-turn-123");
let second_responses_metadata =
with_parent_turn(turn_metadata(&harness, Some("turn-456")), "parent-turn-456");
stream_until_complete_with_metadata(
&mut client_session,
@@ -1797,6 +1800,10 @@ async fn responses_websocket_forwards_turn_metadata_on_initial_and_incremental_c
assert_eq!(first_metadata["turn_id"].as_str(), Some("turn-123"));
assert_eq!(second_metadata["turn_id"].as_str(), Some("turn-456"));
core_test_support::responses::assert_parent_turn(&first, Some("parent-turn-123"))
.expect("initial WebSocket parent-turn metadata");
core_test_support::responses::assert_parent_turn(&second, Some("parent-turn-456"))
.expect("incremental WebSocket parent-turn metadata");
assert_eq!(
first["client_metadata"]["turn_id"].as_str(),
first_metadata["turn_id"].as_str()
codex-rs/core/tests/suite/multi_agent_resume.rs
+149 -9
View File
@@ -5,6 +5,10 @@ use codex_protocol::models::PermissionProfile;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::AgentStatus;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::Submission;
use codex_protocol::user_input::UserInput;
use core_test_support::responses::assert_parent_turn;
use core_test_support::responses::ev_assistant_message;
use core_test_support::responses::ev_completed;
use core_test_support::responses::ev_function_call_with_namespace;
@@ -23,12 +27,17 @@ use tokio::time::sleep;
const COLLABORATION_NAMESPACE: &str = "collaboration";
const SPAWN_CALL_ID: &str = "spawn-worker";
const NESTED_CALL_ID: &str = "spawn-grandchild";
const QUEUE_CALL_ID: &str = "queue-worker-message";
const FOLLOWUP_CALL_ID: &str = "followup-worker";
const SIBLING_SPAWN_CALL_ID: &str = "spawn-survivor";
const SIBLING_FOLLOWUP_CALL_ID: &str = "followup-survivor";
const INTERRUPT_CALL_ID: &str = "interrupt-worker";
const INITIAL_PROMPT: &str = "spawn a durable worker";
const INITIAL_TASK: &str = "inspect the repository";
const NESTED_TASK: &str = "inspect the nested repository";
const QUEUE_PROMPT: &str = "queue context for the durable worker";
const QUEUED_MESSAGE: &str = "queue-only context from an earlier parent turn";
const FOLLOWUP_PROMPT: &str = "continue the durable worker";
const FOLLOWUP_TASK: &str = "inspect the tests too";
const SIBLING_PROMPT: &str = "spawn a second durable worker";
@@ -134,6 +143,7 @@ fn configure_multi_agent_v2_with_role(
.expect("test config should allow feature update");
config.multi_agent_v2.subagent_developer_instructions =
Some(SUBAGENT_DEVELOPER_INSTRUCTIONS.to_string());
config.multi_agent_v2.max_concurrent_threads_per_session = 3;
let role_path = config.codex_home.join("durable-worker-role.toml");
std::fs::write(
&role_path,
@@ -185,11 +195,37 @@ async fn cold_root_resume_restores_agent_identity_and_role_on_followup() -> Resu
},
sse(vec![
ev_response_created("resp-worker-1"),
ev_assistant_message("msg-worker-1", "initial task complete"),
ev_function_call_with_namespace(
NESTED_CALL_ID,
COLLABORATION_NAMESPACE,
"spawn_agent",
r#"{"message":"inspect the nested repository","task_name":"grandchild","fork_turns":"none"}"#,
),
ev_completed("resp-worker-1"),
]),
)
.await;
let nested_mock = mount_sse_once_match(
&server,
|request: &wiremock::Request| {
body_contains(request, NESTED_TASK)
&& request_has_input_type(request, "agent_message")
&& !body_contains(request, NESTED_CALL_ID)
},
sse(vec![ev_completed("resp-parent-turn-assistant")]),
)
.await;
for (text, is_subagent) in [(NESTED_CALL_ID, true), (QUEUE_CALL_ID, false)] {
mount_sse_once_match(
&server,
move |request: &wiremock::Request| {
body_contains(request, text)
&& request_has_input_type(request, "agent_message") == is_subagent
},
sse(vec![ev_completed("resp-parent-turn-assistant")]),
)
.await;
}
mount_sse_once_match(
&server,
|request: &wiremock::Request| {
@@ -215,16 +251,46 @@ async fn cold_root_resume_restores_agent_identity_and_role_on_followup() -> Resu
.rollout_path()
.expect("root rollout path")
.to_path_buf();
initial.submit_turn(INITIAL_PROMPT).await?;
let mut op = vec![UserInput::Text {
text: INITIAL_PROMPT.to_string(),
text_elements: Vec::new(),
}]
.into();
if let Op::UserInput {
thread_settings, ..
} = &mut op
{
thread_settings.permission_profile = Some(PermissionProfile::Disabled);
}
initial
.codex
.submit_with_id(Submission {
id: "spoofed-root-turn".to_string(),
op,
client_user_message_id: None,
trace: None,
parent_turn_id: Some("spoofed-parent-turn".to_string()),
})
.await?;
wait_for_event(&initial.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let deadline = Instant::now() + Duration::from_secs(2);
let worker_thread_id = loop {
if let Some(thread_id) = initial
.thread_manager
.list_thread_ids()
.await
if let Some(thread_id) = initial_child_request
.requests()
.into_iter()
.find(|thread_id| *thread_id != root_thread_id)
.find_map(|request| {
let body = request.body_json();
if body["client_metadata"]["x-codex-parent-thread-id"] != json!(root_thread_id) {
return None;
}
body["client_metadata"]["thread_id"]
.as_str()
.and_then(|thread_id| codex_protocol::ThreadId::from_string(thread_id).ok())
})
{
break thread_id;
}
@@ -299,12 +365,14 @@ async fn cold_root_resume_restores_agent_identity_and_role_on_followup() -> Resu
.await;
initial.submit_turn(SIBLING_PROMPT).await?;
let grandchild = nested_mock.last_request().expect("grandchild").body_json();
let nested_id = &grandchild["client_metadata"]["thread_id"];
let sibling_thread_id = initial
.thread_manager
.list_thread_ids()
.await
.into_iter()
.find(|thread_id| *thread_id != root_thread_id && *thread_id != worker_thread_id)
.find(|id| ![root_thread_id, worker_thread_id].contains(id) && &json!(id) != nested_id)
.ok_or_else(|| anyhow::anyhow!("spawned sibling should be registered"))?;
let sibling_thread = initial.thread_manager.get_thread(sibling_thread_id).await?;
wait_for_event(sibling_thread.as_ref(), |event| {
@@ -343,6 +411,7 @@ async fn cold_root_resume_restores_agent_identity_and_role_on_followup() -> Resu
request_has_model(request, ROLE_MODEL)
&& request_has_input_type(request, "agent_message")
&& body_contains(request, FOLLOWUP_TASK)
&& body_contains(request, QUEUED_MESSAGE)
},
sse(vec![
ev_response_created("resp-worker-2"),
@@ -388,13 +457,29 @@ async fn cold_root_resume_restores_agent_identity_and_role_on_followup() -> Resu
.is_err()
);
mount_sse_once_match(
&server,
|request: &wiremock::Request| body_contains(request, QUEUE_PROMPT),
sse(vec![
ev_response_created("resp-queue"),
ev_function_call_with_namespace(
QUEUE_CALL_ID,
COLLABORATION_NAMESPACE,
"send_message",
r#"{"target":"worker","message":"queue-only context from an earlier parent turn"}"#,
),
ev_completed("resp-queue"),
]),
)
.await;
resumed.submit_turn(QUEUE_PROMPT).await?;
resumed.submit_turn(FOLLOWUP_PROMPT).await?;
let reloaded_worker = resumed
.thread_manager
.get_thread(worker_thread_id)
.await
.expect("follow-up should lazily reload the original worker");
.expect("queued message should lazily reload the original worker");
let deadline = Instant::now() + Duration::from_secs(2);
loop {
if matches!(
@@ -413,6 +498,61 @@ async fn cold_root_resume_restores_agent_identity_and_role_on_followup() -> Resu
&& request.body_contains_text(ROLE_DEVELOPER_INSTRUCTIONS)
&& !request.body_contains_text(SUBAGENT_DEVELOPER_INSTRUCTIONS)
}));
let requests = server
.received_requests()
.await
.expect("captured response requests");
assert!(!followup_child_request.requests().iter().any(|request| {
request.body_json()["client_metadata"]["thread_id"] == json!(worker_thread_id)
&& request.body_contains_text(QUEUED_MESSAGE)
&& !request.body_contains_text(FOLLOWUP_TASK)
}));
let body_for = |text: &str, thread: codex_protocol::ThreadId| {
requests
.iter()
.find_map(|request| {
let body: Value = serde_json::from_slice(&decoded_body(request)?).ok()?;
(body_contains(request, text)
&& body["client_metadata"]["thread_id"] == json!(thread))
.then_some(body)
})
.expect("matching model request for expected thread")
};
let initial_root = body_for(INITIAL_PROMPT, root_thread_id);
let queue_root = body_for(QUEUE_PROMPT, root_thread_id);
let followup_root = body_for(FOLLOWUP_PROMPT, root_thread_id);
let initial_child = body_for(INITIAL_TASK, worker_thread_id);
let followup_child = body_for(FOLLOWUP_TASK, worker_thread_id);
let initial_parent = initial_root["client_metadata"]["turn_id"]
.as_str()
.expect("initial parent turn");
let queue_parent = queue_root["client_metadata"]["turn_id"]
.as_str()
.expect("queue-only parent turn");
let followup_parent = followup_root["client_metadata"]["turn_id"]
.as_str()
.expect("follow-up parent turn");
assert_ne!(followup_parent, initial_parent);
assert_ne!(followup_parent, queue_parent);
let nested_parent = initial_child["client_metadata"]["turn_id"]
.as_str()
.expect("nested worker parent turn");
for (body, parent_thread, parent_turn) in [
(&initial_root, None, None),
(&queue_root, None, None),
(&followup_root, None, None),
(&initial_child, Some(root_thread_id), Some(initial_parent)),
(&followup_child, Some(root_thread_id), Some(followup_parent)),
(&grandchild, Some(worker_thread_id), Some(nested_parent)),
] {
if let Some(parent_thread) = parent_thread {
assert_eq!(
body["client_metadata"]["x-codex-parent-thread-id"],
json!(parent_thread)
);
}
assert_parent_turn(body, parent_turn)?;
}
let reloaded_worker_config = reloaded_worker.config_snapshot().await;
let reloaded_worker_role_config = (
reloaded_worker_config.model,
codex-rs/core/tests/suite/review.rs
+2
View File
@@ -212,6 +212,8 @@ async fn review_op_emits_lifecycle_and_review_output() {
turn_metadata["parent_thread_id"].as_str(),
Some(parent_thread_id.as_str())
);
responses::assert_parent_turn(&request.body_json(), Some(review_turn_id.as_str()))
.expect("review request parent turn metadata");
// Also verify that a user message with the header and a formatted finding
// was recorded back in the parent session's rollout.
codex-rs/core/tests/suite/subagent_notifications.rs
+61 -2
View File
@@ -8,6 +8,7 @@ use codex_protocol::ThreadId;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::models::PermissionProfile;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::AgentStatus;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
@@ -17,6 +18,7 @@ use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::user_input::UserInput;
use core_test_support::hooks::trust_discovered_hooks;
use core_test_support::responses::ResponsesRequest;
use core_test_support::responses::assert_parent_turn;
use core_test_support::responses::ev_assistant_message;
use core_test_support::responses::ev_completed;
use core_test_support::responses::ev_function_call_with_namespace;
@@ -951,17 +953,22 @@ async fn spawned_child_receives_forked_parent_context(
config.agent_default_subagent_model = Some(REQUESTED_MODEL.to_string());
config.agent_default_subagent_reasoning_effort = Some(REQUESTED_REASONING_EFFORT);
});
let test = builder.build(&server).await?;
let test = builder.build_with_auto_env(&server).await?;
test.submit_turn(TURN_0_FORK_PROMPT).await?;
let _ = seed_turn.single_request();
test.submit_turn(TURN_1_PROMPT).await?;
let _ = spawn_turn.single_request();
let parent_body = spawn_turn.single_request().body_json();
let child_request = wait_for_request_with_model(&child_request_log, REQUESTED_MODEL).await?;
assert!(child_request.body_contains_text(TURN_0_FORK_PROMPT));
let child_body = child_request.body_json();
let original_parent_turn_id = parent_body["client_metadata"]["turn_id"]
.as_str()
.expect("legacy spawn parent turn id");
assert_parent_turn(&parent_body, /*expected*/ None)?;
assert_parent_turn(&child_body, Some(original_parent_turn_id))?;
assert_eq!(
(
child_body["model"].clone(),
@@ -972,7 +979,59 @@ async fn spawned_child_receives_forked_parent_context(
json!(REQUESTED_REASONING_EFFORT.to_string()),
)
);
let child_thread_id = ThreadId::from_string(
child_body["client_metadata"]["thread_id"]
.as_str()
.expect("legacy child thread id"),
)?;
let child_thread = test.thread_manager.get_thread(child_thread_id).await?;
tokio::time::timeout(Duration::from_secs(2), async {
while !matches!(child_thread.agent_status().await, AgentStatus::Completed(_)) {
sleep(Duration::from_millis(10)).await;
}
})
.await?;
let args = serde_json::to_string(&json!({
"target": child_thread_id.to_string(),
"message": "legacy child follow-up",
}))?;
let parent = mount_sse_once_match(
&server,
|request: &wiremock::Request| body_contains(request, "reuse the legacy child"),
sse(vec![
ev_response_created("resp-legacy-reuse"),
ev_function_call_with_namespace(
"legacy-reuse-call",
MULTI_AGENT_V1_NAMESPACE,
"send_input",
&args,
),
ev_completed("resp-legacy-reuse"),
]),
)
.await;
let followup = mount_sse_sequence(
&server,
vec![
sse(vec![ev_completed("resp-legacy-child-reuse")]),
sse(vec![ev_completed("resp-legacy-reuse-complete")]),
],
)
.await;
test.submit_turn("reuse the legacy child").await?;
let followup_parent_body = parent.single_request().body_json();
let reused_child_body = wait_for_request_with_model(&followup, REQUESTED_MODEL)
.await?
.body_json();
let followup_parent_turn_id = followup_parent_body["client_metadata"]["turn_id"]
.as_str()
.expect("legacy follow-up parent turn id");
assert_ne!(followup_parent_turn_id, original_parent_turn_id);
let metadata = &reused_child_body["client_metadata"];
assert_eq!(metadata["thread_id"], json!(child_thread_id));
assert_parent_turn(&followup_parent_body, /*expected*/ None)?;
assert_parent_turn(&reused_child_body, Some(followup_parent_turn_id))?;
Ok(())
}
codex-rs/mcp-server/src/codex_tool_runner.rs
+1
View File
@@ -119,6 +119,7 @@ pub async fn run_codex_tool_session(
},
client_user_message_id: None,
trace: None,
parent_turn_id: None,
};
if let Err(e) = thread.submit_with_id(submission).await {
codex-rs/mcp-server/src/message_processor.rs
+1
View File
@@ -560,6 +560,7 @@ impl MessageProcessor {
op: codex_protocol::protocol::Op::Interrupt,
client_user_message_id: None,
trace: None,
parent_turn_id: None,
})
.await
{
codex-rs/protocol/src/protocol.rs
+2
View File
@@ -182,6 +182,8 @@ pub struct Submission {
pub client_user_message_id: Option<String>,
/// Optional W3C trace carrier propagated across async submission handoffs.
pub trace: Option<W3cTraceContext>,
/// Core-provided ID of the parent turn that directly initiated this submission.
pub parent_turn_id: Option<String>,
}
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]