Tag realtime transcript tail flush delegations (#33855)

## What changed

- Add `<source>transcript_tail_flush</source>` to realtime delegation payloads
  emitted when a session flushes its remaining transcript tail.
- Keep explicit realtime handoff payloads unchanged.
- Render both delegation variants through a shared contextual user fragment.

## Testing

- Update realtime conversation tests to cover tagged transcript-tail flushes and
  preserve the existing handoff format and XML escaping behavior.

GitOrigin-RevId: 33f70d16cf45b3af8068565c4cf2b1785be3f1d9

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

@@ -23,6 +23,7 @@ mod network_rule_saved;
 mod permissions_instructions;
 mod personality_spec_instructions;
 mod plugin_instructions;
+mod realtime_delegation;
 mod realtime_end_instructions;
 mod realtime_start_instructions;
 mod realtime_start_with_instructions;
@@ -67,6 +68,8 @@ pub use permissions_instructions::ApprovalPromptContext;
 pub use permissions_instructions::PermissionsInstructions;
 pub(crate) use personality_spec_instructions::PersonalitySpecInstructions;
 pub(crate) use plugin_instructions::PluginInstructions;
+pub(crate) use realtime_delegation::RealtimeDelegation;
+pub(crate) use realtime_delegation::RealtimeDelegationSource;
 pub(crate) use realtime_end_instructions::RealtimeEndInstructions;
 pub(crate) use realtime_start_instructions::RealtimeStartInstructions;
 pub(crate) use realtime_start_with_instructions::RealtimeStartWithInstructions;

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

@@ -0,0 +1,67 @@
+use super::ContextualUserFragment;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum RealtimeDelegationSource {
+    Handoff,
+    TranscriptTailFlush,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct RealtimeDelegation<'a> {
+    input: &'a str,
+    transcript_delta: Option<&'a str>,
+    source: RealtimeDelegationSource,
+}
+
+impl<'a> RealtimeDelegation<'a> {
+    pub(crate) fn new(
+        input: &'a str,
+        transcript_delta: Option<&'a str>,
+        source: RealtimeDelegationSource,
+    ) -> Self {
+        Self {
+            input,
+            transcript_delta,
+            source,
+        }
+    }
+}
+
+impl ContextualUserFragment for RealtimeDelegation<'_> {
+    fn role(&self) -> &'static str {
+        "user"
+    }
+
+    fn markers(&self) -> (&'static str, &'static str) {
+        Self::type_markers()
+    }
+
+    fn type_markers() -> (&'static str, &'static str) {
+        ("<realtime_delegation>", "</realtime_delegation>")
+    }
+
+    fn body(&self) -> String {
+        let input = escape_xml_text(self.input);
+        let source = match self.source {
+            RealtimeDelegationSource::Handoff => "",
+            RealtimeDelegationSource::TranscriptTailFlush => {
+                "  <source>transcript_tail_flush</source>\n"
+            }
+        };
+        if let Some(transcript_delta) = self.transcript_delta.filter(|text| !text.is_empty()) {
+            let transcript_delta = escape_xml_text(transcript_delta);
+            return format!(
+                "\n{source}  <input>{input}</input>\n  <transcript_delta>{transcript_delta}</transcript_delta>\n"
+            );
+        }
+
+        format!("\n{source}  <input>{input}</input>\n")
+    }
+}
+
+fn escape_xml_text(input: &str) -> String {
+    input
+        .replace('&', "&amp;")
+        .replace('<', "&lt;")
+        .replace('>', "&gt;")
+}

codex-rs/core/src/realtime_conversation.rs

@@ -1,4 +1,7 @@
 use crate::client::ModelClient;
+use crate::context::ContextualUserFragment;
+use crate::context::RealtimeDelegation;
+use crate::context::RealtimeDelegationSource;
 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;
@@ -1155,26 +1158,16 @@ fn realtime_delegation_from_handoff(handoff: &RealtimeHandoffRequested) -> Optio
     Some(wrap_realtime_delegation_input(
         &input,
         realtime_transcript_delta_from_handoff(handoff).as_deref(),
+        RealtimeDelegationSource::Handoff,
     ))
 }
 
-fn wrap_realtime_delegation_input(input: &str, transcript_delta: Option<&str>) -> String {
-    let input = escape_xml_text(input);
-    if let Some(transcript_delta) = transcript_delta.filter(|text| !text.is_empty()) {
-        let transcript_delta = escape_xml_text(transcript_delta);
-        return format!(
-            "<realtime_delegation>\n  <input>{input}</input>\n  <transcript_delta>{transcript_delta}</transcript_delta>\n</realtime_delegation>"
-        );
-    }
-
-    format!("<realtime_delegation>\n  <input>{input}</input>\n</realtime_delegation>")
-}
-
-fn escape_xml_text(input: &str) -> String {
-    input
-        .replace('&', "&amp;")
-        .replace('<', "&lt;")
-        .replace('>', "&gt;")
+fn wrap_realtime_delegation_input(
+    input: &str,
+    transcript_delta: Option<&str>,
+    source: RealtimeDelegationSource,
+) -> String {
+    RealtimeDelegation::new(input, transcript_delta, source).render()
 }
 
 fn realtime_api_key(auth: Option<&CodexAuth>, provider: &ModelProviderInfo) -> CodexResult<String> {
@@ -1438,6 +1431,7 @@ async fn run_realtime_input_task(input: RealtimeInputTask) {
             .send(wrap_realtime_delegation_input(
                 REALTIME_SESSION_ENDED_HANDOFF_INSTRUCTION,
                 Some(&transcript_delta),
+                RealtimeDelegationSource::TranscriptTailFlush,
             ))
             .await;
     }

codex-rs/core/src/realtime_conversation_tests.rs

@@ -4,6 +4,7 @@ use super::realtime_delegation_from_handoff;
 use super::realtime_request_headers;
 use super::realtime_text_from_handoff_request;
 use super::wrap_realtime_delegation_input;
+use crate::context::RealtimeDelegationSource;
 use async_channel::bounded;
 use codex_api::RealtimeEventParser;
 use codex_protocol::protocol::RealtimeHandoffRequested;
@@ -104,7 +105,11 @@ fn ignores_empty_handoff_request_input_transcript() {
 #[test]
 fn wraps_realtime_delegation_input() {
     assert_eq!(
-        wrap_realtime_delegation_input("hello", /*transcript_delta*/ None),
+        wrap_realtime_delegation_input(
+            "hello",
+            /*transcript_delta*/ None,
+            RealtimeDelegationSource::Handoff,
+        ),
         "<realtime_delegation>\n  <input>hello</input>\n</realtime_delegation>"
     );
 }
@@ -112,7 +117,11 @@ fn wraps_realtime_delegation_input() {
 #[test]
 fn wraps_realtime_delegation_input_with_xml_escaping() {
     assert_eq!(
-        wrap_realtime_delegation_input("use a < b && c > d", Some("saw <that>")),
+        wrap_realtime_delegation_input(
+            "use a < b && c > d",
+            Some("saw <that>"),
+            RealtimeDelegationSource::Handoff,
+        ),
         "<realtime_delegation>\n  <input>use a &lt; b &amp;&amp; c &gt; d</input>\n  <transcript_delta>saw &lt;that&gt;</transcript_delta>\n</realtime_delegation>"
     );
 }
@@ -120,7 +129,11 @@ fn wraps_realtime_delegation_input_with_xml_escaping() {
 #[test]
 fn wraps_realtime_delegation_input_with_xml_escaping_without_transcript() {
     assert_eq!(
-        wrap_realtime_delegation_input("use a < b && c > d", /*transcript_delta*/ None),
+        wrap_realtime_delegation_input(
+            "use a < b && c > d",
+            /*transcript_delta*/ None,
+            RealtimeDelegationSource::Handoff,
+        ),
         "<realtime_delegation>\n  <input>use a &lt; b &amp;&amp; c &gt; d</input>\n</realtime_delegation>"
     );
 }

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

@@ -1370,7 +1370,7 @@ async fn assert_transport_close_tail_flush(
             tokio::time::sleep(Duration::from_millis(10)).await;
         }
         assert!(response_mock.single_request().message_input_texts("user").iter().any(|text| text
-            == "<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: transport tail</transcript_delta>\n</realtime_delegation>"));
+            == "<realtime_delegation>\n  <source>transcript_tail_flush</source>\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: transport tail</transcript_delta>\n</realtime_delegation>"));
     } else {
         tokio::time::sleep(Duration::from_millis(200)).await;
         assert!(response_mock.requests().is_empty());
@@ -3601,7 +3601,7 @@ async fn conversation_close_routes_only_remaining_transcript_tail_once() -> Resu
     assert!(requests[0].message_input_texts("user").iter().any(|text| text
         == "<realtime_delegation>\n  <input>already handed off</input>\n  <transcript_delta>user: already handed off</transcript_delta>\n</realtime_delegation>"));
     assert!(requests[1].message_input_texts("user").iter().any(|text| text
-        == "<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>assistant: remaining answer\nuser: remaining question</transcript_delta>\n</realtime_delegation>"));
+        == "<realtime_delegation>\n  <source>transcript_tail_flush</source>\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>assistant: remaining answer\nuser: remaining question</transcript_delta>\n</realtime_delegation>"));
 
     realtime_server.shutdown().await;
     Ok(())