From 58122e6b93c04d7b558626ece504274a65df87e6 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Thu, 24 Sep 2026 21:53:33 +0800
Subject: [PATCH 1/8] fix(quota): close typed blocked Turns without debit

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 loopx/control_plane/quota/settlement.py       |  2 ++
 loopx/control_plane/quota/settlement_phase.ts |  5 +++-
 .../quota/settlement_readback.ts              | 27 ++++++++++++++++---
 loopx/control_plane/quota/slot_accounting.py  | 15 +++++++++++
 .../quota/unsettled_host_turn_recovery.ts     | 10 ++++++-
 5 files changed, 53 insertions(+), 6 deletions(-)

diff --git a/loopx/control_plane/quota/settlement.py b/loopx/control_plane/quota/settlement.py
index b412095b65..03e045f8c1 100644
--- a/loopx/control_plane/quota/settlement.py
+++ b/loopx/control_plane/quota/settlement.py
@@ -214,6 +214,8 @@ def render_settlement_progress_markdown(payload: dict[str, Any]) -> list[str]:
     if not isinstance(progress, dict):
         return []
     lines = [f"- settlement: `{progress.get('state')}`"]
+    if progress.get("closeout_kind") == "typed_blocked_writeback_no_spend":
+        lines.append("- closeout: typed blocked writeback; no quota slot spent")
     owed = payload.get("settlement_owed")
     if isinstance(owed, dict):
         lines.extend([f"- settlement_owed: {owed['reason']}", "", "```sh", owed["command"], "```"])
diff --git a/loopx/control_plane/quota/settlement_phase.ts b/loopx/control_plane/quota/settlement_phase.ts
index fe4c834280..01058521a1 100644
--- a/loopx/control_plane/quota/settlement_phase.ts
+++ b/loopx/control_plane/quota/settlement_phase.ts
@@ -53,6 +53,8 @@ export interface ReceiptBoundReplaySettlementState {
   completion_receipt_present: boolean;
   durable_writeback_present: boolean;
   quota_spend_present: boolean;
+  /** Exact typed blocked writeback closes a Turn without a quota debit. */
+  no_spend_closeout_present?: boolean;
 }
 
 export function receiptBoundReplayPhase(
@@ -63,7 +65,8 @@ export function receiptBoundReplayPhase(
     ? state.durable_writeback_present
     : state.completion_receipt_present;
   if (!bindingComplete) return "open";
-  return state.durable_writeback_present && state.quota_spend_present
+  return state.durable_writeback_present &&
+      (state.quota_spend_present || state.no_spend_closeout_present === true)
     ? "settled"
     : "settlement_pending";
 }
diff --git a/loopx/control_plane/quota/settlement_readback.ts b/loopx/control_plane/quota/settlement_readback.ts
index 646912dddc..9682d2f017 100644
--- a/loopx/control_plane/quota/settlement_readback.ts
+++ b/loopx/control_plane/quota/settlement_readback.ts
@@ -113,6 +113,7 @@ function settlementProgress(
   identity: SettlementResult, writeback: SettlementResult, spend: SettlementResult,
   writebackRun: JsonObject | null, spendRun: JsonObject | null,
   spendSource: unknown = "heartbeat",
+  blockedNoSpend = false,
 ): JsonObject {
   const source = spendSource ?? "heartbeat";
   if (source !== "heartbeat" && source !== "visible-goal") {
@@ -120,13 +121,17 @@ function settlementProgress(
   }
   const state: SettlementProgressState = identity.failure ? "identity_required"
     : writeback.failure ? (writebackRun ? "writeback_receipt_required" : "writeback_required")
+    : blockedNoSpend ? "settled"
     : spend.failure ? (spendRun ? "spend_receipt_required" : "spend_required")
     : "settled";
   return {
     schema_version: "quota_settlement_progress_v0", state,
     next_step: identity.failure ? "validation" : writeback.failure ? "durable_writeback"
-      : spend.failure ? "quota_spend" : null,
+      : blockedNoSpend ? null : spend.failure ? "quota_spend" : null,
     quota_spend_source: source,
+    ...(blockedNoSpend ? {
+      closeout_kind: "typed_blocked_writeback_no_spend",
+    } : {}),
   };
 }
 
@@ -966,9 +971,22 @@ function readQuotaSettlementFromRequest(
 
   const writeback = writebackResult(identity, writebackRun, writebackEvent);
   const spend = spendResult(identity, spendRun, spendEvent);
+  // The exact Turn-bound blocked writeback is itself a durable no-spend
+  // closeout. It cannot certify Todo completion or become delivery progress.
+  // A spend already committed for this identity remains an ordinary spend
+  // settlement, so readback never erases a historical debit.
+  const blockedNoSpend = writeback.failure === null &&
+    spendRun === null && spendEvent === null &&
+    identity.binding_kind === "todo" &&
+    writebackRun?.delivery_outcome === "outcome_gap" &&
+    isTurnScopedSettlementOutcome(
+      writebackRun.delivery_outcome,
+      writebackRun.progress_observation,
+      identity.todo_id,
+    );
   const terminalCloseout = terminalResult(identity, completionEvent);
   const withWriteback = settlementBindReduce(identityResult, writeback);
-  const settled = settlementBindReduce(withWriteback, spend);
+  const settled = blockedNoSpend ? withWriteback : settlementBindReduce(withWriteback, spend);
   const terminalSettlement = settlementBindReduce(settled, terminalCloseout);
   const monitorPoll = [...runs].reverse().find((run) =>
     run.classification === "quota_monitor_poll" &&
@@ -1020,7 +1038,7 @@ function readQuotaSettlementFromRequest(
     terminal_closeout: bundle(terminalCloseout),
     terminal_settlement: bundle(terminalSettlement),
     progress: settlementProgress(identityResult, writeback, spend, writebackRun, spendRun,
-      receiptDetails.quota_spend_source ?? spendRun?.source),
+      receiptDetails.quota_spend_source ?? spendRun?.source, blockedNoSpend),
     workspace_causality: workspaceCausality,
     semantic_replan_guard: semanticReplanGuard,
     writeback_run: writebackRun,
@@ -1042,10 +1060,11 @@ function readQuotaSettlementFromRequest(
     }),
     replay_phase: receiptBoundReplayPhase({
       binding_kind: identity.binding_kind,
-      writeback_completes_binding: todoBoundReplan,
+      writeback_completes_binding: todoBoundReplan || blockedNoSpend,
       completion_receipt_present: completionEvent !== null,
       durable_writeback_present: writeback.failure === null,
       quota_spend_present: spend.failure === null,
+      no_spend_closeout_present: blockedNoSpend,
     }),
   };
 }
diff --git a/loopx/control_plane/quota/slot_accounting.py b/loopx/control_plane/quota/slot_accounting.py
index 8d0d6f265f..eeacd97b25 100644
--- a/loopx/control_plane/quota/slot_accounting.py
+++ b/loopx/control_plane/quota/slot_accounting.py
@@ -179,6 +179,21 @@ def _resolve_preview_settlement(
     )
     if readback is None:
         return {}
+    if (
+        isinstance(readback.progress, dict)
+        and readback.progress.get("closeout_kind")
+        == "typed_blocked_writeback_no_spend"
+    ):
+        return {
+            "identity": readback.identity.value,
+            "result": readback.settlement,
+            "delivery_run": readback.writeback_run,
+            "reason": (
+                "this Turn already closed with an exact typed blocked writeback "
+                "and must not consume a quota slot; retry the Todo only after "
+                "its external blocker changes or a bounded backoff"
+            ),
+        }
     result = readback.identity
     identity = result.value if result.failure is None else None
     if identity is not None:
diff --git a/loopx/control_plane/quota/unsettled_host_turn_recovery.ts b/loopx/control_plane/quota/unsettled_host_turn_recovery.ts
index 63e03c6c6f..e03cb1b8f1 100644
--- a/loopx/control_plane/quota/unsettled_host_turn_recovery.ts
+++ b/loopx/control_plane/quota/unsettled_host_turn_recovery.ts
@@ -61,6 +61,7 @@ const MISSING_RECEIPT_NAMES = [WRITEBACK_RECEIPT, SPEND_RECEIPT] as const;
 
 export const ACCEPTED_CLOSEOUTS = [
   "validated_writeback_and_quota_spend",
+  "typed_blocked_writeback_no_spend",
   "exact_committed_quota_monitor_poll",
   "typed_external_wait_with_runnable_successor",
   "typed_blocker_or_lifecycle_transition",
@@ -240,6 +241,7 @@ export async function preflightPriorHostTurnCloseout(
     rolloutSnapshot,
   );
   let newestSettledTurn: string | null = null;
+  let newestAcceptedCloseout: AcceptedCloseout = "validated_writeback_and_quota_spend";
   for (const selected of candidates) {
     const readback = readQuotaSettlementFromSnapshot(
       settlementReadbackRequest(request, selected),
@@ -250,6 +252,12 @@ export async function preflightPriorHostTurnCloseout(
       // scanning in persisted newest-first order until the first unsettled
       // candidate is found.
       newestSettledTurn ??= selected.prior_turn_instance_id;
+      if (newestSettledTurn === selected.prior_turn_instance_id) {
+        const progress = jsonObject(readback.progress);
+        if (progress?.closeout_kind === "typed_blocked_writeback_no_spend") {
+          newestAcceptedCloseout = "typed_blocked_writeback_no_spend";
+        }
+      }
       continue;
     }
     const missingReceipts: string[] = [];
@@ -273,7 +281,7 @@ export async function preflightPriorHostTurnCloseout(
     reason: "prior_turn_settlement_validated",
     turns_validated: turnsValidated,
     prior_turn_instance_id: newestSettledTurn,
-    accepted_closeout: "validated_writeback_and_quota_spend",
+    accepted_closeout: newestAcceptedCloseout,
   };
 }
 

From 26976678f23f84456a1aec89a3e746fe1806b176 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Thu, 24 Sep 2026 21:53:44 +0800
Subject: [PATCH 2/8] test(quota): cover blocked no-spend closeout and replay

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 .../test_quota_settlement_cli.py              | 36 ++++++++++--
 .../quota_settlement_readback.test.ts         | 55 +++++++++++++++++++
 .../unsettled_host_turn_recovery.test.ts      | 50 +++++++++++++++++
 3 files changed, 136 insertions(+), 5 deletions(-)

diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py
index c8226c3eac..68c1ae4096 100644
--- a/tests/control_plane/test_quota_settlement_cli.py
+++ b/tests/control_plane/test_quota_settlement_cli.py
@@ -22,6 +22,7 @@
 from loopx.control_plane.quota.settlement_validation import (
     completion_validation_spend_error,
 )
+from loopx.control_plane.quota.settlement import render_settlement_progress_markdown
 from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos
 from loopx.heartbeat_prompt import build_heartbeat_prompt
 from loopx.rollout_event_log import build_rollout_event
@@ -898,6 +899,7 @@ def test_typed_outcome_gap_settles_exact_turn_without_becoming_progress(
         tmp_path,
         required_capability="filesystem_write",
     )
+    _configure_completion_validation_todo(project)
     turn_id = "turn-typed-blocker-settlement"
     binding = (
         "--agent-id",
@@ -1004,6 +1006,14 @@ def test_typed_outcome_gap_settles_exact_turn_without_becoming_progress(
         receipt["step_kind"]
         for receipt in refresh["settlement_result"]["receipts"]
     ] == ["validation", "durable_writeback"]
+    assert refresh["settlement_progress"]["state"] == "settled"
+    assert refresh["settlement_progress"]["closeout_kind"] == (
+        "typed_blocked_writeback_no_spend"
+    )
+    assert refresh.get("settlement_owed") is None
+    assert "- closeout: typed blocked writeback; no quota slot spent" in (
+        render_settlement_progress_markdown(refresh)
+    )
 
     spend_rc, spend = _run_cli(
         registry_path,
@@ -1023,11 +1033,27 @@ def test_typed_outcome_gap_settles_exact_turn_without_becoming_progress(
         cwd=project,
     )
     assert spend_rc == 0, spend
-    assert [
-        receipt["step_kind"]
-        for receipt in spend["settlement_result"]["receipts"]
-    ] == ["validation", "durable_writeback", "quota_spend"]
-    assert _spend_run_count(runtime) == 1
+    assert spend["appended"] is False
+    assert _spend_run_count(runtime) == 0
+
+    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-typed-blocker-settlement",
+        "--scan-path",
+        str(project),
+        cwd=project,
+    )
+    assert next_rc == 0, next_turn
+    assert next_turn["effective_action"] != "unsettled_host_turn_recovery"
 
 
 def test_in_flight_progress_preserves_todo_across_heartbeat_settlements(
diff --git a/tests/control_plane_ts/quota_settlement_readback.test.ts b/tests/control_plane_ts/quota_settlement_readback.test.ts
index 13bfdc5455..dc7af1cba3 100644
--- a/tests/control_plane_ts/quota_settlement_readback.test.ts
+++ b/tests/control_plane_ts/quota_settlement_readback.test.ts
@@ -695,6 +695,59 @@ test("accepts only an attributable typed blocker as an outcome-gap writeback", a
   const qualified = await readQuotaSettlement(request(qualifiedRuntime));
   assert.equal((qualified.writeback as any).payload.ok, true);
   assert.equal((qualified.writeback_run as any).delivery_outcome, "outcome_gap");
+  assert.equal((qualified.settlement as any).payload.ok, true);
+  assert.equal((qualified.spend as any).payload.ok, false);
+  assert.deepEqual((qualified.settlement as any).result.receipts.map(
+    (receipt: any) => receipt.step_kind), ["validation", "durable_writeback"]);
+  assert.equal((qualified.progress as any).state, "settled");
+  assert.equal((qualified.progress as any).next_step, null);
+  assert.equal((qualified.progress as any).closeout_kind,
+    "typed_blocked_writeback_no_spend");
+  assert.equal(qualified.replay_phase, "settled");
+
+  const spentRuntime = await fixture({
+    writeback: true,
+    spend: true,
+    writebackOutcome: "outcome_gap",
+    progressObservation: {
+      schema_version: "typed_progress_observation_v0",
+      result_class: "blocked",
+      work_item_id: todoId,
+      blocker_id: "blocker-runtime-boundary",
+      evidence_ids: ["evidence-runtime-boundary"],
+    },
+  });
+  const spent = await readQuotaSettlement(request(spentRuntime));
+  assert.equal((spent.spend as any).payload.ok, true);
+  assert.equal((spent.progress as any).closeout_kind, undefined);
+  assert.deepEqual((spent.settlement as any).result.receipts.map(
+    (receipt: any) => receipt.step_kind),
+    ["validation", "durable_writeback", "quota_spend"]);
+
+  const incompleteSpendRuntime = await fixture({
+    writeback: true,
+    writebackOutcome: "outcome_gap",
+    progressObservation: {
+      schema_version: "typed_progress_observation_v0",
+      result_class: "blocked",
+      work_item_id: todoId,
+      blocker_id: "blocker-runtime-boundary",
+      evidence_ids: ["evidence-runtime-boundary"],
+    },
+  });
+  await appendFile(join(incompleteSpendRuntime, "goals", goalId,
+    "rollout-event-log.jsonl"), `${JSON.stringify({
+      schema_version: "loopx_rollout_event_v0",
+      event_id: "event-incomplete-spend",
+      event_kind: "quota_spend",
+      goal_id: goalId,
+      agent_id: agentId,
+      run_id: turnId,
+      details: {settlement_effect_id: identity.effect_id},
+    })}\n`);
+  const incompleteSpend = await readQuotaSettlement(request(incompleteSpendRuntime));
+  assert.equal((incompleteSpend.settlement as any).payload.ok, false);
+  assert.equal((incompleteSpend.progress as any).closeout_kind, undefined);
 
   const bareRuntime = await fixture({
     writeback: true,
@@ -703,6 +756,7 @@ test("accepts only an attributable typed blocker as an outcome-gap writeback", a
   const bare = await readQuotaSettlement(request(bareRuntime));
   assert.equal((bare.writeback as any).payload.ok, false);
   assert.equal((bare.writeback as any).result.failure.kind, "writeback_missing");
+  assert.equal((bare.settlement as any).payload.ok, false);
 
   const mismatchedRuntime = await fixture({
     writeback: true,
@@ -717,6 +771,7 @@ test("accepts only an attributable typed blocker as an outcome-gap writeback", a
   });
   const mismatched = await readQuotaSettlement(request(mismatchedRuntime));
   assert.equal((mismatched.writeback as any).payload.ok, false);
+  assert.equal((mismatched.settlement as any).payload.ok, false);
 
   for (const evidenceIds of [
     "evidence-runtime-boundary",
diff --git a/tests/control_plane_ts/unsettled_host_turn_recovery.test.ts b/tests/control_plane_ts/unsettled_host_turn_recovery.test.ts
index 82657f4f0a..445b72422a 100644
--- a/tests/control_plane_ts/unsettled_host_turn_recovery.test.ts
+++ b/tests/control_plane_ts/unsettled_host_turn_recovery.test.ts
@@ -365,6 +365,56 @@ test("a prior Turn that already validates its settlement needs no bound facts",
   // end-to-end through the real entrypoint.
 });
 
+test("a prior typed blocked writeback closes without a quota debit", async () => {
+  const turn = "turn-blocked";
+  const todoId = "todo_blocked";
+  const identity = settlementIdentity({
+    goal_id: GOAL, agent_id: AGENT, todo_id: todoId, turn_instance_id: turn,
+  });
+  const runtime = await runtimeWith([
+    receipt(turn, {
+      todo_id: todoId,
+      settlement_effect_id: identity.effect_id,
+      closeout_required: true,
+    }),
+    {
+      schema_version: "loopx_rollout_event_v0",
+      event_id: "event-blocked-writeback",
+      event_kind: "refresh_state",
+      goal_id: GOAL,
+      agent_id: AGENT,
+      run_id: turn,
+      details: {settlement_effect_id: identity.effect_id},
+    },
+  ]);
+  try {
+    const runsRoot = join(runtime.root, "goals", GOAL, "runs");
+    await mkdir(runsRoot, {recursive: true});
+    await writeFile(join(runsRoot, "index.jsonl"), `${JSON.stringify({
+      classification: "state_refreshed",
+      delivery_outcome: "outcome_gap",
+      goal_id: GOAL,
+      agent_id: AGENT,
+      todo_id: todoId,
+      turn_instance_id: turn,
+      settlement_identity: identity,
+      progress_observation: {
+        schema_version: "typed_progress_observation_v0",
+        result_class: "blocked",
+        work_item_id: todoId,
+        blocker_id: "blocker-lease",
+        evidence_ids: ["evidence-lease"],
+      },
+    })}\n`);
+    const result = await preflight(runtime.root);
+    assert.equal(result.status, "none");
+    assert.equal(result.accepted_closeout, "typed_blocked_writeback_no_spend");
+    assert.equal(result.prior_turn_instance_id, turn);
+  } finally {
+    await runtime.close();
+  }
+});
+
 test("a malformed or foreign log line fails the read instead of erasing a closeout", async () => {
   const runtime = await runtimeWith([
     receipt("turn-a", closeoutRequired("turn-a", "todo_alpha")),

From 3dfc3b645541e95251a9c97ec2453b587cb50951 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Fri, 25 Sep 2026 00:17:53 +0800
Subject: [PATCH 3/8] fix(quota): defer blocked Turns with receipt-owned
 retries

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 .../control_plane/goals/vision_checkpoint.py  |   2 +
 .../control_plane/goals/vision_checkpoint.ts  |  15 +-
 loopx/control_plane/quota/blocked_retry.py    | 296 ++++++++++++++++++
 loopx/control_plane/quota/blocked_retry.ts    |  15 +
 loopx/control_plane/quota/live_decision.py    |   8 +
 .../control_plane/quota/projection_repair.py  |  33 +-
 .../quota/settlement_readback.ts              |   4 +-
 loopx/control_plane/runtime/run_compaction.py |   1 +
 .../runtime/run_context_retention.py          |  27 +-
 loopx/state_refresh.py                        |  35 +++
 10 files changed, 431 insertions(+), 5 deletions(-)
 create mode 100644 loopx/control_plane/quota/blocked_retry.py
 create mode 100644 loopx/control_plane/quota/blocked_retry.ts

diff --git a/loopx/control_plane/goals/vision_checkpoint.py b/loopx/control_plane/goals/vision_checkpoint.py
index 5e7eb6d512..82c338ddb4 100644
--- a/loopx/control_plane/goals/vision_checkpoint.py
+++ b/loopx/control_plane/goals/vision_checkpoint.py
@@ -127,6 +127,7 @@ def build_vision_checkpoint(
     todo_id: str | None = None,
     completion_todo_id: str | None = None,
     autonomous_replan_recorded: bool = False,
+    blocked_retry: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Finalize the TS-owned Vision transaction after replan qualification."""
 
@@ -149,6 +150,7 @@ def build_vision_checkpoint(
                 "todo_id": todo_id,
                 "completion_todo_id": completion_todo_id,
                 "autonomous_replan_recorded": bool(autonomous_replan_recorded),
+                "blocked_retry": blocked_retry,
             },
         )
     except EffectRuntimeRejected as exc:
diff --git a/loopx/control_plane/goals/vision_checkpoint.ts b/loopx/control_plane/goals/vision_checkpoint.ts
index f702ff7c73..779971caf8 100644
--- a/loopx/control_plane/goals/vision_checkpoint.ts
+++ b/loopx/control_plane/goals/vision_checkpoint.ts
@@ -1,5 +1,6 @@
 import type { JsonObject } from "../effect_program.ts";
 import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
+import { isBoundedBlockedRetry } from "../quota/blocked_retry.ts";
 import {
   DELIVERY_BOUNDARIES,
   type DeliveryBoundary,
@@ -148,6 +149,7 @@ interface VisionRefreshFinalizeRequest {
   todo_id: string | null;
   completion_todo_id: string | null;
   autonomous_replan_recorded: boolean;
+  blocked_retry: JsonObject | null;
 }
 
 export type VisionCheckpointDecision =
@@ -721,6 +723,7 @@ export function decodeVisionCheckpointRequest(
       request.autonomous_replan_recorded,
       "autonomous_replan_recorded",
     ),
+    blocked_retry: optionalObject(request.blocked_retry, "blocked_retry"),
   };
 }
 
@@ -763,10 +766,20 @@ export function buildVisionCheckpoint(value: unknown): JsonObject {
   }
   const request = decodeVisionCheckpointRequest(value);
   validateInFlightBoundary(request);
+  if (request.blocked_retry !== null && (
+    request.delivery_outcome !== "outcome_gap" ||
+    request.delivery_boundary !== "semantic_closeout" ||
+    request.todo_id === null ||
+    request.completion_todo_id !== null ||
+    !isBoundedBlockedRetry(request.blocked_retry, request.todo_id)
+  )) {
+    throw new EffectRuntimeRequestError("blocked retry does not bind a typed outcome-gap Todo closeout");
+  }
   const triggers: JsonObject[] = [];
   if (
     isMaterialDeliveryOutcome(request.delivery_outcome) &&
-    request.delivery_boundary === "semantic_closeout"
+    request.delivery_boundary === "semantic_closeout" &&
+    request.blocked_retry === null
   ) {
     triggers.push({
       kind: "material_delivery_outcome",
diff --git a/loopx/control_plane/quota/blocked_retry.py b/loopx/control_plane/quota/blocked_retry.py
new file mode 100644
index 0000000000..ea304be39b
--- /dev/null
+++ b/loopx/control_plane/quota/blocked_retry.py
@@ -0,0 +1,296 @@
+"""Bound a typed blocked Turn's no-spend closeout to a durable retry."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+from typing import Any
+
+from ..todos.contract import normalize_todo_id, normalize_todo_resume_when
+
+BLOCKED_RETRY_SCHEMA_VERSION = "quota_blocked_retry_v0"
+MIN_RETRY_SECONDS = 60
+MAX_RETRY_SECONDS = 30 * 60
+TURN_SETTLEMENT_RETRY_SECONDS = 5 * 60
+
+
+def require_blocked_retry_wait(
+    todo_fields: dict[str, Any] | None,
+    *,
+    todo_id: str,
+    observed_at: str,
+    allow_turn_settlement_retry: bool = False,
+) -> dict[str, Any]:
+    """Require a bounded Todo wait or mint a canonical Turn-owned retry.
+
+    A peer lease can prevent the blocked agent from updating the Todo. In that
+    case the committed Turn owns a five-minute wait and selection projects it
+    without changing the canonical Todo or its validator.
+    """
+
+    summary = (todo_fields or {}).get("agent_todos")
+    items = summary.get("items") if isinstance(summary, dict) else None
+    todo = (
+        next(
+            (
+                item
+                for item in items
+                if isinstance(item, dict)
+                and normalize_todo_id(item.get("todo_id")) == todo_id
+            ),
+            None,
+        )
+        if isinstance(items, list)
+        else None
+    )
+    resume = normalize_todo_resume_when(todo.get("resume_when")) if todo else None
+    condition = todo.get("resume_condition") if isinstance(todo, dict) else None
+    if (
+        not isinstance(todo, dict)
+        or todo.get("status") not in {"open", "deferred"}
+        or todo.get("task_class") != "advancement_task"
+    ):
+        raise ValueError(
+            "typed blocked no-spend closeout requires the same unfinished "
+            "advancement Todo"
+        )
+    try:
+        observed = datetime.fromisoformat(observed_at.replace("Z", "+00:00"))
+        if observed.tzinfo is None:
+            raise ValueError("observation timestamp must be timezone-aware")
+    except (TypeError, ValueError) as exc:
+        raise ValueError("typed blocked retry wait has an invalid timestamp") from exc
+    if (
+        not resume
+        and not todo.get("resume_when")
+        and allow_turn_settlement_retry
+        and todo.get("status") == "open"
+    ):
+        due_at = (
+            (observed + timedelta(seconds=TURN_SETTLEMENT_RETRY_SECONDS))
+            .isoformat()
+            .replace("+00:00", "Z")
+        )
+        return {
+            "schema_version": BLOCKED_RETRY_SCHEMA_VERSION,
+            "source": "turn_settlement",
+            "todo_id": todo_id,
+            "resume_when": f"resume_at:{due_at}",
+            "observed_at": observed_at,
+            "due_at": due_at,
+        }
+    if (
+        not resume
+        or not resume.startswith("resume_at:")
+        or todo.get("resume_ready") is not False
+        or not isinstance(condition, dict)
+        or condition.get("kind") != "resume_at"
+        or condition.get("resume_when") != resume
+        or condition.get("satisfied") is not False
+    ):
+        raise ValueError(
+            "typed blocked no-spend closeout requires the same unfinished Todo to "
+            "have a pending resume_when=resume_at:<timezone-aware-time> wait; "
+            "schedule it with todo update, read it back, then retry this Turn"
+        )
+    try:
+        due_at = resume.partition(":")[2]
+        due = datetime.fromisoformat(due_at.replace("Z", "+00:00"))
+        delay = (due - observed).total_seconds()
+    except (TypeError, ValueError) as exc:
+        raise ValueError("typed blocked retry wait has an invalid timestamp") from exc
+    if not MIN_RETRY_SECONDS <= delay <= MAX_RETRY_SECONDS:
+        raise ValueError(
+            "typed blocked retry wait must be due in 1–30 minutes; update "
+            "the Todo resume_at and retry this same Turn"
+        )
+    return {
+        "schema_version": BLOCKED_RETRY_SCHEMA_VERSION,
+        "source": "todo",
+        "todo_id": todo_id,
+        "resume_when": resume,
+        "observed_at": observed_at,
+        "due_at": due_at,
+    }
+
+
+def active_turn_retry_for_run(
+    run: dict[str, Any], *, observed_at: str
+) -> dict[str, Any] | None:
+    """Accept only the exact active, receipt-owned retry on a blocked work Run."""
+
+    todo_id = normalize_todo_id(run.get("todo_id"))
+    retry = run.get("blocked_retry")
+    progress = run.get("progress_observation")
+    if (
+        not todo_id
+        or not isinstance(retry, dict)
+        or retry.get("schema_version") != BLOCKED_RETRY_SCHEMA_VERSION
+        or retry.get("source") != "turn_settlement"
+        or retry.get("todo_id") != todo_id
+        or run.get("delivery_outcome") != "outcome_gap"
+        or not isinstance(progress, dict)
+        or progress.get("result_class") != "blocked"
+        or not run.get("turn_instance_id")
+    ):
+        return None
+    try:
+        now = datetime.fromisoformat(observed_at.replace("Z", "+00:00"))
+        observed = datetime.fromisoformat(
+            str(retry["observed_at"]).replace("Z", "+00:00")
+        )
+        due = datetime.fromisoformat(str(retry["due_at"]).replace("Z", "+00:00"))
+        delay = (due - observed).total_seconds()
+    except (KeyError, ValueError, TypeError):
+        return None
+    if (
+        now.tzinfo is None
+        or observed.tzinfo is None
+        or due.tzinfo is None
+        or not MIN_RETRY_SECONDS <= delay <= MAX_RETRY_SECONDS
+        or due <= now
+        or retry.get("resume_when") != f"resume_at:{retry['due_at']}"
+    ):
+        return None
+    return retry
+
+
+def overlay_active_turn_retries(
+    status_payload: dict[str, Any],
+    *,
+    goal_id: str,
+    agent_id: str | None,
+    observed_at: str,
+) -> dict[str, Any]:
+    """Project a receipt-owned wait into selection without editing hard-lease Todos.
+
+    The committed Run is the retry authority.  Its five-minute wait is scoped
+    to the same Goal/Agent/Todo and expires without a maintenance write.  A
+    newer work Run on that Todo supersedes it.  The canonical Todo remains
+    open and keeps its completion validator and lease rules intact.
+    """
+
+    if not agent_id:
+        return status_payload
+    history = status_payload.get("run_history")
+    goals = history.get("goals") if isinstance(history, dict) else None
+    goal = (
+        next(
+            (
+                value
+                for value in goals
+                if isinstance(value, dict) and value.get("id") == goal_id
+            ),
+            None,
+        )
+        if isinstance(goals, list)
+        else None
+    )
+    semantic = goal.get("semantic_history") if isinstance(goal, dict) else None
+    runs = (
+        semantic.get("active_blocked_retry_runs")
+        if isinstance(semantic, dict)
+        and isinstance(semantic.get("active_blocked_retry_runs"), list)
+        else goal.get("latest_runs")
+        if isinstance(goal, dict)
+        else None
+    )
+    if not isinstance(runs, list):
+        return status_payload
+    waits: dict[str, dict[str, Any]] = {}
+    seen: set[str] = set()
+    for run in runs:
+        if not isinstance(run, dict) or run.get("agent_id") != agent_id:
+            continue
+        todo_id = normalize_todo_id(run.get("todo_id"))
+        if not todo_id or todo_id in seen:
+            continue
+        classification = str(run.get("classification") or "")
+        if classification.startswith(("quota_slot_", "quota_scheduler_")):
+            continue
+        seen.add(todo_id)
+        retry = active_turn_retry_for_run(run, observed_at=observed_at)
+        if retry is not None:
+            waits[todo_id] = retry
+    if not waits:
+        return status_payload
+    queue = status_payload.get("attention_queue")
+    items = queue.get("items") if isinstance(queue, dict) else None
+    if not isinstance(items, list):
+        return status_payload
+    projected_items = []
+    changed = False
+
+    def overlay_summary(summary: Any) -> tuple[Any, bool]:
+        if not isinstance(summary, dict):
+            return summary, False
+        projected: dict[str, Any] = {}
+        summary_changed = False
+        for key, value in summary.items():
+            if not isinstance(value, list):
+                projected[key] = value
+                continue
+            projected_rows = []
+            for todo in value:
+                retry = (
+                    waits.get(normalize_todo_id(todo.get("todo_id")))
+                    if isinstance(todo, dict)
+                    else None
+                )
+                if (
+                    retry is not None
+                    and todo.get("status") == "open"
+                    and todo.get("task_class") == "advancement_task"
+                    and not todo.get("resume_when")
+                ):
+                    resume = retry["resume_when"]
+                    projected_rows.append(
+                        {
+                            **todo,
+                            "resume_when": resume,
+                            "resume_ready": False,
+                            "resume_condition": {
+                                "kind": "resume_at",
+                                "resume_when": resume,
+                                "satisfied": False,
+                            },
+                        }
+                    )
+                    summary_changed = True
+                else:
+                    projected_rows.append(todo)
+            projected[key] = projected_rows
+        return (projected if summary_changed else summary), summary_changed
+
+    for item in items:
+        if not isinstance(item, dict) or item.get("goal_id") != goal_id:
+            projected_items.append(item)
+            continue
+        summary = item.get("agent_todos")
+        projected_summary, summary_changed = overlay_summary(summary)
+        asset = item.get("project_asset")
+        asset_summary, asset_changed = overlay_summary(
+            asset.get("agent_todos") if isinstance(asset, dict) else None
+        )
+        if summary_changed or asset_changed:
+            changed = True
+            projected_items.append(
+                {
+                    **item,
+                    "agent_todos": projected_summary,
+                    **(
+                        {"project_asset": {**asset, "agent_todos": asset_summary}}
+                        if asset_changed
+                        else {}
+                    ),
+                }
+            )
+        else:
+            projected_items.append(item)
+    return (
+        {
+            **status_payload,
+            "attention_queue": {**queue, "items": projected_items},
+        }
+        if changed
+        else status_payload
+    )
diff --git a/loopx/control_plane/quota/blocked_retry.ts b/loopx/control_plane/quota/blocked_retry.ts
new file mode 100644
index 0000000000..bc8aa9e056
--- /dev/null
+++ b/loopx/control_plane/quota/blocked_retry.ts
@@ -0,0 +1,15 @@
+import { jsonObject } from "../runtime_decode.ts";
+
+/** A typed blocked Turn may close without spend only with a bounded retry. */
+export function isBoundedBlockedRetry(value: unknown, todoId: string | null): boolean {
+  const retry = jsonObject(value);
+  if (!retry || retry.schema_version !== "quota_blocked_retry_v0" ||
+      (retry.source !== "todo" && retry.source !== "turn_settlement") ||
+      typeof todoId !== "string" || retry.todo_id !== todoId ||
+      typeof retry.observed_at !== "string" || typeof retry.due_at !== "string" ||
+      retry.resume_when !== `resume_at:${retry.due_at}`) return false;
+  const observed = Date.parse(retry.observed_at);
+  const due = Date.parse(retry.due_at);
+  const delay = (due - observed) / 1000;
+  return Number.isFinite(delay) && delay >= 60 && delay <= 30 * 60;
+}
diff --git a/loopx/control_plane/quota/live_decision.py b/loopx/control_plane/quota/live_decision.py
index 2a9f1e0eef..a58010a08a 100644
--- a/loopx/control_plane/quota/live_decision.py
+++ b/loopx/control_plane/quota/live_decision.py
@@ -9,11 +9,13 @@
 from ...quota import build_quota_should_run
 from ...agent_registry import load_goal_from_registry
 from ..agent_context import project_agent_context, project_goal_agent_context
+from ..runtime.time import now_utc_iso
 from ..capability_hooks import (
     InteractionProjectionHookRegistration,
     dispatch_interaction_projection_hooks,
 )
 from .effect_program import ReceiptBoundReplayPhase
+from .blocked_retry import overlay_active_turn_retries
 from .settlement import (
     read_heartbeat_settlement,
 )
@@ -564,6 +566,12 @@ def build_live_quota_should_run_decision(
                 for item in queue.get("items") or []
             ],
         }
+    decision_status_payload = overlay_active_turn_retries(
+        decision_status_payload,
+        goal_id=goal_id,
+        agent_id=agent_id,
+        observed_at=now_utc_iso(),
+    )
     payload = build_quota_should_run(
         decision_status_payload,
         goal_id=goal_id,
diff --git a/loopx/control_plane/quota/projection_repair.py b/loopx/control_plane/quota/projection_repair.py
index fb201de35c..0150dd94a6 100644
--- a/loopx/control_plane/quota/projection_repair.py
+++ b/loopx/control_plane/quota/projection_repair.py
@@ -4,7 +4,7 @@
 import fnmatch
 from typing import Any
 
-from ...state_projection import is_user_wait_text
+from ...state_projection import actions_are_projection_aligned, is_user_wait_text
 from ..todos.contract import (
     TODO_TASK_CLASS_ADVANCEMENT,
     normalize_required_write_scopes,
@@ -99,6 +99,37 @@ def build_state_projection_gap_repair_hint(
         return None
     if open_todo_count(user_todo_summary) > 0 or open_todo_count(agent_todo_summary) > 0:
         return None
+    # A deferred advancement Todo with a pending dated resume is already the
+    # concrete projection of Next Action.  It is temporarily unrunnable, not
+    # missing, so asking for Todo expansion would defeat its bounded wait.
+    evidence = gap.get("first_evidence")
+    deferred = (
+        agent_todo_summary.get("deferred_items")
+        if isinstance(agent_todo_summary, dict)
+        else None
+    )
+    if (
+        isinstance(evidence, list)
+        and evidence
+        and gap.get("evidence_count") == len(evidence)
+        and isinstance(deferred, list)
+        and all(
+            isinstance(entry, dict)
+            and entry.get("kind") == "next_action_executable_without_agent_todo"
+            and entry.get("target_role") == "agent"
+            and any(
+                isinstance(item, dict)
+                and item.get("task_class") == TODO_TASK_CLASS_ADVANCEMENT
+                and item.get("status") == "deferred"
+                and item.get("resume_ready") is False
+                and str(item.get("resume_when") or "").startswith("resume_at:")
+                and actions_are_projection_aligned(entry.get("text"), item.get("text"))
+                for item in deferred
+            )
+            for entry in evidence
+        )
+    ):
+        return None
     must_attempt = bool(
         work_lane_contract
         and work_lane_contract.get("must_attempt_work") is True
diff --git a/loopx/control_plane/quota/settlement_readback.ts b/loopx/control_plane/quota/settlement_readback.ts
index 9682d2f017..726169f022 100644
--- a/loopx/control_plane/quota/settlement_readback.ts
+++ b/loopx/control_plane/quota/settlement_readback.ts
@@ -35,6 +35,7 @@ import {
   receiptBoundReplayPhase,
 } from "./settlement_phase.ts";
 import { isTurnScopedSettlementOutcome } from "../work_items/delivery_outcome.ts";
+import { isBoundedBlockedRetry } from "./blocked_retry.ts";
 import {
   decodeRefreshRetry,
   isMaterialMonitorPoll,
@@ -978,7 +979,8 @@ function readQuotaSettlementFromRequest(
   const blockedNoSpend = writeback.failure === null &&
     spendRun === null && spendEvent === null &&
     identity.binding_kind === "todo" &&
-    writebackRun?.delivery_outcome === "outcome_gap" &&
+    writebackRun !== null && writebackRun.delivery_outcome === "outcome_gap" &&
+    isBoundedBlockedRetry(writebackRun.blocked_retry, identity.todo_id) &&
     isTurnScopedSettlementOutcome(
       writebackRun.delivery_outcome,
       writebackRun.progress_observation,
diff --git a/loopx/control_plane/runtime/run_compaction.py b/loopx/control_plane/runtime/run_compaction.py
index 9f1fddd9e3..89996e8347 100644
--- a/loopx/control_plane/runtime/run_compaction.py
+++ b/loopx/control_plane/runtime/run_compaction.py
@@ -94,6 +94,7 @@
     "progress_scope",
     "todo_id",
     "progress_observation",
+    "blocked_retry",
     "delivery_batch_scale",
     "delivery_outcome",
     "lifecycle_phase",
diff --git a/loopx/control_plane/runtime/run_context_retention.py b/loopx/control_plane/runtime/run_context_retention.py
index fae8667df6..1b0149aa49 100644
--- a/loopx/control_plane/runtime/run_context_retention.py
+++ b/loopx/control_plane/runtime/run_context_retention.py
@@ -3,6 +3,7 @@
 from collections.abc import Callable
 from typing import Any
 
+from ..quota.blocked_retry import active_turn_retry_for_run
 from ..work_items.delivery_outcome import (
     MATERIAL_DELIVERY_OUTCOMES,
     PROGRESS_DELIVERY_OUTCOMES,
@@ -13,6 +14,7 @@
     autonomous_replan_ack_recorded,
 )
 from ..work_items.autonomous_replan_obligation import run_history_agent_id
+from .time import now_utc_iso
 
 GOAL_SEMANTIC_HISTORY_SCHEMA_VERSION = "goal_semantic_history_v0"
 SEMANTIC_CONTEXT_RUN_FIELDS = (
@@ -125,13 +127,17 @@ def goal_semantic_history_from_runs(
 ) -> dict[str, Any]:
     """Select time-bounded control semantics from newest-first run history.
 
-    The result grows with participating agents, not heartbeat count. Recent
-    drill-down rows remain a separate strictly bounded list.
+    The result grows with participating agents and currently waiting Todos,
+    not heartbeat count. Recent drill-down rows remain a separate strictly
+    bounded list.
     """
 
     contexts: dict[str, dict[str, Any]] = {}
     resolved_agent_vision: set[str] = set()
     latest_owner_correction_run: dict[str, Any] | None = None
+    active_blocked_retry_runs: list[dict[str, Any]] = []
+    seen_retry_todos: set[tuple[str, str]] = set()
+    observed_at = now_utc_iso()
 
     for run in runs:
         if latest_owner_correction_run is None and isinstance(
@@ -142,6 +148,17 @@ def goal_semantic_history_from_runs(
         agent_id = _agent_id_for_run(run)
         if not agent_id:
             continue
+        todo_id = str(run.get("todo_id") or "").strip()
+        classification = str(run.get("classification") or "")
+        if todo_id and not classification.startswith(("quota_slot_", "quota_scheduler_")):
+            key = (agent_id, todo_id)
+            if key not in seen_retry_todos:
+                seen_retry_todos.add(key)
+                if (
+                    active_turn_retry_for_run(run, observed_at=observed_at)
+                    is not None
+                ):
+                    active_blocked_retry_runs.append(run)
         context = contexts.setdefault(agent_id, {"agent_id": agent_id})
 
         checkpoint = run.get("vision_checkpoint")
@@ -202,6 +219,7 @@ def goal_semantic_history_from_runs(
 
     semantic_history: dict[str, Any] = {
         "schema_version": GOAL_SEMANTIC_HISTORY_SCHEMA_VERSION,
+        "active_blocked_retry_runs": active_blocked_retry_runs,
         "agents": [
             context
             for context in contexts.values()
@@ -250,6 +268,11 @@ def compact_goal_semantic_history(
         "schema_version": GOAL_SEMANTIC_HISTORY_SCHEMA_VERSION,
         "agents": agents,
     }
+    compact["active_blocked_retry_runs"] = [
+        compact_run(run)
+        for run in value.get("active_blocked_retry_runs") or []
+        if isinstance(run, dict)
+    ]
     owner_correction_run = value.get("latest_owner_correction_run")
     if isinstance(owner_correction_run, dict):
         compacted_run = compact_run(owner_correction_run)
diff --git a/loopx/state_refresh.py b/loopx/state_refresh.py
index c5ae786f3f..83ebdb851c 100644
--- a/loopx/state_refresh.py
+++ b/loopx/state_refresh.py
@@ -28,6 +28,9 @@
 from .control_plane.quota.refresh_external_delivery import (
     finish_external_delivery_refresh, refresh_recovery_payload,
 )
+from .control_plane.quota.blocked_retry import require_blocked_retry_wait
+from .control_plane.coordination.local_authority import local_authority_is_promoted
+from .control_plane.todos.active_state_todo_parser import parse_active_state_todos
 from .control_plane.quota.settlement import (
     SettlementIdentity,
     attach_settlement_progress,
@@ -515,6 +518,7 @@ def _build_state_refresh_output_projections(
         "delivery_outcome",
         "delivery_workspace",
         "settlement_identity",
+        "blocked_retry",
         "refresh_recovery",
         "turn_instance_id",
         "todo_id",
@@ -1174,6 +1178,34 @@ def refresh_state_run(
         effective_autonomous_replan_recorded = (
             replan_qualification.autonomous_replan_recorded
         )
+        blocked_retry = None
+        if checkpoint_supplement and prior_writeback_run is not None:
+            blocked_retry = prior_writeback_run.get("blocked_retry")
+        elif (
+            settlement_identity is not None
+            and settlement_identity.binding_kind.value == "todo"
+            and normalized_delivery_outcome == "outcome_gap"
+            and isinstance(normalized_progress_observation, dict)
+            and normalized_progress_observation.get("result_class") == "blocked"
+        ):
+            blocked_todo_fields = todo_fields
+            if blocked_todo_fields is None:
+                blocked_todo_fields = parse_active_state_todos(
+                    state_text,
+                    goal=registry_goal,
+                    state_path=resolved_state_file,
+                    preferred_todo_ids={settlement_identity.todo_id or ""},
+                    rollout_events=planning_events,
+                    item_limit=None,
+                )
+            blocked_retry = require_blocked_retry_wait(
+                blocked_todo_fields,
+                todo_id=settlement_identity.todo_id or "",
+                observed_at=generated_at,
+                allow_turn_settlement_retry=local_authority_is_promoted(
+                    runtime_root=runtime_root, goal_id=safe_goal_id,
+                ),
+            )
         # read_heartbeat_settlement admits checkpoint_supplement only for a
         # checkpoint-only retry of an already committed writeback with the
         # exact Goal/Agent/Todo/Turn and delivery identity. It rejects replayed
@@ -1206,6 +1238,7 @@ def refresh_state_run(
             todo_id=(settlement_identity.todo_id if settlement_identity else None),
             completion_todo_id=completion_todo_id,
             autonomous_replan_recorded=effective_autonomous_replan_recorded,
+            blocked_retry=blocked_retry,
         )
         if checkpoint_supplement and not vision_checkpoint.get("satisfied"):
             raise ValueError(
@@ -1312,6 +1345,8 @@ def refresh_state_run(
             settlement_identity=settlement_identity,
             todo_fields=todo_fields,
         )
+        if blocked_retry is not None:
+            record["blocked_retry"] = blocked_retry
         if delivery_workspace_causality:
             record["delivery_workspace_causality"] = delivery_workspace_causality
         if refresh_recovery:

From ad536d4aa3b7c2db7c39661a3fc0a60fb7a22673 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Fri, 25 Sep 2026 00:18:03 +0800
Subject: [PATCH 4/8] test(quota): prove bounded blocked retry under hard
 leases

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 .../references/repair-patterns.md             |   1 +
 .../test_quota_blocked_retry_projection.py    | 297 ++++++++++++++++++
 .../test_quota_settlement_cli.py              | 239 +++++++++++++-
 .../quota_settlement_readback.test.ts         |  27 ++
 .../unsettled_host_turn_recovery.test.ts      |   8 +
 .../vision_checkpoint.test.ts                 |  29 ++
 6 files changed, 598 insertions(+), 3 deletions(-)
 create mode 100644 tests/control_plane/test_quota_blocked_retry_projection.py

diff --git a/skills/loopx-self-repair/references/repair-patterns.md b/skills/loopx-self-repair/references/repair-patterns.md
index 3db7ad547d..271caa8e79 100644
--- a/skills/loopx-self-repair/references/repair-patterns.md
+++ b/skills/loopx-self-repair/references/repair-patterns.md
@@ -32,6 +32,7 @@ teaches a reusable control-plane lesson.
 | `pending_capability_intent_projection_gap` | A durable post-writeback sidecar says `intent_recorded`, but quota remains `monitor_quiet_skip` or `terminal_no_followup`; no governed executor wakes, so no local artifact or exact approval Todo appears. | Exact Goal/Agent sidecar, intent schema and authority, consumption receipt, quota interaction contract, generated-artifact count, approval-Todo count, and external-effect count. | The producer journal was durable but no read model projected unconsumed capability intents into quota arbitration. Terminal or quiet lifecycle state therefore hid required capability work forever. | Add a provider-neutral TypeScript-validated pending-intent interaction slot, let the opted-in capability read only exact eligible sidecars, and give the pending action precedence over quiet/terminal routes. Consume it through an idempotent local executor that freezes checked artifacts and one digest-bound user gate; keep external delivery unauthorized and suppress consumed intents on replay. |
 | `approved_capability_delivery_successor_gap` | A governed report or other capability payload is frozen and its exact user gate is approved, but quota returns quiet terminal/monitor state and no provider action runs. A later caller may also improvise a destination or default sender because the approval receipt contains no executable route. | Frozen generation/consumption receipt, completed gate decision and scope, linked agent Todo, required decision scopes before/after approval, Goal Channel binding, selected Todo, provider effect and exact sender/destination readback. | The consumer created only a user gate. Approval recorded authority but had no explicitly linked, typed agent successor for Todo/quota selection, so the external effect existed only in chat memory or caller convention. | Before the gate, create one blocked agent successor bound to the frozen generation, exact decision scope, provider capability, and safe write scope; link the gate with `unblocks_todo_id`. Approval must atomically consume only that scope and resume the successor. The provider then resolves route and sender from the durable project binding, rejects caller overrides/default fallbacks, and records success only after native effect readback. |
 | `host_closeout_presentation_lookup_gap` | A prior heartbeat remains in closeout recovery after a successful blocked/deferred/completed Todo writeback; adding unrelated Todos changes recovery. | Receipt-bound Todo id, exact `todo list --todo-id` readback, compact quota visibility lanes, same-Turn retry and quota-spend count. | Recovery treated absence from a bounded presentation list as absence of a durable lifecycle transition. | Resolve the exact receipt-bound Todo through the existing provider-aware read path before evaluating closeout; preserve provider errors, identity binding and no-spend recovery. Cover crowded versus small inventories and real legacy/File/SQLite CLI writes; never raise display limits or fabricate settlement receipts. |
+| `blocked_turn_retry_history_gap` | An exact blocked Turn can close without spending, but its open Todo is immediately selected again or loses its wait when other runs crowd the status display; a peer hard lease prevents the blocked agent from updating the Todo. | Exact Goal/Agent/Todo/Turn receipt, peer lease, canonical Todo validator, persisted retry due time, full run index versus compact recent runs, next-Turn selection and spend count. | The settlement had no bounded retry authority, or quota selection reconstructed it from a truncated display list and tried to mutate a peer-gated Todo. | Persist one bounded retry on the blocked Turn receipt, retain it in the full-index semantic projection, and overlay it only for the same agent's selection until due or superseded by a newer work run. Keep the canonical Todo and terminal validator intact; prove replay, no spend, crowded history, and real File/SQLite hard-lease readback. |
 | `host_closeout_history_scan_amplification` | Quota entry approaches or exceeds its runtime timeout only on a long-lived Goal, especially when many prior Turns already have valid closeouts. | Turn and receipt counts, rollout-log and runs-index sizes, candidate count, per-stage elapsed time, declared runtime budget, and the same synthetic settled history before and after repair. | Preflight rebuilt append-order recency with repeated list searches, then each candidate reread, reparsed, and rescanned the complete event and run history. A larger timeout hid the multiplicative work without bounding it. | Parse each persisted source once, retain strict-versus-tolerant read semantics in the shared snapshot, build append-order and exact Turn/effect indexes, and reduce every candidate from that snapshot. Restore the ordinary runtime budget and cover a sufficiently large all-settled history that must traverse every candidate; never truncate history, skip identity validation, or raise the timeout as the repair. |
 | `turn_replay_recovery_semantic_gap` | A real Turn resumes from a safe Journal prefix, but `inspect-journal` presents `replay_legal=false` as if recovery were forbidden; scheduler-only or saved-Host-result recovery is especially misleading. | Journal integrity fields, replay decision, executor-adopted recovery decision, completed phase prefix, complete typed settlement identity, authoritative selected-Todo lineage, conditional Host Session Binding check, prepared-effect presence, and bounded recovery outcome. | Effect-free terminal replay and effectful executor recovery were collapsed into one user-facing decision even though the executor used separate status/phase rules; initial repairs also omitted either the settlement identity or its binding-to-Turn cross-check, allowing a drifted goal, agent, or canonical Todo to reach a later effect. | Keep replay legality and Journal consistency distinct. Make one typed recovery decision the source used by both executor and inspection; validate and cross-bind the canonical settlement goal, agent, Turn instance, binding, and effect id before authorization, including the selected Todo or adaptive primary Todo; project continue/resume phase/Host reinvocation/reason plus only participating checks, and persist a bounded planned-versus-actual audit. Cover identity and binding drift with zero-provider-call regressions. Preserve the existing prepared-effect readback owner and do not claim general exactly-once semantics. |
 | `managed_chat_resume_turn_identity_gap` | A managed Chat Session resumes after its adapter becomes unhealthy, returns to `ready`, and clears `active_turn_id`, but the interrupted Turn remains nonterminal and unowned. | Pre-resume Session snapshot, post-prepare persisted Session, adapter health, interrupted Turn status and error code, final Session state. | Resume preparation cleared the persisted active Turn reference before adapter recovery, then recovery re-read only the mutated Session and lost the pre-resume Turn identity. | Carry the pre-mutation Turn id through the fresh closed-state check, terminalize that Turn as `failed/server_restarted` before restoring the Session, and cover unhealthy-adapter resume with a focused regression while retaining the per-Session lifecycle lock. |
diff --git a/tests/control_plane/test_quota_blocked_retry_projection.py b/tests/control_plane/test_quota_blocked_retry_projection.py
new file mode 100644
index 0000000000..c84665dbdd
--- /dev/null
+++ b/tests/control_plane/test_quota_blocked_retry_projection.py
@@ -0,0 +1,297 @@
+import pytest
+
+from loopx.control_plane.quota.blocked_retry import (
+    overlay_active_turn_retries,
+    require_blocked_retry_wait,
+)
+from loopx.control_plane.quota.projection_repair import (
+    build_state_projection_gap_repair_hint,
+)
+from loopx.control_plane.runtime.run_context_retention import (
+    compact_goal_semantic_history,
+    goal_semantic_history_from_runs,
+)
+
+
+def test_deferred_retry_suppresses_only_its_own_projection_gap() -> None:
+    gap = {
+        "requires_todo_expansion": True,
+        "evidence_count": 1,
+        "first_evidence": [
+            {
+                "kind": "next_action_executable_without_agent_todo",
+                "target_role": "agent",
+                "text": "Validate and settle the selected delivery.",
+            }
+        ],
+    }
+    summary = {
+        "open_count": 0,
+        "deferred_items": [
+            {
+                "task_class": "advancement_task",
+                "status": "deferred",
+                "text": "[P1] Validate and settle the selected delivery.",
+                "resume_when": "resume_at:2026-09-24T14:30:00Z",
+                "resume_ready": False,
+            }
+        ],
+    }
+    kwargs = {
+        "candidate_should_run": True,
+        "user_todo_summary": {"open_count": 0},
+        "work_lane_contract": None,
+    }
+    assert (
+        build_state_projection_gap_repair_hint(
+            gap, agent_todo_summary=summary, **kwargs
+        )
+        is None
+    )
+
+    unrelated = {
+        **gap,
+        "first_evidence": [
+            {
+                **gap["first_evidence"][0],
+                "text": "Review a separate blocked source.",
+            }
+        ],
+    }
+    repair = build_state_projection_gap_repair_hint(
+        unrelated, agent_todo_summary=summary, **kwargs
+    )
+    assert repair is not None
+    assert repair["trigger"] == "state_projection_gap"
+
+
+def test_blocked_retry_requires_the_exact_pending_todo_and_bounded_due_time() -> None:
+    def fields(due: str) -> dict:
+        resume = f"resume_at:{due}"
+        return {
+            "agent_todos": {
+                "items": [
+                    {
+                        "todo_id": "todo_current001",
+                        "status": "deferred",
+                        "task_class": "advancement_task",
+                        "resume_when": resume,
+                        "resume_ready": False,
+                        "resume_condition": {
+                            "kind": "resume_at",
+                            "resume_when": resume,
+                            "satisfied": False,
+                        },
+                    }
+                ]
+            }
+        }
+
+    observed = "2026-09-24T14:00:00Z"
+    valid = require_blocked_retry_wait(
+        fields("2026-09-24T14:05:00Z"),
+        todo_id="todo_current001",
+        observed_at=observed,
+    )
+    assert valid["due_at"] == "2026-09-24T14:05:00Z"
+    with pytest.raises(ValueError, match="same unfinished advancement Todo"):
+        require_blocked_retry_wait(
+            fields("2026-09-24T14:05:00Z"),
+            todo_id="todo_other",
+            observed_at=observed,
+        )
+    with pytest.raises(ValueError, match="1–30 minutes"):
+        require_blocked_retry_wait(
+            fields("2026-09-24T14:31:00Z"),
+            todo_id="todo_current001",
+            observed_at=observed,
+        )
+
+
+def test_turn_owned_retry_is_selection_only_and_expires() -> None:
+    retry = {
+        "schema_version": "quota_blocked_retry_v0",
+        "source": "turn_settlement",
+        "todo_id": "todo_current001",
+        "resume_when": "resume_at:2026-09-24T14:05:00Z",
+        "observed_at": "2026-09-24T14:00:00Z",
+        "due_at": "2026-09-24T14:05:00Z",
+    }
+    original = {
+        "todo_id": "todo_current001",
+        "task_class": "advancement_task",
+        "status": "open",
+    }
+    status = {
+        "run_history": {
+            "goals": [
+                {
+                    "id": "goal-a",
+                    "latest_runs": [
+                        {
+                            "agent_id": "agent-a",
+                            "todo_id": "todo_current001",
+                            "turn_instance_id": "turn-a",
+                            "delivery_outcome": "outcome_gap",
+                            "progress_observation": {"result_class": "blocked"},
+                            "blocked_retry": retry,
+                        }
+                    ],
+                }
+            ]
+        },
+        "attention_queue": {
+            "items": [
+                {
+                    "goal_id": "goal-a",
+                    "agent_todos": {
+                        "items": [original],
+                    },
+                }
+            ]
+        },
+    }
+    waiting = overlay_active_turn_retries(
+        status,
+        goal_id="goal-a",
+        agent_id="agent-a",
+        observed_at="2026-09-24T14:02:00Z",
+    )
+    projected = waiting["attention_queue"]["items"][0]["agent_todos"]["items"][0]
+    assert projected["resume_when"] == retry["resume_when"]
+    assert projected["resume_ready"] is False
+    assert "resume_when" not in original
+    assert (
+        overlay_active_turn_retries(
+            status,
+            goal_id="goal-a",
+            agent_id="agent-a",
+            observed_at="2026-09-24T14:05:00Z",
+        )
+        is status
+    )
+    assert (
+        overlay_active_turn_retries(
+            status,
+            goal_id="goal-a",
+            agent_id="agent-b",
+            observed_at="2026-09-24T14:02:00Z",
+        )
+        is status
+    )
+    superseded = {
+        **status,
+        "run_history": {
+            "goals": [
+                {
+                    "id": "goal-a",
+                    "latest_runs": [
+                        {
+                            "agent_id": "agent-a",
+                            "todo_id": "todo_current001",
+                            "delivery_outcome": "outcome_progress",
+                        },
+                        status["run_history"]["goals"][0]["latest_runs"][0],
+                    ],
+                }
+            ]
+        },
+    }
+    assert (
+        overlay_active_turn_retries(
+            superseded,
+            goal_id="goal-a",
+            agent_id="agent-a",
+            observed_at="2026-09-24T14:02:00Z",
+        )
+        is superseded
+    )
+
+
+def test_blocked_retry_survives_recent_run_display_limit(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(
+        "loopx.control_plane.runtime.run_context_retention.now_utc_iso",
+        lambda: "2026-09-24T14:02:00Z",
+    )
+    blocked = {
+        "agent_id": "agent-a",
+        "todo_id": "todo_current001",
+        "turn_instance_id": "turn-a",
+        "delivery_outcome": "outcome_gap",
+        "progress_observation": {"result_class": "blocked"},
+        "blocked_retry": {
+            "schema_version": "quota_blocked_retry_v0",
+            "source": "turn_settlement",
+            "todo_id": "todo_current001",
+            "resume_when": "resume_at:2026-09-24T14:05:00Z",
+            "observed_at": "2026-09-24T14:00:00Z",
+            "due_at": "2026-09-24T14:05:00Z",
+        },
+    }
+    newer = [
+        {"agent_id": "agent-a", "todo_id": f"todo_other{i:03d}"} for i in range(40)
+    ]
+    semantic = compact_goal_semantic_history(
+        goal_semantic_history_from_runs([*newer, blocked]),
+        compact_run=dict,
+    )
+    assert semantic is not None
+    assert semantic["active_blocked_retry_runs"] == [blocked]
+    status = {
+        "run_history": {
+            "goals": [
+                {
+                    "id": "goal-a",
+                    "latest_runs": newer[:3],
+                    "semantic_history": semantic,
+                }
+            ]
+        },
+        "attention_queue": {
+            "items": [
+                {
+                    "goal_id": "goal-a",
+                    "agent_todos": {
+                        "items": [
+                            {
+                                "todo_id": "todo_current001",
+                                "task_class": "advancement_task",
+                                "status": "open",
+                            }
+                        ],
+                    },
+                }
+            ]
+        },
+    }
+    projected = overlay_active_turn_retries(
+        status,
+        goal_id="goal-a",
+        agent_id="agent-a",
+        observed_at="2026-09-24T14:02:00Z",
+    )
+    assert (
+        projected["attention_queue"]["items"][0]["agent_todos"]["items"][0][
+            "resume_ready"
+        ]
+        is False
+    )
+    assert (
+        "resume_when"
+        not in status["attention_queue"]["items"][0]["agent_todos"]["items"][0]
+    )
+
+    superseded = goal_semantic_history_from_runs(
+        [
+            {
+                "agent_id": "agent-a",
+                "todo_id": "todo_current001",
+                "delivery_outcome": "outcome_progress",
+            },
+            *newer,
+            blocked,
+        ]
+    )
+    assert superseded["active_blocked_retry_runs"] == []
diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py
index 68c1ae4096..09049bd6f9 100644
--- a/tests/control_plane/test_quota_settlement_cli.py
+++ b/tests/control_plane/test_quota_settlement_cli.py
@@ -5,6 +5,7 @@
 import shlex
 import subprocess
 import sys
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Any
 from urllib.parse import quote
@@ -310,8 +311,9 @@ def _configure_completion_validation_todo(project: Path) -> Path:
     state_text = state_path.read_text(encoding="utf-8")
     state_path.write_text(
         state_text.replace(
-            "action_kind=validate -->",
-            "action_kind=validate validation_command=pytest -->",
+            "action_kind=validate",
+            "action_kind=validate validation_command=pytest",
+            1,
         ),
         encoding="utf-8",
     )
@@ -900,6 +902,7 @@ def test_typed_outcome_gap_settles_exact_turn_without_becoming_progress(
         required_capability="filesystem_write",
     )
     _configure_completion_validation_todo(project)
+    _configure_selectable_alternative(project)
     turn_id = "turn-typed-blocker-settlement"
     binding = (
         "--agent-id",
@@ -986,6 +989,55 @@ def test_typed_outcome_gap_settles_exact_turn_without_becoming_progress(
     assert mismatch_rc == 1, mismatch
     assert "settlement binding does not match" in mismatch["error"]
 
+    unscheduled_rc, unscheduled = _run_cli(
+        registry_path,
+        runtime,
+        *common_refresh_args,
+        "--progress-result-class",
+        "blocked",
+        "--progress-blocker-id",
+        "blocker:runtime-boundary",
+        "--progress-evidence-id",
+        "evidence:runtime-boundary",
+        cwd=project,
+    )
+    assert unscheduled_rc == 1, unscheduled
+    assert "pending resume_when=resume_at" in unscheduled["error"]
+    assert _classification_count(runtime, "typed_blocker_writeback") == 0
+
+    due_at = (datetime.now(timezone.utc) + timedelta(minutes=5)).replace(
+        microsecond=0
+    ).isoformat().replace("+00:00", "Z")
+    wait_rc, wait = _run_cli(
+        registry_path,
+        runtime,
+        "todo",
+        "update",
+        "--goal-id",
+        GOAL_ID,
+        "--todo-id",
+        TODO_ID,
+        "--agent-id",
+        AGENT_ID,
+        "--status",
+        "open",
+        "--resume-when",
+        f"resume_at:{due_at}",
+        "--successor-todo-id",
+        ALTERNATIVE_TODO_ID,
+    )
+    assert wait_rc == 0, wait
+    listed_rc, listed = _run_cli(
+        registry_path, runtime, "todo", "list", "--goal-id", GOAL_ID
+    )
+    assert listed_rc == 0, listed
+    waiting_todo = next(
+        item for item in listed["todos"] if item["todo_id"] == TODO_ID
+    )
+    assert waiting_todo["resume_when"] == f"resume_at:{due_at}"
+    assert waiting_todo["resume_ready"] is False
+    assert waiting_todo["successor_todo_ids"] == [ALTERNATIVE_TODO_ID]
+
     refresh_rc, refresh = _run_cli(
         registry_path,
         runtime,
@@ -998,10 +1050,11 @@ def test_typed_outcome_gap_settles_exact_turn_without_becoming_progress(
         "evidence:runtime-boundary",
         cwd=project,
     )
-    assert refresh_rc == 0, refresh
+    assert refresh_rc == 0, refresh.get("error") or refresh
     assert refresh["delivery_outcome"] == "outcome_gap"
     assert refresh["progress_observation"]["result_class"] == "blocked"
     assert refresh["progress_observation"]["work_item_id"] == TODO_ID
+    assert refresh["blocked_retry"]["resume_when"] == f"resume_at:{due_at}"
     assert [
         receipt["step_kind"]
         for receipt in refresh["settlement_result"]["receipts"]
@@ -1054,6 +1107,186 @@ def test_typed_outcome_gap_settles_exact_turn_without_becoming_progress(
     )
     assert next_rc == 0, next_turn
     assert next_turn["effective_action"] != "unsettled_host_turn_recovery"
+    assert next_turn["selected_todo"]["todo_id"] == ALTERNATIVE_TODO_ID
+
+
+def test_typed_blocked_retry_without_successor_defers_the_only_todo(
+    tmp_path: Path,
+) -> None:
+    project, runtime, registry_path = _write_fixture(tmp_path)
+    _configure_completion_validation_todo(project)
+    turn_id = "turn-typed-blocker-only-todo"
+    binding = (
+        "--agent-id", AGENT_ID,
+        "--todo-id", TODO_ID,
+        "--turn-instance-id", turn_id,
+    )
+    guard_rc, guard = _run_cli(
+        registry_path, runtime, "quota", "should-run", "--codex-app",
+        "--goal-id", GOAL_ID, *binding, "--scan-path", str(project), cwd=project,
+    )
+    assert guard_rc == 0, guard
+    due_at = (datetime.now(timezone.utc) + timedelta(minutes=5)).replace(
+        microsecond=0
+    ).isoformat().replace("+00:00", "Z")
+    wait_rc, wait = _run_cli(
+        registry_path, runtime, "todo", "update", "--goal-id", GOAL_ID,
+        "--todo-id", TODO_ID, "--agent-id", AGENT_ID,
+        "--status", "deferred", "--resume-when", f"resume_at:{due_at}",
+    )
+    assert wait_rc == 0, wait
+    listed_rc, listed = _run_cli(
+        registry_path, runtime, "todo", "list", "--goal-id", GOAL_ID
+    )
+    assert listed_rc == 0, listed
+    assert listed["todos"][0]["status"] == "deferred"
+    assert listed["todos"][0]["resume_ready"] is False
+
+    refresh_rc, refresh = _run_cli(
+        registry_path, runtime, "refresh-state", "--goal-id", GOAL_ID,
+        "--classification", "typed_blocker_writeback",
+        "--delivery-batch-scale", "single_surface",
+        "--delivery-outcome", "outcome_gap", *binding,
+        "--progress-result-class", "blocked",
+        "--progress-blocker-id", "blocker:runtime-boundary",
+        "--progress-evidence-id", "evidence:runtime-boundary",
+        "--no-global-sync", "--suppress-external-sinks", cwd=project,
+    )
+    assert refresh_rc == 0, refresh.get("error") or refresh
+    assert refresh["settlement_progress"]["state"] == "settled"
+    assert refresh["blocked_retry"]["resume_when"] == f"resume_at:{due_at}"
+    assert _spend_run_count(runtime) == 0
+
+    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-only-todo-blocked",
+        "--scan-path", str(project), cwd=project,
+    )
+    assert next_rc == 0, next_turn
+    assert next_turn["effective_action"] != "unsettled_host_turn_recovery"
+    assert (next_turn.get("selected_todo") or {}).get("todo_id") != TODO_ID
+    assert next_turn["should_run"] is False, next_turn
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_typed_blocked_retry_with_peer_hard_lease(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    provider: str,
+) -> None:
+    """A peer's overlapping execution lease must not strand exact closeout."""
+    from canonical_authority_fixture import (
+        initialize_canonical_authority,
+        isolate_sqlite_runtime,
+    )
+    from loopx.control_plane.coordination.runtime_shadow import (
+        build_todo_runtime_shadow_projection,
+    )
+
+    if provider == "sqlite":
+        isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry_path = _write_fixture(tmp_path)
+    state = _configure_completion_validation_todo(project)
+    _configure_selectable_alternative(project)
+    registry = json.loads(registry_path.read_text(encoding="utf-8"))
+    registry["goals"][0]["coordination"]["registered_agents"].append("peer-agent")
+    registry["goals"][0]["coordination"]["write_scope"] = ["src/**"]
+    registry["goals"][0]["workspace_guard_policy"] = {
+        "peer_independent_worktree_required": False,
+    }
+    registry_path.write_text(json.dumps(registry), encoding="utf-8")
+    rc, listed = _run_cli(registry_path, runtime, "todo", "list", "--goal-id", GOAL_ID)
+    assert rc == 0, listed
+    todos = listed["todos"]
+    for todo in todos:
+        todo["required_write_scopes"] = ["src/**"]
+    projection = build_todo_runtime_shadow_projection(
+        goal_id=GOAL_ID, todos=todos, handoff_mode="hard_lease", leases=[],
+    )
+    initialize_canonical_authority(
+        runtime, GOAL_ID, projection, state_path=state, provider=provider,
+    )
+
+    turn_id = f"turn-hard-lease-blocker-{provider}"
+    binding = (
+        "--agent-id", AGENT_ID, "--todo-id", TODO_ID,
+        "--turn-instance-id", turn_id,
+    )
+    guard_rc, guard = _run_cli(
+        registry_path, runtime, "quota", "should-run", "--codex-app",
+        "--goal-id", GOAL_ID, *binding, "--scan-path", str(project), cwd=project,
+    )
+    assert guard_rc == 0, guard
+    assert guard["heartbeat_receipt"]["settlement_identity"]["todo_id"] == TODO_ID
+
+    lease_rc, lease = _run_cli(
+        registry_path, runtime, "task-lease", "acquire", "--goal-id", GOAL_ID,
+        "--todo-id", ALTERNATIVE_TODO_ID, "--owner", "peer-agent",
+        "--idempotency-key", f"peer-overlap-{provider}", "--expected-version", "0",
+        "--ttl-seconds", "600", "--write-scope", "src/**",
+    )
+    assert lease_rc == 0, lease
+    assert lease["acquired"] is True
+    assert lease["source_authority"] == f"{provider}_v0"
+    refresh_rc, refresh = _run_cli(
+        registry_path, runtime, "refresh-state", "--goal-id", GOAL_ID,
+        "--classification", "peer_hard_lease_blocker",
+        "--delivery-batch-scale", "single_surface",
+        "--delivery-outcome", "outcome_gap", *binding,
+        "--progress-result-class", "blocked",
+        "--progress-blocker-id", "blocker:peer-hard-lease",
+        "--progress-evidence-id", "evidence:peer-hard-lease",
+        "--no-global-sync", "--suppress-external-sinks", cwd=project,
+    )
+    assert refresh_rc == 0, refresh.get("error") or refresh
+    assert refresh["settlement_progress"]["closeout_kind"] == (
+        "typed_blocked_writeback_no_spend"
+    )
+    assert refresh["blocked_retry"]["source"] == "turn_settlement"
+    due_at = datetime.fromisoformat(refresh["blocked_retry"]["due_at"].replace("Z", "+00:00"))
+    observed = datetime.fromisoformat(refresh["blocked_retry"]["observed_at"].replace("Z", "+00:00"))
+    assert due_at - observed == timedelta(minutes=5)
+    assert _spend_run_count(runtime) == 0
+    replay_rc, replay = _run_cli(
+        registry_path, runtime, "refresh-state", "--goal-id", GOAL_ID,
+        "--classification", "peer_hard_lease_blocker",
+        "--delivery-batch-scale", "single_surface",
+        "--delivery-outcome", "outcome_gap", *binding,
+        "--progress-result-class", "blocked",
+        "--progress-blocker-id", "blocker:peer-hard-lease",
+        "--progress-evidence-id", "evidence:peer-hard-lease",
+        "--no-global-sync", "--suppress-external-sinks", cwd=project,
+    )
+    assert replay_rc == 0, replay.get("error") or replay
+    assert replay["idempotent_replay"] is True
+    assert replay["blocked_retry"] == refresh["blocked_retry"]
+    assert _classification_count(runtime, "peer_hard_lease_blocker") == 1
+    spend_rc, spend = _run_cli(
+        registry_path, runtime, "quota", "spend-slot", "--goal-id", GOAL_ID,
+        "--slots", "1", "--source", "heartbeat", "--execute", *binding,
+        "--scan-path", str(project), cwd=project,
+    )
+    assert spend_rc == 0, spend
+    assert spend["appended"] is False
+    assert _spend_run_count(runtime) == 0
+    rc, after = _run_cli(registry_path, runtime, "todo", "list", "--goal-id", GOAL_ID)
+    assert rc == 0, after
+    original = next(todo for todo in after["todos"] if todo["todo_id"] == TODO_ID)
+    assert original["status"] == "open"
+    assert not original.get("resume_when")
+    assert original["completion_validation_required"] is True
+    assert original["completion_validation_sha256"]
+
+    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", f"turn-after-hard-lease-blocker-{provider}",
+        "--scan-path", str(project), cwd=project,
+    )
+    assert next_rc == 0, next_turn
+    assert next_turn["effective_action"] != "unsettled_host_turn_recovery"
+    assert (next_turn.get("selected_todo") or {}).get("todo_id") != TODO_ID
 
 
 def test_in_flight_progress_preserves_todo_across_heartbeat_settlements(
diff --git a/tests/control_plane_ts/quota_settlement_readback.test.ts b/tests/control_plane_ts/quota_settlement_readback.test.ts
index dc7af1cba3..f97096dee4 100644
--- a/tests/control_plane_ts/quota_settlement_readback.test.ts
+++ b/tests/control_plane_ts/quota_settlement_readback.test.ts
@@ -78,6 +78,7 @@ async function fixture(options: {
   monitor?: boolean;
   writebackOutcome?: string;
   progressObservation?: Record<string, unknown>;
+  blockedRetry?: boolean;
 } = {}) {
   const runtimeRoot = await mkdtemp(join(tmpdir(), "loopx-settlement-readback-"));
   const goalRoot = join(runtimeRoot, "goals", goalId);
@@ -150,6 +151,14 @@ async function fixture(options: {
       todo_id: todoId,
       turn_instance_id: turnId,
       settlement_identity: identity,
+      ...(options.blockedRetry ? {blocked_retry: {
+        schema_version: "quota_blocked_retry_v0",
+        source: "todo",
+        todo_id: todoId,
+        resume_when: "resume_at:2026-09-24T10:05:00Z",
+        observed_at: "2026-09-24T10:00:00Z",
+        due_at: "2026-09-24T10:05:00Z",
+      }} : {}),
       ...(options.progressObservation
         ? { progress_observation: options.progressObservation }
         : {}),
@@ -684,6 +693,7 @@ test("accepts only an attributable typed blocker as an outcome-gap writeback", a
   const qualifiedRuntime = await fixture({
     writeback: true,
     writebackOutcome: "outcome_gap",
+    blockedRetry: true,
     progressObservation: {
       schema_version: "typed_progress_observation_v0",
       result_class: "blocked",
@@ -709,6 +719,7 @@ test("accepts only an attributable typed blocker as an outcome-gap writeback", a
     writeback: true,
     spend: true,
     writebackOutcome: "outcome_gap",
+    blockedRetry: true,
     progressObservation: {
       schema_version: "typed_progress_observation_v0",
       result_class: "blocked",
@@ -727,6 +738,7 @@ test("accepts only an attributable typed blocker as an outcome-gap writeback", a
   const incompleteSpendRuntime = await fixture({
     writeback: true,
     writebackOutcome: "outcome_gap",
+    blockedRetry: true,
     progressObservation: {
       schema_version: "typed_progress_observation_v0",
       result_class: "blocked",
@@ -749,6 +761,21 @@ test("accepts only an attributable typed blocker as an outcome-gap writeback", a
   assert.equal((incompleteSpend.settlement as any).payload.ok, false);
   assert.equal((incompleteSpend.progress as any).closeout_kind, undefined);
 
+  const unscheduledRuntime = await fixture({
+    writeback: true,
+    writebackOutcome: "outcome_gap",
+    progressObservation: {
+      schema_version: "typed_progress_observation_v0",
+      result_class: "blocked",
+      work_item_id: todoId,
+      blocker_id: "blocker-runtime-boundary",
+      evidence_ids: ["evidence-runtime-boundary"],
+    },
+  });
+  const unscheduled = await readQuotaSettlement(request(unscheduledRuntime));
+  assert.equal((unscheduled.progress as any).state, "spend_required");
+  assert.equal((unscheduled.progress as any).closeout_kind, undefined);
+
   const bareRuntime = await fixture({
     writeback: true,
     writebackOutcome: "outcome_gap",
diff --git a/tests/control_plane_ts/unsettled_host_turn_recovery.test.ts b/tests/control_plane_ts/unsettled_host_turn_recovery.test.ts
index 445b72422a..82fae03862 100644
--- a/tests/control_plane_ts/unsettled_host_turn_recovery.test.ts
+++ b/tests/control_plane_ts/unsettled_host_turn_recovery.test.ts
@@ -398,6 +398,14 @@ test("a prior typed blocked writeback closes without a quota debit", async () =>
       todo_id: todoId,
       turn_instance_id: turn,
       settlement_identity: identity,
+      blocked_retry: {
+        schema_version: "quota_blocked_retry_v0",
+        source: "turn_settlement",
+        todo_id: todoId,
+        resume_when: "resume_at:2026-09-24T10:05:00Z",
+        observed_at: "2026-09-24T10:00:00Z",
+        due_at: "2026-09-24T10:05:00Z",
+      },
       progress_observation: {
         schema_version: "typed_progress_observation_v0",
         result_class: "blocked",
diff --git a/tests/control_plane_ts/vision_checkpoint.test.ts b/tests/control_plane_ts/vision_checkpoint.test.ts
index 549ebcf218..a52cac96f3 100644
--- a/tests/control_plane_ts/vision_checkpoint.test.ts
+++ b/tests/control_plane_ts/vision_checkpoint.test.ts
@@ -417,6 +417,35 @@ test("semantic closeout retains the strict material vision checkpoint", () => {
   ]);
 });
 
+test("a bounded blocked retry does not invent a vision change", () => {
+  const blockedRetry = {
+    schema_version: "quota_blocked_retry_v0",
+    source: "turn_settlement",
+    todo_id: "todo_current001",
+    observed_at: "2026-09-24T14:00:00Z",
+    due_at: "2026-09-24T14:05:00Z",
+    resume_when: "resume_at:2026-09-24T14:05:00Z",
+  };
+  const blocked = buildVisionCheckpoint(finalizeRequest({
+    delivery_outcome: "outcome_gap", blocked_retry: blockedRetry,
+  }));
+  assert.equal(blocked.required, false);
+  assert.equal(blocked.satisfied, true);
+  assert.equal(blocked.decision, "not_required");
+  assert.deepEqual(blocked.triggers, []);
+
+  const changedAction = buildVisionCheckpoint(finalizeRequest({
+    delivery_outcome: "outcome_gap", blocked_retry: blockedRetry,
+    active_state_next_action_would_update: true,
+  }));
+  assert.equal(changedAction.required, true);
+  assert.deepEqual(changedAction.triggers, [{kind: "durable_next_action_update"}]);
+  assert.throws(() => buildVisionCheckpoint(finalizeRequest({
+    delivery_outcome: "outcome_gap",
+    blocked_retry: {...blockedRetry, todo_id: "todo_other"},
+  })), /blocked retry does not bind/);
+});
+
 test("explicit in-flight progress records continuity without vision repetition", () => {
   const result = buildVisionCheckpoint(finalizeRequest({
     delivery_boundary: "in_flight_continuation",

From 121bf22700095f7e25817292217091670e1efc5c Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Fri, 25 Sep 2026 00:31:00 +0800
Subject: [PATCH 5/8] fix(quota): invalidate stale scheduler run-index caches

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 loopx/cli_commands/quota_context.py           | 12 ++++++
 .../runtime/status_projection_cache.py        | 40 +++++++++++++++++++
 .../references/repair-patterns.md             |  2 +-
 .../test_quota_settlement_cli.py              |  5 ++-
 4 files changed, 56 insertions(+), 3 deletions(-)

diff --git a/loopx/cli_commands/quota_context.py b/loopx/cli_commands/quota_context.py
index db797371b1..df18f3e141 100644
--- a/loopx/cli_commands/quota_context.py
+++ b/loopx/cli_commands/quota_context.py
@@ -10,6 +10,7 @@
 )
 from ..control_plane.quota.error_codes import QuotaCommandValidationError
 from ..control_plane.runtime.status_projection_cache import (
+    cached_goal_run_index_is_current,
     load_status_projection_cache,
     resolve_status_projection_cache_runtime_root,
     write_status_projection_cache,
@@ -321,6 +322,17 @@ def prepare_quota_command_context(
             available_capabilities=args.available_capabilities,
             agent_lane_id=args.agent_id,
         )
+        if (
+            status_payload is not None
+            and command in QUOTA_SCHEDULER_COMMANDS
+            and status_goal_id
+            and not cached_goal_run_index_is_current(
+                status_payload, runtime_root=runtime_root, goal_id=status_goal_id,
+            )
+        ):
+            status_payload = None
+            cache_metadata["hit"] = False
+            cache_metadata["miss_reason"] = "run_index_changed"
     if status_payload is None:
         collector = status_collector or collect_status
         status_payload = collector(
diff --git a/loopx/control_plane/runtime/status_projection_cache.py b/loopx/control_plane/runtime/status_projection_cache.py
index 9954a64366..7d9dafe241 100644
--- a/loopx/control_plane/runtime/status_projection_cache.py
+++ b/loopx/control_plane/runtime/status_projection_cache.py
@@ -86,6 +86,46 @@ def status_projection_cache_path(runtime_root: Path, key: str) -> Path:
     return status_projection_cache_dir(runtime_root) / f"{key}.json"
 
 
+def cached_goal_run_index_is_current(
+    payload: dict[str, Any], *, runtime_root: Path, goal_id: str
+) -> bool:
+    """Fence scheduler cache reads against the durable Goal Run index.
+
+    The status projection stores the index digest computed over raw index
+    bytes. Hashing those bytes avoids reparsing run artifacts on a cache hit,
+    while any new blocked-settlement or other Run forces a fresh projection.
+    """
+
+    if not goal_id or goal_id in {".", ".."} or Path(goal_id).name != goal_id:
+        return False
+    history = payload.get("run_history")
+    goals = history.get("goals") if isinstance(history, dict) else None
+    goal = next(
+        (
+            item
+            for item in goals
+            if isinstance(item, dict) and item.get("id") == goal_id
+        ),
+        None,
+    ) if isinstance(goals, list) else None
+    if not isinstance(goal, dict) or "index_digest" not in goal:
+        return False
+    expected = goal["index_digest"]
+    if expected is not None and not isinstance(expected, str):
+        return False
+    index_path = runtime_root / "goals" / goal_id / "runs" / "index.jsonl"
+    digest = hashlib.sha256()
+    try:
+        with index_path.open("rb") as stream:
+            for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+                digest.update(chunk)
+    except FileNotFoundError:
+        return expected is None
+    except OSError:
+        return False
+    return digest.hexdigest() == expected
+
+
 def status_projection_cache_metadata(
     *,
     registry_path: Path,
diff --git a/skills/loopx-self-repair/references/repair-patterns.md b/skills/loopx-self-repair/references/repair-patterns.md
index 271caa8e79..77f8339376 100644
--- a/skills/loopx-self-repair/references/repair-patterns.md
+++ b/skills/loopx-self-repair/references/repair-patterns.md
@@ -32,7 +32,7 @@ teaches a reusable control-plane lesson.
 | `pending_capability_intent_projection_gap` | A durable post-writeback sidecar says `intent_recorded`, but quota remains `monitor_quiet_skip` or `terminal_no_followup`; no governed executor wakes, so no local artifact or exact approval Todo appears. | Exact Goal/Agent sidecar, intent schema and authority, consumption receipt, quota interaction contract, generated-artifact count, approval-Todo count, and external-effect count. | The producer journal was durable but no read model projected unconsumed capability intents into quota arbitration. Terminal or quiet lifecycle state therefore hid required capability work forever. | Add a provider-neutral TypeScript-validated pending-intent interaction slot, let the opted-in capability read only exact eligible sidecars, and give the pending action precedence over quiet/terminal routes. Consume it through an idempotent local executor that freezes checked artifacts and one digest-bound user gate; keep external delivery unauthorized and suppress consumed intents on replay. |
 | `approved_capability_delivery_successor_gap` | A governed report or other capability payload is frozen and its exact user gate is approved, but quota returns quiet terminal/monitor state and no provider action runs. A later caller may also improvise a destination or default sender because the approval receipt contains no executable route. | Frozen generation/consumption receipt, completed gate decision and scope, linked agent Todo, required decision scopes before/after approval, Goal Channel binding, selected Todo, provider effect and exact sender/destination readback. | The consumer created only a user gate. Approval recorded authority but had no explicitly linked, typed agent successor for Todo/quota selection, so the external effect existed only in chat memory or caller convention. | Before the gate, create one blocked agent successor bound to the frozen generation, exact decision scope, provider capability, and safe write scope; link the gate with `unblocks_todo_id`. Approval must atomically consume only that scope and resume the successor. The provider then resolves route and sender from the durable project binding, rejects caller overrides/default fallbacks, and records success only after native effect readback. |
 | `host_closeout_presentation_lookup_gap` | A prior heartbeat remains in closeout recovery after a successful blocked/deferred/completed Todo writeback; adding unrelated Todos changes recovery. | Receipt-bound Todo id, exact `todo list --todo-id` readback, compact quota visibility lanes, same-Turn retry and quota-spend count. | Recovery treated absence from a bounded presentation list as absence of a durable lifecycle transition. | Resolve the exact receipt-bound Todo through the existing provider-aware read path before evaluating closeout; preserve provider errors, identity binding and no-spend recovery. Cover crowded versus small inventories and real legacy/File/SQLite CLI writes; never raise display limits or fabricate settlement receipts. |
-| `blocked_turn_retry_history_gap` | An exact blocked Turn can close without spending, but its open Todo is immediately selected again or loses its wait when other runs crowd the status display; a peer hard lease prevents the blocked agent from updating the Todo. | Exact Goal/Agent/Todo/Turn receipt, peer lease, canonical Todo validator, persisted retry due time, full run index versus compact recent runs, next-Turn selection and spend count. | The settlement had no bounded retry authority, or quota selection reconstructed it from a truncated display list and tried to mutate a peer-gated Todo. | Persist one bounded retry on the blocked Turn receipt, retain it in the full-index semantic projection, and overlay it only for the same agent's selection until due or superseded by a newer work run. Keep the canonical Todo and terminal validator intact; prove replay, no spend, crowded history, and real File/SQLite hard-lease readback. |
+| `blocked_turn_retry_history_gap` | An exact blocked Turn can close without spending, but its open Todo is immediately selected again or loses its wait when other runs crowd the status display; a peer hard lease prevents the blocked agent from updating the Todo. | Exact Goal/Agent/Todo/Turn receipt, peer lease, canonical Todo validator, persisted retry due time, full run index versus compact recent runs and cached index digest, next-Turn selection and spend count. | The settlement had no bounded retry authority, or quota selection reconstructed it from truncated or stale status projection and tried to mutate a peer-gated Todo. | Persist one bounded retry on the blocked Turn receipt, retain it in the full-index semantic projection, and fence cached scheduler status against the current Run index before projecting it for the same agent. Keep the canonical Todo and terminal validator intact; prove replay, no spend, crowded history, stale cache, and real File/SQLite hard-lease readback. |
 | `host_closeout_history_scan_amplification` | Quota entry approaches or exceeds its runtime timeout only on a long-lived Goal, especially when many prior Turns already have valid closeouts. | Turn and receipt counts, rollout-log and runs-index sizes, candidate count, per-stage elapsed time, declared runtime budget, and the same synthetic settled history before and after repair. | Preflight rebuilt append-order recency with repeated list searches, then each candidate reread, reparsed, and rescanned the complete event and run history. A larger timeout hid the multiplicative work without bounding it. | Parse each persisted source once, retain strict-versus-tolerant read semantics in the shared snapshot, build append-order and exact Turn/effect indexes, and reduce every candidate from that snapshot. Restore the ordinary runtime budget and cover a sufficiently large all-settled history that must traverse every candidate; never truncate history, skip identity validation, or raise the timeout as the repair. |
 | `turn_replay_recovery_semantic_gap` | A real Turn resumes from a safe Journal prefix, but `inspect-journal` presents `replay_legal=false` as if recovery were forbidden; scheduler-only or saved-Host-result recovery is especially misleading. | Journal integrity fields, replay decision, executor-adopted recovery decision, completed phase prefix, complete typed settlement identity, authoritative selected-Todo lineage, conditional Host Session Binding check, prepared-effect presence, and bounded recovery outcome. | Effect-free terminal replay and effectful executor recovery were collapsed into one user-facing decision even though the executor used separate status/phase rules; initial repairs also omitted either the settlement identity or its binding-to-Turn cross-check, allowing a drifted goal, agent, or canonical Todo to reach a later effect. | Keep replay legality and Journal consistency distinct. Make one typed recovery decision the source used by both executor and inspection; validate and cross-bind the canonical settlement goal, agent, Turn instance, binding, and effect id before authorization, including the selected Todo or adaptive primary Todo; project continue/resume phase/Host reinvocation/reason plus only participating checks, and persist a bounded planned-versus-actual audit. Cover identity and binding drift with zero-provider-call regressions. Preserve the existing prepared-effect readback owner and do not claim general exactly-once semantics. |
 | `managed_chat_resume_turn_identity_gap` | A managed Chat Session resumes after its adapter becomes unhealthy, returns to `ready`, and clears `active_turn_id`, but the interrupted Turn remains nonterminal and unowned. | Pre-resume Session snapshot, post-prepare persisted Session, adapter health, interrupted Turn status and error code, final Session state. | Resume preparation cleared the persisted active Turn reference before adapter recovery, then recovery re-read only the mutated Session and lost the pre-resume Turn identity. | Carry the pre-mutation Turn id through the fresh closed-state check, terminalize that Turn as `failed/server_restarted` before restoring the Session, and cover unhealthy-adapter resume with a focused regression while retaining the per-Session lifecycle lock. |
diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py
index 09049bd6f9..bf381a15b1 100644
--- a/tests/control_plane/test_quota_settlement_cli.py
+++ b/tests/control_plane/test_quota_settlement_cli.py
@@ -1215,7 +1215,8 @@ def test_typed_blocked_retry_with_peer_hard_lease(
     )
     guard_rc, guard = _run_cli(
         registry_path, runtime, "quota", "should-run", "--codex-app",
-        "--goal-id", GOAL_ID, *binding, "--scan-path", str(project), cwd=project,
+        "--goal-id", GOAL_ID, *binding, "--scan-path", str(project),
+        "--write-projection-cache", cwd=project,
     )
     assert guard_rc == 0, guard
     assert guard["heartbeat_receipt"]["settlement_identity"]["todo_id"] == TODO_ID
@@ -1282,7 +1283,7 @@ def test_typed_blocked_retry_with_peer_hard_lease(
         registry_path, runtime, "quota", "should-run", "--codex-app",
         "--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
         "--turn-instance-id", f"turn-after-hard-lease-blocker-{provider}",
-        "--scan-path", str(project), cwd=project,
+        "--scan-path", str(project), "--use-projection-cache", cwd=project,
     )
     assert next_rc == 0, next_turn
     assert next_turn["effective_action"] != "unsettled_host_turn_recovery"

From 91887f66e11e10e6ed3bdd8e82718bd54b92c0c9 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Fri, 25 Sep 2026 00:37:02 +0800
Subject: [PATCH 6/8] test(quota): preserve scheduler cache hits before run
 changes

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 loopx/control_plane/runtime/status_projection_cache.py | 2 +-
 tests/control_plane/test_quota_settlement_cli.py       | 8 ++++++++
 2 files changed, 9 insertions(+), 1 deletion(-)

diff --git a/loopx/control_plane/runtime/status_projection_cache.py b/loopx/control_plane/runtime/status_projection_cache.py
index 7d9dafe241..77dc52d103 100644
--- a/loopx/control_plane/runtime/status_projection_cache.py
+++ b/loopx/control_plane/runtime/status_projection_cache.py
@@ -123,7 +123,7 @@ def cached_goal_run_index_is_current(
         return expected is None
     except OSError:
         return False
-    return digest.hexdigest() == expected
+    return f"sha256:{digest.hexdigest()}" == expected
 
 
 def status_projection_cache_metadata(
diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py
index bf381a15b1..cc125fd686 100644
--- a/tests/control_plane/test_quota_settlement_cli.py
+++ b/tests/control_plane/test_quota_settlement_cli.py
@@ -1220,6 +1220,13 @@ def test_typed_blocked_retry_with_peer_hard_lease(
     )
     assert guard_rc == 0, guard
     assert guard["heartbeat_receipt"]["settlement_identity"]["todo_id"] == TODO_ID
+    cached_rc, cached_guard = _run_cli(
+        registry_path, runtime, "quota", "should-run", "--codex-app",
+        "--goal-id", GOAL_ID, *binding, "--scan-path", str(project),
+        "--use-projection-cache", cwd=project,
+    )
+    assert cached_rc == 0, cached_guard
+    assert cached_guard["status_projection_cache"]["hit"] is True
 
     lease_rc, lease = _run_cli(
         registry_path, runtime, "task-lease", "acquire", "--goal-id", GOAL_ID,
@@ -1287,6 +1294,7 @@ def test_typed_blocked_retry_with_peer_hard_lease(
     )
     assert next_rc == 0, next_turn
     assert next_turn["effective_action"] != "unsettled_host_turn_recovery"
+    assert next_turn["status_projection_cache"]["miss_reason"] == "run_index_changed"
     assert (next_turn.get("selected_todo") or {}).get("todo_id") != TODO_ID
 
 

From 9f952192641ee87a47df534366be0cddbfba1127 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Fri, 25 Sep 2026 00:40:15 +0800
Subject: [PATCH 7/8] docs(quota): disclose blocked Turn no-spend closeout

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 docs/quota-allocation.md | 32 ++++++++++++++++++++++++--------
 1 file changed, 24 insertions(+), 8 deletions(-)

diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md
index faebc62606..27ab1bde7c 100644
--- a/docs/quota-allocation.md
+++ b/docs/quota-allocation.md
@@ -135,12 +135,17 @@ Turn-scoped `refresh-state` and `quota spend-slot` expose
 `settlement_progress` from the TypeScript receipt readback. The states are
 `identity_required`, `writeback_required`, `writeback_receipt_required`,
 `spend_required`, `spend_receipt_required`, and `settled`. A durable run without
-its matching receipt is incomplete. `settled` certifies this writeback/spend
-chain; Todo completion and Goal acceptance retain their separate checks.
-
-After verified writeback, `settlement_owed.command` carries the original Goal,
-Agent, Todo or replan obligation, Turn, registry/runtime route and spend source.
-Execute it unchanged. In `spend_receipt_required`, the same idempotent spend
+its matching receipt is incomplete. Ordinarily, `settled` certifies the
+writeback/spend chain. An exact typed blocked writeback with a bounded retry
+instead sets `closeout_kind=typed_blocked_writeback_no_spend` and settles the
+Turn without a quota debit. Todo completion and Goal acceptance retain their
+separate checks in both cases.
+
+When a quota spend remains owed after verified writeback,
+`settlement_owed.command` carries the original Goal, Agent, Todo or replan
+obligation, Turn, registry/runtime route and spend source. Execute it unchanged.
+The typed blocked no-spend closeout has no spend command. In
+`spend_receipt_required`, the same idempotent spend
 writer restores the receipt without another debit. Refresh and recovery never
 spend automatically. JSON and normal/recovery Markdown expose the same step.
 Rejected recovery reports observed progress without offering a spend command.
@@ -313,13 +318,24 @@ from `classification`. New writes should use one of:
 execution-profile hints only as a compatibility fallback for historical runs;
 new control-plane decisions should be driven by the enum above.
 
-An `outcome_gap` does not become delivery progress. It may settle and spend one
-exact Todo-bound Turn only when the same writeback includes a
+An `outcome_gap` does not become delivery progress. A blocked writeback is
+eligible for exact Todo-bound Turn settlement only when it includes a
 `typed_progress_observation_v0` with `result_class=blocked`, the matching
 `work_item_id`, a stable `blocker_id`, and a non-empty array of stable
 `evidence_ids`. Missing schemas, prose-only blockers, malformed evidence, and
 Todo identity mismatches remain fail-closed.
 
+New Turn-bound blocked writebacks also require a bounded retry on the same
+unfinished advancement Todo. A legacy Todo must have a pending
+`resume_when=resume_at:<timezone-aware-time>` due in 1–30 minutes. With promoted
+File/SQLite authority, an open Todo without its own resume condition can use
+a five-minute retry stored on the committed Turn instead; the peer-gated Todo
+and its completion validator stay unchanged. That exact writeback settles the
+Turn without spending quota. The retry suppresses only the blocked Todo for
+the same Agent until due or superseded by newer work, so independent eligible
+Todos can still be selected. Historical blocked writebacks without a bounded
+retry retain their prior spend readback; no debit is retroactively erased.
+
 `quota should-run` also separates long-running observation from work that should
 advance the selected goal. When the selected goal's current projection is a
 dependency-only observation, the payload includes `work_lane_contract` with

From d2c81995e0cfdbef3706e5c6840e8025fd41d146 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Fri, 25 Sep 2026 00:48:27 +0800
Subject: [PATCH 8/8] test(quota): keep later Todos outside blocked retry scope

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 .../test_quota_blocked_retry_projection.py    | 28 +++++++++++++++++++
 1 file changed, 28 insertions(+)

diff --git a/tests/control_plane/test_quota_blocked_retry_projection.py b/tests/control_plane/test_quota_blocked_retry_projection.py
index c84665dbdd..23c5825dce 100644
--- a/tests/control_plane/test_quota_blocked_retry_projection.py
+++ b/tests/control_plane/test_quota_blocked_retry_projection.py
@@ -161,6 +161,34 @@ def test_turn_owned_retry_is_selection_only_and_expires() -> None:
     assert projected["resume_when"] == retry["resume_when"]
     assert projected["resume_ready"] is False
     assert "resume_when" not in original
+    new_subject = {
+        **status,
+        "attention_queue": {
+            "items": [
+                {
+                    **status["attention_queue"]["items"][0],
+                    "agent_todos": {
+                        "items": [
+                            original,
+                            {
+                                "todo_id": "todo_created_after_block",
+                                "task_class": "advancement_task",
+                                "status": "open",
+                            },
+                        ]
+                    },
+                }
+            ],
+        },
+    }
+    new_subject_rows = overlay_active_turn_retries(
+        new_subject,
+        goal_id="goal-a",
+        agent_id="agent-a",
+        observed_at="2026-09-24T14:02:00Z",
+    )["attention_queue"]["items"][0]["agent_todos"]["items"]
+    assert new_subject_rows[0]["resume_ready"] is False
+    assert "resume_when" not in new_subject_rows[1]
     assert (
         overlay_active_turn_retries(
             status,