C1: unify work sources and validate recovery

benchmark/edgebench/README.md

@@ -994,30 +994,25 @@ budget or submissions. Freeze it with exact sources under a new run ID; do not
 change live workers. Receipts qualify requested configuration, while actual
 review, proposal adoption and task gains require trajectory evidence.
 
-## Optional short-context treatment
-
-Add `--turn-envelope` to an otherwise identical `heartbeat-resume` or
-`heartbeat-explore` command to use the product's short normal packet and local
-same-invocation full detail. Other workers reject the option before creating a
-trial. The default remains off; the runtime and worker-profile receipts record
-it explicitly. Use a new run ID and freeze this as a separate context treatment,
-keeping objective guidance, task/scorer version, model and budgets matched.
-Omit the flag on the next run to disable it. No active experiment is changed by
-installing this code. Inspect wake receipts and the saved decisions before
-claiming adoption; score countability still requires the usual integrity review.
-
-For an explicitly complete v1 Host contract, choose `--complete-turn` instead
-on either heartbeat profile. The shared worker forwards that choice through
-the real heartbeat renderer and guard; it retains the original ready claim,
-receipt and single-command settlement. It cannot be combined with the legacy
-short-envelope option. Context delivery remains independent, and unselected
-runs keep their existing behavior. See the [Host protocol](../../docs/reference/protocols/turn-envelope-v0.md#explicit-complete-v1-host-entry--显式完整-v1-host-入口)
-for source readback, failure recovery and rollback. User task time requirements
-remain readable; private Host deadlines stay outside model context.
-
-中文：heartbeat profile 可显式选择 `--complete-turn` 以消费完整 v1，不能与旧短包
-选项同时启用；共用原 ready claim、回执和单条结算 owner，不改变未选择的运行。
-按原 source reader 核实采用与恢复；该开关不新增模型、子 worker、时钟或权限。
+## Complete Turn contract
+
+New `heartbeat-resume` and `heartbeat-explore` runs use the complete v1
+decision (C1) with the original ready claim, receipt and single-command
+settlement. `--complete-turn` records an explicit selection in the worker
+and runtime receipts. `--context-delivery off` disables native context
+delivery while keeping the same decision contract.
+
+Use `--turn-envelope` only to reproduce the v0 presentation of a signed
+decision. Give that comparison a new run ID and keep task/scorer version,
+model and budgets matched. To compare against the original quota source, pin
+the reference runner and LoopX revision recorded by the experiment. Inspect
+wake receipts and saved decisions to verify adoption. See the
+[Host protocol](../../docs/reference/protocols/turn-envelope-v0.md#explicit-complete-v1-host-entry--显式完整-v1-host-入口)
+for source readback and recovery.
+
+中文：新 heartbeat 运行默认用 C1 决策，关闭上下文交付后仍用同一份决策契约。
+运行回执记录默认或显式选择。旧短包选项仅用于复现展示，原 quota 对照需要固定
+对照组的 runner 与 LoopX 版本。
 
 ## Close a managed native attempt
 

benchmark/edgebench/run.py

@@ -126,9 +126,9 @@ def main(argv=None):
                         help="Opt-in private binding for native terminal proof, consumed by the controller")
     turn_view = parser.add_mutually_exclusive_group()
     turn_view.add_argument("--turn-envelope", action="store_true",
-                        help="Opt-in short heartbeat context with same-invocation full captures")
+                        help="Compatibility capture option; quota uses the shared complete v1 contract, with context delivery configured separately")
     turn_view.add_argument("--complete-turn", action="store_true",
-                        help="Explicit complete v1 Turn contract with shared exact-Turn settlement; heartbeat workers only")
+                        help="Explicitly select the complete v1 Turn contract, already the default for heartbeat workers")
     parser.add_argument("--context-delivery", choices=("off", "on"),
                         help="Native Codex context delivery: on by default for heartbeat workers; pass off to disable")
     parser.add_argument("--execution-observation", choices=("local_probe",),
@@ -548,7 +548,7 @@ def main(argv=None):
         "loopx_commit": pins[0], "runner_commit": pins[1],
         "sforge_source": sforge_source if native_result_scope is not None else active_source_receipt(),
         "local_docker_images": local_docker_images,
-        **turn_envelope_receipt(explicit=args.turn_envelope, complete=args.complete_turn),
+        **turn_envelope_receipt(execution),
         **({"context_delivery": "on"} if args.context_delivery != "off" else {}),
         **({"execution_observation": args.execution_observation}
            if args.execution_observation is not None else {}),

benchmark/runtime/RUNTIME.md

@@ -176,6 +176,14 @@ a complete-mode binding cannot verify a pending command that uses
 `--complete-turn`. Historical ordinary and envelope guards remain readable.
 The existing Darwin native-checkpoint refusal remains in force.
 
+New heartbeat work uses the complete C1 guard by default, with context delivery
+on or off. The explicit v0 display still reads a signed C1 source. An omitted
+display flag does not by itself add a checkpoint or capture directory. The
+worker and trial receipts record whether C1 was selected explicitly or by default.
+Preflight and live handoff require the signed source before selecting a replan
+point. Native checkpoint recovery can still read previously recorded unversioned
+originals with their receipt and scope checks.
+
 The host-side checkpoint preflight captures the complete original decision.
 With context delivery enabled, its prompt supplies `quota context-project`
 with that original path and hash. The shared presentation owner prepares the
@@ -193,9 +201,13 @@ original full-read reference and receipt; use that explicit reference without
 rerunning admission. An optional cache write after recovery is established can
 still fall back to the same full input. A bad source hash/scope is refused before
 publication. This does not renew decision
-freshness or establish autonomous solver benefit. Default-off preloading is
-unchanged. Include the projection command, descriptor, recovery and page reads
-in subsequent total-cost comparisons.
+freshness or establish autonomous solver benefit. With context delivery off,
+preflight forwards an explicit short-envelope request and pins that choice in
+same-Turn Todo selection. The short view links to the original complete
+capture used by the checkpoint selector. The wake receipt records the
+requested treatment, so check the delivered view when qualifying a comparison.
+Include the projection command, descriptor, recovery and page reads in
+subsequent total-cost comparisons.
 Across distinct worker wakes, settle the prior Turn before selecting new work;
 reading an old prompt from resumed history cannot supply the next wake's guard.
 Do not modify an active/frozen trial. Set `context_delivery: "off"` for a new trial

benchmark/runtime/checkpoint.py

@@ -21,7 +21,7 @@
 from loopx.control_plane.quota.settlement import read_heartbeat_settlement
 from loopx.hosts.research_checkpoint import (
     _participant_id, native_binding_participant, verify_native_binding_participant,
-    research_participants, verify_research_participants,
+    research_participants, select_checkpoint_replan, verify_research_participants,
 )
 from .checkpoint_handoff import CHECKPOINT_CAPTURE_SECONDS
 from .codex import explicit_turn_display
@@ -122,10 +122,8 @@ def _prior_replan_guard_seen(env: dict[str, str], current: Path, obligation_id:
         if path.resolve(strict=True) != path:
             raise ValueError("prior checkpoint guard must be a physical Host file")
         guard = json.loads(path.read_text())
-        prior = effect_runtime_result("runtime.checkpoint_artifacts", {
-            "action": "select_replan", "quota": guard, "goal_id": env["LOOPX_GOAL_ID"],
-            "agent_id": env["LOOPX_AGENT_ID"], "turn_instance_id": path.parent.name,
-        }, large_local_snapshot=True)
+        prior = select_checkpoint_replan(guard, goal_id=env["LOOPX_GOAL_ID"],
+            agent_id=env["LOOPX_AGENT_ID"], turn_instance_id=path.parent.name, allow_legacy_source=True)
         if prior["capture"] and prior["obligation_id"] == obligation_id:
             return True
     return False
@@ -222,19 +220,21 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
         raise ValueError("fresh_replan checkpoint trigger requires a heartbeat execute wake")
     started = time.monotonic()
     guard_path = Path(env["LOOPX_WAKE_LOG_DIR"]) / receipt["turn_id"] / "checkpoint-start-guard.json"
-    # Use the CLI's original-decision capture, including with TurnEnvelope off.
-    # The checkpoint selector needs the full source; the model consumes the
-    # ordinary CLI projection of those same bytes, never a second guard.
+    # The selector and the model both use the original CLI guard. The explicit
+    # compatibility display can still render v0 from that signed source.
     capture_root = Path(tempfile.mkdtemp(prefix="checkpoint-start-", dir=guard_path.parent))
     command = [env["LOOPX_CLI"], "--cli-bin", env["LOOPX_CLI"], "--format", "json", "--registry", env["LOOPX_REGISTRY"],
                "--runtime-root", env["LOOPX_RUNTIME_ROOT"], "quota", "should-run",
                "--goal-id", env["LOOPX_GOAL_ID"], "--agent-id", env["LOOPX_AGENT_ID"],
                "--runtime-profile", "generic_cli", "--turn-instance-id", receipt["turn_id"],
                "--available-capability", "shell", "--available-capability", "filesystem_write"]
     display = explicit_turn_display(env)
-    if display.complete_turn:
-        command.append("--complete-turn")
     native_context = env.get("LOOPX_CONTEXT_DELIVERY") == "on" and bool(env.get("LOOPX_CODEX_CONTEXT_DIR"))
+    if display.uses_complete_heartbeat_guard:
+        command.append("--complete-turn")
+    elif display.turn_envelope:
+        command.append("--turn-envelope")
+    legacy_display = display.turn_envelope and env.get("LOOPX_CONTEXT_DELIVERY", "off") == "off"
     if native_context:
         # The native reader selects full/delta after SessionStart. Ask the
         # existing preparation route for the committed capture reference now.
@@ -244,6 +244,10 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
                     "--defer-context-presentation"]
     else:
         command += ["--decision-output-root", str(capture_root)]
+        if legacy_display:
+            # Forward the explicit raw display to the same CLI that renders
+            # ordinary wakes. The checkpoint selector still reads the source.
+            command += ["--context-delivery", "off"]
     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)
@@ -259,6 +263,9 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
         raise ValueError("checkpoint requires a successful original quota guard")
     if native_context and projected.get("capture_sha256") != guard_digest:
         raise ValueError("checkpoint prepared capture differs from the original guard; do not replay the guard")
+    # A fresh guard always captures signed C1, including for the v0 display.
+    intent = select_checkpoint_replan(guard, goal_id=env["LOOPX_GOAL_ID"],
+        agent_id=env["LOOPX_AGENT_ID"], turn_instance_id=receipt["turn_id"])
     # A previous preflight on this wake may already have allowed model work,
     # even if its optional attempt marker (or final receipt) could not be saved.
     # Such reentry can verify an existing point, but cannot capture a later one.
@@ -275,17 +282,14 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
         view_path = guard_path.with_name("checkpoint-start-view.json")
         view_path.write_text(guard_run.stdout, encoding="utf-8")
         view_path.chmod(0o600)
-        reading = (f"Read the complete CLI view at {view_path} before work, then follow its interaction_contract "
+        action = "action, required reads and settlement instructions" if legacy_display else "interaction_contract"
+        reading = (f"Read the complete CLI view at {view_path} before work, then follow its {action} "
             "and replan review/writeback reads when present. ")
     preloaded_body = ("Host checkpoint preflight has already executed this wake's original quota guard once. "
         + reading +
         "Do not repeat the initial guard below; use its selection or recovery command when required.\n\n" + body)
     receipt["checkpoint"] = {"captured": False, "research_point_qualified": False, "phase": "before_replan"}
     try:
-        intent = effect_runtime_result("runtime.checkpoint_artifacts", {
-            "action": "select_replan", "quota": guard, "goal_id": env["LOOPX_GOAL_ID"],
-            "agent_id": env["LOOPX_AGENT_ID"], "turn_instance_id": receipt["turn_id"],
-        }, large_local_snapshot=True)
         from .checkpoint_continuation import restored_capture_gate
         restore_gate = restored_capture_gate(env, intent.get("obligation_id"),
             full_history=not request.get("native_session", False))
@@ -676,11 +680,9 @@ def remaining():
         if failed_readback:
             intent = {"capture": False, "reason": "quota_readback_failed", "execution_authority_granted": False}
         else:
-            intent = effect_runtime_result("runtime.checkpoint_artifacts", {
-                "action": "select_replan", "quota": quota, "goal_id": goal_id,
-                "agent_id": agent_id, "turn_instance_id": (turn_instance_id if turn_instance_id is not None
-                    else (quota.get("heartbeat_receipt") or {}).get("turn_instance_id")),
-            }, timeout=remaining(), large_local_snapshot=True)
+            intent = select_checkpoint_replan(quota, goal_id=goal_id, agent_id=agent_id,
+                turn_instance_id=(turn_instance_id if turn_instance_id is not None
+                    else (quota.get("heartbeat_receipt") or {}).get("turn_instance_id")), timeout=remaining())
         if not failed_readback and native_process is not None and interrupt_when is not None and interrupt_when(intent):
             from .worker import interrupt_checkpoint_process
 

benchmark/runtime/checkpoint_continuation.py

@@ -12,6 +12,7 @@
 from loopx.control_plane.effect_runtime import effect_runtime_result
 from loopx.control_plane.quota.settlement import read_heartbeat_settlement
 from loopx.control_plane.turn_driver.codex_sessions import load_codex_cli_session
+from loopx.hosts.research_checkpoint import select_checkpoint_replan
 
 
 def retained_native_history(binding, metadata):
@@ -82,11 +83,9 @@ def _restored_context(env, current_obligation_id=None, *, verify_native, include
         if hashlib.sha256(raw).hexdigest() != metadata.get("decision_sha256"):
             raise ValueError("restored pending replan packet digest differs")
         packet = json.loads(raw)
-        intent = effect_runtime_result("runtime.checkpoint_artifacts", {
-            "action": "select_replan", "quota": packet,
-            "goal_id": metadata["loopx_state"]["goal_id"], "agent_id": metadata["loopx_state"]["agent_id"],
-            "turn_instance_id": packet["heartbeat_receipt"]["turn_instance_id"],
-        }, large_local_snapshot=True)
+        intent = select_checkpoint_replan(packet, goal_id=metadata["loopx_state"]["goal_id"],
+            agent_id=metadata["loopx_state"]["agent_id"], turn_instance_id=packet["heartbeat_receipt"]["turn_instance_id"],
+            allow_legacy_source=True)
         if intent != pending or intent.get("capture") is not True:
             raise ValueError("restored original replan binding differs")
     result = effect_runtime_result("runtime.checkpoint_artifacts", {

benchmark/runtime/checkpoint_session_refresh.py

@@ -45,6 +45,7 @@ def _prepare_fresh_replan(env, session, wake, declaration):
     is manufactured or fed to the fresh decision maker.
     """
     from loopx.control_plane.effect_runtime import effect_runtime_result
+    from loopx.hosts.research_checkpoint import select_checkpoint_replan
     from benchmark.runtime.sforge_warm_start import admit_warm_session
 
     unit = Path(declaration["root"])
@@ -61,11 +62,9 @@ def _prepare_fresh_replan(env, session, wake, declaration):
     if hashlib.sha256(raw).hexdigest() != metadata.get("decision_sha256"):
         raise ValueError("fresh replan pending packet digest differs")
     packet = json.loads(raw)
-    selected = effect_runtime_result("runtime.checkpoint_artifacts", {
-        "action": "select_replan", "quota": packet,
-        "goal_id": metadata["loopx_state"]["goal_id"],
-        "agent_id": metadata["loopx_state"]["agent_id"],
-        "turn_instance_id": packet["heartbeat_receipt"]["turn_instance_id"]}, large_local_snapshot=True)
+    selected = select_checkpoint_replan(packet, goal_id=metadata["loopx_state"]["goal_id"],
+        agent_id=metadata["loopx_state"]["agent_id"], turn_instance_id=packet["heartbeat_receipt"]["turn_instance_id"],
+        allow_legacy_source=True)
     if selected.get("capture") is not True or selected != (
             metadata.get("replan") or metadata.get("operator_metadata", {}).get("replan")):
         raise ValueError("fresh replan requires the original pending decision boundary")

benchmark/runtime/codex.py

@@ -88,11 +88,18 @@ def uses_loopx(self) -> bool:
     def native_goal(self) -> bool:
         return self.mode in {"native-goal", "loopx-goal"}
 
+    @property
+    def uses_complete_heartbeat_guard(self) -> bool:
+        return self.mode == "heartbeat" and not self.turn_envelope
+
 
-def turn_envelope_receipt(*, explicit: bool, complete: bool = False) -> dict:
-    """Record only the explicit display opt-in; checkpoint capture is independent."""
-    return {**({"turn_envelope": True, "turn_envelope_source": "explicit"} if explicit else {}),
-        **({"complete_turn": True, "complete_turn_source": "explicit"} if complete else {})}
+def turn_envelope_receipt(execution: Execution) -> dict:
+    """Record the actual heartbeat source choice independently of capture."""
+    if execution.turn_envelope:
+        return {"turn_envelope": True, "turn_envelope_source": "explicit"}
+    if execution.uses_complete_heartbeat_guard:
+        return {"complete_turn": True, "complete_turn_source": "explicit" if execution.complete_turn else "default"}
+    return {}
 
 
 def explicit_turn_display(env: dict[str, str]) -> Execution:

benchmark/runtime/harbor.py

@@ -194,7 +194,7 @@ def __init__(
                 raise ValueError("provider_request_admission requires the Responses wire API")
 
     def _turn_envelope_receipt(self) -> dict:
-        return {**turn_envelope_receipt(explicit=self.execution.turn_envelope, complete=self.execution.complete_turn),
+        return {**turn_envelope_receipt(self.execution),
                 **({"native_child_limit": self.execution.native_child_limit}
                    if self.execution.native_child_limit is not None else {}),
                 **({"context_delivery": "on"} if self.execution.context_delivery != "off" else {})}