refactor: centralize resolution, unify auth detection, harden agent/config

.gitignore

@@ -8,6 +8,7 @@ build/
 .ruff_cache/
 .mypy_cache/
 .coverage
+coverage.xml
 htmlcov/
 
 # Editor/agent local artifacts: keep personal settings local, but track the

aai_cli/agent/audio.py

@@ -98,9 +98,14 @@ def __init__(
         self._device = device
         self._factory = stream_factory or _default_duplex_stream
         self._blocksize = max(1, self._device_rate // 10)  # ~100 ms
+        # Thread ownership: `_in` is a thread-safe Queue handed from the PortAudio
+        # callback thread to capture_frames(). `_out` (device-rate playback bytes) is
+        # shared between feed()/flush() (caller thread) and the callback, so every
+        # access goes through `_lock`. `_out_state` (the target->device ratecv state)
+        # is touched ONLY by feed(), never the callback, so it needs no lock.
         self._in: queue.Queue[bytes | None] = queue.Queue()
-        self._out = bytearray()  # device-rate playback bytes
-        self._out_state: Any = None  # ratecv state for target -> device
+        self._out = bytearray()
+        self._out_state: Any = None
         self._lock = threading.Lock()
         self._stream: Any = None
         self._started = False

aai_cli/agent/session.py

@@ -4,6 +4,7 @@
 import contextlib
 import json
 import threading
+from collections.abc import Callable
 from typing import Any
 
 from aai_cli.errors import APIError, CLIError, auth_failure, is_auth_failure
@@ -40,58 +41,76 @@ def __init__(
         # to the spoken input, so the process exits instead of waiting forever.
         self.exit_after_reply = exit_after_reply
         self.ready_event = ready_event
+        # `ready`/`muted` are the duplex gate: written by dispatch() on the receive
+        # loop, read by should_send_audio() on the capture thread. The lock makes that
+        # cross-thread handoff explicit (and the reads consistent) instead of relying on
+        # the GIL. `finished`/`_saw_user` are touched only by dispatch (one thread).
+        self._lock = threading.Lock()
         self.ready = False
         self.muted = False
         self.finished = False
         self._saw_user = False
 
     def should_send_audio(self) -> bool:
         """True when captured mic frames should be forwarded to the server."""
-        return self.ready and not self.muted
+        with self._lock:
+            return self.ready and not self.muted
 
     def dispatch(self, event: dict) -> None:
-        etype = event.get("type")
+        """Route one server event to its handler; unknown types are ignored.
 
-        if etype == "session.ready":
+        Handlers are registered in ``_EVENT_HANDLERS`` (below the class). Events with
+        no handler — ``input.speech.stopped``, ``tool.call``, anything new — are no-ops.
+        """
+        handler = _EVENT_HANDLERS.get(event.get("type", ""))
+        if handler is not None:
+            handler(self, event)
+
+    def _on_session_ready(self, _event: dict) -> None:
+        with self._lock:
             self.ready = True
-            if self.ready_event is not None:
-                self.ready_event.set()
-            self.renderer.connected()
-        elif etype == "input.speech.started":
-            if self.full_duplex:
-                self.player.flush()
-        elif etype == "input.speech.stopped":
-            pass
-        elif etype == "transcript.user.delta":
-            self.renderer.user_partial(event.get("text", ""))
-        elif etype == "transcript.user":
-            self._saw_user = True
-            self.renderer.user_final(event.get("text", ""))
-        elif etype == "reply.started":
-            if not self.full_duplex:
+        if self.ready_event is not None:
+            self.ready_event.set()
+        self.renderer.connected()
+
+    def _on_speech_started(self, _event: dict) -> None:
+        if self.full_duplex:
+            self.player.flush()
+
+    def _on_user_delta(self, event: dict) -> None:
+        self.renderer.user_partial(event.get("text", ""))
+
+    def _on_user_final(self, event: dict) -> None:
+        self._saw_user = True
+        self.renderer.user_final(event.get("text", ""))
+
+    def _on_reply_started(self, _event: dict) -> None:
+        if not self.full_duplex:
+            with self._lock:
                 self.muted = True
-            self.renderer.reply_started()
-        elif etype == "reply.audio":
-            data = event.get("data")
-            if data:
-                self.player.enqueue(base64.b64decode(data))
-        elif etype == "transcript.agent":
-            self.renderer.agent_transcript(
-                event.get("text", ""), interrupted=bool(event.get("interrupted", False))
-            )
-        elif etype == "reply.done":
-            if not self.full_duplex:
+        self.renderer.reply_started()
+
+    def _on_reply_audio(self, event: dict) -> None:
+        data = event.get("data")
+        if data:
+            self.player.enqueue(base64.b64decode(data))
+
+    def _on_agent_transcript(self, event: dict) -> None:
+        self.renderer.agent_transcript(
+            event.get("text", ""), interrupted=bool(event.get("interrupted", False))
+        )
+
+    def _on_reply_done(self, event: dict) -> None:
+        if not self.full_duplex:
+            with self._lock:
                 self.muted = False
-            interrupted = event.get("status") == "interrupted"
-            if interrupted:
-                self.player.flush()
-            self.renderer.reply_done(interrupted=interrupted)
-            # File-driven run: the agent has answered the spoken input, so stop.
-            if self.exit_after_reply and self._saw_user and not interrupted:
-                self.finished = True
-        elif etype == "session.error":
-            self._raise_error(event)
-        # tool.call and unknown event types: intentionally ignored.
+        interrupted = event.get("status") == "interrupted"
+        if interrupted:
+            self.player.flush()
+        self.renderer.reply_done(interrupted=interrupted)
+        # File-driven run: the agent has answered the spoken input, so stop.
+        if self.exit_after_reply and self._saw_user and not interrupted:
+            self.finished = True
 
     def _raise_error(self, event: dict) -> None:
         code = event.get("code", "")
@@ -105,6 +124,21 @@ def _raise_error(self, event: dict) -> None:
         raise APIError(f"Voice agent error ({code}): {message}")
 
 
+# Server event type -> the VoiceAgentSession method that handles it. Types absent
+# here (input.speech.stopped, tool.call, anything unrecognized) are ignored.
+_EVENT_HANDLERS: dict[str, Callable[[VoiceAgentSession, dict], None]] = {
+    "session.ready": VoiceAgentSession._on_session_ready,
+    "input.speech.started": VoiceAgentSession._on_speech_started,
+    "transcript.user.delta": VoiceAgentSession._on_user_delta,
+    "transcript.user": VoiceAgentSession._on_user_final,
+    "reply.started": VoiceAgentSession._on_reply_started,
+    "reply.audio": VoiceAgentSession._on_reply_audio,
+    "transcript.agent": VoiceAgentSession._on_agent_transcript,
+    "reply.done": VoiceAgentSession._on_reply_done,
+    "session.error": VoiceAgentSession._raise_error,
+}
+
+
 def _send_audio_loop(ws: Any, session: VoiceAgentSession, mic: Any) -> None:
     """Forward mic PCM as input.audio while the session gate allows it."""
     # File-driven runs wait for session.ready before consuming the source, so a
@@ -121,22 +155,20 @@ def _send_audio_loop(ws: Any, session: VoiceAgentSession, mic: Any) -> None:
             return
 
 
-def _is_auth_rejection(exc: BaseException) -> bool:
-    """True when a connect/session failure means the credentials were rejected.
+def _auth_or_api_error(exc: Exception, 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_auth_failure(exc):
+        return auth_failure()
+    return APIError(f"{message}: {exc}")
 
-    Detected structurally where possible — the Voice Agent closes with WebSocket
-    code 1008 on a bad key, and a pre-upgrade HTTP 401/403 carries a status code —
-    then falls back to the shared text heuristic.
-    """
-    code = getattr(exc, "code", None)
-    if code is None:
-        code = getattr(getattr(exc, "rcvd", None), "code", None)
-    if code == 1008:
-        return True
-    response = getattr(exc, "response", None)
-    if getattr(response, "status_code", None) in (401, 403):
-        return True
-    return is_auth_failure(exc)
+
+def _open_ws(connect: Any, api_key: str) -> Any:
+    """Open the Voice Agent socket, mapping a connect failure to a clean CLIError."""
+    try:
+        return connect(WS_URL, additional_headers={"Authorization": f"Bearer {api_key}"})
+    except Exception as exc:
+        raise _auth_or_api_error(exc, "Could not connect to the voice agent") from exc
 
 
 def run_session(
@@ -171,16 +203,14 @@ def run_session(
         ready_event=ready_event,
     )
 
-    try:
-        ws = connect(WS_URL, additional_headers={"Authorization": f"Bearer {api_key}"})
-    except Exception as exc:
-        if _is_auth_rejection(exc):
-            raise auth_failure() from exc
-        raise APIError(f"Could not connect to the voice agent: {exc}") from exc
+    ws = _open_ws(connect, api_key)
 
     # The mic opens lazily on first iteration, inside the capture thread; a failure
     # there (no device, sounddevice missing) must reach the user instead of vanishing
     # with the daemon thread. Capture it and close the socket to end the receive loop.
+    # The append below happens-before the main thread reads `capture_error`: the
+    # capture thread appends, then ws.close() ends the `for raw in ws` loop, so the
+    # error is always visible by the time we inspect it after the loop.
     capture_error: list[CLIError] = []
 
     def _capture() -> None:
@@ -217,9 +247,7 @@ def _capture() -> None:
     except Exception as exc:
         if capture_error:
             raise capture_error[0] from exc  # a mic-open failure is the real cause
-        if _is_auth_rejection(exc):
-            raise auth_failure() from exc
-        raise APIError(f"Voice agent session failed: {exc}") from exc
+        raise _auth_or_api_error(exc, "Voice agent session failed") from exc
     finally:
         with contextlib.suppress(Exception):
             ws.close()

aai_cli/auth/loopback.py

@@ -63,6 +63,7 @@ def log_message(self, *args: object) -> None:  # silence stderr logging
         if not done.wait(timeout):
             result.error = "timeout"
     finally:
-        server.shutdown()
+        server.shutdown()  # stop serve_forever()
         thread.join(timeout=5)
+        server.server_close()  # close the listening socket (shutdown() leaves it open)
     return result