loopx-project · huangruiteng · Sep 21, 2026 · Sep 21, 2026 · Sep 21, 2026 · Sep 21, 2026
@@ -304,9 +304,11 @@ candidates. A dormant or non-resumable lane is projected under
 That blocked diagnostic does not replace the coordinator's own runnable lane or
 re-arm an activation obligation on every heartbeat. If the coordinator also
 has no in-scope runnable fallback, the final interaction mode is
-`peer_coordination_blocked`: schedulers return the bundle to its owner and stop
-the recurring heartbeat until peer capability/readiness, coordinator
-configuration, or the coordinator's own work frontier materially changes.
+`peer_coordination_blocked`: schedulers keep a no-spend observer alive with a
+10/20/30/60 minute stateful backoff. Peer capability/readiness, coordinator
+configuration, reassignment, or the coordinator's own work frontier changes
+the reset identity and restores the initial cadence; this recoverable state
+does not pause or delete the recurring heartbeat.
 
 Disable registered-peer coordination without changing peer registration or
 child-worker policy:

@@ -340,7 +340,7 @@ stateDiagram-v2
 | --- | --- | --- | --- |
 | `run_now` | `active_work` | 3 / 10 minutes | Work or repair must be attempted. |
 | `backoff_waiting_for_user` | `human_gate` | 30 / 120 minutes | Concrete user/controller action is next. |
-| `backoff_until_reassigned` | `agent_scope_wait` | 10 / 60 minutes, progression 10/20/30/60 | Handoff owner or reassignment may unblock this agent. |
+| `backoff_until_reassigned` | `agent_scope_wait` or `peer_coordination_wait` | 10 / 60 minutes, progression 10/20/30/60 | Handoff owner, peer readiness, coordinator configuration, reassignment, or new local work may unblock this agent. |
 | `backoff_until_material_transition` | `monitor_wait` | 15 / 60 minutes | Monitor-only liveness without compute spend. |
 | `backoff_until_fresh_evidence` | `unchanged_noop` | 60 / 240 minutes | Wait for fresh mapped or post-handoff evidence. |
 | `backoff_until_state_change` | `quiet_wait` | 30 / 120 minutes | No specific user/monitor path is projected. |

@@ -753,6 +753,14 @@ gap, autonomy blocker, or replan obligation fail closed. This keeps recurring
 controllers alive during ordinary waits while honoring an explicit completed
 goal shutdown without another quota-spending turn.
 
+An explicit `peer_coordination_blocked` decision is a recoverable typed wait,
+not a terminal host stop. It keeps the recurring heartbeat alive without
+spending quota and uses the existing TypeScript-owned stateful backoff
+transition with a 10/20/30/60 minute progression. Peer activation capability,
+peer runtime readiness, coordinator configuration, or newly projected local
+work changes the reset identity and restores the initial cadence. Goal stopped,
+quota paused, and validated terminal no-follow-up remain the stop cases.
+
 An individual registered peer can instead be put in `monitor_only` work mode:
 
 ```bash

@@ -267,8 +267,13 @@ def assert_ready_deferred_p0_preempts_open_p1_for_successor_replan() -> None:
         "top_ready_todo_id"
     ] == READY_DEFERRED_ID, quota_payload
 
-    for item in agent_todos["deferred_items"]:
-        item["priority"] = "P1"
+    # The bounded display lane and the lossless candidate lane are two
+    # projections of the same canonical Todo, and the explicit candidate lane
+    # owns the row when both carry it. Move the priority in every projection so
+    # the fixture still models one real Todo instead of two different ones.
+    for lane in ("deferred_items", "deferred_resume_candidates"):
+        for item in agent_todos.get(lane, []):
+            item["priority"] = "P1"
     equal_priority = build_quota_should_run(
         status_payload(agent_todos, next_action=FALLBACK_ACTION),
         goal_id=GOAL_ID,

@@ -286,15 +286,30 @@ def main() -> int:
         assert blocked_turn["interaction_contract"]["mode"] == (
             "peer_coordination_blocked"
         ), blocked_turn
+        assert blocked_turn["automation_liveness"]["keep_active"] is True, (
+            blocked_turn
+        )
+        assert blocked_turn["automation_liveness"]["pause_allowed"] is False, (
+            blocked_turn
+        )
+        assert blocked_turn["automation_liveness"]["automation_action"] == (
+            "keep_active_peer_coordination_backoff"
+        ), blocked_turn
         assert blocked_turn["scheduler_hint"]["action"] == (
-            "return_to_owner_until_material_change"
+            "backoff_until_reassigned"
         ), blocked_turn
-        assert blocked_turn["scheduler_hint"]["codex_app"]["host_action"] == (
-            "pause_or_delete_current_heartbeat"
+        assert blocked_turn["scheduler_hint"]["cadence_class"] == (
+            "peer_coordination_wait"
         ), blocked_turn
-        assert blocked_turn["scheduler_hint"]["unchanged_poll"][
+        assert blocked_turn["scheduler_hint"]["codex_app"][
+            "recommended_interval_minutes"
+        ] == 10, blocked_turn
+        assert blocked_turn["scheduler_hint"]["codex_app"][
+            "example_progression_minutes"
+        ] == [10, 20, 30, 60], blocked_turn
+        assert blocked_turn["scheduler_hint"]["unchanged_poll"]["limits"][
             "local_scheduler"
-        ] == "stop", blocked_turn
+        ] == 3, blocked_turn
         assert blocked_turn["interaction_contract"]["cli_channel"][
             "spend_after_validation"
         ] is False, blocked_turn

@@ -13,7 +13,7 @@
 
 class SchedulerDisposition(str, Enum):
     TERMINAL_STOP = "terminal_stop"
-    PEER_COORDINATION_STOP = "peer_coordination_stop"
+    PEER_COORDINATION_WAIT = "peer_coordination_wait"
     AGENT_MONITOR_ONLY_WAIT = "agent_monitor_only_wait"
     ACTIVE_WORK = "active_work"
     AGENT_SCOPE_WAIT = "agent_scope_wait"
@@ -80,7 +80,7 @@ def _classify_disposition(
     if mode == "terminal_no_followup":
         return SchedulerDisposition.TERMINAL_STOP, mode
     if mode == "peer_coordination_blocked":
-        return SchedulerDisposition.PEER_COORDINATION_STOP, mode
+        return SchedulerDisposition.PEER_COORDINATION_WAIT, mode
     if mode == "agent_monitor_only":
         return SchedulerDisposition.AGENT_MONITOR_ONLY_WAIT, mode
     if user_required and not must_attempt:

@@ -99,6 +99,20 @@ def build_automation_liveness(payload: dict[str, Any]) -> dict[str, Any]:
             "next_trigger": "explicit goal resume or newly projected work",
             "spend_policy": "no quota spend for terminal automation shutdown",
         }
+    if effective_action == EffectiveAction.PEER_COORDINATION_BLOCKED.value:
+        return {
+            **base,
+            "automation_action": "keep_active_peer_coordination_backoff",
+            "reason": (
+                "peer coordination is recoverable from asynchronous peer, runtime, "
+                "configuration, or local-frontier changes"
+            ),
+            "next_trigger": (
+                "peer activation capability, peer runtime readiness, coordinator "
+                "configuration, or newly projected local work"
+            ),
+            "spend_policy": "no quota spend while explicit peer coordination is blocked",
+        }
     if (
         effective_action == EffectiveAction.MONITOR_QUIET_SKIP.value
         or recommended_mode == "monitor_quiet_until_material_transition"

@@ -1313,26 +1313,6 @@ def build_scheduler_hint(
             unchanged_spend_policy="no quota spend for terminal loop stop",
         )
 
-    if arbitration.disposition == SchedulerDisposition.PEER_COORDINATION_STOP:
-        cadence_class = "peer_coordination_blocked"
-        return _build_scheduler_stop_hint(
-            execution_context=execution_context,
-            action="return_to_owner_until_material_change",
-            cadence_class=cadence_class,
-            reason_code=arbitration.reason_code,
-            reason=(
-                "explicit peer coordination has no executable peer lane or local "
-                "fallback; recurring polling must stop until its inputs change"
-            ),
-            spend_policy=("no quota spend while explicit peer coordination is blocked"),
-            resume_trigger=(
-                "peer activation capability, peer runtime readiness, coordinator "
-                "configuration, or newly projected local work"
-            ),
-            ssh_goal_runtime_action="return_to_owner",
-            unchanged_spend_policy=("no quota spend for blocked coordination stop"),
-        )
-
     builder = _SchedulerHintBuilder(
         payload=payload,
         execution_context=execution_context,
@@ -1344,6 +1324,21 @@ def build_scheduler_hint(
         codex_app_automation_id=codex_app_automation_id,
         include_detail=include_detail,
     )
+    if arbitration.disposition == SchedulerDisposition.PEER_COORDINATION_WAIT:
+        return builder.build(
+            action="backoff_until_reassigned",
+            cadence_class="peer_coordination_wait",
+            reason=(
+                "explicit peer coordination has no executable peer lane or local "
+                "fallback; keep a bounded no-spend observer because peer readiness, "
+                "configuration, or the local frontier can change asynchronously"
+            ),
+            codex_interval=10,
+            codex_max=60,
+            cli_limit=3,
+            claude_limit=3,
+            cadence_progression_override=[10, 20, 30, 60],
+        )
     if arbitration.disposition == SchedulerDisposition.AGENT_MONITOR_ONLY_WAIT:
         return builder.build(
             action="backoff_agent_monitor_only",

@@ -66,10 +66,19 @@ function ordered(items: readonly Item[]): Item[] {
   return [...items].sort((a, b) => a.priority - b.priority || a.index - b.index);
 }
 
+function itemIdentity(entry: Item): string {
+  // A canonical Todo id owns identity across compact/display and lossless
+  // candidate projections. Fall back to the legacy presentation identity only
+  // for rows that predate typed Todo ids.
+  return entry.id
+    ? JSON.stringify(["todo", entry.id])
+    : JSON.stringify(["projection", entry.payload.text, entry.payload.index]);
+}
+
 function unique(items: readonly Item[]): Item[] {
   const seen = new Set<string>();
   return items.filter((entry) => {
-    const key = JSON.stringify([entry.id ?? "", entry.payload.text, entry.payload.index]);
+    const key = itemIdentity(entry);
     if (seen.has(key)) return false;
     seen.add(key);
     return true;
@@ -167,12 +176,24 @@ function decodeSources(value: unknown): Record<SourceKey, Item[]> {
 
 function deferredPlan(sources: Record<SourceKey, Item[]>, capabilities: unknown) {
   let deferredItems = deferred(sources.deferred_items.length ? sources.deferred_items : sources.items);
-  let candidates = deferred(sources.deferred_resume_candidates).filter((entry) => entry.payload.resume_ready === true);
+  const projectedCandidates = deferred(sources.deferred_resume_candidates);
+  let candidates = projectedCandidates.filter((entry) => entry.payload.resume_ready === true);
   let capacityFields: JsonObject | null = null;
   if (capabilities !== null) {
-    deferredItems = resolveCapacity(deferredItems, sources.items,
+    // deferred_items is a bounded presentation lane while
+    // deferred_resume_candidates retains semantically selected rows that may
+    // fall outside that display window. Re-evaluate their union so supplying
+    // runtime capabilities cannot erase an already projected ready successor.
+    // Put the explicit candidate lane first so its lossless payload wins when
+    // the same Todo is also present in a compact display lane.
+    const displayIdentities = new Set(deferredItems.map(itemIdentity));
+    const reevaluated = resolveCapacity(ordered(unique([
+      ...projectedCandidates,
+      ...deferredItems,
+    ])), sources.items,
       requireStringArray(capabilities, "available_capabilities"));
-    candidates = deferredItems.filter((entry) => entry.payload.resume_ready === true);
+    deferredItems = reevaluated.filter((entry) => displayIdentities.has(itemIdentity(entry)));
+    candidates = reevaluated.filter((entry) => entry.payload.resume_ready === true);
     capacityFields = { deferred_items: payloads(deferredItems), deferred_resume_candidates: payloads(candidates) };
     // Existing summary readers treat an empty explicit deferred lane as absent,
     // including after capacity resolution; retain that compatibility fallback.

@@ -485,7 +485,7 @@
         "operator_gate_notify": "Existing result of quota_effective_action; previously missed because the literal scan did not inspect declared return functions.",
         "operator_inbox_material_review_due": "Captured operator-inbox material is pending bounded review, and normal delivery is allowed for that review.",
         "outcome_floor_recovery": "Normal delivery is refused but recovery delivery is allowed, through the quota safe bypass of kind outcome_floor_recovery.",
-        "peer_coordination_blocked": "Peer coordination blocks this lane. The scheduler stops it rather than holding it in a monitor wait.",
+        "peer_coordination_blocked": "Peer coordination blocks this lane. The scheduler keeps a no-spend observer alive with bounded stateful backoff because peer readiness, coordinator configuration, reassignment, or local work can change asynchronously.",
         "quota_skip": "The final fallback of quota_effective_action: no delivery is allowed, no repair applies, and no more specific blocked state matched.",
         "runtime_user_gate_projection_repair": "A runtime capability user-gate projection needs repair; recommended_mode repair_user_gate_projection, blocked scope user_gate_projection.",
         "scoped_user_gate_fallback": "A scoped user-gate fallback applies while replan decisions are not allowed. It replaces quota_skip, monitor_quiet_skip or an absent action, and obliges one non-gated fallback segment after the user-gate notice.",

@@ -397,6 +397,23 @@ def test_app_heartbeat_settlement_keeps_automation_active_until_terminal(
     assert terminal["keep_active"] is False
 
 
+def test_peer_coordination_block_keeps_recoverable_heartbeat_active() -> None:
+    liveness = build_automation_liveness(
+        {
+            "effective_action": "peer_coordination_blocked",
+            "heartbeat_recommendation": {
+                "recommended_mode": "peer_coordination_blocked"
+            },
+            "execution_obligation": {"must_attempt_work": False},
+        }
+    )
+
+    assert liveness["keep_active"] is True
+    assert liveness["pause_allowed"] is False
+    assert liveness["automation_action"] == "keep_active_peer_coordination_backoff"
+    assert "peer activation capability" in liveness["next_trigger"]
+
+
 def test_goal_runtime_projects_typed_immediate_continuation() -> None:
     context = scheduler_execution_context_for_runtime_profile(
         SchedulerRuntimeProfile.ARK_MANAGED_AGENT_GOAL

@@ -171,8 +171,8 @@ def _payload(
                 delivery_allowed=False,
                 quiet_noop_allowed=True,
             ),
-            SchedulerDisposition.PEER_COORDINATION_STOP,
-            "peer_coordination_blocked",
+            SchedulerDisposition.PEER_COORDINATION_WAIT,
+            "peer_coordination_wait",
         ),
     ],
 )
@@ -502,7 +502,7 @@ def test_terminal_contract_with_open_action_fails_closed() -> None:
     )
 
 
-def test_blocked_peer_coordination_returns_to_owner_without_polling() -> None:
+def test_blocked_peer_coordination_uses_stateful_backoff() -> None:
     payload = _payload(
         mode="peer_coordination_blocked",
         should_run=False,
@@ -517,11 +517,13 @@ def test_blocked_peer_coordination_returns_to_owner_without_polling() -> None:
         agent_scope_frontier_actions=AGENT_SCOPE_ACTIONS,
     )
 
-    assert hint["action"] == "return_to_owner_until_material_change"
-    assert hint["codex_app"]["host_action"] == ("pause_or_delete_current_heartbeat")
-    assert hint["codex_app"]["host_action_required"] is True
-    assert hint["unchanged_poll"]["local_scheduler"] == "stop"
-    assert hint["unchanged_poll"]["final_quota_replan_check_enabled"] is False
+    assert hint["action"] == "backoff_until_reassigned"
+    assert hint["cadence_class"] == "peer_coordination_wait"
+    assert hint["codex_app"]["host_action"] == "update_current_heartbeat_rrule"
+    assert hint["codex_app"]["stateful_backoff"]["apply_needed"] is True
+    assert hint["codex_app"]["recommended_interval_minutes"] == 10
+    assert hint["codex_app"]["example_progression_minutes"] == [10, 20, 30, 60]
+    assert hint["unchanged_poll"]["final_quota_replan_check_enabled"] is True
 
 
 def test_structurally_invalid_contract_fails_closed() -> None:

@@ -44,6 +44,55 @@ test("capacity evaluation and claim lanes use one snapshot without mutating it",
   assert.equal(unavailable.current_agent_deferred_resume_count, 0);
 });
 
+test("capability reevaluation retains ready candidates outside the deferred display bound", () => {
+  const visible = fact("todo_visible", { claim: "agent-a" });
+  const ready = fact("todo_ready", {
+    claim: "agent-a", ready: true, ready_truthy: true,
+  });
+  (ready.payload as JsonObject).resume_ready = true;
+  const input = request({ available_capabilities: ["compiler"] });
+  (input.sources as JsonObject).deferred_items = [visible];
+  (input.sources as JsonObject).deferred_resume_candidates = [ready];
+
+  const result = projectTodoResumePlanning(input);
+  const lanes = result.deferred_lanes as JsonObject;
+
+  assert.deepEqual(
+    (lanes.current_agent_deferred_resume_candidates as JsonObject[]).map(
+      (row) => row.todo_id,
+    ),
+    ["todo_ready", "todo_visible"],
+  );
+  assert.equal(lanes.current_agent_deferred_resume_count, 2);
+  assert.deepEqual(
+    ((result.capacity_fields as JsonObject).deferred_items as JsonObject[]).map(
+      (row) => row.todo_id,
+    ),
+    ["todo_visible"],
+  );
+});
+
+test("explicit candidate payload wins when the display lane has the same Todo id", () => {
+  const display = fact("todo_shared", { claim: "agent-a" });
+  (display.payload as JsonObject).text = "Compact display copy";
+  const candidate = fact("todo_shared", {
+    claim: "agent-a", ready: true, ready_truthy: true,
+  });
+  (candidate.payload as JsonObject).text = "Lossless candidate copy";
+  (candidate.payload as JsonObject).resume_ready = true;
+  const input = request({ available_capabilities: ["compiler"] });
+  (input.sources as JsonObject).deferred_items = [display];
+  (input.sources as JsonObject).deferred_resume_candidates = [candidate];
+
+  const result = projectTodoResumePlanning(input);
+  const lanes = result.deferred_lanes as JsonObject;
+  const candidates = lanes.current_agent_deferred_resume_candidates as JsonObject[];
+
+  assert.equal(candidates.length, 1);
+  assert.equal(candidates[0].todo_id, "todo_shared");
+  assert.equal(candidates[0].text, "Lossless candidate copy");
+});
+
 test("a large wait source retains total counts independently from the display bound", () => {
   const input = request();
   const rows = Array.from({ length: 257 }, (_, i) => fact(`todo_wait_${i}`));