From 815d9e5267dcc47edf512a62e6d1f4da46783239 Mon Sep 17 00:00:00 2001
From: song <liusongstep@gmail.com>
Date: Sat, 17 Oct 2026 09:53:43 +0800
Subject: [PATCH 1/4] fix(benchmark): verify admitted warm preparation before
 model entry

Carry the current continuation request on the existing preparation Todo and read current Todo/quota facts within the original phase. Keep raw terminal archives, historical authority, native opportunities and later research qualification outside this readiness claim.

Signed-off-by: song <liusongstep@gmail.com>
---
 benchmark/README.md                           | 20 ++++
 benchmark/runtime/sforge_warm_start.py        | 44 +++++++--
 .../hierarchical-agent-stride-control-v0.md   |  8 ++
 ...rarchical-agent-stride-control-v0.zh-CN.md |  5 +
 .../test_warm_continuation_readback.py        | 98 +++++++++++++++++++
 5 files changed, 167 insertions(+), 8 deletions(-)
 create mode 100644 tests/control_plane/test_warm_continuation_readback.py

diff --git a/benchmark/README.md b/benchmark/README.md
index 28396d14e8..4c7d286c82 100644
--- a/benchmark/README.md
+++ b/benchmark/README.md
@@ -1011,6 +1011,26 @@ returns. Preparation and context-reading cost belong in the comparison. Omit all
 restore flags for the unchanged cold entry; never reuse an ambiguous trial
 folder as a new attempt.
 
+Warm preparation carries the admitted continuation request on its existing Todo.
+Earlier Host deadlines and Host-budget closeout messages in retained history
+belong to the source phase; the receiver follows its current Goal and guard.
+Task-owned time constraints, native cooldowns, user gates and stop authority
+remain binding. Preparation reads the current claimed Todo and actual quota
+result before reporting `preparation_actionable=true` in the existing `checkpoint-context.json`
+receipt. Closed or unavailable work and a deadline that expires during readback
+fail before model execution; an old ready plan is not current admission. This
+readback neither grants execution nor qualifies subsequent research or recapture.
+It applies only to the existing explicitly admitted complete-checkpoint path;
+ordinary terminal archives are not converted into complete checkpoints.
+
+Warm 准备在既有 Todo 中传达本次已准入的继续请求：历史中的旧 Host 截止和预算收尾
+信息属于源阶段，目标阶段遵循当前 Goal 与 guard。题目时限、原生冷却、用户门槛及
+停止权限仍有效。准备步骤须读回当前 claimed Todo 和实际 quota 结果，才在既有
+`checkpoint-context.json` 回执中报告 `preparation_actionable=true`；工作已闭合、不可用或读回
+期间期限耗尽时，在模型执行前失败。旧 ready 计划不等于当前准入；此读回不授予
+执行权，也不认证后续研究或再次捕获，仅作用于显式准入的完整检查点路径。
+普通终态资料归档不会因此被升级为完整检查点。
+
 To test a new decision from the same pre-replan state, use
 `--restore-fresh-replan`. The point must contain a verified pending replan packet.
 The first destination model call starts a fresh native conversation, before
diff --git a/benchmark/runtime/sforge_warm_start.py b/benchmark/runtime/sforge_warm_start.py
index 86a15f4c9e..96323127ee 100644
--- a/benchmark/runtime/sforge_warm_start.py
+++ b/benchmark/runtime/sforge_warm_start.py
@@ -22,6 +22,18 @@
 
 
 _CHECKPOINT_VERIFY_TIMEOUT_SECONDS = 60
+_PREPARATION_TASK = (
+    "Read the full retained checkpoint research and pending decision, verify the current code, "
+    "incumbent and remaining native opportunities. Do not modify candidates or submit during "
+    "this preparation. Settle actual findings under current authority, then complete the first "
+    "current replan before choosing a complete research experiment. "
+    "This receiver was explicitly admitted to continue the retained work. Earlier Host deadlines "
+    "and Host-budget closeout messages belong to the source phase; follow the current Goal and "
+    "guard rather than using those historical messages to close this continuation. The Host "
+    "enforces the admitted allowance externally. Original task time constraints, submission "
+    "cooldowns, user gates and stop authority remain binding. No additional time or authority "
+    "is granted by this instruction."
+)
 
 
 @dataclass(frozen=True)
@@ -287,7 +299,8 @@ def prepare_context(env: dict[str, str], declaration: dict) -> dict:
     _prepare_native_skills(Path(env["CODEX_HOME"]), Path(env["LOOPX_SHARED_SKILLS"]))
     def cli(*args):
         result = subprocess.run([*loopx_command(env), *args], env=env,
-            cwd=env["LOOPX_PROJECT"], text=True, capture_output=True, timeout=60, check=True)
+            cwd=env["LOOPX_PROJECT"], text=True, capture_output=True,
+            timeout=_phase_read_timeout(declaration, 60), check=True)
         value = json.loads(result.stdout)
         if value.get("ok") is not True:
             raise ValueError("Checkpoint destination operation failed: " + args[0])
@@ -297,11 +310,7 @@ def cli(*args):
     # model/guard must inspect the retained evidence and settle this Todo.
     todo = cli("todo", "add", "--goal-id", goal, "--role", "agent", "--claimed-by", agent,
         "--operation-id", "checkpoint-preparation", "--required-capability", "shell",
-        "--task-class", "advancement_task", "--text",
-        "Read the full retained checkpoint research and pending decision, verify the current code, "
-        "incumbent and remaining native opportunities. Do not modify candidates or submit during "
-        "this preparation. Settle actual findings under current authority, then complete the first "
-        "current replan before choosing a complete research experiment.")["todo_id"]
+        "--task-class", "advancement_task", "--text", _PREPARATION_TASK)["todo_id"]
     key = "checkpoint-preparation"
     lease = cli("task-lease", "acquire", "--goal-id", goal, "--todo-id", todo,
                 "--owner", agent, "--idempotency-key", key)["lease"]
@@ -315,9 +324,28 @@ def cli(*args):
     cli("task-lease", "release", "--goal-id", goal, "--todo-id", todo,
         "--owner", agent, "--idempotency-key", key, "--expected-version", str(lease["version"]))
     cli("configure-goal", "--goal-id", goal, "--execution-replan-after-turns", "1", "--execute")
+    _read_preparation(cli, goal, agent, todo, declaration)
     return {"session": native, "context": context, "preparation_todo_id": todo,
-            "model_calls": 0, "research_continuation_qualified": False}
-
+            "preparation_actionable": True, "model_calls": 0,
+            "research_continuation_qualified": False}
+
+
+
+def _read_preparation(cli, goal, agent, todo, declaration):
+    """Read current owner facts; a saved plan alone cannot admit continuation."""
+    _phase_read_timeout(declaration, 60)
+    preparation = cli("todo", "list", "--goal-id", goal, "--todo-id", todo)["todo"]
+    if preparation.get("status") != "open" or preparation.get("claimed_by") != agent:
+        raise ValueError("Warm preparation Todo is not open under the destination agent")
+    guard = cli("quota", "should-run", "--goal-id", goal, "--agent-id", agent,
+        "--available-capability", "shell", "--available-capability", "filesystem_write")
+    channel = (guard.get("interaction_contract") or {}).get("agent_channel") or {}
+    selected = guard.get("selected_todo") or {}
+    pending_review = guard.get("autonomous_replan_obligation") or {}
+    if (guard.get("should_run") is not True or channel.get("delivery_allowed") is not True or
+            not (selected.get("todo_id") == todo or pending_review.get("required") is True)):
+        raise ValueError("Warm continuation has no actionable current preparation or review")
+    _phase_read_timeout(declaration, 60)
 
 
 def finish_preparation(env: dict[str, str]) -> dict:
diff --git a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md
index 49336fd626..03908a2a8f 100644
--- a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md
+++ b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md
@@ -1043,6 +1043,14 @@ strategy comparison declares the same extension on both arms and reports the
 shared prefix separately; it is not a matched-total-budget comparison with the
 original source. See the warm-entry options below for activation and rollback.
 
+The warm receiver now places the admitted continuation intent on its existing
+preparation Todo, distinguishing historical Host phase messages from current
+Goal authority without reintroducing a countdown. Before reporting preparation
+ready, it reads the live Todo and quota/interaction owner within the unchanged
+phase. This qualifies current preparation or review readiness only; native work,
+current semantic replan and research benefit retain their separate acceptance.
+Unqualified terminal archives and historical source authority remain inert.
+
 The ordinary SForge worker's explicit warm entry composes these
 owners. It uses a fresh current preparation Todo with full retained evidence,
 then requires an accepted current replan before restoring the normal cadence.
diff --git a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md
index 17d58145c3..7bca0f90ad 100644
--- a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md
+++ b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md
@@ -835,6 +835,11 @@ Warm 接续默认保留捕获的剩余 phase。显式获得 operator 授权的
 相同增量，共享前缀单独报告，不能再与原 source 声称总预算相同。激活与回退
 沿用下方 warm 入口文档中的参数；运行中实验和原结果不重写。
 
+Warm 目标在既有准备 Todo 中传达本次已准入的继续意图，将旧 Host 阶段信息与当前
+Goal 权威区分，不重新注入倒计时。报告准备就绪前，须在原 phase 内读回实际 Todo
+与 quota/交互 owner。此项仅验证当前准备或复盘可执行；原生工作、当前语义复盘和
+研究收益仍按各自验收。未资格化的终态归档与历史来源权限保持惰性。
+
 显式 Docker controller 现提供有界的文件写入屏障：暂停自己拥有的运行中 worker，
 经 daemon archive 复制声明文件及 phase 时钟，捕获原生 evaluator capsule，恢复
 worker 后再发布惰性联合产物。真实持续写入容器与 native HTTP 验证了文件稳定、
diff --git a/tests/control_plane/test_warm_continuation_readback.py b/tests/control_plane/test_warm_continuation_readback.py
new file mode 100644
index 0000000000..3cf02851d4
--- /dev/null
+++ b/tests/control_plane/test_warm_continuation_readback.py
@@ -0,0 +1,98 @@
+"""Continuation readiness uses live Goal owners, not old plan or clock prose."""
+from __future__ import annotations
+
+import time
+
+import pytest
+
+from benchmark.runtime.sforge_warm_start import _PREPARATION_TASK, _read_preparation, restore_workspace
+from loopx.control_plane.effect_runtime import effect_runtime_result
+from test_replan_successor_guard_reentry import AGENT, GOAL, _fixture
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_current_preparation_readback_is_idempotent_and_respects_user_gate(tmp_path, monkeypatch, provider):
+    call, _, _ = _fixture(tmp_path, monkeypatch, provider, future_monitor=False)
+    project = tmp_path / "project"
+    candidate = project / "candidate.txt"
+    candidate.write_text("retained candidate; old Host deadline belongs to the prior phase")
+    component = {"id": "workspace", "source_path": str(project)}
+    probe = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", "components": [component]})
+    point = tmp_path / "captured-point"
+    sealed = effect_runtime_result("runtime.checkpoint_artifacts", {
+        "action": "capture", "destination": str(point), "components": [component],
+        "expected_source_sha256": probe["source_sha256"], "point_metadata": {
+            "native_session": {"guest_workspace": str(project)},
+            "worker_file_snapshot": {"components": [component]},
+        },
+    })
+    (project / "image-only-file").write_text("not part of the retained work")
+    restore_workspace({"root": str(point), "sha256": sealed["sha256"],
+                       "deadline_epoch": int(time.time()) + 600}, project)
+    assert candidate.read_text().startswith("retained candidate")
+    assert not (project / "image-only-file").exists()
+    created = call("todo", "add", "--goal-id", GOAL, "--role", "agent", "--claimed-by", AGENT,
+        "--text", _PREPARATION_TASK, "--task-class", "advancement_task", "--required-capability", "shell",
+        "--operation-id", "checkpoint-preparation")
+    todo = created["todo_id"]
+    deadline = int(time.time()) + 600
+    declaration = {"deadline_epoch": deadline}
+    state = project / "ACTIVE_GOAL_STATE.md"
+    state_before = state.read_bytes()
+    _read_preparation(call, GOAL, AGENT, todo, declaration)
+    _read_preparation(call, GOAL, AGENT, todo, declaration)
+    assert state.read_bytes() == state_before
+    assert declaration["deadline_epoch"] == deadline
+    listed = call("todo", "list", "--goal-id", GOAL, "--todo-id", todo)["todo"]
+    assert listed["status"] == "open" and listed["claimed_by"] == AGENT
+    assert "Earlier Host deadlines" in listed["text"]
+    assert "Original task time constraints" in listed["text"]
+    assert "No additional time or authority" in listed["text"]
+    call("todo", "add", "--goal-id", GOAL, "--role", "user", "--task-class", "user_gate",
+        "--global-gate", "--text", "Authorize the separate destination operation.")
+    with pytest.raises(ValueError, match="no actionable"):
+        _read_preparation(call, GOAL, AGENT, todo, declaration)
+
+
+def test_expired_preparation_does_not_read_or_renew_a_phase():
+    declaration = {"deadline_epoch": 1}
+    calls = []
+    with pytest.raises(ValueError, match="expired"):
+        _read_preparation(lambda *args: calls.append(args), "goal", "agent", "todo", declaration)
+    assert calls == [] and declaration == {"deadline_epoch": 1}
+
+
+@pytest.mark.parametrize("todo", [
+    {"status": "done", "claimed_by": "agent"},
+    {"status": "open", "claimed_by": "another-agent"},
+])
+def test_closed_or_foreign_preparation_cannot_use_a_saved_ready_plan(todo):
+    calls = []
+
+    def read(*args):
+        calls.append(args)
+        return {"todo": todo}
+
+    with pytest.raises(ValueError, match="not open"):
+        _read_preparation(read, "goal", "agent", "todo", {"deadline_epoch": int(time.time()) + 60})
+    assert len(calls) == 1 and calls[0][:2] == ("todo", "list")
+
+
+def test_expiry_during_readback_cannot_report_continuation_ready(tmp_path, monkeypatch):
+    from benchmark.runtime import sforge_warm_start as warm
+
+    call, _, _ = _fixture(tmp_path, monkeypatch, "file", future_monitor=False)
+    todo = call("todo", "add", "--goal-id", GOAL, "--role", "agent", "--claimed-by", AGENT,
+        "--text", _PREPARATION_TASK, "--task-class", "advancement_task")["todo_id"]
+    clock = [time.time()]
+    deadline = int(clock[0]) + 600
+    monkeypatch.setattr(warm, "time", type("Clock", (), {"time": staticmethod(lambda: clock[0])}))
+
+    def expires_after_actual_guard(*args):
+        result = call(*args)
+        if args[:2] == ("quota", "should-run"):
+            clock[0] = deadline
+        return result
+
+    with pytest.raises(ValueError, match="expired"):
+        _read_preparation(expires_after_actual_guard, GOAL, AGENT, todo, {"deadline_epoch": deadline})

From d72cd3dcdbb9450a5d37b51aaf4c989ff7c8d2f9 Mon Sep 17 00:00:00 2001
From: song <liusongstep@gmail.com>
Date: Sat, 17 Oct 2026 10:10:24 +0800
Subject: [PATCH 2/4] test(benchmark): qualify warm preparation through real
 native resume

Signed-off-by: song <liusongstep@gmail.com>
---
 benchmark/tests/test_native_checkpoint.py | 127 ++++++++++++++++------
 1 file changed, 94 insertions(+), 33 deletions(-)

diff --git a/benchmark/tests/test_native_checkpoint.py b/benchmark/tests/test_native_checkpoint.py
index 9c6b58bc90..7204191d0e 100644
--- a/benchmark/tests/test_native_checkpoint.py
+++ b/benchmark/tests/test_native_checkpoint.py
@@ -291,7 +291,7 @@ def test_real_native_workspace_readback_rejects_foreign_code_basis(tmp_path):
 @pytest.mark.skipif(not os.environ.get("LOOPX_NATIVE_CHECKPOINT_BIN"),
                     reason="Requires an explicit real native Host binary; only a local static provider is used")
 @pytest.mark.parametrize("case", ["resume", "scope_change", "profile_change", "joint", "binding_reply_loss",
-                                 "fork_reply_loss", "fork_missing", "fork_ambiguous"])
+                                 "fork_reply_loss", "fork_missing", "fork_ambiguous", "warm_preparation"])
 def test_real_fork_admission_and_native_resume_use_retained_state(tmp_path, capsys, monkeypatch, case, request):
     """Real Host/CLI/Goal/session/artifact owners; deterministic synthetic work.
 
@@ -301,6 +301,7 @@ def test_real_fork_admission_and_native_resume_use_retained_state(tmp_path, caps
     outside this fixture. No benchmark cell or paid model is launched.
     """
     import shutil
+    import shlex
     import subprocess
     import sys
     from pathlib import Path
@@ -323,6 +324,8 @@ def test_real_fork_admission_and_native_resume_use_retained_state(tmp_path, caps
     research.write_text("candidate 3 was rejected by the original synthetic check\n")
     source_goal, source_agent = "source-research", "source-researcher"
     destination_goal, destination_agent = "branch-research", "branch-researcher"
+    from benchmark.runtime.sforge_warm_start import _PREPARATION_TASK, _read_preparation
+
     def register(name, agent):
         registry = tmp_path / name / "registry.json"
         runtime = tmp_path / name / "runtime"
@@ -336,7 +339,9 @@ def register(name, agent):
         capsys.readouterr()
         assert main([*prefix, "todo", "add", "--goal-id", name, "--role", "agent", "--claimed-by", agent,
             "--task-class", "advancement_task", "--required-capability", "shell",
-            "--text", "Verify the retained candidate and preserve applicable negative evidence."]) == 0
+            *(["--operation-id", "checkpoint-preparation"] if case == "warm_preparation" and name == destination_goal else []),
+            "--text", _PREPARATION_TASK if case == "warm_preparation" and name == destination_goal else
+            "Verify the retained candidate and preserve applicable negative evidence."]) == 0
         added = json.loads(capsys.readouterr().out)
         return registry, runtime, added["todo_id"]
     source_registry, source_runtime, source_todo = register(source_goal, source_agent)
@@ -345,9 +350,10 @@ def register(name, agent):
     turn_id = "checkpoint-continuation-fixture"
     binding = ["--goal-id", destination_goal, "--agent-id", destination_agent,
                "--todo-id", destination_todo, "--turn-instance-id", turn_id]
-    assert main([*prefix, "quota", "should-run", "--runtime-profile", "generic_cli", *binding]) == 0
-    guard = json.loads(capsys.readouterr().out)
-    assert guard["interaction_contract"]["agent_channel"]["delivery_allowed"] is True
+    if case != "warm_preparation":
+        assert main([*prefix, "quota", "should-run", "--runtime-profile", "generic_cli", *binding]) == 0
+        guard = json.loads(capsys.readouterr().out)
+        assert guard["interaction_contract"]["agent_channel"]["delivery_allowed"] is True
     from loopx.hosts.research_checkpoint import canonical_context, research_context, adopt_todo_context, _encoded
     source_prefix = ["--format", "json", "--registry", str(source_registry), "--runtime-root", str(source_runtime)]
     def invoke(arguments):
@@ -388,21 +394,45 @@ def invoke(arguments):
 
     def response(request, index, message):
         requests.append(request)
+        if case == "warm_preparation" and index == 1:
+            message["content"][0]["text"] = "The previous Host deadline has passed; the prior phase was closed."
         if index == 2:
+            current_command = [sys.executable, "-m", "loopx.cli", *prefix, "todo", "list",
+                "--goal-id", destination_goal, "--todo-id", destination_todo]
+            # Read the actual Todo but emit only its task text. Whole authority
+            # envelopes can hide the adjacent retained files in native output
+            # truncation, which is unrelated to continuation admission.
+            read_current = (shlex.join([sys.executable, "-c",
+                "import json, subprocess; print(json.loads(subprocess.check_output("
+                + repr(current_command) + ", text=True))['todo']['text'])"]) + " && "
+                if case == "warm_preparation" else "")
             return {"type": "function_call", "id": "fc_retained", "call_id": "call_retained",
                 "name": "exec_command", "arguments": json.dumps({
-                    "cmd": "cat candidate.txt research.txt " + retained_context_path + joint_command, "max_output_tokens": 6000})}
+                    "cmd": read_current + "cat candidate.txt research.txt " + retained_context_path + joint_command,
+                    "max_output_tokens": 6000})}
         if index == 3:
             message["content"][0]["text"] = "Retained candidate 7 verified; candidate 3 remains rejected."
         return message
 
     binary = os.environ["LOOPX_NATIVE_CHECKPOINT_BIN"]
-    execution = Execution(context="resume", sandbox="workspace-write" if case == "joint" else "read-only")
+    # The warm receiver uses the benchmark's execution profile inside its
+    # isolated worker. Other cases retain their original sandbox contract.
+    sandbox = ("danger-full-access" if case == "warm_preparation" else
+               "workspace-write" if case == "joint" else "read-only")
+    execution = Execution(context="resume", sandbox=sandbox)
     with real_native_source(tmp_path / "native", binary, workspace, close_source=True,
                             response_builder=response, sandbox=execution.sandbox) as (env, source_session, _, calls):
         env.update(LOOPX_REGISTRY=str(source_registry), LOOPX_RUNTIME_ROOT=str(source_runtime),
                    LOOPX_GOAL_ID=source_goal, LOOPX_AGENT_ID=source_agent,
                    MODEL_NAME="fixture", REASONING_EFFORT="high", LOOPX_CODEX_SANDBOX=execution.sandbox)
+        if case == "warm_preparation":
+            executable = tmp_path / "receiver-cli"
+            executable.write_text("#!/bin/sh\nexec " + shlex.quote(sys.executable) + ' -m loopx.cli "$@"\n')
+            executable.chmod(0o700)
+            skills = tmp_path / "shared-skills"
+            skills.mkdir()
+            env.update(LOOPX_CLI=str(executable), LOOPX_SHARED_SKILLS=str(skills),
+                       PYTHONPATH=str(Path(__file__).resolve().parents[2]) + os.pathsep + env.get("PYTHONPATH", ""))
         source_wake = BenchmarkSessionWake(env, execution, {})
         source_wake.admission.accept_result(lambda: _store_codex_cli_session(source_runtime,
             lineage={"goal_id": source_goal, "agent_id": source_agent}, session_scope="agent",
@@ -643,26 +673,50 @@ def forbidden_fork(*args, **kwargs):
 
         # Both participants meet at the real destination continuation owner.
         # No source claim/lease is replayed; adoption alone still starts no work.
-        lease = invoke([*prefix, "task-lease", "acquire", "--goal-id", destination_goal,
-            "--todo-id", destination_todo, "--owner", destination_agent, "--idempotency-key", "branch-context"])
-        current = invoke([*prefix, "todo", "list", "--goal-id", destination_goal, "--todo-id", destination_todo])
-        context_command = [sys.executable, "-m", "loopx.hosts.research_checkpoint",
-            "--root", str(unit), "--expected-sha256", digest, "--source-todo-id", source_todo,
-            "--todo-id", destination_todo, "--operation-id", "branch-context",
-            "--expected-revision", current["authority_read"]["provider_revision"],
-            "--lease-key", "branch-context", "--lease-version", str(lease["lease"]["version"])]
-        assert not explore_result_log_path(runtime, destination_goal).exists(), "session admission must not adopt knowledge"
-        adopted_cli = subprocess.run(context_command, env=env, text=True, capture_output=True, timeout=60)
-        assert adopted_cli.returncode == 0, adopted_cli.stderr
-        context_result = json.loads(adopted_cli.stdout)
-        # Retrying the exact CLI after a lost result reuses both original
-        # receipts; it must not fork, charge, edit the Todo or run a model.
-        replay_cli = subprocess.run(context_command, env=env, text=True, capture_output=True, timeout=60)
-        assert replay_cli.returncode == 0, replay_cli.stderr
-        replay_result = json.loads(replay_cli.stdout)
-        assert replay_result["continuation"]["adoption"]["status"] == "replayed"
-        assert replay_result["continuation"]["provider_revision"] == context_result["continuation"]["provider_revision"]
-        assert len(calls) == 1
+        if case == "warm_preparation":
+            declaration = tmp_path / "warm-start.json"
+            payload = {"root": str(unit), "sha256": digest, "config": str(config),
+                "goal_id": destination_goal, "agent_id": destination_agent,
+                "source_todo_id": source_todo, "normal_cadence": 6,
+                "deadline_epoch": int(time.time()) + 600}
+            declaration.write_text(json.dumps(payload))
+            prepared_cli = subprocess.run([sys.executable, "-m", "benchmark.runtime.sforge_warm_start",
+                "context", "--declaration", str(declaration)], env=env, cwd=workspace,
+                text=True, capture_output=True, timeout=90)
+            assert prepared_cli.returncode == 0, prepared_cli.stderr
+            prepared = json.loads(prepared_cli.stdout)
+            assert prepared["preparation_todo_id"] == destination_todo
+            assert prepared["preparation_actionable"] and prepared["model_calls"] == 0
+            assert prepared["research_continuation_qualified"] is False
+            context_result = prepared["context"]
+            _read_preparation(lambda *args: invoke([*prefix, *args]), destination_goal,
+                destination_agent, destination_todo, payload)
+            assert json.loads(declaration.read_text()) == payload
+            guard = invoke([*prefix, "quota", "should-run", "--runtime-profile", "generic_cli", *binding,
+                "--available-capability", "shell", "--available-capability", "filesystem_write"])
+            assert guard["interaction_contract"]["agent_channel"]["delivery_allowed"]
+            assert len(calls) == 1
+        else:
+            lease = invoke([*prefix, "task-lease", "acquire", "--goal-id", destination_goal,
+                "--todo-id", destination_todo, "--owner", destination_agent, "--idempotency-key", "branch-context"])
+            current = invoke([*prefix, "todo", "list", "--goal-id", destination_goal, "--todo-id", destination_todo])
+            context_command = [sys.executable, "-m", "loopx.hosts.research_checkpoint",
+                "--root", str(unit), "--expected-sha256", digest, "--source-todo-id", source_todo,
+                "--todo-id", destination_todo, "--operation-id", "branch-context",
+                "--expected-revision", current["authority_read"]["provider_revision"],
+                "--lease-key", "branch-context", "--lease-version", str(lease["lease"]["version"])]
+            assert not explore_result_log_path(runtime, destination_goal).exists(), "session admission must not adopt knowledge"
+            adopted_cli = subprocess.run(context_command, env=env, text=True, capture_output=True, timeout=60)
+            assert adopted_cli.returncode == 0, adopted_cli.stderr
+            context_result = json.loads(adopted_cli.stdout)
+            # Retrying the exact CLI after a lost result reuses both original
+            # receipts; it must not fork, charge, edit the Todo or run a model.
+            replay_cli = subprocess.run(context_command, env=env, text=True, capture_output=True, timeout=60)
+            assert replay_cli.returncode == 0, replay_cli.stderr
+            replay_result = json.loads(replay_cli.stdout)
+            assert replay_result["continuation"]["adoption"]["status"] == "replayed"
+            assert replay_result["continuation"]["provider_revision"] == context_result["continuation"]["provider_revision"]
+            assert len(calls) == 1
         retained_context_path = str(Path(context_result["artifact"]).relative_to(workspace))
         assert Path(context_result["artifact"]).read_bytes() == context_raw and len(calls) == 1
         adopted_todo = invoke([*prefix, "todo", "list", "--goal-id", destination_goal, "--todo-id", destination_todo])
@@ -734,7 +788,9 @@ def forbidden_fork(*args, **kwargs):
         (logs / "stdout.jsonl").write_text(result.stdout)
         wake.observe(logs / "stdout.jsonl")
         assert wake.session_id == admitted["session_id"] and len(calls) == 3
-        assert "Synthetic source result." in json.dumps(requests[1])
+        retained_message = ("previous Host deadline has passed" if case == "warm_preparation"
+                            else "Synthetic source result.")
+        assert retained_message in json.dumps(requests[1])
         assert destination_goal in json.dumps(requests[1]) and destination_todo in json.dumps(requests[1])
         assert "retained candidate: 7" in json.dumps(requests[2])
         assert "candidate 3 was rejected" in json.dumps(requests[2])
@@ -774,6 +830,10 @@ def forbidden_fork(*args, **kwargs):
             for secret in (source_token, fresh["token"], source_judge.state.judge.admin_secret,
                            branch_judge.state.judge.admin_secret):
                 assert secret not in json.dumps(requests)
+        if case == "warm_preparation":
+            assert "previous Host deadline has passed" in json.dumps(requests[1])
+            assert "Earlier Host deadlines" in json.dumps(requests[2])
+            assert "No additional time or authority" in json.dumps(requests[2])
         assert "Retained candidate 7 verified" in result.stdout
         retained = native_session_checkpoint(env, admitted["session_id"], timeout=20)
         assert retained["metadata"]["turn_ids"][:-1] == native["metadata"]["turn_ids"]
@@ -785,11 +845,12 @@ def forbidden_fork(*args, **kwargs):
             admit_native_checkpoint_session(env, unit, digest, config, sandbox=(execution).sandbox, timeout=30)
         assert load_codex_cli_session(runtime, lineage={"goal_id": destination_goal,
             "agent_id": destination_agent}, session_scope="agent") == after_work
-        with pytest.raises(ValueError, match="native history or configuration moved"):
-            adopt_todo_context(env, unit, digest, source_todo_id=source_todo,
-                todo_id=destination_todo, operation_id="must-not-adopt-later-session",
-                expected_revision=adopted_todo["authority_read"]["provider_revision"],
-                lease_key="branch-context", lease_version=lease["lease"]["version"])
+        if case != "warm_preparation":
+            with pytest.raises(ValueError, match="native history or configuration moved"):
+                adopt_todo_context(env, unit, digest, source_todo_id=source_todo,
+                    todo_id=destination_todo, operation_id="must-not-adopt-later-session",
+                    expected_revision=adopted_todo["authority_read"]["provider_revision"],
+                    lease_key="branch-context", lease_version=lease["lease"]["version"])
         assert len(calls) == 3
         assert invoke([*prefix, "todo", "list", "--goal-id", destination_goal,
                        "--todo-id", destination_todo])["todo"] == adopted_todo["todo"]

From 2600db453a59769b3f5ce013ffb0b8e34b2c01cf Mon Sep 17 00:00:00 2001
From: song <liusongstep@gmail.com>
Date: Sat, 17 Oct 2026 11:50:32 +0800
Subject: [PATCH 3/4] fix(benchmark): bind warm preparation readback to one
 current revision

Bracket the actual guard with the exact preparation Todo authority revision. Refuse interleaved completion or transfer, and expose the observed Todo, revision and current replan identity in the existing context receipt.

Preserve phase deadlines, normal worker admission and owed first review. Cover real File/SQLite interleavings and the actual native warm resume entry without paid inference; retain the parent proposal's other recovery boundaries.

Signed-off-by: song <liusongstep@gmail.com>
---
 benchmark/README.md                           | 19 +++++++++
 benchmark/runtime/sforge_warm_start.py        | 21 +++++++++-
 .../test_warm_continuation_readback.py        | 39 ++++++++++++++++++-
 3 files changed, 75 insertions(+), 4 deletions(-)

diff --git a/benchmark/README.md b/benchmark/README.md
index 4c7d286c82..87a33053b2 100644
--- a/benchmark/README.md
+++ b/benchmark/README.md
@@ -1023,6 +1023,18 @@ readback neither grants execution nor qualifies subsequent research or recapture
 It applies only to the existing explicitly admitted complete-checkpoint path;
 ordinary terminal archives are not converted into complete checkpoints.
 
+The receiver brackets the guard with the exact preparation Todo's canonical
+authority revision. Completion, transfer or another canonical edit during that
+observation rejects readiness; a later required review cannot revive an earlier
+open snapshot. The existing context receipt includes `preparation_readback` with
+that revision, the selected Todo and current replan identity, so an operator can
+inspect what was actually observed instead of relying on a readiness boolean.
+This is a bounded observation, not an execution lease; the worker must still use
+fresh ordinary admission. A rejected attempt retains its context, cadence-one
+preparation and owed review for inspection rather than silently resetting them.
+Cross-version partial receivers and actual model/research adoption retain their
+separate qualification boundaries.
+
 Warm 准备在既有 Todo 中传达本次已准入的继续请求：历史中的旧 Host 截止和预算收尾
 信息属于源阶段，目标阶段遵循当前 Goal 与 guard。题目时限、原生冷却、用户门槛及
 停止权限仍有效。准备步骤须读回当前 claimed Todo 和实际 quota 结果，才在既有
@@ -1031,6 +1043,13 @@ Warm 准备在既有 Todo 中传达本次已准入的继续请求：历史中的
 执行权，也不认证后续研究或再次捕获，仅作用于显式准入的完整检查点路径。
 普通终态资料归档不会因此被升级为完整检查点。
 
+接收端用准确准备 Todo 的规范权威修订号夹住 guard 读取；期间完成、转交或其他规范
+修改都会拒绝 ready，后来的 required review 不能复活较早的 open 快照。既有上下文
+回执的 `preparation_readback` 返回该修订、当前选中 Todo 与 replan 身份，便于检查
+实际观察依据；它不是执行租约，worker 仍须重新走普通准入。拒绝的接收端保留上下文、
+cadence-one 准备和欠下的复盘供检查，不默默重置；跨版本部分接收端与真实模型/研究采用
+仍需各自资格化。
+
 To test a new decision from the same pre-replan state, use
 `--restore-fresh-replan`. The point must contain a verified pending replan packet.
 The first destination model call starts a fresh native conversation, before
diff --git a/benchmark/runtime/sforge_warm_start.py b/benchmark/runtime/sforge_warm_start.py
index 96323127ee..fc8bd0afd0 100644
--- a/benchmark/runtime/sforge_warm_start.py
+++ b/benchmark/runtime/sforge_warm_start.py
@@ -324,8 +324,9 @@ def cli(*args):
     cli("task-lease", "release", "--goal-id", goal, "--todo-id", todo,
         "--owner", agent, "--idempotency-key", key, "--expected-version", str(lease["version"]))
     cli("configure-goal", "--goal-id", goal, "--execution-replan-after-turns", "1", "--execute")
-    _read_preparation(cli, goal, agent, todo, declaration)
+    readback = _read_preparation(cli, goal, agent, todo, declaration)
     return {"session": native, "context": context, "preparation_todo_id": todo,
+            "preparation_readback": readback,
             "preparation_actionable": True, "model_calls": 0,
             "research_continuation_qualified": False}
 
@@ -334,7 +335,8 @@ def cli(*args):
 def _read_preparation(cli, goal, agent, todo, declaration):
     """Read current owner facts; a saved plan alone cannot admit continuation."""
     _phase_read_timeout(declaration, 60)
-    preparation = cli("todo", "list", "--goal-id", goal, "--todo-id", todo)["todo"]
+    before = cli("todo", "list", "--goal-id", goal, "--todo-id", todo)
+    preparation = before["todo"]
     if preparation.get("status") != "open" or preparation.get("claimed_by") != agent:
         raise ValueError("Warm preparation Todo is not open under the destination agent")
     guard = cli("quota", "should-run", "--goal-id", goal, "--agent-id", agent,
@@ -345,7 +347,22 @@ def _read_preparation(cli, goal, agent, todo, declaration):
     if (guard.get("should_run") is not True or channel.get("delivery_allowed") is not True or
             not (selected.get("todo_id") == todo or pending_review.get("required") is True)):
         raise ValueError("Warm continuation has no actionable current preparation or review")
+    # The guard can legitimately require a review after this Todo completes.
+    # That review cannot make an older open/owned preparation snapshot current.
+    # Bracket it with the existing canonical read revision; do not invent a
+    # ready receipt, retry budget, or permission from imported history.
+    after = cli("todo", "list", "--goal-id", goal, "--todo-id", todo)
+    revision = (before.get("authority_read") or {}).get("provider_revision")
+    current = after["todo"]
+    if (not isinstance(revision, str) or not revision
+            or (after.get("authority_read") or {}).get("provider_revision") != revision
+            or current.get("status") != "open" or current.get("claimed_by") != agent):
+        raise ValueError("Warm preparation state changed during current readiness readback; inspect this receiver")
     _phase_read_timeout(declaration, 60)
+    return {"authority_read": after["authority_read"], "preparation_todo_id": todo,
+            "selected_todo_id": selected.get("todo_id"),
+            "current_replan_obligation_id": pending_review.get("obligation_id"),
+            "current_replan_required": pending_review.get("required") is True}
 
 
 def finish_preparation(env: dict[str, str]) -> dict:
diff --git a/tests/control_plane/test_warm_continuation_readback.py b/tests/control_plane/test_warm_continuation_readback.py
index 3cf02851d4..eaaceff9d7 100644
--- a/tests/control_plane/test_warm_continuation_readback.py
+++ b/tests/control_plane/test_warm_continuation_readback.py
@@ -39,8 +39,12 @@ def test_current_preparation_readback_is_idempotent_and_respects_user_gate(tmp_p
     declaration = {"deadline_epoch": deadline}
     state = project / "ACTIVE_GOAL_STATE.md"
     state_before = state.read_bytes()
-    _read_preparation(call, GOAL, AGENT, todo, declaration)
-    _read_preparation(call, GOAL, AGENT, todo, declaration)
+    readback = _read_preparation(call, GOAL, AGENT, todo, declaration)
+    repeated = _read_preparation(call, GOAL, AGENT, todo, declaration)
+    assert readback == repeated
+    assert readback["preparation_todo_id"] == todo
+    assert readback["authority_read"]["provider_revision"]
+    assert readback["selected_todo_id"] == todo or readback["current_replan_required"]
     assert state.read_bytes() == state_before
     assert declaration["deadline_epoch"] == deadline
     listed = call("todo", "list", "--goal-id", GOAL, "--todo-id", todo)["todo"]
@@ -62,6 +66,37 @@ def test_expired_preparation_does_not_read_or_renew_a_phase():
     assert calls == [] and declaration == {"deadline_epoch": 1}
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("boundary", ["before_guard", "after_guard"])
+@pytest.mark.parametrize("change", ["complete", "transfer"])
+def test_changed_preparation_cannot_borrow_an_unrelated_current_review(
+    tmp_path, monkeypatch, provider, boundary, change,
+):
+    call, _, _ = _fixture(tmp_path, monkeypatch, provider, future_monitor=False)
+    todo = call("todo", "add", "--goal-id", GOAL, "--role", "agent", "--claimed-by", AGENT,
+        "--text", _PREPARATION_TASK, "--task-class", "advancement_task")["todo_id"]
+    changed = False
+
+    def interleaved(*args):
+        nonlocal changed
+        result = call(*args)
+        trigger = ("todo", "list") if boundary == "before_guard" else ("quota", "should-run")
+        if not changed and args[:2] == trigger:
+            changed = True
+            if change == "complete":
+                call("todo", "complete", "--goal-id", GOAL, "--agent-id", AGENT, "--todo-id", todo)
+            else:
+                call("todo", "update", "--goal-id", GOAL, "--agent-id", AGENT, "--todo-id", todo,
+                     "--claimed-by", "different-agent")
+        return result
+
+    with pytest.raises(ValueError, match="preparation.*changed|not open"):
+        _read_preparation(interleaved, GOAL, AGENT, todo, {"deadline_epoch": int(time.time()) + 600})
+    assert changed
+    final = call("todo", "list", "--goal-id", GOAL, "--todo-id", todo)["todo"]
+    assert final["status"] == "done" if change == "complete" else final["claimed_by"] == "different-agent"
+
+
 @pytest.mark.parametrize("todo", [
     {"status": "done", "claimed_by": "agent"},
     {"status": "open", "claimed_by": "another-agent"},

From 79c00981c72491851034d6f4e1cf5b471e6ac714 Mon Sep 17 00:00:00 2001
From: song <liusongstep@gmail.com>
Date: Sat, 17 Oct 2026 12:27:02 +0800
Subject: [PATCH 4/4] fix(benchmark): preserve preparation intent and defer
 cadence until ready

Signed-off-by: song <liusongstep@gmail.com>
---
 benchmark/README.md                           | 37 +++++----
 benchmark/runtime/sforge_warm_start.py        | 48 +++++++++---
 benchmark/tests/test_native_checkpoint.py     |  4 +-
 .../test_replan_successor_guard_reentry.py    |  4 +-
 .../test_warm_continuation_readback.py        | 76 ++++++++++++++++++-
 5 files changed, 136 insertions(+), 33 deletions(-)

diff --git a/benchmark/README.md b/benchmark/README.md
index 87a33053b2..3242892821 100644
--- a/benchmark/README.md
+++ b/benchmark/README.md
@@ -1011,10 +1011,9 @@ returns. Preparation and context-reading cost belong in the comparison. Omit all
 restore flags for the unchanged cold entry; never reuse an ambiguous trial
 folder as a new attempt.
 
-Warm preparation carries the admitted continuation request on its existing Todo.
-Earlier Host deadlines and Host-budget closeout messages in retained history
-belong to the source phase; the receiver follows its current Goal and guard.
-Task-owned time constraints, native cooldowns, user gates and stop authority
+Warm preparation retains the original evidence-reading task on its existing Todo;
+it does not add a prompt paragraph interpreting historical deadlines. Current
+Goal/guard authority, task time constraints, native cooldowns and user stops
 remain binding. Preparation reads the current claimed Todo and actual quota
 result before reporting `preparation_actionable=true` in the existing `checkpoint-context.json`
 receipt. Closed or unavailable work and a deadline that expires during readback
@@ -1030,14 +1029,19 @@ open snapshot. The existing context receipt includes `preparation_readback` with
 that revision, the selected Todo and current replan identity, so an operator can
 inspect what was actually observed instead of relying on a readiness boolean.
 This is a bounded observation, not an execution lease; the worker must still use
-fresh ordinary admission. A rejected attempt retains its context, cadence-one
-preparation and owed review for inspection rather than silently resetting them.
-Cross-version partial receivers and actual model/research adoption retain their
-separate qualification boundaries.
-
-Warm 准备在既有 Todo 中传达本次已准入的继续请求：历史中的旧 Host 截止和预算收尾
-信息属于源阶段，目标阶段遵循当前 Goal 与 guard。题目时限、原生冷却、用户门槛及
-停止权限仍有效。准备步骤须读回当前 claimed Todo 和实际 quota 结果，才在既有
+fresh ordinary admission. Readiness is validated before changing cadence, so a
+rejected preparation does not change the Goal's cadence. An ambiguous cadence
+write must be inspected, not blindly rolled back across an owed current review.
+Creation retries read the original `checkpoint-preparation` receipt and replay
+its exact accepted task through the canonical create owner, including receivers
+created with the earlier longer task. They never mint a second operation or Todo.
+This covers partial creation, not arbitrary retries after context adoption or
+worker execution. Such uncertain operations retain their original receipt,
+revision and lease recovery boundary; actual model/research adoption is separate.
+
+Warm 准备保留既有 Todo 中原本的证据读取任务，不再增加解释历史截止时间的提示段落。
+当前 Goal、guard、题目时限、原生冷却、用户门槛及停止权限仍有效。
+准备步骤须读回当前 claimed Todo 和实际 quota 结果，才在既有
 `checkpoint-context.json` 回执中报告 `preparation_actionable=true`；工作已闭合、不可用或读回
 期间期限耗尽时，在模型执行前失败。旧 ready 计划不等于当前准入；此读回不授予
 执行权，也不认证后续研究或再次捕获，仅作用于显式准入的完整检查点路径。
@@ -1046,9 +1050,12 @@ Warm 准备在既有 Todo 中传达本次已准入的继续请求：历史中的
 接收端用准确准备 Todo 的规范权威修订号夹住 guard 读取；期间完成、转交或其他规范
 修改都会拒绝 ready，后来的 required review 不能复活较早的 open 快照。既有上下文
 回执的 `preparation_readback` 返回该修订、当前选中 Todo 与 replan 身份，便于检查
-实际观察依据；它不是执行租约，worker 仍须重新走普通准入。拒绝的接收端保留上下文、
-cadence-one 准备和欠下的复盘供检查，不默默重置；跨版本部分接收端与真实模型/研究采用
-仍需各自资格化。
+实际观察依据；它不是执行租约，worker 仍须重新走普通准入。先验证准备资格，再改
+cadence，因此拒绝准备不会改变 Goal 的 cadence；不确定的配置写入须检查，不能盲目
+回滚而抹掉欠下的当前复盘。创建重试读回原 `checkpoint-preparation` 回执，通过规范
+创建 owner 重放当时接受的准确任务（兼容此前较长任务），不创建第二个操作或 Todo。
+此兼容范围是部分创建，不是上下文采用或 worker 执行后的任意重试；不确定的操作仍须
+保留原回执、修订与租约的恢复边界，真实模型/研究采用也需单独验证。
 
 To test a new decision from the same pre-replan state, use
 `--restore-fresh-replan`. The point must contain a verified pending replan packet.
diff --git a/benchmark/runtime/sforge_warm_start.py b/benchmark/runtime/sforge_warm_start.py
index fc8bd0afd0..ae5ef89007 100644
--- a/benchmark/runtime/sforge_warm_start.py
+++ b/benchmark/runtime/sforge_warm_start.py
@@ -26,14 +26,11 @@
     "Read the full retained checkpoint research and pending decision, verify the current code, "
     "incumbent and remaining native opportunities. Do not modify candidates or submit during "
     "this preparation. Settle actual findings under current authority, then complete the first "
-    "current replan before choosing a complete research experiment. "
-    "This receiver was explicitly admitted to continue the retained work. Earlier Host deadlines "
-    "and Host-budget closeout messages belong to the source phase; follow the current Goal and "
-    "guard rather than using those historical messages to close this continuation. The Host "
-    "enforces the admitted allowance externally. Original task time constraints, submission "
-    "cooldowns, user gates and stop authority remain binding. No additional time or authority "
-    "is granted by this instruction."
+    "current replan before choosing a complete research experiment."
 )
+# Compatibility for the already accepted longer preparation request. Never
+# submit that guidance to a new receiver or rewrite an existing create receipt.
+_LEGACY_PREPARATION_TEXT_SHA256 = "c1480d8fcec20909567f39bac1b0af2a0a3d3703c56556f4356e5b3e43e0ba45"
 
 
 @dataclass(frozen=True)
@@ -283,6 +280,34 @@ def admit_warm_session(env: dict[str, str], declaration: dict) -> dict:
     return result
 
 
+def _preparation_todo(cli, goal: str, agent: str) -> str:
+    """Replay the exact accepted create payload; history grants no readiness."""
+    operation = "checkpoint-preparation"
+    found = cli("todo", "receipt", "--goal-id", goal, "--operation-id", operation)
+    text = _PREPARATION_TASK
+    if found.get("status") == "found":
+        receipts = found.get("receipts")
+        original = receipts[0] if isinstance(receipts, list) and len(receipts) == 1 else {}
+        todo = original.get("todo") or {}
+        # Canonical presentation may decorate text with a priority prefix;
+        # title retains the create intent. The create owner checks its digest.
+        text = todo.get("title", todo.get("text"))
+        if (original.get("schema_version") != "loopx_coordination_todo_create_receipt_v0"
+                or original.get("goal_id") != goal or original.get("operation_id") != operation
+                or not isinstance(text, str)
+                or (text != _PREPARATION_TASK and hashlib.sha256(text.encode()).hexdigest()
+                    != _LEGACY_PREPARATION_TEXT_SHA256)):
+            raise ValueError("Warm preparation operation names another request; inspect its original receipt")
+    elif found.get("status") != "missing":
+        raise ValueError("Warm preparation create receipt is unavailable; preserve its operation identity")
+    # The canonical owner still compares the complete request digest, including
+    # owner/capabilities/class. Do not derive a second Todo or accept a receipt
+    # as permission to execute a completed/transferred preparation.
+    return cli("todo", "add", "--goal-id", goal, "--role", "agent", "--claimed-by", agent,
+        "--operation-id", operation, "--required-capability", "shell",
+        "--task-class", "advancement_task", "--text", text)["todo_id"]
+
+
 def prepare_context(env: dict[str, str], declaration: dict) -> dict:
     """Join actual session + research owners on one freshly bootstrapped Goal."""
     from loopx.hosts.research_checkpoint import adopt_todo_context
@@ -308,9 +333,7 @@ def cli(*args):
     goal, agent = declaration["goal_id"], declaration["agent_id"]
     # Preparation is actual current work, not imported completion. The normal
     # model/guard must inspect the retained evidence and settle this Todo.
-    todo = cli("todo", "add", "--goal-id", goal, "--role", "agent", "--claimed-by", agent,
-        "--operation-id", "checkpoint-preparation", "--required-capability", "shell",
-        "--task-class", "advancement_task", "--text", _PREPARATION_TASK)["todo_id"]
+    todo = _preparation_todo(cli, goal, agent)
     key = "checkpoint-preparation"
     lease = cli("task-lease", "acquire", "--goal-id", goal, "--todo-id", todo,
                 "--owner", agent, "--idempotency-key", key)["lease"]
@@ -323,8 +346,11 @@ def cli(*args):
     # finished before this lease is released; no historical lease is imported.
     cli("task-lease", "release", "--goal-id", goal, "--todo-id", todo,
         "--owner", agent, "--idempotency-key", key, "--expected-version", str(lease["version"]))
-    cli("configure-goal", "--goal-id", goal, "--execution-replan-after-turns", "1", "--execute")
     readback = _read_preparation(cli, goal, agent, todo, declaration)
+    # Rejected readiness must not mutate cadence. Once this succeeds the worker
+    # still acquires fresh ordinary admission; no best-effort rollback may erase
+    # an owed current review after an ambiguous configuration response.
+    cli("configure-goal", "--goal-id", goal, "--execution-replan-after-turns", "1", "--execute")
     return {"session": native, "context": context, "preparation_todo_id": todo,
             "preparation_readback": readback,
             "preparation_actionable": True, "model_calls": 0,
diff --git a/benchmark/tests/test_native_checkpoint.py b/benchmark/tests/test_native_checkpoint.py
index 7204191d0e..c3e85b19e7 100644
--- a/benchmark/tests/test_native_checkpoint.py
+++ b/benchmark/tests/test_native_checkpoint.py
@@ -832,8 +832,8 @@ def forbidden_fork(*args, **kwargs):
                 assert secret not in json.dumps(requests)
         if case == "warm_preparation":
             assert "previous Host deadline has passed" in json.dumps(requests[1])
-            assert "Earlier Host deadlines" in json.dumps(requests[2])
-            assert "No additional time or authority" in json.dumps(requests[2])
+            assert _PREPARATION_TASK in json.dumps(requests[2])
+            assert "Earlier Host deadlines" not in json.dumps(requests[2])
         assert "Retained candidate 7 verified" in result.stdout
         retained = native_session_checkpoint(env, admitted["session_id"], timeout=20)
         assert retained["metadata"]["turn_ids"][:-1] == native["metadata"]["turn_ids"]
diff --git a/tests/control_plane/test_replan_successor_guard_reentry.py b/tests/control_plane/test_replan_successor_guard_reentry.py
index de8326e6ff..fc64a76a31 100644
--- a/tests/control_plane/test_replan_successor_guard_reentry.py
+++ b/tests/control_plane/test_replan_successor_guard_reentry.py
@@ -24,7 +24,7 @@
 ROOT = Path(loopx.__file__).resolve().parents[1]
 
 
-def _fixture(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, *, later_vision: bool = False, future_monitor: bool = True, initial_todos: list[dict] | None = None):
+def _fixture(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, *, later_vision: bool = False, future_monitor: bool = True, initial_todos: list[dict] | None = None, handoff_mode: str = "soft_claim"):
     if provider == "sqlite":
         isolate_sqlite_runtime(tmp_path, monkeypatch)
     project, runtime = tmp_path / "project", tmp_path / "runtime"
@@ -67,7 +67,7 @@ def _fixture(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, *,
         "coordination": {"agent_model": "peer_v1", "registered_agents": [AGENT, "different-agent"]},
     }]}))
     initialize_canonical_authority(runtime, GOAL, build_todo_runtime_shadow_projection(
-        goal_id=GOAL, todos=initial_todos or [], handoff_mode="soft_claim", leases=[],
+        goal_id=GOAL, todos=initial_todos or [], handoff_mode=handoff_mode, leases=[],
     ), state_path=state, provider=provider)
 
     def call(*args: str, expected_code: int = 0) -> dict:
diff --git a/tests/control_plane/test_warm_continuation_readback.py b/tests/control_plane/test_warm_continuation_readback.py
index eaaceff9d7..c7ca5d548f 100644
--- a/tests/control_plane/test_warm_continuation_readback.py
+++ b/tests/control_plane/test_warm_continuation_readback.py
@@ -49,9 +49,7 @@ def test_current_preparation_readback_is_idempotent_and_respects_user_gate(tmp_p
     assert declaration["deadline_epoch"] == deadline
     listed = call("todo", "list", "--goal-id", GOAL, "--todo-id", todo)["todo"]
     assert listed["status"] == "open" and listed["claimed_by"] == AGENT
-    assert "Earlier Host deadlines" in listed["text"]
-    assert "Original task time constraints" in listed["text"]
-    assert "No additional time or authority" in listed["text"]
+    assert listed["title"] == _PREPARATION_TASK
     call("todo", "add", "--goal-id", GOAL, "--role", "user", "--task-class", "user_gate",
         "--global-gate", "--text", "Authorize the separate destination operation.")
     with pytest.raises(ValueError, match="no actionable"):
@@ -131,3 +129,75 @@ def expires_after_actual_guard(*args):
 
     with pytest.raises(ValueError, match="expired"):
         _read_preparation(expires_after_actual_guard, GOAL, AGENT, todo, {"deadline_epoch": deadline})
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("legacy", [False, True])
+def test_preparation_create_reuses_original_payload_without_a_second_todo(tmp_path, monkeypatch, provider, legacy):
+    from benchmark.runtime.sforge_warm_start import _preparation_todo
+
+    call, _, _ = _fixture(tmp_path, monkeypatch, provider, future_monitor=False)
+    text = _PREPARATION_TASK
+    if legacy:
+        text += (
+            " This receiver was explicitly admitted to continue the retained work. Earlier Host deadlines "
+            "and Host-budget closeout messages belong to the source phase; follow the current Goal and "
+            "guard rather than using those historical messages to close this continuation. The Host "
+            "enforces the admitted allowance externally. Original task time constraints, submission "
+            "cooldowns, user gates and stop authority remain binding. No additional time or authority "
+            "is granted by this instruction."
+        )
+    original = call("todo", "add", "--goal-id", GOAL, "--role", "agent", "--claimed-by", AGENT,
+        "--text", text, "--task-class", "advancement_task", "--required-capability", "shell",
+        "--operation-id", "checkpoint-preparation")["todo_id"]
+    before = call("todo", "list", "--goal-id", GOAL)
+    assert _preparation_todo(call, GOAL, AGENT) == original
+    assert _preparation_todo(call, GOAL, AGENT) == original
+    after = call("todo", "list", "--goal-id", GOAL)
+    assert after["authority_read"] == before["authority_read"]
+    assert after["todos"] == before["todos"]
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_preparation_operation_cannot_be_repurposed(tmp_path, monkeypatch, provider):
+    from benchmark.runtime.sforge_warm_start import _preparation_todo
+
+    call, _, _ = _fixture(tmp_path, monkeypatch, provider, future_monitor=False)
+    call("todo", "add", "--goal-id", GOAL, "--role", "agent", "--claimed-by", AGENT,
+        "--text", "Different work", "--operation-id", "checkpoint-preparation")
+    before = call("todo", "list", "--goal-id", GOAL)
+    with pytest.raises(ValueError, match="another request"):
+        _preparation_todo(call, GOAL, AGENT)
+    assert call("todo", "list", "--goal-id", GOAL)["authority_read"] == before["authority_read"]
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_rejected_context_preparation_does_not_change_cadence(tmp_path, monkeypatch, provider):
+    import json
+    import os
+    import sys
+    from benchmark.runtime import sforge_warm_start as warm, worker
+    from loopx.hosts import research_checkpoint
+
+    call, _, _ = _fixture(tmp_path, monkeypatch, provider, future_monitor=False, handoff_mode="hard_lease")
+    registry = tmp_path / "registry.json"
+    call("configure-goal", "--goal-id", GOAL, "--execution-replan-after-turns", "6", "--execute")
+    call("todo", "add", "--goal-id", GOAL, "--role", "user", "--task-class", "user_gate",
+         "--global-gate", "--text", "Wait for the user's authorization")
+    before = registry.read_bytes()
+    home, skills = tmp_path / "home", tmp_path / "skills"
+    home.mkdir()
+    skills.mkdir()
+    env = {**os.environ, "LOOPX_GOAL_ID": GOAL, "LOOPX_AGENT_ID": AGENT,
+           "CODEX_HOME": str(home), "LOOPX_SHARED_SKILLS": str(skills),
+           "LOOPX_PROJECT": str(tmp_path / "project")}
+    monkeypatch.setattr(warm, "admit_warm_session", lambda *_: {})
+    monkeypatch.setattr(research_checkpoint, "adopt_todo_context", lambda *_, **__: {})
+    monkeypatch.setattr(worker, "loopx_command", lambda _: [sys.executable, "-m", "loopx.cli",
+        "--registry", str(registry), "--runtime-root", str(tmp_path / "runtime"), "--format", "json"])
+    with pytest.raises(ValueError, match="no actionable"):
+        warm.prepare_context(env, {"root": str(tmp_path / "point"), "sha256": "a" * 64,
+            "goal_id": GOAL, "agent_id": AGENT, "source_todo_id": "source", "normal_cadence": 6,
+            "deadline_epoch": int(time.time()) + 600})
+    assert registry.read_bytes() == before
+    assert json.loads(registry.read_bytes()) == json.loads(before)