Restore deferred rollout materialization semantics

Reinstate main's deferred rollout persistence behavior while keeping the shared rollout source groundwork. Materialize the rollout on first persisted user-visible activity, restore app-server's unmaterialized-thread handling, and keep in-memory rollout state aligned with persisted session metadata.

Co-authored-by: Codex <noreply@openai.com>

codex-rs/app-server/src/bespoke_event_handling.rs

@@ -1608,7 +1608,7 @@ pub(crate) async fn apply_bespoke_event_handling(
                 let Some(rollout_path) = conversation.rollout_path() else {
                     let error = JSONRPCErrorError {
                         code: INVALID_REQUEST_ERROR_CODE,
-                        message: "thread has no rollout".to_string(),
+                        message: "thread has no persisted rollout".to_string(),
                         data: None,
                     };
                     outgoing.send_error(request_id, error).await;

codex-rs/app-server/src/codex_message_processor.rs

@@ -3100,6 +3100,16 @@ impl CodexMessageProcessor {
                 Ok(items) => {
                     thread.turns = build_turns_from_rollout_items(&items);
                 }
+                Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
+                    self.send_invalid_request_error(
+                        request_id,
+                        format!(
+                            "thread {thread_uuid} is not materialized yet; includeTurns is unavailable before first user message"
+                        ),
+                    )
+                    .await;
+                    return;
+                }
                 Err(err) => {
                     self.send_internal_error(
                         request_id,
@@ -3387,7 +3397,34 @@ impl CodexMessageProcessor {
             }
 
             let rollout_path = if let Some(path) = existing_thread.rollout_path() {
-                path
+                if path.exists() {
+                    path
+                } else {
+                    match find_thread_path_by_id_str(
+                        &self.config.codex_home,
+                        &existing_thread_id.to_string(),
+                    )
+                    .await
+                    {
+                        Ok(Some(path)) => path,
+                        Ok(None) => {
+                            self.send_invalid_request_error(
+                                request_id,
+                                format!("no rollout found for thread id {existing_thread_id}"),
+                            )
+                            .await;
+                            return true;
+                        }
+                        Err(err) => {
+                            self.send_invalid_request_error(
+                                request_id,
+                                format!("failed to locate thread id {existing_thread_id}: {err}"),
+                            )
+                            .await;
+                            return true;
+                        }
+                    }
+                }
             } else {
                 match find_thread_path_by_id_str(
                     &self.config.codex_home,

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

@@ -2,6 +2,7 @@ use anyhow::Result;
 use app_test_support::McpProcess;
 use app_test_support::create_mock_responses_server_repeating_assistant;
 use app_test_support::to_response;
+use codex_app_server_protocol::JSONRPCError;
 use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::ThreadArchiveParams;
@@ -27,7 +28,7 @@ use tokio::time::timeout;
 const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
 
 #[tokio::test]
-async fn thread_archive_works_before_first_user_message() -> Result<()> {
+async fn thread_archive_requires_materialized_rollout() -> Result<()> {
     let server = create_mock_responses_server_repeating_assistant("Done").await;
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
@@ -51,14 +52,72 @@ async fn thread_archive_works_before_first_user_message() -> Result<()> {
     assert!(!thread.id.is_empty());
 
     let rollout_path = thread.path.clone().expect("thread path");
-    assert!(rollout_path.exists(), "fresh thread rollout should exist");
+    assert!(
+        !rollout_path.exists(),
+        "fresh thread rollout should not exist yet at {}",
+        rollout_path.display()
+    );
     assert!(
         find_thread_path_by_id_str(codex_home.path(), &thread.id)
             .await?
-            .is_some(),
-        "thread id should be discoverable immediately after startup"
+            .is_none(),
+        "thread id should not be discoverable before rollout materialization"
     );
 
+    // Archive should fail before the rollout is materialized.
+    let archive_id = mcp
+        .send_thread_archive_request(ThreadArchiveParams {
+            thread_id: thread.id.clone(),
+        })
+        .await?;
+    let archive_err: JSONRPCError = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_error_message(RequestId::Integer(archive_id)),
+    )
+    .await??;
+    assert!(
+        archive_err
+            .error
+            .message
+            .contains("no rollout found for thread id"),
+        "unexpected archive error: {}",
+        archive_err.error.message
+    );
+
+    // Materialize rollout via a real user turn and confirm archive succeeds.
+    let turn_start_id = mcp
+        .send_turn_start_request(TurnStartParams {
+            thread_id: thread.id.clone(),
+            input: vec![UserInput::Text {
+                text: "materialize".to_string(),
+                text_elements: Vec::new(),
+            }],
+            ..Default::default()
+        })
+        .await?;
+    let turn_start_response: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(turn_start_id)),
+    )
+    .await??;
+    let _: TurnStartResponse = to_response::<TurnStartResponse>(turn_start_response)?;
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("turn/completed"),
+    )
+    .await??;
+
+    assert!(
+        rollout_path.exists(),
+        "expected rollout path {} to exist after first user message",
+        rollout_path.display()
+    );
+
+    let discovered_path = find_thread_path_by_id_str(codex_home.path(), &thread.id)
+        .await?
+        .expect("expected rollout path for thread id to exist after materialization");
+    assert_paths_match_on_disk(&discovered_path, &rollout_path)?;
+
     let archive_id = mcp
         .send_thread_archive_request(ThreadArchiveParams {
             thread_id: thread.id.clone(),
@@ -70,7 +129,6 @@ async fn thread_archive_works_before_first_user_message() -> Result<()> {
     )
     .await??;
     let _: ThreadArchiveResponse = to_response::<ThreadArchiveResponse>(archive_resp)?;
-
     let archive_notification = timeout(
         DEFAULT_READ_TIMEOUT,
         mcp.read_stream_until_notification_message("thread/archived"),
@@ -258,3 +316,10 @@ stream_max_retries = 0
 "#
     )
 }
+
+fn assert_paths_match_on_disk(actual: &Path, expected: &Path) -> std::io::Result<()> {
+    let actual = actual.canonicalize()?;
+    let expected = expected.canonicalize()?;
+    assert_eq!(actual, expected);
+    Ok(())
+}

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

@@ -3,6 +3,7 @@ use app_test_support::McpProcess;
 use app_test_support::create_fake_rollout;
 use app_test_support::create_mock_responses_server_repeating_assistant;
 use app_test_support::to_response;
+use codex_app_server_protocol::JSONRPCError;
 use codex_app_server_protocol::JSONRPCMessage;
 use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::RequestId;
@@ -163,7 +164,7 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
 }
 
 #[tokio::test]
-async fn thread_fork_creates_empty_loaded_thread() -> Result<()> {
+async fn thread_fork_rejects_unmaterialized_thread() -> Result<()> {
     let server = create_mock_responses_server_repeating_assistant("Done").await;
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
@@ -184,24 +185,25 @@ async fn thread_fork_creates_empty_loaded_thread() -> Result<()> {
     .await??;
     let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
 
-    let thread_id = thread.id.clone();
     let fork_id = mcp
         .send_thread_fork_request(ThreadForkParams {
-            thread_id,
+            thread_id: thread.id,
             ..Default::default()
         })
         .await?;
-    let fork_resp: JSONRPCResponse = timeout(
+    let fork_err: JSONRPCError = timeout(
         DEFAULT_READ_TIMEOUT,
-        mcp.read_stream_until_response_message(RequestId::Integer(fork_id)),
+        mcp.read_stream_until_error_message(RequestId::Integer(fork_id)),
     )
     .await??;
-    let ThreadForkResponse { thread: forked, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
-
-    assert_ne!(forked.id, thread.id);
-    assert!(forked.preview.is_empty());
-    assert_eq!(forked.turns.len(), 0);
-    assert_eq!(forked.status, ThreadStatus::Idle);
+    assert!(
+        fork_err
+            .error
+            .message
+            .contains("no rollout found for thread id"),
+        "unexpected fork error: {}",
+        fork_err.error.message
+    );
 
     Ok(())
 }

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

@@ -3,6 +3,7 @@ use app_test_support::McpProcess;
 use app_test_support::create_fake_rollout_with_text_elements;
 use app_test_support::create_mock_responses_server_repeating_assistant;
 use app_test_support::to_response;
+use codex_app_server_protocol::JSONRPCError;
 use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::SessionSource;
@@ -152,7 +153,7 @@ async fn thread_read_can_include_turns() -> Result<()> {
 }
 
 #[tokio::test]
-async fn thread_read_loaded_thread_returns_startup_rollout_path() -> Result<()> {
+async fn thread_read_loaded_thread_returns_precomputed_path_before_materialization() -> Result<()> {
     let server = create_mock_responses_server_repeating_assistant("Done").await;
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
@@ -174,8 +175,8 @@ async fn thread_read_loaded_thread_returns_startup_rollout_path() -> Result<()>
     let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
     let thread_path = thread.path.clone().expect("thread path");
     assert!(
-        thread_path.exists(),
-        "fresh thread rollout should be readable at startup"
+        !thread_path.exists(),
+        "fresh thread rollout should not be materialized yet"
     );
 
     let read_id = mcp
@@ -356,7 +357,7 @@ async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> {
 }
 
 #[tokio::test]
-async fn thread_read_include_turns_returns_empty_loaded_thread() -> Result<()> {
+async fn thread_read_include_turns_rejects_unmaterialized_loaded_thread() -> Result<()> {
     let server = create_mock_responses_server_repeating_assistant("Done").await;
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
@@ -377,7 +378,10 @@ async fn thread_read_include_turns_returns_empty_loaded_thread() -> Result<()> {
     .await??;
     let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
     let thread_path = thread.path.clone().expect("thread path");
-    assert!(thread_path.exists(), "fresh thread rollout should exist");
+    assert!(
+        !thread_path.exists(),
+        "fresh thread rollout should not be materialized yet"
+    );
 
     let read_id = mcp
         .send_thread_read_request(ThreadReadParams {
@@ -385,18 +389,20 @@ async fn thread_read_include_turns_returns_empty_loaded_thread() -> Result<()> {
             include_turns: true,
         })
         .await?;
-    let read_resp: JSONRPCResponse = timeout(
+    let read_err: JSONRPCError = timeout(
         DEFAULT_READ_TIMEOUT,
-        mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
+        mcp.read_stream_until_error_message(RequestId::Integer(read_id)),
     )
     .await??;
-    let ThreadReadResponse { thread: read } = to_response::<ThreadReadResponse>(read_resp)?;
 
-    assert_eq!(read.id, thread.id);
-    assert_eq!(read.path, Some(thread_path));
-    assert!(read.preview.is_empty());
-    assert_eq!(read.turns.len(), 0);
-    assert_eq!(read.status, ThreadStatus::Idle);
+    assert!(
+        read_err
+            .error
+            .message
+            .contains("includeTurns is unavailable before first user message"),
+        "unexpected error: {}",
+        read_err.error.message
+    );
 
     Ok(())
 }

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

@@ -95,7 +95,7 @@ async fn wait_for_responses_request_count(
 }
 
 #[tokio::test]
-async fn thread_resume_returns_empty_loaded_thread() -> Result<()> {
+async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> {
     let server = create_mock_responses_server_repeating_assistant("Done").await;
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
@@ -117,25 +117,26 @@ async fn thread_resume_returns_empty_loaded_thread() -> Result<()> {
     .await??;
     let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
 
+    // Resume should fail before the first user message materializes rollout storage.
     let resume_id = mcp
         .send_thread_resume_request(ThreadResumeParams {
             thread_id: thread.id.clone(),
             ..Default::default()
         })
         .await?;
-    let resume_resp: JSONRPCResponse = timeout(
+    let resume_err: JSONRPCError = timeout(
         DEFAULT_READ_TIMEOUT,
-        mcp.read_stream_until_response_message(RequestId::Integer(resume_id)),
+        mcp.read_stream_until_error_message(RequestId::Integer(resume_id)),
     )
     .await??;
-    let ThreadResumeResponse {
-        thread: resumed, ..
-    } = to_response::<ThreadResumeResponse>(resume_resp)?;
-
-    assert_eq!(resumed.id, thread.id);
-    assert!(resumed.preview.is_empty());
-    assert_eq!(resumed.turns.len(), 0);
-    assert_eq!(resumed.status, ThreadStatus::Idle);
+    assert!(
+        resume_err
+            .error
+            .message
+            .contains("no rollout found for thread id"),
+        "unexpected resume error: {}",
+        resume_err.error.message
+    );
 
     Ok(())
 }

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

@@ -73,8 +73,8 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
     let thread_path = thread.path.clone().expect("thread path should be present");
     assert!(thread_path.is_absolute(), "thread path should be absolute");
     assert!(
-        thread_path.exists(),
-        "fresh thread rollout should be readable at startup"
+        !thread_path.exists(),
+        "fresh thread rollout should not be materialized until first user message"
     );
 
     // Wire contract: thread title field is `name`, serialized as null when unset.

codex-rs/core/src/agent/control.rs

@@ -144,7 +144,12 @@ impl AgentControl {
                     let parent_thread = state.get_thread(parent_thread_id).await.ok();
                     if let Some(parent_thread) = parent_thread.as_ref() {
                         // `record_conversation_items` only queues rollout writes asynchronously.
-                        // Flush the live parent before snapshotting JSONL for a fork.
+                        // Flush/materialize the live parent before snapshotting JSONL for a fork.
+                        parent_thread
+                            .codex
+                            .session
+                            .ensure_rollout_materialized()
+                            .await;
                         parent_thread.codex.session.flush_rollout().await;
                     }
                     let rollout_path = parent_thread
@@ -883,6 +888,11 @@ mod tests {
             .session
             .record_conversation_items(turn_context.as_ref(), &[parent_spawn_call])
             .await;
+        parent_thread
+            .codex
+            .session
+            .ensure_rollout_materialized()
+            .await;
         parent_thread.codex.session.flush_rollout().await;
 
         let child_thread_id = harness
@@ -960,6 +970,11 @@ mod tests {
             .session
             .record_conversation_items(turn_context.as_ref(), &[parent_spawn_call])
             .await;
+        parent_thread
+            .codex
+            .session
+            .ensure_rollout_materialized()
+            .await;
         parent_thread.codex.session.flush_rollout().await;
 
         let child_thread_id = harness

codex-rs/core/src/codex.rs

@@ -1816,6 +1816,18 @@ impl Session {
         }
     }
 
+    pub(crate) async fn ensure_rollout_materialized(&self) {
+        let recorder = {
+            let guard = self.services.rollout.lock().await;
+            guard.clone()
+        };
+        if let Some(rec) = recorder
+            && let Err(e) = rec.persist().await
+        {
+            warn!("failed to materialize rollout recorder: {e}");
+        }
+    }
+
     fn next_internal_sub_id(&self) -> String {
         let id = self
             .next_internal_sub_id
@@ -2052,6 +2064,9 @@ impl Session {
                     state.set_reference_context_item(Some(turn_context.to_turn_context_item()));
                 }
 
+                // Forked threads should remain file-backed immediately after startup.
+                self.ensure_rollout_materialized().await;
+
                 // Flush after seeding history and any persisted rollout copy.
                 if !is_subagent {
                     self.flush_rollout().await;
@@ -3673,6 +3688,7 @@ impl Session {
         let turn_item = TurnItem::UserMessage(UserMessageItem::new(input));
         self.emit_turn_item_started(turn_context, &turn_item).await;
         self.emit_turn_item_completed(turn_context, turn_item).await;
+        self.ensure_rollout_materialized().await;
     }
 
     pub(crate) async fn notify_background_event(

codex-rs/core/src/codex_tests.rs

@@ -1824,6 +1824,7 @@ async fn attach_rollout_recorder(session: &Arc<Session>) -> PathBuf {
         let mut rollout = session.services.rollout.lock().await;
         *rollout = Some(recorder);
     }
+    session.ensure_rollout_materialized().await;
     session.flush_rollout().await;
     rollout_path
 }
@@ -3410,6 +3411,7 @@ async fn record_context_updates_and_set_reference_context_item_persists_baseline
         serde_json::to_value(Some(turn_context.to_turn_context_item()))
             .expect("serialize expected context item")
     );
+    session.ensure_rollout_materialized().await;
     session.flush_rollout().await;
 
     let InitialHistory::Resumed(resumed) = RolloutStore::get_rollout_history(&rollout_path)
@@ -3511,6 +3513,7 @@ async fn record_context_updates_and_set_reference_context_item_persists_full_rei
     session
         .record_context_updates_and_set_reference_context_item(&turn_context)
         .await;
+    session.ensure_rollout_materialized().await;
     session.flush_rollout().await;
 
     let InitialHistory::Resumed(resumed) = RolloutStore::get_rollout_history(&rollout_path)

codex-rs/core/src/codex_thread.rs

@@ -133,6 +133,8 @@ impl CodexThread {
     }
 
     /// Return the rollout path for this thread, if any.
+    ///
+    /// Fresh persistent threads may reserve a rollout path before the file is materialized.
     pub fn rollout_path(&self) -> Option<PathBuf> {
         self.rollout_path.clone()
     }

codex-rs/core/src/rollout/recorder.rs

@@ -519,15 +519,23 @@ impl RolloutStore {
 
     /// Attempt to create a new [`RolloutStore`].
     ///
-    /// For both new and resumed sessions, this opens the rollout file immediately so the path is
-    /// readable as soon as startup returns.
+    /// For resumed sessions, this immediately opens the existing rollout file. Fresh sessions keep
+    /// their rollout buffered in memory until an explicit `persist()` call materializes the file.
     pub async fn new(
         config: &Config,
         params: RolloutStoreParams,
         state_db_ctx: Option<StateDbHandle>,
         state_builder: Option<ThreadMetadataBuilder>,
     ) -> std::io::Result<Self> {
-        let (file, rollout_path, source, meta, event_persistence_mode) = match params {
+        let (
+            file,
+            deferred_log_file_info,
+            rollout_path,
+            source,
+            meta,
+            git_info_handle,
+            event_persistence_mode,
+        ) = match params {
             RolloutStoreParams::Create {
                 conversation_id,
                 forked_from_id,
@@ -573,14 +581,19 @@ impl RolloutStore {
                     meta: session_meta,
                     git: None,
                 };
+                let cwd = config.cwd.clone();
 
                 (
-                    tokio::fs::File::from_std(open_log_file(path.as_path())?),
+                    None,
+                    Some(log_file_info),
                     path,
                     InMemoryRolloutSource::new(vec![RolloutItem::SessionMeta(
                         session_meta_line.clone(),
                     )]),
                     Some(session_meta_line),
+                    Some(tokio::task::spawn(async move {
+                        collect_git_info(cwd.as_path()).await
+                    })),
                     event_persistence_mode,
                 )
             }
@@ -594,13 +607,17 @@ impl RolloutStore {
                     None => Self::load_source(path.as_path()).await?.source,
                 };
                 (
-                    tokio::fs::OpenOptions::new()
-                        .append(true)
-                        .open(&path)
-                        .await?,
+                    Some(
+                        tokio::fs::OpenOptions::new()
+                            .append(true)
+                            .open(&path)
+                            .await?,
+                    ),
+                    None,
                     path,
                     source,
                     None,
+                    None,
                     event_persistence_mode,
                 )
             }
@@ -612,30 +629,20 @@ impl RolloutStore {
         let (tx, rx) = mpsc::channel::<RolloutCmd>(256);
 
         let source = Arc::new(Mutex::new(source));
-        let mut writer = JsonlWriter { file };
-        let mut state_builder = state_builder;
-        if let Some(session_meta_line) = meta {
-            write_session_meta(
-                &mut writer,
-                session_meta_line,
-                &rollout_path,
-                &source,
-                state_db_ctx.as_deref(),
-                &mut state_builder,
-                config.model_provider_id.as_str(),
-            )
-            .await?;
-        }
         // Spawn a Tokio task that owns the file handle and performs async
         // writes. Using `tokio::fs::File` keeps everything on the async I/O
         // driver instead of blocking the runtime.
         tokio::task::spawn(rollout_writer(
-            writer.file,
+            file,
+            deferred_log_file_info,
             rx,
+            meta,
             rollout_path.clone(),
+            Arc::clone(&source),
             state_db_ctx.clone(),
             state_builder,
             config.model_provider_id.clone(),
+            git_info_handle,
         ));
 
         Ok(Self {
@@ -886,18 +893,39 @@ fn open_log_file(path: &Path) -> std::io::Result<File> {
 
 #[allow(clippy::too_many_arguments)]
 async fn rollout_writer(
-    file: tokio::fs::File,
+    file: Option<tokio::fs::File>,
+    mut deferred_log_file_info: Option<LogFileInfo>,
     mut rx: mpsc::Receiver<RolloutCmd>,
+    mut meta: Option<SessionMetaLine>,
     rollout_path: PathBuf,
+    source: Arc<Mutex<InMemoryRolloutSource>>,
     state_db_ctx: Option<StateDbHandle>,
     mut state_builder: Option<ThreadMetadataBuilder>,
     default_provider: String,
+    mut git_info_handle: Option<tokio::task::JoinHandle<Option<codex_protocol::protocol::GitInfo>>>,
 ) -> std::io::Result<()> {
-    let mut writer = JsonlWriter { file };
+    let mut writer = file.map(|file| JsonlWriter { file });
+    let mut buffered_items = Vec::<RolloutItem>::new();
     if let Some(builder) = state_builder.as_mut() {
         builder.rollout_path = rollout_path.clone();
     }
 
+    if let Some(writer) = writer.as_mut()
+        && let Some(session_meta_line) = meta.take()
+    {
+        write_session_meta(
+            writer,
+            session_meta_line,
+            &rollout_path,
+            &source,
+            state_db_ctx.as_deref(),
+            &mut state_builder,
+            default_provider.as_str(),
+            &mut git_info_handle,
+        )
+        .await?;
+    }
+
     // Process rollout commands
     while let Some(cmd) = rx.recv().await {
         match cmd {
@@ -906,8 +934,16 @@ async fn rollout_writer(
                     continue;
                 }
 
+                if writer.is_none() {
+                    buffered_items.extend(items);
+                    continue;
+                }
+
+                let Some(writer) = writer.as_mut() else {
+                    continue;
+                };
                 write_and_reconcile_items(
-                    &mut writer,
+                    writer,
                     items.as_slice(),
                     &rollout_path,
                     state_db_ctx.as_deref(),
@@ -917,14 +953,76 @@ async fn rollout_writer(
                 .await?;
             }
             RolloutCmd::Persist { ack } => {
-                if let Err(e) = writer.file.flush().await {
+                if writer.is_none() {
+                    let result = async {
+                        let Some(log_file_info) = deferred_log_file_info.take() else {
+                            return Err(IoError::other(
+                                "deferred rollout store missing log file metadata",
+                            ));
+                        };
+                        let file = open_log_file(log_file_info.path.as_path())?;
+                        writer = Some(JsonlWriter {
+                            file: tokio::fs::File::from_std(file),
+                        });
+
+                        if let Some(session_meta_line) = meta.take() {
+                            let Some(writer) = writer.as_mut() else {
+                                return Err(IoError::other(
+                                    "writer missing after rollout materialization",
+                                ));
+                            };
+                            write_session_meta(
+                                writer,
+                                session_meta_line,
+                                &rollout_path,
+                                &source,
+                                state_db_ctx.as_deref(),
+                                &mut state_builder,
+                                default_provider.as_str(),
+                                &mut git_info_handle,
+                            )
+                            .await?;
+                        }
+
+                        if !buffered_items.is_empty() {
+                            let Some(writer) = writer.as_mut() else {
+                                return Err(IoError::other(
+                                    "writer missing after rollout materialization",
+                                ));
+                            };
+                            write_and_reconcile_items(
+                                writer,
+                                buffered_items.as_slice(),
+                                &rollout_path,
+                                state_db_ctx.as_deref(),
+                                state_builder.as_ref(),
+                                default_provider.as_str(),
+                            )
+                            .await?;
+                            buffered_items.clear();
+                        }
+
+                        Ok(())
+                    }
+                    .await;
+
+                    if let Err(err) = result {
+                        let _ = ack.send(());
+                        return Err(err);
+                    }
+                }
+                if let Some(writer) = writer.as_mut()
+                    && let Err(e) = writer.file.flush().await
+                {
                     let _ = ack.send(());
                     return Err(e);
                 }
                 let _ = ack.send(());
             }
             RolloutCmd::Flush { ack } => {
-                if let Err(e) = writer.file.flush().await {
+                if let Some(writer) = writer.as_mut()
+                    && let Err(e) = writer.file.flush().await
+                {
                     let _ = ack.send(());
                     return Err(e);
                 }
@@ -948,9 +1046,21 @@ async fn write_session_meta(
     state_db_ctx: Option<&StateRuntime>,
     state_builder: &mut Option<ThreadMetadataBuilder>,
     default_provider: &str,
+    git_info_handle: &mut Option<
+        tokio::task::JoinHandle<Option<codex_protocol::protocol::GitInfo>>,
+    >,
 ) -> std::io::Result<()> {
     if session_meta_line.git.is_none() {
-        session_meta_line.git = collect_git_info(&session_meta_line.meta.cwd).await;
+        session_meta_line.git = match git_info_handle.take() {
+            Some(handle) => match handle.await {
+                Ok(git_info) => git_info,
+                Err(err) => {
+                    warn!("failed waiting for startup git info: {err}");
+                    collect_git_info(&session_meta_line.meta.cwd).await
+                }
+            },
+            None => collect_git_info(&session_meta_line.meta.cwd).await,
+        };
     }
     let memory_mode = session_meta_line.meta.memory_mode.clone();
     if state_db_ctx.is_some() {
@@ -1272,7 +1382,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn store_creates_rollout_immediately() -> std::io::Result<()> {
+    async fn recorder_materializes_only_after_explicit_persist() -> std::io::Result<()> {
         let home = TempDir::new().expect("temp dir");
         let config = ConfigBuilder::default()
             .codex_home(home.path().to_path_buf())
@@ -1296,10 +1406,45 @@ mod tests {
 
         let rollout_path = store.rollout_path().to_path_buf();
         assert!(
-            rollout_path.exists(),
-            "rollout file should exist at startup"
+            !rollout_path.exists(),
+            "rollout file should not exist before first user message"
         );
 
+        store
+            .record_items(&[RolloutItem::EventMsg(EventMsg::AgentMessage(
+                AgentMessageEvent {
+                    message: "buffered-event".to_string(),
+                    phase: None,
+                },
+            ))])
+            .await?;
+        store.flush().await?;
+        assert!(
+            !rollout_path.exists(),
+            "rollout file should remain deferred before first user message"
+        );
+
+        store
+            .record_items(&[RolloutItem::EventMsg(EventMsg::UserMessage(
+                UserMessageEvent {
+                    message: "first-user-message".to_string(),
+                    images: None,
+                    local_images: Vec::new(),
+                    text_elements: Vec::new(),
+                },
+            ))])
+            .await?;
+        store.flush().await?;
+        assert!(
+            !rollout_path.exists(),
+            "user-message-like items should not materialize without explicit persist"
+        );
+
+        store.persist().await?;
+        // Second call verifies `persist()` is idempotent after materialization.
+        store.persist().await?;
+        assert!(rollout_path.exists(), "rollout file should be materialized");
+
         let source_session_meta = {
             let source = store.source.lock().await;
             source
@@ -1334,33 +1479,6 @@ mod tests {
             serde_json::to_value(collect_git_info(&config.cwd).await)?,
         );
 
-        store
-            .record_items(&[RolloutItem::EventMsg(EventMsg::AgentMessage(
-                AgentMessageEvent {
-                    message: "buffered-event".to_string(),
-                    phase: None,
-                },
-            ))])
-            .await?;
-        store.flush().await?;
-
-        store
-            .record_items(&[RolloutItem::EventMsg(EventMsg::UserMessage(
-                UserMessageEvent {
-                    message: "first-user-message".to_string(),
-                    images: None,
-                    local_images: Vec::new(),
-                    text_elements: Vec::new(),
-                },
-            ))])
-            .await?;
-        store.flush().await?;
-
-        store.persist().await?;
-        // Second call verifies `persist()` is idempotent after startup materialization.
-        store.persist().await?;
-        assert!(rollout_path.exists(), "rollout file should exist");
-
         let text = std::fs::read_to_string(&rollout_path)?;
         assert!(
             text.contains("\"type\":\"session_meta\""),

codex-rs/core/src/tasks/review.rs

@@ -265,7 +265,8 @@ pub(crate) async fn exit_review_mode(
         )
         .await;
 
-    // Review turns record rollout items asynchronously, so explicitly flush
-    // before any follow-up path snapshots the thread's JSONL.
-    session.flush_rollout().await;
+    // Review turns can run before any regular user turn, so explicitly
+    // materialize rollout persistence. Do this after emitting review output so
+    // file creation + git metadata collection cannot delay client-facing items.
+    session.ensure_rollout_materialized().await;
 }

codex-rs/core/tests/suite/sqlite_state.rs

@@ -60,10 +60,18 @@ async fn new_thread_is_recorded_in_state_db() -> Result<()> {
 
     let db = test.codex.state_db().expect("state db enabled");
     assert!(
-        rollout_path.exists(),
-        "fresh thread rollout should exist at startup"
+        !rollout_path.exists(),
+        "fresh thread rollout should not be materialized before first user message"
     );
 
+    let initial_metadata = db.get_thread(thread_id).await?;
+    assert!(
+        initial_metadata.is_none(),
+        "fresh thread should not be recorded in state db before first user message"
+    );
+
+    test.submit_turn("materialize rollout").await?;
+
     let mut metadata = None;
     for _ in 0..100 {
         metadata = db.get_thread(thread_id).await?;
@@ -76,6 +84,10 @@ async fn new_thread_is_recorded_in_state_db() -> Result<()> {
     let metadata = metadata.expect("thread should exist in state db");
     assert_eq!(metadata.id, thread_id);
     assert_eq!(metadata.rollout_path, rollout_path);
+    assert!(
+        rollout_path.exists(),
+        "rollout should be materialized after first user message"
+    );
 
     Ok(())
 }

codex-rs/protocol/src/protocol.rs

@@ -3011,7 +3011,8 @@ pub struct SessionConfiguredEvent {
     #[ts(optional)]
     pub network_proxy: Option<SessionNetworkProxyRuntime>,
 
-    /// Path in which the rollout is stored. Can be `None` for ephemeral threads.
+    /// Path in which the rollout is stored. Fresh persistent threads may reserve this path
+    /// before the rollout file is materialized on disk. Can be `None` for ephemeral threads.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub rollout_path: Option<PathBuf>,
 }