openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-29 19:08:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Separate realtime v2 and v1 runtime behavior

Browse Source
This commit is contained in:
Ahmed Ibrahim
2026-03-04 21:17:13 -08:00
parent 374bff8cc0
commit 1e083ef2dc
14 changed files with 787 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/codex_message_processor.rs
+1 -1
View File
@@ -5188,7 +5188,7 @@ impl CodexMessageProcessor {
}
}
if !thread.enabled(Feature::RealtimeConversation) {
if !thread.realtime_voice_enabled() {
self.send_invalid_request_error(
request_id,
format!("thread {thread_id} does not support realtime conversation"),
codex-rs/app-server/tests/suite/v2/realtime_conversation.rs
+104 -6
View File
@@ -37,6 +37,13 @@ use tokio::time::timeout;
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum RealtimeFeatureGate {
RealtimeV2,
RealtimeConversation,
Disabled,
}
#[tokio::test]
async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
skip_if_no_network!(Ok(()));
@@ -77,7 +84,7 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
codex_home.path(),
&responses_server.uri(),
realtime_server.uri(),
true,
RealtimeFeatureGate::RealtimeV2,
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
@@ -178,6 +185,11 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
let connections = realtime_server.connections();
assert_eq!(connections.len(), 1);
let connection = &connections[0];
assert_eq!(connection[0].body_json()["type"], json!("session.update"));
assert_eq!(
connection[0].body_json()["session"]["type"],
json!("realtime")
);
assert_eq!(connection.len(), 3);
assert_eq!(
connection[0].body_json()["type"].as_str(),
@@ -225,7 +237,7 @@ async fn realtime_conversation_stop_emits_closed_notification() -> Result<()> {
codex_home.path(),
&responses_server.uri(),
realtime_server.uri(),
true,
RealtimeFeatureGate::RealtimeV2,
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
@@ -297,7 +309,7 @@ async fn realtime_conversation_requires_feature_flag() -> Result<()> {
codex_home.path(),
&responses_server.uri(),
realtime_server.uri(),
false,
RealtimeFeatureGate::Disabled,
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
@@ -337,6 +349,74 @@ async fn realtime_conversation_requires_feature_flag() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn realtime_conversation_accepts_legacy_feature_flag() -> Result<()> {
skip_if_no_network!(Ok(()));
let responses_server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
let realtime_server = start_websocket_server(vec![vec![vec![json!({
"type": "session.updated",
"session": { "id": "sess_backend", "instructions": "backend prompt" }
})]]])
.await;
let codex_home = TempDir::new()?;
create_config_toml(
codex_home.path(),
&responses_server.uri(),
realtime_server.uri(),
RealtimeFeatureGate::RealtimeConversation,
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
mcp.initialize().await?;
login_with_api_key(&mut mcp, "sk-test-key").await?;
let thread_start_request_id = mcp
.send_thread_start_request(ThreadStartParams::default())
.await?;
let thread_start_response: JSONRPCResponse = timeout(
DEFAULT_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(thread_start_request_id)),
)
.await??;
let thread_start: ThreadStartResponse = to_response(thread_start_response)?;
let start_request_id = mcp
.send_thread_realtime_start_request(ThreadRealtimeStartParams {
thread_id: thread_start.thread.id.clone(),
prompt: "backend prompt".to_string(),
session_id: None,
})
.await?;
let start_response: JSONRPCResponse = timeout(
DEFAULT_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(start_request_id)),
)
.await??;
let _: ThreadRealtimeStartResponse = to_response(start_response)?;
let started =
read_notification::<ThreadRealtimeStartedNotification>(&mut mcp, "thread/realtime/started")
.await?;
assert_eq!(started.thread_id, thread_start.thread.id);
let connections = realtime_server.connections();
assert_eq!(connections.len(), 1);
let connection = &connections[0];
assert_eq!(connection[0].body_json()["type"], json!("session.update"));
assert_eq!(
connection[0].body_json()["session"]["type"],
json!("quicksilver")
);
assert_eq!(
realtime_server.handshakes()[0].uri(),
"/v1/realtime?intent=quicksilver"
);
realtime_server.shutdown().await;
Ok(())
}
async fn read_notification<T: DeserializeOwned>(mcp: &mut McpProcess, method: &str) -> Result<T> {
let notification = timeout(
DEFAULT_TIMEOUT,
@@ -366,13 +446,31 @@ fn create_config_toml(
codex_home: &Path,
responses_server_uri: &str,
realtime_server_uri: &str,
realtime_enabled: bool,
realtime_feature_gate: RealtimeFeatureGate,
) -> std::io::Result<()> {
let realtime_feature_key = FEATURES
let realtime_v2_feature_key = FEATURES
.iter()
.find(|spec| spec.id == Feature::RealtimeV2)
.map(|spec| spec.key)
.unwrap_or("realtime_v2");
let realtime_conversation_feature_key = FEATURES
.iter()
.find(|spec| spec.id == Feature::RealtimeConversation)
.map(|spec| spec.key)
.unwrap_or("realtime_conversation");
let feature_toggles = match realtime_feature_gate {
RealtimeFeatureGate::RealtimeV2 => {
format!("{realtime_v2_feature_key} = true\n{realtime_conversation_feature_key} = false")
}
RealtimeFeatureGate::RealtimeConversation => {
format!("{realtime_v2_feature_key} = false\n{realtime_conversation_feature_key} = true")
}
RealtimeFeatureGate::Disabled => {
format!(
"{realtime_v2_feature_key} = false\n{realtime_conversation_feature_key} = false"
)
}
};
std::fs::write(
codex_home.join("config.toml"),
@@ -385,7 +483,7 @@ model_provider = "mock_provider"
experimental_realtime_ws_base_url = "{realtime_server_uri}"
[features]
{realtime_feature_key} = {realtime_enabled}
{feature_toggles}
[model_providers.mock_provider]
name = "Mock provider for test"
codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs
+197 -58
View File
@@ -1,20 +1,26 @@
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::SessionAudio;
use crate::endpoint::realtime_websocket::protocol::SessionAudioFormat;
use crate::endpoint::realtime_websocket::protocol::SessionAudioInput;
use crate::endpoint::realtime_websocket::protocol::SessionAudioOutput;
use crate::endpoint::realtime_websocket::protocol::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::error::ApiError;
use crate::provider::Provider;
@@ -200,12 +206,14 @@ pub struct RealtimeWebsocketConnection {
pub struct RealtimeWebsocketWriter {
stream: Arc<WsStream>,
is_closed: Arc<AtomicBool>,
mode: RealtimeApiMode,
}
#[derive(Clone)]
pub struct RealtimeWebsocketEvents {
rx_message: Arc<Mutex<mpsc::UnboundedReceiver<Result<Message, WsError>>>>,
is_closed: Arc<AtomicBool>,
mode: RealtimeApiMode,
}
impl RealtimeWebsocketConnection {
@@ -260,6 +268,7 @@ impl RealtimeWebsocketConnection {
fn new(
stream: WsStream,
rx_message: mpsc::UnboundedReceiver<Result<Message, WsError>>,
mode: RealtimeApiMode,
) -> Self {
let stream = Arc::new(stream);
let is_closed = Arc::new(AtomicBool::new(false));
@@ -267,10 +276,12 @@ impl RealtimeWebsocketConnection {
writer: RealtimeWebsocketWriter {
stream: Arc::clone(&stream),
is_closed: Arc::clone(&is_closed),
mode,
},
events: RealtimeWebsocketEvents {
rx_message: Arc::new(Mutex::new(rx_message)),
is_closed,
mode,
},
}
}
@@ -283,13 +294,14 @@ 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(),
};
self.send_json(RealtimeOutboundMessage::ConversationItemCreate {
item: ConversationItem::Message {
role: "user".to_string(),
content: vec![ConversationItemContent {
kind: "input_text".to_string(),
text,
}],
content: vec![ConversationItemContent { kind, text }],
},
})
.await
@@ -297,19 +309,30 @@ impl RealtimeWebsocketWriter {
pub async fn send_conversation_handoff_append(
&self,
_handoff_id: String,
handoff_id: String,
output_text: String,
) -> Result<(), ApiError> {
self.send_json(RealtimeOutboundMessage::ConversationItemCreate {
item: ConversationItem::Message {
role: "assistant".to_string(),
content: vec![ConversationItemContent {
kind: "output_text".to_string(),
text: output_text,
}],
},
})
.await
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
}
}
}
pub async fn send_function_call_output(
@@ -317,29 +340,54 @@ impl RealtimeWebsocketWriter {
call_id: String,
output_text: String,
) -> Result<(), ApiError> {
let output = json!({
"content": output_text,
})
.to_string();
self.send_json(RealtimeOutboundMessage::ConversationItemCreate {
item: ConversationItem::FunctionCallOutput { call_id, output },
})
.await
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
}
}
}
pub async fn send_response_create(&self) -> Result<(), ApiError> {
self.send_json(RealtimeOutboundMessage::ResponseCreate)
.await
match self.mode {
RealtimeApiMode::V1 => Ok(()),
RealtimeApiMode::V2 => {
self.send_json(RealtimeOutboundMessage::ResponseCreate)
.await
}
}
}
pub async fn send_session_update(&self, instructions: String) -> Result<(), ApiError> {
self.send_json(RealtimeOutboundMessage::SessionUpdate {
session: SessionUpdateSession {
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: SessionAudio {
input: SessionAudioInput {
audio: SessionAudioV2 {
input: SessionAudioInputV2 {
format: SessionAudioFormat {
kind: "audio/pcm".to_string(),
rate: 24_000,
@@ -350,7 +398,7 @@ impl RealtimeWebsocketWriter {
create_response: true,
},
},
output: SessionAudioOutput {
output: SessionAudioOutputV2 {
format: SessionAudioOutputFormat {
kind: "audio/pcm".to_string(),
rate: 24_000,
@@ -376,9 +424,11 @@ impl RealtimeWebsocketWriter {
},
}],
tool_choice: "auto".to_string(),
},
})
.await
}),
};
self.send_json(RealtimeOutboundMessage::SessionUpdate { session })
.await
}
pub async fn close(&self) -> Result<(), ApiError> {
@@ -415,6 +465,10 @@ impl RealtimeWebsocketWriter {
}
impl RealtimeWebsocketEvents {
fn mode(&self) -> RealtimeApiMode {
self.mode
}
pub async fn next_event(&self) -> Result<Option<RealtimeEvent>, ApiError> {
if self.is_closed.load(Ordering::SeqCst) {
return Ok(None);
@@ -439,7 +493,7 @@ impl RealtimeWebsocketEvents {
match msg {
Message::Text(text) => {
if let Some(event) = parse_realtime_event(&text) {
if let Some(event) = parse_realtime_event(&text, self.mode()) {
debug!(?event, "realtime websocket parsed event");
return Ok(Some(event));
}
@@ -485,6 +539,7 @@ impl RealtimeWebsocketClient {
self.provider.base_url.as_str(),
self.provider.query_params.as_ref(),
config.model.as_deref(),
config.mode,
)?;
let mut request = ws_url
@@ -512,7 +567,7 @@ impl RealtimeWebsocketClient {
);
let (stream, rx_message) = WsStream::new(stream);
let connection = RealtimeWebsocketConnection::new(stream, rx_message);
let connection = RealtimeWebsocketConnection::new(stream, rx_message, config.mode);
debug!(
session_id = config.session_id.as_deref().unwrap_or("<none>"),
"realtime websocket sending session.update"
@@ -564,6 +619,7 @@ fn websocket_url_from_api_url(
api_url: &str,
query_params: Option<&HashMap<String, String>>,
model: Option<&str>,
mode: RealtimeApiMode,
) -> Result<Url, ApiError> {
let mut url = Url::parse(api_url)
.map_err(|err| ApiError::Stream(format!("failed to parse realtime api_url: {err}")))?;
@@ -583,16 +639,19 @@ fn websocket_url_from_api_url(
}
}
let has_additional_query_params = query_params
.is_some_and(|params| params.keys().any(|key| key != "model" || model.is_none()));
if model.is_some() || has_additional_query_params {
{
let mut query = url.query_pairs_mut();
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() {
if (key == "model" && model.is_some())
|| (key == "intent" && mode == RealtimeApiMode::V1)
{
continue;
}
query.append_pair(key, value);
@@ -653,7 +712,7 @@ mod tests {
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str()),
parse_realtime_event(payload.as_str(), RealtimeApiMode::V2),
Some(RealtimeEvent::SessionUpdated {
session_id: "sess_123".to_string(),
instructions: Some("backend prompt".to_string()),
@@ -672,7 +731,7 @@ mod tests {
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str()),
parse_realtime_event(payload.as_str(), RealtimeApiMode::V2),
Some(RealtimeEvent::AudioOut(RealtimeAudioFrame {
data: "AAA=".to_string(),
sample_rate: 48000,
@@ -691,7 +750,7 @@ mod tests {
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str()),
parse_realtime_event(payload.as_str(), RealtimeApiMode::V2),
Some(RealtimeEvent::AudioOut(RealtimeAudioFrame {
data: "AAA=".to_string(),
sample_rate: 24_000,
@@ -709,7 +768,7 @@ mod tests {
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str()),
parse_realtime_event(payload.as_str(), RealtimeApiMode::V2),
Some(RealtimeEvent::ConversationItemAdded(
json!({"type": "message", "seq": 7})
))
@@ -724,7 +783,7 @@ mod tests {
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str()),
parse_realtime_event(payload.as_str(), RealtimeApiMode::V2),
Some(RealtimeEvent::ConversationItemDone {
item_id: "item_123".to_string(),
})
@@ -750,7 +809,7 @@ mod tests {
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str()),
parse_realtime_event(payload.as_str(), RealtimeApiMode::V2),
Some(RealtimeEvent::HandoffRequested(RealtimeHandoffRequested {
handoff_id: "handoff_123".to_string(),
item_id: "item_123".to_string(),
@@ -763,6 +822,54 @@ mod tests {
);
}
#[test]
fn parse_v1_handoff_requested_event() {
let payload = json!({
"type": "conversation.handoff.requested",
"handoff_id": "handoff_legacy",
"item_id": "item_legacy",
"input_transcript": "delegate legacy",
"messages": [
{"role": "user", "text": "delegate legacy"}
]
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str(), RealtimeApiMode::V1),
Some(RealtimeEvent::HandoffRequested(RealtimeHandoffRequested {
handoff_id: "handoff_legacy".to_string(),
item_id: "item_legacy".to_string(),
input_transcript: "delegate legacy".to_string(),
messages: vec![RealtimeHandoffMessage {
role: "user".to_string(),
text: "delegate legacy".to_string(),
}],
}))
);
}
#[test]
fn parse_v1_audio_delta_event() {
let payload = json!({
"type": "conversation.output_audio.delta",
"delta": "AAA=",
"sample_rate": 48000,
"channels": 1,
"samples_per_channel": 960
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str(), RealtimeApiMode::V1),
Some(RealtimeEvent::AudioOut(RealtimeAudioFrame {
data: "AAA=".to_string(),
sample_rate: 48000,
num_channels: 1,
samples_per_channel: Some(960),
}))
);
}
#[test]
fn parse_unknown_event_as_conversation_item_added() {
let payload = json!({
@@ -772,7 +879,7 @@ mod tests {
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str()),
parse_realtime_event(payload.as_str(), RealtimeApiMode::V2),
Some(RealtimeEvent::ConversationItemAdded(json!({
"type": "response.output_text.delta",
"delta": "hello",
@@ -817,15 +924,20 @@ mod tests {
#[test]
fn websocket_url_from_http_base_defaults_to_ws_path() {
let url =
websocket_url_from_api_url("http://127.0.0.1:8011", None, None).expect("build ws url");
websocket_url_from_api_url("http://127.0.0.1:8011", None, None, RealtimeApiMode::V2)
.expect("build ws url");
assert_eq!(url.as_str(), "ws://127.0.0.1:8011/v1/realtime");
}
#[test]
fn websocket_url_from_ws_base_defaults_to_ws_path() {
let url =
websocket_url_from_api_url("wss://example.com", None, Some("realtime-test-model"))
.expect("build ws url");
let url = websocket_url_from_api_url(
"wss://example.com",
None,
Some("realtime-test-model"),
RealtimeApiMode::V2,
)
.expect("build ws url");
assert_eq!(
url.as_str(),
"wss://example.com/v1/realtime?model=realtime-test-model"
@@ -834,8 +946,13 @@ mod tests {
#[test]
fn websocket_url_from_v1_base_appends_realtime_path() {
let url = websocket_url_from_api_url("https://api.openai.com/v1", None, Some("snapshot"))
.expect("build ws url");
let url = websocket_url_from_api_url(
"https://api.openai.com/v1",
None,
Some("snapshot"),
RealtimeApiMode::V2,
)
.expect("build ws url");
assert_eq!(
url.as_str(),
"wss://api.openai.com/v1/realtime?model=snapshot"
@@ -844,9 +961,13 @@ mod tests {
#[test]
fn websocket_url_from_nested_v1_base_appends_realtime_path() {
let url =
websocket_url_from_api_url("https://example.com/openai/v1", None, Some("snapshot"))
.expect("build ws url");
let url = websocket_url_from_api_url(
"https://example.com/openai/v1",
None,
Some("snapshot"),
RealtimeApiMode::V2,
)
.expect("build ws url");
assert_eq!(
url.as_str(),
"wss://example.com/openai/v1/realtime?model=snapshot"
@@ -859,6 +980,7 @@ mod tests {
"https://example.com/v1/realtime?foo=bar",
Some(&HashMap::from([("trace".to_string(), "1".to_string())])),
Some("snapshot"),
RealtimeApiMode::V2,
)
.expect("build ws url");
assert_eq!(
@@ -867,6 +989,21 @@ mod tests {
);
}
#[test]
fn websocket_url_v1_mode_adds_quicksilver_intent() {
let url = websocket_url_from_api_url(
"https://example.com/v1/realtime?foo=bar",
Some(&HashMap::from([("trace".to_string(), "1".to_string())])),
Some("snapshot"),
RealtimeApiMode::V1,
)
.expect("build ws url");
assert_eq!(
url.as_str(),
"wss://example.com/v1/realtime?foo=bar&intent=quicksilver&model=snapshot&trace=1"
);
}
#[tokio::test]
async fn e2e_connect_and_exchange_events_against_mock_ws_server() {
let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
@@ -1076,6 +1213,7 @@ mod tests {
instructions: "backend prompt".to_string(),
model: Some("realtime-test-model".to_string()),
session_id: Some("conv_1".to_string()),
mode: RealtimeApiMode::V2,
},
HeaderMap::new(),
HeaderMap::new(),
@@ -1224,6 +1362,7 @@ mod tests {
instructions: "backend prompt".to_string(),
model: Some("realtime-test-model".to_string()),
session_id: Some("conv_1".to_string()),
mode: RealtimeApiMode::V2,
},
HeaderMap::new(),
HeaderMap::new(),
codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs
+1
View File
@@ -7,4 +7,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;
codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs
+183 -68
View File
@@ -8,11 +8,18 @@ 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)]
@@ -20,6 +27,11 @@ pub struct RealtimeSessionConfig {
pub(super) enum RealtimeOutboundMessage {
#[serde(rename = "input_audio_buffer.append")]
InputAudioBufferAppend { audio: String },
#[serde(rename = "conversation.handoff.append")]
ConversationHandoffAppend {
handoff_id: String,
output_text: String,
},
#[serde(rename = "response.create")]
ResponseCreate,
#[serde(rename = "session.update")]
@@ -29,24 +41,50 @@ pub(super) enum RealtimeOutboundMessage {
}
#[derive(Debug, Clone, Serialize)]
pub(super) struct SessionUpdateSession {
#[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: SessionAudio,
pub(super) audio: SessionAudioV2,
pub(super) tools: Vec<SessionTool>,
pub(super) tool_choice: String,
}
#[derive(Debug, Clone, Serialize)]
pub(super) struct SessionAudio {
pub(super) input: SessionAudioInput,
pub(super) output: SessionAudioOutput,
pub(super) struct SessionAudioV1 {
pub(super) input: SessionAudioInputV1,
pub(super) output: SessionAudioOutputV1,
}
#[derive(Debug, Clone, Serialize)]
pub(super) struct SessionAudioInput {
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,
}
@@ -67,7 +105,12 @@ pub(super) struct SessionTurnDetection {
}
#[derive(Debug, Clone, Serialize)]
pub(super) struct SessionAudioOutput {
pub(super) struct SessionAudioOutputV1 {
pub(super) voice: String,
}
#[derive(Debug, Clone, Serialize)]
pub(super) struct SessionAudioOutputV2 {
pub(super) format: SessionAudioOutputFormat,
pub(super) voice: String,
}
@@ -127,7 +170,7 @@ pub(super) struct ConversationItemContent {
pub(super) text: String,
}
pub(super) fn parse_realtime_event(payload: &str) -> Option<RealtimeEvent> {
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) => {
@@ -143,52 +186,44 @@ pub(super) fn parse_realtime_event(payload: &str) -> Option<RealtimeEvent> {
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.created" | "session.updated" => {
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,
})
}
"response.output_audio.delta" => {
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()),
}))
"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()
@@ -201,30 +236,110 @@ pub(super) fn parse_realtime_event(payload: &str) -> Option<RealtimeEvent> {
.map(str::to_string)
.map(|item_id| RealtimeEvent::ConversationItemDone { item_id }),
"response.done" => {
if let Some(handoff) = parse_handoff_requested(&parsed) {
if let Some(handoff) = parse_handoff_requested_v2(&parsed) {
return Some(RealtimeEvent::HandoffRequested(handoff));
}
Some(RealtimeEvent::ConversationItemAdded(parsed))
}
"error" => 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),
"error" => parse_realtime_error(&parsed),
_ => Some(RealtimeEvent::ConversationItemAdded(parsed)),
}
}
fn parse_handoff_requested(parsed: &Value) -> Option<RealtimeHandoffRequested> {
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)
codex-rs/codex-api/src/lib.rs
+1
View File
@@ -27,6 +27,7 @@ pub use crate::common::create_text_param_for_request;
pub use crate::endpoint::compact::CompactClient;
pub use crate::endpoint::memories::MemoriesClient;
pub use crate::endpoint::models::ModelsClient;
pub use crate::endpoint::realtime_websocket::RealtimeApiMode;
pub use crate::endpoint::realtime_websocket::RealtimeSessionConfig;
pub use crate::endpoint::realtime_websocket::RealtimeWebsocketClient;
pub use crate::endpoint::realtime_websocket::RealtimeWebsocketConnection;
codex-rs/codex-api/tests/realtime_websocket_e2e.rs
+5
View File
@@ -2,6 +2,7 @@ use std::collections::HashMap;
use std::future::Future;
use std::time::Duration;
use codex_api::RealtimeApiMode;
use codex_api::RealtimeAudioFrame;
use codex_api::RealtimeEvent;
use codex_api::RealtimeSessionConfig;
@@ -163,6 +164,7 @@ async fn realtime_ws_e2e_session_create_and_event_flow() {
instructions: "backend prompt".to_string(),
model: Some("realtime-test-model".to_string()),
session_id: Some("conv_123".to_string()),
mode: RealtimeApiMode::V2,
},
HeaderMap::new(),
HeaderMap::new(),
@@ -255,6 +257,7 @@ async fn realtime_ws_e2e_send_while_next_event_waits() {
instructions: "backend prompt".to_string(),
model: Some("realtime-test-model".to_string()),
session_id: Some("conv_123".to_string()),
mode: RealtimeApiMode::V2,
},
HeaderMap::new(),
HeaderMap::new(),
@@ -318,6 +321,7 @@ async fn realtime_ws_e2e_disconnected_emitted_once() {
instructions: "backend prompt".to_string(),
model: Some("realtime-test-model".to_string()),
session_id: Some("conv_123".to_string()),
mode: RealtimeApiMode::V2,
},
HeaderMap::new(),
HeaderMap::new(),
@@ -378,6 +382,7 @@ async fn realtime_ws_e2e_forwards_unknown_text_events() {
instructions: "backend prompt".to_string(),
model: Some("realtime-test-model".to_string()),
session_id: Some("conv_123".to_string()),
mode: RealtimeApiMode::V2,
},
HeaderMap::new(),
HeaderMap::new(),
codex-rs/core/config.schema.json
+7 -1
View File
@@ -398,6 +398,9 @@
"prevent_idle_sleep": {
"type": "boolean"
},
"realtime_v2": {
"type": "boolean"
},
"realtime_conversation": {
"type": "boolean"
},
@@ -1789,6 +1792,9 @@
"prevent_idle_sleep": {
"type": "boolean"
},
"realtime_v2": {
"type": "boolean"
},
"realtime_conversation": {
"type": "boolean"
},
@@ -2248,4 +2254,4 @@
},
"title": "ConfigToml",
"type": "object"
}
}
codex-rs/core/src/codex_thread.rs
+9
View File
@@ -4,6 +4,7 @@ use crate::codex::SteerInputError;
use crate::config::ConstraintResult;
use crate::error::Result as CodexResult;
use crate::features::Feature;
use crate::features::RealtimeVoiceMode;
use crate::file_watcher::WatchRegistration;
use crate::protocol::Event;
use crate::protocol::Op;
@@ -143,4 +144,12 @@ impl CodexThread {
pub fn enabled(&self, feature: Feature) -> bool {
self.codex.enabled(feature)
}
pub fn realtime_voice_mode(&self) -> Option<RealtimeVoiceMode> {
self.codex.features().realtime_voice_mode()
}
pub fn realtime_voice_enabled(&self) -> bool {
self.realtime_voice_mode().is_some()
}
}
codex-rs/core/src/features.rs
+82
View File
@@ -155,6 +155,8 @@ pub enum Feature {
FastMode,
/// Enable voice transcription in the TUI composer.
VoiceTranscription,
/// Enable realtime voice conversation mode in the TUI.
RealtimeV2,
/// Enable experimental realtime voice conversation mode in the TUI.
RealtimeConversation,
/// Prevent idle system sleep while a turn is actively running.
@@ -165,6 +167,12 @@ pub enum Feature {
ResponsesWebsocketsV2,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RealtimeVoiceMode {
V1,
V2,
}
impl Feature {
pub fn key(self) -> &'static str {
self.info().key
@@ -237,6 +245,18 @@ impl Features {
self.enabled.contains(&f)
}
pub fn realtime_voice_mode(&self) -> Option<RealtimeVoiceMode> {
if self.enabled(Feature::RealtimeV2) {
return Some(RealtimeVoiceMode::V2);
}
self.enabled(Feature::RealtimeConversation)
.then_some(RealtimeVoiceMode::V1)
}
pub fn realtime_voice_enabled(&self) -> bool {
self.realtime_voice_mode().is_some()
}
pub fn enable(&mut self, f: Feature) -> &mut Self {
self.enabled.insert(f);
self
@@ -717,6 +737,12 @@ pub const FEATURES: &[FeatureSpec] = &[
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::RealtimeV2,
key: "realtime_v2",
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::RealtimeConversation,
key: "realtime_conversation",
@@ -887,4 +913,60 @@ mod tests {
assert_eq!(feature_for_key("multi_agent"), Some(Feature::Collab));
assert_eq!(feature_for_key("collab"), Some(Feature::Collab));
}
#[test]
fn realtime_voice_enabled_is_false_when_no_realtime_flags_are_enabled() {
let features = Features::with_defaults();
assert_eq!(features.realtime_voice_enabled(), false);
}
#[test]
fn realtime_voice_enabled_is_true_when_realtime_v2_is_enabled() {
let mut features = Features::with_defaults();
features.enable(Feature::RealtimeV2);
assert_eq!(features.realtime_voice_enabled(), true);
}
#[test]
fn realtime_voice_enabled_is_true_when_legacy_realtime_conversation_is_enabled() {
let mut features = Features::with_defaults();
features.enable(Feature::RealtimeConversation);
assert_eq!(features.realtime_voice_enabled(), true);
}
#[test]
fn realtime_voice_enabled_is_true_when_both_realtime_flags_are_enabled() {
let mut features = Features::with_defaults();
features.enable(Feature::RealtimeV2);
features.enable(Feature::RealtimeConversation);
assert_eq!(features.realtime_voice_enabled(), true);
}
#[test]
fn realtime_voice_mode_is_none_when_no_realtime_flags_are_enabled() {
let features = Features::with_defaults();
assert_eq!(features.realtime_voice_mode(), None);
}
#[test]
fn realtime_voice_mode_is_v2_when_realtime_v2_is_enabled() {
let mut features = Features::with_defaults();
features.enable(Feature::RealtimeV2);
assert_eq!(features.realtime_voice_mode(), Some(RealtimeVoiceMode::V2));
}
#[test]
fn realtime_voice_mode_is_v1_when_only_legacy_realtime_flag_is_enabled() {
let mut features = Features::with_defaults();
features.enable(Feature::RealtimeConversation);
assert_eq!(features.realtime_voice_mode(), Some(RealtimeVoiceMode::V1));
}
#[test]
fn realtime_voice_mode_prefers_v2_when_both_flags_are_enabled() {
let mut features = Features::with_defaults();
features.enable(Feature::RealtimeV2);
features.enable(Feature::RealtimeConversation);
assert_eq!(features.realtime_voice_mode(), Some(RealtimeVoiceMode::V2));
}
}
codex-rs/core/src/realtime_conversation.rs
+48 -6
View File
@@ -5,10 +5,12 @@ 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;
@@ -54,6 +56,7 @@ 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)]
@@ -69,11 +72,12 @@ enum HandoffOutput {
}
impl RealtimeHandoffState {
fn new(output_tx: Sender<HandoffOutput>) -> Self {
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,
}
}
@@ -94,6 +98,9 @@ impl RealtimeHandoffState {
}
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(());
};
@@ -116,6 +123,7 @@ struct ConversationState {
audio_tx: Sender<RealtimeAudioFrame>,
user_text_tx: Sender<String>,
handoff: RealtimeHandoffState,
mode: RealtimeApiMode,
task: JoinHandle<()>,
realtime_active: Arc<AtomicBool>,
}
@@ -141,6 +149,7 @@ impl RealtimeConversationManager {
extra_headers: Option<HeaderMap>,
prompt: String,
model: Option<String>,
mode: RealtimeApiMode,
session_id: Option<String>,
) -> CodexResult<(Receiver<RealtimeEvent>, Arc<AtomicBool>)> {
let previous_state = {
@@ -157,6 +166,7 @@ impl RealtimeConversationManager {
instructions: prompt,
model,
session_id,
mode,
};
let client = RealtimeWebsocketClient::new(api_provider);
let connection = client
@@ -180,7 +190,7 @@ impl RealtimeConversationManager {
async_channel::bounded::<RealtimeEvent>(OUTPUT_EVENTS_QUEUE_CAPACITY);
let realtime_active = Arc::new(AtomicBool::new(true));
let handoff = RealtimeHandoffState::new(handoff_output_tx);
let handoff = RealtimeHandoffState::new(handoff_output_tx, mode);
let task = spawn_realtime_input_task(
writer,
events,
@@ -196,6 +206,7 @@ impl RealtimeConversationManager {
audio_tx,
user_text_tx,
handoff,
mode,
task,
realtime_active: Arc::clone(&realtime_active),
});
@@ -321,7 +332,20 @@ pub(crate) async fn handle_start(
.experimental_realtime_ws_backend_prompt
.clone()
.unwrap_or(params.prompt);
let model = Some(DEFAULT_REALTIME_MODEL.to_string());
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
@@ -336,6 +360,7 @@ pub(crate) async fn handle_start(
extra_headers,
prompt,
model,
realtime_api_mode,
requested_session_id.clone(),
)
.await
@@ -644,6 +669,7 @@ mod tests {
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;
@@ -699,7 +725,7 @@ mod tests {
#[tokio::test]
async fn clears_active_handoff_explicitly() {
let (tx, _rx) = bounded(1);
let state = RealtimeHandoffState::new(tx);
let state = RealtimeHandoffState::new(tx, RealtimeApiMode::V2);
*state.active_handoff.lock().await = Some("handoff_1".to_string());
assert_eq!(
@@ -714,7 +740,7 @@ mod tests {
#[tokio::test]
async fn sends_multiple_handoff_outputs_until_cleared() {
let (tx, rx) = bounded(4);
let state = RealtimeHandoffState::new(tx);
let state = RealtimeHandoffState::new(tx, RealtimeApiMode::V2);
state
.send_output("ignored".to_string())
@@ -758,7 +784,7 @@ mod tests {
#[tokio::test]
async fn sends_final_tool_call_output_for_active_handoff() {
let (tx, rx) = bounded(4);
let state = RealtimeHandoffState::new(tx);
let state = RealtimeHandoffState::new(tx, RealtimeApiMode::V2);
*state.active_handoff.lock().await = Some("handoff_2".to_string());
state
@@ -777,4 +803,20 @@ mod tests {
}
);
}
#[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());
}
}
codex-rs/core/tests/suite/realtime_conversation.rs
+90
View File
@@ -1,6 +1,7 @@
use anyhow::Result;
use codex_core::CodexAuth;
use codex_core::auth::OPENAI_API_KEY_ENV_VAR;
use codex_core::features::Feature;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::ConversationAudioParams;
use codex_protocol::protocol::ConversationStartParams;
@@ -177,6 +178,95 @@ async fn conversation_start_audio_text_close_round_trip() -> Result<()> {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn conversation_start_legacy_feature_uses_v1_realtime_protocol() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_websocket_server(vec![
vec![],
vec![
vec![json!({
"type": "session.updated",
"session": { "id": "sess_legacy", "instructions": "backend prompt" }
})],
vec![json!({
"type": "conversation.output_audio.delta",
"delta": "AQID",
"sample_rate": 24000,
"channels": 1
})],
vec![],
],
])
.await;
let mut builder = test_codex().with_config(|config| {
config
.features
.enable(Feature::RealtimeConversation)
.expect("enable legacy realtime feature");
config
.features
.disable(Feature::RealtimeV2)
.expect("disable realtime_v2 feature");
});
let test = builder.build_with_websocket_server(&server).await?;
assert!(server.wait_for_handshakes(1, Duration::from_secs(2)).await);
test.codex
.submit(Op::RealtimeConversationStart(ConversationStartParams {
prompt: "backend prompt".to_string(),
session_id: None,
}))
.await?;
let _started = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::RealtimeConversationStarted(started) => Some(Ok(started.clone())),
EventMsg::Error(err) => Some(Err(err.clone())),
_ => None,
})
.await
.unwrap_or_else(|err: ErrorEvent| panic!("conversation start failed: {err:?}"));
test.codex
.submit(Op::RealtimeConversationText(ConversationTextParams {
text: "hello legacy".to_string(),
}))
.await?;
let audio_out = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::AudioOut(frame),
}) => Some(frame.clone()),
_ => None,
})
.await;
assert_eq!(audio_out.data, "AQID");
let connections = server.connections();
assert_eq!(connections.len(), 2);
let connection = &connections[1];
assert_eq!(connection[0].body_json()["type"], json!("session.update"));
assert_eq!(
connection[0].body_json()["session"]["type"],
json!("quicksilver")
);
assert_eq!(
server.handshakes()[1].uri(),
"/v1/realtime?intent=quicksilver&model=realtime-test-model"
);
assert_eq!(
connection[1].body_json()["type"],
json!("conversation.item.create")
);
assert_eq!(
connection[1].body_json()["item"]["content"][0]["type"],
json!("text")
);
server.shutdown().await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth() -> Result<()> {
skip_if_no_network!(Ok(()));
codex-rs/tui/src/chatwidget.rs
+2 -3
View File
@@ -926,8 +926,7 @@ enum ReplayKind {
impl ChatWidget {
fn realtime_conversation_enabled(&self) -> bool {
self.config.features.enabled(Feature::RealtimeConversation)
&& cfg!(not(target_os = "linux"))
self.config.features.realtime_voice_enabled() && cfg!(not(target_os = "linux"))
}
fn realtime_audio_device_selection_enabled(&self) -> bool {
@@ -7074,7 +7073,7 @@ impl ChatWidget {
if feature == Feature::VoiceTranscription {
self.bottom_pane.set_voice_transcription_enabled(enabled);
}
if feature == Feature::RealtimeConversation {
if matches!(feature, Feature::RealtimeConversation | Feature::RealtimeV2) {
let realtime_conversation_enabled = self.realtime_conversation_enabled();
self.bottom_pane
.set_realtime_conversation_enabled(realtime_conversation_enabled);
codex-rs/tui/src/chatwidget/tests.rs
+57
View File
@@ -1882,6 +1882,63 @@ fn set_chatgpt_auth(chat: &mut ChatWidget) {
));
}
#[tokio::test]
async fn realtime_feature_gate_uses_dual_read_feature_flags() {
let (mut chat, _rx, _op_rx) = make_chatwidget_manual(None).await;
let platform_realtime_supported = cfg!(not(target_os = "linux"));
assert_eq!(chat.realtime_conversation_enabled(), false);
chat.set_feature_enabled(Feature::RealtimeV2, true);
assert_eq!(
chat.realtime_conversation_enabled(),
platform_realtime_supported
);
chat.set_feature_enabled(Feature::RealtimeV2, false);
assert_eq!(chat.realtime_conversation_enabled(), false);
chat.set_feature_enabled(Feature::RealtimeConversation, true);
assert_eq!(
chat.realtime_conversation_enabled(),
platform_realtime_supported
);
}
#[tokio::test]
async fn disabling_realtime_v2_resets_live_realtime_session() {
let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(None).await;
chat.set_feature_enabled(Feature::RealtimeV2, true);
chat.realtime_conversation.phase = super::realtime::RealtimeConversationPhase::Active;
chat.set_feature_enabled(Feature::RealtimeV2, false);
assert!(!chat.realtime_conversation.is_live());
assert!(matches!(
chat.realtime_conversation.phase,
super::realtime::RealtimeConversationPhase::Inactive
));
assert!(matches!(
op_rx.try_recv(),
Ok(Op::RealtimeConversationClose)
));
}
#[tokio::test]
async fn legacy_realtime_feature_keeps_realtime_enabled_when_realtime_v2_is_off() {
let (mut chat, _rx, _op_rx) = make_chatwidget_manual(None).await;
let platform_realtime_supported = cfg!(not(target_os = "linux"));
chat.set_feature_enabled(Feature::RealtimeConversation, true);
chat.set_feature_enabled(Feature::RealtimeV2, true);
chat.set_feature_enabled(Feature::RealtimeV2, false);
assert_eq!(
chat.realtime_conversation_enabled(),
platform_realtime_supported
);
}
#[tokio::test]
async fn prefetch_rate_limits_is_gated_on_chatgpt_auth_provider() {
let (mut chat, _rx, _op_rx) = make_chatwidget_manual(None).await;