refactor(quota): isolate deferred receipt projection

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>

Author: huangruiteng

loopx/control_plane/quota/settlement_precedence.py

@@ -9,6 +9,10 @@
     "the receipt-bound work binding and required settlement receipts "
     "are complete for this heartbeat turn; defer successor selection to a new turn"
 )
+RECEIPT_BOUND_DEFERRED_REASON = (
+    "the Todo bound to this heartbeat receipt is deferred; do not "
+    "select or spend an independent successor in the same Turn"
+)
 
 _ACTION_PROJECTION_KEYS = (
     "agent_command",
@@ -89,3 +93,26 @@ def settled_replay_fields() -> dict[str, Any]:
             "spend_policy": "no quota spend for an already-settled heartbeat turn",
         },
     }
+
+
+def deferred_receipt_bound_skip_fields(
+    quota: dict[str, Any],
+    heartbeat_recommendation: dict[str, Any],
+) -> tuple[str, str, dict[str, Any], dict[str, Any]]:
+    """Project a deferred receipt without borrowing a successor's authority."""
+
+    effective_action = EffectiveAction.QUOTA_SKIP.value
+    reason = RECEIPT_BOUND_DEFERRED_REASON
+    return (
+        effective_action,
+        reason,
+        {**quota, "safe_bypass_allowed": False},
+        {
+            **heartbeat_recommendation,
+            "recommended_mode": effective_action,
+            "notify": "DONT_NOTIFY",
+            "reason": reason,
+            "spend_policy": "no quota spend for a deferred receipt-bound Todo",
+            "stop_if_unchanged": True,
+        },
+    )

loopx/control_plane/quota/should_run_packet.py

@@ -129,6 +129,7 @@
     work_lane_contract_is_receipt_bound_monitor_settled,
 )
 from .settlement_precedence import (
+    deferred_receipt_bound_skip_fields,
     settled_replay_fields,
     HEARTBEAT_SETTLED_REPLAY_REASON,
     clear_quota_action_projections,
@@ -889,31 +890,32 @@ def _resolve_quota_should_run_route(
             "reason": reason,
             "spend_policy": "no quota spend for an already-settled heartbeat turn",
         }
-    receipt_bound_deferred_wait = bool(
+    if (
         prepared.receipt_bound_todo_id
         and isinstance(prepared.work_lane_contract, dict)
         and prepared.work_lane_contract.get("obligation")
         == WORK_LANE_RECEIPT_BOUND_DEFERRED_OBLIGATION
-    )
-    if receipt_bound_deferred_wait:
-        normal_delivery_allowed = recovery_allowed = self_repair_allowed = False
-        capability_repair_allowed = workspace_repair_allowed = False
-        replan_decision_allowed = receipt_bound_replan_decision = False
-        should_run = False
-        effective_action = EffectiveAction.QUOTA_SKIP.value
-        reason = (
-            "the Todo bound to this heartbeat receipt is deferred; do not "
-            "select or spend an independent successor in the same Turn"
+    ):
+        # Every action path is closed for this immutable, deferred receipt.
+        (
+            normal_delivery_allowed,
+            recovery_allowed,
+            self_repair_allowed,
+            capability_repair_allowed,
+            workspace_repair_allowed,
+            replan_decision_allowed,
+            receipt_bound_replan_decision,
+            should_run,
+        ) = (False,) * 8
+        (
+            effective_action,
+            reason,
+            quota,
+            heartbeat_recommendation,
+        ) = deferred_receipt_bound_skip_fields(
+            quota,
+            heartbeat_recommendation,
         )
-        quota = {**quota, "safe_bypass_allowed": False}
-        heartbeat_recommendation = {
-            **heartbeat_recommendation,
-            "recommended_mode": effective_action,
-            "notify": "DONT_NOTIFY",
-            "reason": reason,
-            "spend_policy": "no quota spend for a deferred receipt-bound Todo",
-            "stop_if_unchanged": True,
-        }
     monitor_quiet_skip = (
         not replan_decision_allowed
         and normal_delivery_allowed

loopx/control_plane/quota/should_run_prepare.py

@@ -452,6 +452,20 @@ def _build_agent_work_lane(
     return monitor_only, work_lane, task_orchestration
 
 
+def _deferred_receipt_bound_work_lane(
+    *, todo_id: str, source_items: list[dict[str, Any]],
+) -> dict[str, Any] | None:
+    """Keep a deferred Todo's old receipt visible without selecting new work."""
+
+    if any(
+        normalize_todo_id(source_item.get("todo_id")) == todo_id
+        and normalize_todo_status(source_item.get("status")) == TODO_STATUS_DEFERRED
+        for source_item in source_items
+    ):
+        return receipt_bound_deferred_work_lane(todo_id=todo_id)
+    return None
+
+
 def _prepare_quota_should_run_item(
     status_payload: dict[str, Any],
     *,
@@ -738,24 +752,24 @@ def _prepare_quota_should_run_item(
             and candidate.get("selection_binding") == "heartbeat_receipt"
         ):
             receipt_bound_agent_next_action = candidate
-            preserved_work_lane = preserve_heartbeat_receipt_bound_work_lane(
-                work_lane_contract,
-                selected_todo=candidate,
+            work_lane_contract = (
+                preserve_heartbeat_receipt_bound_work_lane(
+                    work_lane_contract,
+                    selected_todo=candidate,
+                )
+                or work_lane_contract
             )
-            if isinstance(preserved_work_lane, dict):
-                work_lane_contract = preserved_work_lane
-        elif any(
-            normalize_todo_id(source_item.get("todo_id")) == receipt_bound_todo_id
-            and normalize_todo_status(source_item.get("status"))
-            == TODO_STATUS_DEFERRED
-            for source_item in agent_todo_planning_source_items
-        ):
+        else:
             # The old Turn still owns its committed settlement identity, but a
             # deferred Todo is not an executable candidate.  A successor may be
             # selected only by a fresh Turn; do not leak it through work-lane
             # fallback on this replay.
-            work_lane_contract = receipt_bound_deferred_work_lane(
-                todo_id=receipt_bound_todo_id,
+            work_lane_contract = (
+                _deferred_receipt_bound_work_lane(
+                    todo_id=receipt_bound_todo_id,
+                    source_items=agent_todo_planning_source_items,
+                )
+                or work_lane_contract
             )
     if inbox_priority_due:
         task_orchestration_contract = capability_gate = capability_monitor_contract = None