diff --git a/codex-rs/core/src/agent/control/runtime.rs b/codex-rs/core/src/agent/control/runtime.rs index b48740f4aa..1672d2aee7 100644 --- a/codex-rs/core/src/agent/control/runtime.rs +++ b/codex-rs/core/src/agent/control/runtime.rs @@ -8,8 +8,6 @@ use crate::agent::api::AgentControl; use crate::agent::registry::AgentRegistry; use crate::config::RolloutBudgetConfig; use crate::rollout_budget::RolloutBudget; -use crate::session::session::Session; -use crate::state::ContextWindowIdGenerator; use crate::thread_manager::AgentTreeShutdownFailure; use crate::thread_manager::AgentTreeShutdownFailureReason; use crate::thread_manager::AgentTreeShutdownReport; @@ -18,7 +16,6 @@ use crate::thread_manager::ThreadManagerState; use arc_swap::ArcSwapOption; use codex_extension_api::ThreadInstructionsProvider; use codex_protocol::AgentPath; -use codex_protocol::ItemIdGenerator; use codex_protocol::SessionId; use codex_protocol::ThreadId; use codex_protocol::error::AgentErrorContext; @@ -195,8 +192,6 @@ pub(crate) struct LocalAgentRuntime { pub(super) manager: Weak, /// Captured at construction so delegates retain their manager's allocation policy. pub(super) thread_id_generator: ThreadIdGenerator, - pub(crate) item_id_generator: ItemIdGenerator, - pub(crate) context_window_id_generator: ContextWindowIdGenerator, pub(super) agent_execution_limiter: Arc, /// Session-scoped state shared by the root thread and every cloned sub-agent control handle. pub(super) rollout_budget: Arc, @@ -222,8 +217,6 @@ impl LocalAgentRuntime { let runtime = Self { manager, thread_id_generator, - item_id_generator: Session::new_item_id_generator(), - context_window_id_generator: Arc::new(codex_protocol::ContextWindowId::new_uuid_v7), registry: Arc::default(), residency: Arc::default(), mailboxes: Arc::default(), diff --git a/codex-rs/core/src/agent/control/sender_context.rs b/codex-rs/core/src/agent/control/sender_context.rs index 39afaca3d6..b84082fff7 100644 --- a/codex-rs/core/src/agent/control/sender_context.rs +++ b/codex-rs/core/src/agent/control/sender_context.rs @@ -5,6 +5,7 @@ use crate::context::GuardianSenderMessages; use crate::context_manager::ContextManager; +use crate::session::session::Session; use codex_history::RolloutItem; use codex_protocol::ThreadId; use codex_protocol::models::ResponseItem; @@ -103,9 +104,7 @@ impl LocalAgentRuntime { ContextManager::reconstruct_rollout( &stored.items, meta.history_mode, - ContextManager::new(std::sync::Arc::clone( - &self.item_id_generator, - )), + ContextManager::new(Session::new_item_id_generator()), // Tool bodies are not sender authorization evidence. TruncationPolicy::Bytes(0), ) diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index ea95f784bd..b932d39dca 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -274,6 +274,7 @@ mod mcp_prewarm; mod mcp_refresh; mod mcp_runtime; mod model_catalog_settings; +mod model_ids; pub(crate) mod multi_agents; mod plugin_selection; mod realtime_history; @@ -601,6 +602,21 @@ pub(crate) fn assign_missing_item_id( } impl Session { + fn compaction_resume_metadata(&self, state: &SessionState) -> CompactionResumeMetadata { + CompactionResumeMetadata { + next_item_ordinal: None, + rejected_audio_file_ids: state.history.rejected_audio_file_ids(), + dynamic_tools: Some(Box::new(DynamicToolsSnapshot { + thread_id: self.thread_id(), + tools: state.persisted_dynamic_tools.to_vec(), + })), + multi_agent_version: self.multi_agent_version(), + last_started_turn_id: state.last_started_turn_id.clone(), + turn_attribution: state.turn_attribution.clone(), + previous_turn_settings: state.previous_turn_settings(), + } + } + pub(crate) async fn resolve_startup_model( models_manager: &dyn ModelsManager, config: &Config, @@ -4958,18 +4974,7 @@ impl Session { window_id: Some(metadata.window_ids.window_id), compaction_response_id: metadata.compaction_response_id, latest_token_usage_record: state.latest_token_usage_record.clone(), - resume_metadata: Some(CompactionResumeMetadata { - next_item_ordinal: None, - rejected_audio_file_ids: state.history.rejected_audio_file_ids(), - dynamic_tools: Some(Box::new(DynamicToolsSnapshot { - thread_id: self.thread_id(), - tools: state.persisted_dynamic_tools.to_vec(), - })), - multi_agent_version: self.multi_agent_version(), - last_started_turn_id: state.last_started_turn_id.clone(), - turn_attribution: state.turn_attribution.clone(), - previous_turn_settings: state.previous_turn_settings(), - }), + resume_metadata: Some(self.compaction_resume_metadata(&state)), }; (compacted_item, started_item) }; diff --git a/codex-rs/core/src/session/model_ids.rs b/codex-rs/core/src/session/model_ids.rs new file mode 100644 index 0000000000..6214c5ba7c --- /dev/null +++ b/codex-rs/core/src/session/model_ids.rs @@ -0,0 +1,52 @@ +//! Creates this thread's item and window sequences from the history it will actually replay. + +use std::sync::Arc; + +use codex_history::InitialHistory; +use codex_history::RolloutItem; +use codex_protocol::ContextWindowId; +use codex_protocol::ItemIdGenerator; + +use super::session::Session; +use crate::state::AutoCompactWindowIds; +use crate::state::ContextWindowIdGenerator; + +pub(super) struct InitialModelIds { + pub item_id_generator: ItemIdGenerator, + pub context_window_id_generator: ContextWindowIdGenerator, + pub initial_auto_compact_window_ids: AutoCompactWindowIds, +} + +pub(super) fn from_history(history: &InitialHistory) -> InitialModelIds { + // A resume already owns its window. Allocating a provisional one would skip an ordinal. + let resumed_window = match history { + InitialHistory::Resumed(resumed) => { + resumed + .history + .iter() + .rev() + .find_map(|record| match record { + RolloutItem::Compacted(checkpoint) => { + checkpoint.window_id.as_ref()?.validated() + } + RolloutItem::SessionMeta(meta) if meta.meta.id == resumed.conversation_id => { + meta.meta.context_window.as_ref()?.window_id.validated() + } + _ => None, + }) + } + InitialHistory::New | InitialHistory::Cleared | InitialHistory::Forked(_) => None, + }; + let initialize = |item_id_generator, context_window_id_generator: ContextWindowIdGenerator| { + let initial = resumed_window.unwrap_or_else(|| context_window_id_generator()); + InitialModelIds { + item_id_generator, + context_window_id_generator, + initial_auto_compact_window_ids: AutoCompactWindowIds::new_initial(initial), + } + }; + initialize( + Session::new_item_id_generator(), + Arc::new(ContextWindowId::new_uuid_v7), + ) +} diff --git a/codex-rs/core/src/session/rollout_reconstruction.rs b/codex-rs/core/src/session/rollout_reconstruction.rs index 1df79fc656..b5e37406b3 100644 --- a/codex-rs/core/src/session/rollout_reconstruction.rs +++ b/codex-rs/core/src/session/rollout_reconstruction.rs @@ -200,7 +200,9 @@ impl Session { turn_context.root_capabilities_restricted, &turn_context.config.features, turn_context.config.guardian_context_budgeting(), - std::sync::Arc::clone(&self.item_id_generator), + // Reconstructing retained evidence may mint temporary revisions before restoring + // the durable originals. Replay must not consume the live thread's item sequence. + Self::new_item_id_generator(), ), turn_context.model_info().truncation_policy.into(), ) diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 822f0ec90e..6ab58bc676 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1008,11 +1008,31 @@ impl Session { session_configuration.thread_source.as_ref(), Some(ThreadSource::Subagent | ThreadSource::GuardianReview) ); - let item_id_generator = Arc::clone(&agent_control.runtime().item_id_generator); - if let InitialHistory::Forked(items) = &mut initial_history { - for record in items { - if let RolloutItem::ResponseItem(item) = record { - assign_missing_item_id(&mut item.item, &item_id_generator); + let prepare_ids = super::model_ids::from_history; + // New threads need their window ID to create the rollout. Resumes initialize once the + // writer has validated or refreshed the history snapshot. + let new_thread_ids = (!matches!(initial_history, InitialHistory::Resumed(_))) + .then(|| prepare_ids(&initial_history)); + let restore_child_window = matches!(&initial_history, InitialHistory::Forked(_)) + && session_configuration.session_source.is_non_root_agent() + && config.features.enabled(Feature::TokenBudget); + if let (InitialHistory::Forked(records), Some(ids)) = + (&mut initial_history, &new_thread_ids) + { + for record in records { + match record { + RolloutItem::ResponseItem(item) => { + assign_missing_item_id(&mut item.item, &ids.item_id_generator) + } + RolloutItem::Compacted(checkpoint) if restore_child_window => { + checkpoint.window_number = Some(0); + checkpoint.first_window_id = + Some(ids.initial_auto_compact_window_ids.window_id.clone()); + checkpoint.previous_window_id = None; + checkpoint.window_id = + Some(ids.initial_auto_compact_window_ids.window_id.clone()); + } + _ => {} } } } @@ -1096,24 +1116,6 @@ impl Session { SessionId::from(thread_id) } }); - let context_window_id_generator = - Arc::clone(&agent_control.runtime().context_window_id_generator); - let initial_auto_compact_window_ids = - AutoCompactWindowIds::new_initial(context_window_id_generator()); - let restore_child_window = matches!(&initial_history, InitialHistory::Forked(_)) - && session_configuration.session_source.is_non_root_agent() - && config.features.enabled(Feature::TokenBudget); - if restore_child_window && let InitialHistory::Forked(items) = &mut initial_history { - let child_window_id = initial_auto_compact_window_ids.window_id.clone(); - for item in items { - if let RolloutItem::Compacted(checkpoint) = item { - checkpoint.window_number = Some(0); - checkpoint.first_window_id = Some(child_window_id.clone()); - checkpoint.previous_window_id = None; - checkpoint.window_id = Some(child_window_id.clone()); - } - } - } let has_provided_agent_control = matches!(&agent_control, AgentControlInit::Provided { .. }); let (agent_control, local_agent_runtime): (Arc, _) = match agent_control { @@ -1215,6 +1217,9 @@ impl Session { let guard = managed_guard.as_deref_mut().unwrap_or(&mut local_guard); let live_thread = match &initial_history { InitialHistory::New | InitialHistory::Cleared | InitialHistory::Forked(_) => { + let ids = new_thread_ids.as_ref().ok_or_else(|| { + anyhow::anyhow!("new thread missing its initial window") + })?; let auth = persistence_auth.await; let params = CreateThreadParams { attached_thread, @@ -1241,7 +1246,10 @@ impl Session { ForkPersistence::Referenced { history_base, .. } => *history_base, }, subagent_history_start_ordinal: None, - initial_window_id: initial_auto_compact_window_ids.window_id.clone(), + initial_window_id: ids + .initial_auto_compact_window_ids + .window_id + .clone(), runtime_workspace_roots: Some(config.workspace_roots.clone()), metadata: ThreadPersistenceMetadata { cwd: Some(config.cwd.to_path_buf()), @@ -1401,6 +1409,11 @@ impl Session { resumed.history = history; resumed.history_revision = None; } + let super::model_ids::InitialModelIds { + item_id_generator, + context_window_id_generator, + initial_auto_compact_window_ids, + } = new_thread_ids.unwrap_or_else(|| prepare_ids(&initial_history)); let rollout_path = if let Some(live_thread) = live_thread.as_ref() { live_thread.local_rollout_path().await? } else { diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index e05db21e83..fd53a60943 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -6789,7 +6789,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { config.model_provider.clone(), ); let agent_control = LocalAgentControl::default(); - let item_id_generator = Arc::clone(&agent_control.runtime.item_id_generator); + let item_id_generator = Session::new_item_id_generator(); let exec_policy = Arc::new(ExecPolicyManager::default()); let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit); let model = get_model_offline_for_tests(config.model.as_deref()); @@ -9134,7 +9134,7 @@ where config.model_provider.clone(), ); let agent_control = LocalAgentControl::default(); - let item_id_generator = Arc::clone(&agent_control.runtime.item_id_generator); + let item_id_generator = Session::new_item_id_generator(); let exec_policy = Arc::new(ExecPolicyManager::default()); let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit); let model = get_model_offline_for_tests(config.model.as_deref()); diff --git a/codex-rs/core/tests/suite/multi_agent_restore_tests.rs b/codex-rs/core/tests/suite/multi_agent_restore_tests.rs index e616b4a922..0c9167da37 100644 --- a/codex-rs/core/tests/suite/multi_agent_restore_tests.rs +++ b/codex-rs/core/tests/suite/multi_agent_restore_tests.rs @@ -302,7 +302,7 @@ async fn cold_root_resume_overlaps_child_reads_and_applies_identities_in_graph_o .thread_manager .ensure_multi_agent_v2_child_loaded(first) .await?; - assert!(resumed.thread_manager.get_thread(first).await.is_ok()); + let first_child = resumed.thread_manager.get_thread(first).await?; assert!(resumed.thread_manager.get_thread(last).await.is_err()); assert!(resumed.thread_manager.get_thread(failed).await.is_err()); @@ -321,7 +321,7 @@ async fn cold_root_resume_overlaps_child_reads_and_applies_identities_in_graph_o ) .await; resumed.submit_turn("continue restored child").await?; - let child = resumed.thread_manager.get_thread(first).await?; + let child = first_child; wait_for_event(child.as_ref(), |event| { matches!(event, EventMsg::TurnComplete(_)) }) diff --git a/codex-rs/core/tests/suite/scenarios.rs b/codex-rs/core/tests/suite/scenarios.rs index 40513c867e..171054ad85 100644 --- a/codex-rs/core/tests/suite/scenarios.rs +++ b/codex-rs/core/tests/suite/scenarios.rs @@ -513,13 +513,11 @@ async fn gif_generation_preserves_animation_and_replays_a_still_preview() -> Res async fn admitted_prompt_id_reaches_responses() -> Result<()> { skip_if_no_network!(Ok(())); let server = start_mock_server().await; - let test = test_codex() - .with_config(|config| { - configure_scenario_catalog(config); - config.include_environment_context = false; - }) - .build_with_auto_env(&server) - .await?; + let mut builder = test_codex().with_config(|config| { + configure_scenario_catalog(config); + config.include_environment_context = false; + }); + let test = builder.build_with_auto_env(&server).await?; let mock = mount_sse_sequence( &server, vec![ diff --git a/codex-rs/history/src/lib.rs b/codex-rs/history/src/lib.rs index 54792bf89b..cee005424e 100644 --- a/codex-rs/history/src/lib.rs +++ b/codex-rs/history/src/lib.rs @@ -292,6 +292,7 @@ impl JsonSchema for RolloutItem { mod guardian_history; mod model_id_ordinals; pub use model_id_ordinals::next_model_id_ordinals; +pub use model_id_ordinals::stamp_item_ordinals; mod reconciled_retained_context; mod retained_context; mod sender_user_messages; diff --git a/codex-rs/history/src/model_id_ordinals.rs b/codex-rs/history/src/model_id_ordinals.rs index 41e397e9d1..2b3bde1b97 100644 --- a/codex-rs/history/src/model_id_ordinals.rs +++ b/codex-rs/history/src/model_id_ordinals.rs @@ -23,22 +23,25 @@ pub fn next_model_id_ordinals(records: &[RolloutItem]) -> (u64, u64) { .as_ref() .and_then(|m| m.next_item_ordinal) { - item = next.checked_sub(1); + item = item.max(next.checked_sub(1)); + } + for entry in checkpoint.replacement_history.iter().flatten().chain( + checkpoint + .guardian_history + .iter() + .flat_map(|history| &history.0), + ) { + item = item.max(envelope_ordinal(entry)); + } + if let Some(context) = &checkpoint.retained_context { + item = item.max(context.max_source_ordinal()); } if let Some(id) = &checkpoint.window_id { window = window.max(id.allocation().map(ModelFacingId::ordinal)); } } RolloutItem::ResponseItem(envelope) => { - item = item.max(envelope.item.id().and_then(item_ordinal)); - // The retained revision is local. Guardian delivery metadata quotes other threads. - if let Some(source) = envelope - .metadata - .as_ref() - .and_then(|m| m.retained_source.as_ref()) - { - item = item.max(item_ordinal(&source.revision)); - } + item = item.max(envelope_ordinal(envelope)); } RolloutItem::InterAgentCommunication(message) => { item = item.max(message.id.as_ref().and_then(item_ordinal)); @@ -64,7 +67,31 @@ pub fn next_model_id_ordinals(records: &[RolloutItem]) -> (u64, u64) { (next(item), next(window)) } -fn item_ordinal(id: &ResponseItemId) -> Option { +/// Advances the durable cursor in append order, including allocations first seen in a checkpoint. +pub fn stamp_item_ordinals(mut next: u64, items: &mut [RolloutItem]) -> u64 { + for item in items { + next = next.max(next_model_id_ordinals(std::slice::from_ref(item)).0); + if let RolloutItem::Compacted(checkpoint) = item + && let Some(metadata) = &mut checkpoint.resume_metadata + && next != 0 + { + metadata.next_item_ordinal = Some(next); + } + } + next +} + +fn envelope_ordinal(entry: &crate::ResponseItemEnvelope) -> Option { + entry.item.id().and_then(item_ordinal).max( + entry + .metadata + .as_ref() + .and_then(|metadata| metadata.retained_source.as_ref()) + .and_then(|source| item_ordinal(&source.revision)), + ) +} + +pub(super) fn item_ordinal(id: &ResponseItemId) -> Option { id.allocation() .map(|allocation| allocation.model_id.ordinal()) } diff --git a/codex-rs/history/src/retained_context.rs b/codex-rs/history/src/retained_context.rs index 50782c183c..9868b3af7a 100644 --- a/codex-rs/history/src/retained_context.rs +++ b/codex-rs/history/src/retained_context.rs @@ -555,6 +555,19 @@ impl RetainedContext { } } + pub(super) fn max_source_ordinal(&self) -> Option { + self.user_messages + .iter() + .chain(&self.assistant_messages) + .filter_map(|entry| { + entry + .revision + .as_ref() + .and_then(crate::model_id_ordinals::item_ordinal) + }) + .max() + } + /// Restoring a saved thread must not bypass the live storage limits. /// A missing checkpoint cannot establish complete historical user instructions. /// Surviving local input metadata also advances the counter when its retained record is missing. diff --git a/codex-rs/history/src/tests.rs b/codex-rs/history/src/tests.rs index a8965d8e2e..7e9fadb84d 100644 --- a/codex-rs/history/src/tests.rs +++ b/codex-rs/history/src/tests.rs @@ -1200,7 +1200,7 @@ fn model_id_ordinals_resume_at_the_checkpoint_and_advance_for_local_items() -> R "guardian_review_ids": [{"id": "review_item008", "id_allocation": {"prefix": "review", "salt": "item", "ordinal": 301, "value": 8}}] }); - let records: Vec = serde_json::from_value(json!([ + let mut records: Vec = serde_json::from_value(json!([ record(19), {"type": "compacted", "payload": { "message": "", "window_number": 4, @@ -1214,6 +1214,9 @@ fn model_id_ordinals_resume_at_the_checkpoint_and_advance_for_local_items() -> R assert_eq!(crate::next_model_id_ordinals(&records[1..2]), (24, 5)); assert_eq!(crate::next_model_id_ordinals(&records[1..]), (26, 5)); assert_eq!(crate::next_model_id_ordinals(&records), (26, 5)); + // An append containing a checkpoint and a later allocation stamps only the prefix. + assert_eq!(crate::stamp_item_ordinals(/*next*/ 25, &mut records), 26); + assert_eq!(crate::next_model_id_ordinals(&records[1..2]), (25, 5)); Ok(()) } diff --git a/codex-rs/protocol/src/item_id_generator.rs b/codex-rs/protocol/src/item_id_generator.rs index 60472a3c50..04f0d17ad8 100644 --- a/codex-rs/protocol/src/item_id_generator.rs +++ b/codex-rs/protocol/src/item_id_generator.rs @@ -1,9 +1,9 @@ -//! Response item ID generation shared by sessions and their consumers. +//! Response item ID allocation for a thread and its consumers. use crate::ResponseItemId; use std::sync::Arc; -/// Generates response item IDs from a type prefix for an owning session. Clones share the same callable. +/// Generates IDs for an item kind. Clones share the owning thread's sequence. pub type ItemIdGenerator = Arc ResponseItemId + Send + Sync>; #[cfg(test)] diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 9f640a526c..dad54bb1e6 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -36,7 +36,9 @@ pub use codex_history::RetainedContextEvent; pub use codex_history::RetainedInputSource; pub use codex_history::RolloutItem; pub use codex_history::RolloutLine; +pub use codex_history::next_model_id_ordinals; pub use codex_history::resume_multi_agent_version; +pub use codex_history::stamp_item_ordinals; pub(crate) use codex_protocol::protocol; /// Decodes a persisted rollout record without Serde's flattened-envelope buffering. diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index 6939111b81..6ed5b6a3b7 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -665,16 +665,22 @@ impl InMemoryThreadStore { } let mut state = self.state.lock().await; let history_mode = history_mode_from_state(&state, params.thread_id); - let persisted_items = into_persisted_rollout_items(params.items, history_mode); + let mut persisted_items = into_persisted_rollout_items(params.items, history_mode); if persisted_items.is_empty() { return Ok(()); } state.calls.append_items += 1; - state - .histories - .entry(params.thread_id) - .or_default() - .extend(persisted_items); + let history = state.histories.entry(params.thread_id).or_default(); + if persisted_items + .iter() + .any(|item| matches!(item, RolloutItem::Compacted(_))) + { + codex_rollout::stamp_item_ordinals( + codex_rollout::next_model_id_ordinals(history).0, + &mut persisted_items, + ); + } + history.extend(persisted_items); Ok(()) } diff --git a/codex-rs/thread-store/src/local/live_writer.rs b/codex-rs/thread-store/src/local/live_writer.rs index 04a623930d..0ffd8c7276 100644 --- a/codex-rs/thread-store/src/local/live_writer.rs +++ b/codex-rs/thread-store/src/local/live_writer.rs @@ -28,6 +28,15 @@ pub(super) async fn create_thread( params: CreateThreadParams, ) -> ThreadStoreResult<()> { let thread_id = params.thread_id; + let inherited_cursor = if params.initial_window_id.allocation().is_some() + && let (Some(parent), Some(base)) = (params.forked_from_id, params.history_base) + { + let lineage = store.resolve_rollout_lineage_for_reference(parent).await?; + let prefix = super::model_context::load_for_fork(lineage, Some(base)).await?; + codex_rollout::next_model_id_ordinals(&prefix).0 + } else { + 0 + }; let attachment = if params.attached_thread { let parent = params .forked_from_id @@ -53,7 +62,7 @@ pub(super) async fn create_thread( } else { None }; - let _live_writer_guard = store.live_writer_locks.lock(thread_id).await; + let mut cursor = 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 = store.acquire_writer_lock(thread_id)?; @@ -69,14 +78,16 @@ pub(super) async fn create_thread( } store .insert_live_recorder(thread_id, recorder, thread_id, history_mode, writer_lock) - .await + .await?; + *cursor = inherited_cursor; + Ok(()) } pub(super) async fn resume_thread( store: &LocalThreadStore, params: ResumeThreadParams, ) -> ThreadStoreResult>> { - let _live_writer_guard = store.live_writer_locks.lock(params.thread_id).await; + let mut cursor = store.live_writer_locks.lock(params.thread_id).await; store.ensure_live_recorder_absent(params.thread_id).await?; let writer_lock = store.acquire_writer_lock(params.thread_id)?; let rollout_path = match params.rollout_path { @@ -167,6 +178,7 @@ pub(super) async fn resume_thread( writer_lock, ) .await?; + *cursor = codex_rollout::next_model_id_ordinals(&history).0; Ok(history) } @@ -420,27 +432,30 @@ async fn write_and_project( // Every live write should have a recorder: create/resume installs one, while // shutdown/discard/delete removes it. Keep the lookup defensive so late writes fail after // teardown. - let _live_writer_guard = store.live_writer_locks.lock(thread_id).await; + let mut cursor = store.live_writer_locks.lock(thread_id).await; let (recorder, rollout_id, history_mode) = live_writer_parts(store, thread_id).await?; let sync_rollout_path = matches!(&write_op, RolloutWriteOp::Persist | RolloutWriteOp::Flush); - let write_op = match write_op { + match write_op { RolloutWriteOp::AppendItems(items) => { - let items = into_persisted_rollout_items(items, history_mode); + let mut items = into_persisted_rollout_items(items, history_mode); if items.is_empty() { return Ok(()); } - RolloutWriteOp::AppendItems(items) + let next = codex_rollout::stamp_item_ordinals(*cursor, &mut items); + recorder + .record_canonical_items(&items) + .await + .map_err(thread_store_io_error)?; + // Once queued, the recorder owns the batch even if flush is cancelled or retried. + *cursor = next; + recorder.flush().await.map_err(thread_store_io_error)?; } - RolloutWriteOp::Persist => RolloutWriteOp::Persist, - RolloutWriteOp::Flush => RolloutWriteOp::Flush, - }; - if matches!(history_mode, ThreadHistoryMode::Legacy) { - durable_write(&recorder, write_op).await?; - } else { + RolloutWriteOp::Persist => recorder.persist().await.map_err(thread_store_io_error)?, + RolloutWriteOp::Flush => recorder.flush().await.map_err(thread_store_io_error)?, + } + if matches!(history_mode, ThreadHistoryMode::Paginated) { + // SQLite is a rebuildable view and must never get ahead of canonical history. let rollout_path = recorder.rollout_path(); - // SQLite is a rebuildable view. The flush barrier must win before projection starts so it - // can lag JSONL after failure, but can never get ahead of canonical history. - durable_write(&recorder, write_op).await?; if let Err(err) = super::thread_history_materialization::materialize_to_sqlite( store, rollout_id, @@ -456,17 +471,3 @@ async fn write_and_project( } Ok(()) } - -async fn durable_write(recorder: &RolloutRecorder, write: RolloutWriteOp) -> ThreadStoreResult<()> { - match write { - RolloutWriteOp::AppendItems(items) => { - recorder - .record_canonical_items(items.as_slice()) - .await - .map_err(thread_store_io_error)?; - recorder.flush().await.map_err(thread_store_io_error) - } - RolloutWriteOp::Persist => recorder.persist().await.map_err(thread_store_io_error), - RolloutWriteOp::Flush => recorder.flush().await.map_err(thread_store_io_error), - } -} diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 680a9de531..c30641e7a4 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -187,8 +187,8 @@ struct LiveWriterLocks { #[derive(Default)] struct ThreadCoordination { - // Serialize writes and capture consistent fork snapshots. - writer: Arc>, + // Serializes writes and holds the next item ordinal observed in this writer’s rollout. + writer: Arc>, // Forks hold a shared lease until their child reference is durable; deletion, archive, and // unarchive require exclusive access. Keeping this separate from `writer` lets the source // accept writes during child initialization, including MCP startup that can take 30 seconds. @@ -206,7 +206,7 @@ impl LiveWriterLocks { .clone() } - async fn lock(&self, thread_id: ThreadId) -> OwnedMutexGuard<()> { + async fn lock(&self, thread_id: ThreadId) -> OwnedMutexGuard { self.coordination(thread_id) .await .writer diff --git a/codex-rs/thread-store/src/local/rollout_migration.rs b/codex-rs/thread-store/src/local/rollout_migration.rs index 412c5823fb..968af560a0 100644 --- a/codex-rs/thread-store/src/local/rollout_migration.rs +++ b/codex-rs/thread-store/src/local/rollout_migration.rs @@ -85,7 +85,7 @@ struct MigrationLocks { _maintenance: Option>, _writer: WriterLockGuard, // Let local writers proceed only after releasing the cross-process locks. - _live_writer: OwnedMutexGuard<()>, + _live_writer: OwnedMutexGuard, } enum CanonicalizationAttempt {