fix(context): replay frozen selections in native read verification

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:

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

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

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