fix(heartbeat): keep checkpoint capture independent of input display

benchmark/edgebench/run.py

@@ -329,9 +329,7 @@ def main(argv=None):
            if agent.task_entry == "loopx-planned" else {}),
         "model": args.model, "effort": args.effort, "timeout_seconds": args.timeout,
         "loopx_commit": pins[0], "runner_commit": pins[1],
-        **turn_envelope_receipt(explicit=args.turn_envelope,
-            mode="heartbeat" if args.worker.startswith("heartbeat-") else "plain",
-            checkpoint_request=declaration if checkpoint_request is not None else None),
+        **turn_envelope_receipt(explicit=args.turn_envelope),
         **({"native_goal_continue": True} if args.native_goal_continue else {}),
         "task_sha256": hashlib.sha256(task_file.read_bytes()).hexdigest(),
         "feedback": args.feedback, "internet": task.internet,

benchmark/runtime/RUNTIME.md

@@ -45,6 +45,13 @@ fail before execution. Set it back to `false` for the next run to roll back;
 do not change a running trial's treatment. Captures contain private Goal context.
 Transport tests do not establish lower token cost or better model decisions.
 
+Checkpoint capture is independent of this display experiment. A `fresh_replan`
+declaration may establish a private capture/handoff socket while
+`turn_envelope` remains false: the worker keeps the ordinary dispatcher and quota
+input, and only an explicit true setting selects the short view. This replaces
+the earlier implicit activation by checkpoint handoff. Do not relabel older
+receipts or trajectories as feature-off; qualify their actual input separately.
+
 | Mode | Execution/continuation | LoopX skills and state |
 | --- | --- | --- |
 | `plain` | One Codex exec, native Goals disabled | Absent |

benchmark/runtime/codex.py

@@ -67,19 +67,10 @@ def native_goal(self) -> bool:
         return self.mode in {"native-goal", "loopx-goal"}
 
 
-def turn_envelope_receipt(*, explicit: bool, mode: str,
-                          checkpoint_request: dict | None = None) -> dict:
-    """Project known configuration; a guest path alone proves no activation.
-
-    Callers pass the existing parser's validated declaration. Actual wake
-    receipts still own whether a handoff socket was successfully established.
-    These source labels remain local derived receipt values, not admission.
-    """
+def turn_envelope_receipt(*, explicit: bool) -> dict:
+    """Record only the explicit display opt-in; checkpoint capture is independent."""
     if explicit:
         return {"turn_envelope": True, "turn_envelope_source": "explicit"}
-    if mode == "heartbeat" and checkpoint_request is not None and (
-            checkpoint_request.get("trigger") == "fresh_replan"):
-        return {"turn_envelope": True, "turn_envelope_source": "checkpoint_handoff"}
     return {}
 
 

benchmark/runtime/harbor.py

@@ -145,15 +145,7 @@ def __init__(
         super().__init__(*args, **kwargs)
 
     def _turn_envelope_receipt(self) -> dict:
-        """Project the effective TurnEnvelope setting, not only the explicit opt-in.
-
-        A checkpoint handoff socket makes the worker capture TurnEnvelope on
-        fresh_replan execute wakes; per-wake receipts remain authoritative.
-        """
-        return turn_envelope_receipt(
-            explicit=self.execution.turn_envelope, mode=self.execution.mode,
-            checkpoint_request=self._checkpoint_declaration,
-        )
+        return turn_envelope_receipt(explicit=self.execution.turn_envelope)
 
     def _planning_receipt(self) -> dict:
         return {"native_goal_continue": True} if self.native_goal_continue else {}

benchmark/runtime/worker.py

@@ -67,6 +67,8 @@ def heartbeat_body(env: dict[str, str], turn_id: str, *, native_goal: bool,
         root = Path(env["LOOPX_WAKE_LOG_DIR"]) / turn_id / "decisions"
         root.mkdir(mode=0o700, exist_ok=False)
         command += ["--decision-output-root", str(root)]
+        if env.get("LOOPX_TURN_ENVELOPE") == "1":
+            command += ["--turn-envelope"]
         if decision_handoff_socket is not None:
             command += ["--decision-handoff-socket", decision_handoff_socket]
     result = subprocess.run(
@@ -417,11 +419,6 @@ def run_once(env: dict[str, str]) -> dict:
         elif execution.mode in {"heartbeat", "loopx-goal"}:
             body = heartbeat_body(env, turn_id, native_goal=execution.native_goal,
                                   decision_handoff_socket=checkpoint_wake.socket_path)
-            if checkpoint_wake.socket_path is not None and not execution.turn_envelope:
-                # The handoff socket enables TurnEnvelope capture without the
-                # explicit opt-in; the receipt records the effective setting.
-                receipt["turn_envelope"] = True
-                receipt["turn_envelope_source"] = "checkpoint_handoff"
         elif execution.mode == "plain":
             body = Path(env["LOOPX_TASK_DOC"]).read_text(encoding="utf-8")
         session_wake = None

benchmark/tests/test_native_checkpoint_handoff.py

@@ -529,8 +529,9 @@ def settle_work(result, expected):
         result = run_once({**env, "LOOPX_CHECKPOINT_REQUEST": str(request)})
         assert result["ok"] and result["checkpoint"]["captured"], result
         assert "turn_envelope" not in prior and "turn_envelope_source" not in prior
-        assert result["turn_envelope"] is True
-        assert result["turn_envelope_source"] == "checkpoint_handoff"
+        assert "turn_envelope" not in result and "turn_envelope_source" not in result
+        assert 'Use TurnEnvelope:' not in json.dumps(requests[13])
+        assert 'interaction_contract' in guards[0] and 'action_signature' not in guards[0]
         assert result["checkpoint"]["phase"] == "before_replan"
         assert result["checkpoint"]["source_turn_id"] == prior["turn_id"]
         assert result["session"]["session_id"] == prior["session"]["session_id"]
@@ -710,6 +711,8 @@ def response(payload, index, message):
             assert not list(logs.glob("*/decisions"))
             return
         assert len(result["checkpoint_handoffs"]) == 1
+        assert '--turn-envelope' not in shlex.split(guards[0])
+        assert 'turn_envelope' not in result and 'turn_envelope_source' not in result
         event = result["checkpoint_handoffs"][0]
         if json.loads(request.read_text())["native_session"] and sys.platform == "darwin":
             # Unsupported native interruption is refused before any signal;

benchmark/tests/test_sforge_runtime.py

@@ -182,11 +182,11 @@ async def installed(self, environment):
 
 @pytest.mark.parametrize("explicit,trigger,expected", [
     (False, None, None),
-    (False, "fresh_replan", "checkpoint_handoff"),
+    (False, "fresh_replan", None),
     (True, None, "explicit"),
     (True, "fresh_replan", "explicit"),
 ])
-def test_sforge_profile_uses_validated_checkpoint_trigger(tmp_path, monkeypatch, explicit, trigger, expected):
+def test_sforge_profile_does_not_enable_envelope_for_checkpoint(tmp_path, monkeypatch, explicit, trigger, expected):
     from types import SimpleNamespace
     pytest.importorskip("sforge")
     pytest.importorskip("harbor")

benchmark/tests/test_shared_codex_runtime.py

@@ -651,10 +651,10 @@ def test_default_execution_has_no_independent_turn_deadline(tmp_path):
     (False, None, None),
     (True, None, "explicit"),
     (False, "after_wake", None),
-    (False, "fresh_replan", "checkpoint_handoff"),
+    (False, "fresh_replan", None),
     (True, "fresh_replan", "explicit"),
 ])
-def test_harbor_install_receipt_matches_actual_checkpoint_trigger(tmp_path, monkeypatch, enabled, trigger, expected):
+def test_harbor_install_receipt_keeps_capture_independent_from_display(tmp_path, monkeypatch, enabled, trigger, expected):
     pytest.importorskip("harbor")
     from benchmark.runtime.harbor import BenchmarkCodex, CodexOffline
     from benchmark.runtime.checkpoint_wake import CheckpointWake
@@ -679,7 +679,7 @@ def test_harbor_install_receipt_matches_actual_checkpoint_trigger(tmp_path, monk
         # Real declaration parser and socket lifecycle, independent of the
         # metadata helper. Omitted trigger is legal after-wake capture.
         assert bool(wake.socket_path) is (trigger == "fresh_replan")
-        assert bool(enabled or wake.socket_path) is (expected is not None)
+        assert enabled is (expected is not None)
     commands = []
     command_envs = {}
     async def no_op(*args, **kwargs):

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

@@ -109,7 +109,8 @@ Turn ID. The returned Todo selection command retains the root and requested
 Other repair/replan guard commands still require the caller to retain those
 transport flags. A saved decision is not permission to skip revalidation.
 
-The heartbeat renderer can opt into the same transport:
+The heartbeat renderer can capture decisions while retaining its ordinary
+dispatcher and quota display:
 
 ```bash
 mkdir -m 700 ./private-decisions
@@ -118,13 +119,29 @@ loopx --format json heartbeat-prompt --thin --runtime-profile generic_cli \
   --decision-output-root ./private-decisions
 ```
 
-Execute its guard, read the short action/contract/writeback fields, and use
+Capture and display are independent. `--decision-output-root` saves complete
+decisions; `--decision-handoff-socket` optionally holds their output for a
+caller-owned checkpoint. Neither enables TurnEnvelope. This corrects the old
+renderer behavior where requesting capture also selected the short dispatcher.
+Existing capture callers that intentionally want the short view must now add
+`--turn-envelope` to `heartbeat-prompt`; the generated guard and thin regeneration
+command retain that explicit choice. Quota's existing explicit flag is unchanged.
+
+With that display opt-in, execute its guard, read the short action/contract/writeback fields, and use
 `detail_ref.full_decision` to read the **same invocation** when selection, replan,
 capability context or truncation requires detail. Replan history remains with the
 existing owner; this option neither invents another history store nor chooses a
 strategy. The renderer rejects full/compact/brief modes, missing Turn identity,
 native Goal-owned Turns and invalid roots. Stable bootstrap prompts cannot pin a
-Turn. Omit the capture option to return to the unchanged full-decision dispatcher.
+Turn. Omit `--turn-envelope` to use the ordinary dispatcher, including when
+checkpoint capture remains enabled. Captured files retain the full decision in
+both display modes.
+
+中文：捕获和模型输入投影独立。`--decision-output-root` 仅保存完整决策，
+`--decision-handoff-socket` 仅保留检查点交接；二者均不隐式开启 TurnEnvelope。
+此前捕获目录同时启用短包的行为已修正。需要短包的调用者须对
+`heartbeat-prompt` 显式传 `--turn-envelope`；省略该开关时仍使用普通输入，
+选择、结算、身份与完整私有捕获不变。旧运行的实际输入与设置不得追溯改写。
 
 This is Python filesystem/command transport under the existing quota/context
 owner, not a new capability or Python decision rule. The shared benchmark

loopx/cli_commands/support_control.py

@@ -400,6 +400,7 @@ def handle_support_control_command(
                 turn_instance_id=args.turn_instance_id,
                 decision_output_root=args.decision_output_root,
                 decision_handoff_socket=args.decision_handoff_socket,
+                turn_envelope=args.turn_envelope,
                 reward_memory_enabled=reward_memory_enabled,
             )
             if args.bootstrap and payload.get("ok"):

loopx/cli_commands/support_control_heartbeat_registration.py

@@ -159,7 +159,13 @@ def register_heartbeat_control_commands(
 
     heartbeat_prompt_parser.add_argument(
         "--decision-output-root",
-        help="Opt into the short TurnEnvelope with full local captures; requires thin mode and an explicit host-owned Turn ID.",
+        help="Save full local decisions without changing the display; requires thin mode and an explicit host-owned Turn ID.",
+    )
+
+    heartbeat_prompt_parser.add_argument(
+        "--turn-envelope",
+        action="store_true",
+        help="Explicitly use the short TurnEnvelope dispatcher; requires --decision-output-root. Capture alone keeps the ordinary full decision.",
     )
 
     heartbeat_prompt_parser.add_argument(

loopx/control_plane/heartbeat/builder.py

@@ -176,6 +176,7 @@ def _heartbeat_prompt_commands(
     turn_identity_arg: str,
     decision_output_root: Path | None,
     decision_handoff_socket: str | None,
+    turn_envelope: bool,
 ) -> dict[str, str | None]:
     quota_guard_command = render_quota_guard_command(
         goal_id,
@@ -279,9 +280,10 @@ def _heartbeat_prompt_commands(
         capture_arg = " --decision-output-root " + shlex.quote(str(decision_output_root))
         if decision_handoff_socket is not None:
             capture_arg += " --decision-handoff-socket " + shlex.quote(decision_handoff_socket)
+        display_arg = " --turn-envelope" if turn_envelope else ""
         for key in ("quota_guard_command", "task_body_quota_guard_command"):
-            commands[key] = str(commands[key]) + " --turn-envelope" + capture_arg
-        commands["thin_prompt_command"] = str(commands["thin_prompt_command"]) + capture_arg
+            commands[key] = str(commands[key]) + display_arg + capture_arg
+        commands["thin_prompt_command"] = str(commands["thin_prompt_command"]) + display_arg + capture_arg
     return commands
 
 
@@ -313,6 +315,7 @@ def build_heartbeat_prompt(
     reward_memory_enabled: bool = True,
     decision_output_root: str | Path | None = None,
     decision_handoff_socket: str | None = None,
+    turn_envelope: bool = False,
 ) -> dict[str, Any]:
     if not (full or compact or brief or thin):
         thin = True
@@ -377,6 +380,8 @@ def build_heartbeat_prompt(
         turn_instance_id=normalized_turn_instance_id,
         decision_handoff_socket=decision_handoff_socket,
     )
+    if turn_envelope and decision_output_root is None:
+        raise ValueError("TurnEnvelope requires an explicit decision capture root")
     explicit_agent_scopes = normalize_agent_scopes(agent_scopes)
     if explicit_agent_scopes:
         normalized_agent_scopes = explicit_agent_scopes
@@ -464,6 +469,7 @@ def build_heartbeat_prompt(
         turn_identity_arg=turn_identity_arg,
         decision_output_root=decision_output_root,
         decision_handoff_socket=decision_handoff_socket,
+        turn_envelope=turn_envelope,
     )
     cli_preflight = render_cli_preflight(cli_bin=cli_bin)
     task_body_renderer = _select_task_body_renderer(
@@ -499,13 +505,13 @@ def build_heartbeat_prompt(
         compact_prompt_command=str(commands["compact_prompt_command"]),
         brief_prompt_command=str(commands["brief_prompt_command"]),
         thin_prompt_command=str(commands["thin_prompt_command"]),
-        **({"captured_envelope": True} if decision_output_root is not None else {}),
+        **({"captured_envelope": True} if turn_envelope else {}),
         **reward_memory_rule_kwargs,
     )
     task_body = bind_exact_turn_settlement_task_body(
         task_body,
         turn_instance_id=normalized_turn_instance_id,
-        captured_envelope=decision_output_root is not None,
+        captured_envelope=turn_envelope,
     )
     if fine_grained:
         task_body = f"{task_body}\n\n{FINE_GRAINED_TURN_RULE}"