From eca9406620d8a9bd8c7b7c3c64cebf971abd8248 Mon Sep 17 00:00:00 2001
From: fanng <“fanng@apache.org”>
Date: Wed, 15 Jul 2026 17:26:11 +0900
Subject: [PATCH] Add distributed Daft audio benchmark

---
 .gitignore                       |   1 +
 README.md                        |   3 +
 benchmark/README.md              |  52 +++++
 benchmark/__init__.py            |   1 +
 benchmark/audio/__init__.py      |   1 +
 benchmark/audio/__main__.py      |   5 +
 benchmark/audio/cli.py           | 225 +++++++++++++++++++++
 benchmark/audio/config.py        |  94 +++++++++
 benchmark/audio/data.py          | 126 ++++++++++++
 benchmark/audio/metrics.py       | 111 +++++++++++
 benchmark/audio/mock_llm.py      | 126 ++++++++++++
 benchmark/audio/pipeline.py      | 327 +++++++++++++++++++++++++++++++
 benchmark/audio/report.py        | 161 +++++++++++++++
 benchmark/audio/storage.py       |  64 ++++++
 benchmark/audio/submit.py        |  54 +++++
 docs/audio-benchmark.md          | 258 ++++++++++++++++++++++++
 pyproject.toml                   |   5 +-
 tests/benchmark/__init__.py      |   1 +
 tests/benchmark/test_config.py   |  34 ++++
 tests/benchmark/test_data.py     |  15 ++
 tests/benchmark/test_e2e.py      |  12 ++
 tests/benchmark/test_mock_llm.py |  36 ++++
 tests/benchmark/test_report.py   |  43 ++++
 tests/benchmark/test_submit.py   |  28 +++
 uv.lock                          |  20 ++
 25 files changed, 1802 insertions(+), 1 deletion(-)
 create mode 100644 benchmark/README.md
 create mode 100644 benchmark/__init__.py
 create mode 100644 benchmark/audio/__init__.py
 create mode 100644 benchmark/audio/__main__.py
 create mode 100644 benchmark/audio/cli.py
 create mode 100644 benchmark/audio/config.py
 create mode 100644 benchmark/audio/data.py
 create mode 100644 benchmark/audio/metrics.py
 create mode 100644 benchmark/audio/mock_llm.py
 create mode 100644 benchmark/audio/pipeline.py
 create mode 100644 benchmark/audio/report.py
 create mode 100644 benchmark/audio/storage.py
 create mode 100644 benchmark/audio/submit.py
 create mode 100644 docs/audio-benchmark.md
 create mode 100644 tests/benchmark/__init__.py
 create mode 100644 tests/benchmark/test_config.py
 create mode 100644 tests/benchmark/test_data.py
 create mode 100644 tests/benchmark/test_e2e.py
 create mode 100644 tests/benchmark/test_mock_llm.py
 create mode 100644 tests/benchmark/test_report.py
 create mode 100644 tests/benchmark/test_submit.py

diff --git a/.gitignore b/.gitignore
index 4c4c331..2e52fc4 100644
--- a/.gitignore
+++ b/.gitignore
@@ -25,3 +25,4 @@ __pycache__/
 /data/images/*.bmp
 /data/images/*.webp
 .coverage
+.benchmarks/
diff --git a/README.md b/README.md
index 28c039e..49f53fe 100644
--- a/README.md
+++ b/README.md
@@ -4,6 +4,9 @@ Audio call-centre analysis POC: ingest recordings from S3, store audio as Lance
 
 Also includes an image analysis workflow (face presence + clarity detection), fully isolated from the audio pipeline — see [Image workflow](#image-workflow) below.
 
+For repeatable Daft audio performance and memory testing—from a local MinIO/Ray
+smoke test to a 4-worker cluster—see the [audio benchmark guide](docs/audio-benchmark.md).
+
 ## Architecture overview
 
 Media-specific analysis lives under each media package (`audio/workflow/` and
diff --git a/benchmark/README.md b/benchmark/README.md
new file mode 100644
index 0000000..1d8ac6d
--- /dev/null
+++ b/benchmark/README.md
@@ -0,0 +1,52 @@
+# Daft audio stability benchmark
+
+For complete local and multi-node Ray instructions, sizing guidance, metrics,
+and extension points, see the [audio benchmark guide](../docs/audio-benchmark.md).
+
+This package exercises the real audio path without calling a real LLM:
+
+```
+MinIO -> Daft/Ray download -> SenseVoice ASR -> HTTP mock LLM -> Lance blob v2
+```
+
+It always uses the MinIO credentials and endpoint from the repository `.env`.
+It never starts, stops, or deletes the configured MinIO service. Each run writes
+under its own `s3://<bucket>/<prefix>/<run-id>/` prefix.
+
+## Local smoke test
+
+Prerequisites: the configured MinIO is running, the SenseVoice and FSMN VAD
+models are cached or downloadable, and `data/audio/` contains seed audio.
+
+```bash
+uv run python -m benchmark.audio local-smoke --count 4
+```
+
+Results are written to `.benchmarks/<run-id>/`, including `metadata.json`,
+`summary.json`, `resources.csv`, `report.json`, `report.md`, the Daft physical
+plan, and Daft event logs.
+
+## Separate commands
+
+```bash
+uv run python -m benchmark.audio generate --profile smoke --count 4
+
+uv run python -m benchmark.audio serve-mock --host 0.0.0.0 --port 8010 --profile standard
+
+uv run python -m benchmark.audio run \
+  --manifest s3://benchmark/audio/RUN/manifest.parquet \
+  --lance-uri s3://benchmark/audio/RUN/output.lance \
+  --mock-url http://MOCK_HOST:8010 \
+  --ray-address auto --max-minutes 60
+
+uv run python -m benchmark.audio submit \
+  --dashboard-address http://RAY_HEAD:8265 \
+  --manifest s3://benchmark/audio/RUN/manifest.parquet \
+  --lance-uri s3://benchmark/audio/RUN/output.lance \
+  --mock-url http://MOCK_HOST:8010 \
+  --ray-address auto --max-minutes 60 --wait
+```
+
+Use `generate --profile fixed` for controlled fixed-duration data and
+`--profile mixed` for a 70% short / 25% medium / 5% long distribution. Data
+generation happens before the timed benchmark.
diff --git a/benchmark/__init__.py b/benchmark/__init__.py
new file mode 100644
index 0000000..22f3670
--- /dev/null
+++ b/benchmark/__init__.py
@@ -0,0 +1 @@
+"""Standalone benchmark tooling for multimodal-toolkit."""
diff --git a/benchmark/audio/__init__.py b/benchmark/audio/__init__.py
new file mode 100644
index 0000000..8ff1eed
--- /dev/null
+++ b/benchmark/audio/__init__.py
@@ -0,0 +1 @@
+"""Daft audio stability benchmark."""
diff --git a/benchmark/audio/__main__.py b/benchmark/audio/__main__.py
new file mode 100644
index 0000000..2f05ddc
--- /dev/null
+++ b/benchmark/audio/__main__.py
@@ -0,0 +1,5 @@
+from .cli import main
+
+
+if __name__ == "__main__":
+    main()
diff --git a/benchmark/audio/cli.py b/benchmark/audio/cli.py
new file mode 100644
index 0000000..bdd5764
--- /dev/null
+++ b/benchmark/audio/cli.py
@@ -0,0 +1,225 @@
+from __future__ import annotations
+
+import argparse
+import json
+from dataclasses import asdict
+from pathlib import Path
+
+from .config import BenchmarkConfig, DEFAULT_RUN_ROOT, new_run_id
+
+
+def _add_run_arguments(parser: argparse.ArgumentParser, *, ray_default: str = "local") -> None:
+    parser.add_argument("--run-id", default=None)
+    parser.add_argument("--manifest", required=True)
+    parser.add_argument("--lance-uri", required=True)
+    parser.add_argument("--mock-url", required=True)
+    parser.add_argument("--ray-address", default=ray_default)
+    parser.add_argument("--max-minutes", type=float, default=15.0)
+    parser.add_argument("--asr-actor-cpus", type=float, default=4.0)
+    parser.add_argument("--asr-actor-concurrency", type=int, default=1)
+    parser.add_argument("--asr-batch-size", type=int, default=1)
+    parser.add_argument("--llm-concurrency", type=int, default=8)
+    parser.add_argument("--llm-timeout-s", type=float, default=15.0)
+    parser.add_argument("--llm-max-attempts", type=int, default=3)
+    parser.add_argument("--sample-interval-s", type=float, default=5.0)
+    parser.add_argument(
+        "--num-partitions",
+        type=int,
+        default=None,
+        help="manifest partitions before download; default is 2x cluster CPUs",
+    )
+    parser.add_argument("--run-dir", default=None)
+
+
+def _config_from_args(args) -> BenchmarkConfig:
+    run_id = args.run_id or new_run_id()
+    run_dir = args.run_dir
+    if run_dir is None and args.command != "submit":
+        run_dir = str(DEFAULT_RUN_ROOT / run_id)
+    return BenchmarkConfig(
+        run_id=run_id,
+        manifest_uri=args.manifest,
+        lance_uri=args.lance_uri,
+        mock_url=args.mock_url,
+        ray_address=args.ray_address,
+        max_minutes=args.max_minutes,
+        asr_actor_cpus=args.asr_actor_cpus,
+        asr_actor_concurrency=args.asr_actor_concurrency,
+        asr_batch_size=args.asr_batch_size,
+        llm_concurrency=args.llm_concurrency,
+        llm_timeout_s=args.llm_timeout_s,
+        llm_max_attempts=args.llm_max_attempts,
+        sample_interval_s=args.sample_interval_s,
+        num_partitions=args.num_partitions,
+        run_dir=run_dir or "",
+    )
+
+
+def _cmd_generate(args) -> None:
+    from .data import generate_dataset
+    from .storage import preflight
+
+    run_id = args.run_id or new_run_id()
+    preflight(args.bucket)
+    result = generate_dataset(
+        run_id=run_id,
+        bucket=args.bucket,
+        prefix=args.prefix,
+        profile=args.profile,
+        count=args.count,
+        fixed_duration_s=args.duration_s,
+        source_dir=Path(args.source_dir) if args.source_dir else None,
+        seed=args.seed,
+    )
+    print(json.dumps({"run_id": run_id, **asdict(result)}, ensure_ascii=False, indent=2))
+
+
+def _cmd_run(args) -> None:
+    from .pipeline import run_pipeline
+    from .report import build_report
+
+    cfg = _config_from_args(args)
+    try:
+        summary = run_pipeline(cfg)
+    except Exception:
+        if (cfg.output_dir / "summary.json").exists():
+            build_report(cfg.output_dir)
+        raise
+    else:
+        report = build_report(cfg.output_dir)
+        print(json.dumps({"summary": summary, "report": report}, ensure_ascii=False, indent=2))
+
+
+def _cmd_submit(args) -> None:
+    from .submit import submit_job
+
+    cfg = _config_from_args(args)
+    job_id = submit_job(cfg, args.dashboard_address, wait=args.wait)
+    print(job_id)
+
+
+def _cmd_report(args) -> None:
+    from .report import build_report
+
+    print(json.dumps(build_report(Path(args.run_dir)), ensure_ascii=False, indent=2))
+
+
+def _cmd_local_smoke(args) -> None:
+    import ray
+
+    from .data import generate_dataset
+    from .mock_llm import start_in_thread
+    from .pipeline import run_pipeline
+    from .report import build_report
+    from .storage import preflight
+
+    run_id = args.run_id or new_run_id()
+    print(f"[preflight] checking configured MinIO bucket {args.bucket}")
+    print(json.dumps(preflight(args.bucket), ensure_ascii=False))
+    print(f"[generate] creating {args.count} smoke objects")
+    dataset = generate_dataset(
+        run_id=run_id,
+        bucket=args.bucket,
+        prefix=args.prefix,
+        profile="smoke",
+        count=args.count,
+        source_dir=Path(args.source_dir) if args.source_dir else None,
+    )
+    server, thread = start_in_thread(port=0, profile="fast")
+    port = server.server_address[1]
+    lance_uri = f"s3://{args.bucket}/{args.prefix.strip('/')}/{run_id}/output.lance"
+    cfg = BenchmarkConfig(
+        run_id=run_id,
+        manifest_uri=dataset.manifest_uri,
+        lance_uri=lance_uri,
+        mock_url=f"http://127.0.0.1:{port}",
+        ray_address="local",
+        max_minutes=args.max_minutes,
+        asr_actor_cpus=args.asr_actor_cpus,
+        asr_actor_concurrency=1,
+        asr_batch_size=1,
+        llm_concurrency=4,
+        llm_timeout_s=5.0,
+        sample_interval_s=1.0,
+        num_partitions=min(args.count, 2),
+        run_dir=str(DEFAULT_RUN_ROOT / run_id),
+    )
+    try:
+        print(f"[run] {dataset.manifest_uri} -> {lance_uri}")
+        try:
+            summary = run_pipeline(cfg)
+        except Exception:
+            if (cfg.output_dir / "summary.json").exists():
+                build_report(cfg.output_dir)
+            raise
+        report = build_report(cfg.output_dir)
+        if summary.get("rows") != dataset.rows:
+            raise RuntimeError(f"row count mismatch: manifest={dataset.rows}, lance={summary.get('rows')}")
+        if summary.get("unique_doc_ids") != dataset.rows:
+            raise RuntimeError("output doc_id values are not unique")
+        print(json.dumps(report, ensure_ascii=False, indent=2))
+        print(f"[ok] report: {cfg.output_dir / 'report.md'}")
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+        if ray.is_initialized():
+            ray.shutdown()
+
+
+def _cmd_serve_mock(args) -> None:
+    from .mock_llm import serve
+
+    serve(args.host, args.port, args.profile)
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Daft audio stability benchmark")
+    commands = parser.add_subparsers(dest="command", required=True)
+
+    generate = commands.add_parser("generate", help="create and upload benchmark audio")
+    generate.add_argument("--run-id", default=None)
+    generate.add_argument("--bucket", default="benchmark")
+    generate.add_argument("--prefix", default="audio")
+    generate.add_argument("--profile", choices=("smoke", "fixed", "mixed"), default="smoke")
+    generate.add_argument("--count", type=int, default=4)
+    generate.add_argument("--duration-s", type=float, default=30.0)
+    generate.add_argument("--source-dir", default=None)
+    generate.add_argument("--seed", type=int, default=20260715)
+    generate.set_defaults(func=_cmd_generate)
+
+    mock = commands.add_parser("serve-mock", help="run the OpenAI-compatible mock LLM")
+    mock.add_argument("--host", default="0.0.0.0")
+    mock.add_argument("--port", type=int, default=8010)
+    mock.add_argument("--profile", choices=("fast", "standard"), default="fast")
+    mock.set_defaults(func=_cmd_serve_mock)
+
+    run = commands.add_parser("run", help="execute the benchmark directly")
+    _add_run_arguments(run)
+    run.set_defaults(func=_cmd_run)
+
+    submit = commands.add_parser("submit", help="submit the benchmark with Ray Jobs")
+    _add_run_arguments(submit, ray_default="auto")
+    submit.add_argument("--dashboard-address", required=True)
+    submit.add_argument("--wait", action="store_true")
+    submit.set_defaults(func=_cmd_submit)
+
+    report = commands.add_parser("report", help="rebuild reports for a run")
+    report.add_argument("--run-dir", required=True)
+    report.set_defaults(func=_cmd_report)
+
+    smoke = commands.add_parser("local-smoke", help="run a real local end-to-end smoke test")
+    smoke.add_argument("--run-id", default=None)
+    smoke.add_argument("--bucket", default="benchmark")
+    smoke.add_argument("--prefix", default="audio")
+    smoke.add_argument("--count", type=int, default=4)
+    smoke.add_argument("--source-dir", default=None)
+    smoke.add_argument("--max-minutes", type=float, default=15.0)
+    smoke.add_argument("--asr-actor-cpus", type=float, default=4.0)
+    smoke.set_defaults(func=_cmd_local_smoke)
+    return parser
+
+
+def main() -> None:
+    args = build_parser().parse_args()
+    args.func(args)
diff --git a/benchmark/audio/config.py b/benchmark/audio/config.py
new file mode 100644
index 0000000..efee465
--- /dev/null
+++ b/benchmark/audio/config.py
@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+import json
+import os
+import platform
+import subprocess
+import sys
+from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from multimodal_toolkit import config as project_config
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+DEFAULT_RUN_ROOT = REPO_ROOT / ".benchmarks"
+
+
+def new_run_id() -> str:
+    return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+
+
+@dataclass(frozen=True)
+class BenchmarkConfig:
+    run_id: str
+    manifest_uri: str
+    lance_uri: str
+    mock_url: str
+    ray_address: str = "local"
+    max_minutes: float = 15.0
+    asr_actor_cpus: float = 4.0
+    asr_actor_concurrency: int = 1
+    asr_batch_size: int = 1
+    llm_concurrency: int = 8
+    llm_timeout_s: float = 15.0
+    llm_max_attempts: int = 3
+    sample_interval_s: float = 5.0
+    num_partitions: int | None = None
+    run_dir: str = ""
+
+    @property
+    def output_dir(self) -> Path:
+        return Path(self.run_dir) if self.run_dir else DEFAULT_RUN_ROOT / self.run_id
+
+    def worker_env(self) -> dict[str, str]:
+        return {
+            "BENCH_ASR_ACTOR_CPUS": str(self.asr_actor_cpus),
+            "BENCH_ASR_ACTOR_CONCURRENCY": str(self.asr_actor_concurrency),
+            "BENCH_ASR_BATCH_SIZE": str(self.asr_batch_size),
+            "BENCH_LLM_CONCURRENCY": str(self.llm_concurrency),
+            "BENCH_LLM_TIMEOUT_S": str(self.llm_timeout_s),
+            "BENCH_LLM_MAX_ATTEMPTS": str(self.llm_max_attempts),
+            "ASR_MODEL": os.getenv("ASR_MODEL", "iic/SenseVoiceSmall"),
+            "ASR_DEVICE": os.getenv("ASR_DEVICE", "cpu"),
+            "MINIO_ENDPOINT": project_config.S3_ENDPOINT,
+            "MINIO_ROOT_USER": project_config.S3_KEY,
+            "MINIO_ROOT_PASSWORD": project_config.S3_SECRET,
+            "MINIO_REGION": project_config.S3_REGION,
+        }
+
+
+def git_commit() -> str:
+    try:
+        return subprocess.check_output(
+            ["git", "rev-parse", "HEAD"], cwd=REPO_ROOT, text=True, stderr=subprocess.DEVNULL
+        ).strip()
+    except (OSError, subprocess.CalledProcessError):
+        return "unknown"
+
+
+def runtime_metadata(cfg: BenchmarkConfig, *, cluster: dict[str, Any] | None = None) -> dict[str, Any]:
+    import daft
+    import funasr
+    import ray
+
+    return {
+        "config": asdict(cfg),
+        "git_commit": git_commit(),
+        "versions": {
+            "python": sys.version.split()[0],
+            "daft": daft.__version__,
+            "ray": ray.__version__,
+            "funasr": getattr(funasr, "__version__", "unknown"),
+        },
+        "platform": platform.platform(),
+        "cluster": cluster or {},
+        "started_at": datetime.now(timezone.utc).isoformat(),
+    }
+
+
+def write_json(path: Path, value: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(value, ensure_ascii=False, indent=2, default=str) + "\n")
diff --git a/benchmark/audio/data.py b/benchmark/audio/data.py
new file mode 100644
index 0000000..bf6069a
--- /dev/null
+++ b/benchmark/audio/data.py
@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+import itertools
+import random
+import subprocess
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import soundfile as sf
+
+from .config import REPO_ROOT
+from .storage import arrow_path, ensure_bucket, s3_filesystem, upload_file
+
+
+SUPPORTED = {".wav", ".mp3", ".m4a", ".flac", ".ogg"}
+
+
+@dataclass(frozen=True)
+class GeneratedDataset:
+    manifest_uri: str
+    rows: int
+    total_bytes: int
+    total_audio_seconds: float
+
+
+def _duration(path: Path) -> float:
+    try:
+        info = sf.info(path)
+        return float(info.frames) / info.samplerate if info.samplerate else 0.0
+    except Exception:
+        probe = subprocess.run(
+            [
+                "ffprobe", "-v", "error", "-show_entries", "format=duration",
+                "-of", "default=noprint_wrappers=1:nokey=1", str(path),
+            ],
+            check=True,
+            text=True,
+            capture_output=True,
+        )
+        return float(probe.stdout.strip())
+
+
+def _transcode(source: Path, output: Path, duration_s: float) -> None:
+    codec = ["-c:a", "pcm_s16le"] if output.suffix == ".wav" else ["-c:a", "libmp3lame", "-b:a", "64k"]
+    subprocess.run(
+        [
+            "ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
+            "-stream_loop", "-1", "-i", str(source), "-t", str(duration_s),
+            "-ac", "1", "-ar", "16000", *codec, str(output),
+        ],
+        check=True,
+    )
+
+
+def _mixed_duration(rng: random.Random) -> float:
+    bucket = rng.random()
+    if bucket < 0.70:
+        return rng.uniform(15, 90)
+    if bucket < 0.95:
+        return rng.uniform(90, 300)
+    return rng.uniform(300, 1800)
+
+
+def generate_dataset(
+    *,
+    run_id: str,
+    bucket: str = "benchmark",
+    prefix: str = "audio",
+    profile: str = "smoke",
+    count: int = 4,
+    fixed_duration_s: float = 30.0,
+    source_dir: Path | None = None,
+    seed: int = 20260715,
+) -> GeneratedDataset:
+    if count < 1:
+        raise ValueError("count must be at least 1")
+    if profile not in {"smoke", "fixed", "mixed"}:
+        raise ValueError(f"unsupported profile: {profile}")
+
+    source_dir = source_dir or REPO_ROOT / "data" / "audio"
+    sources = sorted(p for p in source_dir.iterdir() if p.suffix.lower() in SUPPORTED)
+    if not sources:
+        raise ValueError(f"No supported seed audio found in {source_dir}")
+
+    fs = s3_filesystem()
+    ensure_bucket(fs, bucket)
+    rng = random.Random(seed)
+    rows: list[dict[str, object]] = []
+    total_bytes = 0
+    total_seconds = 0.0
+
+    with tempfile.TemporaryDirectory(prefix="audio-benchmark-") as tmp:
+        tmp_path = Path(tmp)
+        for index, source in zip(range(count), itertools.cycle(sources)):
+            if profile == "smoke":
+                duration_s = _duration(source)
+                generated = source
+            else:
+                duration_s = fixed_duration_s if profile == "fixed" else _mixed_duration(rng)
+                suffix = ".mp3" if rng.random() < 0.8 else ".wav"
+                generated = tmp_path / f"audio_{index:06d}{suffix}"
+                _transcode(source, generated, duration_s)
+
+            suffix = generated.suffix.lower()
+            doc_id = f"{run_id}_{index:06d}{suffix}"
+            uri = f"s3://{bucket}/{prefix.strip('/')}/{run_id}/input/{doc_id}"
+            size = upload_file(fs, generated, uri)
+            rows.append(
+                {
+                    "doc_id": doc_id,
+                    "s3_url": uri,
+                    "expected_bytes": size,
+                    "expected_duration_s": duration_s,
+                }
+            )
+            total_bytes += size
+            total_seconds += duration_s
+
+    manifest_uri = f"s3://{bucket}/{prefix.strip('/')}/{run_id}/manifest.parquet"
+    table = pa.Table.from_pylist(rows)
+    with fs.open_output_stream(arrow_path(manifest_uri)) as out:
+        pq.write_table(table, out)
+    return GeneratedDataset(manifest_uri, len(rows), total_bytes, total_seconds)
diff --git a/benchmark/audio/metrics.py b/benchmark/audio/metrics.py
new file mode 100644
index 0000000..5849a1b
--- /dev/null
+++ b/benchmark/audio/metrics.py
@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+import csv
+import os
+import socket
+import threading
+import time
+from pathlib import Path
+from typing import Any
+
+
+def _node_probe_class():
+    import ray
+
+    @ray.remote(num_cpus=0)
+    class NodeProbe:
+        def sample(self) -> dict[str, Any]:
+            import psutil
+            import ray
+
+            vm = psutil.virtual_memory()
+            ray_rss = 0
+            for proc in psutil.process_iter(["memory_info", "cmdline"]):
+                try:
+                    cmdline = " ".join(proc.info.get("cmdline") or ())
+                    if "ray" in cmdline.lower() or "daft" in cmdline.lower():
+                        ray_rss += proc.info["memory_info"].rss
+                except (psutil.AccessDenied, psutil.NoSuchProcess, TypeError):
+                    continue
+            object_store_used = None
+            try:
+                from ray._private.internal_api import get_store_used_memory
+
+                object_store_used = int(get_store_used_memory())
+            except Exception:
+                pass
+            return {
+                "timestamp": time.time(),
+                "hostname": socket.gethostname(),
+                "node_id": ray.get_runtime_context().get_node_id(),
+                "pid": os.getpid(),
+                "cpu_percent": psutil.cpu_percent(interval=None),
+                "memory_total_bytes": vm.total,
+                "memory_used_bytes": vm.used,
+                "memory_percent": vm.percent,
+                "ray_process_rss_bytes": ray_rss,
+                "object_store_used_bytes": object_store_used,
+            }
+
+    return NodeProbe
+
+
+class ResourceSampler:
+    def __init__(self, output: Path, interval_s: float = 5.0):
+        self.output = output
+        self.interval_s = interval_s
+        self._stop = threading.Event()
+        self._thread: threading.Thread | None = None
+        self._actors: list[Any] = []
+
+    def start(self) -> None:
+        import ray
+        from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
+
+        probe_cls = _node_probe_class()
+        for node in ray.nodes():
+            if not node.get("Alive"):
+                continue
+            self._actors.append(
+                probe_cls.options(
+                    scheduling_strategy=NodeAffinitySchedulingStrategy(node["NodeID"], soft=False)
+                ).remote()
+            )
+        self.output.parent.mkdir(parents=True, exist_ok=True)
+        self._thread = threading.Thread(target=self._run, name="benchmark-resource-sampler", daemon=True)
+        self._thread.start()
+
+    def stop(self) -> None:
+        import ray
+
+        self._stop.set()
+        if self._thread:
+            self._thread.join(timeout=max(10.0, self.interval_s * 2))
+        for actor in self._actors:
+            try:
+                ray.kill(actor, no_restart=True)
+            except Exception:
+                pass
+        self._actors.clear()
+
+    def _run(self) -> None:
+        import ray
+
+        fieldnames = [
+            "timestamp", "hostname", "pid", "cpu_percent", "memory_total_bytes",
+            "node_id",
+            "memory_used_bytes", "memory_percent", "ray_process_rss_bytes",
+            "object_store_used_bytes",
+        ]
+        with self.output.open("w", newline="") as out:
+            writer = csv.DictWriter(out, fieldnames=fieldnames)
+            writer.writeheader()
+            while not self._stop.is_set():
+                try:
+                    rows = ray.get([actor.sample.remote() for actor in self._actors], timeout=self.interval_s)
+                    writer.writerows(rows)
+                    out.flush()
+                except Exception:
+                    # Metrics are diagnostic and must never fail the benchmark query.
+                    pass
+                self._stop.wait(self.interval_s)
diff --git a/benchmark/audio/mock_llm.py b/benchmark/audio/mock_llm.py
new file mode 100644
index 0000000..5088032
--- /dev/null
+++ b/benchmark/audio/mock_llm.py
@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import math
+import random
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any
+
+
+PROFILES = {
+    "fast": {"median": 0.02, "p95": 0.05, "failure_rate": 0.0},
+    "standard": {"median": 2.0, "p95": 8.0, "failure_rate": 0.01},
+}
+
+
+def _rng(doc_id: str, attempt: int) -> random.Random:
+    digest = hashlib.sha256(f"{doc_id}:{attempt}".encode()).digest()
+    return random.Random(int.from_bytes(digest[:8], "big"))
+
+
+def response_plan(doc_id: str, attempt: int, profile: str) -> tuple[float, int]:
+    cfg = PROFILES[profile]
+    rng = _rng(doc_id, attempt)
+    sigma = math.log(cfg["p95"] / cfg["median"]) / 1.6448536269514722
+    delay = rng.lognormvariate(math.log(cfg["median"]), sigma)
+    # Only the first attempt is faulted, so the benchmark exercises recovery
+    # without making final row counts depend on random retry exhaustion.
+    status = 200
+    if attempt == 1 and rng.random() < cfg["failure_rate"]:
+        status = 429 if rng.random() < 0.5 else 500
+    return delay, status
+
+
+class MockHandler(BaseHTTPRequestHandler):
+    server_version = "BenchmarkMockLLM/1.0"
+
+    def do_GET(self) -> None:  # noqa: N802
+        if self.path == "/health":
+            self._json(200, {"status": "ok", "profile": self.server.profile})
+        else:
+            self._json(404, {"error": "not found"})
+
+    def do_POST(self) -> None:  # noqa: N802
+        if self.path != "/v1/chat/completions":
+            self._json(404, {"error": "not found"})
+            return
+        length = int(self.headers.get("Content-Length", "0"))
+        body = json.loads(self.rfile.read(length) or b"{}")
+        doc_id = self.headers.get("X-Benchmark-Doc-Id", "unknown")
+        attempt = int(self.headers.get("X-Benchmark-Attempt", "1"))
+        delay, status = response_plan(doc_id, attempt, self.server.profile)
+        time.sleep(delay)
+        if status != 200:
+            self._json(status, {"error": {"message": "injected benchmark failure"}})
+            return
+        prompt = json.dumps(body, ensure_ascii=False)
+        content = json.dumps(
+            {
+                "summary": f"mock summary for {doc_id}",
+                "category": "benchmark",
+                "confidence": 0.99,
+                "request_chars": len(prompt),
+                "padding": "x" * 3800,
+            },
+            ensure_ascii=False,
+        )
+        self._json(
+            200,
+            {
+                "id": f"mock-{doc_id}",
+                "object": "chat.completion",
+                "choices": [{"index": 0, "message": {"role": "assistant", "content": content}}],
+                "usage": {"prompt_tokens": max(1, len(prompt) // 4), "completion_tokens": len(content) // 4},
+            },
+        )
+
+    def log_message(self, format: str, *args: Any) -> None:
+        return
+
+    def _json(self, status: int, payload: dict[str, Any]) -> None:
+        encoded = json.dumps(payload, ensure_ascii=False).encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(encoded)))
+        self.end_headers()
+        try:
+            self.wfile.write(encoded)
+        except BrokenPipeError:
+            pass
+
+
+class MockServer(ThreadingHTTPServer):
+    daemon_threads = True
+
+    def __init__(self, address: tuple[str, int], profile: str):
+        super().__init__(address, MockHandler)
+        self.profile = profile
+
+
+def start_in_thread(host: str = "127.0.0.1", port: int = 8010, profile: str = "fast"):
+    server = MockServer((host, port), profile)
+    thread = threading.Thread(target=server.serve_forever, name="benchmark-mock-llm", daemon=True)
+    thread.start()
+    return server, thread
+
+
+def serve(host: str, port: int, profile: str) -> None:
+    server = MockServer((host, port), profile)
+    try:
+        print(f"Mock LLM listening on http://{host}:{port} ({profile})")
+        server.serve_forever()
+    finally:
+        server.server_close()
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--host", default="127.0.0.1")
+    parser.add_argument("--port", type=int, default=8010)
+    parser.add_argument("--profile", choices=sorted(PROFILES), default="fast")
+    args = parser.parse_args()
+    serve(args.host, args.port, args.profile)
diff --git a/benchmark/audio/pipeline.py b/benchmark/audio/pipeline.py
new file mode 100644
index 0000000..a05595f
--- /dev/null
+++ b/benchmark/audio/pipeline.py
@@ -0,0 +1,327 @@
+from __future__ import annotations
+
+import io
+import json
+import os
+import signal
+import time
+from contextlib import contextmanager
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+import daft
+from daft import col, lit
+from daft.functions import download, when
+
+from multimodal_toolkit import config as project_config
+from multimodal_toolkit.storage.blob import validate_blob_v2
+from multimodal_toolkit.storage.io import daft_io_config, lance_storage_options
+
+from .config import BenchmarkConfig, runtime_metadata, write_json
+from .metrics import ResourceSampler
+
+
+ASR_DTYPE = daft.DataType.struct(
+    {
+        "input_bytes": daft.DataType.int64(),
+        "duration_s": daft.DataType.float64(),
+        "transcript": daft.DataType.string(),
+        "acoustic_emotion": daft.DataType.string(),
+        "asr_ms": daft.DataType.float64(),
+        "error": daft.DataType.string(),
+    }
+)
+
+LLM_DTYPE = daft.DataType.struct(
+    {
+        "content": daft.DataType.string(),
+        "llm_ms": daft.DataType.float64(),
+        "attempts": daft.DataType.int64(),
+        "error": daft.DataType.string(),
+    }
+)
+
+
+def _asr_udf(cfg: BenchmarkConfig):
+    class AsrBenchmarkUDF:
+        def __init__(self) -> None:
+            from multimodal_toolkit.audio.asr import SenseVoiceASR
+
+            self._asr = SenseVoiceASR()
+
+        @daft.method.batch(return_dtype=ASR_DTYPE, batch_size=cfg.asr_batch_size)
+        def __call__(self, audio_bytes_col, doc_ids):
+            import soundfile as sf
+
+            rows = []
+            for audio_bytes, doc_id in zip(audio_bytes_col.to_pylist(), doc_ids.to_pylist()):
+                started = time.perf_counter()
+                duration_s = 0.0
+                if not audio_bytes:
+                    rows.append(
+                        {
+                            "input_bytes": 0,
+                            "duration_s": 0.0,
+                            "transcript": None,
+                            "acoustic_emotion": None,
+                            "asr_ms": 0.0,
+                            "error": "download_failed",
+                        }
+                    )
+                    continue
+                try:
+                    info = sf.info(io.BytesIO(audio_bytes))
+                    duration_s = float(info.frames) / info.samplerate if info.samplerate else 0.0
+                    suffix = Path(doc_id or "audio.wav").suffix or ".wav"
+                    result = self._asr.transcribe_bytes(audio_bytes, suffix)
+                    rows.append(
+                        {
+                            "input_bytes": len(audio_bytes),
+                            "duration_s": duration_s,
+                            "transcript": result["transcript"],
+                            "acoustic_emotion": result["acoustic_emotion"],
+                            "asr_ms": (time.perf_counter() - started) * 1000,
+                            "error": None,
+                        }
+                    )
+                except Exception as exc:
+                    rows.append(
+                        {
+                            "input_bytes": len(audio_bytes),
+                            "duration_s": duration_s,
+                            "transcript": None,
+                            "acoustic_emotion": None,
+                            "asr_ms": (time.perf_counter() - started) * 1000,
+                            "error": f"{type(exc).__name__}: {str(exc)[:500]}",
+                        }
+                    )
+            return rows
+
+    decorated = daft.cls(
+        cpus=cfg.asr_actor_cpus,
+        max_concurrency=cfg.asr_actor_concurrency,
+        max_retries=1,
+        name_override="SenseVoiceBenchmark",
+        ray_options={"scheduling_strategy": "SPREAD"},
+    )(AsrBenchmarkUDF)
+    return decorated()
+
+
+def _llm_udf(cfg: BenchmarkConfig):
+    class MockLLMBenchmarkUDF:
+        def __init__(self) -> None:
+            import httpx
+
+            self._client = httpx.AsyncClient(timeout=cfg.llm_timeout_s)
+
+        @daft.method(return_dtype=LLM_DTYPE)
+        async def __call__(self, transcript: str | None, doc_id: str, asr_error: str | None):
+            import asyncio
+            import httpx
+
+            if asr_error:
+                return {"content": None, "llm_ms": 0.0, "attempts": 0, "error": "skipped_after_asr"}
+            started = time.perf_counter()
+            error = None
+            for attempt in range(1, cfg.llm_max_attempts + 1):
+                try:
+                    response = await self._client.post(
+                        f"{cfg.mock_url.rstrip('/')}/v1/chat/completions",
+                        headers={
+                            "X-Benchmark-Doc-Id": doc_id,
+                            "X-Benchmark-Attempt": str(attempt),
+                        },
+                        json={
+                            "model": "benchmark-mock",
+                            "messages": [{"role": "user", "content": transcript or ""}],
+                            "temperature": 0,
+                        },
+                    )
+                    response.raise_for_status()
+                    content = response.json()["choices"][0]["message"]["content"]
+                    return {
+                        "content": content,
+                        "llm_ms": (time.perf_counter() - started) * 1000,
+                        "attempts": attempt,
+                        "error": None,
+                    }
+                except (httpx.HTTPError, KeyError, ValueError) as exc:
+                    error = f"{type(exc).__name__}: {str(exc)[:500]}"
+                    if attempt < cfg.llm_max_attempts:
+                        await asyncio.sleep(0.5 * (2 ** (attempt - 1)))
+            return {
+                "content": None,
+                "llm_ms": (time.perf_counter() - started) * 1000,
+                "attempts": cfg.llm_max_attempts,
+                "error": error,
+            }
+
+    decorated = daft.cls(
+        cpus=0.1,
+        max_concurrency=cfg.llm_concurrency,
+        name_override="MockLLMBenchmark",
+    )(MockLLMBenchmarkUDF)
+    return decorated()
+
+
+def _configure_daft(cfg: BenchmarkConfig) -> dict[str, Any]:
+    import ray
+
+    if not ray.is_initialized():
+        ray.init(
+            address=cfg.ray_address,
+            ignore_reinit_error=True,
+            log_to_driver=False,
+            runtime_env={"env_vars": cfg.worker_env()},
+        )
+    daft.set_runner_ray(address=cfg.ray_address, noop_if_initialized=True)
+    daft.set_execution_config(
+        default_morsel_size=project_config.DAFT_DEFAULT_MORSEL_SIZE,
+        enable_dynamic_batching=project_config.DAFT_ENABLE_DYNAMIC_BATCHING,
+        enable_scan_task_split_and_merge=project_config.DAFT_ENABLE_SCAN_TASK_SPLIT_AND_MERGE,
+        scan_tasks_min_size_bytes=project_config.DAFT_SCAN_TASKS_MIN_SIZE_BYTES,
+        scan_tasks_max_size_bytes=project_config.DAFT_SCAN_TASKS_MAX_SIZE_BYTES,
+        max_sources_per_scan_task=project_config.DAFT_MAX_SOURCES_PER_SCAN_TASK,
+        scantask_max_parallel=project_config.DAFT_SCANTASK_MAX_PARALLEL,
+        maintain_order=project_config.DAFT_MAINTAIN_ORDER,
+        actor_udf_ready_timeout=project_config.DAFT_ACTOR_UDF_READY_TIMEOUT,
+    )
+    return {
+        "resources": ray.cluster_resources(),
+        "nodes": [
+            {"node_id": n["NodeID"], "address": n["NodeManagerAddress"], "alive": n["Alive"]}
+            for n in ray.nodes()
+        ],
+    }
+
+
+@contextmanager
+def _deadline(minutes: float):
+    if minutes <= 0 or not hasattr(signal, "setitimer"):
+        yield
+        return
+
+    def expired(_signum, _frame):
+        raise TimeoutError(f"benchmark exceeded {minutes} minute limit")
+
+    previous = signal.signal(signal.SIGALRM, expired)
+    signal.setitimer(signal.ITIMER_REAL, minutes * 60)
+    try:
+        yield
+    finally:
+        signal.setitimer(signal.ITIMER_REAL, 0)
+        signal.signal(signal.SIGALRM, previous)
+
+
+def _build_dataframe(cfg: BenchmarkConfig):
+    import ray
+
+    io_config = daft_io_config()
+    df = daft.read_parquet(cfg.manifest_uri, io_config=io_config).select("doc_id", "s3_url")
+    num_partitions = cfg.num_partitions
+    if num_partitions is None:
+        num_partitions = max(1, 2 * int(ray.cluster_resources().get("CPU", 1)))
+    if num_partitions > 1:
+        df = df.into_partitions(num_partitions)
+    df = df.with_column("audio_blob", download(col("s3_url"), on_error="null", io_config=io_config))
+    asr = _asr_udf(cfg)
+    df = df.with_column("asr", asr(col("audio_blob"), col("doc_id")))
+    for name in ("input_bytes", "duration_s", "transcript", "acoustic_emotion", "asr_ms"):
+        df = df.with_column(name, col("asr")[name])
+    df = df.with_column("asr_error", col("asr")["error"])
+
+    llm = _llm_udf(cfg)
+    df = df.with_column("llm", llm(col("transcript"), col("doc_id"), col("asr_error")))
+    df = df.with_column("llm_content", col("llm")["content"])
+    df = df.with_column("llm_ms", col("llm")["llm_ms"])
+    df = df.with_column("llm_attempts", col("llm")["attempts"])
+    df = df.with_column("llm_error", col("llm")["error"])
+    df = df.with_column(
+        "status",
+        when(col("audio_blob").is_null(), "download_failed")
+        .when(~col("asr_error").is_null(), "asr_failed")
+        .when(~col("llm_error").is_null(), "llm_failed")
+        .otherwise("ok"),
+    )
+    df = df.with_column(
+        "error_code",
+        when(col("audio_blob").is_null(), "download_failed")
+        .when(~col("asr_error").is_null(), "asr_error")
+        .when(~col("llm_error").is_null(), "llm_error"),
+    )
+    df = df.with_column("run_id", lit(cfg.run_id))
+    df = df.with_column(
+        "processed_at",
+        lit(datetime.now(timezone.utc)).cast(daft.DataType.timestamp("us", "UTC")),
+    )
+    return df.select(
+        "run_id", "doc_id", "s3_url", "input_bytes", "duration_s", "transcript",
+        "acoustic_emotion", "llm_content", "asr_ms", "llm_ms", "llm_attempts",
+        "status", "error_code", "asr_error", "llm_error", "audio_blob", "processed_at",
+    )
+
+
+def run_pipeline(cfg: BenchmarkConfig) -> dict[str, Any]:
+    import lance
+    from daft.subscribers.event_log import disable_event_log, enable_event_log
+
+    run_dir = cfg.output_dir
+    run_dir.mkdir(parents=True, exist_ok=True)
+    cluster = _configure_daft(cfg)
+    metadata = runtime_metadata(cfg, cluster=cluster)
+    write_json(run_dir / "metadata.json", metadata)
+    enable_event_log(run_dir / "daft-events")
+
+    df = _build_dataframe(cfg)
+    with (run_dir / "plan.txt").open("w") as plan:
+        df.explain(show_all=True, file=plan)
+
+    sampler = ResourceSampler(run_dir / "resources.csv", cfg.sample_interval_s)
+    sampler.start()
+    started = time.perf_counter()
+    state = "success"
+    error = None
+    try:
+        with _deadline(cfg.max_minutes):
+            df.write_lance(
+                cfg.lance_uri,
+                mode="overwrite",
+                io_config=daft_io_config(),
+                blob_columns=["audio_blob"],
+                max_rows_per_file=project_config.LANCE_MAX_ROWS_PER_FILE,
+                max_bytes_per_file=project_config.LANCE_MAX_BYTES_PER_FILE,
+            )
+        validate_blob_v2(cfg.lance_uri, "audio_blob")
+    except Exception as exc:
+        state = "timeout" if isinstance(exc, TimeoutError) else "failed"
+        error = f"{type(exc).__name__}: {exc}"
+        raise
+    finally:
+        sampler.stop()
+        disable_event_log()
+        elapsed = time.perf_counter() - started
+        summary: dict[str, Any] = {
+            "run_id": cfg.run_id,
+            "state": state,
+            "error": error,
+            "elapsed_s": elapsed,
+            "lance_uri": cfg.lance_uri,
+        }
+        if state == "success":
+            ds = lance.dataset(cfg.lance_uri, storage_options=lance_storage_options(cfg.lance_uri))
+            table = ds.to_table(columns=["doc_id", "status", "input_bytes", "duration_s"])
+            summary.update(
+                {
+                    "rows": table.num_rows,
+                    "unique_doc_ids": len(set(table.column("doc_id").to_pylist())),
+                    "status_counts": {
+                        status: table.column("status").to_pylist().count(status)
+                        for status in sorted(set(table.column("status").to_pylist()))
+                    },
+                    "input_bytes": sum(x or 0 for x in table.column("input_bytes").to_pylist()),
+                    "audio_seconds": sum(x or 0 for x in table.column("duration_s").to_pylist()),
+                }
+            )
+        write_json(run_dir / "summary.json", summary)
+    return summary
diff --git a/benchmark/audio/report.py b/benchmark/audio/report.py
new file mode 100644
index 0000000..49e54dc
--- /dev/null
+++ b/benchmark/audio/report.py
@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+import csv
+import json
+import math
+from collections import defaultdict
+from pathlib import Path
+from typing import Iterable
+
+import numpy as np
+
+from multimodal_toolkit.storage.io import lance_storage_options
+
+from .config import write_json
+
+
+def _percentile(values: Iterable[float], q: float) -> float | None:
+    clean = [float(v) for v in values if v is not None and not math.isnan(float(v))]
+    return float(np.percentile(clean, q)) if clean else None
+
+
+def _memory_slope_bytes_per_hour(rows: list[dict[str, str]]) -> float | None:
+    if len(rows) < 2:
+        return None
+    x = np.asarray([float(row["timestamp"]) for row in rows], dtype=np.float64)
+    y = np.asarray([float(row["memory_used_bytes"]) for row in rows], dtype=np.float64)
+    x -= x[0]
+    if x[-1] <= 0:
+        return None
+    return float(np.polyfit(x, y, 1)[0] * 3600)
+
+
+def build_report(run_dir: Path) -> dict:
+    import lance
+
+    summary = json.loads((run_dir / "summary.json").read_text())
+    metadata = json.loads((run_dir / "metadata.json").read_text())
+    report = {"run_id": summary["run_id"], "state": summary["state"], "summary": summary}
+    if summary["state"] == "success":
+        ds = lance.dataset(summary["lance_uri"], storage_options=lance_storage_options(summary["lance_uri"]))
+        table = ds.to_table(
+            columns=["asr_ms", "llm_ms", "input_bytes", "duration_s", "llm_attempts", "status"]
+        )
+        elapsed = summary["elapsed_s"]
+        report["throughput"] = {
+            "files_per_s": summary["rows"] / elapsed if elapsed else None,
+            "mb_per_s": summary["input_bytes"] / elapsed / 1_000_000 if elapsed else None,
+            "audio_seconds_per_wall_second": summary["audio_seconds"] / elapsed if elapsed else None,
+        }
+        asr_values = [x for x in table.column("asr_ms").to_pylist() if x is not None and x > 0]
+        attempts = table.column("llm_attempts").to_pylist()
+        llm_values = [
+            value
+            for value, attempt in zip(table.column("llm_ms").to_pylist(), attempts)
+            if value is not None and attempt is not None and attempt > 0
+        ]
+        report["latency_ms"] = {
+            "asr_ms": {"p50": _percentile(asr_values, 50), "p95": _percentile(asr_values, 95), "p99": _percentile(asr_values, 99)},
+            "llm_ms": {"p50": _percentile(llm_values, 50), "p95": _percentile(llm_values, 95), "p99": _percentile(llm_values, 99)},
+        }
+        report["llm_retry_rows"] = sum(x > 1 for x in attempts if x)
+
+    resource_rows = []
+    resource_path = run_dir / "resources.csv"
+    if resource_path.exists():
+        with resource_path.open(newline="") as inp:
+            resource_rows = list(csv.DictReader(inp))
+    if resource_rows:
+        used = [float(row["memory_used_bytes"]) for row in resource_rows]
+        rss = [float(row["ray_process_rss_bytes"]) for row in resource_rows]
+        report["resources"] = {
+            "samples": len(resource_rows),
+            "node_memory_used_peak_bytes": max(used),
+            "node_memory_used_p95_bytes": _percentile(used, 95),
+            "ray_process_rss_peak_bytes": max(rss),
+            "ray_process_rss_p95_bytes": _percentile(rss, 95),
+            "cpu_percent_p95": _percentile(
+                [float(row["cpu_percent"]) for row in resource_rows if row.get("cpu_percent")], 95
+            ),
+            "object_store_used_peak_bytes": max(
+                [float(row["object_store_used_bytes"]) for row in resource_rows if row.get("object_store_used_bytes")]
+                or [0.0]
+            ),
+        }
+        by_host: dict[str, list[dict[str, str]]] = defaultdict(list)
+        for row in resource_rows:
+            by_host[row.get("node_id") or row.get("hostname", "unknown")].append(row)
+        node_reports = {}
+        for hostname, rows in by_host.items():
+            node_used = [float(row["memory_used_bytes"]) for row in rows]
+            total = max(float(row["memory_total_bytes"]) for row in rows)
+            peak = max(node_used)
+            p95 = _percentile(node_used, 95)
+            slope = _memory_slope_bytes_per_hour(rows)
+            node_reports[hostname] = {
+                "memory_total_bytes": total,
+                "memory_used_peak_bytes": peak,
+                "memory_used_p95_bytes": p95,
+                "memory_growth_bytes_per_hour": slope,
+                "passes_memory_slo": bool(
+                    p95 is not None
+                    and p95 <= total * 0.80
+                    and peak <= total * 0.90
+                    and (slope is None or slope < 500_000_000)
+                ),
+            }
+        report["resources"]["nodes"] = node_reports
+        report["resources"]["passes_memory_slo"] = all(
+            node["passes_memory_slo"] for node in node_reports.values()
+        )
+
+    report["versions"] = metadata.get("versions", {})
+    write_json(run_dir / "report.json", report)
+    _write_markdown(run_dir / "report.md", report)
+    return report
+
+
+def _write_markdown(path: Path, report: dict) -> None:
+    summary = report["summary"]
+    lines = [
+        f"# Audio benchmark {report['run_id']}",
+        "",
+        f"- State: `{report['state']}`",
+        f"- Elapsed: {summary.get('elapsed_s', 0):.2f}s",
+        f"- Rows: {summary.get('rows', 0)}",
+        f"- Status: `{json.dumps(summary.get('status_counts', {}), ensure_ascii=False)}`",
+    ]
+    throughput = report.get("throughput")
+    if throughput:
+        lines.extend(
+            [
+                "",
+                "## Throughput",
+                "",
+                f"- Files/s: {throughput['files_per_s']:.4f}",
+                f"- MB/s: {throughput['mb_per_s']:.4f}",
+                f"- Audio seconds / wall second: {throughput['audio_seconds_per_wall_second']:.4f}",
+            ]
+        )
+    latency = report.get("latency_ms")
+    if latency:
+        lines.extend(["", "## Latency", ""])
+        for stage, values in latency.items():
+            rendered = ", ".join(
+                f"{name.upper()}={value:.2f}ms" for name, value in values.items() if value is not None
+            )
+            lines.append(f"- {stage}: {rendered or 'no executed rows'}")
+    resources = report.get("resources")
+    if resources:
+        lines.extend(
+            [
+                "",
+                "## Resources",
+                "",
+                f"- Ray process RSS peak: {resources['ray_process_rss_peak_bytes'] / 1_000_000_000:.2f}GB",
+                f"- Node memory SLO: `{'PASS' if resources['passes_memory_slo'] else 'FAIL'}`",
+                f"- CPU P95: {resources.get('cpu_percent_p95') or 0:.2f}%",
+                f"- Object store peak: {resources.get('object_store_used_peak_bytes') or 0:.0f} bytes",
+            ]
+        )
+    path.write_text("\n".join(lines) + "\n")
diff --git a/benchmark/audio/storage.py b/benchmark/audio/storage.py
new file mode 100644
index 0000000..13ea3c1
--- /dev/null
+++ b/benchmark/audio/storage.py
@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from pathlib import Path
+from urllib.parse import urlparse
+
+import pyarrow.fs as pafs
+
+from multimodal_toolkit import config
+
+
+def s3_filesystem() -> pafs.S3FileSystem:
+    parsed = urlparse(config.S3_ENDPOINT)
+    endpoint = parsed.netloc or parsed.path
+    return pafs.S3FileSystem(
+        access_key=config.S3_KEY,
+        secret_key=config.S3_SECRET,
+        endpoint_override=endpoint,
+        scheme="https" if config.S3_USE_SSL else "http",
+        region=config.S3_REGION,
+        allow_bucket_creation=True,
+    )
+
+
+def split_s3_uri(uri: str) -> tuple[str, str]:
+    parsed = urlparse(uri)
+    if parsed.scheme != "s3" or not parsed.netloc:
+        raise ValueError(f"Expected s3:// URI, got {uri!r}")
+    return parsed.netloc, parsed.path.lstrip("/")
+
+
+def arrow_path(uri: str) -> str:
+    bucket, key = split_s3_uri(uri)
+    return f"{bucket}/{key}" if key else bucket
+
+
+def ensure_bucket(fs: pafs.S3FileSystem, bucket: str) -> None:
+    if fs.get_file_info(bucket).type == pafs.FileType.NotFound:
+        fs.create_dir(bucket)
+
+
+def upload_file(fs: pafs.S3FileSystem, source: Path, uri: str) -> int:
+    path = arrow_path(uri)
+    with source.open("rb") as src, fs.open_output_stream(path) as dst:
+        while chunk := src.read(8 * 1024 * 1024):
+            dst.write(chunk)
+    return source.stat().st_size
+
+
+def preflight(bucket: str) -> dict[str, object]:
+    """Validate the configured MinIO without modifying existing user objects."""
+    import uuid
+
+    fs = s3_filesystem()
+    ensure_bucket(fs, bucket)
+    key = f"{bucket}/_benchmark_preflight/{uuid.uuid4().hex}.txt"
+    payload = b"multimodal-toolkit benchmark preflight"
+    with fs.open_output_stream(key) as out:
+        out.write(payload)
+    with fs.open_input_file(key) as inp:
+        actual = inp.read()
+    fs.delete_file(key)
+    if actual != payload:
+        raise RuntimeError("MinIO preflight read did not match the written payload")
+    return {"endpoint": config.S3_ENDPOINT, "bucket": bucket, "read_write": True}
diff --git a/benchmark/audio/submit.py b/benchmark/audio/submit.py
new file mode 100644
index 0000000..3fc9910
--- /dev/null
+++ b/benchmark/audio/submit.py
@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import asyncio
+import shlex
+from dataclasses import asdict
+
+from ray.job_submission import JobSubmissionClient
+
+from .config import BenchmarkConfig, REPO_ROOT
+
+
+def build_entrypoint_args(cfg: BenchmarkConfig) -> list[str]:
+    args = [
+        "python", "-m", "benchmark.audio", "run",
+        "--run-id", cfg.run_id,
+        "--manifest", cfg.manifest_uri,
+        "--lance-uri", cfg.lance_uri,
+        "--mock-url", cfg.mock_url,
+        "--ray-address", cfg.ray_address,
+        "--max-minutes", str(cfg.max_minutes),
+        "--asr-actor-cpus", str(cfg.asr_actor_cpus),
+        "--asr-actor-concurrency", str(cfg.asr_actor_concurrency),
+        "--asr-batch-size", str(cfg.asr_batch_size),
+        "--llm-concurrency", str(cfg.llm_concurrency),
+        "--llm-timeout-s", str(cfg.llm_timeout_s),
+        "--llm-max-attempts", str(cfg.llm_max_attempts),
+        "--sample-interval-s", str(cfg.sample_interval_s),
+    ]
+    if cfg.num_partitions is not None:
+        args.extend(["--num-partitions", str(cfg.num_partitions)])
+    if cfg.run_dir:
+        args.extend(["--run-dir", cfg.run_dir])
+    return args
+
+
+def submit_job(cfg: BenchmarkConfig, dashboard_address: str, *, wait: bool = False) -> str:
+    args = build_entrypoint_args(cfg)
+    client = JobSubmissionClient(dashboard_address)
+    job_id = client.submit_job(
+        entrypoint=shlex.join(args),
+        runtime_env={
+            "working_dir": str(REPO_ROOT),
+            "excludes": [".git", ".venv", ".benchmarks", "data", ".claude"],
+            "env_vars": cfg.worker_env(),
+        },
+        metadata={"benchmark_run_id": cfg.run_id, "config": str(asdict(cfg))[:1024]},
+    )
+    if wait:
+        async def tail() -> None:
+            async for line in client.tail_job_logs(job_id):
+                print(line, end="")
+
+        asyncio.run(tail())
+    return job_id
diff --git a/docs/audio-benchmark.md b/docs/audio-benchmark.md
new file mode 100644
index 0000000..e33f99f
--- /dev/null
+++ b/docs/audio-benchmark.md
@@ -0,0 +1,258 @@
+# Daft 音频稳定性与性能 Benchmark
+
+本文档说明如何使用 `benchmark/audio` 工具验证以下链路，并从本地小数据扩展到多节点 Ray 集群：
+
+```text
+MinIO/S3 → Daft download → SenseVoice ASR → HTTP Mock LLM → Lance Blob v2
+```
+
+Benchmark 不调用真实 LLM，也不会启动或停止 MinIO。每次运行使用独立的 `run_id` 和 S3 prefix，不覆盖其他运行的数据。
+
+## 1. 命令与产物
+
+统一入口为：
+
+```bash
+uv run python -m benchmark.audio --help
+```
+
+| 命令 | 用途 |
+|---|---|
+| `generate` | 构造音频对象并写入 MinIO，同时生成 Parquet manifest |
+| `serve-mock` | 启动 OpenAI-compatible Mock LLM |
+| `run` | 直接连接 local/remote Ray 执行 Daft pipeline |
+| `submit` | 通过 Ray Jobs API 提交同一个 pipeline |
+| `report` | 从已有运行产物重建报告 |
+| `local-smoke` | 一键完成 MinIO 预检、造数、本地 Ray 执行和校验 |
+
+每次运行在 `.benchmarks/<run_id>/` 或 `--run-dir` 指定目录生成：
+
+| 文件 | 内容 |
+|---|---|
+| `metadata.json` | Git commit、依赖版本、运行参数和集群节点 |
+| `summary.json` | 状态、总耗时、行数、字节数和音频时长 |
+| `resources.csv` | 每 5 秒采集的节点 CPU、内存、Ray RSS 和 object store |
+| `report.json` | 机器可读的吞吐、延迟、重试和内存 SLO |
+| `report.md` | 适合人工阅读的摘要 |
+| `plan.txt` | Daft 优化前后的执行计划 |
+| `daft-events/` | Daft query、operator 和 process event log |
+
+结果表保存在 `--lance-uri`，包含 ASR/LLM 结果、逐行耗时、状态、错误信息和原始 `audio_blob`。写入后会强制校验 Lance Blob v2。
+
+## 2. 本地快速验证
+
+准备条件：
+
+- `.env` 中的 `MINIO_ENDPOINT` 和凭证可用；
+- MinIO 当前已启动；benchmark 不负责启动服务；
+- `data/audio/` 至少有一个支持的种子文件；
+- SenseVoice 和 FSMN VAD 模型已缓存或当前环境可以下载；
+- 安装了 `ffmpeg` 和 `libsndfile`。
+
+执行真实 ASR 的四行 smoke：
+
+```bash
+uv sync --frozen
+uv run python -m benchmark.audio local-smoke --count 4
+```
+
+`local-smoke` 会用随机端口启动 fast Mock LLM，只清理该 Mock 进程。它会向当前 MinIO 写入：
+
+```text
+s3://benchmark/audio/<run_id>/
+├── input/
+├── manifest.parquet
+└── output.lance/
+```
+
+如需使用其他 bucket、prefix 或种子目录：
+
+```bash
+uv run python -m benchmark.audio local-smoke \
+  --bucket perf-test \
+  --prefix daft/audio \
+  --source-dir /path/to/audio \
+  --count 8
+```
+
+## 3. 构造测试数据
+
+三种 profile 的定位如下：
+
+| Profile | 行为 | 适用场景 |
+|---|---|---|
+| `smoke` | 直接复用种子文件内容，但每行使用唯一对象 key | 功能验证、依赖验证 |
+| `fixed` | 循环/裁剪种子并转为 16kHz 单声道 MP3/WAV | 参数间公平对比 |
+| `mixed` | 70% 15–90s、25% 90–300s、5% 300–1800s；80% MP3、20% WAV | 调度、长尾和内存稳定性 |
+
+固定 60 秒、1000 条：
+
+```bash
+RUN_ID=audio-fixed-$(date +%Y%m%d-%H%M%S)
+uv run python -m benchmark.audio generate \
+  --run-id "$RUN_ID" \
+  --profile fixed \
+  --duration-s 60 \
+  --count 1000 \
+  --bucket benchmark \
+  --prefix audio
+```
+
+混合时长数据：
+
+```bash
+uv run python -m benchmark.audio generate \
+  --run-id "$RUN_ID" \
+  --profile mixed \
+  --count 10000 \
+  --seed 20260715
+```
+
+同一个 seed、种子目录和参数会得到相同的时长/编码选择。所有音频会使用唯一 S3 key，避免因重复 URL 缓存而低估 I/O 成本。造数时间不计入 benchmark 运行时间。
+
+## 4. 扩展到 4-worker Ray 集群
+
+基准拓扑：一个不承担 ASR 的 Ray head，加 4 个 worker；每个 worker 为 8 CPU、24GB 内存。启动 head 时应将可调度 CPU 设为 0，建议每个 worker 预留约 6GB Ray object store，并配置本地 SSD spill 目录。
+
+```bash
+ray start --head --num-cpus=0
+```
+
+### 4.1 集群前置检查
+
+所有节点必须使用相同的 Python/系统环境，并具备：
+
+- 本项目及 `uv.lock` 中的依赖；
+- `ffmpeg`、`libsndfile`；
+- 相同的 SenseVoice/VAD 模型缓存；
+- 可写的临时目录和 Ray spill 目录；
+- 到 MinIO 和 Mock LLM 的网络路由。
+
+多节点时不能使用 `127.0.0.1` 作为 MinIO 或 Mock 地址。将 `.env` 中的端点改为所有 worker 可访问的地址，并配置 `NO_PROXY/no_proxy`：
+
+```bash
+MINIO_ENDPOINT=http://minio.internal:9000
+MINIO_ROOT_USER=...
+MINIO_ROOT_PASSWORD=...
+MINIO_REGION=us-east-1
+NO_PROXY=minio.internal,mock-llm.internal,127.0.0.1,localhost
+no_proxy=minio.internal,mock-llm.internal,127.0.0.1,localhost
+ASR_DEVICE=cpu
+```
+
+逐节点验证：
+
+```bash
+curl -f http://minio.internal:9000/minio/health/live
+curl -f http://mock-llm.internal:8010/health
+ray status
+```
+
+`submit` 会打包项目源码并传递 benchmark 所需环境变量，但不会安装系统库或下载模型。生产式压测应使用预构建且一致的 worker 镜像。
+
+### 4.2 启动 Mock LLM
+
+在所有 worker 可访问的主机运行：
+
+```bash
+uv run python -m benchmark.audio serve-mock \
+  --host 0.0.0.0 \
+  --port 8010 \
+  --profile standard
+```
+
+`standard` profile 的中位延迟约 2 秒、P95 约 8 秒、响应约 4KB，并确定性注入少量首请求 429/500；后续 retry 可恢复。Mock 自身也可能成为瓶颈，应独立观察其 CPU、连接数和错误率。
+
+### 4.3 提交第一轮安全基线
+
+4×8 CPU 集群先使用每个 ASR actor 4 CPU、最多 4 个 actor、batch size 1。ASR actor 使用 Ray 的 `SPREAD` 调度策略，但这只是调度倾向，不是严格的“一节点一个 actor”约束；必须从 Ray Dashboard 和逐节点 RSS 确认实际放置。`--num-partitions 64` 对应总 CPU 的两倍，确保小 manifest 能在 download/ASR 前充分拆分：
+
+```bash
+uv run python -m benchmark.audio submit \
+  --dashboard-address http://ray-head.internal:8265 \
+  --run-id "$RUN_ID" \
+  --manifest "s3://benchmark/audio/$RUN_ID/manifest.parquet" \
+  --lance-uri "s3://benchmark/audio/$RUN_ID/output.lance" \
+  --mock-url http://mock-llm.internal:8010 \
+  --ray-address auto \
+  --max-minutes 60 \
+  --num-partitions 64 \
+  --asr-actor-cpus 4 \
+  --asr-actor-concurrency 4 \
+  --asr-batch-size 1 \
+  --llm-concurrency 32 \
+  --wait
+```
+
+不传 `--num-partitions` 时会自动使用集群 CPU 数的两倍。显式参数更适合可重复的横向比较。
+
+`--asr-actor-concurrency` 控制可并行的同步 ASR actor 数；`--llm-concurrency` 是单个 Daft 异步 LLM actor 的最大协程数，不代表全集群 HTTP 请求总数。总并发必须结合 Ray actor 数量和 Mock 端连接数实测。
+
+Ray Jobs 的默认本地报告位于 job 工作目录，可能随 runtime environment 清理。集群验收应把 `--run-dir` 指向 Ray head 可写的持久卷，例如：
+
+```bash
+--run-dir "/mnt/benchmark/$RUN_ID"
+```
+
+结果 Lance 始终保存在 MinIO；报告目录只保存指标和诊断信息。
+
+## 5. 一小时迭代与调参顺序
+
+ASR 工作量更接近“音频时长”而不是文件字节数。不要直接假设固定 GB 数能在一小时完成。
+
+1. 用 100–500 条代表性数据跑 10–15 分钟校准。
+2. 从 `report.json` 读取 `audio_seconds_per_wall_second = R`。
+3. 正式批次目标音频秒数取 `R × 2700`，即约 45 分钟有效负载，给冷启动和 Lance commit 留 15 分钟。
+4. `mixed` profile 的理论平均时长约 138 秒，可用 `目标音频秒数 ÷ 138` 粗估 `--count`。
+5. 同一配置至少重复两次；吞吐差异超过 10% 时先排查 MinIO、Mock 和节点噪声。
+
+建议一次只改变一个参数：
+
+1. 固定 batch 1，比较 ASR actor concurrency 4 与 8；只有内存 SLO 通过时才提高。
+2. 固定 actor 数，比较 ASR batch size 1 与 2。
+3. 比较 `DAFT_DEFAULT_MORSEL_SIZE=8/16/32`；大音频优先从 8 开始。
+4. 最后调整 `--llm-concurrency`、S3 连接数和 partition 数。
+
+更多 Daft 参数含义见 [Daft tuning](daft-tuning.md)。
+
+当前 `generate` 使用单进程串行 ffmpeg 和串行上传，适合提前准备小到中等规模数据，不适合临时生成 TB 级数据。大规模造数应在正式计时前完成；需要 TB 级数据时可沿用同一 manifest schema，另行实现并行生成器。
+
+## 6. 报告与验收
+
+关键指标：
+
+- `files_per_s`：小文件调度能力；
+- `mb_per_s`：S3/Lance 字节吞吐；
+- `audio_seconds_per_wall_second`：ASR 核心吞吐；
+- `asr_ms`、`llm_ms` P50/P95/P99：只统计实际执行过对应阶段的行，跳过的 0ms 不进入分位数；
+- `ray_process_rss_*`：Ray/Daft 相关进程内存；
+- `resources.nodes.*`：逐节点总内存、增长斜率和 SLO；
+- `status_counts`、`llm_retry_rows`：正确性与恢复情况。
+
+默认内存 SLO：逐节点 P95 不超过总内存 80%，峰值不超过 90%，增长斜率小于 0.5GB/小时。节点总内存包含同机其他进程，容器中还可能反映宿主机而非 cgroup；Ray RSS 也是按进程命令行近似归因。因此这些值是诊断性指标，正式结果应来自隔离 worker，并结合 Ray Dashboard/Prometheus 复核。当前工具不采集磁盘 spill 字节数。
+
+一次运行至少满足：
+
+- query 在 `--max-minutes` 内完成；该参数是 driver 侧 best-effort deadline，集群硬时限还应使用外部 watchdog 或 `ray job stop`；
+- 无 OOM、无限等待或未恢复 Ray task failure；
+- manifest 行数与 Lance 行数一致，`doc_id` 唯一；
+- 每行均为 `ok` 或明确的失败状态；
+- `audio_blob` 通过 Lance Blob v2 校验；
+- 两次相同配置吞吐变异系数不超过 10%。
+
+## 7. 扩展开发
+
+主要模块：
+
+| 模块 | 扩展点 |
+|---|---|
+| `data.py` | 新的数据分布、编码组合、坏文件和 TTS 生成器 |
+| `mock_llm.py` | 新延迟模型、限流、超时和响应大小 |
+| `pipeline.py` | 新 Daft UDF、资源声明、输出列和失败语义 |
+| `metrics.py` | Prometheus、GPU、磁盘 spill 和网络指标 |
+| `report.py` | 新 SLO、横向对比和趋势分析 |
+| `submit.py` | Kubernetes、容器镜像和 Ray runtime environment |
+
+新增 profile 或指标时，应同时补充 `tests/benchmark/`，保持默认 pytest 不依赖 MinIO/Ray/模型；真实端到端测试继续使用 `benchmark_e2e` marker。
+
+当前限制：pipeline 只接受 Parquet manifest，只覆盖 CPU SenseVoice 和音频；尚未在真实 4-worker 集群验收，不包含 GPU、视频/图片、真实 LLM 或主动 worker 故障注入。每轮必须使用唯一 `run_id`/Lance URI，因为写入模式为 `overwrite`。
diff --git a/pyproject.toml b/pyproject.toml
index 4c1a94e..44316a1 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -8,6 +8,7 @@ dependencies = [
   "daft-lance>=0.4.0",
   "funasr",
   "getdaft",
+  "httpx",
   "insightface",
   "lance-ray",
   "modelscope",
@@ -16,6 +17,7 @@ dependencies = [
   "opencv-python-headless",
   "openai",
   "pillow",
+  "psutil",
   "pyarrow",
   "pylance<8.0.0",
   "python-dotenv",
@@ -29,8 +31,9 @@ dependencies = [
 [tool.pytest.ini_options]
 markers = [
   "ray: hard dependency on a local Ray cluster; excluded by default, run with `pytest -m ray`",
+  "benchmark_e2e: real MinIO + local Ray + SenseVoice benchmark; excluded by default",
 ]
-addopts = "-m 'not ray'"
+addopts = "-m 'not ray and not benchmark_e2e'"
 
 [build-system]
 requires = ["hatchling"]
diff --git a/tests/benchmark/__init__.py b/tests/benchmark/__init__.py
new file mode 100644
index 0000000..ce00eba
--- /dev/null
+++ b/tests/benchmark/__init__.py
@@ -0,0 +1 @@
+"""Tests for the standalone benchmark tooling."""
diff --git a/tests/benchmark/test_config.py b/tests/benchmark/test_config.py
new file mode 100644
index 0000000..65498ac
--- /dev/null
+++ b/tests/benchmark/test_config.py
@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from benchmark.audio.cli import build_parser
+from benchmark.audio.config import BenchmarkConfig
+
+
+def test_worker_env_contains_resource_controls():
+    cfg = BenchmarkConfig(
+        run_id="run",
+        manifest_uri="s3://b/m.parquet",
+        lance_uri="s3://b/o.lance",
+        mock_url="http://mock:8010",
+        asr_actor_cpus=2.0,
+        asr_batch_size=3,
+        llm_concurrency=7,
+    )
+    env = cfg.worker_env()
+    assert env["BENCH_ASR_ACTOR_CPUS"] == "2.0"
+    assert env["BENCH_ASR_BATCH_SIZE"] == "3"
+    assert env["BENCH_LLM_CONCURRENCY"] == "7"
+
+
+def test_cli_parses_run_controls():
+    args = build_parser().parse_args(
+        [
+            "run", "--manifest", "s3://b/m.parquet", "--lance-uri", "s3://b/o.lance",
+            "--mock-url", "http://mock", "--max-minutes", "60", "--asr-batch-size", "2",
+            "--num-partitions", "64",
+        ]
+    )
+    assert args.command == "run"
+    assert args.max_minutes == 60
+    assert args.asr_batch_size == 2
+    assert args.num_partitions == 64
diff --git a/tests/benchmark/test_data.py b/tests/benchmark/test_data.py
new file mode 100644
index 0000000..3641a17
--- /dev/null
+++ b/tests/benchmark/test_data.py
@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+import random
+
+from benchmark.audio.data import _mixed_duration
+
+
+def test_mixed_duration_stays_in_supported_range():
+    rng = random.Random(42)
+    values = [_mixed_duration(rng) for _ in range(1000)]
+    assert min(values) >= 15
+    assert max(values) <= 1800
+    assert any(value <= 90 for value in values)
+    assert any(90 < value <= 300 for value in values)
+    assert any(value > 300 for value in values)
diff --git a/tests/benchmark/test_e2e.py b/tests/benchmark/test_e2e.py
new file mode 100644
index 0000000..3e29f0f
--- /dev/null
+++ b/tests/benchmark/test_e2e.py
@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+import pytest
+
+
+@pytest.mark.benchmark_e2e
+def test_real_local_smoke():
+    """Run explicitly: pytest -m benchmark_e2e tests/benchmark/test_e2e.py -s."""
+    from benchmark.audio.cli import build_parser
+
+    args = build_parser().parse_args(["local-smoke", "--count", "2", "--max-minutes", "15"])
+    args.func(args)
diff --git a/tests/benchmark/test_mock_llm.py b/tests/benchmark/test_mock_llm.py
new file mode 100644
index 0000000..15d2835
--- /dev/null
+++ b/tests/benchmark/test_mock_llm.py
@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import json
+import urllib.request
+
+from benchmark.audio.mock_llm import response_plan, start_in_thread
+
+
+def test_response_plan_is_deterministic():
+    assert response_plan("doc-1", 1, "standard") == response_plan("doc-1", 1, "standard")
+    delay, status = response_plan("doc-1", 2, "fast")
+    assert delay >= 0
+    assert status == 200
+
+
+def test_mock_has_health_and_openai_response():
+    server, thread = start_in_thread(port=0, profile="fast")
+    port = server.server_address[1]
+    try:
+        with urllib.request.urlopen(f"http://127.0.0.1:{port}/health") as response:
+            assert json.load(response)["status"] == "ok"
+        request = urllib.request.Request(
+            f"http://127.0.0.1:{port}/v1/chat/completions",
+            data=json.dumps({"messages": [{"role": "user", "content": "hello"}]}).encode(),
+            headers={"Content-Type": "application/json", "X-Benchmark-Doc-Id": "doc-1"},
+            method="POST",
+        )
+        with urllib.request.urlopen(request) as response:
+            payload = json.load(response)
+        content = payload["choices"][0]["message"]["content"]
+        assert json.loads(content)["category"] == "benchmark"
+        assert len(content) >= 3800
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
diff --git a/tests/benchmark/test_report.py b/tests/benchmark/test_report.py
new file mode 100644
index 0000000..fa1f6eb
--- /dev/null
+++ b/tests/benchmark/test_report.py
@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+import csv
+import json
+
+from benchmark.audio.report import build_report
+
+
+def test_failed_run_report_does_not_require_lance(tmp_path):
+    (tmp_path / "summary.json").write_text(
+        json.dumps({"run_id": "run-1", "state": "failed", "elapsed_s": 2.5, "error": "boom"})
+    )
+    (tmp_path / "metadata.json").write_text(json.dumps({"versions": {"daft": "test"}}))
+    with (tmp_path / "resources.csv").open("w", newline="") as out:
+        writer = csv.DictWriter(
+            out,
+            fieldnames=[
+                "timestamp", "hostname", "memory_total_bytes", "memory_used_bytes",
+                "ray_process_rss_bytes", "cpu_percent", "object_store_used_bytes",
+            ],
+        )
+        writer.writeheader()
+        writer.writerow(
+            {
+                "timestamp": 1, "hostname": "node", "memory_total_bytes": 1000,
+                "memory_used_bytes": 100, "ray_process_rss_bytes": 50,
+                "cpu_percent": 10, "object_store_used_bytes": 20,
+            }
+        )
+        writer.writerow(
+            {
+                "timestamp": 2, "hostname": "node", "memory_total_bytes": 1000,
+                "memory_used_bytes": 200, "ray_process_rss_bytes": 75,
+                "cpu_percent": 20, "object_store_used_bytes": 30,
+            }
+        )
+
+    report = build_report(tmp_path)
+    assert report["state"] == "failed"
+    assert report["resources"]["node_memory_used_peak_bytes"] == 200
+    assert report["resources"]["nodes"]["node"]["memory_used_p95_bytes"] == 195
+    assert (tmp_path / "report.json").exists()
+    assert (tmp_path / "report.md").exists()
diff --git a/tests/benchmark/test_submit.py b/tests/benchmark/test_submit.py
new file mode 100644
index 0000000..d40d9f0
--- /dev/null
+++ b/tests/benchmark/test_submit.py
@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from benchmark.audio.config import BenchmarkConfig
+from benchmark.audio.submit import build_entrypoint_args
+
+
+def test_submit_forwards_all_run_controls():
+    cfg = BenchmarkConfig(
+        run_id="run",
+        manifest_uri="s3://b/m.parquet",
+        lance_uri="s3://b/o.lance",
+        mock_url="http://mock",
+        ray_address="auto",
+        max_minutes=60,
+        llm_timeout_s=22,
+        llm_max_attempts=5,
+        sample_interval_s=3,
+        num_partitions=64,
+        run_dir="/mnt/bench/run",
+    )
+    args = build_entrypoint_args(cfg)
+    rendered = " ".join(args)
+    assert "--ray-address auto" in rendered
+    assert "--llm-timeout-s 22" in rendered
+    assert "--llm-max-attempts 5" in rendered
+    assert "--sample-interval-s 3" in rendered
+    assert "--num-partitions 64" in rendered
+    assert "--run-dir /mnt/bench/run" in rendered
diff --git a/uv.lock b/uv.lock
index 7052353..cef60ab 100644
--- a/uv.lock
+++ b/uv.lock
@@ -891,6 +891,7 @@ dependencies = [
     { name = "daft-lance" },
     { name = "funasr" },
     { name = "getdaft" },
+    { name = "httpx" },
     { name = "insightface" },
     { name = "lance-ray" },
     { name = "modelscope" },
@@ -899,6 +900,7 @@ dependencies = [
     { name = "openai" },
     { name = "opencv-python-headless" },
     { name = "pillow" },
+    { name = "psutil" },
     { name = "pyarrow" },
     { name = "pylance" },
     { name = "python-dotenv" },
@@ -914,6 +916,7 @@ requires-dist = [
     { name = "daft-lance", specifier = ">=0.4.0" },
     { name = "funasr" },
     { name = "getdaft" },
+    { name = "httpx" },
     { name = "insightface" },
     { name = "lance-ray" },
     { name = "modelscope" },
@@ -922,6 +925,7 @@ requires-dist = [
     { name = "openai" },
     { name = "opencv-python-headless" },
     { name = "pillow" },
+    { name = "psutil" },
     { name = "pyarrow" },
     { name = "pylance", specifier = "<8.0.0" },
     { name = "python-dotenv" },
@@ -1354,6 +1358,22 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/19/c7/5f7c636ec43e0c545e28d1f1db71990108306f7bdcb89f069ba97e428e7f/protobuf-7.35.1-py3-none-any.whl", hash = "sha256:4bc97768d8fe4ad6743c8a19403e314511ed9f6d13205b687e52421c023ac1b9", size = 171659, upload-time = "2026-06-11T21:55:39.155Z" },
 ]
 
+[[package]]
+name = "psutil"
+version = "7.2.2"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/aa/c6/d1ddf4abb55e93cebc4f2ed8b5d6dbad109ecb8d63748dd2b20ab5e57ebe/psutil-7.2.2.tar.gz", hash = "sha256:0746f5f8d406af344fd547f1c8daa5f5c33dbc293bb8d6a16d80b4bb88f59372", size = 493740, upload-time = "2026-01-28T18:14:54.428Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/e7/36/5ee6e05c9bd427237b11b3937ad82bb8ad2752d72c6969314590dd0c2f6e/psutil-7.2.2-cp36-abi3-macosx_10_9_x86_64.whl", hash = "sha256:ed0cace939114f62738d808fdcecd4c869222507e266e574799e9c0faa17d486", size = 129090, upload-time = "2026-01-28T18:15:22.168Z" },
+    { url = "https://files.pythonhosted.org/packages/80/c4/f5af4c1ca8c1eeb2e92ccca14ce8effdeec651d5ab6053c589b074eda6e1/psutil-7.2.2-cp36-abi3-macosx_11_0_arm64.whl", hash = "sha256:1a7b04c10f32cc88ab39cbf606e117fd74721c831c98a27dc04578deb0c16979", size = 129859, upload-time = "2026-01-28T18:15:23.795Z" },
+    { url = "https://files.pythonhosted.org/packages/b5/70/5d8df3b09e25bce090399cf48e452d25c935ab72dad19406c77f4e828045/psutil-7.2.2-cp36-abi3-manylinux2010_x86_64.manylinux_2_12_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:076a2d2f923fd4821644f5ba89f059523da90dc9014e85f8e45a5774ca5bc6f9", size = 155560, upload-time = "2026-01-28T18:15:25.976Z" },
+    { url = "https://files.pythonhosted.org/packages/63/65/37648c0c158dc222aba51c089eb3bdfa238e621674dc42d48706e639204f/psutil-7.2.2-cp36-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:b0726cecd84f9474419d67252add4ac0cd9811b04d61123054b9fb6f57df6e9e", size = 156997, upload-time = "2026-01-28T18:15:27.794Z" },
+    { url = "https://files.pythonhosted.org/packages/8e/13/125093eadae863ce03c6ffdbae9929430d116a246ef69866dad94da3bfbc/psutil-7.2.2-cp36-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:fd04ef36b4a6d599bbdb225dd1d3f51e00105f6d48a28f006da7f9822f2606d8", size = 148972, upload-time = "2026-01-28T18:15:29.342Z" },
+    { url = "https://files.pythonhosted.org/packages/04/78/0acd37ca84ce3ddffaa92ef0f571e073faa6d8ff1f0559ab1272188ea2be/psutil-7.2.2-cp36-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:b58fabe35e80b264a4e3bb23e6b96f9e45a3df7fb7eed419ac0e5947c61e47cc", size = 148266, upload-time = "2026-01-28T18:15:31.597Z" },
+    { url = "https://files.pythonhosted.org/packages/b4/90/e2159492b5426be0c1fef7acba807a03511f97c5f86b3caeda6ad92351a7/psutil-7.2.2-cp37-abi3-win_amd64.whl", hash = "sha256:eb7e81434c8d223ec4a219b5fc1c47d0417b12be7ea866e24fb5ad6e84b3d988", size = 137737, upload-time = "2026-01-28T18:15:33.849Z" },
+    { url = "https://files.pythonhosted.org/packages/8c/c7/7bb2e321574b10df20cbde462a94e2b71d05f9bbda251ef27d104668306a/psutil-7.2.2-cp37-abi3-win_arm64.whl", hash = "sha256:8c233660f575a5a89e6d4cb65d9f938126312bca76d8fe087b947b3a1aaac9ee", size = 134617, upload-time = "2026-01-28T18:15:36.514Z" },
+]
+
 [[package]]
 name = "pyarrow"
 version = "24.0.0"