Refactor context module: extract utilities, consolidate error handling

.importlinter

@@ -28,6 +28,7 @@ source_modules =
     aai_cli.microphone
     aai_cli.options
     aai_cli.output
+    aai_cli.procs
     aai_cli.render
     aai_cli.speak_exec
     aai_cli.stdio

aai_cli/argscan.py

@@ -9,13 +9,18 @@
 
 from __future__ import annotations
 
+# The standalone "give me JSON" flag spellings. Shared with main's misplaced-flag
+# hint (which recognizes a `--json`/`-j` passed at the root level), so the two
+# can't drift on which forms count.
+JSON_FLAGS = ("--json", "-j")
+
 
 def requests_json(raw_args: list[str]) -> bool:
     """Whether the token list opts into JSON output: ``--json``, ``-j``,
     ``-o json``, ``--output json``, or their glued forms (``--output=json``,
     ``-ojson``)."""
     for index, token in enumerate(raw_args):
-        if token in ("--json", "-j", "--output=json", "-ojson"):
+        if token in (*JSON_FLAGS, "--output=json", "-ojson"):
             return True
         if token in ("-o", "--output") and raw_args[index + 1 : index + 2] == ["json"]:
             return True

aai_cli/auth/flow.py

@@ -141,24 +141,27 @@ def _reusable_cli_key(token: _Token) -> str | None:
     return None
 
 
+def _no_project_error() -> APIError:
+    """The one failure mode behind both guards in ``find_or_create_cli_key``:
+    nowhere to mint a key (no project entries at all, or an entry without a project)."""
+    return APIError(
+        "Your account has no project to create an API key in.",
+        suggestion="Create a project in the AssemblyAI dashboard, then run 'assembly login' again.",
+    )
+
+
 def find_or_create_cli_key(account_id: int, session_jwt: str) -> str:
     """Return the existing 'AssemblyAI CLI' key, or create one in the first project."""
     projects = _parse(_PROJECT_LIST, ams.list_projects(account_id, session_jwt))
     if not projects:
-        raise APIError(
-            "Your account has no project to create an API key in.",
-            suggestion="Create a project in the AssemblyAI dashboard, then run 'assembly login' again.",
-        )
+        raise _no_project_error()
     for entry in projects:
         for token in entry.tokens:
             if key := _reusable_cli_key(token):
                 return key
     project = projects[0].project
     if project is None:
-        raise APIError(
-            "Your account has no project to create an API key in.",
-            suggestion="Create a project in the AssemblyAI dashboard, then run 'assembly login' again.",
-        )
+        raise _no_project_error()
     created = ams.create_token(account_id, project.id, endpoints.CLI_TOKEN_NAME, session_jwt)
     return _parse(_CREATED_TOKEN, created).api_key
 

aai_cli/commands/account.py

@@ -10,7 +10,7 @@
 
 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.context import AppState, run_command
 from aai_cli.errors import UsageError
 from aai_cli.help_text import examples_epilog
 
@@ -144,7 +144,7 @@ def balance(
     """Show your remaining account balance."""
 
     def body(state: AppState, json_mode: bool) -> None:
-        _, jwt = resolve_session(state)
+        _, jwt = state.resolve_session()
         data = ams.get_balance(jwt)
         cents = jsonshape.as_float(data.get("balance_in_cents"))
         output.emit(
@@ -207,7 +207,7 @@ def body(state: AppState, json_mode: bool) -> None:
             )
         start_date = _utc_day_start(start_day)
         end_date = _utc_day_start(end_day)
-        _, jwt = resolve_session(state)
+        _, jwt = state.resolve_session()
         data = ams.get_usage(jwt, start_date, end_date, window)
 
         def render(d: dict[str, object]) -> object:
@@ -243,13 +243,7 @@ def render(d: dict[str, object]) -> object:
                 if show_breakdown:
                     row.append(escape(breakdown))
                 table.add_row(*row)
-            hidden_note = (
-                output.muted(
-                    f"Hidden: {hidden_count} zero-usage window(s). Use --include-zero to show them."
-                )
-                if hidden_count
-                else None
-            )
+            hidden_note = output.hidden_note(hidden_count, "zero-usage window", "--include-zero")
             return output.stack(summary, table, hidden_note)
 
         output.emit(data, render, json_mode=json_mode)
@@ -273,7 +267,7 @@ def limits(
     """Show your account's rate limits per service."""
 
     def body(state: AppState, json_mode: bool) -> None:
-        account_id, jwt = resolve_session(state)
+        account_id, jwt = state.resolve_session()
         data = ams.get_rate_limits(account_id, jwt)
 
         def render(d: dict[str, object]) -> object:

aai_cli/commands/audit.py

@@ -7,7 +7,7 @@
 
 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.context import AppState, run_command
 from aai_cli.help_text import examples_epilog
 
 app = typer.Typer(help="View your account's audit log.")
@@ -98,21 +98,15 @@ def audit(
     """List recent audit-log entries for your account."""
 
     def body(state: AppState, json_mode: bool) -> None:
-        _, jwt = resolve_session(state)
+        _, jwt = state.resolve_session()
         payload = ams.list_audit_logs(jwt, limit=limit, action_taken=action, resource_type=resource)
         rows = _audit_rows(payload)
 
         def render(data: list[dict[str, object]]) -> object:
             hide_logins = not include_logins and action is None
             shown = [entry for entry in data if not (hide_logins and _is_login(entry))]
             hidden_logins = len(data) - len(shown)
-            hidden_note = (
-                output.muted(
-                    f"Hidden: {hidden_logins} login event(s). Use --include-logins to show them."
-                )
-                if hidden_logins
-                else None
-            )
+            hidden_note = output.hidden_note(hidden_logins, "login event", "--include-logins")
             if not shown:
                 message = (
                     "No notable audit events in the recent log."

aai_cli/commands/doctor.py

@@ -10,7 +10,7 @@
 from rich.markup import escape
 
 from aai_cli import client, coding_agent, config, environments, help_panels, options, output, theme
-from aai_cli.context import AppState, resolve_profile, run_command
+from aai_cli.context import AppState, run_command
 from aai_cli.errors import CLIError, NotAuthenticated
 from aai_cli.help_text import examples_epilog
 
@@ -232,6 +232,22 @@ def _check_coding_agent() -> Check:
     )
 
 
+def render_check_lines(checks: list[Check]) -> list[str]:
+    """The per-check report lines (glyph, name — detail, indented fix hint).
+
+    Shared with the onboarding wizard's environment section (which renders the same
+    checks with its own summary line), so the two renders can't drift."""
+    lines: list[str] = []
+    for c in checks:
+        symbol, style = _SYMBOL.get(c["status"], (theme.SYMBOL_HINT, "aai.muted"))
+        lines.append(
+            f"  [{style}]{escape(symbol)}[/{style}] {escape(c['name'])} — {escape(c['detail'])}"
+        )
+        if c["fix"]:
+            lines.append("      " + output.hint(f"fix: {escape(c['fix'])}"))
+    return lines
+
+
 def render(data: DoctorResult) -> str:
     checks = data["checks"]
     lines = [output.heading("Environment check")]
@@ -240,13 +256,7 @@ def render(data: DoctorResult) -> str:
         lines.append(
             "  " + output.hint(f"profile: {escape(profile)} · environment: {escape(environment)}")
         )
-    for c in checks:
-        symbol, style = _SYMBOL.get(c["status"], (theme.SYMBOL_HINT, "aai.muted"))
-        lines.append(
-            f"  [{style}]{escape(symbol)}[/{style}] {escape(c['name'])} — {escape(c['detail'])}"
-        )
-        if c["fix"]:
-            lines.append("      " + output.hint(f"fix: {escape(c['fix'])}"))
+    lines.extend(render_check_lines(checks))
     if data["ok"]:
         lines.append("  " + output.success("Everything looks good."))
         # Only the real `assembly doctor` carries profile context; the onboarding wizard
@@ -277,7 +287,7 @@ def doctor(
     """Check that your environment is ready to use AssemblyAI."""
 
     def body(state: AppState, json_mode: bool) -> None:
-        profile = resolve_profile(state)
+        profile = state.resolve_profile()
         checks = [
             check_python(),
             _check_credentials(profile),

aai_cli/commands/init.py

@@ -1,13 +1,12 @@
 # aai_cli/commands/init.py
 from __future__ import annotations
 
-import sys
 from pathlib import Path
 
 import typer
 from rich.markup import escape
 
-from aai_cli import __version__, environments, help_panels, options, output, steps
+from aai_cli import __version__, environments, help_panels, options, output, stdio, steps
 from aai_cli.context import AppState, run_command
 from aai_cli.errors import CLIError, UsageError
 from aai_cli.help_text import examples_epilog
@@ -23,7 +22,7 @@
 
 def _pick_template() -> str:
     """Interactive picker; raises a usage error when there's no TTY to prompt on."""
-    if not sys.stdin.isatty() or not sys.stdout.isatty():
+    if not stdio.interactive_stdio():
         raise CLIError(
             "No template given and not running interactively. "
             f"Pass one of: {', '.join(templates.TEMPLATE_ORDER)}.",

aai_cli/commands/keys.py

@@ -5,7 +5,7 @@
 
 from aai_cli import jsonshape, options, output
 from aai_cli.auth import ams
-from aai_cli.context import AppState, resolve_session, run_command
+from aai_cli.context import AppState, run_command
 from aai_cli.errors import APIError, UsageError
 from aai_cli.help_text import examples_epilog
 
@@ -60,7 +60,7 @@ def list_(
     """List API keys across your projects (keys shown masked)."""
 
     def body(state: AppState, json_mode: bool) -> None:
-        account_id, jwt = resolve_session(state)
+        account_id, jwt = state.resolve_session()
         projects = ams.list_projects(account_id, jwt)
         rows: list[dict[str, object]] = []
         for entry in projects:
@@ -126,7 +126,7 @@ def body(state: AppState, json_mode: bool) -> None:
                 "--name must not be empty.",
                 suggestion="Pass a label for the key, e.g. --name ci-pipeline.",
             )
-        account_id, jwt = resolve_session(state)
+        account_id, jwt = state.resolve_session()
         pid = project_id if project_id is not None else _default_project_id(account_id, jwt)
         created = ams.create_token(account_id, pid, name, jwt)
         output.emit(
@@ -158,7 +158,7 @@ def rename(
     """Rename an existing API key."""
 
     def body(state: AppState, json_mode: bool) -> None:
-        account_id, jwt = resolve_session(state)
+        account_id, jwt = state.resolve_session()
         ams.rename_token(account_id, token_id, new_name, jwt)
         output.emit(
             {"id": token_id, "name": new_name},

aai_cli/commands/llm.py

@@ -1,41 +1,28 @@
 from __future__ import annotations
 
-from collections.abc import Callable
-
 import typer
 
 from aai_cli import choices, help_panels, llm_exec, options, output
 from aai_cli import llm as gateway
-from aai_cli.context import AppState, run_command
+from aai_cli.context import run_command
 from aai_cli.errors import UsageError
 from aai_cli.help_text import examples_epilog
 
 app = typer.Typer()
 
 
-def _emit_model_list(_state: AppState, json_mode: bool) -> None:
-    """--list-models body, routed through run_command so --json yields a
-    machine-readable array instead of the human list; needs no auth."""
+def _list_models(output_field: choices.TextOrJson | None, json_mode: bool) -> None:
+    """The --list-models body, routed through run_command so --json yields a
+    machine-readable array instead of the human list; needs no auth. Rejects -o
+    (it only applies to one-shot mode, mirroring how --follow rejects it)."""
+    if output_field is not None:
+        raise UsageError(
+            "--output applies to one-shot mode; --list-models prints the plain "
+            "list (use --json for a machine-readable array)."
+        )
     output.emit(list(gateway.KNOWN_MODELS), "\n".join, json_mode=json_mode)
 
 
-def _list_models_body(
-    output_field: choices.TextOrJson | None,
-) -> Callable[[AppState, bool], None]:
-    """The --list-models command body: rejects -o (it only applies to one-shot
-    mode, mirroring how --follow rejects it) before printing the known models."""
-
-    def body(state: AppState, json_mode: bool) -> None:
-        if output_field is not None:
-            raise UsageError(
-                "--output applies to one-shot mode; --list-models prints the plain "
-                "list (use --json for a machine-readable array)."
-            )
-        _emit_model_list(state, json_mode)
-
-    return body
-
-
 @app.command(
     rich_help_panel=help_panels.TRANSCRIPTION,
     epilog=examples_epilog(
@@ -94,7 +81,9 @@ def llm(
     """
 
     if list_models:
-        run_command(ctx, _list_models_body(output_field), json=json_out)
+        run_command(
+            ctx, lambda _state, json_mode: _list_models(output_field, json_mode), json=json_out
+        )
         return
 
     opts = llm_exec.LlmOptions(

aai_cli/commands/login.py

@@ -5,7 +5,7 @@
 from rich.table import Table
 
 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.context import AppState, persist_browser_login, run_command
 from aai_cli.errors import APIError, CLIError, UsageError
 from aai_cli.help_text import examples_epilog
 
@@ -29,7 +29,7 @@ def login(
     """Authenticate via your browser; stores a CLI API key."""
 
     def body(state: AppState, json_mode: bool) -> None:
-        profile = resolve_profile(state)
+        profile = state.resolve_profile()
         env = environments.active().name
         if api_key is not None and not api_key.strip():
             # An explicitly-passed empty/whitespace key (e.g. --api-key "$UNSET_VAR")
@@ -120,7 +120,7 @@ def logout(
     """Clear stored credentials for the active profile."""
 
     def body(state: AppState, json_mode: bool) -> None:
-        profile = resolve_profile(state)
+        profile = state.resolve_profile()
         # Look before clearing so the report is truthful: "Signed out" on a fresh
         # machine (or a typo'd --profile) would claim something happened when
         # nothing was stored. Still exit 0 either way — logout is idempotent.
@@ -161,7 +161,7 @@ def whoami(
     """Show the active profile and whether its key is usable."""
 
     def body(state: AppState, json_mode: bool) -> None:
-        profile = resolve_profile(state)
+        profile = state.resolve_profile()
         # 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 = state.resolve_api_key()

aai_cli/commands/onboard.py

@@ -1,11 +1,9 @@
 from __future__ import annotations
 
-import sys
-
 import typer
 
-from aai_cli import help_panels, options, output
-from aai_cli.context import AppState, resolve_profile, run_command
+from aai_cli import help_panels, options, output, stdio
+from aai_cli.context import AppState, run_command
 from aai_cli.errors import CLIError
 from aai_cli.help_text import examples_epilog
 from aai_cli.onboard import wizard
@@ -20,7 +18,7 @@ def build_prompter(*, non_interactive: bool = False) -> Prompter:
     otherwise never block for input."""
     if non_interactive:
         return NonInteractivePrompter()
-    if sys.stdin.isatty() and sys.stdout.isatty():
+    if stdio.interactive_stdio():
         return InteractivePrompter()
     return NonInteractivePrompter()
 
@@ -45,7 +43,7 @@ def onboard(
     """Guided setup: sign in, run your first transcription, and start building."""
 
     def body(state: AppState, json_mode: bool) -> None:
-        profile = resolve_profile(state)
+        profile = state.resolve_profile()
         wiz_ctx = WizardContext(state=state, profile=profile, json_mode=json_mode)
         # --json also forces non-interactive: a machine-output run can't block on
         # prompts, and the interactive prompter would write prose onto the JSON stdout.