From 38fcb0eb3686b2c5351b2709381444903cb3cde8 Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Thu, 5 Nov 2026 09:51:37 +0800 Subject: [PATCH 1/5] fix(turn-driver): ask the host for its no-follow flags O_NOFOLLOW, O_DIRECTORY and os.getuid are POSIX-only attributes. Spelling one directly raises AttributeError on a host that does not define it, and it raises before the bounded private regular-file check on the next line can refuse anything, so each of these readers reported an untyped crash in place of its own refusal: - read_native_trace_file is reachable on any host and had no gate, so the observation it exists to produce never appeared. - the recovery readback is reachable without the platform gate verify_state_backup applies, and its os.getuid() failure came after the open had already succeeded. - checkpoint_retention._sync already tested os.name == "posix", but never asked whether the host that satisfies that test also defines O_DIRECTORY. Read the flag this host has, as the peer, completion, decision-capture and effect-runtime readers already do, and refuse a symlink explicitly where the open cannot. The recovery reader reports its own recovery_input_unavailable instead of letting AttributeError escape. Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- loopx/control_plane/state_recovery_io.py | 15 +++++++++++++-- .../turn_driver/codex_input_trace.py | 18 +++++++++++++++++- loopx/hosts/checkpoint_retention.py | 7 +++++-- 3 files changed, 35 insertions(+), 5 deletions(-) diff --git a/loopx/control_plane/state_recovery_io.py b/loopx/control_plane/state_recovery_io.py index f0207b5225..aacfd118bf 100644 --- a/loopx/control_plane/state_recovery_io.py +++ b/loopx/control_plane/state_recovery_io.py @@ -105,14 +105,25 @@ def _audit_owners(request: dict[str, Any]) -> dict[str, Any]: @contextmanager def _regular_input(path: Path, limit: int) -> Iterator[BinaryIO]: + # O_NOFOLLOW and os.getuid are POSIX-only. A bare reference to either raises + # AttributeError on a host that does not define it, so this reader would + # surface an untyped crash instead of its own refusal -- and the readback is + # reachable without the platform gate verify_state_backup applies. Ask for + # what this host has, and refuse a symlink here when the open cannot. + getuid = getattr(os, "getuid", None) + if getuid is None: + raise StateRecoveryError("recovery_input_unavailable") + no_follow = getattr(os, "O_NOFOLLOW", 0) + if not no_follow and path.is_symlink(): + raise StateRecoveryError("recovery_input_unavailable") try: - fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) + fd = os.open(path, os.O_RDONLY | no_follow) except OSError as exc: raise StateRecoveryError("recovery_input_unavailable") from exc with os.fdopen(fd, "rb") as handle: info = os.fstat(handle.fileno()) _require(stat.S_ISREG(info.st_mode), "recovery_input_not_regular") - _require(info.st_uid == os.getuid() and not info.st_mode & 0o077, "recovery_input_not_private") + _require(info.st_uid == getuid() and not info.st_mode & 0o077, "recovery_input_not_private") _require(0 < info.st_size <= limit, "recovery_capacity_exceeded") yield handle diff --git a/loopx/control_plane/turn_driver/codex_input_trace.py b/loopx/control_plane/turn_driver/codex_input_trace.py index 4fd5ecfe5c..5a3ba6dfd3 100644 --- a/loopx/control_plane/turn_driver/codex_input_trace.py +++ b/loopx/control_plane/turn_driver/codex_input_trace.py @@ -13,10 +13,26 @@ from .decision_capture import capture_page_range as _capture_page_range +def _no_follow_flag() -> int: + """The host's no-follow flag, or 0 where it has none. + + `O_NOFOLLOW` and `O_NONBLOCK` are POSIX-only attributes: referencing them + bare raises AttributeError on a host that does not define them, before the + bounded-regular-file check below can refuse anything. Read them per call so + the probe describes this host, as the peer and completion readers do. + """ + return getattr(os, "O_NOFOLLOW", 0) + + def read_native_trace_file(path: Path, *, tail: bool = False) -> bytes: from ..effect_runtime import MAX_REQUEST_BYTES - descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + no_follow = _no_follow_flag() + if not no_follow and path.is_symlink(): + # Without O_NOFOLLOW the open cannot refuse a symlink for us. Refuse it + # here rather than letting a diagnostic read follow an arbitrary target. + raise ValueError("native observation must not be a symlink") + descriptor = os.open(path, os.O_RDONLY | no_follow | getattr(os, "O_NONBLOCK", 0)) with os.fdopen(descriptor, "rb") as stream: metadata = os.fstat(stream.fileno()) if not stat.S_ISREG(metadata.st_mode) or (metadata.st_size > MAX_REQUEST_BYTES and not tail): diff --git a/loopx/hosts/checkpoint_retention.py b/loopx/hosts/checkpoint_retention.py index 7b12f3fe0d..af7f7be8d6 100644 --- a/loopx/hosts/checkpoint_retention.py +++ b/loopx/hosts/checkpoint_retention.py @@ -108,8 +108,11 @@ def retention_plan(bank: Path, policy: dict | None, *, now: float | None = None) def _sync(path: Path) -> None: - if os.name == "posix": - fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY) + # O_DIRECTORY is POSIX-only, so the platform gate and the flag lookup must + # name the same thing rather than trusting `os.name` alone. + directory = getattr(os, "O_DIRECTORY", 0) + if os.name == "posix" and directory: + fd = os.open(path, os.O_RDONLY | directory) try: os.fsync(fd) finally: From 0c988b8bcea83ac921dd176616605f994b4b41ca Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Thu, 5 Nov 2026 09:51:42 +0800 Subject: [PATCH 2/5] test(architecture): pin the bare POSIX flag sites A bare os.O_NOFOLLOW, os.O_DIRECTORY or os.O_NONBLOCK raises AttributeError on a host that does not define it, and it raises before the bounded regular-file check on the next line can refuse anything. The repository already reads these flags with getattr in the peer, completion, decision-capture and effect-runtime readers. Pin every remaining site per file so a new one needs a recorded reason, and assert the readers this change fixed stay at zero. A lower measurement is accepted, so retiring one of these sites does not have to edit a fixture that would then describe a tree it no longer matches. Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- tests/architecture/posix_flag_ceilings.json | 12 ++ .../test_posix_flag_portability.py | 159 ++++++++++++++++++ 2 files changed, 171 insertions(+) create mode 100644 tests/architecture/posix_flag_ceilings.json create mode 100644 tests/architecture/test_posix_flag_portability.py diff --git a/tests/architecture/posix_flag_ceilings.json b/tests/architecture/posix_flag_ceilings.json new file mode 100644 index 0000000000..8a69273a3d --- /dev/null +++ b/tests/architecture/posix_flag_ceilings.json @@ -0,0 +1,12 @@ +{ + "schema": "loopx_posix_flag_ceilings_v0", + "baseline_commit": "b4216d8c79f7f31785cf3bb6845f6847fcf0a6a6", + "note": "Per-file ceiling on bare `os.O_NOFOLLOW` / `os.O_DIRECTORY` / `os.O_NONBLOCK` references under loopx/. A ceiling is a maximum, not a target: the guard fails when a file exceeds it or a new file appears, and accepts a lower measurement, so retiring one of these sites needs no edit here. Lower or drop an entry whenever a file is fixed, so the ceiling does not describe an older tree than the code.", + "ceiling": { + "loopx/cli_commands/quota_capture.py": 2, + "loopx/control_plane/collaboration/result_files.py": 6, + "loopx/control_plane/effect_runtime.py": 2, + "loopx/control_plane/turn_driver/codex_context.py": 4, + "loopx/control_plane/turn_driver/codex_context_history.py": 2 + } +} diff --git a/tests/architecture/test_posix_flag_portability.py b/tests/architecture/test_posix_flag_portability.py new file mode 100644 index 0000000000..86a8da7427 --- /dev/null +++ b/tests/architecture/test_posix_flag_portability.py @@ -0,0 +1,159 @@ +"""No module grows a bare reference to a POSIX-only `os` flag. + +`os.O_NOFOLLOW`, `os.O_DIRECTORY` and `os.O_NONBLOCK` are POSIX-only +attributes. Spelling one directly raises `AttributeError` on a host that does +not define it, and it raises it before the bounded-regular-file check on the +next line can refuse anything: the caller reports an untyped crash in place of +its own refusal, or the diagnostic it was opened to produce never appears. +`from os import O_NOFOLLOW` is worse still: the module fails at import time. + +The repository already reads these flags with `getattr(os, "", 0)` in +the peer, completion, decision-capture and effect-runtime readers, and in +`codex_input_trace`, `state_recovery_io` and `checkpoint_retention` after +this change. This guard keeps a new bare site out — including the sites in +files another change is reworking. + +Ceilings live in `posix_flag_ceilings.json`, following +`test_package_kernel_import_ratchet`: an increase needs a recorded decision, +and a decrease is accepted as-is so that fixing one of these files does not +have to edit a fixture that would then describe a tree it no longer matches. +""" + +from __future__ import annotations + +import ast +import json +import re +from pathlib import Path +from typing import Any + +REPO_ROOT = Path(__file__).resolve().parents[2] +LOOPX_ROOT = REPO_ROOT / "loopx" +CEILINGS_FIXTURE = Path(__file__).with_name("posix_flag_ceilings.json") +CEILINGS_SCHEMA = "loopx_posix_flag_ceilings_v0" +POSIX_ONLY_FLAGS = frozenset({"O_NOFOLLOW", "O_DIRECTORY", "O_NONBLOCK"}) +_COMMIT_SHA = re.compile(r"^[0-9a-f]{40}$") + + +def bare_flag_uses(source: str) -> list[tuple[int, str]]: + """Every direct `os.` or `from os import ` in one module.""" + tree = ast.parse(source) + aliases = { + alias.asname or alias.name + for node in ast.walk(tree) + if isinstance(node, ast.Import) + for alias in node.names + if alias.name == "os" + } + uses: list[tuple[int, str]] = [] + for node in ast.walk(tree): + if ( + isinstance(node, ast.Attribute) + and isinstance(node.value, ast.Name) + and node.value.id in aliases + and node.attr in POSIX_ONLY_FLAGS + ): + uses.append((node.lineno, f"{node.value.id}.{node.attr}")) + elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module == "os": + uses.extend( + (node.lineno, f"from os import {alias.name}") + for alias in node.names + if alias.name in POSIX_ONLY_FLAGS + ) + return sorted(uses) + + +def observed_ceiling(loopx_root: Path | None = None) -> dict[str, list[tuple[int, str]]]: + root = loopx_root or LOOPX_ROOT + observed: dict[str, list[tuple[int, str]]] = {} + for path in sorted(root.rglob("*.py")): + uses = bare_flag_uses(path.read_text(encoding="utf-8")) + if uses: + observed[path.relative_to(REPO_ROOT).as_posix()] = uses + return observed + + +def _ceiling() -> dict[str, Any]: + payload = json.loads(CEILINGS_FIXTURE.read_text(encoding="utf-8")) + assert payload["schema"] == CEILINGS_SCHEMA, ( + f"{CEILINGS_FIXTURE.name} schema is {payload['schema']!r}, expected {CEILINGS_SCHEMA!r}" + ) + assert _COMMIT_SHA.match(str(payload["baseline_commit"])), ( + "baseline_commit must be the full 40-character commit the ceilings were measured at" + ) + ceiling = payload["ceiling"] + assert isinstance(ceiling, dict) and ceiling, "ceiling must map module path -> positive count" + assert all( + isinstance(path, str) and isinstance(count, int) and count > 0 + for path, count in ceiling.items() + ), "every ceiling entry needs a module path and a positive count" + assert list(ceiling) == sorted(ceiling), "ceiling entries must be sorted by module path" + return payload + + +def test_no_module_exceeds_its_bare_posix_flag_ceiling() -> None: + ceiling = _ceiling()["ceiling"] + observed = observed_ceiling() + + exceeded = { + path: {"ceiling": ceiling.get(path, 0), "observed": uses} + for path, uses in observed.items() + if len(uses) > ceiling.get(path, 0) + } + assert not exceeded, ( + "bare POSIX-only os flags grew: " + + json.dumps( + {path: [f"{line}: {what}" for line, what in uses] for path, uses in exceeded.items()}, + indent=2, + ) + + ". Read the flag with `getattr(os, \"\", 0)` the way the peer, completion " + "and decision-capture readers do, so a host without it refuses or degrades on its " + "own terms instead of raising AttributeError first. Raising a ceiling in " + f"{CEILINGS_FIXTURE.name} needs a recorded reason." + ) + + +def test_fixed_readers_keep_no_bare_posix_flag() -> None: + """The readers whose fallback this guard protects stay at zero.""" + observed = observed_ceiling() + fixed = ( + "loopx/control_plane/turn_driver/codex_input_trace.py", + "loopx/control_plane/state_recovery_io.py", + "loopx/hosts/checkpoint_retention.py", + ) + remaining = {path: observed[path] for path in fixed if path in observed} + assert not remaining, ( + f"these readers spell a POSIX-only flag directly again: {remaining}. They are " + "reachable on hosts that do not define the flag, so the open must request it " + "with `getattr(os, ...)`." + ) + + +def test_scan_sees_bare_alias_and_from_import_but_not_getattr() -> None: + source = ( + "import os\n" + "import os as host_os\n" + "from os import O_DIRECTORY\n" + "def f(path):\n" + " first = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)\n" + " second = host_os.open(path, host_os.O_RDONLY | host_os.O_NOFOLLOW)\n" + " third = getattr(host_os, \"O_NONBLOCK\", 0)\n" + " return first, second, third, os.O_RDONLY\n" + ) + assert bare_flag_uses(source) == [ + (3, "from os import O_DIRECTORY"), + (5, "os.O_NOFOLLOW"), + (6, "host_os.O_NOFOLLOW"), + ] + + +def test_scan_ignores_a_name_that_is_not_the_os_module() -> None: + """A local `os` binding is not the module, and a flag name alone is not a use.""" + source = ( + "class Holder:\n" + " O_NOFOLLOW = 1\n" + "def f(holder):\n" + " flag = \"O_DIRECTORY\"\n" + " return holder.O_NONBLOCK, Holder.O_NOFOLLOW, flag\n" + ) + assert bare_flag_uses(source) == [] From 9ff91c9c3595f028d601bbc2b44e34d6a0d2578b Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Thu, 5 Nov 2026 09:53:23 +0800 Subject: [PATCH 3/5] test(portability): cover hosts without the POSIX attributes Deleting os.O_NOFOLLOW, os.O_NONBLOCK, os.O_DIRECTORY or os.getuid for the duration of one call is the branch a host that lacks them takes. Six of these seven cases failed on the pre-fix tree with `AttributeError: module 'os' has no attribute 'O_NOFOLLOW'`, which is the same failure a native Windows host reports, so the cases reproduce the defect on a POSIX checkout. The Windows lane runs the guard and these cases as well, so the branch a real Windows host takes stays qualified in the place it is reached. Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- .github/workflows/python-tests.yml | 11 ++ .../control_plane/test_posix_flag_fallback.py | 100 ++++++++++++++++++ 2 files changed, 111 insertions(+) create mode 100644 tests/control_plane/test_posix_flag_fallback.py diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml index 7bfe92f200..7a309c8eab 100644 --- a/.github/workflows/python-tests.yml +++ b/.github/workflows/python-tests.yml @@ -967,6 +967,17 @@ jobs: tests/control_plane/test_document_io.py::test_ordinary_windows_long_path_publish_preserves_bytes_and_create_only tests/control_plane/test_todo_completion_validation.py::test_private_validation_store_is_owner_only_and_detects_tampering + - name: Qualify portable file-flag probes on native Windows + # read_native_trace_file, the recovery readback and checkpoint retention + # used to reference os.O_NOFOLLOW, os.O_DIRECTORY and os.getuid + # directly, so a host that does not define one raised AttributeError + # before the reader could refuse. Only a native Windows host takes that + # branch for real; the simulated cases cover it on POSIX. + run: >- + python -m pytest -q + tests/architecture/test_posix_flag_portability.py + tests/control_plane/test_posix_flag_fallback.py + - name: Check Windows collaboration lock sidecars run: python -m pytest -q --basetemp .t tests/test_collaboration_sidecar_scan.py diff --git a/tests/control_plane/test_posix_flag_fallback.py b/tests/control_plane/test_posix_flag_fallback.py new file mode 100644 index 0000000000..cfb78d13e2 --- /dev/null +++ b/tests/control_plane/test_posix_flag_fallback.py @@ -0,0 +1,100 @@ +"""Readers keep working where the host does not define a POSIX-only attribute. + +`os.O_NOFOLLOW`, `os.O_NONBLOCK`, `os.O_DIRECTORY` and `os.getuid` are POSIX +attributes. Removing one for the duration of a single call is what +`getattr(os, "", ...)` sees on a host that lacks it, and it is the same +branch a native Windows host takes. Each reader must refuse on its own terms +there: an untyped AttributeError, or an open that quietly follows a symlink, +would replace a diagnosis the caller asked for. +""" + +from __future__ import annotations + +import os + +import pytest + +from loopx.control_plane.state_recovery_io import StateRecoveryError, _regular_input +from loopx.control_plane.turn_driver.codex_input_trace import read_native_trace_file +from loopx.hosts.checkpoint_retention import _sync + +OWNER_MODEL = pytest.mark.skipif( + getattr(os, "getuid", None) is None, + reason="this host has no owner model to check the private regular file against", +) + + +def _symlink(path, target) -> None: + try: + path.symlink_to(target) + except (OSError, NotImplementedError): + pytest.skip("this host cannot create a file symlink") + + +def test_native_trace_read_survives_a_host_without_the_posix_flags(tmp_path, monkeypatch): + trace = tmp_path / "trace.jsonl" + trace.write_bytes(b'{"seq": 1}\n') + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + monkeypatch.delattr(os, "O_NONBLOCK", raising=False) + assert read_native_trace_file(trace) == b'{"seq": 1}\n' + + +def test_native_trace_read_refuses_a_symlink_without_the_flag(tmp_path, monkeypatch): + original = tmp_path / "original.jsonl" + original.write_bytes(b'{"seq": 1}\n') + trace = tmp_path / "trace.jsonl" + _symlink(trace, original) + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + with pytest.raises(ValueError, match="symlink"): + read_native_trace_file(trace) + + +@OWNER_MODEL +def test_recovery_readback_refuses_a_host_without_an_owner_model(tmp_path, monkeypatch): + payload = tmp_path / "recovery-audit.json" + payload.write_bytes(b'{"ok": false}') + payload.chmod(0o600) + monkeypatch.delattr(os, "getuid", raising=False) + with pytest.raises(StateRecoveryError, match="recovery_input_unavailable"): + with _regular_input(payload, 1024): + pass + + +@OWNER_MODEL +def test_recovery_readback_survives_a_host_without_the_no_follow_flag(tmp_path, monkeypatch): + payload = tmp_path / "recovery-audit.json" + payload.write_bytes(b'{"ok": false}') + payload.chmod(0o600) + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + with _regular_input(payload, 1024) as handle: + assert handle.read() == b'{"ok": false}' + + +@OWNER_MODEL +def test_recovery_readback_refuses_a_symlink_without_the_flag(tmp_path, monkeypatch): + original = tmp_path / "original.json" + original.write_bytes(b'{"ok": false}') + original.chmod(0o600) + payload = tmp_path / "recovery-audit.json" + _symlink(payload, original) + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + with pytest.raises(StateRecoveryError, match="recovery_input_unavailable"): + with _regular_input(payload, 1024): + pass + + +def test_checkpoint_sync_skips_a_host_without_the_directory_flag(tmp_path, monkeypatch): + synced = [] + monkeypatch.setattr(os, "fsync", synced.append) + monkeypatch.delattr(os, "O_DIRECTORY", raising=False) + _sync(tmp_path) + assert synced == [] + + +def test_checkpoint_sync_still_syncs_where_the_directory_flag_exists(tmp_path, monkeypatch): + if not getattr(os, "O_DIRECTORY", 0): + pytest.skip("this host does not define O_DIRECTORY") + synced, real = [], os.fsync + monkeypatch.setattr(os, "fsync", lambda descriptor: synced.append(descriptor) or real(descriptor)) + _sync(tmp_path) + assert len(synced) == 1 From b7aee4ea1e6005dcee8d4a45fe99b358e6fa3597 Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Thu, 5 Nov 2026 11:04:06 +0800 Subject: [PATCH 4/5] fix(io): bind the opened entry and keep the directory flush Two holds were open in the previous fallback. Reading without O_NOFOLLOW checked the path and then opened it, so the entry could be replaced by a link to another file in between: nothing was bound to what was actually opened. Open first, then compare the opened descriptor with the directory entry before one byte is read. A link opens to its target, whose device and inode differ from the link's own, so a swapped or linked entry is refused instead of followed. The native trace reader and the recovery readback share the rule. checkpoint retention treated a missing O_DIRECTORY as permission to skip the flush, so a POSIX host without the flag deleted recovery points and reported a rotation that never became durable. O_DIRECTORY only hardens the open: take the flag the host has, refuse a descriptor that is not the directory, and flush it, as file_lock.write_json_atomic already does. Static-link cases could not see either defect. The new cases replace the entry at the instant it is opened, and one runs through codex_context_in_active_input, so head before this commit fails them. Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- loopx/control_plane/state_recovery_io.py | 17 +++- .../turn_driver/codex_input_trace.py | 41 ++++++---- loopx/hosts/checkpoint_retention.py | 28 ++++--- .../test_checkpoint_retention.py | 20 +++++ .../test_codex_input_retention.py | 25 ++++++ .../control_plane/test_posix_flag_fallback.py | 78 ++++++++++++++++++- 6 files changed, 179 insertions(+), 30 deletions(-) diff --git a/loopx/control_plane/state_recovery_io.py b/loopx/control_plane/state_recovery_io.py index aacfd118bf..ebe24345fc 100644 --- a/loopx/control_plane/state_recovery_io.py +++ b/loopx/control_plane/state_recovery_io.py @@ -113,13 +113,22 @@ def _regular_input(path: Path, limit: int) -> Iterator[BinaryIO]: getuid = getattr(os, "getuid", None) if getuid is None: raise StateRecoveryError("recovery_input_unavailable") - no_follow = getattr(os, "O_NOFOLLOW", 0) - if not no_follow and path.is_symlink(): - raise StateRecoveryError("recovery_input_unavailable") try: - fd = os.open(path, os.O_RDONLY | no_follow) + fd = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) except OSError as exc: raise StateRecoveryError("recovery_input_unavailable") from exc + try: + # Where the host has no no-follow flag the open cannot refuse a link, and + # inspecting the path first only moves the race to the open below. Bind the + # descriptor to the entry before anything is read: a link opens to its + # target, whose device and inode differ from the link's own. + opened, entry = os.fstat(fd), os.lstat(path) + _require(not stat.S_ISLNK(entry.st_mode) + and (entry.st_dev, entry.st_ino) == (opened.st_dev, opened.st_ino), + "recovery_input_unavailable") + except BaseException: + os.close(fd) + raise with os.fdopen(fd, "rb") as handle: info = os.fstat(handle.fileno()) _require(stat.S_ISREG(info.st_mode), "recovery_input_not_regular") diff --git a/loopx/control_plane/turn_driver/codex_input_trace.py b/loopx/control_plane/turn_driver/codex_input_trace.py index 5a3ba6dfd3..5f211f7541 100644 --- a/loopx/control_plane/turn_driver/codex_input_trace.py +++ b/loopx/control_plane/turn_driver/codex_input_trace.py @@ -13,26 +13,41 @@ from .decision_capture import capture_page_range as _capture_page_range -def _no_follow_flag() -> int: - """The host's no-follow flag, or 0 where it has none. +def _open_verified(path: Path) -> int: + """Open the entry at `path` without following a link, on any host. - `O_NOFOLLOW` and `O_NONBLOCK` are POSIX-only attributes: referencing them - bare raises AttributeError on a host that does not define them, before the - bounded-regular-file check below can refuse anything. Read them per call so - the probe describes this host, as the peer and completion readers do. + `O_NOFOLLOW` and `O_NONBLOCK` are POSIX-only attributes, so a bare reference + raises AttributeError on a host that does not define them -- here, before the + bounded-regular-file check below can refuse anything. + + Where the host has no no-follow flag, the open cannot refuse a link for us, + and checking the path first would only move the race: the entry can be + replaced between the check and the open. Bind the descriptor to the entry + instead, and do it before the caller reads one byte: a link opens to its + target, whose device and inode differ from the link's own, so a swapped or + linked entry is refused rather than followed. """ - return getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open( + path, + os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0), + ) + try: + opened = os.fstat(descriptor) + entry = os.lstat(path) + if stat.S_ISLNK(entry.st_mode): + raise ValueError("native observation must not be a symlink") + if (entry.st_dev, entry.st_ino) != (opened.st_dev, opened.st_ino): + raise ValueError("native observation changed while it was opened") + except BaseException: + os.close(descriptor) + raise + return descriptor def read_native_trace_file(path: Path, *, tail: bool = False) -> bytes: from ..effect_runtime import MAX_REQUEST_BYTES - no_follow = _no_follow_flag() - if not no_follow and path.is_symlink(): - # Without O_NOFOLLOW the open cannot refuse a symlink for us. Refuse it - # here rather than letting a diagnostic read follow an arbitrary target. - raise ValueError("native observation must not be a symlink") - descriptor = os.open(path, os.O_RDONLY | no_follow | getattr(os, "O_NONBLOCK", 0)) + descriptor = _open_verified(path) with os.fdopen(descriptor, "rb") as stream: metadata = os.fstat(stream.fileno()) if not stat.S_ISREG(metadata.st_mode) or (metadata.st_size > MAX_REQUEST_BYTES and not tail): diff --git a/loopx/hosts/checkpoint_retention.py b/loopx/hosts/checkpoint_retention.py index af7f7be8d6..610887d8e9 100644 --- a/loopx/hosts/checkpoint_retention.py +++ b/loopx/hosts/checkpoint_retention.py @@ -108,15 +108,25 @@ def retention_plan(bank: Path, policy: dict | None, *, now: float | None = None) def _sync(path: Path) -> None: - # O_DIRECTORY is POSIX-only, so the platform gate and the flag lookup must - # name the same thing rather than trusting `os.name` alone. - directory = getattr(os, "O_DIRECTORY", 0) - if os.name == "posix" and directory: - fd = os.open(path, os.O_RDONLY | directory) - try: - os.fsync(fd) - finally: - os.close(fd) + # `_state` qualifies this transport on POSIX and the rotation moves recovery + # points only after that, so skipping the flush here would report a completed + # rotation that never made it durable. O_DIRECTORY is POSIX-only and only + # hardens the open, so take the flag this host has and keep the guarantee by + # refusing anything that is not the directory itself -- the pattern + # `file_lock.write_json_atomic` already uses. + if os.name != "posix": + return + fd = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) + try: + _require_directory(os.fstat(fd).st_mode, path) + os.fsync(fd) + finally: + os.close(fd) + + +def _require_directory(mode: int, path: Path) -> None: + if not stat.S_ISDIR(mode): + raise ValueError(f"checkpoint sync target is not a directory: {path}") def _receipt(path: Path, record: dict) -> None: diff --git a/tests/control_plane/test_checkpoint_retention.py b/tests/control_plane/test_checkpoint_retention.py index 09a01470af..ae7e981a82 100644 --- a/tests/control_plane/test_checkpoint_retention.py +++ b/tests/control_plane/test_checkpoint_retention.py @@ -1,6 +1,7 @@ """Real local artifact owner, retention CLI and filesystem recovery negatives.""" import json import os +import stat import threading from types import SimpleNamespace @@ -59,6 +60,25 @@ def test_preview_is_read_only_and_execute_preserves_pins_latest_and_unsealed(ban assert transport.rotate_bank(bank, policy, execute=True)["deleted"] == [] +def test_rotation_flushes_directories_without_the_directory_flag(bank, policy, monkeypatch): + # A host can report os.name == "posix" without defining O_DIRECTORY. The + # rotation still moves recovery points, so it must still flush the + # directories the moves changed; skipping that would report a completed + # rotation that was never made durable. + policy["keep_latest"] = 1 + flushed, real_fsync = [], os.fsync + + def recording(descriptor): + flushed.append(stat.S_ISDIR(os.fstat(descriptor).st_mode)) + return real_fsync(descriptor) + + monkeypatch.setattr(os, "fsync", recording) + monkeypatch.delattr(os, "O_DIRECTORY", raising=False) + result = transport.rotate_bank(bank, policy, execute=True) + assert result["deleted"] + assert sum(flushed) >= 2, "the bank and the trash parent must be flushed after the moves" + + def test_stale_preview_and_changed_pin_cannot_delete(bank, policy): preview = transport.rotate_bank(bank, policy) os.utime(bank / "p2/checkpoint.json", (9000, 9000)) diff --git a/tests/control_plane/test_codex_input_retention.py b/tests/control_plane/test_codex_input_retention.py index 4febe3f3f1..ad2a523485 100644 --- a/tests/control_plane/test_codex_input_retention.py +++ b/tests/control_plane/test_codex_input_retention.py @@ -3,6 +3,7 @@ import hashlib import json +import os from pathlib import Path import pytest @@ -188,6 +189,30 @@ def test_unknown_stale_lossy_or_unbound_observation_never_reuses(tmp_path, failu assert not codex_context_in_active_input(observation, baseline=baseline, session_id="session") +def test_trace_link_swapped_after_containment_is_refused_before_the_read(tmp_path, monkeypatch): + """The path is replaced at open time, past the checks the bundle passed.""" + observation, baseline, request, rows = fixture(tmp_path) + trace = Path(observation["trace_bundle"]) / "trace.jsonl" + outside = tmp_path / "outside.jsonl" + outside.write_text(trace.read_text(encoding="utf-8"), encoding="utf-8") + # The reader resolves the bundle first, so compare against the resolved entry. + resolved = trace.resolve() + real_open = os.open + + def swapping(target, flags, *args, **kwargs): + if os.fspath(target) == os.fspath(resolved) and not trace.is_symlink(): + trace.unlink() + try: + trace.symlink_to(outside) + except (OSError, NotImplementedError): + pytest.skip("this host cannot create a file symlink") + return real_open(target, flags, *args, **kwargs) + + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + monkeypatch.setattr(os, "open", swapping) + assert not codex_context_in_active_input(observation, baseline=baseline, session_id="session") + + @pytest.mark.parametrize("pragma", ["", '// @exec: {"max_output_tokens":12000}\n', ' \t// @exec: {"yield_time_ms":1000}\r\n']) @pytest.mark.parametrize("failure", [None, "parent_missing", "parent_changed", "parent_ended", "parent_wrong_call", diff --git a/tests/control_plane/test_posix_flag_fallback.py b/tests/control_plane/test_posix_flag_fallback.py index cfb78d13e2..c029f8ba01 100644 --- a/tests/control_plane/test_posix_flag_fallback.py +++ b/tests/control_plane/test_posix_flag_fallback.py @@ -11,6 +11,7 @@ from __future__ import annotations import os +import stat import pytest @@ -83,12 +84,25 @@ def test_recovery_readback_refuses_a_symlink_without_the_flag(tmp_path, monkeypa pass -def test_checkpoint_sync_skips_a_host_without_the_directory_flag(tmp_path, monkeypatch): - synced = [] - monkeypatch.setattr(os, "fsync", synced.append) +def test_checkpoint_sync_flushes_the_directory_without_the_flag(tmp_path, monkeypatch): + flushed, real = [], os.fsync + + def recording(descriptor): + flushed.append(stat.S_ISDIR(os.fstat(descriptor).st_mode)) + return real(descriptor) + + monkeypatch.setattr(os, "fsync", recording) monkeypatch.delattr(os, "O_DIRECTORY", raising=False) _sync(tmp_path) - assert synced == [] + assert flushed == [True], "the directory must be flushed, not silently skipped" + + +def test_checkpoint_sync_refuses_a_target_that_is_not_a_directory(tmp_path, monkeypatch): + target = tmp_path / "bank.json" + target.write_text("{}") + monkeypatch.delattr(os, "O_DIRECTORY", raising=False) + with pytest.raises(ValueError, match="not a directory"): + _sync(target) def test_checkpoint_sync_still_syncs_where_the_directory_flag_exists(tmp_path, monkeypatch): @@ -98,3 +112,59 @@ def test_checkpoint_sync_still_syncs_where_the_directory_flag_exists(tmp_path, m monkeypatch.setattr(os, "fsync", lambda descriptor: synced.append(descriptor) or real(descriptor)) _sync(tmp_path) assert len(synced) == 1 + + +def _swap_at_open(monkeypatch, path, replacement): + """Replace `path` with a link to `replacement` the instant it is opened.""" + real_open = os.open + + def swapping(target, flags, *args, **kwargs): + if os.fspath(target) == os.fspath(path) and not os.path.islink(path): + path.unlink() + try: + path.symlink_to(replacement) + except (OSError, NotImplementedError): + pytest.skip("this host cannot create a file symlink") + return real_open(target, flags, *args, **kwargs) + + monkeypatch.setattr(os, "open", swapping) + + +def test_native_trace_read_refuses_a_link_swapped_in_at_the_open(tmp_path, monkeypatch): + """Binding the descriptor refuses the entry the check-then-open window admits.""" + outside = tmp_path / "outside.jsonl" + outside.write_bytes(b'{"external": true}\n') + trace = tmp_path / "trace.jsonl" + trace.write_bytes(b'{"seq": 1}\n') + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + _swap_at_open(monkeypatch, trace, outside) + with pytest.raises(ValueError, match="symlink"): + read_native_trace_file(trace) + + +def test_native_trace_read_refuses_the_same_swap_with_the_flag(tmp_path, monkeypatch): + """The native flag refuses it in the open, so the no-flag branch must match.""" + if not getattr(os, "O_NOFOLLOW", 0): + pytest.skip("this host has no no-follow flag to compare against") + outside = tmp_path / "outside.jsonl" + outside.write_bytes(b'{"external": true}\n') + trace = tmp_path / "trace.jsonl" + trace.write_bytes(b'{"seq": 1}\n') + _swap_at_open(monkeypatch, trace, outside) + with pytest.raises(OSError): + read_native_trace_file(trace) + + +@OWNER_MODEL +def test_recovery_readback_refuses_a_link_swapped_in_at_the_open(tmp_path, monkeypatch): + outside = tmp_path / "outside.json" + outside.write_bytes(b'{"external": true}') + outside.chmod(0o600) + payload = tmp_path / "recovery-audit.json" + payload.write_bytes(b'{"ok": false}') + payload.chmod(0o600) + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + _swap_at_open(monkeypatch, payload, outside) + with pytest.raises(StateRecoveryError, match="recovery_input_unavailable"): + with _regular_input(payload, 1024): + pass From 99e9b5d5ed5f0a9ce7b9ce206aa6a16888198041 Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Thu, 5 Nov 2026 15:07:53 +0800 Subject: [PATCH 5/5] test(portability): claim the POSIX-only rotation assertions The new native Windows step ran the whole fallback file, but two sync assertions describe a POSIX bank: _sync returns before flushing when the host is not POSIX, so the flush list stayed empty and the non-directory target raised nothing. Mark those two POSIX-only and pin the other branch with an explicit no-op case that holds on both hosts. Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- tests/control_plane/test_posix_flag_fallback.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/tests/control_plane/test_posix_flag_fallback.py b/tests/control_plane/test_posix_flag_fallback.py index c029f8ba01..9ddd8d2fca 100644 --- a/tests/control_plane/test_posix_flag_fallback.py +++ b/tests/control_plane/test_posix_flag_fallback.py @@ -23,6 +23,10 @@ getattr(os, "getuid", None) is None, reason="this host has no owner model to check the private regular file against", ) +POSIX_SYNC = pytest.mark.skipif( + os.name != "posix", + reason="_sync only qualifies a POSIX bank; other hosts are claimed by the no-op case", +) def _symlink(path, target) -> None: @@ -84,6 +88,7 @@ def test_recovery_readback_refuses_a_symlink_without_the_flag(tmp_path, monkeypa pass +@POSIX_SYNC def test_checkpoint_sync_flushes_the_directory_without_the_flag(tmp_path, monkeypatch): flushed, real = [], os.fsync @@ -97,6 +102,7 @@ def recording(descriptor): assert flushed == [True], "the directory must be flushed, not silently skipped" +@POSIX_SYNC def test_checkpoint_sync_refuses_a_target_that_is_not_a_directory(tmp_path, monkeypatch): target = tmp_path / "bank.json" target.write_text("{}") @@ -105,6 +111,17 @@ def test_checkpoint_sync_refuses_a_target_that_is_not_a_directory(tmp_path, monk _sync(target) +def test_checkpoint_sync_claims_nothing_off_posix(tmp_path, monkeypatch): + calls = [] + monkeypatch.setattr(os, "fsync", lambda descriptor: calls.append(descriptor)) + monkeypatch.setattr(os, "name", "nt") + target = tmp_path / "bank.json" + target.write_text("{}") + _sync(tmp_path) + _sync(target) + assert calls == [], "a host outside the POSIX contract must not report a flush" + + def test_checkpoint_sync_still_syncs_where_the_directory_flag_exists(tmp_path, monkeypatch): if not getattr(os, "O_DIRECTORY", 0): pytest.skip("this host does not define O_DIRECTORY")