From 9ec1a1f7440b580e52b7a68d5536d0cb86683c3f Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Thu, 24 Sep 2026 18:18:13 +0800 Subject: [PATCH 1/4] fix: transport monitor lease proof through generated CLI route Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- loopx/cli_commands/quota_monitor_poll.py | 1 + loopx/cli_commands/quota_request.py | 15 +++ loopx/control_plane/quota/monitor_poll.py | 15 ++- .../quota/monitor_poll_lease_transport.py | 107 ++++++++++++++++++ .../work_items/interaction_contract.py | 7 +- loopx/quota.py | 2 + 6 files changed, 145 insertions(+), 2 deletions(-) create mode 100644 loopx/control_plane/quota/monitor_poll_lease_transport.py diff --git a/loopx/cli_commands/quota_monitor_poll.py b/loopx/cli_commands/quota_monitor_poll.py index 2e46f8a371..838776bd5c 100644 --- a/loopx/cli_commands/quota_monitor_poll.py +++ b/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, diff --git a/loopx/cli_commands/quota_request.py b/loopx/cli_commands/quota_request.py index 99fe35f307..59259bc935 100644 --- a/loopx/cli_commands/quota_request.py +++ b/loopx/cli_commands/quota_request.py @@ -94,6 +94,9 @@ 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.") @@ -101,6 +104,18 @@ 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") diff --git a/loopx/control_plane/quota/monitor_poll.py b/loopx/control_plane/quota/monitor_poll.py index 4ac85ba6f4..e98c87073c 100644 --- a/loopx/control_plane/quota/monitor_poll.py +++ b/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, ) diff --git a/loopx/control_plane/quota/monitor_poll_lease_transport.py b/loopx/control_plane/quota/monitor_poll_lease_transport.py new file mode 100644 index 0000000000..fab2e479b5 --- /dev/null +++ b/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 diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py index 1cb2b6aee7..6f2d0ac226 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/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( diff --git a/loopx/quota.py b/loopx/quota.py index 014e42f543..db1e8e02fb 100644 --- a/loopx/quota.py +++ b/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, ) From 3a82f885030ba4801de3402d5d6d0d9987bae822 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Thu, 24 Sep 2026 18:18:24 +0800 Subject: [PATCH 2/4] test: cover leased monitor transport and crash replay Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- ...est_auxiliary_monitor_poll_availability.py | 2 + .../control_plane/test_leased_monitor_poll.py | 86 ++++++++++++++++++- .../test_work_lane_contract_core.py | 6 ++ 3 files changed, 92 insertions(+), 2 deletions(-) diff --git a/tests/control_plane/test_auxiliary_monitor_poll_availability.py b/tests/control_plane/test_auxiliary_monitor_poll_availability.py index c212ccb312..e7eeb214c1 100644 --- a/tests/control_plane/test_auxiliary_monitor_poll_availability.py +++ b/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" diff --git a/tests/control_plane/test_leased_monitor_poll.py b/tests/control_plane/test_leased_monitor_poll.py index f54a361e6e..7c0f9e7300 100644 --- a/tests/control_plane/test_leased_monitor_poll.py +++ b/tests/control_plane/test_leased_monitor_poll.py @@ -59,6 +59,13 @@ def arguments(monitor): "--task-lease-expected-version", str(PROOF["expected_version"]), "--execute"] +def automatic_arguments(monitor, *, turn_id="leased-monitor-automatic"): + explicit = arguments(monitor) + key_index = explicit.index("--task-lease-idempotency-key") + del explicit[key_index:key_index + 4] + return [*explicit, "--turn-instance-id", turn_id, "--use-current-task-lease"] + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("native", [False, True]) def test_leased_monitor_public_cli_settles_once_without_display(tmp_path, monkeypatch, provider, native): @@ -80,6 +87,77 @@ def test_leased_monitor_public_cli_settles_once_without_display(tmp_path, monkey assert all(row.get("classification") != "quota_slot_spend" for row in records) +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_current_lease_cli_transport_replays_after_release(tmp_path, monkeypatch, provider): + isolate_sqlite_runtime(tmp_path, monkeypatch) + registry, runtime, _state, monitor = _canonical(tmp_path, provider=provider, lease=LEASE) + args = automatic_arguments(monitor) + guard = run_json_cli("quota", "should-run", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--runtime-profile", "generic_cli", "--turn-instance-id", "leased-monitor-automatic", + "--available-capability", "network", "--available-capability", "external_evidence_poll", + registry_path=registry, runtime_root=runtime) + assert guard["selected_todo"]["todo_id"] == monitor["todo_id"] + args.extend(["--available-capability", "network", "--available-capability", "external_evidence_poll"]) + first = run_json_cli(*args, registry_path=registry, runtime_root=runtime) + assert first["ok"] is True + assert first["todo_writeback"]["lease_proof"] == PROOF + run_json_cli("task-lease", "release", "--goal-id", GOAL_ID, "--todo-id", monitor["todo_id"], + "--owner", AGENT_ID, "--idempotency-key", PROOF["idempotency_key"], "--expected-version", "3", + registry_path=registry, runtime_root=runtime) + replay = run_json_cli(*args, registry_path=registry, runtime_root=runtime) + assert replay["replayed"] is True + records = [json.loads(line) for line in (runtime / "goals" / GOAL_ID / "runs" / "index.jsonl").read_text().splitlines()] + assert sum(row.get("classification") == "quota_monitor_poll" for row in records) == 1 + assert all(row.get("classification") != "quota_slot_spend" for row in records) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_current_lease_cli_transport_rejects_expired_lease_without_pending(tmp_path, monkeypatch, provider): + isolate_sqlite_runtime(tmp_path, monkeypatch) + registry, runtime, _state, monitor = _canonical(tmp_path, provider=provider, + lease={**LEASE, "expires_at": "2026-01-01T01:00:00Z"}) + guard = run_json_cli("quota", "should-run", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--runtime-profile", "generic_cli", "--turn-instance-id", "leased-monitor-automatic", + "--available-capability", "network", "--available-capability", "external_evidence_poll", + registry_path=registry, runtime_root=runtime) + assert guard["selected_todo"]["todo_id"] == monitor["todo_id"] + before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL_ID, include_leases=True) + code, failure = run_json_cli_result(*automatic_arguments(monitor), registry_path=registry, runtime_root=runtime) + assert code != 0 + assert failure["ok"] is False + assert "current active task lease" in str(failure.get("reason")) + assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL_ID, include_leases=True) == before + pending = runtime / "goals" / GOAL_ID / "runs" / ".transactions" / "quota-monitor-poll" + assert not pending.exists() or not list(pending.glob("*.json")) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_current_lease_cli_transport_keeps_soft_claim_compatible(tmp_path, monkeypatch, provider): + isolate_sqlite_runtime(tmp_path, monkeypatch) + registry, runtime, _state, monitor = _canonical(tmp_path, provider=provider) + turn_id = "soft-claim-monitor-automatic" + guard = 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 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) + assert result["ok"] is True + assert "lease_proof" not in result["todo_writeback"] + + +def test_current_lease_cli_transport_rejects_ambiguous_proof_arguments(tmp_path): + registry, runtime, _state, monitor = _canonical(tmp_path, lease=LEASE) + code, failure = run_json_cli_result(*automatic_arguments(monitor), + "--task-lease-idempotency-key", PROOF["idempotency_key"], + "--task-lease-expected-version", "3", registry_path=registry, runtime_root=runtime) + assert code != 0 + assert failure["error_code"] == "QUOTA_VALIDATION_FAILED" + assert "cannot be combined" in failure["reason"] + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_old_observation_time_cannot_revive_expired_proof(tmp_path, monkeypatch, provider): isolate_sqlite_runtime(tmp_path, monkeypatch) @@ -107,7 +185,8 @@ def test_explicit_proof_never_falls_back_to_legacy_writer(tmp_path): @pytest.mark.parametrize("provider", ["file", "sqlite"]) -def test_process_death_after_business_commit_recovers_after_lease_release(tmp_path, monkeypatch, provider): +@pytest.mark.parametrize("automatic", [False, True]) +def test_process_death_after_business_commit_recovers_after_lease_release(tmp_path, monkeypatch, provider, automatic): isolate_sqlite_runtime(tmp_path, monkeypatch) registry, runtime, _state, monitor = _canonical(tmp_path, provider=provider, lease=LEASE) turn_id = "leased-monitor-crash" @@ -116,7 +195,10 @@ def test_process_death_after_business_commit_recovers_after_lease_release(tmp_pa "--available-capability", "network", "--available-capability", "external_evidence_poll", registry_path=registry, runtime_root=runtime) assert guard["selected_todo"]["todo_id"] == monitor["todo_id"] - args = [*arguments(monitor), "--turn-instance-id", turn_id, + proof_args = automatic_arguments(monitor, turn_id=turn_id) if automatic else [ + *arguments(monitor), "--turn-instance-id", turn_id, + ] + args = [*proof_args, "--available-capability", "network", "--available-capability", "external_evidence_poll"] # Kill the actual CLI process after the authority transaction returned its # receipt, before the separate quota transaction can settle it. diff --git a/tests/control_plane/test_work_lane_contract_core.py b/tests/control_plane/test_work_lane_contract_core.py index 556c738663..d959f7a81c 100644 --- a/tests/control_plane/test_work_lane_contract_core.py +++ b/tests/control_plane/test_work_lane_contract_core.py @@ -158,6 +158,7 @@ def test_due_watch_only_monitor_is_an_auxiliary_no_spend_route() -> None: assert auxiliary_cli["spend_policy"] == "no_spend" assert f"--turn-instance-id {turn_instance_id}" in auxiliary_cli["command"] assert "--todo-id todo_watch_due" in auxiliary_cli["command"] + assert "--use-current-task-lease" in auxiliary_cli["command"] assert '--result-hash "${LOOPX_MONITOR_RESULT_HASH:?}"' in auxiliary_cli[ "command" ] @@ -177,6 +178,11 @@ def test_due_watch_only_monitor_is_an_auxiliary_no_spend_route() -> None: "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, + }, } From 0e687fc1332e181881166b1faa3f1b91094c96f0 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Thu, 24 Sep 2026 18:18:33 +0800 Subject: [PATCH 3/4] docs: explain monitor lease proof and replay contract Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- docs/project-agent-todo-contract.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/docs/project-agent-todo-contract.md b/docs/project-agent-todo-contract.md index d19d05ff37..d4c531eff0 100644 --- a/docs/project-agent-todo-contract.md +++ b/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 路由。 From 717535c9cc7b3a3f40dfe1667f6464a64e60d610 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Thu, 24 Sep 2026 18:49:31 +0800 Subject: [PATCH 4/4] test(quota): reject foreign monitor lease transport Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../control_plane/test_leased_monitor_poll.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/tests/control_plane/test_leased_monitor_poll.py b/tests/control_plane/test_leased_monitor_poll.py index 7c0f9e7300..b3e89e2735 100644 --- a/tests/control_plane/test_leased_monitor_poll.py +++ b/tests/control_plane/test_leased_monitor_poll.py @@ -11,6 +11,8 @@ from test_native_monitor_poll import _canonical from test_monitor_followthrough_contract import GOAL_ID, AGENT_ID, _write_fixture, _add_monitor from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted +from loopx.control_plane.quota.error_codes import QuotaCommandValidationError +from loopx.control_plane.quota.monitor_poll_lease_transport import current_monitor_lease_proof from loopx.control_plane.scheduler.monitor_poll_writeback import write_monitor_poll_todo_state from loopx.control_plane.testing.canary_harness import run_json_cli, run_json_cli_result @@ -131,6 +133,23 @@ def test_current_lease_cli_transport_rejects_expired_lease_without_pending(tmp_p assert not pending.exists() or not list(pending.glob("*.json")) +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_current_lease_cli_transport_rejects_foreign_owner_without_pending(tmp_path, monkeypatch, provider): + isolate_sqlite_runtime(tmp_path, monkeypatch) + registry, runtime, _state, monitor = _canonical( + tmp_path, provider=provider, lease={**LEASE, "owner": "another-agent"}, + ) + # The public command also requires a committed same-Turn receipt first; + # exercise the read-only transport directly to prove its owner fence. + with pytest.raises(QuotaCommandValidationError, match="owned by --agent-id"): + current_monitor_lease_proof( + runtime_root=runtime, goal_id=GOAL_ID, todo_id=monitor["todo_id"], + agent_id=AGENT_ID, effect_id="foreign-owner-monitor-poll", + ) + pending = runtime / "goals" / GOAL_ID / "runs" / ".transactions" / "quota-monitor-poll" + assert not pending.exists() or not list(pending.glob("*.json")) + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_current_lease_cli_transport_keeps_soft_claim_compatible(tmp_path, monkeypatch, provider): isolate_sqlite_runtime(tmp_path, monkeypatch)