diff --git a/docs/integrations/codex-subagent-orchestration.md b/docs/integrations/codex-subagent-orchestration.md index b758c1ae82..9f51b18a24 100644 --- a/docs/integrations/codex-subagent-orchestration.md +++ b/docs/integrations/codex-subagent-orchestration.md @@ -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: diff --git a/docs/product/core-control-plane/state-machine.md b/docs/product/core-control-plane/state-machine.md index 0a21e82184..b88511b548 100644 --- a/docs/product/core-control-plane/state-machine.md +++ b/docs/product/core-control-plane/state-machine.md @@ -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. | diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index 0d4df49b73..faebc62606 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -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 diff --git a/examples/control_plane/quota-resume-gated-open-todo-smoke.py b/examples/control_plane/quota-resume-gated-open-todo-smoke.py index cc5f3d4313..a5c68d8c24 100644 --- a/examples/control_plane/quota-resume-gated-open-todo-smoke.py +++ b/examples/control_plane/quota-resume-gated-open-todo-smoke.py @@ -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, diff --git a/examples/control_plane/task-orchestration-smoke.py b/examples/control_plane/task-orchestration-smoke.py index aa878277cb..05664b1e6a 100644 --- a/examples/control_plane/task-orchestration-smoke.py +++ b/examples/control_plane/task-orchestration-smoke.py @@ -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 diff --git a/loopx/control_plane/scheduler/arbitration.py b/loopx/control_plane/scheduler/arbitration.py index 36d8b887a7..74be6750a4 100644 --- a/loopx/control_plane/scheduler/arbitration.py +++ b/loopx/control_plane/scheduler/arbitration.py @@ -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: diff --git a/loopx/control_plane/scheduler/automation_liveness.py b/loopx/control_plane/scheduler/automation_liveness.py index ffd8dde899..a6d51f88b8 100644 --- a/loopx/control_plane/scheduler/automation_liveness.py +++ b/loopx/control_plane/scheduler/automation_liveness.py @@ -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" diff --git a/loopx/control_plane/scheduler/scheduler_hint.py b/loopx/control_plane/scheduler/scheduler_hint.py index f385ce5f05..5085ea625f 100644 --- a/loopx/control_plane/scheduler/scheduler_hint.py +++ b/loopx/control_plane/scheduler/scheduler_hint.py @@ -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", diff --git a/loopx/control_plane/todos/resume_planning.ts b/loopx/control_plane/todos/resume_planning.ts index 3391de346e..04169f1354 100644 --- a/loopx/control_plane/todos/resume_planning.ts +++ b/loopx/control_plane/todos/resume_planning.ts @@ -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(); 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 { function deferredPlan(sources: Record, 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. diff --git a/loopx/semantics/vocabulary_v0.json b/loopx/semantics/vocabulary_v0.json index a390dc626c..1805be97ff 100644 --- a/loopx/semantics/vocabulary_v0.json +++ b/loopx/semantics/vocabulary_v0.json @@ -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.", diff --git a/tests/control_plane/test_scheduler_execution_context.py b/tests/control_plane/test_scheduler_execution_context.py index b1f87d1daa..ad246061d8 100644 --- a/tests/control_plane/test_scheduler_execution_context.py +++ b/tests/control_plane/test_scheduler_execution_context.py @@ -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 diff --git a/tests/control_plane/test_scheduler_interaction_arbitration.py b/tests/control_plane/test_scheduler_interaction_arbitration.py index bcc6723c63..990a7fb8ec 100644 --- a/tests/control_plane/test_scheduler_interaction_arbitration.py +++ b/tests/control_plane/test_scheduler_interaction_arbitration.py @@ -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: diff --git a/tests/control_plane_ts/resume_planning.test.ts b/tests/control_plane_ts/resume_planning.test.ts index 208cd5d13f..39f23c7dfb 100644 --- a/tests/control_plane_ts/resume_planning.test.ts +++ b/tests/control_plane_ts/resume_planning.test.ts @@ -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}`));