fix(quota): replay unleased monitor poll receipts

loopx/control_plane/quota/monitor_poll_lease_transport.py

@@ -36,15 +36,15 @@ def _proof(value: object) -> tuple[str, int] | None:
 
 def _receipt_proof(
     *, runtime_root: Path, goal_id: str, effect_id: str
-) -> tuple[str, int] | None:
+) -> tuple[bool, 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
+        return False, None
     except (OSError, UnicodeError, json.JSONDecodeError) as exc:
         raise QuotaCommandValidationError(
             "prior monitor-poll transaction receipt cannot be read for lease-proof replay"
@@ -55,21 +55,28 @@ def _receipt_proof(
         )
     status = receipt.get("status")
     if status == "provider_pending":
-        plan = receipt.get("provider_plan")
-        value = plan.get("lease_proof") if isinstance(plan, Mapping) else None
+        source = receipt.get("provider_plan")
+        unleased_schema = "monitor_poll_todo_provider_plan_v0"
     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
+        source = event.get("todo_writeback") if isinstance(event, Mapping) else None
+        unleased_schema = "monitor_poll_todo_writeback_v0"
     else:
-        value = None
+        source = None
+        unleased_schema = ""
+    value = source.get("lease_proof") if isinstance(source, Mapping) else None
+    if value is None and isinstance(source, Mapping) and source.get("schema_version") == unleased_schema:
+        # A committed legacy or soft-claim observation has no lease to replay.
+        # Keep it distinct from a missing receipt so the caller can still
+        # reject a later hard-lease authority transition.
+        return True, 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
+    return True, proof
 
 
 def current_monitor_lease_proof(
@@ -79,13 +86,22 @@ def current_monitor_lease_proof(
     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:
+    has_prior, prior = _receipt_proof(
+        runtime_root=runtime_root, goal_id=goal_id, effect_id=effect_id
+    )
+    if has_prior:
+        if prior is not None and snapshot is None:
             raise QuotaCommandValidationError(
                 "prior monitor-poll lease receipt cannot replay without promoted canonical authority"
             )
-        return prior
+        if prior is None and snapshot is not None and (
+            snapshot.get("handoff_mode") == "hard_lease"
+            or any(lease.get("todo_id") == todo_id for lease in snapshot["leases"])
+        ):
+            raise QuotaCommandValidationError(
+                "prior monitor-poll transaction receipt has no lease proof for current canonical authority"
+            )
+        return prior if prior is not None else (None, None)
     if snapshot is None:
         return None, None
     leases = [lease for lease in snapshot["leases"] if lease.get("todo_id") == todo_id]

tests/control_plane/test_leased_monitor_poll.py

@@ -160,11 +160,36 @@ def test_current_lease_cli_transport_keeps_soft_claim_compatible(tmp_path, monke
         "--available-capability", "network", "--available-capability", "external_evidence_poll",
         registry_path=registry, runtime_root=runtime)
     assert guard["selected_todo"]["todo_id"] == monitor["todo_id"]
-    result = run_json_cli(*automatic_arguments(monitor, turn_id=turn_id),
-        "--available-capability", "network", "--available-capability", "external_evidence_poll",
-        registry_path=registry, runtime_root=runtime)
+    args = [*automatic_arguments(monitor, turn_id=turn_id),
+        "--available-capability", "network", "--available-capability", "external_evidence_poll"]
+    result = run_json_cli(*args, registry_path=registry, runtime_root=runtime)
     assert result["ok"] is True
     assert "lease_proof" not in result["todo_writeback"]
+    replay = run_json_cli(*args, registry_path=registry, runtime_root=runtime)
+    assert replay["replayed"] is True
+
+
+def test_existing_hard_lease_receipt_cannot_lose_its_proof(tmp_path):
+    registry, runtime, _state, monitor = _canonical(tmp_path, lease=LEASE)
+    turn_id = "hard-lease-proof-retained"
+    args = [*automatic_arguments(monitor, turn_id=turn_id),
+        "--available-capability", "network", "--available-capability", "external_evidence_poll"]
+    run_json_cli("quota", "should-run", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
+        "--runtime-profile", "generic_cli", "--turn-instance-id", turn_id,
+        "--available-capability", "network", "--available-capability", "external_evidence_poll",
+        registry_path=registry, runtime_root=runtime)
+    assert run_json_cli(*args, registry_path=registry, runtime_root=runtime)["ok"] is True
+    receipt_dir = runtime / "goals" / GOAL_ID / "runs" / ".transactions" / "quota-monitor-poll"
+    receipt_path, = receipt_dir.glob("*.json")
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    receipt["record"]["monitor_event"]["todo_writeback"].pop("lease_proof")
+    receipt_path.write_text(json.dumps(receipt), encoding="utf-8")
+
+    with pytest.raises(QuotaCommandValidationError, match="no lease proof"):
+        current_monitor_lease_proof(
+            runtime_root=runtime, goal_id=GOAL_ID, todo_id=monitor["todo_id"],
+            agent_id=AGENT_ID, effect_id=receipt["effect_id"],
+        )
 
 
 def test_current_lease_cli_transport_rejects_ambiguous_proof_arguments(tmp_path):