fix(execution): write the file heartbeat in place — os.replace races concurrent probe reads on Windows (ALP-951)

src/alphamind/execution/process_supervision/heartbeat.py

@@ -14,8 +14,12 @@
 separate watchdog process. Both processes read the same wall clock, so age is
 comparable across the boundary.
 
-The write is atomic (write-temp + ``os.replace``) so a watchdog reading
-concurrently never observes a half-written or empty file.
+The beat is a single in-place write to the heartbeat path. Renaming onto the
+path is not an option: Windows denies it while the watchdog probe holds the
+file open for its concurrent read (ALP-951). The liveness contract is
+per-write freshness with read-side tolerance — a probe read landing inside
+the writer's truncate-write window sees an empty file and reports ``None``,
+which the watchdog treats as one missed probe, never a restart trigger.
 """
 
 from __future__ import annotations
@@ -39,7 +43,7 @@ def beat(self) -> None: ...
 
 
 class FileHeartbeatSink:
-    """Writes the supervised process's latest beat timestamp to a file (atomically)."""
+    """Writes the supervised process's latest beat timestamp to a file (in place)."""
 
     def __init__(self, *, path: Path, clock: Callable[[], float] = time.time) -> None:
         # ``clock`` is the injected wall-clock port (default ``time.time``); tests
@@ -55,13 +59,17 @@ def __init__(self, *, path: Path, clock: Callable[[], float] = time.time) -> Non
     def beat(self) -> None:
         """Record the current timestamp as the latest heartbeat.
 
-        Atomic: writes to a sibling ``*.tmp`` then ``os.replace`` onto the real
-        path, so a concurrent :class:`FileHeartbeatProbe` read never sees a
-        torn or empty file.
+        A single in-place write to the heartbeat path — never a rename onto it,
+        which Windows denies (``PermissionError``) while the watchdog's
+        :class:`FileHeartbeatProbe` holds the file open for its concurrent read
+        (ALP-951). A probe read landing inside the truncate-write window sees an
+        empty file, which :meth:`FileHeartbeatProbe.age` reports as ``None``. The
+        timestamp itself lands in one small buffered ``write``, which concurrent
+        reads observe all-or-nothing on both platforms — empty is the only
+        intermediate state a probe can see, never a torn numeric prefix (which
+        could parse as an ancient epoch and trip a false restart).
         """
-        tmp = self._path.with_suffix(self._path.suffix + ".tmp")
-        tmp.write_text(repr(self._now()), encoding="utf-8")
-        tmp.replace(self._path)
+        self._path.write_text(repr(self._now()), encoding="utf-8")
 
 
 class FileHeartbeatProbe:
@@ -75,11 +83,12 @@ def age(self, *, now: float) -> float | None:
 
         ``None`` means the heartbeat file does not exist yet (the supervised
         process has not started / not yet written one) — the watchdog treats
-        that as startup grace, not a wedge. A transiently unreadable file (e.g.
-        Windows briefly locking it around the writer's ``os.replace``) or a
-        malformed one (hand-corrupted) also reads as ``None`` so a single bad
-        read never trips a false restart — and never crashes the watchdog,
-        which must outlive any single probe failure to keep supervising.
+        that as startup grace, not a wedge. A transiently unreadable file
+        (e.g. an antivirus scan holding it), an empty one (a read landing
+        inside the writer's in-place truncate-write window), or a malformed
+        one (hand-corrupted) also reads as ``None`` so a single bad read never
+        trips a false restart — and never crashes the watchdog, which must
+        outlive any single probe failure to keep supervising.
         """
         try:
             raw = self._path.read_text(encoding="utf-8")

tests/execution/process_supervision/test_heartbeat.py

@@ -41,26 +41,55 @@ def test_probe_missing_heartbeat_is_none(tmp_path: Path) -> None:
 def test_probe_unreadable_heartbeat_is_none_not_a_crash(tmp_path: Path) -> None:
     """A transient OS-level read failure reads as None — the watchdog survives.
 
-    Windows can briefly deny the read while the writer's os.replace holds the
-    file; the probe must treat any OSError as an unreadable beat and re-probe
-    next tick rather than crash the watchdog process. A directory at the
-    heartbeat path raises an OSError subclass on both platforms, standing in
-    for that transient denial.
+    The environment can briefly deny the read (an antivirus scan or SMB layer
+    holding the file); the probe must treat any OSError as an unreadable beat
+    and re-probe next tick rather than crash the watchdog process. A directory
+    at the heartbeat path raises an OSError subclass on both platforms,
+    standing in for that transient denial.
     """
     probe = FileHeartbeatProbe(path=tmp_path)  # a directory: read_text -> OSError
 
     assert probe.age(now=100.0) is None
 
 
-def test_beat_overwrites_previous(tmp_path: Path) -> None:
-    """A later beat resets the timer — age is measured from the most recent beat."""
+def test_beat_lands_in_place_while_reader_holds_the_file_open(tmp_path: Path) -> None:
+    """A beat must land in the file a concurrent probe holds open (ALP-951).
+
+    The watchdog probe reads the heartbeat from a separate process; its open
+    handle must never make ``beat()`` fail (Windows share modes deny rename
+    over an open destination) and the fresh beat must be visible through that
+    same handle (an in-place write, not a directory-entry swap).
+    """
     clock = {"now": 100.0}
-    path = tmp_path / "safety_core.heartbeat"
+    path = tmp_path / "monitor.heartbeat"
+    sink = FileHeartbeatSink(path=path, clock=lambda: clock["now"])
+
+    sink.beat()
+    clock["now"] = 123.0
+    with path.open("r", encoding="utf-8") as held:
+        sink.beat()  # must not raise while ``held`` keeps the file open
+        held.seek(0)
+        assert held.read() == repr(123.0)
+
+
+def test_beat_writes_only_the_heartbeat_file(tmp_path: Path) -> None:
+    """Beats leave exactly the heartbeat file — no ``.tmp`` sibling (ALP-951)."""
+    clock = {"now": 100.0}
+    path = tmp_path / "monitor.heartbeat"
     sink = FileHeartbeatSink(path=path, clock=lambda: clock["now"])
-    probe = FileHeartbeatProbe(path=path)
 
     sink.beat()
     clock["now"] = 200.0
     sink.beat()
 
-    assert probe.age(now=205.0) == 5.0
+    assert [p.name for p in tmp_path.iterdir()] == ["monitor.heartbeat"]
+    assert path.read_text(encoding="utf-8") == repr(200.0)
+
+
+def test_probe_empty_heartbeat_is_none(tmp_path: Path) -> None:
+    """An empty file — a probe read inside the writer's truncate-write window — is None."""
+    path = tmp_path / "monitor.heartbeat"
+    path.write_text("", encoding="utf-8")
+    probe = FileHeartbeatProbe(path=path)
+
+    assert probe.age(now=100.0) is None