diff --git a/loopx/capabilities/benchmark_toolkit/native_codex_isolation.py b/loopx/capabilities/benchmark_toolkit/native_codex_isolation.py index 4418d494af..f687f0e330 100644 --- a/loopx/capabilities/benchmark_toolkit/native_codex_isolation.py +++ b/loopx/capabilities/benchmark_toolkit/native_codex_isolation.py @@ -295,7 +295,10 @@ def _atomic_write_text(path: Path, text: str) -> None: ) temporary_path = Path(temporary) try: - os.fchmod(descriptor, path.stat().st_mode & 0o777) + # Windows has no fchmod; the POSIX permission bits of the source file do + # not exist there either, so the copy keeps the platform default. + if hasattr(os, "fchmod"): + os.fchmod(descriptor, path.stat().st_mode & 0o777) with os.fdopen(descriptor, "w", encoding="utf-8") as stream: stream.write(text) stream.flush() diff --git a/loopx/capabilities/decision_context/capture.py b/loopx/capabilities/decision_context/capture.py index 6b3c64d747..5994517f77 100644 --- a/loopx/capabilities/decision_context/capture.py +++ b/loopx/capabilities/decision_context/capture.py @@ -41,7 +41,9 @@ def _open_spool(path: Path, *, goal_id: str, agent_id: str) -> sqlite3.Connectio descriptor = os.open( path, os.O_CREAT | os.O_RDWR | getattr(os, "O_NOFOLLOW", 0), 0o600 ) - os.fchmod(descriptor, 0o600) + # Windows has no fchmod; os.open already applied the mode above. + if hasattr(os, "fchmod"): + os.fchmod(descriptor, 0o600) os.close(descriptor) db = sqlite3.connect(path, timeout=1) db.row_factory = sqlite3.Row diff --git a/loopx/control_plane/goals/botmux_runtime.py b/loopx/control_plane/goals/botmux_runtime.py index 51b89cedc2..2f55c65fc8 100644 --- a/loopx/control_plane/goals/botmux_runtime.py +++ b/loopx/control_plane/goals/botmux_runtime.py @@ -163,7 +163,9 @@ def _write_private_json_atomic(path: Path, payload: Mapping[str, Any]) -> None: ) temporary = Path(temporary_name) try: - os.fchmod(descriptor, 0o600) + # Windows has no fchmod; mkstemp already created the file for this user only. + if hasattr(os, "fchmod"): + os.fchmod(descriptor, 0o600) with os.fdopen(descriptor, "w", encoding="utf-8") as handle: json.dump(dict(payload), handle, ensure_ascii=False, indent=2) handle.write("\n") diff --git a/loopx/extensions/lark/private_json.py b/loopx/extensions/lark/private_json.py index c1f7b2b1e0..9ba1ea0ef9 100644 --- a/loopx/extensions/lark/private_json.py +++ b/loopx/extensions/lark/private_json.py @@ -21,7 +21,9 @@ def write_private_json_atomic( ) temporary = Path(temporary_name) try: - os.fchmod(descriptor, 0o600) + # Windows has no fchmod; mkstemp already created the file for this user only. + if hasattr(os, "fchmod"): + os.fchmod(descriptor, 0o600) with os.fdopen(descriptor, "w", encoding="utf-8") as handle: json.dump(dict(payload), handle, ensure_ascii=False, indent=2) handle.write("\n") diff --git a/tests/test_windows_atomic_writes.py b/tests/test_windows_atomic_writes.py index 8816434be9..639d5fca50 100644 --- a/tests/test_windows_atomic_writes.py +++ b/tests/test_windows_atomic_writes.py @@ -10,7 +10,9 @@ import pytest from loopx.capabilities.decision_context import private_state +from loopx.capabilities.benchmark_toolkit import native_codex_isolation from loopx.control_plane.heartbeat import automation_upgrade +from loopx.control_plane.goals import botmux_runtime from loopx.extensions import presentation from loopx.extensions.lark import private_json @@ -22,7 +24,10 @@ def __init__(self, directory: Path) -> None: self._directory = directory def __getattr__(self, name: str) -> Any: - if name == "O_DIRECTORY": + # Native Windows Python has neither directory fsync nor fchmod; an + # attribute that exists only on POSIX must stay missing here, or the + # POSIX lane stops covering the Windows surface. + if name in {"O_DIRECTORY", "fchmod"}: raise AttributeError(name) return getattr(os, name) @@ -48,6 +53,14 @@ def _write_heartbeat_automation(path: Path) -> None: automation_upgrade._atomic(path, 'prompt = "ready"\n') +def _write_botmux_binding(path: Path) -> None: + botmux_runtime._write_private_json_atomic(path, {"status": "ready"}) + + +def _write_native_isolation(path: Path) -> None: + native_codex_isolation._atomic_write_text(path, "prompt = \"ready\"\n") + + @pytest.mark.parametrize( ("module", "writer", "expected"), [ @@ -75,6 +88,18 @@ def _write_heartbeat_automation(path: Path) -> None: 'prompt = "ready"\n', id="heartbeat-automation", ), + pytest.param( + botmux_runtime, + _write_botmux_binding, + {"status": "ready"}, + id="botmux-binding", + ), + pytest.param( + native_codex_isolation, + _write_native_isolation, + 'prompt = "ready"\n', + id="native-isolation", + ), ], ) def test_atomic_writers_skip_unsupported_windows_directory_fsync(