From 1f4ea798538d2c947ce215a65bc166db517a2df6 Mon Sep 17 00:00:00 2001 From: Eddie Chen <173840017+eddie-openai@users.noreply.github.com> Date: Wed, 12 Aug 2026 17:58:42 +0000 Subject: [PATCH] Track root turns across delegated Codex requests (#38232) ## Why `parent_turn_id` identifies the immediate caller of a nested request, but not the top-level turn that initiated a multi-level chain of work. ## What changed - Add `root_turn_id` to turn state, submissions, and Responses API metadata. - Propagate the root through spawned agents, delegated work, review sessions, and turn-triggering inter-agent messages. - Omit the root when background work has no user turn or combined input makes the root ambiguous. ## Testing Added coverage for nested agent resumes, reviews, mailbox delivery, steered input, and reserved metadata handling. GitOrigin-RevId: 60057db2ee81f4efefa946a9f63c74ff98b93929 --- codex-rs/core/src/agent/control.rs | 24 +++- codex-rs/core/src/agent/control/legacy.rs | 14 +- codex-rs/core/src/agent/control/spawn.rs | 2 + codex-rs/core/src/agent/control_tests.rs | 5 + codex-rs/core/src/codex_delegate.rs | 3 + codex-rs/core/src/codex_delegate_tests.rs | 1 + codex-rs/core/src/codex_thread.rs | 7 +- codex-rs/core/src/guardian/review_session.rs | 4 +- codex-rs/core/src/responses_metadata.rs | 10 ++ codex-rs/core/src/session/handlers.rs | 32 ++++- codex-rs/core/src/session/inject.rs | 9 ++ codex-rs/core/src/session/input_queue.rs | 126 ++++++++++++++---- codex-rs/core/src/session/mod.rs | 21 ++- codex-rs/core/src/session/review.rs | 3 + codex-rs/core/src/session/tests.rs | 70 ++++++++-- codex-rs/core/src/tasks/mod.rs | 19 ++- codex-rs/core/src/thread_manager.rs | 3 +- .../tools/handlers/multi_agents/send_input.rs | 7 +- .../src/tools/handlers/multi_agents/spawn.rs | 1 + .../src/tools/handlers/multi_agents_tests.rs | 5 + .../handlers/multi_agents_v2/message_tool.rs | 8 +- .../tools/handlers/multi_agents_v2/spawn.rs | 1 + codex-rs/core/src/turn_metadata.rs | 39 ++++++ codex-rs/core/src/turn_metadata_tests.rs | 12 ++ codex-rs/core/tests/common/responses.rs | 9 +- codex-rs/core/tests/suite/git_enrichment.rs | 11 +- codex-rs/core/tests/suite/guardian_review.rs | 22 +++ .../core/tests/suite/mcp_turn_metadata.rs | 19 ++- .../core/tests/suite/multi_agent_resume.rs | 12 ++ codex-rs/core/tests/suite/pending_input.rs | 8 ++ codex-rs/core/tests/suite/review.rs | 5 +- ...pi_auth_prompt_cache_key_request_diff.snap | 3 +- ...ce_tier_prompt_cache_key_request_diff.snap | 3 +- .../tests/suite/subagent_notifications.rs | 5 + .../tests/suite/user_message_admission.rs | 7 +- codex-rs/mcp-server/src/codex_tool_runner.rs | 2 + codex-rs/mcp-server/src/message_processor.rs | 1 + codex-rs/protocol/src/protocol.rs | 2 + 38 files changed, 470 insertions(+), 65 deletions(-) diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 36690f937b..d8c2f1f271 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -73,6 +73,7 @@ pub(crate) struct SpawnAgentOptions { pub(crate) fork_mode: Option, pub(crate) parent_thread_id: Option, pub(crate) parent_turn_id: Option, + pub(crate) root_turn_id: Option, pub(crate) environments: Option>, } @@ -169,11 +170,12 @@ impl AgentControl { agent_id: ThreadId, input: Vec, parent_turn_id: Option, + root_turn_id: Option, ) -> CodexResult { 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, parent_turn_id) + self.send_input_after_capacity_check(agent_id, &state, input, parent_turn_id, root_turn_id) .await } @@ -183,11 +185,14 @@ impl AgentControl { state: &Arc, input: Vec, parent_turn_id: Option, + root_turn_id: Option, ) -> CodexResult { self.handle_thread_request_result( agent_id, state, - state.send_op(agent_id, input.into(), parent_turn_id).await, + state + .send_op(agent_id, input.into(), parent_turn_id, root_turn_id) + .await, ) .await } @@ -198,6 +203,7 @@ impl AgentControl { communication: InterAgentCommunication, agent_communication_context: AgentCommunicationContext, parent_turn_id: Option, + root_turn_id: Option, ) -> CodexResult { let state = self.upgrade()?; self.ensure_execution_capacity_for_turn_start(agent_id, communication.trigger_turn) @@ -208,6 +214,7 @@ impl AgentControl { communication, agent_communication_context, parent_turn_id, + root_turn_id, ) .await } @@ -219,6 +226,7 @@ impl AgentControl { communication: InterAgentCommunication, context: AgentCommunicationContext, parent_turn_id: Option, + root_turn_id: Option, ) -> CodexResult { self.submit_inter_agent_communication( agent_id, @@ -226,6 +234,7 @@ impl AgentControl { communication, context, parent_turn_id, + root_turn_id, ) .await } @@ -237,10 +246,12 @@ impl AgentControl { communication: InterAgentCommunication, context: AgentCommunicationContext, parent_turn_id: Option, + root_turn_id: Option, ) -> CodexResult { let communication_for_log = crate::agent_communication::logging_enabled().then(|| communication.clone()); let parent_turn_id = parent_turn_id.filter(|_| communication.trigger_turn); + let root_turn_id = root_turn_id.filter(|_| communication.trigger_turn); let result = self .handle_thread_request_result( agent_id, @@ -250,6 +261,7 @@ impl AgentControl { agent_id, Op::InterAgentCommunication { communication }, parent_turn_id, + root_turn_id, ) .await, ) @@ -274,7 +286,12 @@ impl AgentControl { agent_id, &state, state - .send_op(agent_id, Op::Interrupt, /*parent_turn_id*/ None) + .send_op( + agent_id, + Op::Interrupt, + /*parent_turn_id*/ None, + /*root_turn_id*/ None, + ) .await, ) .await @@ -556,6 +573,7 @@ impl AgentControl { communication, context, /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await; return; diff --git a/codex-rs/core/src/agent/control/legacy.rs b/codex-rs/core/src/agent/control/legacy.rs index 72f757ab2b..f6bff6cd90 100644 --- a/codex-rs/core/src/agent/control/legacy.rs +++ b/codex-rs/core/src/agent/control/legacy.rs @@ -17,14 +17,24 @@ impl AgentControl { Ok(String::new()) } else { state - .send_op(agent_id, Op::Shutdown {}, /*parent_turn_id*/ None) + .send_op( + agent_id, + Op::Shutdown {}, + /*parent_turn_id*/ None, + /*root_turn_id*/ None, + ) .await }; thread.wait_until_terminated().await; result } else { state - .send_op(agent_id, Op::Shutdown {}, /*parent_turn_id*/ None) + .send_op( + agent_id, + Op::Shutdown {}, + /*parent_turn_id*/ None, + /*root_turn_id*/ None, + ) .await }; let _ = state.remove_thread(&agent_id).await; diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index a589ef84af..1de4768e65 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -548,6 +548,7 @@ impl AgentControl { &state, input, options.parent_turn_id, + options.root_turn_id, ) .await?; } @@ -558,6 +559,7 @@ impl AgentControl { communication, context, options.parent_turn_id, + options.root_turn_id, ) .await?; } diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index bb2525d02b..df13e1bf66 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -418,6 +418,7 @@ async fn send_input_errors_when_manager_dropped() { text_elements: Vec::new(), }], /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await .expect_err("send_input should fail without a manager"); @@ -533,6 +534,7 @@ async fn send_input_errors_when_thread_missing() { text_elements: Vec::new(), }], /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await .expect_err("send_input should fail for missing thread"); @@ -606,6 +608,7 @@ async fn send_input_submits_user_message() { text_elements: Vec::new(), }], /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await .expect("send_input should succeed"); @@ -650,6 +653,7 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig communication.clone(), AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()), /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await .expect("send_inter_agent_communication should succeed"); @@ -815,6 +819,7 @@ async fn ensure_v2_agent_loaded_reloads_registered_unloaded_agent() { communication.clone(), AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()), /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await .expect("send_inter_agent_communication should succeed after reload"); diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index c779296ff8..3f74617b82 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -173,6 +173,7 @@ pub(crate) async fn run_codex_thread_one_shot( let child_cancel = cancel_token.child_token(); let parent_turn_id = parent_ctx.sub_id.clone(); let parent_environments = parent_ctx.environments.clone(); + let root_turn_id = parent_ctx.turn_metadata_state.root_turn_id(); let (session, io) = Box::pin(run_codex_thread_interactive( config, auth_manager, @@ -199,6 +200,7 @@ pub(crate) async fn run_codex_thread_one_shot( }, /*trace*/ None, Some(parent_turn_id), + root_turn_id, ) .await?; @@ -223,6 +225,7 @@ pub(crate) async fn run_codex_thread_one_shot( client_user_message_id: None, trace: None, parent_turn_id: None, + root_turn_id: None, }) .await; child_cancel.cancel(); diff --git a/codex-rs/core/src/codex_delegate_tests.rs b/codex-rs/core/src/codex_delegate_tests.rs index 159f026e05..c8cbfde2f7 100644 --- a/codex-rs/core/src/codex_delegate_tests.rs +++ b/codex-rs/core/src/codex_delegate_tests.rs @@ -136,6 +136,7 @@ async fn forward_ops_preserves_submission_trace_context() { tracestate: Some("vendor=state".to_string()), }), parent_turn_id: Some("parent-turn".to_string()), + root_turn_id: Some("root-turn".to_string()), }; tx_ops.send(submission).await.unwrap(); drop(tx_ops); diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index d1212f90bf..16956a112b 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -294,7 +294,9 @@ impl CodexThread { trace: Option, ) -> CodexResult { self.io - .submit_with_trace(op, trace, /*parent_turn_id*/ None) + .submit_with_trace( + op, trace, /*parent_turn_id*/ None, /*root_turn_id*/ None, + ) .await } @@ -343,6 +345,7 @@ impl CodexThread { client_user_message_id, trace, parent_turn_id: None, + root_turn_id: None, }) .await?; tokio::select! { @@ -374,6 +377,7 @@ impl CodexThread { expected_turn_id, client_user_message_id, responsesapi_client_metadata, + /*incoming_turn_metadata*/ None, ) .await } @@ -483,6 +487,7 @@ impl CodexThread { /// Use sparingly: this is intended to be removed soon. pub async fn submit_with_id(&self, mut sub: Submission) -> CodexResult<()> { sub.parent_turn_id = None; + sub.root_turn_id = None; self.io.submit_with_id(sub).await } diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 3899b0a5e5..21f903a186 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -887,6 +887,7 @@ async fn run_review_on_session( .and_then(|environment| environment.cwd().to_abs_path().ok()) .unwrap_or_else(|| params.parent_context.turn().config.cwd.clone()); + let parent_turn = params.parent_context.turn(); let submission = review_session.io.submit_with_trace( Op::UserInput { items: prompt_items.items, @@ -915,7 +916,8 @@ async fn run_review_on_session( }, }, /*trace*/ None, - Some(params.parent_context.turn().sub_id.clone()), + Some(parent_turn.sub_id.clone()), + parent_turn.turn_metadata_state.root_turn_id(), ); let submit_result = run_before_review_deadline( deadline, diff --git a/codex-rs/core/src/responses_metadata.rs b/codex-rs/core/src/responses_metadata.rs index 6bbdf04de7..e785af35ab 100644 --- a/codex-rs/core/src/responses_metadata.rs +++ b/codex-rs/core/src/responses_metadata.rs @@ -38,6 +38,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 ROOT_TURN_ID_KEY: &str = "root_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"; @@ -66,6 +67,7 @@ const RESERVED_METADATA_KEYS: &[&str] = &[ FORKED_FROM_THREAD_ID_KEY, PARENT_THREAD_ID_KEY, PARENT_TURN_ID_KEY, + ROOT_TURN_ID_KEY, SUBAGENT_KIND_KEY, THREAD_SOURCE_KEY, SANDBOX_KEY, @@ -204,6 +206,7 @@ pub struct CodexResponsesMetadata { pub(crate) forked_from_thread_id: Option, pub(crate) parent_thread_id: Option, pub(crate) parent_turn_id: Option, + pub(crate) root_turn_id: Option, pub(crate) subagent_header: Option, pub(crate) subagent_kind: Option, pub(crate) thread_source: Option, @@ -234,6 +237,7 @@ impl CodexResponsesMetadata { forked_from_thread_id: None, parent_thread_id: None, parent_turn_id: None, + root_turn_id: None, subagent_header: None, subagent_kind: None, thread_source: None, @@ -287,6 +291,9 @@ impl CodexResponsesMetadata { if let Some(parent_turn_id) = &self.parent_turn_id { client_metadata.insert(PARENT_TURN_ID_KEY.to_string(), parent_turn_id.clone()); } + if let Some(root_turn_id) = &self.root_turn_id { + client_metadata.insert(ROOT_TURN_ID_KEY.to_string(), root_turn_id.clone()); + } if self.has_turn_metadata() && let Some(turn_metadata_json) = self.turn_metadata_json() { @@ -347,6 +354,7 @@ impl CodexResponsesMetadata { 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(), + root_turn_id: self.root_turn_id.as_deref(), subagent_kind: self.subagent_kind.as_deref(), thread_source: self.thread_source.as_ref(), sandbox: self.sandbox.as_deref(), @@ -468,6 +476,8 @@ struct CodexTurnMetadataPayload<'a> { #[serde(default, skip_serializing_if = "Option::is_none")] parent_turn_id: Option<&'a str>, #[serde(default, skip_serializing_if = "Option::is_none")] + root_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>, diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index aeaa975af5..558527e4bf 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -88,6 +88,7 @@ pub async fn user_input_or_turn( op: Op, client_user_message_id: Option, parent_turn_id: Option, + root_turn_id: Option, ) { let admission = user_input_or_turn_inner( sess, @@ -95,6 +96,7 @@ pub async fn user_input_or_turn( op, client_user_message_id, parent_turn_id, + root_turn_id, ) .await; sess.pending_user_message_admissions @@ -193,6 +195,7 @@ pub(super) async fn user_input_or_turn_inner( op: Op, client_user_message_id: Option, parent_turn_id: Option, + root_turn_id: Option, ) -> CodexResult { let Op::UserInput { items, @@ -223,6 +226,14 @@ pub(super) async fn user_input_or_turn_inner( } sess.maybe_emit_model_warnings_for_turn(current_context.as_ref()) .await; + if let Some(id) = root_turn_id.as_ref() { + current_context + .turn_metadata_state + .set_root_turn_id(id.clone()); + } + let incoming_turn_metadata = parent_turn_id + .as_ref() + .map(|_| current_context.turn_metadata_state.as_ref()); match sess .steer_input( items.clone(), @@ -230,6 +241,7 @@ pub(super) async fn user_input_or_turn_inner( /*expected_turn_id*/ None, client_user_message_id.clone(), responsesapi_client_metadata.clone(), + incoming_turn_metadata, ) .await { @@ -238,6 +250,17 @@ pub(super) async fn user_input_or_turn_inner( Ok(UserMessageAdmission::Steered { turn_id }) } Err(SteerInputError::NoActiveTurn(items)) => { + if root_turn_id.is_none() + && parent_turn_id.is_none() + && !items.is_empty() + && current_context + .turn_metadata_state + .can_start_root_turn(¤t_context.session_source) + { + current_context + .turn_metadata_state + .set_root_turn_id(sub_id.clone()); + } if let Some(id) = parent_turn_id { current_context.turn_metadata_state.set_parent_turn_id(id); } @@ -290,10 +313,15 @@ pub async fn inter_agent_communication( sub_id: String, communication: InterAgentCommunication, parent_turn_id: Option, + root_turn_id: Option, ) { let trigger_turn = communication.trigger_turn; sess.input_queue - .enqueue_mailbox_communication(communication, parent_turn_id.filter(|_| trigger_turn)) + .enqueue_mailbox_communication( + communication, + parent_turn_id.filter(|_| trigger_turn), + root_turn_id.filter(|_| trigger_turn), + ) .await; crate::agent_communication::emit_agent_communication_receive(&sub_id); if trigger_turn || sess.has_outstanding_durable_sleep() { @@ -765,6 +793,7 @@ pub(super) async fn submission_loop( op, sub.client_user_message_id, sub.parent_turn_id, + sub.root_turn_id, ) .await; false @@ -779,6 +808,7 @@ pub(super) async fn submission_loop( sub.id.clone(), communication, sub.parent_turn_id, + sub.root_turn_id, ) .await; false diff --git a/codex-rs/core/src/session/inject.rs b/codex-rs/core/src/session/inject.rs index 9b73b05370..4b3fc7c752 100644 --- a/codex-rs/core/src/session/inject.rs +++ b/codex-rs/core/src/session/inject.rs @@ -123,6 +123,15 @@ impl Session { )); } + if has_user_input + && turn_context + .turn_metadata_state + .can_start_root_turn(&turn_context.session_source) + { + turn_context + .turn_metadata_state + .set_root_turn_id(turn_context.sub_id.clone()); + } let task_input = if has_user_input { self.clear_connector_selection().await; for item in &input { diff --git a/codex-rs/core/src/session/input_queue.rs b/codex-rs/core/src/session/input_queue.rs index 9f35a335e2..9b97abf127 100644 --- a/codex-rs/core/src/session/input_queue.rs +++ b/codex-rs/core/src/session/input_queue.rs @@ -47,6 +47,7 @@ pub(crate) struct InputQueue { struct PendingMailboxCommunication { communication: InterAgentCommunication, parent_turn_id: Option, + root_turn_id: Option, _diagnostics_guard: GaugeGuard, } @@ -86,6 +87,7 @@ impl InputQueue { &self, communication: InterAgentCommunication, parent_turn_id: Option, + root_turn_id: Option, ) { self.mailbox_pending_mails .lock() @@ -93,6 +95,7 @@ impl InputQueue { .push_back(PendingMailboxCommunication { communication, parent_turn_id, + root_turn_id, _diagnostics_guard: PENDING_MAILBOX_MESSAGES.track(), }); self.activity_tx.send_replace(InputQueueActivity::Mailbox); @@ -110,7 +113,9 @@ impl InputQueue { .any(|mail| mail.communication.trigger_turn) } - pub(crate) async fn drain_mailbox_input_items(&self) -> (Vec, Option) { + pub(crate) async fn drain_mailbox_input_items( + &self, + ) -> (Vec, Option, Option) { let pending_mails = self .mailbox_pending_mails .lock() @@ -123,11 +128,22 @@ impl InputQueue { .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 root_turn_id = pending_mails + .iter() + .filter(|mail| mail.communication.trigger_turn) + .map(|mail| { + mail.parent_turn_id + .as_deref() + .filter(|id| !id.trim().is_empty()) + .and(mail.root_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) + (items, parent_turn_id, root_turn_id) } pub(crate) async fn turn_state_for_sub_id( @@ -233,11 +249,15 @@ impl InputQueue { pub(crate) async fn get_pending_input( &self, active_turn: &Mutex>, - ) -> (Vec, Option) { - let (pending_input, accepts_mailbox_delivery) = { + ) -> (Vec, Option, Option) { + let (pending_input, accepts_mailbox_delivery, active_turn_metadata) = { let mut active = active_turn.lock().await; match active.as_mut() { Some(active_turn) => { + let active_turn_metadata = active_turn + .task + .as_ref() + .map(|task| Arc::clone(&task.turn_context.turn_metadata_state)); let mut turn_state = active_turn.turn_state.lock().await; let accepts_mailbox_delivery = turn_state.accepts_mailbox_delivery_for_current_turn(); @@ -246,21 +266,37 @@ impl InputQueue { } else { Vec::new() }; - (pending_input, accepts_mailbox_delivery) + ( + pending_input, + accepts_mailbox_delivery, + active_turn_metadata, + ) } - None => (Vec::new(), true), + None => (Vec::new(), true, None), } }; if !accepts_mailbox_delivery { - return (pending_input, None); + return (pending_input, None, None); + } + let (mailbox_items, parent_turn_id, root_turn_id) = self.drain_mailbox_input_items().await; + if let Some(active_turn_metadata) = active_turn_metadata + && mailbox_items.iter().any(|item| { + matches!( + item, + TurnInput::InterAgentCommunication(communication) + if communication.trigger_turn + ) + }) + && (root_turn_id.is_none() || active_turn_metadata.root_turn_id() != root_turn_id) + { + active_turn_metadata.mark_root_turn_ambiguous(); } - let (mailbox_items, parent_turn_id) = self.drain_mailbox_input_items().await; if pending_input.is_empty() { - (mailbox_items, parent_turn_id) + (mailbox_items, parent_turn_id, root_turn_id) } else { let mut pending_input = pending_input; pending_input.extend(mailbox_items); - (pending_input, parent_turn_id) + (pending_input, parent_turn_id, root_turn_id) } } @@ -335,7 +371,9 @@ mod tests { /*trigger_turn*/ false, ); input_queue - .enqueue_mailbox_communication(mail_one, /*parent_turn_id*/ None) + .enqueue_mailbox_communication( + mail_one, /*parent_turn_id*/ None, /*root_turn_id*/ None, + ) .await; let mail_two = make_mail( AgentPath::root(), @@ -344,7 +382,9 @@ mod tests { /*trigger_turn*/ false, ); input_queue - .enqueue_mailbox_communication(mail_two, /*parent_turn_id*/ None) + .enqueue_mailbox_communication( + mail_two, /*parent_turn_id*/ None, /*root_turn_id*/ None, + ) .await; activity_rx.changed().await.expect("mailbox update"); @@ -419,10 +459,18 @@ mod tests { ); input_queue - .enqueue_mailbox_communication(mail_one.clone(), /*parent_turn_id*/ None) + .enqueue_mailbox_communication( + mail_one.clone(), + /*parent_turn_id*/ None, + /*root_turn_id*/ None, + ) .await; input_queue - .enqueue_mailbox_communication(mail_two.clone(), /*parent_turn_id*/ None) + .enqueue_mailbox_communication( + mail_two.clone(), + /*parent_turn_id*/ None, + /*root_turn_id*/ None, + ) .await; assert_eq!( @@ -437,28 +485,42 @@ mod tests { #[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 (parent, peer, root, root2) = (Some("a"), Some("b"), Some("r"), Some("s")); + for (pending_mails, expected_parent_turn_id, expected_root_turn_id) in [ + (Vec::new(), None, None), + (vec![(false, Some("q"), root)], None, None), + (vec![(true, Some(""), root)], None, None), + (vec![(true, Some(" "), root)], None, None), + (vec![(true, None, root)], None, None), + (vec![(true, parent, None)], parent, None), + (vec![(true, parent, Some(""))], parent, None), + (vec![(true, parent, root), (true, peer, root)], None, root), + (vec![(true, parent, root), (true, peer, root2)], None, None), + (vec![(true, parent, root), (true, None, root)], None, None), + ( + vec![(true, parent, root), (true, parent, root)], + parent, + root, + ), + ( + vec![(false, Some("q"), root2), (true, parent, root)], + parent, + root, + ), ] { let input_queue = InputQueue::new(); - for (trigger_turn, parent_turn_id) in pending_mails { + for (trigger_turn, parent_turn_id, root_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), + root_turn_id.map(str::to_string), ) .await; } - let (_, parent_turn_id) = input_queue.drain_mailbox_input_items().await; + let (_, parent_turn_id, root_turn_id) = input_queue.drain_mailbox_input_items().await; assert_eq!(parent_turn_id.as_deref(), expected_parent_turn_id); + assert_eq!(root_turn_id.as_deref(), expected_root_turn_id); } } @@ -473,7 +535,11 @@ mod tests { /*trigger_turn*/ false, ); input_queue - .enqueue_mailbox_communication(queued_mail, /*parent_turn_id*/ None) + .enqueue_mailbox_communication( + queued_mail, + /*parent_turn_id*/ None, + /*root_turn_id*/ None, + ) .await; assert!(!input_queue.has_trigger_turn_mailbox_items().await); @@ -484,7 +550,11 @@ mod tests { /*trigger_turn*/ true, ); input_queue - .enqueue_mailbox_communication(trigger_mail, /*parent_turn_id*/ None) + .enqueue_mailbox_communication( + trigger_mail, + /*parent_turn_id*/ None, + /*root_turn_id*/ None, + ) .await; assert!(input_queue.has_trigger_turn_mailbox_items().await); } diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 5dd12bc91b..dd77f8504d 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -830,8 +830,10 @@ impl Session { impl SessionIo { /// Submit the `op` wrapped in a `Submission` with a unique ID. pub(crate) async fn submit(&self, op: Op) -> CodexResult { - self.submit_with_trace(op, /*trace*/ None, /*parent_turn_id*/ None) - .await + self.submit_with_trace( + op, /*trace*/ None, /*parent_turn_id*/ None, /*root_turn_id*/ None, + ) + .await } pub(crate) async fn submit_with_trace( @@ -839,6 +841,7 @@ impl SessionIo { op: Op, trace: Option, parent_turn_id: Option, + root_turn_id: Option, ) -> CodexResult { let id = new_submission_id(); let sub = Submission { @@ -847,6 +850,7 @@ impl SessionIo { client_user_message_id: None, trace, parent_turn_id, + root_turn_id, }; self.submit_with_id(sub).await?; Ok(id) @@ -866,6 +870,7 @@ impl SessionIo { client_user_message_id, trace, parent_turn_id: None, + root_turn_id: None, }; self.submit_with_id(sub).await?; Ok(id) @@ -1259,6 +1264,7 @@ impl Session { }, /*client_user_message_id*/ None, /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await; } @@ -2046,6 +2052,7 @@ impl Session { communication, context, /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await { @@ -4025,6 +4032,7 @@ impl Session { expected_turn_id: Option<&str>, client_user_message_id: Option, responsesapi_client_metadata: Option>, + incoming_turn_metadata: Option<&TurnMetadataState>, ) -> Result { let mut active = self.active_turn.lock().await; let Some(active_turn) = active.as_mut() else { @@ -4084,6 +4092,15 @@ impl Session { content: input, client_id: client_user_message_id.clone(), }); + if let Some(incoming_turn_metadata) = incoming_turn_metadata + && active_task.turn_context.turn_metadata_state.root_turn_id() + != incoming_turn_metadata.root_turn_id() + { + active_task + .turn_context + .turn_metadata_state + .mark_root_turn_ambiguous(); + } self.input_queue .extend_pending_input_and_accept_mailbox_delivery_for_turn_state( active_turn.turn_state.as_ref(), diff --git a/codex-rs/core/src/session/review.rs b/codex-rs/core/src/session/review.rs index f7719fd365..9bb5c33428 100644 --- a/codex-rs/core/src/session/review.rs +++ b/codex-rs/core/src/session/review.rs @@ -121,6 +121,9 @@ pub(super) async fn spawn_review_thread( parent_turn_context.network.is_some(), auto_review_enabled, )); + if turn_metadata_state.can_start_root_turn(&session_source) { + turn_metadata_state.set_root_turn_id(review_turn_id.clone()); + } let extension_data = Arc::new(codex_extension_api::ExtensionData::new( review_turn_id.clone(), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 3a9cec895b..5f2a3ebdd2 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -6939,6 +6939,7 @@ async fn submit_with_id_captures_current_span_trace_context() { op: Op::Interrupt, client_user_message_id: None, parent_turn_id: None, + root_turn_id: None, trace: None, }) .await @@ -7012,6 +7013,7 @@ fn submission_dispatch_span_prefers_submission_trace_context() { op: Op::Interrupt, client_user_message_id: None, parent_turn_id: None, + root_turn_id: None, trace: Some(submission_trace), }) }); @@ -7040,6 +7042,7 @@ fn submission_dispatch_span_uses_debug_for_realtime_audio() { }), client_user_message_id: None, parent_turn_id: None, + root_turn_id: None, trace: None, }); @@ -7106,6 +7109,7 @@ async fn user_turn_updates_approvals_reviewer() { }, /*client_user_message_id*/ None, /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await; @@ -7403,6 +7407,7 @@ async fn spawn_task_turn_span_inherits_dispatch_trace_context() { op: Op::Interrupt, client_user_message_id: None, parent_turn_id: None, + root_turn_id: None, trace: Some(submission_trace.clone()), }); let dispatch_span_id = dispatch_span.context().span().span_context().span_id(); @@ -10435,6 +10440,7 @@ async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input() Some(&tc.sub_id), /*client_user_message_id*/ None, /*responsesapi_client_metadata*/ None, + /*incoming_turn_metadata*/ None, ) .await .expect("steer pending input into active turn"); @@ -10606,6 +10612,7 @@ async fn thread_idle_lifecycle_waits_for_trigger_turn_mailbox_work() { /*trigger_turn*/ true, ), /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await; @@ -10638,7 +10645,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::::new(), None), + (Vec::::new(), None, None), sess.input_queue.get_pending_input(&sess.active_turn).await ); @@ -10665,7 +10672,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::::new(), None), + (Vec::::new(), None, None), sess.input_queue.get_pending_input(&sess.active_turn).await ); } @@ -10736,6 +10743,7 @@ async fn try_start_turn_if_idle_rejects_pending_trigger_turn_without_injecting() /*trigger_turn*/ true, ), /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await; @@ -10776,7 +10784,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::::new(), None), + (Vec::::new(), None, None), sess.input_queue.get_pending_input(&sess.active_turn).await ); @@ -10798,6 +10806,7 @@ async fn steer_input_requires_active_turn() { /*expected_turn_id*/ None, /*client_user_message_id*/ None, /*responsesapi_client_metadata*/ None, + /*incoming_turn_metadata*/ None, ) .await .expect_err("steering without active turn should fail"); @@ -10836,6 +10845,7 @@ async fn steer_input_enforces_expected_turn_id() { Some("different-turn-id"), /*client_user_message_id*/ None, /*responsesapi_client_metadata*/ None, + /*incoming_turn_metadata*/ None, ) .await .expect_err("mismatched expected turn id should fail"); @@ -10857,7 +10867,10 @@ async fn steer_input_rejects_non_regular_turns() { (TaskKind::Review, NonSteerableTurnKind::Review), (TaskKind::Compact, NonSteerableTurnKind::Compact), ] { - let (sess, _tc, _rx) = make_session_and_context_with_rx().await; + let (sess, incoming_turn_context, _rx) = make_session_and_context_with_rx().await; + incoming_turn_context + .turn_metadata_state + .set_root_turn_id("incoming-root".to_string()); let input = vec![TurnInput::UserInput { content: vec![UserInput::Text { text: "hello".to_string(), @@ -10866,8 +10879,11 @@ async fn steer_input_rejects_non_regular_turns() { client_id: None, }]; let turn_context = sess.new_default_turn_with_sub_id("turn".to_string()).await; + turn_context + .turn_metadata_state + .set_root_turn_id("active-root".to_string()); sess.spawn_task( - turn_context, + Arc::clone(&turn_context), input, NeverEndingTask { kind: task_kind, @@ -10887,11 +10903,16 @@ async fn steer_input_rejects_non_regular_turns() { /*expected_turn_id*/ None, /*client_user_message_id*/ None, /*responsesapi_client_metadata*/ None, + Some(incoming_turn_context.turn_metadata_state.as_ref()), ) .await .expect_err("steering a non-regular turn should fail"); assert_eq!(err, SteerInputError::ActiveTurnNotSteerable { turn_kind }); + assert_eq!( + turn_context.turn_metadata_state.root_turn_id().as_deref(), + Some("active-root") + ); sess.abort_all_tasks(TurnAbortReason::Interrupted).await; } @@ -10900,6 +10921,14 @@ async fn steer_input_rejects_non_regular_turns() { #[tokio::test] async fn steer_input_returns_active_turn_id() { let (sess, tc, _rx) = make_session_and_context_with_rx().await; + tc.turn_metadata_state + .set_root_turn_id("active-root".to_string()); + let incoming_turn_context = sess + .new_default_turn_with_sub_id("incoming".to_string()) + .await; + incoming_turn_context + .turn_metadata_state + .set_root_turn_id("incoming-root".to_string()); let input = vec![TurnInput::UserInput { content: vec![UserInput::Text { text: "hello".to_string(), @@ -10928,11 +10957,13 @@ async fn steer_input_returns_active_turn_id() { Some(&tc.sub_id), /*client_user_message_id*/ None, /*responsesapi_client_metadata*/ None, + Some(incoming_turn_context.turn_metadata_state.as_ref()), ) .await .expect("steering with matching expected turn id should succeed"); assert_eq!(turn_id, tc.sub_id); + assert_eq!(tc.turn_metadata_state.root_turn_id(), None); assert!(sess.input_queue.has_pending_input(&sess.active_turn).await); } @@ -11004,7 +11035,11 @@ 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(), /*parent_turn_id*/ None) + .enqueue_mailbox_communication( + communication.clone(), + /*parent_turn_id*/ None, + /*root_turn_id*/ None, + ) .await; assert!( @@ -11013,7 +11048,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::new(), None) + (Vec::new(), None, None) ); sess.abort_all_tasks(TurnAbortReason::Replaced).await; @@ -11050,6 +11085,7 @@ async fn trigger_turn_mailbox_mail_waits_for_next_turn_after_answer_boundary() { /*trigger_turn*/ true, ), /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await; @@ -11087,7 +11123,11 @@ 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(), /*parent_turn_id*/ None) + .enqueue_mailbox_communication( + communication.clone(), + /*parent_turn_id*/ None, + /*root_turn_id*/ None, + ) .await; sess.steer_input( vec![UserInput::Text { @@ -11098,6 +11138,7 @@ async fn steered_input_reopens_mailbox_delivery_for_current_turn() { Some(&tc.sub_id), /*client_user_message_id*/ None, /*responsesapi_client_metadata*/ None, + /*incoming_turn_metadata*/ None, ) .await .expect("steered input should be accepted"); @@ -11141,7 +11182,11 @@ 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(), /*parent_turn_id*/ None) + .enqueue_mailbox_communication( + communication.clone(), + /*parent_turn_id*/ None, + /*root_turn_id*/ None, + ) .await; sess.steer_input( vec![UserInput::Text { @@ -11152,6 +11197,7 @@ async fn stale_defer_mailbox_delivery_does_not_override_steered_input() { Some(&tc.sub_id), /*client_user_message_id*/ None, /*responsesapi_client_metadata*/ None, + /*incoming_turn_metadata*/ None, ) .await .expect("steered input should be accepted"); @@ -11199,7 +11245,11 @@ 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(), /*parent_turn_id*/ None) + .enqueue_mailbox_communication( + communication.clone(), + /*parent_turn_id*/ None, + /*root_turn_id*/ None, + ) .await; let item = ResponseItem::FunctionCall { diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 3418ba83c1..3f1c7f9c0f 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -317,12 +317,23 @@ impl Session { .await .clear_turn(&turn_context.sub_id); - let (pending_items, parent_turn_id) = + let (pending_items, parent_turn_id, root_turn_id) = self.input_queue.get_pending_input(&self.active_turn).await; - if let (MailboxParentProvenance::Attribute, Some(id)) = - (mailbox_parent_provenance, parent_turn_id) + if let MailboxParentProvenance::Attribute = mailbox_parent_provenance { + if let Some(id) = parent_turn_id { + turn_context.turn_metadata_state.set_parent_turn_id(id); + } + if let Some(id) = root_turn_id { + turn_context.turn_metadata_state.set_root_turn_id(id); + } + } else if pending_items.iter().any(|item| { + matches!( + item, + TurnInput::InterAgentCommunication(communication) if communication.trigger_turn + ) + }) && turn_context.turn_metadata_state.root_turn_id() != root_turn_id { - turn_context.turn_metadata_state.set_parent_turn_id(id); + turn_context.turn_metadata_state.mark_root_turn_ambiguous(); } let turn_state = { let mut active = self.active_turn.lock().await; diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index d5297df04f..e46ac6c99d 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -1374,6 +1374,7 @@ impl ThreadManagerState { thread_id: ThreadId, op: Op, parent_turn_id: Option, + root_turn_id: Option, ) -> CodexResult { let thread = self.get_thread(thread_id).await?; if let Some(ops_log) = &self.ops_log @@ -1384,7 +1385,7 @@ impl ThreadManagerState { } thread .io - .submit_with_trace(op, /*trace*/ None, parent_turn_id) + .submit_with_trace(op, /*trace*/ None, parent_turn_id, root_turn_id) .await } diff --git a/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs b/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs index 1104fc2070..a5e1c454d7 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs @@ -86,7 +86,12 @@ impl Handler { .await; let agent_control = session.services.agent_control.clone(); let result = agent_control - .send_input(receiver_thread_id, input_items, Some(turn.sub_id.clone())) + .send_input( + receiver_thread_id, + input_items, + Some(turn.sub_id.clone()), + turn.turn_metadata_state.root_turn_id(), + ) .await .map_err(|err| collab_agent_error(receiver_thread_id, err)); let status = session diff --git a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs index cb159993f2..d7ffb2c0bb 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs @@ -136,6 +136,7 @@ async fn handle_spawn_agent( fork_mode: args.fork_context.then_some(SpawnAgentForkMode::FullHistory), parent_thread_id: Some(session.thread_id), parent_turn_id: Some(turn.sub_id.clone()), + root_turn_id: turn.turn_metadata_state.root_turn_id(), environments: Some(step_context.environments.to_selections()), }, )) diff --git a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs index e37a63588f..1de54d29f1 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -2976,6 +2976,7 @@ async fn multi_agent_v2_wait_agent_accepts_timeout_only_argument() { /*trigger_turn*/ false, ), /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await; @@ -3480,6 +3481,7 @@ async fn multi_agent_v2_wait_agent_returns_summary_for_mailbox_activity() { /*trigger_turn*/ false, ), /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await; @@ -3556,6 +3558,7 @@ async fn multi_agent_v2_wait_agent_returns_for_already_queued_mail() { /*trigger_turn*/ false, ), /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await; @@ -3658,6 +3661,7 @@ async fn multi_agent_v2_wait_agent_wakes_on_any_mailbox_notification() { /*trigger_turn*/ false, ), /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await; @@ -3749,6 +3753,7 @@ async fn multi_agent_v2_wait_agent_does_not_return_completed_content() { /*trigger_turn*/ false, ), /*parent_turn_id*/ None, + /*root_turn_id*/ None, ) .await; diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs index eb16bc336b..b217fa2dc1 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs @@ -112,7 +112,13 @@ pub(crate) async fn handle_message_string_tool( let result = session .services .agent_control - .send_inter_agent_communication(receiver_thread_id, communication, context, parent_turn_id) + .send_inter_agent_communication( + receiver_thread_id, + communication, + context, + parent_turn_id, + turn.turn_metadata_state.root_turn_id(), + ) .await .map_err(|err| collab_agent_error(receiver_thread_id, err)); result?; diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs index d166a325c4..c8289c0fd5 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs @@ -136,6 +136,7 @@ async fn handle_spawn_agent( fork_mode, parent_thread_id: Some(session.thread_id), parent_turn_id: Some(turn.sub_id.clone()), + root_turn_id: turn.turn_metadata_state.root_turn_id(), environments: Some(step_context.environments.to_selections()), }, ), diff --git a/codex-rs/core/src/turn_metadata.rs b/codex-rs/core/src/turn_metadata.rs index f473cae228..27ead5c714 100644 --- a/codex-rs/core/src/turn_metadata.rs +++ b/codex-rs/core/src/turn_metadata.rs @@ -15,6 +15,7 @@ use tokio::task::JoinHandle; use crate::responses_metadata::CodexResponsesMetadata; use crate::responses_metadata::CodexResponsesRequestKind; use crate::responses_metadata::PARENT_TURN_ID_KEY; +use crate::responses_metadata::ROOT_TURN_ID_KEY; use crate::responses_metadata::TurnMetadataWorkspace; use crate::responses_metadata::TurnToolNamespacesInfo; use crate::responses_metadata::filter_extra_metadata; @@ -101,6 +102,7 @@ pub(crate) struct TurnMetadataState { forked_from_thread_id: Option, parent_thread_id: Option, parent_turn_id: OnceLock, + root_turn_id: OnceLock, subagent_header: Option, subagent_kind: Option, thread_source: Option, @@ -113,6 +115,7 @@ pub(crate) struct TurnMetadataState { turn_started_at_unix_ms: RwLock>, responses_api_metadata: RwLock>, responsesapi_client_metadata: RwLock>, + root_turn_ambiguous: AtomicBool, user_input_requested_during_turn: AtomicBool, enrichment_task: Mutex>>, } @@ -152,6 +155,7 @@ impl TurnMetadataState { forked_from_thread_id, parent_thread_id, parent_turn_id: OnceLock::new(), + root_turn_id: OnceLock::new(), subagent_header: subagent_header_value(session_source), subagent_kind: subagent_metadata_kind(session_source), thread_source, @@ -164,6 +168,7 @@ impl TurnMetadataState { turn_started_at_unix_ms: RwLock::new(None), responses_api_metadata: RwLock::new(BTreeMap::new()), responsesapi_client_metadata: RwLock::new(BTreeMap::new()), + root_turn_ambiguous: AtomicBool::new(false), user_input_requested_during_turn: AtomicBool::new(false), enrichment_task: Mutex::new(None), } @@ -180,6 +185,7 @@ impl TurnMetadataState { return None; }; metadata.remove(PARENT_TURN_ID_KEY); + metadata.remove(ROOT_TURN_ID_KEY); metadata.insert( MODEL_KEY.to_string(), Value::String(context.model.to_string()), @@ -243,6 +249,38 @@ impl TurnMetadataState { let _ = self.parent_turn_id.set(parent_turn_id); } + pub(crate) fn set_root_turn_id(&self, root_turn_id: String) { + if root_turn_id.trim().is_empty() { + return; + } + let _ = self.root_turn_id.set(root_turn_id); + } + + pub(crate) fn root_turn_id(&self) -> Option { + self.root_turn_id + .get() + .filter(|_| !self.root_turn_ambiguous.load(Ordering::Relaxed)) + .cloned() + } + + pub(crate) fn mark_root_turn_ambiguous(&self) { + self.root_turn_ambiguous.store(true, Ordering::Relaxed); + } + + pub(crate) fn can_start_root_turn(&self, session_source: &SessionSource) -> bool { + if session_source.is_non_root_agent() { + return false; + } + match &self.thread_source { + // Desktop create/fork/send lacks trusted app-server provenance; fail closed. + Some(ThreadSource::Subagent | ThreadSource::MemoryConsolidation) => false, + Some(ThreadSource::Feature(feature)) => { + !matches!(feature.as_str(), "system" | "title") && !feature.starts_with("ambient") + } + Some(ThreadSource::User) | None => true, + } + } + pub(crate) fn set_responsesapi_client_metadata( &self, responsesapi_client_metadata: HashMap, @@ -302,6 +340,7 @@ impl TurnMetadataState { 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(), + root_turn_id: self.root_turn_id(), subagent_header: self.subagent_header.clone(), subagent_kind: self.subagent_kind.clone(), thread_source: self.thread_source.clone(), diff --git a/codex-rs/core/src/turn_metadata_tests.rs b/codex-rs/core/src/turn_metadata_tests.rs index 6266057d8f..aa0c9375de 100644 --- a/codex-rs/core/src/turn_metadata_tests.rs +++ b/codex-rs/core/src/turn_metadata_tests.rs @@ -6,6 +6,7 @@ use crate::responses_metadata::CompactionTurnMetadata; use crate::responses_metadata::INSTALLATION_ID_KEY; use crate::responses_metadata::LEGACY_CODE_MODE_TOOL_NAMES_KEY; use crate::responses_metadata::PARENT_TURN_ID_KEY; +use crate::responses_metadata::ROOT_TURN_ID_KEY; use crate::responses_metadata::SANDBOX_MODE_KEY; use crate::responses_metadata::TOOL_NAMESPACES_INFO_KEY; use crate::responses_metadata::TurnToolFunctionInfo; @@ -164,6 +165,7 @@ async fn detached_memory_responses_metadata_omits_turn_identity() { assert!(parsed.get("thread_id").is_none()); assert!(parsed.get("forked_from_thread_id").is_none()); assert!(parsed.get("turn_id").is_none()); + assert!(parsed.get(ROOT_TURN_ID_KEY).is_none()); assert!(parsed.get(WINDOW_ID_KEY).is_none()); let expected_repo_path = repo_path.to_string_lossy().into_owned(); @@ -594,6 +596,7 @@ fn turn_metadata_state_ignores_client_reserved_metadata_before_start() { "client-supplied".to_string(), ), ("parent_turn_id".to_string(), "client-supplied".to_string()), + (ROOT_TURN_ID_KEY.to_string(), "client-supplied".to_string()), ("subagent_kind".to_string(), "client-supplied".to_string()), ( SANDBOX_MODE_KEY.to_string(), @@ -611,6 +614,7 @@ fn turn_metadata_state_ignores_client_reserved_metadata_before_start() { 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(ROOT_TURN_ID_KEY).is_none()); assert!(json.get("subagent_kind").is_none()); assert_eq!(json[SANDBOX_MODE_KEY].as_str(), Some("read-only")); assert_eq!(json[AUTO_REVIEW_ENABLED_KEY].as_bool(), Some(false)); @@ -651,6 +655,7 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields( "sdk".to_string(), )])); state.set_parent_turn_id("parent-turn-a".to_string()); + state.set_root_turn_id("root-turn-a".to_string()); state.set_responsesapi_client_metadata(HashMap::from([ ( "codex_security_surface".to_string(), @@ -688,6 +693,7 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields( "client-supplied".to_string(), ), ("parent_turn_id".to_string(), "client-supplied".to_string()), + (ROOT_TURN_ID_KEY.to_string(), "client-supplied".to_string()), ("subagent_kind".to_string(), "client-supplied".to_string()), ( LEGACY_CODE_MODE_TOOL_NAMES_KEY.to_string(), @@ -773,6 +779,7 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields( Some("55555555-5555-4555-8555-555555555555") ); assert_eq!(json["parent_turn_id"].as_str(), Some("parent-turn-a")); + assert_eq!(json[ROOT_TURN_ID_KEY].as_str(), Some("root-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")); @@ -787,6 +794,10 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields( let model_request_json: Value = serde_json::from_str(&model_request_header).expect("model request json"); assert_eq!(model_request_json["request_kind"].as_str(), Some("turn")); + assert_eq!( + model_request_json[ROOT_TURN_ID_KEY].as_str(), + Some("root-turn-a") + ); assert_eq!( model_request_json["thread_source"].as_str(), Some("automation") @@ -838,6 +849,7 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields( assert!(meta.get(LEGACY_CODE_MODE_TOOL_NAMES_KEY).is_none()); assert!(meta.get(TOOL_NAMESPACES_INFO_KEY).is_none()); assert!(meta.get(PARENT_TURN_ID_KEY).is_none()); + assert!(meta.get(ROOT_TURN_ID_KEY).is_none()); assert!(meta.get(WINDOW_ID_KEY).is_none()); assert!(meta.get("codex_security_surface").is_none()); assert_eq!(state.workspace_kind().as_deref(), Some("projectless")); diff --git a/codex-rs/core/tests/common/responses.rs b/codex-rs/core/tests/common/responses.rs index 1103cbfd86..231f7f4166 100644 --- a/codex-rs/core/tests/common/responses.rs +++ b/codex-rs/core/tests/common/responses.rs @@ -81,13 +81,20 @@ impl ResponseMock { } pub fn assert_parent_turn(body: &Value, expected: Option<&str>) -> Result<()> { + assert_turn_id(body, "parent_turn_id", expected) +} + +pub fn assert_root_turn(body: &Value, expected: Option<&str>) -> Result<()> { + assert_turn_id(body, "root_turn_id", expected) +} + +fn assert_turn_id(body: &Value, key: &str, 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(()) diff --git a/codex-rs/core/tests/suite/git_enrichment.rs b/codex-rs/core/tests/suite/git_enrichment.rs index d6d8266759..7d1c726b48 100644 --- a/codex-rs/core/tests/suite/git_enrichment.rs +++ b/codex-rs/core/tests/suite/git_enrichment.rs @@ -15,6 +15,7 @@ use codex_protocol::protocol::Op; use codex_protocol::protocol::ThreadSource; use codex_protocol::user_input::UserInput; use core_test_support::PathBufExt; +use core_test_support::responses::assert_root_turn; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; use core_test_support::responses::ev_function_call; @@ -30,6 +31,7 @@ use std::path::Path; use std::process::Command; use std::time::Duration; use tempfile::TempDir; +use test_case::test_case; const ORIGIN_URL: &str = "https://example.invalid/cxa5426/repo.git"; @@ -230,8 +232,12 @@ async fn guardian_prewarm_and_review_skip_redundant_git_enrichment() -> Result<( Ok(()) } +#[test_case("system"; "system background thread")] +#[test_case("ambient_background"; "ambient background thread")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn ephemeral_system_thread_prewarm_skips_and_turn_observes_fresh_state() -> Result<()> { +async fn ephemeral_system_thread_prewarm_skips_and_turn_observes_fresh_state( + thread_source: &str, +) -> Result<()> { skip_if_no_network!(Ok(())); let (repo, head) = create_git_repo()?; @@ -274,7 +280,7 @@ async fn ephemeral_system_thread_prewarm_skips_and_turn_observes_fresh_state() - let system_thread = test .thread_manager .start_thread(StartThreadOptions { - thread_source: Some(ThreadSource::Feature("system".to_string())), + thread_source: Some(ThreadSource::Feature(thread_source.to_string())), ..StartThreadOptions::new(config) }) .await?; @@ -310,6 +316,7 @@ async fn ephemeral_system_thread_prewarm_skips_and_turn_observes_fresh_state() - .and_then(|connection| connection.get(2)) .context("system turn follow-up request")? .body_json(); + assert_root_turn(&turn, /*expected*/ None)?; assert_eq!( turn_metadata(&turn)?["workspaces"], expected_workspace(repo.path(), &head, /*has_changes*/ true) diff --git a/codex-rs/core/tests/suite/guardian_review.rs b/codex-rs/core/tests/suite/guardian_review.rs index e835cbd35f..7d246b4380 100644 --- a/codex-rs/core/tests/suite/guardian_review.rs +++ b/codex-rs/core/tests/suite/guardian_review.rs @@ -27,6 +27,8 @@ use codex_protocol::protocol::SandboxPolicy; use codex_protocol::protocol::TurnAbortReason; use codex_protocol::user_input::UserInput; use core_test_support::fs_wait; +use core_test_support::responses::assert_parent_turn; +use core_test_support::responses::assert_root_turn; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; use core_test_support::responses::ev_custom_tool_call; @@ -209,6 +211,9 @@ async fn guardian_session_prewarms_and_is_reused_for_first_review( "startup prewarm must not request the external clock" ); let prewarm_requests = [first.body_json(), second.body_json()]; + for prewarm in &prewarm_requests { + assert_root_turn(prewarm, /*expected*/ None)?; + } let guardian_prewarm = prewarm_requests .iter() .find(|request| { @@ -267,6 +272,15 @@ async fn guardian_session_prewarms_and_is_reused_for_first_review( ) .await? .body_json(); + let parent_request = server.connections()[2][0].body_json(); + let parent_turn_id = parent_request["client_metadata"]["turn_id"] + .as_str() + .expect("reviewed parent turn id"); + assert_parent_turn(&parent_request, /*expected*/ None)?; + assert_parent_turn(&guardian_review, Some(parent_turn_id))?; + for request in [&parent_request, &guardian_review] { + assert_root_turn(request, Some(parent_turn_id))?; + } assert_eq!( guardian_review["client_metadata"]["x-openai-subagent"].as_str(), Some("guardian") @@ -466,6 +480,14 @@ async fn guardian_session_is_reused_for_consecutive_tool_reviews_without_prewarm assert_eq!(guardian_requests.len(), 2); let first_guardian_request = guardian_requests[0].body_json(); let second_guardian_request = guardian_requests[1].body_json(); + let parent_request = responses.requests()[0].body_json(); + let parent_turn_id = parent_request["client_metadata"]["turn_id"] + .as_str() + .expect("reviewed parent turn id"); + for request in [&first_guardian_request, &second_guardian_request] { + assert_parent_turn(request, Some(parent_turn_id))?; + assert_root_turn(request, Some(parent_turn_id))?; + } let first_guardian_thread_id = first_guardian_request["client_metadata"]["thread_id"] .as_str() .expect("first Guardian review should have a thread id"); diff --git a/codex-rs/core/tests/suite/mcp_turn_metadata.rs b/codex-rs/core/tests/suite/mcp_turn_metadata.rs index e1cd35e240..cb47f28e63 100644 --- a/codex-rs/core/tests/suite/mcp_turn_metadata.rs +++ b/codex-rs/core/tests/suite/mcp_turn_metadata.rs @@ -33,6 +33,7 @@ use core_test_support::apps_test_server::SEARCH_CALENDAR_LIST_TOOL; use core_test_support::apps_test_server::SEARCH_CALENDAR_NAMESPACE; use core_test_support::apps_test_server::recorded_apps_tool_call_by_call_id; use core_test_support::apps_test_server::search_capable_apps_builder; +use core_test_support::responses::assert_root_turn; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; use core_test_support::responses::ev_function_call; @@ -309,8 +310,24 @@ async fn approved_mcp_tool_call_metadata_records_prior_user_input_request( }) .await; - assert_eq!(mock.requests().len(), 2 + usize::from(strict_auto_review)); + let response_requests = mock.requests(); + assert_eq!(response_requests.len(), 2 + usize::from(strict_auto_review)); + let response_body = response_requests[0].body_json(); + let turn_id = response_body["client_metadata"]["turn_id"] + .as_str() + .expect("Responses request turn id"); + assert_root_turn(&response_body, Some(turn_id))?; let apps_tool_call = recorded_apps_tool_call_by_call_id(&server, call_id).await; + let mcp_turn_metadata = apps_tool_call + .pointer("/params/_meta/x-codex-turn-metadata") + .expect("MCP tools/call turn metadata"); + assert_eq!( + ( + mcp_turn_metadata.get("root_turn_id"), + mcp_turn_metadata.get("parent_turn_id"), + ), + (None, None) + ); assert_eq!( apps_tool_call.pointer("/params/_meta/callId"), diff --git a/codex-rs/core/tests/suite/multi_agent_resume.rs b/codex-rs/core/tests/suite/multi_agent_resume.rs index e0438e9d24..1016d5a18c 100644 --- a/codex-rs/core/tests/suite/multi_agent_resume.rs +++ b/codex-rs/core/tests/suite/multi_agent_resume.rs @@ -9,6 +9,7 @@ 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::assert_root_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; @@ -264,6 +265,7 @@ async fn cold_root_resume_restores_agent_identity_and_role_on_followup() -> Resu client_user_message_id: None, trace: None, parent_turn_id: Some("spoofed-parent-turn".to_string()), + root_turn_id: Some("spoofed-foreign-root-turn".to_string()), }) .await?; wait_for_event(&initial.codex, |event| { @@ -551,6 +553,16 @@ async fn cold_root_resume_restores_agent_identity_and_role_on_followup() -> Resu } assert_parent_turn(body, parent_turn)?; } + for (body, root_turn) in [ + (&initial_root, initial_parent), + (&queue_root, queue_parent), + (&followup_root, followup_parent), + (&initial_child, initial_parent), + (&followup_child, followup_parent), + (&grandchild, initial_parent), + ] { + assert_root_turn(body, Some(root_turn))?; + } let reloaded_worker_config = reloaded_worker.config_snapshot().await; let reloaded_worker_role_config = ( reloaded_worker_config.model, diff --git a/codex-rs/core/tests/suite/pending_input.rs b/codex-rs/core/tests/suite/pending_input.rs index e0dd9ffc57..4f6fd96778 100644 --- a/codex-rs/core/tests/suite/pending_input.rs +++ b/codex-rs/core/tests/suite/pending_input.rs @@ -81,6 +81,9 @@ async fn idle_response_items_include_pending_mailbox_in_first_request() -> anyho wait_for_turn_complete(test.codex.as_ref()).await; let request = response.single_request(); + let request_body = request.body_json(); + responses::assert_root_turn(&request_body, /*expected*/ None)?; + responses::assert_parent_turn(&request_body, /*expected*/ None)?; let user_messages = request.message_input_texts("user"); assert!( user_messages @@ -167,6 +170,11 @@ async fn assert_idle_user_input_reaches_the_first_model_request( wait_for_turn_complete(test.codex.as_ref()).await; let request = response.single_request(); + let request_body = request.body_json(); + let turn_id = request_body["client_metadata"]["turn_id"] + .as_str() + .expect("idle user turn id"); + responses::assert_root_turn(&request_body, Some(turn_id))?; assert!( request .message_input_texts("user") diff --git a/codex-rs/core/tests/suite/review.rs b/codex-rs/core/tests/suite/review.rs index a75a2d14f0..86d55e4c87 100644 --- a/codex-rs/core/tests/suite/review.rs +++ b/codex-rs/core/tests/suite/review.rs @@ -226,7 +226,10 @@ 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())) + let request_body = request.body_json(); + responses::assert_root_turn(&request_body, Some(review_turn_id.as_str())) + .expect("review request root turn metadata"); + responses::assert_parent_turn(&request_body, 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 diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_api_auth_prompt_cache_key_request_diff.snap b/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_api_auth_prompt_cache_key_request_diff.snap index 9628f7b293..9518f2912e 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_api_auth_prompt_cache_key_request_diff.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_api_auth_prompt_cache_key_request_diff.snap @@ -7,11 +7,12 @@ Scenario: After five varied API-key-auth turns, remote manual compaction omits s --- Last Normal /responses Request +++ Remote /responses/compact Request - "client_metadata": { +- "root_turn_id": "", - "session_id": "", - "thread_id": "", - "turn_id": "", - "x-codex-installation-id": "", -- "x-codex-turn-metadata": "{\"installation_id\":\"\",\"session_id\":\"\",\"thread_id\":\"\",\"turn_id\":\"\",\"window_id\":\":0\",\"request_kind\":\"turn\",\"sandbox\":\"\",\"sandbox_mode\":\"read-only\",\"auto_review_enabled\":false,\"turn_started_at_unix_ms\":}", +- "x-codex-turn-metadata": "{\"installation_id\":\"\",\"session_id\":\"\",\"thread_id\":\"\",\"turn_id\":\"\",\"window_id\":\":0\",\"request_kind\":\"turn\",\"root_turn_id\":\"\",\"sandbox\":\"\",\"sandbox_mode\":\"read-only\",\"auto_review_enabled\":false,\"turn_started_at_unix_ms\":}", - "x-codex-window-id": ":0" - }, - "include": [ diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_chatgpt_auth_service_tier_prompt_cache_key_request_diff.snap b/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_chatgpt_auth_service_tier_prompt_cache_key_request_diff.snap index 9aefd7520c..97dd35e113 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_chatgpt_auth_service_tier_prompt_cache_key_request_diff.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_chatgpt_auth_service_tier_prompt_cache_key_request_diff.snap @@ -7,11 +7,12 @@ Scenario: After five varied ChatGPT-auth turns, remote manual compaction reuses --- Last Normal /responses Request +++ Remote /responses/compact Request - "client_metadata": { +- "root_turn_id": "", - "session_id": "", - "thread_id": "", - "turn_id": "", - "x-codex-installation-id": "", -- "x-codex-turn-metadata": "{\"installation_id\":\"\",\"session_id\":\"\",\"thread_id\":\"\",\"turn_id\":\"\",\"window_id\":\":0\",\"request_kind\":\"turn\",\"sandbox\":\"\",\"sandbox_mode\":\"read-only\",\"auto_review_enabled\":false,\"turn_started_at_unix_ms\":}", +- "x-codex-turn-metadata": "{\"installation_id\":\"\",\"session_id\":\"\",\"thread_id\":\"\",\"turn_id\":\"\",\"window_id\":\":0\",\"request_kind\":\"turn\",\"root_turn_id\":\"\",\"sandbox\":\"\",\"sandbox_mode\":\"read-only\",\"auto_review_enabled\":false,\"turn_started_at_unix_ms\":}", - "x-codex-window-id": ":0" - }, - "include": [ diff --git a/codex-rs/core/tests/suite/subagent_notifications.rs b/codex-rs/core/tests/suite/subagent_notifications.rs index 246681dbbd..fe02afc684 100644 --- a/codex-rs/core/tests/suite/subagent_notifications.rs +++ b/codex-rs/core/tests/suite/subagent_notifications.rs @@ -20,6 +20,7 @@ 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::assert_root_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; @@ -971,6 +972,8 @@ async fn spawned_child_receives_forked_parent_context( .expect("legacy spawn parent turn id"); assert_parent_turn(&parent_body, /*expected*/ None)?; assert_parent_turn(&child_body, Some(original_parent_turn_id))?; + assert_root_turn(&parent_body, Some(original_parent_turn_id))?; + assert_root_turn(&child_body, Some(original_parent_turn_id))?; assert_eq!( ( child_body["model"].clone(), @@ -1034,6 +1037,8 @@ async fn spawned_child_receives_forked_parent_context( 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))?; + assert_root_turn(&followup_parent_body, Some(followup_parent_turn_id))?; + assert_root_turn(&reused_child_body, Some(followup_parent_turn_id))?; Ok(()) } diff --git a/codex-rs/core/tests/suite/user_message_admission.rs b/codex-rs/core/tests/suite/user_message_admission.rs index 188acafba9..289476b7e2 100644 --- a/codex-rs/core/tests/suite/user_message_admission.rs +++ b/codex-rs/core/tests/suite/user_message_admission.rs @@ -93,9 +93,10 @@ async fn user_message_admission_starts_turn_for_empty_input() { } } - let user_input_groups = response_mock - .single_request() - .message_input_text_groups("user"); + let request = response_mock.single_request(); + responses::assert_root_turn(&request.body_json(), /*expected*/ None) + .expect("synthetic clock wake must not become a root"); + let user_input_groups = request.message_input_text_groups("user"); assert_eq!(user_input_groups.len(), 1); assert_eq!(user_input_groups[0].len(), 1); assert!(user_input_groups[0][0].starts_with("")); diff --git a/codex-rs/mcp-server/src/codex_tool_runner.rs b/codex-rs/mcp-server/src/codex_tool_runner.rs index e056e9b5c5..1dadc465fb 100644 --- a/codex-rs/mcp-server/src/codex_tool_runner.rs +++ b/codex-rs/mcp-server/src/codex_tool_runner.rs @@ -114,6 +114,7 @@ pub async fn run_codex_tool_session( client_user_message_id: None, trace: None, parent_turn_id: None, + root_turn_id: None, }; if let Err(e) = thread.submit_with_id(submission).await { @@ -159,6 +160,7 @@ pub async fn run_codex_tool_session_reply( client_user_message_id: None, trace: None, parent_turn_id: None, + root_turn_id: None, }) .await { diff --git a/codex-rs/mcp-server/src/message_processor.rs b/codex-rs/mcp-server/src/message_processor.rs index eb5cec4e81..f8242a6555 100644 --- a/codex-rs/mcp-server/src/message_processor.rs +++ b/codex-rs/mcp-server/src/message_processor.rs @@ -565,6 +565,7 @@ impl MessageProcessor { client_user_message_id: None, trace: None, parent_turn_id: None, + root_turn_id: None, }) .await { diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 64e85a6452..2bd7a8ab97 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -184,6 +184,8 @@ pub struct Submission { pub trace: Option, /// Core-provided ID of the parent turn that directly initiated this submission. pub parent_turn_id: Option, + /// Core-provided ID of the top-level turn that causally initiated this submission. + pub root_turn_id: Option, } #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]