openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-18 17:18:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Enforce single-writer ownership for all thread histories (#36389)

Browse Source 
## 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
This commit is contained in:
Owen Lin
2026-07-31 20:49:08 +00:00
committed by copyberry
parent 1bef168976
commit dc60dadce6
16 changed files with 240 additions and 366 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+17 -39
View File
@@ -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()
codex-rs/core/tests/common/test_codex.rs
+15
View File
@@ -549,6 +549,21 @@ impl TestCodexBuilder {
.await
}
pub async fn restart(
&mut self,
server: &MockServer,
previous: &TestCodex,
) -> Result<TestCodex> {
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,
codex-rs/core/tests/suite/collaboration_instructions.rs
+1 -7
View File
@@ -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 {
codex-rs/core/tests/suite/compact.rs
+1 -10
View File
@@ -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![
codex-rs/core/tests/suite/hooks.rs
+2 -14
View File
@@ -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?;
codex-rs/core/tests/suite/mcp_tool_exposure.rs
+2 -4
View File
@@ -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")
codex-rs/core/tests/suite/model_visible_layout.rs
+2 -14
View File
@@ -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();
codex-rs/core/tests/suite/multi_agent_mode.rs
+4 -16
View File
@@ -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();
codex-rs/core/tests/suite/multi_agent_resume.rs
+4 -8
View File
@@ -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]
codex-rs/core/tests/suite/permissions_messages.rs
+2 -9
View File
@@ -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 {
codex-rs/core/tests/suite/resume.rs
+4 -108
View File
@@ -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<TempDir>,
rollout_path: PathBuf,
predicate: impl Fn(&[EventMsg]) -> bool,
) -> Result<TestCodex> {
let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
let poll_interval = Duration::from_millis(10);
let mut last_initial_messages = "<missing 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 {
codex-rs/core/tests/suite/sqlite_state.rs
+1
View File
@@ -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
codex-rs/thread-store/src/local/archive_thread.rs
+47 -44
View File
@@ -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]
codex-rs/thread-store/src/local/delete_thread.rs
+48 -45
View File
@@ -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]
codex-rs/thread-store/src/local/live_writer.rs
+2 -10
View File
@@ -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 {
codex-rs/thread-store/src/local/mod.rs
+88 -38
View File
@@ -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<WriterLockGuard>,
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<Vec<WriterLockGuard>> {
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<WriterLockGuard>,
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]