Extract command-line JSON detection to shared argscan module

.importlinter

@@ -7,6 +7,7 @@ name = Core modules do not import command modules
 type = forbidden
 source_modules =
     aai_cli.agent
+    aai_cli.argscan
     aai_cli.auth
     aai_cli.client
     aai_cli.code_gen
@@ -67,6 +68,7 @@ modules =
 name = Library layers do not depend on Rich rendering
 type = forbidden
 source_modules =
+    aai_cli.argscan
     aai_cli.client
     aai_cli.config
     aai_cli.config_builder

AGENTS.md

@@ -93,7 +93,7 @@ Lessons that cost iterations getting the patch-coverage and mutation tail gates
   patch it must accept it or the call `TypeError`s.
 - **`--json` / `-j` is a per-command flag, not a root flag**: `assembly --json transcribe …` fails
   with "No such option"; it's `assembly transcribe … --json`. (The root callback still sniffs the
-  whole token list via `_command_line_requests_json`, so a callback-level failure like a bad
+  whole token list via `argscan.requests_json`, so a callback-level failure like a bad
   `--env` keeps the JSON error shape — but the flag itself lives on the subcommand.)
 
 ### Manual QA / running the CLI in sandboxed sessions

aai_cli/argscan.py

@@ -0,0 +1,22 @@
+"""Sniffing the raw, not-yet-parsed command line for output-mode flags.
+
+Both the root callback (`main`) and telemetry's first-run notice run before any
+subcommand parses its own ``--json``, so honoring a pipeline's request for
+machine-readable output at that point means scanning the raw token list. The
+shared definition lives here — free of Rich and import cycles — so the two
+callers can't drift on which flag forms count.
+"""
+
+from __future__ import annotations
+
+
+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"):
+            return True
+        if token in ("-o", "--output") and raw_args[index + 1 : index + 2] == ["json"]:
+            return True
+    return False

aai_cli/commands/agent.py

@@ -6,7 +6,7 @@
 
 import typer
 
-from aai_cli import choices, client, code_gen, config, help_panels, options, output
+from aai_cli import choices, client, code_gen, 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 (
@@ -182,7 +182,7 @@ def body(state: AppState, json_mode: bool) -> None:
             # 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)
+        api_key = state.resolve_api_key()
 
         renderer = AgentRenderer(
             json_mode=json_mode,

aai_cli/commands/evaluate.py

@@ -16,7 +16,7 @@
 import typer
 from rich.console import RenderableType
 
-from aai_cli import client, config, der, eval_data, help_panels, jsonshape, options, output, wer
+from aai_cli import client, der, eval_data, help_panels, jsonshape, options, output, wer
 from aai_cli.context import AppState, run_command
 from aai_cli.errors import CLIError, NotAuthenticated, UsageError
 from aai_cli.help_text import examples_epilog
@@ -345,7 +345,7 @@ def body(state: AppState, json_mode: bool) -> None:
             )
         # Resolve credentials before any dataset download: a signed-out user must
         # not pull the whole dataset only to fail at the first transcription.
-        api_key = config.resolve_api_key(profile=state.profile)
+        api_key = state.resolve_api_key()
         data = eval_data.load(
             dataset,
             split=split,

aai_cli/commands/llm.py

@@ -5,7 +5,7 @@
 import typer
 from rich.markup import escape
 
-from aai_cli import choices, client, config, help_panels, options, output, stdio
+from aai_cli import choices, client, help_panels, options, output, stdio
 from aai_cli import llm as gateway
 from aai_cli.context import AppState, run_command
 from aai_cli.errors import UsageError
@@ -151,7 +151,7 @@ def llm(
 
     def follow_body(state: AppState, json_mode: bool) -> None:
         prompt_text = _validate_follow_args(prompt, output_field, transcript_id)
-        api_key = config.resolve_api_key(profile=state.profile)
+        api_key = state.resolve_api_key()
 
         def ask(transcript_text: str) -> str:
             messages = gateway.build_messages(
@@ -185,7 +185,7 @@ def body(state: AppState, json_mode: bool) -> None:
             )
         prompt_text = prompt
         stdin_text = _stdin_transcript_text(state, json_mode, transcript_id)
-        api_key = config.resolve_api_key(profile=state.profile)
+        api_key = state.resolve_api_key()
         messages = gateway.build_messages(
             prompt_text, system=system, transcript_id=transcript_id, transcript_text=stdin_text
         )

aai_cli/commands/login.py

@@ -162,7 +162,7 @@ def body(state: AppState, json_mode: bool) -> None:
         profile = resolve_profile(state)
         # 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)
+        key = state.resolve_api_key()
         masked = output.mask_secret(key)
         env = environments.active().name
         # A network failure must not suppress the local table: profile, env, masked

aai_cli/commands/speak.py

@@ -5,7 +5,7 @@
 
 import typer
 
-from aai_cli import config, help_panels, options, output
+from aai_cli import help_panels, options, output
 from aai_cli.context import AppState, run_command
 from aai_cli.errors import CLIError, UsageError
 from aai_cli.help_text import examples_epilog
@@ -231,7 +231,7 @@ def body(state: AppState, json_mode: bool) -> None:
                 "(--sandbox goes before the command; or use --env sandbox000).",
             )
         spoken = _read_text(text)
-        api_key = config.resolve_api_key(profile=state.profile)
+        api_key = state.resolve_api_key()
         bare_voice, overrides = dialogue.parse_voice_overrides(voice)
         if dialogue.looks_like_speaker_labeled(spoken):
             _speak_dialogue(

aai_cli/commands/stream.py

@@ -10,7 +10,6 @@
     choices,
     client,
     code_gen,
-    config,
     config_builder,
     help_panels,
     llm,
@@ -430,7 +429,7 @@ def body(state: AppState, json_mode: bool) -> None:
         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)
+        api_key = state.resolve_api_key()
 
         llm_prompts = list(llm_prompt or [])
         session = StreamSession(

aai_cli/commands/transcribe.py

@@ -9,7 +9,6 @@
     choices,
     client,
     code_gen,
-    config,
     config_builder,
     help_panels,
     llm,
@@ -429,7 +428,7 @@ def body(state: AppState, json_mode: bool) -> None:
                 out=out, output_field=output_field, llm_prompt=llm_prompt, show_code=show_code
             )
             transcribe_batch.run_batch(
-                config.resolve_api_key(profile=state.profile),
+                state.resolve_api_key(),
                 sources,
                 transcription_config=config_builder.construct_transcription_config(merged),
                 concurrency=concurrency,
@@ -466,7 +465,7 @@ def body(state: AppState, json_mode: bool) -> None:
         transcribe_exec.check_source_exists(source, sample=sample)
         transcribe_exec.warn_unrecognized_extension(source, json_mode=json_mode, quiet=state.quiet)
 
-        api_key = config.resolve_api_key(profile=state.profile)
+        api_key = state.resolve_api_key()
         with output.status("Transcribing…", json_mode=json_mode, quiet=state.quiet):
             transcript = transcribe_exec.run_transcription(
                 api_key,

aai_cli/commands/transcripts.py

@@ -3,7 +3,7 @@
 import typer
 from rich.markup import escape
 
-from aai_cli import choices, client, config, options, output, theme, timeparse
+from aai_cli import choices, client, options, output, theme, timeparse
 from aai_cli.context import AppState, run_command
 from aai_cli.errors import APIError
 from aai_cli.help_text import examples_epilog
@@ -36,7 +36,7 @@ def list_(
     """List recent transcripts."""
 
     def body(state: AppState, json_mode: bool) -> None:
-        api_key = config.resolve_api_key(profile=state.profile)
+        api_key = state.resolve_api_key()
         rows = client.list_transcripts(api_key, limit=limit)
 
         def render(data: list[dict[str, object]]) -> object:
@@ -83,7 +83,7 @@ def body(state: AppState, json_mode: bool) -> None:
         # Cheap local id validation first: a malformed id is a usage error whether
         # or not the user is signed in, so it must not trigger auth/login first.
         client.validate_transcript_id(transcript_id)
-        api_key = config.resolve_api_key(profile=state.profile)
+        api_key = state.resolve_api_key()
         transcript = client.get_transcript(api_key, transcript_id)
         if client.status_str(transcript) == "error":
             raise APIError(

aai_cli/context.py

@@ -18,7 +18,7 @@
 @dataclass
 class AppState:
     """Request-scoped CLI state (the global --profile / --env) and the single place
-    that turns it into a concrete profile, environment, or session.
+    that turns it into a concrete profile, environment, session, or API key.
 
     Centralizing resolution here keeps the precedence rules in one spot instead of
     being re-derived per command. The module-level ``resolve_*`` functions below are
@@ -46,6 +46,11 @@ def resolve_environment(self) -> Environment:
         profile_env = config.get_profile_env(self.resolve_profile())
         return environments.resolve(self.env, profile_env)
 
+    def resolve_api_key(self) -> str:
+        """The API key for SDK/gateway calls: ASSEMBLYAI_API_KEY, else the profile's
+        keyring entry. Raises NotAuthenticated when neither is set."""
+        return config.resolve_api_key(profile=self.profile)
+
     def resolve_session(self) -> tuple[int, str]:
         """Account id + Stytch session JWT for AMS self-service commands.
 

aai_cli/main.py

@@ -19,7 +19,7 @@
     # context type, not the upstream click.Context. Imported for typing only.
     from typer._click.core import Context as ClickContext
 
-from aai_cli import __version__, config, environments, help_panels, output, stdio, theme
+from aai_cli import __version__, argscan, environments, help_panels, output, stdio, theme
 from aai_cli.commands import (
     account,
     agent,
@@ -102,7 +102,7 @@ def list_commands(self, ctx: ClickContext) -> list[str]:
     def parse_args(self, ctx: ClickContext, args: list[str]) -> list[str]:
         # Stash the full token list before anything is parsed, so the root callback can
         # tell whether the (not-yet-parsed) subcommand opted into JSON — see
-        # `_command_line_requests_json`. Recorded here because Click clears the pending
+        # `argscan.requests_json`. Recorded here because Click clears the pending
         # args off the context before the group callback runs.
         ctx.meta[_RAW_ARGS_META_KEY] = list(args)
         return super().parse_args(ctx, args)
@@ -202,7 +202,7 @@ def _click_error_requests_json(err: ClickException) -> bool:
         raw_args: list[str] = ctx.meta[_RAW_ARGS_META_KEY]
     else:
         raw_args = sys.argv[1:]
-    return _command_line_requests_json(raw_args)
+    return argscan.requests_json(raw_args)
 
 
 def _format_click_error_fixed(self: ClickException) -> None:
@@ -265,7 +265,7 @@ def _version_callback(value: bool) -> None:
 
 def _profile_has_key(state: AppState) -> bool:
     try:
-        config.resolve_api_key(profile=state.profile)
+        state.resolve_api_key()
     except NotAuthenticated:
         return False
     return True
@@ -276,27 +276,15 @@ def _interactive_session() -> bool:
     return sys.stdin.isatty() and sys.stdout.isatty()
 
 
+# The root callback runs before the subcommand parses its own ``--json``, so a failure
+# raised there (e.g. a bad ``--env``) would otherwise always render human text — leaving a
+# ``… --json`` pipeline without the uniform ``{"error": …}`` shape it relies on. The group
+# stashes the raw token list in ``ctx.meta`` (see ``_OrderedGroup.parse_args``) before the
+# callback runs, so sniffing it with ``argscan.requests_json`` lets every failure class
+# honor the request.
 _RAW_ARGS_META_KEY = "aai_raw_args"
 
 
-def _command_line_requests_json(raw_args: list[str]) -> bool:
-    """Whether the token list opts into JSON (``--json``, ``-o json``, ``--output json``,
-    or their glued forms).
-
-    The root callback runs before the subcommand parses its own ``--json``, so a failure
-    raised here (e.g. a bad ``--env``) would otherwise always render human text — leaving a
-    ``… --json`` pipeline without the uniform ``{"error": …}`` shape it relies on. The group
-    stashes the raw token list in ``ctx.meta`` (see ``_OrderedGroup.parse_args``) before the
-    callback runs, so sniffing it lets every failure class honor the request.
-    """
-    for index, token in enumerate(raw_args):
-        if token in ("--json", "-j", "--output=json", "-ojson"):
-            return True
-        if token in ("-o", "--output") and raw_args[index + 1 : index + 2] == ["json"]:
-            return True
-    return False
-
-
 def _sandbox_conflict_warning(sandbox: bool, env: str | None) -> str | None:
     """A warning when ``--sandbox`` and a contradictory ``--env`` are both passed.
 
@@ -367,7 +355,7 @@ def main(
     # a root-callback failure (e.g. bad --env) still emits the JSON error shape when the
     # invocation opted into JSON, and renders human text on stderr otherwise.
     raw_args: list[str] = ctx.meta.get(_RAW_ARGS_META_KEY, [])
-    json_mode = output.resolve_json(explicit=_command_line_requests_json(raw_args))
+    json_mode = output.resolve_json(explicit=argscan.requests_json(raw_args))
     conflict_warning = _sandbox_conflict_warning(sandbox, env)
     if sandbox and env is None:
         env = "sandbox000"

aai_cli/telemetry.py

@@ -27,7 +27,7 @@
 
 import typer
 
-from aai_cli import __version__, config
+from aai_cli import __version__, argscan, config
 from aai_cli.errors import CLIError
 
 ENV_DISABLED = "AAI_TELEMETRY_DISABLED"
@@ -104,15 +104,9 @@ def _notice_suppressed(raw_args: list[str]) -> bool:
 
     The one-time disclosure is human-facing chrome: it must not decorate a
     ``--quiet`` run nor pollute the machine-readable stderr a ``--json`` (or
-    ``-o json``) pipeline relies on. Mirrors ``main._command_line_requests_json``
-    (telemetry can't import main without a cycle) plus the quiet flags.
+    ``-o json``) pipeline relies on.
     """
-    for index, token in enumerate(raw_args):
-        if token in ("--quiet", "-q", "--json", "-j", "--output=json", "-ojson"):
-            return True
-        if token in ("-o", "--output") and raw_args[index + 1 : index + 2] == ["json"]:
-            return True
-    return False
+    return any(token in ("--quiet", "-q") for token in raw_args) or argscan.requests_json(raw_args)
 
 
 def _maybe_emit_first_run_notice() -> None:

tests/test_main_module.py

@@ -4,10 +4,11 @@
 import pytest
 
 import aai_cli.main as main_mod
+from aai_cli import argscan
 
 
 def test_command_line_requests_json_recognizes_every_form():
-    f = main_mod._command_line_requests_json
+    f = argscan.requests_json
     assert f(["whoami", "--json"])
     assert f(["transcribe", "a.mp3", "-o", "json"])
     assert f(["transcribe", "a.mp3", "--output", "json"])
@@ -19,7 +20,7 @@ def test_command_line_requests_json_recognizes_every_form():
 
 
 def test_command_line_requests_json_false_for_text_and_bare():
-    f = main_mod._command_line_requests_json
+    f = argscan.requests_json
     assert not f(["transcribe", "a.mp3", "-o", "text"])
     assert not f(["transcribe", "a.mp3"])
     assert not f([])