diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index e6f0ffeed3..670430896f 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -139,6 +139,29 @@ 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. + """ + result: dict[str, Any] = {} + pending: list[tuple[Any, Any]] = [(payload, result)] + while pending: + source, target = pending.pop() + for key, value in source.items() if isinstance(source, dict) else enumerate(source): + if isinstance(value, (dict, list)): + child: Any = {} if isinstance(value, dict) else [None] * len(value) + target[key] = child + pending.append((value, child)) + else: + target[key] = None if isinstance(value, str) and len(value) > 2 else value + return result + + 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]: @@ -233,7 +256,8 @@ 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} - plan = _turn_envelope_result("context_select", context=payload, list_fields=list_fields, + planning_context = payload if initial_read or list_fields else _context_selection_transport(payload) + plan = _turn_envelope_result("context_select", context=planning_context, list_fields=list_fields, recover_selection_errors=not initial_read, **selection_options, **depth_options) if rejection := plan.get("rejected_selection"): # TS alone resolves the nearest container. Reuse the original diff --git a/tests/control_plane/test_context_selection_transport.py b/tests/control_plane/test_context_selection_transport.py new file mode 100644 index 0000000000..8825f3c6f0 --- /dev/null +++ b/tests/control_plane/test_context_selection_transport.py @@ -0,0 +1,99 @@ +"""Selection planning may omit scalar prose; captured values and navigation stay exact.""" +import argparse +import hashlib +import json + +import pytest + +from loopx.cli_commands.quota_capture import read_context_capture +from loopx.control_plane.quota import turn_envelope + + +def captured_reader(tmp_path, value, **options): + raw = json.dumps(value, ensure_ascii=False).encode() + path = tmp_path / "capture.json" + path.write_bytes(raw) + args = argparse.Namespace(capture_file=str(path), capture_sha256=hashlib.sha256(raw).hexdigest(), + offset=0, page_bytes=64000, max_page_output_bytes=64000, format="json", + select=options.pop("select", None), **options) + return path, raw, args + + +@pytest.fixture +def transported_contexts(monkeypatch): + observed = [] + original = turn_envelope._turn_envelope_result + + def record(operation, **request): + if operation == "context_select": + observed.append(request["context"]) + return original(operation, **request) + + monkeypatch.setattr(turn_envelope, "_turn_envelope_result", record) + return observed + + +def test_explicit_reader_transports_less_prose_and_returns_original_values(tmp_path, transported_contexts): + text = "Keep the complete final stop. 保留末尾约束🙂 " * 90 + value = {"rows": [{"text": text, "small": "ab", "empty": {}, "array": []}], + "a/b": {"~": None}, "numbers": [0, False, None, -0.0, 9007199254740993], + "optional": "Unselected historical supporting material. " * 200} + path, raw, args = captured_reader(tmp_path, value, select=["/rows/0", "/a~1b/~0", "/numbers"]) + page = read_context_capture(args) + assert page["ok"] and page["eof"], page + selected = {row["pointer"]: row["value"] for row in json.loads(page["content"])["selections"]} + assert selected == {"/rows/0": value["rows"][0], "/a~1b/~0": None, "/numbers": value["numbers"]} + assert "-0.0" in page["content"] and "9007199254740993" in page["content"] + assert path.read_bytes() == raw + assert len(transported_contexts) == 1 + transported = transported_contexts[0] + assert len(json.dumps(transported, ensure_ascii=False).encode()) < len(raw) // 2 + assert transported["rows"][0] == {"text": None, "small": "ab", "empty": {}, "array": []} + assert transported["numbers"] == value["numbers"] + assert text not in json.dumps(transported, ensure_ascii=False) + + +@pytest.mark.parametrize("mode", ["initial", "root_index", "nested_index"]) +def test_navigation_keeps_full_scalar_input(tmp_path, transported_contexts, mode): + instruction = "Preserve all current requirements and stop before publication. " * 10 + value = {"ok": True, "goal_id": "synthetic", + "interaction_contract": {"schema_version": "loopx_interaction_contract_v0", + "agent_channel": {"work_context": {"instruction": instruction, "sources": []}}}} + options = {"initial_read": True} if mode == "initial" else { + "list_fields": True, "select": [""] if mode == "root_index" else ["/interaction_contract"]} + path, raw, args = captured_reader(tmp_path, value, **options) + page = read_context_capture(args) + assert page["ok"] and page["eof"], page + assert transported_contexts == [value] + assert path.read_bytes() == raw + if mode == "initial": + assert instruction in page["content"] + + +def test_short_scalar_arrays_do_not_grow_the_planning_payload(tmp_path, transported_contexts): + value = {"values": [0, 1, False, True, None, "", "a", "ab"] * 60} + _, _, args = captured_reader(tmp_path, value, select=["/values"]) + page = read_context_capture(args) + assert page["ok"] and page["eof"], page + assert transported_contexts == [value] + assert json.loads(page["content"])["selections"][0]["value"] == value["values"] + + +def test_deep_selection_keeps_the_existing_json_reader_depth(tmp_path, transported_contexts): + text = "Preserve the original deep tail constraint. " * 20 + value = text + for _ in range(650): + value = {"v": value} + _, _, args = captured_reader(tmp_path, value, select=["/v" * 650]) + page = read_context_capture(args) + assert page["ok"] and page["eof"], page + assert json.loads(page["content"])["selections"][0]["value"] == text + assert len(transported_contexts) == 1 + + +def test_full_reader_stays_independent_of_selection_runtime(tmp_path, transported_contexts): + value = {"stop": "Keep the complete original document.", "unknown": {"enabled": False}} + _, raw, args = captured_reader(tmp_path, value) + page = read_context_capture(args) + assert page["ok"] and page["eof"] and page["content"].encode() == raw + assert transported_contexts == []