diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 6e81beb3a1..1a166bcb81 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -4949,6 +4949,10 @@ ], "type": "object" }, + "ThreadItemOutputDeferral": { + "properties": {}, + "type": "object" + }, "ThreadItemsListAnchor": { "description": "An exclusive item position within the requested visible turn.", "oneOf": [ @@ -4998,6 +5002,17 @@ ], "description": "Opaque continuation cursor or an exclusive item anchor in the requested visible turn. An item anchor requires a non-empty `turnId`; ascending (the default) returns items after it, and descending returns items before it. Continue with the returned string cursor." }, + "deferOutput": { + "anyOf": [ + { + "$ref": "#/definitions/ThreadItemOutputDeferral" + }, + { + "type": "null" + } + ], + "description": "For a single turn, choose which saved outputs to fetch separately by item ID." + }, "limit": { "description": "Optional item page size.", "format": "uint32", 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 9f93f7b98a..1c46e02254 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 @@ -24234,6 +24234,10 @@ ], "type": "object" }, + "ThreadItemOutputDeferral": { + "properties": {}, + "type": "object" + }, "ThreadItemsListAnchor": { "description": "An exclusive item position within the requested visible turn.", "oneOf": [ @@ -24284,6 +24288,17 @@ ], "description": "Opaque continuation cursor or an exclusive item anchor in the requested visible turn. An item anchor requires a non-empty `turnId`; ascending (the default) returns items after it, and descending returns items before it. Continue with the returned string cursor." }, + "deferOutput": { + "anyOf": [ + { + "$ref": "#/definitions/v2/ThreadItemOutputDeferral" + }, + { + "type": "null" + } + ], + "description": "For a single turn, choose which saved outputs to fetch separately by item ID." + }, "limit": { "description": "Optional item page size.", "format": "uint32", 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 8964219b27..0f4fac64b9 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 @@ -21937,6 +21937,10 @@ ], "type": "object" }, + "ThreadItemOutputDeferral": { + "properties": {}, + "type": "object" + }, "ThreadItemsListAnchor": { "description": "An exclusive item position within the requested visible turn.", "oneOf": [ @@ -21987,6 +21991,17 @@ ], "description": "Opaque continuation cursor or an exclusive item anchor in the requested visible turn. An item anchor requires a non-empty `turnId`; ascending (the default) returns items after it, and descending returns items before it. Continue with the returned string cursor." }, + "deferOutput": { + "anyOf": [ + { + "$ref": "#/definitions/ThreadItemOutputDeferral" + }, + { + "type": "null" + } + ], + "description": "For a single turn, choose which saved outputs to fetch separately by item ID." + }, "limit": { "description": "Optional item page size.", "format": "uint32", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadItemsListParams.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadItemsListParams.json index 6fb586ee02..72cde348cc 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadItemsListParams.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadItemsListParams.json @@ -8,6 +8,10 @@ ], "type": "string" }, + "ThreadItemOutputDeferral": { + "properties": {}, + "type": "object" + }, "ThreadItemsListAnchor": { "description": "An exclusive item position within the requested visible turn.", "oneOf": [ @@ -57,6 +61,17 @@ ], "description": "Opaque continuation cursor or an exclusive item anchor in the requested visible turn. An item anchor requires a non-empty `turnId`; ascending (the default) returns items after it, and descending returns items before it. Continue with the returned string cursor." }, + "deferOutput": { + "anyOf": [ + { + "$ref": "#/definitions/ThreadItemOutputDeferral" + }, + { + "type": "null" + } + ], + "description": "For a single turn, choose which saved outputs to fetch separately by item ID." + }, "limit": { "description": "Optional item page size.", "format": "uint32", diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index 778dc980cf..0a64014227 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 114085e8f3..3ba1e80d21 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemEntry.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemEntry.ts index 5815a3afcf..184a01781a 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemEntry.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemEntry.ts @@ -3,16 +3,13 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. import type { ThreadItem } from "./ThreadItem"; -export type ThreadItemEntry = { -/** +export type ThreadItemEntry = {/** * Turn containing this item. */ -turnId: string, item: ThreadItem, -/** +turnId: string, item: ThreadItem, /** * Unix timestamp (milliseconds) when the item started, if recorded by the producer. */ -startedAtMs: number | null, -/** +startedAtMs: number | null, /** * Unix timestamp (milliseconds) when the item completed, if recorded by the producer. */ -completedAtMs: number | null, }; +completedAtMs: number | null}; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemOutputDeferral.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemOutputDeferral.ts new file mode 100644 index 0000000000..80acf3323a --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemOutputDeferral.ts @@ -0,0 +1,5 @@ +// 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. + +export type ThreadItemOutputDeferral = {}; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemsListParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemsListParams.ts index fe2cce685e..0555704a00 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemsListParams.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemsListParams.ts @@ -2,6 +2,7 @@ // 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 { ThreadItemOutputDeferral } from "./ThreadItemOutputDeferral"; import type { ThreadItemsListCursor } from "./ThreadItemsListCursor"; export type ThreadItemsListParams = { threadId: string, @@ -22,4 +23,8 @@ limit?: number | null, /** * Optional item pagination direction; defaults to ascending. */ -sortDirection?: SortDirection | null, }; +sortDirection?: SortDirection | null, +/** + * For a single turn, choose which saved outputs to fetch separately by item ID. + */ +deferOutput?: ThreadItemOutputDeferral | 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 b7bca86b20..4667924b9e 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -546,6 +546,7 @@ export type { ThreadInjectItemsParams } from "./ThreadInjectItemsParams"; export type { ThreadInjectItemsResponse } from "./ThreadInjectItemsResponse"; export type { ThreadItem } from "./ThreadItem"; export type { ThreadItemEntry } from "./ThreadItemEntry"; +export type { ThreadItemOutputDeferral } from "./ThreadItemOutputDeferral"; export type { ThreadItemsListAnchor } from "./ThreadItemsListAnchor"; export type { ThreadItemsListCursor } from "./ThreadItemsListCursor"; export type { ThreadItemsListParams } from "./ThreadItemsListParams"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 210142ffc6..73ce890e55 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -896,6 +896,7 @@ client_request_definitions! { }, ThreadItemsList => "thread/items/list" { params: v2::ThreadItemsListParams, + inspect_params: true, // Explicitly concurrent: this primarily reads append-only rollout storage. serialization: None, response: v2::ThreadItemsListResponse, @@ -2757,6 +2758,7 @@ mod tests { cursor: None, limit: None, sort_direction: None, + defer_output: None, }, }; assert_eq!(thread_items_list.serialization_scope(), 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 6990a4cf07..546296b610 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -421,6 +421,7 @@ fn thread_items_list_round_trips() { cursor: Some(ThreadItemsListCursor::Opaque("cursor_1".to_string())), limit: Some(50), sort_direction: Some(SortDirection::Asc), + defer_output: None, }; assert_eq!( @@ -462,6 +463,7 @@ fn thread_items_list_round_trips() { let response = ThreadItemsListResponse { data: vec![ThreadItemEntry { turn_id: "turn_456".to_string(), + output_deferred: false, item: ThreadItem { payload: ThreadItemPayload::ContextCompaction { id: "item_1".to_string(), @@ -501,6 +503,7 @@ fn thread_items_list_round_trips() { cursor: None, limit: None, sort_direction: None, + defer_output: None, }; assert_eq!( 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 510f42a5d3..9e2dc9ebf9 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -1837,7 +1837,7 @@ pub struct ThreadTurnsListResponse { pub backwards_cursor: Option, } -#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS, ExperimentalApi)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] pub struct ThreadItemsListParams { @@ -1856,6 +1856,21 @@ pub struct ThreadItemsListParams { /// Optional item pagination direction; defaults to ascending. #[ts(optional = nullable)] pub sort_direction: Option, + /// For a single turn, choose which saved outputs to fetch separately by item ID. + #[experimental(nested)] + #[serde(skip_serializing_if = "Option::is_none")] + #[ts(optional = nullable)] + pub defer_output: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS, ExperimentalApi)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadItemOutputDeferral { + /// Defer nonempty output from successful commands. + #[experimental("thread/items/list.deferOutput.commandExecution")] + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub command_execution: bool, } /// Starting position for an item-history page. @@ -1879,13 +1894,17 @@ pub enum ThreadItemsListAnchor { }, } -#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS, ExperimentalApi)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] pub struct ThreadItemEntry { /// Turn containing this item. pub turn_id: String, pub item: ThreadItem, + /// The saved command output was omitted. Fetch the complete item with `thread/items/read`. + #[experimental("thread/items/list.outputDeferred")] + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub output_deferred: bool, /// Unix timestamp (milliseconds) when the item started, if recorded by the producer. #[ts(type = "number | null")] pub started_at_ms: Option, @@ -1894,13 +1913,30 @@ pub struct ThreadItemEntry { pub completed_at_ms: Option, } -impl From for ThreadItem { - fn from(entry: ThreadItemEntry) -> Self { - Self { +/// Identifies the item to fetch with `thread/items/read` to obtain its full output. +#[derive(Debug, thiserror::Error)] +#[error("output for item {item_id} in turn {turn_id} is deferred")] +pub struct DeferredItemOutput { + pub turn_id: String, + pub item_id: String, +} + +impl TryFrom for ThreadItem { + type Error = DeferredItemOutput; + + fn try_from(entry: ThreadItemEntry) -> Result { + if entry.output_deferred { + return Err(DeferredItemOutput { + item_id: entry.item.id().to_owned(), + turn_id: entry.turn_id, + }); + } + + Ok(Self { started_at_ms: entry.item.started_at_ms.or(entry.started_at_ms), completed_at_ms: entry.item.completed_at_ms.or(entry.completed_at_ms), ..entry.item - } + }) } } diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 0c8241c489..50cd437090 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -718,6 +718,14 @@ Omitted or null cursors preserve normal first-page behavior. Continue anchored pages with the returned opaque string `nextCursor`; response fields and `backwardsCursor` semantics are unchanged. +Experimental clients can set `"deferOutput": {"commandExecution": true}` +with a `turnId`. Omitted or false options keep the full output. Unknown fields in +`deferOutput` are ignored. +For a completed command with a recorded completion time, exit code zero, and +nonempty saved output, the entry returns `outputDeferred: true` and +`item.aggregatedOutput: null`. Use `thread/items/read` with that turn and item ID +to retrieve the full output. All other entries retain their recorded output. + ## Workspace switching (experimental) For a managed ChatGPT login, `accountSession/list` returns an opaque 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 a2da8e7251..b37330deae 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -26,6 +26,7 @@ use crate::thread_sample::generate_sample; use codex_app_server_protocol::SelectedCapabilityRoot; use codex_app_server_protocol::ThreadHistoryMode as ApiThreadHistoryMode; use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::ThreadItemPayload; use codex_app_server_protocol::ThreadItemsListAnchor; use codex_app_server_protocol::ThreadItemsListCursor; use codex_app_server_protocol::ThreadRevertParams; @@ -3728,7 +3729,10 @@ impl ThreadRequestProcessor { cursor, limit, sort_direction, + defer_output, } = params; + let defer_command_output = + turn_id.is_some() && defer_output.is_some_and(|options| options.command_execution); let position = cursor.map(|cursor| match cursor { ThreadItemsListCursor::Opaque(cursor) => StoreListItemsPosition::Cursor(cursor), ThreadItemsListCursor::Anchor(ThreadItemsListAnchor::Item { item_id }) => { @@ -3777,11 +3781,26 @@ impl ThreadRequestProcessor { } err => internal_error(format!("failed to list thread items: {err}")), })?; - let data = page - .items - .into_iter() - .map(stored_thread_item_to_entry) - .collect::, _>>()?; + let mut data = Vec::with_capacity(page.items.len()); + for stored_item in page.items { + let mut entry = stored_thread_item_to_entry(stored_item)?; + if defer_command_output + && entry.completed_at_ms.is_some() + && let ThreadItemPayload::CommandExecution { + status: codex_app_server_protocol::CommandExecutionStatus::Completed, + exit_code: Some(0), + aggregated_output, + .. + } = &mut entry.item.payload + && aggregated_output + .as_ref() + .is_some_and(|output| !output.is_empty()) + { + *aggregated_output = None; + entry.output_deferred = true; + } + data.push(entry); + } Ok(ThreadItemsListResponse { data, @@ -6475,6 +6494,7 @@ fn stored_thread_item_to_entry( let item = stored_thread_item_to_api_item(&stored_item)?; Ok(ThreadItemEntry { turn_id: stored_item.turn_id, + output_deferred: false, started_at_ms: item.started_at_ms, completed_at_ms: item.completed_at_ms, item, diff --git a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs index 6c33d9e11a..4fbdd37934 100644 --- a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs +++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs @@ -1879,6 +1879,7 @@ async fn realtime_conversation_stop_emits_closed_notification( cursor: None, limit: None, sort_direction: None, + defer_output: None, }) .await?; let items: ThreadItemsListResponse = 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 ff22262fe8..945508376c 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -25,6 +25,7 @@ use codex_app_server_protocol::SortDirection; use codex_app_server_protocol::ThreadForkParams; use codex_app_server_protocol::ThreadForkResponse; use codex_app_server_protocol::ThreadHistoryMode; +use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadItemEntry; use codex_app_server_protocol::ThreadItemPayload; use codex_app_server_protocol::ThreadItemsListAnchor; @@ -66,12 +67,14 @@ use codex_feedback::CodexFeedback; use codex_protocol::ContextWindowId; use codex_protocol::items::AgentMessageContent; use codex_protocol::items::AgentMessageItem; +use codex_protocol::items::CommandExecutionItem; use codex_protocol::items::TurnItem as CoreTurnItem; use codex_protocol::items::UserMessageItem; use codex_protocol::models::BaseInstructions; use codex_protocol::models::MessagePhase; use codex_protocol::protocol::AgentMessageEvent; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::ExecCommandSource; use codex_protocol::protocol::ItemCompletedEvent; use codex_protocol::protocol::SessionSource as ProtocolSessionSource; use codex_protocol::protocol::ThreadMemoryMode; @@ -1828,55 +1831,174 @@ async fn thread_turns_list_rejects_unmaterialized_loaded_thread() -> Result<()> Ok(()) } +#[tokio::test] +async fn thread_items_list_defers_successful_command_output() -> Result<()> { + use codex_protocol::items::CommandExecutionStatus::Completed; + use codex_protocol::items::CommandExecutionStatus::Declined; + use codex_protocol::items::CommandExecutionStatus::Failed; + use codex_protocol::items::CommandExecutionStatus::InProgress; + + async fn list(server: &mut TestAppServer, params: Value) -> Result { + let id = server + .send_request("thread/items/list", Some(params)) + .await?; + timeout(DEFAULT_READ_TIMEOUT, server.read_response(id)).await? + } + + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + MockResponsesConfig::new(&server.uri()).write(codex_home.path())?; + let thread_id = codex_protocol::ThreadId::default(); + let store = create_paginated_thread(codex_home.path(), thread_id).await?; + let large = "é".repeat(2048); + let small = "x".to_owned(); + let empty = String::new(); + let commands = [ + ("large", Completed, Some(0), Some(&large)), + ("small", Completed, Some(0), Some(&small)), + ("empty", Completed, Some(0), Some(&empty)), + ("missing-output", Completed, Some(0), None), + ("missing-exit", Completed, None, Some(&large)), + ("nonzero", Completed, Some(1), Some(&large)), + ("failed", Failed, Some(0), Some(&large)), + ("declined", Declined, Some(0), Some(&large)), + ("running", InProgress, Some(0), Some(&large)), + ]; + let mut items = vec![paginated_turn_started("turn")]; + for (id, status, exit_code, output) in commands { + items.push(paginated_completed_item( + thread_id, + "turn", + CoreTurnItem::CommandExecution(CommandExecutionItem { + id: id.to_string(), + command: vec!["ls".to_string()], + cwd: codex_home.path().abs().into(), + parsed_cmd: Vec::new(), + source: ExecCommandSource::Agent, + status, + aggregated_output: output.cloned(), + exit_code, + sandbox_type: None, + model_context: None, + plugin_id: None, + script_path: None, + process_id: None, + interaction_input: None, + duration: None, + }), + )); + } + store + .append_items(AppendThreadItemsParams { thread_id, items }) + .await?; + store.shutdown_thread(thread_id).await?; + + let mut stable = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .build() + .await?; + stable + .initialize_with_capabilities( + ClientInfo { + name: "stable-history-client".to_string(), + title: None, + version: "1".to_string(), + }, + /*capabilities*/ None, + ) + .await?; + let mut params = json!({"threadId": thread_id, "turnId": "turn"}); + let full = list(&mut stable, params.clone()).await?; + let entries = full["data"].as_array().expect("list entries"); + assert_eq!(entries.len(), commands.len()); + for (entry, (id, _, _, output)) in entries.iter().zip(commands) { + assert_eq!(entry["item"]["id"], id); + assert_eq!( + entry["item"]["aggregatedOutput"], + json!(output.filter(|text| !text.is_empty())) + ); + assert!(entry.get("outputDeferred").is_none()); + } + + for options in [ + Value::Null, + json!({}), + json!({"commandExecution": false}), + json!({"unknownItemKind": true}), + ] { + params["deferOutput"] = options; + assert_eq!(list(&mut stable, params.clone()).await?, full); + } + + params["deferOutput"] = json!({"commandExecution": true, "unknownItemKind": true}); + let id = stable + .send_request("thread/items/list", Some(params.clone())) + .await?; + let error = timeout( + DEFAULT_READ_TIMEOUT, + stable.read_stream_until_error_message(RequestId::Integer(id)), + ) + .await??; + assert_eq!(error.error.code, -32600); + + let mut experimental = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .build_initialized() + .await?; + assert_eq!( + list( + &mut experimental, + json!({"threadId": thread_id, "turnId": "turn", "deferLargeCommandOutput": true}), + ) + .await?, + full + ); + let deferred = list(&mut experimental, params).await?; + let mut expected = full.clone(); + for i in 0..2 { + expected["data"][i]["outputDeferred"] = json!(true); + expected["data"][i]["item"]["aggregatedOutput"] = Value::Null; + } + assert_eq!(deferred, expected); + + let projected: ThreadItemsListResponse = serde_json::from_value(deferred)?; + let mut entries_to_fetch = projected + .data + .into_iter() + .filter_map(|entry| ThreadItem::try_from(entry).err()); + let large_command = entries_to_fetch.next().expect("large command needs a read"); + let small_command = entries_to_fetch.next().expect("small command needs a read"); + assert!(entries_to_fetch.next().is_none()); + + let across_turns = list( + &mut experimental, + json!({"threadId": thread_id, "deferOutput": {"commandExecution": true}}), + ) + .await?; + assert_eq!(across_turns["data"], full["data"]); + let exact: Value = experimental + .request(|request_id| ClientRequest::ThreadItemsRead { + request_id, + params: ThreadItemsReadParams { + thread_id: thread_id.to_string(), + turn_id: small_command.turn_id, + item_ids: vec![large_command.item_id, small_command.item_id], + }, + }) + .await?; + assert_eq!(exact, json!({"data": [entries[0], entries[1]]})); + Ok(()) +} + #[tokio::test] async fn paginated_history_lists_and_legacy_reads_use_projected_turns_and_items() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; MockResponsesConfig::new(&server.uri()).write(codex_home.path())?; let thread_id = codex_protocol::ThreadId::default(); - let sqlite = codex_state::SqliteConfig::new_for_testing(codex_home.path().abs()); - let state_db = - codex_state::StateRuntime::init(sqlite.clone(), "mock_provider".to_string()).await?; - let store = LocalThreadStore::new( - LocalThreadStoreConfig { - codex_home: codex_home.path().to_path_buf(), - sqlite, - default_model_provider_id: "mock_provider".to_string(), - }, - Some(state_db), - ); - store - .create_thread(CreateThreadParams { - attached_thread: false, - creator_user_id: None, - creator_account_id: None, - session_id: thread_id.into(), - thread_id, - extra_config: None, - forked_from_id: None, - parent_thread_id: None, - source: ProtocolSessionSource::Cli, - thread_source: None, - originator: "test_originator".to_string(), - base_instructions: BaseInstructions::default(), - dynamic_tools: Vec::new(), - selected_capability_roots: Vec::new(), - multi_agent_version: None, - history_mode: codex_protocol::protocol::ThreadHistoryMode::Paginated, - history_base: None, - subagent_history_start_ordinal: None, - initial_window_id: ContextWindowId::new_uuid_v7().to_string(), - runtime_workspace_roots: None, - metadata: ThreadPersistenceMetadata { - cwd: Some(codex_home.path().to_path_buf()), - model_provider: "mock_provider".to_string(), - memory_mode: ThreadMemoryMode::Enabled, - }, - }) - .await?; - store - .persist_thread(thread_id, PersistContext::Standard) - .await?; + let store = create_paginated_thread(codex_home.path(), thread_id).await?; store .append_items(AppendThreadItemsParams { thread_id, @@ -2254,6 +2376,7 @@ async fn paginated_history_lists_and_legacy_reads_use_projected_turns_and_items( first_items_page.data, vec![ThreadItemEntry { turn_id: "turn-1".to_string(), + output_deferred: false, item: codex_app_server_protocol::ThreadItem { payload: ThreadItemPayload::UserMessage { id: "user-1".to_string(), @@ -2300,6 +2423,7 @@ async fn paginated_history_lists_and_legacy_reads_use_projected_turns_and_items( cursor: None, limit: Some(1), sort_direction: None, + defer_output: None, }; for direction in [None, Some(SortDirection::Desc)] { let request = mcp @@ -2390,6 +2514,7 @@ async fn paginated_history_lists_and_legacy_reads_use_projected_turns_and_items( vec![ ThreadItemEntry { turn_id: "turn-1".to_string(), + output_deferred: false, item: codex_app_server_protocol::ThreadItem { payload: ThreadItemPayload::UserMessage { id: "steer-1".to_string(), @@ -2404,6 +2529,7 @@ async fn paginated_history_lists_and_legacy_reads_use_projected_turns_and_items( }, ThreadItemEntry { turn_id: "turn-1".to_string(), + output_deferred: false, item: codex_app_server_protocol::ThreadItem { payload: ThreadItemPayload::AgentMessage { id: "agent-1".to_string(), @@ -2479,6 +2605,7 @@ async fn thread_items_read_validates_ids_before_history_access() -> Result<()> { cursor: None, limit: None, sort_direction: None, + defer_output: None, }) .await?; let read_err: JSONRPCError = timeout( @@ -2721,6 +2848,7 @@ async fn read_items_page( cursor: cursor.map(ThreadItemsListCursor::Opaque), limit, sort_direction: Some(sort_direction), + defer_output: None, }) .await?; let response: JSONRPCResponse = timeout( @@ -2731,6 +2859,56 @@ async fn read_items_page( to_response(response) } +async fn create_paginated_thread( + codex_home: &Path, + thread_id: codex_protocol::ThreadId, +) -> Result { + let sqlite = codex_state::SqliteConfig::new_for_testing(codex_home.abs()); + let state_db = + codex_state::StateRuntime::init(sqlite.clone(), "mock_provider".to_string()).await?; + let store = LocalThreadStore::new( + LocalThreadStoreConfig { + codex_home: codex_home.to_path_buf(), + sqlite, + default_model_provider_id: "mock_provider".to_string(), + }, + Some(state_db), + ); + store + .create_thread(CreateThreadParams { + attached_thread: false, + creator_user_id: None, + creator_account_id: None, + session_id: thread_id.into(), + thread_id, + extra_config: None, + forked_from_id: None, + parent_thread_id: None, + source: ProtocolSessionSource::Cli, + thread_source: None, + originator: "test_originator".to_string(), + base_instructions: BaseInstructions::default(), + dynamic_tools: Vec::new(), + selected_capability_roots: Vec::new(), + multi_agent_version: None, + history_mode: codex_protocol::protocol::ThreadHistoryMode::Paginated, + history_base: None, + subagent_history_start_ordinal: None, + initial_window_id: ContextWindowId::new_uuid_v7().to_string(), + runtime_workspace_roots: None, + metadata: ThreadPersistenceMetadata { + cwd: Some(codex_home.to_path_buf()), + model_provider: "mock_provider".to_string(), + memory_mode: ThreadMemoryMode::Enabled, + }, + }) + .await?; + store + .persist_thread(thread_id, PersistContext::Standard) + .await?; + Ok(store) +} + fn paginated_turn_started(turn_id: &str) -> RolloutItem { RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_attribution: None, diff --git a/codex-rs/app-server/tests/suite/v2/thread_revert.rs b/codex-rs/app-server/tests/suite/v2/thread_revert.rs index 8ba0f07f07..224ef5ee8d 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_revert.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_revert.rs @@ -505,6 +505,7 @@ async fn thread_revert_replaces_paginated_history_before_turn() -> Result<()> { cursor: items_backwards_cursor.map(ThreadItemsListCursor::Opaque), limit: None, sort_direction: None, + defer_output: None, }, }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_timeline.rs b/codex-rs/app-server/tests/suite/v2/thread_timeline.rs index 68ebe2be11..9f9e62e171 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_timeline.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_timeline.rs @@ -223,6 +223,7 @@ async fn timeline_pages_mix_items_and_resolve_the_opening_realtime_session() -> cursor: None, limit: Some(10), sort_direction: Some(SortDirection::Asc), + defer_output: None, }, }) .await?; diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index d841ebf65a..40cfc4a491 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -1788,6 +1788,7 @@ async fn maybe_backfill_turn_completed_items( cursor, limit: None, sort_direction: None, + defer_output: None, }, }) .await @@ -1798,7 +1799,17 @@ async fn maybe_backfill_turn_completed_items( return; } }; - items.extend(page.data.into_iter().map(ThreadItem::from)); + for entry in page.data { + match ThreadItem::try_from(entry) { + Ok(item) => items.push(item), + Err(err) => { + warn!( + "thread/items/list returned an incomplete item during turn completion: {err}" + ); + return; + } + } + } let Some(next) = page.next_cursor else { break }; if !seen_cursors.insert(next.clone()) { warn!("thread/items/list returned a repeated cursor during turn completion"); diff --git a/codex-rs/tui/src/app/safety_buffering.rs b/codex-rs/tui/src/app/safety_buffering.rs index 659bcbacaf..aae4a2f19d 100644 --- a/codex-rs/tui/src/app/safety_buffering.rs +++ b/codex-rs/tui/src/app/safety_buffering.rs @@ -153,8 +153,8 @@ impl App { .data .into_iter() .rev() - .map(ThreadItem::from) - .collect(); + .map(ThreadItem::try_from) + .collect::>()?; turn.items_view = TurnItemsView::Full; } } diff --git a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs index ca94cd6ebf..4407ec76ea 100644 --- a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs +++ b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs @@ -2590,6 +2590,7 @@ async fn older_pagination_reconciles_review_prompts_across_page_boundaries() -> cursor: Some(ThreadItemsListCursor::Opaque(cursor.clone())), limit: Some(crate::app_server_session::HISTORY_ITEM_PAGE_LIMIT), sort_direction: Some(SortDirection::Desc), + defer_output: None, }, }) .await?; diff --git a/codex-rs/tui/src/app_server_session/history.rs b/codex-rs/tui/src/app_server_session/history.rs index ab0382b04e..505b5b469d 100644 --- a/codex-rs/tui/src/app_server_session/history.rs +++ b/codex-rs/tui/src/app_server_session/history.rs @@ -112,6 +112,7 @@ pub(crate) fn thread_items_page_params( cursor: cursor.map(ThreadItemsListCursor::Opaque), limit: Some(limit), sort_direction: Some(SortDirection::Desc), + defer_output: None, } } @@ -311,8 +312,9 @@ impl AppServerSession { if let Some(turn) = turns.iter_mut().find(|turn| turn.id == entry.turn_id) && !turn.items.iter().any(|item| item.id() == entry.item.id()) { - items.push(entry.item.payload.clone()); - turn.items.insert(/*index*/ 0, ThreadItem::from(entry)); + let item = ThreadItem::try_from(entry)?; + items.push(item.payload.clone()); + turn.items.insert(/*index*/ 0, item); turn.items_view = TurnItemsView::Summary; } } diff --git a/codex-rs/tui/src/dynamic_tools.rs b/codex-rs/tui/src/dynamic_tools.rs index 3d99d3aba1..abb57970fb 100644 --- a/codex-rs/tui/src/dynamic_tools.rs +++ b/codex-rs/tui/src/dynamic_tools.rs @@ -897,6 +897,7 @@ async fn execute_inner( cursor: None, limit: Some(20), sort_direction: Some(SortDirection::Desc), + defer_output: None, }, } }), @@ -914,6 +915,7 @@ async fn execute_inner( .cloned() .map(|item| ThreadItemEntry { turn_id: turn.id.clone(), + output_deferred: false, started_at_ms: item.started_at_ms, completed_at_ms: item.completed_at_ms, item, diff --git a/codex-rs/tui/src/resume_picker_transcript_preview_tests.rs b/codex-rs/tui/src/resume_picker_transcript_preview_tests.rs index fe516c6513..1986c7c1f5 100644 --- a/codex-rs/tui/src/resume_picker_transcript_preview_tests.rs +++ b/codex-rs/tui/src/resume_picker_transcript_preview_tests.rs @@ -413,6 +413,7 @@ async fn transcript_preview_for_history_mode( cursor: None, limit: Some(TRANSCRIPT_PREVIEW_ITEMS_PAGE_SIZE), sort_direction: Some(SortDirection::Desc), + defer_output: None, }, }) .await diff --git a/sdk/python/src/openai_codex/generated/v2_all.py b/sdk/python/src/openai_codex/generated/v2_all.py index 516ffdee3e..5ffaa9da4f 100644 --- a/sdk/python/src/openai_codex/generated/v2_all.py +++ b/sdk/python/src/openai_codex/generated/v2_all.py @@ -5959,6 +5959,12 @@ class ContextCompactionThreadItem(BaseModel): type: Annotated[Literal["contextCompaction"], Field(title="ContextCompactionThreadItemType")] +class ThreadItemOutputDeferral(BaseModel): + model_config = ConfigDict( + populate_by_name=True, + ) + + class ItemThreadItemsListAnchor(BaseModel): model_config = ConfigDict( populate_by_name=True, @@ -5991,6 +5997,13 @@ class ThreadItemsListParams(BaseModel): description="Opaque continuation cursor or an exclusive item anchor in the requested visible turn. An item anchor requires a non-empty `turnId`; ascending (the default) returns items after it, and descending returns items before it. Continue with the returned string cursor." ), ] = None + defer_output: Annotated[ + ThreadItemOutputDeferral | None, + Field( + alias="deferOutput", + description="For a single turn, choose which saved outputs to fetch separately by item ID.", + ), + ] = None limit: Annotated[int | None, Field(description="Optional item page size.", ge=0)] = None sort_direction: Annotated[ SortDirection | None,