fix(quota): include hard-lease execution preparation in settlement recipes

docs/reference/handoff-mode.md

@@ -62,6 +62,15 @@ A read-only quota guard never claims or renews. Long-running delegated Hosts ret
 their existing supervision; this CLI path does not supervise an idle external
 process. The App's manual completion boundary described above is unchanged.
 
+For a selected `hard_lease` Todo, the quota settlement plan now gives the ordinary
+claim command before task execution and carries its returned reference in the
+completion template. Quota and TurnEnvelope share that plan. The policy comes
+from the canonical Todo snapshot through `goal_boundary.handoff_mode`, including
+reads that omit lease records; display Markdown does not supply it. `soft_claim` retains its existing completion
+recipe. Supplied explicit proof is preserved, and an unavailable policy is not
+inferred as a new lease requirement. This is caller guidance, not acquisition,
+renewal or a replacement for validation and the existing fences.
+
 Goal-mode MCP uses the same task reference: retain `claim_operation_id` from
 `claim_task` and pass it to `complete_task`, including retries. This is a breaking
 tool-schema change: `complete_task` no longer accepts
@@ -375,6 +384,13 @@ CLI 由明确的 `--agent-id` 认领给自己时，会在同一 canonical 事务
 重复写入或续租。只读 quota guard 不做认领或续租；长期委派 Host 仍走已有监督，
 本 CLI 路径不监督空闲外部进程，也不改变 App 手工完成的现有边界。
 
+选中的 Todo 使用 `hard_lease` 时，quota 结算计划会在执行前给出普通认领命令，完成
+模板携带其返回的原引用；quota 与 TurnEnvelope 共用该计划。策略取自 canonical Todo
+快照，经 `goal_boundary.handoff_mode` 传递；即使读取未请求租约记录，也不从展示
+Markdown 推断。`soft_claim` 保持原完成命令；调用方已有显式证明时原样保留，策略
+不可用时不臆造租约要求。这是调用指引，
+不执行认领或续租，也不替代验收及已有 fence。
+
 Goal-mode MCP 同样传递任务引用：保存 `claim_task` 返回的 `claim_operation_id`，
 在 `complete_task` 及其重试中原样使用。本次为 breaking change，`complete_task`
 移除 `task_lease_idempotency_key` 和 `task_lease_expected_version` 参数；调用方须

loopx/cli_commands/todo.py

@@ -144,6 +144,7 @@ def _todo_settlement_plan(
     path_args = ""
     for name, option in (
         ("project", "--project"), ("state_file", "--state-file"),
+        ("claim_operation_id", "--claim-operation-id"),
         ("task_lease_idempotency_key", "--task-lease-idempotency-key"),
         ("task_lease_expected_version", "--task-lease-expected-version"),
     ):

loopx/control_plane/coordination/canonical_snapshot_page.ts

@@ -109,7 +109,7 @@ export async function readCanonicalSnapshotFromStore(value: unknown, store: Auth
   const metadata = {todo_read_model: readModel,
     ...(!!configuredGoalChecks(acceptance) || acceptance.outcome_requirements || acceptance.goal_objective ? {goal_acceptance_contract: acceptance} : {}),
     ...(oracleObservationBasis === null ? {} : {goal_acceptance_oracle_observation_basis: oracleObservationBasis}),
-    ...(leases === null ? {} : {handoff_mode: head.head.handoff_mode ?? "legacy"}),
+    handoff_mode: head.head.handoff_mode ?? "legacy",
     ...(readback === null ? {} : {projection_readback: confirmProjectionReadback(readback, head.provider_revision)})};
   function result(chosenTodos: readonly string[], chosenLeases: readonly string[], guards: JsonObject): JsonObject {
     const todoEnd = todoStart + chosenTodos.length, leaseEnd = leaseStart + chosenLeases.length;

loopx/control_plane/quota/effect_program.py

@@ -110,6 +110,7 @@ def build_codex_app_settlement_plan(
     quota_spend_source: str = "heartbeat",
     goal_ref: Mapping[str, object] | None = None,
     autonomous_replan: bool = False,
+    handoff_mode: str | None = None,
 ) -> SettlementPlan:
     return build_turn_scoped_cli_settlement_plan(
         goal_id=goal_id,
@@ -125,6 +126,7 @@ def build_codex_app_settlement_plan(
         quota_spend_source=quota_spend_source,
         goal_ref=goal_ref,
         autonomous_replan=autonomous_replan,
+        handoff_mode=handoff_mode,
     )
 
 
@@ -144,6 +146,7 @@ def build_turn_scoped_cli_settlement_plan(
     quota_spend_source: str = "heartbeat",
     goal_ref: Mapping[str, object] | None = None,
     autonomous_replan: bool = False,
+    handoff_mode: str | None = None,
 ) -> SettlementPlan:
     if sum(bool(value) for value in (todo_id, replan_obligation_id, request_subject)) != 1:
         raise ValueError(
@@ -186,6 +189,17 @@ def build_turn_scoped_cli_settlement_plan(
     # Returned settlement commands are machine-readable without caller surgery.
     # Keep this global option before every subcommand, including quota spend-slot.
     cli_prefix = f"{command_prefix.strip() or 'loopx'} --format json"
+    task_claim = (
+        f"{cli_prefix} todo claim --goal-id {shlex.quote(goal_id)}{binding_arg}"
+        f" --agent-id {shlex.quote(agent_id)} --claimed-by {shlex.quote(agent_id)}"
+        f"{goal_ref_arg}"
+    )
+    # Preserve a caller's accepted execution reference or explicit lease proof.
+    # This only renders CLI syntax; the canonical lifecycle still checks it.
+    proof_options = {"--claim-operation-id", "--task-lease-idempotency-key", "--task-lease-expected-version"}
+    completion_proof_supplied = any(
+        token.partition("=")[0] in proof_options for token in shlex.split(lifecycle_actor_args)
+    )
     ordinary_completion = (
         f"{cli_prefix} todo complete --goal-id {shlex.quote(goal_id)}{binding_arg}"
         f"{lifecycle_actor_args}{turn_arg}{goal_ref_arg} --evidence '<validated evidence>'"
@@ -205,9 +219,12 @@ def build_turn_scoped_cli_settlement_plan(
     payload = effect_runtime_result("settlement.turn_scoped_cli_plan", {
         "identity": identity.as_dict(), "delivery_boundary": delivery_boundary,
         "autonomous_replan": autonomous_replan,
+        "handoff_mode": handoff_mode,
+        "completion_proof_supplied": completion_proof_supplied,
         "delivery_batch_scale_choices": list(DELIVERY_BATCH_SCALE_INPUT_CHOICES),
         **({"goal_ref": normalized_goal_ref} if normalized_goal_ref is not None else {}),
         "command_templates": {
+            **({"todo_claim": task_claim} if todo_id else {}),
             "todo_completion": ordinary_completion, "durable_writeback": writeback,
             "quota_spend": spend, "terminal_closeout": terminal_closeout,
         },

loopx/control_plane/quota/settlement_plan.ts

@@ -6,10 +6,11 @@ import {
 } from "../effect_program.ts";
 import {VISION_MATERIAL_CLOSEOUT_HINT, visionAuthoringContract} from "../goals/vision_checkpoint.ts";
 import {parseExactGoalRef} from "../goals/goal_instance_identity.ts";
-import {requireBoolean, requireJsonObject, requireNonEmptyString, requireStringArray} from "../runtime_decode.ts";
+import {requireBoolean, requireJsonObject, requireNonEmptyString, requireStringArray, requireStringLiteral} from "../runtime_decode.ts";
 import {MATERIAL_DELIVERY_OUTCOMES} from "../work_items/delivery_outcome.ts";
 import {DELIVERY_BOUNDARIES} from "../turn_driver/delivery_continuity.ts";
 import {PROGRESS_RESULT_CLASSES} from "../work_items/replan_history.ts";
+import {HANDOFF_MODES} from "../coordination/handoff_mode_vocabulary.ts";
 
 export function turnScopedCliSettlementPlan(params: JsonObject): SettlementPlan {
   const identity = settlementIdentity(
@@ -54,15 +55,28 @@ export function turnScopedCliSettlementPlan(params: JsonObject): SettlementPlan
   const inFlight = params.delivery_boundary === "in_flight_continuation";
   const autonomousReplan = (params.autonomous_replan === undefined ? false
     : requireBoolean(params.autonomous_replan, "autonomous_replan")) || identity.binding_kind === "autonomous_replan";
+  const mode = params.handoff_mode == null ? null
+    : requireStringLiteral(params.handoff_mode, HANDOFF_MODES, "handoff_mode");
+  const managedClaim = identity.binding_kind === "todo" && mode === "hard_lease";
+  const proofSupplied = params.completion_proof_supplied === undefined ? false
+    : requireBoolean(params.completion_proof_supplied, "completion_proof_supplied");
+  const completionProof = managedClaim && !proofSupplied
+    ? " --claim-operation-id '<original claim_operation_id>'" : "";
   const validation: SettlementStep = {
     kind: "validation", owner: "agent", idempotency_key_ref: "$.identity.effect_id",
     expected_receipt: "validation_receipt",
     precondition: "delivery result is independently validated",
   };
   if (identity.binding_kind === "todo") {
-    validation.command_template = requireNonEmptyString(commands.todo_completion, "todo_completion");
+    validation.command_template = requireNonEmptyString(commands.todo_completion, "todo_completion") + completionProof;
     validation.command_condition = "todo_deliverable_complete";
-    validation.precondition = "validate the original Todo deliverable; when complete, run ordinary Todo completion " +
+    const claimHint = managedClaim ? "Before task execution, prepare it with ordinary Todo claim: `" +
+      requireNonEmptyString(commands.todo_claim, "todo_claim") +
+      "`. Reuse an already accepted current claim or explicit lease proof; keep the original " +
+      "claim_operation_id for managed progress/completion retries. claimed_by alone is not an execution lease. " +
+      "An expired, released or replaced execution requires normal ownership recovery; " +
+      "this plan neither acquires nor renews execution. " : "";
+    validation.precondition = claimHint + "validate the original Todo deliverable; when complete, run ordinary Todo completion " +
       "with its original declaration and current lease before writeback/spend. Use --evidence for " +
       "an artifact pointer; --result-file requires approved Goal acceptance criteria bound to this Todo, " +
       "not merely a Todo validator. No invented successor " +
@@ -108,7 +122,7 @@ export function turnScopedCliSettlementPlan(params: JsonObject): SettlementPlan
     precondition: "the selected Todo is final with no runnable successor and matching " +
       "writeback and quota spend receipts exist",
     idempotency_key_ref: "$.identity.effect_id", expected_receipt: "terminal_closeout_receipt",
-    command_template: requireNonEmptyString(commands.terminal_closeout, "terminal_closeout"),
+    command_template: requireNonEmptyString(commands.terminal_closeout, "terminal_closeout") + completionProof,
     conditional: true,
   });
   return {

loopx/control_plane/quota/should_run_packet.py

@@ -1364,7 +1364,10 @@ def _quota_payload_context(prepared: _QuotaDecisionPreparation, route: _QuotaDec
             if isinstance(prepared.item.get("long_task_cadence_hint"), dict)
             else None,
         "handoff_readiness": prepared.item.get("handoff_readiness"),
-        "goal_boundary": prepared.goal_boundary,
+        "goal_boundary": (
+            {**(prepared.goal_boundary or {}), "handoff_mode": prepared.item["handoff_mode"]}
+            if prepared.item.get("handoff_mode") is not None else prepared.goal_boundary
+        ),
         "plan_summary": prepared.plan.get("summary")
     }
 

loopx/control_plane/todos/active_state_todos.py

@@ -111,6 +111,9 @@ def active_state_todo_fields(
     from .goal_todo_projection import observe_todo_acceptance
     fields = observe_todo_acceptance(canonical["_todo_fields"], canonical,
         registry_path=source_registry, goal_id=goal_id)
+    # Execution guidance must use the same canonical policy as the Todo read,
+    # never the historical Markdown frontmatter or a registry default.
+    fields["handoff_mode"] = canonical.get("handoff_mode")
     # Material existence is a local read projection, not persisted Todo state.
     # Derive it from the canonical text so stale display links cannot take over.
     for key in ("user_todos", "agent_todos"):

loopx/control_plane/work_items/accountable_settlement.py

@@ -29,6 +29,7 @@ def build_accountable_work_item_settlement_plan(
     command_prefix: str = "loopx",
     goal_ref: Mapping[str, object] | None = None,
     autonomous_replan: bool = False,
+    handoff_mode: str | None = None,
 ) -> SettlementPlan | None:
     if runtime_profile in APP_HEARTBEAT_SETTLEMENT_RUNTIME_PROFILES:
         normalized_turn_instance_id = normalize_turn_instance_id(turn_instance_id)
@@ -44,6 +45,7 @@ def build_accountable_work_item_settlement_plan(
             delivery_boundary=delivery_boundary,
             goal_ref=goal_ref,
             autonomous_replan=autonomous_replan,
+            handoff_mode=handoff_mode,
         )
     if runtime_profile in VISIBLE_GOAL_SETTLEMENT_RUNTIME_PROFILES:
         normalized_turn_instance_id = normalize_turn_instance_id(turn_instance_id)
@@ -62,6 +64,7 @@ def build_accountable_work_item_settlement_plan(
             quota_spend_source="visible-goal",
             goal_ref=goal_ref,
             autonomous_replan=autonomous_replan,
+            handoff_mode=handoff_mode,
         )
     if runtime_profile is not SchedulerRuntimeProfile.GENERIC_CLI_AGENT_LOOP:
         return None
@@ -80,4 +83,5 @@ def build_accountable_work_item_settlement_plan(
         delivery_boundary=delivery_boundary,
         goal_ref=goal_ref,
         autonomous_replan=autonomous_replan,
+        handoff_mode=handoff_mode,
     )

loopx/control_plane/work_items/interaction_contract.py

@@ -475,6 +475,7 @@ def _turn_scoped_cli_settlement_context(
         replan_obligation_id=replan_obligation_id,
         scoped_cli_args=scoped_cli_args,
         lifecycle_actor_args=f" --agent-id {shlex.quote(agent_id)}",
+        handoff_mode=(payload.get("goal_boundary") or {}).get("handoff_mode"),
         turn_instance_id=turn_instance_id,
         autonomous_replan=semantic_replan_obligation_id is not None,
         delivery_boundary=(

loopx/quota.py

@@ -925,6 +925,7 @@ def build_quota_plan(
             "missing_gates",
             "next_handoff_condition",
             "handoff_readiness",
+            "handoff_mode",
             "user_todos",
             "agent_todos",
             "active_state_next_action",

tests/control_plane/test_quota_settlement.py

@@ -112,6 +112,23 @@ def test_command_plan_retains_one_matching_actor(arguments):
     assert sum(arg == "--agent-id" or arg.startswith("--agent-id=") for arg in argv) == 1
 
 
+@pytest.mark.parametrize("proof", [
+    "--claim-operation-id original-claim",
+    "--claim-operation-id=original-claim",
+    "--task-lease-idempotency-key original-lease --task-lease-expected-version 3",
+])
+def test_hard_lease_plan_preserves_supplied_proof_without_a_second_reference(proof):
+    plan = quota_effect_program.build_turn_scoped_cli_settlement_plan(
+        goal_id=GOAL_ID, agent_id=AGENT_ID, todo_id=TODO_ID,
+        turn_instance_id=TURN_ID, scoped_cli_args="", lifecycle_actor_args=proof,
+        handoff_mode="hard_lease",
+    ).as_dict()
+    for kind in (SettlementStepKind.VALIDATION, SettlementStepKind.TERMINAL_CLOSEOUT):
+        command = settlement_step_command(plan, kind)
+        assert proof in command
+        assert "<original claim_operation_id>" not in command
+
+
 def _receipt(step: SettlementStepKind, marker: str) -> SettlementReceipt:
     return SettlementReceipt(
         step_kind=step,

tests/control_plane/test_settlement_command_guidance_cli.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import shlex
+import re
 from pathlib import Path
 
 import pytest
@@ -13,6 +14,51 @@
 )
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("profile", ["codex_app", "generic_cli"])
+@pytest.mark.parametrize("handoff_mode", ["soft_claim", "hard_lease"])
+def test_returned_completion_prepares_and_preserves_original_execution(
+    tmp_path: Path, provider: str, profile: str, handoff_mode: str,
+) -> None:
+    project, runtime, registry = _write_fixture(tmp_path)
+    _initialize_fixture_authority(project, runtime, provider=provider, handoff_mode=handoff_mode)
+    host = ["--codex-app"] if profile == "codex_app" else ["--runtime-profile", profile]
+    rc, guard = _run_cli(registry, runtime, "quota", "should-run", *host,
+                        "--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
+                        "--todo-id", TODO_ID, "--turn-instance-id", "claim-guidance")
+    assert rc == 0, guard
+    assert guard["goal_boundary"]["handoff_mode"] == handoff_mode
+    plan = guard["interaction_contract"]["cli_channel"]["settlement_plan"]
+    assert build_turn_envelope(guard)["writeback"]["settlement_plan"] == plan
+    step = next(s for s in plan["ordered_steps"] if s["kind"] == "validation")
+    preparation = re.search(r"ordinary Todo claim: `([^`]+)`", step["precondition"])
+    command = step["command_template"].replace("<validated evidence>", "evidence:accepted-deliverable")
+    if handoff_mode == "hard_lease":
+        assert preparation is not None
+        # Follow only the returned plan, without a failed completion or global help.
+        rc, claim = _run_generated_cli(preparation[1], registry_path=registry)
+        assert rc == 0 and claim["execution_managed"] is True, claim
+        command = command.replace("<original claim_operation_id>", claim["claim_operation_id"])
+        # A guessed reference must not adopt the active execution or complete work.
+        rc, refused = _run_generated_cli(
+            command.replace(claim["claim_operation_id"], "claim-not-issued"), registry_path=registry,
+        )
+        assert rc != 0 and refused["ok"] is False, refused
+        rc, todo = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID,
+                            "--todo-id", TODO_ID)
+        assert rc == 0 and todo["todo"]["status"] == "open", todo
+    else:
+        assert preparation is None
+        assert "--claim-operation-id" not in command
+    rc, completed = _run_generated_cli(command, registry_path=registry)
+    assert rc == 0, completed
+    rc, todo = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID,
+                        "--todo-id", TODO_ID)
+    assert rc == 0 and todo["todo"]["status"] == "done", todo
+    rc, replay = _run_generated_cli(command, registry_path=registry)
+    assert rc == 0 and replay["changed"] is False, replay
+
+
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 @pytest.mark.parametrize("profile", ["codex_app", "generic_cli"])
 def test_returned_writeback_can_record_progress_and_settle_once(

tests/control_plane_ts/canonical_snapshot_conformance.ts

@@ -82,10 +82,13 @@ export function registerCanonicalSnapshotConformance(name: string, factory: Auth
       assert.deepEqual(todoOnly.todos, result.todos);
       assert.deepEqual(todoOnly.leases, []);
       assert.ok(todoOnly.pages.every(page => !("leases" in page)));
+      assert.ok(todoOnly.pages.every(page =>
+        (page.metadata as JsonObject).handoff_mode === fixture.projection.handoff_mode));
       const completeTodoOnly = await readCanonicalSnapshotFromStore(snapshotRequest({include_leases: false}),
         store, CANONICAL_SNAPSHOT_LOCAL_BYTES);
       assert.deepEqual(completeTodoOnly.todos, result.todos);
       assert.equal("leases" in completeTodoOnly, false);
+      assert.equal((completeTodoOnly.metadata as JsonObject).handoff_mode, fixture.projection.handoff_mode);
     });
   }