fix(auth): refresh Nous entitlement in tool menus

This commit is contained in:
Robin Fernandes 2026-05-28 09:53:22 +10:00 committed by Teknium
parent 406901b27d
commit 1cf5e639b3
8 changed files with 304 additions and 108 deletions

View file

@ -228,6 +228,8 @@ def _resolve_browser_feature_state(
def get_nous_subscription_features(
config: Optional[Dict[str, object]] = None,
*,
force_fresh: bool = False,
) -> NousSubscriptionFeatures:
if config is None:
config = load_config() or {}
@ -236,7 +238,10 @@ def get_nous_subscription_features(
provider_is_nous = str(model_cfg.get("provider") or "").strip().lower() == "nous"
try:
account_info = get_nous_portal_account_info()
if force_fresh:
account_info = get_nous_portal_account_info(force_fresh=True)
else:
account_info = get_nous_portal_account_info()
except Exception:
account_info = None
@ -322,6 +327,7 @@ def get_nous_subscription_features(
modal_mode,
has_direct=direct_modal,
managed_ready=managed_modal_available,
managed_enabled=managed_tools_flag,
)
web_managed = web_backend == "firecrawl" and managed_web_available and not direct_firecrawl
@ -499,11 +505,15 @@ def apply_nous_managed_defaults(
config: Dict[str, object],
*,
enabled_toolsets: Optional[Iterable[str]] = None,
force_fresh: bool = False,
) -> set[str]:
if not managed_nous_tools_enabled():
features = get_nous_subscription_features(config, force_fresh=force_fresh)
if not (
features.account_info
and features.account_info.logged_in
and features.account_info.paid_service_access is True
):
return set()
features = get_nous_subscription_features(config)
if not features.provider_is_nous:
return set()
@ -600,6 +610,8 @@ _ALL_GATEWAY_KEYS = ("web", "image_gen", "tts", "browser")
def get_gateway_eligible_tools(
config: Optional[Dict[str, object]] = None,
*,
force_fresh: bool = False,
) -> tuple[list[str], list[str], list[str]]:
"""Return (unconfigured, has_direct, already_managed) tool key lists.
@ -610,7 +622,11 @@ def get_gateway_eligible_tools(
All lists are empty when the user is not a paid Nous subscriber or
is not using Nous as their provider.
"""
if not managed_nous_tools_enabled():
if force_fresh:
managed_enabled = managed_nous_tools_enabled(force_fresh=True)
else:
managed_enabled = managed_nous_tools_enabled()
if not managed_enabled:
return [], [], []
if config is None:
@ -701,7 +717,11 @@ def apply_gateway_defaults(
return changed
def prompt_enable_tool_gateway(config: Dict[str, object]) -> set[str]:
def prompt_enable_tool_gateway(
config: Dict[str, object],
*,
force_fresh: bool = True,
) -> set[str]:
"""If eligible tools exist, prompt the user to enable the Tool Gateway.
Uses prompt_choice() with a description parameter so the curses TUI
@ -710,7 +730,10 @@ def prompt_enable_tool_gateway(config: Dict[str, object]) -> set[str]:
Returns the set of tools that were enabled, or empty set if the user
declined or no tools were eligible.
"""
unconfigured, has_direct, already_managed = get_gateway_eligible_tools(config)
unconfigured, has_direct, already_managed = get_gateway_eligible_tools(
config,
force_fresh=force_fresh,
)
if not unconfigured and not has_direct:
return set()

View file

@ -29,7 +29,7 @@ from hermes_cli.nous_subscription import (
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 tools.tool_backend_helpers import fal_key_is_configured
from utils import base_url_hostname, is_truthy_value
logger = logging.getLogger(__name__)
@ -1400,7 +1400,12 @@ def _save_platform_tools(config: dict, platform: str, enabled_toolset_keys: Set[
save_config(config)
def _toolset_has_keys(ts_key: str, config: dict = None) -> bool:
def _toolset_has_keys(
ts_key: str,
config: dict = None,
*,
force_fresh: bool = False,
) -> bool:
"""Check if a toolset's required API keys are configured."""
if config is None:
config = load_config()
@ -1415,7 +1420,7 @@ def _toolset_has_keys(ts_key: str, config: dict = None) -> bool:
return False
if ts_key in {"web", "image_gen", "tts", "browser"}:
features = get_nous_subscription_features(config)
features = get_nous_subscription_features(config, force_fresh=force_fresh)
feature = features.features.get(ts_key)
if feature and (feature.available or feature.managed_by_nous):
return True
@ -1423,7 +1428,7 @@ def _toolset_has_keys(ts_key: str, config: dict = None) -> bool:
# Check TOOL_CATEGORIES first (provider-aware)
cat = TOOL_CATEGORIES.get(ts_key)
if cat:
for provider in _visible_providers(cat, config):
for provider in _visible_providers(cat, config, force_fresh=force_fresh):
env_vars = provider.get("env_vars", [])
if not env_vars:
return True # No-key provider (e.g. Local Browser, Edge TTS)
@ -1494,7 +1499,13 @@ def _estimate_tool_tokens() -> Dict[str, int]:
return _tool_token_cache
def _prompt_toolset_checklist(platform_label: str, enabled: Set[str], platform: str = "cli") -> Set[str]:
def _prompt_toolset_checklist(
platform_label: str,
enabled: Set[str],
platform: str = "cli",
*,
force_fresh: bool = True,
) -> Set[str]:
"""Multi-select checklist of toolsets. Returns set of selected toolset keys."""
from hermes_cli.curses_ui import curses_checklist
from toolsets import resolve_toolset
@ -1512,7 +1523,10 @@ def _prompt_toolset_checklist(platform_label: str, enabled: Set[str], platform:
labels = []
for ts_key, ts_label, ts_desc in effective:
suffix = ""
if not _toolset_has_keys(ts_key) and (TOOL_CATEGORIES.get(ts_key) or TOOLSET_ENV_REQUIREMENTS.get(ts_key)):
if (
not _toolset_has_keys(ts_key, force_fresh=force_fresh)
and (TOOL_CATEGORIES.get(ts_key) or TOOLSET_ENV_REQUIREMENTS.get(ts_key))
):
suffix = " [no API key]"
labels.append(f"{ts_label} ({ts_desc}){suffix}")
@ -1548,7 +1562,12 @@ def _prompt_toolset_checklist(platform_label: str, enabled: Set[str], platform:
# ─── Provider-Aware Configuration ────────────────────────────────────────────
def _configure_toolset(ts_key: str, config: dict):
def _configure_toolset(
ts_key: str,
config: dict,
*,
force_fresh: bool = True,
):
"""Configure a toolset - provider selection + API keys.
Uses TOOL_CATEGORIES for provider-aware config, falls back to simple
@ -1557,7 +1576,7 @@ def _configure_toolset(ts_key: str, config: dict):
cat = TOOL_CATEGORIES.get(ts_key)
if cat:
_configure_tool_category(ts_key, cat, config)
_configure_tool_category(ts_key, cat, config, force_fresh=force_fresh)
else:
# Simple fallback for vision, moa, etc.
_configure_simple_requirements(ts_key)
@ -1810,12 +1829,22 @@ def _plugin_tts_providers() -> list[dict]:
return rows
def _visible_providers(cat: dict, config: dict) -> list[dict]:
def _visible_providers(
cat: dict,
config: dict,
*,
force_fresh: bool = False,
) -> list[dict]:
"""Return provider entries visible for the current auth/config state."""
features = get_nous_subscription_features(config)
features = get_nous_subscription_features(config, force_fresh=force_fresh)
managed_available = bool(
features.account_info
and features.account_info.logged_in
and features.account_info.paid_service_access is True
)
visible = []
for provider in cat.get("providers", []):
if provider.get("managed_nous_feature") and not managed_nous_tools_enabled():
if provider.get("managed_nous_feature") and not managed_available:
continue
if provider.get("requires_nous_auth") and not features.nous_auth_present:
continue
@ -1856,13 +1885,24 @@ def _visible_providers(cat: dict, config: dict) -> list[dict]:
return visible
def _hidden_nous_gateway_message(cat: dict, config: dict, capability: str) -> str:
def _hidden_nous_gateway_message(
cat: dict,
config: dict,
capability: str,
*,
force_fresh: bool = False,
) -> str:
"""Return a reason when a category's Nous provider is hidden."""
if managed_nous_tools_enabled():
features = get_nous_subscription_features(config, force_fresh=force_fresh)
managed_available = bool(
features.account_info
and features.account_info.logged_in
and features.account_info.paid_service_access is True
)
if managed_available:
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,
@ -1901,17 +1941,22 @@ def _post_setup_already_installed(post_setup_key: str) -> bool:
return True
def _toolset_needs_configuration_prompt(ts_key: str, config: dict) -> bool:
def _toolset_needs_configuration_prompt(
ts_key: str,
config: dict,
*,
force_fresh: bool = False,
) -> bool:
"""Return True when enabling this toolset should open provider setup."""
cat = TOOL_CATEGORIES.get(ts_key)
if not cat:
return not _toolset_has_keys(ts_key, config)
return not _toolset_has_keys(ts_key, config, force_fresh=force_fresh)
# If any visible provider has a registered post_setup install-state
# check that hasn't been satisfied (e.g. cua-driver binary not on
# PATH yet), force the configuration flow so `_configure_provider`
# invokes `_run_post_setup` and the install actually runs.
for provider in _visible_providers(cat, config):
for provider in _visible_providers(cat, config, force_fresh=force_fresh):
post_setup = provider.get("post_setup")
if post_setup and not _post_setup_already_installed(post_setup):
return True
@ -1962,18 +2007,25 @@ def _toolset_needs_configuration_prompt(ts_key: str, config: dict) -> bool:
pass
return True
return not _toolset_has_keys(ts_key, config)
return not _toolset_has_keys(ts_key, config, force_fresh=force_fresh)
def _configure_tool_category(ts_key: str, cat: dict, config: dict):
def _configure_tool_category(
ts_key: str,
cat: dict,
config: dict,
*,
force_fresh: bool = True,
):
"""Configure a tool category with provider selection."""
icon = cat.get("icon", "")
name = cat["name"]
providers = _visible_providers(cat, config)
providers = _visible_providers(cat, config, force_fresh=force_fresh)
hidden_nous_message = _hidden_nous_gateway_message(
cat,
config,
f"the Nous Subscription provider for {name}",
force_fresh=force_fresh,
)
# Check Python version requirement
@ -1998,7 +2050,7 @@ def _configure_tool_category(ts_key: str, cat: dict, config: dict):
if hidden_nous_message:
for line in hidden_nous_message.splitlines():
_print_warning(f" {line}")
_configure_provider(provider, config)
_configure_provider(provider, config, force_fresh=force_fresh)
else:
# Multiple providers - let user choose
print()
@ -2018,7 +2070,10 @@ def _configure_tool_category(ts_key: str, cat: dict, config: dict):
# obvious which options cost extra vs. cost nothing on top of Nous.
try:
_nous_logged_in = bool(
get_nous_subscription_features(config).nous_auth_present
get_nous_subscription_features(
config,
force_fresh=force_fresh,
).nous_auth_present
)
except Exception:
_nous_logged_in = False
@ -2030,7 +2085,7 @@ def _configure_tool_category(ts_key: str, cat: dict, config: dict):
configured = ""
env_vars = p.get("env_vars", [])
if not env_vars or all(get_env_value(v["key"]) for v in env_vars):
if _is_provider_active(p, config):
if _is_provider_active(p, config, force_fresh=force_fresh):
configured = " [active]"
elif not env_vars:
configured = ""
@ -2050,7 +2105,11 @@ def _configure_tool_category(ts_key: str, cat: dict, config: dict):
provider_choices.append("Skip — keep defaults / configure later")
# Detect current provider as default
default_idx = _detect_active_provider_index(providers, config)
default_idx = _detect_active_provider_index(
providers,
config,
force_fresh=force_fresh,
)
provider_idx = _prompt_choice(f" {title}:", provider_choices, default_idx)
@ -2059,10 +2118,15 @@ def _configure_tool_category(ts_key: str, cat: dict, config: dict):
_print_info(f" Skipped {name}")
return
_configure_provider(providers[provider_idx], config)
_configure_provider(providers[provider_idx], config, force_fresh=force_fresh)
def _is_provider_active(provider: dict, config: dict) -> bool:
def _is_provider_active(
provider: dict,
config: dict,
*,
force_fresh: bool = False,
) -> bool:
"""Check if a provider entry matches the currently active config."""
plugin_name = provider.get("image_gen_plugin_name")
if plugin_name:
@ -2076,7 +2140,7 @@ def _is_provider_active(provider: dict, config: dict) -> bool:
managed_feature = provider.get("managed_nous_feature")
if managed_feature:
features = get_nous_subscription_features(config)
features = get_nous_subscription_features(config, force_fresh=force_fresh)
feature = features.features.get(managed_feature)
if feature is None:
return False
@ -2123,10 +2187,15 @@ def _is_provider_active(provider: dict, config: dict) -> bool:
return False
def _detect_active_provider_index(providers: list, config: dict) -> int:
def _detect_active_provider_index(
providers: list,
config: dict,
*,
force_fresh: bool = False,
) -> int:
"""Return the index of the currently active provider, or 0."""
for i, p in enumerate(providers):
if _is_provider_active(p, config):
if _is_provider_active(p, config, force_fresh=force_fresh):
return i
# Fallback: env vars present → likely configured
env_vars = p.get("env_vars", [])
@ -2429,13 +2498,18 @@ def _select_plugin_video_gen_provider(plugin_name: str, config: dict) -> None:
_configure_videogen_model_for_plugin(plugin_name, config)
def _configure_provider(provider: dict, config: dict):
def _configure_provider(
provider: dict,
config: dict,
*,
force_fresh: bool = True,
):
"""Configure a single provider - prompt for API keys and set config."""
env_vars = provider.get("env_vars", [])
managed_feature = provider.get("managed_nous_feature")
if provider.get("requires_nous_auth"):
features = get_nous_subscription_features(config)
features = get_nous_subscription_features(config, force_fresh=force_fresh)
entitled = bool(
features.account_info and features.account_info.paid_service_access is True
)
@ -2536,7 +2610,10 @@ def _configure_provider(provider: dict, config: dict):
_has_managed_sibling = True
break
if _has_managed_sibling:
_features = get_nous_subscription_features(config)
_features = get_nous_subscription_features(
config,
force_fresh=force_fresh,
)
_show_portal_hint = not _features.nous_auth_present
except Exception:
_show_portal_hint = False
@ -2654,7 +2731,11 @@ def _configure_simple_requirements(ts_key: str):
_print_warning(" Skipped")
def _reconfigure_tool(config: dict):
def _reconfigure_tool(
config: dict,
*,
force_fresh: bool = True,
):
"""Let user reconfigure an existing tool's provider or API key."""
# Build list of configurable tools that are currently set up
configurable = []
@ -2662,7 +2743,10 @@ def _reconfigure_tool(config: dict):
cat = TOOL_CATEGORIES.get(ts_key)
reqs = TOOLSET_ENV_REQUIREMENTS.get(ts_key)
if cat or reqs:
if _toolset_has_keys(ts_key, config) or _toolset_enabled_for_reconfigure(ts_key, config):
if (
_toolset_has_keys(ts_key, config, force_fresh=force_fresh)
or _toolset_enabled_for_reconfigure(ts_key, config)
):
configurable.append((ts_key, ts_label))
if not configurable:
@ -2681,7 +2765,12 @@ def _reconfigure_tool(config: dict):
cat = TOOL_CATEGORIES.get(ts_key)
if cat:
_configure_tool_category_for_reconfig(ts_key, cat, config)
_configure_tool_category_for_reconfig(
ts_key,
cat,
config,
force_fresh=force_fresh,
)
else:
_reconfigure_simple_requirements(ts_key)
@ -2710,15 +2799,22 @@ def _toolset_enabled_for_reconfigure(ts_key: str, config: dict) -> bool:
return False
def _configure_tool_category_for_reconfig(ts_key: str, cat: dict, config: dict):
def _configure_tool_category_for_reconfig(
ts_key: str,
cat: dict,
config: dict,
*,
force_fresh: bool = True,
):
"""Reconfigure a tool category - provider selection + API key update."""
icon = cat.get("icon", "")
name = cat["name"]
providers = _visible_providers(cat, config)
providers = _visible_providers(cat, config, force_fresh=force_fresh)
hidden_nous_message = _hidden_nous_gateway_message(
cat,
config,
f"the Nous Subscription provider for {name}",
force_fresh=force_fresh,
)
if len(providers) == 1:
@ -2728,7 +2824,7 @@ def _configure_tool_category_for_reconfig(ts_key: str, cat: dict, config: dict):
if hidden_nous_message:
for line in hidden_nous_message.splitlines():
_print_warning(f" {line}")
_reconfigure_provider(provider, config)
_reconfigure_provider(provider, config, force_fresh=force_fresh)
else:
print()
print(color(f" --- {icon} {name} - Choose a provider ---", Colors.CYAN))
@ -2744,7 +2840,7 @@ def _configure_tool_category_for_reconfig(ts_key: str, cat: dict, config: dict):
configured = ""
env_vars = p.get("env_vars", [])
if not env_vars or all(get_env_value(v["key"]) for v in env_vars):
if _is_provider_active(p, config):
if _is_provider_active(p, config, force_fresh=force_fresh):
configured = " [active]"
elif not env_vars:
configured = ""
@ -2752,19 +2848,32 @@ def _configure_tool_category_for_reconfig(ts_key: str, cat: dict, config: dict):
configured = " [configured]"
provider_choices.append(f"{p['name']}{badge}{tag}{configured}")
default_idx = _detect_active_provider_index(providers, config)
default_idx = _detect_active_provider_index(
providers,
config,
force_fresh=force_fresh,
)
provider_idx = _prompt_choice(" Select provider:", provider_choices, default_idx)
_reconfigure_provider(providers[provider_idx], config)
_reconfigure_provider(
providers[provider_idx],
config,
force_fresh=force_fresh,
)
def _reconfigure_provider(provider: dict, config: dict):
def _reconfigure_provider(
provider: dict,
config: dict,
*,
force_fresh: bool = True,
):
"""Reconfigure a provider - update API keys."""
env_vars = provider.get("env_vars", [])
managed_feature = provider.get("managed_nous_feature")
if provider.get("requires_nous_auth"):
features = get_nous_subscription_features(config)
features = get_nous_subscription_features(config, force_fresh=force_fresh)
entitled = bool(
features.account_info and features.account_info.paid_service_access is True
)
@ -2976,11 +3085,11 @@ def tools_command(args=None, first_install: bool = False, config: dict = None):
auto_configured = apply_nous_managed_defaults(
config,
enabled_toolsets=new_enabled,
force_fresh=True,
)
if managed_nous_tools_enabled():
for ts_key in sorted(auto_configured):
label = next((l for k, l, _ in CONFIGURABLE_TOOLSETS if k == ts_key), ts_key)
print(color(f"{label}: using your Nous subscription defaults", Colors.GREEN))
for ts_key in sorted(auto_configured):
label = next((l for k, l, _ in CONFIGURABLE_TOOLSETS if k == ts_key), ts_key)
print(color(f"{label}: using your Nous subscription defaults", Colors.GREEN))
# Walk through ALL selected tools that have provider options or
# need API keys. This ensures browser (Local vs Browserbase),
@ -3048,7 +3157,7 @@ def tools_command(args=None, first_install: bool = False, config: dict = None):
# "Reconfigure" selected
if idx == _reconfig_idx:
_reconfigure_tool(config)
_reconfigure_tool(config, force_fresh=True)
print()
continue
@ -3064,7 +3173,11 @@ def tools_command(args=None, first_install: bool = False, config: dict = None):
all_current = set()
for pk in platform_keys:
all_current |= _get_platform_tools(config, pk, include_default_mcp_servers=False)
new_enabled = _prompt_toolset_checklist("All platforms", all_current)
new_enabled = _prompt_toolset_checklist(
"All platforms",
all_current,
force_fresh=True,
)
if new_enabled != all_current:
for pk in platform_keys:
prev = _get_platform_tools(config, pk, include_default_mcp_servers=False)
@ -3082,7 +3195,11 @@ def tools_command(args=None, first_install: bool = False, config: dict = None):
# Configure API keys for newly enabled tools
for ts_key in sorted(added):
if (TOOL_CATEGORIES.get(ts_key) or TOOLSET_ENV_REQUIREMENTS.get(ts_key)):
if _toolset_needs_configuration_prompt(ts_key, config):
if _toolset_needs_configuration_prompt(
ts_key,
config,
force_fresh=True,
):
_configure_toolset(ts_key, config)
_save_platform_tools(config, pk, new_enabled)
save_config(config)
@ -3104,7 +3221,11 @@ def tools_command(args=None, first_install: bool = False, config: dict = None):
current_enabled = _get_platform_tools(config, pkey, include_default_mcp_servers=False)
# Show checklist
new_enabled = _prompt_toolset_checklist(pinfo["label"], current_enabled)
new_enabled = _prompt_toolset_checklist(
pinfo["label"],
current_enabled,
force_fresh=True,
)
if new_enabled != current_enabled:
added = new_enabled - current_enabled
@ -3122,7 +3243,11 @@ def tools_command(args=None, first_install: bool = False, config: dict = None):
# Configure newly enabled toolsets that need API keys
for ts_key in sorted(added):
if (TOOL_CATEGORIES.get(ts_key) or TOOLSET_ENV_REQUIREMENTS.get(ts_key)):
if _toolset_needs_configuration_prompt(ts_key, config):
if _toolset_needs_configuration_prompt(
ts_key,
config,
force_fresh=True,
):
_configure_toolset(ts_key, config)
_save_platform_tools(config, pkey, new_enabled)

View file

@ -237,7 +237,7 @@ class TestConfigWriting:
monkeypatch.setattr(
tools_config,
"get_nous_subscription_features",
lambda config: SimpleNamespace(
lambda config, **kwargs: SimpleNamespace(
features={"image_gen": SimpleNamespace(managed_by_nous=True)}
),
)

View file

@ -34,6 +34,28 @@ def test_get_nous_subscription_features_recognizes_direct_exa_backend(monkeypatc
assert features.web.current_provider == "exa"
def test_get_nous_subscription_features_force_fresh_forwards_account_request(monkeypatch):
calls = []
def fake_account_info(*, force_fresh=False):
calls.append(force_fresh)
return _account(logged_in=True, paid=True)
monkeypatch.setattr(ns, "get_env_value", lambda name: "")
monkeypatch.setattr(ns, "get_nous_portal_account_info", fake_account_info)
monkeypatch.setattr(ns, "_toolset_enabled", lambda config, key: False)
monkeypatch.setattr(ns, "_has_agent_browser", lambda: False)
monkeypatch.setattr(ns, "resolve_openai_audio_api_key", lambda: "")
monkeypatch.setattr(ns, "has_direct_modal_credentials", lambda: False)
monkeypatch.setattr(ns, "is_managed_tool_gateway_ready", lambda vendor: False)
features = ns.get_nous_subscription_features({}, force_fresh=True)
assert features.account_info is not None
assert features.account_info.paid_service_access is True
assert calls == [True]
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: "")

View file

@ -133,37 +133,6 @@ def test_show_status_reports_nous_inference_key_without_portal_login(monkeypatch
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

@ -1,5 +1,6 @@
"""Tests for hermes_cli.tools_config platform tool persistence."""
from types import SimpleNamespace
from unittest.mock import patch
import pytest
@ -554,7 +555,6 @@ def test_save_platform_tools_still_preserves_mcp_with_platform_default_present()
def test_visible_providers_include_nous_subscription_when_logged_in(monkeypatch):
monkeypatch.setattr("hermes_cli.tools_config.managed_nous_tools_enabled", lambda: True)
config = {"model": {"provider": "nous"}}
monkeypatch.setattr(
@ -572,18 +572,48 @@ def test_visible_providers_include_nous_subscription_when_logged_in(monkeypatch)
assert providers[0]["name"].startswith("Nous Subscription")
def test_visible_providers_hide_nous_subscription_when_feature_flag_is_off(monkeypatch):
monkeypatch.setattr("hermes_cli.tools_config.managed_nous_tools_enabled", lambda: False)
def test_visible_providers_force_fresh_shows_nous_subscription_after_upgrade(monkeypatch):
calls = []
def fake_subscription_features(config, *, force_fresh=False):
calls.append(("features", force_fresh))
return SimpleNamespace(
nous_auth_present=True,
account_info=NousPortalAccountInfo(
logged_in=True,
source="account_api" if force_fresh else "jwt",
fresh=force_fresh,
paid_service_access=True if force_fresh else False,
),
features={},
)
monkeypatch.setattr(
"hermes_cli.tools_config.get_nous_subscription_features",
fake_subscription_features,
)
providers = _visible_providers(
TOOL_CATEGORIES["browser"],
{"model": {"provider": "nous"}},
force_fresh=True,
)
assert providers[0]["name"].startswith("Nous Subscription")
assert ("features", True) in calls
def test_visible_providers_hide_nous_subscription_when_paid_access_is_false(monkeypatch):
config = {"model": {"provider": "nous"}}
monkeypatch.setattr(
"hermes_cli.nous_subscription.get_nous_portal_account_info",
lambda: NousPortalAccountInfo(
logged_in=True,
source="jwt",
fresh=False,
paid_service_access=True,
),
logged_in=True,
source="jwt",
fresh=False,
paid_service_access=False,
),
)
providers = _visible_providers(TOOL_CATEGORIES["browser"], config)
@ -612,7 +642,7 @@ def test_reconfigure_lists_enabled_web_without_existing_provider_config(monkeypa
monkeypatch.setattr(
"hermes_cli.tools_config._toolset_has_keys",
lambda ts_key, config=None: False,
lambda ts_key, config=None, **kwargs: False,
)
def fake_prompt_choice(question, choices, default=0):
@ -622,7 +652,7 @@ def test_reconfigure_lists_enabled_web_without_existing_provider_config(monkeypa
monkeypatch.setattr("hermes_cli.tools_config._prompt_choice", fake_prompt_choice)
monkeypatch.setattr(
"hermes_cli.tools_config._configure_tool_category_for_reconfig",
lambda ts_key, cat, config: configured.append(ts_key),
lambda ts_key, cat, config, **kwargs: configured.append(ts_key),
)
monkeypatch.setattr("hermes_cli.tools_config.save_config", lambda config: None)
@ -633,7 +663,6 @@ def test_reconfigure_lists_enabled_web_without_existing_provider_config(monkeypa
def test_first_install_nous_auto_configures_managed_defaults(monkeypatch):
monkeypatch.setattr("hermes_cli.tools_config.managed_nous_tools_enabled", lambda: True)
monkeypatch.setattr("hermes_cli.nous_subscription.managed_nous_tools_enabled", lambda: True)
config = {
"model": {"provider": "nous"},
@ -669,7 +698,7 @@ def test_first_install_nous_auto_configures_managed_defaults(monkeypatch):
)
monkeypatch.setattr(
"hermes_cli.nous_subscription.get_nous_portal_account_info",
lambda: NousPortalAccountInfo(
lambda *args, **kwargs: NousPortalAccountInfo(
logged_in=True,
source="jwt",
fresh=False,

View file

@ -71,6 +71,26 @@ class TestManagedNousToolsEnabled:
)
assert managed_nous_tools_enabled() is True
def test_force_fresh_is_forwarded(self, monkeypatch):
calls = []
def fake_account_info(*, force_fresh=False):
calls.append(force_fresh)
return NousPortalAccountInfo(
logged_in=True,
source="account_api",
fresh=True,
paid_service_access=True,
)
monkeypatch.setattr(
"hermes_cli.nous_account.get_nous_portal_account_info",
fake_account_info,
)
assert managed_nous_tools_enabled(force_fresh=True) is True
assert calls == [True]
def test_returns_false_on_exception(self, monkeypatch):
"""Should never crash — returns False on any exception."""
monkeypatch.setattr(

View file

@ -14,16 +14,21 @@ _DEFAULT_MODAL_MODE = "auto"
_VALID_MODAL_MODES = {"auto", "direct", "managed"}
def managed_nous_tools_enabled() -> bool:
def managed_nous_tools_enabled(*, force_fresh: bool = False) -> bool:
"""Return True when the user has paid Nous Portal service access.
Tool Gateway availability fails closed on unknown/error entitlement. We
intentionally catch all exceptions and return False never block startup.
``force_fresh=True`` is for interactive configuration flows that should
reflect a just-purchased subscription or credits immediately.
"""
try:
from hermes_cli.nous_account import get_nous_portal_account_info
account_info = get_nous_portal_account_info()
if force_fresh:
account_info = get_nous_portal_account_info(force_fresh=True)
else:
account_info = get_nous_portal_account_info()
if not account_info.logged_in:
return False
return account_info.paid_service_access is True
@ -90,6 +95,7 @@ def resolve_modal_backend_state(
*,
has_direct: bool,
managed_ready: bool,
managed_enabled: bool | None = None,
) -> Dict[str, Any]:
"""Resolve direct vs managed Modal backend selection.
@ -100,16 +106,18 @@ def resolve_modal_backend_state(
"""
requested_mode = coerce_modal_mode(modal_mode)
normalized_mode = normalize_modal_mode(modal_mode)
if managed_enabled is None:
managed_enabled = managed_nous_tools_enabled()
managed_mode_blocked = (
requested_mode == "managed" and not managed_nous_tools_enabled()
requested_mode == "managed" and not managed_enabled
)
if normalized_mode == "managed":
selected_backend = "managed" if managed_nous_tools_enabled() and managed_ready else None
selected_backend = "managed" if managed_enabled and managed_ready else None
elif normalized_mode == "direct":
selected_backend = "direct" if has_direct else None
else:
selected_backend = "managed" if managed_nous_tools_enabled() and managed_ready else "direct" if has_direct else None
selected_backend = "managed" if managed_enabled and managed_ready else "direct" if has_direct else None
return {
"requested_mode": requested_mode,