From b3de6c7f2b5fd5f8e5a2cf98e06f1615d858329b Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Sat, 7 Feb 2026 23:05:03 -0800 Subject: [PATCH] Defer persistence of rollout file (#11028) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Defer rollout persistence for fresh threads (`InitialHistory::New`): keep rollout events in memory and only materialize rollout file + state DB row on first `EventMsg::UserMessage`. - Keep precomputed rollout path available before materialization. - Change `thread/start` to build thread response from live config snapshot and optional precomputed path. - Improve pre-materialization behavior in app-server/TUI: clearer invalid-request errors for file-backed ops and a friendlier `/fork` “not ready yet” UX. - Update tests to match deferred semantics across start/read/archive/unarchive/fork/resume/review flows. - Improved resilience of user_shell test, which should be unrelated to this change but must be affected by timing changes For Reviewers: * The primary change is in recorder.rs * Most of the other changes were to fix up broken assumptions in existing tests Testing: * Manually tested CLI * Exercised app server paths by manually running IDE Extension with rebuilt CLI binary * Only user-visible change is that `/fork` in TUI generates visible error if used prior to first turn --- .../app-server/src/codex_message_processor.rs | 56 ++- .../app-server/tests/suite/archive_thread.rs | 72 +++- codex-rs/app-server/tests/suite/v2/review.rs | 27 ++ .../tests/suite/v2/thread_archive.rs | 104 ++++- .../app-server/tests/suite/v2/thread_fork.rs | 48 +++ .../app-server/tests/suite/v2/thread_read.rs | 102 +++++ .../tests/suite/v2/thread_resume.rs | 89 +++- .../app-server/tests/suite/v2/thread_start.rs | 37 ++ .../tests/suite/v2/thread_unarchive.rs | 60 ++- codex-rs/core/src/codex.rs | 19 +- codex-rs/core/src/rollout/recorder.rs | 393 ++++++++++++++---- codex-rs/core/src/tasks/review.rs | 6 + codex-rs/core/src/tools/handlers/collab.rs | 22 +- codex-rs/core/tests/suite/memory_tool.rs | 17 + .../core/tests/suite/rollout_list_find.rs | 1 + codex-rs/core/tests/suite/sqlite_state.rs | 16 + codex-rs/core/tests/suite/user_shell_cmd.rs | 22 +- codex-rs/tui/src/app.rs | 81 ++-- codex-rs/tui/src/chatwidget.rs | 6 + 19 files changed, 983 insertions(+), 195 deletions(-) diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 29fbd3d74..3f65cf287 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -1878,31 +1878,11 @@ impl CodexMessageProcessor { .. } = new_conv; let config_snapshot = thread.config_snapshot().await; - let fallback_provider = self.config.model_provider_id.as_str(); - - // A bit hacky, but the summary contains a lot of useful information for the thread - // that unfortunately does not get returned from thread_manager.start_thread(). - let thread = match session_configured.rollout_path.as_ref() { - Some(rollout_path) => { - match read_summary_from_rollout(rollout_path.as_path(), fallback_provider) - .await - { - Ok(summary) => summary_to_thread(summary), - Err(err) => { - self.send_internal_error( - request_id, - format!( - "failed to load rollout `{}` for thread {thread_id}: {err}", - rollout_path.display() - ), - ) - .await; - return; - } - } - } - None => build_ephemeral_thread(thread_id, &config_snapshot), - }; + let thread = build_thread_from_snapshot( + thread_id, + &config_snapshot, + session_configured.rollout_path.clone(), + ); let response = ThreadStartResponse { thread: thread.clone(), @@ -2498,7 +2478,8 @@ impl CodexMessageProcessor { return; }; let config_snapshot = thread.config_snapshot().await; - if include_turns { + let loaded_rollout_path = thread.rollout_path(); + if include_turns && loaded_rollout_path.is_none() { self.send_invalid_request_error( request_id, "ephemeral threads do not support includeTurns".to_string(), @@ -2506,7 +2487,10 @@ impl CodexMessageProcessor { .await; return; } - build_ephemeral_thread(thread_uuid, &config_snapshot) + if include_turns { + rollout_path = loaded_rollout_path.clone(); + } + build_thread_from_snapshot(thread_uuid, &config_snapshot, loaded_rollout_path) }; if include_turns && let Some(rollout_path) = rollout_path.as_ref() { @@ -2514,6 +2498,16 @@ impl CodexMessageProcessor { Ok(events) => { thread.turns = build_turns_from_event_msgs(&events); } + 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, @@ -5793,7 +5787,11 @@ async fn read_updated_at(path: &Path, created_at: Option<&str>) -> Option Thread { +fn build_thread_from_snapshot( + thread_id: ThreadId, + config_snapshot: &ThreadConfigSnapshot, + path: Option, +) -> Thread { let now = time::OffsetDateTime::now_utc().unix_timestamp(); Thread { id: thread_id.to_string(), @@ -5801,7 +5799,7 @@ fn build_ephemeral_thread(thread_id: ThreadId, config_snapshot: &ThreadConfigSna model_provider: config_snapshot.model_provider_id.clone(), created_at: now, updated_at: now, - path: None, + path, cwd: config_snapshot.cwd.clone(), cli_version: env!("CARGO_PKG_VERSION").to_string(), source: config_snapshot.session_source.clone().into(), diff --git a/codex-rs/app-server/tests/suite/archive_thread.rs b/codex-rs/app-server/tests/suite/archive_thread.rs index b6e85936f..a47b247bf 100644 --- a/codex-rs/app-server/tests/suite/archive_thread.rs +++ b/codex-rs/app-server/tests/suite/archive_thread.rs @@ -1,12 +1,19 @@ 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::AddConversationListenerParams; +use codex_app_server_protocol::AddConversationSubscriptionResponse; use codex_app_server_protocol::ArchiveConversationParams; use codex_app_server_protocol::ArchiveConversationResponse; +use codex_app_server_protocol::InputItem; +use codex_app_server_protocol::JSONRPCNotification; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::NewConversationParams; use codex_app_server_protocol::NewConversationResponse; use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::SendUserMessageParams; +use codex_app_server_protocol::SendUserMessageResponse; use codex_core::ARCHIVED_SESSIONS_SUBDIR; use std::path::Path; use tempfile::TempDir; @@ -16,8 +23,9 @@ const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn archive_conversation_moves_rollout_into_archived_directory() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; - create_config_toml(codex_home.path())?; + create_config_toml(codex_home.path(), &server.uri())?; let mut mcp = McpProcess::new(codex_home.path()).await?; timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; @@ -40,9 +48,50 @@ async fn archive_conversation_moves_rollout_into_archived_directory() -> Result< .. } = to_response::(new_response)?; + assert!( + !rollout_path.exists(), + "expected rollout path {} to be deferred until first user message", + rollout_path.display() + ); + + let add_listener_request_id = mcp + .send_add_conversation_listener_request(AddConversationListenerParams { + conversation_id, + experimental_raw_events: false, + }) + .await?; + let add_listener_response: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(add_listener_request_id)), + ) + .await??; + let AddConversationSubscriptionResponse { subscription_id: _ } = + to_response::(add_listener_response)?; + + let send_request_id = mcp + .send_send_user_message_request(SendUserMessageParams { + conversation_id, + items: vec![InputItem::Text { + text: "materialize".to_string(), + text_elements: Vec::new(), + }], + }) + .await?; + let send_response: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(send_request_id)), + ) + .await??; + let _: SendUserMessageResponse = to_response::(send_response)?; + let _: JSONRPCNotification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("codex/event/task_complete"), + ) + .await??; + assert!( rollout_path.exists(), - "expected rollout path {} to exist", + "expected rollout path {} to exist after first user message", rollout_path.display() ); @@ -81,14 +130,25 @@ async fn archive_conversation_moves_rollout_into_archived_directory() -> Result< Ok(()) } -fn create_config_toml(codex_home: &Path) -> std::io::Result<()> { +fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> { let config_toml = codex_home.join("config.toml"); - std::fs::write(config_toml, config_contents()) + std::fs::write(config_toml, config_contents(server_uri)) } -fn config_contents() -> &'static str { - r#"model = "mock-model" +fn config_contents(server_uri: &str) -> String { + format!( + r#"model = "mock-model" approval_policy = "never" sandbox_mode = "read-only" + +model_provider = "mock_provider" + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "{server_uri}/v1" +wire_api = "responses" +request_max_retries = 0 +stream_max_retries = 0 "# + ) } diff --git a/codex-rs/app-server/tests/suite/v2/review.rs b/codex-rs/app-server/tests/suite/v2/review.rs index 5b4b04297..2f919eee3 100644 --- a/codex-rs/app-server/tests/suite/v2/review.rs +++ b/codex-rs/app-server/tests/suite/v2/review.rs @@ -19,7 +19,9 @@ use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStatus; +use codex_app_server_protocol::UserInput as V2UserInput; use serde_json::json; use tempfile::TempDir; use tokio::time::timeout; @@ -270,6 +272,7 @@ async fn review_start_with_detached_delivery_returns_new_thread_id() -> Result<( timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; let thread_id = start_default_thread(&mut mcp).await?; + materialize_thread_rollout(&mut mcp, &thread_id).await?; let review_req = mcp .send_review_start_request(ReviewStartParams { @@ -387,6 +390,30 @@ async fn start_default_thread(mcp: &mut McpProcess) -> Result { Ok(thread.id) } +async fn materialize_thread_rollout(mcp: &mut McpProcess, thread_id: &str) -> Result<()> { + let turn_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread_id.to_string(), + input: vec![V2UserInput::Text { + text: "materialize rollout".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_req)), + ) + .await??; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + Ok(()) +} + fn create_config_toml(codex_home: &std::path::Path, server_uri: &str) -> std::io::Result<()> { create_config_toml_with_approval_policy(codex_home, server_uri, "never") } 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 b8cdd426a..8fccd2227 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_archive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_archive.rs @@ -1,12 +1,17 @@ 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; use codex_app_server_protocol::ThreadArchiveResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::UserInput; use codex_core::ARCHIVED_SESSIONS_SUBDIR; use codex_core::find_thread_path_by_id_str; use std::path::Path; @@ -16,9 +21,10 @@ use tokio::time::timeout; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); #[tokio::test] -async fn thread_archive_moves_rollout_into_archived_directory() -> 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())?; + create_config_toml(codex_home.path(), &server.uri())?; let mut mcp = McpProcess::new(codex_home.path()).await?; timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; @@ -38,17 +44,73 @@ async fn thread_archive_moves_rollout_into_archived_directory() -> Result<()> { let ThreadStartResponse { thread, .. } = to_response::(start_resp)?; assert!(!thread.id.is_empty()); - // Locate the rollout path recorded for this thread id. - let rollout_path = find_thread_path_by_id_str(codex_home.path(), &thread.id) - .await? - .expect("expected rollout path for thread id to exist"); + let rollout_path = thread.path.clone().expect("thread path"); + 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_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 {} to exist", + "expected rollout path {} to exist after first user message", rollout_path.display() ); - // Archive the thread. + 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(), @@ -80,14 +142,32 @@ async fn thread_archive_moves_rollout_into_archived_directory() -> Result<()> { Ok(()) } -fn create_config_toml(codex_home: &Path) -> std::io::Result<()> { +fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> { let config_toml = codex_home.join("config.toml"); - std::fs::write(config_toml, config_contents()) + std::fs::write(config_toml, config_contents(server_uri)) } -fn config_contents() -> &'static str { - r#"model = "mock-model" +fn config_contents(server_uri: &str) -> String { + format!( + r#"model = "mock-model" approval_policy = "never" sandbox_mode = "read-only" + +model_provider = "mock_provider" + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "{server_uri}/v1" +wire_api = "responses" +request_max_retries = 0 +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 c06f387fd..1e2e172cf 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::JSONRPCNotification; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::RequestId; @@ -10,6 +11,8 @@ use codex_app_server_protocol::SessionSource; use codex_app_server_protocol::ThreadForkParams; use codex_app_server_protocol::ThreadForkResponse; use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStartedNotification; use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::UserInput; @@ -117,6 +120,51 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> { Ok(()) } +#[tokio::test] +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())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let start_id = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response::(start_resp)?; + + let fork_id = mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: thread.id, + ..Default::default() + }) + .await?; + let fork_err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(fork_id)), + ) + .await??; + assert!( + fork_err + .error + .message + .contains("no rollout found for thread id"), + "unexpected fork error: {}", + fork_err.error.message + ); + + Ok(()) +} + // Helper to create a config.toml pointing at the mock model server. fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> { let config_toml = codex_home.join("config.toml"); 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 d8ca3aa69..338728a12 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -3,12 +3,15 @@ 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; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadReadParams; use codex_app_server_protocol::ThreadReadResponse; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::UserInput; use codex_protocol::user_input::ByteRange; @@ -134,6 +137,105 @@ async fn thread_read_can_include_turns() -> Result<()> { Ok(()) } +#[tokio::test] +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())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let start_id = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .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 not be materialized yet" + ); + + let read_id = mcp + .send_thread_read_request(ThreadReadParams { + thread_id: thread.id.clone(), + include_turns: false, + }) + .await?; + let read_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_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); + + Ok(()) +} + +#[tokio::test] +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())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let start_id = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .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 not be materialized yet" + ); + + let read_id = mcp + .send_thread_read_request(ThreadReadParams { + thread_id: thread.id.clone(), + include_turns: true, + }) + .await?; + let read_err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(read_id)), + ) + .await??; + + assert!( + read_err + .error + .message + .contains("includeTurns is unavailable before first user message"), + "unexpected error: {}", + read_err.error.message + ); + + Ok(()) +} + // Helper to create a config.toml pointing at the mock model server. fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> { let config_toml = codex_home.join("config.toml"); 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 f9dcf49c9..ce4f300f0 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -35,7 +35,7 @@ const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs const CODEX_5_2_INSTRUCTIONS_TEMPLATE_DEFAULT: &str = "You are Codex, a coding agent based on GPT-5. You and the user share the same workspace and collaborate to achieve the user's goals."; #[tokio::test] -async fn thread_resume_returns_original_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())?; @@ -57,24 +57,26 @@ async fn thread_resume_returns_original_thread() -> Result<()> { .await??; let ThreadStartResponse { thread, .. } = to_response::(start_resp)?; - // Resume it via v2 API. + // 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)?; - let mut expected = thread; - expected.updated_at = resumed.updated_at; - assert_eq!(resumed, expected); + assert!( + resume_err + .error + .message + .contains("no rollout found for thread id"), + "unexpected resume error: {}", + resume_err.error.message + ); Ok(()) } @@ -322,6 +324,27 @@ async fn thread_resume_prefers_path_over_thread_id() -> Result<()> { .await??; let ThreadStartResponse { thread, .. } = to_response::(start_resp)?; + let turn_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?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_id)), + ) + .await??; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + let thread_path = thread.path.clone().expect("thread path"); let resume_id = mcp .send_thread_resume_request(ThreadResumeParams { @@ -339,9 +362,8 @@ async fn thread_resume_prefers_path_over_thread_id() -> Result<()> { let ThreadResumeResponse { thread: resumed, .. } = to_response::(resume_resp)?; - let mut expected = thread; - expected.updated_at = resumed.updated_at; - assert_eq!(resumed, expected); + assert_eq!(resumed.id, thread.id); + assert_eq!(resumed.path, thread.path); Ok(()) } @@ -412,12 +434,17 @@ async fn thread_resume_accepts_personality_override() -> Result<()> { skip_if_no_network!(Ok(())); let server = responses::start_mock_server().await; - let body = responses::sse(vec![ + let first_body = responses::sse(vec![ responses::ev_response_created("resp-1"), responses::ev_assistant_message("msg-1", "Done"), responses::ev_completed("resp-1"), ]); - let response_mock = responses::mount_sse_once(&server, body).await; + let second_body = responses::sse(vec![ + responses::ev_response_created("resp-2"), + responses::ev_assistant_message("msg-2", "Done"), + responses::ev_completed("resp-2"), + ]); + let response_mock = responses::mount_sse_sequence(&server, vec![first_body, second_body]).await; let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; @@ -438,9 +465,30 @@ async fn thread_resume_accepts_personality_override() -> Result<()> { .await??; let ThreadStartResponse { thread, .. } = to_response::(start_resp)?; + let materialize_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "seed history".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(materialize_id)), + ) + .await??; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + let resume_id = mcp .send_thread_resume_request(ThreadResumeParams { - thread_id: thread.id.clone(), + thread_id: thread.id, model: Some("gpt-5.2-codex".to_string()), personality: Some(Personality::Friendly), ..Default::default() @@ -451,11 +499,11 @@ async fn thread_resume_accepts_personality_override() -> Result<()> { mcp.read_stream_until_response_message(RequestId::Integer(resume_id)), ) .await??; - let _resume: ThreadResumeResponse = to_response::(resume_resp)?; + let resume: ThreadResumeResponse = to_response::(resume_resp)?; let turn_id = mcp .send_turn_start_request(TurnStartParams { - thread_id: thread.id, + thread_id: resume.thread.id, input: vec![UserInput::Text { text: "Hello".to_string(), text_elements: Vec::new(), @@ -475,7 +523,10 @@ async fn thread_resume_accepts_personality_override() -> Result<()> { ) .await??; - let request = response_mock.single_request(); + let requests = response_mock.requests(); + let request = requests + .last() + .expect("expected request for resumed thread turn"); let developer_texts = request.message_input_texts("developer"); assert!( developer_texts 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 9b9104acf..b7ee9a13e 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_start.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_start.rs @@ -59,6 +59,12 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> { thread.created_at > 0, "created_at should be a positive UNIX timestamp" ); + 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 not be materialized until first user message" + ); // A corresponding thread/started notification should arrive. let notif: JSONRPCNotification = timeout( @@ -114,6 +120,37 @@ model_reasoning_effort = "high" Ok(()) } +#[tokio::test] +async fn thread_start_ephemeral_remains_pathless() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let req_id = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("gpt-5.1".to_string()), + ephemeral: Some(true), + ..Default::default() + }) + .await?; + + let resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(req_id)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response::(resp)?; + assert_eq!( + thread.path, None, + "ephemeral threads should not expose a path" + ); + + Ok(()) +} + #[tokio::test] async fn thread_start_fails_when_required_mcp_server_fails_to_initialize() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; diff --git a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs index dada1cbf2..a6c81728d 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs @@ -1,5 +1,6 @@ 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::JSONRPCResponse; use codex_app_server_protocol::RequestId; @@ -9,6 +10,9 @@ use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadUnarchiveParams; use codex_app_server_protocol::ThreadUnarchiveResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::UserInput; use codex_core::find_archived_thread_path_by_id_str; use codex_core::find_thread_path_by_id_str; use std::fs::FileTimes; @@ -23,8 +27,9 @@ const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs #[tokio::test] async fn thread_unarchive_moves_rollout_back_into_sessions_directory() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; - create_config_toml(codex_home.path())?; + create_config_toml(codex_home.path(), &server.uri())?; let mut mcp = McpProcess::new(codex_home.path()).await?; timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; @@ -42,9 +47,34 @@ async fn thread_unarchive_moves_rollout_back_into_sessions_directory() -> Result .await??; let ThreadStartResponse { thread, .. } = to_response::(start_resp)?; - let rollout_path = find_thread_path_by_id_str(codex_home.path(), &thread.id) + let rollout_path = thread.path.clone().expect("thread path"); + + 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??; + + let found_rollout_path = find_thread_path_by_id_str(codex_home.path(), &thread.id) .await? .expect("expected rollout path for thread id to exist"); + assert_paths_match_on_disk(&found_rollout_path, &rollout_path)?; let archive_id = mcp .send_thread_archive_request(ThreadArchiveParams { @@ -108,14 +138,32 @@ async fn thread_unarchive_moves_rollout_back_into_sessions_directory() -> Result Ok(()) } -fn create_config_toml(codex_home: &Path) -> std::io::Result<()> { +fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> { let config_toml = codex_home.join("config.toml"); - std::fs::write(config_toml, config_contents()) + std::fs::write(config_toml, config_contents(server_uri)) } -fn config_contents() -> &'static str { - r#"model = "mock-model" +fn config_contents(server_uri: &str) -> String { + format!( + r#"model = "mock-model" approval_policy = "never" sandbox_mode = "read-only" + +model_provider = "mock_provider" + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "{server_uri}/v1" +wire_api = "responses" +request_max_retries = 0 +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/core/src/codex.rs b/codex-rs/core/src/codex.rs index 3840aa708..8280b5a9e 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -314,7 +314,7 @@ impl Codex { // Resolve base instructions for the session. Priority order: // 1. config.base_instructions override // 2. conversation history => session_meta.base_instructions - // 3. base_intructions for current model + // 3. base_instructions for current model let model_info = models_manager.get_model_info(model.as_str(), &config).await; let base_instructions = config .base_instructions @@ -1211,6 +1211,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 @@ -1326,6 +1338,10 @@ impl Session { let mut state = self.state.lock().await; state.initial_context_seeded = true; } + + // Forked threads should remain file-backed immediately after startup. + self.ensure_rollout_materialized().await; + // Flush after seeding history and any persisted rollout copy. self.flush_rollout().await; } @@ -2347,6 +2363,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/rollout/recorder.rs b/codex-rs/core/src/rollout/recorder.rs index c42bbb797..9520da8de 100644 --- a/codex-rs/core/src/rollout/recorder.rs +++ b/codex-rs/core/src/rollout/recorder.rs @@ -47,6 +47,7 @@ use codex_protocol::protocol::RolloutLine; use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; +use codex_state::StateRuntime; use codex_state::ThreadMetadataBuilder; /// Records all [`ResponseItem`]s for a session and flushes them to disk after @@ -81,6 +82,9 @@ pub enum RolloutRecorderParams { enum RolloutCmd { AddItems(Vec), + Persist { + ack: oneshot::Sender<()>, + }, /// Ensure all prior writes are processed; respond when flushed. Flush { ack: oneshot::Sender<()>, @@ -279,16 +283,19 @@ impl RolloutRecorder { } } - /// Attempt to create a new [`RolloutRecorder`]. If the sessions directory - /// cannot be created or the rollout file cannot be opened we return the - /// error so the caller can decide whether to disable persistence. + /// Attempt to create a new [`RolloutRecorder`]. + /// + /// For newly created sessions, this precomputes path/metadata and defers + /// file creation/open until an explicit `persist()` call. + /// + /// For resumed sessions, this immediately opens the existing rollout file. pub async fn new( config: &Config, params: RolloutRecorderParams, state_db_ctx: Option, state_builder: Option, ) -> std::io::Result { - let (file, rollout_path, meta) = match params { + let (file, deferred_log_file_info, rollout_path, meta) = match params { RolloutRecorderParams::Create { conversation_id, forked_from_id, @@ -296,47 +303,46 @@ impl RolloutRecorder { base_instructions, dynamic_tools, } => { - let LogFileInfo { - file, - path, - conversation_id: session_id, - timestamp, - } = create_log_file(config, conversation_id)?; + let log_file_info = precompute_log_file_info(config, conversation_id)?; + let path = log_file_info.path.clone(); + let session_id = log_file_info.conversation_id; + let started_at = log_file_info.timestamp; let timestamp_format: &[FormatItem] = format_description!( "[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z" ); - let timestamp = timestamp + let timestamp = started_at .to_offset(time::UtcOffset::UTC) .format(timestamp_format) .map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?; - ( - tokio::fs::File::from_std(file), - path, - Some(SessionMeta { - id: session_id, - forked_from_id, - timestamp, - cwd: config.cwd.clone(), - originator: originator().value, - cli_version: env!("CARGO_PKG_VERSION").to_string(), - source, - model_provider: Some(config.model_provider_id.clone()), - base_instructions: Some(base_instructions), - dynamic_tools: if dynamic_tools.is_empty() { - None - } else { - Some(dynamic_tools) - }, - }), - ) + let session_meta = SessionMeta { + id: session_id, + forked_from_id, + timestamp, + cwd: config.cwd.clone(), + originator: originator().value, + cli_version: env!("CARGO_PKG_VERSION").to_string(), + source, + model_provider: Some(config.model_provider_id.clone()), + base_instructions: Some(base_instructions), + dynamic_tools: if dynamic_tools.is_empty() { + None + } else { + Some(dynamic_tools) + }, + }; + + (None, Some(log_file_info), path, Some(session_meta)) } RolloutRecorderParams::Resume { path } => ( - tokio::fs::OpenOptions::new() - .append(true) - .open(&path) - .await?, + Some( + tokio::fs::OpenOptions::new() + .append(true) + .open(&path) + .await?, + ), + None, path, None, ), @@ -355,6 +361,7 @@ impl RolloutRecorder { // driver instead of blocking the runtime. tokio::task::spawn(rollout_writer( file, + deferred_log_file_info, rx, meta, cwd, @@ -398,6 +405,19 @@ impl RolloutRecorder { .map_err(|e| IoError::other(format!("failed to queue rollout items: {e}"))) } + /// Materialize the rollout file and persist all buffered items. + /// + /// This is idempotent; after first materialization, repeated calls are no-ops. + pub async fn persist(&self) -> std::io::Result<()> { + let (tx, rx) = oneshot::channel(); + self.tx + .send(RolloutCmd::Persist { ack: tx }) + .await + .map_err(|e| IoError::other(format!("failed to queue rollout persist: {e}")))?; + rx.await + .map_err(|e| IoError::other(format!("failed waiting for rollout persist: {e}"))) + } + /// Flush all queued writes and wait until they are committed by the writer task. pub async fn flush(&self) -> std::io::Result<()> { let (tx, rx) = oneshot::channel(); @@ -508,9 +528,6 @@ impl RolloutRecorder { } struct LogFileInfo { - /// Opened file handle to the rollout file. - file: File, - /// Full path to the rollout file. path: PathBuf, @@ -521,8 +538,11 @@ struct LogFileInfo { timestamp: OffsetDateTime, } -fn create_log_file(config: &Config, conversation_id: ThreadId) -> std::io::Result { - // Resolve ~/.codex/sessions/YYYY/MM/DD and create it if missing. +fn precompute_log_file_info( + config: &Config, + conversation_id: ThreadId, +) -> std::io::Result { + // Resolve ~/.codex/sessions/YYYY/MM/DD path. let timestamp = OffsetDateTime::now_local() .map_err(|e| IoError::other(format!("failed to get local time: {e}")))?; let mut dir = config.codex_home.clone(); @@ -530,7 +550,6 @@ fn create_log_file(config: &Config, conversation_id: ThreadId) -> std::io::Resul dir.push(timestamp.year().to_string()); dir.push(format!("{:02}", u8::from(timestamp.month()))); dir.push(format!("{:02}", timestamp.day())); - fs::create_dir_all(&dir)?; // Custom format for YYYY-MM-DDThh-mm-ss. Use `-` instead of `:` for // compatibility with filesystems that do not allow colons in filenames. @@ -543,22 +562,32 @@ fn create_log_file(config: &Config, conversation_id: ThreadId) -> std::io::Resul let filename = format!("rollout-{date_str}-{conversation_id}.jsonl"); let path = dir.join(filename); - let file = std::fs::OpenOptions::new() - .append(true) - .create(true) - .open(&path)?; Ok(LogFileInfo { - file, path, conversation_id, timestamp, }) } +fn open_log_file(path: &Path) -> std::io::Result { + let Some(parent) = path.parent() else { + return Err(IoError::other(format!( + "rollout path has no parent: {}", + path.display() + ))); + }; + fs::create_dir_all(parent)?; + std::fs::OpenOptions::new() + .append(true) + .create(true) + .open(path) +} + #[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, cwd: std::path::PathBuf, @@ -567,35 +596,27 @@ async fn rollout_writer( mut state_builder: Option, default_provider: String, ) -> 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 we have a meta, collect git info asynchronously and write meta first - if let Some(session_meta) = meta.take() { - let git_info = collect_git_info(&cwd).await; - let session_meta_line = SessionMetaLine { - meta: session_meta, - git: git_info, - }; - if state_db_ctx.is_some() { - state_builder = - metadata::builder_from_session_meta(&session_meta_line, rollout_path.as_path()); - } - - // Write the SessionMeta as the first item in the file, wrapped in a rollout line - let rollout_item = RolloutItem::SessionMeta(session_meta_line); - writer.write_rollout_item(&rollout_item).await?; - state_db::reconcile_rollout( + // Resumed sessions already have a file handle open, so session metadata can + // be written immediately if present. + if writer.is_some() + && let Some(session_meta) = meta.take() + { + write_session_meta( + writer.as_mut(), + session_meta, + &cwd, + &rollout_path, state_db_ctx.as_deref(), - rollout_path.as_path(), + &mut state_builder, default_provider.as_str(), - state_builder.as_ref(), - std::slice::from_ref(&rollout_item), - None, ) - .await; + .await?; } // Process rollout commands @@ -605,29 +626,83 @@ async fn rollout_writer( let mut persisted_items = Vec::new(); for item in items { if is_persisted_response_item(&item) { - writer.write_rollout_item(&item).await?; persisted_items.push(item); } } if persisted_items.is_empty() { continue; } - if let Some(builder) = state_builder.as_mut() { - builder.rollout_path = rollout_path.clone(); + + if writer.is_none() { + buffered_items.extend(persisted_items); + continue; } - state_db::apply_rollout_items( - state_db_ctx.as_deref(), - rollout_path.as_path(), - default_provider.as_str(), - state_builder.as_ref(), + + write_and_reconcile_items( + writer.as_mut(), persisted_items.as_slice(), - "rollout_writer", + &rollout_path, + state_db_ctx.as_deref(), + &mut state_builder, + default_provider.as_str(), ) - .await; + .await?; + } + RolloutCmd::Persist { ack } => { + if writer.is_none() { + let result = async { + let Some(log_file_info) = deferred_log_file_info.take() else { + return Err(IoError::other( + "deferred rollout recorder 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) = meta.take() { + write_session_meta( + writer.as_mut(), + session_meta, + &cwd, + &rollout_path, + state_db_ctx.as_deref(), + &mut state_builder, + default_provider.as_str(), + ) + .await?; + } + + if !buffered_items.is_empty() { + write_and_reconcile_items( + writer.as_mut(), + buffered_items.as_slice(), + &rollout_path, + state_db_ctx.as_deref(), + &mut state_builder, + default_provider.as_str(), + ) + .await?; + buffered_items.clear(); + } + + Ok(()) + } + .await; + + if let Err(err) = result { + let _ = ack.send(()); + return Err(err); + } + } + let _ = ack.send(()); } RolloutCmd::Flush { ack } => { - // Ensure underlying file is flushed and then ack. - if let Err(e) = writer.file.flush().await { + // Deferred fresh threads may not have an initialized file yet. + if let Some(writer) = writer.as_mut() + && let Err(e) = writer.file.flush().await + { let _ = ack.send(()); return Err(e); } @@ -642,6 +717,68 @@ async fn rollout_writer( Ok(()) } +async fn write_session_meta( + mut writer: Option<&mut JsonlWriter>, + session_meta: SessionMeta, + cwd: &Path, + rollout_path: &Path, + state_db_ctx: Option<&StateRuntime>, + state_builder: &mut Option, + default_provider: &str, +) -> std::io::Result<()> { + let git_info = collect_git_info(cwd).await; + let session_meta_line = SessionMetaLine { + meta: session_meta, + git: git_info, + }; + if state_db_ctx.is_some() { + *state_builder = metadata::builder_from_session_meta(&session_meta_line, rollout_path); + } + + let rollout_item = RolloutItem::SessionMeta(session_meta_line); + if let Some(writer) = writer.as_mut() { + writer.write_rollout_item(&rollout_item).await?; + } + state_db::reconcile_rollout( + state_db_ctx, + rollout_path, + default_provider, + state_builder.as_ref(), + std::slice::from_ref(&rollout_item), + None, + ) + .await; + Ok(()) +} + +async fn write_and_reconcile_items( + mut writer: Option<&mut JsonlWriter>, + items: &[RolloutItem], + rollout_path: &Path, + state_db_ctx: Option<&StateRuntime>, + state_builder: &mut Option, + default_provider: &str, +) -> std::io::Result<()> { + if let Some(writer) = writer.as_mut() { + for item in items { + writer.write_rollout_item(item).await?; + } + } + if let Some(builder) = state_builder.as_mut() { + builder.rollout_path = rollout_path.to_path_buf(); + } + state_db::apply_rollout_items( + state_db_ctx, + rollout_path, + default_provider, + state_builder.as_ref(), + items, + "rollout_writer", + ) + .await; + Ok(()) +} + struct JsonlWriter { file: tokio::fs::File, } @@ -751,3 +888,97 @@ fn cwd_matches(session_cwd: &Path, cwd: &Path) -> bool { } session_cwd == cwd } + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::ConfigBuilder; + use codex_protocol::protocol::AgentMessageEvent; + use codex_protocol::protocol::EventMsg; + use codex_protocol::protocol::UserMessageEvent; + use tempfile::TempDir; + + #[tokio::test] + 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()) + .build() + .await?; + let thread_id = ThreadId::new(); + let recorder = RolloutRecorder::new( + &config, + RolloutRecorderParams::new( + thread_id, + None, + SessionSource::Exec, + BaseInstructions::default(), + Vec::new(), + ), + None, + None, + ) + .await?; + + let rollout_path = recorder.rollout_path().to_path_buf(); + assert!( + !rollout_path.exists(), + "rollout file should not exist before first user message" + ); + + recorder + .record_items(&[RolloutItem::EventMsg(EventMsg::AgentMessage( + AgentMessageEvent { + message: "buffered-event".to_string(), + }, + ))]) + .await?; + recorder.flush().await?; + assert!( + !rollout_path.exists(), + "rollout file should remain deferred before first user message" + ); + + recorder + .record_items(&[RolloutItem::EventMsg(EventMsg::UserMessage( + UserMessageEvent { + message: "first-user-message".to_string(), + images: None, + local_images: Vec::new(), + text_elements: Vec::new(), + }, + ))]) + .await?; + recorder.flush().await?; + assert!( + !rollout_path.exists(), + "user-message-like items should not materialize without explicit persist" + ); + + recorder.persist().await?; + // Second call verifies `persist()` is idempotent after materialization. + recorder.persist().await?; + assert!(rollout_path.exists(), "rollout file should be materialized"); + + let text = std::fs::read_to_string(&rollout_path)?; + assert!( + text.contains("\"type\":\"session_meta\""), + "expected session metadata in rollout" + ); + let buffered_idx = text + .find("buffered-event") + .expect("buffered event in rollout"); + let user_idx = text + .find("first-user-message") + .expect("first user message in rollout"); + assert!( + buffered_idx < user_idx, + "buffered items should preserve ordering" + ); + let text_after_second_persist = std::fs::read_to_string(&rollout_path)?; + assert_eq!(text_after_second_persist, text); + + recorder.shutdown().await?; + Ok(()) + } +} diff --git a/codex-rs/core/src/tasks/review.rs b/codex-rs/core/src/tasks/review.rs index e2b11dd54..daa394df9 100644 --- a/codex-rs/core/src/tasks/review.rs +++ b/codex-rs/core/src/tasks/review.rs @@ -240,6 +240,7 @@ pub(crate) async fn exit_review_mode( }], ) .await; + session .send_event( ctx.as_ref(), @@ -260,4 +261,9 @@ pub(crate) async fn exit_review_mode( }, ) .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/src/tools/handlers/collab.rs b/codex-rs/core/src/tools/handlers/collab.rs index bfe213d11..833c3d355 100644 --- a/codex-rs/core/src/tools/handlers/collab.rs +++ b/codex-rs/core/src/tools/handlers/collab.rs @@ -797,6 +797,7 @@ fn build_agent_shared_config( #[cfg(test)] mod tests { use super::*; + use crate::AuthManager; use crate::CodexAuth; use crate::ThreadManager; use crate::agent::MAX_THREAD_SPAWN_DEPTH; @@ -811,6 +812,10 @@ mod tests { use crate::protocol::SubAgentSource; use crate::turn_diff_tracker::TurnDiffTracker; use codex_protocol::ThreadId; + use codex_protocol::models::ContentItem; + use codex_protocol::models::ResponseItem; + use codex_protocol::protocol::InitialHistory; + use codex_protocol::protocol::RolloutItem; use pretty_assertions::assert_eq; use serde::Deserialize; use serde_json::json; @@ -1142,7 +1147,22 @@ mod tests { let manager = thread_manager(); session.services.agent_control = manager.agent_control(); let config = turn.config.as_ref().clone(); - let thread = manager.start_thread(config).await.expect("start thread"); + let thread = manager + .resume_thread_with_history( + config, + InitialHistory::Forked(vec![RolloutItem::ResponseItem(ResponseItem::Message { + id: None, + role: "user".to_string(), + content: vec![ContentItem::InputText { + text: "materialized".to_string(), + }], + end_turn: None, + phase: None, + })]), + AuthManager::from_auth_for_testing(CodexAuth::from_api_key("dummy")), + ) + .await + .expect("start thread"); let agent_id = thread.thread_id; let _ = manager .agent_control() diff --git a/codex-rs/core/tests/suite/memory_tool.rs b/codex-rs/core/tests/suite/memory_tool.rs index 09d1ee3ce..7fbd15db4 100644 --- a/codex-rs/core/tests/suite/memory_tool.rs +++ b/codex-rs/core/tests/suite/memory_tool.rs @@ -2,7 +2,12 @@ use anyhow::Result; use codex_core::features::Feature; +use core_test_support::responses::ev_assistant_message; +use core_test_support::responses::ev_completed; +use core_test_support::responses::ev_response_created; use core_test_support::responses::mount_function_call_agent_response; +use core_test_support::responses::mount_sse_once; +use core_test_support::responses::sse; use core_test_support::responses::start_mock_server; use core_test_support::skip_if_no_network; use core_test_support::test_codex::test_codex; @@ -26,6 +31,18 @@ async fn get_memory_tool_returns_persisted_thread_memory() -> Result<()> { let thread_id = test.session_configured.session_id; let thread_id_string = thread_id.to_string(); + mount_sse_once( + &server, + sse(vec![ + ev_response_created("resp-init"), + ev_assistant_message("msg-init", "Materialized"), + ev_completed("resp-init"), + ]), + ) + .await; + test.submit_turn("materialize thread before memory write") + .await?; + let mut thread_exists = false; // Wait for DB creation. for _ in 0..100 { diff --git a/codex-rs/core/tests/suite/rollout_list_find.rs b/codex-rs/core/tests/suite/rollout_list_find.rs index 236a70279..059ca929e 100644 --- a/codex-rs/core/tests/suite/rollout_list_find.rs +++ b/codex-rs/core/tests/suite/rollout_list_find.rs @@ -176,6 +176,7 @@ async fn find_locates_rollout_file_written_by_recorder() -> std::io::Result<()> None, ) .await?; + recorder.persist().await?; recorder.flush().await?; let index_path = home.path().join("session_index.jsonl"); diff --git a/codex-rs/core/tests/suite/sqlite_state.rs b/codex-rs/core/tests/suite/sqlite_state.rs index 688b034ac..b37683a43 100644 --- a/codex-rs/core/tests/suite/sqlite_state.rs +++ b/codex-rs/core/tests/suite/sqlite_state.rs @@ -43,6 +43,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 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 { @@ -56,6 +68,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/core/tests/suite/user_shell_cmd.rs b/codex-rs/core/tests/suite/user_shell_cmd.rs index 88a5caba4..a021509ee 100644 --- a/codex-rs/core/tests/suite/user_shell_cmd.rs +++ b/codex-rs/core/tests/suite/user_shell_cmd.rs @@ -23,6 +23,7 @@ use core_test_support::skip_if_no_network; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; use core_test_support::wait_for_event_match; +use core_test_support::wait_for_event_with_timeout; use regex_lite::escape; use std::path::PathBuf; use tempfile::TempDir; @@ -99,11 +100,11 @@ async fn user_shell_cmd_can_be_interrupted() { // Set up isolated config and conversation. let server = start_mock_server().await; let mut builder = test_codex(); - let codex = builder + let fixture = builder .build(&server) .await - .expect("create new conversation") - .codex; + .expect("create new conversation"); + let codex = &fixture.codex; // Start a long-running command and then interrupt it. let sleep_cmd = "sleep 5".to_string(); @@ -113,11 +114,22 @@ async fn user_shell_cmd_can_be_interrupted() { .unwrap(); // Wait until it has started (ExecCommandBegin), then interrupt. - let _ = wait_for_event(&codex, |ev| matches!(ev, EventMsg::ExecCommandBegin(_))).await; + let _begin = wait_for_event_match(codex, |ev| match ev { + EventMsg::ExecCommandBegin(event) if event.source == ExecCommandSource::UserShell => { + Some(event.clone()) + } + _ => None, + }) + .await; codex.submit(Op::Interrupt).await.unwrap(); // Expect a TurnAborted(Interrupted) notification. - let msg = wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnAborted(_))).await; + let msg = wait_for_event_with_timeout( + codex, + |ev| matches!(ev, EventMsg::TurnAborted(_)), + Duration::from_secs(60), + ) + .await; let EventMsg::TurnAborted(ev) = msg else { unreachable!() }; diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index ea41749e9..4d40c0175 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -1445,46 +1445,57 @@ impl App { self.chat_widget .add_plain_history_lines(vec!["/fork".magenta().into()]); if let Some(path) = self.chat_widget.rollout_path() { - match self - .server - .fork_thread(usize::MAX, self.config.clone(), path.clone()) - .await - { - Ok(forked) => { - self.shutdown_current_thread().await; - let init = self.chatwidget_init_for_forked_or_resumed_thread( - tui, - self.config.clone(), - ); - self.chat_widget = ChatWidget::new_from_existing( - init, - forked.thread, - forked.session_configured, - ); - self.reset_thread_event_state(); - if let Some(summary) = summary { - let mut lines: Vec> = - vec![summary.usage_line.clone().into()]; - if let Some(command) = summary.resume_command { - let spans = vec![ - "To continue this session, run ".into(), - command.cyan(), - ]; - lines.push(spans.into()); + // Fresh threads expose a precomputed path, but the file is + // materialized lazily on first user message. + if path.exists() { + match self + .server + .fork_thread(usize::MAX, self.config.clone(), path.clone()) + .await + { + Ok(forked) => { + self.shutdown_current_thread().await; + let init = self.chatwidget_init_for_forked_or_resumed_thread( + tui, + self.config.clone(), + ); + self.chat_widget = ChatWidget::new_from_existing( + init, + forked.thread, + forked.session_configured, + ); + self.reset_thread_event_state(); + if let Some(summary) = summary { + let mut lines: Vec> = + vec![summary.usage_line.clone().into()]; + if let Some(command) = summary.resume_command { + let spans = vec![ + "To continue this session, run ".into(), + command.cyan(), + ]; + lines.push(spans.into()); + } + self.chat_widget.add_plain_history_lines(lines); } - self.chat_widget.add_plain_history_lines(lines); + } + Err(err) => { + let path_display = path.display(); + self.chat_widget.add_error_message(format!( + "Failed to fork current session from {path_display}: {err}" + )); } } - Err(err) => { - let path_display = path.display(); - self.chat_widget.add_error_message(format!( - "Failed to fork current session from {path_display}: {err}" - )); - } + } else { + self.chat_widget.add_error_message( + "A thread must contain at least one turn before it can be forked." + .to_string(), + ); } } else { - self.chat_widget - .add_error_message("Current session is not ready to fork yet.".to_string()); + self.chat_widget.add_error_message( + "A thread must contain at least one turn before it can be forked." + .to_string(), + ); } tui.frame_requester().schedule_frame(); diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index f94e279bb..9746a0974 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -6845,6 +6845,12 @@ impl ChatWidget { pub(crate) fn thread_name(&self) -> Option { self.thread_name.clone() } + + /// Returns the current thread's precomputed rollout path. + /// + /// For fresh non-ephemeral threads this path may exist before the file is + /// materialized; rollout persistence is deferred until the first user + /// message is recorded. pub(crate) fn rollout_path(&self) -> Option { self.current_rollout_path.clone() }