fix(picker): keep max_models=0 distinct from unlimited; lock cap semantics

Follow-up to the cap-removal salvage. The contributor guarded the new
unlimited default with `[:max_models] if max_models else ...`, which conflates
max_models=0 (used by slug-only callers that want an empty model list) with
None (unlimited). Tighten to `is not None` at all five slicing sites in
list_authenticated_providers / list_picker_providers, and add a regression test
asserting the three-way contract: None=full, 0=empty, N=first N.

agent/billing_view.py

@@ -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)

agent/conversation_compression.py

@@ -512,6 +512,16 @@ def compress_context(
             old_title = agent._session_db.get_session_title(agent.session_id)
             # Trigger memory extraction on the old session before it rotates.
             agent.commit_memory_session(messages)
+            # Flush any un-persisted messages from the current turn to the
+            # old session *before* rotating.  compress_context() can be
+            # called mid-turn (auto-compress when context exceeds threshold)
+            # at a point when _flush_messages_to_session_db() has not yet
+            # run.  Without this, messages generated during the current turn
+            # are silently lost on session rotation (#47202).
+            try:
+                agent._flush_messages_to_session_db(messages)
+            except Exception:
+                pass  # best-effort — don't block compression on a flush error
             agent._session_db.end_session(agent.session_id, "compression")
             old_session_id = agent.session_id
             agent.session_id = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}"

agent/prompt_builder.py

@@ -320,9 +320,11 @@ TASK_COMPLETION_GUIDANCE = (
 # concurrently when they are independent (read-only tools always; path-scoped
 # file ops when their targets don't overlap — see
 # run_agent._execute_tool_calls / tool_dispatch_helpers). The missing piece
-# was telling the *model* to emit those calls together in the first place;
-# nothing in the open-source system prompt encouraged batching. This block
-# closes that gap.
+# was telling the *model* to emit those calls together in the first place.
+# Until now the only batching steer in the prompt lived in
+# GOOGLE_MODEL_OPERATIONAL_GUIDANCE — Gemini/Gemma got it, every other model
+# got nothing. This block makes the steer universal; the now-redundant
+# Google-only bullet has been dropped so no model receives it twice.
 #
 # Short on purpose — shipped in the cached system prompt to every user, every
 # session. Token cost is paid once at install and amortised across all
@@ -425,9 +427,10 @@ GOOGLE_MODEL_OPERATIONAL_GUIDANCE = (
     "package.json, requirements.txt, Cargo.toml, etc. before importing.\n"
     "- **Conciseness:** Keep explanatory text brief — a few sentences, not "
     "paragraphs. Focus on actions and results over narration.\n"
-    "- **Parallel tool calls:** When you need to perform multiple independent "
-    "operations (e.g. reading several files), make all the tool calls in a "
-    "single response rather than sequentially.\n"
+    # Parallel-tool-call steering now lives in the universal
+    # PARALLEL_TOOL_CALL_GUIDANCE block (injected for all models), so it is no
+    # longer duplicated here — keeping it would send Gemini/Gemma the same
+    # instruction twice.
     "- **Non-interactive commands:** Use flags like -y, --yes, --non-interactive "
     "to prevent CLI tools from hanging on prompts.\n"
     "- **Keep going:** Work autonomously until the task is fully resolved. "

apps/desktop/src/app/session/hooks/use-message-stream.ts

@@ -13,6 +13,7 @@ import {
   type GatewayEventPayload,
   reasoningPart,
   renderMediaTags,
+  textPart,
   upsertToolPart
 } from '@/lib/chat-messages'
 import { coerceGatewayText, coerceThinkingText, normalizePersonalityValue } from '@/lib/chat-runtime'
@@ -1080,6 +1081,32 @@ export function useMessageStream({
           // completions / watch matches here — re-sync the status stack.
           void refreshBackgroundProcesses(sessionId)
         }
+      } else if (event.type === 'review.summary') {
+        // Self-improvement background review saved something to memory/skills
+        // and emitted a persistent summary (Python formats it as
+        // "💾 Self-improvement review: …"). The CLI prints this via
+        // prompt_toolkit and the Ink TUI renders it as a system line; the
+        // desktop has neither, so without this handler the skill/memory
+        // change happens silently. Surface it as a persistent system message
+        // in the transcript so the user is always informed — it must not be a
+        // transient toast that can be missed.
+        const text = coerceGatewayText(payload?.text).trim()
+
+        if (text && sessionId) {
+          flushQueuedDeltas(sessionId)
+          updateSessionState(sessionId, state => ({
+            ...state,
+            messages: [
+              ...state.messages,
+              {
+                id: `review-summary-${Date.now()}`,
+                role: 'system',
+                parts: [textPart(text)],
+                timestamp: Math.floor(Date.now() / 1000)
+              }
+            ]
+          }))
+        }
       } else if (event.type === 'error') {
         const errorMessage = payload?.message || 'Hermes reported an error'
         const looksLikeProviderSetup = isProviderSetupErrorMessage(errorMessage)

apps/desktop/src/components/assistant-ui/thread.tsx

@@ -827,7 +827,7 @@ function StickyHumanMessageContainer({ attachments, children }: { attachments?:
 // so without the carve-out, clicking a stuck bubble drags the window instead of
 // opening the edit composer.
 const USER_BUBBLE_BASE_CLASS =
-  'composer-human-message standalone-glass relative flex w-full min-w-0 max-w-full flex-col gap-1.5 overflow-hidden rounded-xl border bg-(--dt-user-bubble) px-3 py-2 text-left [-webkit-app-region:no-drag]'
+  'composer-human-message standalone-glass relative flex w-full min-w-0 max-w-full flex-col gap-1.5 overflow-y-auto rounded-xl border bg-(--dt-user-bubble) px-3 py-2 text-left [-webkit-app-region:no-drag]'
 
 const USER_ACTION_ICON_BUTTON_CLASS =
   'grid place-items-center rounded-md bg-transparent text-(--ui-text-secondary) transition-colors hover:bg-(--ui-control-active-background) hover:text-foreground disabled:cursor-default disabled:text-(--ui-text-quaternary) disabled:opacity-70'

cli.py

@@ -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()
 
@@ -5957,6 +5975,18 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
 
         old_session_id = self.session_id
         if self._session_db and old_session_id:
+            # Flush any un-persisted messages from the current turn to the
+            # old session *before* rotating.  /new can be called mid-turn
+            # when _flush_messages_to_session_db() has not yet run — without
+            # this, messages generated during the current turn are silently
+            # lost on session rotation (#47202).
+            if self.agent:
+                try:
+                    self.agent._flush_messages_to_session_db(
+                        self.conversation_history
+                    )
+                except Exception:
+                    pass  # best-effort
             try:
                 self._session_db.end_session(old_session_id, "new_session")
             except Exception:
@@ -6359,6 +6389,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
@@ -6930,7 +6971,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
             try:
                 if ctx is None:
                     raise RuntimeError("inventory context unavailable")
-                providers = build_models_payload(ctx, max_models=50)["providers"]
+                providers = build_models_payload(ctx)["providers"]
             except Exception:
                 providers = []
 
@@ -7137,53 +7178,6 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
         except Exception:
             return False
 
-    def _should_handle_notify_command_inline(self, text: str, has_images: bool = False) -> bool:
-        """Return True when /notify should be dispatched mid-task.
-
-        Same pattern as /steer: write the sentinel file without queuing
-        through _pending_input (which would miss the mid-run window).
-        """
-        if not text or has_images or not _looks_like_slash_command(text):
-            return False
-        if not getattr(self, "_agent_running", False):
-            return False
-        try:
-            from hermes_cli.commands import resolve_command
-            base = text.split(None, 1)[0].lower().lstrip('/')
-            cmd = resolve_command(base)
-            return bool(cmd and cmd.name == "notify")
-        except Exception:
-            return False
-
-    def _handle_notify_command(self, cmd_original: str):
-        """Handle /notify [prompt | cancel].
-
-        - /notify <prompt>  — set flag + submit prompt
-        - /notify           — set flag only (mid-task or pre-turn)
-        - /notify cancel    — clear pending notification
-        """
-        from tools.notify_utils import set_notify_flag, clear_notify_flag
-
-        parts = cmd_original.split(None, 1)
-        sub = parts[1].strip() if len(parts) > 1 else ""
-
-        if sub.lower() == "cancel":
-            if clear_notify_flag():
-                _cprint("  🔕 Notification cancelled")
-            else:
-                _cprint("  No pending notification")
-            return
-
-        set_notify_flag()
-
-        if sub:
-            # Has a prompt — set flag AND submit
-            self._pending_input.put(sub)
-            _cprint(f"  🔔 Will notify when done: "
-                    f"{sub[:80]}{'...' if len(sub) > 80 else ''}")
-        else:
-            _cprint("  🔔 Will notify when this turn finishes")
-
     def _output_console(self):
         """Use prompt_toolkit-safe Rich rendering once the TUI is live."""
         if getattr(self, "_app", None):
@@ -7553,6 +7547,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":
@@ -7672,8 +7668,6 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
                     _cprint(f"  Queued for the next turn: {payload[:80]}{'...' if len(payload) > 80 else ''}")
                 else:
                     _cprint(f"  Queued: {payload[:80]}{'...' if len(payload) > 80 else ''}")
-        elif canonical == "notify":
-            self._handle_notify_command(cmd_original)
         elif canonical == "steer":
             # Inject a message after the next tool call without interrupting.
             # If the agent is actively running, push the text into the agent's
@@ -8474,7 +8468,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
 
@@ -8536,6 +8530,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
@@ -10438,15 +11054,6 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
             # Flush any remaining streamed text and close the box
             self._flush_stream()
 
-            # Notify check — fire if /notify was set for this turn.
-            # Must run BEFORE goal continuation so the notification
-            # reflects actual turn completion (mirrors the TUI path).
-            try:
-                from tools.notify_utils import consume_pending_notification
-                consume_pending_notification()
-            except Exception as e:
-                logging.debug("notify idle-check failed: %s", e)
-
             # Signal end-of-text to TTS consumer and wait for it to finish
             if use_streaming_tts and text_queue is not None:
                 text_queue.put(None)  # sentinel
@@ -11337,14 +11944,6 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
                     event.app.invalidate()
                     return
 
-                # Handle /notify while the agent is running — same pattern
-                # as /steer: write the sentinel file on the UI thread so it
-                # survives mid-run without queueing through _pending_input.
-                if self._should_handle_notify_command_inline(text, has_images=has_images):
-                    self.process_command(text)
-                    event.app.current_buffer.reset(append_to_history=True)
-                    return
-
                 # Snapshot and clear attached images
                 images = list(self._attached_images)
                 self._attached_images.clear()
@@ -13120,16 +13719,6 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
                         self._pending_tool_info.clear()
                         self._last_scrollback_tool = ""
 
-                        # Notify check — fire if /notify was set during this turn.
-                        # Must run BEFORE goal continuation so the notification
-                        # reflects the actual turn completion, not a
-                        # potentially-auto-continued turn.
-                        try:
-                            from tools.notify_utils import consume_pending_notification
-                            consume_pending_notification()
-                        except Exception as e:
-                            logging.debug("notify idle-check failed: %s", e)
-
                         app.invalidate()  # Refresh status line
 
                         # Goal continuation: if a standing goal is active, ask

gateway/slash_commands.py

@@ -2588,14 +2588,29 @@ class GatewaySlashCommandsMixin:
                 # session_id for the continuation.  Write the compressed messages
                 # into the NEW session so the original history stays searchable.
                 new_session_id = tmp_agent.session_id
-                if new_session_id != session_entry.session_id:
+                rotated = new_session_id != session_entry.session_id
+                if rotated:
                     session_entry.session_id = new_session_id
                     self.session_store._save()
                     self._sync_telegram_topic_binding(
                         source, session_entry, reason="compress-command",
                     )
 
+                # Only rewrite the transcript when rotation actually produced a
+                # NEW session id. If _compress_context could not rotate (e.g.
+                # _session_db unavailable, or the DB split raised), session_id
+                # is unchanged and rewrite_transcript() would DELETE the
+                # original messages and replace them with only the compressed
+                # summary — permanent data loss (#44794, #39704). In that case
+                # leave the original transcript intact.
+                if rotated:
                     self.session_store.rewrite_transcript(new_session_id, compressed)
+                else:
+                    logger.warning(
+                        "Manual /compress: session rotation did not occur "
+                        "(session_id unchanged) — preserving original transcript "
+                        "instead of overwriting it (#44794)."
+                    )
                 # Reset stored token count — transcript changed, old value is stale
                 self.session_store.update_session(
                     session_entry.session_key, last_prompt_tokens=0

hermes_cli/auth.py

@@ -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

hermes_cli/commands.py

@@ -103,8 +103,6 @@ COMMAND_REGISTRY: list[CommandDef] = [
                aliases=("tasks",)),
     CommandDef("queue", "Queue a prompt for the next turn (doesn't interrupt)", "Session",
                aliases=("q",), args_hint="<prompt>"),
-    CommandDef("notify", "Set a desktop notification when Hermes finishes this turn", "Session",
-               args_hint="[prompt | cancel]", cli_only=True),
     CommandDef("steer", "Inject a message after the next tool call without interrupting", "Session",
                args_hint="<prompt>"),
     CommandDef("goal", "Set a standing goal Hermes works on across turns until achieved", "Session",
@@ -217,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",
@@ -1055,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:

hermes_cli/inventory.py

@@ -117,7 +117,7 @@ def build_models_payload(
     pricing: bool = False,
     capabilities: bool = False,
     force_fresh_nous_tier: bool = False,
-    max_models: int = 50,
+    max_models: int | None = None,
 ) -> dict:
     """Build the ``{providers, model, provider}`` shape every consumer
     needs from a single substrate call.

hermes_cli/model_switch.py

@@ -1188,7 +1188,6 @@ def prewarm_picker_cache_async() -> Optional["_threading.Thread"]:
                 current_model=ctx.current_model,
                 user_providers=ctx.user_providers,
                 custom_providers=ctx.custom_providers,
-                max_models=50,
             )
         except Exception:
             # Best-effort warmup — never surface errors into the session.
@@ -1206,7 +1205,7 @@ def list_authenticated_providers(
     custom_providers: list | None = None,
     *,
     force_fresh_nous_tier: bool = False,
-    max_models: int = 8,
+    max_models: int | None = None,
     current_model: str = "",
 ) -> List[dict]:
     """Detect which providers have credentials and list their curated models.
@@ -1426,7 +1425,7 @@ def list_authenticated_providers(
             if hermes_id in _MODELS_DEV_PREFERRED:
                 model_ids = _merge_with_models_dev(hermes_id, model_ids)
         total = len(model_ids)
-        top = model_ids[:max_models]
+        top = model_ids[:max_models] if max_models is not None else model_ids
 
         slug = hermes_id
         pinfo = _mdev_pinfo(mdev_id)
@@ -1589,7 +1588,7 @@ def list_authenticated_providers(
                 if hermes_slug in _MODELS_DEV_PREFERRED:
                     model_ids = _merge_with_models_dev(hermes_slug, model_ids)
         total = len(model_ids)
-        top = model_ids[:max_models]
+        top = model_ids[:max_models] if max_models is not None else model_ids
 
         results.append({
             "slug": hermes_slug,
@@ -1664,7 +1663,7 @@ def list_authenticated_providers(
             if not _cp_model_ids:
                 _cp_model_ids = curated.get(_cp.slug, [])
         _cp_total = len(_cp_model_ids)
-        _cp_top = _cp_model_ids[:max_models]
+        _cp_top = _cp_model_ids[:max_models] if max_models is not None else _cp_model_ids
 
         results.append({
             "slug": _cp.slug,
@@ -1813,7 +1812,7 @@ def list_authenticated_providers(
             "name": "Custom endpoint",
             "is_current": True,
             "is_user_defined": True,
-            "models": _models[:max_models] if max_models else _models,
+            "models": _models[:max_models] if max_models is not None else _models,
             "total_models": len(_models),
             "source": "model-config",
             "api_url": str(current_base_url).strip().rstrip("/"),
@@ -2040,7 +2039,7 @@ def list_picker_providers(
     current_base_url: str = "",
     user_providers: dict = None,
     custom_providers: list | None = None,
-    max_models: int = 8,
+    max_models: int | None = None,
     current_model: str = "",
 ) -> List[dict]:
     """Interactive-picker variant of :func:`list_authenticated_providers`.
@@ -2083,7 +2082,7 @@ def list_picker_providers(
             except Exception:
                 live_ids = list(p.get("models", []))
             p = dict(p)
-            p["models"] = live_ids[:max_models]
+            p["models"] = live_ids[:max_models] if max_models is not None else live_ids
             p["total_models"] = len(live_ids)
 
         has_models = bool(p.get("models"))

hermes_cli/nous_billing.py

@@ -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)

hermes_cli/web_server.py

@@ -3323,7 +3323,6 @@ def get_model_options(profile: Optional[str] = None):
         with _profile_scope(profile):
             return build_models_payload(
                 load_picker_context(),
-                max_models=50,
                 include_unconfigured=True,
                 picker_hints=True,
                 canonical_order=True,
@@ -3398,7 +3397,7 @@ def get_recommended_default_model(provider: str = ""):
     try:
         from hermes_cli.inventory import build_models_payload, load_picker_context
 
-        payload = build_models_payload(load_picker_context(), max_models=50)
+        payload = build_models_payload(load_picker_context())
         for row in payload.get("providers", []):
             if str(row.get("slug", "")).lower() == slug:
                 models = row.get("models") or []

tests/agent/test_billing_view.py

@@ -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()

tests/agent/test_prompt_builder.py

@@ -28,6 +28,7 @@ from agent.prompt_builder import (
     TOOL_USE_ENFORCEMENT_MODELS,
     OPENAI_MODEL_EXECUTION_GUIDANCE,
     PARALLEL_TOOL_CALL_GUIDANCE,
+    GOOGLE_MODEL_OPERATIONAL_GUIDANCE,
     MEMORY_GUIDANCE,
     SESSION_SEARCH_GUIDANCE,
     PLATFORM_HINTS,
@@ -1512,8 +1513,9 @@ class TestParallelToolCallGuidance:
 
     def test_steers_batching_into_one_response(self):
         text = PARALLEL_TOOL_CALL_GUIDANCE.lower()
-        # Must tell the model to group independent calls together.
-        assert "single response" in text or "same" in text and "turn" in text
+        # Must tell the model to group independent calls together — accept any
+        # phrasing that means "one turn" without freezing exact wording.
+        assert "single response" in text or ("same" in text and "turn" in text)
         assert "independent" in text
 
     def test_carves_out_dependent_calls(self):
@@ -1533,6 +1535,11 @@ class TestParallelToolCallGuidance:
         # Heading delimits it as its own section in the assembled prompt.
         assert PARALLEL_TOOL_CALL_GUIDANCE.lstrip().startswith("#")
 
+    def test_not_duplicated_in_google_guidance(self):
+        # The universal block is now the single source of parallel-batching
+        # steer. The Google-only block must NOT carry its own copy, otherwise
+        # Gemini/Gemma would receive the instruction twice in one prompt.
+        assert "parallel tool call" not in GOOGLE_MODEL_OPERATIONAL_GUIDANCE.lower()
 
 
 # =========================================================================

tests/cli/test_prompt_text_input_thread_safety.py

@@ -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."""

tests/hermes_cli/test_billing_cli.py

@@ -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

tests/hermes_cli/test_billing_portal_url.py

@@ -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"
+    )

tests/hermes_cli/test_billing_scope_stepup.py

@@ -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"

tests/hermes_cli/test_inventory.py

@@ -660,3 +660,31 @@ def test_two_custom_providers_with_overlap_both_survive():
     assert a_row["total_models"] == 2
     assert b_row["total_models"] == 2
 
+
+def test_build_models_payload_no_max_models_returns_full_list():
+    """When max_models is not passed (None), build_models_payload must
+    return the full model list — not truncate to the old default of 50.
+    Regression for #48279: Kilo Gateway picker was capped at 50 of 336
+    models, making most models undiscoverable via search."""
+    full_models = [f"model-{i}" for i in range(100)]
+    rows = [
+        {
+            "slug": "kilocode",
+            "name": "Kilo Code",
+            "models": full_models,
+            "total_models": len(full_models),
+            "is_current": False,
+            "is_user_defined": False,
+            "source": "built-in",
+        },
+    ]
+    ctx = _empty_ctx()
+    with _list_auth_returning(rows):
+        # No max_models argument — should return all 100 models
+        payload = build_models_payload(ctx)
+
+    kilo_row = next(r for r in payload["providers"] if r["slug"] == "kilocode")
+    assert kilo_row["models"] == full_models
+    assert kilo_row["total_models"] == 100
+    assert len(kilo_row["models"]) == 100
+

tests/hermes_cli/test_model_catalog.py

@@ -423,6 +423,71 @@ class TestIntegrationWithModelsModule:
         assert nous_row is not None, "nous row must appear when authed"
         assert nous_row["models"] == expected
 
+    def test_picker_max_models_cap_semantics(self, tmp_path, monkeypatch):
+        """The cap argument has three distinct meanings on the real slicing
+        path: ``None`` = unlimited (the cap-removal fix, #48297), ``0`` = no
+        models (preserved for slug-only callers), an int N = first N. Guards
+        the ``is not None`` distinction the cap-removal follow-up introduced —
+        a ``if max_models`` (falsy) check would conflate ``0`` with unlimited.
+        """
+        import importlib
+        from hermes_cli import model_catalog
+        from hermes_cli.models import get_curated_nous_model_ids
+        importlib.reload(model_catalog)
+        try:
+            from hermes_cli.model_switch import (
+                list_authenticated_providers,
+                list_picker_providers,
+            )
+
+            active_home = Path(os.environ["HERMES_HOME"])
+            (active_home / "auth.json").write_text(
+                json.dumps(
+                    {
+                        "providers": {"nous": {"access_token": "fake"}},
+                        "credential_pool": {},
+                    }
+                )
+            )
+            with patch.object(
+                model_catalog, "_fetch_manifest", return_value=_valid_manifest()
+            ), patch("hermes_cli.models.check_nous_free_tier", return_value=False), patch(
+                "hermes_cli.models.union_with_portal_free_recommendations",
+                side_effect=lambda ids, *a, **k: (ids, {}),
+            ), patch(
+                "hermes_cli.models.union_with_portal_paid_recommendations",
+                side_effect=lambda ids, *a, **k: (ids, {}),
+            ):
+                expected = get_curated_nous_model_ids()
+                full = list_picker_providers(current_provider="nous", max_models=None)
+                one = list_picker_providers(current_provider="nous", max_models=1)
+                # 0 is exercised on list_authenticated_providers (the slug-only
+                # path); the picker variant drops empty-model rows entirely, so
+                # the empty-list contract lives on the auth-providers call.
+                zero = list_authenticated_providers(
+                    current_provider="nous", max_models=0
+                )
+        finally:
+            model_catalog.reset_cache()
+
+        def _nous(rows):
+            return next((r for r in rows if r["slug"] == "nous"), None)
+
+        # Only meaningful when the curated list actually exceeds 1 entry.
+        assert len(expected) > 1, "test needs a multi-model curated nous list"
+
+        full_row = _nous(full)
+        assert full_row is not None and full_row["models"] == expected
+
+        one_row = _nous(one)
+        assert one_row is not None and one_row["models"] == expected[:1]
+
+        zero_row = _nous(zero)
+        # 0 means an empty model list — NOT unlimited. total_models still real.
+        assert zero_row is not None
+        assert zero_row["models"] == []
+        assert zero_row["total_models"] == len(expected)
+
 
 # -----------------------------------------------------------------------------
 # Drift guard — prevent the in-repo curated lists from going out of sync with

tests/test_notify_approval.py

@@ -1,140 +0,0 @@
-"""Tests for /notify approval-request notifications."""
-
-
-def test_fire_approval_request_notification_uses_input_needed_message(monkeypatch):
-    from tools import notify_utils
-
-    calls = []
-    monkeypatch.setattr(
-        notify_utils,
-        "fire_notification",
-        lambda *, title="Hermes Agent", message="Task complete", config=None: calls.append(
-            {"title": title, "message": message, "config": config}
-        ),
-    )
-
-    notify_utils.fire_approval_request_notification()
-
-    assert calls == [
-        {"title": "Hermes Agent", "message": "Input needed: approval required", "config": None}
-    ]
-
-
-def test_fire_approval_request_notification_does_not_clear_pending_notify(monkeypatch, tmp_path):
-    from tools import notify_utils
-
-    monkeypatch.setattr(notify_utils, "_hermes_home", lambda: tmp_path)
-    monkeypatch.setattr(notify_utils, "fire_notification", lambda **kwargs: None)
-    notify_utils.set_notify_flag()
-
-    notify_utils.fire_approval_request_notification()
-
-    assert notify_utils.is_notify_pending() is True
-
-
-def test_approval_request_notification_skips_messaging_gateway_platform(monkeypatch):
-    from gateway import session_context
-    from tools import approval
-    from tools import notify_utils
-
-    calls = []
-    monkeypatch.setattr(
-        session_context,
-        "get_session_env",
-        lambda name, default="": "telegram" if name == "HERMES_SESSION_PLATFORM" else default,
-    )
-    monkeypatch.setattr(notify_utils, "is_notify_pending", lambda: True)
-    monkeypatch.setattr(notify_utils, "fire_approval_request_notification", lambda: calls.append("approval"))
-
-    approval._notify_approval_request_if_pending()
-
-    assert calls == []
-
-
-def test_check_all_command_guards_notifies_when_cli_approval_requested(monkeypatch):
-    from tools import approval
-    from tools import notify_utils
-
-    calls = []
-    monkeypatch.setenv("HERMES_INTERACTIVE", "1")
-    monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
-    monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
-    monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
-    monkeypatch.setattr(approval, "_get_approval_mode", lambda: "manual")
-    monkeypatch.setattr(approval, "detect_hardline_command", lambda command: (False, None))
-    monkeypatch.setattr(
-        approval,
-        "detect_dangerous_command",
-        lambda command: (True, "dangerous:test", "test approval"),
-    )
-    monkeypatch.setattr(approval, "is_approved", lambda session_key, pattern_key: False)
-    monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *args, **kwargs: "deny")
-    monkeypatch.setattr(notify_utils, "is_notify_pending", lambda: True)
-    monkeypatch.setattr(notify_utils, "fire_approval_request_notification", lambda: calls.append("approval"))
-
-    result = approval.check_all_command_guards("rm -rf /tmp/demo", "local")
-
-    assert result["approved"] is False
-    assert calls == ["approval"]
-
-
-def test_gateway_approval_prompt_is_emitted_before_desktop_notification(monkeypatch):
-    from tools import approval
-    from tools import notify_utils
-
-    calls = []
-    session_key = "notify-order-session"
-
-    def notify_cb(_approval_data):
-        calls.append("approval-prompt")
-        approval.resolve_gateway_approval(session_key, "deny")
-
-    monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1")
-    monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
-    monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
-    monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
-    monkeypatch.setattr(approval, "get_current_session_key", lambda default="default": session_key)
-    monkeypatch.setattr(approval, "_get_approval_mode", lambda: "manual")
-    monkeypatch.setattr(approval, "_get_approval_config", lambda: {"gateway_timeout": 1})
-    monkeypatch.setattr(approval, "detect_hardline_command", lambda command: (False, None))
-    monkeypatch.setattr(
-        approval,
-        "detect_dangerous_command",
-        lambda command: (True, "dangerous:test", "test approval"),
-    )
-    monkeypatch.setattr(approval, "is_approved", lambda session_key, pattern_key: False)
-    monkeypatch.setattr(notify_utils, "is_notify_pending", lambda: True)
-    monkeypatch.setattr(notify_utils, "fire_approval_request_notification", lambda: calls.append("desktop-notify"))
-    approval.register_gateway_notify(session_key, notify_cb)
-
-    result = approval.check_all_command_guards("rm -rf /tmp/demo", "local")
-
-    assert result["approved"] is False
-    assert calls == ["approval-prompt", "desktop-notify"]
-
-
-def test_check_all_command_guards_skips_approval_notification_without_notify_pending(monkeypatch):
-    from tools import approval
-    from tools import notify_utils
-
-    calls = []
-    monkeypatch.setenv("HERMES_INTERACTIVE", "1")
-    monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
-    monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
-    monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
-    monkeypatch.setattr(approval, "_get_approval_mode", lambda: "manual")
-    monkeypatch.setattr(approval, "detect_hardline_command", lambda command: (False, None))
-    monkeypatch.setattr(
-        approval,
-        "detect_dangerous_command",
-        lambda command: (True, "dangerous:test", "test approval"),
-    )
-    monkeypatch.setattr(approval, "is_approved", lambda session_key, pattern_key: False)
-    monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *args, **kwargs: "deny")
-    monkeypatch.setattr(notify_utils, "is_notify_pending", lambda: False)
-    monkeypatch.setattr(notify_utils, "fire_approval_request_notification", lambda: calls.append("approval"))
-
-    result = approval.check_all_command_guards("rm -rf /tmp/demo", "local")
-
-    assert result["approved"] is False
-    assert calls == []

tests/tools/test_notify_utils.py

@@ -1,201 +0,0 @@
-"""Tests for the per-session /notify sentinel + consume helper.
-
-Covers the behavior the supersede added on top of the original PR:
-the sentinel is scoped per session so a /notify in one TUI/dashboard
-session never fires on another session's turn completion.
-"""
-
-import pytest
-
-from tools import notify_utils
-
-
-@pytest.fixture
-def home(tmp_path, monkeypatch):
-    monkeypatch.setattr(notify_utils, "_hermes_home", lambda: tmp_path)
-    # Never hit the OS during tests.
-    monkeypatch.setattr(notify_utils, "_show_desktop_notification", lambda *a, **k: None)
-    return tmp_path
-
-
-def test_set_is_pending_clear_roundtrip(home):
-    assert notify_utils.is_notify_pending("sess-a") is False
-    assert notify_utils.set_notify_flag("sess-a") is True
-    assert notify_utils.is_notify_pending("sess-a") is True
-    assert notify_utils.clear_notify_flag("sess-a") is True
-    assert notify_utils.is_notify_pending("sess-a") is False
-    # Clearing again is a no-op (nothing to remove).
-    assert notify_utils.clear_notify_flag("sess-a") is False
-
-
-def test_sessions_are_independent(home):
-    """A /notify set in session A must not register as pending for B."""
-    notify_utils.set_notify_flag("sess-a")
-    assert notify_utils.is_notify_pending("sess-a") is True
-    assert notify_utils.is_notify_pending("sess-b") is False
-
-
-def test_distinct_keys_get_distinct_sentinels(home):
-    pa = notify_utils.get_notify_sentinel_path("sess-a")
-    pb = notify_utils.get_notify_sentinel_path("sess-b")
-    assert pa != pb
-
-
-def test_empty_key_uses_default_sentinel(home):
-    # Classic CLI (no session key) gets the unsuffixed default file.
-    assert notify_utils.get_notify_sentinel_path("").name == ".notify_pending"
-    assert notify_utils.get_notify_sentinel_path(None).name == ".notify_pending"
-
-
-def test_consume_fires_once_and_clears(home, monkeypatch):
-    fired = []
-    monkeypatch.setattr(
-        notify_utils, "fire_notification",
-        lambda **kw: fired.append(kw),
-    )
-    notify_utils.set_notify_flag("sess-a")
-
-    assert notify_utils.consume_pending_notification("sess-a") is True
-    assert len(fired) == 1
-    # Sentinel consumed — a second consume is a no-op.
-    assert notify_utils.consume_pending_notification("sess-a") is False
-    assert len(fired) == 1
-
-
-def test_consume_is_scoped_to_its_session(home, monkeypatch):
-    fired = []
-    monkeypatch.setattr(
-        notify_utils, "fire_notification",
-        lambda **kw: fired.append(kw),
-    )
-    notify_utils.set_notify_flag("sess-a")
-
-    # Another session completing must NOT consume A's pending notify.
-    assert notify_utils.consume_pending_notification("sess-b") is False
-    assert fired == []
-    assert notify_utils.is_notify_pending("sess-a") is True
-
-
-@pytest.mark.parametrize(
-    "env,expected",
-    [
-        ({"TERM_PROGRAM": "iTerm.app"}, "osc9"),
-        ({"TERM_PROGRAM": "WarpTerminal"}, "osc9"),
-        ({"KITTY_WINDOW_ID": "1"}, "osc9"),
-        ({"WEZTERM_PANE": "0"}, "osc777"),
-        # VS Code / Cursor and Apple Terminal parse but DON'T render OSC
-        # notifications — must fall through to the OS-level path.
-        ({"TERM_PROGRAM": "vscode"}, None),
-        ({"TERM_PROGRAM": "Apple_Terminal"}, None),
-        ({}, None),
-    ],
-)
-def test_detect_terminal_osc(monkeypatch, env, expected):
-    for k in ("TERM_PROGRAM", "KITTY_WINDOW_ID", "WEZTERM_PANE",
-              "GHOSTTY_RESOURCES_DIR"):
-        monkeypatch.delenv(k, raising=False)
-    for k, v in env.items():
-        monkeypatch.setenv(k, v)
-    assert notify_utils._detect_terminal_osc() == expected
-
-
-def test_emit_terminal_notification_writes_osc9(monkeypatch):
-    written = []
-    monkeypatch.setattr(notify_utils, "_detect_terminal_osc", lambda: "osc9")
-    monkeypatch.delenv("TMUX", raising=False)
-    monkeypatch.setattr(notify_utils, "_write_tty",
-                        lambda payload: written.append(payload) or True)
-
-    assert notify_utils._emit_terminal_notification("Hermes", "done") is True
-    assert written == ["\033]9;Hermes: done\007"]
-
-
-def test_emit_terminal_notification_writes_osc777(monkeypatch):
-    written = []
-    monkeypatch.setattr(notify_utils, "_detect_terminal_osc", lambda: "osc777")
-    monkeypatch.delenv("TMUX", raising=False)
-    monkeypatch.setattr(notify_utils, "_write_tty",
-                        lambda payload: written.append(payload) or True)
-
-    assert notify_utils._emit_terminal_notification("Hermes", "done") is True
-    assert written == ["\033]777;notify;Hermes;done\007"]
-
-
-def test_emit_terminal_notification_unknown_terminal_returns_false(monkeypatch):
-    monkeypatch.setattr(notify_utils, "_detect_terminal_osc", lambda: None)
-    called = []
-    monkeypatch.setattr(notify_utils, "_write_tty",
-                        lambda payload: called.append(payload) or True)
-
-    assert notify_utils._emit_terminal_notification("Hermes", "done") is False
-    assert called == []  # no tty write attempted
-
-
-def test_emit_terminal_notification_wraps_for_tmux(monkeypatch):
-    written = []
-    monkeypatch.setattr(notify_utils, "_detect_terminal_osc", lambda: "osc9")
-    monkeypatch.setenv("TMUX", "/tmp/tmux-1000/default,123,0")
-    monkeypatch.setattr(notify_utils, "_write_tty",
-                        lambda payload: written.append(payload) or True)
-
-    notify_utils._emit_terminal_notification("Hermes", "done")
-    assert written and written[0].startswith("\033Ptmux;")
-    assert written[0].endswith("\033\\")
-
-
-def test_desktop_notification_prefers_terminal_over_os(monkeypatch):
-    """Terminal OSC short-circuits the OS-level fallbacks."""
-    monkeypatch.setattr(notify_utils, "_emit_terminal_notification",
-                        lambda t, m: True)
-    macos_called = []
-    monkeypatch.setattr(notify_utils, "_show_notification_macos",
-                        lambda t, m: macos_called.append((t, m)))
-    notify_utils._show_desktop_notification("T", "M")
-    assert macos_called == []
-
-
-def test_osascript_permission_hint_fires_once(monkeypatch, caplog):
-    import logging
-    monkeypatch.setattr(notify_utils, "_OSASCRIPT_HINT_SHOWN", False)
-    with caplog.at_level(logging.WARNING, logger=notify_utils.logger.name):
-        notify_utils._osascript_permission_hint_once()
-        notify_utils._osascript_permission_hint_once()
-    hits = [r for r in caplog.records if "Script Editor" in r.getMessage()]
-    assert len(hits) == 1
-
-
-def test_macos_prefers_terminal_notifier_when_present(monkeypatch):
-    runs = []
-    monkeypatch.setattr(notify_utils.shutil, "which",
-                        lambda name: "/opt/homebrew/bin/terminal-notifier")
-    monkeypatch.setattr(notify_utils.subprocess, "run",
-                        lambda argv, **kw: runs.append(argv))
-
-    notify_utils._show_notification_macos("T", "M")
-
-    assert len(runs) == 1
-    assert runs[0][0].endswith("terminal-notifier")
-    assert "-message" in runs[0] and "M" in runs[0]
-
-
-def test_macos_falls_back_to_osascript_without_terminal_notifier(monkeypatch):
-    runs = []
-    monkeypatch.setattr(notify_utils.shutil, "which", lambda name: None)
-    monkeypatch.setattr(notify_utils.subprocess, "run",
-                        lambda argv, **kw: runs.append(argv))
-
-    notify_utils._show_notification_macos("T", "M")
-
-    assert len(runs) == 1
-    assert runs[0][0] == "osascript"
-
-
-def test_key_resolves_from_session_env(home, monkeypatch):
-    """When no explicit key is passed, the current session context is used."""
-    monkeypatch.setattr(
-        notify_utils, "_resolve_session_key",
-        lambda sk: "ctx-key" if sk is None else sk,
-    )
-    notify_utils.set_notify_flag()  # resolves to "ctx-key"
-    assert notify_utils.is_notify_pending("ctx-key") is True
-    assert notify_utils.is_notify_pending() is True

tests/tui_gateway/test_billing_rpc.py

@@ -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

tests/tui_gateway/test_review_summary_callback.py

@@ -120,3 +120,48 @@ def test_review_summary_callback_survives_agent_without_attribute(server, monkey
     # LockedAgent's __slots__ blocks background_review_callback assignment.
     server._init_session("sid-x", "key-x", LockedAgent(), [], cols=80)
     # If we got here, _init_session swallowed the AttributeError gracefully.
+
+
+def test_init_session_sets_memory_notifications_from_config(server, monkeypatch):
+    """_init_session must apply display.memory_notifications to the agent so
+    the TUI/desktop honors the same off/on/verbose toggle as the messaging
+    gateway and CLI. Without this the review always behaved as 'on'."""
+    monkeypatch.setattr(server, "_SlashWorker", lambda *a, **kw: object())
+    monkeypatch.setattr(server, "_wire_callbacks", lambda sid: None)
+    monkeypatch.setattr(server, "_notify_session_boundary", lambda *a, **kw: None)
+    monkeypatch.setattr(server, "_session_info", lambda agent, session=None: {"model": "m"})
+    monkeypatch.setattr(server, "_load_show_reasoning", lambda: False)
+    monkeypatch.setattr(server, "_load_tool_progress_mode", lambda: "all")
+    monkeypatch.setattr(server, "_emit", lambda *a, **kw: None)
+    monkeypatch.setattr(server, "_load_memory_notifications", lambda: "verbose")
+
+    class FakeAgent:
+        model = "fake/model"
+        background_review_callback = None
+        memory_notifications = "on"
+
+    agent = FakeAgent()
+    server._init_session("sid-mn", "key-mn", agent, [], cols=80)
+
+    assert agent.memory_notifications == "verbose"
+
+
+@pytest.mark.parametrize(
+    "raw,expected",
+    [
+        (None, "on"),       # unset → default on
+        ("on", "on"),
+        ("off", "off"),
+        ("verbose", "verbose"),
+        ("VERBOSE", "verbose"),  # case-normalized
+        (True, "on"),       # bool back-compat
+        (False, "off"),
+    ],
+)
+def test_load_memory_notifications_normalization(server, monkeypatch, raw, expected):
+    """_load_memory_notifications mirrors the gateway's bool→str normalization
+    and defaults to 'on' when the key is absent."""
+    display = {} if raw is None else {"memory_notifications": raw}
+    monkeypatch.setattr(server, "_load_cfg", lambda: {"display": display})
+    assert server._load_memory_notifications() == expected
+

tools/approval.py

@@ -151,33 +151,6 @@ def _is_gateway_approval_context() -> bool:
         return True
     return bool(_get_session_platform())
 
-
-def _notify_approval_request_if_pending() -> None:
-    """Fire a /notify input-needed notification for approval prompts.
-
-    Best-effort only: approval safety flow must not depend on desktop
-    notification delivery. Do not clear the sentinel here; the final
-    turn-complete notification should still fire after the user responds.
-    """
-    try:
-        from tools.notify_utils import (
-            fire_approval_request_notification,
-            is_notify_pending,
-        )
-
-        # /notify is local CLI/TUI-only.  TUI gateway sessions set only a
-        # session key; messaging gateway sessions also set a platform.  Do not
-        # let a local sentinel trigger desktop notifications from Telegram,
-        # Discord, Slack, etc. approval flows.
-        if _get_session_platform():
-            return
-
-        if is_notify_pending():
-            fire_approval_request_notification()
-    except Exception as exc:
-        logger.debug("Approval-request notification failed: %s", exc)
-
-
 # Sensitive write targets that should trigger approval even when referenced
 # via shell expansions like $HOME or $HERMES_HOME, or by the resolved absolute
 # active profile home path such as /home/hermes/.hermes/config.yaml. The
@@ -1234,7 +1207,6 @@ def check_dangerous_command(command: str, env_type: str,
             "pattern_key": pattern_key,
             "description": description,
         })
-        _notify_approval_request_if_pending()
         return {
             "approved": False,
             "pattern_key": pattern_key,
@@ -1247,7 +1219,6 @@ def check_dangerous_command(command: str, env_type: str,
             ),
         }
 
-    _notify_approval_request_if_pending()
     choice = prompt_dangerous_approval(command, description,
                                        approval_callback=approval_callback)
 
@@ -1576,10 +1547,6 @@ def check_all_command_guards(command: str, env_type: str,
                     "pattern_key": primary_key,
                     "description": combined_desc,
                 }
-            # The approval prompt reached the user — surface a local /notify
-            # input-needed desktop notification if one is pending (no-op on
-            # messaging-gateway sessions, which carry a platform).
-            _notify_approval_request_if_pending()
             resolved = decision["resolved"]
             choice = decision["choice"]
 
@@ -1629,9 +1596,7 @@ def check_all_command_guards(command: str, env_type: str,
                     "user_approved": True, "description": combined_desc}
 
         # Fallback: no gateway callback registered (e.g. cron, batch).
-        # Return approval_required for backward compat.  Do not fire the local
-        # desktop /notify hook here because there is no local UI prompt to pair
-        # it with.
+        # Return approval_required for backward compat.
         submit_pending(session_key, {
             "command": command,
             "pattern_key": primary_key,
@@ -1661,7 +1626,6 @@ def check_all_command_guards(command: str, env_type: str,
         session_key=session_key,
         surface="cli",
     )
-    _notify_approval_request_if_pending()
     choice = prompt_dangerous_approval(command, combined_desc,
                                        allow_permanent=not has_tirith,
                                        approval_callback=approval_callback)

tools/notify_utils.py

@@ -1,441 +0,0 @@
-"""Desktop notification delivery for the /notify slash command.
-
-All functions are fail-safe — notification errors are logged but never
-propagate to the agent loop.
-
-Cross-platform: Linux (notify-send), macOS (osascript),
-Windows (PowerShell), and WSL (bridges to Windows via powershell.exe,
-preferring notify-send via WSLg when available).
-"""
-
-import hashlib
-import logging
-import os
-import platform
-import shutil
-import subprocess
-from pathlib import Path
-from typing import Optional
-
-logger = logging.getLogger(__name__)
-
-_SYSTEM = platform.system()
-
-
-def _hermes_home() -> Path:
-    from hermes_constants import get_hermes_home
-    return get_hermes_home()
-
-
-# ---------------------------------------------------------------------------
-# WSL detection
-# ---------------------------------------------------------------------------
-
-_WSL_CACHE: Optional[bool] = None
-
-
-def _is_wsl() -> bool:
-    """Return True when running under Windows Subsystem for Linux."""
-    global _WSL_CACHE
-    if _WSL_CACHE is not None:
-        return _WSL_CACHE
-    try:
-        with open("/proc/version", "r") as f:
-            _WSL_CACHE = "microsoft" in f.read().lower()
-    except Exception:
-        _WSL_CACHE = False
-    return _WSL_CACHE
-
-
-# ---------------------------------------------------------------------------
-# Sentinel file (per-session)
-# ---------------------------------------------------------------------------
-#
-# The pending-notify flag is scoped to a *session*, not the whole process.
-# The TUI gateway and dashboard serve many concurrent sessions from one
-# process sharing one HERMES_HOME; a single global sentinel would let a
-# ``/notify`` set in session A fire on session B's turn completion. Keying
-# the sentinel by HERMES_SESSION_KEY keeps each session's pending flag
-# independent. Classic single-process CLI has no session key and falls back
-# to the unsuffixed default file — same behavior as before.
-
-
-def _resolve_session_key(session_key: Optional[str]) -> str:
-    """Resolve the session key for the current context.
-
-    Explicit *session_key* wins (used by the TUI gateway, which serves many
-    sessions from one process and must name them explicitly). Otherwise read
-    ``HERMES_SESSION_KEY`` from the session context — a contextvar bound
-    per-turn in the gateway, or ``os.environ`` in the classic CLI and the
-    slash worker. Falls back to ``""`` (the default sentinel).
-    """
-    if session_key is not None:
-        return session_key
-    try:
-        from gateway.session_context import get_session_env
-        return get_session_env("HERMES_SESSION_KEY", "") or ""
-    except Exception:
-        return ""
-
-
-def _sentinel_name(session_key: str) -> str:
-    key = (session_key or "").strip()
-    if not key:
-        return ".notify_pending"
-    digest = hashlib.sha1(key.encode("utf-8", "replace")).hexdigest()[:16]
-    return f".notify_pending-{digest}"
-
-
-def get_notify_sentinel_path(session_key: Optional[str] = None) -> Path:
-    return _hermes_home() / _sentinel_name(_resolve_session_key(session_key))
-
-
-def set_notify_flag(session_key: Optional[str] = None) -> bool:
-    """Write the sentinel file to signal a pending notification."""
-    try:
-        p = get_notify_sentinel_path(session_key)
-        p.parent.mkdir(parents=True, exist_ok=True)
-        p.touch()
-        return True
-    except Exception as e:
-        logger.warning("Failed to write notify sentinel: %s", e)
-        return False
-
-
-def clear_notify_flag(session_key: Optional[str] = None) -> bool:
-    """Remove the sentinel file (cancel or consume notification)."""
-    try:
-        p = get_notify_sentinel_path(session_key)
-        if not p.exists():
-            return False
-        p.unlink()
-        return True
-    except Exception as e:
-        logger.warning("Failed to clear notify sentinel: %s", e)
-        return False
-
-
-def is_notify_pending(session_key: Optional[str] = None) -> bool:
-    """Check if a notification is pending for this session."""
-    return get_notify_sentinel_path(session_key).exists()
-
-
-# ---------------------------------------------------------------------------
-# Desktop notification
-# ---------------------------------------------------------------------------
-
-def _notify_send_available() -> bool:
-    """Return True if notify-send is available and D-Bus is reachable."""
-    if not shutil.which("notify-send"):
-        return False
-    # Quick smoke-test: verify D-Bus notification service exists
-    try:
-        result = subprocess.run(
-            ["notify-send", "--version"],
-            timeout=3, capture_output=True,
-        )
-        return result.returncode == 0
-    except Exception:
-        return False
-
-
-def _show_notification_linux(title: str, message: str) -> None:
-    """Desktop notification on native Linux via notify-send."""
-    try:
-        subprocess.run(
-            ["notify-send", title, message],
-            timeout=5, capture_output=True,
-        )
-        logger.debug("notify: Linux notification sent via notify-send")
-    except FileNotFoundError:
-        logger.debug("notify: notify-send not found on Linux")
-    except subprocess.TimeoutExpired:
-        logger.debug("notify: notify-send timed out on Linux")
-
-
-def _ps_single_quote(value: str) -> str:
-    """Quote a string for a single-quoted PowerShell literal."""
-    return "'" + value.replace("'", "''") + "'"
-
-
-def _show_notification_wsl(title: str, message: str) -> None:
-    """Desktop notification in WSL via Windows balloon tip (PowerShell)."""
-    logger.debug("notify: attempting WSL notification via PowerShell")
-    try:
-        ps_code = (
-            "Add-Type -AssemblyName System.Windows.Forms; "
-            "$n = New-Object System.Windows.Forms.NotifyIcon; "
-            "$n.Icon = [System.Drawing.SystemIcons]::Information; "
-            f"$n.BalloonTipTitle = {_ps_single_quote(title)}; "
-            f"$n.BalloonTipText = {_ps_single_quote(message)}; "
-            "$n.Visible = $true; "
-            "$n.ShowBalloonTip(3000); "
-            "[System.Windows.Forms.Application]::DoEvents(); "
-            "Start-Sleep -Seconds 4; "
-            "$n.Dispose()"
-        )
-        result = subprocess.run(
-            ["powershell.exe", "-c", ps_code],
-            timeout=8, capture_output=True,
-        )
-        if result.returncode != 0:
-            logger.debug("notify: PowerShell balloon failed (rc=%d, stderr=%s)",
-                         result.returncode, result.stderr.decode(errors="replace")[:200])
-        else:
-            logger.debug("notify: WSL notification sent via PowerShell")
-    except subprocess.TimeoutExpired:
-        logger.debug("notify: PowerShell balloon timed out")
-    except FileNotFoundError:
-        logger.debug("notify: powershell.exe not found — is WSL properly configured?")
-    except Exception as e:
-        logger.warning("WSL notification failed: %s", e)
-
-
-def _show_notification_macos(title: str, message: str) -> None:
-    """Desktop notification on macOS.
-
-    Prefer ``terminal-notifier`` when it's on PATH: it ships a real app
-    bundle, so notifications attribute to it, show as banners, and are
-    grantable in System Settings → Notifications. Plain ``osascript display
-    notification`` attributes to the *launching* process — for an unsigned
-    CLI that often can't register an app entry, so macOS delivers it silently
-    to Notification Center with no banner (and no toggle the user can flip).
-    Install with ``brew install terminal-notifier`` for reliable banners;
-    otherwise fall back to osascript.
-    """
-    tn = shutil.which("terminal-notifier")
-    if tn:
-        try:
-            subprocess.run(
-                [tn, "-title", title, "-message", message],
-                timeout=5, capture_output=True,
-            )
-            logger.debug("notify: macOS notification sent via terminal-notifier")
-            return
-        except Exception as e:
-            logger.debug(
-                "notify: terminal-notifier failed (%s), falling back to osascript", e
-            )
-    try:
-        escaped_title = title.replace('\\', '\\\\').replace('"', '\\"')
-        escaped_message = message.replace('\\', '\\\\').replace('"', '\\"')
-        subprocess.run(
-            ["osascript", "-e",
-             f"display notification \"{escaped_message}\" with title \"{escaped_title}\""],
-            timeout=5, capture_output=True,
-        )
-        logger.debug("notify: macOS notification sent via osascript")
-        _osascript_permission_hint_once()
-    except Exception as e:
-        logger.debug("notify: osascript notification failed: %s", e)
-
-
-_OSASCRIPT_HINT_SHOWN = False
-
-
-def _osascript_permission_hint_once() -> None:
-    """Warn once that osascript notifications need Script Editor permission.
-
-    On macOS Sequoia+, ``osascript display notification`` is attributed to
-    ``com.apple.ScriptEditor2`` and is silently dropped until the user grants
-    Script Editor notification permission — the command still exits 0, so a
-    user sees "nothing happened" with no error. Surface the one-time fix so
-    they aren't stuck. (Native terminals using the OSC path never reach here.)
-    """
-    global _OSASCRIPT_HINT_SHOWN
-    if _OSASCRIPT_HINT_SHOWN:
-        return
-    _OSASCRIPT_HINT_SHOWN = True
-    logger.warning(
-        "Desktop notifications use osascript on this terminal, which macOS "
-        "delivers as 'Script Editor' — banners are suppressed until you grant "
-        "permission once: run `open -a 'Script Editor'`, execute "
-        "`display notification \"test\" with title \"test\"` inside it, click "
-        "Allow, then check System Settings > Notifications > Script Editor. "
-        "For native banners, run Hermes in iTerm2/Ghostty/kitty/WezTerm, or "
-        "install the VS Code 'Terminal Notification' extension for Cursor."
-    )
-
-
-# ---------------------------------------------------------------------------
-# Terminal-native notifications (OSC escape sequences)
-# ---------------------------------------------------------------------------
-#
-# The most reliable way to notify from an unsigned CLI — especially on modern
-# macOS, where ``osascript`` notifications are permanently stuck under the
-# "Script Editor" identity (Apple removed sender override in Monterey) and
-# ``terminal-notifier`` is broken on recent releases. Here the *terminal
-# emulator itself* raises the banner: it's attributed to the terminal (which
-# the user already trusts), clicking it focuses the terminal, and there's no
-# extra dependency.
-#
-# We emit the escape sequence directly to the controlling terminal
-# (``/dev/tty``), NOT stdout — the TUI/slash-worker capture stdout, and the
-# sequences are non-rendering so they don't disturb a live TUI screen.
-#
-# Two flavors cover the field (see terminfo.dev): OSC 9 (iTerm2-style, single
-# string) and OSC 777 (urxvt-style, title+body). We pick ONE per terminal via
-# TERM_PROGRAM/env so a terminal that supports both doesn't double-fire.
-# Apple Terminal ignores both → returns False so the caller falls back.
-
-def _detect_terminal_osc() -> Optional[str]:
-    """Return the OSC notification flavor for the current terminal, or None.
-
-    Only terminals that actually *render* an OS notification from the escape
-    sequence are listed. terminfo.dev marks many terminals as "supporting"
-    OSC 9/777, but that only means their parser consumes the sequence — VS
-    Code/Cursor and Apple Terminal silently drop it without showing anything
-    (confirmed: microsoft/vscode#294247, anthropics/claude-code#28338). Those
-    return None so the caller falls back to an OS-level notifier (osascript).
-
-    VS Code / Cursor users who want click-to-focus can install the "Terminal
-    Notification" extension (it parses OSC 9/777 from the terminal stream);
-    that's a user-side opt-in, not something we can assume here.
-    """
-    if os.environ.get("KITTY_WINDOW_ID"):
-        return "osc9"  # kitty also speaks the legacy OSC 9
-    if os.environ.get("WEZTERM_PANE") or os.environ.get("GHOSTTY_RESOURCES_DIR"):
-        return "osc777"
-    tp = os.environ.get("TERM_PROGRAM", "")
-    return {
-        "iTerm.app": "osc9",
-        "WarpTerminal": "osc9",
-        "ghostty": "osc777",
-        "WezTerm": "osc777",
-    }.get(tp)
-
-
-def _tmux_wrap(seq: str) -> str:
-    """Wrap an escape sequence for tmux passthrough so it reaches the outer
-    terminal. Requires ``set -g allow-passthrough on`` in tmux >= 3.3."""
-    return "\033Ptmux;" + seq.replace("\033", "\033\033") + "\033\\"
-
-
-def _emit_terminal_notification(title: str, message: str) -> bool:
-    """Emit an OSC desktop-notification sequence to the controlling terminal.
-
-    Returns True when written to a terminal known to support it. Fully
-    fail-safe.
-    """
-    kind = _detect_terminal_osc()
-    if not kind:
-        return False
-    if kind == "osc777":
-        seq = f"\033]777;notify;{title};{message}\007"
-    else:  # osc9 — single string
-        seq = f"\033]9;{title}: {message}\007" if title else f"\033]9;{message}\007"
-    if os.environ.get("TMUX"):
-        seq = _tmux_wrap(seq)
-    if _write_tty(seq):
-        logger.debug("notify: terminal notification sent via %s", kind)
-        return True
-    return False
-
-
-def _write_tty(payload: str) -> bool:
-    """Write *payload* to the controlling terminal (/dev/tty). Fail-safe."""
-    try:
-        with open("/dev/tty", "w") as tty:
-            tty.write(payload)
-            tty.flush()
-        return True
-    except Exception as e:
-        logger.debug("notify: /dev/tty write failed: %s", e)
-        return False
-
-
-def _show_desktop_notification(title: str, message: str) -> None:
-    """Show a desktop notification bubble.
-
-    Order of preference:
-    1. Terminal-native OSC sequence (works in iTerm2/Ghostty/kitty/WezTerm/
-       Warp/VS Code/Cursor; reliable for an unsigned CLI on modern macOS).
-    2. OS-level fallback per platform (notify-send / terminal-notifier /
-       osascript / PowerShell). Used for Apple Terminal and unknown terminals.
-    """
-    try:
-        if _emit_terminal_notification(title, message):
-            return
-        if _is_wsl():
-            # WSLg path: notify-send bridges to native Windows notifications
-            if _notify_send_available():
-                logger.debug("notify: WSLg notify-send available, using D-Bus path")
-                _show_notification_linux(title, message)
-                return
-            logger.debug("notify: notify-send not available in WSL, falling back to PowerShell")
-            _show_notification_wsl(title, message)
-        elif _SYSTEM == "Linux":
-            _show_notification_linux(title, message)
-        elif _SYSTEM == "Darwin":
-            _show_notification_macos(title, message)
-        elif _SYSTEM == "Windows":
-            ps_code = (
-                "Add-Type -AssemblyName System.Windows.Forms; "
-                "$n = New-Object System.Windows.Forms.NotifyIcon; "
-                "$n.Icon = [System.Drawing.SystemIcons]::Information; "
-                f"$n.BalloonTipTitle = {_ps_single_quote(title)}; "
-                f"$n.BalloonTipText = {_ps_single_quote(message)}; "
-                "$n.Visible = $true; "
-                "$n.ShowBalloonTip(3000); "
-                "Start-Sleep -Seconds 4"
-            )
-            subprocess.run(
-                ["powershell", "-c", ps_code],
-                timeout=8, capture_output=True,
-            )
-            logger.debug("notify: Windows notification sent via PowerShell")
-    except Exception as e:
-        logger.debug("Desktop notification failed: %s", e)
-
-
-# ---------------------------------------------------------------------------
-# Public API
-# ---------------------------------------------------------------------------
-
-def fire_notification(
-    *,
-    title: str = "Hermes Agent",
-    message: str = "Task complete",
-) -> None:
-    """Fire a desktop notification.
-
-    All errors are caught silently — notification failure must never
-    crash the idle loop.
-
-    Args:
-        title: Desktop notification title.
-        message: Desktop notification body.
-    """
-    _show_desktop_notification(title, message)
-
-
-def fire_approval_request_notification() -> None:
-    """Notify that Hermes is blocked waiting for command approval.
-
-    This intentionally does not clear the /notify sentinel; the final
-    turn-complete notification should still fire after the user responds.
-    """
-    fire_notification(message="Input needed: approval required")
-
-
-def consume_pending_notification(
-    session_key: Optional[str] = None,
-    *,
-    title: str = "Hermes Agent",
-    message: str = "Task complete",
-) -> bool:
-    """Fire-and-clear the pending notification for *session_key*, if any.
-
-    Single entry point for the turn-complete sites (CLI idle loop, CLI
-    process loop, TUI gateway success/error paths) so the
-    check→clear→fire sequence lives in one place. Returns True when a
-    notification was fired. Fully fail-safe.
-    """
-    try:
-        if is_notify_pending(session_key):
-            clear_notify_flag(session_key)
-            fire_notification(title=title, message=message)
-            return True
-    except Exception as e:
-        logger.debug("notify consume failed: %s", e)
-    return False

tui_gateway/server.py

@@ -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",
@@ -1907,6 +1908,22 @@ def _load_show_reasoning() -> bool:
     return bool((_load_cfg().get("display") or {}).get("show_reasoning", False))
 
 
+def _load_memory_notifications() -> str:
+    """Self-improvement review notification mode from config.yaml.
+
+    Parity with the messaging gateway (``gateway/run.py``) and the classic CLI:
+    ``display.memory_notifications`` controls whether the background review's
+    "💾 Self-improvement review: …" summary is surfaced. Without this the
+    TUI/desktop backend always behaved as ``"on"`` and silently ignored a user
+    who set ``off``. Accepts ``off`` / ``on`` (default) / ``verbose``; a bool is
+    normalized for back-compat.
+    """
+    raw = (_load_cfg().get("display") or {}).get("memory_notifications")
+    if isinstance(raw, bool):
+        return "on" if raw else "off"
+    return str(raw).lower() if raw else "on"
+
+
 def _load_tool_progress_mode() -> str:
     env = os.environ.get("HERMES_TUI_TOOL_PROGRESS", "").strip().lower()
     if env in {"off", "new", "all", "verbose"}:
@@ -3770,6 +3787,10 @@ def _init_session(
         agent.background_review_callback = lambda message, _sid=sid: _emit(
             "review.summary", _sid, {"text": str(message)}
         )
+        # Honor display.memory_notifications (off | on | verbose) like the
+        # messaging gateway and CLI do — otherwise the review always behaved as
+        # "on" on the TUI/desktop and a user who set "off" was ignored.
+        agent.memory_notifications = _load_memory_notifications()
     except Exception:
         # Bare AIAgents that don't expose the attribute (unlikely, but keep
         # session startup resilient).
@@ -5170,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)
@@ -6330,16 +6566,6 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None:
                         file=sys.stderr,
                     )
 
-            # Notify check — fire if /notify was set for THIS session's turn.
-            # The sentinel is per-session (keyed by session_key), so a
-            # /notify in one TUI session never fires on another session's
-            # completion. Set by the SlashWorker that handled /notify.
-            try:
-                from tools.notify_utils import consume_pending_notification
-                consume_pending_notification(session.get("session_key"))
-            except Exception as e:
-                logging.debug("tui notify idle-check failed: %s", e)
-
             # Apply pending_title now that the DB row exists.
             _pending = session.get("pending_title")
             if _pending and status == "complete":
@@ -6420,13 +6646,6 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None:
                 f"[gateway-turn] {type(e).__name__}: {e}", file=sys.stderr, flush=True
             )
             _emit("error", sid, {"message": str(e)})
-            # If /notify was set for this failed turn, consume it here too so
-            # the (per-session) sentinel doesn't survive into a later turn.
-            try:
-                from tools.notify_utils import consume_pending_notification
-                consume_pending_notification(session.get("session_key"))
-            except Exception as notify_exc:
-                logging.debug("tui notify error-path check failed: %s", notify_exc)
         finally:
             try:
                 if approval_token is not None:
@@ -9277,7 +9496,6 @@ def _(rid, params: dict) -> dict:
             canonical_order=True,
             pricing=True,
             capabilities=True,
-            max_models=50,
         )
         return _ok(rid, payload)
     except Exception as e:

ui-tui/src/__tests__/billingCommand.test.ts

@@ -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')
+  })
+})

ui-tui/src/__tests__/createGatewayEventHandler.test.ts

@@ -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()
+    })
+  })
 })

ui-tui/src/app/createGatewayEventHandler.ts

@@ -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)
 

ui-tui/src/app/interfaces.ts

@@ -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

ui-tui/src/app/overlayStore.ts

@@ -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
     )
 )
 

ui-tui/src/app/slash/commands/billing.ts

@@ -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)
+    }
+  }
+]

ui-tui/src/app/slash/commands/ops.ts

@@ -62,28 +62,6 @@ interface SkillsReloadResponse {
 }
 
 export const opsCommands: SlashCommand[] = [
-  {
-    help: 'notify when this turn finishes; optionally submit a prompt',
-    name: 'notify',
-    run: (arg, ctx, cmd) => {
-      const trimmed = arg.trim()
-
-      ctx.gateway
-        .rpc<SlashExecResponse>('slash.exec', { command: cmd.slice(1), session_id: ctx.sid })
-        .then(
-          ctx.guarded<SlashExecResponse>(r => {
-            const body = r?.output || '/notify: no output'
-            ctx.transcript.sys(body)
-
-            if (trimmed && trimmed.toLowerCase() !== 'cancel') {
-              ctx.transcript.send(trimmed)
-            }
-          })
-        )
-        .catch(ctx.guardedErr)
-    }
-  },
-
   {
     help: 'stop background processes',
     name: 'stop',

ui-tui/src/app/slash/registry.ts

@@ -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,

ui-tui/src/app/useInputHandlers.ts

@@ -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) {

ui-tui/src/components/appOverlays.tsx

@@ -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
 

ui-tui/src/components/billingOverlay.tsx

@@ -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>
+  )
+}

ui-tui/src/gatewayTypes.ts

@@ -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' }