From 61a0323a32c18173a189f7d3c0fd8d41dfcc5e1f Mon Sep 17 00:00:00 2001 From: Charles Cunningham Date: Fri, 6 Mar 2026 13:11:24 -0800 Subject: [PATCH] Introduce rollout source groundwork Co-authored-by: Codex --- .../app-server/src/bespoke_event_handling.rs | 2 +- .../app-server/src/codex_message_processor.rs | 79 +-- .../tests/suite/v2/thread_archive.rs | 75 +- .../app-server/tests/suite/v2/thread_fork.rs | 24 +- .../app-server/tests/suite/v2/thread_read.rs | 32 +- .../tests/suite/v2/thread_resume.rs | 23 +- .../app-server/tests/suite/v2/thread_start.rs | 4 +- codex-rs/core/src/agent/control.rs | 29 +- codex-rs/core/src/codex.rs | 211 +++--- .../core/src/codex/rollout_reconstruction.rs | 23 +- .../src/codex/rollout_reconstruction_tests.rs | 43 +- codex-rs/core/src/codex_tests.rs | 219 +++++- codex-rs/core/src/codex_thread.rs | 1 + codex-rs/core/src/lib.rs | 4 +- codex-rs/core/src/memories/phase1.rs | 21 +- codex-rs/core/src/rollout/metadata.rs | 72 +- codex-rs/core/src/rollout/mod.rs | 6 +- codex-rs/core/src/rollout/recorder.rs | 642 +++++++++++------- codex-rs/core/src/rollout/tests.rs | 6 +- codex-rs/core/src/state/service.rs | 4 +- codex-rs/core/src/tasks/review.rs | 7 +- codex-rs/core/src/thread_manager.rs | 20 +- codex-rs/core/tests/suite/cli_stream.rs | 2 +- .../core/tests/suite/rollout_list_find.rs | 8 +- codex-rs/core/tests/suite/sqlite_state.rs | 16 +- codex-rs/exec/src/lib.rs | 2 +- codex-rs/protocol/src/protocol.rs | 5 +- codex-rs/tui/src/lib.rs | 6 +- codex-rs/tui/src/resume_picker.rs | 10 +- 29 files changed, 935 insertions(+), 661 deletions(-) diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 85ca56c915..eda1f2fe62 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 persisted rollout".to_string(), + message: "thread has no 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 2955b0da6d..9244f98458 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -182,7 +182,7 @@ use codex_core::CodexAuth; use codex_core::CodexThread; use codex_core::Cursor as RolloutCursor; use codex_core::NewThread; -use codex_core::RolloutRecorder; +use codex_core::RolloutStore; use codex_core::SessionMeta; use codex_core::SteerInputError; use codex_core::ThreadConfigSnapshot; @@ -2456,17 +2456,18 @@ impl CodexMessageProcessor { "ephemeral thread does not support metadata updates: {thread_uuid}" ))); }; - - reconcile_rollout( - Some(state_db_ctx), - rollout_path.as_path(), - self.config.model_provider_id.as_str(), - None, - &[], - None, - None, - ) - .await; + if thread.rollout_path().is_some() { + reconcile_rollout( + Some(state_db_ctx), + rollout_path.as_path(), + self.config.model_provider_id.as_str(), + None, + &[], + None, + None, + ) + .await; + } match state_db_ctx.get_thread(thread_uuid).await { Ok(Some(_)) => return Ok(()), @@ -3101,16 +3102,6 @@ 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, @@ -3398,34 +3389,7 @@ impl CodexMessageProcessor { } let rollout_path = if let Some(path) = existing_thread.rollout_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; - } - } - } + path } else { match find_thread_path_by_id_str( &self.config.codex_home, @@ -3610,7 +3574,11 @@ impl CodexMessageProcessor { } }; - match RolloutRecorder::get_rollout_history(&rollout_path).await { + // App-server should not need direct rollout-file access. + // TODO(ccunningham): make resumed startup stop requiring eager + // `InitialHistory::Resumed(Vec)` materialization so core can own rollout + // loading all the way through thread startup. + match RolloutStore::get_rollout_history(&rollout_path).await { Ok(initial_history) => Some(initial_history), Err(err) => { self.send_invalid_request_error( @@ -4048,7 +4016,7 @@ impl CodexMessageProcessor { while remaining > 0 { let page_size = remaining.min(THREAD_LIST_MAX_LIMIT); let page = if archived { - RolloutRecorder::list_archived_threads( + RolloutStore::list_archived_threads( &self.config, page_size, cursor_obj.as_ref(), @@ -4065,7 +4033,7 @@ impl CodexMessageProcessor { data: None, })? } else { - RolloutRecorder::list_threads( + RolloutStore::list_threads( &self.config, page_size, cursor_obj.as_ref(), @@ -7667,7 +7635,10 @@ pub(crate) async fn read_summary_from_rollout( pub(crate) async fn read_rollout_items_from_rollout( path: &Path, ) -> std::io::Result> { - let items = match RolloutRecorder::get_rollout_history(path).await? { + // This helper exists specifically for callers that still need owned rollout items after + // loading by path. When startup and downstream consumers become `RolloutSource`-backed, they + // should bypass this eager conversion. + let items = match RolloutStore::get_rollout_history(path).await? { InitialHistory::New => Vec::new(), InitialHistory::Forked(items) => items, InitialHistory::Resumed(resumed) => resumed.history, 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 20fd6fd884..391b5db798 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_archive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_archive.rs @@ -2,7 +2,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::JSONRPCError; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ThreadArchiveParams; @@ -28,7 +27,7 @@ use tokio::time::timeout; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); #[tokio::test] -async fn thread_archive_requires_materialized_rollout() -> Result<()> { +async fn thread_archive_works_before_first_user_message() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; @@ -52,72 +51,14 @@ async fn thread_archive_requires_materialized_rollout() -> Result<()> { assert!(!thread.id.is_empty()); 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!(rollout_path.exists(), "fresh thread rollout should exist"); assert!( find_thread_path_by_id_str(codex_home.path(), &thread.id) .await? - .is_none(), - "thread id should not be discoverable before rollout materialization" + .is_some(), + "thread id should be discoverable immediately after startup" ); - // 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(), @@ -129,6 +70,7 @@ async fn thread_archive_requires_materialized_rollout() -> Result<()> { ) .await??; let _: ThreadArchiveResponse = to_response::(archive_resp)?; + let archive_notification = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_notification_message("thread/archived"), @@ -316,10 +258,3 @@ 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 1f19ae8b97..30bee751d0 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -3,7 +3,6 @@ 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; @@ -164,7 +163,7 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> { } #[tokio::test] -async fn thread_fork_rejects_unmaterialized_thread() -> Result<()> { +async fn thread_fork_creates_empty_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())?; @@ -185,25 +184,24 @@ async fn thread_fork_rejects_unmaterialized_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_err: JSONRPCError = timeout( + let fork_resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_error_message(RequestId::Integer(fork_id)), + mcp.read_stream_until_response_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 - ); + 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); 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 dcc411538e..2634bab726 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -3,7 +3,6 @@ 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; @@ -153,7 +152,7 @@ async fn thread_read_can_include_turns() -> Result<()> { } #[tokio::test] -async fn thread_read_loaded_thread_returns_precomputed_path_before_materialization() -> Result<()> { +async fn thread_read_loaded_thread_returns_startup_rollout_path() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; @@ -175,8 +174,8 @@ async fn thread_read_loaded_thread_returns_precomputed_path_before_materializati 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" + thread_path.exists(), + "fresh thread rollout should be readable at startup" ); let read_id = mcp @@ -357,7 +356,7 @@ async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> { } #[tokio::test] -async fn thread_read_include_turns_rejects_unmaterialized_loaded_thread() -> Result<()> { +async fn thread_read_include_turns_returns_empty_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())?; @@ -378,10 +377,7 @@ async fn thread_read_include_turns_rejects_unmaterialized_loaded_thread() -> Res .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" - ); + assert!(thread_path.exists(), "fresh thread rollout should exist"); let read_id = mcp .send_thread_read_request(ThreadReadParams { @@ -389,20 +385,18 @@ async fn thread_read_include_turns_rejects_unmaterialized_loaded_thread() -> Res include_turns: true, }) .await?; - let read_err: JSONRPCError = timeout( + let read_resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_error_message(RequestId::Integer(read_id)), + mcp.read_stream_until_response_message(RequestId::Integer(read_id)), ) .await??; + let ThreadReadResponse { thread: read } = to_response::(read_resp)?; - assert!( - read_err - .error - .message - .contains("includeTurns is unavailable before first user message"), - "unexpected error: {}", - read_err.error.message - ); + 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); 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 12a74ed6e6..3df602d4c9 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_rejects_unmaterialized_thread() -> Result<()> { +async fn thread_resume_returns_empty_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())?; @@ -117,26 +117,25 @@ async fn thread_resume_rejects_unmaterialized_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_err: JSONRPCError = timeout( + let resume_resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_error_message(RequestId::Integer(resume_id)), + mcp.read_stream_until_response_message(RequestId::Integer(resume_id)), ) .await??; - assert!( - resume_err - .error - .message - .contains("no rollout found for thread id"), - "unexpected resume error: {}", - resume_err.error.message - ); + 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); 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 bc383a19d1..03b9647f52 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 not be materialized until first user message" + thread_path.exists(), + "fresh thread rollout should be readable at startup" ); // 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 38f2cf73cb..26100b744d 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -6,7 +6,7 @@ use crate::agent::status::is_final; use crate::error::CodexErr; use crate::error::Result as CodexResult; use crate::find_thread_path_by_id_str; -use crate::rollout::RolloutRecorder; +use crate::rollout::RolloutStore; use crate::session_prefix::format_subagent_context_line; use crate::session_prefix::format_subagent_notification_message; use crate::shell_snapshot::ShellSnapshot; @@ -144,12 +144,7 @@ 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/materialize the live parent before snapshotting JSONL for a fork. - parent_thread - .codex - .session - .ensure_rollout_materialized() - .await; + // Flush the live parent before snapshotting JSONL for a fork. parent_thread.codex.session.flush_rollout().await; } let rollout_path = parent_thread @@ -165,10 +160,12 @@ impl AgentControl { "parent thread rollout unavailable for fork: {parent_thread_id}" )) })?; - let mut forked_rollout_items = - RolloutRecorder::get_rollout_history(&rollout_path) - .await? - .get_rollout_items(); + // TODO(ccunningham): delete this eager owned-history path once fork startup + // becomes `RolloutSource`-backed end-to-end; today agent forking still + // appends a fresh tool output onto the parent rollout items before startup. + let mut forked_rollout_items = RolloutStore::get_rollout_history(&rollout_path) + .await? + .get_rollout_items(); let mut output = FunctionCallOutputPayload::from_text( FORKED_SPAWN_AGENT_OUTPUT_MESSAGE.to_string(), ); @@ -886,11 +883,6 @@ 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 @@ -968,11 +960,6 @@ 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 32a3ee3a32..5b45ee36c8 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -256,11 +256,12 @@ use crate::protocol::TokenUsage; use crate::protocol::TokenUsageInfo; use crate::protocol::TurnDiffEvent; use crate::protocol::WarningEvent; -use crate::rollout::RolloutRecorder; -use crate::rollout::RolloutRecorderParams; +use crate::rollout::RolloutStore; +use crate::rollout::RolloutStoreParams; use crate::rollout::map_session_init_error; use crate::rollout::metadata; use crate::rollout::policy::EventPersistenceMode; +use crate::rollout::recorder::InMemoryRolloutSource; use crate::shell; use crate::shell_snapshot::ShellSnapshot; use crate::skills::SkillError; @@ -1254,7 +1255,7 @@ impl Session { exec_policy: ExecPolicyManager, tx_event: Sender, agent_status: watch::Sender, - initial_history: InitialHistory, + mut initial_history: InitialHistory, session_source: SessionSource, skills_manager: Arc, plugins_manager: Arc, @@ -1275,40 +1276,12 @@ impl Session { } let forked_from_id = initial_history.forked_from_id(); - - let (conversation_id, rollout_params) = match &initial_history { - InitialHistory::New | InitialHistory::Forked(_) => { - let conversation_id = ThreadId::default(); - ( - conversation_id, - RolloutRecorderParams::new( - conversation_id, - forked_from_id, - session_source, - BaseInstructions { - text: session_configuration.base_instructions.clone(), - }, - session_configuration.dynamic_tools.clone(), - if session_configuration.persist_extended_history { - EventPersistenceMode::Extended - } else { - EventPersistenceMode::Limited - }, - ), - ) - } - InitialHistory::Resumed(resumed_history) => ( - resumed_history.conversation_id, - RolloutRecorderParams::resume( - resumed_history.rollout_path.clone(), - if session_configuration.persist_extended_history { - EventPersistenceMode::Extended - } else { - EventPersistenceMode::Limited - }, - ), - ), + let event_persistence_mode = if session_configuration.persist_extended_history { + EventPersistenceMode::Extended + } else { + EventPersistenceMode::Limited }; + let initial_messages = initial_history.get_event_msgs(); let state_builder = match &initial_history { InitialHistory::Resumed(resumed) => metadata::builder_from_items( resumed.history.as_slice(), @@ -1317,24 +1290,62 @@ impl Session { InitialHistory::New | InitialHistory::Forked(_) => None, }; + let (conversation_id, rollout_params) = match &mut initial_history { + InitialHistory::New | InitialHistory::Forked(_) => { + let conversation_id = ThreadId::default(); + ( + conversation_id, + RolloutStoreParams::new( + conversation_id, + forked_from_id, + session_source, + BaseInstructions { + text: session_configuration.base_instructions.clone(), + }, + session_configuration.dynamic_tools.clone(), + event_persistence_mode, + ), + ) + } + InitialHistory::Resumed(resumed_history) => ( + resumed_history.conversation_id, + if config.ephemeral { + RolloutStoreParams::resume( + resumed_history.rollout_path.clone(), + event_persistence_mode, + ) + } else { + let resumed_source = + InMemoryRolloutSource::new(std::mem::take(&mut resumed_history.history)); + RolloutStoreParams::resume_with_source( + resumed_history.rollout_path.clone(), + event_persistence_mode, + resumed_source, + ) + }, + ), + }; + // Kick off independent async setup tasks in parallel to reduce startup latency. // - // - initialize RolloutRecorder with new or resumed session info + // - initialize RolloutStore with new or resumed session info // - perform default shell discovery // - load history metadata (skipped for subagents) let rollout_fut = async { if config.ephemeral { Ok::<_, anyhow::Error>((None, None)) } else { + let state_db_ctx = state_db::init(&config, None).await; + let rollout_store = RolloutStore::new( let state_db_ctx = state_db::init(&config).await; - let rollout_recorder = RolloutRecorder::new( + let rollout_store = RolloutStore::new( &config, rollout_params, state_db_ctx.clone(), state_builder.clone(), ) .await?; - Ok((Some(rollout_recorder), state_db_ctx)) + Ok((Some(rollout_store), state_db_ctx)) } }; @@ -1364,18 +1375,16 @@ impl Session { // Join all independent futures. let ( - rollout_recorder_and_state_db, + rollout_store_and_state_db, (history_log_id, history_entry_count), (auth, mcp_servers, auth_statuses), ) = tokio::join!(rollout_fut, history_meta_fut, auth_and_mcp_fut); - let (rollout_recorder, state_db_ctx) = rollout_recorder_and_state_db.map_err(|e| { - error!("failed to initialize rollout recorder: {e:#}"); + let (rollout_store, state_db_ctx) = rollout_store_and_state_db.map_err(|e| { + error!("failed to initialize rollout store: {e:#}"); e })?; - let rollout_path = rollout_recorder - .as_ref() - .map(|rec| rec.rollout_path.clone()); + let rollout_path = rollout_store.as_ref().map(|rec| rec.rollout_path.clone()); let mut post_session_configured_events = Vec::::new(); @@ -1672,7 +1681,6 @@ impl Session { } // Dispatch the SessionConfiguredEvent first and then report any errors. // If resuming, include converted initial messages in the payload so UIs can render them immediately. - let initial_messages = initial_history.get_event_msgs(); let events = std::iter::once(Event { id: INITIAL_SUBMIT_ID.to_owned(), msg: EventMsg::SessionConfigured(SessionConfiguredEvent { @@ -1808,18 +1816,6 @@ 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 @@ -1942,13 +1938,20 @@ impl Session { } } InitialHistory::Resumed(resumed_history) => { - let rollout_items = resumed_history.history; + let rollout_source = if resumed_history.history.is_empty() { + let rollout = self.services.rollout.lock().await; + match rollout.as_ref() { + Some(rollout) => rollout.source.lock().await.clone(), + None => InMemoryRolloutSource::new(Vec::new()), + } + } else { + InMemoryRolloutSource::new(resumed_history.history) + }; let restored_tool_selection = - Self::extract_mcp_tool_selection_from_rollout(&rollout_items); + Self::extract_mcp_tool_selection_from_rollout_source(&rollout_source); - let reconstructed_rollout = self - .reconstruct_history_from_rollout(&turn_context, &rollout_items) - .await; + let reconstructed_rollout = + self.reconstruct_history_from_rollout(&turn_context, &rollout_source); let previous_turn_settings = reconstructed_rollout.previous_turn_settings.clone(); self.set_previous_turn_settings(previous_turn_settings.clone()) .await; @@ -1986,7 +1989,7 @@ impl Session { // Seed usage info from the recorded rollout so UIs can show token counts // immediately on resume/fork. - if let Some(info) = Self::last_token_info_from_rollout(&rollout_items) { + if let Some(info) = Self::last_token_info_from_rollout_source(&rollout_source) { let mut state = self.state.lock().await; state.set_token_info(Some(info)); } @@ -2001,12 +2004,12 @@ impl Session { } } InitialHistory::Forked(rollout_items) => { + let rollout_source = InMemoryRolloutSource::new(rollout_items.clone()); let restored_tool_selection = - Self::extract_mcp_tool_selection_from_rollout(&rollout_items); + Self::extract_mcp_tool_selection_from_rollout_source(&rollout_source); - let reconstructed_rollout = self - .reconstruct_history_from_rollout(&turn_context, &rollout_items) - .await; + let reconstructed_rollout = + self.reconstruct_history_from_rollout(&turn_context, &rollout_source); self.set_previous_turn_settings( reconstructed_rollout.previous_turn_settings.clone(), ) @@ -2027,7 +2030,7 @@ impl Session { // Seed usage info from the recorded rollout so UIs can show token counts // immediately on resume/fork. - if let Some(info) = Self::last_token_info_from_rollout(&rollout_items) { + if let Some(info) = Self::last_token_info_from_rollout_source(&rollout_source) { let mut state = self.state.lock().await; state.set_token_info(Some(info)); } @@ -2049,9 +2052,6 @@ 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; @@ -2060,20 +2060,26 @@ impl Session { } } - fn last_token_info_from_rollout(rollout_items: &[RolloutItem]) -> Option { - rollout_items.iter().rev().find_map(|item| match item { - RolloutItem::EventMsg(EventMsg::TokenCount(ev)) => ev.info.clone(), - _ => None, - }) + fn last_token_info_from_rollout_source( + rollout_source: &InMemoryRolloutSource, + ) -> Option { + rollout_source + .iter_reverse_from(rollout_source.exclusive_end_of_rollout_index()) + .find_map(|(_, item)| match item { + RolloutItem::EventMsg(EventMsg::TokenCount(ev)) => ev.info.clone(), + _ => None, + }) } - fn extract_mcp_tool_selection_from_rollout( - rollout_items: &[RolloutItem], + fn extract_mcp_tool_selection_from_rollout_source( + rollout_source: &InMemoryRolloutSource, ) -> Option> { let mut search_call_ids = HashSet::new(); let mut active_selected_tools: Option> = None; - for item in rollout_items { + for (_, item) in + rollout_source.iter_forward_from(rollout_source.inclusive_start_of_rollout_index()) + { let RolloutItem::ResponseItem(response_item) = item else { continue; }; @@ -3667,7 +3673,6 @@ 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( @@ -4275,7 +4280,7 @@ mod handlers { use crate::mcp::auth::compute_auth_statuses; use crate::mcp::collect_mcp_snapshot_from_manager; use crate::review_prompts::resolve_review_request; - use crate::rollout::RolloutRecorder; + use crate::rollout::RolloutStore; use crate::rollout::session_index; use crate::tasks::CompactTask; use crate::tasks::UndoTask; @@ -4925,24 +4930,24 @@ mod handlers { return; } - let initial_history = - match RolloutRecorder::get_rollout_history(rollout_path.as_path()).await { - Ok(history) => history, - Err(err) => { - sess.send_event_raw(Event { - id: turn_context.sub_id.clone(), - msg: EventMsg::Error(ErrorEvent { - message: format!( - "failed to load rollout `{}` for rollback replay: {err}", - rollout_path.display() - ), - codex_error_info: Some(CodexErrorInfo::ThreadRollbackFailed), - }), - }) - .await; - return; - } - }; + let initial_history = match RolloutStore::get_rollout_history(rollout_path.as_path()).await + { + Ok(history) => history, + Err(err) => { + sess.send_event_raw(Event { + id: turn_context.sub_id.clone(), + msg: EventMsg::Error(ErrorEvent { + message: format!( + "failed to load rollout `{}` for rollback replay: {err}", + rollout_path.display() + ), + codex_error_info: Some(CodexErrorInfo::ThreadRollbackFailed), + }), + }) + .await; + return; + } + }; let rollback_event = ThreadRolledBackEvent { num_turns }; let replay_items = initial_history @@ -4952,10 +4957,10 @@ mod handlers { EventMsg::ThreadRolledBack(rollback_event.clone()), ))) .collect::>(); + let replay_source = crate::rollout::recorder::InMemoryRolloutSource::new(replay_items); - let reconstructed = sess - .reconstruct_history_from_rollout(turn_context.as_ref(), replay_items.as_slice()) - .await; + let reconstructed = + sess.reconstruct_history_from_rollout(turn_context.as_ref(), &replay_source); sess.replace_history( reconstructed.history, reconstructed.reference_context_item.clone(), diff --git a/codex-rs/core/src/codex/rollout_reconstruction.rs b/codex-rs/core/src/codex/rollout_reconstruction.rs index 11ddbc1928..d55c4fe8ce 100644 --- a/codex-rs/core/src/codex/rollout_reconstruction.rs +++ b/codex-rs/core/src/codex/rollout_reconstruction.rs @@ -1,4 +1,5 @@ use super::*; +use crate::rollout::recorder::InMemoryRolloutSource; // Return value of `Session::reconstruct_history_from_rollout`, bundling the rebuilt history with // the resume/fork hydration metadata derived from the same replay. @@ -83,10 +84,10 @@ fn finalize_active_segment<'a>( } impl Session { - pub(super) async fn reconstruct_history_from_rollout( + pub(super) fn reconstruct_history_from_rollout( &self, turn_context: &TurnContext, - rollout_items: &[RolloutItem], + source: &InMemoryRolloutSource, ) -> RolloutReconstruction { // Replay metadata should already match the shape of the future lazy reverse loader, even // while history materialization still uses an eager bridge. Scan newest-to-oldest, @@ -99,14 +100,15 @@ impl Session { // Rollback is "drop the newest N user turns". While scanning in reverse, that becomes // "skip the next N user-turn segments we finalize". let mut pending_rollback_turns = 0usize; - // Borrowed suffix of rollout items newer than the newest surviving replacement-history - // checkpoint. If no such checkpoint exists, this remains the full rollout. - let mut rollout_suffix = rollout_items; + // Default to replaying from the beginning of the rollout. If reverse replay later finds a + // surviving compaction with `replacement_history`, it advances this start index to the + // first rollout row after that compaction checkpoint. + let mut rollout_suffix_start = source.inclusive_start_of_rollout_index(); // Reverse replay accumulates rollout items into the newest in-progress turn segment until // we hit its matching `TurnStarted`, at which point the segment can be finalized. let mut active_segment: Option> = None; - for (index, item) in rollout_items.iter().enumerate().rev() { + for (index, item) in source.iter_reverse_from(source.exclusive_end_of_rollout_index()) { match item { RolloutItem::Compacted(compacted) => { let active_segment = @@ -123,7 +125,7 @@ impl Session { && let Some(replacement_history) = &compacted.replacement_history { active_segment.base_replacement_history = Some(replacement_history); - rollout_suffix = &rollout_items[index + 1..]; + rollout_suffix_start = index.next_newer(); } } RolloutItem::EventMsg(EventMsg::ThreadRolledBack(rollback)) => { @@ -235,7 +237,7 @@ impl Session { // Materialize exact history semantics from the replay-derived suffix. The eventual lazy // design should keep this same replay shape, but drive it from a resumable reverse source // instead of an eagerly loaded `&[RolloutItem]`. - for item in rollout_suffix { + for (_, item) in source.iter_forward_from(rollout_suffix_start) { match item { RolloutItem::ResponseItem(response_item) => { history.record_items( @@ -256,8 +258,9 @@ impl Session { // `reference_context_item`, reinject canonical context at the end of the // resumed conversation, and accept the temporary out-of-distribution // prompt shape. - // TODO(ccunningham): if we drop support for None replacement_history compaction items, - // we can get rid of this second loop entirely and just build `history` directly in the first loop. + // If we eventually drop support for legacy compaction items without + // `replacement_history`, this second pass can go away and the first replay + // loop can materialize history directly. let user_messages = collect_user_messages(history.raw_items()); let rebuilt = compact::build_compacted_history( Vec::new(), diff --git a/codex-rs/core/src/codex/rollout_reconstruction_tests.rs b/codex-rs/core/src/codex/rollout_reconstruction_tests.rs index 6cc99a2907..e81e930f5f 100644 --- a/codex-rs/core/src/codex/rollout_reconstruction_tests.rs +++ b/codex-rs/core/src/codex/rollout_reconstruction_tests.rs @@ -3,6 +3,7 @@ use super::*; use crate::protocol::CompactedItem; use crate::protocol::InitialHistory; use crate::protocol::ResumedHistory; +use crate::rollout::recorder::InMemoryRolloutSource; use codex_protocol::ThreadId; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; @@ -219,9 +220,10 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_com )), ]; - let reconstructed = session - .reconstruct_history_from_rollout(&turn_context, &rollout_items) - .await; + let reconstructed = session.reconstruct_history_from_rollout( + &turn_context, + &InMemoryRolloutSource::new(rollout_items), + ); assert_eq!( reconstructed.history, @@ -301,9 +303,10 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_inc )), ]; - let reconstructed = session - .reconstruct_history_from_rollout(&turn_context, &rollout_items) - .await; + let reconstructed = session.reconstruct_history_from_rollout( + &turn_context, + &InMemoryRolloutSource::new(rollout_items), + ); assert_eq!( reconstructed.history, @@ -407,9 +410,10 @@ async fn reconstruct_history_rollback_skips_non_user_turns_for_history_and_metad )), ]; - let reconstructed = session - .reconstruct_history_from_rollout(&turn_context, &rollout_items) - .await; + let reconstructed = session.reconstruct_history_from_rollout( + &turn_context, + &InMemoryRolloutSource::new(rollout_items), + ); assert_eq!( reconstructed.history, @@ -468,9 +472,10 @@ async fn reconstruct_history_rollback_clears_history_and_metadata_when_exceeding )), ]; - let reconstructed = session - .reconstruct_history_from_rollout(&turn_context, &rollout_items) - .await; + let reconstructed = session.reconstruct_history_from_rollout( + &turn_context, + &InMemoryRolloutSource::new(rollout_items), + ); assert_eq!(reconstructed.history, Vec::new()); assert_eq!(reconstructed.previous_turn_settings, None); @@ -674,9 +679,10 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_does_ }), ]; - let reconstructed = session - .reconstruct_history_from_rollout(&turn_context, &rollout_items) - .await; + let reconstructed = session.reconstruct_history_from_rollout( + &turn_context, + &InMemoryRolloutSource::new(rollout_items), + ); assert_eq!( reconstructed.history, @@ -727,9 +733,10 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_clear )), ]; - let reconstructed = session - .reconstruct_history_from_rollout(&turn_context, &rollout_items) - .await; + let reconstructed = session.reconstruct_history_from_rollout( + &turn_context, + &InMemoryRolloutSource::new(rollout_items), + ); assert!(reconstructed.reference_context_item.is_none()); } diff --git a/codex-rs/core/src/codex_tests.rs b/codex-rs/core/src/codex_tests.rs index 7a17bdd98d..4f3b82da6b 100644 --- a/codex-rs/core/src/codex_tests.rs +++ b/codex-rs/core/src/codex_tests.rs @@ -40,8 +40,9 @@ use crate::protocol::TokenUsageInfo; use crate::protocol::TurnCompleteEvent; use crate::protocol::UserMessageEvent; use crate::rollout::policy::EventPersistenceMode; -use crate::rollout::recorder::RolloutRecorder; -use crate::rollout::recorder::RolloutRecorderParams; +use crate::rollout::recorder::InMemoryRolloutSource; +use crate::rollout::recorder::RolloutStore; +use crate::rollout::recorder::RolloutStoreParams; use crate::state::TaskKind; use crate::tasks::SessionTask; use crate::tasks::SessionTaskContext; @@ -746,11 +747,11 @@ fn extract_mcp_tool_selection_from_rollout_returns_none_without_valid_search_out async fn reconstruct_history_matches_live_compactions() { let (session, turn_context) = make_session_and_context().await; let (rollout_items, expected) = sample_rollout(&session, &turn_context).await; + let rollout_source = InMemoryRolloutSource::new(rollout_items); let reconstruction_turn = session.new_default_turn().await; - let reconstructed = session - .reconstruct_history_from_rollout(reconstruction_turn.as_ref(), &rollout_items) - .await; + let reconstructed = + session.reconstruct_history_from_rollout(reconstruction_turn.as_ref(), &rollout_source); assert_eq!(expected, reconstructed.history); } @@ -768,7 +769,7 @@ async fn reconstruct_history_uses_replacement_history_verbatim() { phase: None, }; let replacement_history = vec![ - summary_item.clone(), + summary_item, ResponseItem::Message { id: None, role: "developer".to_string(), @@ -783,10 +784,9 @@ async fn reconstruct_history_uses_replacement_history_verbatim() { message: String::new(), replacement_history: Some(replacement_history.clone()), })]; + let rollout_source = InMemoryRolloutSource::new(rollout_items); - let reconstructed = session - .reconstruct_history_from_rollout(&turn_context, &rollout_items) - .await; + let reconstructed = session.reconstruct_history_from_rollout(&turn_context, &rollout_source); assert_eq!(reconstructed.history, replacement_history); } @@ -1114,7 +1114,7 @@ async fn thread_rollback_drops_last_turn_from_history() { assert_eq!(sess.previous_turn_settings().await, None); assert!(sess.reference_context_item().await.is_none()); - let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path) + let InitialHistory::Resumed(resumed) = RolloutStore::get_rollout_history(&rollout_path) .await .expect("read rollout history") else { @@ -1366,7 +1366,7 @@ async fn thread_rollback_persists_marker_and_replays_cumulatively() { ] ); - let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path) + let InitialHistory::Resumed(resumed) = RolloutStore::get_rollout_history(&rollout_path) .await .expect("read rollout history") else { @@ -1800,9 +1800,9 @@ async fn wait_for_thread_rollback_failed(rx: &async_channel::Receiver) -> async fn attach_rollout_recorder(session: &Arc) -> PathBuf { let config = session.get_config().await; - let recorder = RolloutRecorder::new( + let recorder = RolloutStore::new( config.as_ref(), - RolloutRecorderParams::new( + RolloutStoreParams::new( ThreadId::default(), None, SessionSource::Exec, @@ -1820,7 +1820,6 @@ 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 } @@ -3367,9 +3366,9 @@ async fn record_context_updates_and_set_reference_context_item_persists_baseline state.set_reference_context_item(Some(previous_context_item.clone())); } let config = session.get_config().await; - let recorder = RolloutRecorder::new( + let recorder = RolloutStore::new( config.as_ref(), - RolloutRecorderParams::new( + RolloutStoreParams::new( ThreadId::default(), None, SessionSource::Exec, @@ -3407,10 +3406,9 @@ 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) = RolloutRecorder::get_rollout_history(&rollout_path) + let InitialHistory::Resumed(resumed) = RolloutStore::get_rollout_history(&rollout_path) .await .expect("read rollout history") else { @@ -3464,9 +3462,9 @@ async fn record_context_updates_and_set_reference_context_item_persists_full_rei .with_model(next_model.to_string(), &session.services.models_manager) .await; let config = session.get_config().await; - let recorder = RolloutRecorder::new( + let recorder = RolloutStore::new( config.as_ref(), - RolloutRecorderParams::new( + RolloutStoreParams::new( ThreadId::default(), None, SessionSource::Exec, @@ -3509,10 +3507,9 @@ 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) = RolloutRecorder::get_rollout_history(&rollout_path) + let InitialHistory::Resumed(resumed) = RolloutStore::get_rollout_history(&rollout_path) .await .expect("read rollout history") else { @@ -3531,6 +3528,184 @@ async fn record_context_updates_and_set_reference_context_item_persists_full_rei ); } +#[tokio::test] +async fn session_new_ephemeral_resume_keeps_in_memory_history() { + let (seed_session, turn_context) = make_session_and_context().await; + let (rollout_items, expected) = sample_rollout(&seed_session, &turn_context).await; + + let codex_home = tempfile::tempdir().expect("create temp dir"); + let mut config = build_test_config(codex_home.path()).await; + config.ephemeral = true; + let config = Arc::new(config); + + let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("Test API Key")); + let models_manager = Arc::new(ModelsManager::new( + config.codex_home.clone(), + auth_manager.clone(), + None, + CollaborationModesConfig::default(), + )); + let model = ModelsManager::get_model_offline_for_tests(config.model.as_deref()); + let model_info = ModelsManager::construct_model_info_offline_for_tests(model.as_str(), &config); + let collaboration_mode = CollaborationMode { + mode: ModeKind::Default, + settings: Settings { + model, + reasoning_effort: config.model_reasoning_effort, + developer_instructions: None, + }, + }; + let session_configuration = SessionConfiguration { + provider: config.model_provider.clone(), + collaboration_mode, + model_reasoning_summary: config.model_reasoning_summary, + developer_instructions: config.developer_instructions.clone(), + user_instructions: config.user_instructions.clone(), + personality: config.personality, + base_instructions: config + .base_instructions + .clone() + .unwrap_or_else(|| model_info.get_model_instructions(config.personality)), + compact_prompt: config.compact_prompt.clone(), + approval_policy: config.permissions.approval_policy.clone(), + sandbox_policy: config.permissions.sandbox_policy.clone(), + windows_sandbox_level: WindowsSandboxLevel::from_config(&config), + cwd: config.cwd.clone(), + codex_home: config.codex_home.clone(), + thread_name: None, + original_config_do_not_use: Arc::clone(&config), + service_tier: config.service_tier, + metrics_service_name: None, + app_server_client_name: None, + session_source: SessionSource::Exec, + dynamic_tools: Vec::new(), + persist_extended_history: false, + inherited_shell_snapshot: None, + }; + + let (tx_event, _rx_event) = async_channel::unbounded(); + let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit); + let plugins_manager = Arc::new(PluginsManager::new(config.codex_home.clone())); + let mcp_manager = Arc::new(McpManager::new(Arc::clone(&plugins_manager))); + let skills_manager = Arc::new(SkillsManager::new( + config.codex_home.clone(), + Arc::clone(&plugins_manager), + )); + let session = Session::new( + session_configuration, + Arc::clone(&config), + auth_manager, + models_manager, + ExecPolicyManager::default(), + tx_event, + agent_status_tx, + InitialHistory::Resumed(ResumedHistory { + conversation_id: ThreadId::default(), + history: rollout_items, + rollout_path: PathBuf::from("/tmp/resume.jsonl"), + }), + SessionSource::Exec, + skills_manager, + plugins_manager, + mcp_manager, + Arc::new(FileWatcher::noop()), + AgentControl::default(), + ) + .await + .expect("create ephemeral resumed session"); + + assert_eq!(expected, session.clone_history().await.raw_items()); +} + +#[tokio::test] +async fn session_new_ephemeral_resume_with_empty_history_does_not_panic() { + let codex_home = tempfile::tempdir().expect("create temp dir"); + let mut config = build_test_config(codex_home.path()).await; + config.ephemeral = true; + let config = Arc::new(config); + + let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("Test API Key")); + let models_manager = Arc::new(ModelsManager::new( + config.codex_home.clone(), + auth_manager.clone(), + None, + CollaborationModesConfig::default(), + )); + let model = ModelsManager::get_model_offline_for_tests(config.model.as_deref()); + let model_info = ModelsManager::construct_model_info_offline_for_tests(model.as_str(), &config); + let collaboration_mode = CollaborationMode { + mode: ModeKind::Default, + settings: Settings { + model, + reasoning_effort: config.model_reasoning_effort, + developer_instructions: None, + }, + }; + let session_configuration = SessionConfiguration { + provider: config.model_provider.clone(), + collaboration_mode, + model_reasoning_summary: config.model_reasoning_summary, + developer_instructions: config.developer_instructions.clone(), + user_instructions: config.user_instructions.clone(), + personality: config.personality, + base_instructions: config + .base_instructions + .clone() + .unwrap_or_else(|| model_info.get_model_instructions(config.personality)), + compact_prompt: config.compact_prompt.clone(), + approval_policy: config.permissions.approval_policy.clone(), + sandbox_policy: config.permissions.sandbox_policy.clone(), + windows_sandbox_level: WindowsSandboxLevel::from_config(&config), + cwd: config.cwd.clone(), + codex_home: config.codex_home.clone(), + thread_name: None, + original_config_do_not_use: Arc::clone(&config), + service_tier: config.service_tier, + metrics_service_name: None, + app_server_client_name: None, + session_source: SessionSource::Exec, + dynamic_tools: Vec::new(), + persist_extended_history: false, + inherited_shell_snapshot: None, + }; + + let (tx_event, _rx_event) = async_channel::unbounded(); + let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit); + let plugins_manager = Arc::new(PluginsManager::new(config.codex_home.clone())); + let mcp_manager = Arc::new(McpManager::new(Arc::clone(&plugins_manager))); + let skills_manager = Arc::new(SkillsManager::new( + config.codex_home.clone(), + Arc::clone(&plugins_manager), + )); + let session = Session::new( + session_configuration, + Arc::clone(&config), + auth_manager, + models_manager, + ExecPolicyManager::default(), + tx_event, + agent_status_tx, + InitialHistory::Resumed(ResumedHistory { + conversation_id: ThreadId::default(), + history: Vec::new(), + rollout_path: PathBuf::from("/tmp/resume.jsonl"), + }), + SessionSource::Exec, + skills_manager, + plugins_manager, + mcp_manager, + Arc::new(FileWatcher::noop()), + AgentControl::default(), + ) + .await + .expect("create ephemeral resumed session with empty history"); + + assert_eq!( + Vec::::new(), + session.clone_history().await.raw_items() + ); +} + #[tokio::test] async fn run_user_shell_command_does_not_set_reference_context_item() { let (session, _turn_context, rx) = make_session_and_context_with_rx().await; diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index b33a66c3ed..d625d9691c 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -132,6 +132,7 @@ impl CodexThread { .await; } + /// Return the rollout path for this thread, if any. pub fn rollout_path(&self) -> Option { self.rollout_path.clone() } diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 7e84577ee6..f1b282e930 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -118,8 +118,8 @@ mod turn_metadata; mod turn_timing; pub use rollout::ARCHIVED_SESSIONS_SUBDIR; pub use rollout::INTERACTIVE_SESSION_SOURCES; -pub use rollout::RolloutRecorder; -pub use rollout::RolloutRecorderParams; +pub use rollout::RolloutStore; +pub use rollout::RolloutStoreParams; pub use rollout::SESSIONS_SUBDIR; pub use rollout::SessionMeta; pub use rollout::append_thread_name; diff --git a/codex-rs/core/src/memories/phase1.rs b/codex-rs/core/src/memories/phase1.rs index ad4f29a0d2..46d8095c9c 100644 --- a/codex-rs/core/src/memories/phase1.rs +++ b/codex-rs/core/src/memories/phase1.rs @@ -1,5 +1,5 @@ use crate::Prompt; -use crate::RolloutRecorder; +use crate::RolloutStore; use crate::codex::Session; use crate::codex::TurnContext; use crate::config::Config; @@ -315,8 +315,17 @@ mod job { rollout_cwd: &Path, stage_one_context: &RequestContext, ) -> anyhow::Result<(StageOneOutput, Option)> { - let (rollout_items, _, _) = RolloutRecorder::load_rollout_items(rollout_path).await?; - let rollout_contents = serialize_filtered_rollout_response_items(&rollout_items)?; + let loaded_rollout = RolloutStore::load_source(rollout_path).await?; + let source = loaded_rollout.source; + // TODO(ccunningham): avoid serializing the whole rollout here; once this path has a + // token-budget-aware head/tail collector, it should use `RolloutSource` forward/reverse + // iteration to build only the sections that will survive into the stage-1 sampling + // request. + let rollout_contents = serialize_filtered_rollout_response_items( + source + .iter_forward_from(source.inclusive_start_of_rollout_index()) + .map(|(_, item)| item), + )?; let prompt = Prompt { input: vec![ResponseItem::Message { @@ -463,11 +472,11 @@ mod job { } /// Serializes filtered stage-1 memory items for prompt inclusion. - fn serialize_filtered_rollout_response_items( - items: &[RolloutItem], + fn serialize_filtered_rollout_response_items<'a>( + items: impl IntoIterator, ) -> crate::error::Result { let filtered = items - .iter() + .into_iter() .filter_map(|item| { if let RolloutItem::ResponseItem(item) = item && should_persist_response_item_for_memories(item) diff --git a/codex-rs/core/src/rollout/metadata.rs b/codex-rs/core/src/rollout/metadata.rs index 3b18520ee0..e0a80c6dd0 100644 --- a/codex-rs/core/src/rollout/metadata.rs +++ b/codex-rs/core/src/rollout/metadata.rs @@ -1,7 +1,7 @@ use crate::config::Config; use crate::rollout; +use crate::rollout::RolloutStore; use crate::rollout::list::parse_timestamp_uuid_from_filename; -use crate::rollout::recorder::RolloutRecorder; use crate::state_db::normalize_cwd_for_state_db; use chrono::DateTime; use chrono::NaiveDateTime; @@ -61,11 +61,11 @@ pub(crate) fn builder_from_session_meta( Some(builder) } -pub(crate) fn builder_from_items( - items: &[RolloutItem], +pub(crate) fn builder_from_items<'a>( + items: impl IntoIterator, rollout_path: &Path, ) -> Option { - if let Some(session_meta) = items.iter().find_map(|item| match item { + if let Some(session_meta) = items.into_iter().find_map(|item| match item { RolloutItem::SessionMeta(meta_line) => Some(meta_line), RolloutItem::ResponseItem(_) | RolloutItem::Compacted(_) @@ -96,22 +96,31 @@ pub(crate) async fn extract_metadata_from_rollout( rollout_path: &Path, default_provider: &str, ) -> anyhow::Result { - let (items, _thread_id, parse_errors) = - RolloutRecorder::load_rollout_items(rollout_path).await?; - if items.is_empty() { - return Err(anyhow::anyhow!( - "empty session file: {}", + let loaded_rollout = RolloutStore::load_source(rollout_path).await?; + let parse_errors = loaded_rollout.parse_errors; + let source = loaded_rollout.source; + let rollout_start = source.inclusive_start_of_rollout_index(); + let has_any_parsed_items = source.iter_forward_from(rollout_start).next().is_some(); + if parse_errors > 0 && !has_any_parsed_items { + anyhow::bail!( + "rollout contains parse errors and no readable items: {}", rollout_path.display() - )); + ); } - let builder = builder_from_items(items.as_slice(), rollout_path).ok_or_else(|| { + let builder = builder_from_items( + source + .iter_forward_from(rollout_start) + .map(|(_, item)| item), + rollout_path, + ) + .ok_or_else(|| { anyhow::anyhow!( "rollout missing metadata builder: {}", rollout_path.display() ) })?; let mut metadata = builder.build(default_provider); - for item in &items { + for (_, item) in source.iter_forward_from(rollout_start) { apply_rollout_item(&mut metadata, item, default_provider); } if let Some(updated_at) = file_modified_time_utc(rollout_path).await { @@ -119,13 +128,15 @@ pub(crate) async fn extract_metadata_from_rollout( } Ok(ExtractionOutcome { metadata, - memory_mode: items.iter().rev().find_map(|item| match item { - RolloutItem::SessionMeta(meta_line) => meta_line.meta.memory_mode.clone(), - RolloutItem::ResponseItem(_) - | RolloutItem::Compacted(_) - | RolloutItem::TurnContext(_) - | RolloutItem::EventMsg(_) => None, - }), + memory_mode: source + .iter_reverse_from(source.exclusive_end_of_rollout_index()) + .find_map(|(_, item)| match item { + RolloutItem::SessionMeta(meta_line) => meta_line.meta.memory_mode.clone(), + RolloutItem::ResponseItem(_) + | RolloutItem::Compacted(_) + | RolloutItem::TurnContext(_) + | RolloutItem::EventMsg(_) => None, + }), parse_errors, }) } @@ -573,6 +584,29 @@ mod tests { assert_eq!(outcome.memory_mode.as_deref(), Some("polluted")); } + #[tokio::test] + async fn extract_metadata_from_rollout_rejects_unparseable_rollout_with_no_items() { + let dir = tempdir().expect("tempdir"); + let uuid = Uuid::new_v4(); + let path = dir + .path() + .join(format!("rollout-2026-01-27T12-34-56-{uuid}.jsonl")); + + let mut file = File::create(&path).expect("create rollout"); + writeln!(file, "{{").expect("write invalid rollout"); + + let err = extract_metadata_from_rollout(&path, "openai", None) + .await + .expect_err("unparseable rollout should fail metadata extraction"); + assert_eq!( + err.to_string(), + format!( + "rollout contains parse errors and no readable items: {}", + path.display() + ) + ); + } + #[test] fn builder_from_items_falls_back_to_filename() { let dir = tempdir().expect("tempdir"); diff --git a/codex-rs/core/src/rollout/mod.rs b/codex-rs/core/src/rollout/mod.rs index 31ee26dcaa..293f9b588a 100644 --- a/codex-rs/core/src/rollout/mod.rs +++ b/codex-rs/core/src/rollout/mod.rs @@ -11,7 +11,7 @@ pub(crate) mod error; pub mod list; pub(crate) mod metadata; pub(crate) mod policy; -pub mod recorder; +pub(crate) mod recorder; pub(crate) mod session_index; pub(crate) mod truncation; @@ -22,8 +22,8 @@ pub use list::find_thread_path_by_id_str; #[deprecated(note = "use find_thread_path_by_id_str")] pub use list::find_thread_path_by_id_str as find_conversation_path_by_id_str; pub use list::rollout_date_parts; -pub use recorder::RolloutRecorder; -pub use recorder::RolloutRecorderParams; +pub use recorder::RolloutStore; +pub use recorder::RolloutStoreParams; pub use session_index::append_thread_name; pub use session_index::find_thread_name_by_id; pub use session_index::find_thread_path_by_name_str; diff --git a/codex-rs/core/src/rollout/recorder.rs b/codex-rs/core/src/rollout/recorder.rs index 851dcfb727..a84240a1df 100644 --- a/codex-rs/core/src/rollout/recorder.rs +++ b/codex-rs/core/src/rollout/recorder.rs @@ -5,6 +5,7 @@ use std::fs::{self}; use std::io::Error as IoError; use std::path::Path; use std::path::PathBuf; +use std::sync::Arc; use chrono::SecondsFormat; use chrono::Utc; @@ -17,6 +18,7 @@ use time::format_description::FormatItem; use time::format_description::well_known::Rfc3339; use time::macros::format_description; use tokio::io::AsyncWriteExt; +use tokio::sync::Mutex; use tokio::sync::mpsc::Sender; use tokio::sync::mpsc::{self}; use tokio::sync::oneshot; @@ -41,7 +43,6 @@ use super::policy::EventPersistenceMode; use super::policy::is_persisted_response_item; use crate::config::Config; use crate::default_client::originator; -use crate::git_info::collect_git_info; use crate::path_utils; use crate::state_db; use crate::state_db::StateDbHandle; @@ -58,8 +59,134 @@ 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 -/// every update. +/// Logical position in the rollout stream. +/// +/// This typed wrapper keeps rollout positions distinct from raw slice offsets. +/// +/// `-1` is the newest rollout row that already existed when this source was created. Older +/// persisted rows are more negative, and any rows appended after startup are `0`, `1`, `2`, and +/// so on. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct RolloutIndex(i64); + +impl RolloutIndex { + /// Return the next newer logical rollout position. + pub(crate) fn next_newer(self) -> Self { + Self(self.0 + 1) + } +} + +/// Parsed rollout source plus file-level metadata gathered while loading by path. +pub(crate) struct LoadedRolloutSource { + pub(crate) source: InMemoryRolloutSource, + pub(crate) thread_id: Option, + pub(crate) parse_errors: usize, +} + +/// In-memory `RolloutSource` for the current startup/replay paths that still materialize the +/// entire rollout before reconstructing history or extracting metadata. +/// +/// The long-term plan is to make this source lazy over the rollout file so startup and replay can +/// support unbounded conversation history without loading the entire rollout into memory up front. +/// When that happens, the equivalent source should keep the same "load older items on demand" +/// contract, but page older rollout rows from disk instead of cloning them out of a fully loaded +/// `Vec`. +#[derive(Clone, Debug)] +pub struct InMemoryRolloutSource { + rollout_items: Vec, + startup_rollout_len: i64, +} + +impl InMemoryRolloutSource { + /// Build an in-memory `RolloutSource` from a fully loaded rollout item stream. + pub(crate) fn new(rollout_items: Vec) -> Self { + let startup_rollout_len = i64::try_from(rollout_items.len()).unwrap_or(i64::MAX); + Self { + rollout_items, + startup_rollout_len, + } + } + + /// Load a persisted rollout file into the eager in-memory `RolloutSource` used today. + pub(crate) async fn load_from_path(path: &Path) -> std::io::Result { + let (items, thread_id, parse_errors) = RolloutStore::load_rollout_items(path).await?; + Ok(LoadedRolloutSource { + source: Self::new(items), + thread_id, + parse_errors, + }) + } + + /// Append newly recorded rollout items after startup. + pub(crate) fn append_items(&mut self, items: Vec) { + self.rollout_items.extend(items); + } + + /// Consume the source back into owned rollout items. + pub(crate) fn into_items(self) -> Vec { + self.rollout_items + } + + /// Return the inclusive index of the start of the persisted rollout file. + pub(crate) fn inclusive_start_of_rollout_index(&self) -> RolloutIndex { + RolloutIndex(-self.startup_rollout_len) + } + + /// Return the exclusive end index of the persisted rollout file plus any items + /// appended in memory after startup. + pub(crate) fn exclusive_end_of_rollout_index(&self) -> RolloutIndex { + let rollout_len = i64::try_from(self.rollout_items.len()).unwrap_or(i64::MAX); + RolloutIndex(rollout_len.saturating_sub(self.startup_rollout_len)) + } + + /// Iterate forward from the inclusive `start` position. + pub(crate) fn iter_forward_from( + &self, + start: RolloutIndex, + ) -> impl Iterator + '_ { + let start = self.loaded_offset_from_rollout_index(start); + self.rollout_items[start..] + .iter() + .enumerate() + .map(move |(offset, item)| { + let offset = i64::try_from(offset).unwrap_or(i64::MAX); + ( + RolloutIndex(start as i64 + offset - self.startup_rollout_len), + item, + ) + }) + } + + /// Iterate backward over rows strictly older than the exclusive `end` position. + pub(crate) fn iter_reverse_from( + &self, + end: RolloutIndex, + ) -> impl Iterator + '_ { + let end = self.loaded_offset_from_rollout_index(end); + self.rollout_items[..end] + .iter() + .enumerate() + .rev() + .map(move |(actual_index, item)| { + let actual_index = i64::try_from(actual_index).unwrap_or(i64::MAX); + (RolloutIndex(actual_index - self.startup_rollout_len), item) + }) + } + + fn loaded_offset_from_rollout_index(&self, rollout_index: RolloutIndex) -> usize { + let actual_index = rollout_index.0.saturating_add(self.startup_rollout_len); + if actual_index <= 0 { + return 0; + } + + match usize::try_from(actual_index) { + Ok(actual_index) => actual_index.min(self.rollout_items.len()), + Err(_) => self.rollout_items.len(), + } + } +} + +/// Manages the canonical rollout state for the current process and persists it to disk. /// /// Rollouts are recorded as JSONL and can be inspected with tools such as: /// @@ -68,15 +195,20 @@ use codex_state::ThreadMetadataBuilder; /// $ fx ~/.codex/sessions/rollout-2025-05-07T17-24-21-5973b6c0-94b8-487b-a530-2aeb6098ae0e.jsonl /// ``` #[derive(Clone)] -pub struct RolloutRecorder { +pub struct RolloutStore { tx: Sender, pub(crate) rollout_path: PathBuf, state_db: Option, event_persistence_mode: EventPersistenceMode, + // Canonical in-memory `RolloutSource` for this process. Startup loading and runtime writes + // both update this source so replay sees the same rollout items and session metadata that the + // writer is responsible for persisting. + pub(crate) source: Arc>, } +/// Parameters for creating or resuming a rollout store. #[derive(Clone)] -pub enum RolloutRecorderParams { +pub enum RolloutStoreParams { Create { conversation_id: ThreadId, forked_from_id: Option, @@ -88,6 +220,7 @@ pub enum RolloutRecorderParams { Resume { path: PathBuf, event_persistence_mode: EventPersistenceMode, + source: Option, }, } @@ -105,7 +238,8 @@ enum RolloutCmd { }, } -impl RolloutRecorderParams { +impl RolloutStoreParams { + /// Create rollout parameters for a brand-new session. pub fn new( conversation_id: ThreadId, forked_from_id: Option, @@ -124,10 +258,26 @@ impl RolloutRecorderParams { } } + /// Create rollout parameters for resuming an existing persisted rollout. pub fn resume(path: PathBuf, event_persistence_mode: EventPersistenceMode) -> Self { Self::Resume { path, event_persistence_mode, + source: None, + } + } + + /// Create rollout parameters for resume when startup already has a preloaded + /// in-memory rollout source and should avoid reading the rollout file again. + pub fn resume_with_source( + path: PathBuf, + event_persistence_mode: EventPersistenceMode, + source: InMemoryRolloutSource, + ) -> Self { + Self::Resume { + path, + event_persistence_mode, + source: Some(source), } } } @@ -159,7 +309,12 @@ fn sanitize_rollout_item_for_persistence( } } -impl RolloutRecorder { +impl RolloutStore { + /// Load a persisted rollout file into the eager in-memory `RolloutSource` used today. + pub(crate) async fn load_source(path: &Path) -> std::io::Result { + InMemoryRolloutSource::load_from_path(path).await + } + /// List threads (rollout files) under the provided Codex home directory. #[allow(clippy::too_many_arguments)] pub async fn list_threads( @@ -361,108 +516,124 @@ impl RolloutRecorder { } } - /// Attempt to create a new [`RolloutRecorder`]. + /// Attempt to create a new [`RolloutStore`]. /// - /// 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. + /// For both new and resumed sessions, this opens the rollout file immediately so the path is + /// readable as soon as startup returns. pub async fn new( config: &Config, - params: RolloutRecorderParams, + params: RolloutStoreParams, state_db_ctx: Option, state_builder: Option, ) -> std::io::Result { - let (file, deferred_log_file_info, rollout_path, meta, event_persistence_mode) = - match params { - RolloutRecorderParams::Create { - conversation_id, + let (file, rollout_path, source, meta, event_persistence_mode) = match params { + RolloutStoreParams::Create { + conversation_id, + forked_from_id, + source, + base_instructions, + dynamic_tools, + event_persistence_mode, + } => { + 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 = started_at + .to_offset(time::UtcOffset::UTC) + .format(timestamp_format) + .map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?; + + 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(), + agent_nickname: source.get_nickname(), + agent_role: source.get_agent_role(), source, - base_instructions, - dynamic_tools, - event_persistence_mode, - } => { - 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; + 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) + }, + memory_mode: (!config.memories.generate_memories) + .then_some("disabled".to_string()), + }; + let session_meta_line = SessionMetaLine { + meta: session_meta, + git: None, + }; - let timestamp_format: &[FormatItem] = format_description!( - "[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z" - ); - let timestamp = started_at - .to_offset(time::UtcOffset::UTC) - .format(timestamp_format) - .map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?; - - 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(), - agent_nickname: source.get_nickname(), - agent_role: source.get_agent_role(), - 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) - }, - memory_mode: (!config.memories.generate_memories) - .then_some("disabled".to_string()), - }; - - ( - None, - Some(log_file_info), - path, - Some(session_meta), - event_persistence_mode, - ) - } - RolloutRecorderParams::Resume { + ( + tokio::fs::File::from_std(open_log_file(path.as_path())?), path, + InMemoryRolloutSource::new(vec![RolloutItem::SessionMeta( + session_meta_line.clone(), + )]), + Some(session_meta_line), event_persistence_mode, - } => ( - Some( - tokio::fs::OpenOptions::new() - .append(true) - .open(&path) - .await?, - ), - None, + ) + } + RolloutStoreParams::Resume { + path, + event_persistence_mode, + source, + } => { + let source = match source { + Some(source) => source, + None => Self::load_source(path.as_path()).await?.source, + }; + ( + tokio::fs::OpenOptions::new() + .append(true) + .open(&path) + .await?, path, + source, None, event_persistence_mode, - ), - }; - - // Clone the cwd for the spawned task to collect git info asynchronously - let cwd = config.cwd.clone(); + ) + } + }; // A reasonably-sized bounded channel. If the buffer fills up the send // future will yield, which is fine – we only need to ensure we do not // perform *blocking* I/O on the caller's thread. 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, + 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( - file, - deferred_log_file_info, + writer.file, rx, - meta, - cwd, rollout_path.clone(), state_db_ctx.clone(), state_builder, config.model_provider_id.clone(), - config.memories.generate_memories, )); Ok(Self { @@ -470,13 +641,16 @@ impl RolloutRecorder { rollout_path, state_db: state_db_ctx, event_persistence_mode, + source, }) } + /// Return the path for this rollout file. pub fn rollout_path(&self) -> &Path { self.rollout_path.as_path() } + /// Return the state DB handle used for rollout reconciliation, when enabled. pub fn state_db(&self) -> Option { self.state_db.clone() } @@ -497,15 +671,16 @@ impl RolloutRecorder { if filtered.is_empty() { return Ok(()); } + let mut source = self.source.lock().await; self.tx - .send(RolloutCmd::AddItems(filtered)) + .send(RolloutCmd::AddItems(filtered.clone())) .await - .map_err(|e| IoError::other(format!("failed to queue rollout items: {e}"))) + .map_err(|e| IoError::other(format!("failed to queue rollout items: {e}")))?; + source.append_items(filtered); + Ok(()) } - /// Materialize the rollout file and persist all buffered items. - /// - /// This is idempotent; after first materialization, repeated calls are no-ops. + /// Ensure all queued rollout writes have been persisted to disk. pub async fn persist(&self) -> std::io::Result<()> { let (tx, rx) = oneshot::channel(); self.tx @@ -527,6 +702,8 @@ impl RolloutRecorder { .map_err(|e| IoError::other(format!("failed waiting for rollout flush: {e}"))) } + // TODO(ccunningham): move this parser under the source implementation + // (not done in this PR to reduce diff) pub(crate) async fn load_rollout_items( path: &Path, ) -> std::io::Result<(Vec, Option, usize)> { @@ -552,29 +729,18 @@ impl RolloutRecorder { } }; - // Parse the rollout line structure match serde_json::from_value::(v.clone()) { Ok(rollout_line) => match rollout_line.item { RolloutItem::SessionMeta(session_meta_line) => { - // Use the FIRST SessionMeta encountered in the file as the canonical - // thread id and main session information. Keep all items intact. if thread_id.is_none() { thread_id = Some(session_meta_line.meta.id); } items.push(RolloutItem::SessionMeta(session_meta_line)); } - RolloutItem::ResponseItem(item) => { - items.push(RolloutItem::ResponseItem(item)); - } - RolloutItem::Compacted(item) => { - items.push(RolloutItem::Compacted(item)); - } - RolloutItem::TurnContext(item) => { - items.push(RolloutItem::TurnContext(item)); - } - RolloutItem::EventMsg(_ev) => { - items.push(RolloutItem::EventMsg(_ev)); - } + RolloutItem::ResponseItem(item) => items.push(RolloutItem::ResponseItem(item)), + RolloutItem::Compacted(item) => items.push(RolloutItem::Compacted(item)), + RolloutItem::TurnContext(item) => items.push(RolloutItem::TurnContext(item)), + RolloutItem::EventMsg(event) => items.push(RolloutItem::EventMsg(event)), }, Err(e) => { trace!("failed to parse rollout line: {e}"); @@ -592,10 +758,21 @@ impl RolloutRecorder { Ok((items, thread_id, parse_errors)) } + /// Load a persisted rollout file into the eager startup history shape used today. pub async fn get_rollout_history(path: &Path) -> std::io::Result { - let (items, thread_id, _parse_errors) = Self::load_rollout_items(path).await?; + let LoadedRolloutSource { + source, + thread_id, + parse_errors: _, + } = Self::load_source(path).await?; let conversation_id = thread_id .ok_or_else(|| IoError::other("failed to parse thread ID from rollout file"))?; + // `InitialHistory::Resumed` still carries an owned `Vec`, so this is the + // boundary where the store's canonical `RolloutSource` snapshot is flattened back into + // the protocol-owned rollout payload for resume/fork startup. + // TODO(ccunningham): when resume startup becomes lazy, replace this eager + // `InitialHistory::Resumed` materialization with a `RolloutSource`-backed history input. + let items = source.into_items(); if items.is_empty() { return Ok(InitialHistory::New); @@ -609,6 +786,7 @@ impl RolloutRecorder { })) } + /// Shut down the background writer after draining all previously queued work. pub async fn shutdown(&self) -> std::io::Result<()> { let (tx_done, rx_done) = oneshot::channel(); match self.tx.send(RolloutCmd::Shutdown { ack: tx_done }).await { @@ -706,41 +884,18 @@ fn open_log_file(path: &Path) -> std::io::Result { #[allow(clippy::too_many_arguments)] async fn rollout_writer( - file: Option, - mut deferred_log_file_info: Option, + file: tokio::fs::File, mut rx: mpsc::Receiver, - mut meta: Option, - cwd: std::path::PathBuf, rollout_path: PathBuf, state_db_ctx: Option, mut state_builder: Option, default_provider: String, - generate_memories: bool, ) -> std::io::Result<()> { - let mut writer = file.map(|file| JsonlWriter { file }); - let mut buffered_items = Vec::::new(); + let mut writer = JsonlWriter { file }; if let Some(builder) = state_builder.as_mut() { builder.rollout_path = rollout_path.clone(); } - // 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(), - &mut state_builder, - default_provider.as_str(), - generate_memories, - ) - .await?; - } - // Process rollout commands while let Some(cmd) = rx.recv().await { match cmd { @@ -749,13 +904,8 @@ async fn rollout_writer( continue; } - if writer.is_none() { - buffered_items.extend(items); - continue; - } - write_and_reconcile_items( - writer.as_mut(), + &mut writer, items.as_slice(), &rollout_path, state_db_ctx.as_deref(), @@ -765,61 +915,14 @@ async fn rollout_writer( .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(), - generate_memories, - ) - .await?; - } - - if !buffered_items.is_empty() { - write_and_reconcile_items( - writer.as_mut(), - 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 Err(e) = writer.file.flush().await { + let _ = ack.send(()); + return Err(e); } let _ = ack.send(()); } RolloutCmd::Flush { ack } => { - // Deferred fresh threads may not have an initialized file yet. - if let Some(writer) = writer.as_mut() - && let Err(e) = writer.file.flush().await - { + if let Err(e) = writer.file.flush().await { let _ = ack.send(()); return Err(e); } @@ -836,52 +939,43 @@ async fn rollout_writer( #[allow(clippy::too_many_arguments)] async fn write_session_meta( - mut writer: Option<&mut JsonlWriter>, - session_meta: SessionMeta, - cwd: &Path, + writer: &mut JsonlWriter, + session_meta_line: SessionMetaLine, rollout_path: &Path, state_db_ctx: Option<&StateRuntime>, state_builder: &mut Option, default_provider: &str, - generate_memories: bool, ) -> std::io::Result<()> { - let git_info = collect_git_info(cwd).await; - let session_meta_line = SessionMetaLine { - meta: session_meta, - git: git_info, - }; + let memory_mode = session_meta_line.meta.memory_mode.clone(); 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?; - } - sync_thread_state_after_write( + 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), - default_provider, - (!generate_memories).then_some("disabled"), + None, + memory_mode.as_deref(), ) .await; Ok(()) } async fn write_and_reconcile_items( - mut writer: Option<&mut JsonlWriter>, + writer: &mut JsonlWriter, items: &[RolloutItem], rollout_path: &Path, state_db_ctx: Option<&StateRuntime>, state_builder: Option<&ThreadMetadataBuilder>, default_provider: &str, ) -> std::io::Result<()> { - if let Some(writer) = writer.as_mut() { - for item in items { - writer.write_rollout_item(item).await?; - } + for item in items { + writer.write_rollout_item(item).await?; } sync_thread_state_after_write( state_db_ctx, @@ -1049,14 +1143,20 @@ async fn resume_candidate_matches_cwd( return true; } - if let Ok((items, _, _)) = RolloutRecorder::load_rollout_items(rollout_path).await - && let Some(latest_turn_context_cwd) = items.iter().rev().find_map(|item| match item { - RolloutItem::TurnContext(turn_context) => Some(turn_context.cwd.as_path()), - RolloutItem::SessionMeta(_) - | RolloutItem::ResponseItem(_) - | RolloutItem::Compacted(_) - | RolloutItem::EventMsg(_) => None, - }) + if let Ok(LoadedRolloutSource { + source, + thread_id: _, + parse_errors: _, + }) = RolloutStore::load_source(rollout_path).await + && let Some(latest_turn_context_cwd) = source + .iter_reverse_from(source.exclusive_end_of_rollout_index()) + .find_map(|(_, item)| match item { + RolloutItem::TurnContext(turn_context) => Some(turn_context.cwd.as_path()), + RolloutItem::SessionMeta(_) + | RolloutItem::ResponseItem(_) + | RolloutItem::Compacted(_) + | RolloutItem::EventMsg(_) => None, + }) { return cwd_matches(latest_turn_context_cwd, cwd); } @@ -1157,16 +1257,16 @@ mod tests { } #[tokio::test] - async fn recorder_materializes_only_after_explicit_persist() -> std::io::Result<()> { + async fn store_creates_rollout_immediately() -> 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( + let store = RolloutStore::new( &config, - RolloutRecorderParams::new( + RolloutStoreParams::new( thread_id, None, SessionSource::Exec, @@ -1179,13 +1279,13 @@ mod tests { ) .await?; - let rollout_path = recorder.rollout_path().to_path_buf(); + let rollout_path = store.rollout_path().to_path_buf(); assert!( - !rollout_path.exists(), - "rollout file should not exist before first user message" + rollout_path.exists(), + "rollout file should exist at startup" ); - recorder + store .record_items(&[RolloutItem::EventMsg(EventMsg::AgentMessage( AgentMessageEvent { message: "buffered-event".to_string(), @@ -1193,13 +1293,9 @@ mod tests { }, ))]) .await?; - recorder.flush().await?; - assert!( - !rollout_path.exists(), - "rollout file should remain deferred before first user message" - ); + store.flush().await?; - recorder + store .record_items(&[RolloutItem::EventMsg(EventMsg::UserMessage( UserMessageEvent { message: "first-user-message".to_string(), @@ -1209,16 +1305,12 @@ mod tests { }, ))]) .await?; - recorder.flush().await?; - assert!( - !rollout_path.exists(), - "user-message-like items should not materialize without explicit persist" - ); + store.flush().await?; - recorder.persist().await?; - // Second call verifies `persist()` is idempotent after materialization. - recorder.persist().await?; - assert!(rollout_path.exists(), "rollout file should be materialized"); + 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!( @@ -1238,7 +1330,75 @@ mod tests { let text_after_second_persist = std::fs::read_to_string(&rollout_path)?; assert_eq!(text_after_second_persist, text); - recorder.shutdown().await?; + store.shutdown().await?; + Ok(()) + } + + #[tokio::test] + async fn store_keeps_source_order_consistent_with_queue_order() -> std::io::Result<()> { + let home = TempDir::new().expect("temp dir"); + let config = ConfigBuilder::default() + .codex_home(home.path().to_path_buf()) + .build() + .await?; + let store = Arc::new( + RolloutStore::new( + &config, + RolloutStoreParams::new( + ThreadId::new(), + None, + SessionSource::Exec, + BaseInstructions::default(), + Vec::new(), + EventPersistenceMode::Limited, + ), + None, + None, + ) + .await?, + ); + + let user_message_a = RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: "A".to_string(), + images: None, + local_images: Vec::new(), + text_elements: Vec::new(), + })); + let user_message_b = RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: "B".to_string(), + images: None, + local_images: Vec::new(), + text_elements: Vec::new(), + })); + + let source_guard = store.source.lock().await; + + let store_a = Arc::clone(&store); + let user_message_a_for_task = user_message_a.clone(); + let task_a = + tokio::spawn(async move { store_a.record_items(&[user_message_a_for_task]).await }); + tokio::task::yield_now().await; + + let store_b = Arc::clone(&store); + let user_message_b_for_task = user_message_b.clone(); + let task_b = + tokio::spawn(async move { store_b.record_items(&[user_message_b_for_task]).await }); + tokio::task::yield_now().await; + + drop(source_guard); + task_a.await.expect("join task A")?; + task_b.await.expect("join task B")?; + + let actual_items = store.source.lock().await.clone().into_items(); + assert!(matches!( + actual_items.first(), + Some(RolloutItem::SessionMeta(_)) + )); + let actual_tail = serde_json::to_value(&actual_items[1..])?; + let expected_tail = serde_json::to_value(vec![user_message_a, user_message_b])?; + assert_eq!(actual_tail, expected_tail); + + store.shutdown().await?; Ok(()) } @@ -1264,9 +1424,9 @@ mod tests { .expect("backfill should be complete"); let thread_id = ThreadId::new(); - let recorder = RolloutRecorder::new( + let recorder = RolloutStore::new( &config, - RolloutRecorderParams::new( + RolloutStoreParams::new( thread_id, None, SessionSource::Cli, @@ -1399,7 +1559,7 @@ mod tests { write_session_file(home.path(), "2025-01-01T12-00-00", Uuid::from_u128(9003))?; let default_provider = config.model_provider_id.clone(); - let page1 = RolloutRecorder::list_threads( + let page1 = RolloutStore::list_threads( &config, 1, None, @@ -1414,7 +1574,7 @@ mod tests { assert_eq!(page1.items[0].path, newest); let cursor = page1.next_cursor.clone().expect("cursor should be present"); - let page2 = RolloutRecorder::list_threads( + let page2 = RolloutStore::list_threads( &config, 1, Some(&cursor), @@ -1478,7 +1638,7 @@ mod tests { .expect("state db upsert should succeed"); let default_provider = config.model_provider_id.clone(); - let page = RolloutRecorder::list_threads( + let page = RolloutStore::list_threads( &config, 10, None, @@ -1547,7 +1707,7 @@ mod tests { .expect("state db upsert should succeed"); let default_provider = config.model_provider_id.clone(); - let page = RolloutRecorder::list_threads( + let page = RolloutStore::list_threads( &config, 1, None, diff --git a/codex-rs/core/src/rollout/tests.rs b/codex-rs/core/src/rollout/tests.rs index 12af36b783..d55f7131d4 100644 --- a/codex-rs/core/src/rollout/tests.rs +++ b/codex-rs/core/src/rollout/tests.rs @@ -109,7 +109,7 @@ async fn insert_state_db_thread( // )); // insert_state_db_thread(home, db_thread_id, db_rollout_path.as_path(), false).await; // -// let page = RolloutRecorder::list_threads( +// let page = RolloutStore::list_threads( // home, // 10, // None, @@ -160,7 +160,7 @@ async fn insert_state_db_thread( // ) // .await; // -// let page = RolloutRecorder::list_threads( +// let page = RolloutStore::list_threads( // home, // 10, // None, @@ -190,7 +190,7 @@ async fn insert_state_db_thread( // ) // .unwrap(); // -// let page = RolloutRecorder::list_threads( +// let page = RolloutStore::list_threads( // home, // 10, // None, diff --git a/codex-rs/core/src/state/service.rs b/codex-rs/core/src/state/service.rs index 012e17bbf6..49ea1f6223 100644 --- a/codex-rs/core/src/state/service.rs +++ b/codex-rs/core/src/state/service.rs @@ -2,7 +2,7 @@ use std::collections::HashMap; use std::sync::Arc; use crate::AuthManager; -use crate::RolloutRecorder; +use crate::RolloutStore; use crate::agent::AgentControl; use crate::analytics_client::AnalyticsEventsClient; use crate::client::ModelClient; @@ -38,7 +38,7 @@ pub(crate) struct SessionServices { pub(crate) main_execve_wrapper_exe: Option, pub(crate) analytics_events_client: AnalyticsEventsClient, pub(crate) hooks: Hooks, - pub(crate) rollout: Mutex>, + pub(crate) rollout: Mutex>, pub(crate) user_shell: Arc, pub(crate) shell_snapshot_tx: watch::Sender>>, pub(crate) show_raw_agent_reasoning: bool, diff --git a/codex-rs/core/src/tasks/review.rs b/codex-rs/core/src/tasks/review.rs index 1146be615d..2aa60904ce 100644 --- a/codex-rs/core/src/tasks/review.rs +++ b/codex-rs/core/src/tasks/review.rs @@ -265,8 +265,7 @@ 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; + // Review turns record rollout items asynchronously, so explicitly flush + // before any follow-up path snapshots the thread's JSONL. + session.flush_rollout().await; } diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index cc6de47104..9eb41a6683 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -18,7 +18,7 @@ use crate::plugins::PluginsManager; use crate::protocol::Event; use crate::protocol::EventMsg; use crate::protocol::SessionConfiguredEvent; -use crate::rollout::RolloutRecorder; +use crate::rollout::RolloutStore; use crate::rollout::truncation; use crate::shell_snapshot::ShellSnapshot; use crate::skills::SkillsManager; @@ -358,8 +358,10 @@ impl ThreadManager { rollout_path: PathBuf, auth_manager: Arc, ) -> CodexResult { - let initial_history = RolloutRecorder::get_rollout_history(&rollout_path).await?; - Box::pin(self.resume_thread_with_history(config, initial_history, auth_manager, false)) + // Thread startup still expects an eager `InitialHistory`, so the rollout store + // materializes one here before resume. + let initial_history = RolloutStore::get_rollout_history(&rollout_path).await?; + self.resume_thread_with_history(config, initial_history, auth_manager, false) .await } @@ -409,7 +411,9 @@ impl ThreadManager { path: PathBuf, persist_extended_history: bool, ) -> CodexResult { - let history = RolloutRecorder::get_rollout_history(&path).await?; + // Fork truncation still operates on an owned `InitialHistory`, so the rollout + // store materializes the rollout before trimming it by user-turn boundary. + let history = RolloutStore::get_rollout_history(&path).await?; let history = truncate_before_nth_user_message(history, nth_user_message); Box::pin(self.state.spawn_thread( config, @@ -515,8 +519,10 @@ impl ThreadManagerState { session_source: SessionSource, inherited_shell_snapshot: Option>, ) -> CodexResult { - let initial_history = RolloutRecorder::get_rollout_history(&rollout_path).await?; - Box::pin(self.spawn_thread_with_source( + // This follows the same eager startup boundary as `resume_thread_from_rollout`: + // the rollout store materializes an owned `InitialHistory` before thread startup. + let initial_history = RolloutStore::get_rollout_history(&rollout_path).await?; + self.spawn_thread_with_source( config, initial_history, Arc::clone(&self.auth_manager), @@ -526,7 +532,7 @@ impl ThreadManagerState { false, None, inherited_shell_snapshot, - )) + ) .await } diff --git a/codex-rs/core/tests/suite/cli_stream.rs b/codex-rs/core/tests/suite/cli_stream.rs index 07faed70a8..bc935cca28 100644 --- a/codex-rs/core/tests/suite/cli_stream.rs +++ b/codex-rs/core/tests/suite/cli_stream.rs @@ -70,7 +70,7 @@ async fn responses_mode_stream_cli() { // TODO(jif) fix // // Verify a new session rollout was created and is discoverable via list_conversations // let provider_filter = vec!["mock".to_string()]; - // let page = RolloutRecorder::list_threads( + // let page = RolloutStore::list_threads( // home.path(), // 10, // None, diff --git a/codex-rs/core/tests/suite/rollout_list_find.rs b/codex-rs/core/tests/suite/rollout_list_find.rs index a9ea2b9c8a..164ec3c2d4 100644 --- a/codex-rs/core/tests/suite/rollout_list_find.rs +++ b/codex-rs/core/tests/suite/rollout_list_find.rs @@ -5,8 +5,8 @@ use std::path::PathBuf; use chrono::Utc; use codex_core::EventPersistenceMode; -use codex_core::RolloutRecorder; -use codex_core::RolloutRecorderParams; +use codex_core::RolloutStore; +use codex_core::RolloutStoreParams; use codex_core::config::ConfigBuilder; use codex_core::find_archived_thread_path_by_id_str; use codex_core::find_thread_path_by_id_str; @@ -164,9 +164,9 @@ async fn find_locates_rollout_file_written_by_recorder() -> std::io::Result<()> .await?; let thread_id = ThreadId::new(); let thread_name = "named thread"; - let recorder = RolloutRecorder::new( + let recorder = RolloutStore::new( &config, - RolloutRecorderParams::new( + RolloutStoreParams::new( thread_id, None, SessionSource::Exec, diff --git a/codex-rs/core/tests/suite/sqlite_state.rs b/codex-rs/core/tests/suite/sqlite_state.rs index b17219e5f1..bf4d8d08dc 100644 --- a/codex-rs/core/tests/suite/sqlite_state.rs +++ b/codex-rs/core/tests/suite/sqlite_state.rs @@ -60,18 +60,10 @@ 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" + rollout_path.exists(), + "fresh thread rollout should exist at startup" ); - 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?; @@ -84,10 +76,6 @@ 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/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index d0a6ac2c3e..20751d58a8 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -1382,7 +1382,7 @@ async fn resolve_resume_path( } else { Some(config.cwd.as_path()) }; - match codex_core::RolloutRecorder::find_latest_thread_path( + match codex_core::RolloutStore::find_latest_thread_path( config, 1, None, diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 98ff8f7f55..117fa7aca4 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -3011,7 +3011,10 @@ pub struct SessionConfiguredEvent { #[ts(optional)] pub network_proxy: Option, - /// Path in which the rollout is stored. Can be `None` for ephemeral threads + /// Path reserved for this rollout. Can be `None` for ephemeral threads. + /// + /// Fresh non-ephemeral sessions may advertise this path before the rollout file has been + /// materialized on disk. #[serde(skip_serializing_if = "Option::is_none")] pub rollout_path: Option, } diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index 43e0e2afe9..3ed9b596ac 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -11,7 +11,7 @@ use codex_cloud_requirements::cloud_requirements_loader; use codex_core::AuthManager; use codex_core::CodexAuth; use codex_core::INTERACTIVE_SESSION_SOURCES; -use codex_core::RolloutRecorder; +use codex_core::RolloutStore; use codex_core::ThreadSortKey; use codex_core::auth::AuthMode; use codex_core::auth::enforce_login_restrictions; @@ -687,7 +687,7 @@ async fn run_ratatui_app( } } else if cli.fork_last { let provider_filter = vec![config.model_provider_id.clone()]; - match RolloutRecorder::list_threads( + match RolloutStore::list_threads( &config, 1, None, @@ -782,7 +782,7 @@ async fn run_ratatui_app( } else { Some(config.cwd.as_path()) }; - match RolloutRecorder::find_latest_thread_path( + match RolloutStore::find_latest_thread_path( &config, 1, None, diff --git a/codex-rs/tui/src/resume_picker.rs b/codex-rs/tui/src/resume_picker.rs index 793e0b947d..972c8a89ab 100644 --- a/codex-rs/tui/src/resume_picker.rs +++ b/codex-rs/tui/src/resume_picker.rs @@ -14,7 +14,7 @@ use chrono::DateTime; use chrono::Utc; use codex_core::Cursor; use codex_core::INTERACTIVE_SESSION_SOURCES; -use codex_core::RolloutRecorder; +use codex_core::RolloutStore; use codex_core::ThreadItem; use codex_core::ThreadSortKey; use codex_core::ThreadsPage; @@ -111,7 +111,7 @@ enum BackgroundEvent { /// between sorting by creation time and last-updated time using the Tab key. /// /// Sessions are loaded on-demand via cursor-based pagination. The backend -/// `RolloutRecorder::list_threads` returns pages ordered by the selected sort key, +/// `RolloutStore::list_threads` returns pages ordered by the selected sort key, /// and the picker deduplicates across pages to handle overlapping windows when /// new sessions appear during pagination. /// @@ -159,7 +159,7 @@ async fn run_session_picker( let config = config.clone(); tokio::spawn(async move { let provider_filter = vec![request.default_provider.clone()]; - let page = RolloutRecorder::list_threads( + let page = RolloutStore::list_threads( &config, PAGE_SIZE, request.cursor.as_ref(), @@ -1456,7 +1456,7 @@ mod tests { // set_rollout_mtime(&path_a, now - Duration::minutes(10)); // set_rollout_mtime(&path_b, now - Duration::seconds(10)); // - // let page = RolloutRecorder::list_threads( + // let page = RolloutStore::list_threads( // tempdir.path(), // PAGE_SIZE, // None, @@ -1765,7 +1765,7 @@ mod tests { // SessionPickerAction::Resume, // ); // - // let page = RolloutRecorder::list_threads( + // let page = RolloutStore::list_threads( // &state.codex_home, // PAGE_SIZE, // None,