From f13917d50edeee34b45dd8f5063a283138d525c1 Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Wed, 4 Mar 2026 16:55:47 -0800 Subject: [PATCH] realtime: use codex tool handoff and fix audio playback --- .../tests/suite/v2/realtime_conversation.rs | 2 +- .../endpoint/realtime_websocket/methods.rs | 258 +++++++++++++++--- .../endpoint/realtime_websocket/protocol.rs | 201 ++++++++++---- .../codex-api/tests/realtime_websocket_e2e.rs | 35 ++- codex-rs/core/src/codex.rs | 3 + codex-rs/core/src/realtime_conversation.rs | 121 ++++++-- .../core/tests/suite/realtime_conversation.rs | 245 +++++++++++------ codex-rs/tui/src/chatwidget/realtime.rs | 2 +- 8 files changed, 667 insertions(+), 200 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs index 71150d7126..d906d74ef9 100644 --- a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs +++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs @@ -50,7 +50,7 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> { vec![], vec![ json!({ - "type": "conversation.output_audio.delta", + "type": "response.output_audio.delta", "delta": "AQID", "sample_rate": 24_000, "channels": 1, diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs index cb6583ded7..7fd0c2b979 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -8,6 +8,11 @@ use crate::endpoint::realtime_websocket::protocol::SessionAudio; use crate::endpoint::realtime_websocket::protocol::SessionAudioFormat; 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::SessionTool; +use crate::endpoint::realtime_websocket::protocol::SessionToolParameters; +use crate::endpoint::realtime_websocket::protocol::SessionToolProperties; +use crate::endpoint::realtime_websocket::protocol::SessionToolProperty; use crate::endpoint::realtime_websocket::protocol::SessionUpdateSession; use crate::endpoint::realtime_websocket::protocol::parse_realtime_event; use crate::error::ApiError; @@ -17,6 +22,7 @@ use futures::SinkExt; use futures::StreamExt; use http::HeaderMap; use http::HeaderValue; +use serde_json::json; use std::collections::HashMap; use std::sync::Arc; use std::sync::atomic::AtomicBool; @@ -220,6 +226,20 @@ impl RealtimeWebsocketConnection { .await } + pub async fn send_function_call_output( + &self, + call_id: String, + output_text: String, + ) -> Result<(), ApiError> { + self.writer + .send_function_call_output(call_id, output_text) + .await + } + + pub async fn send_response_create(&self) -> Result<(), ApiError> { + self.writer.send_response_create().await + } + pub async fn close(&self) -> Result<(), ApiError> { self.writer.close().await } @@ -263,11 +283,10 @@ impl RealtimeWebsocketWriter { pub async fn send_conversation_item_create(&self, text: String) -> Result<(), ApiError> { self.send_json(RealtimeOutboundMessage::ConversationItemCreate { - item: ConversationItem { - kind: "message".to_string(), + item: ConversationItem::Message { role: "user".to_string(), content: vec![ConversationItemContent { - kind: "text".to_string(), + kind: "input_text".to_string(), text, }], }, @@ -277,21 +296,47 @@ impl RealtimeWebsocketWriter { pub async fn send_conversation_handoff_append( &self, - handoff_id: String, + _handoff_id: String, output_text: String, ) -> Result<(), ApiError> { - self.send_json(RealtimeOutboundMessage::ConversationHandoffAppend { - handoff_id, - output_text, + self.send_json(RealtimeOutboundMessage::ConversationItemCreate { + item: ConversationItem::Message { + role: "assistant".to_string(), + content: vec![ConversationItemContent { + kind: "output_text".to_string(), + text: output_text, + }], + }, }) .await } + pub async fn send_function_call_output( + &self, + call_id: String, + output_text: String, + ) -> Result<(), ApiError> { + let output = json!({ + "content": output_text, + }) + .to_string(); + self.send_json(RealtimeOutboundMessage::ConversationItemCreate { + item: ConversationItem::FunctionCallOutput { call_id, output }, + }) + .await + } + + pub async fn send_response_create(&self) -> Result<(), ApiError> { + self.send_json(RealtimeOutboundMessage::ResponseCreate) + .await + } + pub async fn send_session_update(&self, instructions: String) -> Result<(), ApiError> { self.send_json(RealtimeOutboundMessage::SessionUpdate { session: SessionUpdateSession { - kind: "quicksilver".to_string(), + kind: "realtime".to_string(), instructions, + output_modalities: vec!["audio".to_string()], audio: SessionAudio { input: SessionAudioInput { format: SessionAudioFormat { @@ -300,9 +345,30 @@ impl RealtimeWebsocketWriter { }, }, output: SessionAudioOutput { - voice: "mundo".to_string(), + format: SessionAudioOutputFormat { + kind: "audio/pcm".to_string(), + }, + voice: "marin".to_string(), }, }, + tools: vec![SessionTool { + kind: "function".to_string(), + name: "codex".to_string(), + description: + "Delegate a request to Codex and return the final result to the user." + .to_string(), + parameters: SessionToolParameters { + kind: "object".to_string(), + properties: SessionToolProperties { + prompt: SessionToolProperty { + kind: "string".to_string(), + description: "The user request to delegate to Codex.".to_string(), + }, + }, + required: vec!["prompt".to_string()], + }, + }], + tool_choice: "auto".to_string(), }, }) .await @@ -510,15 +576,16 @@ fn websocket_url_from_api_url( } } - { + let has_additional_query_params = query_params + .is_some_and(|params| params.keys().any(|key| key != "model" || model.is_none())); + if model.is_some() || has_additional_query_params { let mut query = url.query_pairs_mut(); - query.append_pair("intent", "quicksilver"); if let Some(model) = model { query.append_pair("model", model); } if let Some(query_params) = query_params { for (key, value) in query_params { - if key == "intent" || (key == "model" && model.is_some()) { + if key == "model" && model.is_some() { continue; } query.append_pair(key, value); @@ -590,7 +657,7 @@ mod tests { #[test] fn parse_audio_delta_event() { let payload = json!({ - "type": "conversation.output_audio.delta", + "type": "response.output_audio.delta", "delta": "AAA=", "sample_rate": 48000, "channels": 1, @@ -608,6 +675,25 @@ mod tests { ); } + #[test] + fn parse_audio_delta_event_defaults_audio_shape() { + let payload = json!({ + "type": "response.output_audio.delta", + "delta": "AAA=" + }) + .to_string(); + + assert_eq!( + parse_realtime_event(payload.as_str()), + Some(RealtimeEvent::AudioOut(RealtimeAudioFrame { + data: "AAA=".to_string(), + sample_rate: 24_000, + num_channels: 1, + samples_per_channel: None, + })) + ); + } + #[test] fn parse_conversation_item_added_event() { let payload = json!({ @@ -641,13 +727,18 @@ mod tests { #[test] fn parse_handoff_requested_event() { let payload = json!({ - "type": "conversation.handoff.requested", - "handoff_id": "handoff_123", - "item_id": "item_123", - "input_transcript": "delegate this", - "messages": [ - {"role": "user", "text": "delegate this"} - ] + "type": "response.done", + "response": { + "output": [ + { + "id": "item_123", + "type": "function_call", + "name": "codex", + "call_id": "handoff_123", + "arguments": "{\"prompt\":\"delegate this\"}" + } + ] + } }) .to_string(); @@ -665,6 +756,24 @@ mod tests { ); } + #[test] + fn parse_unknown_event_as_conversation_item_added() { + let payload = json!({ + "type": "response.output_text.delta", + "delta": "hello", + "response_id": "resp_1", + }) + .to_string(); + assert_eq!( + parse_realtime_event(payload.as_str()), + Some(RealtimeEvent::ConversationItemAdded(json!({ + "type": "response.output_text.delta", + "delta": "hello", + "response_id": "resp_1", + }))) + ); + } + #[test] fn merge_request_headers_matches_http_precedence() { let mut provider_headers = HeaderMap::new(); @@ -702,10 +811,7 @@ mod tests { fn websocket_url_from_http_base_defaults_to_ws_path() { let url = websocket_url_from_api_url("http://127.0.0.1:8011", None, None).expect("build ws url"); - assert_eq!( - url.as_str(), - "ws://127.0.0.1:8011/v1/realtime?intent=quicksilver" - ); + assert_eq!(url.as_str(), "ws://127.0.0.1:8011/v1/realtime"); } #[test] @@ -715,7 +821,7 @@ mod tests { .expect("build ws url"); assert_eq!( url.as_str(), - "wss://example.com/v1/realtime?intent=quicksilver&model=realtime-test-model" + "wss://example.com/v1/realtime?model=realtime-test-model" ); } @@ -725,7 +831,7 @@ mod tests { .expect("build ws url"); assert_eq!( url.as_str(), - "wss://api.openai.com/v1/realtime?intent=quicksilver&model=snapshot" + "wss://api.openai.com/v1/realtime?model=snapshot" ); } @@ -736,7 +842,7 @@ mod tests { .expect("build ws url"); assert_eq!( url.as_str(), - "wss://example.com/openai/v1/realtime?intent=quicksilver&model=snapshot" + "wss://example.com/openai/v1/realtime?model=snapshot" ); } @@ -744,16 +850,13 @@ mod tests { fn websocket_url_preserves_existing_realtime_path_and_extra_query_params() { let url = websocket_url_from_api_url( "https://example.com/v1/realtime?foo=bar", - Some(&HashMap::from([ - ("trace".to_string(), "1".to_string()), - ("intent".to_string(), "ignored".to_string()), - ])), + Some(&HashMap::from([("trace".to_string(), "1".to_string())])), Some("snapshot"), ) .expect("build ws url"); assert_eq!( url.as_str(), - "wss://example.com/v1/realtime?foo=bar&intent=quicksilver&model=snapshot&trace=1" + "wss://example.com/v1/realtime?foo=bar&model=snapshot&trace=1" ); } @@ -777,12 +880,16 @@ mod tests { assert_eq!(first_json["type"], "session.update"); assert_eq!( first_json["session"]["type"], - Value::String("quicksilver".to_string()) + Value::String("realtime".to_string()) ); assert_eq!( first_json["session"]["instructions"], Value::String("backend prompt".to_string()) ); + assert_eq!( + first_json["session"]["output_modalities"][0], + Value::String("audio".to_string()) + ); assert_eq!( first_json["session"]["audio"]["input"]["format"]["type"], Value::String("audio/pcm".to_string()) @@ -791,9 +898,29 @@ mod tests { first_json["session"]["audio"]["input"]["format"]["rate"], Value::from(24_000) ); + assert_eq!( + first_json["session"]["audio"]["output"]["format"]["type"], + Value::String("audio/pcm".to_string()) + ); assert_eq!( first_json["session"]["audio"]["output"]["voice"], - Value::String("mundo".to_string()) + Value::String("marin".to_string()) + ); + assert_eq!( + first_json["session"]["tool_choice"], + Value::String("auto".to_string()) + ); + assert_eq!( + first_json["session"]["tools"][0]["type"], + Value::String("function".to_string()) + ); + assert_eq!( + first_json["session"]["tools"][0]["name"], + Value::String("codex".to_string()) + ); + assert_eq!( + first_json["session"]["tools"][0]["parameters"]["required"][0], + Value::String("prompt".to_string()) ); ws.send(Message::Text( @@ -836,13 +963,43 @@ mod tests { .into_text() .expect("text"); let fourth_json: Value = serde_json::from_str(&fourth).expect("json"); - assert_eq!(fourth_json["type"], "conversation.handoff.append"); - assert_eq!(fourth_json["handoff_id"], "handoff_1"); - assert_eq!(fourth_json["output_text"], "hello from codex"); + assert_eq!(fourth_json["type"], "conversation.item.create"); + assert_eq!(fourth_json["item"]["type"], "message"); + assert_eq!(fourth_json["item"]["role"], "assistant"); + assert_eq!( + fourth_json["item"]["content"][0]["type"], + Value::String("output_text".to_string()) + ); + assert_eq!( + fourth_json["item"]["content"][0]["text"], + Value::String("hello from codex".to_string()) + ); + + let fifth = ws + .next() + .await + .expect("fifth msg") + .expect("fifth msg ok") + .into_text() + .expect("text"); + let fifth_json: Value = serde_json::from_str(&fifth).expect("json"); + assert_eq!(fifth_json["type"], "conversation.item.create"); + assert_eq!(fifth_json["item"]["type"], "function_call_output"); + assert_eq!(fifth_json["item"]["call_id"], "handoff_1"); + + let sixth = ws + .next() + .await + .expect("sixth msg") + .expect("sixth msg ok") + .into_text() + .expect("text"); + let sixth_json: Value = serde_json::from_str(&sixth).expect("json"); + assert_eq!(sixth_json["type"], "response.create"); ws.send(Message::Text( json!({ - "type": "conversation.output_audio.delta", + "type": "response.output_audio.delta", "delta": "AQID", "sample_rate": 48000, "channels": 1 @@ -855,11 +1012,18 @@ mod tests { ws.send(Message::Text( json!({ - "type": "conversation.handoff.requested", - "handoff_id": "handoff_1", - "item_id": "item_2", - "input_transcript": "delegate now", - "messages": [{"role": "user", "text": "delegate now"}] + "type": "response.done", + "response": { + "output": [ + { + "id": "item_2", + "type": "function_call", + "name": "codex", + "call_id": "handoff_1", + "arguments": "{\"prompt\":\"delegate now\"}" + } + ] + } }) .to_string() .into(), @@ -929,6 +1093,14 @@ mod tests { ) .await .expect("send handoff"); + connection + .send_function_call_output("handoff_1".to_string(), "final from codex".to_string()) + .await + .expect("send function output"); + connection + .send_response_create() + .await + .expect("send response.create"); let audio_event = connection .next_event() diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs index 82957c5ec4..9a62163ff3 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs @@ -2,8 +2,10 @@ pub use codex_protocol::protocol::RealtimeAudioFrame; pub use codex_protocol::protocol::RealtimeEvent; pub use codex_protocol::protocol::RealtimeHandoffMessage; pub use codex_protocol::protocol::RealtimeHandoffRequested; +use serde::Deserialize; use serde::Serialize; use serde_json::Value; +use std::string::ToString; use tracing::debug; #[derive(Debug, Clone, PartialEq, Eq)] @@ -18,11 +20,8 @@ pub struct RealtimeSessionConfig { pub(super) enum RealtimeOutboundMessage { #[serde(rename = "input_audio_buffer.append")] InputAudioBufferAppend { audio: String }, - #[serde(rename = "conversation.handoff.append")] - ConversationHandoffAppend { - handoff_id: String, - output_text: String, - }, + #[serde(rename = "response.create")] + ResponseCreate, #[serde(rename = "session.update")] SessionUpdate { session: SessionUpdateSession }, #[serde(rename = "conversation.item.create")] @@ -34,7 +33,10 @@ pub(super) struct SessionUpdateSession { #[serde(rename = "type")] pub(super) kind: String, pub(super) instructions: String, + pub(super) output_modalities: Vec, pub(super) audio: SessionAudio, + pub(super) tools: Vec, + pub(super) tool_choice: String, } #[derive(Debug, Clone, Serialize)] @@ -57,15 +59,55 @@ pub(super) struct SessionAudioFormat { #[derive(Debug, Clone, Serialize)] pub(super) struct SessionAudioOutput { + pub(super) format: SessionAudioOutputFormat, pub(super) voice: String, } #[derive(Debug, Clone, Serialize)] -pub(super) struct ConversationItem { +pub(super) struct SessionAudioOutputFormat { #[serde(rename = "type")] pub(super) kind: String, - pub(super) role: String, - pub(super) content: Vec, +} + +#[derive(Debug, Clone, Serialize)] +pub(super) struct SessionTool { + #[serde(rename = "type")] + pub(super) kind: String, + pub(super) name: String, + pub(super) description: String, + pub(super) parameters: SessionToolParameters, +} + +#[derive(Debug, Clone, Serialize)] +pub(super) struct SessionToolParameters { + #[serde(rename = "type")] + pub(super) kind: String, + pub(super) properties: SessionToolProperties, + pub(super) required: Vec, +} + +#[derive(Debug, Clone, Serialize)] +pub(super) struct SessionToolProperties { + pub(super) prompt: SessionToolProperty, +} + +#[derive(Debug, Clone, Serialize)] +pub(super) struct SessionToolProperty { + #[serde(rename = "type")] + pub(super) kind: String, + pub(super) description: String, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(tag = "type")] +pub(super) enum ConversationItem { + #[serde(rename = "message")] + Message { + role: String, + content: Vec, + }, + #[serde(rename = "function_call_output")] + FunctionCallOutput { call_id: String, output: String }, } #[derive(Debug, Clone, Serialize)] @@ -92,7 +134,7 @@ pub(super) fn parse_realtime_event(payload: &str) -> Option { } }; match message_type { - "session.updated" => { + "session.created" | "session.updated" => { let session_id = parsed .get("session") .and_then(Value::as_object) @@ -110,7 +152,7 @@ pub(super) fn parse_realtime_event(payload: &str) -> Option { instructions, }) } - "conversation.output_audio.delta" => { + "response.output_audio.delta" => { let data = parsed .get("delta") .and_then(Value::as_str) @@ -119,12 +161,14 @@ pub(super) fn parse_realtime_event(payload: &str) -> Option { let sample_rate = parsed .get("sample_rate") .and_then(Value::as_u64) - .and_then(|v| u32::try_from(v).ok())?; + .and_then(|v| u32::try_from(v).ok()) + .unwrap_or(24_000); let num_channels = parsed .get("channels") .or_else(|| parsed.get("num_channels")) .and_then(Value::as_u64) - .and_then(|v| u16::try_from(v).ok())?; + .and_then(|v| u16::try_from(v).ok()) + .unwrap_or(1); Some(RealtimeEvent::AudioOut(RealtimeAudioFrame { data, sample_rate, @@ -146,35 +190,11 @@ pub(super) fn parse_realtime_event(payload: &str) -> Option { .and_then(Value::as_str) .map(str::to_string) .map(|item_id| RealtimeEvent::ConversationItemDone { item_id }), - "conversation.handoff.requested" => { - let handoff_id = parsed - .get("handoff_id") - .and_then(Value::as_str) - .map(str::to_string)?; - let item_id = parsed - .get("item_id") - .and_then(Value::as_str) - .map(str::to_string)?; - let input_transcript = parsed - .get("input_transcript") - .and_then(Value::as_str) - .map(str::to_string)?; - let messages = parsed - .get("messages") - .and_then(Value::as_array)? - .iter() - .filter_map(|message| { - let role = message.get("role").and_then(Value::as_str)?.to_string(); - let text = message.get("text").and_then(Value::as_str)?.to_string(); - Some(RealtimeHandoffMessage { role, text }) - }) - .collect(); - Some(RealtimeEvent::HandoffRequested(RealtimeHandoffRequested { - handoff_id, - item_id, - input_transcript, - messages, - })) + "response.done" => { + if let Some(handoff) = parse_handoff_requested(&parsed) { + return Some(RealtimeEvent::HandoffRequested(handoff)); + } + Some(RealtimeEvent::ConversationItemAdded(parsed)) } "error" => parsed .get("message") @@ -188,11 +208,100 @@ pub(super) fn parse_realtime_event(payload: &str) -> Option { .and_then(Value::as_str) .map(str::to_string) }) - .or_else(|| parsed.get("error").map(std::string::ToString::to_string)) + .or_else(|| parsed.get("error").map(ToString::to_string)) .map(RealtimeEvent::Error), - _ => { - debug!("received unsupported realtime event type: {message_type}, data: {payload}"); - None - } + _ => Some(RealtimeEvent::ConversationItemAdded(parsed)), } } + +fn parse_handoff_requested(parsed: &Value) -> Option { + let outputs = parsed + .get("response") + .and_then(Value::as_object) + .and_then(|response| response.get("output")) + .and_then(Value::as_array)?; + let function_call = outputs.iter().find(|item| { + item.get("type").and_then(Value::as_str) == Some("function_call") + && item.get("name").and_then(Value::as_str) == Some("codex") + })?; + let handoff_id = function_call + .get("call_id") + .and_then(Value::as_str) + .map(str::to_string)?; + let item_id = function_call + .get("id") + .and_then(Value::as_str) + .map(str::to_string) + .unwrap_or_else(|| handoff_id.clone()); + let arguments = function_call + .get("arguments") + .and_then(Value::as_str) + .unwrap_or_default(); + let (input_transcript, messages) = parse_handoff_arguments(arguments); + Some(RealtimeHandoffRequested { + handoff_id, + item_id, + input_transcript, + messages, + }) +} + +fn parse_handoff_arguments(arguments: &str) -> (String, Vec) { + #[derive(Debug, Deserialize)] + struct HandoffArguments { + #[serde(default)] + prompt: Option, + #[serde(default)] + text: Option, + #[serde(default)] + input: Option, + #[serde(default)] + message: Option, + #[serde(default)] + input_transcript: Option, + #[serde(default)] + messages: Vec, + } + + let Some(parsed) = serde_json::from_str::(arguments).ok() else { + return ( + arguments.to_string(), + vec![RealtimeHandoffMessage { + role: "user".to_string(), + text: arguments.to_string(), + }], + ); + }; + let messages = parsed + .messages + .into_iter() + .filter(|message| !message.text.is_empty()) + .collect::>(); + for value in [ + parsed.prompt, + parsed.text, + parsed.input, + parsed.message, + parsed.input_transcript, + ] + .into_iter() + .flatten() + { + if !value.is_empty() { + if messages.is_empty() { + return ( + value.clone(), + vec![RealtimeHandoffMessage { + role: "user".to_string(), + text: value, + }], + ); + } + return (value, messages); + } + } + if let Some(first_message) = messages.first() { + return (first_message.text.clone(), messages); + } + (String::new(), messages) +} diff --git a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs index aa11b79584..66b99e286c 100644 --- a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs +++ b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs @@ -81,7 +81,7 @@ async fn realtime_ws_e2e_session_create_and_event_flow() { assert_eq!(first_json["type"], "session.update"); assert_eq!( first_json["session"]["type"], - Value::String("quicksilver".to_string()) + Value::String("realtime".to_string()) ); assert_eq!( first_json["session"]["instructions"], @@ -95,6 +95,18 @@ async fn realtime_ws_e2e_session_create_and_event_flow() { first_json["session"]["audio"]["input"]["format"]["rate"], Value::from(24_000) ); + assert_eq!( + first_json["session"]["tool_choice"], + Value::String("auto".to_string()) + ); + assert_eq!( + first_json["session"]["tools"][0]["type"], + Value::String("function".to_string()) + ); + assert_eq!( + first_json["session"]["tools"][0]["name"], + Value::String("codex".to_string()) + ); ws.send(Message::Text( json!({ @@ -119,7 +131,7 @@ async fn realtime_ws_e2e_session_create_and_event_flow() { ws.send(Message::Text( json!({ - "type": "conversation.output_audio.delta", + "type": "response.output_audio.delta", "delta": "AQID", "sample_rate": 48000, "channels": 1 @@ -311,7 +323,7 @@ async fn realtime_ws_e2e_disconnected_emitted_once() { } #[tokio::test] -async fn realtime_ws_e2e_ignores_unknown_text_events() { +async fn realtime_ws_e2e_forwards_unknown_text_events() { let (addr, server) = spawn_realtime_ws_server(|mut ws: RealtimeWsStream| async move { let first = ws .next() @@ -361,13 +373,26 @@ async fn realtime_ws_e2e_ignores_unknown_text_events() { .await .expect("connect"); - let event = connection + let first_event = connection .next_event() .await .expect("next event") .expect("event"); assert_eq!( - event, + first_event, + RealtimeEvent::ConversationItemAdded(json!({ + "type": "response.created", + "response": {"id": "resp_unknown"} + })) + ); + + let second_event = connection + .next_event() + .await + .expect("next event") + .expect("event"); + assert_eq!( + second_event, RealtimeEvent::SessionUpdated { session_id: "sess_after_unknown".to_string(), instructions: Some("backend prompt".to_string()), diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 6cb7b0f616..3460e18a4f 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -2447,6 +2447,9 @@ impl Session { if !matches!(msg, EventMsg::TurnComplete(_)) { return; } + if let Err(err) = self.conversation.handoff_complete().await { + debug!("failed to send final realtime handoff tool output: {err}"); + } self.conversation.clear_active_handoff().await; } diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index e614be1061..00b8f5359e 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -43,6 +43,7 @@ const AUDIO_IN_QUEUE_CAPACITY: usize = 256; const USER_TEXT_IN_QUEUE_CAPACITY: usize = 64; const HANDOFF_OUT_QUEUE_CAPACITY: usize = 64; const OUTPUT_EVENTS_QUEUE_CAPACITY: usize = 256; +const DEFAULT_REALTIME_MODEL: &str = "gpt-realtime-1.5"; pub(crate) struct RealtimeConversationManager { state: Mutex>, @@ -52,12 +53,19 @@ pub(crate) struct RealtimeConversationManager { struct RealtimeHandoffState { output_tx: Sender, active_handoff: Arc>>, + last_output_text: Arc>>, } #[derive(Debug, PartialEq, Eq)] -struct HandoffOutput { - handoff_id: String, - output_text: String, +enum HandoffOutput { + TextUpdate { + handoff_id: String, + output_text: String, + }, + FinalToolCall { + call_id: String, + output_text: String, + }, } impl RealtimeHandoffState { @@ -65,6 +73,7 @@ impl RealtimeHandoffState { Self { output_tx, active_handoff: Arc::new(Mutex::new(None)), + last_output_text: Arc::new(Mutex::new(None)), } } @@ -72,9 +81,10 @@ impl RealtimeHandoffState { let Some(handoff_id) = self.active_handoff.lock().await.clone() else { return Ok(()); }; + *self.last_output_text.lock().await = Some(output_text.clone()); self.output_tx - .send(HandoffOutput { + .send(HandoffOutput::TextUpdate { handoff_id, output_text, }) @@ -82,6 +92,23 @@ impl RealtimeHandoffState { .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?; Ok(()) } + + async fn send_final_output(&self) -> CodexResult<()> { + let Some(call_id) = self.active_handoff.lock().await.clone() else { + return Ok(()); + }; + let Some(output_text) = self.last_output_text.lock().await.clone() else { + return Ok(()); + }; + self.output_tx + .send(HandoffOutput::FinalToolCall { + call_id, + output_text, + }) + .await + .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?; + Ok(()) + } } #[allow(dead_code)] @@ -232,6 +259,17 @@ impl RealtimeConversationManager { handoff.send_output(output_text).await } + pub(crate) async fn handoff_complete(&self) -> CodexResult<()> { + let handoff = { + let guard = self.state.lock().await; + guard.as_ref().map(|state| state.handoff.clone()) + }; + let Some(handoff) = handoff else { + return Ok(()); + }; + handoff.send_final_output().await + } + pub(crate) async fn active_handoff_id(&self) -> Option { let handoff = { let guard = self.state.lock().await; @@ -247,6 +285,7 @@ impl RealtimeConversationManager { }; if let Some(handoff) = handoff { *handoff.active_handoff.lock().await = None; + *handoff.last_output_text.lock().await = None; } } @@ -282,7 +321,7 @@ pub(crate) async fn handle_start( .experimental_realtime_ws_backend_prompt .clone() .unwrap_or(params.prompt); - let model = config.experimental_realtime_ws_model.clone(); + let model = Some(DEFAULT_REALTIME_MODEL.to_string()); let requested_session_id = params .session_id @@ -489,17 +528,39 @@ fn spawn_realtime_input_task( } handoff_output = handoff_output_rx.recv() => { match handoff_output { - Ok(HandoffOutput { - handoff_id, - output_text, - }) => { - if let Err(err) = writer - .send_conversation_handoff_append(handoff_id, output_text) - .await - { - let mapped_error = map_api_error(err); - warn!("failed to send handoff output: {mapped_error}"); - break; + Ok(handoff_output) => { + match handoff_output { + HandoffOutput::TextUpdate { + handoff_id, + output_text, + } => { + if let Err(err) = writer + .send_conversation_handoff_append(handoff_id, output_text) + .await + { + let mapped_error = map_api_error(err); + warn!("failed to send handoff output: {mapped_error}"); + break; + } + } + HandoffOutput::FinalToolCall { + call_id, + output_text, + } => { + if let Err(err) = writer + .send_function_call_output(call_id, output_text) + .await + { + let mapped_error = map_api_error(err); + warn!("failed to send handoff tool output: {mapped_error}"); + break; + } + if let Err(err) = writer.send_response_create().await { + let mapped_error = map_api_error(err); + warn!("failed to send handoff response.create: {mapped_error}"); + break; + } + } } } Err(_) => break, @@ -511,6 +572,7 @@ fn spawn_realtime_input_task( if let RealtimeEvent::HandoffRequested(handoff) = &event { *handoff_state.active_handoff.lock().await = Some(handoff.handoff_id.clone()); + *handoff_state.last_output_text.lock().await = None; } let should_stop = matches!(&event, RealtimeEvent::Error(_)); if events_tx.send(event).await.is_err() { @@ -670,7 +732,7 @@ mod tests { let output_1 = rx.recv().await.expect("recv"); assert_eq!( output_1, - HandoffOutput { + HandoffOutput::TextUpdate { handoff_id: "handoff_1".to_string(), output_text: "result".to_string(), } @@ -679,7 +741,7 @@ mod tests { let output_2 = rx.recv().await.expect("recv"); assert_eq!( output_2, - HandoffOutput { + HandoffOutput::TextUpdate { handoff_id: "handoff_1".to_string(), output_text: "result 2".to_string(), } @@ -692,4 +754,27 @@ mod tests { .expect("send"); assert!(rx.is_empty()); } + + #[tokio::test] + async fn sends_final_tool_call_output_for_active_handoff() { + let (tx, rx) = bounded(4); + let state = RealtimeHandoffState::new(tx); + *state.active_handoff.lock().await = Some("handoff_2".to_string()); + + state + .send_output("final text".to_string()) + .await + .expect("send"); + let _ = rx.recv().await.expect("recv text update"); + + state.send_final_output().await.expect("send final output"); + let final_output = rx.recv().await.expect("recv final output"); + assert_eq!( + final_output, + HandoffOutput::FinalToolCall { + call_id: "handoff_2".to_string(), + output_text: "final text".to_string(), + } + ); + } } diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index ff4e745dd2..7975fd9f3f 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -42,7 +42,7 @@ async fn conversation_start_audio_text_close_round_trip() -> Result<()> { vec![], vec![ json!({ - "type": "conversation.output_audio.delta", + "type": "response.output_audio.delta", "delta": "AQID", "sample_rate": 24000, "channels": 1 @@ -141,7 +141,7 @@ async fn conversation_start_audio_text_close_round_trip() -> Result<()> { ); assert_eq!( server.handshakes()[1].uri(), - "/v1/realtime?intent=quicksilver&model=realtime-test-model" + "/v1/realtime?model=gpt-realtime-1.5" ); let mut request_types = [ connection[1].body_json()["type"] @@ -387,7 +387,7 @@ async fn conversation_second_start_replaces_runtime() -> Result<()> { "session": { "id": "sess_new", "instructions": "new" } })], vec![json!({ - "type": "conversation.output_audio.delta", + "type": "response.output_audio.delta", "delta": "AQID", "sample_rate": 24000, "channels": 1 @@ -600,15 +600,11 @@ async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Re "type": "session.updated", "session": { "id": "sess_1", "instructions": "backend prompt" } }), - json!({ - "type": "conversation.handoff.requested", - "handoff_id": "handoff_1", - "item_id": "item_1", - "input_transcript": "delegate hello", - "messages": [{ "role": "user", "text": "delegate hello" }] - }), + realtime_handoff_requested_event("handoff_1", "item_1", "delegate hello"), ], vec![], + vec![], + vec![], ]]) .await; @@ -652,7 +648,7 @@ async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Re let deadline = tokio::time::Instant::now() + Duration::from_secs(2); while tokio::time::Instant::now() < deadline { let connections = realtime_server.connections(); - if connections.len() == 1 && connections[0].len() >= 2 { + if connections.len() == 1 && connections[0].len() >= 4 { break; } tokio::time::sleep(Duration::from_millis(10)).await; @@ -660,22 +656,46 @@ async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Re let realtime_connections = realtime_server.connections(); assert_eq!(realtime_connections.len(), 1); - assert_eq!(realtime_connections[0].len(), 2); + assert_eq!(realtime_connections[0].len(), 4); assert_eq!( realtime_connections[0][0].body_json()["type"].as_str(), Some("session.update") ); assert_eq!( realtime_connections[0][1].body_json()["type"].as_str(), - Some("conversation.handoff.append") + Some("conversation.item.create") ); assert_eq!( - realtime_connections[0][1].body_json()["handoff_id"].as_str(), + realtime_connections[0][1].body_json()["item"]["type"].as_str(), + Some("message") + ); + assert_eq!( + realtime_connections[0][1].body_json()["item"]["role"].as_str(), + Some("assistant") + ); + assert_eq!( + realtime_connections[0][1].body_json()["item"]["content"][0]["type"].as_str(), + Some("output_text") + ); + assert_eq!( + realtime_connections[0][1].body_json()["item"]["content"][0]["text"].as_str(), + Some("assistant says hi") + ); + assert_eq!( + realtime_connections[0][2].body_json()["type"].as_str(), + Some("conversation.item.create") + ); + assert_eq!( + realtime_connections[0][2].body_json()["item"]["type"].as_str(), + Some("function_call_output") + ); + assert_eq!( + realtime_connections[0][2].body_json()["item"]["call_id"].as_str(), Some("handoff_1") ); assert_eq!( - realtime_connections[0][1].body_json()["output_text"].as_str(), - Some("assistant says hi") + realtime_connections[0][3].body_json()["type"].as_str(), + Some("response.create") ); realtime_server.shutdown().await; @@ -719,19 +739,15 @@ async fn conversation_handoff_persists_across_item_done_until_turn_complete() -> "type": "session.updated", "session": { "id": "sess_item_done", "instructions": "backend prompt" } }), - json!({ - "type": "conversation.handoff.requested", - "handoff_id": "handoff_item_done", - "item_id": "item_item_done", - "input_transcript": "delegate now", - "messages": [{ "role": "user", "text": "delegate now" }] - }), + realtime_handoff_requested_event("handoff_item_done", "item_item_done", "delegate now"), ], vec![json!({ "type": "conversation.item.done", "item": { "id": "item_item_done" } })], vec![], + vec![], + vec![], ]]) .await; @@ -769,14 +785,22 @@ async fn conversation_handoff_persists_across_item_done_until_turn_complete() -> let first_append = realtime_server.wait_for_request(0, 1).await; assert_eq!( first_append.body_json()["type"].as_str(), - Some("conversation.handoff.append") + Some("conversation.item.create") ); assert_eq!( - first_append.body_json()["handoff_id"].as_str(), - Some("handoff_item_done") + first_append.body_json()["item"]["type"].as_str(), + Some("message") ); assert_eq!( - first_append.body_json()["output_text"].as_str(), + first_append.body_json()["item"]["role"].as_str(), + Some("assistant") + ); + assert_eq!( + first_append.body_json()["item"]["content"][0]["type"].as_str(), + Some("output_text") + ); + assert_eq!( + first_append.body_json()["item"]["content"][0]["text"].as_str(), Some("assistant message 1") ); @@ -793,14 +817,22 @@ async fn conversation_handoff_persists_across_item_done_until_turn_complete() -> let second_append = realtime_server.wait_for_request(0, 2).await; assert_eq!( second_append.body_json()["type"].as_str(), - Some("conversation.handoff.append") + Some("conversation.item.create") ); assert_eq!( - second_append.body_json()["handoff_id"].as_str(), - Some("handoff_item_done") + second_append.body_json()["item"]["type"].as_str(), + Some("message") ); assert_eq!( - second_append.body_json()["output_text"].as_str(), + second_append.body_json()["item"]["role"].as_str(), + Some("assistant") + ); + assert_eq!( + second_append.body_json()["item"]["content"][0]["type"].as_str(), + Some("output_text") + ); + assert_eq!( + second_append.body_json()["item"]["content"][0]["text"].as_str(), Some("assistant message 2") ); @@ -816,6 +848,30 @@ async fn conversation_handoff_persists_across_item_done_until_turn_complete() -> }) .await; + let final_tool_call = realtime_server.wait_for_request(0, 3).await; + assert_eq!( + final_tool_call.body_json()["type"].as_str(), + Some("conversation.item.create") + ); + assert_eq!( + final_tool_call.body_json()["item"]["type"].as_str(), + Some("function_call_output") + ); + assert_eq!( + final_tool_call.body_json()["item"]["call_id"].as_str(), + Some("handoff_item_done") + ); + assert_eq!( + final_tool_call.body_json()["item"]["output"].as_str(), + Some("{\"content\":\"assistant message 2\"}") + ); + + let response_create = realtime_server.wait_for_request(0, 4).await; + assert_eq!( + response_create.body_json()["type"].as_str(), + Some("response.create") + ); + realtime_server.shutdown().await; api_server.shutdown().await; Ok(()) @@ -825,6 +881,23 @@ fn sse_event(event: Value) -> String { responses::sse(vec![event]) } +fn realtime_handoff_requested_event(handoff_id: &str, item_id: &str, prompt: &str) -> Value { + json!({ + "type": "response.done", + "response": { + "output": [ + { + "id": item_id, + "type": "function_call", + "name": "codex", + "call_id": handoff_id, + "arguments": json!({ "prompt": prompt }).to_string(), + } + ] + } + }) +} + fn message_input_texts(body: &Value, role: &str) -> Vec { body.get("input") .and_then(Value::as_array) @@ -859,13 +932,7 @@ async fn inbound_handoff_request_starts_turn() -> Result<()> { "type": "session.updated", "session": { "id": "sess_inbound", "instructions": "backend prompt" } }), - json!({ - "type": "conversation.handoff.requested", - "handoff_id": "handoff_inbound", - "item_id": "item_inbound", - "input_transcript": "text from realtime", - "messages": [{ "role": "user", "text": "text from realtime" }] - }), + realtime_handoff_requested_event("handoff_inbound", "item_inbound", "text from realtime"), ]]]) .await; @@ -939,15 +1006,25 @@ async fn inbound_handoff_request_uses_all_messages() -> Result<()> { "session": { "id": "sess_inbound_multi", "instructions": "backend prompt" } }), json!({ - "type": "conversation.handoff.requested", - "handoff_id": "handoff_inbound_multi", - "item_id": "item_inbound_multi", - "input_transcript": "ignored", - "messages": [ - { "role": "assistant", "text": "assistant context" }, - { "role": "user", "text": "delegated query" }, - { "role": "assistant", "text": "assist confirm" }, - ] + "type": "response.done", + "response": { + "output": [ + { + "id": "item_inbound_multi", + "type": "function_call", + "name": "codex", + "call_id": "handoff_inbound_multi", + "arguments": json!({ + "input_transcript": "ignored", + "messages": [ + { "role": "assistant", "text": "assistant context" }, + { "role": "user", "text": "delegated query" }, + { "role": "assistant", "text": "assist confirm" }, + ] + }).to_string(), + } + ] + } }), ]]]) .await; @@ -1012,7 +1089,7 @@ async fn inbound_conversation_item_does_not_start_turn_and_still_forwards_audio( } }), json!({ - "type": "conversation.output_audio.delta", + "type": "response.output_audio.delta", "delta": "AQID", "sample_rate": 24000, "channels": 1 @@ -1102,13 +1179,11 @@ async fn delegated_turn_user_role_echo_does_not_redelegate_and_still_forwards_au "type": "session.updated", "session": { "id": "sess_echo_guard", "instructions": "backend prompt" } }), - json!({ - "type": "conversation.handoff.requested", - "handoff_id": "handoff_echo_guard", - "item_id": "item_echo_guard", - "input_transcript": "delegate now", - "messages": [{"role": "user", "text": "delegate now"}] - }), + realtime_handoff_requested_event( + "handoff_echo_guard", + "item_echo_guard", + "delegate now", + ), ], vec![ json!({ @@ -1120,7 +1195,7 @@ async fn delegated_turn_user_role_echo_does_not_redelegate_and_still_forwards_au } }), json!({ - "type": "conversation.output_audio.delta", + "type": "response.output_audio.delta", "delta": "AQID", "sample_rate": 24000, "channels": 1 @@ -1168,22 +1243,30 @@ async fn delegated_turn_user_role_echo_does_not_redelegate_and_still_forwards_au let mirrored_request = realtime_server.wait_for_request(0, 1).await; let mirrored_request_body = mirrored_request.body_json(); eprintln!( - "[realtime test +{}ms] saw mirrored request type={:?} handoff_id={:?} text={:?}", + "[realtime test +{}ms] saw mirrored request type={:?} role={:?} text={:?}", start.elapsed().as_millis(), mirrored_request_body["type"].as_str(), - mirrored_request_body["handoff_id"].as_str(), - mirrored_request_body["output_text"].as_str(), + mirrored_request_body["item"]["role"].as_str(), + mirrored_request_body["item"]["content"][0]["text"].as_str(), ); assert_eq!( mirrored_request_body["type"].as_str(), - Some("conversation.handoff.append") + Some("conversation.item.create") ); assert_eq!( - mirrored_request_body["handoff_id"].as_str(), - Some("handoff_echo_guard") + mirrored_request_body["item"]["type"].as_str(), + Some("message") ); assert_eq!( - mirrored_request_body["output_text"].as_str(), + mirrored_request_body["item"]["role"].as_str(), + Some("assistant") + ); + assert_eq!( + mirrored_request_body["item"]["content"][0]["type"].as_str(), + Some("output_text") + ); + assert_eq!( + mirrored_request_body["item"]["content"][0]["text"].as_str(), Some("assistant says hi") ); @@ -1250,15 +1333,13 @@ async fn inbound_handoff_request_does_not_block_realtime_event_forwarding() -> R "type": "session.updated", "session": { "id": "sess_non_blocking", "instructions": "backend prompt" } }), + realtime_handoff_requested_event( + "handoff_non_blocking", + "item_non_blocking", + "delegate now", + ), json!({ - "type": "conversation.handoff.requested", - "handoff_id": "handoff_non_blocking", - "item_id": "item_non_blocking", - "input_transcript": "delegate now", - "messages": [{"role": "user", "text": "delegate now"}] - }), - json!({ - "type": "conversation.output_audio.delta", + "type": "response.output_audio.delta", "delta": "AQID", "sample_rate": 24000, "channels": 1 @@ -1377,13 +1458,11 @@ async fn inbound_handoff_request_steers_active_turn() -> Result<()> { "type": "session.updated", "session": { "id": "sess_steer", "instructions": "backend prompt" } })], - vec![json!({ - "type": "conversation.handoff.requested", - "handoff_id": "handoff_steer", - "item_id": "item_steer", - "input_transcript": "steer via realtime", - "messages": [{ "role": "user", "text": "steer via realtime" }] - })], + vec![realtime_handoff_requested_event( + "handoff_steer", + "item_steer", + "steer via realtime", + )], ]]) .await; @@ -1500,15 +1579,9 @@ async fn inbound_handoff_request_starts_turn_and_does_not_block_realtime_audio() "type": "session.updated", "session": { "id": "sess_handoff_request", "instructions": "backend prompt" } }), + realtime_handoff_requested_event("handoff_audio", "item_audio", delegated_text), json!({ - "type": "conversation.handoff.requested", - "handoff_id": "handoff_audio", - "item_id": "item_audio", - "input_transcript": delegated_text, - "messages": [{ "role": "user", "text": delegated_text }] - }), - json!({ - "type": "conversation.output_audio.delta", + "type": "response.output_audio.delta", "delta": "AQID", "sample_rate": 24000, "channels": 1 diff --git a/codex-rs/tui/src/chatwidget/realtime.rs b/codex-rs/tui/src/chatwidget/realtime.rs index 32f0a15287..9e62694510 100644 --- a/codex-rs/tui/src/chatwidget/realtime.rs +++ b/codex-rs/tui/src/chatwidget/realtime.rs @@ -304,7 +304,7 @@ impl ChatWidget { } } - #[cfg(not(target_os = "linux"))] + #[cfg(all(not(target_os = "linux"), feature = "voice-input"))] fn start_realtime_local_audio(&mut self) { if self.realtime_conversation.capture_stop_flag.is_some() { return;