Flush trailing realtime transcript on session end

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

@@ -219,6 +219,8 @@ pub struct RealtimeWebsocketEvents {
 struct ActiveTranscriptState {
     entries: Vec<RealtimeTranscriptEntry>,
     last_handoff_entry_count: usize,
+    last_requested_handoff_entry_count: usize,
+    handoff_entry_counts: HashMap<String, usize>,
     new_input_entry: bool,
     new_output_entry: bool,
 }
@@ -391,6 +393,25 @@ impl RealtimeWebsocketWriter {
 }
 
 impl RealtimeWebsocketEvents {
+    /// Marks the transcript included in a handoff as durably accepted by the caller.
+    pub async fn commit_transcript_handoff(&self, handoff_id: &str) {
+        let mut active_transcript = self.active_transcript.lock().await;
+        if let Some(entry_count) = active_transcript.handoff_entry_counts.remove(handoff_id) {
+            active_transcript.last_handoff_entry_count =
+                active_transcript.last_handoff_entry_count.max(entry_count);
+        }
+    }
+
+    /// Takes transcript entries not committed through a handoff, advancing the cursor once.
+    pub async fn take_transcript_tail(&self) -> Vec<RealtimeTranscriptEntry> {
+        let mut active_transcript = self.active_transcript.lock().await;
+        let tail = active_transcript.entries[active_transcript.last_handoff_entry_count..].to_vec();
+        active_transcript.last_handoff_entry_count = active_transcript.entries.len();
+        active_transcript.last_requested_handoff_entry_count = active_transcript.entries.len();
+        active_transcript.handoff_entry_counts.clear();
+        tail
+    }
+
     pub async fn next_event(&self) -> Result<Option<RealtimeEvent>, ApiError> {
         if self.is_closed.load(Ordering::SeqCst) {
             return Ok(None);
@@ -485,10 +506,14 @@ impl RealtimeWebsocketEvents {
             }
             RealtimeEvent::HandoffRequested(handoff) => {
                 append_handoff_input(&mut active_transcript.entries, &handoff.input_transcript);
+                let entry_count = active_transcript.entries.len();
                 handoff.active_transcript = active_transcript.entries
-                    [active_transcript.last_handoff_entry_count..]
+                    [active_transcript.last_requested_handoff_entry_count..entry_count]
                     .to_vec();
-                active_transcript.last_handoff_entry_count = active_transcript.entries.len();
+                active_transcript.last_requested_handoff_entry_count = entry_count;
+                active_transcript
+                    .handoff_entry_counts
+                    .insert(handoff.handoff_id.clone(), entry_count);
                 active_transcript.new_input_entry = true;
                 active_transcript.new_output_entry = true;
             }
@@ -872,6 +897,94 @@ mod tests {
     use tokio_tungstenite::accept_async;
     use tokio_tungstenite::tungstenite::Message;
 
+    fn transcript_events() -> RealtimeWebsocketEvents {
+        let (_tx_message, rx_message) = async_channel::unbounded();
+        RealtimeWebsocketEvents {
+            rx_message,
+            active_transcript: Arc::new(Mutex::new(ActiveTranscriptState::default())),
+            event_parser: RealtimeEventParser::RealtimeV2,
+            is_closed: Arc::new(AtomicBool::new(false)),
+        }
+    }
+
+    #[tokio::test]
+    async fn empty_transcript_has_no_tail() {
+        assert_eq!(transcript_events().take_transcript_tail().await, Vec::new());
+    }
+
+    #[tokio::test]
+    async fn takes_trailing_transcript_once() {
+        let events = transcript_events();
+        events
+            .update_active_transcript(&mut RealtimeEvent::InputTranscriptDelta(
+                RealtimeTranscriptDelta {
+                    delta: "trailing words".to_string(),
+                },
+            ))
+            .await;
+
+        assert_eq!(
+            events.take_transcript_tail().await,
+            vec![RealtimeTranscriptEntry {
+                role: "user".to_string(),
+                text: "trailing words".to_string(),
+            }]
+        );
+        assert_eq!(events.take_transcript_tail().await, Vec::new());
+    }
+
+    #[tokio::test]
+    async fn committed_handoff_is_not_returned_as_tail() {
+        let events = transcript_events();
+        events
+            .update_active_transcript(&mut RealtimeEvent::InputTranscriptDelta(
+                RealtimeTranscriptDelta {
+                    delta: "already handed off".to_string(),
+                },
+            ))
+            .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::new(),
+        });
+        events.update_active_transcript(&mut handoff).await;
+        events.commit_transcript_handoff("handoff_1").await;
+
+        assert_eq!(events.take_transcript_tail().await, Vec::new());
+    }
+
+    #[tokio::test]
+    async fn uncommitted_handoff_is_still_returned_as_tail() {
+        let events = transcript_events();
+        events
+            .update_active_transcript(&mut RealtimeEvent::InputTranscriptDelta(
+                RealtimeTranscriptDelta {
+                    delta: "queued handoff".to_string(),
+                },
+            ))
+            .await;
+        events
+            .update_active_transcript(&mut RealtimeEvent::HandoffRequested(
+                RealtimeHandoffRequested {
+                    handoff_id: "handoff_1".to_string(),
+                    item_id: "item_1".to_string(),
+                    input_transcript: "queued handoff".to_string(),
+                    active_transcript: Vec::new(),
+                },
+            ))
+            .await;
+
+        assert_eq!(
+            events.take_transcript_tail().await,
+            vec![RealtimeTranscriptEntry {
+                role: "user".to_string(),
+                text: "queued handoff".to_string(),
+            }]
+        );
+    }
+
     #[test]
     fn parse_session_updated_event() {
         let payload = json!({

codex-rs/core/src/context/mod.rs

@@ -25,6 +25,7 @@ mod plugin_instructions;
 mod realtime_end_instructions;
 mod realtime_start_instructions;
 mod realtime_start_with_instructions;
+mod realtime_transcript_tail;
 mod recommended_plugins_instructions;
 mod rollout_budget;
 mod subagent_notification;
@@ -68,6 +69,7 @@ pub(crate) use plugin_instructions::PluginInstructions;
 pub(crate) use realtime_end_instructions::RealtimeEndInstructions;
 pub(crate) use realtime_start_instructions::RealtimeStartInstructions;
 pub(crate) use realtime_start_with_instructions::RealtimeStartWithInstructions;
+pub(crate) use realtime_transcript_tail::RealtimeTranscriptTail;
 pub(crate) use recommended_plugins_instructions::RecommendedPluginsInstructions;
 pub(crate) use rollout_budget::RolloutBudgetContext;
 pub(crate) use subagent_notification::SubagentNotification;

codex-rs/core/src/context/realtime_transcript_tail.rs

@@ -0,0 +1,59 @@
+use super::ContextualUserFragment;
+use crate::realtime_context::truncate_realtime_text_to_token_budget;
+
+const REALTIME_TRANSCRIPT_TAIL_OPEN_TAG: &str = "<realtime_delegation>";
+const REALTIME_TRANSCRIPT_TAIL_CLOSE_TAG: &str = "</realtime_delegation>";
+const REALTIME_TRANSCRIPT_TAIL_TOKEN_BUDGET: usize = 8_000;
+const REALTIME_TRANSCRIPT_TAIL_INSTRUCTIONS: &str = "The user just ended their realtime session. Here is the remaining handoff/transcript tail. You probably do not have to do anything; acknowledge the handoff unless the transcript itself asks for something.";
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct RealtimeTranscriptTail {
+    transcript_delta: String,
+}
+
+impl RealtimeTranscriptTail {
+    pub(crate) fn new(transcript_delta: impl AsRef<str>) -> Self {
+        Self {
+            transcript_delta: truncate_realtime_text_to_token_budget(
+                transcript_delta.as_ref(),
+                REALTIME_TRANSCRIPT_TAIL_TOKEN_BUDGET,
+            ),
+        }
+    }
+}
+
+impl ContextualUserFragment for RealtimeTranscriptTail {
+    fn role(&self) -> &'static str {
+        "developer"
+    }
+
+    fn markers(&self) -> (&'static str, &'static str) {
+        Self::type_markers()
+    }
+
+    fn type_markers() -> (&'static str, &'static str) {
+        (
+            REALTIME_TRANSCRIPT_TAIL_OPEN_TAG,
+            REALTIME_TRANSCRIPT_TAIL_CLOSE_TAG,
+        )
+    }
+
+    fn body(&self) -> String {
+        let instructions = escape_xml_text(REALTIME_TRANSCRIPT_TAIL_INSTRUCTIONS);
+        let transcript_delta = escape_xml_text(&self.transcript_delta);
+        format!(
+            "\n  <input>{instructions}</input>\n  <transcript_delta>{transcript_delta}</transcript_delta>\n"
+        )
+    }
+}
+
+fn escape_xml_text(input: &str) -> String {
+    input
+        .replace('&', "&amp;")
+        .replace('<', "&lt;")
+        .replace('>', "&gt;")
+}
+
+#[cfg(test)]
+#[path = "realtime_transcript_tail_tests.rs"]
+mod tests;

codex-rs/core/src/context/realtime_transcript_tail_tests.rs

@@ -0,0 +1,10 @@
+use super::*;
+use pretty_assertions::assert_eq;
+
+#[test]
+fn renders_tail_as_hidden_realtime_delegation_context() {
+    assert_eq!(
+        RealtimeTranscriptTail::new("user: ship <this> & that").render(),
+        "<realtime_delegation>\n  <input>The user just ended their realtime session. Here is the remaining handoff/transcript tail. You probably do not have to do anything; acknowledge the handoff unless the transcript itself asks for something.</input>\n  <transcript_delta>user: ship &lt;this&gt; &amp; that</transcript_delta>\n</realtime_delegation>"
+    );
+}

codex-rs/core/src/lib.rs

@@ -12,6 +12,7 @@ mod client_common;
 mod realtime_context;
 mod realtime_conversation;
 mod realtime_prompt;
+mod realtime_transcript;
 mod responses_metadata;
 mod responses_retry;
 pub(crate) mod session;

codex-rs/core/src/realtime_conversation.rs

@@ -3,6 +3,8 @@ use crate::client::add_originator_header;
 use crate::realtime_context::build_realtime_startup_context;
 use crate::realtime_context::truncate_realtime_text_to_token_budget;
 use crate::realtime_prompt::prepare_realtime_backend_prompt;
+use crate::realtime_transcript::format_realtime_transcript;
+use crate::realtime_transcript::persist_realtime_transcript_tail;
 use crate::session::session::Session;
 use anyhow::Context;
 use async_channel::Receiver;
@@ -48,6 +50,7 @@ use codex_protocol::protocol::RealtimeConversationSdpEvent;
 use codex_protocol::protocol::RealtimeConversationStartedEvent;
 use codex_protocol::protocol::RealtimeHandoffRequested;
 use codex_protocol::protocol::RealtimeOutputModality;
+use codex_protocol::protocol::RealtimeTranscriptEntry;
 use codex_protocol::protocol::RealtimeVoice;
 use codex_protocol::protocol::RealtimeVoicesList;
 use http::HeaderMap;
@@ -58,6 +61,7 @@ use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering;
 use tokio::sync::Mutex;
+use tokio::sync::Semaphore;
 use tokio::task::JoinHandle;
 use tracing::debug;
 use tracing::error;
@@ -242,6 +246,8 @@ struct ConversationState {
     input_task: JoinHandle<()>,
     fanout_task: Option<JoinHandle<()>>,
     realtime_active: Arc<AtomicBool>,
+    transcript_events: Arc<Mutex<Option<RealtimeWebsocketEvents>>>,
+    transcript_flush_lock: Arc<Semaphore>,
 }
 
 struct RealtimeStart {
@@ -261,6 +267,8 @@ struct RealtimeStartOutput {
     realtime_active: Arc<AtomicBool>,
     events_rx: Receiver<RealtimeEvent>,
     sdp: Option<String>,
+    transcript_events: Arc<Mutex<Option<RealtimeWebsocketEvents>>>,
+    transcript_flush_lock: Arc<Semaphore>,
 }
 
 #[allow(dead_code)]
@@ -329,6 +337,8 @@ impl RealtimeConversationManager {
             async_channel::bounded::<RealtimeEvent>(OUTPUT_EVENTS_QUEUE_CAPACITY);
 
         let realtime_active = Arc::new(AtomicBool::new(true));
+        let transcript_events = Arc::new(Mutex::new(None));
+        let transcript_flush_lock = Arc::new(Semaphore::new(1));
         let handoff = RealtimeHandoffState::new(
             handoff_output_tx,
             client_managed_handoffs,
@@ -364,6 +374,7 @@ impl RealtimeConversationManager {
                 session_kind,
                 event_parser,
                 realtime_active: Arc::clone(&realtime_active),
+                transcript_events: Arc::clone(&transcript_events),
             });
             (task, Some(call.sdp))
         } else {
@@ -375,9 +386,11 @@ impl RealtimeConversationManager {
                 )
                 .await
                 .map_err(map_api_error)?;
+            let connection_events = connection.events();
+            *transcript_events.lock().await = Some(connection_events.clone());
             let task = spawn_realtime_input_task(RealtimeInputTask {
                 writer: connection.writer(),
-                events: connection.events(),
+                events: connection_events,
                 text_rx: input_channels.text_rx,
                 handoff_output_rx: input_channels.handoff_output_rx,
                 audio_rx: input_channels.audio_rx,
@@ -398,11 +411,15 @@ impl RealtimeConversationManager {
             input_task: task,
             fanout_task: None,
             realtime_active: Arc::clone(&realtime_active),
+            transcript_events: Arc::clone(&transcript_events),
+            transcript_flush_lock: Arc::clone(&transcript_flush_lock),
         });
         Ok(RealtimeStartOutput {
             realtime_active,
             events_rx,
             sdp,
+            transcript_events,
+            transcript_flush_lock,
         })
     }
 
@@ -427,7 +444,10 @@ impl RealtimeConversationManager {
         }
     }
 
-    pub(crate) async fn finish_if_active(&self, realtime_active: &Arc<AtomicBool>) {
+    pub(crate) async fn finish_if_active(
+        &self,
+        realtime_active: &Arc<AtomicBool>,
+    ) -> Vec<RealtimeTranscriptEntry> {
         let state = {
             let mut guard = self.state.lock().await;
             match guard.as_ref() {
@@ -436,8 +456,9 @@ impl RealtimeConversationManager {
             }
         };
 
-        if let Some(state) = state {
-            stop_conversation_state(state, RealtimeFanoutTaskStop::Detach).await;
+        match state {
+            Some(state) => stop_conversation_state(state, RealtimeFanoutTaskStop::Detach).await,
+            None => Vec::new(),
         }
     }
 
@@ -645,23 +666,30 @@ impl RealtimeConversationManager {
         }
     }
 
-    pub(crate) async fn shutdown(&self) -> CodexResult<()> {
+    pub(crate) async fn shutdown(&self) -> Vec<RealtimeTranscriptEntry> {
         let state = {
             let mut guard = self.state.lock().await;
             guard.take()
         };
 
-        if let Some(state) = state {
-            stop_conversation_state(state, RealtimeFanoutTaskStop::Abort).await;
+        match state {
+            Some(state) => stop_conversation_state(state, RealtimeFanoutTaskStop::Abort).await,
+            None => Vec::new(),
         }
-        Ok(())
     }
 }
 
 async fn stop_conversation_state(
     mut state: ConversationState,
     fanout_task_stop: RealtimeFanoutTaskStop,
-) {
+) -> Vec<RealtimeTranscriptEntry> {
+    let _flush_permit = match state.transcript_flush_lock.acquire().await {
+        Ok(permit) => permit,
+        Err(err) => {
+            warn!("failed to acquire transcript flush permit: {err}");
+            return Vec::new();
+        }
+    };
     state.realtime_active.store(false, Ordering::Relaxed);
     state.input_task.abort();
     let _ = state.input_task.await;
@@ -675,6 +703,12 @@ async fn stop_conversation_state(
             RealtimeFanoutTaskStop::Detach => {}
         }
     }
+
+    let transcript_events = state.transcript_events.lock().await.clone();
+    match transcript_events {
+        Some(transcript_events) => transcript_events.take_transcript_tail().await,
+        None => Vec::new(),
+    }
 }
 
 pub(crate) async fn handle_start(
@@ -992,6 +1026,7 @@ async fn handle_start_inner(
         model_client: sess.services.model_client.clone(),
         sdp,
     };
+    shutdown_realtime_conversation(sess).await;
     let start_output = sess.conversation.start(start).await?;
 
     info!("realtime conversation started");
@@ -1009,6 +1044,8 @@ async fn handle_start_inner(
         realtime_active,
         events_rx,
         sdp,
+        transcript_events,
+        transcript_flush_lock,
     } = start_output;
     if let Some(sdp) = sdp {
         sess.send_event_raw(Event {
@@ -1043,16 +1080,26 @@ async fn handle_start_inner(
             if let RealtimeEvent::Error(_) = &event {
                 end = RealtimeConversationEnd::Error;
             }
-            let maybe_routed_text = match &event {
-                RealtimeEvent::HandoffRequested(handoff) => {
-                    realtime_delegation_from_handoff(handoff)
+            if let RealtimeEvent::HandoffRequested(handoff) = &event {
+                let _flush_permit = match transcript_flush_lock.acquire().await {
+                    Ok(permit) => permit,
+                    Err(err) => {
+                        warn!("failed to acquire transcript flush permit: {err}");
+                        continue;
+                    }
+                };
+                if fanout_realtime_active.load(Ordering::Relaxed) {
+                    if let Some(text) = realtime_delegation_from_handoff(handoff) {
+                        debug!(text = %text, "[realtime-text] realtime conversation text output");
+                        sess_clone.route_realtime_text_input(text).await;
+                    }
+                    let transcript_events = transcript_events.lock().await.clone();
+                    if let Some(transcript_events) = transcript_events {
+                        transcript_events
+                            .commit_transcript_handoff(&handoff.handoff_id)
+                            .await;
+                    }
                 }
-                _ => 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 !fanout_realtime_active.load(Ordering::Relaxed) {
                 break;
@@ -1072,10 +1119,11 @@ async fn handle_start_inner(
                 }
                 RealtimeConversationEnd::Requested | RealtimeConversationEnd::Error => {}
             }
-            sess_clone
+            let transcript_tail = sess_clone
                 .conversation
                 .finish_if_active(&fanout_realtime_active)
                 .await;
+            persist_realtime_transcript_tail(&sess_clone, transcript_tail).await;
             send_realtime_conversation_closed(&sess_clone, sub_id, end).await;
         }
     });
@@ -1103,13 +1151,7 @@ pub(crate) async fn handle_audio(
 }
 
 fn realtime_transcript_delta_from_handoff(handoff: &RealtimeHandoffRequested) -> Option<String> {
-    let active_transcript = handoff
-        .active_transcript
-        .iter()
-        .map(|entry| format!("{role}: {text}", role = entry.role, text = entry.text))
-        .collect::<Vec<_>>()
-        .join("\n");
-    (!active_transcript.is_empty()).then_some(active_transcript)
+    format_realtime_transcript(&handoff.active_transcript)
 }
 
 fn realtime_text_from_handoff_request(handoff: &RealtimeHandoffRequested) -> Option<String> {
@@ -1254,6 +1296,7 @@ struct RealtimeWebrtcSidebandInputTask {
     session_kind: RealtimeSessionKind,
     event_parser: RealtimeEventParser,
     realtime_active: Arc<AtomicBool>,
+    transcript_events: Arc<Mutex<Option<RealtimeWebsocketEvents>>>,
 }
 
 fn spawn_webrtc_sideband_input_task(input: RealtimeWebrtcSidebandInputTask) -> JoinHandle<()> {
@@ -1268,6 +1311,7 @@ fn spawn_webrtc_sideband_input_task(input: RealtimeWebrtcSidebandInputTask) -> J
         session_kind,
         event_parser,
         realtime_active,
+        transcript_events,
     } = input;
 
     tokio::spawn(async move {
@@ -1301,9 +1345,11 @@ fn spawn_webrtc_sideband_input_task(input: RealtimeWebrtcSidebandInputTask) -> J
             return;
         }
 
+        let connection_events = connection.events();
+        *transcript_events.lock().await = Some(connection_events.clone());
         run_realtime_input_task(RealtimeInputTask {
             writer: connection.writer(),
-            events: connection.events(),
+            events: connection_events,
             text_rx: input_channels.text_rx,
             handoff_output_rx: input_channels.handoff_output_rx,
             audio_rx: input_channels.audio_rx,
@@ -1769,10 +1815,15 @@ async fn end_realtime_conversation(
     sub_id: String,
     end: RealtimeConversationEnd,
 ) {
-    let _ = sess.conversation.shutdown().await;
+    shutdown_realtime_conversation(sess).await;
     send_realtime_conversation_closed(sess, sub_id, end).await;
 }
 
+pub(crate) async fn shutdown_realtime_conversation(sess: &Arc<Session>) {
+    let transcript_tail = sess.conversation.shutdown().await;
+    persist_realtime_transcript_tail(sess, transcript_tail).await;
+}
+
 async fn send_realtime_conversation_closed(
     sess: &Arc<Session>,
     sub_id: String,

codex-rs/core/src/realtime_transcript.rs

@@ -0,0 +1,31 @@
+use crate::context::ContextualUserFragment;
+use crate::context::RealtimeTranscriptTail;
+use crate::session::session::Session;
+use codex_protocol::protocol::RealtimeTranscriptEntry;
+use std::sync::Arc;
+use tracing::warn;
+
+pub(crate) fn format_realtime_transcript(entries: &[RealtimeTranscriptEntry]) -> Option<String> {
+    let transcript = entries
+        .iter()
+        .map(|entry| format!("{role}: {text}", role = entry.role, text = entry.text))
+        .collect::<Vec<_>>()
+        .join("\n");
+    (!transcript.is_empty()).then_some(transcript)
+}
+
+pub(crate) async fn persist_realtime_transcript_tail(
+    sess: &Arc<Session>,
+    transcript_tail: Vec<RealtimeTranscriptEntry>,
+) {
+    let Some(transcript_delta) = format_realtime_transcript(&transcript_tail) else {
+        return;
+    };
+    let item = ContextualUserFragment::into(RealtimeTranscriptTail::new(transcript_delta));
+    let turn_context = sess.new_default_turn().await;
+    sess.record_conversation_items(&turn_context, std::slice::from_ref(&item))
+        .await;
+    if let Err(err) = sess.flush_rollout().await {
+        warn!("failed to flush final realtime transcript tail: {err}");
+    }
+}

codex-rs/core/src/session/handlers.rs

@@ -3,6 +3,7 @@ use crate::realtime_conversation::handle_close as handle_realtime_conversation_c
 use crate::realtime_conversation::handle_speech as handle_realtime_conversation_speech;
 use crate::realtime_conversation::handle_start as handle_realtime_conversation_start;
 use crate::realtime_conversation::handle_text as handle_realtime_conversation_text;
+use crate::realtime_conversation::shutdown_realtime_conversation;
 use async_channel::Receiver;
 use codex_otel::set_parent_from_w3c_trace_context;
 use codex_protocol::protocol::Submission;
@@ -592,7 +593,7 @@ async fn shutdown_session_runtime(sess: &Arc<Session>) {
         startup_prewarm.abort().await;
     }
     sess.abort_all_tasks(TurnAbortReason::Interrupted).await;
-    let _ = sess.conversation.shutdown().await;
+    shutdown_realtime_conversation(sess).await;
     sess.services
         .unified_exec_manager
         .terminate_all_processes()

codex-rs/core/tests/suite/mod.rs

@@ -89,6 +89,7 @@ mod prompt_caching;
 mod prompt_debug_tests;
 mod quota_exceeded;
 mod realtime_conversation;
+mod realtime_transcript_tail;
 mod remote_env;
 mod remote_models;
 mod request_compression;

codex-rs/core/tests/suite/realtime_conversation.rs

@@ -57,6 +57,8 @@ use wiremock::ResponseTemplate;
 use wiremock::matchers::method;
 use wiremock::matchers::path_regex;
 
+use super::realtime_transcript_tail::persisted_realtime_transcript_tails;
+
 const STARTUP_CONTEXT_HEADER: &str = "Startup context from Codex.";
 const STARTUP_CONTEXT_OPEN_TAG: &str = "<startup_context>";
 const STARTUP_CONTEXT_CLOSE_TAG: &str = "</startup_context>";
@@ -3196,7 +3198,10 @@ async fn inbound_handoff_request_starts_turn() -> Result<()> {
     let user_texts = request.message_input_texts("user");
     assert!(user_texts.iter().any(|text| text
         == "<realtime_delegation>\n  <input>text from realtime</input>\n  <transcript_delta>user: text from realtime</transcript_delta>\n</realtime_delegation>"));
-
+    assert_eq!(
+        persisted_realtime_transcript_tails(&test),
+        Vec::<String>::new()
+    );
     realtime_server.shutdown().await;
     Ok(())
 }

codex-rs/core/tests/suite/realtime_transcript_tail.rs

@@ -0,0 +1,217 @@
+use anyhow::Result;
+use codex_config::config_toml::RealtimeWsVersion;
+use codex_protocol::models::ContentItem;
+use codex_protocol::models::ResponseItem;
+use codex_protocol::protocol::ConversationStartParams;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::Op;
+use codex_protocol::protocol::RealtimeConversationRealtimeEvent;
+use codex_protocol::protocol::RealtimeEvent;
+use codex_protocol::protocol::RealtimeOutputModality;
+use codex_protocol::protocol::RolloutItem;
+use codex_protocol::protocol::RolloutLine;
+use core_test_support::responses::WebSocketConnectionConfig;
+use core_test_support::responses::start_mock_server;
+use core_test_support::responses::start_websocket_server;
+use core_test_support::responses::start_websocket_server_with_headers;
+use core_test_support::skip_if_no_network;
+use core_test_support::test_codex::TestCodex;
+use core_test_support::test_codex::test_codex;
+use core_test_support::wait_for_event;
+use core_test_support::wait_for_event_match;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+use std::fs;
+
+fn realtime_start_params() -> ConversationStartParams {
+    ConversationStartParams {
+        client_managed_handoffs: 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,
+        prompt: Some(Some("backend prompt".to_string())),
+        realtime_session_id: None,
+        transport: None,
+        version: None,
+        voice: None,
+    }
+}
+
+pub(super) fn persisted_realtime_transcript_tails(test: &TestCodex) -> Vec<String> {
+    let rollout_path = test.codex.rollout_path().expect("rollout path");
+    let Ok(rollout) = fs::read_to_string(rollout_path) else {
+        return Vec::new();
+    };
+    rollout
+        .lines()
+        .filter_map(|line| serde_json::from_str::<RolloutLine>(line).ok())
+        .filter_map(|line| match line.item {
+            RolloutItem::ResponseItem(ResponseItem::Message { role, content, .. })
+                if role == "developer" =>
+            {
+                Some(content)
+            }
+            _ => None,
+        })
+        .flatten()
+        .filter_map(|content| match content {
+            ContentItem::InputText { text }
+                if text.starts_with("<realtime_delegation>")
+                    && text.contains("The user just ended their realtime session") =>
+            {
+                Some(text)
+            }
+            _ => None,
+        })
+        .collect()
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn conversation_close_without_transcript_does_not_persist_tail() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let api_server = start_mock_server().await;
+    let realtime_server = start_websocket_server_with_headers(vec![WebSocketConnectionConfig {
+        requests: vec![vec![json!({
+            "type": "session.updated",
+            "session": { "id": "sess_no_tail", "instructions": "backend prompt" }
+        })]],
+        response_headers: Vec::new(),
+        accept_delay: None,
+        close_after_requests: false,
+    }])
+    .await;
+    let realtime_base_url = realtime_server.uri().to_string();
+    let mut builder = test_codex().with_config(move |config| {
+        config.experimental_realtime_ws_base_url = Some(realtime_base_url);
+        config.realtime.version = RealtimeWsVersion::V1;
+    });
+    let test = builder.build(&api_server).await?;
+
+    test.codex
+        .submit(Op::RealtimeConversationStart(realtime_start_params()))
+        .await?;
+    wait_for_event(&test.codex, |event| {
+        matches!(
+            event,
+            EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
+                payload: RealtimeEvent::SessionUpdated { .. }
+            })
+        )
+    })
+    .await;
+
+    test.codex.submit(Op::RealtimeConversationClose).await?;
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::RealtimeConversationClosed(_))
+    })
+    .await;
+
+    assert_eq!(
+        persisted_realtime_transcript_tails(&test),
+        Vec::<String>::new()
+    );
+    realtime_server.shutdown().await;
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn repeated_conversation_close_persists_trailing_transcript_once() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let api_server = start_mock_server().await;
+    let realtime_server = start_websocket_server_with_headers(vec![WebSocketConnectionConfig {
+        requests: vec![vec![
+            json!({
+                "type": "session.updated",
+                "session": { "id": "sess_tail", "instructions": "backend prompt" }
+            }),
+            json!({
+                "type": "conversation.input_transcript.delta",
+                "delta": "trailing words"
+            }),
+        ]],
+        response_headers: Vec::new(),
+        accept_delay: None,
+        close_after_requests: false,
+    }])
+    .await;
+    let realtime_base_url = realtime_server.uri().to_string();
+    let mut builder = test_codex().with_config(move |config| {
+        config.experimental_realtime_ws_base_url = Some(realtime_base_url);
+        config.realtime.version = RealtimeWsVersion::V1;
+    });
+    let test = builder.build(&api_server).await?;
+
+    test.codex
+        .submit(Op::RealtimeConversationStart(realtime_start_params()))
+        .await?;
+    wait_for_event(&test.codex, |event| {
+        matches!(
+            event,
+            EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
+                payload: RealtimeEvent::InputTranscriptDelta(_)
+            })
+        )
+    })
+    .await;
+
+    for _ in 0..2 {
+        test.codex.submit(Op::RealtimeConversationClose).await?;
+        wait_for_event(&test.codex, |event| {
+            matches!(event, EventMsg::RealtimeConversationClosed(_))
+        })
+        .await;
+    }
+
+    assert_eq!(
+        persisted_realtime_transcript_tails(&test),
+        vec![
+            "<realtime_delegation>\n  <input>The user just ended their realtime session. Here is the remaining handoff/transcript tail. You probably do not have to do anything; acknowledge the handoff unless the transcript itself asks for something.</input>\n  <transcript_delta>user: trailing words</transcript_delta>\n</realtime_delegation>"
+                .to_string()
+        ]
+    );
+    realtime_server.shutdown().await;
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn conversation_transport_close_persists_trailing_transcript() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let api_server = start_mock_server().await;
+    let realtime_server = start_websocket_server(vec![vec![vec![
+        json!({
+            "type": "session.updated",
+            "session": { "id": "sess_transport_tail", "instructions": "backend prompt" }
+        }),
+        json!({
+            "type": "conversation.input_transcript.delta",
+            "delta": "transport tail"
+        }),
+    ]]])
+    .await;
+    let realtime_base_url = realtime_server.uri().to_string();
+    let mut builder = test_codex().with_config(move |config| {
+        config.experimental_realtime_ws_base_url = Some(realtime_base_url);
+        config.realtime.version = RealtimeWsVersion::V1;
+    });
+    let test = builder.build(&api_server).await?;
+
+    test.codex
+        .submit(Op::RealtimeConversationStart(realtime_start_params()))
+        .await?;
+    let closed = wait_for_event_match(&test.codex, |event| match event {
+        EventMsg::RealtimeConversationClosed(closed) => Some(closed.clone()),
+        _ => None,
+    })
+    .await;
+
+    assert_eq!(closed.reason.as_deref(), Some("transport_closed"));
+    assert_eq!(persisted_realtime_transcript_tails(&test).len(), 1);
+    realtime_server.shutdown().await;
+    Ok(())
+}