From af1540f9fec5c178905dc06d3d7028e438dfaf55 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 13 Nov 2026 13:51:55 +0800 Subject: [PATCH] Read evaluated readiness for due Monitor polls Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/quota/monitor_poll.py | 26 ++++++ .../test_settled_monitor_user_gate.py | 87 +++++++++++++++++-- 2 files changed, 108 insertions(+), 5 deletions(-) diff --git a/loopx/control_plane/quota/monitor_poll.py b/loopx/control_plane/quota/monitor_poll.py index bc4824b6d4..e4e8422b3f 100644 --- a/loopx/control_plane/quota/monitor_poll.py +++ b/loopx/control_plane/quota/monitor_poll.py @@ -21,6 +21,7 @@ TODO_TASK_CLASS_MONITOR, normalize_todo_claimed_by, normalize_todo_id, + normalize_todo_resume_when, ) from ..todos.external_wait_contract import ( build_monitor_advancement_authoring_contract, @@ -131,6 +132,31 @@ def resolve_due_monitor_candidate( ) except ValueError: return {} + resume_when = normalize_todo_resume_when(item.get("resume_when")) + if resume_when and item.get("resume_ready") is None: + from ..todos.list_readback import list_goal_todos + + # Target selection returns the stored Todo, where readiness may not be + # materialized. Read the canonical evaluation for that same target. + selected_id = normalize_todo_id(item.get("todo_id")) + if selected_id is None: + return {} + evaluated = list_goal_todos( + registry_path=registry_path, + goal_id=goal_id, + role="agent", + todo_id=selected_id, + runtime_root_arg=str(runtime_root) if runtime_root else None, + ).get("todo") + if ( + not evaluated + or evaluated.get("todo_id") != selected_id + or evaluated.get("target_key") != item.get("target_key") + or evaluated.get("archive_state") == "archive" + or normalize_todo_resume_when(evaluated.get("resume_when")) != resume_when + ): + return {} + item = evaluated if not monitor_todo_is_due(item): return {} return { diff --git a/tests/control_plane/test_settled_monitor_user_gate.py b/tests/control_plane/test_settled_monitor_user_gate.py index 7e04703e4c..96abd539d9 100644 --- a/tests/control_plane/test_settled_monitor_user_gate.py +++ b/tests/control_plane/test_settled_monitor_user_gate.py @@ -105,17 +105,24 @@ def call(*args): assert not any(row["classification"] in {"state_refreshed", "quota_slot_spent"} for row in rows) -def _primary_monitor_fixture(tmp_path, monkeypatch, provider, route="source"): +def _primary_monitor_fixture( + tmp_path, monkeypatch, provider, route="source", *, + resume_when=None, handoff_mode="soft_claim", +): isolate_sqlite_runtime(tmp_path, monkeypatch) - source, runtime, _state = _write_fixture(tmp_path, provider=provider) + source, runtime, _state = _write_fixture( + tmp_path, provider=provider, handoff_mode=handoff_mode) gated = add_goal_todo(registry_path=source, goal_id=GOAL_ID, role="agent", text="Await a scoped owner decision", task_class="advancement_task", status="blocked", claimed_by=AGENT_ID, agent_id=AGENT_ID) gate = add_goal_todo(registry_path=source, goal_id=GOAL_ID, role="user", text="Decide the dependent work", task_class="user_gate", bound_agent=AGENT_ID, blocks_agent=AGENT_ID, unblocks_todo_id=gated["todo_id"], agent_id=AGENT_ID) - monitor = _add_monitor(source, text="Observe a synthetic public target", - target_key="primary-target", next_due_at="2000-01-01T00:00:00Z") + monitor = add_goal_todo(registry_path=source, goal_id=GOAL_ID, role="agent", + text="Observe a synthetic public target", task_class="continuous_monitor", + claimed_by=AGENT_ID, resume_when=resume_when, + monitor_metadata={"target_key": "primary-target", "cadence": "1h", + "next_due_at": "2000-01-01T00:00:00Z", "watch_only": "true"}) registry = source if route == "global": registry = runtime / "registry.global.json" @@ -134,6 +141,76 @@ def _primary_monitor_args(command, turn, monitor): "--result-hash", "primary-observation", "--execute"] if command == "monitor-poll" else [])] +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("route", ["source", "global"]) +@pytest.mark.parametrize("condition", [ + "resume_at:1999-01-01T00:00:00Z", + "resume_at:2099-01-01T00:00:00Z", + "todo_done:todo_missing", +], ids=["elapsed", "pending", "unresolved"]) +def test_primary_monitor_poll_uses_canonical_resume_readiness( + tmp_path, monkeypatch, provider, route, condition, +): + elapsed = "resume_at:1999-01-01T00:00:00Z" + source, registry, runtime, _gated, gate, monitor = _primary_monitor_fixture( + tmp_path, monkeypatch, provider, route, resume_when=elapsed, handoff_mode="hard_lease") + turn = "primary-monitor-resume-readiness" + + def call(*args): + return run_json_cli_result(*args, registry_path=registry, runtime_root=runtime) + + rc, admitted = call(*_primary_monitor_args("should-run", turn, monitor)) + binding = admitted["heartbeat_receipt"]["settlement_identity"] + assert rc == 0 and binding["todo_id"] == monitor["todo_id"] + assert admitted["safe_bypass_kind"] == "scoped_user_gate_fallback" + rc, acquired = call("task-lease", "acquire", "--goal-id", GOAL_ID, + "--todo-id", monitor["todo_id"], "--owner", AGENT_ID, "--idempotency-key", "monitor-resume-lease") + assert rc == 0 and acquired["acquired"], acquired + lease = acquired["lease"] + before = list_goal_todos(registry_path=source, goal_id=GOAL_ID, todo_id=monitor["todo_id"])["todo"] + assert before["resume_ready"] is True + if condition != before["resume_when"]: + update_goal_todo(registry_path=source, goal_id=GOAL_ID, todo_id=monitor["todo_id"], + agent_id=AGENT_ID, resume_when=condition, + task_lease_idempotency_key=lease["idempotency_key"], task_lease_expected_version=lease["version"]) + before = list_goal_todos(registry_path=source, goal_id=GOAL_ID) + row = next(item for item in before["todos"] if item["todo_id"] == monitor["todo_id"]) + index = runtime / "goals" / GOAL_ID / "runs" / "index.jsonl" + args = _primary_monitor_args("monitor-poll", turn, monitor) + # Cover the exact-id source route and target-key-only global route. + if route == "global": + pos = args.index("--todo-id") + del args[pos:pos + 2] + if condition == elapsed: + rc, denied = call(*args) + assert rc != 0 and "lease proof" in denied["reason"], denied + assert_todo_read_unchanged(list_goal_todos(registry_path=source, goal_id=GOAL_ID), before) + assert not index.exists() + args += ["--task-lease-idempotency-key", lease["idempotency_key"], + "--task-lease-expected-version", str(lease["version"])] + rc, result = call(*args) + if condition != elapsed: + assert row["resume_ready"] is False + assert rc != 0 and result["appended"] is False, result + assert_todo_read_unchanged(list_goal_todos(registry_path=source, goal_id=GOAL_ID), before) + assert not index.exists() + return + + assert rc == 0 and result["appended"], result + assert result["settlement_todo_id"] == binding["todo_id"] + assert result["turn_continuation"]["current_turn_settled"] is True + committed = list_goal_todos(registry_path=source, goal_id=GOAL_ID) + after = {item["todo_id"]: item for item in committed["todos"]} + assert after[monitor["todo_id"]]["result_hash"] == "primary-observation" + assert after[monitor["todo_id"]]["resume_when"] == condition + assert after[gate["todo_id"]]["status"] == "open" + ledger = index.read_bytes() + rc, replay = call(*args) + assert rc == 0 and replay["replayed"] and not replay["appended"], replay + assert index.read_bytes() == ledger + assert [json.loads(line)["classification"] for line in ledger.splitlines()] == ["quota_monitor_poll"] + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("route", ["source", "global"]) @pytest.mark.parametrize("gate_kind", ["linked", "unknown", "global"]) @@ -205,7 +282,7 @@ def test_primary_monitor_gate_change_after_preflight_prevents_business_write( ): from loopx.control_plane.quota import monitor_poll source, registry, runtime, _gated, gate, monitor = _primary_monitor_fixture( - tmp_path, monkeypatch, provider) + tmp_path, monkeypatch, provider, resume_when="resume_at:1999-01-01T00:00:00Z") invoke = _primary_record_invocation(registry, runtime, monitor, "primary-gate-race") original = monitor_poll._provider_writeback def interleaved(plan, **kwargs):