fix(windows): skip unsupported parent-directory fsync

.github/workflows/python-tests.yml

@@ -567,6 +567,7 @@ jobs:
           tests/control_plane/test_effect_runtime_integration.py
           tests/control_plane/test_local_authority_shadow_outbox.py
           tests/test_self_update_runtime_activation.py
+          tests/test_windows_atomic_writes.py
           tests/test_windows_install.py
 
       - name: Set up the qualified SQLite runtime

loopx/capabilities/decision_context/private_state.py

@@ -177,13 +177,14 @@ def _write_private_json_atomic(
             handle.flush()
             os.fsync(handle.fileno())
         os.replace(temporary, destination)
-        directory_fd = os.open(
-            destination.parent,
-            os.O_RDONLY | getattr(os, "O_DIRECTORY", 0),
-        )
-        try:
-            os.fsync(directory_fd)
-        finally:
-            os.close(directory_fd)
+        if os.name == "posix":
+            directory_fd = os.open(
+                destination.parent,
+                os.O_RDONLY | getattr(os, "O_DIRECTORY", 0),
+            )
+            try:
+                os.fsync(directory_fd)
+            finally:
+                os.close(directory_fd)
     finally:
         temporary.unlink(missing_ok=True)

loopx/control_plane/heartbeat/automation_upgrade.py

@@ -70,11 +70,12 @@ def _atomic(path: Path, text: str) -> None:
             stream.flush()
             os.fsync(stream.fileno())
         os.replace(temporary, path)
-        directory = os.open(path.parent, os.O_RDONLY)
-        try:
-            os.fsync(directory)
-        finally:
-            os.close(directory)
+        if os.name == "posix":
+            directory = os.open(path.parent, os.O_RDONLY)
+            try:
+                os.fsync(directory)
+            finally:
+                os.close(directory)
     finally:
         if os.path.exists(temporary):
             os.unlink(temporary)

loopx/extensions/lark/private_json.py

@@ -28,13 +28,14 @@ def write_private_json_atomic(
             handle.flush()
             os.fsync(handle.fileno())
         os.replace(temporary, destination)
-        directory_fd = os.open(
-            destination.parent,
-            os.O_RDONLY | getattr(os, "O_DIRECTORY", 0),
-        )
-        try:
-            os.fsync(directory_fd)
-        finally:
-            os.close(directory_fd)
+        if os.name == "posix":
+            directory_fd = os.open(
+                destination.parent,
+                os.O_RDONLY | getattr(os, "O_DIRECTORY", 0),
+            )
+            try:
+                os.fsync(directory_fd)
+            finally:
+                os.close(directory_fd)
     finally:
         temporary.unlink(missing_ok=True)

loopx/extensions/presentation.py

@@ -728,14 +728,15 @@ def _atomic_write_projection(path: Path, payload: Mapping[str, Any]) -> None:
             os.fsync(handle.fileno())
         os.chmod(temporary, 0o600)
         os.replace(temporary, path)
-        directory_fd = os.open(
-            path.parent,
-            os.O_RDONLY | getattr(os, "O_DIRECTORY", 0),
-        )
-        try:
-            os.fsync(directory_fd)
-        finally:
-            os.close(directory_fd)
+        if os.name == "posix":
+            directory_fd = os.open(
+                path.parent,
+                os.O_RDONLY | getattr(os, "O_DIRECTORY", 0),
+            )
+            try:
+                os.fsync(directory_fd)
+            finally:
+                os.close(directory_fd)
     finally:
         temporary.unlink(missing_ok=True)
 

tests/test_windows_atomic_writes.py

@@ -0,0 +1,95 @@
+from __future__ import annotations
+
+import json
+import os
+from collections.abc import Callable
+from pathlib import Path
+from types import ModuleType
+from typing import Any
+
+import pytest
+
+from loopx.capabilities.decision_context import private_state
+from loopx.control_plane.heartbeat import automation_upgrade
+from loopx.extensions import presentation
+from loopx.extensions.lark import private_json
+
+
+class _WindowsOs:
+    name = "nt"
+
+    def __init__(self, directory: Path) -> None:
+        self._directory = directory
+
+    def __getattr__(self, name: str) -> Any:
+        if name == "O_DIRECTORY":
+            raise AttributeError(name)
+        return getattr(os, name)
+
+    def open(self, path: str | bytes | os.PathLike[str], flags: int, *args: Any) -> int:
+        if Path(path) == self._directory and flags == os.O_RDONLY:
+            raise PermissionError(13, "Permission denied", str(path))
+        return os.open(path, flags, *args)
+
+
+def _write_lark_private_json(path: Path) -> None:
+    private_json.write_private_json_atomic(path, {"status": "ready"})
+
+
+def _write_decision_context(path: Path) -> None:
+    private_state.write_private_decision_cursors_atomic(path, {"source": "cursor"})
+
+
+def _write_extension_projection(path: Path) -> None:
+    presentation._atomic_write_projection(path, {"status": "ready"})
+
+
+def _write_heartbeat_automation(path: Path) -> None:
+    automation_upgrade._atomic(path, 'prompt = "ready"\n')
+
+
+@pytest.mark.parametrize(
+    ("module", "writer", "expected"),
+    [
+        pytest.param(
+            private_json,
+            _write_lark_private_json,
+            {"status": "ready"},
+            id="lark-private-json",
+        ),
+        pytest.param(
+            private_state,
+            _write_decision_context,
+            {"source": "cursor"},
+            id="decision-context",
+        ),
+        pytest.param(
+            presentation,
+            _write_extension_projection,
+            {"status": "ready"},
+            id="extension-projection",
+        ),
+        pytest.param(
+            automation_upgrade,
+            _write_heartbeat_automation,
+            'prompt = "ready"\n',
+            id="heartbeat-automation",
+        ),
+    ],
+)
+def test_atomic_writers_skip_unsupported_windows_directory_fsync(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    module: ModuleType,
+    writer: Callable[[Path], None],
+    expected: object,
+) -> None:
+    target = tmp_path / f"{module.__name__.rsplit('.', 1)[-1]}.json"
+    if os.name != "nt":
+        monkeypatch.setattr(module, "os", _WindowsOs(target.parent))
+
+    writer(target)
+
+    content = target.read_text(encoding="utf-8")
+    actual = json.loads(content) if isinstance(expected, dict) else content
+    assert actual == expected