fix: guide reopened work back through the original Turn

docs/quota-allocation.md

@@ -1001,6 +1001,19 @@ settlement still closes this resume path. This differs from executable work
 which became deferred after admission: that existing no-execution/no-spend
 boundary remains intact, and independent work is selected under a fresh Turn.
 
+For a Todo-bound resume, the generated CLI sequence therefore places the
+original Turn's guard immediately after reopening, rather than instructing
+the worker to settle and wait for another wake. Read that guard before
+delivery and acquire the required fresh execution claim/lease; only admission
+of the same Todo permits continued work. Validate and settle that original
+binding once after the authorized work. A refusal, changed wait, supersession
+or no-follow-up decision still follows its own recovery/closeout contract.
+
+Todo 绑定的恢复命令因此在重开后直接给出原 Turn 的 guard，而非要求先结算再等一次
+唤醒。执行前须读取 guard 并取得所需的新执行 claim/lease；仅同一 Todo 获得准入后
+才能继续工作，完成授权工作后验证并结算原绑定一次。拒绝、新等待、替代或无需后续
+的决定仍遵循对应的恢复／关闭契约；重开本身不授予执行权限。
+
 Priority remains authoritative across the resume boundary. When a ready
 current-agent or unclaimed deferred successor has strictly higher priority than
 the selected open advancement todo, quota also returns

loopx/control_plane/work_items/interaction_contract.py

@@ -579,6 +579,27 @@ def _render_replan_successor_closeout_guard(
     return typed_quota_guard + settlement_args
 
 
+def _deferred_resume_continuation(
+    todo_id: str, settlement_plan: Mapping[str, Any] | None,
+    scheduler_args: str, typed_quota_guard: str,
+) -> list[str] | None:
+    """Render the existing same-Todo admission path; reopening is not a grant."""
+    identity = (settlement_plan or {}).get("identity")
+    binding_args = settlement_binding_args(settlement_plan)
+    if (not scheduler_args or not binding_args or not isinstance(identity, Mapping)
+            or identity.get("todo_id") != todo_id):
+        return None
+    return [
+        typed_quota_guard + binding_args,
+        "After reopening, read the original Turn's guard above before delivery. "
+        "Only if it admits this same Todo, acquire the required fresh execution "
+        "claim/lease and continue its authorized work in this Turn; then validate "
+        "and settle the original binding once. If it refuses, follow its current "
+        "recovery or stop condition. Reopening alone grants no execution authority "
+        "and does not require a separate settle-and-wake cycle.",
+    ]
+
+
 def _dependency_cycle_cli_actions(
     frontier: dict[str, Any], *, command_prefix: str, goal_id: str,
     lifecycle_actor_args: str, typed_quota_guard: str,
@@ -850,12 +871,17 @@ def _admitted_interaction_next_cli_actions(
         )
         first_candidate = candidates[0] if candidates and isinstance(candidates[0], dict) else {}
         todo_id = str(first_candidate.get("todo_id") or "<todo_id>")
-        return [
-            (
+        reopen = (
             f"{command_prefix} todo update --goal-id {goal_id} --todo-id {todo_id}"
             f"{lifecycle_actor_args} --status open --clear-resume-when "
             "--reason '<public-safe successor replan reason>'"
-            ),
+        )
+        if continuation := _deferred_resume_continuation(
+            todo_id, settlement_plan, scheduler_args, typed_quota_guard,
+        ):
+            return [reopen, *continuation]
+        return [
+            reopen,
             f"{command_prefix} refresh-state --goal-id {goal_id} --classification successor_replan_recorded --delivery-batch-scale single_surface --delivery-outcome outcome_progress{settlement_args}{scoped_cli_args}",
             quota_spend_action,
         ]

tests/control_plane/test_deferred_resume_generated_command.py

@@ -33,12 +33,14 @@
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 @pytest.mark.parametrize("fallback", [False, True])
+@pytest.mark.parametrize("host_args", [("--codex-app",), ("--runtime-profile", "generic_cli")])
 def test_generated_deferred_resume_keeps_proof_fences_with_planning_note(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
     capsys: pytest.CaptureFixture[str],
     provider: str,
     fallback: bool,
+    host_args: tuple[str, ...],
 ) -> None:
     if provider == "sqlite":
         isolate_sqlite_runtime(tmp_path, monkeypatch)
@@ -95,6 +97,8 @@ def test_generated_deferred_resume_keeps_proof_fences_with_planning_note(
     )
 
     def cli(*args: str) -> tuple[int, dict]:
+        if ("quota", "should-run") in zip(args, args[1:]):
+            args = (*args, "--context-delivery", "off")
         with monkeypatch.context() as context:
             context.chdir(project)
             rc = cli_main(
@@ -113,7 +117,7 @@ def cli(*args: str) -> tuple[int, dict]:
     rc, guard = cli(
         "quota",
         "should-run",
-        "--codex-app",
+        *host_args,
         "--goal-id",
         GOAL_ID,
         "--agent-id",
@@ -197,32 +201,14 @@ def cli(*args: str) -> tuple[int, dict]:
         "task-lease", "inspect", "--goal-id", GOAL_ID, "--todo-id", TODO_ID
     )
     assert rc == 0 and absent["active"] is False, absent
-    rc, acquired = cli(
-        "task-lease",
-        "acquire",
-        "--goal-id",
-        GOAL_ID,
-        "--todo-id",
-        TODO_ID,
-        "--owner",
-        AGENT_ID,
-        "--idempotency-key",
-        "fresh-after-resume",
-    )
-    assert rc == 0 and acquired["lease"]["status"] == "active", acquired
-    rc, reentry = cli(
-        "quota",
-        "should-run",
-        "--codex-app",
-        "--goal-id",
-        GOAL_ID,
-        "--agent-id",
-        AGENT_ID,
-        "--turn-instance-id",
-        "resume-command-turn",
-        "--todo-id",
-        TODO_ID,
-    )
+    actions = guard["interaction_contract"]["cli_channel"]["next_cli_actions"]
+    # Execute the delivered continuation instead of constructing a successful
+    # guard that the consumer was never shown. No new Turn or premature spend.
+    reentry_args = shlex.split(actions[1])[1:]
+    assert reentry_args[reentry_args.index("--turn-instance-id") + 1] == "resume-command-turn"
+    assert reentry_args[reentry_args.index("--todo-id") + 1] == TODO_ID
+    assert not any("refresh-state" in action or "spend-slot" in action for action in actions)
+    rc, reentry = cli(*reentry_args)
     assert rc == 0 and reentry["normal_delivery_allowed"], reentry
     assert (
         reentry["heartbeat_receipt"]["settlement_identity"]
@@ -237,12 +223,28 @@ def cli(*args: str) -> tuple[int, dict]:
     qualified = next(row for row in events if row["event_id"] == reentry["heartbeat_receipt"]["event_id"])
     assert qualified["causality"]["source_event_id"] == original["event_id"]
     assert qualified["details"]["delivery_allowed"] is True
-    rc, cold = cli("quota", "should-run", "--codex-app", "--goal-id", GOAL_ID,
+    rc, cold = cli("quota", "should-run", *host_args, "--goal-id", GOAL_ID,
         "--agent-id", AGENT_ID, "--turn-instance-id", "resume-command-turn",
         "--todo-id", TODO_ID, )
     assert rc == 0 and cold["heartbeat_receipt"]["status"] == "replayed", cold
     assert cold["heartbeat_receipt"]["event_id"] == qualified["event_id"]
     assert load_rollout_events(log_path) == events
+    rc, acquired = cli(
+        "task-lease",
+        "acquire",
+        "--goal-id",
+        GOAL_ID,
+        "--todo-id",
+        TODO_ID,
+        "--owner",
+        AGENT_ID,
+        "--idempotency-key",
+        "fresh-after-resume",
+    )
+    assert rc == 0 and acquired["lease"]["status"] == "active", acquired
+    artifact = project / "verified-result.txt"
+    artifact.write_text(str(17 * 23))
+    assert artifact.read_text() == "391"
     child = ("native-child", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
              "--turn-instance-id", "resume-command-turn")
     rc, decision = cli(*child, "record", "--operation-id", "resumed-audit",
@@ -261,3 +263,25 @@ def cli(*args: str) -> tuple[int, dict]:
         "--validation-ref", "synthetic-validation", "--execute")
     assert rc == 0 and reviewed["native_child_activity"]["parent_accepted_count"] == 1
     assert _spend_run_count(runtime) == 0
+    vision = project / "vision.json"
+    vision.write_text(json.dumps({
+        "schema_version": "goal_vision_replan_contract_v0", "state": "vision_patch_proposed",
+        "vision_patch": {"acceptance_summary": "Verify the retained arithmetic result."},
+        "path_delta": {"outcome": "continue", "prior_assumption": "The work waits for its scheduled time.",
+            "observed_reality": "The wait resolved and the resumed work produced the verified result 391.",
+            "retained": ["Original acceptance and execution budget."],
+            "evidence_refs": ["artifact:verified-result"]},
+    }))
+    settlement = ("refresh-state", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
+        "--todo-id", TODO_ID, "--turn-instance-id", "resume-command-turn",
+        "--classification", "resumed_work_verified", "--delivery-batch-scale", "single_surface",
+        "--delivery-outcome", "outcome_progress", "--agent-vision-json", str(vision),
+        "--no-global-sync", "--suppress-external-sinks", "--settle-turn")
+    for _ in range(2):
+        rc, settled = cli(*settlement)
+        assert rc == 0 and settled["settlement_progress"]["state"] == "settled", json.dumps(settled)
+        assert settled["settlement_identity"] == guard["heartbeat_receipt"]["settlement_identity"]
+        assert _spend_run_count(runtime) == 1
+    rc, ended = cli(*reentry_args)
+    assert rc == 0 and ended["effective_action"] == "heartbeat_settled_skip", ended
+    assert artifact.read_text() == "391"