fix(quota): lead saved context responses with the returned page

docs/reference/required-work-context.md

@@ -256,14 +256,20 @@ For a saved benchmark preflight, the ordinary wake prompt carries the command fo
 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.
+navigation attached. Forward the whole response within the tool's output
+budget and read its top-level `content`. When `eof` is false, follow the
+top-level `next_command` for remaining bytes. Use
+`navigation.first_read_command`, when present, to recover missing initial
+context, and `navigation.recovery.command` for other fields. A smaller
+`--max-page-output-bytes` bounds the combined response for tighter tool limits.
+If the response is lost, use the original digest-bound capture in the wake
+prompt. Read recovery reuses the saved decision and its original receipt.
 
 中文：对于已保存的 benchmark 准入结果，唤醒正文直接给出首读命令。一次调用即可
-取得首屏；已知字段可改用多个 `--select`。续页和补读都沿原文件读取。如果没拿到
-任何输出，就用唤醒正文中的原文件及其摘要恢复，无须重做投影或准入。
+取得首屏，已知字段可改用多个 `--select`。在工具输出预算内转发完整结果并阅读顶层
+`content`，`eof` 为 false 时按顶层 `next_command` 续读。`navigation.first_read_command`
+用于补读缺失的首读上下文，其他字段沿 `navigation.recovery.command` 读取。工具预算较小时可
+降低 `--max-page-output-bytes`。结果丢失时，用唤醒正文中的原文件及其摘要恢复，复用原决策和回执。
 
 When the existing acceptance owner supplies complete `outcome_context`
 navigation matching this capture's Goal, Agent, selected Todo, outcome revision

loopx/cli_commands/quota_capture.py

@@ -377,6 +377,7 @@ def _native_context_reference(projected: dict[str, Any], reference: dict[str, An
                               initial_read_pointers: list[str] | None,
                               initial_read_reference: Path | None,
                               page_bytes: int, output_bytes: int,
+                              page_attached: bool = False,
                               reader_argv: list[str] | None = None) -> dict[str, Any]:
     """Build the bounded native reference without publishing or registering it."""
     target = projected.get("target") if projected.get("schema_version") == "context_delivery_v1" else None
@@ -418,8 +419,11 @@ def fits() -> bool:
         if any(pointer.endswith("/vision_authoring/rule") for pointer in initial_read_pointers):
             instruction = ("Before composing a Vision, select the complete vision_authoring object "
                 "at the displayed step in this capture. " + instruction)
-        result["instruction"] = ("Start with first_read_command for current facts and source navigation from this capture, "
-            "without reading an older baseline. "
+        first_read = ("Read the top-level content and follow next_command if eof is false. "
+            "Use first_read_command only to recover missing initial context from this capture"
+            if page_attached else
+            "Start with first_read_command for current facts and source navigation from this capture")
+        result["instruction"] = (first_read + ", without reading an older baseline. "
             "Read each complete optional attachment from its captured step before using it. " + instruction)
         result["first_read_command"] = context_capture_read_command(current_path, current["sha256"],
             page_bytes=page_bytes, max_page_output_bytes=output_bytes, initial_read=True, reader_argv=reader_argv)
@@ -494,6 +498,7 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
                                 captured_recovery: dict[str, Any] | None = None,
                                 metadata: dict[str, Any] | None = None,
                                 max_page_output_bytes: int | None = None,
+                                page_attached: bool = False,
                                 reader_argv: list[str] | None = None,
                                 initial_read_plan: dict[str, Any] | None = None) -> dict[str, Any]:
     """Publish the cheaper readable native transport for an already-decided view.
@@ -576,7 +581,7 @@ def view(kind: str, ref: dict[str, Any], body: bytes) -> dict[str, Any]:
             field_index=field_index, initial_read_pointers=initial_read_pointers,
             progressive_enabled=progressive_enabled,
             initial_read_reference=initial_read_reference,
-            page_bytes=page_bytes, output_bytes=output_bytes, reader_argv=reader_argv)
+            page_bytes=page_bytes, output_bytes=output_bytes, page_attached=page_attached, reader_argv=reader_argv)
 
     if typed and mode == "delta":
         result = None
@@ -698,7 +703,10 @@ def _saved_context_read_page(args: argparse.Namespace, navigation: dict[str, Any
         "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:
+    if page["ok"]:
+        # Keep the already-budgeted page and cursor ahead of recovery metadata.
+        page["navigation"] = page.pop("navigation")
+    elif "recovery_command" not in page:
         page["recovery_command"] = recovery["command"]
     return page
 
@@ -814,7 +822,8 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
             progressive_enabled=activation["progressive_enabled"], initial_read_plan=prepared.initial_read_plan,
             metadata={"recovery": full_reference, "capture_sha256": args.capture_sha256,
                 "context_delivery": activation},
-            max_page_output_bytes=getattr(args, "max_page_output_bytes", None), reader_argv=reader_argv)
+            max_page_output_bytes=getattr(args, "max_page_output_bytes", None), page_attached=saved_page,
+            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")

tests/control_plane/test_saved_context_project.py

@@ -1087,6 +1087,98 @@ def present(turn):
     assert not registry.exists() and not runtime.exists()
 
 
+@pytest.mark.parametrize("repeat", [1, 1200], ids=["eof", "multipage"])
+def test_saved_initial_page_is_primary_and_its_recovery_replays_the_same_selection(tmp_path, monkeypatch, repeat):
+    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")
+    codex_context_hook(str(context), {"hook_event_name": "SessionStart", "source": "startup",
+        "session_id": "session", "cwd": str(tmp_path)})
+    receipt = {"turn_instance_id": "t"}
+    text = "Keep the original requirement. 保留原要求🙂\n" * repeat
+    original = tmp_path / "original.json"
+    original.write_text(json.dumps({"ok": True, "goal_id": "g", "agent_identity": {"agent_id": "a"},
+        "heartbeat_receipt": receipt, "effective_action": "normal_run",
+        "interaction_contract": {"schema_version": "loopx_interaction_contract_v0", "agent_channel": {
+            "work_context": {"sources": [{"source": "goal_state", "ordering": "before_work",
+                "content": {"text": text, "state_file": "goal.md"}}]}}}}, ensure_ascii=False))
+    raw = original.read_bytes()
+    digest = hashlib.sha256(raw).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)]
+    if repeat > 1:
+        project += ["--max-page-output-bytes", "6400"]
+
+    def cli(*args):
+        return _run_cli(registry, runtime, *args, cwd=tmp_path)
+
+    def follow(command):
+        words = shlex.split(command)
+        code, page = cli(*words[words.index("quota"):])
+        assert code == 0 and page["ok"], page
+        return page
+
+    code, reference = cli(*project)
+    assert code == 0 and "content" not in reference
+    assert reference["instruction"].startswith("Start with first_read_command")
+    code, first = cli(*project, "--initial-read")
+    assert code == 0 and first["offset"] == 0 and first["eof"] == (repeat == 1)
+    assert list(first).index("content") < list(first).index("navigation")
+    navigation = first["navigation"]
+    assert "top-level content" in navigation["instruction"]
+    assert "first_read_command only to recover" in navigation["instruction"]
+    assert "without reading an older baseline" in navigation["instruction"]
+    assert "Start with first_read_command" not in navigation["instruction"]
+    assert ("next_command" in first) == (repeat > 1)
+    if "next_command" in first:
+        assert list(first).index("next_command") < list(first).index("navigation")
+    limit = 6400 if repeat > 1 else 64000
+    assert len(json.dumps(first, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= limit
+
+    def finish(page):
+        body = bytearray()
+        while True:
+            assert page["offset"] == len(body)
+            body.extend(page["content"].encode())
+            if page["eof"]:
+                assert len(body) == page["total_bytes"]
+                return bytes(body)
+            assert page["next_offset"] == len(body)
+            page = follow(page["next_command"])
+
+    # A client that forwards only navigation discards the bounded first page.
+    # It can recover this same selection using the separate offset-zero recipe.
+    assert "content" not in navigation and "next_command" not in navigation
+    files = set(context.rglob("*"))
+    journal = (context / "session.json").read_bytes()
+    body = finish(first)
+    assert finish(follow(navigation["first_read_command"])) == body
+    selected = {item["pointer"]: item["value"] for item in json.loads(body)["selections"]}
+    assert selected["/heartbeat_receipt"] == receipt
+    assert selected["/effective_action"] == "normal_run"
+    assert selected["/interaction_contract/agent_channel/work_context/sources/0/content/text"] == text
+    assert original.read_bytes() == raw and (context / "session.json").read_bytes() == journal
+    assert set(context.rglob("*")) == files
+    if repeat == 1:
+        code, detail = cli(*project, "--select", "/heartbeat_receipt")
+        assert code == 0 and detail["eof"] and "next_command" not in detail
+        assert "first_read_command only to recover" in detail["navigation"]["instruction"]
+        assert json.loads(detail["content"])["selections"] == [{"pointer": "/heartbeat_receipt", "value": receipt}]
+        code, rejected = cli(*project, "--select", "/missing")
+        assert code and not rejected["ok"] and "recovery_command" in rejected
+        assert not ({"content", "next_command", "navigation"} & rejected.keys())
+        code, too_small = cli(*project, "--initial-read", "--max-page-output-bytes", "256")
+        assert code and not too_small["ok"] and not ({"content", "next_command", "navigation"} & too_small.keys())
+        assert too_small["heartbeat_receipt"] == receipt and too_small["capture_sha256"] == digest
+        recovered = Path(too_small["recovery"]["path"]).read_bytes()
+        assert hashlib.sha256(recovered).hexdigest() == too_small["recovery"]["sha256"]
+        assert json.loads(recovered) == json.loads(raw)
+    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