diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 8a3613daa814..996f9ea93c69 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -4202,6 +4202,22 @@ ], "type": "object" }, + "ThreadRealtimeInitialItem": { + "description": "EXPERIMENTAL - role-bearing text item included when a realtime V3 session starts.", + "properties": { + "role": { + "$ref": "#/definitions/ConversationTextRole" + }, + "text": { + "type": "string" + } + }, + "required": [ + "role", + "text" + ], + "type": "object" + }, "ThreadRealtimeStartTransport": { "description": "EXPERIMENTAL - transport used by thread realtime.", "oneOf": [ 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 6ffe55efafcc..e096b1149d03 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 @@ -19581,6 +19581,22 @@ "title": "ThreadRealtimeErrorNotification", "type": "object" }, + "ThreadRealtimeInitialItem": { + "description": "EXPERIMENTAL - role-bearing text item included when a realtime V3 session starts.", + "properties": { + "role": { + "$ref": "#/definitions/v2/ConversationTextRole" + }, + "text": { + "type": "string" + } + }, + "required": [ + "role", + "text" + ], + "type": "object" + }, "ThreadRealtimeItemAddedNotification": { "$schema": "http://json-schema.org/draft-07/schema#", "description": "EXPERIMENTAL - raw non-audio thread realtime item emitted by the backend.", 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 0076353c8f38..ce1a37976e61 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 @@ -17360,6 +17360,22 @@ "title": "ThreadRealtimeErrorNotification", "type": "object" }, + "ThreadRealtimeInitialItem": { + "description": "EXPERIMENTAL - role-bearing text item included when a realtime V3 session starts.", + "properties": { + "role": { + "$ref": "#/definitions/ConversationTextRole" + }, + "text": { + "type": "string" + } + }, + "required": [ + "role", + "text" + ], + "type": "object" + }, "ThreadRealtimeItemAddedNotification": { "$schema": "http://json-schema.org/draft-07/schema#", "description": "EXPERIMENTAL - raw non-audio thread realtime item emitted by the backend.", diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeInitialItem.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeInitialItem.ts new file mode 100644 index 000000000000..6801b94faba5 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeInitialItem.ts @@ -0,0 +1,9 @@ +// 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 { ConversationTextRole } from "../ConversationTextRole"; + +/** + * EXPERIMENTAL - role-bearing text item included when a realtime V3 session starts. + */ +export type ThreadRealtimeInitialItem = { role: ConversationTextRole, text: string, }; 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 e190a2790d1e..f818e310ed6b 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -449,6 +449,7 @@ export type { ThreadReadResponse } from "./ThreadReadResponse"; export type { ThreadRealtimeAudioChunk } from "./ThreadRealtimeAudioChunk"; export type { ThreadRealtimeClosedNotification } from "./ThreadRealtimeClosedNotification"; export type { ThreadRealtimeErrorNotification } from "./ThreadRealtimeErrorNotification"; +export type { ThreadRealtimeInitialItem } from "./ThreadRealtimeInitialItem"; export type { ThreadRealtimeItemAddedNotification } from "./ThreadRealtimeItemAddedNotification"; export type { ThreadRealtimeOutputAudioDeltaNotification } from "./ThreadRealtimeOutputAudioDeltaNotification"; export type { ThreadRealtimeSdpNotification } from "./ThreadRealtimeSdpNotification"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 091251a2c53a..a4d87e202c7f 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -1772,6 +1772,7 @@ mod tests { use codex_protocol::models::BUILT_IN_PERMISSION_PROFILE_READ_ONLY; use codex_protocol::parse_command::ParsedCommand; use codex_protocol::protocol::CodexResponseHandoffMode; + use codex_protocol::protocol::ConversationTextRole; use codex_protocol::protocol::RealtimeConversationVersion; use codex_protocol::protocol::RealtimeOutputModality; use codex_protocol::protocol::RealtimeVoice; @@ -3403,6 +3404,16 @@ mod tests { model: Some("realtime-treatment-model".to_string()), output_modality: RealtimeOutputModality::Audio, include_startup_context: Some(false), + initial_items: Some(vec![ + v2::ThreadRealtimeInitialItem { + role: ConversationTextRole::Developer, + text: "Remember this.".to_string(), + }, + v2::ThreadRealtimeInitialItem { + role: ConversationTextRole::Assistant, + text: "Understood.".to_string(), + }, + ]), prompt: Some(Some("You are on a call".to_string())), realtime_session_id: Some("sess_456".to_string()), transport: None, @@ -3424,6 +3435,16 @@ mod tests { "model": "realtime-treatment-model", "outputModality": "audio", "includeStartupContext": false, + "initialItems": [ + { + "role": "developer", + "text": "Remember this." + }, + { + "role": "assistant", + "text": "Understood." + } + ], "prompt": "You are on a call", "realtimeSessionId": "sess_456", "transport": null, @@ -3450,6 +3471,7 @@ mod tests { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: None, + initial_items: None, prompt: None, realtime_session_id: None, transport: None, @@ -3471,6 +3493,7 @@ mod tests { "model": null, "outputModality": "audio", "includeStartupContext": null, + "initialItems": null, "realtimeSessionId": null, "transport": null, "version": null, @@ -3492,6 +3515,7 @@ mod tests { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: None, + initial_items: None, prompt: Some(None), realtime_session_id: None, transport: None, @@ -3513,6 +3537,7 @@ mod tests { "model": null, "outputModality": "audio", "includeStartupContext": null, + "initialItems": null, "prompt": null, "realtimeSessionId": null, "transport": null, @@ -3734,6 +3759,7 @@ mod tests { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: None, + initial_items: None, prompt: Some(Some("You are on a call".to_string())), realtime_session_id: None, transport: None, diff --git a/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs b/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs index 9333f9d356a0..234b2089e97e 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs @@ -96,6 +96,11 @@ pub struct ThreadRealtimeStartParams { /// Set to false to start without Codex's startup context. Omitted or null includes it. #[ts(optional = nullable)] pub include_startup_context: Option, + /// Adds complete role-bearing text items to the initial Frameless Bidi session history. + /// This is only supported by realtime V3 and is sent during session startup. Requests are + /// limited to 128 items and 8,192 estimated text tokens in total. + #[ts(optional = nullable)] + pub initial_items: Option>, #[serde( default, deserialize_with = "crate::protocol::serde_helpers::deserialize_double_option", @@ -115,6 +120,15 @@ pub struct ThreadRealtimeStartParams { pub voice: Option, } +/// EXPERIMENTAL - role-bearing text item included when a realtime V3 session starts. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeInitialItem { + pub role: ConversationTextRole, + pub text: String, +} + /// EXPERIMENTAL - transport used by thread realtime. #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(tag = "type", rename_all = "camelCase")] 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 580a57d592a9..65f8c72b32df 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -4375,3 +4375,14 @@ fn realtime_append_text_defaults_role_to_user() { } ); } + +#[test] +fn realtime_start_omitted_initial_items_remain_none() { + let params = serde_json::from_value::(json!({ + "threadId": "thread_123", + "outputModality": "audio", + })) + .expect("params should deserialize"); + + assert_eq!(params.initial_items, None); +} diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 13a19c3be4e6..465b82a3ff6e 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -173,7 +173,7 @@ Example with notification opt-out: - `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`. - `turn/interrupt` — request cancellation of an in-flight turn by `(thread_id, turn_id)`; success is an empty `{}` response and the turn finishes with `status: "interrupted"`. -- `thread/realtime/start` — start a thread-scoped realtime session (experimental); pass `outputModality: "text"` or `outputModality: "audio"` to choose model output, optionally pass `model` and `version` to override configured realtime selection for this session only, and pass `includeStartupContext: false` to omit Codex's generated startup context. Version `"v1"` uses legacy Bidi `conversation.handoff.*`, `"v2"` uses the Realtime Voice API, and `"v3"` preserves V1 Codex Voice behavior while using Frameless Bidi `delegation.*`. For V3 automatic Codex text, `codexResponseHandoffMode` accepts `"thinking"` (the default; all output uses channel-less thinking appends), `"commentary"` (all output uses the commentary channel), or `"bemTags"` (the raw BEM envelope selects the API channel: BEM `analysis` and `commentary` use `commentary`, while BEM `final` and unparsable output use `speakable`). The BEM envelope remains in the appended text for the frontend model to interpret. V1 and V2 ignore this setting. V3 handoffs do not prepend the legacy `"Agent Final Message"` label. Pass `clientManagedHandoffs: true` to disable automatic Codex response delivery so only the client's explicit append calls produce handoffs. Pass `codexResponsesAsItems: true` to send automatic Codex responses as realtime conversation items instead, and optionally pass `codexResponseItemPrefix` to prepend experiment instructions to those items. Returns `{}` and streams `thread/realtime/*` notifications. Omit `transport` for the websocket transport, or pass `{ "type": "webrtc", "sdp": "..." }` to create a Bidi WebRTC session from a browser-generated SDP offer; the remote answer SDP is emitted as `thread/realtime/sdp`. Conversation `version: "v2"` requests remain unsupported for WebRTC. +- `thread/realtime/start` — start a thread-scoped realtime session (experimental); pass `outputModality: "text"` or `outputModality: "audio"` to choose model output, optionally pass `model` and `version` to override configured realtime selection for this session only, pass `includeStartupContext: false` to omit Codex's generated startup context, and optionally pass `initialItems` to seed V3 with complete role-bearing text messages at session creation. Version `"v1"` uses legacy Bidi `conversation.handoff.*`, `"v2"` uses the Realtime Voice API, and `"v3"` preserves V1 Codex Voice behavior while using Frameless Bidi `delegation.*`. For V3 automatic Codex text, `codexResponseHandoffMode` accepts `"thinking"` (the default; all output uses channel-less thinking appends), `"commentary"` (all output uses the commentary channel), or `"bemTags"` (the raw BEM envelope selects the API channel: BEM `analysis` and `commentary` use `commentary`, while BEM `final` and unparsable output use `speakable`). The BEM envelope remains in the appended text for the frontend model to interpret. V1 and V2 ignore this setting. V3 handoffs do not prepend the legacy `"Agent Final Message"` label. Pass `clientManagedHandoffs: true` to disable automatic Codex response delivery so only the client's explicit append calls produce handoffs. Pass `codexResponsesAsItems: true` to send automatic Codex responses as realtime conversation items instead, and optionally pass `codexResponseItemPrefix` to prepend experiment instructions to those items. Returns `{}` and streams `thread/realtime/*` notifications. Omit `transport` for the websocket transport, or pass `{ "type": "webrtc", "sdp": "..." }` to create a Bidi WebRTC session from a browser-generated SDP offer; the remote answer SDP is emitted as `thread/realtime/sdp`. Conversation `version: "v2"` requests remain unsupported for WebRTC. - `thread/realtime/appendAudio` — append an input audio chunk to the active realtime session (experimental); returns `{}`. - `thread/realtime/appendText` — append text input to the active realtime session with a required `role` of `user`, `developer`, or `assistant` (experimental); returns `{}`. Older clients that omit `role` default to `user`. - `thread/realtime/appendSpeech` — append text that the realtime model should speak to the user (experimental); returns `{}`. @@ -943,6 +943,32 @@ only. WebRTC uses AVAS and supports legacy Bidi `"v1"` or Frameless Bidi `"v3"`; Realtime Voice `"v2"` is rejected for WebRTC. Pass `includeStartupContext: false` to skip Codex's startup context for this session while still using the selected backend prompt. +For V3, clients may pass `initialItems` to seed the session with complete text +messages before live input begins: + +```json +{ + "initialItems": [ + { + "role": "developer", + "text": "Relevant user memory: prefers concise technical answers." + }, + { + "role": "user", + "text": "Continue from the prior discussion." + } + ] +} +``` + +Each item requires a `role` of `"user"`, `"developer"`, or `"assistant"` and a +`text` string. Core serializes these as Frameless Bidi `session.initial_items` +during the initial session bootstrap (including WebRTC call creation). +Requests are limited to 128 items, 8,192 estimated text tokens per item, and +8,192 estimated text tokens across all items. +Omitting `initialItems`, or passing an empty list, preserves the previous +session payload and startup behavior. V1 and V2 reject non-empty +`initialItems`. Pass `clientManagedHandoffs: true` to suppress automatic Codex response handoffs and items. The client can then choose which updates to deliver with `thread/realtime/appendText` or `thread/realtime/appendSpeech`. 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 1b919775eb56..664247950aa5 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -1081,6 +1081,15 @@ impl TurnRequestProcessor { model: params.model, output_modality: params.output_modality, include_startup_context: params.include_startup_context.unwrap_or(true), + initial_items: params + .initial_items + .unwrap_or_default() + .into_iter() + .map(|item| ConversationTextParams { + text: item.text, + role: item.role, + }) + .collect(), prompt: params.prompt, realtime_session_id: params.realtime_session_id, transport: params.transport.map(|transport| match transport { diff --git a/codex-rs/app-server/tests/suite/v2/experimental_api.rs b/codex-rs/app-server/tests/suite/v2/experimental_api.rs index 5bd44f9f5cd1..ca120bfed2fc 100644 --- a/codex-rs/app-server/tests/suite/v2/experimental_api.rs +++ b/codex-rs/app-server/tests/suite/v2/experimental_api.rs @@ -98,6 +98,7 @@ async fn realtime_conversation_start_requires_experimental_api_capability() -> R model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: None, + initial_items: None, prompt: Some(Some("hello".to_string())), realtime_session_id: None, transport: None, @@ -227,6 +228,7 @@ async fn realtime_webrtc_start_requires_experimental_api_capability() -> Result< model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: None, + initial_items: None, prompt: Some(Some("hello".to_string())), realtime_session_id: None, transport: Some(ThreadRealtimeStartTransport::Webrtc { 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 9c6f787a760e..bb7a77f0548a 100644 --- a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs +++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs @@ -22,6 +22,7 @@ use codex_app_server_protocol::ThreadRealtimeAppendTextResponse; use codex_app_server_protocol::ThreadRealtimeAudioChunk; use codex_app_server_protocol::ThreadRealtimeClosedNotification; use codex_app_server_protocol::ThreadRealtimeErrorNotification; +use codex_app_server_protocol::ThreadRealtimeInitialItem; use codex_app_server_protocol::ThreadRealtimeItemAddedNotification; use codex_app_server_protocol::ThreadRealtimeListVoicesParams; use codex_app_server_protocol::ThreadRealtimeListVoicesResponse; @@ -377,6 +378,7 @@ impl RealtimeE2eHarness { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: None, + initial_items: None, prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: Some(ThreadRealtimeStartTransport::Webrtc { @@ -438,6 +440,7 @@ impl RealtimeE2eHarness { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: None, + initial_items: None, prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -460,6 +463,7 @@ impl RealtimeE2eHarness { async fn start_frameless_bidi_realtime( &mut self, codex_response_handoff_mode: Option, + initial_items: Option>, ) -> Result { let start_request_id = self .mcp @@ -473,6 +477,7 @@ impl RealtimeE2eHarness { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: None, + initial_items, prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -745,6 +750,7 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> { model: Some("realtime-treatment-model".to_string()), output_modality: RealtimeOutputModality::Audio, include_startup_context: None, + initial_items: None, prompt: None, realtime_session_id: None, transport: None, @@ -1040,6 +1046,7 @@ async fn realtime_start_can_skip_startup_context() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: Some(false), + initial_items: None, prompt: None, realtime_session_id: None, transport: None, @@ -1142,6 +1149,7 @@ async fn realtime_text_output_modality_requests_text_output_and_final_transcript model: None, output_modality: RealtimeOutputModality::Text, include_startup_context: None, + initial_items: None, prompt: None, realtime_session_id: None, transport: None, @@ -1330,6 +1338,7 @@ async fn realtime_conversation_stop_emits_closed_notification() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: None, + initial_items: None, prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -1438,6 +1447,7 @@ async fn realtime_webrtc_start_emits_sdp_notification() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: None, + initial_items: None, prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: Some(ThreadRealtimeStartTransport::Webrtc { @@ -2222,6 +2232,56 @@ async fn websocket_v2_assistant_output_without_handoff_reaches_realtime_context( Ok(()) } +#[tokio::test] +async fn websocket_v3_passes_initial_items_through_session_start() -> Result<()> { + skip_if_no_network!(Ok(())); + + let mut harness = RealtimeE2eHarness::new( + RealtimeTestVersion::V1, + main_loop_responses(Vec::new()), + realtime_sideband(vec![realtime_sideband_connection(vec![vec![ + session_started("sess_initial_items"), + ]])]), + ) + .await?; + + let started = harness + .start_frameless_bidi_realtime( + /*codex_response_handoff_mode*/ None, + Some(vec![ + ThreadRealtimeInitialItem { + role: ConversationTextRole::Developer, + text: "Remember this.".to_string(), + }, + ThreadRealtimeInitialItem { + role: ConversationTextRole::Assistant, + text: "Understood.".to_string(), + }, + ]), + ) + .await?; + + assert_eq!(started.version, RealtimeConversationVersion::V3); + assert_eq!( + harness.sideband_outbound_request(/*request_index*/ 0).await["session"]["initial_items"], + json!([ + { + "type": "message", + "role": "developer", + "content": [{"type": "input_text", "text": "Remember this."}], + }, + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "Understood."}], + }, + ]) + ); + + harness.shutdown().await; + Ok(()) +} + #[tokio::test] async fn websocket_v3_routes_handoffs_by_session_mode() -> Result<()> { skip_if_no_network!(Ok(())); @@ -2292,7 +2352,9 @@ async fn websocket_v3_routes_handoffs_by_session_mode() -> Result<()> { ) .await?; - let started = harness.start_frameless_bidi_realtime(mode).await?; + let started = harness + .start_frameless_bidi_realtime(mode, /*initial_items*/ None) + .await?; assert_eq!(started.version, RealtimeConversationVersion::V3); let _ = harness .read_notification::("turn/completed") @@ -3012,6 +3074,7 @@ async fn realtime_webrtc_start_surfaces_backend_error() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: None, + initial_items: None, prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: Some(ThreadRealtimeStartTransport::Webrtc { @@ -3082,6 +3145,7 @@ async fn realtime_conversation_requires_feature_flag() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: None, + initial_items: None, prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, diff --git a/codex-rs/codex-api/src/endpoint/realtime_call.rs b/codex-rs/codex-api/src/endpoint/realtime_call.rs index 37b753ee5c77..e005fd4ad06c 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_call.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_call.rs @@ -305,6 +305,8 @@ mod tests { use codex_client::Response; use codex_client::StreamResponse; use codex_client::TransportError; + use codex_protocol::protocol::ConversationTextParams; + use codex_protocol::protocol::ConversationTextRole; use codex_protocol::protocol::RealtimeVoice; use http::StatusCode; use pretty_assertions::assert_eq; @@ -386,6 +388,7 @@ mod tests { fn realtime_session_config(session_id: &str) -> RealtimeSessionConfig { RealtimeSessionConfig { instructions: "hi".to_string(), + initial_items: Vec::new(), model: Some("gpt-realtime".to_string()), session_id: Some(session_id.to_string()), event_parser: RealtimeEventParser::V1, @@ -697,12 +700,20 @@ mod tests { provider("https://chatgpt.com/backend-api/codex"), Arc::new(DummyAuth), ); + let mut session_config = frameless_bidi_session_config("sess-backend"); + session_config.initial_items = vec![ + ConversationTextParams { + text: "Remember this.".to_string(), + role: ConversationTextRole::Developer, + }, + ConversationTextParams { + text: "Understood.".to_string(), + role: ConversationTextRole::Assistant, + }, + ]; let response = client - .create_with_session( - "v=offer\r\n".to_string(), - frameless_bidi_session_config("sess-backend"), - ) + .create_with_session("v=offer\r\n".to_string(), session_config) .await .expect("request should succeed"); @@ -718,6 +729,21 @@ mod tests { }; assert_eq!(body["session"]["delegation"]["type"], "client"); assert!(body["session"].get("id").is_none()); + assert_eq!( + body["session"]["initial_items"], + serde_json::json!([ + { + "type": "message", + "role": "developer", + "content": [{"type": "input_text", "text": "Remember this."}], + }, + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "Understood."}], + }, + ]) + ); } #[tokio::test] diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs index 9d075a89c8bc..fa33f3524bd5 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -23,6 +23,7 @@ use crate::error::ApiError; use crate::provider::Provider; use codex_client::backoff; use codex_http_client::maybe_build_rustls_client_config_with_custom_ca; +use codex_protocol::protocol::ConversationTextParams; use codex_protocol::protocol::ConversationTextRole; use codex_protocol::protocol::RealtimeTranscriptDelta; use codex_utils_rustls_provider::ensure_rustls_crypto_provider; @@ -378,6 +379,7 @@ impl RealtimeWebsocketWriter { pub async fn send_session_update( &self, instructions: String, + initial_items: Vec, session_mode: RealtimeSessionMode, output_modality: RealtimeOutputModality, voice: RealtimeVoice, @@ -386,6 +388,7 @@ impl RealtimeWebsocketWriter { let message = session_update_message( self.event_parser, instructions, + initial_items, session_mode, output_modality, voice, @@ -841,6 +844,7 @@ impl RealtimeWebsocketClient { .writer .send_session_update( config.instructions, + config.initial_items, config.session_mode, config.output_modality, config.voice, @@ -1965,6 +1969,7 @@ mod tests { .connect( RealtimeSessionConfig { instructions: "backend prompt".to_string(), + initial_items: Vec::new(), model: Some("realtime-test-model".to_string()), session_id: Some("conv_1".to_string()), event_parser: RealtimeEventParser::V1, @@ -2289,6 +2294,7 @@ mod tests { .connect( RealtimeSessionConfig { instructions: "backend prompt".to_string(), + initial_items: Vec::new(), model: Some("realtime-test-model".to_string()), session_id: Some("conv_1".to_string()), event_parser: RealtimeEventParser::RealtimeV2, @@ -2414,6 +2420,7 @@ mod tests { .connect( RealtimeSessionConfig { instructions: "backend prompt".to_string(), + initial_items: Vec::new(), model: Some("realtime-test-model".to_string()), session_id: Some("conv_1".to_string()), event_parser: RealtimeEventParser::RealtimeV2, @@ -2518,6 +2525,7 @@ mod tests { .connect( RealtimeSessionConfig { instructions: "backend prompt".to_string(), + initial_items: Vec::new(), model: Some("realtime-test-model".to_string()), session_id: Some("conv_1".to_string()), event_parser: RealtimeEventParser::V1, @@ -2608,6 +2616,7 @@ mod tests { .connect( RealtimeSessionConfig { instructions: "backend prompt".to_string(), + initial_items: Vec::new(), model: Some("realtime-test-model".to_string()), session_id: Some("conv_1".to_string()), event_parser: RealtimeEventParser::V1, diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_common.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_common.rs index e6376be76844..c2cad2aac409 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_common.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_common.rs @@ -17,6 +17,7 @@ use crate::endpoint::realtime_websocket::protocol::RealtimeSessionConfig; use crate::endpoint::realtime_websocket::protocol::RealtimeSessionMode; use crate::endpoint::realtime_websocket::protocol::RealtimeVoice; use crate::endpoint::realtime_websocket::protocol::RealtimeWireAdapter; +use codex_protocol::protocol::ConversationTextParams; use codex_protocol::protocol::ConversationTextRole; use serde_json::Result as JsonResult; use serde_json::Value; @@ -113,6 +114,7 @@ pub(super) fn conversation_function_call_output_message( pub(super) fn session_update_message( wire_adapter: RealtimeWireAdapter, instructions: String, + initial_items: Vec, session_mode: RealtimeSessionMode, output_modality: RealtimeOutputModality, voice: RealtimeVoice, @@ -122,7 +124,9 @@ pub(super) fn session_update_message( RealtimeWireAdapter::V1 => RealtimeOutboundMessage::SessionUpdate { session: v1_session_update_session(instructions, voice), }, - RealtimeWireAdapter::FramelessBidi => frameless_session_update_message(instructions, voice), + RealtimeWireAdapter::FramelessBidi => { + frameless_session_update_message(instructions, initial_items, voice) + } RealtimeWireAdapter::RealtimeV2 => RealtimeOutboundMessage::SessionUpdate { session: v2_session_update_session(instructions, session_mode, output_modality, voice), }, @@ -151,6 +155,7 @@ pub fn session_update_session_json(config: RealtimeSessionConfig) -> JsonResult< RealtimeWireAdapter::FramelessBidi => Ok(frameless_session_json( config.model, config.instructions, + config.initial_items, config.voice, )), } diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_frameless_bidi.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_frameless_bidi.rs index 2f539255d606..d92907fb1458 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_frameless_bidi.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_frameless_bidi.rs @@ -3,6 +3,8 @@ use crate::endpoint::realtime_websocket::protocol::FramelessInputTextContent; use crate::endpoint::realtime_websocket::protocol::RealtimeContextAppendChannel; use crate::endpoint::realtime_websocket::protocol::RealtimeOutboundMessage; use crate::endpoint::realtime_websocket::protocol::RealtimeVoice; +use codex_protocol::protocol::ConversationTextParams; +use codex_protocol::protocol::ConversationTextRole; use serde_json::Value; use serde_json::json; @@ -32,16 +34,18 @@ pub(super) fn session_context_append_message( pub(super) fn session_update_message( instructions: String, + initial_items: Vec, voice: RealtimeVoice, ) -> RealtimeOutboundMessage { RealtimeOutboundMessage::FramelessSessionUpdate { - session: session_json(/*model*/ None, instructions, voice), + session: session_json(/*model*/ None, instructions, initial_items, voice), } } pub(super) fn session_json( model: Option, instructions: String, + initial_items: Vec, voice: RealtimeVoice, ) -> Value { let mut session = json!({ @@ -58,6 +62,29 @@ pub(super) fn session_json( if let Some(model) = model { session["model"] = Value::String(model); } + if !initial_items.is_empty() { + session["initial_items"] = Value::Array( + initial_items + .into_iter() + .map(|item| { + let content_type = match item.role { + ConversationTextRole::User | ConversationTextRole::Developer => { + "input_text" + } + ConversationTextRole::Assistant => "output_text", + }; + json!({ + "type": "message", + "role": item.role, + "content": [{ + "type": content_type, + "text": item.text, + }], + }) + }) + .collect(), + ); + } session } diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_frameless_bidi_tests.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_frameless_bidi_tests.rs index d3651c65b65f..c6d141231802 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_frameless_bidi_tests.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_frameless_bidi_tests.rs @@ -1,5 +1,11 @@ use super::CONTEXT_APPEND_MAX_BYTES; use super::context_append_chunks; +use super::session_json; +use crate::endpoint::realtime_websocket::protocol::RealtimeVoice; +use codex_protocol::protocol::ConversationTextParams; +use codex_protocol::protocol::ConversationTextRole; +use pretty_assertions::assert_eq; +use serde_json::json; #[test] fn context_append_chunks_preserve_text_within_wire_limit() { @@ -13,3 +19,82 @@ fn context_append_chunks_preserve_text_within_wire_limit() { ); } } + +#[test] +fn session_json_omits_initial_items_when_empty() { + let session = session_json( + Some("gpt-live".to_string()), + "instructions".to_string(), + Vec::new(), + RealtimeVoice::Marin, + ); + + assert_eq!( + session, + json!({ + "model": "gpt-live", + "instructions": "instructions", + "audio": { + "output": { + "voice": "marin", + }, + }, + "delegation": { + "type": "client", + }, + }) + ); +} + +#[test] +fn session_json_encodes_role_bearing_initial_items() { + let session = session_json( + Some("gpt-live".to_string()), + "instructions".to_string(), + vec![ + ConversationTextParams { + text: "Remember this.".to_string(), + role: ConversationTextRole::Developer, + }, + ConversationTextParams { + text: "What do you remember?".to_string(), + role: ConversationTextRole::User, + }, + ConversationTextParams { + text: "I remember.".to_string(), + role: ConversationTextRole::Assistant, + }, + ], + RealtimeVoice::Marin, + ); + + assert_eq!( + session["initial_items"], + json!([ + { + "type": "message", + "role": "developer", + "content": [{ + "type": "input_text", + "text": "Remember this.", + }], + }, + { + "type": "message", + "role": "user", + "content": [{ + "type": "input_text", + "text": "What do you remember?", + }], + }, + { + "type": "message", + "role": "assistant", + "content": [{ + "type": "output_text", + "text": "I remember.", + }], + }, + ]) + ); +} diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs index a47d980a7bc0..a4b0037d1b48 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs @@ -1,6 +1,7 @@ use crate::endpoint::realtime_websocket::protocol_frameless_bidi::parse_frameless_bidi_event; use crate::endpoint::realtime_websocket::protocol_v1::parse_realtime_event_v1; use crate::endpoint::realtime_websocket::protocol_v2::parse_realtime_event_v2; +use codex_protocol::protocol::ConversationTextParams; use codex_protocol::protocol::ConversationTextRole; pub use codex_protocol::protocol::RealtimeAudioFrame; pub use codex_protocol::protocol::RealtimeEvent; @@ -36,6 +37,7 @@ pub enum RealtimeContextAppendChannel { #[derive(Debug, Clone, PartialEq, Eq)] pub struct RealtimeSessionConfig { pub instructions: String, + pub initial_items: Vec, pub model: Option, pub session_id: Option, pub event_parser: RealtimeEventParser, diff --git a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs index 3aaf3f5f58c8..5d0f8f910ca2 100644 --- a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs +++ b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs @@ -144,6 +144,7 @@ async fn realtime_ws_e2e_session_create_and_event_flow() { .connect( RealtimeSessionConfig { instructions: "backend prompt".to_string(), + initial_items: Vec::new(), model: Some("realtime-test-model".to_string()), session_id: Some("conv_123".to_string()), event_parser: RealtimeEventParser::V1, @@ -248,6 +249,7 @@ async fn realtime_ws_connect_webrtc_sideband_retries_join_until_server_is_availa .connect_webrtc_sideband( RealtimeSessionConfig { instructions: "backend prompt".to_string(), + initial_items: Vec::new(), model: Some("realtime-test-model".to_string()), session_id: Some("conv_123".to_string()), event_parser: RealtimeEventParser::RealtimeV2, @@ -320,6 +322,7 @@ async fn realtime_ws_e2e_send_while_next_event_waits() { .connect( RealtimeSessionConfig { instructions: "backend prompt".to_string(), + initial_items: Vec::new(), model: Some("realtime-test-model".to_string()), session_id: Some("conv_123".to_string()), event_parser: RealtimeEventParser::V1, @@ -388,6 +391,7 @@ async fn realtime_ws_e2e_disconnected_emitted_once() { .connect( RealtimeSessionConfig { instructions: "backend prompt".to_string(), + initial_items: Vec::new(), model: Some("realtime-test-model".to_string()), session_id: Some("conv_123".to_string()), event_parser: RealtimeEventParser::V1, @@ -452,6 +456,7 @@ async fn realtime_ws_e2e_ignores_unknown_text_events() { .connect( RealtimeSessionConfig { instructions: "backend prompt".to_string(), + initial_items: Vec::new(), model: Some("realtime-test-model".to_string()), session_id: Some("conv_123".to_string()), event_parser: RealtimeEventParser::V1, @@ -559,6 +564,7 @@ async fn realtime_ws_e2e_realtime_v2_parser_emits_handoff_requested() { .connect( RealtimeSessionConfig { instructions: "backend prompt".to_string(), + initial_items: Vec::new(), model: Some("realtime-test-model".to_string()), session_id: Some("conv_123".to_string()), event_parser: RealtimeEventParser::RealtimeV2, diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index 19a3fec14fc8..1532a3d146a0 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -57,6 +57,7 @@ use codex_protocol::protocol::RealtimeTranscriptEntry; use codex_protocol::protocol::RealtimeVoice; use codex_protocol::protocol::RealtimeVoicesList; use codex_utils_output_truncation::approx_bytes_for_tokens; +use codex_utils_string::approx_token_count; use codex_utils_string::take_bytes_at_char_boundary; use http::HeaderMap; use http::HeaderValue; @@ -87,6 +88,8 @@ const HANDOFF_OUT_QUEUE_CAPACITY: usize = 64; const OUTPUT_EVENTS_QUEUE_CAPACITY: usize = 256; const REALTIME_STARTUP_CONTEXT_TOKEN_BUDGET: usize = 5_300; const REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET: usize = 1_000; +const REALTIME_INITIAL_ITEMS_MAX_COUNT: usize = 128; +const REALTIME_INITIAL_ITEMS_MAX_TOKENS: usize = 8_192; const HANDOFF_STREAM_FLUSH_INTERVAL: Duration = Duration::from_millis(200); const HANDOFF_STREAM_TRUNCATION_MARKER: &str = "\n…output truncated…\n"; const AGENT_FINAL_MESSAGE_PREFIX: &str = "\"Agent Final Message\":\n\n"; @@ -1240,6 +1243,31 @@ pub(crate) async fn build_realtime_session_config( (false, true) => prompt, (false, false) => format!("{prompt}\n\n{startup_context}"), }; + if version != RealtimeWsVersion::V3 && !params.initial_items.is_empty() { + return Err(CodexErr::InvalidRequest( + "initial realtime items require realtime v3".to_string(), + )); + } + if params.initial_items.len() > REALTIME_INITIAL_ITEMS_MAX_COUNT { + return Err(CodexErr::InvalidRequest(format!( + "initial realtime items must contain no more than {REALTIME_INITIAL_ITEMS_MAX_COUNT} items" + ))); + } + let mut total_initial_item_tokens: usize = 0; + for item in ¶ms.initial_items { + let item_tokens = approx_token_count(&item.text); + if item_tokens > REALTIME_INITIAL_ITEMS_MAX_TOKENS { + return Err(CodexErr::InvalidRequest(format!( + "each initial realtime item must not exceed {REALTIME_INITIAL_ITEMS_MAX_TOKENS} estimated tokens" + ))); + } + total_initial_item_tokens = total_initial_item_tokens.saturating_add(item_tokens); + } + if total_initial_item_tokens > REALTIME_INITIAL_ITEMS_MAX_TOKENS { + return Err(CodexErr::InvalidRequest(format!( + "initial realtime items must not exceed {REALTIME_INITIAL_ITEMS_MAX_TOKENS} estimated tokens in total" + ))); + } let model = Some( params .model @@ -1277,6 +1305,7 @@ pub(crate) async fn build_realtime_session_config( validate_realtime_voice(version, voice)?; Ok(RealtimeSessionConfig { instructions: prompt, + initial_items: params.initial_items.clone(), model, session_id: Some( params diff --git a/codex-rs/core/tests/suite/compact_remote.rs b/codex-rs/core/tests/suite/compact_remote.rs index b021630bf438..47d373375c65 100644 --- a/codex-rs/core/tests/suite/compact_remote.rs +++ b/codex-rs/core/tests/suite/compact_remote.rs @@ -217,6 +217,7 @@ async fn start_realtime_conversation(codex: &codex_core::CodexThread) -> Result< model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index df143e89628e..b83fc9c2a0ba 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -98,6 +98,7 @@ mod prompt_caching; mod prompt_debug_tests; mod quota_exceeded; mod realtime_conversation; +mod realtime_initial_items; mod remote_env; mod remote_models; mod request_compression; diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index 7cdb200624bf..fc250eb095df 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -292,6 +292,7 @@ async fn conversation_start_audio_text_close_round_trip() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -440,6 +441,7 @@ async fn conversation_start_defaults_to_v2_and_gpt_realtime_1_5() -> Result<()> model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -537,6 +539,7 @@ async fn conversation_webrtc_start_posts_generated_session() -> Result<()> { model: Some("session-override-model".to_string()), output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: Some(ConversationStartTransport::Webrtc { @@ -726,6 +729,7 @@ async fn conversation_webrtc_start_uses_avas_query() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: Some(ConversationStartTransport::Webrtc { @@ -826,6 +830,7 @@ async fn conversation_webrtc_default_v1_ignores_configured_v2_voice() -> Result< model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: Some(ConversationStartTransport::Webrtc { @@ -888,6 +893,7 @@ async fn conversation_webrtc_default_v1_rejects_explicit_v2_voice() -> Result<() model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: Some(ConversationStartTransport::Webrtc { @@ -960,6 +966,7 @@ async fn conversation_webrtc_start_uses_configured_call_base_url_for_avas() -> R model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: Some(ConversationStartTransport::Webrtc { @@ -1056,6 +1063,7 @@ async fn conversation_webrtc_close_while_sideband_connecting_drops_pending_join( model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: Some(ConversationStartTransport::Webrtc { @@ -1149,6 +1157,7 @@ async fn conversation_webrtc_sideband_connect_failure_closes_with_error() -> Res model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: Some(ConversationStartTransport::Webrtc { @@ -1244,6 +1253,7 @@ async fn conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth() -> model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -1337,6 +1347,7 @@ async fn assert_transport_close_tail_flush( model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -1455,6 +1466,7 @@ async fn conversation_start_preflight_failure_emits_realtime_error_only() -> Res model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -1508,6 +1520,7 @@ async fn conversation_start_connect_failure_emits_realtime_error_only() -> Resul model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -1609,6 +1622,7 @@ async fn conversation_second_start_replaces_runtime() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("old".to_string())), realtime_session_id: Some("conv_old".to_string()), transport: None, @@ -1641,6 +1655,7 @@ async fn conversation_second_start_replaces_runtime() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("new".to_string())), realtime_session_id: Some("conv_new".to_string()), transport: None, @@ -1744,6 +1759,7 @@ async fn conversation_uses_experimental_realtime_ws_base_url_override() -> Resul model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -1815,6 +1831,7 @@ async fn conversation_uses_default_realtime_backend_prompt() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: None, realtime_session_id: None, transport: None, @@ -1894,6 +1911,7 @@ async fn conversation_uses_empty_instructions_for_null_or_empty_prompt() -> Resu model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt, realtime_session_id: None, transport: None, @@ -1966,6 +1984,7 @@ async fn conversation_uses_explicit_start_voice() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -2030,6 +2049,7 @@ async fn conversation_uses_configured_realtime_voice() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -2082,6 +2102,7 @@ async fn conversation_rejects_voice_for_wrong_realtime_version() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -2135,6 +2156,7 @@ async fn conversation_uses_experimental_realtime_ws_backend_prompt_override() -> model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("prompt from op".to_string())), realtime_session_id: None, transport: None, @@ -2214,6 +2236,7 @@ async fn conversation_uses_experimental_realtime_ws_startup_context_override() - model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("prompt from op".to_string())), realtime_session_id: None, transport: None, @@ -2287,6 +2310,7 @@ async fn conversation_disables_realtime_startup_context_with_empty_override() -> model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("prompt from op".to_string())), realtime_session_id: None, transport: None, @@ -2353,6 +2377,7 @@ async fn conversation_start_injects_startup_context_from_thread_history() -> Res model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -2474,6 +2499,7 @@ async fn conversation_startup_context_current_thread_selects_many_turns_by_budge model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -2588,6 +2614,7 @@ async fn conversation_startup_context_falls_back_to_workspace_map() -> Result<() model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -2654,6 +2681,7 @@ async fn conversation_startup_context_is_truncated_and_sent_once_per_start() -> model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -2741,6 +2769,7 @@ async fn conversation_user_text_turn_is_not_sent_to_realtime() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -2844,6 +2873,7 @@ async fn realtime_v2_noop_tool_call_returns_empty_function_output_without_respon model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -2949,6 +2979,7 @@ async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Re model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -3115,6 +3146,7 @@ async fn conversation_flushes_assistant_deltas_every_200ms_for_v3_handoff() -> R model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -3283,6 +3315,7 @@ async fn conversation_handoff_persists_across_item_done_until_turn_complete() -> model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -3441,6 +3474,7 @@ async fn inbound_handoff_request_starts_turn() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -3557,6 +3591,7 @@ async fn inbound_handoff_request_uses_active_transcript() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -3666,6 +3701,7 @@ async fn inbound_handoff_request_sends_transcript_delta_after_each_handoff() -> model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -3794,6 +3830,7 @@ async fn conversation_close_routes_only_remaining_transcript_tail_once() -> Resu model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -3883,6 +3920,7 @@ async fn inbound_conversation_item_does_not_start_turn_and_still_forwards_audio( model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -4012,6 +4050,7 @@ async fn delegated_turn_user_role_echo_does_not_redelegate_and_still_forwards_au model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -4171,6 +4210,7 @@ async fn inbound_handoff_request_does_not_block_realtime_event_forwarding() -> R model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -4319,6 +4359,7 @@ async fn inbound_handoff_request_steers_active_turn() -> Result<()> { model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, @@ -4478,6 +4519,7 @@ async fn inbound_handoff_request_starts_turn_and_does_not_block_realtime_audio() model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: true, + initial_items: Vec::new(), prompt: Some(Some("backend prompt".to_string())), realtime_session_id: None, transport: None, diff --git a/codex-rs/core/tests/suite/realtime_initial_items.rs b/codex-rs/core/tests/suite/realtime_initial_items.rs new file mode 100644 index 000000000000..4eb5673857e5 --- /dev/null +++ b/codex-rs/core/tests/suite/realtime_initial_items.rs @@ -0,0 +1,203 @@ +use anyhow::Result; +use codex_config::config_toml::RealtimeWsVersion; +use codex_protocol::protocol::CodexResponseHandoffMode; +use codex_protocol::protocol::ConversationStartParams; +use codex_protocol::protocol::ConversationTextParams; +use codex_protocol::protocol::ConversationTextRole; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::Op; +use codex_protocol::protocol::RealtimeConversationRealtimeEvent; +use codex_protocol::protocol::RealtimeConversationVersion; +use codex_protocol::protocol::RealtimeEvent; +use codex_protocol::protocol::RealtimeOutputModality; +use core_test_support::responses::start_mock_server; +use core_test_support::responses::start_websocket_server; +use core_test_support::skip_if_no_network; +use core_test_support::test_codex::test_codex; +use core_test_support::wait_for_event; +use core_test_support::wait_for_event_match; +use pretty_assertions::assert_eq; +use serde_json::json; +use std::time::Duration; +use tokio::time::timeout; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn frameless_v3_sends_initial_items_in_session_bootstrap() -> Result<()> { + skip_if_no_network!(Ok(())); + + let api_server = start_mock_server().await; + let realtime_server = start_websocket_server(vec![vec![vec![json!({ + "type": "session.started", + "session": { "id": "sess_initial_items", "instructions": "backend prompt" } + })]]]) + .await; + let mut builder = test_codex().with_config({ + let realtime_base_url = realtime_server.uri().to_string(); + move |config| { + config.experimental_realtime_ws_base_url = Some(realtime_base_url); + config.experimental_realtime_ws_startup_context = Some(String::new()); + config.realtime.version = RealtimeWsVersion::V3; + } + }); + let test = builder.build_with_auto_env(&api_server).await?; + + test.codex + .submit(Op::RealtimeConversationStart(start_params( + RealtimeConversationVersion::V3, + ))) + .await?; + + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::RealtimeConversationStarted(_)) + }) + .await; + let request = timeout( + Duration::from_secs(2), + realtime_server.wait_for_request(/*connection_index*/ 0, /*request_index*/ 0), + ) + .await?; + let body = request.body_json(); + + assert_eq!(body["type"], "session.update"); + assert_eq!(body["session"]["instructions"], "backend prompt"); + assert_eq!( + body["session"]["initial_items"], + json!([ + { + "type": "message", + "role": "developer", + "content": [{"type": "input_text", "text": "Remember this."}], + }, + { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "What do you remember?"}], + }, + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "I remember."}], + }, + ]) + ); + + realtime_server.shutdown().await; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn initial_items_require_frameless_v3() -> Result<()> { + skip_if_no_network!(Ok(())); + + assert_start_error( + start_params(RealtimeConversationVersion::V2), + "initial realtime items require realtime v3", + ) + .await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn initial_items_enforce_count_limit() -> Result<()> { + skip_if_no_network!(Ok(())); + + let mut params = start_params(RealtimeConversationVersion::V3); + params.initial_items = vec![ + ConversationTextParams { + text: "item".to_string(), + role: ConversationTextRole::User, + }; + 129 + ]; + assert_start_error( + params, + "initial realtime items must contain no more than 128 items", + ) + .await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn initial_items_enforce_per_item_token_limit() -> Result<()> { + skip_if_no_network!(Ok(())); + + let mut params = start_params(RealtimeConversationVersion::V3); + params.initial_items = vec![ConversationTextParams { + text: "x".repeat(8_192 * 4 + 1), + role: ConversationTextRole::User, + }]; + assert_start_error( + params, + "each initial realtime item must not exceed 8192 estimated tokens", + ) + .await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn initial_items_enforce_aggregate_token_limit() -> Result<()> { + skip_if_no_network!(Ok(())); + + let mut params = start_params(RealtimeConversationVersion::V3); + params.initial_items = vec![ + ConversationTextParams { + text: "x".repeat(8_192 * 2 + 1), + role: ConversationTextRole::User, + }; + 2 + ]; + assert_start_error( + params, + "initial realtime items must not exceed 8192 estimated tokens in total", + ) + .await +} + +async fn assert_start_error(params: ConversationStartParams, expected_error: &str) -> Result<()> { + let api_server = start_mock_server().await; + let test = test_codex().build_with_auto_env(&api_server).await?; + test.codex + .submit(Op::RealtimeConversationStart(params)) + .await?; + let error = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { + payload: RealtimeEvent::Error(message), + }) => Some(message.clone()), + _ => None, + }) + .await; + assert!( + error.contains(expected_error), + "expected error to contain {expected_error:?}, got {error:?}" + ); + Ok(()) +} + +fn start_params(version: RealtimeConversationVersion) -> ConversationStartParams { + ConversationStartParams { + client_managed_handoffs: false, + flush_transcript_tail_on_session_end: false, + codex_responses_as_items: false, + codex_response_item_prefix: None, + codex_response_handoff_mode: CodexResponseHandoffMode::Thinking, + model: None, + output_modality: RealtimeOutputModality::Audio, + include_startup_context: true, + initial_items: vec![ + ConversationTextParams { + text: "Remember this.".to_string(), + role: ConversationTextRole::Developer, + }, + ConversationTextParams { + text: "What do you remember?".to_string(), + role: ConversationTextRole::User, + }, + ConversationTextParams { + text: "I remember.".to_string(), + role: ConversationTextRole::Assistant, + }, + ], + prompt: Some(Some("backend prompt".to_string())), + realtime_session_id: None, + transport: None, + version: Some(version), + voice: None, + } +} diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index b6c407eac6cd..201b3c4834b4 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -222,6 +222,8 @@ pub struct ConversationStartParams { pub output_modality: RealtimeOutputModality, /// Whether to append Codex's startup context to the realtime backend prompt. pub include_startup_context: bool, + /// Complete role-bearing text items to include in the initial realtime session history. + pub initial_items: Vec, pub prompt: Option>, pub realtime_session_id: Option, pub transport: Option, @@ -428,7 +430,7 @@ pub struct ConversationAudioParams { pub frame: RealtimeAudioFrame, } -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct ConversationTextParams { pub text: String, pub role: ConversationTextRole,