Add assembly webhooks listen command for local webhook development

.importlinter

@@ -38,6 +38,7 @@ source_modules =
     aai_cli.transcribe_batch
     aai_cli.transcribe_exec
     aai_cli.transcribe_render
+    aai_cli.webhook_listen
     aai_cli.wer
     aai_cli.ws
     aai_cli.youtube
@@ -66,6 +67,7 @@ modules =
     aai_cli.commands.telemetry
     aai_cli.commands.transcribe
     aai_cli.commands.transcripts
+    aai_cli.commands.webhooks
 
 [importlinter:contract:3]
 name = Library layers do not depend on Rich rendering

AGENTS.md

@@ -162,7 +162,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`, `agent`, `speak`, `llm`, `transcripts`, `login` (login/logout/whoami), `doctor`, `init`, `dev`, `share`, `deploy`, `setup`, `onboard`, `account` (balance/usage/limits), `keys`, `sessions`, `audit`, `telemetry` (status/enable/disable)). 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`, `telemetry` (status/enable/disable), `webhooks` (listen)). 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.
 
 **Options/run split for flag-heavy commands** (gh-CLI style): the Typer function only parses argv into a frozen `<Cmd>Options` dataclass and hands it to a module-level `run_<cmd>(opts, state, *, json_mode)` through a thin lambda adapter in `run_command(ctx, ..., json=...)`. The five run commands follow it — `aai_cli/stream_exec.py` (the reference implementation), `transcribe_exec.py`, `agent_exec.py`, `speak_exec.py`, `llm_exec.py`. Because the run path is a plain function of data, tests construct options directly (`dataclasses.replace` off a defaults instance, see `tests/test_stream_exec.py` and `tests/test_command_options_seam.py`) instead of round-tripping argv through `CliRunner` — which is also the cheap way to kill mutation-gate mutants on orchestration lines. Follow this for new or heavily-reworked commands with long bodies; small commands keep the inline `body()` closure — the dataclass is pure ceremony there.
 

README.md

@@ -58,12 +58,13 @@ neither.
 ## 📋 Key Features
 
 - **Transcription**: `assembly transcribe` handles files, URLs, and YouTube/podcast pages, with flags for speaker labels, PII redaction, summarization, sentiment, chapters, and more.
-- **Batch transcription**: point `assembly transcribe` at a directory or glob (or pipe paths with `--from-stdin`) to transcribe everything concurrently, with sidecar files that make re-runs resumable.
+- **Batch transcription**: point `assembly transcribe` at a directory or glob (or pipe paths with `--from-stdin`) to transcribe everything concurrently, with sidecar files that make re-runs resumable. Add `--llm "prompt"` to run an LLM prompt over each finished transcript, saved into the sidecars.
 - **Real-time streaming**: `assembly stream` transcribes the microphone, a file, or a URL live — on macOS it can capture system audio too.
 - **Voice agent**: `assembly agent` runs a full-duplex spoken conversation in your terminal (use headphones).
 - **LLM Gateway**: `assembly llm` prompts an LLM over a transcript, stdin, or a live stream (`assembly stream --llm "summarize as I talk"`).
 - **Model evaluation**: `assembly eval` transcribes a Hugging Face dataset (with built-in aliases for common benchmarks: `assembly eval tedlium`) or a local `.csv`/`.jsonl` manifest and scores WER against its references — handy for picking a speech model.
 - **Starter apps**: `assembly init` scaffolds a self-contained FastAPI + HTML app (`audio-transcription`, `live-captions`, `voice-agent`).
+- **Webhook testing**: `assembly webhooks listen` opens a public dev URL (cloudflared quick tunnel) that prints webhook deliveries as they arrive and can forward them to your local app with `--forward-to`.
 - **Code generation**: add `--show-code` to `transcribe`/`stream`/`agent` to print the equivalent Python SDK script instead of running.
 - **Account self-service**: `assembly keys` / `balance` / `usage` / `limits` / `sessions` / `audit` via browser login.
 

aai_cli/commands/share.py

@@ -2,16 +2,12 @@
 from __future__ import annotations
 
 import os
-import shutil
-import subprocess
-import sys
-import tempfile
 from pathlib import Path
 
 import typer
 from rich.markup import escape
 
-from aai_cli import config, help_panels, options, output, steps
+from aai_cli import help_panels, options, output, steps
 from aai_cli.context import AppState, run_command
 from aai_cli.errors import CLIError
 from aai_cli.help_text import examples_epilog
@@ -21,30 +17,6 @@
 app = typer.Typer()
 
 
-# brew exists only on macOS; everywhere else point at Cloudflare's install docs.
-_CLOUDFLARED_DOCS = (
-    "https://developers.cloudflare.com/cloudflare-one/connections/connect-networks/downloads/"
-)
-
-
-def _cloudflared_install_hint() -> str:
-    # A ternary (not an if/return) so neither branch reads as unreachable under
-    # mypy --warn-unreachable, which targets one platform at a time: on macOS the
-    # second return looked dead, on Linux the first would.
-    hint = "brew install cloudflared" if sys.platform == "darwin" else _CLOUDFLARED_DOCS
-    return f"Install it: {hint}"
-
-
-def _require_cloudflared() -> None:
-    if shutil.which(tunnel.CLOUDFLARED) is None:
-        raise CLIError(
-            "cloudflared is required to share a public link.",
-            error_type="missing_dependency",
-            exit_code=1,
-            suggestion=_cloudflared_install_hint(),
-        )
-
-
 def _render_share(data: dict[str, object]) -> str:
     return (
         f"[aai.heading]Sharing[/aai.heading] [aai.url]{escape(str(data['url']))}[/aai.url]\n"
@@ -53,11 +25,6 @@ def _render_share(data: dict[str, object]) -> str:
     )
 
 
-def _terminate(proc: subprocess.Popen[str] | None) -> None:
-    if proc is not None and proc.poll() is None:
-        proc.terminate()
-
-
 def run_share(*, port: int, no_install: bool, json_mode: bool, quiet: bool) -> None:
     """Boot the app and expose it on a public cloudflared quick-tunnel URL."""
     target = Path.cwd()
@@ -67,7 +34,7 @@ def run_share(*, port: int, no_install: bool, json_mode: bool, quiet: bool) -> N
     devserver.notify_port_change(port, chosen_port, json_mode=json_mode, quiet=quiet)
     env = {**os.environ, "PORT": str(chosen_port)}
     web = procfile.web_argv(target, env=env)  # validates we're in a scaffolded project
-    _require_cloudflared()
+    tunnel.require_cloudflared("share a public link")
 
     report: list[steps.Step] = [
         devserver.install_step(target, no_install=no_install, use_uv=use_uv)
@@ -77,7 +44,7 @@ def run_share(*, port: int, no_install: bool, json_mode: bool, quiet: bool) -> N
         raise typer.Exit(code=1)
 
     server = runner.spawn(devserver.dev_command(target, web, use_uv=use_uv), cwd=target, env=env)
-    proxy: subprocess.Popen[str] | None = None
+    proxy = None
     log_path: Path | None = None
     keep_log = False
     try:
@@ -87,16 +54,7 @@ def run_share(*, port: int, no_install: bool, json_mode: bool, quiet: bool) -> N
                 error_type="server_error",
                 exit_code=1,
             )
-        fd, name = tempfile.mkstemp(prefix="aai-tunnel-", suffix=".log")
-        os.close(fd)
-        log_path = Path(name)
-        # The tunnel binary only proxies the port; don't hand it the API key the
-        # dev server needs (keeps the secret out of cloudflared's logs/diagnostics).
-        tunnel_env = {k: v for k, v in os.environ.items() if k != config.ENV_API_KEY}
-        proxy = runner.spawn(
-            tunnel.tunnel_command(chosen_port), cwd=target, env=tunnel_env, log_path=log_path
-        )
-        public = tunnel.await_url(log_path)
+        proxy, public, log_path = tunnel.open_quick_tunnel(chosen_port, cwd=target)
         if public is None:
             # Keep the captured cloudflared output: it's the only evidence of why
             # the tunnel never came up.
@@ -119,8 +77,8 @@ def run_share(*, port: int, no_install: bool, json_mode: bool, quiet: bool) -> N
         # block below tears down the tunnel and server.
         pass
     finally:
-        _terminate(proxy)
-        _terminate(server)
+        tunnel.terminate(proxy)
+        tunnel.terminate(server)
         if log_path is not None and not keep_log:
             log_path.unlink(missing_ok=True)
 

aai_cli/commands/transcribe.py

@@ -26,6 +26,7 @@
             ("Redact PII for compliance", "assembly transcribe call.mp3 --redact-pii"),
             ("Summarize a recording", "assembly transcribe call.mp3 --summarization"),
             ("Ask about the transcript", 'assembly transcribe call.mp3 --llm "List action items"'),
+            ("Summarize a whole folder", 'assembly transcribe ./calls --llm "Summarize this call"'),
         ]
     ),
 )
@@ -257,7 +258,7 @@ def transcribe(
     webhook_url: str | None = typer.Option(
         None,
         "--webhook-url",
-        help="Webhook URL for completion.",
+        help="Webhook URL for completion (get a dev URL with `assembly webhooks listen`).",
         rich_help_panel=help_panels.OPT_WEBHOOKS,
     ),
     webhook_auth_header: str | None = typer.Option(
@@ -342,7 +343,9 @@ def transcribe(
 
     Batch mode: pass a directory or glob (or pipe a list with --from-stdin) to
     transcribe many sources concurrently. Each source gets a .aai.json sidecar
-    with the full result, and a re-run skips sources already transcribed.
+    with the full result (including any --llm responses), and a re-run skips
+    sources already transcribed — with changed --llm prompts it replays just
+    the LLM step, never a second transcription.
 
     Curated flags cover common features; --config KEY=VALUE and --config-file reach every other field. Analysis (summary, chapters, ...) renders in human mode.
     """

aai_cli/commands/webhooks.py

@@ -0,0 +1,71 @@
+# aai_cli/commands/webhooks.py
+from __future__ import annotations
+
+import typer
+
+from aai_cli import options, webhook_listen
+from aai_cli.context import AppState, run_command
+from aai_cli.help_text import examples_epilog
+
+app = typer.Typer(help="Receive webhook deliveries on a public dev URL.", no_args_is_help=True)
+
+
+@app.command(
+    epilog=examples_epilog(
+        [
+            ("Listen on a public URL", "assembly webhooks listen"),
+            (
+                "Deliver a transcription to it",
+                "assembly transcribe call.mp3 --webhook-url https://….trycloudflare.com",
+            ),
+            (
+                "Forward deliveries to your app",
+                "assembly webhooks listen --forward-to http://localhost:8000/webhook",
+            ),
+            ("Local-only sink (no tunnel)", "assembly webhooks listen --no-tunnel"),
+            ("Stop after the first delivery", "assembly webhooks listen --max-events 1"),
+        ]
+    ),
+)
+def listen(
+    ctx: typer.Context,
+    port: int = typer.Option(
+        8989, "--port", help="Local listener port (the first free port from here)."
+    ),
+    forward_to: str | None = typer.Option(
+        None, "--forward-to", help="Re-POST each delivery to this URL (e.g. your local app)."
+    ),
+    no_tunnel: bool = typer.Option(
+        False, "--no-tunnel", help="Local-only: skip the cloudflared public URL."
+    ),
+    max_events: int = typer.Option(
+        0,  # pragma: no mutate (0 = serve forever; only observable by never exiting)
+        "--max-events",
+        min=0,
+        help="Exit after this many deliveries (0 = run until Ctrl-C).",
+    ),
+    json_out: bool = options.json_option("One NDJSON record per delivery."),
+) -> None:
+    """Receive AssemblyAI webhooks on a public URL and watch them arrive.
+
+    Opens a cloudflared quick tunnel to a local listener and prints the public
+    URL to pass as --webhook-url (or webhook_url in the API). Each delivery is
+    acknowledged with HTTP 200 and printed; --forward-to relays it to your app,
+    so you can build webhook handlers without deploying a public endpoint.
+
+    Requires cloudflared unless --no-tunnel (macOS: `brew install cloudflared`; other
+    platforms: https://developers.cloudflare.com/cloudflare-one/connections/connect-networks/downloads/).
+    """
+
+    def body(_state: AppState, json_mode: bool) -> None:
+        webhook_listen.run_listen(
+            webhook_listen.ListenOptions(
+                port=port,
+                forward_to=forward_to,
+                public=not no_tunnel,
+                max_events=max_events,
+            ),
+            json_mode=json_mode,
+        )
+
+    run_command(ctx, body, json=json_out)

aai_cli/init/tunnel.py

@@ -1,23 +1,78 @@
 # aai_cli/init/tunnel.py
 from __future__ import annotations
 
+import os
 import re
+import shutil
+import subprocess
+import sys
+import tempfile
 import time
 from collections.abc import Callable
 from pathlib import Path
 
+from aai_cli import config
+from aai_cli.errors import CLIError
+from aai_cli.init import runner
+
 # cloudflared binary name; resolved via shutil.which by callers.
 CLOUDFLARED = "cloudflared"
 
+# brew exists only on macOS; everywhere else point at Cloudflare's install docs.
+_CLOUDFLARED_DOCS = (
+    "https://developers.cloudflare.com/cloudflare-one/connections/connect-networks/downloads/"
+)
+
 # A cloudflared quick tunnel prints an ephemeral https://<slug>.trycloudflare.com URL.
 _URL = re.compile(r"https://[a-z0-9-]+\.trycloudflare\.com")
 
 
+def install_hint() -> str:
+    # A ternary (not an if/return) so neither branch reads as unreachable under
+    # mypy --warn-unreachable, which targets one platform at a time: on macOS the
+    # second return looked dead, on Linux the first would.
+    hint = "brew install cloudflared" if sys.platform == "darwin" else _CLOUDFLARED_DOCS
+    return f"Install it: {hint}"
+
+
+def require_cloudflared(purpose: str) -> None:
+    """Raise a clean missing-dependency error when cloudflared isn't on PATH."""
+    if shutil.which(CLOUDFLARED) is None:
+        raise CLIError(
+            f"cloudflared is required to {purpose}.",
+            error_type="missing_dependency",
+            exit_code=1,
+            suggestion=install_hint(),
+        )
+
+
 def tunnel_command(port: int) -> list[str]:
     """The cloudflared quick-tunnel command pointing at the local server."""
     return [CLOUDFLARED, "tunnel", "--url", f"http://localhost:{port}"]
 
 
+def open_quick_tunnel(port: int, *, cwd: Path) -> tuple[subprocess.Popen[str], str | None, Path]:
+    """Spawn a cloudflared quick tunnel for ``port``: (process, URL or None, log path).
+
+    The tunnel binary only proxies the port, so the API key is stripped from its
+    environment (keeps the secret out of cloudflared's logs/diagnostics). A None
+    URL means cloudflared never reported one — the caller should keep the log
+    file (the only evidence of why) and name it; on success it should unlink it.
+    """
+    fd, name = tempfile.mkstemp(prefix="aai-tunnel-", suffix=".log")
+    os.close(fd)
+    log_path = Path(name)
+    env = {k: v for k, v in os.environ.items() if k != config.ENV_API_KEY}
+    process = runner.spawn(tunnel_command(port), cwd=cwd, env=env, log_path=log_path)
+    return process, await_url(log_path), log_path
+
+
+def terminate(process: subprocess.Popen[str] | None) -> None:
+    """Terminate a spawned process if it's still running (None / exited: no-op)."""
+    if process is not None and process.poll() is None:
+        process.terminate()
+
+
 def find_url(text: str) -> str | None:
     """The first trycloudflare.com URL in `text`, or None."""
     match = _URL.search(text)

aai_cli/main.py

@@ -41,6 +41,7 @@
     telemetry,
     transcribe,
     transcripts,
+    webhooks,
 )
 from aai_cli.context import AppState, env_override_warning, resolve_environment
 from aai_cli.errors import CLIError, NotAuthenticated, UsageError
@@ -67,6 +68,7 @@
     "speak",
     "llm",
     "eval",
+    "webhooks",
     # Setup & Tools — get set up & maintain
     "doctor",
     "setup",
@@ -399,6 +401,7 @@ def main(
 app.add_typer(setup.app, name="setup", rich_help_panel=help_panels.SETUP)
 app.add_typer(telemetry.app, name="telemetry", rich_help_panel=help_panels.SETUP)
 app.add_typer(keys.app, name="keys", rich_help_panel=help_panels.ACCOUNT)
+app.add_typer(webhooks.app, name="webhooks", rich_help_panel=help_panels.TRANSCRIPTION)
 
 
 @app.command(