chore: strengthen CLI quality gates (#17)

* chore: strengthen CLI quality gates

* chore: remove semgrep gate

---------

Co-authored-by: Alex Kroman <alex@assemblyai.com>

Author: alexkroman

.gitignore

@@ -10,6 +10,7 @@ build/
 .coverage
 coverage.xml
 htmlcov/
+mutants/
 
 # Editor/agent local artifacts: keep personal settings local, but track the
 # team-shared bits (.claude/settings.json, agents/, skills/).

.importlinter

@@ -0,0 +1,64 @@
+[importlinter]
+root_package = aai_cli
+include_external_packages = True
+
+[importlinter:contract:1]
+name = Core modules do not import command modules
+type = forbidden
+source_modules =
+    aai_cli.agent
+    aai_cli.auth
+    aai_cli.client
+    aai_cli.code_gen
+    aai_cli.config
+    aai_cli.config_builder
+    aai_cli.context
+    aai_cli.environments
+    aai_cli.errors
+    aai_cli.follow
+    aai_cli.help_panels
+    aai_cli.help_text
+    aai_cli.init
+    aai_cli.llm
+    aai_cli.microphone
+    aai_cli.output
+    aai_cli.render
+    aai_cli.stdio
+    aai_cli.streaming
+    aai_cli.theme
+    aai_cli.transcribe_render
+    aai_cli.youtube
+forbidden_modules =
+    aai_cli.commands
+
+[importlinter:contract:2]
+name = Command modules are independent
+type = independence
+modules =
+    aai_cli.commands.account
+    aai_cli.commands.agent
+    aai_cli.commands.audit
+    aai_cli.commands.claude
+    aai_cli.commands.doctor
+    aai_cli.commands.init
+    aai_cli.commands.keys
+    aai_cli.commands.llm
+    aai_cli.commands.login
+    aai_cli.commands.samples
+    aai_cli.commands.sessions
+    aai_cli.commands.stream
+    aai_cli.commands.transcribe
+    aai_cli.commands.transcripts
+
+[importlinter:contract:3]
+name = Library layers do not depend on Rich rendering
+type = forbidden
+source_modules =
+    aai_cli.client
+    aai_cli.config
+    aai_cli.config_builder
+    aai_cli.environments
+    aai_cli.errors
+    aai_cli.llm
+forbidden_modules =
+    rich

aai_cli/agent/audio.py

@@ -7,7 +7,7 @@
 from typing import Any
 
 from aai_cli.errors import CLIError
-from aai_cli.microphone import _default_rate, _resample, audio_missing_error
+from aai_cli.microphone import audio_missing_error, default_rate, resample_pcm16
 
 SAMPLE_RATE = 24000  # Voice Agent native PCM16 mono rate
 
@@ -19,7 +19,7 @@ def _output_default_rate(device: int | None = None) -> int:
     'paramErr' (-50) from forcing an unsupported one; agent audio (24 kHz) is
     resampled to it. Falls back to a safe default when the device can't be queried.
     """
-    return _default_rate("output", device)
+    return default_rate("output", device)
 
 
 class NullPlayer:
@@ -141,7 +141,7 @@ def start(self) -> None:
     def feed(self, pcm: bytes) -> None:
         """Queue target-rate PCM for playback, resampled to the device rate."""
         if self._device_rate != self._target:
-            pcm, self._out_state = _resample(
+            pcm, self._out_state = resample_pcm16(
                 pcm, self._out_state, src_rate=self._target, dst_rate=self._device_rate
             )
         with self._lock:
@@ -163,7 +163,7 @@ def capture_frames(self) -> Iterator[bytes]:
             if chunk is None:
                 return
             if self._device_rate != self._target:
-                chunk, state = _resample(
+                chunk, state = resample_pcm16(
                     chunk, state, src_rate=self._device_rate, dst_rate=self._target
                 )
             yield chunk

aai_cli/agent/session.py

@@ -71,41 +71,41 @@ def dispatch(self, event: dict[str, Any]) -> None:
         if handler is not None:
             handler(self, event)
 
-    def _on_session_ready(self, _event: dict[str, Any]) -> None:
+    def on_session_ready(self, _event: dict[str, Any]) -> None:
         with self._lock:
             self.ready = True
         if self.ready_event is not None:
             self.ready_event.set()
         self.renderer.connected()
 
-    def _on_speech_started(self, _event: dict[str, Any]) -> None:
+    def on_speech_started(self, _event: dict[str, Any]) -> None:
         if self.full_duplex:
             self.player.flush()
 
-    def _on_user_delta(self, event: dict[str, Any]) -> None:
+    def on_user_delta(self, event: dict[str, Any]) -> None:
         self.renderer.user_partial(event.get("text", ""))
 
-    def _on_user_final(self, event: dict[str, Any]) -> None:
+    def on_user_final(self, event: dict[str, Any]) -> None:
         self._saw_user = True
         self.renderer.user_final(event.get("text", ""))
 
-    def _on_reply_started(self, _event: dict[str, Any]) -> None:
+    def on_reply_started(self, _event: dict[str, Any]) -> None:
         if not self.full_duplex:
             with self._lock:
                 self.muted = True
         self.renderer.reply_started()
 
-    def _on_reply_audio(self, event: dict[str, Any]) -> None:
+    def on_reply_audio(self, event: dict[str, Any]) -> None:
         data = event.get("data")
         if data:
             self.player.enqueue(base64.b64decode(data))
 
-    def _on_agent_transcript(self, event: dict[str, Any]) -> None:
+    def on_agent_transcript(self, event: dict[str, Any]) -> None:
         self.renderer.agent_transcript(
             event.get("text", ""), interrupted=bool(event.get("interrupted", False))
         )
 
-    def _on_reply_done(self, event: dict[str, Any]) -> None:
+    def on_reply_done(self, event: dict[str, Any]) -> None:
         if not self.full_duplex:
             with self._lock:
                 self.muted = False
@@ -117,7 +117,7 @@ def _on_reply_done(self, event: dict[str, Any]) -> None:
         if self.exit_after_reply and self._saw_user and not interrupted:
             self.finished = True
 
-    def _raise_error(self, event: dict[str, Any]) -> None:
+    def raise_error(self, event: dict[str, Any]) -> None:
         code = event.get("code", "")
         message = event.get("message") or code or "Voice agent error."
         if code in _AUTH_ERROR_CODES:
@@ -132,15 +132,15 @@ def _raise_error(self, event: dict[str, Any]) -> None:
 # 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[str, Any]], 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,
+    "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,
 }
 
 

aai_cli/auth/ams.py

@@ -1,35 +1,60 @@
 from __future__ import annotations
 
-from typing import Any, cast
-
 import httpx2 as httpx
+from pydantic import TypeAdapter, ValidationError
 
 from aai_cli.auth import endpoints
 from aai_cli.errors import APIError, NotAuthenticated
 
 _TIMEOUT = 30.0
+_HTTP_ERROR_MIN_STATUS = 400
+_JSON_OBJECT: TypeAdapter[dict[str, object]] = TypeAdapter(dict[str, object])
+_JSON_OBJECTS: TypeAdapter[list[dict[str, object]]] = TypeAdapter(list[dict[str, object]])
 
 
 def _detail(resp: httpx.Response) -> str:
+    fallback = resp.text or f"HTTP {resp.status_code}"
     try:
-        body = resp.json()
-        if isinstance(body, dict) and "detail" in body:
-            return str(body["detail"])
-    except Exception:  # noqa: BLE001,S110 - non-JSON error body; pass is intentional
-        pass
-    return resp.text or f"HTTP {resp.status_code}"
+        body: object = resp.json()
+        mapping = _JSON_OBJECT.validate_python(body)
+        if "detail" in mapping:
+            return str(mapping["detail"])
+    except (TypeError, ValueError, ValidationError):
+        return fallback
+    return fallback
 
 
 def _raise_for_error(resp: httpx.Response) -> None:
     if resp.status_code in (401, 403):
         raise NotAuthenticated(f"AMS rejected the login ({resp.status_code}): {_detail(resp)}")
-    if resp.status_code >= 400:
+    if resp.status_code >= _HTTP_ERROR_MIN_STATUS:
         raise APIError(f"AMS request failed ({resp.status_code}): {_detail(resp)}")
 
 
-def _json_or_raise(resp: httpx.Response) -> Any:
+def _json_or_raise(resp: httpx.Response) -> object:
     _raise_for_error(resp)
-    return resp.json()
+    data: object = resp.json()
+    return data
+
+
+def _json_object_or_raise(resp: httpx.Response) -> dict[str, object]:
+    data = _json_or_raise(resp)
+    try:
+        return _JSON_OBJECT.validate_python(data)
+    except ValidationError as exc:
+        raise APIError(
+            f"AMS request returned unexpected JSON: expected object, got {type(data).__name__}."
+        ) from exc
+
+
+def _json_object_list_or_raise(resp: httpx.Response) -> list[dict[str, object]]:
+    data = _json_or_raise(resp)
+    try:
+        return _JSON_OBJECTS.validate_python(data)
+    except ValidationError as exc:
+        raise APIError(
+            f"AMS request returned unexpected JSON: expected list of objects, got {type(data).__name__}."
+        ) from exc
 
 
 def _client(session_jwt: str | None = None) -> httpx.Client:
@@ -38,17 +63,17 @@ def _client(session_jwt: str | None = None) -> httpx.Client:
     return httpx.Client(base_url=endpoints.ams_base(), timeout=_TIMEOUT, cookies=cookies)
 
 
-def discover(token: str) -> dict[str, Any]:
+def discover(token: str) -> dict[str, object]:
     """POST /v2/auth/discover with a discovery_oauth token -> {orgs, email, IST}."""
     with _client() as client:
         resp = client.post(
             "/v2/auth/discover",
             json={"token": token, "token_type": "discovery_oauth"},
         )
-    return cast(dict[str, Any], _json_or_raise(resp))
+    return _json_object_or_raise(resp)
 
 
-def exchange(intermediate_session_token: str, organization_id: str) -> dict[str, Any]:
+def exchange(intermediate_session_token: str, organization_id: str) -> dict[str, object]:
     """POST /v2/auth/exchange -> SignedInResponse {account, session_jwt, session_token}."""
     with _client() as client:
         resp = client.post(
@@ -58,55 +83,55 @@ def exchange(intermediate_session_token: str, organization_id: str) -> dict[str,
                 "organization_id": organization_id,
             },
         )
-    return cast(dict[str, Any], _json_or_raise(resp))
+    return _json_object_or_raise(resp)
 
 
-def list_projects(account_id: int, session_jwt: str) -> list[dict[str, Any]]:
+def list_projects(account_id: int, session_jwt: str) -> list[dict[str, object]]:
     """GET /v1/users/accounts/{id}/projects -> [{project, tokens[]}]."""
     with _client(session_jwt) as client:
         resp = client.get(f"/v1/users/accounts/{account_id}/projects")
-    return cast(list[dict[str, Any]], _json_or_raise(resp))
+    return _json_object_list_or_raise(resp)
 
 
 def create_token(
     account_id: int, project_id: int, token_name: str, session_jwt: str
-) -> dict[str, Any]:
+) -> dict[str, object]:
     """POST /v1/users/accounts/{id}/tokens -> TokenSchema incl. `api_key`."""
     with _client(session_jwt) as client:
         resp = client.post(
             f"/v1/users/accounts/{account_id}/tokens",
             json={"project_id": project_id, "token_name": token_name},
         )
-    return cast(dict[str, Any], _json_or_raise(resp))
+    return _json_object_or_raise(resp)
 
 
-def get_balance(session_jwt: str) -> dict[str, Any]:
+def get_balance(session_jwt: str) -> dict[str, object]:
     """GET /v2/billing/balance -> {account_id, balance_in_cents, ...}."""
     with _client(session_jwt) as client:
         resp = client.get("/v2/billing/balance")
-    return cast(dict[str, Any], _json_or_raise(resp))
+    return _json_object_or_raise(resp)
 
 
 def get_usage(
     session_jwt: str,
     starting_on: str,
     ending_before: str,
     window_size: str | None = None,
-) -> dict[str, Any]:
+) -> dict[str, object]:
     """POST /v2/billing/usage (ISO dates) -> {usage_items: [...]}."""
-    body: dict[str, Any] = {"starting_on": starting_on, "ending_before": ending_before}
+    body: dict[str, object] = {"starting_on": starting_on, "ending_before": ending_before}
     if window_size:
         body["window_size"] = window_size
     with _client(session_jwt) as client:
         resp = client.post("/v2/billing/usage", json=body)
-    return cast(dict[str, Any], _json_or_raise(resp))
+    return _json_object_or_raise(resp)
 
 
-def get_rate_limits(account_id: int, session_jwt: str) -> dict[str, Any]:
+def get_rate_limits(account_id: int, session_jwt: str) -> dict[str, object]:
     """GET /v1/users/accounts/{id}/rate-limits -> {rate_limits: [...]}."""
     with _client(session_jwt) as client:
         resp = client.get(f"/v1/users/accounts/{account_id}/rate-limits")
-    return cast(dict[str, Any], _json_or_raise(resp))
+    return _json_object_or_raise(resp)
 
 
 def rename_token(account_id: int, token_id: int, token_name: str, session_jwt: str) -> None:
@@ -119,24 +144,28 @@ def rename_token(account_id: int, token_id: int, token_name: str, session_jwt: s
     _raise_for_error(resp)
 
 
-def list_streaming(session_jwt: str, **filters: Any) -> dict[str, Any]:
+def list_streaming(session_jwt: str, **filters: object) -> dict[str, object]:
     """GET /v1/users/streaming -> {page_details, data: [StreamingSessionSchema]}."""
-    params = {k: v for k, v in filters.items() if v is not None}
+    params = {
+        key: value for key, value in filters.items() if isinstance(value, str | int | float | bool)
+    }
     with _client(session_jwt) as client:
         resp = client.get("/v1/users/streaming", params=params)
-    return cast(dict[str, Any], _json_or_raise(resp))
+    return _json_object_or_raise(resp)
 
 
-def get_streaming(session_id: str, session_jwt: str) -> dict[str, Any]:
+def get_streaming(session_id: str, session_jwt: str) -> dict[str, object]:
     """GET /v1/users/streaming/{session_id} -> StreamingSessionSchema."""
     with _client(session_jwt) as client:
         resp = client.get(f"/v1/users/streaming/{session_id}")
-    return cast(dict[str, Any], _json_or_raise(resp))
+    return _json_object_or_raise(resp)
 
 
-def list_audit_logs(session_jwt: str, **filters: Any) -> dict[str, Any]:
+def list_audit_logs(session_jwt: str, **filters: object) -> dict[str, object]:
     """GET /v2/user/audit-logs -> {page_details, data: [AuditLogResponse]}."""
-    params = {k: v for k, v in filters.items() if v is not None}
+    params = {
+        key: value for key, value in filters.items() if isinstance(value, str | int | float | bool)
+    }
     with _client(session_jwt) as client:
         resp = client.get("/v2/user/audit-logs", params=params)
-    return cast(dict[str, Any], _json_or_raise(resp))
+    return _json_object_or_raise(resp)