diff --git a/AGENTS.md b/AGENTS.md index 19ee5d86..6f23f993 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -153,7 +153,7 @@ A Typer CLI. `aai_cli/main.py` builds the `app`, registers each command sub-app, ### Command layer -Each file in `aai_cli/commands/` is a Typer sub-app (`transcribe`, `stream`, `transcripts`, `agent`, `llm`, `login` (login/logout/whoami), `doctor`, `init`, `dev`, `share`, `deploy`, `setup`, `onboard`, `account` (balance/usage/limits), `keys`, `sessions`, `audit`). Command bodies run through `context.run_command(ctx, fn, json=...)`, which maps any `CLIError` to clean stderr output + the error's exit code. Commands never print tracebacks for expected failures. +Each file in `aai_cli/commands/` is a Typer sub-app (`transcribe`, `stream`, `agent`, `speak`, `llm`, `transcripts`, `login` (login/logout/whoami), `doctor`, `init`, `dev`, `share`, `deploy`, `setup`, `onboard`, `account` (balance/usage/limits), `keys`, `sessions`, `audit`). Command bodies run through `context.run_command(ctx, fn, json=...)`, which maps any `CLIError` to clean stderr output + the error's exit code. Commands never print tracebacks for expected failures. ### Cross-cutting state (resolution order matters) @@ -167,6 +167,7 @@ Each file in `aai_cli/commands/` is a Typer sub-app (`transcribe`, `stream`, `tr - **`streaming/`** + `client.stream_audio` — v3 realtime API. Event callbacks run on the SDK reader thread and guard against `BrokenPipeError` (`stdio.silence_stdout()`) so a closed pipe never dumps a thread traceback. - **`agent/`** — full-duplex voice agent (mic in, TTS out via `voices.py`). +- **`tts/`** + `commands/speak.py` — `aai speak` synthesizes text to speech over the sandbox streaming-TTS WebSocket (`streaming-tts.sandbox000.…`). **Sandbox-only:** `session.is_available()` is false in production (empty `Environment.streaming_tts_host`), so the command exits 2 with a `--sandbox` hint. `session.synthesize` drives a Begin→Generate→Flush→Audio→Terminate protocol with an injectable `connect` for hermetic tests (mirrors `agent/session.py`); `audio.py` plays the PCM (default) or writes a WAV (`--out`). - **`code_gen/`** — backs `--show-code` on `transcribe`/`stream`/`agent`: builds a ready-to-run Python SDK script from exactly the flags passed (no API key needed; generated code reads `ASSEMBLYAI_API_KEY`). - **`auth/`** — browser-assisted `aai login` via AMS + **Stytch B2B OAuth discovery** (`discovery.py`, `flow.py`, `loopback.py`, `ams.py`). Not Stytch Connected Apps. - **`init/`** — scaffolds a self-contained FastAPI + HTML starter (`audio-transcription`/`live-captions`/`voice-agent` templates), optionally installs deps and opens the browser; writes the key to a git-ignored `.env`. diff --git a/aai_cli/commands/speak.py b/aai_cli/commands/speak.py new file mode 100644 index 00000000..117cacb6 --- /dev/null +++ b/aai_cli/commands/speak.py @@ -0,0 +1,252 @@ +from __future__ import annotations + +import sys +from pathlib import Path + +import typer + +from aai_cli import config, 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 +from aai_cli.tts import audio, dialogue, session + +app = typer.Typer() + +# The streaming-TTS reference client defaults to the PocketTTS "jane" voice and +# English, so the CLI sends the same and a bare `aai speak` works out of the box. +# Override either with --voice/--language. +DEFAULT_VOICE = "jane" +DEFAULT_LANGUAGE = "English" + + +def _read_text(text: str | None) -> str: + """The text to speak: the non-blank argument, or piped stdin when the argument + is omitted entirely. A *blank* argument (e.g. "") is a usage error, never a + silent fall-through to stdin — so `aai speak "$MSG"` with an empty MSG fails + fast instead of consuming whatever happens to be on the pipe.""" + if text is not None and text.strip(): + return text + # `text is None` (argument omitted), not merely blank: see the docstring rationale. + if text is None and not sys.stdin.isatty(): + piped = sys.stdin.read().strip() + if piped: + return piped + raise UsageError( + "No text to speak.", + suggestion='Pass text as an argument: aai speak "Hello" — or pipe it via stdin.', + ) + + +def _output_audio(result: session.SpeakResult, out: Path | None) -> None: + """Write a WAV when --out is given, else play through the speakers.""" + if out is not None: + audio.write_wav(out, result.pcm, result.sample_rate) + else: + audio.play_pcm(result.pcm, result.sample_rate) + + +def _disposition(out: Path | None) -> str: + return f"saved to {out}" if out is not None else "played" + + +def _emit_single( + result: session.SpeakResult, + cfg: session.SpeakConfig, + out: Path | None, + *, + json_mode: bool, +) -> None: + """Single-voice result: a JSON object on stdout, or a human note on stderr.""" + duration = round(result.audio_duration_seconds, 3) + if json_mode: + output.emit_ndjson( + { + "voice": cfg.voice, + "language": cfg.language, + "sample_rate": result.sample_rate, + "audio_duration_seconds": duration, + "bytes": len(result.pcm), + "out": str(out) if out is not None else None, + } + ) + return + output.error_console.print( + f"[aai.muted]Spoke {duration}s of audio ({_disposition(out)}).[/aai.muted]" + ) + + +def _emit_multi( + result: session.SpeakResult, + speakers: dict[str, str], + segment_count: int, + out: Path | None, + *, + json_mode: bool, +) -> None: + """Multi-voice result: a JSON object on stdout, or a human note on stderr.""" + duration = round(result.audio_duration_seconds, 3) + if json_mode: + output.emit_ndjson( + { + "mode": "multi", + "speakers": speakers, + "segments": segment_count, + "sample_rate": result.sample_rate, + "audio_duration_seconds": duration, + "bytes": len(result.pcm), + "out": str(out) if out is not None else None, + } + ) + return + voices = ", ".join(f"{spk}={voice}" for spk, voice in speakers.items()) + output.error_console.print( + f"[aai.muted]Spoke {duration}s across {len(speakers)} voices " + f"({voices}) ({_disposition(out)}).[/aai.muted]" + ) + + +def _speak_single( + api_key: str, + text: str, + voice: str, + language: str, + sample_rate: int | None, + out: Path | None, + *, + json_mode: bool, + quiet: bool, +) -> None: + cfg = session.SpeakConfig(text=text, voice=voice, language=language, sample_rate=sample_rate) + with output.status("Synthesizing speech…", json_mode=json_mode, quiet=quiet): + result = session.synthesize( + api_key, cfg, on_warning=lambda m: output.emit_warning(m, json_mode=json_mode) + ) + _output_audio(result, out) + _emit_single(result, cfg, out, json_mode=json_mode) + + +def _speak_dialogue( + api_key: str, + text: str, + bare_voice: str | None, + overrides: dict[str, str], + language: str, + sample_rate: int | None, + out: Path | None, + *, + json_mode: bool, + quiet: bool, +) -> None: + segments = dialogue.parse_segments(text) + if not segments: + raise UsageError( + "No text to speak.", + suggestion="The input had speaker labels but no spoken text.", + ) + if bare_voice is not None: + output.emit_warning( + "Ignoring bare --voice in multi-speaker mode; " + "set a voice per speaker with --voice A=NAME.", + json_mode=json_mode, + ) + resolved, speakers = dialogue.assign_voices( + segments, dialogue.DEFAULT_VOICE_ROTATION, overrides + ) + with output.status("Synthesizing speech…", json_mode=json_mode, quiet=quiet): + result = session.synthesize_dialogue( + api_key, + resolved, + language=language, + sample_rate=sample_rate, + on_warning=lambda m: output.emit_warning(m, json_mode=json_mode), + ) + _output_audio(result, out) + _emit_multi(result, speakers, len(resolved), out, json_mode=json_mode) + + +@app.command( + rich_help_panel=help_panels.TRANSCRIPTION, + epilog=examples_epilog( + [ + ("Speak text aloud (sandbox only)", 'aai speak "Hello there, friend." --sandbox'), + ( + "Pick a voice and language", + 'aai speak "Bonjour" --voice jane --language French --sandbox', + ), + ( + "Speak a diarized transcript, one voice per speaker", + "aai transcribe meeting.mp3 --speaker-labels | aai speak --sandbox", + ), + ( + "Override a speaker's voice", + "… | aai speak --voice A=vera --voice B=paul --sandbox", + ), + ( + "Save to a WAV instead of playing", + 'aai speak "Hello" --out /tmp/hello.wav --sandbox', + ), + ] + ), +) +def speak( + ctx: typer.Context, + text: str | None = typer.Argument(None, help="Text to speak. Omit to read from stdin."), + voice: list[str] = typer.Option( + [], + "--voice", + help="Voice id, or SPEAKER=VOICE for diarized input (repeatable, e.g. --voice A=jane).", + ), + language: str = typer.Option(DEFAULT_LANGUAGE, "--language", help="Language of the text."), + sample_rate: int | None = typer.Option( + None, "--sample-rate", help="Output sample rate in Hz. Server default if omitted." + ), + out: Path | None = typer.Option( + None, "--out", help="Write a WAV file instead of playing through the speakers." + ), + json_out: bool = options.json_option("Emit JSON metadata about the synthesized audio."), +) -> None: + """Synthesize speech from text with AssemblyAI streaming TTS (sandbox only). + + Plays the audio through your speakers by default, or writes a WAV with --out. + Speaker-labeled input (from 'aai transcribe --speaker-labels') is detected + automatically: the labels are stripped and each speaker gets a different + voice. This feature only exists in the sandbox today — run it with --sandbox. + """ + + def body(state: AppState, json_mode: bool) -> None: + if not session.is_available(): + raise CLIError( + "aai speak is only available in the sandbox.", + error_type="unsupported_environment", + exit_code=2, + suggestion="Re-run with --sandbox (or --env sandbox000).", + ) + spoken = _read_text(text) + api_key = config.resolve_api_key(profile=state.profile) + bare_voice, overrides = dialogue.parse_voice_overrides(voice) + if dialogue.looks_like_speaker_labeled(spoken): + _speak_dialogue( + api_key, + spoken, + bare_voice, + overrides, + language, + sample_rate, + out, + json_mode=json_mode, + quiet=state.quiet, + ) + else: + _speak_single( + api_key, + spoken, + bare_voice or DEFAULT_VOICE, + language, + sample_rate, + out, + json_mode=json_mode, + quiet=state.quiet, + ) + + run_command(ctx, body, json=json_out) diff --git a/aai_cli/environments.py b/aai_cli/environments.py index 12955f56..4373314d 100644 --- a/aai_cli/environments.py +++ b/aai_cli/environments.py @@ -17,6 +17,7 @@ class Environment: name: str api_base: str # SDK base_url for /v2/upload + /v2/transcript streaming_host: str # StreamingClientOptions.api_host (SDK builds wss://host/v3/ws) + streaming_tts_host: str # streaming TTS host; empty when TTS isn't available (prod) agents_host: str # Voice Agent host; the agent client builds wss://host/v1/ws llm_gateway_base: str # OpenAI base_url for the LLM Gateway (…/v1) ams_base: str # Accounts Management Service @@ -37,6 +38,7 @@ class Environment: name="production", api_base="https://api.assemblyai.com", streaming_host="streaming.assemblyai.com", + streaming_tts_host="", agents_host="agents.assemblyai.com", llm_gateway_base="https://llm-gateway.assemblyai.com/v1", ams_base="https://ams.internal.assemblyai-labs.com", @@ -48,6 +50,7 @@ class Environment: name="sandbox000", api_base="https://api.sandbox000.assemblyai-labs.com", streaming_host="streaming.sandbox000.assemblyai-labs.com", + streaming_tts_host="streaming-tts.sandbox000.assemblyai-labs.com", agents_host="agents.sandbox000.assemblyai-labs.com", llm_gateway_base="https://llm-gateway.sandbox000.assemblyai-labs.com/v1", ams_base="https://ams.sandbox000.assemblyai-labs.com", diff --git a/aai_cli/main.py b/aai_cli/main.py index 11bd8f11..0ff37b90 100644 --- a/aai_cli/main.py +++ b/aai_cli/main.py @@ -34,6 +34,7 @@ sessions, setup, share, + speak, stream, transcribe, transcripts, @@ -60,6 +61,7 @@ "transcribe", "stream", "agent", + "speak", "llm", # Setup & Tools — get set up & maintain "doctor", @@ -313,6 +315,7 @@ def main( app.add_typer(sessions.app, name="sessions", rich_help_panel=help_panels.HISTORY) app.add_typer(audit.app) # audit app.add_typer(agent.app) +app.add_typer(speak.app) app.add_typer(llm.app) app.add_typer(account.app) # balance, usage, limits app.add_typer(login.app) # login, logout, whoami diff --git a/aai_cli/tts/__init__.py b/aai_cli/tts/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/aai_cli/tts/audio.py b/aai_cli/tts/audio.py new file mode 100644 index 00000000..0b608dd0 --- /dev/null +++ b/aai_cli/tts/audio.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +import contextlib +import wave +from collections.abc import Callable +from pathlib import Path +from typing import Protocol + +from aai_cli.errors import CLIError +from aai_cli.microphone import audio_missing_error + + +class _OutputStream(Protocol): + """The slice of a sounddevice output stream play_pcm drives — named as a + Protocol so the untyped library boundary is structurally typed, not opaque.""" + + def start(self) -> None: ... + def write(self, data: bytes, /) -> object: ... # real write returns a bool we ignore + def stop(self) -> None: ... + def abort(self) -> None: ... # immediate stop: discards buffered frames (vs stop's drain) + def close(self) -> None: ... + + +# Write playback in ~4 KiB chunks (≈85 ms of 16-bit mono at 24 kHz) instead of one +# big blocking write, so a Ctrl-C is delivered between chunks and cancels promptly +# rather than only after the entire clip has been queued to the output device. +_PLAYBACK_CHUNK_BYTES = 4096 + + +def write_wav(path: Path, pcm: bytes, sample_rate: int) -> None: + """Write 16-bit mono PCM to a WAV file, creating parent dirs as needed.""" + path.parent.mkdir(parents=True, exist_ok=True) + with wave.open(str(path), "wb") as wav: + wav.setnchannels(1) + wav.setsampwidth(2) + wav.setframerate(sample_rate) + wav.writeframes(pcm) + + +def silence(sample_rate: int, seconds: float) -> bytes: + """Zeroed 16-bit mono PCM of the given duration (2 bytes per frame).""" + return b"\x00" * (int(sample_rate * seconds) * 2) + + +def _default_output_stream(sample_rate: int) -> _OutputStream: + """A started-on-demand raw 16-bit mono output stream from sounddevice.""" + try: + import sounddevice as sd + except ImportError as exc: + raise audio_missing_error() from exc + stream: _OutputStream = sd.RawOutputStream(samplerate=sample_rate, channels=1, dtype="int16") + return stream + + +def _playback_error(exc: Exception) -> CLIError: + return CLIError( + f"Could not play audio: {exc}", + error_type="audio_output_error", + exit_code=1, + suggestion="Check your output device and run 'aai doctor', or use --out to save a WAV.", + ) + + +def play_pcm( + pcm: bytes, + sample_rate: int, + *, + stream_factory: Callable[[int], _OutputStream] | None = None, +) -> None: + """Play 16-bit mono PCM through the default output device (blocks until done). + + Audio is written in short chunks so a Ctrl-C interrupts promptly: on + KeyboardInterrupt the stream is aborted (buffered frames discarded) for an + immediate stop, then the cancel propagates. ``stream_factory`` is injectable + for tests; a device failure is wrapped in a clean CLIError that points at + --out as the headless escape hatch. + """ + factory = stream_factory or _default_output_stream + try: + stream = factory(sample_rate) + except CLIError: + raise # audio_missing_error() is already user-facing + except Exception as exc: + raise _playback_error(exc) from exc + + try: + stream.start() + for offset in range(0, len(pcm), _PLAYBACK_CHUNK_BYTES): + stream.write(pcm[offset : offset + _PLAYBACK_CHUNK_BYTES]) + stream.stop() + except KeyboardInterrupt: + # Cut sound immediately (discard whatever is still buffered in the device) + # instead of letting stop() drain the rest, then propagate the cancel. + with contextlib.suppress(Exception): + stream.abort() + raise + except Exception as exc: + raise _playback_error(exc) from exc + finally: + with contextlib.suppress(Exception): + stream.close() diff --git a/aai_cli/tts/dialogue.py b/aai_cli/tts/dialogue.py new file mode 100644 index 00000000..4cef86ea --- /dev/null +++ b/aai_cli/tts/dialogue.py @@ -0,0 +1,118 @@ +from __future__ import annotations + +import re +from collections.abc import Sequence +from dataclasses import dataclass + +from aai_cli.errors import UsageError + +# A rendered transcript line: "Speaker A: text". The id is a non-space run; the +# colon may be followed by a single space (label-only lines can lack trailing text). +_LABEL_RE = re.compile(r"^Speaker (?P\S+): ?(?P.*)$") + + +@dataclass(frozen=True) +class Segment: + """One speaker turn after labels are stripped and wrapped lines folded in.""" + + speaker_id: str + text: str + + +def looks_like_speaker_labeled(text: str) -> bool: + """True when the first non-blank line is a ``Speaker :`` label line.""" + for line in text.splitlines(): + if line.strip(): + return _LABEL_RE.match(line) is not None + return False + + +def parse_segments(text: str) -> list[Segment]: + """Parse speaker-labeled text into merged per-turn segments. + + A ``Speaker :`` line starts a turn; a following line that is not itself a + label is a wrapped continuation appended to the current turn (joined with a + single space). Consecutive same-speaker turns merge; empty turns are dropped. + """ + turns: list[Segment] = [] + current_id: str | None = None + parts: list[str] = [] + + def flush() -> None: + if current_id is not None: + turns.append(Segment(current_id, " ".join(p for p in parts if p))) + + for line in text.splitlines(): + stripped = line.strip() + if not stripped: + continue + match = _LABEL_RE.match(line) + if match: + flush() + current_id = match.group("id") + parts = [match.group("text").strip()] + elif current_id is not None: + parts.append(stripped) + flush() + + merged: list[Segment] = [] + for turn in turns: + if merged and merged[-1].speaker_id == turn.speaker_id: + joined = f"{merged[-1].text} {turn.text}" + merged[-1] = Segment(turn.speaker_id, joined) + else: + merged.append(turn) + return [turn for turn in merged if turn.text] + + +DEFAULT_VOICE_ROTATION = ("jane", "michael", "mary", "paul", "eve", "george") + + +def parse_voice_overrides(values: list[str]) -> tuple[str | None, dict[str, str]]: + """Split repeatable ``--voice`` values into ``(bare_voice, {speaker_id: voice})``. + + A value containing ``=`` is a ``SPEAKER=VOICE`` mapping (ids casefolded so the + match is case-insensitive); a bare value sets the single-voice default, last + one wins. Raises ``UsageError`` on a malformed pair (empty side). + """ + bare: str | None = None + overrides: dict[str, str] = {} + for value in values: + if "=" in value: + speaker, _, voice = value.partition("=") + speaker, voice = speaker.strip(), voice.strip() + if not speaker or not voice: + raise UsageError( + f"Invalid --voice mapping {value!r}.", + suggestion="Use SPEAKER=VOICE, e.g. --voice A=jane.", + ) + overrides[speaker.casefold()] = voice + elif value.strip(): + bare = value.strip() + return bare, overrides + + +def assign_voices( + segments: list[Segment], + rotation: Sequence[str], + overrides: dict[str, str], +) -> tuple[list[tuple[str, str]], dict[str, str]]: + """Resolve each segment to ``(voice, text)`` and return the id→voice map. + + A speaker uses its override if present, else the next rotation voice in + first-appearance order (wrapping). Overrides do not consume rotation slots. + The returned map is ordered by first appearance. + """ + id_to_voice: dict[str, str] = {} + next_index = 0 + resolved: list[tuple[str, str]] = [] + for segment in segments: + if segment.speaker_id not in id_to_voice: + override = overrides.get(segment.speaker_id.casefold()) + if override is not None: + id_to_voice[segment.speaker_id] = override + else: + id_to_voice[segment.speaker_id] = rotation[next_index % len(rotation)] + next_index += 1 + resolved.append((id_to_voice[segment.speaker_id], segment.text)) + return resolved, id_to_voice diff --git a/aai_cli/tts/session.py b/aai_cli/tts/session.py new file mode 100644 index 00000000..01d2767f --- /dev/null +++ b/aai_cli/tts/session.py @@ -0,0 +1,219 @@ +from __future__ import annotations + +import base64 +import contextlib +import json +import logging +from collections.abc import Callable +from dataclasses import dataclass +from typing import Protocol +from urllib.parse import urlencode + +from aai_cli import environments +from aai_cli.errors import APIError, CLIError, auth_failure, is_auth_failure +from aai_cli.tts import audio + + +class _WebSocket(Protocol): + """The slice of a websockets sync connection this module drives — named as a + Protocol so the untyped library boundary is structurally typed, not opaque.""" + + def recv(self) -> str | bytes: ... + def send(self, data: str, /) -> None: ... # positional-only: matches ws send(message) + def close(self) -> None: ... + + +# The connect factory: returns a fresh _WebSocket. websockets' real sync client +# matches structurally; tests inject a fake with the same surface. +_Connect = Callable[..., _WebSocket] + +# A pre-upgrade HTTP 403 on the WebSocket handshake is NOT a rejected key (it also +# covers WAF/region/plan blocks) — mirrors how agent/stream classify handshakes. +_HTTP_FORBIDDEN = 403 + +# websockets' sync client logs teardown errors through these; silence them so an +# internal "stream ended" traceback never lands on stderr next to our clean error. +_WEBSOCKETS_LOGGERS = ("websockets", "websockets.client") + +# The streaming-TTS server synthesizes at 24 kHz unless a sample_rate is requested, +# and echoes the resolved value back in the Begin frame's configuration. Audio frames +# carry only the PCM payload, so Begin is the single source of truth for the rate; this +# is the fallback if that field is ever absent. +_DEFAULT_SAMPLE_RATE = 24000 + +# Pause inserted between speaker turns in a multi-voice dialogue, for natural pacing. +_INTER_TURN_SILENCE_SECONDS = 0.25 + + +@dataclass(frozen=True) +class SpeakConfig: + """Per-run TTS parameters. Optional fields are only sent when set, so the + server applies its own defaults (voice/language/sample_rate) otherwise.""" + + text: str + voice: str | None = None + language: str | None = None + sample_rate: int | None = None + + def query_params(self) -> dict[str, str]: + params: dict[str, str] = {} + if self.voice is not None: + params["voice"] = self.voice + if self.language is not None: + params["language"] = self.language + if self.sample_rate is not None: + params["sample_rate"] = str(self.sample_rate) + return params + + +@dataclass(frozen=True) +class SpeakResult: + """The synthesized audio: raw 16-bit mono PCM plus its sample rate.""" + + pcm: bytes + sample_rate: int + audio_duration_seconds: float + + +def is_available() -> bool: + """True only where the active environment has a streaming-TTS host (sandbox).""" + return bool(environments.active().streaming_tts_host) + + +def ws_url(params: dict[str, str]) -> str: + """The streaming-TTS socket URL for the active environment, with query params.""" + base = f"wss://{environments.active().streaming_tts_host}/v1/ws/" + return f"{base}?{urlencode(params)}" if params else base + + +def _silence_websockets_logging() -> None: + """Keep websockets' internal teardown logging off the user's stderr.""" + for name in _WEBSOCKETS_LOGGERS: + logging.getLogger(name).setLevel(logging.CRITICAL) + + +def _is_rejected_key(exc: Exception) -> bool: + """Auth-shaped failure (the key was rejected) vs a generic block. A plain HTTP + 403 on the upgrade is NOT a rejected key (WAF/region/plan also 403).""" + status = getattr(getattr(exc, "response", None), "status_code", None) + if status == _HTTP_FORBIDDEN: + return False + return is_auth_failure(exc) + + +def _auth_or_api_error(exc: Exception, message: str) -> CLIError: + """A rejected key becomes auth_failure(); anything else becomes an APIError.""" + if _is_rejected_key(exc): + return auth_failure() + return APIError(f"{message}: {exc}") + + +def _default_connect( + url: str, *, additional_headers: dict[str, str], max_size: int | None +) -> _WebSocket: + """The real websockets sync client, imported lazily so tests can inject a fake.""" + from websockets.sync.client import connect + + return connect(url, additional_headers=additional_headers, max_size=max_size) + + +def _open_ws(connect: _Connect, api_key: str, url: str) -> _WebSocket: + """Open the TTS socket, mapping a connect failure to a clean CLIError.""" + try: + return connect( + url, + additional_headers={"Authorization": f"Bearer {api_key}"}, + max_size=None, + ) + except Exception as exc: + raise _auth_or_api_error(exc, "Could not connect to the TTS service") from exc + + +def _run_protocol( + ws: _WebSocket, config: SpeakConfig, on_warning: Callable[[str], None] | None +) -> SpeakResult: + """Send Generate + ForceFlushTextBuffer, collect Audio until is_final, then Terminate.""" + begin = json.loads(ws.recv()) + if begin.get("type") != "Begin": + raise APIError(f"TTS service did not start the session (got {begin.get('type')!r}).") + sample_rate = int(begin.get("configuration", {}).get("sample_rate", _DEFAULT_SAMPLE_RATE)) + + ws.send(json.dumps({"type": "Generate", "text": config.text})) + ws.send(json.dumps({"type": "ForceFlushTextBuffer"})) + + pcm = bytearray() + while True: + msg = json.loads(ws.recv()) + mtype = msg.get("type") + if mtype == "Audio": + pcm.extend(base64.b64decode(msg["audio"])) + if msg.get("is_final"): + break + elif mtype == "Error": + raise APIError( + f"TTS error ({msg.get('error_code', '')}): {msg.get('error') or 'unknown'}" + ) + elif mtype == "Warning" and on_warning is not None: + on_warning(str(msg.get("warning", ""))) + + with contextlib.suppress(Exception): + ws.send(json.dumps({"type": "Terminate"})) + + duration = len(pcm) / 2 / sample_rate + return SpeakResult(bytes(pcm), sample_rate, duration) + + +def synthesize( + api_key: str, + config: SpeakConfig, + *, + connect: _Connect | None = None, + on_warning: Callable[[str], None] | None = None, +) -> SpeakResult: + """Open the streaming-TTS socket and synthesize ``config.text`` to PCM. + + ``connect`` defaults to websockets' synchronous client; injectable for tests. + Connect/session failures map to a clean CLIError (a rejected key -> exit 4). + """ + _silence_websockets_logging() + if connect is None: + connect = _default_connect + + ws = _open_ws(connect, api_key, ws_url(config.query_params())) + try: + return _run_protocol(ws, config, on_warning) + except (CLIError, KeyboardInterrupt, BrokenPipeError): + raise # clean CLI errors, Ctrl-C, and a closed pipe are handled upstream + except Exception as exc: + raise _auth_or_api_error(exc, "TTS session failed") from exc + finally: + with contextlib.suppress(Exception): + ws.close() + + +def synthesize_dialogue( + api_key: str, + segments: list[tuple[str, str]], + *, + language: str | None = None, + sample_rate: int | None = None, + connect: _Connect | None = None, + on_warning: Callable[[str], None] | None = None, +) -> SpeakResult: + """Synthesize each ``(voice, text)`` segment and concatenate the PCM. + + Each segment opens its own connection (the voice is fixed at connect time). + A short silence is inserted between turns — never at the ends. The result's + sample rate is the rate the server reported for the segments. + """ + pcm = bytearray() + sample_rate_out = _DEFAULT_SAMPLE_RATE + for index, (voice, text) in enumerate(segments): + config = SpeakConfig(text=text, voice=voice, language=language, sample_rate=sample_rate) + result = synthesize(api_key, config, connect=connect, on_warning=on_warning) + if index: + pcm.extend(audio.silence(result.sample_rate, _INTER_TURN_SILENCE_SECONDS)) + pcm.extend(result.pcm) + sample_rate_out = result.sample_rate + duration = len(pcm) / 2 / sample_rate_out + return SpeakResult(bytes(pcm), sample_rate_out, duration) diff --git a/docs/superpowers/plans/2026-06-10-aai-speak-speaker-voices.md b/docs/superpowers/plans/2026-06-10-aai-speak-speaker-voices.md new file mode 100644 index 00000000..83c0af75 --- /dev/null +++ b/docs/superpowers/plans/2026-06-10-aai-speak-speaker-voices.md @@ -0,0 +1,884 @@ +# `aai speak` Speaker-Aware Multi-Voice Playback — Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** When `aai speak` is fed speaker-labeled transcript text (`aai transcribe … --speaker-labels`), strip the `Speaker X:` labels and synthesize each speaker in a different voice, concatenated with short pauses; plain text keeps today's single-voice behavior. + +**Architecture:** A new pure module `aai_cli/tts/dialogue.py` does all detection/parsing/voice-assignment (no I/O). `session.synthesize_dialogue` loops the existing `session.synthesize` per segment and concatenates PCM with an `audio.silence` gap. `commands/speak.py` auto-detects labeled input and dispatches to a dialogue path or the existing single-voice path. Detection is automatic; `--voice` becomes repeatable and accepts `SPEAKER=VOICE` overrides. + +**Tech Stack:** Python 3.12+, Typer, the `websockets` sync client (already wired in `session.py`), pytest. Spec: `docs/superpowers/specs/2026-06-10-aai-speak-speaker-voices-design.md`. + +**Conventions (read before starting):** + +- Every module starts with `from __future__ import annotations`; modern typing (`X | None`). +- Run all tooling through `uv run`. The gate is `./scripts/check.sh` (mutation + 100% patch coverage tail gates — assert *behavior*, not just execution). +- The post-edit hook auto-runs `ruff --fix` + `ruff format` on saved `*.py`, so don't hand-format. +- `errors.UsageError(message, *, suggestion=...)` → exit 2. `errors.CLIError` is its base. +- Help/CLI output is pinned by syrupy `.ambr` snapshots — regenerate with `--snapshot-update`, never hand-edit. + +--- + +## File Structure + +- **Create** `aai_cli/tts/dialogue.py` — `Segment` dataclass, `looks_like_speaker_labeled`, `parse_segments`, `parse_voice_overrides`, `assign_voices`, `DEFAULT_VOICE_ROTATION`. Pure, no I/O. +- **Create** `tests/test_tts_dialogue.py` — unit tests for the above. +- **Modify** `aai_cli/tts/audio.py` — add `silence(sample_rate, seconds) -> bytes`. +- **Modify** `tests/test_tts_audio.py` — test `silence`. +- **Modify** `aai_cli/tts/session.py` — add `synthesize_dialogue(...)` + `_INTER_TURN_SILENCE_SECONDS`. +- **Modify** `tests/test_tts_session.py` — test `synthesize_dialogue`. +- **Modify** `aai_cli/commands/speak.py` — repeatable `--voice`, detection/dispatch, multi-result emit. +- **Modify** `tests/test_speak.py` — multi-voice command tests. +- **Modify** `tests/__snapshots__/test_cli_output_snapshots.ambr` — regenerate `aai speak --help`. + +--- + +## Task 1: `dialogue.py` — detection + segment parsing + +**Files:** + +- Create: `aai_cli/tts/dialogue.py` +- Test: `tests/test_tts_dialogue.py` + +- [ ] **Step 1: Write the failing tests** + +Create `tests/test_tts_dialogue.py`: + +```python +from __future__ import annotations + +from aai_cli.tts import dialogue + + +def test_detects_labeled_input_on_first_nonblank_line(): + assert dialogue.looks_like_speaker_labeled("Speaker A: hi\nSpeaker B: yo") is True + # Leading blank lines are skipped before the check. + assert dialogue.looks_like_speaker_labeled("\n\nSpeaker A: hi") is True + + +def test_plain_prose_is_not_labeled(): + assert dialogue.looks_like_speaker_labeled("Hello there, friend.") is False + # A mid-sentence "Speaker" word must not trigger detection. + assert dialogue.looks_like_speaker_labeled("The Speaker said hello") is False + assert dialogue.looks_like_speaker_labeled("") is False + + +def test_parse_single_line_turns(): + segs = dialogue.parse_segments("Speaker A: Hello.\nSpeaker B: Hi there.") + assert [(s.speaker_id, s.text) for s in segs] == [("A", "Hello."), ("B", "Hi there.")] + + +def test_parse_folds_wrapped_continuation_lines(): + # An utterance wrapped across physical lines (no label on the 2nd line) folds + # back into one segment, joined with single spaces. + text = "Speaker A: This is a long line that wrapped\nonto a second line here\nSpeaker B: Ok." + segs = dialogue.parse_segments(text) + assert [(s.speaker_id, s.text) for s in segs] == [ + ("A", "This is a long line that wrapped onto a second line here"), + ("B", "Ok."), + ] + + +def test_parse_merges_consecutive_same_speaker_turns(): + segs = dialogue.parse_segments("Speaker A: One.\nSpeaker A: Two.\nSpeaker B: Three.") + assert [(s.speaker_id, s.text) for s in segs] == [("A", "One. Two."), ("B", "Three.")] + + +def test_parse_skips_blank_lines_and_drops_empty_turns(): + segs = dialogue.parse_segments("Speaker A: Hi.\n\nSpeaker B: \nSpeaker A: Bye.") + # Speaker B's empty turn is dropped; the two A turns are not merged (B is between). + assert [(s.speaker_id, s.text) for s in segs] == [("A", "Hi."), ("A", "Bye.")] +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/test_tts_dialogue.py -q` +Expected: FAIL — `ModuleNotFoundError: No module named 'aai_cli.tts.dialogue'`. + +- [ ] **Step 3: Implement `dialogue.py` (detection + parsing portion)** + +Create `aai_cli/tts/dialogue.py`: + +```python +from __future__ import annotations + +import re +from dataclasses import dataclass + +# A rendered transcript line: "Speaker A: text". The id is a non-space run; the +# colon may be followed by a single space (label-only lines can lack trailing text). +_LABEL_RE = re.compile(r"^Speaker (?P\S+): ?(?P.*)$") + + +@dataclass(frozen=True) +class Segment: + """One speaker turn after labels are stripped and wrapped lines folded in.""" + + speaker_id: str + text: str + + +def looks_like_speaker_labeled(text: str) -> bool: + """True when the first non-blank line is a ``Speaker :`` label line.""" + for line in text.splitlines(): + if line.strip(): + return _LABEL_RE.match(line) is not None + return False + + +def parse_segments(text: str) -> list[Segment]: + """Parse speaker-labeled text into merged per-turn segments. + + A ``Speaker :`` line starts a turn; a following line that is not itself a + label is a wrapped continuation appended to the current turn (joined with a + single space). Consecutive same-speaker turns merge; empty turns are dropped. + """ + turns: list[Segment] = [] + current_id: str | None = None + parts: list[str] = [] + + def flush() -> None: + if current_id is not None: + turns.append(Segment(current_id, " ".join(p for p in parts if p))) + + for line in text.splitlines(): + stripped = line.strip() + if not stripped: + continue + match = _LABEL_RE.match(line) + if match: + flush() + current_id = match.group("id") + parts = [match.group("text").strip()] + elif current_id is not None: + parts.append(stripped) + flush() + + merged: list[Segment] = [] + for turn in turns: + if merged and merged[-1].speaker_id == turn.speaker_id: + joined = f"{merged[-1].text} {turn.text}".strip() + merged[-1] = Segment(turn.speaker_id, joined) + else: + merged.append(turn) + return [turn for turn in merged if turn.text] +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/test_tts_dialogue.py -q` +Expected: PASS (6 tests). + +- [ ] **Step 5: Commit** + +```bash +git add aai_cli/tts/dialogue.py tests/test_tts_dialogue.py +git commit -m "feat(speak): parse speaker-labeled transcript text into segments" +``` + +--- + +## Task 2: `dialogue.py` — voice overrides + assignment + +**Files:** + +- Modify: `aai_cli/tts/dialogue.py` +- Test: `tests/test_tts_dialogue.py` + +- [ ] **Step 1: Write the failing tests** + +Append to `tests/test_tts_dialogue.py`: + +```python +import pytest + +from aai_cli.errors import UsageError + + +def test_parse_voice_overrides_splits_bare_and_mapped(): + bare, overrides = dialogue.parse_voice_overrides(["A=vera", "mary", "B=paul"]) + assert bare == "mary" + assert overrides == {"a": "vera", "b": "paul"} # ids casefolded + + +def test_parse_voice_overrides_bare_last_wins_and_empty_default(): + assert dialogue.parse_voice_overrides([]) == (None, {}) + assert dialogue.parse_voice_overrides(["jane", "mary"]) == ("mary", {}) + + +def test_parse_voice_overrides_rejects_malformed_pair(): + for bad in ["=vera", "A=", " = "]: + with pytest.raises(UsageError): + dialogue.parse_voice_overrides([bad]) + + +def test_assign_voices_rotates_in_first_appearance_order(): + segs = [dialogue.Segment(s, "x") for s in ("A", "B", "A", "C")] + resolved, mapping = dialogue.assign_voices(segs, ["jane", "michael", "mary"], {}) + assert [v for v, _ in resolved] == ["jane", "michael", "jane", "mary"] + assert mapping == {"A": "jane", "B": "michael", "C": "mary"} + + +def test_assign_voices_wraps_past_rotation_length(): + segs = [dialogue.Segment(s, "x") for s in ("A", "B", "C")] + resolved, _ = dialogue.assign_voices(segs, ["jane", "michael"], {}) + assert [v for v, _ in resolved] == ["jane", "michael", "jane"] + + +def test_assign_voices_override_beats_rotation_without_consuming_a_slot(): + segs = [dialogue.Segment(s, "x") for s in ("A", "B")] + # A is overridden, so B still gets the FIRST rotation voice, not the second. + resolved, mapping = dialogue.assign_voices(segs, ["jane", "michael"], {"a": "vera"}) + assert [v for v, _ in resolved] == ["vera", "jane"] + assert mapping == {"A": "vera", "B": "jane"} + + +def test_default_rotation_is_the_confirmed_working_voices(): + assert dialogue.DEFAULT_VOICE_ROTATION == ("jane", "michael", "mary", "paul", "eve", "george") +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/test_tts_dialogue.py -q` +Expected: FAIL — `AttributeError: module 'aai_cli.tts.dialogue' has no attribute 'parse_voice_overrides'`. + +- [ ] **Step 3: Implement the assignment portion** + +Add to the imports at the top of `aai_cli/tts/dialogue.py`: + +```python +from collections.abc import Sequence + +from aai_cli.errors import UsageError +``` + +Append to `aai_cli/tts/dialogue.py`: + +```python +DEFAULT_VOICE_ROTATION = ("jane", "michael", "mary", "paul", "eve", "george") + + +def parse_voice_overrides(values: list[str]) -> tuple[str | None, dict[str, str]]: + """Split repeatable ``--voice`` values into ``(bare_voice, {speaker_id: voice})``. + + A value containing ``=`` is a ``SPEAKER=VOICE`` mapping (ids casefolded so the + match is case-insensitive); a bare value sets the single-voice default, last + one wins. Raises ``UsageError`` on a malformed pair (empty side). + """ + bare: str | None = None + overrides: dict[str, str] = {} + for value in values: + if "=" in value: + speaker, _, voice = value.partition("=") + speaker, voice = speaker.strip(), voice.strip() + if not speaker or not voice: + raise UsageError( + f"Invalid --voice mapping {value!r}.", + suggestion="Use SPEAKER=VOICE, e.g. --voice A=jane.", + ) + overrides[speaker.casefold()] = voice + elif value.strip(): + bare = value.strip() + return bare, overrides + + +def assign_voices( + segments: list[Segment], + rotation: Sequence[str], + overrides: dict[str, str], +) -> tuple[list[tuple[str, str]], dict[str, str]]: + """Resolve each segment to ``(voice, text)`` and return the id→voice map. + + A speaker uses its override if present, else the next rotation voice in + first-appearance order (wrapping). Overrides do not consume rotation slots. + The returned map is ordered by first appearance. + """ + id_to_voice: dict[str, str] = {} + next_index = 0 + resolved: list[tuple[str, str]] = [] + for segment in segments: + if segment.speaker_id not in id_to_voice: + override = overrides.get(segment.speaker_id.casefold()) + if override is not None: + id_to_voice[segment.speaker_id] = override + else: + id_to_voice[segment.speaker_id] = rotation[next_index % len(rotation)] + next_index += 1 + resolved.append((id_to_voice[segment.speaker_id], segment.text)) + return resolved, id_to_voice +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/test_tts_dialogue.py -q` +Expected: PASS (13 tests total). + +- [ ] **Step 5: Commit** + +```bash +git add aai_cli/tts/dialogue.py tests/test_tts_dialogue.py +git commit -m "feat(speak): resolve speaker voices via rotation + overrides" +``` + +--- + +## Task 3: `audio.silence` helper + +**Files:** + +- Modify: `aai_cli/tts/audio.py` +- Test: `tests/test_tts_audio.py` + +- [ ] **Step 1: Write the failing test** + +Append to `tests/test_tts_audio.py`: + +```python +def test_silence_returns_zeroed_pcm_of_the_right_length(): + # 16-bit mono: 100 ms at 16 kHz = 1600 frames = 3200 zero bytes. + pcm = audio.silence(16000, 0.1) + assert pcm == b"\x00" * 3200 + # Empty duration -> no bytes. + assert audio.silence(24000, 0.0) == b"" +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/test_tts_audio.py::test_silence_returns_zeroed_pcm_of_the_right_length -q` +Expected: FAIL — `AttributeError: module 'aai_cli.tts.audio' has no attribute 'silence'`. + +- [ ] **Step 3: Implement `silence`** + +Add to `aai_cli/tts/audio.py` (after `write_wav`, before `_default_output_stream`): + +```python +def silence(sample_rate: int, seconds: float) -> bytes: + """Zeroed 16-bit mono PCM of the given duration (2 bytes per frame).""" + return b"\x00" * (int(sample_rate * seconds) * 2) +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `uv run pytest tests/test_tts_audio.py -q` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add aai_cli/tts/audio.py tests/test_tts_audio.py +git commit -m "feat(speak): add silence() PCM helper for inter-turn gaps" +``` + +--- + +## Task 4: `session.synthesize_dialogue` + +**Files:** + +- Modify: `aai_cli/tts/session.py` +- Test: `tests/test_tts_session.py` + +- [ ] **Step 1: Write the failing tests** + +Append to `tests/test_tts_session.py` (reuses `FakeWS`, `_begin_frame`, `_audio_frame` already in that file): + +```python +def test_synthesize_dialogue_concatenates_segments_with_silence(): + # One fresh fake socket per segment; record the voice each connection requested. + sockets = [ + FakeWS([_begin_frame(sample_rate=24000), _audio_frame(b"\xaa\xbb", final=True)]), + FakeWS([_begin_frame(sample_rate=24000), _audio_frame(b"\xcc\xdd", final=True)]), + ] + urls: list[str] = [] + + def _connect(url: str, **_kwargs): + urls.append(url) + return sockets.pop(0) + + result = session.synthesize_dialogue( + "k", + [("jane", "Hello."), ("michael", "Hi.")], + language="English", + connect=_connect, + ) + # Each segment connected with its own voice. + assert "voice=jane" in urls[0] + assert "voice=michael" in urls[1] + # 0.25 s of silence (24000 * 0.25 * 2 = 12000 zero bytes) sits BETWEEN the two + # segments' PCM, with none at the ends. + gap = b"\x00" * 12000 + assert result.pcm == b"\xaa\xbb" + gap + b"\xcc\xdd" + assert result.sample_rate == 24000 + + +def test_synthesize_dialogue_single_segment_has_no_silence(): + ws = FakeWS([_begin_frame(sample_rate=24000), _audio_frame(b"\x01\x02", final=True)]) + result = session.synthesize_dialogue("k", [("jane", "Hi.")], connect=lambda *a, **k: ws) + assert result.pcm == b"\x01\x02" # no leading/trailing pad +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/test_tts_session.py -q -k dialogue` +Expected: FAIL — `AttributeError: module 'aai_cli.tts.session' has no attribute 'synthesize_dialogue'`. + +- [ ] **Step 3: Implement `synthesize_dialogue`** + +Add `from aai_cli.tts import audio` to the imports in `aai_cli/tts/session.py` (top, with the other `aai_cli` imports). + +Add a constant near `_DEFAULT_SAMPLE_RATE`: + +```python +# Pause inserted between speaker turns in a multi-voice dialogue, for natural pacing. +_INTER_TURN_SILENCE_SECONDS = 0.25 +``` + +Append after `synthesize`: + +```python +def synthesize_dialogue( + api_key: str, + segments: list[tuple[str, str]], + *, + language: str | None = None, + sample_rate: int | None = None, + connect: _Connect | None = None, + on_warning: Callable[[str], None] | None = None, +) -> SpeakResult: + """Synthesize each ``(voice, text)`` segment and concatenate the PCM. + + Each segment opens its own connection (the voice is fixed at connect time). + A short silence is inserted between turns — never at the ends. The result's + sample rate is the rate the server reported for the segments. + """ + pcm = bytearray() + sample_rate_out = _DEFAULT_SAMPLE_RATE + for index, (voice, text) in enumerate(segments): + config = SpeakConfig(text=text, voice=voice, language=language, sample_rate=sample_rate) + result = synthesize(api_key, config, connect=connect, on_warning=on_warning) + if index: + pcm.extend(audio.silence(result.sample_rate, _INTER_TURN_SILENCE_SECONDS)) + pcm.extend(result.pcm) + sample_rate_out = result.sample_rate + duration = len(pcm) / 2 / sample_rate_out + return SpeakResult(bytes(pcm), sample_rate_out, duration) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/test_tts_session.py -q` +Expected: PASS (all, including the two new ones). + +- [ ] **Step 5: Commit** + +```bash +git add aai_cli/tts/session.py tests/test_tts_session.py +git commit -m "feat(speak): synthesize_dialogue concatenates per-voice segments" +``` + +--- + +## Task 5: Wire detection + dispatch into `commands/speak.py` + +**Files:** + +- Modify: `aai_cli/commands/speak.py` +- Test: `tests/test_speak.py` + +- [ ] **Step 1: Write the failing tests** + +Append to `tests/test_speak.py`: + +```python +from aai_cli.tts import session as tts_session + + +@pytest.fixture +def fake_dialogue(monkeypatch: pytest.MonkeyPatch): + calls: dict[str, object] = {} + + def _fake(api_key, segments, *, language=None, sample_rate=None, connect=None, on_warning=None): + calls["segments"] = segments + calls["language"] = language + return tts_session.SpeakResult( + pcm=b"\x01\x02", sample_rate=24000, audio_duration_seconds=1.5 + ) + + monkeypatch.setattr(tts_session, "synthesize_dialogue", _fake) + monkeypatch.setattr("aai_cli.commands.speak.audio.play_pcm", lambda *a, **k: None) + return calls + + +def test_labeled_stdin_uses_dialogue_path_with_default_rotation(fake_dialogue): + text = "Speaker A: Hello there.\nSpeaker B: Hi.\nSpeaker A: Bye." + result = runner.invoke(app, ["--sandbox", "speak"], input=text) + assert result.exit_code == 0 + # Labels stripped; consecutive A turns are NOT merged (B between); voices rotate. + assert fake_dialogue["segments"] == [ + ("jane", "Hello there."), + ("michael", "Hi."), + ("jane", "Bye."), + ] + + +def test_speaker_voice_override_is_applied(fake_dialogue): + text = "Speaker A: One.\nSpeaker B: Two." + result = runner.invoke( + app, ["--sandbox", "speak", "--voice", "A=vera", "--voice", "B=paul"], input=text + ) + assert result.exit_code == 0 + assert fake_dialogue["segments"] == [("vera", "One."), ("paul", "Two.")] + + +def test_bare_voice_in_dialogue_mode_is_ignored_with_a_note(fake_dialogue): + text = "Speaker A: One.\nSpeaker B: Two." + result = runner.invoke(app, ["--sandbox", "speak", "--voice", "mary"], input=text) + assert result.exit_code == 0 + # The rotation still drives voices (bare voice ignored)... + assert fake_dialogue["segments"] == [("jane", "One."), ("michael", "Two.")] + # ...and the user is told why, pointed at the per-speaker form. + assert "A=NAME" in result.stderr + + +def test_dialogue_json_reports_speaker_voice_map(fake_dialogue): + text = "Speaker A: One.\nSpeaker B: Two." + result = runner.invoke(app, ["--sandbox", "speak", "--json"], input=text) + assert result.exit_code == 0 + payload = json.loads(result.stdout.strip()) + assert payload["mode"] == "multi" + assert payload["speakers"] == {"A": "jane", "B": "michael"} + assert payload["segments"] == 2 + assert payload["sample_rate"] == 24000 + + +def test_unlabeled_text_still_uses_single_voice_path(fake_synthesize, monkeypatch): + # A bare --voice still selects the single-voice voice for ordinary prose. + monkeypatch.setattr("aai_cli.commands.speak.audio.play_pcm", lambda *a, **k: None) + result = runner.invoke(app, ["--sandbox", "speak", "Just prose.", "--voice", "mary"]) + assert result.exit_code == 0 + assert fake_synthesize["cfg"].voice == "mary" + assert fake_synthesize["cfg"].text == "Just prose." +``` + +Note: the existing `test_voice_and_language_flow_into_config` passes `--voice jane` (a bare value) with prose — that keeps working because a bare `--voice` still selects the single-voice voice. No change needed there. + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/test_speak.py -q` +Expected: FAIL — the new tests error (e.g. `--voice` not repeatable / dialogue path absent), single-voice tests still pass. + +- [ ] **Step 3: Rewrite `commands/speak.py`** + +Replace the entire file `aai_cli/commands/speak.py` with: + +```python +from __future__ import annotations + +import sys +from pathlib import Path + +import typer + +from aai_cli import config, 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 +from aai_cli.tts import audio, dialogue, session + +app = typer.Typer() + +# The streaming-TTS reference client defaults to the PocketTTS "jane" voice and +# English, so the CLI sends the same and a bare `aai speak` works out of the box. +# Override either with --voice/--language. +DEFAULT_VOICE = "jane" +DEFAULT_LANGUAGE = "English" + + +def _read_text(text: str | None) -> str: + """The text to speak: the argument if non-blank, else stdin when piped.""" + if text is not None and text.strip(): + return text + if text is None and not sys.stdin.isatty(): + piped = sys.stdin.read().strip() + if piped: + return piped + raise UsageError( + "No text to speak.", + suggestion='Pass text as an argument: aai speak "Hello" — or pipe it via stdin.', + ) + + +def _output_audio(result: session.SpeakResult, out: Path | None) -> None: + """Write a WAV when --out is given, else play through the speakers.""" + if out is not None: + audio.write_wav(out, result.pcm, result.sample_rate) + else: + audio.play_pcm(result.pcm, result.sample_rate) + + +def _disposition(out: Path | None) -> str: + return f"saved to {out}" if out is not None else "played" + + +def _emit_single( + result: session.SpeakResult, + cfg: session.SpeakConfig, + out: Path | None, + *, + json_mode: bool, +) -> None: + """Single-voice result: a JSON object on stdout, or a human note on stderr.""" + duration = round(result.audio_duration_seconds, 3) + if json_mode: + output.emit_ndjson( + { + "voice": cfg.voice, + "language": cfg.language, + "sample_rate": result.sample_rate, + "audio_duration_seconds": duration, + "bytes": len(result.pcm), + "out": str(out) if out is not None else None, + } + ) + return + output.error_console.print( + f"[aai.muted]Spoke {duration}s of audio ({_disposition(out)}).[/aai.muted]" + ) + + +def _emit_multi( + result: session.SpeakResult, + speakers: dict[str, str], + segment_count: int, + out: Path | None, + *, + json_mode: bool, +) -> None: + """Multi-voice result: a JSON object on stdout, or a human note on stderr.""" + duration = round(result.audio_duration_seconds, 3) + if json_mode: + output.emit_ndjson( + { + "mode": "multi", + "speakers": speakers, + "segments": segment_count, + "sample_rate": result.sample_rate, + "audio_duration_seconds": duration, + "bytes": len(result.pcm), + "out": str(out) if out is not None else None, + } + ) + return + voices = ", ".join(f"{spk}={voice}" for spk, voice in speakers.items()) + output.error_console.print( + f"[aai.muted]Spoke {duration}s across {len(speakers)} voices " + f"({voices}) ({_disposition(out)}).[/aai.muted]" + ) + + +def _speak_single( + api_key: str, + text: str, + voice: str, + language: str, + sample_rate: int | None, + out: Path | None, + *, + json_mode: bool, + quiet: bool, +) -> None: + cfg = session.SpeakConfig(text=text, voice=voice, language=language, sample_rate=sample_rate) + with output.status("Synthesizing speech…", json_mode=json_mode, quiet=quiet): + result = session.synthesize( + api_key, cfg, on_warning=lambda m: output.emit_warning(m, json_mode=json_mode) + ) + _output_audio(result, out) + _emit_single(result, cfg, out, json_mode=json_mode) + + +def _speak_dialogue( + api_key: str, + text: str, + bare_voice: str | None, + overrides: dict[str, str], + language: str, + sample_rate: int | None, + out: Path | None, + *, + json_mode: bool, + quiet: bool, +) -> None: + segments = dialogue.parse_segments(text) + if not segments: + raise UsageError( + "No text to speak.", + suggestion="The input had speaker labels but no spoken text.", + ) + if bare_voice is not None: + output.emit_warning( + "Ignoring bare --voice in multi-speaker mode; " + "set a voice per speaker with --voice A=NAME.", + json_mode=json_mode, + ) + resolved, speakers = dialogue.assign_voices( + segments, dialogue.DEFAULT_VOICE_ROTATION, overrides + ) + with output.status("Synthesizing speech…", json_mode=json_mode, quiet=quiet): + result = session.synthesize_dialogue( + api_key, + resolved, + language=language, + sample_rate=sample_rate, + on_warning=lambda m: output.emit_warning(m, json_mode=json_mode), + ) + _output_audio(result, out) + _emit_multi(result, speakers, len(resolved), out, json_mode=json_mode) + + +@app.command( + rich_help_panel=help_panels.TRANSCRIPTION, + epilog=examples_epilog( + [ + ("Speak text aloud (sandbox only)", 'aai speak "Hello there, friend." --sandbox'), + ( + "Pick a voice and language", + 'aai speak "Bonjour" --voice jane --language French --sandbox', + ), + ( + "Speak a diarized transcript, one voice per speaker", + "aai transcribe meeting.mp3 --speaker-labels | aai speak --sandbox", + ), + ( + "Override a speaker's voice", + "… | aai speak --voice A=vera --voice B=paul --sandbox", + ), + ( + "Save to a WAV instead of playing", + 'aai speak "Hello" --out /tmp/hello.wav --sandbox', + ), + ] + ), +) +def speak( + ctx: typer.Context, + text: str | None = typer.Argument(None, help="Text to speak. Omit to read from stdin."), + voice: list[str] = typer.Option( + [], + "--voice", + help="Voice id, or SPEAKER=VOICE for diarized input (repeatable, e.g. --voice A=jane).", + ), + language: str = typer.Option(DEFAULT_LANGUAGE, "--language", help="Language of the text."), + sample_rate: int | None = typer.Option( + None, "--sample-rate", help="Output sample rate in Hz. Server default if omitted." + ), + out: Path | None = typer.Option( + None, "--out", help="Write a WAV file instead of playing through the speakers." + ), + json_out: bool = options.json_option("Emit JSON metadata about the synthesized audio."), +) -> None: + """Synthesize speech from text with AssemblyAI streaming TTS (sandbox only). + + Plays the audio through your speakers by default, or writes a WAV with --out. + Speaker-labeled input (from 'aai transcribe --speaker-labels') is detected + automatically: the labels are stripped and each speaker gets a different + voice. This feature only exists in the sandbox today — run it with --sandbox. + """ + + def body(state: AppState, json_mode: bool) -> None: + if not session.is_available(): + raise CLIError( + "aai speak is only available in the sandbox.", + error_type="unsupported_environment", + exit_code=2, + suggestion="Re-run with --sandbox (or --env sandbox000).", + ) + spoken = _read_text(text) + api_key = config.resolve_api_key(profile=state.profile) + bare_voice, overrides = dialogue.parse_voice_overrides(voice) + if dialogue.looks_like_speaker_labeled(spoken): + _speak_dialogue( + api_key, + spoken, + bare_voice, + overrides, + language, + sample_rate, + out, + json_mode=json_mode, + quiet=state.quiet, + ) + else: + _speak_single( + api_key, + spoken, + bare_voice or DEFAULT_VOICE, + language, + sample_rate, + out, + json_mode=json_mode, + quiet=state.quiet, + ) + + run_command(ctx, body, json=json_out) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/test_speak.py -q` +Expected: PASS (existing + new). If `test_voice_and_language_flow_into_config` fails on `cfg.voice`, confirm it passes a bare `--voice jane` with prose text — it should still set `cfg.voice == "jane"` via the single path. + +- [ ] **Step 5: Commit** + +```bash +git add aai_cli/commands/speak.py tests/test_speak.py +git commit -m "feat(speak): auto-detect diarized input and voice each speaker" +``` + +--- + +## Task 6: Regenerate the help snapshot, then run the full gate + live check + +**Files:** + +- Modify: `tests/__snapshots__/test_cli_output_snapshots.ambr` + +- [ ] **Step 1: Regenerate the `aai speak --help` snapshot** + +The `--voice` help text and the epilog examples changed. + +Run: `uv run pytest tests/test_cli_output_snapshots.py --snapshot-update -q` +Then inspect: `git diff tests/__snapshots__/test_cli_output_snapshots.ambr` +Expected: only the `[speak]` help block changes (new `--voice` help string, new examples). No other snapshot should move. + +- [ ] **Step 2: Run the full gate** + +Run: `./scripts/check.sh` +Expected: ends with `All checks passed.` This is the source of truth — it covers ruff, mypy, pyright, vulture, import-linter, xenon (complexity ≤ B; if `speak.py`'s `body`/helpers trip it, the helper split above should keep each function simple — do not re-inline), 90% branch coverage, 100% patch coverage, and the diff-scoped mutation gate. + +If the mutation gate reports survivors on changed lines, add a behavioral assertion that fails when that line breaks (e.g. assert the *exact* silence byte-length, the *exact* resolved voice list, or the specific note substring) rather than loosening the test. + +- [ ] **Step 3: Live sandbox smoke test (manual, optional but recommended)** + +A sandbox key is in the keyring (profile `default`). Verify the real pipe end-to-end, writing to `/tmp` (never the repo root) and wrapping so a blocking path can't wedge the session: + +```bash +printf 'Speaker A: Hello there friend.\nSpeaker B: I am doing great, thanks for asking.\n' \ + | uv run aai --sandbox speak --out /tmp/dialogue.wav --json +file /tmp/dialogue.wav # expect: RIFF ... WAVE audio, 16 bit, mono 24000 Hz +``` + +Expected JSON: `{"mode":"multi","speakers":{"A":"jane","B":"michael"},"segments":2,...}` and a valid multi-second WAV. (If the sandbox blocks the host or a chosen voice returns `RetryError`, that surfaces as a clean error — note it, don't treat it as a code defect.) + +- [ ] **Step 4: Commit the snapshot** + +```bash +git add tests/__snapshots__/test_cli_output_snapshots.ambr +git commit -m "test(speak): regenerate help snapshot for repeatable --voice" +``` + +--- + +## Self-Review Notes + +- **Spec coverage:** detection (Task 1), parsing/wrap-folding/merge (Task 1), rotation + overrides + bare-voice rule (Tasks 2 & 5), silence gap (Tasks 3 & 4), per-segment synth/concat (Task 4), command dispatch + JSON shapes + stderr note (Task 5), help snapshot (Task 6). Error table: malformed `--voice` (Task 2), empty-after-strip (Task 5 `_speak_dialogue`), bare-voice note (Task 5). +- **Type consistency:** `Segment(speaker_id, text)`, `assign_voices -> (list[tuple[str,str]], dict[str,str])`, `parse_voice_overrides -> (str|None, dict[str,str])`, `synthesize_dialogue(api_key, segments, *, language, sample_rate, connect, on_warning)` are used identically across tasks and the command. +- **`--voice` default `[]`:** lets `_speak_single` fall back to `DEFAULT_VOICE` and lets the multi-mode note fire only on an explicit bare voice. diff --git a/docs/superpowers/plans/2026-06-10-aai-speak.md b/docs/superpowers/plans/2026-06-10-aai-speak.md new file mode 100644 index 00000000..7c8c1b9a --- /dev/null +++ b/docs/superpowers/plans/2026-06-10-aai-speak.md @@ -0,0 +1,1090 @@ +# `aai speak` Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add an `aai speak` command that synthesizes speech from text via the sandbox streaming-TTS WebSocket, playing it through the speakers by default or writing a WAV with `--out`. + +**Architecture:** A new `aai_cli/tts/` subsystem mirrors `aai_cli/agent/`: `session.py` drives the WebSocket protocol (Begin → Generate → Flush → Audio → Terminate) with an injectable `connect` factory for hermetic tests, and `audio.py` writes WAVs / plays PCM. A thin `commands/speak.py` Typer sub-app enforces sandbox-only availability, resolves the API key, reads text from arg or stdin, and routes the result to file or speaker. A new `streaming_tts_host` field on `Environment` is the availability signal (empty on production). + +**Tech Stack:** Python 3.12+, Typer, `websockets` (sync client), `sounddevice`, stdlib `wave`. Tests: pytest + Typer `CliRunner` + syrupy snapshots. All commands run via `uv run`. + +--- + +## File Structure + +- **Create** `aai_cli/tts/__init__.py` — empty package marker. +- **Create** `aai_cli/tts/session.py` — protocol client: `SpeakConfig`, `SpeakResult`, `is_available()`, `ws_url()`, `synthesize()`, plus connect/auth-mapping helpers (mirrors `agent/session.py`). +- **Create** `aai_cli/tts/audio.py` — `write_wav()` and `play_pcm()`. +- **Create** `aai_cli/commands/speak.py` — the `speak` Typer sub-app. +- **Modify** `aai_cli/environments.py` — add `streaming_tts_host: str` to `Environment` and both env literals. +- **Modify** `aai_cli/main.py` — import `speak`, register the sub-app, slot `"speak"` into `_COMMAND_ORDER`. +- **Create** `tests/test_tts_session.py`, `tests/test_tts_audio.py`, `tests/test_speak.py`, and extend `tests/test_environments.py`. +- **Update** `tests/__snapshots__/test_cli_output_snapshots.ambr` via `--snapshot-update`. + +Conventions for every new module: start with `from __future__ import annotations`; modern typing (`X | None`); errors → stderr, data → stdout. The post-edit hook runs `ruff --fix` + `ruff format` on save, so land each import with its first use in the same edit. + +--- + +## Task 1: Add `streaming_tts_host` to `Environment` + +**Files:** +- Modify: `aai_cli/environments.py` +- Test: `tests/test_environments.py` + +- [ ] **Step 1: Write the failing test** + +Add to `tests/test_environments.py`: + +```python +def test_sandbox_has_streaming_tts_host(): + assert ( + environments.get("sandbox000").streaming_tts_host + == "streaming-tts.sandbox000.assemblyai-labs.com" + ) + + +def test_production_has_no_streaming_tts_host(): + # Empty host is the "TTS not available here" signal the speak command checks. + assert environments.get("production").streaming_tts_host == "" +``` + +If `tests/test_environments.py` does not import `environments`, add `from aai_cli import environments` at the top (check first; many test files already have it). + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/test_environments.py -q -k streaming_tts` +Expected: FAIL — `TypeError: __init__() missing ... 'streaming_tts_host'` or `AttributeError`. + +- [ ] **Step 3: Add the field and populate both environments** + +In `aai_cli/environments.py`, add the field to the dataclass (place it right after `streaming_host` so related hosts stay together): + +```python + streaming_host: str # StreamingClientOptions.api_host (SDK builds wss://host/v3/ws) + streaming_tts_host: str # streaming TTS host; empty when TTS isn't available (prod) + agents_host: str # Voice Agent host; the agent client builds wss://host/v1/ws +``` + +In the `production` literal add (after its `streaming_host=` line): + +```python + streaming_host="streaming.assemblyai.com", + streaming_tts_host="", +``` + +In the `sandbox000` literal add (after its `streaming_host=` line): + +```python + streaming_host="streaming.sandbox000.assemblyai-labs.com", + streaming_tts_host="streaming-tts.sandbox000.assemblyai-labs.com", +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `uv run pytest tests/test_environments.py -q` +Expected: PASS (all environment tests, including the two new ones). + +- [ ] **Step 5: Commit** + +```bash +git add aai_cli/environments.py tests/test_environments.py +git commit -m "feat(speak): add streaming_tts_host to Environment + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 2: TTS session — availability + URL building + +**Files:** +- Create: `aai_cli/tts/__init__.py` +- Create: `aai_cli/tts/session.py` +- Test: `tests/test_tts_session.py` + +- [ ] **Step 1: Write the failing test** + +Create `tests/test_tts_session.py`: + +```python +from __future__ import annotations + +from aai_cli import environments +from aai_cli.tts import session + + +def _use_env(name: str) -> None: + environments.set_active(environments.get(name)) + + +def test_is_available_true_in_sandbox(): + _use_env("sandbox000") + assert session.is_available() is True + + +def test_is_available_false_in_production(): + _use_env("production") + assert session.is_available() is False + + +def test_ws_url_includes_set_params_only(): + _use_env("sandbox000") + cfg = session.SpeakConfig(text="hi", voice="jane", language="English") + url = session.ws_url(cfg.query_params()) + assert url.startswith("wss://streaming-tts.sandbox000.assemblyai-labs.com/v1/ws/?") + assert "voice=jane" in url + assert "language=English" in url + assert "sample_rate" not in url + + +def test_ws_url_no_params_has_no_query_string(): + _use_env("sandbox000") + url = session.ws_url(session.SpeakConfig(text="hi").query_params()) + assert url == "wss://streaming-tts.sandbox000.assemblyai-labs.com/v1/ws/" + + +def test_query_params_serializes_sample_rate_as_string(): + cfg = session.SpeakConfig(text="hi", sample_rate=16000) + assert cfg.query_params() == {"sample_rate": "16000"} +``` + +Note: `environments.set_active` mutates a process global; the suite's ordering is randomized, so always set the env inside each test (as above) rather than relying on prior state. + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/test_tts_session.py -q` +Expected: FAIL — `ModuleNotFoundError: No module named 'aai_cli.tts'`. + +- [ ] **Step 3: Create the package and the URL/availability layer** + +Create `aai_cli/tts/__init__.py` (empty file). + +Create `aai_cli/tts/session.py` with this top portion: + +```python +from __future__ import annotations + +import base64 +import contextlib +import json +import logging +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any +from urllib.parse import urlencode + +from aai_cli import environments +from aai_cli.errors import APIError, CLIError, auth_failure, is_auth_failure + +# The websocket connection and the `connect` factory come from a library with no +# usable type stubs; alias that untyped boundary so each role is named and Any +# stays in one place. (Server frames remain dict[str, Any].) +_WebSocket = Any +_Connect = Any + +# A pre-upgrade HTTP 403 on the WebSocket handshake is NOT a rejected key (it also +# covers WAF/region/plan blocks) — mirrors how agent/stream classify handshakes. +_HTTP_FORBIDDEN = 403 + +# websockets' sync client logs teardown errors through these; silence them so an +# internal "stream ended" traceback never lands on stderr next to our clean error. +_WEBSOCKETS_LOGGERS = ("websockets", "websockets.client") + + +@dataclass(frozen=True) +class SpeakConfig: + """Per-run TTS parameters. Optional fields are only sent when set, so the + server applies its own defaults (voice/language/sample_rate) otherwise.""" + + text: str + voice: str | None = None + language: str | None = None + sample_rate: int | None = None + + def query_params(self) -> dict[str, str]: + params: dict[str, str] = {} + if self.voice is not None: + params["voice"] = self.voice + if self.language is not None: + params["language"] = self.language + if self.sample_rate is not None: + params["sample_rate"] = str(self.sample_rate) + return params + + +@dataclass(frozen=True) +class SpeakResult: + """The synthesized audio: raw 16-bit mono PCM plus its sample rate.""" + + pcm: bytes + sample_rate: int + audio_duration_seconds: float + + +def is_available() -> bool: + """True only where the active environment has a streaming-TTS host (sandbox).""" + return bool(environments.active().streaming_tts_host) + + +def ws_url(params: dict[str, str]) -> str: + """The streaming-TTS socket URL for the active environment, with query params.""" + base = f"wss://{environments.active().streaming_tts_host}/v1/ws/" + return f"{base}?{urlencode(params)}" if params else base +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `uv run pytest tests/test_tts_session.py -q` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add aai_cli/tts/__init__.py aai_cli/tts/session.py tests/test_tts_session.py +git commit -m "feat(speak): tts session availability + url building + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 3: TTS session — the synthesize protocol + +**Files:** +- Modify: `aai_cli/tts/session.py` +- Test: `tests/test_tts_session.py` + +- [ ] **Step 1: Write the failing tests** + +Add to `tests/test_tts_session.py` (top imports become `import base64`, `import json`, `import pytest`, and `from aai_cli.errors import APIError, NotAuthenticated`): + +```python +import base64 +import json + +import pytest + +from aai_cli.errors import APIError, NotAuthenticated + + +class FakeWS: + """A minimal stand-in for a websockets sync connection.""" + + def __init__(self, incoming: list[str]) -> None: + self._incoming = list(incoming) + self.sent: list[dict] = [] + self.closed = False + + def recv(self) -> str: + return self._incoming.pop(0) + + def send(self, data: str) -> None: + self.sent.append(json.loads(data)) + + def close(self) -> None: + self.closed = True + + +def _audio_frame(pcm: bytes, *, sample_rate: int = 24000, final: bool) -> str: + return json.dumps( + { + "type": "Audio", + "audio": base64.b64encode(pcm).decode("ascii"), + "sample_rate": sample_rate, + "encoding": "pcm_s16le", + "is_final_for_flush": final, + } + ) + + +def _connect_returning(ws: FakeWS, captured: dict) -> "callable": + def _connect(url: str, **kwargs): + captured["url"] = url + captured["kwargs"] = kwargs + return ws + return _connect + + +def test_synthesize_drives_the_full_protocol(): + captured: dict = {} + ws = FakeWS( + [ + json.dumps({"type": "Begin", "id": "s1", "expires_at": 1}), + _audio_frame(b"\x01\x02\x03\x04", final=False), + _audio_frame(b"\x05\x06", final=True), + ] + ) + cfg = session.SpeakConfig(text="hello", voice="jane") + result = session.synthesize("k", cfg, connect=_connect_returning(ws, captured)) + + # Sends Generate(text), then Flush, then Terminate — in that order. + assert [m["type"] for m in ws.sent] == ["Generate", "Flush", "Terminate"] + assert ws.sent[0]["text"] == "hello" + # Accumulates decoded PCM across chunks and stops on is_final_for_flush. + assert result.pcm == b"\x01\x02\x03\x04\x05\x06" + assert result.sample_rate == 24000 + # 6 bytes / 2 bytes-per-sample / 24000 Hz. + assert result.audio_duration_seconds == pytest.approx(6 / 2 / 24000) + # Auth header carries the key as a Bearer token. + assert captured["kwargs"]["additional_headers"]["Authorization"] == "Bearer k" + assert ws.closed is True + + +def test_synthesize_raises_on_missing_begin(): + ws = FakeWS([json.dumps({"type": "Audio"})]) + with pytest.raises(APIError): + session.synthesize("k", session.SpeakConfig(text="hi"), connect=lambda *a, **k: ws) + + +def test_synthesize_maps_error_frame_to_api_error(): + ws = FakeWS( + [ + json.dumps({"type": "Begin", "id": "s", "expires_at": 1}), + json.dumps({"type": "Error", "error_code": 500, "error": "boom"}), + ] + ) + with pytest.raises(APIError, match="boom"): + session.synthesize("k", session.SpeakConfig(text="hi"), connect=lambda *a, **k: ws) + + +def test_synthesize_invokes_on_warning_then_continues(): + seen: list[str] = [] + ws = FakeWS( + [ + json.dumps({"type": "Begin", "id": "s", "expires_at": 1}), + json.dumps({"type": "Warning", "warning_code": 1, "warning": "slow"}), + _audio_frame(b"\x01\x02", final=True), + ] + ) + result = session.synthesize( + "k", + session.SpeakConfig(text="hi"), + connect=lambda *a, **k: ws, + on_warning=seen.append, + ) + assert seen == ["slow"] + assert result.pcm == b"\x01\x02" + + +def test_synthesize_ignores_warning_without_callback(): + ws = FakeWS( + [ + json.dumps({"type": "Begin", "id": "s", "expires_at": 1}), + json.dumps({"type": "Warning", "warning_code": 1, "warning": "slow"}), + _audio_frame(b"\x01\x02", final=True), + ] + ) + # No on_warning: the warning is silently skipped, not an error. + result = session.synthesize("k", session.SpeakConfig(text="hi"), connect=lambda *a, **k: ws) + assert result.pcm == b"\x01\x02" + + +def test_synthesize_maps_rejected_key_on_connect_to_not_authenticated(): + class Rejected(Exception): + pass + + def _connect(*_a, **_k): + raise Rejected("Unauthorized") + + with pytest.raises(NotAuthenticated): + session.synthesize("k", session.SpeakConfig(text="hi"), connect=_connect) + + +def test_synthesize_maps_forbidden_connect_to_api_error(): + class Resp: + status_code = 403 + + class Forbidden(Exception): + response = Resp() + + def _connect(*_a, **_k): + raise Forbidden("Unauthorized") # 403 -> NOT a rejected key + + with pytest.raises(APIError): + session.synthesize("k", session.SpeakConfig(text="hi"), connect=_connect) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/test_tts_session.py -q -k synthesize` +Expected: FAIL — `AttributeError: module 'aai_cli.tts.session' has no attribute 'synthesize'`. + +- [ ] **Step 3: Implement the protocol** + +Append to `aai_cli/tts/session.py`: + +```python +def _silence_websockets_logging() -> None: + """Keep websockets' internal teardown logging off the user's stderr.""" + for name in _WEBSOCKETS_LOGGERS: + logging.getLogger(name).setLevel(logging.CRITICAL) + + +def _is_rejected_key(exc: Exception) -> bool: + """Auth-shaped failure (the key was rejected) vs a generic block. A plain HTTP + 403 on the upgrade is NOT a rejected key (WAF/region/plan also 403).""" + status = getattr(getattr(exc, "response", None), "status_code", None) + if status == _HTTP_FORBIDDEN: + return False + return is_auth_failure(exc) + + +def _auth_or_api_error(exc: Exception, message: str) -> CLIError: + """A rejected key becomes auth_failure(); anything else becomes an APIError.""" + if _is_rejected_key(exc): + return auth_failure() + return APIError(f"{message}: {exc}") + + +def _open_ws(connect: _Connect, api_key: str, url: str) -> _WebSocket: + """Open the TTS socket, mapping a connect failure to a clean CLIError.""" + try: + return connect( + url, + additional_headers={"Authorization": f"Bearer {api_key}"}, + max_size=None, + ) + except Exception as exc: + raise _auth_or_api_error(exc, "Could not connect to the TTS service") from exc + + +def _run_protocol( + ws: _WebSocket, config: SpeakConfig, on_warning: Callable[[str], None] | None +) -> SpeakResult: + """Send Generate+Flush, collect Audio until is_final_for_flush, then Terminate.""" + begin = json.loads(ws.recv()) + if begin.get("type") != "Begin": + raise APIError(f"TTS service did not start the session (got {begin.get('type')!r}).") + + ws.send(json.dumps({"type": "Generate", "text": config.text})) + ws.send(json.dumps({"type": "Flush"})) + + pcm = bytearray() + sample_rate = 0 + while True: + msg: dict[str, Any] = json.loads(ws.recv()) + mtype = msg.get("type") + if mtype == "Audio": + pcm.extend(base64.b64decode(msg["audio"])) + sample_rate = int(msg["sample_rate"]) + if msg.get("is_final_for_flush"): + break + elif mtype == "Error": + raise APIError( + f"TTS error ({msg.get('error_code', '')}): {msg.get('error') or 'unknown'}" + ) + elif mtype == "Warning" and on_warning is not None: + on_warning(str(msg.get("warning", ""))) + + with contextlib.suppress(Exception): + ws.send(json.dumps({"type": "Terminate"})) + + duration = len(pcm) / 2 / sample_rate + return SpeakResult(bytes(pcm), sample_rate, duration) + + +def synthesize( + api_key: str, + config: SpeakConfig, + *, + connect: _Connect = None, + on_warning: Callable[[str], None] | None = None, +) -> SpeakResult: + """Open the streaming-TTS socket and synthesize ``config.text`` to PCM. + + ``connect`` defaults to websockets' synchronous client; injectable for tests. + Connect/session failures map to a clean CLIError (a rejected key -> exit 4). + """ + _silence_websockets_logging() + if connect is None: + from websockets.sync.client import connect + + ws = _open_ws(connect, api_key, ws_url(config.query_params())) + try: + return _run_protocol(ws, config, on_warning) + except (CLIError, KeyboardInterrupt, BrokenPipeError): + raise # clean CLI errors, Ctrl-C, and a closed pipe are handled upstream + except Exception as exc: + raise _auth_or_api_error(exc, "TTS session failed") from exc + finally: + with contextlib.suppress(Exception): + ws.close() +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/test_tts_session.py -q` +Expected: PASS (all session tests). + +- [ ] **Step 5: Commit** + +```bash +git add aai_cli/tts/session.py tests/test_tts_session.py +git commit -m "feat(speak): tts websocket synthesize protocol + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 4: TTS audio — WAV writing and playback + +**Files:** +- Create: `aai_cli/tts/audio.py` +- Test: `tests/test_tts_audio.py` + +- [ ] **Step 1: Write the failing tests** + +Create `tests/test_tts_audio.py`: + +```python +from __future__ import annotations + +import wave +from pathlib import Path + +import pytest + +from aai_cli.errors import CLIError +from aai_cli.tts import audio + + +def test_write_wav_produces_mono_16bit_wav(tmp_path: Path): + out = tmp_path / "nested" / "out.wav" # parent dirs created on demand + pcm = b"\x01\x02\x03\x04" + audio.write_wav(out, pcm, 24000) + + with wave.open(str(out), "rb") as wav: + assert wav.getnchannels() == 1 + assert wav.getsampwidth() == 2 + assert wav.getframerate() == 24000 + assert wav.readframes(wav.getnframes()) == pcm + + +class FakeStream: + def __init__(self) -> None: + self.events: list[str] = [] + self.written: bytes = b"" + + def start(self) -> None: + self.events.append("start") + + def write(self, data: bytes) -> None: + self.written += bytes(data) + self.events.append("write") + + def stop(self) -> None: + self.events.append("stop") + + def close(self) -> None: + self.events.append("close") + + +def test_play_pcm_writes_to_started_stream_then_closes(): + stream = FakeStream() + audio.play_pcm(b"\x01\x02", 16000, stream_factory=lambda rate: stream) + assert stream.events == ["start", "write", "stop", "close"] + assert stream.written == b"\x01\x02" + + +def test_play_pcm_wraps_device_failure_in_cli_error(): + def _boom(_rate: int): + raise RuntimeError("no device") + + with pytest.raises(CLIError, match="Could not play audio"): + audio.play_pcm(b"\x01\x02", 16000, stream_factory=_boom) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/test_tts_audio.py -q` +Expected: FAIL — `ModuleNotFoundError: No module named 'aai_cli.tts.audio'`. + +- [ ] **Step 3: Implement `audio.py`** + +Create `aai_cli/tts/audio.py`: + +```python +from __future__ import annotations + +import wave +from collections.abc import Callable +from pathlib import Path +from typing import Any + +from aai_cli.errors import CLIError +from aai_cli.microphone import audio_missing_error + + +def write_wav(path: Path, pcm: bytes, sample_rate: int) -> None: + """Write 16-bit mono PCM to a WAV file, creating parent dirs as needed.""" + path.parent.mkdir(parents=True, exist_ok=True) + with wave.open(str(path), "wb") as wav: + wav.setnchannels(1) + wav.setsampwidth(2) + wav.setframerate(sample_rate) + wav.writeframes(pcm) + + +def _default_output_stream(sample_rate: int) -> Any: + """A started-on-demand raw 16-bit mono output stream from sounddevice.""" + try: + import sounddevice as sd + except ImportError as exc: + raise audio_missing_error() from exc + return sd.RawOutputStream(samplerate=sample_rate, channels=1, dtype="int16") + + +def play_pcm( + pcm: bytes, + sample_rate: int, + *, + stream_factory: Callable[[int], Any] | None = None, +) -> None: + """Play 16-bit mono PCM through the default output device (blocks until done). + + ``stream_factory`` is injectable for tests; a device failure is wrapped in a + clean CLIError that points at --out as the headless escape hatch. + """ + factory = stream_factory or _default_output_stream + try: + stream = factory(sample_rate) + stream.start() + stream.write(pcm) + stream.stop() + stream.close() + except CLIError: + raise # audio_missing_error() is already user-facing + except Exception as exc: + raise CLIError( + f"Could not play audio: {exc}", + error_type="audio_output_error", + exit_code=1, + suggestion="Check your output device and run 'aai doctor', or use --out to save a WAV.", + ) from exc +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/test_tts_audio.py -q` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add aai_cli/tts/audio.py tests/test_tts_audio.py +git commit -m "feat(speak): tts wav writing + pcm playback + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 5: The `speak` command + +**Files:** +- Create: `aai_cli/commands/speak.py` +- Modify: `aai_cli/main.py` +- Test: `tests/test_speak.py` + +- [ ] **Step 1: Write the failing tests** + +Create `tests/test_speak.py`: + +```python +from __future__ import annotations + +import json + +import pytest +from typer.testing import CliRunner + +from aai_cli import config +from aai_cli.main import app +from aai_cli.tts import session + +runner = CliRunner() + + +@pytest.fixture(autouse=True) +def _fake_key(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(config, "resolve_api_key", lambda **_: "test-key") + + +@pytest.fixture +def _fake_synthesize(monkeypatch: pytest.MonkeyPatch): + calls: dict = {} + + def _fake(api_key, cfg, *, connect=None, on_warning=None): + calls["api_key"] = api_key + calls["cfg"] = cfg + return session.SpeakResult(pcm=b"\x01\x02\x03\x04", sample_rate=24000, audio_duration_seconds=0.5) + + monkeypatch.setattr(session, "synthesize", _fake) + return calls + + +def test_production_env_is_rejected_with_sandbox_hint(): + result = runner.invoke(app, ["speak", "Hello"]) # default = production + assert result.exit_code == 2 + assert "only available in the sandbox" in result.output + assert "--sandbox" in result.output + + +def test_plays_audio_by_default(monkeypatch, _fake_synthesize): + played: dict = {} + monkeypatch.setattr( + "aai_cli.commands.speak.audio.play_pcm", + lambda pcm, rate, **_: played.update(pcm=pcm, rate=rate), + ) + result = runner.invoke(app, ["--sandbox", "speak", "Hello there"]) + assert result.exit_code == 0 + assert played == {"pcm": b"\x01\x02\x03\x04", "rate": 24000} + assert _fake_synthesize["cfg"].text == "Hello there" + + +def test_out_writes_wav_and_does_not_play(monkeypatch, tmp_path, _fake_synthesize): + monkeypatch.setattr( + "aai_cli.commands.speak.audio.play_pcm", + lambda *a, **k: pytest.fail("should not play when --out is given"), + ) + written: dict = {} + monkeypatch.setattr( + "aai_cli.commands.speak.audio.write_wav", + lambda path, pcm, rate: written.update(path=path, pcm=pcm, rate=rate), + ) + out = tmp_path / "x.wav" + result = runner.invoke(app, ["--sandbox", "speak", "Hi", "--out", str(out)]) + assert result.exit_code == 0 + assert written["pcm"] == b"\x01\x02\x03\x04" + assert str(written["path"]) == str(out) + + +def test_reads_text_from_stdin_when_arg_omitted(monkeypatch, _fake_synthesize): + monkeypatch.setattr("aai_cli.commands.speak.audio.play_pcm", lambda *a, **k: None) + result = runner.invoke(app, ["--sandbox", "speak"], input="piped text\n") + assert result.exit_code == 0 + assert _fake_synthesize["cfg"].text == "piped text" + + +def test_empty_text_is_a_usage_error(monkeypatch): + # No arg and empty stdin -> usage error, before any synthesis. + result = runner.invoke(app, ["--sandbox", "speak"], input="") + assert result.exit_code == 2 + assert "No text to speak" in result.output + + +def test_voice_and_language_flow_into_config(monkeypatch, _fake_synthesize): + monkeypatch.setattr("aai_cli.commands.speak.audio.play_pcm", lambda *a, **k: None) + result = runner.invoke( + app, ["--sandbox", "speak", "Hi", "--voice", "jane", "--language", "English"] + ) + assert result.exit_code == 0 + cfg = _fake_synthesize["cfg"] + assert cfg.voice == "jane" + assert cfg.language == "English" + assert cfg.query_params() == {"voice": "jane", "language": "English"} + + +def test_json_mode_emits_metadata_object_on_stdout(monkeypatch, _fake_synthesize): + monkeypatch.setattr("aai_cli.commands.speak.audio.play_pcm", lambda *a, **k: None) + result = runner.invoke(app, ["--sandbox", "speak", "Hi", "--voice", "jane", "--json"]) + assert result.exit_code == 0 + # The behavioral split: --json yields a parseable object, not human prose. + payload = json.loads(result.stdout.strip()) + assert payload["voice"] == "jane" + assert payload["sample_rate"] == 24000 + assert payload["bytes"] == 4 + + +def test_human_mode_keeps_stdout_clean(monkeypatch, _fake_synthesize): + monkeypatch.setattr("aai_cli.commands.speak.audio.play_pcm", lambda *a, **k: None) + result = runner.invoke(app, ["--sandbox", "speak", "Hi"]) + assert result.exit_code == 0 + # Human summary goes to stderr; stdout stays empty (audio went to the speaker). + assert result.stdout.strip() == "" +``` + +Note: `CliRunner` by default merges stderr into `result.output` but keeps `result.stdout` separate only when constructed with `mix_stderr=False`. This repo's other snapshot/JSON tests read `result.stdout` successfully, so the installed Typer/Click splits them; if `result.stdout` raises or is empty unexpectedly, construct `CliRunner(mix_stderr=False)` at the top of this file (check how `tests/test_speak.py`'s sibling JSON tests do it — e.g. grep `mix_stderr` across `tests/`). + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/test_speak.py -q` +Expected: FAIL — the `speak` command does not exist (`Error: No such command 'speak'`). + +- [ ] **Step 3: Implement the command** + +Create `aai_cli/commands/speak.py`: + +```python +from __future__ import annotations + +import sys +from pathlib import Path + +import typer + +from aai_cli import config, 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 +from aai_cli.tts import audio, session + +app = typer.Typer() + + +def _read_text(text: str | None) -> str: + """The text to speak: the argument if non-blank, else stdin when piped.""" + if text is not None and text.strip(): + return text + if text is None and not sys.stdin.isatty(): + piped = sys.stdin.read().strip() + if piped: + return piped + raise UsageError( + "No text to speak.", + suggestion='Pass text as an argument: aai speak "Hello" — or pipe it via stdin.', + ) + + +def _emit_result( + result: session.SpeakResult, + config_: session.SpeakConfig, + out: Path | None, + *, + json_mode: bool, +) -> None: + """Report what was produced: a JSON object on stdout, or a human note on stderr.""" + duration = round(result.audio_duration_seconds, 3) + if json_mode: + output.emit_ndjson( + { + "voice": config_.voice, + "language": config_.language, + "sample_rate": result.sample_rate, + "audio_duration_seconds": duration, + "bytes": len(result.pcm), + "out": str(out) if out is not None else None, + } + ) + return + where = f"saved to {out}" if out is not None else "played" + output.error_console.print(f"[aai.muted]Spoke {duration}s of audio ({where}).[/aai.muted]") + + +@app.command( + rich_help_panel=help_panels.TRANSCRIPTION, + epilog=examples_epilog( + [ + ("Speak text aloud (sandbox only)", 'aai speak "Hello there, friend." --sandbox'), + ("Pick a voice and language", 'aai speak "Bonjour" --voice jane --language French --sandbox'), + ("Save to a WAV instead of playing", 'aai speak "Hello" --out /tmp/hello.wav --sandbox'), + ("Speak text piped from stdin", 'echo "Hello" | aai speak --sandbox'), + ] + ), +) +def speak( + ctx: typer.Context, + text: str | None = typer.Argument(None, help="Text to speak. Omit to read from stdin."), + voice: str | None = typer.Option( + None, "--voice", help="Voice id. Server default if omitted." + ), + language: str | None = typer.Option( + None, "--language", help="Language. Server default if omitted." + ), + sample_rate: int | None = typer.Option( + None, "--sample-rate", help="Output sample rate in Hz. Server default if omitted." + ), + out: Path | None = typer.Option( + None, "--out", help="Write a WAV file instead of playing through the speakers." + ), + json_out: bool = options.json_option("Emit JSON metadata about the synthesized audio."), +) -> None: + """Synthesize speech from text with AssemblyAI streaming TTS (sandbox only). + + Plays the audio through your speakers by default, or writes a WAV with --out. + This feature only exists in the sandbox today — run it with --sandbox. + """ + + def body(state: AppState, json_mode: bool) -> None: + if not session.is_available(): + raise CLIError( + "aai speak is only available in the sandbox.", + error_type="unsupported_environment", + exit_code=2, + suggestion="Re-run with --sandbox (or --env sandbox000).", + ) + spoken = _read_text(text) + api_key = config.resolve_api_key(profile=state.profile) + cfg = session.SpeakConfig( + text=spoken, voice=voice, language=language, sample_rate=sample_rate + ) + with output.status("Synthesizing speech…", json_mode=json_mode, quiet=state.quiet): + result = session.synthesize( + api_key, cfg, on_warning=lambda m: output.emit_warning(m, json_mode=json_mode) + ) + if out is not None: + audio.write_wav(out, result.pcm, result.sample_rate) + else: + audio.play_pcm(result.pcm, result.sample_rate) + _emit_result(result, cfg, out, json_mode=json_mode) + + run_command(ctx, body, json=json_out) +``` + +- [ ] **Step 4: Register the command in `main.py`** + +In the `from aai_cli.commands import (...)` block, add `speak` in alphabetical order (between `share` and `stream`): + +```python + share, + speak, + stream, +``` + +After the `app.add_typer(agent.app)` line, add: + +```python +app.add_typer(speak.app) +``` + +In `_COMMAND_ORDER`, add `"speak"` in the "Run AssemblyAI" group, right after `"agent"`: + +```python + "transcribe", + "stream", + "agent", + "speak", + "llm", +``` + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `uv run pytest tests/test_speak.py -q` +Expected: PASS. If `result.stdout` access errors, apply the `mix_stderr=False` note from Step 1. + +- [ ] **Step 6: Commit** + +```bash +git add aai_cli/commands/speak.py aai_cli/main.py tests/test_speak.py +git commit -m "feat(speak): add sandbox-only speak command + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 6: Regenerate help snapshots + +**Files:** +- Modify: `tests/__snapshots__/test_cli_output_snapshots.ambr` + +The snapshot suite derives its command list from the live Typer tree, so the new `speak` leaf now requires a `--help` golden, and the root `aai --help` golden changed (new command in the ordering). `test_help_examples_coverage.py` also requires the examples epilog the command already declares. + +- [ ] **Step 1: Confirm the snapshots currently fail** + +Run: `uv run pytest tests/test_cli_output_snapshots.py tests/test_help_examples_coverage.py -q` +Expected: FAIL — a missing/changed snapshot for `speak` and the root help. + +- [ ] **Step 2: Regenerate the snapshots** + +Run: `uv run pytest tests/test_cli_output_snapshots.py --snapshot-update -q` + +Do NOT hand-edit the `.ambr` file (syrupy indentation must stay byte-for-byte). + +- [ ] **Step 3: Verify everything passes** + +Run: `uv run pytest tests/test_cli_output_snapshots.py tests/test_help_examples_coverage.py tests/test_help_rendering.py -q` +Expected: PASS. + +- [ ] **Step 4: Eyeball the diff** + +Run: `git diff tests/__snapshots__/test_cli_output_snapshots.ambr` +Expected: only a new `speak` help block and the root help gaining a `speak` line under "Run AssemblyAI" — no unrelated churn. + +- [ ] **Step 5: Commit** + +```bash +git add tests/__snapshots__/test_cli_output_snapshots.ambr +git commit -m "test(speak): refresh CLI help snapshots + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 7: Documentation + +**Files:** +- Modify: `AGENTS.md` + +- [ ] **Step 1: Document the command and its subsystem** + +In `AGENTS.md`, under the command-layer list, add `speak` to the parenthesized command roster, and under "Feature subsystems" add a bullet: + +```markdown +- **`tts/`** + `commands/speak.py` — `aai speak` synthesizes text to speech over the + sandbox streaming-TTS WebSocket (`streaming-tts.sandbox000.…`). Sandbox-only: + `session.is_available()` is false in production (empty `Environment.streaming_tts_host`), + so the command exits 2 with a `--sandbox` hint. `session.synthesize` drives a + Begin→Generate→Flush→Audio→Terminate protocol with an injectable `connect` for tests; + `audio.py` plays PCM (default) or writes a WAV (`--out`). +``` + +- [ ] **Step 2: Commit** + +```bash +git add AGENTS.md +git commit -m "docs(speak): document the tts subsystem and speak command + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 8: Full gate + +- [ ] **Step 1: Run the authoritative gate** + +Run: `./scripts/check.sh` +Expected: ends with `All checks passed.` + +This runs ruff, mypy, pyright, vulture, deptry, import-linter, xenon, the 90% branch-coverage pytest gate, 100% patch coverage vs `origin/main`, and the diff-scoped mutation gate. Pay attention to: + +- **Patch coverage (100%):** every new line must be executed by a test. The session/audio/command tests above are written to cover each branch (including the `on_warning is None` path and the 403-vs-rejected-key split). +- **Mutation gate:** changed lines need assertions that *fail* if the line breaks. The tests assert behavior (exact PCM bytes, exact sent-frame order, the JSON-vs-stderr split, duration math), not just execution. If a boolean/default survives, add an assertion on the behavioral difference rather than `# pragma: no mutate` (reserve that for genuinely unassertable lines). +- **vulture (dead code):** every new function is wired to a caller; if vulture flags one, it means a call site is missing. +- **"no new escape hatches" diff gate:** avoid net-new `# type: ignore` / `# noqa` / `cast(` / bare `Any` beyond the documented `_WebSocket`/`_Connect` aliases (which mirror `agent/session.py`). + +- [ ] **Step 2: Fix any failures and re-run until green** + +Iterate on `./scripts/check.sh` until it prints `All checks passed.` Do not claim completion before that line appears. + +- [ ] **Step 3: Final commit (if the gate required tweaks)** + +```bash +git add -A +git commit -m "chore(speak): satisfy lint/coverage/mutation gate + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Self-Review + +**Spec coverage** — every spec section maps to a task: +- Sandbox-only enforcement → Task 1 (`streaming_tts_host=""` on prod) + Task 5 (`is_available()` guard, exit 2, `--sandbox` hint). +- Requires API key, `Authorization: Bearer` → Task 3 (`_open_ws`) + Task 5 (`config.resolve_api_key`). +- Text from arg or stdin → Task 5 (`_read_text`). +- Params only sent when passed → Task 2 (`SpeakConfig.query_params`). +- Protocol (Begin→Generate→Flush→Audio→Terminate, base64, is_final_for_flush, Warning/Error) → Task 3. +- Play by default / `--out` writes WAV → Task 4 + Task 5. +- `--json` metadata shape → Task 5 (`_emit_result`). +- Error handling table → Tasks 3 & 5 (bad Begin/Error frame → APIError; rejected key → exit 4; empty text → exit 2; prod → exit 2). +- Testing incl. snapshot regeneration → Tasks 2–6. +- Out of scope (no `--show-code`, single Generate, no voice list, no prod wiring) → honored; not implemented. + +**Placeholder scan** — no TBD/TODO; every code step contains complete code. + +**Type consistency** — `SpeakConfig`/`SpeakResult` field names (`text`, `voice`, `language`, `sample_rate`, `pcm`, `audio_duration_seconds`) are identical across `session.py`, the command, and the tests. `synthesize(api_key, config, *, connect=None, on_warning=None)`, `is_available()`, `ws_url(params)`, `write_wav(path, pcm, sample_rate)`, and `play_pcm(pcm, sample_rate, *, stream_factory=None)` signatures match every call site. diff --git a/docs/superpowers/specs/2026-06-10-aai-speak-design.md b/docs/superpowers/specs/2026-06-10-aai-speak-design.md new file mode 100644 index 00000000..c42c7f94 --- /dev/null +++ b/docs/superpowers/specs/2026-06-10-aai-speak-design.md @@ -0,0 +1,144 @@ +# `aai speak` — streaming text-to-speech (sandbox-only) + +**Date:** 2026-06-10 +**Status:** Approved design + +## Summary + +Add an `aai speak` command that synthesizes speech from text via the sandbox +streaming-TTS WebSocket and plays it through the speakers by default, or writes +a WAV file when `--out` is given. The feature only exists in the sandbox +(`streaming-tts.sandbox000.assemblyai-labs.com`); running it against the default +production environment fails with a clean, actionable error. + +```sh +aai speak "Hello there, friend." # play through speakers +aai speak "Hello there." --voice jane --language English +echo "Hello" | aai speak # text from stdin +aai speak "Hi" --out /tmp/out.wav # write a WAV instead of playing +aai speak "Hi" --json # machine-readable metadata +``` + +## Behavior + +- **Sandbox-only.** TTS only exists at `streaming-tts.sandbox000.assemblyai-labs.com`. + Running against production (the shipped default) raises a `CLIError` + instructing the user to pass `--sandbox` (exit 2). +- **Requires an API key**, resolved the normal way (`ASSEMBLYAI_API_KEY` env → + OS keyring). The key is sent as `Authorization: Bearer `, mirroring the + `agent` command. A missing key takes the standard not-authenticated path + (exit 4). No `--api-key` flag (keys must not leak into shell history / `ps`). +- **Text source:** the positional `TEXT` argument, or stdin when the argument is + omitted (`echo … | aai speak`). Empty text from both → usage error (exit 2). +- **Connection params** (`--voice`, `--language`, `--sample-rate`) are added to + the WebSocket query string. `--voice`/`--language` default to the reference + client's defaults (`jane` / `English`) so a bare `aai speak` works out of the + box; `--sample-rate` is only sent when explicitly passed (the server defaults it + to `24000` and echoes the resolved value back in the `Begin` frame). The server + stays the source of truth for the voice catalog — no hardcoded client-side + voice list. (The server's *own* default voice routes to a backend that currently + errors, which is why the CLI sends `jane` explicitly rather than relying on it.) +- **Output:** with no `--out`, play the audio through the speakers (uses + `sounddevice`, the dependency `agent` already relies on). With `--out PATH`, + write a WAV instead. Headless machines have no speakers — documented; use + `--out` there. + +## Protocol + +Reference: `assemblyai/engineering/projects/realtime/api_tts/scripts/sample_session.py` +in the DeepLearning monorepo. A synchronous `websockets` client (same library and +pattern as `aai_cli/agent/session.py`): + +1. Connect to `wss://{tts_host}/v1/ws/?voice=…&language=…&sample_rate=…`. +2. Receive a `Begin` frame + (`{"type":"Begin","id":…,"expires_at":…,"configuration":{"voice":…,"language":…,"sample_rate":int}}`); + anything else is an error. The server echoes the resolved synthesis settings in + `configuration` — `sample_rate` is read from here (the `Audio` frames don't + carry it), falling back to `24000` if absent. +3. Send one `Generate` frame: `{"type":"Generate","text": }`. +4. Send a `ForceFlushTextBuffer` frame: `{"type":"ForceFlushTextBuffer"}` to + trigger synthesis of the buffered text. (A plain `{"type":"Flush"}` is *not* a + valid message — the server rejects it with `InputParseError`.) +5. Receive `Audio` frames: `{"type":"Audio","audio": , + "is_final": bool}`. Base64-decode `audio`, accumulate PCM, and stop when + `is_final` is `true`. + - `Warning` frames → printed to stderr, non-fatal. + - `Error` frames (`{"type":"Error","error_code":int,"error":str}`) → mapped to + a clean `CLIError`. +6. Send `Terminate` (`{"type":"Terminate"}`); the optional `Termination` frame + (`session_duration_seconds`, `total_input_char_length`, + `audio_duration_seconds`) carries final stats. + +Audio is 16-bit mono PCM. The WAV is written with `nchannels=1`, `sampwidth=2`, +`framerate=sample_rate` (the rate from the `Begin` configuration, not assumed). + +## Components + +Mirrors the existing `streaming/` and `agent/` feature subsystems. + +- **`aai_cli/environments.py`** — add a `streaming_tts_host: str` field to the + frozen `Environment` dataclass. `sandbox000` → + `"streaming-tts.sandbox000.assemblyai-labs.com"`; `production` → `""` (the + empty string is the sandbox-only signal the command checks). + +- **`aai_cli/tts/session.py`** — the protocol client. Connects, drives + Begin→Generate→ForceFlushTextBuffer→Audio→Terminate, and returns + `(pcm_bytes, sample_rate, termination_stats)`. The `connect` factory is + injectable (defaults to the `websockets` sync client) so tests run hermetically + with a fake socket. Connect/auth failures map through the existing + `errors.py` helpers (`auth_failure` / `is_auth_failure`); a rejected key → + not-authenticated, other failures → `APIError`. `websockets` internal logging + is silenced for the session (same helper shape as `agent`). + +- **`aai_cli/tts/audio.py`** — `write_wav(path, pcm, sample_rate)` (stdlib + `wave`) and `play_pcm(pcm, sample_rate)` (`sounddevice`). + +- **`aai_cli/commands/speak.py`** — Typer sub-app run through + `context.run_command`. Parses options, enforces sandbox-only, resolves the key, + reads text (arg or stdin), drives the session, then plays or writes. Human mode + shows a status spinner while synthesizing plus a final note; `--json` emits + `{voice, language, sample_rate, audio_duration_seconds, bytes, out}`. + +- **`aai_cli/main.py`** — register the `speak` sub-app and slot it into + `_COMMAND_ORDER` near `agent` / `stream`. + +## Error handling + +| Condition | Result | +| --- | --- | +| Production environment (no sandbox) | `CLIError`, exit 2, suggestion: re-run with `--sandbox` | +| No API key | not-authenticated path, exit 4 | +| Empty text (arg + stdin both empty) | usage `CLIError`, exit 2 | +| Server `Error` frame / rejected key / unexpected first frame | mapped to `APIError` or auth failure — never a traceback | + +## Testing + +Must clear the project's mutation and 100%-patch-coverage tail gates, so tests +assert *behavior*, not just line execution. + +- `tts/session.py`: a fake injected websocket drives the full + Begin→Generate→ForceFlushTextBuffer→Audio→Terminate flow. Assert the exact + frames sent, the base64 decode, stop-on-`is_final`, the `sample_rate` read from + the `Begin` configuration (and its `24000` fallback), `Error`/`Warning` + handling, and auth-failure mapping (rejected key → exit 4). +- `tts/audio.py`: `write_wav` produces a WAV with the correct header + (mono / 16-bit / given rate) and body bytes; `play_pcm` invokes a monkeypatched + `sounddevice`. +- `commands/speak.py` (Typer `CliRunner`): production → clean error; missing key + → exit 4; stdin path; query-param building per flag (only set params appear); + `--out` writes a file; default plays (monkeypatched playback); `--json` shape + asserted on the behavioral split (a parsed JSON object, not human text). +- `environments`: new field present; production host empty. +- Regenerate the syrupy snapshots for `aai speak --help` and the updated + `aai --help` command ordering (`--snapshot-update`); never hand-edit `.ambr`. + +No replay fixture is added: the replay harness records real REST responses, but +TTS is a WebSocket. The injected-`connect` unit tests provide the offline +end-to-end coverage instead. + +## Out of scope (v1) + +- `--show-code` (the protocol is a custom WebSocket client, not an SDK call). +- Multiple `Generate` segments, `--file` input, and `--keepalive`. +- A hardcoded client-side voice list. +- Any production wiring (TTS does not yet exist in production). diff --git a/docs/superpowers/specs/2026-06-10-aai-speak-speaker-voices-design.md b/docs/superpowers/specs/2026-06-10-aai-speak-speaker-voices-design.md new file mode 100644 index 00000000..95d2a937 --- /dev/null +++ b/docs/superpowers/specs/2026-06-10-aai-speak-speaker-voices-design.md @@ -0,0 +1,165 @@ +# `aai speak` — speaker-aware multi-voice playback + +**Date:** 2026-06-10 +**Status:** Approved design + +## Summary + +Extend `aai speak` so that when its input is speaker-labeled transcript text — +the output of `aai transcribe … --speaker-labels` — it strips the `Speaker X:` +labels and synthesizes each speaker in a different voice, concatenated into one +stream. The motivating pipe should "just work": + +```sh +aai transcribe "https://youtu.be/…" --speaker-labels | aai speak --sandbox +``` + +Plain (unlabeled) text keeps today's behavior exactly: one voice, spoken +literally. Detection is automatic — no new flag is required to opt in. + +## Behavior + +### Detection (automatic) + +After reading the text (argument or stdin, unchanged), inspect it: if the +**first non-blank line matches `^Speaker :`** followed by a space (id = one +or more non-space characters), the input is treated as speaker-labeled and enters multi-voice +mode. Otherwise `speak` behaves as it does today (single voice, literal text). +Anchoring on the first labeled line means ordinary prose never misfires into +multi-voice mode. + +### Parsing (folds the 80-column wrap) + +`aai transcribe --speaker-labels` renders one utterance per logical line, but its +Rich console soft-wraps at ~80 columns when piped, so a long utterance becomes +several physical lines and only the first carries the `Speaker X:` prefix. The +parser accounts for this: + +- A line matching `^Speaker (?P\S+): (?P.*)$` **starts a new turn**. +- Any following line that does **not** match that pattern is a wrapped + continuation and is appended to the current turn's text (joined with a single + space; surrounding whitespace stripped). +- Blank lines are skipped. +- **Consecutive turns by the same speaker are merged** into one synthesis + segment (fewer connections, more natural delivery). + +The result is an ordered list of `Segment(speaker_id, text)` with labels +removed. A label line with empty text contributes no text but still establishes +the speaker for following continuation lines. + +### Voice assignment + +- **Default rotation:** `["jane", "michael", "mary", "paul", "eve", "george"]` + (all confirmed-working PocketTTS English voices, ordered to alternate timbre). + Speakers are assigned in **first-appearance order**, wrapping when there are + more speakers than rotation entries. +- **Per-speaker override:** `--voice` is repeatable and accepts a + `SPEAKER=VOICE` form, e.g. `--voice A=vera --voice B=paul`. An explicit mapping + wins over the rotation for that speaker. The speaker id is matched + case-insensitively against the parsed ids. +- **Bare `--voice NAME`** (no `=`) keeps its current meaning: the voice for + single-voice (unlabeled) mode (falling back to `jane` when none is given). In + multi-voice mode an *explicitly passed* bare `--voice` is **ignored with a + one-line stderr note** pointing at the `--voice A=NAME` form, so it never + silently collapses every speaker to one voice. The note fires only when the + user actually typed a bare voice — see the empty-list default below. +- An override naming an unknown/erroring voice is not validated client-side; it + surfaces as the normal TTS error at synthesis time. + +### Synthesis & output + +- Each segment is synthesized with its assigned voice via `session.synthesize` + (one WebSocket connection per segment — voice is fixed at connect time, so a + voice change needs a new connection; merging consecutive same-speaker turns + keeps this minimal). `--language` and `--sample-rate` apply to every segment. +- Segments are concatenated into one 16-bit mono PCM buffer with **~250 ms of + silence between turns** for natural pacing (no leading/trailing pad). +- Output is unchanged in spirit: play through the speakers (interruptible — + Ctrl-C aborts immediately) or write one concatenated WAV with `--out`. + +### JSON output + +- **Multi-voice mode:** one object — + `{"mode": "multi", "speakers": {"A": "jane", "B": "michael"}, "segments": N, + "sample_rate": int, "audio_duration_seconds": float, "bytes": int, + "out": str|null}`. `speakers` lists the resolved id→voice map in + first-appearance order. +- **Single-voice mode:** the existing shape is unchanged + (`{voice, language, sample_rate, audio_duration_seconds, bytes, out}`). + +## Components + +- **`aai_cli/tts/dialogue.py`** (new) — pure, I/O-free, unit-testable: + - `looks_like_speaker_labeled(text: str) -> bool` + - `parse_segments(text: str) -> list[Segment]` (`Segment` = frozen + `(speaker_id, text)` dataclass) + - `parse_voice_overrides(values: list[str]) -> tuple[str | None, dict[str, str]]` + — splits repeatable `--voice` values into the bare voice (if any) and the + `SPEAKER→VOICE` map. + - `assign_voices(segments, rotation, overrides) -> list[tuple[str, str]]` + — `(voice, text)` per segment, applying overrides then first-appearance + rotation. Also returns/exposes the id→voice map for JSON. + - `DEFAULT_VOICE_ROTATION` constant. + +- **`aai_cli/tts/session.py`** — add + `synthesize_dialogue(api_key, segments, *, language, sample_rate, connect, + on_warning) -> SpeakResult`, looping `synthesize` per `(voice, text)` segment + and concatenating PCM with an inter-turn silence. Reuses the existing + connect/auth-mapping path. `SpeakResult` is unchanged. + +- **`aai_cli/tts/audio.py`** — add `silence(sample_rate, seconds) -> bytes` + (zeroed 16-bit mono PCM) for the inter-turn gap. + +- **`aai_cli/commands/speak.py`** — `--voice` becomes repeatable + (`list[str]`, default **`[]`**); single-voice mode uses the bare value if given + else `jane`. The empty default lets the command tell "user passed a bare voice" + from "default", so the multi-mode note fires only on an explicit bare voice. + Flow: read text → detect → if labeled, parse + resolve voices + + `synthesize_dialogue`; else single-voice `synthesize` as today → play or write + → emit human/JSON result. The bare-voice-in-multi-mode note is printed here. + +## Error handling + +| Condition | Result | +| --- | --- | +| Labeled input but a chosen voice errors mid-stream | the segment's `synthesize` raises the existing clean `APIError`; the command fails (no partial playback) | +| `--voice` value with an empty side (`=jane`, `A=`) | `UsageError` (exit 2) with the expected `SPEAKER=VOICE` form | +| Bare `--voice` in multi-voice mode | honored-but-ignored: one stderr note, synthesis proceeds with rotation/overrides | +| Empty text after stripping (only labels, no content) | existing "No text to speak" usage error (exit 2) | +| Production environment / no key / Ctrl-C | unchanged from current `speak` behavior | + +## Testing + +Must clear the mutation and 100%-patch-coverage tail gates — assert behavior, +not just execution. + +- **`dialogue.py`** (pure functions, the bulk of the logic): + - detection: labeled first line → true; prose / leading blank lines / a + `Speaker`-like word mid-sentence → false. + - parsing: single-line turns; an utterance wrapped across multiple physical + lines folds back into one segment with the right text; consecutive + same-speaker turns merge; blank lines skipped; a label line with empty text. + - override parsing: bare voice captured; `A=vera` mapped; case-insensitive id; + malformed (`=x`, `a=`) → error. + - assignment: first-appearance rotation, wrap past the rotation length, + override beats rotation, single-speaker labeled input → one voice. +- **`session.synthesize_dialogue`**: injected fake socket(s) drive several + segments; assert the per-segment voice query param, PCM concatenation order, + and the inserted silence length between turns (and none at the ends). +- **`audio.silence`**: returns exactly `seconds * sample_rate` frames of zero + bytes (even length). +- **`commands/speak.py`** (Typer `CliRunner`): a labeled stdin pipe → + multi-voice path with the expected id→voice map; `--voice A=vera` override + reflected; bare `--voice` in multi mode emits the note and still rotates; + `--json` multi shape; `--out` writes one WAV; single-voice path unchanged. +- Regenerate the `aai speak --help` syrupy snapshot for the repeatable `--voice` + option; never hand-edit `.ambr`. + +## Out of scope (v1) + +- A `--gap`/`--no-gap` flag (fixed ~250 ms pause for now). +- Non-AssemblyAI label formats / custom label regexes. +- Per-speaker language or sample-rate. +- Reusing one WebSocket across consecutive different-voice segments (a new + connection per voice change is acceptable). +- Client-side validation of voice names against a catalog. diff --git a/tests/__snapshots__/test_cli_output_snapshots.ambr b/tests/__snapshots__/test_cli_output_snapshots.ambr index d41d7ae9..89aa3487 100644 --- a/tests/__snapshots__/test_cli_output_snapshots.ambr +++ b/tests/__snapshots__/test_cli_output_snapshots.ambr @@ -643,6 +643,50 @@ + ''' +# --- +# name: test_command_help_matches_snapshot[speak] + ''' + + Usage: aai speak [OPTIONS] [TEXT] + + Synthesize speech from text with AssemblyAI streaming TTS (sandbox only). + + Plays the audio through your speakers by default, or writes a WAV with --out. + Speaker-labeled input (from 'aai transcribe --speaker-labels') is detected + automatically: the labels are stripped and each speaker gets a different + voice. This feature only exists in the sandbox today — run it with --sandbox. + + ╭─ Arguments ──────────────────────────────────────────────────────────────────╮ + │ text [TEXT] Text to speak. Omit to read from stdin. │ + ╰──────────────────────────────────────────────────────────────────────────────╯ + ╭─ Options ────────────────────────────────────────────────────────────────────╮ + │ --voice TEXT Voice id, or SPEAKER=VOICE for diarized │ + │ input (repeatable, e.g. --voice A=jane). │ + │ --language TEXT Language of the text. [default: English] │ + │ --sample-rate INTEGER Output sample rate in Hz. Server default if │ + │ omitted. │ + │ --out PATH Write a WAV file instead of playing through │ + │ the speakers. │ + │ --json -j Emit JSON metadata about the synthesized │ + │ audio. │ + │ --help Show this message and exit. │ + ╰──────────────────────────────────────────────────────────────────────────────╯ + + Examples + Speak text aloud (sandbox only) + $ aai speak "Hello there, friend." --sandbox + Pick a voice and language + $ aai speak "Bonjour" --voice jane --language French --sandbox + Speak a diarized transcript, one voice per speaker + $ aai transcribe meeting.mp3 --speaker-labels | aai speak --sandbox + Override a speaker's voice + $ … | aai speak --voice A=vera --voice B=paul --sandbox + Save to a WAV instead of playing + $ aai speak "Hello" --out /tmp/hello.wav --sandbox + + + ''' # --- # name: test_command_help_matches_snapshot[stream] diff --git a/tests/test_environments.py b/tests/test_environments.py index dd1b2c56..051bcc68 100644 --- a/tests/test_environments.py +++ b/tests/test_environments.py @@ -55,3 +55,15 @@ def test_profile_env_roundtrip(): assert config.get_profile_env("default") is None config.set_profile_env("default", "sandbox000") assert config.get_profile_env("default") == "sandbox000" + + +def test_sandbox_has_streaming_tts_host(): + assert ( + environments.get("sandbox000").streaming_tts_host + == "streaming-tts.sandbox000.assemblyai-labs.com" + ) + + +def test_production_has_no_streaming_tts_host(): + # Empty host is the "TTS not available here" signal the speak command checks. + assert environments.get("production").streaming_tts_host == "" diff --git a/tests/test_smoke.py b/tests/test_smoke.py index e65f7d8c..1751317d 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -102,6 +102,7 @@ def test_help_lists_commands_in_workflow_order(): "transcribe", "stream", "agent", + "speak", "llm", # Setup & Tools "doctor", diff --git a/tests/test_speak.py b/tests/test_speak.py new file mode 100644 index 00000000..742fc22d --- /dev/null +++ b/tests/test_speak.py @@ -0,0 +1,225 @@ +from __future__ import annotations + +import json + +import pytest +from typer.testing import CliRunner + +from aai_cli import config +from aai_cli.main import app +from aai_cli.tts import session + +runner = CliRunner() + + +@pytest.fixture(autouse=True) +def _fake_key(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(config, "resolve_api_key", lambda **_: "test-key") + + +@pytest.fixture +def fake_synthesize(monkeypatch: pytest.MonkeyPatch): + calls: dict[str, object] = {} + + def _fake(api_key, cfg, *, connect=None, on_warning=None): + calls["api_key"] = api_key + calls["cfg"] = cfg + return session.SpeakResult( + pcm=b"\x01\x02\x03\x04", sample_rate=24000, audio_duration_seconds=0.123456 + ) + + monkeypatch.setattr(session, "synthesize", _fake) + return calls + + +def test_production_env_is_rejected_with_sandbox_hint(): + result = runner.invoke(app, ["speak", "Hello"]) # default = production + assert result.exit_code == 2 + assert "only available in the sandbox" in result.output + assert "--sandbox" in result.output + + +def test_plays_audio_by_default(monkeypatch, fake_synthesize): + played: dict = {} + monkeypatch.setattr( + "aai_cli.commands.speak.audio.play_pcm", + lambda pcm, rate, **_: played.update(pcm=pcm, rate=rate), + ) + result = runner.invoke(app, ["--sandbox", "speak", "Hello there"]) + assert result.exit_code == 0 + assert played == {"pcm": b"\x01\x02\x03\x04", "rate": 24000} + assert fake_synthesize["cfg"].text == "Hello there" + # No --voice given -> single-voice path falls back to the default "jane". + assert fake_synthesize["cfg"].voice == "jane" + # Human summary (stderr) reports the default "played" disposition. + assert "played" in result.stderr + assert "saved to" not in result.stderr + + +def test_out_writes_wav_and_does_not_play(monkeypatch, tmp_path, fake_synthesize): + monkeypatch.setattr( + "aai_cli.commands.speak.audio.play_pcm", + lambda *a, **k: pytest.fail("should not play when --out is given"), + ) + written: dict = {} + monkeypatch.setattr( + "aai_cli.commands.speak.audio.write_wav", + lambda path, pcm, rate: written.update(path=path, pcm=pcm, rate=rate), + ) + out = tmp_path / "x.wav" + result = runner.invoke(app, ["--sandbox", "speak", "Hi", "--out", str(out)]) + assert result.exit_code == 0 + assert written["pcm"] == b"\x01\x02\x03\x04" + assert str(written["path"]) == str(out) + # Human summary (stderr) reports the file disposition, not "played". + assert "saved to" in result.stderr + assert "played" not in result.stderr + + +def test_reads_text_from_stdin_when_arg_omitted(monkeypatch, fake_synthesize): + monkeypatch.setattr("aai_cli.commands.speak.audio.play_pcm", lambda *a, **k: None) + result = runner.invoke(app, ["--sandbox", "speak"], input="piped text\n") + assert result.exit_code == 0 + assert fake_synthesize["cfg"].text == "piped text" + + +def test_empty_text_is_a_usage_error(monkeypatch): + # No arg and empty stdin -> usage error, before any synthesis. + result = runner.invoke(app, ["--sandbox", "speak"], input="") + assert result.exit_code == 2 + assert "No text to speak" in result.output + + +def test_blank_arg_does_not_fall_back_to_stdin(monkeypatch): + # A blank argument is a usage error; stdin is only read when the arg is omitted + # entirely, so an explicit empty arg must NOT silently pull from the pipe. + result = runner.invoke(app, ["--sandbox", "speak", " "], input="from stdin") + assert result.exit_code == 2 + assert "No text to speak" in result.output + + +def test_voice_and_language_flow_into_config(monkeypatch, fake_synthesize): + monkeypatch.setattr("aai_cli.commands.speak.audio.play_pcm", lambda *a, **k: None) + result = runner.invoke( + app, ["--sandbox", "speak", "Hi", "--voice", "jane", "--language", "English"] + ) + assert result.exit_code == 0 + cfg = fake_synthesize["cfg"] + assert cfg.voice == "jane" + assert cfg.language == "English" + assert cfg.query_params() == {"voice": "jane", "language": "English"} + + +def test_json_mode_emits_metadata_object_on_stdout(monkeypatch, fake_synthesize): + monkeypatch.setattr("aai_cli.commands.speak.audio.play_pcm", lambda *a, **k: None) + result = runner.invoke(app, ["--sandbox", "speak", "Hi", "--voice", "jane", "--json"]) + assert result.exit_code == 0 + # The behavioral split: --json yields a parseable object, not human prose. + payload = json.loads(result.stdout.strip()) + assert payload["voice"] == "jane" + assert payload["sample_rate"] == 24000 + assert payload["bytes"] == 4 + # Duration is rounded to 3 decimals (0.123456 -> 0.123, not 0.1235). + assert payload["audio_duration_seconds"] == 0.123 + # No --out -> the reported path is null, not the string "None". + assert payload["out"] is None + + +def test_human_mode_keeps_stdout_clean(monkeypatch, fake_synthesize): + monkeypatch.setattr("aai_cli.commands.speak.audio.play_pcm", lambda *a, **k: None) + result = runner.invoke(app, ["--sandbox", "speak", "Hi"]) + assert result.exit_code == 0 + # Human summary goes to stderr; stdout stays empty (audio went to the speaker). + assert result.stdout.strip() == "" + + +@pytest.fixture +def fake_dialogue(monkeypatch: pytest.MonkeyPatch): + calls: dict[str, object] = {} + + def _fake(api_key, segments, *, language=None, sample_rate=None, connect=None, on_warning=None): + calls["segments"] = segments + calls["language"] = language + return session.SpeakResult( + pcm=b"\x01\x02", sample_rate=24000, audio_duration_seconds=1.23456 + ) + + monkeypatch.setattr(session, "synthesize_dialogue", _fake) + monkeypatch.setattr("aai_cli.commands.speak.audio.play_pcm", lambda *a, **k: None) + return calls + + +def test_labeled_stdin_uses_dialogue_path_with_default_rotation(fake_dialogue): + text = "Speaker A: Hello there.\nSpeaker B: Hi.\nSpeaker A: Bye." + result = runner.invoke(app, ["--sandbox", "speak"], input=text) + assert result.exit_code == 0 + # Labels stripped; consecutive A turns are NOT merged (B between); voices rotate. + assert fake_dialogue["segments"] == [ + ("jane", "Hello there."), + ("michael", "Hi."), + ("jane", "Bye."), + ] + + +def test_speaker_voice_override_is_applied(fake_dialogue): + text = "Speaker A: One.\nSpeaker B: Two." + result = runner.invoke( + app, ["--sandbox", "speak", "--voice", "A=vera", "--voice", "B=paul"], input=text + ) + assert result.exit_code == 0 + assert fake_dialogue["segments"] == [("vera", "One."), ("paul", "Two.")] + + +def test_bare_voice_in_dialogue_mode_is_ignored_with_a_note(fake_dialogue): + text = "Speaker A: One.\nSpeaker B: Two." + result = runner.invoke(app, ["--sandbox", "speak", "--voice", "mary"], input=text) + assert result.exit_code == 0 + # The rotation still drives voices (bare voice ignored)... + assert fake_dialogue["segments"] == [("jane", "One."), ("michael", "Two.")] + # ...and the user is told why, pointed at the per-speaker form. + assert "A=NAME" in result.stderr + # The human note reports the speaker count, pinning len(speakers) in _emit_multi. + assert "2 voices" in result.stderr + + +def test_dialogue_json_reports_speaker_voice_map(fake_dialogue): + text = "Speaker A: One.\nSpeaker B: Two." + result = runner.invoke(app, ["--sandbox", "speak", "--json"], input=text) + assert result.exit_code == 0 + payload = json.loads(result.stdout.strip()) + assert payload["mode"] == "multi" + assert payload["speakers"] == {"A": "jane", "B": "michael"} + assert payload["segments"] == 2 + assert payload["sample_rate"] == 24000 + # 1.23456 rounded to 3 decimals -> pins the round(...) precision in _emit_multi. + assert payload["audio_duration_seconds"] == 1.235 + + +def test_dialogue_json_out_path_is_reported(fake_dialogue, monkeypatch, tmp_path): + # With --out, the multi JSON reports the file path (not null) — pins the + # `str(out) if out is not None else None` branch in _emit_multi. + monkeypatch.setattr("aai_cli.commands.speak.audio.write_wav", lambda *a, **k: None) + out = tmp_path / "dialogue.wav" + text = "Speaker A: One.\nSpeaker B: Two." + result = runner.invoke(app, ["--sandbox", "speak", "--out", str(out), "--json"], input=text) + assert result.exit_code == 0 + payload = json.loads(result.stdout.strip()) + assert payload["out"] == str(out) + + +def test_empty_speaker_labels_raises_usage_error(): + # Speaker-labeled input with no spoken text: detected as labeled, parses to zero + # segments, and raises the usage error before any synthesis. + result = runner.invoke(app, ["--sandbox", "speak"], input="Speaker A:\nSpeaker B:") + assert result.exit_code == 2 + assert "No text to speak" in result.output + assert "speaker labels" in result.output + + +def test_unlabeled_text_still_uses_single_voice_path(fake_synthesize, monkeypatch): + # A bare --voice still selects the single-voice voice for ordinary prose. + monkeypatch.setattr("aai_cli.commands.speak.audio.play_pcm", lambda *a, **k: None) + result = runner.invoke(app, ["--sandbox", "speak", "Just prose.", "--voice", "mary"]) + assert result.exit_code == 0 + assert fake_synthesize["cfg"].voice == "mary" + assert fake_synthesize["cfg"].text == "Just prose." diff --git a/tests/test_tts_audio.py b/tests/test_tts_audio.py new file mode 100644 index 00000000..5622598b --- /dev/null +++ b/tests/test_tts_audio.py @@ -0,0 +1,152 @@ +from __future__ import annotations + +import sys +import types +import wave +from pathlib import Path + +import pytest + +from aai_cli.errors import CLIError +from aai_cli.microphone import audio_missing_error +from aai_cli.tts import audio + + +def test_write_wav_produces_mono_16bit_wav(tmp_path: Path): + # Two missing levels deep, so the write only succeeds if parents are created. + out = tmp_path / "deep" / "nested" / "out.wav" + pcm = b"\x01\x02\x03\x04" + audio.write_wav(out, pcm, 24000) + + with wave.open(str(out), "rb") as wav: + assert wav.getnchannels() == 1 + assert wav.getsampwidth() == 2 + assert wav.getframerate() == 24000 + assert wav.readframes(wav.getnframes()) == pcm + + +def test_write_wav_into_existing_dir(tmp_path: Path): + # The parent already exists; writing must not error (exist_ok must be set). + out = tmp_path / "flat.wav" + audio.write_wav(out, b"\x01\x02", 16000) + assert out.exists() + + +class FakeStream: + def __init__(self, *, raise_on_write: BaseException | None = None) -> None: + self.events: list[str] = [] + self.written: bytes = b"" + self.writes: list[bytes] = [] + self._raise_on_write = raise_on_write + + def start(self) -> None: + self.events.append("start") + + def write(self, data: bytes) -> None: + if self._raise_on_write is not None: + raise self._raise_on_write + chunk = bytes(data) + self.written += chunk + self.writes.append(chunk) + self.events.append("write") + + def stop(self) -> None: + self.events.append("stop") + + def abort(self) -> None: + self.events.append("abort") + + def close(self) -> None: + self.events.append("close") + + +def test_play_pcm_writes_to_started_stream_then_closes(): + stream = FakeStream() + audio.play_pcm(b"\x01\x02", 16000, stream_factory=lambda rate: stream) + assert stream.events == ["start", "write", "stop", "close"] + assert stream.written == b"\x01\x02" + + +def test_play_pcm_writes_audio_in_bounded_chunks(): + # A buffer larger than one chunk is written in fixed-size pieces (so a Ctrl-C + # can land between writes); the chunks reassemble to the original audio. + stream = FakeStream() + pcm = bytes(range(256)) * 40 # 10240 bytes > 2 * chunk + audio.play_pcm(pcm, 24000, stream_factory=lambda rate: stream) + assert [len(c) for c in stream.writes] == [4096, 4096, 2048] + assert b"".join(stream.writes) == pcm + + +def test_play_pcm_aborts_and_propagates_on_ctrl_c(): + # Ctrl-C mid-playback must stop the device immediately (abort, not just stop) + # and re-raise so the cancel reaches the CLI; the stream is still closed. + stream = FakeStream(raise_on_write=KeyboardInterrupt()) + with pytest.raises(KeyboardInterrupt): + audio.play_pcm(b"\x01\x02", 16000, stream_factory=lambda rate: stream) + assert "abort" in stream.events + assert "stop" not in stream.events # aborted, never reached the draining stop() + assert stream.events[-1] == "close" # finally still closed it + + +def test_play_pcm_wraps_write_failure_in_cli_error(): + # A device error mid-stream (not from the factory) maps to the same clean + # CLIError, and the stream is still closed via the finally block. + stream = FakeStream(raise_on_write=RuntimeError("device fell over")) + with pytest.raises(CLIError, match="Could not play audio") as excinfo: + audio.play_pcm(b"\x01\x02", 16000, stream_factory=lambda rate: stream) + assert excinfo.value.exit_code == 1 + assert stream.events[-1] == "close" + + +def test_play_pcm_wraps_device_failure_in_cli_error(): + def _boom(_rate: int): + raise RuntimeError("no device") + + with pytest.raises(CLIError, match="Could not play audio") as excinfo: + audio.play_pcm(b"\x01\x02", 16000, stream_factory=_boom) + assert excinfo.value.exit_code == 1 + + +def test_play_pcm_reraises_cli_error_unchanged(monkeypatch: pytest.MonkeyPatch): + # A CLIError from the factory (e.g. audio_missing_error) is already user-facing, + # so it must propagate as-is, NOT get re-wrapped in "Could not play audio". + def _missing(_rate: int): + raise audio_missing_error() + + with pytest.raises(CLIError) as excinfo: + audio.play_pcm(b"\x01\x02", 16000, stream_factory=_missing) + assert "Could not play audio" not in excinfo.value.message + + +def test_default_output_stream_opens_raw_int16_mono_stream(monkeypatch: pytest.MonkeyPatch): + captured: dict[str, object] = {} + + sentinel = object() + + def _raw_output_stream(**kwargs: object) -> object: + captured.update(kwargs) + return sentinel + + fake_sd = types.ModuleType("sounddevice") + monkeypatch.setattr(fake_sd, "RawOutputStream", _raw_output_stream, raising=False) + monkeypatch.setitem(sys.modules, "sounddevice", fake_sd) + + stream: object = audio._default_output_stream(24000) + assert stream is sentinel # returns exactly what RawOutputStream produced + assert captured == {"samplerate": 24000, "channels": 1, "dtype": "int16"} + + +def test_default_output_stream_missing_sounddevice_raises_audio_missing( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setitem(sys.modules, "sounddevice", None) # import -> ImportError + with pytest.raises(CLIError): + audio._default_output_stream(24000) + + +def test_silence_returns_zeroed_pcm_of_the_right_length(): + # 16-bit mono: 100 ms at 16 kHz = 1600 frames = 3200 zero bytes. + pcm = audio.silence(16000, 0.1) + assert pcm == b"\x00" * 3200 + # Empty duration -> no bytes. + assert audio.silence(24000, 0.0) == b"" diff --git a/tests/test_tts_dialogue.py b/tests/test_tts_dialogue.py new file mode 100644 index 00000000..6f3dcbc8 --- /dev/null +++ b/tests/test_tts_dialogue.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +import pytest + +from aai_cli.errors import UsageError +from aai_cli.tts import dialogue + + +def test_detects_labeled_input_on_first_nonblank_line(): + assert dialogue.looks_like_speaker_labeled("Speaker A: hi\nSpeaker B: yo") is True + # Leading blank lines are skipped before the check. + assert dialogue.looks_like_speaker_labeled("\n\nSpeaker A: hi") is True + + +def test_plain_prose_is_not_labeled(): + assert dialogue.looks_like_speaker_labeled("Hello there, friend.") is False + # A mid-sentence "Speaker" word must not trigger detection. + assert dialogue.looks_like_speaker_labeled("The Speaker said hello") is False + assert dialogue.looks_like_speaker_labeled("") is False + + +def test_segment_is_immutable(): + seg = dialogue.Segment("A", "hi") + attr = "text" # a variable (not a constant) keeps ruff's B010 from rewriting setattr + with pytest.raises(AttributeError): # frozen=True -> assignment is blocked + setattr(seg, attr, "changed") + + +def test_parse_single_line_turns(): + segs = dialogue.parse_segments("Speaker A: Hello.\nSpeaker B: Hi there.") + assert [(s.speaker_id, s.text) for s in segs] == [("A", "Hello."), ("B", "Hi there.")] + + +def test_parse_folds_wrapped_continuation_lines(): + # An utterance wrapped across physical lines (no label on the 2nd line) folds + # back into one segment, joined with single spaces. + text = "Speaker A: This is a long line that wrapped\nonto a second line here\nSpeaker B: Ok." + segs = dialogue.parse_segments(text) + assert [(s.speaker_id, s.text) for s in segs] == [ + ("A", "This is a long line that wrapped onto a second line here"), + ("B", "Ok."), + ] + + +def test_parse_merges_consecutive_same_speaker_turns(): + segs = dialogue.parse_segments("Speaker A: One.\nSpeaker A: Two.\nSpeaker B: Three.") + assert [(s.speaker_id, s.text) for s in segs] == [("A", "One. Two."), ("B", "Three.")] + + +def test_parse_skips_blank_lines_and_drops_empty_turns(): + segs = dialogue.parse_segments("Speaker A: Hi.\n\nSpeaker B: \nSpeaker A: Bye.") + # Speaker B's empty turn is dropped; the two A turns are not merged (B is between). + assert [(s.speaker_id, s.text) for s in segs] == [("A", "Hi."), ("A", "Bye.")] + + +def test_parse_label_only_line_with_continuation(): + # A label line with no inline text, followed by a wrapped continuation line: + # the empty first part must be filtered so the text has no leading space. + segs = dialogue.parse_segments("Speaker A:\ncontinuation text\nSpeaker B: Ok.") + assert [(s.speaker_id, s.text) for s in segs] == [ + ("A", "continuation text"), + ("B", "Ok."), + ] + + +def test_parse_voice_overrides_splits_bare_and_mapped(): + bare, overrides = dialogue.parse_voice_overrides(["A=vera", "mary", "B=paul"]) + assert bare == "mary" + assert overrides == {"a": "vera", "b": "paul"} # ids casefolded + + +def test_parse_voice_overrides_bare_last_wins_and_empty_default(): + assert dialogue.parse_voice_overrides([]) == (None, {}) + assert dialogue.parse_voice_overrides(["jane", "mary"]) == ("mary", {}) + + +@pytest.mark.parametrize("bad", ["=vera", "A=", " = "]) +def test_parse_voice_overrides_rejects_malformed_pair(bad: str) -> None: + with pytest.raises(UsageError): + dialogue.parse_voice_overrides([bad]) + + +def test_assign_voices_rotates_in_first_appearance_order(): + segs = [dialogue.Segment(s, "x") for s in ("A", "B", "A", "C")] + resolved, mapping = dialogue.assign_voices(segs, ["jane", "michael", "mary"], {}) + assert [v for v, _ in resolved] == ["jane", "michael", "jane", "mary"] + assert mapping == {"A": "jane", "B": "michael", "C": "mary"} + + +def test_assign_voices_wraps_past_rotation_length(): + # 3-voice rotation, 4 speakers: the 4th wraps to the 1st voice. This only holds + # when the rotation index advances correctly, so it pins the wrap arithmetic. + segs = [dialogue.Segment(s, "x") for s in ("A", "B", "C", "D")] + resolved, _ = dialogue.assign_voices(segs, ["jane", "michael", "mary"], {}) + assert [v for v, _ in resolved] == ["jane", "michael", "mary", "jane"] + + +def test_assign_voices_override_beats_rotation_without_consuming_a_slot(): + segs = [dialogue.Segment(s, "x") for s in ("A", "B")] + # A is overridden, so B still gets the FIRST rotation voice, not the second. + resolved, mapping = dialogue.assign_voices(segs, ["jane", "michael"], {"a": "vera"}) + assert [v for v, _ in resolved] == ["vera", "jane"] + assert mapping == {"A": "vera", "B": "jane"} + + +def test_default_rotation_is_the_confirmed_working_voices(): + assert dialogue.DEFAULT_VOICE_ROTATION == ("jane", "michael", "mary", "paul", "eve", "george") diff --git a/tests/test_tts_session.py b/tests/test_tts_session.py new file mode 100644 index 00000000..d086a938 --- /dev/null +++ b/tests/test_tts_session.py @@ -0,0 +1,319 @@ +from __future__ import annotations + +import base64 +import json + +import pytest + +from aai_cli import environments +from aai_cli.errors import APIError, CLIError, NotAuthenticated +from aai_cli.tts import session + + +def _use_env(name: str) -> None: + environments.set_active(environments.get(name)) + + +def test_is_available_true_in_sandbox(): + _use_env("sandbox000") + assert session.is_available() is True + + +def test_is_available_false_in_production(): + _use_env("production") + assert session.is_available() is False + + +def test_ws_url_includes_set_params_only(): + _use_env("sandbox000") + cfg = session.SpeakConfig(text="hi", voice="jane", language="English") + url = session.ws_url(cfg.query_params()) + assert url.startswith("wss://streaming-tts.sandbox000.assemblyai-labs.com/v1/ws/?") + assert "voice=jane" in url + assert "language=English" in url + assert "sample_rate" not in url + + +def test_ws_url_no_params_has_no_query_string(): + _use_env("sandbox000") + url = session.ws_url(session.SpeakConfig(text="hi").query_params()) + assert url == "wss://streaming-tts.sandbox000.assemblyai-labs.com/v1/ws/" + + +def test_query_params_serializes_sample_rate_as_string(): + cfg = session.SpeakConfig(text="hi", sample_rate=16000) + assert cfg.query_params() == {"sample_rate": "16000"} + + +def _set_attr(obj: object, name: str, value: object) -> None: + # Indirect setattr: the attribute name is opaque to the type checker (so no + # read-only error) and non-constant (so ruff's B010 leaves it as setattr), + # while still hitting a frozen dataclass's blocking __setattr__ at runtime. + setattr(obj, name, value) + + +def test_speak_config_is_immutable(): + cfg = session.SpeakConfig(text="hi") + with pytest.raises(AttributeError): # frozen=True -> assignment is blocked + _set_attr(cfg, "text", "changed") + + +def test_speak_result_is_immutable(): + result = session.SpeakResult(pcm=b"\x01", sample_rate=24000, audio_duration_seconds=0.0) + with pytest.raises(AttributeError): + _set_attr(result, "sample_rate", 1) + + +class FakeWS: + """A minimal stand-in for a websockets sync connection.""" + + def __init__(self, incoming: list[str]) -> None: + self._incoming = list(incoming) + self.sent: list[dict[str, object]] = [] + self.closed = False + + def recv(self) -> str: + return self._incoming.pop(0) + + def send(self, data: str) -> None: + self.sent.append(json.loads(data)) + + def close(self) -> None: + self.closed = True + + +def _audio_frame(pcm: bytes, *, final: bool) -> str: + # The real server's Audio frames carry only the PCM payload and the final flag; + # the sample rate is reported once, up front, in the Begin frame's configuration. + return json.dumps( + { + "type": "Audio", + "audio": base64.b64encode(pcm).decode("ascii"), + "is_final": final, + } + ) + + +def _begin_frame(*, sample_rate: int = 24000) -> str: + return json.dumps( + { + "type": "Begin", + "id": "s1", + "expires_at": 1, + "configuration": {"voice": "jane", "language": "english", "sample_rate": sample_rate}, + } + ) + + +def _connect_returning(ws: FakeWS, captured: dict[str, object]): + def _connect(url: str, **kwargs): + captured["url"] = url + captured["kwargs"] = kwargs + return ws + + return _connect + + +def test_synthesize_drives_the_full_protocol(): + captured: dict = {} + ws = FakeWS( + [ + _begin_frame(sample_rate=24000), + _audio_frame(b"\x01\x02\x03\x04", final=False), + _audio_frame(b"\x05\x06", final=True), + ] + ) + cfg = session.SpeakConfig(text="hello", voice="jane") + result = session.synthesize("k", cfg, connect=_connect_returning(ws, captured)) + + # Sends Generate(text), then ForceFlushTextBuffer, then Terminate — in that order. + assert [m["type"] for m in ws.sent] == ["Generate", "ForceFlushTextBuffer", "Terminate"] + assert ws.sent[0]["text"] == "hello" + # Accumulates decoded PCM across chunks and stops on the is_final frame. + assert result.pcm == b"\x01\x02\x03\x04\x05\x06" + # The sample rate comes from the Begin frame's configuration, not the Audio frames. + assert result.sample_rate == 24000 + # 6 bytes / 2 bytes-per-sample / 24000 Hz. + assert result.audio_duration_seconds == pytest.approx(6 / 2 / 24000) + # Auth header carries the key as a Bearer token. + assert captured["kwargs"]["additional_headers"]["Authorization"] == "Bearer k" + assert ws.closed is True + + +def test_synthesize_reads_sample_rate_from_begin_configuration(): + # A non-default rate in the Begin frame flows into the result and its duration, + # proving the rate is read from Begin.configuration rather than hardcoded. + ws = FakeWS([_begin_frame(sample_rate=16000), _audio_frame(b"\x01\x02\x03\x04", final=True)]) + result = session.synthesize("k", session.SpeakConfig(text="hi"), connect=lambda *a, **k: ws) + assert result.sample_rate == 16000 + assert result.audio_duration_seconds == pytest.approx(4 / 2 / 16000) + + +def test_synthesize_falls_back_to_default_rate_when_begin_omits_configuration(): + # Begin without a configuration block -> the 24 kHz fallback, not a KeyError. + ws = FakeWS( + [ + json.dumps({"type": "Begin", "id": "s", "expires_at": 1}), + _audio_frame(b"\x01\x02", final=True), + ] + ) + result = session.synthesize("k", session.SpeakConfig(text="hi"), connect=lambda *a, **k: ws) + assert result.sample_rate == 24000 + + +def test_synthesize_raises_on_missing_begin(): + ws = FakeWS([json.dumps({"type": "Audio"})]) + with pytest.raises(APIError): + session.synthesize("k", session.SpeakConfig(text="hi"), connect=lambda *a, **k: ws) + + +def test_synthesize_maps_error_frame_to_api_error(): + ws = FakeWS( + [ + json.dumps({"type": "Begin", "id": "s", "expires_at": 1}), + json.dumps({"type": "Error", "error_code": 500, "error": "boom"}), + ] + ) + with pytest.raises(APIError, match="boom"): + session.synthesize("k", session.SpeakConfig(text="hi"), connect=lambda *a, **k: ws) + + +def test_synthesize_invokes_on_warning_then_continues(): + seen: list[str] = [] + ws = FakeWS( + [ + json.dumps({"type": "Begin", "id": "s", "expires_at": 1}), + json.dumps({"type": "Warning", "warning_code": 1, "warning": "slow"}), + _audio_frame(b"\x01\x02", final=True), + ] + ) + result = session.synthesize( + "k", + session.SpeakConfig(text="hi"), + connect=lambda *a, **k: ws, + on_warning=seen.append, + ) + assert seen == ["slow"] + assert result.pcm == b"\x01\x02" + + +def test_synthesize_ignores_warning_without_callback(): + ws = FakeWS( + [ + json.dumps({"type": "Begin", "id": "s", "expires_at": 1}), + json.dumps({"type": "Warning", "warning_code": 1, "warning": "slow"}), + _audio_frame(b"\x01\x02", final=True), + ] + ) + # No on_warning: the warning is silently skipped, not an error. + result = session.synthesize("k", session.SpeakConfig(text="hi"), connect=lambda *a, **k: ws) + assert result.pcm == b"\x01\x02" + + +def test_synthesize_maps_rejected_key_on_connect_to_not_authenticated(): + class Rejected(Exception): + pass + + def _connect(*_a, **_k): + raise Rejected("Unauthorized") + + with pytest.raises(NotAuthenticated): + session.synthesize("k", session.SpeakConfig(text="hi"), connect=_connect) + + +def test_synthesize_maps_forbidden_connect_to_api_error(): + class Resp: + status_code = 403 + + class Forbidden(Exception): + response = Resp() + + def _connect(*_a, **_k): + raise Forbidden("Unauthorized") # 403 -> NOT a rejected key + + with pytest.raises(APIError): + session.synthesize("k", session.SpeakConfig(text="hi"), connect=_connect) + + +def test_synthesize_error_frame_without_details_says_unknown(): + # An Error frame with no error_code / error message still produces a clean, + # non-empty message (the "unknown" fallback), not a KeyError. + ws = FakeWS( + [ + json.dumps({"type": "Begin", "id": "s", "expires_at": 1}), + json.dumps({"type": "Error"}), + ] + ) + with pytest.raises(APIError, match="unknown"): + session.synthesize("k", session.SpeakConfig(text="hi"), connect=lambda *a, **k: ws) + + +def test_synthesize_maps_unexpected_protocol_error_to_api_error(): + # recv() on an exhausted FakeWS raises IndexError inside _run_protocol; the + # generic-exception arm maps it to a clean APIError rather than leaking. + ws = FakeWS([json.dumps({"type": "Begin", "id": "s", "expires_at": 1})]) + with pytest.raises(APIError, match="TTS session failed"): + session.synthesize("k", session.SpeakConfig(text="hi"), connect=lambda *a, **k: ws) + + +def test_synthesize_without_connect_uses_real_client_and_fails_cleanly(): + # No `connect` provided: synthesize imports websockets' real sync client and + # attempts a connection. pytest-socket blocks socket creation, so this must + # surface as a clean CLIError (mapped in _open_ws), never a raw socket error. + _use_env("sandbox000") + with pytest.raises(CLIError): + session.synthesize("k", session.SpeakConfig(text="hi")) + + +def test_synthesize_dialogue_concatenates_segments_with_silence(): + # One fresh fake socket per segment; record the voice each connection requested. + sockets = [ + FakeWS([_begin_frame(sample_rate=24000), _audio_frame(b"\xaa\xbb", final=True)]), + FakeWS([_begin_frame(sample_rate=24000), _audio_frame(b"\xcc\xdd", final=True)]), + ] + urls: list[str] = [] + + def _connect(url: str, **_kwargs): + urls.append(url) + return sockets.pop(0) + + result = session.synthesize_dialogue( + "k", + [("jane", "Hello."), ("michael", "Hi.")], + language="English", + connect=_connect, + ) + # Each segment connected with its own voice. + assert "voice=jane" in urls[0] + assert "voice=michael" in urls[1] + # 0.25 s of silence (24000 * 0.25 * 2 = 12000 zero bytes) sits BETWEEN the two + # segments' PCM, with none at the ends. + gap = b"\x00" * 12000 + assert result.pcm == b"\xaa\xbb" + gap + b"\xcc\xdd" + assert result.sample_rate == 24000 + # Pin the duration formula (len/2/rate) so its operators survive the mutation gate. + assert result.audio_duration_seconds == pytest.approx(len(result.pcm) / 2 / 24000) + + +def test_synthesize_dialogue_single_segment_has_no_silence(): + ws = FakeWS([_begin_frame(sample_rate=24000), _audio_frame(b"\x01\x02", final=True)]) + result = session.synthesize_dialogue("k", [("jane", "Hi.")], connect=lambda *a, **k: ws) + assert result.pcm == b"\x01\x02" # no leading/trailing pad + + +def test_synthesize_dialogue_uses_server_sample_rate(): + # A non-default server rate must flow into the result (proving the per-segment + # rate is read, not left at the default) and into the duration denominator. + ws = FakeWS([_begin_frame(sample_rate=16000), _audio_frame(b"\x01\x02", final=True)]) + result = session.synthesize_dialogue("k", [("jane", "Hi.")], connect=lambda *a, **k: ws) + assert result.sample_rate == 16000 + assert result.audio_duration_seconds == pytest.approx(2 / 2 / 16000) + + +def test_synthesize_dialogue_empty_segments_returns_silent_default(): + # No segments -> no audio at the default rate, and no crash. connect is omitted + # entirely: the loop body never runs, so no connection is ever attempted. + result = session.synthesize_dialogue("k", []) + assert result.pcm == b"" + assert result.sample_rate == 24000 + assert result.audio_duration_seconds == 0.0