test: write scheduler barrier files atomically

scripts/run-test-wave.py

@@ -17,6 +17,7 @@
 import signal
 import subprocess
 import sys
+import tempfile
 import time
 from dataclasses import dataclass
 
@@ -105,6 +106,31 @@ def append_log(path: pathlib.Path, message: str) -> None:
         stream.write("\n")
 
 
+def publish_barrier_file(path: pathlib.Path, text: str) -> None:
+    """Publish a barrier file so a poller sees either no file or its content.
+
+    Path.write_text creates and truncates before it writes, so a reader that
+    polls for existence and then parses the content can observe the zero-byte
+    window in between.  Write next to the destination and rename into place.
+    """
+    with tempfile.NamedTemporaryFile(
+        mode="w",
+        encoding="utf-8",
+        newline="\n",
+        dir=path.parent,
+        prefix=f".{path.name}.",
+        delete=False,
+    ) as temporary:
+        temporary.write(text)
+        temporary.flush()
+        temporary_path = pathlib.Path(temporary.name)
+    try:
+        os.replace(temporary_path, path)
+    except BaseException:
+        temporary_path.unlink(missing_ok=True)
+        raise
+
+
 def start_suite(
     suite: str,
     runner_command: list[str],
@@ -288,12 +314,12 @@ def wait_for_test_pre_terminate_barrier(
     ready = barrier_dir / f"{active.name}.ready"
     leader_exited = barrier_dir / f"{active.name}.leader-exited"
     release = barrier_dir / f"{active.name}.release"
-    ready.write_text(f"{active.process.pid}\n", encoding="utf-8")
+    publish_barrier_file(ready, f"{active.process.pid}\n")
     deadline = time.monotonic() + 10
     while not release.exists():
         returncode = active.process.poll()
         if returncode is not None and not leader_exited.exists():
-            leader_exited.write_text(f"{returncode}\n", encoding="utf-8")
+            publish_barrier_file(leader_exited, f"{returncode}\n")
         if time.monotonic() >= deadline:
             raise RuntimeError(
                 f"test pre-terminate barrier for {active.name!r} was not released"
@@ -312,7 +338,7 @@ def wait_for_test_post_exit_barrier(
         return
     ready = barrier_dir / f"{suite}.ready"
     release = barrier_dir / f"{suite}.release"
-    ready.write_text("child exited; result intentionally not recorded\n", encoding="utf-8")
+    publish_barrier_file(ready, "child exited; result intentionally not recorded\n")
     deadline = time.monotonic() + 10
     while not release.exists():
         if time.monotonic() >= deadline:

tests/test_parallel_harness_contract.sh

@@ -45,6 +45,23 @@ if [[ "$probe_sites" == *kill_grace* ]]; then
     exit 1
 fi
 
+# Barrier files must be published atomically. Path.write_text creates and
+# truncates before it writes, so a poller that saw `<suite>.ready` appear and
+# then parsed the leader pid could read the zero-byte window and fail on
+# int("") -- a scheduler-side race surfacing as a harness flake. Asserted
+# structurally: no barrier file is written in place, and the scheduler renames
+# a same-directory temp file onto the destination instead.
+barrier_writes=$(grep -nE '^[[:space:]]*(ready|leader_exited)\.write_text\(' "$scheduler" || true)
+if [ -n "$barrier_writes" ]; then
+    echo "FAIL: scheduler barrier files are written in place (non-atomic):" >&2
+    echo "$barrier_writes" >&2
+    exit 1
+fi
+if ! grep -Fq 'os.replace(' "$scheduler"; then
+    echo "FAIL: scheduler does not rename barrier files into place" >&2
+    exit 1
+fi
+
 python3 - "$scheduler" <<'PROBE'
 from __future__ import annotations