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 9182245cf1..b3f473496c 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -1475,8 +1475,10 @@ async fn thread_fork_tracks_thread_initialized_analytics() -> Result<()> { Ok(()) } +#[test_case(ThreadHistoryMode::Legacy; "legacy")] +#[test_case(ThreadHistoryMode::Paginated; "paginated")] #[tokio::test] -async fn thread_fork_rejects_unmaterialized_thread() -> Result<()> { +async fn thread_fork_rejects_unmaterialized_thread(history_mode: ThreadHistoryMode) -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; MockResponsesConfig::new(&server.uri()).write(codex_home.path())?; @@ -1489,6 +1491,7 @@ async fn thread_fork_rejects_unmaterialized_thread() -> Result<()> { let start_id = mcp .send_thread_start_request_with_auto_env(ThreadStartParams { model: Some("mock-model".to_string()), + history_mode: Some(history_mode), ..Default::default() }) .await?; 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 20649549fd..c34a453e6c 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -447,8 +447,15 @@ async fn wait_for_responses_request_count( Ok(()) } +#[test_case(ThreadHistoryMode::Legacy, false; "legacy with turns")] +#[test_case(ThreadHistoryMode::Legacy, true; "legacy without turns")] +#[test_case(ThreadHistoryMode::Paginated, false; "paginated with turns")] +#[test_case(ThreadHistoryMode::Paginated, true; "paginated without turns")] #[tokio::test] -async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> { +async fn thread_resume_rejects_unmaterialized_thread( + history_mode: ThreadHistoryMode, + exclude_turns: bool, +) -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; mock_responses_config(&server.uri()).write(codex_home.path())?; @@ -462,6 +469,7 @@ async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> { let start_id = mcp .send_thread_start_request_with_auto_env(ThreadStartParams { model: Some("gpt-5.4".to_string()), + history_mode: Some(history_mode), ..Default::default() }) .await?; @@ -472,6 +480,7 @@ async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> { let resume_id = mcp .send_thread_resume_request(ThreadResumeParams { thread_id: thread.id.clone(), + exclude_turns, ..Default::default() }) .await?; diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index 4dbd9e5fe1..b02229a377 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -127,6 +127,9 @@ pub enum RolloutRecorderParams { enum RolloutCmd { AddItems(Vec), + ReadUnpersistedSessionMeta { + ack: oneshot::Sender>, + }, Persist { ack: oneshot::Sender>, }, @@ -1045,6 +1048,19 @@ impl RolloutRecorder { self.rollout_path.as_path() } + /// Reads the creation metadata while this writer still defers its rollout file. + /// + /// The writer orders this read with persistence. Resumed or already materialized rollouts + /// return `None`, so callers must keep using their durable metadata. + pub async fn unpersisted_session_meta(&self) -> std::io::Result> { + let (ack, result) = oneshot::channel(); + self.tx + .send(RolloutCmd::ReadUnpersistedSessionMeta { ack }) + .await + .map_err(IoError::other)?; + result.await.map_err(IoError::other) + } + pub async fn record_canonical_items(&self, items: &[RolloutItem]) -> std::io::Result<()> { if items.is_empty() { return Ok(()); @@ -1966,6 +1982,14 @@ async fn rollout_writer( // Process rollout commands while let Some(cmd) = rx.recv().await { match cmd { + RolloutCmd::ReadUnpersistedSessionMeta { ack } => { + let metadata = if state.is_deferred() { + state.meta.clone() + } else { + None + }; + let _ = ack.send(metadata); + } RolloutCmd::AddItems(items) => { state.add_items(items); state.flush_if_materialized().await; diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index e440aa1617..7e7767ec0f 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -822,10 +822,20 @@ async fn referenced_paginated_rollout_starts_at_history_cutoff_and_resumes() -> ) .await?; let rollout_path = recorder.rollout_path().to_path_buf(); + let initial_meta = recorder + .unpersisted_session_meta() + .await? + .expect("new rollout retains its creation metadata"); + assert!(!rollout_path.exists()); recorder.persist().await?; + assert!(recorder.unpersisted_session_meta().await?.is_none()); recorder.shutdown().await?; let meta = crate::read_session_meta_line(&rollout_path).await?.meta; + assert_eq!( + serde_json::to_value(&initial_meta)?, + serde_json::to_value(&meta)? + ); assert_eq!( meta.forked_from_ordinal_exclusive, Some(history_base.end_ordinal_exclusive) @@ -833,6 +843,7 @@ async fn referenced_paginated_rollout_starts_at_history_cutoff_and_resumes() -> let resumed = RolloutRecorder::new(&config, RolloutRecorderParams::resume(rollout_path.clone())).await?; + assert!(resumed.unpersisted_session_meta().await?.is_none()); resumed .record_canonical_items(&[agent_message_item("first child record")]) .await?; diff --git a/codex-rs/thread-store/src/local/live_writer.rs b/codex-rs/thread-store/src/local/live_writer.rs index e1b49e82b7..a8213b8845 100644 --- a/codex-rs/thread-store/src/local/live_writer.rs +++ b/codex-rs/thread-store/src/local/live_writer.rs @@ -2,6 +2,7 @@ use std::path::PathBuf; use std::sync::Arc; use codex_protocol::ThreadId; +use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadMemoryMode; use codex_rollout::RolloutConfig; @@ -324,6 +325,24 @@ pub(super) async fn live_writer_parts( Ok((entry.recorder.clone(), entry.rollout_id, entry.history_mode)) } +/// Uses the writer's creation record only until its rollout has been materialized. +pub(super) async fn unpersisted_session_meta( + store: &LocalThreadStore, + thread_id: ThreadId, +) -> Option<(PathBuf, SessionMeta)> { + let (recorder, _, _) = live_writer_parts(store, thread_id).await.ok()?; + // Materialized reads should not queue behind unrelated transcript writes. + if codex_rollout::existing_rollout_path(recorder.rollout_path()) + .await + .is_some() + { + return None; + } + // A writer may finish shutdown after lookup; durable reads remain the fallback. + let meta = recorder.unpersisted_session_meta().await.ok()??; + Some((recorder.rollout_path().to_path_buf(), meta)) +} + async fn write_and_project( store: &LocalThreadStore, thread_id: ThreadId, diff --git a/codex-rs/thread-store/src/local/pending_thread_metadata_tests.rs b/codex-rs/thread-store/src/local/pending_thread_metadata_tests.rs index 4c26f48384..f9b4c63348 100644 --- a/codex-rs/thread-store/src/local/pending_thread_metadata_tests.rs +++ b/codex-rs/thread-store/src/local/pending_thread_metadata_tests.rs @@ -233,6 +233,97 @@ async fn pending_rollout_compatible_metadata_is_written_with_observed_update() { assert_pending_metadata_consumed(&store, thread_id).await; } +#[tokio::test] +async fn deferred_reads_preserve_creation_metadata_and_updates() { + for history_mode in [ThreadHistoryMode::Legacy, ThreadHistoryMode::Paginated] { + let (_home, store, runtime) = store_with_runtime().await; + let thread_id = ThreadId::new(); + let mut params = create_thread_params(thread_id); + params.history_mode = history_mode; + params.forked_from_id = Some(ThreadId::new()); + params.parent_thread_id = Some(ThreadId::new()); + let live_thread = LiveThread::create(store.clone(), params.clone()) + .await + .expect("create live thread"); + let rollout_path = store + .live_rollout_path(thread_id) + .await + .expect("live rollout path"); + let initial = live_thread + .read_thread( + /*include_archived*/ false, /*include_history*/ false, + ) + .await + .expect_err("untouched threads still have no stored record"); + assert!(matches!( + initial, + ThreadStoreError::InvalidRequest { message } + if message == format!("no rollout found for thread id {thread_id}") + )); + assert!(!rollout_path.exists()); + assert_eq!( + runtime.get_thread(thread_id).await.expect("read index"), + None + ); + + let updated = live_thread + .update_metadata( + ThreadMetadataPatch { + daybreak_enabled: Some(true), + ..Default::default() + }, + /*include_archived*/ false, + ) + .await + .expect("update metadata without materializing history"); + let read = live_thread + .read_thread( + /*include_archived*/ false, /*include_history*/ false, + ) + .await + .expect("read updated metadata"); + assert_eq!( + serde_json::to_value(read).expect("serialize metadata read"), + serde_json::to_value(&updated).expect("serialize update result") + ); + assert_eq!( + ( + updated.forked_from_id, + updated.parent_thread_id, + updated.originator.clone(), + updated.history_mode, + updated.daybreak_enabled, + ), + ( + params.forked_from_id, + params.parent_thread_id, + Some(params.originator), + history_mode, + Some(true), + ) + ); + assert!(!rollout_path.exists()); + + live_thread + .persist(PersistContext::Standard) + .await + .expect("persist history"); + live_thread.shutdown().await.expect("shutdown writer"); + let durable = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: false, + include_history: false, + }) + .await + .expect("read durable metadata"); + assert_eq!( + serde_json::to_value(durable).expect("serialize durable metadata"), + serde_json::to_value(updated).expect("serialize initial metadata") + ); + } +} + #[tokio::test] async fn idle_shutdown_clears_pending_thread_metadata_without_materializing() { let (_home, store, runtime) = store_with_runtime().await; @@ -245,6 +336,24 @@ async fn idle_shutdown_clears_pending_thread_metadata_without_materializing() { let live_thread = LiveThread::create(store.clone(), create_thread_params(thread_id)) .await .expect("create live thread"); + let read_error = live_thread + .read_thread( + /*include_archived*/ false, /*include_history*/ false, + ) + .await + .expect_err("untouched threads still have no stored record"); + assert!(matches!( + read_error, + ThreadStoreError::InvalidRequest { message } + if message == format!("no rollout found for thread id {thread_id}") + )); + assert!( + !store + .live_rollout_path(thread_id) + .await + .expect("live rollout path") + .exists() + ); live_thread.shutdown().await.expect("shutdown idle thread"); assert_eq!( diff --git a/codex-rs/thread-store/src/local/read_thread.rs b/codex-rs/thread-store/src/local/read_thread.rs index c5dd2e085d..13624f82c8 100644 --- a/codex-rs/thread-store/src/local/read_thread.rs +++ b/codex-rs/thread-store/src/local/read_thread.rs @@ -2,6 +2,7 @@ use chrono::DateTime; use chrono::Utc; use codex_protocol::models::PermissionProfile; use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::ThreadHistoryMode; @@ -34,6 +35,24 @@ pub(super) async fn read_thread( ) -> ThreadStoreResult { let thread_id = params.thread_id; let sqlite_metadata = read_sqlite_metadata(store, thread_id).await; + if !params.include_history + && let Some((path, meta)) = + super::live_writer::unpersisted_session_meta(store, thread_id).await + && sqlite_metadata + .as_ref() + .is_none_or(|metadata| metadata.rollout_path == path && metadata.archived_at.is_none()) + { + // A new writer already owns its metadata. Missing JSONL is intentional, so neither + // filesystem discovery nor compression-race retries can find anything more authoritative. + return match sqlite_metadata { + Some(metadata) => Ok(stored_thread_from_metadata(store, metadata, Some(meta)).await), + // Preserve the saved-thread contract used by fork/resume. A live but untouched + // thread has no stored record yet; its absence is known without searching history. + None => Err(ThreadStoreError::InvalidRequest { + message: format!("no rollout found for thread id {thread_id}"), + }), + }; + } let persisted_model_settings = sqlite_metadata .as_ref() .map(|metadata| (metadata.model.clone(), metadata.reasoning_effort.clone())); @@ -376,6 +395,14 @@ pub(super) async fn stored_thread_from_sqlite_metadata( }); } }; + Ok(stored_thread_from_metadata(store, metadata, session_meta).await) +} + +async fn stored_thread_from_metadata( + store: &LocalThreadStore, + metadata: ThreadMetadata, + session_meta: Option, +) -> StoredThread { let forked_from_id = session_meta.as_ref().and_then(|meta| meta.forked_from_id); let parent_thread_id = session_meta.as_ref().and_then(|meta| meta.parent_thread_id); let history_mode = session_meta @@ -392,7 +419,7 @@ pub(super) async fn stored_thread_from_sqlite_metadata( thread.forked_from_id = forked_from_id; thread.history_mode = history_mode; thread.name = name; - Ok(thread) + thread } pub(super) fn stored_thread_from_state_metadata( diff --git a/codex-rs/thread-store/src/local/update_thread_metadata.rs b/codex-rs/thread-store/src/local/update_thread_metadata.rs index 393513eb99..9e72bcdf02 100644 --- a/codex-rs/thread-store/src/local/update_thread_metadata.rs +++ b/codex-rs/thread-store/src/local/update_thread_metadata.rs @@ -695,6 +695,11 @@ async fn canonical_history_mode( thread_id: ThreadId, rollout_path: &Path, ) -> ThreadStoreResult { + if let Some((path, meta)) = live_writer::unpersisted_session_meta(store, thread_id).await + && path == rollout_path + { + return Ok(meta.history_mode); + } let session_meta = match read_session_meta_line(rollout_path).await { Ok(session_meta) => session_meta, Err(err) => {