Preserve thread-local model ID state across replay and compaction (#60942)

## What changed

- Initialize item and context-window ID generators per thread, reuse the stored window on resume, and give history reconstruction a separate item generator.
- Recover item ordinals from checkpoint replacement history, guardian history, and retained context without lowering the cursor established by earlier records.
- Stamp compaction resume metadata with the next item ordinal in append order, tracking the cursor in local and in-memory thread stores.

## Testing

Extend the history ordinal test to verify that a checkpoint records the cursor at its position in a batch, excluding later allocations.

GitOrigin-RevId: 7121b747a039dc5477235078702137d1ff74e789

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<ThreadManagerState>,
     /// 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<AgentExecutionLimiter>,
     /// Session-scoped state shared by the root thread and every cloned sub-agent control handle.
     pub(super) rollout_budget: Arc<RolloutBudget>,
@@ -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(),

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),
                                 )

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)
         };

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),
+    )
+}

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(),
         )

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<dyn AgentControl>, _) = 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 {

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());

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(_))
     })

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![

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;

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<u64> {
+/// 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<u64> {
+    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<u64> {
     id.allocation()
         .map(|allocation| allocation.model_id.ordinal())
 }

codex-rs/history/src/retained_context.rs

@@ -555,6 +555,19 @@ impl RetainedContext {
         }
     }
 
+    pub(super) fn max_source_ordinal(&self) -> Option<u64> {
+        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.

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<RolloutItem> = serde_json::from_value(json!([
+    let mut records: Vec<RolloutItem> = 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(())
 }
 

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<dyn Fn(&str) -> ResponseItemId + Send + Sync>;
 
 #[cfg(test)]

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.

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(())
     }
 

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<Arc<Vec<RolloutItem>>> {
-    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),
-    }
-}

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<Mutex<()>>,
+    // Serializes writes and holds the next item ordinal observed in this writer’s rollout.
+    writer: Arc<Mutex<u64>>,
     // 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<u64> {
         self.coordination(thread_id)
             .await
             .writer

codex-rs/thread-store/src/local/rollout_migration.rs

@@ -85,7 +85,7 @@ struct MigrationLocks {
     _maintenance: Option<Arc<RolloutMaintenanceGuard>>,
     _writer: WriterLockGuard,
     // Let local writers proceed only after releasing the cross-process locks.
-    _live_writer: OwnedMutexGuard<()>,
+    _live_writer: OwnedMutexGuard<u64>,
 }
 
 enum CanonicalizationAttempt {