diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index f64e9f9227..668e51f34b 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/src/protocol/item_builders_tests.rs b/codex-rs/app-server-protocol/src/protocol/item_builders_tests.rs index 3bb88ff493..f08082a8d9 100644 --- a/codex-rs/app-server-protocol/src/protocol/item_builders_tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/item_builders_tests.rs @@ -108,6 +108,7 @@ fn guardian_stdin_reviews_preserve_parent_command_history() { }; let start_turn = |turn_id: &str| { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.into(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index e7128a59cc..c0928c758c 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -2186,6 +2186,7 @@ mod tests { let thread_id = ThreadId::new(); let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.to_string(), root_turn_id: None, trace_id: None, @@ -2252,6 +2253,7 @@ mod tests { })); let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.to_string(), root_turn_id: None, trace_id: None, @@ -2321,6 +2323,7 @@ mod tests { let saved_path = test_path_buf("/tmp/image-1.png").abs(); let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.to_string(), root_turn_id: None, trace_id: None, @@ -2414,6 +2417,7 @@ mod tests { }); let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.to_string(), root_turn_id: None, trace_id: None, @@ -2533,6 +2537,7 @@ mod tests { let thread_id = ThreadId::new(); let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.to_string(), root_turn_id: None, trace_id: None, @@ -2634,6 +2639,7 @@ mod tests { fn replays_image_generation_end_events_into_turn_history() { let items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-image".into(), root_turn_id: None, trace_id: None, @@ -3029,6 +3035,7 @@ mod tests { fn uses_explicit_turn_boundaries_for_mid_turn_steering() { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: None, trace_id: None, @@ -3097,6 +3104,7 @@ mod tests { fn reconstructs_tool_items_from_persisted_completion_events() { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".into(), root_turn_id: None, trace_id: None, @@ -3235,6 +3243,7 @@ mod tests { fn reconstructs_mcp_tool_result_meta_from_persisted_completion_events() { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".into(), root_turn_id: None, trace_id: None, @@ -3318,6 +3327,7 @@ mod tests { fn reconstructs_dynamic_tool_items_from_request_and_response_events() { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".into(), root_turn_id: None, trace_id: None, @@ -3403,6 +3413,7 @@ mod tests { fn reconstructs_declined_exec_and_patch_items() { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".into(), root_turn_id: None, trace_id: None, @@ -3499,6 +3510,7 @@ mod tests { fn reconstructs_declined_guardian_command_item() { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".into(), root_turn_id: None, trace_id: None, @@ -3598,6 +3610,7 @@ mod tests { fn reconstructs_in_progress_guardian_execve_item() { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".into(), root_turn_id: None, trace_id: None, @@ -3706,6 +3719,7 @@ mod tests { }; let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: None, trace_id: None, @@ -3727,6 +3741,7 @@ mod tests { time_to_first_token_ms: None, }), EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-b".into(), root_turn_id: None, trace_id: None, @@ -3767,6 +3782,7 @@ mod tests { fn assigns_late_exec_completion_to_original_turn() { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: None, trace_id: None, @@ -3792,6 +3808,7 @@ mod tests { time_to_first_token_ms: None, }), EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-b".into(), root_turn_id: None, trace_id: None, @@ -3874,6 +3891,7 @@ mod tests { fn drops_late_turn_scoped_item_for_unknown_turn_id() { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: None, trace_id: None, @@ -3899,6 +3917,7 @@ mod tests { time_to_first_token_ms: None, }), EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-b".into(), root_turn_id: None, trace_id: None, @@ -3975,6 +3994,7 @@ mod tests { None ); builder.handle_event(&EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".into(), root_turn_id: None, trace_id: None, @@ -4044,6 +4064,7 @@ mod tests { let mut builder = ThreadHistoryBuilder::new(); assert_eq!(builder.active_turn_metadata_snapshot(), None); builder.handle_event(&EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".into(), root_turn_id: Some("root-turn".into()), trace_id: None, @@ -4092,6 +4113,7 @@ mod tests { let mut builder = ThreadHistoryBuilder::new(); let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.to_string(), root_turn_id: None, trace_id: None, @@ -4161,6 +4183,7 @@ mod tests { let mut builder = ThreadHistoryBuilder::new(); let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.to_string(), root_turn_id: None, trace_id: None, @@ -4230,6 +4253,7 @@ mod tests { fn late_turn_complete_does_not_close_active_turn() { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: None, trace_id: None, @@ -4255,6 +4279,7 @@ mod tests { time_to_first_token_ms: None, }), EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-b".into(), root_turn_id: None, trace_id: None, @@ -4312,6 +4337,7 @@ mod tests { fn late_turn_complete_with_embedded_error_preserves_active_turn() { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: None, trace_id: None, @@ -4328,6 +4354,7 @@ mod tests { ..Default::default() }), EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-b".into(), root_turn_id: None, trace_id: None, @@ -4416,6 +4443,7 @@ mod tests { fn late_turn_aborted_does_not_interrupt_active_turn() { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: None, trace_id: None, @@ -4441,6 +4469,7 @@ mod tests { time_to_first_token_ms: None, }), EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-b".into(), root_turn_id: None, trace_id: None, @@ -4489,6 +4518,7 @@ mod tests { fn preserves_compaction_only_turn() { let items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-compact".into(), root_turn_id: None, trace_id: None, @@ -4763,6 +4793,7 @@ mod tests { fn out_of_turn_error_does_not_create_or_fail_a_turn() { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: None, trace_id: None, @@ -4826,6 +4857,7 @@ mod tests { fn error_then_turn_complete_preserves_failed_status() { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: None, trace_id: None, @@ -4886,6 +4918,7 @@ mod tests { fn turn_complete_with_embedded_error_marks_turn_failed() { let events = vec![ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: None, trace_id: None, @@ -4957,6 +4990,7 @@ mod tests { .expect("hook prompt message"); let items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: None, trace_id: None, @@ -5018,6 +5052,7 @@ mod tests { let expected_item = ThreadItem::from(hook_prompt.clone()); let mut builder = ThreadHistoryBuilder::new(); builder.handle_event(&EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: None, trace_id: None, @@ -5047,6 +5082,7 @@ mod tests { .expect("worker path"); let items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: None, trace_id: None, @@ -5095,6 +5131,7 @@ mod tests { fn ignores_plain_user_response_items_in_rollout_replay() { let items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: None, trace_id: None, @@ -5256,6 +5293,7 @@ mod tests { let start_changes = builder.handle_rollout_item_with_changes(&RolloutItem::EventMsg( EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: Some("root-turn".into()), trace_id: None, @@ -5374,6 +5412,7 @@ mod tests { let mut builder = ThreadHistoryBuilder::new(); let changes = builder.handle_rollout_items_with_changes(&[ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: Some("root-turn".into()), trace_id: None, @@ -5415,6 +5454,7 @@ mod tests { let mut builder = ThreadHistoryBuilder::new(); let changes = builder.handle_rollout_items_with_changes(&[ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-a".into(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs b/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs index 729f673e29..bed87ac9ab 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs @@ -26,6 +26,7 @@ use crate::protocol::v2::TurnError; fn projects_turn_lifecycle_without_prior_builder_state() { let started = project(RolloutItem::EventMsg(EventMsg::TurnStarted( TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".to_string(), root_turn_id: Some("root-turn".into()), trace_id: None, diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index d1b0ab8975..14908318e3 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -3089,6 +3089,7 @@ mod tests { state.track_current_turn_event( "turn-1", &EventMsg::TurnStarted(codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".to_string(), root_turn_id: None, trace_id: None, @@ -3125,6 +3126,7 @@ mod tests { Event { id: "turn-1".to_string(), msg: EventMsg::TurnStarted(codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".to_string(), root_turn_id: None, trace_id: None, @@ -3395,6 +3397,7 @@ mod tests { state.track_current_turn_event( &event_turn_id, &EventMsg::TurnStarted(codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: event_turn_id.clone(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs index af878ff8e0..c4cf0cfc2b 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs @@ -1199,6 +1199,7 @@ mod thread_processor_behavior_tests { state.track_current_turn_event( "turn-1", &EventMsg::TurnStarted(codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/app-server/src/turn_cost_worker_tests.rs b/codex-rs/app-server/src/turn_cost_worker_tests.rs index efd940492c..e6a5f81ae1 100644 --- a/codex-rs/app-server/src/turn_cost_worker_tests.rs +++ b/codex-rs/app-server/src/turn_cost_worker_tests.rs @@ -87,6 +87,7 @@ async fn handle_observes_only_matching_model_provider() { let event = Event { id: "turn-1".to_string(), msg: EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index 3690b3eb1f..35bfda57be 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -1249,6 +1249,7 @@ async fn thread_fork_can_cut_before_unfinished_stored_turn() -> Result<()> { append_rollout_item_to_path( &source_path, &RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: unfinished_turn_id.to_string(), root_turn_id: None, trace_id: None, @@ -1524,6 +1525,7 @@ async fn thread_fork_creates_reference_backed_paginated_thread() -> Result<()> { ); for item in [ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".to_string(), root_turn_id: None, trace_id: None, @@ -1854,6 +1856,7 @@ async fn assert_thread_fork_freezes_active_paginated_turn_as_interrupted( append_rollout_item_to_path( source_path.as_path(), &RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "active-turn".to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index 3ac95063e4..58505d0d61 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -2426,6 +2426,7 @@ async fn read_items_page( fn paginated_turn_started(turn_id: &str) -> RolloutItem { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index 6f58231e30..6e339c75a5 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -3915,6 +3915,7 @@ async fn cold_paginated_resume_restores_usage_without_loading_turns() -> Result< append_rollout_item_to_path( &path, &RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: canonical_turn_id.to_string(), root_turn_id: None, trace_id: None, @@ -4027,6 +4028,7 @@ async fn cold_paginated_resume_omits_usage_when_its_turn_is_ambiguous() -> Resul append_rollout_item_to_path( &path, &RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: interrupted_turn_id.to_string(), root_turn_id: None, trace_id: None, @@ -4170,6 +4172,7 @@ async fn thread_resume_token_usage_replay_ignores_stale_interrupted_tail_turn() "timestamp": meta_rfc3339, "type": "event_msg", "payload": serde_json::to_value(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: stale_turn_id.to_string(), root_turn_id: None, trace_id: None, @@ -4259,6 +4262,7 @@ async fn thread_resume_token_usage_replay_can_belong_to_interrupted_turn() -> Re "timestamp": meta_rfc3339, "type": "event_msg", "payload": serde_json::to_value(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: interrupted_turn_id.to_string(), root_turn_id: None, trace_id: None, @@ -4576,6 +4580,7 @@ async fn thread_resume_and_read_interrupt_incomplete_rollout_turn_when_thread_is "timestamp": meta_rfc3339, "type": "event_msg", "payload": serde_json::to_value(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/app-server/tests/suite/v2/thread_timeline.rs b/codex-rs/app-server/tests/suite/v2/thread_timeline.rs index c0babef2bc..954f09eaf6 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_timeline.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_timeline.rs @@ -98,6 +98,7 @@ async fn timeline_pages_mix_items_and_resolve_the_opening_realtime_session() -> content: RealtimeItemContent::RealtimeSessionStarted, }), RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index 901a8f153e..c5a17767da 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -1201,6 +1201,7 @@ impl LocalAgentControl { compacted.latest_token_usage_record = None; if let Some(resume_metadata) = &mut compacted.resume_metadata { resume_metadata.multi_agent_version = Some(multi_agent_version); + resume_metadata.turn_attribution = None; if !preserve_context_baselines { resume_metadata.previous_turn_settings = None; } diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 7bb29d8e00..17404e0acc 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -565,6 +565,7 @@ async fn get_status_returns_not_found_without_manager() { #[tokio::test] async fn on_event_updates_status_from_task_started() { let status = agent_status_from_event(&EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".to_string(), root_turn_id: None, trace_id: None, @@ -2256,7 +2257,14 @@ async fn spawn_agent_fork_sanitizes_inherited_compaction_metadata() { let (parent_thread_id, parent_thread) = harness.start_paginated_thread().await; let parent_resume_metadata = codex_history::CompactionResumeMetadata { multi_agent_version: Some(MultiAgentVersion::V2), - last_started_turn_id: Some("parent-turn".into()), + last_started_turn_id: Some("parent-turn".to_string()), + turn_attribution: Some(codex_history::TurnAttribution { + turn_id: "parent-turn".to_string(), + turn_trigger: Some("automation".to_string()), + parent_turn_id: Some("initiating-turn".to_string()), + initiating_agent_path: Some(codex_protocol::AgentPath::root()), + root_turn_id: Some("root-turn".to_string()), + }), previous_turn_settings: Some(codex_history::PreviousTurnSettings { model: "parent-model".into(), comp_hash: None, @@ -2382,6 +2390,7 @@ async fn spawn_agent_fork_sanitizes_inherited_compaction_metadata() { inherited_resume_metadata, &codex_history::CompactionResumeMetadata { multi_agent_version: Some(MultiAgentVersion::V1), + turn_attribution: None, ..parent_resume_metadata } ); @@ -3287,6 +3296,7 @@ async fn spawn_agent_full_fork_legacy_compaction_rebuilds_child_instructions_onc resume_metadata: Some(codex_history::CompactionResumeMetadata { multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: None, + turn_attribution: None, previous_turn_settings: Some(codex_history::PreviousTurnSettings { model: "parent-model".into(), comp_hash: None, @@ -3380,6 +3390,7 @@ async fn spawn_agent_full_fork_legacy_compaction_rebuilds_child_instructions_onc Some(codex_history::CompactionResumeMetadata { multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: None, + turn_attribution: None, previous_turn_settings: None, }) ); diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 4000b257a1..7eb186016d 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -474,16 +474,9 @@ impl CodexThread { trace, cyber_access_program, } = request; - let root_turn_id = self - .session - .reference_context_item() - .await - .filter(|context| context.turn_id.as_deref() == Some(turn_id.as_str())) - .and_then(|context| context.root_turn_id); let start_options = TurnStartOptions { cyber_access_program, - root_turn_id, - ..Default::default() + ..self.session.recovered_turn_start_options(&turn_id).await }; match self .io diff --git a/codex-rs/core/src/realtime_conversation/transcript_tail.rs b/codex-rs/core/src/realtime_conversation/transcript_tail.rs index 8bdb9ba22e..eb778e4a0b 100644 --- a/codex-rs/core/src/realtime_conversation/transcript_tail.rs +++ b/codex-rs/core/src/realtime_conversation/transcript_tail.rs @@ -37,6 +37,7 @@ pub(super) async fn record(session: &Session, text: String) -> std::io::Result<( let now = chrono::Utc::now(); if recording_turn { let started = EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn.sub_id.clone(), root_turn_id: Some(turn.sub_id.clone()), trace_id: turn.trace_id.clone(), @@ -47,7 +48,9 @@ pub(super) async fn record(session: &Session, text: String) -> std::io::Result<( session .persist_rollout_items(&[RolloutItem::EventMsg(started)]) .await; - session.record_started_turn(&turn.sub_id).await; + session + .record_started_turn(&turn.sub_id, /*attribution*/ None) + .await; } let item = session.response_item_from_user_input(content.clone()); session diff --git a/codex-rs/core/src/realtime_history_tests.rs b/codex-rs/core/src/realtime_history_tests.rs index fd163ec459..c6b0b977a6 100644 --- a/codex-rs/core/src/realtime_history_tests.rs +++ b/codex-rs/core/src/realtime_history_tests.rs @@ -28,6 +28,7 @@ fn started_state() -> RealtimeHistoryState { let mut state = RealtimeHistoryState::default(); state.observe(&EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".to_string(), root_turn_id: None, trace_id: None, @@ -129,6 +130,7 @@ fn interrupted_turn_is_not_associated_with_a_new_voice_session( let mut state = RealtimeHistoryState::default(); state.observe(&EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/core/src/session/daemon_recovery.rs b/codex-rs/core/src/session/daemon_recovery.rs index 5fdefff870..bbf27dd367 100644 --- a/codex-rs/core/src/session/daemon_recovery.rs +++ b/codex-rs/core/src/session/daemon_recovery.rs @@ -1,5 +1,5 @@ -//! Captures recorded, uncanceled regular turns using one thread-configured local executor. -//! Callers must flush the rollout after capture before persisting the snapshot. +//! Captures interrupted turns for daemon recovery and restores persisted turn attribution. +//! Callers must flush the rollout after capturing a live turn before persisting its snapshot. use super::Session; use codex_protocol::protocol::EnvironmentConfigState; @@ -9,6 +9,16 @@ use codex_protocol::protocol::TurnEnvironmentSelection; pub(super) struct RecordedTurnInput; impl Session { + pub(crate) async fn recovered_turn_start_options( + &self, + turn_id: &str, + ) -> crate::TurnStartOptions { + self.state + .lock() + .await + .recovered_turn_start_options(turn_id) + } + /// Captures a regular turn only after its input is recorded. The caller must flush the rollout. pub(crate) async fn interrupted_turn( &self, diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 427dfdcbd6..7f0bd90d16 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -318,6 +318,7 @@ use crate::state::PendingRequestPermissions; use crate::state::ReasoningEffortPin; use crate::state::SessionServices; use crate::state::SessionState; +use crate::state::TaskKind; #[cfg(test)] use crate::stream_events_utils::HandleOutputCtx; #[cfg(test)] @@ -1754,6 +1755,7 @@ impl Session { last_started_turn_id, previous_turn_settings, reference_context_item, + turn_attribution, world_state_baseline, window_number, first_window_id, @@ -1809,6 +1811,7 @@ impl Session { reviewer_compaction_hash.as_deref(), ); state.last_started_turn_id = last_started_turn_id; + state.turn_attribution = turn_attribution; if let Some(world_state) = world_state_baseline { state.history.set_world_state_baseline(world_state); } @@ -2259,8 +2262,9 @@ impl Session { .trusted_guardian_reviewer } - pub(crate) async fn emit_turn_started(&self, turn_context: &TurnContext) { + pub(crate) async fn emit_turn_started(&self, turn_context: &TurnContext, task_kind: TaskKind) { 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 @@ -4082,6 +4086,7 @@ impl Session { resume_metadata: Some(CompactionResumeMetadata { multi_agent_version: self.multi_agent_version(), last_started_turn_id: state.last_started_turn_id.clone(), + turn_attribution: state.turn_attribution.clone(), previous_turn_settings: state.previous_turn_settings(), }), } diff --git a/codex-rs/core/src/session/rollout_reconstruction.rs b/codex-rs/core/src/session/rollout_reconstruction.rs index ab20e5b30e..4ca5ce1498 100644 --- a/codex-rs/core/src/session/rollout_reconstruction.rs +++ b/codex-rs/core/src/session/rollout_reconstruction.rs @@ -2,6 +2,7 @@ use super::*; use crate::context::world_state::WorldStateSnapshot; use crate::context_manager::is_user_turn_boundary; use codex_history::ResponseItemEnvelope; +use codex_history::TurnAttribution; use codex_protocol::protocol::SessionContextWindow; use codex_protocol::protocol::ThreadHistoryMode; use uuid::Uuid; @@ -16,6 +17,7 @@ pub(super) struct RolloutReconstruction { pub(super) last_started_turn_id: Option, pub(super) previous_turn_settings: Option, pub(super) reference_context_item: Option, + pub(super) turn_attribution: Option, pub(super) world_state_baseline: Option, pub(super) window_number: u64, pub(super) first_window_id: Option, @@ -94,11 +96,25 @@ struct ActiveReplaySegment<'a> { counts_as_user_turn: bool, previous_turn_settings: Option, reference_context_item: TurnReferenceContextItem, + turn_attribution: Option, world_state_replay: Vec<&'a RolloutItem>, history_checkpoint: Option>, window: Option, } +impl ActiveReplaySegment<'_> { + fn take_checkpoint_attribution(&mut self, checkpoint: &mut Option) { + if checkpoint.as_ref().is_some_and(|attribution| { + self.turn_id + .as_ref() + .is_none_or(|id| id == &attribution.turn_id) + }) { + // Consume a matching checkpoint even when rollback discards this segment. + self.turn_attribution = self.turn_attribution.take().or(checkpoint.take()); + } + } +} + fn turn_ids_are_compatible(active_turn_id: Option<&str>, item_turn_id: Option<&str>) -> bool { active_turn_id .is_none_or(|turn_id| item_turn_id.is_none_or(|item_turn_id| item_turn_id == turn_id)) @@ -112,7 +128,7 @@ fn finalize_active_segment<'a>( world_state_replay: &mut Vec<&'a RolloutItem>, window: &mut Option, pending_rollback_turns: &mut usize, -) { +) -> Option { // Thread rollback drops the newest surviving real user-message boundaries. In replay, that // means skipping the next finalized segments that contain a non-contextual // `EventMsg::UserMessage`. @@ -120,7 +136,7 @@ fn finalize_active_segment<'a>( if active_segment.counts_as_user_turn { *pending_rollback_turns -= 1; } - return; + return None; } // Full world-state snapshots are persisted after installing initial context. They still @@ -163,6 +179,9 @@ fn finalize_active_segment<'a>( { *reference_context_item = active_segment.reference_context_item; } + // Automatic turns can persist attribution without establishing a model-context baseline. + // Select recovery metadata from the same surviving segments, independently of that baseline. + active_segment.turn_attribution } impl Session { @@ -197,11 +216,11 @@ impl Session { .and_then(|checkpoint| reconstructed_window_from_compaction(checkpoint.compacted)); // Scan the selected items backward to find the newest surviving turn state. - let last_started_turn_id = replay_items + let mut started_turn_ids = replay_items .iter() .rev() - .find_map(|item| match item { - RolloutItem::EventMsg(EventMsg::TurnStarted(event)) => Some(event.turn_id.clone()), + .filter_map(|item| match item { + RolloutItem::EventMsg(EventMsg::TurnStarted(event)) => Some(event.turn_id.as_str()), RolloutItem::SessionMeta(_) | RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) @@ -215,19 +234,37 @@ impl Session { | RolloutItem::Compacted(_) | RolloutItem::EventMsg(_) => None, }) + .peekable(); + let last_started_turn_id = started_turn_ids + .peek() + .map(|id| (*id).to_owned()) .or_else(|| resume_metadata.and_then(|metadata| metadata.last_started_turn_id.clone())); + // Saved attribution still identifies the turn when continuation eligibility was cleared. + let checkpoint_turn_id = resume_metadata.and_then(|metadata| { + metadata.last_started_turn_id.as_deref().or_else(|| { + metadata + .turn_attribution + .as_ref() + .map(|attribution| attribution.turn_id.as_str()) + }) + }); let mut previous_turn_settings = None; let mut reference_context_item = TurnReferenceContextItem::NeverSet; + let mut turn_attribution = None; let mut world_state_replay = Vec::new(); // Rollback is "drop the newest N user turns". While scanning in reverse, that becomes // "skip the next N user-turn segments we finalize". let mut pending_rollback_turns = 0usize; // Reverse replay accumulates rollout items into the newest in-progress turn segment until // we hit its matching `TurnStarted`, at which point the segment can be finalized. + let mut checkpoint_attribution = + resume_metadata.and_then(|metadata| metadata.turn_attribution.clone()); + let mut late_completed_turn_ids = HashSet::new(); let mut active_segment: Option> = None; for (index, item) in replay_items.iter().enumerate().rev() { + let preceding_turn_id = started_turn_ids.peek().copied().or(checkpoint_turn_id); match item { RolloutItem::Compacted(compacted) => { let active_segment = @@ -261,6 +298,11 @@ impl Session { .saturating_add(usize::try_from(rollback.num_turns).unwrap_or(usize::MAX)); } RolloutItem::EventMsg(EventMsg::TurnComplete(event)) => { + // Stop hooks can finish after a newer turn has started. + if !turn_ids_are_compatible(preceding_turn_id, Some(event.turn_id.as_str())) { + late_completed_turn_ids.insert(event.turn_id.as_str()); + continue; + } let active_segment = active_segment.get_or_insert_with(ActiveReplaySegment::default); active_segment.turn_completed = true; @@ -271,6 +313,9 @@ impl Session { } } RolloutItem::EventMsg(EventMsg::TurnAborted(event)) => { + if !turn_ids_are_compatible(preceding_turn_id, event.turn_id.as_deref()) { + continue; + } if let Some(active_segment) = active_segment.as_mut() { if active_segment.turn_id.is_none() && let Some(turn_id) = &event.turn_id @@ -322,15 +367,34 @@ impl Session { active_segment.world_state_replay.push(item); } RolloutItem::EventMsg(EventMsg::TurnStarted(event)) => { + started_turn_ids.next(); + // Startup can be suspended before any input or context is written. + // Explicit start attribution is authoritative, including unknown fields. + if let Some(attribution) = &event.turn_attribution + && attribution.turn_id == event.turn_id + { + let segment = + active_segment.get_or_insert_with(ActiveReplaySegment::default); + if turn_ids_are_compatible( + segment.turn_id.as_deref(), + Some(event.turn_id.as_str()), + ) { + segment.turn_attribution = Some(attribution.clone()); + } + } // `TurnStarted` is the oldest boundary of the active reverse segment. if active_segment.as_ref().is_some_and(|active_segment| { turn_ids_are_compatible( active_segment.turn_id.as_deref(), Some(event.turn_id.as_str()), ) - }) && let Some(active_segment) = active_segment.take() + }) && let Some(mut active_segment) = active_segment.take() { - finalize_active_segment( + active_segment + .turn_id + .get_or_insert_with(|| event.turn_id.clone()); + active_segment.take_checkpoint_attribution(&mut checkpoint_attribution); + let segment_context = finalize_active_segment( active_segment, &mut history_checkpoint, &mut previous_turn_settings, @@ -339,6 +403,7 @@ impl Session { &mut window, &mut pending_rollback_turns, ); + turn_attribution = turn_attribution.or(segment_context); } } RolloutItem::ResponseItem(response_item) => { @@ -363,12 +428,18 @@ impl Session { } if let Some(mut active_segment) = active_segment.take() { + active_segment.take_checkpoint_attribution(&mut checkpoint_attribution); + active_segment.turn_completed |= active_segment + .turn_id + .as_deref() + .or(checkpoint_turn_id) + .is_some_and(|turn_id| late_completed_turn_ids.contains(turn_id)); // A companion turn context only restores the context baseline. Once that turn // completes, its settings are newer than the compaction metadata. if resume_metadata.is_some() && !active_segment.turn_completed { active_segment.previous_turn_settings = None; } - finalize_active_segment( + let segment_context = finalize_active_segment( active_segment, &mut history_checkpoint, &mut previous_turn_settings, @@ -377,6 +448,11 @@ impl Session { &mut window, &mut pending_rollback_turns, ); + turn_attribution = turn_attribution.or(segment_context); + } + + if pending_rollback_turns == 0 { + turn_attribution = turn_attribution.or(checkpoint_attribution); } if previous_turn_settings.is_none() { @@ -527,6 +603,7 @@ impl Session { history: history.into_annotated_items(), previous_turn_settings, reference_context_item, + turn_attribution, world_state_baseline, window_number: window.number, first_window_id: window.first_id, diff --git a/codex-rs/core/src/session/rollout_reconstruction_tests.rs b/codex-rs/core/src/session/rollout_reconstruction_tests.rs index 7b280be120..ddf567e159 100644 --- a/codex-rs/core/src/session/rollout_reconstruction_tests.rs +++ b/codex-rs/core/src/session/rollout_reconstruction_tests.rs @@ -260,6 +260,7 @@ fn completed_user_turn_rollout( let mut rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.clone(), root_turn_id: None, trace_id: None, @@ -524,6 +525,7 @@ async fn record_initial_history_resumed_hydrates_previous_turn_settings_from_lif let rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.clone(), root_turn_id: None, trace_id: None, @@ -581,14 +583,22 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_com let (session, turn_context) = make_session_and_context().await; let mut first_context_item = turn_context.to_turn_context_item(); first_context_item.cyber_access_program = Some(CyberAccessProgram::DaybreakBlue); + first_context_item.root_turn_id = Some("root-turn".to_owned()); let first_turn_id = first_context_item .turn_id .clone() .expect("turn context should have turn_id"); + let first_attribution: codex_history::TurnAttribution = object!({ + "turn_id": first_turn_id, + "turn_trigger": "composer", + "parent_turn_id": "parent-turn", + "root_turn_id": "root-turn" + }); let mut rolled_back_context_item = first_context_item.clone(); rolled_back_context_item.turn_id = Some("rolled-back-turn".to_string()); rolled_back_context_item.model = "rolled-back-model".to_string(); rolled_back_context_item.cyber_access_program = Some(CyberAccessProgram::DaybreakRed); + rolled_back_context_item.root_turn_id = Some("rolled-back-root".to_owned()); let rolled_back_turn_id = rolled_back_context_item .turn_id .clone() @@ -601,6 +611,7 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_com let rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: Some(first_attribution.clone()), turn_id: first_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -638,6 +649,12 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_com )), RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: Some(object!({ + "turn_id": rolled_back_turn_id, + "turn_trigger": "automation", + "parent_turn_id": "rolled-back-parent", + "root_turn_id": "rolled-back-root" + })), turn_id: rolled_back_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -696,6 +713,7 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_com "cyber_access_program": "daybreak_blue", }) ); + assert_eq!(reconstructed.turn_attribution, Some(first_attribution)); assert_eq!( serde_json::to_value(reconstructed.reference_context_item) .expect("serialize reconstructed reference context item"), @@ -726,6 +744,7 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_inc let rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: first_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -760,6 +779,7 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_inc )), RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: incomplete_turn_id, root_turn_id: None, trace_id: None, @@ -829,6 +849,7 @@ async fn reconstruct_history_rollback_skips_non_user_turns_for_history_and_metad let rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: first_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -863,6 +884,7 @@ async fn reconstruct_history_rollback_skips_non_user_turns_for_history_and_metad )), RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: second_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -896,6 +918,7 @@ async fn reconstruct_history_rollback_skips_non_user_turns_for_history_and_metad )), RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: standalone_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -970,6 +993,7 @@ async fn reconstruct_history_rollback_counts_inter_agent_assistant_turns() { let rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: first_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -1004,6 +1028,7 @@ async fn reconstruct_history_rollback_counts_inter_agent_assistant_turns() { )), RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: assistant_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -1070,6 +1095,7 @@ async fn reconstruct_history_rollback_clears_history_and_metadata_when_exceeding let rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: only_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -1114,6 +1140,7 @@ async fn reconstruct_history_rollback_clears_history_and_metadata_when_exceeding assert_eq!(reconstructed.history, Vec::new()); assert_eq!(reconstructed.previous_turn_settings, None); assert!(reconstructed.reference_context_item.is_none()); + assert!(reconstructed.turn_attribution.is_none()); } #[tokio::test] @@ -1128,6 +1155,7 @@ async fn record_initial_history_resumed_rollback_skips_only_user_turns() { let rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: user_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -1161,6 +1189,7 @@ async fn record_initial_history_resumed_rollback_skips_only_user_turns() { // Standalone task turn (no UserMessage) should not consume rollback skips. RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: standalone_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -1211,6 +1240,7 @@ async fn record_initial_history_resumed_rollback_drops_incomplete_user_turn_comp let rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: previous_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -1243,6 +1273,7 @@ async fn record_initial_history_resumed_rollback_drops_incomplete_user_turn_comp )), RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: incomplete_turn_id, root_turn_id: None, trace_id: None, @@ -1581,6 +1612,7 @@ async fn bounded_replay_matches_full_replay_after_empty_turn_compactions( resume_metadata: current.then(|| codex_history::CompactionResumeMetadata { multi_agent_version: None, last_started_turn_id: Some(format!("wake-{window_number}")), + turn_attribution: None, previous_turn_settings: Some(PreviousTurnSettings { model: format!("metadata-model-{window_number}"), comp_hash: Some(format!("metadata-hash-{window_number}")), @@ -1687,17 +1719,147 @@ async fn paginated_compaction_does_not_restore_missing_companions_from_older_his assert_eq!(reconstructed.previous_turn_settings, None); assert_eq!(reconstructed.reference_context_item, None); + assert_eq!(reconstructed.turn_attribution, None); assert_eq!(reconstructed.world_state_baseline, None); } } +#[derive(Clone, Copy)] +enum AttributionEvidence { + StartAndContext, + StartOnly, +} + +#[test_case(AttributionEvidence::StartAndContext; "late completion")] +#[test_case(AttributionEvidence::StartOnly; "late abort before current context")] #[tokio::test] -async fn completed_turn_suffix_after_compaction_overrides_resume_metadata() { +async fn reconstruct_attribution_ignores_previous_turn_terminal(evidence: AttributionEvidence) { + let (session, turn_context) = make_session_and_context().await; + let older_attribution = turn_context.attribution(); + let expected: codex_history::TurnAttribution = object!({ + "turn_id": "current-turn", + "turn_trigger": "automation", + "parent_turn_id": "calling-turn", + "root_turn_id": "root-turn", + "initiating_agent_path": "/root/requester" + }); + let mut current_context = turn_context.to_turn_context_item(); + current_context.turn_id = Some(expected.turn_id.clone()); + current_context.root_turn_id = expected.root_turn_id.clone(); + let mut rollout = completed_user_turn_rollout(turn_context.to_turn_context_item(), Vec::new()); + let RolloutItem::EventMsg(EventMsg::TurnStarted(start)) = &mut rollout[0] else { + panic!("expected previous turn start"); + }; + start.turn_attribution = Some(older_attribution.clone()); + let mut late_terminal = rollout.pop().expect("previous turn completion"); + let mut current = completed_user_turn_rollout(current_context, Vec::new()); + current.pop().expect("current turn completion"); + let RolloutItem::EventMsg(EventMsg::TurnStarted(start)) = &mut current[0] else { + panic!("expected current turn start"); + }; + start.turn_attribution = Some(expected.clone()); + if matches!(evidence, AttributionEvidence::StartOnly) { + current.truncate(1); + late_terminal = RolloutItem::EventMsg(EventMsg::TurnAborted(object!({ + "turn_id": older_attribution.turn_id, + "reason": "interrupted" + }))); + } + rollout.extend(current); + rollout.push(late_terminal); + + let reconstructed = session + .reconstruct_history_from_rollout(&turn_context, &rollout) + .await; + assert_eq!(reconstructed.turn_attribution, Some(expected)); + + if matches!(evidence, AttributionEvidence::StartAndContext) { + rollout.push(RolloutItem::EventMsg(EventMsg::ThreadRolledBack( + ThreadRolledBackEvent { num_turns: 1 }, + ))); + let rolled_back = session + .reconstruct_history_from_rollout(&turn_context, &rollout) + .await; + assert_eq!(rolled_back.turn_attribution, Some(older_attribution)); + } +} + +#[test_case(Some("current-turn"); "continuation marker retained")] +#[test_case(None; "continuation marker cleared")] +#[tokio::test] +async fn reconstruct_attribution_ignores_previous_terminal_after_checkpoint( + last_started_turn_id: Option<&str>, +) { + let (session, turn_context) = make_session_and_context().await; + let expected: codex_history::TurnAttribution = object!({ + "turn_id": "current-turn", + "turn_trigger": "automation", + "parent_turn_id": "calling-turn", + "root_turn_id": "root-turn" + }); + let mut context = turn_context.to_turn_context_item(); + context.turn_id = Some(expected.turn_id.clone()); + context.root_turn_id = expected.root_turn_id.clone(); + let settings: PreviousTurnSettings = object!({"model": "checkpoint-model"}); + let late_terminal = + completed_user_turn_rollout(turn_context.to_turn_context_item(), Vec::new()) + .pop() + .expect("previous turn completion"); + let mut rollout = vec![ + RolloutItem::Compacted(object!({ + "message": "summary", + "replacement_history": [user_message("original work")], + "window_number": 1, + "resume_metadata": { + "last_started_turn_id": last_started_turn_id, + "turn_attribution": expected, + "previous_turn_settings": settings + } + })), + RolloutItem::ResponseItem(user_message("continued work").into()), + RolloutItem::TurnContext(context.clone()), + late_terminal, + ]; + let reconstructed = session + .reconstruct_history_from_rollout(&turn_context, &rollout) + .await; + assert_eq!(reconstructed.turn_attribution, Some(expected)); + assert_eq!(reconstructed.reference_context_item, Some(context)); + assert_eq!(reconstructed.previous_turn_settings, Some(settings)); + assert_eq!( + reconstructed.last_started_turn_id.as_deref(), + last_started_turn_id + ); + + rollout.push(RolloutItem::EventMsg(EventMsg::ThreadRolledBack( + ThreadRolledBackEvent { num_turns: 1 }, + ))); + let rolled_back = session + .reconstruct_history_from_rollout(&turn_context, &rollout) + .await; + assert_eq!(rolled_back.turn_attribution, None); +} + +#[test_case(false, false; "completion before next turn")] +#[test_case(true, false; "completion after next turn starts")] +#[test_case(true, true; "late completion without checkpoint turn identity")] +#[tokio::test] +async fn completed_turn_suffix_after_compaction_overrides_resume_metadata( + late_completion: bool, + legacy_checkpoint: bool, +) { let (session, turn_context) = make_session_and_context().await; let mut newer_context = turn_context.to_turn_context_item(); newer_context.turn_id = Some("newer-turn".to_string()); newer_context.model = "newer-model".to_string(); newer_context.comp_hash = Some("newer-hash".to_string()); + let checkpoint_attribution = (!legacy_checkpoint).then(|| codex_history::TurnAttribution { + turn_id: "newer-turn".to_string(), + turn_trigger: Some("automation".to_string()), + parent_turn_id: None, + initiating_agent_path: None, + root_turn_id: None, + }); let expected_settings = PreviousTurnSettings { model: newer_context.model.clone(), comp_hash: newer_context.comp_hash.clone(), @@ -1710,6 +1872,8 @@ async fn completed_turn_suffix_after_compaction_overrides_resume_metadata() { "replacement_history": [user_message("seed"), assistant_message("summary")], "window_number": 1, "resume_metadata": { + "last_started_turn_id": checkpoint_attribution.as_ref().map(|value| value.turn_id.as_str()), + "turn_attribution": checkpoint_attribution, "previous_turn_settings": { "model": "metadata-model", "comp_hash": "metadata-hash" @@ -1725,6 +1889,17 @@ async fn completed_turn_suffix_after_compaction_overrides_resume_metadata() { .into_iter() .skip(2), ); + let next_attribution: Option = + late_completion.then(|| object!({"turn_id": "next-turn", "turn_trigger": "user"})); + if let Some(attribution) = &next_attribution { + rollout_items.insert( + rollout_items.len() - 1, + RolloutItem::EventMsg(EventMsg::TurnStarted(object!({ + "turn_id": attribution.turn_id, + "turn_attribution": attribution + }))), + ); + } let reconstructed = session .reconstruct_history_from_rollout(&turn_context, &rollout_items) @@ -1734,7 +1909,35 @@ async fn completed_turn_suffix_after_compaction_overrides_resume_metadata() { reconstructed.previous_turn_settings, Some(expected_settings) ); - assert_eq!(reconstructed.reference_context_item, Some(newer_context)); + assert_eq!( + reconstructed.turn_attribution, + next_attribution.or(checkpoint_attribution) + ); + assert_eq!( + reconstructed.reference_context_item, + Some(newer_context.clone()) + ); + + // A rollback of a newer user turn restores the checkpoint's older regular turn. + let RolloutItem::Compacted(checkpoint) = &mut rollout_items[0] else { + panic!("expected checkpoint"); + }; + let metadata = checkpoint + .resume_metadata + .as_mut() + .expect("resume metadata"); + let older_attribution: codex_history::TurnAttribution = + object!({"turn_id": "older-turn", "turn_trigger": "automation"}); + metadata.turn_attribution = Some(older_attribution.clone()); + rollout_items.truncate(1); + rollout_items.extend(completed_user_turn_rollout(newer_context, Vec::new())); + rollout_items.push(RolloutItem::EventMsg(EventMsg::ThreadRolledBack( + codex_protocol::protocol::ThreadRolledBackEvent { num_turns: 1 }, + ))); + let rolled_back = session + .reconstruct_history_from_rollout(&turn_context, &rollout_items) + .await; + assert_eq!(rolled_back.turn_attribution, Some(older_attribution)); } #[tokio::test] @@ -1870,6 +2073,7 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_clear }), RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: current_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -1948,6 +2152,7 @@ async fn record_initial_history_resumed_turn_context_after_compaction_reestablis let rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: previous_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -2088,6 +2293,7 @@ async fn record_initial_history_resumed_aborted_turn_without_id_clears_active_tu let rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: previous_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -2120,6 +2326,7 @@ async fn record_initial_history_resumed_aborted_turn_without_id_clears_active_tu )), RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: aborted_turn_id, root_turn_id: None, trace_id: None, @@ -2228,6 +2435,7 @@ async fn record_initial_history_resumed_unmatched_abort_preserves_active_turn_fo let rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: previous_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -2260,6 +2468,7 @@ async fn record_initial_history_resumed_unmatched_abort_preserves_active_turn_fo )), RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: current_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -2369,6 +2578,7 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_compaction_clea let rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: previous_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -2401,6 +2611,7 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_compaction_clea )), RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: incomplete_turn_id, root_turn_id: None, trace_id: None, @@ -2468,6 +2679,7 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_preserves_turn_ let rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: current_turn_id, root_turn_id: None, trace_id: None, @@ -2557,6 +2769,7 @@ async fn record_initial_history_resumed_replaced_incomplete_compacted_turn_clear let rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: previous_turn_id.clone(), root_turn_id: None, trace_id: None, @@ -2589,6 +2802,7 @@ async fn record_initial_history_resumed_replaced_incomplete_compacted_turn_clear )), RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: compacted_incomplete_turn_id, root_turn_id: None, trace_id: None, @@ -2625,6 +2839,7 @@ async fn record_initial_history_resumed_replaced_incomplete_compacted_turn_clear // completion/abort for the old one. RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: replacing_turn_id, root_turn_id: None, trace_id: None, diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index a9bcb0477d..81e5a2d050 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -4338,6 +4338,7 @@ async fn record_initial_history_forked_hydrates_previous_turn_settings() { let rollout_items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted( codex_protocol::protocol::TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.clone(), root_turn_id: None, trace_id: None, @@ -6134,6 +6135,18 @@ async fn compaction_persists_resume_metadata_and_companion_records() { let (mut session, turn_context) = make_session_and_context().await; let rollout_path = attach_thread_persistence(&mut session).await; let turn_context = Arc::new(turn_context); + turn_context + .turn_metadata_state + .set_turn_trigger("automation".to_string()); + turn_context + .turn_metadata_state + .set_parent_turn_id("parent-turn".to_string()); + turn_context + .turn_metadata_state + .set_root_turn_id("root-turn".to_string()); + turn_context.turn_metadata_state.set_initiating_agent_path( + codex_protocol::AgentPath::try_from("/root/requester").expect("requester path"), + ); let turn_context_baseline = turn_context.to_turn_context_item(); let world_state = Arc::new(build_world_state_from_turn_context(&session, &turn_context).await); let previous_turn_settings = PreviousTurnSettings { @@ -6146,11 +6159,19 @@ async fn compaction_persists_resume_metadata_and_companion_records() { .set_previous_turn_settings(Some(previous_turn_settings.clone())) .await; - session.state.lock().await.last_started_turn_id = Some("checkpoint-turn".into()); session.multi_agent_version = std::sync::OnceLock::from(MultiAgentVersion::V2); let expected = CompactionResumeMetadata { multi_agent_version: Some(MultiAgentVersion::V2), - last_started_turn_id: Some("checkpoint-turn".into()), + last_started_turn_id: Some(turn_context.sub_id.clone()), + turn_attribution: Some(codex_history::TurnAttribution { + turn_id: turn_context.sub_id.clone(), + turn_trigger: Some("automation".to_string()), + parent_turn_id: Some("parent-turn".to_string()), + initiating_agent_path: Some( + codex_protocol::AgentPath::try_from("/root/requester").expect("requester path"), + ), + root_turn_id: turn_context_baseline.root_turn_id.clone(), + }), previous_turn_settings: Some(previous_turn_settings), }; let expected_settings = codex_protocol::protocol::ThreadSettingsAppliedEvent { @@ -6158,6 +6179,21 @@ async fn compaction_persists_resume_metadata_and_companion_records() { thread_settings: session.thread_settings_snapshot().await, }; + session + .record_started_turn(&turn_context.sub_id, Some(turn_context.attribution())) + .await; + let compact_context = session.new_default_turn().await; + session + .record_started_turn(&compact_context.sub_id, /*attribution*/ None) + .await; + assert_eq!( + session.state.lock().await.turn_attribution, + expected.turn_attribution + ); + + session + .record_started_turn(&turn_context.sub_id, Some(turn_context.attribution())) + .await; let input_goal_ids = crate::context::UserGoalUpdate::message_ids(session.clone_history().await.raw_items()); // The goal edit is accepted after the compaction input was captured. @@ -6217,6 +6253,58 @@ async fn compaction_persists_resume_metadata_and_companion_records() { ); assert_eq!(saved_turn_context, &turn_context_baseline); assert_eq!(saved_settings, &expected_settings); + for same_turn_input in [false, true] { + let mut rollback_items = vec![RolloutItem::Compacted(compacted.clone())]; + if same_turn_input { + rollback_items.push(RolloutItem::ResponseItem( + user_message("steered work").into(), + )); + } + rollback_items.push(RolloutItem::EventMsg(EventMsg::ThreadRolledBack( + codex_protocol::protocol::ThreadRolledBackEvent { num_turns: 1 }, + ))); + let rolled_back = session + .reconstruct_history_from_rollout(&turn_context, &rollback_items) + .await; + assert_eq!(rolled_back.turn_attribution, None); + } + + // Settings invalidate continuation eligibility; standalone compaction replaces its ID. + // Neither changes the regular work whose attribution the checkpoint preserves. + for last_started_turn_id in [None, Some("compact-turn".to_string())] { + let mut checkpoint = compacted.clone(); + checkpoint + .resume_metadata + .as_mut() + .expect("resume metadata") + .last_started_turn_id = last_started_turn_id.clone(); + let mut items = vec![RolloutItem::Compacted(checkpoint)]; + if let Some(turn_id) = last_started_turn_id { + items.push(RolloutItem::EventMsg(EventMsg::TurnComplete( + TurnCompleteEvent { + turn_id, + last_agent_message: None, + error: None, + started_at: None, + completed_at: None, + duration_ms: None, + time_to_first_token_ms: None, + }, + ))); + } + let reconstructed = session + .reconstruct_history_from_rollout(&turn_context, &items) + .await; + assert_eq!(reconstructed.turn_attribution, expected.turn_attribution); + assert_eq!(reconstructed.reference_context_item, None); + items.push(RolloutItem::EventMsg(EventMsg::ThreadRolledBack( + codex_protocol::protocol::ThreadRolledBackEvent { num_turns: 1 }, + ))); + let rolled_back = session + .reconstruct_history_from_rollout(&turn_context, &items) + .await; + assert_eq!(rolled_back.turn_attribution, None); + } } #[tokio::test] diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index d58b7cd888..0c85aa0d7d 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -523,6 +523,16 @@ impl TurnContext { .strict_auto_review_enabled } + pub(crate) fn attribution(&self) -> codex_history::TurnAttribution { + codex_history::TurnAttribution { + turn_id: self.sub_id.clone(), + turn_trigger: self.turn_metadata_state.current_turn_trigger(), + parent_turn_id: self.turn_metadata_state.parent_turn_id(), + initiating_agent_path: self.turn_metadata_state.initiating_agent_path().cloned(), + root_turn_id: self.turn_metadata_state.root_turn_id(), + } + } + /// Captures current model metadata without preparing a step. pub(crate) fn capture_current_model_info(&self) -> Arc { Arc::clone(&self.next_step_settings.load().model_info) diff --git a/codex-rs/core/src/session/turn_input.rs b/codex-rs/core/src/session/turn_input.rs index f6e5c4895d..c1ce885c7c 100644 --- a/codex-rs/core/src/session/turn_input.rs +++ b/codex-rs/core/src/session/turn_input.rs @@ -138,6 +138,7 @@ impl PreparedTurnInputSettings { final_output_json_schema, service_tier, parent_turn_id, + initiating_agent_path, root_turn_id, cyber_access_program, } = self.start_options; @@ -183,6 +184,11 @@ impl PreparedTurnInputSettings { thread_settings::emit_applied(session, submission_id, settings_snapshot).await; } if let Some(parent_turn_id) = parent_turn_id { + if let Some(initiating_agent_path) = initiating_agent_path { + turn_context + .turn_metadata_state + .set_initiating_agent_path(initiating_agent_path); + } turn_context .turn_metadata_state .set_parent_turn_id(parent_turn_id); @@ -288,10 +294,7 @@ pub(super) async fn handle_recovery( } = thread_settings.into(); let request = TurnInputRequest::user_input(Vec::new()) .with_thread_settings(thread_settings) - .on_start(TurnStartOptions { - turn_trigger: Some("retry".to_string()), - ..start_options - }); + .on_start(start_options); let result = start_if_idle( session, WithTurnExtensionData { @@ -633,8 +636,16 @@ async fn steer( impl Session { /// Called under the active-turn lock before running any task or lifecycle callback. - pub(crate) async fn record_started_turn(&self, turn_id: &str) { - self.state.lock().await.last_started_turn_id = Some(turn_id.to_string()); + pub(crate) async fn record_started_turn( + &self, + turn_id: &str, + attribution: Option, + ) { + let mut state = self.state.lock().await; + state.last_started_turn_id = Some(turn_id.to_owned()); + if let Some(attribution) = attribution { + state.turn_attribution = Some(attribution); + } } pub(crate) async fn route_realtime_text_input( diff --git a/codex-rs/core/src/state/session.rs b/codex-rs/core/src/state/session.rs index 6a93e160a4..53857e1cc2 100644 --- a/codex-rs/core/src/state/session.rs +++ b/codex-rs/core/src/state/session.rs @@ -11,6 +11,7 @@ use super::AdditionalContextStore; use super::auto_compact_window::AutoCompactWindow; use super::auto_compact_window::AutoCompactWindowIds; use super::auto_compact_window::AutoCompactWindowSnapshot; +use crate::TurnStartOptions; use crate::context_manager::ContextManager; use crate::context_manager::HistoryReplacement; use crate::session::PreviousTurnSettings; @@ -18,6 +19,7 @@ use crate::session::session::SessionConfiguration; use crate::session::startup_prewarm::SessionStartupPrewarmHandle; use crate::session::time_reminder::CurrentTimeReminderState; use codex_history::ResponseItemEnvelope; +use codex_history::TurnAttribution; use codex_protocol::SessionId; use codex_protocol::ThreadId; use codex_protocol::protocol::RateLimitSnapshot; @@ -84,6 +86,8 @@ pub(crate) struct SessionState { /// Latest task admitted in this runtime, retained across completion and history edits. /// Cleared by standalone settings changes to invalidate pending continuation. pub(crate) last_started_turn_id: Option, + /// Latest regular turn, retained across completion, settings changes, and compaction. + pub(crate) turn_attribution: Option, /// Runtime accounting state for the active auto-compaction window. auto_compact_window: AutoCompactWindow, /// Original request effort for the current model while configuration updates remain active. @@ -129,6 +133,7 @@ impl SessionState { additional_context: AdditionalContextStore::default(), previous_turn_settings: None, last_started_turn_id: None, + turn_attribution: None, auto_compact_window: AutoCompactWindow::new_with_ids(auto_compact_window_ids), reasoning_effort_pin: ReasoningEffortPin::Unset, shutting_down: false, @@ -191,6 +196,7 @@ impl SessionState { .history .replace_compacted(items, reviewer_compaction_hash.as_deref()), HistoryReplacement::Reset => { + self.turn_attribution = None; self.history.replace_annotated(items); true } @@ -253,6 +259,21 @@ impl SessionState { self.history.reference_context_item() } + pub(crate) fn recovered_turn_start_options(&self, turn_id: &str) -> TurnStartOptions { + self.turn_attribution + .as_ref() + .filter(|attribution| attribution.turn_id == turn_id) + .map(TurnAttribution::start_options) + .unwrap_or_else(|| TurnStartOptions { + // Older rollouts only persisted the root on the model-context record. + root_turn_id: self + .reference_context_item() + .filter(|context| context.turn_id.as_deref() == Some(turn_id)) + .and_then(|context| context.root_turn_id), + ..Default::default() + }) + } + // Token/rate limit helpers pub(crate) fn update_token_info_from_usage( &mut self, diff --git a/codex-rs/core/src/tasks/compact.rs b/codex-rs/core/src/tasks/compact.rs index 62808c9e60..823b064fee 100644 --- a/codex-rs/core/src/tasks/compact.rs +++ b/codex-rs/core/src/tasks/compact.rs @@ -41,7 +41,7 @@ impl SessionTask for CompactTask { turn.id = %ctx.sub_id, ); // Preparation errors must reach the task runner, which reports them to the client. - session.emit_turn_started(&ctx).await; + session.emit_turn_started(&ctx, TaskKind::Compact).await; let step_context = session .capture_step_context(Arc::clone(&ctx), &cancellation_token) .await?; diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 5ace22c811..fc9272e911 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -315,7 +315,11 @@ impl Session { let (pending_items, _) = self.input_queue.drain_mailbox_input_items().await; let turn_state = { let mut active = self.active_turn.lock().await; - self.record_started_turn(&turn_context.sub_id).await; + self.record_started_turn( + &turn_context.sub_id, + (task_kind == TaskKind::Regular).then(|| turn_context.attribution()), + ) + .await; let turn = active.get_or_insert_with(ActiveTurn::default); debug_assert!(turn.task.is_none()); Arc::clone(&turn.turn_state) diff --git a/codex-rs/core/src/tasks/regular.rs b/codex-rs/core/src/tasks/regular.rs index bb8f00dc52..0eca40f3d4 100644 --- a/codex-rs/core/src/tasks/regular.rs +++ b/codex-rs/core/src/tasks/regular.rs @@ -48,7 +48,7 @@ impl SessionTask for RegularTask { // Regular turns emit `TurnStarted` inline so first-turn lifecycle does // not wait on startup prewarm resolution. let prewarmed_client_session = async { - sess.emit_turn_started(&ctx).await; + sess.emit_turn_started(&ctx, TaskKind::Regular).await; // Regular-start contributors run once, after the task is visible and interruptible. let prepares_mcp = sess .services diff --git a/codex-rs/core/src/tasks/user_shell.rs b/codex-rs/core/src/tasks/user_shell.rs index d4c76dc11c..f3915ed5c7 100644 --- a/codex-rs/core/src/tasks/user_shell.rs +++ b/codex-rs/core/src/tasks/user_shell.rs @@ -121,7 +121,9 @@ pub(crate) async fn execute_user_shell_command( // standalone lifecycle tasks (for example /shell, and review once it emits TurnStarted). // `/compact` is an intentional exception because compaction requests should not include // freshly reinjected context before the summary/replacement history is applied. - session.emit_turn_started(&turn_context).await; + session + .emit_turn_started(&turn_context, TaskKind::Regular) + .await; } let Some((turn_environment, environment_shell)) = turn_context diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 8a19d054e3..13e0b069bc 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -799,6 +799,7 @@ fn out_of_range_truncation_drops_pre_user_active_turn_prefix() { RolloutItem::ResponseItem(user_msg("u1").into()), RolloutItem::ResponseItem(assistant_msg("a1").into()), RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-2".to_string(), root_turn_id: None, trace_id: None, @@ -3125,6 +3126,7 @@ async fn interrupted_fork_snapshot_preserves_explicit_turn_id() { history_mode: Some(ThreadHistoryMode::Legacy), initial_history: InitialHistory::Forked(vec![ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "turn-explicit".to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/core/src/thread_rollout_truncation_tests.rs b/codex-rs/core/src/thread_rollout_truncation_tests.rs index 9625f74677..1495a9dfe5 100644 --- a/codex-rs/core/src/thread_rollout_truncation_tests.rs +++ b/codex-rs/core/src/thread_rollout_truncation_tests.rs @@ -43,6 +43,7 @@ fn assistant_msg(text: &str) -> ResponseItem { fn turn_started(turn_id: &str) -> RolloutItem { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/core/tests/suite/resume_warning.rs b/codex-rs/core/tests/suite/resume_warning.rs index fd53c8cfe2..14a443f834 100644 --- a/codex-rs/core/tests/suite/resume_warning.rs +++ b/codex-rs/core/tests/suite/resume_warning.rs @@ -62,6 +62,7 @@ fn resume_history( conversation_id: ThreadId::default(), history: Arc::new(vec![ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.clone(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/core/tests/suite/turn_input_submission.rs b/codex-rs/core/tests/suite/turn_input_submission.rs index 2376325027..4686dba102 100644 --- a/codex-rs/core/tests/suite/turn_input_submission.rs +++ b/codex-rs/core/tests/suite/turn_input_submission.rs @@ -28,6 +28,7 @@ use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::protocol::TurnEnvironmentSelections; +use codex_protocol::turn_input::TurnAttribution; use codex_protocol::user_input::UserInput; use core_test_support::responses; use core_test_support::responses::ev_completed; @@ -663,13 +664,318 @@ async fn recover_turn_if_idle_preserves_id_and_resumes_plan_mode() { .expect("recovered turn should include turn metadata"), ) .expect("recovered turn metadata should be valid JSON"); - assert_eq!(turn_metadata["turn_trigger"].as_str(), Some("retry")); + assert_eq!( + ( + turn_metadata.get("turn_trigger"), + turn_metadata.get("parent_turn_id"), + turn_metadata["root_turn_id"].as_str(), + ), + (None, None, Some(turn_id)) + ); 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("")); } +#[derive(Default)] +struct RecoveryLifecycleBarrier { + block_next: AtomicBool, + entered: tokio::sync::Notify, + block_stop: AtomicBool, + stop_entered: tokio::sync::Notify, + release_stop: tokio::sync::Notify, +} + +impl codex_extension_api::TurnLifecycleContributor for RecoveryLifecycleBarrier { + fn turn_start_phase( + &self, + _thread_store: &codex_extension_api::ExtensionData, + ) -> codex_extension_api::TurnStartPhase { + codex_extension_api::TurnStartPhase::RegularTaskStart + } + + fn on_turn_start<'a>( + &'a self, + _input: codex_extension_api::TurnStartInput<'a>, + ) -> codex_extension_api::ExtensionFuture<'a, ()> { + Box::pin(async move { + if self.block_next.swap(false, Ordering::SeqCst) { + self.entered.notify_one(); + std::future::pending::<()>().await; + } + }) + } + + fn on_turn_stop<'a>( + &'a self, + _input: codex_extension_api::TurnStopInput<'a>, + ) -> codex_extension_api::ExtensionFuture<'a, ()> { + Box::pin(async move { + if self.block_stop.swap(false, Ordering::SeqCst) { + self.stop_entered.notify_one(); + self.release_stop.notified().await; + } + }) + } +} + +enum RecoveryPause { + AfterContext, + BeforeContext, + AfterLateCompletion, + LegacyRollout, +} + +/// Cold recovery preserves attribution across repeated restarts, leaves unknown +/// triggers absent, and starts new turns with fresh attribution. +#[test_case(Some("automation"), RecoveryPause::AfterContext; "automation")] +#[test_case(None, RecoveryPause::AfterContext; "unknown_trigger")] +#[test_case(None, RecoveryPause::LegacyRollout; "legacy_root")] +#[test_case(Some("automation"), RecoveryPause::BeforeContext; "before_context_is_written")] +#[test_case(Some("composer"), RecoveryPause::AfterLateCompletion; "after_previous_turn_completes_late")] +#[tokio::test] +async fn recovery_preserves_durable_turn_attribution( + trigger: Option<&str>, + pause: RecoveryPause, +) -> anyhow::Result<()> { + let mut server = responses::start_mock_server().await; + responses::mount_sse_once(&server, responses::sse_completed("prior-user-turn")).await; + let barrier = Arc::new(RecoveryLifecycleBarrier::default()); + let mut extensions = ExtensionRegistryBuilder::new(); + extensions.turn_lifecycle_contributor(barrier.clone()); + let mut builder = test_codex().with_extensions(Arc::new(extensions.build())); + let mut test = builder.build_with_auto_env(&server).await?; + barrier.block_stop.store( + matches!(pause, RecoveryPause::AfterLateCompletion), + Ordering::SeqCst, + ); + let TurnInputSubmission::Started { + turn_id: previous_turn_id, + } = submit_user_message(&test.codex, "establish the initial context baseline").await? + else { + panic!("expected the initial turn"); + }; + if matches!(pause, RecoveryPause::AfterLateCompletion) { + timeout( + Duration::from_secs(/*secs*/ 10), + barrier.stop_entered.notified(), + ) + .await?; + } else { + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(event) if event.turn_id == previous_turn_id) + }) + .await; + } + let request_received = Arc::new(tokio::sync::Notify::new()); + if matches!(pause, RecoveryPause::BeforeContext) { + barrier.block_next.store(true, Ordering::SeqCst); + } else { + let notify = Arc::clone(&request_received); + responses::mount_response_once_match( + &server, + move |_: &wiremock::Request| { + notify.notify_one(); + true + }, + responses::sse_response(responses::sse_completed("initial")) + .set_delay(Duration::from_secs(/*secs*/ 60)), + ) + .await; + } + let parent = trigger.map(|_| "parent-turn"); + let initiating_agent_path = + parent.map(|_| AgentPath::root().join("requester").expect("requester path")); + let root = "root-turn"; + let request = if trigger == Some("automation") { + TurnInputRequest::user_input(Vec::new()) + } else { + user_message_request("recover this work") + }; + let request = request.on_start(TurnStartOptions { + turn_trigger: trigger.map(str::to_owned), + parent_turn_id: parent.map(str::to_owned), + root_turn_id: Some(root.to_owned()), + initiating_agent_path: initiating_agent_path.clone(), + ..Default::default() + }); + let turn_id = if matches!(pause, RecoveryPause::AfterLateCompletion) { + let TurnInputSubmission::Started { turn_id } = + test.codex.start_or_steer_turn(request).await? + else { + panic!("expected a new turn while the previous stop hook is pending"); + }; + turn_id + } else { + let StartIfIdleSubmission::Started { turn_id } = + test.codex.start_turn_if_idle(request).await? + else { + panic!("expected a new turn"); + }; + turn_id + }; + timeout(Duration::from_secs(/*secs*/ 10), async { + if matches!(pause, RecoveryPause::BeforeContext) { + barrier.entered.notified().await; + } else { + request_received.notified().await; + } + }) + .await?; + if matches!(pause, RecoveryPause::AfterLateCompletion) { + barrier.release_stop.notify_one(); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(event) if event.turn_id == previous_turn_id) + }) + .await; + } + let thread_settings = test.codex.restorable_thread_settings().await; + if let Some(url) = test.executor_environment().exec_server_url() { + builder = builder.with_exec_server_url(url); + } + + for attempt in 0..2 { + assert_eq!( + test.codex.suspend_turn_and_shutdown().await?, + codex_core::SuspendTurnOutcome::Suspended { + turn_id: turn_id.clone() + }, + ); + let rollout_path = test.codex.rollout_path().expect("durable rollout"); + test.thread_manager + .remove_thread(&test.session_configured.thread_id) + .await + .expect("suspended thread is loaded"); + if attempt == 0 && matches!(pause, RecoveryPause::LegacyRollout) { + let mut legacy_rollout = String::new(); + for line in std::fs::read_to_string(&rollout_path)?.lines() { + let mut item: Value = serde_json::from_str(line)?; + if let Some(payload) = item["payload"].as_object_mut() { + payload.remove("turn_attribution"); + } + legacy_rollout.push_str(&serde_json::to_string(&item)?); + legacy_rollout.push('\n'); + } + std::fs::write(&rollout_path, legacy_rollout)?; + } + server = responses::start_mock_server().await; + let notify = Arc::clone(&request_received); + let response = responses::mount_response_once_match( + &server, + move |_: &wiremock::Request| { + notify.notify_one(); + true + }, + responses::sse_response(responses::sse_completed("recovered")) + .set_delay(Duration::from_secs(/*secs*/ 60)), + ) + .await; + test = builder + .resume(&server, Arc::clone(&test.home), rollout_path) + .await?; + assert_eq!( + test.codex + .recover_turn_if_idle(RecoverTurnRequest { + turn_id: turn_id.clone(), + thread_settings: ThreadSettingsOverrides { + environments: thread_settings.environments.clone(), + ..Default::default() + }, + trace: None, + cyber_access_program: None, + }) + .await?, + StartIfIdleSubmission::Started { + turn_id: turn_id.clone() + }, + ); + let attribution = wait_for_event_match(&test.codex, |event| match event { + EventMsg::TurnStarted(event) if event.turn_id == turn_id => { + Some(event.turn_attribution.clone()) + } + _ => None, + }) + .await; + assert_eq!( + attribution, + Some(TurnAttribution { + turn_id: turn_id.clone(), + turn_trigger: trigger.map(str::to_owned), + parent_turn_id: parent.map(str::to_owned), + root_turn_id: Some(root.to_owned()), + initiating_agent_path: initiating_agent_path.clone(), + }), + ); + timeout( + Duration::from_secs(/*secs*/ 10), + request_received.notified(), + ) + .await?; + let request = response.single_request(); + let metadata: Value = serde_json::from_str( + &request + .header("x-codex-turn-metadata") + .expect("turn metadata"), + )?; + assert_eq!( + ( + metadata["turn_id"].as_str(), + metadata["turn_trigger"].as_str(), + metadata["parent_turn_id"].as_str(), + metadata["root_turn_id"].as_str(), + ), + (Some(turn_id.as_str()), trigger, parent, Some(root)), + "recovery attempt {attempt}", + ); + } + test.codex.submit(Op::Interrupt).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnAborted(_)) + }) + .await; + if trigger != Some("automation") { + return Ok(()); + } + let next_server = responses::start_mock_server().await; + test = builder.restart(&next_server, &test).await?; + test.codex.restore_thread_settings(thread_settings).await?; + let response = responses::mount_sse_once(&next_server, responses::sse_completed("new")).await; + let TurnInputSubmission::Started { + turn_id: next_turn_id, + } = test + .codex + .start_or_steer_turn( + user_message_request("new human work").on_start(TurnStartOptions { + turn_trigger: Some("composer".to_owned()), + ..Default::default() + }), + ) + .await? + else { + panic!("expected a new turn"); + }; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let request = response.single_request(); + let metadata: Value = serde_json::from_str( + &request + .header("x-codex-turn-metadata") + .expect("turn metadata"), + )?; + assert_eq!( + ( + metadata["turn_trigger"].as_str(), + metadata.get("parent_turn_id"), + metadata["root_turn_id"].as_str() + ), + (Some("composer"), None, Some(next_turn_id.as_str())), + ); + Ok(()) +} + /// Internal continuation creates a new turn without adding user authorization. #[tokio::test] async fn continue_turn_if_idle_starts_new_turn_with_internal_input() { diff --git a/codex-rs/external-agent-migration/src/sessions/export.rs b/codex-rs/external-agent-migration/src/sessions/export.rs index e3b360da3a..fbe2e37003 100644 --- a/codex-rs/external-agent-migration/src/sessions/export.rs +++ b/codex-rs/external-agent-migration/src/sessions/export.rs @@ -105,6 +105,7 @@ pub(super) fn rollout_items_from_messages(messages: Vec) -> let turn_id = format!("external-import-turn-{user_turn_count}"); items.push(RolloutItem::EventMsg(EventMsg::TurnStarted( TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.clone(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/history/src/compaction_resume_metadata.rs b/codex-rs/history/src/compaction_resume_metadata.rs index 4886fcd52b..bd8eb7a267 100644 --- a/codex-rs/history/src/compaction_resume_metadata.rs +++ b/codex-rs/history/src/compaction_resume_metadata.rs @@ -3,6 +3,7 @@ use crate::RolloutItem; use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::turn_input::CyberAccessProgram; +use codex_protocol::turn_input::TurnAttribution; use schemars::JsonSchema; use serde::Deserialize; use serde::Serialize; @@ -17,6 +18,9 @@ pub struct CompactionResumeMetadata { pub multi_agent_version: Option, /// Turn identity used to admit continuations after cold resume. pub last_started_turn_id: Option, + /// Attribution of the latest regular turn, independent of continuation eligibility. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub turn_attribution: Option, pub previous_turn_settings: Option, } diff --git a/codex-rs/history/src/lib.rs b/codex-rs/history/src/lib.rs index 5fb8de4290..41b99f95a0 100644 --- a/codex-rs/history/src/lib.rs +++ b/codex-rs/history/src/lib.rs @@ -6,6 +6,7 @@ pub use heartbeat::Heartbeat; pub use heartbeat::UserInputOrigin; mod compaction_resume_metadata; +pub use codex_protocol::turn_input::TurnAttribution; pub use compaction_resume_metadata::CompactionResumeMetadata; pub use compaction_resume_metadata::PreviousTurnSettings; pub use compaction_resume_metadata::resume_multi_agent_version; diff --git a/codex-rs/history/src/tests.rs b/codex-rs/history/src/tests.rs index 0b04ff6fd1..d021d687bb 100644 --- a/codex-rs/history/src/tests.rs +++ b/codex-rs/history/src/tests.rs @@ -580,6 +580,7 @@ fn compacted_resume_metadata_presence_round_trips_empty_values() -> Result<()> { let resume_metadata = CompactionResumeMetadata { multi_agent_version: None, last_started_turn_id: None, + turn_attribution: None, previous_turn_settings: None, }; let item = CompactedItem { @@ -690,6 +691,7 @@ fn compacted_metadata_remains_compatible_with_legacy_response_item_readers() -> resume_metadata: Some(CompactionResumeMetadata { multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: Some("turn-1".to_string()), + turn_attribution: None, previous_turn_settings: None, }), }))?; diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index e486d66649..6e6b64817a 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -2156,6 +2156,11 @@ pub struct TurnCompleteEvent { #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] pub struct TurnStartedEvent { + /// Provenance of a regular turn, persisted before startup work can be suspended. + /// Absent on older records and non-regular tasks. + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub turn_attribution: Option, pub turn_id: String, /// ID of the originating turn in the root thread; equals `turn_id` for root turns. #[serde(default, skip_serializing_if = "Option::is_none")] @@ -3280,8 +3285,7 @@ pub struct TurnContextNetworkItem { pub struct TurnContextItem { #[serde(default, skip_serializing_if = "Option::is_none")] pub turn_id: Option, - /// Root turn that owns this subagent turn's attribution. - /// Only set for subagent turns; persisted so resume keeps the scope frozen at turn start. + /// Root turn that owns this turn's attribution, retained across recovery. #[serde(default, skip_serializing_if = "Option::is_none")] pub root_turn_id: Option, /// Plugin selection captured for this turn. Absent in older histories. diff --git a/codex-rs/protocol/src/turn_input.rs b/codex-rs/protocol/src/turn_input.rs index cb27cd4f96..c145e0543d 100644 --- a/codex-rs/protocol/src/turn_input.rs +++ b/codex-rs/protocol/src/turn_input.rs @@ -1,5 +1,6 @@ //! Turn-input request and result types shared by Core's submission APIs. +use crate::AgentPath; use crate::models::ResponseItem; use crate::protocol::AdditionalContextEntry; use crate::protocol::InterAgentCommunication; @@ -52,6 +53,29 @@ pub struct TurnInputRequest { pub trace: Option, } +/// Attribution retained across recovery, sleep, and model-context compaction. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema, TS)] +pub struct TurnAttribution { + pub turn_id: String, + pub turn_trigger: Option, + pub parent_turn_id: Option, + pub initiating_agent_path: Option, + pub root_turn_id: Option, +} + +impl TurnAttribution { + /// Restores provenance without changing the thread's execution settings. + pub fn start_options(&self) -> TurnStartOptions { + TurnStartOptions { + turn_trigger: self.turn_trigger.clone(), + parent_turn_id: self.parent_turn_id.clone(), + initiating_agent_path: self.initiating_agent_path.clone(), + root_turn_id: self.root_turn_id.clone(), + ..Default::default() + } + } +} + /// Request to resume sampling for an interrupted regular turn. /// /// Sampling restarts under `turn_id`, which must be the ID already recorded @@ -171,6 +195,8 @@ pub struct TurnStartOptions { pub service_tier: Option, /// Parent turn lineage recorded if this request starts a new turn. pub parent_turn_id: Option, + /// Agent owning the parent turn, retained when recovering delegated work. + pub initiating_agent_path: Option, /// Causal root turn lineage recorded if this request starts a new turn. pub root_turn_id: Option, /// Explicit cyber treatment for this turn. Omission preserves the backend's diff --git a/codex-rs/rollout/src/persistence_metrics_tests.rs b/codex-rs/rollout/src/persistence_metrics_tests.rs index f6af5af10d..48b32f712c 100644 --- a/codex-rs/rollout/src/persistence_metrics_tests.rs +++ b/codex-rs/rollout/src/persistence_metrics_tests.rs @@ -49,6 +49,7 @@ fn retained_message(text: &str) -> RolloutItem { fn turn_started(turn_id: &str) -> RolloutItem { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index d7aa740270..e8dc915728 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -1164,6 +1164,7 @@ mod tests { live_thread .append_items(&[RolloutItem::EventMsg(EventMsg::TurnStarted( TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/thread-store/src/local/model_context_tests.rs b/codex-rs/thread-store/src/local/model_context_tests.rs index 7be08ce485..6b1fad47f8 100644 --- a/codex-rs/thread-store/src/local/model_context_tests.rs +++ b/codex-rs/thread-store/src/local/model_context_tests.rs @@ -219,6 +219,7 @@ async fn fork_version_stops_before_older_segments_once_resolved() { compacted.resume_metadata = Some(codex_rollout::CompactionResumeMetadata { multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: None, + turn_attribution: None, previous_turn_settings: None, }); append_items( @@ -616,6 +617,7 @@ fn append_items(path: &Path, items: impl IntoIterator) { fn turn_started(turn_id: &str) -> RolloutItem { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/thread-store/src/local/revert_thread_tests.rs b/codex-rs/thread-store/src/local/revert_thread_tests.rs index fd9ca13c0c..827bf8c107 100644 --- a/codex-rs/thread-store/src/local/revert_thread_tests.rs +++ b/codex-rs/thread-store/src/local/revert_thread_tests.rs @@ -321,6 +321,7 @@ async fn turn_ids(store: &LocalThreadStore, thread_id: ThreadId) -> Vec fn turn_started(turn_id: &str) -> RolloutItem { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs b/codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs index 1575bcf090..37c353fa08 100644 --- a/codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs +++ b/codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs @@ -345,6 +345,7 @@ impl LegacyRolloutCanonicalizer { writer, timestamp, RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id, root_turn_id: None, trace_id: None, diff --git a/codex-rs/thread-store/src/local/rollout_migration_tests.rs b/codex-rs/thread-store/src/local/rollout_migration_tests.rs index 90e2f11512..589acf7508 100644 --- a/codex-rs/thread-store/src/local/rollout_migration_tests.rs +++ b/codex-rs/thread-store/src/local/rollout_migration_tests.rs @@ -227,6 +227,7 @@ fn item_completed(turn_id: &str, item_id: &str) -> RolloutItem { fn started(turn_id: &str) -> RolloutItem { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs index b7c0658be6..8b691b6a2c 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs +++ b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs @@ -2832,6 +2832,7 @@ async fn create_paginated_subagent_thread( fn turn_started(turn_id: &str) -> RolloutItem { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/thread-store/src/thread_metadata_sync.rs b/codex-rs/thread-store/src/thread_metadata_sync.rs index cb3559c137..0351105fc1 100644 --- a/codex-rs/thread-store/src/thread_metadata_sync.rs +++ b/codex-rs/thread-store/src/thread_metadata_sync.rs @@ -712,6 +712,7 @@ mod tests { let update = sync .observe_appended_items(&[RolloutItem::EventMsg(EventMsg::TurnStarted( TurnStartedEvent { + turn_attribution: None, turn_id: "turn-1".to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index 89d722bd9a..8ed3f84c85 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -8406,6 +8406,7 @@ async fn prompt_edit_reverts_earlier_and_first_visible_prompts_in_place() -> Res ); for item in [ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.to_string(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/tui/src/app/tests/history_hydration_tests.rs b/codex-rs/tui/src/app/tests/history_hydration_tests.rs index 9a5d34fae6..bd976a3dd5 100644 --- a/codex-rs/tui/src/app/tests/history_hydration_tests.rs +++ b/codex-rs/tui/src/app/tests/history_hydration_tests.rs @@ -55,6 +55,7 @@ async fn history_fixture( for (index, count) in item_counts.iter().enumerate() { let turn_id = format!("turn-{index}"); let mut events = vec![EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.clone(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/tui/src/app/tests/pagination_completion_tests.rs b/codex-rs/tui/src/app/tests/pagination_completion_tests.rs index d3d98496c6..5698e5d2eb 100644 --- a/codex-rs/tui/src/app/tests/pagination_completion_tests.rs +++ b/codex-rs/tui/src/app/tests/pagination_completion_tests.rs @@ -57,6 +57,7 @@ pub(super) async fn completed_history_app( let turn_id = format!("turn-{index}"); let finished = completed_at.timestamp() + index as i64 * 60; let mut events = vec![EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: turn_id.clone(), root_turn_id: None, trace_id: None, diff --git a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs index cd366ae49f..2e518843b7 100644 --- a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs +++ b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs @@ -2320,6 +2320,7 @@ async fn older_pagination_reconciles_review_prompts_across_page_boundaries() -> user_item("newer-visible-prompt", "newer visible prompt"), ]); let events = std::iter::once(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "cross-page-review-turn".to_string(), root_turn_id: None, trace_id: None, @@ -2483,6 +2484,7 @@ async fn transcript_alt_beginning_loads_every_older_history_page() -> Result<()> .map(serde_json::from_str::) .collect::, _>>()?; let events = std::iter::once(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "multi-page-turn".to_string(), root_turn_id: None, trace_id: None, @@ -2898,6 +2900,7 @@ async fn underfilled_scrollback_fetches_older_pages_without_opening_the_transcri .map(serde_json::from_str::) .collect::, _>>()?; let events = std::iter::once(EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: "scrollback-pagination-turn".to_string(), root_turn_id: None, trace_id: None, @@ -4964,6 +4967,7 @@ async fn command_center_read_only_open_requests_and_failure_preservation() -> Re let first_ordinal = contents.lines().count(); let events = [ EventMsg::TurnStarted(TurnStartedEvent { + turn_attribution: None, turn_id: format!("saved-turn-{index}"), root_turn_id: None, trace_id: None,