Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
4 changes: 3 additions & 1 deletion loopx/capabilities/decision_context/capture.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 3 additions & 1 deletion loopx/control_plane/goals/botmux_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
4 changes: 3 additions & 1 deletion loopx/extensions/lark/private_json.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
27 changes: 26 additions & 1 deletion tests/test_windows_atomic_writes.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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)

Expand All @@ -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"),
[
Expand Down Expand Up @@ -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(
Expand Down
Loading