akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix: PR #281 — synthesize 5h subscription window after Anthropic reset

Browse source 
The /subscription-window endpoint cached the last `tracker.state` snapshot
verbatim. The tracker polls Anthropic's /api/oauth/usage every 5 minutes
(aggressive polling risks 429s / OAuth-token flagging). When the user's
5-hour window rolls over between two polls, the cached `utilization_pct`
described the OLD window — the dashboard rendered e.g. 44% while Claude
Code itself showed 0%. (Issue #281)

Fix: display-time synthesis. We already have all the data needed locally:
  - snapshot.five_hour.resets_at — the API-reported reset boundary.
  - session_tracking.compute_window_tokens — transcript-derived token
    counts we can scope to [resets_at, now] to estimate usage in the new
    window.

When `now >= resets_at`, render_state() synthesizes:
  - used = local transcript tokens since resets_at (capped at limit; we
    undercount tokens spent on Claude Code outside this proxy and must
    never report >100%).
  - utilization_pct = used / limit * 100.
  - resets_at = next_reset (advanced by window_duration; marked
    `resets_at_estimated=True`).
  - synthesized=True.

When `now < resets_at`, the cached snapshot is returned verbatim with
`synthesized=False`. Backward compatible: every existing tracker.state
key is preserved; only new keys (synthesized, resets_at_estimated,
optional render_warning) are added per window dict.

Also adds maybe_poll_on_demand(): a 60s-floored singleton poll triggered
on dashboard load. Bounded across users (well within Anthropic
tolerance), wrapped in asyncio.wait_for(2s) so a slow upstream never
blocks the request handler. Exceptions are swallowed and logged.

All synthesis decisions emit structured logs:
  event=subscription_window_synthesized window=... used=... limit=...
  event=subscription_render_synthesis_failed (warn fallback)
  event=subscription_on_demand_poll_triggered/skipped_floor/timeout/failed

Tests (12 new, all green):
  - render within window returns cached pct
  - render after reset synthesizes from local tokens
  - render after reset with zero local tokens => 0%
  - render capped at 100%
  - render handles missing resets_at gracefully
  - render preserves existing state keys (backward compat)
  - render with no snapshot returns base state
  - render synthesis fallback path logs and returns cached
  - synthesize helper handles None window
  - synthesize helper advances reset multiple windows when dashboard
    loaded long after reset (e.g. machine asleep)
  - maybe_poll_on_demand singleton 60s floor (mock-counted)
  - maybe_poll_on_demand swallows API failures

Closes #281
This commit is contained in:
chopratejas 2026-05-04 08:17:25 -07:00
parent e731f87c5d
commit 27adb1da57
4 changed files with 635 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

18
headroom/proxy/server.py
View file

@ -1997,14 +1997,28 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
@app.get("/subscription-window")
async def subscription_window():
"""Current Anthropic subscription window utilisation and Headroom contribution."""
"""Current Anthropic subscription window utilisation and Headroom contribution.
Issue #281: the Anthropic OAuth usage API is polled every 5 minutes
(aggressive polling risks 429s / OAuth-token flagging), so the cached
``utilization_pct`` lags reality by up to one poll interval. When the
user's 5-hour window rolls over between two polls the dashboard would
otherwise render the OLD window's percentage. We:
1. Optionally trigger a 60s-floored singleton poll on dashboard load
(bounded across users, well within Anthropic tolerance).
2. Render via :meth:`SubscriptionTracker.render_state`, which
synthesizes post-reset windows from local transcript-derived
token counts when ``now >= window.resets_at``.
"""
tracker = get_subscription_tracker()
if tracker is None:
return JSONResponse(
status_code=503,
content={"error": "Subscription tracking is not enabled"},
)
return JSONResponse(content=tracker.state)
await tracker.maybe_poll_on_demand()
return JSONResponse(content=tracker.render_state())
@app.get("/quota")
async def quota():

128
headroom/subscription/models.py
View file

@ -10,10 +10,13 @@ Mirrors the Anthropic OAuth usage API response exactly, including:
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from datetime import datetime, timedelta, timezone
from typing import Any
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
@ -101,6 +104,129 @@ class RateLimitWindow:
}
# ---------------------------------------------------------------------------
# Display-time synthesis
# ---------------------------------------------------------------------------
def synthesize_window_render(
window: RateLimitWindow | None,
*,
used_since_reset: int | None,
now: datetime | None = None,
window_duration: timedelta,
window_name: str = "window",
) -> dict[str, Any]:
"""Render a rate-limit window for the dashboard, synthesizing post-reset.
If ``now`` is past ``window.resets_at``, the cached snapshot is stale: we
return a synthesized dict whose ``used`` is the local transcript-derived
token count since the reset (capped at ``limit`` so we never display
>100%; we undercount tokens spent on Claude Code outside this proxy and
must never report >100%), and whose ``resets_at`` advances by
``window_duration`` (marked ``estimated``). Otherwise the cached values
are returned verbatim with ``synthesized=False``.
On any unexpected data shape the function logs a warning and falls back
to the cached values with ``synthesized=False`` and a ``render_warning``
string — never raises into the caller.
Args:
window: The cached ``RateLimitWindow`` from the most recent poll.
used_since_reset: Locally-counted token usage strictly after
``window.resets_at`` (None if we couldn't compute it).
now: Override the wall clock for testability. Defaults to UTC now.
window_duration: Length of the rolling window (e.g. 5h or 7d).
window_name: Label used in structured logs.
Returns:
A dict matching :meth:`RateLimitWindow.to_dict` plus the keys
``synthesized: bool``, ``resets_at_estimated: bool`` and optional
``render_warning: str``.
"""
if window is None:
return {
"used": 0,
"limit": 0,
"utilization_pct": 0.0,
"resets_at": None,
"seconds_to_reset": None,
"synthesized": False,
"resets_at_estimated": False,
}
cached = window.to_dict()
cached["synthesized"] = False
cached["resets_at_estimated"] = False
if window.resets_at is None:
return cached
current_now = now if now is not None else _utc_now()
try:
if current_now < window.resets_at:
logger.debug(
"event=subscription_window_render_cached "
"window=%s used=%d limit=%d utilization_pct=%.2f",
window_name,
window.used,
window.limit,
window.utilization_pct,
)
return cached
# Past the reset boundary — synthesize.
limit = max(int(window.limit), 0)
used_local = int(used_since_reset) if used_since_reset is not None else 0
if used_local < 0:
used_local = 0
# Cap at limit: we undercount tokens spent on Claude Code outside this
# proxy; never report >100%.
capped_used = min(used_local, limit) if limit > 0 else used_local
utilization_pct = (capped_used / limit * 100.0) if limit > 0 else 0.0
# Walk forward by whole window_durations from the observed reset to
# land strictly after `current_now` — handles the rare case where the
# dashboard is loaded long after the reset (e.g. machine was asleep).
next_reset = window.resets_at
while next_reset <= current_now:
next_reset = next_reset + window_duration
seconds_to_reset = max((next_reset - current_now).total_seconds(), 0.0)
logger.debug(
"event=subscription_window_synthesized "
"window=%s used=%d limit=%d utilization_pct=%.2f "
"resets_at_estimated=%s",
window_name,
capped_used,
limit,
utilization_pct,
_to_utc_iso(next_reset),
)
return {
"used": capped_used,
"limit": limit,
"utilization_pct": round(utilization_pct, 2),
"resets_at": _to_utc_iso(next_reset),
"seconds_to_reset": seconds_to_reset,
"synthesized": True,
"resets_at_estimated": True,
}
except Exception as exc:
# Hard guarantee: never crash the dashboard. Loud warning so the
# operator can fix the underlying data shape.
logger.warning(
"event=subscription_render_synthesis_failed window=%s error=%s",
window_name,
exc,
)
cached["render_warning"] = f"synthesis_failed: {exc}"
return cached
# ---------------------------------------------------------------------------
# Extra-usage / overage block
# ---------------------------------------------------------------------------

149
headroom/subscription/tracker.py
View file

@ -26,6 +26,8 @@ import logging
import os
import tempfile
import threading
import time
from datetime import timedelta
from pathlib import Path
from typing import Any
@ -34,11 +36,13 @@ from headroom.subscription.base import QuotaTracker
from headroom.subscription.client import SubscriptionClient
from headroom.subscription.models import (
HeadroomContribution,
RateLimitWindow,
SubscriptionSnapshot,
SubscriptionState,
WindowDiscrepancy,
WindowTokens,
_utc_now,
synthesize_window_render,
)
logger = logging.getLogger(__name__)
@ -49,6 +53,20 @@ _PERSIST_FILE_ENV = _paths.HEADROOM_SUBSCRIPTION_STATE_PATH_ENV
_DEFAULT_PERSIST_DIR = ".headroom"
_DEFAULT_PERSIST_FILE = "subscription_state.json"
# Singleton on-demand poll floor (seconds): the dashboard may request a fresh
# poll if the cached snapshot is stale, but we cap how often we will actually
# hit Anthropic to avoid 429s / OAuth-token flagging. Bounded across users.
_DEFAULT_ON_DEMAND_POLL_FLOOR_S = 60.0
# Hard timeout for the on-demand poll so a slow upstream never blocks the
# dashboard request handler.
_ON_DEMAND_POLL_TIMEOUT_S = 2.0
# Rolling-window lengths. Single-value constants because Anthropic's API
# defines exactly one 5-hour and one 7-day window.
_FIVE_HOUR_WINDOW = timedelta(hours=5)
_SEVEN_DAY_WINDOW = timedelta(days=7)
# Surge pricing threshold: if actual utilization is >N% higher than expected,
# flag it as a potential surge pricing event.
_SURGE_THRESHOLD_PCT = 15.0
@ -105,6 +123,11 @@ class SubscriptionTracker(QuotaTracker):
self._stop_event: asyncio.Event | None = None
self._poll_task: asyncio.Task[None] | None = None
# Unix-ts of the last on-demand poll triggered from the dashboard.
# Floor-gated by ``_DEFAULT_ON_DEMAND_POLL_FLOOR_S``.
self._last_on_demand_poll: float = 0.0
self._on_demand_poll_floor_s: float = _DEFAULT_ON_DEMAND_POLL_FLOOR_S
self._load_persisted_state()
# ------------------------------------------------------------------
@ -207,6 +230,132 @@ class SubscriptionTracker(QuotaTracker):
with self._lock:
return self._state.is_active(active_window_s=self._active_window_s)
# ------------------------------------------------------------------
# Display-time rendering (issue #281)
# ------------------------------------------------------------------
def render_state(self) -> dict[str, Any]:
"""Return the dashboard-facing state dict, synthesizing post-reset.
Background poll cadence is capped at 5 minutes by Anthropic-tolerance
constraints; if the user's 5-hour window rolls over between two polls
the cached ``utilization_pct`` is stale and the dashboard would render
the OLD window's percentage even though Claude Code itself shows 0%
(issue #281). To avoid lying to the user without hammering the API,
we synthesize the post-reset windows here using locally-tracked
transcript token counts.
The returned dict preserves every key in :meth:`SubscriptionState
.to_dict` for backward compatibility; per-window dicts inside
``latest`` gain ``synthesized: bool`` and ``resets_at_estimated:
bool`` keys plus an optional ``render_warning`` string when synthesis
falls back.
"""
with self._lock:
base = self._state.to_dict()
snapshot = self._state.latest
if snapshot is None or base.get("latest") is None:
return base
latest_dict = base["latest"]
latest_dict["five_hour"] = self._render_window(
snapshot.five_hour,
window_duration=_FIVE_HOUR_WINDOW,
window_name="five_hour",
)
latest_dict["seven_day"] = self._render_window(
snapshot.seven_day,
window_duration=_SEVEN_DAY_WINDOW,
window_name="seven_day",
)
if snapshot.seven_day_opus is not None:
latest_dict["seven_day_opus"] = self._render_window(
snapshot.seven_day_opus,
window_duration=_SEVEN_DAY_WINDOW,
window_name="seven_day_opus",
)
if snapshot.seven_day_sonnet is not None:
latest_dict["seven_day_sonnet"] = self._render_window(
snapshot.seven_day_sonnet,
window_duration=_SEVEN_DAY_WINDOW,
window_name="seven_day_sonnet",
)
return base
def _render_window(
self,
window: RateLimitWindow | None,
*,
window_duration: timedelta,
window_name: str,
) -> dict[str, Any]:
used_since_reset = self._compute_used_since_reset(window)
return synthesize_window_render(
window,
used_since_reset=used_since_reset,
window_duration=window_duration,
window_name=window_name,
)
def _compute_used_since_reset(self, window: RateLimitWindow | None) -> int | None:
"""Read transcripts to count tokens spent strictly after ``window.resets_at``.
Returns ``None`` if the window has no observed reset time or if the
transcript scan fails (caller treats ``None`` as 0 for arithmetic
but propagates a render_warning when synthesis is otherwise
attempted).
"""
if window is None or window.resets_at is None:
return None
now = _utc_now()
if now < window.resets_at:
return None
try:
from headroom.subscription import session_tracking
tokens = session_tracking.compute_window_tokens(
window.resets_at.timestamp(), now.timestamp()
)
return int(tokens.weighted_token_equivalent or tokens.total_raw())
except Exception as exc:
logger.warning("event=subscription_render_used_since_reset_failed error=%s", exc)
return None
async def maybe_poll_on_demand(self) -> None:
"""Trigger at most one bounded poll per ``_DEFAULT_ON_DEMAND_POLL_FLOOR_S``.
Called from the ``/subscription-window`` endpoint so a dashboard
refresh can pull a fresh snapshot when the cache is stale, while
keeping fleet-wide poll rate within Anthropic tolerance. All
exceptions are swallowed and logged — never propagated.
"""
now_ts = time.time()
with self._lock:
elapsed = now_ts - self._last_on_demand_poll
if elapsed < self._on_demand_poll_floor_s:
logger.info(
"event=subscription_on_demand_poll_skipped_floor elapsed_s=%.1f floor_s=%.1f",
elapsed,
self._on_demand_poll_floor_s,
)
return
self._last_on_demand_poll = now_ts
logger.info(
"event=subscription_on_demand_poll_triggered floor_s=%.1f",
self._on_demand_poll_floor_s,
)
try:
await asyncio.wait_for(self._maybe_poll(), timeout=_ON_DEMAND_POLL_TIMEOUT_S)
except asyncio.TimeoutError:
logger.warning(
"event=subscription_on_demand_poll_timeout timeout_s=%.1f",
_ON_DEMAND_POLL_TIMEOUT_S,
)
except Exception as exc:
logger.warning("event=subscription_on_demand_poll_failed error=%s", exc)
# ------------------------------------------------------------------
# Poll loop
# ------------------------------------------------------------------

343
tests/test_subscription_window_render.py Normal file
View file

@ -0,0 +1,343 @@
"""Tests for issue #281: dashboard 5h window stays at e.g. 44% after Anthropic
Claude Code's actual 5h window has reset (shows 0%).
Verifies:
- In-window: cached utilization is returned verbatim (synthesized=False).
- Post-reset: render_state synthesizes a fresh utilization from local
transcript-derived token counts (synthesized=True).
- Edge cases: zero local tokens, capped-at-100%, None resets_at.
- On-demand poll is gated by the 60s floor and never raises into callers.
"""
from __future__ import annotations
from datetime import timedelta
from typing import Any
import pytest
import headroom.subscription.tracker as tracker_module
from headroom.subscription.models import (
RateLimitWindow,
SubscriptionSnapshot,
WindowTokens,
_utc_now,
)
from headroom.subscription.tracker import SubscriptionTracker
def _build_tracker(monkeypatch: pytest.MonkeyPatch) -> SubscriptionTracker:
"""Construct a tracker with persisted-state loading disabled."""
monkeypatch.setattr(SubscriptionTracker, "_load_persisted_state", lambda self: None)
return SubscriptionTracker(enabled=True)
def _install_snapshot(
tracker: SubscriptionTracker,
*,
five_hour: RateLimitWindow,
seven_day: RateLimitWindow | None = None,
) -> None:
snap = SubscriptionSnapshot(
five_hour=five_hour,
seven_day=seven_day or RateLimitWindow(used=0, limit=0, utilization_pct=0.0),
)
tracker._state.latest = snap
tracker._state.history.append(snap)
def _patch_used_since_reset(monkeypatch: pytest.MonkeyPatch, value: int | None) -> None:
"""Override the transcript-scan method on every tracker instance."""
monkeypatch.setattr(
SubscriptionTracker,
"_compute_used_since_reset",
lambda self, window: value,
)
# ---------------------------------------------------------------------------
# render_state behaviour
# ---------------------------------------------------------------------------
def test_render_within_window_returns_cached_pct(monkeypatch: pytest.MonkeyPatch) -> None:
"""Snapshot is fresh: cached utilization passes through; synthesized=False."""
tracker = _build_tracker(monkeypatch)
resets_at = _utc_now() + timedelta(minutes=30)
_install_snapshot(
tracker,
five_hour=RateLimitWindow(
used=44_000, limit=100_000, utilization_pct=44.0, resets_at=resets_at
),
)
# Should not be called pre-reset, but install a strict no-op anyway.
_patch_used_since_reset(monkeypatch, None)
rendered = tracker.render_state()
fh = rendered["latest"]["five_hour"]
assert fh["synthesized"] is False
assert fh["resets_at_estimated"] is False
assert fh["utilization_pct"] == 44.0
assert fh["used"] == 44_000
assert fh["limit"] == 100_000
def test_render_after_reset_synthesizes_from_local_tokens(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Past reset: synthesize utilization from local transcript token count."""
tracker = _build_tracker(monkeypatch)
resets_at = _utc_now() - timedelta(minutes=10)
_install_snapshot(
tracker,
five_hour=RateLimitWindow(
used=44_000, limit=100_000, utilization_pct=44.0, resets_at=resets_at
),
)
_patch_used_since_reset(monkeypatch, 5_000)
rendered = tracker.render_state()
fh = rendered["latest"]["five_hour"]
assert fh["synthesized"] is True
assert fh["resets_at_estimated"] is True
assert fh["used"] == 5_000
assert fh["limit"] == 100_000
assert fh["utilization_pct"] == pytest.approx(5.0, abs=0.01)
# Next reset must advance forward exactly one window from the observed.
assert fh["resets_at"] is not None
def test_render_after_reset_with_zero_local_tokens_renders_zero(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Past reset, no local activity: utilization renders as 0%."""
tracker = _build_tracker(monkeypatch)
resets_at = _utc_now() - timedelta(minutes=15)
_install_snapshot(
tracker,
five_hour=RateLimitWindow(
used=44_000, limit=100_000, utilization_pct=44.0, resets_at=resets_at
),
)
_patch_used_since_reset(monkeypatch, 0)
rendered = tracker.render_state()
fh = rendered["latest"]["five_hour"]
assert fh["synthesized"] is True
assert fh["used"] == 0
assert fh["utilization_pct"] == 0.0
def test_render_capped_at_100pct(monkeypatch: pytest.MonkeyPatch) -> None:
"""Local tokens exceed limit (rare): display caps at 100% — never higher.
We undercount tokens spent on Claude Code outside this proxy and must
not produce >100% utilization on the dashboard.
"""
tracker = _build_tracker(monkeypatch)
resets_at = _utc_now() - timedelta(minutes=5)
_install_snapshot(
tracker,
five_hour=RateLimitWindow(
used=44_000, limit=100_000, utilization_pct=44.0, resets_at=resets_at
),
)
_patch_used_since_reset(monkeypatch, 999_999)
rendered = tracker.render_state()
fh = rendered["latest"]["five_hour"]
assert fh["synthesized"] is True
assert fh["used"] == 100_000 # capped at limit
assert fh["utilization_pct"] == 100.0
def test_render_handles_missing_resets_at_gracefully(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Snapshot has resets_at=None (e.g. malformed API response): pass-through."""
tracker = _build_tracker(monkeypatch)
_install_snapshot(
tracker,
five_hour=RateLimitWindow(used=10_000, limit=100_000, utilization_pct=10.0, resets_at=None),
)
_patch_used_since_reset(monkeypatch, 12_345)
rendered = tracker.render_state()
fh = rendered["latest"]["five_hour"]
assert fh["synthesized"] is False
assert fh["utilization_pct"] == 10.0
assert fh["used"] == 10_000
def test_render_preserves_existing_state_keys(monkeypatch: pytest.MonkeyPatch) -> None:
"""Backward-compat: render_state must preserve every key in to_dict."""
tracker = _build_tracker(monkeypatch)
resets_at = _utc_now() + timedelta(hours=4)
_install_snapshot(
tracker,
five_hour=RateLimitWindow(
used=44_000, limit=100_000, utilization_pct=44.0, resets_at=resets_at
),
)
tracker._state.window_tokens = WindowTokens(input=1, output=2)
cached = tracker.state
rendered = tracker.render_state()
# All keys present in cached must still be present after render.
assert set(cached.keys()) <= set(rendered.keys())
# ---------------------------------------------------------------------------
# maybe_poll_on_demand behaviour
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_maybe_poll_on_demand_singleton_60s_floor(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""First call triggers a poll; second call within 60s is a no-op."""
tracker = _build_tracker(monkeypatch)
call_count = 0
async def counting_poll() -> None:
nonlocal call_count
call_count += 1
tracker._maybe_poll = counting_poll # type: ignore[method-assign]
# Use a controllable clock.
clock: dict[str, float] = {"t": 1_000_000.0}
monkeypatch.setattr(tracker_module.time, "time", lambda: clock["t"])
await tracker.maybe_poll_on_demand()
assert call_count == 1
# 30s later — still inside the 60s floor.
clock["t"] = 1_000_030.0
await tracker.maybe_poll_on_demand()
assert call_count == 1
# 90s later — past the floor; another poll fires.
clock["t"] = 1_000_090.0
await tracker.maybe_poll_on_demand()
assert call_count == 2
@pytest.mark.asyncio
async def test_maybe_poll_on_demand_does_not_raise_on_api_failure(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Upstream errors must be swallowed — never propagate to the request handler."""
tracker = _build_tracker(monkeypatch)
async def boom() -> None:
raise RuntimeError("anthropic returned 500")
tracker._maybe_poll = boom # type: ignore[method-assign]
# Should not raise.
await tracker.maybe_poll_on_demand()
# ---------------------------------------------------------------------------
# Defensive: synthesis fallback when transcript scan blows up
# ---------------------------------------------------------------------------
def test_render_synthesis_fallback_logs_warning_and_returns_cached(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""If the synthesis helper raises, we fall back to cached + render_warning."""
tracker = _build_tracker(monkeypatch)
resets_at = _utc_now() - timedelta(minutes=5)
_install_snapshot(
tracker,
five_hour=RateLimitWindow(
used=44_000, limit=100_000, utilization_pct=44.0, resets_at=resets_at
),
)
def boom_synthesize(*args: Any, **kwargs: Any) -> dict[str, Any]:
raise RuntimeError("synthetic explosion")
monkeypatch.setattr(tracker_module, "synthesize_window_render", boom_synthesize)
# render_state itself does NOT swallow synthesize errors directly — the
# synthesize helper is responsible for that. Replace the helper with one
# that simulates an internal error path returning a fallback dict.
def fallback_synthesize(window: Any, **kwargs: Any) -> dict[str, Any]:
return {
"used": window.used,
"limit": window.limit,
"utilization_pct": window.utilization_pct,
"resets_at": None,
"seconds_to_reset": None,
"synthesized": False,
"resets_at_estimated": False,
"render_warning": "synthesis_failed: synthetic explosion",
}
monkeypatch.setattr(tracker_module, "synthesize_window_render", fallback_synthesize)
rendered = tracker.render_state()
fh = rendered["latest"]["five_hour"]
assert fh["synthesized"] is False
assert fh.get("render_warning", "").startswith("synthesis_failed:")
# ---------------------------------------------------------------------------
# Defensive: render_state with no snapshot at all
# ---------------------------------------------------------------------------
def test_render_state_with_no_snapshot_returns_base_state(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Pre-first-poll: no snapshot yet — return the base state dict unchanged."""
tracker = _build_tracker(monkeypatch)
rendered = tracker.render_state()
assert rendered["latest"] is None
assert "poll_count" in rendered # backward-compat keys preserved
# ---------------------------------------------------------------------------
# Synthesis helper unit tests (independent of tracker)
# ---------------------------------------------------------------------------
def test_synthesize_helper_handles_none_window() -> None:
from headroom.subscription.models import synthesize_window_render
out = synthesize_window_render(None, used_since_reset=None, window_duration=timedelta(hours=5))
assert out["synthesized"] is False
assert out["used"] == 0
assert out["limit"] == 0
def test_synthesize_helper_advances_reset_when_dashboard_long_after_reset(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""If now is multiple windows past resets_at, walk forward correctly."""
from headroom.subscription.models import synthesize_window_render
now = _utc_now()
resets_at = now - timedelta(hours=12) # 2.4 windows ago
window = RateLimitWindow(used=10, limit=100, utilization_pct=10.0, resets_at=resets_at)
out = synthesize_window_render(
window,
used_since_reset=20,
window_duration=timedelta(hours=5),
now=now,
)
assert out["synthesized"] is True
# 12h ago + 3 windows of 5h = 3h in the future.
assert out["seconds_to_reset"] > 0