Robustness, environment-aware code gen, and config caching

.claude/settings.json

@@ -95,7 +95,7 @@
         "hooks": [
           {
             "type": "command",
-            "command": "f=$(jq -r '.tool_input.file_path // empty'); case \"$f\" in *.py) uv run ruff check --fix \"$f\" >/dev/null 2>&1; uv run ruff format \"$f\" >/dev/null 2>&1;; esac; exit 0"
+            "command": "f=$(jq -r '.tool_input.file_path // empty'); case \"$f\" in *.py) uv run ruff check --fix --unfixable F401 \"$f\" >/dev/null 2>&1; uv run ruff format \"$f\" >/dev/null 2>&1;; esac; exit 0"
           }
         ]
       }

.importlinter

@@ -21,6 +21,7 @@ source_modules =
     aai_cli.init
     aai_cli.llm
     aai_cli.microphone
+    aai_cli.options
     aai_cli.output
     aai_cli.render
     aai_cli.stdio

AGENTS.md

@@ -14,7 +14,7 @@ uv run aai --help            # run the CLI from the locked environment
 ./scripts/check.sh           # the full gate CI runs (scripts/check.sh is the source of truth)
 ```
 
-`scripts/check.sh` is the authoritative gate; keep this list in sync with it. It runs, in order: `uv lock --check` → `ruff check` → `ruff format --check` → `mypy` → `pyright` (src strict) → `pyright` (tests) → `vulture` (dead code) → `deptry` (dependency hygiene) → `lint-imports` (import-linter architecture contracts) → `xenon` (cyclomatic complexity, max grade B / project avg A) → `swiftlint` + swift compile (macOS only, skipped elsewhere) → `markdownlint` → `prettier` (init template JS/CSS) → `shellcheck` → generated `--show-code` compile gate → init template contract gate → `pytest` (90% branch coverage) → `diff-cover` (100% patch coverage vs `origin/main`) → a "no new escape hatches" diff gate (`# type: ignore` / `# noqa` / `pragma: no cover` / net-new `Any` / `cast(`) → `uv build` + `twine check --strict`. The `vulture`/`deptry`/`lint-imports`/`xenon` and patch-coverage stages catch the failures that `ruff`+`mypy` alone won't — don't claim the gate is green until the script prints `All checks passed.`
+`scripts/check.sh` is the authoritative gate; keep this list in sync with it. It runs, in order: `uv lock --check` → `ruff check` → `ruff format --check` → `mypy` → `pyright` (src strict) → `pyright` (tests) → `vulture` (dead code) → `deptry` (dependency hygiene) → `lint-imports` (import-linter architecture contracts) → max-file-length (500 lines) → `xenon` (cyclomatic complexity, max grade B / project avg A) → `swiftlint` + swift compile (macOS only, skipped elsewhere) → `markdownlint` → `prettier` (init template JS/CSS) → `shellcheck` → `actionlint` + `zizmor` (workflow lint/audit) → `gitleaks` (secret scan) → generated `--show-code` compile gate → init template contract gate → `pytest` (90% branch coverage) → `diff-cover` (100% patch coverage vs `origin/main`) → **mutation gate** (diff-scoped: mutates each changed line and reruns the tests that cover it — a surviving mutant fails the gate, so changed lines need assertions that would *fail* if the line broke, not just coverage; suppress a genuinely unassertable line with `# pragma: no mutate`) → a "no new escape hatches" diff gate (`# type: ignore` / `# noqa` / `pragma: no cover` / net-new `Any` / `cast(`) → `uv build` + `twine check --strict`. The `vulture`/`deptry`/`lint-imports`/`xenon`, patch-coverage, and mutation stages catch the failures that `ruff`+`mypy` alone won't — don't claim the gate is green until the script prints `All checks passed.`
 
 Individual tools (all via `uv run`):
 
@@ -41,6 +41,8 @@ uv run pytest -m install_script  # builds a wheel and runs install.sh for real;
 
 CLI output is pinned by **syrupy snapshot tests** (`tests/__snapshots__/*.ambr`). Changing help text, tables, or rendered output will fail those tests until you regenerate them with `uv run pytest --snapshot-update` and commit the updated `.ambr` files. The auto-format hook only touches `*.py`, and pre-commit's whitespace fixers deliberately skip `tests/__snapshots__/` (syrupy's indentation must stay byte-for-byte), so never hand-edit a snapshot — always regenerate.
 
+The post-edit hook (`.claude/settings.json`) runs `ruff check --fix --unfixable F401` + `ruff format` on every edited `*.py`. `--unfixable F401` means a just-added import is **not** auto-deleted while it's momentarily unused — so adding an import in one edit and its usage in the next is safe. The flip side: a genuinely unused import survives the hook and only fails at `ruff check` in the gate, so still prefer making the import and its first usage land in the same edit.
+
 The suite is hermetic by construction, enforced three ways (`tests/conftest.py` + `pyproject.toml` `[tool.pytest.ini_options]`): **pytest-randomly** shuffles order, an autouse `pin_timezone` fixture pins `TZ` to a fixed non-UTC zone (UTC-normalized rendering must be unaffected; use **time-machine** to freeze `now`), and **pytest-socket** (`--disable-socket`) blocks real network so an unmocked SDK/HTTP call fails loudly instead of hitting the API. A test that only binds a loopback server opts back in with the tight `@pytest.mark.allow_hosts(["127.0.0.1"])` (still blocks external hosts). The `e2e`/`install`/`install_script` marker suites legitimately reach the real network in-process (PyPI reachability probes, real-API runs), so a `pytest_collection_modifyitems` hook in `conftest.py` auto-grants them full sockets — adding a network marker is all that's needed, no per-test `enable_socket`.
 
 ## Naming & packaging gotchas

aai_cli/agent/audio.py

@@ -91,6 +91,7 @@ def __init__(
         device_rate: int | None = None,
         stream_factory: Callable[..., Any] | None = None,
         rate_query: Callable[[int | None], int] | None = None,
+        poll_timeout: float = 1.0,
     ) -> None:
         query = rate_query or _output_default_rate
         self._device_rate = device_rate if device_rate is not None else query(device)
@@ -104,6 +105,9 @@ def __init__(
         # 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()
+        # How long capture_frames() waits for a chunk before checking whether the
+        # device stream silently died (e.g. unplugged); injectable for fast tests.
+        self._poll_timeout = poll_timeout
         self._out = bytearray()
         self._out_state: Any = None
         self._lock = threading.Lock()
@@ -156,10 +160,26 @@ def pending(self) -> int:
             return len(self._out) // 2
 
     def capture_frames(self) -> Iterator[bytes]:
-        """Yield target-rate PCM captured from the device until closed."""
+        """Yield target-rate PCM captured from the device until closed.
+
+        Waits in short timeouts rather than blocking forever: if the PortAudio
+        stream dies without close() being called (device unplugged mid-session),
+        no sentinel ever arrives, so a bare get() would hang the capture thread —
+        and with it the whole agent session — instead of surfacing an error.
+        """
         state: Any = None
         while True:
-            chunk = self._in.get()
+            try:
+                chunk = self._in.get(timeout=self._poll_timeout)
+            except queue.Empty:
+                if self._started and not getattr(self._stream, "active", True):
+                    raise CLIError(
+                        "The audio device stopped producing input.",
+                        error_type="audio_input_error",
+                        exit_code=1,
+                        suggestion="Check your microphone/output device, then run 'aai doctor'.",
+                    ) from None
+                continue
             if chunk is None:
                 return
             if self._device_rate != self._target:

aai_cli/agent/session.py

@@ -12,8 +12,12 @@
 from aai_cli.errors import APIError, CLIError, auth_failure, is_auth_failure
 
 
-def _ws_url() -> str:
-    """Voice Agent socket URL for the active environment (set at CLI startup)."""
+def ws_url() -> str:
+    """Voice Agent socket URL for the active environment (set at CLI startup).
+
+    Shared with code_gen so generated agent scripts target the same host the
+    CLI itself would connect to.
+    """
     return f"wss://{environments.active().agents_host}/v1/ws"
 
 
@@ -196,7 +200,7 @@ def _auth_or_api_error(exc: Exception, message: str) -> CLIError:
 def _open_ws(connect: _Connect, api_key: str) -> _WebSocket:
     """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}"})
+        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
 

aai_cli/client.py

@@ -138,12 +138,8 @@ def _transcript_text(transcript: Any) -> str:
     return str(getattr(transcript, "text", "") or "")
 
 
-def _objects(value: object) -> list[object]:
-    return jsonshape.object_list(value)
-
-
 def _render_utterances(transcript: Any) -> str:
-    utterances = _objects(getattr(transcript, "utterances", None))
+    utterances = jsonshape.object_list(getattr(transcript, "utterances", None))
     if not utterances:
         return _transcript_text(transcript)
     return "\n".join(

aai_cli/code_gen/agent.py

@@ -2,9 +2,11 @@
 
 import json
 
-# Injected fields ({voice}/{system_prompt}/{greeting}) are substituted with json.dumps
-# values via str.format in a single pass, so prompt text can't collide with the
-# template's own braces. Every other literal brace below is doubled for str.format.
+from aai_cli.agent import session
+
+# Injected fields ({ws_url}/{voice}/{system_prompt}/{greeting}) are substituted with
+# json.dumps values via str.format in a single pass, so prompt text can't collide with
+# the template's own braces. Every other literal brace below is doubled for str.format.
 _TEMPLATE = """# Live two-way voice conversation with an AssemblyAI voice agent.
 # Requires audio support:  pip install sounddevice websockets
 # Tip: use headphones — the mic stays open while the agent speaks.
@@ -18,7 +20,7 @@
 from websockets.sync.client import connect
 
 API_KEY = os.environ["ASSEMBLYAI_API_KEY"]
-WS_URL = "wss://agents.assemblyai.com/v1/ws"
+WS_URL = {ws_url}
 RATE = 24000  # Voice Agent native PCM16 mono sample rate
 
 # ONE full-duplex stream (mic + speaker together) at the agent's native 24 kHz. Opening
@@ -87,8 +89,13 @@ def send_mic(ws):
 
 
 def render(voice: str, system_prompt: str, greeting: str) -> str:
-    """Generate a runnable voice-agent script with the given session settings."""
+    """Generate a runnable voice-agent script with the given session settings.
+
+    The socket URL comes from the active environment, so a sandbox run generates
+    a script that targets the sandbox its key was minted for, not production.
+    """
     return _TEMPLATE.format(
+        ws_url=json.dumps(session.ws_url()),
         voice=json.dumps(voice),
         system_prompt=json.dumps(system_prompt),
         greeting=json.dumps(greeting),

aai_cli/code_gen/stream.py

@@ -2,7 +2,7 @@
 
 from typing import cast
 
-from aai_cli import llm as gateway
+from aai_cli import environments
 from aai_cli.code_gen import serialize
 
 # Streaming-class imports always used by the generated scaffold. SpeechModel is added
@@ -34,7 +34,7 @@ def on_turn(client: StreamingClient, event: TurnEvent) -> None:
 
 
 client = StreamingClient(
-    StreamingClientOptions(api_key=API_KEY, api_host="streaming.assemblyai.com")
+    StreamingClientOptions(api_key=API_KEY, api_host={api_host!r})
 )
 client.on(StreamingEvents.Turn, on_turn)
 """
@@ -102,7 +102,7 @@ def on_turn(client: StreamingClient, event: TurnEvent) -> None:
 
 
 client = StreamingClient(
-    StreamingClientOptions(api_key=API_KEY, api_host="streaming.assemblyai.com")
+    StreamingClientOptions(api_key=API_KEY, api_host={api_host!r})
 )
 client.on(StreamingEvents.Turn, on_turn)
 """
@@ -136,18 +136,24 @@ def _imports_block(merged: dict[str, object]) -> str:
 
 
 def _build_preamble(imports: str, llm: dict[str, object] | None) -> str:
-    """Pick and fill the plain vs. LLM-Gateway preamble for the given imports."""
+    """Pick and fill the plain vs. LLM-Gateway preamble for the given imports.
+
+    Hosts come from the active environment, so a sandbox run generates a script
+    that targets the sandbox its key was minted for, not production.
+    """
+    env = environments.active()
     if llm:
         prompts = "\n".join(f"    {p!r}," for p in cast("list[str]", llm["prompts"]))
         return _LLM_PREAMBLE.format(
             imports=imports,
-            base_url=gateway.GATEWAY_BASE_URL,
+            api_host=env.streaming_host,
+            base_url=env.llm_gateway_base,
             prompts=prompts,
             model=llm["model"],
             max_tokens=llm["max_tokens"],
             interval=llm.get("interval", 0.0),
         )
-    return _PREAMBLE.format(imports=imports)
+    return _PREAMBLE.format(imports=imports, api_host=env.streaming_host)
 
 
 def _build_connect(merged: dict[str, object]) -> str:

aai_cli/code_gen/transcribe.py

@@ -2,7 +2,7 @@
 
 from typing import cast
 
-from aai_cli import llm
+from aai_cli import environments, llm
 from aai_cli.code_gen import serialize, snippets
 
 
@@ -38,6 +38,13 @@ def render(
         "",
         '# Export your key first:  export ASSEMBLYAI_API_KEY="<your key>"',
         'aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]',
+    ]
+    # The SDK ships pointing at production, so only a non-default environment
+    # (e.g. --sandbox) needs its api base spelled out in the generated script.
+    env = environments.active()
+    if env.api_base != environments.get(environments.DEFAULT_ENV).api_base:
+        parts.append(f"aai.settings.base_url = {env.api_base!r}")
+    parts += [
         "",
         "transcriber = aai.Transcriber()",
     ]
@@ -75,7 +82,7 @@ def _llm_gateway_block(llm_gateway: dict[str, object]) -> list[str]:
         "# Each prompt runs on the previous response; the first runs on the transcript.",
         "gateway = OpenAI(",
         '    api_key=os.environ["ASSEMBLYAI_API_KEY"],',
-        f"    base_url={llm.GATEWAY_BASE_URL!r},",
+        f"    base_url={environments.active().llm_gateway_base!r},",
         ")",
         "prompts = [",
         prompt_lines,

aai_cli/commands/account.py

@@ -7,7 +7,7 @@
 from rich.markup import escape
 from rich.text import Text
 
-from aai_cli import help_panels, jsonshape, output, timeparse
+from aai_cli import help_panels, jsonshape, options, output, timeparse
 from aai_cli.auth import ams
 from aai_cli.context import AppState, resolve_session, run_command
 from aai_cli.errors import UsageError
@@ -112,7 +112,7 @@ def _line_items_summary(item: Mapping[str, object]) -> str:
 )
 def balance(
     ctx: typer.Context,
-    json_out: bool = typer.Option(False, "--json", help="Output raw JSON."),
+    json_out: bool = options.json_option(),
 ) -> None:
     """Show your remaining account balance."""
 
@@ -151,7 +151,7 @@ def usage(
     end: str | None = typer.Option(None, "--end", help="End date (YYYY-MM-DD). Default: today."),
     window: str | None = typer.Option(None, "--window", help="Window size, e.g. 'day' or 'month'."),
     include_zero: bool = typer.Option(False, "--all", help="Include zero-usage windows."),
-    json_out: bool = typer.Option(False, "--json", help="Output raw JSON."),
+    json_out: bool = options.json_option(),
 ) -> None:
     """Show usage over a date range (defaults to the last 30 days)."""
 
@@ -225,7 +225,7 @@ def render(d: dict[str, object]) -> object:
 )
 def limits(
     ctx: typer.Context,
-    json_out: bool = typer.Option(False, "--json", help="Output raw JSON."),
+    json_out: bool = options.json_option(),
 ) -> None:
     """Show your account's rate limits per service."""
 

aai_cli/commands/agent.py

@@ -6,7 +6,7 @@
 
 import typer
 
-from aai_cli import choices, client, code_gen, config, help_panels, output
+from aai_cli import choices, client, code_gen, config, help_panels, options, output
 from aai_cli.agent.audio import SAMPLE_RATE, DuplexAudio, NullPlayer
 from aai_cli.agent.render import AgentRenderer
 from aai_cli.agent.session import (
@@ -103,7 +103,7 @@ def agent(
     greeting: str = typer.Option(DEFAULT_GREETING, "--greeting", help="Spoken greeting."),
     device: int | None = typer.Option(None, "--device", help="Microphone device index."),
     list_voices: bool = typer.Option(False, "--list-voices", help="Print known voices and exit."),
-    json_out: bool = typer.Option(False, "--json", help="Emit newline-delimited JSON events."),
+    json_out: bool = options.json_option("Emit newline-delimited JSON events."),
     output_field: choices.TextOrJson | None = typer.Option(
         None,
         "-o",

aai_cli/commands/audit.py

@@ -5,7 +5,7 @@
 import typer
 from rich.markup import escape
 
-from aai_cli import help_panels, jsonshape, output, timeparse
+from aai_cli import help_panels, jsonshape, options, output, timeparse
 from aai_cli.auth import ams
 from aai_cli.context import AppState, resolve_session, run_command
 from aai_cli.help_text import examples_epilog
@@ -93,7 +93,7 @@ def audit(
     include_logins: bool = typer.Option(
         False, "--include-logins", help="Show successful login events."
     ),
-    json_out: bool = typer.Option(False, "--json", help="Output raw JSON."),
+    json_out: bool = options.json_option(),
 ) -> None:
     """List recent audit-log entries for your account."""
 

aai_cli/commands/dev.py

@@ -7,7 +7,7 @@
 import typer
 from rich.markup import escape
 
-from aai_cli import help_panels, output, steps
+from aai_cli import help_panels, options, output, steps
 from aai_cli.context import AppState, run_command
 from aai_cli.help_text import examples_epilog
 from aai_cli.init import devserver, procfile, runner
@@ -66,7 +66,7 @@ def dev(
     no_install: bool = typer.Option(
         False, "--no-install", help="Skip dependency install; launch directly."
     ),
-    json_out: bool = typer.Option(False, "--json", help="Output raw JSON."),
+    json_out: bool = options.json_option(),
 ) -> None:
     """Launch the dev server for the app in the current directory.
 

aai_cli/commands/doctor.py

@@ -8,7 +8,7 @@
 import typer
 from rich.markup import escape
 
-from aai_cli import client, config, help_panels, output, theme
+from aai_cli import client, config, help_panels, options, output, theme
 from aai_cli.context import AppState, resolve_profile, run_command
 from aai_cli.errors import CLIError, NotAuthenticated
 from aai_cli.help_text import examples_epilog
@@ -224,7 +224,7 @@ def render(data: DoctorResult) -> str:
 )
 def doctor(
     ctx: typer.Context,
-    json_out: bool = typer.Option(False, "--json", help="Output raw JSON."),
+    json_out: bool = options.json_option(),
 ) -> None:
     """Check that your environment is ready to use AssemblyAI."""
 

aai_cli/commands/init.py

@@ -7,7 +7,7 @@
 import typer
 from rich.markup import escape
 
-from aai_cli import __version__, environments, help_panels, output, steps
+from aai_cli import __version__, environments, help_panels, options, output, steps
 from aai_cli.context import AppState, run_command
 from aai_cli.errors import CLIError
 from aai_cli.help_text import examples_epilog
@@ -255,7 +255,7 @@ def init(
     force: bool = typer.Option(False, "--force", help="Overwrite a non-empty target directory."),
     here: bool = typer.Option(False, "--here", help="Scaffold into the current directory."),
     port: int = typer.Option(3000, "--port", help="Local server port."),
-    json_out: bool = typer.Option(False, "--json", help="Output raw JSON."),
+    json_out: bool = options.json_option(),
 ) -> None:
     """Scaffold a new project from a template, then launch it.