diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index fc6a3281b4fd..7269fdacdade 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -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`", diff --git a/codex-rs/thread-store/src/local/create_thread.rs b/codex-rs/thread-store/src/local/create_thread.rs index 9b370ef095f2..b9a75ad7bd25 100644 --- a/codex-rs/thread-store/src/local/create_thread.rs +++ b/codex-rs/thread-store/src/local/create_thread.rs @@ -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 { - reject_paginated_history_mode(params.history_mode)?; let cwd = params .metadata .cwd diff --git a/codex-rs/thread-store/src/local/live_writer.rs b/codex-rs/thread-store/src/local/live_writer.rs index 3e06bdcb76f8..d11589c6e838 100644 --- a/codex-rs/thread-store/src/local/live_writer.rs +++ b/codex-rs/thread-store/src/local/live_writer.rs @@ -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(¶ms.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 diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index dc10cb936221..d4275ff79eaa 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -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 {