From dc60dadce6b5911b9b5852365f6eb9829ec14442 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Fri, 31 Jul 2026 20:42:29 +0000 Subject: [PATCH] Enforce single-writer ownership for all thread histories (#36389) ## Why Legacy thread histories did not use the cross-process writer ownership guard already applied to paginated histories. ## What changed - Acquire and retain a writer lock whenever a legacy or paginated thread is created or resumed. - Apply the same ownership checks when archiving or deleting either history mode. - Release ownership when initialization fails or the active thread shuts down, allowing another process to resume it. ## Testing - Cover competing create and resume operations for both history modes. - Cover archive and deletion conflicts for legacy and paginated descendants. GitOrigin-RevId: 1be79b01966ad2634ed79eb96c0aa0db8eafe2fa --- .../tests/suite/v2/thread_resume.rs | 56 +++----- codex-rs/core/tests/common/test_codex.rs | 15 +++ .../tests/suite/collaboration_instructions.rs | 8 +- codex-rs/core/tests/suite/compact.rs | 11 +- codex-rs/core/tests/suite/hooks.rs | 16 +-- .../core/tests/suite/mcp_tool_exposure.rs | 6 +- .../core/tests/suite/model_visible_layout.rs | 16 +-- codex-rs/core/tests/suite/multi_agent_mode.rs | 20 +-- .../core/tests/suite/multi_agent_resume.rs | 12 +- .../core/tests/suite/permissions_messages.rs | 11 +- codex-rs/core/tests/suite/resume.rs | 112 +--------------- codex-rs/core/tests/suite/sqlite_state.rs | 1 + .../thread-store/src/local/archive_thread.rs | 91 +++++++------ .../thread-store/src/local/delete_thread.rs | 93 ++++++------- .../thread-store/src/local/live_writer.rs | 12 +- codex-rs/thread-store/src/local/mod.rs | 126 ++++++++++++------ 16 files changed, 240 insertions(+), 366 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index 9558034708..17a768b73c 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -57,8 +57,6 @@ use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus; use codex_app_server_protocol::ThreadStatusChangedNotification; -use codex_app_server_protocol::ThreadTurnsListParams; -use codex_app_server_protocol::ThreadTurnsListResponse; use codex_app_server_protocol::ThreadUnsubscribeParams; use codex_app_server_protocol::TurnItemsView; use codex_app_server_protocol::TurnStartParams; @@ -255,8 +253,19 @@ async fn thread_resume_paginated_model_context_preserves_original_metadata() -> Ok(()) } +#[tokio::test] +async fn thread_resume_rejects_legacy_writer_owned_by_another_process() -> Result<()> { + assert_thread_resume_rejects_writer_owned_by_another_process(ThreadHistoryMode::Legacy).await +} + #[tokio::test] async fn thread_resume_rejects_paginated_writer_owned_by_another_process() -> Result<()> { + assert_thread_resume_rejects_writer_owned_by_another_process(ThreadHistoryMode::Paginated).await +} + +async fn assert_thread_resume_rejects_writer_owned_by_another_process( + history_mode: ThreadHistoryMode, +) -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; mock_responses_config(&server.uri()).write(codex_home.path())?; @@ -268,7 +277,7 @@ async fn thread_resume_rejects_paginated_writer_owned_by_another_process() -> Re let ThreadStartResponse { thread, .. } = primary .start_thread(ThreadStartParams { model: Some("gpt-5.4".to_string()), - history_mode: Some(ThreadHistoryMode::Paginated), + history_mode: Some(history_mode), ..Default::default() }) .await?; @@ -295,16 +304,6 @@ async fn thread_resume_rejects_paginated_writer_owned_by_another_process() -> Re )]) .build_initialized() .await?; - let read_id = secondary - .send_thread_read_request(ThreadReadParams { - thread_id: thread.id.clone(), - include_turns: false, - }) - .await?; - let ThreadReadResponse { thread: read, .. } = - timeout(DEFAULT_READ_TIMEOUT, secondary.read_response(read_id)).await??; - assert_eq!(read.id, thread.id); - let resume_id = secondary .send_thread_resume_request(ThreadResumeParams { thread_id: thread.id.clone(), @@ -348,18 +347,6 @@ async fn thread_resume_rejects_paginated_writer_owned_by_another_process() -> Re ) .await??; - let list_id = secondary - .send_thread_turns_list_request(ThreadTurnsListParams { - thread_id: thread.id, - cursor: None, - limit: None, - sort_direction: None, - items_view: None, - }) - .await?; - let ThreadTurnsListResponse { data, .. } = - timeout(DEFAULT_READ_TIMEOUT, secondary.read_response(list_id)).await??; - assert_eq!(data.len(), 2); Ok(()) } @@ -2988,11 +2975,6 @@ async fn thread_resume_keeps_in_flight_turn_streaming() -> Result<()> { .await??; primary.clear_message_buffer(); - let mut secondary = TestAppServer::builder() - .with_codex_home(codex_home.path()) - .build_initialized() - .await?; - let turn_id = primary .send_turn_start_request(TurnStartParams { thread_id: thread.id.clone(), @@ -3015,7 +2997,7 @@ async fn thread_resume_keeps_in_flight_turn_streaming() -> Result<()> { ) .await??; - let resume_id = secondary + let resume_id = primary .send_thread_resume_request(ThreadResumeParams { thread_id: thread.id, ..Default::default() @@ -3024,7 +3006,7 @@ async fn thread_resume_keeps_in_flight_turn_streaming() -> Result<()> { let ThreadResumeResponse { thread: resumed_thread, .. - } = timeout(DEFAULT_READ_TIMEOUT, secondary.read_response(resume_id)).await??; + } = timeout(DEFAULT_READ_TIMEOUT, primary.read_response(resume_id)).await??; assert_ne!(resumed_thread.status, ThreadStatus::NotLoaded); timeout( DEFAULT_READ_TIMEOUT, @@ -3517,12 +3499,7 @@ async fn thread_resume_can_skip_turns_when_thread_is_running() -> Result<()> { ) .await??; - let mut secondary = TestAppServer::builder() - .with_codex_home(codex_home.path()) - .build_initialized() - .await?; - - let resume_id = secondary + let resume_id = primary .send_thread_resume_request(ThreadResumeParams { thread_id: thread.id.clone(), exclude_turns: true, @@ -3531,7 +3508,7 @@ async fn thread_resume_can_skip_turns_when_thread_is_running() -> Result<()> { .await?; let ThreadResumeResponse { thread: resumed, .. - } = timeout(DEFAULT_READ_TIMEOUT, secondary.read_response(resume_id)).await??; + } = timeout(DEFAULT_READ_TIMEOUT, primary.read_response(resume_id)).await??; assert_eq!(resumed.id, thread.id); assert_eq!(resumed.status, ThreadStatus::Idle); @@ -4313,6 +4290,7 @@ async fn thread_resume_accepts_personality_override() -> Result<()> { ) .await??; + timeout(DEFAULT_READ_TIMEOUT, primary.shutdown_gracefully()).await??; let mut secondary = TestAppServer::builder() .with_codex_home(codex_home.path()) .build_initialized() diff --git a/codex-rs/core/tests/common/test_codex.rs b/codex-rs/core/tests/common/test_codex.rs index fbfbb3f25b..1118591ac1 100644 --- a/codex-rs/core/tests/common/test_codex.rs +++ b/codex-rs/core/tests/common/test_codex.rs @@ -549,6 +549,21 @@ impl TestCodexBuilder { .await } + pub async fn restart( + &mut self, + server: &MockServer, + previous: &TestCodex, + ) -> Result { + let rollout_path = previous + .session_configured + .rollout_path + .clone() + .context("rollout path")?; + previous.codex.shutdown_and_wait().await?; + self.resume(server, Arc::clone(&previous.home), rollout_path) + .await + } + async fn build_with_home_and_base_url( &mut self, base_url: String, diff --git a/codex-rs/core/tests/suite/collaboration_instructions.rs b/codex-rs/core/tests/suite/collaboration_instructions.rs index a34969ebcb..a1a6c36460 100644 --- a/codex-rs/core/tests/suite/collaboration_instructions.rs +++ b/codex-rs/core/tests/suite/collaboration_instructions.rs @@ -944,12 +944,6 @@ async fn resume_replays_collaboration_instructions() -> Result<()> { let mut builder = test_codex(); let initial = builder.build(&server).await?; - let rollout_path = initial - .session_configured - .rollout_path - .clone() - .expect("rollout path"); - let home = initial.home.clone(); let collab_text = "resume instructions"; core_test_support::submit_thread_settings( @@ -976,7 +970,7 @@ async fn resume_replays_collaboration_instructions() -> Result<()> { .await?; wait_for_event(&initial.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; - let resumed = builder.resume(&server, home, rollout_path).await?; + let resumed = builder.restart(&server, &initial).await?; resumed .codex .submit(Op::UserInput { diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index 2d39c44f15..dc425387c7 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -2041,12 +2041,6 @@ async fn auto_compact_runs_after_resume_when_token_usage_is_over_limit() { let _ = config.features.disable(Feature::RemoteCompactionV2); }); let initial = builder.build(&server).await.unwrap(); - let home = initial.home.clone(); - let rollout_path = initial - .session_configured - .rollout_path - .clone() - .expect("rollout path"); // A single over-limit completion should not auto-compact until the next user message. mount_sse_once( @@ -2069,10 +2063,7 @@ async fn auto_compact_runs_after_resume_when_token_usage_is_over_limit() { config.model_auto_compact_token_limit = Some(limit); let _ = config.features.disable(Feature::RemoteCompactionV2); }); - let resumed = resume_builder - .resume(&server, home, rollout_path) - .await - .unwrap(); + let resumed = resume_builder.restart(&server, &initial).await.unwrap(); let follow_up_user = "AFTER_RESUME_USER"; let sse_follow_up = sse(vec![ diff --git a/codex-rs/core/tests/suite/hooks.rs b/codex-rs/core/tests/suite/hooks.rs index 865ba7e01b..f52917df03 100644 --- a/codex-rs/core/tests/suite/hooks.rs +++ b/codex-rs/core/tests/suite/hooks.rs @@ -2005,12 +2005,6 @@ async fn resumed_thread_runs_resume_then_compact_session_start_hooks() -> Result trust_discovered_hooks(config); }); let initial = builder.build(&server).await?; - let home = initial.home.clone(); - let rollout_path = initial - .session_configured - .rollout_path - .clone() - .context("rollout path")?; initial.submit_turn("hello before resume").await?; assert_eq!(responses_mock.requests().len(), 1); @@ -2019,7 +2013,7 @@ async fn resumed_thread_runs_resume_then_compact_session_start_hooks() -> Result config.model_auto_compact_token_limit = Some(limit); trust_discovered_hooks(config); }); - let resumed = resume_builder.resume(&server, home, rollout_path).await?; + let resumed = resume_builder.restart(&server, &initial).await?; resumed.submit_turn("hello after resume").await?; let requests = responses_mock.requests(); @@ -2129,12 +2123,6 @@ async fn resumed_thread_keeps_stop_continuation_prompt_in_history() -> Result<() }) .with_config(trust_discovered_hooks); let initial = initial_builder.build(&server).await?; - let home = initial.home.clone(); - let rollout_path = initial - .session_configured - .rollout_path - .clone() - .expect("rollout path"); initial.submit_turn("tell me something").await?; @@ -2151,7 +2139,7 @@ async fn resumed_thread_keeps_stop_continuation_prompt_in_history() -> Result<() .await; let mut resume_builder = test_codex().with_config(trust_discovered_hooks); - let resumed = resume_builder.resume(&server, home, rollout_path).await?; + let resumed = resume_builder.restart(&server, &initial).await?; resumed.submit_turn("and now continue").await?; diff --git a/codex-rs/core/tests/suite/mcp_tool_exposure.rs b/codex-rs/core/tests/suite/mcp_tool_exposure.rs index 0c33196ccc..627d1abd71 100644 --- a/codex-rs/core/tests/suite/mcp_tool_exposure.rs +++ b/codex-rs/core/tests/suite/mcp_tool_exposure.rs @@ -549,7 +549,6 @@ async fn deferred_tool_world_state_survives_resume_without_duplicate_updates() - initial.codex.ensure_rollout_materialized().await; initial.codex.flush_rollout().await?; - let home = initial.home.clone(); let rollout_path = initial .session_configured .rollout_path @@ -573,11 +572,10 @@ async fn deferred_tool_world_state_survives_resume_without_duplicate_updates() - "mcp__codex_apps__calendar": "Plan events and manage your calendar." }) ); - drop(initial); - let mut resume_builder = search_capable_apps_builder(apps_server.chatgpt_base_url) .with_config(enable_deferred_tool_world_state_without_agents); - let resumed = resume_builder.resume(&server, home, rollout_path).await?; + let resumed = resume_builder.restart(&server, &initial).await?; + drop(initial); wait_for_mcp_server(&resumed.codex, CODEX_APPS_MCP_SERVER_NAME).await?; resumed .submit_turn("inspect unchanged deferred tools after resume") diff --git a/codex-rs/core/tests/suite/model_visible_layout.rs b/codex-rs/core/tests/suite/model_visible_layout.rs index 4d63c7ec11..6550e8e26b 100644 --- a/codex-rs/core/tests/suite/model_visible_layout.rs +++ b/codex-rs/core/tests/suite/model_visible_layout.rs @@ -424,12 +424,6 @@ async fn snapshot_model_visible_layout_resume_with_personality_change() -> Resul }); let initial = initial_builder.build(&server).await?; let codex = Arc::clone(&initial.codex); - let home = initial.home.clone(); - let rollout_path = initial - .session_configured - .rollout_path - .clone() - .expect("rollout path"); let initial_mock = mount_sse_once( &server, @@ -475,7 +469,7 @@ async fn snapshot_model_visible_layout_resume_with_personality_change() -> Resul .expect("test config should allow feature update"); config.personality = Some(Personality::Pragmatic); }); - let resumed = resume_builder.resume(&server, home, rollout_path).await?; + let resumed = resume_builder.restart(&server, &initial).await?; let resume_override_cwd = resumed.cwd_path().join(PRETURN_CONTEXT_DIFF_CWD); fs::create_dir_all(&resume_override_cwd)?; let resume_override_cwd = resume_override_cwd.abs(); @@ -543,12 +537,6 @@ async fn snapshot_model_visible_layout_resume_override_matches_rollout_model() - }); let initial = initial_builder.build(&server).await?; let codex = Arc::clone(&initial.codex); - let home = initial.home.clone(); - let rollout_path = initial - .session_configured - .rollout_path - .clone() - .expect("rollout path"); let initial_mock = mount_sse_once( &server, @@ -589,7 +577,7 @@ async fn snapshot_model_visible_layout_resume_override_matches_rollout_model() - .with_config(|config| { config.model = Some("gpt-5.4".to_string()); }); - let resumed = resume_builder.resume(&server, home, rollout_path).await?; + let resumed = resume_builder.restart(&server, &initial).await?; let resume_override_cwd = resumed.cwd_path().join(PRETURN_CONTEXT_DIFF_CWD); fs::create_dir_all(&resume_override_cwd)?; let resume_override_cwd = resume_override_cwd.abs(); diff --git a/codex-rs/core/tests/suite/multi_agent_mode.rs b/codex-rs/core/tests/suite/multi_agent_mode.rs index c6540e629b..585e63bdd8 100644 --- a/codex-rs/core/tests/suite/multi_agent_mode.rs +++ b/codex-rs/core/tests/suite/multi_agent_mode.rs @@ -225,21 +225,15 @@ async fn changing_configured_mode_hint_to_empty_emits_no_update() -> Result<()> .with_config(configure_custom_mode_hint) .build(&server) .await?; - let home = initial.home.clone(); - let rollout_path = initial - .session_configured - .rollout_path - .clone() - .expect("rollout path"); submit_turn(&initial.codex, "before resume", /*effort*/ None).await?; - drop(initial); let mut resume_builder = test_codex().with_config(|config| { configure_multi_agent_v2(config); config.multi_agent_v2.multi_agent_mode_hint_text = Some(String::new()); }); - let resumed = resume_builder.resume(&server, home, rollout_path).await?; + let resumed = resume_builder.restart(&server, &initial).await?; + drop(initial); submit_turn(&resumed.codex, "after resume", /*effort*/ None).await?; let requests = responses.requests(); @@ -362,20 +356,14 @@ async fn leaving_ultra_after_cold_resume_emits_explicit_mode() -> Result<()> { .with_config(configure_ultra) .build(&server) .await?; - let home = initial.home.clone(); - let rollout_path = initial - .session_configured - .rollout_path - .clone() - .expect("rollout path"); submit_turn(&initial.codex, "before resume", /*effort*/ None).await?; - drop(initial); let mut resume_builder = test_codex() .with_model_info_override("gpt-5.4", add_ultra_reasoning) .with_config(configure_ultra); - let resumed = resume_builder.resume(&server, home, rollout_path).await?; + let resumed = resume_builder.restart(&server, &initial).await?; + drop(initial); submit_turn(&resumed.codex, "after resume", Some(ReasoningEffort::High)).await?; let requests = responses.requests(); diff --git a/codex-rs/core/tests/suite/multi_agent_resume.rs b/codex-rs/core/tests/suite/multi_agent_resume.rs index c770c5151e..0ee40d1d79 100644 --- a/codex-rs/core/tests/suite/multi_agent_resume.rs +++ b/codex-rs/core/tests/suite/multi_agent_resume.rs @@ -245,12 +245,6 @@ async fn cold_root_resume_restores_agent_identity_and_role_on_followup() -> Resu }); let initial = initial_builder.build_with_auto_env(&server).await?; let root_thread_id = initial.session_configured.thread_id; - let home = initial.home.clone(); - let rollout_path = initial - .codex - .rollout_path() - .expect("root rollout path") - .to_path_buf(); let mut op = vec![UserInput::Text { text: INITIAL_PROMPT.to_string(), text_elements: Vec::new(), @@ -382,9 +376,10 @@ async fn cold_root_resume_restores_agent_identity_and_role_on_followup() -> Resu sibling_thread.flush_rollout().await?; worker_thread.flush_rollout().await?; initial.codex.flush_rollout().await?; + sibling_thread.shutdown_and_wait().await?; + worker_thread.shutdown_and_wait().await?; drop(sibling_thread); drop(worker_thread); - drop(initial); let followup_args = serde_json::to_string(&json!({ "target": "worker", @@ -437,7 +432,8 @@ async fn cold_root_resume_restores_agent_identity_and_role_on_followup() -> Resu let mut resume_builder = test_codex().with_config(move |config| { configure_multi_agent_v2_with_role(config, &resumed_model_provider_base_url); }); - let resumed = resume_builder.resume(&server, home, rollout_path).await?; + let resumed = resume_builder.restart(&server, &initial).await?; + drop(initial); assert_eq!( resumed.thread_manager.list_thread_ids().await, vec![root_thread_id] diff --git a/codex-rs/core/tests/suite/permissions_messages.rs b/codex-rs/core/tests/suite/permissions_messages.rs index 2199bdba84..f42b154a39 100644 --- a/codex-rs/core/tests/suite/permissions_messages.rs +++ b/codex-rs/core/tests/suite/permissions_messages.rs @@ -631,12 +631,6 @@ async fn resume_replays_permissions_messages() -> Result<()> { config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); }); let initial = builder.build(&server).await?; - let rollout_path = initial - .session_configured - .rollout_path - .clone() - .expect("rollout path"); - let home = initial.home.clone(); initial .codex @@ -677,7 +671,7 @@ async fn resume_replays_permissions_messages() -> Result<()> { .await?; wait_for_event(&initial.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; - let resumed = builder.resume(&server, home, rollout_path).await?; + let resumed = builder.restart(&server, &initial).await?; resumed .codex .submit(Op::UserInput { @@ -736,7 +730,6 @@ async fn resume_and_fork_append_permissions_messages() -> Result<()> { .rollout_path .clone() .expect("rollout path"); - let home = initial.home.clone(); initial .codex @@ -783,7 +776,7 @@ async fn resume_and_fork_append_permissions_messages() -> Result<()> { builder = builder.with_config(|config| { config.permissions.approval_policy = Constrained::allow_any(AskForApproval::UnlessTrusted); }); - let resumed = builder.resume(&server, home, rollout_path.clone()).await?; + let resumed = builder.restart(&server, &initial).await?; resumed .codex .submit(Op::UserInput { diff --git a/codex-rs/core/tests/suite/resume.rs b/codex-rs/core/tests/suite/resume.rs index 83cb4cdc60..759d3c5a5e 100644 --- a/codex-rs/core/tests/suite/resume.rs +++ b/codex-rs/core/tests/suite/resume.rs @@ -13,49 +13,10 @@ use core_test_support::responses::mount_sse_sequence; use core_test_support::responses::sse; use core_test_support::responses::start_mock_server; use core_test_support::skip_if_no_network; -use core_test_support::test_codex::TestCodex; -use core_test_support::test_codex::TestCodexBuilder; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; use pretty_assertions::assert_eq; -use std::path::PathBuf; use std::sync::Arc; -use std::time::Duration; -use tempfile::TempDir; -use wiremock::MockServer; - -async fn resume_until_initial_messages( - builder: &mut TestCodexBuilder, - server: &MockServer, - home: Arc, - rollout_path: PathBuf, - predicate: impl Fn(&[EventMsg]) -> bool, -) -> Result { - let deadline = tokio::time::Instant::now() + Duration::from_secs(2); - let poll_interval = Duration::from_millis(10); - let mut last_initial_messages = "".to_string(); - - loop { - let resumed = builder - .resume(server, Arc::clone(&home), rollout_path.clone()) - .await?; - if let Some(initial_messages) = resumed.session_configured.initial_messages.as_ref() { - if predicate(initial_messages) { - return Ok(resumed); - } - last_initial_messages = format!("{initial_messages:#?}"); - } - - if tokio::time::Instant::now() >= deadline { - panic!( - "timed out waiting for rollout resume messages to stabilize: {last_initial_messages}" - ); - } - - drop(resumed); - tokio::time::sleep(poll_interval).await; - } -} #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn resume_includes_initial_messages_from_rollout_events() -> Result<()> { @@ -65,12 +26,6 @@ async fn resume_includes_initial_messages_from_rollout_events() -> Result<()> { let mut builder = test_codex(); let initial = builder.build(&server).await?; let codex = Arc::clone(&initial.codex); - let home = initial.home.clone(); - let rollout_path = initial - .session_configured - .rollout_path - .clone() - .expect("rollout path"); let initial_sse = sse(vec![ ev_response_created("resp-initial"), @@ -98,26 +53,7 @@ async fn resume_includes_initial_messages_from_rollout_events() -> Result<()> { .await?; wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await; - - let resumed = resume_until_initial_messages( - &mut builder, - &server, - home, - rollout_path, - |initial_messages| { - matches!( - initial_messages, - [ - EventMsg::TurnStarted(_), - EventMsg::UserMessage(_), - EventMsg::AgentMessage(_), - EventMsg::TokenCount(_), - EventMsg::TurnComplete(_), - ] - ) - }, - ) - .await?; + let resumed = builder.restart(&server, &initial).await?; let initial_messages = resumed .session_configured .initial_messages @@ -155,12 +91,6 @@ async fn resume_includes_initial_messages_from_reasoning_events() -> Result<()> }); let initial = builder.build(&server).await?; let codex = Arc::clone(&initial.codex); - let home = initial.home.clone(); - let rollout_path = initial - .session_configured - .rollout_path - .clone() - .expect("rollout path"); let initial_sse = sse(vec![ ev_response_created("resp-initial"), @@ -184,28 +114,7 @@ async fn resume_includes_initial_messages_from_reasoning_events() -> Result<()> .await?; wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await; - - let resumed = resume_until_initial_messages( - &mut builder, - &server, - home, - rollout_path, - |initial_messages| { - matches!( - initial_messages, - [ - EventMsg::TurnStarted(_), - EventMsg::UserMessage(_), - EventMsg::AgentReasoning(_), - EventMsg::AgentReasoningRawContent(_), - EventMsg::AgentMessage(_), - EventMsg::TokenCount(_), - EventMsg::TurnComplete(_), - ] - ) - }, - ) - .await?; + let resumed = builder.restart(&server, &initial).await?; let initial_messages = resumed .session_configured .initial_messages @@ -246,12 +155,6 @@ async fn resume_switches_models_preserves_base_instructions() -> Result<()> { }); let initial = builder.build(&server).await?; let codex = Arc::clone(&initial.codex); - let home = initial.home.clone(); - let rollout_path = initial - .session_configured - .rollout_path - .clone() - .expect("rollout path"); let initial_sse = sse(vec![ ev_response_created("resp-initial"), @@ -273,7 +176,6 @@ async fn resume_switches_models_preserves_base_instructions() -> Result<()> { }) .await?; wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await; - let initial_body = initial_mock.single_request().body_json(); let initial_instructions = initial_body .get("instructions") @@ -301,7 +203,7 @@ async fn resume_switches_models_preserves_base_instructions() -> Result<()> { let mut resume_builder = test_codex().with_config(|config| { config.model = Some("gpt-5.4".to_string()); }); - let resumed = resume_builder.resume(&server, home, rollout_path).await?; + let resumed = resume_builder.restart(&server, &initial).await?; resumed .codex .submit(Op::UserInput { @@ -378,12 +280,6 @@ async fn resume_model_switch_is_not_duplicated_after_pre_turn_override() -> Resu }); let initial = builder.build(&server).await?; let codex = Arc::clone(&initial.codex); - let home = initial.home.clone(); - let rollout_path = initial - .session_configured - .rollout_path - .clone() - .expect("rollout path"); let initial_mock = mount_sse_once( &server, @@ -422,7 +318,7 @@ async fn resume_model_switch_is_not_duplicated_after_pre_turn_override() -> Resu let mut resume_builder = test_codex().with_config(|config| { config.model = Some("gpt-5.5".to_string()); }); - let resumed = resume_builder.resume(&server, home, rollout_path).await?; + let resumed = resume_builder.restart(&server, &initial).await?; core_test_support::submit_thread_settings( &resumed.codex, codex_protocol::protocol::ThreadSettingsOverrides { diff --git a/codex-rs/core/tests/suite/sqlite_state.rs b/codex-rs/core/tests/suite/sqlite_state.rs index 35c65a433c..7abde52e35 100644 --- a/codex-rs/core/tests/suite/sqlite_state.rs +++ b/codex-rs/core/tests/suite/sqlite_state.rs @@ -181,6 +181,7 @@ async fn resume_restores_dynamic_tools_from_rollout_with_sqlite_enabled() -> Res }) .await; + started.thread.shutdown_and_wait().await?; let mut resume_builder = test_codex().with_config(|config| { config .features diff --git a/codex-rs/thread-store/src/local/archive_thread.rs b/codex-rs/thread-store/src/local/archive_thread.rs index 656bc6766d..91af9dbdd3 100644 --- a/codex-rs/thread-store/src/local/archive_thread.rs +++ b/codex-rs/thread-store/src/local/archive_thread.rs @@ -29,21 +29,13 @@ pub(super) async fn archive_threads( let mut _live_writer_guards = Vec::with_capacity(lock_thread_ids.len()); for thread_id in &lock_thread_ids { _live_writer_guards.push(store.live_writer_locks.lock(*thread_id).await); - if store - .live_recorders - .lock() - .await - .get(thread_id) - .is_some_and(|entry| entry.writer_lock.is_some()) - { + if store.live_recorders.lock().await.contains_key(thread_id) { return Err(ThreadStoreError::Conflict { message: format!("thread {thread_id} already has an active writer"), }); } } - let _writer_guards = store - .acquire_paginated_writer_locks(&lock_thread_ids) - .await?; + let _writer_guards = store.acquire_writer_locks(&lock_thread_ids).await?; let parent_thread_id = thread_ids[0]; let mut archived_thread_ids = Vec::new(); @@ -173,42 +165,53 @@ mod tests { let home = TempDir::new().expect("temp dir"); let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); let owner = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); - let parent_uuid = Uuid::from_u128(203); - let parent_thread_id = - ThreadId::from_string(&parent_uuid.to_string()).expect("valid parent thread id"); - let parent_path = write_session_file_with_history_mode( - home.path(), - "2025-01-03T12-00-00", - parent_uuid, - ThreadHistoryMode::Paginated, - ) - .expect("parent session file"); - let child_uuid = Uuid::from_u128(204); - let child_thread_id = - ThreadId::from_string(&child_uuid.to_string()).expect("valid child thread id"); - let child_path = write_session_file_with_history_mode( - home.path(), - "2025-01-03T12-00-01", - child_uuid, - ThreadHistoryMode::Paginated, - ) - .expect("child session file"); - let _owner_guard = owner - .writer_lock_coordinator - .acquire(child_thread_id) - .expect("acquire child writer lock"); + for (parent_uuid, child_uuid, history_mode) in [ + ( + Uuid::from_u128(203), + Uuid::from_u128(204), + ThreadHistoryMode::Legacy, + ), + ( + Uuid::from_u128(206), + Uuid::from_u128(207), + ThreadHistoryMode::Paginated, + ), + ] { + let parent_thread_id = + ThreadId::from_string(&parent_uuid.to_string()).expect("valid parent thread id"); + let parent_path = write_session_file_with_history_mode( + home.path(), + "2025-01-03T12-00-00", + parent_uuid, + history_mode, + ) + .expect("parent session file"); + let child_thread_id = + ThreadId::from_string(&child_uuid.to_string()).expect("valid child thread id"); + let child_path = write_session_file_with_history_mode( + home.path(), + "2025-01-03T12-00-01", + child_uuid, + history_mode, + ) + .expect("child session file"); + let _owner_guard = owner + .writer_lock_coordinator + .acquire(child_thread_id) + .expect("acquire child writer lock"); - let error = store - .archive_threads(ArchiveThreadsParams { - thread_ids: vec![parent_thread_id, child_thread_id], - writer_lock_thread_ids: Vec::new(), - }) - .await - .expect_err("owned descendant should block archive"); + let error = store + .archive_threads(ArchiveThreadsParams { + thread_ids: vec![parent_thread_id, child_thread_id], + writer_lock_thread_ids: Vec::new(), + }) + .await + .expect_err("owned descendant should block archive"); - assert!(matches!(error, ThreadStoreError::Conflict { .. })); - assert!(parent_path.exists()); - assert!(child_path.exists()); + assert!(matches!(error, ThreadStoreError::Conflict { .. })); + assert!(parent_path.exists()); + assert!(child_path.exists()); + } } #[tokio::test] diff --git a/codex-rs/thread-store/src/local/delete_thread.rs b/codex-rs/thread-store/src/local/delete_thread.rs index 3347c1e44d..69c0ee9802 100644 --- a/codex-rs/thread-store/src/local/delete_thread.rs +++ b/codex-rs/thread-store/src/local/delete_thread.rs @@ -35,7 +35,7 @@ pub(super) async fn delete_thread( if reference_index.reference_count(thread_id) > 0 { return Err(referenced_thread_error(thread_id)); } - let mut writer_guards = store.acquire_paginated_writer_locks(&[thread_id]).await?; + let mut writer_guards = store.acquire_writer_locks(&[thread_id]).await?; delete_thread_after_reference_check(store, thread_id, &mut writer_guards).await } @@ -84,9 +84,7 @@ pub(super) async fn delete_threads( } } - let mut writer_guards = store - .acquire_paginated_writer_locks(&lock_thread_ids) - .await?; + let mut writer_guards = store.acquire_writer_locks(&lock_thread_ids).await?; for thread_id in thread_ids { match delete_thread_after_reference_check(store, thread_id, &mut writer_guards).await { Ok(()) | Err(ThreadStoreError::ThreadNotFound { .. }) => {} @@ -157,14 +155,8 @@ async fn delete_thread_after_reference_check( super::thread_history::delete_thread(store, thread_id).await?; // Drop the recorder before removing files, but retain its writer lock until cleanup finishes. - if let Some(writer_lock) = store - .live_recorders - .lock() - .await - .remove(&thread_id) - .and_then(|entry| entry.writer_lock) - { - writer_guards.push(writer_lock); + if let Some(entry) = store.live_recorders.lock().await.remove(&thread_id) { + writer_guards.push(entry.writer_lock); } let found_rollout_path = !rollout_paths.is_empty(); for rollout_path in rollout_paths { @@ -412,41 +404,52 @@ mod tests { let home = TempDir::new().expect("temp dir"); let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); let owner = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); - let parent_uuid = Uuid::from_u128(309); - let parent_thread_id = - ThreadId::from_string(&parent_uuid.to_string()).expect("valid parent thread id"); - let parent_path = write_session_file_with_history_mode( - home.path(), - "2025-01-03T12-00-00", - parent_uuid, - ThreadHistoryMode::Paginated, - ) - .expect("parent session file"); - let child_uuid = Uuid::from_u128(310); - let child_thread_id = - ThreadId::from_string(&child_uuid.to_string()).expect("valid child thread id"); - let child_path = write_session_file_with_history_mode( - home.path(), - "2025-01-03T12-00-01", - child_uuid, - ThreadHistoryMode::Paginated, - ) - .expect("child session file"); - let _owner_guard = owner - .writer_lock_coordinator - .acquire(child_thread_id) - .expect("acquire child writer lock"); + for (parent_uuid, child_uuid, history_mode) in [ + ( + Uuid::from_u128(309), + Uuid::from_u128(310), + ThreadHistoryMode::Legacy, + ), + ( + Uuid::from_u128(312), + Uuid::from_u128(313), + ThreadHistoryMode::Paginated, + ), + ] { + let parent_thread_id = + ThreadId::from_string(&parent_uuid.to_string()).expect("valid parent thread id"); + let parent_path = write_session_file_with_history_mode( + home.path(), + "2025-01-03T12-00-00", + parent_uuid, + history_mode, + ) + .expect("parent session file"); + let child_thread_id = + ThreadId::from_string(&child_uuid.to_string()).expect("valid child thread id"); + let child_path = write_session_file_with_history_mode( + home.path(), + "2025-01-03T12-00-01", + child_uuid, + history_mode, + ) + .expect("child session file"); + let _owner_guard = owner + .writer_lock_coordinator + .acquire(child_thread_id) + .expect("acquire child writer lock"); - let error = store - .delete_threads(DeleteThreadsParams { - thread_ids: vec![parent_thread_id, child_thread_id], - }) - .await - .expect_err("owned descendant should block deletion"); + let error = store + .delete_threads(DeleteThreadsParams { + thread_ids: vec![parent_thread_id, child_thread_id], + }) + .await + .expect_err("owned descendant should block deletion"); - assert!(matches!(error, ThreadStoreError::Conflict { .. })); - assert!(parent_path.exists()); - assert!(child_path.exists()); + assert!(matches!(error, ThreadStoreError::Conflict { .. })); + assert!(parent_path.exists()); + assert!(child_path.exists()); + } } #[tokio::test] diff --git a/codex-rs/thread-store/src/local/live_writer.rs b/codex-rs/thread-store/src/local/live_writer.rs index 1457cc9e55..fc54cea8c4 100644 --- a/codex-rs/thread-store/src/local/live_writer.rs +++ b/codex-rs/thread-store/src/local/live_writer.rs @@ -30,11 +30,7 @@ pub(super) async fn create_thread( let _live_writer_guard = store.live_writer_locks.lock(thread_id).await; let history_mode = params.history_mode; store.ensure_live_recorder_absent(thread_id).await?; - let writer_lock = if matches!(history_mode, ThreadHistoryMode::Paginated) { - Some(store.writer_lock_coordinator.acquire(thread_id)?) - } else { - None - }; + let writer_lock = store.writer_lock_coordinator.acquire(thread_id)?; let recorder = create_thread::create_thread(store, params).await?; store .insert_live_recorder(thread_id, recorder, history_mode, writer_lock) @@ -47,6 +43,7 @@ pub(super) async fn resume_thread( ) -> ThreadStoreResult<()> { let _live_writer_guard = store.live_writer_locks.lock(params.thread_id).await; store.ensure_live_recorder_absent(params.thread_id).await?; + let writer_lock = store.writer_lock_coordinator.acquire(params.thread_id)?; let history_mode = if let Some(history) = params.history.as_deref() { canonical_history_mode_from_rollout_items(history) } else if let Some(rollout_path) = params.rollout_path.as_ref() { @@ -103,11 +100,6 @@ pub(super) async fn resume_thread( model_provider_id: params.metadata.model_provider.clone(), generate_memories: matches!(params.metadata.memory_mode, ThreadMemoryMode::Enabled), }; - let writer_lock = if matches!(history_mode, ThreadHistoryMode::Paginated) { - Some(store.writer_lock_coordinator.acquire(params.thread_id)?) - } else { - None - }; let recorder = RolloutRecorder::new(&config, RolloutRecorderParams::resume(rollout_path)) .await .map_err(|err| ThreadStoreError::Internal { diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 42aae5fd06..23f00a205b 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -107,7 +107,7 @@ struct LiveRecorderEntry { // canonical SessionMeta is durable. Retain the mode captured when live persistence was opened // so missing SQLite rows can still be seeded. history_mode: ThreadHistoryMode, - writer_lock: Option, + writer_lock: WriterLockGuard, } #[derive(Default)] @@ -264,38 +264,16 @@ impl LocalThreadStore { Ok(()) } - async fn acquire_paginated_writer_locks( + async fn acquire_writer_locks( &self, thread_ids: &[ThreadId], ) -> ThreadStoreResult> { - let mut writer_locks = Vec::new(); + let mut writer_locks = Vec::with_capacity(thread_ids.len()); for &thread_id in thread_ids { - if self - .live_recorders - .lock() - .await - .get(&thread_id) - .is_some_and(|entry| entry.writer_lock.is_some()) - { + if self.live_recorders.lock().await.contains_key(&thread_id) { continue; } - - // Only a readable legacy header proves no paginated writer can own this id. Missing - // lazy rollouts and damaged headers must conservatively try the lock. - let history_mode = match read_thread::resolve_rollout_path( - self, thread_id, /*include_archived*/ true, - ) - .await? - { - Some(rollout_path) => codex_rollout::read_session_meta_line(rollout_path.as_path()) - .await - .ok() - .map(|meta_line| meta_line.meta.history_mode), - None => None, - }; - if !matches!(history_mode, Some(ThreadHistoryMode::Legacy)) { - writer_locks.push(self.writer_lock_coordinator.acquire(thread_id)?); - } + writer_locks.push(self.writer_lock_coordinator.acquire(thread_id)?); } Ok(writer_locks) } @@ -305,7 +283,7 @@ impl LocalThreadStore { thread_id: ThreadId, recorder: RolloutRecorder, history_mode: ThreadHistoryMode, - writer_lock: Option, + writer_lock: WriterLockGuard, ) -> ThreadStoreResult<()> { match self.live_recorders.lock().await.entry(thread_id) { Entry::Occupied(entry) => Err(ThreadStoreError::InvalidRequest { @@ -1200,6 +1178,8 @@ mod tests { async fn create_thread_rejects_missing_cwd() { let home = TempDir::new().expect("temp dir"); let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); + let competing_store = + LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); for history_mode in [ThreadHistoryMode::Legacy, ThreadHistoryMode::Paginated] { let thread_id = ThreadId::default(); let mut params = create_thread_params(thread_id); @@ -1219,10 +1199,10 @@ mod tests { let mut valid_params = create_thread_params(thread_id); valid_params.history_mode = history_mode; - store + competing_store .create_thread(valid_params) .await - .expect("failed initialization should release writer ownership"); + .expect("failed initialization should release cross-process writer ownership"); } } @@ -1249,10 +1229,7 @@ mod tests { .path() .join("thread-writer-locks") .join(format!("{thread_id}.lock")); - assert_eq!( - lock_path.exists(), - matches!(history_mode, ThreadHistoryMode::Paginated) - ); + assert!(lock_path.exists()); store .discard_thread(thread_id) .await @@ -1335,6 +1312,61 @@ mod tests { assert_rollout_contains_message(rollout_path.as_path(), "after resume").await; } + #[tokio::test] + async fn live_writers_reject_cross_process_create_and_resume() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let primary = LocalThreadStore::new(config.clone(), /*state_db*/ None); + let secondary = LocalThreadStore::new(config, /*state_db*/ None); + + for history_mode in [ThreadHistoryMode::Legacy, ThreadHistoryMode::Paginated] { + let thread_id = ThreadId::default(); + let mut create_params = create_thread_params(thread_id); + create_params.history_mode = history_mode; + + primary + .create_thread(create_params.clone()) + .await + .expect("create live thread"); + primary + .persist_thread(thread_id) + .await + .expect("persist thread for resume"); + let rollout_path = primary + .live_rollout_path(thread_id) + .await + .expect("load rollout path"); + let resume_params = ResumeThreadParams { + thread_id, + rollout_path: Some(rollout_path), + history: None, + include_archived: true, + metadata: thread_metadata(), + }; + + let error = secondary + .create_thread(create_params) + .await + .expect_err("competing create should fail"); + assert!(matches!(error, ThreadStoreError::Conflict { .. })); + + let error = secondary + .resume_thread(resume_params.clone()) + .await + .expect_err("competing resume should fail"); + assert!(matches!(error, ThreadStoreError::Conflict { .. })); + + primary + .shutdown_thread(thread_id) + .await + .expect("shutdown should release writer ownership"); + secondary + .resume_thread(resume_params) + .await + .expect("resume after shutdown should acquire writer ownership"); + } + } + #[tokio::test] async fn create_thread_rejects_duplicate_live_writer() { let home = TempDir::new().expect("temp dir"); @@ -1387,14 +1419,21 @@ mod tests { async fn resume_thread_rejects_missing_cwd() { let home = TempDir::new().expect("temp dir"); let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); - let uuid = uuid::Uuid::from_u128(407); + let competing_store = + LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); + let uuid = uuid::Uuid::from_u128(408); let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); - let rollout_path = - write_session_file(home.path(), "2025-01-04T11-30-00", uuid).expect("session file"); + let rollout_path = write_session_file_with_history_mode( + home.path(), + "2025-01-04T11-30-00", + uuid, + ThreadHistoryMode::Paginated, + ) + .expect("session file"); let err = store .resume_thread(ResumeThreadParams { thread_id, - rollout_path: Some(rollout_path), + rollout_path: Some(rollout_path.clone()), history: None, include_archived: true, metadata: ThreadPersistenceMetadata { @@ -1408,6 +1447,17 @@ mod tests { assert!(matches!(err, ThreadStoreError::InvalidRequest { .. })); assert!(err.to_string().contains("requires a cwd")); + + competing_store + .resume_thread(ResumeThreadParams { + thread_id, + rollout_path: Some(rollout_path), + history: None, + include_archived: true, + metadata: thread_metadata(), + }) + .await + .expect("failed initialization should release cross-process writer ownership"); } #[tokio::test]