From d266d34bf7c2cdbf8961acc709ef80b9c26eef36 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 12 Oct 2026 16:05:47 +0800 Subject: [PATCH 1/2] fix(quota): retain receipt retries beside existing wait lanes Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/todos/quota_selection.ts | 32 ++++++++--- .../test_blocked_retry_wait_context.py | 54 +++++++++++++++++-- .../test_quota_retry_monitor_journey.py | 19 +++++-- 3 files changed, 89 insertions(+), 16 deletions(-) diff --git a/loopx/control_plane/todos/quota_selection.ts b/loopx/control_plane/todos/quota_selection.ts index 05d877c740..97598183b2 100644 --- a/loopx/control_plane/todos/quota_selection.ts +++ b/loopx/control_plane/todos/quota_selection.ts @@ -325,6 +325,17 @@ function utcIso(micros: bigint): string { const date = new Date(Number(seconds * 1000n)).toISOString().slice(0, -5); return `${date}${fraction ? `.${String(fraction).padStart(6, "0")}` : ""}+00:00`; } +function scopedReceiptRetry(row: JsonObject, due: bigint, agent: string | null): JsonObject | null { + const retry = object(row.blocked_retry); + return row.status === "open" && row.task_class === "advancement_task" && + row.resume_ready === false && retry?.source === "turn_settlement" && + typeof retry.turn_instance_id === "string" && !!retry.turn_instance_id && + retry.agent_id === agent && agent !== null && typeof retry.goal_id === "string" && + (!row.claimed_by || row.claimed_by === retry.agent_id) && + row.resume_when === retry.resume_when && timestamp(retry.due_at) === due && + isBoundedBlockedRetry(retry, typeof row.todo_id === "string" ? row.todo_id : null) + ? retry : null; +} function projectTodoFrontier(summary: JsonObject, now: bigint, agent: string | null): JsonObject { const projected = object(summary.frontier_deadline); const projectedTime = timestamp(projected?.next_due_at); @@ -346,14 +357,8 @@ function projectTodoFrontier(summary: JsonObject, now: bigint, agent: string | n const id = identity(row), key = JSON.stringify([id, String(due)]); if (seen.has(key)) continue; seen.add(key); - const retry = object(row.blocked_retry); - const receiptRetry = row.status === "open" && row.task_class === "advancement_task" && - row.resume_ready === false && retry?.source === "turn_settlement" && - typeof retry.turn_instance_id === "string" && !!retry.turn_instance_id && - retry.agent_id === agent && agent !== null && typeof retry.goal_id === "string" && - (!row.claimed_by || row.claimed_by === retry.agent_id) && - row.resume_when === retry.resume_when && timestamp(retry.due_at) === due && - isBoundedBlockedRetry(retry, typeof row.todo_id === "string" ? row.todo_id : null); + const retry = scopedReceiptRetry(row, due, agent); + const receiptRetry = retry !== null; // A receipt wait owns one unfinished Todo; the earliest wakeup may // belong to a different monitor. Keep both facts from this full source. if (receiptRetry && (retryDue === null || due < retryDue)) { @@ -418,6 +423,16 @@ export function projectTodoQuotaPlanning(value: unknown): JsonObject { const route = hasSourceContract ? routeContinuationLanes(request.route_items, agent, limit) : {}; const lanes = requireJsonObject(selected.lanes, "selection lanes"); const resumeBlocked = requireJsonObject(resume.resume_blocked_lanes, "resume blocked lanes"); + // An overlay can add a receipt wait after the stored resume lane was built. + // Read its scoped full-source row; an existing nonempty display lane must + // neither hide that retry nor impose its visibility limit on retry clocks. + const receiptWaits = now === null ? [] : rows(selection.items, + selection.available === undefined ? undefined : requireStringArray(selection.available, "available"), + observedAt).filter(row => { + const due = timestamp(row.payload.next_due_at); + return executableBy(row, agent) && due !== null && due > now && + scopedReceiptRetry(row.payload, due, agent) !== null; + }).map(row => row.payload); const frontier = now === null ? null : projectTodoFrontier({ ...requireJsonObject(selected.claim_visibility, "claim visibility"), ...requireJsonObject(resume.deferred_lanes, "deferred lanes"), @@ -427,6 +442,7 @@ export function projectTodoQuotaPlanning(value: unknown): JsonObject { resume_blocked_items: [ ...(Array.isArray(resumeBlocked.resume_blocked_items) ? resumeBlocked.resume_blocked_items : []), ...(resume.blocked_successor_items as JsonObject[]), + ...receiptWaits, ], frontier_deadline: request.frontier_deadline ?? null, monitor_open_items: lanes.monitor_items, diff --git a/tests/control_plane/test_blocked_retry_wait_context.py b/tests/control_plane/test_blocked_retry_wait_context.py index cd750a386f..2ba2bf27d4 100644 --- a/tests/control_plane/test_blocked_retry_wait_context.py +++ b/tests/control_plane/test_blocked_retry_wait_context.py @@ -15,7 +15,8 @@ DUE = "2026-10-01T00:05:00Z" -def source(*, user_class="user_action", newer=False, cached_monitor=False, monitor_due="2026-10-01T00:25:00Z"): +def source(*, user_class="user_action", newer=False, cached_monitor=False, + monitor_due="2026-10-01T00:25:00Z", existing_waits=0): retry = {"schema_version": "quota_blocked_retry_v0", "source": "turn_settlement", "todo_id": "todo_work", "observed_at": "2026-10-01T00:00:00Z", "due_at": DUE, "resume_when": f"resume_at:{DUE}"} @@ -34,7 +35,14 @@ def source(*, user_class="user_action", newer=False, cached_monitor=False, monit "text": "Observe the independent source.", "watch_only": True, "cadence": "30m", "next_due_at": monitor_due, "target_key": "independent-source", - }] if cached_monitor else []), + }] if cached_monitor else []) + [{ + "todo_id": f"todo_existing_wait_{index}", "status": "open", "role": "agent", + "task_class": "advancement_task", "claimed_by": "agent-a", + "text": "Wait for an independent prerequisite.", + "resume_when": "resume_at:2035-01-01T00:00:00Z", "resume_ready": False, + "next_due_at": "2035-01-01T00:00:00Z", + "resume_condition": {"kind": "resume_at", "satisfied": False}, + } for index in range(existing_waits)], user_todo_items=[{"todo_id": "todo_notice", "status": "open", "role": "user", "task_class": user_class, "bound_agent": "agent-a", **({"blocks_agent": "agent-a"} if user_class == "user_gate" else {}), @@ -53,16 +61,35 @@ def source(*, user_class="user_action", newer=False, cached_monitor=False, monit def packet(monkeypatch, *, seconds=120, user_class="user_action", newer=False, runtime=None, floor=0, - cached_monitor=False, monitor_due="2026-10-01T00:25:00Z"): + cached_monitor=False, monitor_due="2026-10-01T00:25:00Z", existing_waits=0, + retry_mismatch=None): now = START + timedelta(seconds=seconds) for path in ("loopx.control_plane.todos.quota_selection.now_utc", "loopx.control_plane.todos.quota_summary.now_utc", "loopx.control_plane.scheduler.monitor_todo.now_utc", "loopx.control_plane.scheduler.scheduler_hint.now_utc"): monkeypatch.setattr(path, lambda: now) - raw = source(user_class=user_class, newer=newer, cached_monitor=cached_monitor, monitor_due=monitor_due) + raw = source(user_class=user_class, newer=newer, cached_monitor=cached_monitor, + monitor_due=monitor_due, existing_waits=existing_waits) before = deepcopy(raw) projected = overlay_active_turn_retries(raw, goal_id="goal-a", agent_id="agent-a", observed_at=now.isoformat()) + if retry_mismatch: + for item in projected["attention_queue"]["items"]: + for summary in (item["agent_todos"], item["project_asset"]["agent_todos"]): + for values in summary.values(): + for row in values if isinstance(values, list) else []: + if not isinstance(row, dict) or row.get("todo_id") != "todo_work": + continue + if retry_mismatch == "claim": + row["claimed_by"] = "agent-b" + elif retry_mismatch == "agent": + row["blocked_retry"] = {**row["blocked_retry"], "agent_id": "agent-b"} + elif retry_mismatch == "excluded": + row["excluded_agents"] = ["agent-a"] + elif retry_mismatch == "retry_todo": + row["blocked_retry"] = {**row["blocked_retry"], "todo_id": "todo_other"} + elif retry_mismatch == "canonical_wait": + row["resume_when"] = "todo_done:todo_dependency" result = build_quota_should_run(projected, goal_id="goal-a", agent_id="agent-a", scheduler_execution_context=scheduler_execution_context_for_runtime_profile( runtime or SchedulerRuntimeProfile.CODEX_APP_HEARTBEAT)) @@ -93,6 +120,25 @@ def test_pending_retry_is_not_a_nonblocking_notice_gate(monkeypatch): assert hint["pending_retry"]["execution_authority_granted"] is False +@pytest.mark.parametrize("existing_waits", [1, 12]) +def test_receipt_retry_survives_preexisting_wait_lanes(monkeypatch, existing_waits): + result = packet(monkeypatch, cached_monitor=True, existing_waits=existing_waits, floor=6) + assert result["execution_obligation"]["kind"] == "blocked_wait" + assert result["interaction_contract"]["user_channel"]["non_gating_notice"] is True + assert result["scheduler_hint"]["cadence_class"] == "quiet_wait" + assert result["scheduler_hint"]["pending_retry"]["todo_id"] == "todo_work" + assert result["scheduler_hint"]["pending_retry"]["due_at"] == DUE + assert result["scheduler_hint"]["app_automation"]["recommended_interval_minutes"] == 6 + + +@pytest.mark.parametrize("mismatch", ["claim", "agent", "excluded", "retry_todo", "canonical_wait"]) +def test_full_source_retry_keeps_scope_and_receipt_fences(monkeypatch, mismatch): + result = packet(monkeypatch, cached_monitor=True, existing_waits=1, + retry_mismatch=mismatch) + assert "pending_retry" not in result["scheduler_hint"] + assert not result["agent_todo_summary"].get("blocked_retry_wait") + + @pytest.mark.parametrize("seconds,floor,interval", [(120, 0, 3), (295, 0, 1), (295, 6, 6)]) def test_retry_overlay_precedes_a_later_cached_monitor_deadline(monkeypatch, seconds, floor, interval): result = packet(monkeypatch, seconds=seconds, floor=floor, cached_monitor=True) diff --git a/tests/control_plane/test_quota_retry_monitor_journey.py b/tests/control_plane/test_quota_retry_monitor_journey.py index 3019f576db..7a4c4e4422 100644 --- a/tests/control_plane/test_quota_retry_monitor_journey.py +++ b/tests/control_plane/test_quota_retry_monitor_journey.py @@ -20,7 +20,7 @@ @pytest.mark.parametrize("provider", ["file", "sqlite"]) -@pytest.mark.parametrize("scenario", ["waiting", "notice_waiting", "notice_independent", "due_monitor", "independent", "no_retry", "earlier_notice", "equal_notice", "earlier_gate", "earlier_independent"]) +@pytest.mark.parametrize("scenario", ["waiting", "notice_waiting", "notice_existing_wait", "notice_independent", "due_monitor", "independent", "no_retry", "earlier_notice", "equal_notice", "earlier_gate", "earlier_independent"]) def test_retry_monitor_guard_and_replay_keep_receipt_and_work_authority( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, scenario: str, ) -> None: @@ -46,7 +46,14 @@ def test_retry_monitor_guard_and_replay_keep_receipt_and_work_authority( if scenario in {"independent", "notice_independent", "earlier_independent"} else "" )) - if scenario in {"notice_waiting", "notice_independent", "earlier_notice", "equal_notice", "earlier_gate", "earlier_independent"}: + if scenario == "notice_existing_wait": + state.write_text(state.read_text() + + "\n- [ ] Wait for another prerequisite.\n" + f" \n") + + if scenario in {"notice_waiting", "notice_existing_wait", "notice_independent", "earlier_notice", "equal_notice", "earlier_gate", "earlier_independent"}: state.write_text(state.read_text() + "\n## User Todo\n\n- [ ] Review an independent draft.\n" f"