diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs index 71c931d9cc..f176f5b228 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -402,6 +402,10 @@ impl RealtimeWebsocketEvents { } } + pub async fn take_active_transcript(&self) -> Vec { + std::mem::take(&mut self.active_transcript.lock().await.entries) + } + async fn update_active_transcript(&self, event: &mut RealtimeEvent) { let mut active_transcript = self.active_transcript.lock().await; match event { diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index ea719f5f8b..763e1011eb 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -49,6 +49,7 @@ use serde_json::json; use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; +use std::time::Duration; use tokio::sync::Mutex; use tokio::task::JoinHandle; use tracing::debug; @@ -62,6 +63,7 @@ const HANDOFF_OUT_QUEUE_CAPACITY: usize = 64; const OUTPUT_EVENTS_QUEUE_CAPACITY: usize = 256; const REALTIME_STARTUP_CONTEXT_TOKEN_BUDGET: usize = 5_000; const DEFAULT_REALTIME_MODEL: &str = "gpt-realtime-1.5"; +const V1_HANDOFF_DELEGATION_GRACE: Duration = Duration::from_secs(1); const ACTIVE_RESPONSE_CONFLICT_ERROR_PREFIX: &str = "Conversation already has an active response in progress:"; @@ -81,18 +83,12 @@ pub(crate) struct RealtimeConversationManager { state: Mutex>, } -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -enum RealtimeSessionKind { - V1, - V2, -} - #[derive(Clone, Debug)] struct RealtimeHandoffState { output_tx: Sender, active_handoff: Arc>>, last_output_text: Arc>>, - session_kind: RealtimeSessionKind, + event_parser: RealtimeEventParser, } #[derive(Debug, PartialEq, Eq)] @@ -121,16 +117,16 @@ struct RealtimeInputTask { audio_rx: Receiver, events_tx: Sender, handoff_state: RealtimeHandoffState, - session_kind: RealtimeSessionKind, + event_parser: RealtimeEventParser, } impl RealtimeHandoffState { - fn new(output_tx: Sender, session_kind: RealtimeSessionKind) -> Self { + fn new(output_tx: Sender, event_parser: RealtimeEventParser) -> Self { Self { output_tx, active_handoff: Arc::new(Mutex::new(None)), last_output_text: Arc::new(Mutex::new(None)), - session_kind, + event_parser, } } } @@ -195,10 +191,7 @@ impl RealtimeConversationManager { model_client, sdp, } = start; - let session_kind = match session_config.event_parser { - RealtimeEventParser::V1 => RealtimeSessionKind::V1, - RealtimeEventParser::RealtimeV2 => RealtimeSessionKind::V2, - }; + let event_parser = session_config.event_parser; let client = RealtimeWebsocketClient::new(api_provider); let (connection, sdp) = if let Some(sdp) = sdp { @@ -243,7 +236,7 @@ impl RealtimeConversationManager { async_channel::bounded::(OUTPUT_EVENTS_QUEUE_CAPACITY); let realtime_active = Arc::new(AtomicBool::new(true)); - let handoff = RealtimeHandoffState::new(handoff_output_tx, session_kind); + let handoff = RealtimeHandoffState::new(handoff_output_tx, event_parser); let task = spawn_realtime_input_task(RealtimeInputTask { writer: writer.clone(), events, @@ -252,7 +245,7 @@ impl RealtimeConversationManager { audio_rx, events_tx, handoff_state: handoff.clone(), - session_kind, + event_parser, }); let mut guard = self.state.lock().await; @@ -366,7 +359,7 @@ impl RealtimeConversationManager { }; *handoff.last_output_text.lock().await = Some(output_text.clone()); - if matches!(handoff.session_kind, RealtimeSessionKind::V1) { + if matches!(handoff.event_parser, RealtimeEventParser::V1) { handoff .output_tx .send(HandoffOutput::ImmediateAppend { @@ -387,7 +380,7 @@ impl RealtimeConversationManager { let Some(handoff) = handoff else { return Ok(()); }; - if matches!(handoff.session_kind, RealtimeSessionKind::V1) { + if matches!(handoff.event_parser, RealtimeEventParser::V1) { return Ok(()); } @@ -708,16 +701,8 @@ async fn handle_start_inner( if matches!(event, RealtimeEvent::Error(_)) { end = RealtimeConversationEnd::Error; } - let maybe_routed_text = match &event { - RealtimeEvent::HandoffRequested(handoff) => { - realtime_text_from_handoff_request(handoff) - } - _ => None, - }; - if let Some(text) = maybe_routed_text { - debug!(text = %text, "[realtime-text] realtime conversation text output"); - let sess_for_routed_text = Arc::clone(&sess_clone); - sess_for_routed_text.route_realtime_text_input(text).await; + if let RealtimeEvent::HandoffRequested(handoff) = &event { + route_realtime_handoff(&sess_clone, handoff).await; } if !fanout_realtime_active.load(Ordering::Relaxed) { break; @@ -764,6 +749,13 @@ pub(crate) async fn handle_audio( } } +async fn route_realtime_handoff(sess: &Arc, handoff: &RealtimeHandoffRequested) { + if let Some(text) = realtime_text_from_handoff_request(handoff) { + debug!(text = %text, "[realtime-text] realtime conversation text output"); + Arc::clone(sess).route_realtime_text_input(text).await; + } +} + fn realtime_text_from_handoff_request(handoff: &RealtimeHandoffRequested) -> Option { let active_transcript = handoff .active_transcript @@ -854,7 +846,7 @@ fn spawn_realtime_input_task(input: RealtimeInputTask) -> JoinHandle<()> { audio_rx, events_tx, handoff_state, - session_kind, + event_parser, } = input; tokio::spawn(async move { @@ -875,7 +867,7 @@ fn spawn_realtime_input_task(input: RealtimeInputTask) -> JoinHandle<()> { .await; break; } - if matches!(session_kind, RealtimeSessionKind::V2) { + if matches!(event_parser, RealtimeEventParser::RealtimeV2) { if response_in_progress { pending_response_create = true; } else if let Err(err) = writer.send_response_create().await { @@ -929,7 +921,7 @@ fn spawn_realtime_input_task(input: RealtimeInputTask) -> JoinHandle<()> { .await; break; } - if matches!(session_kind, RealtimeSessionKind::V2) { + if matches!(event_parser, RealtimeEventParser::RealtimeV2) { if response_in_progress { pending_response_create = true; } else if let Err(err) = writer.send_response_create().await { @@ -962,12 +954,12 @@ fn spawn_realtime_input_task(input: RealtimeInputTask) -> JoinHandle<()> { RealtimeEvent::ConversationItemAdded(item) => { match item.get("type").and_then(Value::as_str) { Some("response.created") - if matches!(session_kind, RealtimeSessionKind::V2) => + if matches!(event_parser, RealtimeEventParser::RealtimeV2) => { response_in_progress = true; } Some("response.done") - if matches!(session_kind, RealtimeSessionKind::V2) => + if matches!(event_parser, RealtimeEventParser::RealtimeV2) => { response_in_progress = false; output_audio_state = None; @@ -992,12 +984,12 @@ fn spawn_realtime_input_task(input: RealtimeInputTask) -> JoinHandle<()> { } } RealtimeEvent::AudioOut(frame) => { - if matches!(session_kind, RealtimeSessionKind::V2) { + if matches!(event_parser, RealtimeEventParser::RealtimeV2) { update_output_audio_state(&mut output_audio_state, frame); } } RealtimeEvent::InputAudioSpeechStarted(event) => { - if matches!(session_kind, RealtimeSessionKind::V2) + if matches!(event_parser, RealtimeEventParser::RealtimeV2) && let Some(output_audio_state) = output_audio_state.take() && event @@ -1021,7 +1013,7 @@ fn spawn_realtime_input_task(input: RealtimeInputTask) -> JoinHandle<()> { RealtimeEvent::ResponseCancelled(_) => { response_in_progress = false; output_audio_state = None; - if matches!(session_kind, RealtimeSessionKind::V2) + if matches!(event_parser, RealtimeEventParser::RealtimeV2) && pending_response_create { if let Err(err) = writer.send_response_create().await { @@ -1044,9 +1036,24 @@ fn spawn_realtime_input_task(input: RealtimeInputTask) -> JoinHandle<()> { *handoff_state.last_output_text.lock().await = None; response_in_progress = false; output_audio_state = None; + if event_parser == RealtimeEventParser::V1 { + let mut handoff = handoff.clone(); + let events = events.clone(); + let events_tx = events_tx.clone(); + forward_event = false; + tokio::spawn(async move { + tokio::time::sleep(V1_HANDOFF_DELEGATION_GRACE).await; + handoff + .active_transcript + .extend(events.take_active_transcript().await); + let _ = events_tx + .send(RealtimeEvent::HandoffRequested(handoff)) + .await; + }); + } } RealtimeEvent::Error(message) - if matches!(session_kind, RealtimeSessionKind::V2) + if matches!(event_parser, RealtimeEventParser::RealtimeV2) && message.starts_with(ACTIVE_RESPONSE_CONFLICT_ERROR_PREFIX) => { warn!( diff --git a/codex-rs/core/src/realtime_conversation_tests.rs b/codex-rs/core/src/realtime_conversation_tests.rs index 0a32d063c0..41550cccb5 100644 --- a/codex-rs/core/src/realtime_conversation_tests.rs +++ b/codex-rs/core/src/realtime_conversation_tests.rs @@ -1,7 +1,7 @@ use super::RealtimeHandoffState; -use super::RealtimeSessionKind; use super::realtime_text_from_handoff_request; use async_channel::bounded; +use codex_api::RealtimeEventParser; use codex_protocol::protocol::RealtimeHandoffRequested; use codex_protocol::protocol::RealtimeTranscriptEntry; use pretty_assertions::assert_eq; @@ -57,7 +57,7 @@ fn ignores_empty_handoff_request_input_transcript() { #[tokio::test] async fn clears_active_handoff_explicitly() { let (tx, _rx) = bounded(1); - let state = RealtimeHandoffState::new(tx, RealtimeSessionKind::V1); + let state = RealtimeHandoffState::new(tx, RealtimeEventParser::V1); *state.active_handoff.lock().await = Some("handoff_1".to_string()); assert_eq!( diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index 0f68af428e..55d58d548c 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -1992,7 +1992,7 @@ async fn inbound_handoff_request_starts_turn() -> Result<()> { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn inbound_handoff_request_uses_active_transcript() -> Result<()> { +async fn inbound_handoff_request_collects_late_transcript_before_routing() -> Result<()> { skip_if_no_network!(Ok(())); let api_server = start_mock_server().await; @@ -2017,11 +2017,7 @@ async fn inbound_handoff_request_uses_active_transcript() -> Result<()> { }), json!({ "type": "conversation.input_transcript.delta", - "delta": "delegated query" - }), - json!({ - "type": "conversation.output_transcript.delta", - "delta": "assist confirm" + "delta": "delegated " }), json!({ "type": "conversation.handoff.requested", @@ -2029,6 +2025,14 @@ async fn inbound_handoff_request_uses_active_transcript() -> Result<()> { "item_id": "item_inbound_multi", "input_transcript": "ignored" }), + json!({ + "type": "conversation.input_transcript.delta", + "delta": "query" + }), + json!({ + "type": "conversation.output_transcript.delta", + "delta": "assist confirm" + }), ]]]) .await;