From 7eed30ad1beaebc20b7d848ce9ee887525a379b0 Mon Sep 17 00:00:00 2001 From: Benjamin Carlsson Date: Wed, 14 Oct 2026 05:55:29 +0000 Subject: [PATCH] Preserve running turn recovery across thread settings updates (#53879) ## Why Standalone thread settings updates cleared the running turn's continuation lineage, preventing recovery after a daemon restart. ## What changed Preserve continuation lineage while a task is registered. Continue invalidating pending continuations for idle, reserved, and finishing turns, and record started-turn lineage atomically with task registration. ## Testing Extend settings invalidation coverage to idle, reserved, and finishing turns. Exercise an in-flight reasoning effort update before compaction, then verify cold resume preserves the turn's frozen settings and allows continuation. Wait for thread idle before testing idle continuation. GitOrigin-RevId: d7c316af6a4c871790bb57f0af9892bb30871ac8 --- codex-rs/core/src/session/tests.rs | 11 ++++- codex-rs/core/src/session/thread_settings.rs | 15 +++++- codex-rs/core/src/session/turn_input.rs | 2 +- codex-rs/core/src/tasks/mod.rs | 13 ++--- codex-rs/core/tests/suite/compact.rs | 47 ++++++++++++++++++- ...mpaction_cold_resume_after_completion.snap | 4 +- .../core/tests/suite/turn_input_submission.rs | 5 ++ 7 files changed, 83 insertions(+), 14 deletions(-) diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 14a88521ba..e18314e5cc 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -6134,12 +6134,21 @@ async fn mcp_attribution_checkpoints_cover_batch_prefixes_compaction_and_restore assert_eq!(restored.mcp_attribution_snapshot(), expected); } +#[test_case(None; "idle")] +#[test_case(Some(ActiveTurn::default()); "reserved turn")] +#[test_case(Some(ActiveTurn { + finishing: Some(tokio_util::sync::CancellationToken::new()), + ..Default::default() +}); "finishing turn")] #[tokio::test] -async fn standalone_settings_invalidate_continuation_before_delivering_acceptance() { +async fn standalone_settings_invalidate_continuation_before_delivering_acceptance( + active_turn: Option, +) { let (mut session, turn_context) = make_session_and_context().await; let (tx, rx) = async_channel::bounded(1); session.tx_event = tx; session.state.lock().await.last_started_turn_id = Some("superseded-turn".into()); + *session.active_turn.lock().await = active_turn; session .tx_event .send(Event { diff --git a/codex-rs/core/src/session/thread_settings.rs b/codex-rs/core/src/session/thread_settings.rs index c45ce098c6..f07d921dc2 100644 --- a/codex-rs/core/src/session/thread_settings.rs +++ b/codex-rs/core/src/session/thread_settings.rs @@ -31,14 +31,25 @@ impl Session { } /// Applies standalone thread settings. The caller holds the persistence permit through notification. +#[expect( + clippy::await_holding_invalid_type, + reason = "the active-turn guard makes continuation invalidation atomic with turn admission" +)] pub(super) async fn update( session: &Session, overrides: impl Into>, ) -> ConstraintResult { let updates = prepare_update(overrides, &session.services.selected_capability_roots); let commit = session.update_settings(updates).await?; - // Standalone settings changes supersede a pending automatic continuation. - session.state.lock().await.last_started_turn_id = None; + let active_turn = session.active_turn.lock().await; + // A registered running task retains its lineage so it can be recovered after a daemon + // restart. Reservations and finishing turns have no interrupted work to continue. + if active_turn + .as_ref() + .is_none_or(|active_turn| active_turn.task.is_none()) + { + session.state.lock().await.last_started_turn_id = None; + } Ok(commit.snapshot) } diff --git a/codex-rs/core/src/session/turn_input.rs b/codex-rs/core/src/session/turn_input.rs index 74a7592de8..d9cbe2f2e1 100644 --- a/codex-rs/core/src/session/turn_input.rs +++ b/codex-rs/core/src/session/turn_input.rs @@ -688,7 +688,7 @@ async fn steer( } impl Session { - /// Called under the active-turn lock before running any task or lifecycle callback. + /// Called under the active-turn lock when registering a task (or a realtime transcript turn). pub(crate) async fn record_started_turn( &self, turn_id: &str, diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 6a3e95ee80..3aa4c55b04 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -283,7 +283,7 @@ impl Session { #[expect( clippy::await_holding_invalid_type, - reason = "record the started turn atomically with its active reservation" + reason = "record the started turn atomically with task registration" )] pub(crate) async fn start_task( self: &Arc, @@ -328,11 +328,6 @@ 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, - (task_kind == TaskKind::Regular).then(|| turn_context.attribution()), - ) - .await; let turn = active.get_or_insert_with(ActiveTurn::default); debug_assert!(turn.task.is_none()); turn.finishing = None; @@ -352,6 +347,12 @@ impl Session { let mut active = self.active_turn.lock().await; let turn = active.get_or_insert_with(ActiveTurn::default); debug_assert!(turn.task.is_none()); + // Record lineage only when the task starts: settings changes can invalidate a reservation. + self.record_started_turn( + &turn_context.sub_id, + (task_kind == TaskKind::Regular).then(|| turn_context.attribution()), + ) + .await; let agent_execution_guard = self.services.agent_control.admit_turn( turn_context.multi_agent_version, &turn_context.session_source, diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index ee8b2e6985..93b183f472 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -22,6 +22,7 @@ use codex_protocol::models::PermissionProfile; use codex_protocol::openai_models::ModelInfo; use codex_protocol::openai_models::ModelServiceTier; use codex_protocol::openai_models::ModelsResponse; +use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::EventMsg; @@ -32,6 +33,8 @@ use codex_protocol::protocol::ItemStartedEvent; use codex_protocol::protocol::Op; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::protocol::WarningEvent; +use codex_protocol::request_user_input::RequestUserInputAnswer; +use codex_protocol::request_user_input::RequestUserInputResponse; use codex_protocol::user_input::UserInput; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_path_uri::PathUri; @@ -4690,7 +4693,18 @@ async fn paginated_compaction_cold_resume_from_bounded_suffix() -> Result<()> { &server, vec![ sse(vec![ - ev_function_call(DUMMY_CALL_ID, DUMMY_FUNCTION_NAME, "{}"), + ev_function_call( + DUMMY_CALL_ID, + "request_user_input", + &json!({"questions": [{ + "id": "next", "header": "Next", "question": "Continue?", + "options": [ + {"label": "Yes", "description": "Continue."}, + {"label": "No", "description": "Stop."} + ] + }]}) + .to_string(), + ), ev_completed_with_tokens("before-compact", /*total_tokens*/ 96), ]), sse(vec![ @@ -4744,6 +4758,8 @@ async fn paginated_compaction_cold_resume_from_bounded_suffix() -> Result<()> { }, ) .await?; + // The resumed turn retains its frozen settings, even after an in-flight update. + let expected_settings = test.codex.thread_settings_snapshot().await; let codex_core::TurnInputSubmission::Started { turn_id, .. } = test .codex .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { @@ -4754,11 +4770,38 @@ async fn paginated_compaction_cold_resume_from_bounded_suffix() -> Result<()> { else { panic!("initial turn did not start"); }; + // Change settings while the turn is blocked, before compaction checkpoints its lineage. + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::RequestUserInput(_)) + }) + .await; + core_test_support::submit_thread_settings( + &test.codex, + ThreadSettingsOverrides { + effort: Some(Some(ReasoningEffort::High)), + ..Default::default() + }, + ) + .await?; + test.codex.checkpoint_thread_settings().await?; + test.codex + .submit(Op::UserInputAnswer { + id: turn_id.clone(), + response: RequestUserInputResponse { + answers: [( + "next".into(), + RequestUserInputAnswer { + answers: vec!["Yes".into()], + }, + )] + .into(), + }, + }) + .await?; wait_for_event(&test.codex, |event| { matches!(event, EventMsg::TurnComplete(_)) }) .await; - let expected_settings = test.codex.thread_settings_snapshot().await; let path = test.codex.rollout_path().expect("rollout path"); test.codex.shutdown_and_wait().await?; test.thread_manager diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__compact__compaction_cold_resume_after_completion.snap b/codex-rs/core/tests/suite/snapshots/all__suite__compact__compaction_cold_resume_after_completion.snap index 2477eb2974..93d6715ad1 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__compact__compaction_cold_resume_after_completion.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__compact__compaction_cold_resume_after_completion.snap @@ -63,8 +63,8 @@ Scenario: Real paginated mid-turn compaction and cold resume from the bounded su 04:message/user: function call limit push -05:function_call/test_tool:{} -06:function_call_output:unsupported call: test_tool +05:function_call/request_user_input:{"questions":[{"header":"Next","id":"next","options":[{"description":"Continue.","label":"Yes"},{"description":"Stop.","label":"No"}],"question":"Continue?"}]} +06:function_call_output:{"answers":{"next":{"answers":["Yes"]}}} 07:message/user: diff --git a/codex-rs/core/tests/suite/turn_input_submission.rs b/codex-rs/core/tests/suite/turn_input_submission.rs index 6e3628c6e6..2dfa42279c 100644 --- a/codex-rs/core/tests/suite/turn_input_submission.rs +++ b/codex-rs/core/tests/suite/turn_input_submission.rs @@ -997,8 +997,11 @@ async fn recovery_preserves_durable_turn_attribution( #[tokio::test] async fn continue_turn_if_idle_starts_new_turn_with_internal_input() { let server = responses::start_mock_server().await; + let mut extensions = ExtensionRegistryBuilder::new(); + extensions.thread_lifecycle_contributor(Arc::new(ThreadIdle)); let test = test_codex() .with_model("gpt-5.6-sol") + .with_extensions(Arc::new(extensions.build())) .with_config(|config| { config .features @@ -1027,6 +1030,7 @@ async fn continue_turn_if_idle_starts_new_turn_with_internal_input() { matches!(event, EventMsg::TurnComplete(_)) }) .await; + ThreadIdle::wait(&test.codex).await; let mock = responses::mount_sse_once(&server, responses::sse_completed("continued")).await; let input = ContextualUserFragment::into(InternalModelContextFragment::new( InternalContextSource::from_static("daemon_recovery"), @@ -1058,6 +1062,7 @@ async fn continue_turn_if_idle_starts_new_turn_with_internal_input() { matches!(event, EventMsg::TurnComplete(_)) }) .await; + ThreadIdle::wait(&test.codex).await; let request = mock.single_request(); assert!(request.has_content_kinds(&["daemon_recovery.internal_context"])); let metadata: Value = serde_json::from_str(