hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 32 Packages Projects Releases Wiki Activity

refactor(agent/subscription_view): share tier-id guard, hoist fixture tables, compact docstrings

Browse Source
This commit is contained in:
Teknium
2026-09-02 18:26:40 -07:00
parent 53895ae69f
commit 33cd2e0063
1 changed files with 35 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
agent/subscription_view.py
View File

@@ -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=<id>[&plan=<tier_id>]``.
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)