perf(context): reuse full recovery for native presentation

docs/architecture/rfcs/controller-executor-unified-contract-v0.md

@@ -130,6 +130,19 @@ the existing context-delivery owner. C1 must converge the remaining composition
 and consumer assumptions, rather than introduce another capture service or
 rewrite already shared delta/native-resume logic.
 
+A separate native-presentation slice reuses the committed full recovery as its
+reader target, avoiding an unused second wire publication. Delta and unwrapped
+inputs retain their own immutable files and the same reader/lifecycle owners.
+On a matched 107,289-byte synthetic source, capture writes fall from 216,754 to
+107,289 bytes; full reader bytes and the 3,937-byte warm delta are unchanged.
+This qualifies only duplicate capture IO removal, not model-input reduction,
+complete cost, accepted outcomes or C1 writer/consumer cutover.
+For a small explicit output ceiling, native presentation can also omit repeated
+directory/selection commands while retaining the complete recovery command:
+append `--list` or `--select` to discover/read current facts without a baseline.
+The original 4,000-byte cold/warm/compaction cases retain their requirements,
+Stop, original identity and full recovery; this is transport qualification.
+
 Journal persistence is an independent source-fidelity prerequisite. The proposed
 Python adapter reuses context delivery's sparse number-token collector; both
 TypeScript consumers reuse one source-number validator. Journals remain JSON

docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md

@@ -104,6 +104,16 @@ Quota capture 与 Turn journal capture 都调用
 context-delivery owner。C1 应收敛剩余的组合逻辑与消费者假设，不另建 capture service，
 也不重写已经共享的差分与 native-resume 逻辑。
 
+独立的原生展示切片复用已提交的完整恢复原件作为 reader 目标，避免再次发布未使用的
+wire 副本。Delta 与未包装输入仍保留各自的不可变文件，reader/lifecycle owner 不变。
+同一 107,289 字节合成来源的 capture 写入从 216,754 降为 107,289 字节；完整 reader
+字节及 3,937 字节的 warm delta 均不变。这只验证重复 capture IO 的消除，不证明模型
+输入减少、完整成本收益、成果验收或 C1 writer/consumer 切换。
+显式小预算下，原生展示还可省略重复的目录/选读快捷命令，保留完整恢复命令；
+在其后追加 `--list` 或 `--select`，无须基线即可发现或读取当前事实。
+原有 4,000 字节冷态、温态及压缩恢复用例保留完整要求、Stop、原身份和完整恢复；
+这属于传输验证。
+
 Journal 持久化是独立的来源保真前置。本次提议由 Python adapter 复用 context
 delivery 的稀疏数值 token 收集器；两个 TypeScript 消费者复用同一来源数值校验器。
 Journal 继续传递 JSON 对象，仅为整值浮点、带符号零和非安全整数增加元数据。

docs/reference/required-work-context.md

@@ -443,6 +443,9 @@ capture and return a small file reference, not an eager first page. The response
 names the delta/full `path`, digest and reader `command`; `recovery` names the
 current complete file. A file delta additionally carries `baseline_reference`.
 `index_command` lists current fields and `select_command` batches exact pointers.
+Under a smaller explicit ceiling, these shortcut commands may be omitted;
+append `--list` or `--select <pointer>` to `recovery.command` instead. Both
+operations address the current complete capture, including during delta delivery.
 The reference stays below 7900 serialized bytes (or the explicit smaller output
 ceiling); if metadata cannot fit, recovery retains the current capture. Reader
 pages remain independently bounded. A complete raw tool log is not proof that
@@ -732,7 +735,9 @@ SSH Goal 使用 `--runtime-profile codex_app_ssh_goal` 生成 context-config 和
 不内联第一页。`path`、摘要与 `command` 指向当前差分或全量；`recovery` 指向当前
 完整文件；文件差分另带 `baseline_reference`。`index_command` 展示当前字段目录，
 `select_command` 批量选读精确路径。引用输出不超过 7900 序列化字节或调用者显式更小
-的上限；元数据放不下时保留当前捕获恢复入口。读取页另受输出预算约束。
+的上限；较小预算下可省略这两个快捷命令，改在 `recovery.command` 后追加
+`--list` 或 `--select <pointer>`，差分投递时也直接读取当前完整捕获。
+元数据仍放不下时保留当前捕获恢复入口。读取页另受输出预算约束。
 原始工具日志完整，不证明宿主保留给模型的字节也完整。
 
 原生 `captured_file` 基线来自校验后的前一完整文件，可差分复用所有未变字段，包括

examples/cli-control-plane-command-modularization-smoke.py

@@ -16,6 +16,7 @@
 INIT_PATH = REPO_ROOT / "loopx" / "cli_commands" / "__init__.py"
 TODO_MODULE = REPO_ROOT / "loopx" / "cli_commands" / "todo.py"
 QUOTA_MODULE = REPO_ROOT / "loopx" / "cli_commands" / "quota.py"
+QUOTA_PRESENTATION_MODULE = REPO_ROOT / "loopx" / "cli_commands" / "quota_presentation.py"
 
 
 def run_cli(*args: str) -> subprocess.CompletedProcess[str]:
@@ -41,6 +42,7 @@ def assert_source_shape() -> None:
     init_text = INIT_PATH.read_text(encoding="utf-8")
     todo_text = TODO_MODULE.read_text(encoding="utf-8")
     quota_text = QUOTA_MODULE.read_text(encoding="utf-8")
+    quota_presentation_text = QUOTA_PRESENTATION_MODULE.read_text(encoding="utf-8")
 
     assert_contains(init_text, "register_todo_command", "__init__")
     assert_contains(init_text, "handle_todo_command", "__init__")
@@ -61,7 +63,8 @@ def assert_source_shape() -> None:
     assert_contains(runtime_text, "handle_quota_command(", "cli_runtime.py")
     assert_contains(todo_text, "render_todo_markdown", "todo module")
     assert_contains(todo_text, "append_cli_rollout_event", "todo module")
-    assert_contains(quota_text, "render_quota_should_run_markdown", "quota module")
+    assert_contains(quota_text, "from .quota_presentation import", "quota module")
+    assert_contains(quota_presentation_text, "render_quota_should_run_markdown", "quota presentation")
     assert_contains(quota_text, "append_cli_rollout_event", "quota module")
 
 

loopx/cli_commands/quota_capture.py

@@ -209,21 +209,25 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
                                 max_page_output_bytes: int | None = None) -> dict[str, Any]:
     """Publish file references for an already-decided native capture.
 
-    Persist the complete wire and expose navigation without eagerly importing
-    content into model history. Reading never collects or admits a decision.
+    Reuse full recovery; persist delta or unwrapped wire before exposing it.
+    Reading never collects or admits a decision.
     """
     from ..control_plane.turn_driver.decision_capture import write_decision_capture
 
     if max_page_output_bytes is not None and max_page_output_bytes <= 0:
         raise QuotaDecisionTransportError("Native context output limit must be positive.")
-    wire = directory / "delivery.json"
-    write_decision_capture(wire, projected)
-    reference = {"path": str(wire), "sha256": hashlib.sha256(wire.read_bytes()).hexdigest()}
+    reference = None
     mode, target = "full", None
     if projected.get("schema_version") == "context_delivery_v1":
         mode, target = projected["mode"], projected["target"]
         if mode == "full":
             reference = {key: projected["recovery"][key] for key in ("path", "sha256")}
+    if reference is None:
+        # Full delivery already reads its committed recovery. Delta and
+        # unwrapped inputs still need their own immutable reader target.
+        wire = directory / "delivery.json"
+        write_decision_capture(wire, projected)
+        reference = {"path": str(wire), "sha256": hashlib.sha256(wire.read_bytes()).hexdigest()}
     page_bytes, output_bytes = native_context_page_budget()
     if max_page_output_bytes is not None:
         output_bytes = min(output_bytes, max_page_output_bytes)
@@ -265,7 +269,7 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
     result["instruction"] = CONTEXT_CAPTURE_NAVIGATION_GUIDANCE
     if mode != "full":
         result["instruction"] += (" command reads the delta; baseline_reference identifies its exact saved base when file-backed. "
-            "index_command/select_command read current facts directly without that baseline.")
+            "recovery.command reads current facts directly without that baseline.")
     # The reference must fit even under an explicit small caller ceiling.
     def fits():
         return len(json.dumps(result, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= min(output_bytes, 7900)
@@ -275,6 +279,9 @@ def fits():
         result.pop("select_command")  # --select also works on the current recovery command.
     if not fits() and "baseline_reference" in result:
         result["baseline_reference"].pop("command", None)  # Exact path/hash remain usable.
+    if not fits():
+        result.pop("index_command")
+        result["instruction"] += " Append --list or --select to recovery.command to discover or read current fields."
     if not fits():
         raise QuotaDecisionTransportError("Native context reference exceeds the output limit; "
             "read the saved original decision and receipt without rerunning admission.")

tests/control_plane/test_progressive_native_context.py

@@ -33,6 +33,29 @@ def outcome(document):
     return document["planning_horizon"]["core_goal"]["acceptance_contract"]["outcome_requirements"]["outcomes"][0]
 
 
+def test_full_native_reference_reuses_the_committed_recovery(tmp_path, monkeypatch):
+    from loopx.cli_commands.quota_capture import native_context_capture_view
+    from loopx.control_plane.turn_driver import decision_capture
+
+    setup(tmp_path, monkeypatch)
+    original, full = project(tmp_path, payload=payload())
+    recovery = full["recovery"]
+    saved = Path(recovery["path"]).read_bytes()
+
+    def unexpected_publication(*args, **kwargs):
+        pytest.fail("Full presentation must reuse its already committed recovery")
+
+    monkeypatch.setattr(decision_capture, "write_decision_capture", unexpected_publication)
+    view = native_context_capture_view(full, Path(recovery["path"]).parent,
+        context_directory=str(tmp_path))
+    assert (view["path"], view["sha256"]) == (recovery["path"], recovery["sha256"])
+    assert Path(view["path"]).read_bytes() == saved
+    pages = read_pages(tmp_path, full)
+    assert json.loads("".join(page["content"] for page in pages)) == full["context"]
+    details = full["context"]["context_details"]
+    assert json.loads(Path(details["path"]).read_bytes()) == original
+
+
 def test_cold_warm_and_repeated_compaction_keep_requirements_without_replaying_material(tmp_path, monkeypatch):
     event = setup(tmp_path, monkeypatch)
     original = payload()
@@ -124,7 +147,17 @@ def check_recovery(view):
             key: journal[key] for key in ("path", "sha256")}
         argv = shlex.split(view["recovery"]["command"])
         if output_limit is not None:
+            assert len((json.dumps(view, ensure_ascii=False, separators=(",", ":")) + "\n").encode()) <= output_limit
             assert int(argv[argv.index("--max-page-output-bytes") + 1]) <= output_limit
+            # Discovery and selection remain usable from the complete current
+            # recovery, independently of optional shortcut command fields.
+            code, index = _run_cli(registry, runtime, *argv[argv.index("quota"):], "--list", cwd=tmp_path)
+            assert code == 0 and index["ok"], index
+            fields = json.loads(index["content"])["selections"][0]["fields"]
+            assert {"/stop", "/heartbeat_receipt"} <= {field["pointer"] for field in fields}
+            code, selected = _run_cli(registry, runtime, *argv[argv.index("quota"):], "--select", "/stop", cwd=tmp_path)
+            assert code == 0 and selected["ok"], selected
+            assert json.loads(selected["content"])["selections"][0]["value"] == original["stop"]
         pages, combined = [], b""
         while True:
             code, recovered = _run_cli(registry, runtime, *argv[argv.index("quota"):], cwd=tmp_path)

tests/control_plane/test_saved_context_project.py

@@ -490,6 +490,7 @@ def test_native_saved_preflight_uses_lifecycle_without_guard_replay(tmp_path, mo
     from loopx.control_plane.quota.turn_envelope import prepare_context_delivery, restore_context_delivery
     from tests.control_plane.codex_checkpoint_fixture import local_inference_provider
     from tests.control_plane.test_codex_context_boundary import _app_server, _configure_provider, _native_fixture, _input_text
+    from tests.control_plane.codex_checkpoint_fixture import last_context_page as _last_page
     from tests.control_plane.test_codex_context_capture import _delivered
     from tests.control_plane.test_quota_settlement_cli import (
         AGENT_ID, GOAL_ID, TODO_ID, _configure_selectable_alternative,
@@ -575,18 +576,16 @@ def response(request, number, message):
                 assert descriptor["ok"] and descriptor["mode"] == ("full" if name == "cold" else "delta")
                 if benchmark:
                     assert len(json.dumps(descriptor, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 32000
-                wire = json.loads((Path(descriptor["path"]).parent / "delivery.json").read_text())
-                assert descriptor["target"] == wire["target"]
-                observed[name] = (wire, descriptor)
                 data = Path(descriptor["path"]).read_bytes()
-                # The existing page protocol also admits the projector's first
-                # page. Old installed projectors still start with command.
-                offset = descriptor["next_offset"] if "content" in descriptor else 0
-                for offset in range(len(data) if offset is None else offset, len(data), 4096):
-                    # Synthetic captures are ASCII JSON; public reader owns every byte.
-                    queue.append(shlex.join([*cli, "--format", "json", "quota", "context-read",
-                        "--capture-file", descriptor["path"], "--capture-sha256", descriptor["sha256"],
-                        "--offset", str(offset)]))
+                assert hashlib.sha256(data).hexdigest() == descriptor["sha256"]
+                published = json.loads(data)
+                if descriptor["mode"] == "delta":
+                    assert descriptor["target"] == published["target"]
+                observed[name] = (published, descriptor)
+                # Follow the public reader's actual page cursor rather than
+                # guessing offsets from the capture size. This keeps each
+                # request's real page budget and avoids overlapping reads.
+                queue.append(descriptor["command"])
                 if progressive:
                     # Exercise the actual small reference and selective public
                     # reader in native model input; never consume either file in full.
@@ -614,11 +613,17 @@ def response(request, number, message):
                     native_context = home / "loopx-context" if benchmark else context
                     assert "candidate" not in json.loads((native_context / "session.json").read_bytes())
                 else:
+                    page = _last_page(request)
+                    assert page is not None and page["sha256"] == descriptor["sha256"]
+                    if not page["eof"]:
+                        assert page["next_offset"] > page["offset"] and page["next_command"]
+                        queue.append(page["next_command"])
+                        return response(request, number, message)
                     assert _delivered(request, descriptor["sha256"]) == Path(descriptor["path"]).read_text().rstrip("\n")
                 if benchmark and name in ("cold", "warm"):
                     # Consume the command actually delivered to the native
                     # worker, rather than manually supplying missing flags.
-                    presented = json.loads(Path(observed[name][0]["recovery"]["path"]).read_text())
+                    presented = json.loads(Path(descriptor["recovery"]["path"]).read_text())
                     selection = presented["interaction_contract"]["cli_channel"]["selection_command"]
                     command = selection["route_prefix"] + " " + selection["command_args_template"].replace("{todo_id}", TODO_ID)
                     argv = shlex.split(command)
@@ -705,12 +710,16 @@ def response(request, number, message):
         assert _heartbeat_receipt_count(runtime, "saved-native-turn") == receipts and _spend_run_count(runtime) == 0
     cold = observed["cold"][0]
     baseline = None
-    for wire, _ in observed.values():
-        projected = json.loads(Path(wire["recovery"]["path"]).read_text())
+    for published, descriptor in observed.values():
+        projected = json.loads(Path(descriptor["recovery"]["path"]).read_text())
         current = prepare_context_delivery(projected,
-            scope={key: wire["target"][key] for key in
-                ("goal_id", "agent_id", "session_id", "context_epoch", "continuity_generation") if key in wire["target"]},
-            capture_id=wire["target"]["capture_id"], surface="quota_packet")
+            scope={key: descriptor["target"][key] for key in
+                ("goal_id", "agent_id", "session_id", "context_epoch", "continuity_generation") if key in descriptor["target"]},
+            capture_id=descriptor["target"]["capture_id"], surface="quota_packet")
+        # Full presentation publishes the original document; only deltas have
+        # a separate wire target. Both retain the same typed recovery check.
+        wire = current if descriptor["mode"] == "full" else published
+        assert current["target"] == descriptor["target"]
         restored = restore_context_delivery(wire, baseline=baseline, current=current)
         assert restored["context"] == projected
         if wire["mode"] == "delta":
@@ -720,7 +729,7 @@ def response(request, number, message):
         baseline = {**current, "ancestry": 0 if wire.get("continuity_basis") == "captured_file" else wire["ancestry"]}
     if benchmark:
         for name, (_, raw) in zip(("cold", "warm"), sources):
-            presentation = json.loads(Path(observed[name][0]["recovery"]["path"]).read_text())
+            presentation = json.loads(Path(observed[name][1]["recovery"]["path"]).read_text())
             original = json.loads(raw)
             selected = presentation["interaction_contract"]["cli_channel"]["selection_command"]
             original_command = original["interaction_contract"]["cli_channel"]["selection_command"]["command_args_template"]
@@ -749,7 +758,7 @@ def response(request, number, message):
             presentation.pop("context_reference_instruction", None)
             assert presentation == original  # only presentation transport differs
     else:
-        assert cold["context"] == json.loads(sources[0][1]) and projected == json.loads(sources[1][1])
+        assert cold == json.loads(sources[0][1]) and projected == json.loads(sources[1][1])
     assert all(path.read_bytes() == raw for path, raw in sources)