fix(quota): clamp window slot spend against the voids that target it

docs/status-data-contract.md

@@ -2323,7 +2323,12 @@ The summary currently reports:
 - `runs_24h` / `runs_7d`: observed compact run records in the current status
   sample.
 - `quota_spend_slots_24h` / `quota_spend_slots_7d`: slots from
-  `quota_slot_spent` events in that sample.
+  `quota_slot_spent` events in that sample, using the same rule as the quota
+  spend ledger (`goal_quota_with_spend_ledger`): the event's `event_type`
+  decides, a spend is keyed by the run it was recorded against, and a void is
+  clamped against the spend it names. A spend whose `quota_event` cannot be
+  read contributes no slot, and a void that targets a spend outside the window
+  does not reduce that window.
 - `automation_run_count_24h` / `automation_run_count_7d`: quota spend events
   whose compact `quota_event.source` is `heartbeat`, `automation`, or `cron`.
   If the compact run index does not retain a source, `quota_slot_spent` is

examples/control_plane/status-runtime-summaries-readmodel-smoke.py

@@ -67,7 +67,13 @@ def build_history() -> dict[str, Any]:
         "goal_id": GOAL_ID,
         "classification": "quota_slot_spent",
         "generated_at": utc_isoformat(now),
-        "quota_event": {"slots": 1, "source": "heartbeat"},
+        # event_type is what the spend commit always stamps; the ledger and the
+        # usage summary both read the slot accounting from it.
+        "quota_event": {
+            "event_type": "quota_slot_spent",
+            "slots": 1,
+            "source": "heartbeat",
+        },
     }
     decision_run = {
         "goal_id": GOAL_ID,

examples/usage-summary-smoke.py

@@ -141,6 +141,8 @@ def main() -> int:
         registry_path = write_registry(root)
         runtime = root / "runtime"
         append_run(runtime, goal_id="project-a", generated_at=now - timedelta(hours=1), classification="state_refreshed")
+        # No quota_event: the spend ledger records no slot for a spend whose
+        # event it cannot read, and the usage summary reports the same number.
         append_run(runtime, goal_id="project-a", generated_at=now - timedelta(minutes=30), classification="quota_slot_spent")
         append_run(
             runtime,
@@ -226,8 +228,8 @@ def main() -> int:
         assert usage["sample_run_count"] == 9, usage
         assert totals["runs_24h"] == 6, totals
         assert totals["runs_7d"] == 7, totals
-        assert totals["quota_spend_slots_24h"] == 3, totals
-        assert totals["quota_spend_slots_7d"] == 3, totals
+        assert totals["quota_spend_slots_24h"] == 2, totals
+        assert totals["quota_spend_slots_7d"] == 2, totals
         assert totals["automation_run_count_24h"] == 2, totals
         assert totals["automation_run_count_7d"] == 2, totals
         assert totals["progress_signal_run_count_24h"] == 3, totals
@@ -237,6 +239,12 @@ def main() -> int:
         assert goals["project-a"]["runs_24h"] == 3, goals
         assert goals["project-a"]["runs_7d"] == 4, goals
         assert goals["project-b"]["runs_24h"] == 3, goals
+        # project-a's spend run has no readable quota_event, so it contributes
+        # no slot; project-b's carries a 2-slot event.
+        assert goals["project-a"]["quota_spend_slots_24h"] == 0, goals
+        assert goals["project-a"]["quota_spend_slots_7d"] == 0, goals
+        assert goals["project-b"]["quota_spend_slots_24h"] == 2, goals
+        assert goals["project-b"]["quota_spend_slots_7d"] == 2, goals
         assert goals["project-a"]["progress_signal_run_count_24h"] == 1, goals
         assert goals["project-a"]["progress_signal_run_count_7d"] == 2, goals
         assert goals["project-b"]["progress_signal_run_count_24h"] == 2, goals

loopx/control_plane/quota/slot_accounting.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import json
-from collections.abc import Callable
+from collections.abc import Callable, Iterable
 from copy import deepcopy
 from pathlib import Path
 from typing import Any
@@ -910,3 +910,55 @@ def load_quota_event_from_run(run: dict[str, Any]) -> dict[str, Any] | None:
         return None
     event = record.get("quota_event") if isinstance(record.get("quota_event"), dict) else None
     return event
+
+
+def quota_slot_contribution(run: dict[str, Any]) -> tuple[str, str, int] | None:
+    """Classify one run's contribution to the rolling-window slot ledger.
+
+    ``goal_quota_with_spend_ledger`` enforces quota from this rule and the
+    usage summary reports from it, so both read an event the same way: the
+    quota event's ``event_type`` decides, a spend is keyed by the run it was
+    recorded against, and a void by the run it targets. A run with no usable
+    event contributes no slot rather than a default one, which is what the
+    ledger already assumed.
+    """
+
+    event = load_quota_event_from_run(run)
+    if not event:
+        return None
+    slots = max(0, _int_number(event.get("slots"), default=0))
+    if slots <= 0:
+        return None
+    event_type = str(event.get("event_type") or "")
+    if event_type == QUOTA_SLOT_SPENT_CLASSIFICATION:
+        run_key = str(event.get("run_generated_at") or run.get("generated_at") or "")
+        if not run_key:
+            return None
+        return ("spent", run_key, slots)
+    if event_type == QUOTA_SLOT_VOIDED_CLASSIFICATION:
+        voided_run_generated_at = str(event.get("voided_run_generated_at") or "")
+        if not voided_run_generated_at:
+            return None
+        return ("voided", voided_run_generated_at, slots)
+    return None
+
+
+def net_quota_slot_spend(
+    contributions: Iterable[tuple[Any, str, int]],
+) -> dict[Any, int]:
+    """Clamp each spend bucket against the voids that target it.
+
+    A void only cancels the spend recorded against the key it names, so a
+    window that no longer holds that spend is never pushed negative and a void
+    never cancels an unrelated spend.
+    """
+
+    spent: dict[Any, int] = {}
+    voided: dict[Any, int] = {}
+    for bucket, kind, slots in contributions:
+        target = spent if kind == "spent" else voided
+        target[bucket] = target.get(bucket, 0) + slots
+    return {
+        bucket: max(0, slots - voided.get(bucket, 0))
+        for bucket, slots in spent.items()
+    }

loopx/control_plane/quota/usage_summary.py

@@ -4,6 +4,7 @@
 from datetime import timedelta
 from typing import Any, Callable
 
+from .slot_accounting import net_quota_slot_spend, quota_slot_contribution
 from .spend_sources import VISIBLE_GOAL_SLOT_SPEND_SOURCE
 from .usage_collector import UsageRowError, UsageSample, collect_usage_for_run
 from ..runtime.time import now_utc
@@ -14,6 +15,7 @@
 )
 
 ParseTimestamp = Callable[[Any], Any]
+SlotContribution = tuple[tuple[str, str], str, int]
 USAGE_METRIC_NAMES = (
     "input_tokens",
     "output_tokens",
@@ -23,19 +25,21 @@
 )
 
 
-def quota_spend_slots(run: dict[str, Any]) -> int:
-    classification = str(run.get("classification") or "")
-    if classification not in {"quota_slot_spent", "quota_slot_voided"}:
-        return 0
-    quota_event = run.get("quota_event") if isinstance(run.get("quota_event"), dict) else {}
-    raw_slots = quota_event.get("slots", 1)
-    try:
-        slots = max(0, int(raw_slots))
-    except (TypeError, ValueError):
-        slots = 1
-    if classification == "quota_slot_voided" or str(quota_event.get("event_type") or "") == "quota_slot_voided":
-        return -slots
-    return slots
+def _goal_quota_spend_slots(
+    contributions: list[SlotContribution],
+) -> tuple[dict[str, int], int]:
+    """Fold each run's slot contribution into per-goal window spend.
+
+    Buckets are keyed by ``(goal_id, run key)`` and clamped by the shared
+    ledger rule, so a goal only ever loses the spend a void actually targets.
+    """
+
+    per_goal: dict[str, int] = {}
+    total = 0
+    for (goal_id, _run_key), slots in net_quota_slot_spend(contributions).items():
+        per_goal[goal_id] = per_goal.get(goal_id, 0) + slots
+        total += slots
+    return per_goal, total
 
 
 def is_automation_run(run: dict[str, Any]) -> bool:
@@ -165,6 +169,8 @@ def build_usage_summary(
     goals: dict[str, dict[str, Any]] = {}
     observed_usage_metrics: set[str] = set()
     goal_usage_metrics: dict[str, set[str]] = {}
+    slot_contributions_24h: list[SlotContribution] = []
+    slot_contributions_7d: list[SlotContribution] = []
     sample_count = 0
 
     for run in history.get("runs") or []:
@@ -176,7 +182,7 @@ def build_usage_summary(
             continue
         goal_id = str(run.get("goal_id") or "unknown-goal")
         goal = goals.setdefault(goal_id, blank_usage_goal(goal_id))
-        slots = quota_spend_slots(run)
+        slot_contribution = quota_slot_contribution(run)
         automation_event = is_automation_run(run)
         progress_signal = is_progress_signal_run(run)
         # Present-but-illegal usage fails closed inside collect_usage_for_run.
@@ -186,8 +192,9 @@ def build_usage_summary(
         if generated_at >= cutoff_7d:
             totals["runs_7d"] += 1
             goal["runs_7d"] += 1
-            totals["quota_spend_slots_7d"] += slots
-            goal["quota_spend_slots_7d"] += slots
+            if slot_contribution is not None:
+                kind, run_key, slots = slot_contribution
+                slot_contributions_7d.append(((goal_id, run_key), kind, slots))
             if automation_event:
                 totals["automation_run_count_7d"] += 1
                 goal["automation_run_count_7d"] += 1
@@ -200,8 +207,9 @@ def build_usage_summary(
         if generated_at >= cutoff_24h:
             totals["runs_24h"] += 1
             goal["runs_24h"] += 1
-            totals["quota_spend_slots_24h"] += slots
-            goal["quota_spend_slots_24h"] += slots
+            if slot_contribution is not None:
+                kind, run_key, slots = slot_contribution
+                slot_contributions_24h.append(((goal_id, run_key), kind, slots))
             if automation_event:
                 totals["automation_run_count_24h"] += 1
                 goal["automation_run_count_24h"] += 1
@@ -212,6 +220,14 @@ def build_usage_summary(
                 _accumulate_usage(totals, usage_sample, "24h", observed_usage_metrics)
                 _accumulate_usage(goal, usage_sample, "24h", goal_metrics)
 
+    goal_slots_24h, total_slots_24h = _goal_quota_spend_slots(slot_contributions_24h)
+    goal_slots_7d, total_slots_7d = _goal_quota_spend_slots(slot_contributions_7d)
+    totals["quota_spend_slots_24h"] = total_slots_24h
+    totals["quota_spend_slots_7d"] = total_slots_7d
+    for goal_id, goal in goals.items():
+        goal["quota_spend_slots_24h"] = goal_slots_24h.get(goal_id, 0)
+        goal["quota_spend_slots_7d"] = goal_slots_7d.get(goal_id, 0)
+
     if totals["runs_24h"]:
         for goal in goals.values():
             goal["project_share_24h"] = round(goal["runs_24h"] / totals["runs_24h"], 3)

loopx/quota.py

@@ -63,7 +63,8 @@
     QUOTA_SLOT_VOIDED_CLASSIFICATION,
     build_quota_slot_preview_for_decision,
     build_quota_slot_spend_event as _build_quota_slot_spend_event,
-    load_quota_event_from_run,
+    net_quota_slot_spend,
+    quota_slot_contribution,
     record_quota_slot_spend_from_preview,
 )
 from .control_plane.quota.spend_commit import replay_quota_spend_by_effect_ref
@@ -356,10 +357,6 @@ def goal_quota_config(goal: dict[str, Any] | None) -> dict[str, Any]:
     return payload
 
 
-def _quota_event_run_key(run: dict[str, Any], event: dict[str, Any]) -> str:
-    return str(event.get("run_generated_at") or run.get("generated_at") or "")
-
-
 def goal_quota_with_spend_ledger(
     goal: dict[str, Any] | None,
     runs: list[dict[str, Any]],
@@ -372,8 +369,7 @@ def goal_quota_with_spend_ledger(
     if current_time.tzinfo is None:
         current_time = current_time.replace(tzinfo=timezone.utc)
     window_start = current_time - timedelta(hours=int(payload["window_hours"]))
-    spent_by_run: dict[str, int] = {}
-    voided_by_run: dict[str, int] = {}
+    contributions: list[tuple[str, str, int]] = []
     spend_event_count = 0
     void_event_count = 0
 
@@ -389,32 +385,17 @@ def goal_quota_with_spend_ledger(
             or generated_at > current_time
         ):
             continue
-        event = load_quota_event_from_run(run)
-        if not event:
-            continue
-        event_type = str(event.get("event_type") or "")
-        slots = max(0, _int_number(event.get("slots"), default=0))
-        if slots <= 0:
+        contribution = quota_slot_contribution(run)
+        if contribution is None:
             continue
-        if event_type == QUOTA_SLOT_SPENT_CLASSIFICATION:
-            run_key = _quota_event_run_key(run, event)
-            if not run_key:
-                continue
-            spent_by_run[run_key] = spent_by_run.get(run_key, 0) + slots
+        kind, run_key, slots = contribution
+        contributions.append((run_key, kind, slots))
+        if kind == "spent":
             spend_event_count += 1
-        elif event_type == QUOTA_SLOT_VOIDED_CLASSIFICATION:
-            voided_run_generated_at = str(event.get("voided_run_generated_at") or "")
-            if not voided_run_generated_at:
-                continue
-            voided_by_run[voided_run_generated_at] = (
-                voided_by_run.get(voided_run_generated_at, 0) + slots
-            )
+        else:
             void_event_count += 1
 
-    spent_slots = 0
-    for run_key, slots in spent_by_run.items():
-        spent_slots += max(0, slots - voided_by_run.get(run_key, 0))
-    payload["spent_slots"] = spent_slots
+    payload["spent_slots"] = sum(net_quota_slot_spend(contributions).values())
     payload["spend_source"] = "runtime_events"
     payload["spend_event_count"] = spend_event_count
     if void_event_count: