diff --git a/codex-rs/app-server/tests/suite/v2/realtime_transcript_tests.rs b/codex-rs/app-server/tests/suite/v2/realtime_transcript_tests.rs index c8e9e6ea8a..e15a6ad3ad 100644 --- a/codex-rs/app-server/tests/suite/v2/realtime_transcript_tests.rs +++ b/codex-rs/app-server/tests/suite/v2/realtime_transcript_tests.rs @@ -128,10 +128,9 @@ async fn websocket_v3_retains_both_speakers_from_same_frame_at_handoff() -> Resu RealtimeTestVersion::V1, main_loop_responses(vec![create_final_assistant_message_sse_response("Done.")?]), realtime_sideband(vec![open_realtime_sideband_connection(vec![vec![ - session_started("bounded-transcript"), + session_started("interleaved-transcript"), json!({"type": "input_transcript.added", "start_ms": 0, "end_ms": 200, "item": {"text": old_user}}), json!({"type": "output_transcript.added", "start_ms": 200, "end_ms": 400, "item": {"text": assistant}}), - // Input alone crosses the cap before output receives the same frame age. json!({"type": "input_transcript.added", "start_ms": 400, "end_ms": 600, "item": {"text": newest_user}}), json!({"type": "output_transcript.added", "start_ms": 400, "end_ms": 600, "item": {"text": assistant_suffix}}), json!({"type": "turn.done", "turn": {"role": "assistant", "transcript": assistant_final}}), @@ -196,9 +195,8 @@ async fn websocket_v3_retains_both_speakers_from_same_frame_at_handoff() -> Resu "item_id": "handoff", "input_transcript": user, "active_transcript": [ - {"role": "user", "text": format!("…{}{newest_user}", "u".repeat(3874))}, - {"role": "assistant", "text": format!("…{}{assistant_suffix}", "a".repeat(3870))}, {"role": "user", "text": user}, + {"role": "assistant", "text": assistant_final}, ], }), }, @@ -208,11 +206,9 @@ async fn websocket_v3_retains_both_speakers_from_same_frame_at_handoff() -> Resu .await?; let requests = harness.main_loop_responses_requests().await?; assert_eq!(requests.len(), 1); - // The 4 KiB input prefix omits the latest instruction. It must survive the transcript tail, - // even when the preceding assistant entry alone fills that field's 4 KiB cap. assert!(response_request_contains_text( &requests[0], - &format!("{newest_user}\n"), + &format!("user: {user}\nassistant: {assistant_final}"), )); harness.shutdown().await; Ok(()) 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 2ce624c85a..e7a8bf1bd4 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -38,7 +38,9 @@ use http::HeaderMap; use http::HeaderValue; use http::StatusCode; use std::collections::HashMap; +use std::collections::HashSet; use std::collections::VecDeque; +use std::hash::BuildHasher; use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; @@ -62,8 +64,6 @@ use url::Url; const REALTIME_WIRE_LOG_TARGET: &str = "codex_api::realtime_websocket::wire"; const OPENAI_REALTIME_API_BASE_URL: &str = "https://api.openai.com/v1"; -const MAX_ACTIVE_TRANSCRIPT_BYTES: usize = 8 * 1024; -const TRUNCATED_TRANSCRIPT_PREFIX: &str = "…"; struct WsStream { tx_command: mpsc::Sender, @@ -271,48 +271,20 @@ pub struct RealtimeTranscriptState { #[derive(Default)] struct ActiveTranscriptState { - entries: Vec, - last_frame: Option, - frame_sequence: u64, - input_last_raw_frame: Option, - output_last_raw_frame: Option, + entries: Vec, + // Handoff input can echo an utterance already written to child history. + flushed_user_inputs: HashSet, new_input_entry: bool, new_output_entry: bool, } impl ActiveTranscriptState { fn take_entries(&mut self) -> Vec { - self.input_last_raw_frame = None; - self.output_last_raw_frame = None; - self.entries - .drain(..) - .map(|entry| entry.transcript) - .collect() + self.flushed_user_inputs.clear(); + std::mem::take(&mut self.entries) } } -struct ActiveTranscriptEntry { - transcript: RealtimeTranscriptEntry, - last_raw_frame: Option, - truncated_prefix_bytes: usize, -} - -impl ActiveTranscriptEntry { - fn retained_text(&self) -> &str { - if self.truncated_prefix_bytes > 0 { - &self.transcript.text[TRUNCATED_TRANSCRIPT_PREFIX.len()..] - } else { - &self.transcript.text - } - } -} - -#[derive(Clone, Copy, PartialEq, Eq, serde::Deserialize)] -struct TranscriptFrame { - start_ms: u64, - end_ms: u64, -} - impl RealtimeWebsocketConnection { pub async fn send_audio_frame(&self, frame: RealtimeAudioFrame) -> Result<(), ApiError> { self.writer.send_audio_frame(frame).await @@ -622,17 +594,7 @@ impl RealtimeWebsocketEvents { Message::Text(text) => { trace!(target: REALTIME_WIRE_LOG_TARGET, "realtime websocket event: {text}"); if let Some(mut event) = parse_realtime_event(&text, self.event_parser) { - let frame = if self.event_parser == RealtimeEventParser::FramelessBidi - && matches!( - event, - RealtimeEvent::InputTranscriptDelta(_) - | RealtimeEvent::OutputTranscriptDelta(_) - ) { - serde_json::from_str::(&text).ok() - } else { - None - }; - self.update_active_transcript(&mut event, frame).await; + self.update_active_transcript(&mut event).await; debug!(?event, "realtime websocket parsed event"); return Ok(Some(event)); } @@ -688,26 +650,11 @@ impl RealtimeWebsocketEvents { } } - async fn update_active_transcript( - &self, - event: &mut RealtimeEvent, - frame: Option, - ) { + async fn update_active_transcript(&self, event: &mut RealtimeEvent) { let mut active_transcript = self.transcript_state.active.lock().await; - let frame_sequence = frame.map(|frame| { - // Both speakers in one frame have equal recency, regardless of wire event order. - // Count observed frames so reconnects do not depend on an absolute timestamp origin. - if active_transcript.last_frame != Some(frame) { - truncate_active_transcript(&mut active_transcript.entries); - active_transcript.frame_sequence += 1; - active_transcript.last_frame = Some(frame); - } - active_transcript.frame_sequence - }); match event { RealtimeEvent::InputAudioSpeechStarted(_) => { active_transcript.new_input_entry = true; - active_transcript.input_last_raw_frame = None; } RealtimeEvent::InputTranscriptDelta(RealtimeTranscriptDelta { delta, .. }) => { let force_new = active_transcript.new_input_entry; @@ -717,11 +664,9 @@ impl RealtimeWebsocketEvents { delta, force_new, self.event_parser, - frame_sequence, ); if !delta.is_empty() || self.event_parser != RealtimeEventParser::FramelessBidi { active_transcript.new_input_entry = false; - active_transcript.input_last_raw_frame = frame_sequence; } } RealtimeEvent::OutputTranscriptDelta(RealtimeTranscriptDelta { delta, .. }) => { @@ -732,53 +677,43 @@ impl RealtimeWebsocketEvents { delta, force_new, self.event_parser, - frame_sequence, ); if !delta.is_empty() || self.event_parser != RealtimeEventParser::FramelessBidi { active_transcript.new_output_entry = false; - active_transcript.output_last_raw_frame = frame_sequence; } } RealtimeEvent::InputTranscriptDone(done) => { let force_new = active_transcript.new_input_entry; - let last_raw_frame = active_transcript.input_last_raw_frame; apply_transcript_done( &mut active_transcript.entries, "user", &done.text, force_new, self.event_parser, - last_raw_frame, ); active_transcript.new_input_entry = self.event_parser == RealtimeEventParser::FramelessBidi; - active_transcript.input_last_raw_frame = None; } RealtimeEvent::OutputTranscriptDone(done) => { let force_new = active_transcript.new_output_entry; - let last_raw_frame = active_transcript.output_last_raw_frame; apply_transcript_done( &mut active_transcript.entries, "assistant", &done.text, force_new, self.event_parser, - last_raw_frame, ); active_transcript.new_output_entry = self.event_parser == RealtimeEventParser::FramelessBidi; - active_transcript.output_last_raw_frame = None; } RealtimeEvent::HandoffRequested(handoff) => { - truncate_active_transcript(&mut active_transcript.entries); - append_handoff_input(&mut active_transcript.entries, &handoff.input_transcript); + append_handoff_input(&mut active_transcript, &handoff.input_transcript); handoff.active_transcript = active_transcript.take_entries(); active_transcript.new_input_entry = true; active_transcript.new_output_entry = true; } RealtimeEvent::ResponseCreated(_) => { active_transcript.new_output_entry = true; - active_transcript.output_last_raw_frame = None; } RealtimeEvent::SessionUpdated { .. } | RealtimeEvent::AudioOut(_) @@ -792,113 +727,68 @@ impl RealtimeWebsocketEvents { | RealtimeEvent::HistoryTranscriptDelta { .. } | RealtimeEvent::HistoryItemCompleted(_) => {} } - if let Some(frame_sequence) = frame_sequence { - // Wait for both speakers before evicting utterances from this frame. Each updated - // utterance is bounded independently, so pending transcript text is at most the previous - // 8 KiB transcript plus 8 KiB per speaker, even if no further event arrives. - for entry in &mut active_transcript.entries { - if entry.last_raw_frame == Some(frame_sequence) { - truncate_transcript_entry(entry, MAX_ACTIVE_TRANSCRIPT_BYTES); - } - } - } else { - truncate_active_transcript(&mut active_transcript.entries); + } + + pub async fn take_completed_transcript( + &self, + event: &RealtimeEvent, + threshold: usize, + ) -> Vec { + if self.event_parser != RealtimeEventParser::FramelessBidi + || !matches!( + event, + RealtimeEvent::InputTranscriptDone(_) | RealtimeEvent::OutputTranscriptDone(_) + ) + { + return Vec::new(); } + let mut active = self.transcript_state.active.lock().await; + // Only the other speaker can still be open after turn.done. Keep their utterance + // and everything after it in place so history stays in conversation order. + let open_role = match event { + RealtimeEvent::InputTranscriptDone(_) if !active.new_output_entry => Some("assistant"), + RealtimeEvent::OutputTranscriptDone(_) if !active.new_input_entry => Some("user"), + _ => None, + }; + let entries = &active.entries; + let end = open_role + .and_then(|role| entries.iter().rposition(|entry| entry.role == role)) + .unwrap_or(entries.len()); + if end == 0 || transcript_entries_bytes(entries) <= threshold { + return Vec::new(); + } + let completed: Vec<_> = active.entries.drain(..end).collect(); + let recorded = &mut active.flushed_user_inputs; + let hash = recorded.hasher().clone(); + recorded.extend( + completed + .iter() + .filter(|entry| entry.role == "user") + .map(|entry| hash.hash_one(entry.text.trim())), + ); + completed } } impl RealtimeTranscriptState { pub async fn take_transcript_tail(&self) -> Vec { - let mut active = self.active.lock().await; - truncate_active_transcript(&mut active.entries); - active.take_entries() + self.active.lock().await.take_entries() } } -fn truncate_active_transcript(entries: &mut Vec) { - let mut total_bytes = transcript_entries_bytes(entries.iter().map(|entry| &entry.transcript)); - while total_bytes > MAX_ACTIVE_TRANSCRIPT_BYTES && entries.len() > 1 { - // An earlier utterance can contain the newest speech. Evict by raw frame recency, - // preserving the legacy FIFO policy without timing. - let oldest = if entries.iter().all(|entry| entry.last_raw_frame.is_some()) { - entries - .iter() - .enumerate() - .min_by_key(|(_, entry)| entry.last_raw_frame) - .map(|(index, _)| index) - .unwrap_or_default() - } else { - 0 - }; - if let Some(frame) = entries[oldest].last_raw_frame - && entries - .iter() - .all(|entry| entry.last_raw_frame == Some(frame)) - { - // Both speakers have equally recent speech. Keep their suffixes in conversation - // order, letting short entries keep their text before dividing the remaining budget. - let mut remaining_bytes = MAX_ACTIVE_TRANSCRIPT_BYTES; - let mut remaining_entries = entries.len(); - let mut by_size: Vec<_> = entries.iter_mut().collect(); - by_size.sort_unstable_by_key(|entry| transcript_entry_bytes(&entry.transcript)); - for entry in by_size { - let entry_budget = remaining_bytes / remaining_entries; - truncate_transcript_entry(entry, entry_budget); - remaining_bytes -= transcript_entry_bytes(&entry.transcript); - remaining_entries -= 1; - } - return; - } - total_bytes -= transcript_entry_bytes(&entries.remove(oldest).transcript); - } - if let Some(entry) = entries.first_mut() { - truncate_transcript_entry(entry, MAX_ACTIVE_TRANSCRIPT_BYTES); - } -} - -fn truncate_transcript_entry(entry: &mut ActiveTranscriptEntry, max_bytes: usize) { - let entry_overhead = entry.transcript.role.len() + 3; - let max_text_bytes = max_bytes.saturating_sub(entry_overhead); - if entry.transcript.text.len() <= max_text_bytes { - return; - } - let mut start = entry - .transcript - .text - .len() - .saturating_sub(max_text_bytes.saturating_sub(TRUNCATED_TRANSCRIPT_PREFIX.len())); - while !entry.transcript.text.is_char_boundary(start) { - start += 1; - } - let previous_marker_bytes = if entry.truncated_prefix_bytes > 0 { - TRUNCATED_TRANSCRIPT_PREFIX.len() - } else { - 0 - }; - entry.truncated_prefix_bytes += start - previous_marker_bytes; - entry.transcript.text = format!( - "{TRUNCATED_TRANSCRIPT_PREFIX}{}", - &entry.transcript.text[start..] - ); -} - -fn transcript_entries_bytes<'a>( - entries: impl IntoIterator, -) -> usize { - entries.into_iter().map(transcript_entry_bytes).sum() -} - -fn transcript_entry_bytes(entry: &RealtimeTranscriptEntry) -> usize { - entry.role.len() + entry.text.len() + 3 +fn transcript_entries_bytes(entries: &[RealtimeTranscriptEntry]) -> usize { + entries + .iter() + .map(|entry| entry.role.len() + entry.text.len() + 3) + .sum() } fn append_transcript_delta( - entries: &mut Vec, + entries: &mut Vec, role: &str, delta: &str, force_new: bool, event_parser: RealtimeEventParser, - frame_sequence: Option, ) { if delta.is_empty() { return; @@ -907,37 +797,30 @@ fn append_transcript_delta( // V3 interleaves the two speakers' transcripts. Legacy protocols retain // adjacent speaker runs and their existing response lifecycle semantics. let entry = match event_parser { - RealtimeEventParser::FramelessBidi => entries - .iter_mut() - .rev() - .find(|entry| entry.transcript.role == role), - RealtimeEventParser::V1 | RealtimeEventParser::RealtimeV2 => entries - .last_mut() - .filter(|entry| entry.transcript.role == role), + RealtimeEventParser::FramelessBidi => { + entries.iter_mut().rev().find(|entry| entry.role == role) + } + RealtimeEventParser::V1 | RealtimeEventParser::RealtimeV2 => { + entries.last_mut().filter(|entry| entry.role == role) + } }; if !force_new && let Some(last_entry) = entry { - last_entry.transcript.text.push_str(delta); - last_entry.last_raw_frame = frame_sequence; + last_entry.text.push_str(delta); return; } - entries.push(ActiveTranscriptEntry { - transcript: RealtimeTranscriptEntry { - role: role.to_string(), - text: delta.to_string(), - }, - last_raw_frame: frame_sequence, - truncated_prefix_bytes: 0, + entries.push(RealtimeTranscriptEntry { + role: role.to_string(), + text: delta.to_string(), }); } fn apply_transcript_done( - entries: &mut Vec, + entries: &mut Vec, role: &str, text: &str, force_new: bool, event_parser: RealtimeEventParser, - last_raw_frame: Option, ) { if text.is_empty() { return; @@ -946,59 +829,49 @@ fn apply_transcript_done( // V3 interleaves the two speakers' transcripts. Legacy protocols retain // adjacent speaker runs and their existing response lifecycle semantics. let entry = match event_parser { - RealtimeEventParser::FramelessBidi => entries - .iter_mut() - .rev() - .find(|entry| entry.transcript.role == role), - RealtimeEventParser::V1 | RealtimeEventParser::RealtimeV2 => entries - .last_mut() - .filter(|entry| entry.transcript.role == role), + RealtimeEventParser::FramelessBidi => { + entries.iter_mut().rev().find(|entry| entry.role == role) + } + RealtimeEventParser::V1 | RealtimeEventParser::RealtimeV2 => { + entries.last_mut().filter(|entry| entry.role == role) + } }; if !force_new && let Some(last_entry) = entry { // A delayed V3 final can arrive after deltas from the next utterance. // Preserve accumulated speech unless the final extends it. - if event_parser != RealtimeEventParser::FramelessBidi - || text - .get(last_entry.truncated_prefix_bytes..) - .is_some_and(|text| text.starts_with(last_entry.retained_text())) + if event_parser != RealtimeEventParser::FramelessBidi || text.starts_with(&last_entry.text) { - last_entry.transcript.text = text.to_string(); - last_entry.truncated_prefix_bytes = 0; + last_entry.text = text.to_string(); } return; } - entries.push(ActiveTranscriptEntry { - transcript: RealtimeTranscriptEntry { - role: role.to_string(), - text: text.to_string(), - }, - // A final for an evicted current utterance must not become newer than retained speech. - last_raw_frame: if force_new { None } else { last_raw_frame }, - truncated_prefix_bytes: 0, + entries.push(RealtimeTranscriptEntry { + role: role.to_string(), + text: text.to_string(), }); } -fn append_handoff_input(entries: &mut Vec, input: &str) { +fn append_handoff_input(active: &mut ActiveTranscriptState, input: &str) { let input = input.trim(); - if input.is_empty() || contains_transcript_entry(entries, "user", input) { + let recorded = &active.flushed_user_inputs; + if input.is_empty() + || contains_transcript_entry(&active.entries, "user", input) + || (!recorded.is_empty() && recorded.contains(&recorded.hasher().hash_one(input))) + { return; } - entries.push(ActiveTranscriptEntry { - transcript: RealtimeTranscriptEntry { - role: "user".to_string(), - text: input.to_string(), - }, - last_raw_frame: None, - truncated_prefix_bytes: 0, + active.entries.push(RealtimeTranscriptEntry { + role: "user".to_string(), + text: input.to_string(), }); } -fn contains_transcript_entry(entries: &[ActiveTranscriptEntry], role: &str, text: &str) -> bool { +fn contains_transcript_entry(entries: &[RealtimeTranscriptEntry], role: &str, text: &str) -> bool { entries .iter() - .any(|entry| entry.transcript.role == role && entry.transcript.text.trim() == text.trim()) + .any(|entry| entry.role == role && entry.text.trim() == text.trim()) } pub struct RealtimeWebsocketClient { @@ -1611,18 +1484,14 @@ mod tests { let mut covered = RealtimeEvent::InputTranscriptDelta(RealtimeTranscriptDelta { delta: "already handed off".to_string(), }); - events - .update_active_transcript(&mut covered, /*frame*/ None) - .await; + events.update_active_transcript(&mut covered).await; let mut handoff = RealtimeEvent::HandoffRequested(RealtimeHandoffRequested { handoff_id: "handoff_1".to_string(), item_id: "item_1".to_string(), input_transcript: "already handed off".to_string(), active_transcript: vec![], }); - events - .update_active_transcript(&mut handoff, /*frame*/ None) - .await; + events.update_active_transcript(&mut handoff).await; assert_eq!( handoff, RealtimeEvent::HandoffRequested(RealtimeHandoffRequested { @@ -1639,9 +1508,7 @@ mod tests { let mut tail = RealtimeEvent::OutputTranscriptDelta(RealtimeTranscriptDelta { delta: "tail".to_string(), }); - events - .update_active_transcript(&mut tail, /*frame*/ None) - .await; + events.update_active_transcript(&mut tail).await; assert_eq!( events.take_transcript_tail().await, vec![RealtimeTranscriptEntry { @@ -1666,9 +1533,7 @@ mod tests { let mut partial = RealtimeEvent::InputTranscriptDelta(RealtimeTranscriptDelta { delta: "hello wor".to_string(), }); - first_events - .update_active_transcript(&mut partial, /*frame*/ None) - .await; + first_events.update_active_transcript(&mut partial).await; let (_second_tx, second_rx) = async_channel::unbounded(); let second_events = RealtimeWebsocketEvents { @@ -1681,9 +1546,7 @@ mod tests { let mut done = RealtimeEvent::InputTranscriptDone(RealtimeTranscriptDone { text: "hello world".to_string(), }); - second_events - .update_active_transcript(&mut done, /*frame*/ None) - .await; + second_events.update_active_transcript(&mut done).await; assert_eq!( second_events.take_transcript_tail().await, @@ -1695,35 +1558,6 @@ mod tests { assert_eq!(second_events.take_transcript_tail().await, vec![]); } - #[tokio::test] - async fn active_transcript_retains_a_bounded_suffix() { - let (_tx_message, rx_message) = async_channel::unbounded(); - let events = RealtimeWebsocketEvents { - rx_message, - pending_events: Arc::new(Mutex::new(VecDeque::new())), - transcript_state: RealtimeTranscriptState::default(), - event_parser: RealtimeEventParser::FramelessBidi, - is_closed: Arc::new(AtomicBool::new(false)), - }; - let mut oversized = RealtimeEvent::InputTranscriptDelta(RealtimeTranscriptDelta { - delta: format!("old{}new", "x".repeat(MAX_ACTIVE_TRANSCRIPT_BYTES)), - }); - for frame in [ - None, - Some(TranscriptFrame { - start_ms: 0, - end_ms: 200, - }), - ] { - events.update_active_transcript(&mut oversized, frame).await; - - let tail = events.take_transcript_tail().await; - assert!(transcript_entries_bytes(&tail) <= MAX_ACTIVE_TRANSCRIPT_BYTES); - assert!(tail[0].text.starts_with(TRUNCATED_TRANSCRIPT_PREFIX)); - assert!(tail[0].text.ends_with("new")); - } - } - #[test] fn terminal_sideband_handshake_statuses_are_not_retryable() { for status in [StatusCode::NOT_FOUND, StatusCode::GONE] { diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/transcript_tests.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/transcript_tests.rs index 28eae40cd7..fa89609bb6 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/transcript_tests.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/transcript_tests.rs @@ -24,9 +24,7 @@ fn entry(role: &str, text: &str) -> RealtimeTranscriptEntry { async fn observe(events: &RealtimeWebsocketEvents, mut event: RealtimeEvent) { let original = event.clone(); - events - .update_active_transcript(&mut event, /*frame*/ None) - .await; + events.update_active_transcript(&mut event).await; assert_eq!( event, original, "live transcript events must pass through unchanged" @@ -131,176 +129,3 @@ async fn completed_utterances_keep_repeated_speech_and_final_only_turns_distinct ); } } - -#[tokio::test] -async fn transcript_budget_uses_frame_recency_without_reordering_speakers() { - use serde_json::json; - - let old_user = "u".repeat(4000); - let assistant = "a".repeat(4000); - let newest_user = "n".repeat(211); - let current_user = "u".repeat(4160); - let same_frame_user = "n".repeat(22); - let unicode_user = "é".repeat(2000); - let oversized_user = "u".repeat(MAX_ACTIVE_TRANSCRIPT_BYTES); - let extended_assistant = format!("{assistant}{newest_user}!"); - for (frames, final_assistant, expected) in [ - ( - vec![ - ("user", old_user.as_str(), Some(0)), - ("assistant", assistant.as_str(), Some(200)), - ("user", newest_user.as_str(), Some(400)), - // Commit eviction before the delayed final arrives. - ("user", "", Some(600)), - ], - // A late final must keep the evicted assistant's old raw-frame age. - Some(assistant.as_str()), - vec![entry("user", &format!("{old_user}{newest_user}"))], - ), - ( - vec![ - ("assistant", assistant.as_str(), Some(0)), - ("user", current_user.as_str(), Some(200)), - // The first event already exceeds the budget, before output updates its age. - ("user", same_frame_user.as_str(), Some(400)), - // Same-frame input is emitted first; output must not become newer. - ("assistant", "bbbbbbbbbbbbbbbbbbbb", Some(400)), - ], - None, - vec![ - entry("assistant", &format!("{assistant}bbbbbbbbbbbbbbbbbbbb")), - entry("user", &format!("…{}{same_frame_user}", "u".repeat(4128))), - ], - ), - ( - vec![ - ("user", old_user.as_str(), Some(0)), - ("assistant", assistant.as_str(), Some(200)), - ("user", newest_user.as_str(), Some(400)), - ("assistant", "b", Some(400)), - ], - None, - vec![ - entry("user", &format!("…{}{newest_user}", "u".repeat(3958))), - entry("assistant", &format!("{assistant}b")), - ], - ), - ( - vec![ - ("user", old_user.as_str(), Some(0)), - ("assistant", assistant.as_str(), Some(200)), - // Reversing the same-frame events must retain the same suffixes. - ("assistant", "b", Some(400)), - ("user", newest_user.as_str(), Some(400)), - // Advancing the frame also commits eviction before the tail is taken. - ("user", "", Some(600)), - ], - None, - vec![ - entry("user", &format!("…{}{newest_user}", "u".repeat(3958))), - entry("assistant", &format!("{assistant}b")), - ], - ), - ( - vec![ - ("user", unicode_user.as_str(), Some(0)), - ("assistant", assistant.as_str(), Some(200)), - ("user", newest_user.as_str(), Some(400)), - ("assistant", newest_user.as_str(), Some(400)), - ], - None, - vec![ - entry("user", &format!("…{}{newest_user}", "é".repeat(1937))), - entry("assistant", &format!("…{}{newest_user}", "a".repeat(3871))), - ], - ), - ( - vec![ - ("user", oversized_user.as_str(), Some(0)), - ("assistant", "b", Some(0)), - ], - None, - vec![ - entry("user", &format!("…{}", "u".repeat(8169))), - entry("assistant", "b"), - ], - ), - ( - vec![ - ("assistant", assistant.as_str(), Some(0)), - ("user", old_user.as_str(), Some(200)), - ("assistant", newest_user.as_str(), Some(400)), - ("user", "b", Some(400)), - // Trim the tied utterances before a final extends the retained suffix. - ("user", "", Some(600)), - ], - Some(extended_assistant.as_str()), - vec![ - entry("assistant", &format!("…{}{newest_user}!", "a".repeat(3957))), - entry("user", &format!("{old_user}b")), - ], - ), - ( - vec![ - ("user", old_user.as_str(), None), - ("assistant", assistant.as_str(), None), - ("user", newest_user.as_str(), None), - ], - None, - vec![entry("assistant", &assistant)], - ), - ( - vec![ - ("assistant", "I will", Some(0)), - ("user", "Stop!", Some(200)), - ("assistant", " do it", Some(200)), - ], - Some("I will do it"), - vec![entry("assistant", "I will do it"), entry("user", "Stop!")], - ), - ] { - let state = RealtimeTranscriptState::default(); - let mut events = events(state.clone()); - let (sender, receiver) = async_channel::unbounded(); - events.rx_message = receiver; - for (role, text, start_ms) in frames { - let event_type = if role == "user" { - "input_transcript.added" - } else { - "output_transcript.added" - }; - let mut payload = json!({"type": event_type, "item": {"text": text}}); - if let Some(start_ms) = start_ms { - payload["start_ms"] = json!(start_ms); - payload["end_ms"] = json!(start_ms + 200); - } - sender - .send(Ok(Message::Text(payload.to_string().into()))) - .await - .unwrap(); - let delta = RealtimeTranscriptDelta { delta: text.into() }; - assert_eq!( - events.next_event().await.unwrap(), - Some(if role == "user" { - RealtimeEvent::InputTranscriptDelta(delta) - } else { - RealtimeEvent::OutputTranscriptDelta(delta) - }), - ); - } - if let Some(text) = final_assistant { - sender.send(Ok(Message::Text(json!({ - "type": "turn.done", - "turn": {"role": "assistant", "transcript": text, "start_ms": 0, "end_ms": 800} - }).to_string().into()))).await.unwrap(); - assert_eq!( - events.next_event().await.unwrap(), - Some(RealtimeEvent::OutputTranscriptDone( - RealtimeTranscriptDone { text: text.into() } - )), - ); - } - assert_eq!(state.take_transcript_tail().await, expected); - assert_eq!(state.take_transcript_tail().await, vec![]); - } -} diff --git a/codex-rs/core/src/context/realtime_delegation.rs b/codex-rs/core/src/context/realtime_delegation.rs index af66caae42..5853544fd3 100644 --- a/codex-rs/core/src/context/realtime_delegation.rs +++ b/codex-rs/core/src/context/realtime_delegation.rs @@ -3,7 +3,6 @@ use codex_protocol::models::ContentItem; use codex_protocol::models::ContentItemKind; use codex_protocol::models::input_text; -const MAX_REALTIME_DELEGATION_FIELD_BYTES: usize = 4 * 1024; const TRUNCATION_MARKER: &str = "…"; #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -17,18 +16,23 @@ pub(crate) struct RealtimeDelegation<'a> { input: &'a str, transcript_delta: Option<&'a str>, source: RealtimeDelegationSource, + preserve_transcript: bool, } impl<'a> RealtimeDelegation<'a> { + pub(crate) const FIELD_BYTES: usize = 4 * 1024; + pub(crate) fn new( input: &'a str, transcript_delta: Option<&'a str>, source: RealtimeDelegationSource, + preserve_transcript: bool, ) -> Self { Self { input, transcript_delta, source, + preserve_transcript, } } } @@ -59,7 +63,11 @@ impl ContextualUserFragment for RealtimeDelegation<'_> { } }; if let Some(transcript_delta) = self.transcript_delta.filter(|text| !text.is_empty()) { - let transcript_delta = escape_xml_text_bounded(transcript_delta, Retain::End); + let transcript_delta = if self.preserve_transcript { + escape_xml_text(transcript_delta) + } else { + escape_xml_text_bounded(transcript_delta, Retain::End) + }; return input_text(format!( "\n{source} {input}\n {transcript_delta}\n" )); @@ -77,10 +85,10 @@ enum Retain { fn escape_xml_text_bounded(input: &str, retain: Retain) -> String { let escaped = escape_xml_text(input); - if escaped.len() <= MAX_REALTIME_DELEGATION_FIELD_BYTES { + if escaped.len() <= RealtimeDelegation::FIELD_BYTES { return escaped; } - let retained_bytes = MAX_REALTIME_DELEGATION_FIELD_BYTES - TRUNCATION_MARKER.len(); + let retained_bytes = RealtimeDelegation::FIELD_BYTES - TRUNCATION_MARKER.len(); match retain { Retain::Start => { let mut end = retained_bytes; diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index b689988cfd..ffa74f075d 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -160,6 +160,16 @@ impl RealtimeHandoffAdmission { Ok(()) } + async fn record_transcript(&self, session: &Session, text: String) -> std::io::Result<()> { + let Ok(_permit) = self.gate.acquire().await else { + return Ok(()); + }; + if !self.retired.load(Ordering::Acquire) { + transcript_tail::record(session, text).await?; + } + Ok(()) + } + async fn retire(&self) { let Ok(_permit) = self.gate.acquire().await else { return; @@ -477,13 +487,16 @@ impl RealtimeResponseCreateQueue { } } +// Any completed transcript travels with its event so the fanout persists it before later handoffs. +type RealtimeOutput = (RealtimeEvent, Option); + struct RealtimeInputTask { writer: RealtimeWebsocketWriter, events: RealtimeWebsocketEvents, text_rx: Receiver, handoff_output_rx: Receiver, audio_rx: Receiver, - events_tx: Sender, + events_tx: Sender, handoff_state: RealtimeHandoffState, session_kind: RealtimeSessionKind, event_parser: RealtimeEventParser, @@ -492,6 +505,7 @@ struct RealtimeInputTask { struct RealtimeTranscriptTailFlush { enabled: bool, + event_parser: RealtimeEventParser, tx: Sender, } @@ -552,7 +566,7 @@ struct RealtimeStart { struct RealtimeStartOutput { realtime_active: Arc, route_handoffs: Arc, - events_rx: Receiver, + events_rx: Receiver, transcript_tail_rx: Receiver, sdp: Option, } @@ -678,7 +692,7 @@ impl RealtimeConversationManager { let (handoff_output_tx, handoff_output_rx) = async_channel::bounded::(HANDOFF_OUT_QUEUE_CAPACITY); let (events_tx, events_rx) = - async_channel::bounded::(OUTPUT_EVENTS_QUEUE_CAPACITY); + async_channel::bounded::(OUTPUT_EVENTS_QUEUE_CAPACITY); let (transcript_tail_tx, transcript_tail_rx) = async_channel::bounded::(1); let realtime_active = Arc::new(AtomicBool::new(true)); @@ -712,6 +726,7 @@ impl RealtimeConversationManager { }; let transcript_tail_flush = RealtimeTranscriptTailFlush { enabled: flush_transcript_tail_on_session_end, + event_parser, tx: transcript_tail_tx, }; let (task, sdp) = if let Some(sdp) = sdp { @@ -1739,6 +1754,7 @@ async fn handle_start_inner( start: realtime_start_instructions, end: realtime_end_instructions, }; + let event_parser = session_config.event_parser; let start = RealtimeStart { api_provider, http_client_factory, @@ -1799,7 +1815,20 @@ async fn handle_start_inner( let mut end = RealtimeConversationEnd::TransportClosed; let mut handoff_error = None; // Drain already-parsed events so a queued handoff is routed before the final tail. - while let Ok(event) = events_rx.recv().await { + while let Ok((event, completed)) = events_rx.recv().await { + if let Some(delta) = completed { + let text = wrap_realtime_delegation_input( + "", + Some(&delta), + RealtimeDelegationSource::Handoff, + /*preserve_transcript*/ true, + ); + if let Err(err) = route_handoffs.record_transcript(&sess_clone, text).await { + warn!("failed to save realtime transcript: {err}"); + handoff_error = Some("failed to save the realtime transcript"); + break; + } + } match &event { RealtimeEvent::AudioOut(_) => {} _ => { @@ -1812,9 +1841,10 @@ async fn handle_start_inner( end = RealtimeConversationEnd::Error; } let maybe_routed_text = match &event { - RealtimeEvent::HandoffRequested(handoff) => { - realtime_delegation_from_handoff(handoff) - } + RealtimeEvent::HandoffRequested(handoff) => realtime_delegation_from_handoff( + handoff, + event_parser == RealtimeEventParser::FramelessBidi, + ), _ => None, }; if let Some(text) = maybe_routed_text { @@ -1833,16 +1863,13 @@ async fn handle_start_inner( break; } } + drop(events_rx); if handoff_error.is_none() && let Ok(text) = transcript_tail_rx.recv().await + && let Err(err) = route_handoffs.record_transcript(&sess_clone, text).await { - let _permit = route_handoffs.gate.acquire().await; - if !route_handoffs.retired.load(Ordering::Acquire) - && let Err(err) = transcript_tail::record(&sess_clone, text).await - { - warn!("failed to flush realtime transcript before closure: {err}"); - handoff_error = Some("failed to save the realtime transcript before closure"); - } + warn!("failed to flush realtime transcript before closure: {err}"); + handoff_error = Some("failed to save the realtime transcript before closure"); } // Make realtime_end eligible only after the tail has reached history. { @@ -1918,12 +1945,16 @@ fn realtime_text_from_handoff_request(handoff: &RealtimeHandoffRequested) -> Opt .or_else(|| realtime_transcript_delta_from_handoff(handoff)) } -fn realtime_delegation_from_handoff(handoff: &RealtimeHandoffRequested) -> Option { +fn realtime_delegation_from_handoff( + handoff: &RealtimeHandoffRequested, + preserve_transcript: bool, +) -> Option { let input = realtime_text_from_handoff_request(handoff)?; Some(wrap_realtime_delegation_input( &input, realtime_transcript_delta_from_handoff(handoff).as_deref(), RealtimeDelegationSource::Handoff, + preserve_transcript, )) } @@ -1931,8 +1962,9 @@ fn wrap_realtime_delegation_input( input: &str, transcript_delta: Option<&str>, source: RealtimeDelegationSource, + preserve_transcript: bool, ) -> String { - RealtimeDelegation::new(input, transcript_delta, source) + RealtimeDelegation::new(input, transcript_delta, source, preserve_transcript) .render() .into_text() .unwrap_or_default() @@ -2119,7 +2151,7 @@ struct RealtimeWebrtcSidebandInputTask { sideband_headers: HeaderMap, session_initialization: RealtimeSidebandSessionInitialization, input_channels: RealtimeInputChannels, - events_tx: Sender, + events_tx: Sender, handoff_state: RealtimeHandoffState, session_kind: RealtimeSessionKind, event_parser: RealtimeEventParser, @@ -2176,8 +2208,9 @@ async fn run_realtime_input_task( realtime_event = events.next_event() => { match realtime_event { Ok(Some(event)) => { + let completed = events.take_completed_transcript(&event, RealtimeDelegation::FIELD_BYTES).await; handle_realtime_server_event( - event, + (event, realtime_transcript_delta(&completed)), &writer, &events_tx, &handoff_state, @@ -2290,11 +2323,11 @@ async fn send_realtime_pending_outbound( } } -async fn report_realtime_transport_loss(events_tx: &Sender, err: ApiError) { +async fn report_realtime_transport_loss(events_tx: &Sender, err: ApiError) { let mapped_error = map_api_error(err); error!("realtime stream closed: {mapped_error}"); let _ = events_tx - .send(RealtimeEvent::Error(mapped_error.to_string())) + .send((RealtimeEvent::Error(mapped_error.to_string()), None)) .await; } @@ -2311,6 +2344,7 @@ async fn flush_realtime_transcript_tail( REALTIME_SESSION_ENDED_HANDOFF_INSTRUCTION, Some(&transcript_delta), RealtimeDelegationSource::TranscriptTailFlush, + transcript_tail_flush.event_parser == RealtimeEventParser::FramelessBidi, )) .await; } @@ -2606,9 +2640,9 @@ async fn handle_handoff_output( } async fn handle_realtime_server_event( - event: RealtimeEvent, + (event, completed): RealtimeOutput, writer: &RealtimeWebsocketWriter, - events_tx: &Sender, + events_tx: &Sender, handoff_state: &RealtimeHandoffState, session_kind: RealtimeSessionKind, output_audio_state: &mut Option, @@ -2754,7 +2788,7 @@ async fn handle_realtime_server_event( | RealtimeEvent::HistoryItemCompleted(_) => false, }; - if events_tx.send(event).await.is_err() { + if events_tx.send((event, completed)).await.is_err() { anyhow::bail!("realtime output event channel closed"); } if should_stop { diff --git a/codex-rs/core/src/realtime_conversation/existing_call.rs b/codex-rs/core/src/realtime_conversation/existing_call.rs index 21ee18dcb3..11ed523ec7 100644 --- a/codex-rs/core/src/realtime_conversation/existing_call.rs +++ b/codex-rs/core/src/realtime_conversation/existing_call.rs @@ -1,5 +1,6 @@ use super::RealtimeHandoffState; use super::RealtimeInputChannels; +use super::RealtimeOutput; use super::RealtimeSessionKind; use super::RealtimeSidebandSessionInitialization; use super::RealtimeTranscriptTailFlush; @@ -7,7 +8,6 @@ use super::RealtimeWebrtcSidebandInputTask; use super::spawn_webrtc_sideband_input_task; use crate::client::ModelClient; use async_channel::Sender; -use codex_api::RealtimeEvent; use codex_api::RealtimeEventParser; use codex_api::RealtimeSessionConfig; use codex_api::RealtimeTranscriptState; @@ -28,7 +28,7 @@ pub(super) struct ExistingCallAttachment { pub(super) call_id: String, pub(super) extra_headers: HeaderMap, pub(super) input_channels: RealtimeInputChannels, - pub(super) events_tx: Sender, + pub(super) events_tx: Sender, pub(super) handoff_state: RealtimeHandoffState, pub(super) session_kind: RealtimeSessionKind, pub(super) event_parser: RealtimeEventParser, diff --git a/codex-rs/core/src/realtime_conversation/sideband.rs b/codex-rs/core/src/realtime_conversation/sideband.rs index e378592c5a..1884722fdd 100644 --- a/codex-rs/core/src/realtime_conversation/sideband.rs +++ b/codex-rs/core/src/realtime_conversation/sideband.rs @@ -107,7 +107,7 @@ pub(super) fn spawn_webrtc_sideband_input_task( "failed to connect realtime sideband: {mapped_error}" ); let _ = events_tx - .send(RealtimeEvent::Error(mapped_error.to_string())) + .send((RealtimeEvent::Error(mapped_error.to_string()), None)) .await; break; } diff --git a/codex-rs/core/src/realtime_conversation/transcript_tail.rs b/codex-rs/core/src/realtime_conversation/transcript_tail.rs index db8e85da72..86e27dd8e7 100644 --- a/codex-rs/core/src/realtime_conversation/transcript_tail.rs +++ b/codex-rs/core/src/realtime_conversation/transcript_tail.rs @@ -1,4 +1,4 @@ -//! Saves terminal speech in model and client history before realtime closure, without inference. +//! Records realtime transcript chunks in model and client history without inference. use crate::session::new_submission_id; use crate::session::session::Session; diff --git a/codex-rs/core/src/realtime_conversation_tests.rs b/codex-rs/core/src/realtime_conversation_tests.rs index 985fc17149..1d45d45fa2 100644 --- a/codex-rs/core/src/realtime_conversation_tests.rs +++ b/codex-rs/core/src/realtime_conversation_tests.rs @@ -307,7 +307,7 @@ fn wraps_handoff_with_transcript_delta() { ], }; assert_eq!( - realtime_delegation_from_handoff(&handoff), + realtime_delegation_from_handoff(&handoff, /*preserve_transcript*/ false), Some( "\n delegate this\n user: hello\nassistant: hi there\n" .to_string() @@ -347,6 +347,7 @@ fn wraps_realtime_delegation_input() { "hello", /*transcript_delta*/ None, RealtimeDelegationSource::Handoff, + /*preserve_transcript*/ false, ), "\n hello\n" ); @@ -359,6 +360,7 @@ fn wraps_realtime_delegation_input_with_xml_escaping() { "use a < b && c > d", Some("saw "), RealtimeDelegationSource::Handoff, + /*preserve_transcript*/ false, ), "\n use a < b && c > d\n saw <that>\n" ); @@ -371,6 +373,7 @@ fn wraps_realtime_delegation_input_with_xml_escaping_without_transcript() { "use a < b && c > d", /*transcript_delta*/ None, RealtimeDelegationSource::Handoff, + /*preserve_transcript*/ false, ), "\n use a < b && c > d\n" ); @@ -384,6 +387,7 @@ fn bounds_realtime_delegation_fields_and_keeps_latest_transcript() { &input, Some(&transcript), RealtimeDelegationSource::Handoff, + /*preserve_transcript*/ false, ); assert!(rendered.len() < 9 * 1024); diff --git a/codex-rs/core/tests/suite/realtime_attachment.rs b/codex-rs/core/tests/suite/realtime_attachment.rs index 1edadc7977..ff7b3738d3 100644 --- a/codex-rs/core/tests/suite/realtime_attachment.rs +++ b/codex-rs/core/tests/suite/realtime_attachment.rs @@ -319,3 +319,87 @@ async fn failed_attachment_leaves_core_disconnected() -> Result<()> { test.codex.shutdown_and_wait().await?; Ok(()) } + +#[test_case::test_case(true; "record unfinished tail")] +#[test_case::test_case(false; "drop unfinished tail")] +#[tokio::test] +async fn completed_turns_reach_history_before_the_next_handoff(flush_tail: bool) -> Result<()> { + use codex_protocol::protocol::RealtimeEvent; + use core_test_support::responses; + use core_test_support::wait_for_event; + + let api = start_mock_server().await; + let replies = ["handoff", "text"].map(|id| { + responses::sse(vec![ + responses::ev_response_created(id), + responses::ev_completed(id), + ]) + }); + let model = responses::mount_sse_sequence(&api, replies.into()).await; + let test = test_codex().build_with_auto_env(&api).await?; + let mut sideband = Sideband::new().await?; + let mut params = params("transcript", &sideband.url); + params.flush_transcript_tail_on_session_end = flush_tail; + let (reply, attached) = oneshot::channel(); + test.codex + .submit(Op::RealtimeConversationAttach { params, reply }) + .await?; + let connection = sideband.connected().await; + connection + .accept + .send(/*value*/ true) + .expect("pending handshake"); + timeout(Duration::from_secs(10), attached).await???; + let send = |event| connection.events.send(event); + + let speech = format!("{}&", "u".repeat(12 * 1024)); + let final_text = format!("{speech}!"); + send(json!({"type":"input_transcript.added", "item":{"text":speech}}))?; + send(json!({"type":"turn.done", "turn":{"role":"assistant","transcript":"yes"}}))?; + send(json!({"type":"turn.done", "turn":{"role":"user","transcript":final_text}}))?; + wait_for_event(&test.codex, |event| matches!(event, EventMsg::RealtimeConversationRealtime(event) if matches!(event.payload, RealtimeEvent::InputTranscriptDone(_)))).await; + assert!(model.requests().is_empty()); + + // The handoff repeats the user input. Only the new partial belongs in its transcript. + let partial = "p".repeat(12 * 1024); + send(json!({"type":"output_transcript.added", "item":{"text":partial}}))?; + send( + json!({"type":"delegation.created","item":{"id":"handoff","type":"delegation","target":"client","content":[{"type":"input_text","text":final_text}]}}), + )?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let transcript = |request: ResponsesRequest| { + request + .message_input_texts("user") + .into_iter() + .filter(|text| text.contains("")) + .collect::>() + }; + let completed = transcript(model.single_request()); + assert_eq!(completed.len(), 2); + let expected = format!( + "user: {}\nassistant: yes", + final_text.replace('&', "&") + ); + assert!(completed[0].contains("") && completed[0].contains(&expected)); + assert!(completed[1].contains(&format!( + "assistant: {partial}" + ))); + + let unfinished = "a".repeat(12 * 1024); + send(json!({"type":"output_transcript.added", "item":{"text":unfinished}}))?; + wait_for_event(&test.codex, |event| matches!(event, EventMsg::RealtimeConversationRealtime(event) if matches!(event.payload, RealtimeEvent::OutputTranscriptDelta(_)))).await; + detach(&test, "transcript").await?; + test.submit_text_turn("continue").await?; + let after_close = transcript(model.last_request().expect("text turn")); + assert_eq!(after_close.len(), completed.len() + usize::from(flush_tail)); + assert_eq!(&after_close[..completed.len()], completed.as_slice()); + if flush_tail { + assert!(after_close[completed.len()].contains(&format!( + "assistant: {unfinished}" + ))); + } + Ok(()) +}