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 4abf680c1d..8b15ae56c6 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -1,27 +1,12 @@ -use crate::endpoint::realtime_websocket::protocol::ConversationItem; -use crate::endpoint::realtime_websocket::protocol::ConversationItemContent; -use crate::endpoint::realtime_websocket::protocol::RealtimeApiMode; -use crate::endpoint::realtime_websocket::protocol::RealtimeAudioFrame; -use crate::endpoint::realtime_websocket::protocol::RealtimeEvent; -use crate::endpoint::realtime_websocket::protocol::RealtimeOutboundMessage; -use crate::endpoint::realtime_websocket::protocol::RealtimeSessionConfig; -use crate::endpoint::realtime_websocket::protocol::SessionAudioFormat; -use crate::endpoint::realtime_websocket::protocol::SessionAudioInputV1; -use crate::endpoint::realtime_websocket::protocol::SessionAudioInputV2; -use crate::endpoint::realtime_websocket::protocol::SessionAudioOutputFormat; -use crate::endpoint::realtime_websocket::protocol::SessionAudioOutputV1; -use crate::endpoint::realtime_websocket::protocol::SessionAudioOutputV2; -use crate::endpoint::realtime_websocket::protocol::SessionAudioV1; -use crate::endpoint::realtime_websocket::protocol::SessionAudioV2; -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::SessionTurnDetection; -use crate::endpoint::realtime_websocket::protocol::SessionUpdateSession; -use crate::endpoint::realtime_websocket::protocol::SessionUpdateSessionV1; -use crate::endpoint::realtime_websocket::protocol::SessionUpdateSessionV2; -use crate::endpoint::realtime_websocket::protocol::parse_realtime_event; +use crate::endpoint::realtime_websocket::mode_v1; +use crate::endpoint::realtime_websocket::mode_v2; +use crate::endpoint::realtime_websocket::protocol_v1; +use crate::endpoint::realtime_websocket::protocol_v2; +use crate::endpoint::realtime_websocket::types::RealtimeApiMode; +use crate::endpoint::realtime_websocket::types::RealtimeAudioFrame; +use crate::endpoint::realtime_websocket::types::RealtimeEvent; +use crate::endpoint::realtime_websocket::types::RealtimeOutboundMessage; +use crate::endpoint::realtime_websocket::types::RealtimeSessionConfig; use crate::error::ApiError; use crate::provider::Provider; use codex_utils_rustls_provider::ensure_rustls_crypto_provider; @@ -29,7 +14,7 @@ use futures::SinkExt; use futures::StreamExt; use http::HeaderMap; use http::HeaderValue; -use serde_json::json; +use http::header::Entry; use std::collections::HashMap; use std::sync::Arc; use std::sync::atomic::AtomicBool; @@ -294,17 +279,11 @@ impl RealtimeWebsocketWriter { } pub async fn send_conversation_item_create(&self, text: String) -> Result<(), ApiError> { - let kind = match self.mode { - RealtimeApiMode::V1 => "text".to_string(), - RealtimeApiMode::V2 => "input_text".to_string(), + let message = match self.mode { + RealtimeApiMode::V1 => mode_v1::conversation_item_create(text), + RealtimeApiMode::V2 => mode_v2::conversation_item_create(text), }; - self.send_json(RealtimeOutboundMessage::ConversationItemCreate { - item: ConversationItem::Message { - role: "user".to_string(), - content: vec![ConversationItemContent { kind, text }], - }, - }) - .await + self.send_json(message).await } pub async fn send_conversation_handoff_append( @@ -312,27 +291,11 @@ impl RealtimeWebsocketWriter { handoff_id: String, output_text: String, ) -> Result<(), ApiError> { - match self.mode { - RealtimeApiMode::V1 => { - self.send_json(RealtimeOutboundMessage::ConversationHandoffAppend { - handoff_id, - output_text, - }) - .await - } - RealtimeApiMode::V2 => { - self.send_json(RealtimeOutboundMessage::ConversationItemCreate { - item: ConversationItem::Message { - role: "assistant".to_string(), - content: vec![ConversationItemContent { - kind: "output_text".to_string(), - text: output_text, - }], - }, - }) - .await - } - } + let message = match self.mode { + RealtimeApiMode::V1 => mode_v1::handoff_append(handoff_id, output_text), + RealtimeApiMode::V2 => mode_v2::handoff_append(output_text), + }; + self.send_json(message).await } pub async fn send_function_call_output( @@ -343,14 +306,8 @@ impl RealtimeWebsocketWriter { match self.mode { RealtimeApiMode::V1 => Ok(()), RealtimeApiMode::V2 => { - let output = json!({ - "content": output_text, - }) - .to_string(); - self.send_json(RealtimeOutboundMessage::ConversationItemCreate { - item: ConversationItem::FunctionCallOutput { call_id, output }, - }) - .await + self.send_json(mode_v2::function_call_output(call_id, output_text)) + .await } } } @@ -366,69 +323,11 @@ impl RealtimeWebsocketWriter { } pub async fn send_session_update(&self, instructions: String) -> Result<(), ApiError> { - let session = match self.mode { - RealtimeApiMode::V1 => SessionUpdateSession::V1(SessionUpdateSessionV1 { - kind: "quicksilver".to_string(), - instructions, - audio: SessionAudioV1 { - input: SessionAudioInputV1 { - format: SessionAudioFormat { - kind: "audio/pcm".to_string(), - rate: 24_000, - }, - }, - output: SessionAudioOutputV1 { - voice: "mundo".to_string(), - }, - }, - }), - RealtimeApiMode::V2 => SessionUpdateSession::V2(SessionUpdateSessionV2 { - kind: "realtime".to_string(), - instructions, - output_modalities: vec!["audio".to_string()], - audio: SessionAudioV2 { - input: SessionAudioInputV2 { - format: SessionAudioFormat { - kind: "audio/pcm".to_string(), - rate: 24_000, - }, - turn_detection: SessionTurnDetection { - kind: "semantic_vad".to_string(), - interrupt_response: false, - create_response: true, - }, - }, - output: SessionAudioOutputV2 { - format: SessionAudioOutputFormat { - kind: "audio/pcm".to_string(), - rate: 24_000, - }, - 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(), - }), + let message = match self.mode { + RealtimeApiMode::V1 => mode_v1::session_update(instructions), + RealtimeApiMode::V2 => mode_v2::session_update(instructions), }; - - self.send_json(RealtimeOutboundMessage::SessionUpdate { session }) - .await + self.send_json(message).await } pub async fn close(&self) -> Result<(), ApiError> { @@ -519,6 +418,13 @@ impl RealtimeWebsocketEvents { } } +fn parse_realtime_event(payload: &str, mode: RealtimeApiMode) -> Option { + match mode { + RealtimeApiMode::V1 => protocol_v1::parse_realtime_event(payload), + RealtimeApiMode::V2 => protocol_v2::parse_realtime_event(payload), + } +} + pub struct RealtimeWebsocketClient { provider: Provider, } @@ -588,7 +494,7 @@ fn merge_request_headers( let mut headers = provider_headers.clone(); headers.extend(extra_headers); for (name, value) in &default_headers { - if let http::header::Entry::Vacant(entry) = headers.entry(name) { + if let Entry::Vacant(entry) = headers.entry(name) { entry.insert(value.clone()); } } @@ -639,24 +545,9 @@ fn websocket_url_from_api_url( } } - { - let mut query = url.query_pairs_mut(); - if mode == RealtimeApiMode::V1 { - 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 == "model" && model.is_some()) - || (key == "intent" && mode == RealtimeApiMode::V1) - { - continue; - } - query.append_pair(key, value); - } - } + match mode { + RealtimeApiMode::V1 => mode_v1::append_query_params(&mut url, query_params, model), + RealtimeApiMode::V2 => mode_v2::append_query_params(&mut url, query_params, model), } Ok(url) @@ -691,8 +582,8 @@ fn normalize_realtime_path(url: &mut Url) { #[cfg(test)] mod tests { use super::*; - use crate::endpoint::realtime_websocket::protocol::RealtimeHandoffMessage; - use crate::endpoint::realtime_websocket::protocol::RealtimeHandoffRequested; + use crate::endpoint::realtime_websocket::types::RealtimeHandoffMessage; + use crate::endpoint::realtime_websocket::types::RealtimeHandoffRequested; use http::HeaderValue; use pretty_assertions::assert_eq; use serde_json::Value; diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs index 651579f174..a937c44203 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs @@ -1,5 +1,9 @@ pub mod methods; -pub mod protocol; +mod mode_v1; +mod mode_v2; +mod protocol_v1; +mod protocol_v2; +mod types; pub use codex_protocol::protocol::RealtimeAudioFrame; pub use codex_protocol::protocol::RealtimeEvent; @@ -7,5 +11,5 @@ pub use methods::RealtimeWebsocketClient; pub use methods::RealtimeWebsocketConnection; pub use methods::RealtimeWebsocketEvents; pub use methods::RealtimeWebsocketWriter; -pub use protocol::RealtimeApiMode; -pub use protocol::RealtimeSessionConfig; +pub use types::RealtimeApiMode; +pub use types::RealtimeSessionConfig; diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/mode_v1.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/mode_v1.rs new file mode 100644 index 0000000000..439c1e33ad --- /dev/null +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/mode_v1.rs @@ -0,0 +1,70 @@ +use crate::endpoint::realtime_websocket::types::ConversationItem; +use crate::endpoint::realtime_websocket::types::ConversationItemContent; +use crate::endpoint::realtime_websocket::types::RealtimeOutboundMessage; +use crate::endpoint::realtime_websocket::types::SessionAudioFormat; +use crate::endpoint::realtime_websocket::types::SessionAudioInputV1; +use crate::endpoint::realtime_websocket::types::SessionAudioOutputV1; +use crate::endpoint::realtime_websocket::types::SessionAudioV1; +use crate::endpoint::realtime_websocket::types::SessionUpdateSession; +use crate::endpoint::realtime_websocket::types::SessionUpdateSessionV1; +use std::collections::HashMap; +use url::Url; + +pub(super) fn conversation_item_create(text: String) -> RealtimeOutboundMessage { + RealtimeOutboundMessage::ConversationItemCreate { + item: ConversationItem::Message { + role: "user".to_string(), + content: vec![ConversationItemContent { + kind: "text".to_string(), + text, + }], + }, + } +} + +pub(super) fn handoff_append(handoff_id: String, output_text: String) -> RealtimeOutboundMessage { + RealtimeOutboundMessage::ConversationHandoffAppend { + handoff_id, + output_text, + } +} + +pub(super) fn session_update(instructions: String) -> RealtimeOutboundMessage { + RealtimeOutboundMessage::SessionUpdate { + session: SessionUpdateSession::V1(SessionUpdateSessionV1 { + kind: "quicksilver".to_string(), + instructions, + audio: SessionAudioV1 { + input: SessionAudioInputV1 { + format: SessionAudioFormat { + kind: "audio/pcm".to_string(), + rate: 24_000, + }, + }, + output: SessionAudioOutputV1 { + voice: "mundo".to_string(), + }, + }, + }), + } +} + +pub(super) fn append_query_params( + url: &mut Url, + query_params: Option<&HashMap>, + model: Option<&str>, +) { + 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 == "model" && model.is_some()) || key == "intent" { + continue; + } + query.append_pair(key, value); + } + } +} diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/mode_v2.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/mode_v2.rs new file mode 100644 index 0000000000..12f164c15d --- /dev/null +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/mode_v2.rs @@ -0,0 +1,121 @@ +use crate::endpoint::realtime_websocket::types::ConversationItem; +use crate::endpoint::realtime_websocket::types::ConversationItemContent; +use crate::endpoint::realtime_websocket::types::RealtimeOutboundMessage; +use crate::endpoint::realtime_websocket::types::SessionAudioFormat; +use crate::endpoint::realtime_websocket::types::SessionAudioInputV2; +use crate::endpoint::realtime_websocket::types::SessionAudioOutputFormat; +use crate::endpoint::realtime_websocket::types::SessionAudioOutputV2; +use crate::endpoint::realtime_websocket::types::SessionAudioV2; +use crate::endpoint::realtime_websocket::types::SessionTool; +use crate::endpoint::realtime_websocket::types::SessionToolParameters; +use crate::endpoint::realtime_websocket::types::SessionToolProperties; +use crate::endpoint::realtime_websocket::types::SessionToolProperty; +use crate::endpoint::realtime_websocket::types::SessionTurnDetection; +use crate::endpoint::realtime_websocket::types::SessionUpdateSession; +use crate::endpoint::realtime_websocket::types::SessionUpdateSessionV2; +use serde_json::json; +use std::collections::HashMap; +use url::Url; + +pub(super) fn conversation_item_create(text: String) -> RealtimeOutboundMessage { + RealtimeOutboundMessage::ConversationItemCreate { + item: ConversationItem::Message { + role: "user".to_string(), + content: vec![ConversationItemContent { + kind: "input_text".to_string(), + text, + }], + }, + } +} + +pub(super) fn handoff_append(output_text: String) -> RealtimeOutboundMessage { + RealtimeOutboundMessage::ConversationItemCreate { + item: ConversationItem::Message { + role: "assistant".to_string(), + content: vec![ConversationItemContent { + kind: "output_text".to_string(), + text: output_text, + }], + }, + } +} + +pub(super) fn function_call_output( + call_id: String, + output_text: String, +) -> RealtimeOutboundMessage { + let output = json!({ + "content": output_text, + }) + .to_string(); + RealtimeOutboundMessage::ConversationItemCreate { + item: ConversationItem::FunctionCallOutput { call_id, output }, + } +} + +pub(super) fn session_update(instructions: String) -> RealtimeOutboundMessage { + RealtimeOutboundMessage::SessionUpdate { + session: SessionUpdateSession::V2(SessionUpdateSessionV2 { + kind: "realtime".to_string(), + instructions, + output_modalities: vec!["audio".to_string()], + audio: SessionAudioV2 { + input: SessionAudioInputV2 { + format: SessionAudioFormat { + kind: "audio/pcm".to_string(), + rate: 24_000, + }, + turn_detection: SessionTurnDetection { + kind: "semantic_vad".to_string(), + interrupt_response: false, + create_response: true, + }, + }, + output: SessionAudioOutputV2 { + format: SessionAudioOutputFormat { + kind: "audio/pcm".to_string(), + rate: 24_000, + }, + 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(), + }), + } +} + +pub(super) fn append_query_params( + url: &mut Url, + query_params: Option<&HashMap>, + model: Option<&str>, +) { + let mut query = url.query_pairs_mut(); + 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 == "model" && model.is_some() { + continue; + } + query.append_pair(key, value); + } + } +} diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs deleted file mode 100644 index aa8687d537..0000000000 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs +++ /dev/null @@ -1,432 +0,0 @@ -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, Copy, PartialEq, Eq)] -pub enum RealtimeApiMode { - V1, - V2, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct RealtimeSessionConfig { - pub instructions: String, - pub model: Option, - pub session_id: Option, - pub mode: RealtimeApiMode, -} - -#[derive(Debug, Clone, Serialize)] -#[serde(tag = "type")] -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")] - ConversationItemCreate { item: ConversationItem }, -} - -#[derive(Debug, Clone, Serialize)] -#[serde(untagged)] -pub(super) enum SessionUpdateSession { - V1(SessionUpdateSessionV1), - V2(SessionUpdateSessionV2), -} - -#[derive(Debug, Clone, Serialize)] -pub(super) struct SessionUpdateSessionV1 { - #[serde(rename = "type")] - pub(super) kind: String, - pub(super) instructions: String, - pub(super) audio: SessionAudioV1, -} - -#[derive(Debug, Clone, Serialize)] -pub(super) struct SessionUpdateSessionV2 { - #[serde(rename = "type")] - pub(super) kind: String, - pub(super) instructions: String, - pub(super) output_modalities: Vec, - pub(super) audio: SessionAudioV2, - pub(super) tools: Vec, - pub(super) tool_choice: String, -} - -#[derive(Debug, Clone, Serialize)] -pub(super) struct SessionAudioV1 { - pub(super) input: SessionAudioInputV1, - pub(super) output: SessionAudioOutputV1, -} - -#[derive(Debug, Clone, Serialize)] -pub(super) struct SessionAudioV2 { - pub(super) input: SessionAudioInputV2, - pub(super) output: SessionAudioOutputV2, -} - -#[derive(Debug, Clone, Serialize)] -pub(super) struct SessionAudioInputV1 { - pub(super) format: SessionAudioFormat, -} - -#[derive(Debug, Clone, Serialize)] -pub(super) struct SessionAudioInputV2 { - pub(super) format: SessionAudioFormat, - pub(super) turn_detection: SessionTurnDetection, -} - -#[derive(Debug, Clone, Serialize)] -pub(super) struct SessionAudioFormat { - #[serde(rename = "type")] - pub(super) kind: String, - pub(super) rate: u32, -} - -#[derive(Debug, Clone, Serialize)] -pub(super) struct SessionTurnDetection { - #[serde(rename = "type")] - pub(super) kind: String, - pub(super) interrupt_response: bool, - pub(super) create_response: bool, -} - -#[derive(Debug, Clone, Serialize)] -pub(super) struct SessionAudioOutputV1 { - pub(super) voice: String, -} - -#[derive(Debug, Clone, Serialize)] -pub(super) struct SessionAudioOutputV2 { - pub(super) format: SessionAudioOutputFormat, - pub(super) voice: String, -} - -#[derive(Debug, Clone, Serialize)] -pub(super) struct SessionAudioOutputFormat { - #[serde(rename = "type")] - pub(super) kind: String, - pub(super) rate: u32, -} - -#[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)] -pub(super) struct ConversationItemContent { - #[serde(rename = "type")] - pub(super) kind: String, - pub(super) text: String, -} - -pub(super) fn parse_realtime_event(payload: &str, mode: RealtimeApiMode) -> Option { - let parsed: Value = match serde_json::from_str(payload) { - Ok(msg) => msg, - Err(err) => { - debug!("failed to parse realtime event: {err}, data: {payload}"); - return None; - } - }; - - let message_type = match parsed.get("type").and_then(Value::as_str) { - Some(message_type) => message_type, - None => { - debug!("received realtime event without type field: {payload}"); - return None; - } - }; - match mode { - RealtimeApiMode::V1 => parse_realtime_event_v1(&parsed, message_type, payload), - RealtimeApiMode::V2 => parse_realtime_event_v2(parsed, message_type), - } -} - -fn parse_realtime_event_v1( - parsed: &Value, - message_type: &str, - payload: &str, -) -> Option { - match message_type { - "session.updated" => parse_session_updated(parsed), - "conversation.output_audio.delta" => parse_audio_delta(parsed, false), - "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.handoff.requested" => parse_handoff_requested_v1(parsed), - "error" => parse_realtime_error(parsed), - _ => { - debug!("received unsupported realtime event type: {message_type}, data: {payload}"); - None - } - } -} - -fn parse_realtime_event_v2(parsed: Value, message_type: &str) -> Option { - match message_type { - "session.created" | "session.updated" => parse_session_updated(&parsed), - "response.output_audio.delta" => parse_audio_delta(&parsed, true), - "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 }), - "response.done" => { - if let Some(handoff) = parse_handoff_requested_v2(&parsed) { - return Some(RealtimeEvent::HandoffRequested(handoff)); - } - Some(RealtimeEvent::ConversationItemAdded(parsed)) - } - "error" => parse_realtime_error(&parsed), - _ => Some(RealtimeEvent::ConversationItemAdded(parsed)), - } -} - -fn parse_session_updated(parsed: &Value) -> Option { - let session_id = parsed - .get("session") - .and_then(Value::as_object) - .and_then(|session| session.get("id")) - .and_then(Value::as_str) - .map(str::to_string); - let instructions = parsed - .get("session") - .and_then(Value::as_object) - .and_then(|session| session.get("instructions")) - .and_then(Value::as_str) - .map(str::to_string); - session_id.map(|session_id| RealtimeEvent::SessionUpdated { - session_id, - instructions, - }) -} - -fn parse_audio_delta(parsed: &Value, default_shape: bool) -> Option { - let data = parsed - .get("delta") - .and_then(Value::as_str) - .or_else(|| parsed.get("data").and_then(Value::as_str)) - .map(str::to_string)?; - let sample_rate = parsed - .get("sample_rate") - .and_then(Value::as_u64) - .and_then(|v| u32::try_from(v).ok()); - 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()); - Some(RealtimeEvent::AudioOut(RealtimeAudioFrame { - data, - sample_rate: sample_rate.or_else(|| default_shape.then_some(24_000))?, - num_channels: num_channels.or_else(|| default_shape.then_some(1))?, - samples_per_channel: parsed - .get("samples_per_channel") - .and_then(Value::as_u64) - .and_then(|v| u32::try_from(v).ok()), - })) -} - -fn parse_realtime_error(parsed: &Value) -> Option { - parsed - .get("message") - .and_then(Value::as_str) - .map(str::to_string) - .or_else(|| { - parsed - .get("error") - .and_then(Value::as_object) - .and_then(|error| error.get("message")) - .and_then(Value::as_str) - .map(str::to_string) - }) - .or_else(|| parsed.get("error").map(ToString::to_string)) - .map(RealtimeEvent::Error) -} - -fn parse_handoff_requested_v1(parsed: &Value) -> Option { - 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, - })) -} - -fn parse_handoff_requested_v2(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/src/endpoint/realtime_websocket/protocol_v1.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v1.rs new file mode 100644 index 0000000000..254d0c55cc --- /dev/null +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v1.rs @@ -0,0 +1,140 @@ +use crate::endpoint::realtime_websocket::types::RealtimeAudioFrame; +use crate::endpoint::realtime_websocket::types::RealtimeEvent; +use crate::endpoint::realtime_websocket::types::RealtimeHandoffMessage; +use crate::endpoint::realtime_websocket::types::RealtimeHandoffRequested; +use serde_json::Value; +use std::string::ToString; +use tracing::debug; + +pub(super) fn parse_realtime_event(payload: &str) -> Option { + let parsed: Value = match serde_json::from_str(payload) { + Ok(msg) => msg, + Err(err) => { + debug!("failed to parse realtime event: {err}, data: {payload}"); + return None; + } + }; + + let message_type = match parsed.get("type").and_then(Value::as_str) { + Some(message_type) => message_type, + None => { + debug!("received realtime event without type field: {payload}"); + return None; + } + }; + + match message_type { + "session.updated" => parse_session_updated(&parsed), + "conversation.output_audio.delta" => parse_audio_delta(&parsed), + "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.handoff.requested" => parse_handoff_requested(&parsed), + "error" => parse_realtime_error(&parsed), + _ => { + debug!("received unsupported realtime event type: {message_type}, data: {payload}"); + None + } + } +} + +fn parse_session_updated(parsed: &Value) -> Option { + let session_id = parsed + .get("session") + .and_then(Value::as_object) + .and_then(|session| session.get("id")) + .and_then(Value::as_str) + .map(str::to_string); + let instructions = parsed + .get("session") + .and_then(Value::as_object) + .and_then(|session| session.get("instructions")) + .and_then(Value::as_str) + .map(str::to_string); + session_id.map(|session_id| RealtimeEvent::SessionUpdated { + session_id, + instructions, + }) +} + +fn parse_audio_delta(parsed: &Value) -> Option { + let data = parsed + .get("delta") + .and_then(Value::as_str) + .or_else(|| parsed.get("data").and_then(Value::as_str)) + .map(str::to_string)?; + let sample_rate = parsed + .get("sample_rate") + .and_then(Value::as_u64) + .and_then(|v| u32::try_from(v).ok())?; + 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())?; + Some(RealtimeEvent::AudioOut(RealtimeAudioFrame { + data, + sample_rate, + num_channels, + samples_per_channel: parsed + .get("samples_per_channel") + .and_then(Value::as_u64) + .and_then(|v| u32::try_from(v).ok()), + })) +} + +fn parse_realtime_error(parsed: &Value) -> Option { + parsed + .get("message") + .and_then(Value::as_str) + .map(str::to_string) + .or_else(|| { + parsed + .get("error") + .and_then(Value::as_object) + .and_then(|error| error.get("message")) + .and_then(Value::as_str) + .map(str::to_string) + }) + .or_else(|| parsed.get("error").map(ToString::to_string)) + .map(RealtimeEvent::Error) +} + +fn parse_handoff_requested(parsed: &Value) -> Option { + 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, + })) +} diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs new file mode 100644 index 0000000000..c60a6089fa --- /dev/null +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs @@ -0,0 +1,206 @@ +use crate::endpoint::realtime_websocket::types::RealtimeAudioFrame; +use crate::endpoint::realtime_websocket::types::RealtimeEvent; +use crate::endpoint::realtime_websocket::types::RealtimeHandoffMessage; +use crate::endpoint::realtime_websocket::types::RealtimeHandoffRequested; +use serde::Deserialize; +use serde_json::Value; +use std::string::ToString; +use tracing::debug; + +pub(super) fn parse_realtime_event(payload: &str) -> Option { + let parsed: Value = match serde_json::from_str(payload) { + Ok(msg) => msg, + Err(err) => { + debug!("failed to parse realtime event: {err}, data: {payload}"); + return None; + } + }; + + let message_type = match parsed.get("type").and_then(Value::as_str) { + Some(message_type) => message_type, + None => { + debug!("received realtime event without type field: {payload}"); + return None; + } + }; + + match message_type { + "session.created" | "session.updated" => parse_session_updated(&parsed), + "response.output_audio.delta" => parse_audio_delta(&parsed), + "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 }), + "response.done" => { + if let Some(handoff) = parse_handoff_requested(&parsed) { + return Some(RealtimeEvent::HandoffRequested(handoff)); + } + Some(RealtimeEvent::ConversationItemAdded(parsed)) + } + "error" => parse_realtime_error(&parsed), + _ => Some(RealtimeEvent::ConversationItemAdded(parsed)), + } +} + +fn parse_session_updated(parsed: &Value) -> Option { + let session_id = parsed + .get("session") + .and_then(Value::as_object) + .and_then(|session| session.get("id")) + .and_then(Value::as_str) + .map(str::to_string); + let instructions = parsed + .get("session") + .and_then(Value::as_object) + .and_then(|session| session.get("instructions")) + .and_then(Value::as_str) + .map(str::to_string); + session_id.map(|session_id| RealtimeEvent::SessionUpdated { + session_id, + instructions, + }) +} + +fn parse_audio_delta(parsed: &Value) -> Option { + let data = parsed + .get("delta") + .and_then(Value::as_str) + .or_else(|| parsed.get("data").and_then(Value::as_str)) + .map(str::to_string)?; + let sample_rate = parsed + .get("sample_rate") + .and_then(Value::as_u64) + .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()) + .unwrap_or(1); + Some(RealtimeEvent::AudioOut(RealtimeAudioFrame { + data, + sample_rate, + num_channels, + samples_per_channel: parsed + .get("samples_per_channel") + .and_then(Value::as_u64) + .and_then(|v| u32::try_from(v).ok()), + })) +} + +fn parse_realtime_error(parsed: &Value) -> Option { + parsed + .get("message") + .and_then(Value::as_str) + .map(str::to_string) + .or_else(|| { + parsed + .get("error") + .and_then(Value::as_object) + .and_then(|error| error.get("message")) + .and_then(Value::as_str) + .map(str::to_string) + }) + .or_else(|| parsed.get("error").map(ToString::to_string)) + .map(RealtimeEvent::Error) +} + +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/src/endpoint/realtime_websocket/types.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/types.rs new file mode 100644 index 0000000000..b0abb54279 --- /dev/null +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/types.rs @@ -0,0 +1,167 @@ +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::Serialize; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RealtimeApiMode { + V1, + V2, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RealtimeSessionConfig { + pub instructions: String, + pub model: Option, + pub session_id: Option, + pub mode: RealtimeApiMode, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(tag = "type")] +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")] + ConversationItemCreate { item: ConversationItem }, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(untagged)] +pub(super) enum SessionUpdateSession { + V1(SessionUpdateSessionV1), + V2(SessionUpdateSessionV2), +} + +#[derive(Debug, Clone, Serialize)] +pub(super) struct SessionUpdateSessionV1 { + #[serde(rename = "type")] + pub(super) kind: String, + pub(super) instructions: String, + pub(super) audio: SessionAudioV1, +} + +#[derive(Debug, Clone, Serialize)] +pub(super) struct SessionUpdateSessionV2 { + #[serde(rename = "type")] + pub(super) kind: String, + pub(super) instructions: String, + pub(super) output_modalities: Vec, + pub(super) audio: SessionAudioV2, + pub(super) tools: Vec, + pub(super) tool_choice: String, +} + +#[derive(Debug, Clone, Serialize)] +pub(super) struct SessionAudioV1 { + pub(super) input: SessionAudioInputV1, + pub(super) output: SessionAudioOutputV1, +} + +#[derive(Debug, Clone, Serialize)] +pub(super) struct SessionAudioV2 { + pub(super) input: SessionAudioInputV2, + pub(super) output: SessionAudioOutputV2, +} + +#[derive(Debug, Clone, Serialize)] +pub(super) struct SessionAudioInputV1 { + pub(super) format: SessionAudioFormat, +} + +#[derive(Debug, Clone, Serialize)] +pub(super) struct SessionAudioInputV2 { + pub(super) format: SessionAudioFormat, + pub(super) turn_detection: SessionTurnDetection, +} + +#[derive(Debug, Clone, Serialize)] +pub(super) struct SessionAudioFormat { + #[serde(rename = "type")] + pub(super) kind: String, + pub(super) rate: u32, +} + +#[derive(Debug, Clone, Serialize)] +pub(super) struct SessionTurnDetection { + #[serde(rename = "type")] + pub(super) kind: String, + pub(super) interrupt_response: bool, + pub(super) create_response: bool, +} + +#[derive(Debug, Clone, Serialize)] +pub(super) struct SessionAudioOutputV1 { + pub(super) voice: String, +} + +#[derive(Debug, Clone, Serialize)] +pub(super) struct SessionAudioOutputV2 { + pub(super) format: SessionAudioOutputFormat, + pub(super) voice: String, +} + +#[derive(Debug, Clone, Serialize)] +pub(super) struct SessionAudioOutputFormat { + #[serde(rename = "type")] + pub(super) kind: String, + pub(super) rate: u32, +} + +#[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)] +pub(super) struct ConversationItemContent { + #[serde(rename = "type")] + pub(super) kind: String, + pub(super) text: String, +} diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs deleted file mode 100644 index dc086b79d0..0000000000 --- a/codex-rs/core/src/realtime_conversation.rs +++ /dev/null @@ -1,822 +0,0 @@ -use crate::CodexAuth; -use crate::api_bridge::map_api_error; -use crate::auth::read_openai_api_key_from_env; -use crate::codex::Session; -use crate::default_client::default_headers; -use crate::error::CodexErr; -use crate::error::Result as CodexResult; -use crate::features::RealtimeVoiceMode; -use async_channel::Receiver; -use async_channel::Sender; -use async_channel::TrySendError; -use codex_api::Provider as ApiProvider; -use codex_api::RealtimeApiMode; -use codex_api::RealtimeAudioFrame; -use codex_api::RealtimeEvent; -use codex_api::RealtimeSessionConfig; -use codex_api::RealtimeWebsocketClient; -use codex_api::endpoint::realtime_websocket::RealtimeWebsocketEvents; -use codex_api::endpoint::realtime_websocket::RealtimeWebsocketWriter; -use codex_protocol::protocol::CodexErrorInfo; -use codex_protocol::protocol::ConversationAudioParams; -use codex_protocol::protocol::ConversationStartParams; -use codex_protocol::protocol::ConversationTextParams; -use codex_protocol::protocol::ErrorEvent; -use codex_protocol::protocol::Event; -use codex_protocol::protocol::EventMsg; -use codex_protocol::protocol::RealtimeConversationClosedEvent; -use codex_protocol::protocol::RealtimeConversationRealtimeEvent; -use codex_protocol::protocol::RealtimeConversationStartedEvent; -use codex_protocol::protocol::RealtimeHandoffRequested; -use http::HeaderMap; -use http::HeaderValue; -use http::header::AUTHORIZATION; -use std::sync::Arc; -use std::sync::atomic::AtomicBool; -use std::sync::atomic::Ordering; -use tokio::sync::Mutex; -use tokio::task::JoinHandle; -use tracing::debug; -use tracing::error; -use tracing::info; -use tracing::warn; - -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>, -} - -#[derive(Clone, Debug)] -struct RealtimeHandoffState { - output_tx: Sender, - active_handoff: Arc>>, - last_output_text: Arc>>, - mode: RealtimeApiMode, -} - -#[derive(Debug, PartialEq, Eq)] -enum HandoffOutput { - TextUpdate { - handoff_id: String, - output_text: String, - }, - FinalToolCall { - call_id: String, - output_text: String, - }, -} - -impl RealtimeHandoffState { - fn new(output_tx: Sender, mode: RealtimeApiMode) -> Self { - Self { - output_tx, - active_handoff: Arc::new(Mutex::new(None)), - last_output_text: Arc::new(Mutex::new(None)), - mode, - } - } - - async fn send_output(&self, output_text: String) -> CodexResult<()> { - 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::TextUpdate { - handoff_id, - output_text, - }) - .await - .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?; - Ok(()) - } - - async fn send_final_output(&self) -> CodexResult<()> { - if self.mode == RealtimeApiMode::V1 { - return Ok(()); - } - 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)] -struct ConversationState { - audio_tx: Sender, - user_text_tx: Sender, - handoff: RealtimeHandoffState, - mode: RealtimeApiMode, - task: JoinHandle<()>, - realtime_active: Arc, -} - -#[allow(dead_code)] -impl RealtimeConversationManager { - pub(crate) fn new() -> Self { - Self { - state: Mutex::new(None), - } - } - - pub(crate) async fn running_state(&self) -> Option<()> { - let state = self.state.lock().await; - state - .as_ref() - .and_then(|state| state.realtime_active.load(Ordering::Relaxed).then_some(())) - } - - pub(crate) async fn start( - &self, - api_provider: ApiProvider, - extra_headers: Option, - prompt: String, - model: Option, - mode: RealtimeApiMode, - session_id: Option, - ) -> CodexResult<(Receiver, Arc)> { - let previous_state = { - let mut guard = self.state.lock().await; - guard.take() - }; - if let Some(state) = previous_state { - state.realtime_active.store(false, Ordering::Relaxed); - state.task.abort(); - let _ = state.task.await; - } - - let session_config = RealtimeSessionConfig { - instructions: prompt, - model, - session_id, - mode, - }; - let client = RealtimeWebsocketClient::new(api_provider); - let connection = client - .connect( - session_config, - extra_headers.unwrap_or_default(), - default_headers(), - ) - .await - .map_err(map_api_error)?; - - let writer = connection.writer(); - let events = connection.events(); - let (audio_tx, audio_rx) = - async_channel::bounded::(AUDIO_IN_QUEUE_CAPACITY); - let (user_text_tx, user_text_rx) = - async_channel::bounded::(USER_TEXT_IN_QUEUE_CAPACITY); - let (handoff_output_tx, handoff_output_rx) = - async_channel::bounded::(HANDOFF_OUT_QUEUE_CAPACITY); - let (events_tx, events_rx) = - async_channel::bounded::(OUTPUT_EVENTS_QUEUE_CAPACITY); - - let realtime_active = Arc::new(AtomicBool::new(true)); - let handoff = RealtimeHandoffState::new(handoff_output_tx, mode); - let task = spawn_realtime_input_task( - writer, - events, - user_text_rx, - handoff_output_rx, - audio_rx, - events_tx, - handoff.clone(), - ); - - let mut guard = self.state.lock().await; - *guard = Some(ConversationState { - audio_tx, - user_text_tx, - handoff, - mode, - task, - realtime_active: Arc::clone(&realtime_active), - }); - Ok((events_rx, realtime_active)) - } - - pub(crate) async fn audio_in(&self, frame: RealtimeAudioFrame) -> CodexResult<()> { - let sender = { - let guard = self.state.lock().await; - guard.as_ref().map(|state| state.audio_tx.clone()) - }; - - let Some(sender) = sender else { - return Err(CodexErr::InvalidRequest( - "conversation is not running".to_string(), - )); - }; - - match sender.try_send(frame) { - Ok(()) => Ok(()), - Err(TrySendError::Full(_)) => { - warn!("dropping input audio frame due to full queue"); - Ok(()) - } - Err(TrySendError::Closed(_)) => Err(CodexErr::InvalidRequest( - "conversation is not running".to_string(), - )), - } - } - - pub(crate) async fn text_in(&self, text: String) -> CodexResult<()> { - let sender = { - let guard = self.state.lock().await; - guard.as_ref().map(|state| state.user_text_tx.clone()) - }; - - let Some(sender) = sender else { - return Err(CodexErr::InvalidRequest( - "conversation is not running".to_string(), - )); - }; - - sender - .send(text) - .await - .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?; - Ok(()) - } - - pub(crate) async fn handoff_out(&self, output_text: String) -> CodexResult<()> { - let handoff = { - let guard = self.state.lock().await; - let Some(state) = guard.as_ref() else { - return Err(CodexErr::InvalidRequest( - "conversation is not running".to_string(), - )); - }; - state.handoff.clone() - }; - - 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; - guard.as_ref().map(|state| state.handoff.clone()) - }?; - handoff.active_handoff.lock().await.clone() - } - - pub(crate) async fn clear_active_handoff(&self) { - let handoff = { - let guard = self.state.lock().await; - guard.as_ref().map(|state| state.handoff.clone()) - }; - if let Some(handoff) = handoff { - *handoff.active_handoff.lock().await = None; - *handoff.last_output_text.lock().await = None; - } - } - - pub(crate) async fn shutdown(&self) -> CodexResult<()> { - let state = { - let mut guard = self.state.lock().await; - guard.take() - }; - - if let Some(state) = state { - state.realtime_active.store(false, Ordering::Relaxed); - state.task.abort(); - let _ = state.task.await; - } - Ok(()) - } -} - -pub(crate) async fn handle_start( - sess: &Arc, - sub_id: String, - params: ConversationStartParams, -) -> CodexResult<()> { - let provider = sess.provider().await; - let auth = sess.services.auth_manager.auth().await; - let realtime_api_key = realtime_api_key(auth.as_ref(), &provider)?; - let mut api_provider = provider.to_api_provider(Some(crate::auth::AuthMode::ApiKey))?; - let config = sess.get_config().await; - if let Some(realtime_ws_base_url) = &config.experimental_realtime_ws_base_url { - api_provider.base_url = realtime_ws_base_url.clone(); - } - let prompt = config - .experimental_realtime_ws_backend_prompt - .clone() - .unwrap_or(params.prompt); - let mode = config - .features - .realtime_voice_mode() - .unwrap_or(RealtimeVoiceMode::V2); - let (realtime_api_mode, model) = match mode { - RealtimeVoiceMode::V1 => ( - RealtimeApiMode::V1, - config.experimental_realtime_ws_model.clone(), - ), - RealtimeVoiceMode::V2 => ( - RealtimeApiMode::V2, - Some(DEFAULT_REALTIME_MODEL.to_string()), - ), - }; - - let requested_session_id = params - .session_id - .or_else(|| Some(sess.conversation_id.to_string())); - let extra_headers = - realtime_request_headers(requested_session_id.as_deref(), realtime_api_key.as_str())?; - info!("starting realtime conversation"); - let (events_rx, realtime_active) = match sess - .conversation - .start( - api_provider, - extra_headers, - prompt, - model, - realtime_api_mode, - requested_session_id.clone(), - ) - .await - { - Ok(events_rx) => events_rx, - Err(err) => { - error!("failed to start realtime conversation: {err}"); - send_conversation_error(sess, sub_id, err.to_string(), CodexErrorInfo::Other).await; - return Ok(()); - } - }; - - info!("realtime conversation started"); - - sess.send_event_raw(Event { - id: sub_id.clone(), - msg: EventMsg::RealtimeConversationStarted(RealtimeConversationStartedEvent { - session_id: requested_session_id, - }), - }) - .await; - - let sess_clone = Arc::clone(sess); - tokio::spawn(async move { - let ev = |msg| Event { - id: sub_id.clone(), - msg, - }; - while let Ok(event) = events_rx.recv().await { - debug!(conversation_id = %sess_clone.conversation_id, "received realtime conversation event"); - let maybe_routed_text = match &event { - RealtimeEvent::HandoffRequested(handoff) => { - realtime_text_from_handoff_request(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(text).await; - } - sess_clone - .send_event_raw(ev(EventMsg::RealtimeConversationRealtime( - RealtimeConversationRealtimeEvent { - payload: event.clone(), - }, - ))) - .await; - } - if realtime_active.swap(false, Ordering::Relaxed) { - info!("realtime conversation transport closed"); - sess_clone - .send_event_raw(ev(EventMsg::RealtimeConversationClosed( - RealtimeConversationClosedEvent { - reason: Some("transport_closed".to_string()), - }, - ))) - .await; - } - }); - - Ok(()) -} - -pub(crate) async fn handle_audio( - sess: &Arc, - sub_id: String, - params: ConversationAudioParams, -) { - if let Err(err) = sess.conversation.audio_in(params.frame).await { - error!("failed to append realtime audio: {err}"); - send_conversation_error(sess, sub_id, err.to_string(), CodexErrorInfo::BadRequest).await; - } -} - -fn realtime_text_from_handoff_request(handoff: &RealtimeHandoffRequested) -> Option { - let messages = handoff - .messages - .iter() - .map(|message| message.text.as_str()) - .collect::>() - .join("\n"); - (!messages.is_empty()).then_some(messages).or_else(|| { - (!handoff.input_transcript.is_empty()).then(|| handoff.input_transcript.clone()) - }) -} - -fn realtime_api_key( - auth: Option<&CodexAuth>, - provider: &crate::ModelProviderInfo, -) -> CodexResult { - if let Some(api_key) = provider.api_key()? { - return Ok(api_key); - } - - if let Some(token) = provider.experimental_bearer_token.clone() { - return Ok(token); - } - - if let Some(api_key) = auth.and_then(CodexAuth::api_key) { - return Ok(api_key.to_string()); - } - - // TODO(aibrahim): Remove this temporary fallback once realtime auth no longer - // requires API key auth for ChatGPT/SIWC sessions. - if provider.is_openai() - && let Some(api_key) = read_openai_api_key_from_env() - { - return Ok(api_key); - } - - Err(CodexErr::InvalidRequest( - "realtime conversation requires API key auth".to_string(), - )) -} - -fn realtime_request_headers( - session_id: Option<&str>, - api_key: &str, -) -> CodexResult> { - let mut headers = HeaderMap::new(); - - if let Some(session_id) = session_id - && let Ok(session_id) = HeaderValue::from_str(session_id) - { - headers.insert("x-session-id", session_id); - } - - let auth_value = HeaderValue::from_str(&format!("Bearer {api_key}")).map_err(|err| { - CodexErr::InvalidRequest(format!("invalid realtime api key header: {err}")) - })?; - headers.insert(AUTHORIZATION, auth_value); - - Ok(Some(headers)) -} - -pub(crate) async fn handle_text( - sess: &Arc, - sub_id: String, - params: ConversationTextParams, -) { - debug!(text = %params.text, "[realtime-text] appending realtime conversation text input"); - - if let Err(err) = sess.conversation.text_in(params.text).await { - error!("failed to append realtime text: {err}"); - send_conversation_error(sess, sub_id, err.to_string(), CodexErrorInfo::BadRequest).await; - } -} - -pub(crate) async fn handle_close(sess: &Arc, sub_id: String) { - match sess.conversation.shutdown().await { - Ok(()) => { - sess.send_event_raw(Event { - id: sub_id, - msg: EventMsg::RealtimeConversationClosed(RealtimeConversationClosedEvent { - reason: Some("requested".to_string()), - }), - }) - .await; - } - Err(err) => { - send_conversation_error(sess, sub_id, err.to_string(), CodexErrorInfo::Other).await; - } - } -} - -fn spawn_realtime_input_task( - writer: RealtimeWebsocketWriter, - events: RealtimeWebsocketEvents, - user_text_rx: Receiver, - handoff_output_rx: Receiver, - audio_rx: Receiver, - events_tx: Sender, - handoff_state: RealtimeHandoffState, -) -> JoinHandle<()> { - tokio::spawn(async move { - loop { - tokio::select! { - text = user_text_rx.recv() => { - match text { - Ok(text) => { - if let Err(err) = writer.send_conversation_item_create(text).await { - let mapped_error = map_api_error(err); - warn!("failed to send input text: {mapped_error}"); - break; - } - } - Err(_) => break, - } - } - handoff_output = handoff_output_rx.recv() => { - match handoff_output { - 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, - } - } - event = events.next_event() => { - match event { - Ok(Some(event)) => { - 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() { - break; - } - if should_stop { - error!("realtime stream error event received"); - break; - } - } - Ok(None) => { - let _ = events_tx - .send(RealtimeEvent::Error( - "realtime websocket connection is closed".to_string(), - )) - .await; - break; - } - Err(err) => { - let mapped_error = map_api_error(err); - if events_tx - .send(RealtimeEvent::Error(mapped_error.to_string())) - .await - .is_err() - { - break; - } - error!("realtime stream closed: {mapped_error}"); - break; - } - } - } - frame = audio_rx.recv() => { - match frame { - Ok(frame) => { - if let Err(err) = writer.send_audio_frame(frame).await { - let mapped_error = map_api_error(err); - error!("failed to send input audio: {mapped_error}"); - break; - } - } - Err(_) => break, - } - } - } - } - }) -} - -async fn send_conversation_error( - sess: &Arc, - sub_id: String, - message: String, - codex_error_info: CodexErrorInfo, -) { - sess.send_event_raw(Event { - id: sub_id, - msg: EventMsg::Error(ErrorEvent { - message, - codex_error_info: Some(codex_error_info), - }), - }) - .await; -} - -#[cfg(test)] -mod tests { - use super::HandoffOutput; - use super::RealtimeHandoffState; - use super::realtime_text_from_handoff_request; - use async_channel::bounded; - use codex_api::RealtimeApiMode; - use codex_protocol::protocol::RealtimeHandoffMessage; - use codex_protocol::protocol::RealtimeHandoffRequested; - use pretty_assertions::assert_eq; - - #[test] - fn extracts_text_from_handoff_request_messages() { - let handoff = RealtimeHandoffRequested { - handoff_id: "handoff_1".to_string(), - item_id: "item_1".to_string(), - input_transcript: "ignored".to_string(), - messages: vec![ - RealtimeHandoffMessage { - role: "user".to_string(), - text: "hello".to_string(), - }, - RealtimeHandoffMessage { - role: "assistant".to_string(), - text: "hi there".to_string(), - }, - ], - }; - assert_eq!( - realtime_text_from_handoff_request(&handoff), - Some("hello\nhi there".to_string()) - ); - } - - #[test] - fn extracts_text_from_handoff_request_input_transcript_if_messages_missing() { - let handoff = RealtimeHandoffRequested { - handoff_id: "handoff_1".to_string(), - item_id: "item_1".to_string(), - input_transcript: "ignored".to_string(), - messages: vec![], - }; - assert_eq!( - realtime_text_from_handoff_request(&handoff), - Some("ignored".to_string()) - ); - } - - #[test] - fn ignores_empty_handoff_request_input_transcript() { - let handoff = RealtimeHandoffRequested { - handoff_id: "handoff_1".to_string(), - item_id: "item_1".to_string(), - input_transcript: String::new(), - messages: vec![], - }; - assert_eq!(realtime_text_from_handoff_request(&handoff), None); - } - - #[tokio::test] - async fn clears_active_handoff_explicitly() { - let (tx, _rx) = bounded(1); - let state = RealtimeHandoffState::new(tx, RealtimeApiMode::V2); - - *state.active_handoff.lock().await = Some("handoff_1".to_string()); - assert_eq!( - state.active_handoff.lock().await.clone(), - Some("handoff_1".to_string()) - ); - - *state.active_handoff.lock().await = None; - assert_eq!(state.active_handoff.lock().await.clone(), None); - } - - #[tokio::test] - async fn sends_multiple_handoff_outputs_until_cleared() { - let (tx, rx) = bounded(4); - let state = RealtimeHandoffState::new(tx, RealtimeApiMode::V2); - - state - .send_output("ignored".to_string()) - .await - .expect("send"); - assert!(rx.is_empty()); - - *state.active_handoff.lock().await = Some("handoff_1".to_string()); - state.send_output("result".to_string()).await.expect("send"); - state - .send_output("result 2".to_string()) - .await - .expect("send"); - - let output_1 = rx.recv().await.expect("recv"); - assert_eq!( - output_1, - HandoffOutput::TextUpdate { - handoff_id: "handoff_1".to_string(), - output_text: "result".to_string(), - } - ); - - let output_2 = rx.recv().await.expect("recv"); - assert_eq!( - output_2, - HandoffOutput::TextUpdate { - handoff_id: "handoff_1".to_string(), - output_text: "result 2".to_string(), - } - ); - - *state.active_handoff.lock().await = None; - state - .send_output("ignored after clear".to_string()) - .await - .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, RealtimeApiMode::V2); - *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(), - } - ); - } - - #[tokio::test] - async fn does_not_send_final_tool_call_output_in_v1_mode() { - let (tx, rx) = bounded(4); - let state = RealtimeHandoffState::new(tx, RealtimeApiMode::V1); - *state.active_handoff.lock().await = Some("handoff_3".to_string()); - - state - .send_output("legacy final text".to_string()) - .await - .expect("send"); - let _ = rx.recv().await.expect("recv text update"); - - state.send_final_output().await.expect("send final output"); - assert!(rx.is_empty()); - } -} diff --git a/codex-rs/core/src/realtime_conversation/common.rs b/codex-rs/core/src/realtime_conversation/common.rs new file mode 100644 index 0000000000..ff12e5bbe8 --- /dev/null +++ b/codex-rs/core/src/realtime_conversation/common.rs @@ -0,0 +1,94 @@ +use crate::CodexAuth; +use crate::ModelProviderInfo; +use crate::auth::read_openai_api_key_from_env; +use crate::codex::Session; +use crate::error::CodexErr; +use crate::error::Result as CodexResult; +use codex_protocol::protocol::CodexErrorInfo; +use codex_protocol::protocol::ErrorEvent; +use codex_protocol::protocol::Event; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::RealtimeHandoffRequested; +use http::HeaderMap; +use http::HeaderValue; +use http::header::AUTHORIZATION; +use std::sync::Arc; + +pub(super) fn realtime_text_from_handoff_request( + handoff: &RealtimeHandoffRequested, +) -> Option { + let messages = handoff + .messages + .iter() + .map(|message| message.text.as_str()) + .collect::>() + .join("\n"); + (!messages.is_empty()).then_some(messages).or_else(|| { + (!handoff.input_transcript.is_empty()).then(|| handoff.input_transcript.clone()) + }) +} + +pub(super) fn realtime_api_key( + auth: Option<&CodexAuth>, + provider: &ModelProviderInfo, +) -> CodexResult { + if let Some(api_key) = provider.api_key()? { + return Ok(api_key); + } + + if let Some(token) = provider.experimental_bearer_token.clone() { + return Ok(token); + } + + if let Some(api_key) = auth.and_then(CodexAuth::api_key) { + return Ok(api_key.to_string()); + } + + // TODO(aibrahim): Remove this temporary fallback once realtime auth no longer + // requires API key auth for ChatGPT/SIWC sessions. + if provider.is_openai() + && let Some(api_key) = read_openai_api_key_from_env() + { + return Ok(api_key); + } + + Err(CodexErr::InvalidRequest( + "realtime conversation requires API key auth".to_string(), + )) +} + +pub(super) fn realtime_request_headers( + session_id: Option<&str>, + api_key: &str, +) -> CodexResult> { + let mut headers = HeaderMap::new(); + + if let Some(session_id) = session_id + && let Ok(session_id) = HeaderValue::from_str(session_id) + { + headers.insert("x-session-id", session_id); + } + + let auth_value = HeaderValue::from_str(&format!("Bearer {api_key}")).map_err(|err| { + CodexErr::InvalidRequest(format!("invalid realtime api key header: {err}")) + })?; + headers.insert(AUTHORIZATION, auth_value); + + Ok(Some(headers)) +} + +pub(super) async fn send_conversation_error( + sess: &Arc, + sub_id: String, + message: String, + codex_error_info: CodexErrorInfo, +) { + sess.send_event_raw(Event { + id: sub_id, + msg: EventMsg::Error(ErrorEvent { + message, + codex_error_info: Some(codex_error_info), + }), + }) + .await; +} diff --git a/codex-rs/core/src/realtime_conversation/mod.rs b/codex-rs/core/src/realtime_conversation/mod.rs new file mode 100644 index 0000000000..317fd1b377 --- /dev/null +++ b/codex-rs/core/src/realtime_conversation/mod.rs @@ -0,0 +1,430 @@ +mod common; +mod v1; +mod v2; + +use crate::auth::AuthMode; +use crate::codex::Session; +use crate::error::CodexErr; +use crate::error::Result as CodexResult; +use crate::features::RealtimeVoiceMode; +use async_channel::Receiver; +use codex_api::Provider as ApiProvider; +use codex_api::RealtimeApiMode; +use codex_api::RealtimeAudioFrame; +use codex_api::RealtimeEvent; +use codex_protocol::protocol::CodexErrorInfo; +use codex_protocol::protocol::ConversationAudioParams; +use codex_protocol::protocol::ConversationStartParams; +use codex_protocol::protocol::ConversationTextParams; +use codex_protocol::protocol::Event; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::RealtimeConversationClosedEvent; +use codex_protocol::protocol::RealtimeConversationRealtimeEvent; +use codex_protocol::protocol::RealtimeConversationStartedEvent; +use http::HeaderMap; +use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; +use tokio::sync::Mutex; +use tracing::debug; +use tracing::error; +use tracing::info; + +const DEFAULT_REALTIME_MODEL: &str = "gpt-realtime-1.5"; + +pub(crate) struct RealtimeConversationManager { + state: Mutex>, +} + +enum ConversationState { + V1(v1::ConversationState), + V2(v2::ConversationState), +} + +impl ConversationState { + fn is_running(&self) -> bool { + match self { + Self::V1(state) => state.is_running(), + Self::V2(state) => state.is_running(), + } + } + + async fn audio_in(&self, frame: RealtimeAudioFrame) -> CodexResult<()> { + match self { + Self::V1(state) => state.audio_in(frame).await, + Self::V2(state) => state.audio_in(frame).await, + } + } + + async fn text_in(&self, text: String) -> CodexResult<()> { + match self { + Self::V1(state) => state.text_in(text).await, + Self::V2(state) => state.text_in(text).await, + } + } + + async fn handoff_out(&self, output_text: String) -> CodexResult<()> { + match self { + Self::V1(state) => state.handoff_out(output_text).await, + Self::V2(state) => state.handoff_out(output_text).await, + } + } + + async fn handoff_complete(&self) -> CodexResult<()> { + match self { + Self::V1(state) => state.handoff_complete().await, + Self::V2(state) => state.handoff_complete().await, + } + } + + async fn active_handoff_id(&self) -> Option { + match self { + Self::V1(state) => state.active_handoff_id().await, + Self::V2(state) => state.active_handoff_id().await, + } + } + + async fn clear_active_handoff(&self) { + match self { + Self::V1(state) => state.clear_active_handoff().await, + Self::V2(state) => state.clear_active_handoff().await, + } + } + + async fn shutdown(self) { + match self { + Self::V1(state) => state.shutdown().await, + Self::V2(state) => state.shutdown().await, + } + } + + fn realtime_active(&self) -> Arc { + match self { + Self::V1(state) => state.realtime_active(), + Self::V2(state) => state.realtime_active(), + } + } +} + +impl RealtimeConversationManager { + pub(crate) fn new() -> Self { + Self { + state: Mutex::new(None), + } + } + + pub(crate) async fn running_state(&self) -> Option<()> { + let state = self.state.lock().await; + state + .as_ref() + .and_then(|state| state.is_running().then_some(())) + } + + pub(crate) async fn start( + &self, + api_provider: ApiProvider, + extra_headers: Option, + prompt: String, + model: Option, + mode: RealtimeApiMode, + session_id: Option, + ) -> CodexResult<(Receiver, Arc)> { + let previous_state = { + let mut guard = self.state.lock().await; + guard.take() + }; + if let Some(state) = previous_state { + state.shutdown().await; + } + + let (state, events_rx) = match mode { + RealtimeApiMode::V1 => { + let (state, events_rx) = + v1::start(api_provider, extra_headers, prompt, model, session_id).await?; + (ConversationState::V1(state), events_rx) + } + RealtimeApiMode::V2 => { + let (state, events_rx) = + v2::start(api_provider, extra_headers, prompt, model, session_id).await?; + (ConversationState::V2(state), events_rx) + } + }; + + let realtime_active = state.realtime_active(); + let mut guard = self.state.lock().await; + *guard = Some(state); + Ok((events_rx, realtime_active)) + } + + pub(crate) async fn audio_in(&self, frame: RealtimeAudioFrame) -> CodexResult<()> { + let guard = self.state.lock().await; + let Some(state) = guard.as_ref() else { + return Err(CodexErr::InvalidRequest( + "conversation is not running".to_string(), + )); + }; + state.audio_in(frame).await + } + + pub(crate) async fn text_in(&self, text: String) -> CodexResult<()> { + let guard = self.state.lock().await; + let Some(state) = guard.as_ref() else { + return Err(CodexErr::InvalidRequest( + "conversation is not running".to_string(), + )); + }; + state.text_in(text).await + } + + pub(crate) async fn handoff_out(&self, output_text: String) -> CodexResult<()> { + let guard = self.state.lock().await; + let Some(state) = guard.as_ref() else { + return Err(CodexErr::InvalidRequest( + "conversation is not running".to_string(), + )); + }; + state.handoff_out(output_text).await + } + + pub(crate) async fn handoff_complete(&self) -> CodexResult<()> { + let guard = self.state.lock().await; + let Some(state) = guard.as_ref() else { + return Ok(()); + }; + state.handoff_complete().await + } + + pub(crate) async fn active_handoff_id(&self) -> Option { + let guard = self.state.lock().await; + let state = guard.as_ref()?; + state.active_handoff_id().await + } + + pub(crate) async fn clear_active_handoff(&self) { + let guard = self.state.lock().await; + if let Some(state) = guard.as_ref() { + state.clear_active_handoff().await; + } + } + + pub(crate) async fn shutdown(&self) -> CodexResult<()> { + let state = { + let mut guard = self.state.lock().await; + guard.take() + }; + + if let Some(state) = state { + state.shutdown().await; + } + Ok(()) + } +} + +pub(crate) async fn handle_start( + sess: &Arc, + sub_id: String, + params: ConversationStartParams, +) -> CodexResult<()> { + let provider = sess.provider().await; + let auth = sess.services.auth_manager.auth().await; + let realtime_api_key = common::realtime_api_key(auth.as_ref(), &provider)?; + let mut api_provider = provider.to_api_provider(Some(AuthMode::ApiKey))?; + let config = sess.get_config().await; + if let Some(realtime_ws_base_url) = &config.experimental_realtime_ws_base_url { + api_provider.base_url = realtime_ws_base_url.clone(); + } + let prompt = config + .experimental_realtime_ws_backend_prompt + .clone() + .unwrap_or(params.prompt); + let mode = config + .features + .realtime_voice_mode() + .unwrap_or(RealtimeVoiceMode::V2); + let (realtime_api_mode, model) = match mode { + RealtimeVoiceMode::V1 => ( + RealtimeApiMode::V1, + config.experimental_realtime_ws_model.clone(), + ), + RealtimeVoiceMode::V2 => ( + RealtimeApiMode::V2, + Some(DEFAULT_REALTIME_MODEL.to_string()), + ), + }; + + let requested_session_id = params + .session_id + .or_else(|| Some(sess.conversation_id.to_string())); + let extra_headers = common::realtime_request_headers( + requested_session_id.as_deref(), + realtime_api_key.as_str(), + )?; + info!("starting realtime conversation"); + let (events_rx, realtime_active) = match sess + .conversation + .start( + api_provider, + extra_headers, + prompt, + model, + realtime_api_mode, + requested_session_id.clone(), + ) + .await + { + Ok(events_rx) => events_rx, + Err(err) => { + error!("failed to start realtime conversation: {err}"); + common::send_conversation_error(sess, sub_id, err.to_string(), CodexErrorInfo::Other) + .await; + return Ok(()); + } + }; + + info!("realtime conversation started"); + + sess.send_event_raw(Event { + id: sub_id.clone(), + msg: EventMsg::RealtimeConversationStarted(RealtimeConversationStartedEvent { + session_id: requested_session_id, + }), + }) + .await; + + let sess_clone = Arc::clone(sess); + tokio::spawn(async move { + let ev = |msg| Event { + id: sub_id.clone(), + msg, + }; + while let Ok(event) = events_rx.recv().await { + debug!(conversation_id = %sess_clone.conversation_id, "received realtime conversation event"); + if let RealtimeEvent::HandoffRequested(handoff) = &event + && let Some(text) = common::realtime_text_from_handoff_request(handoff) + { + debug!(text = %text, "[realtime-text] realtime conversation text output"); + let sess_for_routed_text = Arc::clone(&sess_clone); + sess_for_routed_text.route_realtime_text_input(text).await; + } + sess_clone + .send_event_raw(ev(EventMsg::RealtimeConversationRealtime( + RealtimeConversationRealtimeEvent { + payload: event.clone(), + }, + ))) + .await; + } + if realtime_active.swap(false, Ordering::Relaxed) { + info!("realtime conversation transport closed"); + sess_clone + .send_event_raw(ev(EventMsg::RealtimeConversationClosed( + RealtimeConversationClosedEvent { + reason: Some("transport_closed".to_string()), + }, + ))) + .await; + } + }); + + Ok(()) +} + +pub(crate) async fn handle_audio( + sess: &Arc, + sub_id: String, + params: ConversationAudioParams, +) { + if let Err(err) = sess.conversation.audio_in(params.frame).await { + error!("failed to append realtime audio: {err}"); + common::send_conversation_error(sess, sub_id, err.to_string(), CodexErrorInfo::BadRequest) + .await; + } +} + +pub(crate) async fn handle_text( + sess: &Arc, + sub_id: String, + params: ConversationTextParams, +) { + debug!(text = %params.text, "[realtime-text] appending realtime conversation text input"); + + if let Err(err) = sess.conversation.text_in(params.text).await { + error!("failed to append realtime text: {err}"); + common::send_conversation_error(sess, sub_id, err.to_string(), CodexErrorInfo::BadRequest) + .await; + } +} + +pub(crate) async fn handle_close(sess: &Arc, sub_id: String) { + match sess.conversation.shutdown().await { + Ok(()) => { + sess.send_event_raw(Event { + id: sub_id, + msg: EventMsg::RealtimeConversationClosed(RealtimeConversationClosedEvent { + reason: Some("requested".to_string()), + }), + }) + .await; + } + Err(err) => { + common::send_conversation_error(sess, sub_id, err.to_string(), CodexErrorInfo::Other) + .await; + } + } +} + +#[cfg(test)] +mod tests { + use super::common::realtime_text_from_handoff_request; + use codex_protocol::protocol::RealtimeHandoffMessage; + use codex_protocol::protocol::RealtimeHandoffRequested; + use pretty_assertions::assert_eq; + + #[test] + fn extracts_text_from_handoff_request_messages() { + let handoff = RealtimeHandoffRequested { + handoff_id: "handoff_1".to_string(), + item_id: "item_1".to_string(), + input_transcript: "ignored".to_string(), + messages: vec![ + RealtimeHandoffMessage { + role: "user".to_string(), + text: "hello".to_string(), + }, + RealtimeHandoffMessage { + role: "assistant".to_string(), + text: "hi there".to_string(), + }, + ], + }; + assert_eq!( + realtime_text_from_handoff_request(&handoff), + Some("hello\nhi there".to_string()) + ); + } + + #[test] + fn extracts_text_from_handoff_request_input_transcript_if_messages_missing() { + let handoff = RealtimeHandoffRequested { + handoff_id: "handoff_1".to_string(), + item_id: "item_1".to_string(), + input_transcript: "ignored".to_string(), + messages: vec![], + }; + assert_eq!( + realtime_text_from_handoff_request(&handoff), + Some("ignored".to_string()) + ); + } + + #[test] + fn ignores_empty_handoff_request_input_transcript() { + let handoff = RealtimeHandoffRequested { + handoff_id: "handoff_1".to_string(), + item_id: "item_1".to_string(), + input_transcript: String::new(), + messages: vec![], + }; + assert_eq!(realtime_text_from_handoff_request(&handoff), None); + } +} diff --git a/codex-rs/core/src/realtime_conversation/v1.rs b/codex-rs/core/src/realtime_conversation/v1.rs new file mode 100644 index 0000000000..f0130e05b3 --- /dev/null +++ b/codex-rs/core/src/realtime_conversation/v1.rs @@ -0,0 +1,319 @@ +use crate::api_bridge::map_api_error; +use crate::default_client::default_headers; +use crate::error::CodexErr; +use crate::error::Result as CodexResult; +use async_channel::Receiver; +use async_channel::Sender; +use async_channel::TrySendError; +use codex_api::Provider as ApiProvider; +use codex_api::RealtimeApiMode; +use codex_api::RealtimeAudioFrame; +use codex_api::RealtimeEvent; +use codex_api::RealtimeSessionConfig; +use codex_api::RealtimeWebsocketClient; +use codex_api::endpoint::realtime_websocket::RealtimeWebsocketEvents; +use codex_api::endpoint::realtime_websocket::RealtimeWebsocketWriter; +use http::HeaderMap; +use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; +use tokio::sync::Mutex; +use tokio::task::JoinHandle; +use tracing::error; +use tracing::warn; + +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; + +#[derive(Clone, Debug)] +struct RealtimeHandoffState { + output_tx: Sender, + active_handoff: Arc>>, +} + +#[derive(Debug, PartialEq, Eq)] +struct HandoffOutput { + handoff_id: String, + output_text: String, +} + +impl RealtimeHandoffState { + fn new(output_tx: Sender) -> Self { + Self { + output_tx, + active_handoff: Arc::new(Mutex::new(None)), + } + } + + async fn send_output(&self, output_text: String) -> CodexResult<()> { + let Some(handoff_id) = self.active_handoff.lock().await.clone() else { + return Ok(()); + }; + + self.output_tx + .send(HandoffOutput { + handoff_id, + output_text, + }) + .await + .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?; + Ok(()) + } +} + +pub(super) struct ConversationState { + audio_tx: Sender, + user_text_tx: Sender, + handoff: RealtimeHandoffState, + task: JoinHandle<()>, + realtime_active: Arc, +} + +impl ConversationState { + pub(super) fn is_running(&self) -> bool { + self.realtime_active.load(Ordering::Relaxed) + } + + pub(super) async fn audio_in(&self, frame: RealtimeAudioFrame) -> CodexResult<()> { + match self.audio_tx.try_send(frame) { + Ok(()) => Ok(()), + Err(TrySendError::Full(_)) => { + warn!("dropping input audio frame due to full queue"); + Ok(()) + } + Err(TrySendError::Closed(_)) => Err(CodexErr::InvalidRequest( + "conversation is not running".to_string(), + )), + } + } + + pub(super) async fn text_in(&self, text: String) -> CodexResult<()> { + self.user_text_tx + .send(text) + .await + .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?; + Ok(()) + } + + pub(super) async fn handoff_out(&self, output_text: String) -> CodexResult<()> { + self.handoff.send_output(output_text).await + } + + pub(super) async fn handoff_complete(&self) -> CodexResult<()> { + Ok(()) + } + + pub(super) async fn active_handoff_id(&self) -> Option { + self.handoff.active_handoff.lock().await.clone() + } + + pub(super) async fn clear_active_handoff(&self) { + *self.handoff.active_handoff.lock().await = None; + } + + pub(super) async fn shutdown(self) { + self.realtime_active.store(false, Ordering::Relaxed); + self.task.abort(); + let _ = self.task.await; + } + + pub(super) fn realtime_active(&self) -> Arc { + Arc::clone(&self.realtime_active) + } +} + +pub(super) async fn start( + api_provider: ApiProvider, + extra_headers: Option, + prompt: String, + model: Option, + session_id: Option, +) -> CodexResult<(ConversationState, Receiver)> { + let session_config = RealtimeSessionConfig { + instructions: prompt, + model, + session_id, + mode: RealtimeApiMode::V1, + }; + let client = RealtimeWebsocketClient::new(api_provider); + let connection = client + .connect( + session_config, + extra_headers.unwrap_or_default(), + default_headers(), + ) + .await + .map_err(map_api_error)?; + + let writer = connection.writer(); + let events = connection.events(); + let (audio_tx, audio_rx) = + async_channel::bounded::(AUDIO_IN_QUEUE_CAPACITY); + let (user_text_tx, user_text_rx) = + async_channel::bounded::(USER_TEXT_IN_QUEUE_CAPACITY); + let (handoff_output_tx, handoff_output_rx) = + async_channel::bounded::(HANDOFF_OUT_QUEUE_CAPACITY); + let (events_tx, events_rx) = + async_channel::bounded::(OUTPUT_EVENTS_QUEUE_CAPACITY); + + let realtime_active = Arc::new(AtomicBool::new(true)); + let handoff = RealtimeHandoffState::new(handoff_output_tx); + let task = spawn_realtime_input_task( + writer, + events, + user_text_rx, + handoff_output_rx, + audio_rx, + events_tx, + handoff.clone(), + ); + + Ok(( + ConversationState { + audio_tx, + user_text_tx, + handoff, + task, + realtime_active, + }, + events_rx, + )) +} + +fn spawn_realtime_input_task( + writer: RealtimeWebsocketWriter, + events: RealtimeWebsocketEvents, + user_text_rx: Receiver, + handoff_output_rx: Receiver, + audio_rx: Receiver, + events_tx: Sender, + handoff_state: RealtimeHandoffState, +) -> JoinHandle<()> { + tokio::spawn(async move { + loop { + tokio::select! { + text = user_text_rx.recv() => { + match text { + Ok(text) => { + if let Err(err) = writer.send_conversation_item_create(text).await { + let mapped_error = map_api_error(err); + warn!("failed to send input text: {mapped_error}"); + break; + } + } + Err(_) => break, + } + } + 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; + } + } + Err(_) => break, + } + } + event = events.next_event() => { + match event { + Ok(Some(event)) => { + if let RealtimeEvent::HandoffRequested(handoff) = &event { + *handoff_state.active_handoff.lock().await = + Some(handoff.handoff_id.clone()); + } + let should_stop = matches!(&event, RealtimeEvent::Error(_)); + if events_tx.send(event).await.is_err() { + break; + } + if should_stop { + error!("realtime stream error event received"); + break; + } + } + Ok(None) => { + let _ = events_tx + .send(RealtimeEvent::Error( + "realtime websocket connection is closed".to_string(), + )) + .await; + break; + } + Err(err) => { + let mapped_error = map_api_error(err); + if events_tx + .send(RealtimeEvent::Error(mapped_error.to_string())) + .await + .is_err() + { + break; + } + error!("realtime stream closed: {mapped_error}"); + break; + } + } + } + frame = audio_rx.recv() => { + match frame { + Ok(frame) => { + if let Err(err) = writer.send_audio_frame(frame).await { + let mapped_error = map_api_error(err); + error!("failed to send input audio: {mapped_error}"); + break; + } + } + Err(_) => break, + } + } + } + } + }) +} + +#[cfg(test)] +mod tests { + use super::HandoffOutput; + use super::RealtimeHandoffState; + use async_channel::bounded; + use pretty_assertions::assert_eq; + + #[tokio::test] + async fn sends_handoff_outputs_when_handoff_is_active() { + let (tx, rx) = bounded(2); + let state = RealtimeHandoffState::new(tx); + *state.active_handoff.lock().await = Some("handoff_1".to_string()); + + state + .send_output("legacy output".to_string()) + .await + .expect("send"); + + let output = rx.recv().await.expect("recv"); + assert_eq!( + output, + HandoffOutput { + handoff_id: "handoff_1".to_string(), + output_text: "legacy output".to_string(), + } + ); + } + + #[tokio::test] + async fn ignores_handoff_output_when_handoff_is_not_active() { + let (tx, rx) = bounded(1); + let state = RealtimeHandoffState::new(tx); + + state + .send_output("ignored".to_string()) + .await + .expect("send"); + + assert!(rx.is_empty()); + } +} diff --git a/codex-rs/core/src/realtime_conversation/v2.rs b/codex-rs/core/src/realtime_conversation/v2.rs new file mode 100644 index 0000000000..58c6f98b54 --- /dev/null +++ b/codex-rs/core/src/realtime_conversation/v2.rs @@ -0,0 +1,416 @@ +use crate::api_bridge::map_api_error; +use crate::default_client::default_headers; +use crate::error::CodexErr; +use crate::error::Result as CodexResult; +use async_channel::Receiver; +use async_channel::Sender; +use async_channel::TrySendError; +use codex_api::Provider as ApiProvider; +use codex_api::RealtimeApiMode; +use codex_api::RealtimeAudioFrame; +use codex_api::RealtimeEvent; +use codex_api::RealtimeSessionConfig; +use codex_api::RealtimeWebsocketClient; +use codex_api::endpoint::realtime_websocket::RealtimeWebsocketEvents; +use codex_api::endpoint::realtime_websocket::RealtimeWebsocketWriter; +use http::HeaderMap; +use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; +use tokio::sync::Mutex; +use tokio::task::JoinHandle; +use tracing::error; +use tracing::warn; + +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; + +#[derive(Clone, Debug)] +struct RealtimeHandoffState { + output_tx: Sender, + active_handoff: Arc>>, + last_output_text: Arc>>, +} + +#[derive(Debug, PartialEq, Eq)] +enum HandoffOutput { + TextUpdate { + handoff_id: String, + output_text: String, + }, + FinalToolCall { + call_id: String, + output_text: String, + }, +} + +impl RealtimeHandoffState { + fn new(output_tx: Sender) -> Self { + Self { + output_tx, + active_handoff: Arc::new(Mutex::new(None)), + last_output_text: Arc::new(Mutex::new(None)), + } + } + + async fn send_output(&self, output_text: String) -> CodexResult<()> { + 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::TextUpdate { + handoff_id, + output_text, + }) + .await + .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(()) + } +} + +pub(super) struct ConversationState { + audio_tx: Sender, + user_text_tx: Sender, + handoff: RealtimeHandoffState, + task: JoinHandle<()>, + realtime_active: Arc, +} + +impl ConversationState { + pub(super) fn is_running(&self) -> bool { + self.realtime_active.load(Ordering::Relaxed) + } + + pub(super) async fn audio_in(&self, frame: RealtimeAudioFrame) -> CodexResult<()> { + match self.audio_tx.try_send(frame) { + Ok(()) => Ok(()), + Err(TrySendError::Full(_)) => { + warn!("dropping input audio frame due to full queue"); + Ok(()) + } + Err(TrySendError::Closed(_)) => Err(CodexErr::InvalidRequest( + "conversation is not running".to_string(), + )), + } + } + + pub(super) async fn text_in(&self, text: String) -> CodexResult<()> { + self.user_text_tx + .send(text) + .await + .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?; + Ok(()) + } + + pub(super) async fn handoff_out(&self, output_text: String) -> CodexResult<()> { + self.handoff.send_output(output_text).await + } + + pub(super) async fn handoff_complete(&self) -> CodexResult<()> { + self.handoff.send_final_output().await + } + + pub(super) async fn active_handoff_id(&self) -> Option { + self.handoff.active_handoff.lock().await.clone() + } + + pub(super) async fn clear_active_handoff(&self) { + *self.handoff.active_handoff.lock().await = None; + *self.handoff.last_output_text.lock().await = None; + } + + pub(super) async fn shutdown(self) { + self.realtime_active.store(false, Ordering::Relaxed); + self.task.abort(); + let _ = self.task.await; + } + + pub(super) fn realtime_active(&self) -> Arc { + Arc::clone(&self.realtime_active) + } +} + +pub(super) async fn start( + api_provider: ApiProvider, + extra_headers: Option, + prompt: String, + model: Option, + session_id: Option, +) -> CodexResult<(ConversationState, Receiver)> { + let session_config = RealtimeSessionConfig { + instructions: prompt, + model, + session_id, + mode: RealtimeApiMode::V2, + }; + let client = RealtimeWebsocketClient::new(api_provider); + let connection = client + .connect( + session_config, + extra_headers.unwrap_or_default(), + default_headers(), + ) + .await + .map_err(map_api_error)?; + + let writer = connection.writer(); + let events = connection.events(); + let (audio_tx, audio_rx) = + async_channel::bounded::(AUDIO_IN_QUEUE_CAPACITY); + let (user_text_tx, user_text_rx) = + async_channel::bounded::(USER_TEXT_IN_QUEUE_CAPACITY); + let (handoff_output_tx, handoff_output_rx) = + async_channel::bounded::(HANDOFF_OUT_QUEUE_CAPACITY); + let (events_tx, events_rx) = + async_channel::bounded::(OUTPUT_EVENTS_QUEUE_CAPACITY); + + let realtime_active = Arc::new(AtomicBool::new(true)); + let handoff = RealtimeHandoffState::new(handoff_output_tx); + let task = spawn_realtime_input_task( + writer, + events, + user_text_rx, + handoff_output_rx, + audio_rx, + events_tx, + handoff.clone(), + ); + + Ok(( + ConversationState { + audio_tx, + user_text_tx, + handoff, + task, + realtime_active, + }, + events_rx, + )) +} + +fn spawn_realtime_input_task( + writer: RealtimeWebsocketWriter, + events: RealtimeWebsocketEvents, + user_text_rx: Receiver, + handoff_output_rx: Receiver, + audio_rx: Receiver, + events_tx: Sender, + handoff_state: RealtimeHandoffState, +) -> JoinHandle<()> { + tokio::spawn(async move { + loop { + tokio::select! { + text = user_text_rx.recv() => { + match text { + Ok(text) => { + if let Err(err) = writer.send_conversation_item_create(text).await { + let mapped_error = map_api_error(err); + warn!("failed to send input text: {mapped_error}"); + break; + } + } + Err(_) => break, + } + } + handoff_output = handoff_output_rx.recv() => { + match handoff_output { + Ok(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; + } + } + Ok(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, + } + } + event = events.next_event() => { + match event { + Ok(Some(event)) => { + 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() { + break; + } + if should_stop { + error!("realtime stream error event received"); + break; + } + } + Ok(None) => { + let _ = events_tx + .send(RealtimeEvent::Error( + "realtime websocket connection is closed".to_string(), + )) + .await; + break; + } + Err(err) => { + let mapped_error = map_api_error(err); + if events_tx + .send(RealtimeEvent::Error(mapped_error.to_string())) + .await + .is_err() + { + break; + } + error!("realtime stream closed: {mapped_error}"); + break; + } + } + } + frame = audio_rx.recv() => { + match frame { + Ok(frame) => { + if let Err(err) = writer.send_audio_frame(frame).await { + let mapped_error = map_api_error(err); + error!("failed to send input audio: {mapped_error}"); + break; + } + } + Err(_) => break, + } + } + } + } + }) +} + +#[cfg(test)] +mod tests { + use super::HandoffOutput; + use super::RealtimeHandoffState; + use async_channel::bounded; + use pretty_assertions::assert_eq; + + #[tokio::test] + async fn clears_active_handoff_explicitly() { + let (tx, _rx) = bounded(1); + let state = RealtimeHandoffState::new(tx); + + *state.active_handoff.lock().await = Some("handoff_1".to_string()); + assert_eq!( + state.active_handoff.lock().await.clone(), + Some("handoff_1".to_string()) + ); + + *state.active_handoff.lock().await = None; + assert_eq!(state.active_handoff.lock().await.clone(), None); + } + + #[tokio::test] + async fn sends_multiple_handoff_outputs_until_cleared() { + let (tx, rx) = bounded(4); + let state = RealtimeHandoffState::new(tx); + + state + .send_output("ignored".to_string()) + .await + .expect("send"); + assert!(rx.is_empty()); + + *state.active_handoff.lock().await = Some("handoff_1".to_string()); + state.send_output("result".to_string()).await.expect("send"); + state + .send_output("result 2".to_string()) + .await + .expect("send"); + + let output_1 = rx.recv().await.expect("recv"); + assert_eq!( + output_1, + HandoffOutput::TextUpdate { + handoff_id: "handoff_1".to_string(), + output_text: "result".to_string(), + } + ); + + let output_2 = rx.recv().await.expect("recv"); + assert_eq!( + output_2, + HandoffOutput::TextUpdate { + handoff_id: "handoff_1".to_string(), + output_text: "result 2".to_string(), + } + ); + + *state.active_handoff.lock().await = None; + state + .send_output("ignored after clear".to_string()) + .await + .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(), + } + ); + } +}