Add portable benchmark run artifacts

runner/README.md

@@ -46,6 +46,12 @@ docker compose up -d api         # FastAPI on http://localhost:8000
 # Trigger a single-item benchmark run (writes to the local Postgres):
 docker compose run --rm runner coval-bench run --smoke --kind tts
 
+# Optional: write a portable JSONL artifact for the run:
+mkdir -p artifacts
+docker compose run --rm -v "$PWD/artifacts:/artifacts" \
+  -e RUN_ARTIFACT_DIR=/artifacts runner \
+  coval-bench run --smoke --kind tts
+
 # Probe one TTS provider without DB writes:
 docker compose run --rm runner coval-bench tts-smoke \
   --provider cartesia --model sonic-3 --voice <voice-id> --text "hello"

runner/src/coval_bench/config.py

@@ -62,6 +62,7 @@ def _dataset_id_not_reserved(cls, value: str) -> str:
     # --- Runner ---
     runner_sha: str = "dev"
     log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"
+    run_artifact_dir: Path | None = None
 
     # Scheduler period in seconds. The runner floors its start time to this grid
     # to compute each run's scheduled_at. MUST stay in sync with the Cloud

runner/src/coval_bench/runner/artifacts.py

@@ -0,0 +1,177 @@
+# Copyright 2026 The Coval Benchmarks Authors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Portable JSONL artifacts for benchmark runs.
+
+The database is the source of truth for public aggregates. These artifacts are
+for run-level debugging: one file contains the reproducibility metadata, every
+metric row that was produced, and a compact failure summary.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from collections import Counter
+from datetime import datetime
+from pathlib import Path
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict
+
+from coval_bench.db.models import Result
+
+
+class ArtifactRunHeader(BaseModel):
+    """First JSONL record in a run artifact."""
+
+    record_type: Literal["run"] = "run"
+    schema_version: int = 1
+    run_id: int
+    runner_sha: str
+    dataset_id: str
+    dataset_sha256: str
+    benchmark_kind: str
+    smoke: bool
+    scheduled_at: datetime | None
+    started_at: datetime
+    finished_at: datetime
+
+
+class ArtifactResultRow(BaseModel):
+    """Sanitized metric row copied from the in-memory Result model."""
+
+    model_config = ConfigDict(use_enum_values=True)
+
+    record_type: Literal["result"] = "result"
+    provider: str
+    model: str
+    voice: str | None
+    benchmark: str
+    metric_type: str
+    metric_value: float | None
+    metric_units: str | None
+    audio_filename: str | None
+    status: str
+    error: str | None
+    transcript_sha256: str | None
+    transcript_chars: int | None
+    http_version: str | None
+    submit_to_headers_ms: float | None
+
+
+class ArtifactFailureBucket(BaseModel):
+    """One grouped failure reason in the artifact summary."""
+
+    provider: str
+    model: str
+    metric_type: str
+    error: str
+    count: int
+
+
+class ArtifactSummary(BaseModel):
+    """Last JSONL record in a run artifact."""
+
+    record_type: Literal["summary"] = "summary"
+    status: str
+    total_results: int
+    success_count: int
+    fail_count: int
+    failure_buckets: list[ArtifactFailureBucket]
+
+
+def write_run_artifact(
+    *,
+    artifact_dir: Path,
+    run_id: int,
+    runner_sha: str,
+    dataset_id: str,
+    dataset_sha256: str,
+    benchmark_kind: str,
+    smoke: bool,
+    scheduled_at: datetime | None,
+    started_at: datetime,
+    finished_at: datetime,
+    status: str,
+    results: list[Result],
+) -> Path:
+    """Write a single atomic JSONL artifact for a completed or partial run."""
+    artifact_dir.mkdir(parents=True, exist_ok=True)
+    final_path = artifact_dir / f"run-{run_id}.jsonl"
+    tmp_path = artifact_dir / f".run-{run_id}.jsonl.tmp"
+
+    rows = [_result_row(r) for r in results]
+    summary = ArtifactSummary(
+        status=status,
+        total_results=len(rows),
+        success_count=sum(1 for r in rows if r.status == "success"),
+        fail_count=sum(1 for r in rows if r.status == "failed"),
+        failure_buckets=_failure_buckets(rows),
+    )
+    header = ArtifactRunHeader(
+        run_id=run_id,
+        runner_sha=runner_sha,
+        dataset_id=dataset_id,
+        dataset_sha256=dataset_sha256,
+        benchmark_kind=benchmark_kind,
+        smoke=smoke,
+        scheduled_at=scheduled_at,
+        started_at=started_at,
+        finished_at=finished_at,
+    )
+
+    with tmp_path.open("w", encoding="utf-8") as fh:
+        fh.write(header.model_dump_json() + "\n")
+        for row in rows:
+            fh.write(row.model_dump_json() + "\n")
+        fh.write(summary.model_dump_json() + "\n")
+    tmp_path.replace(final_path)
+    return final_path
+
+
+def _result_row(result: Result) -> ArtifactResultRow:
+    transcript = getattr(result, "transcript", None)
+    return ArtifactResultRow(
+        provider=str(result.provider),
+        model=str(result.model),
+        voice=result.voice,
+        benchmark=str(result.benchmark),
+        metric_type=str(result.metric_type),
+        metric_value=result.metric_value,
+        metric_units=result.metric_units,
+        audio_filename=result.audio_filename,
+        status=str(result.status),
+        error=result.error,
+        transcript_sha256=_transcript_sha256(transcript),
+        transcript_chars=len(transcript) if isinstance(transcript, str) else None,
+        http_version=result.http_version,
+        submit_to_headers_ms=result.submit_to_headers_ms,
+    )
+
+
+def _transcript_sha256(transcript: str | None) -> str | None:
+    if not isinstance(transcript, str):
+        return None
+    return hashlib.sha256(transcript.encode("utf-8")).hexdigest()
+
+
+def _failure_buckets(rows: list[ArtifactResultRow]) -> list[ArtifactFailureBucket]:
+    counts: Counter[tuple[str, str, str, str]] = Counter()
+    for row in rows:
+        if row.status != "failed" or not row.error:
+            continue
+        counts[(row.provider, row.model, row.metric_type, row.error)] += 1
+
+    return [
+        ArtifactFailureBucket(
+            provider=provider,
+            model=model,
+            metric_type=metric_type,
+            error=error,
+            count=count,
+        )
+        for (provider, model, metric_type, error), count in sorted(
+            counts.items(),
+            key=lambda item: (-item[1], item[0]),
+        )
+    ]

runner/src/coval_bench/runner/orchestrator.py

@@ -60,6 +60,7 @@
     ModelStatus,
     RegisteredModel,
 )
+from coval_bench.runner.artifacts import write_run_artifact
 from coval_bench.runner.retry import with_retry
 
 if TYPE_CHECKING:
@@ -820,6 +821,46 @@ async def _refresh_series_bucket(writer: Any, run_id: int, settings: Settings) -
             return
 
 
+async def _write_run_artifact_if_enabled(
+    *,
+    settings: Settings,
+    run_id: int,
+    runner_sha: str,
+    dataset_id: str,
+    dataset_sha256: str,
+    benchmark_kind: str,
+    smoke: bool,
+    scheduled_at: datetime | None,
+    started_at: datetime,
+    finished_at: datetime,
+    status: str,
+    results: list[Any],
+) -> None:
+    """Best-effort local artifact write; DB writes remain the source of truth."""
+    if settings.run_artifact_dir is None:
+        return
+    try:
+        artifact_path = await asyncio.to_thread(
+            write_run_artifact,
+            artifact_dir=settings.run_artifact_dir,
+            run_id=run_id,
+            runner_sha=runner_sha,
+            dataset_id=dataset_id,
+            dataset_sha256=dataset_sha256,
+            benchmark_kind=benchmark_kind,
+            smoke=smoke,
+            scheduled_at=scheduled_at,
+            started_at=started_at,
+            finished_at=finished_at,
+            status=status,
+            results=results,
+        )
+    except Exception:
+        logger.warning("run_artifact_write_failed", exc_info=True)
+    else:
+        logger.info("run_artifact_written", path=str(artifact_path))
+
+
 async def run_benchmarks(
     *,
     settings: Settings,
@@ -1121,6 +1162,20 @@ def _on_sigterm() -> None:
                 success_count=success_count,
                 fail_count=fail_count,
             )
+            await _write_run_artifact_if_enabled(
+                settings=settings,
+                run_id=run_id,
+                runner_sha=settings.runner_sha,
+                dataset_id=run_dataset_id,
+                dataset_sha256=dataset_sha256,
+                benchmark_kind=benchmark_kind,
+                smoke=smoke,
+                scheduled_at=scheduled_at,
+                started_at=started_at,
+                finished_at=finished_at,
+                status=str(final_status),
+                results=typed_results,
+            )
 
             duration_s = (finished_at - started_at).total_seconds()
             logger.info(
@@ -1177,6 +1232,20 @@ def _on_sigterm() -> None:
                 await asyncio.shield(_refresh_series_bucket(writer, run_id, settings))
             finished_at = datetime.now(tz=UTC)
             sigterm_duration_s = (finished_at - started_at).total_seconds()
+            await _write_run_artifact_if_enabled(
+                settings=settings,
+                run_id=run_id,
+                runner_sha=settings.runner_sha,
+                dataset_id=run_dataset_id,
+                dataset_sha256=dataset_sha256,
+                benchmark_kind=benchmark_kind,
+                smoke=smoke,
+                scheduled_at=scheduled_at,
+                started_at=started_at,
+                finished_at=finished_at,
+                status=str(RunStatus.PARTIAL),
+                results=typed_results,
+            )
             logger.warning(
                 "benchmark_run_finished_early_sigterm",
                 status=str(RunStatus.PARTIAL),
@@ -1215,13 +1284,28 @@ def _on_sigterm() -> None:
             # metric), update run row, then re-raise so the job exits non-zero.
             err_msg = _truncate(str(exc))
             log_run_failed(err_msg, exc)
+            typed_results = [r for r in all_results if isinstance(r, Result)]
             try:
                 await writer.finish_run(run_id, status=RunStatus.FAILED, error=err_msg)
             except Exception as write_exc:
                 logger.error(
                     "run_row_update_failed_after_failure",
                     exc_info=write_exc,
                 )
+            await _write_run_artifact_if_enabled(
+                settings=settings,
+                run_id=run_id,
+                runner_sha=settings.runner_sha,
+                dataset_id=run_dataset_id,
+                dataset_sha256=dataset_sha256,
+                benchmark_kind=benchmark_kind,
+                smoke=smoke,
+                scheduled_at=scheduled_at,
+                started_at=started_at,
+                finished_at=datetime.now(tz=UTC),
+                status=str(RunStatus.FAILED),
+                results=typed_results,
+            )
             _emit_posthog(
                 posthog_client,
                 "benchmark_run_failed",

runner/tests/unit/test_orchestrator.py

@@ -24,6 +24,7 @@
 
 import asyncio
 import contextlib
+import json
 import tempfile
 import wave
 from collections.abc import AsyncIterator, MutableMapping
@@ -330,6 +331,49 @@ async def test_smoke_run_stt(audio_file: Path, settings: Settings) -> None:
     )
 
 
+@pytest.mark.asyncio
+async def test_run_writes_artifact_when_configured(
+    audio_file: Path, settings: Settings, tmp_path: Path
+) -> None:
+    """Opt-in run artifacts capture the final in-memory result set."""
+    provider = MagicMock()
+    provider.measure_ttft = AsyncMock(return_value=_good_transcription())
+    provider_cls = MagicMock(return_value=provider)
+
+    run = _make_run()
+    writer = _make_stub_writer(run)
+    artifact_settings = settings.model_copy(update={"run_artifact_dir": tmp_path})
+
+    async with _orchestrator_env(
+        audio_path=audio_file,
+        stt_items=[_make_dataset_item(audio_file)],
+        stt_providers={"deepgram": provider_cls},
+        run=run,
+        writer=writer,
+    ) as _:
+        summary = await run_benchmarks(
+            settings=artifact_settings,
+            benchmark_kind="stt",
+            smoke=True,
+            matrix_overrides=[*_paused_registry(Benchmark.STT), _stt_entry("deepgram", "nova-2")],
+        )
+
+    artifact = tmp_path / f"run-{summary.run_id}.jsonl"
+    records = [json.loads(line) for line in artifact.read_text().splitlines()]
+    assert records[0]["record_type"] == "run"
+    assert records[0]["run_id"] == summary.run_id
+    assert records[-1]["record_type"] == "summary"
+    assert records[-1]["status"] == str(RunStatus.SUCCEEDED)
+    assert records[-1]["total_results"] == summary.total_results
+    assert {r["metric_type"] for r in records if r["record_type"] == "result"} >= {
+        "TTFT",
+        "AudioToFinal",
+        "RTF",
+        "TTFS",
+        "WER",
+    }
+
+
 # ---------------------------------------------------------------------------
 # 2. test_partial_run
 # ---------------------------------------------------------------------------