diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 2dd4409f27..fc348cc26d 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -2768,6 +2768,12 @@ "data": { "type": "string" }, + "itemId": { + "type": [ + "string", + "null" + ] + }, "numChannels": { "format": "uint16", "minimum": 0.0, diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index acfdfb9214..3f8875f809 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -2749,6 +2749,12 @@ "data": { "type": "string" }, + "itemId": { + "type": [ + "string", + "null" + ] + }, "numChannels": { "format": "uint16", "minimum": 0.0, diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 75054b0d6a..aec1fabb7e 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -12838,6 +12838,12 @@ "data": { "type": "string" }, + "itemId": { + "type": [ + "string", + "null" + ] + }, "numChannels": { "format": "uint16", "minimum": 0.0, diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index b4868f09ad..e617a17f97 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -10555,6 +10555,12 @@ "data": { "type": "string" }, + "itemId": { + "type": [ + "string", + "null" + ] + }, "numChannels": { "format": "uint16", "minimum": 0.0, diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeOutputAudioDeltaNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeOutputAudioDeltaNotification.json index d4df6194fa..6c75f67553 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeOutputAudioDeltaNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeOutputAudioDeltaNotification.json @@ -7,6 +7,12 @@ "data": { "type": "string" }, + "itemId": { + "type": [ + "string", + "null" + ] + }, "numChannels": { "format": "uint16", "minimum": 0.0, diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeAudioChunk.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeAudioChunk.ts index 078f642247..eefb79dd65 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeAudioChunk.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeAudioChunk.ts @@ -5,4 +5,4 @@ /** * EXPERIMENTAL - thread realtime audio chunk. */ -export type ThreadRealtimeAudioChunk = { data: string, sampleRate: number, numChannels: number, samplesPerChannel: number | null, }; +export type ThreadRealtimeAudioChunk = { data: string, sampleRate: number, numChannels: number, samplesPerChannel: number | null, itemId: string | null, }; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 75aa7768d1..73139a2e09 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -1577,6 +1577,7 @@ mod tests { sample_rate: 24_000, num_channels: 1, samples_per_channel: Some(512), + item_id: None, }, }, ); @@ -1589,7 +1590,8 @@ mod tests { "data": "AQID", "sampleRate": 24000, "numChannels": 1, - "samplesPerChannel": 512 + "samplesPerChannel": 512, + "itemId": null } } }), @@ -1641,6 +1643,7 @@ mod tests { sample_rate: 24_000, num_channels: 1, samples_per_channel: Some(512), + item_id: None, }, }, ); diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index e86900cbc7..d1f2759930 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -3650,6 +3650,7 @@ pub struct ThreadRealtimeAudioChunk { pub sample_rate: u32, pub num_channels: u16, pub samples_per_channel: Option, + pub item_id: Option, } impl From for ThreadRealtimeAudioChunk { @@ -3659,12 +3660,14 @@ impl From for ThreadRealtimeAudioChunk { sample_rate, num_channels, samples_per_channel, + item_id, } = value; Self { data, sample_rate, num_channels, samples_per_channel, + item_id, } } } @@ -3676,12 +3679,14 @@ impl From for CoreRealtimeAudioFrame { sample_rate, num_channels, samples_per_channel, + item_id, } = value; Self { data, sample_rate, num_channels, samples_per_channel, + item_id, } } } diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 4afd4cc244..0632308b85 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -351,6 +351,20 @@ pub(crate) async fn apply_bespoke_event_handling( if let ApiVersion::V2 = api_version { match event.payload { RealtimeEvent::SessionUpdated { .. } => {} + RealtimeEvent::InputAudioSpeechStarted(event) => { + let notification = ThreadRealtimeItemAddedNotification { + thread_id: conversation_id.to_string(), + item: serde_json::json!({ + "type": "input_audio_buffer.speech_started", + "item_id": event.item_id, + }), + }; + outgoing + .send_server_notification(ServerNotification::ThreadRealtimeItemAdded( + notification, + )) + .await; + } RealtimeEvent::InputTranscriptDelta(_) => {} RealtimeEvent::OutputTranscriptDelta(_) => {} RealtimeEvent::AudioOut(audio) => { @@ -364,6 +378,20 @@ pub(crate) async fn apply_bespoke_event_handling( ) .await; } + RealtimeEvent::ResponseCancelled(event) => { + let notification = ThreadRealtimeItemAddedNotification { + thread_id: conversation_id.to_string(), + item: serde_json::json!({ + "type": "response.cancelled", + "response_id": event.response_id, + }), + }; + outgoing + .send_server_notification(ServerNotification::ThreadRealtimeItemAdded( + notification, + )) + .await; + } RealtimeEvent::ConversationItemAdded(item) => { let notification = ThreadRealtimeItemAddedNotification { thread_id: conversation_id.to_string(), diff --git a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs index a771fb8744..7d472c5036 100644 --- a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs +++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs @@ -135,6 +135,7 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> { sample_rate: 24_000, num_channels: 1, samples_per_channel: Some(480), + item_id: None, }, }) .await?; diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs index e62649e42e..dbd9a5296f 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -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(""), session_id = config.session_id.as_deref().unwrap_or(""), + 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 diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs index 73c2c1052d..a6bef882ea 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs @@ -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, + #[serde(skip_serializing_if = "Option::is_none")] + pub(super) output_modalities: Option>, pub(super) audio: SessionAudio, #[serde(skip_serializing_if = "Option::is_none")] pub(super) tools: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub(super) tool_choice: Option, } #[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, + #[serde(skip_serializing_if = "Option::is_none")] + pub(super) turn_detection: Option, } #[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, 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)] diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v1.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v1.rs index 04e76fb447..b66cf2b24b 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v1.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v1.rs @@ -35,6 +35,7 @@ pub(super) fn parse_realtime_event_v1(payload: &str) -> Option { .get("samples_per_channel") .and_then(Value::as_u64) .and_then(|value| u32::try_from(value).ok()), + item_id: None, })) } "conversation.input_transcript.delta" => { diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs index 7ef318d3f9..b33007519e 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs @@ -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 { 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 { "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 { .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 { .map(|item_id| RealtimeEvent::ConversationItemDone { item_id }) } +fn parse_response_done_event(parsed: Value) -> Option { + 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 { + 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) -> Option { let item_type = item.get("type").and_then(Value::as_str); let item_name = item.get("name").and_then(Value::as_str); diff --git a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs index 30786ad92d..130ab6fd35 100644 --- a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs +++ b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs @@ -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 diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index d5a9ad7719..28a3703ada 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -37,6 +37,7 @@ use crate::parse_command::parse_command; use crate::parse_turn_item; use crate::realtime_conversation::RealtimeConversationManager; use crate::realtime_conversation::handle_audio as handle_realtime_conversation_audio; +use crate::realtime_conversation::handle_audio_truncate as handle_realtime_conversation_audio_truncate; use crate::realtime_conversation::handle_close as handle_realtime_conversation_close; use crate::realtime_conversation::handle_start as handle_realtime_conversation_start; use crate::realtime_conversation::handle_text as handle_realtime_conversation_text; @@ -2597,6 +2598,9 @@ impl Session { if !matches!(msg, EventMsg::TurnComplete(_)) { return; } + if let Err(err) = self.conversation.handoff_complete().await { + debug!("failed to finalize realtime handoff output: {err}"); + } self.conversation.clear_active_handoff().await; } @@ -4148,6 +4152,11 @@ async fn submission_loop(sess: Arc, config: Arc, rx_sub: Receiv handle_realtime_conversation_audio(&sess, sub.id.clone(), params).await; false } + Op::RealtimeConversationAudioTruncate(params) => { + handle_realtime_conversation_audio_truncate(&sess, sub.id.clone(), params) + .await; + false + } Op::RealtimeConversationText(params) => { handle_realtime_conversation_text(&sess, sub.id.clone(), params).await; false @@ -4349,7 +4358,7 @@ fn submission_dispatch_span(sub: &Submission) -> tracing::Span { let op_name = sub.op.kind(); let span_name = format!("op.dispatch.{op_name}"); let dispatch_span = match &sub.op { - Op::RealtimeConversationAudio(_) => { + Op::RealtimeConversationAudio(_) | Op::RealtimeConversationAudioTruncate(_) => { debug_span!( "submission_dispatch", otel.name = span_name.as_str(), diff --git a/codex-rs/core/src/codex_tests.rs b/codex-rs/core/src/codex_tests.rs index efb43b0e5b..84373a625b 100644 --- a/codex-rs/core/src/codex_tests.rs +++ b/codex-rs/core/src/codex_tests.rs @@ -2485,6 +2485,7 @@ fn submission_dispatch_span_uses_debug_for_realtime_audio() { sample_rate: 16_000, num_channels: 1, samples_per_channel: Some(160), + item_id: None, }, }), trace: None, diff --git a/codex-rs/core/src/realtime_context.rs b/codex-rs/core/src/realtime_context.rs index 8016763514..f2d603425e 100644 --- a/codex-rs/core/src/realtime_context.rs +++ b/codex-rs/core/src/realtime_context.rs @@ -1,8 +1,11 @@ use crate::codex::Session; +use crate::compact::content_items_to_text; +use crate::event_mapping::is_contextual_user_message_content; use crate::git_info::resolve_root_git_project_for_trust; use crate::truncate::TruncationPolicy; use crate::truncate::truncate_text; use chrono::Utc; +use codex_protocol::models::ResponseItem; use codex_state::SortKey; use codex_state::ThreadMetadata; use dirs::home_dir; @@ -19,9 +22,11 @@ use tracing::info; use tracing::warn; const STARTUP_CONTEXT_HEADER: &str = "Startup context from Codex.\nThis is background context about recent work and machine/workspace layout. It may be incomplete or stale. Use it to inform responses, and do not repeat it back unless relevant."; +const CURRENT_THREAD_SECTION_TOKEN_BUDGET: usize = 1_200; const RECENT_WORK_SECTION_TOKEN_BUDGET: usize = 2_200; const WORKSPACE_SECTION_TOKEN_BUDGET: usize = 1_600; const NOTES_SECTION_TOKEN_BUDGET: usize = 300; +const MAX_CURRENT_THREAD_TURNS: usize = 2; const MAX_RECENT_THREADS: usize = 40; const MAX_RECENT_WORK_GROUPS: usize = 8; const MAX_CURRENT_CWD_ASKS: usize = 8; @@ -49,20 +54,33 @@ pub(crate) async fn build_realtime_startup_context( ) -> Option { let config = sess.get_config().await; let cwd = config.cwd.clone(); + let history = sess.clone_history().await; + let current_thread_section = build_current_thread_section(history.raw_items()); let recent_threads = load_recent_threads(sess).await; let recent_work_section = build_recent_work_section(&cwd, &recent_threads); let workspace_section = build_workspace_section(&cwd); - if recent_work_section.is_none() && workspace_section.is_none() { + if current_thread_section.is_none() + && recent_work_section.is_none() + && workspace_section.is_none() + { debug!("realtime startup context unavailable; skipping injection"); return None; } let mut parts = vec![STARTUP_CONTEXT_HEADER.to_string()]; + let has_current_thread_section = current_thread_section.is_some(); let has_recent_work_section = recent_work_section.is_some(); let has_workspace_section = workspace_section.is_some(); + if let Some(section) = format_section( + "Current Thread", + current_thread_section, + CURRENT_THREAD_SECTION_TOKEN_BUDGET, + ) { + parts.push(section); + } if let Some(section) = format_section( "Recent Work", recent_work_section, @@ -79,7 +97,7 @@ pub(crate) async fn build_realtime_startup_context( } if let Some(section) = format_section( "Notes", - Some("Built at realtime startup from persisted thread metadata in the state DB and a bounded local workspace scan. This excludes repo memory instructions, AGENTS files, project-doc prompt blends, and memory summaries.".to_string()), + Some("Built at realtime startup from the current thread history, persisted thread metadata in the state DB, and a bounded local workspace scan. This excludes repo memory instructions, AGENTS files, project-doc prompt blends, and memory summaries.".to_string()), NOTES_SECTION_TOKEN_BUDGET, ) { parts.push(section); @@ -89,6 +107,7 @@ pub(crate) async fn build_realtime_startup_context( debug!( approx_tokens = approx_token_count(&context), bytes = context.len(), + has_current_thread_section, has_recent_work_section, has_workspace_section, "built realtime startup context" @@ -167,6 +186,90 @@ fn build_recent_work_section(cwd: &Path, recent_threads: &[ThreadMetadata]) -> O (!sections.is_empty()).then(|| sections.join("\n\n")) } +fn build_current_thread_section(items: &[ResponseItem]) -> Option { + let mut turns = Vec::new(); + let mut current_user = Vec::new(); + let mut current_assistant = Vec::new(); + + for item in items { + match item { + ResponseItem::Message { role, content, .. } if role == "user" => { + if is_contextual_user_message_content(content) { + continue; + } + let Some(text) = content_items_to_text(content) + .map(|text| text.trim().to_string()) + .filter(|text| !text.is_empty()) + else { + continue; + }; + if !current_user.is_empty() || !current_assistant.is_empty() { + turns.push(( + std::mem::take(&mut current_user), + std::mem::take(&mut current_assistant), + )); + } + current_user.push(text); + } + ResponseItem::Message { role, content, .. } if role == "assistant" => { + let Some(text) = content_items_to_text(content) + .map(|text| text.trim().to_string()) + .filter(|text| !text.is_empty()) + else { + continue; + }; + if current_user.is_empty() && current_assistant.is_empty() { + continue; + } + current_assistant.push(text); + } + _ => {} + } + } + + if !current_user.is_empty() || !current_assistant.is_empty() { + turns.push((current_user, current_assistant)); + } + + let retained_turns = turns + .into_iter() + .rev() + .take(MAX_CURRENT_THREAD_TURNS) + .collect::>() + .into_iter() + .rev() + .collect::>(); + if retained_turns.is_empty() { + return None; + } + + let mut lines = vec![ + "Most recent user/assistant turns from this exact thread. Use them for continuity when responding.".to_string(), + ]; + + let retained_turn_count = retained_turns.len(); + for (index, (user_messages, assistant_messages)) in retained_turns.into_iter().enumerate() { + lines.push(String::new()); + if retained_turn_count == 1 || index + 1 == retained_turn_count { + lines.push("### Latest turn".to_string()); + } else { + lines.push(format!("### Prior turn {}", index + 1)); + } + + if !user_messages.is_empty() { + lines.push("User:".to_string()); + lines.push(user_messages.join("\n\n")); + } + if !assistant_messages.is_empty() { + lines.push(String::new()); + lines.push("Assistant:".to_string()); + lines.push(assistant_messages.join("\n\n")); + } + } + + Some(lines.join("\n")) +} + fn build_workspace_section(cwd: &Path) -> Option { build_workspace_section_with_user_root(cwd, home_dir()) } diff --git a/codex-rs/core/src/realtime_context_tests.rs b/codex-rs/core/src/realtime_context_tests.rs index b23c2743cf..5079890ec2 100644 --- a/codex-rs/core/src/realtime_context_tests.rs +++ b/codex-rs/core/src/realtime_context_tests.rs @@ -1,9 +1,13 @@ +use super::build_current_thread_section; use super::build_recent_work_section; use super::build_workspace_section; use super::build_workspace_section_with_user_root; +use crate::contextual_user_message::ENVIRONMENT_CONTEXT_FRAGMENT; use chrono::TimeZone; use chrono::Utc; use codex_protocol::ThreadId; +use codex_protocol::models::ContentItem; +use codex_protocol::models::ResponseItem; use codex_state::ThreadMetadata; use pretty_assertions::assert_eq; use std::fs; @@ -131,3 +135,60 @@ fn recent_work_section_groups_threads_by_cwd() { assert!(section.contains(&format!("### Directory: {}", outside.display()))); assert!(section.contains(&format!("- {}: Inspect flaky test", outside.display()))); } + +fn text_message(role: &str, text: &str, output: bool) -> ResponseItem { + ResponseItem::Message { + id: None, + role: role.to_string(), + content: vec![if output { + ContentItem::OutputText { + text: text.to_string(), + } + } else { + ContentItem::InputText { + text: text.to_string(), + } + }], + end_turn: None, + phase: None, + } +} + +#[test] +fn current_thread_section_keeps_latest_two_turns() { + let section = build_current_thread_section(&[ + text_message("user", "first question", false), + text_message("assistant", "first answer", true), + text_message("user", "second question", false), + text_message("assistant", "second answer", true), + text_message("user", "third question", false), + text_message("assistant", "third answer", true), + ]) + .expect("current thread section"); + + assert!(!section.contains("first question")); + assert!(section.contains("### Prior turn 1")); + assert!(section.contains("second question")); + assert!(section.contains("second answer")); + assert!(section.contains("### Latest turn")); + assert!(section.contains("third question")); + assert!(section.contains("third answer")); +} + +#[test] +fn current_thread_section_skips_contextual_user_messages() { + let section = build_current_thread_section(&[ + ENVIRONMENT_CONTEXT_FRAGMENT.into_message( + ENVIRONMENT_CONTEXT_FRAGMENT.wrap("stale environment context".to_string()), + ), + text_message("assistant", "ignore contextual pairing", true), + text_message("user", "actual question", false), + text_message("assistant", "actual answer", true), + ]) + .expect("current thread section"); + + assert!(!section.contains("stale environment context")); + assert!(!section.contains("ignore contextual pairing")); + assert!(section.contains("actual question")); + assert!(section.contains("actual answer")); +} diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index 243f4d8f22..54f1b24480 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -11,6 +11,8 @@ use crate::realtime_context::build_realtime_startup_context; use async_channel::Receiver; use async_channel::Sender; use async_channel::TrySendError; +use base64::Engine; +use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; use codex_api::Provider as ApiProvider; use codex_api::RealtimeAudioFrame; use codex_api::RealtimeEvent; @@ -22,6 +24,7 @@ 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::ConversationAudioTruncateParams; use codex_protocol::protocol::ConversationStartParams; use codex_protocol::protocol::ConversationTextParams; use codex_protocol::protocol::ErrorEvent; @@ -34,6 +37,7 @@ use codex_protocol::protocol::RealtimeHandoffRequested; use http::HeaderMap; use http::HeaderValue; use http::header::AUTHORIZATION; +use serde_json::Value; use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; @@ -49,6 +53,8 @@ const USER_TEXT_IN_QUEUE_CAPACITY: usize = 64; const HANDOFF_OUT_QUEUE_CAPACITY: usize = 64; const OUTPUT_EVENTS_QUEUE_CAPACITY: usize = 256; const REALTIME_STARTUP_CONTEXT_TOKEN_BUDGET: usize = 5_000; +const ACTIVE_RESPONSE_CONFLICT_ERROR_PREFIX: &str = + "Conversation already has an active response in progress:"; pub(crate) struct RealtimeConversationManager { state: Mutex>, @@ -58,19 +64,46 @@ pub(crate) struct RealtimeConversationManager { struct RealtimeHandoffState { output_tx: Sender, active_handoff: Arc>>, + last_output_text: Arc>>, + use_final_tool_output: bool, } #[derive(Debug, PartialEq, Eq)] -struct HandoffOutput { - handoff_id: String, - output_text: String, +enum HandoffOutput { + ImmediateAppend { + handoff_id: String, + output_text: String, + }, + FinalToolCall { + handoff_id: String, + output_text: String, + }, +} + +#[derive(Debug, PartialEq, Eq)] +struct OutputAudioState { + item_id: String, + audio_end_ms: u32, +} + +struct RealtimeInputTask { + writer: RealtimeWebsocketWriter, + events: RealtimeWebsocketEvents, + user_text_rx: Receiver, + handoff_output_rx: Receiver, + audio_rx: Receiver, + events_tx: Sender, + handoff_state: RealtimeHandoffState, + use_response_create_flow: bool, } impl RealtimeHandoffState { - fn new(output_tx: Sender) -> Self { + fn new(output_tx: Sender, use_final_tool_output: bool) -> Self { Self { output_tx, active_handoff: Arc::new(Mutex::new(None)), + last_output_text: Arc::new(Mutex::new(None)), + use_final_tool_output, } } @@ -79,8 +112,33 @@ impl RealtimeHandoffState { return Ok(()); }; + *self.last_output_text.lock().await = Some(output_text.clone()); + if !self.use_final_tool_output { + self.output_tx + .send(HandoffOutput::ImmediateAppend { + 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.use_final_tool_output { + return Ok(()); + } + + let Some(handoff_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 { + .send(HandoffOutput::FinalToolCall { handoff_id, output_text, }) @@ -94,6 +152,7 @@ impl RealtimeHandoffState { struct ConversationState { audio_tx: Sender, user_text_tx: Sender, + writer: RealtimeWebsocketWriter, handoff: RealtimeHandoffState, task: JoinHandle<()>, realtime_active: Arc, @@ -129,6 +188,8 @@ impl RealtimeConversationManager { state.task.abort(); let _ = state.task.await; } + let use_response_create_flow = + session_config.event_parser == RealtimeEventParser::RealtimeV2; let client = RealtimeWebsocketClient::new(api_provider); let connection = client @@ -152,21 +213,23 @@ impl RealtimeConversationManager { async_channel::bounded::(OUTPUT_EVENTS_QUEUE_CAPACITY); let realtime_active = Arc::new(AtomicBool::new(true)); - let handoff = RealtimeHandoffState::new(handoff_output_tx); - let task = spawn_realtime_input_task( - writer, + let handoff = RealtimeHandoffState::new(handoff_output_tx, use_response_create_flow); + let task = spawn_realtime_input_task(RealtimeInputTask { + writer: writer.clone(), events, user_text_rx, handoff_output_rx, audio_rx, events_tx, - handoff.clone(), - ); + handoff_state: handoff.clone(), + use_response_create_flow, + }); let mut guard = self.state.lock().await; *guard = Some(ConversationState { audio_tx, user_text_tx, + writer, handoff, task, realtime_active: Arc::clone(&realtime_active), @@ -198,6 +261,32 @@ impl RealtimeConversationManager { } } + pub(crate) async fn audio_truncate( + &self, + params: ConversationAudioTruncateParams, + ) -> CodexResult<()> { + let writer = { + let guard = self.state.lock().await; + guard.as_ref().map(|state| state.writer.clone()) + }; + + let Some(writer) = writer else { + return Err(CodexErr::InvalidRequest( + "conversation is not running".to_string(), + )); + }; + + writer + .send_conversation_item_truncate( + params.item_id, + params.content_index, + params.audio_end_ms, + ) + .await + .map_err(map_api_error)?; + Ok(()) + } + pub(crate) async fn text_in(&self, text: String) -> CodexResult<()> { let sender = { let guard = self.state.lock().await; @@ -231,6 +320,17 @@ impl RealtimeConversationManager { handoff.send_output(output_text).await } + pub(crate) async fn handoff_complete(&self) -> CodexResult<()> { + let handoff = { + let guard = self.state.lock().await; + guard.as_ref().map(|state| state.handoff.clone()) + }; + let Some(handoff) = handoff else { + return Ok(()); + }; + handoff.send_final_output().await + } + pub(crate) async fn active_handoff_id(&self) -> Option { let handoff = { let guard = self.state.lock().await; @@ -246,6 +346,7 @@ impl RealtimeConversationManager { }; if let Some(handoff) = handoff { *handoff.active_handoff.lock().await = None; + *handoff.last_output_text.lock().await = None; } } @@ -398,6 +499,17 @@ pub(crate) async fn handle_audio( } } +pub(crate) async fn handle_audio_truncate( + sess: &Arc, + sub_id: String, + params: ConversationAudioTruncateParams, +) { + if let Err(err) = sess.conversation.audio_truncate(params).await { + error!("failed to truncate realtime audio: {err}"); + send_conversation_error(sess, sub_id, err.to_string(), CodexErrorInfo::BadRequest).await; + } +} + fn realtime_text_from_handoff_request(handoff: &RealtimeHandoffRequested) -> Option { let active_transcript = handoff .active_transcript @@ -491,16 +603,23 @@ pub(crate) async fn handle_close(sess: &Arc, sub_id: String) { } } -fn spawn_realtime_input_task( - writer: RealtimeWebsocketWriter, - events: RealtimeWebsocketEvents, - user_text_rx: Receiver, - handoff_output_rx: Receiver, - audio_rx: Receiver, - events_tx: Sender, - handoff_state: RealtimeHandoffState, -) -> JoinHandle<()> { +fn spawn_realtime_input_task(input: RealtimeInputTask) -> JoinHandle<()> { + let RealtimeInputTask { + writer, + events, + user_text_rx, + handoff_output_rx, + audio_rx, + events_tx, + handoff_state, + use_response_create_flow, + } = input; + tokio::spawn(async move { + let mut pending_response_create = false; + let mut response_in_progress = false; + let mut output_audio_state: Option = None; + loop { tokio::select! { text = user_text_rx.recv() => { @@ -511,23 +630,66 @@ fn spawn_realtime_input_task( warn!("failed to send input text: {mapped_error}"); break; } + if use_response_create_flow { + if response_in_progress { + pending_response_create = true; + } else if let Err(err) = writer.send_response_create().await { + let mapped_error = map_api_error(err); + warn!("failed to send text response.create: {mapped_error}"); + break; + } else { + pending_response_create = false; + response_in_progress = true; + } + } } 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; + Ok(handoff_output) => { + match handoff_output { + HandoffOutput::ImmediateAppend { + 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 { + 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; + } + if use_response_create_flow { + if response_in_progress { + pending_response_create = true; + } else 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; + } else { + pending_response_create = false; + response_in_progress = true; + } + } + } } } Err(_) => break, @@ -536,12 +698,98 @@ fn spawn_realtime_input_task( 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 mut should_stop = false; + let mut forward_event = true; + + match &event { + RealtimeEvent::ConversationItemAdded(item) => { + match item.get("type").and_then(Value::as_str) { + Some("response.created") if use_response_create_flow => { + response_in_progress = true; + } + Some("response.done") if use_response_create_flow => { + response_in_progress = false; + output_audio_state = None; + if pending_response_create { + if let Err(err) = writer.send_response_create().await { + let mapped_error = map_api_error(err); + warn!( + "failed to send deferred response.create: {mapped_error}" + ); + break; + } + pending_response_create = false; + response_in_progress = true; + } + } + _ => {} + } + } + RealtimeEvent::AudioOut(frame) => { + if use_response_create_flow { + update_output_audio_state(&mut output_audio_state, frame); + } + } + RealtimeEvent::InputAudioSpeechStarted(event) => { + if use_response_create_flow + && let Some(truncate) = output_audio_truncate_params( + &mut output_audio_state, + event.item_id.as_deref(), + ) + && let Err(err) = writer + .send_conversation_item_truncate( + truncate.item_id, + truncate.content_index, + truncate.audio_end_ms, + ) + .await + { + let mapped_error = map_api_error(err); + warn!("failed to truncate realtime audio: {mapped_error}"); + } + } + RealtimeEvent::ResponseCancelled(_) => { + response_in_progress = false; + output_audio_state = None; + if use_response_create_flow && pending_response_create { + if let Err(err) = writer.send_response_create().await { + let mapped_error = map_api_error(err); + warn!( + "failed to send deferred response.create after cancellation: {mapped_error}" + ); + break; + } + pending_response_create = false; + response_in_progress = true; + } + } + RealtimeEvent::HandoffRequested(handoff) => { + *handoff_state.active_handoff.lock().await = + Some(handoff.handoff_id.clone()); + *handoff_state.last_output_text.lock().await = None; + response_in_progress = false; + output_audio_state = None; + } + RealtimeEvent::Error(message) + if use_response_create_flow + && message.starts_with(ACTIVE_RESPONSE_CONFLICT_ERROR_PREFIX) => + { + warn!( + "realtime rejected response.create because a response is already in progress; deferring follow-up response.create" + ); + pending_response_create = true; + response_in_progress = true; + forward_event = false; + } + RealtimeEvent::Error(_) => { + should_stop = true; + } + RealtimeEvent::SessionUpdated { .. } + | RealtimeEvent::InputTranscriptDelta(_) + | RealtimeEvent::OutputTranscriptDelta(_) + | RealtimeEvent::ConversationItemDone { .. } => {} } - let should_stop = matches!(&event, RealtimeEvent::Error(_)); - if events_tx.send(event).await.is_err() { + if forward_event && events_tx.send(event).await.is_err() { break; } if should_stop { @@ -588,6 +836,67 @@ fn spawn_realtime_input_task( }) } +fn update_output_audio_state( + output_audio_state: &mut Option, + frame: &RealtimeAudioFrame, +) { + let Some(item_id) = frame.item_id.clone() else { + return; + }; + let audio_end_ms = audio_duration_ms(frame); + if audio_end_ms == 0 { + return; + } + + if let Some(current) = output_audio_state.as_mut() + && current.item_id == item_id + { + current.audio_end_ms = current.audio_end_ms.saturating_add(audio_end_ms); + return; + } + + *output_audio_state = Some(OutputAudioState { + item_id, + audio_end_ms, + }); +} + +fn audio_duration_ms(frame: &RealtimeAudioFrame) -> u32 { + let Some(samples_per_channel) = frame + .samples_per_channel + .or_else(|| decoded_samples_per_channel(frame)) + else { + return 0; + }; + let sample_rate = u64::from(frame.sample_rate.max(1)); + ((u64::from(samples_per_channel) * 1_000) / sample_rate) as u32 +} + +fn decoded_samples_per_channel(frame: &RealtimeAudioFrame) -> Option { + let bytes = BASE64_STANDARD.decode(&frame.data).ok()?; + let channels = usize::from(frame.num_channels.max(1)); + let samples = bytes.len().checked_div(2)?.checked_div(channels)?; + u32::try_from(samples).ok() +} + +fn output_audio_truncate_params( + output_audio_state: &mut Option, + item_id: Option<&str>, +) -> Option { + let state = output_audio_state.take()?; + if let Some(item_id) = item_id + && item_id != state.item_id + { + return None; + } + + Some(ConversationAudioTruncateParams { + item_id: state.item_id, + content_index: 0, + audio_end_ms: state.audio_end_ms, + }) +} + async fn send_conversation_error( sess: &Arc, sub_id: String, diff --git a/codex-rs/core/src/realtime_conversation_tests.rs b/codex-rs/core/src/realtime_conversation_tests.rs index d6b85a92da..54f457004f 100644 --- a/codex-rs/core/src/realtime_conversation_tests.rs +++ b/codex-rs/core/src/realtime_conversation_tests.rs @@ -57,7 +57,7 @@ fn ignores_empty_handoff_request_input_transcript() { #[tokio::test] async fn clears_active_handoff_explicitly() { let (tx, _rx) = bounded(1); - let state = RealtimeHandoffState::new(tx); + let state = RealtimeHandoffState::new(tx, false); *state.active_handoff.lock().await = Some("handoff_1".to_string()); assert_eq!( @@ -72,7 +72,7 @@ async fn clears_active_handoff_explicitly() { #[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, false); state .send_output("ignored".to_string()) @@ -90,7 +90,7 @@ async fn sends_multiple_handoff_outputs_until_cleared() { let output_1 = rx.recv().await.expect("recv"); assert_eq!( output_1, - HandoffOutput { + HandoffOutput::ImmediateAppend { handoff_id: "handoff_1".to_string(), output_text: "result".to_string(), } @@ -99,7 +99,7 @@ async fn sends_multiple_handoff_outputs_until_cleared() { let output_2 = rx.recv().await.expect("recv"); assert_eq!( output_2, - HandoffOutput { + HandoffOutput::ImmediateAppend { handoff_id: "handoff_1".to_string(), output_text: "result 2".to_string(), } @@ -112,3 +112,23 @@ async fn sends_multiple_handoff_outputs_until_cleared() { .expect("send"); assert!(rx.is_empty()); } + +#[tokio::test] +async fn final_tool_output_waits_for_completion_in_realtime_v2() { + let (tx, rx) = bounded(2); + let state = RealtimeHandoffState::new(tx, true); + + *state.active_handoff.lock().await = Some("handoff_1".to_string()); + state.send_output("result".to_string()).await.expect("send"); + assert!(rx.is_empty()); + + state.send_final_output().await.expect("final send"); + let output = rx.recv().await.expect("recv"); + assert_eq!( + output, + HandoffOutput::FinalToolCall { + handoff_id: "handoff_1".to_string(), + output_text: "result".to_string(), + } + ); +} diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index 0d49f8c8d5..4ab9871214 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -176,6 +176,7 @@ async fn conversation_start_audio_text_close_round_trip() -> Result<()> { sample_rate: 24000, num_channels: 1, samples_per_channel: Some(480), + item_id: None, }, })) .await?; @@ -409,6 +410,7 @@ async fn conversation_audio_before_start_emits_error() -> Result<()> { sample_rate: 24000, num_channels: 1, samples_per_channel: Some(480), + item_id: None, }, })) .await?; @@ -518,6 +520,7 @@ async fn conversation_second_start_replaces_runtime() -> Result<()> { sample_rate: 24000, num_channels: 1, samples_per_channel: Some(480), + item_id: None, }, })) .await?; @@ -1469,6 +1472,7 @@ async fn inbound_handoff_request_clears_active_transcript_after_each_handoff() - sample_rate: 24000, num_channels: 1, samples_per_channel: Some(480), + item_id: None, }, })) .await?; @@ -1954,6 +1958,7 @@ async fn inbound_handoff_request_steers_active_turn() -> Result<()> { sample_rate: 24000, num_channels: 1, samples_per_channel: Some(480), + item_id: None, }, })) .await?; diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 8b5d490a2b..83ba73730f 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -133,6 +133,8 @@ pub struct RealtimeAudioFrame { pub num_channels: u16, #[serde(skip_serializing_if = "Option::is_none")] pub samples_per_channel: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub item_id: Option, } #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] @@ -154,15 +156,27 @@ pub struct RealtimeHandoffRequested { pub active_transcript: Vec, } +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] +pub struct RealtimeInputAudioSpeechStarted { + pub item_id: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] +pub struct RealtimeResponseCancelled { + pub response_id: Option, +} + #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] pub enum RealtimeEvent { SessionUpdated { session_id: String, instructions: Option, }, + InputAudioSpeechStarted(RealtimeInputAudioSpeechStarted), InputTranscriptDelta(RealtimeTranscriptDelta), OutputTranscriptDelta(RealtimeTranscriptDelta), AudioOut(RealtimeAudioFrame), + ResponseCancelled(RealtimeResponseCancelled), ConversationItemAdded(Value), ConversationItemDone { item_id: String, @@ -176,6 +190,13 @@ pub struct ConversationAudioParams { pub frame: RealtimeAudioFrame, } +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, JsonSchema, TS)] +pub struct ConversationAudioTruncateParams { + pub item_id: String, + pub content_index: u32, + pub audio_end_ms: u32, +} + #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, JsonSchema, TS)] pub struct ConversationTextParams { pub text: String, @@ -200,6 +221,9 @@ pub enum Op { /// Send audio input to the running realtime conversation stream. RealtimeConversationAudio(ConversationAudioParams), + /// Truncate output audio for the running realtime conversation stream. + RealtimeConversationAudioTruncate(ConversationAudioTruncateParams), + /// Send text input to the running realtime conversation stream. RealtimeConversationText(ConversationTextParams), @@ -493,6 +517,7 @@ impl Op { Self::CleanBackgroundTerminals => "clean_background_terminals", Self::RealtimeConversationStart(_) => "realtime_conversation_start", Self::RealtimeConversationAudio(_) => "realtime_conversation_audio", + Self::RealtimeConversationAudioTruncate(_) => "realtime_conversation_audio_truncate", Self::RealtimeConversationText(_) => "realtime_conversation_text", Self::RealtimeConversationClose => "realtime_conversation_close", Self::UserInput { .. } => "user_input", @@ -4058,8 +4083,14 @@ mod tests { sample_rate: 24_000, num_channels: 1, samples_per_channel: Some(480), + item_id: None, }, }); + let truncate = Op::RealtimeConversationAudioTruncate(ConversationAudioTruncateParams { + item_id: "item_1".to_string(), + content_index: 0, + audio_end_ms: 123, + }); let start = Op::RealtimeConversationStart(ConversationStartParams { prompt: "be helpful".to_string(), session_id: Some("conv_1".to_string()), @@ -4089,6 +4120,15 @@ mod tests { } }) ); + assert_eq!( + serde_json::to_value(&truncate).unwrap(), + json!({ + "type": "realtime_conversation_audio_truncate", + "item_id": "item_1", + "content_index": 0, + "audio_end_ms": 123 + }) + ); assert_eq!( serde_json::from_value::(serde_json::to_value(&text).unwrap()).unwrap(), text diff --git a/codex-rs/tui/src/chatwidget/realtime.rs b/codex-rs/tui/src/chatwidget/realtime.rs index 4e4f2f0e70..b33f46ad89 100644 --- a/codex-rs/tui/src/chatwidget/realtime.rs +++ b/codex-rs/tui/src/chatwidget/realtime.rs @@ -264,9 +264,11 @@ impl ChatWidget { RealtimeEvent::SessionUpdated { session_id, .. } => { self.realtime_conversation.session_id = Some(session_id); } + RealtimeEvent::InputAudioSpeechStarted(_) => self.interrupt_realtime_audio_playback(), RealtimeEvent::InputTranscriptDelta(_) => {} RealtimeEvent::OutputTranscriptDelta(_) => {} RealtimeEvent::AudioOut(frame) => self.enqueue_realtime_audio_out(&frame), + RealtimeEvent::ResponseCancelled(_) => self.interrupt_realtime_audio_playback(), RealtimeEvent::ConversationItemAdded(_item) => {} RealtimeEvent::ConversationItemDone { .. } => {} RealtimeEvent::HandoffRequested(_) => {} @@ -306,6 +308,16 @@ impl ChatWidget { } } + #[cfg(not(target_os = "linux"))] + fn interrupt_realtime_audio_playback(&mut self) { + if let Some(player) = &self.realtime_conversation.audio_player { + player.clear(); + } + } + + #[cfg(target_os = "linux")] + fn interrupt_realtime_audio_playback(&mut self) {} + #[cfg(not(target_os = "linux"))] fn start_realtime_local_audio(&mut self) { if self.realtime_conversation.capture_stop_flag.is_some() { diff --git a/codex-rs/tui/src/voice.rs b/codex-rs/tui/src/voice.rs index 7e4d8a85e8..07adcfd0a8 100644 --- a/codex-rs/tui/src/voice.rs +++ b/codex-rs/tui/src/voice.rs @@ -428,6 +428,7 @@ fn send_realtime_audio_chunk( sample_rate: MODEL_AUDIO_SAMPLE_RATE, num_channels: MODEL_AUDIO_CHANNELS, samples_per_channel: Some(samples_per_channel), + item_id: None, }, }, )));