diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index c95bca57e5..408453fb8c 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -312,6 +312,8 @@ fn sample_thread_resume_response_with_source( reasoning_effort: None, multi_agent_mode: Default::default(), initial_turns_page: None, + turns_backwards_cursor: None, + items_backwards_cursor: None, }) } diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index db4cf92d61..935d56fceb 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -513,6 +513,8 @@ fn sample_thread_resume_response() -> ClientResponsePayload { reasoning_effort: None, multi_agent_mode: Default::default(), initial_turns_page: None, + turns_backwards_cursor: None, + items_backwards_cursor: None, }) } diff --git a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs index c0df9d92ad..915e37c5d7 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -219,6 +219,8 @@ fn thread_resume_response_round_trips_initial_turns_page() { next_cursor: Some("cursor_next".to_string()), backwards_cursor: Some("cursor_back".to_string()), }), + turns_backwards_cursor: Some("turns_head".to_string()), + items_backwards_cursor: Some("items_head".to_string()), }; let value = serde_json::to_value(&response).expect("serialize thread resume response"); @@ -230,6 +232,14 @@ fn thread_resume_response_round_trips_initial_turns_page() { "backwardsCursor": "cursor_back", })) ); + assert_eq!( + value.get("turnsBackwardsCursor"), + Some(&json!("turns_head")) + ); + assert_eq!( + value.get("itemsBackwardsCursor"), + Some(&json!("items_head")) + ); let decoded = serde_json::from_value::(value) .expect("deserialize thread resume response"); assert_eq!(decoded, response); diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index 0c884f2bac..ae685cc7ca 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -435,6 +435,20 @@ pub struct ThreadResumeResponse { #[experimental("thread/resume.initialTurnsPage")] #[serde(default)] pub initial_turns_page: Option, + /// Opaque head cursor for hydrating paginated turns backwards. + /// + /// Pass this as `cursor` to `thread/turns/list` with + /// `sortDirection: "desc"`. The first page includes the cursor's head turn. + #[experimental("thread/resume.turnsBackwardsCursor")] + #[serde(default)] + pub turns_backwards_cursor: Option, + /// Opaque head cursor for hydrating paginated items backwards. + /// + /// Pass this as `cursor` to `thread/items/list` with + /// `sortDirection: "desc"`. The first page includes the cursor's head item. + #[experimental("thread/resume.itemsBackwardsCursor")] + #[serde(default)] + pub items_backwards_cursor: Option, } impl ThreadResumeResponse { diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 35d240b8e1..869e8e0efe 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -137,7 +137,7 @@ Example with notification opt-out: ## API Overview -- `thread/start` — create a new thread; emits `thread/started` (including the current `thread.status`) and auto-subscribes you to turn/item events for that thread. When the request includes a `cwd` and the resolved sandbox is `workspace-write` or full access, app-server also marks that project as trusted in the user `config.toml`. Pass `sessionStartSource: "clear"` when starting a replacement thread after clearing the current session so `SessionStart` hooks receive `source: "clear"` instead of the default `"startup"`. Experimental `allowProviderModelFallback` lets providers backed by an authoritative static model catalog replace an unavailable requested `model` with the catalog default; dynamic or cached catalogs preserve the requested model. Experimental `runtimeWorkspaceRoots` supplies the runtime workspace roots used when app-server creates default environment selections; paths must be absolute. For permissions, prefer experimental `permissions` profile selection by id; the legacy `sandbox` shorthand is still accepted but cannot be combined with `permissions`. Deprecated experimental `multiAgentMode` is ignored; use Ultra reasoning effort for proactive multi-agent behavior. Experimental `environments` selects the sticky execution environments for turns on the thread; omit it to use the server default, pass `[]` to disable environments, or pass explicit environment ids with per-environment `cwd` and optional environment-native `runtimeWorkspaceRoots`. Explicit environments ignore the top-level roots; omitted per-environment roots default to that environment's `cwd`, while an empty list explicitly selects no roots. Experimental `selectedCapabilityRoots` selects environment-owned plugin or standalone-skill roots using environment-native absolute paths. Skills found below those roots are listed and read through the owning environment. Stdio MCP servers declared by selected plugins are started in that environment, and HTTP MCP connections use that environment's HTTP client. +- `thread/start` — create a new thread; emits `thread/started` (including the current `thread.status`) and auto-subscribes you to turn/item events for that thread. Experimental `historyMode: "paginated"` selects projection-backed durable history. When the request includes a `cwd` and the resolved sandbox is `workspace-write` or full access, app-server also marks that project as trusted in the user `config.toml`. Pass `sessionStartSource: "clear"` when starting a replacement thread after clearing the current session so `SessionStart` hooks receive `source: "clear"` instead of the default `"startup"`. Experimental `allowProviderModelFallback` lets providers backed by an authoritative static model catalog replace an unavailable requested `model` with the catalog default; dynamic or cached catalogs preserve the requested model. Experimental `runtimeWorkspaceRoots` supplies the runtime workspace roots used when app-server creates default environment selections; paths must be absolute. For permissions, prefer experimental `permissions` profile selection by id; the legacy `sandbox` shorthand is still accepted but cannot be combined with `permissions`. Deprecated experimental `multiAgentMode` is ignored; use Ultra reasoning effort for proactive multi-agent behavior. Experimental `environments` selects the sticky execution environments for turns on the thread; omit it to use the server default, pass `[]` to disable environments, or pass explicit environment ids with per-environment `cwd` and optional environment-native `runtimeWorkspaceRoots`. Explicit environments ignore the top-level roots; omitted per-environment roots default to that environment's `cwd`, while an empty list explicitly selects no roots. Experimental `selectedCapabilityRoots` selects environment-owned plugin or standalone-skill roots using environment-native absolute paths. Skills found below those roots are listed and read through the owning environment. Stdio MCP servers declared by selected plugins are started in that environment, and HTTP MCP connections use that environment's HTTP client. - `thread/resume` — reopen an existing thread by id so subsequent `turn/start` calls append to it. Accepts the same permission override rules as `thread/start`. - `thread/fork` — fork an existing thread into a new thread id by copying the stored history; pass an optional `lastTurnId` to copy history only through that turn, inclusive, and drop later turns from the fork. An in-progress `lastTurnId` boundary is rejected. Experimental `beforeTurnId` instead copies history strictly before the referenced turn, including when that turn is in progress, and cannot be combined with `lastTurnId`. If both boundaries are null while the source thread is mid-turn, the fork records the same interruption marker as `turn/interrupt` instead of inheriting an unmarked partial turn suffix. The returned `thread.forkedFromId` points at the source thread when known. Accepts `ephemeral: true` for an in-memory temporary fork, emits `thread/started` (including the current `thread.status`), and auto-subscribes you to turn/item events for the new thread. Experimental clients can pass `excludeTurns: true` when they plan to page fork history via `thread/turns/list` instead of receiving the full turn array immediately, or `deferGoalContinuation: true` to carry the source thread's current goal into the fork and run an explicit turn before automatic continuation resumes. Deferred goal continuation is persisted until that turn starts and cannot be combined with `ephemeral: true`. Accepts the same permission override rules as `thread/start`. - `thread/start`, `thread/resume`, and `thread/fork` responses include the legacy `sandbox` compatibility projection. `instructionSources` lists loaded instruction files using each source environment's native absolute path syntax, including files loaded from remote environments. Experimental clients can read `runtimeWorkspaceRoots` for the thread-scoped runtime roots and `activePermissionProfile` for the named or implicit built-in profile identity/provenance when known. Their deprecated experimental `multiAgentMode` field, and the corresponding thread setting, always report `explicitRequestOnly`; Ultra reasoning effort is the source of proactive multi-agent behavior. @@ -167,7 +167,7 @@ Example with notification opt-out: - `thread/backgroundTerminals/clean` — terminate all running background terminals for a thread (experimental; requires `capabilities.experimentalApi`); returns `{}` when the cleanup request is accepted. - `thread/backgroundTerminals/list` — list running background terminals for a loaded thread (experimental; requires `capabilities.experimentalApi`); returns `data` with the running terminal ids. - `thread/backgroundTerminals/terminate` — terminate one running background terminal by app-server `processId` (experimental; requires `capabilities.experimentalApi`); returns whether a process was terminated. -- `thread/rollback` — deprecated and will be removed soon. Drop the last N turns from the agent’s in-memory context and persist a rollback marker in the rollout so future resumes see the pruned history; returns the updated `thread` (with `turns` populated) on success. +- `thread/rollback` — deprecated and will be removed soon. Drop the last N turns from the agent’s in-memory context and persist a rollback marker in the rollout so future resumes see the pruned history; returns the updated `thread` (with `turns` populated) on success. Paginated threads do not support rollback. - `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Experimental `runtimeWorkspaceRoots` supplies the default roots for newly resolved environment selections. Explicit `environments[].runtimeWorkspaceRoots` override that fallback with environment-native absolute paths. Prefer experimental `permissions` profile selection by id for permission overrides; the legacy `sandboxPolicy` field is still accepted but cannot be combined with `permissions`. For `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode". Deprecated experimental `multiAgentMode` is ignored; Ultra reasoning effort selects proactive behavior. - `thread/inject_items` — append raw Responses API items to a loaded thread’s model-visible history without starting a user turn; returns `{}` on success. - `turn/steer` — add user input to an already in-flight regular turn without starting a new turn; returns the active `turnId` that accepted the input. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Review and manual compaction turns reject `turn/steer`. @@ -319,6 +319,8 @@ To continue a stored session, call `thread/resume` with the `thread.id` you prev By default, `thread/resume` includes the reconstructed turn history in `thread.turns`. Experimental clients can pass `excludeTurns: true` to return only thread metadata and live resume state, then call `thread/turns/list` separately if they want to page the turn history over the network. In that mode the server also skips replaying restored `thread/tokenUsage/updated`, which avoids rebuilding turns just to attribute historical usage. +Paginated threads require `excludeTurns: true`. Their response leaves `thread.turns` empty and includes `turnsBackwardsCursor` and `itemsBackwardsCursor` for the durable history visible at the resume boundary. Pass each cursor directly to its matching list API with `sortDirection: "desc"`; the first page includes the cursor's head row, while newer records arrive through live notifications. Either cursor is `null` when there is no durable row yet. Full-history resume and `initialTurnsPage` remain unavailable for paginated threads. + Experimental clients that want the live resume subscription plus a turns page in one round trip can pass `initialTurnsPage`. It accepts the same `limit`, `sortDirection`, and `itemsView` controls as `thread/turns/list`; omitted controls use its defaults. The response includes `initialTurnsPage` with `nextCursor` and `backwardsCursor` for follow-up pagination. By default, resume uses the latest persisted `model` and `reasoningEffort` values associated with the thread. Supplying any of `model`, `modelProvider`, `config.model`, or `config.model_reasoning_effort` disables that persisted fallback and uses the explicit overrides plus normal config resolution instead. @@ -336,7 +338,11 @@ Example: "threadId": "thr_123", "excludeTurns": true } } -{ "id": 12, "result": { "thread": { "id": "thr_123", "turns": [], … } } } +{ "id": 12, "result": { + "thread": { "id": "thr_123", "turns": [], … }, + "turnsBackwardsCursor": "turn-head-cursor-or-null", + "itemsBackwardsCursor": "item-head-cursor-or-null" +} } { "method": "thread/resume", "id": 13, "params": { "threadId": "thr_123", @@ -483,6 +489,8 @@ Later, after the idle unload timeout: Use `thread/read` to fetch a stored thread by id without resuming it. Pass `includeTurns` when you want thread history loaded into `thread.turns`. The returned thread includes `parentThreadId`, `agentNickname`, and `agentRole` for subagent threads when available. +Paginated threads support metadata-only reads; `includeTurns: true` is unsupported for them. + ```json { "method": "thread/read", "id": 22, "params": { "threadId": "thr_123" } } { "id": 22, "result": { @@ -531,8 +539,9 @@ Paginated threads support `summary` and `notLoaded`; `full` returns an invalid-r ``` Each returned entry includes the containing `turnId` and its full `item`, so clients can group -unfiltered pages into turns. Omit `turnId` or pass `null` to page items across the thread. Thread -stores that do not implement item pagination return JSON-RPC `-32601` with message +unfiltered pages into turns. Omit `turnId` or pass `null` to page items across the thread. Item +cursors can be reused with or without `turnId`; the filter does not change the cursor's scope. +Thread stores that do not implement item pagination return JSON-RPC `-32601` with message `thread/items/list is not supported yet`. ### Example: Update stored thread metadata @@ -1031,6 +1040,8 @@ Example request/response: For a detached review, use `"delivery": "detached"`. The response is the same shape, but `reviewThreadId` will be the id of the new review thread (different from the original `threadId`). The server also emits a `thread/started` notification for that new thread before streaming the review turn. Internally, this is a normal forked thread and turn whose prompt mentions the bundled `$review-agent` skill, so normal turn steering, tool, permission, and item-stream behavior applies. +Detached review is unsupported when the parent thread is paginated. + For an inline review, Codex streams the usual `turn/started` notification followed by an `item/started` with an `enteredReviewMode` item so clients can show progress: diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index a6aa1c288b..0f8e3a0eff 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -461,6 +461,7 @@ use codex_thread_store::GitInfoPatch as StoreGitInfoPatch; use codex_thread_store::ListItemsParams as StoreListItemsParams; use codex_thread_store::ListThreadsParams as StoreListThreadsParams; use codex_thread_store::ListTurnsParams as StoreListTurnsParams; +use codex_thread_store::LoadThreadHistoryParams as StoreLoadThreadHistoryParams; use codex_thread_store::LocalThreadStore; use codex_thread_store::ReadThreadByRolloutPathParams as StoreReadThreadByRolloutPathParams; use codex_thread_store::ReadThreadParams as StoreReadThreadParams; diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index da5db51e48..3af419da1d 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -622,6 +622,25 @@ pub(super) async fn handle_pending_thread_resume_request( } } + let (turns_backwards_cursor, items_backwards_cursor) = if let Some(thread_store) = + pending.resume_cursor_store.as_ref() + { + match super::thread_processor::ThreadRequestProcessor::paginated_resume_backwards_cursors( + thread_store.as_ref(), + conversation_id, + ) + .await + { + Ok(cursors) => cursors, + Err(error) => { + outgoing.send_error(request_id, error).await; + return; + } + } + } else { + (None, None) + }; + let config_snapshot = pending.config_snapshot; let sandbox = config_snapshot.sandbox_policy().into(); let cwd = config_snapshot.cwd().clone(); @@ -658,6 +677,8 @@ pub(super) async fn handle_pending_thread_resume_request( reasoning_effort, multi_agent_mode: MultiAgentMode::ExplicitRequestOnly, initial_turns_page, + turns_backwards_cursor, + items_backwards_cursor, }; outgoing .send_response_with_thread_originator(request_id, response, originator) 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 85937f9de4..f57224f2e9 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -967,8 +967,11 @@ impl ThreadRequestProcessor { if matches!( history_mode, Some(codex_app_server_protocol::ThreadHistoryMode::Paginated) - ) { - return Err(method_not_found("paginated_threads is not supported yet")); + ) && !self.thread_store.supports_paginated_history_lists() + { + return Err(invalid_request( + "paginated threads require thread/turns/list and thread/items/list support", + )); } if sandbox.is_some() && permissions.is_some() { return Err(invalid_request( @@ -1783,6 +1786,14 @@ impl ThreadRequestProcessor { } let (thread_id, thread) = self.load_thread(&thread_id).await?; + if matches!( + thread.config_snapshot().await.history_mode, + ThreadHistoryMode::Paginated + ) { + return Err(invalid_request( + "paginated threads do not support thread/rollback", + )); + } let request = request_id.clone(); @@ -2329,23 +2340,45 @@ impl ThreadRequestProcessor { include_turns: bool, ) -> Result, ThreadReadViewError> { let fallback_provider = self.config.model_provider_id.as_str(); + if include_turns + && self + .read_stored_thread_for_read(thread_id, /*include_history*/ false) + .await? + .is_some_and(|thread| matches!(thread.history_mode, ThreadHistoryMode::Paginated)) + { + return Err(ThreadReadViewError::InvalidRequest( + "paginated threads do not support thread/read(includeTurns=true)".to_string(), + )); + } + let Some(stored_thread) = self + .read_stored_thread_for_read(thread_id, /*include_history*/ include_turns) + .await? + else { + return Ok(None); + }; + let (mut thread, history) = + thread_from_stored_thread(stored_thread, fallback_provider, &self.config.cwd); + if include_turns && let Some(history) = history { + thread.turns = build_legacy_api_turns_from_rollout_items(&history.items); + } + Ok(Some(thread)) + } + + async fn read_stored_thread_for_read( + &self, + thread_id: ThreadId, + include_history: bool, + ) -> Result, ThreadReadViewError> { match self .thread_store .read_thread(StoreReadThreadParams { thread_id, include_archived: true, - include_history: include_turns, + include_history, }) .await { - Ok(stored_thread) => { - let (mut thread, history) = - thread_from_stored_thread(stored_thread, fallback_provider, &self.config.cwd); - if include_turns && let Some(history) = history { - thread.turns = build_legacy_api_turns_from_rollout_items(&history.items); - } - Ok(Some(thread)) - } + Ok(stored_thread) => Ok(Some(stored_thread)), Err(ThreadStoreError::InvalidRequest { message }) if message == format!("no rollout found for thread id {thread_id}") => { @@ -2380,6 +2413,11 @@ impl ThreadRequestProcessor { "ephemeral threads do not support includeTurns".to_string(), )); } + if include_turns && matches!(config_snapshot.history_mode, ThreadHistoryMode::Paginated) { + return Err(ThreadReadViewError::InvalidRequest( + "paginated threads do not support thread/read(includeTurns=true)".to_string(), + )); + } let fallback_thread = build_thread_from_loaded_snapshot(thread_id, &config_snapshot, loaded_thread); let mut thread = if let Some(mut thread) = persisted_thread { @@ -2573,6 +2611,35 @@ impl ThreadRequestProcessor { }) } + pub(super) async fn paginated_resume_backwards_cursors( + thread_store: &dyn ThreadStore, + thread_id: ThreadId, + ) -> Result<(Option, Option), JSONRPCErrorError> { + let turns_page = thread_store + .list_turns(StoreListTurnsParams { + thread_id, + include_archived: true, + cursor: None, + page_size: 1, + sort_direction: StoreSortDirection::Desc, + items_view: StoredTurnItemsView::NotLoaded, + }) + .await + .map_err(paginated_history_list_error)?; + let items_page = thread_store + .list_items(StoreListItemsParams { + thread_id, + turn_id: None, + include_archived: true, + cursor: None, + page_size: 1, + sort_direction: StoreSortDirection::Desc, + }) + .await + .map_err(paginated_history_list_error)?; + Ok((turns_page.backwards_cursor, items_page.backwards_cursor)) + } + async fn thread_items_list_response_inner( &self, params: ThreadItemsListParams, @@ -2850,14 +2917,33 @@ impl ThreadRequestProcessor { self.resume_thread_from_history(history.as_slice()) .await .map(|thread_history| (thread_history, None)) - } else if let Some(mut stored_thread) = stored_thread_from_running_probe { - self.stored_thread_to_initial_history(&mut stored_thread) - .await - .map(|thread_history| (thread_history, Some(*stored_thread))) + } else if let Some(stored_thread) = stored_thread_from_running_probe { + self.load_resume_initial_history_from_stored_thread( + *stored_thread, + exclude_turns, + initial_turns_page.is_some(), + ) + .await + .map(|(thread_history, stored_thread)| (thread_history, Some(stored_thread))) } else { - self.resume_thread_from_rollout(&thread_id, path.as_ref()) + match self + .read_stored_thread_for_resume( + &thread_id, + path.as_ref(), + /*include_history*/ false, + ) .await - .map(|(thread_history, stored_thread)| (thread_history, Some(stored_thread))) + { + Ok(stored_thread) => self + .load_resume_initial_history_from_stored_thread( + stored_thread, + exclude_turns, + initial_turns_page.is_some(), + ) + .await + .map(|(thread_history, stored_thread)| (thread_history, Some(stored_thread))), + Err(error) => Err(error), + } }; let (thread_history, resume_source_thread) = match resume_result { Ok(value) => value, @@ -3004,6 +3090,23 @@ impl ThreadRequestProcessor { /*has_live_in_progress_turn*/ false, ); let config_snapshot = codex_thread.config_snapshot().await; + let (turns_backwards_cursor, items_backwards_cursor) = + if matches!(config_snapshot.history_mode, ThreadHistoryMode::Paginated) { + match Self::paginated_resume_backwards_cursors( + self.thread_store.as_ref(), + thread_id, + ) + .await + { + Ok(cursors) => cursors, + Err(error) => { + self.outgoing.send_error(request_id, error).await; + return Ok(()); + } + } + } else { + (None, None) + }; let sandbox = config_snapshot.sandbox_policy().into(); let active_permission_profile = thread_response_active_permission_profile( config_snapshot.active_permission_profile, @@ -3049,6 +3152,8 @@ impl ThreadRequestProcessor { reasoning_effort: session_configured.reasoning_effort, multi_agent_mode: MultiAgentMode::ExplicitRequestOnly, initial_turns_page, + turns_backwards_cursor, + items_backwards_cursor, }; let connection_id = request_id.connection_id; @@ -3139,7 +3244,7 @@ impl ThreadRequestProcessor { .read_stored_thread_for_resume( ¶ms.thread_id, /*path*/ None, - /*include_history*/ true, + /*include_history*/ false, ) .await?; Some((existing_thread_id, existing_thread, source_thread)) @@ -3148,7 +3253,7 @@ impl ThreadRequestProcessor { .read_stored_thread_for_resume( ¶ms.thread_id, params.path.as_ref(), - /*include_history*/ true, + /*include_history*/ false, ) .await?; let existing_thread_id = source_thread.thread_id; @@ -3163,6 +3268,11 @@ impl ThreadRequestProcessor { }; if let Some((existing_thread_id, existing_thread, mut source_thread)) = running_thread { + ensure_paginated_resume_supported( + source_thread.history_mode, + params.exclude_turns, + params.initial_turns_page.is_some(), + )?; let existing_thread_rollout_path = existing_thread.rollout_path(); let active_path = existing_thread_rollout_path .as_ref() @@ -3222,15 +3332,32 @@ impl ThreadRequestProcessor { } let redact_resume_payloads = should_redact_thread_resume_payloads(app_server_client_name.as_deref()); - let history_items = source_thread - .history - .take() - .map(|history| history.items) - .ok_or_else(|| { - internal_error(format!( - "thread {existing_thread_id} did not include persisted history" - )) - })?; + let include_turns = !params.exclude_turns; + let needs_history = include_turns || params.initial_turns_page.is_some(); + if needs_history { + let source_thread_id = source_thread.thread_id.to_string(); + let source_rollout_path = source_thread.rollout_path.clone(); + source_thread = self + .read_stored_thread_for_resume( + &source_thread_id, + source_rollout_path.as_ref(), + /*include_history*/ true, + ) + .await?; + } + let history_items = if needs_history { + source_thread + .history + .take() + .map(|history| history.items) + .ok_or_else(|| { + internal_error(format!( + "thread {existing_thread_id} did not include persisted history" + )) + })? + } else { + Vec::new() + }; let thread_state = self .thread_state_manager @@ -3255,6 +3382,7 @@ impl ThreadRequestProcessor { /*include_turns*/ false, ); thread_summary.session_id = existing_thread.session_configured().session_id.to_string(); + thread_summary.thread_source = config_snapshot.thread_source.clone().map(Into::into); let instruction_sources = existing_thread.legacy_instruction_sources().await; let listener_command_tx = { @@ -3271,6 +3399,9 @@ impl ThreadRequestProcessor { .thread_goal_processor .pending_resume_goal_state(existing_thread.as_ref()) .await; + let resume_cursor_store = + matches!(config_snapshot.history_mode, ThreadHistoryMode::Paginated) + .then(|| Arc::clone(&self.thread_store)); let command = crate::thread_state::ThreadListenerCommand::SendThreadResumeResponse( Box::new(crate::thread_state::PendingThreadResumeRequest { @@ -3281,8 +3412,9 @@ impl ThreadRequestProcessor { thread_summary, emit_thread_goal_update, thread_goal_state_db, - include_turns: !params.exclude_turns, + include_turns, initial_turns_page: params.initial_turns_page.clone(), + resume_cursor_store, redact_resume_payloads, }), ); @@ -3313,14 +3445,42 @@ impl ThreadRequestProcessor { )) } - #[tracing::instrument(level = "trace", skip_all)] - async fn resume_thread_from_rollout( + async fn load_resume_initial_history_from_stored_thread( &self, - thread_id: &str, - path: Option<&PathBuf>, + stored_thread: StoredThread, + exclude_turns: bool, + has_initial_turns_page: bool, ) -> Result<(InitialHistory, StoredThread), JSONRPCErrorError> { + ensure_paginated_resume_supported( + stored_thread.history_mode, + exclude_turns, + has_initial_turns_page, + )?; + if matches!(stored_thread.history_mode, ThreadHistoryMode::Paginated) { + let model_context = self + .thread_store + .load_latest_model_context(StoreLoadThreadHistoryParams { + thread_id: stored_thread.thread_id, + include_archived: true, + }) + .await + .map_err(thread_store_resume_read_error)?; + let history = InitialHistory::Resumed(ResumedHistory { + conversation_id: model_context.thread_id, + history: Arc::new(model_context.items), + rollout_path: stored_thread.rollout_path.clone(), + }); + return Ok((history, stored_thread)); + } + + let thread_id = stored_thread.thread_id.to_string(); + let rollout_path = stored_thread.rollout_path.clone(); let mut stored_thread = self - .read_stored_thread_for_resume(thread_id, path, /*include_history*/ true) + .read_stored_thread_for_resume( + &thread_id, + rollout_path.as_ref(), + /*include_history*/ true, + ) .await?; let history = self .stored_thread_to_initial_history(&mut stored_thread) @@ -4311,6 +4471,19 @@ fn thread_read_view_error(err: ThreadReadViewError) -> JSONRPCErrorError { } } +fn paginated_history_list_error(err: ThreadStoreError) -> JSONRPCErrorError { + match err { + ThreadStoreError::InvalidRequest { message } => invalid_request(message), + ThreadStoreError::Unsupported { operation } => { + unsupported_thread_store_operation(operation) + } + ThreadStoreError::ThreadNotFound { thread_id } => { + invalid_request(format!("no rollout found for thread id {thread_id}")) + } + err => internal_error(format!("failed to list thread history: {err}")), + } +} + fn deserialize_stored_thread_item( item: codex_thread_store::StoredThreadItem, ) -> Result { @@ -4354,6 +4527,27 @@ fn stored_turn_to_api_turn( }) } +fn ensure_paginated_resume_supported( + history_mode: ThreadHistoryMode, + exclude_turns: bool, + has_initial_turns_page: bool, +) -> Result<(), JSONRPCErrorError> { + if !matches!(history_mode, ThreadHistoryMode::Paginated) { + return Ok(()); + } + if !exclude_turns { + return Err(invalid_request( + "paginated threads do not support full-history thread/resume; pass excludeTurns=true", + )); + } + if has_initial_turns_page { + return Err(invalid_request( + "paginated threads do not support initialTurnsPage; use turnsBackwardsCursor and itemsBackwardsCursor", + )); + } + Ok(()) +} + pub(super) fn unsupported_thread_store_operation(operation: &'static str) -> JSONRPCErrorError { method_not_found(format!("{operation} is not supported yet")) } diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index c6c2143311..2771cf5410 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -1258,6 +1258,17 @@ impl TurnRequestProcessor { parent_thread: Arc, prompt: &str, ) -> std::result::Result<(), JSONRPCErrorError> { + // AgentRunner::start still delegates to spawn_subagent, which forks from the parent's + // full history. Paginated threads only allow bounded model-context reads, so keep this + // closed until detached review has a bounded fork path. + if matches!( + parent_thread.config_snapshot().await.history_mode, + codex_protocol::protocol::ThreadHistoryMode::Paginated + ) { + return Err(invalid_request( + "paginated threads do not support detached review", + )); + } let mut config = self.config.as_ref().clone(); if let Some(review_model) = &config.review_model { config.model = Some(review_model.clone()); diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index fec786373c..0e634e3f7f 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -40,6 +40,7 @@ pub(crate) struct PendingThreadResumeRequest { pub(crate) include_turns: bool, pub(crate) initial_turns_page: Option, + pub(crate) resume_cursor_store: Option>, pub(crate) redact_resume_payloads: bool, } diff --git a/codex-rs/app-server/tests/common/lib.rs b/codex-rs/app-server/tests/common/lib.rs index d39340a899..e065e77a9e 100644 --- a/codex-rs/app-server/tests/common/lib.rs +++ b/codex-rs/app-server/tests/common/lib.rs @@ -41,6 +41,7 @@ pub use responses::create_final_assistant_message_sse_response; pub use responses::create_request_permissions_sse_response; pub use responses::create_request_user_input_sse_response; pub use responses::create_shell_command_sse_response; +pub use rollout::create_fake_paginated_rollout; pub use rollout::create_fake_parented_rollout_with_source; pub use rollout::create_fake_rollout; pub use rollout::create_fake_rollout_with_source; diff --git a/codex-rs/app-server/tests/common/rollout.rs b/codex-rs/app-server/tests/common/rollout.rs index a563e48d92..847ae7a5bf 100644 --- a/codex-rs/app-server/tests/common/rollout.rs +++ b/codex-rs/app-server/tests/common/rollout.rs @@ -6,6 +6,7 @@ use codex_protocol::protocol::GitInfo; use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::TokenCountEvent; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TokenUsageInfo; @@ -56,6 +57,41 @@ pub fn create_fake_rollout( ) } +/// Creates a minimal paginated rollout with ordinalized JSONL records. +pub fn create_fake_paginated_rollout( + codex_home: &Path, + filename_ts: &str, + meta_rfc3339: &str, + preview: &str, + model_provider: Option<&str>, + git_info: Option, +) -> Result { + let thread_id = create_fake_rollout( + codex_home, + filename_ts, + meta_rfc3339, + preview, + model_provider, + git_info, + )?; + let path = rollout_path(codex_home, filename_ts, &thread_id); + let mut lines = fs::read_to_string(path.as_path())? + .lines() + .map(serde_json::from_str::) + .collect::, _>>()?; + lines[0]["payload"]["history_mode"] = serde_json::to_value(ThreadHistoryMode::Paginated)?; + for (ordinal, line) in lines.iter_mut().enumerate() { + line["ordinal"] = serde_json::to_value(ordinal)?; + } + let contents = lines + .into_iter() + .map(|line| line.to_string()) + .collect::>() + .join("\n"); + fs::write(path, format!("{contents}\n"))?; + Ok(thread_id) +} + /// Creates a minimal rollout whose history includes a persisted token usage event. /// /// Resume and fork tests use this fixture to verify lifecycle replay of restored diff --git a/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs b/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs index 2002b08a88..b15b32188d 100644 --- a/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs +++ b/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs @@ -18,6 +18,7 @@ use std::path::Path; use std::sync::Arc; use anyhow::Result; +use app_test_support::TestAppServer; use app_test_support::create_mock_responses_server_repeating_assistant; use codex_app_server::in_process; use codex_app_server::in_process::InProcessClientHandle; @@ -26,10 +27,12 @@ use codex_app_server::in_process::InProcessStartArgs; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::InitializeParams; +use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ThreadDeleteParams; use codex_app_server_protocol::ThreadDeleteResponse; +use codex_app_server_protocol::ThreadHistoryMode; use codex_app_server_protocol::ThreadListParams; use codex_app_server_protocol::ThreadListResponse; use codex_app_server_protocol::ThreadResumeParams; @@ -60,6 +63,48 @@ use uuid::Uuid; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +#[tokio::test] +async fn thread_start_rejects_paginated_history_without_list_support() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + let store_id = Uuid::new_v4().to_string(); + create_config_toml_with_thread_store(codex_home.path(), &server.uri(), &store_id)?; + + let _in_memory_store = InMemoryThreadStoreId { store_id }; + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build() + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.initialize_with_client_info(ClientInfo { + name: "codex-app-server-tests".to_string(), + title: None, + version: "0.1.0".to_string(), + }), + ) + .await??; + let request_id = mcp + .send_thread_start_request_with_auto_env(ThreadStartParams { + history_mode: Some(ThreadHistoryMode::Paginated), + ..Default::default() + }) + .await?; + let error: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(request_id)), + ) + .await??; + + assert_eq!(error.error.code, -32600); + assert_eq!( + error.error.message, + "paginated threads require thread/turns/list and thread/items/list support" + ); + + Ok(()) +} + #[tokio::test] async fn thread_delete_with_non_local_thread_store_does_not_create_local_persistence() -> Result<()> { diff --git a/codex-rs/app-server/tests/suite/v2/review.rs b/codex-rs/app-server/tests/suite/v2/review.rs index aace8de3c9..9c78cb8a63 100644 --- a/codex-rs/app-server/tests/suite/v2/review.rs +++ b/codex-rs/app-server/tests/suite/v2/review.rs @@ -17,6 +17,7 @@ use codex_app_server_protocol::ReviewStartParams; use codex_app_server_protocol::ReviewStartResponse; use codex_app_server_protocol::ReviewTarget; use codex_app_server_protocol::ServerRequest; +use codex_app_server_protocol::ThreadHistoryMode; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; @@ -37,6 +38,54 @@ const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs const INVALID_REQUEST_ERROR_CODE: i64 = -32600; const COLLIDING_REVIEW_SKILL_MARKER: &str = "COLLIDING_REVIEW_SKILL_MARKER"; +#[tokio::test] +async fn review_start_rejects_detached_delivery_for_paginated_parent() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .build() + .await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + let start_id = mcp + .send_thread_start_request(ThreadStartParams { + history_mode: Some(ThreadHistoryMode::Paginated), + ..Default::default() + }) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response(start_resp)?; + + let review_id = mcp + .send_review_start_request(ReviewStartParams { + thread_id: thread.id, + delivery: Some(ReviewDelivery::Detached), + target: ReviewTarget::Custom { + instructions: "detached review".to_string(), + }, + }) + .await?; + let review_err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(review_id)), + ) + .await??; + assert_eq!(review_err.error.code, -32600); + assert_eq!( + review_err.error.message, + "paginated threads do not support detached review" + ); + + Ok(()) +} + #[tokio::test] async fn review_start_runs_review_turn_and_emits_code_review_item() -> Result<()> { let review_payload = json!({ diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index 191d8f9b69..f80c7dce56 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -1,5 +1,6 @@ use anyhow::Result; use app_test_support::TestAppServer; +use app_test_support::create_fake_paginated_rollout; use app_test_support::create_fake_rollout_with_text_elements; use app_test_support::create_mock_responses_server_repeating_assistant; use app_test_support::rollout_path; @@ -223,19 +224,14 @@ async fn paginated_stored_thread_routes_projected_turns_and_rejects_legacy_histo let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; - let conversation_id = create_fake_rollout_with_text_elements( + let conversation_id = create_fake_paginated_rollout( codex_home.path(), "2025-01-05T12-00-00", "2025-01-05T12:00:00Z", "Saved user message", - Vec::new(), Some("mock_provider"), /*git_info*/ None, )?; - set_rollout_history_mode( - rollout_path(codex_home.path(), "2025-01-05T12-00-00", &conversation_id).as_path(), - ThreadHistoryMode::Paginated, - )?; let mut mcp = TestAppServer::builder() .with_codex_home(codex_home.path()) @@ -293,12 +289,15 @@ async fn paginated_stored_thread_routes_projected_turns_and_rejects_legacy_histo include_turns: true, }) .await?; - assert_paginated_threads_unsupported( - timeout( - DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_error_message(RequestId::Integer(read_id)), - ) - .await??, + let read_err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(read_id)), + ) + .await??; + assert_eq!(read_err.error.code, -32600); + assert_eq!( + read_err.error.message, + "paginated threads do not support thread/read(includeTurns=true)" ); let turns_list_id = mcp @@ -324,20 +323,6 @@ async fn paginated_stored_thread_routes_projected_turns_and_rejects_legacy_histo } ); - let resume_id = mcp - .send_thread_resume_request(ThreadResumeParams { - thread_id: conversation_id, - ..Default::default() - }) - .await?; - assert_paginated_threads_unsupported( - timeout( - DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_error_message(RequestId::Integer(resume_id)), - ) - .await??, - ); - Ok(()) } @@ -1412,11 +1397,108 @@ async fn paginated_history_lists_use_projected_turns_and_items() -> Result<()> { .await?; timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + let resume_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread_id.to_string(), + exclude_turns: true, + ..Default::default() + }) + .await?; + let resume_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(resume_id)), + ) + .await??; + let ThreadResumeResponse { + thread, + turns_backwards_cursor, + items_backwards_cursor, + .. + } = to_response::(resume_resp)?; + assert!(thread.turns.is_empty()); + let turns_backwards_cursor = + turns_backwards_cursor.expect("resume should return a turn head cursor"); + let items_backwards_cursor = + items_backwards_cursor.expect("resume should return an item head cursor"); + + let rejoin_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread_id.to_string(), + exclude_turns: true, + ..Default::default() + }) + .await?; + let rejoin_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(rejoin_id)), + ) + .await??; + let ThreadResumeResponse { + turns_backwards_cursor: rejoin_turns_backwards_cursor, + items_backwards_cursor: rejoin_items_backwards_cursor, + .. + } = to_response::(rejoin_resp)?; + assert_eq!( + rejoin_turns_backwards_cursor.as_deref(), + Some(turns_backwards_cursor.as_str()) + ); + assert_eq!( + rejoin_items_backwards_cursor.as_deref(), + Some(items_backwards_cursor.as_str()) + ); + + let ThreadTurnsListResponse { data, .. } = read_turns_page( + &mut mcp, + thread_id, + Some(turns_backwards_cursor), + Some(2), + SortDirection::Desc, + Some(TurnItemsView::NotLoaded), + ) + .await?; + assert_eq!( + data.into_iter().map(|turn| turn.id).collect::>(), + vec!["turn-2", "turn-1"] + ); + + let ThreadItemsListResponse { data, .. } = read_items_page( + &mut mcp, + thread_id, + /*turn_id*/ None, + Some(items_backwards_cursor.clone()), + Some(3), + SortDirection::Desc, + ) + .await?; + assert_eq!( + data.into_iter() + .map(|entry| entry.item.id().to_string()) + .collect::>(), + vec!["user-2", "agent-1", "user-1"] + ); + + let ThreadItemsListResponse { data, .. } = read_items_page( + &mut mcp, + thread_id, + Some("turn-1"), + Some(items_backwards_cursor), + Some(2), + SortDirection::Desc, + ) + .await?; + assert_eq!( + data.into_iter() + .map(|entry| entry.item.id().to_string()) + .collect::>(), + vec!["agent-1", "user-1"] + ); + let first_page = read_turns_page( &mut mcp, thread_id, /*cursor*/ None, Some(1), + SortDirection::Asc, Some(TurnItemsView::Summary), ) .await?; @@ -1451,6 +1533,7 @@ async fn paginated_history_lists_use_projected_turns_and_items() -> Result<()> { thread_id, Some(next_cursor), Some(1), + SortDirection::Asc, Some(TurnItemsView::NotLoaded), ) .await?; @@ -1488,15 +1571,25 @@ async fn paginated_history_lists_use_projected_turns_and_items() -> Result<()> { "thread/turns/list itemsView full is not supported for paginated threads; use thread/items/list" ); - let first_items_page = read_items_page(&mut mcp, thread_id, /*cursor*/ None, Some(1)).await?; + let first_items_page = read_items_page( + &mut mcp, + thread_id, + /*turn_id*/ None, + /*cursor*/ None, + Some(1), + SortDirection::Asc, + ) + .await?; assert_eq!(first_items_page.data.len(), 1); assert_eq!(first_items_page.data[0].turn_id, "turn-1"); assert_eq!(first_items_page.data[0].item.id(), "user-1"); let second_items_page = read_items_page( &mut mcp, thread_id, + /*turn_id*/ None, Some(first_items_page.next_cursor.expect("next item cursor")), Some(1), + SortDirection::Asc, ) .await?; assert_eq!(second_items_page.data.len(), 1); @@ -1505,8 +1598,10 @@ async fn paginated_history_lists_use_projected_turns_and_items() -> Result<()> { let third_items_page = read_items_page( &mut mcp, thread_id, + /*turn_id*/ None, Some(second_items_page.next_cursor.expect("next item cursor")), Some(1), + SortDirection::Asc, ) .await?; assert_eq!(third_items_page.data.len(), 1); @@ -1706,6 +1801,7 @@ async fn read_turns_page( thread_id: codex_protocol::ThreadId, cursor: Option, limit: Option, + sort_direction: SortDirection, items_view: Option, ) -> Result { let request_id = mcp @@ -1713,7 +1809,7 @@ async fn read_turns_page( thread_id: thread_id.to_string(), cursor, limit, - sort_direction: Some(SortDirection::Asc), + sort_direction: Some(sort_direction), items_view, }) .await?; @@ -1728,16 +1824,18 @@ async fn read_turns_page( async fn read_items_page( mcp: &mut TestAppServer, thread_id: codex_protocol::ThreadId, + turn_id: Option<&str>, cursor: Option, limit: Option, + sort_direction: SortDirection, ) -> Result { let request_id = mcp .send_thread_items_list_request(ThreadItemsListParams { thread_id: thread_id.to_string(), - turn_id: None, + turn_id: turn_id.map(str::to_string), cursor, limit, - sort_direction: Some(SortDirection::Asc), + sort_direction: Some(sort_direction), }) .await?; let response: JSONRPCResponse = timeout( @@ -1879,26 +1977,6 @@ fn store_history_items() -> Vec { ))] } -fn set_rollout_history_mode(path: &Path, history_mode: ThreadHistoryMode) -> Result<()> { - let mut lines = std::fs::read_to_string(path)? - .lines() - .map(serde_json::from_str::) - .collect::, _>>()?; - lines[0]["payload"]["history_mode"] = serde_json::to_value(history_mode)?; - let contents = lines - .into_iter() - .map(|line| line.to_string()) - .collect::>() - .join("\n"); - std::fs::write(path, format!("{contents}\n"))?; - Ok(()) -} - -fn assert_paginated_threads_unsupported(err: JSONRPCError) { - assert_eq!(err.error.code, -32601); - assert_eq!(err.error.message, "paginated_threads is not supported yet"); -} - fn create_config_toml_with_thread_store(codex_home: &Path, store_id: &str) -> std::io::Result<()> { let config_toml = codex_home.join("config.toml"); std::fs::write( 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 352d96409f..505c9c6856 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -2,6 +2,7 @@ use anyhow::Result; use app_test_support::ChatGptAuthFixture; use app_test_support::TestAppServer; use app_test_support::create_apply_patch_sse_response; +use app_test_support::create_fake_paginated_rollout; use app_test_support::create_fake_rollout; use app_test_support::create_fake_rollout_with_text_elements; use app_test_support::create_fake_rollout_with_token_usage; @@ -34,6 +35,7 @@ use codex_app_server_protocol::SessionSource; use codex_app_server_protocol::ThreadGoalClearResponse; use codex_app_server_protocol::ThreadGoalSetResponse; use codex_app_server_protocol::ThreadGoalStatus; +use codex_app_server_protocol::ThreadHistoryMode; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadListResponse; use codex_app_server_protocol::ThreadMetadataGitInfoUpdateParams; @@ -124,6 +126,91 @@ const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); const CODEX_5_2_INSTRUCTIONS_TEMPLATE_DEFAULT: &str = "You are Codex, a coding agent based on GPT-5. You and the user share the same workspace and collaborate to achieve the user's goals."; +#[tokio::test] +async fn thread_resume_paginated_uses_model_context_without_history() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + let conversation_id = create_fake_paginated_rollout( + codex_home.path(), + "2025-01-05T12-00-00", + "2025-01-05T12:00:00Z", + "Saved user message", + Some("mock_provider"), + /*git_info*/ None, + )?; + + let mut primary = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .build() + .await?; + timeout(DEFAULT_READ_TIMEOUT, primary.initialize()).await??; + + let full_resume_id = primary + .send_thread_resume_request(ThreadResumeParams { + thread_id: conversation_id.clone(), + ..Default::default() + }) + .await?; + let full_resume_err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + primary.read_stream_until_error_message(RequestId::Integer(full_resume_id)), + ) + .await??; + assert_eq!(full_resume_err.error.code, -32600); + assert_eq!( + full_resume_err.error.message, + "paginated threads do not support full-history thread/resume; pass excludeTurns=true" + ); + + let initial_page_resume_id = primary + .send_thread_resume_request(ThreadResumeParams { + thread_id: conversation_id.clone(), + exclude_turns: true, + initial_turns_page: Some(ThreadResumeInitialTurnsPageParams { + limit: None, + sort_direction: None, + items_view: None, + }), + ..Default::default() + }) + .await?; + let initial_page_err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + primary.read_stream_until_error_message(RequestId::Integer(initial_page_resume_id)), + ) + .await??; + assert_eq!(initial_page_err.error.code, -32600); + assert_eq!( + initial_page_err.error.message, + "paginated threads do not support initialTurnsPage; use turnsBackwardsCursor and itemsBackwardsCursor" + ); + + // LocalThreadStore rejects paginated full-history reads, so this cold resume only succeeds + // when app-server asks for the bounded latest model context. + let resume_id = primary + .send_thread_resume_request(ThreadResumeParams { + thread_id: conversation_id.clone(), + exclude_turns: true, + ..Default::default() + }) + .await?; + let resume_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + primary.read_stream_until_response_message(RequestId::Integer(resume_id)), + ) + .await??; + let ThreadResumeResponse { + thread: resumed, .. + } = to_response::(resume_resp)?; + assert_eq!(resumed.id, conversation_id); + assert_eq!(resumed.history_mode, ThreadHistoryMode::Paginated); + assert!(resumed.turns.is_empty()); + + Ok(()) +} + fn normalized_existing_path(path: impl AsRef) -> Result { Ok(AbsolutePathBuf::from_absolute_path(path.as_ref().canonicalize()?)?.into_path_buf()) } diff --git a/codex-rs/app-server/tests/suite/v2/thread_rollback.rs b/codex-rs/app-server/tests/suite/v2/thread_rollback.rs index 53f549c2e2..a9237f9a9a 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_rollback.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_rollback.rs @@ -1,13 +1,16 @@ use anyhow::Result; use app_test_support::TestAppServer; use app_test_support::create_final_assistant_message_sse_response; +use app_test_support::create_mock_responses_server_repeating_assistant; use app_test_support::create_mock_responses_server_sequence_unchecked; use app_test_support::to_response; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::DeprecationNoticeNotification; +use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCMessage; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadHistoryMode; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; @@ -25,6 +28,51 @@ use tokio::time::timeout; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +#[tokio::test] +async fn thread_rollback_rejects_paginated_thread() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .build() + .await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + let start_id = mcp + .send_thread_start_request(ThreadStartParams { + history_mode: Some(ThreadHistoryMode::Paginated), + ..Default::default() + }) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response(start_resp)?; + + let rollback_id = mcp + .send_thread_rollback_request(ThreadRollbackParams { + thread_id: thread.id, + num_turns: 1, + }) + .await?; + let rollback_err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(rollback_id)), + ) + .await??; + assert_eq!(rollback_err.error.code, -32600); + assert_eq!( + rollback_err.error.message, + "paginated threads do not support thread/rollback" + ); + + Ok(()) +} + #[tokio::test] async fn thread_rollback_does_not_emit_deprecation_notice_to_codex_tui() -> Result<()> { let codex_home = TempDir::new()?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_start.rs b/codex-rs/app-server/tests/suite/v2/thread_start.rs index f6dbc85ba7..cf5cc13332 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_start.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_start.rs @@ -470,7 +470,7 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> { } #[tokio::test] -async fn thread_start_history_mode_accepts_legacy_and_rejects_paginated() -> Result<()> { +async fn thread_start_history_mode_accepts_legacy_and_paginated() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; create_config_toml_without_approval_policy(codex_home.path(), &server.uri())?; @@ -502,17 +502,14 @@ async fn thread_start_history_mode_accepts_legacy_and_rejects_paginated() -> Res ..Default::default() }) .await?; - let error: JSONRPCError = timeout( + let response: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_error_message(RequestId::Integer(request_id)), + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; + let ThreadStartResponse { thread, .. } = to_response(response)?; - assert_eq!(error.error.code, -32601); - assert_eq!( - error.error.message, - "paginated_threads is not supported yet" - ); + assert_eq!(thread.history_mode, ThreadHistoryMode::Paginated); Ok(()) } diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index cbaccceb36..63b04be1fc 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -39,7 +39,6 @@ use crate::ThreadStoreFuture; use crate::ThreadStoreResult; use crate::UpdateThreadMetadataParams; use crate::error::reject_paginated_history_mode; -use crate::types::canonical_history_mode_from_rollout_items; static IN_MEMORY_THREAD_STORES: OnceLock>>> = OnceLock::new(); @@ -200,7 +199,7 @@ mod tests { } #[tokio::test] - async fn paginated_threads_allow_metadata_reads_and_reject_legacy_history_paths() { + async fn paginated_threads_allow_metadata_reads_and_resume_but_reject_legacy_history_paths() { let store = InMemoryThreadStore::default(); let thread_id = ThreadId::default(); let rollout_path = PathBuf::from("/tmp/paginated-thread.jsonl"); @@ -287,18 +286,16 @@ mod tests { .await .expect_err("history load should fail"), ); - assert_paginated_threads_unsupported( - store - .resume_thread(ResumeThreadParams { - thread_id, - rollout_path: None, - history: None, - include_archived: false, - metadata: thread_metadata(), - }) - .await - .expect_err("resume should fail"), - ); + store + .resume_thread(ResumeThreadParams { + thread_id, + rollout_path: None, + history: None, + include_archived: false, + metadata: thread_metadata(), + }) + .await + .expect("resume should succeed"); assert_paginated_threads_unsupported( store .create_thread(create_thread_params( @@ -463,13 +460,6 @@ impl InMemoryThreadStore { async fn resume_thread(&self, params: ResumeThreadParams) -> ThreadStoreResult<()> { let mut state = self.state.lock().await; state.calls.resume_thread += 1; - let history_mode = params - .history - .as_deref() - .map(Vec::as_slice) - .map(canonical_history_mode_from_rollout_items) - .unwrap_or_else(|| history_mode_from_state(&state, params.thread_id)); - reject_paginated_history_mode(history_mode)?; if let Some(history) = params.history { state .histories diff --git a/codex-rs/thread-store/src/local/live_writer.rs b/codex-rs/thread-store/src/local/live_writer.rs index a6535a77bc..66e01243ce 100644 --- a/codex-rs/thread-store/src/local/live_writer.rs +++ b/codex-rs/thread-store/src/local/live_writer.rs @@ -18,7 +18,6 @@ use crate::ReadThreadParams; use crate::ResumeThreadParams; use crate::ThreadStoreError; use crate::ThreadStoreResult; -use crate::error::reject_paginated_history_mode; use crate::types::canonical_history_mode_from_rollout_items; const ROLLOUT_SIZE_BYTES_METRIC: &str = "codex.rollout.size_bytes"; @@ -66,7 +65,6 @@ pub(super) async fn resume_thread( .await? .history_mode }; - reject_paginated_history_mode(history_mode)?; let rollout_path = match (params.rollout_path, params.history) { (Some(rollout_path), _history) => rollout_path, (None, history) => { diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 4671e2ba74..f0074609a7 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -344,6 +344,10 @@ impl ThreadStore for LocalThreadStore { Box::pin(async move { list_threads::list_threads(self, params).await }) } + fn supports_paginated_history_lists(&self) -> bool { + true + } + fn list_turns(&self, params: ListTurnsParams) -> ThreadStoreFuture<'_, TurnPage> { Box::pin(LocalThreadStore::list_turns(self, params)) } @@ -1244,7 +1248,7 @@ mod tests { } #[tokio::test] - async fn paginated_threads_allow_metadata_reads_and_reject_legacy_history_paths() { + async fn paginated_threads_allow_metadata_reads_and_resume_but_reject_legacy_history_paths() { let home = TempDir::new().expect("temp dir"); let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); let uuid = uuid::Uuid::from_u128(408); @@ -1308,18 +1312,16 @@ mod tests { .await .expect_err("history load should fail"), ); - assert_paginated_threads_unsupported( - store - .resume_thread(ResumeThreadParams { - thread_id, - rollout_path: Some(rollout_path), - history: None, - include_archived: false, - metadata: thread_metadata(), - }) - .await - .expect_err("resume should fail"), - ); + store + .resume_thread(ResumeThreadParams { + thread_id, + rollout_path: Some(rollout_path), + history: None, + include_archived: false, + metadata: thread_metadata(), + }) + .await + .expect("resume should succeed"); } #[tokio::test] diff --git a/codex-rs/thread-store/src/local/test_support.rs b/codex-rs/thread-store/src/local/test_support.rs index 3537555ee4..24f1e41769 100644 --- a/codex-rs/thread-store/src/local/test_support.rs +++ b/codex-rs/thread-store/src/local/test_support.rs @@ -89,7 +89,7 @@ pub(super) fn write_session_file_with_fork( fs::create_dir_all(&day_dir)?; let path = day_dir.join(format!("rollout-{ts}-{uuid}.jsonl")); let mut file = fs::File::create(&path)?; - let meta = serde_json::json!({ + let mut meta = serde_json::json!({ "timestamp": ts, "type": "session_meta", "payload": { @@ -110,6 +110,9 @@ pub(super) fn write_session_file_with_fork( } }, }); + if matches!(history_mode, ThreadHistoryMode::Paginated) { + meta["ordinal"] = serde_json::json!(0); + } writeln!(file, "{meta}")?; if matches!(history_mode, ThreadHistoryMode::Legacy) { let user_event = serde_json::json!({ diff --git a/codex-rs/thread-store/src/local/thread_history/read.rs b/codex-rs/thread-store/src/local/thread_history/read.rs index de36e3ed4c..79b9a16ec5 100644 --- a/codex-rs/thread-store/src/local/thread_history/read.rs +++ b/codex-rs/thread-store/src/local/thread_history/read.rs @@ -38,7 +38,7 @@ struct HistoryCursor { #[serde(tag = "kind", rename_all = "camelCase")] enum CursorScope { Turns, - Items { turn_id: Option }, + Items, } struct StoredTurnRow { @@ -148,9 +148,7 @@ pub(in crate::local) async fn list_items( "list_items", ) .await?; - let scope = CursorScope::Items { - turn_id: params.turn_id.clone(), - }; + let scope = CursorScope::Items; let cursor = parse_cursor(params.cursor.as_deref(), params.thread_id, &scope)?; let pool = store.thread_history_db().await?; let limit = page_limit(params.page_size)?; diff --git a/codex-rs/thread-store/src/local/thread_history/read_tests.rs b/codex-rs/thread-store/src/local/thread_history/read_tests.rs index cfef95313d..70ee07d519 100644 --- a/codex-rs/thread-store/src/local/thread_history/read_tests.rs +++ b/codex-rs/thread-store/src/local/thread_history/read_tests.rs @@ -184,6 +184,20 @@ async fn list_items_pages_whole_thread_and_per_turn_rows() { .await .expect("turn item page"); assert_eq!(item_ids(&turn_page), vec!["item-5", "item-4"]); + let whole_thread_from_turn_cursor = store + .list_items(item_params( + thread_id, + /*turn_id*/ None, + turn_page.backwards_cursor.clone(), + /*page_size*/ 2, + SortDirection::Desc, + )) + .await + .expect("whole-thread page from turn cursor"); + assert_eq!( + item_ids(&whole_thread_from_turn_cursor), + vec!["item-5", "item-4"] + ); let next_turn_page = store .list_items(item_params( thread_id, diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index 67834ef93d..9d12ad2e32 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/codex-rs/thread-store/src/store.rs @@ -105,6 +105,11 @@ pub trait ThreadStore: Any + Send + Sync { /// Lists stored threads matching the supplied filters. fn list_threads(&self, params: ListThreadsParams) -> ThreadStoreFuture<'_, ThreadPage>; + /// Whether paginated threads can hydrate durable history through turn and item lists. + fn supports_paginated_history_lists(&self) -> bool { + false + } + /// Searches stored threads and returns search-only preview metadata. fn search_threads( &self, diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index 9341cc3db1..3de1088fb6 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -2544,6 +2544,8 @@ mod tests { reasoning_effort: None, multi_agent_mode: Default::default(), initial_turns_page: None, + turns_backwards_cursor: None, + items_backwards_cursor: None, }; let started = started_thread_from_resume_response(