mirror of
https://github.com/openai/codex.git
synced 2026-09-14 11:57:03 +00:00
Split realtime v1/v2 logic into isolated modules
This commit is contained in:
Ahmed Ibrahim
@@ -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<RealtimeEvent> {
|
||||
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;
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<String, String>>,
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
121
codex-rs/codex-api/src/endpoint/realtime_websocket/mode_v2.rs
Normal file
121
codex-rs/codex-api/src/endpoint/realtime_websocket/mode_v2.rs
Normal file
@@ -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<String, String>>,
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<String>,
|
||||
pub session_id: Option<String>,
|
||||
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<String>,
|
||||
pub(super) audio: SessionAudioV2,
|
||||
pub(super) tools: Vec<SessionTool>,
|
||||
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<String>,
|
||||
}
|
||||
|
||||
#[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<ConversationItemContent>,
|
||||
},
|
||||
#[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<RealtimeEvent> {
|
||||
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<RealtimeEvent> {
|
||||
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<RealtimeEvent> {
|
||||
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<RealtimeEvent> {
|
||||
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<RealtimeEvent> {
|
||||
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<RealtimeEvent> {
|
||||
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<RealtimeHandoffRequested> {
|
||||
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<RealtimeHandoffRequested> {
|
||||
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<RealtimeHandoffMessage>) {
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct HandoffArguments {
|
||||
#[serde(default)]
|
||||
prompt: Option<String>,
|
||||
#[serde(default)]
|
||||
text: Option<String>,
|
||||
#[serde(default)]
|
||||
input: Option<String>,
|
||||
#[serde(default)]
|
||||
message: Option<String>,
|
||||
#[serde(default)]
|
||||
input_transcript: Option<String>,
|
||||
#[serde(default)]
|
||||
messages: Vec<RealtimeHandoffMessage>,
|
||||
}
|
||||
|
||||
let Some(parsed) = serde_json::from_str::<HandoffArguments>(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::<Vec<_>>();
|
||||
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)
|
||||
}
|
||||
@@ -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<RealtimeEvent> {
|
||||
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<RealtimeEvent> {
|
||||
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<RealtimeEvent> {
|
||||
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<RealtimeEvent> {
|
||||
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<RealtimeEvent> {
|
||||
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,
|
||||
}))
|
||||
}
|
||||
@@ -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<RealtimeEvent> {
|
||||
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<RealtimeEvent> {
|
||||
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<RealtimeEvent> {
|
||||
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<RealtimeEvent> {
|
||||
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<RealtimeHandoffRequested> {
|
||||
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<RealtimeHandoffMessage>) {
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct HandoffArguments {
|
||||
#[serde(default)]
|
||||
prompt: Option<String>,
|
||||
#[serde(default)]
|
||||
text: Option<String>,
|
||||
#[serde(default)]
|
||||
input: Option<String>,
|
||||
#[serde(default)]
|
||||
message: Option<String>,
|
||||
#[serde(default)]
|
||||
input_transcript: Option<String>,
|
||||
#[serde(default)]
|
||||
messages: Vec<RealtimeHandoffMessage>,
|
||||
}
|
||||
|
||||
let Some(parsed) = serde_json::from_str::<HandoffArguments>(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::<Vec<_>>();
|
||||
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)
|
||||
}
|
||||
167
codex-rs/codex-api/src/endpoint/realtime_websocket/types.rs
Normal file
167
codex-rs/codex-api/src/endpoint/realtime_websocket/types.rs
Normal file
@@ -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<String>,
|
||||
pub session_id: Option<String>,
|
||||
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<String>,
|
||||
pub(super) audio: SessionAudioV2,
|
||||
pub(super) tools: Vec<SessionTool>,
|
||||
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<String>,
|
||||
}
|
||||
|
||||
#[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<ConversationItemContent>,
|
||||
},
|
||||
#[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,
|
||||
}
|
||||
@@ -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<Option<ConversationState>>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
struct RealtimeHandoffState {
|
||||
output_tx: Sender<HandoffOutput>,
|
||||
active_handoff: Arc<Mutex<Option<String>>>,
|
||||
last_output_text: Arc<Mutex<Option<String>>>,
|
||||
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<HandoffOutput>, 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<RealtimeAudioFrame>,
|
||||
user_text_tx: Sender<String>,
|
||||
handoff: RealtimeHandoffState,
|
||||
mode: RealtimeApiMode,
|
||||
task: JoinHandle<()>,
|
||||
realtime_active: Arc<AtomicBool>,
|
||||
}
|
||||
|
||||
#[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<HeaderMap>,
|
||||
prompt: String,
|
||||
model: Option<String>,
|
||||
mode: RealtimeApiMode,
|
||||
session_id: Option<String>,
|
||||
) -> CodexResult<(Receiver<RealtimeEvent>, Arc<AtomicBool>)> {
|
||||
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::<RealtimeAudioFrame>(AUDIO_IN_QUEUE_CAPACITY);
|
||||
let (user_text_tx, user_text_rx) =
|
||||
async_channel::bounded::<String>(USER_TEXT_IN_QUEUE_CAPACITY);
|
||||
let (handoff_output_tx, handoff_output_rx) =
|
||||
async_channel::bounded::<HandoffOutput>(HANDOFF_OUT_QUEUE_CAPACITY);
|
||||
let (events_tx, events_rx) =
|
||||
async_channel::bounded::<RealtimeEvent>(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<String> {
|
||||
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<Session>,
|
||||
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<Session>,
|
||||
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<String> {
|
||||
let messages = handoff
|
||||
.messages
|
||||
.iter()
|
||||
.map(|message| message.text.as_str())
|
||||
.collect::<Vec<_>>()
|
||||
.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<String> {
|
||||
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<Option<HeaderMap>> {
|
||||
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<Session>,
|
||||
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<Session>, 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<String>,
|
||||
handoff_output_rx: Receiver<HandoffOutput>,
|
||||
audio_rx: Receiver<RealtimeAudioFrame>,
|
||||
events_tx: Sender<RealtimeEvent>,
|
||||
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<Session>,
|
||||
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());
|
||||
}
|
||||
}
|
||||
94
codex-rs/core/src/realtime_conversation/common.rs
Normal file
94
codex-rs/core/src/realtime_conversation/common.rs
Normal file
@@ -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<String> {
|
||||
let messages = handoff
|
||||
.messages
|
||||
.iter()
|
||||
.map(|message| message.text.as_str())
|
||||
.collect::<Vec<_>>()
|
||||
.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<String> {
|
||||
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<Option<HeaderMap>> {
|
||||
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<Session>,
|
||||
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;
|
||||
}
|
||||
430
codex-rs/core/src/realtime_conversation/mod.rs
Normal file
430
codex-rs/core/src/realtime_conversation/mod.rs
Normal file
@@ -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<Option<ConversationState>>,
|
||||
}
|
||||
|
||||
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<String> {
|
||||
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<AtomicBool> {
|
||||
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<HeaderMap>,
|
||||
prompt: String,
|
||||
model: Option<String>,
|
||||
mode: RealtimeApiMode,
|
||||
session_id: Option<String>,
|
||||
) -> CodexResult<(Receiver<RealtimeEvent>, Arc<AtomicBool>)> {
|
||||
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<String> {
|
||||
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<Session>,
|
||||
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<Session>,
|
||||
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<Session>,
|
||||
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<Session>, 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);
|
||||
}
|
||||
}
|
||||
319
codex-rs/core/src/realtime_conversation/v1.rs
Normal file
319
codex-rs/core/src/realtime_conversation/v1.rs
Normal file
@@ -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<HandoffOutput>,
|
||||
active_handoff: Arc<Mutex<Option<String>>>,
|
||||
}
|
||||
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
struct HandoffOutput {
|
||||
handoff_id: String,
|
||||
output_text: String,
|
||||
}
|
||||
|
||||
impl RealtimeHandoffState {
|
||||
fn new(output_tx: Sender<HandoffOutput>) -> 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<RealtimeAudioFrame>,
|
||||
user_text_tx: Sender<String>,
|
||||
handoff: RealtimeHandoffState,
|
||||
task: JoinHandle<()>,
|
||||
realtime_active: Arc<AtomicBool>,
|
||||
}
|
||||
|
||||
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<String> {
|
||||
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<AtomicBool> {
|
||||
Arc::clone(&self.realtime_active)
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) async fn start(
|
||||
api_provider: ApiProvider,
|
||||
extra_headers: Option<HeaderMap>,
|
||||
prompt: String,
|
||||
model: Option<String>,
|
||||
session_id: Option<String>,
|
||||
) -> CodexResult<(ConversationState, Receiver<RealtimeEvent>)> {
|
||||
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::<RealtimeAudioFrame>(AUDIO_IN_QUEUE_CAPACITY);
|
||||
let (user_text_tx, user_text_rx) =
|
||||
async_channel::bounded::<String>(USER_TEXT_IN_QUEUE_CAPACITY);
|
||||
let (handoff_output_tx, handoff_output_rx) =
|
||||
async_channel::bounded::<HandoffOutput>(HANDOFF_OUT_QUEUE_CAPACITY);
|
||||
let (events_tx, events_rx) =
|
||||
async_channel::bounded::<RealtimeEvent>(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<String>,
|
||||
handoff_output_rx: Receiver<HandoffOutput>,
|
||||
audio_rx: Receiver<RealtimeAudioFrame>,
|
||||
events_tx: Sender<RealtimeEvent>,
|
||||
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());
|
||||
}
|
||||
}
|
||||
416
codex-rs/core/src/realtime_conversation/v2.rs
Normal file
416
codex-rs/core/src/realtime_conversation/v2.rs
Normal file
@@ -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<HandoffOutput>,
|
||||
active_handoff: Arc<Mutex<Option<String>>>,
|
||||
last_output_text: Arc<Mutex<Option<String>>>,
|
||||
}
|
||||
|
||||
#[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<HandoffOutput>) -> 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<RealtimeAudioFrame>,
|
||||
user_text_tx: Sender<String>,
|
||||
handoff: RealtimeHandoffState,
|
||||
task: JoinHandle<()>,
|
||||
realtime_active: Arc<AtomicBool>,
|
||||
}
|
||||
|
||||
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<String> {
|
||||
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<AtomicBool> {
|
||||
Arc::clone(&self.realtime_active)
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) async fn start(
|
||||
api_provider: ApiProvider,
|
||||
extra_headers: Option<HeaderMap>,
|
||||
prompt: String,
|
||||
model: Option<String>,
|
||||
session_id: Option<String>,
|
||||
) -> CodexResult<(ConversationState, Receiver<RealtimeEvent>)> {
|
||||
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::<RealtimeAudioFrame>(AUDIO_IN_QUEUE_CAPACITY);
|
||||
let (user_text_tx, user_text_rx) =
|
||||
async_channel::bounded::<String>(USER_TEXT_IN_QUEUE_CAPACITY);
|
||||
let (handoff_output_tx, handoff_output_rx) =
|
||||
async_channel::bounded::<HandoffOutput>(HANDOFF_OUT_QUEUE_CAPACITY);
|
||||
let (events_tx, events_rx) =
|
||||
async_channel::bounded::<RealtimeEvent>(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<String>,
|
||||
handoff_output_rx: Receiver<HandoffOutput>,
|
||||
audio_rx: Receiver<RealtimeAudioFrame>,
|
||||
events_tx: Sender<RealtimeEvent>,
|
||||
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(),
|
||||
}
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user