"""Nous free-tier identity: the ``anonymous`` auth method of the ``nous`` provider. The identity is created in two places: by the desktop's first install, which claims the welcome trial (``hermes_cli.welcome_trial``), and on the first hosted connector action for anyone without a Nous identity (``tools.connectors.gateway.config.ensure_guest_identity``, gated by ``guest_allowed``). Either mints an anonymous Nous account (``POST /api/anonymous/create``); its ``anon_`` credential is later exchanged for short-lived JWTs (``POST /api/anonymous/token``). The result is persisted as the singleton ``providers.nous``. Only a guest stamped with the trial claim carries inference (:func:`free_inference_active`), on every surface of the install once the desktop claimed it: chats, cron, side agents, workers, the CLI. Every other guest serves connectors and managed tools. In the resolver ladder (``resolve_provider``) a guest that carries inference sits directly above the implicit AWS Bedrock chain (NS-829): any explicit provider (env key, ``model.provider``, OpenRouter pool, a logged-in ``active_provider``) beats it, and the ladder never creates one. Only two mechanics differ from an OAuth login and both are isolated behind ``is_guest_state``: token acquisition (re-exchange the ``anon_`` credential; there is no refresh token) and routing (the welcome inference host, single model ``nous/welcome``). Users are never shown the words guest / anonymous / account for this state: surfaces say "Nous · free tier". Two user-facing verbs reach the same flow, both keeping the identity's connectors: ``hermes auth upgrade`` in a terminal and ``/login`` inside a chat. Lifecycle lives in ONE primitive, :func:`_reconcile_and_provision`: adopt what the shared store already holds, else mint under the shared-store lock. :func:`ensure_portal_identity` (connectors) and the welcome-trial claim are its only callers; nothing else calls :func:`mint_guest`. """ from __future__ import annotations import logging import math import os import time from dataclasses import dataclass from datetime import datetime, timedelta, timezone, UTC from pathlib import Path from typing import Any, Callable, Dict, Optional from agent.retry_utils import parse_retry_after_seconds from hermes_cli.auth_constants import ( AuthError, DEFAULT_NOUS_PORTAL_URL, DEFAULT_NOUS_WELCOME_URL, NOUS_INFERENCE_INVOKE_SCOPE, WELCOME_CLAIM_FLAG, _decode_jwt_claims, httpx) logger = logging.getLogger("hermes_cli.auth") ANON_AUTH_METHOD = "anonymous" ANON_CLIENT_ID = "nas-anonymous" ANON_ACCOUNT_TIER = "anonymous" GUEST_MODEL = "nous/welcome" ANON_SECRET_HEADER = "x-anonymous-api-secret" # The shared secret gates the anonymous surface during its integration phase. It is a deployment # secret (Sid's), read from the environment only. ANON_SECRET_ENV = "HERMES_ANON_API_SECRET" # Preview cohort for the free tier's connector set, sent once on account creation so the account # service can record it on the account. "1" / "true" send true and "0" / "false" send false; anything # else (unset included) omits the field and the service applies its default. Self-reported and # baked into desktop bundles in plain text: the service must treat it as a preference, never proof. PREVIEW_FULL_CONNECTORS_ENV = "HERMES_PREVIEW_FULL_CONNECTORS" GUEST_MINT_TIMEOUT_SECONDS = 5.0 # Copy shared by every surface that names the free tier (R-USR-1): never guest / anonymous / account. FREE_TIER_LABEL = "Nous · free tier" UPGRADE_HINT = "Run `hermes auth upgrade` to sign in with a Nous account, or /login inside a chat." FREE_TIER_NOT_SIGNED_IN = ( "You're not signed in. Free inference and connectors are always on. " "Run `hermes auth` to sign in with a Nous account.") class AnonCredentialDead(AuthError): """NAS no longer knows this ``anon_`` credential (reaped, or claimed into a real account). The one client rule for reap AND claim: mark dead, re-mint on the next need. """ def _anon_err(message: str, code: str, *, retry_after: Optional[float] = None) -> AuthError: """An ``AuthError`` for a free-tier failure *code*; terminal-ness is the code's (``ANON_TERMINAL_CODES``).""" return AuthError(message, code=code, retry_after=retry_after, retryable=code not in ANON_TERMINAL_CODES) # --- Free-tier failure codes ------------------------------------------------------------------------ # # Every way the account service (NAS) or the wire can refuse the free tier, as one ``AuthError.code`` # each. Surfaces key their copy on the code; the message on the error is the surface-agnostic # fallback (no ``/login``, no ``hermes`` verb, no guest / anonymous / credential). ``retryable`` # says whether a later attempt can succeed at all; ``retry_after`` is the wait the server named. # # What NAS actually sends (nous-account-service ``api/anonymous/gate.ts`` and the routes behind it): # 404 ``not_found`` the surface is not enabled on this deployment (terminal) # 503 ``temporarily_disabled`` the ops breaker is tripped (transient, no hint) # 429 ``temporarily_unavailable`` + Retry-After per-address / per-credential limits # 428 ``pow_required`` / ``pow_invalid`` / ``pow_replayed`` proof-of-work enforced (not implemented here) # 428 ``challenge_required`` + ``challenges[]`` a browser challenge first (``anon_challenge``) # 403 ``signin_required`` / ``access_denied`` this client is refused without an account # 404 ``unknown_token`` the credential was reaped or claimed (re-mint) # 403 ``account_locked`` the account is locked (dead; never re-mint from it) # 401 an outstanding JWT whose account is gone (re-mint) ANON_GATE_CLOSED = "anon_gate_closed" # not enabled here: sign in, or another provider ANON_GATE_PAUSED = "anon_gate_paused" # ops breaker: keep checking in the background ANON_RATE_LIMITED = "anon_rate_limited" # too many sign-ups / exchanges: wait Retry-After ANON_POW_REQUIRED = "anon_pow_required" # proof of work requested: deferred, sign in instead ANON_ACCOUNT_LOCKED = "anon_account_locked" # dead, and no replacement is minted from it ANON_CREDENTIAL_DEAD = "anon_credential_dead" # reaped or claimed: replaced silently, once ANON_UNREACHABLE = "anon_unreachable" # timeout, DNS, refused connection ANON_SERVER_ERROR = "anon_server_error" # 5xx, non-JSON, malformed success body ANON_CHALLENGE_REQUIRED = "anon_challenge_required" # a browser check is still pending (``anon_challenge``) ANON_SIGNIN_REQUIRED = "anon_signin_required" # refused without an account, or a 428 this version can't run # Codes a later attempt cannot fix (for this process / this version). ANON_TERMINAL_CODES = frozenset({ ANON_GATE_CLOSED, ANON_POW_REQUIRED, ANON_ACCOUNT_LOCKED, ANON_SIGNIN_REQUIRED}) # Codes that mean the account service itself is not answering: a sign-in (which goes through the # same service) cannot help either, so surfaces offer "try again" / "another provider" only. ANON_UNREACHABLE_CODES = frozenset({ANON_UNREACHABLE, ANON_SERVER_ERROR}) # Copy per code: what happened, then the one honest way forward. The free MODEL is never "off": # what is unavailable is using Hermes without signing in, and signing in is free. _SIGNIN_IS_FREE = "Signing in is free." ANON_FAILURE_COPY = { ANON_GATE_CLOSED: f"This version can't be used without a Nous account. {_SIGNIN_IS_FREE}", ANON_GATE_PAUSED: f"Using Hermes without signing in is paused for a moment. {_SIGNIN_IS_FREE}", ANON_RATE_LIMITED: "Lots of people are getting started right now. Try again in {wait}. " "Signing in is free and skips the wait.", ANON_POW_REQUIRED: "The Nous server asked for a proof of work, but that isn't implemented in your " "Agent yet. Sign in with a Nous account to continue.", ANON_CHALLENGE_REQUIRED: "Finish the quick check in your browser, then try again.", ANON_SIGNIN_REQUIRED: "Free guest access isn't available here. Sign in with a Nous account to continue.", ANON_ACCOUNT_LOCKED: f"This session can't continue without signing in. {_SIGNIN_IS_FREE}", ANON_CREDENTIAL_DEAD: "Your session ended. A new one starts on its own.", ANON_UNREACHABLE: "The Nous service couldn't be reached. Check your internet connection and try again.", ANON_SERVER_ERROR: "The Nous service had a hiccup. Try again in a moment.", } def friendly_wait(seconds: Any) -> str: """A rounded, spoken duration for user copy: "a few seconds", "about a minute", "about 5 minutes", "about an hour". Never a raw second count.""" try: s = max(0.0, float(seconds or 0)) except (TypeError, ValueError): s = 0.0 if s <= 15: return "a few seconds" if s < 90: return "about a minute" if s < 3600: return f"about {round(s / 60)} minutes" hours = round(s / 3600) return "about an hour" if hours <= 1 else f"about {hours} hours" def anon_failure_copy(code: str, *, retry_after: Any = None) -> str: """The surface-agnostic sentence for a free-tier failure *code* (``ANON_FAILURE_COPY``).""" template = ANON_FAILURE_COPY.get(code) or ANON_FAILURE_COPY[ANON_SERVER_ERROR] return template.format(wait=friendly_wait(retry_after if retry_after else 60)) def guest_allowed() -> bool: """``nous.guest`` (default True) has not switched guests off: no connector guest, no trial.""" try: from hermes_cli.config import load_config_readonly nous_cfg = load_config_readonly().get("nous") except Exception: # config unreadable: keep today's behaviour (no guest) rather than mint logger.debug("guest: config unreadable, treating nous.guest as false", exc_info=True) return False if not isinstance(nous_cfg, dict): return True return bool(nous_cfg.get("guest", True)) def is_guest_state(state: Any) -> bool: return isinstance(state, dict) and state.get("auth_method") == ANON_AUTH_METHOD def is_anonymous_request(provider: Any, api_key: Any) -> bool: """Select anonymous error UX from the credential actually sent, never the saved profile or URL. This is display/recovery metadata, not token verification; the gateway authenticates the JWT. Named free accounts and opaque API keys must retain normal provider errors. """ from hermes_cli.auth_constants import _decode_jwt_claims return provider == "nous" and _decode_jwt_claims(api_key).get("account_tier") == ANON_ACCOUNT_TIER def is_anonymous_agent(agent: Any) -> bool: """:func:`is_anonymous_request` for a live agent: read at call time, since the credential rotates.""" return is_anonymous_request(getattr(agent, "provider", ""), getattr(agent, "api_key", None)) def current_nous_state() -> Optional[dict[str, Any]]: """The profile's ``providers.nous`` state without locking or network (status/picker reads).""" from hermes_cli.auth import _load_auth_store, _load_provider_state try: return _load_provider_state(_load_auth_store(), "nous") except Exception as exc: logger.debug("guest: auth store unreadable: %s", exc) return None def has_guest() -> bool: return is_guest_state(current_nous_state()) def on_local_desktop_backend() -> bool: """This activity is work for the Desktop app on this machine: a request from, or a turn of a chat created by, a connection ``process_identity.local_desktop_client`` accepted. Never an SSH, URL or cloud backend, a shell the desktop launched, or the gateway's other clients.""" from hermes_cli.process_identity import serves_local_desktop_now return serves_local_desktop_now() def free_inference_active(state: Any = None) -> bool: """The welcome trial carries inference: a guest that claimed it and guests still allowed. Only the claim is the local Desktop's (``welcome_trial``); once claimed, every surface of the install runs on it. *state* defaults to the profile's.""" state = current_nous_state() if state is None else state return is_guest_state(state) and state.get(WELCOME_CLAIM_FLAG) is True and guest_allowed() def trial_model() -> str: """The last rung of model resolution, below every model the user named (``model.default``, a cron or task pin): ``nous/welcome`` while the welcome trial carries inference, else ''. Never written to config.yaml, so it stops applying the moment the user configures a real model.""" return GUEST_MODEL if free_inference_active() else "" def has_free_tier_account() -> bool: """True when the profile's Nous identity is a guest that carries inference here. Profile-level: use for status, picker and notice surfaces. Routing decisions (which model a request may carry) must use :func:`route_is_welcome_host` on the SELECTED runtime instead: a credential-pool entry can pick a paid Nous key while the profile singleton is still a guest. """ return free_inference_active() def guest_without_free_tier(state: Any) -> bool: """A guest that does not carry inference here: it serves connectors and managed tools only, so inference, account and status readers treat the profile as not signed in.""" return is_guest_state(state) and not free_inference_active(state) def free_tier_route() -> bool: from hermes_cli.auth import resolve_provider return has_free_tier_account() and resolve_provider("auto") == "nous" WELCOME_HOSTS = frozenset({"welcome-api.nousresearch.com"}) # Dev-only: extra hostnames that count as the welcome host, comma-separated (for example # ``127.0.0.1`` while ``NOUS_INFERENCE_BASE_URL`` points at a local stand-in). Read from the # environment, which the user controls, so it sits at the same trust level as the URL override # itself; it never widens the NETWORK-side allowlist in ``auth_nous``. EXTRA_WELCOME_HOSTS_ENV = "HERMES_EXTRA_WELCOME_HOSTS" def welcome_hosts() -> frozenset[str]: """``WELCOME_HOSTS`` plus any ``HERMES_EXTRA_WELCOME_HOSTS`` entries (lowercased hostnames).""" raw = os.environ.get(EXTRA_WELCOME_HOSTS_ENV) or "" extra = {part.strip().lower() for part in raw.split(",") if part.strip()} return WELCOME_HOSTS | frozenset(extra) if extra else WELCOME_HOSTS def pin_model_for_route(provider: Any, base_url: Any, model: Any) -> Any: """Model policy at agent START: on the Nous welcome host the model is ``nous/welcome``; anywhere else the caller's model stands. Used once, when the route is first finalized. Mid-conversation route changes go through :func:`route_can_serve_model` instead: a conversation's model is never silently rewritten by a credential rotation. """ if provider == "nous" and route_is_welcome_host(base_url): if model and model != GUEST_MODEL: logger.info("Nous free tier: using %s instead of configured model %s", GUEST_MODEL, model) return GUEST_MODEL return model def route_can_serve_model(provider: Any, base_url: Any, model: Any) -> bool: """Eligibility for a credential ROTATION: the welcome host serves only ``nous/welcome``, so a conversation on any other model must not be rotated onto it (and a ``nous/welcome`` conversation may move to the portal host, which serves it too). Non-Nous routes are always eligible.""" if provider != "nous" or not route_is_welcome_host(base_url): return True return not model or model == GUEST_MODEL def route_is_welcome_host(base_url: Any) -> bool: """The routing predicate for the free tier: the welcome host serves exactly ``nous/welcome``. Keyed on the resolved endpoint, never on profile state, so a paid pool credential routed to the portal host keeps its model even when a guest singleton exists beside it. """ from urllib.parse import urlparse try: host = (urlparse(str(base_url or "")).hostname or "").lower() except ValueError: return False return host in welcome_hosts() def on_free_model(agent: Any, base_url: Any) -> bool: """The request went to the free tier's host on a free-tier credential.""" return route_is_welcome_host(base_url) and is_anonymous_agent(agent) def anon_secret() -> str: return (os.environ.get(ANON_SECRET_ENV) or "").strip() def _anon_headers() -> dict[str, str]: from hermes_cli.anon_challenge import user_agent headers = {"content-type": "application/json", "user-agent": user_agent()} if secret := anon_secret(): headers[ANON_SECRET_HEADER] = secret return headers # (status, NAS ``error``) -> (exception class, code). ``None`` matches any error string for that # status; an exact pair wins over the wildcard. Anything unlisted is a server error. _NAS_REFUSALS: dict[tuple, tuple] = { (404, "unknown_token"): (AnonCredentialDead, ANON_CREDENTIAL_DEAD), (404, None): (AuthError, ANON_GATE_CLOSED), # uniform with a nonexistent route, on purpose (401, "invalid_shared_secret"): (AuthError, ANON_GATE_CLOSED), # pre-launch NAS builds only (401, None): (AnonCredentialDead, ANON_CREDENTIAL_DEAD), (403, "account_locked"): (AnonCredentialDead, ANON_ACCOUNT_LOCKED), (403, "anonymous_accounts_disabled"): (AuthError, ANON_GATE_PAUSED), # pre-launch names (403, "circuit_open"): (AuthError, ANON_GATE_PAUSED), (428, None): (AuthError, ANON_POW_REQUIRED), (429, None): (AuthError, ANON_RATE_LIMITED), (503, "temporarily_disabled"): (AuthError, ANON_GATE_PAUSED), } # An endpoint-specific verdict: builds the error from the refusal's JSON body. _Verdict = Callable[[dict[str, Any]], AuthError] def _raise_for_anon_status( response: httpx.Response, *, action: str, overrides: Optional[dict[tuple, _Verdict]] = None, ) -> dict[str, Any]: try: payload = response.json() except ValueError: payload = {} if not isinstance(payload, dict): payload = {} error = str(payload.get("error") or "") status = response.status_code if status in (200, 201): return payload if error.startswith("pow_"): error = "pow_" # pow_required / pow_invalid / pow_replayed are one verdict # An endpoint's own verdicts (same keys as ``_NAS_REFUSALS``) win over the table. if overrides: build = overrides.get((status, error)) or overrides.get((status, None)) if build: raise build(payload) cls, code = (_NAS_REFUSALS.get((status, error)) or _NAS_REFUSALS.get((status, None)) or ((AuthError, ANON_POW_REQUIRED) if error == "pow_" else (AuthError, ANON_SERVER_ERROR))) if code == ANON_SERVER_ERROR: logger.info("Nous free tier %s failed (%s%s)", action, status, f": {error}" if error else "") retry_after = parse_retry_after_seconds(response.headers) raise cls(anon_failure_copy(code, retry_after=retry_after), code=code, retry_after=retry_after, retryable=code not in ANON_TERMINAL_CODES) def mint_request_body() -> dict[str, Any]: """The ``/api/anonymous/create`` body: ``{"preview_full_connectors": bool}`` when the env var is ``1`` / ``true`` or ``0`` / ``false``, else ``{}``.""" raw = (os.environ.get(PREVIEW_FULL_CONNECTORS_ENV) or "").strip() return {"preview_full_connectors": raw in ("1", "true")} if raw in ("1", "true", "0", "false") else {} def mint_guest(client: httpx.Client, portal_base_url: str) -> dict[str, Any]: """``POST /api/anonymous/create`` -> ``{user_id, org_id, token, idle_ttl_days}``. Token shown once.""" response = client.post( f"{portal_base_url.rstrip('/')}/api/anonymous/create", headers=_anon_headers(), json=mint_request_body()) payload = _raise_for_anon_status(response, action="sign-up") token = payload.get("token") if not isinstance(token, str) or not token.startswith("anon_"): logger.info("Nous free tier sign-up returned no credential") raise _anon_err(ANON_FAILURE_COPY[ANON_SERVER_ERROR], ANON_SERVER_ERROR) return payload # ``purpose: "connectors"`` (NAS hermes-portal#1516) mints a JWT whose only scope is # ``connectors:invoke``: connector routes and the tool gateway accept it, the welcome host refuses it, # and NAS never puts the free-inference challenge in front of it. A NAS from before #1516 ignores the # field and answers with an inference token (a superset), so which slot a token fills is read off its # ``scope`` claim, never off the purpose that asked for it. def is_connectors_only(token: Any) -> bool: """*token* carries a scope claim and it lacks ``inference:invoke`` (a scopeless JWT reads as today's).""" from hermes_cli.auth_nous import _scope_values claims = _decode_jwt_claims(token) scopes = _scope_values(claims.get("scope")) | _scope_values(claims.get("scp")) return bool(scopes) and NOUS_INFERENCE_INVOKE_SCOPE not in scopes def held_connectors_token(state: Any, *, skew_seconds: Optional[int] = None) -> Optional[str]: """The bearer a connectors / managed-tools call sends: the profile's token (for a guest the inference one, a superset), else a guest's connectors-only token, preferring one that outlives *skew_seconds*. ``None`` also accepts an expiring token (availability probes).""" from hermes_cli.auth import _is_expiring if not isinstance(state, dict): return None slots = (state, state.get("connectors_token") if is_guest_state(state) else None) held = [(s["access_token"], s.get("expires_at")) for s in slots if isinstance(s, dict) and isinstance(s.get("access_token"), str) and s["access_token"]] live = next((token for token, exp in held if not _is_expiring(exp, skew_seconds or 0)), None) return live or (held[0][0] if held and skew_seconds is None else None) def exchange_anon_jwt( client: httpx.Client, portal_base_url: str, anon_token: str, *, auth_state: dict[str, Any], purpose: str, ) -> dict[str, Any]: """``POST /api/anonymous/token {token}`` -> ``{access_token, expires_in, inference_base_url, ...}``. Raises :class:`AnonCredentialDead` on 404 ``unknown_token`` / 401 (reaped or claimed), and :class:`~hermes_cli.anon_challenge.AnonChallengeRequired` (carrying what the status poll needs: portal, credential, and the *auth_state* whose ``tls`` block the mint used) on a browser challenge. """ from hermes_cli import anon_challenge response = client.post( f"{portal_base_url.rstrip('/')}/api/anonymous/token", headers=_anon_headers(), json={"token": anon_token, "client": anon_challenge.client_info(), "purpose": purpose}) def challenge(body: dict[str, Any]) -> AuthError: return anon_challenge.challenge_error( body, portal_base_url=portal_base_url, anon_token=anon_token, auth_state=auth_state) def signin(body: dict[str, Any]) -> AuthError: # Refused without an account, or a 428 this version has no primitive for: either way the # honest way forward is a sign-in, in the service's own words when it sent some. return anon_challenge.signin_required_error(body.get("message")) def proof_of_work(_body: dict[str, Any]) -> AuthError: # The PoW verdict, kept out of the 428 sign-in catch-all below. return _anon_err(ANON_FAILURE_COPY[ANON_POW_REQUIRED], ANON_POW_REQUIRED) # The challenge gate sits on the token exchange only; every other endpoint keeps the table. payload = _raise_for_anon_status(response, action="token exchange", overrides={ (428, "challenge_required"): challenge, (428, "pow_"): proof_of_work, (428, None): signin, (403, "signin_required"): signin, (403, "access_denied"): signin}) if not isinstance(payload.get("access_token"), str) or not payload["access_token"]: logger.info("Nous free tier token exchange returned no token") raise _anon_err(ANON_FAILURE_COPY[ANON_SERVER_ERROR], ANON_SERVER_ERROR) # A connectors-only grant must neither clear nor announce an inference challenge. if not is_connectors_only(payload["access_token"]): anon_challenge.note_optional_challenges(payload, portal_base_url) return payload def apply_exchange_to_state(state: dict[str, Any], exchanged: dict[str, Any]) -> None: """Write a fresh exchange result into a guest state in place: an inference token (token, expiry, routing) into the state itself, a connectors-only one into ``connectors_token`` beside it.""" from hermes_cli.auth_nous import _validate_nous_inference_url_from_network access_token = exchanged["access_token"] claims = _decode_jwt_claims(access_token) now = datetime.now(UTC) exp = claims.get("exp") if isinstance(exp, (int, float)): expires_at = datetime.fromtimestamp(float(exp), tz=UTC) else: expires_at = now + timedelta(seconds=int(exchanged.get("expires_in") or 900)) if is_connectors_only(access_token): state["connectors_token"] = {"access_token": access_token, "expires_at": expires_at.isoformat()} return # NAS names the welcome host on every exchange; absent (older NAS) or outside the allowlist # (a staging host without NOUS_INFERENCE_BASE_URL set), the literal stands in. Never the paid # host: the gateway cross-refuses an anonymous JWT there. inference_url = (_validate_nous_inference_url_from_network(exchanged.get("inference_base_url")) or DEFAULT_NOUS_WELCOME_URL) scope = claims.get("scope") or claims.get("scp") or state.get("scope") if isinstance(scope, (list, tuple)): scope = " ".join(str(s) for s in scope) state.update( access_token=access_token, token_type="Bearer", scope=scope, obtained_at=now.isoformat(), expires_at=expires_at.isoformat(), expires_in=max(0, int((expires_at - now).total_seconds())), account_tier=str(claims.get("account_tier") or ANON_ACCOUNT_TIER)) state["inference_base_url"] = inference_url for key in ("user_id", "org_id"): if exchanged.get(key): state[key] = exchanged[key] state.pop("refresh_token", None) def _portal_base_url() -> str: from hermes_cli.auth_nous import _nous_portal_env_override return (_nous_portal_env_override() or DEFAULT_NOUS_PORTAL_URL).rstrip("/") def _shared_identity_key(state: Any) -> Optional[str]: """Stable identity of a Nous credential: the anon_ token for a guest, the refresh token for an account. Used to decide whether two stores hold the SAME identity.""" if not isinstance(state, dict): return None return state.get("anon_token") if is_guest_state(state) else state.get("refresh_token") def _mint_locked(client: httpx.Client, portal: str, auth_store: dict[str, Any], *, welcome_claim: bool) -> dict[str, Any]: """Mint under the caller's locks. The identity is persisted as soon as ``create`` succeeds, BEFORE the exchange: a 429 or timeout on the exchange must not lose a credential NAS still honours (the next attempt exchanges the stored one instead of minting again). *welcome_claim* is written in that same save, so a claimed guest is never left unstamped.""" from hermes_cli.auth import _store_provider_state, _save_auth_store from hermes_cli.auth_nous import _write_shared_nous_state minted = mint_guest(client, portal) state: dict[str, Any] = { "auth_method": ANON_AUTH_METHOD, "account_tier": ANON_ACCOUNT_TIER, "anon_token": minted["token"], "client_id": ANON_CLIENT_ID, "portal_base_url": portal.rstrip("/"), "user_id": minted.get("user_id"), "org_id": minted.get("org_id"), "idle_ttl_days": minted.get("idle_ttl_days"), WELCOME_CLAIM_FLAG: welcome_claim, } _store_provider_state(auth_store, "nous", state, set_active=False) _save_auth_store(auth_store) _write_shared_nous_state(state) logger.info("Nous free tier ready (identity minted)") return state # Per-process mint memo: a failed mint is not retried until its cooldown has passed (a closed gate # is never retried; a 429 waits out ``Retry-After``; an unreachable portal climbs a short ladder), # so a boot-time blip cannot hammer the portal AND cannot disable the free tier for the whole # process the way a plain "tried once" flag did. ``clear_dead_guest`` resets it because a retired # credential is a reason to mint again; an explicit user retry (``force=True``) bypasses it. # Keyed per profile home ("" for the unscoped launch profile): profile A's 429 must not stop # profile B from ever getting an identity. _MINT_RETRY_LADDER = (15.0, 60.0, 300.0) # unreachable / server error, attempt 1, 2, 3+ _MINT_PAUSED_MIN_WAIT = 60.0 # ops breaker: never poll it faster than this @dataclass class MintFailure: """Why the last mint for a profile failed and when the next one may run.""" code: str message: str retryable: bool retry_after: float # the wait this failure asked for, in seconds (0 for a terminal code) not_before: float # ``time.monotonic()`` before which ``ensure_portal_identity`` stays quiet attempts: int = 1 def remaining(self) -> float: # Rounded to the millisecond: ``(now + wait) - now`` is not exactly ``wait`` in floating point, # and the ceil below turned that dust into an extra whole second ("retry in 61s"). return 0.0 if not self.retryable else max(0.0, round(self.not_before - time.monotonic(), 3)) def as_payload(self) -> dict[str, Any]: """The wire shape every status RPC carries: ``{error_code, error, retryable, retry_after}`` with ``retry_after`` the seconds still to wait (whole, rounded up).""" return {"error_code": self.code, "error": self.message, "retryable": self.retryable, "retry_after": math.ceil(self.remaining()) if self.retryable else 0} _mint_failures: dict[str, MintFailure] = {} def _mint_memo_key() -> str: from hermes_constants import get_hermes_home_override, hermes_home_key return "" if get_hermes_home_override() is None else hermes_home_key() def _mint_failure_for_profile() -> Optional[MintFailure]: return _mint_failures.get(_mint_memo_key()) def _clear_mint_failure() -> None: _mint_failures.pop(_mint_memo_key(), None) def reset_mint_memo_for_tests() -> None: _mint_failures.clear() def last_mint_failure() -> Optional[dict[str, Any]]: """The most recent mint failure for this profile as a wire payload, or None (never failed, or cleared by a later success / retirement).""" failure = _mint_failure_for_profile() return failure.as_payload() if failure else None def classify_mint_exception(exc: BaseException) -> AuthError: """Every mint failure as one ``AuthError`` with a code: the portal's own refusals already are; the wire's (timeout, DNS, refused connection) and anything else get a code here. Pure: *exc* is never mutated (an uncoded ``AuthError`` is re-raised as a server-error twin).""" if isinstance(exc, AuthError) and exc.code: return exc transport = (TimeoutError, ConnectionError, OSError) try: transport = transport + (httpx.TimeoutException, httpx.TransportError) except Exception: # httpx unavailable (lazy proxy): the stdlib set stands pass code = ANON_UNREACHABLE if isinstance(exc, transport) else ANON_SERVER_ERROR wrapped = _anon_err(ANON_FAILURE_COPY[code], code, retry_after=getattr(exc, "retry_after", None)) wrapped.__cause__ = exc return wrapped def _note_mint_failure(err: AuthError) -> MintFailure: code = str(err.code or ANON_SERVER_ERROR) previous = _mint_failure_for_profile() attempts = previous.attempts + 1 if previous and previous.code == code else 1 retryable = code not in ANON_TERMINAL_CODES if not retryable: wait, not_before = 0.0, float("inf") else: hinted = err.retry_after wait = float(hinted) if hinted else _MINT_RETRY_LADDER[min(attempts, len(_MINT_RETRY_LADDER)) - 1] if code == ANON_GATE_PAUSED: wait = max(wait, _MINT_PAUSED_MIN_WAIT) wait = max(1.0, wait) not_before = time.monotonic() + wait failure = MintFailure(code=code, message=str(err), retryable=retryable, retry_after=wait, not_before=not_before, attempts=attempts) _mint_failures[_mint_memo_key()] = failure return failure def _reconcile_and_provision(*, timeout_seconds: float, force: bool = False, welcome_claim: bool = False) -> Optional[dict[str, Any]]: """The lifecycle body, run under profile lock THEN shared lock (the documented order). 1. The shared store is the identity of record for this Hermes root. If it holds an identity that differs from the profile's, the profile adopts it (a stale guest never outlives a sibling profile's sign-in, and never overwrites it). 2. Otherwise the profile's own identity stands. 3. Nothing anywhere: mint, persisting the credential before exchanging it. """ from hermes_cli.auth import ( _auth_store_lock, _load_auth_store, _load_provider_state, _save_auth_store, _store_provider_state, _resolve_verify) from hermes_cli.auth_nous import ( _nous_http_client, _nous_shared_store_lock, _read_shared_nous_state, _write_shared_nous_state) portal = _portal_base_url() with _auth_store_lock(): auth_store = _load_auth_store() profile_state = _load_provider_state(auth_store, "nous") with _nous_shared_store_lock(timeout_seconds=max(timeout_seconds, 5.0)): shared = _read_shared_nous_state() if shared and _shared_identity_key(shared) != _shared_identity_key(profile_state): state = dict(shared) _store_provider_state(auth_store, "nous", state, set_active=not is_guest_state(state)) _save_auth_store(auth_store) logger.debug("Nous identity adopted from the shared store") return state if profile_state: if not shared: _write_shared_nous_state(profile_state) return profile_state # A caller queued on these locks behind a mint that just failed waits out its cooldown. failure = _mint_failure_for_profile() if failure and not force and time.monotonic() < failure.not_before: return None verify = _resolve_verify(insecure=None, ca_bundle=None, auth_state=None) try: with _nous_http_client(timeout_seconds, verify) as client: return _mint_locked(client, portal, auth_store, welcome_claim=welcome_claim) except Exception as exc: # Noted before the locks are released, so a caller queued on them sees the cooldown. err = classify_mint_exception(exc) err.mint_failure = _note_mint_failure(err) if err is exc: raise raise err from exc def ensure_portal_identity( *, explicit: bool, timeout_seconds: float = GUEST_MINT_TIMEOUT_SECONDS, force: bool = False, welcome_claim: bool = False, ) -> Optional[dict[str, Any]]: """Make sure this profile has a Nous identity (guest or account); mint a guest only if the shared store has none. Returns the ``providers.nous`` state, or None (disabled / failed once already). ``explicit`` is required and must be True: the only callers are the first hosted connector action (``tools.connectors.gateway.config.ensure_guest_identity``), the connectors dead-credential replacement (``managed_tool_gateway._replace_dead_guest_token``) and the desktop's welcome-trial claim (``welcome_trial.claim_desktop_free_tier``), which passes ``welcome_claim=True`` so a guest it mints carries the stamp from its first write. Every other guest is minted with ``welcome_claim: False`` and serves connectors only. Nothing creates an identity as a side effect of reading status, resolving a provider or fetching a connector bearer (NS-845 Q1.2). Order: gate (``guest_allowed``) -> reconcile with the shared store -> mint. Locks are taken profile first, then shared, matching every other Nous path. Blocking, bounded by ``timeout_seconds``. A failed mint is memoised with a cooldown (``MintFailure``): until it passes, and for a terminal code forever, this returns None without touching the portal. ``force=True`` is the user's own retry: it makes exactly one attempt regardless of the cooldown. Every failure raises an ``AuthError`` whose ``code`` is one of the ``ANON_*`` codes and whose ``retry_after`` / ``retryable`` say what a caller may do next. """ if not explicit: raise ValueError("ensure_portal_identity: only explicit creators may call this (explicit=True)") if not guest_allowed(): return None failure = _mint_failure_for_profile() if failure and not force and not current_nous_state() and time.monotonic() < failure.not_before: return None # in cooldown (or terminal) for this profile; do not hammer the portal try: state = _reconcile_and_provision(timeout_seconds=timeout_seconds, force=force, welcome_claim=welcome_claim) except Exception as exc: err = classify_mint_exception(exc) noted = getattr(err, "mint_failure", None) or _note_mint_failure(err) logger.info("Nous free tier not set up (%s, attempt %d%s)", noted.code, noted.attempts, f", next try in {noted.retry_after:.0f}s" if noted.retryable else ", not retried") if err is exc: raise raise err from exc if state is not None: _clear_mint_failure() # Adopting a sibling profile's sign-in replaced this profile's guest under the locks above. from hermes_cli.guest_handoff import drain drain() return state def refresh_guest_state(state: dict[str, Any], client: httpx.Client, *, purpose: str) -> str: """Token-acquisition seam for a guest: re-exchange the ``anon_`` credential in place and return the minted token. Every caller names its *purpose*: ``"connectors"`` for connector and managed-tool bearers, ``"inference"`` only for a guest that carries the welcome trial. The portal URL is the resolver's canonical one (env override, else the validated stored URL, else the default), never a raw stored value on its own. Raises :class:`AnonCredentialDead` when NAS no longer knows the credential; the caller owns re-minting (:func:`ensure_portal_identity` after :func:`clear_dead_guest`). """ if purpose == "inference" and not free_inference_active(state): # The one place a guest's inference token is minted: only the welcome trial on the desktop's # local backend carries one. Not terminal: the guest stays for connectors. raise AuthError("Hermes is not logged into Nous Portal.", provider="nous", code="nous_auth_missing") anon_token = state.get("anon_token") if not isinstance(anon_token, str) or not anon_token: raise AnonCredentialDead(ANON_FAILURE_COPY[ANON_CREDENTIAL_DEAD], code=ANON_CREDENTIAL_DEAD) from hermes_cli.auth import _nous_portal_base_url exchanged = exchange_anon_jwt( client, _nous_portal_base_url(state), anon_token, auth_state=state, purpose=purpose) apply_exchange_to_state(state, exchanged) return exchanged["access_token"] def clear_dead_guest(reason: str, *, dead_token: Optional[str] = None) -> None: """Drop a dead guest so the next need re-mints. Only the identity that actually failed is removed: a stale profile whose credential NAS rejected must not erase a sibling profile's newer sign-in or replacement guest from the shared store. When *dead_token* is None the profile's current guest is treated as the failed one. """ from hermes_cli.auth import ( _auth_store_lock, _load_auth_store, _load_provider_state, _save_auth_store, _store_section) from hermes_cli.auth_nous import _clear_shared_nous_state, _nous_shared_store_lock, _read_shared_nous_state with _auth_store_lock(): auth_store = _load_auth_store() state = _load_provider_state(auth_store, "nous") if is_guest_state(state): token = dead_token or state.get("anon_token") if state.get("anon_token") == token: _store_section(auth_store, "providers").pop("nous", None) _store_section(auth_store, "credential_pool").pop("nous", None) if auth_store.get("active_provider") == "nous": auth_store["active_provider"] = None _save_auth_store(auth_store) else: token = dead_token with _nous_shared_store_lock(): shared = _read_shared_nous_state() if token and is_guest_state(shared) and shared.get("anon_token") == token: _clear_shared_nous_state(reason) _clear_mint_failure() logger.info("Nous free-tier identity retired (%s); a new one is set up on next use", reason) # --- Gateway welcome-tier contract: structured refusals and the model-switch header ------------------ # # The inference gateway answers a welcome-tier request it will not serve with a structured 429 # (``{status, message, reason, retry_after, alternates?, upgrade_url?}``), and a request on the wrong # host with a 400 (or a 403 while the tier is dark) whose message names the right host. A NAMED # account that still asks for ``nous/welcome`` is served the id's backing model and told what to # switch to in the ``x-nous-model-switch`` response header. Every rule for reading those lives here; # the error classifier and the turn loop only call in. MODEL_SWITCH_HEADER = "x-nous-model-switch" # Fairshare refusal reasons the welcome tier can answer with (api ``FairshareRefusalReason``). WELCOME_REFUSAL_REASONS = frozenset( {"model_not_free", "feature_not_free", "at_capacity", "admission_closed", "rate_limited"}) # Reasons that mean "not on this tier, ever": no retry helps, only a sign-in or another provider. WELCOME_TIER_GATE_REASONS = frozenset({"model_not_free", "feature_not_free"}) # Gateway messages (lowercased substrings) for a request on the wrong host or a dark tier. _WELCOME_ROUTE_REFUSALS = ( ("anonymous accounts must use", "anon_on_paid_host"), ("serves anonymous hermes agent accounts only", "named_on_welcome_host"), ("anonymous accounts are not accepted", "tier_disabled"), # The gateway's answer to an expired or unreadable bearer: the credential, not the tier. A # retry that waited out a long rate limit outlives the 15-minute free-tier JWT and lands here. ("invalid jwt", "session_expired"), ) _WELCOME_ROUTE_COPY = { # Only reachable when the route heal (``turn_recovery._recover_welcome_tier``) could not move # the session: the one cause left is a user-set NOUS_INFERENCE_BASE_URL naming the paid host. "anon_on_paid_host": "This install is set to use a different Nous server (NOUS_INFERENCE_BASE_URL). " "Unset it to use the free model, or sign in. {signin}", "named_on_welcome_host": "This Nous account needs to reconnect. {model_hint}", "tier_disabled": "Using Hermes without signing in is switched off right now. " "Sign in to keep chatting, it's free. {signin}", # Only reachable when re-exchanging the free credential failed (``turn_recovery._recover_welcome_tier``). "session_expired": "Hermes couldn't renew its connection to the free model. " "Send your message again, or sign in to keep chatting, it's free. {signin}", } # The sign-in door, phrased for a chat surface (slash command) and for a terminal. _SIGNIN_CHAT = "To sign in: /login." _SIGNIN_TERMINAL = "To sign in: `hermes auth upgrade`." _MODEL_HINT_CHAT = "Run /model and pick the Nous row again." _MODEL_HINT_TERMINAL = "Run `hermes model` and pick the Nous row again." # Terminal copy for a free-model outage once the retries are spent (5xx, transport failure). FREE_TIER_OUTAGE_COPY = ("The free model is having trouble responding right now. " "Try sending your message again in a minute.") def parse_welcome_refusal(body: Any) -> Optional[dict[str, Any]]: """The structured welcome-tier refusal in a gateway 429 body, or None for any other shape. Returns ``{"reason", "retry_after", "alternates", "upgrade_url"}`` with ``retry_after`` an int of whole seconds (0 when the gateway sent none) and ``alternates`` a list of model ids. """ if not isinstance(body, dict): return None reason = body.get("reason") if not isinstance(reason, str) or reason not in WELCOME_REFUSAL_REASONS: return None raw_retry = body.get("retry_after") try: retry_after = max(0, int(float(raw_retry))) if raw_retry not in (None, "") else 0 except (TypeError, ValueError): retry_after = 0 raw_alternates = body.get("alternates") alternates = [str(a) for a in raw_alternates if isinstance(a, str) and a] if isinstance(raw_alternates, list) else [] upgrade_url = body.get("upgrade_url") return {"reason": reason, "retry_after": retry_after, "alternates": alternates, "upgrade_url": upgrade_url if isinstance(upgrade_url, str) else ""} def welcome_refusal_copy(refusal: dict[str, Any], *, model: str = "", in_chat: bool = True, door: bool = True) -> str: """User copy for a structured welcome-tier refusal: what happened and the one way forward. Never guest / anonymous / claim; ``in_chat`` picks ``/login`` over the terminal verb. ``door=False`` leaves the "To sign in: …" tail off, for a surface that renders the sign-in as a button beside the sentence (the desktop's error card).""" signin = (_SIGNIN_CHAT if in_chat else _SIGNIN_TERMINAL) if door else "" reason = str(refusal.get("reason") or "") alternates = refusal.get("alternates") or [] serves = alternates[0] if alternates else GUEST_MODEL retry = int(refusal.get("retry_after") or 0) wait = friendly_wait(retry) if retry > 0 else "a little while" if reason == "model_not_free": what = f"{model} isn't" if model else "That model isn't" return (f"{what} available without signing in, so Hermes uses {serves} for now. " f"Sign in for more models. {signin}").rstrip() if reason == "feature_not_free": return f"That isn't available without signing in. Sign in to use it, it's free. {signin}".rstrip() if reason == "at_capacity": return ("Chatting without signing in is really busy right now. Sign in to skip the queue, " f"it's free, or try again in {wait}. {signin}").rstrip() if reason == "admission_closed": return ("Chatting without signing in is full right now. Sign in to keep going, " f"it's free, or try again in {wait}. {signin}").rstrip() if reason == "rate_limited": return (f"You've used up the allowance for chatting without signing in. It refreshes in {wait}. " f"Sign in for a bigger allowance, it's free. {signin}").rstrip() return f"Hermes couldn't send that without signing in. Signing in is free. {signin}".rstrip() def welcome_route_refusal(status: Any, message: Any, base_url: Any = None) -> Optional[str]: """Which host cross-refusal a gateway 400/403 is; None for any other error. ``"anon_on_paid_host"``: a free-tier JWT reached the paid host. ``"named_on_welcome_host"``: an account or API key reached the free tier's host. ``"tier_disabled"``: the tier is dark (``WELCOME_MODE=off``). Each is deterministic for the request: retrying cannot help. ``"session_expired"``: the 403 names the bearer (``invalid jwt``); a fresh credential heals it. The dark-tier 403 is keyed on the ROUTE, not the message: the gateway's permission error carries only its generic sentence (the detail stays in its logs), so any 403 answered by the welcome host means the tier refused this install. The message needles remain for gateways that do spell it out, and for the two wrong-host 400s.""" if status not in (400, 403): return None text = str(message or "").lower() kind = next((kind for needle, kind in _WELCOME_ROUTE_REFUSALS if needle in text), None) if kind is None and status == 403 and route_is_welcome_host(base_url): return "tier_disabled" return kind def welcome_route_refusal_copy(kind: str, *, in_chat: bool = True, door: bool = True) -> str: template = _WELCOME_ROUTE_COPY.get(kind) or "Hermes couldn't reach the free model on this route." return template.format( host=DEFAULT_NOUS_WELCOME_URL, signin=(_SIGNIN_CHAT if in_chat else _SIGNIN_TERMINAL) if door else "", model_hint=_MODEL_HINT_CHAT if in_chat else _MODEL_HINT_TERMINAL).rstrip() def note_model_switch(agent: Any, headers: Any) -> Optional[str]: """Record the gateway's ``x-nous-model-switch`` header on *agent* for the next call, if present. The header arrives on a NAMED account's response that asked for ``nous/welcome`` (the gateway served the backing model and billed it normally): the free tier's model no longer belongs in this install's configuration. Recorded here, applied by :func:`apply_model_switch` between calls so a response still streaming is never re-labelled under itself. Returns the backing id. """ if headers is None: return None value = None try: value = headers.get(MODEL_SWITCH_HEADER) if value is None and hasattr(headers, "items"): value = next((v for k, v in headers.items() if str(k).lower() == MODEL_SWITCH_HEADER), None) except Exception: return None backing = str(value or "").strip() if not backing: return None requested = str(getattr(agent, "model", "") or "") if backing == requested: return None try: agent._nous_pending_model_switch = (requested, backing) except Exception: return None return backing def apply_model_switch(agent: Any) -> Optional[str]: """Move *agent* (and the config default, when it still names the switched id) to the backing model the gateway named. Returns the new model, or None when nothing was pending. Runs once per recorded header, between calls. The conversation keeps its history; only the id the next request carries changes, so a promoted account stops relying on the gateway's reverse map. The config write is the same one a sign-in completion uses, so ``hermes model`` and the gateway's config re-read agree with the live session. """ pending = getattr(agent, "_nous_pending_model_switch", None) if not pending: return None agent._nous_pending_model_switch = None requested, backing = pending if str(getattr(agent, "model", "") or "") != requested: return None # the session already moved (a /model, a sign-in sweep) agent.model = backing # The gateway's cache check compares agent.model with the config default and evicts on a # mismatch it did not cause; this pair names the move so the check can recognise exactly this # server-driven switch even when the config write below did not land. agent._nous_model_switch = (requested, backing) logger.info("Nous gateway asked to switch %s -> %s; applied for this session", requested, backing) try: from hermes_cli.config import load_config_readonly raw = load_config_readonly().get("model") model_cfg = raw if isinstance(raw, dict) else ({"default": raw} if isinstance(raw, str) else {}) if str(model_cfg.get("default") or "").strip() == requested: from hermes_cli.auth import _update_config_for_provider _update_config_for_provider( "nous", str(getattr(agent, "base_url", "") or ""), default_model=backing) logger.info("Config default model moved %s -> %s", requested, backing) except Exception as exc: logger.debug("model switch: config default left as is: %s", exc) status = getattr(agent, "_buffer_status", None) if callable(status): try: status(f"Model is now {backing} (your account's model; {requested} is the free tier's).") except Exception: pass return backing # One-time CLI notice: an install whose inference is carried by an explicit provider learns once that # the free tier (inference + connectors) now exists. The flag lives on the guest state itself so it # dies with the identity; a fresh guest (re-mint, new profile) may announce itself once more. GUEST_NOTICE_FLAG = "guest_notice_shown" def guest_notice_pending() -> bool: """True when a guest identity exists and the one-time availability notice has not been shown.""" state = current_nous_state() return free_inference_active(state) and not bool(state.get(GUEST_NOTICE_FLAG)) def mark_guest_notice_shown() -> bool: """Persist ``guest_notice_shown`` on the guest's ``providers.nous`` state (whichever store holds it). Returns True when a flag was written; False when there is no guest to mark.""" return set_guest_flag(GUEST_NOTICE_FLAG) is not None def set_guest_flag(flag: str) -> Optional[dict[str, Any]]: """Set *flag* on the guest's ``providers.nous`` state in whichever store holds it, and return the updated state; None when there is no guest.""" from hermes_cli.auth import _provider_state_transaction with _provider_state_transaction("nous") as (auth_store, state, source_path): if not is_guest_state(state) or source_path is None: return None if state.get(flag): return state state = {**state, flag: True} _save_guest_state(auth_store, state, source_path) return state def _save_guest_state(auth_store: dict[str, Any], state: dict[str, Any], source_path: Path) -> None: """Write the guest's ``providers.nous`` back to the store it was read from (the caller holds its lock).""" from hermes_cli.auth import _auth_file_path, _load_auth_store, _same_path, _save_auth_store, _store_section if _same_path(source_path, _auth_file_path()): _store_section(auth_store, "providers")["nous"] = state _save_auth_store(auth_store) else: source_store = _load_auth_store(source_path) _store_section(source_store, "providers")["nous"] = state _save_auth_store(source_store, target_path=source_path) # --- Sign-in completion: the account state from a token grant, and the config settle ------------- # # Every Nous sign-in is the plain device-code login (``anon_sign_in.run_sign_in``). The free tier it # replaces hands its apps over in ``guest_handoff`` after the save. User-facing copy never says # guest / anonymous / claim. UPGRADED_AUTH_METHOD = "oauth_device_code" def _account_state_from_token( token_data: dict[str, Any], *, portal_base_url: str, client_id: str, scope: Optional[str], verify: Any, timeout_seconds: float, ) -> dict[str, Any]: """The ``providers.nous`` shape for the signed-in account (same fields the device-code login writes).""" from hermes_cli.auth import PROVIDER_REGISTRY, _coerce_ttl_seconds, _optional_base_url, _tls_state_from_verify from hermes_cli.auth_nous import _NOUS_EMPTY_AGENT_KEY_FIELDS, _iso_after, refresh_nous_oauth_from_state now = datetime.now(UTC) ttl = _coerce_ttl_seconds(token_data.get("expires_in", 0)) inference_url = ( _optional_base_url(token_data.get("inference_base_url")) or PROVIDER_REGISTRY["nous"].inference_base_url.rstrip("/")) state = { "portal_base_url": portal_base_url, "inference_base_url": inference_url, "client_id": client_id, "scope": token_data.get("scope") or scope, "token_type": token_data.get("token_type", "Bearer"), "access_token": token_data["access_token"], "refresh_token": token_data.get("refresh_token"), "obtained_at": now.isoformat(), "expires_at": _iso_after(now, ttl), "expires_in": ttl, "tls": _tls_state_from_verify(verify), **_NOUS_EMPTY_AGENT_KEY_FIELDS} state = refresh_nous_oauth_from_state(state, timeout_seconds=timeout_seconds, force_refresh=False) state["auth_method"] = UPGRADED_AUTH_METHOD return state def settle_after_upgrade(account_state: dict[str, Any]) -> dict[str, Any]: """After a sign-in from the free tier persisted the account: move the config off the free tier's route. Picking the free-tier row may have written ``model.default: nous/welcome`` and ``model.base_url`` = welcome host. An account cannot keep either: the welcome host refuses account tokens, and the portal host serves ``nous/welcome`` as a paid model. When the config is on the free tier's route, ``model.base_url`` becomes the account's inference host and ``model.default`` the recommended default for the account's tier (:func:`hermes_cli.models.recommended_nous_default_model`, the same pick as ``GET /api/model/recommended-default``), through the same config write a plain Nous login uses. A config on the user's own model and host is left alone. Every sign-in completion (CLI ``hermes auth upgrade``, the desktop poller) calls this once, after ``persist_nous_credentials``. Returns ``{"model": str, "changed": bool}``: ``model`` is the default the config now carries (``""`` when it carries none); ``changed`` says whether this call wrote it. Never raises: a failed pick or write is logged and reported as ``changed: False`` so the sign-in itself still counts. """ from hermes_cli.config import load_config_readonly try: raw = load_config_readonly().get("model") except Exception as exc: logger.warning("sign-in completion: config unreadable, default model left as is: %s", exc) return {"model": "", "changed": False} model_cfg = raw if isinstance(raw, dict) else ({"default": raw} if isinstance(raw, str) else {}) current = str(model_cfg.get("default") or "").strip() on_welcome_model = current == GUEST_MODEL on_welcome_host = route_is_welcome_host(model_cfg.get("base_url")) if not (on_welcome_model or on_welcome_host): return {"model": current, "changed": False} model = current if on_welcome_model: from hermes_cli.models import recommended_nous_default_model try: model = str(recommended_nous_default_model().get("model") or "") except Exception as exc: logger.debug("sign-in completion: recommended default unavailable: %s", exc) model = "" try: from hermes_cli.auth import _update_config_for_provider # One write: host and default move together, so a failure leaves the config as it was # rather than the account host paired with the welcome model. No eligible recommendation # (Portal unreachable, or the plan and org policy admit nothing) clears the default in that # same write; the runtime's silent default applies until the user picks one with `hermes model`. _update_config_for_provider( "nous", str(account_state.get("inference_base_url") or ""), default_model=model if on_welcome_model else None, clear_default=on_welcome_model and not model) except Exception as exc: logger.warning("sign-in completion: could not update the default model: %s", exc) return {"model": current, "changed": False} return {"model": model, "changed": True} def _poll_for_token(*args, **kwargs) -> dict[str, Any]: """Keep both the sign-in module seam and the device-flow seam live at call time.""" from hermes_cli.auth_device_flow import _poll_for_token as poll return poll(*args, **kwargs) def persist_nous_credentials(*args, **kwargs): """Keep the existing auth_nous persistence seam behind the sign-in entry point.""" from hermes_cli.auth_nous import persist_nous_credentials as persist return persist(*args, **kwargs) # Public sign-in imports remain here for existing callers and module-attribute patches. # The flow imports this module only inside calls, so either module can be imported first. from hermes_cli.anon_sign_in import ( AlreadySignedIn, Cancelled, Code, Completed, FREE_TIER_RATE_LIMIT_CARD, FREE_TIER_RATE_LIMIT_CHAT, Failed, LOGIN_BUSY_ELSEWHERE, LOGIN_COMMAND, LOGIN_DM_ONLY, LOGIN_NOT_ALLOWED, LOGIN_STARTING, SignInState, TimedOut, UPGRADE_ALREADY_SIGNED_IN, UPGRADE_CANCELLED, UPGRADE_DO_NOT_SHARE, UPGRADE_NOT_COMPLETED, UPGRADE_NO_DEFAULT_CHAT, UPGRADE_NO_DEFAULT_TERMINAL, UPGRADE_START, UPGRADE_TIMED_OUT, UPGRADE_WAITING, UPGRADE_WAITING_UP_TO, Waiting, _default_persist_guard, format_wait_line, run_sign_in, ) from hermes_cli.anon_sign_in_cli import ( drain_sign_in_copy, render_sign_in_cli, render_sign_in_cli_code, upgrade_guest, )