fix(codex): keep Windows rollout reads from blocking native rotation

.github/workflows/python-tests.yml

@@ -866,6 +866,12 @@ jobs:
       - name: Qualify TypeScript shard paths on native Windows
         run: python -m pytest -q tests/test_python_ci_workflow.py::test_typescript_shards_select_every_test_file_exactly_once
 
+      - name: Qualify read-only Codex rollout rotation on native Windows
+        run: >-
+          python -m pytest -q
+          tests/test_codex_rollout_file.py
+          tests/test_codex_rollout_cursor.py
+
       - name: Qualify the canary premerge contract smoke on native Windows
         run: python examples/canary/premerge-validation-gate-smoke.py
 

docs/reference/protocols/host-integration-surface-v0.md

@@ -373,6 +373,18 @@ source changes; it does not authenticate arbitrary edits to the host store.
 Cache loss or denied cache writes do not grant permissions or prevent a fresh
 bounded read. The provider never writes into the Codex home.
 
+On Windows, the rollout reader requests read access with read/write/delete
+sharing, allowing the host to rename its log or replace it with the native
+`ReplaceFile` operation while an observation is open. The reader cannot write,
+create or delete the source. Existing file-identity checks keep a mid-read
+replacement unknown until a subsequent observation recovers the new source.
+Windows `os.replace` uses a different operation that cannot overwrite an open
+destination even with delete sharing; the native rotation fixture uses
+[`ReplaceFile`](https://learn.microsoft.com/en-us/windows/win32/api/winbase/nf-winbase-replacefilew).
+Visible conversation paging and content reads reuse the same read-only transport.
+An in-progress read retains its opened snapshot; the next page detects the new
+source identity and resets its cursor, while old content references are rejected.
+
 These are provider IO/cache changes under the existing host activity contract,
 not a new execution authority, scheduler, model call or configurable capability.
 

loopx/capabilities/native_chat/codex_history.py

@@ -15,6 +15,7 @@
 from typing import Any
 
 from ...public_safe_text import CATEGORY_CREDENTIAL, classify_private_text
+from ...hosts.codex_rollout_file import open_rollout
 
 WINDOW_BYTES = 2 * 1024 * 1024
 RECORD_BYTES = 8 * 1024 * 1024
@@ -206,7 +207,7 @@ def read_messages(
         raise ValueError("Choose one history direction")
     if any(position is not None and position < 0 for position in (before, after)):
         raise ValueError("History position must be nonnegative")
-    with path.open("rb") as stream:
+    with open_rollout(path) as stream:
         stamp = os.fstat(stream.fileno())
         size = stamp.st_size
         source = source_ref(stamp)
@@ -273,7 +274,7 @@ def read_content(
 ) -> dict[str, Any]:
     if entry_id < 0 or offset < 0:
         raise ValueError("Content position must be nonnegative")
-    with path.open("rb") as stream:
+    with open_rollout(path) as stream:
         stamp = os.fstat(stream.fileno())
         if source_ref(stamp) != expected_source or entry_id >= stamp.st_size:
             raise ValueError("History source changed")

loopx/codex_app_thread_activity.py

@@ -20,6 +20,7 @@
 from typing import Any
 
 from .hosts.codex_rollout_cursor import Reader, RolloutCursor
+from .hosts.codex_rollout_file import open_rollout
 from .control_plane.agents.host_thread_activity import (
     HostDeliveryExpectation,
     HostDeliveryScope,
@@ -176,7 +177,7 @@ def _rollout_activity(
     try:
         if not path.is_file():
             return HostThreadActivity.unknown(HostThreadUnknownReason.STORE_UNAVAILABLE)
-        with path.open("rb") as handle:
+        with open_rollout(path) as handle:
             stat = os.fstat(handle.fileno())
             if Path(os.path.realpath(path)) != path or (stat.st_dev, stat.st_ino) != (path.stat().st_dev, path.stat().st_ino):
                 return HostThreadActivity.unknown(HostThreadUnknownReason.RECORD_UNRECOGNIZED)

loopx/hosts/codex_rollout_file.py

@@ -0,0 +1,47 @@
+"""Read-only Codex provider IO that does not prevent native log rotation."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import BinaryIO
+
+
+def open_rollout(path: Path) -> BinaryIO:
+    if os.name != "nt":
+        return path.open("rb")
+
+    import ctypes
+    import msvcrt
+    from ctypes import wintypes
+
+    kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+    create_file = kernel32.CreateFileW
+    create_file.argtypes = (
+        wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, wintypes.LPVOID,
+        wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE,
+    )
+    create_file.restype = wintypes.HANDLE
+    close_handle = kernel32.CloseHandle
+    close_handle.argtypes = (wintypes.HANDLE,)
+    close_handle.restype = wintypes.BOOL
+    filename = os.path.abspath(path)
+    if not filename.startswith("\\\\?\\"):
+        filename = "\\\\?\\UNC\\" + filename[2:] if filename.startswith("\\\\") else "\\\\?\\" + filename
+    # GENERIC_READ, FILE_SHARE_READ|WRITE|DELETE, OPEN_EXISTING. Sharing delete
+    # permits the host's rename/replace; it grants this reader no write access.
+    handle = create_file(filename, 0x80000000, 0x7, None, 3, 0, None)
+    if handle == wintypes.HANDLE(-1).value:
+        raise ctypes.WinError(ctypes.get_last_error())
+    try:
+        descriptor = msvcrt.open_osfhandle(handle, os.O_RDONLY | os.O_BINARY | os.O_NOINHERIT)
+    except BaseException:
+        close_handle(handle)
+        raise
+    # open_osfhandle transfers handle ownership to the descriptor, and fdopen
+    # transfers it to the returned stream. Release it if stream creation fails.
+    try:
+        return os.fdopen(descriptor, "rb")
+    except BaseException:
+        os.close(descriptor)
+        raise

tests/test_codex_rollout_cursor.py

@@ -55,6 +55,72 @@ def write(path, data):
     path.write_bytes(data)
 
 
+def replace_rollout(replacement, path):
+    if os.name != "nt":
+        os.replace(replacement, path)
+        return
+    import ctypes
+    from ctypes import wintypes
+
+    # MoveFileEx (os.replace) cannot replace an open Windows destination even
+    # with delete sharing. Exercise the native atomic replacement operation.
+    replace = ctypes.WinDLL("kernel32", use_last_error=True).ReplaceFileW
+    replace.argtypes = (wintypes.LPCWSTR, wintypes.LPCWSTR, wintypes.LPCWSTR,
+                        wintypes.DWORD, wintypes.LPVOID, wintypes.LPVOID)
+    replace.restype = wintypes.BOOL
+    if not replace(str(path), str(replacement), None, 0, None, None):
+        raise ctypes.WinError(ctypes.get_last_error())
+
+
+@pytest.mark.parametrize("reader", ["messages", "content"])
+def test_visible_history_read_does_not_block_native_rollout_rotation(
+    tmp_path, monkeypatch, reader
+):
+    from loopx.capabilities.native_chat import codex_history as history
+
+    path, replacement = tmp_path / "rollout.jsonl", tmp_path / "replacement.jsonl"
+
+    def message(text):
+        return encode({
+            "type": "response_item",
+            "payload": {"type": "message", "role": "assistant", "channel": "final",
+                        "content": [{"type": "output_text", "text": text}]},
+        })
+
+    path.write_bytes(message("Original visible result"))
+    replacement.write_bytes(message("Replacement visible result"))
+    previous = history.read_messages(path)
+    row = previous["messages"][0]
+    content_args = dict(entry_id=int(row["id"]), offset=0,
+                        expected_source=previous["source_ref"],
+                        content_ref=row["content_ref"])
+    original_display = history._display_record
+    rotations = []
+
+    def display(raw):
+        # Both production readers still own their source handle here; the
+        # message reader also owns a real read-only mmap of the old snapshot.
+        replace_rollout(replacement, path)
+        rotations.append(True)
+        return original_display(raw)
+
+    with monkeypatch.context() as during_read:
+        during_read.setattr(history, "_display_record", display)
+        if reader == "messages":
+            snapshot = history.read_messages(path)["messages"][0]
+        else:
+            snapshot = history.read_content(path, **content_args)
+        assert snapshot["text"] == "Original visible result"
+    assert rotations == [True]
+    current = history.read_messages(
+        path, after=previous["after"], expected_source=previous["source_ref"]
+    )
+    assert current["reset"] is True
+    assert [row["text"] for row in current["messages"]] == ["Replacement visible result"]
+    with pytest.raises(ValueError, match="History source changed"):
+        history.read_content(path, **content_args)
+
+
 @pytest.fixture
 def store(tmp_path, monkeypatch):
     root = tmp_path / "host"
@@ -345,6 +411,7 @@ def test_cached_start_and_file_mtime_do_not_refresh_the_event_clock(store):
 def test_every_cold_and_incremental_read_obeys_the_byte_budget(store, monkeypatch):
     path = write_long(store)
     original = Path.open
+    source_open = observer.open_rollout
     reads = [0]
 
     class Counted:
@@ -373,9 +440,10 @@ def opened(target, mode="r", *args, **kwargs):
         stream = original(target, mode, *args, **kwargs)
         if target.parent == store[2] and mode == "rb":
             return Counted(stream, host=False)
-        return Counted(stream) if target == path and mode == "rb" else stream
+        return stream
 
     monkeypatch.setattr(Path, "open", opened)
+    monkeypatch.setattr(observer, "open_rollout", lambda target: Counted(source_open(target)))
     for _ in range(30):
         reads[0] = 0
         result = observe(store, cached=True)
@@ -395,13 +463,16 @@ def opened(target, mode="r", *args, **kwargs):
 
 
 @pytest.mark.parametrize("replace", [False, True])
+@pytest.mark.parametrize("cached", [False, True])
 def test_a_new_end_or_rotation_during_read_is_not_reported_as_open(
-    store, monkeypatch, replace
+    store, monkeypatch, replace, cached
 ):
     path = store[1]
     write(path, encode(event("task_started")))
     original = Path.open
+    source_open = observer.open_rollout
     injected = [False]
+    rotation_errors = []
 
     class Racing:
         def __init__(self, stream):
@@ -414,7 +485,11 @@ def read(self, count=-1):
                 if replace:
                     replacement = path.with_suffix(".replacement")
                     write(replacement, encode(event("task_complete", T2)))
-                    os.replace(replacement, path)
+                    try:
+                        replace_rollout(replacement, path)
+                    except OSError as error:
+                        rotation_errors.append(error)
+                        raise
                 else:
                     with original(path, "ab") as stream:
                         stream.write(encode(event("task_complete", T2)))
@@ -429,13 +504,15 @@ def __enter__(self):
         def __exit__(self, *args):
             return self.stream.__exit__(*args)
 
-    def opened(target, mode="r", *args, **kwargs):
-        stream = original(target, mode, *args, **kwargs)
-        return Racing(stream) if target == path and mode == "rb" else stream
+    def opened(target):
+        return Racing(source_open(target))
 
-    monkeypatch.setattr(Path, "open", opened)
-    assert observe(store).state is HostThreadState.UNKNOWN
-    assert converge(store).state is HostThreadState.IDLE
+    monkeypatch.setattr(observer, "open_rollout", opened)
+    assert observe(store, cached=cached).state is HostThreadState.UNKNOWN
+    assert not rotation_errors, f"read-only observation blocked host rotation: {rotation_errors}"
+    result = converge(store, cached=cached)
+    assert result.state is HostThreadState.IDLE
+    assert result.last_turn_ended_at == T2
 
 
 @pytest.mark.parametrize("chunk", [1, 7, 41, 4096])

tests/test_codex_rollout_file.py

@@ -0,0 +1,103 @@
+"""Native read-only handle semantics, independent of activity classification."""
+
+import os
+
+import pytest
+
+from loopx.hosts.codex_rollout_file import open_rollout
+
+
+def test_reader_is_binary_read_only_and_not_inherited(tmp_path):
+    path = tmp_path / "rollout 中文.jsonl"
+    content = b"old\r\n\x1a\x00\xff"
+    path.write_bytes(content)
+    with open_rollout(path) as reader:
+        assert not os.get_inheritable(reader.fileno())
+        assert not reader.writable()
+        with pytest.raises(OSError):
+            os.write(reader.fileno(), b"changed")
+        assert reader.read() == content
+    assert reader.closed
+    assert path.read_bytes() == content
+
+
+def test_host_can_rename_an_open_rollout_and_publish_a_new_one(tmp_path):
+    path = tmp_path / "rollout 中文.jsonl"
+    rotated = tmp_path / "previous.jsonl"
+    path.write_bytes(b"old")
+    with open_rollout(path) as reader:
+        path.rename(rotated)
+        path.write_bytes(b"new")
+        assert reader.read() == b"old"
+        assert path.read_bytes() == b"new"
+    assert rotated.read_bytes() == b"old"
+
+
+def test_missing_rollout_is_not_created(tmp_path):
+    missing = tmp_path / "missing.jsonl"
+    with pytest.raises(FileNotFoundError):
+        open_rollout(missing)
+    assert not missing.exists()
+
+
+def test_exception_closes_reader_without_deleting_source(tmp_path):
+    path = tmp_path / "rollout.jsonl"
+    path.write_bytes(b"retained")
+    with pytest.raises(ArithmeticError):
+        with open_rollout(path) as reader:
+            raise ArithmeticError("synthetic consumer failure")
+    assert reader.closed
+    assert path.read_bytes() == b"retained"
+
+
+@pytest.mark.skipif(os.name != "nt", reason="Native Windows handle ownership")
+@pytest.mark.parametrize("failure_at", ["descriptor", "stream"])
+def test_native_handle_is_released_when_reader_construction_fails(tmp_path, monkeypatch, failure_at):
+    import ctypes
+    import msvcrt
+    from ctypes import wintypes
+
+    kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+    information = kernel32.GetHandleInformation
+    information.argtypes = (wintypes.HANDLE, ctypes.POINTER(wintypes.DWORD))
+    information.restype = wintypes.BOOL
+    close_handle = kernel32.CloseHandle
+    close_handle.argtypes = (wintypes.HANDLE,)
+    close_handle.restype = wintypes.BOOL
+    path = tmp_path / "rollout.jsonl"
+    path.write_bytes(b"retained")
+    handles, descriptors = [], []
+    original = msvcrt.open_osfhandle
+    failure = OSError("synthetic reader construction failure")
+
+    def adopt(handle, flags):
+        handles.append(handle)
+        assert information(handle, ctypes.byref(wintypes.DWORD()))
+        if failure_at == "descriptor":
+            raise failure
+        descriptor = original(handle, flags)
+        descriptors.append(descriptor)
+        return descriptor
+
+    def stream_failure(*args, **kwargs):
+        raise failure
+
+    monkeypatch.setattr(msvcrt, "open_osfhandle", adopt)
+    if failure_at == "stream":
+        monkeypatch.setattr(os, "fdopen", stream_failure)
+    try:
+        with pytest.raises(OSError) as caught:
+            open_rollout(path)
+        assert caught.value is failure
+        assert len(handles) == 1
+        assert not information(handles[0], ctypes.byref(wintypes.DWORD()))
+        assert ctypes.get_last_error() == 6  # ERROR_INVALID_HANDLE
+        assert path.read_bytes() == b"retained"
+    finally:
+        # A failing regression must not itself leak the real native resource.
+        for handle in handles:
+            if information(handle, ctypes.byref(wintypes.DWORD())):
+                if descriptors:
+                    os.close(descriptors[0])
+                else:
+                    close_handle(handle)