github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-28 16:06:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Route realtime V3 handoffs by response channel (#33903)

Browse Source 
## What changed

- Add `codexResponseHandoffMode` to `thread/realtime/start` for V3 sessions, with `thinking` as the default and `commentary` and `bemTags` routing modes.
- Map BEM `analysis` and `commentary` output to the commentary channel and `final` output to the speakable channel, while preserving the original BEM envelope. Unrecognized output falls back to speakable.
- Mark explicit speech appends as speakable and stop adding the legacy `"Agent Final Message"` prefix to V3 handoffs. V1 and V2 continue to ignore the new setting.

## Testing

- Cover channel selection for every routing mode, streamed BEM header parsing, fallback behavior, explicit speech, and V1 compatibility.

GitOrigin-RevId: 1974578257488f2914b252c9c1990ba38831c96b
This commit is contained in:
jiayuhuang-openai
2026-07-17 21:12:40 +00:00
committed by copyberry
parent 161748a68e
commit 025db22058
26 changed files with 851 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/realtime_conversation.rs
+279 -31
View File
@@ -16,6 +16,7 @@ use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use codex_api::ApiError;
use codex_api::Provider as ApiProvider;
use codex_api::RealtimeAudioFrame;
use codex_api::RealtimeContextAppendChannel;
use codex_api::RealtimeEvent;
use codex_api::RealtimeEventParser;
use codex_api::RealtimeSessionConfig;
@@ -36,6 +37,7 @@ use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::models::MessagePhase;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::CodexResponseHandoffMode;
use codex_protocol::protocol::ConversationAudioParams;
use codex_protocol::protocol::ConversationSpeechParams;
use codex_protocol::protocol::ConversationStartParams;
@@ -74,6 +76,11 @@ use tracing::error;
use tracing::info;
use tracing::warn;
mod bem;
use self::bem::ChannelParser as BemChannelParser;
use self::bem::message_phase as bem_message_phase;
const AUDIO_IN_QUEUE_CAPACITY: usize = 256;
const TEXT_IN_QUEUE_CAPACITY: usize = 64;
const HANDOFF_OUT_QUEUE_CAPACITY: usize = 64;
@@ -121,15 +128,22 @@ enum RealtimeSessionKind {
#[derive(Clone, Debug)]
struct RealtimeHandoffState {
output_tx: Sender<RealtimeOutbound>,
last_output_text: Arc<Mutex<Option<String>>>,
last_output: Arc<Mutex<Option<RealtimeHandoffOutput>>>,
stream: Arc<Mutex<RealtimeHandoffStreamState>>,
client_managed_handoffs: bool,
codex_responses_as_items: bool,
codex_response_item_prefix: Option<String>,
codex_response_handoff_mode: CodexResponseHandoffMode,
session_kind: RealtimeSessionKind,
event_parser: RealtimeEventParser,
}
#[derive(Clone, Debug)]
struct RealtimeHandoffOutput {
text: String,
phase: Option<MessagePhase>,
}
#[derive(Debug, Default)]
struct RealtimeHandoffStreamState {
active_handoff: Option<String>,
@@ -140,6 +154,8 @@ struct RealtimeHandoffStreamState {
struct RealtimeStreamedItem {
handoff_id: String,
phase: Option<MessagePhase>,
bem_channel_parser: Option<BemChannelParser>,
prefix_final_message: bool,
sent_bytes: usize,
buffered_text: String,
tail_text: String,
@@ -154,7 +170,10 @@ impl RealtimeStreamedItem {
}
fn output_prefix(&self) -> &'static str {
if self.sent_bytes == 0 && !matches!(self.phase, Some(MessagePhase::Commentary)) {
if self.prefix_final_message
&& self.sent_bytes == 0
&& !matches!(self.phase, Some(MessagePhase::Commentary))
{
AGENT_FINAL_MESSAGE_PREFIX
} else {
""
@@ -182,6 +201,36 @@ impl RealtimeStreamedItem {
if text.is_empty() {
return;
}
let text = if let Some(parser) = self.bem_channel_parser.as_mut() {
let Some(text) = parser.push(text) else {
return;
};
self.phase = parser.phase();
text
} else {
text.to_string()
};
self.push_output_text(&text);
}
fn finish_input(&mut self) {
let Some(parser) = self.bem_channel_parser.as_mut() else {
return;
};
self.phase = parser.phase();
let text = parser.finish();
if self.phase.is_none() && !text.is_empty() {
warn!("BEM output ended before a recognized channel header was received");
self.phase = Some(MessagePhase::FinalAnswer);
}
self.push_output_text(&text);
}
fn push_output_text(&mut self, text: &str) {
if text.is_empty() {
return;
}
if self.truncated {
self.tail_text.push_str(text);
self.tail_text =
@@ -256,12 +305,35 @@ fn take_last_bytes_at_char_boundary(text: &str, max_bytes: usize) -> &str {
#[derive(Debug, PartialEq, Eq)]
enum RealtimeOutbound {
StandaloneHandoff { text: String },
HandoffUpdate { handoff_id: String, text: String },
HandoffAppend { handoff_id: String, text: String },
CompletedHandoff { handoff_id: String, text: String },
ConversationItem { text: String },
HandoffCompleteAck { handoff_id: String },
StandaloneHandoff {
text: String,
phase: Option<MessagePhase>,
},
StandaloneSpeech {
text: String,
},
HandoffUpdate {
handoff_id: String,
text: String,
phase: Option<MessagePhase>,
},
HandoffAppend {
handoff_id: String,
text: String,
phase: Option<MessagePhase>,
},
CompletedHandoff {
handoff_id: String,
text: String,
phase: Option<MessagePhase>,
},
ConversationItem {
text: String,
phase: Option<MessagePhase>,
},
HandoffCompleteAck {
handoff_id: String,
},
}
#[derive(Debug, PartialEq, Eq)]
@@ -359,16 +431,18 @@ impl RealtimeHandoffState {
client_managed_handoffs: bool,
codex_responses_as_items: bool,
codex_response_item_prefix: Option<String>,
codex_response_handoff_mode: CodexResponseHandoffMode,
session_kind: RealtimeSessionKind,
event_parser: RealtimeEventParser,
) -> Self {
Self {
output_tx,
last_output_text: Arc::new(Mutex::new(None)),
last_output: Arc::new(Mutex::new(None)),
stream: Arc::new(Mutex::new(RealtimeHandoffStreamState::default())),
client_managed_handoffs,
codex_responses_as_items,
codex_response_item_prefix,
codex_response_handoff_mode,
session_kind,
event_parser,
}
@@ -379,6 +453,11 @@ impl RealtimeHandoffState {
&& !self.client_managed_handoffs
&& !self.codex_responses_as_items
}
fn routes_handoff_by_bem(&self) -> bool {
self.event_parser == RealtimeEventParser::FramelessBidi
&& self.codex_response_handoff_mode == CodexResponseHandoffMode::BemTags
}
}
#[allow(dead_code)]
@@ -400,6 +479,7 @@ struct RealtimeStart {
flush_transcript_tail_on_session_end: bool,
codex_responses_as_items: bool,
codex_response_item_prefix: Option<String>,
codex_response_handoff_mode: CodexResponseHandoffMode,
realtime_call_api_provider: Option<ApiProvider>,
session_config: RealtimeSessionConfig,
model_client: ModelClient,
@@ -458,6 +538,7 @@ impl RealtimeConversationManager {
flush_transcript_tail_on_session_end,
codex_responses_as_items,
codex_response_item_prefix,
codex_response_handoff_mode,
realtime_call_api_provider,
session_config,
model_client,
@@ -486,6 +567,7 @@ impl RealtimeConversationManager {
client_managed_handoffs,
codex_responses_as_items,
codex_response_item_prefix,
codex_response_handoff_mode,
session_kind,
event_parser,
);
@@ -668,28 +750,45 @@ impl RealtimeConversationManager {
if handoff.client_managed_handoffs {
return Ok(());
}
let phase = if handoff.routes_handoff_by_bem() {
match bem_message_phase(&output_text) {
Some(phase) => Some(phase),
None => {
warn!("BEM output did not contain a recognized channel header");
Some(MessagePhase::FinalAnswer)
}
}
} else {
phase
};
let is_commentary = matches!(phase, Some(MessagePhase::Commentary));
let active_handoff = handoff.stream.lock().await.active_handoff.clone();
let output = match active_handoff {
Some(handoff_id) => {
let output_text = realtime_backend_output(output_text, handoff.session_kind);
*handoff.last_output_text.lock().await = Some(output_text.clone());
*handoff.last_output.lock().await = Some(RealtimeHandoffOutput {
text: output_text.clone(),
phase: phase.clone(),
});
if handoff.codex_responses_as_items {
RealtimeOutbound::ConversationItem {
text: realtime_backend_item(
output_text,
handoff.codex_response_item_prefix.as_deref(),
),
phase,
}
} else if handoff.session_kind == RealtimeSessionKind::V1 && is_commentary {
} else if handoff.event_parser == RealtimeEventParser::V1 && is_commentary {
RealtimeOutbound::HandoffAppend {
handoff_id,
text: output_text,
phase,
}
} else {
RealtimeOutbound::HandoffUpdate {
handoff_id,
text: output_text,
phase,
}
}
}
@@ -702,14 +801,16 @@ impl RealtimeConversationManager {
output_text,
handoff.codex_response_item_prefix.as_deref(),
),
phase,
}
} else {
RealtimeOutbound::StandaloneHandoff {
text: if handoff.session_kind == RealtimeSessionKind::V1 && !is_commentary {
text: if handoff.event_parser == RealtimeEventParser::V1 && !is_commentary {
format!("{AGENT_FINAL_MESSAGE_PREFIX}{output_text}")
} else {
output_text
},
phase,
}
}
}
@@ -745,7 +846,15 @@ impl RealtimeConversationManager {
};
let mut streamed_item = RealtimeStreamedItem {
handoff_id,
phase,
phase: if handoff.routes_handoff_by_bem() {
None
} else {
phase
},
bem_channel_parser: handoff
.routes_handoff_by_bem()
.then(BemChannelParser::default),
prefix_final_message: handoff.event_parser == RealtimeEventParser::V1,
sent_bytes: 0,
buffered_text: String::new(),
tail_text: String::new(),
@@ -821,6 +930,7 @@ impl RealtimeConversationManager {
let Some(mut streamed_item) = handoff.stream.lock().await.items.remove(item_id) else {
return false;
};
streamed_item.finish_input();
let chunk = streamed_item.drain_final_chunk();
let sent_output = streamed_item.sent_bytes > 0;
if let Some(text) = chunk {
@@ -829,6 +939,7 @@ impl RealtimeConversationManager {
.send(RealtimeOutbound::HandoffAppend {
handoff_id: streamed_item.handoff_id,
text,
phase: streamed_item.phase,
})
.await;
}
@@ -852,7 +963,7 @@ impl RealtimeConversationManager {
handoff
.output_tx
.send(RealtimeOutbound::StandaloneHandoff {
.send(RealtimeOutbound::StandaloneSpeech {
text: realtime_backend_output(text, handoff.session_kind),
})
.await
@@ -879,7 +990,7 @@ impl RealtimeConversationManager {
let Some(handoff_id) = handoff.stream.lock().await.active_handoff.clone() else {
return Ok(());
};
let Some(output_text) = handoff.last_output_text.lock().await.clone() else {
let Some(last_output) = handoff.last_output.lock().await.clone() else {
return Ok(());
};
@@ -888,7 +999,8 @@ impl RealtimeConversationManager {
} else {
RealtimeOutbound::CompletedHandoff {
handoff_id,
text: output_text,
text: last_output.text,
phase: last_output.phase,
}
};
@@ -910,7 +1022,7 @@ impl RealtimeConversationManager {
stream.active_handoff = None;
stream.items.clear();
}
*handoff.last_output_text.lock().await = None;
*handoff.last_output.lock().await = None;
}
}
@@ -987,6 +1099,7 @@ struct PreparedRealtimeConversationStart {
flush_transcript_tail_on_session_end: bool,
codex_responses_as_items: bool,
codex_response_item_prefix: Option<String>,
codex_response_handoff_mode: CodexResponseHandoffMode,
realtime_call_api_provider: Option<ApiProvider>,
requested_realtime_session_id: Option<String>,
version: RealtimeWsVersion,
@@ -1072,6 +1185,7 @@ async fn prepare_realtime_start(
flush_transcript_tail_on_session_end: params.flush_transcript_tail_on_session_end,
codex_responses_as_items: params.codex_responses_as_items,
codex_response_item_prefix: params.codex_response_item_prefix,
codex_response_handoff_mode: params.codex_response_handoff_mode,
realtime_call_api_provider,
requested_realtime_session_id,
version,
@@ -1243,6 +1357,7 @@ async fn handle_start_inner(
flush_transcript_tail_on_session_end,
codex_responses_as_items,
codex_response_item_prefix,
codex_response_handoff_mode,
realtime_call_api_provider,
requested_realtime_session_id,
version,
@@ -1261,6 +1376,7 @@ async fn handle_start_inner(
flush_transcript_tail_on_session_end,
codex_responses_as_items,
codex_response_item_prefix,
codex_response_handoff_mode,
realtime_call_api_provider,
session_config,
model_client: sess.services.model_client.clone(),
@@ -1700,7 +1816,7 @@ async fn handle_text_input(
}
async fn flush_streamed_handoff_item(handoff: &RealtimeHandoffState, item_id: &str) {
let (handoff_id, text) = {
let (handoff_id, text, phase) = {
let mut stream = handoff.stream.lock().await;
let Some(streamed_item) = stream.items.get_mut(item_id) else {
return;
@@ -1710,11 +1826,19 @@ async fn flush_streamed_handoff_item(handoff: &RealtimeHandoffState, item_id: &s
return;
};
streamed_item.last_flush_at = Instant::now();
(streamed_item.handoff_id.clone(), text)
(
streamed_item.handoff_id.clone(),
text,
streamed_item.phase.clone(),
)
};
let _ = handoff
.output_tx
.send(RealtimeOutbound::HandoffAppend { handoff_id, text })
.send(RealtimeOutbound::HandoffAppend {
handoff_id,
text,
phase,
})
.await;
}
@@ -1730,6 +1854,26 @@ fn schedule_streamed_handoff_flush(
});
}
fn v3_output_writer(
writer: &RealtimeWebsocketWriter,
phase: Option<&MessagePhase>,
handoff_mode: CodexResponseHandoffMode,
) -> RealtimeWebsocketWriter {
let channel = match handoff_mode {
CodexResponseHandoffMode::Thinking => None,
CodexResponseHandoffMode::Commentary => Some(RealtimeContextAppendChannel::Commentary),
CodexResponseHandoffMode::BemTags => match phase {
Some(MessagePhase::FinalAnswer) => Some(RealtimeContextAppendChannel::Speakable),
Some(MessagePhase::Commentary) => Some(RealtimeContextAppendChannel::Commentary),
None => Some(RealtimeContextAppendChannel::Speakable),
},
};
match channel {
Some(channel) => writer.clone().with_context_append_channel(channel),
None => writer.clone(),
}
}
async fn handle_handoff_output(
handoff_output: Result<RealtimeOutbound, RecvError>,
writer: &RealtimeWebsocketWriter,
@@ -1739,34 +1883,117 @@ async fn handle_handoff_output(
response_create_queue: &mut RealtimeResponseCreateQueue,
) -> anyhow::Result<()> {
let handoff_output = handoff_output.context("handoff output channel closed")?;
let result = match event_parser {
RealtimeEventParser::V1 | RealtimeEventParser::FramelessBidi => match handoff_output {
RealtimeOutbound::StandaloneHandoff { text } => {
RealtimeEventParser::V1 => match handoff_output {
RealtimeOutbound::StandaloneHandoff { text, phase: _ } => {
writer
.send_standalone_handoff(STANDALONE_HANDOFF_ID.to_string(), text)
.await
}
RealtimeOutbound::HandoffUpdate { handoff_id, text }
| RealtimeOutbound::CompletedHandoff { handoff_id, text } => {
RealtimeOutbound::StandaloneSpeech { text } => {
writer
.send_standalone_handoff(STANDALONE_HANDOFF_ID.to_string(), text)
.await
}
RealtimeOutbound::HandoffUpdate {
handoff_id,
text,
phase: _,
}
| RealtimeOutbound::CompletedHandoff {
handoff_id,
text,
phase: _,
} => {
writer
.send_conversation_function_call_output(handoff_id, text)
.await
}
RealtimeOutbound::HandoffAppend { handoff_id, text } => {
RealtimeOutbound::HandoffAppend {
handoff_id,
text,
phase: _,
} => {
writer
.send_conversation_handoff_append(handoff_id, text)
.await
}
RealtimeOutbound::ConversationItem { text } => {
RealtimeOutbound::ConversationItem { text, phase: _ } => {
writer
.send_conversation_item_create(text, ConversationTextRole::Developer)
.await
}
RealtimeOutbound::HandoffCompleteAck { .. } => Ok(()),
},
RealtimeEventParser::FramelessBidi => match handoff_output {
RealtimeOutbound::StandaloneHandoff { text, phase } => {
v3_output_writer(
writer,
phase.as_ref(),
handoff_state.codex_response_handoff_mode,
)
.send_standalone_handoff(STANDALONE_HANDOFF_ID.to_string(), text)
.await
}
RealtimeOutbound::StandaloneSpeech { text } => {
writer
.clone()
.with_context_append_channel(RealtimeContextAppendChannel::Speakable)
.send_standalone_handoff(STANDALONE_HANDOFF_ID.to_string(), text)
.await
}
RealtimeOutbound::HandoffUpdate {
handoff_id,
text,
phase,
} => {
v3_output_writer(
writer,
phase.as_ref(),
handoff_state.codex_response_handoff_mode,
)
.send_conversation_function_call_output(handoff_id, text)
.await
}
RealtimeOutbound::HandoffAppend {
handoff_id,
text,
phase,
} => {
v3_output_writer(
writer,
phase.as_ref(),
handoff_state.codex_response_handoff_mode,
)
.send_conversation_handoff_append(handoff_id, text)
.await
}
RealtimeOutbound::CompletedHandoff {
handoff_id,
text,
phase,
} => {
v3_output_writer(
writer,
phase.as_ref(),
handoff_state.codex_response_handoff_mode,
)
.send_conversation_function_call_output(handoff_id, text)
.await
}
RealtimeOutbound::ConversationItem { text, phase } => {
v3_output_writer(
writer,
phase.as_ref(),
handoff_state.codex_response_handoff_mode,
)
.send_conversation_item_create(text, ConversationTextRole::Developer)
.await
}
RealtimeOutbound::HandoffCompleteAck { .. } => Ok(()),
},
RealtimeEventParser::RealtimeV2 => match handoff_output {
RealtimeOutbound::StandaloneHandoff { text } => {
RealtimeOutbound::StandaloneHandoff { text, phase: _ } => {
if let Err(err) = writer
.send_conversation_item_create(text, ConversationTextRole::User)
.await
@@ -1778,8 +2005,28 @@ async fn handle_handoff_output(
.await;
}
}
RealtimeOutbound::HandoffUpdate { handoff_id, text }
| RealtimeOutbound::HandoffAppend { handoff_id, text } => {
RealtimeOutbound::StandaloneSpeech { text } => {
if let Err(err) = writer
.send_conversation_item_create(text, ConversationTextRole::User)
.await
{
Err(err)
} else {
return response_create_queue
.request_create(writer, events_tx, "standalone handoff")
.await;
}
}
RealtimeOutbound::HandoffUpdate {
handoff_id,
text,
phase: _,
}
| RealtimeOutbound::HandoffAppend {
handoff_id,
text,
phase: _,
} => {
let active_handoff = handoff_state.stream.lock().await.active_handoff.clone();
match active_handoff {
Some(active_handoff) if active_handoff == handoff_id => {}
@@ -1795,6 +2042,7 @@ async fn handle_handoff_output(
RealtimeOutbound::CompletedHandoff {
handoff_id,
text: _,
phase: _,
} => {
if let Err(err) = writer
.send_conversation_function_call_output(
@@ -1810,7 +2058,7 @@ async fn handle_handoff_output(
.await;
}
}
RealtimeOutbound::ConversationItem { text } => {
RealtimeOutbound::ConversationItem { text, phase: _ } => {
writer
.send_conversation_item_create(text, ConversationTextRole::Developer)
.await
codex-rs/core/src/realtime_conversation/bem.rs
+49
View File
@@ -0,0 +1,49 @@
use codex_protocol::models::MessagePhase;
const BEM_ASSISTANT_CHANNEL_PREFIX: &str = "<|start|>assistant<|channel|>";
const BEM_MESSAGE_MARKER: &str = "<|message|>";
pub(super) fn message_phase(text: &str) -> Option<MessagePhase> {
let channel_and_message = text.strip_prefix(BEM_ASSISTANT_CHANNEL_PREFIX)?;
let (channel, _) = channel_and_message.split_once(BEM_MESSAGE_MARKER)?;
match channel {
"analysis" | "commentary" => Some(MessagePhase::Commentary),
"final" => Some(MessagePhase::FinalAnswer),
_ => None,
}
}
/// Buffers a streamed BEM message until its channel header is complete.
///
/// Once the channel is known, the original envelope is released unchanged so
/// the frontend model can distinguish BEM `analysis` from `commentary`.
#[derive(Debug, Default)]
pub(super) struct ChannelParser {
buffered_text: String,
phase: Option<MessagePhase>,
}
impl ChannelParser {
pub(super) fn push(&mut self, text: &str) -> Option<String> {
if self.phase.is_some() {
return Some(text.to_string());
}
self.buffered_text.push_str(text);
self.phase = message_phase(&self.buffered_text);
self.phase.as_ref()?;
Some(std::mem::take(&mut self.buffered_text))
}
pub(super) fn phase(&self) -> Option<MessagePhase> {
self.phase.clone()
}
pub(super) fn finish(&mut self) -> String {
std::mem::take(&mut self.buffered_text)
}
}
#[cfg(test)]
#[path = "bem_tests.rs"]
mod tests;
codex-rs/core/src/realtime_conversation/bem_tests.rs
+42
View File
@@ -0,0 +1,42 @@
use super::ChannelParser;
use super::message_phase;
use codex_protocol::models::MessagePhase;
use pretty_assertions::assert_eq;
#[test]
fn maps_bem_channels_to_realtime_phases() {
for (channel, expected) in [
("analysis", MessagePhase::Commentary),
("commentary", MessagePhase::Commentary),
("final", MessagePhase::FinalAnswer),
] {
assert_eq!(
message_phase(&format!(
"<|start|>assistant<|channel|>{channel}<|message|>text<|end|>"
)),
Some(expected)
);
}
}
#[test]
fn buffers_streamed_text_until_the_bem_channel_is_complete() {
let mut parser = ChannelParser::default();
assert_eq!(parser.push("<|start|>assistant<|channel|>com"), None);
assert_eq!(
parser.push("mentary<|message|>progress"),
Some("<|start|>assistant<|channel|>commentary<|message|>progress".to_string())
);
assert_eq!(parser.phase(), Some(MessagePhase::Commentary));
assert_eq!(parser.push("<|end|>"), Some("<|end|>".to_string()));
}
#[test]
fn preserves_unrecognized_output_when_the_stream_finishes() {
let mut parser = ChannelParser::default();
assert_eq!(parser.push("plain output"), None);
assert_eq!(parser.finish(), "plain output");
assert_eq!(parser.phase(), None);
}
codex-rs/core/src/realtime_conversation_tests.rs
+23
View File
@@ -11,6 +11,7 @@ use crate::context::RealtimeDelegationSource;
use async_channel::bounded;
use codex_api::RealtimeEventParser;
use codex_protocol::models::MessagePhase;
use codex_protocol::protocol::CodexResponseHandoffMode;
use codex_protocol::protocol::RealtimeHandoffRequested;
use codex_protocol::protocol::RealtimeTranscriptEntry;
use pretty_assertions::assert_eq;
@@ -151,6 +152,7 @@ async fn clears_active_handoff_explicitly() {
/*client_managed_handoffs*/ false,
/*codex_responses_as_items*/ false,
/*codex_response_item_prefix*/ None,
CodexResponseHandoffMode::Thinking,
RealtimeSessionKind::V1,
/*event_parser*/ RealtimeEventParser::V1,
);
@@ -170,6 +172,8 @@ fn streamed_handoff_preserves_a_bounded_final_tail() {
let mut item = RealtimeStreamedItem {
handoff_id: "handoff_1".to_string(),
phase: Some(MessagePhase::FinalAnswer),
bem_channel_parser: None,
prefix_final_message: true,
sent_bytes: 0,
buffered_text: String::new(),
tail_text: String::new(),
@@ -193,6 +197,25 @@ fn streamed_handoff_preserves_a_bounded_final_tail() {
assert!(output.ends_with("TAIL"));
}
#[test]
fn streamed_v3_handoff_omits_the_final_message_prefix() {
let mut item = RealtimeStreamedItem {
handoff_id: "handoff_1".to_string(),
phase: Some(MessagePhase::FinalAnswer),
bem_channel_parser: None,
prefix_final_message: false,
sent_bytes: 0,
buffered_text: String::new(),
tail_text: String::new(),
truncated: false,
last_flush_at: Instant::now(),
flush_scheduled: false,
};
item.push_text("done");
assert_eq!(item.drain_final_chunk(), Some("done".to_string()));
}
#[test]
fn uses_quicksilver_alpha_header_for_realtime_v1() {
let headers = realtime_request_headers(
codex-rs/core/tests/suite/compact_remote.rs
+2
View File
@@ -212,6 +212,8 @@ async fn start_realtime_conversation(codex: &codex_core::CodexThread) -> Result<
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
codex-rs/core/tests/suite/realtime_conversation.rs
+98 -14
View File
@@ -287,6 +287,8 @@ async fn conversation_start_audio_text_close_round_trip() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -433,6 +435,8 @@ async fn conversation_start_defaults_to_v2_and_gpt_realtime_1_5() -> Result<()>
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -528,6 +532,8 @@ async fn conversation_webrtc_start_posts_generated_session() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: Some("session-override-model".to_string()),
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -715,6 +721,8 @@ async fn conversation_webrtc_start_uses_avas_query() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -813,6 +821,8 @@ async fn conversation_webrtc_default_v1_ignores_configured_v2_voice() -> Result<
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -873,6 +883,8 @@ async fn conversation_webrtc_default_v1_rejects_explicit_v2_voice() -> Result<()
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -943,6 +955,8 @@ async fn conversation_webrtc_start_uses_configured_call_base_url_for_avas() -> R
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1037,6 +1051,8 @@ async fn conversation_webrtc_close_while_sideband_connecting_drops_pending_join(
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1128,6 +1144,8 @@ async fn conversation_webrtc_sideband_connect_failure_closes_with_error() -> Res
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1221,6 +1239,8 @@ async fn conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth() ->
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1312,6 +1332,8 @@ async fn assert_transport_close_tail_flush(
flush_transcript_tail_on_session_end,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1428,6 +1450,8 @@ async fn conversation_start_preflight_failure_emits_realtime_error_only() -> Res
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1479,6 +1503,8 @@ async fn conversation_start_connect_failure_emits_realtime_error_only() -> Resul
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1578,6 +1604,8 @@ async fn conversation_second_start_replaces_runtime() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1608,6 +1636,8 @@ async fn conversation_second_start_replaces_runtime() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1709,6 +1739,8 @@ async fn conversation_uses_experimental_realtime_ws_base_url_override() -> Resul
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1778,6 +1810,8 @@ async fn conversation_uses_default_realtime_backend_prompt() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1855,6 +1889,8 @@ async fn conversation_uses_empty_instructions_for_null_or_empty_prompt() -> Resu
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1925,6 +1961,8 @@ async fn conversation_uses_explicit_start_voice() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -1987,6 +2025,8 @@ async fn conversation_uses_configured_realtime_voice() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2037,6 +2077,8 @@ async fn conversation_rejects_voice_for_wrong_realtime_version() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2088,6 +2130,8 @@ async fn conversation_uses_experimental_realtime_ws_backend_prompt_override() ->
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2165,6 +2209,8 @@ async fn conversation_uses_experimental_realtime_ws_startup_context_override() -
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2236,6 +2282,8 @@ async fn conversation_disables_realtime_startup_context_with_empty_override() ->
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2300,6 +2348,8 @@ async fn conversation_start_injects_startup_context_from_thread_history() -> Res
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2419,6 +2469,8 @@ async fn conversation_startup_context_current_thread_selects_many_turns_by_budge
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2531,6 +2583,8 @@ async fn conversation_startup_context_falls_back_to_workspace_map() -> Result<()
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2595,6 +2649,8 @@ async fn conversation_startup_context_is_truncated_and_sent_once_per_start() ->
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2680,6 +2736,8 @@ async fn conversation_user_text_turn_is_not_sent_to_realtime() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2781,6 +2839,8 @@ async fn realtime_v2_noop_tool_call_returns_empty_function_output_without_respon
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2884,6 +2944,8 @@ async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Re
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -2958,22 +3020,20 @@ async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Re
async fn conversation_flushes_assistant_deltas_every_200ms_for_v3_handoff() -> Result<()> {
skip_if_no_network!(Ok(()));
let initial_commentary_text = "seed ";
let first_commentary_delta = "first ";
let second_commentary_delta = "x".repeat(94);
let initial_commentary_text = "<|start|>assistant<|channel|>com";
let first_commentary_delta = "mentary<|message|>seed first ";
let second_commentary_delta = format!("{}<|end|>", "x".repeat(94));
let commentary_text =
format!("{initial_commentary_text}{first_commentary_delta}{second_commentary_delta}");
let (gate_commentary_done_tx, gate_commentary_done_rx) = oneshot::channel();
let mut commentary_item_added =
let commentary_item_added =
responses::ev_message_item_added("msg_commentary", initial_commentary_text);
commentary_item_added["item"]["phase"] = json!("commentary");
let mut commentary_item_done =
responses::ev_assistant_message("msg_commentary", &commentary_text);
commentary_item_done["item"]["phase"] = json!("commentary");
let mut final_item_added = responses::ev_message_item_added("msg_final", "");
final_item_added["item"]["phase"] = json!("final_answer");
let mut final_item_done = responses::ev_assistant_message("msg_final", "done");
final_item_done["item"]["phase"] = json!("final_answer");
let commentary_item_done = responses::ev_assistant_message("msg_commentary", &commentary_text);
let initial_final_text = "<|start|>assistant<|channel|>fi";
let final_delta = "nal<|message|>done<|end|>";
let final_text = format!("{initial_final_text}{final_delta}");
let final_item_added = responses::ev_message_item_added("msg_final", initial_final_text);
let final_item_done = responses::ev_assistant_message("msg_final", &final_text);
let response_chunks = vec![
StreamingSseChunk {
gate: None,
@@ -3001,7 +3061,7 @@ async fn conversation_flushes_assistant_deltas_every_200ms_for_v3_handoff() -> R
},
StreamingSseChunk {
gate: None,
body: sse_event(responses::ev_output_text_delta("done")),
body: sse_event(responses::ev_output_text_delta(final_delta)),
},
StreamingSseChunk {
gate: None,
@@ -3050,6 +3110,8 @@ async fn conversation_flushes_assistant_deltas_every_200ms_for_v3_handoff() -> R
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::BemTags,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -3100,6 +3162,7 @@ async fn conversation_flushes_assistant_deltas_every_200ms_for_v3_handoff() -> R
json!({
"type": "delegation.context.append",
"delegation_item_id": "delegation_stream",
"channel": "commentary",
"content": [{ "type": "input_text", "text": commentary_text }]
})
);
@@ -3116,9 +3179,10 @@ async fn conversation_flushes_assistant_deltas_every_200ms_for_v3_handoff() -> R
json!({
"type": "delegation.context.append",
"delegation_item_id": "delegation_stream",
"channel": "speakable",
"content": [{
"type": "input_text",
"text": "\"Agent Final Message\":\n\ndone"
"text": final_text
}]
})
);
@@ -3214,6 +3278,8 @@ async fn conversation_handoff_persists_across_item_done_until_turn_complete() ->
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::BemTags,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -3370,6 +3436,8 @@ async fn inbound_handoff_request_starts_turn() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -3484,6 +3552,8 @@ async fn inbound_handoff_request_uses_active_transcript() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -3591,6 +3661,8 @@ async fn inbound_handoff_request_sends_transcript_delta_after_each_handoff() ->
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -3717,6 +3789,8 @@ async fn conversation_close_routes_only_remaining_transcript_tail_once() -> Resu
flush_transcript_tail_on_session_end: true,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -3804,6 +3878,8 @@ async fn inbound_conversation_item_does_not_start_turn_and_still_forwards_audio(
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -3931,6 +4007,8 @@ async fn delegated_turn_user_role_echo_does_not_redelegate_and_still_forwards_au
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -4088,6 +4166,8 @@ async fn inbound_handoff_request_does_not_block_realtime_event_forwarding() -> R
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -4234,6 +4314,8 @@ async fn inbound_handoff_request_steers_active_turn() -> Result<()> {
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
@@ -4391,6 +4473,8 @@ async fn inbound_handoff_request_starts_turn_and_does_not_block_realtime_audio()
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,