Validate saved complete Turn context before recovery

benchmark/runtime/context_delivery.py

@@ -27,17 +27,18 @@ def preflight_context_instruction(env: dict[str, str], path: Path, digest: str,
     project = [*prefix, "context-project", *source,
         "--goal-id", env["LOOPX_GOAL_ID"], "--agent-id", env["LOOPX_AGENT_ID"],
         "--turn-instance-id", turn_id, "--initial-read", "--context-delivery", "on", "--runtime-profile", "generic_cli", "--codex-context-dir", env["LOOPX_CODEX_CONTEXT_DIR"]]
-    recovery_index = [*prefix, "context-read", *source, "--list"]
+    receipt_reader = [*prefix, "context-read", *source, "--select", "/heartbeat_receipt"]
     return ("Run this command directly and deliver its complete bounded stdout to the model. "
         "Keep top-level content and next_command when present; navigation alone omits the returned page. "
         "Preserve any failure response.\n```sh\n" + shlex.join(project)
-        + "\n```\n" + ("" if guide_in_host else CONTEXT_CAPTURE_NAVIGATION_GUIDANCE)
-        + "If the projection command itself fails, recover necessary facts from the same original capture. "
-        "When pointers are unknown, discover them with:\n```sh\n" + shlex.join(recovery_index)
-        + "\n```\nUse the same capture's generic reader and selectors for the facts the current decision needs. "
-        "For known fields, replace --initial-read with repeated --select pointers in the first command. "
-        "Continue the returned page or use navigation.recovery.command to reread; do not project again. "
-        "Discovery does not require draining the packet. "
+        + "\n```\n" + ("" if guide_in_host else "After projection succeeds: " + CONTEXT_CAPTURE_NAVIGATION_GUIDANCE)
+        + "For known fields, replace --initial-read with repeated --select pointers in the first command. "
+        "After a successful projection, continue the returned page or use navigation.recovery.command to reread. "
+        "If projection fails, follow its recovery instructions and preserve the original receipt. "
+        "If it provides no verified context or usable instructions, restore the intact capture and retry the same "
+        "context-project with its original digest and Turn. Use this command for diagnostic receipt readback:\n```sh\n"
+        + shlex.join(receipt_reader) + "\n```\n"
+        "Use verified context before work. Do not rerun admission to recover presentation. "
         "Neither presentation nor recovery renews this decision's authority or freshness. ")
 
 

benchmark/tests/test_context_delivery.py

@@ -68,6 +68,7 @@ def test_projection_failure_keeps_original_capture_selectively_readable(tmp_path
         for command in re.findall(r"```sh\n(.*?)\n```", instruction, re.S)
     ]
     assert len(commands) == 2
+    assert commands[1][commands[1].index("--select") + 1:] == ["/heartbeat_receipt"]
 
     def invoke(argv):
         return subprocess.run(
@@ -95,6 +96,14 @@ def invoke(argv):
     assert json.loads(index["content"])["schema_version"] == "context_capture_index_v1"
     assert len(discovery.stdout.encode()) < len(raw) // 5
 
+    diagnostic = invoke(commands[1])
+    assert diagnostic.returncode == 0, diagnostic.stderr
+    receipt_page = json.loads(diagnostic.stdout)
+    assert receipt_page["sha256"] == digest and receipt_page["eof"]
+    assert json.loads(receipt_page["content"])["selections"] == [
+        {"pointer": "/heartbeat_receipt", "value": source["heartbeat_receipt"]},
+    ]
+
     reader = recovery[:recovery.index("context-read") + 1]
     branch = invoke([
         *reader, "--capture-file", str(capture), "--capture-sha256", digest,

docs/reference/protocols/turn-envelope-v0.md

@@ -56,6 +56,20 @@ An older CLI rejects the new flag before guard execution. Omit it to roll back;
 existing complete captures remain readable through their original digest-bound
 `quota context-read`/`context-project` recovery.
 
+`quota context-project` checks a saved complete decision before presenting it
+as current context, including with Delta alone or with context delivery off.
+If the saved source cannot be verified, the returned digest-bound
+`context-read` command selects only its recorded `heartbeat_receipt` for
+diagnosis. Restore the intact, supported source and retry `context-project`
+with the original Goal, Agent, Turn and digest before using it for work.
+Native recovery publishes the current session source after the same typed
+check succeeds. Rejection retires the implicit lookup for the same native
+session, Goal and Agent. The original captures remain available by their
+explicit digest-bound references. For a rejected ready capture, the native
+hook can still read back its recorded claim and lease references. A
+diagnostic read cannot acknowledge delivery of ready work context. Optional
+presentation failures after verification can still refer to the verified capture.
+
 For eligible original Todo plans, the same typed settlement owner projects one
 `refresh-state ... --settle-turn --settlement-receipt` command. It retains the
 original effect identity, validation and terminal conditions. A non-ready
@@ -80,6 +94,15 @@ model context. Actual live adoption and efficiency still require separate proof.
 可执行的原 Todo 结算计划复用同一 typed owner，合并为一条写回并保留原 effect id，
 不把未 ready 的准备或错误回执转成授权。部分失败只沿原 `settlement_owed` 恢复。
 旧 CLI 在 guard 前拒绝新选项，回滚时移除该选项；旧完整 capture 仍用原 reader。
+
+`quota context-project` 先校验已保存的完整决定，再将它呈现为当前上下文。仅开启差分
+或关闭上下文交付时同样校验。来源校验失败或中断时，返回的摘要绑定
+`context-read` 命令只选取原记录的 `heartbeat_receipt` 用于诊断。恢复完整且受支持
+的来源后，以原 Goal、Agent、Turn 和摘要重试 `context-project`，通过后再作为工作
+上下文。原生会话通过同一校验后发布当前来源。拒绝来源时撤下同一会话、Goal 和 Agent
+的隐式读回入口。原捕获仍可通过显式摘要引用读取。准备就绪的捕获被拒绝后，原生
+Hook 仍能读回已记录的领取与租约引用。诊断读取不会确认已交付就绪的工作上下文。
+校验成功后如遇可选呈现失败，仍可读取已验证的捕获。
 这只是 Host 开发协议入口，不新增设置、模型/子 worker、预算或权限，也不证明
 已运行的 Host 采用或效率收益。
 

loopx/cli_commands/quota_capture.py

@@ -278,6 +278,39 @@ def progressive_context_capture_view(payload: dict[str, Any], args: argparse.Nam
     return readable if page_attached or len(_native_output_bytes(readable)) < len(_native_output_bytes(required)) else required
 
 
+def _saved_context_transport_failure(
+    error: Exception, *, args: argparse.Namespace, receipt: dict[str, Any],
+    recovery: dict[str, Any], reader_argv: list[str], source_unverified: bool = False,
+) -> dict[str, Any]:
+    page_bytes, output_bytes = native_context_page_budget()
+    limit = getattr(args, "max_page_output_bytes", None)
+    if isinstance(limit, int) and limit > 0:
+        output_bytes = min(output_bytes, limit)
+        page_bytes = min(page_bytes, output_bytes)
+    if source_unverified:
+        instruction = (
+            "The saved source could not be verified. Use recovery.command for diagnostic receipt readback. "
+            "Restore an intact, supported capture and retry context-project with its original Goal, Agent, Turn "
+            "and digest before using it as work context. Preserve the original receipt and do not rerun admission."
+        )
+    else:
+        instruction = (
+            "Context presentation was not published. Discover current fields with recovery.command, then select "
+            "only the facts the current decision needs from the original capture; do not drain unrelated optional "
+            "material or rerun admission. Preserve its Turn receipt. If a selected value spans pages, follow "
+            "next_command only for that selection and verify contiguous offsets, total_bytes and sha256. "
+            "Authority and freshness are unchanged. " + CONTEXT_CAPTURE_NAVIGATION_GUIDANCE
+        )
+    return {"ok": False, "error_code": QuotaDecisionTransportError.error_code,
+        "reason": " ".join(str(error).split())[:500],
+        "heartbeat_receipt": receipt, "capture_sha256": args.capture_sha256,
+        "recovery": {**recovery, "command": context_capture_read_command(
+            Path(recovery["path"]), recovery["sha256"], page_bytes=page_bytes,
+            max_page_output_bytes=output_bytes, list_fields=not source_unverified,
+            select=["/heartbeat_receipt"] if source_unverified else None, reader_argv=reader_argv)},
+        "instruction": instruction}
+
+
 def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str | None = None) -> dict[str, Any]:
     """Present an immutable already-decided quota capture, without authority IO.
 
@@ -289,6 +322,9 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
     Invalid run-mode opt-ins fail before that optional transport path.
     """
     from ..command_invocation import current_loopx_cli_argv
+    from ..control_plane.quota.turn_envelope import (
+        COMPLETE_TURN_DECISION_VERSION, assert_complete_turn_decision, is_legacy_saved_quota_decision,
+    )
 
     reader_argv = current_loopx_cli_argv()
     try:
@@ -327,6 +363,23 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
     reference = {"path": str(path), "sha256": args.capture_sha256,
         "command": context_capture_read_command(path, args.capture_sha256, reader_argv=reader_argv)}
     full_reference = dict(reference)
+    version = payload.get("schema_version")
+    complete = version == COMPLETE_TURN_DECISION_VERSION
+    try:
+        if not complete and not is_legacy_saved_quota_decision(payload):
+            if activation["enabled"]:
+                from ..control_plane.turn_driver.codex_context import retire_rejected_saved_context
+
+                retire_rejected_saved_context(directory=args.codex_context_dir, context_root=args.codex_context_root,
+                    goal_id=args.goal_id, agent_id=args.agent_id, capture_sha256=args.capture_sha256,
+                    hook_session=hook_session)
+            raise QuotaDecisionTransportError("Unsupported saved quota decision version or mixed Turn binding.")
+        if complete and not activation["enabled"] and not activation["progressive_enabled"]:
+            assert_complete_turn_decision(payload)
+    except (QuotaDecisionTransportError, EffectRuntimeRemoteError, EffectRuntimeStartupError) as exc:
+        return _saved_context_transport_failure(exc, args=args, receipt=receipt,
+            recovery=full_reference, reader_argv=reader_argv, source_unverified=True)
+
     mode = "full"
     target = None
     native_prepared = False
@@ -342,15 +395,13 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
                 directory=None, reader_argv=reader_argv, captured_decision_path=str(path), page_attached=saved_page)
             if saved_page:
                 return _saved_context_read_page(args, navigation, initial_reference=None,
-                    initial_read_plan=None, output_limit=caller_output_limit, reader_argv=reader_argv)
+                    initial_read_plan=None, output_limit=output_limit, reader_argv=reader_argv)
             return navigation
-        except EffectRuntimeRejected as exc:
-            if payload.get("schema_version") == "loopx_turn_envelope_v1":
-                failure = QuotaDecisionTransportError(str(exc))
-                return {"ok": False, "error_code": failure.error_code, "reason": str(failure),
-                    "capture_sha256": args.capture_sha256, "recovery": full_reference}
-        except (QuotaDecisionTransportError, OSError, ValueError, EffectRuntimeRemoteError, EffectRuntimeStartupError):
-            pass  # The complete original reference remains the fallback below.
+        except (QuotaDecisionTransportError, OSError, ValueError, EffectRuntimeRemoteError, EffectRuntimeStartupError) as exc:
+            if complete:
+                return _saved_context_transport_failure(exc, args=args, receipt=receipt,
+                    recovery=full_reference, reader_argv=reader_argv, source_unverified=True)
+            # Historical readers keep their original fallback below.
     try:
         from ..control_plane.turn_driver.codex_context import codex_context_directory, prepare_codex_context
 
@@ -385,7 +436,6 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
         # Carry the now-active transport into its selection presentation, not
         # into the immutable source or a second admission. Native lifecycle
         # still owns when the following full/delta view can be published.
-        complete = payload.get("schema_version") == "loopx_turn_envelope_v1"
         transport = _capture_selection_transport_template(payload, args, capture_root=root)
         presentation = payload
         if args.initial_read and activation["progressive_enabled"] and not complete:
@@ -430,6 +480,11 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
         mode, target = native_view["mode"], native_view.get("target")
         initial_reference = native_view.get("first_read_reference")
         native_view = model_context_reference(native_view, reader_argv=reader_argv)
+    except (EffectRuntimeRemoteError, EffectRuntimeStartupError) as exc:
+        if not complete:
+            raise
+        return _saved_context_transport_failure(exc, args=args, receipt=receipt,
+            recovery=full_reference, reader_argv=reader_argv, source_unverified=not native_prepared)
     except (QuotaDecisionTransportError, OSError, ValueError, TypeError, RecursionError) as exc:
         if native_view is not None and not isinstance(exc, QuotaDecisionTransportError):
             # The native view and journal are already valid. Publishing the
@@ -438,28 +493,9 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
             pass
         elif isinstance(exc, QuotaDecisionTransportError) or (activation["enabled"] and not native_prepared):
             if not isinstance(exc, QuotaDecisionTransportError):
-                exc = QuotaDecisionTransportError(
-                    "Current context recovery could not be prepared. Read the saved "
-                    "original decision and receipt; do not rerun admission."
-                )
-            page_bytes, output_bytes = native_context_page_budget()
-            if isinstance(caller_output_limit, int) and caller_output_limit > 0:
-                output_bytes = min(output_bytes, caller_output_limit)
-                page_bytes = min(page_bytes, output_bytes)
-            recovery_index = {**full_reference, "command": context_capture_read_command(
-                Path(full_reference["path"]), full_reference["sha256"],
-                page_bytes=page_bytes, max_page_output_bytes=output_bytes, list_fields=True,
-                reader_argv=reader_argv)}
-            return {"ok": False, "error_code": exc.error_code,
-                "reason": " ".join(str(exc).split())[:500],
-                "heartbeat_receipt": receipt, "capture_sha256": args.capture_sha256,
-                "recovery": recovery_index,
-                "instruction": "Context presentation was not published. Discover current fields with "
-                    "recovery.command, then select only the facts the current decision needs from the original "
-                    "capture; do not drain unrelated optional material or rerun admission. Preserve its Turn "
-                    "receipt. If a selected value spans pages, follow next_command only for that selection and "
-                    "verify contiguous offsets, total_bytes and sha256. Authority and freshness are unchanged. "
-                    + CONTEXT_CAPTURE_NAVIGATION_GUIDANCE}
+                exc = QuotaDecisionTransportError("Current context recovery could not be prepared.")
+            return _saved_context_transport_failure(exc, args=args, receipt=receipt,
+                recovery=full_reference, reader_argv=reader_argv, source_unverified=complete and not native_prepared)
         # Off is inert; after native preparation an optional wire write can fall
         # back to the committed required view without leaving an older selector live.
     result: dict[str, Any] = {"ok": True, "mode": mode, **reference, **({"target": target} if target else {}),
@@ -915,11 +951,16 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca
                                 "goal_id": request["goal_id"], "agent_id": request["agent_id"],
                                 "turn_instance_id": request["turn_instance_id"],
                             }), hook_session=event["session_id"])
-                            if receipt["context"].get("ok") is False:
-                                from ..command_invocation import current_loopx_cli_argv
-
-                                receipt["context"]["recovery_command"] = context_capture_read_command(
-                                    source, digest, reader_argv=current_loopx_cli_argv())
+                            context = receipt["context"]
+                            if context.get("ok") is False:
+                                recovery = context.get("recovery")
+                                if isinstance(recovery, dict) and isinstance(recovery.get("command"), str):
+                                    context["recovery_command"] = recovery["command"]
+                                else:
+                                    from ..command_invocation import current_loopx_cli_argv
+
+                                    context["recovery_command"] = context_capture_read_command(
+                                        source, digest, reader_argv=current_loopx_cli_argv())
                             preparation_context = codex_execution_preparation_context(receipt)
                 except LegacyNativePreparationCapture as error:
                     preparation_context = str(error)

loopx/control_plane/quota/turn_envelope.py

@@ -302,6 +302,14 @@ def compose_turn_decision(
     return _apply_context_projection_changes(source, result)
 
 
+def is_legacy_saved_quota_decision(payload: Mapping[str, Any]) -> bool:
+    """Recognize saved raw/v0 carriers that predate complete-source checking."""
+    return (
+        "contract_binding" not in payload
+        and payload.get("schema_version", TURN_ENVELOPE_SCHEMA_VERSION) == TURN_ENVELOPE_SCHEMA_VERSION
+    )
+
+
 def assert_complete_turn_decision(payload: Mapping[str, Any]) -> None:
     """Validate the original contract; keep its host binding and caller-owned bytes."""
     _turn_envelope_result("complete_consumer", payload=dict(payload))