diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md index 6da703ece8..350eaec1a5 100644 --- a/docs/reference/required-work-context.md +++ b/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. diff --git a/loopx/control_plane/quota/native_context_lifecycle.ts b/loopx/control_plane/quota/native_context_lifecycle.ts index f188a21f34..3d556830cf 100644 --- a/loopx/control_plane/quota/native_context_lifecycle.ts +++ b/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(); diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index 6f7bfa82df..30d86245e9 100644 --- a/loopx/control_plane/quota/turn_envelope.py +++ b/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( diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py index cd9595e833..0382e189b0 100644 --- a/loopx/control_plane/turn_driver/codex_context.py +++ b/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) diff --git a/loopx/control_plane/turn_driver/codex_context_history.py b/loopx/control_plane/turn_driver/codex_context_history.py index f26c75de60..2e782436b5 100644 --- a/loopx/control_plane/turn_driver/codex_context_history.py +++ b/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. diff --git a/loopx/control_plane/turn_driver/native_context_selection.py b/loopx/control_plane/turn_driver/native_context_selection.py index 63cd0cae3a..5f73d8e116 100644 --- a/loopx/control_plane/turn_driver/native_context_selection.py +++ b/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 {})} diff --git a/tests/control_plane/test_codex_context_resume.py b/tests/control_plane/test_codex_context_resume.py index 4b405f71ee..1b65d0a91f 100644 --- a/tests/control_plane/test_codex_context_resume.py +++ b/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])