diff --git a/.importlinter b/.importlinter index 9a666fdc..9013d138 100644 --- a/.importlinter +++ b/.importlinter @@ -19,16 +19,20 @@ source_modules = aai_cli.config_builder aai_cli.context aai_cli.debuglog + aai_cli.deploy_exec + aai_cli.dev_exec aai_cli.dictate_exec aai_cli.dub_exec aai_cli.environments aai_cli.errors aai_cli.eval_data + aai_cli.evaluate_exec aai_cli.follow aai_cli.help_panels aai_cli.help_text aai_cli.hotkey aai_cli.init + aai_cli.init_exec aai_cli.llm aai_cli.llm_exec aai_cli.microphone @@ -37,6 +41,7 @@ source_modules = aai_cli.procs aai_cli.remotefs aai_cli.render + aai_cli.share_exec aai_cli.speak_exec aai_cli.stdio aai_cli.stream_exec diff --git a/aai_cli/commands/deploy.py b/aai_cli/commands/deploy.py index b7dade95..a3a0bf9d 100644 --- a/aai_cli/commands/deploy.py +++ b/aai_cli/commands/deploy.py @@ -1,132 +1,16 @@ # aai_cli/commands/deploy.py from __future__ import annotations -import shutil -import subprocess -import sys -from dataclasses import dataclass -from pathlib import Path - import typer -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 import deploy_exec, help_panels, options +from aai_cli.context import run_command from aai_cli.help_text import examples_epilog -from aai_cli.init import procfile # Flattened single-command sub-typer (same pattern as `assembly dev`). app = typer.Typer() -@dataclass(frozen=True) -class Target: - name: str # human label, e.g. "Vercel" - bin: str # executable resolved via shutil.which - flag: str # CLI selector, e.g. "--vercel" - install: str # hint sentence shown when the CLI is missing (everywhere, or macOS-only) - deploy_args: tuple[str, ...] # subcommand(s) appended after `bin` - supports_prod: bool = False # whether `--prod` adds a production flag - post_deploy_args: tuple[str, ...] | None = None # command run after a successful deploy - install_non_darwin: str | None = None # hint off-macOS, when `install` is brew-specific - - def command(self, *, prod: bool) -> list[str]: - argv = [self.bin, *self.deploy_args] - if prod and self.supports_prod: - argv.append("--prod") - return argv - - -VERCEL = Target( - name="Vercel", - bin="vercel", - flag="--vercel", - install="Install it with `npm i -g vercel`.", - deploy_args=("deploy",), - supports_prod=True, -) -RAILWAY = Target( - name="Railway", - bin="railway", - flag="--railway", - install="Install it with `npm i -g @railway/cli`.", - deploy_args=("up",), - post_deploy_args=("domain",), -) -FLY = Target( - name="Fly", - bin="fly", - flag="--fly", - # brew is macOS-specific; elsewhere point at the official install docs. - install="Install it with `brew install flyctl`.", - install_non_darwin="Install it: https://fly.io/docs/flyctl/install/", - # `fly launch` does it all: creates the app, generates fly.toml (detecting the - # shipped Dockerfile), and deploys — so no fly.toml needs to exist beforehand. - deploy_args=("launch",), -) - -TARGETS = (VERCEL, RAILWAY, FLY) - - -def _resolve_target(selected: list[Target]) -> Target: - if len(selected) > 1: - flags = " / ".join(t.flag for t in TARGETS) - raise UsageError(f"Pass at most one deploy target ({flags}).") - return selected[0] if selected else VERCEL # Vercel is the default - - -def _install_hint(target: Target) -> str: - """The platform-appropriate install hint: brew on macOS, docs URL elsewhere.""" - if target.install_non_darwin is not None and sys.platform != "darwin": - return target.install_non_darwin - return target.install - - -def _require_cli(target: Target) -> None: - if shutil.which(target.bin) is None: - raise CLIError( - f"The {target.name} CLI is required to deploy. {_install_hint(target)}", - error_type="missing_dependency", - exit_code=1, - ) - - -def _confirmed(target: Target, *, assume_yes: bool) -> bool: - """True when the deploy should proceed: --yes, or an interactive yes. - - Refuses to guess in a non-interactive/agent session.""" - if assume_yes: - return True - if output.is_agentic(): - raise UsageError( - "Refusing to deploy without confirmation in a non-interactive session. " - "Pass --yes to deploy." - ) - return typer.confirm(f"Deploy this project to {target.name}?") - - -def run_deploy(*, target: Target, prod: bool, assume_yes: bool, json_mode: bool) -> None: - """Confirm, then run the target's deploy command in the current directory.""" - if prod and not target.supports_prod: - raise UsageError( - "--prod is only supported for Vercel deploys.", - suggestion=f"Drop --prod, or drop {target.flag} to deploy to Vercel.", - ) - # Same not-a-project guard as `assembly dev`/`assembly share`, checked before CLI presence - # so an empty directory says "run `assembly init`", not "install the Vercel CLI". - procfile.require_procfile(Path.cwd()) - _require_cli(target) - if not _confirmed(target, assume_yes=assume_yes): - aborted = {"status": "aborted", "target": target.name} - output.emit(aborted, lambda _d: "Aborted.", json_mode=json_mode) - return - result = subprocess.run(target.command(prod=prod), cwd=Path.cwd(), check=False) - if result.returncode: - raise typer.Exit(code=result.returncode) - if target.post_deploy_args is not None: - subprocess.run([target.bin, *target.post_deploy_args], cwd=Path.cwd(), check=False) - - @app.command( rich_help_panel=help_panels.BUILD, epilog=examples_epilog( @@ -153,11 +37,11 @@ def deploy( `railway up`, or `fly launch`). Requires that target's CLI to be installed. (Render deploys from a connected Git repo — see the project README.) """ - - def body(_state: AppState, json_mode: bool) -> None: - selected = [t for t, on in ((VERCEL, vercel), (RAILWAY, railway), (FLY, fly)) if on] - run_deploy( - target=_resolve_target(selected), prod=prod, assume_yes=assume_yes, json_mode=json_mode - ) - - run_command(ctx, body, json=json_out) + opts = deploy_exec.DeployOptions( + prod=prod, vercel=vercel, railway=railway, fly=fly, assume_yes=assume_yes + ) + run_command( + ctx, + lambda state, json_mode: deploy_exec.run_deploy(opts, state, json_mode=json_mode), + json=json_out, + ) diff --git a/aai_cli/commands/dev.py b/aai_cli/commands/dev.py index e6ba43f6..eb9be5e4 100644 --- a/aai_cli/commands/dev.py +++ b/aai_cli/commands/dev.py @@ -1,59 +1,18 @@ # aai_cli/commands/dev.py from __future__ import annotations -import os -from pathlib import Path - import typer -from rich.markup import escape -from aai_cli import help_panels, options, output, steps -from aai_cli.context import AppState, run_command +from aai_cli import dev_exec, help_panels, options +from aai_cli.context import run_command from aai_cli.help_text import examples_epilog -from aai_cli.init import devserver, procfile, runner +from aai_cli.init import devserver # Flattened single-command sub-typer (same pattern as `assembly init`): one # @app.command() registered via app.add_typer(dev.app) with no name. app = typer.Typer() -def run_dev( - *, port: int, host: str, no_install: bool, no_open: bool, json_mode: bool, quiet: bool -) -> None: - """Boot the project's Procfile `web:` process locally, with live reload.""" - target = Path.cwd() - use_uv = runner.has_uv() - - chosen_port = runner.find_free_port(port) - devserver.notify_port_change(port, chosen_port, json_mode=json_mode, quiet=quiet) - env = {**os.environ, "PORT": str(chosen_port)} - # Resolves the start command AND validates we're inside a scaffolded project. - web = procfile.web_argv(target, env=env) - - report: list[steps.Step] = [ - devserver.install_step(target, no_install=no_install, use_uv=use_uv) - ] - output.emit(report, lambda d: steps.render_steps(d, heading="Dev"), json_mode=json_mode) - if any(s["status"] == "failed" for s in report): - raise typer.Exit(code=1) - - command = devserver.dev_command(target, web, use_uv=use_uv, host=host) - # The printed URL reflects the actual bind: "localhost" for the loopback - # default, the literal host for an explicit --host. - url_host = "localhost" if host == devserver.LOCAL_HOST else host - url = f"http://{url_host}:{chosen_port}" - if not json_mode: - output.console.print( - f"[aai.heading]Starting[/aai.heading] [aai.url]{escape(url)}[/aai.url]" - " [aai.muted](Ctrl-C to stop)[/aai.muted]" - ) - code = runner.run_server( - target, command=command, port=chosen_port, env=env, open_browser=not no_open - ) - if code: - raise typer.Exit(code=code) - - @app.command( rich_help_panel=help_panels.BUILD, epilog=examples_epilog( @@ -84,15 +43,9 @@ def dev( Run this from inside a project created by `assembly init`. It installs dependencies if needed, then starts the FastAPI server with live reload and opens the browser. """ - - def body(state: AppState, json_mode: bool) -> None: - run_dev( - port=port, - host=host, - no_install=no_install, - no_open=no_open, - json_mode=json_mode, - quiet=state.quiet, - ) - - run_command(ctx, body, json=json_out) + opts = dev_exec.DevOptions(port=port, host=host, no_install=no_install, no_open=no_open) + run_command( + ctx, + lambda state, json_mode: dev_exec.run_dev(opts, state, json_mode=json_mode), + json=json_out, + ) diff --git a/aai_cli/commands/evaluate.py b/aai_cli/commands/evaluate.py index 2a04b511..1ac7381a 100644 --- a/aai_cli/commands/evaluate.py +++ b/aai_cli/commands/evaluate.py @@ -1,205 +1,22 @@ """`assembly eval` — transcribe an evaluation dataset and score it against references. -WER (via jiwer) against the dataset's reference texts. The module is named -``evaluate`` because importing a module named ``eval`` would shadow the builtin; -the command itself registers as ``eval``. +The module is named ``evaluate`` because importing a module named ``eval`` would +shadow the builtin; the command itself registers as ``eval``. The scoring/render +logic lives in ``aai_cli.evaluate_exec`` (the options/run split, see AGENTS.md). """ from __future__ import annotations -from concurrent.futures import ThreadPoolExecutor, as_completed -from dataclasses import dataclass -from enum import StrEnum - -import assemblyai as aai import typer -from rich.console import RenderableType -from aai_cli import client, eval_data, help_panels, jsonshape, options, output, wer -from aai_cli.context import AppState, run_command -from aai_cli.errors import CLIError, NotAuthenticated +from aai_cli import evaluate_exec, help_panels, options +from aai_cli.context import run_command +from aai_cli.evaluate_exec import EvalSpeechModel from aai_cli.help_text import examples_epilog app = typer.Typer() -class EvalSpeechModel(StrEnum): - """The current-generation models, requested via the SDK's ``speech_models`` - list parameter (its legacy ``SpeechModel`` enum predates them).""" - - universal_3_pro = "universal-3-pro" - universal_2 = "universal-2" - - -def _pct(value: object) -> str: - return f"{jsonshape.as_float(value):.2%}" - - -@dataclass(frozen=True) -class _ItemResult: - """One scored row: the emitted dict plus the score kept for pooling.""" - - row: dict[str, object] - words: wer.Score | None - - -def _failed_result(item: eval_data.EvalItem, err: CLIError) -> _ItemResult: - """A row whose transcription failed: the error rides along, no scores pooled.""" - return _ItemResult(row={"item": item.item_id, "error": err.message}, words=None) - - -def _score_item(item: eval_data.EvalItem, transcript: aai.Transcript) -> _ItemResult: - words = wer.score(item.reference, str(transcript.text or "")) - row: dict[str, object] = { - "item": item.item_id, - "words": words.words, - "errors": words.errors, - "wer": words.wer, - } - return _ItemResult(row=row, words=words) - - -def _pooled_metrics(results: list[_ItemResult]) -> dict[str, object]: - """The summary scores pooled over the scored rows (failed rows carry none).""" - metrics: dict[str, object] = {} - word_scores = [result.words for result in results if result.words is not None] - if word_scores: - total = wer.pooled(word_scores) - metrics.update({"words": total.words, "errors": total.errors, "wer": total.wer}) - return metrics - - -def _transcribe_one( - api_key: str, item: eval_data.EvalItem, config: aai.TranscriptionConfig -) -> aai.Transcript | CLIError: - """One item's outcome: its transcript, or the CLIError it failed with. - - A bad item must not discard the other (paid) items, so per-item failures - are recorded rather than raised — except ``NotAuthenticated`` (one rejected - key fails every row identically) and non-CLIError bugs, which propagate and - abort the run. - """ - try: - return client.transcribe(api_key, item.audio, config=config) - except NotAuthenticated: - raise - except CLIError as err: - return err - - -def _concurrent_transcripts( - api_key: str, - items: list[eval_data.EvalItem], - *, - transcription_config: aai.TranscriptionConfig, - concurrency: int, -) -> list[aai.Transcript | CLIError]: - with ThreadPoolExecutor(max_workers=concurrency) as pool: - futures = [ - pool.submit(_transcribe_one, api_key, item, transcription_config) for item in items - ] - for future in as_completed(futures): - if (exc := future.exception()) is not None: - # Only aborting failures escape _transcribe_one: drop the - # not-yet-started items rather than burn an API call on each. - pool.shutdown(cancel_futures=True) - raise exc - return [future.result() for future in futures] - - -def _transcripts( - api_key: str, - items: list[eval_data.EvalItem], - *, - transcription_config: aai.TranscriptionConfig, - concurrency: int, - json_mode: bool, - quiet: bool, -) -> list[aai.Transcript | CLIError]: - """Each item's transcript — or the CLIError it failed with — in dataset order. - - Sequential by default, with a per-item spinner; ``--concurrency`` fans the - API calls out across a thread pool (see ``_transcribe_one`` for which - failures are per-item outcomes and which abort the run). - """ - if concurrency == 1: - outcomes: list[aai.Transcript | CLIError] = [] - for index, item in enumerate(items, start=1): - with output.status( - f"[{index}/{len(items)}] Transcribing {item.item_id}…", - json_mode=json_mode, - quiet=quiet, - ): - outcomes.append(_transcribe_one(api_key, item, transcription_config)) - return outcomes - with output.status( - f"Transcribing {len(items)} items (concurrency {concurrency})…", - json_mode=json_mode, - quiet=quiet, - ): - return _concurrent_transcripts( - api_key, items, transcription_config=transcription_config, concurrency=concurrency - ) - - -def _payload( - label: str, speech_model: EvalSpeechModel | None, results: list[_ItemResult] -) -> dict[str, object]: - payload: dict[str, object] = { - "dataset": label, - "speech_model": speech_model.value if speech_model else None, - "items": len(results), - "rows": [result.row for result in results], - } - payload.update(_pooled_metrics(results)) - failed = sum(1 for result in results if "error" in result.row) - if failed: - payload["failed"] = failed - return payload - - -def _summary(payload: dict[str, object]) -> str: - parts: list[str] = [] - if "wer" in payload: - errors = jsonshape.as_int(payload.get("errors")) - noun = "error" if errors == 1 else "errors" - parts.append( - f"WER {_pct(payload.get('wer'))} ({errors} {noun} / {payload.get('words')} words)" - ) - return output.heading(" ".join(parts)) - - -def _cell(row: dict[str, object], key: str) -> str: - """The row's value as table text — blank when absent (e.g. a failed row's scores).""" - return str(row[key]) if key in row else "" - - -def _pct_cell(row: dict[str, object], key: str) -> str: - return _pct(row[key]) if key in row else "" - - -def _render(payload: dict[str, object]) -> RenderableType: - has_wer = "wer" in payload - has_failed = "failed" in payload - columns = [ - "ITEM", - *(["WORDS", "ERRORS", "WER"] if has_wer else []), - *(["ERROR"] if has_failed else []), - ] - table = output.data_table(*columns) - for row in jsonshape.mapping_list(payload.get("rows")): - cells = [str(row.get("item"))] - if has_wer: - cells += [_cell(row, "words"), _cell(row, "errors"), _pct_cell(row, "wer")] - if has_failed: - cells.append(_cell(row, "error")) - table.add_row(*cells) - model = payload.get("speech_model") or "default model" - return output.stack( - output.muted(f"{payload.get('dataset')} · {model}"), table, _summary(payload) - ) - - @app.command( name="eval", rich_help_panel=help_panels.TRANSCRIPTION, @@ -276,49 +93,19 @@ def evaluate( (parliament speech), switchboard (phone calls), expresso (expressive speech), loquacious, and callhome (phone calls). """ - - 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 = state.resolve_api_key() - data = eval_data.load( - dataset, - split=split, - subset=subset, - audio_column=audio_column, - text_column=text_column, - limit=limit, - ) - transcription_config = aai.TranscriptionConfig( - speech_models=[speech_model.value] if speech_model else None, - language_code=language_code, - ) - outcomes = _transcripts( - api_key, - data.items, - transcription_config=transcription_config, - concurrency=concurrency, - json_mode=json_mode, - quiet=state.quiet, - ) - results = [ - _failed_result(item, outcome) - if isinstance(outcome, CLIError) - else _score_item(item, outcome) - for item, outcome in zip( - data.items, - outcomes, - strict=True, # pragma: no mutate (defensive invariant; _transcripts returns one outcome per item) - ) - ] - payload = _payload(data.label, speech_model, results) - output.emit(payload, _render, json_mode=json_mode) - failed = jsonshape.as_int(payload.get("failed")) - if failed: - raise CLIError( - f"{failed} of {len(results)} items failed to transcribe.", - error_type="eval_failed", - suggestion="The summary covers only the items that transcribed.", - ) - - run_command(ctx, body, json=json_out) + opts = evaluate_exec.EvalOptions( + dataset=dataset, + split=split, + subset=subset, + limit=limit, + audio_column=audio_column, + text_column=text_column, + speech_model=speech_model, + language_code=language_code, + concurrency=concurrency, + ) + run_command( + ctx, + lambda state, json_mode: evaluate_exec.run_evaluate(opts, state, json_mode=json_mode), + json=json_out, + ) diff --git a/aai_cli/commands/init.py b/aai_cli/commands/init.py index 705a9bcf..f23c8d63 100644 --- a/aai_cli/commands/init.py +++ b/aai_cli/commands/init.py @@ -1,18 +1,12 @@ # aai_cli/commands/init.py from __future__ import annotations -from pathlib import Path - import typer -from rich.markup import escape -from aai_cli import __version__, environments, help_panels, options, output, stdio, steps +from aai_cli import help_panels, init_exec, options from aai_cli.context import AppState, run_command -from aai_cli.errors import CLIError, UsageError from aai_cli.help_text import examples_epilog -from aai_cli.init import keys, runner, scaffold, templates - -_DEFAULT_PORT = 3000 +from aai_cli.init import templates # Single-command sub-typer flattened to `assembly init` (the exact pattern `assembly transcribe` # uses): one @app.command() named `init`, registered via app.add_typer(init.app) with @@ -20,271 +14,6 @@ app = typer.Typer() -def _pick_template() -> str: - """Interactive picker; raises a usage error when there's no TTY to prompt on.""" - if not stdio.interactive_stdio(): - raise CLIError( - "No template given and not running interactively. " - f"Pass one of: {', '.join(templates.TEMPLATE_ORDER)}.", - error_type="usage_error", - exit_code=1, - ) - try: - import questionary - except ImportError as exc: # a broken/stale install missing the declared dep - raise CLIError( - "The interactive picker needs 'questionary'. Reinstall the CLI " - "(e.g. `uv tool install --reinstall aai-cli`), or pass a template " - f"directly: {', '.join(templates.TEMPLATE_ORDER)}.", - error_type="missing_dependency", - exit_code=1, - ) from exc - - choice = questionary.select( - "Pick a template", - choices=[ - questionary.Choice(title=templates.title_for(t), value=t) - for t in templates.TEMPLATE_ORDER - ], - ).ask() - if choice is None: # user pressed Ctrl-C - raise typer.Exit(code=130) - return str(choice) - - -def _resolve_dir(directory: str | None, template: str, *, here: bool) -> Path: - if here: - return Path.cwd() - if directory: - return Path(directory) - return Path.cwd() / template - - -def _resolve_template(template: str | None) -> str: - """Resolve the template name: the picker when omitted, else validate the arg.""" - chosen = template if template is not None else _pick_template() - if not templates.is_template(chosen): - raise CLIError( - f"Unknown template {chosen!r}. Choose one of: {', '.join(templates.TEMPLATE_ORDER)}.", - error_type="usage_error", - exit_code=1, - ) - return chosen - - -def _active_env_vars() -> dict[str, str]: - """Pin the scaffolded app to the active environment's hosts. - - A sandbox key (minted by `assembly login` against a non-prod env) would otherwise be - rejected by the production defaults baked into the template. - """ - env = environments.active() - return { - "ASSEMBLYAI_BASE_URL": env.api_base, - "ASSEMBLYAI_LLM_GATEWAY_URL": env.llm_gateway_base, - "ASSEMBLYAI_STREAMING_HOST": env.streaming_host, - # Voice Agent host mirrors the streaming host's naming across environments. - "ASSEMBLYAI_AGENTS_HOST": env.streaming_host.replace("streaming", "agents", 1), - } - - -def _install_step( - target: Path, *, no_install: bool, api_key: str | None, use_uv: bool -) -> tuple[list[steps.Step], bool]: - """Run (or skip) dependency install, returning the report rows and whether to launch. - - Launch only happens when deps are installed and there's a key; an install failure - flips `will_launch` off so the caller exits non-zero instead of starting a server. - """ - will_launch = not no_install and api_key is not None - if no_install: - return [{"name": "install", "status": "skipped", "detail": "--no-install"}], will_launch - setup = runner.run_setup(target, use_uv=use_uv) - if setup.returncode != 0: - row: steps.Step = { - "name": "install", - "status": "failed", - "detail": (setup.stderr or setup.stdout).strip()[:300], - } - # The False (don't-launch) is an equivalent mutant: run_init raises Exit(1) on - # any failed step before it ever consults will_launch, so the value is unused - # on this branch. - return [row], False # pragma: no mutate - return [ - { - "name": "install", - "status": "installed", - "detail": "uv" if use_uv else "venv + pip", - } - ], will_launch - - -def _resolve_target( - directory: str | None, chosen: str, *, here: bool, force: bool -) -> tuple[Path, bool]: - """Resolve the target directory, rejecting --here+DIRECTORY, an existing file, or - a non-empty conflict. Returns the target and whether --force is overlaying it.""" - if here and directory: - raise CLIError( - "Pass either a DIRECTORY or --here, not both.", - error_type="usage_error", - exit_code=1, - ) - target = _resolve_dir(directory, chosen, here=here) - if target.exists() and not target.is_dir(): - raise UsageError(f"{target} exists and is not a directory.") - conflict = scaffold.target_conflict(target) - if conflict and not force: - raise CLIError( - f"{target} already exists and is not empty. " - f"Use --force to overwrite or pick another directory.", - error_type="usage_error", - exit_code=1, - ) - return target, conflict - - -def _key_row(api_key: str | None, key_source: str | None, preserved: str | None) -> steps.Step: - """The report's `key` row — emitted symmetrically whether a key resolved or not.""" - if api_key is not None: - # Literal branches rather than interpolating key_source: it rode in the same - # return tuple as the API key, so CodeQL's coarse tuple taint marks it - # sensitive and flags the report emit (py/clear-text-logging-sensitive-data). - detail = "from environment" if key_source == "environment" else "from keyring" - return {"name": "key", "status": "written", "detail": detail} - if preserved is not None: - return {"name": "key", "status": "kept", "detail": "existing .env key preserved"} - return { - "name": "key", - "status": "skipped", - "detail": "no API key found; wrote a placeholder to .env (run `assembly login`)", - } - - -def _scaffold_report( - chosen: str, - target: Path, - *, - api_key: str | None, - key_source: str | None, - preserved: str | None, -) -> list[steps.Step]: - """Write the template to `target` and return the opening report rows.""" - scaffold.scaffold(chosen, target, api_key=api_key or preserved, env_vars=_active_env_vars()) - return [ - {"name": "scaffold", "status": "created", "detail": str(target)}, - _key_row(api_key, key_source, preserved), - ] - - -def _dev_hint(port: int) -> str: - """The `assembly dev` invocation matching the chosen port (the default needs no flag).""" - return "assembly dev" if port == _DEFAULT_PORT else f"assembly dev --port {port}" - - -def launch_app(target: Path, *, port: int, use_uv: bool, no_open: bool, json_mode: bool) -> None: - """Start the scaffolded app on a free port and open the browser, then block. - - Public (not underscore-private) because the onboarding wizard launches the - scaffolded app as its final step, after the remaining wizard sections have run. - """ - chosen_port = runner.find_free_port(port) - url = f"http://localhost:{chosen_port}" - if not json_mode: - output.console.print( - f"[aai.heading]Starting[/aai.heading] [aai.url]{escape(url)}[/aai.url]" - " [aai.muted](Ctrl-C to stop)[/aai.muted]" - ) - code = runner.launch_and_open(target, port=chosen_port, use_uv=use_uv, open_browser=not no_open) - if code: - raise typer.Exit(code=code) - - -def _build_report( - state: AppState, chosen: str, target: Path, *, no_install: bool, use_uv: bool, port: int -) -> tuple[list[steps.Step], bool]: - """Scaffold and assemble the report rows; returns them plus whether to launch.""" - api_key, key_source = keys.resolve_optional_api_key(profile=state.profile) - # A configured (non-placeholder) .env key must survive a re-scaffold when no key - # resolves — otherwise --force would silently reset it to the placeholder. - preserved = scaffold.existing_env_key(target) if api_key is None else None - effective_key = api_key or preserved - report = _scaffold_report( - chosen, target, api_key=api_key, key_source=key_source, preserved=preserved - ) - - install_rows, will_launch = _install_step( - target, no_install=no_install, api_key=effective_key, use_uv=use_uv - ) - report.extend(install_rows) - - # Deps are installed but there's no key, so the server can't start — say so - # rather than exiting silently. - if not no_install and effective_key is None: - report.append( - { - "name": "launch", - "status": "skipped", - "detail": f"no API key; run `assembly login`, then: cd {target} && {_dev_hint(port)}", - } - ) - return report, will_launch - - -def run_init( - state: AppState, - *, - template: str | None, - directory: str | None, - no_install: bool, - no_open: bool, - force: bool, - here: bool, - port: int, - json_mode: bool, - launch: bool = True, -) -> Path: - """Scaffold (and optionally install/launch) a template; return the target dir. - - `launch=False` is for callers like the onboarding wizard that must not block on a - running dev server mid-flow — it stops after install and leaves the run command as - a hint (the wizard calls `launch_app` itself once its remaining sections are done). - """ - chosen = _resolve_template(template) - target, overwriting = _resolve_target(directory, chosen, here=here, force=force) - if not json_mode: - # Vercel-style banner, printed only once validation passes so pure error runs - # (unknown template, conflicting target) stay undecorated like the sibling - # commands. Decoration goes to stderr (data → stdout): it must never pollute - # a piped stdout. - output.error_console.print( - f"[aai.heading]AssemblyAI CLI[/aai.heading] [aai.muted]{__version__}[/aai.muted]" - ) - if overwriting: - output.emit_warning( - f"--force: overwriting existing files in {target} " - "(the template is overlaid; files not in the template are kept).", - json_mode=json_mode, - ) - - use_uv = runner.has_uv() - report, will_launch = _build_report( - state, chosen, target, no_install=no_install, use_uv=use_uv, port=port - ) - - output.emit(report, lambda d: steps.render_steps(d, heading="Setup"), json_mode=json_mode) - if any(s["status"] == "failed" for s in report): - raise typer.Exit(code=1) - - if launch and will_launch: - launch_app(target, port=port, use_uv=use_uv, no_open=no_open, json_mode=json_mode) - elif not json_mode: - # Scaffolded but not launched (no key, or --no-install, or launch=False): leave the - # user with the one command that starts their app, the way `vercel`/`supabase` sign off. - output.console.print(output.hint(f"Run `cd {escape(str(target))} && {_dev_hint(port)}`.")) - return target - - @app.command( rich_help_panel=help_panels.BUILD, epilog=examples_epilog( @@ -330,7 +59,7 @@ def init( ), ), here: bool = typer.Option(False, "--here", help="Scaffold into the current directory."), - port: int = typer.Option(_DEFAULT_PORT, "--port", help="Local server port."), + port: int = typer.Option(init_exec.DEFAULT_PORT, "--port", help="Local server port."), json_out: bool = options.json_option(), ) -> None: """Scaffold a new project from a template, then launch it. @@ -339,18 +68,19 @@ def init( ('assembly init voice-agent'). The 'assembly agent' command only runs a live mic conversation and writes no code. """ + opts = init_exec.InitOptions( + template=template, + directory=directory, + no_install=no_install, + no_open=no_open, + force=force, + here=here, + port=port, + ) + # run_init returns the scaffolded path (for the onboarding wizard); the command + # path discards it, so a thin body adapts the run_command (-> None) signature. def body(state: AppState, json_mode: bool) -> None: - run_init( - state, - template=template, - directory=directory, - no_install=no_install, - no_open=no_open, - force=force, - here=here, - port=port, - json_mode=json_mode, - ) + init_exec.run_init(opts, state, json_mode=json_mode) run_command(ctx, body, json=json_out) diff --git a/aai_cli/commands/share.py b/aai_cli/commands/share.py index 83291380..6e564a49 100644 --- a/aai_cli/commands/share.py +++ b/aai_cli/commands/share.py @@ -1,88 +1,16 @@ # aai_cli/commands/share.py from __future__ import annotations -import os -from pathlib import Path - import typer -from rich.markup import escape -from aai_cli import help_panels, options, output, steps -from aai_cli.context import AppState, run_command -from aai_cli.errors import CLIError +from aai_cli import help_panels, options, share_exec +from aai_cli.context import run_command from aai_cli.help_text import examples_epilog -from aai_cli.init import devserver, procfile, runner, tunnel # Flattened single-command sub-typer (same pattern as `assembly dev`). app = typer.Typer() -def _render_share(data: dict[str, object]) -> str: - return ( - f"[aai.heading]Sharing[/aai.heading] [aai.url]{escape(str(data['url']))}[/aai.url]\n" - f"[aai.muted]→ serving[/aai.muted] [aai.url]{escape(str(data['local']))}[/aai.url]" - " [aai.muted](Ctrl-C to stop)[/aai.muted]" - ) - - -def run_share(*, port: int, no_install: bool, json_mode: bool, quiet: bool) -> None: - """Boot the app and expose it on a public cloudflared quick-tunnel URL.""" - target = Path.cwd() - use_uv = runner.has_uv() - - chosen_port = runner.find_free_port(port) - devserver.notify_port_change(port, chosen_port, json_mode=json_mode, quiet=quiet) - env = {**os.environ, "PORT": str(chosen_port)} - web = procfile.web_argv(target, env=env) # validates we're in a scaffolded project - tunnel.require_cloudflared("share a public link") - - report: list[steps.Step] = [ - devserver.install_step(target, no_install=no_install, use_uv=use_uv) - ] - output.emit(report, lambda d: steps.render_steps(d, heading="Share"), json_mode=json_mode) - if any(s["status"] == "failed" for s in report): - raise typer.Exit(code=1) - - server = runner.spawn(devserver.dev_command(target, web, use_uv=use_uv), cwd=target, env=env) - proxy = None - log_path: Path | None = None - keep_log = False - try: - if not runner.wait_for_port(chosen_port): - raise CLIError( - "The dev server didn't start, so there's nothing to share.", - error_type="server_error", - exit_code=1, - ) - proxy, public, log_path = tunnel.open_quick_tunnel(chosen_port, cwd=target) - if public is None: - # Keep the captured cloudflared output: it's the only evidence of why - # the tunnel never came up. - keep_log = True - raise CLIError( - "cloudflared didn't report a tunnel URL in time.", - error_type="tunnel_error", - exit_code=1, - suggestion=f"cloudflared's output was kept at {log_path} — check it for errors.", - ) - payload: dict[str, object] = { - "url": public, - "local": f"http://localhost:{chosen_port}", - "port": chosen_port, - } - output.emit(payload, _render_share, json_mode=json_mode) - server.wait() - except KeyboardInterrupt: - # Ctrl-C is the expected way to stop a foreground share; the finally - # block below tears down the tunnel and server. - pass - finally: - tunnel.terminate(proxy) - tunnel.terminate(server) - if log_path is not None and not keep_log: - log_path.unlink(missing_ok=True) - - @app.command( rich_help_panel=help_panels.BUILD, epilog=examples_epilog( @@ -108,8 +36,9 @@ def share( URL. Requires cloudflared (macOS: `brew install cloudflared`; other platforms: https://developers.cloudflare.com/cloudflare-one/connections/connect-networks/downloads/). """ - - def body(state: AppState, json_mode: bool) -> None: - run_share(port=port, no_install=no_install, json_mode=json_mode, quiet=state.quiet) - - run_command(ctx, body, json=json_out) + opts = share_exec.ShareOptions(port=port, no_install=no_install) + run_command( + ctx, + lambda state, json_mode: share_exec.run_share(opts, state, json_mode=json_mode), + json=json_out, + ) diff --git a/aai_cli/deploy_exec.py b/aai_cli/deploy_exec.py new file mode 100644 index 00000000..f99f468b --- /dev/null +++ b/aai_cli/deploy_exec.py @@ -0,0 +1,147 @@ +"""Run logic for `assembly deploy`: ship the current project to a PaaS target. + +The command module (aai_cli/commands/deploy.py) only parses argv — it builds a +``DeployOptions`` and hands it to ``run_deploy`` via ``context.run_command`` (the +options/run split, see AGENTS.md), so tests drive target resolution and the +subprocess orchestration by constructing options directly instead of +round-tripping argv. +""" + +from __future__ import annotations + +import shutil +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path + +import typer + +from aai_cli import output +from aai_cli.context import AppState +from aai_cli.errors import CLIError, UsageError +from aai_cli.init import procfile + + +@dataclass(frozen=True) +class Target: + name: str # human label, e.g. "Vercel" + bin: str # executable resolved via shutil.which + flag: str # CLI selector, e.g. "--vercel" + install: str # hint sentence shown when the CLI is missing (everywhere, or macOS-only) + deploy_args: tuple[str, ...] # subcommand(s) appended after `bin` + supports_prod: bool = False # whether `--prod` adds a production flag + post_deploy_args: tuple[str, ...] | None = None # command run after a successful deploy + install_non_darwin: str | None = None # hint off-macOS, when `install` is brew-specific + + def command(self, *, prod: bool) -> list[str]: + argv = [self.bin, *self.deploy_args] + if prod and self.supports_prod: + argv.append("--prod") + return argv + + +VERCEL = Target( + name="Vercel", + bin="vercel", + flag="--vercel", + install="Install it with `npm i -g vercel`.", + deploy_args=("deploy",), + supports_prod=True, +) +RAILWAY = Target( + name="Railway", + bin="railway", + flag="--railway", + install="Install it with `npm i -g @railway/cli`.", + deploy_args=("up",), + post_deploy_args=("domain",), +) +FLY = Target( + name="Fly", + bin="fly", + flag="--fly", + # brew is macOS-specific; elsewhere point at the official install docs. + install="Install it with `brew install flyctl`.", + install_non_darwin="Install it: https://fly.io/docs/flyctl/install/", + # `fly launch` does it all: creates the app, generates fly.toml (detecting the + # shipped Dockerfile), and deploys — so no fly.toml needs to exist beforehand. + deploy_args=("launch",), +) + +TARGETS = (VERCEL, RAILWAY, FLY) + + +@dataclass(frozen=True) +class DeployOptions: + """Every `assembly deploy` flag as plain data (``--json`` excluded: run_command + resolves it into the ``json_mode`` argument).""" + + prod: bool + vercel: bool + railway: bool + fly: bool + assume_yes: bool + + +def _resolve_target(opts: DeployOptions) -> Target: + selected = [ + t for t, on in ((VERCEL, opts.vercel), (RAILWAY, opts.railway), (FLY, opts.fly)) if on + ] + if len(selected) > 1: + flags = " / ".join(t.flag for t in TARGETS) + raise UsageError(f"Pass at most one deploy target ({flags}).") + return selected[0] if selected else VERCEL # Vercel is the default + + +def _install_hint(target: Target) -> str: + """The platform-appropriate install hint: brew on macOS, docs URL elsewhere.""" + if target.install_non_darwin is not None and sys.platform != "darwin": + return target.install_non_darwin + return target.install + + +def _require_cli(target: Target) -> None: + if shutil.which(target.bin) is None: + raise CLIError( + f"The {target.name} CLI is required to deploy. {_install_hint(target)}", + error_type="missing_dependency", + exit_code=1, + ) + + +def _confirmed(target: Target, *, assume_yes: bool) -> bool: + """True when the deploy should proceed: --yes, or an interactive yes. + + Refuses to guess in a non-interactive/agent session.""" + if assume_yes: + return True + if output.is_agentic(): + raise UsageError( + "Refusing to deploy without confirmation in a non-interactive session. " + "Pass --yes to deploy." + ) + return typer.confirm(f"Deploy this project to {target.name}?") + + +def run_deploy(opts: DeployOptions, _state: AppState, *, json_mode: bool) -> None: + """Confirm, then run the target's deploy command in the current directory.""" + target = _resolve_target(opts) + if opts.prod and not target.supports_prod: + raise UsageError( + "--prod is only supported for Vercel deploys.", + suggestion=f"Drop --prod, or drop {target.flag} to deploy to Vercel.", + ) + # Same not-a-project guard as `assembly dev`/`assembly share`, checked before CLI presence + # so an empty directory says "run `assembly init`", not "install the Vercel CLI". + procfile.require_procfile(Path.cwd()) + _require_cli(target) + if not _confirmed(target, assume_yes=opts.assume_yes): + aborted = {"status": "aborted", "target": target.name} + output.emit(aborted, lambda _d: "Aborted.", json_mode=json_mode) + return + result = subprocess.run(target.command(prod=opts.prod), cwd=Path.cwd(), check=False) + if result.returncode: + raise typer.Exit(code=result.returncode) + if target.post_deploy_args is not None: + subprocess.run([target.bin, *target.post_deploy_args], cwd=Path.cwd(), check=False) diff --git a/aai_cli/dev_exec.py b/aai_cli/dev_exec.py new file mode 100644 index 00000000..a683bdbc --- /dev/null +++ b/aai_cli/dev_exec.py @@ -0,0 +1,66 @@ +"""Run logic for `assembly dev`: boot the scaffolded project's dev server. + +The command module (aai_cli/commands/dev.py) only parses argv — it builds a +``DevOptions`` and hands it to ``run_dev`` via ``context.run_command`` (the +options/run split, see AGENTS.md), so tests drive the server orchestration by +constructing options directly instead of round-tripping argv. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path + +import typer +from rich.markup import escape + +from aai_cli import output, steps +from aai_cli.context import AppState +from aai_cli.init import devserver, procfile, runner + + +@dataclass(frozen=True) +class DevOptions: + """Every `assembly dev` flag as plain data (``--json`` excluded: run_command + resolves it into the ``json_mode`` argument).""" + + port: int + host: str + no_install: bool + no_open: bool + + +def run_dev(opts: DevOptions, state: AppState, *, json_mode: bool) -> None: + """Boot the project's Procfile `web:` process locally, with live reload.""" + target = Path.cwd() + use_uv = runner.has_uv() + + chosen_port = runner.find_free_port(opts.port) + devserver.notify_port_change(opts.port, chosen_port, json_mode=json_mode, quiet=state.quiet) + env = {**os.environ, "PORT": str(chosen_port)} + # Resolves the start command AND validates we're inside a scaffolded project. + web = procfile.web_argv(target, env=env) + + report: list[steps.Step] = [ + devserver.install_step(target, no_install=opts.no_install, use_uv=use_uv) + ] + output.emit(report, lambda d: steps.render_steps(d, heading="Dev"), json_mode=json_mode) + if any(s["status"] == "failed" for s in report): + raise typer.Exit(code=1) + + command = devserver.dev_command(target, web, use_uv=use_uv, host=opts.host) + # The printed URL reflects the actual bind: "localhost" for the loopback + # default, the literal host for an explicit --host. + url_host = "localhost" if opts.host == devserver.LOCAL_HOST else opts.host + url = f"http://{url_host}:{chosen_port}" + if not json_mode: + output.console.print( + f"[aai.heading]Starting[/aai.heading] [aai.url]{escape(url)}[/aai.url]" + " [aai.muted](Ctrl-C to stop)[/aai.muted]" + ) + code = runner.run_server( + target, command=command, port=chosen_port, env=env, open_browser=not opts.no_open + ) + if code: + raise typer.Exit(code=code) diff --git a/aai_cli/evaluate_exec.py b/aai_cli/evaluate_exec.py new file mode 100644 index 00000000..aa4fe8a9 --- /dev/null +++ b/aai_cli/evaluate_exec.py @@ -0,0 +1,263 @@ +"""Run logic for `assembly eval`: transcribe a dataset and score WER against references. + +The command module (aai_cli/commands/evaluate.py) only parses argv — it builds an +``EvalOptions`` and hands it to ``run_evaluate`` via ``context.run_command`` (the +options/run split, see AGENTS.md), so tests drive the scoring and rendering by +constructing options directly instead of round-tripping argv. + +WER (via jiwer) against the dataset's reference texts. The sibling module is named +``evaluate`` because importing a module named ``eval`` would shadow the builtin; +the command itself registers as ``eval``. +""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import dataclass +from enum import StrEnum + +import assemblyai as aai +from rich.console import RenderableType + +from aai_cli import client, eval_data, jsonshape, output, wer +from aai_cli.context import AppState +from aai_cli.errors import CLIError, NotAuthenticated + + +class EvalSpeechModel(StrEnum): + """The current-generation models, requested via the SDK's ``speech_models`` + list parameter (its legacy ``SpeechModel`` enum predates them).""" + + universal_3_pro = "universal-3-pro" + universal_2 = "universal-2" + + +@dataclass(frozen=True) +class EvalOptions: + """Every `assembly eval` flag as plain data (``--json`` excluded: run_command + resolves it into the ``json_mode`` argument).""" + + dataset: str + split: str | None + subset: str | None + limit: int + audio_column: str | None + text_column: str | None + speech_model: EvalSpeechModel | None + language_code: str | None + concurrency: int + + +def _pct(value: object) -> str: + return f"{jsonshape.as_float(value):.2%}" + + +@dataclass(frozen=True) +class _ItemResult: + """One scored row: the emitted dict plus the score kept for pooling.""" + + row: dict[str, object] + words: wer.Score | None + + +def _failed_result(item: eval_data.EvalItem, err: CLIError) -> _ItemResult: + """A row whose transcription failed: the error rides along, no scores pooled.""" + return _ItemResult(row={"item": item.item_id, "error": err.message}, words=None) + + +def _score_item(item: eval_data.EvalItem, transcript: aai.Transcript) -> _ItemResult: + words = wer.score(item.reference, str(transcript.text or "")) + row: dict[str, object] = { + "item": item.item_id, + "words": words.words, + "errors": words.errors, + "wer": words.wer, + } + return _ItemResult(row=row, words=words) + + +def _pooled_metrics(results: list[_ItemResult]) -> dict[str, object]: + """The summary scores pooled over the scored rows (failed rows carry none).""" + metrics: dict[str, object] = {} + word_scores = [result.words for result in results if result.words is not None] + if word_scores: + total = wer.pooled(word_scores) + metrics.update({"words": total.words, "errors": total.errors, "wer": total.wer}) + return metrics + + +def _transcribe_one( + api_key: str, item: eval_data.EvalItem, config: aai.TranscriptionConfig +) -> aai.Transcript | CLIError: + """One item's outcome: its transcript, or the CLIError it failed with. + + A bad item must not discard the other (paid) items, so per-item failures + are recorded rather than raised — except ``NotAuthenticated`` (one rejected + key fails every row identically) and non-CLIError bugs, which propagate and + abort the run. + """ + try: + return client.transcribe(api_key, item.audio, config=config) + except NotAuthenticated: + raise + except CLIError as err: + return err + + +def _concurrent_transcripts( + api_key: str, + items: list[eval_data.EvalItem], + *, + transcription_config: aai.TranscriptionConfig, + concurrency: int, +) -> list[aai.Transcript | CLIError]: + with ThreadPoolExecutor(max_workers=concurrency) as pool: + futures = [ + pool.submit(_transcribe_one, api_key, item, transcription_config) for item in items + ] + for future in as_completed(futures): + if (exc := future.exception()) is not None: + # Only aborting failures escape _transcribe_one: drop the + # not-yet-started items rather than burn an API call on each. + pool.shutdown(cancel_futures=True) + raise exc + return [future.result() for future in futures] + + +def _transcripts( + api_key: str, + items: list[eval_data.EvalItem], + *, + transcription_config: aai.TranscriptionConfig, + concurrency: int, + json_mode: bool, + quiet: bool, +) -> list[aai.Transcript | CLIError]: + """Each item's transcript — or the CLIError it failed with — in dataset order. + + Sequential by default, with a per-item spinner; ``--concurrency`` fans the + API calls out across a thread pool (see ``_transcribe_one`` for which + failures are per-item outcomes and which abort the run). + """ + if concurrency == 1: + outcomes: list[aai.Transcript | CLIError] = [] + for index, item in enumerate(items, start=1): + with output.status( + f"[{index}/{len(items)}] Transcribing {item.item_id}…", + json_mode=json_mode, + quiet=quiet, + ): + outcomes.append(_transcribe_one(api_key, item, transcription_config)) + return outcomes + with output.status( + f"Transcribing {len(items)} items (concurrency {concurrency})…", + json_mode=json_mode, + quiet=quiet, + ): + return _concurrent_transcripts( + api_key, items, transcription_config=transcription_config, concurrency=concurrency + ) + + +def _payload( + label: str, speech_model: EvalSpeechModel | None, results: list[_ItemResult] +) -> dict[str, object]: + payload: dict[str, object] = { + "dataset": label, + "speech_model": speech_model.value if speech_model else None, + "items": len(results), + "rows": [result.row for result in results], + } + payload.update(_pooled_metrics(results)) + failed = sum(1 for result in results if "error" in result.row) + if failed: + payload["failed"] = failed + return payload + + +def _summary(payload: dict[str, object]) -> str: + parts: list[str] = [] + if "wer" in payload: + errors = jsonshape.as_int(payload.get("errors")) + noun = "error" if errors == 1 else "errors" + parts.append( + f"WER {_pct(payload.get('wer'))} ({errors} {noun} / {payload.get('words')} words)" + ) + return output.heading(" ".join(parts)) + + +def _cell(row: dict[str, object], key: str) -> str: + """The row's value as table text — blank when absent (e.g. a failed row's scores).""" + return str(row[key]) if key in row else "" + + +def _pct_cell(row: dict[str, object], key: str) -> str: + return _pct(row[key]) if key in row else "" + + +def _render(payload: dict[str, object]) -> RenderableType: + has_wer = "wer" in payload + has_failed = "failed" in payload + columns = [ + "ITEM", + *(["WORDS", "ERRORS", "WER"] if has_wer else []), + *(["ERROR"] if has_failed else []), + ] + table = output.data_table(*columns) + for row in jsonshape.mapping_list(payload.get("rows")): + cells = [str(row.get("item"))] + if has_wer: + cells += [_cell(row, "words"), _cell(row, "errors"), _pct_cell(row, "wer")] + if has_failed: + cells.append(_cell(row, "error")) + table.add_row(*cells) + model = payload.get("speech_model") or "default model" + return output.stack( + output.muted(f"{payload.get('dataset')} · {model}"), table, _summary(payload) + ) + + +def run_evaluate(opts: EvalOptions, state: AppState, *, json_mode: bool) -> None: + """Transcribe an evaluation dataset and score WER against its reference texts.""" + # 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 = state.resolve_api_key() + data = eval_data.load( + opts.dataset, + split=opts.split, + subset=opts.subset, + audio_column=opts.audio_column, + text_column=opts.text_column, + limit=opts.limit, + ) + transcription_config = aai.TranscriptionConfig( + speech_models=[opts.speech_model.value] if opts.speech_model else None, + language_code=opts.language_code, + ) + outcomes = _transcripts( + api_key, + data.items, + transcription_config=transcription_config, + concurrency=opts.concurrency, + json_mode=json_mode, + quiet=state.quiet, + ) + results = [ + _failed_result(item, outcome) + if isinstance(outcome, CLIError) + else _score_item(item, outcome) + for item, outcome in zip( + data.items, + outcomes, + strict=True, # pragma: no mutate (defensive invariant; _transcripts returns one outcome per item) + ) + ] + payload = _payload(data.label, opts.speech_model, results) + output.emit(payload, _render, json_mode=json_mode) + failed = jsonshape.as_int(payload.get("failed")) + if failed: + raise CLIError( + f"{failed} of {len(results)} items failed to transcribe.", + error_type="eval_failed", + suggestion="The summary covers only the items that transcribed.", + ) diff --git a/aai_cli/init_exec.py b/aai_cli/init_exec.py new file mode 100644 index 00000000..85390a8f --- /dev/null +++ b/aai_cli/init_exec.py @@ -0,0 +1,301 @@ +"""Run logic for `assembly init`: scaffold (and optionally launch) a starter app. + +The command module (aai_cli/commands/init.py) only parses argv — it builds an +``InitOptions`` and hands it to ``run_init`` via ``context.run_command`` (the +options/run split, see AGENTS.md), so tests drive scaffolding, install, and +launch by constructing options directly instead of round-tripping argv. + +``run_init`` and ``launch_app`` are public because the onboarding wizard +(aai_cli/onboard/sections.py) scaffolds with ``launch=False`` and starts the +dev server itself once its remaining sections have run. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +import typer +from rich.markup import escape + +from aai_cli import __version__, environments, output, stdio, steps +from aai_cli.context import AppState +from aai_cli.errors import CLIError, UsageError +from aai_cli.init import keys, runner, scaffold, templates + +DEFAULT_PORT = 3000 + + +@dataclass(frozen=True) +class InitOptions: + """Every `assembly init` flag as plain data (``--json`` excluded: run_command + resolves it into the ``json_mode`` argument).""" + + template: str | None + directory: str | None + no_install: bool + no_open: bool + force: bool + here: bool + port: int + + +def _pick_template() -> str: + """Interactive picker; raises a usage error when there's no TTY to prompt on.""" + if not stdio.interactive_stdio(): + raise CLIError( + "No template given and not running interactively. " + f"Pass one of: {', '.join(templates.TEMPLATE_ORDER)}.", + error_type="usage_error", + exit_code=1, + ) + try: + import questionary + except ImportError as exc: # a broken/stale install missing the declared dep + raise CLIError( + "The interactive picker needs 'questionary'. Reinstall the CLI " + "(e.g. `uv tool install --reinstall aai-cli`), or pass a template " + f"directly: {', '.join(templates.TEMPLATE_ORDER)}.", + error_type="missing_dependency", + exit_code=1, + ) from exc + + choice = questionary.select( + "Pick a template", + choices=[ + questionary.Choice(title=templates.title_for(t), value=t) + for t in templates.TEMPLATE_ORDER + ], + ).ask() + if choice is None: # user pressed Ctrl-C + raise typer.Exit(code=130) + return str(choice) + + +def _resolve_dir(directory: str | None, template: str, *, here: bool) -> Path: + if here: + return Path.cwd() + if directory: + return Path(directory) + return Path.cwd() / template + + +def _resolve_template(template: str | None) -> str: + """Resolve the template name: the picker when omitted, else validate the arg.""" + chosen = template if template is not None else _pick_template() + if not templates.is_template(chosen): + raise CLIError( + f"Unknown template {chosen!r}. Choose one of: {', '.join(templates.TEMPLATE_ORDER)}.", + error_type="usage_error", + exit_code=1, + ) + return chosen + + +def _active_env_vars() -> dict[str, str]: + """Pin the scaffolded app to the active environment's hosts. + + A sandbox key (minted by `assembly login` against a non-prod env) would otherwise be + rejected by the production defaults baked into the template. + """ + env = environments.active() + return { + "ASSEMBLYAI_BASE_URL": env.api_base, + "ASSEMBLYAI_LLM_GATEWAY_URL": env.llm_gateway_base, + "ASSEMBLYAI_STREAMING_HOST": env.streaming_host, + # Voice Agent host mirrors the streaming host's naming across environments. + "ASSEMBLYAI_AGENTS_HOST": env.streaming_host.replace("streaming", "agents", 1), + } + + +def _install_step( + target: Path, *, no_install: bool, api_key: str | None, use_uv: bool +) -> tuple[list[steps.Step], bool]: + """Run (or skip) dependency install, returning the report rows and whether to launch. + + Launch only happens when deps are installed and there's a key; an install failure + flips `will_launch` off so the caller exits non-zero instead of starting a server. + """ + will_launch = not no_install and api_key is not None + if no_install: + return [{"name": "install", "status": "skipped", "detail": "--no-install"}], will_launch + setup = runner.run_setup(target, use_uv=use_uv) + if setup.returncode != 0: + row: steps.Step = { + "name": "install", + "status": "failed", + "detail": (setup.stderr or setup.stdout).strip()[:300], + } + # The False (don't-launch) is an equivalent mutant: run_init raises Exit(1) on + # any failed step before it ever consults will_launch, so the value is unused + # on this branch. + return [row], False # pragma: no mutate + return [ + { + "name": "install", + "status": "installed", + "detail": "uv" if use_uv else "venv + pip", + } + ], will_launch + + +def _resolve_target( + directory: str | None, chosen: str, *, here: bool, force: bool +) -> tuple[Path, bool]: + """Resolve the target directory, rejecting --here+DIRECTORY, an existing file, or + a non-empty conflict. Returns the target and whether --force is overlaying it.""" + if here and directory: + raise CLIError( + "Pass either a DIRECTORY or --here, not both.", + error_type="usage_error", + exit_code=1, + ) + target = _resolve_dir(directory, chosen, here=here) + if target.exists() and not target.is_dir(): + raise UsageError(f"{target} exists and is not a directory.") + conflict = scaffold.target_conflict(target) + if conflict and not force: + raise CLIError( + f"{target} already exists and is not empty. " + f"Use --force to overwrite or pick another directory.", + error_type="usage_error", + exit_code=1, + ) + return target, conflict + + +def _key_row(api_key: str | None, key_source: str | None, preserved: str | None) -> steps.Step: + """The report's `key` row — emitted symmetrically whether a key resolved or not.""" + if api_key is not None: + # Literal branches rather than interpolating key_source: it rode in the same + # return tuple as the API key, so CodeQL's coarse tuple taint marks it + # sensitive and flags the report emit (py/clear-text-logging-sensitive-data). + detail = "from environment" if key_source == "environment" else "from keyring" + return {"name": "key", "status": "written", "detail": detail} + if preserved is not None: + return {"name": "key", "status": "kept", "detail": "existing .env key preserved"} + return { + "name": "key", + "status": "skipped", + "detail": "no API key found; wrote a placeholder to .env (run `assembly login`)", + } + + +def _scaffold_report( + chosen: str, + target: Path, + *, + api_key: str | None, + key_source: str | None, + preserved: str | None, +) -> list[steps.Step]: + """Write the template to `target` and return the opening report rows.""" + scaffold.scaffold(chosen, target, api_key=api_key or preserved, env_vars=_active_env_vars()) + return [ + {"name": "scaffold", "status": "created", "detail": str(target)}, + _key_row(api_key, key_source, preserved), + ] + + +def _dev_hint(port: int) -> str: + """The `assembly dev` invocation matching the chosen port (the default needs no flag).""" + return "assembly dev" if port == DEFAULT_PORT else f"assembly dev --port {port}" + + +def launch_app(target: Path, *, port: int, use_uv: bool, no_open: bool, json_mode: bool) -> None: + """Start the scaffolded app on a free port and open the browser, then block. + + Public (not underscore-private) because the onboarding wizard launches the + scaffolded app as its final step, after the remaining wizard sections have run. + """ + chosen_port = runner.find_free_port(port) + url = f"http://localhost:{chosen_port}" + if not json_mode: + output.console.print( + f"[aai.heading]Starting[/aai.heading] [aai.url]{escape(url)}[/aai.url]" + " [aai.muted](Ctrl-C to stop)[/aai.muted]" + ) + code = runner.launch_and_open(target, port=chosen_port, use_uv=use_uv, open_browser=not no_open) + if code: + raise typer.Exit(code=code) + + +def _build_report( + state: AppState, chosen: str, target: Path, *, no_install: bool, use_uv: bool, port: int +) -> tuple[list[steps.Step], bool]: + """Scaffold and assemble the report rows; returns them plus whether to launch.""" + api_key, key_source = keys.resolve_optional_api_key(profile=state.profile) + # A configured (non-placeholder) .env key must survive a re-scaffold when no key + # resolves — otherwise --force would silently reset it to the placeholder. + preserved = scaffold.existing_env_key(target) if api_key is None else None + effective_key = api_key or preserved + report = _scaffold_report( + chosen, target, api_key=api_key, key_source=key_source, preserved=preserved + ) + + install_rows, will_launch = _install_step( + target, no_install=no_install, api_key=effective_key, use_uv=use_uv + ) + report.extend(install_rows) + + # Deps are installed but there's no key, so the server can't start — say so + # rather than exiting silently. + if not no_install and effective_key is None: + report.append( + { + "name": "launch", + "status": "skipped", + "detail": f"no API key; run `assembly login`, then: cd {target} && {_dev_hint(port)}", + } + ) + return report, will_launch + + +def run_init( + opts: InitOptions, + state: AppState, + *, + json_mode: bool, + launch: bool = True, +) -> Path: + """Scaffold (and optionally install/launch) a template; return the target dir. + + `launch=False` is for callers like the onboarding wizard that must not block on a + running dev server mid-flow — it stops after install and leaves the run command as + a hint (the wizard calls `launch_app` itself once its remaining sections are done). + """ + chosen = _resolve_template(opts.template) + target, overwriting = _resolve_target(opts.directory, chosen, here=opts.here, force=opts.force) + if not json_mode: + # Vercel-style banner, printed only once validation passes so pure error runs + # (unknown template, conflicting target) stay undecorated like the sibling + # commands. Decoration goes to stderr (data → stdout): it must never pollute + # a piped stdout. + output.error_console.print( + f"[aai.heading]AssemblyAI CLI[/aai.heading] [aai.muted]{__version__}[/aai.muted]" + ) + if overwriting: + output.emit_warning( + f"--force: overwriting existing files in {target} " + "(the template is overlaid; files not in the template are kept).", + json_mode=json_mode, + ) + + use_uv = runner.has_uv() + report, will_launch = _build_report( + state, chosen, target, no_install=opts.no_install, use_uv=use_uv, port=opts.port + ) + + output.emit(report, lambda d: steps.render_steps(d, heading="Setup"), json_mode=json_mode) + if any(s["status"] == "failed" for s in report): + raise typer.Exit(code=1) + + if launch and will_launch: + launch_app(target, port=opts.port, use_uv=use_uv, no_open=opts.no_open, json_mode=json_mode) + elif not json_mode: + # Scaffolded but not launched (no key, or --no-install, or launch=False): leave the + # user with the one command that starts their app, the way `vercel`/`supabase` sign off. + output.console.print( + output.hint(f"Run `cd {escape(str(target))} && {_dev_hint(opts.port)}`.") + ) + return target diff --git a/aai_cli/onboard/sections.py b/aai_cli/onboard/sections.py index 0f1340d5..80695c45 100644 --- a/aai_cli/onboard/sections.py +++ b/aai_cli/onboard/sections.py @@ -7,9 +7,8 @@ import assemblyai as aai import typer -from aai_cli import config, environments, output, transcribe_exec, transcribe_render +from aai_cli import config, environments, init_exec, output, transcribe_exec, transcribe_render from aai_cli.commands import doctor as doctor_cmd -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 @@ -151,15 +150,17 @@ def build_path(prompter: Prompter, ctx: WizardContext) -> SectionResult: # launch=False: the dev server blocks until Ctrl-C, so launching here would stop # the remaining sections from running. launch_app starts it once the wizard is done. try: - ctx.scaffolded = init_cmd.run_init( + ctx.scaffolded = init_exec.run_init( + init_exec.InitOptions( + template=choice, + directory=None, + no_install=False, + no_open=True, + force=False, + here=False, + port=3000, + ), ctx.state, - template=choice, - directory=None, - no_install=False, - no_open=True, - force=False, - here=False, - port=3000, json_mode=ctx.json_mode, launch=False, ) @@ -210,7 +211,7 @@ def launch_app(prompter: Prompter, ctx: WizardContext) -> SectionResult: prompter.note(f"Launch it any time with `{run_hint}`.") return SectionResult.SKIPPED try: - init_cmd.launch_app( + init_exec.launch_app( ctx.scaffolded, port=3000, use_uv=runner.has_uv(), diff --git a/aai_cli/share_exec.py b/aai_cli/share_exec.py new file mode 100644 index 00000000..c284db6a --- /dev/null +++ b/aai_cli/share_exec.py @@ -0,0 +1,96 @@ +"""Run logic for `assembly share`: expose the local dev server on a public URL. + +The command module (aai_cli/commands/share.py) only parses argv — it builds a +``ShareOptions`` and hands it to ``run_share`` via ``context.run_command`` (the +options/run split, see AGENTS.md), so tests drive the tunnel orchestration by +constructing options directly instead of round-tripping argv. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path + +import typer +from rich.markup import escape + +from aai_cli import output, steps +from aai_cli.context import AppState +from aai_cli.errors import CLIError +from aai_cli.init import devserver, procfile, runner, tunnel + + +@dataclass(frozen=True) +class ShareOptions: + """Every `assembly share` flag as plain data (``--json`` excluded: run_command + resolves it into the ``json_mode`` argument).""" + + port: int + no_install: bool + + +def _render_share(data: dict[str, object]) -> str: + return ( + f"[aai.heading]Sharing[/aai.heading] [aai.url]{escape(str(data['url']))}[/aai.url]\n" + f"[aai.muted]→ serving[/aai.muted] [aai.url]{escape(str(data['local']))}[/aai.url]" + " [aai.muted](Ctrl-C to stop)[/aai.muted]" + ) + + +def run_share(opts: ShareOptions, state: AppState, *, json_mode: bool) -> None: + """Boot the app and expose it on a public cloudflared quick-tunnel URL.""" + target = Path.cwd() + use_uv = runner.has_uv() + + chosen_port = runner.find_free_port(opts.port) + devserver.notify_port_change(opts.port, chosen_port, json_mode=json_mode, quiet=state.quiet) + env = {**os.environ, "PORT": str(chosen_port)} + web = procfile.web_argv(target, env=env) # validates we're in a scaffolded project + tunnel.require_cloudflared("share a public link") + + report: list[steps.Step] = [ + devserver.install_step(target, no_install=opts.no_install, use_uv=use_uv) + ] + output.emit(report, lambda d: steps.render_steps(d, heading="Share"), json_mode=json_mode) + if any(s["status"] == "failed" for s in report): + raise typer.Exit(code=1) + + server = runner.spawn(devserver.dev_command(target, web, use_uv=use_uv), cwd=target, env=env) + proxy = None + log_path: Path | None = None + keep_log = False + try: + if not runner.wait_for_port(chosen_port): + raise CLIError( + "The dev server didn't start, so there's nothing to share.", + error_type="server_error", + exit_code=1, + ) + proxy, public, log_path = tunnel.open_quick_tunnel(chosen_port, cwd=target) + if public is None: + # Keep the captured cloudflared output: it's the only evidence of why + # the tunnel never came up. + keep_log = True + raise CLIError( + "cloudflared didn't report a tunnel URL in time.", + error_type="tunnel_error", + exit_code=1, + suggestion=f"cloudflared's output was kept at {log_path} — check it for errors.", + ) + payload: dict[str, object] = { + "url": public, + "local": f"http://localhost:{chosen_port}", + "port": chosen_port, + } + output.emit(payload, _render_share, json_mode=json_mode) + server.wait() + except KeyboardInterrupt: + # Ctrl-C is the expected way to stop a foreground share; the finally + # block below tears down the tunnel and server. + pass + finally: + tunnel.terminate(proxy) + tunnel.terminate(server) + if log_path is not None and not keep_log: + log_path.unlink(missing_ok=True) diff --git a/tests/test_deploy.py b/tests/test_deploy.py index 77269c20..73ba1d82 100644 --- a/tests/test_deploy.py +++ b/tests/test_deploy.py @@ -9,7 +9,7 @@ import pytest from typer.testing import CliRunner -from aai_cli.commands.deploy import FLY, RAILWAY, VERCEL, Target +from aai_cli.deploy_exec import FLY, RAILWAY, VERCEL, Target from aai_cli.main import app runner = CliRunner() @@ -67,7 +67,7 @@ def fake_run(cmd: list[str], *, cwd: Path, check: bool) -> types.SimpleNamespace runs.append({"cmd": cmd, "cwd": cwd, "check": check}) return types.SimpleNamespace(returncode=returncode) - monkeypatch.setattr("aai_cli.commands.deploy.subprocess.run", fake_run) + monkeypatch.setattr("aai_cli.deploy_exec.subprocess.run", fake_run) return calls @@ -358,7 +358,7 @@ def test_deploy_prod_error_suggests_dropping_the_flag(monkeypatch: pytest.Monkey def test_install_hint_platform_selection(monkeypatch: pytest.MonkeyPatch) -> None: - from aai_cli.commands import deploy + from aai_cli import deploy_exec as deploy monkeypatch.setattr("sys.platform", "darwin") assert deploy._install_hint(FLY) == "Install it with `brew install flyctl`." diff --git a/tests/test_eval_command.py b/tests/test_eval_command.py index 5a384b04..19da9424 100644 --- a/tests/test_eval_command.py +++ b/tests/test_eval_command.py @@ -42,7 +42,7 @@ def _write_wer_manifest(tmp_path): def _mock_transcribe(mocker, results): return mocker.patch( - "aai_cli.commands.evaluate.client.transcribe", + "aai_cli.evaluate_exec.client.transcribe", autospec=True, side_effect=list(results), ) @@ -141,7 +141,7 @@ def _assign(obj, attribute, value): def test_item_results_are_immutable(): - from aai_cli.commands.evaluate import _ItemResult + from aai_cli.evaluate_exec import _ItemResult result = _ItemResult(row={}, words=None) with pytest.raises(dataclasses.FrozenInstanceError): @@ -165,7 +165,7 @@ def _loaded_dataset(): def test_loader_defaults(tmp_path, mocker): _auth() load = mocker.patch( - "aai_cli.commands.evaluate.eval_data.load", autospec=True, return_value=_loaded_dataset() + "aai_cli.evaluate_exec.eval_data.load", autospec=True, return_value=_loaded_dataset() ) _mock_transcribe(mocker, [_transcript("hello")]) result = runner.invoke(app, ["eval", "org/ds"]) @@ -179,7 +179,7 @@ def test_loader_defaults(tmp_path, mocker): def test_explicit_loader_flags_pass_through(tmp_path, mocker): _auth() load = mocker.patch( - "aai_cli.commands.evaluate.eval_data.load", autospec=True, return_value=_loaded_dataset() + "aai_cli.evaluate_exec.eval_data.load", autospec=True, return_value=_loaded_dataset() ) _mock_transcribe(mocker, [_transcript("hello")]) argv = [ @@ -206,7 +206,7 @@ def test_limit_out_of_range_is_a_usage_error(limit): def test_limit_bounds_are_inclusive(tmp_path, mocker, limit): _auth() mocker.patch( - "aai_cli.commands.evaluate.eval_data.load", autospec=True, return_value=_loaded_dataset() + "aai_cli.evaluate_exec.eval_data.load", autospec=True, return_value=_loaded_dataset() ) _mock_transcribe(mocker, [_transcript("hello")]) assert runner.invoke(app, ["eval", "org/ds", "--limit", limit]).exit_code == 0 @@ -223,7 +223,7 @@ def fake_status(message, *, json_mode, quiet): seen.append(message) yield - monkeypatch.setattr("aai_cli.commands.evaluate.output.status", fake_status) + monkeypatch.setattr("aai_cli.evaluate_exec.output.status", fake_status) assert runner.invoke(app, ["eval", "manifest.csv"]).exit_code == 0 assert seen == ["[1/2] Transcribing a.wav…", "[2/2] Transcribing b.wav…"] diff --git a/tests/test_eval_failures.py b/tests/test_eval_failures.py index 6797fd3a..4c53c09e 100644 --- a/tests/test_eval_failures.py +++ b/tests/test_eval_failures.py @@ -13,7 +13,7 @@ import pytest from typer.testing import CliRunner -from aai_cli.commands import evaluate +from aai_cli import evaluate_exec as evaluate from aai_cli.errors import APIError, auth_failure from aai_cli.main import app from tests.test_eval_command import ( @@ -46,7 +46,7 @@ def fake_transcribe(api_key, audio, *, config): return _transcript(texts[Path(audio).name]) mocker.patch( - "aai_cli.commands.evaluate.client.transcribe", autospec=True, side_effect=fake_transcribe + "aai_cli.evaluate_exec.client.transcribe", autospec=True, side_effect=fake_transcribe ) result = runner.invoke(app, ["eval", "manifest.csv", "--concurrency", "2", "--json"]) assert result.exit_code == 0 @@ -67,7 +67,7 @@ def fake_status(message, *, json_mode, quiet): seen.append(message) yield - monkeypatch.setattr("aai_cli.commands.evaluate.output.status", fake_status) + monkeypatch.setattr("aai_cli.evaluate_exec.output.status", fake_status) assert runner.invoke(app, ["eval", "manifest.csv", "--concurrency", "2"]).exit_code == 0 assert seen == ["Transcribing 2 items (concurrency 2)…"] @@ -211,7 +211,7 @@ def test_rejected_key_aborts_eval_with_auth_exit_code(tmp_path, mocker): def test_unauthenticated_fails_before_dataset_download(mocker): # Credentials resolve before the dataset loads: a signed-out user must not # pull the whole dataset first. - load = mocker.patch("aai_cli.commands.evaluate.eval_data.load", autospec=True) + load = mocker.patch("aai_cli.evaluate_exec.eval_data.load", autospec=True) result = runner.invoke(app, ["eval", "org/ds"]) assert result.exit_code == 4 load.assert_not_called() diff --git a/tests/test_init_command.py b/tests/test_init_command.py index 6b1838ef..460701f5 100644 --- a/tests/test_init_command.py +++ b/tests/test_init_command.py @@ -7,6 +7,7 @@ import typer from typer.testing import CliRunner +from aai_cli import init_exec from aai_cli.commands import init as init_cmd from aai_cli.errors import CLIError from aai_cli.main import app @@ -255,7 +256,7 @@ def test_pick_template_interactive_returns_choice(monkeypatch): monkeypatch.setattr("sys.stdin", _Tty()) monkeypatch.setattr("sys.stdout", _Tty()) monkeypatch.setitem(sys.modules, "questionary", _fake_questionary(TEMPLATE)) - assert init_cmd._pick_template() == TEMPLATE + assert init_exec._pick_template() == TEMPLATE def test_pick_template_ctrl_c_exits_130(monkeypatch): @@ -264,7 +265,7 @@ def test_pick_template_ctrl_c_exits_130(monkeypatch): monkeypatch.setattr("sys.stdout", _Tty()) monkeypatch.setitem(sys.modules, "questionary", _fake_questionary(None)) with pytest.raises(typer.Exit) as exc: - init_cmd._pick_template() + init_exec._pick_template() assert exc.value.exit_code == 130 @@ -274,7 +275,7 @@ def test_pick_template_missing_questionary_errors(monkeypatch): monkeypatch.setattr("sys.stdout", _Tty()) monkeypatch.setitem(sys.modules, "questionary", None) # makes `import questionary` raise with pytest.raises(CLIError) as exc: - init_cmd._pick_template() + init_exec._pick_template() assert exc.value.error_type == "missing_dependency" assert exc.value.exit_code == 1 @@ -289,7 +290,7 @@ def test_pick_template_errors_when_either_stream_not_a_tty(monkeypatch, stdin_tt monkeypatch.setattr("sys.stdout", _Tty() if stdout_tty else io.StringIO()) monkeypatch.setitem(sys.modules, "questionary", None) with pytest.raises(CLIError) as exc: - init_cmd._pick_template() + init_exec._pick_template() assert exc.value.error_type == "usage_error" assert exc.value.exit_code == 1 @@ -302,8 +303,8 @@ def test_active_env_vars_agents_host_replaces_only_first_streaming(monkeypatch): llm_gateway_base="https://llm.x", streaming_host="streaming.streaming.example.com", ) - monkeypatch.setattr(init_cmd.environments, "active", lambda: fake_env) - assert init_cmd._active_env_vars()["ASSEMBLYAI_AGENTS_HOST"] == "agents.streaming.example.com" + monkeypatch.setattr(init_exec.environments, "active", lambda: fake_env) + assert init_exec._active_env_vars()["ASSEMBLYAI_AGENTS_HOST"] == "agents.streaming.example.com" def test_init_install_failure_reports_and_exits(tmp_path, monkeypatch): diff --git a/tests/test_onboard_sections.py b/tests/test_onboard_sections.py index 3a4939d8..c2f9600f 100644 --- a/tests/test_onboard_sections.py +++ b/tests/test_onboard_sections.py @@ -7,8 +7,7 @@ import pytest import typer -from aai_cli import output, transcribe_exec, transcribe_render -from aai_cli.commands import init as init_cmd +from aai_cli import init_exec, output, transcribe_exec, transcribe_render from aai_cli.commands import setup as setup_cmd from aai_cli.context import AppState from aai_cli.errors import CLIError @@ -161,7 +160,7 @@ def _fake_run_init(*a: object, **k: object) -> Path: called = True return Path() - monkeypatch.setattr(init_cmd, "run_init", _fake_run_init) + monkeypatch.setattr(init_exec, "run_init", _fake_run_init) # NonInteractivePrompter.select returns the default; build_path's default is "skip". assert sections.build_path(NonInteractivePrompter(), ctx) is SectionResult.SKIPPED assert called is False @@ -255,34 +254,28 @@ def test_build_path_scaffolds(ctx: WizardContext, monkeypatch: pytest.MonkeyPatc seen: dict[str, object] = {} def _fake_run_init( + opts: init_exec.InitOptions, state: object, *, - template: str | None, - directory: str | None, - no_install: bool, - no_open: bool, - force: bool, - here: bool, - port: int, json_mode: bool, launch: bool = True, ) -> Path: nonlocal calls calls += 1 seen.update( - template=template, - directory=directory, - no_install=no_install, - no_open=no_open, - force=force, - here=here, - port=port, + template=opts.template, + directory=opts.directory, + no_install=opts.no_install, + no_open=opts.no_open, + force=opts.force, + here=opts.here, + port=opts.port, json_mode=json_mode, launch=launch, ) return Path("/scaffolded/app") - monkeypatch.setattr(init_cmd, "run_init", _fake_run_init) + monkeypatch.setattr(init_exec, "run_init", _fake_run_init) prompter = _ScriptedPrompter(select="audio-transcription", confirm=True) result = sections.build_path(prompter, ctx) assert result is SectionResult.DONE @@ -314,7 +307,7 @@ def _fake_run_init(*a: object, **k: object) -> Path: called = True return Path() - monkeypatch.setattr(init_cmd, "run_init", _fake_run_init) + monkeypatch.setattr(init_exec, "run_init", _fake_run_init) result = sections.build_path(_ScriptedPrompter(select="voice-agent", confirm=False), ctx) assert result is SectionResult.SKIPPED assert called is False @@ -324,7 +317,7 @@ def test_build_path_run_init_failure(ctx: WizardContext, monkeypatch: pytest.Mon def _boom(*a: object, **k: object) -> Path: raise typer.Exit(code=1) - monkeypatch.setattr(init_cmd, "run_init", _boom) + monkeypatch.setattr(init_exec, "run_init", _boom) result = sections.build_path(_ScriptedPrompter(select="live-captions", confirm=True), ctx) assert result is SectionResult.FAILED # Nothing scaffolded, so launch_app must have nothing to launch. @@ -382,7 +375,7 @@ def _fake_launch( captured["target"] = target captured.update(port=port, use_uv=use_uv, no_open=no_open, json_mode=json_mode) - monkeypatch.setattr(init_cmd, "launch_app", _fake_launch) + monkeypatch.setattr(init_exec, "launch_app", _fake_launch) return captured @@ -452,6 +445,6 @@ def test_launch_app_failure( def _boom(*a: object, **k: object) -> None: raise exc - monkeypatch.setattr(init_cmd, "launch_app", _boom) + monkeypatch.setattr(init_exec, "launch_app", _boom) ctx.scaffolded = Path("/scaffolded/app") assert sections.launch_app(_ScriptedPrompter(confirm=True), ctx) is SectionResult.FAILED