openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-30 19:18:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Preserve realtime V3 transcripts in conversation history (#58221)

Browse Source 
## Why

Long realtime conversations could lose speech when active transcripts and handoff transcript fields were truncated.

## What changed

- Record completed V3 transcript chunks in model and client history once the active transcript exceeds 4 KiB, without triggering inference.
- Keep unfinished utterances pending to preserve conversation order, and persist completed chunks before subsequent handoffs.
- Preserve full V3 transcript text at handoff and session-end tail flush, and avoid adding handoff input already recorded in transcript history.

## Testing

Add integration coverage for long transcripts, XML escaping, recording before handoff without inference, repeated handoff input, and both session-end tail flush settings.

GitOrigin-RevId: 62218b811809c4cf5b16fb0247a335eb474f9ab9
This commit is contained in:
guinness-oai
2026-10-30 02:43:02 +00:00
committed by copyberry
parent cc2f80d515
commit c9750d7f50
10 changed files with 260 additions and 475 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/tests/suite/v2/realtime_transcript_tests.rs
+3 -7
View File
@@ -128,10 +128,9 @@ async fn websocket_v3_retains_both_speakers_from_same_frame_at_handoff() -> Resu
RealtimeTestVersion::V1,
main_loop_responses(vec![create_final_assistant_message_sse_response("Done.")?]),
realtime_sideband(vec![open_realtime_sideband_connection(vec![vec![
session_started("bounded-transcript"),
session_started("interleaved-transcript"),
json!({"type": "input_transcript.added", "start_ms": 0, "end_ms": 200, "item": {"text": old_user}}),
json!({"type": "output_transcript.added", "start_ms": 200, "end_ms": 400, "item": {"text": assistant}}),
// Input alone crosses the cap before output receives the same frame age.
json!({"type": "input_transcript.added", "start_ms": 400, "end_ms": 600, "item": {"text": newest_user}}),
json!({"type": "output_transcript.added", "start_ms": 400, "end_ms": 600, "item": {"text": assistant_suffix}}),
json!({"type": "turn.done", "turn": {"role": "assistant", "transcript": assistant_final}}),
@@ -196,9 +195,8 @@ async fn websocket_v3_retains_both_speakers_from_same_frame_at_handoff() -> Resu
"item_id": "handoff",
"input_transcript": user,
"active_transcript": [
{"role": "user", "text": format!("…{}{newest_user}", "u".repeat(3874))},
{"role": "assistant", "text": format!("…{}{assistant_suffix}", "a".repeat(3870))},
{"role": "user", "text": user},
{"role": "assistant", "text": assistant_final},
],
}),
},
@@ -208,11 +206,9 @@ async fn websocket_v3_retains_both_speakers_from_same_frame_at_handoff() -> Resu
.await?;
let requests = harness.main_loop_responses_requests().await?;
assert_eq!(requests.len(), 1);
// The 4 KiB input prefix omits the latest instruction. It must survive the transcript tail,
// even when the preceding assistant entry alone fills that field's 4 KiB cap.
assert!(response_request_contains_text(
&requests[0],
&format!("{newest_user}</transcript_delta>\n</realtime_delegation>"),
&format!("<transcript_delta>user: {user}\nassistant: {assistant_final}</transcript_delta>"),
));
harness.shutdown().await;
Ok(())
codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs
+94 -260
View File
@@ -38,7 +38,9 @@ use http::HeaderMap;
use http::HeaderValue;
use http::StatusCode;
use std::collections::HashMap;
use std::collections::HashSet;
use std::collections::VecDeque;
use std::hash::BuildHasher;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
@@ -62,8 +64,6 @@ use url::Url;
const REALTIME_WIRE_LOG_TARGET: &str = "codex_api::realtime_websocket::wire";
const OPENAI_REALTIME_API_BASE_URL: &str = "https://api.openai.com/v1";
const MAX_ACTIVE_TRANSCRIPT_BYTES: usize = 8 * 1024;
const TRUNCATED_TRANSCRIPT_PREFIX: &str = "…";
struct WsStream {
tx_command: mpsc::Sender<WsCommand>,
@@ -271,48 +271,20 @@ pub struct RealtimeTranscriptState {
#[derive(Default)]
struct ActiveTranscriptState {
entries: Vec<ActiveTranscriptEntry>,
last_frame: Option<TranscriptFrame>,
frame_sequence: u64,
input_last_raw_frame: Option<u64>,
output_last_raw_frame: Option<u64>,
entries: Vec<RealtimeTranscriptEntry>,
// Handoff input can echo an utterance already written to child history.
flushed_user_inputs: HashSet<u64>,
new_input_entry: bool,
new_output_entry: bool,
}
impl ActiveTranscriptState {
fn take_entries(&mut self) -> Vec<RealtimeTranscriptEntry> {
self.input_last_raw_frame = None;
self.output_last_raw_frame = None;
self.entries
.drain(..)
.map(|entry| entry.transcript)
.collect()
self.flushed_user_inputs.clear();
std::mem::take(&mut self.entries)
}
}
struct ActiveTranscriptEntry {
transcript: RealtimeTranscriptEntry,
last_raw_frame: Option<u64>,
truncated_prefix_bytes: usize,
}
impl ActiveTranscriptEntry {
fn retained_text(&self) -> &str {
if self.truncated_prefix_bytes > 0 {
&self.transcript.text[TRUNCATED_TRANSCRIPT_PREFIX.len()..]
} else {
&self.transcript.text
}
}
}
#[derive(Clone, Copy, PartialEq, Eq, serde::Deserialize)]
struct TranscriptFrame {
start_ms: u64,
end_ms: u64,
}
impl RealtimeWebsocketConnection {
pub async fn send_audio_frame(&self, frame: RealtimeAudioFrame) -> Result<(), ApiError> {
self.writer.send_audio_frame(frame).await
@@ -622,17 +594,7 @@ impl RealtimeWebsocketEvents {
Message::Text(text) => {
trace!(target: REALTIME_WIRE_LOG_TARGET, "realtime websocket event: {text}");
if let Some(mut event) = parse_realtime_event(&text, self.event_parser) {
let frame = if self.event_parser == RealtimeEventParser::FramelessBidi
&& matches!(
event,
RealtimeEvent::InputTranscriptDelta(_)
| RealtimeEvent::OutputTranscriptDelta(_)
) {
serde_json::from_str::<TranscriptFrame>(&text).ok()
} else {
None
};
self.update_active_transcript(&mut event, frame).await;
self.update_active_transcript(&mut event).await;
debug!(?event, "realtime websocket parsed event");
return Ok(Some(event));
}
@@ -688,26 +650,11 @@ impl RealtimeWebsocketEvents {
}
}
async fn update_active_transcript(
&self,
event: &mut RealtimeEvent,
frame: Option<TranscriptFrame>,
) {
async fn update_active_transcript(&self, event: &mut RealtimeEvent) {
let mut active_transcript = self.transcript_state.active.lock().await;
let frame_sequence = frame.map(|frame| {
// Both speakers in one frame have equal recency, regardless of wire event order.
// Count observed frames so reconnects do not depend on an absolute timestamp origin.
if active_transcript.last_frame != Some(frame) {
truncate_active_transcript(&mut active_transcript.entries);
active_transcript.frame_sequence += 1;
active_transcript.last_frame = Some(frame);
}
active_transcript.frame_sequence
});
match event {
RealtimeEvent::InputAudioSpeechStarted(_) => {
active_transcript.new_input_entry = true;
active_transcript.input_last_raw_frame = None;
}
RealtimeEvent::InputTranscriptDelta(RealtimeTranscriptDelta { delta, .. }) => {
let force_new = active_transcript.new_input_entry;
@@ -717,11 +664,9 @@ impl RealtimeWebsocketEvents {
delta,
force_new,
self.event_parser,
frame_sequence,
);
if !delta.is_empty() || self.event_parser != RealtimeEventParser::FramelessBidi {
active_transcript.new_input_entry = false;
active_transcript.input_last_raw_frame = frame_sequence;
}
}
RealtimeEvent::OutputTranscriptDelta(RealtimeTranscriptDelta { delta, .. }) => {
@@ -732,53 +677,43 @@ impl RealtimeWebsocketEvents {
delta,
force_new,
self.event_parser,
frame_sequence,
);
if !delta.is_empty() || self.event_parser != RealtimeEventParser::FramelessBidi {
active_transcript.new_output_entry = false;
active_transcript.output_last_raw_frame = frame_sequence;
}
}
RealtimeEvent::InputTranscriptDone(done) => {
let force_new = active_transcript.new_input_entry;
let last_raw_frame = active_transcript.input_last_raw_frame;
apply_transcript_done(
&mut active_transcript.entries,
"user",
&done.text,
force_new,
self.event_parser,
last_raw_frame,
);
active_transcript.new_input_entry =
self.event_parser == RealtimeEventParser::FramelessBidi;
active_transcript.input_last_raw_frame = None;
}
RealtimeEvent::OutputTranscriptDone(done) => {
let force_new = active_transcript.new_output_entry;
let last_raw_frame = active_transcript.output_last_raw_frame;
apply_transcript_done(
&mut active_transcript.entries,
"assistant",
&done.text,
force_new,
self.event_parser,
last_raw_frame,
);
active_transcript.new_output_entry =
self.event_parser == RealtimeEventParser::FramelessBidi;
active_transcript.output_last_raw_frame = None;
}
RealtimeEvent::HandoffRequested(handoff) => {
truncate_active_transcript(&mut active_transcript.entries);
append_handoff_input(&mut active_transcript.entries, &handoff.input_transcript);
append_handoff_input(&mut active_transcript, &handoff.input_transcript);
handoff.active_transcript = active_transcript.take_entries();
active_transcript.new_input_entry = true;
active_transcript.new_output_entry = true;
}
RealtimeEvent::ResponseCreated(_) => {
active_transcript.new_output_entry = true;
active_transcript.output_last_raw_frame = None;
}
RealtimeEvent::SessionUpdated { .. }
| RealtimeEvent::AudioOut(_)
@@ -792,113 +727,68 @@ impl RealtimeWebsocketEvents {
| RealtimeEvent::HistoryTranscriptDelta { .. }
| RealtimeEvent::HistoryItemCompleted(_) => {}
}
if let Some(frame_sequence) = frame_sequence {
// Wait for both speakers before evicting utterances from this frame. Each updated
// utterance is bounded independently, so pending transcript text is at most the previous
// 8 KiB transcript plus 8 KiB per speaker, even if no further event arrives.
for entry in &mut active_transcript.entries {
if entry.last_raw_frame == Some(frame_sequence) {
truncate_transcript_entry(entry, MAX_ACTIVE_TRANSCRIPT_BYTES);
}
}
} else {
truncate_active_transcript(&mut active_transcript.entries);
}
pub async fn take_completed_transcript(
&self,
event: &RealtimeEvent,
threshold: usize,
) -> Vec<RealtimeTranscriptEntry> {
if self.event_parser != RealtimeEventParser::FramelessBidi
|| !matches!(
event,
RealtimeEvent::InputTranscriptDone(_) | RealtimeEvent::OutputTranscriptDone(_)
)
{
return Vec::new();
}
let mut active = self.transcript_state.active.lock().await;
// Only the other speaker can still be open after turn.done. Keep their utterance
// and everything after it in place so history stays in conversation order.
let open_role = match event {
RealtimeEvent::InputTranscriptDone(_) if !active.new_output_entry => Some("assistant"),
RealtimeEvent::OutputTranscriptDone(_) if !active.new_input_entry => Some("user"),
_ => None,
};
let entries = &active.entries;
let end = open_role
.and_then(|role| entries.iter().rposition(|entry| entry.role == role))
.unwrap_or(entries.len());
if end == 0 || transcript_entries_bytes(entries) <= threshold {
return Vec::new();
}
let completed: Vec<_> = active.entries.drain(..end).collect();
let recorded = &mut active.flushed_user_inputs;
let hash = recorded.hasher().clone();
recorded.extend(
completed
.iter()
.filter(|entry| entry.role == "user")
.map(|entry| hash.hash_one(entry.text.trim())),
);
completed
}
}
impl RealtimeTranscriptState {
pub async fn take_transcript_tail(&self) -> Vec<RealtimeTranscriptEntry> {
let mut active = self.active.lock().await;
truncate_active_transcript(&mut active.entries);
active.take_entries()
self.active.lock().await.take_entries()
}
}
fn truncate_active_transcript(entries: &mut Vec<ActiveTranscriptEntry>) {
let mut total_bytes = transcript_entries_bytes(entries.iter().map(|entry| &entry.transcript));
while total_bytes > MAX_ACTIVE_TRANSCRIPT_BYTES && entries.len() > 1 {
// An earlier utterance can contain the newest speech. Evict by raw frame recency,
// preserving the legacy FIFO policy without timing.
let oldest = if entries.iter().all(|entry| entry.last_raw_frame.is_some()) {
entries
.iter()
.enumerate()
.min_by_key(|(_, entry)| entry.last_raw_frame)
.map(|(index, _)| index)
.unwrap_or_default()
} else {
0
};
if let Some(frame) = entries[oldest].last_raw_frame
&& entries
.iter()
.all(|entry| entry.last_raw_frame == Some(frame))
{
// Both speakers have equally recent speech. Keep their suffixes in conversation
// order, letting short entries keep their text before dividing the remaining budget.
let mut remaining_bytes = MAX_ACTIVE_TRANSCRIPT_BYTES;
let mut remaining_entries = entries.len();
let mut by_size: Vec<_> = entries.iter_mut().collect();
by_size.sort_unstable_by_key(|entry| transcript_entry_bytes(&entry.transcript));
for entry in by_size {
let entry_budget = remaining_bytes / remaining_entries;
truncate_transcript_entry(entry, entry_budget);
remaining_bytes -= transcript_entry_bytes(&entry.transcript);
remaining_entries -= 1;
}
return;
}
total_bytes -= transcript_entry_bytes(&entries.remove(oldest).transcript);
}
if let Some(entry) = entries.first_mut() {
truncate_transcript_entry(entry, MAX_ACTIVE_TRANSCRIPT_BYTES);
}
}
fn truncate_transcript_entry(entry: &mut ActiveTranscriptEntry, max_bytes: usize) {
let entry_overhead = entry.transcript.role.len() + 3;
let max_text_bytes = max_bytes.saturating_sub(entry_overhead);
if entry.transcript.text.len() <= max_text_bytes {
return;
}
let mut start = entry
.transcript
.text
.len()
.saturating_sub(max_text_bytes.saturating_sub(TRUNCATED_TRANSCRIPT_PREFIX.len()));
while !entry.transcript.text.is_char_boundary(start) {
start += 1;
}
let previous_marker_bytes = if entry.truncated_prefix_bytes > 0 {
TRUNCATED_TRANSCRIPT_PREFIX.len()
} else {
0
};
entry.truncated_prefix_bytes += start - previous_marker_bytes;
entry.transcript.text = format!(
"{TRUNCATED_TRANSCRIPT_PREFIX}{}",
&entry.transcript.text[start..]
);
}
fn transcript_entries_bytes<'a>(
entries: impl IntoIterator<Item = &'a RealtimeTranscriptEntry>,
) -> usize {
entries.into_iter().map(transcript_entry_bytes).sum()
}
fn transcript_entry_bytes(entry: &RealtimeTranscriptEntry) -> usize {
entry.role.len() + entry.text.len() + 3
fn transcript_entries_bytes(entries: &[RealtimeTranscriptEntry]) -> usize {
entries
.iter()
.map(|entry| entry.role.len() + entry.text.len() + 3)
.sum()
}
fn append_transcript_delta(
entries: &mut Vec<ActiveTranscriptEntry>,
entries: &mut Vec<RealtimeTranscriptEntry>,
role: &str,
delta: &str,
force_new: bool,
event_parser: RealtimeEventParser,
frame_sequence: Option<u64>,
) {
if delta.is_empty() {
return;
@@ -907,37 +797,30 @@ fn append_transcript_delta(
// 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.transcript.role == role),
RealtimeEventParser::V1 | RealtimeEventParser::RealtimeV2 => entries
.last_mut()
.filter(|entry| entry.transcript.role == role),
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.transcript.text.push_str(delta);
last_entry.last_raw_frame = frame_sequence;
last_entry.text.push_str(delta);
return;
}
entries.push(ActiveTranscriptEntry {
transcript: RealtimeTranscriptEntry {
role: role.to_string(),
text: delta.to_string(),
},
last_raw_frame: frame_sequence,
truncated_prefix_bytes: 0,
entries.push(RealtimeTranscriptEntry {
role: role.to_string(),
text: delta.to_string(),
});
}
fn apply_transcript_done(
entries: &mut Vec<ActiveTranscriptEntry>,
entries: &mut Vec<RealtimeTranscriptEntry>,
role: &str,
text: &str,
force_new: bool,
event_parser: RealtimeEventParser,
last_raw_frame: Option<u64>,
) {
if text.is_empty() {
return;
@@ -946,59 +829,49 @@ fn apply_transcript_done(
// 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.transcript.role == role),
RealtimeEventParser::V1 | RealtimeEventParser::RealtimeV2 => entries
.last_mut()
.filter(|entry| entry.transcript.role == role),
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
.get(last_entry.truncated_prefix_bytes..)
.is_some_and(|text| text.starts_with(last_entry.retained_text()))
if event_parser != RealtimeEventParser::FramelessBidi || text.starts_with(&last_entry.text)
{
last_entry.transcript.text = text.to_string();
last_entry.truncated_prefix_bytes = 0;
last_entry.text = text.to_string();
}
return;
}
entries.push(ActiveTranscriptEntry {
transcript: RealtimeTranscriptEntry {
role: role.to_string(),
text: text.to_string(),
},
// A final for an evicted current utterance must not become newer than retained speech.
last_raw_frame: if force_new { None } else { last_raw_frame },
truncated_prefix_bytes: 0,
entries.push(RealtimeTranscriptEntry {
role: role.to_string(),
text: text.to_string(),
});
}
fn append_handoff_input(entries: &mut Vec<ActiveTranscriptEntry>, input: &str) {
fn append_handoff_input(active: &mut ActiveTranscriptState, input: &str) {
let input = input.trim();
if input.is_empty() || contains_transcript_entry(entries, "user", input) {
let recorded = &active.flushed_user_inputs;
if input.is_empty()
|| contains_transcript_entry(&active.entries, "user", input)
|| (!recorded.is_empty() && recorded.contains(&recorded.hasher().hash_one(input)))
{
return;
}
entries.push(ActiveTranscriptEntry {
transcript: RealtimeTranscriptEntry {
role: "user".to_string(),
text: input.to_string(),
},
last_raw_frame: None,
truncated_prefix_bytes: 0,
active.entries.push(RealtimeTranscriptEntry {
role: "user".to_string(),
text: input.to_string(),
});
}
fn contains_transcript_entry(entries: &[ActiveTranscriptEntry], role: &str, text: &str) -> bool {
fn contains_transcript_entry(entries: &[RealtimeTranscriptEntry], role: &str, text: &str) -> bool {
entries
.iter()
.any(|entry| entry.transcript.role == role && entry.transcript.text.trim() == text.trim())
.any(|entry| entry.role == role && entry.text.trim() == text.trim())
}
pub struct RealtimeWebsocketClient {
@@ -1611,18 +1484,14 @@ mod tests {
let mut covered = RealtimeEvent::InputTranscriptDelta(RealtimeTranscriptDelta {
delta: "already handed off".to_string(),
});
events
.update_active_transcript(&mut covered, /*frame*/ None)
.await;
events.update_active_transcript(&mut covered).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![],
});
events
.update_active_transcript(&mut handoff, /*frame*/ None)
.await;
events.update_active_transcript(&mut handoff).await;
assert_eq!(
handoff,
RealtimeEvent::HandoffRequested(RealtimeHandoffRequested {
@@ -1639,9 +1508,7 @@ mod tests {
let mut tail = RealtimeEvent::OutputTranscriptDelta(RealtimeTranscriptDelta {
delta: "tail".to_string(),
});
events
.update_active_transcript(&mut tail, /*frame*/ None)
.await;
events.update_active_transcript(&mut tail).await;
assert_eq!(
events.take_transcript_tail().await,
vec![RealtimeTranscriptEntry {
@@ -1666,9 +1533,7 @@ mod tests {
let mut partial = RealtimeEvent::InputTranscriptDelta(RealtimeTranscriptDelta {
delta: "hello wor".to_string(),
});
first_events
.update_active_transcript(&mut partial, /*frame*/ None)
.await;
first_events.update_active_transcript(&mut partial).await;
let (_second_tx, second_rx) = async_channel::unbounded();
let second_events = RealtimeWebsocketEvents {
@@ -1681,9 +1546,7 @@ mod tests {
let mut done = RealtimeEvent::InputTranscriptDone(RealtimeTranscriptDone {
text: "hello world".to_string(),
});
second_events
.update_active_transcript(&mut done, /*frame*/ None)
.await;
second_events.update_active_transcript(&mut done).await;
assert_eq!(
second_events.take_transcript_tail().await,
@@ -1695,35 +1558,6 @@ mod tests {
assert_eq!(second_events.take_transcript_tail().await, vec![]);
}
#[tokio::test]
async fn active_transcript_retains_a_bounded_suffix() {
let (_tx_message, rx_message) = async_channel::unbounded();
let events = RealtimeWebsocketEvents {
rx_message,
pending_events: Arc::new(Mutex::new(VecDeque::new())),
transcript_state: RealtimeTranscriptState::default(),
event_parser: RealtimeEventParser::FramelessBidi,
is_closed: Arc::new(AtomicBool::new(false)),
};
let mut oversized = RealtimeEvent::InputTranscriptDelta(RealtimeTranscriptDelta {
delta: format!("old{}new", "x".repeat(MAX_ACTIVE_TRANSCRIPT_BYTES)),
});
for frame in [
None,
Some(TranscriptFrame {
start_ms: 0,
end_ms: 200,
}),
] {
events.update_active_transcript(&mut oversized, frame).await;
let tail = events.take_transcript_tail().await;
assert!(transcript_entries_bytes(&tail) <= MAX_ACTIVE_TRANSCRIPT_BYTES);
assert!(tail[0].text.starts_with(TRUNCATED_TRANSCRIPT_PREFIX));
assert!(tail[0].text.ends_with("new"));
}
}
#[test]
fn terminal_sideband_handshake_statuses_are_not_retryable() {
for status in [StatusCode::NOT_FOUND, StatusCode::GONE] {
codex-rs/codex-api/src/endpoint/realtime_websocket/transcript_tests.rs
+1 -176
View File
@@ -24,9 +24,7 @@ fn entry(role: &str, text: &str) -> RealtimeTranscriptEntry {
async fn observe(events: &RealtimeWebsocketEvents, mut event: RealtimeEvent) {
let original = event.clone();
events
.update_active_transcript(&mut event, /*frame*/ None)
.await;
events.update_active_transcript(&mut event).await;
assert_eq!(
event, original,
"live transcript events must pass through unchanged"
@@ -131,176 +129,3 @@ async fn completed_utterances_keep_repeated_speech_and_final_only_turns_distinct
);
}
}
#[tokio::test]
async fn transcript_budget_uses_frame_recency_without_reordering_speakers() {
use serde_json::json;
let old_user = "u".repeat(4000);
let assistant = "a".repeat(4000);
let newest_user = "n".repeat(211);
let current_user = "u".repeat(4160);
let same_frame_user = "n".repeat(22);
let unicode_user = "é".repeat(2000);
let oversized_user = "u".repeat(MAX_ACTIVE_TRANSCRIPT_BYTES);
let extended_assistant = format!("{assistant}{newest_user}!");
for (frames, final_assistant, expected) in [
(
vec![
("user", old_user.as_str(), Some(0)),
("assistant", assistant.as_str(), Some(200)),
("user", newest_user.as_str(), Some(400)),
// Commit eviction before the delayed final arrives.
("user", "", Some(600)),
],
// A late final must keep the evicted assistant's old raw-frame age.
Some(assistant.as_str()),
vec![entry("user", &format!("{old_user}{newest_user}"))],
),
(
vec![
("assistant", assistant.as_str(), Some(0)),
("user", current_user.as_str(), Some(200)),
// The first event already exceeds the budget, before output updates its age.
("user", same_frame_user.as_str(), Some(400)),
// Same-frame input is emitted first; output must not become newer.
("assistant", "bbbbbbbbbbbbbbbbbbbb", Some(400)),
],
None,
vec![
entry("assistant", &format!("{assistant}bbbbbbbbbbbbbbbbbbbb")),
entry("user", &format!("…{}{same_frame_user}", "u".repeat(4128))),
],
),
(
vec![
("user", old_user.as_str(), Some(0)),
("assistant", assistant.as_str(), Some(200)),
("user", newest_user.as_str(), Some(400)),
("assistant", "b", Some(400)),
],
None,
vec![
entry("user", &format!("…{}{newest_user}", "u".repeat(3958))),
entry("assistant", &format!("{assistant}b")),
],
),
(
vec![
("user", old_user.as_str(), Some(0)),
("assistant", assistant.as_str(), Some(200)),
// Reversing the same-frame events must retain the same suffixes.
("assistant", "b", Some(400)),
("user", newest_user.as_str(), Some(400)),
// Advancing the frame also commits eviction before the tail is taken.
("user", "", Some(600)),
],
None,
vec![
entry("user", &format!("…{}{newest_user}", "u".repeat(3958))),
entry("assistant", &format!("{assistant}b")),
],
),
(
vec![
("user", unicode_user.as_str(), Some(0)),
("assistant", assistant.as_str(), Some(200)),
("user", newest_user.as_str(), Some(400)),
("assistant", newest_user.as_str(), Some(400)),
],
None,
vec![
entry("user", &format!("…{}{newest_user}", "é".repeat(1937))),
entry("assistant", &format!("…{}{newest_user}", "a".repeat(3871))),
],
),
(
vec![
("user", oversized_user.as_str(), Some(0)),
("assistant", "b", Some(0)),
],
None,
vec![
entry("user", &format!("…{}", "u".repeat(8169))),
entry("assistant", "b"),
],
),
(
vec![
("assistant", assistant.as_str(), Some(0)),
("user", old_user.as_str(), Some(200)),
("assistant", newest_user.as_str(), Some(400)),
("user", "b", Some(400)),
// Trim the tied utterances before a final extends the retained suffix.
("user", "", Some(600)),
],
Some(extended_assistant.as_str()),
vec![
entry("assistant", &format!("…{}{newest_user}!", "a".repeat(3957))),
entry("user", &format!("{old_user}b")),
],
),
(
vec![
("user", old_user.as_str(), None),
("assistant", assistant.as_str(), None),
("user", newest_user.as_str(), None),
],
None,
vec![entry("assistant", &assistant)],
),
(
vec![
("assistant", "I will", Some(0)),
("user", "Stop!", Some(200)),
("assistant", " do it", Some(200)),
],
Some("I will do it"),
vec![entry("assistant", "I will do it"), entry("user", "Stop!")],
),
] {
let state = RealtimeTranscriptState::default();
let mut events = events(state.clone());
let (sender, receiver) = async_channel::unbounded();
events.rx_message = receiver;
for (role, text, start_ms) in frames {
let event_type = if role == "user" {
"input_transcript.added"
} else {
"output_transcript.added"
};
let mut payload = json!({"type": event_type, "item": {"text": text}});
if let Some(start_ms) = start_ms {
payload["start_ms"] = json!(start_ms);
payload["end_ms"] = json!(start_ms + 200);
}
sender
.send(Ok(Message::Text(payload.to_string().into())))
.await
.unwrap();
let delta = RealtimeTranscriptDelta { delta: text.into() };
assert_eq!(
events.next_event().await.unwrap(),
Some(if role == "user" {
RealtimeEvent::InputTranscriptDelta(delta)
} else {
RealtimeEvent::OutputTranscriptDelta(delta)
}),
);
}
if let Some(text) = final_assistant {
sender.send(Ok(Message::Text(json!({
"type": "turn.done",
"turn": {"role": "assistant", "transcript": text, "start_ms": 0, "end_ms": 800}
}).to_string().into()))).await.unwrap();
assert_eq!(
events.next_event().await.unwrap(),
Some(RealtimeEvent::OutputTranscriptDone(
RealtimeTranscriptDone { text: text.into() }
)),
);
}
assert_eq!(state.take_transcript_tail().await, expected);
assert_eq!(state.take_transcript_tail().await, vec![]);
}
}
codex-rs/core/src/context/realtime_delegation.rs
+12 -4
View File
@@ -3,7 +3,6 @@ use codex_protocol::models::ContentItem;
use codex_protocol::models::ContentItemKind;
use codex_protocol::models::input_text;
const MAX_REALTIME_DELEGATION_FIELD_BYTES: usize = 4 * 1024;
const TRUNCATION_MARKER: &str = "…";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -17,18 +16,23 @@ pub(crate) struct RealtimeDelegation<'a> {
input: &'a str,
transcript_delta: Option<&'a str>,
source: RealtimeDelegationSource,
preserve_transcript: bool,
}
impl<'a> RealtimeDelegation<'a> {
pub(crate) const FIELD_BYTES: usize = 4 * 1024;
pub(crate) fn new(
input: &'a str,
transcript_delta: Option<&'a str>,
source: RealtimeDelegationSource,
preserve_transcript: bool,
) -> Self {
Self {
input,
transcript_delta,
source,
preserve_transcript,
}
}
}
@@ -59,7 +63,11 @@ impl ContextualUserFragment for RealtimeDelegation<'_> {
}
};
if let Some(transcript_delta) = self.transcript_delta.filter(|text| !text.is_empty()) {
let transcript_delta = escape_xml_text_bounded(transcript_delta, Retain::End);
let transcript_delta = if self.preserve_transcript {
escape_xml_text(transcript_delta)
} else {
escape_xml_text_bounded(transcript_delta, Retain::End)
};
return input_text(format!(
"\n{source} <input>{input}</input>\n <transcript_delta>{transcript_delta}</transcript_delta>\n"
));
@@ -77,10 +85,10 @@ enum Retain {
fn escape_xml_text_bounded(input: &str, retain: Retain) -> String {
let escaped = escape_xml_text(input);
if escaped.len() <= MAX_REALTIME_DELEGATION_FIELD_BYTES {
if escaped.len() <= RealtimeDelegation::FIELD_BYTES {
return escaped;
}
let retained_bytes = MAX_REALTIME_DELEGATION_FIELD_BYTES - TRUNCATION_MARKER.len();
let retained_bytes = RealtimeDelegation::FIELD_BYTES - TRUNCATION_MARKER.len();
match retain {
Retain::Start => {
let mut end = retained_bytes;
codex-rs/core/src/realtime_conversation.rs
+57 -23
View File
@@ -160,6 +160,16 @@ impl RealtimeHandoffAdmission {
Ok(())
}
async fn record_transcript(&self, session: &Session, text: String) -> std::io::Result<()> {
let Ok(_permit) = self.gate.acquire().await else {
return Ok(());
};
if !self.retired.load(Ordering::Acquire) {
transcript_tail::record(session, text).await?;
}
Ok(())
}
async fn retire(&self) {
let Ok(_permit) = self.gate.acquire().await else {
return;
@@ -477,13 +487,16 @@ impl RealtimeResponseCreateQueue {
}
}
// Any completed transcript travels with its event so the fanout persists it before later handoffs.
type RealtimeOutput = (RealtimeEvent, Option<String>);
struct RealtimeInputTask {
writer: RealtimeWebsocketWriter,
events: RealtimeWebsocketEvents,
text_rx: Receiver<ConversationTextParams>,
handoff_output_rx: Receiver<RealtimeOutbound>,
audio_rx: Receiver<RealtimeAudioFrame>,
events_tx: Sender<RealtimeEvent>,
events_tx: Sender<RealtimeOutput>,
handoff_state: RealtimeHandoffState,
session_kind: RealtimeSessionKind,
event_parser: RealtimeEventParser,
@@ -492,6 +505,7 @@ struct RealtimeInputTask {
struct RealtimeTranscriptTailFlush {
enabled: bool,
event_parser: RealtimeEventParser,
tx: Sender<String>,
}
@@ -552,7 +566,7 @@ struct RealtimeStart {
struct RealtimeStartOutput {
realtime_active: Arc<AtomicBool>,
route_handoffs: Arc<RealtimeHandoffAdmission>,
events_rx: Receiver<RealtimeEvent>,
events_rx: Receiver<RealtimeOutput>,
transcript_tail_rx: Receiver<String>,
sdp: Option<String>,
}
@@ -678,7 +692,7 @@ impl RealtimeConversationManager {
let (handoff_output_tx, handoff_output_rx) =
async_channel::bounded::<RealtimeOutbound>(HANDOFF_OUT_QUEUE_CAPACITY);
let (events_tx, events_rx) =
async_channel::bounded::<RealtimeEvent>(OUTPUT_EVENTS_QUEUE_CAPACITY);
async_channel::bounded::<RealtimeOutput>(OUTPUT_EVENTS_QUEUE_CAPACITY);
let (transcript_tail_tx, transcript_tail_rx) = async_channel::bounded::<String>(1);
let realtime_active = Arc::new(AtomicBool::new(true));
@@ -712,6 +726,7 @@ impl RealtimeConversationManager {
};
let transcript_tail_flush = RealtimeTranscriptTailFlush {
enabled: flush_transcript_tail_on_session_end,
event_parser,
tx: transcript_tail_tx,
};
let (task, sdp) = if let Some(sdp) = sdp {
@@ -1739,6 +1754,7 @@ async fn handle_start_inner(
start: realtime_start_instructions,
end: realtime_end_instructions,
};
let event_parser = session_config.event_parser;
let start = RealtimeStart {
api_provider,
http_client_factory,
@@ -1799,7 +1815,20 @@ async fn handle_start_inner(
let mut end = RealtimeConversationEnd::TransportClosed;
let mut handoff_error = None;
// Drain already-parsed events so a queued handoff is routed before the final tail.
while let Ok(event) = events_rx.recv().await {
while let Ok((event, completed)) = events_rx.recv().await {
if let Some(delta) = completed {
let text = wrap_realtime_delegation_input(
"",
Some(&delta),
RealtimeDelegationSource::Handoff,
/*preserve_transcript*/ true,
);
if let Err(err) = route_handoffs.record_transcript(&sess_clone, text).await {
warn!("failed to save realtime transcript: {err}");
handoff_error = Some("failed to save the realtime transcript");
break;
}
}
match &event {
RealtimeEvent::AudioOut(_) => {}
_ => {
@@ -1812,9 +1841,10 @@ async fn handle_start_inner(
end = RealtimeConversationEnd::Error;
}
let maybe_routed_text = match &event {
RealtimeEvent::HandoffRequested(handoff) => {
realtime_delegation_from_handoff(handoff)
}
RealtimeEvent::HandoffRequested(handoff) => realtime_delegation_from_handoff(
handoff,
event_parser == RealtimeEventParser::FramelessBidi,
),
_ => None,
};
if let Some(text) = maybe_routed_text {
@@ -1833,16 +1863,13 @@ async fn handle_start_inner(
break;
}
}
drop(events_rx);
if handoff_error.is_none()
&& let Ok(text) = transcript_tail_rx.recv().await
&& let Err(err) = route_handoffs.record_transcript(&sess_clone, text).await
{
let _permit = route_handoffs.gate.acquire().await;
if !route_handoffs.retired.load(Ordering::Acquire)
&& let Err(err) = transcript_tail::record(&sess_clone, text).await
{
warn!("failed to flush realtime transcript before closure: {err}");
handoff_error = Some("failed to save the realtime transcript before closure");
}
warn!("failed to flush realtime transcript before closure: {err}");
handoff_error = Some("failed to save the realtime transcript before closure");
}
// Make realtime_end eligible only after the tail has reached history.
{
@@ -1918,12 +1945,16 @@ fn realtime_text_from_handoff_request(handoff: &RealtimeHandoffRequested) -> Opt
.or_else(|| realtime_transcript_delta_from_handoff(handoff))
}
fn realtime_delegation_from_handoff(handoff: &RealtimeHandoffRequested) -> Option<String> {
fn realtime_delegation_from_handoff(
handoff: &RealtimeHandoffRequested,
preserve_transcript: bool,
) -> Option<String> {
let input = realtime_text_from_handoff_request(handoff)?;
Some(wrap_realtime_delegation_input(
&input,
realtime_transcript_delta_from_handoff(handoff).as_deref(),
RealtimeDelegationSource::Handoff,
preserve_transcript,
))
}
@@ -1931,8 +1962,9 @@ fn wrap_realtime_delegation_input(
input: &str,
transcript_delta: Option<&str>,
source: RealtimeDelegationSource,
preserve_transcript: bool,
) -> String {
RealtimeDelegation::new(input, transcript_delta, source)
RealtimeDelegation::new(input, transcript_delta, source, preserve_transcript)
.render()
.into_text()
.unwrap_or_default()
@@ -2119,7 +2151,7 @@ struct RealtimeWebrtcSidebandInputTask {
sideband_headers: HeaderMap,
session_initialization: RealtimeSidebandSessionInitialization,
input_channels: RealtimeInputChannels,
events_tx: Sender<RealtimeEvent>,
events_tx: Sender<RealtimeOutput>,
handoff_state: RealtimeHandoffState,
session_kind: RealtimeSessionKind,
event_parser: RealtimeEventParser,
@@ -2176,8 +2208,9 @@ async fn run_realtime_input_task(
realtime_event = events.next_event() => {
match realtime_event {
Ok(Some(event)) => {
let completed = events.take_completed_transcript(&event, RealtimeDelegation::FIELD_BYTES).await;
handle_realtime_server_event(
event,
(event, realtime_transcript_delta(&completed)),
&writer,
&events_tx,
&handoff_state,
@@ -2290,11 +2323,11 @@ async fn send_realtime_pending_outbound(
}
}
async fn report_realtime_transport_loss(events_tx: &Sender<RealtimeEvent>, err: ApiError) {
async fn report_realtime_transport_loss(events_tx: &Sender<RealtimeOutput>, err: ApiError) {
let mapped_error = map_api_error(err);
error!("realtime stream closed: {mapped_error}");
let _ = events_tx
.send(RealtimeEvent::Error(mapped_error.to_string()))
.send((RealtimeEvent::Error(mapped_error.to_string()), None))
.await;
}
@@ -2311,6 +2344,7 @@ async fn flush_realtime_transcript_tail(
REALTIME_SESSION_ENDED_HANDOFF_INSTRUCTION,
Some(&transcript_delta),
RealtimeDelegationSource::TranscriptTailFlush,
transcript_tail_flush.event_parser == RealtimeEventParser::FramelessBidi,
))
.await;
}
@@ -2606,9 +2640,9 @@ async fn handle_handoff_output(
}
async fn handle_realtime_server_event(
event: RealtimeEvent,
(event, completed): RealtimeOutput,
writer: &RealtimeWebsocketWriter,
events_tx: &Sender<RealtimeEvent>,
events_tx: &Sender<RealtimeOutput>,
handoff_state: &RealtimeHandoffState,
session_kind: RealtimeSessionKind,
output_audio_state: &mut Option<OutputAudioState>,
@@ -2754,7 +2788,7 @@ async fn handle_realtime_server_event(
| RealtimeEvent::HistoryItemCompleted(_) => false,
};
if events_tx.send(event).await.is_err() {
if events_tx.send((event, completed)).await.is_err() {
anyhow::bail!("realtime output event channel closed");
}
if should_stop {
codex-rs/core/src/realtime_conversation/existing_call.rs
+2 -2
View File
@@ -1,5 +1,6 @@
use super::RealtimeHandoffState;
use super::RealtimeInputChannels;
use super::RealtimeOutput;
use super::RealtimeSessionKind;
use super::RealtimeSidebandSessionInitialization;
use super::RealtimeTranscriptTailFlush;
@@ -7,7 +8,6 @@ use super::RealtimeWebrtcSidebandInputTask;
use super::spawn_webrtc_sideband_input_task;
use crate::client::ModelClient;
use async_channel::Sender;
use codex_api::RealtimeEvent;
use codex_api::RealtimeEventParser;
use codex_api::RealtimeSessionConfig;
use codex_api::RealtimeTranscriptState;
@@ -28,7 +28,7 @@ pub(super) struct ExistingCallAttachment {
pub(super) call_id: String,
pub(super) extra_headers: HeaderMap,
pub(super) input_channels: RealtimeInputChannels,
pub(super) events_tx: Sender<RealtimeEvent>,
pub(super) events_tx: Sender<RealtimeOutput>,
pub(super) handoff_state: RealtimeHandoffState,
pub(super) session_kind: RealtimeSessionKind,
pub(super) event_parser: RealtimeEventParser,
codex-rs/core/src/realtime_conversation/sideband.rs
+1 -1
View File
@@ -107,7 +107,7 @@ pub(super) fn spawn_webrtc_sideband_input_task(
"failed to connect realtime sideband: {mapped_error}"
);
let _ = events_tx
.send(RealtimeEvent::Error(mapped_error.to_string()))
.send((RealtimeEvent::Error(mapped_error.to_string()), None))
.await;
break;
}
codex-rs/core/src/realtime_conversation/transcript_tail.rs
+1 -1
View File
@@ -1,4 +1,4 @@
//! Saves terminal speech in model and client history before realtime closure, without inference.
//! Records realtime transcript chunks in model and client history without inference.
use crate::session::new_submission_id;
use crate::session::session::Session;
codex-rs/core/src/realtime_conversation_tests.rs
+5 -1
View File
@@ -307,7 +307,7 @@ fn wraps_handoff_with_transcript_delta() {
],
};
assert_eq!(
realtime_delegation_from_handoff(&handoff),
realtime_delegation_from_handoff(&handoff, /*preserve_transcript*/ false),
Some(
"<realtime_delegation>\n <input>delegate this</input>\n <transcript_delta>user: hello\nassistant: hi there</transcript_delta>\n</realtime_delegation>"
.to_string()
@@ -347,6 +347,7 @@ fn wraps_realtime_delegation_input() {
"hello",
/*transcript_delta*/ None,
RealtimeDelegationSource::Handoff,
/*preserve_transcript*/ false,
),
"<realtime_delegation>\n <input>hello</input>\n</realtime_delegation>"
);
@@ -359,6 +360,7 @@ fn wraps_realtime_delegation_input_with_xml_escaping() {
"use a < b && c > d",
Some("saw <that>"),
RealtimeDelegationSource::Handoff,
/*preserve_transcript*/ false,
),
"<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>"
);
@@ -371,6 +373,7 @@ fn wraps_realtime_delegation_input_with_xml_escaping_without_transcript() {
"use a < b && c > d",
/*transcript_delta*/ None,
RealtimeDelegationSource::Handoff,
/*preserve_transcript*/ false,
),
"<realtime_delegation>\n <input>use a &lt; b &amp;&amp; c &gt; d</input>\n</realtime_delegation>"
);
@@ -384,6 +387,7 @@ fn bounds_realtime_delegation_fields_and_keeps_latest_transcript() {
&input,
Some(&transcript),
RealtimeDelegationSource::Handoff,
/*preserve_transcript*/ false,
);
assert!(rendered.len() < 9 * 1024);
codex-rs/core/tests/suite/realtime_attachment.rs
+84
View File
@@ -319,3 +319,87 @@ async fn failed_attachment_leaves_core_disconnected() -> Result<()> {
test.codex.shutdown_and_wait().await?;
Ok(())
}
#[test_case::test_case(true; "record unfinished tail")]
#[test_case::test_case(false; "drop unfinished tail")]
#[tokio::test]
async fn completed_turns_reach_history_before_the_next_handoff(flush_tail: bool) -> Result<()> {
use codex_protocol::protocol::RealtimeEvent;
use core_test_support::responses;
use core_test_support::wait_for_event;
let api = start_mock_server().await;
let replies = ["handoff", "text"].map(|id| {
responses::sse(vec![
responses::ev_response_created(id),
responses::ev_completed(id),
])
});
let model = responses::mount_sse_sequence(&api, replies.into()).await;
let test = test_codex().build_with_auto_env(&api).await?;
let mut sideband = Sideband::new().await?;
let mut params = params("transcript", &sideband.url);
params.flush_transcript_tail_on_session_end = flush_tail;
let (reply, attached) = oneshot::channel();
test.codex
.submit(Op::RealtimeConversationAttach { params, reply })
.await?;
let connection = sideband.connected().await;
connection
.accept
.send(/*value*/ true)
.expect("pending handshake");
timeout(Duration::from_secs(10), attached).await???;
let send = |event| connection.events.send(event);
let speech = format!("{}&", "u".repeat(12 * 1024));
let final_text = format!("{speech}!");
send(json!({"type":"input_transcript.added", "item":{"text":speech}}))?;
send(json!({"type":"turn.done", "turn":{"role":"assistant","transcript":"yes"}}))?;
send(json!({"type":"turn.done", "turn":{"role":"user","transcript":final_text}}))?;
wait_for_event(&test.codex, |event| matches!(event, EventMsg::RealtimeConversationRealtime(event) if matches!(event.payload, RealtimeEvent::InputTranscriptDone(_)))).await;
assert!(model.requests().is_empty());
// The handoff repeats the user input. Only the new partial belongs in its transcript.
let partial = "p".repeat(12 * 1024);
send(json!({"type":"output_transcript.added", "item":{"text":partial}}))?;
send(
json!({"type":"delegation.created","item":{"id":"handoff","type":"delegation","target":"client","content":[{"type":"input_text","text":final_text}]}}),
)?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let transcript = |request: ResponsesRequest| {
request
.message_input_texts("user")
.into_iter()
.filter(|text| text.contains("<transcript_delta>"))
.collect::<Vec<_>>()
};
let completed = transcript(model.single_request());
assert_eq!(completed.len(), 2);
let expected = format!(
"<transcript_delta>user: {}\nassistant: yes</transcript_delta>",
final_text.replace('&', "&amp;")
);
assert!(completed[0].contains("<input></input>") && completed[0].contains(&expected));
assert!(completed[1].contains(&format!(
"<transcript_delta>assistant: {partial}</transcript_delta>"
)));
let unfinished = "a".repeat(12 * 1024);
send(json!({"type":"output_transcript.added", "item":{"text":unfinished}}))?;
wait_for_event(&test.codex, |event| matches!(event, EventMsg::RealtimeConversationRealtime(event) if matches!(event.payload, RealtimeEvent::OutputTranscriptDelta(_)))).await;
detach(&test, "transcript").await?;
test.submit_text_turn("continue").await?;
let after_close = transcript(model.last_request().expect("text turn"));
assert_eq!(after_close.len(), completed.len() + usize::from(flush_tail));
assert_eq!(&after_close[..completed.len()], completed.as_slice());
if flush_tail {
assert!(after_close[completed.len()].contains(&format!(
"<transcript_delta>assistant: {unfinished}</transcript_delta>"
)));
}
Ok(())
}