diff --git a/codex-rs/app-server-transport/src/daemon_recovery.rs b/codex-rs/app-server-transport/src/daemon_recovery.rs index 9b3b590d18..c19c51d8a8 100644 --- a/codex-rs/app-server-transport/src/daemon_recovery.rs +++ b/codex-rs/app-server-transport/src/daemon_recovery.rs @@ -22,6 +22,9 @@ pub struct InterruptedTurn { pub output_schema: Option, pub service_tier: Option, pub cyber_access_program: Option, + /// Only local execution with thread-owned configuration can continue automatically. + /// Older snapshots without this identity are reloaded without continuation. + pub local_environment: Option, } // Old servers accept the array and skip this non-thread entry during best-effort diff --git a/codex-rs/app-server/src/daemon_thread_recovery.rs b/codex-rs/app-server/src/daemon_thread_recovery.rs index 7460cc0499..ade73c616e 100644 --- a/codex-rs/app-server/src/daemon_thread_recovery.rs +++ b/codex-rs/app-server/src/daemon_thread_recovery.rs @@ -34,7 +34,7 @@ pub(crate) async fn start_recovery( processor: std::sync::Arc, ) -> io::Result> { let candidates = tokio::task::spawn_blocking(move || { - let candidates = daemon_recovery::read_candidates(&path); + let candidates = daemon_recovery::read_snapshot(&path); // Even malformed or temporarily unreadable snapshots belong to this generation only. match std::fs::remove_file(&path) { Ok(()) => candidates, diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 17b6d2d4e5..7c2e274b36 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -765,18 +765,18 @@ impl MessageProcessor { pub(crate) async fn restore_daemon_threads( &self, - candidates: std::collections::BTreeSet, + mut snapshot: codex_app_server_transport::daemon_recovery::RecoverySnapshot, ) { - for thread_id in candidates { + for thread_id in snapshot.loaded { let Ok(_permit) = self.turn_admission.admit() else { break; }; if let Err(err) = self .thread_processor .thread_resume( - ThreadResumeTarget::DaemonRecovery, + ThreadResumeTarget::DaemonRecovery(snapshot.interrupted.remove(&thread_id)), codex_app_server_protocol::ThreadResumeParams { - thread_id, + thread_id: thread_id.clone(), exclude_turns: true, ..Default::default() }, diff --git a/codex-rs/app-server/src/request_processors/daemon_continuation.rs b/codex-rs/app-server/src/request_processors/daemon_continuation.rs new file mode 100644 index 0000000000..04e05693d9 --- /dev/null +++ b/codex-rs/app-server/src/request_processors/daemon_continuation.rs @@ -0,0 +1,135 @@ +//! Attempts one new continuation turn immediately after daemon thread restoration. +//! Completed or superseded work is discarded; recovery never waits for a later idle event. +//! Permissions and the saved local executor selection must match the resumed runtime. + +use super::ThreadRequestProcessor; +use codex_app_server_protocol::ThreadEnvironment; +use codex_app_server_transport::daemon_recovery::InterruptedTurn; +use codex_core::TurnInput; +use codex_core::TurnInputRequest; +use codex_core::TurnInputSubmission; +use codex_core::TurnStartOptions; +use codex_core::context::ContextualUserFragment; +use codex_core::context::InternalContextSource; +use codex_core::context::InternalModelContextFragment; +use codex_protocol::ThreadId; +use codex_protocol::protocol::EnvironmentConfigState; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::TurnAbortReason; +use codex_protocol::protocol::TurnAbortedEvent; +use codex_rollout::RolloutItem; + +impl ThreadRequestProcessor { + pub(crate) async fn continue_daemon_turn(&self, thread_id: &str, saved: InterruptedTurn) { + let Ok(thread_id) = ThreadId::from_string(thread_id) else { + return; + }; + let Ok(thread) = self.thread_manager.get_thread(thread_id).await else { + return; + }; + let Some(path) = thread.rollout_path() else { + return; + }; + let history = match codex_rollout::RolloutRecorder::load_rollout_items(&path).await { + Ok((history, _, _)) => history, + Err(err) => { + tracing::warn!(%thread_id, %err, "failed to read interrupted turn history"); + return; + } + }; + // The snapshot has complete recorded input, but the old process may have + // completed or replaced its turn during the shutdown grace period. + let Some(start) = history + .iter() + .rposition(|item| matches!(item, RolloutItem::EventMsg(EventMsg::TurnStarted(_)))) + else { + return; + }; + if !matches!(&history[start], RolloutItem::EventMsg(EventMsg::TurnStarted(event)) + if event.turn_id == saved.turn_id) + || history[start..].iter().any(|item| match item { + RolloutItem::EventMsg(EventMsg::TurnComplete(event)) => { + event.turn_id == saved.turn_id + } + RolloutItem::EventMsg(EventMsg::TurnAborted(event)) => event + .turn_id + .as_ref() + .is_none_or(|turn_id| turn_id == &saved.turn_id), + _ => false, + }) + { + return; + } + let previous_context = history.iter().rev().find_map(|item| match item { + RolloutItem::TurnContext(context) + if context.turn_id.as_ref() == Some(&saved.turn_id) => + { + Some(context) + } + _ => None, + }); + let Some(previous_context) = previous_context else { + return; + }; + let config = thread.config_snapshot().await; + let [environment] = config.environment_selections() else { + return; + }; + if environment.environment_id != codex_exec_server::LOCAL_ENVIRONMENT_ID + || environment.config != EnvironmentConfigState::FromThread + || saved.local_environment.as_ref() != Some(&ThreadEnvironment::from(environment)) + { + return; + } + // Recovery must not override a stricter saved or newly configured policy. + if previous_context.permission_profile() != config.permission_profile { + return; + } + let continuation = ContextualUserFragment::into(InternalModelContextFragment::new( + InternalContextSource::from_static("daemon_recovery"), + "The server restarted and interrupted the previous turn. Continue the unfinished work from the saved conversation. Check the current state before repeating actions that may already have completed.", + )); + let request = TurnInputRequest::new(TurnInput::ResponseItem(continuation)).on_start( + TurnStartOptions { + turn_trigger: Some("daemon_recovery".to_string()), + final_output_json_schema: saved.output_schema, + service_tier: saved.service_tier, + cyber_access_program: saved.cyber_access_program, + root_turn_id: previous_context.root_turn_id.clone(), + ..Default::default() + }, + ); + // Close the old turn in persisted history before exposing a new running turn. + if let Err(err) = thread + .append_rollout_items(&[RolloutItem::EventMsg(EventMsg::TurnAborted( + TurnAbortedEvent { + turn_id: Some(saved.turn_id.clone()), + reason: TurnAbortReason::Interrupted, + started_at: None, + completed_at: None, + duration_ms: None, + }, + ))]) + .await + { + tracing::warn!(%thread_id, %err, "failed to close interrupted turn history"); + return; + } + match thread.continue_turn_if_idle(request, saved.turn_id).await { + Ok(TurnInputSubmission::Started { .. }) => { + crate::extensions::send_thread_warning( + &self.outgoing, + &self.thread_state_manager, + thread_id, + "Resuming interrupted work".to_string(), + ) + .await; + } + Ok(TurnInputSubmission::NotSubmitted { reason }) => { + tracing::debug!(%thread_id, ?reason, "recovery continuation was not started"); + } + Ok(TurnInputSubmission::Steered { .. }) => unreachable!("continuation cannot steer"), + Err(err) => tracing::warn!(%thread_id, %err, "recovery continuation failed"), + } + } +} diff --git a/codex-rs/app-server/src/request_processors/daemon_snapshot.rs b/codex-rs/app-server/src/request_processors/daemon_snapshot.rs index e1238f4c06..038373f3a1 100644 --- a/codex-rs/app-server/src/request_processors/daemon_snapshot.rs +++ b/codex-rs/app-server/src/request_processors/daemon_snapshot.rs @@ -37,7 +37,7 @@ impl ThreadRequestProcessor { continue; } snapshot.loaded.insert(thread_id.to_string()); - if let Some((turn_id, options)) = interrupted { + if let Some((turn_id, options, environment)) = interrupted { snapshot.interrupted.insert( thread_id.to_string(), InterruptedTurn { @@ -45,6 +45,7 @@ impl ThreadRequestProcessor { output_schema: options.final_output_json_schema, service_tier: options.service_tier, cyber_access_program: options.cyber_access_program, + local_environment: Some((&environment).into()), }, ); } diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 04c2229652..e8e2c3ab0a 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -1,3 +1,6 @@ +#[path = "daemon_continuation.rs"] +mod daemon_continuation; + #[path = "daemon_snapshot.rs"] mod daemon_snapshot; @@ -460,7 +463,7 @@ pub(crate) struct ThreadRequestProcessor { /// Whether resume attaches a client or restores a cold runtime during daemon startup. pub(crate) enum ThreadResumeTarget { Client(ConnectionRequestId), - DaemonRecovery, + DaemonRecovery(Option), } /// Outcome of trying to satisfy a resume request from an already loaded thread. @@ -3630,11 +3633,15 @@ impl ThreadRequestProcessor { } RunningThreadResumeResult::NotRunning(stored_thread) => stored_thread, }, - ThreadResumeTarget::DaemonRecovery => { + ThreadResumeTarget::DaemonRecovery(saved) => { let thread_id = ThreadId::from_string(¶ms.thread_id) .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; // Recheck under the same permit as client resume, including after config loading. if self.thread_manager.get_thread(thread_id).await.is_ok() { + if let Some(saved) = saved { + self.continue_daemon_turn(¶ms.thread_id, saved.clone()) + .await; + } return Ok(ControlFlow::Break(())); } None @@ -3918,7 +3925,7 @@ impl ThreadRequestProcessor { ThreadResumeTarget::Client(request_id) => { self.request_trace_context(request_id).await } - ThreadResumeTarget::DaemonRecovery => None, + ThreadResumeTarget::DaemonRecovery(_) => None, }, client_mcp_extensions, ) @@ -3939,6 +3946,10 @@ impl ThreadRequestProcessor { codex_thread .emit_thread_idle_lifecycle_if_idle(ThreadIdleCause::Completed) .await; + if let ThreadResumeTarget::DaemonRecovery(Some(saved)) = target { + self.continue_daemon_turn(&thread_id.to_string(), saved.clone()) + .await; + } let state = self.thread_state_manager.thread_state(thread_id).await; self.ensure_listener_task_running(thread_id, Arc::clone(&codex_thread), state) .await?; diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index 1de7f03f1b..142d8a70f6 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -1150,7 +1150,9 @@ impl TurnRequestProcessor { None, None, ), - NotSubmittedReason::PendingTriggerTurn | NotSubmittedReason::PlanMode => ( + NotSubmittedReason::PendingTriggerTurn + | NotSubmittedReason::PlanMode + | NotSubmittedReason::Superseded => ( "no active turn to steer".to_string(), None, Some(AnalyticsJsonRpcError::TurnSteer( diff --git a/codex-rs/app-server/tests/suite/v2/daemon_update_recovery.rs b/codex-rs/app-server/tests/suite/v2/daemon_update_recovery.rs index 1aefa8f132..019d327bbb 100644 --- a/codex-rs/app-server/tests/suite/v2/daemon_update_recovery.rs +++ b/codex-rs/app-server/tests/suite/v2/daemon_update_recovery.rs @@ -460,6 +460,11 @@ async fn assert_still_running(server: &mut Child, message: &str) { } #[test_case::test_case("running")] +#[test_case::test_case("plan")] +#[test_case::test_case("read_only")] +#[test_case::test_case("stricter_config")] +#[test_case::test_case("different_environment")] +#[test_case::test_case("legacy_snapshot")] #[test_case::test_case("completed")] #[test_case::test_case("canceled")] #[test_case::test_case("compacting")] @@ -467,7 +472,7 @@ async fn assert_still_running(server: &mut Child, message: &str) { async fn managed_shutdown_records_interrupted_turn(outcome: &str) -> Result<()> { let home = TempDir::new()?; let (release, gate) = oneshot::channel(); - let (_release_compaction, compaction_gate) = oneshot::channel(); + let (release_compaction, compaction_gate) = oneshot::channel(); let (mock, _completions) = start_streaming_sse_server(vec![ vec![stream_chunk(Some(gate), "Original")?], vec![stream_chunk(Some(compaction_gate), "Compacted")?], @@ -476,16 +481,36 @@ async fn managed_shutdown_records_interrupted_turn(outcome: &str) -> Result<()> create_config_toml(home.path(), mock.uri(), "never")?; let socket = home.path().join("control/server.sock"); let mut server = spawn_server(home.path(), &socket)?; - let mut client = connect_default_daemon_client(&socket).await?; + let mut client = connect_daemon_client( + &socket, + InitializeCapabilities { + experimental_api: true, + ..Default::default() + }, + ) + .await?; let thread = start_thread(&mut client, /*id*/ 2, json!({})).await?; + let local_environment = thread + .thread + .environments + .as_ref() + .and_then(|environments| environments.first()) + .cloned(); let id = thread.thread.id; + let mode = if outcome == "plan" { "plan" } else { "default" }; let schema = json!({"type":"object","properties":{},"additionalProperties":false}); let turn = request( &mut client, /*id*/ 3, "turn/start", json!({ - "threadId":id,"input":[{"type":"text","text":"Do the work"}],"outputSchema":schema + "threadId":id,"input":[{"type":"text","text":"Do the work"}],"outputSchema":schema, + "sandboxPolicy": match outcome { + "read_only" => json!({"type":"readOnly"}), + "stricter_config" => json!({"type":"workspaceWrite","writableRoots":[],"networkAccess":false,"excludeTmpdirEnvVar":false,"excludeSlashTmp":false}), + _ => serde_json::Value::Null, + }, + "collaborationMode":{"mode":mode,"settings":{"model":"mock-model","reasoning_effort":null,"developer_instructions":null}} }), ) .await?; @@ -527,7 +552,12 @@ async fn managed_shutdown_records_interrupted_turn(outcome: &str) -> Result<()> .await?; wait_for_requests(&mock, /*count*/ 2).await?; } - "running" => {} + "running" + | "plan" + | "read_only" + | "stricter_config" + | "different_environment" + | "legacy_snapshot" => {} _ => unreachable!(), } request_shutdown(&server, &socket).await?; @@ -538,14 +568,16 @@ async fn managed_shutdown_records_interrupted_turn(outcome: &str) -> Result<()> } }) .await?; - let expected = if outcome == "running" { + let interrupted = !matches!(outcome, "completed" | "canceled" | "compacting"); + let expected = if interrupted { [( id.clone(), daemon_recovery::InterruptedTurn { turn_id: turn["turn"]["id"].as_str().context("turn ID")?.to_string(), - output_schema: Some(schema), + output_schema: Some(schema.clone()), service_tier: Some("default".into()), cyber_access_program: None, + local_environment, }, )] .into() @@ -560,11 +592,168 @@ async fn managed_shutdown_records_interrupted_turn(outcome: &str) -> Result<()> } ); // Existing binaries can still read the candidate array and ignore the metadata entry. - let legacy: std::collections::BTreeSet = serde_json::from_slice(&std::fs::read(path)?)?; + let legacy: std::collections::BTreeSet = + serde_json::from_slice(&std::fs::read(&path)?)?; assert!(legacy.contains(&id)); if server.try_wait()?.is_none() { server.kill().await?; } + if !interrupted { + return Ok(()); + } + if outcome == "running" { + // A prior recovery can append an older turn's abort after this turn started. + codex_rollout::append_rollout_item_to_path( + thread.thread.path.as_ref().context("rollout path")?, + &serde_json::from_value(json!({ + "type":"event_msg", + "payload":{"type":"turn_aborted","turn_id":"older-turn","reason":"interrupted"} + }))?, + ) + .await?; + } + if outcome == "read_only" { + let config = home.path().join("config.toml"); + let contents = std::fs::read_to_string(&config)?; + std::fs::write( + config, + contents.replace( + "sandbox_mode = \"read-only\"", + "default_permissions = \":workspace\"", + ), + )?; + } + if matches!(outcome, "different_environment" | "legacy_snapshot") { + let mut snapshot = daemon_recovery::read_snapshot(&path)?; + let saved = snapshot + .interrupted + .get_mut(&id) + .context("interrupted turn")?; + if outcome == "legacy_snapshot" { + saved.local_environment = None; + } else { + saved + .local_environment + .as_mut() + .context("local environment")? + .environment_id = "remote".into(); + } + daemon_recovery::write_snapshot(&path, &snapshot)?; + } + let mut successor = spawn_server(home.path(), &socket)?; + if !matches!(outcome, "running" | "plan") { + let mut client = connect_default_daemon_client(&socket).await?; + timeout(DEFAULT_READ_TIMEOUT, async { + loop { + let loaded = + request(&mut client, /*id*/ 2, "thread/loaded/list", json!({})).await?; + if loaded["data"] + .as_array() + .context("loaded IDs")? + .contains(&json!(id)) + { + break; + } + sleep(Duration::from_millis(25)).await; + } + Ok::<(), anyhow::Error>(()) + }) + .await??; + // Resume serializes with the ongoing daemon restore, so the recovery attempt has finished. + request( + &mut client, + /*id*/ 3, + "thread/resume", + json!({"threadId":id,"excludeTurns":true}), + ) + .await?; + let response = request( + &mut client, + /*id*/ 2, + "thread/read", + json!({"threadId":id,"includeTurns":true}), + ) + .await?; + assert_eq!(response["thread"]["status"]["type"], "idle"); + assert_eq!(mock.requests().await.len(), 1); + successor.kill().await?; + return Ok(()); + } + // Recovery starts without a client, then its internal message survives reconnect. + wait_for_requests(&mock, /*count*/ 2).await?; + let mut client = connect_default_daemon_client(&socket).await?; + request( + &mut client, + /*id*/ 2, + "thread/resume", + json!({"threadId":id,"excludeTurns":true}), + ) + .await?; + let response = request( + &mut client, + /*id*/ 3, + "thread/read", + json!({"threadId":id,"includeTurns":true}), + ) + .await?; + let statuses: Vec<_> = response["thread"]["turns"] + .as_array() + .context("turns")? + .iter() + .filter(|turn| turn["id"] != "older-turn") + .map(|turn| turn["status"].clone()) + .collect(); + assert_eq!(statuses, vec![json!("interrupted"), json!("inProgress")]); + release_compaction + .send(()) + .expect("continuation is waiting"); + timeout(DEFAULT_READ_TIMEOUT, async { + loop { + let response = request( + &mut client, + /*id*/ 3, + "thread/read", + json!({"threadId":id,"includeTurns":true}), + ) + .await?; + let turns = response["thread"]["turns"].as_array().context("turns")?; + if let Some(continued) = turns.last() + && continued["status"] == "completed" + { + assert_ne!(continued["id"], turn["turn"]["id"]); + assert!( + continued["items"] + .as_array() + .context("items")? + .iter() + .all(|item| item["type"] != "userMessage") + ); + break; + } + sleep(Duration::from_millis(25)).await; + } + Ok::<(), anyhow::Error>(()) + }) + .await??; + let requests = mock.requests().await; + assert_eq!(requests.len(), 2); + let body: serde_json::Value = serde_json::from_slice(&requests[1])?; + assert_eq!(body["text"]["format"]["schema"], schema); + let input = body["input"].to_string(); + assert!( + input.contains("Do the work") && input.contains("Continue the unfinished work"), + "{input}" + ); + let collaboration = input + .rsplit("") + .next() + .context("collaboration instructions")?; + assert_eq!( + collaboration.contains("# Plan Mode"), + mode == "plan", + "{input}" + ); + successor.kill().await?; Ok(()) } diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index a2f4e9bf63..d2bbbe84f2 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -357,6 +357,23 @@ impl CodexThread { } } + /// Starts a new internal continuation turn when idle, including in Plan mode. + /// Rejects if a newer task has started, even if it has already finished. + /// The input must be a response item; it is never treated as user authorization. + pub async fn continue_turn_if_idle( + &self, + request: TurnInputRequest, + expected_previous_turn_id: String, + ) -> CodexResult { + self.submit_turn_input_with_mode( + request, + TurnInputMode::ContinueIfIdle { + expected_previous_turn_id, + }, + ) + .await + } + /// Resumes an interrupted regular turn only when the thread is idle. /// /// Recovery starts no new user input and preserves the turn ID that was @@ -498,7 +515,9 @@ impl CodexThread { } /// Captures a regular turn only after its input is recorded. The caller must flush the rollout. - pub async fn interrupted_turn(&self) -> Option<(String, TurnStartOptions)> { + pub async fn interrupted_turn( + &self, + ) -> Option<(String, TurnStartOptions, TurnEnvironmentSelection)> { self.session.interrupted_turn().await } diff --git a/codex-rs/core/src/session/daemon_recovery.rs b/codex-rs/core/src/session/daemon_recovery.rs index 8e12ee026b..1c8fa947d6 100644 --- a/codex-rs/core/src/session/daemon_recovery.rs +++ b/codex-rs/core/src/session/daemon_recovery.rs @@ -1,14 +1,18 @@ -//! Captures recorded, uncanceled regular turns for managed daemon recovery. +//! Captures recorded, uncanceled regular turns using one thread-configured local executor. //! Callers must flush the rollout after capture before persisting the snapshot. use super::Session; +use codex_protocol::protocol::EnvironmentConfigState; +use codex_protocol::protocol::TurnEnvironmentSelection; /// Turn input and turn-start injections have entered the rollout writer. pub(super) struct RecordedTurnInput; impl Session { /// Captures a regular turn only after its input is recorded. The caller must flush the rollout. - pub(crate) async fn interrupted_turn(&self) -> Option<(String, crate::TurnStartOptions)> { + pub(crate) async fn interrupted_turn( + &self, + ) -> Option<(String, crate::TurnStartOptions, TurnEnvironmentSelection)> { let active = self.active_turn.lock().await; let task = active.as_ref()?.task.as_ref()?; if task.kind != crate::state::TaskKind::Regular || task.cancellation_token.is_cancelled() { @@ -16,6 +20,14 @@ impl Session { } let context = &task.turn_context; context.extension_data.get::()?; + // Remote identities/configuration are not persisted across daemon restarts. + if context.environments.environments.len() != 1 { + return None; + } + let environment = context.environments.single_local_environment()?.selection(); + if environment.config != EnvironmentConfigState::FromThread { + return None; + } Some(( context.sub_id.clone(), crate::TurnStartOptions { @@ -34,6 +46,7 @@ impl Session { cyber_access_program: context.cyber_access_program, ..Default::default() }, + environment, )) } } diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 97e067c759..e5e96e32b4 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1445,6 +1445,12 @@ impl Session { &session_configuration.session_source, ), ); + state.last_started_turn_id = initial_history.get_rollout_items().iter().rev().find_map(|item| { + match item { + RolloutItem::EventMsg(EventMsg::TurnStarted(event)) => Some(event.turn_id.clone()), + _ => None, + } + }); state.base_instructions_provenance = base_instructions_provenance.clone(); state.active_disabled_plugin_ids = session_configuration.disabled_plugin_ids.clone(); let managed_network_requirements_configured = config diff --git a/codex-rs/core/src/session/thread_settings.rs b/codex-rs/core/src/session/thread_settings.rs index 2cc34cca8e..c0831ed4ba 100644 --- a/codex-rs/core/src/session/thread_settings.rs +++ b/codex-rs/core/src/session/thread_settings.rs @@ -50,6 +50,9 @@ pub(super) async fn update( }), }) .await; + } else { + // Standalone settings changes supersede a pending automatic continuation. + session.state.lock().await.last_started_turn_id = None; } } diff --git a/codex-rs/core/src/session/turn_input.rs b/codex-rs/core/src/session/turn_input.rs index d08dcd62a9..ba366c67e1 100644 --- a/codex-rs/core/src/session/turn_input.rs +++ b/codex-rs/core/src/session/turn_input.rs @@ -216,7 +216,31 @@ pub(super) async fn handle( | SubmittedTurnInput::ResponseItem(_) | SubmittedTurnInput::InterAgentCommunication(_) => TurnStartKind::Automatic, }; - start_if_idle(session, request, submission_id, kind).await + start_if_idle( + session, + request, + submission_id, + kind, + /*expected_previous_turn_id*/ None, + ) + .await + } + TurnInputMode::ContinueIfIdle { + expected_previous_turn_id, + } => { + if !matches!(&request.input, SubmittedTurnInput::ResponseItem(_)) { + return Err(CodexErr::InvalidRequest( + "continuation requires internal response input".to_string(), + )); + } + start_if_idle( + session, + request, + submission_id, + TurnStartKind::Recovery, + Some(expected_previous_turn_id), + ) + .await } TurnInputMode::Steer { expected_turn_id } => { steer(session, request, expected_turn_id, submission_id).await @@ -236,7 +260,14 @@ pub(super) async fn handle_recovery( turn_trigger: Some("retry".to_string()), ..start_options }); - start_if_idle(session, request, submission_id, TurnStartKind::Recovery).await + start_if_idle( + session, + request, + submission_id, + TurnStartKind::Recovery, + /*expected_previous_turn_id*/ None, + ) + .await } async fn start_or_steer( @@ -335,11 +366,16 @@ async fn start_or_steer( } } +#[expect( + clippy::await_holding_invalid_type, + reason = "the previous turn check and idle reservation must be atomic" +)] async fn start_if_idle( session: &Arc, request: TurnInputRequest, submission_id: String, kind: TurnStartKind, + expected_previous_turn_id: Option, ) -> CodexResult { let TurnInputRequest { input, @@ -392,6 +428,13 @@ async fn start_if_idle( reason: NotSubmittedReason::NotIdle, }); } + if let Some(expected) = expected_previous_turn_id + && session.state.lock().await.last_started_turn_id.as_ref() != Some(&expected) + { + return Ok(TurnInputSubmission::NotSubmitted { + reason: NotSubmittedReason::Superseded, + }); + } let active_turn = active_turn.get_or_insert_with(ActiveTurn::default); Arc::clone(&active_turn.turn_state) }; @@ -445,7 +488,7 @@ async fn start_if_idle( } task_input.push(pending_turn_input(session, input).await); } - TurnStartKind::Automatic => { + TurnStartKind::Automatic | TurnStartKind::Recovery => { // Empty automatic user input resumes sampling without a new message. if !matches!(&input, SubmittedTurnInput::UserInput { .. }) { session @@ -457,9 +500,6 @@ async fn start_if_idle( .await; } } - TurnStartKind::Recovery => { - // Recovery resumes an existing turn without a new empty user message. - } } session .start_task(turn_context, task_input, RegularTask::new()) @@ -508,6 +548,11 @@ 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 route_realtime_text_input( self: &Arc, text: String, diff --git a/codex-rs/core/src/state/session.rs b/codex-rs/core/src/state/session.rs index ebb9fac454..e3ddd2cc24 100644 --- a/codex-rs/core/src/state/session.rs +++ b/codex-rs/core/src/state/session.rs @@ -84,6 +84,9 @@ pub(crate) struct SessionState { /// model/realtime handling on subsequent regular turns (including full-context /// reinjection after resume or `/compact`). previous_turn_settings: Option, + /// 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, /// 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. @@ -127,6 +130,7 @@ impl SessionState { mcp_dependency_prompted: HashSet::new(), additional_context: AdditionalContextStore::default(), previous_turn_settings: None, + last_started_turn_id: None, auto_compact_window: AutoCompactWindow::new_with_ids(auto_compact_window_ids), reasoning_effort_pin: ReasoningEffortPin::Unset, startup_prewarm: None, diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 4c5cd6dfd5..5bae1532d8 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -278,6 +278,10 @@ impl Session { self.start_task(turn_context, input, task).await; } + #[expect( + clippy::await_holding_invalid_type, + reason = "record the started turn atomically with its active reservation" + )] pub(crate) async fn start_task( self: &Arc, turn_context: Arc, @@ -309,6 +313,7 @@ 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; 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/tests/suite/turn_input_submission.rs b/codex-rs/core/tests/suite/turn_input_submission.rs index ff1d56c003..07b6855c33 100644 --- a/codex-rs/core/tests/suite/turn_input_submission.rs +++ b/codex-rs/core/tests/suite/turn_input_submission.rs @@ -8,6 +8,9 @@ use codex_core::TurnInputRequest; use codex_core::TurnInputSubmission; use codex_core::TurnStartOptions; use codex_core::config::Constrained; +use codex_core::context::ContextualUserFragment; +use codex_core::context::InternalContextSource; +use codex_core::context::InternalModelContextFragment; use codex_extension_api::ExtensionRegistryBuilder; use codex_extension_api::TurnStartAdmission; use codex_protocol::AgentPath; @@ -516,6 +519,125 @@ async fn recover_turn_if_idle_preserves_id_and_resumes_plan_mode() { assert!(user_input_groups[0][0].starts_with("")); } +/// Internal continuation creates a new turn without adding user authorization. +#[tokio::test] +async fn continue_turn_if_idle_starts_new_turn_with_internal_input() { + let server = responses::start_mock_server().await; + let test = test_codex() + .with_model("gpt-5.6-sol") + .with_config(|config| { + config + .features + .enable(codex_features::Feature::FastMode) + .unwrap(); + }) + .build_with_auto_env(&server) + .await + .unwrap(); + responses::mount_sse_once(&server, responses::sse_completed("original")).await; + let TurnInputSubmission::Started { + turn_id: previous_turn_id, + } = test + .codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Do the work".to_string(), + text_elements: Vec::new(), + }])) + .await + .unwrap() + else { + panic!("original turn did not start") + }; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let mock = responses::mount_sse_once(&server, responses::sse_completed("continued")).await; + let input = ContextualUserFragment::into(InternalModelContextFragment::new( + InternalContextSource::from_static("daemon_recovery"), + "Continue the interrupted work.", + )); + let schema = serde_json::json!({"type":"object","properties":{},"additionalProperties":false}); + let submission = test + .codex + .continue_turn_if_idle( + TurnInputRequest::new(TurnInput::ResponseItem(input.clone())).on_start( + TurnStartOptions { + final_output_json_schema: Some(schema.clone()), + service_tier: Some("priority".to_string()), + root_turn_id: Some("originating-turn".to_string()), + ..Default::default() + }, + ), + previous_turn_id.clone(), + ) + .await + .unwrap(); + let TurnInputSubmission::Started { turn_id } = submission else { + panic!("continuation did not start") + }; + wait_for_event(&test.codex, |event| { + assert!(!matches!(event, EventMsg::UserMessage(_))); + assert!(!matches!(event, EventMsg::ItemCompleted(event) + if matches!(event.item, codex_protocol::items::TurnItem::UserMessage(_)))); + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let request = mock.single_request(); + assert!(request.has_content_kinds(&["daemon_recovery.internal_context"])); + let metadata: Value = serde_json::from_str( + &request + .header("x-codex-turn-metadata") + .expect("turn metadata"), + ) + .unwrap(); + assert_eq!(metadata["root_turn_id"], "originating-turn"); + let body = request.body_json(); + assert_eq!(body["text"]["format"]["schema"], schema); + assert_eq!(body["service_tier"], "priority"); + // The continuation has completed, but the saved previous ID is still stale. + assert_eq!( + test.codex + .continue_turn_if_idle( + TurnInputRequest::new(TurnInput::ResponseItem(ContextualUserFragment::into( + InternalModelContextFragment::new( + InternalContextSource::from_static("daemon_recovery"), + "Continue." + ), + ))), + previous_turn_id, + ) + .await + .unwrap(), + TurnInputSubmission::NotSubmitted { + reason: NotSubmittedReason::Superseded + } + ); + assert_eq!(mock.requests().len(), 1); + submit_thread_settings( + &test.codex, + ThreadSettingsOverrides { + approval_policy: Some(AskForApproval::OnRequest), + ..Default::default() + }, + ) + .await + .expect("update permissions before continuation admission"); + assert_eq!( + test.codex + .continue_turn_if_idle( + TurnInputRequest::new(TurnInput::ResponseItem(input)), + turn_id, + ) + .await + .unwrap(), + TurnInputSubmission::NotSubmitted { + reason: NotSubmittedReason::Superseded + } + ); + assert_eq!(mock.requests().len(), 1); +} + /// Concurrent submissions must start exactly one turn and steer the other message. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn turn_input_submission_reports_started_and_steered_for_concurrent_submissions() { @@ -833,17 +955,32 @@ async fn start_or_steer_turn_requires_matching_active_output_schema() { server.shutdown().await; } -#[test_case(Vec::new(); "automatic")] -#[test_case(vec![UserInput::Text { text: "Do the work".into(), text_elements: Vec::new() }]; "user")] +#[test_case(Vec::new(), "local"; "automatic")] +#[test_case(vec![UserInput::Text { text: "Do the work".into(), text_elements: Vec::new() }], "local"; "user")] +#[cfg_attr(unix, test_case(Vec::new(), "remote"; "remote_stays_idle"))] #[tokio::test] -async fn sampling_is_ready_for_daemon_recovery(input: Vec) -> anyhow::Result<()> { +async fn sampling_is_ready_for_daemon_recovery( + input: Vec, + executor: &str, +) -> anyhow::Result<()> { let (release, gate) = oneshot::channel(); let (server, _completions) = start_streaming_sse_server(vec![vec![StreamingSseChunk { gate: Some(gate), body: responses::sse_completed("automatic"), }]]) .await; - let test = test_codex().build_with_streaming_server(&server).await?; + #[cfg(unix)] + let remote = if executor == "remote" { + Some(super::multi_exec_server_sandbox::ExecServerProcess::start().await?) + } else { + None + }; + let mut builder = test_codex(); + #[cfg(unix)] + if let Some(remote) = &remote { + builder = builder.with_exec_server_url(&remote.websocket_url); + } + let test = builder.build_with_streaming_server(&server).await?; let StartIfIdleSubmission::Started { turn_id } = test .codex .start_turn_if_idle(TurnInputRequest::user_input(input)) @@ -857,7 +994,10 @@ async fn sampling_is_ready_for_daemon_recovery(input: Vec) -> anyhow: ) .await?; let active = test.codex.interrupted_turn().await; - assert_eq!(active.map(|(id, _)| id), Some(turn_id)); + assert_eq!( + active.map(|(id, _, _)| id), + (executor == "local").then_some(turn_id) + ); release.send(()).expect("sampling is waiting"); wait_for_event(&test.codex, |event| { matches!(event, EventMsg::TurnComplete(_)) diff --git a/codex-rs/protocol/src/turn_input.rs b/codex-rs/protocol/src/turn_input.rs index 61b1ffe893..ceece0461c 100644 --- a/codex-rs/protocol/src/turn_input.rs +++ b/codex-rs/protocol/src/turn_input.rs @@ -135,6 +135,9 @@ pub enum TurnInputMode { StartOrSteer, /// Start only when the thread is idle. StartIfIdle, + /// Start an internal continuation when idle. + /// Reject if another task has started since the expected previous turn. + ContinueIfIdle { expected_previous_turn_id: String }, /// Steer only if this exact turn is active. Steer { expected_turn_id: String }, } @@ -215,6 +218,9 @@ pub enum SteerSubmission { /// Why Core did not accept submitted turn input for turn processing. #[derive(Clone, Debug, Eq, PartialEq)] pub enum NotSubmittedReason { + /// New work superseded the expected previous turn of an internal continuation. + Superseded, + /// The host is draining and no longer permits new regular turns. ServerDraining, diff --git a/codex-rs/tui/src/app/agents_overview_tests.rs b/codex-rs/tui/src/app/agents_overview_tests.rs index 0aa841741a..1623289cb8 100644 --- a/codex-rs/tui/src/app/agents_overview_tests.rs +++ b/codex-rs/tui/src/app/agents_overview_tests.rs @@ -1564,8 +1564,10 @@ async fn root_switch_loads_local_preferences_from_disk() -> Result<()> { async fn root_switch_preserves_idle_root_with_running_subagent() -> Result<()> { let mut app = make_test_app().await; trust_fixture_folders(&mut app); - let mut app_server = - crate::start_embedded_app_server_for_picker(app.chat_widget.config_ref()).await?; + let mut app_server = Box::pin(crate::start_embedded_app_server_for_picker( + app.chat_widget.config_ref(), + )) + .await?; let previous = app_server.start_thread(&app.config).await?; let previous_root_id = previous.session.thread_id; app.enqueue_primary_thread_session(previous.session, previous.turns) @@ -1662,7 +1664,8 @@ async fn overview_selection_applies_user_permissions_only_to_unloaded_threads() .expect("create historical session"), )?); } - let mut app_server = crate::start_embedded_app_server_for_picker(&server_config).await?; + let mut app_server = + Box::pin(crate::start_embedded_app_server_for_picker(&server_config)).await?; let loaded = app_server .resume_thread( &crate::local_settings::LocalSettings::from(&server_config), diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index 2d0548dff8..af1ea40e27 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -7952,12 +7952,12 @@ async fn remembered_current_cwd_stays_at_launch_across_in_app_resumes() -> Resul let state_db = crate::init_state_db_for_app_server_target(&config, &crate::AppServerTarget::Embedded) .await?; - let mut app_server = crate::start_app_server_for_picker( + let mut app_server = Box::pin(crate::start_app_server_for_picker( &config, &crate::AppServerTarget::Embedded, state_db.clone(), Arc::new(EnvironmentManager::default_for_tests()), - ) + )) .await?; let (mut app, _app_event_rx, _op_rx) = make_test_app_with_channels().await; app.config = config; @@ -7968,9 +7968,8 @@ async fn remembered_current_cwd_stays_at_launch_across_in_app_resumes() -> Resul let mut tui = crate::tui::test_support::make_test_tui()?; for target_session in targets { - let control = app - .resume_target_session(&mut tui, &mut app_server, target_session) - .await?; + let control = + Box::pin(app.resume_target_session(&mut tui, &mut app_server, target_session)).await?; assert!(matches!(control, AppRunControl::Continue)); assert!(!crate::session_resume::cwds_differ( 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 efacbcf5df..a4bf3563cb 100644 --- a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs +++ b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs @@ -805,11 +805,11 @@ fn spawn_approved_task_tool_call( #[tokio::test] async fn external_transport_registers_dynamic_tools_and_finds_task_mentions() -> Result<()> { let (app, _codex_home) = make_history_test_app().await?; - let (mut app_server, requests, proxy) = start_recording_app_server( + let (mut app_server, requests, proxy) = Box::pin(start_recording_app_server( &app.config, /*blocked_thread_list*/ None, /*failed_thread_name*/ None, - ) + )) .await?; let started = app_server.start_thread(&app.config).await?; @@ -858,25 +858,27 @@ async fn external_transport_registers_dynamic_tools_and_finds_task_mentions() -> app_server.shutdown().await?; proxy.await??; let (mut restarted_app_server, _restarted_requests, restarted_proxy) = - start_recording_app_server( + Box::pin(start_recording_app_server( &app.config, /*blocked_thread_list*/ None, /*failed_thread_name*/ None, - ) - .await?; - let resumed = restarted_app_server - .resume_thread( - &app.local_settings, - app.config.clone(), - target_id, - crate::app_server_session::ResumeModelSettings::RestoreFromThread, - ) + )) .await?; + let resumed = Box::pin(restarted_app_server.resume_thread( + &app.local_settings, + app.config.clone(), + target_id, + crate::app_server_session::ResumeModelSettings::RestoreFromThread, + )) + .await?; assert!(resumed.task_tools_available); assert!(restarted_app_server.task_tools_available(target_id)); - let forked = restarted_app_server - .fork_thread(&app.local_settings, app.config.clone(), target_id) - .await?; + let forked = Box::pin(restarted_app_server.fork_thread( + &app.local_settings, + app.config.clone(), + target_id, + )) + .await?; assert!(forked.task_tools_available); assert!(restarted_app_server.task_tools_available(forked.session.thread_id)); restarted_app_server diff --git a/codex-rs/tui/src/app/tests/startup.rs b/codex-rs/tui/src/app/tests/startup.rs index 8398ae08b5..b178368a68 100644 --- a/codex-rs/tui/src/app/tests/startup.rs +++ b/codex-rs/tui/src/app/tests/startup.rs @@ -1377,7 +1377,7 @@ async fn owned_subagent_approval_before_thread_started_is_preserved() -> Result< let codex_home = tempdir()?; app.config.codex_home = codex_home.path().to_path_buf().abs(); app.config.sqlite = codex_state::SqliteConfig::new_for_testing(codex_home.path().abs()); - let mut app_server = crate::start_embedded_app_server_for_picker(&app.config).await?; + let mut app_server = Box::pin(crate::start_embedded_app_server_for_picker(&app.config)).await?; let parent = app_server.start_thread(&app.config).await?; let parent_thread_id = parent.session.thread_id; app.enqueue_primary_thread_session(parent.session, parent.turns)