fix: qualify observed Codex replay profiles and native validation

loopx/control_plane/turn_driver/codex_context_history.py

@@ -12,12 +12,12 @@
 import hashlib
 import json
 import os
-from contextlib import contextmanager
+import re
+import stat
 from collections import Counter
 from collections.abc import Iterator
+from contextlib import contextmanager
 from pathlib import Path
-import re
-import stat
 from typing import Any, BinaryIO
 
 # A compatibility profile of the native rollout, not a shared control vocabulary.
@@ -33,7 +33,7 @@
 # Bound one decoded protocol record, not the history or the replay interval.
 _MAX_RECORD_BYTES = 16 * 1024 * 1024
 _SHELL_OUTPUT = re.compile(r"\AChunk ID: [^\r\n]+\nWall time: [0-9.]+ seconds\n"
-    r"Process exited with code 0\n(?:Original token count: [0-9]+\n)?Output:\n(.*)\Z", re.S)
+    r"Process exited with code 0\n(?:Original token count: [0-9]+\n)?Output:\n(.*)\Z", re.DOTALL)
 
 
 def _digest(value: object) -> str:
@@ -63,7 +63,9 @@ def _snapshot(event: dict[str, Any]) -> Iterator[tuple[dict[str, Any], BinaryIO]
         if (first.get("type") != "session_meta" or meta.get("id") != event.get("session_id")
                 or meta.get("cwd") != event.get("cwd") or meta.get("forked_from_id")
                 or meta.get("history_mode") != "paginated"
-                or meta.get("cli_version") not in ("0.153.4", "0.162.1", "0.163.0-alpha.11.1")
+                or meta.get("cli_version") not in (
+                    "0.153.4", "0.162.1", "0.163.0-alpha.11.1", "0.163.0-alpha.11.2",
+                    "0.166.0-alpha.5.1")
                 or not event.get("model") or not event.get("permission_mode")):
             raise ValueError("unqualified native replay metadata")
         stream.seek(info.st_size - 1)
@@ -214,9 +216,7 @@ def _matches(anchor: dict[str, Any], snapshot: dict[str, Any], stream: BinaryIO,
         if start > cursor:
             return False
         cursor = max(cursor, stop)
-    if seen_pages != expected_pages or cursor != len(body):
-        return False
-    return True
+    return seen_pages == expected_pages and cursor == len(body)
 
 
 def _append_only(record: dict[str, Any], event: dict[str, Any]) -> bool:
@@ -228,9 +228,8 @@ def _append_only(record: dict[str, Any], event: dict[str, Any]) -> bool:
             return False
     elif kind not in _APPEND_RECORDS:
         return False
-    if kind == "turn_context" and (payload.get("cwd") != event["cwd"] or payload.get("model") != event["model"]):
-        return False
-    return True
+    return kind != "turn_context" or (
+        payload.get("cwd") == event["cwd"] and payload.get("model") == event["model"])
 
 
 def observe_native_read(state: dict[str, Any], event: dict[str, Any],

tests/control_plane/test_codex_context_capture.py

@@ -305,8 +305,19 @@ def response(request, number, message):
             read_commands[name] = descriptor["command"]
             # The public native CLI now shares context-project's paged reader.
             # Keep the independent full-wire oracle separate from its page view.
-            wire_file = Path(descriptor["path"]).parent / "delivery.json"
-            wires[name] = json.loads(wire_file.read_text())
+            wire_file = Path(descriptor["path"])
+            wire_bytes = wire_file.read_bytes()
+            assert hashlib.sha256(wire_bytes).hexdigest() == descriptor["sha256"]
+            if descriptor["mode"] == "full":
+                # Full native delivery reads the advertised original directly;
+                # the retired duplicate delivery.json is not a public oracle.
+                assert str(wire_file) == descriptor["recovery"]["path"]
+                assert descriptor["sha256"] == descriptor["recovery"]["sha256"]
+                wires[name] = {"schema_version": "context_delivery_v1", "mode": "full",
+                    "context": json.loads(wire_bytes), "target": descriptor["target"],
+                    "recovery": descriptor["recovery"], "ancestry": 0}
+            else:
+                wires[name] = json.loads(wire_bytes)
             assert descriptor["mode"] == wires[name]["mode"] == ("full" if name == "cold" or resume_loss else "delta")
             assert "Native input observation independently verifies" not in descriptor["instruction"]
             if name == "warm" and not resume_loss and "content" in descriptor:

tests/control_plane/test_codex_context_resume.py

@@ -3,15 +3,15 @@
 
 import json
 import os
-from pathlib import Path
 import subprocess
 import sys
+from pathlib import Path
 
 import pytest
 
-from loopx.control_plane.turn_driver import codex_context as context
-from tests.control_plane.test_codex_context_lifecycle import setup, project, read_pages
 from loopx.control_plane.quota.turn_envelope import restore_context_delivery
+from loopx.control_plane.turn_driver import codex_context as context
+from tests.control_plane.test_codex_context_lifecycle import project, read_pages, setup
 
 
 def assert_file_delta(wire, previous, expected):
@@ -81,8 +81,49 @@ def observed_read(root, monkeypatch, *, complete=True, shell_wrapper=False, lost
     return event, history, payload
 
 
+@pytest.mark.parametrize("version", ["0.163.0-alpha.11.2", "0.166.0-alpha.5.1"])
+def test_failed_tool_result_cannot_qualify_new_native_writer(tmp_path, monkeypatch, version):
+    """A successful page nested in a failed tool result is not replay evidence."""
+    event = setup(tmp_path, monkeypatch)
+    history = tmp_path / "rollout.jsonl"
+    append(history, "session_meta", {"id": "session", "cwd": str(tmp_path),
+        "cli_version": version, "history_mode": "paginated"})
+    append(history, "turn_context", {"cwd": str(tmp_path), "model": "fixture"})
+    history.chmod(0o600)
+    event.update(transcript_path=str(history), model="fixture", permission_mode="bypassPermissions")
+    _, first = project(tmp_path, payload={"ok": True,
+        "plan_summary": {"requirements": "Keep original evidence. " * 1000},
+        "heartbeat_receipt": {"turn_instance_id": "first"}})
+
+    repo = Path(__file__).resolve().parents[2]
+    command = [sys.executable, "-m", "loopx.entrypoint", "--format", "json", "quota",
+        "context-hook", "--context-delivery", "on", "--codex-app",
+        "--codex-context-dir", str(tmp_path)]
+    environment = {**os.environ, "PYTHONPATH": str(repo), "CODEX_THREAD_ID": "session"}
+    for index, page in enumerate(read_pages(tmp_path, first)):
+        call = f"failed-reader-{index}"
+        response = {"isError": True,
+            "content": [{"type": "text", "text": json.dumps(page)}]}
+        append(history, "response_item", {"type": "function_call_output", "call_id": call,
+            "output": json.dumps(response)})
+        result = subprocess.run(command, input=json.dumps({**event, "hook_event_name": "PostToolUse",
+            "tool_use_id": call, "tool_response": response}), capture_output=True, text=True,
+            cwd=tmp_path, env=environment, timeout=20, check=False)
+        assert result.returncode == 0, result.stderr
+        assert json.loads(result.stdout) == {}
+
+    before = context._state(tmp_path)
+    assert "candidate" not in before
+    resumed = subprocess.run(command, input=json.dumps({**event, "source": "resume"}),
+        capture_output=True, text=True, cwd=tmp_path, env=environment, timeout=20, check=False)
+    assert resumed.returncode == 0, resumed.stderr
+    recovery = json.loads(resumed.stdout)
+    assert "quota context-read" in recovery["hookSpecificOutput"]["additionalContext"]
+    assert "candidate" not in context._state(tmp_path)
+
+
 @pytest.mark.parametrize("shell_wrapper", [False, True])
-@pytest.mark.parametrize("version", ["0.162.1", "0.163.0-alpha.11.1"])
+@pytest.mark.parametrize("version", ["0.162.1", "0.163.0-alpha.11.1", "0.163.0-alpha.11.2", "0.166.0-alpha.5.1"])
 def test_completed_native_resume_keeps_delta_without_old_capture_prefetch(tmp_path, monkeypatch, shell_wrapper, version):
     event, history, payload = observed_read(tmp_path, monkeypatch, shell_wrapper=shell_wrapper, version=version)
     before = context._state(tmp_path)
@@ -102,6 +143,7 @@ def test_completed_native_resume_keeps_delta_without_old_capture_prefetch(tmp_pa
 @pytest.mark.parametrize("code_mode", [False, True])
 def test_compact_recovery_preserves_turn_but_next_turn_starts_new_file_baseline(tmp_path, monkeypatch, single_page, lose_first, code_mode):
     from argparse import Namespace
+
     from loopx.cli_commands.quota_capture import read_context_capture
 
     event, history, payload = observed_read(tmp_path, monkeypatch)
@@ -160,7 +202,7 @@ def test_compact_recovery_preserves_turn_but_next_turn_starts_new_file_baseline(
 
 @pytest.mark.parametrize("change", ["compact", "rollback", "retained", "unknown", "truncate",
     "replace", "rewrite", "missing", "output", "duplicate", "model", "workspace", "session", "permission", "writable", "malformed"])
-@pytest.mark.parametrize("version", ["0.162.1", "0.163.0-alpha.11.1"])
+@pytest.mark.parametrize("version", ["0.162.1", "0.163.0-alpha.11.1", "0.163.0-alpha.11.2", "0.166.0-alpha.5.1"])
 def test_resume_loss_or_unknown_continuity_falls_back_to_current_full(tmp_path, monkeypatch, change, version):
     event, history, payload = observed_read(tmp_path, monkeypatch, version=version)
     before = context._state(tmp_path)
@@ -201,7 +243,7 @@ def test_resume_loss_or_unknown_continuity_falls_back_to_current_full(tmp_path,
         assert wire["mode"] == "full" and wire["context"] == payload
 
 
-@pytest.mark.parametrize("version", ["0.163.0", "0.163.0-alpha.11.2", "unknown"])
+@pytest.mark.parametrize("version", ["0.163.0", "0.163.0-alpha.11.3", "0.166.0-alpha.5.2", "unknown"])
 def test_unqualified_native_version_cannot_retain_resume_baseline(tmp_path, monkeypatch, version):
     event, _, payload = observed_read(tmp_path, monkeypatch, version=version)
     before = context._state(tmp_path)
@@ -252,7 +294,7 @@ def test_real_hook_rejects_fifo_history_without_waiting_for_a_writer(tmp_path, m
 
     def hook(values):
         result = subprocess.run(cli, input=json.dumps({**event, **values}), capture_output=True,
-            text=True, timeout=5, cwd=Path(__file__).resolve().parents[2])
+            text=True, timeout=5, cwd=Path(__file__).resolve().parents[2], check=False)
         assert result.returncode == 0, result.stderr
         return json.loads(result.stdout)
 
@@ -301,6 +343,7 @@ def test_complete_eof_cannot_hide_a_prior_page_missing_from_native_replay(tmp_pa
 @pytest.mark.parametrize("prefix_mib", [0, 96])
 def test_unrelated_history_size_does_not_change_replay_or_decode_cost(tmp_path, monkeypatch, prefix_mib):
     import tracemalloc
+
     from loopx.control_plane.turn_driver import codex_context_history as history_io
 
     event, history, payload = observed_read(tmp_path, monkeypatch, history_prefix_mib=prefix_mib)

tests/managed_effect_runtime_fixture.py

@@ -1,6 +1,7 @@
 """Retire runtimes in a finished test's disposable namespace, via their owner."""
 from __future__ import annotations
 
+import os
 from pathlib import Path
 import tempfile
 from uuid import uuid4
@@ -13,15 +14,29 @@ def capture_disposable_effect_runtimes(root: Path) -> list[tuple[Path, str]]:
         return []
     owned_root = root.resolve()
     captured = []
-    for locator in root.glob("**/loopx-effect-runtime-*/runtime-*.json"):
-        # Never follow a fixture's borrowed namespace or locator outside the
-        # test directory. Copied packages can legitimately own several revisions.
-        if locator.is_symlink() or not locator.resolve().is_relative_to(owned_root):
-            continue
-        try:
-            captured.append((locator, locator.read_text(encoding="utf-8")))
-        except (OSError, UnicodeError):
+
+    def tolerate_retirement(error: OSError) -> None:
+        if not isinstance(error, (FileNotFoundError, PermissionError)):
+            raise error
+
+    # Runtime retirement can remove a directory while this census traverses
+    # it. Walk skips that vanished branch and still captures surviving owners.
+    for directory, _, names in os.walk(root, onerror=tolerate_retirement):
+        folder = Path(directory)
+        if folder == root or not folder.name.startswith("loopx-effect-runtime-"):
             continue
+        for name in names:
+            if not name.startswith("runtime-") or not name.endswith(".json"):
+                continue
+            locator = folder / name
+            # Never follow a fixture's borrowed namespace or locator outside
+            # the test directory. Copied packages can own several revisions.
+            if locator.is_symlink() or not locator.resolve().is_relative_to(owned_root):
+                continue
+            try:
+                captured.append((locator, locator.read_text(encoding="utf-8")))
+            except (OSError, UnicodeError):
+                continue
     return captured
 
 

tests/test_managed_effect_runtime_fixture.py

@@ -2,8 +2,10 @@
 from __future__ import annotations
 
 import json
+import errno
 import os
 from pathlib import Path
+import shutil
 import subprocess
 import sys
 import textwrap
@@ -14,6 +16,44 @@
 from loopx.control_plane import effect_runtime
 
 
+def test_capture_continues_after_a_runtime_directory_disappears(tmp_path, monkeypatch):
+    disappearing = tmp_path / "retiring"
+    removed = disappearing / "loopx-effect-runtime-gone" / "runtime-gone.json"
+    retained = tmp_path / "remaining" / "loopx-effect-runtime-live" / "runtime-live.json"
+    for locator in (removed, retained):
+        locator.parent.mkdir(parents=True)
+        locator.write_text("synthetic locator", encoding="utf-8")
+    scandir = os.scandir
+    removals = []
+
+    def retire_before_scan(path):
+        if isinstance(path, (str, bytes, os.PathLike)) and Path(path) == disappearing:
+            if not removals:
+                shutil.rmtree(disappearing)
+                removals.append(disappearing)
+        return scandir(path)
+
+    monkeypatch.setattr(os, "scandir", retire_before_scan)
+    captured = capture_disposable_effect_runtimes(tmp_path)
+    assert removals == [disappearing]
+    assert captured == [(retained, "synthetic locator")]
+    assert retained.read_text(encoding="utf-8") == "synthetic locator"
+
+
+def test_capture_preserves_unexpected_directory_read_errors(tmp_path, monkeypatch):
+    scandir = os.scandir
+
+    def unreadable(path):
+        if isinstance(path, (str, bytes, os.PathLike)) and Path(path) == tmp_path:
+            raise OSError(errno.EIO, "synthetic directory read failure")
+        return scandir(path)
+
+    with monkeypatch.context() as failing:
+        failing.setattr(os, "scandir", unreadable)
+        with pytest.raises(OSError, match="synthetic directory read failure"):
+            capture_disposable_effect_runtimes(tmp_path)
+
+
 @pytest.fixture
 def nested_pytest_environment(tmp_path):
     repository = Path(__file__).resolve().parents[1]