diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index fa64b46d7..48379cc39 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -263,6 +263,7 @@ fn sample_thread_resume_response_with_source( sandbox: AppServerSandboxPolicy::DangerFullAccess, active_permission_profile: None, reasoning_effort: None, + initial_turns_page: None, }) } diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index c6a928afa..d0f8fc40e 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -172,6 +172,7 @@ fn sample_thread_resume_response() -> ClientResponsePayload { sandbox: AppServerSandboxPolicy::DangerFullAccess, active_permission_profile: None, reasoning_effort: None, + initial_turns_page: None, }) } diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 63e8fbde1..a09d793bf 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -3543,6 +3543,42 @@ } ] }, + "ThreadResumeInitialTurnsPageParams": { + "properties": { + "itemsView": { + "anyOf": [ + { + "$ref": "#/definitions/TurnItemsView" + }, + { + "type": "null" + } + ], + "description": "How much item detail to include for each returned turn; defaults to summary." + }, + "limit": { + "description": "Optional turn page size.", + "format": "uint32", + "minimum": 0.0, + "type": [ + "integer", + "null" + ] + }, + "sortDirection": { + "anyOf": [ + { + "$ref": "#/definitions/SortDirection" + }, + { + "type": "null" + } + ], + "description": "Optional turn pagination direction; defaults to descending." + } + }, + "type": "object" + }, "ThreadResumeParams": { "description": "There are three ways to resume a thread: 1. By thread_id: load the thread from disk by thread_id and resume it. 2. By history: instantiate the thread from memory and resume it. 3. By path: load the thread from disk by path and resume it.\n\nFor non-running threads, the precedence is: history > non-empty path > thread_id. If using history or a non-empty path for a non-running thread, the thread_id param will be ignored.\n\nIf thread_id identifies a running thread, app-server rejoins that thread and treats a non-empty path as a consistency check against the active rollout path. Empty string path values are treated as absent.\n\nPrefer using thread_id whenever possible.", "properties": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 28dc006b5..1d1a64a2f 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -17150,6 +17150,42 @@ "title": "ThreadRealtimeTranscriptDoneNotification", "type": "object" }, + "ThreadResumeInitialTurnsPageParams": { + "properties": { + "itemsView": { + "anyOf": [ + { + "$ref": "#/definitions/v2/TurnItemsView" + }, + { + "type": "null" + } + ], + "description": "How much item detail to include for each returned turn; defaults to summary." + }, + "limit": { + "description": "Optional turn page size.", + "format": "uint32", + "minimum": 0.0, + "type": [ + "integer", + "null" + ] + }, + "sortDirection": { + "anyOf": [ + { + "$ref": "#/definitions/v2/SortDirection" + }, + { + "type": "null" + } + ], + "description": "Optional turn pagination direction; defaults to descending." + } + }, + "type": "object" + }, "ThreadResumeParams": { "$schema": "http://json-schema.org/draft-07/schema#", "description": "There are three ways to resume a thread: 1. By thread_id: load the thread from disk by thread_id and resume it. 2. By history: instantiate the thread from memory and resume it. 3. By path: load the thread from disk by path and resume it.\n\nFor non-running threads, the precedence is: history > non-empty path > thread_id. If using history or a non-empty path for a non-running thread, the thread_id param will be ignored.\n\nIf thread_id identifies a running thread, app-server rejoins that thread and treats a non-empty path as a consistency check against the active rollout path. Empty string path values are treated as absent.\n\nPrefer using thread_id whenever possible.", @@ -18488,6 +18524,32 @@ "title": "TurnSteerResponse", "type": "object" }, + "TurnsPage": { + "properties": { + "backwardsCursor": { + "type": [ + "string", + "null" + ] + }, + "data": { + "items": { + "$ref": "#/definitions/v2/Turn" + }, + "type": "array" + }, + "nextCursor": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "data" + ], + "type": "object" + }, "UserInput": { "oneOf": [ { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index d1c59e985..1b49b3864 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -14974,6 +14974,42 @@ "title": "ThreadRealtimeTranscriptDoneNotification", "type": "object" }, + "ThreadResumeInitialTurnsPageParams": { + "properties": { + "itemsView": { + "anyOf": [ + { + "$ref": "#/definitions/TurnItemsView" + }, + { + "type": "null" + } + ], + "description": "How much item detail to include for each returned turn; defaults to summary." + }, + "limit": { + "description": "Optional turn page size.", + "format": "uint32", + "minimum": 0.0, + "type": [ + "integer", + "null" + ] + }, + "sortDirection": { + "anyOf": [ + { + "$ref": "#/definitions/SortDirection" + }, + { + "type": "null" + } + ], + "description": "Optional turn pagination direction; defaults to descending." + } + }, + "type": "object" + }, "ThreadResumeParams": { "$schema": "http://json-schema.org/draft-07/schema#", "description": "There are three ways to resume a thread: 1. By thread_id: load the thread from disk by thread_id and resume it. 2. By history: instantiate the thread from memory and resume it. 3. By path: load the thread from disk by path and resume it.\n\nFor non-running threads, the precedence is: history > non-empty path > thread_id. If using history or a non-empty path for a non-running thread, the thread_id param will be ignored.\n\nIf thread_id identifies a running thread, app-server rejoins that thread and treats a non-empty path as a consistency check against the active rollout path. Empty string path values are treated as absent.\n\nPrefer using thread_id whenever possible.", @@ -16312,6 +16348,32 @@ "title": "TurnSteerResponse", "type": "object" }, + "TurnsPage": { + "properties": { + "backwardsCursor": { + "type": [ + "string", + "null" + ] + }, + "data": { + "items": { + "$ref": "#/definitions/Turn" + }, + "type": "array" + }, + "nextCursor": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "data" + ], + "type": "object" + }, "UserInput": { "oneOf": [ { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeParams.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeParams.json index 58b85d41a..364cb3661 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeParams.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeParams.json @@ -983,6 +983,74 @@ "danger-full-access" ], "type": "string" + }, + "SortDirection": { + "enum": [ + "asc", + "desc" + ], + "type": "string" + }, + "ThreadResumeInitialTurnsPageParams": { + "properties": { + "itemsView": { + "anyOf": [ + { + "$ref": "#/definitions/TurnItemsView" + }, + { + "type": "null" + } + ], + "description": "How much item detail to include for each returned turn; defaults to summary." + }, + "limit": { + "description": "Optional turn page size.", + "format": "uint32", + "minimum": 0.0, + "type": [ + "integer", + "null" + ] + }, + "sortDirection": { + "anyOf": [ + { + "$ref": "#/definitions/SortDirection" + }, + { + "type": "null" + } + ], + "description": "Optional turn pagination direction; defaults to descending." + } + }, + "type": "object" + }, + "TurnItemsView": { + "oneOf": [ + { + "description": "`items` was not loaded for this turn. The field is intentionally empty.", + "enum": [ + "notLoaded" + ], + "type": "string" + }, + { + "description": "`items` contains only a display summary for this turn.", + "enum": [ + "summary" + ], + "type": "string" + }, + { + "description": "`items` contains every ThreadItem available from persisted app-server history for this turn.", + "enum": [ + "full" + ], + "type": "string" + } + ] } }, "description": "There are three ways to resume a thread: 1. By thread_id: load the thread from disk by thread_id and resume it. 2. By history: instantiate the thread from memory and resume it. 3. By path: load the thread from disk by path and resume it.\n\nFor non-running threads, the precedence is: history > non-empty path > thread_id. If using history or a non-empty path for a non-running thread, the thread_id param will be ignored.\n\nIf thread_id identifies a running thread, app-server rejoins that thread and treats a non-empty path as a consistency check against the active rollout path. Empty string path values are treated as absent.\n\nPrefer using thread_id whenever possible.", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json index b4fa181c8..8f76893af 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json @@ -1995,6 +1995,32 @@ ], "type": "string" }, + "TurnsPage": { + "properties": { + "backwardsCursor": { + "type": [ + "string", + "null" + ] + }, + "data": { + "items": { + "$ref": "#/definitions/Turn" + }, + "type": "array" + }, + "nextCursor": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "data" + ], + "type": "object" + }, "UserInput": { "oneOf": [ { diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadResumeInitialTurnsPageParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadResumeInitialTurnsPageParams.ts new file mode 100644 index 000000000..2dbcd9780 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadResumeInitialTurnsPageParams.ts @@ -0,0 +1,19 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { SortDirection } from "./SortDirection"; +import type { TurnItemsView } from "./TurnItemsView"; + +export type ThreadResumeInitialTurnsPageParams = { +/** + * Optional turn page size. + */ +limit?: number | null, +/** + * Optional turn pagination direction; defaults to descending. + */ +sortDirection?: SortDirection | null, +/** + * How much item detail to include for each returned turn; defaults to summary. + */ +itemsView?: TurnItemsView | null, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/TurnsPage.ts b/codex-rs/app-server-protocol/schema/typescript/v2/TurnsPage.ts new file mode 100644 index 000000000..e91865aee --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/TurnsPage.ts @@ -0,0 +1,6 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { Turn } from "./Turn"; + +export type TurnsPage = { data: Array, nextCursor: string | null, backwardsCursor: string | null, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts index 2a4220cb6..0b7316fc4 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -398,6 +398,7 @@ export type { ThreadRealtimeStartTransport } from "./ThreadRealtimeStartTranspor export type { ThreadRealtimeStartedNotification } from "./ThreadRealtimeStartedNotification"; export type { ThreadRealtimeTranscriptDeltaNotification } from "./ThreadRealtimeTranscriptDeltaNotification"; export type { ThreadRealtimeTranscriptDoneNotification } from "./ThreadRealtimeTranscriptDoneNotification"; +export type { ThreadResumeInitialTurnsPageParams } from "./ThreadResumeInitialTurnsPageParams"; export type { ThreadResumeParams } from "./ThreadResumeParams"; export type { ThreadResumeResponse } from "./ThreadResumeResponse"; export type { ThreadRollbackParams } from "./ThreadRollbackParams"; @@ -450,6 +451,7 @@ export type { TurnStartedNotification } from "./TurnStartedNotification"; export type { TurnStatus } from "./TurnStatus"; export type { TurnSteerParams } from "./TurnSteerParams"; export type { TurnSteerResponse } from "./TurnSteerResponse"; +export type { TurnsPage } from "./TurnsPage"; export type { UserInput } from "./UserInput"; export type { WarningNotification } from "./WarningNotification"; export type { WebSearchAction } from "./WebSearchAction"; 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 8bc01e93d..0465a4e01 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -111,6 +111,85 @@ fn thread_turns_list_params_accepts_items_view() { assert_eq!(params.items_view, Some(TurnItemsView::NotLoaded)); } +#[test] +fn thread_resume_params_accept_turns_page_bootstrap() { + let params = serde_json::from_value::(json!({ + "threadId": "thr_123", + "initialTurnsPage": { + "limit": 25, + "sortDirection": "asc", + "itemsView": "full", + }, + })) + .expect("thread resume params should deserialize"); + + assert_eq!(params.thread_id, "thr_123"); + assert_eq!( + params.initial_turns_page, + Some(ThreadResumeInitialTurnsPageParams { + limit: Some(25), + sort_direction: Some(SortDirection::Asc), + items_view: Some(TurnItemsView::Full), + }) + ); +} + +#[test] +fn thread_resume_response_round_trips_initial_turns_page() { + let response = ThreadResumeResponse { + thread: Thread { + id: "thr_123".to_string(), + session_id: "thr_123".to_string(), + forked_from_id: None, + preview: String::new(), + ephemeral: false, + model_provider: "openai".to_string(), + created_at: 1, + updated_at: 1, + status: ThreadStatus::Idle, + path: None, + cwd: absolute_path("tmp"), + cli_version: "0.0.0".to_string(), + source: SessionSource::Exec, + thread_source: None, + agent_nickname: None, + agent_role: None, + git_info: None, + name: None, + turns: Vec::new(), + }, + model: "gpt-5".to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: absolute_path("tmp"), + runtime_workspace_roots: Vec::new(), + instruction_sources: Vec::new(), + approval_policy: AskForApproval::OnFailure, + approvals_reviewer: ApprovalsReviewer::User, + sandbox: SandboxPolicy::DangerFullAccess, + active_permission_profile: None, + reasoning_effort: None, + initial_turns_page: Some(TurnsPage { + data: Vec::new(), + next_cursor: Some("cursor_next".to_string()), + backwards_cursor: Some("cursor_back".to_string()), + }), + }; + + let value = serde_json::to_value(&response).expect("serialize thread resume response"); + assert_eq!( + value.get("initialTurnsPage"), + Some(&json!({ + "data": [], + "nextCursor": "cursor_next", + "backwardsCursor": "cursor_back", + })) + ); + let decoded = serde_json::from_value::(value) + .expect("deserialize thread resume response"); + assert_eq!(decoded, response); +} + #[test] fn thread_turns_items_list_round_trips() { let params = ThreadTurnsItemsListParams { @@ -3406,6 +3485,7 @@ fn thread_lifecycle_responses_default_missing_optional_fields() { assert_eq!(fork.instruction_sources, Vec::::new()); assert_eq!(start.active_permission_profile, None); assert_eq!(resume.active_permission_profile, None); + assert_eq!(resume.initial_turns_page, None); assert_eq!(fork.active_permission_profile, None); } 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 4971d5f4b..8157fce69 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -397,6 +397,11 @@ pub struct ThreadResumeParams { #[experimental("thread/resume.excludeTurns")] #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub exclude_turns: bool, + /// When present, include a `thread/turns/list` page in the resume response + /// so clients can bootstrap recent turns without a second request. + #[experimental("thread/resume.initialTurnsPage")] + #[ts(optional = nullable)] + pub initial_turns_page: Option, /// Deprecated and ignored by app-server. Kept only so older clients can /// continue sending the field while rollout persistence always uses the /// limited history policy. @@ -435,6 +440,44 @@ pub struct ThreadResumeResponse { #[serde(default)] pub active_permission_profile: Option, pub reasoning_effort: Option, + /// `thread/turns/list` page returned when requested by `initialTurnsPage`. + #[experimental("thread/resume.initialTurnsPage")] + #[serde(default)] + pub initial_turns_page: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadResumeInitialTurnsPageParams { + /// Optional turn page size. + #[ts(optional = nullable)] + pub limit: Option, + /// Optional turn pagination direction; defaults to descending. + #[ts(optional = nullable)] + pub sort_direction: Option, + /// How much item detail to include for each returned turn; defaults to summary. + #[ts(optional = nullable)] + pub items_view: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct TurnsPage { + pub data: Vec, + pub next_cursor: Option, + pub backwards_cursor: Option, +} + +impl From for TurnsPage { + fn from(response: ThreadTurnsListResponse) -> Self { + Self { + data: response.data, + next_cursor: response.next_cursor, + backwards_cursor: response.backwards_cursor, + } + } } #[derive( diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 88edc22f9..d7dfb00fc 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -281,6 +281,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. +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. Example: @@ -297,6 +299,24 @@ Example: "excludeTurns": true } } { "id": 12, "result": { "thread": { "id": "thr_123", "turns": [], … } } } + +{ "method": "thread/resume", "id": 13, "params": { + "threadId": "thr_123", + "excludeTurns": true, + "initialTurnsPage": { + "limit": 20, + "sortDirection": "desc", + "itemsView": "summary" + } +} } +{ "id": 13, "result": { + "thread": { "id": "thr_123", "turns": [], … }, + "initialTurnsPage": { + "data": [ ... ], + "nextCursor": "older-turns-cursor-or-null", + "backwardsCursor": "newer-turns-cursor-or-null" + } +} } ``` To branch from a stored session, call `thread/fork` with the `thread.id`. This creates a new thread id and emits a `thread/started` notification for it. The returned `thread.sessionId` identifies the current live session tree root. Root threads use their own `thread.id` as `thread.sessionId`; stored threads that are not loaded also report their own `thread.id`, because resuming one makes it the root of a new live session tree. When the source history includes persisted token usage, the server also emits `thread/tokenUsage/updated` for the new thread immediately after the response. If the source thread is actively running, the fork snapshots it as if the current turn had been interrupted first. Pass `ephemeral: true` when the fork should stay in-memory only: diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 66ab8e3a1..8408a3f67 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -213,6 +213,7 @@ use codex_app_server_protocol::ThreadRealtimeStartResponse; use codex_app_server_protocol::ThreadRealtimeStartTransport; use codex_app_server_protocol::ThreadRealtimeStopParams; use codex_app_server_protocol::ThreadRealtimeStopResponse; +use codex_app_server_protocol::ThreadResumeInitialTurnsPageParams; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadRollbackParams; 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 985baac91..f72cfb5a6 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -565,8 +565,28 @@ pub(super) async fn handle_pending_thread_resume_request( has_live_in_progress_turn, ); let token_usage_thread = pending.include_turns.then(|| thread.clone()); + let mut initial_turns_page = if let Some(params) = pending.initial_turns_page.as_ref() { + match super::thread_processor::build_thread_resume_initial_turns_page( + &pending.history_items, + thread.status.clone(), + has_live_in_progress_turn, + active_turn, + params, + ) { + Ok(page) => Some(page), + Err(error) => { + outgoing.send_error(request_id, error).await; + return; + } + } + } else { + None + }; if pending.redact_resume_payloads { - redact_thread_resume_payloads(&mut thread); + redact_thread_resume_payloads(&mut thread.turns); + if let Some(initial_turns_page) = initial_turns_page.as_mut() { + redact_thread_resume_payloads(&mut initial_turns_page.data); + } } { @@ -635,6 +655,7 @@ pub(super) async fn handle_pending_thread_resume_request( sandbox, active_permission_profile, reasoning_effort, + initial_turns_page, }; outgoing.send_response(request_id, response).await; // Match cold resume: metadata-only resume should attach the listener without 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 5a2e1a4de..939c3df51 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2241,8 +2241,6 @@ impl ThreadRequestProcessor { sort_direction, items_view, } = params; - let items_view = items_view.unwrap_or(TurnItemsView::Summary); - let thread_uuid = ThreadId::from_string(&thread_id) .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; @@ -2270,60 +2268,20 @@ impl ThreadRequestProcessor { } else { None }; - let mut turns = reconstruct_thread_turns_for_turns_list( + build_thread_turns_page_response( &items, self.thread_watch_manager .loaded_status_for_thread(&thread_uuid.to_string()) .await, has_live_running_thread, active_turn, - ); - for turn in &mut turns { - match items_view { - TurnItemsView::NotLoaded => { - turn.items.clear(); - turn.items_view = TurnItemsView::NotLoaded; - } - TurnItemsView::Summary => { - let first_user_message = turn - .items - .iter() - .find(|item| matches!(item, ThreadItem::UserMessage { .. })) - .cloned(); - let final_agent_message = turn - .items - .iter() - .rev() - .find(|item| matches!(item, ThreadItem::AgentMessage { .. })) - .cloned(); - turn.items = match (first_user_message, final_agent_message) { - (Some(user_message), Some(agent_message)) - if user_message.id() != agent_message.id() => - { - vec![user_message, agent_message] - } - (Some(user_message), _) => vec![user_message], - (None, Some(agent_message)) => vec![agent_message], - (None, None) => Vec::new(), - }; - turn.items_view = TurnItemsView::Summary; - } - TurnItemsView::Full => { - turn.items_view = TurnItemsView::Full; - } - } - } - let page = paginate_thread_turns( - turns, - cursor.as_deref(), - limit, - sort_direction.unwrap_or(SortDirection::Desc), - )?; - Ok(ThreadTurnsListResponse { - data: page.turns, - next_cursor: page.next_cursor, - backwards_cursor: page.backwards_cursor, - }) + ThreadTurnsPageOptions { + cursor: cursor.as_deref(), + limit, + sort_direction: sort_direction.unwrap_or(SortDirection::Desc), + items_view: items_view.unwrap_or(TurnItemsView::Summary), + }, + ) } async fn load_thread_turns_list_history( @@ -2531,6 +2489,7 @@ impl ThreadRequestProcessor { developer_instructions, personality, exclude_turns, + initial_turns_page, persist_extended_history: _persist_extended_history, } = params; let include_turns = !exclude_turns; @@ -2685,8 +2644,28 @@ impl ThreadRequestProcessor { config_snapshot.active_permission_profile, ); let token_usage_thread = include_turns.then(|| thread.clone()); + let mut initial_turns_page = if let Some(params) = initial_turns_page.as_ref() { + match build_thread_resume_initial_turns_page( + &response_history.get_rollout_items(), + thread.status.clone(), + /*has_live_running_thread*/ false, + /*active_turn*/ None, + params, + ) { + Ok(page) => Some(page), + Err(error) => { + self.outgoing.send_error(request_id, error).await; + return Ok(()); + } + } + } else { + None + }; if redact_resume_payloads { - redact_thread_resume_payloads(&mut thread); + redact_thread_resume_payloads(&mut thread.turns); + if let Some(initial_turns_page) = initial_turns_page.as_mut() { + redact_thread_resume_payloads(&mut initial_turns_page.data); + } } let response = ThreadResumeResponse { @@ -2702,6 +2681,7 @@ impl ThreadRequestProcessor { sandbox, active_permission_profile, reasoning_effort: session_configured.reasoning_effort, + initial_turns_page, }; let connection_id = request_id.connection_id; @@ -2926,6 +2906,7 @@ impl ThreadRequestProcessor { emit_thread_goal_update, thread_goal_state_db, include_turns: !params.exclude_turns, + initial_turns_page: params.initial_turns_page.clone(), redact_resume_payloads, }), ); @@ -3706,6 +3687,95 @@ fn parse_thread_turns_cursor(cursor: &str) -> Result { + cursor: Option<&'a str>, + limit: Option, + sort_direction: SortDirection, + items_view: TurnItemsView, +} + +fn build_thread_turns_page_response( + items: &[RolloutItem], + loaded_status: ThreadStatus, + has_live_running_thread: bool, + active_turn: Option, + options: ThreadTurnsPageOptions<'_>, +) -> Result { + let mut turns = reconstruct_thread_turns_for_turns_list( + items, + loaded_status, + has_live_running_thread, + active_turn, + ); + apply_thread_turns_items_view(&mut turns, options.items_view); + let page = paginate_thread_turns(turns, options.cursor, options.limit, options.sort_direction)?; + Ok(ThreadTurnsListResponse { + data: page.turns, + next_cursor: page.next_cursor, + backwards_cursor: page.backwards_cursor, + }) +} + +pub(super) fn build_thread_resume_initial_turns_page( + items: &[RolloutItem], + loaded_status: ThreadStatus, + has_live_running_thread: bool, + active_turn: Option, + params: &ThreadResumeInitialTurnsPageParams, +) -> Result { + build_thread_turns_page_response( + items, + loaded_status, + has_live_running_thread, + active_turn, + ThreadTurnsPageOptions { + cursor: None, + limit: params.limit, + sort_direction: params.sort_direction.unwrap_or(SortDirection::Desc), + items_view: params.items_view.unwrap_or(TurnItemsView::Summary), + }, + ) + .map(Into::into) +} + +fn apply_thread_turns_items_view(turns: &mut [Turn], items_view: TurnItemsView) { + for turn in turns { + match items_view { + TurnItemsView::NotLoaded => { + turn.items.clear(); + turn.items_view = TurnItemsView::NotLoaded; + } + TurnItemsView::Summary => { + let first_user_message = turn + .items + .iter() + .find(|item| matches!(item, ThreadItem::UserMessage { .. })) + .cloned(); + let final_agent_message = turn + .items + .iter() + .rev() + .find(|item| matches!(item, ThreadItem::AgentMessage { .. })) + .cloned(); + turn.items = match (first_user_message, final_agent_message) { + (Some(user_message), Some(agent_message)) + if user_message.id() != agent_message.id() => + { + vec![user_message, agent_message] + } + (Some(user_message), _) => vec![user_message], + (None, Some(agent_message)) => vec![agent_message], + (None, None) => Vec::new(), + }; + turn.items_view = TurnItemsView::Summary; + } + TurnItemsView::Full => { + turn.items_view = TurnItemsView::Full; + } + } + } +} + fn reconstruct_thread_turns_for_turns_list( items: &[RolloutItem], loaded_status: ThreadStatus, diff --git a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs index ab8f52ffc..2fa95b3d2 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs @@ -652,6 +652,7 @@ mod thread_processor_behavior_tests { developer_instructions: None, personality: None, exclude_turns: false, + initial_turns_page: None, persist_extended_history: false, }; let config_snapshot = ThreadConfigSnapshot { diff --git a/codex-rs/app-server/src/request_processors/thread_resume_redaction.rs b/codex-rs/app-server/src/request_processors/thread_resume_redaction.rs index b1e860884..b265e3a63 100644 --- a/codex-rs/app-server/src/request_processors/thread_resume_redaction.rs +++ b/codex-rs/app-server/src/request_processors/thread_resume_redaction.rs @@ -1,6 +1,6 @@ use codex_app_server_protocol::McpToolCallResult; -use codex_app_server_protocol::Thread; use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::Turn; use serde_json::Value as JsonValue; // Temporary bandaid for remote clients: thread/resume can include large MCP and @@ -14,8 +14,8 @@ pub(super) fn should_redact_thread_resume_payloads(client_name: Option<&str>) -> client_name.is_some_and(|client_name| CHATGPT_REMOTE_CLIENT_NAMES.contains(&client_name)) } -pub(super) fn redact_thread_resume_payloads(thread: &mut Thread) { - for turn in &mut thread.turns { +pub(super) fn redact_thread_resume_payloads(turns: &mut [Turn]) { + for turn in turns { turn.items.retain_mut(|item| match item { ThreadItem::McpToolCall { arguments, @@ -55,8 +55,8 @@ mod tests { use codex_app_server_protocol::McpToolCallError; use codex_app_server_protocol::McpToolCallStatus; use codex_app_server_protocol::SessionSource; + use codex_app_server_protocol::Thread; use codex_app_server_protocol::ThreadStatus; - use codex_app_server_protocol::Turn; use codex_app_server_protocol::TurnItemsView; use codex_app_server_protocol::TurnStatus; use codex_utils_absolute_path::test_support::PathBufExt; @@ -100,7 +100,7 @@ mod tests { }, ]); - redact_thread_resume_payloads(&mut thread); + redact_thread_resume_payloads(&mut thread.turns); assert_eq!(thread.turns[0].items.len(), 2); assert_eq!( @@ -146,7 +146,7 @@ mod tests { duration_ms: Some(8), }]); - redact_thread_resume_payloads(&mut thread); + redact_thread_resume_payloads(&mut thread.turns); assert_eq!( thread.turns[0].items[0], diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index 7f864265b..2d932f032 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -35,6 +35,8 @@ pub(crate) struct PendingThreadResumeRequest { pub(crate) emit_thread_goal_update: bool, pub(crate) thread_goal_state_db: Option, pub(crate) include_turns: bool, + pub(crate) initial_turns_page: + Option, pub(crate) redact_resume_payloads: bool, } 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 c8e3e179f..5ba9fa9d2 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -24,6 +24,7 @@ use codex_app_server_protocol::ThreadListResponse; use codex_app_server_protocol::ThreadNameUpdatedNotification; use codex_app_server_protocol::ThreadReadParams; use codex_app_server_protocol::ThreadReadResponse; +use codex_app_server_protocol::ThreadResumeInitialTurnsPageParams; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadSetNameParams; @@ -631,6 +632,77 @@ async fn thread_read_can_return_archived_threads_by_id() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_resume_initial_turns_page_matches_requested_turns_list_page() -> 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 filename_ts = "2025-01-05T12-00-00"; + let conversation_id = create_fake_rollout_with_text_elements( + codex_home.path(), + filename_ts, + "2025-01-05T12:00:00Z", + "first", + vec![], + Some("mock_provider"), + /*git_info*/ None, + )?; + let rollout_path = rollout_path(codex_home.path(), filename_ts, &conversation_id); + append_user_message(rollout_path.as_path(), "2025-01-05T12:01:00Z", "second")?; + append_user_message(rollout_path.as_path(), "2025-01-05T12:02:00Z", "third")?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let turns_list_id = mcp + .send_thread_turns_list_request(ThreadTurnsListParams { + thread_id: conversation_id.clone(), + cursor: None, + limit: Some(2), + sort_direction: Some(SortDirection::Asc), + items_view: Some(TurnItemsView::NotLoaded), + }) + .await?; + let turns_list_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turns_list_id)), + ) + .await??; + let expected_page = to_response::(turns_list_resp)?; + + let resume_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: conversation_id, + exclude_turns: true, + initial_turns_page: Some(ThreadResumeInitialTurnsPageParams { + limit: Some(2), + sort_direction: Some(SortDirection::Asc), + items_view: Some(TurnItemsView::NotLoaded), + }), + ..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, + initial_turns_page, + .. + } = to_response::(resume_resp)?; + + assert!(thread.turns.is_empty()); + assert_eq!( + initial_turns_page, + Some(codex_app_server_protocol::TurnsPage::from(expected_page)) + ); + + Ok(()) +} + #[tokio::test] async fn thread_turns_list_rejects_cursor_when_anchor_turn_is_rolled_back() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; 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 bb2de2cfb..2b8542724 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -37,6 +37,7 @@ use codex_app_server_protocol::ThreadMetadataGitInfoUpdateParams; use codex_app_server_protocol::ThreadMetadataUpdateParams; use codex_app_server_protocol::ThreadReadParams; use codex_app_server_protocol::ThreadReadResponse; +use codex_app_server_protocol::ThreadResumeInitialTurnsPageParams; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadSource; @@ -530,48 +531,59 @@ async fn thread_resume_returns_rollout_history() -> Result<()> { #[tokio::test] async fn thread_resume_redacts_payloads_for_chatgpt_remote_clients() -> Result<()> { for client_name in ["codex_chatgpt_android_remote", "codex_chatgpt_ios_remote"] { - let remote_thread = resume_redaction_fixture(Some(client_name)).await?; - let remote_turn = remote_thread + let remote_resume = resume_redaction_fixture(Some(client_name)).await?; + let remote_turn = remote_resume + .thread .turns .first() .expect("remote resume should include a turn"); - let remote_mcp_item = remote_turn - .items - .iter() - .find(|item| matches!(item, ThreadItem::McpToolCall { .. })) - .expect("remote resume should include redacted MCP item"); - let ThreadItem::McpToolCall { - arguments, - result, - error, - .. - } = remote_mcp_item - else { - unreachable!("matched MCP item"); - }; - assert_eq!(arguments, &json!("[redacted]")); - let result = result.as_ref().expect("redacted MCP result"); - assert_eq!( - result.content, - vec![json!({ - "type": "text", - "text": "[redacted]", - })] - ); - assert_eq!(result.structured_content, None); - assert_eq!(result.meta, None); - assert_eq!(error, &None); - assert!( - !remote_turn + let remote_page_turn = remote_resume + .initial_turns_page + .as_ref() + .expect("remote resume should include the requested initial turns page") + .data + .first() + .expect("remote initial turns page should include a turn"); + for remote_turn in [remote_turn, remote_page_turn] { + let remote_mcp_item = remote_turn .items .iter() - .any(|item| matches!(item, ThreadItem::ImageGeneration { .. })), - "remote resume should drop image generation items for {client_name}" - ); + .find(|item| matches!(item, ThreadItem::McpToolCall { .. })) + .expect("remote resume should include redacted MCP item"); + let ThreadItem::McpToolCall { + arguments, + result, + error, + .. + } = remote_mcp_item + else { + unreachable!("matched MCP item"); + }; + assert_eq!(arguments, &json!("[redacted]")); + let result = result.as_ref().expect("redacted MCP result"); + assert_eq!( + result.content, + vec![json!({ + "type": "text", + "text": "[redacted]", + })] + ); + assert_eq!(result.structured_content, None); + assert_eq!(result.meta, None); + assert_eq!(error, &None); + assert!( + !remote_turn + .items + .iter() + .any(|item| matches!(item, ThreadItem::ImageGeneration { .. })), + "remote resume should drop image generation items for {client_name}" + ); + } } - let normal_thread = resume_redaction_fixture(Some("some_other_client")).await?; - let normal_turn = normal_thread + let normal_resume = resume_redaction_fixture(Some("some_other_client")).await?; + let normal_turn = normal_resume + .thread .turns .first() .expect("normal resume should include a turn"); @@ -616,9 +628,7 @@ async fn thread_resume_redacts_payloads_for_chatgpt_remote_clients() -> Result<( Ok(()) } -async fn resume_redaction_fixture( - client_name: Option<&str>, -) -> Result { +async fn resume_redaction_fixture(client_name: Option<&str>) -> Result { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; @@ -658,6 +668,11 @@ async fn resume_redaction_fixture( let resume_id = mcp .send_thread_resume_request(ThreadResumeParams { thread_id: conversation_id, + initial_turns_page: Some(ThreadResumeInitialTurnsPageParams { + limit: None, + sort_direction: None, + items_view: Some(TurnItemsView::Full), + }), ..Default::default() }) .await?; @@ -666,8 +681,7 @@ async fn resume_redaction_fixture( mcp.read_stream_until_response_message(RequestId::Integer(resume_id)), ) .await??; - let ThreadResumeResponse { thread, .. } = to_response::(resume_resp)?; - Ok(thread) + to_response::(resume_resp) } fn append_resume_redaction_history( @@ -2448,11 +2462,13 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R ..Default::default() }) .await?; - timeout( + let running_turn_resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, primary.read_stream_until_response_message(RequestId::Integer(running_turn_id)), ) .await??; + let TurnStartResponse { turn: running_turn } = + to_response::(running_turn_resp)?; timeout( DEFAULT_READ_TIMEOUT, primary.read_stream_until_notification_message("turn/started"), @@ -2464,6 +2480,11 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R thread_id: thread.id.clone(), model: Some("not-the-running-model".to_string()), cwd: Some("/tmp".to_string()), + initial_turns_page: Some(ThreadResumeInitialTurnsPageParams { + limit: None, + sort_direction: None, + items_view: None, + }), ..Default::default() }) .await?; @@ -2472,9 +2493,23 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R primary.read_stream_until_response_message(RequestId::Integer(resume_id)), ) .await??; - let ThreadResumeResponse { thread, model, .. } = - to_response::(resume_resp)?; + let ThreadResumeResponse { + thread, + model, + initial_turns_page, + .. + } = to_response::(resume_resp)?; assert_eq!(model, "gpt-5.4"); + let initial_turns_page = initial_turns_page.expect("resume should include initial turns page"); + let resumed_running_turn = initial_turns_page + .data + .first() + .expect("resume page should include the running turn"); + assert_eq!(resumed_running_turn.id, running_turn.id); + assert_eq!(resumed_running_turn.items_view, TurnItemsView::Summary); + assert_eq!(resumed_running_turn.status, TurnStatus::InProgress); + assert!(initial_turns_page.backwards_cursor.is_some()); + assert_eq!(initial_turns_page.next_cursor, None); // The running-thread resume response is queued onto the thread listener task. // If the in-flight turn completes before that queued command runs, the response // can legitimately observe the thread as idle. diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index aa2d5a645..79f8e274b 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -2300,6 +2300,7 @@ mod tests { .into(), active_permission_profile: None, reasoning_effort: None, + initial_turns_page: None, }; let started = started_thread_from_resume_response(