fix(context): activate native checkpoint capture before deferred presentation

benchmark/runtime/checkpoint.py

@@ -149,6 +149,12 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
                "--runtime-profile", "generic_cli", "--turn-instance-id", receipt["turn_id"],
                "--available-capability", "shell", "--available-capability", "filesystem_write",
                "--decision-output-root", str(capture_root)]
+    if env.get("LOOPX_CONTEXT_DELIVERY") == "on" and env.get("LOOPX_CODEX_CONTEXT_DIR"):
+        # Capture the complete Goal now, but let the actual native tool select
+        # full/delta after SessionStart. Do not consume or invalidate continuity
+        # while the Host is preparing a checkpoint outside the model session.
+        command += ["--context-delivery", "on", "--codex-context-dir", env["LOOPX_CODEX_CONTEXT_DIR"],
+                    "--defer-context-presentation"]
     guard_run = subprocess.run(command, cwd=env["LOOPX_PROJECT"], env=env, capture_output=True,
                                text=True, timeout=120, check=True)
     projected = json.loads(guard_run.stdout)

docs/reference/required-work-context.md

@@ -72,6 +72,17 @@ context-loss recovery retain the complete original text, including arbitrary
 sections, comments and final stops. This does not establish model retention,
 execution authority, token savings or accepted outcome gains.
 
+Checkpoint Host preflight uses the hidden `--defer-context-presentation` option
+with an explicitly enabled native JSON `should-run` capture. It collects the
+same complete Goal source for admitted work-context reads and writes the ordinary
+immutable CLI view, without
+reading or replacing the model session's continuity state. After native lifecycle
+hooks run, the model presents that saved view with `context-project`; this does
+not re-enter admission. Deferred presentation requires a capture destination and
+cannot be combined with socket handoff. Ordinary callers keep their existing
+presentation, and context delivery off keeps its original pending Goal reads.
+Unbound staged replans retain their existing review-owned original Goal read.
+
 The Goal availability adapter still reads and UTF-8-validates the complete file
 and hashes all bytes for its revision. With native context delivery off, only the path and revision of unmarked or mixed sources cross the
 internal context-projection bridge, preserving the complete progressive read
@@ -480,6 +491,13 @@ baseline 继承，当前来源凭据和未知字段仍明文提供。冷态及
 原文，包括任意章节、注释和末尾停止条件。不据此证明模型保留、执行权限、token 节省
 或已验收成果收益。
 
+Checkpoint Host 预检在显式开启 native JSON `should-run` 捕获时使用隐藏参数
+`--defer-context-presentation`。它为已准入工作上下文读取捕获完整 Goal，并保存普通
+不可变 CLI 视图，不读取或替换模型会话的连续性状态。原生生命周期 hook 运行后，模型
+通过 `context-project` 呈现同一视图，不重复准入。延迟呈现必须指定捕获位置，不能
+与 socket handoff 组合。普通调用保持既有呈现；关闭差分时保留原来的待完成 Goal
+读取。未绑定任务的分阶段重规划继续沿用 review 所属的原始 Goal 读取。
+
 Goal adapter 仍完整读取文件、校验 UTF-8，并以全部字节计算修订摘要；关闭 native
 context delivery 时，未标记或混合归属仅在内部桥接传路径与修订，避免未被消费的正文
 副本，Agent 仍须渐进式全文读取。显式

loopx/cli_commands/quota.py

@@ -422,13 +422,14 @@ def _emit_quota_result(
     # actual-input observation owns its projection when explicitly selected;
     # never wrap one context-delivery wire in a second projection.
     native_context = context_delivery_enabled and bool(getattr(args, "codex_context_dir", None) or getattr(args, "codex_context_root", None))
-    if (args.format == "json" and succeeded and delivery_request is not None and not native_context
+    deferred_context = getattr(args, "defer_context_presentation", False)
+    if (not deferred_context and args.format == "json" and succeeded and delivery_request is not None and not native_context
             and context_delivery_enabled):
         payload = project_captured_context(payload, directory=capture_directory,
             request=delivery_request[0], capture_id=delivery_request[1],
             goal_id=decision_identity[0], agent_id=decision_identity[1],
             surface="turn_envelope" if getattr(args, "turn_envelope", False) else "quota_packet")
-    if (args.format == "json" and succeeded and capture_directory
+    if (not deferred_context and args.format == "json" and succeeded and capture_directory
             and native_context):
         import hashlib
         from ..control_plane.turn_driver.codex_context import codex_context_directory, project_codex_context

loopx/cli_commands/quota_capture.py

@@ -303,6 +303,14 @@ def project_saved_context_capture(args: argparse.Namespace) -> dict[str, Any]:
 
 def prepare_quota_capture(args: argparse.Namespace, directory: Path | None) -> Path | None:
     """Validate transport options before allocating a new guard capture."""
+    if getattr(args, "defer_context_presentation", False) and (
+        args.quota_command != "should-run" or args.format != "json"
+        or not native_context_activation(args)["enabled"]
+        or getattr(args, "decision_handoff_socket", None)
+        or not (getattr(args, "decision_output_dir", None) or getattr(args, "decision_output_root", None))
+    ):
+        raise QuotaCommandValidationError(
+            "deferred context presentation requires a native JSON should-run capture without socket handoff")
     if directory is None and (
         getattr(args, "decision_output_dir", None) is not None
         or getattr(args, "decision_output_root", None) is not None

loopx/cli_commands/quota_registration.py

@@ -207,6 +207,9 @@ def register_quota_command(
               "Waits at most 120s for a digest-bound release; failure retains the original receipt."),
     )
     quota_parser.add_argument("--capture-file", help="For context-read/context-project: original captured JSON file; no guard is evaluated.")
+    # Host preflight captures the complete native input, then presents that same
+    # immutable view after the model's lifecycle hooks establish continuity.
+    quota_parser.add_argument("--defer-context-presentation", action="store_true", help=argparse.SUPPRESS)
     from ..control_plane.turn_driver.codex_context import CONTEXT_DELIVERY_SETTINGS
 
     quota_parser.add_argument("--context-delivery", choices=CONTEXT_DELIVERY_SETTINGS,

tests/control_plane/test_quota_decision_capture.py

@@ -31,6 +31,21 @@ def test_disabled_capture_does_not_touch_disk(tmp_path):
     assert list(tmp_path.iterdir()) == []
 
 
+@pytest.mark.parametrize('overrides', [
+    {'context_delivery': 'off'}, {'format': 'markdown'}, {'quota_command': 'plan'},
+    {'decision_output_dir': None}, {'decision_handoff_socket': '/unused/socket'},
+])
+def test_deferred_native_presentation_requires_original_capture_before_admission(tmp_path, overrides):
+    from loopx.cli_commands.quota_capture import prepare_quota_capture
+    values = dict(defer_context_presentation=True, format='json', context_delivery='on',
+                  codex_app=True, decision_output_dir=tmp_path / 'capture')
+    values.update(overrides)
+    path = values.pop('decision_output_dir')
+    with pytest.raises(QuotaCommandValidationError, match='deferred context presentation'):
+        prepare_quota_capture(request(path, **values), None)
+    assert list(tmp_path.iterdir()) == []
+
+
 @pytest.mark.parametrize('overrides', [
     {'quota_command': 'spend-slot'}, {'turn_instance_id': None},
 ])

tests/control_plane/test_replan_staged_read.py

@@ -193,6 +193,8 @@ def test_checkpoint_preflight_delivers_the_real_cli_view_and_retains_full_source
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     fixture = make_progress_cli_fixture(tmp_path, provider, "advanced")
     enable_explore(fixture)
+    with fixture["state"].open("a") as stream:
+        stream.write("\n## Original additional requirements\nPreserve this final independent stop.\n")
     if not replan:
         fixture["index"].write_text("")
     wake = tmp_path / "wakes" / TURN
@@ -231,6 +233,21 @@ def observe(command, **kwargs):
         assert str(view_path) in body and str(source_path) not in body
         assert body.endswith("original heartbeat instructions")
         if context_delivery:
+            assert "--defer-context-presentation" in commands[0]
+            assert "--context-delivery" in commands[0]
+            assert not list(native.iterdir()), "Host preflight must not alter native continuity"
+            assert not any(read.get("source") == "goal_state" for read in
+                view["interaction_contract"]["agent_channel"]["required_reads"])
+            delivered = [row["content"] for row in view["interaction_contract"]["agent_channel"]
+                ["work_context"]["sources"] if row.get("source") == "goal_state"]
+            if replan:
+                # This fixture has no bound work: the staged review owns the
+                # original Goal read, not ordinary work-context delivery.
+                core = source["autonomous_replan_obligation"]["replan_context"]["core_goal"]
+                assert shlex.split(core["read_action"]) == ["cat", "--", str(fixture["state"].resolve())]
+            else:
+                assert len(delivered) == 1 and delivered[0]["full_source"] is True
+                assert delivered[0]["text"] == fixture["state"].read_text()
             # The optional full/delta transport must present the staged CLI
             # view too. Its full recovery must not restore old inline hints.
             command = body.split("```sh\n", 1)[1].split("\n```", 1)[0]