diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index eda1f2fe62..85ca56c915 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/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; diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 70a7d87c61..cdb72e4b39 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/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, diff --git a/codex-rs/app-server/tests/suite/v2/thread_archive.rs b/codex-rs/app-server/tests/suite/v2/thread_archive.rs index 391b5db798..20fd6fd884 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_archive.rs +++ b/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::(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::(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(()) +} diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index 30bee751d0..1f19ae8b97 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/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::(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::(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(()) } diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index 2634bab726..dcc411538e 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/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::(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::(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::(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(()) } diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index 3df602d4c9..12a74ed6e6 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/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::(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::(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(()) } diff --git a/codex-rs/app-server/tests/suite/v2/thread_start.rs b/codex-rs/app-server/tests/suite/v2/thread_start.rs index 03b9647f52..bc383a19d1 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_start.rs +++ b/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. diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 26100b744d..00e403ce82 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/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 diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 5b45ee36c8..b0ea5818bb 100644 --- a/codex-rs/core/src/codex.rs +++ b/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( diff --git a/codex-rs/core/src/codex_tests.rs b/codex-rs/core/src/codex_tests.rs index dbb9e0318a..47f64cc402 100644 --- a/codex-rs/core/src/codex_tests.rs +++ b/codex-rs/core/src/codex_tests.rs @@ -1824,6 +1824,7 @@ async fn attach_rollout_recorder(session: &Arc) -> 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) diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index d625d9691c..e1a265fdf1 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/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 { self.rollout_path.clone() } diff --git a/codex-rs/core/src/rollout/recorder.rs b/codex-rs/core/src/rollout/recorder.rs index 1fc7ab44c6..b6ff516a7a 100644 --- a/codex-rs/core/src/rollout/recorder.rs +++ b/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, state_builder: Option, ) -> std::io::Result { - 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::(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 { #[allow(clippy::too_many_arguments)] async fn rollout_writer( - file: tokio::fs::File, + file: Option, + mut deferred_log_file_info: Option, mut rx: mpsc::Receiver, + mut meta: Option, rollout_path: PathBuf, + source: Arc>, state_db_ctx: Option, mut state_builder: Option, default_provider: String, + mut git_info_handle: Option>>, ) -> std::io::Result<()> { - let mut writer = JsonlWriter { file }; + let mut writer = file.map(|file| JsonlWriter { file }); + let mut buffered_items = Vec::::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, default_provider: &str, + git_info_handle: &mut Option< + tokio::task::JoinHandle>, + >, ) -> 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\""), diff --git a/codex-rs/core/src/tasks/review.rs b/codex-rs/core/src/tasks/review.rs index 2aa60904ce..1146be615d 100644 --- a/codex-rs/core/src/tasks/review.rs +++ b/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; } diff --git a/codex-rs/core/tests/suite/sqlite_state.rs b/codex-rs/core/tests/suite/sqlite_state.rs index bf4d8d08dc..b17219e5f1 100644 --- a/codex-rs/core/tests/suite/sqlite_state.rs +++ b/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(()) } diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 0a80cbae35..372d8894ca 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -3011,7 +3011,8 @@ pub struct SessionConfiguredEvent { #[ts(optional)] pub network_proxy: Option, - /// 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, }