fix(quota): fail closed on unreadable ledger events

docs/status-data-contract.md

@@ -280,6 +280,20 @@ For spend accounting, status derives `spent_slots` from compact
 `quota_slot_spent` runtime events in the current quota window. The registry
 remains the policy source for compute share and window size, not the spend
 ledger.
+An indexed quota event whose artifact cannot be read back is a ledger
+integrity failure, not a spend of zero: status reports it through
+`quota.unreadable_event_count` (with `quota.unreadable_event_reasons` naming
+the cause), keeps the quota state at `blocked_health`, and blocks automatic
+agent turns until the artifact is readable again.
+An indexed quota event that reads back but does not name a known event type,
+or that names a known type without the fields that type requires, is the same
+integrity failure: the reason is `event_invalid`, and the event is never read
+as a zero spend. A readable event that names a known type and spends zero
+slots is a legal zero contribution, not an unreadable one.
+The legacy `from loopx.quota import quota_slot_contribution` export stays an
+identity alias of `loopx.control_plane.quota.ledger_readback.quota_slot_contribution`;
+it reports the contribution of one readable event only, and it is not the
+surface that carries the unreadable count.
 `quota_slot_spent` is status-neutral accounting: it must remain visible in run
 history for quota audit, but status and attention queues should use the latest
 non-accounting run as the current work state.
@@ -2512,6 +2526,17 @@ The summary currently reports:
   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.
+- `unreadable_event_count_24h` / `unreadable_event_count_7d`: optional; omitted
+  when no unreadable event was observed. When present, they count indexed quota
+  events (`quota_slot_spent` / `quota_slot_voided`) in that sample whose
+  artifact cannot be read back, or whose slot count is missing or unparseable.
+  The real slot count is unknown, so these events are counted instead of being
+  silently read as a zero spend (an explicitly written `slots: 0` stays a legal
+  zero, not an unreadable event), and the
+  quota projection fails closed while the count is non-zero. The counters only
+  appear when they describe a real integrity failure, so a healthy summary
+  keeps its original shape and the agent-facing status output stays inside its
+  growth budget.
 - `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

loopx/control_plane/quota/ledger_readback.py

@@ -8,13 +8,40 @@
 from __future__ import annotations
 
 import json
+import math
 from collections.abc import Iterable
 from pathlib import Path
 from typing import Any
 
 QUOTA_SLOT_SPENT_CLASSIFICATION = "quota_slot_spent"
 QUOTA_SLOT_VOIDED_CLASSIFICATION = "quota_slot_voided"
 
+# Why an indexed quota event could not be read back. The ledger must not
+# confuse "this event spends nothing" with "this event cannot be read", so
+# every unreadable case carries a stable machine-readable reason.
+QUOTA_EVENT_UNREADABLE_MISSING_SOURCE = "missing_source"
+QUOTA_EVENT_UNREADABLE_ARTIFACT_MISSING = "artifact_missing"
+QUOTA_EVENT_UNREADABLE_ARTIFACT_UNREADABLE = "artifact_unreadable"
+QUOTA_EVENT_UNREADABLE_ARTIFACT_INVALID_JSON = "artifact_invalid_json"
+QUOTA_EVENT_UNREADABLE_ARTIFACT_NOT_OBJECT = "artifact_not_object"
+QUOTA_EVENT_UNREADABLE_EVENT_MISSING = "event_missing"
+QUOTA_EVENT_UNREADABLE_EVENT_INVALID = "event_invalid"
+
+# The closed vocabulary of unreadable reasons. The public quota projection
+# only reports a reason from this set, so an unreadable event cannot smuggle
+# free-form text into the status surface.
+QUOTA_EVENT_UNREADABLE_REASONS = frozenset(
+    {
+        QUOTA_EVENT_UNREADABLE_MISSING_SOURCE,
+        QUOTA_EVENT_UNREADABLE_ARTIFACT_MISSING,
+        QUOTA_EVENT_UNREADABLE_ARTIFACT_UNREADABLE,
+        QUOTA_EVENT_UNREADABLE_ARTIFACT_INVALID_JSON,
+        QUOTA_EVENT_UNREADABLE_ARTIFACT_NOT_OBJECT,
+        QUOTA_EVENT_UNREADABLE_EVENT_MISSING,
+        QUOTA_EVENT_UNREADABLE_EVENT_INVALID,
+    }
+)
+
 
 def _int_number(value: Any, *, default: int) -> int:
     if isinstance(value, bool):
@@ -34,6 +61,67 @@ def _int_number(value: Any, *, default: int) -> int:
     return default
 
 
+def _slot_count(value: Any) -> int | None:
+    """Read one quota event's slot count, or ``None`` when it is unknown.
+
+    A missing or unparseable count means the remaining budget is unknown, which
+    the ledger must not fold into "this event spends nothing". Integers, finite
+    floats and numeric strings are the supported representations; booleans and
+    unmapped shapes are not counts. A readable negative count clamps to zero,
+    preserving the existing contribution semantics.
+    """
+
+    if isinstance(value, bool):
+        return None
+    if isinstance(value, int):
+        return max(0, value)
+    if isinstance(value, float):
+        if not math.isfinite(value):
+            return None
+        return max(0, int(value))
+    if isinstance(value, str):
+        text = value.strip()
+        if not text:
+            return None
+        try:
+            return max(0, int(float(text)))
+        except (ValueError, OverflowError):
+            return None
+    return None
+
+
+def _read_quota_event_artifact(raw_json_path: str) -> tuple[dict[str, Any] | None, str | None]:
+    """Read the quota event out of one durable artifact.
+
+    Returns ``(event, unreadable_reason)``. A readable artifact without a
+    usable quota event is still an integrity problem: the index row claims a
+    spend or void happened, so the missing event must be reported instead of
+    being treated as "no spend".
+    """
+
+    json_path = Path(raw_json_path).expanduser()
+    if not json_path.exists():
+        return None, QUOTA_EVENT_UNREADABLE_ARTIFACT_MISSING
+    try:
+        raw = json_path.read_bytes()
+    except OSError:
+        return None, QUOTA_EVENT_UNREADABLE_ARTIFACT_UNREADABLE
+    try:
+        text = raw.decode("utf-8")
+    except UnicodeDecodeError:
+        return None, QUOTA_EVENT_UNREADABLE_ARTIFACT_UNREADABLE
+    try:
+        record = json.loads(text)
+    except json.JSONDecodeError:
+        return None, QUOTA_EVENT_UNREADABLE_ARTIFACT_INVALID_JSON
+    if not isinstance(record, dict):
+        return None, QUOTA_EVENT_UNREADABLE_ARTIFACT_NOT_OBJECT
+    event = record.get("quota_event") if isinstance(record.get("quota_event"), dict) else None
+    if not event:
+        return None, QUOTA_EVENT_UNREADABLE_EVENT_MISSING
+    return event, None
+
+
 def load_quota_event_from_run(run: dict[str, Any]) -> dict[str, Any] | None:
     if str(run.get("classification") or "") not in {
         QUOTA_SLOT_SPENT_CLASSIFICATION,
@@ -47,19 +135,35 @@ def load_quota_event_from_run(run: dict[str, Any]) -> dict[str, Any] | None:
     raw_json_path = str(run.get("json_path") or "")
     if not raw_json_path:
         return None
-    json_path = Path(raw_json_path).expanduser()
-    if not json_path.exists():
-        return None
-    try:
-        record = json.loads(json_path.read_text(encoding="utf-8"))
-    except (OSError, json.JSONDecodeError):
-        return None
-    if not isinstance(record, dict):
-        return None
-    event = record.get("quota_event") if isinstance(record.get("quota_event"), dict) else None
+    event, _reason = _read_quota_event_artifact(raw_json_path)
     return event
 
 
+def resolve_quota_event_from_run(
+    run: dict[str, Any],
+) -> tuple[dict[str, Any] | None, str | None]:
+    """Resolve one run's quota event and report why it could not be read.
+
+    ``(event, None)`` means the event was read, ``(None, reason)`` means the
+    index row claims a quota event but the event cannot be read back, and
+    ``(None, None)`` means the row is not a quota event row at all.
+    """
+
+    classification = str(run.get("classification") or "")
+    if classification not in {
+        QUOTA_SLOT_SPENT_CLASSIFICATION,
+        QUOTA_SLOT_VOIDED_CLASSIFICATION,
+    }:
+        return None, None
+    event = run.get("quota_event") if isinstance(run.get("quota_event"), dict) else None
+    if event:
+        return event, None
+    raw_json_path = str(run.get("json_path") or "")
+    if not raw_json_path:
+        return None, QUOTA_EVENT_UNREADABLE_MISSING_SOURCE
+    return _read_quota_event_artifact(raw_json_path)
+
+
 def quota_slot_contribution(run: dict[str, Any]) -> tuple[str, str, int] | None:
     """Classify one run's contribution to the rolling-window slot ledger.
 
@@ -71,24 +175,59 @@ def quota_slot_contribution(run: dict[str, Any]) -> tuple[str, str, int] | None:
     ledger already assumed.
     """
 
-    event = load_quota_event_from_run(run)
+    contribution, _reason = quota_slot_observation(run)
+    return contribution
+
+
+def quota_slot_observation(
+    run: dict[str, Any],
+) -> tuple[tuple[str, str, int] | None, str | None]:
+    """Observe one run's slot contribution and its readability.
+
+    ``(contribution, None)`` means the run's quota event was read, whether or
+    not it spends a slot. ``(None, reason)`` means the index row claims a quota
+    event that cannot be read back, which is an integrity failure the ledger
+    must report instead of silently dropping.
+    """
+
+    event, reason = resolve_quota_event_from_run(run)
+    if reason:
+        return None, reason
     if not event:
-        return None
-    slots = max(0, _int_number(event.get("slots"), default=0))
-    if slots <= 0:
-        return None
+        return None, 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)
+            return None, QUOTA_EVENT_UNREADABLE_EVENT_INVALID
+        slots = _slot_count(event.get("slots"))
+        if slots is None:
+            # The event names a spend but not how much it spent. That is an
+            # unknown quantity, not a zero spend: guessing zero would hand the
+            # Goal back the quota it already used.
+            return None, QUOTA_EVENT_UNREADABLE_EVENT_INVALID
+        if slots == 0:
+            # An explicit, readable zero is a legal zero contribution.
+            return None, None
+        return ("spent", run_key, slots), None
     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
+            return None, QUOTA_EVENT_UNREADABLE_EVENT_INVALID
+        slots = _slot_count(event.get("slots"))
+        if slots is None:
+            # Same rule as a spend: an unknown cancelled quantity cannot be
+            # read as "this void cancels nothing".
+            return None, QUOTA_EVENT_UNREADABLE_EVENT_INVALID
+        if slots == 0:
+            # An explicit, readable zero is a legal zero contribution.
+            return None, None
+        return ("voided", voided_run_generated_at, slots), None
+    # The index row claims a quota event, but the event does not name a type
+    # this ledger understands. Reading it as "this event spends nothing" would
+    # hand the Goal back the quota it already used, so the malformed event is
+    # reported as unreadable instead.
+    return None, QUOTA_EVENT_UNREADABLE_EVENT_INVALID
 
 
 def net_quota_slot_spend(

loopx/control_plane/quota/usage_summary.py

@@ -4,7 +4,7 @@
 from datetime import timedelta
 from typing import Any, Callable
 
-from .ledger_readback import net_quota_slot_spend, quota_slot_contribution
+from .ledger_readback import net_quota_slot_spend, quota_slot_observation
 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
@@ -182,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))
-        slot_contribution = quota_slot_contribution(run)
+        slot_contribution, unreadable_reason = quota_slot_observation(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.
@@ -192,7 +192,13 @@ def build_usage_summary(
         if generated_at >= cutoff_7d:
             totals["runs_7d"] += 1
             goal["runs_7d"] += 1
-            if slot_contribution is not None:
+            if unreadable_reason is not None:
+                # The summary must not report a spend it cannot read back as a
+                # zero-spend run; the integrity failure is reported next to
+                # the slots that were readable.
+                totals["unreadable_event_count_7d"] = totals.get("unreadable_event_count_7d", 0) + 1
+                goal["unreadable_event_count_7d"] = goal.get("unreadable_event_count_7d", 0) + 1
+            elif slot_contribution is not None:
                 kind, run_key, slots = slot_contribution
                 slot_contributions_7d.append(((goal_id, run_key), kind, slots))
             if automation_event:
@@ -207,7 +213,10 @@ def build_usage_summary(
         if generated_at >= cutoff_24h:
             totals["runs_24h"] += 1
             goal["runs_24h"] += 1
-            if slot_contribution is not None:
+            if unreadable_reason is not None:
+                totals["unreadable_event_count_24h"] = totals.get("unreadable_event_count_24h", 0) + 1
+                goal["unreadable_event_count_24h"] = goal.get("unreadable_event_count_24h", 0) + 1
+            elif slot_contribution is not None:
                 kind, run_key, slots = slot_contribution
                 slot_contributions_24h.append(((goal_id, run_key), kind, slots))
             if automation_event: