feat(billing): /billing terminal billing — interactive TUI + CLI client (#45449)

* feat(billing): nous_billing http client + BillingState core (phase 2b)

Phase 2b terminal-billing client foundation:
- hermes_cli/nous_billing.py: typed client for the 4 /api/billing/* endpoints
  (state/charge/poll/auto-top-up). Raises typed errors (BillingScopeRequired,
  BillingRateLimited, BillingAuthError) mapped from the live-verified contract;
  fail-open is the caller's job. Idempotency-Key enforced client-side.
- agent/billing_view.py: surface-agnostic BillingState core + Decimal money
  parsing (server emits decimal strings, not 2dp), fail-open builder,
  idempotency-key gen, custom-amount validation.
- 51 unit tests (decimal parse/format, payload tiering, error->exception
  matrix, fail-open, amount validation).

Plan: docs/plans/2026-06-13-001-phase-2b-terminal-billing-tui-plan.md

* feat(billing): billing:manage scope + lazy step-up re-auth (phase 2b)

- NOUS_BILLING_MANAGE_SCOPE constant.
- nous_token_has_billing_scope(): split-based scope check (no false-positive
  substring match).
- step_up_nous_billing_scope(): re-runs the device flow requesting
  billing:manage, reusing the held credential's portal/inference URLs + client_id
  (so a preview stays a preview), persists like _login_nous but WITHOUT the model
  picker. Returns True iff the minted token carries the scope (False when NAS
  silently downscopes a non-admin / unticked grant).

Lazy step-up (plan D-A): normal login path unchanged; 403 insufficient_scope
from a billing call triggers this. 7 unit tests.

* feat(billing): billing JSON-RPC methods for the TUI (phase 2b)

billing.state / charge / charge_status / auto_reload / step_up in
tui_gateway/server.py. Return STRUCTURED success envelopes (result.ok +
result.error=<code>) rather than JSON-RPC-level errors, so the Ink rpc() promise
always resolves and the TUI branches on the typed billing error code
(insufficient_scope, rate_limited, no_payment_method, …) to render the right
affordance. Money serialized as decimal STRINGS + display strings. charge mints
+ echoes an idempotency_key for retry reuse. 16 unit tests.

* feat(billing): /billing CLI handler + command registry (phase 2b)

- CommandDef("billing", subcommands=buy|auto-reload|limit), added to
  _SLACK_VIA_HERMES_ONLY so it routes via /hermes on Slack (keeps the 50-cap
  parity test green, same as /credits).
- cli.py::_show_billing + screen helpers: all 5 screens (overview, buy→confirm→
  poll, auto-reload, monthly-limit read-only). Reuses _prompt_text_input_modal /
  _prompt_text_input (D-C). Non-interactive (_app is None) renders text + portal
  deep-link, never prompts (R7). Decimal money end-to-end. 2s/5-min cancellable
  poll loop; 429/503 = retry not failure; settled = ledger truth. Lazy step-up on
  403 insufficient_scope. no_payment_method treated as mainline funnel-to-portal.
- 6 CLI tests; 156 command tests (incl. Slack/Telegram parity) green.

* feat(billing): /billing Ink TUI screens + tests (phase 2b)

- ui-tui/src/app/slash/commands/billing.ts: /billing TUI command covering all 5
  screens — overview (text), buy <amt> → ConfirmReq → charge → non-blocking 2s/
  5-min poll loop → settled/failed/timeout branches, auto-reload <below> <to> →
  ConfirmReq → PATCH, limit (read-only). Reuses the existing ConfirmReq overlay
  (D-C) — no bespoke component. Typed-error envelope branching: insufficient_scope
  arms the lazy step-up confirm; no_payment_method/rate_limited/cap funnel to
  portal. Client-side amount validation mirrors the server (bounds + 2dp).
- gatewayTypes.ts: Billing* response interfaces.
- registry.ts: register billingCommands.
- billingCommand.test.ts: 12 vitest cases (overview/gating/buy-confirm-poll-
  settled/no_payment_method/step-up/limit/auto-reload/validation).

TUI build green; 12/12 vitest pass; slash tests pass once @hermes/ink is built.

* docs(billing): scrub private cross-repo references

NAS is a private repo — remove all references to it from the public PR:
- drop the cross-repo planning doc (planning scaffolding, not a deliverable;
  the PR description documents the design)
- replace 'NAS' / 'PR #412 preview' mentions in code + test comments with
  generic 'the server' / 'a preview deployment'

* docs(billing): scrub final NAS reference in step-up docstring

* docs(billing): drop dangling plan-doc refs

The phase-2b plan doc was removed in the cross-repo scrub (300afcc0b)
but two module docstrings still pointed at it. Drop the dead refs.

* feat(billing): interactive /billing overlay + step-up UX, portal-URL & token fixes

Adds the interactive /billing TUI overlay and hardens the terminal-billing
client across CLI and TUI.

- TUI: full /billing overlay state machine (overview to buy to confirm,
  auto-reload, read-only monthly limit) reusing the existing confirm overlay.
- Step-up: surface the verification link in-transcript and open the browser
  via the TUI's own opener (the device flow runs in the headless gateway, so a
  printed URL was being dropped); run the step-up handler off the main loop and
  emit the link as an out-of-band event so the gateway stays responsive.
- Step-up copy is scope-accurate ("Billing permission granted") and re-checks
  /state so it never claims "enabled" when the org kill-switch is still off.
- Portal deep-links resolve to absolute URLs against the active portal base
  (the server emits them relative) - fixes a bare "/billing?topup=open" link.
- Billing calls refresh an expired access token via the stored refresh token
  instead of reporting a false "not logged in".
- Optimistic funnel: advise "set up a saved card on the portal" up front when
  no card is on file (advisory, not a hard gate).
- Token resolution is cached briefly so the 2s charge poll loop stops
  re-locking + re-reading the auth store on every tick; 401 re-resolves fresh.
- Remove the temporary demo-mode shims.

Validation: 87 Python billing tests, 88 TS tests (billing command + gateway
event handler), tsc clean, ink + ui-tui builds green.

* docs(billing): add /billing TUI screenshots for PR

* fix(cli): guard _last_invalidate on bare instances; update stale prompt-fallback test

The UI-invalidate throttle read self._last_invalidate unconditionally, which
raised AttributeError on HermesCLI instances built without __init__ (the
thread-safety test's object.__new__ shell). Guard the read with getattr.

The off-main-thread branch of _prompt_text_input was changed (#23185) to cancel
cleanly to None instead of falling back to a bare input() that would hang on the
slash-worker thread; the test still asserted the old direct-input fallback.
Update it to assert the current intended behavior: returns None, calls neither
run_in_terminal nor input(), and does not hang.
This commit is contained in:
Siddharth Balyan 2026-06-19 01:53:32 +05:30 committed by GitHub
parent 81eaedd0f5
commit 73cd8622f9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
25 changed files with 4203 additions and 22 deletions

Binary file not shown.

After

Width:  |  Height:  |  Size: 138 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 148 KiB

295
agent/billing_view.py Normal file
View file

@ -0,0 +1,295 @@
"""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)

657
cli.py
View file

@ -1984,6 +1984,24 @@ _ACCENT = _SkinAwareAnsi("response_border", "#FFD700", bold=True)
_DIM = "\x1b[2;3m"
def _b(s: str) -> str:
"""Bold if stdout is a real TTY; plain text otherwise (slash-worker safe)."""
import sys as _sys
try:
return f"\x1b[1m{s}\x1b[0m" if _sys.stdout.isatty() else str(s)
except Exception:
return str(s)
def _d(s: str) -> str:
"""Dim-italic if stdout is a real TTY; plain text otherwise."""
import sys as _sys
try:
return f"\x1b[2;3m{s}\x1b[0m" if _sys.stdout.isatty() else str(s)
except Exception:
return str(s)
def _accent_hex() -> str:
"""Return the active skin accent color for legacy CLI output lines."""
try:
@ -3664,7 +3682,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
if getattr(self, "_resize_recovery_pending", False):
return
now = time.monotonic()
if hasattr(self, "_app") and self._app and (now - self._last_invalidate) >= min_interval:
if hasattr(self, "_app") and self._app and (now - getattr(self, "_last_invalidate", 0.0)) >= min_interval:
self._last_invalidate = now
self._app.invalidate()
@ -6359,6 +6377,17 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
in_main_thread = threading.current_thread() is threading.main_thread()
# Slash-worker guard (#23185 / billing auto-reload hang): when a
# prompt_toolkit app is running but we're on a non-main thread (the
# process_loop / TUI slash-worker daemon thread), stdin is owned by the
# event loop / JSON-RPC pipe. A bare input() there blocks forever until
# the worker's 45s timeout fires. We cannot safely prompt off the main
# thread, so cancel cleanly (None) instead of hanging — mirrors the
# _stdin_fallback discipline in _prompt_text_input_modal.
if self._app and not in_main_thread:
self._invalidate()
return None
if self._app and in_main_thread:
from prompt_toolkit.application import run_in_terminal
was_visible = self._status_bar_visible
@ -7506,6 +7535,8 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
self._show_usage()
elif canonical == "credits":
self._show_credits()
elif canonical == "billing":
self._show_billing(cmd_original)
elif canonical == "insights":
self._show_insights(cmd_original)
elif canonical == "copy":
@ -8425,7 +8456,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
if not view.logged_in:
print()
print(f" 💳 {_DIM}Not logged into Nous Portal.{_RST}")
_cprint(f" 💳 {_d('Not logged into Nous Portal.')}")
print(" Run `hermes portal` to log in, then /credits.")
return
@ -8487,6 +8518,628 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
else:
print(" 🟡 Cancelled. No credits added.")
# ------------------------------------------------------------------
# /billing — Phase 2b terminal billing (CLI surface, all 5 screens)
# ------------------------------------------------------------------
def _show_billing(self, command: str = "/billing"):
"""`/billing` — terminal billing for Nous (one interactive modal).
ZERO sub-commands: any argument is ignored. Bare ``/billing`` always
opens the Overview (Screen 1), whose numbered menu is the *only* way to
reach the Buy / Auto-reload / Monthly-limit sub-screens. (Per the unified
UX spec §0.4 ``/billing buy`` etc. are gone; we don't error on a stray
arg, we just open the menu.)
Interactive CLI uses the prompt_toolkit modal; non-interactive contexts
(TUI slash-worker / no live app) render text + the portal deep-link, never
prompting (the URL is the affordance), same discipline as ``_show_credits``.
All money is Decimal end-to-end; the terminal never collects card details.
"""
from agent.billing_view import build_billing_state
state = build_billing_state()
if not state.logged_in:
print()
if state.error:
_msg = f"Couldn't load billing: {state.error}"
_cprint(f" 💳 {_d(_msg)}")
else:
_cprint(f" 💳 {_d('Not logged into Nous Portal.')}")
print(" Run `hermes portal` to log in, then /billing.")
return
# Any sub-arg is intentionally ignored — always open the menu.
self._billing_overview(state)
def _billing_portal_hint(self, state, *, reason: str = "") -> None:
"""Print a portal deep-link line (the funnel for portal-only actions)."""
url = getattr(state, "portal_url", None)
if not url:
return
if reason:
print(f" {reason}")
print(f" Manage on portal: {url}")
def _billing_overview(self, state):
"""Screen 1 — overview: balance, spend bar, role-gated action menu."""
from agent.billing_view import format_money
print()
_cprint(f" 💳 {_b('Usage credits')}")
print(f" {'' * 41}")
cap = state.monthly_cap
if cap is not None and cap.limit_usd is not None:
spent = format_money(cap.spent_this_month_usd)
limit = format_money(cap.limit_usd)
ceiling = " (default ceiling)" if cap.is_default_ceiling else ""
bar, pct = self._billing_spend_bar(
cap.spent_this_month_usd, cap.limit_usd
)
print(f" {spent} of {limit} used{ceiling} {bar} {pct}%")
print(f" Balance: {format_money(state.balance_usd)}")
ar = state.auto_reload
if ar is not None:
if ar.enabled:
print(
f" Auto-reload: on — below {format_money(ar.threshold_usd)} "
f"→ reload to {format_money(ar.reload_to_usd)}"
)
else:
print(" Auto-reload: off")
if state.org_name:
role = (state.role or "").title()
_org_line = f"Org: {state.org_name}{f' · {role}' if role else ''}"
_cprint(f" {_d(_org_line)}")
print(f" {'' * 41}")
# Action gating: admin + kill-switch for charge/auto-reload; everyone gets portal.
if not state.is_admin:
_cprint(f" {_d('Billing actions require an org admin/owner.')}")
self._billing_portal_hint(state)
return
if not state.cli_billing_enabled:
_cprint(f" {_d('Terminal billing is turned off for this org.')}")
self._billing_portal_hint(state, reason="Enable it on the portal to buy credits here.")
return
# Optimistic funnel: no card on file → a charge will 403 no_payment_method.
# Surface that up front (with the portal link) but DON'T hide Buy — /state.card
# can't fully prove CLI-chargeability, so we advise rather than gate.
if state.card is None:
_cprint(
f" {_d('No saved card for terminal charges yet — set one up on the portal first.')}"
)
self._billing_portal_hint(state)
# Non-interactive (slash-worker / no live app): no modal, no sub-command
# advertising — just the portal funnel (the URL is the affordance).
if not getattr(self, "_app", None):
self._billing_portal_hint(state)
return
choices = [
("buy", "Buy credits", "purchase a one-time credit top-up"),
("auto", "Adjust auto-reload", "configure automatic top-ups"),
("limit", "Adjust monthly limit", "show the monthly spend cap (read-only)"),
("portal", "Manage on portal", "open the billing page in your browser"),
("cancel", "Cancel", "do nothing"),
]
# The overview summary is already printed above; the modal only needs to
# present the action menu — repeating the title/balance reads as a dupe.
raw = self._prompt_text_input_modal(
title="💳 Choose an action", detail="",
choices=choices,
)
choice = self._normalize_slash_confirm_choice(raw, choices)
if choice == "buy":
self._billing_buy_flow(state)
elif choice == "auto":
self._billing_auto_reload_flow(state)
elif choice == "limit":
self._billing_limit_screen(state)
elif choice == "portal":
self._billing_open_portal(state)
else:
print(" 🟡 Cancelled.")
def _billing_spend_bar(self, spent, limit, *, cells: int = 10):
"""Render a 10-cell `█`/`░` spend bar + integer percent from spent/limit.
Returns ``(bar, pct)`` where ``bar`` is like ``[]`` and ``pct``
is the spent/limit percentage clamped to 0..100. Box-drawing glyphs are
not SGR codes, so this is leak-safe even without ``_b()``/``_d()``.
"""
from decimal import Decimal
try:
s = Decimal(str(spent)) if spent is not None else Decimal("0")
l = Decimal(str(limit)) if limit is not None else Decimal("0")
except Exception:
s, l = Decimal("0"), Decimal("0")
if l <= 0:
pct = 0
else:
pct = int((s / l) * 100)
pct = max(0, min(100, pct))
filled = int(round(pct / 100 * cells))
filled = max(0, min(cells, filled))
bar = ("" * filled) + ("" * (cells - filled))
return bar, pct
def _billing_open_portal(self, state):
url = getattr(state, "portal_url", None)
if not url:
print(" No portal URL available.")
return
opened = False
try:
import webbrowser
opened = webbrowser.open(url)
except Exception:
opened = False
if not opened:
print(f" Open this URL: {url}")
print(" Complete billing changes in the browser.")
def _billing_require_admin(self, state) -> bool:
"""Guard charge/auto-reload entry points; print + return False if blocked."""
if not state.is_admin:
print()
_cprint(f" 💳 {_d('Billing actions require an org admin/owner.')}")
self._billing_portal_hint(state)
return False
if not state.cli_billing_enabled:
print()
_cprint(f" 💳 {_d('Terminal billing is turned off for this org.')}")
self._billing_portal_hint(state, reason="Enable it on the portal first.")
return False
return True
def _billing_buy_flow(self, state):
"""Screen 2 (preset select) → Screen 3 (confirm + charge + poll)."""
from agent.billing_view import format_money, validate_charge_amount
if not self._billing_require_admin(state):
return
# Screen 3 — preset selection.
if not getattr(self, "_app", None):
presets = ", ".join(format_money(p) for p in state.charge_presets)
print()
_cprint(f" 💳 {_b('Buy usage credits')}")
print(f" Presets: {presets}")
print(" Run this in the interactive CLI to complete a purchase.")
self._billing_portal_hint(state)
return
preset_choices = []
for p in state.charge_presets:
preset_choices.append((str(p), format_money(p), "one-time credit purchase"))
preset_choices.append(("custom", "Custom amount…", "enter your own amount"))
preset_choices.append(("cancel", "Cancel", "do nothing"))
card = state.card
detail = f"Payment: {card.masked}" if card else "No saved card on file"
raw = self._prompt_text_input_modal(
title="💳 Buy usage credits", detail=detail, choices=preset_choices,
)
choice = self._normalize_slash_confirm_choice(raw, preset_choices)
if not choice or choice == "cancel":
print(" 🟡 Cancelled. No credits added.")
return
from decimal import Decimal
if choice == "custom":
entered = self._prompt_text_input(" Amount (USD): ")
if entered is None:
# None = cancelled (e.g. slash-worker can't prompt off-thread).
print(" 🟡 Cancelled. No credits added.")
return
v = validate_charge_amount(
entered or "", min_usd=state.min_usd, max_usd=state.max_usd
)
if not v.ok:
print(f" 🔴 {v.error}")
return
amount = v.amount
else:
try:
amount = Decimal(choice)
except Exception:
print(" 🔴 Invalid selection.")
return
self._billing_confirm_and_charge(state, amount)
def _billing_confirm_and_charge(self, state, amount):
"""Screen 3 — confirm total + consent, charge, then poll to settlement."""
from agent.billing_view import format_money, new_idempotency_key
card = state.card
print()
_cprint(f" 💳 {_b('Confirm purchase')}")
print(f" {'' * 41}")
print(f" Total: {format_money(amount)}")
if card:
print(f" Payment: {card.masked}")
print(f" {'' * 41}")
_consent = (
"By confirming, you allow Nous Research to charge your card."
)
_cprint(f" {_d(_consent)}")
confirm_choices = [
("pay", f"Pay {format_money(amount)} now", "submit the charge"),
("cancel", "Go back", "do not charge"),
]
if not getattr(self, "_app", None):
print(" Run in the interactive CLI to confirm a purchase.")
return
raw = self._prompt_text_input_modal(
title=f"💳 Pay {format_money(amount)}?",
detail=(card.masked if card else "no saved card"),
choices=confirm_choices,
)
choice = self._normalize_slash_confirm_choice(raw, confirm_choices)
if choice != "pay":
print(" 🟡 Cancelled. No credits added.")
return
# Submit the charge with a fresh idempotency key (reused on retry).
from hermes_cli.nous_billing import (
BillingError,
BillingScopeRequired,
post_charge,
)
key = new_idempotency_key()
try:
result = post_charge(amount_usd=amount, idempotency_key=key)
except BillingScopeRequired:
self._billing_handle_scope_required(state)
return
except BillingError as exc:
self._billing_render_charge_error(state, exc)
return
charge_id = result.get("chargeId")
if not charge_id:
print(" 🔴 No charge id returned; please check the portal.")
return
_cprint(f" {_d('Charge submitted — confirming settlement…')}")
self._billing_poll_charge(state, charge_id, amount)
def _billing_poll_charge(self, state, charge_id, amount):
"""Poll loop: 2s interval, 5-min cap, cancellable. settled = ledger truth."""
import time as _time
from agent.billing_view import format_money
from hermes_cli.nous_billing import (
BillingError,
BillingRateLimited,
get_charge_status,
)
deadline = _time.time() + 300 # 5-minute cap
interval = 2.0
while _time.time() < deadline:
try:
status = get_charge_status(charge_id)
except BillingRateLimited as exc:
# Retry-after, NOT a failure — back off and keep polling.
wait = exc.retry_after or 5
_time.sleep(min(wait, 30))
continue
except BillingError as exc:
print(f" 🔴 Could not check the charge: {exc}")
return
state_str = status.get("status")
if state_str == "settled":
amt = status.get("amountUsd")
from agent.billing_view import parse_money
shown = format_money(parse_money(amt)) if amt else format_money(amount)
print(f"{shown} in credits added.")
return
if state_str == "failed":
self._billing_render_charge_failed(state, status.get("reason"))
return
# pending → wait and poll again
_time.sleep(interval)
# Past the cap with no terminal state = timeout (not an error).
print(f" 🟡 Still processing after 5 minutes — this is a timeout, not a "
f"failure. Check /billing or the portal shortly.")
self._billing_portal_hint(state)
def _billing_render_charge_failed(self, state, reason):
"""Branch the poll `failed` reasons to the right copy + portal funnel."""
reason = (reason or "").strip()
if reason == "authentication_required":
print(" 🔴 Your bank requires verification (3DS). Complete it on the "
"portal to finish this purchase.")
elif reason == "payment_method_expired":
print(" 🔴 Your card has expired. Update it on the portal.")
elif reason == "card_declined":
print(" 🔴 Your card was declined. Try another card on the portal.")
else:
print(f" 🔴 The charge didn't go through ({reason or 'processing_error'}).")
self._billing_portal_hint(state)
def _billing_render_charge_error(self, state, exc):
"""Render a typed BillingError at submit time (pre-poll)."""
from hermes_cli.nous_billing import BillingRateLimited
code = getattr(exc, "error", None)
portal_url = getattr(exc, "portal_url", None) or getattr(state, "portal_url", None)
if code == "no_payment_method":
print(" 💳 No saved card for terminal charges yet. Set one up on the "
"portal (one-time credit buys don't save a reusable card).")
elif code == "cli_billing_disabled":
print(" 🔴 Terminal billing is turned off for this org — an admin must enable it on the portal.")
elif code == "monthly_cap_exceeded":
remaining = (getattr(exc, "payload", {}) or {}).get("remainingUsd")
if remaining is not None:
print(f" 🔴 Monthly spend cap reached — ${remaining} headroom left.")
else:
print(" 🔴 Monthly spend cap reached.")
elif isinstance(exc, BillingRateLimited):
wait = getattr(exc, "retry_after", None)
mins = f" (try again in ~{max(1, round(wait / 60))} min)" if wait else ""
print(f" 🟡 Too many charges right now{mins}. This isn't a payment failure.")
else:
print(f" 🔴 {exc}")
if portal_url:
print(f" Portal: {portal_url}")
def _billing_handle_scope_required(self, state):
"""403 insufficient_scope → lazy step-up re-auth (plan D-A)."""
print()
print(" 💳 Terminal billing needs an extra permission (billing:manage).")
_scope_msg = (
"An org admin/owner must tick \"Allow terminal billing\" during "
"login."
)
_cprint(f" {_d(_scope_msg)}")
if not getattr(self, "_app", None):
print(" Run `hermes portal` and approve terminal billing, then retry.")
return
confirm_choices = [
("yes", "Re-authorize now", "open the portal to grant billing access"),
("no", "Not now", "cancel"),
]
raw = self._prompt_text_input_modal(
title="💳 Grant terminal billing access?",
detail="Opens the portal device-authorization page.",
choices=confirm_choices,
)
choice = self._normalize_slash_confirm_choice(raw, confirm_choices)
if choice != "yes":
print(" 🟡 Cancelled.")
return
try:
from hermes_cli.auth import step_up_nous_billing_scope
granted = step_up_nous_billing_scope(open_browser=True)
except Exception as exc:
print(f" 🔴 Re-authorization failed: {exc}")
return
if granted:
print(" ✅ Billing permission granted.")
# Step-up only grants the billing:manage TOKEN scope; the ORG
# kill-switch (cli_billing_enabled) is a separate gate. Re-fetch
# /state so we don't over-promise when a charge would still hit
# cli_billing_disabled.
from agent.billing_view import build_billing_state
fresh = build_billing_state()
if fresh.logged_in and fresh.cli_billing_enabled:
print(" Run /billing buy again to continue.")
else:
print(" 🟡 Permission granted, but terminal billing is still turned "
"off for this org. Enable it in the portal, then run /billing again.")
self._billing_portal_hint(fresh)
else:
print(" 🟡 Terminal billing was not granted (an admin must tick the box).")
def _billing_auto_reload_flow(self, state):
"""Screen 4 — auto-reload config: threshold + reload-to → PATCH.
Prefills the current values from ``state.auto_reload``. Validates both
amounts (2dp, within bounds, ``reload_to > threshold``). When auto-reload
is already on, offers a "Turn off" path (PATCH ``enabled:false``).
"""
from agent.billing_view import format_money, validate_charge_amount
if not self._billing_require_admin(state):
return
card = state.card
ar = state.auto_reload
currently_on = bool(ar and ar.enabled)
print()
_cprint(f" 💳 {_b('Auto-reload')}")
print(f" {'' * 41}")
_cprint(f" {_d('Automatically buy more credits when your balance is low.')}")
if card:
print(f" Card on file: {card.masked}")
else:
print(" No saved card — set one up on the portal first.")
self._billing_portal_hint(state)
return
if currently_on:
print(
f" Currently: below {format_money(ar.threshold_usd)}"
f"reload to {format_money(ar.reload_to_usd)}"
)
if not getattr(self, "_app", None):
print(" Run in the interactive CLI to configure auto-reload.")
self._billing_portal_hint(state)
return
# When already enabled, let the user turn it off without re-entering values.
if currently_on:
top_choices = [
("edit", "Edit thresholds", "change when / how much to reload"),
("off", "Turn off", "disable auto-reload"),
("cancel", "Cancel", "do nothing"),
]
raw = self._prompt_text_input_modal(
title="💳 Auto-reload",
detail=(
f"On — below {format_money(ar.threshold_usd)}"
f"reload to {format_money(ar.reload_to_usd)}"
),
choices=top_choices,
)
top = self._normalize_slash_confirm_choice(raw, top_choices)
if top == "off":
self._billing_auto_reload_disable(state)
return
if top != "edit":
print(" 🟡 Cancelled.")
return
# Field 1 — threshold (prefilled when editing an existing config).
cur_thr = format_money(ar.threshold_usd) if currently_on else None
thr_prompt = " When balance falls below (USD)"
thr_prompt += f" [{cur_thr}]: " if cur_thr else ": "
threshold_raw = self._prompt_text_input(thr_prompt)
if threshold_raw is None:
# None = cancelled (e.g. slash-worker can't prompt off-thread).
print(" 🟡 Cancelled.")
return
if not (threshold_raw or "").strip() and currently_on:
threshold_amt = ar.threshold_usd # keep current value on empty input
else:
tv = validate_charge_amount(
threshold_raw or "", min_usd=state.min_usd, max_usd=state.max_usd
)
if not tv.ok or tv.amount is None:
print(f" 🔴 {tv.error}")
return
threshold_amt = tv.amount
# Field 2 — reload-to (prefilled when editing an existing config).
cur_rel = format_money(ar.reload_to_usd) if currently_on else None
rel_prompt = " Reload balance to (USD)"
rel_prompt += f" [{cur_rel}]: " if cur_rel else ": "
reload_raw = self._prompt_text_input(rel_prompt)
if reload_raw is None:
print(" 🟡 Cancelled.")
return
if not (reload_raw or "").strip() and currently_on:
reload_amt = ar.reload_to_usd # keep current value on empty input
else:
rv = validate_charge_amount(
reload_raw or "", min_usd=state.min_usd, max_usd=state.max_usd
)
if not rv.ok or rv.amount is None:
print(f" 🔴 {rv.error}")
return
reload_amt = rv.amount
if reload_amt is None or threshold_amt is None or reload_amt <= threshold_amt:
print(" 🔴 Reload-to amount must be greater than the threshold.")
return
print()
_ar_consent = (
f"By confirming, you authorize Nous Research to charge {card.masked} "
f"whenever your balance reaches {format_money(threshold_amt)}. "
f"Turn off any time here or on the portal."
)
_cprint(f" {_d(_ar_consent)}")
confirm_choices = [
("agree", "Agree and turn on", "enable auto-reload"),
("cancel", "Cancel", "do nothing"),
]
raw = self._prompt_text_input_modal(
title="💳 Turn on auto-reload?",
detail=f"Below {format_money(threshold_amt)} → reload to {format_money(reload_amt)}",
choices=confirm_choices,
)
choice = self._normalize_slash_confirm_choice(raw, confirm_choices)
if choice != "agree":
print(" 🟡 Cancelled.")
return
from hermes_cli.nous_billing import (
BillingError,
BillingScopeRequired,
patch_auto_top_up,
)
try:
patch_auto_top_up(
enabled=True, threshold=float(threshold_amt), top_up_amount=float(reload_amt)
)
except BillingScopeRequired:
self._billing_handle_scope_required(state)
return
except BillingError as exc:
self._billing_render_charge_error(state, exc)
return
print(f" ✅ Auto-reload on: below {format_money(threshold_amt)}"
f"reload to {format_money(reload_amt)}.")
def _billing_auto_reload_disable(self, state):
"""Turn off auto-reload (PATCH ``enabled:false``).
The endpoint requires ``threshold``/``topUpAmount`` in the body even when
disabling, so we echo back the current values (falling back to 0).
"""
from hermes_cli.nous_billing import (
BillingError,
BillingScopeRequired,
patch_auto_top_up,
)
ar = state.auto_reload
thr = float(ar.threshold_usd) if ar and ar.threshold_usd is not None else 0.0
rel = float(ar.reload_to_usd) if ar and ar.reload_to_usd is not None else 0.0
try:
patch_auto_top_up(enabled=False, threshold=thr, top_up_amount=rel)
except BillingScopeRequired:
self._billing_handle_scope_required(state)
return
except BillingError as exc:
self._billing_render_charge_error(state, exc)
return
print(" ✅ Auto-reload turned off.")
def _billing_limit_screen(self, state):
"""Screen 5 — monthly spend limit (read-only; cap is portal-only)."""
from agent.billing_view import format_money
print()
_cprint(f" 💳 {_b('Monthly spend limit')}")
print(f" {'' * 41}")
cap = state.monthly_cap
if cap is None or cap.limit_usd is None:
_cprint(f" {_d('No monthly cap visible (managed on the portal).')}")
else:
spent = format_money(cap.spent_this_month_usd)
limit = format_money(cap.limit_usd)
ceiling = " (default ceiling)" if cap.is_default_ceiling else ""
print(f" {spent} of {limit} used this month{ceiling}")
_limit_note = (
"The monthly limit is set on the portal — the terminal shows "
"it read-only."
)
_cprint(f" {_d(_limit_note)}")
self._billing_portal_hint(state)
def _show_insights(self, command: str = "/insights"):
"""Show usage insights and analytics from session history."""
# Parse optional --days flag

View file

@ -71,6 +71,7 @@ DEFAULT_NOUS_PORTAL_URL = "https://portal.nousresearch.com"
DEFAULT_NOUS_INFERENCE_URL = "https://inference-api.nousresearch.com/v1"
DEFAULT_NOUS_CLIENT_ID = "hermes-cli"
NOUS_INFERENCE_INVOKE_SCOPE = "inference:invoke"
NOUS_BILLING_MANAGE_SCOPE = "billing:manage"
DEFAULT_NOUS_SCOPE = NOUS_INFERENCE_INVOKE_SCOPE
NOUS_DEVICE_CODE_SOURCE = "device_code"
NOUS_AUTH_PATH_INVOKE_JWT = "invoke_jwt"
@ -7865,6 +7866,7 @@ def _nous_device_code_login(
timeout_seconds: float = 15.0,
insecure: bool = False,
ca_bundle: Optional[str] = None,
on_verification: Optional[Callable[[str, str], None]] = None,
) -> Dict[str, Any]:
"""Run the Nous device-code flow and return full OAuth state without persisting."""
pconfig = PROVIDER_REGISTRY["nous"]
@ -7919,6 +7921,16 @@ def _nous_device_code_login(
else:
print(" Could not open browser automatically — use the URL above.")
# Surface the verification URL/code to an out-of-band consumer (e.g. the
# TUI gateway, whose stdout is a JSON-RPC pipe — a plain print() there is
# dropped). Fired AFTER the print/browser block and BEFORE polling blocks,
# so the consumer can render the link while we wait. Best-effort.
if on_verification is not None:
try:
on_verification(verification_url, user_code)
except Exception:
pass
effective_interval = max(1, min(interval, DEVICE_AUTH_POLL_INTERVAL_CAP_SECONDS))
print(f"Waiting for approval (polling every {effective_interval}s)...")
@ -7984,6 +7996,91 @@ def _nous_device_code_login(
raise
def nous_token_has_billing_scope() -> bool:
"""Return True if the currently-held Nous token carries ``billing:manage``.
Reads the persisted ``scope`` string saved at login (``_save_provider_state``
stores ``token_data.get("scope") or scope``). A space-delimited match. Used by
the lazy step-up: if False, the first billing call will 403 ``insufficient_scope``
anyway, but checking up front lets a surface skip a doomed round-trip.
"""
try:
state = get_provider_auth_state("nous") or {}
except Exception:
return False
scope = state.get("scope")
if not isinstance(scope, str):
return False
return NOUS_BILLING_MANAGE_SCOPE in scope.split()
def step_up_nous_billing_scope(
*,
open_browser: bool = True,
timeout_seconds: float = 15.0,
on_verification: Optional[Callable[[str, str], None]] = None,
) -> bool:
"""Re-run the device flow requesting ``billing:manage`` and persist the result.
The lazy step-up (plan D-A): triggered when a billing endpoint returns
``403 insufficient_scope``. Runs a fresh device-connect with
``inference:invoke tool:invoke billing:manage`` on the scope. The user must be
an ADMIN/OWNER and tick "Allow terminal billing" in the portal for the minted
token to actually carry the scope; otherwise the server silently downscopes and this
returns False.
Reuses the held credential's portal/inference URLs + client_id so the step-up
targets the same deployment (incl. a preview via ``HERMES_PORTAL_BASE_URL`` set
at the original login). Persists to the auth store + shared store + pool, exactly
like ``_login_nous`` but WITHOUT the model picker (this is a scope upgrade, not
a fresh login).
Returns True iff the new token carries ``billing:manage``.
"""
prior = get_provider_auth_state("nous") or {}
pconfig = PROVIDER_REGISTRY["nous"]
# Build the step-up scope: existing scopes (if any) + billing:manage, deduped,
# order-stable. Fall back to the standard inference+tool+billing set.
_raw_scope = prior.get("scope")
prior_scope = _raw_scope if isinstance(_raw_scope, str) else ""
requested: list[str] = []
for tok in (prior_scope.split() or [NOUS_INFERENCE_INVOKE_SCOPE, "tool:invoke"]):
if tok and tok not in requested:
requested.append(tok)
if NOUS_BILLING_MANAGE_SCOPE not in requested:
requested.append(NOUS_BILLING_MANAGE_SCOPE)
scope = " ".join(requested)
auth_state = _nous_device_code_login(
portal_base_url=prior.get("portal_base_url") or None,
inference_base_url=prior.get("inference_base_url") or None,
client_id=prior.get("client_id") or pconfig.client_id,
scope=scope,
open_browser=open_browser,
timeout_seconds=timeout_seconds,
on_verification=on_verification,
)
with _auth_store_lock():
auth_store = _load_auth_store()
_save_provider_state(auth_store, "nous", auth_state)
_save_auth_store(auth_store)
# Mirror to shared store + reseed the pool (best-effort), same as _login_nous.
try:
_write_shared_nous_state(auth_state)
except Exception:
pass
try:
_sync_nous_pool_from_auth_store()
except Exception:
pass
granted = auth_state.get("scope")
return isinstance(granted, str) and NOUS_BILLING_MANAGE_SCOPE in granted.split()
def _login_nous(args, pconfig: ProviderConfig) -> None:
"""Nous Portal device authorization flow."""
timeout_seconds = getattr(args, "timeout", None) or 15.0

View file

@ -215,6 +215,7 @@ COMMAND_REGISTRY: list[CommandDef] = [
gateway_only=True),
CommandDef("usage", "Show token usage and rate limits for the current session", "Info"),
CommandDef("credits", "Show Nous credit balance and top up", "Info"),
CommandDef("billing", "Manage Nous terminal billing — buy credits, auto-reload, limits", "Info"),
CommandDef("insights", "Show usage insights and analytics", "Info",
args_hint="[days]"),
CommandDef("platforms", "Show gateway/messaging platform status", "Info",
@ -1053,8 +1054,9 @@ _SLACK_PRIORITY_ALIASES = ("btw", "bg")
# the telegram-parity test reads it so an entry here is a deliberate
# "Slack-via-/hermes" decision, not a silent clamp.
# - credits: the billing/top-up surface; reached via /hermes credits on Slack.
# - billing: the terminal-billing surface (buy/auto-reload/limit); /hermes billing.
# - debug: the log/report upload surface; reached via /hermes debug on Slack.
_SLACK_VIA_HERMES_ONLY = frozenset({"credits", "debug"})
_SLACK_VIA_HERMES_ONLY = frozenset({"credits", "billing", "debug"})
def _sanitize_slack_name(raw: str) -> str:

406
hermes_cli/nous_billing.py Normal file
View file

@ -0,0 +1,406 @@
"""Nous Portal terminal-billing HTTP client (Phase 2b).
Thin, fail-loud client for the four ``/api/billing/*`` endpoints the terminal
billing screens drive. Companion to ``hermes_cli/nous_account.py`` (which owns
read-only entitlement/balance) this module owns the *write* side: buy credits,
poll a charge, configure auto-reload.
Design rules:
- **Money is decimal, never float.** The server emits decimal STRINGS
(``"142.5"`` not fixed 2dp). We parse with :class:`decimal.Decimal` and never
round-trip through float.
- **This client raises typed exceptions; it does NOT fail open.** Fail-open is the
*caller's* job (the ``agent/billing_view.py`` builders) so each surface can
decide how to degrade. A raw network/HTTP error here surfaces as
:class:`BillingError` (or a subclass) carrying the parsed server ``error`` code,
HTTP status, ``portalUrl`` deep-link, and ``retry_after``.
- **Auth** = the OAuth bearer JWT Hermes already holds for inference
(``get_provider_auth_state("nous")["access_token"]``). No API-key auth on these.
- **Portal base URL** resolves with the same precedence as the device-flow login
(``auth.py``): ``HERMES_PORTAL_BASE_URL`` ``NOUS_PORTAL_BASE_URL`` the
stored auth-state ``portal_base_url`` the registry default. This is how the
E2E run points the client at a preview deployment with zero code change.
"""
from __future__ import annotations
import json
import os
import urllib.error
import urllib.parse
import urllib.request
from typing import Any, Optional
DEFAULT_PORTAL_BASE_URL = "https://portal.nousresearch.com"
# Default HTTP timeout (seconds). Charge/poll calls are quick; keep this tight so
# a hung portal doesn't freeze the TUI.
DEFAULT_TIMEOUT = 15.0
# Scope the privileged billing endpoints require. Mirrored from
# hermes_cli.auth.NOUS_BILLING_MANAGE_SCOPE (kept here too so this module has no
# import-time dependency on the much heavier auth module).
BILLING_MANAGE_SCOPE = "billing:manage"
# =============================================================================
# Typed errors
# =============================================================================
class BillingError(Exception):
"""A billing HTTP call failed.
Carries everything a surface needs to render the right message + affordance:
the server ``error`` code, HTTP ``status``, an optional human ``message``, the
``portalUrl`` deep-link (present on every gate denial), and ``retry_after``
seconds (429/503). ``payload`` is the full parsed JSON body when available.
"""
def __init__(
self,
message: str,
*,
status: Optional[int] = None,
error: Optional[str] = None,
portal_url: Optional[str] = None,
retry_after: Optional[int] = None,
payload: Optional[dict[str, Any]] = None,
) -> None:
super().__init__(message)
self.status = status
self.error = error
self.portal_url = portal_url
self.retry_after = retry_after
self.payload = payload or {}
class BillingScopeRequired(BillingError):
"""``403 insufficient_scope`` — the held token lacks ``billing:manage``.
The lazy step-up trigger: catching this kicks off a fresh device-connect that
requests ``billing:manage`` (and tells the user an ADMIN must tick "Allow
terminal billing"). Also fires mid-session if the scope is stripped on refresh
after the user loses ADMIN.
"""
class BillingRateLimited(BillingError):
"""``429 rate_limited`` or ``503 temporarily_unavailable``.
NOT a payment failure. Carries ``retry_after`` (seconds) back off and tell
the user "try again in N min"; never auto-retry-spam (the limiter is
5/org/hr + 5/token/hr and easy to dig deeper into).
"""
class BillingAuthError(BillingError):
"""``401`` — missing/invalid bearer token (not logged in / expired)."""
# =============================================================================
# Base-URL + auth resolution
# =============================================================================
def resolve_portal_base_url(state: Optional[dict[str, Any]] = None) -> str:
"""Resolve the portal base URL with login-time precedence.
``HERMES_PORTAL_BASE_URL`` ``NOUS_PORTAL_BASE_URL`` stored auth-state
``portal_base_url`` registry default. Trailing slash stripped.
"""
env = os.getenv("HERMES_PORTAL_BASE_URL") or os.getenv("NOUS_PORTAL_BASE_URL")
if env and env.strip():
return env.strip().rstrip("/")
if state:
stored = state.get("portal_base_url")
if isinstance(stored, str) and stored.strip():
return stored.strip().rstrip("/")
return DEFAULT_PORTAL_BASE_URL
def _absolutize_portal_url(portal_url: Optional[str]) -> Optional[str]:
"""Resolve a (possibly relative) server portalUrl to an absolute URL.
The server emits ``portalUrl`` relative by design (e.g. ``/billing?topup=open``)
it doesn't know which deployment the client points at. Resolve it against the
client's portal base (preview / staging / prod) so deep-links are clickable.
Idempotent: an already-absolute URL is returned unchanged (urljoin keeps it).
"""
if not (isinstance(portal_url, str) and portal_url.strip()):
return portal_url
base = resolve_portal_base_url()
# urljoin needs a trailing slash on the base to treat it as a directory and
# join an absolute path like "/billing?..." against the host. An already-
# absolute portal_url (with its own scheme/host) is returned as-is.
return urllib.parse.urljoin(base.rstrip("/") + "/", portal_url)
# Short-lived cache for the resolved (token, base). `resolve_nous_access_token`
# acquires two cross-process file locks + reads two files on every call (even on
# its fast path), which is wasteful when the 2s/5-min charge poll loop calls a
# billing endpoint ~150x per purchase. Cache the result briefly: the resolver
# only ever returns a token with >=120s of life (its refresh skew), so a 30s
# cache can never hand back an about-to-expire token. A 401 still surfaces
# normally (the cache holds a valid token, not the HTTP outcome).
_TOKEN_CACHE_TTL_SECONDS = 30.0
_token_cache: tuple[float, str, str] | None = None # (cached_at, token, base)
def _billing_not_logged_in(exc: Optional[BaseException] = None) -> "BillingAuthError":
"""Build the canonical 'not logged in' BillingAuthError (single source)."""
err = BillingAuthError(
"Not logged into Nous Portal — run `hermes portal` to log in.",
status=401,
error="invalid_token",
)
if exc is not None:
err.__cause__ = exc
return err
def _resolve_token_and_base(*, use_cache: bool = True) -> tuple[str, str]:
"""Return ``(access_token, portal_base_url)`` for billing calls.
Uses the same refresh-aware resolver the inference path uses
(``resolve_nous_access_token``), so a short-lived (~15 min) access token that
has expired is transparently refreshed via the stored ``refresh_token``
instead of failing as "not logged in". Raises :class:`BillingAuthError` only
when there is no usable Nous session at all.
The result is cached for ``_TOKEN_CACHE_TTL_SECONDS`` to keep the charge poll
loop from re-locking + re-reading the auth store on every 2s tick. Pass
``use_cache=False`` to force a fresh resolution (e.g. after a 401).
"""
global _token_cache
import time as _time
if use_cache and _token_cache is not None:
cached_at, token, base = _token_cache
if (_time.time() - cached_at) < _TOKEN_CACHE_TTL_SECONDS:
return token, base
try:
from hermes_cli.auth import get_provider_auth_state
state = get_provider_auth_state("nous") or {}
except Exception:
state = {}
base = resolve_portal_base_url(state)
try:
from hermes_cli.auth import AuthError, resolve_nous_access_token
except ImportError:
# auth module unavailable — fall back to the raw stored token.
token = state.get("access_token")
if isinstance(token, str) and token.strip():
resolved = (token.strip(), base)
_token_cache = (_time.time(), *resolved)
return resolved
raise _billing_not_logged_in()
try:
token = resolve_nous_access_token()
except AuthError as exc:
raise _billing_not_logged_in(exc) from exc
resolved = (token.strip(), base)
_token_cache = (_time.time(), *resolved)
return resolved
# =============================================================================
# HTTP plumbing
# =============================================================================
def _retry_after_seconds(headers: Any) -> Optional[int]:
"""Parse a ``Retry-After`` header (integer seconds) — None if absent/bad."""
if headers is None:
return None
try:
raw = headers.get("Retry-After")
except Exception:
raw = None
if raw is None:
return None
try:
return int(str(raw).strip())
except (TypeError, ValueError):
return None
def _raise_for_error(
status: int, payload: dict[str, Any], headers: Any = None
) -> None:
"""Map an HTTP error response to the right typed :class:`BillingError`."""
error = payload.get("error") if isinstance(payload, dict) else None
message = payload.get("message") if isinstance(payload, dict) else None
portal_url = _absolutize_portal_url(
payload.get("portalUrl") if isinstance(payload, dict) else None
)
retry_after = _retry_after_seconds(headers)
common = {
"status": status,
"error": error,
"portal_url": portal_url,
"retry_after": retry_after,
"payload": payload if isinstance(payload, dict) else None,
}
if status == 401:
raise BillingAuthError(message or "Authentication required.", **common)
if status == 403 and error == "insufficient_scope":
raise BillingScopeRequired(
message or "This action needs the billing:manage scope.", **common
)
if status in (429, 503):
raise BillingRateLimited(
message or "Rate limited — try again shortly.", **common
)
raise BillingError(message or error or f"Billing request failed ({status}).", **common)
def _request(
method: str,
path: str,
*,
body: Optional[dict[str, Any]] = None,
extra_headers: Optional[dict[str, str]] = None,
timeout: float = DEFAULT_TIMEOUT,
_retried_auth: bool = False,
) -> dict[str, Any]:
"""Make an authenticated billing request; return the parsed JSON dict.
Raises a typed :class:`BillingError` on any non-2xx response (or transport
failure). 2xx with an empty body returns ``{}``. A 401 triggers exactly one
retry with a freshly-resolved token (bypassing the short token cache) so a
cached-but-just-expired token self-heals instead of failing the call.
"""
token, base = _resolve_token_and_base(use_cache=not _retried_auth)
url = f"{base}{path}"
headers = {
"Authorization": f"Bearer {token}",
"Accept": "application/json",
}
if body is not None:
headers["Content-Type"] = "application/json"
if extra_headers:
headers.update(extra_headers)
data = json.dumps(body).encode("utf-8") if body is not None else None
req = urllib.request.Request(url, data=data, headers=headers, method=method)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read().decode("utf-8")
return json.loads(raw) if raw.strip() else {}
except urllib.error.HTTPError as exc:
# A 401 on a cached token → drop the cache and retry once with a fresh
# (refresh-aware) resolve before surfacing the auth error.
if exc.code == 401 and not _retried_auth:
global _token_cache
_token_cache = None
return _request(
method,
path,
body=body,
extra_headers=extra_headers,
timeout=timeout,
_retried_auth=True,
)
raw = ""
try:
raw = exc.read().decode("utf-8")
except Exception:
raw = ""
try:
payload = json.loads(raw) if raw.strip() else {}
except json.JSONDecodeError:
payload = {}
_raise_for_error(exc.code, payload, getattr(exc, "headers", None))
raise # unreachable; _raise_for_error always raises
except urllib.error.URLError as exc:
raise BillingError(
f"Could not reach Nous Portal: {exc.reason}", error="network_error"
) from exc
# =============================================================================
# The four endpoints
# =============================================================================
def get_billing_state(*, timeout: float = DEFAULT_TIMEOUT) -> dict[str, Any]:
"""``GET /api/billing/state`` — role-tiered overview (no scope required)."""
return _request("GET", "/api/billing/state", timeout=timeout)
def patch_auto_top_up(
*,
enabled: bool,
threshold: float | str,
top_up_amount: float | str,
timeout: float = DEFAULT_TIMEOUT,
) -> dict[str, Any]:
"""``PATCH /api/billing/auto-top-up`` — configure auto-reload (scope required).
Body is strict server-side: extra keys (``maxMonthlySpend``, a payment method)
are rejected with 400. Numbers are sent as JSON numbers per the contract.
"""
return _request(
"PATCH",
"/api/billing/auto-top-up",
body={
"enabled": bool(enabled),
"threshold": float(threshold),
"topUpAmount": float(top_up_amount),
},
timeout=timeout,
)
def post_charge(
*,
amount_usd: float | str,
idempotency_key: str,
timeout: float = DEFAULT_TIMEOUT,
) -> dict[str, Any]:
"""``POST /api/billing/charge`` — buy credits (scope required).
``Idempotency-Key`` header is MANDATORY (a missing header is a server 400, not
a default): generate a UUID per user-confirmed purchase and reuse it on retry.
Returns ``202 {chargeId}`` money is NOT confirmed yet; poll with
:func:`get_charge_status`.
"""
if not (isinstance(idempotency_key, str) and idempotency_key.strip()):
raise BillingError(
"Idempotency-Key is required for a charge.",
error="idempotency_key_required",
)
return _request(
"POST",
"/api/billing/charge",
body={"amountUsd": float(amount_usd)},
extra_headers={"Idempotency-Key": idempotency_key.strip()},
timeout=timeout,
)
def get_charge_status(
charge_id: str, *, timeout: float = DEFAULT_TIMEOUT
) -> dict[str, Any]:
"""``GET /api/billing/charge/{id}`` — poll a charge (scope required).
Returns ``{status: "pending"|"settled"|"failed", ...}``. An unknown or foreign
id returns ``{status:"pending"}`` (never 404, never another org's data) — so a
``pending`` that never resolves past the 5-min cap is a *timeout*, not an error.
"""
if not (isinstance(charge_id, str) and charge_id.strip()):
raise BillingError("A charge id is required.", error="invalid_charge_id")
# urllib does not need manual quoting for the opaque ids the server mints, but
# guard against a stray slash that would change the path shape.
safe_id = urllib.parse.quote(charge_id.strip(), safe="")
return _request("GET", f"/api/billing/charge/{safe_id}", timeout=timeout)

View file

@ -0,0 +1,377 @@
"""Unit tests for the Phase 2b terminal-billing core + HTTP client.
Covers:
- Decimal money parsing/formatting (server emits decimal strings, not 2dp).
- BillingState payload parsing (role tiering, presets, bounds, sub-structs).
- Error-code typed-exception mapping (the live-verified contract matrix).
- Fail-open builder behavior.
- Idempotency key generation.
- Custom-amount validation against bounds + multipleOf 0.01.
No network: HTTP-layer tests drive _raise_for_error directly and monkeypatch the
request function for the builder.
"""
from __future__ import annotations
from decimal import Decimal
import pytest
import agent.billing_view as bv
from agent.billing_view import (
AutoReload,
BillingState,
CardInfo,
MonthlyCap,
billing_state_from_payload,
build_billing_state,
format_money,
new_idempotency_key,
parse_money,
validate_charge_amount,
)
import hermes_cli.nous_billing as nb
from hermes_cli.nous_billing import (
BillingAuthError,
BillingError,
BillingRateLimited,
BillingScopeRequired,
_raise_for_error,
resolve_portal_base_url,
)
# ---------------------------------------------------------------------------
# Decimal money
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"raw,expected",
[
("142.5", Decimal("142.5")), # decimal string, NOT 2dp — the headline case
("100", Decimal("100")),
("10000", Decimal("10000")),
("0.01", Decimal("0.01")),
(250, Decimal("250")),
(" 50 ", Decimal("50")),
],
)
def test_parse_money_valid(raw, expected):
assert parse_money(raw) == expected
@pytest.mark.parametrize("raw", [None, "", "abc", "1.2.3", "$5", {}])
def test_parse_money_invalid_returns_none(raw):
assert parse_money(raw) is None
def test_parse_money_never_uses_binary_float():
# If a float ever sneaks through, we still get an exact decimal, not 0.1+0.2 junk.
assert parse_money(0.1) == Decimal("0.1")
@pytest.mark.parametrize(
"value,expected",
[
(Decimal("142.5"), "$142.50"),
(Decimal("100"), "$100"),
(Decimal("0.01"), "$0.01"),
(Decimal("1000"), "$1000"),
(None, ""),
],
)
def test_format_money(value, expected):
assert format_money(value) == expected
# ---------------------------------------------------------------------------
# BillingState payload parsing
# ---------------------------------------------------------------------------
def _member_payload() -> dict:
return {
"org": {"id": "o1", "slug": "acme", "name": "Acme", "role": "MEMBER"},
"balanceUsd": "142.5",
"cliBillingEnabled": True,
"chargePresets": ["100", "250", "500"],
"bounds": {"minUsd": "10", "maxUsd": "10000"},
"card": None,
"monthlyCap": None,
"autoReload": None,
}
def _owner_payload() -> dict:
p = _member_payload()
p["org"]["role"] = "OWNER"
p["card"] = {"brand": "visa", "last4": "4242"}
p["monthlyCap"] = {
"limitUsd": "1000",
"spentThisMonthUsd": "180",
"isDefaultCeiling": True,
}
p["autoReload"] = {"enabled": True, "thresholdUsd": "20", "reloadToUsd": "100"}
return p
def test_state_member_tier_parse():
s = billing_state_from_payload(_member_payload())
assert s.logged_in
assert s.role == "MEMBER"
assert s.balance_usd == Decimal("142.5")
assert s.cli_billing_enabled is True
assert s.charge_presets == (Decimal("100"), Decimal("250"), Decimal("500"))
assert s.min_usd == Decimal("10") and s.max_usd == Decimal("10000")
assert s.card is None and s.monthly_cap is None and s.auto_reload is None
assert s.is_admin is False
assert s.can_charge is False # not admin
def test_state_owner_tier_parse():
s = billing_state_from_payload(_owner_payload())
assert s.is_admin is True
assert s.can_charge is True # admin + kill-switch on
assert s.card == CardInfo(brand="visa", last4="4242")
assert s.card is not None and s.card.masked == "visa ····4242"
assert s.monthly_cap == MonthlyCap(
limit_usd=Decimal("1000"),
spent_this_month_usd=Decimal("180"),
is_default_ceiling=True,
)
assert s.auto_reload == AutoReload(
enabled=True, threshold_usd=Decimal("20"), reload_to_usd=Decimal("100")
)
def test_state_can_charge_false_when_killswitch_off():
p = _owner_payload()
p["cliBillingEnabled"] = False
s = billing_state_from_payload(p)
assert s.is_admin is True
assert s.can_charge is False # kill-switch off gates the action
def test_state_handles_garbage_substructs():
p = _member_payload()
p["card"] = "not-a-dict"
p["monthlyCap"] = 42
p["chargePresets"] = ["100", "bad", "250"] # bad preset dropped, not crash
s = billing_state_from_payload(p)
assert s.card is None and s.monthly_cap is None
assert s.charge_presets == (Decimal("100"), Decimal("250"))
# ---------------------------------------------------------------------------
# Error-code → typed-exception mapping (live-verified contract)
# ---------------------------------------------------------------------------
class _Headers:
def __init__(self, d):
self._d = d
def get(self, k):
return self._d.get(k)
def test_401_maps_to_auth_error():
with pytest.raises(BillingAuthError) as ei:
_raise_for_error(401, {"error": "invalid_token"})
assert ei.value.status == 401
def test_403_insufficient_scope_maps_to_scope_required():
with pytest.raises(BillingScopeRequired) as ei:
_raise_for_error(403, {"error": "insufficient_scope", "portalUrl": "/billing"})
assert ei.value.error == "insufficient_scope"
# portalUrl is resolved to an absolute URL (relative-by-design from the server).
assert (ei.value.portal_url or "").startswith("http")
assert (ei.value.portal_url or "").endswith("/billing")
@pytest.mark.parametrize("status", [429, 503])
def test_rate_limited_maps_with_retry_after(status):
with pytest.raises(BillingRateLimited) as ei:
_raise_for_error(
status,
{"error": "rate_limited"},
_Headers({"Retry-After": "60"}),
)
assert ei.value.retry_after == 60
# Critically: a rate limit is NOT a generic BillingError-only — surfaces branch on type.
assert isinstance(ei.value, BillingRateLimited)
@pytest.mark.parametrize(
"error",
[
"no_payment_method",
"cli_billing_disabled",
"role_required",
"monthly_cap_exceeded",
"org_access_denied",
],
)
def test_other_403s_map_to_base_error_with_portal_url(error):
with pytest.raises(BillingError) as ei:
_raise_for_error(403, {"error": error, "portalUrl": "/billing?topup=open"})
# Not a scope/auth/rate subclass — the generic gate-denial path.
assert not isinstance(ei.value, (BillingScopeRequired, BillingAuthError, BillingRateLimited))
assert ei.value.error == error
# portalUrl resolved to an absolute deep-link (server sends it relative).
assert (ei.value.portal_url or "").startswith("http")
assert (ei.value.portal_url or "").endswith("/billing?topup=open")
def test_monthly_cap_exceeded_carries_remaining_in_payload():
with pytest.raises(BillingError) as ei:
_raise_for_error(
403,
{
"error": "monthly_cap_exceeded",
"remainingUsd": "12.50",
"isDefaultCeiling": True,
"portalUrl": "/billing",
},
)
assert ei.value.payload["remainingUsd"] == "12.50"
assert ei.value.payload["isDefaultCeiling"] is True
def test_400_amount_out_of_bounds_is_base_error():
with pytest.raises(BillingError) as ei:
_raise_for_error(400, {"error": "amount_out_of_bounds", "message": "too big"})
assert ei.value.status == 400
assert "too big" in str(ei.value)
# ---------------------------------------------------------------------------
# post_charge requires idempotency key (client-side guard)
# ---------------------------------------------------------------------------
def test_post_charge_requires_idempotency_key():
with pytest.raises(BillingError) as ei:
nb.post_charge(amount_usd=50, idempotency_key="")
assert ei.value.error == "idempotency_key_required"
def test_get_charge_status_requires_id():
with pytest.raises(BillingError) as ei:
nb.get_charge_status("")
assert ei.value.error == "invalid_charge_id"
# ---------------------------------------------------------------------------
# Base-URL resolution precedence
# ---------------------------------------------------------------------------
def test_portal_base_url_env_override(monkeypatch):
monkeypatch.setenv("HERMES_PORTAL_BASE_URL", "https://preview.example.com/")
assert resolve_portal_base_url() == "https://preview.example.com"
def test_portal_base_url_falls_back_to_state(monkeypatch):
monkeypatch.delenv("HERMES_PORTAL_BASE_URL", raising=False)
monkeypatch.delenv("NOUS_PORTAL_BASE_URL", raising=False)
assert (
resolve_portal_base_url({"portal_base_url": "https://stored.example.com/"})
== "https://stored.example.com"
)
def test_portal_base_url_default(monkeypatch):
monkeypatch.delenv("HERMES_PORTAL_BASE_URL", raising=False)
monkeypatch.delenv("NOUS_PORTAL_BASE_URL", raising=False)
assert resolve_portal_base_url() == nb.DEFAULT_PORTAL_BASE_URL
# ---------------------------------------------------------------------------
# Fail-open builder
# ---------------------------------------------------------------------------
def test_build_billing_state_logged_out_on_auth_error(monkeypatch):
def _auth(*a, **kw):
raise BillingAuthError("nope", status=401)
monkeypatch.setattr(nb, "get_billing_state", _auth)
s = build_billing_state()
assert s.logged_in is False
assert s.error is None # cleanly logged out, not an error
def test_build_billing_state_fail_open_on_http_error(monkeypatch):
def _boom(*a, **kw):
raise BillingError("portal exploded", status=500)
monkeypatch.setattr(nb, "get_billing_state", _boom)
s = build_billing_state()
assert s.logged_in is False
assert "portal exploded" in (s.error or "")
def test_build_billing_state_parses_and_prefers_server_portal_url(monkeypatch):
payload = _owner_payload()
payload["portalUrl"] = "https://portal.example.com/billing?topup=open"
monkeypatch.setattr(nb, "get_billing_state", lambda *a, **kw: payload)
s = build_billing_state()
assert s.logged_in is True
assert s.portal_url == "https://portal.example.com/billing?topup=open"
assert s.balance_usd == Decimal("142.5")
def test_build_billing_state_builds_fallback_portal_url(monkeypatch):
payload = _member_payload() # no portalUrl key
monkeypatch.setattr(nb, "get_billing_state", lambda *a, **kw: payload)
monkeypatch.setattr(bv, "_fallback_portal_url", lambda base: "FALLBACK")
# resolve_portal_base_url is imported into bv via local import; patch nb's.
s = build_billing_state()
assert s.portal_url == "FALLBACK"
# ---------------------------------------------------------------------------
# Idempotency
# ---------------------------------------------------------------------------
def test_new_idempotency_key_unique_and_uuid_shaped():
a, b = new_idempotency_key(), new_idempotency_key()
assert a != b
assert len(a) == 36 and a.count("-") == 4
# ---------------------------------------------------------------------------
# Amount validation (Screen 3 custom input)
# ---------------------------------------------------------------------------
def test_validate_amount_ok():
v = validate_charge_amount("100", min_usd=Decimal("10"), max_usd=Decimal("10000"))
assert v.ok and v.amount == Decimal("100")
def test_validate_amount_strips_dollar_sign():
v = validate_charge_amount("$250", min_usd=Decimal("10"), max_usd=Decimal("10000"))
assert v.ok and v.amount == Decimal("250")
@pytest.mark.parametrize(
"raw,err_substr",
[
("", "dollar amount"),
("0", "greater than"),
("-5", "greater than"),
("10.005", "cent"), # multipleOf 0.01 — sub-cent rejected
("5", "Minimum"), # below bounds.minUsd
("99999", "Maximum"), # above bounds.maxUsd
],
)
def test_validate_amount_rejections(raw, err_substr):
v = validate_charge_amount(raw, min_usd=Decimal("10"), max_usd=Decimal("10000"))
assert not v.ok
assert err_substr.lower() in (v.error or "").lower()

View file

@ -34,37 +34,35 @@ class TestPromptTextInputThreadSafety:
# not the orphaned-coroutine result.
assert mock_rit.called
def test_background_thread_falls_back_to_direct_input(self):
"""On a daemon thread, skip run_in_terminal and call input() directly.
def test_background_thread_cancels_instead_of_hanging(self):
"""On a daemon thread with an active app, cancel cleanly (return None).
This preserves the fallback for any prompt that still runs off the main
UI thread: run_in_terminal's coroutine would otherwise be orphaned.
stdin is owned by the prompt_toolkit event loop / JSON-RPC pipe on the
non-main (process_loop / slash-worker) thread, so a bare input() there
would block until the worker's timeout (#23185 / billing auto-reload
hang). The guard cancels to None instead of hanging it must NOT call
run_in_terminal (orphaned coroutine) and must NOT call input().
"""
cli = _make_cli()
captured = {}
def fake_input(prompt):
captured["prompt"] = prompt
return "1"
result_holder = {}
def run_on_daemon():
with patch("prompt_toolkit.application.run_in_terminal") as mock_rit, \
patch("builtins.input", side_effect=fake_input):
patch("builtins.input", side_effect=AssertionError("input() must not be called off-main-thread")) as mock_input:
result_holder["value"] = cli._prompt_text_input("Choice [1/2/3]: ")
result_holder["rit_called"] = mock_rit.called
result_holder["input_called"] = mock_input.called
t = threading.Thread(target=run_on_daemon, daemon=True)
t.start()
t.join(timeout=2.0)
assert not t.is_alive(), "daemon thread hung — input() was not driven"
assert not t.is_alive(), "daemon thread hung — guard did not cancel cleanly"
# run_in_terminal was bypassed entirely on the background thread.
# Cancelled cleanly: None returned, neither run_in_terminal nor input() called.
assert result_holder["value"] is None
assert result_holder["rit_called"] is False
# input() was invoked with the prompt and its return value was captured.
assert captured.get("prompt") == "Choice [1/2/3]: "
assert result_holder["value"] == "1"
assert result_holder["input_called"] is False
def test_no_app_uses_direct_input(self):
"""Without an active prompt_toolkit app, always call input() directly."""

View file

@ -0,0 +1,136 @@
"""Tests for the /billing CLI handler (cli.py::_show_billing).
Focus on the non-interactive (no live prompt_toolkit app) path the same
discipline as the /credits non-interactive test: it must render text, never
invoke the modal (which would read the slash-worker's JSON-RPC stdin and hang).
Plus role/kill-switch gating and logged-out handling.
"""
from __future__ import annotations
from decimal import Decimal
import pytest
import agent.billing_view as bv
from agent.billing_view import BillingState, CardInfo, MonthlyCap
from cli import HermesCLI
@pytest.fixture
def cli():
obj = HermesCLI.__new__(HermesCLI) # bypass __init__ (no full app needed)
obj._app = None # non-interactive: forces the text path
return obj
def _boom_modal(*a, **kw):
raise AssertionError("modal must NOT be called in non-interactive mode")
def test_billing_logged_out(cli, monkeypatch, capsys):
monkeypatch.setattr(bv, "build_billing_state", lambda *a, **kw: BillingState(logged_in=False))
cli._show_billing("/billing")
out = capsys.readouterr().out
assert "Not logged into Nous Portal" in out
assert "hermes portal" in out
def test_billing_overview_non_interactive_renders_text_not_modal(cli, monkeypatch, capsys):
monkeypatch.setattr(HermesCLI, "_prompt_text_input_modal", _boom_modal, raising=False)
state = BillingState(
logged_in=True,
org_name="Acme",
role="OWNER",
balance_usd=Decimal("142.5"),
cli_billing_enabled=True,
charge_presets=(Decimal("100"),),
monthly_cap=MonthlyCap(limit_usd=Decimal("1000"), spent_this_month_usd=Decimal("180"),
is_default_ceiling=True),
portal_url="https://portal/billing?topup=open",
)
monkeypatch.setattr(bv, "build_billing_state", lambda *a, **kw: state)
cli._show_billing("/billing")
out = capsys.readouterr().out
assert "Usage credits" in out
assert "$142.50" in out
assert "$180 of $1000 used (default ceiling)" in out
# New design: a spend bar with a percentage on the overview.
assert "%" in out and ("" in out or "" in out)
# ZERO sub-commands: no /billing buy|auto-reload|limit advertising.
assert "/billing buy" not in out
assert "Actions:" not in out
# Non-interactive funnels to the portal (the URL is the affordance).
assert "Manage on portal:" in out
def test_billing_member_cannot_charge(cli, monkeypatch, capsys):
state = BillingState(
logged_in=True, role="MEMBER", balance_usd=Decimal("10"),
cli_billing_enabled=True, portal_url="https://portal/billing",
)
monkeypatch.setattr(bv, "build_billing_state", lambda *a, **kw: state)
cli._show_billing("/billing")
out = capsys.readouterr().out
assert "require an org admin/owner" in out
def test_billing_killswitch_off_blocks(cli, monkeypatch, capsys):
state = BillingState(
logged_in=True, role="OWNER", balance_usd=Decimal("10"),
cli_billing_enabled=False, portal_url="https://portal/billing",
)
monkeypatch.setattr(bv, "build_billing_state", lambda *a, **kw: state)
cli._show_billing("/billing")
out = capsys.readouterr().out
assert "turned off for this org" in out
def test_billing_limit_screen_readonly(cli, monkeypatch, capsys):
state = BillingState(
logged_in=True, role="OWNER", cli_billing_enabled=True,
monthly_cap=MonthlyCap(limit_usd=Decimal("1000"), spent_this_month_usd=Decimal("250"),
is_default_ceiling=True),
portal_url="https://portal/billing",
)
monkeypatch.setattr(bv, "build_billing_state", lambda *a, **kw: state)
# ZERO sub-commands: the limit screen is reached via the menu, never a
# sub-command — call it directly the way the overview menu would.
cli._billing_limit_screen(state)
out = capsys.readouterr().out
assert "Monthly spend limit" in out
assert "$250 of $1000 used" in out
assert "read-only" in out
def test_billing_sub_arg_ignored_opens_overview(cli, monkeypatch, capsys):
# A stray sub-arg must NOT error and must NOT dispatch to a sub-screen —
# it just opens the overview (spec §0.4: zero sub-commands).
monkeypatch.setattr(HermesCLI, "_prompt_text_input_modal", _boom_modal, raising=False)
state = BillingState(
logged_in=True, role="OWNER", balance_usd=Decimal("142.5"),
cli_billing_enabled=True, charge_presets=(Decimal("25"),),
portal_url="https://portal/billing",
)
monkeypatch.setattr(bv, "build_billing_state", lambda *a, **kw: state)
cli._show_billing("/billing buy") # arg is ignored
out = capsys.readouterr().out
assert "Usage credits" in out # overview, NOT the buy screen
assert "Buy usage credits" not in out
def test_billing_buy_non_interactive_defers_to_portal(cli, monkeypatch, capsys):
monkeypatch.setattr(HermesCLI, "_prompt_text_input_modal", _boom_modal, raising=False)
state = BillingState(
logged_in=True, role="OWNER", cli_billing_enabled=True,
charge_presets=(Decimal("25"), Decimal("50"), Decimal("100")),
card=CardInfo(brand="visa", last4="4242"),
portal_url="https://portal/billing",
)
monkeypatch.setattr(bv, "build_billing_state", lambda *a, **kw: state)
# Reached via the menu in real use; non-interactively it defers to the portal.
cli._billing_buy_flow(state)
out = capsys.readouterr().out
assert "Buy usage credits" in out
assert "$25" in out and "$50" in out and "$100" in out
assert "interactive CLI" in out # defers; no charge attempted non-interactively

View file

@ -0,0 +1,53 @@
"""Portal-URL resolution for Phase 2b billing errors (nous_billing).
The server emits ``portalUrl`` relative by design (``/billing?topup=open``); the
client must resolve it against the active portal base so deep-links are clickable
on whatever deployment (preview / staging / prod) the user is pointed at.
"""
from __future__ import annotations
import pytest
from hermes_cli.nous_billing import (
BillingError,
_absolutize_portal_url,
_raise_for_error,
)
@pytest.fixture
def _preview(monkeypatch):
monkeypatch.setenv("HERMES_PORTAL_BASE_URL", "https://nas-pr-412.nousresearch.wtf")
def test_absolutize_resolves_relative(_preview):
assert (
_absolutize_portal_url("/billing?topup=open")
== "https://nas-pr-412.nousresearch.wtf/billing?topup=open"
)
def test_absolutize_leaves_absolute_unchanged(_preview):
# Idempotent: an already-absolute URL must NOT be double-prefixed.
url = "https://other.example/billing?topup=open"
assert _absolutize_portal_url(url) == url
def test_absolutize_passthrough_empty(_preview):
assert _absolutize_portal_url(None) is None
assert _absolutize_portal_url("") == ""
def test_raise_for_error_attaches_absolute_portal_url(_preview):
# The 403 no_payment_method envelope carries a RELATIVE portalUrl; the raised
# BillingError must expose it as ABSOLUTE so CLI + TUI render a clickable link.
with pytest.raises(BillingError) as exc_info:
_raise_for_error(
403,
{"error": "no_payment_method", "portalUrl": "/billing?topup=open"},
)
assert (
exc_info.value.portal_url
== "https://nas-pr-412.nousresearch.wtf/billing?topup=open"
)

View file

@ -0,0 +1,193 @@
"""Tests for the Phase 2b billing:manage scope step-up (auth.py)."""
from __future__ import annotations
import pytest
import hermes_cli.auth as auth
from hermes_cli.auth import (
NOUS_BILLING_MANAGE_SCOPE,
nous_token_has_billing_scope,
step_up_nous_billing_scope,
)
# ---------------------------------------------------------------------------
# nous_token_has_billing_scope
# ---------------------------------------------------------------------------
def test_has_scope_true_when_present(monkeypatch):
monkeypatch.setattr(
auth,
"get_provider_auth_state",
lambda p: {"scope": "inference:invoke tool:invoke billing:manage"},
)
assert nous_token_has_billing_scope() is True
def test_has_scope_false_when_absent(monkeypatch):
monkeypatch.setattr(
auth, "get_provider_auth_state", lambda p: {"scope": "inference:invoke tool:invoke"}
)
assert nous_token_has_billing_scope() is False
def test_has_scope_false_when_no_state(monkeypatch):
monkeypatch.setattr(auth, "get_provider_auth_state", lambda p: None)
assert nous_token_has_billing_scope() is False
def test_has_scope_no_substring_false_positive(monkeypatch):
# "billing:manage-lite" must NOT match billing:manage (split-based, not substring).
monkeypatch.setattr(
auth, "get_provider_auth_state", lambda p: {"scope": "billing:manage-lite"}
)
assert nous_token_has_billing_scope() is False
# ---------------------------------------------------------------------------
# step_up_nous_billing_scope
# ---------------------------------------------------------------------------
@pytest.fixture
def _stub_persist(monkeypatch):
"""Neutralize the persistence side-effects so step-up tests are pure."""
monkeypatch.setattr(auth, "_auth_store_lock", lambda: _NullCtx())
monkeypatch.setattr(auth, "_load_auth_store", lambda: {})
monkeypatch.setattr(auth, "_save_provider_state", lambda *a, **kw: None)
monkeypatch.setattr(auth, "_save_auth_store", lambda *a, **kw: "auth.json")
monkeypatch.setattr(auth, "_write_shared_nous_state", lambda *a, **kw: None)
monkeypatch.setattr(auth, "_sync_nous_pool_from_auth_store", lambda: None)
class _NullCtx:
def __enter__(self):
return self
def __exit__(self, *a):
return False
def test_step_up_requests_billing_scope_and_reuses_prior_urls(monkeypatch, _stub_persist):
monkeypatch.setattr(
auth,
"get_provider_auth_state",
lambda p: {
"scope": "inference:invoke tool:invoke",
"portal_base_url": "https://preview.example.com",
"inference_base_url": "https://inf.example.com",
"client_id": "hermes-cli",
},
)
captured = {}
def _fake_login(**kw):
captured.update(kw)
# Simulate the admin ticking the box → token comes back WITH the scope.
return {"scope": "inference:invoke tool:invoke billing:manage", "access_token": "t"}
monkeypatch.setattr(auth, "_nous_device_code_login", _fake_login)
granted = step_up_nous_billing_scope()
assert granted is True
# Requested scope must include billing:manage, preserving prior scopes.
assert NOUS_BILLING_MANAGE_SCOPE in captured["scope"].split()
assert "inference:invoke" in captured["scope"].split()
# Reuses the prior credential's deployment URLs (so a preview stays a preview).
assert captured["portal_base_url"] == "https://preview.example.com"
assert captured["client_id"] == "hermes-cli"
def test_step_up_returns_false_when_downscoped(monkeypatch, _stub_persist):
# Non-admin / unticked → the server silently downscopes; token comes back WITHOUT scope.
monkeypatch.setattr(auth, "get_provider_auth_state", lambda p: {"scope": "inference:invoke"})
monkeypatch.setattr(
auth,
"_nous_device_code_login",
lambda **kw: {"scope": "inference:invoke", "access_token": "t"},
)
assert step_up_nous_billing_scope() is False
def test_step_up_falls_back_to_standard_scope_when_no_prior(monkeypatch, _stub_persist):
monkeypatch.setattr(auth, "get_provider_auth_state", lambda p: {})
captured = {}
def _fake_login(**kw):
captured.update(kw)
return {"scope": "inference:invoke tool:invoke billing:manage"}
monkeypatch.setattr(auth, "_nous_device_code_login", _fake_login)
step_up_nous_billing_scope()
requested = captured["scope"].split()
assert "inference:invoke" in requested
assert "tool:invoke" in requested
assert NOUS_BILLING_MANAGE_SCOPE in requested
# ---------------------------------------------------------------------------
# on_verification callback plumbing (TUI surfaces the device-flow URL via this)
# ---------------------------------------------------------------------------
def test_step_up_forwards_on_verification_callback(monkeypatch, _stub_persist):
monkeypatch.setattr(auth, "get_provider_auth_state", lambda p: {})
captured = {}
def _fake_login(**kw):
captured.update(kw)
return {"scope": "inference:invoke tool:invoke billing:manage"}
monkeypatch.setattr(auth, "_nous_device_code_login", _fake_login)
def _cb(url, code):
pass
step_up_nous_billing_scope(on_verification=_cb)
# The callback must be threaded straight through to the device-code login.
assert captured["on_verification"] is _cb
def test_device_login_fires_on_verification_before_polling(monkeypatch):
"""on_verification(url, code) must fire BEFORE _poll_for_token (so the TUI
can render the link while the flow blocks waiting for approval)."""
order: list[str] = []
monkeypatch.setattr(
auth,
"_request_device_code",
lambda **kw: {
"verification_uri_complete": "https://portal.example/device?code=ABCD",
"user_code": "ABCD-1234",
"device_code": "dev",
"expires_in": 600,
"interval": 5,
},
)
def _fake_poll(**kw):
order.append("poll")
return {"access_token": "t", "scope": "inference:invoke", "expires_in": 3600}
monkeypatch.setattr(auth, "_poll_for_token", _fake_poll)
seen = {}
def _cb(url, code):
order.append("verify")
seen["url"] = url
seen["code"] = code
# We only assert the callback fires before polling. Post-poll token
# validation (JWT usability checks) is out of scope and may raise on the
# synthetic token — swallow it; the ordering assertion is what matters.
try:
auth._nous_device_code_login(open_browser=False, on_verification=_cb)
except Exception:
pass
assert order[:2] == ["verify", "poll"], "callback must fire before polling"
assert seen["url"] == "https://portal.example/device?code=ABCD"
assert seen["code"] == "ABCD-1234"

View file

@ -0,0 +1,206 @@
"""Tests for the Phase 2b billing JSON-RPC methods (tui_gateway/server.py).
Verifies the structured envelope contract the Ink side branches on:
- billing.state serializes BillingState (Decimals strings) + fails open.
- billing.charge / charge_status / auto_reload return typed error envelopes
(result.ok=false, result.error=<code>) instead of JSON-RPC errors.
- billing.charge mints + echoes an idempotency_key for retry reuse.
"""
from __future__ import annotations
from decimal import Decimal
import pytest
import tui_gateway.server as srv
import hermes_cli.nous_billing as nb
import agent.billing_view as bv
from agent.billing_view import BillingState, CardInfo, MonthlyCap
def _call(method: str, params: dict) -> dict:
"""Invoke a registered RPC method and return its result dict."""
envelope = srv._methods[method](1, params)
return envelope["result"]
# ---------------------------------------------------------------------------
# billing.state
# ---------------------------------------------------------------------------
def test_billing_state_serializes_decimals_as_strings(monkeypatch):
state = BillingState(
logged_in=True,
org_name="Acme",
role="OWNER",
balance_usd=Decimal("142.5"),
cli_billing_enabled=True,
charge_presets=(Decimal("100"), Decimal("250")),
min_usd=Decimal("10"),
max_usd=Decimal("10000"),
card=CardInfo(brand="visa", last4="4242"),
monthly_cap=MonthlyCap(
limit_usd=Decimal("1000"), spent_this_month_usd=Decimal("180"), is_default_ceiling=True
),
portal_url="https://portal/billing?topup=open",
)
monkeypatch.setattr(bv, "build_billing_state", lambda *a, **kw: state)
res = _call("billing.state", {})
assert res["ok"] is True and res["logged_in"] is True
# Money on the wire is STRING, not float/number.
assert res["balance_usd"] == "142.5"
assert res["balance_display"] == "$142.50"
assert res["charge_presets"] == ["100", "250"]
assert res["card"]["masked"] == "visa ····4242"
assert res["monthly_cap"]["is_default_ceiling"] is True
assert res["is_admin"] is True and res["can_charge"] is True
def test_billing_state_fail_open(monkeypatch):
def _boom(*a, **kw):
raise RuntimeError("portal down")
monkeypatch.setattr(bv, "build_billing_state", _boom)
res = _call("billing.state", {})
assert res["ok"] is True and res["logged_in"] is False
# ---------------------------------------------------------------------------
# billing.charge — typed error envelopes
# ---------------------------------------------------------------------------
def test_billing_charge_success_echoes_charge_id(monkeypatch):
monkeypatch.setattr(nb, "post_charge", lambda **kw: {"chargeId": "ch_123"})
res = _call("billing.charge", {"amount_usd": "100", "idempotency_key": "key-1"})
assert res["ok"] is True
assert res["charge_id"] == "ch_123"
assert res["idempotency_key"] == "key-1"
def test_billing_charge_mints_key_when_absent(monkeypatch):
seen = {}
def _post(**kw):
seen["key"] = kw["idempotency_key"]
return {"chargeId": "ch_x"}
monkeypatch.setattr(nb, "post_charge", _post)
res = _call("billing.charge", {"amount_usd": "50"})
assert res["ok"] is True
assert res["idempotency_key"] == seen["key"] # minted key echoed back
assert len(res["idempotency_key"]) == 36
def test_billing_charge_insufficient_scope_envelope(monkeypatch):
def _post(**kw):
raise nb.BillingScopeRequired("need scope", status=403, error="insufficient_scope")
monkeypatch.setattr(nb, "post_charge", _post)
res = _call("billing.charge", {"amount_usd": "100", "idempotency_key": "k"})
assert res["ok"] is False
assert res["error"] == "insufficient_scope"
assert res["idempotency_key"] == "k" # preserved for reuse post-stepup
def test_billing_charge_no_payment_method_envelope(monkeypatch):
def _post(**kw):
raise nb.BillingError(
"no reusable card", status=403, error="no_payment_method",
portal_url="/billing?topup=open",
)
monkeypatch.setattr(nb, "post_charge", _post)
res = _call("billing.charge", {"amount_usd": "100", "idempotency_key": "k"})
assert res["ok"] is False
assert res["error"] == "no_payment_method"
assert res["portal_url"] == "/billing?topup=open"
def test_billing_charge_rate_limited_envelope(monkeypatch):
def _post(**kw):
raise nb.BillingRateLimited("slow down", status=429, error="rate_limited", retry_after=60)
monkeypatch.setattr(nb, "post_charge", _post)
res = _call("billing.charge", {"amount_usd": "100", "idempotency_key": "k"})
assert res["error"] == "rate_limited"
assert res["retry_after"] == 60
# ---------------------------------------------------------------------------
# billing.charge_status — the poll
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"server_resp,expected",
[
({"status": "pending"}, {"status": "pending"}),
(
{"status": "settled", "amountUsd": "50", "settledAt": "2026-06-13T00:00:00Z"},
{"status": "settled", "amount_usd": "50"},
),
({"status": "failed", "reason": "card_declined"}, {"status": "failed", "reason": "card_declined"}),
],
)
def test_billing_charge_status_maps_fields(monkeypatch, server_resp, expected):
monkeypatch.setattr(nb, "get_charge_status", lambda cid, **kw: server_resp)
res = _call("billing.charge_status", {"charge_id": "ch_1"})
assert res["ok"] is True
for k, v in expected.items():
assert res[k] == v
def test_billing_charge_status_requires_id():
res = _call("billing.charge_status", {})
assert res["ok"] is False and res["error"] == "invalid_charge_id"
# ---------------------------------------------------------------------------
# billing.auto_reload
# ---------------------------------------------------------------------------
def test_billing_auto_reload_success(monkeypatch):
seen = {}
monkeypatch.setattr(nb, "patch_auto_top_up", lambda **kw: seen.update(kw) or {"ok": True})
res = _call("billing.auto_reload", {"enabled": True, "threshold": 20, "top_up_amount": 100})
assert res["ok"] is True
assert seen == {"enabled": True, "threshold": 20, "top_up_amount": 100}
def test_billing_auto_reload_validation_error_envelope(monkeypatch):
def _patch(**kw):
raise nb.BillingError("bad", status=400, error="validation_failed")
monkeypatch.setattr(nb, "patch_auto_top_up", _patch)
res = _call("billing.auto_reload", {"enabled": True, "threshold": 20, "top_up_amount": 100})
assert res["ok"] is False and res["error"] == "validation_failed"
def test_billing_auto_reload_requires_fields():
res = _call("billing.auto_reload", {"enabled": True})
assert res["ok"] is False and res["error"] == "invalid_request"
# ---------------------------------------------------------------------------
# billing.step_up
# ---------------------------------------------------------------------------
def test_billing_step_up_granted(monkeypatch):
import hermes_cli.auth as auth
monkeypatch.setattr(auth, "step_up_nous_billing_scope", lambda **kw: True)
res = _call("billing.step_up", {})
assert res["ok"] is True and res["granted"] is True
def test_billing_step_up_downscoped(monkeypatch):
import hermes_cli.auth as auth
monkeypatch.setattr(auth, "step_up_nous_billing_scope", lambda **kw: False)
res = _call("billing.step_up", {})
assert res["ok"] is True and res["granted"] is False

View file

@ -174,6 +174,7 @@ _DETAIL_MODES = frozenset({"hidden", "collapsed", "expanded"})
# response writes are safe.
_LONG_HANDLERS = frozenset(
{
"billing.step_up",
"browser.manage",
"cli.exec",
"plugins.manage",
@ -5190,6 +5191,221 @@ def _(rid, params: dict) -> dict:
return _ok(rid, {"logged_in": False, "balance_lines": [], "identity_line": None, "topup_url": None, "depleted": False})
# ===========================================================================
# Phase 2b terminal billing RPC methods
# ===========================================================================
#
# These return STRUCTURED success envelopes (result.ok / result.error) rather
# than JSON-RPC-level errors, so the TUI's rpc() promise always resolves and the
# Ink side can branch on the typed billing error code (insufficient_scope,
# rate_limited, no_payment_method, …) to render the right affordance instead of
# landing in a generic catch. The data-building lives in the shared core
# (agent/billing_view.py + hermes_cli/nous_billing.py) — same as /credits.
def _serialize_billing_error(exc) -> dict:
"""Map a BillingError into the result.error envelope the TUI branches on."""
from hermes_cli.nous_billing import (
BillingRateLimited,
BillingScopeRequired,
)
kind = "error"
if isinstance(exc, BillingScopeRequired):
kind = "insufficient_scope"
elif isinstance(exc, BillingRateLimited):
kind = "rate_limited"
elif getattr(exc, "error", None):
kind = str(exc.error)
return {
"ok": False,
"error": kind,
"message": str(exc),
"portal_url": getattr(exc, "portal_url", None),
"retry_after": getattr(exc, "retry_after", None),
"payload": getattr(exc, "payload", {}) or {},
}
def _serialize_billing_state(state) -> dict:
"""Serialize a BillingState for the wire (Decimals → strings, money-safe)."""
from agent.billing_view import format_money
def _s(value):
return None if value is None else str(value)
card = None
if state.card is not None:
card = {"brand": state.card.brand, "last4": state.card.last4, "masked": state.card.masked}
monthly_cap = None
if state.monthly_cap is not None:
mc = state.monthly_cap
monthly_cap = {
"limit_usd": _s(mc.limit_usd),
"limit_display": format_money(mc.limit_usd),
"spent_this_month_usd": _s(mc.spent_this_month_usd),
"spent_display": format_money(mc.spent_this_month_usd),
"is_default_ceiling": mc.is_default_ceiling,
}
auto_reload = None
if state.auto_reload is not None:
ar = state.auto_reload
auto_reload = {
"enabled": ar.enabled,
"threshold_usd": _s(ar.threshold_usd),
"threshold_display": format_money(ar.threshold_usd),
"reload_to_usd": _s(ar.reload_to_usd),
"reload_to_display": format_money(ar.reload_to_usd),
}
return {
"ok": True,
"logged_in": state.logged_in,
"org_name": state.org_name,
"org_slug": state.org_slug,
"role": state.role,
"is_admin": state.is_admin,
"can_charge": state.can_charge,
"balance_usd": _s(state.balance_usd),
"balance_display": format_money(state.balance_usd),
"cli_billing_enabled": state.cli_billing_enabled,
"charge_presets": [_s(p) for p in state.charge_presets],
"charge_presets_display": [format_money(p) for p in state.charge_presets],
"min_usd": _s(state.min_usd),
"max_usd": _s(state.max_usd),
"card": card,
"monthly_cap": monthly_cap,
"auto_reload": auto_reload,
"portal_url": state.portal_url,
"error": state.error,
}
@method("billing.state")
def _(rid, params: dict) -> dict:
"""GET /api/billing/state → serialized BillingState (Screen 1 + 5).
Fail-open like credits.view: a logged-out / unreachable portal yields
{ok:true, logged_in:false}. No scope required for this endpoint.
"""
try:
from agent.billing_view import build_billing_state
state = build_billing_state()
return _ok(rid, _serialize_billing_state(state))
except Exception:
return _ok(rid, {"ok": True, "logged_in": False, "error": "could not load billing state"})
@method("billing.charge")
def _(rid, params: dict) -> dict:
"""POST /api/billing/charge → {ok, chargeId} or a typed error envelope.
params: {amount_usd: str|number, idempotency_key?: str}. If no key is
supplied, the server-side core mints a fresh one and returns it so the TUI can
reuse it on retry of the SAME purchase.
"""
from hermes_cli.nous_billing import BillingError, post_charge
from agent.billing_view import new_idempotency_key
amount = params.get("amount_usd")
if amount is None:
return _ok(rid, {"ok": False, "error": "invalid_request", "message": "amount_usd is required"})
key = params.get("idempotency_key") or new_idempotency_key()
try:
result = post_charge(amount_usd=amount, idempotency_key=key)
return _ok(rid, {"ok": True, "charge_id": result.get("chargeId"), "idempotency_key": key})
except BillingError as exc:
env = _serialize_billing_error(exc)
env["idempotency_key"] = key # so the TUI can reuse on retry
return _ok(rid, env)
except Exception as exc:
return _ok(rid, {"ok": False, "error": "error", "message": str(exc), "idempotency_key": key})
@method("billing.charge_status")
def _(rid, params: dict) -> dict:
"""GET /api/billing/charge/{id}{ok, status, ...} or typed error.
The poll. Caller drives the 2s/5-min cadence; this is a single status read.
"""
from hermes_cli.nous_billing import BillingError, get_charge_status
charge_id = params.get("charge_id")
if not charge_id:
return _ok(rid, {"ok": False, "error": "invalid_charge_id", "message": "charge_id is required"})
try:
result = get_charge_status(charge_id)
return _ok(
rid,
{
"ok": True,
"status": result.get("status"),
"amount_usd": result.get("amountUsd"),
"settled_at": result.get("settledAt"),
"reason": result.get("reason"),
},
)
except BillingError as exc:
return _ok(rid, _serialize_billing_error(exc))
except Exception as exc:
return _ok(rid, {"ok": False, "error": "error", "message": str(exc)})
@method("billing.auto_reload")
def _(rid, params: dict) -> dict:
"""PATCH /api/billing/auto-top-up → {ok:true} or typed error (Screen 2).
params: {enabled: bool, threshold: number, top_up_amount: number}.
"""
from hermes_cli.nous_billing import BillingError, patch_auto_top_up
try:
enabled = bool(params.get("enabled"))
threshold = params.get("threshold")
top_up_amount = params.get("top_up_amount")
if threshold is None or top_up_amount is None:
return _ok(rid, {"ok": False, "error": "invalid_request", "message": "threshold and top_up_amount are required"})
patch_auto_top_up(enabled=enabled, threshold=threshold, top_up_amount=top_up_amount)
return _ok(rid, {"ok": True})
except BillingError as exc:
return _ok(rid, _serialize_billing_error(exc))
except Exception as exc:
return _ok(rid, {"ok": False, "error": "error", "message": str(exc)})
@method("billing.step_up")
def _(rid, params: dict) -> dict:
"""Run the lazy billing:manage step-up device flow → {ok, granted}.
Triggered by the TUI after a billing call returns error=insufficient_scope.
Returns granted:false when the server silently downscopes (non-admin / unticked).
Runs on the thread pool (in _LONG_HANDLERS): the device flow blocks for the
whole device-code lifetime (minutes), so it must not stall the main stdin loop.
The verification URL/code reach the TUI via an out-of-band ``billing.step_up.
verification`` event (a plain print would be dropped by the JSON-RPC stdout
pipe), and the browser is opened TUI-side via openExternalUrl never with the
gateway's headless webbrowser.open (hence open_browser=False).
"""
sid = params.get("session_id") or ""
try:
from hermes_cli.auth import step_up_nous_billing_scope
def _on_verification(url: str, code: str) -> None:
_emit(
"billing.step_up.verification",
sid,
{"verification_url": url, "user_code": code},
)
granted = step_up_nous_billing_scope(
open_browser=False, on_verification=_on_verification
)
return _ok(rid, {"ok": True, "granted": bool(granted)})
except Exception as exc:
return _ok(rid, {"ok": False, "error": "error", "message": str(exc), "granted": False})
@method("session.status")
def _(rid, params: dict) -> dict:
session, err = _sess_nowait(params, rid)

View file

@ -0,0 +1,301 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
import { getOverlayState, resetOverlayState } from '../app/overlayStore.js'
import { billingCommands } from '../app/slash/commands/billing.js'
import type { BillingStateResponse } from '../gatewayTypes.js'
vi.mock('../lib/openExternalUrl.js', () => ({
openExternalUrl: vi.fn(() => true)
}))
const billingCommand = billingCommands.find(cmd => cmd.name === 'billing')!
const ownerState = (overrides: Partial<BillingStateResponse> = {}): BillingStateResponse => ({
auto_reload: {
enabled: false,
reload_to_display: '—',
reload_to_usd: null,
threshold_display: '—',
threshold_usd: null
},
balance_display: '$142.50',
balance_usd: '142.5',
can_charge: true,
card: { brand: 'visa', last4: '4242', masked: 'visa ····4242' },
charge_presets: ['25', '50', '100'],
charge_presets_display: ['$25', '$50', '$100'],
cli_billing_enabled: true,
is_admin: true,
logged_in: true,
max_usd: '10000',
min_usd: '10',
monthly_cap: {
is_default_ceiling: true,
limit_display: '$1000',
limit_usd: '1000',
spent_display: '$180',
spent_this_month_usd: '180'
},
ok: true,
org_name: 'Acme',
portal_url: 'https://portal/billing?topup=open',
role: 'OWNER',
...overrides
})
const guarded =
<T>(fn: (r: T) => void) =>
(r: null | T) => {
if (r) {
fn(r)
}
}
/** Build a ctx whose rpc routes by method name to a supplied map of results. */
const buildCtx = (results: Record<string, unknown>) => {
const sys = vi.fn()
const calls: Array<{ method: string; params: unknown }> = []
const rpc = vi.fn((method: string, params: unknown) => {
calls.push({ method, params })
return Promise.resolve(results[method])
})
const ctx = {
gateway: { rpc },
guarded,
guardedErr: vi.fn(),
sid: 'sid-1',
stale: () => false,
transcript: { page: vi.fn(), panel: vi.fn(), sys }
}
const run = async (arg: string) => {
billingCommand.run(arg, ctx as any, 'billing')
await rpc.mock.results[0]?.value
await Promise.resolve()
await Promise.resolve()
}
return { calls, ctx, rpc, run, sys }
}
const printed = (sys: ReturnType<typeof vi.fn>) => sys.mock.calls.map(c => c[0]).join('\n')
describe('/billing slash command (overlay-driven)', () => {
beforeEach(() => {
resetOverlayState()
})
it('not logged in → prompts to log in, no overlay', async () => {
const { run, sys } = buildCtx({ 'billing.state': { ...ownerState(), logged_in: false, ok: true } })
await run('')
expect(printed(sys)).toContain('Not logged into Nous Portal')
expect(getOverlayState().billing).toBeNull()
})
it('bare /billing opens the overlay on the overview screen with state', async () => {
const { run, rpc } = buildCtx({ 'billing.state': ownerState() })
await run('')
expect(rpc).toHaveBeenCalledWith('billing.state', {})
const billing = getOverlayState().billing
expect(billing).toBeTruthy()
expect(billing?.screen).toBe('overview')
expect(billing?.state.balance_display).toBe('$142.50')
expect(billing?.state.charge_presets_display).toEqual(['$25', '$50', '$100'])
})
it('any sub-command arg is ignored — still opens the overview overlay', async () => {
const { run } = buildCtx({ 'billing.state': ownerState() })
await run('buy 100')
const billing = getOverlayState().billing
expect(billing?.screen).toBe('overview')
// No confirm overlay armed directly by the command anymore.
expect(getOverlayState().confirm).toBeNull()
})
it('member overview carries the non-admin state for component-side gating', async () => {
const { run } = buildCtx({
'billing.state': ownerState({
is_admin: false,
can_charge: false,
role: 'MEMBER',
card: null,
monthly_cap: null,
auto_reload: null
})
})
await run('')
const billing = getOverlayState().billing
expect(billing?.state.is_admin).toBe(false)
expect(billing?.screen).toBe('overview')
})
// ── Overlay ctx behaviors (RPC + error mapping live in billing.ts) ──
it('ctx.validate rejects out-of-bounds and sub-cent amounts, accepts valid', async () => {
const { run } = buildCtx({ 'billing.state': ownerState() })
await run('')
const ctx = getOverlayState().billing!.ctx
expect(ctx.validate('5').error).toContain('Minimum is $10')
expect(ctx.validate('10.005').error).toContain('2 decimal places')
expect(ctx.validate('100').amount).toBe('100')
expect(ctx.validate('$50').amount).toBe('50')
})
it('ctx.charge → poll → settled', async () => {
vi.useFakeTimers()
try {
const { run, sys } = buildCtx({
'billing.state': ownerState(),
'billing.charge': { ok: true, charge_id: 'ch_1', idempotency_key: 'k' },
'billing.charge_status': { ok: true, status: 'settled', amount_usd: '100' }
})
await run('')
const ctx = getOverlayState().billing!.ctx
ctx.charge('100')
await vi.runAllTimersAsync()
const out = printed(sys)
expect(out).toContain('Charge submitted')
expect(out).toContain('✅ $100 added.')
} finally {
vi.useRealTimers()
}
})
it('ctx.charge → poll → failed adds the portal funnel line', async () => {
vi.useFakeTimers()
try {
const { run, sys } = buildCtx({
'billing.state': ownerState(),
'billing.charge': { ok: true, charge_id: 'ch_1', idempotency_key: 'k' },
'billing.charge_status': { ok: true, status: 'failed', reason: 'card_declined' }
})
await run('')
getOverlayState().billing!.ctx.charge('100')
await vi.runAllTimersAsync()
const out = printed(sys)
expect(out).toContain('Your card was declined')
// Parity with the CLI: a failed poll funnels to the portal (from state.portal_url).
expect(out).toContain('Portal: https://portal/billing?topup=open')
} finally {
vi.useRealTimers()
}
})
it('ctx.charge monthly_cap_exceeded surfaces remaining headroom', async () => {
const { run, sys } = buildCtx({
'billing.state': ownerState(),
'billing.charge': {
ok: false,
error: 'monthly_cap_exceeded',
message: 'Monthly spend cap reached.',
payload: { remainingUsd: '42.50' },
portal_url: '/billing?topup=open',
idempotency_key: 'k'
}
})
await run('')
getOverlayState().billing!.ctx.charge('100')
await Promise.resolve()
await Promise.resolve()
const out = printed(sys)
expect(out).toContain('Monthly spend cap reached — $42.50 headroom left.')
expect(out).toContain('Portal: /billing?topup=open')
})
it('ctx.charge no_payment_method → portal funnel copy', async () => {
const { run, sys } = buildCtx({
'billing.state': ownerState(),
'billing.charge': {
ok: false,
error: 'no_payment_method',
portal_url: '/billing?topup=open',
idempotency_key: 'k'
}
})
await run('')
getOverlayState().billing!.ctx.charge('100')
await Promise.resolve()
await Promise.resolve()
const out = printed(sys)
expect(out).toContain('No saved card for terminal charges')
expect(out).toContain('Portal: /billing?topup=open')
})
it('ctx.charge insufficient_scope → arms step-up confirm', async () => {
const { run } = buildCtx({
'billing.state': ownerState(),
'billing.charge': { ok: false, error: 'insufficient_scope', idempotency_key: 'k' }
})
await run('')
getOverlayState().billing!.ctx.charge('100')
await Promise.resolve()
await Promise.resolve()
// The charge failed with insufficient_scope → a NEW confirm (step-up) is armed.
const stepUp = getOverlayState().confirm
expect(stepUp?.title).toBe('Grant terminal billing access?')
})
it('ctx.applyAutoReload(true, …) → billing.auto_reload RPC, resolves true', async () => {
const { run, calls } = buildCtx({
'billing.state': ownerState(),
'billing.auto_reload': { ok: true }
})
await run('')
const ok = await getOverlayState().billing!.ctx.applyAutoReload(true, 20, 100)
expect(ok).toBe(true)
const ar = calls.find(c => c.method === 'billing.auto_reload')
expect(ar?.params).toEqual({ enabled: true, threshold: 20, top_up_amount: 100 })
})
it('ctx.applyAutoReload(false) → disables (enabled:false, no amounts)', async () => {
const { run, calls } = buildCtx({
'billing.state': ownerState({
auto_reload: {
enabled: true,
reload_to_display: '$100',
reload_to_usd: '100',
threshold_display: '$20',
threshold_usd: '20'
}
}),
'billing.auto_reload': { ok: true }
})
await run('')
const ok = await getOverlayState().billing!.ctx.applyAutoReload(false)
expect(ok).toBe(true)
const ar = calls.find(c => c.method === 'billing.auto_reload')
expect(ar?.params).toEqual({ enabled: false })
})
it('ctx.applyAutoReload error → resolves false + maps the error', async () => {
const { run, sys } = buildCtx({
'billing.state': ownerState(),
'billing.auto_reload': { ok: false, error: 'monthly_cap_exceeded', message: 'Monthly spend cap reached.' }
})
await run('')
const ok = await getOverlayState().billing!.ctx.applyAutoReload(true, 20, 100)
expect(ok).toBe(false)
expect(printed(sys)).toContain('Monthly spend cap reached.')
})
it('ctx.openPortal opens the URL + echoes a transcript line', async () => {
const { run, sys } = buildCtx({ 'billing.state': ownerState() })
await run('')
getOverlayState().billing!.ctx.openPortal('https://portal/x')
expect(printed(sys)).toContain('Opening portal: https://portal/x')
})
})

View file

@ -8,6 +8,13 @@ import { getUiState, patchUiState, resetUiState } from '../app/uiStore.js'
import { estimateTokensRough } from '../lib/text.js'
import type { Msg } from '../types.js'
// Mock the external-URL opener so the billing.step_up.verification test can
// assert it's invoked without spawning a real browser process.
const openExternalUrlMock = vi.fn((_url: string) => true)
vi.mock('../lib/openExternalUrl.js', () => ({
openExternalUrl: (url: string) => openExternalUrlMock(url)
}))
const ref = <T>(current: T) => ({ current })
const buildCtx = (appended: Msg[]) =>
@ -1561,4 +1568,34 @@ describe('createGatewayEventHandler', () => {
expect(getUiState().notice).toBeNull()
})
})
describe('billing.step_up.verification', () => {
beforeEach(() => {
openExternalUrlMock.mockClear()
})
it('renders the verification link + code and opens the browser', () => {
const ctx = buildCtx([])
const onEvent = createGatewayEventHandler(ctx)
onEvent({
payload: { user_code: 'WXYZ-9999', verification_url: 'https://portal.example/device?code=WXYZ' },
type: 'billing.step_up.verification'
} as any)
const printed = (ctx.system.sys as ReturnType<typeof vi.fn>).mock.calls.map(c => c[0]).join('\n')
expect(printed).toContain('https://portal.example/device?code=WXYZ')
expect(printed).toContain('WXYZ-9999')
expect(openExternalUrlMock).toHaveBeenCalledWith('https://portal.example/device?code=WXYZ')
})
it('no-ops on a missing verification_url (never opens a browser)', () => {
const ctx = buildCtx([])
const onEvent = createGatewayEventHandler(ctx)
onEvent({ payload: { verification_url: '' }, type: 'billing.step_up.verification' } as any)
expect(openExternalUrlMock).not.toHaveBeenCalled()
})
})
})

View file

@ -10,6 +10,7 @@ import type {
SessionMostRecentResponse
} from '../gatewayTypes.js'
import { rpcErrorMessage } from '../lib/rpc.js'
import { openExternalUrl } from '../lib/openExternalUrl.js'
import { topLevelSubagents } from '../lib/subagentTree.js'
import { formatAbandonedClarify, formatToolCall, stripAnsi } from '../lib/text.js'
import { fromSkin } from '../theme.js'
@ -533,6 +534,29 @@ export function createGatewayEventHandler(ctx: GatewayEventHandlerContext): (ev:
turnController.clearNotice(ev.payload?.key)
return
case 'billing.step_up.verification': {
// The billing step-up device flow runs in the headless gateway, so it
// can't open a browser or print the URL where the user sees it. Surface
// the link here (clickable/copyable in the transcript) and best-effort
// open it via the TUI process's own opener. This event arrives while the
// billing.step_up RPC is still polling (and may even outlive the RPC's
// 120s timeout), so the link — not the RPC result — is the source of truth.
const url = ev.payload.verification_url
const code = ev.payload.user_code
if (!url) {
return
}
sys('💳 Open this link to grant terminal billing access:')
sys(url)
if (code) {
sys(`If prompted, enter code: ${code}`)
}
void openExternalUrl(url)
return
}
case 'gateway.stderr': {
const line = String(ev.payload.line).slice(0, 120)

View file

@ -3,7 +3,7 @@ import type { MutableRefObject, ReactNode, RefObject, SetStateAction } from 'rea
import type { PasteEvent } from '../components/textInput.js'
import type { GatewayClient } from '../gatewayClient.js'
import type { ImageAttachResponse, SessionCloseResponse } from '../gatewayTypes.js'
import type { BillingStateResponse, ImageAttachResponse, SessionCloseResponse } from '../gatewayTypes.js'
import type { ParsedVoiceRecordKey } from '../lib/platform.js'
import type { RpcResult } from '../lib/rpc.js'
import type { Theme } from '../theme.js'
@ -85,10 +85,53 @@ export interface GatewayProviderProps {
value: GatewayServices
}
// ── Billing overlay (Phase 2b: full-modal TUI parity) ────────────────
// The /billing command no longer parses sub-commands; bare `/billing`
// fetches `billing.state` and opens this overlay. The overlay is a small
// state machine (overview → buy|autoreload|limit → confirm) that performs
// the SAME RPCs as the old slash flows (billing.charge / charge_status /
// auto_reload / step_up). Backend is unchanged & shared with the CLI.
export type BillingScreen = 'autoreload' | 'buy' | 'confirm' | 'limit' | 'overview'
/**
* The functions the overlay needs to talk to the gateway and emit
* transcript lines. Built once in `billing.ts` (closing over the live
* SlashRunCtx) and stashed in the overlay slot, mirroring how a ConfirmReq
* stashes its `onConfirm` closure. Keeps all RPC + error-mapping logic in
* billing.ts (single source of truth) the overlay only renders + routes.
*/
export interface BillingOverlayCtx {
/** Run `billing.auto_reload` (enabled/threshold/top_up) → resolve ok/false. */
applyAutoReload: (enabled: boolean, threshold?: number, topUp?: number) => Promise<boolean>
/** Submit `billing.charge` for `amount` and poll to settlement (non-blocking). */
charge: (amount: string) => void
/** Open the portal in the browser + echo a transcript line. */
openPortal: (url: string) => void
/** Emit a transcript system line. */
sys: (text: string) => void
/** Validate a custom amount against state bounds + 2dp (mirrors the server). */
validate: (raw: string) => { amount?: string; error?: string }
}
/** Pending confirm built when leaving the buy/autoreload screen. */
export interface BillingPendingCharge {
amount: string
}
export interface BillingOverlayState {
ctx: BillingOverlayCtx
/** Set when on the 'confirm' screen for a buy. */
pendingCharge?: BillingPendingCharge | null
screen: BillingScreen
state: BillingStateResponse
}
export interface OverlayState {
agents: boolean
agentsInitialHistoryIndex: number
approval: ApprovalReq | null
billing: BillingOverlayState | null
clarify: ClarifyReq | null
confirm: ConfirmReq | null
modelPicker: boolean

View file

@ -6,6 +6,7 @@ const buildOverlayState = (): OverlayState => ({
agents: false,
agentsInitialHistoryIndex: 0,
approval: null,
billing: null,
clarify: null,
confirm: null,
modelPicker: false,
@ -21,9 +22,20 @@ export const $overlayState = atom<OverlayState>(buildOverlayState())
export const $isBlocked = computed(
$overlayState,
({ agents, approval, clarify, confirm, modelPicker, pager, pluginsHub, secret, sessions, skillsHub, sudo }) =>
({ agents, approval, billing, clarify, confirm, modelPicker, pager, pluginsHub, secret, sessions, skillsHub, sudo }) =>
Boolean(
agents || approval || clarify || confirm || modelPicker || pager || pluginsHub || secret || sessions || skillsHub || sudo
agents ||
approval ||
billing ||
clarify ||
confirm ||
modelPicker ||
pager ||
pluginsHub ||
secret ||
sessions ||
skillsHub ||
sudo
)
)

View file

@ -0,0 +1,332 @@
import type {
BillingChargeResponse,
BillingChargeStatusResponse,
BillingErrorPayload,
BillingMutationResponse,
BillingStateResponse
} from '../../../gatewayTypes.js'
import { openExternalUrl } from '../../../lib/openExternalUrl.js'
import type { BillingOverlayCtx } from '../../interfaces.js'
import { patchOverlayState } from '../../overlayStore.js'
import type { SlashCommand, SlashRunCtx } from '../types.js'
// Poll cadence (plan §5, frozen): 2s interval, 5-minute cap.
const POLL_INTERVAL_MS = 2000
const POLL_CAP_MS = 5 * 60 * 1000
type Sys = (text: string) => void
/** Map a typed billing error envelope to user-facing copy + portal funnel. */
const renderBillingError = (
sys: Sys,
ctx: SlashRunCtx,
env: {
error?: string
message?: string
payload?: BillingErrorPayload
portal_url?: string | null
retry_after?: number | null
}
): void => {
const portal = env.portal_url
switch (env.error) {
case 'insufficient_scope':
armStepUp(sys, ctx)
return
case 'no_payment_method':
sys(
'💳 No saved card for terminal charges yet. Set one up on the portal ' +
"(one-time credit buys don't save a reusable card)."
)
break
case 'cli_billing_disabled':
sys('🔴 Terminal billing is turned off for this org — an admin must enable it on the portal.')
break
case 'monthly_cap_exceeded': {
// Surface the remaining headroom the server attaches (parity with the CLI).
const remaining = env.payload?.remainingUsd
sys(remaining != null ? `🔴 Monthly spend cap reached — $${remaining} headroom left.` : '🔴 Monthly spend cap reached.')
break
}
case 'rate_limited': {
const mins = env.retry_after ? ` (try again in ~${Math.max(1, Math.round(env.retry_after / 60))} min)` : ''
sys(`🟡 Too many charges right now${mins}. This isn't a payment failure.`)
break
}
default:
sys(`🔴 ${env.message || env.error || 'Billing request failed.'}`)
}
if (portal) {
sys(`Portal: ${portal}`)
}
}
/** 403 insufficient_scope → arm a ConfirmReq that runs the lazy step-up. */
const armStepUp = (sys: Sys, ctx: SlashRunCtx): void => {
sys('💳 Terminal billing needs an extra permission (billing:manage).')
patchOverlayState({
confirm: {
cancelLabel: 'Not now',
confirmLabel: 'Re-authorize',
detail: 'An org admin/owner must tick "Allow terminal billing" in the portal.',
onConfirm: () => {
// session_id lets the gateway route the billing.step_up.verification
// event (the verification link) back to this session — the device flow
// runs headless in the gateway, so the link can't be printed there.
ctx.gateway
.rpc<BillingMutationResponse>('billing.step_up', { session_id: ctx.sid ?? undefined })
.then(
ctx.guarded<BillingMutationResponse>(r => {
if (r.ok && r.granted) {
// Step-up only grants the billing:manage TOKEN scope — the ORG
// kill-switch (cli_billing_enabled) is a separate gate. Re-fetch
// /state so we don't over-promise "enabled" when a charge would
// still hit cli_billing_disabled.
sys('✅ Billing permission granted.')
ctx.gateway
.rpc<BillingStateResponse>('billing.state', {})
.then(
ctx.guarded<BillingStateResponse>(s => {
if (s.cli_billing_enabled) {
sys('Run /billing again to continue.')
} else {
sys(
'🟡 Permission granted, but terminal billing is still turned off ' +
'for this org. Enable it in the portal, then run /billing again.'
)
if (s.portal_url) {
sys(`Portal: ${s.portal_url}`)
}
}
})
)
.catch(() => {
sys('Run /billing again to continue.')
})
} else {
sys('🟡 Terminal billing was not granted (an admin must tick the box).')
}
})
)
.catch(() => {
// The device flow can outlive the RPC's 120s timeout while the user
// is still authorizing in the browser. A reject here is NOT a hard
// failure — the grant (if it lands) is persisted gateway-side; tell
// the user to re-run /billing rather than reporting an error.
sys('🟡 Still waiting on approval — finish in the browser, then run /billing again.')
})
},
title: 'Grant terminal billing access?'
}
})
}
/** Poll a charge to a terminal state (settled/failed/timeout). Non-blocking. */
const pollCharge = (sys: Sys, ctx: SlashRunCtx, chargeId: string, portalUrl?: string | null): void => {
const start = Date.now()
const tick = (): void => {
if (ctx.stale()) {
return
}
ctx.gateway
.rpc<BillingChargeStatusResponse>('billing.charge_status', { charge_id: chargeId })
.then(
ctx.guarded<BillingChargeStatusResponse>(r => {
if (!r.ok) {
// 429/503 while polling = retry-after, NOT a failure. Back off + continue.
if (r.error === 'rate_limited') {
const wait = (r.retry_after ?? 5) * 1000
setTimeout(tick, Math.min(wait, 30000))
return
}
sys(`🔴 Could not check the charge: ${r.message || r.error || 'error'}`)
return
}
if (r.status === 'settled') {
sys(`${r.amount_usd ? `$${r.amount_usd}` : 'Credits'} added.`)
return
}
if (r.status === 'failed') {
renderChargeFailed(sys, r.reason, portalUrl)
return
}
// pending → keep polling until the 5-min cap, then call it a timeout.
if (Date.now() - start >= POLL_CAP_MS) {
sys(
'🟡 Still processing after 5 minutes — this is a timeout, not a failure. ' +
'Check /billing or the portal shortly.'
)
if (portalUrl) {
sys(`Portal: ${portalUrl}`)
}
return
}
setTimeout(tick, POLL_INTERVAL_MS)
})
)
.catch(ctx.guardedErr)
}
tick()
}
const renderChargeFailed = (sys: Sys, reason?: string | null, portalUrl?: string | null): void => {
switch ((reason || '').trim()) {
case 'authentication_required':
sys('🔴 Your bank requires verification (3DS). Complete it on the portal to finish this purchase.')
break
case 'payment_method_expired':
sys('🔴 Your card has expired. Update it on the portal.')
break
case 'card_declined':
sys('🔴 Your card was declined. Try another card on the portal.')
break
default:
sys(`🔴 The charge didn't go through (${reason || 'processing_error'}).`)
}
// Funnel to the portal after any failure (parity with cli.py _billing_portal_hint).
if (portalUrl) {
sys(`Portal: ${portalUrl}`)
}
}
/** Validate a custom amount against state bounds + 2dp, mirroring the server. */
const validateAmount = (raw: string, s: BillingStateResponse): { amount?: string; error?: string } => {
const cleaned = raw.trim().replace(/^\$/, '').trim()
if (!cleaned || !/^\d+(\.\d{1,2})?$/.test(cleaned)) {
return { error: 'Enter a dollar amount, e.g. 100 (max 2 decimal places).' }
}
const value = Number(cleaned)
if (!(value > 0)) {
return { error: 'Amount must be greater than $0.' }
}
if (s.min_usd != null && value < Number(s.min_usd)) {
return { error: `Minimum is $${s.min_usd}.` }
}
if (s.max_usd != null && value > Number(s.max_usd)) {
return { error: `Maximum is $${s.max_usd}.` }
}
return { amount: cleaned }
}
/**
* Build the closure bundle the BillingOverlay needs to talk to the gateway
* and emit transcript lines. Keeps ALL RPC + error-mapping logic here
* (single source of truth) the overlay only renders + routes keys.
*/
const buildOverlayCtx = (ctx: SlashRunCtx, sys: Sys, s: BillingStateResponse): BillingOverlayCtx => ({
applyAutoReload: (enabled, threshold, topUp) =>
ctx.gateway
.rpc<BillingMutationResponse>('billing.auto_reload', {
enabled,
...(threshold != null ? { threshold } : {}),
...(topUp != null ? { top_up_amount: topUp } : {})
})
.then(r => {
if (r && r.ok) {
return true
}
if (r) {
renderBillingError(sys, ctx, r)
}
return false
})
.catch(e => {
ctx.guardedErr(e)
return false
}),
charge: (amount: string) => {
sys('💳 Charge submitted — confirming settlement…')
ctx.gateway
.rpc<BillingChargeResponse>('billing.charge', { amount_usd: amount })
.then(
ctx.guarded<BillingChargeResponse>(r => {
if (r.ok && r.charge_id) {
pollCharge(sys, ctx, r.charge_id, s.portal_url)
} else {
renderBillingError(sys, ctx, r)
}
})
)
.catch(ctx.guardedErr)
},
openPortal: (url: string) => {
openExternalUrl(url)
sys(`Opening portal: ${url}`)
},
sys,
validate: (raw: string) => validateAmount(raw, s)
})
export const billingCommands: SlashCommand[] = [
{
help: 'Manage Nous terminal billing — buy credits, auto-reload, limits',
name: 'billing',
// ZERO sub-commands (plan §0.4): any arg is ignored. Bare `/billing`
// fetches state and opens the interactive overlay (CLI/TUI parity).
run: (_arg, ctx) => {
const sys: Sys = ctx.transcript.sys
ctx.gateway
.rpc<BillingStateResponse>('billing.state', {})
.then(
ctx.guarded<BillingStateResponse>(s => {
if (!s.logged_in) {
sys('💳 Not logged into Nous Portal — run /portal to log in, then /billing.')
return
}
patchOverlayState({
billing: {
ctx: buildOverlayCtx(ctx, sys, s),
pendingCharge: null,
screen: 'overview',
state: s
}
})
})
)
.catch(ctx.guardedErr)
}
}
]

View file

@ -1,4 +1,5 @@
import { coreCommands } from './commands/core.js'
import { billingCommands } from './commands/billing.js'
import { creditsCommands } from './commands/credits.js'
import { debugCommands } from './commands/debug.js'
import { opsCommands } from './commands/ops.js'
@ -8,6 +9,7 @@ import type { SlashCommand } from './types.js'
export const SLASH_COMMANDS: SlashCommand[] = [
...coreCommands,
...billingCommands,
...creditsCommands,
...sessionCommands,
...opsCommands,

View file

@ -147,6 +147,10 @@ export function useInputHandlers(ctx: InputHandlerContext): InputHandlerResult {
return patchOverlayState({ modelPicker: false })
}
if (overlay.billing) {
return patchOverlayState({ billing: null })
}
if (overlay.skillsHub) {
return patchOverlayState({ skillsHub: false })
}
@ -272,7 +276,7 @@ export function useInputHandlers(ctx: InputHandlerContext): InputHandlerResult {
// answering felt like the prompt had locked the entire UI. Explicitly
// skip the prompt-overlay early-return for scroll keys so they fall
// through to the wheel / PageUp / Shift+arrow handlers below.
const promptOverlay = overlay.approval || overlay.clarify || overlay.confirm
const promptOverlay = overlay.approval || overlay.billing || overlay.clarify || overlay.confirm
const fallThroughForScroll = promptOverlay && shouldFallThroughForScroll(key)
if (promptOverlay && !fallThroughForScroll) {

View file

@ -8,6 +8,7 @@ import { $uiSessionId, $uiTheme } from '../app/uiStore.js'
import { ActiveSessionSwitcher } from './activeSessionSwitcher.js'
import { FloatBox } from './appChrome.js'
import { BillingOverlay } from './billingOverlay.js'
import { MaskedPrompt } from './maskedPrompt.js'
import { ModelPicker } from './modelPicker.js'
import { OverlayHint } from './overlayControls.js'
@ -35,6 +36,21 @@ export function PromptZone({
)
}
if (overlay.billing) {
const current = overlay.billing
const onPatch = (next: Partial<typeof current>) =>
patchOverlayState(prev => (prev.billing ? { ...prev, billing: { ...prev.billing, ...next } } : prev))
const onClose = () => patchOverlayState({ billing: null })
return (
<Box flexDirection="column" flexShrink={0} paddingX={1} paddingY={1}>
<BillingOverlay onClose={onClose} onPatch={onPatch} overlay={current} t={theme} />
</Box>
)
}
if (overlay.confirm) {
const req = overlay.confirm

View file

@ -0,0 +1,684 @@
import { Box, Text, useInput } from '@hermes/ink'
import { useState } from 'react'
import type { BillingOverlayState } from '../app/interfaces.js'
import type { BillingStateResponse } from '../gatewayTypes.js'
import type { Theme } from '../theme.js'
import { TextInput } from './textInput.js'
const SPEND_BAR_CELLS = 10
interface BillingOverlayProps {
/** Replace the overlay slot (screen transitions + pending data). */
onPatch: (next: Partial<BillingOverlayState>) => void
/** Close the overlay entirely. */
onClose: () => void
overlay: BillingOverlayState
t: Theme
}
/** A numbered menu row with the ▸ cursor (mirrors ClarifyPrompt). */
function MenuRow({ active, index, label, t }: { active: boolean; index: number; label: string; t: Theme }) {
return (
<Text>
<Text bold={active} color={active ? t.color.label : t.color.muted} inverse={active}>
{active ? '▸ ' : ' '}
{index}. {label}
</Text>
</Text>
)
}
/** Plain (non-numbered) action row with the ▸ cursor (confirm screens). */
function ActionRow({ active, label, color, t }: { active: boolean; label: string; color?: string; t: Theme }) {
return (
<Text>
<Text color={active ? t.color.accent : t.color.muted}>{active ? '▸ ' : ' '}</Text>
<Text bold={active} color={active ? (color ?? t.color.text) : t.color.muted}>
{label}
</Text>
</Text>
)
}
/** 10-cell spend bar + percent (omit entirely when there's no usable cap). */
function spendBar(s: BillingStateResponse): null | string {
const cap = s.monthly_cap
if (!cap || cap.limit_usd == null) {
return null
}
const limit = Number(cap.limit_usd)
const spent = Number(cap.spent_this_month_usd ?? '0')
if (!(limit > 0) || Number.isNaN(spent)) {
return null
}
const ratio = Math.max(0, Math.min(1, spent / limit))
const filled = Math.round(ratio * SPEND_BAR_CELLS)
const bar = '█'.repeat(filled) + '░'.repeat(SPEND_BAR_CELLS - filled)
const pct = Math.round(ratio * 100)
const ceiling = cap.is_default_ceiling ? ' (default ceiling)' : ''
return `${cap.spent_display} of ${cap.limit_display} used ${bar} ${pct}%${ceiling}`
}
function autoReloadLine(s: BillingStateResponse): null | string {
if (!s.auto_reload) {
return null
}
return s.auto_reload.enabled
? `Auto-reload: on (below ${s.auto_reload.threshold_display}${s.auto_reload.reload_to_display})`
: 'Auto-reload: off'
}
const footer = (extra: string, t: Theme) => <Text color={t.color.muted}>{extra}</Text>
/**
* The /billing modal. A self-contained state machine:
* overview buy | autoreload | limit (and buy confirm).
* Esc from a sub-screen returns to overview; Esc from overview closes.
* All RPCs + error mapping live in billing.ts and are reached through
* `overlay.ctx` this component only renders + routes keys.
*/
export function BillingOverlay({ onClose, onPatch, overlay, t }: BillingOverlayProps) {
const { ctx, screen, state: s } = overlay
return (
<Box borderColor={t.color.accent} borderStyle="round" flexDirection="column" paddingX={1}>
{screen === 'overview' && <OverviewScreen ctx={ctx} onClose={onClose} onPatch={onPatch} s={s} t={t} />}
{screen === 'buy' && <BuyScreen ctx={ctx} onClose={onClose} onPatch={onPatch} s={s} t={t} />}
{screen === 'confirm' && (
<ConfirmScreen
amount={overlay.pendingCharge?.amount ?? ''}
ctx={ctx}
onBack={() => onPatch({ pendingCharge: null, screen: 'buy' })}
onClose={onClose}
s={s}
t={t}
/>
)}
{screen === 'autoreload' && <AutoReloadScreen ctx={ctx} onClose={onClose} onPatch={onPatch} s={s} t={t} />}
{screen === 'limit' && <LimitScreen ctx={ctx} onClose={onClose} onPatch={onPatch} s={s} t={t} />}
</Box>
)
}
// ── Screen 1: Overview ────────────────────────────────────────────────
interface ScreenProps {
ctx: BillingOverlayState['ctx']
onClose: () => void
onPatch: (next: Partial<BillingOverlayState>) => void
s: BillingStateResponse
t: Theme
}
function OverviewScreen({ ctx, onClose, onPatch, s, t }: ScreenProps) {
// Gate: full menu only for an admin with the kill-switch on. Otherwise the
// menu collapses to Manage-on-portal / Cancel + a one-line note.
const full = s.is_admin && s.cli_billing_enabled
const note = !s.is_admin
? 'Billing actions need an org admin/owner.'
: !s.cli_billing_enabled
? 'Terminal billing is off for this org — enable it on the portal.'
: null
// Optimistic funnel: admin + kill-switch on but no saved card → a charge will
// 403 no_payment_method. Advise up front (Buy stays available — /state.card
// can't fully prove CLI-chargeability, so we hint rather than hide).
const cardHint = full && !s.card ? 'No saved card for terminal charges yet — set one up on the portal first.' : null
const items = full
? ['Buy credits', 'Adjust auto-reload', 'Adjust monthly limit', 'Manage on portal', 'Cancel']
: ['Manage on portal', 'Cancel']
const [sel, setSel] = useState(0)
const choose = (i: number) => {
if (full) {
if (i === 0) {
onPatch({ screen: 'buy' })
} else if (i === 1) {
onPatch({ screen: 'autoreload' })
} else if (i === 2) {
onPatch({ screen: 'limit' })
} else if (i === 3) {
if (s.portal_url) {
ctx.openPortal(s.portal_url)
}
onClose()
} else {
onClose()
}
} else {
if (i === 0 && s.portal_url) {
ctx.openPortal(s.portal_url)
}
onClose()
}
}
useInput((ch, key) => {
if (key.escape) {
return onClose()
}
if (key.upArrow && sel > 0) {
setSel(v => v - 1)
}
if (key.downArrow && sel < items.length - 1) {
setSel(v => v + 1)
}
if (key.return) {
return choose(sel)
}
const n = parseInt(ch, 10)
if (n >= 1 && n <= items.length) {
return choose(n - 1)
}
})
const bar = spendBar(s)
const auto = autoReloadLine(s)
return (
<Box flexDirection="column">
<Text bold color={t.color.accent}>
Usage credits
</Text>
{bar && <Text color={t.color.text}>{bar}</Text>}
<Text color={t.color.text}>Balance: {s.balance_display}</Text>
{auto && <Text color={t.color.muted}>{auto}</Text>}
{s.org_name && (
<Text color={t.color.muted}>
Org: {s.org_name}
{s.role ? ` · ${s.role}` : ''}
</Text>
)}
{note && (
<Box marginTop={1}>
<Text color={t.color.warn}>{note}</Text>
</Box>
)}
{cardHint && (
<Box marginTop={1}>
<Text color={t.color.warn}>{cardHint}</Text>
</Box>
)}
{cardHint && s.portal_url && <Text color={t.color.muted}>Portal: {s.portal_url}</Text>}
<Text />
{items.map((label, i) => (
<MenuRow active={sel === i} index={i + 1} key={label} label={label} t={t} />
))}
<Text />
{footer(`↑/↓ select · 1-${items.length} quick pick · Enter confirm · Esc close`, t)}
</Box>
)
}
// ── Screen 2: Buy credits ─────────────────────────────────────────────
function BuyScreen({ ctx, onPatch, s, t }: ScreenProps) {
const presets = s.charge_presets_display
const rawPresets = s.charge_presets
// rows: [...presets, 'Custom amount…', 'Cancel']
const rows = [...presets, 'Custom amount…', 'Cancel']
const customIdx = presets.length
const [sel, setSel] = useState(0)
const [typing, setTyping] = useState(false)
const [custom, setCustom] = useState('')
const [error, setError] = useState<null | string>(null)
const toConfirm = (amount: string) => {
onPatch({ pendingCharge: { amount }, screen: 'confirm' })
}
const pickPreset = (i: number) => {
// Prefer the raw (numeric) preset for the amount; fall back to stripping $.
const raw = (rawPresets[i] ?? presets[i] ?? '').replace(/^\$/, '').trim()
const v = ctx.validate(raw)
if (v.error || !v.amount) {
setError(v.error ?? 'Invalid preset.')
return
}
toConfirm(v.amount)
}
const submitCustom = (raw: string) => {
const v = ctx.validate(raw)
if (v.error || !v.amount) {
setError(v.error ?? 'Invalid amount.')
return
}
toConfirm(v.amount)
}
const choose = (i: number) => {
if (i < presets.length) {
pickPreset(i)
} else if (i === customIdx) {
setError(null)
setTyping(true)
} else {
onPatch({ screen: 'overview' })
}
}
useInput((ch, key) => {
if (key.escape) {
return typing ? (setTyping(false), setError(null)) : onPatch({ screen: 'overview' })
}
if (typing) {
return
}
if (key.upArrow && sel > 0) {
setSel(v => v - 1)
}
if (key.downArrow && sel < rows.length - 1) {
setSel(v => v + 1)
}
if (key.return) {
return choose(sel)
}
const n = parseInt(ch, 10)
if (n >= 1 && n <= rows.length) {
return choose(n - 1)
}
})
const payLine = s.card ? `Payment: ${s.card.masked}` : 'No saved card on file'
if (typing) {
return (
<Box flexDirection="column">
<Text bold color={t.color.accent}>
Buy usage credits
</Text>
<Text color={t.color.muted}>{payLine}</Text>
<Text />
<Text color={t.color.label}>Enter a custom amount:</Text>
<Box>
<Text color={t.color.label}>{'$'}</Text>
<TextInput columns={20} onChange={setCustom} onSubmit={submitCustom} value={custom} />
</Box>
{error && <Text color={t.color.error}>{error}</Text>}
<Text />
{footer('Enter confirm · Esc back', t)}
</Box>
)
}
return (
<Box flexDirection="column">
<Text bold color={t.color.accent}>
Buy usage credits
</Text>
<Text color={t.color.muted}>{payLine}</Text>
<Text />
{rows.map((label, i) => (
<MenuRow active={sel === i} index={i + 1} key={label} label={label} t={t} />
))}
{error && <Text color={t.color.error}>{error}</Text>}
<Text />
{footer(`↑/↓ select · 1-${rows.length} quick pick · Enter confirm · Esc back`, t)}
</Box>
)
}
// ── Screen 3: Confirm purchase ────────────────────────────────────────
function ConfirmScreen({
amount,
ctx,
onBack,
onClose,
s,
t
}: {
amount: string
ctx: BillingOverlayState['ctx']
onBack: () => void
onClose: () => void
s: BillingStateResponse
t: Theme
}) {
// rows: Pay $X now / Cancel
const [sel, setSel] = useState(0)
const pay = () => {
ctx.charge(amount)
// Settlement is reported via transcript lines; close the overlay now.
onClose()
}
const back = () => onBack()
useInput((ch, key) => {
if (key.escape) {
return back()
}
const lower = ch.toLowerCase()
if (lower === 'y') {
return pay()
}
if (lower === 'n') {
return back()
}
if (key.upArrow) {
setSel(0)
}
if (key.downArrow) {
setSel(1)
}
if (key.return) {
return sel === 0 ? pay() : back()
}
})
const payLine = s.card ? `Payment: ${s.card.masked}` : 'No saved card on file'
return (
<Box flexDirection="column">
<Text bold color={t.color.accent}>
Confirm purchase
</Text>
<Text color={t.color.text}>Total: ${amount}</Text>
<Text color={t.color.muted}>{payLine}</Text>
<Text color={t.color.muted}>By confirming, you allow Nous Research to charge your card.</Text>
<Text />
<ActionRow active={sel === 0} color={t.color.ok} label={`Pay $${amount} now`} t={t} />
<ActionRow active={sel === 1} label="Cancel" t={t} />
<Text />
{footer('↑/↓ select · Enter confirm · Y/N quick · Esc back', t)}
</Box>
)
}
// ── Screen 4: Auto-reload (the 2-field form) ──────────────────────────
function AutoReloadScreen({ ctx, onClose, onPatch, s, t }: ScreenProps) {
const ar = s.auto_reload
const enabled = Boolean(ar?.enabled)
// Prefill from state (strip the $ from the *_usd raw fields if present).
const prefill = (raw?: null | string) => (raw == null ? '' : String(raw).replace(/^\$/, '').trim())
const [threshold, setThreshold] = useState(prefill(ar?.threshold_usd))
const [reloadTo, setReloadTo] = useState(prefill(ar?.reload_to_usd))
const [field, setField] = useState<'reloadTo' | 'threshold'>('threshold')
const [error, setError] = useState<null | string>(null)
// focusRow: 0=threshold field, 1=reloadTo field, 2=Agree, 3=Turn off (if enabled), last=Cancel
const actionRows = enabled ? ['Agree and turn on', 'Turn off', 'Cancel'] : ['Agree and turn on', 'Cancel']
const FIELD_ROWS = 2
const [row, setRow] = useState(0)
const noCard = !s.card
const validatePair = (): null | { reloadTo: string; threshold: string } => {
const tv = ctx.validate(threshold)
if (tv.error || !tv.amount) {
setError(`Threshold: ${tv.error ?? 'invalid'}`)
return null
}
const rv = ctx.validate(reloadTo)
if (rv.error || !rv.amount) {
setError(`Reload-to: ${rv.error ?? 'invalid'}`)
return null
}
if (Number(rv.amount) <= Number(tv.amount)) {
setError('Reload-to amount must be greater than the threshold.')
return null
}
setError(null)
return { reloadTo: rv.amount, threshold: tv.amount }
}
const turnOn = () => {
if (noCard) {
ctx.sys('🔴 No saved card — set one up on the portal first.')
if (s.portal_url) {
ctx.openPortal(s.portal_url)
}
onClose()
return
}
const pair = validatePair()
if (!pair) {
return
}
void ctx.applyAutoReload(true, Number(pair.threshold), Number(pair.reloadTo)).then(ok => {
if (ok) {
ctx.sys(`✅ Auto-reload on: below $${pair.threshold} → reload to $${pair.reloadTo}.`)
}
})
onClose()
}
const turnOff = () => {
void ctx.applyAutoReload(false).then(ok => {
if (ok) {
ctx.sys('✅ Auto-reload turned off.')
}
})
onClose()
}
const onAction = (label: string) => {
if (label === 'Agree and turn on') {
turnOn()
} else if (label === 'Turn off') {
turnOff()
} else {
onPatch({ screen: 'overview' })
}
}
const editingField = row < FIELD_ROWS
useInput((ch, key) => {
if (key.escape) {
return onPatch({ screen: 'overview' })
}
if (key.upArrow && row > 0) {
setRow(v => v - 1)
setField(row - 1 === 0 ? 'threshold' : 'reloadTo')
}
if (key.downArrow && row < FIELD_ROWS + actionRows.length - 1) {
setRow(v => v + 1)
setField(row + 1 === 0 ? 'threshold' : 'reloadTo')
}
// Tab cycles between the two fields when focused on a field.
if (key.tab && editingField) {
const next = field === 'threshold' ? 'reloadTo' : 'threshold'
setField(next)
setRow(next === 'threshold' ? 0 : 1)
}
if (key.return && !editingField) {
const idx = row - FIELD_ROWS
return onAction(actionRows[idx] ?? 'Cancel')
}
// a number quick-picks an action row (1..actionRows.length)
if (!editingField) {
const n = parseInt(ch, 10)
if (n >= 1 && n <= actionRows.length) {
return onAction(actionRows[n - 1]!)
}
}
})
const cardLine = s.card ? `Card on file: ${s.card.masked}` : 'No saved card on file'
const fieldBox = (label: string, value: string, onChange: (v: string) => void, focused: boolean, key: string) => (
<Box flexDirection="column" key={key}>
<Text color={focused ? t.color.label : t.color.muted}>{label}</Text>
<Box borderColor={focused ? t.color.accent : t.color.border} borderStyle="round" paddingX={1}>
<Text color={t.color.label}>{'$'}</Text>
<TextInput
columns={16}
focus={focused}
onChange={onChange}
onSubmit={() => {
// Enter inside the threshold field jumps to reload-to; inside
// reload-to jumps to the Agree action.
if (key === 'threshold') {
setField('reloadTo')
setRow(1)
} else {
setRow(FIELD_ROWS)
}
}}
value={value}
/>
</Box>
</Box>
)
return (
<Box flexDirection="column">
<Text bold color={t.color.accent}>
Auto-reload
</Text>
<Text color={t.color.muted}>Automatically buy more credits when your balance is low.</Text>
<Text color={t.color.muted}>{cardLine}</Text>
<Text />
{fieldBox('When balance falls below:', threshold, setThreshold, row === 0, 'threshold')}
{fieldBox('Reload balance to:', reloadTo, setReloadTo, row === 1, 'reloadTo')}
<Text />
<Text color={t.color.muted}>
By confirming, you authorize Nous Research to charge {s.card ? s.card.masked : 'your card'} whenever your
balance falls below the threshold. Turn off any time here or on the portal.
</Text>
{error && <Text color={t.color.error}>{error}</Text>}
<Text />
{actionRows.map((label, i) => (
<ActionRow
active={!editingField && row - FIELD_ROWS === i}
color={label === 'Turn off' ? t.color.warn : label === 'Agree and turn on' ? t.color.ok : t.color.text}
key={label}
label={label}
t={t}
/>
))}
<Text />
{footer('↑/↓ move · Tab switch field · Enter next/confirm · Esc back', t)}
</Box>
)
}
// ── Screen 5: Monthly spend limit (read-only) ─────────────────────────
function LimitScreen({ ctx, onClose, onPatch, s, t }: ScreenProps) {
const rows = ['Manage on portal', 'Cancel']
const [sel, setSel] = useState(0)
const choose = (i: number) => {
if (i === 0 && s.portal_url) {
ctx.openPortal(s.portal_url)
return onClose()
}
onPatch({ screen: 'overview' })
}
useInput((ch, key) => {
if (key.escape) {
return onPatch({ screen: 'overview' })
}
if (key.upArrow && sel > 0) {
setSel(v => v - 1)
}
if (key.downArrow && sel < rows.length - 1) {
setSel(v => v + 1)
}
if (key.return) {
return choose(sel)
}
const n = parseInt(ch, 10)
if (n >= 1 && n <= rows.length) {
return choose(n - 1)
}
})
const cap = s.monthly_cap
const usageLine =
cap && cap.limit_usd != null
? `${cap.spent_display} of ${cap.limit_display} used this month${cap.is_default_ceiling ? ' (default ceiling)' : ''}`
: 'No monthly cap visible (managed on the portal).'
return (
<Box flexDirection="column">
<Text bold color={t.color.accent}>
Monthly spend limit
</Text>
<Text color={t.color.text}>{usageLine}</Text>
<Text color={t.color.muted}>The monthly limit is set on the portal shown here read-only.</Text>
<Text />
{rows.map((label, i) => (
<MenuRow active={sel === i} index={i + 1} key={label} label={label} t={t} />
))}
<Text />
{footer(`↑/↓ select · 1-${rows.length} quick pick · Enter confirm · Esc back`, t)}
</Box>
)
}

View file

@ -53,6 +53,95 @@ export interface CreditsViewResponse {
topup_url: string | null
}
// ── Terminal billing (Phase 2b) ──────────────────────────────────────
export interface BillingCardInfo {
brand: string
last4: string
masked: string
}
export interface BillingMonthlyCap {
is_default_ceiling: boolean
limit_display: string
limit_usd: string | null
spent_display: string
spent_this_month_usd: string | null
}
export interface BillingAutoReload {
enabled: boolean
reload_to_display: string
reload_to_usd: string | null
threshold_display: string
threshold_usd: string | null
}
export interface BillingStateResponse {
auto_reload: BillingAutoReload | null
balance_display: string
balance_usd: string | null
can_charge: boolean
card: BillingCardInfo | null
charge_presets: string[]
charge_presets_display: string[]
cli_billing_enabled: boolean
error?: string | null
is_admin: boolean
logged_in: boolean
max_usd: string | null
min_usd: string | null
monthly_cap: BillingMonthlyCap | null
ok: boolean
org_name: string | null
portal_url: string | null
role: string | null
}
/**
* Raw error payload echoed from the server (`_serialize_billing_error`). Carries
* the extra fields a few error codes attach notably `remainingUsd` on
* `monthly_cap_exceeded` so the client can render the same detail the CLI does.
*/
export interface BillingErrorPayload {
isDefaultCeiling?: boolean
remainingUsd?: string
}
export interface BillingChargeResponse {
charge_id?: string
error?: string
idempotency_key?: string
message?: string
ok: boolean
payload?: BillingErrorPayload
portal_url?: string | null
retry_after?: number | null
}
export interface BillingChargeStatusResponse {
amount_usd?: string | null
error?: string
message?: string
ok: boolean
payload?: BillingErrorPayload
portal_url?: string | null
reason?: string | null
retry_after?: number | null
settled_at?: string | null
status?: string
}
export interface BillingMutationResponse {
error?: string
granted?: boolean
message?: string
ok: boolean
payload?: BillingErrorPayload
portal_url?: string | null
retry_after?: number | null
}
export type CommandDispatchResponse =
| { output?: string; type: 'exec' | 'plugin' }
| { target: string; type: 'alias' }
@ -538,6 +627,11 @@ export type GatewayEvent =
type: 'notification.show'
}
| { payload?: { key?: string }; session_id?: string; type: 'notification.clear' }
| {
payload: { user_code?: string; verification_url: string }
session_id?: string
type: 'billing.step_up.verification'
}
| { payload?: { state?: 'idle' | 'listening' | 'transcribing' }; session_id?: string; type: 'voice.status' }
| { payload?: { no_speech_limit?: boolean; text?: string }; session_id?: string; type: 'voice.transcript' }
| { payload: { line: string }; session_id?: string; type: 'gateway.stderr' }