From 1f1ba64e9b4d85c30a82aa54067ae2293dd8bafb Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 07:22:37 +0800 Subject: [PATCH] Reuse resolved paths when paging a pinned context selection Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/reference/required-work-context.md | 6 ++ loopx/cli_commands/quota_context_reader.py | 11 ++- .../turn_driver/decision_capture.py | 71 +++++++++++++---- .../turn_driver/native_context_selection.py | 5 +- .../turn_driver/native_first_read.py | 2 +- .../control_plane/test_context_first_read.py | 78 ++++++++++++++++++- 6 files changed, 151 insertions(+), 22 deletions(-) diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md index 7ac6fe66e1..fdf3276986 100644 --- a/docs/reference/required-work-context.md +++ b/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 diff --git a/loopx/cli_commands/quota_context_reader.py b/loopx/cli_commands/quota_context_reader.py index 2b503cf58d..abcbd0bc70 100644 --- a/loopx/cli_commands/quota_context_reader.py +++ b/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. diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py index cf6c63a147..138789e4e6 100644 --- a/loopx/control_plane/turn_driver/decision_capture.py +++ b/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, diff --git a/loopx/control_plane/turn_driver/native_context_selection.py b/loopx/control_plane/turn_driver/native_context_selection.py index 471e270ac5..2c16ef868b 100644 --- a/loopx/control_plane/turn_driver/native_context_selection.py +++ b/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 diff --git a/loopx/control_plane/turn_driver/native_first_read.py b/loopx/control_plane/turn_driver/native_first_read.py index 1f0c646a5e..24713f2a49 100644 --- a/loopx/control_plane/turn_driver/native_first_read.py +++ b/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 diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py index 915e4ceaef..26eebe8fca 100644 --- a/tests/control_plane/test_context_first_read.py +++ b/tests/control_plane/test_context_first_read.py @@ -260,11 +260,21 @@ def observe(operation, **params): page, values = read_selection(navigation["first_read_command"]) assert page["content"] == original["content"] assert values["/selected_todo"]["future_number"] == 10**45 + 17 - assert len(observed) == 1 if opaque: + from loopx.control_plane.turn_driver.decision_capture import read_context_read_reference, write_context_read_reference + + assert observed == [] + words = shlex.split(navigation["first_read_command"]) + _, metadata = read_context_read_reference(words[words.index("--reference-file") + 1]) + legacy = write_context_read_reference(Path(metadata["path"]), metadata["sha256"], selection=metadata["selection"]) + words[words.index("--reference-file") + 1] = str(legacy) + legacy_page, _ = read_selection(shlex.join(words)) + assert legacy_page["content"] == original["content"] + assert len(observed) == 1 assert b"Historical evidence remains available." not in observed[0] assert len(observed[0]) < len(raw) // 2 else: + assert len(observed) == 1 assert b"Historical evidence remains available." in observed[0] assert Path(view["recovery"]["path"]).read_bytes() == raw @@ -814,6 +824,72 @@ def test_initial_read_pages_pin_the_exact_selection_for_continuation(tmp_path, d assert not changed["ok"] and "digest changed" in changed["reason"] +def test_pinned_value_pages_replay_the_resolved_paths(tmp_path, monkeypatch): + from loopx.control_plane.quota import turn_envelope + from loopx.control_plane.turn_driver.decision_capture import ( + read_context_read_reference, write_context_read_reference, write_decision_capture, + ) + + capture = tmp_path / "decision.json" + payload = {"a/~": {"0": "原捕获的精确正文。" * 160}, "array": [{"number": 9007199254740993}]} + digest = write_decision_capture(capture, payload) + reference = write_context_read_reference(capture, digest) + start = read_context_capture(Namespace(reference_file=str(reference), format="json", offset=0, page_bytes=256, + select=["/a~1~0/0", "/array/0/number"], max_page_output_bytes=5000)) + assert start["ok"] and not start["eof"] + pinned = start["reference_file"] + _, saved = read_context_read_reference(pinned) + assert saved["selection_paths"] == [["array", 0, "number"], ["a/~", "0"]] + + legacy = write_context_read_reference(capture, digest, selection=saved["selection"]) + legacy_page = read_context_capture(Namespace(reference_file=str(legacy), format="json", offset=0, page_bytes=256)) + assert legacy_page["ok"] and legacy_page["content"] == start["content"] + + def unavailable(*_args, **_kwargs): + pytest.fail("a pinned value page should reuse the writer's resolved paths") + + monkeypatch.setattr(turn_envelope, "_turn_envelope_result", unavailable) + page, parts = start, [start["content"]] + while not page["eof"]: + page = read_context_capture(Namespace(reference_file=pinned, format="json", offset=page["next_offset"], + page_bytes=256, max_page_output_bytes=5000)) + assert page["ok"] + parts.append(page["content"]) + assert json.loads("".join(parts))["selections"] == [ + {"pointer": "/array/0/number", "value": payload["array"][0]["number"]}, + {"pointer": "/a~1~0/0", "value": payload["a/~"]["0"]}, + ] + + +@pytest.mark.parametrize("damage", ["different_path", "wrong_key_type", "view_digest", "capture_digest"]) +def test_pinned_value_page_rejects_changed_path_or_bytes(tmp_path, damage): + from loopx.control_plane.turn_driver.decision_capture import ( + read_context_read_reference, write_context_read_reference, write_decision_capture, + ) + + capture = tmp_path / "decision.json" + digest = write_decision_capture(capture, {"array": ["value"], "alias": ["value"]}) + original = write_context_read_reference(capture, digest) + first = read_context_capture(Namespace(reference_file=str(original), format="json", offset=0, page_bytes=256, + select=["/array/0"])) + assert first["ok"] + _, saved = read_context_read_reference(first["reference_file"]) + selection, paths = saved["selection"], saved["selection_paths"] + if damage == "different_path": + paths = [["alias", 0]] # The value itself is identical. + elif damage == "wrong_key_type": + paths = [["array", "0"]] + elif damage == "view_digest": + selection = {**selection, "sha256": "0" * 64} + else: + capture.write_text('{"array":["changed"],"alias":["value"]}', encoding="utf-8") + reference = write_context_read_reference(capture, digest, selection=selection, selection_paths=paths) + rejected = read_context_capture(Namespace(reference_file=str(reference), format="json", offset=0, page_bytes=256)) + assert not rejected["ok"] and "content" not in rejected + reason = "digest" if "digest" in damage else "address" + assert reason in rejected["reason"] + + def test_small_initial_pages_progress_without_repeating_the_pointer_vector(tmp_path, decision): view = native_context_capture_view(decision, tmp_path, max_page_output_bytes=4500) assert "first_read_command" in view