github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-25 23:57:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

Track root turns across delegated Codex requests (#38232)

Browse Source 
## 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
This commit is contained in:
Eddie Chen
2026-08-12 18:11:13 +00:00
committed by copyberry
parent 9df9ff6ad9
commit 1f4ea79853
38 changed files with 470 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control.rs
+21 -3
View File
@@ -73,6 +73,7 @@ pub(crate) struct SpawnAgentOptions {
pub(crate) fork_mode: Option<SpawnAgentForkMode>,
pub(crate) parent_thread_id: Option<ThreadId>,
pub(crate) parent_turn_id: Option<String>,
pub(crate) root_turn_id: Option<String>,
pub(crate) environments: Option<Vec<TurnEnvironmentSelection>>,
}
@@ -169,11 +170,12 @@ impl AgentControl {
agent_id: ThreadId,
input: Vec<UserInput>,
parent_turn_id: Option<String>,
root_turn_id: Option<String>,
) -> CodexResult<String> {
let state = self.upgrade()?;
self.ensure_execution_capacity_for_turn_start(agent_id, /*starts_turn*/ true)
.await?;
self.send_input_after_capacity_check(agent_id, &state, input, 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<ThreadManagerState>,
input: Vec<UserInput>,
parent_turn_id: Option<String>,
root_turn_id: Option<String>,
) -> CodexResult<String> {
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<String>,
root_turn_id: Option<String>,
) -> CodexResult<String> {
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<String>,
root_turn_id: Option<String>,
) -> CodexResult<String> {
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<String>,
root_turn_id: Option<String>,
) -> CodexResult<String> {
let communication_for_log =
crate::agent_communication::logging_enabled().then(|| communication.clone());
let parent_turn_id = parent_turn_id.filter(|_| communication.trigger_turn);
let 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;
codex-rs/core/src/agent/control/legacy.rs
+12 -2
View File
@@ -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;
codex-rs/core/src/agent/control/spawn.rs
+2
View File
@@ -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?;
}
codex-rs/core/src/agent/control_tests.rs
+5
View File
@@ -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");
codex-rs/core/src/codex_delegate.rs
+3
View File
@@ -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();
codex-rs/core/src/codex_delegate_tests.rs
+1
View File
@@ -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);
codex-rs/core/src/codex_thread.rs
+6 -1
View File
@@ -294,7 +294,9 @@ impl CodexThread {
trace: Option<W3cTraceContext>,
) -> CodexResult<String> {
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
}
codex-rs/core/src/guardian/review_session.rs
+3 -1
View File
@@ -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,
codex-rs/core/src/responses_metadata.rs
+10
View File
@@ -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<ThreadId>,
pub(crate) parent_thread_id: Option<ThreadId>,
pub(crate) parent_turn_id: Option<String>,
pub(crate) root_turn_id: Option<String>,
pub(crate) subagent_header: Option<String>,
pub(crate) subagent_kind: Option<String>,
pub(crate) thread_source: Option<ThreadSource>,
@@ -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>,
codex-rs/core/src/session/handlers.rs
+31 -1
View File
@@ -88,6 +88,7 @@ pub async fn user_input_or_turn(
op: Op,
client_user_message_id: Option<String>,
parent_turn_id: Option<String>,
root_turn_id: Option<String>,
) {
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<String>,
parent_turn_id: Option<String>,
root_turn_id: Option<String>,
) -> CodexResult<UserMessageAdmission> {
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(&current_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<String>,
root_turn_id: Option<String>,
) {
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
codex-rs/core/src/session/inject.rs
+9
View File
@@ -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 {
codex-rs/core/src/session/input_queue.rs
+98 -28
View File
@@ -47,6 +47,7 @@ pub(crate) struct InputQueue {
struct PendingMailboxCommunication {
communication: InterAgentCommunication,
parent_turn_id: Option<String>,
root_turn_id: Option<String>,
_diagnostics_guard: GaugeGuard,
}
@@ -86,6 +87,7 @@ impl InputQueue {
&self,
communication: InterAgentCommunication,
parent_turn_id: Option<String>,
root_turn_id: Option<String>,
) {
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<TurnInput>, Option<String>) {
pub(crate) async fn drain_mailbox_input_items(
&self,
) -> (Vec<TurnInput>, Option<String>, Option<String>) {
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<Option<ActiveTurn>>,
) -> (Vec<TurnInput>, Option<String>) {
let (pending_input, accepts_mailbox_delivery) = {
) -> (Vec<TurnInput>, Option<String>, Option<String>) {
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);
}
codex-rs/core/src/session/mod.rs
+19 -2
View File
@@ -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<String> {
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<W3cTraceContext>,
parent_turn_id: Option<String>,
root_turn_id: Option<String>,
) -> CodexResult<String> {
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<String>,
responsesapi_client_metadata: Option<HashMap<String, String>>,
incoming_turn_metadata: Option<&TurnMetadataState>,
) -> Result<String, SteerInputError> {
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(),
codex-rs/core/src/session/review.rs
+3
View File
@@ -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(),
codex-rs/core/src/session/tests.rs
+60 -10
View File
@@ -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::<TurnInput>::new(), None),
(Vec::<TurnInput>::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::<TurnInput>::new(), None),
(Vec::<TurnInput>::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::<TurnInput>::new(), None),
(Vec::<TurnInput>::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 {
codex-rs/core/src/tasks/mod.rs
+15 -4
View File
@@ -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;
codex-rs/core/src/thread_manager.rs
+2 -1
View File
@@ -1374,6 +1374,7 @@ impl ThreadManagerState {
thread_id: ThreadId,
op: Op,
parent_turn_id: Option<String>,
root_turn_id: Option<String>,
) -> CodexResult<String> {
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
}
codex-rs/core/src/tools/handlers/multi_agents/send_input.rs
+6 -1
View File
@@ -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
codex-rs/core/src/tools/handlers/multi_agents/spawn.rs
+1
View File
@@ -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()),
},
))
codex-rs/core/src/tools/handlers/multi_agents_tests.rs
+5
View File
@@ -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;
codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs
+7 -1
View File
@@ -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?;
codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs
+1
View File
@@ -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()),
},
),
codex-rs/core/src/turn_metadata.rs
+39
View File
@@ -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<ThreadId>,
parent_thread_id: Option<ThreadId>,
parent_turn_id: OnceLock<String>,
root_turn_id: OnceLock<String>,
subagent_header: Option<String>,
subagent_kind: Option<String>,
thread_source: Option<ThreadSource>,
@@ -113,6 +115,7 @@ pub(crate) struct TurnMetadataState {
turn_started_at_unix_ms: RwLock<Option<i64>>,
responses_api_metadata: RwLock<BTreeMap<String, String>>,
responsesapi_client_metadata: RwLock<BTreeMap<String, String>>,
root_turn_ambiguous: AtomicBool,
user_input_requested_during_turn: AtomicBool,
enrichment_task: Mutex<Option<JoinHandle<()>>>,
}
@@ -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<String> {
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<String, String>,
@@ -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(),
codex-rs/core/src/turn_metadata_tests.rs
+12
View File
@@ -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"));
codex-rs/core/tests/common/responses.rs
+8 -1
View File
@@ -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(())
codex-rs/core/tests/suite/git_enrichment.rs
+9 -2
View File
@@ -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)
codex-rs/core/tests/suite/guardian_review.rs
+22
View File
@@ -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");
codex-rs/core/tests/suite/mcp_turn_metadata.rs
+18 -1
View File
@@ -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"),
codex-rs/core/tests/suite/multi_agent_resume.rs
+12
View File
@@ -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,
codex-rs/core/tests/suite/pending_input.rs
+8
View File
@@ -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")
codex-rs/core/tests/suite/review.rs
+4 -1
View File
@@ -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
codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_api_auth_prompt_cache_key_request_diff.snap
+2 -1
View File
@@ -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": "<UUID>",
- "session_id": "<UUID>",
- "thread_id": "<UUID>",
- "turn_id": "<UUID>",
- "x-codex-installation-id": "<UUID>",
- "x-codex-turn-metadata": "{\"installation_id\":\"<UUID>\",\"session_id\":\"<UUID>\",\"thread_id\":\"<UUID>\",\"turn_id\":\"<UUID>\",\"window_id\":\"<UUID>:0\",\"request_kind\":\"turn\",\"sandbox\":\"<SANDBOX>\",\"sandbox_mode\":\"read-only\",\"auto_review_enabled\":false,\"turn_started_at_unix_ms\":<UNIX_MS>}",
- "x-codex-turn-metadata": "{\"installation_id\":\"<UUID>\",\"session_id\":\"<UUID>\",\"thread_id\":\"<UUID>\",\"turn_id\":\"<UUID>\",\"window_id\":\"<UUID>:0\",\"request_kind\":\"turn\",\"root_turn_id\":\"<UUID>\",\"sandbox\":\"<SANDBOX>\",\"sandbox_mode\":\"read-only\",\"auto_review_enabled\":false,\"turn_started_at_unix_ms\":<UNIX_MS>}",
- "x-codex-window-id": "<UUID>:0"
- },
- "include": [
codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_chatgpt_auth_service_tier_prompt_cache_key_request_diff.snap
+2 -1
View File
@@ -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": "<UUID>",
- "session_id": "<UUID>",
- "thread_id": "<UUID>",
- "turn_id": "<UUID>",
- "x-codex-installation-id": "<UUID>",
- "x-codex-turn-metadata": "{\"installation_id\":\"<UUID>\",\"session_id\":\"<UUID>\",\"thread_id\":\"<UUID>\",\"turn_id\":\"<UUID>\",\"window_id\":\"<UUID>:0\",\"request_kind\":\"turn\",\"sandbox\":\"<SANDBOX>\",\"sandbox_mode\":\"read-only\",\"auto_review_enabled\":false,\"turn_started_at_unix_ms\":<UNIX_MS>}",
- "x-codex-turn-metadata": "{\"installation_id\":\"<UUID>\",\"session_id\":\"<UUID>\",\"thread_id\":\"<UUID>\",\"turn_id\":\"<UUID>\",\"window_id\":\"<UUID>:0\",\"request_kind\":\"turn\",\"root_turn_id\":\"<UUID>\",\"sandbox\":\"<SANDBOX>\",\"sandbox_mode\":\"read-only\",\"auto_review_enabled\":false,\"turn_started_at_unix_ms\":<UNIX_MS>}",
- "x-codex-window-id": "<UUID>:0"
- },
- "include": [
codex-rs/core/tests/suite/subagent_notifications.rs
+5
View File
@@ -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(())
}
codex-rs/core/tests/suite/user_message_admission.rs
+4 -3
View File
@@ -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("<environment_context>"));
codex-rs/mcp-server/src/codex_tool_runner.rs
+2
View File
@@ -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
{
codex-rs/mcp-server/src/message_processor.rs
+1
View File
@@ -565,6 +565,7 @@ impl MessageProcessor {
client_user_message_id: None,
trace: None,
parent_turn_id: None,
root_turn_id: None,
})
.await
{
codex-rs/protocol/src/protocol.rs
+2
View File
@@ -184,6 +184,8 @@ pub struct Submission {
pub trace: Option<W3cTraceContext>,
/// Core-provided ID of the parent turn that directly initiated this submission.
pub parent_turn_id: Option<String>,
/// Core-provided ID of the top-level turn that causally initiated this submission.
pub root_turn_id: Option<String>,
}
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]