diff --git a/codex-rs/core/tests/suite/fork_thread.rs b/codex-rs/core/tests/suite/fork_thread.rs index 2d1532d678..831bfed7dd 100644 --- a/codex-rs/core/tests/suite/fork_thread.rs +++ b/codex-rs/core/tests/suite/fork_thread.rs @@ -371,17 +371,17 @@ fn latest_multi_agent_version( std::fs::read_to_string(path) .unwrap_or_else(|err| panic!("failed to read rollout file {}: {err}", path.display())) .lines() + .rev() .filter(|line| !line.trim().is_empty()) .map(|line| { serde_json::from_str::(line) .unwrap_or_else(|err| panic!("failed to parse rollout line `{line}`: {err}")) }) - .filter_map(|line| match line.item { + .find_map(|line| match line.item { RolloutItem::SessionMeta(meta_line) if meta_line.meta.id == thread_id => { Some(meta_line.meta.multi_agent_version) } _ => None, }) - .last() .flatten() } diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index ea5cf8613f..c7fd7705f5 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -21,6 +21,7 @@ use std::collections::hash_map::Entry; use std::path::PathBuf; use std::sync::Arc; use tokio::sync::Mutex; +use tokio::sync::Semaphore; use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; @@ -59,7 +60,7 @@ use codex_protocol::protocol::MultiAgentVersion; pub struct LocalThreadStore { pub(super) config: LocalThreadStoreConfig, live_recorders: Arc>>, - multi_agent_version_seed_lock: Arc>, + multi_agent_version_seed_semaphore: Arc, state_db: Option, } @@ -99,7 +100,7 @@ impl LocalThreadStore { Self { config, live_recorders: Arc::new(Mutex::new(HashMap::new())), - multi_agent_version_seed_lock: Arc::new(Mutex::new(())), + multi_agent_version_seed_semaphore: Arc::new(Semaphore::new(1)), state_db, } } diff --git a/codex-rs/thread-store/src/local/set_multi_agent_version.rs b/codex-rs/thread-store/src/local/set_multi_agent_version.rs index d077d073c3..d8ea67ab1a 100644 --- a/codex-rs/thread-store/src/local/set_multi_agent_version.rs +++ b/codex-rs/thread-store/src/local/set_multi_agent_version.rs @@ -16,7 +16,13 @@ pub(super) async fn set_multi_agent_version_if_unset( store: &LocalThreadStore, params: SetMultiAgentVersionIfUnsetParams, ) -> ThreadStoreResult { - let _guard = store.multi_agent_version_seed_lock.lock().await; + let _permit = store + .multi_agent_version_seed_semaphore + .acquire() + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to lock thread multi-agent version: {err}"), + })?; let has_live_writer = live_writer::rollout_path(store, params.thread_id) .await .is_ok();