Skip to content

Reorganize code_gen and command tests by concern #68

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alexkroman merged 1 commit into from
Jun 11, 2026
Merged

Reorganize code_gen and command tests by concern #68

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 12 additions & 47 deletions aai_cli/commands/transcribe.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@

import assemblyai as aai
import typer
from rich.markup import escape

from aai_cli import (
choices,
Expand All @@ -17,14 +16,12 @@
options,
output,
transcribe_exec,
transcribe_render,
)

# The package attribute `code_gen.transcribe` is the wrapper function, so the module's
# render() (which also takes the -o output field) is imported from the submodule itself.
from aai_cli.code_gen.transcribe import render as render_transcribe_code
from aai_cli.context import AppState, run_command
from aai_cli.errors import UsageError
from aai_cli.help_text import examples_epilog

app = typer.Typer()
Expand Down Expand Up @@ -398,12 +395,7 @@ def body(state: AppState, json_mode: bool) -> None:
}
flags.update(config_builder.auth_header_flags(webhook_auth_header))

if out is not None and llm_prompt:
# --out captures the transcript itself; an LLM transform is a separate step.
raise UsageError(
"--out can't be combined with --llm.",
suggestion='Pipe the transform instead, e.g. -o text | assembly llm -f "…".',
)
transcribe_exec.validate_out_with_llm(out, llm_prompt)

merged = config_builder.merge_transcribe_config(
flags=flags, overrides=config_kv, config_file=config_file
Expand Down Expand Up @@ -447,43 +439,16 @@ def body(state: AppState, json_mode: bool) -> None:
download_sections=list(download_sections or []),
)

if out is not None:
# Write a clean file artifact and confirm on stderr; stdout stays empty.
if ".." in out.parts: # reject path-traversal segments in --out
raise UsageError(f"--out path can't contain '..': {out}")
out.write_text(
transcribe_exec.out_payload(transcript, output_field, json_mode=json_mode) + "\n"
)
if not state.quiet:
output.error_console.print(output.success(f"Saved to {escape(str(out))}"))
return

if output_field is not None:
# Raw single-field output for pipelines (overrides --json and analysis render).
output.emit_text(client.select_transcript_field(transcript, output_field))
return

if llm_prompt:
# Chain the prompts: the first runs over the transcript (injected server-side
# via transcript_id); each subsequent prompt runs over the prior response.
steps = llm.run_chain_steps(
api_key,
llm_prompt,
transcript_id=transcript.id,
model=model,
max_tokens=max_tokens,
)
output.emit(
client.transcript_summary(transcript)
| {"transform": {"model": model, "steps": steps}},
transcribe_exec.render_transform_steps,
json_mode=json_mode,
)
return

if json_mode:
output.emit(client.transcript_json_payload(transcript), lambda d: d, json_mode=True)
else:
transcribe_render.render_transcript_result(transcript, output.console)
transcribe_exec.deliver_result(
transcript,
api_key=api_key,
out=out,
output_field=output_field,
transform=transcribe_exec.TransformOptions(
prompts=list(llm_prompt or []), model=model, max_tokens=max_tokens
),
json_mode=json_mode,
quiet=state.quiet,
)

run_command(ctx, body, json=json_out)
74 changes: 71 additions & 3 deletions aai_cli/transcribe_exec.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
"""Transcription execution and non-Rich output shaping for the ``transcribe`` command.
"""Transcription execution and result delivery for the ``transcribe`` command.

Kept out of ``commands/transcribe.py`` so the command stays a thin option surface, and
so ``run_transcription`` lives in a core module that ``onboard`` can import directly
Expand All @@ -10,11 +10,12 @@
import json
import tempfile
from pathlib import Path
from typing import Any
from typing import Any, NamedTuple

import assemblyai as aai
from rich.markup import escape

from aai_cli import choices, client, stdio, youtube
from aai_cli import choices, client, llm, output, stdio, transcribe_render, youtube
from aai_cli.errors import UsageError

# The PII policy strings the SDK accepts, validated client-side so a typo'd
Expand Down Expand Up @@ -47,6 +48,15 @@ def validate_speakers_expected(merged: dict[str, object]) -> None:
)


def validate_out_with_llm(out: Path | None, llm_prompts: list[str] | None) -> None:
if out is not None and llm_prompts:
# --out captures the transcript itself; an LLM transform is a separate step.
raise UsageError(
"--out can't be combined with --llm.",
suggestion='Pipe the transform instead, e.g. -o text | assembly llm -f "…".',
)


def render_transform_steps(d: dict[str, Any]) -> str:
"""Human view of chained LLM-Gateway steps: the lone output, or each step labeled."""
steps = d["transform"]["steps"]
Expand Down Expand Up @@ -105,3 +115,61 @@ def run_transcription(
local = youtube.download_audio(audio, Path(td), download_sections=download_sections)
return client.transcribe(api_key, str(local), config=transcription_config)
return client.transcribe(api_key, audio, config=transcription_config)


class TransformOptions(NamedTuple):
"""The ``--llm`` chain options: the prompts plus the gateway model settings."""

prompts: list[str]
model: str
max_tokens: int


def deliver_result(
transcript: aai.Transcript,
*,
api_key: str,
out: Path | None,
output_field: choices.TranscriptOutput | None,
transform: TransformOptions,
json_mode: bool,
quiet: bool,
) -> None:
"""Route the finished transcript: ``--out`` file, single ``-o`` field, ``--llm``
transform chain, or the default JSON/human render — first match wins."""
if out is not None:
# Write a clean file artifact and confirm on stderr; stdout stays empty.
if ".." in out.parts: # reject path-traversal segments in --out
raise UsageError(f"--out path can't contain '..': {out}")
out.write_text(out_payload(transcript, output_field, json_mode=json_mode) + "\n")
if not quiet:
output.error_console.print(output.success(f"Saved to {escape(str(out))}"))
return

if output_field is not None:
# Raw single-field output for pipelines (overrides --json and analysis render).
output.emit_text(client.select_transcript_field(transcript, output_field))
return

if transform.prompts:
# Chain the prompts: the first runs over the transcript (injected server-side
# via transcript_id); each subsequent prompt runs over the prior response.
steps = llm.run_chain_steps(
api_key,
transform.prompts,
transcript_id=transcript.id,
model=transform.model,
max_tokens=transform.max_tokens,
)
output.emit(
client.transcript_summary(transcript)
| {"transform": {"model": transform.model, "steps": steps}},
render_transform_steps,
json_mode=json_mode,
)
return

if json_mode:
output.emit(client.transcript_json_payload(transcript), lambda d: d, json_mode=True)
else:
transcribe_render.render_transcript_result(transcript, output.console)
125 changes: 3 additions & 122 deletions tests/test_code_gen.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Example-based code_gen tests: serializers, snippets, rendered scripts.
"""Example-based code_gen tests: serializers, snippets, transcribe rendering.

Hypothesis fuzz/property tests live in test_code_gen_fuzz.py.
Stream/agent scaffold tests live in test_code_gen_stream_agent.py; hypothesis
fuzz/property tests live in test_code_gen_fuzz.py.
"""

from __future__ import annotations
Expand Down Expand Up @@ -182,55 +183,6 @@ def test_transcribe_render_plain_url_is_not_downloaded():
assert "transcriber.transcribe('https://assembly.ai/wildfires.mp3')" in code


def test_stream_render_parses_and_is_runnable_shape():
from assemblyai.streaming.v3 import SpeechModel

code = code_gen.stream(
{"sample_rate": 16000, "format_turns": True, "speech_model": SpeechModel.u3_rt_pro}
)
ast.parse(code)
assert "StreamingClient(" in code
assert "StreamingParameters(" in code
assert "SpeechModel.u3_rt_pro" in code
assert "MicrophoneStream" in code
assert 'os.environ["ASSEMBLYAI_API_KEY"]' in code


def test_stream_render_mic_rate_matches_params():
code = code_gen.stream({"sample_rate": 8000})
ast.parse(code)
assert "StreamingParameters(\n sample_rate=8000," in code
assert "MicrophoneStream(sample_rate=8000)" in code


def test_stream_render_empty_is_clean_and_has_no_speechmodel_import():
code = code_gen.stream({})
ast.parse(code)
assert "StreamingParameters()" in code
assert " SpeechModel," not in code # not imported when unused (keeps script lint-clean)
assert "MicrophoneStream(sample_rate=16000)" in code # default rate


def test_agent_render_parses_and_injects_session_fields():
code = code_gen.agent(voice="ivy", system_prompt="Be terse.", greeting="Hi there")
ast.parse(code)
assert '"voice": "ivy"' in code
assert "Be terse." in code
assert "Hi there" in code
assert "agents.assemblyai.com" in code
assert 'os.environ["ASSEMBLYAI_API_KEY"]' in code


def test_agent_render_escapes_quotes_in_prompt():
import json as _json

tricky = 'Say "hi"\nand stop'
code = code_gen.agent(voice="ivy", system_prompt=tricky, greeting="Hello")
ast.parse(code) # valid Python despite embedded quotes/newlines
# The prompt is injected via json.dumps, so its escaped form appears verbatim.
assert _json.dumps(tricky) in code


# ---------------------------------------------------------------------------
# Validity & fidelity checks (the exhaustive hypothesis harness — Task 10 —
# lives in test_code_gen_fuzz.py).
Expand Down Expand Up @@ -378,77 +330,6 @@ def test_transcribe_show_code_without_gateway_has_no_openai_import():
assert "transcript.utterances" in code # normal result handling instead


def test_agent_show_code_uses_single_full_duplex_stream():
# ONE sd.RawStream (mic+speaker); two separate streams fail on macOS CoreAudio.
code = code_gen.agent(voice="ivy", system_prompt="p", greeting="g")
ast.parse(code)
assert "sd.RawStream(" in code
assert "samplerate=RATE" in code # opens at the agent's native 24 kHz, no resampling
assert "RawInputStream" not in code
assert "RawOutputStream" not in code
# No audioop: it's deprecated and removed in Python 3.13, so the script stays portable.
assert "audioop" not in code


def test_stream_show_code_includes_llm_follow_loop():
code = code_gen.stream(
{"speech_model": "universal_streaming"},
llm={
"prompts": ["summarize", "translate to french"],
"model": "claude-haiku-4-5-20251001",
"max_tokens": 500,
"interval": 30.0,
},
)
ast.parse(code)
assert "from openai import OpenAI" in code
assert "llm-gateway.assemblyai.com" in code
# Both prompts appear, in order, for the chain.
assert code.index("summarize") < code.index("translate to french")
# Still streams from the mic, refreshing the answer on the interval.
assert "MicrophoneStream" in code
assert "end_of_turn" in code
assert "claude-haiku-4-5-20251001" in code
# The generated loop mirrors --llm-interval: a baked-in throttle plus a closing flush.
assert "LLM_INTERVAL = 30.0" in code
assert "now - _last_summary < LLM_INTERVAL" in code
assert "summarize(final=True)" in code


def test_gateway_options_defaults_interval_to_per_turn():
# Called without an explicit interval (transcribe's path), the baked-in cadence is
# per-turn (0.0); pins the default so it can't drift.
opts = code_gen.gateway_options(["summarize"], "m", 100)
assert opts is not None
assert opts["interval"] == 0.0


def test_stream_show_code_defaults_interval_when_absent():
# An llm dict with no "interval" key falls back to per-turn (LLM_INTERVAL = 0.0).
code = code_gen.stream({}, llm={"prompts": ["s"], "model": "m", "max_tokens": 1})
ast.parse(code)
assert "LLM_INTERVAL = 0.0" in code


def test_stream_show_code_llm_interval_zero_is_per_turn():
# --llm-interval 0 bakes in the legacy per-turn cadence (LLM_INTERVAL = 0.0 makes the
# throttle a no-op), while still emitting the closing flush.
code = code_gen.stream(
{},
llm=code_gen.gateway_options(["summarize"], "m", 100, interval=0.0),
)
ast.parse(code)
assert "LLM_INTERVAL = 0.0" in code
assert "summarize(final=True)" in code


def test_stream_show_code_without_llm_is_plain_scaffold():
code = code_gen.stream({})
ast.parse(code)
assert "from openai import OpenAI" not in code # no gateway when --llm absent
assert "MicrophoneStream" in code


def test_generated_code_targets_active_environment():
# --show-code embeds hosts from the active environment, so a sandbox user's
# generated script talks to the sandbox that minted their key, not production.
Expand Down
Loading
Loading