diff --git a/docs/guides/context-delivery.md b/docs/guides/context-delivery.md
index 93934c0632..d4af7f341a 100644
--- a/docs/guides/context-delivery.md
+++ b/docs/guides/context-delivery.md
@@ -134,6 +134,13 @@ same registered source in the current session and epoch. The current source
 revision and Turn controls still appear. A change to the narrative or its
 source, an incomplete read, or compaction restores the current text.
 
+For ready work on the same registered route and Todo, the native first read can
+reference an exact write-scope list that the session already received in full.
+The rest of the current boundary, including guards and stops, still appears.
+If unsure about a path, select `/goal_boundary/write_scope` from the current
+recovery reader. Changes to the list or route, incomplete reads and compaction
+restore the current list.
+
 ### Read and recover the right object
 
 Use the path, digest and command returned by the current response. A native
diff --git a/docs/guides/context-delivery.zh-CN.md b/docs/guides/context-delivery.zh-CN.md
index 757807833c..3464dbe298 100644
--- a/docs/guides/context-delivery.zh-CN.md
+++ b/docs/guides/context-delivery.zh-CN.md
@@ -106,6 +106,11 @@ Full 表示完整保存视图可用，不要求判断任何问题前先遍历所
 首读仍交付当前来源修订和 Turn 控制条件。正文或来源变化、读取不完整或发生压缩时，
 读者恢复当前正文。
 
+同一注册路由和 Todo 的工作已就绪时，原生首读可以引用会话此前完整收到的同一份写入
+范围列表。当前边界的其他部分仍会显示，包括防护和停止条件。不确定某个路径是否在
+范围内时，通过当前恢复读取命令选择 `/goal_boundary/write_scope`。列表或路由变化、
+读取不完整或发生压缩时，会恢复当前完整列表。
+
 ### 读对对象，再恢复
 
 使用当前响应返回的路径、摘要与命令。原生响应的 `recovery` 指向当前完整必要视图；
diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md
index 7ac6fe66e1..07aede4d92 100644
--- a/docs/reference/required-work-context.md
+++ b/docs/reference/required-work-context.md
@@ -303,6 +303,12 @@ immutable capture.
 original digest-bound capture in the wake prompt. Recovery reuses the saved
 decision and its original receipt.
 
+The native `quota should-run` navigation may also link two reads when a ready
+decision has a large write scope. Read `first_read_command` for the decision,
+stop conditions and required reads. If `remaining_read_command` is present,
+follow it and its cursor before work, even when the opening reports EOF. It
+starts with the complete current write scope and uses the same capture.
+
 中文：对于已保存的 benchmark 准入结果，唤醒正文直接给出首读命令。一次调用即可
 取得首屏，已知字段可改用多个 `--select`。首读默认最多 12,000 个 UTF-8 字节，含导航。
 已知工具直出与历史保留上限时，可用 `--max-page-output-bytes` 指定匹配的字节预算，最高
@@ -317,6 +323,11 @@ decision and its original receipt.
 `navigation.first_read_command` 可重读首屏，其他字段沿
 `navigation.recovery.command` 读取。结果丢失时，用唤醒正文中的原文件及摘要恢复，复用原决策和回执。
 
+原生 `quota should-run` 的已就绪决策含较长的写入范围时，导航也可能给出两段读取。
+先用 `first_read_command` 取得决策、停止条件和必读项。如果给出
+`remaining_read_command`，即使首段已到 EOF，也要在开工前沿该命令及其游标读完。
+续读从完整的当前写入范围开始，两段绑定同一份捕获。
+
 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
diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index 512335a3e5..f2d96a633d 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -28,6 +28,7 @@
     native_context_page_budget,
 )
 from .quota_context_reader import (
+    _bounded_initial_read,
     _capture_selection_transport_template,
     bind_capture_selection_transport as bind_capture_selection_transport,
     _saved_context_read_page,
@@ -108,17 +109,19 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
     field_index = None
     initial_read_pointers = None
     initial_read_reference = None
+    initial_read_size = 0
     # A stateless attached page uses the ordinary verified selector below.
     # Its recovery recipe does not require an extra root index to be planned.
     stateless_first_read = progressive_enabled and page_attached and captured_recovery is not None
 
     def pin_initial_selection(plan: dict[str, Any]) -> None:
-        nonlocal initial_read_pointers, initial_read_reference
+        nonlocal initial_read_pointers, initial_read_reference, initial_read_size
         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, content = write_initial_context_read_reference(Path(current["path"]), current["sha256"],
                 current_document, selections)
+            initial_read_size = len(content)
         except OSError:
             # Read-only capture roots keep the existing initial recipe;
             # metadata is optional and never a read/authority receipt.
@@ -195,6 +198,10 @@ def view(kind: str, ref: dict[str, Any], body: bytes) -> dict[str, Any]:
         result["first_read_command"] = context_capture_read_command(Path(current["path"]), current["sha256"],
             reference_file=first, page_bytes=first_page_bytes,
             max_page_output_bytes=first_output_bytes, reader_argv=reader_argv)
+    elif (not page_attached and initial_read_plan and initial_read_plan.get("prefer_linked_opening") is True
+            and initial_read_size > first_page_bytes):
+        result = _bounded_initial_read(result, initial_read_plan, page_bytes=first_page_bytes,
+            output_bytes=first_output_bytes, reader_argv=reader_argv, attach_page=False) or result
     return result
 
 
diff --git a/loopx/cli_commands/quota_context_reader.py b/loopx/cli_commands/quota_context_reader.py
index 2b503cf58d..09ac026f46 100644
--- a/loopx/cli_commands/quota_context_reader.py
+++ b/loopx/cli_commands/quota_context_reader.py
@@ -445,6 +445,8 @@ def reference(value: dict[str, Any], *, navigation: bool = False) -> dict[str, s
             result["first_read_command"] = context_capture_read_command(Path(recovery["path"]), recovery["sha256"],
                 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)
+        if "remaining_read_command" in view:
+            result["remaining_read_command"] = view["remaining_read_command"]
     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. "
         "A reference or digest grants no authority or freshness.")
@@ -582,8 +584,9 @@ def _native_context_read_bytes(reference: dict[str, Any], raw: bytes, *,
         offset = page["next_offset"]
 
 
-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:
+def _bounded_initial_read(navigation: dict[str, Any], plan: dict[str, Any], *,
+                          page_bytes: int, output_bytes: int, reader_argv: list[str] | None,
+                          attach_page: bool = True) -> 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
@@ -599,14 +602,20 @@ 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)
 
+        guidance = ("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. "
+            "Both views pin the same capture.")
         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"])}
+            "instruction": (guidance + " " + navigation["instruction"] if attach_page
+                else navigation["instruction"] + " " + guidance)}
         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.
+        if not attach_page:
+            navigation["first_read_reference"] = str(opening)
+            return (navigation if len(_native_output_bytes(page)) <= output_bytes
+                and len(_native_output_bytes(navigation)) <= min(output_bytes, 7900) else None)
         page["read_instruction"] = (
             "EOF ends the opening. Before work, read its content and follow navigation.remaining_read_command "
             "and its next_command cursor through the remaining initial context. Both views pin the same capture. "
@@ -633,7 +642,7 @@ def _saved_context_read_page(args: argparse.Namespace, navigation: dict[str, Any
     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,
+                and (opening := _bounded_initial_read(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
diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts
index 1593589897..a1b0697c29 100644
--- a/loopx/control_plane/quota/progressive_context.ts
+++ b/loopx/control_plane/quota/progressive_context.ts
@@ -18,7 +18,8 @@ import {createHash} from "node:crypto";
 
 
 type Path = (string | number)[];
-type ReusableReadField = {path: Path; freshKeys: string[]; minimumBytes: number; sourceFile?: string};
+type ReusableReadField = {path: Path; freshKeys: string[]; minimumBytes: number; sourceIdentity?: string | string[]};
+export type ReusableExactBody = {path: Path; sourceIdentity: string};
 const MIN_REUSABLE_BODY_BYTES = 1024;
 const pointer = (path: Path): string => path.length ? "/" + path.map(key =>
   String(key).replaceAll("~", "~0").replaceAll("/", "~1")).join("/") : "";
@@ -28,11 +29,13 @@ const known = (row: JsonObject, fields: string[]): boolean => Object.keys(row).e
 
 /** One registry for bodies eligible for observed reuse, regardless of whether
  * the reader delivered the body itself or packaged it in a complete parent. */
-function reusableFirstReadFields(context: JsonObject): ReusableReadField[] {
+function reusableFirstReadFields(context: JsonObject, exactBody?: ReusableExactBody): ReusableReadField[] {
+  const exactFields = exactBody && at(context, exactBody.path) !== undefined
+    ? [{...exactBody, freshKeys: [], minimumBytes: MIN_REUSABLE_BODY_BYTES}] : [];
   const base: Path = ["interaction_contract", "agent_channel", "work_context", "sources"];
   const sources = at(context, base);
-  if (!Array.isArray(sources)) return [];
-  return sources.flatMap<ReusableReadField>((value, i) => {
+  if (!Array.isArray(sources)) return exactFields;
+  return [...exactFields, ...sources.flatMap<ReusableReadField>((value, i) => {
     const source = jsonObject(value), content = jsonObject(source?.content);
     if (!content) return [];
     const path = [...base, i, "content"];
@@ -42,7 +45,7 @@ function reusableFirstReadFields(context: JsonObject): ReusableReadField[] {
         (content.status === undefined || content.status === "loaded") &&
         typeof content.state_file === "string" && completeGoalNarrative(content))
       return [{path: [...path, "text"], freshKeys: [], minimumBytes: MIN_REUSABLE_BODY_BYTES,
-        sourceFile: content.state_file}];
+        sourceIdentity: content.state_file}];
     if (source?.kind === "agent_preferences")
       return [{path, freshKeys: ["ok", "revision", "instruction"], minimumBytes: MIN_REUSABLE_BODY_BYTES}];
     if (source?.source !== "goal_acceptance") return [];
@@ -72,22 +75,22 @@ function reusableFirstReadFields(context: JsonObject): ReusableReadField[] {
       }
       return fields;
     }) : [];
-  });
+  })];
 }
 
 function observedSectionDigest(body: string, field?: ReusableReadField): string {
-  // Same text at a reused array index is insufficient after a Goal file rebind.
-  const value = field?.sourceFile === undefined ? body : JSON.stringify([field.sourceFile, body]);
+  // Matching content is insufficient when a field is bound to another source.
+  const value = field?.sourceIdentity === undefined ? body : JSON.stringify([field.sourceIdentity, body]);
   return createHash("sha256").update(value).digest("hex");
 }
 
 /** Digests stay in the private observed-read journal, never repeated to the model.
  * The existing exact-source identity preserves non-JS numeric spellings. */
-export function contextReadSectionDigests(request: JsonObject, paths: Path[]): JsonObject {
+export function contextReadSectionDigests(request: JsonObject, paths: Path[], exactBody?: ReusableExactBody): JsonObject {
   const context = requireJsonObject(request.context, "read section source");
   const identity = contextSourceIdentity(request, request.source_number_tokens);
   const selected = new Set(paths.map(pointer));
-  const fields = new Map(reusableFirstReadFields(context).map(field => [pointer(field.path), field]));
+  const fields = new Map(reusableFirstReadFields(context, exactBody).map(field => [pointer(field.path), field]));
   const covered = new Map(paths.map(path => [pointer(path), {path, minimumBytes: MIN_REUSABLE_BODY_BYTES}]));
   for (const [address, field] of fields)
     if (paths.some(parent => pathStartsWith(field.path, parent))) covered.set(address, field);
@@ -124,11 +127,11 @@ export function contextReadSectionDigests(request: JsonObject, paths: Path[]): J
 
 /** Only registered, previously delivered bodies may use a reference. Current
  * decisions, stops, selection, settlement and unknown source shapes stay inline. */
-function observedFirstRead(request: JsonObject, pointers: string[]): JsonObject | undefined {
+function observedFirstRead(request: JsonObject, pointers: string[], exactBody?: ReusableExactBody): JsonObject | undefined {
   const context = requireJsonObject(request.context, "current first read");
   const seen = requireJsonObject(request.consumed_pointers, "observed sections");
   if (typeof jsonObject(context.selected_todo)?.todo_id !== "string") return undefined;
-  const fields = new Map(reusableFirstReadFields(context).map(field => [pointer(field.path), field]));
+  const fields = new Map(reusableFirstReadFields(context, exactBody).map(field => [pointer(field.path), field]));
   const parents = new Set([...fields.values()].flatMap(({path}) => path.map((_, i) => pointer(path.slice(0, i)))));
   const identity = contextSourceIdentity(request, request.source_number_tokens);
   const reused: JsonObject = {};
@@ -182,7 +185,7 @@ function observedFirstRead(request: JsonObject, pointers: string[]): JsonObject
   return {selections: grouped, reused};
 }
 
-function inlineReadSections(request: JsonObject, required: JsonObject): JsonObject {
+function inlineReadSections(request: JsonObject, required: JsonObject, exactBody?: ReusableExactBody): JsonObject {
   const original = requireJsonObject(request.context, "inline source"), paths: Path[] = [];
   const visit = (source: unknown, shown: unknown, path: Path) => {
     if (source === shown) { paths.push(path); return; }
@@ -191,7 +194,7 @@ function inlineReadSections(request: JsonObject, required: JsonObject): JsonObje
       (source as JsonObject)[key], (shown as JsonObject)[key], [...path, Array.isArray(source) ? Number(key) : key]);
   };
   visit(original, required, []);
-  return contextReadSectionDigests(request, paths);
+  return contextReadSectionDigests(request, paths, exactBody);
 }
 
 /** Return paths instead of round-tripping the caller's JSON numeric values
@@ -449,7 +452,7 @@ function nativeContextContinuity(codec: JsonObject): JsonObject {
 /** Plan and encode one native capture without sending its intermediate view
  * back through Python. Capture publication and native locking remain Host IO. */
 export function prepareProgressiveContextDelivery(request: JsonObject, legacyFields: readonly string[],
-    decisionFields: readonly string[], plan?: JsonObject): JsonObject {
+    decisionFields: readonly string[], plan?: JsonObject, exactBody?: ReusableExactBody): JsonObject {
   if (request.include_initial_read_plan !== undefined)
     requireBoolean(request.include_initial_read_plan, "include_initial_read_plan");
   const sourceValues: ProjectionSourceValues = new WeakMap();
@@ -464,13 +467,13 @@ export function prepareProgressiveContextDelivery(request: JsonObject, legacyFie
   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 observed = observe && pointers.length ? observedFirstRead(codec, pointers, exactBody) : 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),
+    read_sections: inlineReadSections(request, required.context, exactBody),
     ...(request.include_initial_read_plan === true && groups
-      ? {initial_read_plan: nativeContextReadPlan(required.context, pointers, groups)} : {})}, sourceValues, undefined, true);
+      ? {initial_read_plan: nativeContextReadPlan(required.context, pointers, groups, exactBody?.path)} : {})}, sourceValues, undefined, true);
 }
 
 /** Stateless presentation uses the same plan without inventing a native session,
@@ -485,12 +488,31 @@ 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, pointers: string[],
-    groups: ReturnType<typeof initialContextReadGroups>): JsonObject {
+    groups: ReturnType<typeof initialContextReadGroups>, openingDetail?: Path): 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)} : {})};
+  let {opening, remaining} = groups;
+  let linkedOpening = false;
+  if (openingDetail) {
+    const parent = openingDetail.slice(0, -1), detail = openingDetail.at(-1)!;
+    const body = jsonObject(at(context, parent)), address = pointer(parent);
+    // Small lists stay in the ordinary opening. The IO adapter also checks
+    // the serialized size against the actual tool window before linking.
+    if (opening.includes(address) && body && Object.hasOwn(body, detail)
+        && Buffer.byteLength(JSON.stringify(body[detail])) >= 8192) {
+      const preparation = "/interaction_contract/cli_channel/execution_preparation";
+      opening = opening.flatMap(field => field === address
+        ? Object.keys(body).filter(key => key !== detail).map(key => pointer([...parent, key])) : [field]);
+      if (remaining.includes(preparation)) opening.push(preparation);
+      remaining = [pointer(openingDetail), ...remaining.filter(field => field !== preparation)];
+      linkedOpening = true;
+    }
+  }
+  // Keep the ordinary selection (and historical raw cursors) byte-compatible.
+  return {initial: true, plan: select(pointers), ...(opening.length && remaining.length
+    ? {opening: select(opening), remaining: select(remaining)} : {}),
+    ...(linkedOpening ? {prefer_linked_opening: true} : {})};
 }
 
 export function initialContextReadPointers(context: JsonObject, decisionFields: readonly string[]): string[] {
diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts
index 649ffaad59..d8aeafc36b 100644
--- a/loopx/control_plane/quota/turn_envelope.ts
+++ b/loopx/control_plane/quota/turn_envelope.ts
@@ -1,6 +1,7 @@
 import {EXECUTION_PREPARATION_SCHEMA_VERSION} from "../work_items/execution_preparation.ts";
 import {capturedContextSelection} from "./context_selection.ts";
-import {contextReadSectionDigests, initialContextReadPointers, prepareProgressiveContext, prepareProgressiveContextDelivery, progressiveContextPlan} from "./progressive_context.ts";
+import {contextReadSectionDigests, initialContextReadPointers, prepareProgressiveContext, prepareProgressiveContextDelivery,
+  progressiveContextPlan, type ReusableExactBody} from "./progressive_context.ts";
 import type { LoopXTurnRouteValue } from "../turn_driver/turn_contract_generated.ts";
 import { EFFECTIVE_ACTIONS, EffectiveAction, type QuotaEffectiveActionValue } from "./effective_action.generated.ts";
 import { AGENT_SCOPE_FRONTIER_ACTIONS } from "../agents/agent_scope_frontier.generated.ts";
@@ -2296,6 +2297,29 @@ function replanCliView(payload: JsonObject, identity: ReplanValueIdentity = cano
     replan_action_packet: replanActionPacket(payload, reads)};
 }
 
+/** Only the exact list can carry forward. Current boundary controls stay inline. */
+function readyWriteScope(context: JsonObject): ReusableExactBody | undefined {
+  const interaction = jsonObject(context.interaction_contract);
+  const channel = jsonObject(interaction?.agent_channel);
+  const preparation = jsonObject(jsonObject(interaction?.cli_channel)?.execution_preparation);
+  const boundary = jsonObject(context.goal_boundary);
+  const identityAgent = jsonObject(context.agent_identity)?.agent_id;
+  const agent = context.agent_id ?? identityAgent;
+  const todo = jsonObject(context.selected_todo)?.todo_id;
+  const route = [context.registry, context.runtime_root, context.goal_id, agent, todo];
+  if (context.ok !== true || context.should_run !== true || context.effective_action !== EffectiveAction.NORMAL_RUN ||
+      context.normal_delivery_allowed !== true || channel?.delivery_allowed !== true ||
+      jsonObject(channel.work_context)?.complete !== true || jsonObject(interaction?.user_channel)?.action_required !== false ||
+      (identityAgent !== undefined && identityAgent !== agent) ||
+      preparation?.schema_version !== EXECUTION_PREPARATION_SCHEMA_VERSION || preparation.state !== "ready" ||
+      preparation.delivery_allowed !== true || preparation.agent_id !== agent || preparation.todo_id !== todo ||
+      boundary?.rule !== "stay_in_scope_or_stop" || Object.hasOwn(boundary, "write_scope_pointer") ||
+      !Array.isArray(boundary.write_scope) || boundary.write_scope.some(scope => typeof scope !== "string" || !scope) ||
+      !route.every(part => typeof part === "string" && !!part)) return undefined;
+  return {path: ["goal_boundary", "write_scope"],
+    sourceIdentity: JSON.stringify([route, context.goal_ref ?? null, boundary.handoff_mode ?? null])};
+}
+
 /** Present the long scope list on demand only while an ordinary Turn prepares work. */
 function progressivePreparationScope(payload: JsonObject):
   {path: string[]; marker_path: string[]; instruction: string} | undefined {
@@ -2501,14 +2525,15 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
   if (operation === "context_read_sections") {
     const selected = capturedContextSelection({...request, list_fields: false});
     return {sections: contextReadSectionDigests(request,
-      (selected.selections as {path: (string | number)[]}[]).map(row => row.path))};
+      (selected.selections as {path: (string | number)[]}[]).map(row => row.path),
+      readyWriteScope(requireJsonObject(request.context, "read section source")))};
   }
   if (operation === "progressive_context_delivery")
     return prepareProgressiveContext(request, checkedContextPresentationPlan(request));
   if (operation === "context_delivery") return request.native_projection === undefined
     ? prepareContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS)
     : prepareProgressiveContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS, INITIAL_CONTEXT_DECISION_FIELDS,
-      checkedContextPresentationPlan(request));
+      checkedContextPresentationPlan(request), readyWriteScope(requireJsonObject(request.context, "current projection")));
   if (operation === "native_context_lifecycle") return updateNativeContextLifecycle(request);
   if (operation === "context_recovery") return restoreContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS);
   if (operation === "bind_capture") return bindTurnDecisionCapture(request);
diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py
index 915e4ceaef..7791f14f09 100644
--- a/tests/control_plane/test_context_first_read.py
+++ b/tests/control_plane/test_context_first_read.py
@@ -455,6 +455,84 @@ def test_cli_current_goal_and_action_precede_complete_scoped_preferences(tmp_pat
     assert not registry.exists() and not runtime.exists()
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_ready_opening_keeps_controls_readable_with_a_large_write_scope(tmp_path, monkeypatch, provider):
+    from test_codex_context_lifecycle import setup
+    from test_quota_settlement_cli import (
+        AGENT_ID, GOAL_ID, TODO_ID, _heartbeat_receipt_count, _initialize_fixture_authority,
+        _run_cli, _spend_run_count, _write_fixture,
+    )
+    from loopx.control_plane.turn_driver.decision_capture import read_context_read_reference
+    from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = _write_fixture(tmp_path)
+    configuration = json.loads(registry.read_text())
+    scopes = [f"docs/areas/{index:03d}/fixture-entrypoints/published/**" for index in range(290)]
+    configuration["goals"][0]["coordination"].update(write_scope=scopes, requires_parent_approval=["publish"])
+    registry.write_text(json.dumps(configuration))
+    _initialize_fixture_authority(project, runtime, provider=provider, handoff_mode="hard_lease")
+    journal, captures = tmp_path / "journal", tmp_path / "captures"
+    journal.mkdir(mode=0o700)
+    captures.mkdir(mode=0o700)
+    setup(journal, monkeypatch)
+
+    def cli(*args):
+        code, result = _run_cli(registry, runtime, *args, cwd=project)
+        assert code == 0 and result["ok"], result
+        return result
+
+    binding = ("--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID)
+    cli("todo", "claim", *binding, "--claimed-by", AGENT_ID)
+    view = cli("quota", "should-run", "--codex-app", "--context-delivery", "on",
+        *binding, "--turn-instance-id", "wide-ready", "--decision-output-root", str(captures),
+        "--codex-context-dir", str(journal))
+
+    def read(command):
+        pages, body = [], bytearray()
+        while command:
+            words = shlex.split(command)
+            page = cli(*words[words.index("quota"):])
+            assert page["offset"] == len(body)
+            body.extend(page["content"].encode())
+            pages.append(page)
+            command = page.get("next_command")
+        assert pages[-1]["eof"] and len(body) == pages[-1]["total_bytes"]
+        return pages, selected_context_values(body.decode())
+
+    assert "Before work" in view["instruction"] and "remaining_read_command" in view
+    _, source_ref = read_context_read_reference(view["recovery"]["reference_file"])
+    path = Path(source_ref["path"])
+    raw = path.read_bytes()
+    source = json.loads(raw)
+    pages, opening = read(view["first_read_command"])
+    assert len(pages) == 1 and pages[0]["eof"]
+    _, opening_ref = read_context_read_reference(pages[0]["reference_file"])
+    assert opening_ref["sha256"] == source_ref["sha256"]
+    assert opening["/effective_action"] == "normal_run"
+    assert opening["/heartbeat_receipt/turn_instance_id"] == "wide-ready"
+    assert opening["/selected_todo/todo_id"] == TODO_ID
+    assert opening["/interaction_contract/cli_channel/execution_preparation/state"] == "ready"
+    assert opening["/goal_boundary/requires_parent_approval"] == ["publish"]
+    assert opening["/goal_boundary/stop_condition"] == source["goal_boundary"]["stop_condition"]
+    assert "/goal_boundary/write_scope" not in opening
+
+    pages, remaining = read(view["remaining_read_command"])
+    assert len(pages) > 1 and remaining["/goal_boundary/write_scope"] == scopes
+    _, remainder_ref = read_context_read_reference(pages[0]["reference_file"])
+    assert remainder_ref["sha256"] == source_ref["sha256"]
+    _, ordinary = read(view["recovery"]["command"] + " --initial-read")
+    # The pinned split and historical one-view selector cover the same leaves.
+    def leaves(values):
+        return {pointer: value for pointer, value in values.items()
+            if not isinstance(value, (dict, list)) or not value}
+    assert not opening.keys() & remaining.keys()
+    assert leaves(opening | remaining) == leaves(ordinary)
+    assert path.read_bytes() == raw and hashlib.sha256(raw).hexdigest() == source_ref["sha256"]
+    assert _heartbeat_receipt_count(runtime, "wide-ready") == 1
+    assert _spend_run_count(runtime) == 0
+
+
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 def test_advertised_native_reader_keeps_the_complete_current_contract_across_pages(tmp_path, monkeypatch, provider):
     from test_quota_settlement_cli import (
@@ -1185,3 +1263,135 @@ def read(command):
     assert Path(fallback[0]["reference_file"]) == ordinary
     assert {key: value for key, value in after.items() if key != body_pointer} == facts
     assert fallback_bytes - warm_bytes > 3000
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_native_first_read_reuses_only_observed_write_scope(tmp_path, monkeypatch, provider):
+    from test_codex_context_lifecycle import deliver, setup
+    from test_quota_settlement_cli import (
+        AGENT_ID, GOAL_ID, TODO_ID, _heartbeat_receipt_count,
+        _initialize_fixture_authority, _run_cli, _spend_run_count, _write_fixture,
+    )
+    from loopx.control_plane.turn_driver.decision_capture import read_context_read_reference
+    from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = _write_fixture(tmp_path)
+    configuration = json.loads(registry.read_text())
+    scopes = [f"docs/areas/{i:03d}/entrypoints/public-fixture/**" for i in range(270)]
+    configuration["goals"][0]["quota"]["allowed_slots"] = 6
+    coordination = configuration["goals"][0]["coordination"]
+    coordination["write_scope"] = scopes
+    registry.write_text(json.dumps(configuration))
+    _initialize_fixture_authority(project, runtime, provider=provider, handoff_mode="hard_lease")
+    journal, captures = tmp_path / "journal", tmp_path / "captures"
+    journal.mkdir(mode=0o700)
+    captures.mkdir(mode=0o700)
+    event = setup(journal, monkeypatch)
+    scope_path = "/goal_boundary/write_scope"
+
+    def cli(*args, registry_path=registry):
+        code, payload = _run_cli(registry_path, runtime, *args, cwd=project)
+        assert code == 0 and payload["ok"], payload
+        return payload
+
+    def binding(turn):
+        return ("--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID, "--turn-instance-id", turn)
+
+    def capture(turn, registry_path=registry):
+        return cli("quota", "should-run", "--codex-app", "--context-delivery", "auto", "--delta-delivery", "auto",
+                   *binding(turn), "--decision-output-root", str(captures), "--codex-context-dir", str(journal),
+                   registry_path=registry_path)
+
+    def settle(turn, registry_path=registry):
+        cli("refresh-state", *binding(turn), "--classification", "fixture_validated",
+            "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress",
+            "--delivery-boundary", "in_flight_continuation", "--delivery-workspace-path", str(project),
+            "--no-global-sync", "--suppress-external-sinks", registry_path=registry_path)
+        spent = cli("quota", "spend-slot", *binding(turn), "--slots", "1", "--source", "heartbeat", "--execute",
+                    registry_path=registry_path)
+        assert spent["appended"]
+
+    def read(command):
+        pages, size, offset = [], 0, 0
+        while command:
+            words = shlex.split(command)
+            stdout = subprocess.check_output([sys.executable, "-m", "loopx.entrypoint", "--format", "json",
+                *words[words.index("quota"):]], cwd=project)
+            page = json.loads(stdout)
+            assert page["ok"] and page["offset"] == offset, page
+            pages.append(page)
+            size += len(stdout)
+            offset += len(page["content"].encode())
+            command = page.get("next_command")
+        assert pages[-1]["eof"] and offset == pages[-1]["total_bytes"]
+        return pages, selected_context_values("".join(page["content"] for page in pages)), size
+
+    def read_initial(view):
+        pages, facts, size = read(view["first_read_command"])
+        if continuation := view.get("remaining_read_command"):
+            more_pages, more_facts, more_size = read(continuation)
+            assert not facts.keys() & more_facts.keys()
+            return pages + more_pages, facts | more_facts, size + more_size
+        return pages, facts, size
+
+    cli("todo", "claim", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID, "--claimed-by", AGENT_ID)
+    cold, cold_facts, _ = read_initial(capture("scope-cold"))
+    assert cold_facts[scope_path] == scopes and len(cold) > 1
+    assert scope_path not in cold[0].get("previously_delivered_fields", [])
+    deliver(journal, event, cold[:1])
+    settle("scope-cold")
+
+    first_complete, complete_facts, _ = read_initial(capture("scope-unobserved"))
+    assert scope_path not in first_complete[0].get("previously_delivered_fields", [])
+    assert complete_facts[scope_path] == scopes
+    deliver(journal, event, first_complete)
+    settle("scope-unobserved")
+
+    guard = "Check the current authorization before writing outside the project."
+    configuration["goals"][0]["guards"] = [guard]
+    registry.write_text(json.dumps(configuration))
+    current = capture("scope-warm")
+    _, source_ref = read_context_read_reference(current["recovery"]["reference_file"])
+    original = Path(source_ref["path"])
+    raw = original.read_bytes()
+    boundary = json.loads(raw)["goal_boundary"]
+    warm, warm_facts, warm_bytes = read_initial(current)
+    _, restored_scope, scope_bytes = read(current["recovery"]["command"] + " --select " + scope_path)
+    assert restored_scope[scope_path] == scopes
+    codex_context.codex_context_hook(str(journal), {**event, "hook_event_name": "PostCompact"})
+    restored, restored_facts, restored_bytes = read_initial(current)
+    assert scope_path not in restored[0].get("previously_delivered_fields", [])
+    assert restored_facts[scope_path] == scopes and original.read_bytes() == raw
+    deliver(journal, event, restored)
+    settle("scope-warm")
+
+    alternate_registry = registry.with_name("alternate-registry.json")
+    alternate_registry.write_bytes(registry.read_bytes())
+    rebound, rebound_facts, _ = read_initial(capture("scope-rebound", alternate_registry))
+    assert scope_path not in rebound[0].get("previously_delivered_fields", [])
+    assert rebound_facts[scope_path] == scopes
+    settle("scope-rebound", alternate_registry)
+
+    coordination["write_scope"] = scopes[:-1] + ["docs/areas/fresh-scope/**"]
+    registry.write_text(json.dumps(configuration))
+    changed, changed_facts, _ = read_initial(capture("scope-revised"))
+    assert scope_path not in changed[0].get("previously_delivered_fields", [])
+    assert changed_facts[scope_path] == coordination["write_scope"]
+    settle("scope-revised")
+    off = cli("quota", "should-run", "--codex-app", "--context-delivery", "off", *binding("scope-off"))
+    assert off["goal_boundary"]["write_scope"] == coordination["write_scope"]
+    assert "first_read_command" not in off
+    assert _heartbeat_receipt_count(runtime, "scope-warm") == 1 and _spend_run_count(runtime) == 5
+
+    print(json.dumps({"provider": provider, "warm_bytes": warm_bytes, "recovery_bytes": restored_bytes,
+                      "warm_pages": len(warm), "recovery_pages": len(restored), "scope_selection_bytes": scope_bytes}))
+    assert scope_path in warm[0].get("previously_delivered_fields", []) and scope_path not in warm_facts
+    assert warm_facts["/goal_boundary/guards"] == [guard]
+    for key, value in boundary.items():
+        if key != "write_scope":
+            assert warm_facts[f"/goal_boundary/{key}"] == value
+    assert warm_facts["/interaction_contract/cli_channel/settlement_plan/identity/turn_instance_id"] == "scope-warm"
+    assert {key: value for key, value in restored_facts.items()
+            if key != "/goal_boundary" and not key.startswith(scope_path)} == warm_facts
+    assert restored_bytes - warm_bytes > 10_000