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

Stream realtime V3 Codex handoff output (#33856)

Browse Source 
## What changed

- Stream V3 agent message deltas to `delegation.context.append` at 200 ms intervals instead of waiting for each completed message.
- Send commentary without a prefix and label final or phase-less V1/V3 output with `"Agent Final Message":`.
- Bound streamed output while preserving both its beginning and final tail when truncation is required.
- Remove `codexResponseHandoffPrefix` from `thread/realtime/start`; unknown fields from older clients remain ignored.

## Testing

- Cover incremental V3 commentary and final output, Unicode-safe truncation, V1/V3 phase labeling, and legacy request compatibility.

GitOrigin-RevId: 633c18b5eb9a4f5756a264b63b7bfc185b073951
This commit is contained in:
jiayuhuang-openai
2026-07-17 15:54:33 +00:00
committed by copyberry
parent 22f413bdbf
commit b03545b5b8
13 changed files with 583 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/src/protocol/common.rs
+2 -7
View File
@@ -3390,7 +3390,6 @@ mod tests {
flush_transcript_tail_on_session_end: Some(true),
codex_responses_as_items: None,
codex_response_item_prefix: None,
codex_response_handoff_prefix: Some("silent context".to_string()),
thread_id: "thr_123".to_string(),
model: Some("realtime-treatment-model".to_string()),
output_modality: RealtimeOutputModality::Audio,
@@ -3412,7 +3411,6 @@ mod tests {
"flushTranscriptTailOnSessionEnd": true,
"codexResponsesAsItems": null,
"codexResponseItemPrefix": null,
"codexResponseHandoffPrefix": "silent context",
"model": "realtime-treatment-model",
"outputModality": "audio",
"includeStartupContext": false,
@@ -3437,7 +3435,6 @@ mod tests {
flush_transcript_tail_on_session_end: None,
codex_responses_as_items: None,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
thread_id: "thr_123".to_string(),
model: None,
output_modality: RealtimeOutputModality::Audio,
@@ -3459,7 +3456,6 @@ mod tests {
"flushTranscriptTailOnSessionEnd": null,
"codexResponsesAsItems": null,
"codexResponseItemPrefix": null,
"codexResponseHandoffPrefix": null,
"model": null,
"outputModality": "audio",
"includeStartupContext": null,
@@ -3479,7 +3475,6 @@ mod tests {
flush_transcript_tail_on_session_end: None,
codex_responses_as_items: None,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
thread_id: "thr_123".to_string(),
model: None,
output_modality: RealtimeOutputModality::Audio,
@@ -3501,7 +3496,6 @@ mod tests {
"flushTranscriptTailOnSessionEnd": null,
"codexResponsesAsItems": null,
"codexResponseItemPrefix": null,
"codexResponseHandoffPrefix": null,
"model": null,
"outputModality": "audio",
"includeStartupContext": null,
@@ -3520,6 +3514,8 @@ mod tests {
"id": 9,
"params": {
"threadId": "thr_123",
// Retain runtime compatibility with clients that have not yet removed this field.
"codexResponseHandoffPrefix": "",
"outputModality": "audio",
"realtimeSessionId": null,
"transport": null,
@@ -3719,7 +3715,6 @@ mod tests {
flush_transcript_tail_on_session_end: None,
codex_responses_as_items: None,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
thread_id: "thr_123".to_string(),
model: None,
output_modality: RealtimeOutputModality::Audio,
codex-rs/app-server-protocol/src/protocol/v2/realtime.rs
-5
View File
@@ -80,11 +80,6 @@ pub struct ThreadRealtimeStartParams {
/// Optional prefix added to automatic Codex response items when `codexResponsesAsItems` is true.
#[ts(optional = nullable)]
pub codex_response_item_prefix: Option<String>,
/// Optional prefix added to automatic V1 or V3 Codex commentary sent through the selected
/// Bidi handoff wire event when `codexResponsesAsItems` is not true. Final answers are sent
/// without the prefix.
#[ts(optional = nullable)]
pub codex_response_handoff_prefix: Option<String>,
/// Overrides the configured realtime model for this session only.
#[ts(optional = nullable)]
pub model: Option<String>,
codex-rs/app-server/README.md
+6 -5
View File
@@ -172,7 +172,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.*`. By default, automatic Codex text follows the protocol's speakable output path. 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. For V1 and V3 sessions, pass `codexResponseHandoffPrefix` while item mode is disabled to route automatic Codex commentary through the selected Bidi handoff wire event with that prefix; final answers remain unprefixed. 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, 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.*`. V1 and V3 send commentary as raw handoff appends and label final or phase-less agent output with `"Agent Final Message":`; V3 streams these appends as Codex produces text deltas. 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 `{}`.
@@ -921,10 +921,11 @@ Pass `codexResponsesAsItems: true` to inject automatic Codex responses with
path. When using that mode, `codexResponseItemPrefix` can prepend short
experiment instructions to each automatic Codex response item. Omit
`codexResponsesAsItems`, or pass `false`, to preserve the default speakable
behavior. For V1 and V3 sessions, `codexResponseHandoffPrefix` instead routes
automatic Codex commentary through the selected Bidi handoff wire event and
prepends the provided text. Final answers remain unprefixed. Item mode takes
precedence when `codexResponsesAsItems` is true.
behavior. In that default mode, V1 and V3 append commentary without a prefix
and label final or phase-less agent output with `"Agent Final Message":`. V3
emits each text delta as a `delegation.context.append` instead of waiting for
the completed agent message. Older clients may continue to send the removed
`codexResponseHandoffPrefix` field; the server ignores unknown request fields.
Call
`thread/realtime/appendText` to append app-provided realtime text items, or
`thread/realtime/appendSpeech` when the app decides a realtime update should be
codex-rs/app-server/src/request_processors/turn_processor.rs
-1
View File
@@ -1077,7 +1077,6 @@ impl TurnRequestProcessor {
.unwrap_or(false),
codex_responses_as_items: params.codex_responses_as_items.unwrap_or(false),
codex_response_item_prefix: params.codex_response_item_prefix,
codex_response_handoff_prefix: params.codex_response_handoff_prefix,
model: params.model,
output_modality: params.output_modality,
include_startup_context: params.include_startup_context.unwrap_or(true),
codex-rs/app-server/tests/suite/v2/experimental_api.rs
-2
View File
@@ -93,7 +93,6 @@ async fn realtime_conversation_start_requires_experimental_api_capability() -> R
flush_transcript_tail_on_session_end: None,
codex_responses_as_items: None,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
thread_id: "thr_123".to_string(),
model: None,
output_modality: RealtimeOutputModality::Audio,
@@ -222,7 +221,6 @@ async fn realtime_webrtc_start_requires_experimental_api_capability() -> Result<
flush_transcript_tail_on_session_end: None,
codex_responses_as_items: None,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
thread_id: "thr_123".to_string(),
model: None,
output_modality: RealtimeOutputModality::Audio,
codex-rs/app-server/tests/suite/v2/realtime_conversation.rs
+1 -70
View File
@@ -88,8 +88,6 @@ const V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT: &str =
"Background agent finished. Use the preceding [BACKEND] messages as the result.";
const RESPONSE_ITEM_PREFIX: &str =
"Use the following context to inform future responses, but do not speak it to the user.";
const RESPONSE_HANDOFF_PREFIX: &str =
"Silent Codex context. Do not speak, acknowledge, or summarize this item.";
#[derive(Debug, Clone, Copy)]
enum StartupContextConfig<'a> {
@@ -334,7 +332,6 @@ impl RealtimeE2eHarness {
offer_sdp,
/*client_managed_handoffs*/ None,
/*codex_responses_as_items*/ None,
/*codex_response_handoff_prefix*/ None,
RealtimeConversationVersion::V1,
)
.await
@@ -348,7 +345,6 @@ impl RealtimeE2eHarness {
offer_sdp,
/*client_managed_handoffs*/ None,
/*codex_responses_as_items*/ Some(true),
/*codex_response_handoff_prefix*/ None,
RealtimeConversationVersion::V1,
)
.await
@@ -359,7 +355,6 @@ impl RealtimeE2eHarness {
offer_sdp: &str,
client_managed_handoffs: Option<bool>,
codex_responses_as_items: Option<bool>,
codex_response_handoff_prefix: Option<&str>,
version: RealtimeConversationVersion,
) -> Result<StartedWebrtcRealtime> {
// Starts realtime through the public JSON-RPC method, then waits for the same client-visible
@@ -373,7 +368,6 @@ impl RealtimeE2eHarness {
codex_response_item_prefix: codex_responses_as_items
.unwrap_or(false)
.then(|| RESPONSE_ITEM_PREFIX.to_string()),
codex_response_handoff_prefix: codex_response_handoff_prefix.map(str::to_string),
codex_responses_as_items,
model: None,
output_modality: RealtimeOutputModality::Audio,
@@ -434,7 +428,6 @@ impl RealtimeE2eHarness {
codex_response_item_prefix: codex_responses_as_items
.unwrap_or(false)
.then(|| RESPONSE_ITEM_PREFIX.to_string()),
codex_response_handoff_prefix: None,
codex_responses_as_items,
model: None,
output_modality: RealtimeOutputModality::Audio,
@@ -466,7 +459,6 @@ impl RealtimeE2eHarness {
client_managed_handoffs: None,
flush_transcript_tail_on_session_end: None,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
codex_responses_as_items: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
@@ -738,7 +730,6 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
flush_transcript_tail_on_session_end: None,
codex_responses_as_items: None,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
thread_id: thread_start.thread.id.clone(),
model: Some("realtime-treatment-model".to_string()),
output_modality: RealtimeOutputModality::Audio,
@@ -1033,7 +1024,6 @@ async fn realtime_start_can_skip_startup_context() -> Result<()> {
flush_transcript_tail_on_session_end: None,
codex_responses_as_items: None,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
thread_id: thread_start.thread.id.clone(),
model: None,
output_modality: RealtimeOutputModality::Audio,
@@ -1135,7 +1125,6 @@ async fn realtime_text_output_modality_requests_text_output_and_final_transcript
flush_transcript_tail_on_session_end: None,
codex_responses_as_items: None,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
thread_id: thread_start.thread.id.clone(),
model: None,
output_modality: RealtimeOutputModality::Text,
@@ -1323,7 +1312,6 @@ async fn realtime_conversation_stop_emits_closed_notification() -> Result<()> {
flush_transcript_tail_on_session_end: None,
codex_responses_as_items: None,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
thread_id: thread_start.thread.id.clone(),
model: None,
output_modality: RealtimeOutputModality::Audio,
@@ -1431,7 +1419,6 @@ async fn realtime_webrtc_start_emits_sdp_notification() -> Result<()> {
flush_transcript_tail_on_session_end: None,
codex_responses_as_items: None,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
thread_id: thread_id.clone(),
model: None,
output_modality: RealtimeOutputModality::Audio,
@@ -1625,7 +1612,6 @@ async fn webrtc_v3_start_posts_live_session_and_joins_without_session_update() -
"v=offer\r\n",
/*client_managed_handoffs*/ None,
/*codex_responses_as_items*/ None,
/*codex_response_handoff_prefix*/ None,
RealtimeConversationVersion::V3,
)
.await?;
@@ -1727,7 +1713,7 @@ async fn webrtc_v1_default_automatic_output_uses_handoff_append() -> Result<()>
json!({
"type": "conversation.handoff.append",
"handoff_id": "codex",
"output_text": "legacy automatic speech",
"output_text": "\"Agent Final Message\":\n\nlegacy automatic speech",
})
);
@@ -1756,7 +1742,6 @@ async fn webrtc_v1_client_managed_handoffs_disable_automatic_output() -> Result<
"v=offer\r\n",
/*client_managed_handoffs*/ Some(true),
/*codex_responses_as_items*/ None,
/*codex_response_handoff_prefix*/ None,
RealtimeConversationVersion::V1,
)
.await?;
@@ -1814,58 +1799,6 @@ async fn webrtc_v1_client_managed_handoffs_disable_automatic_output() -> Result<
Ok(())
}
#[tokio::test]
async fn webrtc_v1_final_automatic_handoff_omits_silent_prefix() -> Result<()> {
skip_if_no_network!(Ok(()));
let mut harness = RealtimeE2eHarness::new(
RealtimeTestVersion::V1,
main_loop_responses(vec![create_final_assistant_message_sse_response(
"background progress",
)?]),
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![
session_updated("sess_v1_prefixed_handoff"),
json!({
"type": "conversation.handoff.requested",
"handoff_id": "handoff_prefixed",
"item_id": "item_prefixed",
"input_transcript": "run the background task"
}),
],
vec![],
vec![],
])]),
)
.await?;
let started = harness
.start_webrtc_realtime_with_codex_response_routing(
"v=offer\r\n",
/*client_managed_handoffs*/ None,
/*codex_responses_as_items*/ None,
Some(RESPONSE_HANDOFF_PREFIX),
RealtimeConversationVersion::V1,
)
.await?;
assert_eq!(started.started.version, RealtimeConversationVersion::V1);
let _ = harness
.read_notification::<TurnCompletedNotification>("turn/completed")
.await?;
assert_eq!(
harness.sideband_outbound_request(/*request_index*/ 1).await,
json!({
"type": "conversation.handoff.append",
"handoff_id": "handoff_prefixed",
"output_text": "background progress",
})
);
harness.shutdown().await;
Ok(())
}
#[tokio::test]
async fn webrtc_v1_handoff_request_delegates_context_and_manual_append_speaks() -> Result<()> {
skip_if_no_network!(Ok(()));
@@ -2980,7 +2913,6 @@ async fn realtime_webrtc_start_surfaces_backend_error() -> Result<()> {
flush_transcript_tail_on_session_end: None,
codex_responses_as_items: None,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
thread_id: thread_start.thread.id,
model: None,
output_modality: RealtimeOutputModality::Audio,
@@ -3050,7 +2982,6 @@ async fn realtime_conversation_requires_feature_flag() -> Result<()> {
flush_transcript_tail_on_session_end: None,
codex_responses_as_items: None,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
thread_id: thread_start.thread.id.clone(),
model: None,
output_modality: RealtimeOutputModality::Audio,
codex-rs/core/src/realtime_conversation.rs
+306 -32
View File
@@ -54,13 +54,18 @@ use codex_protocol::protocol::RealtimeOutputModality;
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::take_bytes_at_char_boundary;
use http::HeaderMap;
use http::HeaderValue;
use http::header::AUTHORIZATION;
use serde_json::json;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::time::Duration;
use std::time::Instant;
use tokio::sync::Mutex;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
@@ -75,6 +80,9 @@ 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 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";
const STANDALONE_HANDOFF_ID: &str = "codex";
const DEFAULT_REALTIME_MODEL: &str = "gpt-realtime-1.5";
const DEFAULT_FRAMELESS_REALTIME_MODEL: &str = "gpt-live-1-boulder-alpha";
@@ -113,13 +121,137 @@ enum RealtimeSessionKind {
#[derive(Clone, Debug)]
struct RealtimeHandoffState {
output_tx: Sender<RealtimeOutbound>,
active_handoff: Arc<Mutex<Option<String>>>,
last_output_text: Arc<Mutex<Option<String>>>,
stream: Arc<Mutex<RealtimeHandoffStreamState>>,
client_managed_handoffs: bool,
codex_responses_as_items: bool,
codex_response_item_prefix: Option<String>,
codex_response_handoff_prefix: Option<String>,
session_kind: RealtimeSessionKind,
event_parser: RealtimeEventParser,
}
#[derive(Debug, Default)]
struct RealtimeHandoffStreamState {
active_handoff: Option<String>,
items: HashMap<String, RealtimeStreamedItem>,
}
#[derive(Debug)]
struct RealtimeStreamedItem {
handoff_id: String,
phase: Option<MessagePhase>,
sent_bytes: usize,
buffered_text: String,
tail_text: String,
truncated: bool,
last_flush_at: Instant,
flush_scheduled: bool,
}
impl RealtimeStreamedItem {
fn next_flush_delay(&self) -> Duration {
HANDOFF_STREAM_FLUSH_INTERVAL.saturating_sub(self.last_flush_at.elapsed())
}
fn output_prefix(&self) -> &'static str {
if self.sent_bytes == 0 && !matches!(self.phase, Some(MessagePhase::Commentary)) {
AGENT_FINAL_MESSAGE_PREFIX
} else {
""
}
}
fn stream_head_byte_limit(&self) -> usize {
let output_byte_limit = approx_bytes_for_tokens(REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET);
output_byte_limit.saturating_sub(HANDOFF_STREAM_TRUNCATION_MARKER.len()) / 2
}
fn tail_byte_limit(&self) -> usize {
approx_bytes_for_tokens(REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET)
.saturating_sub(self.stream_head_byte_limit())
.saturating_sub(HANDOFF_STREAM_TRUNCATION_MARKER.len())
}
fn streamable_text_bytes(&self) -> usize {
self.stream_head_byte_limit()
.saturating_sub(self.sent_bytes)
.saturating_sub(self.output_prefix().len())
}
fn push_text(&mut self, text: &str) {
if text.is_empty() {
return;
}
if self.truncated {
self.tail_text.push_str(text);
self.tail_text =
take_last_bytes_at_char_boundary(&self.tail_text, self.tail_byte_limit())
.to_string();
return;
}
self.buffered_text.push_str(text);
let output_byte_limit = approx_bytes_for_tokens(REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET);
let remaining_text_bytes = output_byte_limit
.saturating_sub(self.sent_bytes)
.saturating_sub(self.output_prefix().len());
if self.buffered_text.len() <= remaining_text_bytes {
return;
}
let head_bytes =
take_bytes_at_char_boundary(&self.buffered_text, self.streamable_text_bytes()).len();
self.tail_text =
take_last_bytes_at_char_boundary(&self.buffered_text, self.tail_byte_limit())
.to_string();
self.buffered_text.truncate(head_bytes);
self.truncated = true;
}
fn drain_stream_chunk(&mut self) -> Option<String> {
let prefix = self.output_prefix();
let available_text_bytes = self.streamable_text_bytes();
if self.buffered_text.is_empty() || available_text_bytes == 0 {
return None;
}
let requested_bytes = available_text_bytes.min(self.buffered_text.len());
let split_at = take_bytes_at_char_boundary(&self.buffered_text, requested_bytes).len();
if split_at == 0 {
return None;
}
let text = self.buffered_text.drain(..split_at).collect::<String>();
let text = format!("{prefix}{text}");
self.sent_bytes += text.len();
Some(text)
}
fn drain_final_chunk(&mut self) -> Option<String> {
let prefix = self.output_prefix();
if !self.truncated {
if self.buffered_text.is_empty() {
return None;
}
let text = self.buffered_text.drain(..).collect::<String>();
let text = format!("{prefix}{text}");
self.sent_bytes += text.len();
return Some(text);
}
let head = self.buffered_text.drain(..).collect::<String>();
let tail = self.tail_text.drain(..).collect::<String>();
let text = format!("{prefix}{head}{HANDOFF_STREAM_TRUNCATION_MARKER}{tail}");
self.sent_bytes += text.len();
Some(text)
}
}
fn take_last_bytes_at_char_boundary(text: &str, max_bytes: usize) -> &str {
let mut start = text.len().saturating_sub(max_bytes);
while start < text.len() && !text.is_char_boundary(start) {
start += 1;
}
&text[start..]
}
#[derive(Debug, PartialEq, Eq)]
@@ -227,20 +359,26 @@ impl RealtimeHandoffState {
client_managed_handoffs: bool,
codex_responses_as_items: bool,
codex_response_item_prefix: Option<String>,
codex_response_handoff_prefix: Option<String>,
session_kind: RealtimeSessionKind,
event_parser: RealtimeEventParser,
) -> Self {
Self {
output_tx,
active_handoff: Arc::new(Mutex::new(None)),
last_output_text: Arc::new(Mutex::new(None)),
stream: Arc::new(Mutex::new(RealtimeHandoffStreamState::default())),
client_managed_handoffs,
codex_responses_as_items,
codex_response_item_prefix,
codex_response_handoff_prefix,
session_kind,
event_parser,
}
}
fn streams_handoff_append(&self) -> bool {
self.event_parser == RealtimeEventParser::FramelessBidi
&& !self.client_managed_handoffs
&& !self.codex_responses_as_items
}
}
#[allow(dead_code)]
@@ -262,7 +400,6 @@ struct RealtimeStart {
flush_transcript_tail_on_session_end: bool,
codex_responses_as_items: bool,
codex_response_item_prefix: Option<String>,
codex_response_handoff_prefix: Option<String>,
realtime_call_api_provider: Option<ApiProvider>,
session_config: RealtimeSessionConfig,
model_client: ModelClient,
@@ -321,7 +458,6 @@ impl RealtimeConversationManager {
flush_transcript_tail_on_session_end,
codex_responses_as_items,
codex_response_item_prefix,
codex_response_handoff_prefix,
realtime_call_api_provider,
session_config,
model_client,
@@ -350,8 +486,8 @@ impl RealtimeConversationManager {
client_managed_handoffs,
codex_responses_as_items,
codex_response_item_prefix,
codex_response_handoff_prefix,
session_kind,
event_parser,
);
let input_channels = RealtimeInputChannels {
text_rx,
@@ -532,11 +668,8 @@ impl RealtimeConversationManager {
if handoff.client_managed_handoffs {
return Ok(());
}
let response_handoff_prefix = match phase {
Some(MessagePhase::Commentary) => handoff.codex_response_handoff_prefix.clone(),
Some(MessagePhase::FinalAnswer) | None => None,
};
let active_handoff = handoff.active_handoff.lock().await.clone();
let is_commentary = matches!(phase, Some(MessagePhase::Commentary));
let active_handoff = handoff.stream.lock().await.active_handoff.clone();
let output = match active_handoff {
Some(handoff_id) => {
let output_text = realtime_backend_output(output_text, handoff.session_kind);
@@ -548,15 +681,10 @@ impl RealtimeConversationManager {
handoff.codex_response_item_prefix.as_deref(),
),
}
} else if handoff.session_kind == RealtimeSessionKind::V1
&& handoff.codex_response_handoff_prefix.is_some()
{
} else if handoff.session_kind == RealtimeSessionKind::V1 && is_commentary {
RealtimeOutbound::HandoffAppend {
handoff_id,
text: realtime_backend_item(
output_text,
response_handoff_prefix.as_deref(),
),
text: output_text,
}
} else {
RealtimeOutbound::HandoffUpdate {
@@ -577,8 +705,8 @@ impl RealtimeConversationManager {
}
} else {
RealtimeOutbound::StandaloneHandoff {
text: if handoff.session_kind == RealtimeSessionKind::V1 {
realtime_backend_item(output_text, response_handoff_prefix.as_deref())
text: if handoff.session_kind == RealtimeSessionKind::V1 && !is_commentary {
format!("{AGENT_FINAL_MESSAGE_PREFIX}{output_text}")
} else {
output_text
},
@@ -594,6 +722,119 @@ impl RealtimeConversationManager {
Ok(())
}
pub(crate) async fn register_handoff_stream_item(
&self,
item_id: String,
phase: Option<MessagePhase>,
initial_text: String,
) {
let handoff = {
let guard = self.state.lock().await;
guard.as_ref().map(|state| state.handoff.clone())
};
let Some(handoff) = handoff else {
return;
};
if !handoff.streams_handoff_append() {
return;
}
let flush_delay = {
let mut stream = handoff.stream.lock().await;
let Some(handoff_id) = stream.active_handoff.clone() else {
return;
};
let mut streamed_item = RealtimeStreamedItem {
handoff_id,
phase,
sent_bytes: 0,
buffered_text: String::new(),
tail_text: String::new(),
truncated: false,
last_flush_at: Instant::now(),
flush_scheduled: false,
};
streamed_item.push_text(&initial_text);
let flush_delay = if streamed_item.buffered_text.is_empty() {
None
} else {
streamed_item.flush_scheduled = true;
Some(streamed_item.next_flush_delay())
};
stream.items.insert(item_id.clone(), streamed_item);
flush_delay
};
if let Some(flush_delay) = flush_delay {
schedule_streamed_handoff_flush(&handoff, item_id, flush_delay);
}
}
pub(crate) async fn stream_handoff_delta(
&self,
item_id: &str,
delta: String,
) -> CodexResult<()> {
if delta.is_empty() {
return Ok(());
}
let handoff = {
let guard = self.state.lock().await;
let Some(state) = guard.as_ref() else {
return Err(CodexErr::InvalidRequest(
"conversation is not running".to_string(),
));
};
state.handoff.clone()
};
if !handoff.streams_handoff_append() {
return Ok(());
}
let flush_delay = {
let mut stream = handoff.stream.lock().await;
let Some(streamed_item) = stream.items.get_mut(item_id) else {
return Ok(());
};
streamed_item.push_text(&delta);
if streamed_item.flush_scheduled || streamed_item.streamable_text_bytes() == 0 {
None
} else {
streamed_item.flush_scheduled = true;
Some(streamed_item.next_flush_delay())
}
};
if let Some(flush_delay) = flush_delay {
schedule_streamed_handoff_flush(&handoff, item_id.to_string(), flush_delay);
}
Ok(())
}
pub(crate) async fn finish_handoff_stream_item(&self, item_id: &str) -> bool {
let handoff = {
let guard = self.state.lock().await;
guard.as_ref().map(|state| state.handoff.clone())
};
let Some(handoff) = handoff else {
return false;
};
if !handoff.streams_handoff_append() {
return false;
}
let Some(mut streamed_item) = handoff.stream.lock().await.items.remove(item_id) else {
return false;
};
let chunk = streamed_item.drain_final_chunk();
let sent_output = streamed_item.sent_bytes > 0;
if let Some(text) = chunk {
let _ = handoff
.output_tx
.send(RealtimeOutbound::HandoffAppend {
handoff_id: streamed_item.handoff_id,
text,
})
.await;
}
sent_output
}
pub(crate) async fn append_speech(&self, text: String) -> CodexResult<()> {
if text.trim().is_empty() {
return Ok(());
@@ -635,7 +876,7 @@ impl RealtimeConversationManager {
RealtimeSessionKind::V2 => {}
}
let Some(handoff_id) = handoff.active_handoff.lock().await.clone() else {
let Some(handoff_id) = handoff.stream.lock().await.active_handoff.clone() else {
return Ok(());
};
let Some(output_text) = handoff.last_output_text.lock().await.clone() else {
@@ -664,7 +905,11 @@ impl RealtimeConversationManager {
guard.as_ref().map(|state| state.handoff.clone())
};
if let Some(handoff) = handoff {
*handoff.active_handoff.lock().await = None;
{
let mut stream = handoff.stream.lock().await;
stream.active_handoff = None;
stream.items.clear();
}
*handoff.last_output_text.lock().await = None;
}
}
@@ -742,7 +987,6 @@ struct PreparedRealtimeConversationStart {
flush_transcript_tail_on_session_end: bool,
codex_responses_as_items: bool,
codex_response_item_prefix: Option<String>,
codex_response_handoff_prefix: Option<String>,
realtime_call_api_provider: Option<ApiProvider>,
requested_realtime_session_id: Option<String>,
version: RealtimeWsVersion,
@@ -828,7 +1072,6 @@ async fn prepare_realtime_start(
flush_transcript_tail_on_session_end: params.flush_transcript_tail_on_session_end,
codex_responses_as_items: params.codex_responses_as_items,
codex_response_item_prefix: params.codex_response_item_prefix,
codex_response_handoff_prefix: params.codex_response_handoff_prefix,
realtime_call_api_provider,
requested_realtime_session_id,
version,
@@ -1000,7 +1243,6 @@ async fn handle_start_inner(
flush_transcript_tail_on_session_end,
codex_responses_as_items,
codex_response_item_prefix,
codex_response_handoff_prefix,
realtime_call_api_provider,
requested_realtime_session_id,
version,
@@ -1019,7 +1261,6 @@ async fn handle_start_inner(
flush_transcript_tail_on_session_end,
codex_responses_as_items,
codex_response_item_prefix,
codex_response_handoff_prefix,
realtime_call_api_provider,
session_config,
model_client: sess.services.model_client.clone(),
@@ -1458,6 +1699,37 @@ async fn handle_text_input(
Ok(())
}
async fn flush_streamed_handoff_item(handoff: &RealtimeHandoffState, item_id: &str) {
let (handoff_id, text) = {
let mut stream = handoff.stream.lock().await;
let Some(streamed_item) = stream.items.get_mut(item_id) else {
return;
};
streamed_item.flush_scheduled = false;
let Some(text) = streamed_item.drain_stream_chunk() else {
return;
};
streamed_item.last_flush_at = Instant::now();
(streamed_item.handoff_id.clone(), text)
};
let _ = handoff
.output_tx
.send(RealtimeOutbound::HandoffAppend { handoff_id, text })
.await;
}
fn schedule_streamed_handoff_flush(
handoff: &RealtimeHandoffState,
item_id: String,
flush_delay: Duration,
) {
let handoff = handoff.clone();
let _flush_task = tokio::spawn(async move {
tokio::time::sleep(flush_delay).await;
flush_streamed_handoff_item(&handoff, &item_id).await;
});
}
async fn handle_handoff_output(
handoff_output: Result<RealtimeOutbound, RecvError>,
writer: &RealtimeWebsocketWriter,
@@ -1508,7 +1780,7 @@ async fn handle_handoff_output(
}
RealtimeOutbound::HandoffUpdate { handoff_id, text }
| RealtimeOutbound::HandoffAppend { handoff_id, text } => {
let active_handoff = handoff_state.active_handoff.lock().await.clone();
let active_handoff = handoff_state.stream.lock().await.active_handoff.clone();
match active_handoff {
Some(active_handoff) if active_handoff == handoff_id => {}
Some(_) | None => {
@@ -1661,10 +1933,12 @@ async fn handle_realtime_server_event(
match session_kind {
RealtimeSessionKind::V1 => {
*handoff_state.active_handoff.lock().await = Some(handoff.handoff_id.clone());
let mut stream = handoff_state.stream.lock().await;
stream.items.clear();
stream.active_handoff = Some(handoff.handoff_id.clone());
}
RealtimeSessionKind::V2 => {
let active_handoff = handoff_state.active_handoff.lock().await.clone();
let active_handoff = handoff_state.stream.lock().await.active_handoff.clone();
match active_handoff {
Some(_) => {
if let Err(err) = writer
@@ -1688,7 +1962,7 @@ async fn handle_realtime_server_event(
.await?;
}
None => {
*handoff_state.active_handoff.lock().await =
handoff_state.stream.lock().await.active_handoff =
Some(handoff.handoff_id.clone());
}
}
codex-rs/core/src/realtime_conversation_tests.rs
+38 -5
View File
@@ -1,5 +1,8 @@
use super::AGENT_FINAL_MESSAGE_PREFIX;
use super::HANDOFF_STREAM_TRUNCATION_MARKER;
use super::RealtimeHandoffState;
use super::RealtimeSessionKind;
use super::RealtimeStreamedItem;
use super::realtime_delegation_from_handoff;
use super::realtime_request_headers;
use super::realtime_text_from_handoff_request;
@@ -7,9 +10,11 @@ use super::wrap_realtime_delegation_input;
use crate::context::RealtimeDelegationSource;
use async_channel::bounded;
use codex_api::RealtimeEventParser;
use codex_protocol::models::MessagePhase;
use codex_protocol::protocol::RealtimeHandoffRequested;
use codex_protocol::protocol::RealtimeTranscriptEntry;
use pretty_assertions::assert_eq;
use std::time::Instant;
#[test]
fn prefers_handoff_input_transcript_over_active_transcript() {
@@ -146,18 +151,46 @@ async fn clears_active_handoff_explicitly() {
/*client_managed_handoffs*/ false,
/*codex_responses_as_items*/ false,
/*codex_response_item_prefix*/ None,
/*codex_response_handoff_prefix*/ None,
RealtimeSessionKind::V1,
/*event_parser*/ RealtimeEventParser::V1,
);
*state.active_handoff.lock().await = Some("handoff_1".to_string());
state.stream.lock().await.active_handoff = Some("handoff_1".to_string());
assert_eq!(
state.active_handoff.lock().await.clone(),
state.stream.lock().await.active_handoff.clone(),
Some("handoff_1".to_string())
);
*state.active_handoff.lock().await = None;
assert_eq!(state.active_handoff.lock().await.clone(), None);
state.stream.lock().await.active_handoff = None;
assert_eq!(state.stream.lock().await.active_handoff.clone(), None);
}
#[test]
fn streamed_handoff_preserves_a_bounded_final_tail() {
let mut item = RealtimeStreamedItem {
handoff_id: "handoff_1".to_string(),
phase: Some(MessagePhase::FinalAnswer),
sent_bytes: 0,
buffered_text: String::new(),
tail_text: String::new(),
truncated: false,
last_flush_at: Instant::now(),
flush_scheduled: false,
};
item.push_text(&format!("HEAD{}TAIL", "x".repeat(/*n*/ 5_000)));
let first = item
.drain_stream_chunk()
.expect("oversized output should retain a streamable head");
let final_chunk = item
.drain_final_chunk()
.expect("oversized output should retain a final tail");
let output = format!("{first}{final_chunk}");
assert!(output.len() <= 4_000);
assert!(output.starts_with(&format!("{AGENT_FINAL_MESSAGE_PREFIX}HEAD")));
assert!(output.contains(HANDOFF_STREAM_TRUNCATION_MARKER));
assert!(output.ends_with("TAIL"));
}
#[test]
codex-rs/core/src/session/mod.rs
+36 -3
View File
@@ -236,6 +236,7 @@ use self::session::SessionConfiguration;
pub(crate) use self::session::SessionSettingsUpdate;
#[cfg(test)]
use self::turn::AssistantMessageStreamParsers;
use self::turn::agent_message_text;
#[cfg(test)]
use self::turn::collect_explicit_app_ids_from_skill_items;
use self::turn::realtime_text_for_event;
@@ -1885,12 +1886,44 @@ impl Session {
}
async fn maybe_mirror_event_text_to_realtime(&self, msg: &EventMsg) {
let Some((text, phase)) = realtime_text_for_event(msg) else {
return;
};
if self.conversation.running_state().await.is_none() {
return;
}
match msg {
EventMsg::ItemStarted(event) => {
if let TurnItem::AgentMessage(item) = &event.item {
self.conversation
.register_handoff_stream_item(
item.id.clone(),
item.phase.clone(),
agent_message_text(item),
)
.await;
}
return;
}
EventMsg::AgentMessageContentDelta(event) => {
if let Err(err) = self
.conversation
.stream_handoff_delta(&event.item_id, event.delta.clone())
.await
{
debug!("failed to stream event text to realtime conversation: {err}");
}
return;
}
EventMsg::ItemCompleted(event) => {
if let TurnItem::AgentMessage(item) = &event.item
&& self.conversation.finish_handoff_stream_item(&item.id).await
{
return;
}
}
_ => {}
}
let Some((text, phase)) = realtime_text_for_event(msg) else {
return;
};
if let Err(err) = self.conversation.handoff_out(text, phase).await {
debug!("failed to mirror event text to realtime conversation: {err}");
}
codex-rs/core/src/session/turn.rs
+1 -1
View File
@@ -1523,7 +1523,7 @@ async fn maybe_emit_pending_agent_message_start(
}
/// Agent messages are text-only today; concatenate all text entries.
fn agent_message_text(item: &codex_protocol::items::AgentMessageItem) -> String {
pub(super) fn agent_message_text(item: &codex_protocol::items::AgentMessageItem) -> String {
item.content
.iter()
.map(|entry| match entry {
codex-rs/core/tests/suite/compact_remote.rs
-1
View File
@@ -212,7 +212,6 @@ async fn start_realtime_conversation(codex: &codex_core::CodexThread) -> Result<
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
codex-rs/core/tests/suite/realtime_conversation.rs
+193 -44
View File
@@ -66,7 +66,6 @@ const MEMORY_PROMPT_PHRASE: &str =
"You have access to a memory folder with guidance from prior runs.";
const REALTIME_CONVERSATION_TEST_SUBPROCESS_ENV_VAR: &str =
"CODEX_REALTIME_CONVERSATION_TEST_SUBPROCESS";
const SILENT_CONTEXT_PREFIX: &str = "[BACKEND] Silent Codex context. Do not speak, acknowledge, or summarize this item. Wait for an explicit speakable handoff or direct user request.";
#[derive(Debug, Clone)]
struct RealtimeCallRequestCapture {
@@ -288,7 +287,6 @@ async fn conversation_start_audio_text_close_round_trip() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -435,7 +433,6 @@ async fn conversation_start_defaults_to_v2_and_gpt_realtime_1_5() -> Result<()>
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -531,7 +528,6 @@ async fn conversation_webrtc_start_posts_generated_session() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: Some("session-override-model".to_string()),
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -719,7 +715,6 @@ async fn conversation_webrtc_start_uses_avas_query() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -818,7 +813,6 @@ async fn conversation_webrtc_default_v1_ignores_configured_v2_voice() -> Result<
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -879,7 +873,6 @@ async fn conversation_webrtc_default_v1_rejects_explicit_v2_voice() -> Result<()
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -950,7 +943,6 @@ async fn conversation_webrtc_start_uses_configured_call_base_url_for_avas() -> R
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1045,7 +1037,6 @@ async fn conversation_webrtc_close_while_sideband_connecting_drops_pending_join(
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1137,7 +1128,6 @@ async fn conversation_webrtc_sideband_connect_failure_closes_with_error() -> Res
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1231,7 +1221,6 @@ async fn conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth() ->
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1323,7 +1312,6 @@ async fn assert_transport_close_tail_flush(
flush_transcript_tail_on_session_end,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1440,7 +1428,6 @@ async fn conversation_start_preflight_failure_emits_realtime_error_only() -> Res
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1492,7 +1479,6 @@ async fn conversation_start_connect_failure_emits_realtime_error_only() -> Resul
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1592,7 +1578,6 @@ async fn conversation_second_start_replaces_runtime() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1623,7 +1608,6 @@ async fn conversation_second_start_replaces_runtime() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1725,7 +1709,6 @@ async fn conversation_uses_experimental_realtime_ws_base_url_override() -> Resul
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1795,7 +1778,6 @@ async fn conversation_uses_default_realtime_backend_prompt() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1873,7 +1855,6 @@ async fn conversation_uses_empty_instructions_for_null_or_empty_prompt() -> Resu
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1944,7 +1925,6 @@ async fn conversation_uses_explicit_start_voice() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2007,7 +1987,6 @@ async fn conversation_uses_configured_realtime_voice() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2058,7 +2037,6 @@ async fn conversation_rejects_voice_for_wrong_realtime_version() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2110,7 +2088,6 @@ async fn conversation_uses_experimental_realtime_ws_backend_prompt_override() ->
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2188,7 +2165,6 @@ async fn conversation_uses_experimental_realtime_ws_startup_context_override() -
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2260,7 +2236,6 @@ async fn conversation_disables_realtime_startup_context_with_empty_override() ->
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2325,7 +2300,6 @@ async fn conversation_start_injects_startup_context_from_thread_history() -> Res
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2445,7 +2419,6 @@ async fn conversation_startup_context_current_thread_selects_many_turns_by_budge
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2558,7 +2531,6 @@ async fn conversation_startup_context_falls_back_to_workspace_map() -> Result<()
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2623,7 +2595,6 @@ async fn conversation_startup_context_is_truncated_and_sent_once_per_start() ->
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2709,7 +2680,6 @@ async fn conversation_user_text_turn_is_not_sent_to_realtime() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2811,7 +2781,6 @@ async fn realtime_v2_noop_tool_call_returns_empty_function_output_without_respon
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2915,7 +2884,6 @@ async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Re
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2986,6 +2954,197 @@ async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Re
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn conversation_flushes_assistant_deltas_every_200ms_for_v3_handoff() -> Result<()> {
skip_if_no_network!(Ok(()));
let initial_commentary_text = "seed ";
let first_commentary_delta = "first ";
let second_commentary_delta = "x".repeat(94);
let commentary_text =
format!("{initial_commentary_text}{first_commentary_delta}{second_commentary_delta}");
let (gate_commentary_done_tx, gate_commentary_done_rx) = oneshot::channel();
let mut commentary_item_added =
responses::ev_message_item_added("msg_commentary", initial_commentary_text);
commentary_item_added["item"]["phase"] = json!("commentary");
let mut commentary_item_done =
responses::ev_assistant_message("msg_commentary", &commentary_text);
commentary_item_done["item"]["phase"] = json!("commentary");
let mut final_item_added = responses::ev_message_item_added("msg_final", "");
final_item_added["item"]["phase"] = json!("final_answer");
let mut final_item_done = responses::ev_assistant_message("msg_final", "done");
final_item_done["item"]["phase"] = json!("final_answer");
let response_chunks = vec![
StreamingSseChunk {
gate: None,
body: sse_event(responses::ev_response_created("resp_stream")),
},
StreamingSseChunk {
gate: None,
body: sse_event(commentary_item_added),
},
StreamingSseChunk {
gate: None,
body: sse_event(responses::ev_output_text_delta(first_commentary_delta)),
},
StreamingSseChunk {
gate: None,
body: sse_event(responses::ev_output_text_delta(&second_commentary_delta)),
},
StreamingSseChunk {
gate: Some(gate_commentary_done_rx),
body: sse_event(commentary_item_done),
},
StreamingSseChunk {
gate: None,
body: sse_event(final_item_added),
},
StreamingSseChunk {
gate: None,
body: sse_event(responses::ev_output_text_delta("done")),
},
StreamingSseChunk {
gate: None,
body: sse_event(final_item_done),
},
StreamingSseChunk {
gate: None,
body: sse_event(responses::ev_completed("resp_stream")),
},
];
let (api_server, completions) = start_streaming_sse_server(vec![response_chunks]).await;
let realtime_server = start_websocket_server(vec![vec![
vec![
json!({
"type": "session.started",
"session": { "id": "sess_stream", "instructions": "backend prompt" }
}),
json!({
"type": "delegation.created",
"item": {
"id": "delegation_stream",
"type": "delegation",
"target": "client",
"content": [{ "type": "input_text", "text": "delegate streaming" }]
}
}),
],
vec![],
vec![],
]])
.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.realtime.version = RealtimeWsVersion::V3;
}
});
let test = builder.build_with_streaming_server(&api_server).await?;
test.codex
.submit(Op::RealtimeConversationStart(ConversationStartParams {
client_managed_handoffs: false,
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
prompt: Some(Some("backend prompt".to_string())),
realtime_session_id: None,
transport: None,
version: None,
voice: None,
}))
.await?;
let _ = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload:
RealtimeEvent::SessionUpdated {
realtime_session_id,
..
},
}) if realtime_session_id == "sess_stream" => Some(()),
_ => None,
})
.await;
let _ = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::HandoffRequested(handoff),
}) if handoff.handoff_id == "delegation_stream" => Some(()),
_ => None,
})
.await;
let _ = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::AgentMessageContentDelta(event)
if event.item_id == "msg_commentary" && event.delta == first_commentary_delta =>
{
Some(())
}
_ => None,
})
.await;
assert_eq!(
wait_for_websocket_request(
&realtime_server,
/*connection_index*/ 0,
/*request_index*/ 1,
)
.await?
.body_json(),
json!({
"type": "delegation.context.append",
"delegation_item_id": "delegation_stream",
"content": [{ "type": "input_text", "text": commentary_text }]
})
);
let _ = gate_commentary_done_tx.send(());
assert_eq!(
wait_for_websocket_request(
&realtime_server,
/*connection_index*/ 0,
/*request_index*/ 2,
)
.await?
.body_json(),
json!({
"type": "delegation.context.append",
"delegation_item_id": "delegation_stream",
"content": [{
"type": "input_text",
"text": "\"Agent Final Message\":\n\ndone"
}]
})
);
completions
.into_iter()
.next()
.expect("missing delegated turn completion")
.await
.expect("delegated turn request did not complete");
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
tokio::time::sleep(Duration::from_millis(50)).await;
assert_eq!(
realtime_server.single_connection().len(),
3,
"completed assistant item must not resend the full text"
);
realtime_server.shutdown().await;
api_server.shutdown().await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn conversation_handoff_persists_across_item_done_until_turn_complete() -> Result<()> {
skip_if_no_network!(Ok(()));
@@ -3055,7 +3214,6 @@ async fn conversation_handoff_persists_across_item_done_until_turn_complete() ->
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: Some(SILENT_CONTEXT_PREFIX.to_string()),
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -3100,7 +3258,7 @@ async fn conversation_handoff_persists_across_item_done_until_turn_complete() ->
);
assert_eq!(
first_append.body_json()["output_text"].as_str(),
Some(format!("{SILENT_CONTEXT_PREFIX}\n\nassistant message 1").as_str())
Some("assistant message 1")
);
let _ = wait_for_event_match(&test.codex, |msg| match msg {
@@ -3126,7 +3284,7 @@ async fn conversation_handoff_persists_across_item_done_until_turn_complete() ->
);
assert_eq!(
second_append.body_json()["output_text"].as_str(),
Some("assistant message 2")
Some("\"Agent Final Message\":\n\nassistant message 2")
);
let completion = completions
@@ -3212,7 +3370,6 @@ async fn inbound_handoff_request_starts_turn() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -3327,7 +3484,6 @@ async fn inbound_handoff_request_uses_active_transcript() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -3435,7 +3591,6 @@ async fn inbound_handoff_request_sends_transcript_delta_after_each_handoff() ->
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -3562,7 +3717,6 @@ async fn conversation_close_routes_only_remaining_transcript_tail_once() -> Resu
flush_transcript_tail_on_session_end: true,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -3650,7 +3804,6 @@ async fn inbound_conversation_item_does_not_start_turn_and_still_forwards_audio(
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -3778,7 +3931,6 @@ async fn delegated_turn_user_role_echo_does_not_redelegate_and_still_forwards_au
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -3936,7 +4088,6 @@ async fn inbound_handoff_request_does_not_block_realtime_event_forwarding() -> R
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -4083,7 +4234,6 @@ async fn inbound_handoff_request_steers_active_turn() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -4241,7 +4391,6 @@ async fn inbound_handoff_request_starts_turn_and_does_not_block_realtime_audio()
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_prefix: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
codex-rs/protocol/src/protocol.rs
-4
View File
@@ -213,10 +213,6 @@ pub struct ConversationStartParams {
pub codex_responses_as_items: bool,
/// Optional prefix added to automatic Codex response items when `codex_responses_as_items` is set.
pub codex_response_item_prefix: Option<String>,
/// Optional prefix added to automatic V1 or V3 Codex commentary sent through the selected
/// Bidi handoff wire event when `codex_responses_as_items` is not set. Final answers are sent
/// without the prefix.
pub codex_response_handoff_prefix: Option<String>,
/// Overrides the configured realtime model for this session only.
pub model: Option<String>,
/// Selects whether the realtime session should produce text or audio output.