diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml index 2c208e9c61..93643fbbb4 100644 --- a/.github/workflows/python-tests.yml +++ b/.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 diff --git a/docs/reference/protocols/host-integration-surface-v0.md b/docs/reference/protocols/host-integration-surface-v0.md index 6542357027..8efd14a653 100644 --- a/docs/reference/protocols/host-integration-surface-v0.md +++ b/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. diff --git a/loopx/capabilities/native_chat/codex_history.py b/loopx/capabilities/native_chat/codex_history.py index 19c789080f..fbeac75a91 100644 --- a/loopx/capabilities/native_chat/codex_history.py +++ b/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") diff --git a/loopx/codex_app_thread_activity.py b/loopx/codex_app_thread_activity.py index f7af51b876..258db796ff 100644 --- a/loopx/codex_app_thread_activity.py +++ b/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) diff --git a/loopx/hosts/codex_rollout_file.py b/loopx/hosts/codex_rollout_file.py new file mode 100644 index 0000000000..8e6061aca9 --- /dev/null +++ b/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 diff --git a/tests/test_codex_rollout_cursor.py b/tests/test_codex_rollout_cursor.py index 7582ef0c5f..4b59ab8d4d 100644 --- a/tests/test_codex_rollout_cursor.py +++ b/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]) diff --git a/tests/test_codex_rollout_file.py b/tests/test_codex_rollout_file.py new file mode 100644 index 0000000000..2d1a5406b8 --- /dev/null +++ b/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)