feat(auth) normalise the way in which we check whether a user has free/paid access to nous portal so we can expose behaviour and error messages accordingly.

This commit is contained in:
Robin Fernandes 2026-05-25 15:10:14 +10:00 committed by Teknium
parent 0bf9b867cf
commit 406901b27d
32 changed files with 2470 additions and 181 deletions

View file

@ -2244,11 +2244,15 @@ def _is_payment_error(exc: Exception) -> bool:
# but sometimes wrap them in 429 or other codes.
# Daily quota exhaustion from Bedrock, Vertex AI, and similar providers
# uses different language but is semantically identical to credit exhaustion.
if status in {402, 429, None}:
if status in {402, 404, 429, None}:
if any(kw in err_lower for kw in (
"credits", "insufficient funds",
"can only afford", "billing",
"payment required",
"out of funds", "run out of funds",
"balance_depleted", "no usable credits",
"model_not_supported_on_free_tier",
"not available on the free tier",
# Daily / monthly / weekly quota exhaustion keywords
"quota exceeded", "quota_exceeded",
"too many tokens per day", "daily limit",
@ -2260,6 +2264,18 @@ def _is_payment_error(exc: Exception) -> bool:
return False
def _nous_portal_account_has_fresh_paid_access() -> bool:
"""Return True only when the fresh Nous account API says paid access is allowed."""
try:
from hermes_cli.nous_account import get_nous_portal_account_info
account_info = get_nous_portal_account_info(force_fresh=True)
return account_info.paid_service_access is True
except Exception as exc:
logger.debug("Auxiliary Nous paid-entitlement refresh check failed: %s", exc)
return False
def _is_rate_limit_error(exc: Exception) -> bool:
"""Detect rate-limit errors that warrant provider fallback.
@ -2288,6 +2304,10 @@ def _is_rate_limit_error(exc: Exception) -> bool:
if not any(kw in err_lower for kw in (
"credits", "insufficient funds", "billing",
"payment required", "can only afford",
"out of funds", "run out of funds",
"balance_depleted", "no usable credits",
"model_not_supported_on_free_tier",
"not available on the free tier",
)):
return True
return False
@ -4937,6 +4957,41 @@ def call_llm(
resolved_provider == "nous"
or base_url_host_matches(_base_info, "inference-api.nousresearch.com")
)
if (
_is_payment_error(first_err)
and client_is_nous
and _nous_portal_account_has_fresh_paid_access()
):
refreshed_client, refreshed_model = _refresh_nous_auxiliary_client(
cache_provider=resolved_provider or "nous",
model=final_model,
async_mode=False,
base_url=resolved_base_url,
api_key=resolved_api_key,
api_mode=resolved_api_mode,
main_runtime=main_runtime,
is_vision=(task == "vision"),
)
if refreshed_client is not None:
logger.info(
"Auxiliary %s: refreshed Nous runtime credentials after paid account check, retrying",
task or "call",
)
if refreshed_model and refreshed_model != kwargs.get("model"):
kwargs["model"] = refreshed_model
try:
return _validate_llm_response(
refreshed_client.chat.completions.create(**kwargs), task)
except Exception as retry_err:
if not (
_is_auth_error(retry_err)
or _is_payment_error(retry_err)
or _is_connection_error(retry_err)
or _is_rate_limit_error(retry_err)
):
raise
first_err = retry_err
if _is_auth_error(first_err) and client_is_nous:
refreshed_client, refreshed_model = _refresh_nous_auxiliary_client(
cache_provider=resolved_provider or "nous",
@ -5339,6 +5394,40 @@ async def async_call_llm(
resolved_provider == "nous"
or base_url_host_matches(_client_base, "inference-api.nousresearch.com")
)
if (
_is_payment_error(first_err)
and client_is_nous
and _nous_portal_account_has_fresh_paid_access()
):
refreshed_client, refreshed_model = _refresh_nous_auxiliary_client(
cache_provider=resolved_provider or "nous",
model=final_model,
async_mode=True,
base_url=resolved_base_url,
api_key=resolved_api_key,
api_mode=resolved_api_mode,
is_vision=(task == "vision"),
)
if refreshed_client is not None:
logger.info(
"Auxiliary %s (async): refreshed Nous runtime credentials after paid account check, retrying",
task or "call",
)
if refreshed_model and refreshed_model != kwargs.get("model"):
kwargs["model"] = refreshed_model
try:
return _validate_llm_response(
await refreshed_client.chat.completions.create(**kwargs), task)
except Exception as retry_err:
if not (
_is_auth_error(retry_err)
or _is_payment_error(retry_err)
or _is_connection_error(retry_err)
or _is_rate_limit_error(retry_err)
):
raise
first_err = retry_err
if _is_auth_error(first_err) and client_is_nous:
refreshed_client, refreshed_model = _refresh_nous_auxiliary_client(
cache_provider=resolved_provider or "nous",

View file

@ -127,6 +127,106 @@ def _ra():
return run_agent
def _nous_entitlement_message(capability: str) -> str:
try:
from hermes_cli.nous_account import (
format_nous_portal_entitlement_message,
get_nous_portal_account_info,
)
account_info = get_nous_portal_account_info(force_fresh=True)
message = format_nous_portal_entitlement_message(
account_info,
capability=capability,
)
return message or ""
except Exception:
return ""
def _print_nous_entitlement_guidance(agent, capability: str) -> bool:
message = _nous_entitlement_message(capability)
if not message:
return False
for line in message.splitlines():
agent._vprint(f"{agent.log_prefix} 💡 {line}", force=True)
return True
def _is_nous_inference_route(provider: str, base_url: str) -> bool:
provider = (provider or "").strip().lower()
if provider == "nous":
return True
base = str(base_url or "")
return (
base_url_host_matches(base, "inference-api.nousresearch.com")
or base_url_host_matches(base, "inference.nousresearch.com")
)
def _billing_or_entitlement_message(
*,
capability: str,
provider: str,
base_url: str,
model: str,
) -> str:
if _is_nous_inference_route(provider, base_url):
return _nous_entitlement_message(capability)
provider_label = (provider or "").strip() or "the selected provider"
model_label = (model or "").strip() or "the selected model"
lines = [
(
f"{provider_label} reported that billing, credits, or account "
f"entitlement is exhausted for {model_label}."
),
"Add credits or update billing with that provider, then retry.",
]
if base_url_host_matches(str(base_url or ""), "openrouter.ai"):
lines.append("OpenRouter credits: https://openrouter.ai/settings/credits")
lines.append("You can switch providers temporarily with /model <model> --provider <provider>.")
return "\n".join(lines)
def _print_billing_or_entitlement_guidance(
agent,
*,
capability: str,
provider: str,
base_url: str,
model: str,
) -> bool:
message = _billing_or_entitlement_message(
capability=capability,
provider=provider,
base_url=base_url,
model=model,
)
if not message:
return False
for line in message.splitlines():
agent._vprint(f"{agent.log_prefix} 💡 {line}", force=True)
return True
def _try_refresh_nous_paid_entitlement_credentials(agent) -> bool:
"""Refresh Nous runtime credentials after a fresh paid-entitlement check."""
try:
from hermes_cli.auth import NOUS_INFERENCE_AUTH_MODE_LEGACY
from hermes_cli.nous_account import get_nous_portal_account_info
account_info = get_nous_portal_account_info(force_fresh=True)
if account_info.paid_service_access is not True:
return False
return agent._try_refresh_nous_client_credentials(
force=False,
inference_auth_mode=NOUS_INFERENCE_AUTH_MODE_LEGACY,
)
except Exception:
return False
def _restore_or_build_system_prompt(agent, system_message, conversation_history):
"""Restore the cached system prompt from the session DB or build it fresh.
@ -1017,6 +1117,7 @@ def run_conversation(
codex_auth_retry_attempted=False
anthropic_auth_retry_attempted=False
nous_auth_retry_attempted=False
nous_paid_entitlement_refresh_attempted=False
copilot_auth_retry_attempted=False
thinking_sig_retry_attempted = False
invalid_encrypted_content_retry_attempted = False
@ -2093,6 +2194,23 @@ def run_conversation(
classified.should_rotate_credential, classified.should_fallback,
)
if (
classified.reason == FailoverReason.billing
and _is_nous_inference_route(
getattr(agent, "provider", "") or "",
getattr(agent, "base_url", "") or "",
)
and not nous_paid_entitlement_refresh_attempted
):
nous_paid_entitlement_refresh_attempted = True
if _try_refresh_nous_paid_entitlement_credentials(agent):
agent._vprint(
f"{agent.log_prefix}🔐 Nous paid access verified — "
"refreshed runtime credentials and retrying request...",
force=True,
)
continue
recovered_with_pool, has_retried_429 = agent._recover_with_credential_pool(
status_code=status_code,
has_retried_429=has_retried_429,
@ -2217,7 +2335,8 @@ def run_conversation(
print(f"{agent.log_prefix}🔐 Nous 401 — Portal authentication failed.")
if _body_text:
print(f"{agent.log_prefix} Response: {_body_text}")
print(f"{agent.log_prefix} Most likely: Portal OAuth expired, account out of credits, or agent key revoked.")
if not _print_nous_entitlement_guidance(agent, "Nous model access"):
print(f"{agent.log_prefix} Most likely: Portal OAuth expired, account out of credits, or agent key revoked.")
print(f"{agent.log_prefix} Troubleshooting:")
print(f"{agent.log_prefix} • Re-authenticate: hermes auth add nous")
print(f"{agent.log_prefix} • Check credits / billing: https://portal.nousresearch.com")
@ -2538,7 +2657,12 @@ def run_conversation(
base_url=getattr(agent, "base_url", None),
)
if not pool_may_recover:
agent._emit_status("⚠️ Rate limited — switching to fallback provider...")
if classified.reason == FailoverReason.billing:
agent._emit_status(
"⚠️ Billing or credits exhausted — switching to fallback provider..."
)
else:
agent._emit_status("⚠️ Rate limited — switching to fallback provider...")
if agent._try_activate_fallback(reason=classified.reason):
retry_count = 0
compression_attempts = 0
@ -2948,7 +3072,20 @@ def run_conversation(
agent._vprint(f"{agent.log_prefix} 🌐 Endpoint: {_base}", force=True)
# Actionable guidance for common auth errors
if classified.is_auth or classified.reason == FailoverReason.billing:
if _provider in {"openai-codex", "xai-oauth", "nous"} and status_code == 401:
if classified.reason == FailoverReason.billing and _print_billing_or_entitlement_guidance(
agent,
capability="model access",
provider=_provider,
base_url=str(_base),
model=_model,
):
pass
elif _provider == "nous" and _print_nous_entitlement_guidance(
agent,
"Nous model access",
):
pass
elif _provider in {"openai-codex", "xai-oauth", "nous"} and status_code == 401:
if _provider == "openai-codex":
agent._vprint(f"{agent.log_prefix} 💡 Codex OAuth token was rejected (HTTP 401). Your token may have been", force=True)
agent._vprint(f"{agent.log_prefix} refreshed by another client (Codex CLI, VS Code). To fix:", force=True)
@ -3018,7 +3155,23 @@ def run_conversation(
primary_recovery_attempted = False
continue
_final_summary = agent._summarize_api_error(api_error)
if is_rate_limited:
_billing_guidance = ""
if classified.reason == FailoverReason.billing:
agent._emit_status(f"❌ Billing or credits exhausted — {_final_summary}")
_billing_guidance = _billing_or_entitlement_message(
capability="model access",
provider=_provider,
base_url=str(_base),
model=_model,
)
_print_billing_or_entitlement_guidance(
agent,
capability="model access",
provider=_provider,
base_url=str(_base),
model=_model,
)
elif is_rate_limited:
agent._emit_status(f"❌ Rate limited after {max_retries} retries — {_final_summary}")
else:
agent._emit_status(f"❌ API failed after {max_retries} retries — {_final_summary}")
@ -3063,7 +3216,12 @@ def run_conversation(
api_kwargs, reason="max_retries_exhausted", error=api_error,
)
agent._persist_session(messages, conversation_history)
_final_response = f"API call failed after {max_retries} retries: {_final_summary}"
if classified.reason == FailoverReason.billing:
_final_response = f"Billing or credits exhausted: {_final_summary}"
if _billing_guidance:
_final_response += f"\n\n{_billing_guidance}"
else:
_final_response = f"API call failed after {max_retries} retries: {_final_summary}"
if _is_stream_drop:
_final_response += (
"\n\nThe provider's stream connection keeps "

View file

@ -97,13 +97,20 @@ _BILLING_PATTERNS = [
"insufficient_quota",
"insufficient balance",
"credit balance",
"credits exhausted",
"credits have been exhausted",
"no usable credits",
"top up your credits",
"payment required",
"billing hard limit",
"exceeded your current quota",
"account is deactivated",
"plan does not include",
"out of funds",
"run out of funds",
"balance_depleted",
"model_not_supported_on_free_tier",
"not available on the free tier",
]
# Patterns that indicate rate limiting (transient, will resolve)
@ -690,8 +697,13 @@ def _classify_by_status(
)
if status_code == 403:
# OpenRouter 403 "key limit exceeded" is actually billing
if "key limit exceeded" in error_msg or "spending limit" in error_msg:
# OpenRouter 403 "key limit exceeded" is actually billing. Other
# providers also use 403 for account-plan or credit exhaustion.
if (
"key limit exceeded" in error_msg
or "spending limit" in error_msg
or any(p in error_msg for p in _BILLING_PATTERNS)
):
return result_fn(
FailoverReason.billing,
retryable=False,
@ -708,6 +720,17 @@ def _classify_by_status(
return _classify_402(error_msg, result_fn)
if status_code == 404:
# Nous API currently surfaces HA/NAS credit depletion as a paid model
# becoming unavailable on the Free Tier, returned as 404 rather than
# 402. Treat that as entitlement/billing exhaustion, not a missing
# model, so the retry loop can show credit/top-up guidance.
if any(p in error_msg for p in _BILLING_PATTERNS):
return result_fn(
FailoverReason.billing,
retryable=False,
should_rotate_credential=True,
should_fallback=True,
)
# OpenRouter policy-block 404 — distinct from "model not found".
# The model exists; the user's account privacy setting excludes the
# only endpoint serving it. Falling back to another provider won't
@ -973,7 +996,15 @@ def _classify_by_error_code(
should_rotate_credential=True,
)
if code_lower in {"insufficient_quota", "billing_not_active", "payment_required"}:
if code_lower in {
"insufficient_quota",
"billing_not_active",
"payment_required",
"insufficient_credits",
"no_usable_credits",
"balance_depleted",
"model_not_supported_on_free_tier",
}:
return result_fn(
FailoverReason.billing,
retryable=False,

View file

@ -802,16 +802,18 @@ def format_auth_error(error: Exception) -> str:
return f"{error} Run `hermes model` to re-authenticate."
if error.code == "subscription_required":
return (
"No active paid subscription found on Nous Portal. "
"Please purchase/activate a subscription, then retry."
)
if error.provider == "nous":
return _format_nous_entitlement_auth_error(error)
return "No active paid subscription found. Please purchase/activate a subscription, then retry."
if error.code == "insufficient_credits":
return (
"Subscription credits are exhausted. "
"Top up/renew credits in Nous Portal, then retry."
)
if error.provider == "nous":
return _format_nous_entitlement_auth_error(error)
return "Subscription credits are exhausted. Top up/renew credits, then retry."
if error.code in {"subscription_expired", "no_usable_credits", "account_missing"}:
if error.provider == "nous":
return _format_nous_entitlement_auth_error(error)
if error.code == "temporarily_unavailable":
return f"{error} Please retry in a few seconds."
@ -819,6 +821,25 @@ def format_auth_error(error: Exception) -> str:
return str(error)
def _format_nous_entitlement_auth_error(error: AuthError) -> str:
try:
from hermes_cli.nous_account import (
format_nous_portal_entitlement_message,
get_nous_portal_account_info,
)
account_info = get_nous_portal_account_info(force_fresh=True)
message = format_nous_portal_entitlement_message(
account_info,
capability="Nous model access",
)
if message:
return message
except Exception:
pass
return f"{error} Check credits or billing in Nous Portal, then retry."
def _token_fingerprint(token: Any) -> Optional[str]:
"""Return a short hash fingerprint for telemetry without leaking token bytes."""
if not isinstance(token, str):
@ -5627,6 +5648,8 @@ def _empty_nous_auth_status() -> Dict[str, Any]:
"access_expires_at": None,
"agent_key_expires_at": None,
"has_refresh_token": False,
"inference_credential_present": False,
"credential_source": None,
}
@ -5655,24 +5678,36 @@ def _snapshot_nous_pool_status() -> Dict[str, Any]:
return (agent_exp, access_exp, -priority)
entry = max(entries, key=_entry_sort_key)
access_token = (
getattr(entry, "access_token", None)
or getattr(entry, "runtime_api_key", "")
)
if not access_token:
runtime_key = getattr(entry, "runtime_api_key", None) or getattr(entry, "access_token", "")
if not runtime_key:
return _empty_nous_auth_status()
access_token = getattr(entry, "access_token", None)
auth_type = str(getattr(entry, "auth_type", "") or "").strip().lower()
refresh_token = getattr(entry, "refresh_token", None)
is_portal_oauth = bool(access_token) and (
auth_type.startswith("oauth") or bool(refresh_token)
)
label = getattr(entry, "label", "unknown")
portal_status_url = None
if is_portal_oauth:
portal_status_url = (
getattr(entry, "portal_base_url", None)
or DEFAULT_NOUS_PORTAL_URL
)
return {
"logged_in": True,
"portal_base_url": getattr(entry, "portal_base_url", None)
or getattr(entry, "base_url", None),
"logged_in": is_portal_oauth,
"portal_base_url": portal_status_url,
"inference_base_url": getattr(entry, "inference_base_url", None)
or getattr(entry, "runtime_base_url", None)
or getattr(entry, "base_url", None),
"access_token": access_token,
"access_token": access_token if is_portal_oauth else None,
"access_expires_at": getattr(entry, "expires_at", None),
"agent_key_expires_at": getattr(entry, "agent_key_expires_at", None),
"has_refresh_token": bool(getattr(entry, "refresh_token", None)),
"source": f"pool:{getattr(entry, 'label', 'unknown')}",
"has_refresh_token": bool(refresh_token),
"inference_credential_present": True,
"credential_source": f"pool:{label}",
"source": f"pool:{label}",
}
except Exception:
return _empty_nous_auth_status()
@ -5755,6 +5790,10 @@ def _compute_nous_auth_status() -> Dict[str, Any]:
"agent_key_expires_at": state.get("agent_key_expires_at"),
"has_refresh_token": bool(state.get("refresh_token")),
"access_token": state.get("access_token"),
"inference_credential_present": bool(
state.get("access_token") or state.get("agent_key")
),
"credential_source": "auth_store",
"source": "auth_store",
}
try:
@ -5772,6 +5811,8 @@ def _compute_nous_auth_status() -> Dict[str, Any]:
or refreshed_state.get("agent_key_expires_at")
or base_status.get("agent_key_expires_at"),
"has_refresh_token": bool(refreshed_state.get("refresh_token")),
"inference_credential_present": True,
"credential_source": "auth_store",
"source": f"runtime:{creds.get('source', 'portal')}",
"key_id": creds.get("key_id"),
}
@ -6283,6 +6324,7 @@ def _prompt_model_selection(
pricing: Optional[Dict[str, Dict[str, str]]] = None,
unavailable_models: Optional[List[str]] = None,
portal_url: str = "",
unavailable_message: str = "",
) -> Optional[str]:
"""Interactive model selection. Puts current_model first with a marker. Returns chosen model ID or None.
@ -6374,18 +6416,22 @@ def _prompt_model_selection(
choices.append(" Enter custom model name")
choices.append(" Skip (keep current)")
_upgrade_url = (portal_url or DEFAULT_NOUS_PORTAL_URL).rstrip("/")
unavailable_footer = unavailable_message.strip()
if not unavailable_footer and _unavailable:
unavailable_footer = f"Upgrade at {_upgrade_url} for paid models"
# Print the unavailable block BEFORE the menu via regular print().
# simple_term_menu pads title lines to terminal width (causes wrapping),
# so we keep the title minimal and use stdout for the static block.
# clear_screen=False means our printed output stays visible above.
_upgrade_url = (portal_url or DEFAULT_NOUS_PORTAL_URL).rstrip("/")
if _unavailable:
print(menu_title)
print()
for mid in _unavailable:
print(f"{_DIM} {_label(mid)}{_RESET}")
print()
print(f"{_DIM} ── Upgrade at {_upgrade_url} for paid models ──{_RESET}")
print(f"{_DIM} ── {unavailable_footer} ──{_RESET}")
print()
effective_title = "Available free models:"
else:
@ -6427,8 +6473,11 @@ def _prompt_model_selection(
if _unavailable:
_upgrade_url = (portal_url or DEFAULT_NOUS_PORTAL_URL).rstrip("/")
unavailable_footer = unavailable_message.strip() or (
f"Unavailable models (requires paid tier — upgrade at {_upgrade_url})"
)
print()
print(f" {_DIM}── Unavailable models (requires paid tier — upgrade at {_upgrade_url}) ──{_RESET}")
print(f" {_DIM}── {unavailable_footer} ──{_RESET}")
for mid in _unavailable:
print(f" {'':>{num_width}} {_DIM}{_label(mid)}{_RESET}")
print()
@ -7626,8 +7675,9 @@ def _nous_device_code_login(
portal_url = auth_state.get(
"portal_base_url", DEFAULT_NOUS_PORTAL_URL
).rstrip("/")
message = format_auth_error(exc)
print()
print("Your Nous Portal account does not have an active subscription.")
print(message)
print(f" Subscribe here: {portal_url}/billing")
print()
print("After subscribing, run `hermes model` again to finish setup.")
@ -7737,11 +7787,30 @@ def _login_nous(args, pconfig: ProviderConfig) -> None:
print()
unavailable_models: list = []
unavailable_message = ""
if model_ids:
pricing = get_pricing_for_provider("nous")
free_tier = check_nous_free_tier()
# Force fresh account data for model selection so recent credit
# purchases are reflected immediately.
free_tier = check_nous_free_tier(force_fresh=True)
_portal_for_recs = auth_state.get("portal_base_url", "")
if free_tier:
try:
from hermes_cli.nous_account import (
format_nous_portal_entitlement_message,
get_nous_portal_account_info,
)
_account_info = get_nous_portal_account_info(force_fresh=True)
unavailable_message = (
format_nous_portal_entitlement_message(
_account_info,
capability="paid Nous models",
)
or ""
)
except Exception:
unavailable_message = ""
# The Portal's freeRecommendedModels endpoint is the
# source of truth for what's free *right now*. Augment
# the curated list with anything new the Portal flags
@ -7768,11 +7837,12 @@ def _login_nous(args, pconfig: ProviderConfig) -> None:
model_ids, pricing=pricing,
unavailable_models=unavailable_models,
portal_url=_portal,
unavailable_message=unavailable_message,
)
elif unavailable_models:
_url = (_portal or DEFAULT_NOUS_PORTAL_URL).rstrip("/")
print("No free models currently available.")
print(f"Upgrade at {_url} to access paid models.")
print(unavailable_message or f"Upgrade at {_url} to access paid models.")
else:
print("No curated models available for Nous Portal.")
except Exception as exc:

View file

@ -2997,6 +2997,7 @@ def _model_flow_nous(config, current_model="", args=None):
"""Nous Portal provider: ensure logged in, then pick model."""
from hermes_cli.auth import (
get_provider_auth_state,
NOUS_INFERENCE_AUTH_MODE_LEGACY,
_prompt_model_selection,
_save_model_choice,
_update_config_for_provider,
@ -3092,8 +3093,21 @@ def _model_flow_nous(config, current_model="", args=None):
# Fetch live pricing (non-blocking — returns empty dict on failure)
pricing = get_pricing_for_provider("nous")
# Check if user is on free tier
free_tier = check_nous_free_tier()
# Force fresh account data for model selection so recent credit purchases
# are reflected immediately.
free_tier = check_nous_free_tier(force_fresh=True)
if not free_tier:
try:
refreshed_creds = resolve_nous_runtime_credentials(
min_key_ttl_seconds=5 * 60,
inference_auth_mode=NOUS_INFERENCE_AUTH_MODE_LEGACY,
)
if refreshed_creds:
creds = refreshed_creds
except Exception:
# Runtime inference has its own paid-entitlement recovery path; do
# not block model selection if this opportunistic remint fails.
pass
# Resolve portal URL early — needed both for upgrade links and for the
# freeRecommendedModels endpoint below.
@ -3115,7 +3129,24 @@ def _model_flow_nous(config, current_model="", args=None):
# newly-launched paid models surface in the picker too — independent
# of CLI release cadence.
unavailable_models: list[str] = []
unavailable_message = ""
if free_tier:
try:
from hermes_cli.nous_account import (
format_nous_portal_entitlement_message,
get_nous_portal_account_info,
)
_account_info = get_nous_portal_account_info(force_fresh=True)
unavailable_message = (
format_nous_portal_entitlement_message(
_account_info,
capability="paid Nous models",
)
or ""
)
except Exception:
unavailable_message = ""
model_ids, pricing = union_with_portal_free_recommendations(
model_ids, pricing, _nous_portal_url,
)
@ -3137,7 +3168,7 @@ def _model_flow_nous(config, current_model="", args=None):
from hermes_cli.auth import DEFAULT_NOUS_PORTAL_URL
_url = (_nous_portal_url or DEFAULT_NOUS_PORTAL_URL).rstrip("/")
print(f"Upgrade at {_url} to access paid models.")
print(unavailable_message or f"Upgrade at {_url} to access paid models.")
return
print(
@ -3150,6 +3181,7 @@ def _model_flow_nous(config, current_model="", args=None):
pricing=pricing,
unavailable_models=unavailable_models,
portal_url=_nous_portal_url,
unavailable_message=unavailable_message,
)
if selected:
_save_model_choice(selected)

View file

@ -518,9 +518,19 @@ def fetch_nous_account_tier(access_token: str, portal_base_url: str = "") -> dic
def is_nous_free_tier(account_info: dict[str, Any]) -> bool:
"""Return True if the account info indicates a free (unpaid) tier.
Checks ``subscription.monthly_charge == 0``. Returns False when
the field is missing or unparseable (assumes paid don't block users).
Prefer the Portal's explicit ``paid_service_access.allowed`` entitlement
decision. Legacy payloads fall back to ``subscription.monthly_charge == 0``.
Returns False when both signals are missing or unparseable.
"""
paid_access = account_info.get("paid_service_access")
if isinstance(paid_access, dict):
allowed = paid_access.get("allowed")
if isinstance(allowed, bool):
return not allowed
paid = paid_access.get("paid_access")
if isinstance(paid, bool):
return not paid
sub = account_info.get("subscription")
if not isinstance(sub, dict):
return False
@ -699,40 +709,28 @@ _FREE_TIER_CACHE_TTL: int = 180 # seconds (3 minutes)
_free_tier_cache: tuple[bool, float] | None = None # (result, timestamp)
def check_nous_free_tier() -> bool:
def check_nous_free_tier(*, force_fresh: bool = False) -> bool:
"""Check if the current Nous Portal user is on a free (unpaid) tier.
Results are cached for ``_FREE_TIER_CACHE_TTL`` seconds to avoid
hitting the Portal API on every call. The cache is short-lived so
that an account upgrade is reflected within a few minutes.
Returns False (assume paid) on any error never blocks paying users.
Returns True only when entitlement is known to be free. Unknown/error
states return False so this compatibility wrapper does not block users.
"""
global _free_tier_cache
now = time.monotonic()
if _free_tier_cache is not None:
if not force_fresh and _free_tier_cache is not None:
cached_result, cached_at = _free_tier_cache
if now - cached_at < _FREE_TIER_CACHE_TTL:
return cached_result
try:
from hermes_cli.auth import get_provider_auth_state, resolve_nous_runtime_credentials
from hermes_cli.nous_account import get_nous_portal_account_info
# Ensure we have a fresh token (triggers refresh if needed)
resolve_nous_runtime_credentials(min_key_ttl_seconds=60)
state = get_provider_auth_state("nous")
if not state:
_free_tier_cache = (False, now)
return False
access_token = state.get("access_token", "")
portal_url = state.get("portal_base_url", "")
if not access_token:
_free_tier_cache = (False, now)
return False
account_info = fetch_nous_account_tier(access_token, portal_url)
result = is_nous_free_tier(account_info)
account_info = get_nous_portal_account_info(force_fresh=force_fresh)
result = account_info.is_free_tier
_free_tier_cache = (result, now)
return result
except Exception:

678
hermes_cli/nous_account.py Normal file
View file

@ -0,0 +1,678 @@
"""Normalized Nous Portal account entitlement helpers."""
from __future__ import annotations
import hashlib
import json
import time
import urllib.request
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Literal, Optional
NousAccountInfoSource = Literal["jwt", "account_api", "inference_key", "none", "error"]
_ACCOUNT_INFO_CACHE_TTL = 60
_account_info_cache: tuple[str, float, "NousPortalAccountInfo"] | None = None
@dataclass(frozen=True)
class NousPortalSubscriptionInfo:
plan: Optional[str] = None
tier: Optional[int] = None
monthly_charge: Optional[float] = None
current_period_end: Optional[str] = None
credits_remaining: Optional[float] = None
rollover_credits: Optional[float] = None
@dataclass(frozen=True)
class NousPaidServiceAccessInfo:
allowed: Optional[bool] = None
paid_access: Optional[bool] = None
reason: Optional[str] = None
organisation_id: Optional[str] = None
effective_at_ms: Optional[int] = None
has_active_subscription: Optional[bool] = None
active_subscription_is_paid: Optional[bool] = None
subscription_tier: Optional[int] = None
subscription_monthly_charge: Optional[float] = None
subscription_credits_remaining: Optional[float] = None
purchased_credits_remaining: Optional[float] = None
total_usable_credits: Optional[float] = None
@dataclass(frozen=True)
class NousPortalAccountInfo:
logged_in: bool
source: NousAccountInfoSource
fresh: bool
user_id: Optional[str] = None
org_id: Optional[str] = None
client_id: Optional[str] = None
product_id: Optional[str] = None
nous_client: Optional[str] = None
portal_base_url: Optional[str] = None
inference_base_url: Optional[str] = None
inference_credential_present: bool = False
credential_source: Optional[str] = None
expires_at: Optional[datetime] = None
email: Optional[str] = None
privy_did: Optional[str] = None
subscription: Optional[NousPortalSubscriptionInfo] = None
paid_service_access: Optional[bool] = None
paid_service_access_info: Optional[NousPaidServiceAccessInfo] = None
raw_claims: Optional[dict[str, Any]] = None
raw_account: Optional[dict[str, Any]] = None
error: Optional[str] = None
@property
def is_paid(self) -> bool:
return self.paid_service_access is True
@property
def is_free_tier(self) -> bool:
return self.paid_service_access is False
@property
def tool_gateway_entitled(self) -> bool:
return self.paid_service_access is True
def nous_portal_billing_url(account_info: Optional[NousPortalAccountInfo] = None) -> str:
"""Return the billing URL for a normalized Nous account snapshot."""
try:
from hermes_cli.auth import DEFAULT_NOUS_PORTAL_URL
except Exception:
DEFAULT_NOUS_PORTAL_URL = "https://portal.nousresearch.com"
base = None
if account_info is not None:
base = account_info.portal_base_url
if not isinstance(base, str) or not base.strip():
base = DEFAULT_NOUS_PORTAL_URL
return f"{base.rstrip('/')}/billing"
def format_nous_portal_entitlement_message(
account_info: Optional[NousPortalAccountInfo],
*,
capability: str = "this feature",
include_refresh_hint: bool = True,
) -> Optional[str]:
"""Return user-facing guidance for a missing Nous paid entitlement.
``None`` means the account is known to have paid service access. The
message intentionally works from normalized entitlement fields rather than
subscription price alone: purchased credits without a subscription still
count as paid access, while a paid subscription with exhausted usable
credits does not.
"""
billing_url = nous_portal_billing_url(account_info)
if account_info is not None and account_info.paid_service_access is True:
return None
if account_info is None:
return (
f"Hermes could not verify your Nous Portal entitlement, so {capability} "
f"is unavailable. Run `hermes model` to refresh your login, or check "
f"billing at {billing_url}."
)
if not account_info.logged_in:
if account_info.inference_credential_present:
return (
f"Nous inference credentials are configured, but Hermes cannot verify "
f"your Nous Portal paid access for {capability}. Log in with "
f"`hermes model` to enable Portal-managed features. Billing and "
f"credits are managed at {billing_url}."
)
return (
f"Log in to Nous Portal to use {capability}: run `hermes model`. "
f"Billing and credits are managed at {billing_url}."
)
if account_info.paid_service_access is None:
detail = (
f"Hermes could not verify your Nous Portal paid access, so {capability} "
f"is unavailable."
)
if account_info.error:
detail += f" Account lookup failed: {account_info.error}."
if include_refresh_hint:
detail += " Run `hermes model` to refresh your session."
detail += f" Check billing at {billing_url}."
return detail
access = account_info.paid_service_access_info
reason = access.reason if access else None
if reason == "account_missing":
return (
f"Hermes could not find a Nous Portal account or organisation for this "
f"login, so {capability} is unavailable. Run `hermes model` to "
f"authenticate again; if the problem persists, contact Nous support."
)
if reason == "no_usable_credits" or account_info.paid_service_access is False:
message = _no_paid_access_message(account_info, capability, billing_url)
if include_refresh_hint and not account_info.fresh:
message += " If you recently bought credits, run `hermes model` to refresh Hermes."
return message
return (
f"Your Nous Portal account does not currently have paid service access, "
f"so {capability} is unavailable. Add credits or update billing at {billing_url}."
)
def _no_paid_access_message(
account_info: NousPortalAccountInfo,
capability: str,
billing_url: str,
) -> str:
access = account_info.paid_service_access_info
has_active_subscription = access.has_active_subscription if access else None
active_subscription_is_paid = access.active_subscription_is_paid if access else None
total_usable = access.total_usable_credits if access else None
subscription_credits = access.subscription_credits_remaining if access else None
purchased_credits = access.purchased_credits_remaining if access else None
if has_active_subscription and active_subscription_is_paid:
credit_detail = _credit_detail(total_usable, subscription_credits, purchased_credits)
return (
f"Your Nous Portal credits are exhausted{credit_detail}, so {capability} "
f"is unavailable. Top up or renew credits at {billing_url}."
)
if has_active_subscription and active_subscription_is_paid is False:
return (
f"Your current Nous Portal plan does not include paid service access, "
f"so {capability} is unavailable. Upgrade or add credits at {billing_url}."
)
if has_active_subscription is False:
credit_detail = _credit_detail(total_usable, subscription_credits, purchased_credits)
return (
f"Your Nous Portal account has no active subscription or usable credits"
f"{credit_detail}, so {capability} is unavailable. Subscribe or add credits "
f"at {billing_url}."
)
credit_detail = _credit_detail(total_usable, subscription_credits, purchased_credits)
return (
f"Your Nous Portal account has no usable paid credits{credit_detail}, so "
f"{capability} is unavailable. Add credits or update billing at {billing_url}."
)
def _credit_detail(
total_usable: Optional[float],
subscription_credits: Optional[float],
purchased_credits: Optional[float],
) -> str:
parts: list[str] = []
if total_usable is not None:
parts.append(f"usable ${total_usable:.2f}")
if subscription_credits is not None:
parts.append(f"subscription ${subscription_credits:.2f}")
if purchased_credits is not None:
parts.append(f"purchased ${purchased_credits:.2f}")
if not parts:
return ""
return f" ({', '.join(parts)})"
def reset_nous_portal_account_info_cache() -> None:
"""Clear the short-lived account-info cache used by tests."""
global _account_info_cache
_account_info_cache = None
def get_nous_portal_account_info(
*,
force_fresh: bool = False,
min_jwt_ttl_seconds: int = 60,
) -> NousPortalAccountInfo:
"""Return normalized Nous Portal account entitlement information.
By default, a valid unexpired OAuth access JWT is used as a low-latency
local account snapshot. ``force_fresh=True`` always calls
``/api/oauth/account`` and bypasses the short-lived cache. JWT claims are
decoded locally for UX gating only; server APIs remain authoritative.
"""
try:
from hermes_cli.auth import get_provider_auth_state
state = get_provider_auth_state("nous") or {}
except Exception as exc:
return _error_info(error=exc, logged_in=False)
access_token = state.get("access_token")
portal_base_url = _portal_base_url(state)
if not isinstance(access_token, str) or not access_token.strip():
pool_oauth_info = _info_from_oauth_pool(
force_fresh=force_fresh,
min_jwt_ttl_seconds=min_jwt_ttl_seconds,
portal_base_url=portal_base_url,
)
if pool_oauth_info is not None:
return pool_oauth_info
pool_info = _info_from_inference_key_pool(portal_base_url)
if pool_info is not None:
return pool_info
return NousPortalAccountInfo(
logged_in=False,
source="none",
fresh=False,
portal_base_url=portal_base_url,
)
if not force_fresh:
jwt_info = _info_from_valid_jwt(
access_token,
state=state,
portal_base_url=portal_base_url,
min_jwt_ttl_seconds=min_jwt_ttl_seconds,
)
if jwt_info is not None:
return jwt_info
return _fresh_account_info(
state=state,
force_fresh=force_fresh,
portal_base_url=portal_base_url,
)
def _fresh_account_info(
*,
state: dict[str, Any],
force_fresh: bool,
portal_base_url: Optional[str],
) -> NousPortalAccountInfo:
global _account_info_cache
try:
from hermes_cli.auth import get_provider_auth_state, resolve_nous_access_token
access_token = resolve_nous_access_token()
refreshed_state = get_provider_auth_state("nous") or state
portal_base_url = _portal_base_url(refreshed_state) or portal_base_url
cache_key = _cache_key(access_token, portal_base_url)
if not force_fresh and _account_info_cache is not None:
cached_key, cached_at, cached_info = _account_info_cache
if cached_key == cache_key and (time.monotonic() - cached_at) < _ACCOUNT_INFO_CACHE_TTL:
return cached_info
payload = _fetch_nous_account_info(access_token, portal_base_url)
if not payload:
return _error_info(
error="empty_account_response",
logged_in=True,
portal_base_url=portal_base_url,
)
if isinstance(payload.get("error"), str):
return _error_info(
error=payload.get("error") or "account_response_error",
logged_in=True,
portal_base_url=portal_base_url,
raw_account=payload,
)
info = _info_from_account_payload(
payload,
state=refreshed_state,
portal_base_url=portal_base_url,
)
_account_info_cache = (cache_key, time.monotonic(), info)
return info
except Exception as exc:
return _error_info(
error=exc,
logged_in=bool(state.get("access_token")),
portal_base_url=portal_base_url,
)
def _info_from_inference_key_pool(
portal_base_url: Optional[str],
) -> Optional[NousPortalAccountInfo]:
"""Return an explicit unknown-entitlement snapshot for opaque Nous keys."""
try:
entry = _select_nous_pool_entry()
if entry is None:
return None
runtime_key = getattr(entry, "runtime_api_key", None) or getattr(entry, "access_token", "")
if not isinstance(runtime_key, str) or not runtime_key.strip():
return None
return NousPortalAccountInfo(
logged_in=False,
source="inference_key",
fresh=False,
portal_base_url=(
getattr(entry, "portal_base_url", None)
or portal_base_url
),
inference_base_url=(
getattr(entry, "inference_base_url", None)
or getattr(entry, "runtime_base_url", None)
or getattr(entry, "base_url", None)
),
inference_credential_present=True,
credential_source=f"pool:{getattr(entry, 'label', 'unknown')}",
error="portal_oauth_missing",
)
except Exception:
return None
def _info_from_oauth_pool(
*,
force_fresh: bool,
min_jwt_ttl_seconds: int,
portal_base_url: Optional[str],
) -> Optional[NousPortalAccountInfo]:
try:
entry = _select_nous_pool_entry()
except Exception:
return None
if entry is None or not _pool_entry_is_portal_oauth(entry):
return None
access_token = getattr(entry, "access_token", None)
if not isinstance(access_token, str) or not access_token.strip():
return None
entry_portal_url = (
getattr(entry, "portal_base_url", None)
or portal_base_url
)
state = {
"access_token": access_token,
"client_id": getattr(entry, "client_id", None),
"inference_base_url": (
getattr(entry, "inference_base_url", None)
or getattr(entry, "runtime_base_url", None)
or getattr(entry, "base_url", None)
),
"agent_key": getattr(entry, "agent_key", None),
"credential_source": f"pool:{getattr(entry, 'label', 'unknown')}",
}
if not force_fresh:
jwt_info = _info_from_valid_jwt(
access_token,
state=state,
portal_base_url=entry_portal_url,
min_jwt_ttl_seconds=min_jwt_ttl_seconds,
)
if jwt_info is not None:
return jwt_info
try:
payload = _fetch_nous_account_info(access_token, entry_portal_url)
except Exception as exc:
return _error_info(
error=exc,
logged_in=True,
portal_base_url=entry_portal_url,
)
if not payload:
return _error_info(
error="empty_account_response",
logged_in=True,
portal_base_url=entry_portal_url,
)
if isinstance(payload.get("error"), str):
return _error_info(
error=payload.get("error") or "account_response_error",
logged_in=True,
portal_base_url=entry_portal_url,
raw_account=payload,
)
return _info_from_account_payload(
payload,
state=state,
portal_base_url=entry_portal_url,
)
def _select_nous_pool_entry() -> Optional[Any]:
from agent.credential_pool import load_pool
pool = load_pool("nous")
if not pool or not pool.has_credentials():
return None
entries = list(pool.entries())
if not entries:
return None
def _entry_sort_key(entry: Any) -> tuple[float, float, int]:
agent_exp = _parse_iso_timestamp(getattr(entry, "agent_key_expires_at", None)) or 0.0
access_exp = _parse_iso_timestamp(getattr(entry, "expires_at", None)) or 0.0
priority = int(getattr(entry, "priority", 0) or 0)
return (agent_exp, access_exp, -priority)
return max(entries, key=_entry_sort_key)
def _pool_entry_is_portal_oauth(entry: Any) -> bool:
access_token = getattr(entry, "access_token", None)
if not isinstance(access_token, str) or not access_token.strip():
return False
auth_type = str(getattr(entry, "auth_type", "") or "").strip().lower()
refresh_token = getattr(entry, "refresh_token", None)
return auth_type.startswith("oauth") or bool(refresh_token)
def _fetch_nous_account_info(
access_token: str,
portal_base_url: Optional[str] = None,
) -> dict[str, Any]:
base = (portal_base_url or "https://portal.nousresearch.com").rstrip("/")
url = f"{base}/api/oauth/account"
headers = {
"Authorization": f"Bearer {access_token}",
"Accept": "application/json",
}
req = urllib.request.Request(url, headers=headers)
with urllib.request.urlopen(req, timeout=8) as resp:
payload = json.loads(resp.read().decode())
return payload if isinstance(payload, dict) else {}
def _info_from_valid_jwt(
token: str,
*,
state: dict[str, Any],
portal_base_url: Optional[str],
min_jwt_ttl_seconds: int,
) -> Optional[NousPortalAccountInfo]:
try:
from hermes_cli.auth import _decode_jwt_claims
except Exception:
return None
claims = _decode_jwt_claims(token)
if not claims:
return None
exp = _coerce_float(claims.get("exp"))
if exp is None or exp <= time.time() + max(0, int(min_jwt_ttl_seconds)):
return None
paid_access = _coerce_bool(claims.get("paid_access"))
subscription_tier = _coerce_int(claims.get("subscription_tier"))
access_info = NousPaidServiceAccessInfo(
allowed=paid_access,
paid_access=paid_access,
organisation_id=_coerce_str(claims.get("org_id")),
subscription_tier=subscription_tier,
)
return NousPortalAccountInfo(
logged_in=True,
source="jwt",
fresh=False,
user_id=_coerce_str(claims.get("sub")),
org_id=_coerce_str(claims.get("org_id")),
client_id=_coerce_str(claims.get("client_id") or state.get("client_id")),
product_id=_coerce_str(claims.get("product_id")),
nous_client=_coerce_str(claims.get("nous_client")),
portal_base_url=portal_base_url,
inference_base_url=_coerce_str(state.get("inference_base_url")),
inference_credential_present=True,
credential_source=_coerce_str(state.get("credential_source")) or "auth_store",
expires_at=datetime.fromtimestamp(exp, tz=timezone.utc),
paid_service_access=paid_access,
paid_service_access_info=access_info,
raw_claims=dict(claims),
)
def _info_from_account_payload(
payload: dict[str, Any],
*,
state: dict[str, Any],
portal_base_url: Optional[str],
) -> NousPortalAccountInfo:
user = payload.get("user") if isinstance(payload.get("user"), dict) else {}
organisation = (
payload.get("organisation")
if isinstance(payload.get("organisation"), dict)
else {}
)
subscription = _subscription_from_payload(payload.get("subscription"))
access = _paid_service_access_from_payload(payload.get("paid_service_access"))
paid_access = access.allowed if access else None
if paid_access is None and access is not None:
paid_access = access.paid_access
return NousPortalAccountInfo(
logged_in=True,
source="account_api",
fresh=True,
org_id=_coerce_str(organisation.get("id")) or (access.organisation_id if access else None),
client_id=_coerce_str(state.get("client_id")),
portal_base_url=portal_base_url,
inference_base_url=_coerce_str(state.get("inference_base_url")),
inference_credential_present=bool(state.get("access_token") or state.get("agent_key")),
credential_source=_coerce_str(state.get("credential_source")) or "auth_store",
email=_coerce_str(user.get("email")),
privy_did=_coerce_str(user.get("privy_did")),
subscription=subscription,
paid_service_access=paid_access,
paid_service_access_info=access,
raw_account=dict(payload),
)
def _subscription_from_payload(value: Any) -> Optional[NousPortalSubscriptionInfo]:
if not isinstance(value, dict):
return None
return NousPortalSubscriptionInfo(
plan=_coerce_str(value.get("plan")),
tier=_coerce_int(value.get("tier")),
monthly_charge=_coerce_float(value.get("monthly_charge")),
current_period_end=_coerce_str(value.get("current_period_end")),
credits_remaining=_coerce_float(value.get("credits_remaining")),
rollover_credits=_coerce_float(value.get("rollover_credits")),
)
def _paid_service_access_from_payload(value: Any) -> Optional[NousPaidServiceAccessInfo]:
if not isinstance(value, dict):
return None
allowed = _coerce_bool(value.get("allowed"))
paid_access = _coerce_bool(value.get("paid_access"))
return NousPaidServiceAccessInfo(
allowed=allowed,
paid_access=paid_access,
reason=_coerce_str(value.get("reason")),
organisation_id=_coerce_str(value.get("organisation_id")),
effective_at_ms=_coerce_int(value.get("effective_at_ms")),
has_active_subscription=_coerce_bool(value.get("has_active_subscription")),
active_subscription_is_paid=_coerce_bool(value.get("active_subscription_is_paid")),
subscription_tier=_coerce_int(value.get("subscription_tier")),
subscription_monthly_charge=_coerce_float(value.get("subscription_monthly_charge")),
subscription_credits_remaining=_coerce_float(value.get("subscription_credits_remaining")),
purchased_credits_remaining=_coerce_float(value.get("purchased_credits_remaining")),
total_usable_credits=_coerce_float(value.get("total_usable_credits")),
)
def _error_info(
*,
error: object,
logged_in: bool,
portal_base_url: Optional[str] = None,
raw_account: Optional[dict[str, Any]] = None,
) -> NousPortalAccountInfo:
return NousPortalAccountInfo(
logged_in=logged_in,
source="error",
fresh=False,
portal_base_url=portal_base_url,
raw_account=raw_account,
error=str(error),
)
def _portal_base_url(state: dict[str, Any]) -> Optional[str]:
value = state.get("portal_base_url")
if not isinstance(value, str) or not value.strip():
return None
return value.strip().rstrip("/")
def _cache_key(access_token: str, portal_base_url: Optional[str]) -> str:
digest = hashlib.sha256(access_token.encode("utf-8")).hexdigest()
return f"{portal_base_url or ''}:{digest}"
def _parse_iso_timestamp(value: Any) -> Optional[float]:
if not isinstance(value, str) or not value:
return None
text = value.strip()
if text.endswith("Z"):
text = text[:-1] + "+00:00"
try:
return datetime.fromisoformat(text).timestamp()
except Exception:
return None
def _coerce_str(value: Any) -> Optional[str]:
if isinstance(value, str) and value:
return value
return None
def _coerce_bool(value: Any) -> Optional[bool]:
return value if isinstance(value, bool) else None
def _coerce_int(value: Any) -> Optional[int]:
if isinstance(value, bool):
return None
try:
if value is None:
return None
return int(value)
except (TypeError, ValueError):
return None
def _coerce_float(value: Any) -> Optional[float]:
if isinstance(value, bool):
return None
try:
if value is None:
return None
return float(value)
except (TypeError, ValueError):
return None

View file

@ -6,8 +6,8 @@ from dataclasses import dataclass
from pathlib import Path
from typing import Dict, Iterable, Optional, Set
from hermes_cli.auth import get_nous_auth_status
from hermes_cli.config import get_env_value, load_config
from hermes_cli.nous_account import NousPortalAccountInfo, get_nous_portal_account_info
from tools.managed_tool_gateway import is_managed_tool_gateway_ready
from utils import is_truthy_value
from tools.tool_backend_helpers import (
@ -53,6 +53,7 @@ class NousSubscriptionFeatures:
nous_auth_present: bool
provider_is_nous: bool
features: Dict[str, NousFeatureState]
account_info: Optional[NousPortalAccountInfo] = None
@property
def web(self) -> NousFeatureState:
@ -235,12 +236,16 @@ def get_nous_subscription_features(
provider_is_nous = str(model_cfg.get("provider") or "").strip().lower() == "nous"
try:
nous_status = get_nous_auth_status()
account_info = get_nous_portal_account_info()
except Exception:
nous_status = {}
account_info = None
managed_tools_flag = managed_nous_tools_enabled()
nous_auth_present = bool(nous_status.get("logged_in"))
managed_tools_flag = bool(
account_info
and account_info.logged_in
and account_info.paid_service_access is True
)
nous_auth_present = bool(account_info and account_info.logged_in)
subscribed = provider_is_nous or nous_auth_present
web_tool_enabled = _toolset_enabled(config, "web")
@ -483,6 +488,7 @@ def get_nous_subscription_features(
nous_auth_present=nous_auth_present,
provider_is_nous=provider_is_nous,
features=features,
account_info=account_info,
)

View file

@ -16,6 +16,10 @@ from hermes_cli.auth import AuthError, resolve_provider
from hermes_cli.colors import Colors, color
from hermes_cli.config import get_env_path, get_env_value, get_hermes_home, load_config
from hermes_cli.models import provider_label
from hermes_cli.nous_account import (
format_nous_portal_entitlement_message,
get_nous_portal_account_info,
)
from hermes_cli.nous_subscription import get_nous_subscription_features
from hermes_cli.runtime_provider import resolve_requested_provider
from hermes_constants import OPENROUTER_MODELS_URL
@ -193,26 +197,57 @@ def show_status(args):
qwen_status = {}
minimax_status = {}
nous_logged_in = bool(nous_status.get("logged_in"))
nous_account_info = None
if (
nous_status.get("logged_in")
or nous_status.get("access_token")
or nous_status.get("portal_base_url")
or nous_status.get("inference_credential_present")
or nous_status.get("error_code")
):
try:
nous_account_info = get_nous_portal_account_info()
except Exception:
nous_account_info = None
nous_logged_in = bool(
nous_status.get("logged_in")
or (nous_account_info and nous_account_info.logged_in)
)
nous_inference_present = bool(
nous_status.get("inference_credential_present")
or (nous_account_info and nous_account_info.inference_credential_present)
)
nous_error = nous_status.get("error")
nous_label = "logged in" if nous_logged_in else "not logged in (run: hermes auth add nous --type oauth)"
if nous_logged_in:
nous_label = "logged in"
elif nous_inference_present:
nous_label = "not logged in (Nous inference key configured)"
else:
nous_label = "not logged in (run: hermes auth add nous --type oauth)"
print(
f" {'Nous Portal':<12} {check_mark(nous_logged_in)} "
f"{nous_label}"
)
portal_url = nous_status.get("portal_base_url") or "(unknown)"
inference_url = (
nous_status.get("inference_base_url")
or (nous_account_info.inference_base_url if nous_account_info else None)
)
access_exp = _format_iso_timestamp(nous_status.get("access_expires_at"))
key_exp = _format_iso_timestamp(nous_status.get("agent_key_expires_at"))
refresh_label = "yes" if nous_status.get("has_refresh_token") else "no"
if nous_logged_in or portal_url != "(unknown)" or nous_error:
print(f" Portal URL: {portal_url}")
if nous_inference_present and inference_url:
print(f" Inference: {inference_url}")
if nous_logged_in or nous_status.get("access_expires_at"):
print(f" Access exp: {access_exp}")
if nous_logged_in or nous_status.get("agent_key_expires_at"):
if nous_logged_in or nous_inference_present or nous_status.get("agent_key_expires_at"):
print(f" Key exp: {key_exp}")
if nous_logged_in or nous_status.get("has_refresh_token"):
print(f" Refresh: {refresh_label}")
if nous_error and not nous_logged_in:
if nous_error:
print(f" Error: {nous_error}")
codex_logged_in = bool(codex_status.get("logged_in"))
@ -303,18 +338,18 @@ def show_status(args):
else:
state = "not configured"
print(f" {feature.label:<15} {check_mark(feature.available or feature.active or feature.managed_by_nous)} {state}")
elif nous_logged_in:
# Logged into Nous but on the free tier — show upgrade nudge
elif nous_logged_in or nous_inference_present:
# Nous OAuth without entitlement, or an opaque inference key without
# Portal account information, cannot enable the Tool Gateway.
print()
print(color("◆ Nous Tool Gateway", Colors.CYAN, Colors.BOLD))
print(" Your free-tier Nous account does not include Tool Gateway access.")
print(" Upgrade your subscription to unlock managed web, image, TTS, and browser tools.")
try:
portal_url = nous_status.get("portal_base_url", "").rstrip("/")
if portal_url:
print(f" Upgrade: {portal_url}")
except Exception:
pass
message = format_nous_portal_entitlement_message(
nous_account_info,
capability="managed web, image, TTS, browser, and Modal tools",
)
if message:
for line in message.splitlines():
print(f" {line}")
# =========================================================================
# API-Key Providers

View file

@ -28,6 +28,7 @@ from hermes_cli.nous_subscription import (
apply_nous_managed_defaults,
get_nous_subscription_features,
)
from hermes_cli.nous_account import format_nous_portal_entitlement_message
from tools.tool_backend_helpers import fal_key_is_configured, managed_nous_tools_enabled
from utils import base_url_hostname, is_truthy_value
@ -1855,6 +1856,20 @@ def _visible_providers(cat: dict, config: dict) -> list[dict]:
return visible
def _hidden_nous_gateway_message(cat: dict, config: dict, capability: str) -> str:
"""Return a reason when a category's Nous provider is hidden."""
if managed_nous_tools_enabled():
return ""
if not any(p.get("managed_nous_feature") for p in cat.get("providers", [])):
return ""
features = get_nous_subscription_features(config)
message = format_nous_portal_entitlement_message(
features.account_info,
capability=capability,
)
return message or ""
_POST_SETUP_INSTALLED: dict = {
# post_setup_key -> predicate(): True when the install side-effect
# is already satisfied. Used by `_toolset_needs_configuration_prompt`
@ -1955,6 +1970,11 @@ def _configure_tool_category(ts_key: str, cat: dict, config: dict):
icon = cat.get("icon", "")
name = cat["name"]
providers = _visible_providers(cat, config)
hidden_nous_message = _hidden_nous_gateway_message(
cat,
config,
f"the Nous Subscription provider for {name}",
)
# Check Python version requirement
if cat.get("requires_python"):
@ -1975,6 +1995,9 @@ def _configure_tool_category(ts_key: str, cat: dict, config: dict):
# For single-provider tools, show a note if available
if cat.get("setup_note"):
_print_info(f" {cat['setup_note']}")
if hidden_nous_message:
for line in hidden_nous_message.splitlines():
_print_warning(f" {line}")
_configure_provider(provider, config)
else:
# Multiple providers - let user choose
@ -1984,6 +2007,9 @@ def _configure_tool_category(ts_key: str, cat: dict, config: dict):
print(color(f" --- {icon} {name} - {title} ---", Colors.CYAN))
if cat.get("setup_note"):
_print_info(f" {cat['setup_note']}")
if hidden_nous_message:
for line in hidden_nous_message.splitlines():
_print_warning(f" {line}")
print()
# Plain text labels only (no ANSI codes in menu items)
@ -2410,8 +2436,17 @@ def _configure_provider(provider: dict, config: dict):
if provider.get("requires_nous_auth"):
features = get_nous_subscription_features(config)
if not features.nous_auth_present:
_print_warning(" Nous Subscription is only available after logging into Nous Portal.")
entitled = bool(
features.account_info and features.account_info.paid_service_access is True
)
if not features.nous_auth_present or not entitled:
message = format_nous_portal_entitlement_message(
features.account_info,
capability=f"{provider.get('name', 'Nous Subscription')}",
)
_print_warning(
f" {message or 'Nous Subscription is only available after logging into Nous Portal.'}"
)
return
# Set TTS provider in config if applicable
@ -2680,15 +2715,26 @@ def _configure_tool_category_for_reconfig(ts_key: str, cat: dict, config: dict):
icon = cat.get("icon", "")
name = cat["name"]
providers = _visible_providers(cat, config)
hidden_nous_message = _hidden_nous_gateway_message(
cat,
config,
f"the Nous Subscription provider for {name}",
)
if len(providers) == 1:
provider = providers[0]
print()
print(color(f" --- {icon} {name} ({provider['name']}) ---", Colors.CYAN))
if hidden_nous_message:
for line in hidden_nous_message.splitlines():
_print_warning(f" {line}")
_reconfigure_provider(provider, config)
else:
print()
print(color(f" --- {icon} {name} - Choose a provider ---", Colors.CYAN))
if hidden_nous_message:
for line in hidden_nous_message.splitlines():
_print_warning(f" {line}")
print()
provider_choices = []
@ -2719,8 +2765,17 @@ def _reconfigure_provider(provider: dict, config: dict):
if provider.get("requires_nous_auth"):
features = get_nous_subscription_features(config)
if not features.nous_auth_present:
_print_warning(" Nous Subscription is only available after logging into Nous Portal.")
entitled = bool(
features.account_info and features.account_info.paid_service_access is True
)
if not features.nous_auth_present or not entitled:
message = format_nous_portal_entitlement_message(
features.account_info,
capability=f"{provider.get('name', 'Nous Subscription')}",
)
_print_warning(
f" {message or 'Nous Subscription is only available after logging into Nous Portal.'}"
)
return
if provider.get("tts_provider"):

View file

@ -196,9 +196,13 @@ def _raise_web_backend_configuration_error() -> None:
)
if _wt.managed_nous_tools_enabled():
message += (
" With your Nous subscription you can also use the Tool Gateway "
" With your Nous subscription you can also use the Tool Gateway. "
"run `hermes tools` and select Nous Subscription as the web provider."
)
else:
message += " " + _wt.nous_tool_gateway_unavailable_message(
"managed Firecrawl web tools",
)
raise ValueError(message)

View file

@ -2847,7 +2847,12 @@ class AIAgent:
return True
def _try_refresh_nous_client_credentials(self, *, force: bool = True) -> bool:
def _try_refresh_nous_client_credentials(
self,
*,
force: bool = True,
inference_auth_mode: str | None = None,
) -> bool:
if self.api_mode != "chat_completions" or self.provider != "nous":
return False
@ -2858,14 +2863,15 @@ class AIAgent:
resolve_nous_runtime_credentials,
)
selected_auth_mode = inference_auth_mode or (
NOUS_INFERENCE_AUTH_MODE_LEGACY
if force
else NOUS_INFERENCE_AUTH_MODE_AUTO
)
creds = resolve_nous_runtime_credentials(
min_key_ttl_seconds=max(60, int(os.getenv("HERMES_NOUS_MIN_KEY_TTL_SECONDS", "1800"))),
timeout_seconds=float(os.getenv("HERMES_NOUS_TIMEOUT_SECONDS", "15")),
inference_auth_mode=(
NOUS_INFERENCE_AUTH_MODE_LEGACY
if force
else NOUS_INFERENCE_AUTH_MODE_AUTO
),
inference_auth_mode=selected_auth_mode,
)
except Exception as exc:
logger.debug("Nous credential refresh failed: %s", exc)

View file

@ -992,6 +992,47 @@ class TestAuxiliaryPoolAwareness:
assert stale_client.chat.completions.create.call_count == 1
assert fresh_client.chat.completions.create.call_count == 1
def test_call_llm_refreshes_nous_after_free_tier_block_when_account_paid(self):
from hermes_cli.nous_account import NousPortalAccountInfo
class _Payment404(Exception):
status_code = 404
stale_client = MagicMock()
stale_client.base_url = "https://inference-api.nousresearch.com/v1"
stale_client.chat.completions.create.side_effect = _Payment404(
"model_not_supported_on_free_tier: model is not available on the free tier"
)
fresh_client = MagicMock()
fresh_client.base_url = "https://inference-api.nousresearch.com/v1"
fresh_client.chat.completions.create.return_value = {"ok": True}
with (
patch("agent.auxiliary_client._resolve_task_provider_model", return_value=("nous", "nous-model", None, None, None)),
patch("agent.auxiliary_client._get_cached_client", return_value=(stale_client, "nous-model")),
patch("agent.auxiliary_client.OpenAI", return_value=fresh_client),
patch("agent.auxiliary_client._validate_llm_response", side_effect=lambda resp, _task: resp),
patch("agent.auxiliary_client._resolve_nous_runtime_api", return_value=("fresh-agent-key", "https://inference-api.nousresearch.com/v1")),
patch(
"hermes_cli.nous_account.get_nous_portal_account_info",
return_value=NousPortalAccountInfo(
logged_in=True,
source="account_api",
fresh=True,
paid_service_access=True,
),
),
):
result = call_llm(
task="compression",
messages=[{"role": "user", "content": "hi"}],
)
assert result == {"ok": True}
assert stale_client.chat.completions.create.call_count == 1
assert fresh_client.chat.completions.create.call_count == 1
@pytest.mark.asyncio
async def test_async_call_llm_retries_nous_after_401(self):
class _Auth401(Exception):
@ -1021,6 +1062,48 @@ class TestAuxiliaryPoolAwareness:
assert stale_client.chat.completions.create.await_count == 1
assert fresh_async_client.chat.completions.create.await_count == 1
@pytest.mark.asyncio
async def test_async_call_llm_refreshes_nous_after_free_tier_block_when_account_paid(self):
from hermes_cli.nous_account import NousPortalAccountInfo
class _Payment404(Exception):
status_code = 404
stale_client = MagicMock()
stale_client.base_url = "https://inference-api.nousresearch.com/v1"
stale_client.chat.completions.create = AsyncMock(side_effect=_Payment404(
"model_not_supported_on_free_tier: model is not available on the free tier"
))
fresh_async_client = MagicMock()
fresh_async_client.base_url = "https://inference-api.nousresearch.com/v1"
fresh_async_client.chat.completions.create = AsyncMock(return_value={"ok": True})
with (
patch("agent.auxiliary_client._resolve_task_provider_model", return_value=("nous", "nous-model", None, None, None)),
patch("agent.auxiliary_client._get_cached_client", return_value=(stale_client, "nous-model")),
patch("agent.auxiliary_client._to_async_client", return_value=(fresh_async_client, "nous-model")),
patch("agent.auxiliary_client._validate_llm_response", side_effect=lambda resp, _task: resp),
patch("agent.auxiliary_client._resolve_nous_runtime_api", return_value=("fresh-agent-key", "https://inference-api.nousresearch.com/v1")),
patch(
"hermes_cli.nous_account.get_nous_portal_account_info",
return_value=NousPortalAccountInfo(
logged_in=True,
source="account_api",
fresh=True,
paid_service_access=True,
),
),
):
result = await async_call_llm(
task="session_search",
messages=[{"role": "user", "content": "hi"}],
)
assert result == {"ok": True}
assert stale_client.chat.completions.create.await_count == 1
assert fresh_async_client.chat.completions.create.await_count == 1
def test_cached_gmi_client_keeps_explicit_slash_model_override(self):
import agent.auxiliary_client as aux
@ -1076,6 +1159,19 @@ class TestIsPaymentError:
exc.status_code = 429
assert _is_payment_error(exc) is True
def test_404_free_tier_model_block_is_payment(self):
exc = Exception(
"Model 'gpt-5' is not available on the Free Tier. "
"Upgrade at https://portal.nousresearch.com or pick a free model."
)
exc.status_code = 404
assert _is_payment_error(exc) is True
def test_404_generic_not_found_is_not_payment(self):
exc = Exception("Not Found")
exc.status_code = 404
assert _is_payment_error(exc) is False
def test_429_without_credits_message_is_not_payment(self):
"""Normal rate limits should NOT be treated as payment errors."""
exc = Exception("Rate limit exceeded, try again in 2 seconds")

View file

@ -254,12 +254,51 @@ class TestClassifyApiError:
assert result.reason == FailoverReason.billing
assert result.retryable is False
def test_402_out_of_funds_billing(self):
e = MockAPIError(
"Payment Required",
status_code=402,
body={
"status": 402,
"message": (
"Your API key has run out of funds. Please go visit the "
"portal to sort that out: https://portal.nousresearch.com"
),
},
)
result = classify_api_error(e)
assert result.reason == FailoverReason.billing
assert result.retryable is False
def test_402_transient_usage_limit(self):
e = MockAPIError("usage limit exceeded, try again later", status_code=402)
result = classify_api_error(e)
assert result.reason == FailoverReason.rate_limit
assert result.retryable is True
def test_403_plan_entitlement_billing(self):
e = MockAPIError("This plan does not include the requested model", status_code=403)
result = classify_api_error(e)
assert result.reason == FailoverReason.billing
assert result.retryable is False
def test_404_free_tier_model_block_is_billing(self):
e = MockAPIError(
"Not Found",
status_code=404,
body={
"status": 404,
"message": (
"Model 'gpt-5' is not available on the Free Tier. "
"Upgrade at https://portal.nousresearch.com or pick a free model."
),
},
)
result = classify_api_error(e, provider="nous", model="gpt-5")
assert result.reason == FailoverReason.billing
assert result.retryable is False
assert result.should_fallback is True
# ── Rate limit ──
def test_429_rate_limit(self):
@ -753,6 +792,19 @@ class TestClassifyApiError:
result = classify_api_error(e)
assert result.reason == FailoverReason.context_overflow
def test_error_code_model_not_supported_on_free_tier_is_billing(self):
e = MockAPIError(
"Model unavailable",
body={
"error": {
"code": "model_not_supported_on_free_tier",
"message": "Model 'gpt-5' is not available on the Free Tier.",
}
},
)
result = classify_api_error(e, provider="nous", model="gpt-5")
assert result.reason == FailoverReason.billing
# ── Message-only patterns (no status code) ──
def test_message_billing_pattern(self):
@ -760,6 +812,11 @@ class TestClassifyApiError:
result = classify_api_error(e)
assert result.reason == FailoverReason.billing
def test_message_free_tier_model_block_is_billing(self):
e = Exception("Model 'gpt-5' is not available on the Free Tier.")
result = classify_api_error(e, provider="nous", model="gpt-5")
assert result.reason == FailoverReason.billing
def test_message_rate_limit_pattern(self):
e = Exception("rate limit reached for this model")
result = classify_api_error(e)

View file

@ -667,6 +667,42 @@ def test_get_nous_auth_status_checks_credential_pool(tmp_path, monkeypatch):
assert "example.com" in str(status.get("portal_base_url", ""))
def test_get_nous_auth_status_pool_opaque_key_is_not_portal_login(tmp_path, monkeypatch):
from hermes_cli.auth import get_nous_auth_status, invalidate_nous_auth_status_cache
hermes_home = tmp_path / "hermes"
hermes_home.mkdir(parents=True, exist_ok=True)
(hermes_home / "auth.json").write_text(json.dumps({
"version": 1, "providers": {},
}))
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
invalidate_nous_auth_status_cache()
from agent.credential_pool import PooledCredential, load_pool
pool = load_pool("nous")
entry = PooledCredential.from_dict("nous", {
"access_token": "",
"agent_key": "opaque-agent-key",
"agent_key_expires_at": "2099-01-01T00:00:00+00:00",
"label": "manual opaque key",
"auth_type": "api_key",
"source": "manual",
"base_url": "https://inference.example.com/v1",
"inference_base_url": "https://inference.example.com/v1",
})
pool.add_entry(entry)
status = get_nous_auth_status()
assert status["logged_in"] is False
assert status["inference_credential_present"] is True
assert status["credential_source"] == "pool:manual opaque key"
assert status.get("access_token") is None
assert status.get("portal_base_url") is None
assert status.get("inference_base_url") == "https://inference.example.com/v1"
invalidate_nous_auth_status_cache()
def test_get_nous_auth_status_auth_store_fallback(tmp_path, monkeypatch):
"""get_nous_auth_status() falls back to auth store when credential
pool is empty.
@ -1023,12 +1059,19 @@ class TestLoginNousSkipKeepsCurrent:
lambda *a, **kw: prompt_returns,
)
monkeypatch.setattr(models_mod, "get_pricing_for_provider", lambda p: {})
monkeypatch.setattr(models_mod, "check_nous_free_tier", lambda: None)
free_tier_calls = []
def _check_nous_free_tier(**kwargs):
free_tier_calls.append(kwargs)
return None
monkeypatch.setattr(models_mod, "check_nous_free_tier", _check_nous_free_tier)
monkeypatch.setattr(
models_mod, "partition_nous_models_by_tier",
lambda ids, p, free_tier=False: (ids, []),
)
monkeypatch.setattr(ns, "prompt_enable_tool_gateway", lambda cfg: None)
return free_tier_calls
def test_skip_keep_current_preserves_provider_and_model(self, tmp_path, monkeypatch):
"""User picks Skip → config.yaml untouched, Nous creds still saved."""
@ -1070,7 +1113,7 @@ class TestLoginNousSkipKeepsCurrent:
hermes_home, config_path, auth_path = self._setup_home_with_openrouter(
tmp_path, monkeypatch,
)
self._patch_login_internals(
free_tier_calls = self._patch_login_internals(
monkeypatch, prompt_returns="xiaomi/mimo-v2-pro",
)
@ -1083,6 +1126,7 @@ class TestLoginNousSkipKeepsCurrent:
cfg_after = yaml.safe_load(config_path.read_text())
assert cfg_after["model"]["provider"] == "nous"
assert cfg_after["model"]["default"] == "xiaomi/mimo-v2-pro"
assert free_tier_calls == [{"force_fresh": True}]
auth_after = json.loads(auth_path.read_text())
assert auth_after["active_provider"] == "nous"

View file

@ -2,6 +2,7 @@
from unittest.mock import patch, MagicMock
from hermes_cli.nous_account import NousPortalAccountInfo
from hermes_cli.models import (
OPENROUTER_MODELS, fetch_openrouter_models, model_ids, detect_provider_for_model,
is_nous_free_tier, partition_nous_models_by_tier,
@ -308,6 +309,15 @@ class TestDetectProviderForModel:
class TestIsNousFreeTier:
"""Tests for is_nous_free_tier — account tier detection."""
def test_paid_service_access_allowed_true_is_not_free(self):
assert is_nous_free_tier({"paid_service_access": {"allowed": True}}) is False
def test_paid_service_access_allowed_false_is_free(self):
assert is_nous_free_tier({"paid_service_access": {"allowed": False}}) is True
def test_paid_service_access_paid_access_fallback(self):
assert is_nous_free_tier({"paid_service_access": {"paid_access": False}}) is True
def test_paid_plus_tier(self):
assert is_nous_free_tier({"subscription": {"plan": "Plus", "tier": 2, "monthly_charge": 20}}) is False
@ -657,39 +667,58 @@ class TestCheckNousFreeTierCache:
def teardown_method(self):
_models_mod._free_tier_cache = None
@patch("hermes_cli.models.fetch_nous_account_tier")
@patch("hermes_cli.models.is_nous_free_tier", return_value=True)
def test_result_is_cached(self, mock_is_free, mock_fetch):
"""Second call within TTL returns cached result without API call."""
mock_fetch.return_value = {"subscription": {"monthly_charge": 0}}
with patch("hermes_cli.auth.get_provider_auth_state", return_value={"access_token": "tok"}), \
patch("hermes_cli.auth.resolve_nous_runtime_credentials"):
result1 = check_nous_free_tier()
result2 = check_nous_free_tier()
@patch("hermes_cli.nous_account.get_nous_portal_account_info")
def test_result_is_cached(self, mock_account):
"""Second call within TTL returns cached result without account lookup."""
mock_account.return_value = NousPortalAccountInfo(
logged_in=True,
source="jwt",
fresh=False,
paid_service_access=False,
)
result1 = check_nous_free_tier()
result2 = check_nous_free_tier()
assert result1 is True
assert result2 is True
assert mock_fetch.call_count == 1
assert mock_account.call_count == 1
@patch("hermes_cli.models.fetch_nous_account_tier")
@patch("hermes_cli.models.is_nous_free_tier", return_value=False)
def test_cache_expires_after_ttl(self, mock_is_free, mock_fetch):
"""After TTL expires, the API is called again."""
mock_fetch.return_value = {"subscription": {"monthly_charge": 20}}
with patch("hermes_cli.auth.get_provider_auth_state", return_value={"access_token": "tok"}), \
patch("hermes_cli.auth.resolve_nous_runtime_credentials"):
result1 = check_nous_free_tier()
assert mock_fetch.call_count == 1
@patch("hermes_cli.nous_account.get_nous_portal_account_info")
def test_cache_expires_after_ttl(self, mock_account):
"""After TTL expires, account info is resolved again."""
mock_account.return_value = NousPortalAccountInfo(
logged_in=True,
source="jwt",
fresh=False,
paid_service_access=True,
)
result1 = check_nous_free_tier()
assert mock_account.call_count == 1
cached_result, cached_at = _models_mod._free_tier_cache
_models_mod._free_tier_cache = (cached_result, cached_at - _FREE_TIER_CACHE_TTL - 1)
cached_result, cached_at = _models_mod._free_tier_cache
_models_mod._free_tier_cache = (cached_result, cached_at - _FREE_TIER_CACHE_TTL - 1)
result2 = check_nous_free_tier()
assert mock_fetch.call_count == 2
result2 = check_nous_free_tier()
assert mock_account.call_count == 2
assert result1 is False
assert result2 is False
@patch("hermes_cli.nous_account.get_nous_portal_account_info")
def test_force_fresh_bypasses_cache(self, mock_account):
mock_account.return_value = NousPortalAccountInfo(
logged_in=True,
source="account_api",
fresh=True,
paid_service_access=True,
)
assert check_nous_free_tier() is False
assert check_nous_free_tier(force_fresh=True) is False
assert mock_account.call_count == 2
mock_account.assert_called_with(force_fresh=True)
def test_cache_ttl_is_short(self):
"""TTL should be short enough to catch upgrades quickly (<=5 min)."""
assert _FREE_TIER_CACHE_TTL <= 300

View file

@ -0,0 +1,547 @@
"""Tests for normalized Nous Portal account entitlement helpers."""
from __future__ import annotations
import base64
import json
import time
from typing import Any
import pytest
from hermes_cli.nous_account import (
NousPaidServiceAccessInfo,
NousPortalAccountInfo,
format_nous_portal_entitlement_message,
get_nous_portal_account_info,
reset_nous_portal_account_info_cache,
)
def _jwt(claims: dict[str, Any]) -> str:
def _part(payload: dict[str, Any]) -> str:
raw = json.dumps(payload, separators=(",", ":")).encode()
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
return f"{_part({'alg': 'none', 'typ': 'JWT'})}.{_part(claims)}.sig"
def _state(token: str) -> dict[str, Any]:
return {
"access_token": token,
"portal_base_url": "https://portal.example.test",
"client_id": "hermes-cli",
}
def _account_payload(
*,
allowed: bool,
subscription: dict[str, Any] | None,
subscription_credits: float,
purchased_credits: float,
) -> dict[str, Any]:
return {
"user": {
"email": "alice@example.test",
"privy_did": "did:privy:alice",
},
"organisation": {
"id": "org_123",
},
"subscription": subscription,
"purchased_credits_remaining": purchased_credits,
"paid_service_access": {
"allowed": allowed,
"paid_access": allowed,
"reason": "usable_credits" if allowed else "no_usable_credits",
"organisation_id": "org_123",
"effective_at_ms": 123456789,
"has_active_subscription": subscription is not None,
"active_subscription_is_paid": bool(
subscription and subscription.get("monthly_charge", 0) > 0
),
"subscription_tier": subscription.get("tier") if subscription else None,
"subscription_monthly_charge": (
subscription.get("monthly_charge") if subscription else None
),
"subscription_credits_remaining": subscription_credits,
"purchased_credits_remaining": purchased_credits,
"total_usable_credits": subscription_credits + purchased_credits,
},
}
@pytest.fixture(autouse=True)
def _reset_cache():
reset_nous_portal_account_info_cache()
yield
reset_nous_portal_account_info_cache()
def test_valid_jwt_with_paid_access_true(monkeypatch):
token = _jwt(
{
"sub": "user_123",
"org_id": "org_123",
"client_id": "hermes-cli",
"product_id": "nous-hermes-agent",
"nous_client": "hermes-agent",
"exp": int(time.time()) + 900,
"paid_access": True,
"subscription_tier": 2,
}
)
monkeypatch.setattr("hermes_cli.auth.get_provider_auth_state", lambda provider: _state(token))
info = get_nous_portal_account_info()
assert info.source == "jwt"
assert info.fresh is False
assert info.logged_in is True
assert info.user_id == "user_123"
assert info.org_id == "org_123"
assert info.product_id == "nous-hermes-agent"
assert info.paid_service_access is True
assert info.is_paid is True
assert info.is_free_tier is False
def test_valid_jwt_with_paid_access_false(monkeypatch):
token = _jwt(
{
"sub": "user_123",
"org_id": "org_123",
"exp": int(time.time()) + 900,
"paid_access": False,
}
)
monkeypatch.setattr("hermes_cli.auth.get_provider_auth_state", lambda provider: _state(token))
info = get_nous_portal_account_info()
assert info.source == "jwt"
assert info.paid_service_access is False
assert info.is_paid is False
assert info.is_free_tier is True
def test_valid_jwt_missing_paid_access_is_unknown_not_paid(monkeypatch):
token = _jwt(
{
"sub": "user_123",
"org_id": "org_123",
"exp": int(time.time()) + 900,
}
)
monkeypatch.setattr("hermes_cli.auth.get_provider_auth_state", lambda provider: _state(token))
info = get_nous_portal_account_info()
assert info.source == "jwt"
assert info.paid_service_access is None
assert info.is_paid is False
assert info.is_free_tier is False
def test_expired_jwt_falls_back_to_fresh_account(monkeypatch):
token = _jwt(
{
"sub": "user_123",
"org_id": "org_123",
"exp": int(time.time()) - 60,
"paid_access": False,
}
)
payload = _account_payload(
allowed=True,
subscription={
"plan": "Tier 2",
"tier": 2,
"monthly_charge": 20,
"current_period_end": "2026-05-01T00:00:00.000Z",
"credits_remaining": 12.25,
"rollover_credits": 3.5,
},
subscription_credits=12.25,
purchased_credits=7.75,
)
monkeypatch.setattr("hermes_cli.auth.get_provider_auth_state", lambda provider: _state(token))
monkeypatch.setattr("hermes_cli.auth.resolve_nous_access_token", lambda: "fresh-token")
monkeypatch.setattr("hermes_cli.nous_account._fetch_nous_account_info", lambda *a, **kw: payload)
info = get_nous_portal_account_info()
assert info.source == "account_api"
assert info.fresh is True
assert info.paid_service_access is True
assert info.subscription is not None
assert info.subscription.monthly_charge == 20
assert info.paid_service_access_info is not None
assert info.paid_service_access_info.total_usable_credits == 20
@pytest.mark.parametrize(
("payload", "expected_paid"),
[
(
_account_payload(
allowed=True,
subscription={
"plan": "Tier 2",
"tier": 2,
"monthly_charge": 20,
"current_period_end": "2026-05-01T00:00:00.000Z",
"credits_remaining": 12.25,
"rollover_credits": 3.5,
},
subscription_credits=12.25,
purchased_credits=7.75,
),
True,
),
(
_account_payload(
allowed=False,
subscription={
"plan": "Tier 2",
"tier": 2,
"monthly_charge": 20,
"current_period_end": "2026-05-01T00:00:00.000Z",
"credits_remaining": 0,
"rollover_credits": 0,
},
subscription_credits=0,
purchased_credits=0,
),
False,
),
(
_account_payload(
allowed=True,
subscription=None,
subscription_credits=0,
purchased_credits=7.75,
),
True,
),
(
_account_payload(
allowed=False,
subscription=None,
subscription_credits=0,
purchased_credits=0,
),
False,
),
],
)
def test_fresh_account_payload_normalization(monkeypatch, payload, expected_paid):
token = _jwt({"sub": "user_123", "org_id": "org_123", "exp": int(time.time()) + 900})
monkeypatch.setattr("hermes_cli.auth.get_provider_auth_state", lambda provider: _state(token))
monkeypatch.setattr("hermes_cli.auth.resolve_nous_access_token", lambda: "fresh-token")
monkeypatch.setattr("hermes_cli.nous_account._fetch_nous_account_info", lambda *a, **kw: payload)
info = get_nous_portal_account_info(force_fresh=True)
assert isinstance(info, NousPortalAccountInfo)
assert info.source == "account_api"
assert info.fresh is True
assert info.email == "alice@example.test"
assert info.privy_did == "did:privy:alice"
assert info.org_id == "org_123"
assert info.paid_service_access is expected_paid
assert info.is_paid is expected_paid
assert info.is_free_tier is (not expected_paid)
def test_force_fresh_uses_account_api_even_when_jwt_is_valid(monkeypatch):
token = _jwt(
{
"sub": "user_123",
"org_id": "org_123",
"exp": int(time.time()) + 900,
"paid_access": False,
}
)
payload = _account_payload(
allowed=True,
subscription=None,
subscription_credits=0,
purchased_credits=5,
)
monkeypatch.setattr("hermes_cli.auth.get_provider_auth_state", lambda provider: _state(token))
monkeypatch.setattr("hermes_cli.auth.resolve_nous_access_token", lambda: "fresh-token")
monkeypatch.setattr("hermes_cli.nous_account._fetch_nous_account_info", lambda *a, **kw: payload)
info = get_nous_portal_account_info(force_fresh=True)
assert info.source == "account_api"
assert info.paid_service_access is True
def test_no_oauth_token_reports_inference_key_present(monkeypatch):
monkeypatch.setattr("hermes_cli.auth.get_provider_auth_state", lambda provider: {})
class _Entry:
label = "manual-nous"
access_token = ""
agent_key = "opaque-runtime-key"
agent_key_expires_at = "2099-01-01T00:00:00+00:00"
expires_at = None
inference_base_url = "https://inference.example.test/v1"
base_url = "https://inference.example.test/v1"
priority = 0
@property
def runtime_api_key(self):
return self.agent_key
@property
def runtime_base_url(self):
return self.inference_base_url
class _Pool:
def has_credentials(self):
return True
def entries(self):
return [_Entry()]
monkeypatch.setattr("agent.credential_pool.load_pool", lambda provider: _Pool())
info = get_nous_portal_account_info()
assert info.logged_in is False
assert info.source == "inference_key"
assert info.inference_credential_present is True
assert info.credential_source == "pool:manual-nous"
assert info.paid_service_access is None
def test_pool_oauth_entry_uses_jwt_snapshot(monkeypatch):
token = _jwt(
{
"sub": "user_123",
"org_id": "org_123",
"client_id": "hermes-cli",
"exp": int(time.time()) + 900,
"paid_access": True,
}
)
monkeypatch.setattr("hermes_cli.auth.get_provider_auth_state", lambda provider: {})
class _Entry:
label = "dashboard device_code"
auth_type = "oauth"
access_token = token
refresh_token = "refresh-token"
agent_key = "opaque-runtime-key"
agent_key_expires_at = "2099-01-01T00:00:00+00:00"
expires_at = "2099-01-01T00:00:00+00:00"
portal_base_url = "https://portal.example.test"
inference_base_url = "https://inference.example.test/v1"
base_url = "https://inference.example.test/v1"
priority = 0
@property
def runtime_api_key(self):
return self.agent_key
@property
def runtime_base_url(self):
return self.inference_base_url
class _Pool:
def has_credentials(self):
return True
def entries(self):
return [_Entry()]
monkeypatch.setattr("agent.credential_pool.load_pool", lambda provider: _Pool())
info = get_nous_portal_account_info()
assert info.logged_in is True
assert info.source == "jwt"
assert info.paid_service_access is True
assert info.credential_source == "pool:dashboard device_code"
def test_pool_oauth_entry_force_fresh_uses_account_api(monkeypatch):
token = _jwt(
{
"sub": "user_123",
"org_id": "org_123",
"exp": int(time.time()) + 900,
"paid_access": False,
}
)
payload = _account_payload(
allowed=True,
subscription=None,
subscription_credits=0,
purchased_credits=3,
)
monkeypatch.setattr("hermes_cli.auth.get_provider_auth_state", lambda provider: {})
monkeypatch.setattr("hermes_cli.nous_account._fetch_nous_account_info", lambda *a, **kw: payload)
class _Entry:
label = "dashboard device_code"
auth_type = "oauth"
access_token = token
refresh_token = "refresh-token"
agent_key = "opaque-runtime-key"
agent_key_expires_at = "2099-01-01T00:00:00+00:00"
expires_at = "2099-01-01T00:00:00+00:00"
portal_base_url = "https://portal.example.test"
inference_base_url = "https://inference.example.test/v1"
base_url = "https://inference.example.test/v1"
priority = 0
@property
def runtime_api_key(self):
return self.agent_key
@property
def runtime_base_url(self):
return self.inference_base_url
class _Pool:
def has_credentials(self):
return True
def entries(self):
return [_Entry()]
monkeypatch.setattr("agent.credential_pool.load_pool", lambda provider: _Pool())
info = get_nous_portal_account_info(force_fresh=True)
assert info.logged_in is True
assert info.source == "account_api"
assert info.fresh is True
assert info.paid_service_access is True
assert info.credential_source == "pool:dashboard device_code"
def test_entitlement_message_returns_none_for_paid_access():
info = NousPortalAccountInfo(
logged_in=True,
source="account_api",
fresh=True,
paid_service_access=True,
portal_base_url="https://portal.example.test",
)
assert format_nous_portal_entitlement_message(info, capability="paid models") is None
def test_entitlement_message_for_inference_key_without_portal_login():
info = NousPortalAccountInfo(
logged_in=False,
source="inference_key",
fresh=False,
inference_credential_present=True,
portal_base_url="https://portal.example.test",
)
message = format_nous_portal_entitlement_message(
info,
capability="managed tools",
)
assert message is not None
assert "Nous inference credentials are configured" in message
assert "cannot verify your Nous Portal paid access" in message
assert "Log in with `hermes model`" in message
def test_entitlement_message_for_active_paid_subscription_with_no_credits():
info = NousPortalAccountInfo(
logged_in=True,
source="account_api",
fresh=True,
paid_service_access=False,
portal_base_url="https://portal.example.test",
paid_service_access_info=NousPaidServiceAccessInfo(
allowed=False,
reason="no_usable_credits",
has_active_subscription=True,
active_subscription_is_paid=True,
subscription_credits_remaining=0,
purchased_credits_remaining=0,
total_usable_credits=0,
),
)
message = format_nous_portal_entitlement_message(
info,
capability="managed tools",
)
assert message is not None
assert "credits are exhausted" in message
assert "managed tools" in message
assert "https://portal.example.test/billing" in message
def test_entitlement_message_for_no_subscription_or_credits():
info = NousPortalAccountInfo(
logged_in=True,
source="account_api",
fresh=True,
paid_service_access=False,
portal_base_url="https://portal.example.test",
paid_service_access_info=NousPaidServiceAccessInfo(
allowed=False,
reason="no_usable_credits",
has_active_subscription=False,
subscription_credits_remaining=0,
purchased_credits_remaining=0,
total_usable_credits=0,
),
)
message = format_nous_portal_entitlement_message(info, capability="paid models")
assert message is not None
assert "no active subscription or usable credits" in message
assert "Subscribe or add credits" in message
def test_entitlement_message_for_unknown_entitlement_is_explicit():
info = NousPortalAccountInfo(
logged_in=True,
source="error",
fresh=False,
paid_service_access=None,
portal_base_url="https://portal.example.test",
error="account_api_timeout",
)
message = format_nous_portal_entitlement_message(info, capability="Tool Gateway")
assert message is not None
assert "could not verify" in message
assert "account_api_timeout" in message
assert "Run `hermes model`" in message
def test_entitlement_message_for_account_missing():
info = NousPortalAccountInfo(
logged_in=True,
source="account_api",
fresh=True,
paid_service_access=False,
paid_service_access_info=NousPaidServiceAccessInfo(
allowed=False,
reason="account_missing",
),
)
message = format_nous_portal_entitlement_message(info, capability="Tool Gateway")
assert message is not None
assert "could not find a Nous Portal account or organisation" in message

View file

@ -1,14 +1,25 @@
"""Tests for Nous subscription feature detection."""
from hermes_cli.nous_account import NousPortalAccountInfo
from hermes_cli import nous_subscription as ns
def _account(*, logged_in: bool, paid: bool | None = None) -> NousPortalAccountInfo:
return NousPortalAccountInfo(
logged_in=logged_in,
source="jwt" if logged_in else "none",
fresh=False,
paid_service_access=paid,
)
def test_get_nous_subscription_features_recognizes_direct_exa_backend(monkeypatch):
env = {"EXA_API_KEY": "exa-test"}
monkeypatch.setattr(ns, "get_env_value", lambda name: env.get(name, ""))
monkeypatch.setattr(ns, "get_nous_auth_status", lambda: {})
monkeypatch.setattr(ns, "managed_nous_tools_enabled", lambda: False)
monkeypatch.setattr(
ns, "get_nous_portal_account_info", lambda: _account(logged_in=False)
)
monkeypatch.setattr(ns, "_toolset_enabled", lambda config, key: key == "web")
monkeypatch.setattr(ns, "_has_agent_browser", lambda: False)
monkeypatch.setattr(ns, "resolve_openai_audio_api_key", lambda: "")
@ -26,8 +37,9 @@ def test_get_nous_subscription_features_recognizes_direct_exa_backend(monkeypatc
def test_get_nous_subscription_features_prefers_managed_modal_in_auto_mode(monkeypatch):
monkeypatch.setattr("tools.tool_backend_helpers.managed_nous_tools_enabled", lambda: True)
monkeypatch.setattr(ns, "get_env_value", lambda name: "")
monkeypatch.setattr(ns, "get_nous_auth_status", lambda: {"logged_in": True})
monkeypatch.setattr(ns, "managed_nous_tools_enabled", lambda: True)
monkeypatch.setattr(
ns, "get_nous_portal_account_info", lambda: _account(logged_in=True, paid=True)
)
monkeypatch.setattr(ns, "_toolset_enabled", lambda config, key: key == "terminal")
monkeypatch.setattr(ns, "_has_agent_browser", lambda: False)
monkeypatch.setattr(ns, "resolve_openai_audio_api_key", lambda: "")
@ -46,8 +58,9 @@ def test_get_nous_subscription_features_prefers_managed_modal_in_auto_mode(monke
def test_get_nous_subscription_features_marks_browser_use_as_managed_when_gateway_ready(monkeypatch):
monkeypatch.setattr(ns, "get_env_value", lambda name: "")
monkeypatch.setattr(ns, "get_nous_auth_status", lambda: {"logged_in": True})
monkeypatch.setattr(ns, "managed_nous_tools_enabled", lambda: True)
monkeypatch.setattr(
ns, "get_nous_portal_account_info", lambda: _account(logged_in=True, paid=True)
)
monkeypatch.setattr(ns, "_toolset_enabled", lambda config, key: key == "browser")
monkeypatch.setattr(ns, "_has_agent_browser", lambda: True)
monkeypatch.setattr(ns, "resolve_openai_audio_api_key", lambda: "")
@ -78,8 +91,9 @@ def test_get_nous_subscription_features_uses_direct_browserbase_when_no_managed_
}
monkeypatch.setattr(ns, "get_env_value", lambda name: env.get(name, ""))
monkeypatch.setattr(ns, "get_nous_auth_status", lambda: {"logged_in": True})
monkeypatch.setattr(ns, "managed_nous_tools_enabled", lambda: True)
monkeypatch.setattr(
ns, "get_nous_portal_account_info", lambda: _account(logged_in=True, paid=True)
)
monkeypatch.setattr(ns, "_toolset_enabled", lambda config, key: key == "browser")
monkeypatch.setattr(ns, "_has_agent_browser", lambda: True)
monkeypatch.setattr(ns, "resolve_openai_audio_api_key", lambda: "")
@ -103,8 +117,9 @@ def test_get_nous_subscription_features_prefers_camofox_over_managed_browser_use
env = {"CAMOFOX_URL": "http://localhost:9377"}
monkeypatch.setattr(ns, "get_env_value", lambda name: env.get(name, ""))
monkeypatch.setattr(ns, "get_nous_auth_status", lambda: {"logged_in": True})
monkeypatch.setattr(ns, "managed_nous_tools_enabled", lambda: True)
monkeypatch.setattr(
ns, "get_nous_portal_account_info", lambda: _account(logged_in=True, paid=True)
)
monkeypatch.setattr(ns, "_toolset_enabled", lambda config, key: key == "browser")
monkeypatch.setattr(ns, "_has_agent_browser", lambda: False)
monkeypatch.setattr(ns, "resolve_openai_audio_api_key", lambda: "")
@ -133,8 +148,9 @@ def test_get_nous_subscription_features_requires_agent_browser_for_browserbase(m
}
monkeypatch.setattr(ns, "get_env_value", lambda name: env.get(name, ""))
monkeypatch.setattr(ns, "get_nous_auth_status", lambda: {})
monkeypatch.setattr(ns, "managed_nous_tools_enabled", lambda: False)
monkeypatch.setattr(
ns, "get_nous_portal_account_info", lambda: _account(logged_in=False)
)
monkeypatch.setattr(ns, "_toolset_enabled", lambda config, key: key == "browser")
monkeypatch.setattr(ns, "_has_agent_browser", lambda: False)
monkeypatch.setattr(ns, "resolve_openai_audio_api_key", lambda: "")
@ -155,8 +171,9 @@ def test_get_nous_subscription_features_does_not_treat_quoted_false_as_gateway_o
env = {"EXA_API_KEY": "exa-test"}
monkeypatch.setattr(ns, "get_env_value", lambda name: env.get(name, ""))
monkeypatch.setattr(ns, "get_nous_auth_status", lambda: {"logged_in": True})
monkeypatch.setattr(ns, "managed_nous_tools_enabled", lambda: True)
monkeypatch.setattr(
ns, "get_nous_portal_account_info", lambda: _account(logged_in=True, paid=True)
)
monkeypatch.setattr(ns, "_toolset_enabled", lambda config, key: key == "web")
monkeypatch.setattr(ns, "_has_agent_browser", lambda: False)
monkeypatch.setattr(ns, "resolve_openai_audio_api_key", lambda: "")

View file

@ -83,6 +83,87 @@ def test_show_status_reports_nous_auth_error(monkeypatch, capsys, tmp_path):
assert "Key exp:" in output
def test_show_status_reports_nous_inference_key_without_portal_login(monkeypatch, capsys, tmp_path):
from hermes_cli import status as status_mod
from hermes_cli.nous_account import NousPortalAccountInfo
import hermes_cli.auth as auth_mod
import hermes_cli.gateway as gateway_mod
monkeypatch.setattr(status_mod, "get_env_path", lambda: tmp_path / ".env", raising=False)
monkeypatch.setattr(status_mod, "get_hermes_home", lambda: tmp_path, raising=False)
monkeypatch.setattr(status_mod, "load_config", lambda: {"model": "gpt-5.4"}, raising=False)
monkeypatch.setattr(status_mod, "resolve_requested_provider", lambda requested=None: "openai-codex", raising=False)
monkeypatch.setattr(status_mod, "resolve_provider", lambda requested=None, **kwargs: "openai-codex", raising=False)
monkeypatch.setattr(status_mod, "provider_label", lambda provider: "OpenAI Codex", raising=False)
monkeypatch.setattr(
auth_mod,
"get_nous_auth_status",
lambda: {
"logged_in": False,
"inference_credential_present": True,
"credential_source": "pool:manual opaque key",
"inference_base_url": "https://inference.example.com/v1",
"agent_key_expires_at": "2099-01-01T00:00:00+00:00",
},
raising=False,
)
monkeypatch.setattr(
status_mod,
"get_nous_portal_account_info",
lambda: NousPortalAccountInfo(
logged_in=False,
source="inference_key",
fresh=False,
inference_credential_present=True,
inference_base_url="https://inference.example.com/v1",
),
raising=False,
)
monkeypatch.setattr(status_mod, "managed_nous_tools_enabled", lambda: False, raising=False)
monkeypatch.setattr(auth_mod, "get_codex_auth_status", lambda: {}, raising=False)
monkeypatch.setattr(auth_mod, "get_qwen_auth_status", lambda: {}, raising=False)
monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status", lambda: {}, raising=False)
monkeypatch.setattr(gateway_mod, "find_gateway_pids", lambda exclude_pids=None: [], raising=False)
status_mod.show_status(SimpleNamespace(all=False, deep=False))
output = capsys.readouterr().out
assert "Nous Portal ✗ not logged in (Nous inference key configured)" in output
assert "Inference: https://inference.example.com/v1" in output
assert "Nous inference credentials are configured" in output
def test_show_status_reports_vercel_backend_contract(monkeypatch, capsys, tmp_path):
from hermes_cli import status as status_mod
import hermes_cli.auth as auth_mod
import hermes_cli.gateway as gateway_mod
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("TERMINAL_ENV", "vercel_sandbox")
monkeypatch.setenv("TERMINAL_VERCEL_RUNTIME", "python3.13")
monkeypatch.setenv("TERMINAL_CONTAINER_PERSISTENT", "true")
monkeypatch.setenv("VERCEL_OIDC_TOKEN", "oidc-token")
monkeypatch.setattr(status_mod.importlib.util, "find_spec", lambda name: object() if name == "vercel" else None)
monkeypatch.setattr(status_mod, "load_config", lambda: {"terminal": {"backend": "vercel_sandbox"}}, raising=False)
monkeypatch.setattr(auth_mod, "get_nous_auth_status", lambda: {}, raising=False)
monkeypatch.setattr(auth_mod, "get_codex_auth_status", lambda: {}, raising=False)
monkeypatch.setattr(auth_mod, "get_qwen_auth_status", lambda: {}, raising=False)
monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status", lambda: {}, raising=False)
monkeypatch.setattr(gateway_mod, "find_gateway_pids", lambda exclude_pids=None: [], raising=False)
status_mod.show_status(SimpleNamespace(all=False, deep=False))
output = capsys.readouterr().out
assert "Backend: vercel_sandbox" in output
assert "Runtime: python3.13" in output
assert "Auth:" in output and "OIDC token via VERCEL_OIDC_TOKEN" in output
assert "Auth detail: mode: OIDC" in output
assert "Auth detail: active env: VERCEL_OIDC_TOKEN" in output
assert "oidc-token" not in output
assert "snapshot filesystem" in output
assert "live processes do not survive" in output
# ---------------------------------------------------------------------------
# Helpers shared by xAI OAuth status tests
# ---------------------------------------------------------------------------

View file

@ -2,6 +2,7 @@
from types import SimpleNamespace
from hermes_cli.nous_account import NousPaidServiceAccessInfo, NousPortalAccountInfo
from hermes_cli.nous_subscription import NousFeatureState, NousSubscriptionFeatures
@ -124,6 +125,59 @@ def test_show_status_hides_nous_subscription_section_when_feature_flag_is_off(mo
assert "Nous Tool Gateway" not in out
def test_show_status_reports_exhausted_nous_credits(monkeypatch, capsys, tmp_path):
monkeypatch.setattr("hermes_cli.status.managed_nous_tools_enabled", lambda: False)
from hermes_cli import status as status_mod
import hermes_cli.auth as auth_mod
_patch_common_status_deps(monkeypatch, status_mod, tmp_path)
monkeypatch.setattr(
auth_mod,
"get_nous_auth_status",
lambda: {
"logged_in": False,
"access_token": "jwt",
"portal_base_url": "https://portal.example.test",
"error": "credits exhausted",
"error_code": "insufficient_credits",
},
raising=False,
)
monkeypatch.setattr(
status_mod,
"get_nous_portal_account_info",
lambda: NousPortalAccountInfo(
logged_in=True,
source="account_api",
fresh=True,
paid_service_access=False,
portal_base_url="https://portal.example.test",
paid_service_access_info=NousPaidServiceAccessInfo(
allowed=False,
reason="no_usable_credits",
has_active_subscription=True,
active_subscription_is_paid=True,
subscription_credits_remaining=0,
purchased_credits_remaining=0,
total_usable_credits=0,
),
),
raising=False,
)
monkeypatch.setattr(status_mod, "load_config", lambda: {"model": {"provider": "nous"}}, raising=False)
monkeypatch.setattr(status_mod, "resolve_requested_provider", lambda requested=None: "nous", raising=False)
monkeypatch.setattr(status_mod, "resolve_provider", lambda requested=None, **kwargs: "nous", raising=False)
monkeypatch.setattr(status_mod, "provider_label", lambda provider: "Nous Portal", raising=False)
status_mod.show_status(SimpleNamespace(all=False, deep=False))
out = capsys.readouterr().out
assert "Nous Tool Gateway" in out
assert "credits are exhausted" in out
assert "https://portal.example.test/billing" in out
assert "free-tier Nous account" not in out
def test_show_status_reports_empty_lmstudio_listing_as_reachable(monkeypatch, capsys, tmp_path):
from hermes_cli import status as status_mod

View file

@ -4,6 +4,7 @@ from unittest.mock import patch
import pytest
from hermes_cli.nous_account import NousPortalAccountInfo
from hermes_cli.tools_config import (
_DEFAULT_OFF_TOOLSETS,
_apply_toolset_change,
@ -557,8 +558,13 @@ def test_visible_providers_include_nous_subscription_when_logged_in(monkeypatch)
config = {"model": {"provider": "nous"}}
monkeypatch.setattr(
"hermes_cli.nous_subscription.get_nous_auth_status",
lambda: {"logged_in": True},
"hermes_cli.nous_subscription.get_nous_portal_account_info",
lambda: NousPortalAccountInfo(
logged_in=True,
source="jwt",
fresh=False,
paid_service_access=True,
),
)
providers = _visible_providers(TOOL_CATEGORIES["browser"], config)
@ -571,8 +577,13 @@ def test_visible_providers_hide_nous_subscription_when_feature_flag_is_off(monke
config = {"model": {"provider": "nous"}}
monkeypatch.setattr(
"hermes_cli.nous_subscription.get_nous_auth_status",
lambda: {"logged_in": True},
"hermes_cli.nous_subscription.get_nous_portal_account_info",
lambda: NousPortalAccountInfo(
logged_in=True,
source="jwt",
fresh=False,
paid_service_access=True,
),
)
providers = _visible_providers(TOOL_CATEGORIES["browser"], config)
@ -657,8 +668,13 @@ def test_first_install_nous_auto_configures_managed_defaults(monkeypatch):
lambda: ["cli"],
)
monkeypatch.setattr(
"hermes_cli.nous_subscription.get_nous_auth_status",
lambda: {"logged_in": True},
"hermes_cli.nous_subscription.get_nous_portal_account_info",
lambda: NousPortalAccountInfo(
logged_in=True,
source="jwt",
fresh=False,
paid_service_access=True,
),
)
configured = []

View file

@ -489,6 +489,42 @@ class TestErrorResponseShapes:
assert "error" in result["results"][0]
assert result["results"][0]["url"] == "https://example.com"
def test_firecrawl_config_error_points_paid_users_to_nous_subscription(self, monkeypatch):
from plugins.web.firecrawl import provider as firecrawl_provider
monkeypatch.setattr(
"tools.web_tools.managed_nous_tools_enabled",
lambda: True,
raising=False,
)
with pytest.raises(ValueError) as exc_info:
firecrawl_provider._raise_web_backend_configuration_error()
message = str(exc_info.value)
assert "With your Nous subscription you can also use the Tool Gateway" in message
assert "select Nous Subscription as the web provider" in message
assert "managed Firecrawl web tools is unavailable" not in message
def test_firecrawl_config_error_uses_entitlement_message_when_not_paid(self, monkeypatch):
from plugins.web.firecrawl import provider as firecrawl_provider
monkeypatch.setattr(
"tools.web_tools.managed_nous_tools_enabled",
lambda: False,
raising=False,
)
monkeypatch.setattr(
"tools.web_tools.nous_tool_gateway_unavailable_message",
lambda capability: f"{capability} denied by test entitlement.",
raising=False,
)
with pytest.raises(ValueError) as exc_info:
firecrawl_provider._raise_web_backend_configuration_error()
assert "managed Firecrawl web tools denied by test entitlement" in str(exc_info.value)
def test_xai_search_returns_error_dict_when_unconfigured(self) -> None:
"""xAI returns a typed error dict (no XAI_API_KEY)."""
_ensure_plugins_loaded()

View file

@ -3875,6 +3875,33 @@ class TestNousCredentialRefresh:
assert "default_headers" not in rebuilt["kwargs"]
assert isinstance(agent.client, _RebuiltClient)
def test_try_refresh_nous_client_credentials_accepts_explicit_auth_mode(
self, agent, monkeypatch
):
agent.provider = "nous"
agent.api_mode = "chat_completions"
captured = {}
def _fake_resolve(**kwargs):
captured.update(kwargs)
return {
"api_key": "new-nous-key",
"base_url": "https://inference-api.nousresearch.com/v1",
}
monkeypatch.setattr(
"hermes_cli.auth.resolve_nous_runtime_credentials", _fake_resolve
)
with patch("run_agent.OpenAI", return_value=MagicMock()):
ok = agent._try_refresh_nous_client_credentials(
force=False,
inference_auth_mode="legacy",
)
assert ok is True
assert captured["inference_auth_mode"] == "legacy"
class TestCredentialPoolRecovery:
def test_recover_with_pool_rotates_on_402(self, agent):

View file

@ -9,6 +9,8 @@ from unittest.mock import patch
import pytest
from hermes_cli.nous_account import NousPortalAccountInfo
REPO_ROOT = Path(__file__).resolve().parents[2]
TOOLS_DIR = REPO_ROOT / "tools"
@ -69,10 +71,17 @@ def _enable_managed_nous_tools(monkeypatch):
The _install_fake_tools_package() helper resets and reimports tool modules,
so a simple monkeypatch on tool_backend_helpers doesn't survive. We patch
the *source* modules that the reimported modules will import from both
hermes_cli.auth and hermes_cli.models so the function body returns True.
hermes_cli.nous_account so the function body returns True.
"""
monkeypatch.setattr("hermes_cli.auth.get_nous_auth_status", lambda: {"logged_in": True})
monkeypatch.setattr("hermes_cli.models.check_nous_free_tier", lambda: False)
monkeypatch.setattr(
"hermes_cli.nous_account.get_nous_portal_account_info",
lambda: NousPortalAccountInfo(
logged_in=True,
source="jwt",
fresh=False,
paid_service_access=True,
),
)
def _install_fake_tools_package():

View file

@ -5,6 +5,8 @@ from pathlib import Path
import pytest
from hermes_cli.nous_account import NousPortalAccountInfo
TOOLS_DIR = Path(__file__).resolve().parents[2] / "tools"
@ -48,8 +50,15 @@ def _restore_tool_and_agent_modules():
def _enable_managed_nous_tools(monkeypatch):
"""Patch the source modules so managed_nous_tools_enabled() returns True
even after tool modules are dynamically reloaded."""
monkeypatch.setattr("hermes_cli.auth.get_nous_auth_status", lambda: {"logged_in": True})
monkeypatch.setattr("hermes_cli.models.check_nous_free_tier", lambda: False)
monkeypatch.setattr(
"hermes_cli.nous_account.get_nous_portal_account_info",
lambda: NousPortalAccountInfo(
logged_in=True,
source="jwt",
fresh=False,
paid_service_access=True,
),
)
def _install_fake_tools_package():

View file

@ -16,10 +16,12 @@ from unittest.mock import patch
import pytest
from hermes_cli.nous_account import NousPaidServiceAccessInfo, NousPortalAccountInfo
from tools.tool_backend_helpers import (
coerce_modal_mode,
has_direct_modal_credentials,
managed_nous_tools_enabled,
nous_tool_gateway_unavailable_message,
normalize_browser_cloud_provider,
normalize_modal_mode,
prefers_gateway,
@ -40,42 +42,73 @@ class TestManagedNousToolsEnabled:
def test_disabled_when_not_logged_in(self, monkeypatch):
monkeypatch.setattr(
"hermes_cli.auth.get_nous_auth_status",
lambda: {},
"hermes_cli.nous_account.get_nous_portal_account_info",
lambda: NousPortalAccountInfo(logged_in=False, source="none", fresh=False),
)
assert managed_nous_tools_enabled() is False
def test_disabled_for_free_tier(self, monkeypatch):
monkeypatch.setattr(
"hermes_cli.auth.get_nous_auth_status",
lambda: {"logged_in": True},
)
monkeypatch.setattr(
"hermes_cli.models.check_nous_free_tier",
lambda: True,
"hermes_cli.nous_account.get_nous_portal_account_info",
lambda: NousPortalAccountInfo(
logged_in=True,
source="jwt",
fresh=False,
paid_service_access=False,
),
)
assert managed_nous_tools_enabled() is False
def test_enabled_for_paid_subscriber(self, monkeypatch):
monkeypatch.setattr(
"hermes_cli.auth.get_nous_auth_status",
lambda: {"logged_in": True},
)
monkeypatch.setattr(
"hermes_cli.models.check_nous_free_tier",
lambda: False,
"hermes_cli.nous_account.get_nous_portal_account_info",
lambda: NousPortalAccountInfo(
logged_in=True,
source="jwt",
fresh=False,
paid_service_access=True,
),
)
assert managed_nous_tools_enabled() is True
def test_returns_false_on_exception(self, monkeypatch):
"""Should never crash — returns False on any exception."""
monkeypatch.setattr(
"hermes_cli.auth.get_nous_auth_status",
"hermes_cli.nous_account.get_nous_portal_account_info",
_raise_import,
)
assert managed_nous_tools_enabled() is False
class TestNousToolGatewayUnavailableMessage:
def test_uses_entitlement_reason_for_logged_in_user(self, monkeypatch):
monkeypatch.setattr(
"hermes_cli.nous_account.get_nous_portal_account_info",
lambda force_fresh=False: NousPortalAccountInfo(
logged_in=True,
source="account_api",
fresh=True,
paid_service_access=False,
portal_base_url="https://portal.example.test",
paid_service_access_info=NousPaidServiceAccessInfo(
allowed=False,
reason="no_usable_credits",
has_active_subscription=True,
active_subscription_is_paid=True,
subscription_credits_remaining=0,
purchased_credits_remaining=0,
total_usable_credits=0,
),
),
)
message = nous_tool_gateway_unavailable_message("managed image generation")
assert "credits are exhausted" in message
assert "managed image generation" in message
assert "https://portal.example.test/billing" in message
# ---------------------------------------------------------------------------
# normalize_browser_cloud_provider
# ---------------------------------------------------------------------------

View file

@ -66,6 +66,7 @@ from tools.managed_tool_gateway import resolve_managed_tool_gateway
from tools.tool_backend_helpers import (
fal_key_is_configured,
managed_nous_tools_enabled,
nous_tool_gateway_unavailable_message,
prefers_gateway,
)
@ -452,12 +453,22 @@ def _submit_fal_request(model: str, arguments: Dict[str, Any]):
# of a raw HTTP error from httpx.
status = _extract_http_status(exc)
if status is not None and 400 <= status < 500:
gateway_message = ""
if status in {401, 402, 403}:
gateway_message = (
"\n\n"
+ nous_tool_gateway_unavailable_message(
"managed FAL image generation",
force_fresh=True,
)
)
raise ValueError(
f"Nous Subscription gateway rejected model '{model}' "
f"(HTTP {status}). This model may not yet be enabled on "
f"the Nous Portal's FAL proxy. Either:\n"
f" • Set FAL_KEY in your environment to use FAL.ai directly, or\n"
f" • Pick a different model via `hermes tools` → Image Generation."
f"{gateway_message}"
) from exc
raise
@ -767,6 +778,11 @@ def _build_no_backend_setup_message() -> str:
)
else:
lines.append(" - FAL_KEY environment variable is not set")
gateway_message = nous_tool_gateway_unavailable_message(
"managed FAL image generation",
)
if gateway_message:
lines.append(f" - {gateway_message}")
lines.append("")
lines.append("To enable image generation, do one of:")
lines.append(

View file

@ -71,6 +71,7 @@ from tools.tool_backend_helpers import (
coerce_modal_mode,
has_direct_modal_credentials,
managed_nous_tools_enabled,
nous_tool_gateway_unavailable_message,
resolve_modal_backend_state,
)
@ -1118,13 +1119,19 @@ def _create_environment(env_type: str, image: str, cwd: str, timeout: int,
if modal_state["managed_mode_blocked"]:
raise ValueError(
"Modal backend is configured for managed mode, but "
"a paid Nous subscription is required for the Tool Gateway and no direct "
"Modal credentials/config were found. Log in with `hermes model` or "
"choose TERMINAL_MODAL_MODE=direct/auto."
"Nous Tool Gateway access is not currently available and no direct "
"Modal credentials/config were found. "
+ nous_tool_gateway_unavailable_message(
"managed Modal execution",
)
+ " Choose TERMINAL_MODAL_MODE=direct/auto to use direct Modal credentials."
)
if modal_state["mode"] == "managed":
raise ValueError(
"Modal backend is configured for managed mode, but the managed tool gateway is unavailable."
"Modal backend is configured for managed mode, but the managed tool gateway is unavailable. "
+ nous_tool_gateway_unavailable_message(
"managed Modal execution",
)
)
if modal_state["mode"] == "direct":
raise ValueError(
@ -2214,16 +2221,21 @@ def check_terminal_requirements() -> bool:
if modal_state["managed_mode_blocked"]:
logger.error(
"Modal backend selected with TERMINAL_MODAL_MODE=managed, but "
"a paid Nous subscription is required for the Tool Gateway and no direct "
"Modal credentials/config were found. Log in with `hermes model` "
"or choose TERMINAL_MODAL_MODE=direct/auto."
"Nous Tool Gateway access is not currently available and no direct "
"Modal credentials/config were found. %s Choose "
"TERMINAL_MODAL_MODE=direct/auto to use direct Modal credentials.",
nous_tool_gateway_unavailable_message(
"managed Modal execution",
),
)
return False
if modal_state["mode"] == "managed":
logger.error(
"Modal backend selected with TERMINAL_MODAL_MODE=managed, but the managed "
"tool gateway is unavailable. Configure the managed gateway or choose "
"TERMINAL_MODAL_MODE=direct/auto."
"tool gateway is unavailable. %s",
nous_tool_gateway_unavailable_message(
"managed Modal execution",
),
)
return False
elif modal_state["mode"] == "direct":

View file

@ -15,28 +15,49 @@ _VALID_MODAL_MODES = {"auto", "direct", "managed"}
def managed_nous_tools_enabled() -> bool:
"""Return True when the user has an active paid Nous subscription.
"""Return True when the user has paid Nous Portal service access.
The Tool Gateway is available to any Nous subscriber who is NOT on
the free tier. We intentionally catch all exceptions and return
False never block the agent startup path.
Tool Gateway availability fails closed on unknown/error entitlement. We
intentionally catch all exceptions and return False never block startup.
"""
try:
from hermes_cli.auth import get_nous_auth_status
from hermes_cli.nous_account import get_nous_portal_account_info
status = get_nous_auth_status()
if not status.get("logged_in"):
account_info = get_nous_portal_account_info()
if not account_info.logged_in:
return False
from hermes_cli.models import check_nous_free_tier
if check_nous_free_tier():
return False # free-tier users don't get gateway access
return True
return account_info.paid_service_access is True
except Exception:
return False
def nous_tool_gateway_unavailable_message(
capability: str = "the Nous Tool Gateway",
*,
force_fresh: bool = False,
) -> str:
"""Return account-aware guidance for an unavailable Nous Tool Gateway path."""
try:
from hermes_cli.nous_account import (
format_nous_portal_entitlement_message,
get_nous_portal_account_info,
)
account_info = get_nous_portal_account_info(force_fresh=force_fresh)
message = format_nous_portal_entitlement_message(
account_info,
capability=capability,
)
if message:
return message
except Exception:
pass
return (
f"{capability} is unavailable. Run `hermes model` to refresh your "
"Nous Portal login and billing status."
)
def normalize_browser_cloud_provider(value: object | None) -> str:
"""Return a normalized browser provider key."""
provider = str(value or _DEFAULT_BROWSER_PROVIDER).strip().lower()

View file

@ -38,7 +38,11 @@ from urllib.parse import urljoin
from utils import is_truthy_value
from tools.managed_tool_gateway import resolve_managed_tool_gateway
from tools.tool_backend_helpers import managed_nous_tools_enabled, resolve_openai_audio_api_key
from tools.tool_backend_helpers import (
managed_nous_tools_enabled,
nous_tool_gateway_unavailable_message,
resolve_openai_audio_api_key,
)
logger = logging.getLogger(__name__)
@ -1643,7 +1647,12 @@ def _resolve_openai_audio_client_config() -> tuple[str, str]:
if managed_gateway is None:
message = "Neither stt.openai.api_key in config nor VOICE_TOOLS_OPENAI_KEY/OPENAI_API_KEY is set"
if managed_nous_tools_enabled():
message += ", and the managed OpenAI audio gateway is unavailable"
message += (
". "
+ nous_tool_gateway_unavailable_message(
"managed OpenAI audio for transcription",
)
)
raise ValueError(message)
return managed_gateway.nous_user_token, urljoin(

View file

@ -69,7 +69,12 @@ def get_env_value(name, default=None):
value = _get_env_value(name)
return default if value is None else value
from tools.managed_tool_gateway import resolve_managed_tool_gateway
from tools.tool_backend_helpers import managed_nous_tools_enabled, prefers_gateway, resolve_openai_audio_api_key
from tools.tool_backend_helpers import (
managed_nous_tools_enabled,
nous_tool_gateway_unavailable_message,
prefers_gateway,
resolve_openai_audio_api_key,
)
from tools.xai_http import hermes_xai_user_agent
# ---------------------------------------------------------------------------
@ -2206,8 +2211,13 @@ def _resolve_openai_audio_client_config() -> tuple[str, str]:
managed_gateway = resolve_managed_tool_gateway("openai-audio")
if managed_gateway is None:
message = "Neither VOICE_TOOLS_OPENAI_KEY nor OPENAI_API_KEY is set"
if managed_nous_tools_enabled():
message += ", and the managed OpenAI audio gateway is unavailable"
if managed_nous_tools_enabled() or prefers_gateway("tts"):
message += (
". "
+ nous_tool_gateway_unavailable_message(
"managed OpenAI audio for TTS",
)
)
raise ValueError(message)
return managed_gateway.nous_user_token, urljoin(

View file

@ -110,7 +110,11 @@ from tools.managed_tool_gateway import ( # noqa: F401 — backward-compat names
read_nous_access_token as _read_nous_access_token,
resolve_managed_tool_gateway,
)
from tools.tool_backend_helpers import managed_nous_tools_enabled, prefers_gateway # noqa: F401
from tools.tool_backend_helpers import ( # noqa: F401
managed_nous_tools_enabled,
nous_tool_gateway_unavailable_message,
prefers_gateway,
)
from tools.url_safety import is_safe_url
from tools.website_policy import check_website_access
import sys