Preserve thread metadata before rollout files are materialized (#55380)

## Why

A live thread can have updated SQLite metadata before its rollout file exists.
Reading metadata from the missing file loses creation details such as fork and
parent relationships and triggers unnecessary filesystem discovery and retries.

## What changed

- Read deferred `SessionMeta` through the rollout writer, ordered with persistence.
  Return no in-memory metadata once the rollout is materialized or resumed.
- Combine that creation metadata with SQLite metadata for reads that exclude
  history, and use it to determine the canonical history mode during updates.
- Keep untouched threads without a stored record unavailable for fork and resume.

## Testing

Add coverage for metadata reads and updates before persistence in both legacy and
paginated history modes, including equality with metadata read after persistence.
Verify the recorder exposes creation metadata only while deferred, and expand
fork/resume rejection tests across history modes and resume turn inclusion.

GitOrigin-RevId: 8d293cc8f87156409f195130437d93858a4099b9

codex-rs/app-server/tests/suite/v2/thread_fork.rs

@@ -1475,8 +1475,10 @@ async fn thread_fork_tracks_thread_initialized_analytics() -> Result<()> {
     Ok(())
 }
 
+#[test_case(ThreadHistoryMode::Legacy; "legacy")]
+#[test_case(ThreadHistoryMode::Paginated; "paginated")]
 #[tokio::test]
-async fn thread_fork_rejects_unmaterialized_thread() -> Result<()> {
+async fn thread_fork_rejects_unmaterialized_thread(history_mode: ThreadHistoryMode) -> Result<()> {
     let server = create_mock_responses_server_repeating_assistant("Done").await;
     let codex_home = TempDir::new()?;
     MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
@@ -1489,6 +1491,7 @@ async fn thread_fork_rejects_unmaterialized_thread() -> Result<()> {
     let start_id = mcp
         .send_thread_start_request_with_auto_env(ThreadStartParams {
             model: Some("mock-model".to_string()),
+            history_mode: Some(history_mode),
             ..Default::default()
         })
         .await?;

codex-rs/app-server/tests/suite/v2/thread_resume.rs

@@ -447,8 +447,15 @@ async fn wait_for_responses_request_count(
     Ok(())
 }
 
+#[test_case(ThreadHistoryMode::Legacy, false; "legacy with turns")]
+#[test_case(ThreadHistoryMode::Legacy, true; "legacy without turns")]
+#[test_case(ThreadHistoryMode::Paginated, false; "paginated with turns")]
+#[test_case(ThreadHistoryMode::Paginated, true; "paginated without turns")]
 #[tokio::test]
-async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> {
+async fn thread_resume_rejects_unmaterialized_thread(
+    history_mode: ThreadHistoryMode,
+    exclude_turns: bool,
+) -> 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())?;
@@ -462,6 +469,7 @@ async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> {
     let start_id = mcp
         .send_thread_start_request_with_auto_env(ThreadStartParams {
             model: Some("gpt-5.4".to_string()),
+            history_mode: Some(history_mode),
             ..Default::default()
         })
         .await?;
@@ -472,6 +480,7 @@ async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> {
     let resume_id = mcp
         .send_thread_resume_request(ThreadResumeParams {
             thread_id: thread.id.clone(),
+            exclude_turns,
             ..Default::default()
         })
         .await?;

codex-rs/rollout/src/recorder.rs

@@ -127,6 +127,9 @@ pub enum RolloutRecorderParams {
 
 enum RolloutCmd {
     AddItems(Vec<RolloutItem>),
+    ReadUnpersistedSessionMeta {
+        ack: oneshot::Sender<Option<SessionMeta>>,
+    },
     Persist {
         ack: oneshot::Sender<std::io::Result<()>>,
     },
@@ -1045,6 +1048,19 @@ impl RolloutRecorder {
         self.rollout_path.as_path()
     }
 
+    /// Reads the creation metadata while this writer still defers its rollout file.
+    ///
+    /// The writer orders this read with persistence. Resumed or already materialized rollouts
+    /// return `None`, so callers must keep using their durable metadata.
+    pub async fn unpersisted_session_meta(&self) -> std::io::Result<Option<SessionMeta>> {
+        let (ack, result) = oneshot::channel();
+        self.tx
+            .send(RolloutCmd::ReadUnpersistedSessionMeta { ack })
+            .await
+            .map_err(IoError::other)?;
+        result.await.map_err(IoError::other)
+    }
+
     pub async fn record_canonical_items(&self, items: &[RolloutItem]) -> std::io::Result<()> {
         if items.is_empty() {
             return Ok(());
@@ -1966,6 +1982,14 @@ async fn rollout_writer(
     // Process rollout commands
     while let Some(cmd) = rx.recv().await {
         match cmd {
+            RolloutCmd::ReadUnpersistedSessionMeta { ack } => {
+                let metadata = if state.is_deferred() {
+                    state.meta.clone()
+                } else {
+                    None
+                };
+                let _ = ack.send(metadata);
+            }
             RolloutCmd::AddItems(items) => {
                 state.add_items(items);
                 state.flush_if_materialized().await;

codex-rs/rollout/src/recorder_tests.rs

@@ -822,10 +822,20 @@ async fn referenced_paginated_rollout_starts_at_history_cutoff_and_resumes() ->
     )
     .await?;
     let rollout_path = recorder.rollout_path().to_path_buf();
+    let initial_meta = recorder
+        .unpersisted_session_meta()
+        .await?
+        .expect("new rollout retains its creation metadata");
+    assert!(!rollout_path.exists());
     recorder.persist().await?;
+    assert!(recorder.unpersisted_session_meta().await?.is_none());
     recorder.shutdown().await?;
 
     let meta = crate::read_session_meta_line(&rollout_path).await?.meta;
+    assert_eq!(
+        serde_json::to_value(&initial_meta)?,
+        serde_json::to_value(&meta)?
+    );
     assert_eq!(
         meta.forked_from_ordinal_exclusive,
         Some(history_base.end_ordinal_exclusive)
@@ -833,6 +843,7 @@ async fn referenced_paginated_rollout_starts_at_history_cutoff_and_resumes() ->
 
     let resumed =
         RolloutRecorder::new(&config, RolloutRecorderParams::resume(rollout_path.clone())).await?;
+    assert!(resumed.unpersisted_session_meta().await?.is_none());
     resumed
         .record_canonical_items(&[agent_message_item("first child record")])
         .await?;

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

@@ -2,6 +2,7 @@ use std::path::PathBuf;
 use std::sync::Arc;
 
 use codex_protocol::ThreadId;
+use codex_protocol::protocol::SessionMeta;
 use codex_protocol::protocol::ThreadHistoryMode;
 use codex_protocol::protocol::ThreadMemoryMode;
 use codex_rollout::RolloutConfig;
@@ -324,6 +325,24 @@ pub(super) async fn live_writer_parts(
     Ok((entry.recorder.clone(), entry.rollout_id, entry.history_mode))
 }
 
+/// Uses the writer's creation record only until its rollout has been materialized.
+pub(super) async fn unpersisted_session_meta(
+    store: &LocalThreadStore,
+    thread_id: ThreadId,
+) -> Option<(PathBuf, SessionMeta)> {
+    let (recorder, _, _) = live_writer_parts(store, thread_id).await.ok()?;
+    // Materialized reads should not queue behind unrelated transcript writes.
+    if codex_rollout::existing_rollout_path(recorder.rollout_path())
+        .await
+        .is_some()
+    {
+        return None;
+    }
+    // A writer may finish shutdown after lookup; durable reads remain the fallback.
+    let meta = recorder.unpersisted_session_meta().await.ok()??;
+    Some((recorder.rollout_path().to_path_buf(), meta))
+}
+
 async fn write_and_project(
     store: &LocalThreadStore,
     thread_id: ThreadId,

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

@@ -233,6 +233,97 @@ async fn pending_rollout_compatible_metadata_is_written_with_observed_update() {
     assert_pending_metadata_consumed(&store, thread_id).await;
 }
 
+#[tokio::test]
+async fn deferred_reads_preserve_creation_metadata_and_updates() {
+    for history_mode in [ThreadHistoryMode::Legacy, ThreadHistoryMode::Paginated] {
+        let (_home, store, runtime) = store_with_runtime().await;
+        let thread_id = ThreadId::new();
+        let mut params = create_thread_params(thread_id);
+        params.history_mode = history_mode;
+        params.forked_from_id = Some(ThreadId::new());
+        params.parent_thread_id = Some(ThreadId::new());
+        let live_thread = LiveThread::create(store.clone(), params.clone())
+            .await
+            .expect("create live thread");
+        let rollout_path = store
+            .live_rollout_path(thread_id)
+            .await
+            .expect("live rollout path");
+        let initial = live_thread
+            .read_thread(
+                /*include_archived*/ false, /*include_history*/ false,
+            )
+            .await
+            .expect_err("untouched threads still have no stored record");
+        assert!(matches!(
+            initial,
+            ThreadStoreError::InvalidRequest { message }
+                if message == format!("no rollout found for thread id {thread_id}")
+        ));
+        assert!(!rollout_path.exists());
+        assert_eq!(
+            runtime.get_thread(thread_id).await.expect("read index"),
+            None
+        );
+
+        let updated = live_thread
+            .update_metadata(
+                ThreadMetadataPatch {
+                    daybreak_enabled: Some(true),
+                    ..Default::default()
+                },
+                /*include_archived*/ false,
+            )
+            .await
+            .expect("update metadata without materializing history");
+        let read = live_thread
+            .read_thread(
+                /*include_archived*/ false, /*include_history*/ false,
+            )
+            .await
+            .expect("read updated metadata");
+        assert_eq!(
+            serde_json::to_value(read).expect("serialize metadata read"),
+            serde_json::to_value(&updated).expect("serialize update result")
+        );
+        assert_eq!(
+            (
+                updated.forked_from_id,
+                updated.parent_thread_id,
+                updated.originator.clone(),
+                updated.history_mode,
+                updated.daybreak_enabled,
+            ),
+            (
+                params.forked_from_id,
+                params.parent_thread_id,
+                Some(params.originator),
+                history_mode,
+                Some(true),
+            )
+        );
+        assert!(!rollout_path.exists());
+
+        live_thread
+            .persist(PersistContext::Standard)
+            .await
+            .expect("persist history");
+        live_thread.shutdown().await.expect("shutdown writer");
+        let durable = store
+            .read_thread(ReadThreadParams {
+                thread_id,
+                include_archived: false,
+                include_history: false,
+            })
+            .await
+            .expect("read durable metadata");
+        assert_eq!(
+            serde_json::to_value(durable).expect("serialize durable metadata"),
+            serde_json::to_value(updated).expect("serialize initial metadata")
+        );
+    }
+}
+
 #[tokio::test]
 async fn idle_shutdown_clears_pending_thread_metadata_without_materializing() {
     let (_home, store, runtime) = store_with_runtime().await;
@@ -245,6 +336,24 @@ async fn idle_shutdown_clears_pending_thread_metadata_without_materializing() {
     let live_thread = LiveThread::create(store.clone(), create_thread_params(thread_id))
         .await
         .expect("create live thread");
+    let read_error = live_thread
+        .read_thread(
+            /*include_archived*/ false, /*include_history*/ false,
+        )
+        .await
+        .expect_err("untouched threads still have no stored record");
+    assert!(matches!(
+        read_error,
+        ThreadStoreError::InvalidRequest { message }
+            if message == format!("no rollout found for thread id {thread_id}")
+    ));
+    assert!(
+        !store
+            .live_rollout_path(thread_id)
+            .await
+            .expect("live rollout path")
+            .exists()
+    );
     live_thread.shutdown().await.expect("shutdown idle thread");
 
     assert_eq!(

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

@@ -2,6 +2,7 @@ use chrono::DateTime;
 use chrono::Utc;
 use codex_protocol::models::PermissionProfile;
 use codex_protocol::protocol::AskForApproval;
+use codex_protocol::protocol::SessionMeta;
 use codex_protocol::protocol::SessionMetaLine;
 use codex_protocol::protocol::SessionSource;
 use codex_protocol::protocol::ThreadHistoryMode;
@@ -34,6 +35,24 @@ pub(super) async fn read_thread(
 ) -> ThreadStoreResult<StoredThread> {
     let thread_id = params.thread_id;
     let sqlite_metadata = read_sqlite_metadata(store, thread_id).await;
+    if !params.include_history
+        && let Some((path, meta)) =
+            super::live_writer::unpersisted_session_meta(store, thread_id).await
+        && sqlite_metadata
+            .as_ref()
+            .is_none_or(|metadata| metadata.rollout_path == path && metadata.archived_at.is_none())
+    {
+        // A new writer already owns its metadata. Missing JSONL is intentional, so neither
+        // filesystem discovery nor compression-race retries can find anything more authoritative.
+        return match sqlite_metadata {
+            Some(metadata) => Ok(stored_thread_from_metadata(store, metadata, Some(meta)).await),
+            // Preserve the saved-thread contract used by fork/resume. A live but untouched
+            // thread has no stored record yet; its absence is known without searching history.
+            None => Err(ThreadStoreError::InvalidRequest {
+                message: format!("no rollout found for thread id {thread_id}"),
+            }),
+        };
+    }
     let persisted_model_settings = sqlite_metadata
         .as_ref()
         .map(|metadata| (metadata.model.clone(), metadata.reasoning_effort.clone()));
@@ -376,6 +395,14 @@ pub(super) async fn stored_thread_from_sqlite_metadata(
             });
         }
     };
+    Ok(stored_thread_from_metadata(store, metadata, session_meta).await)
+}
+
+async fn stored_thread_from_metadata(
+    store: &LocalThreadStore,
+    metadata: ThreadMetadata,
+    session_meta: Option<SessionMeta>,
+) -> StoredThread {
     let forked_from_id = session_meta.as_ref().and_then(|meta| meta.forked_from_id);
     let parent_thread_id = session_meta.as_ref().and_then(|meta| meta.parent_thread_id);
     let history_mode = session_meta
@@ -392,7 +419,7 @@ pub(super) async fn stored_thread_from_sqlite_metadata(
     thread.forked_from_id = forked_from_id;
     thread.history_mode = history_mode;
     thread.name = name;
-    Ok(thread)
+    thread
 }
 
 pub(super) fn stored_thread_from_state_metadata(

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

@@ -695,6 +695,11 @@ async fn canonical_history_mode(
     thread_id: ThreadId,
     rollout_path: &Path,
 ) -> ThreadStoreResult<ThreadHistoryMode> {
+    if let Some((path, meta)) = live_writer::unpersisted_session_meta(store, thread_id).await
+        && path == rollout_path
+    {
+        return Ok(meta.history_mode);
+    }
     let session_meta = match read_session_meta_line(rollout_path).await {
         Ok(session_meta) => session_meta,
         Err(err) => {