Read saved quota fields in the first native tool call

benchmark/runtime/context_delivery.py

@@ -26,13 +26,15 @@ def preflight_context_instruction(env: dict[str, str], path: Path, digest: str,
     source = ["--capture-file", str(path), "--capture-sha256", digest]
     project = [*prefix, "context-project", *source,
         "--goal-id", env["LOOPX_GOAL_ID"], "--agent-id", env["LOOPX_AGENT_ID"],
-        "--turn-instance-id", turn_id, "--context-delivery", "on", "--runtime-profile", "generic_cli", "--codex-context-dir", env["LOOPX_CODEX_CONTEXT_DIR"]]
+        "--turn-instance-id", turn_id, "--initial-read", "--context-delivery", "on", "--runtime-profile", "generic_cli", "--codex-context-dir", env["LOOPX_CODEX_CONTEXT_DIR"]]
     recovery_index = [*prefix, "context-read", *source, "--list"]
     return ("Present that same immutable packet with:\n```sh\n" + shlex.join(project)
         + "\n```\n" + ("" if guide_in_host else CONTEXT_CAPTURE_NAVIGATION_GUIDANCE)
         + "If the projection command itself fails, recover necessary facts from the same original capture. "
         "When pointers are unknown, discover them with:\n```sh\n" + shlex.join(recovery_index)
         + "\n```\nUse the same capture's generic reader and selectors for the facts the current decision needs. "
+        "For known fields, replace --initial-read with repeated --select pointers in the first command. "
+        "Continue the returned page or use navigation.recovery.command to reread; do not project again. "
         "Discovery does not require draining the packet. "
         "Neither presentation nor recovery renews this decision's authority or freshness. ")
 

docs/reference/required-work-context.md

@@ -252,6 +252,19 @@ evidence and the complete operation/settlement contract before depending on them
 The original SHA-bound full recovery remains available; a selected page's EOF
 does not prove complete source delivery, model retention or fresh authority.
 
+For a saved benchmark preflight, the ordinary wake prompt carries the command for
+the first page: native JSON `quota context-project --initial-read` publishes the
+saved decision and reads it in one call. Replace `--initial-read` with repeated
+`--select` for known fields. The output is a normal bounded reader page with its
+navigation attached. Follow `next_command` for remaining bytes or use
+`navigation.recovery.command` for other fields. If no output arrived, read the
+original digest-bound capture identified in the wake prompt. Neither retry
+requires another projection, guard, notification or model acknowledgement.
+
+中文：对于已保存的 benchmark 准入结果，唤醒正文直接给出首读命令。一次调用即可
+取得首屏；已知字段可改用多个 `--select`。续页和补读都沿原文件读取。如果没拿到
+任何输出，就用唤醒正文中的原文件及其摘要恢复，无须重做投影或准入。
+
 When the existing acceptance owner supplies complete `outcome_context`
 navigation matching this capture's Goal, Agent, selected Todo, outcome revision
 and provider revision, the initial read expands its suggested outcomes in full

loopx/cli_commands/quota_capture.py

@@ -661,6 +661,22 @@ def progressive_context_capture_view(payload: dict[str, Any], args: argparse.Nam
     return readable if len(_native_output_bytes(readable)) < len(_native_output_bytes(required)) else required
 
 
+def _saved_context_read_page(args: argparse.Namespace, navigation: dict[str, Any], *,
+                             initial_reference: str | None, reader_argv: list[str]) -> dict[str, Any]:
+    """Use the ordinary verified reader; the first page and navigation share its output budget."""
+    recovery = navigation["recovery"]
+    pinned = initial_reference if args.initial_read else None
+    reference = pinned or recovery.get("reference_file")
+    read_args = argparse.Namespace(**{**vars(args), "quota_command": "context-read",
+        "reference_file": reference, "capture_file": None if reference else recovery["path"],
+        "capture_sha256": None if reference else recovery["sha256"],
+        "codex_context_dir": None, "codex_context_root": None, "initial_read": bool(args.initial_read and not pinned)})
+    page = read_context_capture(read_args, presentation={"navigation": navigation}, reader_argv=reader_argv)
+    if not page["ok"] and "recovery_command" not in page:
+        page["recovery_command"] = recovery["command"]
+    return page
+
+
 def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str | None = None) -> dict[str, Any]:
     """Present an immutable already-decided quota capture, without authority IO.
 
@@ -676,7 +692,14 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
     reader_argv = current_loopx_cli_argv()
     try:
         activation = native_context_activation(args)
-        path, raw, payload = _read_capture(args.capture_file, args.capture_sha256, private=hook_session is not None)
+        saved_page = bool(getattr(args, "initial_read", False) or getattr(args, "select", None))
+        if saved_page and (args.format != "json" or not activation["enabled"] or args.offset != 0
+                or args.reference_file or args.selection_sha256 or args.page_bytes is not None
+                or (args.initial_read and args.select)):
+            raise ValueError("Use an enabled native JSON context-project with either --initial-read or --select "
+                "at offset zero. Use context-read for replay or continuation.")
+        path, raw, payload = _read_capture(args.capture_file, args.capture_sha256,
+                                           private=hook_session is not None)
         identity = payload.get("agent_identity")
         receipt = payload.get("heartbeat_receipt")
         if (payload.get("ok") is not True or not isinstance(identity, dict) or not isinstance(receipt, dict)
@@ -699,6 +722,7 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
     target = None
     native_prepared = False
     native_view = None
+    initial_reference = None
     if not activation["enabled"] and activation["progressive_enabled"]:
         # Recovery is presentation of this exact saved decision, including its
         # original Turn. It does not need a supported host or a native session.
@@ -765,14 +789,15 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
             max_page_output_bytes=getattr(args, "max_page_output_bytes", None), reader_argv=reader_argv)
         reference = {key: native_view[key] for key in ("path", "sha256", "command")}
         mode, target = native_view["mode"], native_view.get("target")
+        initial_reference = native_view.get("first_read_reference")
         native_view = model_context_reference(native_view, reader_argv=reader_argv)
     except (QuotaDecisionTransportError, OSError, ValueError, TypeError, RecursionError) as exc:
         if native_view is not None and not isinstance(exc, QuotaDecisionTransportError):
             # The native view and journal are already valid. Publishing the
             # shorter reader metadata is optional; retain their exact legacy
             # recovery rather than losing the original capture or admitting again.
-            return native_view
-        if isinstance(exc, QuotaDecisionTransportError) or (activation["enabled"] and not native_prepared):
+            pass
+        elif isinstance(exc, QuotaDecisionTransportError) or (activation["enabled"] and not native_prepared):
             if not isinstance(exc, QuotaDecisionTransportError):
                 exc = QuotaDecisionTransportError(
                     "Current context recovery could not be prepared. Read the saved "
@@ -805,9 +830,10 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
         **({"context_delivery": activation} if activation["enabled"] else {}),
         "instruction": "Presentation of the original decision only; authority and freshness are unchanged. "
             + (CONTEXT_CAPTURE_NAVIGATION_GUIDANCE if activation["enabled"] else CONTEXT_CAPTURE_READ_GUIDANCE)}
-    if native_view is not None:
-        return native_view
-    return result
+    navigation = native_view or result
+    if saved_page:
+        return _saved_context_read_page(args, navigation, initial_reference=initial_reference, reader_argv=reader_argv)
+    return navigation
 
 
 def prepare_quota_capture(args: argparse.Namespace, directory: Path | None) -> Path | None:
@@ -1089,8 +1115,8 @@ def handoff_decision(directory: Path | None, socket_path: str | None,
 def handle_native_context_command(args: argparse.Namespace, *, print_payload: Callable[..., None],
                                   registry_path: Path | None = None, runtime_root_arg: str | None = None) -> int | None:
     """Dispatch native presentation/lifecycle IO before any quota authority read."""
-    if getattr(args, "initial_read", False) and args.quota_command != "context-read":
-        print_payload({"ok": False, "reason": "--initial-read is only supported by context-read"}, args.format, str)
+    if getattr(args, "initial_read", False) and args.quota_command not in ("context-read", "context-project"):
+        print_payload({"ok": False, "reason": "--initial-read is only supported by context-read or enabled native JSON context-project"}, args.format, str)
         return 1
     list_depth = getattr(args, "list_depth", None)
     if list_depth is not None and (args.quota_command != "context-read" or not getattr(args, "list_fields", False)

loopx/cli_commands/quota_registration.py

@@ -239,10 +239,10 @@ def register_quota_command(
              "context-config renders native configuration for the exact Goal/Agent; no host configuration, trust, guard or spend are changed.")
     quota_parser.add_argument("--capture-sha256", help="For context-read/context-project: exact SHA-256 supplied with the capture reference.")
     quota_parser.add_argument("--initial-read", action="store_true",
-        help="For JSON context-read: read current controls, task text and source navigation from a recognized capture. "
+        help="For JSON context-read or enabled native context-project: read current controls, task text and source navigation from a recognized capture. "
              "Detailed criteria and operation contracts remain required before dependent work. Start at offset zero; follow next_command for continuation.")
     quota_parser.add_argument("--select", action="append",
-        help="For context-read: repeat an exact RFC 6901 JSON pointer to read a batch from the original capture. "
+        help="For context-read or enabled native JSON context-project: repeat an exact RFC 6901 JSON pointer to read a batch from the original capture. "
              "Selected detail pages never acknowledge full-context retention; omit for unchanged full capture reads.")
     quota_parser.add_argument("--list", dest="list_fields", action="store_true",
         help="For context-read: list immediate fields at the selected pointers (root by default), "

tests/control_plane/test_saved_context_project.py

@@ -407,6 +407,7 @@ def test_preflight_instruction_pins_both_projection_and_full_fallback(tmp_path):
         assert argv[argv.index("--capture-sha256") + 1] == "a" * 64
         assert "should-run" not in argv and "spend-slot" not in argv
     assert commands[0][commands[0].index("--turn-instance-id") + 1] == "original-turn"
+    assert "--initial-read" in commands[0]
     assert "--decision-output-root" not in commands[0]
 
 
@@ -490,9 +491,9 @@ def consume(wire):
     if enabled:
         command.extend(("--max-page-output-bytes", "32000"))
         rc, cold = _run_cli(registry, runtime, *command[command.index("quota"):], cwd=project)
-        assert rc == 0 and cold["mode"] == "full", cold
+        assert rc == 0 and cold["navigation"]["mode"] == "full" and cold["offset"] == 0, cold
         assert len(json.dumps(cold, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 32000
-        presented = consume(cold)
+        presented = consume(cold["navigation"])
         if compacted:
             codex_context_hook(str(native), {**event, "hook_event_name": "PreCompact"})
     else:
@@ -512,7 +513,7 @@ def consume(wire):
         assert len(json.dumps(bound, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 32000
         delivered = consume(bound)
         if not compacted:
-            assert delivered["base"]["capture_id"] == machine_reference_oracle(cold)["target"]["capture_id"]
+            assert delivered["base"]["capture_id"] == machine_reference_oracle(cold["navigation"])["target"]["capture_id"]
     else:
         assert bound["mode"] == "should-run" and not list(native.iterdir())
     assert _heartbeat_receipt_count(runtime, turn) == 2 and _spend_run_count(runtime) == 0
@@ -670,6 +671,10 @@ def response(request, number, message):
                     sources.append((source, source.read_bytes()))
                     receipts = _heartbeat_receipt_count(runtime, actual_turn)
                     assert hashlib.sha256(source.read_bytes()).hexdigest() == argv[argv.index("--capture-sha256") + 1]
+                    assert "--initial-read" in argv
+                    if progressive:
+                        command = shlex.join([word for word in argv if word != "--initial-read"] + [
+                            "--select", "/goal_id", "--select", "/heartbeat_receipt"])
                     command += " --max-page-output-bytes 32000"
                 else:
                     source, raw = sources[0 if phase == "cold" else 1]
@@ -686,6 +691,15 @@ def response(request, number, message):
                 if isinstance(output, list):
                     output = [part["text"] for part in output if part.get("type") == "input_text"][-1]
                 navigation = json.loads(output.split("\nOutput:\n", 1)[-1])
+                if benchmark and name in ("cold", "warm"):
+                    page = navigation
+                    assert page["ok"] and len(json.dumps(page, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 32000
+                    assert page["offset"] == 0 and page["content"]
+                    navigation = page["navigation"]
+                    if progressive:
+                        assert page["eof"]
+                        values = {row["pointer"]: row["value"] for row in json.loads(page["content"])["selections"]}
+                        assert values["/goal_id"] == GOAL_ID and values["/heartbeat_receipt"]["turn_instance_id"] == actual_turn
                 descriptor = machine_reference_oracle(navigation)
                 # A selectively read large capture cannot become a retained
                 # full baseline; its next projection must remain full.
@@ -739,6 +753,8 @@ def response(request, number, message):
                         argv[index] = str(damaged)
                         queue.insert(0, shlex.join(argv))
                         rejected_read = True
+                if benchmark and progressive and name in ("cold", "warm"):
+                    queue.clear()  # The first tool result already contains the requested facts.
                 phase = name + "_read"
                 if not queue:
                     return response(request, number, message)
@@ -1065,3 +1081,55 @@ def present(turn):
     codex_context_hook(str(journal), {**event, "source": "compact"})
     assert present("warm")["mode"] == "full"
     assert not registry.exists() and not runtime.exists()
+
+
+def test_saved_page_continues_and_replays_after_compaction_without_republication(tmp_path, monkeypatch):
+    from concurrent.futures import ThreadPoolExecutor
+    from loopx.control_plane.turn_driver.codex_context import codex_context_hook
+
+    runtime, registry, context = tmp_path / "runtime", tmp_path / "registry.json", tmp_path / "native"
+    context.mkdir(mode=0o700)
+    monkeypatch.setenv("CODEX_THREAD_ID", "session")
+    monkeypatch.chdir(tmp_path)
+    event = {"source": "startup", "session_id": "session", "cwd": str(tmp_path)}
+    codex_context_hook(str(context), {**event, "hook_event_name": "SessionStart"})
+    value = "scope and stops: 保留原要求。" * 1200
+    original = tmp_path / "original.json"
+    original.write_text(json.dumps({"ok": True, "goal_id": "g", "agent_identity": {"agent_id": "a"},
+        "heartbeat_receipt": {"turn_instance_id": "t"}, "instructions": value}, ensure_ascii=False))
+    digest = hashlib.sha256(original.read_bytes()).hexdigest()
+    project = ["quota", "context-project", "--capture-file", str(original), "--capture-sha256", digest,
+        "--goal-id", "g", "--agent-id", "a", "--turn-instance-id", "t", "--codex-app",
+        "--context-delivery", "on", "--codex-context-dir", str(context), "--max-page-output-bytes", "6400"]
+    def cli(*args):
+        return _run_cli(registry, runtime, *args, cwd=tmp_path)
+    initial_files = set(context.rglob("*"))
+    for invalid in (["--initial-read", "--select", "/instructions"], ["--offset", "1", "--select", "/instructions"],
+                    ["--context-delivery", "off", "--initial-read"], ["--goal-id", "foreign", "--initial-read"]):
+        code, error = cli(*project, *invalid)
+        assert code and not error["ok"] and "content" not in error
+    assert set(context.rglob("*")) == initial_files
+    code, first = cli(*project, "--select", "/instructions")
+    assert code == 0 and first["offset"] == 0 and not first["eof"]
+    assert first["navigation"]["mode"] == "full"
+    codex_context_hook(str(context), {**event, "hook_event_name": "PreCompact"})
+    journal = (context / "session.json").read_bytes()
+    files = set(context.rglob("*"))
+    def read_reference():
+        ref = first["reference_file"]
+        result = bytearray()
+        args = ["quota", "context-read", "--reference-file", ref, "--max-page-output-bytes", "6400"]
+        while True:
+            code, page = cli(*args)
+            assert code == 0 and page["offset"] == len(result)
+            result.extend(page["content"].encode())
+            if page["eof"]:
+                assert len(result) == page["total_bytes"]
+                return json.loads(result)["selections"][0]["value"]
+            words = shlex.split(page["next_command"])
+            args = words[words.index("quota"):]
+    with ThreadPoolExecutor(2) as pool:
+        assert list(pool.map(lambda _: read_reference(), range(2))) == [value, value]
+    assert (context / "session.json").read_bytes() == journal and set(context.rglob("*")) == files
+    assert hashlib.sha256(original.read_bytes()).hexdigest() == digest
+    assert not registry.exists() and not runtime.exists()