Persist paginated items in the local thread store (#32289)

## What changed

- Allow the local thread store to create paginated threads while keeping them unsupported through the app-server API.
- Filter live append items using the history mode recorded with the thread's live recorder, so paginated threads persist canonical `ItemCompleted` events instead of legacy history events.

## Testing

- Add a local-store test that creates a paginated thread and verifies that a paginated item is persisted while a legacy user-message event is omitted.

GitOrigin-RevId: 83d93660afd5f0217213d4f21b5bf2a97745ff19
This commit is contained in:
Owen Lin
2026-07-10 23:26:58 +00:00
committed by copyberry
parent 610f09abcf
commit bfe31598c7
4 changed files with 74 additions and 14 deletions
@@ -963,6 +963,12 @@ impl ThreadRequestProcessor {
thread_source,
environments,
} = params;
if matches!(
history_mode,
Some(codex_app_server_protocol::ThreadHistoryMode::Paginated)
) {
return Err(method_not_found("paginated_threads is not supported yet"));
}
if sandbox.is_some() && permissions.is_some() {
return Err(invalid_request(
"`permissions` cannot be combined with `sandbox`",
@@ -2,7 +2,6 @@ use super::LocalThreadStore;
use crate::CreateThreadParams;
use crate::ThreadStoreError;
use crate::ThreadStoreResult;
use crate::error::reject_paginated_history_mode;
use codex_protocol::protocol::ThreadMemoryMode;
use codex_rollout::RolloutConfig;
use codex_rollout::RolloutRecorder;
@@ -12,7 +11,6 @@ pub(super) async fn create_thread(
store: &LocalThreadStore,
params: CreateThreadParams,
) -> ThreadStoreResult<RolloutRecorder> {
reject_paginated_history_mode(params.history_mode)?;
let cwd = params
.metadata
.cwd
+13 -5
View File
@@ -1,7 +1,6 @@
use std::path::PathBuf;
use codex_protocol::ThreadId;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::ThreadMemoryMode;
use codex_rollout::RolloutConfig;
use codex_rollout::RolloutRecorder;
@@ -116,13 +115,22 @@ pub(super) async fn append_items(
store: &LocalThreadStore,
params: AppendThreadItemsParams,
) -> ThreadStoreResult<()> {
// LocalThreadStore rejects paginated threads before opening a writer.
let persisted_items =
persisted_rollout_items(params.items.as_slice(), ThreadHistoryMode::Legacy);
// A live append should always have a recorder: create/resume installs one, while
// shutdown/discard/delete removes it. Keep the lookup defensive so late appends fail after
// teardown.
let (recorder, history_mode) = store
.live_recorders
.lock()
.await
.get(&params.thread_id)
.map(|entry| (entry.recorder.clone(), entry.history_mode))
.ok_or(ThreadStoreError::ThreadNotFound {
thread_id: params.thread_id,
})?;
let persisted_items = persisted_rollout_items(params.items.as_slice(), history_mode);
if persisted_items.is_empty() {
return Ok(());
}
let recorder = store.live_recorder(params.thread_id).await?;
recorder
.record_canonical_items(persisted_items.as_slice())
.await
+55 -7
View File
@@ -326,12 +326,15 @@ mod tests {
use std::sync::Arc;
use codex_protocol::ThreadId;
use codex_protocol::items::TurnItem;
use codex_protocol::items::UserMessageItem;
use codex_protocol::models::BaseInstructions;
use codex_protocol::models::FunctionCallOutputPayload;
use codex_protocol::models::MessagePhase;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::AgentMessageEvent;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ItemCompletedEvent;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::ThreadHistoryMode;
@@ -1259,15 +1262,60 @@ mod tests {
.await
.expect_err("resume should fail"),
);
}
let mut create_params = create_thread_params(ThreadId::default());
#[tokio::test]
async fn paginated_live_appends_use_paginated_history_mode() {
let home = TempDir::new().expect("temp dir");
let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
let thread_id = ThreadId::default();
let mut create_params = create_thread_params(thread_id);
create_params.history_mode = ThreadHistoryMode::Paginated;
assert_paginated_threads_unsupported(
store
.create_thread(create_params)
.await
.expect_err("paginated create should fail"),
);
store
.create_thread(create_params)
.await
.expect("create paginated thread");
let paginated_item = RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent {
thread_id,
turn_id: "turn-1".to_string(),
item: TurnItem::UserMessage(UserMessageItem {
id: "item-1".to_string(),
client_id: None,
content: Vec::new(),
}),
completed_at_ms: 1,
}));
store
.append_items(AppendThreadItemsParams {
thread_id,
items: vec![
user_message_item("legacy event should not persist"),
paginated_item,
],
})
.await
.expect("append paginated item");
let rollout_path = store
.live_rollout_path(thread_id)
.await
.expect("paginated rollout path");
let (items, _, _) = RolloutRecorder::load_rollout_items(rollout_path.as_path())
.await
.expect("load paginated rollout");
assert!(items.iter().any(|item| {
matches!(
item,
RolloutItem::EventMsg(EventMsg::ItemCompleted(event))
if event.turn_id == "turn-1"
)
}));
assert!(!items.iter().any(|item| {
matches!(
item,
RolloutItem::EventMsg(EventMsg::UserMessage(event))
if event.message == "legacy event should not persist"
)
}));
}
fn create_thread_params(thread_id: ThreadId) -> CreateThreadParams {