Reduce the runtime request for scoped capture reads and discovery

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 {}

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"]}