Fix realtime V3 transcript reconciliation for handoffs (#46922)

## Why

Interleaved speaker transcripts can fragment handoff context, and delayed final transcripts can overwrite newer speech.

## What changed

- Reconcile V3 transcript deltas and finals with the latest entry for the same speaker.
- Replace accumulated text only when the final transcript extends it, preserving newer speech when a final arrives late.
- Start a new entry after each completed utterance, keeping repeated speech and final-only turns distinct even when empty deltas arrive.
- Preserve live transcript notifications and legacy protocol behavior.

## Testing

Add unit coverage for interleaved speakers across reconnects, either completion order, repeated utterances, final-only turns, and empty deltas. Add app-server integration coverage for expanded and delayed finals, checking live notifications and reconciled handoff context.

GitOrigin-RevId: 961729281f61b95fd3e0c5cc3ec72512987980b3

codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs

@@ -603,8 +603,16 @@ impl RealtimeWebsocketEvents {
             }
             RealtimeEvent::InputTranscriptDelta(RealtimeTranscriptDelta { delta, .. }) => {
                 let force_new = active_transcript.new_input_entry;
-                append_transcript_delta(&mut active_transcript.entries, "user", delta, force_new);
-                active_transcript.new_input_entry = false;
+                append_transcript_delta(
+                    &mut active_transcript.entries,
+                    "user",
+                    delta,
+                    force_new,
+                    self.event_parser,
+                );
+                if !delta.is_empty() || self.event_parser != RealtimeEventParser::FramelessBidi {
+                    active_transcript.new_input_entry = false;
+                }
             }
             RealtimeEvent::OutputTranscriptDelta(RealtimeTranscriptDelta { delta, .. }) => {
                 let force_new = active_transcript.new_output_entry;
@@ -613,8 +621,11 @@ impl RealtimeWebsocketEvents {
                     "assistant",
                     delta,
                     force_new,
+                    self.event_parser,
                 );
-                active_transcript.new_output_entry = false;
+                if !delta.is_empty() || self.event_parser != RealtimeEventParser::FramelessBidi {
+                    active_transcript.new_output_entry = false;
+                }
             }
             RealtimeEvent::InputTranscriptDone(done) => {
                 let force_new = active_transcript.new_input_entry;
@@ -623,8 +634,10 @@ impl RealtimeWebsocketEvents {
                     "user",
                     &done.text,
                     force_new,
+                    self.event_parser,
                 );
-                active_transcript.new_input_entry = false;
+                active_transcript.new_input_entry =
+                    self.event_parser == RealtimeEventParser::FramelessBidi;
             }
             RealtimeEvent::OutputTranscriptDone(done) => {
                 let force_new = active_transcript.new_output_entry;
@@ -633,8 +646,10 @@ impl RealtimeWebsocketEvents {
                     "assistant",
                     &done.text,
                     force_new,
+                    self.event_parser,
                 );
-                active_transcript.new_output_entry = false;
+                active_transcript.new_output_entry =
+                    self.event_parser == RealtimeEventParser::FramelessBidi;
             }
             RealtimeEvent::HandoffRequested(handoff) => {
                 append_handoff_input(&mut active_transcript.entries, &handoff.input_transcript);
@@ -704,15 +719,23 @@ fn append_transcript_delta(
     role: &str,
     delta: &str,
     force_new: bool,
+    event_parser: RealtimeEventParser,
 ) {
     if delta.is_empty() {
         return;
     }
 
-    if !force_new
-        && let Some(last_entry) = entries.last_mut()
-        && last_entry.role == role
-    {
+    // 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.role == role)
+        }
+        RealtimeEventParser::V1 | RealtimeEventParser::RealtimeV2 => {
+            entries.last_mut().filter(|entry| entry.role == role)
+        }
+    };
+    if !force_new && let Some(last_entry) = entry {
         last_entry.text.push_str(delta);
         return;
     }
@@ -728,16 +751,29 @@ fn apply_transcript_done(
     role: &str,
     text: &str,
     force_new: bool,
+    event_parser: RealtimeEventParser,
 ) {
     if text.is_empty() {
         return;
     }
 
-    if !force_new
-        && let Some(last_entry) = entries.last_mut()
-        && last_entry.role == role
-    {
-        last_entry.text = text.to_string();
+    // 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.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.starts_with(&last_entry.text)
+        {
+            last_entry.text = text.to_string();
+        }
         return;
     }
 
@@ -1212,6 +1248,10 @@ fn normalize_realtime_path(url: &mut Url, event_parser: RealtimeEventParser) {
     }
 }
 
+#[cfg(test)]
+#[path = "transcript_tests.rs"]
+mod transcript_tests;
+
 #[cfg(test)]
 mod tests {
     use super::*;

codex-rs/codex-api/src/endpoint/realtime_websocket/transcript_tests.rs

@@ -0,0 +1,131 @@
+//! V3 transcript reconciliation across interleaved speakers, reconnects, and turn boundaries.
+
+use super::*;
+use codex_protocol::protocol::RealtimeTranscriptDone;
+use pretty_assertions::assert_eq;
+
+fn events(transcript_state: RealtimeTranscriptState) -> RealtimeWebsocketEvents {
+    let (_, rx_message) = async_channel::unbounded();
+    RealtimeWebsocketEvents {
+        rx_message,
+        pending_events: Arc::new(Mutex::new(VecDeque::new())),
+        transcript_state,
+        event_parser: RealtimeEventParser::FramelessBidi,
+        is_closed: Arc::new(AtomicBool::new(false)),
+    }
+}
+
+fn entry(role: &str, text: &str) -> RealtimeTranscriptEntry {
+    RealtimeTranscriptEntry {
+        role: role.to_string(),
+        text: text.to_string(),
+    }
+}
+
+async fn observe(events: &RealtimeWebsocketEvents, mut event: RealtimeEvent) {
+    let original = event.clone();
+    events.update_active_transcript(&mut event).await;
+    assert_eq!(
+        event, original,
+        "live transcript events must pass through unchanged"
+    );
+}
+
+#[tokio::test]
+async fn interleaved_transcripts_reconcile_across_reconnect_in_either_completion_order() {
+    for assistant_finishes_first in [false, true] {
+        let state = RealtimeTranscriptState::default();
+        let first = events(state.clone());
+        for event in [
+            RealtimeEvent::InputTranscriptDelta(RealtimeTranscriptDelta {
+                delta: "What can ".into(),
+            }),
+            RealtimeEvent::OutputTranscriptDelta(RealtimeTranscriptDelta {
+                delta: "A bunch ".into(),
+            }),
+            RealtimeEvent::InputTranscriptDelta(RealtimeTranscriptDelta {
+                delta: "you do?".into(),
+            }),
+            RealtimeEvent::OutputTranscriptDelta(RealtimeTranscriptDelta {
+                delta: "of stuff".into(),
+            }),
+        ] {
+            observe(&first, event).await;
+        }
+        let reconnected = events(state);
+        let user = RealtimeEvent::InputTranscriptDone(RealtimeTranscriptDone {
+            text: "What can you do?".into(),
+        });
+        let assistant = RealtimeEvent::OutputTranscriptDone(RealtimeTranscriptDone {
+            text: "A bunch of stuff, including coding.".into(),
+        });
+        let finals = if assistant_finishes_first {
+            [assistant, user]
+        } else {
+            [user, assistant]
+        };
+        for event in finals {
+            observe(&reconnected, event).await;
+        }
+        assert_eq!(
+            reconnected.take_transcript_tail().await,
+            vec![
+                entry("user", "What can you do?"),
+                entry("assistant", "A bunch of stuff, including coding."),
+            ]
+        );
+        assert_eq!(reconnected.take_transcript_tail().await, vec![]);
+    }
+}
+
+#[tokio::test]
+async fn completed_utterances_keep_repeated_speech_and_final_only_turns_distinct() {
+    let events = events(RealtimeTranscriptState::default());
+    for role in ["user", "assistant"] {
+        for with_delta in [true, true, false] {
+            if with_delta {
+                let delta = RealtimeTranscriptDelta {
+                    delta: "Again".into(),
+                };
+                observe(
+                    &events,
+                    if role == "user" {
+                        RealtimeEvent::InputTranscriptDelta(delta)
+                    } else {
+                        RealtimeEvent::OutputTranscriptDelta(delta)
+                    },
+                )
+                .await;
+            }
+            let done = RealtimeTranscriptDone {
+                text: "Again.".into(),
+            };
+            observe(
+                &events,
+                if role == "user" {
+                    RealtimeEvent::InputTranscriptDone(done)
+                } else {
+                    RealtimeEvent::OutputTranscriptDone(done)
+                },
+            )
+            .await;
+            // An empty delta must not reopen the utterance we just completed.
+            let empty = RealtimeTranscriptDelta {
+                delta: String::new(),
+            };
+            observe(
+                &events,
+                if role == "user" {
+                    RealtimeEvent::InputTranscriptDelta(empty)
+                } else {
+                    RealtimeEvent::OutputTranscriptDelta(empty)
+                },
+            )
+            .await;
+        }
+        assert_eq!(
+            events.take_transcript_tail().await,
+            vec![entry(role, "Again."); 3]
+        );
+    }
+}