|
11 | 11 | from test_native_monitor_poll import _canonical |
12 | 12 | from test_monitor_followthrough_contract import GOAL_ID, AGENT_ID, _write_fixture, _add_monitor |
13 | 13 | from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted |
| 14 | +from loopx.control_plane.quota.error_codes import QuotaCommandValidationError |
| 15 | +from loopx.control_plane.quota.monitor_poll_lease_transport import current_monitor_lease_proof |
14 | 16 | from loopx.control_plane.scheduler.monitor_poll_writeback import write_monitor_poll_todo_state |
15 | 17 | from loopx.control_plane.testing.canary_harness import run_json_cli, run_json_cli_result |
16 | 18 |
|
@@ -100,6 +102,23 @@ def test_current_lease_cli_transport_rejects_expired_lease_without_pending(tmp_p |
100 | 102 | assert not pending.exists() or not list(pending.glob("*.json")) |
101 | 103 |
|
102 | 104 |
|
| 105 | +@pytest.mark.parametrize("provider", ["file", "sqlite"]) |
| 106 | +def test_current_lease_cli_transport_rejects_foreign_owner_without_pending(tmp_path, monkeypatch, provider): |
| 107 | + isolate_sqlite_runtime(tmp_path, monkeypatch) |
| 108 | + registry, runtime, _state, monitor = _canonical( |
| 109 | + tmp_path, provider=provider, lease={**LEASE, "owner": "another-agent"}, |
| 110 | + ) |
| 111 | + # The public command also requires a committed same-Turn receipt first; |
| 112 | + # exercise the read-only transport directly to prove its owner fence. |
| 113 | + with pytest.raises(QuotaCommandValidationError, match="owned by --agent-id"): |
| 114 | + current_monitor_lease_proof( |
| 115 | + runtime_root=runtime, goal_id=GOAL_ID, todo_id=monitor["todo_id"], |
| 116 | + agent_id=AGENT_ID, effect_id="foreign-owner-monitor-poll", |
| 117 | + ) |
| 118 | + pending = runtime / "goals" / GOAL_ID / "runs" / ".transactions" / "quota-monitor-poll" |
| 119 | + assert not pending.exists() or not list(pending.glob("*.json")) |
| 120 | + |
| 121 | + |
103 | 122 | @pytest.mark.parametrize("provider", ["file", "sqlite"]) |
104 | 123 | def test_current_lease_cli_transport_keeps_soft_claim_compatible(tmp_path, monkeypatch, provider): |
105 | 124 | isolate_sqlite_runtime(tmp_path, monkeypatch) |
|
0 commit comments