feat: add benchmark reports and cli

Problem: Benchmark runs had no packaged command or durable aggregate reports, making comparisons hard to automate and review.

Risk: Report generation and CLI failures could leak credentials or leave half-published artifacts; centralized redaction, strict validation, atomic cleanup, and regression tests cover these paths.

pyproject.toml

@@ -21,6 +21,7 @@ dev = [
 
 [project.scripts]
 agent-lab = "agent_lab.main:run"
+agent-lab-benchmark = "agent_lab.presentation.benchmark_cli:main"
 
 [build-system]
 requires = ["setuptools>=68"]

src/agent_lab/infrastructure/benchmark_reporting.py

@@ -0,0 +1,501 @@
+from __future__ import annotations
+
+import json
+import os
+import re
+from collections import defaultdict
+from collections.abc import Iterable, Sequence
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Literal
+from uuid import uuid4
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from agent_lab.application.benchmark import (
+    BenchmarkCaseId,
+    BenchmarkConfig,
+    BenchmarkMode,
+    BenchmarkRunResult,
+)
+
+
+REPORT_SCHEMA_VERSION = 1
+ERROR_LIMIT = 2048
+REDACTED = "[REDACTED]"
+
+
+class _StrictReportModel(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+        strict=True,
+        hide_input_in_errors=True,
+    )
+
+
+class BenchmarkLatencyAggregate(_StrictReportModel):
+    count: int = Field(ge=0)
+    p50: float | None
+    p95: float | None
+
+
+class BenchmarkValueAggregate(_StrictReportModel):
+    count: int = Field(ge=0)
+    total: int
+    mean: float | None
+
+
+class BenchmarkAggregate(_StrictReportModel):
+    case_id: BenchmarkCaseId
+    mode: BenchmarkMode
+    attempted: int = Field(ge=0)
+    passed: int = Field(ge=0)
+    failed: int = Field(ge=0)
+    initial_provider_ttft_ms: BenchmarkLatencyAggregate
+    visible_ttft_ms: BenchmarkLatencyAggregate
+    turn_wall_time_ms: BenchmarkLatencyAggregate
+    prompt_tokens: BenchmarkValueAggregate
+    completion_tokens: BenchmarkValueAggregate
+    total_tokens: BenchmarkValueAggregate
+    cached_tokens: BenchmarkValueAggregate
+    model_call_count: BenchmarkValueAggregate
+    fallback_count: BenchmarkValueAggregate
+    tool_count: BenchmarkValueAggregate
+    warnings: list[str]
+
+
+class BenchmarkReportTarget(_StrictReportModel):
+    base_url: str
+    model: str
+    mock: bool
+
+
+class BenchmarkReportConfig(_StrictReportModel):
+    runs_per_case: int = Field(ge=1)
+    cases: list[BenchmarkCaseId]
+    modes: list[BenchmarkMode]
+
+
+class BenchmarkOverall(_StrictReportModel):
+    attempted: int = Field(ge=0)
+    passed: int = Field(ge=0)
+    failed: int = Field(ge=0)
+    warnings: list[str]
+
+
+class BenchmarkReport(_StrictReportModel):
+    schema_version: Literal[1]
+    generated_at: datetime
+    target: BenchmarkReportTarget
+    config: BenchmarkReportConfig
+    runs: list[BenchmarkRunResult]
+    aggregates: list[BenchmarkAggregate]
+    overall: BenchmarkOverall
+
+
+class BenchmarkReportWriteError(RuntimeError):
+    pass
+
+
+def percentile(values: Sequence[int | float], quantile: float) -> float | None:
+    if not 0.0 <= quantile <= 1.0:
+        raise ValueError("quantile must be between 0 and 1")
+    if not values:
+        return None
+    ordered = sorted(float(value) for value in values)
+    index = (len(ordered) - 1) * quantile
+    lower_index = int(index)
+    upper_index = min(lower_index + 1, len(ordered) - 1)
+    fraction = index - lower_index
+    return ordered[lower_index] + (
+        ordered[upper_index] - ordered[lower_index]
+    ) * fraction
+
+
+def redact_text(value: str, *, secrets: Iterable[str] = ()) -> str:
+    redacted = value
+    for secret in sorted(
+        {secret for secret in secrets if secret},
+        key=len,
+        reverse=True,
+    ):
+        redacted = redacted.replace(secret, REDACTED)
+
+    redacted = re.sub(
+        r"(?i)\bBearer\s+[^\s,;]+",
+        f"Bearer {REDACTED}",
+        redacted,
+    )
+    redacted = re.sub(
+        r"(?i)\b(Authorization|X-API-Key|API-Key)\b"
+        r"(\s*[:=]\s*)"
+        r"(?:\"[^\"]*\"|'[^']*'|[^\s,;]+)",
+        lambda match: f"{match.group(1)}{match.group(2)}{REDACTED}",
+        redacted,
+    )
+    redacted = re.sub(
+        r"(?i)([\"']?(?:api[_-]?key|access[_-]?token|auth[_-]?token|token|key)"
+        r"[\"']?\s*(?:=|:)\s*)"
+        r"(?:\"[^\"]*\"|'[^']*'|[^&\s,;}]+)",
+        lambda match: f"{match.group(1)}{REDACTED}",
+        redacted,
+    )
+    return redacted
+
+
+def sanitize_error(value: str | None, *, secrets: Iterable[str] = ()) -> str | None:
+    if value is None:
+        return None
+    return redact_text(value, secrets=secrets)[:ERROR_LIMIT]
+
+
+def build_benchmark_report(
+    config: BenchmarkConfig,
+    runs: Sequence[BenchmarkRunResult],
+    *,
+    mock: bool,
+    generated_at: datetime | None = None,
+    secrets: Iterable[str] = (),
+) -> BenchmarkReport:
+    secret_values = tuple(secret for secret in secrets if secret)
+    sanitized_runs = [
+        _sanitize_run(run, secrets=secret_values) for run in runs
+    ]
+    aggregates = _aggregate_runs(sanitized_runs)
+    warnings = [
+        warning
+        for aggregate in aggregates
+        for warning in aggregate.warnings
+    ]
+    passed = sum(run.status == "passed" for run in sanitized_runs)
+    failed = len(sanitized_runs) - passed
+    resolved_generated_at = generated_at or datetime.now(timezone.utc)
+    if resolved_generated_at.tzinfo is None:
+        resolved_generated_at = resolved_generated_at.replace(tzinfo=timezone.utc)
+    return BenchmarkReport(
+        schema_version=REPORT_SCHEMA_VERSION,
+        generated_at=resolved_generated_at,
+        target=BenchmarkReportTarget(
+            base_url=redact_text(config.base_url, secrets=secret_values),
+            model=redact_text(config.model, secrets=secret_values),
+            mock=mock,
+        ),
+        config=BenchmarkReportConfig(
+            runs_per_case=config.runs_per_case,
+            cases=list(config.cases),
+            modes=list(config.modes),
+        ),
+        runs=sanitized_runs,
+        aggregates=aggregates,
+        overall=BenchmarkOverall(
+            attempted=len(sanitized_runs),
+            passed=passed,
+            failed=failed,
+            warnings=warnings,
+        ),
+    )
+
+
+def _sanitize_run(
+    run: BenchmarkRunResult,
+    *,
+    secrets: Sequence[str],
+) -> BenchmarkRunResult:
+    string_list_fields = (
+        "event_names",
+        "batch_event_names",
+        "tool_event_names",
+        "event_sources",
+        "tool_statuses",
+        "semantic_failures",
+    )
+    updates: dict[str, object] = {
+        field_name: [
+            redact_text(value, secrets=secrets)
+            for value in getattr(run, field_name)
+        ]
+        for field_name in string_list_fields
+    }
+    updates["error"] = sanitize_error(run.error, secrets=secrets)
+    updates["model_calls"] = [
+        call.model_copy(
+            update={
+                "first_item_kind": (
+                    redact_text(call.first_item_kind, secrets=secrets)
+                    if call.first_item_kind is not None
+                    else None
+                )
+            }
+        )
+        for call in run.model_calls
+    ]
+    return run.model_copy(update=updates)
+
+
+def _aggregate_runs(
+    runs: Sequence[BenchmarkRunResult],
+) -> list[BenchmarkAggregate]:
+    grouped: dict[
+        tuple[BenchmarkCaseId, BenchmarkMode],
+        list[BenchmarkRunResult],
+    ] = defaultdict(list)
+    for run in runs:
+        grouped[(run.case_id, run.mode)].append(run)
+
+    return [
+        _aggregate_group(case_id, mode, grouped[(case_id, mode)])
+        for case_id, mode in sorted(
+            grouped,
+            key=lambda item: (item[0].value, item[1].value),
+        )
+    ]
+
+
+def _aggregate_group(
+    case_id: BenchmarkCaseId,
+    mode: BenchmarkMode,
+    runs: Sequence[BenchmarkRunResult],
+) -> BenchmarkAggregate:
+    passed_runs = [run for run in runs if run.status == "passed"]
+    latency_fields = (
+        "initial_provider_ttft_ms",
+        "visible_ttft_ms",
+        "turn_wall_time_ms",
+    )
+    latency_values = {
+        field_name: [
+            value
+            for run in passed_runs
+            if (value := getattr(run, field_name)) is not None
+        ]
+        for field_name in latency_fields
+    }
+    warnings = [
+        f"{case_id.value}/{mode.value} {field_name} p95 low confidence: "
+        f"{len(values)} samples; fewer than 20"
+        for field_name, values in latency_values.items()
+        if 0 < len(values) < 20
+    ]
+    value_fields = (
+        "prompt_tokens",
+        "completion_tokens",
+        "total_tokens",
+        "cached_tokens",
+        "model_call_count",
+        "fallback_count",
+        "tool_count",
+    )
+    value_aggregates = {
+        field_name: _value_aggregate(
+            [
+                value
+                for run in passed_runs
+                if (value := getattr(run, field_name)) is not None
+            ]
+        )
+        for field_name in value_fields
+    }
+    return BenchmarkAggregate(
+        case_id=case_id,
+        mode=mode,
+        attempted=len(runs),
+        passed=len(passed_runs),
+        failed=len(runs) - len(passed_runs),
+        initial_provider_ttft_ms=_latency_aggregate(
+            latency_values["initial_provider_ttft_ms"]
+        ),
+        visible_ttft_ms=_latency_aggregate(
+            latency_values["visible_ttft_ms"]
+        ),
+        turn_wall_time_ms=_latency_aggregate(
+            latency_values["turn_wall_time_ms"]
+        ),
+        prompt_tokens=value_aggregates["prompt_tokens"],
+        completion_tokens=value_aggregates["completion_tokens"],
+        total_tokens=value_aggregates["total_tokens"],
+        cached_tokens=value_aggregates["cached_tokens"],
+        model_call_count=value_aggregates["model_call_count"],
+        fallback_count=value_aggregates["fallback_count"],
+        tool_count=value_aggregates["tool_count"],
+        warnings=warnings,
+    )
+
+
+def _latency_aggregate(values: Sequence[int]) -> BenchmarkLatencyAggregate:
+    return BenchmarkLatencyAggregate(
+        count=len(values),
+        p50=percentile(values, 0.50),
+        p95=percentile(values, 0.95),
+    )
+
+
+def _value_aggregate(values: Sequence[int]) -> BenchmarkValueAggregate:
+    total = sum(values)
+    return BenchmarkValueAggregate(
+        count=len(values),
+        total=total,
+        mean=(float(total) / len(values)) if values else None,
+    )
+
+
+def report_to_json(report: BenchmarkReport) -> str:
+    return json.dumps(
+        report.model_dump(mode="json"),
+        ensure_ascii=False,
+        indent=2,
+    ) + "\n"
+
+
+def render_markdown(report: BenchmarkReport) -> str:
+    payload = report.model_dump(mode="json")
+    target = payload["target"]
+    overall = payload["overall"]
+    lines = [
+        "# Agent Lab Benchmark Report",
+        "",
+        f"- Schema version: {payload['schema_version']}",
+        f"- Generated at: {payload['generated_at']}",
+        f"- Base URL: `{_escape_markdown(str(target['base_url']))}`",
+        f"- Model: `{_escape_markdown(str(target['model']))}`",
+        f"- Mock: `{str(target['mock']).lower()}`",
+        "",
+        "## Overall",
+        "",
+        f"- Attempted: {overall['attempted']}",
+        f"- Passed: {overall['passed']}",
+        f"- Failed: {overall['failed']}",
+        f"- Warnings: {len(overall['warnings'])}",
+        "",
+        "## Aggregates",
+        "",
+        "| Case | Mode | Attempted | Passed | Failed | Provider TTFT p50/p95 ms | Visible TTFT p50/p95 ms | Turn wall p50/p95 ms | Tokens total/mean | Model calls total/mean | Fallbacks total/mean | Tools total/mean |",
+        "| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
+    ]
+    for aggregate in payload["aggregates"]:
+        lines.append(
+            "| "
+            + " | ".join(
+                [
+                    _escape_markdown(str(aggregate["case_id"])),
+                    _escape_markdown(str(aggregate["mode"])),
+                    str(aggregate["attempted"]),
+                    str(aggregate["passed"]),
+                    str(aggregate["failed"]),
+                    _percentile_pair(aggregate["initial_provider_ttft_ms"]),
+                    _percentile_pair(aggregate["visible_ttft_ms"]),
+                    _percentile_pair(aggregate["turn_wall_time_ms"]),
+                    _total_mean(aggregate["total_tokens"]),
+                    _total_mean(aggregate["model_call_count"]),
+                    _total_mean(aggregate["fallback_count"]),
+                    _total_mean(aggregate["tool_count"]),
+                ]
+            )
+            + " |"
+        )
+
+    lines.extend(["", "## Warnings", ""])
+    if overall["warnings"]:
+        lines.extend(
+            f"- {_escape_markdown(str(warning))}"
+            for warning in overall["warnings"]
+        )
+    else:
+        lines.append("- None")
+
+    lines.extend(
+        [
+            "",
+            "## Runs",
+            "",
+            "| Case | Mode | Iteration | Status | Error |",
+            "| --- | --- | ---: | --- | --- |",
+        ]
+    )
+    for run in payload["runs"]:
+        failure_text = "; ".join(run["semantic_failures"])
+        error_text = run["error"] or failure_text or "-"
+        lines.append(
+            "| "
+            + " | ".join(
+                [
+                    _escape_markdown(str(run["case_id"])),
+                    _escape_markdown(str(run["mode"])),
+                    str(run["iteration"]),
+                    str(run["status"]),
+                    _escape_markdown(str(error_text)),
+                ]
+            )
+            + " |"
+        )
+    return "\n".join(lines) + "\n"
+
+
+def _percentile_pair(metric: dict[str, object]) -> str:
+    return f"{_display(metric['p50'])}/{_display(metric['p95'])}"
+
+
+def _total_mean(metric: dict[str, object]) -> str:
+    return f"{_display(metric['total'])}/{_display(metric['mean'])}"
+
+
+def _display(value: object) -> str:
+    return "-" if value is None else str(value)
+
+
+def _escape_markdown(value: str) -> str:
+    return value.replace("|", "\\|").replace("\r", " ").replace("\n", " ")
+
+
+def write_benchmark_report(
+    report: BenchmarkReport,
+    output_dir: str | Path,
+    *,
+    timestamp: str | None = None,
+) -> tuple[Path, Path]:
+    directory = Path(output_dir)
+    temp_paths: list[Path] = []
+    published_paths: list[Path] = []
+    try:
+        directory.mkdir(parents=True, exist_ok=True)
+        stem = timestamp or report.generated_at.astimezone(timezone.utc).strftime(
+            "%Y%m%dT%H%M%SZ"
+        )
+        json_path, markdown_path = _unique_report_paths(directory, stem)
+        nonce = uuid4().hex
+        json_temp = directory / f".{json_path.name}.{nonce}.tmp"
+        markdown_temp = directory / f".{markdown_path.name}.{nonce}.tmp"
+        temp_paths.extend([json_temp, markdown_temp])
+        _write_text(json_temp, report_to_json(report))
+        _write_text(markdown_temp, render_markdown(report))
+        os.replace(json_temp, json_path)
+        published_paths.append(json_path)
+        os.replace(markdown_temp, markdown_path)
+        published_paths.append(markdown_path)
+        return json_path, markdown_path
+    except Exception as exc:
+        for path in temp_paths + published_paths:
+            try:
+                path.unlink(missing_ok=True)
+            except OSError:
+                pass
+        raise BenchmarkReportWriteError(
+            "failed to write benchmark reports"
+        ) from exc
+
+
+def _write_text(path: Path, content: str) -> None:
+    with path.open("x", encoding="utf-8") as handle:
+        handle.write(content)
+        handle.flush()
+        os.fsync(handle.fileno())
+
+
+def _unique_report_paths(directory: Path, stem: str) -> tuple[Path, Path]:
+    suffix = 0
+    while True:
+        candidate = stem if suffix == 0 else f"{stem}-{suffix}"
+        json_path = directory / f"{candidate}.json"
+        markdown_path = directory / f"{candidate}.md"
+        if not json_path.exists() and not markdown_path.exists():
+            return json_path, markdown_path
+        suffix += 1

src/agent_lab/presentation/benchmark_cli.py

@@ -0,0 +1,216 @@
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import sys
+from collections.abc import Awaitable, Callable, Sequence
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, NoReturn, TextIO
+
+from agent_lab.application.benchmark import (
+    BenchmarkCaseId,
+    BenchmarkConfig,
+    BenchmarkMode,
+    BenchmarkRunResult,
+    BenchmarkRunner,
+)
+from agent_lab.infrastructure.benchmark_reporting import (
+    BenchmarkReport,
+    build_benchmark_report,
+    redact_text,
+    sanitize_error,
+    write_benchmark_report,
+)
+from agent_lab.settings import Settings
+
+
+DEFAULT_OUTPUT_DIR = Path("outputs/benchmarks")
+
+
+class _RedactingArgumentParser(argparse.ArgumentParser):
+    def error(self, message: str) -> NoReturn:
+        self.print_usage(sys.stderr)
+        self.exit(2, f"{self.prog}: error: {redact_text(message)}\n")
+
+
+@dataclass(frozen=True)
+class BenchmarkCliDependencies:
+    settings_factory: Callable[[], object] = Settings
+    runner_factory: Callable[..., object] = BenchmarkRunner
+    report_builder: Callable[..., BenchmarkReport] = build_benchmark_report
+    report_writer: Callable[..., tuple[Path, Path]] = write_benchmark_report
+    run_async: Callable[[Awaitable[list[BenchmarkRunResult]]], list[BenchmarkRunResult]] = (
+        asyncio.run
+    )
+    now: Callable[[], datetime] = lambda: datetime.now(timezone.utc)
+    stdout: TextIO = field(default_factory=lambda: sys.stdout)
+    stderr: TextIO = field(default_factory=lambda: sys.stderr)
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = _RedactingArgumentParser(
+        prog="agent-lab-benchmark",
+        description="Run Agent Lab invocation benchmarks and write JSON/Markdown reports.",
+    )
+    parser.add_argument("--config", required=True, help="Benchmark JSON config path.")
+    parser.add_argument("--base-url", help="Override the JSON base_url.")
+    parser.add_argument("--model", help="Override the JSON model.")
+    parser.add_argument(
+        "--runs",
+        type=int,
+        help="Override runs_per_case.",
+    )
+    parser.add_argument(
+        "--output-dir",
+        default=str(DEFAULT_OUTPUT_DIR),
+        help=f"Report directory (default: {DEFAULT_OUTPUT_DIR}).",
+    )
+    parser.add_argument(
+        "--mock",
+        action="store_true",
+        help="Use deterministic mock provider streams; no API key is required.",
+    )
+    parser.add_argument(
+        "--case",
+        action="append",
+        choices=[case_id.value for case_id in BenchmarkCaseId],
+        help="Replace JSON cases; repeat to select multiple cases.",
+    )
+    parser.add_argument(
+        "--mode",
+        action="append",
+        choices=[mode.value for mode in BenchmarkMode],
+        help="Replace JSON modes; repeat to select multiple modes.",
+    )
+    return parser
+
+
+def run_cli(
+    argv: Sequence[str] | None = None,
+    *,
+    dependencies: BenchmarkCliDependencies | None = None,
+) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    deps = dependencies or BenchmarkCliDependencies()
+    api_key = ""
+
+    try:
+        config = _load_config(Path(args.config), args)
+        if not args.mock:
+            settings = deps.settings_factory()
+            api_key = str(getattr(settings, "openai_api_key", "")).strip()
+            if not api_key:
+                raise ValueError("API key is required for live benchmark runs")
+    except KeyboardInterrupt:
+        return 130
+    except Exception as exc:
+        _print_error(exc, deps.stderr, secrets=())
+        return 2
+
+    results: list[BenchmarkRunResult] = []
+    runner_error: Exception | None = None
+    try:
+        runner = deps.runner_factory(
+            config=config,
+            api_key=api_key or None,
+            mock=args.mock,
+        )
+        run = getattr(runner, "run")
+        results = deps.run_async(run())
+    except KeyboardInterrupt:
+        return 130
+    except Exception as exc:
+        runner_error = exc
+
+    try:
+        report = deps.report_builder(
+            config,
+            results,
+            mock=args.mock,
+            generated_at=deps.now(),
+            secrets=(api_key,),
+        )
+        json_path, markdown_path = deps.report_writer(
+            report,
+            Path(args.output_dir),
+        )
+    except KeyboardInterrupt:
+        return 130
+    except Exception as exc:
+        _print_error(exc, deps.stderr, secrets=(api_key,))
+        return 2
+
+    _print_summary(
+        report,
+        json_path,
+        markdown_path,
+        deps.stdout,
+        secrets=(api_key,),
+    )
+    if runner_error is not None:
+        _print_error(runner_error, deps.stderr, secrets=(api_key,))
+        return 2
+    return 1 if report.overall.failed else 0
+
+
+def _load_config(path: Path, args: argparse.Namespace) -> BenchmarkConfig:
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(payload, dict):
+        raise ValueError("benchmark config must be a JSON object")
+    resolved: dict[str, Any] = dict(payload)
+    if args.base_url is not None:
+        resolved["base_url"] = args.base_url
+    if args.model is not None:
+        resolved["model"] = args.model
+    if args.runs is not None:
+        resolved["runs_per_case"] = args.runs
+    if args.case is not None:
+        resolved["cases"] = list(args.case)
+    if args.mode is not None:
+        resolved["modes"] = list(args.mode)
+    return BenchmarkConfig.model_validate(resolved)
+
+
+def _print_summary(
+    report: BenchmarkReport,
+    json_path: Path,
+    markdown_path: Path,
+    stream: TextIO,
+    *,
+    secrets: Sequence[str],
+) -> None:
+    summary = (
+        f"attempted={report.overall.attempted} "
+        f"passed={report.overall.passed} "
+        f"failed={report.overall.failed} "
+        f"warnings={len(report.overall.warnings)}"
+    )
+    lines = (
+        summary,
+        f"json_report={json_path}",
+        f"markdown_report={markdown_path}",
+    )
+    for line in lines:
+        print(redact_text(line, secrets=secrets), file=stream)
+
+
+def _print_error(
+    exc: Exception,
+    stream: TextIO,
+    *,
+    secrets: Sequence[str],
+) -> None:
+    message = sanitize_error(str(exc), secrets=secrets) or exc.__class__.__name__
+    print(f"benchmark error: {message}", file=stream)
+
+
+def main() -> NoReturn:
+    raise SystemExit(run_cli())
+
+
+if __name__ == "__main__":
+    main()

tests/test_benchmark_cli.py

@@ -0,0 +1,465 @@
+import importlib
+import importlib.util
+import json
+import tomllib
+from datetime import datetime, timezone
+from io import StringIO
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from agent_lab.application.benchmark import (
+    BenchmarkCaseId,
+    BenchmarkMode,
+    BenchmarkRunResult,
+)
+
+
+def _cli_module():
+    spec = importlib.util.find_spec("agent_lab.presentation.benchmark_cli")
+    assert spec is not None, "benchmark_cli module must be implemented"
+    return importlib.import_module("agent_lab.presentation.benchmark_cli")
+
+
+def _write_config(path: Path, **overrides: object) -> Path:
+    payload = {
+        "schema_version": 1,
+        "base_url": "https://json.example/v1",
+        "model": "json-model",
+        "runs_per_case": 1,
+        "cases": ["ordinary_chat"],
+        "modes": ["dual_agent"],
+    }
+    payload.update(overrides)
+    path.write_text(json.dumps(payload))
+    return path
+
+
+def _run_result(
+    *,
+    case_id: BenchmarkCaseId = BenchmarkCaseId.ORDINARY_CHAT,
+    mode: BenchmarkMode = BenchmarkMode.DUAL_AGENT,
+    iteration: int = 1,
+    status: str = "passed",
+    error: str | None = None,
+) -> BenchmarkRunResult:
+    return BenchmarkRunResult(
+        case_id=case_id,
+        mode=mode,
+        iteration=iteration,
+        status=status,
+        initial_provider_ttft_ms=1,
+        visible_ttft_ms=2,
+        turn_wall_time_ms=3,
+        prompt_tokens=2,
+        completion_tokens=3,
+        total_tokens=5,
+        cached_tokens=0,
+        model_call_count=1,
+        fallback_count=0,
+        tool_count=0,
+        event_names=[],
+        batch_event_names=[],
+        tool_event_names=[],
+        event_sources=[],
+        tool_statuses=[],
+        tool_latencies_ms=[],
+        semantic_failures=[],
+        error=error,
+        model_calls=[],
+    )
+
+
+class FakeRunner:
+    def __init__(self, results: list[BenchmarkRunResult]) -> None:
+        self.results = results
+
+    async def run(self) -> list[BenchmarkRunResult]:
+        return self.results
+
+
+def _dependencies(
+    cli,
+    *,
+    results: list[BenchmarkRunResult],
+    api_key: str = "",
+    runner_calls: list[dict[str, object]] | None = None,
+    report_writer=None,
+    stderr: StringIO | None = None,
+    stdout: StringIO | None = None,
+):
+    calls = runner_calls if runner_calls is not None else []
+
+    def runner_factory(**kwargs):
+        calls.append(kwargs)
+        return FakeRunner(results)
+
+    return cli.BenchmarkCliDependencies(
+        settings_factory=lambda: SimpleNamespace(
+            openai_api_key=api_key,
+            openai_base_url="https://env-must-not-win.example/v1",
+            openai_default_model="env-must-not-win",
+        ),
+        runner_factory=runner_factory,
+        report_writer=report_writer or cli.write_benchmark_report,
+        now=lambda: datetime(2026, 7, 14, 1, 2, 3, tzinfo=timezone.utc),
+        stdout=stdout or StringIO(),
+        stderr=stderr or StringIO(),
+    )
+
+
+def test_cli_overrides_json_and_repeatable_case_mode_replace_lists(tmp_path: Path):
+    cli = _cli_module()
+    config_path = _write_config(
+        tmp_path / "config.json",
+        cases=["ordinary_chat", "web_search_two_answers"],
+        modes=["dual_agent", "chat_agent_tools"],
+    )
+    output_dir = tmp_path / "reports"
+    runner_calls: list[dict[str, object]] = []
+    dependencies = _dependencies(
+        cli,
+        results=[
+            _run_result(
+                case_id=BenchmarkCaseId.SESSION_TERMINATE,
+                mode=BenchmarkMode.CHAT_AGENT_TOOLS,
+            )
+        ],
+        api_key="env-key",
+        runner_calls=runner_calls,
+    )
+
+    exit_code = cli.run_cli(
+        [
+            "--config",
+            str(config_path),
+            "--base-url",
+            "https://cli.example/v1",
+            "--model",
+            "cli-model",
+            "--runs",
+            "3",
+            "--case",
+            "ordinary_chat",
+            "--case",
+            "session_terminate",
+            "--mode",
+            "chat_agent_tools",
+            "--output-dir",
+            str(output_dir),
+        ],
+        dependencies=dependencies,
+    )
+
+    assert exit_code == 0
+    assert len(runner_calls) == 1
+    resolved = runner_calls[0]
+    assert resolved["api_key"] == "env-key"
+    assert resolved["mock"] is False
+    config = resolved["config"]
+    assert config.base_url == "https://cli.example/v1"
+    assert config.model == "cli-model"
+    assert config.runs_per_case == 3
+    assert config.cases == [
+        BenchmarkCaseId.ORDINARY_CHAT,
+        BenchmarkCaseId.SESSION_TERMINATE,
+    ]
+    assert config.modes == [BenchmarkMode.CHAT_AGENT_TOOLS]
+    assert len(list(output_dir.glob("*.json"))) == 1
+    assert len(list(output_dir.glob("*.md"))) == 1
+
+
+def test_mock_run_needs_no_api_key_and_exits_zero(tmp_path: Path):
+    cli = _cli_module()
+    config_path = _write_config(tmp_path / "config.json")
+    output_dir = tmp_path / "reports"
+    stdout = StringIO()
+    stderr = StringIO()
+    dependencies = _dependencies(
+        cli,
+        results=[_run_result()],
+        api_key="",
+        stdout=stdout,
+        stderr=stderr,
+    )
+
+    exit_code = cli.run_cli(
+        [
+            "--config",
+            str(config_path),
+            "--mock",
+            "--output-dir",
+            str(output_dir),
+        ],
+        dependencies=dependencies,
+    )
+
+    assert exit_code == 0
+    assert "attempted=1 passed=1 failed=0" in stdout.getvalue()
+    assert "json_report=" in stdout.getvalue()
+    assert "markdown_report=" in stdout.getvalue()
+    assert stderr.getvalue() == ""
+
+
+def test_partial_failed_run_exits_one_writes_redacted_reports(tmp_path: Path):
+    cli = _cli_module()
+    secret = "sk-cli-real-secret"
+    config_path = _write_config(tmp_path / "config.json")
+    output_dir = tmp_path / "reports"
+    stdout = StringIO()
+    stderr = StringIO()
+    dependencies = _dependencies(
+        cli,
+        results=[
+            _run_result(iteration=1),
+            _run_result(
+                iteration=2,
+                status="failed",
+                error=f"Authorization: Bearer {secret}",
+            ),
+        ],
+        api_key=secret,
+        stdout=stdout,
+        stderr=stderr,
+    )
+
+    exit_code = cli.run_cli(
+        [
+            "--config",
+            str(config_path),
+            "--output-dir",
+            str(output_dir),
+        ],
+        dependencies=dependencies,
+    )
+
+    assert exit_code == 1
+    assert "attempted=2 passed=1 failed=1" in stdout.getvalue()
+    assert secret not in stdout.getvalue()
+    assert stderr.getvalue() == ""
+    report_text = "\n".join(
+        path.read_text() for path in sorted(output_dir.iterdir())
+    )
+    assert secret not in report_text
+    assert "[REDACTED]" in report_text
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        {"schema_version": 1, "api_key": "secret-in-json"},
+        {"schema_version": 1, "base_url": "not-a-url", "model": "m"},
+    ],
+)
+def test_bad_config_exits_two_without_leaking_secret(
+    tmp_path: Path, payload: dict[str, object]
+):
+    cli = _cli_module()
+    config_path = tmp_path / "bad.json"
+    config_path.write_text(json.dumps(payload))
+    stderr = StringIO()
+    runner_calls: list[dict[str, object]] = []
+    dependencies = _dependencies(
+        cli,
+        results=[],
+        runner_calls=runner_calls,
+        stderr=stderr,
+    )
+
+    exit_code = cli.run_cli(
+        ["--config", str(config_path), "--mock"],
+        dependencies=dependencies,
+    )
+
+    assert exit_code == 2
+    assert runner_calls == []
+    assert "secret-in-json" not in stderr.getvalue()
+    assert "benchmark error:" in stderr.getvalue()
+
+
+def test_final_validation_allows_cli_to_supply_missing_base_url_and_model(
+    tmp_path: Path,
+):
+    cli = _cli_module()
+    config_path = tmp_path / "minimal.json"
+    config_path.write_text(
+        json.dumps(
+            {
+                "schema_version": 1,
+                "cases": ["ordinary_chat"],
+                "modes": ["dual_agent"],
+            }
+        )
+    )
+    runner_calls: list[dict[str, object]] = []
+    dependencies = _dependencies(
+        cli,
+        results=[_run_result()],
+        runner_calls=runner_calls,
+    )
+
+    exit_code = cli.run_cli(
+        [
+            "--config",
+            str(config_path),
+            "--base-url",
+            "https://cli.example/v1",
+            "--model",
+            "cli-model",
+            "--mock",
+            "--output-dir",
+            str(tmp_path / "reports"),
+        ],
+        dependencies=dependencies,
+    )
+
+    assert exit_code == 0
+    assert runner_calls[0]["config"].model == "cli-model"
+
+
+def test_live_run_without_api_key_exits_two(tmp_path: Path):
+    cli = _cli_module()
+    config_path = _write_config(tmp_path / "config.json")
+    stderr = StringIO()
+    runner_calls: list[dict[str, object]] = []
+    dependencies = _dependencies(
+        cli,
+        results=[],
+        api_key="",
+        runner_calls=runner_calls,
+        stderr=stderr,
+    )
+
+    exit_code = cli.run_cli(
+        ["--config", str(config_path)],
+        dependencies=dependencies,
+    )
+
+    assert exit_code == 2
+    assert runner_calls == []
+    assert "api key is required" in stderr.getvalue().lower()
+
+
+def test_report_failure_exits_two_and_redacts_stderr(tmp_path: Path):
+    cli = _cli_module()
+    secret = "sk-report-failure-secret"
+    config_path = _write_config(tmp_path / "config.json")
+    stderr = StringIO()
+
+    def failing_writer(report, output_dir):
+        del report, output_dir
+        raise OSError(f"API-Key: {secret}")
+
+    dependencies = _dependencies(
+        cli,
+        results=[_run_result()],
+        api_key=secret,
+        report_writer=failing_writer,
+        stderr=stderr,
+    )
+
+    exit_code = cli.run_cli(
+        ["--config", str(config_path)],
+        dependencies=dependencies,
+    )
+
+    assert exit_code == 2
+    assert secret not in stderr.getvalue()
+    assert "[REDACTED]" in stderr.getvalue()
+
+
+def test_unexpected_runner_failure_still_attempts_report_write(tmp_path: Path):
+    cli = _cli_module()
+    config_path = _write_config(tmp_path / "config.json")
+    writes: list[object] = []
+    stderr = StringIO()
+
+    class FailingRunner:
+        async def run(self):
+            raise RuntimeError("runner failed")
+
+    def write_report(report, output_dir):
+        writes.append(report)
+        output = Path(output_dir)
+        output.mkdir(parents=True)
+        json_path = output / "report.json"
+        md_path = output / "report.md"
+        json_path.write_text("{}")
+        md_path.write_text("# report")
+        return json_path, md_path
+
+    dependencies = cli.BenchmarkCliDependencies(
+        settings_factory=lambda: SimpleNamespace(openai_api_key="key"),
+        runner_factory=lambda **kwargs: FailingRunner(),
+        report_writer=write_report,
+        now=lambda: datetime(2026, 7, 14, tzinfo=timezone.utc),
+        stdout=StringIO(),
+        stderr=stderr,
+    )
+
+    exit_code = cli.run_cli(
+        [
+            "--config",
+            str(config_path),
+            "--output-dir",
+            str(tmp_path / "reports"),
+        ],
+        dependencies=dependencies,
+    )
+
+    assert exit_code == 2
+    assert len(writes) == 1
+    assert writes[0].overall.attempted == 0
+    assert "runner failed" in stderr.getvalue()
+
+
+def test_help_and_required_config_use_argparse(capsys: pytest.CaptureFixture[str]):
+    cli = _cli_module()
+
+    with pytest.raises(SystemExit) as help_exit:
+        cli.run_cli(["--help"])
+    assert help_exit.value.code == 0
+    assert "--config" in capsys.readouterr().out
+
+    with pytest.raises(SystemExit) as missing_exit:
+        cli.run_cli([])
+    assert missing_exit.value.code == 2
+
+
+def test_argparse_errors_are_redacted(capsys: pytest.CaptureFixture[str]):
+    cli = _cli_module()
+    secret = "argument-secret"
+
+    with pytest.raises(SystemExit) as exc_info:
+        cli.run_cli(
+            [
+                "--config",
+                "unused.json",
+                "--mode",
+                f"api_key={secret}",
+            ]
+        )
+
+    assert exc_info.value.code == 2
+    stderr = capsys.readouterr().err
+    assert secret not in stderr
+    assert "[REDACTED]" in stderr
+
+
+def test_main_raises_system_exit(monkeypatch: pytest.MonkeyPatch):
+    cli = _cli_module()
+    monkeypatch.setattr(cli, "run_cli", lambda argv=None: 1)
+
+    with pytest.raises(SystemExit) as exc_info:
+        cli.main()
+
+    assert exc_info.value.code == 1
+
+
+def test_pyproject_registers_benchmark_console_script():
+    pyproject = tomllib.loads(Path("pyproject.toml").read_text())
+
+    assert pyproject["project"]["scripts"]["agent-lab-benchmark"] == (
+        "agent_lab.presentation.benchmark_cli:main"
+    )

tests/test_benchmark_reporting.py

@@ -0,0 +1,414 @@
+import importlib
+import importlib.util
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+from pydantic import ValidationError
+
+from agent_lab.application.benchmark import (
+    BenchmarkCaseId,
+    BenchmarkConfig,
+    BenchmarkMode,
+    BenchmarkRunResult,
+)
+
+
+def _reporting_module():
+    spec = importlib.util.find_spec(
+        "agent_lab.infrastructure.benchmark_reporting"
+    )
+    assert spec is not None, "benchmark_reporting module must be implemented"
+    return importlib.import_module(
+        "agent_lab.infrastructure.benchmark_reporting"
+    )
+
+
+def _config(**overrides: object) -> BenchmarkConfig:
+    payload = {
+        "schema_version": 1,
+        "base_url": "https://provider.example/v1",
+        "model": "benchmark-model",
+        "runs_per_case": 1,
+        "cases": [BenchmarkCaseId.ORDINARY_CHAT],
+        "modes": [BenchmarkMode.DUAL_AGENT],
+    }
+    payload.update(overrides)
+    return BenchmarkConfig.model_validate(payload)
+
+
+def _run(
+    *,
+    case_id: BenchmarkCaseId = BenchmarkCaseId.ORDINARY_CHAT,
+    mode: BenchmarkMode = BenchmarkMode.DUAL_AGENT,
+    iteration: int = 1,
+    status: str = "passed",
+    initial_provider_ttft_ms: int | None = 10,
+    visible_ttft_ms: int | None = 20,
+    turn_wall_time_ms: int | None = 30,
+    prompt_tokens: int | None = 2,
+    completion_tokens: int | None = 3,
+    total_tokens: int | None = 5,
+    cached_tokens: int | None = 1,
+    model_call_count: int | None = 1,
+    fallback_count: int | None = 0,
+    tool_count: int | None = 0,
+    semantic_failures: list[str] | None = None,
+    error: str | None = None,
+) -> BenchmarkRunResult:
+    return BenchmarkRunResult(
+        case_id=case_id,
+        mode=mode,
+        iteration=iteration,
+        status=status,
+        initial_provider_ttft_ms=initial_provider_ttft_ms,
+        visible_ttft_ms=visible_ttft_ms,
+        turn_wall_time_ms=turn_wall_time_ms,
+        prompt_tokens=prompt_tokens,
+        completion_tokens=completion_tokens,
+        total_tokens=total_tokens,
+        cached_tokens=cached_tokens,
+        model_call_count=model_call_count,
+        fallback_count=fallback_count,
+        tool_count=tool_count,
+        event_names=[],
+        batch_event_names=[],
+        tool_event_names=[],
+        event_sources=[],
+        tool_statuses=[],
+        tool_latencies_ms=[],
+        semantic_failures=semantic_failures or [],
+        error=error,
+        model_calls=[],
+    )
+
+
+@pytest.mark.parametrize(
+    ("values", "quantile", "expected"),
+    [
+        ([], 0.50, None),
+        ([7], 0.50, 7.0),
+        ([7], 0.95, 7.0),
+        ([1, 3, 5, 7], 0.50, 4.0),
+        ([1, 3, 5, 7], 0.95, 6.7),
+    ],
+)
+def test_percentile_uses_linear_interpolation(
+    values: list[int], quantile: float, expected: float | None
+):
+    reporting = _reporting_module()
+
+    assert reporting.percentile(values, quantile) == pytest.approx(expected)
+
+
+def test_report_models_are_strict_and_include_required_summary_fields():
+    reporting = _reporting_module()
+    generated_at = datetime(2026, 7, 14, 1, 2, 3, tzinfo=timezone.utc)
+
+    report = reporting.build_benchmark_report(
+        _config(),
+        [_run()],
+        mock=True,
+        generated_at=generated_at,
+    )
+
+    assert isinstance(report, reporting.BenchmarkReport)
+    assert report.schema_version == 1
+    assert report.generated_at == generated_at
+    assert report.target.model_dump() == {
+        "base_url": "https://provider.example/v1",
+        "model": "benchmark-model",
+        "mock": True,
+    }
+    assert report.config.model_dump(mode="json") == {
+        "runs_per_case": 1,
+        "cases": ["ordinary_chat"],
+        "modes": ["dual_agent"],
+    }
+    assert report.overall.model_dump() == {
+        "attempted": 1,
+        "passed": 1,
+        "failed": 0,
+        "warnings": report.overall.warnings,
+    }
+    with pytest.raises(ValidationError):
+        reporting.BenchmarkReport.model_validate(
+            report.model_dump() | {"schema_version": "1"}
+        )
+    with pytest.raises(ValidationError):
+        reporting.BenchmarkAggregate.model_validate(
+            report.aggregates[0].model_dump() | {"unexpected": True}
+        )
+
+
+def test_aggregate_excludes_failed_and_null_samples_from_metrics():
+    reporting = _reporting_module()
+    runs = [
+        _run(
+            iteration=1,
+            initial_provider_ttft_ms=10,
+            visible_ttft_ms=20,
+            turn_wall_time_ms=30,
+            prompt_tokens=2,
+            completion_tokens=3,
+            total_tokens=5,
+            cached_tokens=1,
+            model_call_count=1,
+            fallback_count=0,
+            tool_count=2,
+        ),
+        _run(
+            iteration=2,
+            initial_provider_ttft_ms=None,
+            visible_ttft_ms=40,
+            turn_wall_time_ms=None,
+            prompt_tokens=None,
+            completion_tokens=7,
+            total_tokens=7,
+            cached_tokens=None,
+            model_call_count=3,
+            fallback_count=1,
+            tool_count=None,
+        ),
+        _run(
+            iteration=3,
+            status="failed",
+            initial_provider_ttft_ms=999,
+            visible_ttft_ms=999,
+            turn_wall_time_ms=999,
+            prompt_tokens=999,
+            completion_tokens=999,
+            total_tokens=999,
+            cached_tokens=999,
+            model_call_count=999,
+            fallback_count=999,
+            tool_count=999,
+            error="failed",
+        ),
+    ]
+
+    aggregate = reporting.build_benchmark_report(
+        _config(runs_per_case=3), runs, mock=True
+    ).aggregates[0]
+
+    assert (aggregate.attempted, aggregate.passed, aggregate.failed) == (3, 2, 1)
+    assert aggregate.initial_provider_ttft_ms.model_dump() == {
+        "count": 1,
+        "p50": 10.0,
+        "p95": 10.0,
+    }
+    assert aggregate.visible_ttft_ms.model_dump() == {
+        "count": 2,
+        "p50": 30.0,
+        "p95": 39.0,
+    }
+    assert aggregate.turn_wall_time_ms.count == 1
+    assert aggregate.prompt_tokens.model_dump() == {
+        "count": 1,
+        "total": 2,
+        "mean": 2.0,
+    }
+    assert aggregate.completion_tokens.model_dump() == {
+        "count": 2,
+        "total": 10,
+        "mean": 5.0,
+    }
+    assert aggregate.total_tokens.total == 12
+    assert aggregate.cached_tokens.total == 1
+    assert aggregate.model_call_count.model_dump() == {
+        "count": 2,
+        "total": 4,
+        "mean": 2.0,
+    }
+    assert aggregate.fallback_count.total == 1
+    assert aggregate.tool_count.model_dump() == {
+        "count": 1,
+        "total": 2,
+        "mean": 2.0,
+    }
+
+
+def test_aggregates_are_stably_sorted_and_warn_for_low_confidence_p95():
+    reporting = _reporting_module()
+    runs = [
+        _run(
+            case_id=BenchmarkCaseId.WEB_SEARCH_TWO_ANSWERS,
+            mode=BenchmarkMode.DUAL_AGENT,
+        ),
+        _run(
+            case_id=BenchmarkCaseId.ORDINARY_CHAT,
+            mode=BenchmarkMode.CHAT_AGENT_TOOLS,
+        ),
+        _run(
+            case_id=BenchmarkCaseId.ORDINARY_CHAT,
+            mode=BenchmarkMode.DUAL_AGENT,
+        ),
+    ]
+
+    report = reporting.build_benchmark_report(
+        _config(
+            cases=[
+                BenchmarkCaseId.WEB_SEARCH_TWO_ANSWERS,
+                BenchmarkCaseId.ORDINARY_CHAT,
+            ],
+            modes=[
+                BenchmarkMode.CHAT_AGENT_TOOLS,
+                BenchmarkMode.DUAL_AGENT,
+            ],
+        ),
+        runs,
+        mock=True,
+    )
+
+    assert [(item.case_id.value, item.mode.value) for item in report.aggregates] == [
+        ("ordinary_chat", "chat_agent_tools"),
+        ("ordinary_chat", "dual_agent"),
+        ("web_search_two_answers", "dual_agent"),
+    ]
+    assert all(
+        any("p95 low confidence" in warning for warning in item.warnings)
+        for item in report.aggregates
+    )
+    assert report.overall.warnings == [
+        warning
+        for aggregate in report.aggregates
+        for warning in aggregate.warnings
+    ]
+
+
+def test_errors_are_redacted_before_truncation_in_json_and_markdown():
+    reporting = _reporting_module()
+    real_key = "sk-live-secret-value"
+    leaked_error = (
+        f"Authorization: Bearer {real_key}; "
+        f"API-Key={real_key}; "
+        f"https://example.test/?api_key={real_key}&token={real_key}; "
+        f'{{"access_token":"{real_key}","key":"{real_key}"}}; '
+        + "x" * 3000
+    )
+
+    report = reporting.build_benchmark_report(
+        _config(),
+        [
+            _run(
+                status="failed",
+                semantic_failures=[f"token={real_key}"],
+                error=leaked_error,
+            )
+        ],
+        mock=False,
+        secrets=[real_key],
+    )
+    payload = report.model_dump(mode="json")
+    json_text = reporting.report_to_json(report)
+    markdown = reporting.render_markdown(report)
+
+    assert len(report.runs[0].error) == 2048
+    assert real_key not in json.dumps(payload)
+    assert real_key not in json_text
+    assert real_key not in markdown
+    assert "[REDACTED]" in json_text
+    assert "[REDACTED]" in markdown
+
+
+def test_configured_secret_is_redacted_from_all_report_string_surfaces():
+    reporting = _reporting_module()
+    real_key = "sk-everywhere-secret"
+    run = _run().model_copy(
+        update={
+            "event_names": [real_key],
+            "event_sources": [f"Bearer {real_key}"],
+            "tool_statuses": [f"token={real_key}"],
+        }
+    )
+
+    report = reporting.build_benchmark_report(
+        _config(
+            base_url=f"https://provider.example/{real_key}",
+            model=f"model-{real_key}",
+        ),
+        [run],
+        mock=False,
+        secrets=[real_key],
+    )
+    rendered = reporting.report_to_json(report) + reporting.render_markdown(report)
+
+    assert real_key not in rendered
+    assert "[REDACTED]" in rendered
+
+
+def test_markdown_is_rendered_from_the_same_json_facts():
+    reporting = _reporting_module()
+    report = reporting.build_benchmark_report(
+        _config(runs_per_case=2),
+        [
+            _run(iteration=1, visible_ttft_ms=10, total_tokens=5),
+            _run(iteration=2, visible_ttft_ms=30, total_tokens=9),
+        ],
+        mock=True,
+        generated_at=datetime(2026, 7, 14, tzinfo=timezone.utc),
+    )
+    payload = json.loads(reporting.report_to_json(report))
+    markdown = reporting.render_markdown(report)
+    aggregate = payload["aggregates"][0]
+
+    assert str(aggregate["visible_ttft_ms"]["p50"]) in markdown
+    assert str(aggregate["visible_ttft_ms"]["p95"]) in markdown
+    assert str(aggregate["total_tokens"]["total"]) in markdown
+    assert str(aggregate["total_tokens"]["mean"]) in markdown
+    assert payload == report.model_dump(mode="json")
+
+
+def test_atomic_write_creates_matching_unique_json_and_markdown_paths(tmp_path: Path):
+    reporting = _reporting_module()
+    report = reporting.build_benchmark_report(
+        _config(), [_run()], mock=True
+    )
+    timestamp = "20260714T010203Z"
+    (tmp_path / f"{timestamp}.json").write_text("existing")
+    (tmp_path / f"{timestamp}.md").write_text("existing")
+
+    json_path, markdown_path = reporting.write_benchmark_report(
+        report,
+        tmp_path,
+        timestamp=timestamp,
+    )
+
+    assert json_path.name == f"{timestamp}-1.json"
+    assert markdown_path.name == f"{timestamp}-1.md"
+    assert json.loads(json_path.read_text()) == report.model_dump(mode="json")
+    assert markdown_path.read_text() == reporting.render_markdown(report)
+    assert list(tmp_path.glob("*.tmp")) == []
+
+
+def test_atomic_write_removes_temps_and_published_half_on_replace_failure(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+):
+    reporting = _reporting_module()
+    report = reporting.build_benchmark_report(
+        _config(), [_run()], mock=True
+    )
+    real_replace = reporting.os.replace
+    calls = 0
+
+    def fail_second_replace(source: Path, destination: Path) -> None:
+        nonlocal calls
+        calls += 1
+        if calls == 2:
+            raise OSError("simulated markdown publish failure")
+        real_replace(source, destination)
+
+    monkeypatch.setattr(reporting.os, "replace", fail_second_replace)
+
+    with pytest.raises(
+        reporting.BenchmarkReportWriteError,
+        match="failed to write benchmark reports",
+    ):
+        reporting.write_benchmark_report(
+            report,
+            tmp_path,
+            timestamp="20260714T010203Z",
+        )
+
+    assert list(tmp_path.iterdir()) == []