diff --git a/loopx/control_plane/collaboration/inbox.py b/loopx/control_plane/collaboration/inbox.py index b397ef7dc4..2e3107327a 100644 --- a/loopx/control_plane/collaboration/inbox.py +++ b/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: diff --git a/tests/test_file_lock_windows_atomic.py b/tests/test_file_lock_windows_atomic.py index 39e5105dba..8dc44ef752 100644 --- a/tests/test_file_lock_windows_atomic.py +++ b/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):