Reduce scalar transport for explicit context selections

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

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 == []