Keep delegation recovery moving through Windows file sharing conflicts

loopx/control_plane/collaboration/inbox.py

@@ -8,14 +8,12 @@
 
 import hashlib
 import json
-import os
 import re
-import tempfile
 from contextlib import contextmanager
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
-from ...file_lock import exclusive_cross_runtime_file_lock
+from ...file_lock import exclusive_cross_runtime_file_lock, write_bytes_atomic
 from ..runtime.file_paths import windows_extended_path
 from ..content_digest import BARE_SHA256_PATTERN, ENVELOPED_SHA256_PATTERN
 from ..todos.contract import TODO_ID_PATTERN
@@ -81,17 +79,14 @@ def _root(runtime_root: Path) -> Path:
 
 
 def _write(path: Path, value: dict) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
-    fd, tmp = tempfile.mkstemp(dir=path.parent)
-    try:
-        with os.fdopen(fd, "w", encoding="utf-8") as f:
-            json.dump(value, f, ensure_ascii=False)
-            f.flush()
-            os.fsync(f.fileno())
-        os.replace(tmp, path)
-    finally:
-        if os.path.exists(tmp):
-            os.unlink(tmp)
+    # A concurrent status reader on Windows can briefly deny replacement.
+    # Retry only the atomic publication, inside the caller's existing fence.
+    write_bytes_atomic(
+        path,
+        json.dumps(value, ensure_ascii=False).encode("utf-8"),
+        directory_mode=0o700,
+        replace_retry_seconds=0.5,
+    )
 
 
 def _read(path: Path) -> dict:

tests/test_file_lock_windows_atomic.py

@@ -6,12 +6,13 @@
 import json
 import os
 from pathlib import Path
-import threading
 import time
 from types import SimpleNamespace
 
 import pytest
 import loopx.file_lock as locks
+from loopx.control_plane.collaboration import inbox
+from loopx.control_plane.runtime.file_paths import windows_extended_path
 
 pytestmark = pytest.mark.skipif(os.name != 'nt', reason='Windows file sharing boundary')
 
@@ -37,27 +38,22 @@ def close():
 @contextmanager
 def transient_reader(path, monkeypatch):
     attempts = []
-    first_attempt = threading.Event()
     replace = os.replace
-    def observed_replace(source,destination):
-        if Path(destination) == path:
-            attempts.append(time.monotonic())
-            first_attempt.set()
-        return replace(source,destination)
     with reader_without_delete_share(path) as close:
-        def release():
-            first_attempt.wait(2)
-            time.sleep(0.1)
-            close()
-        worker = threading.Thread(target=release)
-        worker.start()
-        monkeypatch.setattr(locks.os,'replace',observed_replace)
-        try:
+
+        def observed_replace(source, destination):
+            if Path(destination) != path:
+                return replace(source, destination)
+            attempts.append(time.monotonic())
+            try:
+                return replace(source, destination)
+            finally:
+                # Hold the real read handle through the first replacement.
+                close()
+
+        with monkeypatch.context() as patch:
+            patch.setattr(locks.os, "replace", observed_replace)
             yield attempts
-        finally:
-            first_attempt.set()
-            worker.join(3)
-            assert not worker.is_alive()
 
 def test_transient_reader_allows_atomic_holder_replace(tmp_path,monkeypatch):
     path = tmp_path / "holder.json"
@@ -68,6 +64,18 @@ def test_transient_reader_allows_atomic_holder_replace(tmp_path,monkeypatch):
     assert json.loads(path.read_text())=={'sequence':1}
     assert not list(tmp_path.glob('*.tmp'))
 
+
+def test_transient_reader_allows_atomic_delegation_record_replace(tmp_path, monkeypatch):
+    path = windows_extended_path(tmp_path / "delegation.json")
+    inbox._write(path, {"status": "turn_returned"})
+
+    with transient_reader(path, monkeypatch) as attempts:
+        inbox._write(path, {"status": "accepted"})
+
+    assert len(attempts) >= 2
+    assert inbox._read(path) == {"status": "accepted"}
+    assert not list(path.parent.glob("*.tmp"))
+
 def test_transient_reader_preserves_real_kernel_lock_body(tmp_path,monkeypatch):
     target=tmp_path/'state.json'
     with locks.exclusive_file_lock(target):