test: write scheduler barrier files atomically

Path.write_text is open-then-write: it creates and truncates the file
first and the content lands afterwards. The parallel harness contract
polls for `<suite>.ready` to appear and then parses its content as the
leader pid, so a reader that wins that window sees an empty file and
fails with `ValueError: invalid literal for int() with base 10: ''` --
a spurious FAIL of the harness contract with a scheduler-side cause.
The same shape applied to `<suite>.leader-exited`.

Publish the barrier files through a helper that writes a same-directory
temp file and os.replace()s it onto the destination, which is atomic on
POSIX and Windows: a poller now sees either no file or complete content.
All three barrier publications go through it.

Pin the contract structurally (no timing): the scheduler may not write a
barrier file in place, and it must rename into place instead. The pin is
RED on the previous scheduler.

Distilled from #1188 with co-author credit.

Signed-off-by: Martin Vogel <martin.vogel.tech@gmail.com>
Co-authored-by: Kody <kaidi.shi.1121@gmail.com>
Signed-off-by: Martin Vogel <martin.vogel.tech@gmail.com>

scripts/run-test-wave.py

@@ -17,6 +17,7 @@ import re
 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