fix(quota): resume verified replan settlement before stopping

loopx/control_plane/quota/decision_summary.py

@@ -12,6 +12,7 @@
     goal_frontier_is_terminal_no_followup,
 )
 from ..effect_runtime import effect_runtime_result
+from .effect_program import ReceiptBoundReplayPhase
 from ..todos.contract import normalize_todo_claimed_by
 from ..work_items.work_lane import work_lane_contract_is_due_monitor_attempt
 
@@ -263,10 +264,12 @@ def resolve_quota_run_decision(
     registered_agent_ids: list[str],
     goal_frontier_projection: dict[str, Any] | None,
     task_orchestration_contract: dict[str, Any] | None,
+    receipt_bound_replan_phase: ReceiptBoundReplayPhase | None = None,
 ) -> QuotaRunDecision:
     """Apply the ordered final guards for one quota run decision."""
 
     replan_required = autonomous_replan_is_required(replan_obligation)
+    settlement_packet = (replan_obligation or {}).get("settlement_action_packet")
     # Scope evaluation can synthesize a legacy identity; preserve its lazy read boundary.
     scope = autonomous_replan_scope_decision(
         replan_obligation, agent_id=agent_frontier_id,
@@ -294,6 +297,13 @@ def resolve_quota_run_decision(
         "automation_prompt_upgrade_required": automation_prompt_upgrade_required,
         "replan_required": replan_required,
         "replan_applies": bool(scope.get("applies")),
+        "replan_settlement_pending": (
+            receipt_bound_replan_phase is ReceiptBoundReplayPhase.SETTLEMENT_PENDING
+            and (replan_obligation or {}).get("selection_binding") == "heartbeat_receipt"
+            and isinstance(settlement_packet, dict)
+            and settlement_packet.get("settlement_only") is True
+            and settlement_packet.get("decision") == receipt_bound_replan_phase.value
+        ),
         "goal_health_ok": goal_health_ok,
         "inbox_reply_due": inbox_reply_due,
         "inbox_material_review_due": inbox_material_review_due,

loopx/control_plane/quota/live_decision.py

@@ -69,12 +69,32 @@ def _receipt_bound_guard_fields(
 
 
 def _historical_ack_writeback_required(payload: Mapping[str, Any]) -> bool:
-    """Dispatch the typed historical ACK without hiding pending current closeout."""
+    """Recognize the exact historical ACK closeout before durable writeback."""
     packet = payload.get("replan_action_packet")
     return (isinstance(packet, Mapping) and packet.get("settlement_only") is True
             and packet.get("decision") == "historical_ack_recorded")
 
 
+def _original_replan_settlement_required(
+    payload: Mapping[str, Any], readback: QuotaSettlementReadback | None,
+) -> bool:
+    if _historical_ack_writeback_required(payload):
+        return True
+    if readback is None or readback.replay_phase is not ReceiptBoundReplayPhase.SETTLEMENT_PENDING:
+        return False
+    packet = payload.get("replan_action_packet")
+    identity = readback.identity.value
+    return (
+        isinstance(packet, Mapping) and packet.get("settlement_only") is True
+        and packet.get("decision") == readback.replay_phase.value
+        and identity is not None and identity.replan_obligation_id is not None
+        and identity.replan_obligation_id == packet.get("obligation_id")
+        and (readback.progress or {}).get("next_step") == "quota_spend"
+        and payload.get("should_run") is True
+        and payload.get("effective_action") == EffectiveAction.AUTONOMOUS_REPLAN_REQUIRED.value
+    )
+
+
 def _project_receipt_replan_settlement(
     payload: dict[str, Any], readback: QuotaSettlementReadback | None, *,
     available_capabilities: list[str] | None,
@@ -83,7 +103,7 @@ def _project_receipt_replan_settlement(
     goal_ref: Mapping[str, Any] | None,
 ) -> None:
     """Project closeout from the original readback, never the current selection."""
-    if not readback or not _historical_ack_writeback_required(payload):
+    if not readback or not _original_replan_settlement_required(payload, readback):
         return
     attach_settlement_progress(payload, readback, registry_path=registry_path,
         runtime_root=runtime_root, goal_ref=goal_ref)
@@ -871,7 +891,7 @@ def build_live_quota_should_run_decision(
     # unsettled Turn here can overwrite the settled-skip route with a recovery
     # obligation and then select a successor against the immutable receipt
     # identity.  Leave prior-Turn recovery to the next fresh Turn instead.
-    original_replan_settlement = _historical_ack_writeback_required(payload)
+    original_replan_settlement = _original_replan_settlement_required(payload, settlement_readback)
     if receipt_bound_replay_phase is not ReceiptBoundReplayPhase.SETTLED and not original_replan_settlement:
         apply_unsettled_host_turn_recovery_if_required(
             payload,

loopx/control_plane/quota/run_decision.ts

@@ -18,6 +18,7 @@ interface RunFacts {
   automation_prompt_upgrade_required: boolean;
   replan_required: boolean;
   replan_applies: boolean;
+  replan_settlement_pending: boolean;
   goal_health_ok: boolean;
   inbox_reply_due: boolean;
   inbox_material_review_due: boolean;
@@ -69,6 +70,8 @@ function decodeRunFacts(params: JsonObject): RunFacts {
     automation_prompt_upgrade_required: requireBoolean(params.automation_prompt_upgrade_required, "automation_prompt_upgrade_required"),
     replan_required: requireBoolean(params.replan_required, "replan_required"),
     replan_applies: requireBoolean(params.replan_applies, "replan_applies"),
+    replan_settlement_pending: params.replan_settlement_pending === undefined ? false
+      : requireBoolean(params.replan_settlement_pending, "replan_settlement_pending"),
     goal_health_ok: requireBoolean(params.goal_health_ok, "goal_health_ok"),
     inbox_reply_due: requireBoolean(params.inbox_reply_due, "inbox_reply_due"),
     inbox_material_review_due: requireBoolean(params.inbox_material_review_due, "inbox_material_review_due"),
@@ -139,6 +142,15 @@ function applyGoalOverrides(facts: RunFacts, decision: RunDecision): void {
   decision.replan_decision_allowed = !inboxPriority && facts.replan_required && facts.replan_applies
     && facts.goal_health_ok && !facts.workspace_guard && !facts.automation_prompt_upgrade_required;
   if (decision.replan_decision_allowed) {
+    if (facts.replan_settlement_pending) {
+      // A covered frontier can stop new work before its original receipt has
+      // settled. Finish that receipt before applying derived terminal stop.
+      clearExecutionLanes(decision);
+      decision.should_run = true;
+      decision.effective_action = EffectiveAction.AUTONOMOUS_REPLAN_REQUIRED;
+      decision.reason = "verified replan writeback requires settlement for the original Turn";
+      return;
+    }
     decision.normal_delivery_allowed = false;
     decision.recovery_delivery_allowed = false;
     decision.should_run = true;

loopx/control_plane/quota/should_run_packet.py

@@ -935,6 +935,11 @@ def _resolve_quota_should_run_route(
         registered_agent_ids=prepared.registered_agent_ids,
         goal_frontier_projection=prepared.goal_frontier_projection,
         task_orchestration_contract=prepared.task_orchestration_contract,
+        receipt_bound_replan_phase=(
+            prepared.receipt_bound_replay_phase
+            if prepared.receipt_bound_todo_id is None
+            else None
+        ),
     )
     if prepared.receipt_bound_replay_phase is ReceiptBoundReplayPhase.SETTLED:
         fields = settled_replay_fields()

loopx/control_plane/work_items/interaction_contract.py

@@ -447,8 +447,10 @@ def _turn_scoped_cli_settlement_context(
     agent_id = str(agent_identity.get("agent_id") or "").strip()
     selected_todo_id = normalize_todo_id(selected_todo.get("todo_id"))
     receipt = payload.get("heartbeat_receipt")
-    if ((payload.get("replan_action_packet") or {}).get("settlement_only") is True
-            and isinstance(receipt, Mapping) and receipt.get("settlement_identity") is not None):
+    replan_packet = payload.get("replan_action_packet")
+    replan_settlement_only = isinstance(replan_packet, Mapping) and replan_packet.get("settlement_only") is True
+    if (replan_settlement_only and isinstance(receipt, Mapping)
+            and receipt.get("settlement_identity") is not None):
         identity = SettlementIdentity.from_runtime_payload(receipt["settlement_identity"])
         if (identity.goal_id, identity.agent_id, identity.turn_instance_id) != (goal_id, agent_id, turn_instance_id):
             raise ValueError("receipt settlement projection does not match the original Goal/Agent/Turn")
@@ -504,10 +506,20 @@ def _turn_scoped_cli_settlement_context(
             else None
         ),
     )
-    return (
-        plan.as_dict() if plan is not None else None,
-        replan_settlement_contract,
-    )
+    if plan is None:
+        return None, replan_settlement_contract
+    cli_plan = plan.as_dict()
+    owed = payload.get("settlement_owed")
+    if (replan_settlement_only and isinstance(owed, Mapping)
+            and owed.get("effect_id") == plan.identity.effect_id
+            and owed.get("kind") == SettlementStepKind.QUOTA_SPEND.value):
+        # Verified receipt progress already owns the remaining step. Keep
+        # its original command and any conditional terminal closeout.
+        cli_plan["ordered_steps"] = [
+            step for step in cli_plan["ordered_steps"]
+            if step["kind"] not in {SettlementStepKind.VALIDATION.value, SettlementStepKind.DURABLE_WRITEBACK.value}
+        ]
+    return cli_plan, replan_settlement_contract
 
 
 def _terminal_cli_actions(

tests/control_plane/test_replan_successor_guard_reentry.py

@@ -306,7 +306,11 @@ def test_successor_guard_returns_original_settlement_not_repeated_planning(
     assert packet["obligation_id"] == obligation_id
     assert packet["successor_todo_id"] == added["todo_id"]
     assert packet["writeback_contract"]["preferred_input"] == "canonical_successor_transition"
-    actions = reread["interaction_contract"]["cli_channel"]["next_cli_actions"]
+    cli = reread["interaction_contract"]["cli_channel"]
+    assert [step["kind"] for step in cli["settlement_plan"]["ordered_steps"]] == [
+        "validation", "durable_writeback", "quota_spend",
+    ]
+    actions = cli["next_cli_actions"]
     assert len(actions) == 2
     assert all("todo add" not in command for command in actions)
     assert all("--replan-obligation-id " + obligation_id in command for command in actions)
@@ -330,11 +334,18 @@ def test_successor_guard_returns_original_settlement_not_repeated_planning(
     pending = _guard(call)
     assert pending["heartbeat_receipt"]["settlement_identity"] == identity
     assert pending["settlement_progress"]["state"] == "spend_required"
-    spend_actions = pending["interaction_contract"]["cli_channel"]["next_cli_actions"]
+    assert pending["settlement_owed"]["effect_id"] == identity["effect_id"]
+    pending_cli = pending["interaction_contract"]["cli_channel"]
+    spend_actions = pending_cli["next_cli_actions"]
     assert spend_actions == [actions[1]]
+    assert pending["settlement_owed"]["command"] == actions[1]
+    pending_plan = pending_cli["settlement_plan"]
+    assert pending_plan["identity"] == identity
+    assert pending_plan["ordered_steps"] == [cli["settlement_plan"]["ordered_steps"][-1]]
     pending_envelope = call("quota", "should-run", "--codex-app", "--goal-id", GOAL,
         "--agent-id", AGENT, "--turn-instance-id", TURN, "--turn-envelope")
     assert pending_envelope["writeback"]["next_cli_actions"] == spend_actions
+    assert pending_envelope["writeback"]["settlement_plan"] == pending_plan
     assert pending_envelope["action_signature"]["matches"] is True
     spent = call(*shlex.split(spend_actions[0])[1:])
     assert spent["settlement_progress"]["state"] == "settled"
@@ -459,7 +470,15 @@ def test_later_vision_replan_can_close_covered_frontier_and_settle_once(
     assert pending["heartbeat_receipt"]["settlement_identity"] == identity
     assert pending["settlement_progress"]["state"] == "spend_required"
     assert pending.get("selected_todo") is None
-    spent = call("quota", "spend-slot", *binding, "--slots", "1", "--source", "heartbeat", "--execute")
+    assert pending["should_run"] is True
+    assert pending["normal_delivery_allowed"] is False
+    cli = pending["interaction_contract"]["cli_channel"]
+    plan = cli["settlement_plan"]
+    assert plan["identity"] == identity
+    assert [step["kind"] for step in plan["ordered_steps"]] == ["quota_spend"]
+    command = pending["settlement_owed"]["command"]
+    assert cli["next_cli_actions"] == [command]
+    spent = call(*shlex.split(command)[1:])
     assert spent["appended"] is True
     assert spent["settlement_progress"]["state"] == "settled"
     assert call("quota", "spend-slot", *binding, "--slots", "1", "--source", "heartbeat", "--execute")["appended"] is False