diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py index 91f3ea8e21..e81f2fe2b7 100644 --- a/loopx/control_plane/turn_driver/decision_capture.py +++ b/loopx/control_plane/turn_driver/decision_capture.py @@ -462,26 +462,46 @@ def context_index_fields(payload: dict[str, Any], children: list[dict[str, Any]] return fields -def context_selection_transport(payload: dict[str, Any]) -> dict[str, Any]: - """Keep address structure without retransmitting unused scalar prose. - - The TS owner validates pointers; selected values still come from the fully - digest-checked original. Initial navigation and discovery need real scalar - values and do not use this adapter. JSON null takes four bytes, so replacing - only strings longer than two characters cannot increase request bytes. - Iteration preserves the original JSON reader's supported nesting depth. - """ +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.""" + # Root discovery also derives initial navigation from real decision values. + if not select or list_fields and "" in select: + return payload + # Leave malformed requests to the typed owner, including error ordering. + if not all(isinstance(pointer, str) for pointer in select): + return payload + + paths: dict[str | None, Any] = {} + for pointer in select: + branch = paths + for token in pointer.split("/")[1:]: + branch = branch.setdefault(token, {}) + # None cannot collide with a JSON object key (including the empty key). + branch[None] = True + result: dict[str, Any] = {} - pending: list[tuple[Any, Any]] = [(payload, result)] + pending: list[tuple[Any, Any, dict[str | None, Any], int]] = [(payload, result, paths, 0)] while pending: - source, target = pending.pop() + source, target, branch, depth = pending.pop() + if list_fields and None in branch: + depth = max(depth, list_depth or 1) + 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") + if depth <= 0 and token not in branch: + continue if isinstance(value, (dict, list)): + subtree = branch.get(token, {}) + # Length matters when validating array indices or listing children. child: Any = {} if isinstance(value, dict) else [None] * len(value) target[key] = child - pending.append((value, child)) + pending.append((value, child, subtree, depth - 1)) else: - target[key] = None if isinstance(value, str) and len(value) > 2 else value + # Every scalar has the same address behavior. Keep original scalar + # types, precision, and byte counts at the file rendering boundary. + target[key] = None return result @@ -509,7 +529,8 @@ def context_selection_view(payload: dict[str, Any], *, select: list[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 or list_fields else context_selection_transport(payload) + 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 {} diff --git a/tests/control_plane/test_context_read_reference.py b/tests/control_plane/test_context_read_reference.py index 5c5e5f43cc..9086d8327f 100644 --- a/tests/control_plane/test_context_read_reference.py +++ b/tests/control_plane/test_context_read_reference.py @@ -32,6 +32,66 @@ def fixture(tmp_path): return payload, view, model_context_reference(view) +@pytest.mark.parametrize("listing", [False, True]) +@pytest.mark.parametrize("selectors", [ + [""], ["/a~1b~0c/0/number"], ["/a~1b~0c", "/a~1b~0c/0/number"], + ["/", "/0/01"], ["/scalar/child"], ["/null/child"], ["/empty/child"], + ["/a~1b~0c/3"], ["/a~1b~0c/01"], ["/a~1b~0c/-1"], + ["/a~1b~0c/9007199254740992"], ["/a~1b~0c/0/absent/child"], + ["/a~1b~0c/0/number", "/missing"], ["/invalid~2"], ["invalid"], +]) +def test_selected_structure_preserves_original_reader_and_recovery(tmp_path, monkeypatch, selectors, listing): + from loopx.control_plane.turn_driver import decision_capture + + payload = {"a/b~c": [{"number": 10**45 + 17}, {"tail": "完整原值"}], + "": "empty-key", "0": {"01": -0.0}, "scalar": "source text", "null": None, + "empty": {}, "unread": {f"sibling_{index}": {"value": index} for index in range(200)}} + source = tmp_path / "source.json" + original = (json.dumps(payload, ensure_ascii=False) + "\n").encode() + source.write_bytes(original) + source.chmod(0o600) + command = shlex.join(["loopx", "--format", "json", "quota", "context-read", "--capture-file", str(source), + "--capture-sha256", hashlib.sha256(original).hexdigest(), "--page-bytes", "64000", + "--max-page-output-bytes", "64000"]) + # The unpruned transport independently exercises the original typed owner, + # including its failed-batch precedence and nearest-container remediation. + with monkeypatch.context() as old: + old.setattr(decision_capture, "context_selection_transport", lambda value, *args, **kwargs: value) + expected = read(command, select=selectors, list_fields=listing, list_depth=2 if listing else None) + observed = read(command, select=selectors, list_fields=listing, list_depth=2 if listing else None) + assert observed == expected + assert source.read_bytes() == original + if not listing and selectors == ["/a~1b~0c/0/number"]: + assert json.loads(observed["content"])["selections"][0]["value"] == 10**45 + 17 + if not observed["ok"] and "recovery_command" in observed: + assert read(observed["recovery_command"])["ok"], "Recovery must still discover the complete original parent" + + +def test_scoped_discovery_keeps_unrelated_values_out_of_the_runtime_request(tmp_path, monkeypatch): + from loopx.control_plane.quota import turn_envelope + + sizes = [] + evaluate = turn_envelope.effect_runtime_projection_result + + def measure(method, request): + if request.get("operation") == "context_select": + sizes.append(len(json.dumps(request, ensure_ascii=False).encode())) + return evaluate(method, request) + + monkeypatch.setattr(turn_envelope, "effect_runtime_projection_result", measure) + template = b'{"selected":{"answer":42},"unrelated":""}' + raw = template[:-2] + b"x" * 100_000 + template[-2:] + source = tmp_path / "decision.json" + source.write_bytes(raw) + command = shlex.join(["loopx", "--format", "json", "quota", "context-read", + "--capture-file", str(source), "--capture-sha256", hashlib.sha256(raw).hexdigest()]) + page = read(command, select=["/selected"], list_fields=True) + assert page["ok"] and page["eof"], page + assert sizes and sum(sizes) < len(raw) // 100 + assert json.loads(page["content"])["selections"] == [{"pointer": "/selected", "fields": [ + {"pointer": "/selected/answer", "type": "number", "bytes": 2}]}] + + def assert_section_read_only(page, machine): raw = Path(machine["path"]).read_bytes() reference = {"path": machine["path"], "sha256": machine["sha256"]}