From 1df7c442451c488d81b8abdd426c97b8d3a363c6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Stephan=20A=C3=9Fmus?= Date: Fri, 9 Oct 2026 08:36:38 +0200 Subject: [PATCH 1/4] fix(voice): do not offer a ChatGPT subscription login as realtime provider Its token grants no access to the Realtime API, so voice mode could only fail with a missing api_key. The settings no longer list such a provider, and a voice.json that still names one fails with the reason. --- .../code_assistant_core/src/voice/config.rs | 10 ++++++ crates/llm/src/realtime/mod.rs | 33 +++++++++++++------ .../src/settings_screen/voice_section.rs | 5 +-- 3 files changed, 36 insertions(+), 12 deletions(-) diff --git a/crates/code_assistant_core/src/voice/config.rs b/crates/code_assistant_core/src/voice/config.rs index 25ec7b67..8cfe1479 100644 --- a/crates/code_assistant_core/src/voice/config.rs +++ b/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( diff --git a/crates/llm/src/realtime/mod.rs b/crates/llm/src/realtime/mod.rs index 52148854..29ed6130 100644 --- a/crates/llm/src/realtime/mod.rs +++ b/crates/llm/src/realtime/mod.rs @@ -54,19 +54,25 @@ pub trait RealtimeConnector: Send + Sync { } } -/// Provider types that can serve realtime sessions. -pub const REALTIME_PROVIDER_TYPES: &[&str] = &[ - "openai", - "openai-responses", - "openai-responses-ws", - "ai-core", -]; - const DEFAULT_OPENAI_BASE_URL: &str = "https://api.openai.com/v1"; -/// Whether a provider can serve realtime sessions. +/// Whether a provider can serve realtime sessions: OpenAI types with an API +/// key, and AI Core. A ChatGPT subscription login (`codex_auth`) cannot; its +/// token grants no access to the Realtime API. pub fn supports_realtime(provider: &ProviderConfig) -> bool { - REALTIME_PROVIDER_TYPES.contains(&provider.provider.as_str()) + match provider.provider.as_str() { + "openai" | "openai-responses" | "openai-responses-ws" => !uses_codex_auth(provider), + "ai-core" => true, + _ => false, + } +} + +fn uses_codex_auth(provider: &ProviderConfig) -> bool { + provider + .config + .get("codex_auth") + .and_then(Value::as_bool) + .unwrap_or(false) } /// The connector for `model` on a provider from `providers.json`. @@ -91,6 +97,13 @@ pub fn connector_for_provider( }; match provider.provider.as_str() { "openai" | "openai-responses" | "openai-responses-ws" => { + if uses_codex_auth(provider) { + bail!( + "Provider '{}' signs in with a ChatGPT subscription, which has no access to \ + the Realtime API; voice mode needs an OpenAI provider with an API key", + provider.label + ); + } let api_key = text("api_key")?; let base_url = text("base_url").unwrap_or_else(|_| DEFAULT_OPENAI_BASE_URL.into()); let mut endpoint = RealtimeEndpoint::openai(&base_url, &api_key, model); diff --git a/crates/ui_gpui/src/settings_screen/voice_section.rs b/crates/ui_gpui/src/settings_screen/voice_section.rs index 15249d78..e13e87a6 100644 --- a/crates/ui_gpui/src/settings_screen/voice_section.rs +++ b/crates/ui_gpui/src/settings_screen/voice_section.rs @@ -276,8 +276,9 @@ impl Render for VoiceSection { .text_sm() .text_color(cx.theme().muted_foreground) .child( - "No provider with a realtime API yet. Add an OpenAI or AI Core \ - provider under Providers.", + "No provider with a realtime API yet. Add an OpenAI provider with \ + an API key (a ChatGPT subscription has no realtime access) or \ + an AI Core provider under Providers.", ) .into_any_element() } else { From 4164918903315bdc449fc9c3198319289afb7138 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Stephan=20A=C3=9Fmus?= Date: Fri, 9 Oct 2026 08:36:38 +0200 Subject: [PATCH 2/4] fix(voice): fail with the server's reason when it rejects the session OpenAI accepts the WebSocket even with an invalid API key, then sends an error event and closes. connect() counted the handshake as success, so the agent reconnected forever and the status stayed on Connecting. A connection now counts once session.created arrives; an error event or a close before that fails the attempt with the server's message. --- crates/code_assistant_core/src/voice/agent.rs | 71 ++++++++---- crates/code_assistant_core/src/voice/tests.rs | 109 +++++++++++++++++- 2 files changed, 157 insertions(+), 23 deletions(-) diff --git a/crates/code_assistant_core/src/voice/agent.rs b/crates/code_assistant_core/src/voice/agent.rs index ec158088..101cb32e 100644 --- a/crates/code_assistant_core/src/voice/agent.rs +++ b/crates/code_assistant_core/src/voice/agent.rs @@ -15,7 +15,7 @@ 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, @@ -35,7 +35,7 @@ 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. @@ -131,7 +131,7 @@ impl VoiceAgent { timer: None, muted: false, status: VoiceStatus { - activity: VoiceActivity::Connecting, + activity: VoiceActivity::Listening, ..VoiceStatus::default() }, transcript: Vec::new(), @@ -202,12 +202,8 @@ 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), @@ -392,6 +388,8 @@ impl VoiceAgent { } let commands = self.floor.reconnected(); self.execute(commands); + self.status.activity = self.floor_activity(); + self.publish_status(); return Ok(()); } Err(e) => { @@ -456,11 +454,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 +463,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 +479,40 @@ impl VoiceAgent { } } +/// Open a session and wait until the server confirms it. A server that +/// rejects the session (an invalid API key, an unknown model) still accepts +/// the WebSocket, then sends an `error` event and closes; that is a failed +/// attempt carrying the server's message, not a connection. async fn connect(connector: &dyn RealtimeConnector) -> Result { - tokio::time::timeout(CONNECT_TIMEOUT, connector.connect()) - .await - .map_err(|_| { - anyhow!( - "No realtime connection after {}s", - CONNECT_TIMEOUT.as_secs() - ) - })? + tokio::time::timeout(CONNECT_TIMEOUT, async { + let mut connection = connector.connect().await?; + loop { + match connection.incoming.recv().await { + Some(Incoming::Event(ServerEvent::SessionCreated)) => return Ok(connection), + 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 samples_to_ms(samples: u64) -> u64 { diff --git a/crates/code_assistant_core/src/voice/tests.rs b/crates/code_assistant_core/src/voice/tests.rs index 15c72720..7a0839ac 100644 --- a/crates/code_assistant_core/src/voice/tests.rs +++ b/crates/code_assistant_core/src/voice/tests.rs @@ -219,6 +219,8 @@ async fn harness() -> Harness { let dir = tempfile::tempdir().unwrap(); let (sessions, events) = session_service(&dir); let (connection, mut server) = scripted_connection(); + // Like OpenAI, the server confirms the session before the client speaks. + server.send(ServerEvent::SessionCreated); let connector: Arc = Arc::new(ScriptedConnector { connections: Mutex::new(vec![connection]), }); @@ -237,7 +239,6 @@ async fn harness() -> Harness { matches!(e, ClientEvent::SessionUpdate(_)) }) .await; - server.send(ServerEvent::SessionCreated); Harness { sessions, voice, @@ -554,3 +555,109 @@ async fn a_connection_attempt_that_hangs_times_out() { }; assert!(message.contains("No realtime connection"), "{message}"); } + +/// A voice service over scripted connections (handed out in order), with a +/// subscription to its status events. +fn scripted_voice( + dir: &tempfile::TempDir, + mut connections: Vec, +) -> (VoiceService, crate::session::event_stream::Subscription) { + let (sessions, events) = session_service(dir); + let statuses = sessions.subscribe(); + connections.reverse(); + let connector: Arc = Arc::new(ScriptedConnector { + connections: Mutex::new(connections), + }); + let (voice, worker) = VoiceService::with_config_loader( + sessions, + events, + FakeAudio::default().factory(), + Arc::new(move |_| Ok(connector.clone())), + Arc::new(test_config), + ); + tokio::spawn(worker); + (voice, statuses) +} + +/// What OpenAI does with an invalid API key: accept the WebSocket, send an +/// `error` event, close. +fn rejecting_connection() -> RealtimeConnection { + let (connection, server) = scripted_connection(); + server.send(ServerEvent::Error { + error: llm::realtime::ErrorInfo { + message: "Incorrect API key provided".into(), + code: Some("invalid_api_key".into()), + }, + }); + server + .to_client + .send(Incoming::Closed(Some( + "3000 invalid_request_error.invalid_api_key".into(), + ))) + .unwrap(); + connection +} + +async fn wait_for_failure( + statuses: &mut crate::session::event_stream::Subscription, + within: Duration, +) -> String { + let status = wait_for_status(statuses, within, |s| { + matches!(s.activity, VoiceActivity::Failed(_)) + }) + .await; + let VoiceActivity::Failed(message) = status.activity else { + unreachable!() + }; + message +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_rejected_session_fails_with_the_servers_reason() { + let dir = tempfile::tempdir().unwrap(); + let (voice, mut statuses) = scripted_voice(&dir, vec![rejecting_connection()]); + voice.start(); + let message = wait_for_failure(&mut statuses, WAIT).await; + assert!(message.contains("Incorrect API key provided"), "{message}"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_connection_closed_before_the_session_started_fails() { + let dir = tempfile::tempdir().unwrap(); + let (connection, server) = scripted_connection(); + server + .to_client + .send(Incoming::Closed(Some("1011 internal error".into()))) + .unwrap(); + let (voice, mut statuses) = scripted_voice(&dir, vec![connection]); + voice.start(); + let message = wait_for_failure(&mut statuses, WAIT).await; + assert!(message.contains("1011 internal error"), "{message}"); +} + +#[tokio::test(start_paused = true)] +async fn rejected_reconnects_end_voice_mode_instead_of_looping() { + let dir = tempfile::tempdir().unwrap(); + let (first, server) = scripted_connection(); + server.send(ServerEvent::SessionCreated); + let (voice, mut statuses) = scripted_voice( + &dir, + vec![ + first, + rejecting_connection(), + rejecting_connection(), + rejecting_connection(), + ], + ); + voice.start(); + wait_for_status(&mut statuses, WAIT, |s| { + s.activity == VoiceActivity::Listening + }) + .await; + // The session ends (its maximum duration, say); every renewal is refused. + server.to_client.send(Incoming::Closed(None)).unwrap(); + // Paused time runs ahead through the reconnect backoff. + let message = wait_for_failure(&mut statuses, Duration::from_secs(60)).await; + assert!(message.contains("could not be renewed"), "{message}"); + assert!(message.contains("Incorrect API key provided"), "{message}"); +} From 705063267e9b8701af5ce8f70f80eb39c2d9814e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Stephan=20A=C3=9Fmus?= Date: Fri, 9 Oct 2026 09:50:56 +0200 Subject: [PATCH 3/4] feat(realtime): tag client events and expose response failure reasons response.create, response.cancel and conversation.item.truncate carry an event_id with a per-kind prefix, so ErrorInfo::cause tells which of them an error answers. ResponseInfo keeps status_details and names the reason of a failed or incomplete response. --- crates/llm/src/realtime/events.rs | 121 ++++++++++++++++++++++++++++++ crates/llm/src/realtime/mod.rs | 2 +- 2 files changed, 122 insertions(+), 1 deletion(-) diff --git a/crates/llm/src/realtime/events.rs b/crates/llm/src/realtime/events.rs index dd0b3eb2..82174d1e 100644 --- a/crates/llm/src/realtime/events.rs +++ b/crates/llm/src/realtime/events.rs @@ -8,6 +8,7 @@ use base64::Engine; use serde::Deserialize; use serde_json::{Value, json}; +use std::sync::atomic::{AtomicU64, Ordering}; /// Sample rate of the PCM16 mono audio both directions use. pub const SAMPLE_RATE: u32 = 24_000; @@ -59,8 +60,33 @@ pub enum ClientEvent { SystemMessage(String), } +/// Event id prefixes of the client events whose errors callers tell apart +/// (see [`ErrorInfo::cause`]). +const CREATE_ID_PREFIX: &str = "create-"; +const CANCEL_ID_PREFIX: &str = "cancel-"; +const TRUNCATE_ID_PREFIX: &str = "truncate-"; + +fn event_id(prefix: &str) -> String { + static NEXT: AtomicU64 = AtomicU64::new(1); + format!("{prefix}{}", NEXT.fetch_add(1, Ordering::Relaxed)) +} + impl ClientEvent { pub fn to_json(&self) -> Value { + let mut event = self.payload(); + let prefix = match self { + ClientEvent::CreateResponse => Some(CREATE_ID_PREFIX), + ClientEvent::CancelResponse => Some(CANCEL_ID_PREFIX), + ClientEvent::TruncateItem { .. } => Some(TRUNCATE_ID_PREFIX), + _ => None, + }; + if let Some(prefix) = prefix { + event["event_id"] = json!(event_id(prefix)); + } + event + } + + fn payload(&self) -> Value { match self { ClientEvent::SessionUpdate(settings) => { let tools: Vec = settings @@ -162,6 +188,22 @@ pub struct ResponseInfo { /// `completed`, `cancelled`, `failed` or `incomplete` once done. #[serde(default)] pub status: Option, + /// Why a response failed or stopped early. + #[serde(default)] + pub status_details: Option, +} + +impl ResponseInfo { + /// The reason in `status_details`: the error's message for a failed + /// response, the reason (`max_output_tokens`, `content_filter`) for an + /// incomplete one. + pub fn status_reason(&self) -> Option { + let details = self.status_details.as_ref()?; + let text = |value: Option<&Value>| value.and_then(Value::as_str).map(str::to_string); + text(details.pointer("/error/message")) + .or_else(|| text(details.pointer("/error/code"))) + .or_else(|| text(details.get("reason"))) + } } #[derive(Debug, Clone, PartialEq, Deserialize)] @@ -170,6 +212,30 @@ pub struct ErrorInfo { pub message: String, #[serde(default)] pub code: Option, + /// The client event the error answers, when it answers one. + #[serde(default)] + pub event_id: Option, +} + +/// Which of our client events an error answers. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ErrorCause { + CreateResponse, + CancelResponse, + TruncateItem, + /// Another event, or the session itself. + Other, +} + +impl ErrorInfo { + pub fn cause(&self) -> ErrorCause { + match self.event_id.as_deref() { + Some(id) if id.starts_with(CREATE_ID_PREFIX) => ErrorCause::CreateResponse, + Some(id) if id.starts_with(CANCEL_ID_PREFIX) => ErrorCause::CancelResponse, + Some(id) if id.starts_with(TRUNCATE_ID_PREFIX) => ErrorCause::TruncateItem, + _ => ErrorCause::Other, + } + } } /// A server event. Event names follow the GA protocol; the beta names of @@ -259,6 +325,61 @@ mod tests { } } + #[test] + fn errors_name_the_client_event_they_answer() { + let create = ClientEvent::CreateResponse.to_json(); + let id = create["event_id"].as_str().unwrap().to_string(); + let text = format!( + r#"{{"type":"error","error":{{"type":"invalid_request_error", + "code":"conversation_already_has_active_response","message":"busy","event_id":"{id}"}}}}"# + ); + let ServerEvent::Error { error } = ServerEvent::parse(&text).unwrap() else { + panic!("wrong event"); + }; + assert_eq!(error.cause(), ErrorCause::CreateResponse); + + let cancel = ClientEvent::CancelResponse.to_json(); + let truncate = ClientEvent::TruncateItem { + item_id: "i".into(), + audio_end_ms: 1, + } + .to_json(); + let cause = |event: &Value| { + ErrorInfo { + message: String::new(), + code: None, + event_id: event["event_id"].as_str().map(str::to_string), + } + .cause() + }; + assert_eq!(cause(&cancel), ErrorCause::CancelResponse); + assert_eq!(cause(&truncate), ErrorCause::TruncateItem); + assert_eq!(cause(&json!({})), ErrorCause::Other); + assert_ne!( + create["event_id"], + ClientEvent::CreateResponse.to_json()["event_id"] + ); + } + + #[test] + fn response_status_reasons() { + let failed: ResponseInfo = serde_json::from_value(json!({ + "id": "r", "status": "failed", + "status_details": {"type": "failed", "error": {"type": "server_error", "message": "boom"}} + })) + .unwrap(); + assert_eq!(failed.status_reason().as_deref(), Some("boom")); + let incomplete: ResponseInfo = serde_json::from_value(json!({ + "id": "r", "status": "incomplete", + "status_details": {"type": "incomplete", "reason": "max_output_tokens"} + })) + .unwrap(); + assert_eq!( + incomplete.status_reason().as_deref(), + Some("max_output_tokens") + ); + } + #[test] fn unknown_events_parse_as_other() { let event = ServerEvent::parse(r#"{"type":"rate_limits.updated","rate_limits":[]}"#); diff --git a/crates/llm/src/realtime/mod.rs b/crates/llm/src/realtime/mod.rs index 29ed6130..3af6be34 100644 --- a/crates/llm/src/realtime/mod.rs +++ b/crates/llm/src/realtime/mod.rs @@ -15,7 +15,7 @@ mod transport; pub use aicore::AiCoreConnector; pub use events::{ - ClientEvent, ConversationItem, ErrorInfo, ResponseInfo, SAMPLE_RATE, ServerEvent, + ClientEvent, ConversationItem, ErrorCause, ErrorInfo, ResponseInfo, SAMPLE_RATE, ServerEvent, SessionSettings, ToolDefinition, decode_pcm16, encode_pcm16, }; pub use transport::{RealtimeEndpoint, WsConnector}; From 050894739099da98ebea12d64a41d28f68e79378 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Stephan=20A=C3=9Fmus?= Date: Fri, 9 Oct 2026 09:50:57 +0200 Subject: [PATCH 4/4] fix(voice): confirm the session settings and show errors the session survives - connect() sends session.update itself and waits for session.updated. A refused update (unknown voice or transcription model, a tool schema) fails the attempt instead of leaving the session on server defaults, without tools or instructions. - Errors are told apart by the client event they answer: those for our cancel or truncate are logged only, one for our response.create frees the floor (before, any error did, even while a response was coming). Every other error and a failed response show as an Error line in the transcript and on the status chip until an answer completes again. - A watchdog frees the floor when response.create gets no response.created within 10 s. - An answer the user barged into is marked as interrupted in the transcript that seeds a renewed session. --- crates/code_assistant_core/src/voice/agent.rs | 149 ++++++++++++++---- crates/code_assistant_core/src/voice/floor.rs | 70 ++++++-- crates/code_assistant_core/src/voice/mod.rs | 5 + crates/code_assistant_core/src/voice/tests.rs | 142 ++++++++++++++++- .../ui_gpui/src/main_screen/voice_control.rs | 16 +- docs/voice-mode.md | 21 ++- 6 files changed, 346 insertions(+), 57 deletions(-) diff --git a/crates/code_assistant_core/src/voice/agent.rs b/crates/code_assistant_core/src/voice/agent.rs index 101cb32e..b649fc70 100644 --- a/crates/code_assistant_core/src/voice/agent.rs +++ b/crates/code_assistant_core/src/voice/agent.rs @@ -17,9 +17,10 @@ use crate::session::event_stream::{EventStream, StreamError, Subscription}; use crate::ui::UiEvent; 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; @@ -42,6 +43,8 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(20); 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, + /// 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, /// 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(); @@ -135,10 +145,11 @@ impl VoiceAgent { ..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) } @@ -207,8 +218,19 @@ impl VoiceAgent { 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) @@ -228,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 => {} } } @@ -346,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, @@ -378,11 +426,11 @@ 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)); } @@ -409,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(|| { @@ -420,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; } } @@ -479,16 +535,34 @@ impl VoiceAgent { } } -/// Open a session and wait until the server confirms it. A server that -/// rejects the session (an invalid API key, an unknown model) still accepts -/// the WebSocket, then sends an `error` event and closes; that is a failed -/// attempt carrying the server's message, not a connection. -async fn connect(connector: &dyn RealtimeConnector) -> Result { +/// 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 { 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)) => return Ok(connection), + 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: {}", @@ -515,6 +589,17 @@ async fn connect(connector: &dyn RealtimeConnector) -> Result 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 { samples * 1000 / SAMPLE_RATE as u64 } diff --git a/crates/code_assistant_core/src/voice/floor.rs b/crates/code_assistant_core/src/voice/floor.rs index b10fcd88..c0d71d46 100644 --- a/crates/code_assistant_core/src/voice/floor.rs +++ b/crates/code_assistant_core/src/voice/floor.rs @@ -41,6 +41,9 @@ pub struct FloorConfig { /// How long after the user stopped speaking the floor stays theirs when /// no response follows. pub user_turn_grace: Duration, + /// How long our `response.create` may go unanswered before the floor + /// is given back (a lost request must not wedge it). + pub response_start_grace: Duration, } impl Default for FloorConfig { @@ -49,6 +52,7 @@ impl Default for FloorConfig { cooling: Duration::from_millis(1500), playback_grace: Duration::from_secs(5), user_turn_grace: Duration::from_secs(8), + response_start_grace: Duration::from_secs(10), } } } @@ -73,6 +77,8 @@ pub enum Timer { Cooling, /// The user stopped speaking but no response followed. UserTurn, + /// Our `response.create` got no `response.created`. + ResponseStart, } #[derive(Debug, Clone, PartialEq)] @@ -93,8 +99,8 @@ pub enum FloorInput { CaptureMuted, /// The speakers played everything queued. PlaybackDrained, - /// The server reported an error. - ServerError, + /// The server refused our `response.create`. + CreateFailed, /// A tool call started running. ToolStarted, /// A tool call finished. @@ -261,16 +267,7 @@ impl Floor { self.playback_finished(&mut out); } } - FloorInput::ServerError => { - if self.create_pending { - self.create_pending = false; - self.response_open = false; - self.cancel_on_create = false; - if self.state == FloorState::Speaking { - self.playback_finished(&mut out); - } - } - } + FloorInput::CreateFailed => self.create_failed(&mut out), FloorInput::ToolStarted => self.tools_running += 1, FloorInput::ToolCompleted { call_id, output } => { self.tools_running = self.tools_running.saturating_sub(1); @@ -328,6 +325,7 @@ impl Floor { self.flush(&mut out); } } + Timer::ResponseStart => self.create_failed(&mut out), Timer::Cooling | Timer::UserTurn => {} } } @@ -335,6 +333,20 @@ impl Floor { out } + /// Our `response.create` was refused or never answered: give the floor + /// back. + fn create_failed(&mut self, out: &mut Vec) { + if !self.create_pending { + return; + } + self.create_pending = false; + self.response_open = false; + self.cancel_on_create = false; + if self.state == FloorState::Speaking { + self.playback_finished(out); + } + } + fn waiting_for_playback(&self) -> bool { self.state == FloorState::Speaking && !self.response_open && !self.create_pending } @@ -375,6 +387,7 @@ impl Floor { self.response_open = true; self.state = FloorState::Speaking; out.push(FloorCommand::CreateResponse); + self.start_timer(Timer::ResponseStart, self.config.response_start_grace, out); } fn start_timer(&mut self, timer: Timer, after: Duration, out: &mut Vec) { @@ -593,7 +606,8 @@ mod tests { output: "out".into() }, C::CancelTimer, - C::CreateResponse + C::CreateResponse, + C::StartTimer(Timer::ResponseStart, Duration::from_secs(10)), ] ); } @@ -612,7 +626,8 @@ mod tests { call_id: "c1".into(), output: "out".into() }, - C::CreateResponse + C::CreateResponse, + C::StartTimer(Timer::ResponseStart, Duration::from_secs(10)), ] ); } @@ -802,13 +817,38 @@ mod tests { fn failed_create_does_not_wedge_the_floor() { let mut floor = floor(); floor.handle(finished("a")); - floor.handle(I::ServerError); + floor.handle(I::CreateFailed); assert_eq!(floor.state(), FloorState::Cooling); floor.handle(I::TimerFired(Timer::Cooling)); assert_eq!(floor.state(), FloorState::Idle); assert!(creates(&floor.handle(finished("b")))); } + #[test] + fn an_unanswered_create_gives_the_floor_back() { + let mut floor = floor(); + let out = floor.handle(finished("a")); + assert!(out.contains(&C::StartTimer( + Timer::ResponseStart, + Duration::from_secs(10) + ))); + assert_eq!(floor.state(), FloorState::Speaking); + floor.handle(I::TimerFired(Timer::ResponseStart)); + assert_eq!(floor.state(), FloorState::Cooling); + floor.handle(I::TimerFired(Timer::Cooling)); + assert!(creates(&floor.handle(finished("b")))); + } + + #[test] + fn a_created_response_disarms_the_create_watchdog() { + let mut floor = floor(); + floor.handle(finished("a")); + floor.handle(I::ResponseCreated); + // A watchdog firing late (its cancel raced it) changes nothing. + assert!(floor.handle(I::TimerFired(Timer::ResponseStart)).is_empty()); + assert_eq!(floor.state(), FloorState::Speaking); + } + #[test] fn reading_a_conversation_drops_its_notification() { let mut floor = floor(); diff --git a/crates/code_assistant_core/src/voice/mod.rs b/crates/code_assistant_core/src/voice/mod.rs index edd785fa..2b79caa7 100644 --- a/crates/code_assistant_core/src/voice/mod.rs +++ b/crates/code_assistant_core/src/voice/mod.rs @@ -87,6 +87,9 @@ pub struct VoiceStatus { pub muted: bool, /// Notifications waiting for a free floor. pub queued_notifications: usize, + /// The latest error the session survived (a refused request, a failed + /// answer); cleared once an answer completes again. + pub error: Option, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -97,6 +100,8 @@ pub enum TranscriptRole { Tool, /// A background notification delivered to the model. Notification, + /// An error the server reported. + Error, } #[derive(Debug, Clone, PartialEq, Eq)] diff --git a/crates/code_assistant_core/src/voice/tests.rs b/crates/code_assistant_core/src/voice/tests.rs index 7a0839ac..88a80798 100644 --- a/crates/code_assistant_core/src/voice/tests.rs +++ b/crates/code_assistant_core/src/voice/tests.rs @@ -145,10 +145,20 @@ struct Harness { sessions: SessionService, voice: VoiceService, server: Server, + /// The servers of the connections that renew the first one, in order. + renewals: Vec, audio: FakeAudio, _dir: tempfile::TempDir, } +/// Like OpenAI, the server announces the session before the client speaks +/// and then accepts its `session.update` (queued up front: the client reads +/// them in order). +fn confirm_session(server: &Server) { + server.send(ServerEvent::SessionCreated); + server.send(ServerEvent::SessionUpdated); +} + /// Never answers, like a connection attempt into a blackholed network. struct HangingConnector; @@ -216,13 +226,28 @@ fn session_service(dir: &tempfile::TempDir) -> (SessionService, EventStream) { /// A running voice session over [`session_service`]. async fn harness() -> Harness { + harness_with_renewals(0).await +} + +/// Like [`harness`], with `renewals` more connections for reconnects. +async fn harness_with_renewals(renewals: usize) -> Harness { let dir = tempfile::tempdir().unwrap(); let (sessions, events) = session_service(&dir); let (connection, mut server) = scripted_connection(); - // Like OpenAI, the server confirms the session before the client speaks. - server.send(ServerEvent::SessionCreated); + confirm_session(&server); + let mut connections = Vec::new(); + let mut renewal_servers = Vec::new(); + for _ in 0..renewals { + let (connection, server) = scripted_connection(); + confirm_session(&server); + connections.push(connection); + renewal_servers.push(server); + } + // The connector hands them out from the back. + connections.reverse(); + connections.push(connection); let connector: Arc = Arc::new(ScriptedConnector { - connections: Mutex::new(vec![connection]), + connections: Mutex::new(connections), }); let audio = FakeAudio::default(); let (voice, voice_worker) = VoiceService::with_config_loader( @@ -243,6 +268,7 @@ async fn harness() -> Harness { sessions, voice, server, + renewals: renewal_servers, audio, _dir: dir, } @@ -253,6 +279,7 @@ fn response_created() -> ServerEvent { response: ResponseInfo { id: "r".into(), status: None, + status_details: None, }, } } @@ -262,6 +289,7 @@ fn response_done() -> ServerEvent { response: ResponseInfo { id: "r".into(), status: Some("completed".into()), + status_details: None, }, } } @@ -587,6 +615,7 @@ fn rejecting_connection() -> RealtimeConnection { error: llm::realtime::ErrorInfo { message: "Incorrect API key provided".into(), code: Some("invalid_api_key".into()), + event_id: None, }, }); server @@ -639,7 +668,7 @@ async fn a_connection_closed_before_the_session_started_fails() { async fn rejected_reconnects_end_voice_mode_instead_of_looping() { let dir = tempfile::tempdir().unwrap(); let (first, server) = scripted_connection(); - server.send(ServerEvent::SessionCreated); + confirm_session(&server); let (voice, mut statuses) = scripted_voice( &dir, vec![ @@ -661,3 +690,108 @@ async fn rejected_reconnects_end_voice_mode_instead_of_looping() { assert!(message.contains("could not be renewed"), "{message}"); assert!(message.contains("Incorrect API key provided"), "{message}"); } + +fn server_error(message: &str, event_id: Option<&str>) -> ServerEvent { + ServerEvent::Error { + error: llm::realtime::ErrorInfo { + message: message.into(), + code: None, + event_id: event_id.map(str::to_string), + }, + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn rejected_voice_settings_fail_instead_of_running_on_defaults() { + let dir = tempfile::tempdir().unwrap(); + let (connection, server) = scripted_connection(); + server.send(ServerEvent::SessionCreated); + server.send(server_error("Invalid value: 'nobody'", None)); + let (voice, mut statuses) = scripted_voice(&dir, vec![connection]); + voice.start(); + let message = wait_for_failure(&mut statuses, WAIT).await; + assert!(message.contains("rejected the voice settings"), "{message}"); + assert!(message.contains("'nobody'"), "{message}"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_survived_error_shows_until_an_answer_completes() { + let h = harness().await; + let mut events = h.sessions.subscribe(); + // Our cancel raced the end of the response: nothing to show. + h.server + .send(server_error("no active response", Some("cancel-7"))); + h.server.send(server_error("Tool schema invalid", None)); + let status = wait_for_status(&mut events, WAIT, |s| s.error.is_some()).await; + assert_eq!(status.error.as_deref(), Some("Tool schema invalid")); + + h.server.send(response_created()); + h.server.send(response_done()); + wait_for_status(&mut events, WAIT, |s| s.error.is_none()).await; + h.voice.stop(); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_failed_answer_shows_its_reason() { + let h = harness().await; + let mut events = h.sessions.subscribe(); + h.server.send(response_created()); + h.server.send(ServerEvent::ResponseDone { + response: ResponseInfo { + id: "r".into(), + status: Some("failed".into()), + status_details: Some(serde_json::json!({ + "type": "failed", + "error": {"type": "server_error", "message": "The server had an error"} + })), + }, + }); + let status = wait_for_status(&mut events, WAIT, |s| s.error.is_some()).await; + assert!( + status.error.unwrap().contains("The server had an error"), + "reason shown" + ); + h.voice.stop(); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_renewed_session_learns_which_answer_the_user_cut_off() { + let mut h = harness_with_renewals(1).await; + h.server.send(response_created()); + h.server.send(audio_delta("speech", 24_000)); + h.server.send(ServerEvent::AudioTranscriptDone { + item_id: "speech".into(), + transcript: "The build failed in two tests".into(), + }); + h.audio + .wait_for("the queued audio", |a| a.queued == 24_000) + .await; + h.audio.play(12_000); + h.server.send(ServerEvent::SpeechStarted); + h.server + .expect("truncate", |e| { + matches!(e, ClientEvent::TruncateItem { .. }) + }) + .await; + + h.server.to_client.send(Incoming::Closed(None)).unwrap(); + let mut renewal = h.renewals.remove(0); + renewal + .expect("session.update", |e| { + matches!(e, ClientEvent::SessionUpdate(_)) + }) + .await; + let ClientEvent::SystemMessage(summary) = renewal + .expect("the conversation so far", |e| { + matches!(e, ClientEvent::SystemMessage(_)) + }) + .await + else { + unreachable!() + }; + assert!( + summary.contains("The build failed in two tests [interrupted by the user]"), + "{summary}" + ); + h.voice.stop(); +} diff --git a/crates/ui_gpui/src/main_screen/voice_control.rs b/crates/ui_gpui/src/main_screen/voice_control.rs index b6d828da..851f1a16 100644 --- a/crates/ui_gpui/src/main_screen/voice_control.rs +++ b/crates/ui_gpui/src/main_screen/voice_control.rs @@ -96,6 +96,15 @@ impl MainScreen { label.to_string() }; let speaking = status.activity == VoiceActivity::Speaking; + let chip_color = if status.error.is_some() { + cx.theme().danger + } else { + cx.theme().muted_foreground + }; + let chip_tooltip = SharedString::from(match &status.error { + Some(error) => format!("{error}\n\nShow the transcript"), + None => "Show the transcript".to_string(), + }); row = row .child( div() @@ -108,7 +117,7 @@ impl MainScreen { .gap_1() .cursor_pointer() .text_xs() - .text_color(cx.theme().muted_foreground) + .text_color(chip_color) .hover(|s| s.bg(cx.theme().muted)) .when(self.voice_transcript_open, |el| el.bg(cx.theme().muted)) .child( @@ -122,7 +131,9 @@ impl MainScreen { }), ) .child(SharedString::from(label)) - .tooltip(|window, cx| Tooltip::new("Show the transcript").build(window, cx)) + .tooltip(move |window, cx| { + Tooltip::new(chip_tooltip.clone()).build(window, cx) + }) .on_click(cx.listener(|this, _, _, cx| { this.voice_transcript_open = !this.voice_transcript_open; cx.notify(); @@ -182,6 +193,7 @@ impl MainScreen { TranscriptRole::Assistant => ("Assistant", cx.theme().primary), TranscriptRole::Tool => ("Tool", cx.theme().muted_foreground), TranscriptRole::Notification => ("Notification", cx.theme().muted_foreground), + TranscriptRole::Error => ("Error", cx.theme().danger), }; // A notification's first lines name the conversations; the // instructions to the model below them are noise here. diff --git a/docs/voice-mode.md b/docs/voice-mode.md index 15174a5f..7ce095ba 100644 --- a/docs/voice-mode.md +++ b/docs/voice-mode.md @@ -219,7 +219,8 @@ sink. | server `response.done` / cancelled | attach held tool outputs; in `Speaking`, wait for the drain (playback timer: pending audio + 5 s), or act as drained if nothing is pending | | server `input_audio_buffer.speech_started` (barge-in) | cancel timers; if `Speaking`: `response.cancel` (deferred until `response.created` when our create is still in flight) and `StopPlayback` (agent: clear the queue, `conversation.item.truncate` to the samples played). Audio deltas play only in `Speaking`, so late audio of the cancelled response is dropped → `UserTurn` | | server `input_audio_buffer.speech_stopped` | stays `UserTurn`; the server's VAD creates the response. If none follows within 8 s, the model answers unanswered tool outputs, or else the floor is free again | -| server `error` | when our `response.create` was in flight: give the floor back (no wedge) | +| server `error` for our `response.create` (`event_id` prefix `create-`) | give the floor back (no wedge) | +| create watchdog (10 s without `response.created`) | same: give the floor back | | sink `Drained` (or fallback timer) | if tool outputs are unanswered and no tool runs any more → `response.create`; otherwise → `Cooling` + start the cooling timer | | cooling timer fires | `Cooling` → `Idle`; flush notifications if any are queued | | tool completed | response still open → hold until `response.done`; otherwise attach, and the output counts as unanswered until the next response starts. `Speaking` → answered after `Drained`; `UserTurn` → **no** `response.create` (the model uses it in its next answer, or answers when the user's turn ends without one); `Cooling`/`Idle` → `response.create` once the last running tool is done, so several calls of one response get one answer | @@ -250,13 +251,25 @@ session) and to fast ones. ## Session lifetime and reconnect -- Start: open the realtime session → `session.update` (instructions, tools, - voice, turn detection, input transcription on) → start capture. +- Start: open the realtime session, wait for `session.created`, send + `session.update` (instructions, tools, voice, turn detection, input + transcription on), wait for `session.updated` → start capture. OpenAI + accepts the WebSocket even with an invalid key and then sends `error` and + closes; an `error` or a close before `session.updated` fails the attempt + with the server's message (a rejected `session.update` would otherwise + leave the session on server defaults, without tools or instructions). +- Errors during a session: client events whose errors are expected carry + an `event_id` prefix (`create-`, `cancel-`, `truncate-`). Errors for a + cancel or a truncate are logged only (the response or item ended + first). Every other error, and a `response.done` with status `failed`, + is shown: as an *Error* line in the transcript and on the status chip + (`VoiceStatus::error`) until a response completes again. - Realtime sessions have a maximum duration and can drop. The agent keeps a local text transcript (user transcripts + assistant transcripts + tool calls). On a closed connection it reconnects (three attempts, backing off), re-sends `session.update`, seeds the last 30 transcript lines as one - system message, and keeps the notification queue (flushed after a cooling + system message (an answer the user barged into is marked as interrupted, + since its transcript says more than they heard), and keeps the notification queue (flushed after a cooling period). Tool results of the old connection are dropped. - Stop: the toggle, or an error that cannot be retried. Capture and playback stop, the socket closes, and the queue is dropped.