Consolidate error classification for realtime WebSocket failures

aai_cli/agent/audio.py

@@ -7,7 +7,7 @@
 from typing import Any
 
 from aai_cli.errors import CLIError
-from aai_cli.microphone import audio_missing_error, default_rate, resample_pcm16
+from aai_cli.microphone import default_rate, import_sounddevice, resample_pcm16
 
 SAMPLE_RATE = 24000  # Voice Agent native PCM16 mono rate
 
@@ -48,10 +48,7 @@ def close(self) -> None:
 
 def _default_duplex_stream(*, rate: int, blocksize: int, callback: Any, device: int | None) -> Any:
     """Open ONE started full-duplex sounddevice stream (mic + speaker together)."""
-    try:
-        import sounddevice as sd
-    except ImportError as exc:
-        raise audio_missing_error() from exc
+    sd = import_sounddevice()
     try:
         stream = sd.RawStream(
             samplerate=rate,

aai_cli/agent/session.py

@@ -211,14 +211,12 @@ def _open_ws(connect: _Connect, api_key: str) -> _WebSocket:
     A rejected handshake (HTTP 401/403) gets the shared actionable suggestion
     (whoami / environment / network); anything else keeps the wsutil mapping.
     """
-    message = "Could not connect to the voice agent"
     try:
         return connect(ws_url(), additional_headers={"Authorization": f"Bearer {api_key}"})
     except Exception as exc:
-        rejected = diagnostics.handshake_error(exc, message, host=environments.active().agents_host)
-        if rejected is not None:
-            raise rejected from exc
-        raise wsutil.auth_or_api_error(exc, message) from exc
+        raise diagnostics.classify_error(
+            exc, "Could not connect to the voice agent", host=environments.active().agents_host
+        ) from exc
 
 
 def _session_update_message(config: AgentRunConfig) -> str:

aai_cli/client.py

@@ -17,7 +17,7 @@
 
 from aai_cli import environments, jsonshape, stdio
 from aai_cli.errors import APIError, CLIError, UsageError, auth_failure, is_auth_failure
-from aai_cli.streaming.diagnostics import handshake_error, silence_streaming_logging
+from aai_cli.streaming.diagnostics import classify_error, silence_streaming_logging
 
 SAMPLE_AUDIO_URL = "https://assembly.ai/wildfires.mp3"
 _StreamHandler = Callable[[Any, Any], object]
@@ -271,12 +271,7 @@ def _streaming_run_error(error: object) -> CLIError:
     A rejected handshake (HTTP 401/403) gets an actionable suggestion: bare
     "Streaming error: WebSocket handshake rejected (HTTP 403)" left users cold.
     """
-    rejected = handshake_error(error, "Streaming error", host=environments.active().streaming_host)
-    if rejected is not None:
-        return rejected
-    if is_auth_failure(error):
-        return auth_failure()
-    return APIError(f"Streaming error: {error}")
+    return classify_error(error, "Streaming error", host=environments.active().streaming_host)
 
 
 def stream_audio(

aai_cli/microphone.py

@@ -3,6 +3,7 @@
 import warnings
 from abc import abstractmethod
 from collections.abc import Callable, Iterable, Iterator, Mapping
+from types import ModuleType
 from typing import Any, Protocol, cast
 
 from aai_cli.errors import CLIError
@@ -53,12 +54,23 @@ def audio_missing_error() -> CLIError:
     )
 
 
-def _sounddevice() -> _SoundDeviceModule:
+def import_sounddevice() -> ModuleType:
+    """Import sounddevice lazily, mapping an ImportError to ``audio_missing_error``.
+
+    The one import-and-fail path for every audio device opener (mic capture,
+    TTS playback, the agent's duplex stream), so a broken sounddevice install
+    yields the same actionable error no matter which command hit it first.
+    """
     try:
-        import sounddevice as module
+        import sounddevice
     except ImportError as exc:
         raise audio_missing_error() from exc
-    return cast("_SoundDeviceModule", module)
+    module: ModuleType = sounddevice
+    return module
+
+
+def _sounddevice() -> _SoundDeviceModule:
+    return cast("_SoundDeviceModule", import_sounddevice())
 
 
 def default_rate(kind: str, device: int | None = None) -> int:

aai_cli/streaming/diagnostics.py

@@ -62,3 +62,18 @@ def handshake_error(error: object, message: str, *, host: str) -> CLIError | Non
             rejected_key=True,
         )
     return APIError(f"{message}: {error}", suggestion=handshake_suggestion(host))
+
+
+def classify_error(error: object, message: str, *, host: str) -> CLIError:
+    """The one CLIError classification for a realtime WebSocket failure.
+
+    Shared by every realtime path (stream, agent, speak) for connect failures and
+    recorded stream Error events alike, so they can't drift: a rejected handshake
+    (HTTP 401/403) carries the actionable suggestion via ``handshake_error``;
+    anything else keeps the ``aai_cli.ws`` mapping — a rejected key becomes
+    ``auth_failure()``, the rest ``APIError(f"{message}: …")``.
+    """
+    rejected = handshake_error(error, message, host=host)
+    if rejected is not None:
+        return rejected
+    return wsutil.auth_or_api_error(error, message)

aai_cli/tts/audio.py

@@ -7,7 +7,7 @@
 from typing import Protocol
 
 from aai_cli.errors import CLIError
-from aai_cli.microphone import audio_missing_error
+from aai_cli.microphone import import_sounddevice
 
 
 class _OutputStream(Protocol):
@@ -53,10 +53,7 @@ def silence(sample_rate: int, seconds: float) -> bytes:
 
 def _default_output_stream(sample_rate: int) -> _OutputStream:
     """A started-on-demand raw 16-bit mono output stream from sounddevice."""
-    try:
-        import sounddevice as sd
-    except ImportError as exc:
-        raise audio_missing_error() from exc
+    sd = import_sounddevice()
     stream: _OutputStream = sd.RawOutputStream(samplerate=sample_rate, channels=1, dtype="int16")
     return stream
 

aai_cli/tts/session.py

@@ -139,20 +139,18 @@ def _open_ws(connect: _Connect, api_key: str, url: str) -> _WebSocket:
     A rejected handshake (HTTP 401/403) gets the shared actionable suggestion
     (whoami / environment / network); anything else keeps the wsutil mapping.
     """
-    message = "Could not connect to the TTS service"
     try:
         return connect(
             url,
             additional_headers={"Authorization": f"Bearer {api_key}"},
             max_size=None,
         )
     except Exception as exc:
-        rejected = diagnostics.handshake_error(
-            exc, message, host=environments.active().streaming_tts_host
-        )
-        if rejected is not None:
-            raise rejected from exc
-        raise wsutil.auth_or_api_error(exc, message) from exc
+        raise diagnostics.classify_error(
+            exc,
+            "Could not connect to the TTS service",
+            host=environments.active().streaming_tts_host,
+        ) from exc
 
 
 def _run_protocol(

aai_cli/ws.py

@@ -53,7 +53,7 @@ def handshake_status(exc: object) -> int | None:
     return None
 
 
-def is_rejected_key(exc: Exception) -> bool:
+def is_rejected_key(exc: object) -> bool:
     """Is this connect/session failure auth-shaped (the key itself was rejected)?
 
     Mirrors how `stream` classifies handshake failures: a plain HTTP 403 on the
@@ -68,7 +68,7 @@ def is_rejected_key(exc: Exception) -> bool:
     return is_auth_failure(exc)
 
 
-def auth_or_api_error(exc: Exception, message: str) -> CLIError:
+def auth_or_api_error(exc: object, message: str) -> CLIError:
     """Map a connect/session exception to the right CLIError: a rejected key becomes
     auth_failure(), anything else becomes APIError(f"{message}: {exc}")."""
     if is_rejected_key(exc):