diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index cc7e8b00cf..176ab77eff 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -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, diff --git a/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs b/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs index dee420dc65..f375a5ea2e 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs @@ -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, - /// 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, /// Overrides the configured realtime model for this session only. #[ts(optional = nullable)] pub model: Option, diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index fa1d2f1670..021e8043fa 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -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 diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index d48ff7649f..31d00cd7bd 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -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), diff --git a/codex-rs/app-server/tests/suite/v2/experimental_api.rs b/codex-rs/app-server/tests/suite/v2/experimental_api.rs index cc010b93eb..c9c0bd8dfb 100644 --- a/codex-rs/app-server/tests/suite/v2/experimental_api.rs +++ b/codex-rs/app-server/tests/suite/v2/experimental_api.rs @@ -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, diff --git a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs index 4dbcb6f394..3ac137541e 100644 --- a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs +++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs @@ -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, codex_responses_as_items: Option, - codex_response_handoff_prefix: Option<&str>, version: RealtimeConversationVersion, ) -> Result { // 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::("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, diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index 5d3550c3b5..a9a194c83c 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -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, - active_handoff: Arc>>, last_output_text: Arc>>, + stream: Arc>, client_managed_handoffs: bool, codex_responses_as_items: bool, codex_response_item_prefix: Option, - codex_response_handoff_prefix: Option, session_kind: RealtimeSessionKind, + event_parser: RealtimeEventParser, +} + +#[derive(Debug, Default)] +struct RealtimeHandoffStreamState { + active_handoff: Option, + items: HashMap, +} + +#[derive(Debug)] +struct RealtimeStreamedItem { + handoff_id: String, + phase: Option, + 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 { + 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::(); + let text = format!("{prefix}{text}"); + self.sent_bytes += text.len(); + Some(text) + } + + fn drain_final_chunk(&mut self) -> Option { + let prefix = self.output_prefix(); + if !self.truncated { + if self.buffered_text.is_empty() { + return None; + } + let text = self.buffered_text.drain(..).collect::(); + let text = format!("{prefix}{text}"); + self.sent_bytes += text.len(); + return Some(text); + } + + let head = self.buffered_text.drain(..).collect::(); + let tail = self.tail_text.drain(..).collect::(); + 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, - codex_response_handoff_prefix: Option, 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, - codex_response_handoff_prefix: Option, realtime_call_api_provider: Option, 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, + 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, - codex_response_handoff_prefix: Option, realtime_call_api_provider: Option, requested_realtime_session_id: Option, 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, 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()); } } diff --git a/codex-rs/core/src/realtime_conversation_tests.rs b/codex-rs/core/src/realtime_conversation_tests.rs index 15bd4d5456..f9bf00a7c5 100644 --- a/codex-rs/core/src/realtime_conversation_tests.rs +++ b/codex-rs/core/src/realtime_conversation_tests.rs @@ -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] diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index b85a3af54f..da0992cb3d 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -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}"); } diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index ce59f1fa99..7da4c18693 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -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 { diff --git a/codex-rs/core/tests/suite/compact_remote.rs b/codex-rs/core/tests/suite/compact_remote.rs index 10de2b96c8..43caddf0ee 100644 --- a/codex-rs/core/tests/suite/compact_remote.rs +++ b/codex-rs/core/tests/suite/compact_remote.rs @@ -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, diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index 91416a69d0..3d1a858620 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -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, diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index d64b4bb96e..62c0e17ebd 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -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, - /// 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, /// Overrides the configured realtime model for this session only. pub model: Option, /// Selects whether the realtime session should produce text or audio output.