From dcbf1cf32b19964473527c7be75587106c32557e Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Fri, 25 Sep 2026 04:33:28 +0800 Subject: [PATCH] fix(quota): replay unleased monitor poll receipts Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../quota/monitor_poll_lease_transport.py | 40 +++++++++++++------ .../control_plane/test_leased_monitor_poll.py | 31 ++++++++++++-- 2 files changed, 56 insertions(+), 15 deletions(-) diff --git a/loopx/control_plane/quota/monitor_poll_lease_transport.py b/loopx/control_plane/quota/monitor_poll_lease_transport.py index fab2e479b5..be652d5e31 100644 --- a/loopx/control_plane/quota/monitor_poll_lease_transport.py +++ b/loopx/control_plane/quota/monitor_poll_lease_transport.py @@ -36,7 +36,7 @@ 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" @@ -44,7 +44,7 @@ def _receipt_proof( 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] diff --git a/tests/control_plane/test_leased_monitor_poll.py b/tests/control_plane/test_leased_monitor_poll.py index b3e89e2735..ad9ce3bcf1 100644 --- a/tests/control_plane/test_leased_monitor_poll.py +++ b/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):