From 1f77c0cfa63484e1feee46de8dd5669581cd3adb Mon Sep 17 00:00:00 2001 From: Raunak Daga Date: Wed, 30 Sep 2026 20:29:41 +0000 Subject: [PATCH] Persist empty paginated threads when naming them (#49785) ## Why Naming a paginated thread before its first turn must make it durable so it can be resumed immediately and after an app-server restart. ## What changed Persist the live rollout before updating name metadata in SQLite. Propagate persistence failures before saving the name. ## Testing Add thread-store tests for empty-thread persistence and failure handling, plus an app-server regression test that resumes a named empty thread before and after a restart and completes its first turn. GitOrigin-RevId: 2f7556c7de65f6a86177130c39e1a17951496931 --- codex-rs/app-server/tests/suite/v2/mod.rs | 1 + .../tests/suite/v2/thread_name_persistence.rs | 104 ++++++++++++++++++ .../local/pending_thread_metadata_tests.rs | 87 +++++++++++++++ .../src/local/update_thread_metadata.rs | 8 ++ 4 files changed, 200 insertions(+) create mode 100644 codex-rs/app-server/tests/suite/v2/thread_name_persistence.rs diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index b94a7a45c6..804987396f 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -139,6 +139,7 @@ mod thread_list; mod thread_loaded_list; mod thread_memory_mode_set; mod thread_metadata_update; +mod thread_name_persistence; mod thread_name_websocket; mod thread_queue; mod thread_read; diff --git a/codex-rs/app-server/tests/suite/v2/thread_name_persistence.rs b/codex-rs/app-server/tests/suite/v2/thread_name_persistence.rs new file mode 100644 index 0000000000..fac3a545eb --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/thread_name_persistence.rs @@ -0,0 +1,104 @@ +//! Naming persists an empty paginated thread for resume before and after a restart. + +use anyhow::Result; +use app_test_support::MockResponsesConfig; +use app_test_support::TestAppServer; +use app_test_support::create_mock_responses_server_repeating_assistant; +use codex_app_server_protocol::ThreadHistoryMode; +use codex_app_server_protocol::ThreadResumeParams; +use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadSetNameParams; +use codex_app_server_protocol::ThreadSetNameResponse; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStatus; +use codex_app_server_protocol::UserInput; +use pretty_assertions::assert_eq; +use tempfile::TempDir; +use tokio::time::Duration; +use tokio::time::timeout; + +const READ_TIMEOUT: Duration = Duration::from_secs(30); + +#[tokio::test] +async fn named_empty_paginated_thread_resumes_after_restart() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + MockResponsesConfig::new(&server.uri()).write(codex_home.path())?; + let mut app_server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + let start_id = app_server + .send_thread_start_request_with_auto_env(ThreadStartParams { + history_mode: Some(ThreadHistoryMode::Paginated), + ..Default::default() + }) + .await?; + let ThreadStartResponse { thread, .. } = + timeout(READ_TIMEOUT, app_server.read_response(start_id)).await??; + let thread_id = thread.id; + let name = "Scheduled task"; + let set_name_id = app_server + .send_thread_set_name_request(ThreadSetNameParams { + thread_id: thread_id.clone(), + name: name.to_string(), + }) + .await?; + let _: ThreadSetNameResponse = + timeout(READ_TIMEOUT, app_server.read_response(set_name_id)).await??; + + // Run-now resumes immediately after naming, before any user turn exists. + let resume_id = app_server + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread_id.clone(), + exclude_turns: true, + ..Default::default() + }) + .await?; + let ThreadResumeResponse { thread, .. } = + timeout(READ_TIMEOUT, app_server.read_response(resume_id)).await??; + assert_eq!(thread.id, thread_id); + assert_eq!(thread.name.as_deref(), Some(name)); + assert!( + timeout(READ_TIMEOUT, app_server.shutdown_gracefully()) + .await?? + .success() + ); + drop(app_server); + + let mut app_server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + let resume_id = app_server + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread_id.clone(), + exclude_turns: true, + ..Default::default() + }) + .await?; + let ThreadResumeResponse { thread, .. } = + timeout(READ_TIMEOUT, app_server.read_response(resume_id)).await??; + assert_eq!(thread.id, thread_id); + assert_eq!(thread.name.as_deref(), Some(name)); + + let input = vec![UserInput::Text { + text: "Scheduled run".to_string(), + text_elements: Vec::new(), + }]; + let completed = timeout( + READ_TIMEOUT, + app_server.start_turn_and_wait_for_completion(TurnStartParams { + thread_id: thread_id.clone(), + input, + ..Default::default() + }), + ) + .await??; + assert_eq!(completed.thread_id, thread_id); + assert_eq!(completed.turn.status, TurnStatus::Completed); + assert_eq!(completed.turn.error, None); + Ok(()) +} 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 1f7c272faf..4c26f48384 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 @@ -20,6 +20,93 @@ use crate::ThreadPersistenceMetadata; use crate::UpdateThreadMetadataParams; use crate::local::test_support::test_config; +#[tokio::test] +async fn naming_empty_paginated_thread_materializes_rollout() { + let (_home, store, runtime) = store_with_runtime().await; + let thread_id = ThreadId::new(); + let mut params = create_thread_params(thread_id); + params.history_mode = ThreadHistoryMode::Paginated; + let live_thread = LiveThread::create(store.clone(), params) + .await + .expect("create live thread"); + let rollout_path = store + .live_rollout_path(thread_id) + .await + .expect("live rollout path"); + assert!(!rollout_path.exists()); + + let named = live_thread + .update_metadata( + ThreadMetadataPatch { + name: Some(Some("Scheduled task".to_string())), + memory_mode: Some(ThreadMemoryMode::Disabled), + ..Default::default() + }, + /*include_archived*/ false, + ) + .await + .expect("name empty thread"); + + assert!(rollout_path.exists()); + assert_eq!(named.name.as_deref(), Some("Scheduled task")); + assert_eq!(named.history_mode, ThreadHistoryMode::Paginated); + assert_eq!( + runtime + .get_thread_memory_mode(thread_id) + .await + .expect("read memory mode") + .as_deref(), + Some("disabled") + ); + live_thread.shutdown().await.expect("shutdown named thread"); + let stored = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: false, + include_history: false, + }) + .await + .expect("read durable empty thread"); + assert_eq!(stored.thread_id, thread_id); + assert_eq!(stored.name, named.name); + assert_eq!(stored.history_mode, ThreadHistoryMode::Paginated); +} + +#[tokio::test] +async fn naming_empty_paginated_thread_propagates_rollout_persistence_failure() { + let (_home, store, runtime) = store_with_runtime().await; + let thread_id = ThreadId::new(); + let mut params = create_thread_params(thread_id); + params.history_mode = ThreadHistoryMode::Paginated; + let live_thread = LiveThread::create(store.clone(), params) + .await + .expect("create live thread"); + let rollout_path = store + .live_rollout_path(thread_id) + .await + .expect("live rollout path"); + std::fs::create_dir_all(&rollout_path).expect("block rollout file creation"); + + live_thread + .update_metadata( + ThreadMetadataPatch { + name: Some(Some("Unpersistable task".to_string())), + ..Default::default() + }, + /*include_archived*/ false, + ) + .await + .expect_err("naming must fail when the thread cannot be persisted"); + assert_eq!( + runtime + .get_thread(thread_id) + .await + .expect("read thread metadata") + .and_then(|metadata| metadata.name), + None + ); +} + #[tokio::test] async fn pending_thread_metadata_is_consumed_by_first_metadata_update() { let (_home, store, runtime) = store_with_runtime().await; 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 d8074171c1..393513eb99 100644 --- a/codex-rs/thread-store/src/local/update_thread_metadata.rs +++ b/codex-rs/thread-store/src/local/update_thread_metadata.rs @@ -94,6 +94,14 @@ pub(super) async fn update_thread_metadata( None }; let paginated = matches!(history_mode, Some(ThreadHistoryMode::Paginated)); + if paginated + && patch.name.is_some() + && live_writer::rollout_path(store, thread_id).await.is_ok() + { + // Naming saves a new thread even before its first turn. Persist its live recorder before + // updating SQLite so the named thread can be resumed immediately or after a restart. + live_writer::persist_thread(store, thread_id).await?; + } let needs_rollout_compat = requires_rollout_compat || patch.name.is_some(); // Reject competing writers before committing any part of a legacy rollout patch to SQLite. let writer_lock = if !paginated