Validate input before credentials; fix help rendering and closed pipes

AGENTS.md

@@ -9,11 +9,13 @@ this file, so Claude Code reads the same instructions.
 This project uses [uv](https://docs.astral.sh/uv/). **Run every Python tool through `uv run`** so it uses the locked environment (`pyproject.toml` + `uv.lock`), not whatever is on `PATH`:
 
 ```sh
-uv sync --extra dev          # create/refresh the venv with dev dependencies
+uv sync                      # create/refresh the venv (the dev group installs by default)
 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)
 ```
 
+Dev tooling is a PEP 735 `[dependency-groups]` group with `default-groups = ["dev"]`, not a `[project]` extra — `uv sync --extra dev` errors.
+
 `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`):
@@ -28,16 +30,27 @@ uv run pytest tests/test_transcribe.py -q              # a single file
 uv run pytest tests/test_transcribe.py::test_name -q   # a single test
 ```
 
+The two diff-scoped tail gates are the slowest failures to discover via the full
+script; after a gate run (or any pytest run with the coverage flags below) they can
+be re-run alone:
+
+```sh
+uv run pytest -q -n auto --cov=aai_cli --cov-branch --cov-context=test --cov-report=xml  # refresh coverage data
+uv run diff-cover coverage.xml --compare-branch=origin/main --fail-under=100             # patch-coverage gate
+uv run python scripts/mutation_gate.py origin/main                                       # mutation gate
+```
+
 ### Test markers
 
-The default suite **excludes** two slow/credentialed marker sets (see `scripts/check.sh` and `pyproject.toml`):
+The default suite **excludes** three slow/credentialed marker sets — `pyproject.toml`'s `addopts` carries `-m "not e2e and not install and not install_script"`, so a bare `pytest` matches what `check.sh` gates. An explicit command-line `-m` overrides it for the opt-in runs:
 
 ```sh
 uv run pytest -m e2e             # real-API end-to-end; needs ASSEMBLYAI_API_KEY, else skips
+uv run pytest -m install         # installs each init template's requirements for real; needs network + uv
 uv run pytest -m install_script  # builds a wheel and runs install.sh for real; needs network + uv/pipx
 ```
 
-`check.sh` runs `-m "not e2e and not install_script"` with a **90% branch-coverage gate** (`--cov-fail-under=90`). New code generally needs tests to clear that gate.
+`check.sh` runs the default suite with a **90% branch-coverage gate** (`--cov-fail-under=90`). New code generally needs tests to clear that gate.
 
 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.
 

aai_cli/client.py

@@ -2,7 +2,9 @@
 
 import contextlib
 import json
+import re
 from collections.abc import Callable, Generator, Iterable
+from pathlib import Path
 from typing import Any, Literal, Protocol
 
 import assemblyai as aai
@@ -37,10 +39,14 @@ def _make_streaming_client(api_key: str) -> _StreamingClientLike:
     return client
 
 
-def resolve_audio_source(source: str | None, *, sample: bool) -> str:
+def resolve_audio_source(source: str | None, *, sample: bool, check_local: bool = True) -> str:
     """The audio reference to use: the hosted --sample clip, else the given path/URL.
 
     Shared by `transcribe` and `stream` so both accept a file or URL and `--sample`.
+    A local path that doesn't exist fails here — before any credential resolution or
+    request — so a typo'd filename reads as "file not found", not as an API failure.
+    ``--show-code`` paths pass ``check_local=False``: generating code for a file you
+    don't have yet is legitimate.
     """
     if sample:
         return SAMPLE_AUDIO_URL
@@ -49,6 +55,13 @@ def resolve_audio_source(source: str | None, *, sample: bool) -> str:
             "Provide an audio path or URL.",
             suggestion="Or pass --sample to use the hosted demo file.",
         )
+    if check_local and not source.startswith(("http://", "https://")) and not Path(source).exists():
+        raise CLIError(
+            f"File not found: {source}",
+            error_type="file_not_found",
+            exit_code=2,
+            suggestion="Check the path. For remote audio, pass an http(s):// URL.",
+        )
     return source
 
 
@@ -169,7 +182,23 @@ def select_transcript_field(transcript: Any, field: str) -> str:
     return _FIELD_RENDERERS.get(field, _transcript_text)(transcript)
 
 
+# Transcript ids are opaque url-safe tokens. Reject anything else before it is
+# interpolated into the request path: an "id" like ../v2/other would otherwise steer
+# an authenticated GET at an arbitrary API route.
+_TRANSCRIPT_ID_RE = re.compile(r"^[A-Za-z0-9_-]+$")
+
+
+def validate_transcript_id(transcript_id: str) -> str:
+    if not _TRANSCRIPT_ID_RE.match(transcript_id):
+        raise UsageError(
+            f"{transcript_id!r} doesn't look like a transcript id.",
+            suggestion="Ids are letters/digits/dashes; find them with 'aai transcripts list'.",
+        )
+    return transcript_id
+
+
 def get_transcript(api_key: str, transcript_id: str) -> aai.Transcript:
+    validate_transcript_id(transcript_id)
     _configure(api_key)
     with _sdk_errors(f"Could not fetch transcript {transcript_id}"):
         return aai.Transcript.get_by_id(transcript_id)

aai_cli/commands/agent.py

@@ -145,10 +145,14 @@ def body(state: AppState, json_mode: bool) -> None:
             output.print_code(code_gen.agent(voice, system_prompt_text, greeting))
             return
 
-        api_key = config.resolve_api_key(profile=state.profile)
         from_file = bool(source) or sample
         if from_file and device is not None:
             raise UsageError("--device applies only to microphone input.")
+        if from_file:
+            # Existence-check the clip before credentials, so a typo'd path reads as
+            # "file not found" instead of triggering a login.
+            client.resolve_audio_source(source, sample=sample)
+        api_key = config.resolve_api_key(profile=state.profile)
 
         renderer = AgentRenderer(
             json_mode=json_mode,

aai_cli/commands/login.py

@@ -6,7 +6,7 @@
 
 from aai_cli import client, config, environments, help_panels, options, output
 from aai_cli.context import AppState, persist_browser_login, resolve_profile, run_command
-from aai_cli.errors import APIError, NotAuthenticated
+from aai_cli.errors import APIError
 from aai_cli.help_text import examples_epilog
 
 app = typer.Typer()
@@ -107,9 +107,9 @@ def whoami(
 
     def body(state: AppState, json_mode: bool) -> None:
         profile = resolve_profile(state)
-        key = config.get_api_key(profile)
-        if not key:
-            raise NotAuthenticated()
+        # The full env -> keyring chain (raises NotAuthenticated when empty), so a CI
+        # box authenticated via ASSEMBLYAI_API_KEY can use whoami as a preflight check.
+        key = config.resolve_api_key(profile=state.profile)
         masked = output.mask_secret(key)
         env = environments.active().name
         reachable = client.validate_key(key)

aai_cli/commands/stream.py

@@ -162,7 +162,10 @@ def stream(
     # turn detection
     end_of_turn_confidence_threshold: float | None = typer.Option(
         None,
-        "--end-of-turn-confidence-threshold",
+        # Not "--end-of-turn-confidence-threshold": at 34 chars the name can't render
+        # un-clipped in an 80-column help screen, which made it unlearnable from --help.
+        # The full SDK field stays reachable via --config.
+        "--end-of-turn-confidence",
         help="End-of-turn confidence (0-1).",
         min=0.0,
         max=1.0,
@@ -183,15 +186,15 @@ def stream(
     vad_threshold: float | None = typer.Option(
         None,
         "--vad-threshold",
-        help="Voice-activity threshold (0-1).",
+        help="Voice activity threshold (0-1).",
         min=0.0,
         max=1.0,
         rich_help_panel=help_panels.OPT_TURNS,
     ),
     format_turns: bool | None = typer.Option(
         None,
         "--format-turns/--no-format-turns",
-        help="Punctuate/format finalized turns.",
+        help="Format (punctuate) finalized turns.",
         rich_help_panel=help_panels.OPT_TURNS,
     ),
     include_partial_turns: bool | None = typer.Option(
@@ -393,8 +396,12 @@ def body(state: AppState, json_mode: bool) -> None:
             output.print_code(code_gen.stream(merged, llm=gateway))
             return
 
-        api_key = config.resolve_api_key(profile=state.profile)
+        # Validate the requested sources (including that a local file exists) before
+        # credentials, so a typo'd path reads as "file not found" — not as a login.
         validate_sources(opts, has_llm=bool(llm_prompt), text_mode=text_mode)
+        if opts.from_file and not opts.from_stdin:
+            client.resolve_audio_source(opts.source, sample=opts.sample)
+        api_key = config.resolve_api_key(profile=state.profile)
 
         llm_prompts = list(llm_prompt or [])
         session = StreamSession(

aai_cli/commands/transcribe.py

@@ -392,7 +392,7 @@ def body(state: AppState, json_mode: bool) -> None:
             # authenticating (raw stdout, so `--show-code > script.py` runs). No
             # source/--sample needed — fall back to a placeholder path for a pure snippet.
             audio = (
-                client.resolve_audio_source(source, sample=sample)
+                client.resolve_audio_source(source, sample=sample, check_local=False)
                 if source or sample
                 else "your-audio-file.mp3"
             )
@@ -402,6 +402,9 @@ def body(state: AppState, json_mode: bool) -> None:
 
         tc = config_builder.construct_transcription_config(merged)
 
+        # A typo'd path must read as "file not found", not trigger a login.
+        transcribe_exec.check_source_exists(source, sample=sample)
+
         api_key = config.resolve_api_key(profile=state.profile)
         with output.status("Transcribing…", json_mode=json_mode):
             transcript = transcribe_exec.run_transcription(

aai_cli/config.py

@@ -185,8 +185,21 @@ def set_api_key(profile: str, api_key: str) -> None:
     _dump(cfg)
 
 
+def _keyring_get(username: str) -> str | None:
+    """Read a secret, treating an unusable keyring backend as "nothing stored".
+
+    Headless machines (containers, CI, servers) routinely have no keyring backend at
+    all, so keyring raises NoKeyringError on every read. That state must read as "not
+    signed in" — ASSEMBLYAI_API_KEY still works there — never as a crash.
+    """
+    try:
+        return keyring.get_password(KEYRING_SERVICE, username)
+    except keyring.errors.KeyringError:
+        return None
+
+
 def get_api_key(profile: str) -> str | None:
-    return keyring.get_password(KEYRING_SERVICE, profile)
+    return _keyring_get(profile)
 
 
 def get_profile_env(profile: str) -> str | None:
@@ -204,7 +217,9 @@ def set_profile_env(profile: str, env: str) -> None:
 
 
 def clear_api_key(profile: str) -> None:
-    with contextlib.suppress(keyring.errors.PasswordDeleteError):
+    # KeyringError, not just PasswordDeleteError: with no backend at all (headless
+    # boxes) delete raises NoKeyringError, and "nothing stored" is already the goal.
+    with contextlib.suppress(keyring.errors.KeyringError):
         keyring.delete_password(KEYRING_SERVICE, profile)
 
 
@@ -233,7 +248,7 @@ def set_session(profile: str, *, session_jwt: str, session_token: str, account_i
 
 def get_session(profile: str) -> dict[str, str] | None:
     """The stored {'jwt', 'token'} for a profile, or None if absent/corrupt."""
-    raw = keyring.get_password(KEYRING_SERVICE, _session_username(profile))
+    raw = _keyring_get(_session_username(profile))
     if not raw:
         return None
     try:
@@ -250,7 +265,7 @@ def get_account_id(profile: str) -> int | None:
 
 
 def clear_session(profile: str) -> None:
-    with contextlib.suppress(keyring.errors.PasswordDeleteError):
+    with contextlib.suppress(keyring.errors.KeyringError):
         keyring.delete_password(KEYRING_SERVICE, _session_username(profile))
     cfg = _load()
     prof = cfg.profiles.get(profile)
@@ -278,8 +293,8 @@ def persist_login(
     restored best-effort.
     """
     _validate_profile(profile)
-    prior_api_key = keyring.get_password(KEYRING_SERVICE, profile)
-    prior_session = keyring.get_password(KEYRING_SERVICE, _session_username(profile))
+    prior_api_key = _keyring_get(profile)
+    prior_session = _keyring_get(_session_username(profile))
     prior_cfg = _load()
     done = False
     try:

aai_cli/context.py

@@ -171,3 +171,20 @@ def run_command(
     except CLIError as err:
         output.emit_error(err, json_mode=json_mode)
         raise typer.Exit(code=err.exit_code) from None
+    except (typer.Exit, typer.Abort, BrokenPipeError):
+        # Deliberate control flow (and the closed-pipe contract handled in main.run);
+        # these must reach Click/the entry point untouched.
+        raise
+    except Exception as exc:
+        # Last resort: a bug must surface as one clean line (and the JSON error shape
+        # under --json), never as a raw traceback.
+        internal = CLIError(
+            f"Unexpected error: {exc}",
+            error_type="internal_error",
+            suggestion=(
+                "This looks like a bug in the CLI — please report it at "
+                "https://github.com/AssemblyAI/cli/issues."
+            ),
+        )
+        output.emit_error(internal, json_mode=json_mode)
+        raise typer.Exit(code=internal.exit_code) from exc

aai_cli/errors.py

@@ -39,7 +39,8 @@ def __init__(
         message: str = "You're not signed in.",
         *,
         suggestion: str | None = (
-            "Run 'aai onboard' for guided setup, or 'aai login' if you have an account."
+            "Run 'aai onboard' for guided setup, 'aai login' if you have an account, "
+            "or set ASSEMBLYAI_API_KEY."
         ),
     ) -> None:
         super().__init__(

aai_cli/main.py

@@ -1,10 +1,16 @@
 from __future__ import annotations
 
 import sys
+from types import ModuleType
 from typing import TYPE_CHECKING
 
 import typer
+from rich.console import RenderableType
+from rich.style import StyleType
+from rich.table import Table
 from typer import completion, rich_utils
+from typer._click.exceptions import ClickException, NoSuchOption
+from typer._click.utils import PacifyFlushWrapper
 from typer.core import TyperGroup
 
 if TYPE_CHECKING:
@@ -111,6 +117,50 @@ def parse_args(self, ctx: ClickContext, args: list[str]) -> list[str]:
 rich_utils.STYLE_USAGE = theme.MUTED
 rich_utils.STYLE_USAGE_COMMAND = f"bold {theme.BRAND}"
 
+
+# Help tables put flag/command names in the leading columns and wrapping prose
+# (metavar, help text) in the trailing two. Rich's width collapse only spares no_wrap
+# columns, so on a narrow terminal it happily clips a flag name to "--end-of-turn-c…" —
+# unlearnable from the help screen itself. Pin every column except the last two so the
+# prose columns absorb the squeeze instead.
+class _NoClipTable(Table):
+    def add_row(
+        self,
+        *renderables: RenderableType | None,
+        style: StyleType | None = None,
+        end_section: bool = False,
+    ) -> None:
+        super().add_row(*renderables, style=style, end_section=end_section)
+        for column in self.columns[:-2]:
+            column.no_wrap = True
+
+
+def _patch_module(module: ModuleType, **attrs: object) -> None:
+    """Replace module attributes that are imports (not definitions) in their module —
+    strict mypy's no-implicit-reexport rejects plain attribute assignment for those."""
+    for name, value in attrs.items():
+        setattr(module, name, value)
+
+
+# Typer's own help/error consoles must also honor the closed-pipe contract: with
+# Rich's default Console, `aai --help | head -2` exits 1 via Console.on_broken_pipe.
+_patch_module(rich_utils, Table=_NoClipTable, Console=theme.PipeSafeConsole)
+
+_format_click_error = rich_utils.rich_format_error
+
+
+def _format_click_error_fixed(self: ClickException) -> None:
+    # Typer's vendored Click renders flag suggestions as a stringified 1-tuple:
+    # "No such option: --jsno ('(Possible options: --json)',)". Fold the suggestion
+    # into the message ourselves so the user sees "(Possible options: --json)".
+    if isinstance(self, NoSuchOption) and self.possibilities:
+        self.message = f"{self.message} (Possible options: {', '.join(sorted(self.possibilities))})"
+        self.possibilities = None
+    _format_click_error(self)
+
+
+rich_utils.rich_format_error = _format_click_error_fixed
+
 # Typer's built-in `--show-completion` help is long enough to wrap several lines in
 # the options panel. Trim it so it fits on fewer rows. The OptionInfo objects live on
 # the completion placeholder's parameter defaults; reach the (underscore-prefixed)
@@ -204,7 +254,10 @@ def _offer_or_help(ctx: typer.Context, state: AppState) -> None:
                 'aai transcribe call.mp3 --llm "summarize action items"',
             ),
         ]
-    ),
+    )
+    + "\n\n[bold]Authentication[/bold]\n\n"
+    "Run 'aai login', or set ASSEMBLYAI_API_KEY (used before the stored key). "
+    "--env or AAI_ENV selects the backend: production, sandbox000.",
 )
 def main(
     ctx: typer.Context,
@@ -288,3 +341,11 @@ def run() -> None:
     except BrokenPipeError:
         stdio.silence_stdout()
         sys.exit(0)
+    except SystemExit as exc:
+        # Typer's vendored Click handles EPIPE before our handler can see it: it wraps
+        # the streams in PacifyFlushWrapper and exits 1. That wrapper only ever appears
+        # on the closed-pipe path, so rewrite that exit to the documented success code.
+        if exc.code == 1 and isinstance(sys.stdout, PacifyFlushWrapper):
+            stdio.silence_stdout()
+            sys.exit(0)
+        raise