From 8227ad285d36e3e287b60d4429bcab6cfa8cfd1f Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 5 Nov 2026 22:19:09 +0800
Subject: [PATCH 1/6] Reuse observed preference and acceptance fields in native
 first reads

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/reference/required-work-context.md       |  9 +++
 loopx/cli_commands/quota_capture.py           | 40 +++++++++++--
 .../quota/native_context_lifecycle.ts         |  3 +
 .../quota/progressive_context.ts              | 31 ++++++++++
 loopx/control_plane/quota/turn_envelope.py    |  2 +-
 loopx/control_plane/quota/turn_envelope.ts    |  6 +-
 .../turn_driver/codex_context.py              | 12 ++--
 .../turn_driver/decision_capture.py           |  4 +-
 .../turn_driver/native_first_read.py          | 46 +++++++++++++++
 .../control_plane/test_context_first_read.py  | 56 +++++++++++++++++++
 10 files changed, 198 insertions(+), 11 deletions(-)
 create mode 100644 loopx/control_plane/turn_driver/native_first_read.py

diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md
index 3fe0c5993d..d3c10755b1 100644
--- a/docs/reference/required-work-context.md
+++ b/docs/reference/required-work-context.md
@@ -671,6 +671,15 @@ ceiling); if metadata cannot fit, recovery retains the current capture. Reader
 pages remain independently bounded. A complete raw tool log is not proof that
 the Host retained the same bytes in model history.
 
+In a native progressive session, the initial read can reference a previously
+delivered preferences block or acceptance outcome. The first page lists those
+pointers in `previously_delivered_fields`. Their requirements still apply.
+Use `recovery.command --select <pointer>` for the current body when needed.
+Reuse requires complete observed tool delivery in the same task, session and
+epoch, and a matching digest of the current visible field. A revised field
+appears in full. Losing the observation restores the ordinary initial read.
+A saved delta on its own provides no evidence of model retention.
+
 Each progressive `context_details.source_path` names this Turn's complete immutable
 `projection.json`, including when a warm wire omits unchanged material. The current
 `unread_context_index` preserves unobserved or changed supporting pointers and
diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index 4f174164eb..9ac04be4fb 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -205,6 +205,26 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
                     or not isinstance(pinned_selection["sha256"], str)
                     or BARE_SHA256_PATTERN.fullmatch(pinned_selection["sha256"]) is None):
                 raise ValueError("reader selection reference is incomplete or unsupported")
+            if observation := reference.get("observed_read"):
+                from ..control_plane.turn_driver.native_first_read import observation_current
+
+                if not observation_current(observation, path, digest):
+                    if offset:
+                        return {"ok": False, "reason": "Observed first-read scope changed during paging.",
+                            "recovery_command": context_capture_read_command(path, digest, initial_read=True,
+                                page_bytes=size, max_page_output_bytes=output_limit, reader_argv=reader_argv)}
+                    # Restart in the ordinary complete view of this same capture.
+                    # Subsequent pages must pin that view even if the journal changes.
+                    pinned_selection, initial_read = None, True
+                    try:
+                        reference_file = write_context_read_reference(path, digest)
+                    except OSError:
+                        reference_file = None
+                elif offset == 0:
+                    presentation = {**(presentation or {}), "previously_delivered_fields": list(observation["fields"]),
+                        "field_recovery": "These unchanged fields still apply. If needed or uncertain, append --select "
+                            "and the field pointer to recovery.command for its complete current body."}
+        if pinned_selection is not None:
             selectors, initial_read = pinned_selection["select"], pinned_selection["initial_read"]
             list_fields, list_depth = pinned_selection["list_fields"], pinned_selection["list_depth"]
             selection_digest = pinned_selection["sha256"]
@@ -331,9 +351,10 @@ 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_reference = view.get("first_read_reference")
         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,
-            reference_file=Path(result["recovery"]["reference_file"]), reader_argv=reader_argv)
+            page_bytes=page_bytes, max_page_output_bytes=output_bytes, initial_read=not first_reference,
+            reference_file=Path(first_reference or 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. "
         "A reference or digest grants no authority or freshness.")
@@ -456,6 +477,7 @@ def _native_context_read_bytes(reference: dict[str, Any], raw: bytes, *,
 
 
 def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
+                                progressive_enabled: bool = False,
                                 context_directory: str | None = None,
                                 hook_session: str | None = None,
                                 captured_recovery: dict[str, Any] | None = None,
@@ -545,8 +567,17 @@ def view(kind: str, ref: dict[str, Any], body: bytes) -> dict[str, Any]:
         write_decision_capture(wire, projected)
     from ..control_plane.turn_driver.codex_context import register_native_context_view
 
-    register_native_context_view(context_directory,
+    seen = register_native_context_view(context_directory,
         {**result, "ancestry": ancestry}, hook_session=hook_session)
+    if progressive_enabled and seen and initial_read_pointers and "first_read_command" in result:
+        from ..control_plane.turn_driver.native_first_read import observed_first_read_reference
+
+        first = observed_first_read_reference(context_directory, result["target"], current,
+            current_document, initial_read_pointers, seen, output_bytes)
+        if first is not None:
+            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)
     return result
 
 
@@ -568,7 +599,7 @@ def project_native_quota_context(payload: dict[str, Any], args: argparse.Namespa
         capture_id=capture_id, continuity_generation=getattr(args, "_context_delivery_generation", None),
         progressive_enabled=progressive_enabled)
     recovery = payload.get("recovery")
-    view = native_context_capture_view(payload, capture_directory, context_directory=directory,
+    view = native_context_capture_view(payload, capture_directory, context_directory=directory, progressive_enabled=progressive_enabled,
         metadata={"capture_sha256": capture_id, **({"recovery": recovery} if recovery is not None else {})},
         max_page_output_bytes=getattr(args, "max_page_output_bytes", None), reader_argv=reader_argv)
     return model_context_reference(view, reader_argv=reader_argv)
@@ -710,6 +741,7 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
                 page_bytes=page_bytes, output_bytes=output_bytes, reader_argv=reader_argv)
             reference = dict(full_reference)
         native_view = native_context_capture_view(projected, cache, context_directory=directory, hook_session=hook_session,
+            progressive_enabled=activation["progressive_enabled"],
             metadata={"recovery": full_reference, "capture_sha256": args.capture_sha256,
                 "context_delivery": activation},
             max_page_output_bytes=getattr(args, "max_page_output_bytes", None), reader_argv=reader_argv)
diff --git a/loopx/control_plane/quota/native_context_lifecycle.ts b/loopx/control_plane/quota/native_context_lifecycle.ts
index 06e551f741..209ae264ba 100644
--- a/loopx/control_plane/quota/native_context_lifecycle.ts
+++ b/loopx/control_plane/quota/native_context_lifecycle.ts
@@ -68,6 +68,9 @@ export function updateNativeContextLifecycle(request: JsonObject): JsonObject {
       delete state.candidate;
       delete state.pending;
       delete state.active_call;
+      delete state.consumed_pointers;
+      delete state.confirmed_read_sections;
+      delete state.section_reads;
       return state;
     }
     invalidate();
diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts
index 22bb9899d3..5b0aa34721 100644
--- a/loopx/control_plane/quota/progressive_context.ts
+++ b/loopx/control_plane/quota/progressive_context.ts
@@ -27,6 +27,37 @@ export function contextReadSectionDigests(request: JsonObject, paths: Path[]): J
     createHash("sha256").update(identity(at(context, path), ["context", ...path])).digest("hex")]));
 }
 
+/** Only previously delivered supplementary bodies may use a reference. Current
+ * decisions, stops, selection, settlement and unknown source shapes stay inline. */
+export function observedFirstRead(request: JsonObject, selections: {pointer: string; path: Path}[]): JsonObject {
+  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 {selections, reused: {}};
+  const sources = ["interaction_contract", "agent_channel", "work_context", "sources"];
+  const identity = contextSourceIdentity(request, request.source_number_tokens);
+  const reused: JsonObject = {};
+  const current = selections.flatMap(row => {
+    const {path, pointer: address} = row;
+    if (typeof seen[address] !== "string" || !pathStartsWith(path, sources)
+        || typeof path[4] !== "number" || path[5] !== "content") return [row];
+    const source = jsonObject(at(context, path.slice(0, 5)));
+    const preference = source?.kind === "agent_preferences" && path.length === 6;
+    const outcome = source?.source === "goal_acceptance" && path.length === 10
+      && path[6] === "goal_acceptance_contract" && path[7] === "outcome_requirements"
+      && path[8] === "outcomes" && typeof path[9] === "number";
+    const body = jsonObject(at(context, path));
+    if (!body || !(preference || outcome) || (outcome && typeof body.id !== "string")) return [row];
+    const content = identity(body, ["context", ...path]);
+    // A reference and its recovery instructions should cost less than the field.
+    if (Buffer.byteLength(content) < 1024 || createHash("sha256").update(content).digest("hex") !== seen[address])
+      return [row];
+    reused[address] = seen[address];
+    return (outcome ? ["id", "revision"] : ["ok", "revision", "instruction"]).filter(key =>
+      Object.hasOwn(body, key)).map(key => ({pointer: address + "/" + key, path: [...path, key]}));
+  });
+  return {selections: current, reused};
+}
+
 function inlineReadSections(request: JsonObject, required: JsonObject): JsonObject {
   const original = requireJsonObject(request.context, "inline source"), paths: Path[] = [];
   const visit = (source: unknown, shown: unknown, path: Path) => {
diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py
index 85e58fc392..e43c1968ac 100644
--- a/loopx/control_plane/quota/turn_envelope.py
+++ b/loopx/control_plane/quota/turn_envelope.py
@@ -60,7 +60,7 @@ def _turn_envelope_result(
         tokens = source_number_tokens(params["payload"])
         if tokens:
             request["source_number_tokens"] = tokens
-    elif operation in {"context_delivery", "context_recovery", "progressive_context_delivery", "context_read_sections"}:
+    elif operation in {"context_delivery", "context_recovery", "progressive_context_delivery", "context_read_sections", "observed_first_read"}:
         tokens = source_number_tokens(params)
         if tokens:
             request["source_number_tokens"] = tokens
diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts
index b25d761da0..d21659e1ac 100644
--- a/loopx/control_plane/quota/turn_envelope.ts
+++ b/loopx/control_plane/quota/turn_envelope.ts
@@ -1,5 +1,5 @@
 import {capturedContextSelection} from "./context_selection.ts";
-import {contextReadSectionDigests, initialContextReadPointers, prepareProgressiveContext, prepareProgressiveContextDelivery} from "./progressive_context.ts";
+import {observedFirstRead, contextReadSectionDigests, initialContextReadPointers, prepareProgressiveContext, prepareProgressiveContextDelivery} from "./progressive_context.ts";
 import {contextProjectionChanges, contextSourceIdentity, projectQuotaDiagnostics} from "./diagnostic_projection.ts";
 import type { LoopXTurnRouteValue } from "../turn_driver/turn_contract_generated.ts";
 import { EffectiveAction, type QuotaEffectiveActionValue } from "./effective_action.generated.ts";
@@ -1598,6 +1598,10 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
     return contextProjectionChanges(source, payload);
   }
   if (operation === "build") return buildTurnEnvelope(request);
+  if (operation === "observed_first_read") {
+    const selected = capturedContextSelection({...request, preserve_selection_order: true});
+    return observedFirstRead(request, selected.selections as {pointer: string; path: (string | number)[]}[]);
+  }
   if (operation === "context_read_sections") {
     const selected = capturedContextSelection({...request, list_fields: false});
     return {sections: contextReadSectionDigests(request,
diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py
index 99a3f9c84b..7481a7874f 100644
--- a/loopx/control_plane/turn_driver/codex_context.py
+++ b/loopx/control_plane/turn_driver/codex_context.py
@@ -578,10 +578,10 @@ def _native_context_request(state: dict[str, Any], scope: dict[str, Any]) -> dic
     return request
 
 
-def register_native_context_view(directory: str | None, view: dict[str, Any], *, hook_session: str | None = None) -> None:
+def register_native_context_view(directory: str | None, view: dict[str, Any], *, hook_session: str | None = None) -> dict[str, str]:
     """Bind the existing pending delivery to its actual paged wire, under lock."""
     if directory is None or not isinstance(view.get("target"), dict):
-        return
+        return {}
     with _locked(directory) as root:
         state = _state(root)
         pending = state.get("pending")
@@ -592,6 +592,7 @@ def register_native_context_view(directory: str | None, view: dict[str, Any], *,
             ancestry=view.get("ancestry", 0),
             read_sections=pending.get("full_read_sections", {}) if view.get("mode") == "full" else {})
         _save(root, state)
+        return state.get("consumed_pointers", {})
 
 
 def project_codex_context(payload: dict[str, Any], *, directory: str | None, capture_directory: Path,
@@ -634,10 +635,13 @@ def project_codex_context(payload: dict[str, Any], *, directory: str | None, cap
                 scope = dict(goal_id=goal_id, agent_id=agent_id, session_id=session, context_epoch=epoch)
                 if continuity_generation is not None:
                     scope["continuity_generation"] = continuity_generation
-                if state.get("read_scope") != scope:
+                selected = payload.get("selected_todo")
+                read_scope = {**scope, "surface": surface,
+                    "todo_id": selected.get("todo_id") if isinstance(selected, dict) else None}
+                if state.get("read_scope") != read_scope:
                     for key in ("consumed_pointers", "confirmed_read_sections", "section_reads"):
                         state.pop(key, None)
-                state["read_scope"] = scope
+                state["read_scope"] = read_scope
                 request = _native_context_request(state, scope)
                 recovery = capture_context_projection(payload, directory=capture_directory,
                     read_guidance=CONTEXT_CAPTURE_PAGE_GUIDANCE)
diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py
index 1521219608..1c7389b31e 100644
--- a/loopx/control_plane/turn_driver/decision_capture.py
+++ b/loopx/control_plane/turn_driver/decision_capture.py
@@ -291,7 +291,7 @@ def context_capture_read_command(path: Path, digest: str, *, offset: int = 0,
 
 
 def write_context_read_reference(path: Path, digest: str, *, selection: dict[str, Any] | None = None,
-                                 navigation: dict[str, Any] | None = None) -> Path:
+                                 navigation: dict[str, Any] | None = None, observed_read: dict[str, Any] | None = None) -> Path:
     """Immutable reader IO metadata, not a new identity or authority store.
 
     A short filename locates the document; only its complete stored digests are
@@ -302,6 +302,8 @@ def write_context_read_reference(path: Path, digest: str, *, selection: dict[str
         payload["selection"] = selection
     if navigation is not None:
         payload["navigation"] = navigation
+    if observed_read is not None:
+        payload["observed_read"] = observed_read
     suffix = ("." + hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest()[:16]
               if selection is not None or navigation is not None else "")
     reference = path.with_name(path.stem + suffix + ".reader.json")
diff --git a/loopx/control_plane/turn_driver/native_first_read.py b/loopx/control_plane/turn_driver/native_first_read.py
new file mode 100644
index 0000000000..63bd77ccfc
--- /dev/null
+++ b/loopx/control_plane/turn_driver/native_first_read.py
@@ -0,0 +1,46 @@
+"""Pin a smaller initial selection to previously observed native tool output."""
+from __future__ import annotations
+
+import hashlib
+import os
+from pathlib import Path
+from typing import Any
+
+from ..effect_runtime import EffectRuntimeRemoteError, EffectRuntimeStartupError
+from ..quota.turn_envelope import _turn_envelope_result
+from .decision_capture import context_selection_bytes, write_context_read_reference
+
+
+def observed_first_read_reference(directory: str, target: dict, recovery: dict, context: dict,
+                                  pointers: list[str], seen: dict[str, str], output_budget: int) -> Path | None:
+    try:
+        plan = _turn_envelope_result("observed_first_read", context=context, select=pointers, consumed_pointers=seen)
+        if not plan["reused"] or not plan["selections"]:
+            return None
+        # Keep each bounded page usable even when many outcomes match.
+        if sum(len(key.encode()) + 4 for key in plan["reused"]) > output_budget // 4:
+            return None
+        selected = plan["selections"]
+        raw = context_selection_bytes(context, selected, list_fields=False, list_depth=None)
+        return write_context_read_reference(Path(recovery["path"]), recovery["sha256"], selection={
+            "select": [item["pointer"] for item in selected], "initial_read": False,
+            "list_fields": False, "list_depth": None, "sha256": hashlib.sha256(raw).hexdigest()},
+            observed_read={"directory": directory, "target": target, "fields": plan["reused"]})
+    except (OSError, ValueError, EffectRuntimeRemoteError, EffectRuntimeStartupError):
+        return None  # The ordinary first read remains available.
+
+
+def observation_current(observation: dict[str, Any], path: Path, digest: str) -> bool:
+    from .codex_context import _private_root, _state
+
+    try:
+        state = _state(_private_root(observation["directory"]))
+        recovery = state["recovery"]
+        target = observation["target"]
+        return (target == recovery.get("target") and target["context_epoch"] == state["context_epoch"]
+                and target["session_id"] == state["session_id"] == os.environ.get("CODEX_THREAD_ID")
+                and recovery["sha256"] == digest and Path(recovery["path"]).resolve() == path.resolve()
+                and bool(observation["fields"]) and all(state.get("consumed_pointers", {}).get(pointer) == expected
+                    for pointer, expected in observation["fields"].items()))
+    except (OSError, ValueError, KeyError, TypeError, AttributeError):
+        return False
diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py
index 3e32980ec0..8b2de20ca1 100644
--- a/tests/control_plane/test_context_first_read.py
+++ b/tests/control_plane/test_context_first_read.py
@@ -667,3 +667,59 @@ def test_warm_delta_initial_read_uses_current_action_without_the_old_base(tmp_pa
     failure = json.loads(failed.stdout)
     assert failed.returncode != 0 and not failure["ok"] and "content" not in failure
     assert not registry.exists() and not runtime.exists()
+
+
+@pytest.mark.parametrize("invalidate", ["compact", "partial", "task_switch", "off"])
+def test_native_first_read_reuses_only_observed_current_fields(tmp_path, monkeypatch, navigated_decision, invalidate):
+    from test_codex_context_lifecycle import setup, project, deliver
+
+    event = setup(tmp_path, monkeypatch)
+    source = navigated_decision["interaction_contract"]["agent_channel"]["work_context"]["sources"][0]
+    content = source["content"]
+    outcomes = content["goal_acceptance_contract"]["outcome_requirements"]["outcomes"]
+    for outcome in outcomes:
+        outcome["requirements"][0]["description"] *= 240
+    prefix = "/interaction_contract/agent_channel/work_context/sources/0/content/goal_acceptance_contract/outcome_requirements/outcomes"
+    def navigation(capture, enabled=True):
+        _, wire = project(tmp_path, capture, payload=navigated_decision)
+        view = native_context_capture_view(wire, tmp_path / capture, context_directory=str(tmp_path),
+            progressive_enabled=enabled, max_page_output_bytes=6000)
+        return model_context_reference(view)
+
+    def pages(command):
+        found = []
+        while command:
+            words = shlex.split(command)
+            out = subprocess.check_output([sys.executable, "-m", "loopx.entrypoint", "--format", "json",
+                *words[words.index("quota"):]])
+            page = json.loads(out)
+            assert page["ok"], page
+            found.append(page)
+            command = page.get("next_command")
+        return found
+
+    def facts(pages):
+        return {row["pointer"]: row["value"] for row in json.loads(
+            "".join(page["content"] for page in pages))["selections"]}
+
+    first = pages(navigation("first")["first_read_command"])
+    assert len(first) > 1 and facts(first)[prefix + "/0"] == outcomes[0]
+    deliver(tmp_path, event, first[1:] if invalidate == "partial" else first)
+    outcomes[1]["requirements"][0]["description"] = "Revised stop: require the audited result."
+    if invalidate == "task_switch":
+        navigated_decision["selected_todo"]["todo_id"] = content["outcome_context"]["todo_id"] = "todo_next"
+    second = navigation("next", invalidate != "off")
+    warm = pages(second["first_read_command"])
+    observed = facts(warm)
+    assert observed[prefix + "/1"] == outcomes[1]
+    assert observed["/goal_boundary"] == navigated_decision["goal_boundary"]
+    if invalidate != "compact":
+        assert observed[prefix + "/0"] == outcomes[0]
+        assert "previously_delivered_fields" not in warm[0]
+        return
+    assert warm[0]["previously_delivered_fields"] == [prefix + "/0"]
+    assert observed[prefix + "/0/id"] == outcomes[0]["id"] and prefix + "/0" not in observed
+    assert facts(pages(second["recovery"]["command"] + " --select " + prefix + "/0"))[prefix + "/0"] == outcomes[0]
+    codex_context.codex_context_hook(str(tmp_path), {**event, "hook_event_name": "PostCompact"})
+    recovered = pages(second["first_read_command"])
+    assert facts(recovered)[prefix + "/0"] == outcomes[0]

From b1f433dad5886d1df1d1ddcd297ff29803cb9676 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 5 Nov 2026 22:28:52 +0800
Subject: [PATCH 2/6] Plan field reuse inside the existing delivery transaction
 and qualify invalidation

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/cli_commands/quota_capture.py           | 17 ++++++--------
 .../quota/progressive_context.ts              | 16 +++++++++----
 loopx/control_plane/quota/turn_envelope.py    |  6 +++--
 loopx/control_plane/quota/turn_envelope.ts    | 18 +++++++--------
 .../turn_driver/codex_context.py              | 15 ++++++++----
 .../turn_driver/native_first_read.py          | 16 ++++---------
 .../control_plane/test_context_first_read.py  | 23 +++++++++++++++----
 .../test_native_context_delivery_cost.py      |  4 ++--
 8 files changed, 67 insertions(+), 48 deletions(-)

diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index 9ac04be4fb..6cf9d2778e 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -567,17 +567,14 @@ def view(kind: str, ref: dict[str, Any], body: bytes) -> dict[str, Any]:
         write_decision_capture(wire, projected)
     from ..control_plane.turn_driver.codex_context import register_native_context_view
 
-    seen = register_native_context_view(context_directory,
+    observed = register_native_context_view(context_directory,
         {**result, "ancestry": ancestry}, hook_session=hook_session)
-    if progressive_enabled and seen and initial_read_pointers and "first_read_command" in result:
-        from ..control_plane.turn_driver.native_first_read import observed_first_read_reference
-
-        first = observed_first_read_reference(context_directory, result["target"], current,
-            current_document, initial_read_pointers, seen, output_bytes)
-        if first is not None:
-            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)
+    if (progressive_enabled and observed and "first_read_command" in result
+            and observed["pointer_bytes"] <= 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)
     return result
 
 
diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts
index 5b0aa34721..1d726a21ff 100644
--- a/loopx/control_plane/quota/progressive_context.ts
+++ b/loopx/control_plane/quota/progressive_context.ts
@@ -9,6 +9,7 @@ import {FULL_OUTCOME_READ_GUIDANCE, PROGRESSIVE_OUTCOME_READ_GUIDANCE} from "../
 import {quotaSummaryReferences, contextSourceIdentity, contextProjectionChanges,
   type ProjectionSourceValues} from "./diagnostic_projection.ts";
 import {encodeContextDelivery, fileNavigationPaths} from "./context_delivery.ts";
+import {capturedContextSelection} from "./context_selection.ts";
 import {createHash} from "node:crypto";
 
 type Path = (string | number)[];
@@ -29,10 +30,10 @@ export function contextReadSectionDigests(request: JsonObject, paths: Path[]): J
 
 /** Only previously delivered supplementary bodies may use a reference. Current
  * decisions, stops, selection, settlement and unknown source shapes stay inline. */
-export function observedFirstRead(request: JsonObject, selections: {pointer: string; path: Path}[]): JsonObject {
+function observedFirstRead(request: JsonObject, selections: {pointer: string; path: Path}[]): 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 {selections, reused: {}};
+  if (typeof jsonObject(context.selected_todo)?.todo_id !== "string") return undefined;
   const sources = ["interaction_contract", "agent_channel", "work_context", "sources"];
   const identity = contextSourceIdentity(request, request.source_number_tokens);
   const reused: JsonObject = {};
@@ -55,7 +56,7 @@ export function observedFirstRead(request: JsonObject, selections: {pointer: str
     return (outcome ? ["id", "revision"] : ["ok", "revision", "instruction"]).filter(key =>
       Object.hasOwn(body, key)).map(key => ({pointer: address + "/" + key, path: [...path, key]}));
   });
-  return {selections: current, reused};
+  return Object.keys(reused).length ? {selections: current, reused} : undefined;
 }
 
 function inlineReadSections(request: JsonObject, required: JsonObject): JsonObject {
@@ -270,14 +271,21 @@ 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[]): JsonObject {
+export function prepareProgressiveContextDelivery(request: JsonObject, legacyFields: readonly string[],
+    decisionFields: readonly string[]): JsonObject {
   const sourceValues: ProjectionSourceValues = new WeakMap();
   const required = requiredNativeContext(request, sourceValues);
   const codec = nativeContextContinuity({...request, context: required.context,
     ...(request.source_number_tokens === undefined ? {} : {source_number_tokens: required.tokens})});
   const delivery = encodeContextDelivery(codec, legacyFields, sourceValues, fileNavigationPaths(required.context));
+  const observe = request.observe_first_read === true && !!jsonObject(required.context.selected_todo)?.todo_id;
+  const pointers = observe ? initialContextReadPointers(required.context, decisionFields) : [];
+  const selections = pointers.length
+    ? capturedContextSelection({context: required.context, initial_read: true}, pointers).selections : undefined;
+  const observed = observe && selections ? observedFirstRead(codec, selections as {pointer: string; path: Path}[]) : 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)}, sourceValues);
 }
 
diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py
index e43c1968ac..bbfa2aaa54 100644
--- a/loopx/control_plane/quota/turn_envelope.py
+++ b/loopx/control_plane/quota/turn_envelope.py
@@ -60,7 +60,7 @@ def _turn_envelope_result(
         tokens = source_number_tokens(params["payload"])
         if tokens:
             request["source_number_tokens"] = tokens
-    elif operation in {"context_delivery", "context_recovery", "progressive_context_delivery", "context_read_sections", "observed_first_read"}:
+    elif operation in {"context_delivery", "context_recovery", "progressive_context_delivery", "context_read_sections"}:
         tokens = source_number_tokens(params)
         if tokens:
             request["source_number_tokens"] = tokens
@@ -184,7 +184,7 @@ def prepare_context_delivery(
     retained: Mapping[str, Any] | None = None,
     continuity: Mapping[str, Any] | None = None,
     surface: str = "work_context", native_projection: Mapping[str, Any] | None = None,
-    consumed_pointers: Mapping[str, str] | None = None,
+    consumed_pointers: Mapping[str, str] | None = None, observe_first_read: bool = False,
 ) -> dict[str, Any]:
     """Prepare under the Host's explicit continuity basis, not a model assertion.
 
@@ -203,6 +203,8 @@ def prepare_context_delivery(
         source["native_projection"] = dict(native_projection)
     if consumed_pointers is not None:
         source["consumed_pointers"] = dict(consumed_pointers)
+        if consumed_pointers and observe_first_read:
+            source["observe_first_read"] = True
     result = _turn_envelope_result("context_delivery", **source, source_edits=True)
     return _apply_context_projection_changes(source, result)
 
diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts
index d21659e1ac..d41cfa2129 100644
--- a/loopx/control_plane/quota/turn_envelope.ts
+++ b/loopx/control_plane/quota/turn_envelope.ts
@@ -1,5 +1,5 @@
 import {capturedContextSelection} from "./context_selection.ts";
-import {observedFirstRead, contextReadSectionDigests, initialContextReadPointers, prepareProgressiveContext, prepareProgressiveContextDelivery} from "./progressive_context.ts";
+import {contextReadSectionDigests, initialContextReadPointers, prepareProgressiveContext, prepareProgressiveContextDelivery} from "./progressive_context.ts";
 import {contextProjectionChanges, contextSourceIdentity, projectQuotaDiagnostics} from "./diagnostic_projection.ts";
 import type { LoopXTurnRouteValue } from "../turn_driver/turn_contract_generated.ts";
 import { EffectiveAction, type QuotaEffectiveActionValue } from "./effective_action.generated.ts";
@@ -1517,13 +1517,15 @@ export function turnEnvelopeRoute(envelope: JsonObject, historicalText?: JsonObj
 /** Navigation over captured facts, never a summary or a source-read receipt. */
 // Quota field vocabulary stays with its decision owner; the RFC 6901 reader
 // receives addresses and remains independent of Goal/permission policy.
+const INITIAL_CONTEXT_DECISION_FIELDS = ["ok", "schema_version", "goal_id", "agent_id", "should_run", "effective_action",
+    "normal_delivery_allowed", "recovery_delivery_allowed", "self_repair_allowed", "capability_repair_allowed", "workspace_repair_allowed",
+    "safe_bypass_allowed", "safe_bypass_kind", "blocked_action_scope", "reason",
+    "execution_obligation", "goal_boundary", "heartbeat_receipt", "work_lane_contract", "selected_todo"];
+
 function capturedDecisionSelection(request: JsonObject): JsonObject {
   const initial = request.initial_read === true || request.list_fields === true
     && Array.isArray(request.select) && request.select.length === 1 && request.select[0] === "";
-  return capturedContextSelection(request, initial ? initialContextReadPointers(requireJsonObject(request.context, "captured context"), ["ok", "schema_version", "goal_id", "agent_id", "should_run", "effective_action",
-    "normal_delivery_allowed", "recovery_delivery_allowed", "self_repair_allowed", "capability_repair_allowed", "workspace_repair_allowed",
-    "safe_bypass_allowed", "safe_bypass_kind", "blocked_action_scope", "reason",
-    "execution_obligation", "goal_boundary", "heartbeat_receipt", "work_lane_contract", "selected_todo"]) : []);
+  return capturedContextSelection(request, initial ? initialContextReadPointers(requireJsonObject(request.context, "captured context"), INITIAL_CONTEXT_DECISION_FIELDS) : []);
 }
 
 export function evaluateTurnEnvelope(value: unknown): JsonObject {
@@ -1598,10 +1600,6 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
     return contextProjectionChanges(source, payload);
   }
   if (operation === "build") return buildTurnEnvelope(request);
-  if (operation === "observed_first_read") {
-    const selected = capturedContextSelection({...request, preserve_selection_order: true});
-    return observedFirstRead(request, selected.selections as {pointer: string; path: (string | number)[]}[]);
-  }
   if (operation === "context_read_sections") {
     const selected = capturedContextSelection({...request, list_fields: false});
     return {sections: contextReadSectionDigests(request,
@@ -1610,7 +1608,7 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
   if (operation === "progressive_context_delivery") return prepareProgressiveContext(request);
   if (operation === "context_delivery") return request.native_projection === undefined
     ? prepareContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS)
-    : prepareProgressiveContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS);
+    : prepareProgressiveContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS, INITIAL_CONTEXT_DECISION_FIELDS);
   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/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py
index 7481a7874f..4f21a0ea57 100644
--- a/loopx/control_plane/turn_driver/codex_context.py
+++ b/loopx/control_plane/turn_driver/codex_context.py
@@ -578,10 +578,10 @@ def _native_context_request(state: dict[str, Any], scope: dict[str, Any]) -> dic
     return request
 
 
-def register_native_context_view(directory: str | None, view: dict[str, Any], *, hook_session: str | None = None) -> dict[str, str]:
+def register_native_context_view(directory: str | None, view: dict[str, Any], *, hook_session: str | None = None) -> dict | None:
     """Bind the existing pending delivery to its actual paged wire, under lock."""
     if directory is None or not isinstance(view.get("target"), dict):
-        return {}
+        return None
     with _locked(directory) as root:
         state = _state(root)
         pending = state.get("pending")
@@ -592,7 +592,7 @@ def register_native_context_view(directory: str | None, view: dict[str, Any], *,
             ancestry=view.get("ancestry", 0),
             read_sections=pending.get("full_read_sections", {}) if view.get("mode") == "full" else {})
         _save(root, state)
-        return state.get("consumed_pointers", {})
+        return pending.get("observed_first_read")
 
 
 def project_codex_context(payload: dict[str, Any], *, directory: str | None, capture_directory: Path,
@@ -675,7 +675,7 @@ def publish_recovery(reference: dict[str, Any]) -> None:
                             "command_template": context_capture_read_command(Path(recovery["path"]), recovery["sha256"],
                                 page_bytes=page_bytes, max_page_output_bytes=output_bytes)
                                 + " --select '<supporting_material_pointer>/sources'",
-                        }, consumed_pointers=state.get("consumed_pointers"))
+                        }, consumed_pointers=state.get("consumed_pointers"), observe_first_read=progressive_enabled)
                 except (EffectRuntimeRemoteError, EffectRuntimeStartupError):
                     return payload
                 if transaction["projected"]:
@@ -692,6 +692,13 @@ def publish_recovery(reference: dict[str, Any]) -> None:
                     state["pending"].update(target=result["target"], ancestry=result["ancestry"],
                         full_read_sections=transaction["read_sections"],
                         read_sections=transaction["read_sections"] if result["mode"] == "full" else {})
+                    if plan := transaction.get("observed_first_read"):
+                        from .native_first_read import observed_first_read_reference
+
+                        reference = observed_first_read_reference(directory, result["target"], recovery,
+                            transaction["context"], plan)
+                        if reference is not None:
+                            state["pending"]["observed_first_read"] = reference
                     _save(root, state)
                 return result
             except (OSError, ValueError, KeyError, TypeError, RecursionError):
diff --git a/loopx/control_plane/turn_driver/native_first_read.py b/loopx/control_plane/turn_driver/native_first_read.py
index 63bd77ccfc..8e30cba7d4 100644
--- a/loopx/control_plane/turn_driver/native_first_read.py
+++ b/loopx/control_plane/turn_driver/native_first_read.py
@@ -6,27 +6,21 @@
 from pathlib import Path
 from typing import Any
 
-from ..effect_runtime import EffectRuntimeRemoteError, EffectRuntimeStartupError
-from ..quota.turn_envelope import _turn_envelope_result
 from .decision_capture import context_selection_bytes, write_context_read_reference
 
 
 def observed_first_read_reference(directory: str, target: dict, recovery: dict, context: dict,
-                                  pointers: list[str], seen: dict[str, str], output_budget: int) -> Path | None:
+                                  plan: dict) -> dict | None:
+    """Write the typed delivery plan once. The host only renders original Python values."""
     try:
-        plan = _turn_envelope_result("observed_first_read", context=context, select=pointers, consumed_pointers=seen)
-        if not plan["reused"] or not plan["selections"]:
-            return None
-        # Keep each bounded page usable even when many outcomes match.
-        if sum(len(key.encode()) + 4 for key in plan["reused"]) > output_budget // 4:
-            return None
         selected = plan["selections"]
         raw = context_selection_bytes(context, selected, list_fields=False, list_depth=None)
-        return write_context_read_reference(Path(recovery["path"]), recovery["sha256"], selection={
+        reference = write_context_read_reference(Path(recovery["path"]), recovery["sha256"], selection={
             "select": [item["pointer"] for item in selected], "initial_read": False,
             "list_fields": False, "list_depth": None, "sha256": hashlib.sha256(raw).hexdigest()},
             observed_read={"directory": directory, "target": target, "fields": plan["reused"]})
-    except (OSError, ValueError, EffectRuntimeRemoteError, EffectRuntimeStartupError):
+        return {"path": str(reference), "pointer_bytes": sum(len(key.encode()) + 4 for key in plan["reused"])}
+    except (OSError, ValueError):
         return None  # The ordinary first read remains available.
 
 
diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py
index 8b2de20ca1..550f5041c8 100644
--- a/tests/control_plane/test_context_first_read.py
+++ b/tests/control_plane/test_context_first_read.py
@@ -669,9 +669,9 @@ def test_warm_delta_initial_read_uses_current_action_without_the_old_base(tmp_pa
     assert not registry.exists() and not runtime.exists()
 
 
-@pytest.mark.parametrize("invalidate", ["compact", "partial", "task_switch", "off"])
+@pytest.mark.parametrize("invalidate", ["compact", "resume", "partial", "task_switch", "off"])
 def test_native_first_read_reuses_only_observed_current_fields(tmp_path, monkeypatch, navigated_decision, invalidate):
-    from test_codex_context_lifecycle import setup, project, deliver
+    from test_codex_context_lifecycle import setup, deliver
 
     event = setup(tmp_path, monkeypatch)
     source = navigated_decision["interaction_contract"]["agent_channel"]["work_context"]["sources"][0]
@@ -681,7 +681,11 @@ def test_native_first_read_reuses_only_observed_current_fields(tmp_path, monkeyp
         outcome["requirements"][0]["description"] *= 240
     prefix = "/interaction_contract/agent_channel/work_context/sources/0/content/goal_acceptance_contract/outcome_requirements/outcomes"
     def navigation(capture, enabled=True):
-        _, wire = project(tmp_path, capture, payload=navigated_decision)
+        directory = tmp_path / capture
+        directory.mkdir(mode=0o700)
+        wire = codex_context.project_codex_context(navigated_decision, directory=str(tmp_path),
+            capture_directory=directory, goal_id="fixture", agent_id="agent", surface="quota_packet",
+            capture_id=capture, progressive_enabled=enabled)
         view = native_context_capture_view(wire, tmp_path / capture, context_directory=str(tmp_path),
             progressive_enabled=enabled, max_page_output_bytes=6000)
         return model_context_reference(view)
@@ -702,6 +706,7 @@ def facts(pages):
         return {row["pointer"]: row["value"] for row in json.loads(
             "".join(page["content"] for page in pages))["selections"]}
 
+    navigated_decision["goal_boundary"]["stop_condition"] = "Preserve every applicable stop. " * 220
     first = pages(navigation("first")["first_read_command"])
     assert len(first) > 1 and facts(first)[prefix + "/0"] == outcomes[0]
     deliver(tmp_path, event, first[1:] if invalidate == "partial" else first)
@@ -713,13 +718,21 @@ def facts(pages):
     observed = facts(warm)
     assert observed[prefix + "/1"] == outcomes[1]
     assert observed["/goal_boundary"] == navigated_decision["goal_boundary"]
-    if invalidate != "compact":
+    if invalidate not in {"compact", "resume"}:
         assert observed[prefix + "/0"] == outcomes[0]
         assert "previously_delivered_fields" not in warm[0]
         return
     assert warm[0]["previously_delivered_fields"] == [prefix + "/0"]
     assert observed[prefix + "/0/id"] == outcomes[0]["id"] and prefix + "/0" not in observed
     assert facts(pages(second["recovery"]["command"] + " --select " + prefix + "/0"))[prefix + "/0"] == outcomes[0]
-    codex_context.codex_context_hook(str(tmp_path), {**event, "hook_event_name": "PostCompact"})
+    lifecycle = {"hook_event_name": "PostCompact"} if invalidate == "compact" else {"source": "resume"}
+    codex_context.codex_context_hook(str(tmp_path), {**event, **lifecycle})
+    assert len(warm) > 1
+    stale = subprocess.run(shlex.split(warm[0]["next_command"]), capture_output=True, text=True)
+    assert stale.returncode != 0
+    rejection = json.loads(stale.stdout)
+    assert "content" not in rejection
+    restarted = pages(rejection["recovery_command"])
+    assert facts(restarted)[prefix + "/0"] == outcomes[0]
     recovered = pages(second["first_read_command"])
     assert facts(recovered)[prefix + "/0"] == outcomes[0]
diff --git a/tests/control_plane/test_native_context_delivery_cost.py b/tests/control_plane/test_native_context_delivery_cost.py
index 996095af67..131de4fad7 100644
--- a/tests/control_plane/test_native_context_delivery_cost.py
+++ b/tests/control_plane/test_native_context_delivery_cost.py
@@ -107,9 +107,9 @@ def test_small_codec_delta_uses_cheaper_complete_native_delivery(tmp_path, monke
     index_requests = []
     request = turn_envelope._turn_envelope_result
 
-    def observe_registration(directory, view):
+    def observe_registration(directory, view, **kwargs):
         registrations.append((directory, copy.deepcopy(view)))
-        return register(directory, view)
+        return register(directory, view, **kwargs)
 
     def observe_index_request(operation, **kwargs):
         if operation == "context_select":

From 43856e258eca668b03bb4367dca804f881948437 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 5 Nov 2026 22:39:55 +0800
Subject: [PATCH 3/6] Type the mixed native delivery request for the kernel
 contract check

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/control_plane/quota/turn_envelope.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py
index bbfa2aaa54..311eea5963 100644
--- a/loopx/control_plane/quota/turn_envelope.py
+++ b/loopx/control_plane/quota/turn_envelope.py
@@ -195,7 +195,7 @@ def prepare_context_delivery(
     """
     from .cli_projection import _apply_context_projection_changes
 
-    source = dict(context=dict(context), scope=dict(scope), capture_id=capture_id, surface=surface,
+    source: dict[str, Any] = dict(context=dict(context), scope=dict(scope), capture_id=capture_id, surface=surface,
         baseline=dict(baseline) if baseline is not None else None,
         retained=dict(retained) if retained is not None else None,
         continuity=dict(continuity) if continuity is not None else None)

From 40928d668df3c5847b58c1f1d7e95d68f9c49f99 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 5 Nov 2026 22:41:21 +0800
Subject: [PATCH 4/6] Keep private field observation descriptors typed on both
 native routes

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/control_plane/turn_driver/codex_context.py     | 6 ++++--
 loopx/control_plane/turn_driver/native_first_read.py | 4 ++--
 2 files changed, 6 insertions(+), 4 deletions(-)

diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py
index 4f21a0ea57..d8899f0011 100644
--- a/loopx/control_plane/turn_driver/codex_context.py
+++ b/loopx/control_plane/turn_driver/codex_context.py
@@ -578,7 +578,8 @@ def _native_context_request(state: dict[str, Any], scope: dict[str, Any]) -> dic
     return request
 
 
-def register_native_context_view(directory: str | None, view: dict[str, Any], *, hook_session: str | None = None) -> dict | None:
+def register_native_context_view(directory: str | None, view: dict[str, Any], *,
+                                 hook_session: str | None = None) -> dict[str, Any] | None:
     """Bind the existing pending delivery to its actual paged wire, under lock."""
     if directory is None or not isinstance(view.get("target"), dict):
         return None
@@ -592,7 +593,8 @@ def register_native_context_view(directory: str | None, view: dict[str, Any], *,
             ancestry=view.get("ancestry", 0),
             read_sections=pending.get("full_read_sections", {}) if view.get("mode") == "full" else {})
         _save(root, state)
-        return pending.get("observed_first_read")
+        observed = pending.get("observed_first_read")
+        return observed if isinstance(observed, dict) else None
 
 
 def project_codex_context(payload: dict[str, Any], *, directory: str | None, capture_directory: Path,
diff --git a/loopx/control_plane/turn_driver/native_first_read.py b/loopx/control_plane/turn_driver/native_first_read.py
index 8e30cba7d4..a24b2421eb 100644
--- a/loopx/control_plane/turn_driver/native_first_read.py
+++ b/loopx/control_plane/turn_driver/native_first_read.py
@@ -9,8 +9,8 @@
 from .decision_capture import context_selection_bytes, write_context_read_reference
 
 
-def observed_first_read_reference(directory: str, target: dict, recovery: dict, context: dict,
-                                  plan: dict) -> dict | None:
+def observed_first_read_reference(directory: str, target: dict[str, Any], recovery: dict[str, Any],
+                                  context: dict[str, Any], plan: dict[str, Any]) -> dict[str, Any] | None:
     """Write the typed delivery plan once. The host only renders original Python values."""
     try:
         selected = plan["selections"]

From 0fc2a5e87fdbbc8bbf6c22aa6d70084592fdf132 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Fri, 6 Nov 2026 11:54:09 +0800
Subject: [PATCH 5/6] fix: observe reusable fields inside complete parent reads

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../quota/progressive_context.ts              | 64 +++++++++++++------
 .../control_plane/test_context_first_read.py  | 16 ++++-
 2 files changed, 59 insertions(+), 21 deletions(-)

diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts
index 1d726a21ff..5669572c83 100644
--- a/loopx/control_plane/quota/progressive_context.ts
+++ b/loopx/control_plane/quota/progressive_context.ts
@@ -13,50 +13,80 @@ import {capturedContextSelection} from "./context_selection.ts";
 import {createHash} from "node:crypto";
 
 type Path = (string | number)[];
+const MIN_REUSABLE_BODY_BYTES = 1024;
 const pointer = (path: Path): string => path.length ? "/" + path.map(key =>
   String(key).replaceAll("~", "~0").replaceAll("/", "~1")).join("/") : "";
 const at = (root: unknown, path: Path): unknown => path.reduce<unknown>((row, key) =>
   Array.isArray(row) ? row[Number(key)] : jsonObject(row)?.[key], root);
 const known = (row: JsonObject, fields: string[]): boolean => Object.keys(row).every(key => fields.includes(key));
 
+/** 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): {path: Path; freshKeys: string[]}[] {
+  const base: Path = ["interaction_contract", "agent_channel", "work_context", "sources"];
+  const sources = at(context, base);
+  if (!Array.isArray(sources)) return [];
+  return sources.flatMap((value, i) => {
+    const source = jsonObject(value), content = jsonObject(source?.content);
+    if (!content) return [];
+    const path = [...base, i, "content"];
+    if (source?.kind === "agent_preferences")
+      return [{path, freshKeys: ["ok", "revision", "instruction"]}];
+    if (source?.source !== "goal_acceptance") return [];
+    const outcomesPath = [...path, "goal_acceptance_contract", "outcome_requirements", "outcomes"];
+    const outcomes = at(context, outcomesPath);
+    return Array.isArray(outcomes) ? outcomes.flatMap((outcome, index) =>
+      typeof jsonObject(outcome)?.id === "string"
+        ? [{path: [...outcomesPath, index], freshKeys: ["id", "revision"]}] : []) : [];
+  });
+}
+
 /** 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 {
   const context = requireJsonObject(request.context, "read section source");
   const identity = contextSourceIdentity(request, request.source_number_tokens);
-  return Object.fromEntries(paths.map(path => [pointer(path),
-    createHash("sha256").update(identity(at(context, path), ["context", ...path])).digest("hex")]));
+  const selected = new Set(paths.map(pointer));
+  const covered = new Map(paths.map(path => [pointer(path), path]));
+  for (const {path} of reusableFirstReadFields(context))
+    if (paths.some(parent => pathStartsWith(path, parent))) covered.set(pointer(path), path);
+  return Object.fromEntries([...covered].flatMap(([address, path]) => {
+    const content = identity(at(context, path), ["context", ...path]);
+    // Parent delivery must not multiply tiny bodies into a larger journal.
+    return selected.has(address) || Buffer.byteLength(content) >= MIN_REUSABLE_BODY_BYTES
+      ? [[address, createHash("sha256").update(content).digest("hex")]] : [];
+  }));
 }
 
 /** Only previously delivered supplementary bodies may use a reference. Current
  * decisions, stops, selection, settlement and unknown source shapes stay inline. */
-function observedFirstRead(request: JsonObject, selections: {pointer: string; path: Path}[]): JsonObject | undefined {
+function observedFirstRead(request: JsonObject, pointers: string[]): 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 sources = ["interaction_contract", "agent_channel", "work_context", "sources"];
+  const fields = new Map(reusableFirstReadFields(context).map(field => [pointer(field.path), field]));
   const identity = contextSourceIdentity(request, request.source_number_tokens);
   const reused: JsonObject = {};
+  // Match semantic fields before the reader groups complete parents. Otherwise
+  // selecting every outcome hides the individual unchanged bodies from reuse.
+  const selections = capturedContextSelection({context, select: pointers,
+    preserve_selection_order: true}).selections as {pointer: string; path: Path}[];
   const current = selections.flatMap(row => {
     const {path, pointer: address} = row;
-    if (typeof seen[address] !== "string" || !pathStartsWith(path, sources)
-        || typeof path[4] !== "number" || path[5] !== "content") return [row];
-    const source = jsonObject(at(context, path.slice(0, 5)));
-    const preference = source?.kind === "agent_preferences" && path.length === 6;
-    const outcome = source?.source === "goal_acceptance" && path.length === 10
-      && path[6] === "goal_acceptance_contract" && path[7] === "outcome_requirements"
-      && path[8] === "outcomes" && typeof path[9] === "number";
+    const field = fields.get(address);
+    if (typeof seen[address] !== "string" || !field) return [row];
     const body = jsonObject(at(context, path));
-    if (!body || !(preference || outcome) || (outcome && typeof body.id !== "string")) return [row];
+    if (!body) return [row];
     const content = identity(body, ["context", ...path]);
     // A reference and its recovery instructions should cost less than the field.
-    if (Buffer.byteLength(content) < 1024 || createHash("sha256").update(content).digest("hex") !== seen[address])
+    if (Buffer.byteLength(content) < MIN_REUSABLE_BODY_BYTES || createHash("sha256").update(content).digest("hex") !== seen[address])
       return [row];
     reused[address] = seen[address];
-    return (outcome ? ["id", "revision"] : ["ok", "revision", "instruction"]).filter(key =>
+    return field.freshKeys.filter(key =>
       Object.hasOwn(body, key)).map(key => ({pointer: address + "/" + key, path: [...path, key]}));
   });
-  return Object.keys(reused).length ? {selections: current, reused} : undefined;
+  return Object.keys(reused).length ? {selections: capturedContextSelection({context, initial_read: true},
+    current.map(row => row.pointer)).selections, reused} : undefined;
 }
 
 function inlineReadSections(request: JsonObject, required: JsonObject): JsonObject {
@@ -280,9 +310,7 @@ export function prepareProgressiveContextDelivery(request: JsonObject, legacyFie
   const delivery = encodeContextDelivery(codec, legacyFields, sourceValues, fileNavigationPaths(required.context));
   const observe = request.observe_first_read === true && !!jsonObject(required.context.selected_todo)?.todo_id;
   const pointers = observe ? initialContextReadPointers(required.context, decisionFields) : [];
-  const selections = pointers.length
-    ? capturedContextSelection({context: required.context, initial_read: true}, pointers).selections : undefined;
-  const observed = observe && selections ? observedFirstRead(codec, selections as {pointer: string; path: Path}[]) : undefined;
+  const observed = 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} : {}),
diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py
index 550f5041c8..b8a6552bd4 100644
--- a/tests/control_plane/test_context_first_read.py
+++ b/tests/control_plane/test_context_first_read.py
@@ -670,15 +670,24 @@ def test_warm_delta_initial_read_uses_current_action_without_the_old_base(tmp_pa
 
 
 @pytest.mark.parametrize("invalidate", ["compact", "resume", "partial", "task_switch", "off"])
-def test_native_first_read_reuses_only_observed_current_fields(tmp_path, monkeypatch, navigated_decision, invalidate):
+@pytest.mark.parametrize("coalesced", [False, True])
+def test_native_first_read_reuses_only_observed_current_fields(tmp_path, monkeypatch, navigated_decision, invalidate, coalesced):
     from test_codex_context_lifecycle import setup, deliver
 
     event = setup(tmp_path, monkeypatch)
     source = navigated_decision["interaction_contract"]["agent_channel"]["work_context"]["sources"][0]
     content = source["content"]
     outcomes = content["goal_acceptance_contract"]["outcome_requirements"]["outcomes"]
+    if coalesced:
+        # All outcomes are relevant, so the reader may package them as one
+        # complete parent. Delivery still observes each complete outcome.
+        outcomes.pop()
     for outcome in outcomes:
         outcome["requirements"][0]["description"] *= 240
+    if coalesced:
+        outcomes.append({"id": "small", "owner_agent_id": "author", "revision": 1,
+            "requirements": [{"id": "short", "description": "Keep this short current fact."}]})
+        content["outcome_context"]["suggested_outcome_ids"].append("small")
     prefix = "/interaction_contract/agent_channel/work_context/sources/0/content/goal_acceptance_contract/outcome_requirements/outcomes"
     def navigation(capture, enabled=True):
         directory = tmp_path / capture
@@ -703,13 +712,14 @@ def pages(command):
         return found
 
     def facts(pages):
-        return {row["pointer"]: row["value"] for row in json.loads(
-            "".join(page["content"] for page in pages))["selections"]}
+        return selected_context_values("".join(page["content"] for page in pages))
 
     navigated_decision["goal_boundary"]["stop_condition"] = "Preserve every applicable stop. " * 220
     first = pages(navigation("first")["first_read_command"])
     assert len(first) > 1 and facts(first)[prefix + "/0"] == outcomes[0]
     deliver(tmp_path, event, first[1:] if invalidate == "partial" else first)
+    if coalesced:
+        assert prefix + "/2" not in codex_context._state(tmp_path).get("consumed_pointers", {})
     outcomes[1]["requirements"][0]["description"] = "Revised stop: require the audited result."
     if invalidate == "task_switch":
         navigated_decision["selected_todo"]["todo_id"] = content["outcome_context"]["todo_id"] = "todo_next"

From 1003ff12c78b31d19fcb75a20479a1312b644b26 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Fri, 6 Nov 2026 12:16:05 +0800
Subject: [PATCH 6/6] test: distinguish progressive defaults from delta opt-out

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../test_saved_context_project.py             | 51 ++++++++++++++-----
 1 file changed, 38 insertions(+), 13 deletions(-)

diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py
index 3d1cc25ed1..3b36783c2e 100644
--- a/tests/control_plane/test_saved_context_project.py
+++ b/tests/control_plane/test_saved_context_project.py
@@ -13,7 +13,8 @@
 
 
 @pytest.mark.parametrize("failure", [None, "digest", "goal", "agent", "turn", "symlink", "object"])
-def test_saved_projection_checks_original_before_any_publication(tmp_path, failure):
+@pytest.mark.parametrize("setting", [None, "auto", "off"])
+def test_saved_projection_checks_original_before_any_publication(tmp_path, failure, setting):
     original = {"ok": True, "goal_id": "saved-goal", "agent_identity": {"agent_id": "saved-agent"},
         "heartbeat_receipt": {"turn_instance_id": "saved-turn"}, "requirement": "保留完整要求🙂" * 100}
     path = tmp_path / "original.json"
@@ -27,22 +28,38 @@ def test_saved_projection_checks_original_before_any_publication(tmp_path, failu
         path.symlink_to(source)
     args = ["quota", "context-project", "--capture-file", str(path), "--capture-sha256", digest,
         "--goal-id", "saved-goal", "--agent-id", "saved-agent", "--turn-instance-id", "saved-turn"]
+    if setting is not None:
+        args.extend(("--context-delivery", setting))
     for name in ("digest", "goal", "agent", "turn"):
         if failure == name:
             option = {"digest": "--capture-sha256", "goal": "--goal-id", "agent": "--agent-id", "turn": "--turn-instance-id"}[name]
             args[args.index(option) + 1] = "0" * 64 if name == "digest" else "foreign"
     registry, runtime = tmp_path / "absent-registry.json", tmp_path / "absent-runtime"
     code, result = _run_cli(registry, runtime, *args, cwd=tmp_path)
-    assert not registry.exists() and not runtime.exists()
+    assert not registry.exists()
+    if failure or setting == "off":
+        assert not runtime.exists()
+    else:
+        # Main's default Progressive presentation may create private transport;
+        # it may not create authority, evaluate a guard or debit the saved Turn.
+        if runtime.exists():
+            assert {item.name for item in runtime.iterdir()} <= {"context-delivery"}
     if failure:
         assert code != 0 and not result["ok"] and "command" not in result
     else:
-        assert code == 0 and result["ok"] and result["mode"] == "full"
-        assert result["path"] == str(path) and result["sha256"] == digest
-        assert result["recovery"]["path"] == str(path)
-        command = shlex.split(result["command"])
-        code, page = _run_cli(registry, runtime, *command[3:], cwd=tmp_path)
-        assert code == 0 and page["content"].encode() == path.read_bytes()[:page["next_offset"]]
+        assert code == 0 and result["ok"]
+        if setting == "off":
+            assert result["mode"] == "full"
+            assert result["path"] == str(path) and result["sha256"] == digest
+            assert result["recovery"]["path"] == str(path)
+            command = shlex.split(result["command"])
+            code, page = _run_cli(registry, runtime, *command[3:], cwd=tmp_path)
+            assert code == 0 and page["content"].encode() == path.read_bytes()[:page["next_offset"]]
+        else:
+            from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload
+
+            assert read_cli_policy_payload(json.dumps(result), returncode=code) == original
+        assert hashlib.sha256(path.read_bytes()).hexdigest() == digest
 
 
 @pytest.mark.parametrize("root_kind", ["unsafe", "symlink", "missing_parent"])
@@ -920,7 +937,8 @@ def request(method, params):
     assert not list(home.rglob("*.pack")), "discovery must not clone a plugin catalog"
 
 
-def test_saved_projection_default_off_ignores_available_native_cache(tmp_path, monkeypatch):
+@pytest.mark.parametrize("setting", [None, "off"])
+def test_saved_projection_default_delta_off_ignores_available_native_cache(tmp_path, monkeypatch, setting):
     from loopx.cli import build_parser
     from loopx.cli_commands.quota_capture import project_saved_context_capture
     from loopx.control_plane.turn_driver import codex_context
@@ -930,13 +948,20 @@ def test_saved_projection_default_off_ignores_available_native_cache(tmp_path, m
         "heartbeat_receipt": {"turn_instance_id": "t"}}).encode()
     path.write_bytes(raw)
     def unexpected(*args, **kwargs):
-        pytest.fail("default-off saved presentation must not inspect/create native state")
+        pytest.fail("stateless/disabled presentation must not inspect/create native state")
     monkeypatch.setattr(codex_context, "codex_context_directory", unexpected)
-    args = build_parser().parse_args(["quota", "context-project", "--capture-file", str(path),
+    argv = ["quota", "context-project", "--capture-file", str(path),
         "--capture-sha256", hashlib.sha256(raw).hexdigest(), "--goal-id", "g", "--agent-id", "a",
-        "--turn-instance-id", "t", "--codex-context-dir", str(root), "--decision-output-root", str(root)])
+        "--turn-instance-id", "t", "--codex-context-dir", str(root), "--decision-output-root", str(root)]
+    if setting is not None:
+        argv.extend(("--context-delivery", setting))
+    args = build_parser().parse_args(argv)
     result = project_saved_context_capture(args)
-    assert result["ok"] and result["mode"] == "full" and result["path"] == str(path)
+    assert result["ok"]
+    if setting == "off":
+        assert result["mode"] == "full" and result["path"] == str(path)
+    else:
+        assert result == json.loads(raw)
     assert not root.exists() and path.read_bytes() == raw