Add realtime voice controls and interruption handling

README.md

@@ -16,6 +16,7 @@ If you want Codex in your code editor (VS Code, Cursor, Windsurf), <a href="http
 
 Install globally with your preferred package manager:
 
+
 ```shell
 # Install using npm
 npm install -g @openai/codex
@@ -33,6 +34,7 @@ Then simply run `codex` to get started.
 
 Each GitHub Release contains many executables, but in practice, you likely want one of these:
 
+
 - macOS
   - Apple Silicon/arm64: `codex-aarch64-apple-darwin.tar.gz`
   - x86_64 (older Mac hardware): `codex-x86_64-apple-darwin.tar.gz`

codex-rs/app-server/src/bespoke_event_handling.rs

@@ -275,12 +275,13 @@ pub(crate) async fn apply_bespoke_event_handling(
             if let ApiVersion::V2 = api_version {
                 match event.payload {
                     RealtimeEvent::SessionUpdated { .. } => {}
+                    RealtimeEvent::InputAudioSpeechStarted(_) => {}
                     RealtimeEvent::InputTranscriptDelta(_) => {}
                     RealtimeEvent::OutputTranscriptDelta(_) => {}
                     RealtimeEvent::AudioOut(audio) => {
                         let notification = ThreadRealtimeOutputAudioDeltaNotification {
                             thread_id: conversation_id.to_string(),
-                            audio: audio.into(),
+                            audio: audio.frame.into(),
                         };
                         outgoing
                             .send_server_notification(
@@ -288,6 +289,7 @@ pub(crate) async fn apply_bespoke_event_handling(
                             )
                             .await;
                     }
+                    RealtimeEvent::ResponseCancelled(_) => {}
                     RealtimeEvent::ConversationItemAdded(item) => {
                         let notification = ThreadRealtimeItemAddedNotification {
                             thread_id: conversation_id.to_string(),
@@ -317,6 +319,21 @@ pub(crate) async fn apply_bespoke_event_handling(
                             ))
                             .await;
                     }
+                    RealtimeEvent::InterruptRequested(interrupt) => {
+                        let notification = ThreadRealtimeItemAddedNotification {
+                            thread_id: conversation_id.to_string(),
+                            item: serde_json::json!({
+                                "type": "interrupt_request",
+                                "call_id": interrupt.call_id,
+                            }),
+                        };
+                        outgoing
+                            .send_server_notification(ServerNotification::ThreadRealtimeItemAdded(
+                                notification,
+                            ))
+                            .await;
+                    }
+                    RealtimeEvent::CloseRequested(_) => {}
                     RealtimeEvent::Error(message) => {
                         let notification = ThreadRealtimeErrorNotification {
                             thread_id: conversation_id.to_string(),

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(),

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<String, SessionToolProperty>,
     pub(super) required: Vec<String>,
 }
 
-#[derive(Debug, Clone, Serialize)]
-pub(super) struct SessionToolProperties {
-    pub(super) prompt: SessionToolProperty,
-}
-
 #[derive(Debug, Clone, Serialize)]
 pub(super) struct SessionToolProperty {
     #[serde(rename = "type")]
@@ -163,7 +177,10 @@ pub(super) fn parse_realtime_event(payload: &str) -> Option<RealtimeEvent> {
                 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<RealtimeEvent> {
                 .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<RealtimeEvent> {
                 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<RealtimeEvent> {
             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<RealtimeEvent> {
 }
 
 fn parse_handoff_requested(parsed: &Value) -> Option<RealtimeHandoffRequested> {
-    let outputs = parsed
-        .get("response")
-        .and_then(Value::as_object)
-        .and_then(|response| response.get("output"))
-        .and_then(Value::as_array)?;
-    let function_call = outputs.iter().find(|item| {
-        item.get("type").and_then(Value::as_str) == Some("function_call")
-            && item.get("name").and_then(Value::as_str) == Some("codex")
-    })?;
+    let 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<RealtimeHandoffRequested> {
         .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<RealtimeInterruptRequested> {
+    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<RealtimeCloseRequested> {
+    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<RealtimeResponseCancelled> {
+    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<String>,
         #[serde(default)]
+        send_immediately: bool,
+        #[serde(default)]
         messages: Vec<RealtimeTranscriptEntry>,
     }
 
     let Some(parsed) = serde_json::from_str::<HandoffArguments>(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,
+    }
 }

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,
         })
     );
 

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<TurnContext>, CancellationToken)> {
@@ -4053,6 +4059,11 @@ async fn submission_loop(sess: Arc<Session>, config: Arc<Config>, 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<Session>, config: Arc<Config>, 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()),

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<Option<ConversationState>>,
@@ -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<RealtimeAudioFrame>,
     user_text_tx: Sender<String>,
+    writer: RealtimeWebsocketWriter,
     handoff: RealtimeHandoffState,
     task: JoinHandle<()>,
     realtime_active: Arc<AtomicBool>,
@@ -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<String> {
         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<Session>,
+    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<String> {
     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(),
+            }
+        );
+    }
 }

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()

codex-rs/protocol/src/protocol.rs

@@ -131,6 +131,13 @@ pub struct RealtimeAudioFrame {
     pub samples_per_channel: Option<u32>,
 }
 
+#[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<String>,
+}
+
 #[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<RealtimeTranscriptEntry>,
 }
 
+#[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<String>,
+}
+
+#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]
+pub struct RealtimeResponseCancelled {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub response_id: Option<String>,
+}
+
 #[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::<Op>(serde_json::to_value(&truncate).unwrap()).unwrap(),
+            truncate
+        );
         assert_eq!(
             serde_json::from_value::<Op>(serde_json::to_value(&text).unwrap()).unwrap(),
             text

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;
                     };

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<String>,
-    warned_audio_only_submission: bool,
+    warned_unsupported_composer_submission: bool,
     meter_placeholder_id: Option<String>,
     #[cfg(not(target_os = "linux"))]
     capture_stop_flag: Option<Arc<AtomicBool>>,
@@ -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<UserMessage> {
@@ -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::<Vec<_>>();
+            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<String> {
+    let active_transcript = handoff
+        .active_transcript
+        .iter()
+        .map(|entry| format!("{}: {}", entry.role, entry.text))
+        .collect::<Vec<_>>()
+        .join("\n");
+    (!active_transcript.is_empty())
+        .then_some(active_transcript)
+        .or_else(|| {
+            (!handoff.input_transcript.is_empty()).then(|| handoff.input_transcript.clone())
+        })
+}

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

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>) -> Op {
     }
 }
 
+fn next_realtime_text(op_rx: &mut tokio::sync::mpsc::UnboundedReceiver<Op>) -> 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<Op>) {
     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;

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<cpal::Stream>,
     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<RealtimeAudioPlaybackPosition> {
+        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<i16>,
+    spans: VecDeque<OutputAudioSpan>,
+    played_item: Option<PlayedOutputAudioItem>,
     primed: bool,
 }
 
+struct OutputAudioSpan {
+    item_id: Option<String>,
+    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<i16>,
+) {
+    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<RealtimeAudioPlaybackPosition> {
+    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<Mutex<OutputAudioQueue>>,
@@ -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());
+    }
 }