diff --git a/codex-rs/app-server/tests/suite/v2/thread_unload.rs b/codex-rs/app-server/tests/suite/v2/thread_unload.rs index 8c74cbf2b9..7e6c478505 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_unload.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_unload.rs @@ -1,18 +1,22 @@ -//! Idle unloading must finish even when resource teardown exceeds the shutdown warning deadline. +//! Idle unloading must finish when another process holds the writer coordination lock. use anyhow::Context; use anyhow::Result; use app_test_support::MockResponsesConfig; use app_test_support::TestAppServer; -use app_test_support::create_mock_responses_server_sequence_unchecked; +use app_test_support::create_mock_responses_server_repeating_assistant; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ThreadClosedNotification; use codex_app_server_protocol::ThreadLoadedListParams; use codex_app_server_protocol::ThreadLoadedListResponse; +use codex_app_server_protocol::ThreadResumeParams; +use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadUnsubscribeParams; use codex_app_server_protocol::ThreadUnsubscribeResponse; use codex_app_server_protocol::ThreadUnsubscribeStatus; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::UserInput; use pretty_assertions::assert_eq; use std::fs::OpenOptions; use std::time::Duration; @@ -20,15 +24,14 @@ use tempfile::TempDir; use tokio::time::timeout; #[tokio::test] -async fn idle_unload_finishes_after_slow_writer_shutdown() -> Result<()> { - let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await; +async fn idle_unload_finishes_while_writer_coordination_is_locked() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; let home = TempDir::new()?; MockResponsesConfig::new(&server.uri()) .with_root_config("thread_unload_delay_secs = 0\napprovals_reviewer = \"user\"") .write(home.path())?; let mut client = TestAppServer::builder() .with_codex_home(home.path()) - .with_json_logging("warn") // Writer cleanup blocks one worker while the test queries the server. .with_env_overrides(&[("TOKIO_WORKER_THREADS", Some("2"))]) .build_initialized() @@ -36,8 +39,20 @@ async fn idle_unload_finishes_after_slow_writer_shutdown() -> Result<()> { let started = client.start_thread(ThreadStartParams::default()).await?; let thread_id = started.thread.id; - // Block the real cross-process writer-lock cleanup. Other app-server tasks, - // including the ten-second shutdown deadline and JSON-RPC, can still run. + // Persist a turn so a cold resume can verify durable history after unload. + client + .start_turn_and_wait_for_completion(TurnStartParams { + thread_id: thread_id.clone(), + input: vec![UserInput::Text { + text: "hello".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + + // Hold coordination through unload: cleanup must time out without keeping + // the thread loaded or retaining its writer ownership. let coordination_lock = OpenOptions::new() .read(true) .write(true) @@ -57,39 +72,18 @@ async fn idle_unload_finishes_after_slow_writer_shutdown() -> Result<()> { status: ThreadUnsubscribeStatus::Unsubscribed } ); - // Unsubscribe only schedules shutdown. Keep the lock until the shutdown - // future has actually crossed its warning deadline. - client - .wait_for_json_log_event( - "codex.app_server.thread_shutdown_slow", - Duration::from_secs(/*secs*/ 30), - ) - .await - .context("idle unload never reached the shutdown warning deadline")?; - let loaded: ThreadLoadedListResponse = client - .request(|request_id| ClientRequest::ThreadLoadedList { - request_id, - params: ThreadLoadedListParams::default(), - }) - .await?; - assert_eq!( - loaded, - ThreadLoadedListResponse { - data: vec![thread_id.clone()], - next_cursor: None - } - ); - - // Once persistence can finish, delayed shutdown must still remove the runtime - // and report closure without another client subscription or unload request. - drop(coordination_lock); let closed: ThreadClosedNotification = timeout( - Duration::from_secs(/*secs*/ 10), + Duration::from_secs(/*secs*/ 60), client.read_notification("thread/closed"), ) .await - .context("slow shutdown never finalized idle unload")??; - assert_eq!(closed, ThreadClosedNotification { thread_id }); + .context("writer coordination contention prevented idle unload")??; + assert_eq!( + closed, + ThreadClosedNotification { + thread_id: thread_id.clone() + } + ); let loaded: ThreadLoadedListResponse = client .request(|request_id| ClientRequest::ThreadLoadedList { request_id, @@ -103,5 +97,21 @@ async fn idle_unload_finishes_after_slow_writer_shutdown() -> Result<()> { next_cursor: None } ); + // Cleanup may leave a lock file behind, but must release ownership so the + // same thread can resume after the other process releases coordination. + drop(coordination_lock); + let resumed: ThreadResumeResponse = client + .request(|request_id| ClientRequest::ThreadResume { + request_id, + params: ThreadResumeParams { + thread_id: thread_id.clone(), + ..Default::default() + }, + }) + .await?; + assert_eq!( + (resumed.thread.id, resumed.thread.turns.len()), + (thread_id, 1) + ); Ok(()) } diff --git a/codex-rs/rollout/src/writer_lock.rs b/codex-rs/rollout/src/writer_lock.rs index ceae18e743..31068707a2 100644 --- a/codex-rs/rollout/src/writer_lock.rs +++ b/codex-rs/rollout/src/writer_lock.rs @@ -1,5 +1,5 @@ //! Shared cross-process ownership for local thread writers and rollout publication. -//! Thread lock-file creation and removal uses the existing home coordination lock. +//! Thread lock-file creation and removal uses a bounded wait for the home coordination lock. use std::fs; use std::fs::File; @@ -10,6 +10,8 @@ use std::path::PathBuf; use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; +use std::time::Duration; +use std::time::Instant; use codex_protocol::ThreadId; use tracing::warn; @@ -113,11 +115,22 @@ impl WriterLockCoordinator { .create(true) .truncate(false) .open(self.directory.join(COORDINATION_LOCK_FILE))?; - file.lock().map_err(|err| { - io::Error::other(format!( - "failed to acquire thread writer coordination lock: {err}" - )) - })?; + let deadline = Instant::now() + Duration::from_secs(/*secs*/ 30); + loop { + match file.try_lock() { + Ok(()) => break, + Err(fs::TryLockError::WouldBlock) if Instant::now() < deadline => { + std::thread::sleep(Duration::from_millis(/*millis*/ 50)); + } + Err(fs::TryLockError::WouldBlock) => { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "timed out waiting for the thread writer coordination lock; another Codex process may be suspended on this or another host", + )); + } + Err(fs::TryLockError::Error(err)) => return Err(err), + } + } Ok(file) } diff --git a/codex-rs/rollout/src/writer_lock_tests.rs b/codex-rs/rollout/src/writer_lock_tests.rs index 9fad0aea36..89e42c1f40 100644 --- a/codex-rs/rollout/src/writer_lock_tests.rs +++ b/codex-rs/rollout/src/writer_lock_tests.rs @@ -109,6 +109,11 @@ fn publication_skips_live_writers_and_keeps_coordination_locked() { coordination.try_lock(), Err(fs::TryLockError::WouldBlock) )); + let error = writer + .acquire(ThreadId::default()) + .err() + .expect("coordination contention must return an error instead of hanging"); + assert_eq!(error.kind(), ErrorKind::TimedOut); drop(publication); coordination .try_lock()