fix(quota): keep deferred receipt-bound Turn from selecting a successor

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,24 @@ 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, dict[str, Any], dict[str, Any]]:
+    """Project a deferred receipt without borrowing a successor's authority."""
+
+    reason = RECEIPT_BOUND_DEFERRED_REASON
+    return (
+        reason,
+        {**quota, "safe_bypass_allowed": False},
+        {
+            **heartbeat_recommendation,
+            "recommended_mode": EffectiveAction.QUOTA_SKIP.value,
+            "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

@@ -85,6 +85,7 @@
 )
 from ..todos.contract import (
     normalize_todo_claimed_by,
+    normalize_todo_id,
 )
 from ..todos.todo_semantics import (
     todo_item_is_actionable_open as projection_todo_item_is_actionable_open,
@@ -123,10 +124,12 @@
     user_action_owns_empty_agent_lane_from_summaries as _user_action_owns_empty_agent_lane,
 )
 from ..work_items.work_lane import (
+    WORK_LANE_RECEIPT_BOUND_DEFERRED_OBLIGATION,
     work_lane_contract_is_due_monitor_attempt,
     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,
@@ -533,6 +536,15 @@ def _resolve_agent_lane_delivery_route(
         # decision. A newly runnable Todo remains visible in summaries, but it
         # cannot become the selected settlement target in the same packet.
         fallback = None
+    if (
+        prepared.receipt_bound_todo_id
+        and isinstance(fallback, dict)
+        and normalize_todo_id(fallback.get("todo_id"))
+        != prepared.receipt_bound_todo_id
+    ):
+        # Feed only the committed identity into the TS delivery router.  The
+        # independent successor remains discoverable on a fresh Turn.
+        fallback = None
 
     delivery_agent_id = normalize_todo_claimed_by(
         (prepared.agent_identity or {}).get("agent_id")
@@ -617,6 +629,17 @@ def _resolve_agent_lane_delivery_route(
     else:
         selected_action = None
 
+    if (
+        prepared.receipt_bound_todo_id
+        and isinstance(selected_action, dict)
+        and normalize_todo_id(selected_action.get("todo_id"))
+        != prepared.receipt_bound_todo_id
+    ):
+        # The router may find an independent successor after the bound Todo
+        # becomes unavailable.  That successor cannot replace an already
+        # committed settlement identity inside the same heartbeat Turn.
+        return None
+
     boundary = delivery_route.get("boundary")
     if (
         isinstance(selected_action, dict)
@@ -697,6 +720,37 @@ def _planning_projections(
     )
 
 
+def _resolve_external_evidence_observation(
+    prepared: _QuotaDecisionPreparation,
+    *,
+    state: str,
+) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
+    """Suppress unchanged or premature polls before choosing a delivery route."""
+
+    external_evidence_observation = build_external_evidence_observation_obligation(
+        prepared.item,
+        state=state,
+        agent_todo_summary=prepared.agent_todo_summary,
+        work_lane_contract=prepared.work_lane_contract,
+    )
+    external_evidence_observation_recent = None
+    if external_evidence_observation:
+        external_evidence_observation_recent = _recent_external_monitor_observation_unchanged(
+            prepared.status_payload,
+            goal_id=prepared.safe_goal_id,
+            agent_id=(
+                normalize_todo_claimed_by(prepared.agent_identity.get("agent_id"))
+                if isinstance(prepared.agent_identity, dict)
+                else None
+            ),
+        )
+        if external_evidence_observation_recent or (
+            external_evidence_observation.get("poll_window_status") == "before_next_due"
+        ):
+            external_evidence_observation = None
+    return external_evidence_observation, external_evidence_observation_recent
+
+
 def _resolve_quota_should_run_route(
     prepared: _QuotaDecisionPreparation,
 ) -> _QuotaDecisionRoute:
@@ -794,27 +848,10 @@ def _resolve_quota_should_run_route(
         automation_prompt_upgrade_required=prepared.automation_prompt_upgrade_required,
         blocked_priority_fallback=prepared.blocked_priority_fallback,
     )
-    external_evidence_observation = build_external_evidence_observation_obligation(
-        item,
-        state=state,
-        agent_todo_summary=prepared.agent_todo_summary,
-        work_lane_contract=prepared.work_lane_contract,
-    )
-    external_evidence_observation_recent = None
-    if external_evidence_observation:
-        external_evidence_observation_recent = _recent_external_monitor_observation_unchanged(
-            prepared.status_payload,
-            goal_id=prepared.safe_goal_id,
-            agent_id=(
-                normalize_todo_claimed_by(prepared.agent_identity.get("agent_id"))
-                if isinstance(prepared.agent_identity, dict)
-                else None
-            ),
-        )
-        if external_evidence_observation_recent or (
-            external_evidence_observation.get("poll_window_status") == "before_next_due"
-        ):
-            external_evidence_observation = None
+    (
+        external_evidence_observation,
+        external_evidence_observation_recent,
+    ) = _resolve_external_evidence_observation(prepared, state=state)
     ready_deferred_resume_candidates: list[dict[str, Any]] = []
     if isinstance(prepared.agent_identity, dict) and isinstance(
         prepared.agent_todo_summary, dict
@@ -867,6 +904,33 @@ def _resolve_quota_should_run_route(
             "reason": reason,
             "spend_policy": "no quota spend for an already-settled heartbeat turn",
         }
+    receipt_bound_deferred = (
+        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:
+        # 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
+        (
+            reason,
+            quota,
+            heartbeat_recommendation,
+        ) = deferred_receipt_bound_skip_fields(
+            quota,
+            heartbeat_recommendation,
+        )
+        effective_action = EffectiveAction.QUOTA_SKIP.value
     monitor_quiet_skip = (
         not replan_decision_allowed
         and normal_delivery_allowed
@@ -934,7 +998,11 @@ def _resolve_quota_should_run_route(
     )
     agent_scope_frontier = None
     agent_lane_frontier_hint = None
-    if not replan_decision_allowed and not receipt_bound_monitor_settled:
+    if receipt_bound_deferred:
+        # The no-spend route and its public next action must describe the
+        # same committed binding, even when summaries offer independent work.
+        selected_recommended_action = prepared.work_lane_contract["action"]
+    elif not replan_decision_allowed and not receipt_bound_monitor_settled:
         selected_recommended_action = selected_action_with_agent_lane(
             selected_recommended_action,
             agent_lane_next_action=agent_lane_next_action,

loopx/control_plane/quota/should_run_prepare.py

@@ -69,6 +69,7 @@
 )
 from ..todos.contract import (
     TODO_STATUS_BLOCKED,
+    TODO_STATUS_DEFERRED,
     TODO_STATUS_OPEN,
     TODO_TASK_CLASS_ADVANCEMENT,
     TODO_TASK_CLASS_BLOCKER,
@@ -104,6 +105,7 @@
     lark_inbox_reply_due_work_lane_contract,
     operator_inbox_material_review_due_work_lane_contract,
     preserve_heartbeat_receipt_bound_work_lane,
+    receipt_bound_deferred_work_lane,
     scoped_user_gate_due_monitor_contract,
     work_lane_contract_is_lark_inbox_reply_due,
     work_lane_contract_is_operator_inbox_material_review_due,
@@ -450,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],
     *,
@@ -736,12 +752,25 @@ 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
+            )
+        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 = (
+                _deferred_receipt_bound_work_lane(
+                    todo_id=receipt_bound_todo_id,
+                    source_items=agent_todo_planning_source_items,
+                )
+                or work_lane_contract
             )
-            if isinstance(preserved_work_lane, dict):
-                work_lane_contract = preserved_work_lane
     if inbox_priority_due:
         task_orchestration_contract = capability_gate = capability_monitor_contract = None
         capability_monitor_fallback = scoped_user_gate_fallback = workspace_guard = None

loopx/control_plane/work_items/work_lane.py

@@ -39,6 +39,9 @@ def observe_work_lane(
 WORK_LANE_RECEIPT_BOUND_MONITOR_SETTLED_OBLIGATION = (
     "finish_settled_receipt_bound_monitor_turn"
 )
+WORK_LANE_RECEIPT_BOUND_DEFERRED_OBLIGATION = (
+    "wait_for_receipt_bound_deferred_todo"
+)
 WORK_LANE_CURRENT_AGENT_MONITOR_REPAIR_OBLIGATIONS = {
     "attempt_due_monitor",
     "repair_monitor_schedule_metadata",
@@ -278,6 +281,33 @@ def preserve_heartbeat_receipt_bound_work_lane(
     }
 
 
+def receipt_bound_deferred_work_lane(
+    *, todo_id: str,
+) -> dict[str, Any]:
+    """Keep an immutable Turn binding visible without executing a deferred Todo."""
+
+    normalized = normalize_todo_id(todo_id)
+    if not normalized:
+        raise ValueError("receipt-bound deferred work lane requires a Todo id")
+    return {
+        "schema_version": WORK_LANE_CONTRACT_SCHEMA_VERSION,
+        "lane": "advancement_task",
+        "obligation": WORK_LANE_RECEIPT_BOUND_DEFERRED_OBLIGATION,
+        "must_attempt_work": False,
+        "selection_binding": "heartbeat_receipt",
+        "selected_todo_id": normalized,
+        "reason_codes": [
+            "heartbeat_receipt_bound_replay",
+            "receipt_bound_todo_deferred",
+            "successor_requires_fresh_turn",
+        ],
+        "action": (
+            "the Todo bound to this heartbeat turn is deferred; do not execute "
+            "or spend this turn, and select independent work under a fresh turn"
+        ),
+    }
+
+
 def work_lane_contract_is_lark_inbox_reply_due(
     contract: dict[str, Any] | None,
 ) -> bool: