agent-lab/tests/test_benchmark_reporting.py

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

import pytest
from pydantic import ValidationError

from agent_lab.application.benchmark import (
    BenchmarkCaseId,
    BenchmarkConfig,
    BenchmarkModelCallTiming,
    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,
    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,
        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=tool_latencies_ms or [],
        semantic_failures=semantic_failures or [],
        error=error,
        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) -> tuple[int, int]:
        nonlocal first_write
        identity = original_write_text(path, content)
        if first_write:
            first_write = False
            first_temp_barrier.wait()
        return identity

    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"),
    [
        ([], 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,
        "system_errors": [],
    }
    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_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 = [
        _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_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"
    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_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(
        _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_second_placeholder_collision_cleans_first_and_uses_suffix(
    tmp_path: Path,
):
    reporting = _reporting_module()
    timestamp = "20260714T011223Z"
    external_markdown = tmp_path / f"{timestamp}.md"
    external_markdown.write_text("external markdown")

    json_path, markdown_path = reporting.write_benchmark_report(
        _marker_report("second-placeholder-collision"),
        tmp_path,
        timestamp=timestamp,
    )

    assert external_markdown.read_text() == "external markdown"
    assert not (tmp_path / f"{timestamp}.json").exists()
    assert json_path.name == f"{timestamp}-1.json"
    assert markdown_path.name == f"{timestamp}-1.md"
    assert list(tmp_path.glob("*.lock")) == []


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()) == []


def test_temp_fsync_failure_removes_owned_temp_and_placeholders(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
):
    reporting = _reporting_module()

    def failing_fsync(file_descriptor: int) -> None:
        del file_descriptor
        raise OSError("simulated temp fsync failure")

    monkeypatch.setattr(reporting.os, "fsync", failing_fsync)

    with pytest.raises(
        reporting.BenchmarkReportWriteError,
        match="failed to write benchmark reports",
    ):
        reporting.write_benchmark_report(
            _marker_report("temp-fsync-failure"),
            tmp_path,
            timestamp="20260714T012334Z",
        )

    assert list(tmp_path.iterdir()) == []


def test_second_placeholder_failure_removes_first_placeholder(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
):
    reporting = _reporting_module()
    original_open = reporting.os.open
    timestamp = "20260714T040506Z"
    markdown_path = tmp_path / f"{timestamp}.md"

    def failing_open(path, flags, mode=0o777):
        if Path(path) == markdown_path:
            raise OSError("simulated second placeholder failure")
        return original_open(path, flags, mode)

    monkeypatch.setattr(reporting.os, "open", failing_open)

    with pytest.raises(
        reporting.BenchmarkReportWriteError,
        match="failed to write benchmark reports",
    ):
        reporting.write_benchmark_report(
            _marker_report("reservation-failure"),
            tmp_path,
            timestamp=timestamp,
        )

    assert list(tmp_path.iterdir()) == []


def test_reservation_uses_final_placeholders_without_sidecar_lock(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
):
    reporting = _reporting_module()
    timestamp = "20260714T050607Z"
    first_temp_written = threading.Event()
    allow_publish = threading.Event()
    original_write_text = reporting._write_text
    first_write = True

    def paused_write(path: Path, content: str) -> tuple[int, int]:
        nonlocal first_write
        identity = original_write_text(path, content)
        if first_write:
            first_write = False
            first_temp_written.set()
            assert allow_publish.wait(timeout=10)
        return identity

    monkeypatch.setattr(reporting, "_write_text", paused_write)
    with ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(
            reporting.write_benchmark_report,
            _marker_report("placeholder-owner"),
            tmp_path,
            timestamp=timestamp,
        )
        assert first_temp_written.wait(timeout=10)
        json_placeholder = tmp_path / f"{timestamp}.json"
        markdown_placeholder = tmp_path / f"{timestamp}.md"
        try:
            assert json_placeholder.exists()
            assert markdown_placeholder.exists()
            assert json_placeholder.read_bytes() == b""
            assert markdown_placeholder.read_bytes() == b""
            assert list(tmp_path.glob("*.lock")) == []
        finally:
            allow_publish.set()
        json_path, markdown_path = future.result(timeout=10)

    assert json_path == json_placeholder
    assert markdown_path == markdown_placeholder


@pytest.mark.parametrize("competed_suffix", [".json", ".md"])
def test_external_replacement_of_reserved_final_is_not_overwritten_or_deleted(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    competed_suffix: str,
):
    reporting = _reporting_module()
    timestamp = "20260714T060708Z"
    first_temp_written = threading.Event()
    allow_publish = threading.Event()
    original_write_text = reporting._write_text
    first_write = True

    def paused_write(path: Path, content: str) -> tuple[int, int]:
        nonlocal first_write
        identity = original_write_text(path, content)
        if first_write:
            first_write = False
            first_temp_written.set()
            assert allow_publish.wait(timeout=10)
        return identity

    monkeypatch.setattr(reporting, "_write_text", paused_write)
    competed_path = tmp_path / f"{timestamp}{competed_suffix}"
    competitor_temp = tmp_path / f"competitor{competed_suffix}"
    competitor_content = b"external competitor content"

    with ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(
            reporting.write_benchmark_report,
            _marker_report("must-not-overwrite"),
            tmp_path,
            timestamp=timestamp,
        )
        assert first_temp_written.wait(timeout=10)
        competitor_temp.write_bytes(competitor_content)
        reporting.os.replace(competitor_temp, competed_path)
        allow_publish.set()
        with pytest.raises(
            reporting.BenchmarkReportWriteError,
            match="failed to write benchmark reports",
        ):
            future.result(timeout=10)

    assert competed_path.read_bytes() == competitor_content
    other_suffix = ".md" if competed_suffix == ".json" else ".json"
    assert not (tmp_path / f"{timestamp}{other_suffix}").exists()
    assert list(tmp_path.glob("*.lock")) == []
    assert list(tmp_path.glob("*.tmp")) == []


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) -> tuple[int, int]:
        identity = original_write_text(path, content)
        if not getattr(thread_state, "first_write_done", False):
            thread_state.first_write_done = True
            first_temp_barrier.wait()
        return identity

    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)