"""Billing and subscription handlers for the interactive CLI (mixed into ``HermesCLI``). cli.py symbols are imported LAZILY inside methods — never at module load (import cycle). User-facing copy is looked up through ``agent.i18n.t`` at call time (never at import — the active language is not known when this module loads), under ``cli.billing.*`` (shared /topup copy) and ``cli.subscription.*`` (/subscription copy). Modal choice VALUES (index 0) stay English identifiers; only labels/descriptions are translated.""" from __future__ import annotations from agent.i18n import t _RULE = "─" * 41 # Poll `failed` reasons → copy key (default: generic line carrying the raw reason). _CHARGE_FAILED_KEYS = { "authentication_required": "cli.billing.charge_failed_authentication_required", "payment_method_expired": "cli.billing.charge_failed_payment_method_expired", "card_declined": "cli.billing.charge_failed_card_declined"} # Submit-time BillingError codes with a fixed copy key (no payload/type inspection). _CHARGE_ERROR_KEYS = { "no_payment_method": "cli.billing.charge_error_no_payment_method", "cli_billing_disabled": "cli.billing.charge_error_remote_spending_disabled", "remote_spending_disabled": "cli.billing.charge_error_remote_spending_disabled", "role_required": "cli.billing.charge_error_role_required", "idempotency_conflict": "cli.billing.charge_error_idempotency_conflict"} # Upgrade 2xx `status` → (copy key, echo recoveryUrl as "Portal:"). Missing status → ambiguous. _UPGRADE_STATUS_KEYS = { "requires_action": ("cli.subscription.upgrade_requires_action", True), "payment_failed": ("cli.subscription.upgrade_payment_failed", True)} # Upgrade 2xx terminal statuses → dim ✓ copy key ({name} = target tier). _UPGRADE_OK_KEYS = { "already_on_tier": "cli.subscription.upgrade_already_on_tier", "upgraded": "cli.subscription.upgrade_upgraded"} # Pending-change mutations → dim ✓ copy key. _PENDING_OK_KEYS = { "schedule": "cli.subscription.pending_scheduled", "cancel": "cli.subscription.pending_cancel_scheduled", "resume": "cli.subscription.pending_resumed"} # ── Static modal menus (value, label, description) — built at call time so labels follow the # active language; order is the rendered order. ── def _cancel_row(value: str = "cancel", label_key: str = "cli.billing.choice_cancel", desc_key: str = "cli.billing.desc_do_nothing") -> tuple[str, str, str]: return (value, t(label_key), t(desc_key)) def _portal_row(value: str = "portal") -> tuple[str, str, str]: return (value, t("cli.billing.choice_manage_on_portal"), t("cli.billing.desc_open_billing_page")) def _allow_remote_spending_choices() -> list[tuple[str, str, str]]: return [ ("yes", t("cli.billing.choice_allow_remote_spending"), t("cli.billing.desc_allow_remote_spending")), ("no", t("cli.billing.choice_not_now"), t("cli.billing.desc_not_now"))] def _topup_menu_choices() -> list[tuple[str, str, str]]: return [ ("buy", t("cli.billing.choice_add_funds"), t("cli.billing.desc_add_funds")), ("auto", t("cli.billing.choice_auto_reload"), t("cli.billing.desc_auto_reload")), ("limit", t("cli.billing.choice_monthly_limit"), t("cli.billing.desc_monthly_limit")), _portal_row(), _cancel_row()] def _add_card_choices() -> list[tuple[str, str, str]]: return [ ("portal", t("cli.billing.choice_add_card_on_portal"), t("cli.billing.desc_add_card_on_portal")), ("recheck", t("cli.billing.choice_recheck_card"), t("cli.billing.desc_recheck_card")), _cancel_row(label_key="cli.billing.choice_back")] def _auto_reload_top_choices() -> list[tuple[str, str, str]]: return [ ("edit", t("cli.billing.choice_edit_thresholds"), t("cli.billing.desc_edit_thresholds")), ("off", t("cli.billing.choice_turn_off"), t("cli.billing.desc_turn_off")), _cancel_row()] def _auto_reload_agree_choices() -> list[tuple[str, str, str]]: return [ ("agree", t("cli.billing.choice_agree_turn_on"), t("cli.billing.desc_agree_turn_on")), _cancel_row()] def _change_plan_row() -> tuple[str, str, str]: return ("change", t("cli.subscription.choice_change_plan"), t("cli.subscription.desc_change_plan")) def _change_menu_tail() -> list[tuple[str, str, str]]: return [_portal_row(), _cancel_row("close", "cli.billing.choice_close")] def _cancel_sub_choices() -> list[tuple[str, str, str]]: return [ ("yes", t("cli.subscription.choice_cancel_subscription"), t("cli.subscription.desc_cancel_subscription")), ("cancel", t("cli.billing.choice_go_back"), t("cli.subscription.desc_keep_your_plan"))] # Keep going on credits: window plans only (the portal sends the setting only there). def _keep_going_row() -> tuple[str, str, str]: return ("hermes", t("cli.billing.keep_going_choice"), t("cli.billing.keep_going_desc")) def _keep_going_agree_choices() -> list[tuple[str, str, str]]: return [ ("agree", t("cli.billing.keep_going_turn_on"), t("cli.billing.keep_going_desc")), _cancel_row(desc_key="cli.billing.keep_going_keep_off_desc")] def _keep_going_off_choices() -> list[tuple[str, str, str]]: return [ ("off", t("cli.billing.keep_going_turn_off"), t("cli.billing.keep_going_turn_off_desc")), _cancel_row(label_key="cli.billing.keep_going_leave_on")] class CLIBillingMixin: """Mixin holding interactive-CLI billing and subscription handlers.""" # ── Shared helpers ── def _modal_choice(self, title, detail, choices): """Run the choice modal and return the normalized choice value.""" raw = self._prompt_text_input_modal(title=title, detail=detail, choices=choices) return self._normalize_slash_confirm_choice(raw, choices) def _block_header(self, icon, title, *, rule=True) -> None: """Blank line, `` `` via _cprint, then (optionally) the rule via print.""" from cli import _cprint, _b print() _cprint(f" {icon} {_b(title)}") if rule: print(f" {_RULE}") def _dim(self, msg, *, icon="", lead=False) -> None: """Dim `` `` line via _cprint; ``lead`` prints a blank line first.""" from cli import _cprint, _d if lead: print() _cprint(f" {icon}{_d(msg)}") def _ok(self, msg) -> None: """Dim ``✓ `` success line.""" from cli import _cprint, _DIM, _RST _cprint(f" {_DIM}✓ {msg}{_RST}") def _print_logged_out(self, state, load_failed, cmd) -> None: """Logged-out / fetch-failed block shared by /subscription and /topup.""" if state.error: self._dim(t("cli.billing.load_failed", label=load_failed, error=state.error), icon="💳 ", lead=True) else: self._dim(t("cli.billing.not_logged_in"), icon="💳 ", lead=True) print(f" {t('cli.billing.run_portal_then', cmd=cmd)}") def _print_org_line(self, state) -> None: """Dim ``Org: · `` line (skipped when there is no org).""" if state.org_name: role = (state.role or "").title() if role: self._dim(t("cli.billing.org_line_role", org=state.org_name, role=role)) else: self._dim(t("cli.billing.org_line", org=state.org_name)) def _try_usage_model(self): """Shared dollar usage model (the only source with top-up dollars); None on any failure.""" try: from agent.billing_usage import build_usage_model return build_usage_model() except Exception: return None def _usage_bar_lines(self, usage, plan_name) -> list: """Plan + top-up bars as lines (filled = remaining). Caller picks the print fn: ordering differs per surface.""" lines: list = [] pb = usage.plan_bar if usage else None if pb is not None and pb.total_usd > 0: filled = max(0, min(10, round(pb.fill_fraction * 10))) bar = ("█" * filled) + ("░" * (10 - filled)) pct_s = t("cli.billing.bar_pct_used", pct=pb.pct_used) if pb.pct_used is not None else "" label = (plan_name or t("cli.billing.bar_plan_label")).ljust(8)[:8] lines.append(" " + t("cli.billing.bar_plan", label=label, bar=bar, remaining=f"{pb.remaining_usd:,.2f}", total=f"{pb.total_usd:,.2f}", pct=pct_s)) tb = usage.topup_bar if usage else None if tb is not None and tb.remaining_usd > 0: lines.append(" " + t("cli.billing.bar_topup", label=t("cli.billing.bar_topup_label").ljust(8)[:8], bar="█" * 10, remaining=f"{tb.remaining_usd:,.2f}")) return lines def _print_total_spendable(self, usage, print_fn) -> None: if usage and usage.has_topup and usage.total_spendable_usd is not None: print_fn(f" {t('cli.billing.total_spendable', amount=f'{usage.total_spendable_usd:,.2f}')}") def _step_up_remote_spending(self, *, explain, noninteractive_msg, declined_msg, not_granted_msg) -> bool: """"! One-time setup" step-up (explain → confirm → device-flow). True only when granted; refusals print.""" print() print(f" {t('cli.billing.one_time_setup')}") self._dim(explain) if not self._app: print(noninteractive_msg) return False choice = self._modal_choice( t("cli.billing.allow_remote_spending_title"), t("cli.billing.allow_remote_spending_detail"), _allow_remote_spending_choices()) if choice != "yes": print(declined_msg) return False print(f" {t('cli.billing.opening_browser_remote_spending')}") 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" {t('cli.billing.couldnt_allow_remote_spending', error=exc)}") return False if not granted: print(not_granted_msg) return bool(granted) def _print_portal_line(self, exc) -> None: """``Portal: `` via _cprint when the error carries a portal deep-link.""" from cli import _cprint if exc is not None and exc.portal_url: _cprint(f" {t('cli.billing.portal_line', url=exc.portal_url)}") def _open_url_in_browser(self, url: str) -> bool: """The one portal opener. Refuses TTY-hijacking text browsers (w3m/lynx over SSH) via the auth guard.""" if not url: return False try: from hermes_cli.auth import _can_open_graphical_browser, _is_remote_session if _is_remote_session() or not _can_open_graphical_browser(): return False except Exception: pass # guard unavailable → plain best-effort open try: import webbrowser return bool(webbrowser.open(url)) except Exception: return False def _open_or_print_url(self, url) -> None: """Open ``url`` in the browser, or print it when no graphical browser can be used.""" if not self._open_url_in_browser(url): print(f" {t('cli.billing.open_this_url', url=url)}") # ── /usage — Nous balance block ── def _print_nous_credits_block(self) -> bool: """Nous dollar balance block (two bars); True if anything printed. Shared dollar model first, then legacy ``nous_credits_lines``. Agent-independent (TUI slash-worker has no live agent). Fail-open.""" from cli import _cprint, _b, _d usage = self._try_usage_model() if usage is not None and usage.available: from agent.billing_usage import format_renews plan = usage.plan_name or (t("cli.billing.plan_free") if usage.status == "free" else None) renews_display = usage.renews_display or format_renews(usage.renews_at) renews = t("cli.billing.renews_suffix", date=renews_display) if renews_display else "" head = [f" {_b(t('cli.billing.plan_line', plan=plan, renews=renews))}"] if plan else [] head += self._usage_bar_lines(usage, usage.plan_name) tail = [] if usage.status == "free": tail.append(f" {_d(t('cli.billing.free_models_only_hint'))}") elif usage.status == "low": _amt = f"${usage.total_spendable_usd:,.2f}" if usage.total_spendable_usd is not None else t("cli.billing.under_five") tail.append(f" {t('cli.billing.low_balance_usage', amount=_amt)}") # All via _cprint like the Plan line: print()/_cprint() flush to different buffers under # patch_stdout. "Total spendable" alone does not count as printed (legacy lines still follow). if plan: print() for ln in head: _cprint(ln) self._print_total_spendable(usage, _cprint) for ln in tail: _cprint(ln) if head or tail: return True from agent.account_usage import nous_credits_lines lines = nous_credits_lines( credits_state=getattr(getattr(self, "agent", None), "_credits_state", None), ) if not lines: return False print() for line in lines: print(f" {line}") return True def _print_usage_cta(self) -> None: """The `/usage` call-to-action; mirrors the TUI's ``USAGE_CTA``. Nous-account only.""" self._dim(t("cli.billing.usage_cta")) # ── /subscription — view plan + change it (CLI surface) ── def _show_subscription(self): """`/subscription` (alias `/upgrade`). Deep-links NAS's ``/manage-subscription`` (NOT Stripe); never charges.""" from agent.subscription_view import build_subscription_state, subscription_manage_url state = build_subscription_state() if not state.logged_in: self._print_logged_out(state, t("cli.subscription.load_failed_label"), "/subscription") return if state.context == "team": # no personal plan — teams run on a shared balance self._block_header("☤", t("cli.subscription.team_header")) self._print_org_line(state) print(f" {t('cli.subscription.team_connected', org=state.org_name or t('cli.subscription.a_team_org'))}") self._dim(t("cli.subscription.personal_note")) return self._subscription_overview(state, subscription_manage_url(state)) def _subscription_overview(self, state, manage_url): """Plan read block, then the action: portal hand-off (member / non-interactive), catalog (Free), change menu.""" from cli import _cprint, _b, _d from agent.billing_usage import format_renews usage = self._try_usage_model() c = state.current is_free = not (c and c.tier_id) can_change = state.can_change_plan plan_name = (c.tier_name or c.tier_id) if c else (usage.plan_name if usage else None) u_status = usage.status if usage else None _spend = usage.total_spendable_usd if usage else None renews_display = usage.renews_display if usage else None if not renews_display and c and c.cycle_ends_at: renews_display = format_renews(c.cycle_ends_at) # Headline flags the pending change ("→ Plus" / "→ cancels"); the banner (cancel > downgrade) # leads so it can't read as "nothing happened". _flip, _trans = "", None _your_plan = t("cli.subscription.your_plan") if c and c.cancel_at_period_end: _flip = t("cli.subscription.status_flip_cancels") _trans = (c.tier_name or _your_plan, t("cli.subscription.cancels"), format_renews(c.cancellation_effective_at) or t("cli.subscription.end_of_billing_period")) elif c and c.pending_downgrade_tier_name: _flip = t("cli.subscription.status_flip_to", tier=c.pending_downgrade_tier_name) _trans = (c.tier_name or _your_plan, c.pending_downgrade_tier_name, format_renews(c.pending_downgrade_at) or t("cli.subscription.end_of_cycle")) _left = t("cli.subscription.status_left", amount=f"{_spend:,.2f}") if _spend is not None else "" if u_status == "low" and _spend is not None: _tail = "" elif not can_change: _tail = t("cli.subscription.status_view_only") else: _tail = t("cli.subscription.status_renews", date=renews_display) if renews_display else "" if plan_name: status = t("cli.subscription.status_line", plan=plan_name, flip=_flip, left=_left, tail=_tail) else: status = t("cli.subscription.status_free") # All-_cprint (blanks included) so the block orders deterministically even when piped. _cprint("") if _trans: _from, _to, _when = _trans _cprint(f" ⏳ {_b(t('cli.subscription.scheduled_change'))}") _cprint(" " + t("cli.subscription.scheduled_change_line", from_plan=_from, to_plan=_to, when=_d(t("cli.subscription.scheduled_when", when=_when)))) self._dim(t("cli.subscription.keep_until_then", plan=_from)) _cprint("") _cprint(f" ☤ {_b(status)}") print(f" {_RULE}") for _bar_ln in self._usage_bar_lines(usage, plan_name): print(_bar_ln) self._print_total_spendable(usage, print) if is_free: self._dim(t("cli.subscription.paid_models_need_subscription")) elif u_status == "low": _amt = f"${_spend:,.2f}" if _spend is not None else t("cli.billing.under_five") _cprint(f" {t('cli.subscription.low_balance', amount=_amt)}") self._print_org_line(state) print(f" {_RULE}") if not can_change: self._dim(t("cli.subscription.plan_changes_need_admin"), lead=True) if manage_url: print(f" {t('cli.billing.manage_on_portal_url', url=manage_url)}") elif not self._app: # non-interactive (TUI slash-worker / piped): the modal can't run print() if manage_url: print(f" {t('cli.subscription.manage_url_line', url=manage_url)}") print(f" {t('cli.subscription.open_then_rerun')}") elif is_free: # a NEW subscription needs a fresh card → catalog + portal deep-link only self._subscription_free_catalog(state, manage_url) else: self._subscription_change_menu(state, manage_url) def _subscription_free_catalog(self, state, manage_url): """Free + admin + interactive: catalog → pick → portal deep-link ``plan=`` (a new sub needs a card).""" from agent.subscription_view import format_tier_row, selectable_tiers, subscription_manage_url tiers = selectable_tiers(state) if not tiers: self._subscription_open_portal(state, manage_url, verb=t("cli.subscription.start_subscription")) return self._block_header("☤", t("cli.subscription.choose_plan")) for i, tier in enumerate(tiers, 1): print(f" {i}. {format_tier_row(tier)}") self._dim(t("cli.subscription.start_opens_portal")) choices = [(tier.tier_id, format_tier_row(tier), t("cli.subscription.desc_start_on_portal", name=tier.name)) for tier in tiers] choices.append(_cancel_row()) raw = self._prompt_text_input_modal( title=t("cli.subscription.start_subscription"), detail=t("cli.subscription.pick_plan_detail"), choices=choices) # Rows are numbered → accept a bare number (the normalizer only knows confirm-dialog digits). _digit = (raw or "").strip() _by_row = tiers[int(_digit) - 1] if _digit.isdigit() and 1 <= int(_digit) <= len(tiers) else None choice = _by_row.tier_id if _by_row else self._normalize_slash_confirm_choice(raw, choices) if not choice or choice == "cancel": print(f" {t('cli.subscription.cancelled_no_plan_started')}") return tier_url = subscription_manage_url(state, tier_id=choice) or manage_url if not tier_url: self._dim(t("cli.billing.no_manage_url")) return picked = next((tier for tier in tiers if tier.tier_id == choice), None) label = picked.name if picked else t("cli.subscription.your_plan") if self._open_url_in_browser(tier_url): print(f" {t('cli.subscription.opening_portal_to_start', plan=label)}") else: print(f" {t('cli.subscription.open_url_to_start', plan=label, url=tier_url)}") print(f" {t('cli.subscription.finish_in_browser')}") def _subscription_open_portal(self, state, manage_url, *, verb=None): """Open / copy the manage-subscription URL — the portal hand-off.""" verb = verb or t("cli.subscription.manage_your_subscription") print() if not manage_url: self._dim(t("cli.billing.no_manage_url")) return choices = [ ("open", verb, t("cli.subscription.desc_open_subscription_page")), ("copy", t("cli.subscription.choice_copy_link"), t("cli.subscription.desc_copy_link")), _cancel_row()] choice = self._modal_choice(verb, "", choices) if choice == "open": self._open_or_print_url(manage_url) print() print(f" {t('cli.subscription.finish_in_browser')}") elif choice == "copy": try: self._write_osc52_clipboard(manage_url) print(f" {t('cli.subscription.copied', url=manage_url)}") except Exception: print(f" {t('cli.subscription.manage_url', url=manage_url)}") else: print(f" {t('cli.billing.cancelled_yellow')}") def _subscription_change_menu(self, state, manage_url): """The in-terminal change menu for a paid admin/owner (interactive).""" c = state.current # A scheduled change makes undo the likeliest intent → promote it first. The Close row is # "close" (not "cancel") so typing "cancel" can't be confused with "Cancel subscription". if c and (c.cancel_at_period_end or c.pending_downgrade_tier_name): keep_name = c.tier_name or t("cli.subscription.your_plan") head = [("keep", t("cli.subscription.choice_keep_plan", plan=keep_name), t("cli.subscription.desc_keep_plan")), _change_plan_row()] else: head = [_change_plan_row(), ("cancel_sub", t("cli.subscription.choice_cancel_subscription"), t("cli.subscription.desc_cancel_subscription"))] choice = self._modal_choice(t("cli.subscription.manage_your_subscription"), "", head + _change_menu_tail()) action = { "change": lambda: self._subscription_pick_tier(state), "keep": lambda: self._subscription_apply(state, ("resume", None)), "cancel_sub": lambda: self._subscription_confirm_cancel(state), "portal": lambda: self._subscription_open_portal(state, manage_url)}.get(choice) if action: action() else: print(f" {t('cli.subscription.closed_no_change')}") def _subscription_pick_tier(self, state): """Tier picker → preview → confirm. Paid tiers other than current (dropping to free = cancellation).""" from agent.subscription_view import format_tier_row, is_upgrade, selectable_tiers c = state.current selectable = selectable_tiers(state) if not selectable: print(f" {t('cli.subscription.no_other_plans')}") return choices = [] for tier in selectable: direction = t("cli.subscription.direction_upgrade") if is_upgrade(state, tier.tier_id) else t("cli.subscription.direction_downgrade") choices.append((tier.tier_id, t("cli.subscription.tier_row", row=format_tier_row(tier), direction=direction), t("cli.subscription.desc_switch_to", name=tier.name))) choices.append(_cancel_row(label_key="cli.billing.choice_back")) choice = self._modal_choice( t("cli.subscription.change_plan_title"), t("cli.subscription.current_pick_detail", plan=c.tier_name if c else t("cli.billing.plan_free")), choices) if not choice or choice == "cancel": print(f" {t('cli.subscription.cancelled_no_plan_change')}") return self._subscription_preview_and_confirm(state, choice) def _subscription_preview_and_confirm(self, state, tier_id, *, allow_stepup=True): """Preview → effect → confirm+apply. ``allow_stepup=False`` (post-grant replay) never re-prompts a step-up.""" from cli import _cprint, _b, _d from agent.subscription_view import is_upgrade, subscription_change_preview_from_payload, subscription_manage_url from hermes_cli.nous_billing import BillingError, BillingScopeRequired, post_subscription_preview self._dim(t("cli.subscription.checking_change")) try: payload = post_subscription_preview(subscription_type_id=tier_id) except BillingScopeRequired: if allow_stepup: self._subscription_handle_scope_required(state, retry=("preview", tier_id)) else: print(f" {t('cli.billing.stepup_stale')}") return except BillingError as exc: self._subscription_render_error(state, exc) return p = subscription_change_preview_from_payload(payload) effect = p.effect target = p.target_tier_name or t("cli.subscription.the_selected_plan") print() if effect == "no_op": self._dim(t("cli.subscription.already_on_nothing_to_change", plan=target)) return if effect not in ("charge_now", "scheduled"): # blocked OR unknown effect → fail SAFE (never schedule on an unrecognized string) and # re-offer the portal. plan= rides along only for an UPGRADE hand-off (downgrades stay native). _cprint(f" 🟡 {p.reason or t('cli.subscription.cannot_confirm_here')}") _plan = tier_id if is_upgrade(state, tier_id) else None _mu = subscription_manage_url(state, tier_id=_plan) if _mu: print(f" {t('cli.billing.manage_on_portal_url', url=_mu)}") return _tag = t("cli.subscription.tag_charged_now") if effect == "charge_now" else t("cli.subscription.tag_scheduled_not_today") _cprint(f" {_b(t('cli.subscription.confirm_plan_change'))} {_d(_tag)}") if effect == "charge_now": _amt = f"${p.amount_due_now_cents / 100:.2f}" if p.amount_due_now_cents is not None else None _charged = t("cli.subscription.charged_amount_now", amount=_amt) if _amt else t("cli.subscription.charged_prorated_now") _cprint(f" {t('cli.subscription.upgrade_will_be_charged', plan=target, charged=_charged)}") # Best-effort: name the exact card, but only when the resolver rung matches what a # subscription charge actually uses (subPin / customerDefault — Stripe's precedence). _card_line = t("cli.subscription.card_on_subscription_charged") try: from agent.billing_view import build_billing_state _bs = build_billing_state(timeout=6.0) _c = _bs.card if _bs.logged_in else None if _c is not None and _c.resolved_via in ("subPin", "customerDefault"): _card_line = t("cli.subscription.named_card_on_subscription_charged", card=_c.masked) except Exception: pass self._dim(_card_line) pay_label = (t("cli.subscription.choice_pay_and_upgrade", amount=_amt) if _amt else t("cli.subscription.choice_upgrade_now_prorated")) action = ("upgrade", tier_id) # The money-moving row is NOT the default — a bare Enter hits "Go back", so a stray keystroke can't charge. confirm_choices = [ _cancel_row(label_key="cli.billing.choice_go_back", desc_key="cli.billing.desc_do_not_charge"), ("yes", pay_label, t("cli.subscription.desc_charge_and_upgrade"))] else: # scheduled (whitelisted above) _when = (p.effective_at[:10] if (p.effective_at and len(p.effective_at) >= 10) else t("cli.subscription.end_of_billing_period")) _cprint(f" {t('cli.subscription.change_takes_effect', plan=target, when=_when)}") pay_label = t("cli.subscription.choice_schedule_change", plan=target) action = ("schedule", tier_id) confirm_choices = [ ("yes", pay_label, t("cli.subscription.desc_apply_change")), _cancel_row(label_key="cli.billing.choice_go_back", desc_key="cli.subscription.desc_do_not_change")] if p.monthly_credits_delta: self._dim(t("cli.subscription.monthly_credits_change", delta=p.monthly_credits_delta)) if self._modal_choice(pay_label, "", confirm_choices) != "yes": print(f" {t('cli.subscription.cancelled_no_plan_change')}") return self._subscription_apply(state, action, allow_stepup=allow_stepup) def _subscription_confirm_cancel(self, state): """Confirm, then schedule a cancellation at period end.""" from cli import _cprint, _b, _d from agent.billing_usage import format_renews c = state.current _end = ((format_renews(c.cycle_ends_at) if (c and c.cycle_ends_at) else None) or t("cli.subscription.end_of_billing_period")) print() _cprint(f" {_b(t('cli.subscription.confirm_cancellation'))} {_d(t('cli.subscription.tag_scheduled_not_today'))}") _cprint(" " + t("cli.subscription.cancel_stays_active_until", plan=(c.tier_name if c else t("cli.subscription.your_plan")), end=_end)) self._dim(t("cli.subscription.keep_remaining_credits")) if self._modal_choice(t("cli.subscription.cancel_subscription_question"), "", _cancel_sub_choices()) != "yes": print(f" {t('cli.subscription.cancelled_plan_unchanged')}") return self._subscription_apply(state, ("cancel", None)) def _subscription_apply(self, state, action, idempotency_key=None, *, allow_stepup=True): """Run ("upgrade"|"schedule", tier_id) / ("cancel"|"resume", None); scope denial → step-up + ONE replay, same key.""" from cli import _cprint from hermes_cli.nous_billing import ( BillingError, BillingTransient, BillingRemoteSpendingRevoked, BillingScopeRequired, BillingSessionRevoked, delete_subscription_pending_change, post_subscription_upgrade, put_subscription_pending_change) kind, arg = action key = None if kind == "upgrade": from agent.billing_view import new_idempotency_key key = idempotency_key or new_idempotency_key() try: if kind == "upgrade": res = post_subscription_upgrade(subscription_type_id=arg, idempotency_key=key) or {} status = res.get("status") name = res.get("targetTierName") or t("cli.subscription.your_new_plan") if status in _UPGRADE_OK_KEYS: self._ok(t(_UPGRADE_OK_KEYS[status], name=name)) elif status in _UPGRADE_STATUS_KEYS: line_key, echo_url = _UPGRADE_STATUS_KEYS[status] _cprint(f" {t(line_key)}") if echo_url and res.get("recoveryUrl"): _cprint(f" {t('cli.billing.portal_line', url=res.get('recoveryUrl'))}") else: # unknown / absent 2xx status → also ambiguous, not a flat failure self._subscription_render_upgrade_ambiguous(None) return pending = { "schedule": (put_subscription_pending_change, {"subscription_type_id": arg}), "cancel": (put_subscription_pending_change, {"cancel": True}), "resume": (delete_subscription_pending_change, {})}.get(kind) if pending: pending[0](**pending[1]) self._ok(t(_PENDING_OK_KEYS[kind])) self._dim(t("cli.subscription.rerun_to_review")) except BillingScopeRequired: # rejects BEFORE charging → route to the step-up if allow_stepup: self._subscription_handle_scope_required(state, retry=action, idempotency_key=key) else: print(f" {t('cli.billing.stepup_stale')}") except BillingError as exc: # Upgrade only: deterministic PRE-charge rejections (Transient/401/403 types, 4xx codes) # never reached Stripe → recovery copy. Transport / 5xx is INDETERMINATE (NAS may have # charged) → steer to a re-check, never a blind retry (a fresh key can't dedup). _pre_charge = (BillingTransient, BillingSessionRevoked, BillingRemoteSpendingRevoked) _ambiguous = (exc.error in ("network_error", "endpoint_unavailable") or exc.status is None or exc.status >= 500) if kind == "upgrade" and _ambiguous and not isinstance(exc, _pre_charge): self._subscription_render_upgrade_ambiguous(exc) else: self._subscription_render_error(state, exc) def _subscription_handle_scope_required(self, state, *, retry, idempotency_key=None): """insufficient_scope → step-up, then replay `retry` ONCE so the user never re-runs the command.""" granted = self._step_up_remote_spending( explain=t("cli.subscription.stepup_explain"), noninteractive_msg=f" {t('cli.subscription.stepup_noninteractive')}", declined_msg=f" {t('cli.subscription.stepup_declined')}", not_granted_msg=f" {t('cli.subscription.stepup_not_granted')}") if not granted: return self._ok(t("cli.billing.remote_spending_allowed")) # Bust the 30s token cache (it still holds the pre-grant token; _request only busts on 401). try: from hermes_cli import nous_billing as _nb _nb.invalidate_cached_token() except Exception: pass # Re-fetch fresh state, then replay the held action ONCE (allow_stepup=False). from agent.subscription_view import build_subscription_state try: fresh = build_subscription_state() except Exception: fresh = state if retry[0] == "preview": self._subscription_preview_and_confirm(fresh, retry[1], allow_stepup=False) else: self._subscription_apply(fresh, retry, idempotency_key=idempotency_key, allow_stepup=False) def _subscription_render_error(self, state, exc): """Render a subscription BillingError (a lighter _billing_render_charge_error).""" from cli import _cprint msg = str(exc) or t("cli.billing.something_went_wrong") if exc.error == "insufficient_scope": # defensive: the flow routes scope to the step-up before here _cprint(f" {t('cli.subscription.remote_spending_not_allowed_yet')}") elif exc.error in ("subscription_mutation_rejected", "preview_rejected"): _cprint(f" 🟡 {msg}") else: _cprint(f" 🔴 {msg}") self._print_portal_line(exc) def _subscription_render_upgrade_ambiguous(self, exc): """AMBIGUOUS outcome (NAS may have charged) → steer to a re-check, never a blind retry (key isn't persisted).""" from cli import _cprint _cprint(f" {t('cli.subscription.upgrade_ambiguous')}") self._dim(t("cli.subscription.rerun_to_check_plan")) self._print_portal_line(exc) # ── /topup — Remote Spending (CLI surface, all 5 screens) ── def _show_billing(self, command: str = "/topup"): """`/topup` — ZERO sub-commands (argument ignored; Overview is the only route). Non-interactive never prompts. 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: self._print_logged_out(state, t("cli.billing.load_failed_label"), "/topup") return 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).""" if not state.portal_url: return if reason: print(f" {reason}") print(f" {t('cli.billing.manage_on_portal_url', url=state.portal_url)}") def _billing_require_admin(self, state, *, icon="💳 ", off_reason_key="cli.billing.killswitch_reason_buy") -> bool: """Admin + org kill-switch gate; portal funnel + False when blocked. ``icon`` adds a blank line + prefix.""" if state.can_change_plan and state.cli_billing_enabled: return True if icon: print() if not state.can_change_plan: self._dim(t("cli.billing.actions_require_admin"), icon=icon) self._billing_portal_hint(state) else: self._dim(t("cli.billing.remote_spending_off_for_org"), icon=icon) self._billing_portal_hint(state, reason=t(off_reason_key)) return False def _billing_overview(self, state): """Screen 1 — balance, bars, menu. No scope preflight (a charge 403s); a missing card does NOT gate it.""" from cli import _cprint, _b from agent.billing_view import format_money usage = self._try_usage_model() print() _cprint(f" 💳 {_b(t('cli.billing.topup_header', balance=format_money(state.balance_usd)))}") self._print_org_line(state) print(f" {_RULE}") for _bar_ln in self._usage_bar_lines(usage, usage.plan_name if usage else None): print(_bar_ln) ar = state.auto_reload if ar is not None: if ar.enabled: print(f" {t('cli.billing.auto_reload_on_line', threshold=format_money(ar.threshold_usd), reload_to=format_money(ar.reload_to_usd))}") else: print(f" {t('cli.billing.auto_reload_off_line')}") if state.keep_going_on_credits is not None: print(f" {t('cli.billing.keep_going_on_line' if state.keep_going_on_credits else 'cli.billing.keep_going_off_line')}") if state.can_change_plan and state.cli_billing_enabled: # card at a glance, full-menu case only if state.card is not None: print(f" {t('cli.billing.card_line', card=state.card.display)}") else: self._dim(t("cli.billing.no_saved_card_add_funds_hint")) print(f" {_RULE}") # Action gating: admin + kill-switch for charge/auto-reload; everyone gets portal. if not self._billing_require_admin(state, icon="", off_reason_key="cli.billing.killswitch_reason_overview"): self._billing_offer_keep_going_off(state) return if not self._app: # non-interactive: no modal, just the portal funnel self._billing_portal_hint(state) return # One-time vs automatic — the distinction stated up front in each first sentence. self._dim(t("cli.billing.add_funds_now_hint")) _amounts = [ar.reload_to_usd, ar.threshold_usd] if ar is not None and ar.enabled else [None] if all(a is not None and a.is_finite() for a in _amounts): _auto_line = t("cli.billing.refill_when_low_amounts", reload_to=format_money(ar.reload_to_usd), threshold=format_money(ar.threshold_usd)) else: _auto_line = t("cli.billing.refill_when_low_generic") self._dim(_auto_line) print(f" {_RULE}") # No "Allow Remote Spending" item — discovered at pay time. "Add funds" charges the org's # portal-saved card (server-held; no card ref leaves the client). # "hermes" only on window plans (the portal sends the setting only there). choices = _topup_menu_choices() if state.keep_going_on_credits is not None: choices.insert(2, _keep_going_row()) # after Auto-reload action = { "buy": self._billing_buy_flow, "auto": self._billing_auto_reload_flow, "hermes": self._billing_keep_going_on_credits_flow, "limit": self._billing_limit_screen, "portal": self._billing_open_portal}.get(self._modal_choice(t("cli.billing.topup_title"), "", choices)) if action: action(state) else: print(f" {t('cli.billing.cancelled')}") def _billing_open_portal(self, state): if not state.portal_url: print(f" {t('cli.billing.no_portal_url')}") return self._open_or_print_url(state.portal_url) print(f" {t('cli.billing.complete_in_browser')}") def _billing_add_card_flow(self, state): """No card → add it on the portal (never in-terminal), bounded re-check loop. Refreshed state, or None.""" from cli import _cprint self._block_header("💳", t("cli.billing.add_card_first"), rule=False) _cprint(f" {t('cli.billing.no_saved_card')}") self._dim(t("cli.billing.add_card_once_hint")) for _ in range(8): # bounded: portal-open plus a handful of re-checks choice = self._modal_choice(t("cli.billing.add_card_title"), "", _add_card_choices()) if choice == "portal": self._billing_open_portal(state) self._dim(t("cli.billing.add_card_then_recheck")) elif choice == "recheck": from agent.billing_view import build_billing_state try: fresh = build_billing_state() except Exception: fresh = None if fresh is not None and fresh.logged_in: state = fresh if state.card is not None: self._ok(t("cli.billing.card_found_continuing", card=state.card.display)) return state print(f" {t('cli.billing.still_no_card')}") else: break print(f" {t('cli.billing.cancelled_no_funds')}") return None def _billing_buy_flow(self, state): """Screen 2 (presets) → Screen 3 (confirm+charge+poll). No scope preflight: react to the server's 403s.""" from agent.billing_view import format_money, validate_charge_amount if not self._billing_require_admin(state): return if not self._app: self._block_header("💳", t("cli.billing.add_funds"), rule=False) print(f" {t('cli.billing.presets', presets=', '.join(format_money(p) for p in state.charge_presets))}") print(f" {t('cli.billing.run_interactive_to_purchase')}") self._billing_portal_hint(state) return if state.card is None: # guided add-card path first, so the amount pick can't 403 state = self._billing_add_card_flow(state) if state is None or state.card is None: return preset_choices = [(str(p), format_money(p), t("cli.billing.desc_one_time_purchase")) for p in state.charge_presets] preset_choices.append(("custom", t("cli.billing.choice_custom_amount"), t("cli.billing.desc_custom_amount"))) preset_choices.append(_cancel_row()) card = state.card choice = self._modal_choice( t("cli.billing.add_funds"), t("cli.billing.payment_line", card=card.display) if card else t("cli.billing.no_saved_card_short"), preset_choices) if not choice or choice == "cancel": print(f" {t('cli.billing.cancelled_no_funds')}") return from decimal import Decimal if choice == "custom": entered = self._prompt_text_input(f" {t('cli.billing.amount_prompt')} ") if entered is None: # cancelled (e.g. slash-worker can't prompt off-thread) print(f" {t('cli.billing.cancelled_no_funds')}") 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(f" {t('cli.billing.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 self._block_header("💳", t("cli.billing.confirm_purchase")) print(f" {t('cli.billing.total_line', amount=format_money(amount))}") if card: print(f" {t('cli.billing.payment_line', card=card.display)}") if card.provenance is None: # older NAS without provenance → generic line self._dim(t("cli.billing.portal_card_will_be_charged")) print(f" {_RULE}") self._dim(t("cli.billing.consent_charge")) confirm_choices = [ ("pay", t("cli.billing.choice_pay_now", amount=format_money(amount)), t("cli.billing.desc_submit_charge")), ("portal", t("cli.billing.choice_manage_on_portal"), t("cli.billing.desc_manage_card_billing")), _cancel_row(label_key="cli.billing.choice_go_back", desc_key="cli.billing.desc_do_not_charge")] if not self._app: print(f" {t('cli.billing.run_interactive_to_confirm')}") return choice = self._modal_choice( t("cli.billing.pay_question", amount=format_money(amount)), card.display if card else t("cli.billing.no_saved_card_lower"), confirm_choices) if choice == "portal": self._billing_open_portal(state) return if choice != "pay": print(f" {t('cli.billing.cancelled_no_funds')}") return key = new_idempotency_key() # reused on the post-step-up resume so a double-submit collapses self._billing_submit_and_poll( state, amount, key, missing_msg=f" 🔴 {t('cli.billing.no_charge_id')}", status_msg=t("cli.billing.charge_submitted"), on_scope=lambda: self._billing_handle_scope_required(state, amount=amount, idempotency_key=key)) def _billing_submit_and_poll(self, state, amount, key, *, missing_msg, status_msg, on_scope=None): """POST the charge, then poll. ``on_scope`` handles a scope denial (first submit); else it renders.""" from cli import _cprint, _d from hermes_cli.nous_billing import BillingError, BillingScopeRequired, post_charge try: result = post_charge(amount_usd=amount, idempotency_key=key) except BillingError as exc: if on_scope is not None and isinstance(exc, BillingScopeRequired): on_scope() else: self._billing_render_charge_error(state, exc) return charge_id = result.get("chargeId") if not charge_id: print(missing_msg) return _cprint(f" {_d(status_msg)}") 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, parse_money from hermes_cli.nous_billing import BillingError, BillingTransient, get_charge_status deadline = _time.time() + 300 while _time.time() < deadline: try: status = get_charge_status(charge_id) except BillingTransient as exc: # retry-after, NOT a failure — back off and keep polling _time.sleep(min(exc.retry_after or 5, 30)) continue except BillingError as exc: print(f" {t('cli.billing.could_not_check_charge', error=exc)}") return state_str = status.get("status") if state_str == "settled": amt = status.get("amountUsd") _added = format_money(parse_money(amt)) if amt else format_money(amount) print(f" {t('cli.billing.added_to_balance', amount=_added)}") return if state_str == "failed": self._billing_render_charge_failed(state, status.get("reason")) return _time.sleep(2.0) # pending print(f" {t('cli.billing.still_processing')}") self._billing_portal_hint(state) def _billing_render_charge_failed(self, state, reason): """Poll `failed` reasons → the right copy + portal funnel.""" reason = (reason or "").strip() key = _CHARGE_FAILED_KEYS.get(reason) print(f" {t(key)}" if key else f" {t('cli.billing.charge_failed_generic', reason=reason or 'processing_error')}") self._billing_portal_hint(state) def _billing_render_charge_error(self, state, exc): """Submit-time BillingError. Order matters: revoked/session before code lookups; Transient before scope.""" from hermes_cli.nous_billing import BillingTransient, BillingRemoteSpendingRevoked, BillingSessionRevoked code = exc.error portal_url = exc.portal_url or state.portal_url if isinstance(exc, BillingRemoteSpendingRevoked) or code == "remote_spending_revoked": # This terminal's spend was revoked; recovery is reconnect. who = t("cli.billing.revoked_by_admin") if exc.actor == "admin" else t("cli.billing.revoked_by_you") print(f" {t('cli.billing.revoked_reconnect', who=who)}") elif isinstance(exc, BillingSessionRevoked) or code == "session_revoked": print(f" {t('cli.billing.session_logged_out')}") elif code in _CHARGE_ERROR_KEYS or exc.code == "remote_spending_disabled": # Fixed copy by `error`; the gate's dual error/code payload may carry it in `.code` only. print(f" {t(_CHARGE_ERROR_KEYS.get(code) or _CHARGE_ERROR_KEYS['cli_billing_disabled'])}") elif code == "monthly_cap_exceeded": remaining = (exc.payload or {}).get("remainingUsd") if remaining is not None: print(f" {t('cli.billing.monthly_cap_reached_headroom', remaining=remaining)}") else: print(f" {t('cli.billing.monthly_cap_reached')}") elif isinstance(exc, BillingTransient): wait = exc.retry_after mins = t("cli.billing.try_again_in_min", minutes=max(1, round(wait / 60))) if wait else "" print(f" {t('cli.billing.too_many_charges', mins=mins)}") elif code == "insufficient_scope": # Never leak the raw billing:manage scope (a raced post-grant replay can re-raise it). print(f" {t('cli.billing.remote_spending_needs_approval')}") else: print(f" 🔴 {exc}") if portal_url: print(f" {t('cli.billing.portal_line', url=portal_url)}") def _billing_handle_scope_required(self, state, *, amount=None, idempotency_key=None): """403 insufficient_scope → reauth, then resume ``amount`` on explicit confirm, reusing the idempotency key.""" from agent.billing_view import build_billing_state, format_money, new_idempotency_key amount_str = format_money(amount) if amount is not None else t("cli.billing.your_topup") granted = self._step_up_remote_spending( explain=t("cli.billing.charge_stepup_explain", amount=amount_str), noninteractive_msg=f" {t('cli.billing.charge_stepup_noninteractive')}", declined_msg=f" {t('cli.billing.charge_stepup_declined')}", not_granted_msg=f" {t('cli.billing.charge_stepup_not_granted')}") if not granted: return # The token now has the scope, but the ORG kill-switch is a separate gate — re-fetch /state. fresh = build_billing_state() if not (fresh.logged_in and fresh.cli_billing_enabled): print(f" {t('cli.billing.allowed_but_org_off')}") self._billing_portal_hint(fresh) return if fresh.card is None: # half-done state: say so rather than a bare "✓ enabled" print(f" ✓ {t('cli.billing.allowed_no_card')}") self._dim(t("cli.billing.topup_on_portal_to_continue")) self._billing_portal_hint(fresh) return if amount is None: # scope-required hit outside a charge (e.g. auto-reload config) print(f" ✓ {t('cli.billing.allowed_run_topup')}") return print(f" ✓ {t('cli.billing.remote_spending_allowed')}") resume_choices = [ ("resume", t("cli.billing.choice_resume_topup", amount=format_money(amount)), t("cli.billing.desc_resume_topup")), _cancel_row(desc_key="cli.billing.desc_do_not_charge")] if self._modal_choice( t("cli.billing.resume_topup_title"), t("cli.billing.resume_topup_detail", amount=format_money(amount)), resume_choices) != "resume": print(f" {t('cli.billing.cancelled_no_funds')}") return self._billing_submit_and_poll( fresh, amount, idempotency_key or new_idempotency_key(), missing_msg=f" {t('cli.billing.no_charge_id')}", status_msg=t("cli.billing.resuming_topup")) def _billing_auto_reload_flow(self, state): """Screen 4 — threshold + reload-to → PATCH. Prefills; validates ``reload_to > threshold``; "Turn off" if on.""" 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) self._block_header("💳", t("cli.billing.auto_reload")) self._dim(t("cli.billing.auto_reload_hint")) if card: print(f" {t('cli.billing.card_on_file', card=card.masked)}") else: print(f" {t('cli.billing.no_saved_card_manage_portal')}") self._billing_portal_hint(state) return _current = (t("cli.billing.auto_reload_rule", threshold=format_money(ar.threshold_usd), reload_to=format_money(ar.reload_to_usd)) if currently_on else "") if currently_on: print(f" {t('cli.billing.currently', rule=_current)}") if not self._app: print(f" {t('cli.billing.run_interactive_to_configure')}") self._billing_portal_hint(state) return if currently_on: # let the user turn it off without re-entering values top = self._modal_choice(t("cli.billing.auto_reload"), t("cli.billing.auto_reload_on_detail", rule=_current), _auto_reload_top_choices()) if top == "off": self._billing_auto_reload_disable(state) return if top != "edit": print(f" {t('cli.billing.cancelled_yellow')}") return _CANCELLED = object() def _ask_amount(label, current): """One amount; empty keeps `current` when editing. Decimal / kept value, or _CANCELLED (already printed).""" cur = format_money(current) if currently_on else None prompt = f" {t('cli.billing.amount_prompt_labeled', label=label)}" + (f" [{cur}]: " if cur else ": ") raw = self._prompt_text_input(prompt) if raw is None: # cancelled (e.g. slash-worker can't prompt off-thread) print(f" {t('cli.billing.cancelled_yellow')}") return _CANCELLED if not (raw or "").strip() and currently_on: return current v = validate_charge_amount(raw or "", min_usd=state.min_usd, max_usd=state.max_usd) if not v.ok or v.amount is None: print(f" 🔴 {v.error}") return _CANCELLED return v.amount threshold_amt = _ask_amount(t("cli.billing.threshold_label"), ar.threshold_usd if currently_on else None) if threshold_amt is _CANCELLED: return reload_amt = _ask_amount(t("cli.billing.reload_to_label"), ar.reload_to_usd if currently_on else None) if reload_amt is _CANCELLED: return if reload_amt is None or threshold_amt is None or reload_amt <= threshold_amt: print(f" {t('cli.billing.reload_must_exceed_threshold')}") return self._dim(t("cli.billing.auto_reload_consent", card=card.masked, threshold=format_money(threshold_amt)), lead=True) if self._modal_choice( t("cli.billing.turn_on_auto_reload_question"), t("cli.billing.auto_reload_rule_cap", threshold=format_money(threshold_amt), reload_to=format_money(reload_amt)), _auto_reload_agree_choices()) != "agree": print(f" {t('cli.billing.cancelled_yellow')}") return if self._billing_patch_auto_top_up(state, enabled=True, threshold=float(threshold_amt), top_up_amount=float(reload_amt)): print(f" {t('cli.billing.auto_reload_enabled', threshold=format_money(threshold_amt), reload_to=format_money(reload_amt))}") def _billing_patch_auto_top_up(self, state, **kwargs) -> bool: """PATCH auto-top-up; scope denials → step-up, other errors → renderer. True on success.""" from hermes_cli.nous_billing import patch_auto_top_up return self._billing_patch(state, patch_auto_top_up, **kwargs) def _billing_patch(self, state, call, **kwargs) -> bool: from hermes_cli.nous_billing import BillingError, BillingScopeRequired try: call(**kwargs) except BillingScopeRequired: self._billing_handle_scope_required(state) return False except BillingError as exc: self._billing_render_charge_error(state, exc) return False return True def _billing_offer_keep_going_off(self, state): """Turning "Keep going on credits" off only stops spending, so an admin can still do it with Remote Spending off (the portal and the server allow it too).""" if not (state.can_change_plan and state.keep_going_on_credits and getattr(self, "_app", None)): return if self._modal_choice(t("cli.billing.keep_going_is_on_title"), "", _keep_going_off_choices()) == "off": self._billing_keep_going_on_credits_flow(state) def _billing_keep_going_on_credits_flow(self, state): """Window plans: flip "Keep going on credits". Turning it on spends money later, so it asks first; turning it off does not.""" turn_on = not state.keep_going_on_credits if turn_on and self._modal_choice( t("cli.billing.keep_going_confirm_title"), t("cli.billing.keep_going_confirm_body"), _keep_going_agree_choices()) != "agree": print(f" {t('cli.billing.cancelled_yellow')}") return self._billing_set_keep_going(state, turn_on, allow_stepup=True) def _billing_set_keep_going(self, state, turn_on: bool, *, allow_stepup: bool) -> None: """PATCH the setting. A scope denial runs the step-up, then sends the same change ONCE (no card or org Remote Spending check: the setting is not a charge, and turning it off must always work).""" from hermes_cli.nous_billing import BillingError, BillingScopeRequired, patch_keep_going_on_credits try: patch_keep_going_on_credits(enabled=turn_on) except BillingScopeRequired as exc: if not allow_stepup: self._billing_render_charge_error(state, exc) return if self._step_up_remote_spending( explain=t("cli.subscription.stepup_explain"), noninteractive_msg=f" {t('cli.subscription.stepup_noninteractive')}", declined_msg=f" {t('cli.subscription.stepup_declined')}", not_granted_msg=f" {t('cli.subscription.stepup_not_granted')}"): self._billing_set_keep_going(state, turn_on, allow_stepup=False) return except BillingError as exc: self._billing_render_charge_error(state, exc) return print(f" {t('cli.billing.keep_going_turned_on' if turn_on else 'cli.billing.keep_going_turned_off')}") from agent.credits_tracker import apply_keep_going_change apply_keep_going_change(getattr(self, "agent", None), turn_on) def _billing_auto_reload_disable(self, state): """PATCH ``enabled:false``; the endpoint still requires threshold/topUpAmount → echo current (or 0).""" 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 if self._billing_patch_auto_top_up(state, enabled=False, threshold=thr, top_up_amount=rel): print(f" {t('cli.billing.auto_reload_disabled')}") def _billing_limit_screen(self, state): """Screen 5 — monthly spend limit (read-only; cap is portal-only).""" from agent.billing_view import format_money self._block_header("💳", t("cli.billing.monthly_spend_limit")) cap = state.monthly_cap if cap is None or cap.limit_usd is None: self._dim(t("cli.billing.no_monthly_cap")) else: ceiling = t("cli.billing.default_ceiling_suffix") if cap.is_default_ceiling else "" print(f" {t('cli.billing.used_this_month', spent=format_money(cap.spent_this_month_usd), limit=format_money(cap.limit_usd), ceiling=ceiling)}") self._dim(t("cli.billing.monthly_limit_read_only")) self._billing_portal_hint(state)