From 89076f83ce0e5105ae32426f13f8d10810e89527 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 02:54:16 +0800 Subject: [PATCH] fix(context): replay frozen selections in native read verification Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/cli_commands/quota_capture.py | 29 ++--- .../turn_driver/decision_capture.py | 29 +++++ .../turn_driver/native_context_selection.py | 22 ++-- .../test_context_read_reference.py | 106 +++++++++++++++++- 4 files changed, 151 insertions(+), 35 deletions(-) diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 3adc706e88..6b7a407374 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -19,8 +19,7 @@ from ..control_plane.quota.error_codes import QuotaCommandValidationError, QuotaDecisionTransportError from ..control_plane.turn_driver.decision_capture import ( context_index_fields as _context_index_fields, - context_selection_transport as _context_selection_transport, - context_selection_bytes as _context_selection_bytes, + context_selection_view as _context_selection_view, validate_handoff_socket, CONTEXT_CAPTURE_READ_GUIDANCE, CONTEXT_CAPTURE_PAGE_GUIDANCE, context_capture_read_command, CONTEXT_CAPTURE_NAVIGATION_GUIDANCE, read_context_capture_file as _read_capture, @@ -224,15 +223,10 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An raise ValueError("selected context-read requires JSON output") if offset > 0 and selection_digest is None: raise ValueError("selected continuation requires selection-sha256; follow next_command") - from ..control_plane.quota.turn_envelope import _turn_envelope_result - - depth_options = {"list_depth": list_depth} if list_depth is not None else {} - selection_options = {"initial_read": True} if initial_read else {"select": selectors} - planning_context = payload if initial_read or list_fields else _context_selection_transport(payload) pinned_explicit = pinned_selection is not None and not initial_read - order_options = {"preserve_selection_order": True} if pinned_explicit else {} - plan = _turn_envelope_result("context_select", context=planning_context, list_fields=list_fields, - recover_selection_errors=not initial_read, **selection_options, **depth_options, **order_options) + plan, selected_raw, expected = _context_selection_view(payload, select=selectors, + initial_read=initial_read, list_fields=list_fields, list_depth=list_depth, + expected_sha256=selection_digest, preserve_selection_order=pinned_explicit) if rejection := plan.get("rejected_selection"): # TS alone resolves the nearest container. Reuse the original # escaped tokens, without parsing diagnostic prose or traversing @@ -248,18 +242,9 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An "reason": rejection["message"], "recovery_command": command, "recovery": "Use recovery_command to --list the nearest verified container in the same capture, " "then retry the corrected batch. No values were delivered and no guard, admission or spend was performed."} - raw = _context_selection_bytes(payload, plan["selections"], list_fields=list_fields, list_depth=list_depth) - expected = hashlib.sha256(raw).hexdigest() - if pinned_explicit and selection_digest != expected: - # Older explicit metadata could retain caller order while its - # digest covered canonical output. Try that original owner - # once, over the same verified source and exact address set. - plan = _turn_envelope_result("context_select", context=planning_context, - list_fields=list_fields, **selection_options, **depth_options) - raw = _context_selection_bytes(payload, plan["selections"], list_fields=list_fields, list_depth=list_depth) - expected = hashlib.sha256(raw).hexdigest() - if selection_digest is not None and selection_digest != expected: - raise ValueError("selected view digest changed; restore the original selection and reader") + if selected_raw is None or expected is None: + raise ValueError("selected view is unavailable") + raw = selected_raw selection_digest = expected selectors = [item["pointer"] for item in plan["selections"]] if reference_file is not None and pinned_selection is None: diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py index 42ab5779d1..91f3ea8e21 100644 --- a/loopx/control_plane/turn_driver/decision_capture.py +++ b/loopx/control_plane/turn_driver/decision_capture.py @@ -500,3 +500,32 @@ def context_selection_bytes(payload: dict[str, Any], selections: list[dict[str, view_options = {"list_depth": list_depth} if list_depth is not None and list_depth != 1 else {} return json.dumps({"schema_version": "context_capture_index_v1" if list_fields else "context_capture_selection_v1", "selections": views, **view_options}, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n" + + +def context_selection_view(payload: dict[str, Any], *, select: list[str] | None, + initial_read: bool, list_fields: bool, list_depth: int | None, + expected_sha256: str | None, preserve_selection_order: bool = False, + ) -> tuple[dict[str, Any], bytes | None, str | None]: + """Replay the TS selector and verify the original digest-pinned view.""" + from ..quota.turn_envelope import _turn_envelope_result + + context = payload if initial_read or list_fields else context_selection_transport(payload) + selection = {"initial_read": True} if initial_read else {"select": select} + depth = {"list_depth": list_depth} if list_depth is not None else {} + order = {"preserve_selection_order": True} if preserve_selection_order else {} + plan = _turn_envelope_result("context_select", context=context, list_fields=list_fields, + recover_selection_errors=not initial_read, **selection, **depth, **order) + if plan.get("rejected_selection"): + return plan, None, None + raw = context_selection_bytes(payload, plan["selections"], list_fields=list_fields, list_depth=list_depth) + digest = hashlib.sha256(raw).hexdigest() + if preserve_selection_order and expected_sha256 != digest: + # Historical explicit references may pin canonical output despite + # retaining caller order. Only the same source and addresses can replay. + plan = _turn_envelope_result("context_select", context=context, + list_fields=list_fields, **selection, **depth) + raw = context_selection_bytes(payload, plan["selections"], list_fields=list_fields, list_depth=list_depth) + digest = hashlib.sha256(raw).hexdigest() + if expected_sha256 is not None and expected_sha256 != digest: + raise ValueError("selected view digest changed; restore the original selection and reader") + return plan, raw, digest diff --git a/loopx/control_plane/turn_driver/native_context_selection.py b/loopx/control_plane/turn_driver/native_context_selection.py index 1ae4ed84e0..4291766770 100644 --- a/loopx/control_plane/turn_driver/native_context_selection.py +++ b/loopx/control_plane/turn_driver/native_context_selection.py @@ -6,12 +6,11 @@ """ from __future__ import annotations -import hashlib from pathlib import Path from typing import Any from .decision_capture import (read_context_capture_file, read_context_read_reference, - context_selection_transport, context_selection_bytes) + context_selection_view) def verified_context_selection(state: dict[str, Any], page: dict[str, Any]) -> dict[str, Any] | None: @@ -45,7 +44,9 @@ def verified_context_selection(state: dict[str, Any], page: dict[str, Any]) -> d physical = physical.parent.resolve(strict=True) / physical.name trusted = next((entry for entry in references if entry.get("sha256") == source_digest and Path(entry["path"]).parent.resolve(strict=True) / Path(entry["path"]).name == physical), None) - if trusted is None or not isinstance(pointers, list) or not pointers or not all(isinstance(p, str) for p in pointers): + initial_read = selected is not None and selected.get("initial_read") is True + if trusted is None or (initial_read and pointers is not None) or (not initial_read and ( + not isinstance(pointers, list) or not pointers or not all(isinstance(p, str) for p in pointers))): return None offset, total = page.get("offset"), page.get("total_bytes") content = page.get("content") @@ -55,12 +56,17 @@ def verified_context_selection(state: dict[str, Any], page: dict[str, Any]) -> d if not actual or offset + len(actual) > total: return None _, raw, source = read_context_capture_file(str(physical), source_digest, private=True) + actual_digest = source_digest if selected is not None or "selected_pointers" in page: - selection = _turn_envelope_result("context_select", context=context_selection_transport(source), - select=pointers, list_fields=False) - raw = context_selection_bytes(source, selection["selections"], list_fields=False, - list_depth=selected.get("list_depth") if selected else None) - if hashlib.sha256(raw).hexdigest() != view_digest or len(raw) != total or raw[offset:offset + len(actual)] != actual: + selection, selected_raw, actual_digest = context_selection_view(source, select=pointers, + initial_read=initial_read, list_fields=False, + list_depth=selected.get("list_depth") if selected else None, + expected_sha256=view_digest, preserve_selection_order=selected is not None and not initial_read) + if selected_raw is None: + return None + raw = selected_raw + pointers = [item["pointer"] for item in selection["selections"]] + if actual_digest != view_digest or len(raw) != total or raw[offset:offset + len(actual)] != actual: return None end = offset + len(actual) if ((page.get("eof") is True and (end != total or page.get("next_offset") is not None)) diff --git a/tests/control_plane/test_context_read_reference.py b/tests/control_plane/test_context_read_reference.py index 648dd1249f..5c5e5f43cc 100644 --- a/tests/control_plane/test_context_read_reference.py +++ b/tests/control_plane/test_context_read_reference.py @@ -7,7 +7,7 @@ import pytest from loopx.cli_commands.quota_capture import model_context_reference, native_context_capture_view, read_context_capture -from loopx.control_plane.turn_driver.codex_context import _context_pages +from loopx.control_plane.turn_driver.codex_context import _context_pages, _verified_context_pages from loopx.control_plane.turn_driver.decision_capture import ( capture_page_range, read_context_read_reference, write_context_read_reference, ) @@ -32,6 +32,19 @@ def fixture(tmp_path): return payload, view, model_context_reference(view) +def assert_section_read_only(page, machine): + raw = Path(machine["path"]).read_bytes() + reference = {"path": machine["path"], "sha256": machine["sha256"]} + pages = _context_pages(page) + assert pages == [page] + assert capture_page_range(page, raw=raw, digest=machine["sha256"]) is None + observed = _verified_context_pages({"source_recovery": reference, "recovery": reference}, pages) + assert len(observed) == 1, "a valid selected page must reach the section-read owner" + assert observed[0]["source_sha256"] == machine["sha256"] + assert observed[0]["read_sections"] + assert capture_page_range(observed[0], raw=raw, digest=machine["sha256"]) is None + + @pytest.mark.parametrize("mode", ["full", "delta"]) @pytest.mark.parametrize("reference", [False, True], ids=["explicit", "navigation"]) def test_policy_oracle_reads_complete_current_recovery(tmp_path, mode, reference): @@ -97,7 +110,7 @@ def test_selected_continuations_hide_both_digests_and_preserve_exact_values(tmp_ while True: assert page["ok"] and page["schema_version"] == "context_capture_selection_page_v1" assert not any(key in page for key in ("sha256", "selection_sha256", "selected_pointers")) - assert _context_pages(page) == [] # selected EOF is never a full-input receipt + assert_section_read_only(page, machine) parts.append(page["content"]) if page["eof"]: break @@ -170,7 +183,7 @@ def changed_navigation(operation, **kwargs): parts = [first["content"]] while True: assert following["ok"] and following["schema_version"] == "context_capture_selection_page_v1" - assert _context_pages(following) == [] + assert_section_read_only(following, machine) assert following["offset"] == len("".join(parts).encode()) parts.append(following["content"]) if following["eof"]: @@ -209,12 +222,13 @@ def observe(operation, **kwargs): monkeypatch.setattr(turn_envelope, "_turn_envelope_result", observe) command = f"loopx --format json quota context-read --reference-file {shlex.quote(str(reference))} --page-bytes 3000 --max-page-output-bytes 3500" - content, pages = b"", 0 + content, pages, observed_pages = b"", 0, [] while True: page = read(command) assert page["ok"] and page["offset"] == len(content), page - assert _context_pages(page) == [] + assert _context_pages(page) == [page] assert capture_page_range(page, raw=path.read_bytes(), digest=digest) is None + observed_pages.append(page) content += page["content"].encode() pages += 1 if page["eof"]: @@ -225,6 +239,8 @@ def observe(operation, **kwargs): assert requests[0]["preserve_selection_order"] is True if layout == "legacy_canonical": assert all(not request.get("preserve_selection_order", False) for request in requests[1::2]) + for page in observed_pages: + assert_section_read_only(page, machine) assert {f: f.read_bytes() for f in before} == before @@ -265,6 +281,86 @@ def changed_initial(operation, **kwargs): assert {f: f.read_bytes() for f in before} == before +@pytest.mark.parametrize("kind", ["initial", "historical_initial", "pinned"]) +def test_native_hook_records_frozen_selection_without_full_input_retention(tmp_path, monkeypatch, kind): + from loopx.control_plane.turn_driver import codex_context as context + from loopx.control_plane.turn_driver.decision_capture import context_capture_read_command + from test_codex_context_lifecycle import setup, project, deliver + from test_progressive_native_context import payload, outcome + + event = setup(tmp_path, monkeypatch) + original = payload() + original["interaction_contract"] = {"schema_version": "loopx_interaction_contract_v0", + "agent_channel": {"work_context": {"sources": []}}} + original["selected_todo"] = {"todo_id": "todo_work", "claimed_by": "agent", + "text": "Read the complete original requirement. " * 400} + outcome(original)["requirements"][0]["description"] += " Keep the original Stop." * 400 + project(tmp_path, payload=original) + source = context._state(tmp_path)["source_recovery"] + path, digest = Path(source["path"]), source["sha256"] + reference = write_context_read_reference(path, digest) + command = context_capture_read_command(path, digest, reference_file=reference) + if kind == "pinned": + pointers = ["/stop", "/planning_horizon"] + expected = json.dumps({"schema_version": "context_capture_selection_v1", "selections": [ + {"pointer": pointer, "value": original[pointer[1:]]} for pointer in pointers]}, + ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() + b"\n" + reference = write_context_read_reference(path, digest, selection={"select": pointers, + "initial_read": False, "list_fields": False, "list_depth": None, + "sha256": hashlib.sha256(expected).hexdigest()}) + page = read(command, reference_file=str(reference), page_bytes=3000, max_page_output_bytes=3500) + else: + page = read(command, initial_read=True, page_bytes=3000, max_page_output_bytes=3500) + if kind == "historical_initial": + _, descriptor = read_context_read_reference(page["reference_file"]) + reference = write_context_read_reference(path, digest, selection={ + **descriptor["selection"], "select": None, "initial_read": True}) + page = read(command, reference_file=str(reference), page_bytes=3000, max_page_output_bytes=3500) + pages = [] + while True: + assert page["ok"], page + pages.append(page) + if page["eof"]: + break + page = read(page["next_command"]) + assert len(pages) > 2 + before = path.read_bytes() + deliver(tmp_path, event, pages[1:]) + assert not context._state(tmp_path).get("confirmed_read_sections"), "EOF cannot hide unread earlier pages" + deliver(tmp_path, event, pages[:1]) + state = context._state(tmp_path) + assert state["confirmed_read_sections"] + assert state["consumed_pointers"] + assert "candidate" not in state, "section coverage cannot establish full model-input retention" + assert path.read_bytes() == before + _, warm = project(tmp_path, "after-selected-read", payload=original) + current = json.loads(Path(warm["recovery"]["path"]).read_bytes()) + unread = current["context_details"]["unread_context_index"]["pointers"] + for pointer in state["consumed_pointers"]: + assert not any(item == pointer or item.startswith(pointer + "/") for item in unread) + assert "candidate" not in context._state(tmp_path) + + +@pytest.mark.parametrize("damage", ["content", "foreign_capture", "index"]) +def test_native_selection_observer_rejects_unverified_or_discovery_only_reads(tmp_path, damage): + _, machine, model = fixture(tmp_path) + reference = {"path": machine["path"], "sha256": machine["sha256"]} + state = {"source_recovery": reference, "recovery": reference} + page = read(model["command"], select=["/criterion"]) + assert_section_read_only(page, machine) + if damage == "content": + page = {**page, "content": page["content"][:-1] + "!", + "read_sections": {"/criterion": "a" * 64}, "source_sha256": machine["sha256"]} + elif damage == "foreign_capture": + other = tmp_path / "foreign" + other.mkdir(mode=0o700) + _, _, foreign = fixture(other) + page = read(foreign["command"], select=["/criterion"]) + else: + page = read(model["command"], select=[""], list_fields=True) + assert _verified_context_pages(state, _context_pages(page)) == [] + + @pytest.mark.parametrize("selection", ["fields", "index", "initial"]) def test_selected_read_survives_unwritable_reference_metadata(tmp_path, monkeypatch, selection): import errno