From 22f413bdbf27f25fed08fe98e6ac185fad84722e Mon Sep 17 00:00:00 2001 From: guinness-oai Date: Fri, 17 Jul 2026 03:34:19 +0000 Subject: [PATCH] Tag realtime transcript tail flush delegations (#33855) ## What changed - Add `transcript_tail_flush` 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 | 3 + .../core/src/context/realtime_delegation.rs | 67 +++++++++++++++++++ codex-rs/core/src/realtime_conversation.rs | 28 +++----- .../core/src/realtime_conversation_tests.rs | 19 +++++- .../core/tests/suite/realtime_conversation.rs | 4 +- 5 files changed, 99 insertions(+), 22 deletions(-) create mode 100644 codex-rs/core/src/context/realtime_delegation.rs diff --git a/codex-rs/core/src/context/mod.rs b/codex-rs/core/src/context/mod.rs index 4d8f1817adcb..f14f1061ba5e 100644 --- a/codex-rs/core/src/context/mod.rs +++ b/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; diff --git a/codex-rs/core/src/context/realtime_delegation.rs b/codex-rs/core/src/context/realtime_delegation.rs new file mode 100644 index 000000000000..98274df87c71 --- /dev/null +++ b/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) { + ("", "") + } + + fn body(&self) -> String { + let input = escape_xml_text(self.input); + let source = match self.source { + RealtimeDelegationSource::Handoff => "", + RealtimeDelegationSource::TranscriptTailFlush => { + " transcript_tail_flush\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}\n {transcript_delta}\n" + ); + } + + format!("\n{source} {input}\n") + } +} + +fn escape_xml_text(input: &str) -> String { + input + .replace('&', "&") + .replace('<', "<") + .replace('>', ">") +} diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index 33106868ca1c..5d3550c3b59c 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/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!( - "\n {input}\n {transcript_delta}\n" - ); - } - - format!("\n {input}\n") -} - -fn escape_xml_text(input: &str) -> String { - input - .replace('&', "&") - .replace('<', "<") - .replace('>', ">") +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 { @@ -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; } diff --git a/codex-rs/core/src/realtime_conversation_tests.rs b/codex-rs/core/src/realtime_conversation_tests.rs index 412ce1f8ab6a..15bd4d545699 100644 --- a/codex-rs/core/src/realtime_conversation_tests.rs +++ b/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, + ), "\n hello\n" ); } @@ -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 ")), + wrap_realtime_delegation_input( + "use a < b && c > d", + Some("saw "), + RealtimeDelegationSource::Handoff, + ), "\n use a < b && c > d\n saw <that>\n" ); } @@ -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, + ), "\n use a < b && c > d\n" ); } diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index f4262e8e4021..91416a69d069 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/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 - == "\n 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.\n user: transport tail\n")); + == "\n transcript_tail_flush\n 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.\n user: transport tail\n")); } 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 == "\n already handed off\n user: already handed off\n")); assert!(requests[1].message_input_texts("user").iter().any(|text| text - == "\n 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.\n assistant: remaining answer\nuser: remaining question\n")); + == "\n transcript_tail_flush\n 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.\n assistant: remaining answer\nuser: remaining question\n")); realtime_server.shutdown().await; Ok(())