github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 16:17:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Preserve running turn recovery across thread settings updates (#53879)

Browse Source 
## 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
This commit is contained in:
Benjamin Carlsson
2026-10-14 06:03:02 +00:00
committed by copyberry
parent cb03f28a40
commit 7eed30ad1b
7 changed files with 83 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/session/tests.rs
+10 -1
View File
@@ -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<ActiveTurn>,
) {
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 {
codex-rs/core/src/session/thread_settings.rs
+13 -2
View File
@@ -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<WithTurnExtensionData<ThreadSettingsOverrides>>,
) -> ConstraintResult<ThreadSettingsSnapshot> {
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)
}
codex-rs/core/src/session/turn_input.rs
+1 -1
View File
@@ -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,
codex-rs/core/src/tasks/mod.rs
+7 -6
View File
@@ -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<T: SessionTask>(
self: &Arc<Self>,
@@ -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,
codex-rs/core/tests/suite/compact.rs
+45 -2
View File
@@ -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
codex-rs/core/tests/suite/snapshots/all__suite__compact__compaction_cold_resume_after_completion.snap
+2 -2
View File
@@ -63,8 +63,8 @@ Scenario: Real paginated mid-turn compaction and cold resume from the bounded su
<MULTI_AGENT_MODE>
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:
<SUMMARIZATION_PROMPT>
codex-rs/core/tests/suite/turn_input_submission.rs
+5
View File
@@ -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(