diff --git a/loopx/control_plane/turn_driver/codex_context_history.py b/loopx/control_plane/turn_driver/codex_context_history.py index 411aadb113..f26c75de60 100644 --- a/loopx/control_plane/turn_driver/codex_context_history.py +++ b/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], diff --git a/tests/control_plane/test_codex_context_capture.py b/tests/control_plane/test_codex_context_capture.py index 4e31db9d33..376fe5dbea 100644 --- a/tests/control_plane/test_codex_context_capture.py +++ b/tests/control_plane/test_codex_context_capture.py @@ -19,9 +19,11 @@ from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime, read_fixture_todos from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection from loopx.control_plane.quota.turn_envelope import prepare_context_delivery, restore_context_delivery +from loopx.control_plane.turn_driver.decision_capture import read_context_read_reference from test_quota_settlement_cli import AGENT_ID, GOAL_ID, TODO_ID, _spend_run_count, _write_fixture from tests.control_plane.codex_checkpoint_fixture import ( local_inference_provider, delivered_context as _delivered, last_context_page as _last_page, + reference_metadata_oracle, ) from tests.control_plane.test_heartbeat_envelope_transport import private_handoff # noqa: F401 -- shared private socket fixture from tests.control_plane.test_codex_context_boundary import _app_server, _configure_provider, _input_text, _native_fixture @@ -207,28 +209,27 @@ def pages(name, path): offsets.append(start) count = 0 count += size - if host == "benchmark" and name in {"cold", "recovery"}: - # Execute the shipped original recovery route, including its pinned - # executable; no fixture substitute for installed command routing. - recovery = wires["cold" if name == "cold" else "warm"]["recovery"] - routed = shlex.split(recovery["command"]) - assert routed[:3] == [env["LOOPX_CLI"], "--cli-bin", env["LOOPX_CLI"]] + # Execute the model-visible route, preserving its executable, reference + # and output budget. Private metadata only verifies the expected bytes. + routed = shlex.split(read_commands[name] if name != "recovery" + else wires["warm"]["recovery"]["command"]) + if "--reference-file" in routed: + _, reference = read_context_read_reference(routed[routed.index("--reference-file") + 1]) + assert reference["path"] == str(path) and reference["sha256"] == digest + else: assert routed[routed.index("--capture-file") + 1] == str(path) - routed[routed.index("--page-bytes") + 1] = "5000" - commands = [] - for offset in offsets: - routed[routed.index("--offset") + 1] = str(offset) - commands.append(shlex.join(routed)) - return commands - commands = [shlex.join(cli + ["--format", "json", "quota", "context-read", - "--capture-file", str(path), "--capture-sha256", digest, - "--offset", str(offset), "--page-bytes", str(page_size)]) for offset in offsets] + assert routed[routed.index("--capture-sha256") + 1] == digest + routed[routed.index("--page-bytes") + 1] = str(page_size) + commands = [] + for offset in offsets: + routed[routed.index("--offset") + 1] = str(offset) + commands.append(shlex.join(routed)) if adaptive and name == "cold": # An intact initial short read remains in current native input when # subsequent reads choose different boundaries. Production must # accept exact complete coverage without clearing that history. initial = shlex.split(commands[0]) - initial[-1] = "256" + initial[initial.index("--page-bytes") + 1] = "256" commands.insert(0, shlex.join(initial)) return commands @@ -305,8 +306,20 @@ 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_descriptor = reference_metadata_oracle(descriptor) + wire_file = Path(wire_descriptor["path"]) + wire_bytes = wire_file.read_bytes() + assert hashlib.sha256(wire_bytes).hexdigest() == wire_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) == wire_descriptor["recovery"]["path"] + assert wire_descriptor["sha256"] == wire_descriptor["recovery"]["sha256"] + wires[name] = {"schema_version": "context_delivery_v1", "mode": "full", + "context": json.loads(wire_bytes), "target": wire_descriptor["target"], + "recovery": wire_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: @@ -322,7 +335,7 @@ def response(request, number, message): stages["pages"] = [descriptor["command"]] if bounded_pages else pages(name, read_paths[name]) else: stages["pages"] = ([descriptor["next_command"]] if not descriptor["eof"] else []) if bounded_pages else pages(name, read_paths[name]) - page_digests[name] = descriptor["sha256"] + page_digests[name] = wire_descriptor["sha256"] stages["phase"] = name + "_read" if not stages["pages"]: return response(request, number, message) diff --git a/tests/control_plane/test_codex_context_resume.py b/tests/control_plane/test_codex_context_resume.py index 66b5359301..4b405f71ee 100644 --- a/tests/control_plane/test_codex_context_resume.py +++ b/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) diff --git a/tests/managed_effect_runtime_fixture.py b/tests/managed_effect_runtime_fixture.py index ce580c9bc2..f2f6c3338b 100644 --- a/tests/managed_effect_runtime_fixture.py +++ b/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 diff --git a/tests/test_managed_effect_runtime_fixture.py b/tests/test_managed_effect_runtime_fixture.py index 71441cc2fc..066e07456a 100644 --- a/tests/test_managed_effect_runtime_fixture.py +++ b/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]