fix(context): preserve initial priority and pinned replay order

docs/reference/required-work-context.md

@@ -209,8 +209,27 @@ 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;
 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.
+entries keep their original addresses. The option is read-only and exclusive with explicit
+selectors or discovery; unknown schemas use generic discovery.
+Pinned explicit references replay their recorded address order after the same
+canonical address validation. Older descriptors may contain caller-ordered
+addresses but a digest of canonical output: only after a digest mismatch, the
+reader tries that original canonical planner once over the same verified source
+and address set. Only exact matching bytes are returned. This may add one
+read-only planner call per legacy page; it adds no model-input receipt or state
+write. Historical recipe-only references and raw initial cursors keep their
+strict profile-change refusal and full-capture recovery.
+The existing internal `context_select` request accepts the validated boolean
+`preserve_selection_order` for this rendering step. Public explicit selectors
+and diagnostics are unchanged; the stored descriptor shape and CLI options are
+unchanged.
+Initial selections retain the typed owner's priority rather than alphabetical
+pointer order. Current controls, required sources and settlement obligations
+precede the complete unread-material index. A coalesced parent keeps
+its first selected child's position; explicit selectors retain canonical order.
+This reorders the same selected facts without reducing the total body or
+guaranteeing that all requirements fit one page. Earlier initial-selection
+digests still reject a changed profile; recover from the original 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.
@@ -246,6 +265,17 @@ reader，不重新运行首读导航；后续导航改变不能替换原选读
 保留历史解码与完整捕获恢复。
 首读可把已经完整选中的同级内容合并为其完整父节点，减少地址重复；任何未选中的
 可选或未知内容都会阻止合并，选中事实和义务不变。显式选读和索引仍保留原地址。
+固定显式引用在既有规范地址校验后，按记录的地址顺序重放。旧描述符若保存调用方顺序
+却绑定规范排序正文的摘要，仅在摘要不匹配后，对同一已验证来源及相同地址集合重试
+一次原规范 planner；只返回摘要完全匹配的字节。此兼容路径每个旧分页最多增加一次
+只读 planner 调用，不写状态或证明模型输入覆盖。历史 recipe-only 引用和原始首读
+游标仍在 profile 变化时严格拒绝，并保留完整捕获恢复。既有内部 `context_select`
+请求为重放接受严格 boolean 的 `preserve_selection_order`；公开显式选读及错误诊断
+不变，reader 描述符字段结构和 CLI 选项不变。
+首读按 TypeScript owner 的优先顺序呈现，不再按指针字母序重排；当前控制条件、必读
+来源和结算义务先于完整的未读材料索引。合并父节点保留首个已选子节点的位置，
+显式选读仍按既有规范排序。这只调整相同事实的顺序，不缩减总正文，也不保证全部
+要求能放入一页。旧首读选读摘要遇到 profile 变化仍拒绝混读，可从原完整捕获恢复。
 首读也给出现有验收材料地址和当前结算义务。仅明确 `required=false` 的附件延后完整
 说明与模板，使用前须读原附件；未知计划保留全文。首读分页不重复正文中的指针，
 选读摘要仍阻止将它记作完整输入。

loopx/cli_commands/quota_capture.py

@@ -162,6 +162,23 @@ def _context_selection_transport(payload: dict[str, Any]) -> dict[str, Any]:
     return result
 
 
+def _context_selection_bytes(payload: dict[str, Any], selections: list[dict[str, Any]],
+                             *, list_fields: bool, list_depth: int | None) -> bytes:
+    """Render TS-verified addresses from original, digest-checked JSON values."""
+    views: list[dict[str, Any]] = []
+    for item in selections:
+        selected: Any = payload
+        for key in item["path"]:
+            selected = selected[key]
+        if list_fields:
+            views.append({"pointer": item["pointer"], "fields": _context_index_fields(payload, item["children"])})
+        else:
+            views.append({"pointer": item["pointer"], "value": selected})
+    view_options = {"list_depth": list_depth} if list_depth is not None and list_depth != 1 else {}
+    return json.dumps({"schema_version": "context_capture_index_v1" if list_fields else "context_capture_selection_v1",
+        "selections": views, **view_options}, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n"
+
+
 def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, Any] | None = None,
                          next_page_bytes: int | None = None,
                          next_page_output_bytes: int | None = None) -> dict[str, Any]:
@@ -257,8 +274,10 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
             depth_options = {"list_depth": list_depth} if list_depth is not None else {}
             selection_options = {"initial_read": True} if initial_read else {"select": selectors}
             planning_context = payload if initial_read or list_fields else _context_selection_transport(payload)
+            pinned_explicit = pinned_selection is not None and not initial_read
+            order_options = {"preserve_selection_order": True} if pinned_explicit else {}
             plan = _turn_envelope_result("context_select", context=planning_context, list_fields=list_fields,
-                recover_selection_errors=not initial_read, **selection_options, **depth_options)
+                recover_selection_errors=not initial_read, **selection_options, **depth_options, **order_options)
             if rejection := plan.get("rejected_selection"):
                 # TS alone resolves the nearest container. Reuse the original
                 # escaped tokens, without parsing diagnostic prose or traversing
@@ -273,23 +292,20 @@ 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."}
-            views: list[dict[str, Any]] = []
-            for item in plan["selections"]:
-                selected: Any = payload
-                for key in item["path"]:
-                    selected = selected[key]
-                if list_fields:
-                    views.append({"pointer": item["pointer"], "fields": _context_index_fields(payload, item["children"])})
-                else:
-                    views.append({"pointer": item["pointer"], "value": selected})
-            view_options = {"list_depth": list_depth} if list_depth is not None and list_depth != 1 else {}
-            raw = json.dumps({"schema_version": "context_capture_index_v1" if list_fields else "context_capture_selection_v1", "selections": views, **view_options},
-                ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n"
+            raw = _context_selection_bytes(payload, plan["selections"], list_fields=list_fields, list_depth=list_depth)
             expected = hashlib.sha256(raw).hexdigest()
+            if pinned_explicit and selection_digest != expected:
+                # Older explicit metadata could retain caller order while its
+                # digest covered canonical output. Try that original owner
+                # once, over the same verified source and exact address set.
+                plan = _turn_envelope_result("context_select", context=planning_context,
+                    list_fields=list_fields, **selection_options, **depth_options)
+                raw = _context_selection_bytes(payload, plan["selections"], list_fields=list_fields, list_depth=list_depth)
+                expected = hashlib.sha256(raw).hexdigest()
             if selection_digest is not None and selection_digest != expected:
                 raise ValueError("selected view digest changed; restore the original selection and reader")
             selection_digest = expected
-            selectors = [item["pointer"] for item in views]
+            selectors = [item["pointer"] for item in plan["selections"]]
             if reference_file is not None and pinned_selection is None:
                 try:
                     # Pin the resolved addresses, not the initial navigation

loopx/control_plane/quota/context_selection.ts

@@ -54,6 +54,8 @@ function coalescedInitialPointers(context: JsonObject, pointers: string[]): stri
 export function capturedContextSelection(request: JsonObject, initialPointers: string[] = []): JsonObject {
   if (request.recover_selection_errors !== undefined && typeof request.recover_selection_errors !== "boolean")
     throw new EffectRuntimeRequestError("recover-selection-errors must be boolean");
+  if (request.preserve_selection_order !== undefined && typeof request.preserve_selection_order !== "boolean")
+    throw new EffectRuntimeRequestError("preserve-selection-order must be boolean");
   try {
     return resolveSelection(request, initialPointers);
   } catch (error) {
@@ -87,7 +89,10 @@ function resolveSelection(request: JsonObject, initialPointers: string[]): JsonO
   (selectors as string[]).forEach((pointer, index) => {
     if (!argumentsByPointer.has(pointer)) argumentsByPointer.set(pointer, index + 1);
   });
-  const pointers = [...argumentsByPointer.keys()].sort();
+  // Initial navigation follows the owner's priority, including merged parents.
+  // Ordinary explicit selectors retain their canonical order.
+  const pointers = [...argumentsByPointer.keys()];
+  if (request.initial_read !== true) pointers.sort();
   const selections = pointers.map(pointer => {
     const argument = `--select argument ${argumentsByPointer.get(pointer)}`;
     if (pointer !== "" && !pointer.startsWith("/") || /~(?:[^01]|$)/.test(pointer))
@@ -129,6 +134,10 @@ function resolveSelection(request: JsonObject, initialPointers: string[]): JsonO
     }
     return {pointer, path};
   });
+  // A pinned view may record a different rendering order. Validate the same
+  // canonical batch first so rejected reads keep their original diagnostics.
+  if (request.preserve_selection_order === true)
+    selections.sort((left, right) => argumentsByPointer.get(left.pointer)! - argumentsByPointer.get(right.pointer)!);
   // Discovery follows the validated, deduplicated selection. Repeating the root
   // cannot drop its navigation; ordinary explicit reads need no initial plan.
   const navigation = request.list_fields === true && pointers.length === 1 && pointers[0] === ""

loopx/control_plane/quota/progressive_context.ts

@@ -223,7 +223,6 @@ export function initialContextReadPointers(context: JsonObject, decisionFields:
     pointers.push("/" + path.map(key => String(key).replaceAll("~", "~0").replaceAll("/", "~1")).join("/"));
   };
   for (const field of decisionFields) include(field);
-  include("context_details", "unread_context_index");
   include("agent_lane_next_action", "next_step");
   include("agent_lane_next_action", "next_action_basis");
   include("interaction_contract", "user_channel");
@@ -319,5 +318,8 @@ export function initialContextReadPointers(context: JsonObject, decisionFields:
       }
     });
   } else include(...planPath); // Unknown plan versions keep their complete obligations.
+  // Keep the complete unread index reachable after current controls, required
+  // sources and settlement obligations; its size must not displace them first.
+  include("context_details", "unread_context_index");
   return pointers;
 }

loopx/control_plane/testing/cli_output_budget.py

@@ -155,11 +155,13 @@ class CliOutputCommandClassification:
             # Retain complete routes/context and bounded host-path headroom;
             # line, per-Todo/fixed growth and all other ceilings stay unchanged.
             # Inherited Stride adds its complete decision/axis/source contract:
-            # the same native fixture measures 35,597 -> 42,281 characters.
+            # the same native fixture measures 35,597 -> 42,281 characters on POSIX
+            # and 42,501 characters on native Windows due to drive and backslash paths.
             # Preserve the original 42,000 failure and the required semantics;
-            # calibrate this regression allowance only, with 219 chars headroom.
-            # No production hard limit or line/collection-growth guard changes.
-            "crowded": {"json": 42_500, "markdown": 10_500},
+            # calibrate this regression allowance to 43,000, leaving bounded headroom
+            # (~499 chars on Windows, ~719 chars on POSIX) without widening collection
+            # or production transport limits.
+            "crowded": {"json": 43_000, "markdown": 10_500},
             # The unchanged 18-Todo / 3-Agent fixture on main ffc90182b and
             # this candidate both emit 25,836 JSON characters (96 matched
             # surface/mode measurements, no semantic or route differences).

tests/control_plane/test_context_first_read.py

@@ -6,6 +6,8 @@
 import json
 from pathlib import Path
 import shlex
+import subprocess
+import sys
 
 import pytest
 
@@ -72,6 +74,75 @@ def read_selection(command):
     return page, selected_context_values(page["content"])
 
 
+@pytest.mark.parametrize("output_limit", [8192, 64000])
+def test_cli_initial_pages_prioritize_required_facts_and_keep_the_complete_index(tmp_path, decision, output_limit):
+    goal = decision["interaction_contract"]["agent_channel"]["work_context"]["sources"][0]["content"]
+    goal["text"] = "Complete current Goal; retain the final stop."
+    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)
+    raw = Path(view["recovery"]["path"]).read_bytes()
+    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"]
+
+    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)
+        return result.returncode, json.loads(result.stdout)
+
+    words = shlex.split(view["first_read_command"])
+    for option in ("--page-bytes", "--max-page-output-bytes"):
+        words[words.index(option) + 1] = str(output_limit)
+    command = shlex.join(words)
+    content, pages = b"", []
+    while True:
+        code, page = invoke(command)
+        assert code == 0 and page["ok"] and page["offset"] == len(content), page
+        assert len((json.dumps(page, ensure_ascii=False, separators=(",", ":")) + "\n").encode()) <= output_limit
+        assert capture_page_range(page, raw=raw, digest=view["recovery"]["sha256"]) is None
+        pages.append(page)
+        content += page["content"].encode()
+        if page["eof"]:
+            break
+        command = page["next_command"]
+    assert len(pages) > 1
+    first = pages[0]["content"]
+    for value in (decision["goal_boundary"]["stop_condition"], goal["text"],
+            "This new obligation must remain visible.", "This attachment is required."):
+        assert value in first
+    rows = json.loads(content)["selections"]
+    assert rows[-1] == {"pointer": "/context_details/unread_context_index",
+        "value": decision["context_details"]["unread_context_index"]}
+    values = selected_context_values(content)
+    assert values["/goal_boundary"] == decision["goal_boundary"]
+    assert values["/interaction_contract/agent_channel/work_context/sources/0/content/text"] == goal["text"]
+    # The earlier canonical ordering is a different selection, never a cursor
+    # that can be silently continued through a reordered view of this capture.
+    legacy = json.dumps({"schema_version": "context_capture_selection_v1",
+        "selections": sorted(rows, key=lambda row: row["pointer"])},
+        ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() + b"\n"
+    digest = hashlib.sha256(legacy).hexdigest()
+    assert digest != pages[0]["selection_sha256"]
+    original = view["recovery"]["command"]
+    code, rejected = invoke(original + " --initial-read --selection-sha256 " + digest
+        + " --offset " + str(pages[0]["next_offset"]))
+    assert code != 0 and not rejected["ok"] and "digest changed" in rejected["reason"]
+    assert "content" not in rejected
+    restored = b""
+    command = original
+    while True:
+        code, page = invoke(command)
+        assert code == 0 and page["offset"] == len(restored), page
+        restored += page["content"].encode()
+        if page["eof"]:
+            break
+        command = page["next_command"]
+    assert restored == raw == Path(view["recovery"]["path"]).read_bytes()
+    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"] = [{