fix(quota): preserve deferred receipt-bound turn identity

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

Author: huangruiteng

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,6 +124,7 @@
     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,
 )
@@ -533,6 +535,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 +628,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)
@@ -867,6 +889,31 @@ def _resolve_quota_should_run_route(
             "reason": reason,
             "spend_policy": "no quota spend for an already-settled heartbeat turn",
         }
+    receipt_bound_deferred_wait = bool(
+        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"
+        )
+        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

@@ -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,
@@ -742,6 +744,19 @@ def _prepare_quota_should_run_item(
             )
             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
+        ):
+            # 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,
+            )
     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:

tests/control_plane/test_quota_settlement_cli.py

@@ -3424,6 +3424,79 @@ def test_pending_deferred_p0_allows_independent_p1_selection(
     assert payload["action_selection_qualification"]["state"] == "qualified"
 
 
+def test_same_turn_bound_p0_does_not_project_p1_after_p0_becomes_deferred(
+    tmp_path: Path,
+) -> None:
+    project, runtime, registry_path = _write_fixture(tmp_path)
+    _configure_selectable_alternative(project)
+    state_path = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
+    state_path.write_text(
+        state_path.read_text(encoding="utf-8").replace(
+            "[P1] Validate and settle the selected delivery.",
+            "[P0] Validate and settle the selected delivery.",
+        ),
+        encoding="utf-8",
+    )
+    turn_id = "turn-bound-p0-then-deferred"
+    guard = (
+        "quota", "should-run", "--codex-app", "--goal-id", GOAL_ID,
+        "--agent-id", AGENT_ID, "--turn-instance-id", turn_id,
+        "--scan-path", str(project),
+    )
+    first_rc, first = _run_cli(registry_path, runtime, *guard, "--todo-id", TODO_ID)
+    assert first_rc == 0, first
+    assert first["heartbeat_receipt"]["settlement_identity"]["todo_id"] == TODO_ID
+    state_path.write_text(
+        state_path.read_text(encoding="utf-8").replace(
+            f"todo_id={TODO_ID} status=open",
+            f"todo_id={TODO_ID} status=deferred "
+            "resume_when=resume_at:2099-01-01T00:00:00Z",
+        ),
+        encoding="utf-8",
+    )
+    replay_rc, replay = _run_cli(registry_path, runtime, *guard)
+    assert replay_rc == 0, replay
+    assert replay["effective_action"] == "quota_skip"
+    assert replay["should_run"] is False
+    assert replay["heartbeat_receipt"]["status"] == "replayed"
+    assert replay["heartbeat_receipt"]["settlement_identity"]["todo_id"] == TODO_ID
+    assert replay["selected_todo"]["todo_id"] == TODO_ID
+    assert replay["work_lane_contract"]["obligation"] == (
+        "wait_for_receipt_bound_deferred_todo"
+    )
+    assert replay["work_lane_contract"]["must_attempt_work"] is False
+    interaction = replay["interaction_contract"]
+    assert interaction["agent_channel"]["must_attempt"] is False
+    assert interaction["cli_channel"]["spend_after_validation"] is False
+    assert ALTERNATIVE_TODO_ID not in json.dumps(
+        interaction["cli_channel"].get("next_cli_actions", [])
+    )
+    plan = interaction["cli_channel"].get("settlement_plan")
+    assert plan is None or plan["identity"]["todo_id"] == TODO_ID
+    assert _heartbeat_receipt_count(runtime, turn_id) == 1
+    assert _spend_run_count(runtime) == 0
+
+    conflict_rc, conflict = _run_cli(
+        registry_path, runtime, *guard, "--todo-id", ALTERNATIVE_TODO_ID,
+    )
+    assert conflict_rc != 0, conflict
+    assert conflict["error_code"] in {
+        "heartbeat_receipt_identity_conflict",
+        "quota_action_selection_rejected",
+    }
+    assert _heartbeat_receipt_count(runtime, turn_id) == 1
+
+    next_rc, next_turn = _run_cli(
+        registry_path, runtime,
+        "quota", "should-run", "--codex-app",
+        "--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
+        "--turn-instance-id", "turn-after-bound-p0-deferred",
+        "--scan-path", str(project), "--todo-id", ALTERNATIVE_TODO_ID,
+    )
+    assert next_rc == 0, next_turn
+    assert next_turn["selected_todo"]["todo_id"] == ALTERNATIVE_TODO_ID
+
+
 def test_pending_selection_preserves_workspace_repair_then_reenters_same_turn(
     tmp_path: Path,
 ) -> None: