[codex] Move persistence policy application into ThreadStore (#27318)

Move the application of the persistence policy into the thread store, so
thread stores can get raw append items rather than canonical append
items. This will enable store-specific projections over the raw input
items.

codex-rs/app-server-protocol/src/protocol/thread_history.rs

@@ -124,6 +124,14 @@ impl ThreadHistoryBuilder {
             .or_else(|| self.turns.last().cloned())
     }
 
+    pub fn turn_snapshot(&self, turn_id: &str) -> Option<Turn> {
+        self.current_turn
+            .as_ref()
+            .filter(|turn| turn.id == turn_id)
+            .map(Turn::from)
+            .or_else(|| self.turns.iter().find(|turn| turn.id == turn_id).cloned())
+    }
+
     /// Returns the index of the active turn snapshot within the finished turn list.
     ///
     /// When a turn is still open, this is the index it will occupy after

codex-rs/core/src/session/session.rs

@@ -526,53 +526,47 @@ impl Session {
             } else {
                 let live_thread = match &initial_history {
                     InitialHistory::New | InitialHistory::Cleared | InitialHistory::Forked(_) => {
-                        LiveThread::create(
-                            Arc::clone(&thread_store),
-                            CreateThreadParams {
-                                thread_id,
-                                extra_config: config.extra_config.clone(),
-                                forked_from_id,
-                                parent_thread_id,
-                                source: session_source,
-                                thread_source: session_configuration.thread_source.clone(),
-                                base_instructions: BaseInstructions {
-                                    text: session_configuration.base_instructions.clone(),
-                                },
-                                dynamic_tools: session_configuration.dynamic_tools.clone(),
-                                multi_agent_version: initial_multi_agent_version,
-                                metadata: ThreadPersistenceMetadata {
-                                    cwd: Some(config.cwd.to_path_buf()),
-                                    model_provider: config.model_provider_id.clone(),
-                                    memory_mode: if config.memories.generate_memories {
-                                        ThreadMemoryMode::Enabled
-                                    } else {
-                                        ThreadMemoryMode::Disabled
-                                    },
+                        let params = CreateThreadParams {
+                            thread_id,
+                            extra_config: config.extra_config.clone(),
+                            forked_from_id,
+                            parent_thread_id,
+                            source: session_source,
+                            thread_source: session_configuration.thread_source.clone(),
+                            base_instructions: BaseInstructions {
+                                text: session_configuration.base_instructions.clone(),
+                            },
+                            dynamic_tools: session_configuration.dynamic_tools.clone(),
+                            multi_agent_version: initial_multi_agent_version,
+                            metadata: ThreadPersistenceMetadata {
+                                cwd: Some(config.cwd.to_path_buf()),
+                                model_provider: config.model_provider_id.clone(),
+                                memory_mode: if config.memories.generate_memories {
+                                    ThreadMemoryMode::Enabled
+                                } else {
+                                    ThreadMemoryMode::Disabled
                                 },
                             },
-                        )
-                        .await?
+                        };
+                        LiveThread::create(Arc::clone(&thread_store), params).await?
                     }
                     InitialHistory::Resumed(resumed_history) => {
-                        LiveThread::resume(
-                            Arc::clone(&thread_store),
-                            ResumeThreadParams {
-                                thread_id: resumed_history.conversation_id,
-                                rollout_path: resumed_history.rollout_path.clone(),
-                                history: Some(resumed_history.history.clone()),
-                                include_archived: true,
-                                metadata: ThreadPersistenceMetadata {
-                                    cwd: Some(config.cwd.to_path_buf()),
-                                    model_provider: config.model_provider_id.clone(),
-                                    memory_mode: if config.memories.generate_memories {
-                                        ThreadMemoryMode::Enabled
-                                    } else {
-                                        ThreadMemoryMode::Disabled
-                                    },
+                        let params = ResumeThreadParams {
+                            thread_id: resumed_history.conversation_id,
+                            rollout_path: resumed_history.rollout_path.clone(),
+                            history: Some(resumed_history.history.clone()),
+                            include_archived: true,
+                            metadata: ThreadPersistenceMetadata {
+                                cwd: Some(config.cwd.to_path_buf()),
+                                model_provider: config.model_provider_id.clone(),
+                                memory_mode: if config.memories.generate_memories {
+                                    ThreadMemoryMode::Enabled
+                                } else {
+                                    ThreadMemoryMode::Disabled
                                 },
                             },
-                        )
-                        .await?
+                        };
+                        LiveThread::resume(Arc::clone(&thread_store), params).await?
                     }
                 };
                 Ok(Some(live_thread))

codex-rs/thread-store/src/in_memory.rs

@@ -14,6 +14,7 @@ use codex_protocol::protocol::RolloutItem;
 use codex_protocol::protocol::SessionMeta;
 use codex_protocol::protocol::SessionMetaLine;
 use codex_protocol::protocol::ThreadMemoryMode;
+use codex_rollout::persisted_rollout_items;
 
 use crate::AppendThreadItemsParams;
 use crate::ArchiveThreadParams;
@@ -215,13 +216,17 @@ impl ThreadStore for InMemoryThreadStore {
     }
 
     async fn append_items(&self, params: AppendThreadItemsParams) -> ThreadStoreResult<()> {
+        let canonical_items = persisted_rollout_items(params.items.as_slice());
+        if canonical_items.is_empty() {
+            return Ok(());
+        }
         let mut state = self.state.lock().await;
         state.calls.append_items += 1;
         state
             .histories
             .entry(params.thread_id)
             .or_default()
-            .extend(params.items);
+            .extend(canonical_items);
         Ok(())
     }
 

codex-rs/thread-store/src/lib.rs

@@ -40,6 +40,7 @@ pub use types::SearchThreadsParams;
 pub use types::SortDirection;
 pub use types::StoredThread;
 pub use types::StoredThreadHistory;
+pub use types::StoredThreadItem;
 pub use types::StoredThreadSearchResult;
 pub use types::StoredTurn;
 pub use types::StoredTurnError;

codex-rs/thread-store/src/live_thread.rs

@@ -116,7 +116,11 @@ impl LiveThread {
             {
                 Ok(history) => params.history = Some(history.items),
                 Err(err) => {
-                    let _ = thread_store.discard_thread(thread_id).await;
+                    if let Err(discard_err) = thread_store.discard_thread(thread_id).await {
+                        warn!(
+                            "failed to discard thread persistence after resume history load failed: {discard_err}"
+                        );
+                    }
                     return Err(err);
                 }
             }
@@ -131,15 +135,18 @@ impl LiveThread {
 
     pub async fn append_items(&self, items: &[RolloutItem]) -> ThreadStoreResult<()> {
         let canonical_items = persisted_rollout_items(items);
-        if canonical_items.is_empty() {
+        if items.is_empty() {
             return Ok(());
         }
         self.thread_store
             .append_items(AppendThreadItemsParams {
                 thread_id: self.thread_id,
-                items: canonical_items.clone(),
+                items: items.to_vec(),
             })
             .await?;
+        if canonical_items.is_empty() {
+            return Ok(());
+        }
         let update = self
             .metadata_sync
             .lock()

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

@@ -5,6 +5,7 @@ use codex_protocol::protocol::ThreadMemoryMode;
 use codex_rollout::RolloutConfig;
 use codex_rollout::RolloutRecorder;
 use codex_rollout::RolloutRecorderParams;
+use codex_rollout::persisted_rollout_items;
 use tracing::warn;
 
 use super::LocalThreadStore;
@@ -77,9 +78,13 @@ pub(super) async fn append_items(
     store: &LocalThreadStore,
     params: AppendThreadItemsParams,
 ) -> ThreadStoreResult<()> {
+    let canonical_items = persisted_rollout_items(params.items.as_slice());
+    if canonical_items.is_empty() {
+        return Ok(());
+    }
     let recorder = store.live_recorder(params.thread_id).await?;
     recorder
-        .record_canonical_items(params.items.as_slice())
+        .record_canonical_items(canonical_items.as_slice())
         .await
         .map_err(thread_store_io_error)?;
     // LiveThread applies metadata immediately after append_items returns. Wait for the local

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

@@ -36,11 +36,10 @@ pub trait ThreadStore: Any + Send + Sync {
     /// Reopens an existing thread for live appends.
     async fn resume_thread(&self, params: ResumeThreadParams) -> ThreadStoreResult<()>;
 
-    /// Appends canonical rollout items to a live thread.
+    /// Appends raw rollout items to a live thread.
     ///
-    /// This is the raw history API. It does not infer metadata from item contents. Callers that
-    /// need metadata updates should call [`ThreadStore::update_thread_metadata`] with explicit
-    /// metadata facts prepared above the store.
+    /// Implementations should apply the shared rollout persistence policy before writing durable
+    /// replay history and before updating any implementation-owned projections.
     async fn append_items(&self, params: AppendThreadItemsParams) -> ThreadStoreResult<()>;
 
     /// Materializes the thread if persistence is lazy, then persists all queued items.

codex-rs/thread-store/src/types.rs

@@ -101,11 +101,14 @@ pub struct ResumeThreadParams {
 }
 
 /// Parameters for appending rollout items to a live thread.
-#[derive(Clone, Debug, Serialize, Deserialize)]
+#[derive(Clone, Debug)]
 pub struct AppendThreadItemsParams {
     /// Thread id to append to.
     pub thread_id: ThreadId,
-    /// Items to append in order.
+    /// Raw rollout items to append in order.
+    ///
+    /// Store implementations are responsible for applying the shared rollout persistence policy
+    /// before writing durable replay history or any implementation-owned projections.
     pub items: Vec<RolloutItem>,
 }
 
@@ -297,6 +300,11 @@ pub struct StoredTurn {
     pub turn_id: String,
     /// Persisted rollout items associated with this turn, according to `items_view`.
     pub items: Vec<RolloutItem>,
+    /// Opaque serialized turn metadata supplied by a projected durable store.
+    pub metadata_json: Option<Vec<u8>>,
+    /// Semantic turn creation timestamp in milliseconds, when supplied by a projected durable
+    /// store.
+    pub turn_created_at_ms: Option<i64>,
     /// Amount of item detail included in `items`.
     pub items_view: StoredTurnItemsView,
     /// Store-owned status for API layer projection.
@@ -339,11 +347,21 @@ pub struct ListItemsParams {
     pub sort_direction: SortDirection,
 }
 
-/// A page of persisted rollout items within a turn.
+/// A projected app-server `ThreadItem` snapshot within a turn.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct StoredThreadItem {
+    pub turn_id: Option<String>,
+    pub item_key: String,
+    pub item_ordinal: u64,
+    pub item_created_at_ms: i64,
+    pub materialized_thread_item_json: Vec<u8>,
+}
+
+/// A page of persisted items within a turn.
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct ItemPage {
     /// Items returned for this page.
-    pub items: Vec<RolloutItem>,
+    pub items: Vec<StoredThreadItem>,
     /// Opaque cursor to continue listing.
     pub next_cursor: Option<String>,
     /// Opaque cursor for fetching in the opposite direction.