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 499635579f..050e15f932 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -219,6 +219,8 @@ pub struct RealtimeWebsocketEvents { struct ActiveTranscriptState { entries: Vec, last_handoff_entry_count: usize, + last_requested_handoff_entry_count: usize, + handoff_entry_counts: HashMap, 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 { + 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, 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!({ diff --git a/codex-rs/core/src/context/mod.rs b/codex-rs/core/src/context/mod.rs index 4ba6acf029..bd8f66f790 100644 --- a/codex-rs/core/src/context/mod.rs +++ b/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; diff --git a/codex-rs/core/src/context/realtime_transcript_tail.rs b/codex-rs/core/src/context/realtime_transcript_tail.rs new file mode 100644 index 0000000000..644b96b24e --- /dev/null +++ b/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 = ""; +const REALTIME_TRANSCRIPT_TAIL_CLOSE_TAG: &str = ""; +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) -> 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 {instructions}\n {transcript_delta}\n" + ) + } +} + +fn escape_xml_text(input: &str) -> String { + input + .replace('&', "&") + .replace('<', "<") + .replace('>', ">") +} + +#[cfg(test)] +#[path = "realtime_transcript_tail_tests.rs"] +mod tests; diff --git a/codex-rs/core/src/context/realtime_transcript_tail_tests.rs b/codex-rs/core/src/context/realtime_transcript_tail_tests.rs new file mode 100644 index 0000000000..98ec7b3715 --- /dev/null +++ b/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 & that").render(), + "\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: ship <this> & that\n" + ); +} diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index da91954063..1460271a37 100644 --- a/codex-rs/core/src/lib.rs +++ b/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; diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index 0c81133be1..650557390f 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/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>, realtime_active: Arc, + transcript_events: Arc>>, + transcript_flush_lock: Arc, } struct RealtimeStart { @@ -261,6 +267,8 @@ struct RealtimeStartOutput { realtime_active: Arc, events_rx: Receiver, sdp: Option, + transcript_events: Arc>>, + transcript_flush_lock: Arc, } #[allow(dead_code)] @@ -329,6 +337,8 @@ impl RealtimeConversationManager { async_channel::bounded::(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) { + pub(crate) async fn finish_if_active( + &self, + realtime_active: &Arc, + ) -> Vec { 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 { 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 { + 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 { - let active_transcript = handoff - .active_transcript - .iter() - .map(|entry| format!("{role}: {text}", role = entry.role, text = entry.text)) - .collect::>() - .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 { @@ -1254,6 +1296,7 @@ struct RealtimeWebrtcSidebandInputTask { session_kind: RealtimeSessionKind, event_parser: RealtimeEventParser, realtime_active: Arc, + transcript_events: Arc>>, } 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) { + let transcript_tail = sess.conversation.shutdown().await; + persist_realtime_transcript_tail(sess, transcript_tail).await; +} + async fn send_realtime_conversation_closed( sess: &Arc, sub_id: String, diff --git a/codex-rs/core/src/realtime_transcript.rs b/codex-rs/core/src/realtime_transcript.rs new file mode 100644 index 0000000000..f8e8083af8 --- /dev/null +++ b/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 { + let transcript = entries + .iter() + .map(|entry| format!("{role}: {text}", role = entry.role, text = entry.text)) + .collect::>() + .join("\n"); + (!transcript.is_empty()).then_some(transcript) +} + +pub(crate) async fn persist_realtime_transcript_tail( + sess: &Arc, + transcript_tail: Vec, +) { + 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}"); + } +} diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 39e91d6cf9..5f065e0fc8 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/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) { 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() diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index 3d4a191f12..ff33b77148 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/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; diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index 1a95223bff..82229808fc 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/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 = ""; const STARTUP_CONTEXT_CLOSE_TAG: &str = ""; @@ -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 == "\n text from realtime\n user: text from realtime\n")); - + assert_eq!( + persisted_realtime_transcript_tails(&test), + Vec::::new() + ); realtime_server.shutdown().await; Ok(()) } diff --git a/codex-rs/core/tests/suite/realtime_transcript_tail.rs b/codex-rs/core/tests/suite/realtime_transcript_tail.rs new file mode 100644 index 0000000000..50b77b9b6f --- /dev/null +++ b/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 { + 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::(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("") + && 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::::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![ + "\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: trailing words\n" + .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(()) +}