fix(context): preserve read scope when recovering stale cursors

docs/reference/required-work-context.md

@@ -233,10 +233,12 @@ selection references pin the resolved addresses and complete selection digest;
 continuations use the existing explicit reader instead of rerunning initial
 navigation. A later navigation change cannot replace the saved selection.
 Historical recipe-only references and explicit-file initial cursors still reject
-a changed selection digest; recover through the same original full-capture
-reference, without another admission or rewriting the saved descriptor. These
-legacy readers still require reconciliation of their unfinished original Turns
-before retirement; this slice retains their decoder and full-capture recovery.
+a changed selection digest. If the original capture passes file-digest
+verification, the recovery command restarts at offset zero. Historical
+initial-read cursors can recover the full original file. An explicit selection
+or field listing keeps its original scope, including a receipt-only diagnostic.
+The recovery never repeats admission or rewrites the saved descriptor. Retain
+the legacy readers while unfinished Turns still depend on them.
 Initial navigation can replace fully
 selected children with their complete captured parent to avoid repeating addresses.
 An unselected child, including optional or unknown material, prevents that merge.
@@ -376,9 +378,11 @@ full/delta 字节比较计入这些 reader 命令元数据；既有进程内引
 却绑定规范排序正文的摘要，仅在摘要不匹配后，对同一已验证来源及相同地址集合重试
 一次原规范 planner；只返回摘要完全匹配的字节。此兼容路径每个旧分页最多增加一次
 只读 planner 调用，不写状态或证明模型输入覆盖。历史 recipe-only 引用和原始首读
-游标仍在 profile 变化时严格拒绝，并保留完整捕获恢复。既有内部 `context_select`
-请求为重放接受严格 boolean 的 `preserve_selection_order`；公开显式选读及错误诊断
-不变，reader 描述符字段结构和 CLI 选项不变。
+游标仍在 profile 变化时严格拒绝。原捕获摘要通过校验后，恢复命令从字节零开始。
+历史首读可恢复完整原文件；显式选读和字段列表保留原范围，包括只读诊断回执。
+恢复不会再次准入或改写已保存的描述符。既有内部 `context_select`
+请求为重放接受严格 boolean 的 `preserve_selection_order`。常规显式选读和其他错误
+诊断不变，reader 描述符字段结构和 CLI 选项不变。
 首读按 TypeScript owner 的优先顺序呈现，不再按指针字母序重排；当前控制条件、必读
 来源和结算义务先于完整的未读材料索引。合并父节点保留首个已选子节点的位置，
 显式选读仍按既有规范排序。这只调整相同事实的顺序，不缩减总正文，也不保证全部

loopx/cli_commands/quota_context_reader.py

@@ -14,6 +14,7 @@
     CONTEXT_CAPTURE_NAVIGATION_GUIDANCE,
     CONTEXT_CAPTURE_PAGE_GUIDANCE,
     CONTEXT_CAPTURE_READ_GUIDANCE,
+    ContextSelectionDigestChanged,
     context_capture_read_command,
     context_selection_binding,
     context_selection_view as _context_selection_view,
@@ -332,12 +333,25 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
             if offset > 0 and selection_digest is None:
                 raise ValueError("selected continuation requires selection-sha256; follow next_command")
             pinned_explicit = pinned_selection is not None and not initial_read
-            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,
-                present_goal_narrative=(pinned_selection.get("present_goal_narrative") is True
-                    if pinned_selection is not None else None),
-                selection_paths=reference.get("selection_paths") if pinned_selection is not None else None)
+            try:
+                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,
+                    present_goal_narrative=(pinned_selection.get("present_goal_narrative") is True
+                        if pinned_selection is not None else None),
+                    selection_paths=reference.get("selection_paths") if pinned_selection is not None else None)
+            except ContextSelectionDigestChanged as exc:
+                # A verified file digest only binds bytes. Preserve an explicit
+                # selection or listing, including a receipt-only diagnostic.
+                # Historical initial-read has no explicit scope to preserve.
+                return {"ok": False, "schema_version": "context_capture_page_v1", "reason": str(exc),
+                    "recovery_command": context_capture_read_command(path, digest,
+                        page_bytes=next_page_bytes or size,
+                        max_page_output_bytes=next_page_output_bytes or output_limit,
+                        select=selectors, list_fields=list_fields, list_depth=list_depth, reader_argv=reader_argv),
+                    "recovery": ("Use recovery_command to read the original capture from offset zero; "
+                        if initial_read else "Use recovery_command to restart the original selected read from offset zero; ")
+                    + "this read performs no guard, admission or spend."}
             if rejection := plan.get("rejected_selection"):
                 # TS alone resolves the nearest container. Reuse the original
                 # escaped tokens, without parsing diagnostic prose or traversing

loopx/control_plane/turn_driver/decision_capture.py

@@ -772,6 +772,10 @@ def _pinned_selection_plan(select: list[str] | None, paths: list[list[str | int]
     return {"selections": selections}
 
 
+class ContextSelectionDigestChanged(ValueError):
+    """The rendered selection no longer matches its recorded digest."""
+
+
 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,
@@ -815,5 +819,5 @@ def render(plan: dict[str, Any], present: bool) -> tuple[bytes, str, list[str]]:
         # enabled. Its original digest must still match the literal view.
         raw, digest, displayed = render(plan, False)
     if expected_sha256 is not None and expected_sha256 != digest:
-        raise ValueError("selected view digest changed; restore the original selection and reader")
+        raise ContextSelectionDigestChanged("selected view digest changed; restore the original selection and reader")
     return {**plan, "presentation_pointers": displayed} if displayed else plan, raw, digest

tests/control_plane/test_context_first_read.py

@@ -373,8 +373,17 @@ def invoke(command):
         + " --offset " + str(pages[0]["next_offset"]))
     assert code != 0 and not rejected["ok"] and "digest changed" in rejected["reason"]
     assert "content" not in rejected
+    recovery = shlex.split(rejected["recovery_command"])
+    assert recovery[recovery.index("--offset") + 1] == "0"
+    assert recovery[recovery.index("--capture-file") + 1] == view["recovery"]["path"]
+    assert recovery[recovery.index("--capture-sha256") + 1] == view["recovery"]["sha256"]
+    assert not {"--initial-read", "--select", "--selection-sha256"} & set(recovery)
+    wrong_capture = recovery.copy()
+    wrong_capture[wrong_capture.index("--capture-sha256") + 1] = "0" * 64
+    code, invalid = invoke(shlex.join(wrong_capture))
+    assert code != 0 and not invalid["ok"] and "recovery_command" not in invalid
     restored = b""
-    command = original
+    command = rejected["recovery_command"]
     while True:
         code, page = invoke(command)
         assert code == 0 and page["offset"] == len(restored), page
@@ -387,6 +396,59 @@ def invoke(command):
 
 
 
+@pytest.mark.parametrize("reader", ["explicit", "saved-recipe", "list"])
+def test_stale_selected_cursor_restarts_in_the_same_scope(tmp_path, reader):
+    from loopx.control_plane.turn_driver.decision_capture import write_context_read_reference
+
+    receipt = {"event_id": "recorded", "turn_instance_id": "original-turn", "note": "原回执" * 90}
+    marker = "This unselected work instruction must stay out of the recovery."
+    capture = (tmp_path / "capture.json").resolve()
+    raw = json.dumps({"heartbeat_receipt": receipt, "interaction_contract": {"primary_action": marker}}).encode()
+    capture.write_bytes(raw)
+    digest = hashlib.sha256(raw).hexdigest()
+    stale = "0" * 64
+    flags = ["--capture-file", str(capture), "--capture-sha256", digest]
+    registry, runtime = tmp_path / "absent-registry.json", tmp_path / "absent-runtime"
+    base = [sys.executable, "-m", "loopx.entrypoint", "--registry", str(registry),
+        "--runtime-root", str(runtime), "--format", "json"]
+    command = ["quota", "context-read", "--offset", "256", "--page-bytes", "4096"]
+    pointer = "/interaction_contract" if reader == "list" else "/heartbeat_receipt"
+    if reader == "saved-recipe":
+        ref = write_context_read_reference(capture, digest, selection={
+            "select": [pointer], "initial_read": False, "list_fields": False,
+            "list_depth": None, "sha256": stale,
+        })
+        reference_before = ref.read_bytes()
+        command += ["--reference-file", str(ref)]
+    else:
+        command += [*flags, "--select", pointer, "--selection-sha256", stale]
+        if reader == "list":
+            command += ["--list", "--list-depth", "2"]
+
+    def run(args):
+        process = subprocess.run([*base, *args], cwd=tmp_path, capture_output=True, text=True, check=False)
+        return process.returncode, json.loads(process.stdout)
+
+    code, failure = run(command)
+    assert code != 0 and "digest changed" in failure["reason"] and "content" not in failure
+    recovery = shlex.split(failure["recovery_command"])
+    assert recovery[recovery.index("--offset") + 1] == "0"
+    assert recovery[recovery.index("--capture-sha256") + 1] == digest
+    assert recovery[recovery.index("--select") + 1] == pointer
+    assert ("--list" in recovery) == (reader == "list")
+    if reader == "list":
+        assert recovery[recovery.index("--list-depth") + 1] == "2"
+    code, page = run(recovery[recovery.index("quota"):])
+    assert code == 0 and page["ok"] and page["eof"] and marker not in page["content"]
+    selections = json.loads(page["content"])["selections"]
+    assert [selection["pointer"] for selection in selections] == [pointer]
+    if reader != "list":
+        assert selections[0]["value"] == receipt
+    if reader == "saved-recipe":
+        assert ref.read_bytes() == reference_before
+    assert capture.read_bytes() == raw and not registry.exists() and not runtime.exists()
+
+
 @pytest.mark.parametrize("bound", [False, True])
 def test_first_read_delivers_current_task_operation_with_its_continuation(tmp_path, decision, bound):
     action = {"operation": "read_private_source", "target_ref": "fixture/private-source.json",

tests/control_plane/test_context_read_reference.py

@@ -112,7 +112,12 @@ def assert_section_read_only(page, machine):
     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"]
+    # With no terminal range in this observation or the journal, verified
+    # bytes can contribute coverage while section digests are deferred.
+    if page["eof"]:
+        assert observed[0]["read_sections"]
+    else:
+        assert observed[0]["read_sections"] == {}
     assert capture_page_range(observed[0], raw=raw, digest=machine["sha256"]) is None
 
 
@@ -197,10 +202,12 @@ def test_selected_continuations_hide_both_digests_and_preserve_exact_values(tmp_
     assert changed["ok"] is False and "cannot change selectors" in changed["reason"]
 
 
-def test_initial_reference_rejects_changed_selection_profile(tmp_path, monkeypatch):
-    _, _, model = fixture(tmp_path)
-    first = read(model["first_read_command"], max_page_output_bytes=3000, page_bytes=3000)
+def test_explicit_initial_cursor_rejects_changed_selection_profile(tmp_path, monkeypatch):
+    _, machine, _ = fixture(tmp_path)
+    first = read(machine["first_read_command"], max_page_output_bytes=3000, page_bytes=3000)
     assert first["ok"] and not first["eof"]
+    cursor = shlex.split(first["next_command"])
+    assert {"--initial-read", "--selection-sha256"} <= set(cursor)
     from loopx.control_plane.quota import turn_envelope
 
     original = turn_envelope._turn_envelope_result