fix(scheduler): preserve receipt-owned blocked retry deadlines

docs/reference/protocols/quota-blocked-causal-closeout-v0.md

@@ -189,3 +189,39 @@ host Turn 可选择独立合法任务，不新造第三身份或重绑旧 Turn
 这是既有 quota owner 的 R1/S2/S10 有界修复，兼容保留活动调用者使用的
 `isBoundedBlockedRetry` 名称；只有 advancement schema 产生有界重试投影。
 不新增 Python 判断源、开关、前端控件或 Lark 权威；CLI 计划提供同源说明。
+
+### Receipt-owned retry projection / 回执所属的重试投影
+
+A promoted blocked `outcome_gap` Run can own a five-minute retry while its
+canonical Todo remains OPEN with no `resume_when`. Quota selection overlays the
+wait only for the same Goal/Agent/Todo, until the original due instant; a newer
+work Run supersedes it. The read-only projection now retains `blocked_retry`
+with the original `turn_instance_id`, source and clocks. The existing typed
+Todo frontier owns the earliest deadline before display compaction; thin
+scheduler output carries `pending_retry` from that same result.
+
+When that retry temporarily removes the only executable work, an independently
+open `user_action` remains a notice rather than becoming the reason work must
+wait. Execution still waits: `must_attempt_work=false`, no delivery or spend.
+The scheduler caps quiet/human/agent-scope/monitor waiting to the original retry
+horizon, then applies the existing owner cadence floor. A real user gate still
+blocks work. The wakeup is a quota recheck, not execution permission or an
+exact-time promise; all current gates, claims and leases must be read again.
+At or after due, the temporary overlay disappears without a maintenance write.
+
+CLI and managed heartbeat packets share this projection. This is a backend
+prerequisite: packaged Dashboard and live Lark rendering/adoption are separate
+acceptance stages. No new timer, authority, setting, canonical Todo field,
+financial clock, or automatically executing display action is introduced.
+
+原生 blocked/outcome_gap Run 可以拥有五分钟 retry，而 canonical Todo 仍为
+OPEN、没有 resume_when。临时选择投影只作用于同一 Goal/Agent/Todo，到原到期
+时点即消失；更新的工作 Run 会替代旧等待。blocked_retry 保留原 Turn、来源和
+时钟，typed Todo frontier 在显示压缩前选定最早时点；精简 scheduler 包的
+pending_retry 来自同一结果。
+
+retry 暂时移除唯一可执行工作时，独立 user_action 仍是提醒，不是恢复工作的
+前置门禁。此时仍不执行或扣额。scheduler 先按原 retry 时点收紧等待，再应用
+既有角色间隔下限；真实用户门禁继续阻塞执行。唤醒只意味着重新读取 quota、
+claim/lease 与当前门禁，不授权执行，也不承诺精确时刻运行。到期不需修改原
+Todo 或回执。CLI/heartbeat 同源；打包 App、真实 Lark 展示与采用仍须分别验收。

loopx/control_plane/quota/blocked_retry.py

@@ -139,7 +139,10 @@ def overlay_active_turn_retries(
         seen.add(todo_id)
         retry = active_turn_retry_for_run(run, observed_at=observed_at)
         if retry is not None:
-            waits[todo_id] = retry
+            waits[todo_id] = {
+                **retry, "goal_id": goal_id, "agent_id": agent_id,
+                "turn_instance_id": run["turn_instance_id"],
+            }
     if not waits:
         return status_payload
     queue = status_payload.get("attention_queue")
@@ -170,13 +173,18 @@ def overlay_summary(summary: Any) -> tuple[Any, bool]:
                     and todo.get("status") == "open"
                     and todo.get("task_class") == "advancement_task"
                     and not todo.get("resume_when")
+                    and todo.get("claimed_by") in (None, agent_id)
                 ):
                     resume = retry["resume_when"]
                     projected_rows.append(
                         {
                             **todo,
                             "resume_when": resume,
                             "resume_ready": False,
+                            # Read-only Run provenance; never persist these fields
+                            # into the canonical Todo or treat them as admission.
+                            "blocked_retry": retry,
+                            "next_due_at": retry["due_at"],
                             "resume_condition": {
                                 "kind": "resume_at",
                                 "resume_when": resume,

loopx/control_plane/quota/should_run_packet.py

@@ -236,6 +236,7 @@ def _execution_obligation(
     work_lane_contract: dict[str, Any] | None = None,
     external_evidence_observation: dict[str, Any] | None = None,
     user_gate_owns_frontier: bool = False,
+    blocked_retry_wait: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     return build_execution_obligation(
         should_run=should_run,
@@ -244,6 +245,7 @@ def _execution_obligation(
         work_lane_contract=work_lane_contract,
         external_evidence_observation=external_evidence_observation,
         user_gate_owns_frontier=user_gate_owns_frontier,
+        blocked_retry_wait=blocked_retry_wait,
         successor_replan_mode=AgentScopeFrontierAction.SUCCESSOR_REPLAN_REQUIRED.value,
     )
 
@@ -1283,6 +1285,14 @@ def _build_active_quota_payload(
     turn_instance_id: str | None = None,
 ) -> dict[str, Any]:
     agent_scope_action = _agent_scope_frontier_action(route.effective_action)
+    retry_deadline = (prepared.agent_todo_summary or {}).get("frontier_deadline") or {}
+    user_summary = prepared.user_todo_summary or {}
+    # Only an explicitly complete, notice-only user lane can replace the user
+    # gate explanation. Genuine gates and unknown scope keep their old route.
+    notice_only = (user_summary.get("user_action_open_count", 0) > 0 and
+                   user_summary.get("all_open_count") == user_summary.get("user_action_open_count"))
+    retry_wait = (retry_deadline.get("blocked_retry")
+                  if notice_only and retry_deadline.get("source") == "turn_settlement" else None)
     execution_obligation = _execution_obligation(
         should_run=route.should_run,
         effective_action=route.effective_action,
@@ -1293,6 +1303,7 @@ def _build_active_quota_payload(
         ),
         work_lane_contract=route.payload_work_lane_contract,
         external_evidence_observation=route.external_evidence_observation,
+        blocked_retry_wait=retry_wait,
         user_gate_owns_frontier=_user_action_owns_empty_agent_lane(
             prepared.user_todo_summary,
             prepared.agent_todo_summary,

loopx/control_plane/scheduler/arbitration.py

@@ -76,6 +76,7 @@ def _classify_disposition(
     must_attempt: bool,
     quiet_noop_allowed: bool,
     agent_scope_frontier_actions: Collection[str],
+    non_gating_notice_retry: bool = False,
 ) -> tuple[SchedulerDisposition, str]:
     agent_scope_modes = {str(value) for value in agent_scope_frontier_actions}
     if mode == "terminal_no_followup":
@@ -84,6 +85,8 @@ def _classify_disposition(
         return SchedulerDisposition.PEER_COORDINATION_WAIT, mode
     if mode == "agent_monitor_only":
         return SchedulerDisposition.AGENT_MONITOR_ONLY_WAIT, mode
+    if user_required and not must_attempt and non_gating_notice_retry:
+        return SchedulerDisposition.QUIET_WAIT, "blocked_wait"
     if user_required and not must_attempt:
         return SchedulerDisposition.HUMAN_GATE, "interaction_blocking_user_gate"
     if mode == "monitor_quiet_skip":
@@ -177,6 +180,8 @@ def build_scheduler_arbitration(
         must_attempt=must_attempt,
         quiet_noop_allowed=quiet_noop_allowed,
         agent_scope_frontier_actions=agent_scope_frontier_actions,
+        non_gating_notice_retry=(user_channel.get("non_gating_notice") is True and
+            _mapping(agent_channel.get("wait")).get("source") == "turn_settlement"),
     )
 
     if errors:

loopx/control_plane/scheduler/scheduler_hint.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import math
+
 import base64
 import hashlib
 import json
@@ -626,6 +628,30 @@ def build_codex_app_scheduler_fallback_hint(
     }
 
 
+def _pending_retry_projection(payload: dict[str, Any]) -> dict[str, Any]:
+    deadline = _dict_or_empty(_dict_or_empty(payload.get("agent_todo_summary")).get("frontier_deadline"))
+    retry = deadline.get("blocked_retry")
+    if deadline.get("source") != "turn_settlement" or not isinstance(retry, dict):
+        return {}
+    return {"pending_retry": {**retry, "execution_authority_granted": False,
+        "recheck": "rerun quota at or after due_at; current gates and owner cadence still apply"}}
+
+
+def _receipt_retry_wait_profile(
+    payload: dict[str, Any], cadence_class: str, interval_minutes: int,
+) -> dict[str, Any] | None:
+    # This is a wakeup recommendation. TS cadence admission still owns the
+    # floor; a deadline never changes claims, leases or execution authority.
+    retry = _pending_retry_projection(payload).get("pending_retry")
+    current_time = now_utc()
+    due = parse_scheduler_timestamp(retry.get("due_at")) if retry else None
+    if (cadence_class not in {"human_gate", "quiet_wait", "agent_scope_wait", "monitor_wait"}
+            or due is None or due <= current_time):
+        return None
+    horizon = max(1, math.ceil((due - current_time).total_seconds() / 60))
+    return {"interval_minutes": horizon, "retry": retry} if horizon < interval_minutes else None
+
+
 @dataclass(frozen=True)
 class _SchedulerHintBuilder:
     payload: dict[str, Any]
@@ -664,6 +690,18 @@ def _cadence_projections(
             **projections, "min_interval_minutes": floor,
         }) if floor else projections
 
+    def _wait_cadence(
+        self, cadence_class: str, interval: int, maximum: int, multiplier: int,
+        progression: list[int] | None, reset: dict[str, Any] | None,
+        detail: dict[str, Any] | None,
+    ) -> tuple[dict[str, Any], dict[str, Any] | None, dict[str, Any] | None]:
+        retry_wait = _receipt_retry_wait_profile(self.payload, cadence_class, interval)
+        if retry_wait:
+            interval = retry_wait["interval_minutes"]
+            progression, reset = [interval], None
+            detail = {**(detail or {}), "blocked_retry": retry_wait["retry"]}
+        return self._cadence_projections(interval, maximum, multiplier, progression), reset, detail
+
     def build(
         self,
         *,
@@ -693,7 +731,10 @@ def build(
             (self.codex_app_scheduler_state or {}).get("state_key")
             or APP_AUTOMATION_STATEFUL_BACKOFF_STATE_KEY
         )
-        cadence = self._cadence_projections(codex_interval, codex_max, multiplier, cadence_progression_override)
+        cadence, reset_profile_snapshot_override, cadence_context_detail = self._wait_cadence(
+            cadence_class, codex_interval, codex_max, multiplier, cadence_progression_override,
+            reset_profile_snapshot_override, cadence_context_detail,
+        )
         local_cadence_progression, app_cadence_progression = cadence["local"], cadence["app"]
         app_host_max, codex_max, floor = cadence["app_max"], cadence["local_max"], cadence["floor"]
         app_initial_interval = app_cadence_progression[0]
@@ -1057,6 +1098,7 @@ def build(
                 "contains": detail_contains,
             },
         }
+        scheduler_hint.update(_pending_retry_projection(self.payload))
         if app_surface == CODEX_APP_SURFACE:
             scheduler_hint["codex_app"] = build_codex_app_compatibility_projection(
                 app_automation,
@@ -1589,13 +1631,16 @@ def build_scheduler_hint(
             action="backoff_until_state_change",
             cadence_class="quiet_wait",
             reason=(
+                "the receipt-owned retry is pending; recheck its original due time "
+                "and current admission gates without waiting on the non-blocking user notice"
+                if arbitration.reason_code == "blocked_wait" else
                 "quota blocks delivery and no immediate user/monitor-specific path "
                 "is projected; poll at a slower cadence until the status changes"
             ),
-            codex_interval=30,
-            codex_max=120,
             cli_limit=3,
             claude_limit=3,
+            codex_interval=30,
+            codex_max=120,
         )
 
     raise AssertionError(f"unhandled scheduler disposition: {arbitration.disposition}")

loopx/control_plane/todos/quota_selection.ts

@@ -1,3 +1,4 @@
+import {isBoundedBlockedRetry} from "../quota/settlement_phase.ts";
 import {parseTodoTimestampMicros} from "../runtime_timestamp.ts";
 import {countTodoWork, decodeMonitorSchedule, monitorIsDue, monitorHasScheduleGap,
   type MonitorSchedule} from "./summary_lanes.ts";
@@ -324,7 +325,7 @@ 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 projectTodoFrontierDeadline(summary: JsonObject, now: bigint): JsonObject | null {
+function projectTodoFrontierDeadline(summary: JsonObject, now: bigint, agent: string | null): JsonObject | null {
   const projected = object(summary.frontier_deadline);
   const projectedTime = timestamp(projected?.next_due_at);
   if (projected && projectedTime !== null && projectedTime > now) return {...projected};
@@ -343,7 +344,16 @@ function projectTodoFrontierDeadline(summary: JsonObject, now: bigint): JsonObje
       seen.add(key);
       if (earliest === null || due < earliest) {
         earliest = due;
-        selected = {identity: id, source: source(row, lane), next_due_at: utcIso(due)};
+        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);
+        selected = {identity: id, source: receiptRetry ? "turn_settlement" : source(row, lane),
+          next_due_at: utcIso(due), ...(receiptRetry ? {blocked_retry: retry} : {})};
       }
     }
   }
@@ -394,7 +404,7 @@ export function projectTodoQuotaPlanning(value: unknown): JsonObject {
     frontier_deadline: request.frontier_deadline ?? null,
     monitor_open_items: lanes.monitor_items,
     gate_open_items: lanes.gate_items,
-  }, now);
+  }, now, agent);
   // The full gate lane stays inside this batch. Transport only the original
   // three-row display after deadline selection, rather than duplicating bodies.
   if (now !== null) lanes.gate_items = (lanes.gate_items as JsonObject[]).slice(0, 3);

loopx/control_plane/todos/quota_summary.py

@@ -67,6 +67,7 @@
     "resume_monitor_generation",
     "resume_condition",
     "resume_ready",
+    "blocked_retry",
     "blocking_monitor_todo_id",
     "no_followup",
     "successor_todo_ids",

loopx/control_plane/todos/resume_planning.py

@@ -23,7 +23,7 @@
 
 def _planning_item(item: dict[str, Any]) -> dict[str, Any]:
     payload = compact_todo_projection_item(
-        item, text=str(item.get("text") or "").strip(), extra_fields=("content_revision",)
+        item, text=str(item.get("text") or "").strip(), extra_fields=("content_revision", "blocked_retry")
     )
     condition = item.get("resume_condition")
     condition = condition if isinstance(condition, dict) else {}

loopx/control_plane/todos/summary_item.py

@@ -58,6 +58,7 @@
     "resume_monitor_generation",
     "resume_condition",
     "resume_ready",
+    "blocked_retry",
     "no_followup",
     "successor_todo_ids",
     "completion_continuation",

loopx/control_plane/work_items/execution_obligation.py

@@ -17,6 +17,7 @@ def build_execution_obligation(
     external_evidence_observation: dict[str, Any] | None = None,
     user_gate_owns_frontier: bool = False,
     successor_replan_mode: str = SUCCESSOR_REPLAN_REQUIRED_MODE,
+    blocked_retry_wait: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Separate the worker execution contract from user-facing notification."""
 
@@ -218,6 +219,19 @@ def build_execution_obligation(
                 "only controls whether to interrupt the user"
             ),
         }
+    if should_run and blocked_retry_wait:
+        return {
+            "must_attempt_work": False,
+            "kind": EffectiveAction.BLOCKED_WAIT.value,
+            "delivery_allowed": False,
+            "notify_is_execution_gate": False,
+            "wait": blocked_retry_wait,
+            "reason": (
+                "the unfinished Todo has a receipt-owned bounded retry; recheck "
+                "its original due time and all current gates before delivery. "
+                "The separate user-action notice is non-blocking"
+            ),
+        }
     if should_run:
         return {
             "must_attempt_work": False,

loopx/control_plane/work_items/interaction_contract.py

@@ -1721,6 +1721,11 @@ def build_interaction_contract(
             quiet_noop_allowed=quiet_noop_allowed,
             capability_reentry=capability_reentry,
         )
+    retry_wait = execution_obligation.get("wait")
+    if execution_obligation.get("kind") == EffectiveAction.BLOCKED_WAIT.value and isinstance(retry_wait, dict):
+        agent_channel["wait"] = retry_wait
+        agent_channel["primary_action"] = execution_obligation["reason"]
+        user_channel["non_gating_notice"] = True
     contract: dict[str, Any] = {
         "schema_version": INTERACTION_CONTRACT_SCHEMA_VERSION,
         "mode": mode,