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
This commit is contained in:
Raunak Daga
2026-10-01 00:57:29 +00:00
committed by copyberry
parent 3bbf8ec3a1
commit 1f77c0cfa6
4 changed files with 200 additions and 0 deletions
@@ -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;
@@ -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(())
}
@@ -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;
@@ -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