fix(quota): require compute allowance for fresh replan admission

docs/quota-allocation.md

@@ -90,6 +90,22 @@ depending on the executor:
 
 ## Goal Review Cadence
 
+Fresh autonomous replan work shares the Goal's compute window. Once
+`spent_slots >= allowed_slots`, its extra allowance is **zero**: `quota should-run`
+returns the existing typed `throttled_skip` action rather than admitting a
+replan wakeup. A pending replan remains pending until rolling-window accounting
+restores allowance or the owner increases the existing compute budget. This
+also applies when a focus or external wait masks the exhausted window. Replan
+still precedes monitor quiet and lane waits while compute remains available.
+Owner Stop/zero-compute pauses and exact-identity owed settlement, including
+qualified no-debit closeouts, retain their existing owners and rules.
+
+新的自主重规划与 Goal 共用算力窗口。`spent_slots >= allowed_slots` 时，额外额度为
+**零**；`quota should-run` 使用现有的 typed `throttled_skip`，不再为待处理的重规划
+唤醒模型。等待滚动窗口恢复额度，或 owner 提高已有算力预算后重新判断；focus 或
+外部等待不会豁免这一限制。有剩余额度时，重规划仍优先于 monitor quiet 和 lane wait。
+Owner Stop／零算力暂停、原身份欠账结算与合格的免扣额收尾保持原有规则。
+
 The product and benchmark runners default to review after **6 settled effective
 work Turns** in both standard and fine-grained modes. This changes the previous
 product threshold of five Turns and benchmark threshold of three Turns. Review

loopx/control_plane/quota/decision_summary.py

@@ -286,6 +286,8 @@ def resolve_quota_run_decision(
         "self_repair_allowed": self_repair_allowed,
         "repair_effective_action": str(repair_action or ""),
         "state": state, "reason": reason,
+        "quota_allowed_slots": str(quota["allowed_slots"]) if quota.get("allowed_slots") is not None else None,
+        "quota_spent_slots": str(quota["spent_slots"]) if quota.get("spent_slots") is not None else None,
         "quota_focus_wait": bool(quota.get("focus_wait")),
         "capability_gate": _guard_strings(capability_gate, ("action", "reason")),
         "capability_monitor_fallback": bool(capability_monitor_fallback),

loopx/control_plane/quota/run_decision.ts

@@ -11,6 +11,8 @@ interface RunFacts {
   state: string;
   reason: string;
   quota_focus_wait: boolean;
+  quota_allowed_slots: bigint | null;
+  quota_spent_slots: bigint | null;
   capability_gate: JsonObject | null;
   capability_monitor_fallback: boolean;
   workspace_guard: JsonObject | null;
@@ -62,6 +64,8 @@ function decodeRunFacts(params: JsonObject): RunFacts {
     repair_effective_action: text(params.repair_effective_action, "repair_effective_action"),
     state: text(params.state, "state"), reason: text(params.reason, "reason"),
     quota_focus_wait: requireBoolean(params.quota_focus_wait, "quota_focus_wait"),
+    quota_allowed_slots: quotaCount(params.quota_allowed_slots, "quota_allowed_slots"),
+    quota_spent_slots: quotaCount(params.quota_spent_slots, "quota_spent_slots"),
     capability_gate: optionalRecord(params.capability_gate, "capability_gate"),
     capability_monitor_fallback: requireBoolean(params.capability_monitor_fallback, "capability_monitor_fallback"),
     workspace_guard: optionalRecord(params.workspace_guard, "workspace_guard"),
@@ -85,6 +89,14 @@ function nonNegativeCount(value: unknown, label: string): number {
   return Number(value);
 }
 
+function quotaCount(value: unknown, label: string): bigint | null {
+  if (value === null || value === undefined) return null;
+  // Decimal transport preserves Python's integer precision without rounding quota boundaries.
+  if (typeof value !== "string" || !/^\d+$/.test(value))
+    throw new EffectRuntimeRequestError(`${label} must be a non-negative integer string`);
+  return BigInt(value);
+}
+
 function guardReason(guard: JsonObject | null, fallback: string): string {
   return (guard ? text(guard.reason ?? "", "guard.reason") : "") || fallback;
 }
@@ -136,8 +148,18 @@ function quotaEffectiveAction(facts: RunFacts, decision: RunDecision): string {
 
 function applyGoalOverrides(facts: RunFacts, decision: RunDecision): void {
   const inboxPriority = facts.inbox_reply_due || facts.inbox_material_review_due;
-  decision.replan_decision_allowed = !inboxPriority && facts.replan_required && facts.replan_applies
+  const replanSelected = !inboxPriority && facts.replan_required && facts.replan_applies
     && facts.goal_health_ok && !facts.workspace_guard && !facts.automation_prompt_upgrade_required;
+  const quotaExhausted = facts.state === "throttled" || (facts.quota_allowed_slots !== null
+    && facts.quota_spent_slots !== null && facts.quota_spent_slots >= facts.quota_allowed_slots);
+  decision.replan_decision_allowed = replanSelected && !quotaExhausted;
+  if (replanSelected && quotaExhausted) {
+    clearExecutionLanes(decision);
+    decision.should_run = false;
+    decision.effective_action = EffectiveAction.THROTTLED_SKIP;
+    decision.reason = "compute window exhausted; fresh replan has zero extra allowance; "
+      + "wait until the window resets or the owner raises the compute budget";
+  }
   if (decision.replan_decision_allowed) {
     decision.normal_delivery_allowed = false;
     decision.recovery_delivery_allowed = false;

tests/control_plane/test_replan_quota_exhaustion.py

@@ -0,0 +1,137 @@
+"""Fresh replan work needs remaining compute; settlement retains its own owner."""
+import json
+from pathlib import Path
+
+import pytest
+
+from loopx.control_plane.quota.live_decision import build_live_quota_should_run_decision
+from loopx.control_plane.testing.quota_fixtures import quota_status_payload, quota_todo_item
+
+
+GOAL = "quota-replan-fixture"
+AGENT = "fixture-agent"
+
+
+def _status(*, state="throttled", allowed=1, spent=1, compute=1):
+    return quota_status_payload(
+        goal_id=GOAL, status="active", recommended_action="Wait for the dependency.",
+        agent_todo_items=[quota_todo_item(
+            index=1, todo_id="todo_blocked_work", text="Wait for the dependency.",
+            status="open", resume_when="todo_done:todo_missing_dependency",
+            claimed_by=AGENT, task_class="advancement_task",
+        )],
+        quota_state=state,
+        quota_extra={"compute": compute, "allowed_slots": allowed, "spent_slots": spent},
+        coordination={"agent_model": "peer_v1", "registered_agents": [AGENT]},
+        latest_runs=[{
+            "classification": "state_refreshed", "generated_at": "2026-07-04T00:05:00Z",
+            "agent_id": AGENT, "progress_scope": "agent_lane",
+            "delivery_outcome": "outcome_progress",
+            "vision_checkpoint": {
+                "schema_version": "vision_checkpoint_v0", "agent_id": AGENT,
+                "required": True, "satisfied": False, "decision": "missing_required",
+                "triggers": [{"kind": "material_delivery_outcome", "delivery_outcome": "outcome_progress"}],
+                "required_resolution": ["write_agent_vision_patch", "record_unchanged_reason"],
+            },
+        }],
+    )
+
+
+def _read(status, tmp_path, **kwargs):
+    return build_live_quota_should_run_decision(
+        status, goal_id=GOAL, agent_id=AGENT, available_capabilities=["shell"],
+        include_scheduler_detail=False, codex_app_current_rrule=None,
+        registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime",
+        scheduler_execution_context={"host_surface": "generic_cli",
+            "scheduler_owner": "agent_cli_loop", "execution_mode": "interactive"},
+        **kwargs,
+    )
+
+
+@pytest.mark.parametrize("state", ["throttled", "focus_wait", "waiting"])
+def test_exhausted_window_does_not_admit_fresh_replan(tmp_path: Path, state: str):
+    for _ in range(2):
+        packet = _read(_status(state=state), tmp_path)
+        assert packet["should_run"] is False
+        assert packet["effective_action"] == "throttled_skip"
+        assert "replan" in packet["reason"] and "zero" in packet["reason"]
+        assert packet["scheduler_hint"]["action"] != "run_now"
+        assert packet["interaction_contract"]["agent_channel"]["must_attempt"] is False
+        assert packet["interaction_contract"]["agent_channel"]["delivery_allowed"] is False
+
+
+@pytest.mark.parametrize("allowed,spent", [(1, 0), (2, 1)])
+def test_window_reset_or_owner_increase_reopens_replan(tmp_path: Path, allowed: int, spent: int):
+    assert _read(_status(), tmp_path)["should_run"] is False
+    packet = _read(_status(state="eligible", allowed=allowed, spent=spent), tmp_path)
+    assert packet["should_run"] is True
+    assert packet["effective_action"] == "autonomous_replan_required"
+    assert packet["scheduler_hint"]["action"] == "run_now"
+
+
+def test_zero_compute_remains_an_owner_pause(tmp_path: Path):
+    packet = _read(_status(state="paused", compute=0, allowed=0), tmp_path)
+    assert packet["should_run"] is False
+    assert packet["scheduler_hint"]["action"] == "stop_until_explicit_resume"
+
+
+@pytest.mark.parametrize("spent,expected", [(0, True), (1, False)])
+def test_should_run_cli_uses_the_shared_admission(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture,
+    spent: int, expected: bool,
+):
+    from loopx.cli import main
+    from loopx.cli_commands import quota
+
+    status = _status(state="eligible" if expected else "throttled", spent=spent)
+    monkeypatch.setattr(quota, "collect_status", lambda **kwargs: status)
+    (tmp_path / "registry.json").write_text(json.dumps({"goals": [{
+        "id": GOAL, "repo": str(tmp_path), "state_file": "fixture.md",
+    }]}), encoding="utf-8")
+    (tmp_path / "fixture.md").write_text("# Synthetic Goal\n", encoding="utf-8")
+    code = main([
+        "--registry", str(tmp_path / "registry.json"),
+        "--runtime-root", str(tmp_path / "runtime"), "--format", "json",
+        "quota", "should-run", "--goal-id", GOAL, "--agent-id", AGENT,
+        "-H", "generic_cli", "-O", "agent_cli_loop", "-M", "interactive",
+    ])
+    assert code == 0
+    packet = json.loads(capsys.readouterr().out)
+    assert packet["should_run"] is expected
+    assert packet["effective_action"] == (
+        "autonomous_replan_required" if expected else "throttled_skip"
+    )
+
+
+def test_exhaustion_preserves_exact_owed_turn_recovery(tmp_path: Path):
+    from canonical_authority_fixture import initialize_canonical_source_fixture
+    from loopx.rollout_event_log import build_rollout_event
+
+    runtime = tmp_path / "runtime"
+    state = tmp_path / "fixture.md"
+    state.write_text(
+        "# Synthetic Goal\n\n## Agent Todo\n\n- [ ] Wait for the dependency.\n"
+        f"  <!-- loopx:todo todo_id=todo_blocked_work status=open claimed_by={AGENT} "
+        "task_class=advancement_task resume_when=todo_done:todo_missing_dependency -->\n",
+        encoding="utf-8",
+    )
+    (tmp_path / "registry.json").write_text(json.dumps({"goals": [{
+        "id": GOAL, "repo": str(tmp_path), "state_file": str(state),
+    }]}), encoding="utf-8")
+    initialize_canonical_source_fixture(runtime, GOAL, state_path=state, provider="sqlite")
+    event = build_rollout_event(
+        goal_id=GOAL, event_kind="quota_should_run", agent_id=AGENT,
+        todo_id="todo_blocked_work", run_id="prior-turn", status="normal_run",
+        summary="Original Turn requires settlement.", details={
+            "todo_id": "todo_blocked_work", "closeout_required": True,
+            "settlement_effect_id": f"{GOAL}:{AGENT}:todo_blocked_work:prior-turn",
+        },
+    )
+    log = runtime / "goals" / GOAL / "rollout-event-log.jsonl"
+    log.parent.mkdir(parents=True, exist_ok=True)
+    log.write_text(json.dumps(event) + "\n", encoding="utf-8")
+    packet = _read(_status(), tmp_path, turn_instance_id="current-turn")
+    assert packet["effective_action"] == "unsettled_host_turn_recovery"
+    assert packet["unsettled_host_turn_recovery"]["prior_turn_instance_id"] == "prior-turn"
+    assert packet["interaction_contract"]["agent_channel"]["delivery_allowed"] is False
+    assert packet["interaction_contract"]["cli_channel"]["spend_after_validation"] is False

tests/control_plane_ts/quota_run_decision.test.ts

@@ -15,11 +15,32 @@ for (const scenario of fixture.cases) {
 test("malformed admission facts cannot manufacture a runnable decision", () => {
   for (const patch of [{normal_delivery_allowed: "true"}, {replan_applies: null},
     {workspace_guard: "not a guard"}, {quota_focus_wait: 1}, {reason: null},
-    {task_orchestration_lane_count: -1}, {task_orchestration_lane_count: 1.5}]) {
+    {task_orchestration_lane_count: -1}, {task_orchestration_lane_count: 1.5},
+    {quota_allowed_slots: "-1"}, {quota_spent_slots: "1.5"}, {quota_spent_slots: 1}]) {
     assert.throws(() => resolveQuotaRunDecision({...fixture.base_facts, ...patch} as JsonObject));
   }
 });
 
+for (const state of ["throttled", "focus_wait", "waiting", "eligible"]) {
+  test(`fresh replan cannot exceed its compute window under ${state}`, () => {
+    const result = resolveQuotaRunDecision({...fixture.base_facts, state,
+      replan_required: true, replan_applies: true,
+      quota_allowed_slots: "1", quota_spent_slots: "1"});
+    assert.equal(result.should_run, false);
+    assert.equal(result.replan_decision_allowed, false);
+    assert.equal(result.normal_delivery_allowed, false);
+    assert.equal(result.effective_action, "throttled_skip");
+  });
+}
+
+test("quota boundaries retain integer precision and admit only remaining allowance", () => {
+  const facts = {...fixture.base_facts, replan_required: true, replan_applies: true,
+    quota_allowed_slots: "9007199254740993", quota_spent_slots: "9007199254740992"};
+  assert.equal(resolveQuotaRunDecision(facts).replan_decision_allowed, true);
+  assert.equal(resolveQuotaRunDecision({...facts,
+    quota_spent_slots: "9007199254740993"}).should_run, false);
+});
+
 for (const scenario of [
   {
     name: "adaptive counts child lanes",