diff --git a/docs/status-data-contract.md b/docs/status-data-contract.md index 107234f182..aca17225d6 100644 --- a/docs/status-data-contract.md +++ b/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 diff --git a/loopx/control_plane/quota/ledger_readback.py b/loopx/control_plane/quota/ledger_readback.py index d9d229ca91..47cb039e09 100644 --- a/loopx/control_plane/quota/ledger_readback.py +++ b/loopx/control_plane/quota/ledger_readback.py @@ -8,6 +8,7 @@ from __future__ import annotations import json +import math from collections.abc import Iterable from pathlib import Path from typing import Any @@ -15,6 +16,32 @@ 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( diff --git a/loopx/control_plane/quota/usage_summary.py b/loopx/control_plane/quota/usage_summary.py index b744e7d4b4..13a923d671 100644 --- a/loopx/control_plane/quota/usage_summary.py +++ b/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: diff --git a/loopx/quota.py b/loopx/quota.py index ed4b56de23..272c45e122 100644 --- a/loopx/quota.py +++ b/loopx/quota.py @@ -62,10 +62,12 @@ settlement_result_payload, ) from .control_plane.quota.ledger_readback import ( + QUOTA_EVENT_UNREADABLE_REASONS, QUOTA_SLOT_SPENT_CLASSIFICATION, QUOTA_SLOT_VOIDED_CLASSIFICATION, net_quota_slot_spend, - quota_slot_contribution, + quota_slot_contribution, # noqa: F401 + quota_slot_observation, ) from .control_plane.quota.spend_sources import ( DEFAULT_SLOT_SPEND_SOURCE, @@ -113,6 +115,7 @@ "render_quota_markdown": "loopx.presentation.renderers.quota_markdown", "render_quota_scheduler_ack_markdown": "loopx.presentation.renderers.quota_markdown", "render_quota_should_run_markdown": "loopx.presentation.renderers.quota_markdown", + "quota_slot_contribution": "loopx.control_plane.quota.ledger_readback", "build_quota_slot_void_event": "loopx.control_plane.quota.void_commit", "record_quota_slot_void_from_preview": "loopx.control_plane.quota.void_commit", "render_quota_slot_preview_markdown": "loopx.presentation.renderers.quota_event_markdown", @@ -366,6 +369,20 @@ def goal_quota_config(goal: dict[str, Any] | None) -> dict[str, Any]: } if raw.get("next_eligible_at"): payload["next_eligible_at"] = str(raw.get("next_eligible_at")) + # An unreadable quota event is a ledger integrity failure, not a spend of + # zero: the projection that read it reports the count, and every consumer + # of that projection has to fail closed instead of treating the spend as + # absent. + unreadable_event_count = _int_number(raw.get("unreadable_event_count"), default=0) + if unreadable_event_count > 0: + payload["unreadable_event_count"] = unreadable_event_count + # The count says how many events broke; the reasons say why, so the + # read-back can tell an operator which artifact to repair. + reasons = raw.get("unreadable_event_reasons") + if isinstance(reasons, list): + validated = sorted({str(reason) for reason in reasons if str(reason)}) + if validated: + payload["unreadable_event_reasons"] = validated return payload @@ -384,6 +401,8 @@ def goal_quota_with_spend_ledger( contributions: list[tuple[str, str, int]] = [] spend_event_count = 0 void_event_count = 0 + unreadable_event_count = 0 + unreadable_reasons: set[str] = set() for run in runs: if not isinstance(run, dict): @@ -416,7 +435,14 @@ def goal_quota_with_spend_ledger( or generated_at > current_time ): continue - contribution = quota_slot_contribution(run) + contribution, unreadable_reason = quota_slot_observation(run) + if unreadable_reason is not None: + # The index row claims a spend or a void that cannot be read back. + # Counting it as no spend would silently hand the Goal fresh quota, + # so the event is reported as an integrity failure instead. + unreadable_event_count += 1 + unreadable_reasons.add(unreadable_reason) + continue if contribution is None: continue kind, run_key, slots = contribution @@ -429,6 +455,9 @@ def goal_quota_with_spend_ledger( payload["spent_slots"] = sum(net_quota_slot_spend(contributions).values()) payload["spend_source"] = "runtime_events" payload["spend_event_count"] = spend_event_count + if unreadable_event_count: + payload["unreadable_event_count"] = unreadable_event_count + payload["unreadable_event_reasons"] = sorted(unreadable_reasons) if void_event_count: payload["void_event_count"] = void_event_count return payload @@ -468,6 +497,20 @@ def quota_status( elif severity == "high": state = "blocked_health" reason = "health or contract blocker must clear before compute is spent" + elif int(payload.get("unreadable_event_count") or 0) > 0: + # The spend ledger cannot read back an indexed quota event, so the + # remaining quota is unknown. Failing open here would hand the Goal + # back the spend it already used, so the Goal blocks until the + # artifact is readable again. + state = "blocked_health" + reason = ( + f"{int(payload['unreadable_event_count'])} quota ledger event(s) could not " + "be read back; the remaining quota is unknown and cannot be spent" + ) + payload["blocked_action_scope"] = "automatic_agent_turns" + reasons = payload.get("unreadable_event_reasons") + if isinstance(reasons, list) and reasons: + payload["unreadable_event_reasons"] = list(reasons) elif waiting_on in {"user_or_controller", "controller"}: state = "operator_gate" reason = ( @@ -609,6 +652,76 @@ def _todo_summary_claim_scope_agent_id(summary: dict[str, Any]) -> str | None: return projection_todo_summary_claim_scope_agent_id(summary) +def _unreadable_event_reasons( + quota: dict[str, Any], + ledger_quota: dict[str, Any] | None, +) -> list[str]: + """Collect the validated reasons of the unreadable quota events. + + The projection that read the ledger owns the reasons; a projection that + was built before the artifact broke may only have the count, so the ledger + payload is the fallback. The result is validated against the known reason + vocabulary so an unreadable reason cannot become free-form text in the + public projection. + """ + + reasons: set[str] = set() + for source in (quota, ledger_quota): + if not isinstance(source, dict): + continue + raw = source.get("unreadable_event_reasons") + if not isinstance(raw, list): + continue + for reason in raw: + text = str(reason or "").strip() + if text in QUOTA_EVENT_UNREADABLE_REASONS: + reasons.add(text) + return sorted(reasons) + + +def _apply_unreadable_ledger_fail_closed( + quota: dict[str, Any], + *, + ledger_quota: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Keep an unreadable ledger event from reading as a spendable quota. + + The plan layer can reuse a quota projection that was built before the + ledger event became unreadable, or reuse the raw ledger payload without + re-deriving its state. Either way the unreadable count is the authority: + the remaining quota is unknown, so the Goal must not look eligible. + """ + + unreadable_event_count = _int_number(quota.get("unreadable_event_count"), default=0) + if unreadable_event_count <= 0 and isinstance(ledger_quota, dict): + unreadable_event_count = _int_number( + ledger_quota.get("unreadable_event_count"), default=0 + ) + if unreadable_event_count <= 0: + return quota + reasons = _unreadable_event_reasons(quota, ledger_quota) + if str(quota.get("state") or "") == "blocked_health": + # The Goal is already blocked, but the reasons still have to reach the + # public projection: the count alone cannot say what to repair. + if not reasons: + return quota + blocked = dict(quota) + blocked["unreadable_event_count"] = unreadable_event_count + blocked["unreadable_event_reasons"] = reasons + return blocked + blocked = dict(quota) + blocked["state"] = "blocked_health" + blocked["reason"] = ( + f"{unreadable_event_count} quota ledger event(s) could not " + "be read back; the remaining quota is unknown and cannot be spent" + ) + blocked["blocked_action_scope"] = "automatic_agent_turns" + blocked["unreadable_event_count"] = unreadable_event_count + if reasons: + blocked["unreadable_event_reasons"] = reasons + return blocked + + def _quota_plan_goal_quota( *, attention: dict[str, Any], @@ -661,7 +774,7 @@ def _quota_plan_goal_quota( lifecycle_flags=lifecycle_flags, status=status, ) - return quota_with_handoff_outcome_floor( + quota = quota_with_handoff_outcome_floor( quota, waiting_on=waiting_on, project_asset=project_asset, @@ -669,6 +782,8 @@ def _quota_plan_goal_quota( if isinstance(attention.get("handoff_readiness"), dict) else None, ) + ledger_quota = goal.get("quota") if isinstance(goal.get("quota"), dict) else None + return _apply_unreadable_ledger_fail_closed(quota, ledger_quota=ledger_quota) def build_quota_plan( diff --git a/tests/control_plane/test_quota_ledger_integrity.py b/tests/control_plane/test_quota_ledger_integrity.py new file mode 100644 index 0000000000..9e1f948048 --- /dev/null +++ b/tests/control_plane/test_quota_ledger_integrity.py @@ -0,0 +1,709 @@ +"""An unreadable quota artifact must not read as a zero spend. + +The quota ledger is the authority for how much of a Goal's rolling window has +been spent. An index row that claims a ``quota_slot_spent`` or +``quota_slot_voided`` event but whose artifact cannot be read back is an +integrity failure, not a spend of zero: treating it as "no spend" would hand +the Goal back the quota it already used. These tests drive the real +``index.jsonl`` + artifact shape rather than an inline event, because the +inline event is exactly the shortcut that hides the failure. +""" + +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +from loopx.control_plane.quota.ledger_readback import ( + QUOTA_EVENT_UNREADABLE_ARTIFACT_INVALID_JSON, + QUOTA_EVENT_UNREADABLE_ARTIFACT_MISSING, + QUOTA_EVENT_UNREADABLE_ARTIFACT_NOT_OBJECT, + QUOTA_EVENT_UNREADABLE_ARTIFACT_UNREADABLE, + QUOTA_EVENT_UNREADABLE_EVENT_INVALID, + QUOTA_EVENT_UNREADABLE_EVENT_MISSING, + QUOTA_EVENT_UNREADABLE_MISSING_SOURCE, + quota_slot_observation, +) +from loopx.control_plane.quota.usage_summary import build_usage_summary +from loopx.quota import ( + goal_quota_config, + goal_quota_with_spend_ledger, + quota_status, +) + + +GOAL_ID = "quota-ledger-integrity" +NOW = datetime(2026, 3, 1, 12, 0, tzinfo=timezone.utc) + + +def _goal(*, allowed_slots: int = 60) -> dict[str, Any]: + return { + "id": GOAL_ID, + "quota": { + "compute": 1.0, + "window_hours": 24, + "slot_minutes": 1, + "allowed_slots": allowed_slots, + }, + } + + +def _write_artifact(path: Path, event: dict[str, Any] | None) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + if event is None: + path.write_text("{\"quota_event\": ", encoding="utf-8") + return + path.write_text(json.dumps({"quota_event": event}), encoding="utf-8") + + +def _spend_row( + json_path: Path, + *, + generated_at: datetime, + slots: int = 1, +) -> dict[str, Any]: + """An index row for a spend, with the event only in the artifact.""" + + return { + "generated_at": generated_at.isoformat(), + "goal_id": GOAL_ID, + "classification": "quota_slot_spent", + "json_path": str(json_path), + } + + +def _spend_event(generated_at: datetime, *, slots: int = 1) -> dict[str, Any]: + return { + "event_type": "quota_slot_spent", + "slots": slots, + "run_generated_at": generated_at.isoformat(), + } + + +def _index_rows(runtime: Path, rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + index_path = runtime / "goals" / GOAL_ID / "runs" / "index.jsonl" + index_path.parent.mkdir(parents=True, exist_ok=True) + index_path.write_text( + "".join(json.dumps(row) + "\n" for row in rows), + encoding="utf-8", + ) + return rows + + +def _ledger(runtime: Path, rows: list[dict[str, Any]]) -> dict[str, Any]: + return goal_quota_with_spend_ledger(_goal(), _index_rows(runtime, rows), now=NOW) + + +def test_missing_artifact_is_reported_instead_of_counted_as_no_spend(tmp_path: Path) -> None: + artifact = tmp_path / "runs" / "missing.json" + rows = [_spend_row(artifact, generated_at=NOW - timedelta(minutes=10))] + + ledger = _ledger(tmp_path, rows) + + assert ledger["spent_slots"] == 0 + assert ledger["unreadable_event_count"] == 1 + assert ledger["unreadable_event_reasons"] == [QUOTA_EVENT_UNREADABLE_ARTIFACT_MISSING] + + +def test_truncated_artifact_is_reported_instead_of_counted_as_no_spend(tmp_path: Path) -> None: + artifact = tmp_path / "runs" / "truncated.json" + _write_artifact(artifact, None) + rows = [_spend_row(artifact, generated_at=NOW - timedelta(minutes=10))] + + ledger = _ledger(tmp_path, rows) + + assert ledger["unreadable_event_count"] == 1 + assert ledger["unreadable_event_reasons"] == [QUOTA_EVENT_UNREADABLE_ARTIFACT_INVALID_JSON] + + +def test_artifact_without_quota_event_is_reported(tmp_path: Path) -> None: + artifact = tmp_path / "runs" / "no-event.json" + artifact.parent.mkdir(parents=True, exist_ok=True) + artifact.write_text(json.dumps({"recommended_action": "continue"}), encoding="utf-8") + rows = [_spend_row(artifact, generated_at=NOW - timedelta(minutes=10))] + + ledger = _ledger(tmp_path, rows) + + assert ledger["unreadable_event_count"] == 1 + assert ledger["unreadable_event_reasons"] == [QUOTA_EVENT_UNREADABLE_EVENT_MISSING] + + +def test_non_object_artifact_is_reported(tmp_path: Path) -> None: + artifact = tmp_path / "runs" / "list.json" + artifact.parent.mkdir(parents=True, exist_ok=True) + artifact.write_text(json.dumps([1, 2, 3]), encoding="utf-8") + rows = [_spend_row(artifact, generated_at=NOW - timedelta(minutes=10))] + + ledger = _ledger(tmp_path, rows) + + assert ledger["unreadable_event_count"] == 1 + assert ledger["unreadable_event_reasons"] == [QUOTA_EVENT_UNREADABLE_ARTIFACT_NOT_OBJECT] + + +def test_index_row_without_any_event_source_is_reported(tmp_path: Path) -> None: + rows = [ + { + "generated_at": (NOW - timedelta(minutes=10)).isoformat(), + "goal_id": GOAL_ID, + "classification": "quota_slot_spent", + } + ] + + ledger = _ledger(tmp_path, rows) + + assert ledger["unreadable_event_count"] == 1 + assert ledger["unreadable_event_reasons"] == [QUOTA_EVENT_UNREADABLE_MISSING_SOURCE] + + +def test_unreadable_event_keeps_the_goal_blocked_instead_of_eligible(tmp_path: Path) -> None: + artifact = tmp_path / "runs" / "missing.json" + rows = [_spend_row(artifact, generated_at=NOW - timedelta(minutes=10))] + ledger = _ledger(tmp_path, rows) + + status = quota_status( + {**_goal(), "quota": {**_goal()["quota"], **ledger}}, + waiting_on="codex", + ) + + assert status["state"] == "blocked_health" + assert "could not" in status["reason"] + assert status["blocked_action_scope"] == "automatic_agent_turns" + + +def test_readable_events_still_count_while_an_unreadable_one_is_reported( + tmp_path: Path, +) -> None: + readable = tmp_path / "runs" / "readable.json" + _write_artifact(readable, _spend_event(NOW - timedelta(minutes=20), slots=2)) + broken = tmp_path / "runs" / "broken.json" + rows = [ + _spend_row(readable, generated_at=NOW - timedelta(minutes=20)), + _spend_row(broken, generated_at=NOW - timedelta(minutes=10)), + ] + + ledger = _ledger(tmp_path, rows) + + assert ledger["spent_slots"] == 2 + assert ledger["spend_event_count"] == 1 + assert ledger["unreadable_event_count"] == 1 + + +def test_restoring_the_artifact_restores_the_ledger_and_releases_the_goal( + tmp_path: Path, +) -> None: + artifact = tmp_path / "runs" / "spend.json" + rows = [_spend_row(artifact, generated_at=NOW - timedelta(minutes=10))] + + broken = _ledger(tmp_path, rows) + assert broken["unreadable_event_count"] == 1 + + _write_artifact(artifact, _spend_event(NOW - timedelta(minutes=10))) + repaired = _ledger(tmp_path, rows) + + assert repaired.get("unreadable_event_count", 0) == 0 + assert repaired["spent_slots"] == 1 + assert repaired["spend_event_count"] == 1 + status = quota_status( + {**_goal(), "quota": {**_goal()["quota"], **repaired}}, + waiting_on="codex", + ) + assert status["state"] == "eligible" + + +def test_usage_summary_reports_the_same_unreadable_count(tmp_path: Path) -> None: + artifact = tmp_path / "runs" / "missing.json" + # The usage summary windows are relative to the wall clock, so this row + # has to sit inside the live 24h window rather than the fixed ledger one. + live_now = datetime.now(timezone.utc) + rows = [_spend_row(artifact, generated_at=live_now - timedelta(minutes=10))] + + ledger = _ledger(tmp_path, rows) + summary = build_usage_summary( + {"runs": rows}, + parse_timestamp=lambda value: datetime.fromisoformat(str(value)), + ) + + assert summary["totals"]["unreadable_event_count_24h"] == 1 + assert summary["totals"]["unreadable_event_count_7d"] == 1 + assert summary["goals"][0]["unreadable_event_count_24h"] == 1 + # The readable spend stays the same number on both surfaces. + assert summary["totals"]["quota_spend_slots_24h"] == ledger["spent_slots"] + + +def test_a_legal_void_still_cancels_the_spend_it_targets(tmp_path: Path) -> None: + spend = tmp_path / "runs" / "spend.json" + void = tmp_path / "runs" / "void.json" + _write_artifact(spend, _spend_event(NOW - timedelta(minutes=20))) + _write_artifact( + void, + { + "event_type": "quota_slot_voided", + "slots": 1, + "voided_run_generated_at": (NOW - timedelta(minutes=20)).isoformat(), + }, + ) + rows = [ + _spend_row(spend, generated_at=NOW - timedelta(minutes=20)), + { + "generated_at": (NOW - timedelta(minutes=10)).isoformat(), + "goal_id": GOAL_ID, + "classification": "quota_slot_voided", + "json_path": str(void), + }, + ] + + ledger = _ledger(tmp_path, rows) + + assert ledger["spent_slots"] == 0 + assert ledger["void_event_count"] == 1 + assert ledger.get("unreadable_event_count", 0) == 0 + + +def test_observation_distinguishes_an_unreadable_event_from_a_zero_spend_event( + tmp_path: Path, +) -> None: + artifact = tmp_path / "runs" / "missing.json" + broken_row = _spend_row(artifact, generated_at=NOW - timedelta(minutes=10)) + contribution, reason = quota_slot_observation(broken_row) + assert contribution is None + assert reason == QUOTA_EVENT_UNREADABLE_ARTIFACT_MISSING + + zero = { + "generated_at": (NOW - timedelta(minutes=10)).isoformat(), + "goal_id": GOAL_ID, + "classification": "quota_slot_spent", + "quota_event": {"event_type": "quota_slot_spent", "slots": 0}, + } + contribution, reason = quota_slot_observation(zero) + assert contribution is None + assert reason is None + + +def test_a_malformed_quota_event_is_reported_instead_of_counted_as_no_spend() -> None: + """A readable object that names no known event type is still unreadable. + + The index row claims a spend, so reading the event as "this event spends + nothing" would hand the Goal back the quota it already used. + """ + + malformed = { + "generated_at": (NOW - timedelta(minutes=10)).isoformat(), + "goal_id": GOAL_ID, + "classification": "quota_slot_spent", + "quota_event": {"slots": 1}, + } + + contribution, reason = quota_slot_observation(malformed) + assert contribution is None + assert reason == QUOTA_EVENT_UNREADABLE_EVENT_INVALID + + ledger = goal_quota_with_spend_ledger(_goal(), [malformed], now=NOW) + assert ledger["spent_slots"] == 0 + assert ledger["unreadable_event_count"] == 1 + assert ledger["unreadable_event_reasons"] == [QUOTA_EVENT_UNREADABLE_EVENT_INVALID] + + status = quota_status( + {**_goal(), "quota": {**_goal()["quota"], **ledger}}, + waiting_on="codex", + ) + assert status["state"] == "blocked_health" + assert status["blocked_action_scope"] == "automatic_agent_turns" + + +def test_an_unknown_event_type_is_reported_instead_of_counted_as_no_spend() -> None: + unknown = { + "generated_at": (NOW - timedelta(minutes=10)).isoformat(), + "goal_id": GOAL_ID, + "classification": "quota_slot_spent", + "quota_event": {"event_type": "quota_slot_something_else", "slots": 1}, + } + + contribution, reason = quota_slot_observation(unknown) + assert contribution is None + assert reason == QUOTA_EVENT_UNREADABLE_EVENT_INVALID + + ledger = goal_quota_with_spend_ledger(_goal(), [unknown], now=NOW) + assert ledger["unreadable_event_count"] == 1 + + +def test_a_legal_zero_spend_event_is_not_reported_as_unreadable() -> None: + zero = { + "generated_at": (NOW - timedelta(minutes=10)).isoformat(), + "goal_id": GOAL_ID, + "classification": "quota_slot_spent", + "quota_event": {"event_type": "quota_slot_spent", "slots": 0}, + } + + contribution, reason = quota_slot_observation(zero) + assert contribution is None + assert reason is None + + ledger = goal_quota_with_spend_ledger(_goal(), [zero], now=NOW) + assert ledger.get("unreadable_event_count", 0) == 0 + + +def test_the_public_projection_keeps_the_unreadable_reason(tmp_path: Path) -> None: + """The count says how many events broke; the reason says what to repair.""" + + artifact = tmp_path / "runs" / "truncated.json" + _write_artifact(artifact, None) + rows = [_spend_row(artifact, generated_at=NOW - timedelta(minutes=10))] + ledger = _ledger(tmp_path, rows) + + config = goal_quota_config({**_goal(), "quota": {**_goal()["quota"], **ledger}}) + assert config["unreadable_event_count"] == 1 + assert config["unreadable_event_reasons"] == [QUOTA_EVENT_UNREADABLE_ARTIFACT_INVALID_JSON] + + status = quota_status( + {**_goal(), "quota": {**_goal()["quota"], **ledger}}, + waiting_on="codex", + ) + assert status["state"] == "blocked_health" + assert status["unreadable_event_reasons"] == [QUOTA_EVENT_UNREADABLE_ARTIFACT_INVALID_JSON] + + +def test_the_plan_keeps_the_unreadable_reason(tmp_path: Path) -> None: + from loopx.quota import build_quota_plan + + artifact = tmp_path / "runs" / "missing.json" + rows = [_spend_row(artifact, generated_at=NOW - timedelta(minutes=10))] + ledger = _ledger(tmp_path, rows) + payload = _plan_payload({**(_goal()["quota"]), **ledger}) + + plan = build_quota_plan(payload, mode="plan") + + item = plan["groups"]["blocked_health"][0] + assert item["quota"]["unreadable_event_count"] == 1 + assert item["quota"]["unreadable_event_reasons"] == [ + QUOTA_EVENT_UNREADABLE_ARTIFACT_MISSING + ] + + +def test_an_unreadable_void_is_also_reported(tmp_path: Path) -> None: + artifact = tmp_path / "runs" / "missing-void.json" + rows = [ + { + "generated_at": (NOW - timedelta(minutes=10)).isoformat(), + "goal_id": GOAL_ID, + "classification": "quota_slot_voided", + "json_path": str(artifact), + } + ] + + ledger = _ledger(tmp_path, rows) + + assert ledger["unreadable_event_count"] == 1 + assert ledger["unreadable_event_reasons"] == [QUOTA_EVENT_UNREADABLE_ARTIFACT_MISSING] + + +def test_an_artifact_that_is_not_utf8_is_reported(tmp_path: Path) -> None: + artifact = tmp_path / "runs" / "binary.json" + artifact.parent.mkdir(parents=True, exist_ok=True) + artifact.write_bytes(b"\xff\xfe\x00binary") + rows = [_spend_row(artifact, generated_at=NOW - timedelta(minutes=10))] + + ledger = _ledger(tmp_path, rows) + + assert ledger["unreadable_event_count"] == 1 + assert ledger["unreadable_event_reasons"] == [QUOTA_EVENT_UNREADABLE_ARTIFACT_UNREADABLE] + + +def _plan_payload( + goal_quota: dict[str, Any], + *, + attention_quota: dict[str, Any] | None = None, +) -> dict[str, Any]: + """A status payload whose run history carries the Goal's ledger quota. + + The attention queue normally carries the quota status derived from the + same ledger payload, so the projection is built from the ledger rather + than written by hand. + """ + + goal = { + "id": GOAL_ID, + "registry_member": True, + "quota": goal_quota, + } + if attention_quota is None: + attention_quota = quota_status( + {"id": GOAL_ID, "quota": goal_quota}, + waiting_on="codex", + ) + return { + "ok": True, + "registry": None, + "runtime_root": "/tmp/loopx-quota-ledger-integrity", + "goal_count": 1, + "run_count": 1, + "attention_queue": { + "items": [ + {"goal_id": GOAL_ID, "waiting_on": "codex", "quota": attention_quota} + ] + }, + "run_history": {"goals": [goal]}, + } + + +def test_the_quota_plan_does_not_offer_a_goal_with_an_unreadable_event( + tmp_path: Path, +) -> None: + from loopx.quota import build_quota_plan + + artifact = tmp_path / "runs" / "missing.json" + rows = [_spend_row(artifact, generated_at=NOW - timedelta(minutes=10))] + ledger = _ledger(tmp_path, rows) + payload = _plan_payload({**(_goal()["quota"]), **ledger}) + + plan = build_quota_plan(payload, mode="plan") + + assert plan["groups"]["blocked_health"], plan["groups"] + item = plan["groups"]["blocked_health"][0] + assert item["goal_id"] == GOAL_ID + assert item["quota"]["unreadable_event_count"] == 1 + assert item["quota"]["blocked_action_scope"] == "automatic_agent_turns" + assert plan["summary"]["next_automatic_turn"] is None + + +def test_the_quota_plan_still_offers_a_goal_without_an_unreadable_event( + tmp_path: Path, +) -> None: + from loopx.quota import build_quota_plan + + artifact = tmp_path / "runs" / "spend.json" + _write_artifact(artifact, _spend_event(NOW - timedelta(minutes=10))) + rows = [_spend_row(artifact, generated_at=NOW - timedelta(minutes=10))] + ledger = _ledger(tmp_path, rows) + payload = _plan_payload({**(_goal()["quota"]), **ledger}) + + plan = build_quota_plan(payload, mode="plan") + + assert plan["groups"]["eligible"], plan["groups"] + assert plan["summary"]["next_automatic_turn"] == GOAL_ID + + +def test_a_stale_eligible_projection_cannot_hide_an_unreadable_ledger_event( + tmp_path: Path, +) -> None: + """A projection built before the artifact broke must not stay eligible.""" + + from loopx.quota import build_quota_plan + + artifact = tmp_path / "runs" / "missing.json" + rows = [_spend_row(artifact, generated_at=NOW - timedelta(minutes=10))] + ledger = _ledger(tmp_path, rows) + stale_quota = { + "compute": 1.0, + "window_hours": 24, + "slot_minutes": 1, + "allowed_slots": 60, + "spent_slots": 0, + "state": "eligible", + "reason": "1 compute quota; eligible for the next automatic agent turn", + } + payload = _plan_payload( + {**(_goal()["quota"]), **ledger}, + attention_quota=stale_quota, + ) + + plan = build_quota_plan(payload, mode="plan") + + assert plan["groups"]["blocked_health"], plan["groups"] + item = plan["groups"]["blocked_health"][0] + assert item["quota"]["unreadable_event_count"] == 1 + assert plan["summary"]["next_automatic_turn"] is None + + +def _slot_artifact( + path: Path, + *, + generated_at: datetime | None = None, + event: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Write a physical quota artifact and return its index row. + + The index row is identical across the cases below; only the artifact on + disk changes, which is the shape a real repair takes. + """ + + generated_at = generated_at or (NOW - timedelta(minutes=10)) + if event is None: + event = _spend_event(generated_at) + _write_artifact(path, event) + return _spend_row(path, generated_at=generated_at) + + +def test_a_spend_artifact_with_an_unreadable_slot_count_is_not_a_zero_spend( + tmp_path: Path, +) -> None: + """An unknown quantity is not an explicit zero. + + The index row still claims a spend; only the artifact lost its size. That + must be reported as an unreadable event rather than folded into a zero + spend, which would hand the Goal back the quota it already used. + """ + + readable = { + "event_type": "quota_slot_spent", + "run_generated_at": (NOW - timedelta(minutes=10)).isoformat(), + } + broken_counts: list[Any] = ["lost", "", True, ["1"], {"value": 1}] + + for broken in broken_counts: + artifact = tmp_path / f"runs-{abs(hash(str(broken)))}.json" + row = _slot_artifact(artifact, event={**readable, "slots": broken}) + + observation, reason = quota_slot_observation(row) + assert observation is None + assert reason == QUOTA_EVENT_UNREADABLE_EVENT_INVALID, broken + + ledger = _ledger(tmp_path, [row]) + assert ledger["spent_slots"] == 0 + assert ledger["unreadable_event_count"] == 1 + assert ledger["unreadable_event_reasons"] == [ + QUOTA_EVENT_UNREADABLE_EVENT_INVALID + ] + + # The field is gone entirely, not just unparseable. + missing = tmp_path / "runs-missing-slots.json" + row = _slot_artifact(missing, event=dict(readable)) + observation, reason = quota_slot_observation(row) + assert observation is None + assert reason == QUOTA_EVENT_UNREADABLE_EVENT_INVALID + + +def test_an_unreadable_slot_count_keeps_the_goal_out_of_the_plan( + tmp_path: Path, +) -> None: + """The should-run guard must not offer a Goal whose spend size is unknown.""" + + from loopx.quota import build_quota_plan + + artifact = tmp_path / "runs" / "slots-lost.json" + row = _slot_artifact( + artifact, + event={ + "event_type": "quota_slot_spent", + "run_generated_at": (NOW - timedelta(minutes=10)).isoformat(), + "slots": "lost", + }, + ) + ledger = _ledger(tmp_path, [row]) + payload = _plan_payload({**_goal()["quota"], **ledger}) + + plan = build_quota_plan(payload, mode="plan") + + assert plan["groups"]["blocked_health"], plan["groups"] + item = plan["groups"]["blocked_health"][0] + assert item["goal_id"] == GOAL_ID + assert item["quota"]["unreadable_event_count"] == 1 + assert item["quota"]["unreadable_event_reasons"] == [ + QUOTA_EVENT_UNREADABLE_EVENT_INVALID + ] + assert plan["summary"]["next_automatic_turn"] is None + + +def test_a_void_artifact_with_an_unreadable_slot_count_is_not_a_zero_void( + tmp_path: Path, +) -> None: + """A void of an unknown size cannot be read as cancelling nothing.""" + + for slots_marker in ("missing", "lost"): + artifact = tmp_path / f"runs-void-{slots_marker}.json" + event: dict[str, Any] = { + "event_type": "quota_slot_voided", + "voided_run_generated_at": (NOW - timedelta(minutes=20)).isoformat(), + } + if slots_marker != "missing": + event["slots"] = slots_marker + _write_artifact(artifact, event) + row = { + "generated_at": (NOW - timedelta(minutes=10)).isoformat(), + "goal_id": GOAL_ID, + "classification": "quota_slot_voided", + "json_path": str(artifact), + } + + observation, reason = quota_slot_observation(row) + assert observation is None + assert reason == QUOTA_EVENT_UNREADABLE_EVENT_INVALID, slots_marker + + +def test_an_explicit_zero_slot_count_stays_a_legal_zero(tmp_path: Path) -> None: + """A written zero is readable, and stays a zero contribution.""" + + artifact = tmp_path / "runs" / "zero.json" + zero = { + "event_type": "quota_slot_spent", + "run_generated_at": (NOW - timedelta(minutes=10)).isoformat(), + "slots": 0, + } + row = _slot_artifact(artifact, event=zero) + + observation, reason = quota_slot_observation(row) + assert observation is None + assert reason is None + + ledger = _ledger(tmp_path, [row]) + assert ledger["spent_slots"] == 0 + assert "unreadable_event_count" not in ledger + status = quota_status( + {**_goal(), "quota": {**_goal()["quota"], **ledger}}, + waiting_on="codex", + ) + assert status["state"] == "eligible" + + +def test_a_numeric_string_slot_count_is_still_readable(tmp_path: Path) -> None: + """The supported historical representation keeps working.""" + + artifact = tmp_path / "runs" / "string-slots.json" + event = { + "event_type": "quota_slot_spent", + "run_generated_at": (NOW - timedelta(minutes=10)).isoformat(), + "slots": "3", + } + row = _slot_artifact(artifact, event=event) + + observation, reason = quota_slot_observation(row) + assert reason is None + assert observation == ("spent", event["run_generated_at"], 3) + + ledger = _ledger(tmp_path, [row]) + assert ledger["spent_slots"] == 3 + assert "unreadable_event_count" not in ledger + + +def test_restoring_the_slot_count_releases_the_goal_again(tmp_path: Path) -> None: + """Repairing the artifact is the whole recovery: the index row never moves.""" + + artifact = tmp_path / "runs" / "spend.json" + row = _slot_artifact( + artifact, + event={ + "event_type": "quota_slot_spent", + "run_generated_at": (NOW - timedelta(minutes=10)).isoformat(), + "slots": "lost", + }, + ) + + broken = _ledger(tmp_path, [row]) + assert broken["unreadable_event_count"] == 1 + broken_status = quota_status( + {**_goal(), "quota": {**_goal()["quota"], **broken}}, + waiting_on="codex", + ) + assert broken_status["state"] == "blocked_health" + + _write_artifact(artifact, _spend_event(NOW - timedelta(minutes=10), slots=4)) + repaired = _ledger(tmp_path, [row]) + + assert "unreadable_event_count" not in repaired + assert repaired["spent_slots"] == 4 + repaired_status = quota_status( + {**_goal(), "quota": {**_goal()["quota"], **repaired}}, + waiting_on="codex", + ) + assert repaired_status["state"] == "eligible"