Fix atomic benchmark result publication

README.md

@@ -99,10 +99,18 @@ therefore supports no performance, full-skill, prose/remediation, severity,
 provider-authenticity, or statistical-superiority claim. Historical reports
 remain rejected.
 
-Publication reserves the exact bundle directory before writing and adds a final
-`COMPLETE` marker only after both report and manifest are in place. This is a
-fail-closed local publication guard; it does not claim crash-durable or
-independently witnessed publication.
+Publication builds the complete bundle in a private sibling staging directory,
+then exposes all three files at once with the platform's atomic no-replace
+directory rename. macOS uses `renamex_np(RENAME_EXCL)` and Linux uses
+`renameat2(RENAME_NOREPLACE)`; unsupported platforms, kernels, or filesystems
+fail closed. A concurrent destination is never replaced. On handled publication
+failures, `COMPLETE` is removed before staging evidence is retained, and because
+the canonical name remains free, a later retry is allowed. A marker-removal
+failure is reported explicitly. Published bundles inherit the staging
+directory's owner-only permissions; on macOS, inherited ACLs are removed and
+verified before payload creation. This guard does not claim crash durability,
+protection from an arbitrary malicious same-user process, or independent
+witnessing.
 
 ## License
 

scripts/benchmark_scorer.py

@@ -6,12 +6,14 @@
 """
 
 import argparse
+import ctypes
+import errno
 import hashlib
 import json
 import os
 import re
-import shutil
 import subprocess
+import sys
 import tempfile
 from pathlib import Path, PurePosixPath
 
@@ -554,39 +556,100 @@ def build_window_references(run_id: str, artifacts: list[dict], fixtures: list[d
     return refs
 
 
+def atomic_rename_no_replace(source: Path, destination: Path) -> None:
+    """Atomically publish a directory without replacing any destination."""
+    libc = ctypes.CDLL(None, use_errno=True)
+    source_bytes = os.fsencode(source)
+    destination_bytes = os.fsencode(destination)
+    if sys.platform == "darwin":
+        function = getattr(libc, "renamex_np", None)
+        signature = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_uint]
+        arguments = (source_bytes, destination_bytes, 0x4)  # RENAME_EXCL
+    elif sys.platform.startswith("linux"):
+        function = getattr(libc, "renameat2", None)
+        signature = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]
+        arguments = (-100, source_bytes, -100, destination_bytes, 0x1)  # AT_FDCWD, RENAME_NOREPLACE
+    else:
+        raise IntegrityError(f"Atomic no-replace directory publication is unsupported on {sys.platform}")
+    if function is None:
+        raise IntegrityError(f"Atomic no-replace directory publication is unavailable on {sys.platform}")
+    function.argtypes = signature
+    function.restype = ctypes.c_int
+    ctypes.set_errno(0)
+    if function(*arguments) == 0:
+        return
+    error_number = ctypes.get_errno()
+    if error_number in {errno.EEXIST, errno.ENOTEMPTY}:
+        raise FileExistsError(error_number, os.strerror(error_number), destination)
+    unsupported = {errno.EINVAL, errno.ENOSYS, errno.EOPNOTSUPP, errno.ENOTSUP}
+    if error_number in unsupported:
+        raise IntegrityError(f"Atomic no-replace directory publication is unsupported by {destination.parent}")
+    raise OSError(error_number, os.strerror(error_number), destination)
+
+
+def darwin_has_extended_acl(path: Path) -> bool:
+    """Return whether a Darwin path has an extended ACL, failing closed on errors."""
+    libc = ctypes.CDLL(None, use_errno=True)
+    acl_get_file = getattr(libc, "acl_get_file", None)
+    acl_free = getattr(libc, "acl_free", None)
+    if acl_get_file is None or acl_free is None:
+        raise IntegrityError("Darwin ACL inspection is unavailable")
+    acl_get_file.argtypes = [ctypes.c_char_p, ctypes.c_int]
+    acl_get_file.restype = ctypes.c_void_p
+    acl_free.argtypes = [ctypes.c_void_p]
+    acl_free.restype = ctypes.c_int
+    ctypes.set_errno(0)
+    acl = acl_get_file(os.fsencode(path), 0x100)  # ACL_TYPE_EXTENDED
+    if acl:
+        if acl_free(acl) != 0:
+            error_number = ctypes.get_errno()
+            raise OSError(error_number, os.strerror(error_number), path)
+        return True
+    error_number = ctypes.get_errno()
+    if error_number == errno.ENOENT and os.path.lexists(path):
+        return False
+    raise OSError(error_number, os.strerror(error_number), path)
+
+
+def harden_private_stage(stage: Path) -> None:
+    """Make a fresh staging directory effectively owner-only before writing."""
+    stage.chmod(0o700)
+    if sys.platform == "darwin":
+        result = subprocess.run(["/bin/chmod", "-N", os.fspath(stage)], capture_output=True, text=True)
+        if result.returncode:
+            raise IntegrityError(result.stderr.strip() or f"Could not remove inherited ACLs from {stage}")
+        if darwin_has_extended_acl(stage):
+            raise IntegrityError(f"Could not verify ACL removal from {stage}")
+    if stage.stat().st_mode & 0o777 != 0o700:
+        raise IntegrityError(f"Private staging directory is not mode 0700: {stage}")
+
+
+def preserve_unpublished_stage(stage: Path, error: BaseException) -> None:
+    """Remove the success marker while retaining unpublished diagnostic evidence."""
+    try:
+        (stage / "COMPLETE").unlink(missing_ok=True)
+    except OSError as marker_error:
+        error.add_note(f"Could not remove unpublished COMPLETE marker: {marker_error}")
+    error.add_note(f"Unpublished benchmark evidence preserved at {stage}")
+
+
 def atomic_publish(bundle: Path, report: str, manifest: dict) -> tuple[Path, Path]:
     if os.path.lexists(bundle):
         raise IntegrityError(f"Refusing to overwrite existing bundle: {bundle}")
     bundle.parent.mkdir(parents=True, exist_ok=True)
+    stage = Path(tempfile.mkdtemp(prefix=".pending-model-benchmark-", dir=bundle.parent))
     try:
-        bundle.mkdir()
+        harden_private_stage(stage)
+        (stage / "report.md").write_text(report + "\n")
+        (stage / "manifest.json").write_text(json.dumps(manifest, indent=2, allow_nan=False) + "\n")
+        (stage / "COMPLETE").write_text("complete\n")
+        atomic_rename_no_replace(stage, bundle)
     except FileExistsError as error:
-        raise IntegrityError(f"Refusing to overwrite existing bundle: {bundle}") from error
-    reservation = bundle.lstat()
-    temp = None
-    bundle_fd = None
-    try:
-        bundle_fd = os.open(bundle, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
-        opened = os.fstat(bundle_fd)
-        if (opened.st_dev, opened.st_ino) != (reservation.st_dev, reservation.st_ino):
-            raise IntegrityError("Reserved publication bundle was substituted before it was opened")
-        temp = Path(tempfile.mkdtemp(prefix=".pending-model-benchmark-", dir=bundle.parent))
-        (temp / "report.md").write_text(report + "\n")
-        (temp / "manifest.json").write_text(json.dumps(manifest, indent=2, allow_nan=False) + "\n")
-        (temp / "COMPLETE").write_text("complete\n")
-        for name in ("report.md", "manifest.json", "COMPLETE"):
-            os.replace(temp / name, name, dst_dir_fd=bundle_fd)
-        temp.rmdir()
-        current = bundle.lstat()
-        if (current.st_dev, current.st_ino) != (reservation.st_dev, reservation.st_ino):
-            raise IntegrityError("Reserved publication bundle was substituted during publication")
-    except Exception:
-        if temp is not None:
-            shutil.rmtree(temp, ignore_errors=True)
+        preserve_unpublished_stage(stage, error)
+        raise IntegrityError(f"Refusing to overwrite existing bundle: {bundle}; unpublished evidence preserved at {stage}") from error
+    except BaseException as error:
+        preserve_unpublished_stage(stage, error)
         raise
-    finally:
-        if bundle_fd is not None:
-            os.close(bundle_fd)
     return bundle / "report.md", bundle / "manifest.json"
 
 

tests/test_run_benchmark.py

@@ -1,11 +1,13 @@
 import copy
+import errno
 import json
 import os
 import subprocess
 import sys
 import tempfile
 import unittest
 import uuid
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from unittest.mock import patch
 
@@ -338,63 +340,43 @@ def test_invalid_commit_run_id_and_traversal_fail(self):
         with self.assertRaises(run_benchmark.IntegrityError):
             self.execute_benchmark(capture_root="../captures")
 
-    def test_publication_failure_preserves_incomplete_reservation(self):
+    def test_publication_failure_preserves_private_evidence_and_allows_retry(self):
         results = self.repo / "fault-results"
-        with patch.object(run_benchmark.os, "replace", side_effect=OSError("fault")):
-            with self.assertRaises(OSError):
+        failure = OSError(errno.EIO, "fault")
+        with patch.object(benchmark_scorer, "atomic_rename_no_replace", side_effect=failure):
+            with self.assertRaises(OSError) as raised:
                 self.execute_benchmark(results)
         bundle = results / f"model-benchmark-{self.critic}-{self.run_id}"
-        self.assertTrue(bundle.is_dir())
-        self.assertFalse((bundle / "COMPLETE").exists())
-        with self.assertRaises(run_benchmark.IntegrityError):
-            self.execute_benchmark(results)
-
-    def test_substituted_bundle_is_never_deleted_or_completed(self):
-        results = self.repo / "substitution-results"
+        pending = list(results.glob(".pending-model-benchmark-*"))
+        self.assertFalse(bundle.exists())
+        self.assertEqual(1, len(pending))
+        self.assertTrue((pending[0] / "report.md").is_file())
+        self.assertTrue((pending[0] / "manifest.json").is_file())
+        self.assertFalse((pending[0] / "COMPLETE").exists())
+        self.assertIn(str(pending[0]), "\n".join(raised.exception.__notes__))
+        self.execute_benchmark(results)
+        self.assertEqual("complete\n", (bundle / "COMPLETE").read_text())
+
+    def test_destination_created_at_publish_is_preserved(self):
+        results = self.repo / "destination-race-results"
         bundle = results / f"model-benchmark-{self.critic}-{self.run_id}"
-        moved = results / "original-reservation"
-        original_replace = os.replace
-        replaced = False
-
-        def substitute(source, destination, *args, **kwargs):
-            nonlocal replaced
-            if not replaced:
-                replaced = True
-                original_replace(bundle, moved)
-                bundle.mkdir()
-                (bundle / "other-writer.txt").write_text("preserve\n")
-                raise OSError("substituted")
-            return original_replace(source, destination, *args, **kwargs)
-
-        with patch.object(run_benchmark.os, "replace", side_effect=substitute):
-            with self.assertRaises(OSError):
-                self.execute_benchmark(results)
-        self.assertEqual("preserve\n", (bundle / "other-writer.txt").read_text())
-        self.assertFalse((bundle / "COMPLETE").exists())
-        self.assertTrue(moved.is_dir())
+        original_publish = benchmark_scorer.atomic_rename_no_replace
 
-    def test_bundle_substituted_before_open_is_rejected_before_writes(self):
-        results = self.repo / "pre-open-substitution-results"
-        bundle = results / f"model-benchmark-{self.critic}-{self.run_id}"
-        moved = results / "original-reservation"
-        original_open = os.open
-        replaced = False
-
-        def substitute(path, flags, *args, **kwargs):
-            nonlocal replaced
-            if not replaced and Path(path) == bundle:
-                replaced = True
-                os.replace(bundle, moved)
-                bundle.mkdir()
-                (bundle / "other-writer.txt").write_text("preserve\n")
-            return original_open(path, flags, *args, **kwargs)
-
-        with patch.object(run_benchmark.os, "open", side_effect=substitute):
+        def create_other_writer(source, destination):
+            destination.mkdir()
+            (destination / "report.md").write_text("other report\n")
+            (destination / "manifest.json").write_text("other manifest\n")
+            return original_publish(source, destination)
+
+        with patch.object(benchmark_scorer, "atomic_rename_no_replace", side_effect=create_other_writer):
             with self.assertRaises(run_benchmark.IntegrityError):
                 self.execute_benchmark(results)
-        self.assertEqual("preserve\n", (bundle / "other-writer.txt").read_text())
+        self.assertEqual("other report\n", (bundle / "report.md").read_text())
+        self.assertEqual("other manifest\n", (bundle / "manifest.json").read_text())
         self.assertFalse((bundle / "COMPLETE").exists())
-        self.assertEqual([], list(moved.iterdir()))
+        pending = list(results.glob(".pending-model-benchmark-*"))
+        self.assertEqual(1, len(pending))
+        self.assertFalse((pending[0] / "COMPLETE").exists())
 
     def test_annotated_tag_and_unrelated_capture_commit_are_rejected(self):
         git(self.repo, "tag", "-a", "capture-tag", "-m", "capture", self.capture_commit)
@@ -453,6 +435,121 @@ def test_cli_bootstrap_executes_committed_blob_and_rejects_dirty_core(self):
         self.assertIn("Refusing to execute dirty scorer", dirty.stderr)
 
 
+class AtomicNoReplaceTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name)
+
+    def tearDown(self):
+        self.temp.cleanup()
+
+    def source(self, name="source"):
+        source = self.root / name
+        source.mkdir()
+        (source / "owner.txt").write_text(name)
+        return source
+
+    def test_atomic_publish_exposes_only_complete_bundle(self):
+        bundle = self.root / "bundle"
+        report, manifest = run_benchmark.atomic_publish(bundle, "report", {"valid": True})
+        self.assertEqual("report\n", report.read_text())
+        self.assertEqual({"valid": True}, json.loads(manifest.read_text()))
+        self.assertEqual("complete\n", (bundle / "COMPLETE").read_text())
+        self.assertEqual([], list(self.root.glob(".pending-model-benchmark-*")))
+
+    def test_native_no_replace_rejects_every_existing_destination_type(self):
+        for kind in ("empty-dir", "nonempty-dir", "file", "symlink", "dangling-symlink"):
+            with self.subTest(kind=kind):
+                case = self.root / kind
+                case.mkdir()
+                source = case / "source"; source.mkdir()
+                destination = case / "destination"
+                if kind.endswith("dir"):
+                    destination.mkdir()
+                    if kind == "nonempty-dir":
+                        (destination / "other.txt").write_text("preserve")
+                elif kind == "file":
+                    destination.write_text("preserve")
+                else:
+                    target = case / ("target" if kind == "symlink" else "missing")
+                    if kind == "symlink":
+                        target.write_text("preserve")
+                    destination.symlink_to(target)
+                with self.assertRaises(FileExistsError):
+                    run_benchmark.atomic_rename_no_replace(source, destination)
+                self.assertTrue(source.is_dir())
+                self.assertTrue(os.path.lexists(destination))
+
+    def test_two_publishers_have_exactly_one_winner(self):
+        destination = self.root / "winner"
+        sources = [self.source("source-one"), self.source("source-two")]
+
+        def publish(source):
+            try:
+                run_benchmark.atomic_rename_no_replace(source, destination)
+                return "won"
+            except FileExistsError:
+                return "lost"
+
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            outcomes = list(pool.map(publish, sources))
+        self.assertEqual(["lost", "won"], sorted(outcomes))
+        self.assertIn((destination / "owner.txt").read_text(), {"source-one", "source-two"})
+        self.assertEqual(1, sum(source.exists() for source in sources))
+
+    def test_unsupported_platform_and_missing_symbol_fail_closed(self):
+        with patch.object(benchmark_scorer.sys, "platform", "win32"):
+            with self.assertRaises(run_benchmark.IntegrityError):
+                run_benchmark.atomic_rename_no_replace(self.source("unsupported"), self.root / "unsupported-dest")
+        with patch.object(benchmark_scorer.ctypes, "CDLL", return_value=object()):
+            with self.assertRaises(run_benchmark.IntegrityError):
+                run_benchmark.atomic_rename_no_replace(self.source("missing-symbol"), self.root / "missing-dest")
+
+    @unittest.skipUnless(sys.platform == "darwin", "Darwin ACL behavior")
+    def test_inherited_delete_acl_cannot_leave_complete_on_failed_stage(self):
+        results = self.root / "deny-results"
+        results.mkdir()
+        subprocess.run(
+            ["/bin/chmod", "+a", "everyone deny delete_child,directory_inherit", str(results)],
+            check=True,
+        )
+        bundle = results / "bundle"
+        try:
+            with self.assertRaises(PermissionError):
+                run_benchmark.atomic_publish(bundle, "report", {"valid": True})
+            pending = list(results.glob(".pending-model-benchmark-*"))
+            self.assertEqual(1, len(pending))
+            self.assertFalse(bundle.exists())
+            self.assertTrue((pending[0] / "report.md").is_file())
+            self.assertTrue((pending[0] / "manifest.json").is_file())
+            self.assertFalse((pending[0] / "COMPLETE").exists())
+        finally:
+            subprocess.run(["/bin/chmod", "-N", str(results)], check=True)
+
+    @unittest.skipUnless(sys.platform == "darwin", "Darwin ACL behavior")
+    def test_inherited_allow_acl_is_removed_before_payload_creation(self):
+        results = self.root / "allow-results"
+        results.mkdir()
+        subprocess.run(
+            [
+                "/bin/chmod",
+                "+a",
+                "everyone allow list,search,readattr,readextattr,readsecurity,directory_inherit,file_inherit",
+                str(results),
+            ],
+            check=True,
+        )
+        bundle = results / "bundle"
+        try:
+            self.assertTrue(benchmark_scorer.darwin_has_extended_acl(results))
+            run_benchmark.atomic_publish(bundle, "report", {"valid": True})
+            self.assertEqual(0o700, bundle.stat().st_mode & 0o777)
+            for path in (bundle, bundle / "report.md", bundle / "manifest.json", bundle / "COMPLETE"):
+                self.assertFalse(benchmark_scorer.darwin_has_extended_acl(path), path)
+        finally:
+            subprocess.run(["/bin/chmod", "-N", str(results)], check=True)
+
+
 class RealRepoIntegrationTests(unittest.TestCase):
     def test_real_catalogs_and_all_twenty_four_fixtures_validate(self):
         total = 0