gate.quota: Quota tracking#

Multi-harness quota tracking and economic estimation. Tracks quota windows (session, rolling, calendar), estimates USD list-price and physical energy (kWh) from token tallies, and renders compact text tables or JSON for CLI consumption.

Status: library only, not wired to the CLI#

The data models, estimation, window, and rendering functions below are real and tested. The gate quota console command is not: it has no local-ledger harness feed wired up, so it always prints “quota tracking is not implemented” to stderr and exits 2 (plain or --json) rather than rendering the empty QuotaSnapshot this module’s main() would otherwise produce – a well-formed empty success would read as “no harness is near its cap,” which is not something the CLI can honestly claim yet. gate.dashboard’s quota panel is downstream of the same unwired feed.

This module is unrelated to gate grant/gate revoke/gate grants: those mint and spend short-lived, human-issued bypass grants for the superheavy-model hook (One-off grants), not usage quota. Both happen to carry an expiry, but a grant’s TTL and a quota window are different mechanisms with no shared state.

Data models#

class gate.quota.TokenTotals(tokens_in: int = 0, tokens_out: int = 0, cache_creation: int = 0, cache_read: int = 0, reasoning: int = 0)#

A phase-split token tally, summed over some window (pure carrier for econ).

class gate.quota.WindowSpec(name: str, duration: timedelta | None, reset_mode: 'rolling' | 'session' | 'calendar_week' | 'calendar_month', unit: 'tokens' | 'usd' | 'requests' = 'tokens', limit: float | None = None)#

The definition of one quota window for a harness (static, plan-derived).

class gate.quota.WindowState(spec: WindowSpec, used: float, reset_at: datetime | None = None, data_source: 'derived' | 'authoritative' = 'derived', tokens: TokenTotals | None = None)#

Live state of one window: how much is spent, when it resets, on what pace.

class gate.quota.HarnessQuota(harness: str, windows: list[WindowState] = <factory>, fetched_at: datetime = <factory>)#

One harness’s windows plus freshness metadata.

class gate.quota.QuotaSnapshot(harnesses: list[HarnessQuota] = <factory>, fetched_at: datetime = <factory>)#

The whole fleet’s quota picture at one instant.

class gate.quota.QuotaEvent(timestamp: datetime, tokens: TokenTotals, model: str, harness: str)#

A timestamped spend event from a harness ledger.

Economic estimation#

gate.quota.estimate_usd(tokens: TokenTotals, model: str) → Decimal | None#

Compute the list-price USD cost from token totals and model price.

Parameters:

  • tokens – The phase-split token tally.

  • model – The model name (matched by substring against the price table).

Returns:

USD cost as a Decimal, or None if the model is unpriced.

gate.quota.estimate_kwh(tokens: TokenTotals, model: str) → float | None#

Compute the physical energy estimate (kWh) from token totals and model energy coeff.

Parameters:

  • tokens – The phase-split token tally.

  • model – The model name (matched by substring against the energy table).

Returns:

Energy in kWh as a float, or None if the model is unpriced.

gate.quota.price_for_model(model: str) → PriceCoeff | None#

Look up the price coefficient by model name substring (case-insensitive, first match).

Window management#

gate.quota.compute_window_state(spec: WindowSpec, events: list[QuotaEvent], now: datetime) → WindowState#

Compute current usage for a rolling/session/calendar window from a list of events.

Parameters:

  • spec – The window definition.

  • events – Timestamped spend events (any order).

  • now – The instant to resolve the window against.

Returns:

The computed WindowState for this window.

Reconciliation#

gate.quota.detect_reset(previous: WindowState, current: WindowState) → bool#

Detect when a window has reset (used dropped or reset instant jumped forward).

Parameters:

  • previous – The prior window state.

  • current – The current window state.

Returns:

True if a reset is detected.

gate.quota.detect_burst(events: list[QuotaEvent], threshold: float, window_minutes: int) → bool#

Detect burst spending within a sliding time window.

Parameters:

  • events – Timestamped spend events (any order).

  • threshold – Minimum total billable tokens within the window to flag a burst.

  • window_minutes – Size of the sliding window in minutes.

Returns:

True if any window of window_minutes duration exceeds threshold.

Rendering#

gate.quota.render_table(snapshot: QuotaSnapshot) → str#

Render a compact text table showing per-harness, per-window usage.

Parameters:

snapshot – The fleet quota snapshot to render.

Returns:

A multi-line string table with percentages, reset times, and burn rates.

gate.quota.render_json(snapshot: QuotaSnapshot) → str#

Serialize a snapshot as a JSON string.

Parameters:

snapshot – The fleet quota snapshot to serialize.

Returns:

A pretty-printed JSON string.