openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-06 20:20:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Align main realtime v2 flow with port branch

Browse Source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Ahmed Ibrahim
2026-03-15 14:58:50 -07:00
co-authored by Codex
parent 7fa5201365
commit bc1a7cf2f4
25 changed files with 1066 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs
+290 -28
View File
@@ -14,8 +14,11 @@ use crate::endpoint::realtime_websocket::protocol::SessionAudio;
use crate::endpoint::realtime_websocket::protocol::SessionAudioFormat;
use crate::endpoint::realtime_websocket::protocol::SessionAudioInput;
use crate::endpoint::realtime_websocket::protocol::SessionAudioOutput;
use crate::endpoint::realtime_websocket::protocol::SessionAudioOutputFormat;
use crate::endpoint::realtime_websocket::protocol::SessionAudioVoice;
use crate::endpoint::realtime_websocket::protocol::SessionFunctionTool;
use crate::endpoint::realtime_websocket::protocol::SessionNoiseReduction;
use crate::endpoint::realtime_websocket::protocol::SessionTurnDetection;
use crate::endpoint::realtime_websocket::protocol::SessionUpdateSession;
use crate::endpoint::realtime_websocket::protocol::parse_realtime_event;
use crate::error::ApiError;
@@ -48,10 +51,15 @@ use tungstenite::protocol::WebSocketConfig;
use url::Url;
const REALTIME_AUDIO_SAMPLE_RATE: u32 = 24_000;
const REALTIME_AUDIO_FORMAT: &str = "audio/pcm";
const REALTIME_V2_NOISE_REDUCTION: &str = "near_field";
const REALTIME_V2_TURN_DETECTION: &str = "server_vad";
const REALTIME_V2_OUTPUT_MODALITY_AUDIO: &str = "audio";
const REALTIME_V2_TOOL_CHOICE: &str = "auto";
const REALTIME_V1_SESSION_TYPE: &str = "quicksilver";
const REALTIME_V2_SESSION_TYPE: &str = "realtime";
const REALTIME_V2_CODEX_TOOL_NAME: &str = "codex";
const REALTIME_V2_CODEX_TOOL_DESCRIPTION: &str = "Delegate work to Codex and return the result.";
const REALTIME_V2_CODEX_TOOL_DESCRIPTION: &str = "Delegate a request to Codex and return the final result to the user. Use this as the default action. If the user asks to do something next, later, after this, or once current work finishes, call this tool so the work is actually queued instead of merely promising to do it later.";
fn normalized_session_mode(
event_parser: RealtimeEventParser,
@@ -254,6 +262,21 @@ impl RealtimeWebsocketConnection {
.await
}
pub async fn send_response_create(&self) -> Result<(), ApiError> {
self.writer.send_response_create().await
}
pub async fn send_conversation_item_truncate(
&self,
item_id: String,
content_index: u32,
audio_end_ms: u32,
) -> Result<(), ApiError> {
self.writer
.send_conversation_item_truncate(item_id, content_index, audio_end_ms)
.await
}
pub async fn close(&self) -> Result<(), ApiError> {
self.writer.close().await
}
@@ -341,28 +364,106 @@ impl RealtimeWebsocketWriter {
self.send_json(message).await
}
pub async fn send_response_create(&self) -> Result<(), ApiError> {
self.send_json(RealtimeOutboundMessage::ResponseCreate)
.await
}
pub async fn send_conversation_item_truncate(
&self,
item_id: String,
content_index: u32,
audio_end_ms: u32,
) -> Result<(), ApiError> {
self.send_json(RealtimeOutboundMessage::ConversationItemTruncate {
item_id,
content_index,
audio_end_ms,
})
.await
}
pub async fn send_session_update(
&self,
instructions: String,
session_mode: RealtimeSessionMode,
) -> Result<(), ApiError> {
let session_mode = normalized_session_mode(self.event_parser, session_mode);
let (session_kind, session_instructions, output_audio) = match session_mode {
RealtimeSessionMode::Conversational => {
let kind = match self.event_parser {
RealtimeEventParser::V1 => REALTIME_V1_SESSION_TYPE.to_string(),
RealtimeEventParser::RealtimeV2 => REALTIME_V2_SESSION_TYPE.to_string(),
};
let voice = match self.event_parser {
RealtimeEventParser::V1 => SessionAudioVoice::Fathom,
RealtimeEventParser::RealtimeV2 => SessionAudioVoice::Alloy,
};
(kind, Some(instructions), Some(SessionAudioOutput { voice }))
}
RealtimeSessionMode::Transcription => ("transcription".to_string(), None, None),
};
let tools = match (self.event_parser, session_mode) {
(RealtimeEventParser::RealtimeV2, RealtimeSessionMode::Conversational) => {
let (session_kind, session_instructions, output_modalities, input_audio, output_audio) =
match session_mode {
RealtimeSessionMode::Conversational => {
let kind = match self.event_parser {
RealtimeEventParser::V1 => REALTIME_V1_SESSION_TYPE.to_string(),
RealtimeEventParser::RealtimeV2 => REALTIME_V2_SESSION_TYPE.to_string(),
};
let input = match self.event_parser {
RealtimeEventParser::V1 => SessionAudioInput {
format: SessionAudioFormat {
kind: REALTIME_AUDIO_FORMAT.to_string(),
rate: REALTIME_AUDIO_SAMPLE_RATE,
},
noise_reduction: None,
turn_detection: None,
},
RealtimeEventParser::RealtimeV2 => SessionAudioInput {
format: SessionAudioFormat {
kind: REALTIME_AUDIO_FORMAT.to_string(),
rate: REALTIME_AUDIO_SAMPLE_RATE,
},
noise_reduction: Some(SessionNoiseReduction {
kind: REALTIME_V2_NOISE_REDUCTION.to_string(),
}),
turn_detection: Some(SessionTurnDetection {
kind: REALTIME_V2_TURN_DETECTION.to_string(),
interrupt_response: true,
create_response: true,
}),
},
};
let output = match self.event_parser {
RealtimeEventParser::V1 => SessionAudioOutput {
format: None,
voice: SessionAudioVoice::Fathom,
},
RealtimeEventParser::RealtimeV2 => SessionAudioOutput {
format: Some(SessionAudioOutputFormat {
kind: REALTIME_AUDIO_FORMAT.to_string(),
rate: REALTIME_AUDIO_SAMPLE_RATE,
}),
voice: SessionAudioVoice::Marin,
},
};
let output_modalities = match self.event_parser {
RealtimeEventParser::V1 => None,
RealtimeEventParser::RealtimeV2 => {
Some(vec![REALTIME_V2_OUTPUT_MODALITY_AUDIO.to_string()])
}
};
(
kind,
Some(instructions),
output_modalities,
input,
Some(output),
)
}
RealtimeSessionMode::Transcription => (
"transcription".to_string(),
None,
None,
SessionAudioInput {
format: SessionAudioFormat {
kind: REALTIME_AUDIO_FORMAT.to_string(),
rate: REALTIME_AUDIO_SAMPLE_RATE,
},
noise_reduction: None,
turn_detection: None,
},
None,
),
};
let (tools, tool_choice) = match (self.event_parser, session_mode) {
(RealtimeEventParser::RealtimeV2, RealtimeSessionMode::Conversational) => (
Some(vec![SessionFunctionTool {
kind: "function".to_string(),
name: REALTIME_V2_CODEX_TOOL_NAME.to_string(),
@@ -372,32 +473,38 @@ impl RealtimeWebsocketWriter {
"properties": {
"prompt": {
"type": "string",
"description": "Prompt text for the delegated Codex task."
"description": "The user request to delegate to Codex."
}
},
"required": ["prompt"],
"additionalProperties": false
}),
}])
}
}]),
Some(REALTIME_V2_TOOL_CHOICE.to_string()),
),
(RealtimeEventParser::RealtimeV2, RealtimeSessionMode::Transcription)
| (RealtimeEventParser::V1, RealtimeSessionMode::Conversational)
| (RealtimeEventParser::V1, RealtimeSessionMode::Transcription) => None,
| (RealtimeEventParser::V1, RealtimeSessionMode::Transcription) => (None, None),
};
debug!(
event_parser = ?self.event_parser,
session_mode = ?session_mode,
instructions_len = session_instructions.as_ref().map(String::len).unwrap_or_default(),
has_output_audio = output_audio.is_some(),
has_tools = tools.is_some(),
"realtime websocket prepared session.update"
);
self.send_json(RealtimeOutboundMessage::SessionUpdate {
session: SessionUpdateSession {
kind: session_kind,
instructions: session_instructions,
output_modalities,
audio: SessionAudio {
input: SessionAudioInput {
format: SessionAudioFormat {
kind: "audio/pcm".to_string(),
rate: REALTIME_AUDIO_SAMPLE_RATE,
},
},
input: input_audio,
output: output_audio,
},
tools,
tool_choice,
},
})
.await
@@ -490,6 +597,7 @@ impl RealtimeWebsocketEvents {
async fn update_active_transcript(&self, event: &mut RealtimeEvent) {
let mut active_transcript = self.active_transcript.lock().await;
match event {
RealtimeEvent::InputAudioSpeechStarted(_) => {}
RealtimeEvent::InputTranscriptDelta(RealtimeTranscriptDelta { delta }) => {
append_transcript_delta(&mut active_transcript.entries, "user", delta);
}
@@ -497,10 +605,18 @@ impl RealtimeWebsocketEvents {
append_transcript_delta(&mut active_transcript.entries, "assistant", delta);
}
RealtimeEvent::HandoffRequested(handoff) => {
debug!(
handoff_id = handoff.handoff_id,
item_id = handoff.item_id,
input_len = handoff.input_transcript.len(),
transcript_entries = active_transcript.entries.len(),
"realtime websocket parsed codex function call"
);
handoff.active_transcript = std::mem::take(&mut active_transcript.entries);
}
RealtimeEvent::SessionUpdated { .. }
| RealtimeEvent::AudioOut(_)
| RealtimeEvent::ResponseCancelled(_)
| RealtimeEvent::ConversationItemAdded(_)
| RealtimeEvent::ConversationItemDone { .. }
| RealtimeEvent::Error(_) => {}
@@ -584,7 +700,11 @@ impl RealtimeWebsocketClient {
let (stream, rx_message) = WsStream::new(stream);
let connection = RealtimeWebsocketConnection::new(stream, rx_message, config.event_parser);
debug!(
event_parser = ?config.event_parser,
session_mode = ?config.session_mode,
model = config.model.as_deref().unwrap_or("<provider-default>"),
session_id = config.session_id.as_deref().unwrap_or("<none>"),
instructions_len = config.instructions.len(),
"realtime websocket sending session.update"
);
connection
@@ -717,6 +837,8 @@ mod tests {
use crate::endpoint::realtime_websocket::protocol::RealtimeHandoffRequested;
use crate::endpoint::realtime_websocket::protocol::RealtimeTranscriptDelta;
use crate::endpoint::realtime_websocket::protocol::RealtimeTranscriptEntry;
use codex_protocol::protocol::RealtimeInputAudioSpeechStarted;
use codex_protocol::protocol::RealtimeResponseCancelled;
use http::HeaderValue;
use pretty_assertions::assert_eq;
use serde_json::Value;
@@ -761,6 +883,7 @@ mod tests {
sample_rate: 48000,
num_channels: 1,
samples_per_channel: Some(960),
item_id: None,
}))
);
}
@@ -910,10 +1033,112 @@ mod tests {
sample_rate: 24_000,
num_channels: 1,
samples_per_channel: None,
item_id: None,
}))
);
}
#[test]
fn parse_realtime_v2_response_audio_delta_with_item_id() {
let payload = json!({
"type": "response.audio.delta",
"delta": "AQID",
"item_id": "item_audio_1"
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str(), RealtimeEventParser::RealtimeV2),
Some(RealtimeEvent::AudioOut(RealtimeAudioFrame {
data: "AQID".to_string(),
sample_rate: 24_000,
num_channels: 1,
samples_per_channel: None,
item_id: Some("item_audio_1".to_string()),
}))
);
}
#[test]
fn parse_realtime_v2_speech_started_event() {
let payload = json!({
"type": "input_audio_buffer.speech_started",
"item_id": "item_input_1"
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str(), RealtimeEventParser::RealtimeV2),
Some(RealtimeEvent::InputAudioSpeechStarted(
RealtimeInputAudioSpeechStarted {
item_id: Some("item_input_1".to_string()),
}
))
);
}
#[test]
fn parse_realtime_v2_response_cancelled_event() {
let payload = json!({
"type": "response.cancelled",
"response": {"id": "resp_cancelled_1"}
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str(), RealtimeEventParser::RealtimeV2),
Some(RealtimeEvent::ResponseCancelled(
RealtimeResponseCancelled {
response_id: Some("resp_cancelled_1".to_string()),
}
))
);
}
#[test]
fn parse_realtime_v2_response_done_handoff_event() {
let payload = json!({
"type": "response.done",
"response": {
"output": [{
"id": "item_123",
"type": "function_call",
"name": "codex",
"call_id": "call_123",
"arguments": "{\"prompt\":\"delegate from done\"}"
}]
}
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str(), RealtimeEventParser::RealtimeV2),
Some(RealtimeEvent::HandoffRequested(RealtimeHandoffRequested {
handoff_id: "call_123".to_string(),
item_id: "item_123".to_string(),
input_transcript: "delegate from done".to_string(),
active_transcript: Vec::new(),
}))
);
}
#[test]
fn parse_realtime_v2_response_created_event() {
let payload = json!({
"type": "response.created",
"response": {"id": "resp_created_1"}
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str(), RealtimeEventParser::RealtimeV2),
Some(RealtimeEvent::ConversationItemAdded(json!({
"type": "response.created",
"response": {"id": "resp_created_1"}
})))
);
}
#[test]
fn merge_request_headers_matches_http_precedence() {
let mut provider_headers = HeaderMap::new();
@@ -1270,6 +1495,7 @@ mod tests {
sample_rate: 48000,
num_channels: 1,
samples_per_channel: Some(960),
item_id: None,
})
.await
.expect("send audio");
@@ -1297,6 +1523,7 @@ mod tests {
sample_rate: 48000,
num_channels: 1,
samples_per_channel: None,
item_id: None,
})
);
@@ -1386,9 +1613,38 @@ mod tests {
first_json["session"]["type"],
Value::String("realtime".to_string())
);
assert_eq!(first_json["session"]["output_modalities"], json!(["audio"]));
assert_eq!(
first_json["session"]["audio"]["input"]["format"],
json!({
"type": "audio/pcm",
"rate": 24_000,
})
);
assert_eq!(
first_json["session"]["audio"]["input"]["noise_reduction"],
json!({
"type": "near_field",
})
);
assert_eq!(
first_json["session"]["audio"]["input"]["turn_detection"],
json!({
"type": "server_vad",
"interrupt_response": true,
"create_response": true,
})
);
assert_eq!(
first_json["session"]["audio"]["output"]["format"],
json!({
"type": "audio/pcm",
"rate": 24_000,
})
);
assert_eq!(
first_json["session"]["audio"]["output"]["voice"],
Value::String("alloy".to_string())
Value::String("marin".to_string())
);
assert_eq!(
first_json["session"]["tools"][0]["type"],
@@ -1402,6 +1658,10 @@ mod tests {
first_json["session"]["tools"][0]["parameters"]["required"],
json!(["prompt"])
);
assert_eq!(
first_json["session"]["tool_choice"],
Value::String("auto".to_string())
);
ws.send(Message::Text(
json!({
@@ -1612,6 +1872,7 @@ mod tests {
sample_rate: 24_000,
num_channels: 1,
samples_per_channel: Some(480),
item_id: None,
})
.await
.expect("send audio");
@@ -1791,6 +2052,7 @@ mod tests {
sample_rate: 48000,
num_channels: 1,
samples_per_channel: Some(960),
item_id: None,
}),
)
.await
codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs
+41 -2
View File
@@ -39,6 +39,14 @@ pub(super) enum RealtimeOutboundMessage {
handoff_id: String,
output_text: String,
},
#[serde(rename = "response.create")]
ResponseCreate,
#[serde(rename = "conversation.item.truncate")]
ConversationItemTruncate {
item_id: String,
content_index: u32,
audio_end_ms: u32,
},
#[serde(rename = "session.update")]
SessionUpdate { session: SessionUpdateSession },
#[serde(rename = "conversation.item.create")]
@@ -51,9 +59,13 @@ pub(super) struct SessionUpdateSession {
pub(super) kind: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub(super) instructions: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(super) output_modalities: Option<Vec<String>>,
pub(super) audio: SessionAudio,
#[serde(skip_serializing_if = "Option::is_none")]
pub(super) tools: Option<Vec<SessionFunctionTool>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(super) tool_choice: Option<String>,
}
#[derive(Debug, Clone, Serialize)]
@@ -66,6 +78,10 @@ pub(super) struct SessionAudio {
#[derive(Debug, Clone, Serialize)]
pub(super) struct SessionAudioInput {
pub(super) format: SessionAudioFormat,
#[serde(skip_serializing_if = "Option::is_none")]
pub(super) noise_reduction: Option<SessionNoiseReduction>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(super) turn_detection: Option<SessionTurnDetection>,
}
#[derive(Debug, Clone, Serialize)]
@@ -77,6 +93,8 @@ pub(super) struct SessionAudioFormat {
#[derive(Debug, Clone, Serialize)]
pub(super) struct SessionAudioOutput {
#[serde(skip_serializing_if = "Option::is_none")]
pub(super) format: Option<SessionAudioOutputFormat>,
pub(super) voice: SessionAudioVoice,
}
@@ -84,8 +102,29 @@ pub(super) struct SessionAudioOutput {
pub(super) enum SessionAudioVoice {
#[serde(rename = "fathom")]
Fathom,
#[serde(rename = "alloy")]
Alloy,
#[serde(rename = "marin")]
Marin,
}
#[derive(Debug, Clone, Serialize)]
pub(super) struct SessionNoiseReduction {
#[serde(rename = "type")]
pub(super) kind: String,
}
#[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 SessionAudioOutputFormat {
#[serde(rename = "type")]
pub(super) kind: String,
pub(super) rate: u32,
}
#[derive(Debug, Clone, Serialize)]
codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v1.rs
+1
View File
@@ -35,6 +35,7 @@ pub(super) fn parse_realtime_event_v1(payload: &str) -> Option<RealtimeEvent> {
.get("samples_per_channel")
.and_then(Value::as_u64)
.and_then(|value| u32::try_from(value).ok()),
item_id: None,
}))
}
"conversation.input_transcript.delta" => {
codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs
+59 -1
View File
@@ -5,6 +5,8 @@ use crate::endpoint::realtime_websocket::protocol_common::parse_transcript_delta
use codex_protocol::protocol::RealtimeAudioFrame;
use codex_protocol::protocol::RealtimeEvent;
use codex_protocol::protocol::RealtimeHandoffRequested;
use codex_protocol::protocol::RealtimeInputAudioSpeechStarted;
use codex_protocol::protocol::RealtimeResponseCancelled;
use serde_json::Map as JsonMap;
use serde_json::Value;
use tracing::debug;
@@ -19,7 +21,9 @@ pub(super) fn parse_realtime_event_v2(payload: &str) -> Option<RealtimeEvent> {
match message_type.as_str() {
"session.updated" => parse_session_updated_event(&parsed),
"response.output_audio.delta" => parse_output_audio_delta_event(&parsed),
"response.output_audio.delta" | "response.audio.delta" => {
parse_output_audio_delta_event(&parsed)
}
"conversation.item.input_audio_transcription.delta" => {
parse_transcript_delta_event(&parsed, "delta").map(RealtimeEvent::InputTranscriptDelta)
}
@@ -30,11 +34,37 @@ pub(super) fn parse_realtime_event_v2(payload: &str) -> Option<RealtimeEvent> {
"response.output_text.delta" | "response.output_audio_transcript.delta" => {
parse_transcript_delta_event(&parsed, "delta").map(RealtimeEvent::OutputTranscriptDelta)
}
"input_audio_buffer.speech_started" => Some(RealtimeEvent::InputAudioSpeechStarted(
RealtimeInputAudioSpeechStarted {
item_id: parsed
.get("item_id")
.and_then(Value::as_str)
.map(str::to_string),
},
)),
"conversation.item.added" => parsed
.get("item")
.cloned()
.map(RealtimeEvent::ConversationItemAdded),
"conversation.item.done" => parse_conversation_item_done_event(&parsed),
"response.created" => Some(RealtimeEvent::ConversationItemAdded(parsed)),
"response.done" => parse_response_done_event(parsed),
"response.cancelled" => Some(RealtimeEvent::ResponseCancelled(
RealtimeResponseCancelled {
response_id: parsed
.get("response")
.and_then(Value::as_object)
.and_then(|response| response.get("id"))
.and_then(Value::as_str)
.map(str::to_string)
.or_else(|| {
parsed
.get("response_id")
.and_then(Value::as_str)
.map(str::to_string)
}),
},
)),
"error" => parse_error_event(&parsed),
_ => {
debug!("received unsupported realtime v2 event type: {message_type}, data: {payload}");
@@ -67,6 +97,10 @@ fn parse_output_audio_delta_event(parsed: &Value) -> Option<RealtimeEvent> {
.get("samples_per_channel")
.and_then(Value::as_u64)
.and_then(|value| u32::try_from(value).ok()),
item_id: parsed
.get("item_id")
.and_then(Value::as_str)
.map(str::to_string),
}))
}
@@ -82,6 +116,30 @@ fn parse_conversation_item_done_event(parsed: &Value) -> Option<RealtimeEvent> {
.map(|item_id| RealtimeEvent::ConversationItemDone { item_id })
}
fn parse_response_done_event(parsed: Value) -> Option<RealtimeEvent> {
if let Some(handoff) = parse_response_done_handoff_requested_event(&parsed) {
return Some(handoff);
}
Some(RealtimeEvent::ConversationItemAdded(parsed))
}
fn parse_response_done_handoff_requested_event(parsed: &Value) -> Option<RealtimeEvent> {
let item = parsed
.get("response")
.and_then(Value::as_object)
.and_then(|response| response.get("output"))
.and_then(Value::as_array)?
.iter()
.find(|item| {
item.get("type").and_then(Value::as_str) == Some("function_call")
&& item.get("name").and_then(Value::as_str) == Some(CODEX_TOOL_NAME)
})?
.as_object()?;
parse_handoff_requested_event(item)
}
fn parse_handoff_requested_event(item: &JsonMap<String, Value>) -> Option<RealtimeEvent> {
let item_type = item.get("type").and_then(Value::as_str);
let item_name = item.get("name").and_then(Value::as_str);
codex-rs/codex-api/tests/realtime_websocket_e2e.rs
+3
View File
@@ -170,6 +170,7 @@ async fn realtime_ws_e2e_session_create_and_event_flow() {
sample_rate: 48000,
num_channels: 1,
samples_per_channel: Some(960),
item_id: None,
})
.await
.expect("send audio");
@@ -186,6 +187,7 @@ async fn realtime_ws_e2e_session_create_and_event_flow() {
sample_rate: 48000,
num_channels: 1,
samples_per_channel: None,
item_id: None,
})
);
@@ -254,6 +256,7 @@ async fn realtime_ws_e2e_send_while_next_event_waits() {
sample_rate: 48000,
num_channels: 1,
samples_per_channel: Some(960),
item_id: None,
}),
)
.await