diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 74a8bc224b..76a7d19b8b 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -70,6 +70,7 @@ pub(crate) struct SpawnAgentOptions { pub(crate) fork_parent_spawn_call_id: Option, pub(crate) fork_mode: Option, pub(crate) parent_thread_id: Option, + pub(crate) parent_turn_id: Option, pub(crate) environments: Option>, } @@ -143,11 +144,12 @@ impl AgentControl { &self, agent_id: ThreadId, input: Vec, + parent_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) + 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, input: Vec, + parent_turn_id: Option, ) -> CodexResult { 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, ) -> CodexResult { 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, communication: InterAgentCommunication, context: AgentCommunicationContext, + parent_turn_id: Option, ) -> CodexResult { - 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, communication: InterAgentCommunication, context: AgentCommunicationContext, + parent_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 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; } diff --git a/codex-rs/core/src/agent/control/legacy.rs b/codex-rs/core/src/agent/control/legacy.rs index 3fb1717fa6..f5f16f6851 100644 --- a/codex-rs/core/src/agent/control/legacy.rs +++ b/codex-rs/core/src/agent/control/legacy.rs @@ -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); diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index 9e994fa793..a81d896d49 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -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?; } diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index ab732ec3c2..d5b62aca49 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -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"); diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index 2caeb1332d..cc510f2c05 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -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(); diff --git a/codex-rs/core/src/codex_delegate_tests.rs b/codex-rs/core/src/codex_delegate_tests.rs index d1bebf5620..24d2ab7901 100644 --- a/codex-rs/core/src/codex_delegate_tests.rs +++ b/codex-rs/core/src/codex_delegate_tests.rs @@ -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 diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index c0423c87e3..63f2c69aff 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -275,7 +275,9 @@ impl CodexThread { op: Op, trace: Option, ) -> CodexResult { - 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 } diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 403b599082..b7b4939712 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -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 { diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index acd0fe9c86..03b14f91d7 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/codex-rs/core/src/guardian/tests.rs @@ -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") diff --git a/codex-rs/core/src/responses_metadata.rs b/codex-rs/core/src/responses_metadata.rs index 794b50930e..5476c95ab6 100644 --- a/codex-rs/core/src/responses_metadata.rs +++ b/codex-rs/core/src/responses_metadata.rs @@ -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, pub(crate) forked_from_thread_id: Option, pub(crate) parent_thread_id: Option, + pub(crate) parent_turn_id: Option, pub(crate) subagent_header: Option, pub(crate) subagent_kind: Option, pub(crate) thread_source: Option, @@ -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, #[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>, diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 740f1321a5..28fdadf60d 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -83,8 +83,9 @@ pub async fn user_input_or_turn( sub_id: String, op: Op, client_user_message_id: Option, + parent_turn_id: Option, ) { - 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, + parent_turn_id: Option, ) { 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, sub_id: String, communication: InterAgentCommunication, + parent_turn_id: Option, ) { 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 { diff --git a/codex-rs/core/src/session/inject.rs b/codex-rs/core/src/session/inject.rs index 3e086964c3..e772513d00 100644 --- a/codex-rs/core/src/session/inject.rs +++ b/codex-rs/core/src/session/inject.rs @@ -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(()) } diff --git a/codex-rs/core/src/session/input_queue.rs b/codex-rs/core/src/session/input_queue.rs index 3992f44d4a..34f6f778d7 100644 --- a/codex-rs/core/src/session/input_queue.rs +++ b/codex-rs/core/src/session/input_queue.rs @@ -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, - mailbox_pending_mails: Mutex>, + mailbox_pending_mails: Mutex>, +} + +struct PendingMailboxCommunication { + communication: InterAgentCommunication, + parent_turn_id: Option, } impl InputQueue { @@ -72,11 +77,15 @@ impl InputQueue { pub(crate) async fn enqueue_mailbox_communication( &self, communication: InterAgentCommunication, + parent_turn_id: Option, ) { 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 { - self.mailbox_pending_mails + pub(crate) async fn drain_mailbox_input_items(&self) -> (Vec, Option) { + let pending_mails = self + .mailbox_pending_mails .lock() .await .drain(..) - .map(TurnInput::InterAgentCommunication) - .collect() + .collect::>(); + 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>, - ) -> Vec { + ) -> (Vec, Option) { 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); } diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index df798261fb..efe735c1a8 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -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 { - 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, + parent_turn_id: Option, ) -> CodexResult { 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}"); diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index cb3f85eb0d..943d9eb36e 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -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::::new(), + (Vec::::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::::new(), + (Vec::::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::::new(), + (Vec::::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)], ); } diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 6eb6ae4ba6..57691ab81a 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -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() }; diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 54e561de4f..e12e2bfbe1 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -69,6 +69,11 @@ const TASK_COMPACT_METRIC: &str = "codex.task.compact"; pub(crate) type SessionTaskResult = CodexResult>; +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( @@ -327,6 +333,7 @@ impl Session { turn_context: Arc, input: Vec, task: T, + mailbox_parent_provenance: MailboxParentProvenance, ) { let task: Arc = 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, reason: TurnAbortReason) { diff --git a/codex-rs/core/src/test_support.rs b/codex-rs/core/src/test_support.rs index 15bc24cef8..273311e9d6 100644 --- a/codex-rs/core/src/test_support.rs +++ b/codex-rs/core/src/test_support.rs @@ -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 { &TEST_MODEL_PRESETS } diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 8514f47b11..676033eecd 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -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 { + pub(crate) async fn send_op( + &self, + thread_id: ThreadId, + op: Op, + parent_turn_id: Option, + ) -> CodexResult { 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. 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 649eaf1ce0..58830694a8 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 @@ -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 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 a0cbee7f4c..0424539817 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs @@ -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()), }, )) 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 5e5884d27b..1a58b77c56 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -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 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 62defe65c4..0ffe29562c 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 @@ -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?; 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 67f8057e43..2fc0b1fa64 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 @@ -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()), }, ), diff --git a/codex-rs/core/src/turn_metadata.rs b/codex-rs/core/src/turn_metadata.rs index 7f7abd07b7..3a3d9d2083 100644 --- a/codex-rs/core/src/turn_metadata.rs +++ b/codex-rs/core/src/turn_metadata.rs @@ -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, parent_thread_id: Option, + parent_turn_id: OnceLock, subagent_header: Option, subagent_kind: Option, thread_source: Option, @@ -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, @@ -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(), diff --git a/codex-rs/core/src/turn_metadata_tests.rs b/codex-rs/core/src/turn_metadata_tests.rs index 8f56d7db72..cc67515287 100644 --- a/codex-rs/core/src/turn_metadata_tests.rs +++ b/codex-rs/core/src/turn_metadata_tests.rs @@ -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")); } diff --git a/codex-rs/core/tests/common/responses.rs b/codex-rs/core/tests/common/responses.rs index 2b0c178b80..1103cbfd86 100644 --- a/codex-rs/core/tests/common/responses.rs +++ b/codex-rs/core/tests/common/responses.rs @@ -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); diff --git a/codex-rs/core/tests/suite/client_websockets.rs b/codex-rs/core/tests/suite/client_websockets.rs index 988419dec8..0ebaa23538 100755 --- a/codex-rs/core/tests/suite/client_websockets.rs +++ b/codex-rs/core/tests/suite/client_websockets.rs @@ -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() diff --git a/codex-rs/core/tests/suite/multi_agent_resume.rs b/codex-rs/core/tests/suite/multi_agent_resume.rs index 703c1bddd3..c770c5151e 100644 --- a/codex-rs/core/tests/suite/multi_agent_resume.rs +++ b/codex-rs/core/tests/suite/multi_agent_resume.rs @@ -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, diff --git a/codex-rs/core/tests/suite/review.rs b/codex-rs/core/tests/suite/review.rs index b3ba4f2597..f91114f740 100644 --- a/codex-rs/core/tests/suite/review.rs +++ b/codex-rs/core/tests/suite/review.rs @@ -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. diff --git a/codex-rs/core/tests/suite/subagent_notifications.rs b/codex-rs/core/tests/suite/subagent_notifications.rs index 4e51ed2cf0..76efa8c2dc 100644 --- a/codex-rs/core/tests/suite/subagent_notifications.rs +++ b/codex-rs/core/tests/suite/subagent_notifications.rs @@ -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(()) } diff --git a/codex-rs/mcp-server/src/codex_tool_runner.rs b/codex-rs/mcp-server/src/codex_tool_runner.rs index 4be99e9503..39dc9f9e8c 100644 --- a/codex-rs/mcp-server/src/codex_tool_runner.rs +++ b/codex-rs/mcp-server/src/codex_tool_runner.rs @@ -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 { diff --git a/codex-rs/mcp-server/src/message_processor.rs b/codex-rs/mcp-server/src/message_processor.rs index 38824ab002..3bc7f14cdd 100644 --- a/codex-rs/mcp-server/src/message_processor.rs +++ b/codex-rs/mcp-server/src/message_processor.rs @@ -560,6 +560,7 @@ impl MessageProcessor { op: codex_protocol::protocol::Op::Interrupt, client_user_message_id: None, trace: None, + parent_turn_id: None, }) .await { diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index c692d040d2..2df2081504 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -182,6 +182,8 @@ pub struct Submission { pub client_user_message_id: Option, /// Optional W3C trace carrier propagated across async submission handoffs. pub trace: Option, + /// Core-provided ID of the parent turn that directly initiated this submission. + pub parent_turn_id: Option, } #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]