From 33cd2e00638b73677363b0474337651ca70f4ec5 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:26:40 -0700 Subject: [PATCH] refactor(agent/subscription_view): share tier-id guard, hoist fixture tables, compact docstrings --- agent/subscription_view.py | 89 +++++++++++++++----------------------- 1 file changed, 35 insertions(+), 54 deletions(-) diff --git a/agent/subscription_view.py b/agent/subscription_view.py index 4a62017a29..253f0e015c 100644 --- a/agent/subscription_view.py +++ b/agent/subscription_view.py @@ -2,12 +2,9 @@ Companion to :mod:`agent.billing_view` — same fail-open philosophy (``logged_in=False`` when not logged in / portal unreachable; never crash) and decimal money end-to-end. - -The TUI ``SubscriptionOverlay`` drives the plan change in-terminal: preview, then -schedule a downgrade / cancellation / resume (chargeless) or apply an upgrade -(charges the subscription card). The portal deep-link (``portal_url`` + ``org_id``) -remains the fallback for an upgrade that needs 3DS / was declined. Until the NAS -``GET /api/billing/subscription`` endpoint ships, 404s take the fail-open path. +The TUI drives plan changes in-terminal (preview, then schedule a downgrade/cancel/resume +or apply an upgrade); the portal deep-link (``portal_url`` + ``org_id``) is the fallback +for an upgrade that needs 3DS / was declined. A 404 from NAS takes the fail-open path. """ from __future__ import annotations @@ -23,16 +20,12 @@ from agent.billing_view import OrgRoleCapability, fetch_portal_state, format_mon logger = logging.getLogger(__name__) -# ── Parsed sub-structures ──────────────────────────────────────────────────── - - @dataclass(frozen=True) class CurrentSubscription: """The user's active subscription. ``None`` (not this object) = no plan. - NAS guarantees a present ``current`` is fully populated: ``tier_id`` / - ``tier_name`` / ``monthly_credits`` / ``cycle_ends_at`` are always set; only - ``credits_remaining`` and the cancel/downgrade fields are optional. + NAS guarantees a present ``current`` has ``tier_id`` / ``tier_name`` / + ``monthly_credits`` / ``cycle_ends_at`` set; the rest are optional. """ tier_id: Optional[str] = None @@ -91,7 +84,7 @@ class SubscriptionState(OrgRoleCapability): logged_in: bool org_name: Optional[str] = None - org_id: Optional[str] = None # org.id from the NAS response + org_id: Optional[str] = None role: Optional[str] = None # "OWNER" | "ADMIN" | "FINANCE_ADMIN" | "SECURITY_ADMIN" | "MEMBER" can_change_plan_raw: Optional[bool] = None context: str = "personal" # "personal" | "team" @@ -104,12 +97,15 @@ class SubscriptionState(OrgRoleCapability): # ── Payload parsing ────────────────────────────────────────────────────────── -def _parse_current(raw: Any) -> Optional[CurrentSubscription]: - # "No plan" is wire-represented as current:null; a present current is a real - # plan, so guard on a real tier id and return None otherwise. +def _tier_id(raw: Any) -> Optional[str]: + """Real tier id of a NAS dict, else None ("no plan" is ``current: null``; junk is skipped).""" if not isinstance(raw, dict): return None - tier_id = raw.get("tierId") or raw.get("id") + return raw.get("tierId") or raw.get("id") or None + + +def _parse_current(raw: Any) -> Optional[CurrentSubscription]: + tier_id = _tier_id(raw) if not tier_id: return None return CurrentSubscription( @@ -128,17 +124,11 @@ def _parse_current(raw: Any) -> Optional[CurrentSubscription]: def _coalesce(*vals: Any) -> Any: """First non-``None`` value. NAS sends ``0`` for the free tier's ``tierOrder`` / ``dollarsPerMonth``, which a plain ``x or default`` would drop.""" - for v in vals: - if v is not None: - return v - return None + return next((v for v in vals if v is not None), None) def _parse_tier(raw: Any) -> Optional[SubscriptionTier]: - """Map one NAS ``SubscriptionTierOption`` dict into a :class:`SubscriptionTier`.""" - if not isinstance(raw, dict): - return None - tier_id = raw.get("tierId") or raw.get("id") + tier_id = _tier_id(raw) if not tier_id: return None return SubscriptionTier( @@ -152,9 +142,7 @@ def _parse_tier(raw: Any) -> Optional[SubscriptionTier]: ) -def subscription_change_preview_from_payload( - payload: dict[str, Any], -) -> SubscriptionChangePreview: +def subscription_change_preview_from_payload(payload: dict[str, Any]) -> SubscriptionChangePreview: """Map a raw ``/subscription/preview`` JSON dict into :class:`SubscriptionChangePreview`.""" effect = payload.get("effect") cents = payload.get("amountDueNowCents") @@ -179,11 +167,7 @@ def subscription_state_from_payload( org, can_change_plan_raw = parse_org_fields(payload) raw_context = payload.get("context") raw_tiers = payload.get("tiers") - tiers = ( - tuple(t for t in (_parse_tier(x) for x in raw_tiers) if t is not None) - if isinstance(raw_tiers, list) - else () - ) + tiers = tuple(filter(None, map(_parse_tier, raw_tiers))) if isinstance(raw_tiers, list) else () return SubscriptionState( logged_in=True, org_name=org.get("name"), @@ -201,11 +185,9 @@ def subscription_state_from_payload( def build_subscription_state(*, timeout: float = 15.0) -> SubscriptionState: - """Fetch + parse ``GET /api/billing/subscription``. Fail-open (see - :func:`agent.billing_view.fetch_portal_state`). + """Fetch + parse ``GET /api/billing/subscription``; fail-open like ``fetch_portal_state``. - ``HERMES_DEV_SUBSCRIPTION_FIXTURE`` short-circuits to a fixture so every - plan/cancel/downgrade/team/not-admin state is testable on CLI and TUI offline. + ``HERMES_DEV_SUBSCRIPTION_FIXTURE`` short-circuits to a fixture so every state is testable offline. """ fixture = dev_fixture_subscription_state() if fixture is not None: @@ -221,17 +203,14 @@ def build_subscription_state(*, timeout: float = 15.0) -> SubscriptionState: ) -def subscription_manage_url( - state: SubscriptionState, tier_id: Optional[str] = None -) -> Optional[str]: +def subscription_manage_url(state: SubscriptionState, tier_id: Optional[str] = None) -> Optional[str]: """Build ``{portal_origin}/manage-subscription?org_id=[&plan=]``. Mirrors the TUI's ``buildManageUrl``: the target is NAS's OWN ``/manage-subscription`` - page (NOT the Stripe Billing Portal), which routes upgrade→Checkout / - downgrade→scheduled internally. ``org_id`` pins the right account in multi-org - situations. ``tier_id`` (the stable ``tiers[]`` id, never a name/slug) preselects - the picked plan; the portal ignores an unknown tier, so it's appended - unconditionally when picked. None when no portal URL is resolvable. + page (NOT the Stripe Billing Portal), which routes upgrade→Checkout / downgrade→scheduled. + ``org_id`` pins the account in multi-org situations; ``tier_id`` (the stable ``tiers[]`` + id, never a name/slug) preselects the plan — the portal ignores an unknown tier, so it's + appended unconditionally. None when no portal URL is resolvable. """ from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit @@ -281,18 +260,23 @@ def format_tier_row(tier: SubscriptionTier) -> str: def is_upgrade(state: SubscriptionState, tier_id: str) -> bool: """True when ``tier_id`` ranks above the current plan by ``tier_order``. Prefers the active subscription's tier; falls back to the ``tiers[]`` ``is_current`` marker, else 0.""" - orders = {t.tier_id: (t.tier_order or 0) for t in (state.tiers or ())} + tiers = state.tiers or () + orders = {t.tier_id: (t.tier_order or 0) for t in tiers} cur_id = state.current.tier_id if state.current else None if cur_id is not None and cur_id in orders: cur_order = orders[cur_id] else: - cur_order = next((t.tier_order or 0 for t in (state.tiers or ()) if t.is_current), 0) + cur_order = next((t.tier_order or 0 for t in tiers if t.is_current), 0) return orders.get(tier_id, 0) > cur_order -# ── Dev fixtures (throwaway scaffolding — env-var driven, no live portal) ──── +# ── Dev fixtures (env-var driven, no live portal) ──────────────────────────── _DEV_FIXTURE_PORTAL = "https://portal.nousresearch.com/billing" +_DEV_TIER_SPECS = (("free", "Free", 0, "0", "0"), ("plus", "Plus", 1, "20", "1000"), + ("super", "Super", 2, "40", "3000"), ("ultra", "Ultra", 3, "80", "7000")) +_DEV_FIXTURE_ALIASES = {"logged_out": "logged-out", "loggedout": "logged-out", "mid-tier": "mid", + "top-tier": "top", "member": "not-admin"} def _dev_current(**over: Any) -> CurrentSubscription: @@ -304,15 +288,13 @@ def _dev_current(**over: Any) -> CurrentSubscription: def _dev_tiers(current_id: Optional[str]) -> tuple[SubscriptionTier, ...]: - """A sample plan catalog for fixtures (marks ``current_id`` as the active tier).""" - specs = (("free", "Free", 0, "0", "0"), ("plus", "Plus", 1, "20", "1000"), - ("super", "Super", 2, "40", "3000"), ("ultra", "Ultra", 3, "80", "7000")) + """Sample plan catalog for fixtures (marks ``current_id`` as the active tier).""" return tuple( SubscriptionTier( tier_id=tid, name=name, tier_order=order, dollars_per_month=parse_money(dpm), monthly_credits=parse_money(mc), is_current=(tid == current_id), is_enabled=True, ) - for tid, name, order, dpm, mc in specs + for tid, name, order, dpm, mc in _DEV_TIER_SPECS ) @@ -325,8 +307,7 @@ def dev_fixture_subscription_state() -> Optional[SubscriptionState]: name = (os.getenv("HERMES_DEV_SUBSCRIPTION_FIXTURE") or "").strip().lower() if not name: return None - name = {"logged_out": "logged-out", "loggedout": "logged-out", "mid-tier": "mid", - "top-tier": "top", "member": "not-admin"}.get(name, name) + name = _DEV_FIXTURE_ALIASES.get(name, name) if name == "logged-out": return SubscriptionState(logged_in=False)