"""Surface-agnostic core for the Phase 2b terminal-billing screens. One fetch/parse per concern, consumed identically by the CLI handler (``cli.py::_show_billing``), the TUI JSON-RPC methods (``tui_gateway/server.py``), and any other surface. Mirrors the proven ``agent/account_usage.py::build_credits_view`` pattern: parse the server payload into a frozen dataclass; **fail open** — when not logged in or the portal is unreachable, return a struct with ``logged_in=False`` and let the surface degrade gracefully (never crash). Money discipline: the server emits decimal STRINGS (``"142.5"``, not fixed 2dp). We keep them as :class:`decimal.Decimal` end-to-end and only format for display. """ from __future__ import annotations import logging import uuid from dataclasses import dataclass, field from decimal import Decimal, InvalidOperation from typing import Any, Optional logger = logging.getLogger(__name__) # ============================================================================= # Decimal money helpers # ============================================================================= def parse_money(value: Any) -> Optional[Decimal]: """Parse a server money value (decimal string) into :class:`Decimal`. Returns None for missing/invalid input. Never raises. Accepts str/int (and, defensively, float — though the server always sends strings). """ if value is None: return None try: # Decimal(str(...)) avoids binary-float artifacts if a float ever sneaks in. return Decimal(str(value).strip()) except (InvalidOperation, ValueError, TypeError): return None def format_money(value: Optional[Decimal]) -> str: """Format a Decimal as ``$X`` / ``$X.YY`` for display. Whole dollars show no decimals; any fractional amount shows exactly 2dp: ``Decimal("142.5")`` → ``"$142.50"``, ``Decimal("100")`` → ``"$100"``, ``Decimal("0.01")`` → ``"$0.01"``. """ if value is None: return "—" if value == value.to_integral_value(): # Whole dollars — no decimal point. format(..., "f") avoids 1E+3 for 1000. return f"${format(value.to_integral_value(), 'f')}" # Fractional — always show 2dp. return f"${format(value.quantize(Decimal('0.01')), 'f')}" # ============================================================================= # Parsed sub-structures # ============================================================================= @dataclass(frozen=True) class CardInfo: brand: str last4: str @property def masked(self) -> str: return f"{self.brand} ····{self.last4}" @dataclass(frozen=True) class MonthlyCap: limit_usd: Optional[Decimal] = None spent_this_month_usd: Optional[Decimal] = None is_default_ceiling: bool = False @dataclass(frozen=True) class AutoReload: enabled: bool = False threshold_usd: Optional[Decimal] = None reload_to_usd: Optional[Decimal] = None @dataclass(frozen=True) class BillingState: """Parsed ``GET /api/billing/state`` — the overview screen's data. Fail-open: ``logged_in=False`` (and empty fields) when not logged in or the portal is unreachable. """ logged_in: bool org_id: Optional[str] = None org_slug: Optional[str] = None org_name: Optional[str] = None role: Optional[str] = None # "OWNER" | "ADMIN" | "MEMBER" balance_usd: Optional[Decimal] = None cli_billing_enabled: bool = False charge_presets: tuple[Decimal, ...] = () min_usd: Optional[Decimal] = None max_usd: Optional[Decimal] = None card: Optional[CardInfo] = None monthly_cap: Optional[MonthlyCap] = None auto_reload: Optional[AutoReload] = None portal_url: Optional[str] = None # When the fetch failed (vs cleanly not-logged-in), the message for the surface. error: Optional[str] = None @property def is_admin(self) -> bool: """True for OWNER/ADMIN — the roles that can manage billing.""" return (self.role or "").upper() in ("OWNER", "ADMIN") @property def can_charge(self) -> bool: """True when the UI should offer charge/auto-reload actions. Admin role AND the per-org kill-switch on. (The server still enforces; this is just for graying out actions the user can't take.) """ return self.is_admin and self.cli_billing_enabled def _parse_card(raw: Any) -> Optional[CardInfo]: if not isinstance(raw, dict): return None brand = raw.get("brand") last4 = raw.get("last4") if isinstance(brand, str) and isinstance(last4, str): return CardInfo(brand=brand, last4=last4) return None def _parse_monthly_cap(raw: Any) -> Optional[MonthlyCap]: if not isinstance(raw, dict): return None return MonthlyCap( limit_usd=parse_money(raw.get("limitUsd")), spent_this_month_usd=parse_money(raw.get("spentThisMonthUsd")), is_default_ceiling=bool(raw.get("isDefaultCeiling")), ) def _parse_auto_reload(raw: Any) -> Optional[AutoReload]: if not isinstance(raw, dict): return None return AutoReload( enabled=bool(raw.get("enabled")), threshold_usd=parse_money(raw.get("thresholdUsd")), reload_to_usd=parse_money(raw.get("reloadToUsd")), ) def billing_state_from_payload( payload: dict[str, Any], *, portal_url: Optional[str] = None ) -> BillingState: """Map a raw ``/api/billing/state`` JSON dict into :class:`BillingState`.""" raw_org = payload.get("org") org: dict[str, Any] = raw_org if isinstance(raw_org, dict) else {} raw_bounds = payload.get("bounds") bounds: dict[str, Any] = raw_bounds if isinstance(raw_bounds, dict) else {} presets: list[Decimal] = [] for item in payload.get("chargePresets") or (): parsed = parse_money(item) if parsed is not None: presets.append(parsed) return BillingState( logged_in=True, org_id=org.get("id"), org_slug=org.get("slug"), org_name=org.get("name"), role=org.get("role"), balance_usd=parse_money(payload.get("balanceUsd")), cli_billing_enabled=bool(payload.get("cliBillingEnabled")), charge_presets=tuple(presets), min_usd=parse_money(bounds.get("minUsd")), max_usd=parse_money(bounds.get("maxUsd")), card=_parse_card(payload.get("card")), monthly_cap=_parse_monthly_cap(payload.get("monthlyCap")), auto_reload=_parse_auto_reload(payload.get("autoReload")), portal_url=portal_url, ) # ============================================================================= # Fail-open builders (the surface front doors) # ============================================================================= def build_billing_state(*, timeout: float = 15.0) -> BillingState: """Fetch + parse ``/api/billing/state``. Fail-open. Returns ``BillingState(logged_in=False)`` when not logged in. On a portal/HTTP failure, returns ``logged_in=False`` with ``error`` set so the surface can show a clear message rather than crashing. """ try: from hermes_cli.nous_billing import ( BillingAuthError, BillingError, _absolutize_portal_url, get_billing_state, resolve_portal_base_url, ) except Exception: return BillingState(logged_in=False, error="billing client unavailable") try: payload = get_billing_state(timeout=timeout) except BillingAuthError: return BillingState(logged_in=False) except BillingError as exc: logger.debug("billing ▸ /state fetch failed (fail-open)", exc_info=True) return BillingState(logged_in=False, error=str(exc)) except Exception: logger.debug("billing ▸ /state unexpected error (fail-open)", exc_info=True) return BillingState(logged_in=False, error="could not load billing state") # Prefer a server-supplied portalUrl if present (resolved to absolute in case # it's relative); else build the standard one. raw_portal = payload.get("portalUrl") if isinstance(payload, dict) else None portal_url = _absolutize_portal_url(raw_portal) if raw_portal else None if not portal_url: try: portal_url = _fallback_portal_url(resolve_portal_base_url()) except Exception: portal_url = None return billing_state_from_payload(payload, portal_url=portal_url) def _fallback_portal_url(base: str) -> str: """Standard billing deep-link when the server omits ``portalUrl``.""" return f"{base.rstrip('/')}/billing?topup=open" # ============================================================================= # Idempotency # ============================================================================= def new_idempotency_key() -> str: """Fresh UUID for a user-confirmed purchase (reuse on retry of the SAME buy). The ``Idempotency-Key`` header is mandatory on ``POST /charge``; generate one per confirmed purchase and reuse it across retries so a double-submit collapses to a single charge. Never reuse a key across different amounts (the server returns 409 idempotency_conflict). """ return str(uuid.uuid4()) # ============================================================================= # Amount validation (Screen 3 custom input) # ============================================================================= @dataclass(frozen=True) class AmountValidation: ok: bool amount: Optional[Decimal] = None error: Optional[str] = None def validate_charge_amount( raw: str, *, min_usd: Optional[Decimal], max_usd: Optional[Decimal] ) -> AmountValidation: """Validate a custom charge amount against bounds + 2dp (multipleOf 0.01). Mirrors the server's accept/reject so the UI can give instant feedback rather than round-tripping a sure-to-fail charge. The server is still authoritative. """ cleaned = (raw or "").strip().lstrip("$").strip() amount = parse_money(cleaned) if amount is None: return AmountValidation(ok=False, error="Enter a dollar amount, e.g. 100") if amount <= 0: return AmountValidation(ok=False, error="Amount must be greater than $0") # multipleOf 0.01 — reject sub-cent precision. if amount != amount.quantize(Decimal("0.01")): return AmountValidation(ok=False, error="Amount can't be smaller than a cent") if min_usd is not None and amount < min_usd: return AmountValidation(ok=False, error=f"Minimum is {format_money(min_usd)}") if max_usd is not None and amount > max_usd: return AmountValidation(ok=False, error=f"Maximum is {format_money(max_usd)}") return AmountValidation(ok=True, amount=amount)