Reuse resolved addresses when reading pinned context pages

docs/reference/required-work-context.md

@@ -1221,6 +1221,10 @@ profile. Its short filename is only a locator, never a truncated integrity check
 permission or freshness claim. Missing, malformed, public or symlinked reader
 documents and capture/selection digest mismatches fail closed without admission. Selection reads may publish a derived
 private reader document beside the capture; they do not write Goal state.
+For new pinned value selections, the private reference can retain the selector's
+resolved addresses. The reader and native observation reuse those addresses
+against the verified source, then check the rendered selection digest. Older
+references and new discovery requests use the existing selector.
 Compact selected pages use `context_capture_selection_page_v1`, with a reader
 reference and byte cursors instead of repeated hashes or pointer vectors. They
 are never full-input observations. Explicit file/hash readers and full-page
@@ -1235,6 +1239,8 @@ the compact navigation cannot fit; no navigation or source is truncated.
 引用缺失、格式错误、公开可读、为符号链接或摘要不符时拒绝读取，不重新 admission。选读可在
 捕获旁生成派生私有引用，不修改 Goal 状态。精简选读页返回引用及字节游标，使用
 独立 schema，不能作为完整输入证明；原显式文件/摘要接口及完整页证明保持兼容。
+新生成的值选读引用可保存选择器已解析的地址。读页和原生观察使用校验过的原捕获
+按地址取值，再校验渲染后的选读摘要。旧引用和新字段发现继续使用现有选择器。
 导航引用同时保存完整机器元数据；极小输出预算保留原显式摘要引用，不截断正文或导航。
 
 `quota context-read --select /path --select /another/path` reads an explicit

loopx/cli_commands/quota_context_reader.py

@@ -283,7 +283,7 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
                     fallback = observed_first_read_fallback(observation, path, digest)
                     if fallback is not None:
                         try:
-                            _validate_pinned_selection(fallback[1])
+                            _validate_pinned_selection(fallback[1]["selection"])
                         except ValueError:
                             fallback = None
                     if offset:
@@ -293,7 +293,8 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
                                 page_bytes=size, max_page_output_bytes=output_limit, reader_argv=reader_argv)}
                     # Restart in the ordinary complete view of this same capture.
                     if fallback is not None:
-                        reference_file, pinned_selection = fallback
+                        reference_file, reference = fallback
+                        pinned_selection = reference["selection"]
                     else:
                         pinned_selection, initial_read = None, True
                         try:
@@ -328,7 +329,8 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
                 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))
+                    if pinned_selection is not None else None),
+                selection_paths=reference.get("selection_paths") if pinned_selection is not None else None)
             if rejection := plan.get("rejected_selection"):
                 # TS alone resolves the nearest container. Reuse the original
                 # escaped tokens, without parsing diagnostic prose or traversing
@@ -356,7 +358,8 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
                     # bytes after a navigation change, with both digests checked.
                     reference_file = write_context_read_reference(path, digest, selection=context_selection_binding(
                         selectors, list_fields=list_fields, list_depth=list_depth, sha256=expected,
-                        presented=bool(plan.get("presentation_pointers"))))
+                        presented=bool(plan.get("presentation_pointers"))),
+                        selection_paths=[item["path"] for item in plan["selections"]] if not list_fields else None)
                 except OSError:
                     # A verified read must not require a writable capture root.
                     # Legacy cursors pin the same capture, selection and digest.

loopx/control_plane/turn_driver/decision_capture.py

@@ -345,7 +345,8 @@ def context_capture_read_command(path: Path, digest: str, *, offset: int = 0,
 
 
 def write_context_read_reference(path: Path, digest: str, *, selection: dict[str, Any] | None = None,
-                                 navigation: dict[str, Any] | None = None, observed_read: dict[str, Any] | None = None) -> Path:
+                                 navigation: dict[str, Any] | None = None, observed_read: dict[str, Any] | None = None,
+                                 selection_paths: list[list[str | int]] | None = None) -> Path:
     """Immutable reader IO metadata, not a new identity or authority store.
 
     A short filename locates the document; only its complete stored digests are
@@ -358,6 +359,13 @@ def write_context_read_reference(path: Path, digest: str, *, selection: dict[str
         payload["navigation"] = navigation
     if observed_read is not None:
         payload["observed_read"] = observed_read
+    if selection is not None and selection_paths is not None:
+        from ..effect_runtime import MAX_REQUEST_BYTES
+
+        # A large descriptor can still use the older selector replay route.
+        with_paths = {**payload, "selection_paths": selection_paths}
+        if len(_decision_json(with_paths).encode("utf-8")) <= MAX_REQUEST_BYTES:
+            payload = with_paths
     suffix = ("." + hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest()[:16]
               if selection is not None or navigation is not None else "")
     reference = path.with_name(path.stem + suffix + ".reader.json")
@@ -400,7 +408,9 @@ def read_context_read_reference(value: str) -> tuple[Path, dict[str, Any]]:
     if (not isinstance(payload, dict) or payload.get("schema_version") != "context_capture_read_reference_v1"
             or not isinstance(payload.get("path"), str) or not isinstance(payload.get("sha256"), str)
             or BARE_SHA256_PATTERN.fullmatch(payload["sha256"]) is None
-            or ("selection" in payload and not isinstance(payload["selection"], dict))):
+            or ("selection" in payload and not isinstance(payload["selection"], dict))
+            or ("selection_paths" in payload and (
+                "selection" not in payload or not isinstance(payload["selection_paths"], list)))):
         raise ValueError("reader reference is incomplete or unsupported")
     return path, payload
 
@@ -570,6 +580,10 @@ def context_index_fields(payload: dict[str, Any], children: list[dict[str, Any]]
     return fields
 
 
+def _pointer_token(key: str | int) -> str:
+    return str(key).replace("~", "~0").replace("/", "~1")
+
+
 def context_selection_transport(payload: dict[str, Any], select: list[str] | None, *,
                                 list_fields: bool, list_depth: int | None) -> dict[str, Any]:
     """Supply the address shape for the TS selector. Values are read from the verified original."""
@@ -597,7 +611,7 @@ def context_selection_transport(payload: dict[str, Any], select: list[str] | Non
         if depth <= 0 and not any(key is not None for key in branch):
             continue
         for key, value in source.items() if isinstance(source, dict) else enumerate(source):
-            token = str(key).replace("~", "~0").replace("/", "~1")
+            token = _pointer_token(key)
             if depth <= 0 and token not in branch:
                 continue
             if isinstance(value, (dict, list)):
@@ -652,7 +666,12 @@ def _context_selection_bytes(payload: dict[str, Any], selections: list[dict[str,
         selected: Any = payload
         path, pointer = item["path"], item["pointer"]
         for key in path:
-            selected = selected[key]
+            if type(key) is str and isinstance(selected, dict) and key in selected:
+                selected = selected[key]
+            elif type(key) is int and isinstance(selected, list) and 0 <= key < len(selected):
+                selected = selected[key]
+            else:
+                raise ValueError("selected address does not match the original capture")
         if list_fields:
             views.append({"pointer": pointer, "fields": context_index_fields(payload, item["children"])})
         else:
@@ -683,25 +702,47 @@ def write_initial_context_read_reference(path: Path, digest: str, payload: dict[
     reference = write_context_read_reference(path, digest,
         selection=context_selection_binding([item["pointer"] for item in selections], list_fields=False,
             list_depth=None, sha256=hashlib.sha256(raw).hexdigest(), presented=bool(presented)),
+        selection_paths=[item["path"] for item in selections],
         observed_read=observed_read)
     return reference, raw
 
 
+def _pinned_selection_plan(select: list[str] | None, paths: list[list[str | int]]) -> dict[str, Any]:
+    """Replay TypeScript's resolved addresses from the existing reader reference."""
+    if (not isinstance(select, list) or not select or len(select) != len(paths)
+            or not all(isinstance(pointer, str) for pointer in select) or len(set(select)) != len(select)):
+        raise ValueError("reader selection paths do not match its pinned addresses")
+    selections = []
+    for pointer, path in zip(select, paths, strict=True):
+        if (not isinstance(path, list)
+                or not all(type(key) in (str, int) for key in path)
+                or pointer != "".join("/" + _pointer_token(key) for key in path)):
+            raise ValueError("reader selection paths do not match its pinned addresses")
+        selections.append({"pointer": pointer, "path": path})
+    return {"selections": selections}
+
+
 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,
                            present_goal_narrative: bool | None = None,
+                           selection_paths: list[list[str | int]] | None = None,
                            ) -> tuple[dict[str, Any], bytes | None, str | None]:
-    """Replay the TS selector and verify the original digest-pinned view."""
-    from ..quota.turn_envelope import _turn_envelope_result
-
-    context = payload if initial_read else context_selection_transport(
-        payload, select, list_fields=list_fields, list_depth=list_depth)
-    selection = {"initial_read": True} if initial_read else {"select": select}
-    depth = {"list_depth": list_depth} if list_depth is not None else {}
-    order = {"preserve_selection_order": True} if preserve_selection_order else {}
-    plan = _turn_envelope_result("context_select", context=context, list_fields=list_fields,
-        recover_selection_errors=not initial_read, **selection, **depth, **order)
+    """Render the digest-pinned view from a new or previously resolved plan."""
+    if selection_paths is not None:
+        if initial_read or list_fields or list_depth is not None or not preserve_selection_order or expected_sha256 is None:
+            raise ValueError("reader selection paths require a pinned value selection")
+        plan = _pinned_selection_plan(select, selection_paths)
+    else:
+        from ..quota.turn_envelope import _turn_envelope_result
+
+        context = payload if initial_read else context_selection_transport(
+            payload, select, list_fields=list_fields, list_depth=list_depth)
+        selection = {"initial_read": True} if initial_read else {"select": select}
+        depth = {"list_depth": list_depth} if list_depth is not None else {}
+        order = {"preserve_selection_order": True} if preserve_selection_order else {}
+        plan = _turn_envelope_result("context_select", context=context, list_fields=list_fields,
+            recover_selection_errors=not initial_read, **selection, **depth, **order)
     if plan.get("rejected_selection"):
         return plan, None, None
     presentation = initial_read if present_goal_narrative is None else present_goal_narrative
@@ -712,7 +753,7 @@ def render(plan: dict[str, Any], present: bool) -> tuple[bytes, str, list[str]]:
         return raw, hashlib.sha256(raw).hexdigest(), displayed
 
     raw, digest, displayed = render(plan, presentation)
-    if preserve_selection_order and expected_sha256 != digest:
+    if selection_paths is None and preserve_selection_order and expected_sha256 != digest:
         # Historical explicit references may pin canonical output despite
         # retaining caller order. Only the same source and addresses can replay.
         plan = _turn_envelope_result("context_select", context=context,

loopx/control_plane/turn_driver/native_context_selection.py

@@ -24,10 +24,12 @@ def verified_context_selection(state: dict[str, Any], page: dict[str, Any]) -> d
     reference_file = page.get("reference_file")
     selected = None
     observation = None
+    selection_paths = None
     if isinstance(reference_file, str):
         _, metadata = read_context_read_reference(reference_file)
         selected = metadata.get("selection")
         observation = metadata.get("observed_read")
+        selection_paths = metadata.get("selection_paths")
         if not isinstance(selected, dict) or selected.get("list_fields") is not False:
             return None
         path, source_digest = metadata["path"], metadata["sha256"]
@@ -65,7 +67,8 @@ def verified_context_selection(state: dict[str, Any], page: dict[str, Any]) -> d
             initial_read=initial_read, list_fields=False,
             list_depth=selected.get("list_depth") if selected else None,
             expected_sha256=view_digest, preserve_selection_order=selected is not None and not initial_read,
-            present_goal_narrative=bool(selected and selected.get("present_goal_narrative") is True))
+            present_goal_narrative=bool(selected and selected.get("present_goal_narrative") is True),
+            selection_paths=selection_paths)
         if selected_raw is None:
             return None
         raw = selected_raw

loopx/control_plane/turn_driver/native_first_read.py

@@ -52,7 +52,7 @@ def observed_first_read_fallback(observation: dict[str, Any], path: Path, digest
                 and "observed_read" not in metadata and isinstance(selection, dict)
                 and isinstance(fallback.get("selection_sha256"), str)
                 and selection.get("sha256") == fallback["selection_sha256"]):
-            return reference, selection
+            return reference, metadata
     except (OSError, ValueError):
         pass
     return None