Retain verified context selections across native resume

docs/reference/required-work-context.md

@@ -964,6 +964,12 @@ back to full. One decoded record remains bounded; this is not a history-size
 limit. Legacy whole-prefix checkpoints require a fresh complete read. It retains the
 normal-read engineering assumption rather than certifying future model input;
 it neither scans model requests nor asks the model for an ACK.
+For a complete one-page selection from the current capture, the adapter can
+verify that exact selected output across an ordinary resume. The source and
+read scope must still match. Only the replayed literal sections remain
+eligible for reuse in the next progressive view. Partial or inherited
+selections still need their ordinary recovery, and a selected read cannot
+establish a whole-input baseline.
 Opted-in benchmark native children create owner-only files automatically; the
 parent process and off-mode file creation policy stay unchanged. Group-writable
 existing replay sources cannot establish cross-process continuity.

loopx/control_plane/quota/native_context_lifecycle.ts

@@ -55,6 +55,7 @@ export function updateNativeContextLifecycle(request: JsonObject): JsonObject {
     delete state.consumed_pointers;
     delete state.confirmed_read_sections;
     delete state.section_reads;
+    delete state.selected_resume;
   };
   if (name === "SessionStart") {
     // The provider may qualify the native-selected replay source against the
@@ -73,9 +74,21 @@ export function updateNativeContextLifecycle(request: JsonObject): JsonObject {
       delete state.candidate;
       delete state.pending;
       delete state.active_call;
-      delete state.consumed_pointers;
-      delete state.confirmed_read_sections;
       delete state.section_reads;
+      // Selected outputs can establish their own literal sections on a
+      // qualified replay. They never establish a whole-input candidate.
+      const previous = jsonObject(state.consumed_pointers);
+      const replayed = jsonObject(request.resume_sections);
+      const retained = Object.fromEntries(Object.entries(replayed ?? {}).filter(([address, digest]) =>
+        typeof digest === "string" && previous?.[address] === digest));
+      if (Object.keys(retained).length > 0) {
+        state.consumed_pointers = retained;
+        state.confirmed_read_sections = Object.keys(retained);
+      } else {
+        delete state.consumed_pointers;
+        delete state.confirmed_read_sections;
+        delete state.selected_resume;
+      }
       return state;
     }
     invalidate();

loopx/control_plane/quota/turn_envelope.py

@@ -430,11 +430,12 @@ def prepare_progressive_context(context: Mapping[str, Any], *, source_reference:
 def update_native_context_lifecycle(state: Mapping[str, Any], event: Mapping[str, Any], *,
                                     new_epoch: str, pages: list[dict[str, Any]],
                                     pages_verified: bool = False,
-                                    resume_verified: bool = False) -> dict[str, Any]:
+                                    resume_verified: bool = False,
+                                    resume_sections: Mapping[str, str] | None = None) -> dict[str, Any]:
     """Native IO adapter; the shared typed owner reduces continuity assumptions."""
     return _turn_envelope_result("native_context_lifecycle", state=dict(state),
         event=dict(event), new_epoch=new_epoch, pages=pages, pages_verified=pages_verified,
-        resume_verified=resume_verified)
+        resume_verified=resume_verified, **({"resume_sections": dict(resume_sections)} if resume_sections else {}))
 
 
 def restore_context_delivery(

loopx/control_plane/turn_driver/codex_context.py

@@ -372,7 +372,10 @@ def codex_context_hook(directory: str | None, event: dict[str, Any]) -> dict[str
     """
     from ..quota.turn_envelope import update_native_context_lifecycle
     from ..effect_runtime import EffectRuntimeRemoteError, EffectRuntimeStartupError
-    from .codex_context_history import checkpoint_native_read, native_resume_verified, observe_native_read
+    from .codex_context_history import (
+        checkpoint_native_read, checkpoint_native_selection, native_resume_sections, native_resume_verified,
+        observe_native_read,
+    )
 
     if directory is None:
         return {}
@@ -382,6 +385,7 @@ def codex_context_hook(directory: str | None, event: dict[str, Any]) -> dict[str
     if not native_context_event_eligible(event):
         return {}  # no parent recovery reference may escape to a subagent
     pages = _context_pages(event.get("tool_response")) if name == "PostToolUse" else []
+    raw_pages = pages
     if name == "PostToolUse" and not pages:
         return {}  # ordinary tool use neither invalidates nor rewrites the journal
     recovery = None
@@ -404,12 +408,15 @@ def codex_context_hook(directory: str | None, event: dict[str, Any]) -> dict[str
                     pages = _verified_context_pages(state, pages, selected_only=True)
                     if not pages or not isinstance(event.get("cwd"), str) or not event["cwd"]:
                         return {}
+                resume_verified = name == "SessionStart" and native_resume_verified(state, event)
                 updated = update_native_context_lifecycle(read_state, event,
                     new_epoch=uuid.uuid4().hex, pages=pages, pages_verified=True,
-                    resume_verified=name == "SessionStart" and native_resume_verified(state, event))
+                    resume_verified=resume_verified,
+                    resume_sections=native_resume_sections(state, event) if name == "SessionStart" and not resume_verified else None)
                 if name == "PostToolUse" and same:
                     checkpoint_native_read(read_state, updated, event,
                         [page for page in pages if "read_sections" not in page])
+                    checkpoint_native_selection(state, updated, event, raw_pages, pages)
                     previous = updated.get("host_execution_preparation")
                     ready = previous.get("ready_capture") if isinstance(previous, dict) else None
                     ready_digest = ready.get("sha256") if isinstance(ready, dict) else None
@@ -800,7 +807,7 @@ def prepare_codex_context(payload: dict[str, Any], *, directory: str | None, cap
                 read_scope = {**scope, "surface": surface,
                     "todo_id": selected.get("todo_id") if isinstance(selected, dict) else None}
                 if state.get("read_scope") != read_scope:
-                    for key in ("consumed_pointers", "confirmed_read_sections", "section_reads"):
+                    for key in ("consumed_pointers", "confirmed_read_sections", "section_reads", "selected_resume"):
                         state.pop(key, None)
                 state["read_scope"] = read_scope
                 request = _native_context_request(state, scope, capture_id)

loopx/control_plane/turn_driver/codex_context_history.py

@@ -32,6 +32,7 @@
 _IO_CHUNK_BYTES = 64 * 1024
 # Bound one decoded protocol record, not the history or the replay interval.
 _MAX_RECORD_BYTES = 16 * 1024 * 1024
+_MAX_SELECTED_REPLAY_READS = 32
 _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.DOTALL)
 
@@ -152,7 +153,7 @@ def _read_pages(output: object) -> list[dict[str, Any]]:
 
 
 def _matches(anchor: dict[str, Any], snapshot: dict[str, Any], stream: BinaryIO,
-             event: dict[str, Any]) -> bool:
+             event: dict[str, Any], *, selected: bool = False) -> bool:
     from .decision_capture import capture_page_range, read_context_capture_file
 
     if any(anchor.get(k) != snapshot.get(k) for k in
@@ -203,7 +204,7 @@ def _matches(anchor: dict[str, Any], snapshot: dict[str, Any], stream: BinaryIO,
             for group in groups:
                 seen_pages[_digest(group)] += 1
                 for page in group:
-                    if page.get("sha256") != reference["sha256"]:
+                    if selected or page.get("sha256") != reference["sha256"]:
                         continue
                     span = capture_page_range(page, raw=body, digest=reference["sha256"])
                     if span is None:
@@ -216,7 +217,7 @@ def _matches(anchor: dict[str, Any], snapshot: dict[str, Any], stream: BinaryIO,
         if start > cursor:
             return False
         cursor = max(cursor, stop)
-    return seen_pages == expected_pages and cursor == len(body)
+    return seen_pages == expected_pages and (selected or cursor == len(body))
 
 
 def _append_only(record: dict[str, Any], event: dict[str, Any]) -> bool:
@@ -288,6 +289,68 @@ def native_resume_verified(state: dict[str, Any], event: dict[str, Any]) -> bool
         return False
 
 
+def native_resume_sections(state: dict[str, Any], event: dict[str, Any]) -> dict[str, str]:
+    """Verify replay of exact selected outputs without claiming a whole capture read."""
+    anchor = state.get("selected_resume")
+    if (event.get("source") != "resume" or not isinstance(anchor, dict)
+            or state.get("session_id") != event.get("session_id") or state.get("cwd") != event.get("cwd")
+            or anchor.get("scope") != state.get("read_scope") or anchor.get("epoch") != state.get("context_epoch")
+            or anchor.get("read_source") != state.get("source_recovery")
+            or not isinstance(anchor.get("sections"), dict)):
+        return {}
+    try:
+        with _snapshot(event) as (facts, stream):
+            return anchor["sections"] if _matches(anchor, facts, stream, event, selected=True) else {}
+    except (OSError, ValueError, TypeError, KeyError, AttributeError, RecursionError):
+        return {}
+
+
+def checkpoint_native_selection(state: dict[str, Any], updated: dict[str, Any],
+                                event: dict[str, Any], raw_pages: list[dict[str, Any]],
+                                verified_pages: list[dict[str, Any]]) -> None:
+    """Record bounded, self-contained selections after literal byte verification."""
+    if len(raw_pages) != 1 or len(verified_pages) != 1:
+        return
+    raw, page = raw_pages[0], verified_pages[0]
+    sections, source = page.get("read_sections"), state.get("source_recovery")
+    scope, epoch, call = state.get("read_scope"), state.get("context_epoch"), event.get("tool_use_id")
+    consumed = updated.get("consumed_pointers", {})
+    if (not (raw.get("schema_version") == "context_capture_selection_page_v1"
+            or "selected_pointers" in raw and "selection_sha256" in raw)
+            or page.get("ok") is not True or page.get("offset") != 0 or page.get("eof") is not True
+            or page.get("read_sections_inherited") or not isinstance(sections, dict) or not sections
+            or not isinstance(source, dict) or page.get("source_sha256") != source.get("sha256")
+            or not isinstance(scope, dict) or not epoch or scope.get("context_epoch") != epoch
+            or not isinstance(call, str) or not call or not isinstance(consumed, dict)
+            or any(consumed.get(pointer) != digest for pointer, digest in sections.items())):
+        return
+    receipt = {"call_id": call, "pages_sha256": _digest(raw_pages)}
+    previous = state.get("selected_resume")
+    if (isinstance(previous, dict) and previous.get("scope") == scope
+            and previous.get("epoch") == epoch and previous.get("read_source") == source
+            and all(previous.get(k) == event.get(k) for k in ("model", "permission_mode"))
+            and previous.get("path") == event.get("transcript_path")
+            and isinstance(previous.get("reads"), list) and isinstance(previous.get("sections"), dict)
+            and all(isinstance(read, dict) for read in previous["reads"])
+            and len(previous["reads"]) < _MAX_SELECTED_REPLAY_READS):
+        if any(read.get("call_id") == call for read in previous["reads"]):
+            return
+        updated["selected_resume"] = {**previous, "reads": [*previous["reads"], receipt],
+            "sections": {**previous["sections"], **sections}}
+        return
+    try:
+        with _snapshot(event) as (facts, stream):
+            end = facts["observed_bytes"]
+            start = _reader_start(stream, end, raw_pages)
+            if not 0 < start < end:
+                return
+            anchor = {**facts, "reader_start": start, "reader_sha256": _region_digest(stream, start, end),
+                "read_source": source, "reads": [receipt], "sections": sections, "scope": scope, "epoch": epoch}
+        updated["selected_resume"] = anchor
+    except (OSError, ValueError, TypeError, KeyError, AttributeError, RecursionError):
+        return
+
+
 def checkpoint_native_read(state: dict[str, Any], updated: dict[str, Any],
                            event: dict[str, Any], pages: list[dict[str, Any]]) -> None:
     """Attach content-free native source facts only after the typed complete read.

loopx/control_plane/turn_driver/native_context_selection.py

@@ -96,11 +96,11 @@ def verified_context_selection(
     # batch without a previously observed terminal range cannot complete it.
     # The empty map still marks this as a selected page for range tracking.
     sections = {}
+    inherited = {}
     if not defer_section_digests or _terminal_range_seen(state, view_digest, total):
         # A display edit is not an observation of its literal source text.
         literal_pointers = [pointer for pointer in pointers if pointer not in presentation_pointers]
         if literal_pointers:
-            inherited = {}
             if isinstance(observation, dict):
                 from .native_first_read import observation_current
 
@@ -111,4 +111,5 @@ def verified_context_selection(
             sections = _turn_envelope_result("context_read_sections", context=source, select=literal_pointers,
                 consumed_pointers=inherited)["sections"]
     return {**page, "schema_version": "context_capture_page_v1", "sha256": view_digest,
-        "selected_pointers": pointers, "source_sha256": current["sha256"], "read_sections": sections}
+        "selected_pointers": pointers, "source_sha256": current["sha256"], "read_sections": sections,
+        **({"read_sections_inherited": True} if inherited else {})}

tests/control_plane/test_codex_context_resume.py

@@ -138,6 +138,68 @@ def test_completed_native_resume_keeps_delta_without_old_capture_prefetch(tmp_pa
     assert_file_delta(wire, before["recovery"], current)
 
 
+@pytest.mark.parametrize("replay", ["present", "pinned", "lost_output", "partial", "compacted", "changed_source"])
+def test_native_resume_retains_only_replayed_complete_selections(tmp_path, monkeypatch, replay):
+    """A selected read can survive a qualified replay without proving full input."""
+    import hashlib
+
+    event = setup(tmp_path, monkeypatch)
+    history = tmp_path / "rollout.jsonl"
+    append(history, "session_meta", {"id": "session", "cwd": str(tmp_path),
+        "cli_version": "0.162.1", "history_mode": "paginated"})
+    history.chmod(0o600)
+    event.update(transcript_path=str(history), model="fixture", permission_mode="bypassPermissions")
+    append(history, "turn_context", {"cwd": str(tmp_path), "model": "fixture"})
+    address = "/requirements"
+    payload = {"ok": True, "goal_id": "fixture", "requirements": "Preserve the current stop. " * 320,
+        "unread": "This is a separate source field. " * 400}
+    _, wire = project(tmp_path, payload=payload)
+    recovery = wire["recovery"]
+    repo = Path(context.__file__).resolve().parents[3]
+    cli = [sys.executable, "-m", "loopx.entrypoint", "--format", "json", "quota"]
+    environment = {**os.environ, "PYTHONPATH": str(repo), "CODEX_THREAD_ID": "session"}
+
+    def command(args, event=None):
+        result = subprocess.run([*cli, *args], input=json.dumps(event) if event else None,
+            capture_output=True, text=True, cwd=tmp_path, env=environment, timeout=20, check=False)
+        assert result.returncode == 0, result.stdout + result.stderr
+        return json.loads(result.stdout)
+
+    if replay == "pinned":
+        from loopx.control_plane.turn_driver.decision_capture import write_context_read_reference
+
+        ref = write_context_read_reference(Path(recovery["path"]), recovery["sha256"])
+        source = ["--reference-file", str(ref)]
+    else:
+        source = ["--capture-file", recovery["path"], "--capture-sha256", recovery["sha256"]]
+    page = command(["context-read", *source, "--select", address, "--page-bytes", "1000" if replay == "partial" else "64000",
+        "--max-page-output-bytes", "64000"])
+    assert page["ok"] and page["eof"] == (replay != "partial")
+    hook = ["context-hook", "--context-delivery", "on", "--codex-app", "--codex-context-dir", str(tmp_path)]
+    # A code-mode parent may append its actual output after the nested hook.
+    append(history, "response_item", {"type": "custom_tool_call", "call_id": "parent"})
+    command(hook, {**event, "hook_event_name": "PostToolUse", "tool_use_id": "reader",
+        "tool_response": json.dumps(page)})
+    before = context._state(tmp_path)
+    assert (address in before.get("consumed_pointers", {})) == (replay != "partial")
+    assert "candidate" not in before
+    if replay != "lost_output":
+        append(history, "response_item", {"type": "custom_tool_call_output", "call_id": "parent",
+            "output": [{"type": "input_text", "text": json.dumps(page)}]})
+    if replay == "compacted":
+        append(history, "compacted", {"message": "Earlier model context was replaced."})
+    if replay == "changed_source":
+        capture = Path(recovery["path"])
+        changed = capture.read_bytes() + b" "
+        capture.write_bytes(changed)
+        assert hashlib.sha256(changed).hexdigest() != recovery["sha256"]
+    command(hook, {**event, "source": "resume"})
+    after = context._state(tmp_path)
+    assert "candidate" not in after
+    expected = {address: before["consumed_pointers"][address]} if replay in {"present", "pinned"} else {}
+    assert after.get("consumed_pointers", {}) == expected
+
+
 @pytest.mark.parametrize("single_page", [False, True])
 @pytest.mark.parametrize("lose_first", [False, True])
 @pytest.mark.parametrize("code_mode", [False, True])