fix: publish atomic benchmark report bundles

Problem: Flat JSON and Markdown publication required a racy check-then-unlink rollback that could delete a competitor file or expose a mixed report pair. Markdown also hid semantic failures when a runtime error was present.

Risk: Report artifacts now live under timestamp-named symlinked bundle directories, so consumers that scan root-level JSON or Markdown files must adopt the bundle paths. Mutation by another output-directory writer after successful publication remains outside the atomicity guarantee.

src/agent_lab/infrastructure/benchmark_reporting.py

@@ -3,12 +3,13 @@ from __future__ import annotations
 import json
 import os
 import re
+import shutil
+import tempfile
 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
 
@@ -461,7 +462,12 @@ def render_markdown(report: BenchmarkReport) -> str:
     )
     for run in payload["runs"]:
         failure_text = "; ".join(run["semantic_failures"])
-        error_text = run["error"] or failure_text or "-"
+        error_parts: list[str] = []
+        if run["error"]:
+            error_parts.append(str(run["error"]))
+        if failure_text:
+            error_parts.append(f"semantic failures: {failure_text}")
+        error_text = "; ".join(error_parts) or "-"
         lines.append(
             "| "
             + " | ".join(
@@ -501,49 +507,42 @@ def write_benchmark_report(
     timestamp: str | None = None,
 ) -> tuple[Path, Path]:
     directory = Path(output_dir)
-    owned_paths: dict[Path, tuple[int, int]] = {}
+    bundle_dir: Path | None = None
+    published = False
     try:
         directory.mkdir(parents=True, exist_ok=True)
         stem = timestamp or report.generated_at.astimezone(timezone.utc).strftime(
             "%Y%m%dT%H%M%SZ"
         )
-        nonce = uuid4().hex
-        json_temp = directory / f".{stem}.{nonce}.json.tmp"
-        markdown_temp = directory / f".{stem}.{nonce}.md.tmp"
-        json_temp_identity = _write_text(json_temp, report_to_json(report))
-        owned_paths[json_temp] = json_temp_identity
-        markdown_temp_identity = _write_text(
-            markdown_temp,
-            render_markdown(report),
+        bundle_dir = Path(
+            tempfile.mkdtemp(
+                prefix=f".{stem}.",
+                suffix=".bundle",
+                dir=directory,
+            )
         )
-        owned_paths[markdown_temp] = markdown_temp_identity
+        _write_text(bundle_dir / "report.json", report_to_json(report))
+        _write_text(bundle_dir / "report.md", render_markdown(report))
+        _fsync_directory(bundle_dir)
+
         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"
+            published_dir = directory / candidate
             try:
-                os.link(json_temp, json_path)
+                os.symlink(bundle_dir.name, published_dir)
             except FileExistsError:
                 suffix += 1
                 continue
-            owned_paths[json_path] = json_temp_identity
-            try:
-                os.link(markdown_temp, markdown_path)
-            except FileExistsError:
-                _unlink_owned(json_path, json_temp_identity)
-                owned_paths.pop(json_path, None)
-                suffix += 1
-                continue
-            owned_paths[markdown_path] = markdown_temp_identity
-            _unlink_owned(json_temp, json_temp_identity)
-            owned_paths.pop(json_temp, None)
-            _unlink_owned(markdown_temp, markdown_temp_identity)
-            owned_paths.pop(markdown_temp, None)
-            return json_path, markdown_path
+            published = True
+            _fsync_directory(directory)
+            return (
+                published_dir / "report.json",
+                published_dir / "report.md",
+            )
     except BaseException as exc:
-        for path, identity in owned_paths.items():
-            _unlink_if_owned(path, identity)
+        if bundle_dir is not None and not published:
+            shutil.rmtree(bundle_dir, ignore_errors=True)
         if not isinstance(exc, Exception):
             raise
         raise BenchmarkReportWriteError(
@@ -551,38 +550,17 @@ def write_benchmark_report(
         ) from exc
 
 
-def _write_text(path: Path, content: str) -> tuple[int, int]:
-    identity: tuple[int, int] | None = None
-    try:
-        with path.open("x", encoding="utf-8") as handle:
-            file_stat = os.fstat(handle.fileno())
-            identity = file_stat.st_dev, file_stat.st_ino
-            handle.write(content)
-            handle.flush()
-            os.fsync(handle.fileno())
-        return identity
-    except BaseException:
-        if identity is not None:
-            _unlink_if_owned(path, identity)
-        raise
-
-
-def _path_identity(path: Path) -> tuple[int, int]:
-    path_stat = path.stat()
-    return path_stat.st_dev, path_stat.st_ino
-
-
-def _unlink_owned(path: Path, identity: tuple[int, int]) -> None:
-    try:
-        current_identity = _path_identity(path)
-    except FileNotFoundError:
-        return
-    if current_identity == identity:
-        path.unlink()
+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 _unlink_if_owned(path: Path, identity: tuple[int, int]) -> None:
+def _fsync_directory(path: Path) -> None:
+    flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
+    file_descriptor = os.open(path, flags)
     try:
-        _unlink_owned(path, identity)
-    except OSError:
-        pass
+        os.fsync(file_descriptor)
+    finally:
+        os.close(file_descriptor)

tests/test_benchmark_reporting.py

@@ -125,13 +125,12 @@ def _process_report_writer(
     original_write_text = reporting._write_text
     first_write = True
 
-    def synchronized_write(path: Path, content: str) -> tuple[int, int]:
+    def synchronized_write(path: Path, content: str) -> None:
         nonlocal first_write
-        identity = original_write_text(path, content)
+        original_write_text(path, content)
         if first_write:
             first_write = False
             first_temp_barrier.wait()
-        return identity
 
     reporting._write_text = synchronized_write
     try:
@@ -144,7 +143,7 @@ def _process_report_writer(
         result_queue.put(("error", marker, repr(exc)))
         return
     result_queue.put(
-        ("ok", marker, json_path.name, markdown_path.name)
+        ("ok", marker, json_path.parent.name, markdown_path.parent.name)
     )
 
 
@@ -152,37 +151,57 @@ 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
-    }
+    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()
-    for json_path in json_paths:
+    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 = json_path.with_suffix(".md").read_text()
+        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 output_dir.iterdir()
+        for path in entries
         if path.name.endswith((".tmp", ".lock"))
         or "placeholder" in path.name
+        or path.suffix in {".json", ".md"}
     ]
     assert leftovers == []
-    report_paths = [
+    published_dirs = [
         path
-        for path in output_dir.iterdir()
-        if path.suffix in {".json", ".md"}
+        for path in entries
+        if path.is_symlink()
+        and path.readlink().name.startswith(".")
+        and path.readlink().name.endswith(".bundle")
     ]
-    assert all(path.stat().st_size > 0 for path in report_paths)
+    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(
@@ -582,14 +601,39 @@ def test_markdown_is_rendered_from_the_same_json_facts():
     assert payload == report.model_dump(mode="json")
 
 
-def test_atomic_write_creates_matching_unique_json_and_markdown_paths(tmp_path: Path):
+def test_markdown_preserves_runtime_error_and_semantic_failures_together():
     reporting = _reporting_module()
     report = reporting.build_benchmark_report(
-        _config(), [_run()], mock=True
+        _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"
-    (tmp_path / f"{timestamp}.json").write_text("existing")
-    (tmp_path / f"{timestamp}.md").write_text("existing")
+    competitor = tmp_path / timestamp
+    competitor.mkdir()
+    (competitor / "sentinel").write_text("competitor")
 
     json_path, markdown_path = reporting.write_benchmark_report(
         report,
@@ -597,229 +641,219 @@ def test_atomic_write_creates_matching_unique_json_and_markdown_paths(tmp_path:
         timestamp=timestamp,
     )
 
-    assert json_path.name == f"{timestamp}-1.json"
-    assert markdown_path.name == f"{timestamp}-1.md"
+    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)
 
 
-def test_json_created_immediately_before_link_wins_and_writer_retries(
+@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"
-    external_json = tmp_path / f"{timestamp}.json"
-    real_link = reporting.os.link
-    link_calls = 0
+    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 create_before_link(source: Path, destination: Path) -> None:
-        nonlocal link_calls
-        link_calls += 1
+    def collide_before_symlink(source: str, destination: Path) -> None:
+        nonlocal symlink_calls
+        symlink_calls += 1
         destination = Path(destination)
-        if link_calls == 1:
-            destination.write_text("external json")
-        real_link(source, 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, "link", create_before_link)
+    monkeypatch.setattr(reporting.os, "symlink", collide_before_symlink)
 
     json_path, markdown_path = reporting.write_benchmark_report(
-        _marker_report("json-link-collision"),
+        _marker_report(f"symlink-{collision_kind}-collision"),
         tmp_path,
         timestamp=timestamp,
     )
 
-    assert link_calls == 3
-    assert external_json.read_text() == "external json"
-    assert not (tmp_path / f"{timestamp}.md").exists()
-    assert json_path.name == f"{timestamp}-1.json"
-    assert markdown_path.name == f"{timestamp}-1.md"
+    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_json_replaced_immediately_before_link_wins_and_writer_retries(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+def test_bundle_is_complete_and_matching_before_single_publication(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
 ):
     reporting = _reporting_module()
-    timestamp = "20260714T011224Z"
-    external_json = tmp_path / f"{timestamp}.json"
-    external_json.write_text("old external json")
-    replacement = tmp_path / "competitor-replacement.json"
-    real_link = reporting.os.link
-    real_replace = reporting.os.replace
-    link_calls = 0
+    real_symlink = reporting.os.symlink
+    publication_snapshots: list[tuple[str, str]] = []
 
-    def replace_before_link(source: Path, destination: Path) -> None:
-        nonlocal link_calls
-        link_calls += 1
+    def inspect_complete_bundle(source: str, destination: Path) -> None:
         destination = Path(destination)
-        if link_calls == 1:
-            replacement.write_text("new external json")
-            real_replace(replacement, destination)
-        real_link(source, 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, "link", replace_before_link)
+    monkeypatch.setattr(reporting.os, "symlink", inspect_complete_bundle)
 
     json_path, markdown_path = reporting.write_benchmark_report(
-        _marker_report("json-link-replacement"),
+        _marker_report("complete-before-publish"),
         tmp_path,
-        timestamp=timestamp,
+        timestamp="20260714T011224Z",
     )
 
-    assert link_calls == 3
-    assert external_json.read_text() == "new external json"
-    assert json_path.name == f"{timestamp}-1.json"
-    assert markdown_path.name == f"{timestamp}-1.md"
+    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)
 
 
-@pytest.mark.parametrize("collision_kind", ["create", "replace"])
-def test_markdown_link_conflict_removes_owned_json_link_and_retries(
+def test_report_files_and_bundle_directory_are_fsynced_before_publication(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
-    collision_kind: str,
 ):
     reporting = _reporting_module()
-    timestamp = "20260714T011225Z"
-    base_json = tmp_path / f"{timestamp}.json"
-    external_markdown = tmp_path / f"{timestamp}.md"
-    replacement = tmp_path / "competitor-replacement.md"
-    real_link = reporting.os.link
-    real_replace = reporting.os.replace
-    link_calls = 0
+    real_fsync = reporting.os.fsync
+    real_symlink = reporting.os.symlink
+    fsync_calls = 0
+    symlink_calls = 0
 
-    if collision_kind == "replace":
-        external_markdown.write_text("old external markdown")
+    def record_fsync(file_descriptor: int) -> None:
+        nonlocal fsync_calls
+        fsync_calls += 1
+        real_fsync(file_descriptor)
 
-    def conflict_on_markdown(source: Path, destination: Path) -> None:
-        nonlocal link_calls
-        link_calls += 1
-        destination = Path(destination)
-        if destination == external_markdown:
-            if collision_kind == "replace":
-                replacement.write_text("external markdown")
-                real_replace(replacement, destination)
-            else:
-                destination.write_text("external markdown")
-        real_link(source, destination)
+    def inspect_before_symlink(source: str, destination: Path) -> None:
+        nonlocal symlink_calls
+        symlink_calls += 1
+        assert fsync_calls == 3
+        real_symlink(source, destination)
 
-    monkeypatch.setattr(reporting.os, "link", conflict_on_markdown)
+    monkeypatch.setattr(reporting.os, "fsync", record_fsync)
+    monkeypatch.setattr(reporting.os, "symlink", inspect_before_symlink)
 
-    json_path, markdown_path = reporting.write_benchmark_report(
-        _marker_report("markdown-link-collision"),
+    reporting.write_benchmark_report(
+        _marker_report("fsync-before-publish"),
         tmp_path,
-        timestamp=timestamp,
+        timestamp="20260714T011225Z",
     )
 
-    assert link_calls == 4
-    assert not base_json.exists()
-    assert external_markdown.read_text() == "external markdown"
-    assert json_path.name == f"{timestamp}-1.json"
-    assert markdown_path.name == f"{timestamp}-1.md"
+    assert symlink_calls == 1
+    assert fsync_calls == 4
     _assert_no_publication_artifacts(tmp_path)
 
 
-def test_markdown_conflict_does_not_remove_replaced_json_target(
+def test_file_write_failure_removes_only_unpublished_private_bundle(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
 ):
     reporting = _reporting_module()
-    timestamp = "20260714T011226Z"
-    base_json = tmp_path / f"{timestamp}.json"
-    base_markdown = tmp_path / f"{timestamp}.md"
-    competitor_json = tmp_path / "competitor.json"
-    real_link = reporting.os.link
-    real_replace = reporting.os.replace
+    original_write_text = reporting._write_text
+    write_calls = 0
 
-    def replace_json_before_markdown_conflict(
-        source: Path,
-        destination: Path,
-    ) -> None:
-        destination = Path(destination)
-        if destination == base_markdown:
-            competitor_json.write_text("external json")
-            real_replace(competitor_json, base_json)
-            base_markdown.write_text("external markdown")
-        real_link(source, destination)
+    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.os,
-        "link",
-        replace_json_before_markdown_conflict,
-    )
+    monkeypatch.setattr(reporting, "_write_text", fail_second_file)
 
-    json_path, markdown_path = reporting.write_benchmark_report(
-        _marker_report("replaced-json-before-cleanup"),
-        tmp_path,
-        timestamp=timestamp,
-    )
+    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 base_json.read_text() == "external json"
-    assert base_markdown.read_text() == "external markdown"
-    assert json_path.name == f"{timestamp}-1.json"
-    assert markdown_path.name == f"{timestamp}-1.md"
-    _assert_no_publication_artifacts(tmp_path)
+    assert list(tmp_path.iterdir()) == []
 
 
-def test_link_exception_cleans_only_owned_links_and_temps(
+def test_bundle_fsync_failure_removes_unpublished_private_bundle(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
 ):
     reporting = _reporting_module()
-    timestamp = "20260714T011227Z"
-    base_json = tmp_path / f"{timestamp}.json"
-    base_markdown = tmp_path / f"{timestamp}.md"
-    competitor_json = tmp_path / "competitor.json"
-    real_link = reporting.os.link
-    real_replace = reporting.os.replace
+    original_fsync_directory = reporting._fsync_directory
 
-    def fail_markdown_link(source: Path, destination: Path) -> None:
-        destination = Path(destination)
-        if destination == base_markdown:
-            competitor_json.write_text("external json")
-            real_replace(competitor_json, base_json)
-            base_markdown.write_text("external markdown")
-            raise OSError("simulated markdown link failure")
-        real_link(source, destination)
+    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.os, "link", fail_markdown_link)
+    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("link-exception"),
+            _marker_report("bundle-fsync-failure"),
             tmp_path,
-            timestamp=timestamp,
+            timestamp="20260714T011227Z",
         )
 
-    assert base_json.read_text() == "external json"
-    assert base_markdown.read_text() == "external markdown"
-    _assert_no_publication_artifacts(tmp_path)
+    assert list(tmp_path.iterdir()) == []
 
 
-def test_link_interrupt_cleans_owned_links_and_temps_before_propagating(
+def test_symlink_failure_removes_unpublished_bundle_without_public_rollback(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
 ):
     reporting = _reporting_module()
-    real_link = reporting.os.link
-    link_calls = 0
 
-    def interrupt_markdown_link(source: Path, destination: Path) -> None:
-        nonlocal link_calls
-        link_calls += 1
-        if link_calls == 2:
-            raise KeyboardInterrupt
-        real_link(source, destination)
+    def fail_symlink(source: str, destination: Path) -> None:
+        del source, destination
+        raise OSError("simulated symlink failure")
 
-    monkeypatch.setattr(reporting.os, "link", interrupt_markdown_link)
+    monkeypatch.setattr(reporting.os, "symlink", fail_symlink)
 
-    with pytest.raises(KeyboardInterrupt):
+    with pytest.raises(
+        reporting.BenchmarkReportWriteError,
+        match="failed to write benchmark reports",
+    ):
         reporting.write_benchmark_report(
-            _marker_report("link-interrupt"),
+            _marker_report("symlink-failure"),
             tmp_path,
             timestamp="20260714T011228Z",
         )
@@ -827,73 +861,73 @@ def test_link_interrupt_cleans_owned_links_and_temps_before_propagating(
     assert list(tmp_path.iterdir()) == []
 
 
-def test_temp_fsync_failure_removes_owned_temps(
+def test_symlink_interrupt_removes_unpublished_bundle_before_propagating(
     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")
+    def interrupt_symlink(source: str, destination: Path) -> None:
+        del source, destination
+        raise KeyboardInterrupt
 
-    monkeypatch.setattr(reporting.os, "fsync", failing_fsync)
+    monkeypatch.setattr(reporting.os, "symlink", interrupt_symlink)
 
-    with pytest.raises(
-        reporting.BenchmarkReportWriteError,
-        match="failed to write benchmark reports",
-    ):
+    with pytest.raises(KeyboardInterrupt):
         reporting.write_benchmark_report(
-            _marker_report("temp-fsync-failure"),
+            _marker_report("symlink-interrupt"),
             tmp_path,
-            timestamp="20260714T012334Z",
+            timestamp="20260714T011229Z",
         )
 
     assert list(tmp_path.iterdir()) == []
 
 
-def test_both_temps_are_complete_and_fsynced_before_first_link(
+def test_post_publication_directory_fsync_failure_keeps_public_bundle(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
 ):
     reporting = _reporting_module()
-    timestamp = "20260714T040506Z"
-    real_fsync = reporting.os.fsync
-    real_link = reporting.os.link
+    original_fsync_directory = reporting._fsync_directory
     fsync_calls = 0
-    link_calls = 0
 
-    def record_fsync(file_descriptor: int) -> None:
+    def fail_output_directory_fsync(path: Path) -> None:
         nonlocal fsync_calls
         fsync_calls += 1
-        real_fsync(file_descriptor)
+        if fsync_calls == 2:
+            raise OSError("simulated output directory fsync failure")
+        original_fsync_directory(path)
 
-    def inspect_before_link(source: Path, destination: Path) -> None:
-        nonlocal link_calls
-        link_calls += 1
-        temp_paths = sorted(tmp_path.glob("*.tmp"))
-        assert len(temp_paths) == 2
-        assert all(path.stat().st_size > 0 for path in temp_paths)
-        assert fsync_calls == 2
-        real_link(source, destination)
+    monkeypatch.setattr(
+        reporting,
+        "_fsync_directory",
+        fail_output_directory_fsync,
+    )
 
-    monkeypatch.setattr(reporting.os, "fsync", record_fsync)
-    monkeypatch.setattr(reporting.os, "link", inspect_before_link)
+    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",
+        )
 
-    json_path, markdown_path = reporting.write_benchmark_report(
-        _marker_report("complete-temps"),
-        tmp_path,
-        timestamp=timestamp,
+    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 link_calls == 2
-    assert json.loads(json_path.read_text())["target"]["model"] == "complete-temps"
-    assert "- Model: `complete-temps`" in markdown_path.read_text()
     _assert_no_publication_artifacts(tmp_path)
 
 
 @pytest.mark.parametrize("stress_round", range(3))
-def test_concurrent_thread_writers_reserve_distinct_matching_pairs(
+def test_concurrent_thread_writers_publish_distinct_matching_bundles(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
     stress_round: int,
@@ -904,16 +938,15 @@ def test_concurrent_thread_writers_reserve_distinct_matching_pairs(
         f"thread-{stress_round}-writer-{index}"
         for index in range(writer_count)
     ]
-    first_temp_barrier = threading.Barrier(writer_count, timeout=15)
+    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) -> tuple[int, int]:
-        identity = original_write_text(path, content)
+    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()
-        return identity
+            first_file_barrier.wait()
 
     monkeypatch.setattr(reporting, "_write_text", synchronized_write)
 
@@ -929,12 +962,16 @@ def test_concurrent_thread_writers_reserve_distinct_matching_pairs(
             )
         )
 
-    assert len({json_path.stem for json_path, _ in results}) == writer_count
+    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_reserve_distinct_matching_pairs(
+def test_concurrent_process_writers_publish_distinct_matching_bundles(
     tmp_path: Path,
     stress_round: int,
 ):
@@ -944,7 +981,7 @@ def test_concurrent_process_writers_reserve_distinct_matching_pairs(
         f"process-{stress_round}-writer-{index}"
         for index in range(writer_count)
     ]
-    first_temp_barrier = context.Barrier(writer_count, timeout=20)
+    first_file_barrier = context.Barrier(writer_count, timeout=20)
     result_queue = context.Queue()
     processes = [
         context.Process(
@@ -953,7 +990,7 @@ def test_concurrent_process_writers_reserve_distinct_matching_pairs(
                 str(tmp_path),
                 marker,
                 f"20260714T03040{stress_round}Z",
-                first_temp_barrier,
+                first_file_barrier,
                 result_queue,
             ),
         )
@@ -968,5 +1005,6 @@ def test_concurrent_process_writers_reserve_distinct_matching_pairs(
 
     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 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)