fix(effects): open the effect journal in binary mode on Windows (#6811)

Raw os.open() is text-mode on Windows unless O_BINARY is set, so each
\n written by the journal became \r\n. The log tracks byte offsets,
so the next append read the extra byte as a corrupt tail and every tool
call after the first in a turn failed with 'Effect claim could not be
persisted durably'.

OR O_BINARY (0 on POSIX) into the journal and launch-reservation opens.

Co-authored-by: Alexandre Teixeira <general@alexandreteixeira.dev>

src/agent_runtime/effect_log.py

@@ -47,6 +47,7 @@ EFFECTS_DIR = os.path.join(DATA_DIR, "effects")
 _RUN_ID = re.compile(r"[a-f0-9]{32}")
 _TYPES = {"claim": EffectClaim, "outcome": EffectOutcome, "observation": Observation}
 _VERSION = 1
+_O_BINARY = getattr(os, "O_BINARY", 0)  # Windows: raw os.open() is text-mode without it
 
 
 def _fsync_directory(directory: str | os.PathLike) -> None:
@@ -243,7 +244,7 @@ class EffectLog:
         """Hold the file exclusively with every durable record merged."""
         assert self.path is not None
         _ensure_directory(self.path.parent)
-        flags = os.O_RDWR | os.O_APPEND | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0)
+        flags = os.O_RDWR | os.O_APPEND | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) | _O_BINARY
         descriptor = os.open(self.path, flags, 0o600)
         try:
             if fcntl is not None:
@@ -358,7 +359,7 @@ class EffectLog:
         with self._lock:
             try:
                 descriptor = os.open(self.path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
-                                     | getattr(os, "O_CLOEXEC", 0))
+                                     | getattr(os, "O_CLOEXEC", 0) | _O_BINARY)
             except FileNotFoundError:
                 return
             except OSError:
@@ -394,7 +395,7 @@ class EffectLog:
         log = cls(run_id, directory=directory)
         assert log.path is not None
         try:
-            descriptor = os.open(log.path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0))
+            descriptor = os.open(log.path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) | _O_BINARY)
         except FileNotFoundError:
             return log
         except OSError as error:
@@ -460,7 +461,7 @@ class EffectLog:
         temporary = target.with_suffix(".tmp")
         data = json.dumps({"run_id": self.run_id, "effect_id": effect_id}, sort_keys=True).encode()
         _ensure_directory(target.parent)
-        flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0)
+        flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) | _O_BINARY
         descriptor = os.open(temporary, flags, 0o600)
         try:
             view = memoryview(data)
@@ -478,7 +479,7 @@ class EffectLog:
             return None
         base = Path(directory) if directory is not None else effects_dir()
         try:
-            descriptor = os.open(base / f"launch-{generation}.json", os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
+            descriptor = os.open(base / f"launch-{generation}.json", os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | _O_BINARY)
             with os.fdopen(descriptor, "rb") as stream:
                 if os.fstat(stream.fileno()).st_nlink != 1:
                     return None

tests/test_effect_journal_persistence.py

@@ -344,3 +344,13 @@ def test_hardlinked_effect_state_is_control_plane_without_scanning_the_store(tmp
     elsewhere.write_text("y")
     os.link(elsewhere, workspace / "pnpm-style.txt")
     assert resources._control_plane_path(str(workspace / "pnpm-style.txt")) is True
+
+
+def test_consecutive_claims_write_lf_only_records(tmp_path, target):
+    # Raw os.open() is text-mode on Windows unless O_BINARY is set; the \r\n it
+    # wrote made the next append read a stray empty line as a corrupt tail.
+    log = EffectLog(RUN, directory=tmp_path / "fx")
+    claim(log, target, "e1", "act-1")
+    claim(log, target, "e2", "act-2")
+    assert b"\r" not in log.path.read_bytes()
+    assert len(records(log.path)) == 2