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) -> 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.parent.name, markdown_path.parent.name)
    )


def _assert_concurrent_report_pairs(
    output_dir: Path,
    markers: list[str],
) -> None:
    published_dirs = sorted(
        path
        for path in output_dir.iterdir()
        if not path.name.startswith(".")
    )
    assert len(published_dirs) == len(markers)
    assert all(path.is_symlink() for path in published_dirs)
    actual_markers = set()
    targets = set()
    for published_dir in published_dirs:
        target = published_dir.readlink()
        assert not target.is_absolute()
        targets.add(target)
        json_path = published_dir / "report.json"
        markdown_path = published_dir / "report.md"
        marker = json.loads(json_path.read_text())["target"]["model"]
        markdown = markdown_path.read_text()
        actual_markers.add(marker)
        assert f"- Model: `{marker}`" in markdown
    assert actual_markers == set(markers)
    assert len(targets) == len(markers)
    _assert_no_publication_artifacts(output_dir)


def _assert_no_publication_artifacts(output_dir: Path) -> None:
    entries = list(output_dir.iterdir())
    leftovers = [
        path.name
        for path in entries
        if path.name.endswith((".tmp", ".lock"))
        or "placeholder" in path.name
        or path.suffix in {".json", ".md"}
    ]
    assert leftovers == []
    published_dirs = [
        path
        for path in entries
        if path.is_symlink()
        and path.readlink().name.startswith(".")
        and path.readlink().name.endswith(".bundle")
    ]
    published_targets = {path.readlink() for path in published_dirs}
    private_bundles = {
        Path(path.name)
        for path in entries
        if path.name.startswith(".") and path.is_dir()
    }
    assert private_bundles == published_targets
    for published_dir in published_dirs:
        assert (published_dir / "report.json").stat().st_size > 0
        assert (published_dir / "report.md").stat().st_size > 0


@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_markdown_preserves_runtime_error_and_semantic_failures_together():
    reporting = _reporting_module()
    report = reporting.build_benchmark_report(
        _config(),
        [
            _run(
                status="failed",
                error="runtime exploded",
                semantic_failures=[
                    "expected web_search event",
                    "expected two visible answers",
                ],
            )
        ],
        mock=True,
    )

    markdown = reporting.render_markdown(report)

    assert "runtime exploded" in markdown
    assert "expected web_search event" in markdown
    assert "expected two visible answers" in markdown


def test_atomic_write_publishes_complete_bundle_through_relative_symlink(
    tmp_path: Path,
):
    reporting = _reporting_module()
    report = _marker_report("atomic-bundle")
    timestamp = "20260714T010203Z"
    competitor = tmp_path / timestamp
    competitor.mkdir()
    (competitor / "sentinel").write_text("competitor")

    json_path, markdown_path = reporting.write_benchmark_report(
        report,
        tmp_path,
        timestamp=timestamp,
    )

    published_dir = tmp_path / f"{timestamp}-1"
    assert json_path == published_dir / "report.json"
    assert markdown_path == published_dir / "report.md"
    assert published_dir.is_symlink()
    bundle_target = published_dir.readlink()
    assert not bundle_target.is_absolute()
    assert bundle_target.name.startswith(f".{timestamp}.")
    assert bundle_target.name.endswith(".bundle")
    assert competitor.joinpath("sentinel").read_text() == "competitor"
    assert json.loads(json_path.read_text()) == report.model_dump(mode="json")
    assert markdown_path.read_text() == reporting.render_markdown(report)
    _assert_no_publication_artifacts(tmp_path)


@pytest.mark.parametrize("collision_kind", ["create", "replace"])
def test_symlink_collision_retries_without_removing_or_replacing_competitor(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    collision_kind: str,
):
    reporting = _reporting_module()
    timestamp = "20260714T011223Z"
    public_path = tmp_path / timestamp
    real_symlink = reporting.os.symlink
    real_replace = reporting.os.replace
    symlink_calls = 0

    if collision_kind == "replace":
        old_target = tmp_path / "old-competitor"
        old_target.mkdir()
        real_symlink(old_target.name, public_path)

    def collide_before_symlink(source: str, destination: Path) -> None:
        nonlocal symlink_calls
        symlink_calls += 1
        destination = Path(destination)
        if symlink_calls == 1:
            if collision_kind == "create":
                destination.mkdir()
                (destination / "sentinel").write_text("competitor")
            else:
                new_target = tmp_path / "new-competitor"
                new_target.mkdir()
                replacement = tmp_path / "competitor-replacement"
                real_symlink(new_target.name, replacement)
                real_replace(replacement, destination)
        real_symlink(source, destination)

    monkeypatch.setattr(reporting.os, "symlink", collide_before_symlink)

    json_path, markdown_path = reporting.write_benchmark_report(
        _marker_report(f"symlink-{collision_kind}-collision"),
        tmp_path,
        timestamp=timestamp,
    )

    assert symlink_calls == 2
    assert json_path.parent == tmp_path / f"{timestamp}-1"
    assert markdown_path.parent == json_path.parent
    if collision_kind == "create":
        assert public_path.joinpath("sentinel").read_text() == "competitor"
    else:
        assert public_path.readlink() == Path("new-competitor")
    assert json.loads(json_path.read_text())["target"]["model"] == (
        f"symlink-{collision_kind}-collision"
    )
    _assert_no_publication_artifacts(tmp_path)


def test_bundle_is_complete_and_matching_before_single_publication(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
):
    reporting = _reporting_module()
    real_symlink = reporting.os.symlink
    publication_snapshots: list[tuple[str, str]] = []

    def inspect_complete_bundle(source: str, destination: Path) -> None:
        destination = Path(destination)
        bundle = destination.parent / source
        json_text = bundle.joinpath("report.json").read_text()
        markdown_text = bundle.joinpath("report.md").read_text()
        marker = json.loads(json_text)["target"]["model"]
        assert f"- Model: `{marker}`" in markdown_text
        publication_snapshots.append((json_text, markdown_text))
        real_symlink(source, destination)

    monkeypatch.setattr(reporting.os, "symlink", inspect_complete_bundle)

    json_path, markdown_path = reporting.write_benchmark_report(
        _marker_report("complete-before-publish"),
        tmp_path,
        timestamp="20260714T011224Z",
    )

    assert len(publication_snapshots) == 1
    assert json_path.parent == markdown_path.parent
    assert json_path.parent.is_symlink()
    assert json_path.read_text() == publication_snapshots[0][0]
    assert markdown_path.read_text() == publication_snapshots[0][1]
    _assert_no_publication_artifacts(tmp_path)


def test_interrupt_after_symlink_creation_keeps_resolvable_complete_bundle(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
):
    reporting = _reporting_module()
    real_symlink = reporting.os.symlink
    timestamp = "20260714T011224Z-interrupted"

    def publish_then_interrupt(source: str, destination: Path) -> None:
        real_symlink(source, destination)
        raise KeyboardInterrupt

    monkeypatch.setattr(reporting.os, "symlink", publish_then_interrupt)

    with pytest.raises(KeyboardInterrupt):
        reporting.write_benchmark_report(
            _marker_report("interrupt-after-symlink"),
            tmp_path,
            timestamp=timestamp,
        )

    published_dir = tmp_path / timestamp
    assert published_dir.is_symlink()
    assert published_dir.resolve(strict=True).is_dir()
    assert (
        json.loads(published_dir.joinpath("report.json").read_text())["target"][
            "model"
        ]
        == "interrupt-after-symlink"
    )
    assert "- Model: `interrupt-after-symlink`" in (
        published_dir.joinpath("report.md").read_text()
    )
    _assert_no_publication_artifacts(tmp_path)


@pytest.mark.parametrize(
    "raw_timestamp",
    [
        pytest.param("<absolute>", id="absolute-path"),
        pytest.param("../outside-report", id="parent-traversal"),
        pytest.param("segment/report", id="forward-slash"),
        pytest.param(r"segment\report", id="backslash"),
        pytest.param(".", id="dot-segment"),
        pytest.param("..", id="double-dot-segment"),
        pytest.param("", id="empty"),
        pytest.param(" ", id="space"),
        pytest.param("\t", id="tab"),
        pytest.param("2026\n0714", id="control-newline"),
        pytest.param("2026\x00714", id="control-nul"),
        pytest.param("-20260714", id="non-alnum-prefix-hyphen"),
        pytest.param("_20260714", id="non-alnum-prefix-underscore"),
        pytest.param("2026\u00a00714", id="unicode-whitespace"),
        pytest.param("2026\uff0f0714", id="unicode-fullwidth-slash"),
        pytest.param("2026\u22150714", id="unicode-division-slash"),
        pytest.param("２０２６０７１４", id="unicode-fullwidth-digits"),
        pytest.param("2026\u202ejson", id="unicode-bidi-control"),
        pytest.param("a" * 129, id="too-long"),
    ],
)
def test_invalid_explicit_timestamp_is_rejected_before_any_filesystem_write(
    tmp_path: Path,
    raw_timestamp: str,
):
    reporting = _reporting_module()
    output_dir = tmp_path / "reports"
    timestamp = (
        str(tmp_path / "outside-absolute")
        if raw_timestamp == "<absolute>"
        else raw_timestamp
    )
    entries_before = sorted(
        path.relative_to(tmp_path) for path in tmp_path.rglob("*")
    )

    with pytest.raises(
        reporting.BenchmarkReportWriteError,
        match="invalid benchmark report timestamp",
    ):
        reporting.write_benchmark_report(
            _marker_report("invalid-timestamp"),
            output_dir,
            timestamp=timestamp,
        )

    entries_after = sorted(
        path.relative_to(tmp_path) for path in tmp_path.rglob("*")
    )
    assert entries_after == entries_before
    assert not output_dir.exists()


def test_report_files_and_bundle_directory_are_fsynced_before_publication(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
):
    reporting = _reporting_module()
    real_fsync = reporting.os.fsync
    real_symlink = reporting.os.symlink
    fsynced_objects: list[tuple[int, int]] = []
    symlink_calls = 0

    def record_fsync(file_descriptor: int) -> None:
        file_stat = reporting.os.fstat(file_descriptor)
        fsynced_objects.append((file_stat.st_dev, file_stat.st_ino))
        real_fsync(file_descriptor)

    def inspect_before_symlink(source: str, destination: Path) -> None:
        nonlocal symlink_calls
        symlink_calls += 1
        destination = Path(destination)
        bundle_dir = destination.parent / source
        expected_before_publication = {
            (
                bundle_dir.joinpath(name).stat().st_dev,
                bundle_dir.joinpath(name).stat().st_ino,
            )
            for name in ("report.json", "report.md")
        }
        bundle_stat = bundle_dir.stat()
        expected_before_publication.add((bundle_stat.st_dev, bundle_stat.st_ino))
        assert set(fsynced_objects) == expected_before_publication
        assert len(fsynced_objects) == 3
        real_symlink(source, destination)

    monkeypatch.setattr(reporting.os, "fsync", record_fsync)
    monkeypatch.setattr(reporting.os, "symlink", inspect_before_symlink)

    reporting.write_benchmark_report(
        _marker_report("fsync-before-publish"),
        tmp_path,
        timestamp="20260714T011225Z",
    )

    assert symlink_calls == 1
    output_stat = tmp_path.stat()
    assert (output_stat.st_dev, output_stat.st_ino) in fsynced_objects
    assert len(fsynced_objects) == 4
    assert len(set(fsynced_objects)) == 4
    _assert_no_publication_artifacts(tmp_path)


def test_file_write_failure_removes_only_unpublished_private_bundle(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
):
    reporting = _reporting_module()
    original_write_text = reporting._write_text
    write_calls = 0

    def fail_second_file(path: Path, content: str) -> None:
        nonlocal write_calls
        write_calls += 1
        if write_calls == 2:
            raise OSError("simulated report file failure")
        original_write_text(path, content)

    monkeypatch.setattr(reporting, "_write_text", fail_second_file)

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

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


def test_bundle_fsync_failure_removes_unpublished_private_bundle(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
):
    reporting = _reporting_module()
    original_fsync_directory = reporting._fsync_directory

    def fail_bundle_fsync(path: Path) -> None:
        if path.name.endswith(".bundle"):
            raise OSError("simulated bundle fsync failure")
        original_fsync_directory(path)

    monkeypatch.setattr(reporting, "_fsync_directory", fail_bundle_fsync)

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

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


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

    def fail_symlink(source: str, destination: Path) -> None:
        del source, destination
        raise OSError("simulated symlink failure")

    monkeypatch.setattr(reporting.os, "symlink", fail_symlink)

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

    entries = list(tmp_path.iterdir())
    assert len(entries) == 1
    bundle_dir = entries[0]
    assert bundle_dir.name.startswith(".20260714T011228Z.")
    assert bundle_dir.name.endswith(".bundle")
    assert json.loads(bundle_dir.joinpath("report.json").read_text())["target"][
        "model"
    ] == "symlink-failure"
    assert "- Model: `symlink-failure`" in bundle_dir.joinpath(
        "report.md"
    ).read_text()


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

    def interrupt_symlink(source: str, destination: Path) -> None:
        del source, destination
        raise KeyboardInterrupt

    monkeypatch.setattr(reporting.os, "symlink", interrupt_symlink)

    with pytest.raises(KeyboardInterrupt):
        reporting.write_benchmark_report(
            _marker_report("symlink-interrupt"),
            tmp_path,
            timestamp="20260714T011229Z",
        )

    entries = list(tmp_path.iterdir())
    assert len(entries) == 1
    bundle_dir = entries[0]
    assert bundle_dir.name.startswith(".20260714T011229Z.")
    assert bundle_dir.name.endswith(".bundle")
    assert json.loads(bundle_dir.joinpath("report.json").read_text())["target"][
        "model"
    ] == "symlink-interrupt"
    assert "- Model: `symlink-interrupt`" in bundle_dir.joinpath(
        "report.md"
    ).read_text()


def test_post_publication_directory_fsync_failure_keeps_public_bundle(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
):
    reporting = _reporting_module()
    original_fsync_directory = reporting._fsync_directory
    fsync_calls = 0

    def fail_output_directory_fsync(path: Path) -> None:
        nonlocal fsync_calls
        fsync_calls += 1
        if fsync_calls == 2:
            raise OSError("simulated output directory fsync failure")
        original_fsync_directory(path)

    monkeypatch.setattr(
        reporting,
        "_fsync_directory",
        fail_output_directory_fsync,
    )

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

    published_dir = tmp_path / "20260714T011230Z"
    assert published_dir.is_symlink()
    assert (
        json.loads(published_dir.joinpath("report.json").read_text())["target"]["model"]
        == "published-before-fsync-failure"
    )
    assert "- Model: `published-before-fsync-failure`" in (
        published_dir.joinpath("report.md").read_text()
    )
    _assert_no_publication_artifacts(tmp_path)


@pytest.mark.parametrize("stress_round", range(3))
def test_concurrent_thread_writers_publish_distinct_matching_bundles(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    stress_round: int,
):
    reporting = _reporting_module()
    writer_count = 8
    markers = [
        f"thread-{stress_round}-writer-{index}"
        for index in range(writer_count)
    ]
    first_file_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_file_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=f"20260714T02030{stress_round}Z",
                ),
                markers,
            )
        )

    assert len({json_path.parent for json_path, _ in results}) == writer_count
    assert all(
        json_path.parent == markdown_path.parent
        for json_path, markdown_path in results
    )
    _assert_concurrent_report_pairs(tmp_path, markers)


@pytest.mark.parametrize("stress_round", range(2))
def test_concurrent_process_writers_publish_distinct_matching_bundles(
    tmp_path: Path,
    stress_round: int,
):
    context = multiprocessing.get_context("spawn")
    writer_count = 4
    markers = [
        f"process-{stress_round}-writer-{index}"
        for index in range(writer_count)
    ]
    first_file_barrier = context.Barrier(writer_count, timeout=20)
    result_queue = context.Queue()
    processes = [
        context.Process(
            target=_process_report_writer,
            args=(
                str(tmp_path),
                marker,
                f"20260714T03040{stress_round}Z",
                first_file_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 all(result[2] == result[3] for result in results)
    assert len({result[2] for result in results}) == writer_count
    _assert_concurrent_report_pairs(tmp_path, markers)