From e8426bd3c847c5cc031ecc1f6572c427f2780174 Mon Sep 17 00:00:00 2001 From: Ritwij Aryan Parmar Date: Thu, 16 Jul 2026 11:59:16 -0400 Subject: [PATCH 1/2] Add benchmark run artifacts --- runner/README.md | 4 + runner/src/coval_bench/config.py | 1 + runner/src/coval_bench/runner/artifacts.py | 177 ++++++++++++++++++ runner/src/coval_bench/runner/orchestrator.py | 83 ++++++++ runner/tests/unit/test_orchestrator.py | 44 +++++ runner/tests/unit/test_run_artifacts.py | 102 ++++++++++ 6 files changed, 411 insertions(+) create mode 100644 runner/src/coval_bench/runner/artifacts.py create mode 100644 runner/tests/unit/test_run_artifacts.py diff --git a/runner/README.md b/runner/README.md index 00da492e..d54a4d25 100644 --- a/runner/README.md +++ b/runner/README.md @@ -46,6 +46,10 @@ 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: +docker compose run --rm -e RUN_ARTIFACT_DIR=/tmp/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 --text "hello" diff --git a/runner/src/coval_bench/config.py b/runner/src/coval_bench/config.py index 2e5a9b1c..ecc81e1f 100644 --- a/runner/src/coval_bench/config.py +++ b/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 diff --git a/runner/src/coval_bench/runner/artifacts.py b/runner/src/coval_bench/runner/artifacts.py new file mode 100644 index 00000000..96100e94 --- /dev/null +++ b/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 transcript is None: + 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]), + ) + ] diff --git a/runner/src/coval_bench/runner/orchestrator.py b/runner/src/coval_bench/runner/orchestrator.py index d007fea0..9952e163 100644 --- a/runner/src/coval_bench/runner/orchestrator.py +++ b/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,45 @@ async def _refresh_series_bucket(writer: Any, run_id: int, settings: Settings) - return +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 = 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 +1161,20 @@ def _on_sigterm() -> None: success_count=success_count, fail_count=fail_count, ) + _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 +1231,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() + _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,6 +1283,21 @@ 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)] + _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, + ) try: await writer.finish_run(run_id, status=RunStatus.FAILED, error=err_msg) except Exception as write_exc: diff --git a/runner/tests/unit/test_orchestrator.py b/runner/tests/unit/test_orchestrator.py index 2fe78967..1c356c45 100644 --- a/runner/tests/unit/test_orchestrator.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/runner/tests/unit/test_run_artifacts.py b/runner/tests/unit/test_run_artifacts.py new file mode 100644 index 00000000..95d16c2f --- /dev/null +++ b/runner/tests/unit/test_run_artifacts.py @@ -0,0 +1,102 @@ +# Copyright 2026 The Coval Benchmarks Authors +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import json +from datetime import UTC, datetime +from pathlib import Path + +from coval_bench.db.models import Benchmark, Result, ResultStatus +from coval_bench.runner.artifacts import write_run_artifact + + +def test_write_run_artifact_sanitizes_transcripts_and_groups_failures(tmp_path: Path) -> None: + results = [ + Result( + run_id=7, + provider="deepgram", + model="nova-3", + benchmark=Benchmark.STT, + metric_type="WER", + metric_value=4.2, + metric_units="percent", + audio_filename="0001.wav", + transcript="hello world", + status=ResultStatus.SUCCESS, + error=None, + ), + Result( + run_id=7, + provider="deepgram", + model="nova-3", + benchmark=Benchmark.STT, + metric_type="TTFS", + metric_value=None, + metric_units="seconds", + audio_filename="0001.wav", + transcript="hello world", + status=ResultStatus.FAILED, + error="timeout", + ), + Result( + run_id=7, + provider="deepgram", + model="nova-3", + benchmark=Benchmark.STT, + metric_type="TTFT", + metric_value=None, + metric_units="seconds", + audio_filename="0002.wav", + transcript="second transcript", + status=ResultStatus.FAILED, + error="timeout", + ), + ] + + path = write_run_artifact( + artifact_dir=tmp_path, + run_id=7, + runner_sha="abc", + dataset_id="stt-v3", + dataset_sha256="deadbeef", + benchmark_kind="stt", + smoke=False, + scheduled_at=datetime(2026, 7, 16, 10, 0, tzinfo=UTC), + started_at=datetime(2026, 7, 16, 10, 1, tzinfo=UTC), + finished_at=datetime(2026, 7, 16, 10, 2, tzinfo=UTC), + status="partial", + results=results, + ) + + records = [json.loads(line) for line in path.read_text().splitlines()] + assert [r["record_type"] for r in records] == [ + "run", + "result", + "result", + "result", + "summary", + ] + assert records[0]["dataset_id"] == "stt-v3" + assert "hello world" not in path.read_text() + assert records[1]["transcript_sha256"] + assert records[1]["transcript_chars"] == 11 + assert records[-1]["status"] == "partial" + assert records[-1]["success_count"] == 1 + assert records[-1]["fail_count"] == 2 + assert records[-1]["failure_buckets"] == [ + { + "provider": "deepgram", + "model": "nova-3", + "metric_type": "TTFS", + "error": "timeout", + "count": 1, + }, + { + "provider": "deepgram", + "model": "nova-3", + "metric_type": "TTFT", + "error": "timeout", + "count": 1, + }, + ] From d68e3853b9f740e2ac534eb653fd48e6a14c024d Mon Sep 17 00:00:00 2001 From: Ritwij Aryan Parmar Date: Thu, 16 Jul 2026 12:13:10 -0400 Subject: [PATCH 2/2] Address artifact review feedback --- runner/README.md | 4 ++- runner/src/coval_bench/runner/artifacts.py | 2 +- runner/src/coval_bench/runner/orchestrator.py | 25 ++++++++++--------- runner/tests/unit/test_run_artifacts.py | 18 ++++++++++++- 4 files changed, 34 insertions(+), 15 deletions(-) diff --git a/runner/README.md b/runner/README.md index d54a4d25..6c744832 100644 --- a/runner/README.md +++ b/runner/README.md @@ -47,7 +47,9 @@ docker compose up -d api # FastAPI on http://localhost:8000 docker compose run --rm runner coval-bench run --smoke --kind tts # Optional: write a portable JSONL artifact for the run: -docker compose run --rm -e RUN_ARTIFACT_DIR=/tmp/artifacts runner \ +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: diff --git a/runner/src/coval_bench/runner/artifacts.py b/runner/src/coval_bench/runner/artifacts.py index 96100e94..67c931e5 100644 --- a/runner/src/coval_bench/runner/artifacts.py +++ b/runner/src/coval_bench/runner/artifacts.py @@ -150,7 +150,7 @@ def _result_row(result: Result) -> ArtifactResultRow: def _transcript_sha256(transcript: str | None) -> str | None: - if transcript is None: + if not isinstance(transcript, str): return None return hashlib.sha256(transcript.encode("utf-8")).hexdigest() diff --git a/runner/src/coval_bench/runner/orchestrator.py b/runner/src/coval_bench/runner/orchestrator.py index 9952e163..c2a5cf04 100644 --- a/runner/src/coval_bench/runner/orchestrator.py +++ b/runner/src/coval_bench/runner/orchestrator.py @@ -821,7 +821,7 @@ async def _refresh_series_bucket(writer: Any, run_id: int, settings: Settings) - return -def _write_run_artifact_if_enabled( +async def _write_run_artifact_if_enabled( *, settings: Settings, run_id: int, @@ -840,7 +840,8 @@ def _write_run_artifact_if_enabled( if settings.run_artifact_dir is None: return try: - artifact_path = write_run_artifact( + artifact_path = await asyncio.to_thread( + write_run_artifact, artifact_dir=settings.run_artifact_dir, run_id=run_id, runner_sha=runner_sha, @@ -1161,7 +1162,7 @@ def _on_sigterm() -> None: success_count=success_count, fail_count=fail_count, ) - _write_run_artifact_if_enabled( + await _write_run_artifact_if_enabled( settings=settings, run_id=run_id, runner_sha=settings.runner_sha, @@ -1231,7 +1232,7 @@ 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() - _write_run_artifact_if_enabled( + await _write_run_artifact_if_enabled( settings=settings, run_id=run_id, runner_sha=settings.runner_sha, @@ -1284,7 +1285,14 @@ def _on_sigterm() -> None: err_msg = _truncate(str(exc)) log_run_failed(err_msg, exc) typed_results = [r for r in all_results if isinstance(r, Result)] - _write_run_artifact_if_enabled( + 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, @@ -1298,13 +1306,6 @@ def _on_sigterm() -> None: status=str(RunStatus.FAILED), results=typed_results, ) - 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, - ) _emit_posthog( posthog_client, "benchmark_run_failed", diff --git a/runner/tests/unit/test_run_artifacts.py b/runner/tests/unit/test_run_artifacts.py index 95d16c2f..1670a134 100644 --- a/runner/tests/unit/test_run_artifacts.py +++ b/runner/tests/unit/test_run_artifacts.py @@ -52,6 +52,19 @@ def test_write_run_artifact_sanitizes_transcripts_and_groups_failures(tmp_path: status=ResultStatus.FAILED, error="timeout", ), + Result( + run_id=7, + provider="deepgram", + model="nova-3", + benchmark=Benchmark.STT, + metric_type="AudioToFinal", + metric_value=1.2, + metric_units="seconds", + audio_filename="0003.wav", + transcript=None, + status=ResultStatus.SUCCESS, + error=None, + ), ] path = write_run_artifact( @@ -75,6 +88,7 @@ def test_write_run_artifact_sanitizes_transcripts_and_groups_failures(tmp_path: "result", "result", "result", + "result", "summary", ] assert records[0]["dataset_id"] == "stt-v3" @@ -82,7 +96,9 @@ def test_write_run_artifact_sanitizes_transcripts_and_groups_failures(tmp_path: assert records[1]["transcript_sha256"] assert records[1]["transcript_chars"] == 11 assert records[-1]["status"] == "partial" - assert records[-1]["success_count"] == 1 + assert records[4]["transcript_sha256"] is None + assert records[4]["transcript_chars"] is None + assert records[-1]["success_count"] == 2 assert records[-1]["fail_count"] == 2 assert records[-1]["failure_buckets"] == [ {