diff --git a/benchmark/README.md b/benchmark/README.md
index b1d9d6fbce..378af0a938 100644
--- a/benchmark/README.md
+++ b/benchmark/README.md
@@ -1022,6 +1022,52 @@ 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 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
+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.
+
+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. 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 计划不等于当前准入；此读回不授予
+执行权，也不认证后续研究或再次捕获，仅作用于显式准入的完整检查点路径。
+普通终态资料归档不会因此被升级为完整检查点。
+
+接收端用准确准备 Todo 的规范权威修订号夹住 guard 读取；期间完成、转交或其他规范
+修改都会拒绝 ready，后来的 required review 不能复活较早的 open 快照。既有上下文
+回执的 `preparation_readback` 返回该修订、当前选中 Todo 与 replan 身份，便于检查
+实际观察依据；它不是执行租约，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.
 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 58e6471466..91ab1c7a6c 100644
--- a/benchmark/runtime/sforge_warm_start.py
+++ b/benchmark/runtime/sforge_warm_start.py
@@ -22,6 +22,15 @@
 
 
 _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."
+)
+# 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)
@@ -271,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
@@ -287,7 +324,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])
@@ -295,13 +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",
-        "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"]
+    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"]
@@ -314,10 +346,49 @@ 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"]))
+    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,
-            "model_calls": 0, "research_continuation_qualified": False}
-
+            "preparation_readback": readback,
+            "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)
+    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,
+        "--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")
+    # 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/benchmark/tests/test_native_checkpoint.py b/benchmark/tests/test_native_checkpoint.py
index 9c6b58bc90..c3e85b19e7 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 _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"]
@@ -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"]
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_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
new file mode 100644
index 0000000000..c7ca5d548f
--- /dev/null
+++ b/tests/control_plane/test_warm_continuation_readback.py
@@ -0,0 +1,203 @@
+"""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()
+    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"]
+    assert listed["status"] == "open" and listed["claimed_by"] == AGENT
+    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"):
+        _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("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"},
+])
+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})
+
+
+@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)