Fix generated monitor-poll commands under hard leases

docs/project-agent-todo-contract.md

@@ -159,6 +159,12 @@ requires the caller to place the fresh observation digest in
 `LOOPX_MONITOR_RESULT_HASH` and exposes separate unchanged and material-change
 commands; omitting either the Turn binding or result digest fails closed before
 monitor writeback.
+The generated command transports an existing hard-lease proof with
+`--use-current-task-lease`. It reads the canonical provider's active lease for
+the exact Monitor and agent, or reuses the same Turn's durable transaction
+proof during recovery; it never acquires or renews a lease. Missing, expired,
+or foreign leases fail before creating a provider-pending receipt. The
+canonical TypeScript transaction still checks the proof atomically.
 The canonical watch-only/ordinary-due partition is produced inside the existing
 TypeScript Todo summary and quota-planning owners after Agent scope and
 capability admission; Python compatibility code only adapts legacy facts and
@@ -172,6 +178,10 @@ replan 压力，也不会抢占 runnable advancement；二者同时存在时，
 该 CLI 路由仅在绑定当前 Turn 时可用；调用方必须把本次新鲜 observation digest
 写入 `LOOPX_MONITOR_RESULT_HASH`，并在 unchanged 与 material-change 两条命令中
 明确选择。缺少 Turn 绑定或 result digest 时，monitor writeback 会在写入前失败关闭。
+生成命令使用 `--use-current-task-lease` 传递已有的 hard-lease 证明：从
+canonical provider 读取该 Monitor 与 Agent 的有效租约；若是同一 Turn 的恢复，
+则复用持久交易回执中的原证明。该入口不会获取或续租；租约缺失、过期或归属不符
+会在形成 provider-pending 回执前失败，最终仍由 TypeScript 权威事务原子校验。
 watch-only／普通 due 的权威分区由既有 TypeScript Todo summary 与 quota-planning
 owner 在 Agent scope 和 capability admission 之后生成；Python 兼容层只适配旧事实并
 渲染已选中的 CLI／Lark 路由。

loopx/cli_commands/quota_monitor_poll.py

@@ -85,6 +85,7 @@ def record_quota_monitor_poll_for_cli(
         next_claimed_by=args.next_claimed_by,
         task_lease_idempotency_key=getattr(args, "task_lease_idempotency_key", None),
         task_lease_expected_version=getattr(args, "task_lease_expected_version", None),
+        use_current_task_lease=bool(getattr(args, "use_current_task_lease", False)),
         turn_instance_id=turn_instance_id,
         receipt_bound_todo_id=_receipt_bound_monitor_todo_id(
             args,

loopx/cli_commands/quota_request.py

@@ -94,13 +94,28 @@ def register_quota_monitor_poll_request_arguments(
         help="Current Monitor execution key for canonical quota monitor-poll; requires --task-lease-expected-version.")
     quota_parser.add_argument("--task-lease-expected-version", type=int,
         help="Current Monitor lease version, checked atomically with observation and successors; never renews the lease.")
+    quota_parser.add_argument("--use-current-task-lease", action="store_true",
+        help=("For an executing, turn-scoped monitor-poll with --todo-id, resolve the canonical "
+              "lease proof or replay the exact prior transaction proof. Does not acquire or renew a lease."))
     quota_parser.add_argument("--next-claimed-by", help="Registered agent id to claim the `--next-agent-todo` follow-up.")
 
 
 def validate_quota_command_request(args: argparse.Namespace) -> None:
     command = args.quota_command
     lease_key = getattr(args, "task_lease_idempotency_key", None)
     lease_version = getattr(args, "task_lease_expected_version", None)
+    use_current_lease = bool(getattr(args, "use_current_task_lease", False))
+    if use_current_lease:
+        if command != "monitor-poll" or not args.execute:
+            raise QuotaCommandValidationError("--use-current-task-lease requires executing quota monitor-poll")
+        if not args.todo_id or not args.turn_instance_id or not args.agent_id:
+            raise QuotaCommandValidationError(
+                "--use-current-task-lease requires --todo-id, --turn-instance-id, and --agent-id"
+            )
+        if lease_key is not None or lease_version is not None:
+            raise QuotaCommandValidationError(
+                "--use-current-task-lease cannot be combined with explicit task lease proof"
+            )
     if lease_key is not None or lease_version is not None:
         if command != "monitor-poll":
             raise QuotaCommandValidationError("task lease proof is only valid with quota monitor-poll")

loopx/control_plane/quota/monitor_poll.py

@@ -697,6 +697,7 @@ def record_quota_monitor_poll_for_decision(
     next_claimed_by: str | None = None,
     task_lease_idempotency_key: str | None = None,
     task_lease_expected_version: int | None = None,
+    use_current_task_lease: bool = False,
     turn_instance_id: str | None = None,
     _index_lock_held: bool = False,
     status_reloader: Callable[[], dict[str, Any]] | None = None,
@@ -720,7 +721,19 @@ def record_quota_monitor_poll_for_decision(
         todo_id=safe_todo_id,
         target_key=safe_target_key,
     )
-    if execute and (safe_todo_id or safe_target_key):
+    if use_current_task_lease:
+        from .monitor_poll_lease_transport import current_monitor_lease_proof
+
+        if not safe_todo_id or not normalized_turn_id or not decision_agent_id:
+            raise ValueError("current task lease transport requires exact Turn, Todo, and agent identity")
+        task_lease_idempotency_key, task_lease_expected_version = current_monitor_lease_proof(
+            runtime_root=runtime_root,
+            goal_id=goal_id,
+            todo_id=safe_todo_id,
+            agent_id=decision_agent_id,
+            effect_id=effect_id,
+        )
+    if execute and (safe_todo_id or safe_target_key) and not use_current_task_lease:
         from ..scheduler.provider_monitor_poll import (
             require_monitor_poll_source_available,
         )

loopx/control_plane/quota/monitor_poll_lease_transport.py

@@ -0,0 +1,107 @@
+"""Read-only CLI transport for an existing canonical Monitor execution proof.
+
+The coordination writer remains the authority for lease admission and its
+version CAS. A prior quota transaction receipt is preferred so a retry keeps
+the same immutable observation fingerprint after its lease is released.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import Mapping
+from pathlib import Path
+
+from ..coordination.local_authority import read_canonical_todos_if_promoted
+from ..work_items.task_lease import lease_is_active
+from .error_codes import QuotaCommandValidationError
+
+
+def _proof(value: object) -> tuple[str, int] | None:
+    if not isinstance(value, Mapping):
+        return None
+    key = value.get("idempotency_key")
+    version = value.get("expected_version")
+    if (
+        not isinstance(key, str)
+        or not key
+        or key != key.strip()
+        or not isinstance(version, int)
+        or isinstance(version, bool)
+        or not 1 <= version <= 9007199254740991
+    ):
+        return None
+    return key, version
+
+
+def _receipt_proof(
+    *, runtime_root: Path, goal_id: str, effect_id: str
+) -> tuple[str, int] | None:
+    transaction = (
+        runtime_root / "goals" / goal_id / "runs" / ".transactions"
+        / "quota-monitor-poll" / f"{hashlib.sha256(effect_id.encode()).hexdigest()[:24]}.json"
+    )
+    try:
+        receipt = json.loads(transaction.read_text(encoding="utf-8"))
+    except FileNotFoundError:
+        return None
+    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
+        raise QuotaCommandValidationError(
+            "prior monitor-poll transaction receipt cannot be read for lease-proof replay"
+        ) from exc
+    if not isinstance(receipt, Mapping) or receipt.get("effect_id") != effect_id:
+        raise QuotaCommandValidationError(
+            "prior monitor-poll transaction receipt has an invalid effect identity"
+        )
+    status = receipt.get("status")
+    if status == "provider_pending":
+        plan = receipt.get("provider_plan")
+        value = plan.get("lease_proof") if isinstance(plan, Mapping) else None
+    elif status in {"prepared", "committed"}:
+        record = receipt.get("record")
+        event = record.get("monitor_event") if isinstance(record, Mapping) else None
+        writeback = event.get("todo_writeback") if isinstance(event, Mapping) else None
+        value = writeback.get("lease_proof") if isinstance(writeback, Mapping) else None
+    else:
+        value = None
+    proof = _proof(value)
+    if proof is None:
+        raise QuotaCommandValidationError(
+            "prior monitor-poll transaction receipt lacks a valid lease proof; reconcile this effect before retrying"
+        )
+    return proof
+
+
+def current_monitor_lease_proof(
+    *, runtime_root: Path, goal_id: str, todo_id: str, agent_id: str, effect_id: str
+) -> tuple[str | None, int | None]:
+    """Resolve an existing proof; never claim, renew, or relax a lease."""
+    snapshot = read_canonical_todos_if_promoted(
+        runtime_root=runtime_root, goal_id=goal_id, include_leases=True,
+    )
+    prior = _receipt_proof(runtime_root=runtime_root, goal_id=goal_id, effect_id=effect_id)
+    if prior is not None:
+        if snapshot is None:
+            raise QuotaCommandValidationError(
+                "prior monitor-poll lease receipt cannot replay without promoted canonical authority"
+            )
+        return prior
+    if snapshot is None:
+        return None, None
+    leases = [lease for lease in snapshot["leases"] if lease.get("todo_id") == todo_id]
+    if snapshot.get("handoff_mode") != "hard_lease" and not leases:
+        return None, None
+    active = [lease for lease in leases if lease_is_active(lease)]
+    if len(active) != 1 or active[0].get("owner") != agent_id:
+        raise QuotaCommandValidationError(
+            "monitor-poll requires a current active task lease owned by --agent-id; "
+            "acquire or renew that exact Monitor lease before observing"
+        )
+    lease = active[0]
+    proof = _proof({
+        "idempotency_key": lease.get("idempotency_key"),
+        "expected_version": lease.get("version"),
+    })
+    if proof is None:
+        raise QuotaCommandValidationError("current active task lease has no valid execution proof")
+    return proof

loopx/control_plane/work_items/interaction_contract.py

@@ -1343,6 +1343,11 @@ def _build_interaction_cli_channel(
                         "unchanged_command_key": "command",
                         "changed_command_key": "material_change_command",
                     },
+                    "task_lease_proof": {
+                        "required_when": "canonical_hard_lease",
+                        "source": "canonical_lease_or_same_turn_receipt",
+                        "acquires_or_renews_lease": False,
+                    },
                 },
             }
             if _auxiliary_monitor_receipt_binding_required(payload):
@@ -1372,7 +1377,7 @@ def _build_interaction_cli_channel(
                     f"{_scoped_cli_args(agent_identity, available_capabilities=available_capabilities)}"
                     f"{auxiliary_scheduler_args} --turn-instance-id "
                     f"{shlex.quote(safe_turn_instance_id)} --todo-id "
-                    f"{shlex.quote(selected_monitor_id)} --result-hash "
+                    f"{shlex.quote(selected_monitor_id)} --use-current-task-lease --result-hash "
                     f'"${{{AUXILIARY_MONITOR_RESULT_HASH_ENV}:?}}"'
                 )
                 auxiliary_projection.update(

loopx/quota.py

@@ -1077,6 +1077,7 @@ def record_quota_monitor_poll(
     next_claimed_by: str | None = None,
     task_lease_idempotency_key: str | None = None,
     task_lease_expected_version: int | None = None,
+    use_current_task_lease: bool = False,
     turn_instance_id: str | None = None,
     receipt_bound_todo_id: str | None = None,
     scheduler_execution_context: Mapping[str, Any]
@@ -1246,6 +1247,7 @@ def should_run(current_status: dict[str, Any]) -> dict[str, Any]:
         next_claimed_by=next_claimed_by,
         task_lease_idempotency_key=task_lease_idempotency_key,
         task_lease_expected_version=task_lease_expected_version,
+        use_current_task_lease=use_current_task_lease,
         turn_instance_id=turn_instance_id,
         status_reloader=status_reloader,
     )

tests/control_plane/test_auxiliary_monitor_poll_availability.py

@@ -87,3 +87,5 @@ def test_turn_with_a_settlement_binding_still_offers_the_poll_command() -> None:
     assert projection["turn_instance_id"] == "turn-auxiliary-monitor-availability"
     assert "quota monitor-poll" in str(projection["command"])
     assert DUE_MONITOR_TODO_ID in str(projection["command"])
+    assert "--use-current-task-lease" in str(projection["command"])
+    assert projection["input_contract"]["task_lease_proof"]["source"] == "canonical_lease_or_same_turn_receipt"