fix(turn-driver): ask the host for its no-follow flags

.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
 

loopx/control_plane/state_recovery_io.py

@@ -105,14 +105,34 @@ 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")
     try:
-        fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
+        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")
-        _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
 

loopx/control_plane/turn_driver/codex_input_trace.py

@@ -13,10 +13,41 @@
 from .decision_capture import capture_page_range as _capture_page_range
 
 
+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, 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.
+    """
+    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
 
-    descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+    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):

loopx/hosts/checkpoint_retention.py

@@ -108,12 +108,25 @@ 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)
-        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:

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
+  }
+}

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, "<flag>", 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.<flag>` or `from os import <flag>` 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, \"<flag>\", 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) == []

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))

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",