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 10fadc9e7f..c70eda9e66 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -3274,11 +3274,11 @@ async fn thread_goal_keeps_original_root_until_external_objective_edit() -> Resu ); responses::assert_root_turn(&reopened_request, Some(original_turn.turn.id.as_str()))?; let continuation_request = serde_json::from_slice::(&requests[7])?; - assert_ne!( - continuation_request["client_metadata"]["turn_id"].as_str(), - Some(edited_turn_id) - ); - responses::assert_root_turn(&continuation_request, /*expected*/ None)?; + let continuation_turn_id = continuation_request["client_metadata"]["turn_id"] + .as_str() + .expect("independent continuation turn ID"); + assert_ne!(continuation_turn_id, edited_turn_id); + responses::assert_root_turn(&continuation_request, Some(continuation_turn_id))?; responses::assert_parent_turn(&continuation_request, /*expected*/ None)?; server.shutdown().await; @@ -3446,7 +3446,7 @@ async fn thread_goal_lifecycle_emits_analytics_and_clear_deletes_goal() -> Resul goal_request_body["client_metadata"]["turn_id"], causal_turn_id ); - responses::assert_root_turn(&goal_request_body, /*expected*/ None)?; + responses::assert_root_turn(&goal_request_body, Some(causal_turn_id))?; responses::assert_parent_turn(&goal_request_body, /*expected*/ None)?; let clear_id = mcp diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index 397276ab3c..e567bceb4a 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -386,9 +386,9 @@ async fn turn_start_with_empty_input_runs_model_request() -> Result<()> { assert_eq!( ( event["event_params"]["turn_id"].as_str(), - event["event_params"].get("root_turn_id"), + event["event_params"]["root_turn_id"].as_str(), ), - (Some(turn.id.as_str()), Some(&Value::Null)) + (Some(turn.id.as_str()), Some(turn.id.as_str())) ); let requests = server diff --git a/codex-rs/core/src/responses_metadata.rs b/codex-rs/core/src/responses_metadata.rs index 6f64929ec8..8edfbdfc7a 100644 --- a/codex-rs/core/src/responses_metadata.rs +++ b/codex-rs/core/src/responses_metadata.rs @@ -168,7 +168,7 @@ impl CodexResponsesRequestKind { } } - fn has_turn_identity(self) -> bool { + fn has_thread_identity(self) -> bool { !matches!(self, CodexResponsesRequestKind::Memory) } } @@ -380,20 +380,18 @@ impl CodexResponsesMetadata { let (request_kind, compaction) = request_kind.metadata(); (Some(request_kind), compaction) }); - let has_turn_identity = - request_kind.is_none_or(CodexResponsesRequestKind::has_turn_identity); + let has_thread_identity = + request_kind.is_none_or(CodexResponsesRequestKind::has_thread_identity); let has_request_identity = - request_kind.is_some_and(CodexResponsesRequestKind::has_turn_identity); + request_kind.is_some_and(CodexResponsesRequestKind::has_thread_identity); CodexTurnMetadataPayload { installation_id: has_request_identity.then_some(self.installation_id.as_str()), - session_id: has_turn_identity.then_some(self.session_id.as_str()), - thread_id: has_turn_identity.then_some(self.thread_id.as_str()), - agent_name: has_turn_identity + session_id: has_thread_identity.then_some(self.session_id.as_str()), + thread_id: has_thread_identity.then_some(self.thread_id.as_str()), + agent_name: has_thread_identity .then_some(self.agent_name.as_deref()) .flatten(), - turn_id: has_turn_identity - .then_some(self.turn_id.as_deref()) - .flatten(), + turn_id: self.turn_id.as_deref(), window_id: has_request_identity.then_some(self.window_id.as_str()), window_number: has_request_identity.then_some(self.window_number).flatten(), context_window_id: has_request_identity diff --git a/codex-rs/core/src/session/input_queue.rs b/codex-rs/core/src/session/input_queue.rs index 94d3dfeb4d..3efdb90b0c 100644 --- a/codex-rs/core/src/session/input_queue.rs +++ b/codex-rs/core/src/session/input_queue.rs @@ -292,14 +292,10 @@ impl InputQueue { &self, active_turn: &Mutex>, ) -> (Vec, TurnStartOptions) { - let (pending_input, accepts_mailbox_delivery, active_turn_metadata) = { + let (pending_input, accepts_mailbox_delivery) = { let mut active = active_turn.lock().await; match active.as_mut() { Some(active_turn) => { - let active_turn_metadata = active_turn - .task - .as_ref() - .map(|task| Arc::clone(&task.turn_context.turn_metadata_state)); let mut turn_state = active_turn.turn_state.lock().await; let accepts_mailbox_delivery = turn_state.accepts_mailbox_delivery_for_current_turn(); @@ -308,25 +304,15 @@ impl InputQueue { } else { Vec::new() }; - ( - pending_input, - accepts_mailbox_delivery, - active_turn_metadata, - ) + (pending_input, accepts_mailbox_delivery) } - None => (Vec::new(), true, None), + None => (Vec::new(), true), } }; if !accepts_mailbox_delivery { return (pending_input, TurnStartOptions::default()); } let (mailbox_items, start_options) = self.drain_mailbox_input_items().await; - if let Some(active_turn_metadata) = active_turn_metadata - && active_turn_metadata.root_turn_id().is_none() - && let Some(root_turn_id) = start_options.root_turn_id.as_ref() - { - active_turn_metadata.set_root_turn_id(root_turn_id.clone()); - } if pending_input.is_empty() { (mailbox_items, start_options) } else { diff --git a/codex-rs/core/src/session/review.rs b/codex-rs/core/src/session/review.rs index 202aae4aca..695662914a 100644 --- a/codex-rs/core/src/session/review.rs +++ b/codex-rs/core/src/session/review.rs @@ -130,9 +130,6 @@ pub(super) async fn spawn_review_thread( auto_review_enabled, &model_info, )); - if turn_metadata_state.can_start_root_turn(&session_source) { - turn_metadata_state.set_root_turn_id(review_turn_id.clone()); - } let extension_data = Arc::new(codex_extension_api::ExtensionData::new( review_turn_id.clone(), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 0847c59ce2..79dc4f069f 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -12053,20 +12053,14 @@ async fn trigger_turn_mailbox_mail_waits_for_next_turn_after_answer_boundary() { assert!(sess.input_queue.has_trigger_turn_mailbox_items().await); } +#[test_case(None; "independent root")] +#[test_case(Some("root-a"); "inherited root")] #[tokio::test] -async fn active_turn_keeps_first_root_when_mail_coalesces() { +async fn active_turn_keeps_first_root_when_mail_coalesces(inherited_root: Option<&str>) { let (sess, tc, _rx) = make_session_and_context_with_rx().await; - tc.turn_metadata_state - .set_root_turn_id("root-a".to_string()); - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: true, - }, - ) - .await; + if let Some(root) = inherited_root { + tc.turn_metadata_state.set_root_turn_id(root.to_string()); + } let first = InterAgentCommunication::new( AgentPath::try_from("/root/worker_a").expect("worker path should parse"), AgentPath::root(), @@ -12081,10 +12075,13 @@ async fn active_turn_keeps_first_root_when_mail_coalesces() { "second".to_string(), /*trigger_turn*/ true, ); - for (communication, parent_turn_id, root_turn_id) in [ + for (index, (communication, parent_turn_id, root_turn_id)) in [ (first.clone(), "parent-a", "root-a"), (second.clone(), "parent-b", "root-b"), - ] { + ] + .into_iter() + .enumerate() + { sess.input_queue .enqueue_mailbox_communication( communication, @@ -12095,6 +12092,19 @@ async fn active_turn_keeps_first_root_when_mail_coalesces() { }, ) .await; + if index == 0 { + // The first message is already queued when this independent task + // starts; the second arrives after its root is established. + sess.spawn_task( + Arc::clone(&tc), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }, + ) + .await; + } } assert_eq!( @@ -12106,7 +12116,7 @@ async fn active_turn_keeps_first_root_when_mail_coalesces() { ); assert_eq!( tc.turn_metadata_state.root_turn_id().as_deref(), - Some("root-a") + Some(inherited_root.unwrap_or(&tc.sub_id)) ); assert!(!sess.input_queue.has_pending_mailbox_items().await); diff --git a/codex-rs/core/src/session/turn_input.rs b/codex-rs/core/src/session/turn_input.rs index fe7b5ce32b..4bb8a16cfc 100644 --- a/codex-rs/core/src/session/turn_input.rs +++ b/codex-rs/core/src/session/turn_input.rs @@ -262,11 +262,6 @@ async fn start_or_steer( )); } }; - let can_start_root_turn = start.parent_turn_id.is_none() && start.root_turn_id.is_none(); - let incoming_root_turn_id = start - .parent_turn_id - .as_ref() - .map(|_| start.root_turn_id.clone()); let settings = PreparedTurnInputSettings::prepare(session, thread_settings, start).await?; match session .steer_input( @@ -275,7 +270,6 @@ async fn start_or_steer( /*expected_turn_id*/ None, settings.required_active_final_output_json_schema(), responsesapi_client_metadata.clone(), - incoming_root_turn_id, ) .await { @@ -290,16 +284,6 @@ async fn start_or_steer( else { unreachable!("explicit user input can enter Plan mode"); }; - if can_start_root_turn - && has_explicit_input - && turn_context - .turn_metadata_state - .can_start_root_turn(&turn_context.session_source) - { - turn_context - .turn_metadata_state - .set_root_turn_id(submission_id.clone()); - } if let Some(responsesapi_client_metadata) = responsesapi_client_metadata { turn_context .turn_metadata_state @@ -340,7 +324,6 @@ async fn start_if_idle( responsesapi_client_metadata, .. } = request; - let can_start_root_turn = start.parent_turn_id.is_none() && start.root_turn_id.is_none(); if session.input_queue.has_trigger_turn_mailbox_items().await { return Ok(TurnInputSubmission::NotSubmitted { reason: NotSubmittedReason::PendingTriggerTurn, @@ -403,16 +386,6 @@ async fn start_if_idle( .turn_metadata_state .set_responsesapi_client_metadata(responsesapi_client_metadata); } - if kind == TurnStartKind::User - && can_start_root_turn - && turn_context - .turn_metadata_state - .can_start_root_turn(&turn_context.session_source) - { - turn_context - .turn_metadata_state - .set_root_turn_id(submission_id.clone()); - } session .maybe_emit_model_warnings_for_turn(turn_context.as_ref()) .await; @@ -469,10 +442,6 @@ async fn steer( "only user input can steer a turn".to_string(), )); } - let incoming_root_turn_id = start - .parent_turn_id - .as_ref() - .map(|_| start.root_turn_id.clone()); let settings = PreparedTurnInputSettings::prepare(session, thread_settings, start).await?; match session .steer_input( @@ -481,7 +450,6 @@ async fn steer( Some(expected_turn_id.as_str()), settings.required_active_final_output_json_schema(), responsesapi_client_metadata, - incoming_root_turn_id, ) .await { @@ -557,7 +525,6 @@ impl Session { expected_turn_id: Option<&str>, required_final_output_json_schema: Option<&Value>, responsesapi_client_metadata: Option>, - incoming_root_turn_id: Option>, ) -> Result { let mut active = self.active_turn.lock().await; let Some(active_turn) = active.as_mut() else { @@ -627,18 +594,6 @@ impl Session { input => pending_turn_input(self, input.clone()).await, }; pending_input.push(input); - if active_task - .turn_context - .turn_metadata_state - .root_turn_id() - .is_none() - && let Some(Some(incoming_root_turn_id)) = incoming_root_turn_id - { - active_task - .turn_context - .turn_metadata_state - .set_root_turn_id(incoming_root_turn_id); - } self.input_queue .extend_pending_input_and_accept_mailbox_delivery_for_turn_state( active_turn.turn_state.as_ref(), diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 9cf988a295..9344f8262f 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -285,6 +285,11 @@ impl Session { input: Vec, task: T, ) { + // Inherited or recovered roots are applied before task start. Otherwise this + // task owns its turn, including background work. Later mail cannot change it. + turn_context + .turn_metadata_state + .set_root_turn_id(turn_context.sub_id.clone()); let task: Arc = Arc::new(task); let task_kind = task.kind(); let span_name = task.span_name(); @@ -307,14 +312,7 @@ impl Session { .await .clear_turn(&turn_context.sub_id); - let (pending_items, start_options) = self.input_queue.drain_mailbox_input_items().await; - if turn_context.turn_metadata_state.root_turn_id().is_none() - && let Some(root_turn_id) = start_options.root_turn_id - { - turn_context - .turn_metadata_state - .set_root_turn_id(root_turn_id); - } + let (pending_items, _) = self.input_queue.drain_mailbox_input_items().await; let turn_state = { let mut active = self.active_turn.lock().await; let turn = active.get_or_insert_with(ActiveTurn::default); diff --git a/codex-rs/core/src/turn_metadata.rs b/codex-rs/core/src/turn_metadata.rs index 8d3114845c..681475b86e 100644 --- a/codex-rs/core/src/turn_metadata.rs +++ b/codex-rs/core/src/turn_metadata.rs @@ -93,7 +93,10 @@ pub async fn detached_memory_responses_metadata( permission_profile: &PermissionProfile, sandbox: Option<&str>, ) -> CodexResponsesMetadata { + let turn_id = uuid::Uuid::now_v7().to_string(); let mut metadata = CodexResponsesMetadata { + turn_id: Some(turn_id.clone()), + root_turn_id: Some(turn_id), request_kind: Some(CodexResponsesRequestKind::Memory), thread_source: Some(ThreadSource::MemoryConsolidation), subagent_header: subagent_header_value(session_source), @@ -340,24 +343,6 @@ impl TurnMetadataState { self.root_turn_id.get().cloned() } - pub(crate) fn can_start_root_turn(&self, session_source: &SessionSource) -> bool { - if session_source.is_non_root_agent() { - return false; - } - match &self.thread_source { - // Desktop create/fork/send lacks trusted app-server provenance; fail closed. - Some( - ThreadSource::Subagent - | ThreadSource::GuardianReview - | ThreadSource::MemoryConsolidation, - ) => false, - Some(ThreadSource::Feature(feature)) => { - !matches!(feature.as_str(), "system" | "title") && !feature.starts_with("ambient") - } - Some(ThreadSource::User) | None => true, - } - } - pub(crate) fn set_responsesapi_client_metadata( &self, responsesapi_client_metadata: HashMap, diff --git a/codex-rs/core/src/turn_metadata_tests.rs b/codex-rs/core/src/turn_metadata_tests.rs index 247b01c47f..fb6c118cc3 100644 --- a/codex-rs/core/src/turn_metadata_tests.rs +++ b/codex-rs/core/src/turn_metadata_tests.rs @@ -139,7 +139,7 @@ async fn wait_for_git_enrichment(state: &TurnMetadataState) -> Value { } #[tokio::test] -async fn detached_memory_responses_metadata_omits_turn_identity() { +async fn detached_memory_responses_metadata_starts_an_independent_root_turn() { let (_temp_dir, repo_path) = create_clean_git_repo("repo-東京").await; let thread_manager = crate::ThreadManager::with_models_provider_for_tests( @@ -147,7 +147,7 @@ async fn detached_memory_responses_metadata_omits_turn_identity() { crate::config::test_config().await.model_provider, ); - let header = detached_memory_responses_metadata( + let metadata = detached_memory_responses_metadata( &thread_manager, String::new(), String::new(), @@ -158,9 +158,8 @@ async fn detached_memory_responses_metadata_omits_turn_identity() { &PermissionProfile::read_only(), Some("none"), ) - .await - .turn_metadata_json() - .expect("header"); + .await; + let header = metadata.turn_metadata_json().expect("header"); assert!(header.is_ascii()); assert!(!header.contains("東京")); let parsed: Value = serde_json::from_str(&header).expect("valid json"); @@ -173,8 +172,18 @@ async fn detached_memory_responses_metadata_omits_turn_identity() { assert!(parsed.get("session_id").is_none()); assert!(parsed.get("thread_id").is_none()); assert!(parsed.get("forked_from_thread_id").is_none()); - assert!(parsed.get("turn_id").is_none()); - assert!(parsed.get(ROOT_TURN_ID_KEY).is_none()); + let turn_id = parsed["turn_id"].as_str().expect("memory turn ID"); + uuid::Uuid::parse_str(turn_id).expect("memory turn ID is a UUID"); + assert_eq!(parsed[ROOT_TURN_ID_KEY], parsed["turn_id"]); + let client_metadata = metadata.client_metadata(); + assert_eq!( + client_metadata.get("turn_id").map(String::as_str), + Some(turn_id) + ); + assert_eq!( + client_metadata.get(ROOT_TURN_ID_KEY).map(String::as_str), + Some(turn_id) + ); assert!(parsed.get(WINDOW_ID_KEY).is_none()); let expected_repo_path = repo_path.to_string_lossy().into_owned(); @@ -221,10 +230,14 @@ async fn detached_memory_responses_metadata_omits_empty_workspace_metadata() { .turn_metadata_json() .expect("detached memory should emit its request kind"); let parsed: Value = serde_json::from_str(&header).expect("valid json"); + let turn_id = parsed["turn_id"].as_str().expect("memory turn ID"); + uuid::Uuid::parse_str(turn_id).expect("memory turn ID is a UUID"); assert_eq!( parsed, serde_json::json!({ + "turn_id": turn_id, + "root_turn_id": turn_id, "request_kind": "memory", "sandbox_mode": "read-only", "thread_source": "memory_consolidation", diff --git a/codex-rs/core/tests/suite/compact_remote.rs b/codex-rs/core/tests/suite/compact_remote.rs index cfaf139ddd..7de3e92541 100644 --- a/codex-rs/core/tests/suite/compact_remote.rs +++ b/codex-rs/core/tests/suite/compact_remote.rs @@ -1259,7 +1259,7 @@ async fn assert_remote_manual_compact_request_parity( .await?; wait_for_turn_complete(&codex).await; - codex.submit(Op::Compact).await?; + let compact_turn_id = codex.submit(Op::Compact).await?; wait_for_turn_complete(&codex).await; let response_requests = responses_mock.requests(); @@ -1280,6 +1280,13 @@ async fn assert_remote_manual_compact_request_parity( let compact_request = compact_mock.single_request(); let normal_body = normal_request.body_json(); let compact_body = compact_request.body_json(); + let compact_metadata: Value = serde_json::from_str( + &compact_request + .header("x-codex-turn-metadata") + .expect("compaction turn metadata"), + )?; + assert_eq!(compact_metadata["turn_id"], compact_turn_id); + assert_eq!(compact_metadata["root_turn_id"], compact_turn_id); let expected_routing_hint = |body: &Value| { if !uses_codex_backend { return None; diff --git a/codex-rs/core/tests/suite/git_enrichment.rs b/codex-rs/core/tests/suite/git_enrichment.rs index 989683fe82..5286068b10 100644 --- a/codex-rs/core/tests/suite/git_enrichment.rs +++ b/codex-rs/core/tests/suite/git_enrichment.rs @@ -329,6 +329,9 @@ async fn guardian_prewarm_and_review_skip_redundant_git_enrichment() -> Result<( #[test_case("system"; "system background thread")] #[test_case("ambient_background"; "ambient background thread")] +#[test_case("ambient_suggestions"; "ambient suggestions")] +#[test_case("ambient_suggestion_safety"; "ambient suggestion safety")] +#[test_case("title"; "title background thread")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn ephemeral_system_thread_prewarm_skips_and_turn_observes_fresh_state( thread_source: &str, @@ -388,7 +391,7 @@ async fn ephemeral_system_thread_prewarm_skips_and_turn_observes_fresh_state( assert!(turn_metadata(&prewarm)?.get("workspaces").is_none()); std::fs::write(repo.path().join("untracked.txt"), "dirty\n")?; - system_thread + let submission = system_thread .thread .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { text: "generate a thread title".into(), @@ -405,7 +408,10 @@ async fn ephemeral_system_thread_prewarm_skips_and_turn_observes_fresh_state( .and_then(|connection| connection.get(2)) .context("system turn follow-up request")? .body_json(); - assert_root_turn(&turn, /*expected*/ None)?; + let codex_core::TurnInputSubmission::Started { turn_id } = submission else { + panic!("background input should start a turn"); + }; + assert_root_turn(&turn, Some(&turn_id))?; assert_eq!( turn_metadata(&turn)?["workspaces"], expected_workspace(repo.path(), &head, /*has_changes*/ true) diff --git a/codex-rs/core/tests/suite/pending_input.rs b/codex-rs/core/tests/suite/pending_input.rs index c7ea6fe73b..f1086c7e41 100644 --- a/codex-rs/core/tests/suite/pending_input.rs +++ b/codex-rs/core/tests/suite/pending_input.rs @@ -86,12 +86,14 @@ async fn idle_response_items_include_pending_mailbox_in_first_request() -> anyho responses::user_message_item("automatic response item"), ))) .await?; - assert!(matches!(submission, StartIfIdleSubmission::Started { .. })); + let StartIfIdleSubmission::Started { turn_id } = submission else { + panic!("automatic input should start a turn"); + }; wait_for_turn_complete(test.codex.as_ref()).await; let request = response.single_request(); let request_body = request.body_json(); - responses::assert_root_turn(&request_body, /*expected*/ None)?; + responses::assert_root_turn(&request_body, Some(&turn_id))?; responses::assert_parent_turn(&request_body, /*expected*/ None)?; let user_messages = request.message_input_texts("user"); assert!( diff --git a/codex-rs/memories/write/src/startup_tests.rs b/codex-rs/memories/write/src/startup_tests.rs index 53aa77793c..4771d42029 100644 --- a/codex-rs/memories/write/src/startup_tests.rs +++ b/codex-rs/memories/write/src/startup_tests.rs @@ -670,7 +670,18 @@ async fn memories_startup_phase1_uses_live_thread_service_tier_and_detached_meta assert_eq!(metadata["sandbox_mode"].as_str(), Some("workspace-write")); assert!(metadata.get("session_id").is_none()); assert!(metadata.get("thread_id").is_none()); - assert!(metadata.get("turn_id").is_none()); + let turn_id = metadata["turn_id"].as_str().expect("memory turn ID"); + uuid::Uuid::parse_str(turn_id).expect("memory turn ID is a UUID"); + assert_eq!(metadata["root_turn_id"], metadata["turn_id"]); + let request_body = request.body_json(); + assert_eq!( + request_body["client_metadata"]["turn_id"], + metadata["turn_id"] + ); + assert_eq!( + request_body["client_metadata"]["root_turn_id"], + metadata["root_turn_id"] + ); assert!(metadata.get("window_id").is_none()); assert!(metadata.get("workspaces").is_some());