Defer section digests for incomplete native context reads

docs/reference/required-work-context.md

@@ -785,6 +785,8 @@ complete native input's actually included sections. Digests remain in the journa
 they are not repeated in model-facing index rows. A list, partial page, EOF without
 earlier coverage, failed/damaged output, subagent hook or unrelated capture cannot
 acknowledge a section. Selected reads never establish a full retained-input baseline.
+The native hook still verifies each page's bytes. It skips computing section
+digests when the terminal range is absent from both the batch and the journal.
 Changed content remains unread; compaction or a different Goal/Agent/scope retires
 the old epoch's observations. Stateless hosts make no such observation claim and
 keep the complete current source available. These observations do not establish
@@ -1105,7 +1107,9 @@ Codex App SSH Goal（`--runtime-profile codex_app_ssh_goal`），以及现有 Co
 私有原生 journal 的 `consumed_pointers` 与 `confirmed_read_sections` 只记录经过字节核对、
 连续完整交付的选读视图，或完整原生输入中实际包含的部分；摘要留在 journal，不反复展示给模型。
 目录、部分页、仅 EOF、失败／篡改输出、子 Agent hook、无关捕获都不构成读取确认；选读不建立全量
-模型保留基线。内容变化继续标为未读；压缩、Goal／Agent／scope 变化使旧 epoch 观测失效。
+模型保留基线。
+原生 hook 仍逐页核对字节；当本批和 journal 都没有末尾范围时，跳过正文字段的摘要计算。
+内容变化继续标为未读；压缩、Goal／Agent／scope 变化使旧 epoch 观测失效。
 无状态宿主不声称观测确认，完整当前来源始终可达；这些记录不证明理解、后续保留、执行权限或计费下降。
 
 受支持的 Codex 声明尚未完成原生连续性初始化时（包括尚未提供会话 id），

loopx/control_plane/turn_driver/codex_context.py

@@ -636,6 +636,8 @@ def _verified_context_pages(state: dict[str, Any], pages: list[dict[str, Any]])
         references.append(ready_ref)
     files: dict[str, bytes] = {}
     result = []
+    # One tool response can contain the terminal page anywhere in its batch.
+    defer_section_digests = not any(page.get("eof") is True for page in pages)
     for page in pages:
         # These observations originate only below, after byte/source checks.
         page = {key: value for key, value in page.items() if key not in ("read_sections", "source_sha256")}
@@ -646,7 +648,7 @@ def _verified_context_pages(state: dict[str, Any], pages: list[dict[str, Any]])
             from .native_context_selection import verified_context_selection
 
             try:
-                observed = verified_context_selection(state, page)
+                observed = verified_context_selection(state, page, defer_section_digests=defer_section_digests)
             except (OSError, ValueError, KeyError, TypeError, RecursionError):
                 observed = None
             if observed is not None:

loopx/control_plane/turn_driver/native_context_selection.py

@@ -13,7 +13,18 @@
     context_selection_view)
 
 
-def verified_context_selection(state: dict[str, Any], page: dict[str, Any]) -> dict[str, Any] | None:
+def _terminal_range_seen(state: dict[str, Any], digest: str, total: int) -> bool:
+    reads = state.get("section_reads")
+    view = reads.get(digest) if isinstance(reads, dict) else None
+    ranges = view.get("ranges") if isinstance(view, dict) else None
+    return isinstance(ranges, list) and any(
+        isinstance(span, list) and len(span) == 2 and span[1] == total for span in ranges
+    )
+
+
+def verified_context_selection(
+    state: dict[str, Any], page: dict[str, Any], *, defer_section_digests: bool = False,
+) -> dict[str, Any] | None:
     from ..quota.turn_envelope import _turn_envelope_result
 
     current = state.get("source_recovery")
@@ -78,17 +89,23 @@ def verified_context_selection(state: dict[str, Any], page: dict[str, Any]) -> d
             or (page.get("eof") is False and (end >= total or page.get("next_offset") != end))
             or type(page.get("eof")) is not bool):
         return None
-    inherited = {}
-    if isinstance(observation, dict):
-        from .native_first_read import observation_current
+    # The reducer only publishes digests after complete coverage. A nonterminal
+    # batch without a previously observed terminal range cannot complete it.
+    # The empty map still marks this as a selected page for range tracking.
+    sections = {}
+    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
 
-        # The hook already matched its event session/workspace to this journal.
-        # Native hook processes need not inherit the shell's CODEX_THREAD_ID.
-        if observation_current(observation, physical, source_digest, hook_session=state.get("session_id")):
-            inherited = observation["fields"]
-    # 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]
-    sections = _turn_envelope_result("context_read_sections", context=source, select=literal_pointers,
-        consumed_pointers=inherited)["sections"] if literal_pointers else {}
+                # The hook already matched its event session/workspace to this journal.
+                # Native hook processes need not inherit the shell's CODEX_THREAD_ID.
+                if observation_current(observation, physical, source_digest, hook_session=state.get("session_id")):
+                    inherited = observation["fields"]
+            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}

tests/control_plane/test_codex_context_lifecycle.py

@@ -281,6 +281,60 @@ def deliver(root, event, pages):
             "tool_use_id": f"page-{page.get('offset')}", "tool_response": json.dumps(page)}) == {}
 
 
+@pytest.mark.parametrize("order", ["sequential", "terminal_first", "one_batch"])
+def test_selected_read_can_defer_digests_without_losing_coverage(tmp_path, monkeypatch, order):
+    from argparse import Namespace
+    from loopx.cli_commands.quota_capture import read_context_capture
+    from loopx.control_plane.quota import turn_envelope
+
+    event = setup(tmp_path, monkeypatch)
+    body = "Preserve this acceptance condition: 核对原始结果。\n" * 45
+    _, wire = project(tmp_path, payload={"ok": True, "requirements": body})
+    recovery = wire["recovery"]
+    pages = []
+    offset, selection_digest = 0, None
+    while offset is not None:
+        page = read_context_capture(Namespace(format="json", capture_file=recovery["path"],
+            capture_sha256=recovery["sha256"], offset=offset, page_bytes=1024, select=["/requirements"],
+            selection_sha256=selection_digest))
+        assert page["ok"], page
+        pages.append(page)
+        offset = page["next_offset"]
+        selection_digest = page["selection_sha256"]
+    assert len(pages) == 3 and pages[-1]["eof"]
+
+    calls = []
+    run = turn_envelope._turn_envelope_result
+
+    def track(operation, **kwargs):
+        if operation == "context_read_sections":
+            calls.append(operation)
+        return run(operation, **kwargs)
+
+    monkeypatch.setattr(turn_envelope, "_turn_envelope_result", track)
+    if order == "sequential":
+        deliver(tmp_path, event, pages[:-1])
+        assert calls == []
+        assert not context._state(tmp_path).get("consumed_pointers")
+        deliver(tmp_path, event, pages[-1:])
+        assert len(calls) == 1
+    elif order == "terminal_first":
+        deliver(tmp_path, event, [pages[-1], pages[0]])
+        damaged = {**pages[1], "content": "!" + pages[1]["content"][1:]}
+        deliver(tmp_path, event, [damaged])
+        assert not context._state(tmp_path).get("consumed_pointers")
+        deliver(tmp_path, event, pages[1:2])
+    else:
+        batch = {"content": [{"type": "text", "text": json.dumps(page)} for page in reversed(pages)]}
+        assert context.codex_context_hook(str(tmp_path), {**event, "hook_event_name": "PostToolUse",
+            "tool_use_id": "selected-pages", "tool_response": batch}) == {}
+    state = context._state(tmp_path)
+    expected = hashlib.sha256(json.dumps(body, ensure_ascii=False).encode("utf-8")).hexdigest()
+    assert state["consumed_pointers"] == {"/requirements": expected}
+    assert not state.get("section_reads")
+    assert "candidate" not in state and "retained" not in state
+
+
 def test_complete_reader_delivery_allows_delta_without_diagnostics_or_ack(tmp_path, monkeypatch):
     event = setup(tmp_path, monkeypatch)
     payload = {"ok": True, "plan_summary": {"requirements": "Preserve every obligation. " * 1000}}