diff --git a/README.md b/README.md index 1e44875f21..22df7f3346 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,7 @@ If you want Codex in your code editor (VS Code, Cursor, Windsurf), {} RealtimeEvent::Error(message) => { let notification = ThreadRealtimeErrorNotification { thread_id: conversation_id.to_string(), 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 84b0cd0983..de4d3a2d40 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -11,9 +11,9 @@ 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::SessionNoiseReduction; 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; @@ -26,6 +26,7 @@ use futures::StreamExt; use http::HeaderMap; use http::HeaderValue; use serde_json::json; +use std::collections::BTreeMap; use std::collections::HashMap; use std::sync::Arc; use std::sync::atomic::AtomicBool; @@ -249,6 +250,17 @@ impl RealtimeWebsocketConnection { 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,6 +353,20 @@ impl RealtimeWebsocketWriter { .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) -> Result<(), ApiError> { self.send_json(RealtimeOutboundMessage::SessionUpdate { session: SessionUpdateSession { @@ -353,6 +379,9 @@ impl RealtimeWebsocketWriter { kind: "audio/pcm".to_string(), rate: 24_000, }, + noise_reduction: SessionNoiseReduction { + kind: "near_field".to_string(), + }, turn_detection: SessionTurnDetection { kind: "server_vad".to_string(), interrupt_response: true, @@ -375,14 +404,43 @@ impl RealtimeWebsocketWriter { .to_string(), parameters: SessionToolParameters { kind: "object".to_string(), - properties: SessionToolProperties { - prompt: SessionToolProperty { - kind: "string".to_string(), - description: "The user request to delegate to Codex.".to_string(), - }, - }, + properties: BTreeMap::from([ + ( + "prompt".to_string(), + SessionToolProperty { + kind: "string".to_string(), + description: "The user request to delegate to Codex." + .to_string(), + }, + ), + ( + "send_immediately".to_string(), + SessionToolProperty { + kind: "boolean".to_string(), + description: "When true, send this to Codex immediately, steering any running turn. When false or omitted, queue it for the next turn.".to_string(), + }, + ), + ]), required: vec!["prompt".to_string()], }, + }, SessionTool { + kind: "function".to_string(), + name: "cancel_current_operation".to_string(), + description: "Cancel the current Codex operation, equivalent to pressing Ctrl-C without exiting Codex.".to_string(), + parameters: SessionToolParameters { + kind: "object".to_string(), + properties: BTreeMap::new(), + required: Vec::new(), + }, + }, SessionTool { + kind: "function".to_string(), + name: "turn_off_realtime_mode".to_string(), + description: "Turn off realtime voice mode, equivalent to using /realtime to stop live voice.".to_string(), + parameters: SessionToolParameters { + kind: "object".to_string(), + properties: BTreeMap::new(), + required: Vec::new(), + }, }], tool_choice: "auto".to_string(), }, @@ -487,7 +545,11 @@ impl RealtimeWebsocketEvents { handoff.active_transcript = std::mem::take(&mut active_transcript.entries); } RealtimeEvent::SessionUpdated { .. } + | RealtimeEvent::InterruptRequested(_) + | RealtimeEvent::CloseRequested(_) | RealtimeEvent::AudioOut(_) + | RealtimeEvent::InputAudioSpeechStarted(_) + | RealtimeEvent::ResponseCancelled(_) | RealtimeEvent::ConversationItemAdded(_) | RealtimeEvent::ConversationItemDone { .. } | RealtimeEvent::Error(_) => {} @@ -680,7 +742,12 @@ fn normalize_realtime_path(url: &mut Url) { #[cfg(test)] mod tests { use super::*; + use crate::endpoint::realtime_websocket::protocol::RealtimeCloseRequested; use crate::endpoint::realtime_websocket::protocol::RealtimeHandoffRequested; + use crate::endpoint::realtime_websocket::protocol::RealtimeInputAudioSpeechStarted; + use crate::endpoint::realtime_websocket::protocol::RealtimeInterruptRequested; + use crate::endpoint::realtime_websocket::protocol::RealtimeOutputAudioDelta; + use crate::endpoint::realtime_websocket::protocol::RealtimeResponseCancelled; use crate::endpoint::realtime_websocket::protocol::RealtimeTranscriptDelta; use crate::endpoint::realtime_websocket::protocol::RealtimeTranscriptEntry; use http::HeaderValue; @@ -722,11 +789,14 @@ mod tests { .to_string(); assert_eq!( parse_realtime_event(payload.as_str()), - Some(RealtimeEvent::AudioOut(RealtimeAudioFrame { - data: "AAA=".to_string(), - sample_rate: 48000, - num_channels: 1, - samples_per_channel: Some(960), + Some(RealtimeEvent::AudioOut(RealtimeOutputAudioDelta { + frame: RealtimeAudioFrame { + data: "AAA=".to_string(), + sample_rate: 48000, + num_channels: 1, + samples_per_channel: Some(960), + }, + item_id: None, })) ); } @@ -740,11 +810,37 @@ mod tests { .to_string(); assert_eq!( parse_realtime_event(payload.as_str()), - Some(RealtimeEvent::AudioOut(RealtimeAudioFrame { - data: "AAA=".to_string(), - sample_rate: 24_000, - num_channels: 1, - samples_per_channel: None, + Some(RealtimeEvent::AudioOut(RealtimeOutputAudioDelta { + frame: RealtimeAudioFrame { + data: "AAA=".to_string(), + sample_rate: 24_000, + num_channels: 1, + samples_per_channel: None, + }, + item_id: None, + })) + ); + } + + #[test] + fn parse_audio_delta_event_with_item_id() { + let payload = json!({ + "type": "response.audio.delta", + "delta": "AAA=", + "item_id": "item_audio_1" + }) + .to_string(); + + assert_eq!( + parse_realtime_event(payload.as_str()), + Some(RealtimeEvent::AudioOut(RealtimeOutputAudioDelta { + frame: RealtimeAudioFrame { + data: "AAA=".to_string(), + sample_rate: 24_000, + num_channels: 1, + samples_per_channel: None, + }, + item_id: Some("item_audio_1".to_string()), })) ); } @@ -803,11 +899,136 @@ mod tests { handoff_id: "handoff_123".to_string(), item_id: "item_123".to_string(), input_transcript: "delegate this".to_string(), + send_immediately: false, active_transcript: Vec::new(), })) ); } + #[test] + fn parse_handoff_requested_event_with_send_immediately() { + let payload = json!({ + "type": "response.done", + "response": { + "output": [ + { + "id": "item_456", + "type": "function_call", + "name": "codex", + "call_id": "handoff_456", + "arguments": "{\"prompt\":\"delegate this now\",\"send_immediately\":true}" + } + ] + } + }) + .to_string(); + + assert_eq!( + parse_realtime_event(payload.as_str()), + Some(RealtimeEvent::HandoffRequested(RealtimeHandoffRequested { + handoff_id: "handoff_456".to_string(), + item_id: "item_456".to_string(), + input_transcript: "delegate this now".to_string(), + send_immediately: true, + active_transcript: Vec::new(), + })) + ); + } + + #[test] + fn parse_interrupt_requested_event() { + let payload = json!({ + "type": "response.done", + "response": { + "output": [ + { + "id": "item_cancel", + "type": "function_call", + "name": "cancel_current_operation", + "call_id": "cancel_123", + "arguments": "{}" + } + ] + } + }) + .to_string(); + + assert_eq!( + parse_realtime_event(payload.as_str()), + Some(RealtimeEvent::InterruptRequested( + RealtimeInterruptRequested { + call_id: "cancel_123".to_string(), + } + )) + ); + } + + #[test] + fn parse_close_requested_event() { + let payload = json!({ + "type": "response.done", + "response": { + "output": [ + { + "id": "item_close", + "type": "function_call", + "name": "turn_off_realtime_mode", + "call_id": "close_123", + "arguments": "{}" + } + ] + } + }) + .to_string(); + + assert_eq!( + parse_realtime_event(payload.as_str()), + Some(RealtimeEvent::CloseRequested(RealtimeCloseRequested { + call_id: "close_123".to_string(), + })) + ); + } + + #[test] + fn parse_input_audio_speech_started_event() { + let payload = json!({ + "type": "input_audio_buffer.speech_started", + "item_id": "item_user_1" + }) + .to_string(); + + assert_eq!( + parse_realtime_event(payload.as_str()), + Some(RealtimeEvent::InputAudioSpeechStarted( + RealtimeInputAudioSpeechStarted { + item_id: Some("item_user_1".to_string()), + } + )) + ); + } + + #[test] + fn parse_cancelled_response_done_event() { + let payload = json!({ + "type": "response.done", + "response": { + "id": "resp_cancelled", + "status": "cancelled", + "output": [] + } + }) + .to_string(); + + assert_eq!( + parse_realtime_event(payload.as_str()), + Some(RealtimeEvent::ResponseCancelled( + RealtimeResponseCancelled { + response_id: Some("resp_cancelled".to_string()), + } + )) + ); + } + #[test] fn parse_unknown_event_as_conversation_item_added() { let payload = json!({ @@ -986,6 +1207,10 @@ mod tests { first_json["session"]["audio"]["input"]["format"]["rate"], Value::from(24_000) ); + assert_eq!( + first_json["session"]["audio"]["input"]["noise_reduction"]["type"], + Value::String("near_field".to_string()) + ); assert_eq!( first_json["session"]["audio"]["input"]["turn_detection"]["type"], Value::String("server_vad".to_string()) @@ -1026,6 +1251,50 @@ mod tests { first_json["session"]["tools"][0]["parameters"]["required"][0], Value::String("prompt".to_string()) ); + assert_eq!( + first_json["session"]["tools"][0]["parameters"]["properties"]["send_immediately"]["type"], + Value::String("boolean".to_string()) + ); + assert_eq!( + first_json["session"]["tools"][1]["type"], + Value::String("function".to_string()) + ); + assert_eq!( + first_json["session"]["tools"][1]["name"], + Value::String("cancel_current_operation".to_string()) + ); + assert_eq!( + first_json["session"]["tools"][1]["parameters"]["type"], + Value::String("object".to_string()) + ); + assert_eq!( + first_json["session"]["tools"][1]["parameters"]["required"], + Value::Array(Vec::new()) + ); + assert_eq!( + first_json["session"]["tools"][1]["parameters"]["properties"], + json!({}) + ); + assert_eq!( + first_json["session"]["tools"][2]["type"], + Value::String("function".to_string()) + ); + assert_eq!( + first_json["session"]["tools"][2]["name"], + Value::String("turn_off_realtime_mode".to_string()) + ); + assert_eq!( + first_json["session"]["tools"][2]["parameters"]["type"], + Value::String("object".to_string()) + ); + assert_eq!( + first_json["session"]["tools"][2]["parameters"]["required"], + Value::Array(Vec::new()) + ); + assert_eq!( + first_json["session"]["tools"][2]["parameters"]["properties"], + json!({}) + ); ws.send(Message::Text( json!({ @@ -1246,11 +1515,14 @@ mod tests { .expect("event"); assert_eq!( audio_event, - RealtimeEvent::AudioOut(RealtimeAudioFrame { - data: "AQID".to_string(), - sample_rate: 48000, - num_channels: 1, - samples_per_channel: None, + RealtimeEvent::AudioOut(RealtimeOutputAudioDelta { + frame: RealtimeAudioFrame { + data: "AQID".to_string(), + sample_rate: 48000, + num_channels: 1, + samples_per_channel: None, + }, + item_id: None, }) ); @@ -1301,6 +1573,7 @@ mod tests { handoff_id: "handoff_1".to_string(), item_id: "item_2".to_string(), input_transcript: "delegate now".to_string(), + send_immediately: false, active_transcript: vec![ RealtimeTranscriptEntry { role: "user".to_string(), 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 daac248b9a..949cedd04c 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs @@ -1,11 +1,17 @@ pub use codex_protocol::protocol::RealtimeAudioFrame; +pub use codex_protocol::protocol::RealtimeCloseRequested; pub use codex_protocol::protocol::RealtimeEvent; pub use codex_protocol::protocol::RealtimeHandoffRequested; +pub use codex_protocol::protocol::RealtimeInputAudioSpeechStarted; +pub use codex_protocol::protocol::RealtimeInterruptRequested; +pub use codex_protocol::protocol::RealtimeOutputAudioDelta; +pub use codex_protocol::protocol::RealtimeResponseCancelled; pub use codex_protocol::protocol::RealtimeTranscriptDelta; pub use codex_protocol::protocol::RealtimeTranscriptEntry; use serde::Deserialize; use serde::Serialize; use serde_json::Value; +use std::collections::BTreeMap; use std::string::ToString; use tracing::debug; @@ -23,6 +29,12 @@ pub(super) enum RealtimeOutboundMessage { InputAudioBufferAppend { audio: 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")] @@ -49,6 +61,7 @@ pub(super) struct SessionAudio { #[derive(Debug, Clone, Serialize)] pub(super) struct SessionAudioInput { pub(super) format: SessionAudioFormat, + pub(super) noise_reduction: SessionNoiseReduction, pub(super) turn_detection: SessionTurnDetection, } @@ -59,6 +72,12 @@ pub(super) struct SessionAudioFormat { pub(super) rate: u32, } +#[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")] @@ -93,15 +112,10 @@ pub(super) struct SessionTool { pub(super) struct SessionToolParameters { #[serde(rename = "type")] pub(super) kind: String, - pub(super) properties: SessionToolProperties, + pub(super) properties: BTreeMap, pub(super) required: Vec, } -#[derive(Debug, Clone, Serialize)] -pub(super) struct SessionToolProperties { - pub(super) prompt: SessionToolProperty, -} - #[derive(Debug, Clone, Serialize)] pub(super) struct SessionToolProperty { #[serde(rename = "type")] @@ -163,7 +177,10 @@ pub(super) fn parse_realtime_event(payload: &str) -> Option { instructions, }) } - "conversation.output_audio.delta" | "response.output_audio.delta" => { + + "conversation.output_audio.delta" + | "response.output_audio.delta" + | "response.audio.delta" => { let data = parsed .get("delta") .and_then(Value::as_str) @@ -180,16 +197,30 @@ pub(super) fn parse_realtime_event(payload: &str) -> Option { .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()), + Some(RealtimeEvent::AudioOut(RealtimeOutputAudioDelta { + frame: 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()), + }, + item_id: parsed + .get("item_id") + .and_then(Value::as_str) + .map(str::to_string), })) } + "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.input_transcript.delta" => parsed .get("delta") .and_then(Value::as_str) @@ -228,6 +259,7 @@ pub(super) fn parse_realtime_event(payload: &str) -> Option { handoff_id, item_id, input_transcript, + send_immediately: false, active_transcript: Vec::new(), })) } @@ -235,8 +267,33 @@ pub(super) fn parse_realtime_event(payload: &str) -> Option { if let Some(handoff) = parse_handoff_requested(&parsed) { return Some(RealtimeEvent::HandoffRequested(handoff)); } + if let Some(interrupt) = parse_interrupt_requested(&parsed) { + return Some(RealtimeEvent::InterruptRequested(interrupt)); + } + if let Some(close) = parse_close_requested(&parsed) { + return Some(RealtimeEvent::CloseRequested(close)); + } + if let Some(cancelled) = parse_response_cancelled(&parsed) { + return Some(RealtimeEvent::ResponseCancelled(cancelled)); + } Some(RealtimeEvent::ConversationItemAdded(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" => parsed .get("message") .and_then(Value::as_str) @@ -256,15 +313,7 @@ pub(super) fn parse_realtime_event(payload: &str) -> Option { } fn parse_handoff_requested(parsed: &Value) -> Option { - let outputs = parsed - .get("response") - .and_then(Value::as_object) - .and_then(|response| response.get("output")) - .and_then(Value::as_array)?; - let function_call = outputs.iter().find(|item| { - item.get("type").and_then(Value::as_str) == Some("function_call") - && item.get("name").and_then(Value::as_str) == Some("codex") - })?; + let function_call = find_function_call(parsed, "codex")?; let handoff_id = function_call .get("call_id") .and_then(Value::as_str) @@ -278,15 +327,70 @@ fn parse_handoff_requested(parsed: &Value) -> Option { .get("arguments") .and_then(Value::as_str) .unwrap_or_default(); + let parsed_arguments = parse_handoff_arguments(arguments); Some(RealtimeHandoffRequested { handoff_id, item_id, - input_transcript: parse_handoff_arguments(arguments), + input_transcript: parsed_arguments.input_transcript, + send_immediately: parsed_arguments.send_immediately, active_transcript: Vec::new(), }) } -fn parse_handoff_arguments(arguments: &str) -> String { +fn parse_interrupt_requested(parsed: &Value) -> Option { + let function_call = find_function_call(parsed, "cancel_current_operation")?; + Some(RealtimeInterruptRequested { + call_id: function_call + .get("call_id") + .and_then(Value::as_str) + .map(str::to_string)?, + }) +} + +fn parse_close_requested(parsed: &Value) -> Option { + let function_call = find_function_call(parsed, "turn_off_realtime_mode")?; + Some(RealtimeCloseRequested { + call_id: function_call + .get("call_id") + .and_then(Value::as_str) + .map(str::to_string)?, + }) +} + +fn parse_response_cancelled(parsed: &Value) -> Option { + let response = parsed.get("response")?.as_object()?; + let status = response.get("status").and_then(Value::as_str)?; + if status != "cancelled" { + return None; + } + + Some(RealtimeResponseCancelled { + response_id: response + .get("id") + .and_then(Value::as_str) + .map(str::to_string), + }) +} + +fn find_function_call<'a>(parsed: &'a Value, name: &str) -> Option<&'a Value> { + 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(name) + }) +} + +struct ParsedHandoffArguments { + input_transcript: String, + send_immediately: bool, +} + +fn parse_handoff_arguments(arguments: &str) -> ParsedHandoffArguments { #[derive(Debug, Deserialize)] struct HandoffArguments { #[serde(default)] @@ -300,11 +404,16 @@ fn parse_handoff_arguments(arguments: &str) -> String { #[serde(default)] input_transcript: Option, #[serde(default)] + send_immediately: bool, + #[serde(default)] messages: Vec, } let Some(parsed) = serde_json::from_str::(arguments).ok() else { - return arguments.to_string(); + return ParsedHandoffArguments { + input_transcript: arguments.to_string(), + send_immediately: false, + }; }; for value in [ @@ -318,7 +427,10 @@ fn parse_handoff_arguments(arguments: &str) -> String { .flatten() { if !value.is_empty() { - return value; + return ParsedHandoffArguments { + input_transcript: value, + send_immediately: parsed.send_immediately, + }; } } @@ -327,8 +439,14 @@ fn parse_handoff_arguments(arguments: &str) -> String { .into_iter() .find(|message| message.role == "user" && !message.text.is_empty()) { - return message.text; + return ParsedHandoffArguments { + input_transcript: message.text, + send_immediately: parsed.send_immediately, + }; } - String::new() + ParsedHandoffArguments { + input_transcript: String::new(), + send_immediately: parsed.send_immediately, + } } diff --git a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs index e79f0aac07..85f647e920 100644 --- a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs +++ b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs @@ -8,6 +8,7 @@ use codex_api::RealtimeSessionConfig; use codex_api::RealtimeWebsocketClient; use codex_api::provider::Provider; use codex_api::provider::RetryConfig; +use codex_protocol::protocol::RealtimeOutputAudioDelta; use futures::SinkExt; use futures::StreamExt; use http::HeaderMap; @@ -95,6 +96,10 @@ async fn realtime_ws_e2e_session_create_and_event_flow() { first_json["session"]["audio"]["input"]["format"]["rate"], Value::from(24_000) ); + assert_eq!( + first_json["session"]["audio"]["input"]["noise_reduction"]["type"], + Value::String("near_field".to_string()) + ); assert_eq!( first_json["session"]["audio"]["input"]["turn_detection"]["type"], Value::String("server_vad".to_string()) @@ -131,6 +136,31 @@ async fn realtime_ws_e2e_session_create_and_event_flow() { first_json["session"]["tools"][0]["name"], Value::String("codex".to_string()) ); + assert_eq!( + first_json["session"]["tools"][0]["parameters"]["properties"]["send_immediately"] + ["type"], + Value::String("boolean".to_string()) + ); + assert_eq!( + first_json["session"]["tools"][1]["type"], + Value::String("function".to_string()) + ); + assert_eq!( + first_json["session"]["tools"][1]["name"], + Value::String("cancel_current_operation".to_string()) + ); + assert_eq!( + first_json["session"]["tools"][1]["parameters"]["type"], + Value::String("object".to_string()) + ); + assert_eq!( + first_json["session"]["tools"][1]["parameters"]["required"], + Value::Array(Vec::new()) + ); + assert_eq!( + first_json["session"]["tools"][1]["parameters"]["properties"], + json!({}) + ); ws.send(Message::Text( json!({ @@ -212,11 +242,14 @@ async fn realtime_ws_e2e_session_create_and_event_flow() { .expect("event"); assert_eq!( audio_event, - RealtimeEvent::AudioOut(RealtimeAudioFrame { - data: "AQID".to_string(), - sample_rate: 48000, - num_channels: 1, - samples_per_channel: None, + RealtimeEvent::AudioOut(RealtimeOutputAudioDelta { + frame: RealtimeAudioFrame { + data: "AQID".to_string(), + sample_rate: 48000, + num_channels: 1, + samples_per_channel: None, + }, + item_id: None, }) ); diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 2bc25b2997..388e046c7c 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -36,6 +36,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; @@ -2541,13 +2542,14 @@ impl Session { } async fn maybe_clear_realtime_handoff_for_event(&self, msg: &EventMsg) { - if !matches!(msg, EventMsg::TurnComplete(_)) { - return; - } - if let Err(err) = self.conversation.handoff_complete().await { + if matches!(msg, EventMsg::TurnComplete(_)) + && let Err(err) = self.conversation.handoff_complete().await + { debug!("failed to send final realtime handoff tool output: {err}"); } - self.conversation.clear_active_handoff().await; + if matches!(msg, EventMsg::TurnComplete(_) | EventMsg::TurnAborted(_)) { + self.conversation.clear_active_handoff().await; + } } pub(crate) async fn send_event_raw(&self, event: Event) { @@ -2640,6 +2642,10 @@ impl Session { .map(|task| Arc::clone(&task.turn_context)) } + pub(crate) async fn has_active_turn(&self) -> bool { + self.active_turn.lock().await.is_some() + } + async fn active_turn_context_and_cancellation_token( &self, ) -> Option<(Arc, CancellationToken)> { @@ -4053,6 +4059,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 @@ -4250,7 +4261,7 @@ async fn submission_loop(sess: Arc, config: Arc, rx_sub: Receiv fn submission_dispatch_span(sub: &Submission) -> tracing::Span { let dispatch_span = match &sub.op { - Op::RealtimeConversationAudio(_) => { + Op::RealtimeConversationAudio(_) | Op::RealtimeConversationAudioTruncate(_) => { debug_span!("submission_dispatch", submission.id = sub.id.as_str()) } _ => info_span!("submission_dispatch", submission.id = sub.id.as_str()), diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index fff7a495ab..22b0965bd7 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -18,6 +18,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; @@ -46,6 +47,8 @@ const HANDOFF_OUT_QUEUE_CAPACITY: usize = 64; const OUTPUT_EVENTS_QUEUE_CAPACITY: usize = 256; const REALTIME_STARTUP_CONTEXT_TOKEN_BUDGET: usize = 5_000; const DEFAULT_REALTIME_MODEL: &str = "gpt-realtime-1.5"; +const INTERRUPT_TOOL_OUTPUT_CANCELLED: &str = "Cancelled the current Codex operation."; +const INTERRUPT_TOOL_OUTPUT_IDLE: &str = "No active Codex operation was running."; pub(crate) struct RealtimeConversationManager { state: Mutex>, @@ -111,12 +114,24 @@ impl RealtimeHandoffState { .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?; Ok(()) } + + async fn send_tool_call_output(&self, call_id: String, output_text: String) -> CodexResult<()> { + self.output_tx + .send(HandoffOutput::FinalToolCall { + call_id, + output_text, + }) + .await + .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?; + Ok(()) + } } #[allow(dead_code)] struct ConversationState { audio_tx: Sender, user_text_tx: Sender, + writer: RealtimeWebsocketWriter, handoff: RealtimeHandoffState, task: JoinHandle<()>, realtime_active: Arc, @@ -184,7 +199,7 @@ impl RealtimeConversationManager { let realtime_active = Arc::new(AtomicBool::new(true)); let handoff = RealtimeHandoffState::new(handoff_output_tx); let task = spawn_realtime_input_task( - writer, + writer.clone(), events, user_text_rx, handoff_output_rx, @@ -197,6 +212,7 @@ impl RealtimeConversationManager { *guard = Some(ConversationState { audio_tx, user_text_tx, + writer, handoff, task, realtime_active: Arc::clone(&realtime_active), @@ -247,6 +263,32 @@ impl RealtimeConversationManager { Ok(()) } + 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 handoff_out(&self, output_text: String) -> CodexResult<()> { let handoff = { let guard = self.state.lock().await; @@ -272,6 +314,21 @@ impl RealtimeConversationManager { handoff.send_final_output().await } + pub(crate) async fn tool_call_complete( + &self, + call_id: String, + output_text: String, + ) -> 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_tool_call_output(call_id, output_text).await + } + pub(crate) async fn active_handoff_id(&self) -> Option { let handoff = { let guard = self.state.lock().await; @@ -388,7 +445,7 @@ pub(crate) async fn handle_start( ); } let maybe_routed_text = match &event { - RealtimeEvent::HandoffRequested(handoff) => { + RealtimeEvent::HandoffRequested(handoff) if handoff.send_immediately => { realtime_text_from_handoff_request(handoff) } _ => None, @@ -398,6 +455,21 @@ pub(crate) async fn handle_start( let sess_for_routed_text = Arc::clone(&sess_clone); sess_for_routed_text.route_realtime_text_input(text).await; } + if let RealtimeEvent::InterruptRequested(interrupt) = &event { + let output_text = if sess_clone.has_active_turn().await { + INTERRUPT_TOOL_OUTPUT_CANCELLED.to_string() + } else { + INTERRUPT_TOOL_OUTPUT_IDLE.to_string() + }; + sess_clone.interrupt_task().await; + if let Err(err) = sess_clone + .conversation + .tool_call_complete(interrupt.call_id.clone(), output_text) + .await + { + debug!("failed to send realtime interrupt tool output: {err}"); + } + } sess_clone .send_event_raw(ev(EventMsg::RealtimeConversationRealtime( RealtimeConversationRealtimeEvent { @@ -432,6 +504,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 @@ -666,6 +749,7 @@ async fn send_conversation_error( #[cfg(test)] mod tests { use super::HandoffOutput; + use super::INTERRUPT_TOOL_OUTPUT_CANCELLED; use super::RealtimeHandoffState; use super::realtime_text_from_handoff_request; use async_channel::bounded; @@ -679,6 +763,7 @@ mod tests { handoff_id: "handoff_1".to_string(), item_id: "item_1".to_string(), input_transcript: "ignored".to_string(), + send_immediately: false, active_transcript: vec![ RealtimeTranscriptEntry { role: "user".to_string(), @@ -702,6 +787,7 @@ mod tests { handoff_id: "handoff_1".to_string(), item_id: "item_1".to_string(), input_transcript: "ignored".to_string(), + send_immediately: false, active_transcript: vec![], }; assert_eq!( @@ -716,6 +802,7 @@ mod tests { handoff_id: "handoff_1".to_string(), item_id: "item_1".to_string(), input_transcript: String::new(), + send_immediately: false, active_transcript: vec![], }; assert_eq!(realtime_text_from_handoff_request(&handoff), None); @@ -802,4 +889,27 @@ mod tests { } ); } + + #[tokio::test] + async fn sends_explicit_tool_call_output() { + let (tx, rx) = bounded(1); + let state = RealtimeHandoffState::new(tx); + + state + .send_tool_call_output( + "cancel_1".to_string(), + INTERRUPT_TOOL_OUTPUT_CANCELLED.to_string(), + ) + .await + .expect("send tool output"); + + let output = rx.recv().await.expect("recv"); + assert_eq!( + output, + HandoffOutput::FinalToolCall { + call_id: "cancel_1".to_string(), + output_text: INTERRUPT_TOOL_OUTPUT_CANCELLED.to_string(), + } + ); + } } diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index 3750a5fa25..b59fc6e275 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -6,6 +6,7 @@ use codex_core::auth::OPENAI_API_KEY_ENV_VAR; use codex_protocol::ThreadId; 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; @@ -192,7 +193,7 @@ async fn conversation_start_audio_text_close_round_trip() -> Result<()> { _ => None, }) .await; - assert_eq!(audio_out.data, "AQID"); + assert_eq!(audio_out.frame.data, "AQID"); let connections = server.connections(); assert_eq!(connections.len(), 2); @@ -256,6 +257,71 @@ async fn conversation_start_audio_text_close_round_trip() -> Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn conversation_audio_truncate_sends_truncate_event() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_websocket_server(vec![ + vec![], + vec![ + vec![json!({ + "type": "session.updated", + "session": { "id": "sess_truncate", "instructions": "backend prompt" } + })], + vec![], + ], + ]) + .await; + + let mut builder = test_codex(); + 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?; + + wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { + payload: RealtimeEvent::SessionUpdated { session_id, .. }, + }) if session_id == "sess_truncate" => Some(()), + _ => None, + }) + .await; + + test.codex + .submit(Op::RealtimeConversationAudioTruncate( + ConversationAudioTruncateParams { + item_id: "item_audio_1".to_string(), + content_index: 0, + audio_end_ms: 320, + }, + )) + .await?; + + let truncate_request = wait_for_matching_websocket_request( + &server, + "realtime conversation.item.truncate request", + |request| request.body_json()["type"].as_str() == Some("conversation.item.truncate"), + ) + .await; + assert_eq!( + truncate_request.body_json(), + json!({ + "type": "conversation.item.truncate", + "item_id": "item_audio_1", + "content_index": 0, + "audio_end_ms": 320 + }) + ); + + server.shutdown().await; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[serial(openai_api_key_env)] async fn conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth() -> Result<()> { @@ -524,7 +590,7 @@ async fn conversation_second_start_replaces_runtime() -> Result<()> { let _ = wait_for_event_match(&test.codex, |msg| match msg { EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { payload: RealtimeEvent::AudioOut(frame), - }) if frame.data == "AQID" => Some(()), + }) if frame.frame.data == "AQID" => Some(()), _ => None, }) .await; @@ -1269,7 +1335,7 @@ fn realtime_handoff_requested_event(handoff_id: &str, item_id: &str, prompt: &st "type": "function_call", "name": "codex", "call_id": handoff_id, - "arguments": json!({ "prompt": prompt }).to_string(), + "arguments": json!({ "prompt": prompt, "send_immediately": true }).to_string(), } ] } @@ -1629,7 +1695,7 @@ async fn inbound_conversation_item_does_not_start_turn_and_still_forwards_audio( ) .await .expect("timed out waiting for realtime audio after conversation item"); - assert_eq!(audio_out.data, "AQID"); + assert_eq!(audio_out.frame.data, "AQID"); let unexpected_turn_started = tokio::time::timeout( Duration::from_millis(200), @@ -1782,11 +1848,11 @@ async fn delegated_turn_user_role_echo_does_not_redelegate_and_still_forwards_au eprintln!( "[realtime test +{}ms] saw audio out data={} sample_rate={} num_channels={}", start.elapsed().as_millis(), - audio_out.data, - audio_out.sample_rate, - audio_out.num_channels + audio_out.frame.data, + audio_out.frame.sample_rate, + audio_out.frame.num_channels ); - assert_eq!(audio_out.data, "AQID"); + assert_eq!(audio_out.frame.data, "AQID"); let completion = completions .into_iter() @@ -1895,7 +1961,7 @@ async fn inbound_handoff_request_does_not_block_realtime_event_forwarding() -> R ) .await .expect("timed out waiting for realtime audio while delegated turn was still pending"); - assert_eq!(audio_out.data, "AQID"); + assert_eq!(audio_out.frame.data, "AQID"); let completion = completions .into_iter() @@ -2152,7 +2218,7 @@ async fn inbound_handoff_request_starts_turn_and_does_not_block_realtime_audio() ) .await .expect("timed out waiting for realtime audio after handoff request"); - assert_eq!(audio_out.data, "AQID"); + assert_eq!(audio_out.frame.data, "AQID"); let completion = completions .into_iter() diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 98ff8f7f55..5eeebb9d75 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -131,6 +131,13 @@ pub struct RealtimeAudioFrame { pub samples_per_channel: Option, } +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] +pub struct RealtimeOutputAudioDelta { + pub frame: RealtimeAudioFrame, + #[serde(skip_serializing_if = "Option::is_none")] + pub item_id: Option, +} + #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] pub struct RealtimeTranscriptDelta { pub delta: String, @@ -147,9 +154,33 @@ pub struct RealtimeHandoffRequested { pub handoff_id: String, pub item_id: String, pub input_transcript: String, + #[serde(default)] + pub send_immediately: bool, pub active_transcript: Vec, } +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] +pub struct RealtimeInterruptRequested { + pub call_id: String, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] +pub struct RealtimeCloseRequested { + pub call_id: String, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] +pub struct RealtimeInputAudioSpeechStarted { + #[serde(skip_serializing_if = "Option::is_none")] + pub item_id: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] +pub struct RealtimeResponseCancelled { + #[serde(skip_serializing_if = "Option::is_none")] + pub response_id: Option, +} + #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] pub enum RealtimeEvent { SessionUpdated { @@ -158,12 +189,16 @@ pub enum RealtimeEvent { }, InputTranscriptDelta(RealtimeTranscriptDelta), OutputTranscriptDelta(RealtimeTranscriptDelta), - AudioOut(RealtimeAudioFrame), + AudioOut(RealtimeOutputAudioDelta), + InputAudioSpeechStarted(RealtimeInputAudioSpeechStarted), + ResponseCancelled(RealtimeResponseCancelled), ConversationItemAdded(Value), ConversationItemDone { item_id: String, }, HandoffRequested(RealtimeHandoffRequested), + InterruptRequested(RealtimeInterruptRequested), + CloseRequested(RealtimeCloseRequested), Error(String), } @@ -172,6 +207,13 @@ pub struct ConversationAudioParams { pub frame: RealtimeAudioFrame, } +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, 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, @@ -196,6 +238,9 @@ pub enum Op { /// Send audio input to the running realtime conversation stream. RealtimeConversationAudio(ConversationAudioParams), + /// Truncate the current assistant audio item in the running realtime conversation stream. + RealtimeConversationAudioTruncate(ConversationAudioTruncateParams), + /// Send text input to the running realtime conversation stream. RealtimeConversationText(ConversationTextParams), @@ -3957,6 +4002,11 @@ mod tests { samples_per_channel: Some(480), }, }); + let truncate = Op::RealtimeConversationAudioTruncate(ConversationAudioTruncateParams { + item_id: "item_1".to_string(), + content_index: 0, + audio_end_ms: 320, + }); let start = Op::RealtimeConversationStart(ConversationStartParams { prompt: "be helpful".to_string(), session_id: Some("conv_1".to_string()), @@ -3986,6 +4036,19 @@ 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": 320 + }) + ); + assert_eq!( + serde_json::from_value::(serde_json::to_value(&truncate).unwrap()).unwrap(), + truncate + ); assert_eq!( serde_json::from_value::(serde_json::to_value(&text).unwrap()).unwrap(), text diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 7fdc294dac..775c161845 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -3816,7 +3816,7 @@ impl ChatWidget { return; } let Some(user_message) = - self.maybe_defer_user_message_for_realtime(user_message) + self.maybe_submit_user_message_for_realtime(user_message) else { return; }; @@ -3851,7 +3851,7 @@ impl ChatWidget { .take_recent_submission_mention_bindings(), }; let Some(user_message) = - self.maybe_defer_user_message_for_realtime(user_message) + self.maybe_submit_user_message_for_realtime(user_message) else { return; }; diff --git a/codex-rs/tui/src/chatwidget/realtime.rs b/codex-rs/tui/src/chatwidget/realtime.rs index c38615eceb..ea1c359855 100644 --- a/codex-rs/tui/src/chatwidget/realtime.rs +++ b/codex-rs/tui/src/chatwidget/realtime.rs @@ -1,12 +1,29 @@ use super::*; +use codex_protocol::protocol::ConversationAudioTruncateParams; use codex_protocol::protocol::ConversationStartParams; -use codex_protocol::protocol::RealtimeAudioFrame; +use codex_protocol::protocol::ConversationTextParams; use codex_protocol::protocol::RealtimeConversationClosedEvent; use codex_protocol::protocol::RealtimeConversationRealtimeEvent; use codex_protocol::protocol::RealtimeConversationStartedEvent; use codex_protocol::protocol::RealtimeEvent; +use codex_protocol::protocol::RealtimeHandoffRequested; +use codex_protocol::protocol::RealtimeOutputAudioDelta; -const REALTIME_CONVERSATION_PROMPT: &str = "You are in a realtime voice conversation in the Codex TUI. Respond conversationally and concisely."; +const REALTIME_CONVERSATION_PROMPT: &str = concat!( + "You are in a realtime voice conversation in the Codex TUI. ", + "You can call the codex function to delegate work to Codex, this should be your default action. Don't verbalize anything before calling codex. ", + "The codex function queues requests by default when Codex is already busy. Set send_immediately=true only when you need to steer the current Codex turn right away. ", + "If the user asks to stop, cancel, or abort ongoing Codex work, call cancel_current_operation before speaking. ", + "If the user asks to stop voice mode, exit realtime mode, or turn off live voice, call turn_off_realtime_mode and do not speak afterward. ", + "Codex can inspect the repository, read and edit files, run commands, and produce detailed ", + "text output that the user can read in the TUI. ", + "Use codex whenever the user asks for codebase-specific facts, debugging, file changes, ", + "command output, or anything that benefits from tools. ", + "When you speak to the user directly, be extremely concise. ", + "Prefer short spoken answers because the user can see the detailed text output themselves. ", + "Do not read long code snippets, long file paths, diffs, or command output aloud unless the ", + "user explicitly asks for that." +); #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub(super) enum RealtimeConversationPhase { @@ -22,7 +39,7 @@ pub(super) struct RealtimeConversationUiState { phase: RealtimeConversationPhase, requested_close: bool, session_id: Option, - warned_audio_only_submission: bool, + warned_unsupported_composer_submission: bool, meter_placeholder_id: Option, #[cfg(not(target_os = "linux"))] capture_stop_flag: Option>, @@ -47,6 +64,13 @@ impl RealtimeConversationUiState { } } +#[cfg(test)] +impl RealtimeConversationUiState { + pub(super) fn set_phase_for_test(&mut self, phase: RealtimeConversationPhase) { + self.phase = phase; + } +} + #[derive(Clone, Debug, PartialEq)] pub(super) struct RenderedUserMessageEvent { pub(super) message: String, @@ -170,7 +194,7 @@ impl ChatWidget { self.last_rendered_user_message_event.as_ref() != Some(&key) } - pub(super) fn maybe_defer_user_message_for_realtime( + pub(super) fn maybe_submit_user_message_for_realtime( &mut self, user_message: UserMessage, ) -> Option { @@ -178,17 +202,72 @@ impl ChatWidget { return Some(user_message); } - self.restore_user_message_to_composer(user_message); - if !self.realtime_conversation.warned_audio_only_submission { - self.realtime_conversation.warned_audio_only_submission = true; - self.add_info_message( - "Realtime voice mode is audio-only. Use /realtime to stop.".to_string(), - None, - ); - } else { - self.request_redraw(); + let UserMessage { + text, + local_images, + remote_image_urls, + text_elements, + mention_bindings, + } = user_message; + + if !local_images.is_empty() || !remote_image_urls.is_empty() { + self.restore_user_message_to_composer(UserMessage { + text, + local_images, + remote_image_urls, + text_elements, + mention_bindings, + }); + if !self + .realtime_conversation + .warned_unsupported_composer_submission + { + self.realtime_conversation + .warned_unsupported_composer_submission = true; + self.add_info_message( + "Realtime voice mode accepts typed text, but not image attachments. Remove attachments or use /realtime to stop." + .to_string(), + None, + ); + } else { + self.request_redraw(); + } + return None; } + self.realtime_conversation + .warned_unsupported_composer_submission = false; + + if !self.submit_op(Op::RealtimeConversationText(ConversationTextParams { + text: text.clone(), + })) { + return None; + } + + if !text.is_empty() { + let encoded_mentions = mention_bindings + .iter() + .map(|binding| LinkedMention { + mention: binding.mention.clone(), + path: binding.path.clone(), + }) + .collect::>(); + let history_text = encode_history_mentions(&text, &encoded_mentions); + self.codex_op_tx + .send(Op::AddToHistory { text: history_text }) + .unwrap_or_else(|e| { + tracing::error!("failed to send AddHistory op: {e}"); + }); + } + + self.on_user_message_event(UserMessageEvent { + message: text, + images: None, + local_images: Vec::new(), + text_elements, + }); + self.request_redraw(); + None } @@ -200,7 +279,8 @@ impl ChatWidget { self.realtime_conversation.phase = RealtimeConversationPhase::Starting; self.realtime_conversation.requested_close = false; self.realtime_conversation.session_id = None; - self.realtime_conversation.warned_audio_only_submission = false; + self.realtime_conversation + .warned_unsupported_composer_submission = false; self.set_footer_hint_override(Some(Self::realtime_footer_hint_items())); self.submit_op(Op::RealtimeConversationStart(ConversationStartParams { prompt: REALTIME_CONVERSATION_PROMPT.to_string(), @@ -236,7 +316,8 @@ impl ChatWidget { self.realtime_conversation.phase = RealtimeConversationPhase::Inactive; self.realtime_conversation.requested_close = false; self.realtime_conversation.session_id = None; - self.realtime_conversation.warned_audio_only_submission = false; + self.realtime_conversation + .warned_unsupported_composer_submission = false; } pub(super) fn on_realtime_conversation_started( @@ -250,7 +331,8 @@ impl ChatWidget { } self.realtime_conversation.phase = RealtimeConversationPhase::Active; self.realtime_conversation.session_id = ev.session_id; - self.realtime_conversation.warned_audio_only_submission = false; + self.realtime_conversation + .warned_unsupported_composer_submission = false; self.set_footer_hint_override(Some(Self::realtime_footer_hint_items())); self.start_realtime_local_audio(); self.request_redraw(); @@ -264,12 +346,26 @@ 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(_) => {} + RealtimeEvent::HandoffRequested(handoff) => { + self.on_realtime_handoff_requested(handoff); + } + RealtimeEvent::InterruptRequested(_) => {} + RealtimeEvent::CloseRequested(_) => { + self.request_realtime_conversation_close(Some( + "Realtime voice mode turned off.".to_string(), + )); + } RealtimeEvent::Error(message) => { self.add_error_message(format!("Realtime voice error: {message}")); self.reset_realtime_conversation_state(); @@ -287,7 +383,7 @@ impl ChatWidget { self.request_redraw(); } - fn enqueue_realtime_audio_out(&mut self, frame: &RealtimeAudioFrame) { + fn enqueue_realtime_audio_out(&mut self, frame: &RealtimeOutputAudioDelta) { #[cfg(not(target_os = "linux"))] { if self.realtime_conversation.audio_player.is_none() { @@ -306,6 +402,48 @@ impl ChatWidget { } } + #[cfg(not(target_os = "linux"))] + fn interrupt_realtime_audio_playback(&mut self) { + let Some(player) = &self.realtime_conversation.audio_player else { + return; + }; + + let Some(position) = player.interrupt() else { + return; + }; + + self.submit_op(Op::RealtimeConversationAudioTruncate( + ConversationAudioTruncateParams { + item_id: position.item_id, + content_index: 0, + audio_end_ms: position.audio_end_ms, + }, + )); + } + + #[cfg(target_os = "linux")] + fn interrupt_realtime_audio_playback(&mut self) {} + + fn on_realtime_handoff_requested(&mut self, handoff: RealtimeHandoffRequested) { + let Some(text) = realtime_text_from_handoff_request(&handoff) else { + return; + }; + + let user_message = UserMessage { + text, + local_images: Vec::new(), + remote_image_urls: Vec::new(), + text_elements: Vec::new(), + mention_bindings: Vec::new(), + }; + + if handoff.send_immediately { + self.submit_user_message(user_message); + } else { + self.queue_user_message(user_message); + } + } + #[cfg(all(not(target_os = "linux"), feature = "voice-input"))] fn start_realtime_local_audio(&mut self) { if self.realtime_conversation.capture_stop_flag.is_some() { @@ -428,3 +566,17 @@ impl ChatWidget { } } } + +fn realtime_text_from_handoff_request(handoff: &RealtimeHandoffRequested) -> Option { + let active_transcript = handoff + .active_transcript + .iter() + .map(|entry| format!("{}: {}", entry.role, entry.text)) + .collect::>() + .join("\n"); + (!active_transcript.is_empty()) + .then_some(active_transcript) + .or_else(|| { + (!handoff.input_transcript.is_empty()).then(|| handoff.input_transcript.clone()) + }) +} diff --git a/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__realtime_typed_text_submission.snap b/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__realtime_typed_text_submission.snap new file mode 100644 index 0000000000..a92fc4f974 --- /dev/null +++ b/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__realtime_typed_text_submission.snap @@ -0,0 +1,7 @@ +--- +source: tui/src/chatwidget/tests.rs +assertion_line: 10372 +expression: "lines_to_single_string(&inserted[0])" +--- + +› typed via realtime diff --git a/codex-rs/tui/src/chatwidget/tests.rs b/codex-rs/tui/src/chatwidget/tests.rs index 0896adc074..d1dcfa2c12 100644 --- a/codex-rs/tui/src/chatwidget/tests.rs +++ b/codex-rs/tui/src/chatwidget/tests.rs @@ -82,6 +82,9 @@ use codex_protocol::protocol::PatchApplyBeginEvent; use codex_protocol::protocol::PatchApplyEndEvent; use codex_protocol::protocol::PatchApplyStatus as CorePatchApplyStatus; use codex_protocol::protocol::RateLimitWindow; +use codex_protocol::protocol::RealtimeConversationRealtimeEvent; +use codex_protocol::protocol::RealtimeEvent; +use codex_protocol::protocol::RealtimeHandoffRequested; use codex_protocol::protocol::ReviewRequest; use codex_protocol::protocol::ReviewTarget; use codex_protocol::protocol::SessionSource; @@ -1915,6 +1918,19 @@ fn next_submit_op(op_rx: &mut tokio::sync::mpsc::UnboundedReceiver) -> Op { } } +fn next_realtime_text(op_rx: &mut tokio::sync::mpsc::UnboundedReceiver) -> String { + loop { + match op_rx.try_recv() { + Ok(Op::RealtimeConversationText(params)) => return params.text, + Ok(_) => continue, + Err(TryRecvError::Empty) => panic!("expected realtime text op but queue was empty"), + Err(TryRecvError::Disconnected) => { + panic!("expected realtime text op but channel closed") + } + } + } +} + fn next_interrupt_op(op_rx: &mut tokio::sync::mpsc::UnboundedReceiver) { loop { match op_rx.try_recv() { @@ -10334,6 +10350,117 @@ async fn enter_queues_user_messages_while_review_is_running() { assert!(drain_insert_history(&mut rx).is_empty()); } +#[tokio::test] +async fn enter_submits_typed_text_while_realtime_mode_is_live() { + let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await; + chat.realtime_conversation + .set_phase_for_test(super::realtime::RealtimeConversationPhase::Active); + + chat.bottom_pane + .set_composer_text("typed via realtime".to_string(), Vec::new(), Vec::new()); + chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE)); + + assert_eq!(chat.bottom_pane.composer_text(), ""); + assert!(chat.queued_user_messages.is_empty()); + assert_eq!(next_realtime_text(&mut op_rx), "typed via realtime"); + + let inserted = drain_insert_history(&mut rx); + assert_eq!(inserted.len(), 1); + assert_snapshot!( + "realtime_typed_text_submission", + lines_to_single_string(&inserted[0]) + ); +} + +#[tokio::test] +async fn queue_shortcut_submits_realtime_text_while_realtime_mode_is_live() { + let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await; + chat.realtime_conversation + .set_phase_for_test(super::realtime::RealtimeConversationPhase::Active); + chat.on_task_started(); + + chat.bottom_pane + .set_composer_text("queued via realtime".to_string(), Vec::new(), Vec::new()); + chat.handle_key_event(KeyEvent::new(KeyCode::Tab, KeyModifiers::NONE)); + + assert_eq!(chat.bottom_pane.composer_text(), ""); + assert!(chat.queued_user_messages.is_empty()); + assert_eq!(next_realtime_text(&mut op_rx), "queued via realtime"); + + let inserted = drain_insert_history(&mut rx); + let rendered = inserted + .iter() + .map(|lines| lines_to_single_string(lines)) + .find(|text| text.contains("queued via realtime")) + .expect("expected rendered realtime text prompt"); + assert!(rendered.contains("queued via realtime")); +} + +#[tokio::test] +async fn realtime_handoff_queues_message_while_turn_is_running_by_default() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(None).await; + chat.thread_id = Some(ThreadId::new()); + chat.on_task_started(); + + chat.handle_codex_event(Event { + id: "rt-1".into(), + msg: EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { + payload: RealtimeEvent::HandoffRequested(RealtimeHandoffRequested { + handoff_id: "handoff-1".to_string(), + item_id: "item-1".to_string(), + input_transcript: "queued via realtime".to_string(), + send_immediately: false, + active_transcript: Vec::new(), + }), + }), + }); + + assert_eq!(chat.queued_user_messages.len(), 1); + assert_eq!( + chat.queued_user_messages.front().unwrap().text, + "queued via realtime" + ); + assert!(chat.pending_steers.is_empty()); + assert_no_submit_op(&mut op_rx); +} + +#[tokio::test] +async fn realtime_handoff_can_send_immediately_while_turn_is_running() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(None).await; + chat.thread_id = Some(ThreadId::new()); + chat.on_task_started(); + + chat.handle_codex_event(Event { + id: "rt-2".into(), + msg: EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { + payload: RealtimeEvent::HandoffRequested(RealtimeHandoffRequested { + handoff_id: "handoff-2".to_string(), + item_id: "item-2".to_string(), + input_transcript: "send via realtime".to_string(), + send_immediately: true, + active_transcript: Vec::new(), + }), + }), + }); + + assert!(chat.queued_user_messages.is_empty()); + assert_eq!(chat.pending_steers.len(), 1); + assert_eq!( + chat.pending_steers.front().unwrap().user_message.text, + "send via realtime" + ); + + match next_submit_op(&mut op_rx) { + Op::UserTurn { items, .. } => { + assert_matches!( + items.as_slice(), + [UserInput::Text { text, .. }] if text == "send via realtime" + ); + } + other => panic!("unexpected op: {other:?}"), + } +} + #[tokio::test] async fn review_queues_user_messages_snapshot() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await; diff --git a/codex-rs/tui/src/voice.rs b/codex-rs/tui/src/voice.rs index 7bd566bf60..b27f5f16a7 100644 --- a/codex-rs/tui/src/voice.rs +++ b/codex-rs/tui/src/voice.rs @@ -10,6 +10,7 @@ use codex_login::CodexAuth; use codex_protocol::protocol::ConversationAudioParams; use codex_protocol::protocol::Op; use codex_protocol::protocol::RealtimeAudioFrame; +use codex_protocol::protocol::RealtimeOutputAudioDelta; use cpal::traits::DeviceTrait; use cpal::traits::HostTrait; use cpal::traits::StreamTrait; @@ -44,6 +45,12 @@ pub struct RecordedAudio { pub channels: u16, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct RealtimeAudioPlaybackPosition { + pub(crate) item_id: String, + pub(crate) audio_end_ms: u32, +} + pub struct VoiceCapture { stream: Option, sample_rate: u32, @@ -509,7 +516,8 @@ impl RealtimeAudioPlayer { }) } - pub(crate) fn enqueue_frame(&self, frame: &RealtimeAudioFrame) -> Result<(), String> { + pub(crate) fn enqueue_frame(&self, delta: &RealtimeOutputAudioDelta) -> Result<(), String> { + let frame = &delta.frame; if frame.num_channels == 0 || frame.sample_rate == 0 { return Err("invalid realtime audio frame format".to_string()); } @@ -538,16 +546,23 @@ impl RealtimeAudioPlayer { .lock() .map_err(|_| "failed to lock output audio queue".to_string())?; // TODO(aibrahim): Cap or trim this queue if we observe producer bursts outrunning playback. - guard.samples.extend(converted); + push_output_audio_samples(&mut guard, delta.item_id.as_deref(), converted); Ok(()) } pub(crate) fn clear(&self) { if let Ok(mut guard) = self.queue.lock() { guard.samples.clear(); + guard.spans.clear(); + guard.played_item = None; guard.primed = false; } } + + pub(crate) fn interrupt(&self) -> Option { + let mut guard = self.queue.lock().ok()?; + interrupt_output_audio_queue(&mut guard, self.output_sample_rate, self.output_channels) + } } fn build_output_stream( @@ -589,9 +604,21 @@ fn build_output_stream( #[derive(Default)] struct OutputAudioQueue { samples: VecDeque, + spans: VecDeque, + played_item: Option, primed: bool, } +struct OutputAudioSpan { + item_id: Option, + remaining_samples: usize, +} + +struct PlayedOutputAudioItem { + item_id: String, + played_samples: u64, +} + fn output_prebuffer_samples(sample_rate: u32, channels: u16) -> usize { let samples_per_second = (sample_rate as usize).saturating_mul(channels as usize); ((samples_per_second as u64) * 120 / 1_000) as usize @@ -613,6 +640,94 @@ fn should_output_silence(queue: &mut OutputAudioQueue, min_buffer_samples: usize false } +fn push_output_audio_samples( + queue: &mut OutputAudioQueue, + item_id: Option<&str>, + samples: Vec, +) { + if samples.is_empty() { + return; + } + + let sample_count = samples.len(); + queue.samples.extend(samples); + if let Some(span) = queue.spans.back_mut() + && span.item_id.as_deref() == item_id + { + span.remaining_samples += sample_count; + return; + } + queue.spans.push_back(OutputAudioSpan { + item_id: item_id.map(str::to_string), + remaining_samples: sample_count, + }); +} + +fn pop_output_audio_sample(queue: &mut OutputAudioQueue) -> i16 { + let Some(sample) = queue.samples.pop_front() else { + return 0; + }; + + mark_output_audio_sample_played(queue); + sample +} + +fn mark_output_audio_sample_played(queue: &mut OutputAudioQueue) { + let Some(span) = queue.spans.front_mut() else { + return; + }; + + let item_id = span.item_id.clone(); + + span.remaining_samples = span.remaining_samples.saturating_sub(1); + let span_depleted = span.remaining_samples == 0; + if span_depleted { + queue.spans.pop_front(); + } + + if let Some(item_id) = item_id { + match queue.played_item.as_mut() { + Some(played_item) if played_item.item_id == item_id => { + played_item.played_samples += 1; + } + _ => { + queue.played_item = Some(PlayedOutputAudioItem { + item_id, + played_samples: 1, + }); + } + } + } +} + +fn interrupt_output_audio_queue( + queue: &mut OutputAudioQueue, + output_sample_rate: u32, + output_channels: u16, +) -> Option { + let item_id = queue.spans.iter().find_map(|span| span.item_id.clone()); + let played_samples = item_id.as_ref().map_or(0, |item_id| { + queue + .played_item + .as_ref() + .filter(|played_item| played_item.item_id == *item_id) + .map_or(0, |played_item| played_item.played_samples) + }); + let played_frames = played_samples / u64::from(output_channels.max(1)); + let audio_end_ms = ((played_frames * 1_000) / u64::from(output_sample_rate.max(1))) + .min(u64::from(u32::MAX)) as u32; + + queue.samples.clear(); + queue.spans.clear(); + queue.played_item = None; + queue.primed = false; + + item_id.map(|item_id| RealtimeAudioPlaybackPosition { + item_id, + audio_end_ms, + }) +} + fn fill_output_i16( output: &mut [i16], queue: &Arc>, @@ -624,7 +739,7 @@ fn fill_output_i16( return; } for sample in output { - *sample = guard.samples.pop_front().unwrap_or(0); + *sample = pop_output_audio_sample(&mut guard); } return; } @@ -642,7 +757,7 @@ fn fill_output_f32( return; } for sample in output { - let v = guard.samples.pop_front().unwrap_or(0); + let v = pop_output_audio_sample(&mut guard); *sample = (v as f32) / (i16::MAX as f32); } return; @@ -661,7 +776,7 @@ fn fill_output_u16( return; } for sample in output { - let v = guard.samples.pop_front().unwrap_or(0); + let v = pop_output_audio_sample(&mut guard); *sample = (v as i32 + 32768).clamp(0, u16::MAX as i32) as u16; } return; @@ -926,9 +1041,13 @@ async fn transcribe_bytes( #[cfg(test)] mod tests { + use super::OutputAudioQueue; use super::RecordedAudio; use super::convert_pcm16; use super::encode_wav_normalized; + use super::interrupt_output_audio_queue; + use super::pop_output_audio_sample; + use super::push_output_audio_samples; use pretty_assertions::assert_eq; use std::io::Cursor; @@ -959,4 +1078,36 @@ mod tests { assert_eq!(spec.sample_rate, 24_000); assert_eq!(samples, vec![8_426, 29_490]); } + + #[test] + fn interrupt_output_audio_queue_reports_played_duration_for_current_item() { + let mut queue = OutputAudioQueue::default(); + push_output_audio_samples(&mut queue, Some("item_1"), vec![0; 960]); + queue.primed = true; + + for _ in 0..480 { + let _ = pop_output_audio_sample(&mut queue); + } + + assert_eq!( + interrupt_output_audio_queue(&mut queue, 24_000, 1), + Some(super::RealtimeAudioPlaybackPosition { + item_id: "item_1".to_string(), + audio_end_ms: 20, + }) + ); + assert!(queue.samples.is_empty()); + assert!(queue.spans.is_empty()); + } + + #[test] + fn interrupt_output_audio_queue_ignores_untracked_audio() { + let mut queue = OutputAudioQueue::default(); + push_output_audio_samples(&mut queue, None, vec![0; 480]); + queue.primed = true; + + assert_eq!(interrupt_output_audio_queue(&mut queue, 24_000, 1), None); + assert!(queue.samples.is_empty()); + assert!(queue.spans.is_empty()); + } }