"""Shared dollar-denominated usage model for the ``/usage`` and ``/subscription`` bars. Terminal surfaces show **dollars**, never "credits"; the plan allowance and top-up dollars stay distinctly visible as two SEPARATE bars (a three-segment bar is unreadable at terminal widths). Source: ``NousPortalAccountInfo.paid_service_access_info`` (USD floats despite the legacy ``*_credits`` names) plus ``subscription.monthly_credits`` (plan bar denominator) and ``current_period_end``. Fail-open: missing/non-finite fields degrade to fewer bars; logged-out / unreachable portal yields ``available=False``. """ from __future__ import annotations import logging import math import os from dataclasses import dataclass from typing import Any, Optional from hermes_time import safe_strftime logger = logging.getLogger(__name__) # Below this TOTAL spendable ($) a paid account is flagged "low" — the alert state # that nudges top-up/upgrade before a mid-run cutoff (product: "below $5 is an alert"). LOW_BALANCE_THRESHOLD_USD = 5.0 def _finite(value: Any) -> Optional[float]: """Float iff a real finite number (not bool/NaN/Inf); json.loads admits bare NaN, which would render ``$nan``.""" if isinstance(value, bool) or not isinstance(value, (int, float)): return None f = float(value) return f if math.isfinite(f) else None def _fmt_usd(value: Optional[float]) -> str: """``$X,XXX.YY`` for display. ``None`` -> ``$0.00`` (callers gate on presence).""" return f"${(value or 0.0):,.2f}" def nous_logged_in() -> bool: """Cheap local auth-state check: a Nous access token is present. Fail-closed.""" try: from hermes_cli.auth import get_provider_auth_state tok = (get_provider_auth_state("nous") or {}).get("access_token") return isinstance(tok, str) and bool(tok.strip()) except Exception: return False def fetch_nous_account(timeout: float): """Wall-clock-bounded fresh portal account fetch. Raises on failure/timeout. Shares the one bounded implementation so a stalled portal releases the /billing surface at ``timeout`` too (see ``_fetch_portal_account``).""" from agent.account_usage import _fetch_portal_account return _fetch_portal_account(timeout) def format_renews(value: Optional[str]) -> Optional[str]: """ISO date/timestamp -> ``Jul 24, 2026``; unparseable input is returned unchanged.""" from datetime import datetime text = str(value or "").strip() if not text: return None iso = text[:-1] + "+00:00" if text.endswith("Z") else text try: dt = datetime.fromisoformat(iso) except ValueError: try: dt = datetime.strptime(text[:10], "%Y-%m-%d") except ValueError: return text # %-d isn't portable to Windows; build the day without a leading zero. return f"{safe_strftime(dt, '%b')} {dt.day}, {dt.year}" @dataclass(frozen=True) class UsageBar: """One bar: ``spent`` of ``total`` plus remaining. ``plan`` shows % used; ``topup`` has no denominator (``spent`` is 0 and ``total == remaining`` so it renders full).""" kind: str # "plan" | "topup" remaining_usd: float total_usd: float spent_usd: float = 0.0 @property def pct_used(self) -> Optional[int]: if self.kind != "plan" or self.total_usd <= 0: return None return max(0, min(100, round(self.spent_usd / self.total_usd * 100))) @property def fill_fraction(self) -> float: """Fraction of the bar that should read as 'remaining' (filled).""" return max(0.0, min(1.0, self.remaining_usd / self.total_usd)) if self.total_usd > 0 else 0.0 @dataclass(frozen=True) class UsageModel: """Dollar usage model shared by /usage and /subscription. ``status``: ``free`` (no plan / no paid access), ``low`` (spendable < $5, ALERT), ``healthy`` (>= $5), ``depleted`` (paid access lost).""" available: bool status: str = "free" plan_name: Optional[str] = None renews_at: Optional[str] = None renews_display: Optional[str] = None subscription_remaining_usd: Optional[float] = None topup_remaining_usd: Optional[float] = None total_spendable_usd: Optional[float] = None plan_bar: Optional[UsageBar] = None topup_bar: Optional[UsageBar] = None @property def has_topup(self) -> bool: return bool(self.topup_remaining_usd and self.topup_remaining_usd > 0) def usage_model_from_account(account_info: Any) -> UsageModel: """Build a :class:`UsageModel` from a ``NousPortalAccountInfo``. Never raises.""" try: if account_info is None or not getattr(account_info, "logged_in", False): return UsageModel(available=False) # Sub-structs are dataclasses or None; getattr(None, ..., None) is None, so no guards needed. access = getattr(account_info, "paid_service_access_info", None) sub = getattr(account_info, "subscription", None) paid = getattr(account_info, "paid_service_access", None) sub_remaining = _finite(getattr(access, "subscription_credits_remaining", None)) topup_remaining = _finite(getattr(access, "purchased_credits_remaining", None)) total_usable = _finite(getattr(access, "total_usable_credits", None)) plan_name = getattr(sub, "plan", None) renews_at = getattr(sub, "current_period_end", None) monthly = _finite(getattr(sub, "monthly_credits", None)) has_subscription = bool(plan_name) or (monthly is not None and monthly > 0) has_topup = bool(topup_remaining and topup_remaining > 0) # Prefer the server's total; else sum the parts we have. parts = [v for v in (sub_remaining, topup_remaining) if v is not None] total_spendable = total_usable if total_usable is not None else (sum(parts) if parts else None) if paid is False: status = "depleted" elif not has_subscription and not has_topup: status = "free" # no plan and no purchased balance -> free-models-only elif total_spendable is not None and total_spendable < LOW_BALANCE_THRESHOLD_USD: status = "low" else: status = "healthy" # Plan bar needs a positive allowance AND a remaining to place on it; spent is # clamped so a debt/over-cap balance reads fully spent, not negative. plan_bar = topup_bar = None if monthly is not None and monthly > 0 and sub_remaining is not None: plan_bar = UsageBar(kind="plan", remaining_usd=max(0.0, min(monthly, sub_remaining)), total_usd=monthly, spent_usd=max(0.0, monthly - sub_remaining)) # Top-up has no monthly cap, so the bar renders full = balance. if topup_remaining is not None and topup_remaining > 0: topup_bar = UsageBar(kind="topup", remaining_usd=topup_remaining, total_usd=topup_remaining, spent_usd=0.0) return UsageModel( available=True, status=status, plan_name=plan_name, renews_at=renews_at, renews_display=format_renews(renews_at), subscription_remaining_usd=sub_remaining, topup_remaining_usd=topup_remaining, total_spendable_usd=total_spendable, plan_bar=plan_bar, topup_bar=topup_bar, ) except Exception: logger.debug("usage ▸ model build failed (fail-open)", exc_info=True) return UsageModel(available=False) def build_usage_model(*, timeout: float = 10.0) -> UsageModel: """Fetch account-info and build the usage model; fail-open. ``HERMES_DEV_CREDITS_FIXTURE`` short-circuits to a fixture.""" fixture = _dev_fixture_usage_model() if fixture is not None: return fixture if not nous_logged_in(): return UsageModel(available=False) try: return usage_model_from_account(fetch_nous_account(timeout)) except Exception: logger.debug("usage ▸ portal fetch failed (fail-open)", exc_info=True) return UsageModel(available=False) def _plan_bar(remaining: float, spent: float) -> UsageBar: return UsageBar(kind="plan", remaining_usd=remaining, total_usd=20.0, spent_usd=spent) def _dev_fixture_usage_model() -> Optional[UsageModel]: """``HERMES_DEV_CREDITS_FIXTURE`` -> fixture model (``free|healthy|low|topup|depleted``), else None.""" name = (os.getenv("HERMES_DEV_CREDITS_FIXTURE") or "").strip().lower() name = {"mid": "healthy", "top-up": "topup"}.get(name, name) plus = dict(available=True, plan_name="Plus", renews_at="2026-07-01") specs: dict[str, dict] = { "free": dict(available=True, status="free", plan_name=None), "healthy": dict(**plus, status="healthy", subscription_remaining_usd=14.0, total_spendable_usd=14.0, plan_bar=_plan_bar(14.0, 6.0)), "topup": dict( **plus, status="healthy", subscription_remaining_usd=14.0, topup_remaining_usd=12.0, total_spendable_usd=26.0, plan_bar=_plan_bar(14.0, 6.0), topup_bar=UsageBar(kind="topup", remaining_usd=12.0, total_usd=12.0, spent_usd=0.0), ), "low": dict(**plus, status="low", subscription_remaining_usd=3.4, total_spendable_usd=3.4, plan_bar=_plan_bar(3.4, 16.6)), "depleted": dict(**plus, status="depleted", subscription_remaining_usd=0.0, total_spendable_usd=0.0, plan_bar=_plan_bar(0.0, 20.0)), } spec = specs.get(name) return UsageModel(**spec) if spec else None