Files
codex/codex-rs/app-server/src/daemon_thread_recovery.rs
T
Eric Traut f2b5b81f39 Continue interrupted work after managed daemon restarts (#45820)
## Why

Daemon recovery reloads threads but leaves interrupted work unfinished. Resume eligible work automatically from the saved conversation, even without a connected client.

## What changed

- Attempt one new continuation turn immediately after restoration, including in Plan mode. Mark the old turn interrupted and supply recovery context without creating a user message or granting new authorization.
- Require an idle thread, matching permissions, and the same single local environment configured by the thread. Skip completed, aborted, or superseded work and snapshots without environment identity.
- Add `continue_turn_if_idle` with an atomic previous-turn check so newer tasks or standalone settings changes invalidate pending continuation.
- Preserve the output schema, service tier, and root turn ID, and emit a “Resuming interrupted work” warning when continuation starts.

## Testing

Add daemon restart coverage for continuation without a client, Plan mode, permission and environment mismatches, and legacy snapshots. Add core coverage for preserved continuation metadata, absence of user-message events, rejection of superseded continuations, and exclusion of remote execution from recovery snapshots.

GitOrigin-RevId: 2b290b75d9399fc58258bcda2e85c74e9b5b3b09
2026-09-16 00:54:58 +00:00

51 lines
1.9 KiB
Rust

//! Saves persistent root thread IDs and restores them through the shared internal resume path.
//! Recovery uses normal cold-resume semantics without delaying readiness for runtime loading.
//! Already-loaded runtimes remain owned by their current clients.
use std::io;
use std::path::PathBuf;
use codex_app_server_transport::daemon_recovery;
pub(crate) async fn snapshot(
path: PathBuf,
saved: daemon_recovery::RecoverySnapshot,
) -> io::Result<()> {
// A forced exit must not wait for file I/O in Tokio's blocking pool.
let (result_tx, result_rx) = tokio::sync::oneshot::channel();
std::thread::Builder::new()
.name("daemon-snapshot".into())
.spawn(move || {
let result = daemon_recovery::write_snapshot(&path, &saved);
if result.is_err()
&& let Err(err) = std::fs::remove_file(&path)
&& err.kind() != io::ErrorKind::NotFound
{
tracing::warn!("failed to clear stale daemon recovery file: {err}");
}
let _ = result_tx.send(result);
})?;
result_rx.await.map_err(io::Error::other)?
}
/// Consume the handoff before serving requests, then restore runtimes in the background.
pub(crate) async fn start_recovery(
path: PathBuf,
processor: std::sync::Arc<crate::message_processor::MessageProcessor>,
) -> io::Result<tokio::task::JoinHandle<()>> {
let candidates = tokio::task::spawn_blocking(move || {
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,
Err(err) if err.kind() == io::ErrorKind::NotFound => candidates,
Err(err) => Err(err),
}
})
.await
.map_err(io::Error::other)??;
Ok(tokio::spawn(async move {
processor.restore_daemon_threads(candidates).await;
}))
}