fix: complete benchmark report auditability

Problem: Benchmark reports omitted model and tool latency distributions, could race concurrent writers into mixed pairs, hid runner-level failures, and truncated errors by characters instead of UTF-8 bytes.

Risk: Cross-process basename reservations and new overall failure accounting change report output paths and totals; exclusive-lock cleanup, thread/process stress tests, redaction, and full regression coverage protect these boundaries.

src/agent_lab/infrastructure/benchmark_reporting.py

@@ -5,6 +5,7 @@ import os
 import re
 from collections import defaultdict
 from collections.abc import Iterable, Sequence
+from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Literal
@@ -54,6 +55,8 @@ class BenchmarkAggregate(_StrictReportModel):
     initial_provider_ttft_ms: BenchmarkLatencyAggregate
     visible_ttft_ms: BenchmarkLatencyAggregate
     turn_wall_time_ms: BenchmarkLatencyAggregate
+    model_elapsed_ms: BenchmarkLatencyAggregate
+    tool_latency_ms: BenchmarkLatencyAggregate
     prompt_tokens: BenchmarkValueAggregate
     completion_tokens: BenchmarkValueAggregate
     total_tokens: BenchmarkValueAggregate
@@ -81,6 +84,7 @@ class BenchmarkOverall(_StrictReportModel):
     passed: int = Field(ge=0)
     failed: int = Field(ge=0)
     warnings: list[str]
+    system_errors: list[str]
 
 
 class BenchmarkReport(_StrictReportModel):
@@ -97,6 +101,14 @@ class BenchmarkReportWriteError(RuntimeError):
     pass
 
 
+@dataclass(frozen=True)
+class _ReportReservation:
+    json_path: Path
+    markdown_path: Path
+    lock_path: Path
+    lock_fd: int
+
+
 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")
@@ -146,7 +158,11 @@ def redact_text(value: str, *, secrets: Iterable[str] = ()) -> str:
 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]
+    redacted = redact_text(value, secrets=secrets)
+    return redacted.encode("utf-8")[:ERROR_LIMIT].decode(
+        "utf-8",
+        errors="ignore",
+    )
 
 
 def build_benchmark_report(
@@ -156,8 +172,13 @@ def build_benchmark_report(
     mock: bool,
     generated_at: datetime | None = None,
     secrets: Iterable[str] = (),
+    system_errors: Sequence[str] = (),
 ) -> BenchmarkReport:
     secret_values = tuple(secret for secret in secrets if secret)
+    sanitized_system_errors = [
+        sanitize_error(error, secrets=secret_values) or ""
+        for error in system_errors
+    ]
     sanitized_runs = [
         _sanitize_run(run, secrets=secret_values) for run in runs
     ]
@@ -169,6 +190,7 @@ def build_benchmark_report(
     ]
     passed = sum(run.status == "passed" for run in sanitized_runs)
     failed = len(sanitized_runs) - passed
+    system_failure_count = len(sanitized_system_errors)
     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)
@@ -188,10 +210,11 @@ def build_benchmark_report(
         runs=sanitized_runs,
         aggregates=aggregates,
         overall=BenchmarkOverall(
-            attempted=len(sanitized_runs),
+            attempted=len(sanitized_runs) + system_failure_count,
             passed=passed,
-            failed=failed,
+            failed=failed + system_failure_count,
             warnings=warnings,
+            system_errors=sanitized_system_errors,
         ),
     )
 
@@ -270,11 +293,23 @@ def _aggregate_group(
         ]
         for field_name in latency_fields
     }
+    latency_values["model_elapsed_ms"] = [
+        call.elapsed_ms
+        for run in passed_runs
+        for call in run.model_calls
+        if call.elapsed_ms is not None
+    ]
+    latency_values["tool_latency_ms"] = [
+        latency
+        for run in passed_runs
+        for latency in run.tool_latencies_ms
+        if latency is not None
+    ]
     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
+        if len(values) < 20
     ]
     value_fields = (
         "prompt_tokens",
@@ -310,6 +345,12 @@ def _aggregate_group(
         turn_wall_time_ms=_latency_aggregate(
             latency_values["turn_wall_time_ms"]
         ),
+        model_elapsed_ms=_latency_aggregate(
+            latency_values["model_elapsed_ms"]
+        ),
+        tool_latency_ms=_latency_aggregate(
+            latency_values["tool_latency_ms"]
+        ),
         prompt_tokens=value_aggregates["prompt_tokens"],
         completion_tokens=value_aggregates["completion_tokens"],
         total_tokens=value_aggregates["total_tokens"],
@@ -368,8 +409,13 @@ def render_markdown(report: BenchmarkReport) -> str:
         "",
         "## 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 |",
-        "| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
+        "| Case | Mode | Attempted | Passed | Failed | "
+        "Provider TTFT p50/p95 ms | Visible TTFT p50/p95 ms | "
+        "Turn wall p50/p95 ms | Model elapsed p50/p95 ms | "
+        "Tool latency p50/p95 ms | Tokens total/mean | "
+        "Model calls total/mean | Fallbacks total/mean | Tools total/mean |",
+        "| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | "
+        "---: | ---: | ---: | ---: | ---: | ---: |",
     ]
     for aggregate in payload["aggregates"]:
         lines.append(
@@ -384,6 +430,8 @@ def render_markdown(report: BenchmarkReport) -> str:
                     _percentile_pair(aggregate["initial_provider_ttft_ms"]),
                     _percentile_pair(aggregate["visible_ttft_ms"]),
                     _percentile_pair(aggregate["turn_wall_time_ms"]),
+                    _percentile_pair(aggregate["model_elapsed_ms"]),
+                    _percentile_pair(aggregate["tool_latency_ms"]),
                     _total_mean(aggregate["total_tokens"]),
                     _total_mean(aggregate["model_call_count"]),
                     _total_mean(aggregate["fallback_count"]),
@@ -402,6 +450,15 @@ def render_markdown(report: BenchmarkReport) -> str:
     else:
         lines.append("- None")
 
+    lines.extend(["", "## System Errors", ""])
+    if overall["system_errors"]:
+        lines.extend(
+            f"- {_escape_markdown(str(error))}"
+            for error in overall["system_errors"]
+        )
+    else:
+        lines.append("- None")
+
     lines.extend(
         [
             "",
@@ -455,12 +512,15 @@ def write_benchmark_report(
     directory = Path(output_dir)
     temp_paths: list[Path] = []
     published_paths: list[Path] = []
+    reservation: _ReportReservation | None = None
     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)
+        reservation = _reserve_report_paths(directory, stem)
+        json_path = reservation.json_path
+        markdown_path = reservation.markdown_path
         nonce = uuid4().hex
         json_temp = directory / f".{json_path.name}.{nonce}.tmp"
         markdown_temp = directory / f".{markdown_path.name}.{nonce}.tmp"
@@ -481,6 +541,9 @@ def write_benchmark_report(
         raise BenchmarkReportWriteError(
             "failed to write benchmark reports"
         ) from exc
+    finally:
+        if reservation is not None:
+            _release_reservation(reservation)
 
 
 def _write_text(path: Path, content: str) -> None:
@@ -490,12 +553,55 @@ def _write_text(path: Path, content: str) -> None:
         os.fsync(handle.fileno())
 
 
-def _unique_report_paths(directory: Path, stem: str) -> tuple[Path, Path]:
+def _reserve_report_paths(directory: Path, stem: str) -> _ReportReservation:
     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
+        lock_path = directory / f".{candidate}.lock"
+        try:
+            lock_fd = os.open(
+                lock_path,
+                os.O_CREAT | os.O_EXCL | os.O_WRONLY,
+                0o600,
+            )
+        except FileExistsError:
+            suffix += 1
+            continue
+        reservation = _ReportReservation(
+            json_path=json_path,
+            markdown_path=markdown_path,
+            lock_path=lock_path,
+            lock_fd=lock_fd,
+        )
+        try:
+            paths_are_available = (
+                not json_path.exists() and not markdown_path.exists()
+            )
+        except BaseException:
+            _release_reservation(reservation)
+            raise
+        if paths_are_available:
+            return reservation
+        _release_reservation(reservation)
         suffix += 1
+
+
+def _release_reservation(reservation: _ReportReservation) -> None:
+    try:
+        try:
+            path_stat = reservation.lock_path.stat()
+            fd_stat = os.fstat(reservation.lock_fd)
+        except OSError:
+            return
+        if (path_stat.st_dev, path_stat.st_ino) == (
+            fd_stat.st_dev,
+            fd_stat.st_ino,
+        ):
+            reservation.lock_path.unlink(missing_ok=True)
+    finally:
+        try:
+            os.close(reservation.lock_fd)
+        except OSError:
+            pass

src/agent_lab/presentation/benchmark_cli.py

@@ -133,6 +133,11 @@ def run_cli(
             mock=args.mock,
             generated_at=deps.now(),
             secrets=(api_key,),
+            system_errors=(
+                (str(runner_error),)
+                if runner_error is not None
+                else ()
+            ),
         )
         json_path, markdown_path = deps.report_writer(
             report,

tests/test_benchmark_cli.py

@@ -371,13 +371,15 @@ def test_report_failure_exits_two_and_redacts_stderr(tmp_path: Path):
 
 def test_unexpected_runner_failure_still_attempts_report_write(tmp_path: Path):
     cli = _cli_module()
+    secret = "runner-secret"
     config_path = _write_config(tmp_path / "config.json")
     writes: list[object] = []
     stderr = StringIO()
+    stdout = StringIO()
 
     class FailingRunner:
         async def run(self):
-            raise RuntimeError("runner failed")
+            raise RuntimeError(f"Bearer {secret}")
 
     def write_report(report, output_dir):
         writes.append(report)
@@ -390,11 +392,11 @@ def test_unexpected_runner_failure_still_attempts_report_write(tmp_path: Path):
         return json_path, md_path
 
     dependencies = cli.BenchmarkCliDependencies(
-        settings_factory=lambda: SimpleNamespace(openai_api_key="key"),
+        settings_factory=lambda: SimpleNamespace(openai_api_key=secret),
         runner_factory=lambda **kwargs: FailingRunner(),
         report_writer=write_report,
         now=lambda: datetime(2026, 7, 14, tzinfo=timezone.utc),
-        stdout=StringIO(),
+        stdout=stdout,
         stderr=stderr,
     )
 
@@ -410,8 +412,13 @@ def test_unexpected_runner_failure_still_attempts_report_write(tmp_path: Path):
 
     assert exit_code == 2
     assert len(writes) == 1
-    assert writes[0].overall.attempted == 0
-    assert "runner failed" in stderr.getvalue()
+    assert writes[0].overall.attempted == 1
+    assert writes[0].overall.failed == 1
+    assert writes[0].overall.system_errors == ["Bearer [REDACTED]"]
+    assert "attempted=1 passed=0 failed=1" in stdout.getvalue()
+    assert secret not in stdout.getvalue()
+    assert secret not in stderr.getvalue()
+    assert "Bearer [REDACTED]" in stderr.getvalue()
 
 
 def test_help_and_required_config_use_argparse(capsys: pytest.CaptureFixture[str]):

tests/test_benchmark_reporting.py

@@ -1,6 +1,9 @@
 import importlib
 import importlib.util
 import json
+import multiprocessing
+import threading
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -10,6 +13,7 @@ from pydantic import ValidationError
 from agent_lab.application.benchmark import (
     BenchmarkCaseId,
     BenchmarkConfig,
+    BenchmarkModelCallTiming,
     BenchmarkMode,
     BenchmarkRunResult,
 )
@@ -54,8 +58,10 @@ def _run(
     model_call_count: int | None = 1,
     fallback_count: int | None = 0,
     tool_count: int | None = 0,
+    tool_latencies_ms: list[int | None] | None = None,
     semantic_failures: list[str] | None = None,
     error: str | None = None,
+    model_calls: list[BenchmarkModelCallTiming] | None = None,
 ) -> BenchmarkRunResult:
     return BenchmarkRunResult(
         case_id=case_id,
@@ -77,13 +83,92 @@ def _run(
         tool_event_names=[],
         event_sources=[],
         tool_statuses=[],
-        tool_latencies_ms=[],
+        tool_latencies_ms=tool_latencies_ms or [],
         semantic_failures=semantic_failures or [],
         error=error,
-        model_calls=[],
+        model_calls=model_calls or [],
     )
 
 
+def _model_call(
+    call_index: int,
+    elapsed_ms: int | None,
+) -> BenchmarkModelCallTiming:
+    return BenchmarkModelCallTiming(
+        call_index=call_index,
+        call_kind="chat_completion",
+        first_item_kind="raw_chunk",
+        provider_ttft_ms=1,
+        visible_ttft_ms=2,
+        elapsed_ms=elapsed_ms,
+        usage=None,
+    )
+
+
+def _marker_report(marker: str):
+    reporting = _reporting_module()
+    return reporting.build_benchmark_report(
+        _config(model=marker),
+        [_run()],
+        mock=True,
+    )
+
+
+def _process_report_writer(
+    output_dir: str,
+    marker: str,
+    timestamp: str,
+    first_temp_barrier,
+    result_queue,
+) -> None:
+    reporting = _reporting_module()
+    original_write_text = reporting._write_text
+    first_write = True
+
+    def synchronized_write(path: Path, content: str) -> None:
+        nonlocal first_write
+        original_write_text(path, content)
+        if first_write:
+            first_write = False
+            first_temp_barrier.wait()
+
+    reporting._write_text = synchronized_write
+    try:
+        json_path, markdown_path = reporting.write_benchmark_report(
+            _marker_report(marker),
+            output_dir,
+            timestamp=timestamp,
+        )
+    except BaseException as exc:
+        result_queue.put(("error", marker, repr(exc)))
+        return
+    result_queue.put(
+        ("ok", marker, json_path.name, markdown_path.name)
+    )
+
+
+def _assert_concurrent_report_pairs(
+    output_dir: Path,
+    markers: list[str],
+) -> None:
+    json_paths = sorted(output_dir.glob("*.json"))
+    markdown_paths = sorted(output_dir.glob("*.md"))
+    assert len(json_paths) == len(markers)
+    assert len(markdown_paths) == len(markers)
+    assert {path.stem for path in json_paths} == {
+        path.stem for path in markdown_paths
+    }
+    actual_markers = set()
+    for json_path in json_paths:
+        marker = json.loads(json_path.read_text())["target"]["model"]
+        markdown = json_path.with_suffix(".md").read_text()
+        actual_markers.add(marker)
+        assert f"- Model: `{marker}`" in markdown
+    assert actual_markers == set(markers)
+    assert list(output_dir.glob("*.lock")) == []
+    assert list(output_dir.glob("*.tmp")) == []
+
+
 @pytest.mark.parametrize(
     ("values", "quantile", "expected"),
     [
@@ -131,6 +216,7 @@ def test_report_models_are_strict_and_include_required_summary_fields():
         "passed": 1,
         "failed": 0,
         "warnings": report.overall.warnings,
+        "system_errors": [],
     }
     with pytest.raises(ValidationError):
         reporting.BenchmarkReport.model_validate(
@@ -229,6 +315,83 @@ def test_aggregate_excludes_failed_and_null_samples_from_metrics():
     }
 
 
+def test_aggregate_flattens_passed_model_and_tool_latency_samples():
+    reporting = _reporting_module()
+    runs = [
+        _run(
+            iteration=1,
+            model_calls=[
+                _model_call(1, 10),
+                _model_call(2, None),
+                _model_call(3, 30),
+            ],
+            tool_latencies_ms=[5, None, 15],
+        ),
+        _run(
+            iteration=2,
+            model_calls=[_model_call(1, 20)],
+            tool_latencies_ms=[25],
+        ),
+        _run(
+            iteration=3,
+            status="failed",
+            model_calls=[_model_call(1, 999)],
+            tool_latencies_ms=[999],
+        ),
+    ]
+
+    report = reporting.build_benchmark_report(
+        _config(runs_per_case=3), runs, mock=True
+    )
+    aggregate = report.aggregates[0]
+    markdown = reporting.render_markdown(report)
+
+    assert aggregate.model_elapsed_ms.model_dump() == {
+        "count": 3,
+        "p50": 20.0,
+        "p95": 29.0,
+    }
+    assert aggregate.tool_latency_ms.model_dump() == {
+        "count": 3,
+        "p50": 15.0,
+        "p95": 24.0,
+    }
+    assert "Model elapsed p50/p95 ms" in markdown
+    assert "Tool latency p50/p95 ms" in markdown
+    assert "20.0/29.0" in markdown
+    assert "15.0/24.0" in markdown
+
+
+def test_zero_latency_samples_emit_low_confidence_warnings():
+    reporting = _reporting_module()
+
+    aggregate = reporting.build_benchmark_report(
+        _config(),
+        [
+            _run(
+                initial_provider_ttft_ms=None,
+                visible_ttft_ms=None,
+                turn_wall_time_ms=None,
+                model_calls=[],
+                tool_latencies_ms=[],
+            )
+        ],
+        mock=True,
+    ).aggregates[0]
+
+    assert {
+        warning.split(" p95 low confidence:", 1)[0]
+        for warning in aggregate.warnings
+    } == {
+        "ordinary_chat/dual_agent initial_provider_ttft_ms",
+        "ordinary_chat/dual_agent visible_ttft_ms",
+        "ordinary_chat/dual_agent turn_wall_time_ms",
+        "ordinary_chat/dual_agent model_elapsed_ms",
+        "ordinary_chat/dual_agent tool_latency_ms",
+    }
+    assert all("0 samples" in warning for warning in aggregate.warnings)
+
+
 def test_aggregates_are_stably_sorted_and_warn_for_low_confidence_p95():
     reporting = _reporting_module()
     runs = [
@@ -312,6 +475,21 @@ def test_errors_are_redacted_before_truncation_in_json_and_markdown():
     assert "[REDACTED]" in markdown
 
 
+def test_error_truncation_is_utf8_safe_and_bounded_after_redaction():
+    reporting = _reporting_module()
+    real_key = "sk-byte-bound-secret"
+    error = f"Bearer {real_key} " + "密" * 1000
+
+    sanitized = reporting.sanitize_error(error, secrets=[real_key])
+    redacted = reporting.redact_text(error, secrets=[real_key])
+    expected = redacted.encode("utf-8")[:2048].decode("utf-8", errors="ignore")
+
+    assert sanitized == expected
+    assert real_key not in sanitized
+    assert len(sanitized.encode("utf-8")) <= 2048
+    assert "\ufffd" not in sanitized
+
+
 def test_configured_secret_is_redacted_from_all_report_string_surfaces():
     reporting = _reporting_module()
     real_key = "sk-everywhere-secret"
@@ -338,6 +516,34 @@ def test_configured_secret_is_redacted_from_all_report_string_surfaces():
     assert "[REDACTED]" in rendered
 
 
+def test_system_errors_are_typed_redacted_counted_and_rendered():
+    reporting = _reporting_module()
+    real_key = "sk-runner-system-secret"
+
+    report = reporting.build_benchmark_report(
+        _config(),
+        [],
+        mock=False,
+        secrets=[real_key],
+        system_errors=[f"Authorization: Bearer {real_key}"],
+    )
+    payload = json.loads(reporting.report_to_json(report))
+    markdown = reporting.render_markdown(report)
+
+    assert report.overall.model_dump() == {
+        "attempted": 1,
+        "passed": 0,
+        "failed": 1,
+        "warnings": [],
+        "system_errors": ["Authorization: [REDACTED] [REDACTED]"],
+    }
+    assert payload["overall"]["system_errors"] == report.overall.system_errors
+    assert "## System Errors" in markdown
+    assert "Authorization: [REDACTED] [REDACTED]" in markdown
+    assert real_key not in reporting.report_to_json(report)
+    assert real_key not in markdown
+
+
 def test_markdown_is_rendered_from_the_same_json_facts():
     reporting = _reporting_module()
     report = reporting.build_benchmark_report(
@@ -412,3 +618,99 @@ def test_atomic_write_removes_temps_and_published_half_on_replace_failure(
         )
 
     assert list(tmp_path.iterdir()) == []
+
+
+def test_reservation_failure_after_lock_creation_removes_own_lock(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+):
+    reporting = _reporting_module()
+    original_exists = Path.exists
+
+    def failing_exists(path: Path) -> bool:
+        if path.parent == tmp_path and path.suffix == ".json":
+            raise OSError("simulated reservation path check failure")
+        return original_exists(path)
+
+    monkeypatch.setattr(Path, "exists", failing_exists)
+
+    with pytest.raises(
+        reporting.BenchmarkReportWriteError,
+        match="failed to write benchmark reports",
+    ):
+        reporting.write_benchmark_report(
+            _marker_report("reservation-failure"),
+            tmp_path,
+            timestamp="20260714T040506Z",
+        )
+
+    assert list(tmp_path.iterdir()) == []
+
+
+def test_concurrent_thread_writers_reserve_distinct_matching_pairs(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+):
+    reporting = _reporting_module()
+    writer_count = 8
+    markers = [f"thread-writer-{index}" for index in range(writer_count)]
+    first_temp_barrier = threading.Barrier(writer_count, timeout=15)
+    original_write_text = reporting._write_text
+    thread_state = threading.local()
+
+    def synchronized_write(path: Path, content: str) -> None:
+        original_write_text(path, content)
+        if not getattr(thread_state, "first_write_done", False):
+            thread_state.first_write_done = True
+            first_temp_barrier.wait()
+
+    monkeypatch.setattr(reporting, "_write_text", synchronized_write)
+
+    with ThreadPoolExecutor(max_workers=writer_count) as executor:
+        results = list(
+            executor.map(
+                lambda marker: reporting.write_benchmark_report(
+                    _marker_report(marker),
+                    tmp_path,
+                    timestamp="20260714T020304Z",
+                ),
+                markers,
+            )
+        )
+
+    assert len({json_path.stem for json_path, _ in results}) == writer_count
+    _assert_concurrent_report_pairs(tmp_path, markers)
+
+
+def test_concurrent_process_writers_reserve_distinct_matching_pairs(
+    tmp_path: Path,
+):
+    context = multiprocessing.get_context("spawn")
+    writer_count = 4
+    markers = [f"process-writer-{index}" for index in range(writer_count)]
+    first_temp_barrier = context.Barrier(writer_count, timeout=20)
+    result_queue = context.Queue()
+    processes = [
+        context.Process(
+            target=_process_report_writer,
+            args=(
+                str(tmp_path),
+                marker,
+                "20260714T030405Z",
+                first_temp_barrier,
+                result_queue,
+            ),
+        )
+        for marker in markers
+    ]
+
+    for process in processes:
+        process.start()
+    results = [result_queue.get(timeout=30) for _ in processes]
+    for process in processes:
+        process.join(timeout=30)
+
+    assert all(process.exitcode == 0 for process in processes)
+    assert all(result[0] == "ok" for result in results), results
+    assert len({result[2][:-5] for result in results}) == writer_count
+    _assert_concurrent_report_pairs(tmp_path, markers)