Voice mode: surface session and request errors

crates/code_assistant_core/src/voice/agent.rs

@@ -15,11 +15,12 @@ use super::{
 use crate::session::SessionService;
 use crate::session::event_stream::{EventStream, StreamError, Subscription};
 use crate::ui::UiEvent;
-use anyhow::{Context, Result, anyhow};
+use anyhow::{Context, Result, anyhow, bail};
 use llm::realtime::{
-    ClientEvent, Incoming, RealtimeConnection, RealtimeConnector, SAMPLE_RATE, ServerEvent,
-    SessionSettings, decode_pcm16,
+    ClientEvent, ErrorCause, Incoming, RealtimeConnection, RealtimeConnector, SAMPLE_RATE,
+    ServerEvent, SessionSettings, decode_pcm16,
 };
+use std::collections::HashSet;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::mpsc;
@@ -35,13 +36,15 @@ pub(super) enum Control {
 
 const RECONNECT_ATTEMPTS: u32 = 3;
 /// Bounds one connection attempt, including the credentials and endpoint
-/// lookups a connector does first.
+/// lookups a connector does first and the server's `session.created`.
 const CONNECT_TIMEOUT: Duration = Duration::from_secs(20);
 /// Bounds one voice tool call; all of them return without waiting for a
 /// conversation's agent.
 const TOOL_TIMEOUT: Duration = Duration::from_secs(30);
 /// Transcript entries replayed into a renewed realtime session.
 const RESEED_ENTRIES: usize = 30;
+/// Appended to an assistant transcript the user barged into.
+const INTERRUPTED_MARK: &str = "[interrupted by the user]";
 
 const INSTRUCTIONS: &str = "\
 You are the voice assistant of a coding-agent app. The user talks to you hands-free about their \
@@ -80,6 +83,12 @@ pub(super) struct VoiceAgent {
     muted: bool,
     status: VoiceStatus,
     transcript: Vec<TranscriptEntry>,
+    /// The assistant item of the latest assistant transcript entry, and the
+    /// entry's index.
+    last_assistant_entry: Option<(String, usize)>,
+    /// Assistant items the user barged into: their transcript says more than
+    /// the user heard.
+    interrupted_items: HashSet<String>,
     /// Samples handed to the speakers since the devices opened.
     queued_samples: u64,
     /// The assistant item whose audio plays, with the sample offset its
@@ -105,7 +114,8 @@ impl VoiceAgent {
             NotificationSource::new(service.clone(), config.notify, open_requests.clone()).await?;
         let tools = VoiceTools::new(service, open_requests);
 
-        let connection = connect(connector.as_ref()).await?;
+        let settings = session_settings(&config, connector.as_ref());
+        let connection = connect(connector.as_ref(), settings).await?;
         let (audio_tx, audio_rx) = mpsc::unbounded_channel();
         let audio = audio_factory(audio_tx).context("Failed to open the audio devices")?;
         let (tool_tx, tool_rx) = mpsc::unbounded_channel();
@@ -131,14 +141,15 @@ impl VoiceAgent {
             timer: None,
             muted: false,
             status: VoiceStatus {
-                activity: VoiceActivity::Connecting,
+                activity: VoiceActivity::Listening,
                 ..VoiceStatus::default()
             },
             transcript: Vec::new(),
+            last_assistant_entry: None,
+            interrupted_items: HashSet::new(),
             queued_samples: 0,
             audio_item: None,
         };
-        agent.send(ClientEvent::SessionUpdate(agent.session_settings()));
         Ok(agent)
     }
 
@@ -202,17 +213,24 @@ impl VoiceAgent {
 
     fn on_server_event(&mut self, event: ServerEvent) {
         match event {
-            ServerEvent::SessionCreated | ServerEvent::SessionUpdated => {
-                if self.status.activity == VoiceActivity::Connecting {
-                    self.status.activity = VoiceActivity::Listening;
-                    self.publish_status();
-                }
-            }
+            // `connect` waited for `session.created`.
+            ServerEvent::SessionCreated | ServerEvent::SessionUpdated => {}
             ServerEvent::ResponseCreated { .. } => self.floor_input(FloorInput::ResponseCreated),
             ServerEvent::OutputItemAdded { .. } => {}
             ServerEvent::AudioDelta { item_id, delta } => self.on_audio_delta(item_id, &delta),
-            ServerEvent::AudioTranscriptDone { transcript, .. } => {
-                self.record(TranscriptRole::Assistant, transcript)
+            ServerEvent::AudioTranscriptDone {
+                item_id,
+                transcript,
+            } => {
+                let transcript = if self.interrupted_items.contains(&item_id) {
+                    format!("{transcript} {INTERRUPTED_MARK}")
+                } else {
+                    transcript
+                };
+                self.record(TranscriptRole::Assistant, transcript);
+                if self.transcript.last().map(|e| e.role) == Some(TranscriptRole::Assistant) {
+                    self.last_assistant_entry = Some((item_id, self.transcript.len() - 1));
+                }
             }
             ServerEvent::InputTranscription { transcript, .. } => {
                 self.record(TranscriptRole::User, transcript)
@@ -232,19 +250,44 @@ impl VoiceAgent {
                     "Realtime response {} done: {:?}",
                     response.id, response.status
                 );
+                match response.status.as_deref() {
+                    Some("failed") => self.report_error(format!(
+                        "The model's answer failed: {}",
+                        response
+                            .status_reason()
+                            .unwrap_or_else(|| "no reason given".into())
+                    )),
+                    Some("incomplete") => warn!(
+                        "Realtime response {} incomplete: {}",
+                        response.id,
+                        response.status_reason().unwrap_or_default()
+                    ),
+                    Some("completed") if self.status.error.is_some() => {
+                        self.status.error = None;
+                        self.publish_status();
+                    }
+                    _ => {}
+                }
                 let pending = self.pending_playback();
                 self.floor_input(FloorInput::ResponseDone {
                     pending_playback: pending,
                 });
             }
-            ServerEvent::Error { error } => {
-                warn!(
-                    "Realtime error ({}): {}",
+            ServerEvent::Error { error } => match error.cause() {
+                // Cancelling a response that just ended, truncating audio
+                // that raced the item's end: the floor moved on already.
+                ErrorCause::CancelResponse | ErrorCause::TruncateItem => debug!(
+                    "Realtime error for our {:?} ({}): {}",
+                    error.cause(),
                     error.code.as_deref().unwrap_or("-"),
                     error.message
-                );
-                self.floor_input(FloorInput::ServerError);
-            }
+                ),
+                ErrorCause::CreateResponse => {
+                    self.report_error(error.message);
+                    self.floor_input(FloorInput::CreateFailed);
+                }
+                ErrorCause::Other => self.report_error(error.message),
+            },
             ServerEvent::Other => {}
         }
     }
@@ -350,6 +393,7 @@ impl VoiceAgent {
         if let Some((item_id, start)) = self.audio_item.take()
             && played < self.queued_samples
         {
+            self.mark_interrupted(&item_id);
             let heard = played.saturating_sub(start);
             self.send(ClientEvent::TruncateItem {
                 item_id,
@@ -382,16 +426,18 @@ impl VoiceAgent {
             if attempt > 0 {
                 tokio::time::sleep(Duration::from_secs(1 << attempt)).await;
             }
-            match connect(self.connector.as_ref()).await {
+            let settings = session_settings(&self.config, self.connector.as_ref());
+            match connect(self.connector.as_ref(), settings).await {
                 Ok(connection) => {
                     self.connection = connection;
                     self.generation += 1;
-                    self.send(ClientEvent::SessionUpdate(self.session_settings()));
                     if let Some(summary) = self.transcript_summary() {
                         self.send(ClientEvent::SystemMessage(summary));
                     }
                     let commands = self.floor.reconnected();
                     self.execute(commands);
+                    self.status.activity = self.floor_activity();
+                    self.publish_status();
                     return Ok(());
                 }
                 Err(e) => {
@@ -411,7 +457,7 @@ impl VoiceAgent {
                 TranscriptRole::User => Some(format!("User: {}", entry.text)),
                 TranscriptRole::Assistant => Some(format!("You: {}", entry.text)),
                 TranscriptRole::Tool => Some(format!("(tool call: {})", entry.text)),
-                TranscriptRole::Notification => None,
+                TranscriptRole::Notification | TranscriptRole::Error => None,
             })
             .collect();
         (!lines.is_empty()).then(|| {
@@ -422,17 +468,25 @@ impl VoiceAgent {
         })
     }
 
-    fn session_settings(&self) -> SessionSettings {
-        SessionSettings {
-            model: self
-                .connector
-                .declares_model()
-                .then(|| self.config.model.clone()),
-            instructions: INSTRUCTIONS.to_string(),
-            voice: self.config.voice.clone(),
-            tools: VoiceTools::definitions(),
-            turn_detection: self.config.turn_detection(),
-            transcription_model: self.config.transcription(),
+    /// An error the session survives: logged, shown in the transcript and
+    /// on the status until a response completes again.
+    fn report_error(&mut self, message: String) {
+        warn!("Realtime error: {message}");
+        self.record(TranscriptRole::Error, message.clone());
+        self.status.error = Some(message);
+        self.publish_status();
+    }
+
+    /// The user barged into `item_id`. Its transcript, already recorded or
+    /// still to come, says more than the user heard.
+    fn mark_interrupted(&mut self, item_id: &str) {
+        self.interrupted_items.insert(item_id.to_string());
+        if let Some((id, index)) = &self.last_assistant_entry
+            && id == item_id
+            && let Some(entry) = self.transcript.get_mut(*index)
+        {
+            entry.text = format!("{} {INTERRUPTED_MARK}", entry.text);
+            self.last_assistant_entry = None;
         }
     }
 
@@ -456,11 +510,7 @@ impl VoiceAgent {
         if self.status.activity == VoiceActivity::Connecting {
             return;
         }
-        let activity = match self.floor.state() {
-            FloorState::Speaking => VoiceActivity::Speaking,
-            FloorState::UserTurn => VoiceActivity::UserSpeaking,
-            FloorState::Idle | FloorState::Cooling => VoiceActivity::Listening,
-        };
+        let activity = self.floor_activity();
         let queued = self.floor.queued_notifications();
         if activity != self.status.activity || queued != self.status.queued_notifications {
             self.status.activity = activity;
@@ -469,6 +519,14 @@ impl VoiceAgent {
         }
     }
 
+    fn floor_activity(&self) -> VoiceActivity {
+        match self.floor.state() {
+            FloorState::Speaking => VoiceActivity::Speaking,
+            FloorState::UserTurn => VoiceActivity::UserSpeaking,
+            FloorState::Idle | FloorState::Cooling => VoiceActivity::Listening,
+        }
+    }
+
     fn publish_status(&mut self) {
         self.status.muted = self.muted;
         self.events.publish_app(UiEvent::VoiceStatusChanged {
@@ -477,15 +535,69 @@ impl VoiceAgent {
     }
 }
 
-async fn connect(connector: &dyn RealtimeConnector) -> Result<RealtimeConnection> {
-    tokio::time::timeout(CONNECT_TIMEOUT, connector.connect())
-        .await
-        .map_err(|_| {
-            anyhow!(
-                "No realtime connection after {}s",
-                CONNECT_TIMEOUT.as_secs()
-            )
-        })?
+/// Open a session, configure it, and wait until the server accepted both.
+/// A server that rejects the session (an invalid API key, an unknown model)
+/// still accepts the WebSocket, then sends an `error` event and closes; a
+/// rejected `session.update` (an unknown voice or transcription model, a
+/// tool schema) would leave the session on server defaults. Either is a
+/// failed attempt carrying the server's message, not a connection.
+async fn connect(
+    connector: &dyn RealtimeConnector,
+    settings: SessionSettings,
+) -> Result<RealtimeConnection> {
+    tokio::time::timeout(CONNECT_TIMEOUT, async {
+        let mut connection = connector.connect().await?;
+        let mut configuring = false;
+        loop {
+            match connection.incoming.recv().await {
+                Some(Incoming::Event(ServerEvent::SessionCreated)) if !configuring => {
+                    let _ = connection
+                        .outgoing
+                        .send(ClientEvent::SessionUpdate(settings.clone()));
+                    configuring = true;
+                }
+                Some(Incoming::Event(ServerEvent::SessionUpdated)) if configuring => {
+                    return Ok(connection);
+                }
+                Some(Incoming::Event(ServerEvent::Error { error })) if configuring => bail!(
+                    "The realtime server rejected the voice settings: {}",
+                    error.message
+                ),
+                Some(Incoming::Event(ServerEvent::Error { error })) => {
+                    bail!(
+                        "The realtime server rejected the session: {}",
+                        error.message
+                    )
+                }
+                Some(Incoming::Event(_)) => {}
+                Some(Incoming::Closed(reason)) => bail!(
+                    "The realtime server closed the connection before the session started{}",
+                    reason.map(|r| format!(" ({r})")).unwrap_or_default()
+                ),
+                None => {
+                    bail!("The realtime server closed the connection before the session started")
+                }
+            }
+        }
+    })
+    .await
+    .map_err(|_| {
+        anyhow!(
+            "No realtime connection after {}s",
+            CONNECT_TIMEOUT.as_secs()
+        )
+    })?
+}
+
+fn session_settings(config: &VoiceConfig, connector: &dyn RealtimeConnector) -> SessionSettings {
+    SessionSettings {
+        model: connector.declares_model().then(|| config.model.clone()),
+        instructions: INSTRUCTIONS.to_string(),
+        voice: config.voice.clone(),
+        tools: VoiceTools::definitions(),
+        turn_detection: config.turn_detection(),
+        transcription_model: config.transcription(),
+    }
 }
 
 fn samples_to_ms(samples: u64) -> u64 {

crates/code_assistant_core/src/voice/config.rs

@@ -147,6 +147,16 @@ mod tests {
         assert!(!supports_realtime(&provider("anthropic", json!({}))));
     }
 
+    #[test]
+    fn a_chatgpt_subscription_login_serves_no_realtime() {
+        let p = provider("openai-responses-ws", json!({"codex_auth": true}));
+        assert!(!supports_realtime(&p));
+        let err = llm::realtime::connector_for_provider(&p, "gpt-realtime", None)
+            .err()
+            .unwrap();
+        assert!(err.to_string().contains("ChatGPT subscription"));
+    }
+
     #[test]
     fn ai_core_needs_a_deployment_for_the_voice_model() {
         let p = provider(