fix(quota): retain receipt retries beside existing wait lanes

docs/architecture/rfcs/loopx-overall-roadmap-v0.md

@@ -87,6 +87,13 @@ P0 blocks correctness or continuity in the current user journey. P1 enables repe
 | **S12 Release, developer experience and community governance · P0 hygiene/P1** | Install/source validation, registration, DCO/PR, test layers, contributor routes and bilingual docs exist | Qualify first work and upgrade/rollback from clean machines/release artifacts; host/OS support follows the release contract. Reduce localization/test/review effort for useful changes; preserve exact-head evidence, fixtures, compatibility, maintainer routing and contributor credit; retire duplicate protocols/stale evidence |
 | **S13 Adoption, ecosystem and sustainability · P1 discovery/P2 pilots** | Public adoption loop, showcases, licensing/governance and observer-first product contract exist; paid PMF is unproven | Gather independent first/repeat usage and exit reasons; reproducible cases and pilots with fixed budgets/acceptance/rollback. Retain reusable adapters/delivery guides. Account for model/compute/storage/support and maintenance costs; only repeated demand justifies commercial hosting/support/distribution decisions, with no invented SLA or open-source-term change |
 
+**S7 retry projection checkpoint.** Receipt waits introduced after stored resume
+lanes were built also need the claim-scoped full selection source. Synthetic
+composition and real File/SQLite CLI replay cover a pre-existing wait lane,
+visibility limits, ordinary notices, current cadence floors and unchanged
+canonical waits/monitor clocks. Installed adoption and live host readback remain
+separate from this source repair.
+
 ### Important Areas Outside RFCs and Their Owning Sources
 
 These directly determine whether a long-running team is usable. A directory or README provides an implementation/contract entry, not installed readiness; read availability from `loopx capability list/show` and the selected release.

docs/quota-allocation.md

@@ -465,6 +465,10 @@ monitor clock, genuine gate and independent runnable work keep their own rules.
 This corrects the automatic notice explanation without adding configuration or
 persisted state. The pending retry is a recheck deadline, not execution authority:
 current gates and the configured cadence floor still apply at and after due time.
+Receipt waits added after stored resume lanes were built are read from the same
+claim-scoped full selection source. An existing wait lane or its display limit
+cannot hide a new retry; foreign, excluded, mismatched and replaced waits do not
+qualify through this path.
 
 类型化 deadline owner 会把新叠加的、更早到期的 Turn 重试纳入已有的完整来源
 缓存，同时保留显示列表之外的更早监控时钟和未变化的缓存。同一个 typed batch
@@ -474,6 +478,9 @@ current gates and the configured cadence floor still apply at and after due time
 时钟、真实 gate 和独立可执行工作各自保持。这是自动通知解释的修正，不新增
 配置或持久化状态。重试到期仅要求重新检查，不授予执行权限；当前 gate 和
 已配置的调度周期下限仍然有效。
+stored resume lane 建立后新增的 receipt wait，沿同一个 claim-scoped 完整
+selection 来源读回。已有等待列表及其显示上限不能隐藏新重试；其它 Agent、
+被排除的执行者、身份不匹配或已被替换的等待不能由这条路径取得资格。
 
 `quota should-run` also separates long-running observation from work that should
 advance the selected goal. When the selected goal's current projection is a

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,

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)

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"  <!-- loopx:todo todo_id=todo_existing_wait status=open claimed_by={cli.AGENT_ID} "
+            "task_class=advancement_task action_kind=validate "
+            f"resume_when={quote('resume_at:2036-01-01T00:00:00Z', safe='')} -->\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"  <!-- loopx:todo todo_id=todo_retry_notice status=open bound_agent={cli.AGENT_ID} "
@@ -102,10 +109,10 @@ def run(*args: str):
                           "--turn-instance-id", "turn-retry-next")
         assert code == 0 and guard["ok"], guard
         assert guard["heartbeat_receipt"]["status"] in {"committed", "replayed"}
-        if scenario in {"waiting", "notice_waiting", "earlier_notice", "equal_notice"}:
+        if scenario in {"waiting", "notice_waiting", "notice_existing_wait", "earlier_notice", "equal_notice"}:
             assert guard["effective_action"] == "monitor_quiet_skip"
             assert guard["execution_obligation"]["must_attempt_work"] is False
-            if scenario in {"notice_waiting", "earlier_notice", "equal_notice"}:
+            if scenario in {"notice_waiting", "notice_existing_wait", "earlier_notice", "equal_notice"}:
                 assert guard["execution_obligation"]["kind"] == "blocked_wait"
                 assert guard["interaction_contract"]["user_channel"]["non_gating_notice"] is True
                 assert guard["scheduler_hint"]["cadence_class"] == "quiet_wait"
@@ -131,4 +138,8 @@ def run(*args: str):
     assert watch["next_due_at"] == due
     assert watch["result_hash"] == "unchanged"
     assert watch["material_change_generation"] == 7
+    if scenario == "notice_existing_wait":
+        existing = next(todo for todo in after["todos"] if todo["todo_id"] == "todo_existing_wait")
+        assert existing["resume_when"] == "resume_at:2036-01-01T00:00:00Z"
+        assert existing["resume_ready"] is False
     assert cli._spend_run_count(runtime) == 0