agent/codex
1
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Update realtime handoff transcript handling (#18597)

Browse Source 
## Summary

This PR aims to improve integration between the realtime model and the
codex agent by sharing more context with each other. In particular, we
now share full realtime conversation transcript deltas in addition to
the delegation message.

realtime_conversation.rs now turns a handoff into:
```
<realtime_delegation>
  <input>...</input>
  <transcript_delta>...</transcript_delta>
</realtime_delegation>
```

## Implementation notes

The transcript is accumulated in the realtime websocket layer as parsed
realtime events arrive. When a background-agent handoff is requested,
the current transcript snapshot is copied onto the handoff event and
then serialized by `realtime_conversation.rs` into the hidden realtime
delegation envelope that Codex receives as user-turn context.

For Realtime V2, the session now explicitly enables input audio
transcription, and the parser handles the relevant input/output
transcript completion events so the snapshot includes both user speech
and realtime model responses. The delegation `<input>` remains the
actual handoff request, while `<transcript_delta>` carries the
surrounding conversation history for context.

Reviewers should note that the transcript payload is intended for Codex
context sharing, not UI rendering. The realtime delegation envelope
should stay hidden from the user-facing transcript surface, while still
being included in the background-agent turn so Codex can answer with the
same conversational context the realtime model had.
This commit is contained in:
guinness-oai
2026-04-20 14:04:09 -07:00
committed by GitHub
Unverified
parent 14ebfbced9
commit 126bd6e7a8
11 changed files with 572 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/tests/suite/v2/realtime_conversation.rs
+89 -12
View File
@@ -458,7 +458,10 @@ fn v2_background_agent_tool_call(call_id: &str, prompt: &str) -> Value {
async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
skip_if_no_network!(Ok(()));
let responses_server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
let responses_server = create_mock_responses_server_sequence_unchecked(vec![
create_final_assistant_message_sse_response("delegated")?,
])
.await;
let realtime_server = start_websocket_server(vec![vec![
vec![json!({
"type": "session.updated",
@@ -489,6 +492,10 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
"type": "response.output_text.delta",
"delta": "working"
}),
json!({
"type": "response.output_text.done",
"text": "working on it"
}),
json!({
"type": "conversation.item.done",
"item": {
@@ -677,7 +684,13 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
handoff_item_added.item["input_transcript"],
json!("delegate now")
);
assert_eq!(handoff_item_added.item["active_transcript"], json!([]));
assert_eq!(
handoff_item_added.item["active_transcript"],
json!([
{"role": "user", "text": "delegate now"},
{"role": "assistant", "text": "working on it"}
])
);
let realtime_error =
read_notification::<ThreadRealtimeErrorNotification>(&mut mcp, "thread/realtime/error")
@@ -1142,7 +1155,7 @@ async fn realtime_webrtc_start_emits_sdp_notification() -> Result<()> {
Some("multipart/form-data; boundary=codex-realtime-call-boundary")
);
let body = String::from_utf8(request.body).context("multipart body should be utf-8")?;
let session = r#"{"tool_choice":"auto","type":"realtime","model":"gpt-realtime-1.5","instructions":"backend prompt\n\nstartup context","output_modalities":["audio"],"audio":{"input":{"format":{"type":"audio/pcm","rate":24000},"noise_reduction":{"type":"near_field"},"turn_detection":{"type":"server_vad","interrupt_response":true,"create_response":true,"silence_duration_ms":500}},"output":{"format":{"type":"audio/pcm","rate":24000},"voice":"marin"}},"tools":[{"type":"function","name":"background_agent","description":"Send a user request to the background agent. Use this as the default action. Do not rephrase the user's ask or rewrite it in your own words; pass along the user's own words. If the background agent is idle, this starts a new task and returns the final result to the user. If the background agent is already working on a task, this sends the request as guidance to steer that previous task. If the user asks to do something next, later, after this, or once current work finishes, call this tool so the work is actually queued instead of merely promising to do it later.","parameters":{"type":"object","properties":{"prompt":{"type":"string","description":"The user request to delegate to the background agent."}},"required":["prompt"],"additionalProperties":false}}]}"#;
let session = r#"{"tool_choice":"auto","type":"realtime","model":"gpt-realtime-1.5","instructions":"backend prompt\n\nstartup context","output_modalities":["audio"],"audio":{"input":{"format":{"type":"audio/pcm","rate":24000},"noise_reduction":{"type":"near_field"},"transcription":{"model":"gpt-4o-mini-transcribe"},"turn_detection":{"type":"server_vad","interrupt_response":true,"create_response":true,"silence_duration_ms":500}},"output":{"format":{"type":"audio/pcm","rate":24000},"voice":"marin"}},"tools":[{"type":"function","name":"background_agent","description":"Send a user request to the background agent. Use this as the default action. Do not rephrase the user's ask or rewrite it in your own words; pass along the user's own words. If the background agent is idle, this starts a new task and returns the final result to the user. If the background agent is already working on a task, this sends the request as guidance to steer that previous task. If the user asks to do something next, later, after this, or once current work finishes, call this tool so the work is actually queued instead of merely promising to do it later.","parameters":{"type":"object","properties":{"prompt":{"type":"string","description":"The user request to delegate to the background agent."}},"required":["prompt"],"additionalProperties":false}}]}"#;
let session = normalized_json_string(session)?;
assert_eq!(
body,
@@ -1241,8 +1254,16 @@ async fn webrtc_v1_handoff_request_delegates_and_appends_result() -> Result<()>
vec![
session_updated("sess_v1_handoff"),
json!({
"type": "conversation.input_transcript.delta",
"delta": "delegate from v1"
"type": "conversation.item.input_audio_transcription.completed",
"transcript": "delegate from v1"
}),
json!({
"type": "response.output_audio_transcript.delta",
"delta": "the secret word is "
}),
json!({
"type": "response.output_audio_transcript.delta",
"delta": "kumquat"
}),
json!({
"type": "conversation.handoff.requested",
@@ -1258,6 +1279,12 @@ async fn webrtc_v1_handoff_request_delegates_and_appends_result() -> Result<()>
let started = harness.start_webrtc_realtime("v=offer\r\n").await?;
assert_eq!(started.started.version, RealtimeConversationVersion::V1);
assert_call_create_multipart(
harness.call_capture.single_request(),
"v=offer\r\n",
v1_session_create_json(),
)?;
assert_v1_session_update(&harness.sideband_outbound_request(/*request_index*/ 0).await)?;
// Phase 2: wait for the delegated background agent turn that is launched by the handoff request.
let turn_started = harness
@@ -1274,11 +1301,13 @@ async fn webrtc_v1_handoff_request_delegates_and_appends_result() -> Result<()>
let requests = harness.main_loop_responses_requests().await?;
assert_eq!(requests.len(), 1);
assert!(
response_request_contains_text(&requests[0], "user: delegate from v1"),
"delegated Responses request should contain realtime text: {}",
response_request_contains_text(
&requests[0],
"<realtime_delegation>\n <input>delegate from v1</input>\n <transcript_delta>user: delegate from v1\nassistant: the secret word is kumquat</transcript_delta>\n</realtime_delegation>",
),
"delegated Responses request should contain realtime delegation envelope: {}",
requests[0]
);
let handoff_append = harness.sideband_outbound_request(/*request_index*/ 1).await;
assert_eq!(
handoff_append,
@@ -1526,7 +1555,34 @@ async fn webrtc_v2_background_agent_tool_call_delegates_and_returns_function_out
realtime_sideband(vec![realtime_sideband_connection(vec![
vec![
session_updated("sess_v2_tool"),
v2_background_agent_tool_call("call_v2", "delegate from v2"),
json!({
"type": "conversation.item.input_audio_transcription.completed",
"transcript": "Hi how are you"
}),
json!({
"type": "response.output_audio_transcript.done",
"transcript": "Doing well, what can I help you with?"
}),
json!({
"type": "conversation.item.input_audio_transcription.completed",
"transcript": "The secret word is strawberry"
}),
json!({
"type": "conversation.item.created",
"item": {
"type": "message",
"role": "user",
"content": [{
"type": "input_text",
"text": "<realtime_collaboration_update><voice_policy>silent_delegate</voice_policy></realtime_collaboration_update>"
}]
}
}),
json!({
"type": "response.output_audio_transcript.delta",
"delta": "Got it-strawberry. What's next on the menu?"
}),
v2_background_agent_tool_call("call_v2", "run ls"),
],
vec![],
vec![],
@@ -1553,8 +1609,16 @@ async fn webrtc_v2_background_agent_tool_call_delegates_and_returns_function_out
let requests = harness.main_loop_responses_requests().await?;
assert_eq!(requests.len(), 1);
assert!(
response_request_contains_text(&requests[0], "delegate from v2"),
"delegated Responses request should contain tool prompt: {}",
response_request_contains_text(
&requests[0],
"<realtime_delegation>\n <input>run ls</input>\n <transcript_delta>user: Hi how are you\nassistant: Doing well, what can I help you with?\nuser: The secret word is strawberry\nassistant: Got it-strawberry. What's next on the menu?\nuser: run ls</transcript_delta>\n</realtime_delegation>",
),
"delegated Responses request should contain realtime delegation envelope: {}",
requests[0]
);
assert!(
!response_request_contains_text(&requests[0], "<realtime_collaboration_update>"),
"delegated Responses request should not include realtime control injects: {}",
requests[0]
);
@@ -2068,7 +2132,16 @@ async fn responses_requests(server: &MockServer) -> Result<Vec<Value>> {
}
fn response_request_contains_text(request: &Value, text: &str) -> bool {
request.to_string().contains(text)
match request {
Value::String(value) => value.contains(text),
Value::Array(values) => values
.iter()
.any(|value| response_request_contains_text(value, text)),
Value::Object(map) => map
.values()
.any(|value| response_request_contains_text(value, text)),
Value::Null | Value::Bool(_) | Value::Number(_) => false,
}
}
fn realtime_tool_ok_command() -> Vec<String> {
@@ -2187,6 +2260,10 @@ fn assert_v2_session_update(request: &Value) -> Result<()> {
request["session"]["tools"][0]["name"].as_str(),
Some("background_agent")
);
assert_eq!(
request["session"]["audio"]["input"]["transcription"]["model"].as_str(),
Some("gpt-4o-mini-transcribe")
);
Ok(())
}
codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs
+259 -17
View File
@@ -217,6 +217,8 @@ pub struct RealtimeWebsocketEvents {
#[derive(Default)]
struct ActiveTranscriptState {
entries: Vec<RealtimeTranscriptEntry>,
new_input_entry: bool,
new_output_entry: bool,
}
impl RealtimeWebsocketConnection {
@@ -417,38 +419,76 @@ impl RealtimeWebsocketEvents {
async fn update_active_transcript(&self, event: &mut RealtimeEvent) {
let mut active_transcript = self.active_transcript.lock().await;
match event {
RealtimeEvent::InputAudioSpeechStarted(_) => {}
RealtimeEvent::InputAudioSpeechStarted(_) => {
active_transcript.new_input_entry = true;
}
RealtimeEvent::InputTranscriptDelta(RealtimeTranscriptDelta { delta, .. }) => {
append_transcript_delta(&mut active_transcript.entries, "user", delta);
let force_new = active_transcript.new_input_entry;
append_transcript_delta(&mut active_transcript.entries, "user", delta, force_new);
active_transcript.new_input_entry = false;
}
RealtimeEvent::OutputTranscriptDelta(RealtimeTranscriptDelta { delta, .. }) => {
append_transcript_delta(&mut active_transcript.entries, "assistant", delta);
let force_new = active_transcript.new_output_entry;
append_transcript_delta(
&mut active_transcript.entries,
"assistant",
delta,
force_new,
);
active_transcript.new_output_entry = false;
}
RealtimeEvent::InputTranscriptDone(done) => {
let force_new = active_transcript.new_input_entry;
apply_transcript_done(
&mut active_transcript.entries,
"user",
&done.text,
force_new,
);
active_transcript.new_input_entry = false;
}
RealtimeEvent::OutputTranscriptDone(done) => {
let force_new = active_transcript.new_output_entry;
apply_transcript_done(
&mut active_transcript.entries,
"assistant",
&done.text,
force_new,
);
active_transcript.new_output_entry = false;
}
RealtimeEvent::HandoffRequested(handoff) => {
if self.event_parser == RealtimeEventParser::V1 {
handoff.active_transcript = std::mem::take(&mut active_transcript.entries);
}
append_handoff_input(&mut active_transcript.entries, &handoff.input_transcript);
handoff.active_transcript = active_transcript.entries.clone();
active_transcript.new_input_entry = true;
active_transcript.new_output_entry = true;
}
RealtimeEvent::ResponseCreated(_) => {
active_transcript.new_output_entry = true;
}
RealtimeEvent::SessionUpdated { .. }
| RealtimeEvent::InputTranscriptDone(_)
| RealtimeEvent::OutputTranscriptDone(_)
| RealtimeEvent::AudioOut(_)
| RealtimeEvent::ResponseCreated(_)
| RealtimeEvent::ResponseCancelled(_)
| RealtimeEvent::ResponseDone(_)
| RealtimeEvent::ConversationItemAdded(_)
| RealtimeEvent::ConversationItemDone { .. }
| RealtimeEvent::ConversationItemAdded(_)
| RealtimeEvent::Error(_) => {}
}
}
}
fn append_transcript_delta(entries: &mut Vec<RealtimeTranscriptEntry>, role: &str, delta: &str) {
fn append_transcript_delta(
entries: &mut Vec<RealtimeTranscriptEntry>,
role: &str,
delta: &str,
force_new: bool,
) {
if delta.is_empty() {
return;
}
if let Some(last_entry) = entries.last_mut()
if !force_new
&& let Some(last_entry) = entries.last_mut()
&& last_entry.role == role
{
last_entry.text.push_str(delta);
@@ -461,6 +501,48 @@ fn append_transcript_delta(entries: &mut Vec<RealtimeTranscriptEntry>, role: &st
});
}
fn apply_transcript_done(
entries: &mut Vec<RealtimeTranscriptEntry>,
role: &str,
text: &str,
force_new: bool,
) {
if text.is_empty() {
return;
}
if !force_new
&& let Some(last_entry) = entries.last_mut()
&& last_entry.role == role
{
last_entry.text = text.to_string();
return;
}
entries.push(RealtimeTranscriptEntry {
role: role.to_string(),
text: text.to_string(),
});
}
fn append_handoff_input(entries: &mut Vec<RealtimeTranscriptEntry>, input: &str) {
let input = input.trim();
if input.is_empty() || contains_transcript_entry(entries, "user", input) {
return;
}
entries.push(RealtimeTranscriptEntry {
role: "user".to_string(),
text: input.to_string(),
});
}
fn contains_transcript_entry(entries: &[RealtimeTranscriptEntry], role: &str, text: &str) -> bool {
entries
.iter()
.any(|entry| entry.role == role && entry.text.trim() == text.trim())
}
pub struct RealtimeWebsocketClient {
provider: Provider,
}
@@ -867,6 +949,44 @@ mod tests {
);
}
#[test]
fn parse_v1_input_audio_transcription_delta_event() {
let payload = json!({
"type": "conversation.item.input_audio_transcription.delta",
"item_id": "item_input_1",
"content_index": 0,
"delta": "hello"
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str(), RealtimeEventParser::V1),
Some(RealtimeEvent::InputTranscriptDelta(
RealtimeTranscriptDelta {
delta: "hello".to_string(),
}
))
);
}
#[test]
fn parse_v1_input_audio_transcription_completed_event() {
let payload = json!({
"type": "conversation.item.input_audio_transcription.completed",
"item_id": "item_input_1",
"content_index": 0,
"transcript": "hello world"
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str(), RealtimeEventParser::V1),
Some(RealtimeEvent::InputTranscriptDone(RealtimeTranscriptDone {
text: "hello world".to_string(),
}))
);
}
#[test]
fn parse_output_transcript_delta_event() {
let payload = json!({
@@ -885,6 +1005,66 @@ mod tests {
);
}
#[test]
fn parse_v1_output_audio_transcript_delta_event() {
let payload = json!({
"type": "response.output_audio_transcript.delta",
"delta": "hi"
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str(), RealtimeEventParser::V1),
Some(RealtimeEvent::OutputTranscriptDelta(
RealtimeTranscriptDelta {
delta: "hi".to_string(),
}
))
);
}
#[test]
fn parse_v1_output_audio_transcript_done_event() {
let payload = json!({
"type": "response.output_audio_transcript.done",
"transcript": "hi there"
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str(), RealtimeEventParser::V1),
Some(RealtimeEvent::OutputTranscriptDone(
RealtimeTranscriptDone {
text: "hi there".to_string(),
}
))
);
}
#[test]
fn parse_v1_item_done_output_text_event() {
let payload = json!({
"type": "conversation.item.done",
"item": {
"id": "item_output_1",
"type": "message",
"role": "assistant",
"content": [
{"type": "output_text", "text": "hello"},
{"type": "output_text", "text": " world"}
]
}
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str(), RealtimeEventParser::V1),
Some(RealtimeEvent::ConversationItemDone {
item_id: "item_output_1".to_string(),
})
);
}
#[test]
fn parse_realtime_v2_handoff_tool_call_event() {
let payload = json!({
@@ -930,6 +1110,58 @@ mod tests {
);
}
#[test]
fn parse_realtime_v2_output_audio_transcript_done_event() {
let payload = json!({
"type": "response.output_audio_transcript.done",
"transcript": "hello there"
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str(), RealtimeEventParser::RealtimeV2),
Some(RealtimeEvent::OutputTranscriptDone(
RealtimeTranscriptDone {
text: "hello there".to_string(),
}
))
);
}
#[test]
fn parse_realtime_v2_output_text_done_event() {
let payload = json!({
"type": "response.output_text.done",
"text": "hello there"
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str(), RealtimeEventParser::RealtimeV2),
Some(RealtimeEvent::OutputTranscriptDone(
RealtimeTranscriptDone {
text: "hello there".to_string(),
}
))
);
}
#[test]
fn parse_realtime_v2_conversation_item_created_event() {
let payload = json!({
"type": "conversation.item.created",
"item": {"type": "message", "role": "user"}
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str(), RealtimeEventParser::RealtimeV2),
Some(RealtimeEvent::ConversationItemAdded(
json!({"type": "message", "role": "user"})
))
);
}
#[test]
fn parse_realtime_v2_item_done_output_text_event() {
let payload = json!({
@@ -948,11 +1180,9 @@ mod tests {
assert_eq!(
parse_realtime_event(payload.as_str(), RealtimeEventParser::RealtimeV2),
Some(RealtimeEvent::OutputTranscriptDone(
RealtimeTranscriptDone {
text: "hello world".to_string(),
}
))
Some(RealtimeEvent::ConversationItemDone {
item_id: "item_output_1".to_string(),
})
);
}
@@ -1582,6 +1812,12 @@ mod tests {
"type": "near_field",
})
);
assert_eq!(
first_json["session"]["audio"]["input"]["transcription"],
json!({
"model": "gpt-4o-mini-transcribe",
})
);
assert_eq!(
first_json["session"]["audio"]["input"]["turn_detection"],
json!({
@@ -1756,6 +1992,12 @@ mod tests {
Value::String("transcription".to_string())
);
assert!(first_json["session"].get("instructions").is_none());
assert_eq!(
first_json["session"]["audio"]["input"]["transcription"],
json!({
"model": "gpt-4o-mini-transcribe",
})
);
assert!(first_json["session"]["audio"].get("output").is_none());
assert!(first_json["session"].get("tools").is_none());
codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v1.rs
+1
View File
@@ -55,6 +55,7 @@ pub(super) fn session_update_session(
rate: REALTIME_AUDIO_SAMPLE_RATE,
},
noise_reduction: None,
transcription: None,
turn_detection: None,
},
output: Some(SessionAudioOutput {
codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v2.rs
+8
View File
@@ -18,6 +18,7 @@ use crate::endpoint::realtime_websocket::protocol::SessionAudioInput;
use crate::endpoint::realtime_websocket::protocol::SessionAudioOutput;
use crate::endpoint::realtime_websocket::protocol::SessionAudioOutputFormat;
use crate::endpoint::realtime_websocket::protocol::SessionFunctionTool;
use crate::endpoint::realtime_websocket::protocol::SessionInputAudioTranscription;
use crate::endpoint::realtime_websocket::protocol::SessionNoiseReduction;
use crate::endpoint::realtime_websocket::protocol::SessionToolType;
use crate::endpoint::realtime_websocket::protocol::SessionTurnDetection;
@@ -31,6 +32,7 @@ const REALTIME_V2_OUTPUT_MODALITY_TEXT: &str = "text";
const REALTIME_V2_TOOL_CHOICE: &str = "auto";
const REALTIME_V2_BACKGROUND_AGENT_TOOL_NAME: &str = "background_agent";
const REALTIME_V2_BACKGROUND_AGENT_TOOL_DESCRIPTION: &str = "Send a user request to the background agent. Use this as the default action. Do not rephrase the user's ask or rewrite it in your own words; pass along the user's own words. If the background agent is idle, this starts a new task and returns the final result to the user. If the background agent is already working on a task, this sends the request as guidance to steer that previous task. If the user asks to do something next, later, after this, or once current work finishes, call this tool so the work is actually queued instead of merely promising to do it later.";
const REALTIME_V2_INPUT_TRANSCRIPTION_MODEL: &str = "gpt-4o-mini-transcribe";
pub(super) fn conversation_item_create_message(text: String) -> RealtimeOutboundMessage {
RealtimeOutboundMessage::ConversationItemCreate {
@@ -80,6 +82,9 @@ pub(super) fn session_update_session(
noise_reduction: Some(SessionNoiseReduction {
r#type: NoiseReductionType::NearField,
}),
transcription: Some(SessionInputAudioTranscription {
model: REALTIME_V2_INPUT_TRANSCRIPTION_MODEL.to_string(),
}),
turn_detection: Some(SessionTurnDetection {
r#type: TurnDetectionType::ServerVad,
interrupt_response: true,
@@ -126,6 +131,9 @@ pub(super) fn session_update_session(
rate: REALTIME_AUDIO_SAMPLE_RATE,
},
noise_reduction: None,
transcription: Some(SessionInputAudioTranscription {
model: REALTIME_V2_INPUT_TRANSCRIPTION_MODEL.to_string(),
}),
turn_detection: None,
},
output: None,
codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs
+7
View File
@@ -89,9 +89,16 @@ pub(super) struct SessionAudioInput {
#[serde(skip_serializing_if = "Option::is_none")]
pub(super) noise_reduction: Option<SessionNoiseReduction>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(super) transcription: Option<SessionInputAudioTranscription>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(super) turn_detection: Option<SessionTurnDetection>,
}
#[derive(Debug, Clone, Serialize)]
pub(super) struct SessionInputAudioTranscription {
pub(super) model: String,
}
#[derive(Debug, Clone, Serialize)]
pub(super) struct SessionAudioFormat {
#[serde(rename = "type")]
codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v1.rs
+23 -9
View File
@@ -2,6 +2,7 @@ use crate::endpoint::realtime_websocket::protocol_common::parse_error_event;
use crate::endpoint::realtime_websocket::protocol_common::parse_realtime_payload;
use crate::endpoint::realtime_websocket::protocol_common::parse_session_updated_event;
use crate::endpoint::realtime_websocket::protocol_common::parse_transcript_delta_event;
use crate::endpoint::realtime_websocket::protocol_common::parse_transcript_done_event;
use codex_protocol::protocol::RealtimeAudioFrame;
use codex_protocol::protocol::RealtimeEvent;
use codex_protocol::protocol::RealtimeHandoffRequested;
@@ -38,23 +39,28 @@ pub(super) fn parse_realtime_event_v1(payload: &str) -> Option<RealtimeEvent> {
item_id: None,
}))
}
"conversation.input_transcript.delta" => {
"conversation.input_transcript.delta"
| "conversation.item.input_audio_transcription.delta" => {
parse_transcript_delta_event(&parsed, "delta").map(RealtimeEvent::InputTranscriptDelta)
}
"conversation.output_transcript.delta" => {
"conversation.item.input_audio_transcription.completed" => {
parse_transcript_done_event(&parsed, "transcript")
.map(RealtimeEvent::InputTranscriptDone)
}
"conversation.output_transcript.delta"
| "response.output_text.delta"
| "response.output_audio_transcript.delta" => {
parse_transcript_delta_event(&parsed, "delta").map(RealtimeEvent::OutputTranscriptDelta)
}
"response.output_audio_transcript.done" => {
parse_transcript_done_event(&parsed, "transcript")
.map(RealtimeEvent::OutputTranscriptDone)
}
"conversation.item.added" => parsed
.get("item")
.cloned()
.map(RealtimeEvent::ConversationItemAdded),
"conversation.item.done" => parsed
.get("item")
.and_then(Value::as_object)
.and_then(|item| item.get("id"))
.and_then(Value::as_str)
.map(str::to_string)
.map(|item_id| RealtimeEvent::ConversationItemDone { item_id }),
"conversation.item.done" => parse_conversation_item_done_event(&parsed),
"conversation.handoff.requested" => {
let handoff_id = parsed
.get("handoff_id")
@@ -82,3 +88,11 @@ pub(super) fn parse_realtime_event_v1(payload: &str) -> Option<RealtimeEvent> {
}
}
}
fn parse_conversation_item_done_event(parsed: &Value) -> Option<RealtimeEvent> {
let item = parsed.get("item")?.as_object()?;
item.get("id")
.and_then(Value::as_str)
.map(str::to_string)
.map(|item_id| RealtimeEvent::ConversationItemDone { item_id })
}
codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs
+8 -33
View File
@@ -10,7 +10,6 @@ use codex_protocol::protocol::RealtimeInputAudioSpeechStarted;
use codex_protocol::protocol::RealtimeResponseCancelled;
use codex_protocol::protocol::RealtimeResponseCreated;
use codex_protocol::protocol::RealtimeResponseDone;
use codex_protocol::protocol::RealtimeTranscriptDone;
use serde_json::Map as JsonMap;
use serde_json::Value;
use tracing::debug;
@@ -38,6 +37,13 @@ pub(super) fn parse_realtime_event_v2(payload: &str) -> Option<RealtimeEvent> {
"response.output_text.delta" | "response.output_audio_transcript.delta" => {
parse_transcript_delta_event(&parsed, "delta").map(RealtimeEvent::OutputTranscriptDelta)
}
"response.output_text.done" => {
parse_transcript_done_event(&parsed, "text").map(RealtimeEvent::OutputTranscriptDone)
}
"response.output_audio_transcript.done" => {
parse_transcript_done_event(&parsed, "transcript")
.map(RealtimeEvent::OutputTranscriptDone)
}
"input_audio_buffer.speech_started" => Some(RealtimeEvent::InputAudioSpeechStarted(
RealtimeInputAudioSpeechStarted {
item_id: parsed
@@ -46,7 +52,7 @@ pub(super) fn parse_realtime_event_v2(payload: &str) -> Option<RealtimeEvent> {
.map(str::to_string),
},
)),
"conversation.item.added" => parsed
"conversation.item.added" | "conversation.item.created" => parsed
.get("item")
.cloned()
.map(RealtimeEvent::ConversationItemAdded),
@@ -122,43 +128,12 @@ fn parse_conversation_item_done_event(parsed: &Value) -> Option<RealtimeEvent> {
return Some(handoff);
}
if let Some(transcript_done) = parse_item_done_transcript(item) {
return Some(transcript_done);
}
item.get("id")
.and_then(Value::as_str)
.map(str::to_string)
.map(|item_id| RealtimeEvent::ConversationItemDone { item_id })
}
fn parse_item_done_transcript(item: &JsonMap<String, Value>) -> Option<RealtimeEvent> {
let role = item.get("role").and_then(Value::as_str)?;
let text = item
.get("content")
.and_then(Value::as_array)?
.iter()
.filter_map(item_content_text)
.collect::<String>();
if text.is_empty() {
return None;
}
let done = RealtimeTranscriptDone { text };
match role {
"user" => Some(RealtimeEvent::InputTranscriptDone(done)),
"assistant" => Some(RealtimeEvent::OutputTranscriptDone(done)),
_ => None,
}
}
fn item_content_text(content: &Value) -> Option<&str> {
content
.get("text")
.or_else(|| content.get("transcript"))
.and_then(Value::as_str)
}
fn parse_handoff_requested_event(item: &JsonMap<String, Value>) -> Option<RealtimeEvent> {
let item_type = item.get("type").and_then(Value::as_str);
let item_name = item.get("name").and_then(Value::as_str);
codex-rs/codex-api/tests/realtime_websocket_e2e.rs
+84 -1
View File
@@ -12,6 +12,9 @@ use codex_api::RealtimeSessionMode;
use codex_api::RealtimeWebsocketClient;
use codex_api::RetryConfig;
use codex_protocol::protocol::RealtimeHandoffRequested;
use codex_protocol::protocol::RealtimeTranscriptDelta;
use codex_protocol::protocol::RealtimeTranscriptDone;
use codex_protocol::protocol::RealtimeTranscriptEntry;
use codex_protocol::protocol::RealtimeVoice;
use futures::SinkExt;
use futures::StreamExt;
@@ -493,6 +496,46 @@ async fn realtime_ws_e2e_realtime_v2_parser_emits_handoff_requested() {
let first_json: Value = serde_json::from_str(&first).expect("json");
assert_eq!(first_json["type"], "session.update");
ws.send(Message::Text(
json!({
"type": "conversation.item.input_audio_transcription.completed",
"transcript": "delegate now"
})
.to_string()
.into(),
))
.await
.expect("send input transcript");
ws.send(Message::Text(
json!({
"type": "response.output_audio_transcript.delta",
"delta": "secret context"
})
.to_string()
.into(),
))
.await
.expect("send output transcript");
ws.send(Message::Text(
json!({
"type": "conversation.item.created",
"item": {
"type": "message",
"role": "user",
"content": [{
"type": "input_text",
"text": "<realtime_collaboration_update><voice_policy>silent_delegate</voice_policy></realtime_collaboration_update>"
}]
}
})
.to_string()
.into(),
))
.await
.expect("send control item echo");
ws.send(Message::Text(
json!({
"type": "conversation.item.done",
@@ -530,6 +573,37 @@ async fn realtime_ws_e2e_realtime_v2_parser_emits_handoff_requested() {
.await
.expect("connect");
let event = connection
.next_event()
.await
.expect("next event")
.expect("event");
assert_eq!(
event,
RealtimeEvent::InputTranscriptDone(RealtimeTranscriptDone {
text: "delegate now".to_string()
})
);
let event = connection
.next_event()
.await
.expect("next event")
.expect("event");
assert_eq!(
event,
RealtimeEvent::OutputTranscriptDelta(RealtimeTranscriptDelta {
delta: "secret context".to_string()
})
);
let event = connection
.next_event()
.await
.expect("next event")
.expect("event");
assert!(matches!(event, RealtimeEvent::ConversationItemAdded(_)));
let event = connection
.next_event()
.await
@@ -541,7 +615,16 @@ async fn realtime_ws_e2e_realtime_v2_parser_emits_handoff_requested() {
handoff_id: "call_123".to_string(),
item_id: "item_123".to_string(),
input_transcript: "delegate now".to_string(),
active_transcript: Vec::new(),
active_transcript: vec![
RealtimeTranscriptEntry {
role: "user".to_string(),
text: "delegate now".to_string(),
},
RealtimeTranscriptEntry {
role: "assistant".to_string(),
text: "secret context".to_string(),
},
],
})
);
codex-rs/core/src/realtime_conversation.rs
+28 -13
View File
@@ -831,16 +831,14 @@ async fn handle_start_inner(
}
let maybe_routed_text = match &event {
RealtimeEvent::HandoffRequested(handoff) => {
realtime_text_from_handoff_request(handoff)
realtime_delegation_from_handoff(handoff)
}
_ => 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(wrap_realtime_delegation_input(&text))
.await;
sess_for_routed_text.route_realtime_text_input(text).await;
}
if !fanout_realtime_active.load(Ordering::Relaxed) {
break;
@@ -890,23 +888,40 @@ pub(crate) async fn handle_audio(
}
}
fn realtime_text_from_handoff_request(handoff: &RealtimeHandoffRequested) -> Option<String> {
fn realtime_transcript_delta_from_handoff(handoff: &RealtimeHandoffRequested) -> Option<String> {
let active_transcript = handoff
.active_transcript
.iter()
.map(|entry| format!("{role}: {text}", role = entry.role, text = entry.text))
.collect::<Vec<_>>()
.join("\n");
(!active_transcript.is_empty())
.then_some(active_transcript)
.or((!handoff.input_transcript.is_empty()).then_some(handoff.input_transcript.clone()))
(!active_transcript.is_empty()).then_some(active_transcript)
}
fn wrap_realtime_delegation_input(input: &str) -> String {
format!(
"<realtime_delegation>\n <input>{}</input>\n</realtime_delegation>",
escape_xml_text(input)
)
fn realtime_text_from_handoff_request(handoff: &RealtimeHandoffRequested) -> Option<String> {
(!handoff.input_transcript.is_empty())
.then_some(handoff.input_transcript.clone())
.or_else(|| realtime_transcript_delta_from_handoff(handoff))
}
fn realtime_delegation_from_handoff(handoff: &RealtimeHandoffRequested) -> Option<String> {
let input = realtime_text_from_handoff_request(handoff)?;
Some(wrap_realtime_delegation_input(
&input,
realtime_transcript_delta_from_handoff(handoff).as_deref(),
))
}
fn wrap_realtime_delegation_input(input: &str, transcript_delta: Option<&str>) -> String {
let input = escape_xml_text(input);
if let Some(transcript_delta) = transcript_delta.filter(|text| !text.is_empty()) {
let transcript_delta = escape_xml_text(transcript_delta);
return format!(
"<realtime_delegation>\n <input>{input}</input>\n <transcript_delta>{transcript_delta}</transcript_delta>\n</realtime_delegation>"
);
}
format!("<realtime_delegation>\n <input>{input}</input>\n</realtime_delegation>")
}
fn escape_xml_text(input: &str) -> String {
codex-rs/core/src/realtime_conversation_tests.rs
+56 -4
View File
@@ -1,5 +1,6 @@
use super::RealtimeHandoffState;
use super::RealtimeSessionKind;
use super::realtime_delegation_from_handoff;
use super::realtime_text_from_handoff_request;
use super::wrap_realtime_delegation_input;
use async_channel::bounded;
@@ -8,7 +9,7 @@ use codex_protocol::protocol::RealtimeTranscriptEntry;
use pretty_assertions::assert_eq;
#[test]
fn extracts_text_from_handoff_request_active_transcript() {
fn prefers_handoff_input_transcript_over_active_transcript() {
let handoff = RealtimeHandoffRequested {
handoff_id: "handoff_1".to_string(),
item_id: "item_1".to_string(),
@@ -26,7 +27,50 @@ fn extracts_text_from_handoff_request_active_transcript() {
};
assert_eq!(
realtime_text_from_handoff_request(&handoff),
Some("user: hello\nassistant: hi there".to_string())
Some("ignored".to_string())
);
}
#[test]
fn extracts_text_from_handoff_request_active_transcript_if_input_missing() {
let handoff = RealtimeHandoffRequested {
handoff_id: "handoff_1".to_string(),
item_id: "item_1".to_string(),
input_transcript: String::new(),
active_transcript: vec![RealtimeTranscriptEntry {
role: "user".to_string(),
text: "hello".to_string(),
}],
};
assert_eq!(
realtime_text_from_handoff_request(&handoff),
Some("user: hello".to_string())
);
}
#[test]
fn wraps_handoff_with_transcript_delta() {
let handoff = RealtimeHandoffRequested {
handoff_id: "handoff_1".to_string(),
item_id: "item_1".to_string(),
input_transcript: "delegate this".to_string(),
active_transcript: vec![
RealtimeTranscriptEntry {
role: "user".to_string(),
text: "hello".to_string(),
},
RealtimeTranscriptEntry {
role: "assistant".to_string(),
text: "hi there".to_string(),
},
],
};
assert_eq!(
realtime_delegation_from_handoff(&handoff),
Some(
"<realtime_delegation>\n <input>delegate this</input>\n <transcript_delta>user: hello\nassistant: hi there</transcript_delta>\n</realtime_delegation>"
.to_string()
)
);
}
@@ -58,7 +102,7 @@ fn ignores_empty_handoff_request_input_transcript() {
#[test]
fn wraps_realtime_delegation_input() {
assert_eq!(
wrap_realtime_delegation_input("hello"),
wrap_realtime_delegation_input("hello", /*transcript_delta*/ None),
"<realtime_delegation>\n <input>hello</input>\n</realtime_delegation>"
);
}
@@ -66,7 +110,15 @@ fn wraps_realtime_delegation_input() {
#[test]
fn wraps_realtime_delegation_input_with_xml_escaping() {
assert_eq!(
wrap_realtime_delegation_input("use a < b && c > d"),
wrap_realtime_delegation_input("use a < b && c > d", Some("saw <that>")),
"<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>"
);
}
#[test]
fn wraps_realtime_delegation_input_with_xml_escaping_without_transcript() {
assert_eq!(
wrap_realtime_delegation_input("use a < b && c > d", /*transcript_delta*/ None),
"<realtime_delegation>\n <input>use a &lt; b &amp;&amp; c &gt; d</input>\n</realtime_delegation>"
);
}
codex-rs/core/tests/suite/realtime_conversation.rs
+9 -9
View File
@@ -2480,7 +2480,7 @@ async fn inbound_handoff_request_starts_turn() -> Result<()> {
let request = response_mock.single_request();
let user_texts = request.message_input_texts("user");
assert!(user_texts.iter().any(|text| text
== "<realtime_delegation>\n <input>user: text from realtime</input>\n</realtime_delegation>"));
== "<realtime_delegation>\n <input>text from realtime</input>\n <transcript_delta>user: text from realtime</transcript_delta>\n</realtime_delegation>"));
realtime_server.shutdown().await;
Ok(())
@@ -2562,14 +2562,14 @@ async fn inbound_handoff_request_uses_active_transcript() -> Result<()> {
let request = response_mock.single_request();
let user_texts = request.message_input_texts("user");
assert!(user_texts.iter().any(|text| text
== "<realtime_delegation>\n <input>assistant: assistant context\nuser: delegated query\nassistant: assist confirm</input>\n</realtime_delegation>"));
== "<realtime_delegation>\n <input>ignored</input>\n <transcript_delta>assistant: assistant context\nuser: delegated query\nassistant: assist confirm\nuser: ignored</transcript_delta>\n</realtime_delegation>"));
realtime_server.shutdown().await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn inbound_handoff_request_clears_active_transcript_after_each_handoff() -> Result<()> {
async fn inbound_handoff_request_keeps_transcript_history_after_each_handoff() -> Result<()> {
skip_if_no_network!(Ok(()));
let api_server = start_mock_server().await;
@@ -2677,13 +2677,13 @@ async fn inbound_handoff_request_clears_active_transcript_after_each_handoff() -
let first_user_texts = requests[0].message_input_texts("user");
assert!(first_user_texts.iter().any(|text| text
== "<realtime_delegation>\n <input>user: first question</input>\n</realtime_delegation>"));
== "<realtime_delegation>\n <input>first question</input>\n <transcript_delta>user: first question</transcript_delta>\n</realtime_delegation>"));
let second_user_texts = requests[1].message_input_texts("user");
assert!(second_user_texts.iter().any(|text| text
== "<realtime_delegation>\n <input>user: second question</input>\n</realtime_delegation>"));
== "<realtime_delegation>\n <input>second question</input>\n <transcript_delta>user: first question\nuser: second question</transcript_delta>\n</realtime_delegation>"));
assert!(!second_user_texts.iter().any(|text| text
== "<realtime_delegation>\n <input>user: first question\nuser: second question</input>\n</realtime_delegation>"));
== "<realtime_delegation>\n <input>second question</input>\n <transcript_delta>user: first questionsecond question</transcript_delta>\n</realtime_delegation>"));
realtime_server.shutdown().await;
Ok(())
@@ -3207,11 +3207,11 @@ async fn inbound_handoff_request_steers_active_turn() -> Result<()> {
!first_texts
.iter()
.any(|text| text
== "<realtime_delegation>\n <input>user: steer via realtime</input>\n</realtime_delegation>")
== "<realtime_delegation>\n <input>steer via realtime</input>\n <transcript_delta>user: steer via realtime</transcript_delta>\n</realtime_delegation>")
);
assert!(second_texts.iter().any(|text| text == "first prompt"));
assert!(second_texts.iter().any(|text| text
== "<realtime_delegation>\n <input>user: steer via realtime</input>\n</realtime_delegation>"));
== "<realtime_delegation>\n <input>steer via realtime</input>\n <transcript_delta>user: steer via realtime</transcript_delta>\n</realtime_delegation>"));
realtime_server.shutdown().await;
api_server.shutdown().await;
@@ -3327,7 +3327,7 @@ async fn inbound_handoff_request_starts_turn_and_does_not_block_realtime_audio()
let first_body: Value = serde_json::from_slice(&requests[0]).expect("parse first request");
let first_texts = message_input_texts(&first_body, "user");
let expected_text = format!(
"<realtime_delegation>\n <input>user: {delegated_text}</input>\n</realtime_delegation>"
"<realtime_delegation>\n <input>{delegated_text}</input>\n <transcript_delta>user: {delegated_text}</transcript_delta>\n</realtime_delegation>"
);
assert!(first_texts.iter().any(|text| text == &expected_text));