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

Expose and persist turn lineage across the app server (#51415)

Browse Source 
## Why

Clients starting delegated work need to attribute descendant turns to the original turn and retain that attribution through completion and history reads.

## What changed

- Accept optional `parentTurnId` and `rootTurnId` in `turn/start`, and expose `rootTurnId` on returned turns and lifecycle notifications.
- Default new turns to their own root when none is supplied. Preserve the active turn's root when steering, ignoring incoming lineage fields.
- Carry root IDs through completion, interruption, recovery, and fork boundaries, and persist them for paginated history reads. Leave unknown roots null in older history.
- Update protocol schemas and generated TypeScript and Python bindings.

## Testing

Extend coverage for default and explicit roots, steering without changing lineage, interrupted and recovered turns, history reads after restart in both history modes, nested fork pagination, and compatibility with older database writers.

GitOrigin-RevId: f1a570a12bd11742929fd2e8b017e5d35aa07dcb
This commit is contained in:
Eddie Chen
2026-10-06 16:56:39 +00:00
committed by copyberry
parent ff9ab4aed9
commit cfc946f4e1
149 changed files with 896 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control.rs
+1 -1
View File
@@ -142,7 +142,7 @@ impl LocalAgentControl {
.start_or_steer_turn(TurnInputRequest::user_input(input).on_start(start_options))
.await
{
Ok(TurnInputSubmission::Started { turn_id }) => Ok(turn_id),
Ok(TurnInputSubmission::Started { turn_id, .. }) => Ok(turn_id),
Ok(TurnInputSubmission::Steered { .. }) => {
// MAv1 exposes an opaque `submission_id` to the model. The legacy
// `Op::UserInput` path returned a fresh ID for every steer, while the
codex-rs/core/src/agent/control/residency_tests.rs
+2
View File
@@ -194,6 +194,7 @@ async fn mark_thread_completed(thread: &CodexThread) {
.send_event(
turn.as_ref(),
EventMsg::TurnComplete(TurnCompleteEvent {
root_turn_id: None,
turn_id: turn.sub_id.clone(),
started_at: None,
last_agent_message: Some("done".to_string()),
@@ -214,6 +215,7 @@ async fn mark_thread_interrupted(thread: &CodexThread) {
.send_event(
turn.as_ref(),
EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id: None,
turn_id: Some(turn.sub_id.clone()),
started_at: None,
reason: TurnAbortReason::Interrupted,
codex-rs/core/src/agent/control_tests.rs
+4
View File
@@ -590,6 +590,7 @@ async fn on_event_updates_status_from_task_complete() {
),
] {
let status = agent_status_from_event(&EventMsg::TurnComplete(TurnCompleteEvent {
root_turn_id: None,
turn_id: "turn-1".to_string(),
started_at: None,
last_agent_message: Some("done".to_string()),
@@ -617,6 +618,7 @@ async fn on_event_updates_status_from_error() {
#[tokio::test]
async fn on_event_updates_status_from_turn_aborted() {
let status = agent_status_from_event(&EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id: None,
turn_id: Some("turn-1".to_string()),
started_at: None,
reason: TurnAbortReason::Interrupted,
@@ -3948,6 +3950,7 @@ async fn multi_agent_v2_completion_ignores_dead_direct_parent() {
.send_event(
tester_turn.as_ref(),
EventMsg::TurnComplete(TurnCompleteEvent {
root_turn_id: None,
turn_id: tester_turn.sub_id.clone(),
started_at: None,
last_agent_message: Some("done".to_string()),
@@ -4030,6 +4033,7 @@ async fn multi_agent_v2_completion_queues_message_for_direct_parent() {
.send_event(
tester_turn.as_ref(),
EventMsg::TurnComplete(TurnCompleteEvent {
root_turn_id: None,
turn_id: tester_turn.sub_id.clone(),
started_at: None,
last_agent_message: Some("done".to_string()),
codex-rs/core/src/codex_delegate_tests.rs
+1
View File
@@ -51,6 +51,7 @@ async fn forward_events_filters_private_events_before_blocked_send_is_cancelled(
.send(Event {
id: "full".to_string(),
msg: EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id: None,
turn_id: Some("turn-1".to_string()),
started_at: None,
reason: TurnAbortReason::Interrupted,
codex-rs/core/src/codex_thread.rs
+17 -7
View File
@@ -425,9 +425,13 @@ impl CodexThread {
.submit_turn_input_with_mode(request, TurnInputMode::StartIfIdle)
.await?
{
TurnInputSubmission::Started { turn_id } => {
Ok(StartIfIdleSubmission::Started { turn_id })
}
TurnInputSubmission::Started {
turn_id,
root_turn_id,
} => Ok(StartIfIdleSubmission::Started {
turn_id,
root_turn_id,
}),
TurnInputSubmission::NotSubmitted { reason } => {
Ok(StartIfIdleSubmission::NotSubmitted { reason })
}
@@ -489,9 +493,13 @@ impl CodexThread {
)
.await?
{
TurnInputSubmission::Started { turn_id } => {
Ok(StartIfIdleSubmission::Started { turn_id })
}
TurnInputSubmission::Started {
turn_id,
root_turn_id,
} => Ok(StartIfIdleSubmission::Started {
turn_id,
root_turn_id,
}),
TurnInputSubmission::NotSubmitted { reason } => {
Ok(StartIfIdleSubmission::NotSubmitted { reason })
}
@@ -554,7 +562,9 @@ impl CodexThread {
.submit_turn_input_with_mode(request, TurnInputMode::Steer { expected_turn_id })
.await?
{
TurnInputSubmission::Steered { turn_id } => Ok(SteerSubmission::Steered { turn_id }),
TurnInputSubmission::Steered { turn_id, .. } => {
Ok(SteerSubmission::Steered { turn_id })
}
TurnInputSubmission::NotSubmitted { reason } => {
Ok(SteerSubmission::NotSubmitted { reason })
}
codex-rs/core/src/guardian/review_session_tests.rs
+8 -1
View File
@@ -107,6 +107,7 @@ async fn run_review_preserves_evidence_during_parent_compaction() {
let text = serde_json::to_string(&content).unwrap();
reply
.send(Ok(TurnInputSubmission::Started {
root_turn_id: id.clone(),
turn_id: id.clone(),
}))
.unwrap();
@@ -179,6 +180,7 @@ fn turn_complete_event(
Event {
id: turn_id.to_string(),
msg: EventMsg::TurnComplete(TurnCompleteEvent {
root_turn_id: None,
turn_id: turn_id.to_string(),
started_at: None,
last_agent_message: last_agent_message.map(str::to_string),
@@ -194,6 +196,7 @@ fn turn_aborted_event(turn_id: &str) -> Event {
Event {
id: turn_id.to_string(),
msg: EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id: None,
turn_id: Some(turn_id.to_string()),
started_at: None,
reason: TurnAbortReason::Interrupted,
@@ -728,6 +731,7 @@ async fn run_review_on_reused_session_waits_for_submitted_turn() {
};
reply
.send(Ok(TurnInputSubmission::Started {
root_turn_id: id.clone(),
turn_id: id.clone(),
}))
.expect("reply to guardian submission");
@@ -791,7 +795,10 @@ async fn run_review_removes_trunk_when_event_stream_is_broken() {
panic!("expected turn-input submission");
};
reply
.send(Ok(TurnInputSubmission::Started { turn_id: id }))
.send(Ok(TurnInputSubmission::Started {
root_turn_id: id.clone(),
turn_id: id,
}))
.expect("reply to guardian submission");
drop(tx_event);
codex-rs/core/src/guardian/tests.rs
+1
View File
@@ -2971,6 +2971,7 @@ async fn guardian_reused_trunk_ignores_stale_prior_turn_completion() -> anyhow::
.send_trunk_event_raw_for_test(Event {
id: "stale-turn".to_string(),
msg: EventMsg::TurnComplete(TurnCompleteEvent {
root_turn_id: None,
turn_id: "stale-turn".to_string(),
started_at: None,
last_agent_message: Some(
codex-rs/core/src/realtime_conversation/transcript_tail.rs
+1
View File
@@ -73,6 +73,7 @@ pub(super) async fn record(session: &Session, text: String) -> std::io::Result<(
if recording_turn {
events.push(EventMsg::TurnComplete(TurnCompleteEvent {
turn_id: turn.sub_id.clone(),
root_turn_id: Some(turn.sub_id.clone()),
last_agent_message: None,
error: None,
started_at: Some(now.timestamp()),
codex-rs/core/src/realtime_history_tests.rs
+2
View File
@@ -140,6 +140,7 @@ fn interrupted_turn_is_not_associated_with_a_new_voice_session(
},
));
let aborted = EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id: None,
turn_id: aborted_turn_id.map(str::to_string),
reason: TurnAbortReason::Interrupted,
error: None,
@@ -175,6 +176,7 @@ fn interrupted_turn_is_not_associated_with_a_new_voice_session(
fn interrupted_turn_keeps_its_existing_voice_session_for_late_artifacts() {
let mut state = started_state();
state.observe(&EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id: None,
turn_id: Some("turn-1".to_string()),
reason: TurnAbortReason::Interrupted,
error: None,
codex-rs/core/src/session/mod.rs
+1 -6
View File
@@ -2266,12 +2266,7 @@ impl Session {
let event = TurnStartedEvent {
turn_attribution: (task_kind == TaskKind::Regular).then(|| turn_context.attribution()),
turn_id: turn_context.sub_id.clone(),
root_turn_id: Some(
turn_context
.turn_metadata_state
.root_turn_id()
.unwrap_or_else(|| turn_context.sub_id.clone()),
),
root_turn_id: Some(turn_context.root_turn_id()),
trace_id: turn_context.trace_id.clone(),
started_at: turn_context.turn_timing_state.started_at_unix_secs().await,
model_context_window: turn_context.model_context_window(),
codex-rs/core/src/session/rollout_reconstruction_tests.rs
+24
View File
@@ -284,6 +284,7 @@ fn completed_user_turn_rollout(
rollout_items.extend(items);
rollout_items.push(RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id,
last_agent_message: None,
error: None,
@@ -547,6 +548,7 @@ async fn record_initial_history_resumed_hydrates_previous_turn_settings_from_lif
RolloutItem::TurnContext(previous_context_item),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id,
last_agent_message: None,
error: None,
@@ -638,6 +640,7 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_com
RolloutItem::ResponseItem(turn_one_assistant.clone().into()),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: first_turn_id,
started_at: None,
last_agent_message: None,
@@ -681,6 +684,7 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_com
RolloutItem::ResponseItem(turn_two_assistant.into()),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: rolled_back_turn_id,
started_at: None,
last_agent_message: None,
@@ -768,6 +772,7 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_inc
RolloutItem::ResponseItem(turn_one_assistant.clone().into()),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: first_turn_id,
started_at: None,
last_agent_message: None,
@@ -873,6 +878,7 @@ async fn reconstruct_history_rollback_skips_non_user_turns_for_history_and_metad
RolloutItem::ResponseItem(turn_one_assistant.clone().into()),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: first_turn_id,
started_at: None,
last_agent_message: None,
@@ -907,6 +913,7 @@ async fn reconstruct_history_rollback_skips_non_user_turns_for_history_and_metad
RolloutItem::ResponseItem(turn_two_assistant.into()),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: second_turn_id,
started_at: None,
last_agent_message: None,
@@ -935,6 +942,7 @@ async fn reconstruct_history_rollback_skips_non_user_turns_for_history_and_metad
}),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: standalone_turn_id,
started_at: None,
last_agent_message: None,
@@ -1017,6 +1025,7 @@ async fn reconstruct_history_rollback_counts_inter_agent_assistant_turns() {
RolloutItem::ResponseItem(assistant_message("turn 1 assistant").into()),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: first_turn_id,
started_at: None,
last_agent_message: None,
@@ -1042,6 +1051,7 @@ async fn reconstruct_history_rollback_counts_inter_agent_assistant_turns() {
RolloutItem::ResponseItem(assistant_reply.into()),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: assistant_turn_id,
started_at: None,
last_agent_message: None,
@@ -1119,6 +1129,7 @@ async fn reconstruct_history_rollback_clears_history_and_metadata_when_exceeding
RolloutItem::ResponseItem(assistant_message("only assistant").into()),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: only_turn_id,
started_at: None,
last_agent_message: None,
@@ -1177,6 +1188,7 @@ async fn record_initial_history_resumed_rollback_skips_only_user_turns() {
RolloutItem::TurnContext(previous_context_item),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: user_turn_id,
started_at: None,
last_agent_message: None,
@@ -1200,6 +1212,7 @@ async fn record_initial_history_resumed_rollback_skips_only_user_turns() {
)),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: standalone_turn_id,
started_at: None,
last_agent_message: None,
@@ -1262,6 +1275,7 @@ async fn record_initial_history_resumed_rollback_drops_incomplete_user_turn_comp
RolloutItem::TurnContext(previous_context_item.clone()),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: previous_turn_id,
started_at: None,
last_agent_message: None,
@@ -1640,6 +1654,7 @@ async fn bounded_replay_matches_full_replay_after_empty_turn_compactions(
});
wake_items.push(RolloutItem::EventMsg(EventMsg::TurnAborted(
codex_protocol::protocol::TurnAbortedEvent {
root_turn_id: None,
turn_id: Some(format!("wake-{window_number}")),
reason: TurnAbortReason::Interrupted,
error: None,
@@ -2095,6 +2110,7 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_clear
RolloutItem::TurnContext(current_context_item),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: current_turn_id,
started_at: None,
last_agent_message: None,
@@ -2189,6 +2205,7 @@ async fn record_initial_history_resumed_turn_context_after_compaction_reestablis
RolloutItem::TurnContext(previous_context_item),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: previous_turn_id,
started_at: None,
last_agent_message: None,
@@ -2315,6 +2332,7 @@ async fn record_initial_history_resumed_aborted_turn_without_id_clears_active_tu
RolloutItem::TurnContext(previous_context_item),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: previous_turn_id,
started_at: None,
last_agent_message: None,
@@ -2347,6 +2365,7 @@ async fn record_initial_history_resumed_aborted_turn_without_id_clears_active_tu
)),
RolloutItem::EventMsg(EventMsg::TurnAborted(
codex_protocol::protocol::TurnAbortedEvent {
root_turn_id: None,
turn_id: None,
started_at: None,
reason: TurnAbortReason::Interrupted,
@@ -2457,6 +2476,7 @@ async fn record_initial_history_resumed_unmatched_abort_preserves_active_turn_fo
RolloutItem::TurnContext(previous_context_item),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: previous_turn_id,
started_at: None,
last_agent_message: None,
@@ -2489,6 +2509,7 @@ async fn record_initial_history_resumed_unmatched_abort_preserves_active_turn_fo
)),
RolloutItem::EventMsg(EventMsg::TurnAborted(
codex_protocol::protocol::TurnAbortedEvent {
root_turn_id: None,
turn_id: Some(unmatched_abort_turn_id),
started_at: None,
reason: TurnAbortReason::Interrupted,
@@ -2500,6 +2521,7 @@ async fn record_initial_history_resumed_unmatched_abort_preserves_active_turn_fo
RolloutItem::TurnContext(current_context_item.clone()),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: current_turn_id,
started_at: None,
last_agent_message: None,
@@ -2600,6 +2622,7 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_compaction_clea
RolloutItem::TurnContext(previous_context_item),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: previous_turn_id,
started_at: None,
last_agent_message: None,
@@ -2791,6 +2814,7 @@ async fn record_initial_history_resumed_replaced_incomplete_compacted_turn_clear
RolloutItem::TurnContext(previous_context_item),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id: previous_turn_id,
started_at: None,
last_agent_message: None,
codex-rs/core/src/session/tests.rs
+4
View File
@@ -816,6 +816,7 @@ async fn interrupting_regular_turn_waiting_on_startup_prewarm_emits_turn_aborted
.expect("expected turn aborted event")
.expect("channel open");
let EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id,
turn_id,
reason,
error: _,
@@ -826,6 +827,7 @@ async fn interrupting_regular_turn_waiting_on_startup_prewarm_emits_turn_aborted
else {
panic!("expected turn aborted event");
};
assert_eq!(root_turn_id, Some(tc.root_turn_id()));
assert_eq!(turn_id, Some(tc.sub_id.clone()));
assert_eq!(reason, TurnAbortReason::Interrupted);
assert!(started_at.is_some());
@@ -4360,6 +4362,7 @@ async fn record_initial_history_forked_hydrates_previous_turn_settings() {
RolloutItem::TurnContext(previous_context_item.clone()),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
root_turn_id: None,
turn_id,
last_agent_message: None,
error: None,
@@ -6282,6 +6285,7 @@ async fn compaction_persists_resume_metadata_and_companion_records() {
if let Some(turn_id) = last_started_turn_id {
items.push(RolloutItem::EventMsg(EventMsg::TurnComplete(
TurnCompleteEvent {
root_turn_id: None,
turn_id,
last_agent_message: None,
error: None,
codex-rs/core/src/session/turn_context.rs
+7
View File
@@ -399,6 +399,13 @@ enum TurnContextBuildMode {
}
impl TurnContext {
/// Returns the resolved causal root for this admitted turn.
pub(crate) fn root_turn_id(&self) -> String {
self.turn_metadata_state
.root_turn_id()
.unwrap_or_else(|| self.sub_id.clone())
}
/// Builds a review turn with shared session grants and fresh turn-local state.
pub(super) fn for_review(
&self,
codex-rs/core/src/session/turn_input.rs
+23 -11
View File
@@ -268,8 +268,9 @@ pub(super) async fn handle(
}
};
// Link this request's trace to the accepted turn, which may have an older trace.
if let Ok(TurnInputSubmission::Started { turn_id } | TurnInputSubmission::Steered { turn_id }) =
&result
if let Ok(
TurnInputSubmission::Started { turn_id, .. } | TurnInputSubmission::Steered { turn_id, .. },
) = &result
{
tracing::Span::current().record("turn.id", turn_id);
}
@@ -306,7 +307,7 @@ pub(super) async fn handle_recovery(
/*expected_previous_turn_id*/ None,
)
.await;
if let Ok(TurnInputSubmission::Started { turn_id }) = &result {
if let Ok(TurnInputSubmission::Started { turn_id, .. }) = &result {
tracing::Span::current().record("turn.id", turn_id);
}
result
@@ -363,9 +364,12 @@ async fn start_or_steer(
)
.await
{
Ok(turn_id) => {
Ok((turn_id, root_turn_id)) => {
settings.apply_steered(session, submission_id).await?;
Ok(TurnInputSubmission::Steered { turn_id })
Ok(TurnInputSubmission::Steered {
turn_id,
root_turn_id,
})
}
Err(NotSubmittedReason::NoActiveTurn) => {
// MAv1 sends explicit input to spawned agents as part of an existing
@@ -413,10 +417,11 @@ async fn start_or_steer(
.push(pending_turn_input(session, input, &turn_context.sub_id, origin).await);
}
session
.spawn_task(turn_context, task_input, RegularTask::new())
.spawn_task(Arc::clone(&turn_context), task_input, RegularTask::new())
.await;
Ok(TurnInputSubmission::Started {
turn_id: submission_id,
root_turn_id: turn_context.root_turn_id(),
})
}
Err(reason) => Ok(TurnInputSubmission::NotSubmitted { reason }),
@@ -575,10 +580,11 @@ async fn start_if_idle(
}
}
session
.start_task(turn_context, task_input, RegularTask::new())
.start_task(Arc::clone(&turn_context), task_input, RegularTask::new())
.await;
Ok(TurnInputSubmission::Started {
turn_id: submission_id,
root_turn_id: turn_context.root_turn_id(),
})
}
@@ -626,9 +632,12 @@ async fn steer(
)
.await
{
Ok(turn_id) => {
Ok((turn_id, root_turn_id)) => {
settings.apply_steered(session, submission_id).await?;
Ok(TurnInputSubmission::Steered { turn_id })
Ok(TurnInputSubmission::Steered {
turn_id,
root_turn_id,
})
}
Err(reason) => Ok(TurnInputSubmission::NotSubmitted { reason }),
}
@@ -721,7 +730,7 @@ impl Session {
required_final_output_json_schema: Option<&Value>,
responsesapi_client_metadata: Option<HashMap<String, String>>,
origin: UserInputOrigin,
) -> Result<String, NotSubmittedReason> {
) -> Result<(String, String), NotSubmittedReason> {
let mut active = self.active_turn.lock().await;
let Some(active_turn) = active.as_mut() else {
return Err(NotSubmittedReason::NoActiveTurn);
@@ -799,7 +808,10 @@ impl Session {
pending_input,
)
.await;
Ok(active_turn_id.clone())
Ok((
active_turn_id.clone(),
active_task.turn_context.root_turn_id(),
))
}
}
codex-rs/core/src/session/turn_input_tests.rs
+2
View File
@@ -159,6 +159,7 @@ async fn steering_does_not_wait_for_realtime_history() {
assert_eq!(
submission,
TurnInputSubmission::Steered {
root_turn_id: turn_context.sub_id.clone(),
turn_id: turn_context.sub_id.clone()
}
);
@@ -922,6 +923,7 @@ async fn steer_only_enforces_expected_turn_id() {
assert_eq!(
submission,
TurnInputSubmission::Steered {
root_turn_id: turn_context.sub_id.clone(),
turn_id: turn_context.sub_id.clone()
}
);
codex-rs/core/src/tasks/mod.rs
+3
View File
@@ -820,6 +820,7 @@ impl Session {
self.emit_turn_abort_lifecycle(reason.clone(), turn_context.extension_data.as_ref())
.await;
EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id: Some(turn_context.root_turn_id()),
turn_id: Some(turn_context.sub_id.clone()),
reason,
error: None,
@@ -836,6 +837,7 @@ impl Session {
self.emit_turn_stop_lifecycle(turn_context.extension_data.as_ref())
.await;
EventMsg::TurnComplete(TurnCompleteEvent {
root_turn_id: Some(turn_context.root_turn_id()),
turn_id: turn_context.sub_id.clone(),
last_agent_message,
error,
@@ -1003,6 +1005,7 @@ impl Session {
self.emit_turn_abort_lifecycle(reason.clone(), task.turn_context.extension_data.as_ref())
.await;
let event = EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id: Some(task.turn_context.root_turn_id()),
turn_id: Some(task.turn_context.sub_id.clone()),
reason,
error,
codex-rs/core/src/tasks/mod_tests.rs
+15 -1
View File
@@ -137,7 +137,7 @@ async fn turn_completion_metrics_follow_model_switch(scenario: UsageScenario) {
.with_runtime_reader(),
)
.expect("in-memory metrics client");
let (mut session, mut turn_context, _receiver) = make_session_and_context_with_rx().await;
let (mut session, mut turn_context, receiver) = make_session_and_context_with_rx().await;
let session_telemetry = session
.services
.session_telemetry
@@ -243,7 +243,21 @@ async fn turn_completion_metrics_follow_model_switch(scenario: UsageScenario) {
} else {
Ok(None)
};
let expected_root = turn_context.root_turn_id();
session.on_task_finished(turn_context, task_result).await;
let mut terminal_root = None;
while let Ok(event) = receiver.try_recv() {
match event.msg {
codex_protocol::protocol::EventMsg::TurnComplete(event) => {
terminal_root = event.root_turn_id;
}
codex_protocol::protocol::EventMsg::TurnAborted(event) => {
terminal_root = event.root_turn_id;
}
_ => {}
}
}
assert_eq!(terminal_root, Some(expected_root));
let snapshot = metrics.snapshot().expect("runtime metrics snapshot");
let token_usage_metric = find_metric(&snapshot, TURN_TOKEN_USAGE_METRIC);
codex-rs/core/src/thread_manager.rs
+11 -1
View File
@@ -2606,6 +2606,7 @@ struct SnapshotTurnState {
ends_mid_turn: bool,
active_turn_id: Option<String>,
active_turn_started_at: Option<i64>,
active_turn_root_id: Option<String>,
active_turn_start_index: Option<usize>,
}
@@ -2626,6 +2627,7 @@ fn snapshot_turn_state(history: &InitialHistory) -> SnapshotTurnState {
ends_mid_turn: false,
active_turn_id: None,
active_turn_started_at: None,
active_turn_root_id: None,
active_turn_start_index: None,
};
}
@@ -2633,7 +2635,10 @@ fn snapshot_turn_state(history: &InitialHistory) -> SnapshotTurnState {
return SnapshotTurnState {
ends_mid_turn: true,
active_turn_id,
active_turn_started_at: active_turn_snapshot.and_then(|turn| turn.started_at),
active_turn_started_at: active_turn_snapshot
.as_ref()
.and_then(|turn| turn.started_at),
active_turn_root_id: active_turn_snapshot.and_then(|turn| turn.root_turn_id),
active_turn_start_index: builder.active_turn_start_index(),
};
}
@@ -2646,6 +2651,7 @@ fn snapshot_turn_state(history: &InitialHistory) -> SnapshotTurnState {
ends_mid_turn: false,
active_turn_id: None,
active_turn_started_at: None,
active_turn_root_id: None,
active_turn_start_index: None,
};
};
@@ -2662,6 +2668,7 @@ fn snapshot_turn_state(history: &InitialHistory) -> SnapshotTurnState {
}),
active_turn_id: None,
active_turn_started_at: None,
active_turn_root_id: None,
active_turn_start_index: None,
}
}
@@ -2689,6 +2696,7 @@ fn fork_history_from_snapshot(
append_interrupted_boundary(
history,
snapshot_state.active_turn_id,
snapshot_state.active_turn_root_id,
snapshot_state.active_turn_started_at,
interrupted_marker,
)
@@ -2705,10 +2713,12 @@ fn fork_history_from_snapshot(
fn append_interrupted_boundary(
history: InitialHistory,
turn_id: Option<String>,
root_turn_id: Option<String>,
started_at: Option<i64>,
interrupted_marker: InterruptedTurnHistoryMarker,
) -> InitialHistory {
let aborted_event = RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id,
turn_id,
reason: TurnAbortReason::Interrupted,
error: None,
codex-rs/core/src/thread_manager_tests.rs
+20
View File
@@ -718,6 +718,7 @@ fn truncates_before_requested_user_message() {
ends_mid_turn: false,
active_turn_id: None,
active_turn_started_at: None,
active_turn_root_id: None,
active_turn_start_index: None,
},
);
@@ -744,6 +745,7 @@ fn truncates_before_requested_user_message() {
ends_mid_turn: false,
active_turn_id: None,
active_turn_started_at: None,
active_turn_root_id: None,
active_turn_start_index: None,
},
);
@@ -769,6 +771,7 @@ fn out_of_range_truncation_drops_only_unfinished_suffix_mid_turn() {
ends_mid_turn: true,
active_turn_id: None,
active_turn_started_at: None,
active_turn_root_id: None,
active_turn_start_index: None,
},
);
@@ -818,6 +821,7 @@ fn out_of_range_truncation_drops_pre_user_active_turn_prefix() {
ends_mid_turn: true,
active_turn_id: Some("turn-2".to_string()),
active_turn_started_at: None,
active_turn_root_id: None,
active_turn_start_index: Some(2),
},
);
@@ -863,6 +867,7 @@ async fn ignores_session_prefix_messages_when_truncating() {
ends_mid_turn: false,
active_turn_id: None,
active_turn_started_at: None,
active_turn_root_id: None,
active_turn_start_index: None,
},
);
@@ -2772,6 +2777,7 @@ fn interrupted_fork_snapshot_appends_interrupt_boundary() {
append_interrupted_boundary(
committed_history,
/*turn_id*/ None,
/*root_turn_id*/ None,
/*started_at*/ None,
InterruptedTurnHistoryMarker::ContextualUser,
)
@@ -2782,6 +2788,7 @@ fn interrupted_fork_snapshot_appends_interrupt_boundary() {
RolloutItem::ResponseItem(user_msg("hello").into()),
RolloutItem::ResponseItem(contextual_user_interrupted_marker().into()),
RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id: None,
turn_id: None,
started_at: None,
reason: TurnAbortReason::Interrupted,
@@ -2797,6 +2804,7 @@ fn interrupted_fork_snapshot_appends_interrupt_boundary() {
append_interrupted_boundary(
InitialHistory::New,
/*turn_id*/ None,
/*root_turn_id*/ None,
/*started_at*/ None,
InterruptedTurnHistoryMarker::ContextualUser,
)
@@ -2806,6 +2814,7 @@ fn interrupted_fork_snapshot_appends_interrupt_boundary() {
serde_json::to_value(vec![
RolloutItem::ResponseItem(contextual_user_interrupted_marker().into()),
RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id: None,
turn_id: None,
started_at: None,
reason: TurnAbortReason::Interrupted,
@@ -2828,6 +2837,7 @@ fn disabled_interrupted_fork_snapshot_appends_only_interrupt_event() {
append_interrupted_boundary(
committed_history,
/*turn_id*/ None,
/*root_turn_id*/ None,
/*started_at*/ None,
InterruptedTurnHistoryMarker::Disabled,
)
@@ -2837,6 +2847,7 @@ fn disabled_interrupted_fork_snapshot_appends_only_interrupt_event() {
serde_json::to_value(vec![
RolloutItem::ResponseItem(user_msg("hello").into()),
RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id: None,
turn_id: None,
started_at: None,
reason: TurnAbortReason::Interrupted,
@@ -2852,6 +2863,7 @@ fn disabled_interrupted_fork_snapshot_appends_only_interrupt_event() {
append_interrupted_boundary(
InitialHistory::New,
/*turn_id*/ None,
/*root_turn_id*/ None,
/*started_at*/ None,
InterruptedTurnHistoryMarker::Disabled,
)
@@ -2860,6 +2872,7 @@ fn disabled_interrupted_fork_snapshot_appends_only_interrupt_event() {
.expect("serialize disabled interrupted empty fork history"),
serde_json::to_value(vec![RolloutItem::EventMsg(EventMsg::TurnAborted(
TurnAbortedEvent {
root_turn_id: None,
turn_id: None,
started_at: None,
reason: TurnAbortReason::Interrupted,
@@ -2879,6 +2892,7 @@ fn interrupted_snapshot_is_not_mid_turn() {
RolloutItem::ResponseItem(assistant_msg("partial").into()),
RolloutItem::ResponseItem(contextual_user_interrupted_marker().into()),
RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id: None,
turn_id: Some("turn-1".to_string()),
started_at: None,
reason: TurnAbortReason::Interrupted,
@@ -2894,6 +2908,7 @@ fn interrupted_snapshot_is_not_mid_turn() {
ends_mid_turn: false,
active_turn_id: None,
active_turn_started_at: None,
active_turn_root_id: None,
active_turn_start_index: None,
},
);
@@ -2951,6 +2966,7 @@ fn completed_legacy_event_history_is_not_mid_turn() {
ends_mid_turn: false,
active_turn_id: None,
active_turn_started_at: None,
active_turn_root_id: None,
active_turn_start_index: None,
},
);
@@ -2976,6 +2992,7 @@ fn mixed_response_and_legacy_user_event_history_is_mid_turn() {
ends_mid_turn: true,
active_turn_id: None,
active_turn_started_at: None,
active_turn_root_id: None,
active_turn_start_index: None,
},
);
@@ -3060,6 +3077,7 @@ async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_histor
.expect("serialize interrupted marker");
let interrupted_abort_json = serde_json::to_value(RolloutItem::EventMsg(
EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id: None,
turn_id: expected_turn_id,
started_at: None,
reason: TurnAbortReason::Interrupted,
@@ -3155,6 +3173,7 @@ async fn interrupted_fork_snapshot_preserves_explicit_turn_id() {
ends_mid_turn: true,
active_turn_id: Some("turn-explicit".to_string()),
active_turn_started_at: None,
active_turn_root_id: None,
active_turn_start_index: Some(1),
},
);
@@ -3184,6 +3203,7 @@ async fn interrupted_fork_snapshot_preserves_explicit_turn_id() {
matches!(
item,
RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id: None,
turn_id: Some(turn_id),
started_at: None,
reason: TurnAbortReason::Interrupted,
codex-rs/core/src/thread_rollout_truncation_tests.rs
+1
View File
@@ -55,6 +55,7 @@ fn turn_started(turn_id: &str) -> RolloutItem {
fn turn_completed(turn_id: &str) -> RolloutItem {
RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
root_turn_id: None,
turn_id: turn_id.to_string(),
started_at: None,
last_agent_message: None,
codex-rs/core/src/tools/handlers/multi_agents_tests.rs
+4
View File
@@ -1502,6 +1502,7 @@ async fn multi_agent_v2_list_agents_returns_completed_status() {
.send_event(
child_turn.as_ref(),
EventMsg::TurnComplete(TurnCompleteEvent {
root_turn_id: None,
turn_id: child_turn.sub_id.clone(),
started_at: None,
last_agent_message: Some("done".to_string()),
@@ -1939,6 +1940,7 @@ async fn multi_agent_v2_followup_task_completion_notifies_parent_on_every_turn()
.send_event(
first_turn.as_ref(),
EventMsg::TurnComplete(TurnCompleteEvent {
root_turn_id: None,
turn_id: first_turn.sub_id.clone(),
started_at: None,
last_agent_message: Some("first done".to_string()),
@@ -1981,6 +1983,7 @@ async fn multi_agent_v2_followup_task_completion_notifies_parent_on_every_turn()
.send_event(
second_turn.as_ref(),
EventMsg::TurnComplete(TurnCompleteEvent {
root_turn_id: None,
turn_id: second_turn.sub_id.clone(),
started_at: None,
last_agent_message: Some("second done".to_string()),
@@ -2144,6 +2147,7 @@ async fn multi_agent_v2_interrupted_turn_does_not_notify_parent() {
.send_event(
aborted_turn.as_ref(),
EventMsg::TurnAborted(TurnAbortedEvent {
root_turn_id: None,
turn_id: Some(aborted_turn.sub_id.clone()),
started_at: None,
reason: TurnAbortReason::Interrupted,