fix(benchmark): make warm preparation readback revision-consistent

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
@@ -1023,14 +1022,41 @@ 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 结果，才在既有
+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

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,9 +346,13 @@ 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")
-    _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 +361,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 +373,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:

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"]

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: