Simplify: dedupe optional-key resolution, status cells, doctor checks

aai_cli/commands/doctor.py

@@ -44,80 +44,84 @@ def query_devices(self) -> Sequence[Mapping[str, object]]: ...
 }
 
 
+def _check(
+    name: str,
+    status: str,
+    detail: str,
+    *,
+    fix: str | None = None,
+    affects: list[str] | None = None,
+) -> Check:
+    """Assemble a Check. ``affects`` defaults to empty — an 'ok' check blocks nothing."""
+    return {"name": name, "status": status, "affects": affects or [], "detail": detail, "fix": fix}
+
+
 def check_python() -> Check:
     v = sys.version_info
     version = f"{v.major}.{v.minor}.{v.micro}"
     if v >= (3, 12):
-        return {"name": "python", "status": "ok", "affects": [], "detail": version, "fix": None}
-    return {
-        "name": "python",
-        "status": "fail",
-        "affects": ["everything"],
-        "detail": f"Python {version} is too old; the CLI needs 3.12+",
-        "fix": "Install Python 3.12 or newer, then reinstall the CLI.",
-    }
+        return _check("python", "ok", version)
+    return _check(
+        "python",
+        "fail",
+        f"Python {version} is too old; the CLI needs 3.12+",
+        fix="Install Python 3.12 or newer, then reinstall the CLI.",
+        affects=["everything"],
+    )
 
 
 def _check_api_key(profile: str) -> Check:
-    affects = ["everything"]
     try:
         key = config.resolve_api_key(profile=profile)
     except NotAuthenticated:
-        return {
-            "name": "api-key",
-            "status": "fail",
-            "affects": affects,
-            "detail": "No API key found.",
-            "fix": "Run 'aai login' (or set ASSEMBLYAI_API_KEY).",
-        }
+        return _check(
+            "api-key",
+            "fail",
+            "No API key found.",
+            fix="Run 'aai login' (or set ASSEMBLYAI_API_KEY).",
+            affects=["everything"],
+        )
     # validate_key doubles as the connectivity probe: it makes one cheap authed call,
     # so a pass means the key is valid AND api.assemblyai.com is reachable.
     try:
         valid = client.validate_key(key)
     except CLIError as exc:
-        return {
-            "name": "api-key",
-            "status": "fail",
-            "affects": affects,
-            "detail": f"Could not reach AssemblyAI: {exc.message}",
-            "fix": "Check your network/proxy and that api.assemblyai.com is reachable.",
-        }
+        return _check(
+            "api-key",
+            "fail",
+            f"Could not reach AssemblyAI: {exc.message}",
+            fix="Check your network/proxy and that api.assemblyai.com is reachable.",
+            affects=["everything"],
+        )
     if valid:
-        return {
-            "name": "api-key",
-            "status": "ok",
-            "affects": [],
-            "detail": "API key is valid and AssemblyAI is reachable.",
-            "fix": None,
-        }
-    return {
-        "name": "api-key",
-        "status": "fail",
-        "affects": affects,
-        "detail": "API key was rejected (HTTP 401).",
-        "fix": "Run 'aai login' with a valid key.",
-    }
+        return _check("api-key", "ok", "API key is valid and AssemblyAI is reachable.")
+    return _check(
+        "api-key",
+        "fail",
+        "API key was rejected (HTTP 401).",
+        fix="Run 'aai login' with a valid key.",
+        affects=["everything"],
+    )
 
 
 def check_ffmpeg() -> Check:
     # ffmpeg is ONLY used to stream non-WAV files or URLs (stream/agent), where it
     # decodes them to 16 kHz mono PCM on the fly. Plain `transcribe` (including
     # YouTube URLs) uploads the file to AssemblyAI and never invokes ffmpeg, so it is
     # not required for transcription.
-    affects = ["stream/agent (non-WAV file or URL input)"]
     if shutil.which("ffmpeg"):
-        return {"name": "ffmpeg", "status": "ok", "affects": [], "detail": "found", "fix": None}
-    return {
-        "name": "ffmpeg",
-        "status": "warn",
-        "affects": affects,
-        "detail": (
+        return _check("ffmpeg", "ok", "found")
+    return _check(
+        "ffmpeg",
+        "warn",
+        (
             "ffmpeg not found. Only needed to stream non-WAV files or URLs; "
             "transcription (including YouTube) works without it, as does streaming a "
             "16 kHz mono WAV."
         ),
-        "fix": "Install ffmpeg (macOS: brew install ffmpeg; Debian/Ubuntu: apt-get install ffmpeg).",
-    }
+        fix="Install ffmpeg (macOS: brew install ffmpeg; Debian/Ubuntu: apt-get install ffmpeg).",
+        affects=["stream/agent (non-WAV file or URL input)"],
+    )
 
 
 def _probe_input_devices() -> int:
@@ -144,59 +148,50 @@ def check_audio() -> Check:
     try:
         inputs = _probe_input_devices()
     except ImportError:
-        return {
-            "name": "audio",
-            "status": "warn",
-            "affects": affects,
-            "detail": "sounddevice is not importable; the microphone can't be used.",
-            "fix": "pip install --force-reinstall sounddevice",
-        }
+        return _check(
+            "audio",
+            "warn",
+            "sounddevice is not importable; the microphone can't be used.",
+            fix="pip install --force-reinstall sounddevice",
+            affects=affects,
+        )
     except Exception as exc:  # noqa: BLE001 - any PortAudio/device failure is a soft warning
-        return {
-            "name": "audio",
-            "status": "warn",
-            "affects": affects,
-            "detail": f"audio system unavailable: {exc}",
-            "fix": "On Linux install PortAudio: sudo apt-get install libportaudio2",
-        }
+        return _check(
+            "audio",
+            "warn",
+            f"audio system unavailable: {exc}",
+            fix="On Linux install PortAudio: sudo apt-get install libportaudio2",
+            affects=affects,
+        )
     if inputs == 0:
-        return {
-            "name": "audio",
-            "status": "warn",
-            "affects": affects,
-            "detail": "No microphone (input device) found.",
-            "fix": "Connect a microphone; live mic input is needed for stream/agent.",
-        }
-    return {
-        "name": "audio",
-        "status": "ok",
-        "affects": [],
-        "detail": f"{inputs} microphone input device(s) available.",
-        "fix": None,
-    }
+        return _check(
+            "audio",
+            "warn",
+            "No microphone (input device) found.",
+            fix="Connect a microphone; live mic input is needed for stream/agent.",
+            affects=affects,
+        )
+    return _check("audio", "ok", f"{inputs} microphone input device(s) available.")
 
 
 def _check_coding_agent() -> Check:
-    affects = ["aai setup install"]
     missing = [tool for tool in ("claude", "npx") if shutil.which(tool) is None]
     if not missing:
-        return {
-            "name": "coding-agent",
-            "status": "ok",
-            "affects": [],
-            "detail": "claude and npx found; run 'aai setup install' to wire up the docs MCP + skills.",
-            "fix": None,
-        }
-    return {
-        "name": "coding-agent",
-        "status": "warn",
-        "affects": affects,
-        "detail": f"not found: {', '.join(missing)}.",
-        "fix": (
+        return _check(
+            "coding-agent",
+            "ok",
+            "claude and npx found; run 'aai setup install' to wire up the docs MCP + skills.",
+        )
+    return _check(
+        "coding-agent",
+        "warn",
+        f"not found: {', '.join(missing)}.",
+        fix=(
             "Install Claude Code (https://claude.com/claude-code) and Node.js, "
             "then run 'aai setup install'."
         ),
-    }
+        affects=["aai setup install"],
+    )
 
 
 def render(data: DoctorResult) -> str:

aai_cli/commands/sessions.py

@@ -3,7 +3,6 @@
 import typer
 from rich.markup import escape
 from rich.table import Table
-from rich.text import Text
 
 from aai_cli import jsonshape, output, theme
 from aai_cli.auth import ams
@@ -72,10 +71,9 @@ def render(data: list[dict[str, object]]) -> Table:
                 "model",
             )
             for s in data:
-                status_str = str(s["status"])
                 table.add_row(
                     escape(str(s["session_id"])),
-                    Text(status_str, style=theme.status_style(status_str)),
+                    theme.status_text(str(s["status"])),
                     escape(str(s.get("created_at") or "")),
                     escape(str(s.get("audio_duration_sec") or "")),
                     escape(str(s.get("speech_model") or "")),

aai_cli/commands/transcripts.py

@@ -3,7 +3,6 @@
 import typer
 from rich.markup import escape
 from rich.table import Table
-from rich.text import Text
 
 from aai_cli import choices, client, config, output, theme
 from aai_cli.context import AppState, run_command
@@ -86,10 +85,9 @@ def body(state: AppState, json_mode: bool) -> None:
         def render(data: list[dict[str, object]]) -> Table:
             table = output.data_table("id", "status", "created")
             for row in data:
-                status = str(row["status"])
                 table.add_row(
                     escape(str(row["id"])),
-                    Text(status, style=theme.status_style(status)),
+                    theme.status_text(str(row["status"])),
                     escape(str(row.get("created", ""))),
                 )
             return table

aai_cli/config.py

@@ -300,3 +300,13 @@ def resolve_api_key(*, profile: str | None = None, api_key_flag: str | None = No
     if stored:
         return stored
     raise NotAuthenticated()
+
+
+def resolve_api_key_optional(*, profile: str | None = None) -> str | None:
+    """The same key chain as ``resolve_api_key`` (env -> keyring), but ``None`` instead
+    of raising when no key is configured — for callers that work without one
+    (``aai init`` scaffolding, the onboarding wizard's signed-in check)."""
+    try:
+        return resolve_api_key(profile=profile)
+    except NotAuthenticated:
+        return None

aai_cli/init/keys.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from aai_cli import config
-from aai_cli.errors import NotAuthenticated
 
 
 def resolve_optional_api_key(*, profile: str | None) -> str | None:
@@ -10,7 +9,4 @@ def resolve_optional_api_key(*, profile: str | None) -> str | None:
     `aai init` scaffolds even without a key (writing a placeholder), so it must not
     fail the way run commands do.
     """
-    try:
-        return config.resolve_api_key(profile=profile)
-    except NotAuthenticated:
-        return None
+    return config.resolve_api_key_optional(profile=profile)

aai_cli/onboard/sections.py

@@ -11,7 +11,7 @@
 from aai_cli.commands import init as init_cmd
 from aai_cli.commands import setup as setup_cmd
 from aai_cli.context import AppState, persist_browser_login
-from aai_cli.errors import CLIError, NotAuthenticated
+from aai_cli.errors import CLIError
 from aai_cli.onboard.prompter import Prompter
 
 
@@ -29,11 +29,7 @@ class WizardContext:
 
 
 def _has_key(profile: str) -> bool:
-    try:
-        config.resolve_api_key(profile=profile)
-    except NotAuthenticated:
-        return False
-    return True
+    return config.resolve_api_key_optional(profile=profile) is not None
 
 
 def welcome(prompter: Prompter, _ctx: WizardContext) -> SectionResult:

aai_cli/theme.py

@@ -3,6 +3,7 @@
 from typing import IO, Any
 
 from rich.console import Console
+from rich.text import Text
 from rich.theme import Theme
 
 # AssemblyAI brand accent. Defined once so the whole CLI can be re-tinted here.
@@ -87,3 +88,8 @@ def status_style(status: str) -> str:
     if normalized in _WARN:
         return "aai.warn"
     return "aai.muted"
+
+
+def status_text(status: str) -> Text:
+    """A status string rendered in its semantic status color (see ``status_style``)."""
+    return Text(status, style=status_style(status))

tests/test_auth_loopback.py

@@ -9,6 +9,23 @@
 from aai_cli.errors import APIError
 
 
+@pytest.fixture(autouse=True)
+def _unique_loopback_port(monkeypatch):
+    """Give every test its own OS-assigned loopback port.
+
+    Production binds the single fixed ``LOOPBACK_PORT`` (one login at a time), but the
+    suite runs many capture cycles back-to-back. Sharing one fixed port makes the tests
+    flaky under load/random ordering: if one test's server is still bound when the next
+    starts, the bind raises inside the worker thread (surfacing as an unhandled-thread
+    warning) and cascades into the neighbours. A fresh port per test removes that
+    coupling without changing what `capture_callback` does.
+    """
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
+        probe.bind((endpoints.LOOPBACK_HOST, 0))
+        port = probe.getsockname()[1]
+    monkeypatch.setattr(endpoints, "LOOPBACK_PORT", port)
+
+
 def _hit(path: str) -> int | None:
     """Request `path` against the loopback server, returning the HTTP status code.
 

tests/test_doctor.py

@@ -136,6 +136,7 @@ def test_check_python_flags_old_interpreter(monkeypatch):
     check = doctor.check_python()
     assert check["status"] == "fail"
     assert "3.9.0" in check["detail"]
+    assert check["affects"] == ["everything"]
 
 
 def test_check_audio_handles_portaudio_failure(monkeypatch):