diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md
index 024462d427..73966d4bf4 100644
--- a/docs/reference/required-work-context.md
+++ b/docs/reference/required-work-context.md
@@ -263,24 +263,49 @@ 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.
 
+The generated `first_read_command` and its paging cursor default to a 12,000-byte
+serialized UTF-8 output limit when the caller supplied no tool budget. This
+applies after explicit Todo selection as well. An explicit output limit still
+controls the opening, and the ordinary bulk and recovery commands keep their
+existing budgets.
+
+中文：调用方未指定工具预算时，生成的 `first_read_command` 与续页默认每次最多
+返回 12,000 个序列化 UTF-8 字节。显式选择 Todo 后的首读也适用。调用方可以显式
+指定首读预算，完整读取与恢复命令继续使用原有预算。
+
 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. 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.
+`--select` for known fields. The opening defaults to 12,000 serialized UTF-8
+bytes including navigation. With a known tool and retained-history budget,
+set `--max-page-output-bytes` to a compatible byte limit, up to 64,000. A byte
+limit is not a model-token or retention guarantee. An optional `--page-bytes`
+also bounds UTF-8 content on the attached page and its cursor, using the same
+window rules as `context-read`. The output budget still applies to the complete
+serialized response.
+
+Read the top-level `content` and follow its `next_command` until that selection's
+`eof`. If the whole initial selection would need paging and its decision and
+read obligations fit in one page, the opening contains those controls and
+`navigation.remaining_read_command` points to the rest. Read it and follow
+its cursor before work. Both selections pin the same immutable capture.
+`navigation.first_read_command` replays the opening. Use
+`navigation.recovery.command` for other fields. If the response is lost, use the
+original digest-bound capture in the wake prompt. Recovery reuses the saved
+decision and its original receipt.
 
 中文：对于已保存的 benchmark 准入结果，唤醒正文直接给出首读命令。一次调用即可
-取得首屏，已知字段可改用多个 `--select`。在工具输出预算内转发完整结果并阅读顶层
-`content`，`eof` 为 false 时按顶层 `next_command` 续读。`navigation.first_read_command`
-用于补读缺失的首读上下文，其他字段沿 `navigation.recovery.command` 读取。工具预算较小时可
-降低 `--max-page-output-bytes`。结果丢失时，用唤醒正文中的原文件及其摘要恢复，复用原决策和回执。
+取得首屏，已知字段可改用多个 `--select`。首读默认最多 12,000 个 UTF-8 字节，含导航。
+已知工具直出与历史保留上限时，可用 `--max-page-output-bytes` 指定匹配的字节预算，最高
+64,000。字节数不能保证模型实际收到的 token 数或历史保留。附带页及续读也可用
+`--page-bytes` 限制正文的 UTF-8 字节数，沿用 `context-read` 的窗口规则。
+输出预算仍约束整个序列化响应。
+
+阅读顶层 `content`，按 `next_command` 续读到当前选区的 `eof`。首读内容需要分页且决策
+和读取义务能完整放在一页时，首屏先交付这些控制信息，并给出
+`navigation.remaining_read_command`。工作前沿这条命令及其游标读完余下的首读内容。
+两个选区绑定同一份不可变捕获。`navigation.first_read_command` 可重读首屏，其他字段沿
+`navigation.recovery.command` 读取。结果丢失时，用唤醒正文中的原文件及摘要恢复，复用原决策和回执。
 
 When the existing acceptance owner supplies complete `outcome_context`
 navigation matching this capture's Goal, Agent, selected Todo, outcome revision
diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index 81e4b32365..2350050a5e 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -27,7 +27,8 @@
     read_context_read_reference, write_context_read_reference, private_capture_uid, read_turn_observation_text,
 )
 from ..control_plane.turn_driver.native_context_profile import (
-    NATIVE_CONTEXT_PAGE_BYTES, NATIVE_CONTEXT_PAGE_OUTPUT_BYTES, native_context_page_budget,
+    NATIVE_CONTEXT_FIRST_READ_FALLBACK_BYTES, NATIVE_CONTEXT_PAGE_BYTES, NATIVE_CONTEXT_PAGE_OUTPUT_BYTES,
+    native_context_page_budget,
 )
 
 
@@ -133,6 +134,15 @@ def fits(result: dict[str, Any]) -> bool:
 
 
 
+def _validate_context_page_window(offset: int, size: int, output_limit: int | None) -> None:
+    from ..control_plane.effect_runtime import MAX_REQUEST_BYTES
+
+    maximum = max(8192, min(output_limit or 8192, MAX_REQUEST_BYTES))
+    if type(offset) is not int or offset < 0 or type(size) is not int or not 256 <= size <= maximum:
+        raise ValueError(f"Page offset must be nonnegative and --page-bytes must be between 256 and {maximum}; "
+            "larger pages require a sufficient --max-page-output-bytes within the capture limit.")
+
+
 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,
@@ -182,14 +192,7 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
                 goal_id=getattr(args, "goal_id", None), agent_id=getattr(args, "agent_id", None))
         if not isinstance(digest, str) or BARE_SHA256_PATTERN.fullmatch(digest) is None or not value:
             raise ValueError("context-read requires the original capture file and its exact SHA-256")
-        from ..control_plane.effect_runtime import MAX_REQUEST_BYTES
-
-        # Default JSON reads share the native large-page profile. Explicit
-        # smaller budgets can recover a truncated range of this same capture.
-        maximum = max(8192, min(output_limit or 8192, MAX_REQUEST_BYTES))
-        if type(offset) is not int or offset < 0 or type(size) is not int or not 256 <= size <= maximum:
-            raise ValueError("context-read requires a nonnegative byte offset and page-bytes between 256 and 8192; "
-                "larger pages require a sufficient max-page-output-bytes within the capture limit")
+        _validate_context_page_window(offset, size, output_limit)
         path, raw, payload = _read_capture(value, digest)
         list_fields = getattr(args, "list_fields", False)
         list_depth = getattr(args, "list_depth", None)
@@ -331,15 +334,16 @@ def model_context_reference(view: dict[str, Any], *, reader_argv: list[str] | No
     The descriptor holds complete digests locally. Legacy explicit readers and
     full-page observation receipts retain their original transport shape.
     """
-    page_bytes, output_bytes = native_context_page_budget()
-    # Retain an explicitly smaller caller budget from the generated command.
-    words = shlex.split(view["command"])
-    for option in ("--page-bytes", "--max-page-output-bytes"):
-        if option in words:
-            if option == "--page-bytes":
-                page_bytes = int(words[words.index(option) + 1])
-            else:
-                output_bytes = int(words[words.index(option) + 1])
+    def command_budget(command: str) -> tuple[int, int]:
+        words = shlex.split(command)
+        page_bytes, output_bytes = native_context_page_budget()
+        if "--page-bytes" in words:
+            page_bytes = int(words[words.index("--page-bytes") + 1])
+        if "--max-page-output-bytes" in words:
+            output_bytes = int(words[words.index("--max-page-output-bytes") + 1])
+        return page_bytes, output_bytes
+
+    page_bytes, output_bytes = command_budget(view["command"])
 
     def reference(value: dict[str, Any], *, navigation: bool = False) -> dict[str, str]:
         path = Path(value["path"])
@@ -358,15 +362,16 @@ def reference(value: dict[str, Any], *, navigation: bool = False) -> dict[str, s
         result["continuity_basis"] = view["continuity_basis"]
     if "first_read_command" in view:
         recovery = view["recovery"]
+        first_page_bytes, first_output_bytes = command_budget(view["first_read_command"])
         if view.get("first_read_reference"):
             # The producer already pinned this same initial selection. Keep
             # that existing cursor instead of asking the reader to replan it.
             result["first_read_command"] = context_capture_read_command(Path(recovery["path"]), recovery["sha256"],
-                page_bytes=page_bytes, max_page_output_bytes=output_bytes,
+                page_bytes=first_page_bytes, max_page_output_bytes=first_output_bytes,
                 reference_file=Path(view["first_read_reference"]), reader_argv=reader_argv)
         else:
             result["first_read_command"] = context_capture_read_command(Path(recovery["path"]), recovery["sha256"],
-                page_bytes=page_bytes, max_page_output_bytes=output_bytes, initial_read=True,
+                page_bytes=first_page_bytes, max_page_output_bytes=first_output_bytes, initial_read=True,
                 reference_file=Path(result["recovery"]["reference_file"]), reader_argv=reader_argv)
     result["instruction"] += (" Append --select '<pointer>' or --list to recovery.command for current fields. "
         "Reader references keep complete digests locally; the CLI checks them on every page. "
@@ -383,6 +388,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,
+                              first_read_budget: tuple[int, int],
                               page_attached: bool = False,
                               reader_argv: list[str] | None = None) -> dict[str, Any]:
     """Build the bounded native reference without publishing or registering it."""
@@ -431,8 +437,10 @@ def fits() -> bool:
             "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)
+        first_page_bytes, first_output_bytes = first_read_budget
         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)
+            page_bytes=first_page_bytes, max_page_output_bytes=first_output_bytes,
+            initial_read=True, reader_argv=reader_argv)
         if not fits():
             result.pop("first_read_command")  # A complete optional selection or the existing generic recovery.
             result["overview"] = overview
@@ -504,6 +512,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,
+                                first_read_page_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]:
@@ -521,6 +530,13 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
     if max_page_output_bytes is not None:
         output_bytes = min(output_bytes, max_page_output_bytes)
         page_bytes = min(page_bytes, output_bytes)
+    # The generated opening crosses a tool boundary even when returned as a
+    # command. Keep its inferred budget out of bulk and selection recovery.
+    first_output_bytes = (output_bytes if max_page_output_bytes is not None
+        else min(output_bytes, NATIVE_CONTEXT_FIRST_READ_FALLBACK_BYTES))
+    first_page_bytes = (min(page_bytes, first_output_bytes) if first_read_page_bytes is None
+        else first_read_page_bytes)
+    first_read_budget = first_page_bytes, first_output_bytes
     typed = projected.get("schema_version") == "context_delivery_v1"
     mode = projected["mode"] if typed else "full"
     ancestry = projected.get("ancestry", 0)
@@ -553,7 +569,7 @@ def pin_initial_selection(plan: dict[str, Any]) -> None:
         selections = plan["selections"]
         initial_read_pointers = [item["pointer"] for item in selections]
         try:
-            initial_read_reference = write_initial_context_read_reference(Path(current["path"]), current["sha256"],
+            initial_read_reference, _ = write_initial_context_read_reference(Path(current["path"]), current["sha256"],
                 current_document, selections)
         except OSError:
             # Read-only capture roots keep the existing initial recipe;
@@ -585,7 +601,8 @@ 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, page_attached=page_attached, reader_argv=reader_argv)
+            page_bytes=page_bytes, output_bytes=output_bytes, first_read_budget=first_read_budget,
+            page_attached=page_attached, reader_argv=reader_argv)
 
     if typed and mode == "delta":
         result = None
@@ -624,11 +641,12 @@ def view(kind: str, ref: dict[str, Any], body: bytes) -> dict[str, Any]:
     observed = register_native_context_view(context_directory,
         {**result, "ancestry": ancestry}, hook_session=hook_session)
     if (progressive_enabled and observed and "first_read_command" in result
-            and observed["pointer_bytes"] <= output_bytes // 4):
+            and observed["pointer_bytes"] <= first_output_bytes // 4):
         first = Path(observed["path"])
         result["first_read_reference"] = str(first)
         result["first_read_command"] = context_capture_read_command(Path(current["path"]), current["sha256"],
-            reference_file=first, page_bytes=page_bytes, max_page_output_bytes=output_bytes, reader_argv=reader_argv)
+            reference_file=first, page_bytes=first_page_bytes,
+            max_page_output_bytes=first_output_bytes, reader_argv=reader_argv)
     return result
 
 
@@ -696,8 +714,40 @@ 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 _bounded_initial_page(navigation: dict[str, Any], plan: dict[str, Any], *,
+                          page_bytes: int, output_bytes: int, reader_argv: list[str]) -> dict[str, Any] | None:
+    """Try the typed opening and keep the remainder pinned to the same capture."""
+    if "opening" not in plan or "first_read_command" not in navigation:
+        return None
+    try:
+        path, _, document = _read_capture(plan["capture"]["path"], plan["capture"]["sha256"])
+        digest = plan["capture"]["sha256"]
+        opening, content = write_initial_context_read_reference(path, digest, document, plan["opening"]["selections"])
+        if len(content) > page_bytes or len(content) >= output_bytes:
+            return None
+        remaining, _ = write_initial_context_read_reference(path, digest, document, plan["remaining"]["selections"])
+
+        def command(reference: Path) -> str:
+            return context_capture_read_command(path, digest, page_bytes=page_bytes,
+                max_page_output_bytes=output_bytes, reference_file=reference, reader_argv=reader_argv)
+
+        navigation = {**navigation, "first_read_command": command(opening), "remaining_read_command": command(remaining),
+            "instruction": ("This opening covers the decision and read obligations. Before work, read its content, "
+                "then follow remaining_read_command and its cursor through the rest of the initial selection. "
+                + navigation["instruction"])}
+        page = _capture_page(path, content, digest, offset=0, size=len(content), output_limit=None,
+            instruction=False, select=[], reference_file=opening, reader_argv=reader_argv)
+        # An opening is useful only as a complete, independently readable
+        # page. Keep the ordinary cursor when even the opening cannot fit.
+        page["navigation"] = navigation
+        return page if len(_native_output_bytes(page)) <= output_bytes else None
+    except (OSError, ValueError):
+        return None  # The same-capture ordinary reader is already available.
+
+
 def _saved_context_read_page(args: argparse.Namespace, navigation: dict[str, Any], *,
-                             initial_reference: str | None, reader_argv: list[str]) -> dict[str, Any]:
+                             initial_reference: str | None, initial_read_plan: dict[str, Any] | None,
+                             output_limit: int | 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
@@ -705,9 +755,16 @@ def _saved_context_read_page(args: argparse.Namespace, navigation: dict[str, Any
     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"],
+        "max_page_output_bytes": output_limit,
         "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 page["ok"]:
+        if (pinned and initial_read_plan and output_limit is not None and not page["eof"]
+                and "previously_delivered_fields" not in page
+                and (opening := _bounded_initial_page(navigation, initial_read_plan,
+                    page_bytes=args.page_bytes if args.page_bytes is not None else output_limit,
+                    output_bytes=output_limit, reader_argv=reader_argv)) is not None):
+            return opening
         # Keep the already-budgeted page and cursor ahead of recovery metadata.
         page["navigation"] = page.pop("navigation")
     elif "recovery_command" not in page:
@@ -732,10 +789,19 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
         activation = native_context_activation(args)
         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.reference_file or args.selection_sha256
                 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.")
+        caller_output_limit = getattr(args, "max_page_output_bytes", None)
+        output_limit = caller_output_limit
+        if args.initial_read:
+            # Keep the inferred limit on the attached read. Bulk recovery and
+            # same-Turn selection use only the caller's explicit options.
+            output_limit = (NATIVE_CONTEXT_FIRST_READ_FALLBACK_BYTES if output_limit is None
+                else min(output_limit, NATIVE_CONTEXT_PAGE_OUTPUT_BYTES))
+        if saved_page and args.page_bytes is not None:
+            _validate_context_page_window(args.offset, args.page_bytes, output_limit)
         path, raw, payload = _read_capture(args.capture_file, args.capture_sha256,
                                            private=hook_session is not None)
         identity = payload.get("agent_identity")
@@ -761,6 +827,7 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
     native_prepared = False
     native_view = None
     initial_reference = None
+    initial_read_plan = 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.
@@ -776,11 +843,10 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
         if not activation["enabled"]:
             raise ValueError("context delivery is off")
         page_bytes, output_bytes = native_context_page_budget()
-        output_limit = getattr(args, "max_page_output_bytes", None)
-        if output_limit is not None:
-            if output_limit <= 0:
+        if caller_output_limit is not None:
+            if caller_output_limit <= 0:
                 raise QuotaDecisionTransportError("Native context output limit must be positive.")
-            output_bytes = min(output_bytes, output_limit)
+            output_bytes = min(output_bytes, caller_output_limit)
             page_bytes = min(page_bytes, output_bytes)
 
         full_reference = _capture_read_reference(full_reference,
@@ -814,6 +880,7 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
             captured_decision_path=str(path) if args.initial_read and activation["progressive_enabled"] else None,
             hook_session=hook_session)
         projected = prepared.delivery
+        initial_read_plan = prepared.initial_read_plan
         native_prepared = True
         if projected.get("schema_version") == "context_delivery_v1":
             # Native publication owns the complete required view. Keep its
@@ -826,7 +893,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), page_attached=saved_page,
+            max_page_output_bytes=caller_output_limit,
+            first_read_page_bytes=args.page_bytes if args.initial_read else 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")
@@ -845,9 +913,8 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
                     "original decision and receipt; do not rerun admission."
                 )
             page_bytes, output_bytes = native_context_page_budget()
-            output_limit = getattr(args, "max_page_output_bytes", None)
-            if isinstance(output_limit, int) and output_limit > 0:
-                output_bytes = min(output_bytes, output_limit)
+            if isinstance(caller_output_limit, int) and caller_output_limit > 0:
+                output_bytes = min(output_bytes, caller_output_limit)
                 page_bytes = min(page_bytes, output_bytes)
             recovery_index = {**full_reference, "command": context_capture_read_command(
                 Path(full_reference["path"]), full_reference["sha256"],
@@ -873,7 +940,8 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
             + (CONTEXT_CAPTURE_NAVIGATION_GUIDANCE if activation["enabled"] else CONTEXT_CAPTURE_READ_GUIDANCE)}
     navigation = native_view or result
     if saved_page:
-        return _saved_context_read_page(args, navigation, initial_reference=initial_reference, reader_argv=reader_argv)
+        return _saved_context_read_page(args, navigation, initial_reference=initial_reference,
+            initial_read_plan=initial_read_plan, output_limit=output_limit, reader_argv=reader_argv)
     return navigation
 
 
diff --git a/loopx/cli_commands/quota_registration.py b/loopx/cli_commands/quota_registration.py
index 8cced17797..02f68a6b4e 100644
--- a/loopx/cli_commands/quota_registration.py
+++ b/loopx/cli_commands/quota_registration.py
@@ -254,7 +254,9 @@ def register_quota_command(
         help="For selected context-read continuations: exact derived-view digest from the preceding page.")
     quota_parser.add_argument("--offset", type=int, default=0, help="For context-read: byte offset returned by the previous page.")
     quota_parser.add_argument("--page-bytes", type=int,
-        help="For context-read: maximum UTF-8 content bytes per page (JSON default: 64000 under a 64000-byte stdout budget; Markdown: 4096). "
+        help="For context-read or an attached native JSON context-project page: maximum UTF-8 content bytes per page. "
+             "The ordinary reader defaults to 64000 in JSON and 4096 in Markdown. "
+             "The serialized stdout limit includes JSON framing and navigation. "
              "Larger pages require a sufficient --max-page-output-bytes within the capture limit; "
              "use only with known sufficient direct and retained-history output budgets.")
     quota_parser.add_argument("--max-page-output-bytes", type=int,
diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts
index fa7804a55b..500d055109 100644
--- a/loopx/control_plane/quota/progressive_context.ts
+++ b/loopx/control_plane/quota/progressive_context.ts
@@ -416,14 +416,16 @@ export function prepareProgressiveContextDelivery(request: JsonObject, legacyFie
   // Full native delivery can use that copy; delta still carries its own edits.
   if (request.source_edits === true && delivery.mode === "full") delete delivery.context;
   const observe = request.observe_first_read === true && !!jsonObject(required.context.selected_todo)?.todo_id;
-  const pointers = observe ? initialContextReadPointers(required.context, decisionFields) : [];
-  const observed = pointers.length ? observedFirstRead(codec, pointers) : undefined;
+  const groups = observe || request.include_initial_read_plan === true
+    ? initialContextReadGroups(required.context, decisionFields) : undefined;
+  const pointers = groups ? [...groups.opening, ...groups.remaining] : [];
+  const observed = observe && pointers.length ? observedFirstRead(codec, pointers) : undefined;
   const {operation: _operation, source_edits: _edits, source_number_tokens: _numbers, ...source} = request;
   return contextProjectionChanges(source, {context: required.context, delivery, projected: required.projected,
     ...(observed ? {observed_first_read: observed} : {}),
     read_sections: inlineReadSections(request, required.context),
-    ...(request.include_initial_read_plan === true
-      ? {initial_read_plan: nativeContextReadPlan(required.context, decisionFields)} : {})}, sourceValues);
+    ...(request.include_initial_read_plan === true && groups
+      ? {initial_read_plan: nativeContextReadPlan(required.context, pointers, groups)} : {})}, sourceValues);
 }
 
 /** Stateless presentation uses the same plan without inventing a native session,
@@ -437,18 +439,28 @@ export function prepareProgressiveContext(request: JsonObject, plan?: JsonObject
 
 /** Invocation-local navigation, not captured input or a read/retention receipt.
  * Decision vocabulary comes from the same owner as standalone initial reads. */
-function nativeContextReadPlan(context: JsonObject, decisionFields: readonly string[]): JsonObject {
-  const pointers = initialContextReadPointers(context, decisionFields), initial = pointers.length > 0;
-  return {initial, plan: capturedContextSelection(initial ? {context, initial_read: true}
-    : {context, select: [""], list_fields: true}, pointers)};
+function nativeContextReadPlan(context: JsonObject, pointers: string[],
+    groups: ReturnType<typeof initialContextReadGroups>): JsonObject {
+  if (!pointers.length) return {initial: false,
+    plan: capturedContextSelection({context, select: [""], list_fields: true})};
+  const select = (addresses: string[]) => capturedContextSelection({context, initial_read: true}, addresses);
+  return {initial: true, plan: select(pointers), ...(groups.opening.length && groups.remaining.length
+    ? {opening: select(groups.opening), remaining: select(groups.remaining)} : {})};
 }
 
 export function initialContextReadPointers(context: JsonObject, decisionFields: readonly string[]): string[] {
+  const {opening, remaining} = initialContextReadGroups(context, decisionFields);
+  return [...opening, ...remaining];
+}
+
+function initialContextReadGroups(context: JsonObject, decisionFields: readonly string[]): {
+  opening: string[]; remaining: string[];
+} {
   const object = (value: unknown): JsonObject | undefined =>
     value !== null && typeof value === "object" && !Array.isArray(value) ? value as JsonObject : undefined;
   const interaction = object(context.interaction_contract);
   // Unknown envelopes retain generic discovery instead of guessing their policy.
-  if (interaction?.schema_version !== "loopx_interaction_contract_v0") return [];
+  if (interaction?.schema_version !== "loopx_interaction_contract_v0") return {opening: [], remaining: []};
   const pointers: string[] = [];
   const include = (...path: (string | number)[]): void => {
     let value: unknown = context;
@@ -476,6 +488,10 @@ export function initialContextReadPointers(context: JsonObject, decisionFields:
   include("interaction_contract", "agent_channel", "required_reads");
   const work = ["interaction_contract", "agent_channel", "work_context"];
   include(...work, "instruction");
+  // If the full starting selection needs paging, put the current decision and
+  // its read obligations in one independently readable opening. The rest is
+  // still required under the original ordering and authority.
+  const opening = pointers.splice(0);
   include(...work, "user_todos");
   const channel = object(interaction.agent_channel);
   const sources = object(channel?.work_context)?.sources;
@@ -606,5 +622,5 @@ export function initialContextReadPointers(context: JsonObject, decisionFields:
     include("context_details", "instruction");
     include("context_details", "command_template");
   } else include("context_details", "unread_context_index");
-  return pointers;
+  return {opening, remaining: pointers};
 }
diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py
index 5ab7e7ee30..3c5800f3c4 100644
--- a/loopx/control_plane/turn_driver/decision_capture.py
+++ b/loopx/control_plane/turn_driver/decision_capture.py
@@ -621,13 +621,15 @@ def context_selection_binding(select: list[str], *, list_fields: bool, list_dept
 
 def write_initial_context_read_reference(path: Path, digest: str, payload: dict[str, Any],
                                         selections: list[dict[str, Any]], *,
-                                        observed_read: dict[str, Any] | None = None) -> Path:
+                                        observed_read: dict[str, Any] | None = None) -> tuple[Path, bytes]:
+    """Pin the shared rendered selection and return its exact rendered bytes."""
     raw, presented = _context_selection_bytes(payload, selections, list_fields=False, list_depth=None,
         present_goal_narrative=True)
-    return write_context_read_reference(path, digest,
+    reference = write_context_read_reference(path, digest,
         selection=context_selection_binding([item["pointer"] for item in selections], list_fields=False,
             list_depth=None, sha256=hashlib.sha256(raw).hexdigest(), presented=bool(presented)),
         observed_read=observed_read)
+    return reference, raw
 
 
 def context_selection_view(payload: dict[str, Any], *, select: list[str] | None,
diff --git a/loopx/control_plane/turn_driver/native_context_profile.py b/loopx/control_plane/turn_driver/native_context_profile.py
index c4e28af55f..7888d5262a 100644
--- a/loopx/control_plane/turn_driver/native_context_profile.py
+++ b/loopx/control_plane/turn_driver/native_context_profile.py
@@ -7,6 +7,9 @@
 # explicit reader budgets remain available for same-capture recovery.
 NATIVE_CONTEXT_PAGE_BYTES = 64000
 NATIVE_CONTEXT_PAGE_OUTPUT_BYTES = 64000
+# Native openings cross a tool boundary, whether attached or linked. When
+# that tool's cap is unknown, avoid the bulk-reader profile.
+NATIVE_CONTEXT_FIRST_READ_FALLBACK_BYTES = 12000
 # Keep the configuration declaration usable by older installed readers too.
 NATIVE_CONTEXT_LARGE_OUTPUT_BYTES = 64000
 NATIVE_CONTEXT_HISTORY_TOKENS = 20000
diff --git a/loopx/control_plane/turn_driver/native_first_read.py b/loopx/control_plane/turn_driver/native_first_read.py
index e59efb4fd8..9e805f6f70 100644
--- a/loopx/control_plane/turn_driver/native_first_read.py
+++ b/loopx/control_plane/turn_driver/native_first_read.py
@@ -12,7 +12,7 @@ def observed_first_read_reference(directory: str, target: dict[str, Any], recove
                                   context: dict[str, Any], plan: dict[str, Any]) -> dict[str, Any] | None:
     """Write the typed delivery plan once. The host renders it from the immutable source."""
     try:
-        reference = write_initial_context_read_reference(Path(recovery["path"]), recovery["sha256"],
+        reference, _ = write_initial_context_read_reference(Path(recovery["path"]), recovery["sha256"],
             context, plan["selections"],
             observed_read={"directory": directory, "target": target, "fields": plan["reused"]})
         return {"path": str(reference), "pointer_bytes": sum(len(key.encode()) + 4 for key in plan["reused"])}
diff --git a/tests/control_plane/test_context_goal_narrative.py b/tests/control_plane/test_context_goal_narrative.py
index aa5dd122c0..17aa29a23d 100644
--- a/tests/control_plane/test_context_goal_narrative.py
+++ b/tests/control_plane/test_context_goal_narrative.py
@@ -64,17 +64,29 @@ def finish(page):
 
     pointer = "/interaction_contract/agent_channel/work_context/sources/0/content/text"
     first = cli(*project, "--initial-read")
-    assert not first["eof"], "the 6000-byte response must exercise continuation"
+    remainder = first["navigation"].get("remaining_read_command")
+    assert not first["eof"] or remainder, "the 6000-byte response must exercise continuation"
     selection = finish(first)
     assert finish(follow(first["navigation"]["first_read_command"])) == selection
-    rows = {row["pointer"]: row for row in json.loads(selection)["selections"]}
+    pages = [first]
+    sections = json.loads(selection)["selections"]
+    if remainder:
+        next_section = follow(remainder)
+        pages.append(next_section)
+        sections += json.loads(finish(next_section))["selections"]
+    rows = {row["pointer"]: row for row in sections}
+    assert len(rows) == len(sections)
     assert rows["/heartbeat_receipt"]["value"] == receipt
     assert rows[pointer]["value"] == (original_text if conflict else expected_text)
     assert ("presentation" in rows[pointer]) is not conflict
 
-    observed = verified_context_selection(_state(journal), first)
-    assert observed is not None and observed["read_sections"]["/heartbeat_receipt"]
-    assert (pointer in observed["read_sections"]) is conflict
+    read_sections = {}
+    for page in pages:
+        observed = verified_context_selection(_state(journal), page)
+        assert observed is not None
+        read_sections.update(observed["read_sections"])
+    assert read_sections["/heartbeat_receipt"]
+    assert (pointer in read_sections) is conflict
     literal = [{"pointer": pointer, "value": original_text}]
     for page in [cli("quota", "context-read", *source, "--select", pointer), cli(*project, "--select", pointer)]:
         assert json.loads(finish(page))["selections"] == literal
diff --git a/tests/control_plane/test_context_read_reference.py b/tests/control_plane/test_context_read_reference.py
index b66eec7eb1..24a0b75134 100644
--- a/tests/control_plane/test_context_read_reference.py
+++ b/tests/control_plane/test_context_read_reference.py
@@ -30,6 +30,7 @@ def pinned_initial_view(payload, directory):
     plan = turn_envelope._turn_envelope_result("context_select", context=payload, initial_read=True)
     capture = {key: base["recovery"][key] for key in ("path", "sha256")}
     return native_context_capture_view(payload, directory, captured_recovery=base["recovery"],
+        max_page_output_bytes=64000,
         initial_read_plan={"initial": True, "plan": plan, "capture": capture})
 
 
diff --git a/tests/control_plane/test_progressive_native_context.py b/tests/control_plane/test_progressive_native_context.py
index f35629c344..b30279c39b 100644
--- a/tests/control_plane/test_progressive_native_context.py
+++ b/tests/control_plane/test_progressive_native_context.py
@@ -9,6 +9,7 @@
 import pytest
 
 from loopx.cli_commands.quota_capture import read_context_capture
+from loopx.control_plane.testing.canary_harness import run_json_cli_result
 from loopx.control_plane.turn_driver import codex_context as context
 from tests.control_plane.codex_checkpoint_fixture import reference_metadata_oracle
 from test_codex_context_lifecycle import setup, project, deliver, read_pages
@@ -469,11 +470,18 @@ def test_public_quota_cli_delivers_required_view_and_original_evidence(acceptanc
     lifecycle.mkdir(mode=0o700)
     (tmp_path / "captures").mkdir(mode=0o700)
     setup(lifecycle, monkeypatch)
-    code, packet = run("quota", "should-run", "--goal-id", "goal-acceptance", "--agent-id", "agent-a",
+    # Use the raw CLI response here. The acceptance fixture's policy reader
+    # unwraps the navigation before returning the authoritative document.
+    code, packet = run_json_cli_result("quota", "should-run", "--goal-id", "goal-acceptance", "--agent-id", "agent-a",
         "--todo-id", "todo_export", "--context-delivery", "on", "--codex-app",
         "--codex-context-dir", str(lifecycle), "--decision-output-root", str(tmp_path / "captures"),
-        "--turn-instance-id", "original-turn")
+        "--turn-instance-id", "original-turn", registry_path=workspace.parent / "registry.json",
+        runtime_root=workspace.parent / "runtime")
     assert code == 0, {key: packet.get(key) for key in ("error", "error_code", "error_detail", "reason", "reason_code")}
+    first_read = shlex.split(packet["first_read_command"])
+    assert int(first_read[first_read.index("--max-page-output-bytes") + 1]) == 12000
+    bulk_read = shlex.split(packet["recovery"]["command"])
+    assert int(bulk_read[bulk_read.index("--max-page-output-bytes") + 1]) == 64000
     recovery = context._state(lifecycle)["recovery"]
     assert Path(recovery["path"]).name == "required-context.json"
     required = json.loads(Path(recovery["path"]).read_bytes())
diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py
index d493ea49a6..abe20b78e2 100644
--- a/tests/control_plane/test_quota_execution_preparation.py
+++ b/tests/control_plane/test_quota_execution_preparation.py
@@ -454,9 +454,16 @@ def read(*argv):
 
         navigation = read(*args)
         command = shlex.split(navigation["first_read_command"])
-        page = read(*command[command.index("quota"):])
-        assert page["eof"]
-        values = {row["pointer"]: row["value"] for row in json.loads(page["content"])["selections"]}
+        selected = bytearray()
+        while True:
+            page = read(*command[command.index("quota"):])
+            assert page["offset"] == len(selected)
+            selected.extend(page["content"].encode())
+            if page["eof"]:
+                assert len(selected) == page["total_bytes"]
+                break
+            command = shlex.split(page["next_command"])
+        values = {row["pointer"]: row["value"] for row in json.loads(selected)["selections"]}
         boundary = packet["goal_boundary"]
         assert boundary["write_scope"] == scopes and boundary["requires_parent_approval"] == ["publish"]
         assert values["/heartbeat_receipt"] == packet["heartbeat_receipt"]
diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py
index f5bbe435b7..6ccdcf8b83 100644
--- a/tests/control_plane/test_saved_context_project.py
+++ b/tests/control_plane/test_saved_context_project.py
@@ -63,7 +63,12 @@ def test_saved_projection_checks_original_before_any_publication(tmp_path, failu
 
 
 @pytest.mark.parametrize("root_kind", ["unsafe", "symlink", "missing_parent"])
-def test_enabled_cache_preparation_failure_returns_error_with_original_reference(tmp_path, monkeypatch, root_kind):
+@pytest.mark.parametrize(("opening_args", "recovery_limit"), [
+    ([], 64000), (["--initial-read"], 64000),
+    (["--initial-read", "--max-page-output-bytes", "12000"], 12000),
+], ids=["linked", "attached", "explicit"])
+def test_enabled_cache_preparation_failure_returns_error_with_original_reference(
+        tmp_path, monkeypatch, root_kind, opening_args, recovery_limit):
     from loopx.cli import build_parser
     from loopx.cli_commands.quota_capture import project_saved_context_capture
 
@@ -90,10 +95,13 @@ def test_enabled_cache_preparation_failure_returns_error_with_original_reference
     args = build_parser().parse_args(["--format", "json", "quota", "context-project",
         "--capture-file", str(path), "--capture-sha256", digest, "--goal-id", "saved-goal",
         "--agent-id", "saved-agent", "--turn-instance-id", "saved-turn",
-        "--decision-output-root", str(root), "--context-delivery", "on", "--codex-app", "--codex-context-dir", str(context)])
+        "--decision-output-root", str(root), "--context-delivery", "on", "--codex-app",
+        "--codex-context-dir", str(context), *opening_args])
     result = project_saved_context_capture(args)
     assert result["ok"] is False and result["error_code"] == "quota_decision_transport_failed"
     assert result["recovery"]["path"] == str(path) and result["recovery"]["sha256"] == digest
+    recovery = shlex.split(result["recovery"]["command"])
+    assert int(recovery[recovery.index("--max-page-output-bytes") + 1]) == recovery_limit
     assert "prior.json" not in json.dumps(result)
     assert path.read_bytes() == raw and stale.read_bytes() == stale_bytes
 
@@ -1087,8 +1095,13 @@ 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):
+@pytest.mark.parametrize(("repeat", "limit", "staged", "page_bytes", "paged"), [
+    (1, None, False, None, False), (480, None, True, None, False),
+    (480, 6400, True, None, False), (480, 64000, False, None, False),
+    (1100, 40000, True, 40000, False), (98, 12000, False, 900, True),
+], ids=["small", "unknown", "lower", "sufficient", "explicit-reader-window", "tight-reader-window"])
+def test_saved_initial_page_is_primary_and_its_recovery_replays_the_same_selection(
+        tmp_path, monkeypatch, repeat, limit, staged, page_bytes, paged):
     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"
@@ -1098,10 +1111,18 @@ def test_saved_initial_page_is_primary_and_its_recovery_replays_the_same_selecti
         "session_id": "session", "cwd": str(tmp_path)})
     receipt = {"turn_instance_id": "t"}
     text = "Keep the original requirement. 保留原要求🙂\n" * repeat
+    saved_limit = 32000 if repeat > 1 else None
+    selection_args = "quota should-run --todo-id {todo_id}"
+    if saved_limit is not None:
+        selection_args += f" --max-page-output-bytes {saved_limit}"
+    # The paged case must exceed the window even before supporting context.
+    stop = "Stop before publishing. " * 100 if paged else None
     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": {
+        **({"goal_boundary": {"stop": stop}} if stop is not None else {}),
+        "interaction_contract": {"schema_version": "loopx_interaction_contract_v0",
+            "cli_channel": {"selection_command": {"command_args_template": selection_args}}, "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()
@@ -1109,8 +1130,8 @@ def test_saved_initial_page_is_primary_and_its_recovery_replays_the_same_selecti
     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"]
+    if limit is not None:
+        project += ["--max-page-output-bytes", str(limit)]
 
     def cli(*args):
         return _run_cli(registry, runtime, *args, cwd=tmp_path)
@@ -1124,25 +1145,45 @@ def follow(command):
     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)
+    window = ["--page-bytes", str(page_bytes)] if page_bytes is not None else []
+    code, first = cli(*project, "--initial-read", *window)
+    assert code == 0 and first["offset"] == 0 and first["eof"] is not paged
     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 ("remaining_read_command" in navigation) == staged
+    assert ("next_command" in first) is paged
+    if page_bytes is not None:
+        for command in (navigation["first_read_command"], navigation.get("remaining_read_command"),
+                        first.get("next_command")):
+            if command is not None:
+                words = shlex.split(command)
+                assert int(words[words.index("--page-bytes") + 1]) == page_bytes
+    # Read the saved same-Turn selection recipe from the committed projection.
+    # An implicit cap for this page does not become an explicit guard option.
+    operation = follow(navigation["recovery"]["command"] + " --select /interaction_contract/cli_channel/selection_command")
+    assert operation["eof"]
+    template = json.loads(operation["content"])["selections"][0]["value"]["command_args_template"]
+    words = shlex.split(template)
+    budget = int(words[words.index("--max-page-output-bytes") + 1]) if "--max-page-output-bytes" in words else None
+    assert budget == (limit if limit is not None else saved_limit)
+    recovery_limit = limit if limit is not None else 64000
+    limit = limit if limit is not None else 12000
     assert len(json.dumps(first, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= limit
 
-    def finish(page):
+    def finish(page, *, limit=None, content_limit=None):
         body = bytearray()
         while True:
             assert page["offset"] == len(body)
-            body.extend(page["content"].encode())
+            if limit is not None:
+                assert len(json.dumps(page, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= limit
+            content = page["content"].encode()
+            if content_limit is not None:
+                assert len(content) <= content_limit
+            body.extend(content)
             if page["eof"]:
                 assert len(body) == page["total_bytes"]
                 return bytes(body)
@@ -1152,14 +1193,37 @@ def finish(page):
     # 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
+    if staged:
+        codex_context_hook(str(context), {"hook_event_name": "PreCompact", "source": "startup",
+            "session_id": "session", "cwd": str(tmp_path)})
     files = set(context.rglob("*"))
     journal = (context / "session.json").read_bytes()
-    body = finish(first)
-    assert finish(follow(navigation["first_read_command"])) == body
+    body = finish(first, content_limit=page_bytes)
+    assert finish(follow(navigation["first_read_command"]), content_limit=page_bytes) == 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
+    if stop is not None:
+        assert selected["/goal_boundary"] == {"stop": stop}
+    requirement = "/interaction_contract/agent_channel/work_context/sources/0/content/text"
+    if staged:
+        assert requirement not in selected and "Before work" in navigation["instruction"]
+        remaining = finish(follow(navigation["remaining_read_command"]), content_limit=page_bytes)
+        detail = {item["pointer"]: item["value"] for item in json.loads(remaining)["selections"]}
+        assert not selected.keys() & detail.keys()
+        assert detail[requirement] == text
+    else:
+        assert selected[requirement] == text
+    # Clients that receive navigation before the initial read use the same
+    # transport budget even when the capture has much more required content.
+    recipe = shlex.split(reference["first_read_command"])
+    assert int(recipe[recipe.index("--max-page-output-bytes") + 1]) == limit
+    for view in (reference, navigation):
+        recovery = shlex.split(view["recovery"]["command"])
+        assert int(recovery[recovery.index("--max-page-output-bytes") + 1]) == recovery_limit
+    linked = json.loads(finish(follow(reference["first_read_command"]), limit=limit))["selections"]
+    fields = {item["pointer"]: item["value"] for item in linked}
+    assert fields[requirement] == text and fields["/heartbeat_receipt"] == receipt
     assert original.read_bytes() == raw and (context / "session.json").read_bytes() == journal
     assert set(context.rglob("*")) == files
     if repeat == 1:
@@ -1175,10 +1239,168 @@ def finish(page):
         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)
+        view = json.loads(recovered)
+        recipe = view["interaction_contract"]["cli_channel"]["selection_command"]
+        # The recovery view may bind the same-Turn command's transport. All
+        # other decided values, including the original receipt, are exact.
+        assert shlex.split(recipe["command_args_template"])[:4] == shlex.split(selection_args)
+        recipe["command_args_template"] = selection_args
+        assert view == json.loads(raw)
     assert not registry.exists() and not runtime.exists()
 
 
+def test_attached_selected_page_uses_the_same_window_as_its_reader(tmp_path, monkeypatch):
+    from loopx.control_plane.turn_driver.codex_context import codex_context_hook
+
+    context = 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)})
+    registry, runtime = tmp_path / "registry.json", tmp_path / "runtime"
+    source = tmp_path / "decision.json"
+    work = {"constraint": "Keep this source verbatim. 保留原文🙂" * 75}
+    raw = json.dumps({"ok": True, "goal_id": "g", "agent_identity": {"agent_id": "a"},
+        "heartbeat_receipt": {"turn_instance_id": "t"}, "work": work}, ensure_ascii=False).encode()
+    source.write_bytes(raw)
+    project = ["quota", "context-project", "--capture-file", str(source),
+        "--capture-sha256", hashlib.sha256(raw).hexdigest(), "--goal-id", "g", "--agent-id", "a",
+        "--turn-instance-id", "t", "--codex-app", "--context-delivery", "on",
+        "--codex-context-dir", str(context), "--select", "/work", "--max-page-output-bytes", "12000"]
+    before = {path: path.read_bytes() for path in context.rglob("*") if path.is_file()}
+    for invalid in ("255", "12001"):
+        code, rejected = _run_cli(registry, runtime, *project, "--page-bytes", invalid, cwd=tmp_path)
+        assert code and not rejected["ok"] and "--page-bytes" in rejected["reason"]
+    assert {path: path.read_bytes() for path in context.rglob("*") if path.is_file()} == before
+
+    code, page = _run_cli(registry, runtime, *project, "--page-bytes", "800", cwd=tmp_path)
+    assert code == 0 and not page["eof"]
+    content = bytearray()
+    while True:
+        assert page["ok"] and page["offset"] == len(content)
+        data = page["content"].encode()
+        assert 0 < len(data) <= 800
+        content.extend(data)
+        if page["eof"]:
+            assert len(content) == page["total_bytes"]
+            break
+        args = shlex.split(page["next_command"])
+        assert args[args.index("--page-bytes") + 1] == "800"
+        code, page = _run_cli(registry, runtime, *args[args.index("quota"):], cwd=tmp_path)
+        assert code == 0
+    assert json.loads(content)["selections"] == [{"pointer": "/work", "value": work}]
+    assert source.read_bytes() == raw and not registry.exists() and not runtime.exists()
+
+
+@pytest.mark.skipif(not os.environ.get("LOOPX_NATIVE_CHECKPOINT_BIN"), reason="Explicit native Codex; loopback inference only")
+@pytest.mark.parametrize(("tool_tokens", "history_tokens", "byte_limit", "code_mode"), [
+    (6000, 20000, None, False), (22000, 3000, 6400, False), (6000, 20000, None, True),
+], ids=["tool-limited", "history-limited", "code-mode"])
+def test_bounded_initial_read_survives_the_actual_native_tool_limit(
+        tmp_path, tool_tokens, history_tokens, byte_limit, code_mode):
+    from loopx.control_plane.turn_driver.codex_context import codex_context_configuration
+    from loopx.control_plane.turn_driver.decision_capture import CODE_MODE_EXEC_RESULT_FORWARDER
+    from tests.control_plane.codex_checkpoint_fixture import (
+        last_context_page, local_inference_provider, native_process_continuation, selected_context_values,
+    )
+    from tests.control_plane.test_codex_context_boundary import _app_server, _configure_provider, _native_fixture
+
+    binary, env, workspace, home = _native_fixture(tmp_path)
+    env.update(NODE_OPTIONS="--experimental-sqlite", LOOPX_USAGE_PING="0")
+    context = workspace / "context"
+    context.mkdir(mode=0o700)
+    cli = [os.environ.get("LOOPX_NATIVE_CONTEXT_CLI") or str(Path(sys.executable).with_name("loopx"))]
+    cli += ["--registry", str(workspace / "absent-registry.json"), "--runtime-root", str(workspace / "absent-runtime")]
+    configuration = codex_context_configuration(str(context), goal_id="g", agent_id="a", cli_argv=cli,
+        scenario="codex_app_heartbeat")
+    (home / "hooks.json").write_text(json.dumps({"hooks": configuration["hooks"]}))
+    source = workspace / "original.json"
+    requirement = "".join(f"Criterion {i:04}: preserve exact scope before touching outside paths.\n" for i in range(650))
+    raw = json.dumps({"ok": True, "goal_id": "g", "agent_identity": {"agent_id": "a"},
+        "heartbeat_receipt": {"turn_instance_id": "t"}, "effective_action": "normal_run",
+        "goal_boundary": {"stop": "Stop before publishing."}, "interaction_contract": {
+            "schema_version": "loopx_interaction_contract_v0", "agent_channel": {"work_context": {
+                "instruction": "Read the exact requirement before acting.",
+                "sources": [{"source": "goal_state", "ordering": "before_work",
+                    "content": {"text": requirement, "state_file": "goal.md"}}]}}}}).encode()
+    source.write_bytes(raw)
+    project = shlex.join([*cli, "--format", "json", "quota", "context-project", "--capture-file", str(source),
+        "--capture-sha256", hashlib.sha256(raw).hexdigest(), "--goal-id", "g", "--agent-id", "a",
+        "--turn-instance-id", "t", "--codex-app", "--context-delivery", "on",
+        "--codex-context-dir", str(context), "--initial-read"])
+    phase, errors, detail = "start", [], bytearray()
+
+    def latest_output(request):
+        output = next(item["output"] for item in reversed(request["input"])
+            if item.get("type") in ("function_call_output", "custom_tool_call_output"))
+        return output if isinstance(output, str) else next(part["text"] for part in reversed(output)
+            if part.get("type") == "input_text")
+
+    def response(request, number, message):
+        nonlocal phase
+        try:
+            if wait := native_process_continuation(request, str(number), code_mode=code_mode):
+                if code_mode:
+                    wait["input"] = wait["input"].replace('"max_output_tokens": 22000',
+                        f'"max_output_tokens": {tool_tokens}')
+                else:
+                    wait["arguments"] = json.dumps({**json.loads(wait["arguments"]), "max_output_tokens": tool_tokens})
+                return wait
+            page = last_context_page(request, allow_selection=True)
+            if phase == "start":
+                command, phase = project + " --max-page-output-bytes 64000", "truncated"
+            elif phase == "truncated":
+                assert page is None and "context_capture_selection_page_v1" in latest_output(request)
+                command = project + (f" --max-page-output-bytes {byte_limit}" if byte_limit else "")
+                phase = "opening"  # The same original receipt and source.
+            elif phase == "opening":
+                assert page and page["ok"] and page["offset"] == 0 and page["eof"]
+                assert len(json.dumps(page, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= (byte_limit or 12000)
+                values = selected_context_values(page["content"])
+                assert values["/goal_boundary/stop"] == "Stop before publishing."
+                assert values["/heartbeat_receipt/turn_instance_id"] == "t"
+                command, phase = page["navigation"]["remaining_read_command"], "remaining"
+            elif phase == "remaining":
+                assert page and page["ok"] and page["offset"] == len(detail)
+                detail.extend(page["content"].encode())
+                assert len(detail) == (page["total_bytes"] if page["eof"] else page["next_offset"])
+                if not page["eof"]:
+                    command = page["next_command"]
+                else:
+                    text = selected_context_values(detail.decode())[
+                        "/interaction_contract/agent_channel/work_context/sources/0/content/text"]
+                    assert text == requirement
+                    command, phase = shlex.join(["printf", "%s", text.splitlines()[-1]]), "work"
+            else:
+                assert latest_output(request).split("\nOutput:\n", 1)[-1].strip() == requirement.splitlines()[-1]
+                phase = "done"
+                return message
+            call = {"id": f"fc_{number}", "call_id": f"call_{number}"}
+            arguments = json.dumps({"cmd": command, "max_output_tokens": tool_tokens, "yield_time_ms": 10000})
+            if code_mode:
+                return {**call, "type": "custom_tool_call", "name": "exec",
+                    "input": '// @exec: {"max_output_tokens": 12000}\nconst result=await tools.exec_command('
+                        + arguments + ');' + CODE_MODE_EXEC_RESULT_FORWARDER + ';'}
+            return {**call, "type": "function_call", "name": "exec_command", "arguments": arguments}
+        except Exception as exc:
+            errors.append(exc)
+            return message
+
+    with local_inference_provider(response) as (url, _, _):
+        _configure_provider(home, url)
+        config = home / "config.toml"
+        native_config = {**configuration["config_overrides"], "tool_output_token_limit": history_tokens}
+        overrides = "".join(f"{key}={json.dumps(value)}\n" for key, value in native_config.items())
+        text = config.read_text()
+        if code_mode:
+            text = text.replace("[features]\n", "[features]\ncode_mode=true\n")
+        config.write_text(overrides + text + "[sandbox_workspace_write]\nnetwork_access=true\n")
+        _app_server(binary, env, workspace, home, True, sandbox="workspace-write", turn_timeout_sec=120)
+    assert not errors and phase == "done", (errors, phase)
+    assert source.read_bytes() == raw
+    assert not (workspace / "absent-registry.json").exists() and not (workspace / "absent-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