common/hermes
2
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Merge branch 'main' of github.com:NousResearch/hermes-agent into feat/ink-refactor

Browse Source
This commit is contained in:
Brooklyn Nicholson
2026-04-08 19:11:44 -05:00
parent a435c7274a 7d26feb9a3
commit f226e6be10
391 changed files with 20699 additions and 2902 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
hermes_cli/__init__.py
View File

@@ -11,5 +11,5 @@ Provides subcommands for:
- hermes cron - Manage cron jobs
"""
__version__ = "0.7.0"
__release_date__ = "2026.4.3"
__version__ = "0.8.0"
__release_date__ = "2026.4.8"

422
hermes_cli/auth.py
View File

@@ -37,7 +37,7 @@ from typing import Any, Dict, List, Optional
import httpx
import yaml
from hermes_cli.config import get_hermes_home, get_config_path
from hermes_cli.config import get_hermes_home, get_config_path, read_raw_config
from hermes_constants import OPENROUTER_BASE_URL
logger = logging.getLogger(__name__)
@@ -67,12 +67,16 @@ DEFAULT_AGENT_KEY_MIN_TTL_SECONDS = 30 * 60 # 30 minutes
ACCESS_TOKEN_REFRESH_SKEW_SECONDS = 120 # refresh 2 min before expiry
DEVICE_AUTH_POLL_INTERVAL_CAP_SECONDS = 1 # poll at most every 1s
DEFAULT_CODEX_BASE_URL = "https://chatgpt.com/backend-api/codex"
DEFAULT_QWEN_BASE_URL = "https://portal.qwen.ai/v1"
DEFAULT_GITHUB_MODELS_BASE_URL = "https://api.githubcopilot.com"
DEFAULT_COPILOT_ACP_BASE_URL = "acp://copilot"
DEFAULT_GEMINI_BASE_URL = "https://generativelanguage.googleapis.com/v1beta/openai"
CODEX_OAUTH_CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann"
CODEX_OAUTH_TOKEN_URL = "https://auth.openai.com/oauth/token"
CODEX_ACCESS_TOKEN_REFRESH_SKEW_SECONDS = 120
QWEN_OAUTH_CLIENT_ID = "f0304373b74a44d2b584a3fb70ca9e56"
QWEN_OAUTH_TOKEN_URL = "https://chat.qwen.ai/api/v1/oauth2/token"
QWEN_ACCESS_TOKEN_REFRESH_SKEW_SECONDS = 120
# =============================================================================
@@ -112,6 +116,12 @@ PROVIDER_REGISTRY: Dict[str, ProviderConfig] = {
auth_type="oauth_external",
inference_base_url=DEFAULT_CODEX_BASE_URL,
),
"qwen-oauth": ProviderConfig(
id="qwen-oauth",
name="Qwen OAuth",
auth_type="oauth_external",
inference_base_url=DEFAULT_QWEN_BASE_URL,
),
"copilot": ProviderConfig(
id="copilot",
name="GitHub Copilot",
@@ -404,6 +414,47 @@ def detect_zai_endpoint(api_key: str, timeout: float = 8.0) -> Optional[Dict[str
return None
def _resolve_zai_base_url(api_key: str, default_url: str, env_override: str) -> str:
"""Return the correct Z.AI base URL by probing endpoints.
If the user has explicitly set GLM_BASE_URL, that always wins.
Otherwise, probe the candidate endpoints to find one that accepts the
key. The detected endpoint is cached in provider state (auth.json) keyed
on a hash of the API key so subsequent starts skip the probe.
"""
if env_override:
return env_override
# Check provider-state cache for a previously-detected endpoint.
auth_store = _load_auth_store()
state = _load_provider_state(auth_store, "zai") or {}
cached = state.get("detected_endpoint")
if isinstance(cached, dict) and cached.get("base_url"):
key_hash = cached.get("key_hash", "")
if key_hash == hashlib.sha256(api_key.encode()).hexdigest()[:16]:
logger.debug("Z.AI: using cached endpoint %s", cached["base_url"])
return cached["base_url"]
# Probe — may take up to ~8s per endpoint.
detected = detect_zai_endpoint(api_key)
if detected and detected.get("base_url"):
# Persist the detection result keyed on the API key hash.
key_hash = hashlib.sha256(api_key.encode()).hexdigest()[:16]
state["detected_endpoint"] = {
"base_url": detected["base_url"],
"endpoint_id": detected.get("id", ""),
"model": detected.get("model", ""),
"label": detected.get("label", ""),
"key_hash": key_hash,
}
_save_provider_state(auth_store, "zai", state)
logger.info("Z.AI: auto-detected endpoint %s (%s)", detected["label"], detected["base_url"])
return detected["base_url"]
logger.debug("Z.AI: probe failed, falling back to default %s", default_url)
return default_url
# =============================================================================
# Error Types
# =============================================================================
@@ -776,6 +827,7 @@ def resolve_provider(
"github-copilot-acp": "copilot-acp", "copilot-acp-agent": "copilot-acp",
"aigateway": "ai-gateway", "vercel": "ai-gateway", "vercel-ai-gateway": "ai-gateway",
"opencode": "opencode-zen", "zen": "opencode-zen",
"qwen-portal": "qwen-oauth", "qwen-cli": "qwen-oauth", "qwen-oauth": "qwen-oauth",
"hf": "huggingface", "hugging-face": "huggingface", "huggingface-hub": "huggingface",
"go": "opencode-go", "opencode-go-sub": "opencode-go",
"kilo": "kilocode", "kilo-code": "kilocode", "kilo-gateway": "kilocode",
@@ -905,6 +957,176 @@ def _codex_access_token_is_expiring(access_token: Any, skew_seconds: int) -> boo
return float(exp) <= (time.time() + max(0, int(skew_seconds)))
def _qwen_cli_auth_path() -> Path:
return Path.home() / ".qwen" / "oauth_creds.json"
def _read_qwen_cli_tokens() -> Dict[str, Any]:
auth_path = _qwen_cli_auth_path()
if not auth_path.exists():
raise AuthError(
"Qwen CLI credentials not found. Run 'qwen auth qwen-oauth' first.",
provider="qwen-oauth",
code="qwen_auth_missing",
)
try:
data = json.loads(auth_path.read_text(encoding="utf-8"))
except Exception as exc:
raise AuthError(
f"Failed to read Qwen CLI credentials from {auth_path}: {exc}",
provider="qwen-oauth",
code="qwen_auth_read_failed",
) from exc
if not isinstance(data, dict):
raise AuthError(
f"Invalid Qwen CLI credentials in {auth_path}.",
provider="qwen-oauth",
code="qwen_auth_invalid",
)
return data
def _save_qwen_cli_tokens(tokens: Dict[str, Any]) -> Path:
auth_path = _qwen_cli_auth_path()
auth_path.parent.mkdir(parents=True, exist_ok=True)
tmp_path = auth_path.with_suffix(".tmp")
tmp_path.write_text(json.dumps(tokens, indent=2, sort_keys=True) + "\n", encoding="utf-8")
os.chmod(tmp_path, stat.S_IRUSR | stat.S_IWUSR)
tmp_path.replace(auth_path)
return auth_path
def _qwen_access_token_is_expiring(expiry_date_ms: Any, skew_seconds: int = QWEN_ACCESS_TOKEN_REFRESH_SKEW_SECONDS) -> bool:
try:
expiry_ms = int(expiry_date_ms)
except Exception:
return True
return (time.time() + max(0, int(skew_seconds))) * 1000 >= expiry_ms
def _refresh_qwen_cli_tokens(tokens: Dict[str, Any], timeout_seconds: float = 20.0) -> Dict[str, Any]:
refresh_token = str(tokens.get("refresh_token", "") or "").strip()
if not refresh_token:
raise AuthError(
"Qwen OAuth refresh token missing. Re-run 'qwen auth qwen-oauth'.",
provider="qwen-oauth",
code="qwen_refresh_token_missing",
)
try:
response = httpx.post(
QWEN_OAUTH_TOKEN_URL,
headers={
"Content-Type": "application/x-www-form-urlencoded",
"Accept": "application/json",
},
data={
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": QWEN_OAUTH_CLIENT_ID,
},
timeout=timeout_seconds,
)
except Exception as exc:
raise AuthError(
f"Qwen OAuth refresh failed: {exc}",
provider="qwen-oauth",
code="qwen_refresh_failed",
) from exc
if response.status_code >= 400:
body = response.text.strip()
raise AuthError(
"Qwen OAuth refresh failed. Re-run 'qwen auth qwen-oauth'."
+ (f" Response: {body}" if body else ""),
provider="qwen-oauth",
code="qwen_refresh_failed",
)
try:
payload = response.json()
except Exception as exc:
raise AuthError(
f"Qwen OAuth refresh returned invalid JSON: {exc}",
provider="qwen-oauth",
code="qwen_refresh_invalid_json",
) from exc
if not isinstance(payload, dict) or not str(payload.get("access_token", "") or "").strip():
raise AuthError(
"Qwen OAuth refresh response missing access_token.",
provider="qwen-oauth",
code="qwen_refresh_invalid_response",
)
expires_in = payload.get("expires_in")
try:
expires_in_seconds = int(expires_in)
except Exception:
expires_in_seconds = 6 * 60 * 60
refreshed = {
"access_token": str(payload.get("access_token", "") or "").strip(),
"refresh_token": str(payload.get("refresh_token", refresh_token) or refresh_token).strip(),
"token_type": str(payload.get("token_type", tokens.get("token_type", "Bearer")) or "Bearer").strip() or "Bearer",
"resource_url": str(payload.get("resource_url", tokens.get("resource_url", "portal.qwen.ai")) or "portal.qwen.ai").strip(),
"expiry_date": int(time.time() * 1000) + max(1, expires_in_seconds) * 1000,
}
_save_qwen_cli_tokens(refreshed)
return refreshed
def resolve_qwen_runtime_credentials(
*,
force_refresh: bool = False,
refresh_if_expiring: bool = True,
refresh_skew_seconds: int = QWEN_ACCESS_TOKEN_REFRESH_SKEW_SECONDS,
) -> Dict[str, Any]:
tokens = _read_qwen_cli_tokens()
access_token = str(tokens.get("access_token", "") or "").strip()
should_refresh = bool(force_refresh)
if not should_refresh and refresh_if_expiring:
should_refresh = _qwen_access_token_is_expiring(tokens.get("expiry_date"), refresh_skew_seconds)
if should_refresh:
tokens = _refresh_qwen_cli_tokens(tokens)
access_token = str(tokens.get("access_token", "") or "").strip()
if not access_token:
raise AuthError(
"Qwen OAuth access token missing. Re-run 'qwen auth qwen-oauth'.",
provider="qwen-oauth",
code="qwen_access_token_missing",
)
base_url = os.getenv("HERMES_QWEN_BASE_URL", "").strip().rstrip("/") or DEFAULT_QWEN_BASE_URL
return {
"provider": "qwen-oauth",
"base_url": base_url,
"api_key": access_token,
"source": "qwen-cli",
"expires_at_ms": tokens.get("expiry_date"),
"auth_file": str(_qwen_cli_auth_path()),
}
def get_qwen_auth_status() -> Dict[str, Any]:
auth_path = _qwen_cli_auth_path()
try:
creds = resolve_qwen_runtime_credentials(refresh_if_expiring=False)
return {
"logged_in": True,
"auth_file": str(auth_path),
"source": creds.get("source"),
"api_key": creds.get("api_key"),
"expires_at_ms": creds.get("expires_at_ms"),
}
except AuthError as exc:
return {
"logged_in": False,
"auth_file": str(auth_path),
"error": str(exc),
}
# =============================================================================
# SSH / remote session detection
# =============================================================================
@@ -936,7 +1158,7 @@ def _read_codex_tokens(*, _lock: bool = True) -> Dict[str, Any]:
state = _load_provider_state(auth_store, "openai-codex")
if not state:
raise AuthError(
"No Codex credentials stored. Run `hermes login` to authenticate.",
"No Codex credentials stored. Run `hermes auth` to authenticate.",
provider="openai-codex",
code="codex_auth_missing",
relogin_required=True,
@@ -944,7 +1166,7 @@ def _read_codex_tokens(*, _lock: bool = True) -> Dict[str, Any]:
tokens = state.get("tokens")
if not isinstance(tokens, dict):
raise AuthError(
"Codex auth state is missing tokens. Run `hermes login` to re-authenticate.",
"Codex auth state is missing tokens. Run `hermes auth` to re-authenticate.",
provider="openai-codex",
code="codex_auth_invalid_shape",
relogin_required=True,
@@ -953,14 +1175,14 @@ def _read_codex_tokens(*, _lock: bool = True) -> Dict[str, Any]:
refresh_token = tokens.get("refresh_token")
if not isinstance(access_token, str) or not access_token.strip():
raise AuthError(
"Codex auth is missing access_token. Run `hermes login` to re-authenticate.",
"Codex auth is missing access_token. Run `hermes auth` to re-authenticate.",
provider="openai-codex",
code="codex_auth_missing_access_token",
relogin_required=True,
)
if not isinstance(refresh_token, str) or not refresh_token.strip():
raise AuthError(
"Codex auth is missing refresh_token. Run `hermes login` to re-authenticate.",
"Codex auth is missing refresh_token. Run `hermes auth` to re-authenticate.",
provider="openai-codex",
code="codex_auth_missing_refresh_token",
relogin_required=True,
@@ -995,7 +1217,7 @@ def refresh_codex_oauth_pure(
del access_token # Access token is only used by callers to decide whether to refresh.
if not isinstance(refresh_token, str) or not refresh_token.strip():
raise AuthError(
"Codex auth is missing refresh_token. Run `hermes login` to re-authenticate.",
"Codex auth is missing refresh_token. Run `hermes auth` to re-authenticate.",
provider="openai-codex",
code="codex_auth_missing_refresh_token",
relogin_required=True,
@@ -1030,6 +1252,14 @@ def refresh_codex_oauth_pure(
pass
if code in {"invalid_grant", "invalid_token", "invalid_request"}:
relogin_required = True
if code == "refresh_token_reused":
message = (
"Codex refresh token was already consumed by another client "
"(e.g. Codex CLI or VS Code extension). "
"Run `codex` in your terminal to generate fresh tokens, "
"then run `hermes auth` to re-authenticate."
)
relogin_required = True
raise AuthError(
message,
provider="openai-codex",
@@ -1091,7 +1321,8 @@ def _refresh_codex_auth_tokens(
def _import_codex_cli_tokens() -> Optional[Dict[str, str]]:
"""Try to read tokens from ~/.codex/auth.json (Codex CLI shared file).
Returns tokens dict if valid, None otherwise. Does NOT write to the shared file.
Returns tokens dict if valid and not expired, None otherwise.
Does NOT write to the shared file.
"""
codex_home = os.getenv("CODEX_HOME", "").strip()
if not codex_home:
@@ -1104,7 +1335,17 @@ def _import_codex_cli_tokens() -> Optional[Dict[str, str]]:
tokens = payload.get("tokens")
if not isinstance(tokens, dict):
return None
if not tokens.get("access_token") or not tokens.get("refresh_token"):
access_token = tokens.get("access_token")
refresh_token = tokens.get("refresh_token")
if not access_token or not refresh_token:
return None
# Reject expired tokens — importing stale tokens from ~/.codex/
# that can't be refreshed leaves the user stuck with "Login successful!"
# but no working credentials.
if _codex_access_token_is_expiring(access_token, 0):
logger.debug(
"Codex CLI tokens at %s are expired — skipping import.", auth_path,
)
return None
return dict(tokens)
except Exception:
@@ -1132,7 +1373,7 @@ def resolve_codex_runtime_credentials(
logger.info("Migrating Codex credentials from ~/.codex/ to Hermes auth store")
print("⚠️ Migrating Codex credentials to Hermes's own auth store.")
print(" This avoids conflicts with Codex CLI and VS Code.")
print(" Run `hermes login` to create a fully independent session.\n")
print(" Run `hermes auth` to create a fully independent session.\n")
_save_codex_tokens(cli_tokens)
data = _read_codex_tokens()
else:
@@ -1896,7 +2137,36 @@ def get_nous_auth_status() -> Dict[str, Any]:
def get_codex_auth_status() -> Dict[str, Any]:
"""Status snapshot for Codex auth."""
"""Status snapshot for Codex auth.
Checks the credential pool first (where `hermes auth` stores credentials),
then falls back to the legacy provider state.
"""
# Check credential pool first — this is where `hermes auth` and
# `hermes model` store device_code tokens.
try:
from agent.credential_pool import load_pool
pool = load_pool("openai-codex")
if pool and pool.has_credentials():
entry = pool.select()
if entry is not None:
api_key = (
getattr(entry, "runtime_api_key", None)
or getattr(entry, "access_token", "")
)
if api_key and not _codex_access_token_is_expiring(api_key, 0):
return {
"logged_in": True,
"auth_store": str(_auth_file_path()),
"last_refresh": getattr(entry, "last_refresh", None),
"auth_mode": "chatgpt",
"source": f"pool:{getattr(entry, 'label', 'unknown')}",
"api_key": api_key,
}
except Exception:
pass
# Fall back to legacy provider state
try:
creds = resolve_codex_runtime_credentials()
return {
@@ -1905,6 +2175,7 @@ def get_codex_auth_status() -> Dict[str, Any]:
"last_refresh": creds.get("last_refresh"),
"auth_mode": creds.get("auth_mode"),
"source": creds.get("source"),
"api_key": creds.get("api_key"),
}
except AuthError as exc:
return {
@@ -1982,6 +2253,8 @@ def get_auth_status(provider_id: Optional[str] = None) -> Dict[str, Any]:
return get_nous_auth_status()
if target == "openai-codex":
return get_codex_auth_status()
if target == "qwen-oauth":
return get_qwen_auth_status()
if target == "copilot-acp":
return get_external_process_provider_status(target)
# API-key providers
@@ -2014,6 +2287,8 @@ def resolve_api_key_provider_credentials(provider_id: str) -> Dict[str, Any]:
if provider_id == "kimi-coding":
base_url = _resolve_kimi_base_url(api_key, pconfig.inference_base_url, env_url)
elif provider_id == "zai":
base_url = _resolve_zai_base_url(api_key, pconfig.inference_base_url, env_url)
elif env_url:
base_url = env_url.rstrip("/")
else:
@@ -2088,7 +2363,7 @@ def detect_external_credentials() -> List[Dict[str, Any]]:
found.append({
"provider": "openai-codex",
"path": str(codex_path),
"label": f"Codex CLI credentials found ({codex_path}) — run `hermes login` to create a separate session",
"label": f"Codex CLI credentials found ({codex_path}) — run `hermes auth` to create a separate session",
})
return found
@@ -2122,14 +2397,7 @@ def _update_config_for_provider(
config_path = get_config_path()
config_path.parent.mkdir(parents=True, exist_ok=True)
config: Dict[str, Any] = {}
if config_path.exists():
try:
loaded = yaml.safe_load(config_path.read_text()) or {}
if isinstance(loaded, dict):
config = loaded
except Exception:
config = {}
config = read_raw_config()
current_model = config.get("model")
if isinstance(current_model, dict):
@@ -2166,12 +2434,8 @@ def _reset_config_provider() -> Path:
if not config_path.exists():
return config_path
try:
config = yaml.safe_load(config_path.read_text()) or {}
except Exception:
return config_path
if not isinstance(config, dict):
config = read_raw_config()
if not config:
return config_path
model = config.get("model")
@@ -2187,14 +2451,21 @@ def _prompt_model_selection(
model_ids: List[str],
current_model: str = "",
pricing: Optional[Dict[str, Dict[str, str]]] = None,
unavailable_models: Optional[List[str]] = None,
portal_url: str = "",
) -> Optional[str]:
"""Interactive model selection. Puts current_model first with a marker. Returns chosen model ID or None.
If *pricing* is provided (``{model_id: {prompt, completion}}``), a compact
price indicator is shown next to each model in aligned columns.
If *unavailable_models* is provided, those models are shown grayed out
and unselectable, with an upgrade link to *portal_url*.
"""
from hermes_cli.models import _format_price_per_mtok
_unavailable = unavailable_models or []
# Reorder: current model first, then the rest (deduplicated)
ordered = []
if current_model and current_model in model_ids:
@@ -2203,9 +2474,12 @@ def _prompt_model_selection(
if mid not in ordered:
ordered.append(mid)
# All models for column-width computation (selectable + unavailable)
all_models = list(ordered) + list(_unavailable)
# Column-aligned labels when pricing is available
has_pricing = bool(pricing and any(pricing.get(m) for m in ordered))
name_col = max((len(m) for m in ordered), default=0) + 2 if has_pricing else 0
has_pricing = bool(pricing and any(pricing.get(m) for m in all_models))
name_col = max((len(m) for m in all_models), default=0) + 2 if has_pricing else 0
# Pre-compute formatted prices and dynamic column widths
_price_cache: dict[str, tuple[str, str, str]] = {}
@@ -2213,7 +2487,7 @@ def _prompt_model_selection(
cache_col = 0 # only set if any model has cache pricing
has_cache = False
if has_pricing:
for mid in ordered:
for mid in all_models:
p = pricing.get(mid) # type: ignore[union-attr]
if p:
inp = _format_price_per_mtok(p.get("prompt", ""))
@@ -2258,12 +2532,35 @@ def _prompt_model_selection(
header += f" {'Cache':>{cache_col}}"
menu_title += header + " /Mtok"
# ANSI escape for dim text
_DIM = "\033[2m"
_RESET = "\033[0m"
# Try arrow-key menu first, fall back to number input
try:
from simple_term_menu import TerminalMenu
choices = [f" {_label(mid)}" for mid in ordered]
choices.append(" Enter custom model name")
choices.append(" Skip (keep current)")
# 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()
effective_title = "Available free models:"
else:
effective_title = menu_title
menu = TerminalMenu(
choices,
cursor_index=default_idx,
@@ -2272,7 +2569,7 @@ def _prompt_model_selection(
menu_highlight_style=("fg_green",),
cycle_cursor=True,
clear_screen=False,
title=menu_title,
title=effective_title,
)
idx = menu.show()
if idx is None:
@@ -2295,6 +2592,13 @@ def _prompt_model_selection(
n = len(ordered)
print(f" {n + 1:>{num_width}}. Enter custom model name")
print(f" {n + 2:>{num_width}}. Skip (keep current)")
if _unavailable:
_upgrade_url = (portal_url or DEFAULT_NOUS_PORTAL_URL).rstrip("/")
print()
print(f" {_DIM}── Unavailable models (requires paid tier — upgrade at {_upgrade_url}) ──{_RESET}")
for mid in _unavailable:
print(f" {'':>{num_width}} {_DIM}{_label(mid)}{_RESET}")
print()
while True:
@@ -2337,8 +2641,8 @@ def _save_model_choice(model_id: str) -> None:
def login_command(args) -> None:
"""Deprecated: use 'hermes model' or 'hermes setup' instead."""
print("The 'hermes login' command has been removed.")
print("Use 'hermes model' to select a provider and model,")
print("or 'hermes setup' for full interactive setup.")
print("Use 'hermes auth' to manage credentials,")
print("'hermes model' to select a provider, or 'hermes setup' for full setup.")
raise SystemExit(0)
@@ -2348,17 +2652,25 @@ def _login_openai_codex(args, pconfig: ProviderConfig) -> None:
# Check for existing Hermes-owned credentials
try:
existing = resolve_codex_runtime_credentials()
print("Existing Codex credentials found in Hermes auth store.")
try:
reuse = input("Use existing credentials? [Y/n]: ").strip().lower()
except (EOFError, KeyboardInterrupt):
reuse = "y"
if reuse in ("", "y", "yes"):
config_path = _update_config_for_provider("openai-codex", existing.get("base_url", DEFAULT_CODEX_BASE_URL))
print()
print("Login successful!")
print(f" Config updated: {config_path} (model.provider=openai-codex)")
return
# Verify the resolved token is actually usable (not expired).
# resolve_codex_runtime_credentials attempts refresh, so if we get
# here the token should be valid — but double-check before telling
# the user "Login successful!".
_resolved_key = existing.get("api_key", "")
if isinstance(_resolved_key, str) and _resolved_key and not _codex_access_token_is_expiring(_resolved_key, 60):
print("Existing Codex credentials found in Hermes auth store.")
try:
reuse = input("Use existing credentials? [Y/n]: ").strip().lower()
except (EOFError, KeyboardInterrupt):
reuse = "y"
if reuse in ("", "y", "yes"):
config_path = _update_config_for_provider("openai-codex", existing.get("base_url", DEFAULT_CODEX_BASE_URL))
print()
print("Login successful!")
print(f" Config updated: {config_path} (model.provider=openai-codex)")
return
else:
print("Existing Codex credentials are expired. Starting fresh login...")
except AuthError:
pass
@@ -2699,7 +3011,6 @@ def _login_nous(args, pconfig: ProviderConfig) -> None:
)
inference_base_url = auth_state["inference_base_url"]
verify: bool | str = False if insecure else (ca_bundle if ca_bundle else True)
with _auth_store_lock():
auth_store = _load_auth_store()
@@ -2721,16 +3032,37 @@ def _login_nous(args, pconfig: ProviderConfig) -> None:
code="invalid_token",
)
from hermes_cli.models import _PROVIDER_MODELS
from hermes_cli.models import (
_PROVIDER_MODELS, get_pricing_for_provider, filter_nous_free_models,
check_nous_free_tier, partition_nous_models_by_tier,
)
model_ids = _PROVIDER_MODELS.get("nous", [])
print()
unavailable_models: list = []
if model_ids:
pricing = get_pricing_for_provider("nous")
model_ids = filter_nous_free_models(model_ids, pricing)
free_tier = check_nous_free_tier()
if free_tier:
model_ids, unavailable_models = partition_nous_models_by_tier(
model_ids, pricing, free_tier=True,
)
_portal = auth_state.get("portal_base_url", "")
if model_ids:
print(f"Showing {len(model_ids)} curated models — use \"Enter custom model name\" for others.")
selected_model = _prompt_model_selection(model_ids)
selected_model = _prompt_model_selection(
model_ids, pricing=pricing,
unavailable_models=unavailable_models,
portal_url=_portal,
)
if selected_model:
_save_model_choice(selected_model)
print(f"Default model set to: {selected_model}")
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.")
else:
print("No curated models available for Nous Portal.")
except Exception as exc:

51
hermes_cli/auth_commands.py
View File

@@ -18,7 +18,6 @@ from agent.credential_pool import (
STRATEGY_ROUND_ROBIN,
STRATEGY_RANDOM,
STRATEGY_LEAST_USED,
SUPPORTED_POOL_STRATEGIES,
PooledCredential,
_exhausted_until,
_normalize_custom_pool_name,
@@ -33,7 +32,7 @@ from hermes_constants import OPENROUTER_BASE_URL
# Providers that support OAuth login in addition to API keys.
_OAUTH_CAPABLE_PROVIDERS = {"anthropic", "nous", "openai-codex"}
_OAUTH_CAPABLE_PROVIDERS = {"anthropic", "nous", "openai-codex", "qwen-oauth"}
def _get_custom_provider_names() -> list:
@@ -148,7 +147,7 @@ def auth_add_command(args) -> None:
if provider.startswith(CUSTOM_POOL_PREFIX):
requested_type = AUTH_TYPE_API_KEY
else:
requested_type = AUTH_TYPE_OAUTH if provider in {"anthropic", "nous", "openai-codex"} else AUTH_TYPE_API_KEY
requested_type = AUTH_TYPE_OAUTH if provider in {"anthropic", "nous", "openai-codex", "qwen-oauth"} else AUTH_TYPE_API_KEY
pool = load_pool(provider)
@@ -251,6 +250,26 @@ def auth_add_command(args) -> None:
print(f'Added {provider} OAuth credential #{len(pool.entries())}: "{entry.label}"')
return
if provider == "qwen-oauth":
creds = auth_mod.resolve_qwen_runtime_credentials(refresh_if_expiring=False)
label = (getattr(args, "label", None) or "").strip() or label_from_token(
creds["api_key"],
_oauth_default_label(provider, len(pool.entries()) + 1),
)
entry = PooledCredential(
provider=provider,
id=uuid.uuid4().hex[:6],
label=label,
auth_type=AUTH_TYPE_OAUTH,
priority=0,
source=f"{SOURCE_MANUAL}:qwen_cli",
access_token=creds["api_key"],
base_url=creds.get("base_url"),
)
pool.add_entry(entry)
print(f'Added {provider} OAuth credential #{len(pool.entries())}: "{entry.label}"')
return
raise SystemExit(f"`hermes auth add {provider}` is not implemented for auth type {requested_type} yet.")
@@ -305,6 +324,32 @@ def auth_remove_command(args) -> None:
if cleared:
print(f"Cleared {env_var} from .env")
# If this was a singleton-seeded credential (OAuth device_code, hermes_pkce),
# clear the underlying auth store / credential file so it doesn't get
# re-seeded on the next load_pool() call.
elif removed.source == "device_code" and provider in ("openai-codex", "nous"):
from hermes_cli.auth import (
_load_auth_store, _save_auth_store, _auth_store_lock,
)
with _auth_store_lock():
auth_store = _load_auth_store()
providers_dict = auth_store.get("providers")
if isinstance(providers_dict, dict) and provider in providers_dict:
del providers_dict[provider]
_save_auth_store(auth_store)
print(f"Cleared {provider} OAuth tokens from auth store")
elif removed.source == "hermes_pkce" and provider == "anthropic":
from hermes_constants import get_hermes_home
oauth_file = get_hermes_home() / ".anthropic_oauth.json"
if oauth_file.exists():
oauth_file.unlink()
print("Cleared Hermes Anthropic OAuth credentials")
elif removed.source == "claude_code" and provider == "anthropic":
print("Note: Claude Code credentials live in ~/.claude/.credentials.json")
print(" Remove them manually if you want to deauthorize Claude Code.")
def auth_reset_command(args) -> None:
provider = _normalize_provider(getattr(args, "provider", ""))

75
hermes_cli/banner.py
View File

@@ -190,6 +190,79 @@ def check_for_updates() -> Optional[int]:
return behind
def _resolve_repo_dir() -> Optional[Path]:
"""Return the active Hermes git checkout, or None if this isn't a git install."""
hermes_home = get_hermes_home()
repo_dir = hermes_home / "hermes-agent"
if not (repo_dir / ".git").exists():
repo_dir = Path(__file__).parent.parent.resolve()
return repo_dir if (repo_dir / ".git").exists() else None
def _git_short_hash(repo_dir: Path, rev: str) -> Optional[str]:
"""Resolve a git revision to an 8-character short hash."""
try:
result = subprocess.run(
["git", "rev-parse", "--short=8", rev],
capture_output=True,
text=True,
timeout=5,
cwd=str(repo_dir),
)
except Exception:
return None
if result.returncode != 0:
return None
value = (result.stdout or "").strip()
return value or None
def get_git_banner_state(repo_dir: Optional[Path] = None) -> Optional[dict]:
"""Return upstream/local git hashes for the startup banner."""
repo_dir = repo_dir or _resolve_repo_dir()
if repo_dir is None:
return None
upstream = _git_short_hash(repo_dir, "origin/main")
local = _git_short_hash(repo_dir, "HEAD")
if not upstream or not local:
return None
ahead = 0
try:
result = subprocess.run(
["git", "rev-list", "--count", "origin/main..HEAD"],
capture_output=True,
text=True,
timeout=5,
cwd=str(repo_dir),
)
if result.returncode == 0:
ahead = int((result.stdout or "0").strip() or "0")
except Exception:
ahead = 0
return {"upstream": upstream, "local": local, "ahead": max(ahead, 0)}
def format_banner_version_label() -> str:
"""Return the version label shown in the startup banner title."""
base = f"Hermes Agent v{VERSION} ({RELEASE_DATE})"
state = get_git_banner_state()
if not state:
return base
upstream = state["upstream"]
local = state["local"]
ahead = int(state.get("ahead") or 0)
if ahead <= 0 or upstream == local:
return f"{base} · upstream {upstream}"
carried_word = "commit" if ahead == 1 else "commits"
return f"{base} · upstream {upstream} · local {local} (+{ahead} carried {carried_word})"
# =========================================================================
# Non-blocking update check
# =========================================================================
@@ -449,7 +522,7 @@ def build_welcome_banner(console: Console, model: str, cwd: str,
border_color = _skin_color("banner_border", "#CD7F32")
outer_panel = Panel(
layout_table,
title=f"[bold {title_color}]{agent_name} v{VERSION} ({RELEASE_DATE})[/]",
title=f"[bold {title_color}]{format_banner_version_label()}[/]",
border_style=border_color,
padding=(0, 2),
)

43
hermes_cli/callbacks.py
View File

@@ -25,7 +25,7 @@ def clarify_callback(cli, question, choices):
timeout = CLI_CONFIG.get("clarify", {}).get("timeout", 120)
response_queue = queue.Queue()
is_open_ended = not choices or len(choices) == 0
is_open_ended = not choices
cli._clarify_state = {
"question": question,
@@ -63,47 +63,6 @@ def clarify_callback(cli, question, choices):
)
def sudo_password_callback(cli) -> str:
"""Prompt for sudo password through the TUI.
Sets up a password input area and blocks until the user responds.
"""
timeout = 45
response_queue = queue.Queue()
cli._sudo_state = {"response_queue": response_queue}
cli._sudo_deadline = _time.monotonic() + timeout
if hasattr(cli, "_app") and cli._app:
cli._app.invalidate()
while True:
try:
result = response_queue.get(timeout=1)
cli._sudo_state = None
cli._sudo_deadline = 0
if hasattr(cli, "_app") and cli._app:
cli._app.invalidate()
if result:
cprint(f"\n{_DIM} ✓ Password received (cached for session){_RST}")
else:
cprint(f"\n{_DIM} ⏭ Skipped{_RST}")
return result
except queue.Empty:
remaining = cli._sudo_deadline - _time.monotonic()
if remaining <= 0:
break
if hasattr(cli, "_app") and cli._app:
cli._app.invalidate()
cli._sudo_state = None
cli._sudo_deadline = 0
if hasattr(cli, "_app") and cli._app:
cli._app.invalidate()
cprint(f"\n{_DIM} ⏱ Timeout — continuing without sudo{_RST}")
return ""
def prompt_for_secret(cli, var_name: str, prompt: str, metadata=None) -> dict:
"""Prompt for a secret value through the TUI (e.g. API keys for skills).

2
hermes_cli/claw.py
View File

@@ -10,7 +10,6 @@ Usage:
import importlib.util
import logging
import shutil
import sys
from datetime import datetime
from pathlib import Path
@@ -24,7 +23,6 @@ from hermes_cli.setup import (
print_info,
print_success,
print_error,
print_warning,
prompt_yes_no,
)

130
hermes_cli/clipboard.py
View File

@@ -1,4 +1,4 @@
"""Clipboard image extraction for macOS, Linux, and WSL2.
"""Clipboard image extraction for macOS, Windows, Linux, and WSL2.
Provides a single function `save_clipboard_image(dest)` that checks the
system clipboard for image data, saves it to *dest* as PNG, and returns
@@ -6,9 +6,10 @@ True on success. No external Python dependencies — uses only OS-level
CLI tools that ship with the platform (or are commonly installed).
Platform support:
macOS — osascript (always available), pngpaste (if installed)
WSL2 — powershell.exe via .NET System.Windows.Forms.Clipboard
Linux — wl-paste (Wayland), xclip (X11)
macOS — osascript (always available), pngpaste (if installed)
Windows — PowerShell via .NET System.Windows.Forms.Clipboard
WSL2 — powershell.exe via .NET System.Windows.Forms.Clipboard
Linux — wl-paste (Wayland), xclip (X11)
"""
import base64
@@ -32,6 +33,8 @@ def save_clipboard_image(dest: Path) -> bool:
dest.parent.mkdir(parents=True, exist_ok=True)
if sys.platform == "darwin":
return _macos_save(dest)
if sys.platform == "win32":
return _windows_save(dest)
return _linux_save(dest)
@@ -42,6 +45,8 @@ def has_clipboard_image() -> bool:
"""
if sys.platform == "darwin":
return _macos_has_image()
if sys.platform == "win32":
return _windows_has_image()
if _is_wsl():
return _wsl_has_image()
if os.environ.get("WAYLAND_DISPLAY"):
@@ -112,6 +117,104 @@ def _macos_osascript(dest: Path) -> bool:
return False
# ── Shared PowerShell scripts (native Windows + WSL2) ─────────────────────
# .NET System.Windows.Forms.Clipboard — used by both native Windows (powershell)
# and WSL2 (powershell.exe) paths.
_PS_CHECK_IMAGE = (
"Add-Type -AssemblyName System.Windows.Forms;"
"[System.Windows.Forms.Clipboard]::ContainsImage()"
)
_PS_EXTRACT_IMAGE = (
"Add-Type -AssemblyName System.Windows.Forms;"
"Add-Type -AssemblyName System.Drawing;"
"$img = [System.Windows.Forms.Clipboard]::GetImage();"
"if ($null -eq $img) { exit 1 }"
"$ms = New-Object System.IO.MemoryStream;"
"$img.Save($ms, [System.Drawing.Imaging.ImageFormat]::Png);"
"[System.Convert]::ToBase64String($ms.ToArray())"
)
# ── Native Windows ────────────────────────────────────────────────────────
# Native Windows uses ``powershell`` (Windows PowerShell 5.1, always present)
# or ``pwsh`` (PowerShell 7+, optional). Discovery is cached per-process.
def _find_powershell() -> str | None:
"""Return the first available PowerShell executable, or None."""
for name in ("powershell", "pwsh"):
try:
r = subprocess.run(
[name, "-NoProfile", "-NonInteractive", "-Command", "echo ok"],
capture_output=True, text=True, timeout=5,
)
if r.returncode == 0 and "ok" in r.stdout:
return name
except FileNotFoundError:
continue
except Exception:
continue
return None
# Cache the resolved PowerShell executable (checked once per process)
_ps_exe: str | None | bool = False # False = not yet checked
def _get_ps_exe() -> str | None:
global _ps_exe
if _ps_exe is False:
_ps_exe = _find_powershell()
return _ps_exe
def _windows_has_image() -> bool:
"""Check if the Windows clipboard contains an image."""
ps = _get_ps_exe()
if ps is None:
return False
try:
r = subprocess.run(
[ps, "-NoProfile", "-NonInteractive", "-Command", _PS_CHECK_IMAGE],
capture_output=True, text=True, timeout=5,
)
return r.returncode == 0 and "True" in r.stdout
except Exception as e:
logger.debug("Windows clipboard image check failed: %s", e)
return False
def _windows_save(dest: Path) -> bool:
"""Extract clipboard image on native Windows via PowerShell → base64 PNG."""
ps = _get_ps_exe()
if ps is None:
logger.debug("No PowerShell found — Windows clipboard image paste unavailable")
return False
try:
r = subprocess.run(
[ps, "-NoProfile", "-NonInteractive", "-Command", _PS_EXTRACT_IMAGE],
capture_output=True, text=True, timeout=15,
)
if r.returncode != 0:
return False
b64_data = r.stdout.strip()
if not b64_data:
return False
png_bytes = base64.b64decode(b64_data)
dest.write_bytes(png_bytes)
return dest.exists() and dest.stat().st_size > 0
except Exception as e:
logger.debug("Windows clipboard image extraction failed: %s", e)
dest.unlink(missing_ok=True)
return False
# ── Linux ────────────────────────────────────────────────────────────────
def _is_wsl() -> bool:
@@ -142,24 +245,7 @@ def _linux_save(dest: Path) -> bool:
# ── WSL2 (powershell.exe) ────────────────────────────────────────────────
# PowerShell script: get clipboard image as base64-encoded PNG on stdout.
# Using .NET System.Windows.Forms.Clipboard — always available on Windows.
_PS_CHECK_IMAGE = (
"Add-Type -AssemblyName System.Windows.Forms;"
"[System.Windows.Forms.Clipboard]::ContainsImage()"
)
_PS_EXTRACT_IMAGE = (
"Add-Type -AssemblyName System.Windows.Forms;"
"Add-Type -AssemblyName System.Drawing;"
"$img = [System.Windows.Forms.Clipboard]::GetImage();"
"if ($null -eq $img) { exit 1 }"
"$ms = New-Object System.IO.MemoryStream;"
"$img.Save($ms, [System.Drawing.Imaging.ImageFormat]::Png);"
"[System.Convert]::ToBase64String($ms.ToArray())"
)
# Reuses _PS_CHECK_IMAGE / _PS_EXTRACT_IMAGE defined above.
def _wsl_has_image() -> bool:
"""Check if Windows clipboard has an image (via powershell.exe)."""

12
hermes_cli/commands.py
View File

@@ -293,16 +293,8 @@ def _resolve_config_gates() -> set[str]:
if not gated:
return set()
try:
import yaml
config_path = os.path.join(
os.getenv("HERMES_HOME", os.path.expanduser("~/.hermes")),
"config.yaml",
)
if os.path.exists(config_path):
with open(config_path, encoding="utf-8") as f:
cfg = yaml.safe_load(f) or {}
else:
cfg = {}
from hermes_cli.config import read_raw_config
cfg = read_raw_config()
except Exception:
return set()
result: set[str] = set()

76
hermes_cli/config.py
View File

@@ -42,7 +42,7 @@ _EXTRA_ENV_KEYS = frozenset({
"TERMINAL_ENV", "TERMINAL_SSH_KEY", "TERMINAL_SSH_PORT",
"WHATSAPP_MODE", "WHATSAPP_ENABLED",
"MATTERMOST_HOME_CHANNEL", "MATTERMOST_REPLY_MODE",
"MATRIX_PASSWORD", "MATRIX_ENCRYPTION", "MATRIX_HOME_ROOM",
"MATRIX_PASSWORD", "MATRIX_ENCRYPTION", "MATRIX_DEVICE_ID", "MATRIX_HOME_ROOM",
"MATRIX_REQUIRE_MENTION", "MATRIX_FREE_RESPONSE_ROOMS", "MATRIX_AUTO_THREAD",
})
import yaml
@@ -157,7 +157,14 @@ def get_project_root() -> Path:
return Path(__file__).parent.parent.resolve()
def _secure_dir(path):
"""Set directory to owner-only access (0700). No-op on Windows."""
"""Set directory to owner-only access (0700). No-op on Windows.
Skipped in managed mode — the NixOS module sets group-readable
permissions (0750) so interactive users in the hermes group can
share state with the gateway service.
"""
if is_managed():
return
try:
os.chmod(path, 0o700)
except (OSError, NotImplementedError):
@@ -165,7 +172,13 @@ def _secure_dir(path):
def _secure_file(path):
"""Set file to owner-only read/write (0600). No-op on Windows."""
"""Set file to owner-only read/write (0600). No-op on Windows.
Skipped in managed mode — the NixOS activation script sets
group-readable permissions (0640) on config files.
"""
if is_managed():
return
try:
if os.path.exists(str(path)):
os.chmod(path, 0o600)
@@ -413,13 +426,17 @@ DEFAULT_CONFIG = {
"stt": {
"enabled": True,
"provider": "local", # "local" (free, faster-whisper) | "groq" | "openai" (Whisper API)
"provider": "local", # "local" (free, faster-whisper) | "groq" | "openai" (Whisper API) | "mistral" (Voxtral Transcribe)
"local": {
"model": "base", # tiny, base, small, medium, large-v3
"language": "", # auto-detect by default; set to "en", "es", "fr", etc. to force
},
"openai": {
"model": "whisper-1", # whisper-1, gpt-4o-mini-transcribe, gpt-4o-transcribe
},
"mistral": {
"model": "voxtral-mini-latest", # voxtral-mini-latest, voxtral-mini-2602
},
},
"voice": {
@@ -723,6 +740,14 @@ OPTIONAL_ENV_VARS = {
"category": "provider",
"advanced": True,
},
"HERMES_QWEN_BASE_URL": {
"description": "Qwen Portal base URL override (default: https://portal.qwen.ai/v1)",
"prompt": "Qwen Portal base URL (leave empty for default)",
"url": None,
"password": False,
"category": "provider",
"advanced": True,
},
"OPENCODE_ZEN_API_KEY": {
"description": "OpenCode Zen API key (pay-as-you-go access to curated models)",
"prompt": "OpenCode Zen API key",
@@ -974,6 +999,13 @@ OPTIONAL_ENV_VARS = {
"password": False,
"category": "messaging",
},
"DISCORD_REPLY_TO_MODE": {
"description": "Discord reply threading mode: 'off' (no reply references), 'first' (reply on first message only, default), 'all' (reply on every chunk)",
"prompt": "Discord reply mode (off/first/all)",
"url": None,
"password": False,
"category": "messaging",
},
"SLACK_BOT_TOKEN": {
"description": "Slack bot token (xoxb-). Get from OAuth & Permissions after installing your app. "
"Required scopes: chat:write, app_mentions:read, channels:history, groups:history, "
@@ -1079,6 +1111,14 @@ OPTIONAL_ENV_VARS = {
"category": "messaging",
"advanced": True,
},
"MATRIX_DEVICE_ID": {
"description": "Stable Matrix device ID for E2EE persistence across restarts (e.g. HERMES_BOT)",
"prompt": "Matrix device ID (stable across restarts)",
"url": None,
"password": False,
"category": "messaging",
"advanced": True,
},
"GATEWAY_ALLOW_ALL_USERS": {
"description": "Allow all users to interact with messaging bots (true/false). Default: false.",
"prompt": "Allow all users (true/false)",
@@ -1873,6 +1913,24 @@ def _normalize_max_turns_config(config: Dict[str, Any]) -> Dict[str, Any]:
def read_raw_config() -> Dict[str, Any]:
"""Read ~/.hermes/config.yaml as-is, without merging defaults or migrating.
Returns the raw YAML dict, or ``{}`` if the file doesn't exist or can't
be parsed. Use this for lightweight config reads where you just need a
single value and don't want the overhead of ``load_config()``'s deep-merge
+ migration pipeline.
"""
try:
config_path = get_config_path()
if config_path.exists():
with open(config_path, encoding="utf-8") as f:
return yaml.safe_load(f) or {}
except Exception:
pass
return {}
def load_config() -> Dict[str, Any]:
"""Load configuration from ~/.hermes/config.yaml."""
import copy
@@ -1924,8 +1982,8 @@ _FALLBACK_COMMENT = """
#
# Supported providers:
# openrouter (OPENROUTER_API_KEY) — routes to any model
# openai-codex (OAuth — hermes login) — OpenAI Codex
# nous (OAuth — hermes login) — Nous Portal
# openai-codex (OAuth — hermes auth) — OpenAI Codex
# nous (OAuth — hermes auth) — Nous Portal
# zai (ZAI_API_KEY) — Z.AI / GLM
# kimi-coding (KIMI_API_KEY) — Kimi / Moonshot
# minimax (MINIMAX_API_KEY) — MiniMax
@@ -1967,8 +2025,8 @@ _COMMENTED_SECTIONS = """
#
# Supported providers:
# openrouter (OPENROUTER_API_KEY) — routes to any model
# openai-codex (OAuth — hermes login) — OpenAI Codex
# nous (OAuth — hermes login) — Nous Portal
# openai-codex (OAuth — hermes auth) — OpenAI Codex
# nous (OAuth — hermes auth) — Nous Portal
# zai (ZAI_API_KEY) — Z.AI / GLM
# kimi-coding (KIMI_API_KEY) — Kimi / Moonshot
# minimax (MINIMAX_API_KEY) — MiniMax
@@ -2512,7 +2570,7 @@ def set_config_value(key: str, value: str):
'TINKER_API_KEY',
]
if key.upper() in api_keys or key.upper().endswith('_API_KEY') or key.upper().endswith('_TOKEN') or key.upper().startswith('TERMINAL_SSH'):
if key.upper() in api_keys or key.upper().endswith(('_API_KEY', '_TOKEN')) or key.upper().startswith('TERMINAL_SSH'):
save_env_value(key.upper(), value)
print(f"✓ Set {key} in {get_env_path()}")
return

15
hermes_cli/cron.py
View File

@@ -93,6 +93,21 @@ def cron_list(show_all: bool = False):
script = job.get("script")
if script:
print(f" Script: {script}")
# Execution history
last_status = job.get("last_status")
if last_status:
last_run = job.get("last_run_at", "?")
if last_status == "ok":
status_display = color("ok", Colors.GREEN)
else:
status_display = color(f"{last_status}: {job.get('last_error', '?')}", Colors.RED)
print(f" Last run: {last_run} {status_display}")
delivery_err = job.get("last_delivery_error")
if delivery_err:
print(f" {color('⚠ Delivery failed:', Colors.YELLOW)} {delivery_err}")
print()
from hermes_cli.gateway import find_gateway_pids

130
hermes_cli/doctor.py
View File

@@ -812,69 +812,83 @@ def run_doctor(args):
check_warn("No GITHUB_TOKEN", f"(60 req/hr rate limit — set in {_DHH}/.env for better rates)")
# =========================================================================
# Honcho memory
# Memory Provider (only check the active provider, if any)
# =========================================================================
print()
print(color("◆ Honcho Memory", Colors.CYAN, Colors.BOLD))
print(color("◆ Memory Provider", Colors.CYAN, Colors.BOLD))
_active_memory_provider = ""
try:
from plugins.memory.honcho.client import HonchoClientConfig, resolve_config_path
hcfg = HonchoClientConfig.from_global_config()
_honcho_cfg_path = resolve_config_path()
import yaml as _yaml
_mem_cfg_path = HERMES_HOME / "config.yaml"
if _mem_cfg_path.exists():
with open(_mem_cfg_path) as _f:
_raw_cfg = _yaml.safe_load(_f) or {}
_active_memory_provider = (_raw_cfg.get("memory") or {}).get("provider", "")
except Exception:
pass
if not _honcho_cfg_path.exists():
check_warn("Honcho config not found", "run: hermes memory setup")
elif not hcfg.enabled:
check_info(f"Honcho disabled (set enabled: true in {_honcho_cfg_path} to activate)")
elif not (hcfg.api_key or hcfg.base_url):
check_fail("Honcho API key or base URL not set", "run: hermes memory setup")
issues.append("No Honcho API key — run 'hermes memory setup'")
else:
from plugins.memory.honcho.client import get_honcho_client, reset_honcho_client
reset_honcho_client()
try:
get_honcho_client(hcfg)
check_ok(
"Honcho connected",
f"workspace={hcfg.workspace_id} mode={hcfg.memory_mode} freq={hcfg.write_frequency}",
)
except Exception as _e:
check_fail("Honcho connection failed", str(_e))
issues.append(f"Honcho unreachable: {_e}")
except ImportError:
check_warn("honcho-ai not installed", "pip install honcho-ai")
except Exception as _e:
check_warn("Honcho check failed", str(_e))
if not _active_memory_provider:
check_ok("Built-in memory active", "(no external provider configured — this is fine)")
elif _active_memory_provider == "honcho":
try:
from plugins.memory.honcho.client import HonchoClientConfig, resolve_config_path
hcfg = HonchoClientConfig.from_global_config()
_honcho_cfg_path = resolve_config_path()
# =========================================================================
# Mem0 memory
# =========================================================================
print()
print(color("◆ Mem0 Memory", Colors.CYAN, Colors.BOLD))
try:
from plugins.memory.mem0 import _load_config as _load_mem0_config
mem0_cfg = _load_mem0_config()
mem0_key = mem0_cfg.get("api_key", "")
if mem0_key:
check_ok("Mem0 API key configured")
check_info(f"user_id={mem0_cfg.get('user_id', '?')} agent_id={mem0_cfg.get('agent_id', '?')}")
# Check if mem0.json exists but is missing api_key (the bug we fixed)
mem0_json = HERMES_HOME / "mem0.json"
if mem0_json.exists():
if not _honcho_cfg_path.exists():
check_warn("Honcho config not found", "run: hermes memory setup")
elif not hcfg.enabled:
check_info(f"Honcho disabled (set enabled: true in {_honcho_cfg_path} to activate)")
elif not (hcfg.api_key or hcfg.base_url):
check_fail("Honcho API key or base URL not set", "run: hermes memory setup")
issues.append("No Honcho API key — run 'hermes memory setup'")
else:
from plugins.memory.honcho.client import get_honcho_client, reset_honcho_client
reset_honcho_client()
try:
import json as _json
file_cfg = _json.loads(mem0_json.read_text())
if not file_cfg.get("api_key") and mem0_key:
check_info("api_key from .env (not in mem0.json) — this is fine")
except Exception:
pass
else:
check_warn("Mem0 not configured", "(set MEM0_API_KEY in .env or run hermes memory setup)")
except ImportError:
check_warn("Mem0 plugin not loadable", "(optional)")
except Exception as _e:
check_warn("Mem0 check failed", str(_e))
get_honcho_client(hcfg)
check_ok(
"Honcho connected",
f"workspace={hcfg.workspace_id} mode={hcfg.recall_mode} freq={hcfg.write_frequency}",
)
except Exception as _e:
check_fail("Honcho connection failed", str(_e))
issues.append(f"Honcho unreachable: {_e}")
except ImportError:
check_fail("honcho-ai not installed", "pip install honcho-ai")
issues.append("Honcho is set as memory provider but honcho-ai is not installed")
except Exception as _e:
check_warn("Honcho check failed", str(_e))
elif _active_memory_provider == "mem0":
try:
from plugins.memory.mem0 import _load_config as _load_mem0_config
mem0_cfg = _load_mem0_config()
mem0_key = mem0_cfg.get("api_key", "")
if mem0_key:
check_ok("Mem0 API key configured")
check_info(f"user_id={mem0_cfg.get('user_id', '?')} agent_id={mem0_cfg.get('agent_id', '?')}")
else:
check_fail("Mem0 API key not set", "(set MEM0_API_KEY in .env or run hermes memory setup)")
issues.append("Mem0 is set as memory provider but API key is missing")
except ImportError:
check_fail("Mem0 plugin not loadable", "pip install mem0ai")
issues.append("Mem0 is set as memory provider but mem0ai is not installed")
except Exception as _e:
check_warn("Mem0 check failed", str(_e))
else:
# Generic check for other memory providers (openviking, hindsight, etc.)
try:
from plugins.memory import load_memory_provider
_provider = load_memory_provider(_active_memory_provider)
if _provider and _provider.is_available():
check_ok(f"{_active_memory_provider} provider active")
elif _provider:
check_warn(f"{_active_memory_provider} configured but not available", "run: hermes memory status")
else:
check_warn(f"{_active_memory_provider} plugin not found", "run: hermes memory setup")
except Exception as _e:
check_warn(f"{_active_memory_provider} check failed", str(_e))
# =========================================================================
# Profiles
@@ -920,8 +934,8 @@ def run_doctor(args):
pass
except ImportError:
pass
except Exception as _e:
logger.debug("Profile health check failed: %s", _e)
except Exception:
pass
# =========================================================================
# Summary

63
hermes_cli/gateway.py
View File

@@ -267,6 +267,34 @@ def _profile_suffix() -> str:
return hashlib.sha256(str(home).encode()).hexdigest()[:8]
def _profile_arg(hermes_home: str | None = None) -> str:
"""Return ``--profile <name>`` only when HERMES_HOME is a named profile.
For ``~/.hermes/profiles/<name>``, returns ``"--profile <name>"``.
For the default profile or hash-based custom paths, returns the empty string.
Args:
hermes_home: Optional explicit HERMES_HOME path. Defaults to the current
``get_hermes_home()`` value. Should be passed when generating a
service definition for a different user (e.g. system service).
"""
import re
from pathlib import Path as _Path
home = Path(hermes_home or str(get_hermes_home())).resolve()
default = (_Path.home() / ".hermes").resolve()
if home == default:
return ""
profiles_root = (default / "profiles").resolve()
try:
rel = home.relative_to(profiles_root)
parts = rel.parts
if len(parts) == 1 and re.match(r"^[a-z0-9][a-z0-9_-]{0,63}$", parts[0]):
return f"--profile {parts[0]}"
except ValueError:
pass
return ""
def get_service_name() -> str:
"""Derive a systemd service name scoped to this HERMES_HOME.
@@ -626,6 +654,7 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None)
if system:
username, group_name, home_dir = _system_service_identity(run_as_user)
hermes_home = _hermes_home_for_target_user(home_dir)
profile_arg = _profile_arg(hermes_home)
path_entries.extend(_build_user_local_paths(Path(home_dir), path_entries))
path_entries.extend(common_bin_paths)
sane_path = ":".join(path_entries)
@@ -640,7 +669,7 @@ StartLimitBurst=5
Type=simple
User={username}
Group={group_name}
ExecStart={python_path} -m hermes_cli.main gateway run --replace
ExecStart={python_path} -m hermes_cli.main{f" {profile_arg}" if profile_arg else ""} gateway run --replace
WorkingDirectory={working_dir}
Environment="HOME={home_dir}"
Environment="USER={username}"
@@ -661,6 +690,7 @@ WantedBy=multi-user.target
"""
hermes_home = str(get_hermes_home().resolve())
profile_arg = _profile_arg(hermes_home)
path_entries.extend(_build_user_local_paths(Path.home(), path_entries))
path_entries.extend(common_bin_paths)
sane_path = ":".join(path_entries)
@@ -672,7 +702,7 @@ StartLimitBurst=5
[Service]
Type=simple
ExecStart={python_path} -m hermes_cli.main gateway run --replace
ExecStart={python_path} -m hermes_cli.main{f" {profile_arg}" if profile_arg else ""} gateway run --replace
WorkingDirectory={working_dir}
Environment="PATH={sane_path}"
Environment="VIRTUAL_ENV={venv_dir}"
@@ -965,6 +995,7 @@ def generate_launchd_plist() -> str:
log_dir = get_hermes_home() / "logs"
log_dir.mkdir(parents=True, exist_ok=True)
label = get_launchd_label()
profile_arg = _profile_arg(hermes_home)
# Build a sane PATH for the launchd plist. launchd provides only a
# minimal default (/usr/bin:/bin:/usr/sbin:/sbin) which misses Homebrew,
# nvm, cargo, etc. We prepend venv/bin and node_modules/.bin (matching
@@ -986,21 +1017,32 @@ def generate_launchd_plist() -> str:
dict.fromkeys(priority_dirs + [p for p in os.environ.get("PATH", "").split(":") if p])
)
# Build ProgramArguments array, including --profile when using a named profile
prog_args = [
f"<string>{python_path}</string>",
"<string>-m</string>",
"<string>hermes_cli.main</string>",
]
if profile_arg:
for part in profile_arg.split():
prog_args.append(f"<string>{part}</string>")
prog_args.extend([
"<string>gateway</string>",
"<string>run</string>",
"<string>--replace</string>",
])
prog_args_xml = "\n ".join(prog_args)
return f"""<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
<key>Label</key>
<string>{label}</string>
<key>ProgramArguments</key>
<array>
<string>{python_path}</string>
<string>-m</string>
<string>hermes_cli.main</string>
<string>gateway</string>
<string>run</string>
<string>--replace</string>
{prog_args_xml}
</array>
<key>WorkingDirectory</key>
@@ -1803,8 +1845,7 @@ def _setup_signal():
print_warning("signal-cli not found on PATH.")
print_info(" Signal requires signal-cli running as an HTTP daemon.")
print_info(" Install options:")
print_info(" Linux: sudo apt install signal-cli")
print_info(" or download from https://github.com/AsamK/signal-cli")
print_info(" Linux: download from https://github.com/AsamK/signal-cli/releases")
print_info(" macOS: brew install signal-cli")
print_info(" Docker: bbernhard/signal-cli-rest-api")
print()

1
hermes_cli/logs.py
View File

@@ -15,7 +15,6 @@ Usage examples::
hermes logs --since 30m -f # follow, starting 30 min ago
"""
import os
import re
import sys
import time

157
hermes_cli/main.py
View File

@@ -957,6 +957,7 @@ def select_provider_and_model(args=None):
"openrouter": "OpenRouter",
"nous": "Nous Portal",
"openai-codex": "OpenAI Codex",
"qwen-oauth": "Qwen OAuth",
"copilot-acp": "GitHub Copilot ACP",
"copilot": "GitHub Copilot",
"anthropic": "Anthropic",
@@ -986,6 +987,7 @@ def select_provider_and_model(args=None):
("openrouter", "OpenRouter (100+ models, pay-per-use)"),
("anthropic", "Anthropic (Claude models — API key or Claude Code)"),
("openai-codex", "OpenAI Codex"),
("qwen-oauth", "Qwen OAuth (reuses local Qwen CLI login)"),
("copilot", "GitHub Copilot (uses GITHUB_TOKEN or gh auth token)"),
("huggingface", "Hugging Face Inference Providers (20+ open models)"),
]
@@ -1082,6 +1084,8 @@ def select_provider_and_model(args=None):
_model_flow_nous(config, current_model, args=args)
elif selected_provider == "openai-codex":
_model_flow_openai_codex(config, current_model)
elif selected_provider == "qwen-oauth":
_model_flow_qwen_oauth(config, current_model)
elif selected_provider == "copilot-acp":
_model_flow_copilot_acp(config, current_model)
elif selected_provider == "copilot":
@@ -1193,7 +1197,7 @@ def _model_flow_nous(config, current_model="", args=None):
from hermes_cli.auth import (
get_provider_auth_state, _prompt_model_selection, _save_model_choice,
_update_config_for_provider, resolve_nous_runtime_credentials,
fetch_nous_models, AuthError, format_auth_error,
AuthError, format_auth_error,
_login_nous, PROVIDER_REGISTRY,
)
from hermes_cli.config import get_env_value, save_config, save_env_value
@@ -1234,14 +1238,15 @@ def _model_flow_nous(config, current_model="", args=None):
# Already logged in — use curated model list (same as OpenRouter defaults).
# The live /models endpoint returns hundreds of models; the curated list
# shows only agentic models users recognize from OpenRouter.
from hermes_cli.models import _PROVIDER_MODELS, get_pricing_for_provider
from hermes_cli.models import (
_PROVIDER_MODELS, get_pricing_for_provider, filter_nous_free_models,
check_nous_free_tier, partition_nous_models_by_tier,
)
model_ids = _PROVIDER_MODELS.get("nous", [])
if not model_ids:
print("No curated models available for Nous Portal.")
return
print(f"Showing {len(model_ids)} curated models — use \"Enter custom model name\" for others.")
# Verify credentials are still valid (catches expired sessions early)
try:
creds = resolve_nous_runtime_credentials(min_key_ttl_seconds=5 * 60)
@@ -1267,7 +1272,44 @@ 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")
selected = _prompt_model_selection(model_ids, current_model=current_model, pricing=pricing)
# Check if user is on free tier
free_tier = check_nous_free_tier()
# For both tiers: apply the allowlist filter first (removes non-allowlisted
# free models and allowlist models that aren't actually free).
# Then for free users: partition remaining models into selectable/unavailable.
model_ids = filter_nous_free_models(model_ids, pricing)
unavailable_models: list[str] = []
if free_tier:
model_ids, unavailable_models = partition_nous_models_by_tier(model_ids, pricing, free_tier=True)
if not model_ids and not unavailable_models:
print("No models available for Nous Portal after filtering.")
return
# Resolve portal URL for upgrade links (may differ on staging)
_nous_portal_url = ""
try:
_nous_state = get_provider_auth_state("nous")
if _nous_state:
_nous_portal_url = _nous_state.get("portal_base_url", "")
except Exception:
pass
if free_tier and not model_ids:
print("No free models currently available.")
if unavailable_models:
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.")
return
print(f"Showing {len(model_ids)} curated models — use \"Enter custom model name\" for others.")
selected = _prompt_model_selection(
model_ids, current_model=current_model, pricing=pricing,
unavailable_models=unavailable_models, portal_url=_nous_portal_url,
)
if selected:
_save_model_choice(selected)
# Reactivate Nous as the provider and update config
@@ -1315,7 +1357,6 @@ def _model_flow_openai_codex(config, current_model=""):
PROVIDER_REGISTRY, DEFAULT_CODEX_BASE_URL,
)
from hermes_cli.codex_models import get_codex_model_ids
from hermes_cli.config import get_env_value, save_env_value
import argparse
status = get_codex_auth_status()
@@ -1333,12 +1374,21 @@ def _model_flow_openai_codex(config, current_model=""):
return
_codex_token = None
# Prefer credential pool (where `hermes auth` stores device_code tokens),
# fall back to legacy provider state.
try:
from hermes_cli.auth import resolve_codex_runtime_credentials
_codex_creds = resolve_codex_runtime_credentials()
_codex_token = _codex_creds.get("api_key")
_codex_status = get_codex_auth_status()
if _codex_status.get("logged_in"):
_codex_token = _codex_status.get("api_key")
except Exception:
pass
if not _codex_token:
try:
from hermes_cli.auth import resolve_codex_runtime_credentials
_codex_creds = resolve_codex_runtime_credentials()
_codex_token = _codex_creds.get("api_key")
except Exception:
pass
codex_models = get_codex_model_ids(access_token=_codex_token)
@@ -1352,6 +1402,56 @@ def _model_flow_openai_codex(config, current_model=""):
_DEFAULT_QWEN_PORTAL_MODELS = [
"qwen3-coder-plus",
"qwen3-coder",
]
def _model_flow_qwen_oauth(_config, current_model=""):
"""Qwen OAuth provider: reuse local Qwen CLI login, then pick model."""
from hermes_cli.auth import (
get_qwen_auth_status,
resolve_qwen_runtime_credentials,
_prompt_model_selection,
_save_model_choice,
_update_config_for_provider,
DEFAULT_QWEN_BASE_URL,
)
from hermes_cli.models import fetch_api_models
status = get_qwen_auth_status()
if not status.get("logged_in"):
print("Not logged into Qwen CLI OAuth.")
print("Run: qwen auth qwen-oauth")
auth_file = status.get("auth_file")
if auth_file:
print(f"Expected credentials file: {auth_file}")
if status.get("error"):
print(f"Error: {status.get('error')}")
return
# Try live model discovery, fall back to curated list.
models = None
try:
creds = resolve_qwen_runtime_credentials(refresh_if_expiring=True)
models = fetch_api_models(creds["api_key"], creds["base_url"])
except Exception:
pass
if not models:
models = list(_DEFAULT_QWEN_PORTAL_MODELS)
default = current_model or (models[0] if models else "qwen3-coder-plus")
selected = _prompt_model_selection(models, current_model=default)
if selected:
_save_model_choice(selected)
_update_config_for_provider("qwen-oauth", DEFAULT_QWEN_BASE_URL)
print(f"Default model set to: {selected} (via Qwen OAuth)")
else:
print("No change.")
def _model_flow_custom(config):
"""Custom endpoint: collect URL, API key, and model name.
@@ -1359,7 +1459,7 @@ def _model_flow_custom(config):
so it appears in the provider menu on subsequent runs.
"""
from hermes_cli.auth import _save_model_choice, deactivate_provider
from hermes_cli.config import get_env_value, save_env_value, load_config, save_config
from hermes_cli.config import get_env_value, load_config, save_config
current_url = get_env_value("OPENAI_BASE_URL") or ""
current_key = get_env_value("OPENAI_API_KEY") or ""
@@ -1621,7 +1721,7 @@ def _model_flow_named_custom(config, provider_info):
Otherwise probes the endpoint's /models API to let the user pick one.
"""
from hermes_cli.auth import _save_model_choice, deactivate_provider
from hermes_cli.config import save_env_value, load_config, save_config
from hermes_cli.config import load_config, save_config
from hermes_cli.models import fetch_api_models
name = provider_info["name"]
@@ -1831,7 +1931,7 @@ def _model_flow_copilot(config, current_model=""):
deactivate_provider,
resolve_api_key_provider_credentials,
)
from hermes_cli.config import get_env_value, save_env_value, load_config, save_config
from hermes_cli.config import save_env_value, load_config, save_config
from hermes_cli.models import (
fetch_api_models,
fetch_github_model_catalog,
@@ -2422,8 +2522,6 @@ def _model_flow_anthropic(config, current_model=""):
)
from hermes_cli.models import _PROVIDER_MODELS
pconfig = PROVIDER_REGISTRY["anthropic"]
# Check ALL credential sources
existing_key = (
get_env_value("ANTHROPIC_TOKEN")
@@ -3596,7 +3694,7 @@ def cmd_update(args):
try:
from hermes_cli.profiles import list_profiles, get_active_profile_name, seed_profile_skills
active = get_active_profile_name()
other_profiles = [p for p in list_profiles() if not p.is_default and p.name != active]
other_profiles = [p for p in list_profiles() if p.name != active]
if other_profiles:
print()
print("→ Syncing bundled skills to other profiles...")
@@ -3692,7 +3790,7 @@ def cmd_update(args):
try:
from hermes_cli.gateway import (
is_macos, is_linux, _ensure_user_systemd_env,
get_systemd_linger_status, find_gateway_pids,
find_gateway_pids,
_get_service_pids,
)
import signal as _signal
@@ -3848,7 +3946,7 @@ def cmd_profile(args):
"""Profile management — create, delete, list, switch, alias."""
from hermes_cli.profiles import (
list_profiles, create_profile, delete_profile, seed_profile_skills,
get_active_profile, set_active_profile, get_active_profile_name,
set_active_profile, get_active_profile_name,
check_alias_collision, create_wrapper_script, remove_wrapper_script,
_is_wrapper_dir_in_path, _get_wrapper_dir,
)
@@ -3976,7 +4074,6 @@ def cmd_profile(args):
print(f" {name} chat Start chatting")
print(f" {name} gateway start Start the messaging gateway")
if clone or clone_all:
from hermes_constants import get_hermes_home
profile_dir_display = f"~/.hermes/profiles/{name}"
print(f"\n Edit {profile_dir_display}/.env for different API keys")
print(f" Edit {profile_dir_display}/SOUL.md for different personality")
@@ -4411,7 +4508,7 @@ For more help on a command:
gateway_uninstall.add_argument("--system", action="store_true", help="Target the Linux system-level gateway service")
# gateway setup
gateway_setup = gateway_subparsers.add_parser("setup", help="Configure messaging platforms")
gateway_subparsers.add_parser("setup", help="Configure messaging platforms")
gateway_parser.set_defaults(func=cmd_gateway)
@@ -4686,10 +4783,10 @@ For more help on a command:
config_subparsers = config_parser.add_subparsers(dest="config_command")
# config show (default)
config_show = config_subparsers.add_parser("show", help="Show current configuration")
config_subparsers.add_parser("show", help="Show current configuration")
# config edit
config_edit = config_subparsers.add_parser("edit", help="Open config file in editor")
config_subparsers.add_parser("edit", help="Open config file in editor")
# config set
config_set = config_subparsers.add_parser("set", help="Set a configuration value")
@@ -4697,16 +4794,16 @@ For more help on a command:
config_set.add_argument("value", nargs="?", help="Value to set")
# config path
config_path = config_subparsers.add_parser("path", help="Print config file path")
config_subparsers.add_parser("path", help="Print config file path")
# config env-path
config_env = config_subparsers.add_parser("env-path", help="Print .env file path")
config_subparsers.add_parser("env-path", help="Print .env file path")
# config check
config_check = config_subparsers.add_parser("check", help="Check for missing/outdated config")
config_subparsers.add_parser("check", help="Check for missing/outdated config")
# config migrate
config_migrate = config_subparsers.add_parser("migrate", help="Update config with new options")
config_subparsers.add_parser("migrate", help="Update config with new options")
config_parser.set_defaults(func=cmd_config)
@@ -4720,7 +4817,7 @@ For more help on a command:
)
pairing_sub = pairing_parser.add_subparsers(dest="pairing_action")
pairing_list_parser = pairing_sub.add_parser("list", help="Show pending + approved users")
pairing_sub.add_parser("list", help="Show pending + approved users")
pairing_approve_parser = pairing_sub.add_parser("approve", help="Approve a pairing code")
pairing_approve_parser.add_argument("platform", help="Platform name (telegram, discord, slack, whatsapp)")
@@ -4730,7 +4827,7 @@ For more help on a command:
pairing_revoke_parser.add_argument("platform", help="Platform name")
pairing_revoke_parser.add_argument("user_id", help="User ID to revoke")
pairing_clear_parser = pairing_sub.add_parser("clear-pending", help="Clear all pending codes")
pairing_sub.add_parser("clear-pending", help="Clear all pending codes")
def cmd_pairing(args):
from hermes_cli.pairing import pairing_command
@@ -4906,7 +5003,7 @@ For more help on a command:
memory_sub = memory_parser.add_subparsers(dest="memory_command")
memory_sub.add_parser("setup", help="Interactive provider selection and configuration")
memory_sub.add_parser("status", help="Show current memory provider config")
memory_off_p = memory_sub.add_parser("off", help="Disable external provider (built-in only)")
memory_sub.add_parser("off", help="Disable external provider (built-in only)")
def cmd_memory(args):
sub = getattr(args, "memory_command", None)
@@ -5070,7 +5167,7 @@ For more help on a command:
sessions_prune.add_argument("--source", help="Only prune sessions from this source")
sessions_prune.add_argument("--yes", "-y", action="store_true", help="Skip confirmation")
sessions_stats = sessions_subparsers.add_parser("stats", help="Show session store statistics")
sessions_subparsers.add_parser("stats", help="Show session store statistics")
sessions_rename = sessions_subparsers.add_parser("rename", help="Set or change a session's title")
sessions_rename.add_argument("session_id", help="Session ID to rename")
@@ -5430,7 +5527,7 @@ For more help on a command:
)
profile_subparsers = profile_parser.add_subparsers(dest="profile_action")
profile_list = profile_subparsers.add_parser("list", help="List all profiles")
profile_subparsers.add_parser("list", help="List all profiles")
profile_use = profile_subparsers.add_parser("use", help="Set sticky default profile")
profile_use.add_argument("profile_name", help="Profile name (or 'default')")

10
hermes_cli/memory_setup.py
View File

@@ -12,6 +12,8 @@ import os
import sys
from pathlib import Path
from hermes_constants import get_hermes_home
# ---------------------------------------------------------------------------
# Curses-based interactive picker (same pattern as hermes tools)
@@ -275,7 +277,7 @@ def cmd_setup_provider(provider_name: str) -> None:
config["memory"] = {}
if hasattr(provider, "post_setup"):
hermes_home = str(Path(os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes"))))
hermes_home = str(get_hermes_home())
provider.post_setup(hermes_home, config)
return
@@ -326,7 +328,7 @@ def cmd_setup(args) -> None:
# If the provider has a post_setup hook, delegate entirely to it.
# The hook handles its own config, connection test, and activation.
if hasattr(provider, "post_setup"):
hermes_home = str(Path(os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes"))))
hermes_home = str(get_hermes_home())
provider.post_setup(hermes_home, config)
return
@@ -336,7 +338,7 @@ def cmd_setup(args) -> None:
if not isinstance(provider_config, dict):
provider_config = {}
env_path = Path(os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes"))) / ".env"
env_path = get_hermes_home() / ".env"
env_writes = {}
if schema:
@@ -400,7 +402,7 @@ def cmd_setup(args) -> None:
save_config(config)
# Write non-secret config to provider's native location
hermes_home = str(Path(os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes"))))
hermes_home = str(get_hermes_home())
if provider_config and hasattr(provider, "save_config"):
try:
provider.save_config(provider_config, hermes_home)

1
hermes_cli/model_normalize.py
View File

@@ -84,6 +84,7 @@ _PASSTHROUGH_PROVIDERS: frozenset[str] = frozenset({
"minimax",
"minimax-cn",
"alibaba",
"qwen-oauth",
"huggingface",
"openai-codex",
"custom",

18
hermes_cli/model_switch.py
View File

@@ -21,22 +21,16 @@ OpenRouter variant suffixes (``:free``, ``:extended``, ``:fast``).
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from dataclasses import dataclass
from typing import List, NamedTuple, Optional
from hermes_cli.providers import (
ALIASES,
LABELS,
TRANSPORT_TO_API_MODE,
determine_api_mode,
get_label,
get_provider,
is_aggregator,
normalize_provider,
resolve_provider_full,
)
from hermes_cli.model_normalize import (
detect_vendor,
normalize_model_for_provider,
)
from agent.models_dev import (
@@ -797,12 +791,12 @@ def list_authenticated_providers(
if overlay.auth_type in ("oauth_device_code", "oauth_external", "external_process"):
# These use auth stores, not env vars — check for auth.json entries
try:
from hermes_cli.auth import _read_auth_store
store = _read_auth_store()
if store and pid in store:
from hermes_cli.auth import _load_auth_store
store = _load_auth_store()
if store and (pid in store.get("providers", {}) or pid in store.get("credential_pool", {})):
has_creds = True
except Exception:
pass
except Exception as exc:
logger.debug("Auth store check failed for %s: %s", pid, exc)
if not has_creds:
continue

227
hermes_cli/models.py
View File

@@ -44,7 +44,7 @@ OPENROUTER_MODELS: list[tuple[str, str]] = [
("stepfun/step-3.5-flash", ""),
("minimax/minimax-m2.7", ""),
("minimax/minimax-m2.5", ""),
("z-ai/glm-5", ""),
("z-ai/glm-5.1", ""),
("z-ai/glm-5-turbo", ""),
("moonshotai/kimi-k2.5", ""),
("x-ai/grok-4.20-beta", ""),
@@ -75,7 +75,7 @@ _PROVIDER_MODELS: dict[str, list[str]] = {
"stepfun/step-3.5-flash",
"minimax/minimax-m2.7",
"minimax/minimax-m2.5",
"z-ai/glm-5",
"z-ai/glm-5.1",
"z-ai/glm-5-turbo",
"moonshotai/kimi-k2.5",
"x-ai/grok-4.20-beta",
@@ -144,18 +144,22 @@ _PROVIDER_MODELS: dict[str, list[str]] = {
"kimi-k2-0905-preview",
],
"minimax": [
"MiniMax-M2.7",
"MiniMax-M2.7-highspeed",
"MiniMax-M1",
"MiniMax-M1-40k",
"MiniMax-M1-80k",
"MiniMax-M1-128k",
"MiniMax-M1-256k",
"MiniMax-M2.5",
"MiniMax-M2.5-highspeed",
"MiniMax-M2.1",
"MiniMax-M2.7",
],
"minimax-cn": [
"MiniMax-M2.7",
"MiniMax-M2.7-highspeed",
"MiniMax-M1",
"MiniMax-M1-40k",
"MiniMax-M1-80k",
"MiniMax-M1-128k",
"MiniMax-M1-256k",
"MiniMax-M2.5",
"MiniMax-M2.5-highspeed",
"MiniMax-M2.1",
"MiniMax-M2.7",
],
"anthropic": [
"claude-opus-4-6",
@@ -265,6 +269,202 @@ _PROVIDER_MODELS: dict[str, list[str]] = {
],
}
# ---------------------------------------------------------------------------
# Nous Portal free-model filtering
# ---------------------------------------------------------------------------
# Models that are ALLOWED to appear when priced as free on Nous Portal.
# Any other free model is hidden — prevents promotional/temporary free models
# from cluttering the selection when users are paying subscribers.
# Models in this list are ALSO filtered out if they are NOT free (i.e. they
# should only appear in the menu when they are genuinely free).
_NOUS_ALLOWED_FREE_MODELS: frozenset[str] = frozenset({
"xiaomi/mimo-v2-pro",
"xiaomi/mimo-v2-omni",
})
def _is_model_free(model_id: str, pricing: dict[str, dict[str, str]]) -> bool:
"""Return True if *model_id* has zero-cost prompt AND completion pricing."""
p = pricing.get(model_id)
if not p:
return False
try:
return float(p.get("prompt", "1")) == 0 and float(p.get("completion", "1")) == 0
except (TypeError, ValueError):
return False
def filter_nous_free_models(
model_ids: list[str],
pricing: dict[str, dict[str, str]],
) -> list[str]:
"""Filter the Nous Portal model list according to free-model policy.
Rules:
• Paid models that are NOT in the allowlist → keep (normal case).
• Free models that are NOT in the allowlist → drop.
• Allowlist models that ARE free → keep.
• Allowlist models that are NOT free → drop.
"""
if not pricing:
return model_ids # no pricing data — can't filter, show everything
result: list[str] = []
for mid in model_ids:
free = _is_model_free(mid, pricing)
if mid in _NOUS_ALLOWED_FREE_MODELS:
# Allowlist model: only show when it's actually free
if free:
result.append(mid)
else:
# Regular model: keep only when it's NOT free
if not free:
result.append(mid)
return result
# ---------------------------------------------------------------------------
# Nous Portal account tier detection
# ---------------------------------------------------------------------------
def fetch_nous_account_tier(access_token: str, portal_base_url: str = "") -> dict[str, Any]:
"""Fetch the user's Nous Portal account/subscription info.
Calls ``<portal>/api/oauth/account`` with the OAuth access token.
Returns the parsed JSON dict on success, e.g.::
{
"subscription": {
"plan": "Plus",
"tier": 2,
"monthly_charge": 20,
"credits_remaining": 1686.60,
...
},
...
}
Returns an empty dict on any failure (network, auth, parse).
"""
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",
}
try:
req = urllib.request.Request(url, headers=headers)
with urllib.request.urlopen(req, timeout=8) as resp:
return json.loads(resp.read().decode())
except Exception:
return {}
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).
"""
sub = account_info.get("subscription")
if not isinstance(sub, dict):
return False
charge = sub.get("monthly_charge")
if charge is None:
return False
try:
return float(charge) == 0
except (TypeError, ValueError):
return False
def partition_nous_models_by_tier(
model_ids: list[str],
pricing: dict[str, dict[str, str]],
free_tier: bool,
) -> tuple[list[str], list[str]]:
"""Split Nous models into (selectable, unavailable) based on user tier.
For paid-tier users: all models are selectable, none unavailable
(free-model filtering is handled separately by ``filter_nous_free_models``).
For free-tier users: only free models are selectable; paid models
are returned as unavailable (shown grayed out in the menu).
"""
if not free_tier:
return (model_ids, [])
if not pricing:
return (model_ids, []) # can't determine, show everything
selectable: list[str] = []
unavailable: list[str] = []
for mid in model_ids:
if _is_model_free(mid, pricing):
selectable.append(mid)
else:
unavailable.append(mid)
return (selectable, unavailable)
# ---------------------------------------------------------------------------
# TTL cache for free-tier detection — avoids repeated API calls within a
# session while still picking up upgrades quickly.
# ---------------------------------------------------------------------------
_FREE_TIER_CACHE_TTL: int = 180 # seconds (3 minutes)
_free_tier_cache: tuple[bool, float] | None = None # (result, timestamp)
def clear_nous_free_tier_cache() -> None:
"""Invalidate the cached free-tier result (e.g. after login/logout)."""
global _free_tier_cache
_free_tier_cache = None
def check_nous_free_tier() -> 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.
"""
global _free_tier_cache
import time
now = time.monotonic()
if _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
# 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)
_free_tier_cache = (result, now)
return result
except Exception:
_free_tier_cache = (False, now)
return False # default to paid on error — don't block users
_PROVIDER_LABELS = {
"openrouter": "OpenRouter",
"openai-codex": "OpenAI Codex",
@@ -283,6 +483,7 @@ _PROVIDER_LABELS = {
"ai-gateway": "AI Gateway",
"kilocode": "Kilo Code",
"alibaba": "Alibaba Cloud (DashScope)",
"qwen-oauth": "Qwen OAuth (Portal)",
"huggingface": "Hugging Face",
"custom": "Custom endpoint",
}
@@ -322,6 +523,7 @@ _PROVIDER_ALIASES = {
"aliyun": "alibaba",
"qwen": "alibaba",
"alibaba-cloud": "alibaba",
"qwen-portal": "qwen-oauth",
"hf": "huggingface",
"hugging-face": "huggingface",
"huggingface-hub": "huggingface",
@@ -567,6 +769,7 @@ def list_available_providers() -> list[dict[str, str]]:
"openrouter", "nous", "openai-codex", "copilot", "copilot-acp",
"gemini", "huggingface",
"zai", "kimi-coding", "minimax", "minimax-cn", "kilocode", "anthropic", "alibaba",
"qwen-oauth",
"opencode-zen", "opencode-go",
"ai-gateway", "deepseek", "custom",
]
@@ -935,10 +1138,6 @@ def _payload_items(payload: Any) -> list[dict[str, Any]]:
return []
def _extract_model_ids(payload: Any) -> list[str]:
return [item.get("id", "") for item in _payload_items(payload) if item.get("id")]
def copilot_default_headers() -> dict[str, str]:
"""Standard headers for Copilot API requests.

31
hermes_cli/nous_subscription.py
View File

@@ -167,20 +167,20 @@ def _resolve_browser_feature_state(
if browser_provider_explicit:
current_provider = browser_provider or "local"
if current_provider == "browserbase":
provider_available = managed_browser_available or direct_browserbase
available = bool(browser_local_available and direct_browserbase)
active = bool(browser_tool_enabled and available)
return current_provider, available, active, False
if current_provider == "browser-use":
provider_available = managed_browser_available or direct_browser_use
available = bool(browser_local_available and provider_available)
managed = bool(
browser_tool_enabled
and browser_local_available
and managed_browser_available
and not direct_browserbase
and not direct_browser_use
)
active = bool(browser_tool_enabled and available)
return current_provider, available, active, managed
if current_provider == "browser-use":
available = bool(browser_local_available and direct_browser_use)
active = bool(browser_tool_enabled and available)
return current_provider, available, active, False
if current_provider == "firecrawl":
available = bool(browser_local_available and direct_firecrawl)
active = bool(browser_tool_enabled and available)
@@ -193,16 +193,21 @@ def _resolve_browser_feature_state(
active = bool(browser_tool_enabled and available)
return current_provider, available, active, False
if managed_browser_available or direct_browserbase:
if managed_browser_available or direct_browser_use:
available = bool(browser_local_available)
managed = bool(
browser_tool_enabled
and browser_local_available
and managed_browser_available
and not direct_browserbase
and not direct_browser_use
)
active = bool(browser_tool_enabled and available)
return "browserbase", available, active, managed
return "browser-use", available, active, managed
if direct_browserbase:
available = bool(browser_local_available)
active = bool(browser_tool_enabled and available)
return "browserbase", available, active, False
available = bool(browser_local_available)
active = bool(browser_tool_enabled and available)
@@ -266,7 +271,7 @@ def get_nous_subscription_features(
managed_web_available = managed_tools_flag and nous_auth_present and is_managed_tool_gateway_ready("firecrawl")
managed_image_available = managed_tools_flag and nous_auth_present and is_managed_tool_gateway_ready("fal-queue")
managed_tts_available = managed_tools_flag and nous_auth_present and is_managed_tool_gateway_ready("openai-audio")
managed_browser_available = managed_tools_flag and nous_auth_present and is_managed_tool_gateway_ready("browserbase")
managed_browser_available = managed_tools_flag and nous_auth_present and is_managed_tool_gateway_ready("browser-use")
managed_modal_available = managed_tools_flag and nous_auth_present and is_managed_tool_gateway_ready("modal")
modal_state = resolve_modal_backend_state(
modal_mode,
@@ -512,10 +517,10 @@ def apply_nous_managed_defaults(
changed.add("tts")
if "browser" in selected_toolsets and not features.browser.explicit_configured and not (
get_env_value("BROWSERBASE_API_KEY")
or get_env_value("BROWSER_USE_API_KEY")
get_env_value("BROWSER_USE_API_KEY")
or get_env_value("BROWSERBASE_API_KEY")
):
browser_cfg["cloud_provider"] = "browserbase"
browser_cfg["cloud_provider"] = "browser-use"
changed.add("browser")
if "image_gen" in selected_toolsets and not get_env_value("FAL_KEY"):

10
hermes_cli/plugins.py
View File

@@ -36,8 +36,9 @@ import sys
import types
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Set
from typing import Any, Callable, Dict, List, Optional, Set, Union
from hermes_constants import get_hermes_home
from utils import env_var_enabled
try:
@@ -60,6 +61,8 @@ VALID_HOOKS: Set[str] = {
"post_api_request",
"on_session_start",
"on_session_end",
"on_session_finalize",
"on_session_reset",
}
ENTRY_POINTS_GROUP = "hermes_agent.plugins"
@@ -95,7 +98,7 @@ class PluginManifest:
version: str = ""
description: str = ""
author: str = ""
requires_env: List[str] = field(default_factory=list)
requires_env: List[Union[str, Dict[str, Any]]] = field(default_factory=list)
provides_tools: List[str] = field(default_factory=list)
provides_hooks: List[str] = field(default_factory=list)
source: str = "" # "user", "project", or "entrypoint"
@@ -258,8 +261,7 @@ class PluginManager:
manifests: List[PluginManifest] = []
# 1. User plugins (~/.hermes/plugins/)
hermes_home = os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes"))
user_dir = Path(hermes_home) / "plugins"
user_dir = get_hermes_home() / "plugins"
manifests.extend(self._scan_directory(user_dir, source="user"))
# 2. Project plugins (./.hermes/plugins/)

89
hermes_cli/plugins_cmd.py
View File

@@ -16,6 +16,8 @@ import subprocess
import sys
from pathlib import Path
from hermes_constants import get_hermes_home
logger = logging.getLogger(__name__)
# Minimum manifest version this installer understands.
@@ -26,8 +28,7 @@ _SUPPORTED_MANIFEST_VERSION = 1
def _plugins_dir() -> Path:
"""Return the user plugins directory, creating it if needed."""
hermes_home = os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes"))
plugins = Path(hermes_home) / "plugins"
plugins = get_hermes_home() / "plugins"
plugins.mkdir(parents=True, exist_ok=True)
return plugins
@@ -147,6 +148,82 @@ def _copy_example_files(plugin_dir: Path, console) -> None:
)
def _prompt_plugin_env_vars(manifest: dict, console) -> None:
"""Prompt for required environment variables declared in plugin.yaml.
``requires_env`` accepts two formats:
Simple list (backwards-compatible)::
requires_env:
- MY_API_KEY
Rich list with metadata::
requires_env:
- name: MY_API_KEY
description: "API key for Acme service"
url: "https://acme.com/keys"
secret: true
Already-set variables are skipped. Values are saved to the user's ``.env``.
"""
requires_env = manifest.get("requires_env") or []
if not requires_env:
return
from hermes_cli.config import get_env_value, save_env_value # noqa: F811
from hermes_constants import display_hermes_home
# Normalise to list-of-dicts
env_specs: list[dict] = []
for entry in requires_env:
if isinstance(entry, str):
env_specs.append({"name": entry})
elif isinstance(entry, dict) and entry.get("name"):
env_specs.append(entry)
# Filter to only vars that aren't already set
missing = [s for s in env_specs if not get_env_value(s["name"])]
if not missing:
return
plugin_name = manifest.get("name", "this plugin")
console.print(f"\n[bold]{plugin_name}[/bold] requires the following environment variables:\n")
for spec in missing:
name = spec["name"]
desc = spec.get("description", "")
url = spec.get("url", "")
secret = spec.get("secret", False)
label = f" {name}"
if desc:
label += f" — {desc}"
console.print(label)
if url:
console.print(f" [dim]Get yours at: {url}[/dim]")
try:
if secret:
import getpass
value = getpass.getpass(f" {name}: ").strip()
else:
value = input(f" {name}: ").strip()
except (EOFError, KeyboardInterrupt):
console.print(f"\n[dim] Skipped (you can set these later in {display_hermes_home()}/.env)[/dim]")
return
if value:
save_env_value(name, value)
os.environ[name] = value
console.print(f" [green]✓[/green] Saved to {display_hermes_home()}/.env")
else:
console.print(f" [dim] Skipped (set {name} in {display_hermes_home()}/.env later)[/dim]")
console.print()
def _display_after_install(plugin_dir: Path, identifier: str) -> None:
"""Show after-install.md if it exists, otherwise a default message."""
from rich.console import Console
@@ -218,7 +295,7 @@ def cmd_install(identifier: str, force: bool = False) -> None:
sys.exit(1)
# Warn about insecure / local URL schemes
if git_url.startswith("http://") or git_url.startswith("file://"):
if git_url.startswith(("http://", "file://")):
console.print(
"[yellow]Warning:[/yellow] Using insecure/local URL scheme. "
"Consider using https:// or git@ for production installs."
@@ -306,6 +383,12 @@ def cmd_install(identifier: str, force: bool = False) -> None:
# Copy .example files to their real names (e.g. config.yaml.example → config.yaml)
_copy_example_files(target, console)
# Re-read manifest from installed location (for env var prompting)
installed_manifest = _read_manifest(target)
# Prompt for required environment variables before showing after-install docs
_prompt_plugin_env_vars(installed_manifest, console)
_display_after_install(target, identifier)
console.print("[dim]Restart the gateway for the plugin to take effect:[/dim]")

3
hermes_cli/profiles.py
View File

@@ -26,7 +26,7 @@ import shutil
import stat
import subprocess
import sys
from dataclasses import dataclass, field
from dataclasses import dataclass
from pathlib import Path, PurePosixPath, PureWindowsPath
from typing import List, Optional
@@ -517,7 +517,6 @@ def delete_profile(name: str, yes: bool = False) -> Path:
]
# Check for service
from hermes_cli.gateway import _profile_suffix, get_service_name
wrapper_path = _get_wrapper_dir() / name
has_wrapper = wrapper_path.exists()
if has_wrapper:

29
hermes_cli/providers.py
View File

@@ -20,8 +20,7 @@ Other modules import from this file. No parallel registries.
from __future__ import annotations
import logging
import os
from dataclasses import dataclass, field
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple
logger = logging.getLogger(__name__)
@@ -59,6 +58,12 @@ HERMES_OVERLAYS: Dict[str, HermesOverlay] = {
auth_type="oauth_external",
base_url_override="https://chatgpt.com/backend-api/codex",
),
"qwen-oauth": HermesOverlay(
transport="openai_chat",
auth_type="oauth_external",
base_url_override="https://portal.qwen.ai/v1",
base_url_env_var="HERMES_QWEN_BASE_URL",
),
"copilot-acp": HermesOverlay(
transport="codex_responses",
auth_type="external_process",
@@ -345,26 +350,6 @@ def get_label(provider_id: str) -> str:
return canonical
# Build LABELS dict for backward compat
def _build_labels() -> Dict[str, str]:
"""Build labels dict from overlays + overrides. Lazy, cached."""
labels: Dict[str, str] = {}
for pid in HERMES_OVERLAYS:
labels[pid] = get_label(pid)
labels.update(_LABEL_OVERRIDES)
return labels
# Lazy-built on first access
_labels_cache: Optional[Dict[str, str]] = None
@property
def LABELS() -> Dict[str, str]:
"""Backward-compatible labels dict."""
global _labels_cache
if _labels_cache is None:
_labels_cache = _build_labels()
return _labels_cache
# For direct import compat, expose as module-level dict
# Built on demand by get_label() calls
LABELS: Dict[str, str] = {

105
hermes_cli/runtime_provider.py
View File

@@ -14,11 +14,13 @@ from agent.credential_pool import CredentialPool, PooledCredential, get_custom_p
from hermes_cli.auth import (
AuthError,
DEFAULT_CODEX_BASE_URL,
DEFAULT_QWEN_BASE_URL,
PROVIDER_REGISTRY,
format_auth_error,
resolve_provider,
resolve_nous_runtime_credentials,
resolve_codex_runtime_credentials,
resolve_qwen_runtime_credentials,
resolve_api_key_provider_credentials,
resolve_external_process_provider_credentials,
has_usable_secret,
@@ -148,6 +150,9 @@ def _resolve_runtime_from_pool_entry(
if provider == "openai-codex":
api_mode = "codex_responses"
base_url = base_url or DEFAULT_CODEX_BASE_URL
elif provider == "qwen-oauth":
api_mode = "chat_completions"
base_url = base_url or DEFAULT_QWEN_BASE_URL
elif provider == "anthropic":
api_mode = "anthropic_messages"
cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
@@ -163,6 +168,16 @@ def _resolve_runtime_from_pool_entry(
api_mode = _copilot_runtime_api_mode(model_cfg, getattr(entry, "runtime_api_key", ""))
else:
configured_provider = str(model_cfg.get("provider") or "").strip().lower()
# Honour model.base_url from config.yaml when the configured provider
# matches this provider — same pattern as the Anthropic branch above.
# Only override when the pool entry has no explicit base_url (i.e. it
# fell back to the hardcoded default). Env var overrides win (#6039).
pconfig = PROVIDER_REGISTRY.get(provider)
pool_url_is_default = pconfig and base_url.rstrip("/") == pconfig.inference_base_url.rstrip("/")
if configured_provider == provider and pool_url_is_default:
cfg_base_url = str(model_cfg.get("base_url") or "").strip().rstrip("/")
if cfg_base_url:
base_url = cfg_base_url
configured_mode = _parse_api_mode(model_cfg.get("api_mode"))
if configured_mode and _provider_supports_explicit_api_mode(provider, configured_provider):
api_mode = configured_mode
@@ -639,31 +654,65 @@ def resolve_runtime_provider(
)
if provider == "nous":
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")),
)
return {
"provider": "nous",
"api_mode": "chat_completions",
"base_url": creds.get("base_url", "").rstrip("/"),
"api_key": creds.get("api_key", ""),
"source": creds.get("source", "portal"),
"expires_at": creds.get("expires_at"),
"requested_provider": requested_provider,
}
try:
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")),
)
return {
"provider": "nous",
"api_mode": "chat_completions",
"base_url": creds.get("base_url", "").rstrip("/"),
"api_key": creds.get("api_key", ""),
"source": creds.get("source", "portal"),
"expires_at": creds.get("expires_at"),
"requested_provider": requested_provider,
}
except AuthError:
if requested_provider != "auto":
raise
# Auto-detected Nous but credentials are stale/revoked —
# fall through to env-var providers (e.g. OpenRouter).
logger.info("Auto-detected Nous provider but credentials failed; "
"falling through to next provider.")
if provider == "openai-codex":
creds = resolve_codex_runtime_credentials()
return {
"provider": "openai-codex",
"api_mode": "codex_responses",
"base_url": creds.get("base_url", "").rstrip("/"),
"api_key": creds.get("api_key", ""),
"source": creds.get("source", "hermes-auth-store"),
"last_refresh": creds.get("last_refresh"),
"requested_provider": requested_provider,
}
try:
creds = resolve_codex_runtime_credentials()
return {
"provider": "openai-codex",
"api_mode": "codex_responses",
"base_url": creds.get("base_url", "").rstrip("/"),
"api_key": creds.get("api_key", ""),
"source": creds.get("source", "hermes-auth-store"),
"last_refresh": creds.get("last_refresh"),
"requested_provider": requested_provider,
}
except AuthError:
if requested_provider != "auto":
raise
# Auto-detected Codex but credentials are stale/revoked —
# fall through to env-var providers (e.g. OpenRouter).
logger.info("Auto-detected Codex provider but credentials failed; "
"falling through to next provider.")
if provider == "qwen-oauth":
try:
creds = resolve_qwen_runtime_credentials()
return {
"provider": "qwen-oauth",
"api_mode": "chat_completions",
"base_url": creds.get("base_url", "").rstrip("/"),
"api_key": creds.get("api_key", ""),
"source": creds.get("source", "qwen-cli"),
"expires_at_ms": creds.get("expires_at_ms"),
"requested_provider": requested_provider,
}
except AuthError:
if requested_provider != "auto":
raise
logger.info("Qwen OAuth credentials failed; "
"falling through to next provider.")
if provider == "copilot-acp":
creds = resolve_external_process_provider_credentials(provider)
@@ -708,7 +757,15 @@ def resolve_runtime_provider(
pconfig = PROVIDER_REGISTRY.get(provider)
if pconfig and pconfig.auth_type == "api_key":
creds = resolve_api_key_provider_credentials(provider)
base_url = creds.get("base_url", "").rstrip("/")
# Honour model.base_url from config.yaml when the configured provider
# matches this provider — mirrors the Anthropic path above. Without
# this, users who set model.base_url to e.g. api.minimaxi.com/anthropic
# (China endpoint) still get the hardcoded api.minimax.io default (#6039).
cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
cfg_base_url = ""
if cfg_provider == provider:
cfg_base_url = (model_cfg.get("base_url") or "").strip().rstrip("/")
base_url = cfg_base_url or creds.get("base_url", "").rstrip("/")
api_mode = "chat_completions"
if provider == "copilot":
api_mode = _copilot_runtime_api_mode(model_cfg, creds.get("api_key", ""))

51
hermes_cli/setup.py
View File

@@ -21,7 +21,6 @@ from typing import Optional, Dict, Any
from hermes_cli.nous_subscription import (
apply_nous_provider_defaults,
get_nous_subscription_explainer_lines,
get_nous_subscription_features,
)
from tools.tool_backend_helpers import managed_nous_tools_enabled
@@ -43,18 +42,6 @@ def _model_config_dict(config: Dict[str, Any]) -> Dict[str, Any]:
return {}
def _set_model_provider(
config: Dict[str, Any], provider_id: str, base_url: str = ""
) -> None:
model_cfg = _model_config_dict(config)
model_cfg["provider"] = provider_id
if base_url:
model_cfg["base_url"] = base_url.rstrip("/")
else:
model_cfg.pop("base_url", None)
config["model"] = model_cfg
def _set_default_model(config: Dict[str, Any], model_name: str) -> None:
if not model_name:
return
@@ -118,8 +105,8 @@ _DEFAULT_PROVIDER_MODELS = {
],
"zai": ["glm-5", "glm-4.7", "glm-4.5", "glm-4.5-flash"],
"kimi-coding": ["kimi-k2.5", "kimi-k2-thinking", "kimi-k2-turbo-preview"],
"minimax": ["MiniMax-M2.7", "MiniMax-M2.7-highspeed", "MiniMax-M2.5", "MiniMax-M2.5-highspeed", "MiniMax-M2.1"],
"minimax-cn": ["MiniMax-M2.7", "MiniMax-M2.7-highspeed", "MiniMax-M2.5", "MiniMax-M2.5-highspeed", "MiniMax-M2.1"],
"minimax": ["MiniMax-M1", "MiniMax-M1-40k", "MiniMax-M1-80k", "MiniMax-M1-128k", "MiniMax-M1-256k", "MiniMax-M2.5", "MiniMax-M2.7"],
"minimax-cn": ["MiniMax-M1", "MiniMax-M1-40k", "MiniMax-M1-80k", "MiniMax-M1-128k", "MiniMax-M1-256k", "MiniMax-M2.5", "MiniMax-M2.7"],
"ai-gateway": ["anthropic/claude-opus-4.6", "anthropic/claude-sonnet-4.6", "openai/gpt-5", "google/gemini-3-flash"],
"kilocode": ["anthropic/claude-opus-4.6", "anthropic/claude-sonnet-4.6", "openai/gpt-5.4", "google/gemini-3-pro-preview", "google/gemini-3-flash-preview"],
"opencode-zen": ["gpt-5.4", "gpt-5.3-codex", "claude-sonnet-4-6", "gemini-3-flash", "glm-5", "kimi-k2.5", "minimax-m2.7"],
@@ -327,16 +314,6 @@ def _setup_provider_model_selection(config, provider_id, current_model, prompt_c
config["model"] = model_cfg
def _sync_model_from_disk(config: Dict[str, Any]) -> None:
disk_model = load_config().get("model")
if isinstance(disk_model, dict):
model_cfg = _model_config_dict(config)
model_cfg.update(disk_model)
config["model"] = model_cfg
elif isinstance(disk_model, str) and disk_model.strip():
_set_default_model(config, disk_model.strip())
# Import config helpers
from hermes_cli.config import (
get_hermes_home,
@@ -444,10 +421,22 @@ def _curses_prompt_choice(question: str, choices: list, default: int = 0) -> int
curses.init_pair(1, curses.COLOR_GREEN, -1)
curses.init_pair(2, curses.COLOR_YELLOW, -1)
cursor = default
scroll_offset = 0
while True:
stdscr.clear()
max_y, max_x = stdscr.getmaxyx()
# Rows available for list items: rows 2..(max_y-2) inclusive.
visible = max(1, max_y - 3)
# Scroll the viewport so the cursor is always visible.
if cursor < scroll_offset:
scroll_offset = cursor
elif cursor >= scroll_offset + visible:
scroll_offset = cursor - visible + 1
scroll_offset = max(0, min(scroll_offset, max(0, len(choices) - visible)))
try:
stdscr.addnstr(
0,
@@ -459,12 +448,12 @@ def _curses_prompt_choice(question: str, choices: list, default: int = 0) -> int
except curses.error:
pass
for i, choice in enumerate(choices):
y = i + 2
for row, i in enumerate(range(scroll_offset, min(scroll_offset + visible, len(choices)))):
y = row + 2
if y >= max_y - 1:
break
arrow = "→" if i == cursor else " "
line = f" {arrow} {choice}"
line = f" {arrow} {choices[i]}"
attr = curses.A_NORMAL
if i == cursor:
attr = curses.A_BOLD
@@ -660,14 +649,14 @@ def _print_setup_summary(config: dict, hermes_home):
# Browser tools (local Chromium, Camofox, Browserbase, Browser Use, or Firecrawl)
browser_provider = subscription_features.browser.current_provider
if subscription_features.browser.managed_by_nous:
tool_status.append(("Browser Automation (Nous Browserbase)", True, None))
tool_status.append(("Browser Automation (Nous Browser Use)", True, None))
elif subscription_features.browser.available:
label = "Browser Automation"
if browser_provider:
label = f"Browser Automation ({browser_provider})"
tool_status.append((label, True, None))
else:
missing_browser_hint = "npm install -g agent-browser, set CAMOFOX_URL, or configure Browserbase"
missing_browser_hint = "npm install -g agent-browser, set CAMOFOX_URL, or configure Browser Use or Browserbase"
if browser_provider == "Browserbase":
missing_browser_hint = (
"npm install -g agent-browser and set "
@@ -1348,8 +1337,6 @@ def setup_terminal_backend(config: dict):
terminal_choices.append(f"Keep current ({current_backend})")
idx_to_backend[keep_current_idx] = current_backend
default_terminal = backend_to_idx.get(current_backend, 0)
terminal_idx = prompt_choice(
"Select terminal backend:", terminal_choices, keep_current_idx
)

1
hermes_cli/skin_engine.py
View File

@@ -96,7 +96,6 @@ Activate with ``/skin <name>`` in the CLI or ``display.skin: <name>`` in config.
"""
import logging
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

22
hermes_cli/status.py
View File

@@ -123,7 +123,8 @@ def show_status(args):
"MiniMax-CN": "MINIMAX_CN_API_KEY",
"Firecrawl": "FIRECRAWL_API_KEY",
"Tavily": "TAVILY_API_KEY",
"Browserbase": "BROWSERBASE_API_KEY", # Optional — local browser works without this
"Browser Use": "BROWSER_USE_API_KEY", # Optional — local browser works without this
"Browserbase": "BROWSERBASE_API_KEY", # Optional — direct credentials only
"FAL": "FAL_KEY",
"Tinker": "TINKER_API_KEY",
"WandB": "WANDB_API_KEY",
@@ -152,12 +153,14 @@ def show_status(args):
print(color("◆ Auth Providers", Colors.CYAN, Colors.BOLD))
try:
from hermes_cli.auth import get_nous_auth_status, get_codex_auth_status
from hermes_cli.auth import get_nous_auth_status, get_codex_auth_status, get_qwen_auth_status
nous_status = get_nous_auth_status()
codex_status = get_codex_auth_status()
qwen_status = get_qwen_auth_status()
except Exception:
nous_status = {}
codex_status = {}
qwen_status = {}
nous_logged_in = bool(nous_status.get("logged_in"))
print(
@@ -188,6 +191,21 @@ def show_status(args):
if codex_status.get("error") and not codex_logged_in:
print(f" Error: {codex_status.get('error')}")
qwen_logged_in = bool(qwen_status.get("logged_in"))
print(
f" {'Qwen OAuth':<12} {check_mark(qwen_logged_in)} "
f"{'logged in' if qwen_logged_in else 'not logged in (run: qwen auth qwen-oauth)'}"
)
qwen_auth_file = qwen_status.get("auth_file")
if qwen_auth_file:
print(f" Auth file: {qwen_auth_file}")
qwen_exp = qwen_status.get("expires_at_ms")
if qwen_exp:
from datetime import datetime, timezone
print(f" Access exp: {datetime.fromtimestamp(int(qwen_exp) / 1000, tz=timezone.utc).isoformat()}")
if qwen_status.get("error") and not qwen_logged_in:
print(f" Error: {qwen_status.get('error')}")
# =========================================================================
# Nous Subscription Features
# =========================================================================

48
hermes_cli/tools_config.py
View File

@@ -61,22 +61,6 @@ def _prompt(question: str, default: str = None, password: bool = False) -> str:
print()
return default or ""
def _prompt_yes_no(question: str, default: bool = True) -> bool:
default_str = "Y/n" if default else "y/N"
while True:
try:
value = input(color(f"{question} [{default_str}]: ", Colors.YELLOW)).strip().lower()
except (KeyboardInterrupt, EOFError):
print()
return default
if not value:
return default
if value in ('y', 'yes'):
return True
if value in ('n', 'no'):
return False
# ─── Toolset Registry ─────────────────────────────────────────────────────────
# Toolsets shown in the configurator, grouped for display.
@@ -280,21 +264,21 @@ TOOL_CATEGORIES = {
"icon": "🌐",
"providers": [
{
"name": "Nous Subscription (Browserbase cloud)",
"tag": "Managed Browserbase billed to your subscription",
"name": "Nous Subscription (Browser Use cloud)",
"tag": "Managed Browser Use billed to your subscription",
"env_vars": [],
"browser_provider": "browserbase",
"browser_provider": "browser-use",
"requires_nous_auth": True,
"managed_nous_feature": "browser",
"override_env_vars": ["BROWSERBASE_API_KEY", "BROWSERBASE_PROJECT_ID"],
"post_setup": "browserbase",
"override_env_vars": ["BROWSER_USE_API_KEY"],
"post_setup": "agent_browser",
},
{
"name": "Local Browser",
"tag": "Free headless Chromium (no API key needed)",
"env_vars": [],
"browser_provider": "local",
"post_setup": "browserbase", # Same npm install for agent-browser
"post_setup": "agent_browser",
},
{
"name": "Browserbase",
@@ -304,7 +288,7 @@ TOOL_CATEGORIES = {
{"key": "BROWSERBASE_PROJECT_ID", "prompt": "Browserbase project ID"},
],
"browser_provider": "browserbase",
"post_setup": "browserbase",
"post_setup": "agent_browser",
},
{
"name": "Browser Use",
@@ -313,7 +297,7 @@ TOOL_CATEGORIES = {
{"key": "BROWSER_USE_API_KEY", "prompt": "Browser Use API key", "url": "https://browser-use.com"},
],
"browser_provider": "browser-use",
"post_setup": "browserbase",
"post_setup": "agent_browser",
},
{
"name": "Firecrawl",
@@ -322,7 +306,7 @@ TOOL_CATEGORIES = {
{"key": "FIRECRAWL_API_KEY", "prompt": "Firecrawl API key", "url": "https://firecrawl.dev"},
],
"browser_provider": "firecrawl",
"post_setup": "browserbase",
"post_setup": "agent_browser",
},
{
"name": "Camofox",
@@ -381,7 +365,7 @@ TOOLSET_ENV_REQUIREMENTS = {
def _run_post_setup(post_setup_key: str):
"""Run post-setup hooks for tools that need extra installation steps."""
import shutil
if post_setup_key == "browserbase":
if post_setup_key in ("agent_browser", "browserbase"):
node_modules = PROJECT_ROOT / "node_modules" / "agent-browser"
if not node_modules.exists() and shutil.which("npm"):
_print_info(" Installing Node.js dependencies for browser tools...")
@@ -570,6 +554,7 @@ def _get_platform_tools(
# MCP servers are expected to be available on all platforms by default.
# If the platform explicitly lists one or more MCP server names, treat that
# as an allowlist. Otherwise include every globally enabled MCP server.
# Special sentinel: "no_mcp" in the toolset list disables all MCP servers.
mcp_servers = config.get("mcp_servers") or {}
enabled_mcp_servers = {
name
@@ -577,10 +562,15 @@ def _get_platform_tools(
if isinstance(server_cfg, dict)
and _parse_enabled_flag(server_cfg.get("enabled", True), default=True)
}
explicit_mcp_servers = explicit_passthrough & enabled_mcp_servers
enabled_toolsets.update(explicit_passthrough - enabled_mcp_servers)
# Allow "no_mcp" sentinel to opt out of all MCP servers for this platform
if "no_mcp" in toolset_names:
explicit_mcp_servers = set()
enabled_toolsets.update(explicit_passthrough - enabled_mcp_servers - {"no_mcp"})
else:
explicit_mcp_servers = explicit_passthrough & enabled_mcp_servers
enabled_toolsets.update(explicit_passthrough - enabled_mcp_servers)
if include_default_mcp_servers:
if explicit_mcp_servers:
if explicit_mcp_servers or "no_mcp" in toolset_names:
enabled_toolsets.update(explicit_mcp_servers)
else:
enabled_toolsets.update(enabled_mcp_servers)

5
hermes_cli/uninstall.py
View File

@@ -6,7 +6,6 @@ Provides options for:
- Keep data: Remove code but keep ~/.hermes/ (configs, sessions, logs)
"""
import os
import shutil
import subprocess
from pathlib import Path
@@ -24,10 +23,6 @@ def log_success(msg: str):
def log_warn(msg: str):
print(f"{color('⚠', Colors.YELLOW)} {msg}")
def log_error(msg: str):
print(f"{color('✗', Colors.RED)} {msg}")
def get_project_root() -> Path:
"""Get the project installation directory."""
return Path(__file__).parent.parent.resolve()

7
hermes_cli/webhook.py
View File

@@ -16,7 +16,7 @@ import re
import secrets
import time
from pathlib import Path
from typing import Dict, Optional
from typing import Dict
from hermes_constants import display_hermes_home
@@ -25,9 +25,8 @@ _SUBSCRIPTIONS_FILENAME = "webhook_subscriptions.json"
def _hermes_home() -> Path:
return Path(
os.getenv("HERMES_HOME", str(Path.home() / ".hermes"))
).expanduser()
from hermes_constants import get_hermes_home
return get_hermes_home()
def _subscriptions_path() -> Path: