openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-18 17:18:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Seed realtime V3 sessions with initial text items (#34067)

Browse Source 
## What changed

- Add an optional `initialItems` field to `thread/realtime/start` for seeding
  realtime V3 session history with `user`, `developer`, and `assistant` text.
- Encode the items in Frameless Bidi session bootstrap payloads for websocket
  and WebRTC transports while preserving existing payloads when the list is
  omitted or empty.
- Reject non-empty initial items for V1 and V2, more than 128 items, any item
  over 8,192 estimated tokens, or a combined total over 8,192 estimated tokens.

## Testing

- Cover protocol serialization, app-server forwarding, websocket and WebRTC
  payloads, empty-list compatibility, version checks, and size limits.

GitOrigin-RevId: 09af367f9a07543e3f8d59b7fa5387f932dc5a6a
This commit is contained in:
jiayuhuang-openai
2026-07-18 20:59:57 +00:00
committed by copyberry
parent 35eaf3ffb0
commit 312caf176a
25 changed files with 657 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/json/ClientRequest.json
Generated
+16
View File
@@ -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": [
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
Generated
+16
View File
@@ -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.",
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
Generated
+16
View File
@@ -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.",
codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeInitialItem.ts
Generated
+9
View File
@@ -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, };
codex-rs/app-server-protocol/schema/typescript/v2/index.ts
Generated
+1
View File
@@ -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";
codex-rs/app-server-protocol/src/protocol/common.rs
+26
View File
@@ -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,
codex-rs/app-server-protocol/src/protocol/v2/realtime.rs
+14
View File
@@ -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<bool>,
/// 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<Vec<ThreadRealtimeInitialItem>>,
#[serde(
default,
deserialize_with = "crate::protocol::serde_helpers::deserialize_double_option",
@@ -115,6 +120,15 @@ pub struct ThreadRealtimeStartParams {
pub voice: Option<RealtimeVoice>,
}
/// 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")]
codex-rs/app-server-protocol/src/protocol/v2/tests.rs
+11
View File
@@ -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::<ThreadRealtimeStartParams>(json!({
"threadId": "thread_123",
"outputModality": "audio",
}))
.expect("params should deserialize");
assert_eq!(params.initial_items, None);
}
codex-rs/app-server/README.md
+27 -1
View File
@@ -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`.
codex-rs/app-server/src/request_processors/turn_processor.rs
+9
View File
@@ -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 {
codex-rs/app-server/tests/suite/v2/experimental_api.rs
+2
View File
@@ -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 {
codex-rs/app-server/tests/suite/v2/realtime_conversation.rs
+65 -1
View File
@@ -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<CodexResponseHandoffMode>,
initial_items: Option<Vec<ThreadRealtimeInitialItem>>,
) -> Result<ThreadRealtimeStartedNotification> {
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::<TurnCompletedNotification>("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,
codex-rs/codex-api/src/endpoint/realtime_call.rs
+30 -4
View File
@@ -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]
codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs
+9
View File
@@ -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<ConversationTextParams>,
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,
codex-rs/codex-api/src/endpoint/realtime_websocket/methods_common.rs
+6 -1
View File
@@ -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<ConversationTextParams>,
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,
)),
}
codex-rs/codex-api/src/endpoint/realtime_websocket/methods_frameless_bidi.rs
+28 -1
View File
@@ -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<ConversationTextParams>,
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<String>,
instructions: String,
initial_items: Vec<ConversationTextParams>,
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
}
codex-rs/codex-api/src/endpoint/realtime_websocket/methods_frameless_bidi_tests.rs
+85
View File
@@ -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.",
}],
},
])
);
}
codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs
+2
View File
@@ -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<ConversationTextParams>,
pub model: Option<String>,
pub session_id: Option<String>,
pub event_parser: RealtimeEventParser,
codex-rs/codex-api/tests/realtime_websocket_e2e.rs
+6
View File
@@ -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,
codex-rs/core/src/realtime_conversation.rs
+29
View File
@@ -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 &params.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
codex-rs/core/tests/suite/compact_remote.rs
+1
View File
@@ -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,
codex-rs/core/tests/suite/mod.rs
+1
View File
@@ -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;
codex-rs/core/tests/suite/realtime_conversation.rs
+42
View File
@@ -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,
codex-rs/core/tests/suite/realtime_initial_items.rs
+203
View File
@@ -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,
}
}
codex-rs/protocol/src/protocol.rs
+3 -1
View File
@@ -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<ConversationTextParams>,
pub prompt: Option<Option<String>>,
pub realtime_session_id: Option<String>,
pub transport: Option<ConversationStartTransport>,
@@ -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,