fix(context): replay frozen selection order across pinned reader pages

docs/reference/required-work-context.md

@@ -204,6 +204,12 @@ An unselected child, including optional or unknown material, prevents that merge
 the selected facts and obligations stay identical. Explicit selectors and index
 entries keep their original addresses. The option is read-only and exclusive with explicit
 selectors or discovery; unknown schemas use generic discovery.
+An immutable private reference with resolved selected addresses also freezes
+their serialization order. The typed selector validates those same addresses;
+the reader replays the stored order before checking the complete selection
+digest, rather than applying a fresh explicit request's canonical sorting.
+Fresh explicit reads keep canonical order. Historical recipe-only references
+still reject a changed selection and recover from the same full capture.
 The concrete first read replaces the broad scalar preview and root index when it
 fits; unknown schemas or a tight reference budget keep generic discovery. Warm
 deltas keep their existing reader, and disabled native delivery is unchanged.
@@ -235,6 +241,10 @@ activation defaults, saved files, Delta and full-read receipts are unchanged.
 显式选读或索引混用；续页保留短选项和固定的选读摘要，profile 改变则拒绝混读。
 首读可把已经完整选中的同级内容合并为其完整父节点，减少地址重复；任何未选中的
 可选或未知内容都会阻止合并，选中事实和义务不变。显式选读和索引仍保留原地址。
+已固定解析地址的不可变私有引用也固定其正文序列化顺序。typed selector 校验同一组
+地址，reader 按引用中的顺序重放后验证完整选读摘要，不重新采用普通显式请求的
+规范排序。新的显式选读仍按既有规范排序；历史仅保存导航配方的引用在选读变化时
+仍拒绝混读，并沿同一完整原捕获恢复。
 首读也给出现有验收材料地址和当前结算义务。仅明确 `required=false` 的附件延后完整
 说明与模板，使用前须读原附件；未知计划保留全文。首读分页不重复正文中的指针，
 选读摘要仍阻止将它记作完整输入。

loopx/cli_commands/quota_capture.py

@@ -273,6 +273,13 @@ 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."}
+            if pinned_selection is not None and not initial_read and selectors is not None:
+                # TS validates the addresses; this IO reference already froze
+                # their serialization order, independent of fresh selection policy.
+                positions = {pointer: index for index, pointer in enumerate(dict.fromkeys(selectors))}
+                if positions.keys() != {item["pointer"] for item in plan["selections"]}:
+                    raise ValueError("selected view digest changed; restore the original selection and reader")
+                plan["selections"].sort(key=lambda item: positions[item["pointer"]])
             views: list[dict[str, Any]] = []
             for item in plan["selections"]:
                 selected: Any = payload

tests/control_plane/test_context_first_read.py

@@ -72,6 +72,50 @@ def read_selection(command):
     return page, selected_context_values(page["content"])
 
 
+@pytest.mark.parametrize("output_limit", [8192, 64000])
+def test_cli_initial_reference_pages_keep_original_selection(tmp_path, decision, output_limit):
+    import subprocess
+    import sys
+    from loopx.cli_commands.quota_capture import model_context_reference
+
+    decision["context_details"] = {"unread_context_index": {
+        "coverage_confirmed": False, "pointers": [f"/history/{i}" for i in range(6000)]}}
+    view = native_context_capture_view(decision, tmp_path)
+    model = model_context_reference(view)
+    raw = Path(view["recovery"]["path"]).read_bytes()
+    registry, runtime = tmp_path / "absent-registry.json", tmp_path / "absent-runtime"
+    base = [sys.executable, "-m", "loopx.entrypoint", "--format", "json",
+        "--registry", str(registry), "--runtime-root", str(runtime)]
+
+    def drain(command):
+        words = shlex.split(command)
+        for flag in ("--page-bytes", "--max-page-output-bytes"):
+            words[words.index(flag) + 1] = str(output_limit)
+        content, pages = b"", 0
+        while True:
+            result = subprocess.run([*base, *words[words.index("quota"):]], cwd=tmp_path,
+                capture_output=True, text=True, check=False)
+            page = json.loads(result.stdout)
+            assert result.returncode == 0 and page["ok"] and page["offset"] == len(content), page
+            assert len(result.stdout.encode()) <= output_limit
+            assert capture_page_range(page, raw=raw, digest=view["recovery"]["sha256"]) is None
+            content += page["content"].encode()
+            pages += 1
+            if page["eof"]:
+                return content, pages
+            words = shlex.split(page["next_command"])
+
+    expected, pages = drain(view["first_read_command"])
+    actual, reference_pages = drain(model["first_read_command"])
+    assert actual == expected and reference_pages > 1 and pages > 1
+    values = selected_context_values(actual)
+    assert values["/goal_boundary"] == decision["goal_boundary"]
+    assert values["/context_details/unread_context_index"] == decision["context_details"]["unread_context_index"]
+    assert "This new obligation must remain visible." in actual.decode()
+    assert Path(view["recovery"]["path"]).read_bytes() == raw
+    assert not registry.exists() and not runtime.exists()
+
+
 def test_initial_read_coalesces_whole_contract_with_exact_values_and_unknown_obligations(tmp_path, decision):
     cli = decision["interaction_contract"]["cli_channel"]
     cli["settlement_plan"]["ordered_steps"] = [{

tests/control_plane/test_context_read_reference.py

@@ -3,6 +3,8 @@
 import json
 from pathlib import Path
 import shlex
+import subprocess
+import sys
 
 import pytest
 
@@ -132,6 +134,67 @@ def changed(operation, **kwargs):
     assert "content" not in rejected
 
 
+@pytest.mark.parametrize("output_limit", [8192, 64000])
+def test_cli_frozen_selection_replays_its_order_without_changing_explicit_reads(tmp_path, output_limit):
+    payload = {"ok": True, "goal_boundary": {"stop": "Keep the original budget."},
+        "selected_todo": {"text": "Compare 原始条件。" * 4000},
+        "criterion": {"value": 9007199254740993}}
+    machine = native_context_capture_view(payload, tmp_path)
+    capture, digest = Path(machine["path"]), machine["sha256"]
+    before = capture.read_bytes()
+    pointers = ["/goal_boundary", "/selected_todo", "/criterion"]
+
+    def selected_bytes(order):
+        return (json.dumps({"schema_version": "context_capture_selection_v1", "selections": [
+            {"pointer": pointer, "value": payload[pointer[1:]]} for pointer in order]},
+            ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n").encode()
+
+    expected = selected_bytes(pointers)
+    selection = {"select": pointers, "initial_read": False, "list_fields": False,
+        "list_depth": None, "sha256": hashlib.sha256(expected).hexdigest()}
+    reference = write_context_read_reference(capture, digest, selection=selection)
+    registry, runtime = tmp_path / "absent-registry.json", tmp_path / "absent-runtime"
+    base = [sys.executable, "-m", "loopx.entrypoint", "--format", "json",
+        "--registry", str(registry), "--runtime-root", str(runtime)]
+    options = f" --page-bytes {output_limit} --max-page-output-bytes {output_limit}"
+
+    def invoke(command):
+        words = shlex.split(command)
+        result = subprocess.run([*base, *words[words.index("quota"):]], cwd=tmp_path,
+            capture_output=True, text=True, check=False)
+        assert len(result.stdout.encode()) <= output_limit
+        return result.returncode, json.loads(result.stdout)
+
+    def drain(command):
+        content, count = b"", 0
+        while True:
+            code, page = invoke(command)
+            assert code == 0 and page["ok"] and page["offset"] == len(content), page
+            assert capture_page_range(page, raw=before, digest=digest) is None
+            content += page["content"].encode()
+            count += 1
+            if page["eof"]:
+                return content, count
+            command = page["next_command"]
+
+    frozen = "loopx quota context-read --reference-file " + shlex.quote(str(reference)) + options
+    actual, pages = drain(frozen)
+    assert actual == expected and pages > 1
+    _, pinned = read_context_read_reference(str(reference))
+    assert pinned["selection"] == selection
+    # A fresh explicit request still has canonical order, even with the same
+    # arguments that an immutable reference has already resolved and frozen.
+    original = write_context_read_reference(capture, digest)
+    explicit = "loopx quota context-read --reference-file " + shlex.quote(str(original)) + options
+    explicit += " " + " ".join("--select " + pointer for pointer in pointers)
+    assert drain(explicit)[0] == selected_bytes(sorted(pointers))
+    damaged = write_context_read_reference(capture, digest, selection={**selection, "sha256": "f" * 64})
+    code, rejected = invoke("loopx quota context-read --reference-file " + shlex.quote(str(damaged)) + options)
+    assert code != 0 and not rejected["ok"] and "digest changed" in rejected["reason"]
+    assert "content" not in rejected
+    assert capture.read_bytes() == before and not registry.exists() and not runtime.exists()
+
+
 @pytest.mark.parametrize("selection", ["fields", "index", "initial"])
 def test_selected_read_survives_unwritable_reference_metadata(tmp_path, monkeypatch, selection):
     import errno