Refactor test modules and improve CLI error handling

.gitignore

@@ -15,6 +15,8 @@ htmlcov/
 # Editor/agent local artifacts: keep personal settings local, but track the
 # team-shared bits (.claude/settings.json, agents/, skills/).
 .claude/settings.local.json
+# Temporary git worktrees created for isolated subagents
+.claude/worktrees/
 
 # Local scratch scripts (often contain live keys)
 transcribe/

aai_cli/agent/session.py

@@ -11,6 +11,7 @@
 from aai_cli import environments
 from aai_cli import ws as wsutil
 from aai_cli.errors import APIError, CLIError, NotAuthenticated
+from aai_cli.streaming import diagnostics
 
 
 def ws_url() -> str:
@@ -205,11 +206,19 @@ def _send_audio_loop(ws: _WebSocket, session: VoiceAgentSession, mic: _IO) -> No
 
 
 def _open_ws(connect: _Connect, api_key: str) -> _WebSocket:
-    """Open the Voice Agent socket, mapping a connect failure to a clean CLIError."""
+    """Open the Voice Agent socket, mapping a connect failure to a clean CLIError.
+
+    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:
-        raise wsutil.auth_or_api_error(exc, "Could not connect to the voice agent") from 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
 
 
 def _session_update_message(config: AgentRunConfig) -> str:

aai_cli/agent/voices.py

@@ -1,54 +1,74 @@
 from __future__ import annotations
 
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class Voice:
+    """A known Voice Agent voice id and the language group it belongs to."""
+
+    name: str
+    language: str
+
+
+ENGLISH = "English"
+MULTILINGUAL = "Multilingual"
+
 # Known Voice Agent voice IDs (from the Voice Agent quickstart). The server is
 # the source of truth; this list backs --list-voices and catches obvious typos.
-VOICES: list[str] = [
-    # English
-    "ivy",
-    "james",
-    "tyler",
-    "winter",
-    "sam",
-    "mia",
-    "bella",
-    "david",
-    "jack",
-    "kyle",
-    "helen",
-    "martha",
-    "river",
-    "emma",
-    "victor",
-    "eleanor",
-    "sophie",
-    "oliver",
-    # Multilingual
-    "arjun",
-    "ethan",
-    "dmitri",
-    "lukas",
-    "lena",
-    "pierre",
-    "mina",
-    "ren",
-    "mei",
-    "joon",
-    "giulia",
-    "luca",
-    "lucia",
-    "hana",
-    "mateo",
-    "diego",
+VOICES: list[Voice] = [
+    Voice("ivy", ENGLISH),
+    Voice("james", ENGLISH),
+    Voice("tyler", ENGLISH),
+    Voice("winter", ENGLISH),
+    Voice("sam", ENGLISH),
+    Voice("mia", ENGLISH),
+    Voice("bella", ENGLISH),
+    Voice("david", ENGLISH),
+    Voice("jack", ENGLISH),
+    Voice("kyle", ENGLISH),
+    Voice("helen", ENGLISH),
+    Voice("martha", ENGLISH),
+    Voice("river", ENGLISH),
+    Voice("emma", ENGLISH),
+    Voice("victor", ENGLISH),
+    Voice("eleanor", ENGLISH),
+    Voice("sophie", ENGLISH),
+    Voice("oliver", ENGLISH),
+    Voice("arjun", MULTILINGUAL),
+    Voice("ethan", MULTILINGUAL),
+    Voice("dmitri", MULTILINGUAL),
+    Voice("lukas", MULTILINGUAL),
+    Voice("lena", MULTILINGUAL),
+    Voice("pierre", MULTILINGUAL),
+    Voice("mina", MULTILINGUAL),
+    Voice("ren", MULTILINGUAL),
+    Voice("mei", MULTILINGUAL),
+    Voice("joon", MULTILINGUAL),
+    Voice("giulia", MULTILINGUAL),
+    Voice("luca", MULTILINGUAL),
+    Voice("lucia", MULTILINGUAL),
+    Voice("hana", MULTILINGUAL),
+    Voice("mateo", MULTILINGUAL),
+    Voice("diego", MULTILINGUAL),
 ]
 
+# The plain ids, for membership checks and completion.
+VOICE_NAMES: list[str] = [voice.name for voice in VOICES]
+
 DEFAULT_VOICE = "ivy"
 
 
 def format_voice_list() -> str:
-    """Human-readable, newline-separated voice IDs for --list-voices."""
-    return "\n".join(VOICES)
+    """Human-readable voice IDs for --list-voices, grouped by language."""
+    groups = dict.fromkeys(voice.language for voice in VOICES)
+    blocks: list[str] = []
+    for language in groups:
+        names = "\n".join(f"  {voice.name}" for voice in VOICES if voice.language == language)
+        blocks.append(f"{language}:\n{names}")
+    return "\n\n".join(blocks)
 
 
 def complete_voice(incomplete: str) -> list[str]:
     """Shell-completion callback for ``--voice``: known voice ids matching the prefix."""
-    return [v for v in VOICES if v.startswith(incomplete)]
+    return [name for name in VOICE_NAMES if name.startswith(incomplete)]

aai_cli/auth/flow.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import json
+import sys
 import webbrowser
 from dataclasses import dataclass
 
@@ -84,16 +86,42 @@ def _parse[T](adapter: TypeAdapter[T], data: object) -> T:
         ) from exc
 
 
-def _open_browser(url: str) -> None:
+def _note(*, json_mode: bool, human: str, hint: str, url: str | None = None) -> None:
+    """One stderr progress note for the login flow.
+
+    Humans get Rich prose; ``--json`` mode gets one ``{"hint": …}`` object per note,
+    keeping stderr machine-readable (the same contract as ``output.emit_warning``).
+    """
+    if json_mode:
+        payload: dict[str, object] = {"hint": hint}
+        if url is not None:
+            payload["url"] = url
+        sys.stderr.write(json.dumps(payload) + "\n")
+    else:
+        output.error_console.print(human)
+
+
+def _open_browser(url: str, *, json_mode: bool) -> None:
     """Open the system browser, falling back to printing the URL."""
-    output.error_console.print(
-        f"Opening your browser to sign in:\n  [aai.url]{escape(url)}[/aai.url]"
+    _note(
+        json_mode=json_mode,
+        human=f"Opening your browser to sign in:\n  [aai.url]{escape(url)}[/aai.url]",
+        hint="Opening your browser to sign in.",
+        url=url,
     )
     try:
-        webbrowser.open(url)
+        opened = webbrowser.open(url)
     except Exception:  # noqa: BLE001 - opening a browser is best-effort
-        output.error_console.print(
-            "[aai.muted]Could not open a browser; open the URL above manually.[/aai.muted]"
+        opened = False
+    # webbrowser.open returns False — without raising — on headless boxes with no
+    # usable browser, so the fallback must fire on the boolean too; otherwise the
+    # user sits out the 120s timeout with no hint that nothing opened.
+    if not opened:
+        _note(
+            json_mode=json_mode,
+            human="[aai.muted]Could not open a browser; open the URL above manually.[/aai.muted]",
+            hint="Could not open a browser; open the URL manually.",
+            url=url,
         )
 
 
@@ -135,16 +163,23 @@ def find_or_create_cli_key(account_id: int, session_jwt: str) -> str:
     return _parse(_CREATED_TOKEN, created).api_key
 
 
-def run_login_flow() -> LoginResult:
+def run_login_flow(*, json_mode: bool = False) -> LoginResult:
     """Drive the full browser + AMS login and return a LoginResult."""
     # Bind the loopback callback server *before* opening the browser: if the port is
     # taken, fail cleanly now instead of stranding the user mid-OAuth in a flow that
     # can never call back.
     capture = _start_capture()
-    _open_browser(discovery.build_start_url())
-    output.error_console.print(
-        "[aai.muted]Waiting up to 2 minutes for you to finish signing in…[/aai.muted]\n"
-        "[aai.muted]No browser here? Run 'assembly login --api-key <KEY>' instead.[/aai.muted]"
+    _open_browser(discovery.build_start_url(), json_mode=json_mode)
+    _note(
+        json_mode=json_mode,
+        human=(
+            "[aai.muted]Waiting up to 2 minutes for you to finish signing in…[/aai.muted]\n"
+            "[aai.muted]No browser here? Run 'assembly login --api-key <KEY>' instead.[/aai.muted]"
+        ),
+        hint=(
+            "Waiting up to 2 minutes for you to finish signing in. "
+            "No browser here? Run 'assembly login --api-key <KEY>' instead."
+        ),
     )
     result = capture.wait()
 
@@ -167,10 +202,11 @@ def run_login_flow() -> LoginResult:
         )
     org = disc.organizations[0]
     if len(disc.organizations) > 1:
-        output.error_console.print(
-            f"[aai.muted]Found {len(disc.organizations)} organizations; signing in to "
-            f"'{org.organization_name or org.organization_id}'.[/aai.muted]"
+        chosen = (
+            f"Found {len(disc.organizations)} organizations; signing in to "
+            f"'{org.organization_name or org.organization_id}'."
         )
+        _note(json_mode=json_mode, human=f"[aai.muted]{chosen}[/aai.muted]", hint=chosen)
 
     # `exchange` already returns the signed-in account, so read the id from it
     # rather than making a second GET /v1/auth round-trip.

aai_cli/client.py

@@ -17,6 +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
 
 SAMPLE_AUDIO_URL = "https://assembly.ai/wildfires.mp3"
 _StreamHandler = Callable[[Any, Any], object]
@@ -104,7 +105,14 @@ def _sdk_errors(message: str) -> Generator[None]:
     except Exception as exc:
         if is_auth_failure(exc):
             raise auth_failure() from exc
-        raise APIError(f"{message}: {exc}") from exc
+        # Compact the reason (httpx-backed SDK errors embed a multi-line
+        # `Request: <…>` repr) and drop the SDK's own "failed to …" preamble when
+        # `message` already says the same thing, so the error reads once, cleanly.
+        reason = _SDK_PREAMBLE_RE.sub("", _compact_reason(exc))
+        raise APIError(
+            f"{message}: {reason}",
+            suggestion="Check your network and try again.",
+        ) from exc
 
 
 def _list_transcript_params(limit: int) -> aai.ListTranscriptParameters:
@@ -124,6 +132,13 @@ def _list_transcript_params(limit: int) -> aai.ListTranscriptParameters:
 # httpx-backed SDK errors embed a multi-line repr ("…\nReason: …\nRequest: <Request(…)>").
 _REQUEST_REPR_RE = re.compile(r"Request: <[^>]*>")
 
+# The SDK prefixes transcript-history errors with its own "failed to retrieve
+# transcript(s) …:" preamble, doubling up with the wrapper's message ("Could not
+# list transcripts: failed to retrieve transcripts: …"). Strip just that known
+# preamble; transcript ids are url-safe tokens, so the optional id segment can
+# never swallow a colon-bearing reason (e.g. a URL).
+_SDK_PREAMBLE_RE = re.compile(r"^failed to retrieve transcripts?(?: [A-Za-z0-9_-]+)?: ")
+
 
 def _compact_reason(exc: object) -> str:
     """``str(exc)`` as a single clean line: drop the trailing ``Request: <…>`` repr and
@@ -246,6 +261,20 @@ def get_transcript(api_key: str, transcript_id: str) -> aai.Transcript:
         return aai.Transcript.get_by_id(transcript_id)
 
 
+def _streaming_run_error(error: object) -> CLIError:
+    """Classify a recorded streaming Error event into the CLIError to raise.
+
+    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}")
+
+
 def stream_audio(
     api_key: str,
     source: Iterable[bytes],
@@ -260,6 +289,9 @@ def stream_audio(
     Forwards Begin/Turn/Termination events to the callbacks; raises APIError on a stream error.
     `params` is a fully-built StreamingParameters (sample_rate/speech_model/etc).
     """
+    # The SDK's reader thread and websockets both log connection failures at ERROR;
+    # with no logging configured those lines would duplicate the CLIError on stderr.
+    silence_streaming_logging()
     sc = _make_streaming_client(api_key)
 
     def _guard(cb: Callable[[Any], Any]) -> _StreamHandler:
@@ -299,6 +331,4 @@ def _record_error(_client: object, error: object) -> None:
         sc.disconnect(terminate=True)
 
     if errors:
-        if is_auth_failure(errors[0]):
-            raise auth_failure()
-        raise APIError(f"Streaming error: {errors[0]}")
+        raise _streaming_run_error(errors[0])