common/hermes
2
0
Fork 0
Code Issues Pull Requests Actions 1 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-06 17:56:45 -05:00
parent 4c7d5ec778 e651e04100
commit e2b3b1c5e4
158 changed files with 21437 additions and 2022 deletions
Download Patch File Download Diff File Expand all files Collapse all files

159
hermes_cli/auth.py
View File

@@ -69,6 +69,7 @@ DEVICE_AUTH_POLL_INTERVAL_CAP_SECONDS = 1 # poll at most every 1s
DEFAULT_CODEX_BASE_URL = "https://chatgpt.com/backend-api/codex"
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
@@ -125,6 +126,14 @@ PROVIDER_REGISTRY: Dict[str, ProviderConfig] = {
inference_base_url=DEFAULT_COPILOT_ACP_BASE_URL,
base_url_env_var="COPILOT_ACP_BASE_URL",
),
"gemini": ProviderConfig(
id="gemini",
name="Google AI Studio",
auth_type="api_key",
inference_base_url="https://generativelanguage.googleapis.com/v1beta/openai",
api_key_env_vars=("GOOGLE_API_KEY", "GEMINI_API_KEY"),
base_url_env_var="GEMINI_BASE_URL",
),
"zai": ProviderConfig(
id="zai",
name="Z.AI / GLM",
@@ -711,6 +720,32 @@ def deactivate_provider() -> None:
# Provider Resolution — picks which provider to use
# =============================================================================
def _get_config_hint_for_unknown_provider(provider_name: str) -> str:
"""Return a helpful hint string when provider resolution fails.
Checks for common config.yaml mistakes (malformed custom_providers, etc.)
and returns a human-readable diagnostic, or empty string if nothing found.
"""
try:
from hermes_cli.config import validate_config_structure
issues = validate_config_structure()
if not issues:
return ""
lines = ["Config issue detected — run 'hermes doctor' for full diagnostics:"]
for ci in issues:
prefix = "ERROR" if ci.severity == "error" else "WARNING"
lines.append(f" [{prefix}] {ci.message}")
# Show first line of hint
first_hint = ci.hint.splitlines()[0] if ci.hint else ""
if first_hint:
lines.append(f" → {first_hint}")
return "\n".join(lines)
except Exception:
return ""
def resolve_provider(
requested: Optional[str] = None,
*,
@@ -732,6 +767,7 @@ def resolve_provider(
# Normalize provider aliases
_PROVIDER_ALIASES = {
"glm": "zai", "z-ai": "zai", "z.ai": "zai", "zhipu": "zai",
"google": "gemini", "google-gemini": "gemini", "google-ai-studio": "gemini",
"kimi": "kimi-coding", "moonshot": "kimi-coding",
"minimax-china": "minimax-cn", "minimax_cn": "minimax-cn",
"claude": "anthropic", "claude-code": "anthropic",
@@ -757,10 +793,14 @@ def resolve_provider(
if normalized in PROVIDER_REGISTRY:
return normalized
if normalized != "auto":
raise AuthError(
f"Unknown provider '{normalized}'.",
code="invalid_provider",
)
# Check for common config.yaml issues that cause this error
_config_hint = _get_config_hint_for_unknown_provider(normalized)
msg = f"Unknown provider '{normalized}'."
if _config_hint:
msg += f"\n\n{_config_hint}"
else:
msg += " Check 'hermes model' for available providers, or run 'hermes doctor' to diagnose config issues."
raise AuthError(msg, code="invalid_provider")
# Explicit one-off CLI creds always mean openrouter/custom
if explicit_api_key or explicit_base_url:
@@ -2143,8 +2183,18 @@ def _reset_config_provider() -> Path:
return config_path
def _prompt_model_selection(model_ids: List[str], current_model: str = "") -> Optional[str]:
"""Interactive model selection. Puts current_model first with a marker. Returns chosen model ID or None."""
def _prompt_model_selection(
model_ids: List[str],
current_model: str = "",
pricing: Optional[Dict[str, Dict[str, str]]] = None,
) -> 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.
"""
from hermes_cli.models import _format_price_per_mtok
# Reorder: current model first, then the rest (deduplicated)
ordered = []
if current_model and current_model in model_ids:
@@ -2153,15 +2203,61 @@ def _prompt_model_selection(model_ids: List[str], current_model: str = "") -> Op
if mid not in ordered:
ordered.append(mid)
# Build display labels with marker on current
# 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
# Pre-compute formatted prices and dynamic column widths
_price_cache: dict[str, tuple[str, str, str]] = {}
price_col = 3 # minimum width
cache_col = 0 # only set if any model has cache pricing
has_cache = False
if has_pricing:
for mid in ordered:
p = pricing.get(mid) # type: ignore[union-attr]
if p:
inp = _format_price_per_mtok(p.get("prompt", ""))
out = _format_price_per_mtok(p.get("completion", ""))
cache_read = p.get("input_cache_read", "")
cache = _format_price_per_mtok(cache_read) if cache_read else ""
if cache:
has_cache = True
else:
inp, out, cache = "", "", ""
_price_cache[mid] = (inp, out, cache)
price_col = max(price_col, len(inp), len(out))
cache_col = max(cache_col, len(cache))
if has_cache:
cache_col = max(cache_col, 5) # minimum: "Cache" header
def _label(mid):
if has_pricing:
inp, out, cache = _price_cache.get(mid, ("", "", ""))
price_part = f" {inp:>{price_col}} {out:>{price_col}}"
if has_cache:
price_part += f" {cache:>{cache_col}}"
base = f"{mid:<{name_col}}{price_part}"
else:
base = mid
if mid == current_model:
return f"{mid} ← currently in use"
return mid
base += " ← currently in use"
return base
# Default cursor on the current model (index 0 if it was reordered to top)
default_idx = 0
# Build a pricing header hint for the menu title
menu_title = "Select default model:"
if has_pricing:
# Align the header with the model column.
# Each choice is " {label}" (2 spaces) and simple_term_menu prepends
# a 3-char cursor region ("-> " or " "), so content starts at col 5.
pad = " " * 5
header = f"\n{pad}{'':>{name_col}} {'In':>{price_col}} {'Out':>{price_col}}"
if has_cache:
header += f" {'Cache':>{cache_col}}"
menu_title += header + " /Mtok"
# Try arrow-key menu first, fall back to number input
try:
from simple_term_menu import TerminalMenu
@@ -2176,7 +2272,7 @@ def _prompt_model_selection(model_ids: List[str], current_model: str = "") -> Op
menu_highlight_style=("fg_green",),
cycle_cursor=True,
clear_screen=False,
title="Select default model:",
title=menu_title,
)
idx = menu.show()
if idx is None:
@@ -2192,12 +2288,13 @@ def _prompt_model_selection(model_ids: List[str], current_model: str = "") -> Op
pass
# Fallback: numbered list
print("Select default model:")
print(menu_title)
num_width = len(str(len(ordered) + 2))
for i, mid in enumerate(ordered, 1):
print(f" {i}. {_label(mid)}")
print(f" {i:>{num_width}}. {_label(mid)}")
n = len(ordered)
print(f" {n + 1}. Enter custom model name")
print(f" {n + 2}. Skip (keep current)")
print(f" {n + 1:>{num_width}}. Enter custom model name")
print(f" {n + 2:>{num_width}}. Skip (keep current)")
print()
while True:
@@ -2556,13 +2653,26 @@ def _nous_device_code_login(
"agent_key_reused": None,
"agent_key_obtained_at": None,
}
return refresh_nous_oauth_from_state(
auth_state,
min_key_ttl_seconds=min_key_ttl_seconds,
timeout_seconds=timeout_seconds,
force_refresh=False,
force_mint=True,
)
try:
return refresh_nous_oauth_from_state(
auth_state,
min_key_ttl_seconds=min_key_ttl_seconds,
timeout_seconds=timeout_seconds,
force_refresh=False,
force_mint=True,
)
except AuthError as exc:
if exc.code == "subscription_required":
portal_url = auth_state.get(
"portal_base_url", DEFAULT_NOUS_PORTAL_URL
).rstrip("/")
print()
print("Your Nous Portal account does not have an active subscription.")
print(f" Subscribe here: {portal_url}/billing")
print()
print("After subscribing, run `hermes model` again to finish setup.")
raise SystemExit(1)
raise
def _login_nous(args, pconfig: ProviderConfig) -> None:
@@ -2577,8 +2687,8 @@ def _login_nous(args, pconfig: ProviderConfig) -> None:
try:
auth_state = _nous_device_code_login(
portal_base_url=getattr(args, "portal_url", None) or pconfig.portal_base_url,
inference_base_url=getattr(args, "inference_url", None) or pconfig.inference_base_url,
portal_base_url=getattr(args, "portal_url", None),
inference_base_url=getattr(args, "inference_url", None),
client_id=getattr(args, "client_id", None) or pconfig.client_id,
scope=getattr(args, "scope", None) or pconfig.scope,
open_browser=not getattr(args, "no_browser", False),
@@ -2587,6 +2697,7 @@ def _login_nous(args, pconfig: ProviderConfig) -> None:
ca_bundle=ca_bundle,
min_key_ttl_seconds=5 * 60,
)
inference_base_url = auth_state["inference_base_url"]
verify: bool | str = False if insecure else (ca_bundle if ca_bundle else True)
@@ -2610,8 +2721,6 @@ def _login_nous(args, pconfig: ProviderConfig) -> None:
code="invalid_token",
)
# Use curated model list (same as OpenRouter defaults) instead
# of the full /models dump which returns hundreds of models.
from hermes_cli.models import _PROVIDER_MODELS
model_ids = _PROVIDER_MODELS.get("nous", [])

272
hermes_cli/commands.py
View File

@@ -366,21 +366,46 @@ def telegram_bot_commands() -> list[tuple[str, str]]:
for cmd in COMMAND_REGISTRY:
if not _is_gateway_available(cmd, overrides):
continue
tg_name = cmd.name.replace("-", "_")
result.append((tg_name, cmd.description))
tg_name = _sanitize_telegram_name(cmd.name)
if tg_name:
result.append((tg_name, cmd.description))
return result
_TG_NAME_LIMIT = 32
_CMD_NAME_LIMIT = 32
"""Max command name length shared by Telegram and Discord."""
# Backward-compat alias — tests and external code may reference the old name.
_TG_NAME_LIMIT = _CMD_NAME_LIMIT
# Telegram Bot API allows only lowercase a-z, 0-9, and underscores in
# command names. This regex strips everything else after initial conversion.
_TG_INVALID_CHARS = re.compile(r"[^a-z0-9_]")
_TG_MULTI_UNDERSCORE = re.compile(r"_{2,}")
def _clamp_telegram_names(
def _sanitize_telegram_name(raw: str) -> str:
"""Convert a command/skill/plugin name to a valid Telegram command name.
Telegram requires: 1-32 chars, lowercase a-z, digits 0-9, underscores only.
Steps: lowercase → replace hyphens with underscores → strip all other
invalid characters → collapse consecutive underscores → strip leading/
trailing underscores.
"""
name = raw.lower().replace("-", "_")
name = _TG_INVALID_CHARS.sub("", name)
name = _TG_MULTI_UNDERSCORE.sub("_", name)
return name.strip("_")
def _clamp_command_names(
entries: list[tuple[str, str]],
reserved: set[str],
) -> list[tuple[str, str]]:
"""Enforce Telegram's 32-char command name limit with collision avoidance.
"""Enforce 32-char command name limit with collision avoidance.
Names exceeding 32 chars are truncated. If truncation creates a duplicate
Both Telegram and Discord cap slash command names at 32 characters.
Names exceeding the limit are truncated. If truncation creates a duplicate
(against *reserved* names or earlier entries in the same batch), the name is
shortened to 31 chars and a digit ``0``-``9`` is appended to differentiate.
If all 10 digit slots are taken the entry is silently dropped.
@@ -388,10 +413,10 @@ def _clamp_telegram_names(
used: set[str] = set(reserved)
result: list[tuple[str, str]] = []
for name, desc in entries:
if len(name) > _TG_NAME_LIMIT:
candidate = name[:_TG_NAME_LIMIT]
if len(name) > _CMD_NAME_LIMIT:
candidate = name[:_CMD_NAME_LIMIT]
if candidate in used:
prefix = name[:_TG_NAME_LIMIT - 1]
prefix = name[:_CMD_NAME_LIMIT - 1]
for digit in range(10):
candidate = f"{prefix}{digit}"
if candidate not in used:
@@ -407,6 +432,129 @@ def _clamp_telegram_names(
return result
# Backward-compat alias.
_clamp_telegram_names = _clamp_command_names
# ---------------------------------------------------------------------------
# Shared skill/plugin collection for gateway platforms
# ---------------------------------------------------------------------------
def _collect_gateway_skill_entries(
platform: str,
max_slots: int,
reserved_names: set[str],
desc_limit: int = 100,
sanitize_name: "Callable[[str], str] | None" = None,
) -> tuple[list[tuple[str, str, str]], int]:
"""Collect plugin + skill entries for a gateway platform.
Priority order:
1. Plugin slash commands (take precedence over skills)
2. Built-in skill commands (fill remaining slots, alphabetical)
Only skills are trimmed when the cap is reached.
Hub-installed skills are excluded. Per-platform disabled skills are
excluded.
Args:
platform: Platform identifier for per-platform skill filtering
(``"telegram"``, ``"discord"``, etc.).
max_slots: Maximum number of entries to return (remaining slots after
built-in/core commands).
reserved_names: Names already taken by built-in commands. Mutated
in-place as new names are added.
desc_limit: Max description length (40 for Telegram, 100 for Discord).
sanitize_name: Optional name transform applied before clamping, e.g.
:func:`_sanitize_telegram_name` for Telegram. May return an
empty string to signal "skip this entry".
Returns:
``(entries, hidden_count)`` where *entries* is a list of
``(name, description, cmd_key)`` triples and *hidden_count* is the
number of skill entries dropped due to the cap. ``cmd_key`` is the
original ``/skill-name`` key from :func:`get_skill_commands`.
"""
all_entries: list[tuple[str, str, str]] = []
# --- Tier 1: Plugin slash commands (never trimmed) ---------------------
plugin_pairs: list[tuple[str, str]] = []
try:
from hermes_cli.plugins import get_plugin_manager
pm = get_plugin_manager()
plugin_cmds = getattr(pm, "_plugin_commands", {})
for cmd_name in sorted(plugin_cmds):
name = sanitize_name(cmd_name) if sanitize_name else cmd_name
if not name:
continue
desc = "Plugin command"
if len(desc) > desc_limit:
desc = desc[:desc_limit - 3] + "..."
plugin_pairs.append((name, desc))
except Exception:
pass
plugin_pairs = _clamp_command_names(plugin_pairs, reserved_names)
reserved_names.update(n for n, _ in plugin_pairs)
# Plugins have no cmd_key — use empty string as placeholder
for n, d in plugin_pairs:
all_entries.append((n, d, ""))
# --- Tier 2: Built-in skill commands (trimmed at cap) -----------------
_platform_disabled: set[str] = set()
try:
from agent.skill_utils import get_disabled_skill_names
_platform_disabled = get_disabled_skill_names(platform=platform)
except Exception:
pass
skill_triples: list[tuple[str, str, str]] = []
try:
from agent.skill_commands import get_skill_commands
from tools.skills_tool import SKILLS_DIR
_skills_dir = str(SKILLS_DIR.resolve())
_hub_dir = str((SKILLS_DIR / ".hub").resolve())
skill_cmds = get_skill_commands()
for cmd_key in sorted(skill_cmds):
info = skill_cmds[cmd_key]
skill_path = info.get("skill_md_path", "")
if not skill_path.startswith(_skills_dir):
continue
if skill_path.startswith(_hub_dir):
continue
skill_name = info.get("name", "")
if skill_name in _platform_disabled:
continue
raw_name = cmd_key.lstrip("/")
name = sanitize_name(raw_name) if sanitize_name else raw_name
if not name:
continue
desc = info.get("description", "")
if len(desc) > desc_limit:
desc = desc[:desc_limit - 3] + "..."
skill_triples.append((name, desc, cmd_key))
except Exception:
pass
# Clamp names; _clamp_command_names works on (name, desc) pairs so we
# need to zip/unzip.
skill_pairs = [(n, d) for n, d, _ in skill_triples]
key_by_pair = {(n, d): k for n, d, k in skill_triples}
skill_pairs = _clamp_command_names(skill_pairs, reserved_names)
# Skills fill remaining slots — only tier that gets trimmed
remaining = max(0, max_slots - len(all_entries))
hidden_count = max(0, len(skill_pairs) - remaining)
for n, d in skill_pairs[:remaining]:
all_entries.append((n, d, key_by_pair.get((n, d), "")))
return all_entries[:max_slots], hidden_count
# ---------------------------------------------------------------------------
# Platform-specific wrappers
# ---------------------------------------------------------------------------
def telegram_menu_commands(max_commands: int = 100) -> tuple[list[tuple[str, str]], int]:
"""Return Telegram menu commands capped to the Bot API limit.
@@ -425,80 +573,52 @@ def telegram_menu_commands(max_commands: int = 100) -> tuple[list[tuple[str, str
skill commands omitted due to the cap.
"""
core_commands = list(telegram_bot_commands())
# Reserve core names so plugin/skill truncation can't collide with them
reserved_names = {n for n, _ in core_commands}
all_commands = list(core_commands)
# Plugin slash commands get priority over skills
plugin_entries: list[tuple[str, str]] = []
try:
from hermes_cli.plugins import get_plugin_manager
pm = get_plugin_manager()
plugin_cmds = getattr(pm, "_plugin_commands", {})
for cmd_name in sorted(plugin_cmds):
tg_name = cmd_name.replace("-", "_")
desc = "Plugin command"
if len(desc) > 40:
desc = desc[:37] + "..."
plugin_entries.append((tg_name, desc))
except Exception:
pass
# Clamp plugin names to 32 chars with collision avoidance
plugin_entries = _clamp_telegram_names(plugin_entries, reserved_names)
reserved_names.update(n for n, _ in plugin_entries)
all_commands.extend(plugin_entries)
# Load per-platform disabled skills so they don't consume menu slots.
# get_skill_commands() already filters the *global* disabled list, but
# per-platform overrides (skills.platform_disabled.telegram) were never
# applied here — that's what this block fixes.
_platform_disabled: set[str] = set()
try:
from agent.skill_utils import get_disabled_skill_names
_platform_disabled = get_disabled_skill_names(platform="telegram")
except Exception:
pass
# Remaining slots go to built-in skill commands (not hub-installed).
skill_entries: list[tuple[str, str]] = []
try:
from agent.skill_commands import get_skill_commands
from tools.skills_tool import SKILLS_DIR
_skills_dir = str(SKILLS_DIR.resolve())
_hub_dir = str((SKILLS_DIR / ".hub").resolve())
skill_cmds = get_skill_commands()
for cmd_key in sorted(skill_cmds):
info = skill_cmds[cmd_key]
skill_path = info.get("skill_md_path", "")
if not skill_path.startswith(_skills_dir):
continue
if skill_path.startswith(_hub_dir):
continue
# Skip skills disabled for telegram
skill_name = info.get("name", "")
if skill_name in _platform_disabled:
continue
name = cmd_key.lstrip("/").replace("-", "_")
desc = info.get("description", "")
# Keep descriptions short — setMyCommands has an undocumented
# total payload limit. 40 chars fits 100 commands safely.
if len(desc) > 40:
desc = desc[:37] + "..."
skill_entries.append((name, desc))
except Exception:
pass
# Clamp skill names to 32 chars with collision avoidance
skill_entries = _clamp_telegram_names(skill_entries, reserved_names)
# Skills fill remaining slots — they're the only tier that gets trimmed
remaining_slots = max(0, max_commands - len(all_commands))
hidden_count = max(0, len(skill_entries) - remaining_slots)
all_commands.extend(skill_entries[:remaining_slots])
entries, hidden_count = _collect_gateway_skill_entries(
platform="telegram",
max_slots=remaining_slots,
reserved_names=reserved_names,
desc_limit=40,
sanitize_name=_sanitize_telegram_name,
)
# Drop the cmd_key — Telegram only needs (name, desc) pairs.
all_commands.extend((n, d) for n, d, _k in entries)
return all_commands[:max_commands], hidden_count
def discord_skill_commands(
max_slots: int,
reserved_names: set[str],
) -> tuple[list[tuple[str, str, str]], int]:
"""Return skill entries for Discord slash command registration.
Same priority and filtering logic as :func:`telegram_menu_commands`
(plugins > skills, hub excluded, per-platform disabled excluded), but
adapted for Discord's constraints:
- Hyphens are allowed in names (no ``-`` → ``_`` sanitization)
- Descriptions capped at 100 chars (Discord's per-field max)
Args:
max_slots: Available command slots (100 minus existing built-in count).
reserved_names: Names of already-registered built-in commands.
Returns:
``(entries, hidden_count)`` where *entries* is a list of
``(discord_name, description, cmd_key)`` triples. ``cmd_key`` is
the original ``/skill-name`` key needed for the slash handler callback.
"""
return _collect_gateway_skill_entries(
platform="discord",
max_slots=max_slots,
reserved_names=set(reserved_names), # copy — don't mutate caller's set
desc_limit=100,
)
def slack_subcommand_map() -> dict[str, str]:
"""Return subcommand -> /command mapping for Slack /hermes handler.

320
hermes_cli/config.py
View File

@@ -19,6 +19,7 @@ import stat
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, Any, Optional, List, Tuple
@@ -205,6 +206,11 @@ DEFAULT_CONFIG = {
"toolsets": ["hermes-cli"],
"agent": {
"max_turns": 90,
# Inactivity timeout for gateway agent execution (seconds).
# The agent can run indefinitely as long as it's actively calling
# tools or receiving API responses. Only fires when the agent has
# been completely idle for this duration. 0 = unlimited.
"gateway_timeout": 1800,
# Tool-use enforcement: injects system prompt guidance that tells the
# model to actually call tools instead of describing intended actions.
# Values: "auto" (default — applies to gpt/codex models), true/false
@@ -531,6 +537,14 @@ DEFAULT_CONFIG = {
"wrap_response": True,
},
# Logging — controls file logging to ~/.hermes/logs/.
# agent.log captures INFO+ (all agent activity); errors.log captures WARNING+.
"logging": {
"level": "INFO", # Minimum level for agent.log: DEBUG, INFO, WARNING
"max_size_mb": 5, # Max size per log file before rotation
"backup_count": 3, # Number of rotated backup files to keep
},
# Config schema version - bump this when adding new required fields
"_config_version": 12,
}
@@ -576,6 +590,30 @@ OPTIONAL_ENV_VARS = {
"category": "provider",
"advanced": True,
},
"GOOGLE_API_KEY": {
"description": "Google AI Studio API key (also recognized as GEMINI_API_KEY)",
"prompt": "Google AI Studio API key",
"url": "https://aistudio.google.com/app/apikey",
"password": True,
"category": "provider",
"advanced": True,
},
"GEMINI_API_KEY": {
"description": "Google AI Studio API key (alias for GOOGLE_API_KEY)",
"prompt": "Gemini API key",
"url": "https://aistudio.google.com/app/apikey",
"password": True,
"category": "provider",
"advanced": True,
},
"GEMINI_BASE_URL": {
"description": "Google AI Studio base URL override",
"prompt": "Gemini base URL (leave empty for default)",
"url": None,
"password": False,
"category": "provider",
"advanced": True,
},
"GLM_API_KEY": {
"description": "Z.AI / GLM API key (also recognized as ZAI_API_KEY / Z_AI_API_KEY)",
"prompt": "Z.AI / GLM API key",
@@ -830,6 +868,13 @@ OPTIONAL_ENV_VARS = {
"password": True,
"category": "tool",
},
"FIRECRAWL_BROWSER_TTL": {
"description": "Firecrawl browser session TTL in seconds (optional, default 300)",
"prompt": "Browser session TTL (seconds)",
"tools": ["browser_navigate", "browser_click"],
"password": False,
"category": "tool",
},
"CAMOFOX_URL": {
"description": "Camofox browser server URL for local anti-detection browsing (e.g. http://localhost:9377)",
"prompt": "Camofox server URL",
@@ -1226,6 +1271,43 @@ def get_missing_config_fields() -> List[Dict[str, Any]]:
return missing
def get_missing_skill_config_vars() -> List[Dict[str, Any]]:
"""Return skill-declared config vars that are missing or empty in config.yaml.
Scans all enabled skills for ``metadata.hermes.config`` entries, then checks
which ones are absent or empty under ``skills.config.<key>`` in the user's
config.yaml. Returns a list of dicts suitable for prompting.
"""
try:
from agent.skill_utils import discover_all_skill_config_vars, SKILL_CONFIG_PREFIX
except Exception:
return []
all_vars = discover_all_skill_config_vars()
if not all_vars:
return []
config = load_config()
missing: List[Dict[str, Any]] = []
for var in all_vars:
# Skill config is stored under skills.config.<logical_key>
storage_key = f"{SKILL_CONFIG_PREFIX}.{var['key']}"
parts = storage_key.split(".")
current = config
value = None
for part in parts:
if isinstance(current, dict) and part in current:
current = current[part]
value = current
else:
value = None
break
# Missing = key doesn't exist or is empty string
if value is None or (isinstance(value, str) and not value.strip()):
missing.append(var)
return missing
def check_config_version() -> Tuple[int, int]:
"""
Check config version.
@@ -1238,6 +1320,182 @@ def check_config_version() -> Tuple[int, int]:
return current, latest
# =============================================================================
# Config structure validation
# =============================================================================
# Fields that are valid at root level of config.yaml
_KNOWN_ROOT_KEYS = {
"_config_version", "model", "providers", "fallback_model",
"fallback_providers", "credential_pool_strategies", "toolsets",
"agent", "terminal", "display", "compression", "delegation",
"auxiliary", "custom_providers", "memory", "gateway",
}
# Valid fields inside a custom_providers list entry
_VALID_CUSTOM_PROVIDER_FIELDS = {
"name", "base_url", "api_key", "api_mode", "models",
"context_length", "rate_limit_delay",
}
# Fields that look like they should be inside custom_providers, not at root
_CUSTOM_PROVIDER_LIKE_FIELDS = {"base_url", "api_key", "rate_limit_delay", "api_mode"}
@dataclass
class ConfigIssue:
"""A detected config structure problem."""
severity: str # "error", "warning"
message: str
hint: str
def validate_config_structure(config: Optional[Dict[str, Any]] = None) -> List["ConfigIssue"]:
"""Validate config.yaml structure and return a list of detected issues.
Catches common YAML formatting mistakes that produce confusing runtime
errors (like "Unknown provider") instead of clear diagnostics.
Can be called with a pre-loaded config dict, or will load from disk.
"""
if config is None:
try:
config = load_config()
except Exception:
return [ConfigIssue("error", "Could not load config.yaml", "Run 'hermes setup' to create a valid config")]
issues: List[ConfigIssue] = []
# ── custom_providers must be a list, not a dict ──────────────────────
cp = config.get("custom_providers")
if cp is not None:
if isinstance(cp, dict):
issues.append(ConfigIssue(
"error",
"custom_providers is a dict — it must be a YAML list (items prefixed with '-')",
"Change to:\n"
" custom_providers:\n"
" - name: my-provider\n"
" base_url: https://...\n"
" api_key: ...",
))
# Check if dict keys look like they should be list-entry fields
cp_keys = set(cp.keys()) if isinstance(cp, dict) else set()
suspicious = cp_keys & _CUSTOM_PROVIDER_LIKE_FIELDS
if suspicious:
issues.append(ConfigIssue(
"warning",
f"Root-level keys {sorted(suspicious)} look like custom_providers entry fields",
"These should be indented under a '- name: ...' list entry, not at root level",
))
elif isinstance(cp, list):
# Validate each entry in the list
for i, entry in enumerate(cp):
if not isinstance(entry, dict):
issues.append(ConfigIssue(
"warning",
f"custom_providers[{i}] is not a dict (got {type(entry).__name__})",
"Each entry should have at minimum: name, base_url",
))
continue
if not entry.get("name"):
issues.append(ConfigIssue(
"warning",
f"custom_providers[{i}] is missing 'name' field",
"Add a name, e.g.: name: my-provider",
))
if not entry.get("base_url"):
issues.append(ConfigIssue(
"warning",
f"custom_providers[{i}] is missing 'base_url' field",
"Add the API endpoint URL, e.g.: base_url: https://api.example.com/v1",
))
# ── fallback_model must be a top-level dict with provider + model ────
fb = config.get("fallback_model")
if fb is not None:
if not isinstance(fb, dict):
issues.append(ConfigIssue(
"error",
f"fallback_model should be a dict with 'provider' and 'model', got {type(fb).__name__}",
"Change to:\n"
" fallback_model:\n"
" provider: openrouter\n"
" model: anthropic/claude-sonnet-4",
))
elif fb:
if not fb.get("provider"):
issues.append(ConfigIssue(
"warning",
"fallback_model is missing 'provider' field — fallback will be disabled",
"Add: provider: openrouter (or another provider)",
))
if not fb.get("model"):
issues.append(ConfigIssue(
"warning",
"fallback_model is missing 'model' field — fallback will be disabled",
"Add: model: anthropic/claude-sonnet-4 (or another model)",
))
# ── Check for fallback_model accidentally nested inside custom_providers ──
if isinstance(cp, dict) and "fallback_model" not in config and "fallback_model" in (cp or {}):
issues.append(ConfigIssue(
"error",
"fallback_model appears inside custom_providers instead of at root level",
"Move fallback_model to the top level of config.yaml (no indentation)",
))
# ── model section: should exist when custom_providers is configured ──
model_cfg = config.get("model")
if cp and not model_cfg:
issues.append(ConfigIssue(
"warning",
"custom_providers defined but no 'model' section — Hermes won't know which provider to use",
"Add a model section:\n"
" model:\n"
" provider: custom\n"
" default: your-model-name\n"
" base_url: https://...",
))
# ── Root-level keys that look misplaced ──────────────────────────────
for key in config:
if key.startswith("_"):
continue
if key not in _KNOWN_ROOT_KEYS and key in _CUSTOM_PROVIDER_LIKE_FIELDS:
issues.append(ConfigIssue(
"warning",
f"Root-level key '{key}' looks misplaced — should it be under 'model:' or inside a 'custom_providers' entry?",
f"Move '{key}' under the appropriate section",
))
return issues
def print_config_warnings(config: Optional[Dict[str, Any]] = None) -> None:
"""Print config structure warnings to stderr at startup.
Called early in CLI and gateway init so users see problems before
they hit cryptic "Unknown provider" errors. Prints nothing if
config is healthy.
"""
try:
issues = validate_config_structure(config)
except Exception:
return
if not issues:
return
import sys
lines = ["\033[33m⚠ Config issues detected in config.yaml:\033[0m"]
for ci in issues:
marker = "\033[31m✗\033[0m" if ci.severity == "error" else "\033[33m⚠\033[0m"
lines.append(f" {marker} {ci.message}")
lines.append(" \033[2mRun 'hermes doctor' for fix suggestions.\033[0m")
sys.stderr.write("\n".join(lines) + "\n\n")
def migrate_config(interactive: bool = True, quiet: bool = False) -> Dict[str, Any]:
"""
Migrate config to latest version, prompting for new required fields.
@@ -1481,7 +1739,50 @@ def migrate_config(interactive: bool = True, quiet: bool = False) -> Dict[str, A
config = load_config()
config["_config_version"] = latest_ver
save_config(config)
# ── Skill-declared config vars ──────────────────────────────────────
# Skills can declare config.yaml settings they need via
# metadata.hermes.config in their SKILL.md frontmatter.
# Prompt for any that are missing/empty.
missing_skill_config = get_missing_skill_config_vars()
if missing_skill_config and interactive and not quiet:
print(f"\n {len(missing_skill_config)} skill setting(s) not configured:")
for var in missing_skill_config:
skill_name = var.get("skill", "unknown")
print(f" • {var['key']} — {var['description']} (from skill: {skill_name})")
print()
try:
answer = input(" Configure skill settings? [y/N]: ").strip().lower()
except (EOFError, KeyboardInterrupt):
answer = "n"
if answer in ("y", "yes"):
print()
config = load_config()
try:
from agent.skill_utils import SKILL_CONFIG_PREFIX
except Exception:
SKILL_CONFIG_PREFIX = "skills.config"
for var in missing_skill_config:
default = var.get("default", "")
default_hint = f" (default: {default})" if default else ""
value = input(f" {var['prompt']}{default_hint}: ").strip()
if not value and default:
value = str(default)
if value:
storage_key = f"{SKILL_CONFIG_PREFIX}.{var['key']}"
_set_nested(config, storage_key, value)
results["config_added"].append(var["key"])
print(f" ✓ Saved {var['key']} = {value}")
else:
results["warnings"].append(
f"Skipped {var['key']} — skill '{var.get('skill', '?')}' may ask for it later"
)
print()
save_config(config)
else:
print(" Set later with: hermes config set <key> <value>")
return results
@@ -2135,6 +2436,23 @@ def show_config():
print(f" Telegram: {'configured' if telegram_token else color('not configured', Colors.DIM)}")
print(f" Discord: {'configured' if discord_token else color('not configured', Colors.DIM)}")
# Skill config
try:
from agent.skill_utils import discover_all_skill_config_vars, resolve_skill_config_values
skill_vars = discover_all_skill_config_vars()
if skill_vars:
resolved = resolve_skill_config_values(skill_vars)
print()
print(color("◆ Skill Settings", Colors.CYAN, Colors.BOLD))
for var in skill_vars:
key = var["key"]
value = resolved.get(key, "")
skill_name = var.get("skill", "")
display_val = str(value) if value else color("(not set)", Colors.DIM)
print(f" {key:<20s} {display_val} {color(f'[{skill_name}]', Colors.DIM)}")
except Exception:
pass
print()
print(color("─" * 60, Colors.DIM))
print(color(" hermes config edit # Edit config file", Colors.DIM))

19
hermes_cli/doctor.py
View File

@@ -318,6 +318,25 @@ def run_doctor(args):
except Exception:
pass
# Validate config structure (catches malformed custom_providers, etc.)
try:
from hermes_cli.config import validate_config_structure
config_issues = validate_config_structure()
if config_issues:
print()
print(color("◆ Config Structure", Colors.CYAN, Colors.BOLD))
for ci in config_issues:
if ci.severity == "error":
check_fail(ci.message)
else:
check_warn(ci.message)
# Show the hint indented
for hint_line in ci.hint.splitlines():
check_info(hint_line)
issues.append(ci.message)
except Exception:
pass
# =========================================================================
# Check: Auth providers
# =========================================================================

218
hermes_cli/gateway.py
View File

@@ -28,9 +28,78 @@ from hermes_cli.colors import Colors, color
# Process Management (for manual gateway runs)
# =============================================================================
def find_gateway_pids() -> list:
"""Find PIDs of running gateway processes."""
def _get_service_pids() -> set:
"""Return PIDs currently managed by systemd or launchd gateway services.
Used to avoid killing freshly-restarted service processes when sweeping
for stale manual gateway processes after a service restart. Relies on the
service manager having committed the new PID before the restart command
returns (true for both systemd and launchd in practice).
"""
pids: set = set()
# --- systemd (Linux): user and system scopes ---
if is_linux():
for scope_args in [["systemctl", "--user"], ["systemctl"]]:
try:
result = subprocess.run(
scope_args + ["list-units", "hermes-gateway*",
"--plain", "--no-legend", "--no-pager"],
capture_output=True, text=True, timeout=5,
)
for line in result.stdout.strip().splitlines():
parts = line.split()
if not parts or not parts[0].endswith(".service"):
continue
svc = parts[0]
try:
show = subprocess.run(
scope_args + ["show", svc,
"--property=MainPID", "--value"],
capture_output=True, text=True, timeout=5,
)
pid = int(show.stdout.strip())
if pid > 0:
pids.add(pid)
except (ValueError, subprocess.TimeoutExpired):
pass
except (FileNotFoundError, subprocess.TimeoutExpired):
pass
# --- launchd (macOS) ---
if is_macos():
try:
label = get_launchd_label()
result = subprocess.run(
["launchctl", "list", label],
capture_output=True, text=True, timeout=5,
)
if result.returncode == 0:
# Output: "PID\tStatus\tLabel" header, then one data line
for line in result.stdout.strip().splitlines():
parts = line.split()
if len(parts) >= 3 and parts[2] == label:
try:
pid = int(parts[0])
if pid > 0:
pids.add(pid)
except ValueError:
pass
except (FileNotFoundError, subprocess.TimeoutExpired):
pass
return pids
def find_gateway_pids(exclude_pids: set | None = None) -> list:
"""Find PIDs of running gateway processes.
Args:
exclude_pids: PIDs to exclude from the result (e.g. service-managed
PIDs that should not be killed during a stale-process sweep).
"""
pids = []
_exclude = exclude_pids or set()
patterns = [
"hermes_cli.main gateway",
"hermes_cli/main.py gateway",
@@ -43,7 +112,7 @@ def find_gateway_pids() -> list:
# Windows: use wmic to search command lines
result = subprocess.run(
["wmic", "process", "get", "ProcessId,CommandLine", "/FORMAT:LIST"],
capture_output=True, text=True
capture_output=True, text=True, timeout=10
)
# Parse WMIC LIST output: blocks of "CommandLine=...\nProcessId=...\n"
current_cmd = ""
@@ -56,7 +125,7 @@ def find_gateway_pids() -> list:
if any(p in current_cmd for p in patterns):
try:
pid = int(pid_str)
if pid != os.getpid() and pid not in pids:
if pid != os.getpid() and pid not in pids and pid not in _exclude:
pids.append(pid)
except ValueError:
pass
@@ -65,7 +134,8 @@ def find_gateway_pids() -> list:
result = subprocess.run(
["ps", "aux"],
capture_output=True,
text=True
text=True,
timeout=10,
)
for line in result.stdout.split('\n'):
# Skip grep and current process
@@ -77,7 +147,7 @@ def find_gateway_pids() -> list:
if len(parts) > 1:
try:
pid = int(parts[1])
if pid not in pids:
if pid not in pids and pid not in _exclude:
pids.append(pid)
except ValueError:
continue
@@ -88,9 +158,15 @@ def find_gateway_pids() -> list:
return pids
def kill_gateway_processes(force: bool = False) -> int:
"""Kill ALL running gateway processes (across all profiles). Returns count killed."""
pids = find_gateway_pids()
def kill_gateway_processes(force: bool = False, exclude_pids: set | None = None) -> int:
"""Kill any running gateway processes. Returns count killed.
Args:
force: Use SIGKILL instead of SIGTERM.
exclude_pids: PIDs to skip (e.g. service-managed PIDs that were just
restarted and should not be killed).
"""
pids = find_gateway_pids(exclude_pids=exclude_pids)
killed = 0
for pid in pids:
@@ -402,6 +478,7 @@ def get_systemd_linger_status() -> tuple[bool | None, str]:
capture_output=True,
text=True,
check=False,
timeout=10,
)
except Exception as e:
return None, str(e)
@@ -636,7 +713,7 @@ def refresh_systemd_unit_if_needed(system: bool = False) -> bool:
expected_user = _read_systemd_user_from_unit(unit_path) if system else None
unit_path.write_text(generate_systemd_unit(system=system, run_as_user=expected_user), encoding="utf-8")
subprocess.run(_systemctl_cmd(system) + ["daemon-reload"], check=True)
subprocess.run(_systemctl_cmd(system) + ["daemon-reload"], check=True, timeout=30)
print(f"↻ Updated gateway {_service_scope_label(system)} service definition to match the current Hermes install")
return True
@@ -687,6 +764,7 @@ def _ensure_linger_enabled() -> None:
capture_output=True,
text=True,
check=False,
timeout=30,
)
except Exception as e:
_print_linger_enable_warning(username, str(e))
@@ -717,7 +795,7 @@ def systemd_install(force: bool = False, system: bool = False, run_as_user: str
if not systemd_unit_is_current(system=system):
print(f"↻ Repairing outdated {_service_scope_label(system)} systemd service at: {unit_path}")
refresh_systemd_unit_if_needed(system=system)
subprocess.run(_systemctl_cmd(system) + ["enable", get_service_name()], check=True)
subprocess.run(_systemctl_cmd(system) + ["enable", get_service_name()], check=True, timeout=30)
print(f"✓ {_service_scope_label(system).capitalize()} service definition updated")
return
print(f"Service already installed at: {unit_path}")
@@ -728,8 +806,8 @@ def systemd_install(force: bool = False, system: bool = False, run_as_user: str
print(f"Installing {_service_scope_label(system)} systemd service to: {unit_path}")
unit_path.write_text(generate_systemd_unit(system=system, run_as_user=run_as_user), encoding="utf-8")
subprocess.run(_systemctl_cmd(system) + ["daemon-reload"], check=True)
subprocess.run(_systemctl_cmd(system) + ["enable", get_service_name()], check=True)
subprocess.run(_systemctl_cmd(system) + ["daemon-reload"], check=True, timeout=30)
subprocess.run(_systemctl_cmd(system) + ["enable", get_service_name()], check=True, timeout=30)
print()
print(f"✓ {_service_scope_label(system).capitalize()} service installed and enabled!")
@@ -755,15 +833,15 @@ def systemd_uninstall(system: bool = False):
if system:
_require_root_for_system_service("uninstall")
subprocess.run(_systemctl_cmd(system) + ["stop", get_service_name()], check=False)
subprocess.run(_systemctl_cmd(system) + ["disable", get_service_name()], check=False)
subprocess.run(_systemctl_cmd(system) + ["stop", get_service_name()], check=False, timeout=90)
subprocess.run(_systemctl_cmd(system) + ["disable", get_service_name()], check=False, timeout=30)
unit_path = get_systemd_unit_path(system=system)
if unit_path.exists():
unit_path.unlink()
print(f"✓ Removed {unit_path}")
subprocess.run(_systemctl_cmd(system) + ["daemon-reload"], check=True)
subprocess.run(_systemctl_cmd(system) + ["daemon-reload"], check=True, timeout=30)
print(f"✓ {_service_scope_label(system).capitalize()} service uninstalled")
@@ -772,7 +850,7 @@ def systemd_start(system: bool = False):
if system:
_require_root_for_system_service("start")
refresh_systemd_unit_if_needed(system=system)
subprocess.run(_systemctl_cmd(system) + ["start", get_service_name()], check=True)
subprocess.run(_systemctl_cmd(system) + ["start", get_service_name()], check=True, timeout=30)
print(f"✓ {_service_scope_label(system).capitalize()} service started")
@@ -781,7 +859,7 @@ def systemd_stop(system: bool = False):
system = _select_systemd_scope(system)
if system:
_require_root_for_system_service("stop")
subprocess.run(_systemctl_cmd(system) + ["stop", get_service_name()], check=True)
subprocess.run(_systemctl_cmd(system) + ["stop", get_service_name()], check=True, timeout=90)
print(f"✓ {_service_scope_label(system).capitalize()} service stopped")
@@ -791,7 +869,7 @@ def systemd_restart(system: bool = False):
if system:
_require_root_for_system_service("restart")
refresh_systemd_unit_if_needed(system=system)
subprocess.run(_systemctl_cmd(system) + ["restart", get_service_name()], check=True)
subprocess.run(_systemctl_cmd(system) + ["restart", get_service_name()], check=True, timeout=90)
print(f"✓ {_service_scope_label(system).capitalize()} service restarted")
@@ -818,12 +896,14 @@ def systemd_status(deep: bool = False, system: bool = False):
subprocess.run(
_systemctl_cmd(system) + ["status", get_service_name(), "--no-pager"],
capture_output=False,
timeout=10,
)
result = subprocess.run(
_systemctl_cmd(system) + ["is-active", get_service_name()],
capture_output=True,
text=True,
timeout=10,
)
status = result.stdout.strip()
@@ -860,7 +940,7 @@ def systemd_status(deep: bool = False, system: bool = False):
if deep:
print()
print("Recent logs:")
subprocess.run(_journalctl_cmd(system) + ["-u", get_service_name(), "-n", "20", "--no-pager"])
subprocess.run(_journalctl_cmd(system) + ["-u", get_service_name(), "-n", "20", "--no-pager"], timeout=10)
# =============================================================================
@@ -979,8 +1059,8 @@ def refresh_launchd_plist_if_needed() -> bool:
plist_path.write_text(generate_launchd_plist(), encoding="utf-8")
label = get_launchd_label()
# Bootout/bootstrap so launchd picks up the new definition
subprocess.run(["launchctl", "bootout", f"{_launchd_domain()}/{label}"], check=False)
subprocess.run(["launchctl", "bootstrap", _launchd_domain(), str(plist_path)], check=False)
subprocess.run(["launchctl", "bootout", f"{_launchd_domain()}/{label}"], check=False, timeout=90)
subprocess.run(["launchctl", "bootstrap", _launchd_domain(), str(plist_path)], check=False, timeout=30)
print("↻ Updated gateway launchd service definition to match the current Hermes install")
return True
@@ -1002,7 +1082,7 @@ def launchd_install(force: bool = False):
print(f"Installing launchd service to: {plist_path}")
plist_path.write_text(generate_launchd_plist())
subprocess.run(["launchctl", "bootstrap", _launchd_domain(), str(plist_path)], check=True)
subprocess.run(["launchctl", "bootstrap", _launchd_domain(), str(plist_path)], check=True, timeout=30)
print()
print("✓ Service installed and loaded!")
@@ -1015,7 +1095,7 @@ def launchd_install(force: bool = False):
def launchd_uninstall():
plist_path = get_launchd_plist_path()
label = get_launchd_label()
subprocess.run(["launchctl", "bootout", f"{_launchd_domain()}/{label}"], check=False)
subprocess.run(["launchctl", "bootout", f"{_launchd_domain()}/{label}"], check=False, timeout=90)
if plist_path.exists():
plist_path.unlink()
@@ -1032,25 +1112,25 @@ def launchd_start():
print("↻ launchd plist missing; regenerating service definition")
plist_path.parent.mkdir(parents=True, exist_ok=True)
plist_path.write_text(generate_launchd_plist(), encoding="utf-8")
subprocess.run(["launchctl", "bootstrap", _launchd_domain(), str(plist_path)], check=True)
subprocess.run(["launchctl", "kickstart", f"{_launchd_domain()}/{label}"], check=True)
subprocess.run(["launchctl", "bootstrap", _launchd_domain(), str(plist_path)], check=True, timeout=30)
subprocess.run(["launchctl", "kickstart", f"{_launchd_domain()}/{label}"], check=True, timeout=30)
print("✓ Service started")
return
refresh_launchd_plist_if_needed()
try:
subprocess.run(["launchctl", "kickstart", f"{_launchd_domain()}/{label}"], check=True)
subprocess.run(["launchctl", "kickstart", f"{_launchd_domain()}/{label}"], check=True, timeout=30)
except subprocess.CalledProcessError as e:
if e.returncode != 3:
if e.returncode not in (3, 113):
raise
print("↻ launchd job was unloaded; reloading service definition")
subprocess.run(["launchctl", "bootstrap", _launchd_domain(), str(plist_path)], check=True)
subprocess.run(["launchctl", "kickstart", f"{_launchd_domain()}/{label}"], check=True)
subprocess.run(["launchctl", "bootstrap", _launchd_domain(), str(plist_path)], check=True, timeout=30)
subprocess.run(["launchctl", "kickstart", f"{_launchd_domain()}/{label}"], check=True, timeout=30)
print("✓ Service started")
def launchd_stop():
label = get_launchd_label()
subprocess.run(["launchctl", "kill", "SIGTERM", f"{_launchd_domain()}/{label}"], check=True)
subprocess.run(["launchctl", "kill", "SIGTERM", f"{_launchd_domain()}/{label}"], check=True, timeout=30)
print("✓ Service stopped")
def _wait_for_gateway_exit(timeout: float = 10.0, force_after: float = 5.0):
@@ -1100,26 +1180,33 @@ def launchd_restart():
# A two-step stop/start from inside the gateway's own process tree
# would kill the shell before the start command is reached.
try:
subprocess.run(["launchctl", "kickstart", "-k", target], check=True)
subprocess.run(["launchctl", "kickstart", "-k", target], check=True, timeout=90)
print("✓ Service restarted")
except subprocess.CalledProcessError as e:
if e.returncode != 3:
if e.returncode not in (3, 113):
raise
# Job not loaded — bootstrap and start fresh
print("↻ launchd job was unloaded; reloading")
plist_path = get_launchd_plist_path()
subprocess.run(["launchctl", "bootstrap", _launchd_domain(), str(plist_path)], check=True)
subprocess.run(["launchctl", "kickstart", target], check=True)
subprocess.run(["launchctl", "bootstrap", _launchd_domain(), str(plist_path)], check=True, timeout=30)
subprocess.run(["launchctl", "kickstart", target], check=True, timeout=30)
print("✓ Service restarted")
def launchd_status(deep: bool = False):
plist_path = get_launchd_plist_path()
label = get_launchd_label()
result = subprocess.run(
["launchctl", "list", label],
capture_output=True,
text=True
)
try:
result = subprocess.run(
["launchctl", "list", label],
capture_output=True,
text=True,
timeout=10,
)
loaded = result.returncode == 0
loaded_output = result.stdout
except subprocess.TimeoutExpired:
loaded = False
loaded_output = ""
print(f"Launchd plist: {plist_path}")
if launchd_plist_is_current():
@@ -1127,10 +1214,10 @@ def launchd_status(deep: bool = False):
else:
print("⚠ Service definition is stale relative to the current Hermes install")
print(" Run: hermes gateway start")
if result.returncode == 0:
if loaded:
print("✓ Gateway service is loaded")
print(result.stdout)
print(loaded_output)
else:
print("✗ Gateway service is not loaded")
print(" Service definition exists locally but launchd has not loaded it.")
@@ -1141,7 +1228,7 @@ def launchd_status(deep: bool = False):
if log_file.exists():
print()
print("Recent logs:")
subprocess.run(["tail", "-20", str(log_file)])
subprocess.run(["tail", "-20", str(log_file)], timeout=10)
# =============================================================================
@@ -1658,28 +1745,37 @@ def _is_service_running() -> bool:
system_unit_exists = get_systemd_unit_path(system=True).exists()
if user_unit_exists:
result = subprocess.run(
_systemctl_cmd(False) + ["is-active", get_service_name()],
capture_output=True, text=True
)
if result.stdout.strip() == "active":
return True
try:
result = subprocess.run(
_systemctl_cmd(False) + ["is-active", get_service_name()],
capture_output=True, text=True, timeout=10,
)
if result.stdout.strip() == "active":
return True
except subprocess.TimeoutExpired:
pass
if system_unit_exists:
result = subprocess.run(
_systemctl_cmd(True) + ["is-active", get_service_name()],
capture_output=True, text=True
)
if result.stdout.strip() == "active":
return True
try:
result = subprocess.run(
_systemctl_cmd(True) + ["is-active", get_service_name()],
capture_output=True, text=True, timeout=10,
)
if result.stdout.strip() == "active":
return True
except subprocess.TimeoutExpired:
pass
return False
elif is_macos() and get_launchd_plist_path().exists():
result = subprocess.run(
["launchctl", "list", get_launchd_label()],
capture_output=True, text=True
)
return result.returncode == 0
try:
result = subprocess.run(
["launchctl", "list", get_launchd_label()],
capture_output=True, text=True, timeout=10,
)
return result.returncode == 0
except subprocess.TimeoutExpired:
return False
# Check for manual processes
return len(find_gateway_pids()) > 0

336
hermes_cli/logs.py Normal file
View File

@@ -0,0 +1,336 @@
"""``hermes logs`` — view and filter Hermes log files.
Supports tailing, following, session filtering, level filtering, and
relative time ranges. All log files live under ``~/.hermes/logs/``.
Usage examples::
hermes logs # last 50 lines of agent.log
hermes logs -f # follow agent.log in real time
hermes logs errors # last 50 lines of errors.log
hermes logs gateway -n 100 # last 100 lines of gateway.log
hermes logs --level WARNING # only WARNING+ lines
hermes logs --session abc123 # filter by session ID substring
hermes logs --since 1h # lines from the last hour
hermes logs --since 30m -f # follow, starting 30 min ago
"""
import os
import re
import sys
import time
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional
from hermes_constants import get_hermes_home, display_hermes_home
# Known log files (name → filename)
LOG_FILES = {
"agent": "agent.log",
"errors": "errors.log",
"gateway": "gateway.log",
}
# Log line timestamp regex — matches "2026-04-05 22:35:00,123" or
# "2026-04-05 22:35:00" at the start of a line.
_TS_RE = re.compile(r"^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})")
# Level extraction — matches " INFO ", " WARNING ", " ERROR ", " DEBUG ", " CRITICAL "
_LEVEL_RE = re.compile(r"\s(DEBUG|INFO|WARNING|ERROR|CRITICAL)\s")
# Level ordering for >= filtering
_LEVEL_ORDER = {"DEBUG": 0, "INFO": 1, "WARNING": 2, "ERROR": 3, "CRITICAL": 4}
def _parse_since(since_str: str) -> Optional[datetime]:
"""Parse a relative time string like '1h', '30m', '2d' into a datetime cutoff.
Returns None if the string can't be parsed.
"""
since_str = since_str.strip().lower()
match = re.match(r"^(\d+)\s*([smhd])$", since_str)
if not match:
return None
value = int(match.group(1))
unit = match.group(2)
delta = {
"s": timedelta(seconds=value),
"m": timedelta(minutes=value),
"h": timedelta(hours=value),
"d": timedelta(days=value),
}[unit]
return datetime.now() - delta
def _parse_line_timestamp(line: str) -> Optional[datetime]:
"""Extract timestamp from a log line. Returns None if not parseable."""
m = _TS_RE.match(line)
if not m:
return None
try:
return datetime.strptime(m.group(1), "%Y-%m-%d %H:%M:%S")
except ValueError:
return None
def _extract_level(line: str) -> Optional[str]:
"""Extract the log level from a line."""
m = _LEVEL_RE.search(line)
return m.group(1) if m else None
def _matches_filters(
line: str,
*,
min_level: Optional[str] = None,
session_filter: Optional[str] = None,
since: Optional[datetime] = None,
) -> bool:
"""Check if a log line passes all active filters."""
if since is not None:
ts = _parse_line_timestamp(line)
if ts is not None and ts < since:
return False
if min_level is not None:
level = _extract_level(line)
if level is not None:
if _LEVEL_ORDER.get(level, 0) < _LEVEL_ORDER.get(min_level, 0):
return False
if session_filter is not None:
if session_filter not in line:
return False
return True
def tail_log(
log_name: str = "agent",
*,
num_lines: int = 50,
follow: bool = False,
level: Optional[str] = None,
session: Optional[str] = None,
since: Optional[str] = None,
) -> None:
"""Read and display log lines, optionally following in real time.
Parameters
----------
log_name
Which log to read: ``"agent"``, ``"errors"``, ``"gateway"``.
num_lines
Number of recent lines to show (before follow starts).
follow
If True, keep watching for new lines (Ctrl+C to stop).
level
Minimum log level to show (e.g. ``"WARNING"``).
session
Session ID substring to filter on.
since
Relative time string (e.g. ``"1h"``, ``"30m"``).
"""
filename = LOG_FILES.get(log_name)
if filename is None:
print(f"Unknown log: {log_name!r}. Available: {', '.join(sorted(LOG_FILES))}")
sys.exit(1)
log_path = get_hermes_home() / "logs" / filename
if not log_path.exists():
print(f"Log file not found: {log_path}")
print(f"(Logs are created when Hermes runs — try 'hermes chat' first)")
sys.exit(1)
# Parse --since into a datetime cutoff
since_dt = None
if since:
since_dt = _parse_since(since)
if since_dt is None:
print(f"Invalid --since value: {since!r}. Use format like '1h', '30m', '2d'.")
sys.exit(1)
min_level = level.upper() if level else None
if min_level and min_level not in _LEVEL_ORDER:
print(f"Invalid --level: {level!r}. Use DEBUG, INFO, WARNING, ERROR, or CRITICAL.")
sys.exit(1)
has_filters = min_level is not None or session is not None or since_dt is not None
# Read and display the tail
try:
lines = _read_tail(log_path, num_lines, has_filters=has_filters,
min_level=min_level, session_filter=session,
since=since_dt)
except PermissionError:
print(f"Permission denied: {log_path}")
sys.exit(1)
# Print header
filter_parts = []
if min_level:
filter_parts.append(f"level>={min_level}")
if session:
filter_parts.append(f"session={session}")
if since:
filter_parts.append(f"since={since}")
filter_desc = f" [{', '.join(filter_parts)}]" if filter_parts else ""
if follow:
print(f"--- {display_hermes_home()}/logs/{filename}{filter_desc} (Ctrl+C to stop) ---")
else:
print(f"--- {display_hermes_home()}/logs/{filename}{filter_desc} (last {num_lines}) ---")
for line in lines:
print(line, end="")
if not follow:
return
# Follow mode — poll for new content
try:
_follow_log(log_path, min_level=min_level, session_filter=session,
since=since_dt)
except KeyboardInterrupt:
print("\n--- stopped ---")
def _read_tail(
path: Path,
num_lines: int,
*,
has_filters: bool = False,
min_level: Optional[str] = None,
session_filter: Optional[str] = None,
since: Optional[datetime] = None,
) -> list:
"""Read the last *num_lines* matching lines from a log file.
When filters are active, we read more raw lines to find enough matches.
"""
if has_filters:
# Read more lines to ensure we get enough after filtering.
# For large files, read last 10K lines and filter down.
raw_lines = _read_last_n_lines(path, max(num_lines * 20, 2000))
filtered = [
l for l in raw_lines
if _matches_filters(l, min_level=min_level,
session_filter=session_filter, since=since)
]
return filtered[-num_lines:]
else:
return _read_last_n_lines(path, num_lines)
def _read_last_n_lines(path: Path, n: int) -> list:
"""Efficiently read the last N lines from a file.
For files under 1MB, reads the whole file (fast, simple).
For larger files, reads chunks from the end.
"""
try:
size = path.stat().st_size
if size == 0:
return []
# For files up to 1MB, just read the whole thing — simple and correct.
if size <= 1_048_576:
with open(path, "r", encoding="utf-8", errors="replace") as f:
all_lines = f.readlines()
return all_lines[-n:]
# For large files, read chunks from the end.
with open(path, "rb") as f:
chunk_size = 8192
lines = []
pos = size
while pos > 0 and len(lines) <= n + 1:
read_size = min(chunk_size, pos)
pos -= read_size
f.seek(pos)
chunk = f.read(read_size)
chunk_lines = chunk.split(b"\n")
if lines:
# Merge the last partial line of the new chunk with the
# first partial line of what we already have.
lines[0] = chunk_lines[-1] + lines[0]
lines = chunk_lines[:-1] + lines
else:
lines = chunk_lines
chunk_size = min(chunk_size * 2, 65536)
# Decode and return last N non-empty lines.
decoded = []
for raw in lines:
if not raw.strip():
continue
try:
decoded.append(raw.decode("utf-8", errors="replace") + "\n")
except Exception:
decoded.append(raw.decode("latin-1") + "\n")
return decoded[-n:]
except Exception:
# Fallback: read entire file
with open(path, "r", encoding="utf-8", errors="replace") as f:
all_lines = f.readlines()
return all_lines[-n:]
def _follow_log(
path: Path,
*,
min_level: Optional[str] = None,
session_filter: Optional[str] = None,
since: Optional[datetime] = None,
) -> None:
"""Poll a log file for new content and print matching lines."""
with open(path, "r", encoding="utf-8", errors="replace") as f:
# Seek to end
f.seek(0, 2)
while True:
line = f.readline()
if line:
if _matches_filters(line, min_level=min_level,
session_filter=session_filter, since=since):
print(line, end="")
sys.stdout.flush()
else:
time.sleep(0.3)
def list_logs() -> None:
"""Print available log files with sizes."""
log_dir = get_hermes_home() / "logs"
if not log_dir.exists():
print(f"No logs directory at {display_hermes_home()}/logs/")
return
print(f"Log files in {display_hermes_home()}/logs/:\n")
found = False
for entry in sorted(log_dir.iterdir()):
if entry.is_file() and entry.suffix == ".log":
size = entry.stat().st_size
mtime = datetime.fromtimestamp(entry.stat().st_mtime)
if size < 1024:
size_str = f"{size}B"
elif size < 1024 * 1024:
size_str = f"{size / 1024:.1f}KB"
else:
size_str = f"{size / (1024 * 1024):.1f}MB"
age = datetime.now() - mtime
if age.total_seconds() < 60:
age_str = "just now"
elif age.total_seconds() < 3600:
age_str = f"{int(age.total_seconds() / 60)}m ago"
elif age.total_seconds() < 86400:
age_str = f"{int(age.total_seconds() / 3600)}h ago"
else:
age_str = mtime.strftime("%Y-%m-%d")
print(f" {entry.name:<25} {size_str:>8} {age_str}")
found = True
if not found:
print(" (no log files yet — run 'hermes chat' to generate logs)")

275
hermes_cli/main.py
View File

@@ -142,6 +142,13 @@ from hermes_cli.config import get_hermes_home
from hermes_cli.env_loader import load_hermes_dotenv
load_hermes_dotenv(project_env=PROJECT_ROOT / '.env')
# Initialize centralized file logging early — all `hermes` subcommands
# (chat, setup, gateway, config, etc.) write to agent.log + errors.log.
try:
from hermes_logging import setup_logging as _setup_logging
_setup_logging(mode="cli")
except Exception:
pass # best-effort — don't crash the CLI if logging setup fails
import logging
import time as _time
@@ -916,7 +923,7 @@ def select_provider_and_model(args=None):
try:
active = resolve_provider("auto")
except AuthError:
active = "openrouter" # no provider yet; show full picker
active = None # no provider yet; default to first in list
# Detect custom endpoint
if active == "openrouter" and get_env_value("OPENAI_BASE_URL"):
@@ -929,6 +936,7 @@ def select_provider_and_model(args=None):
"copilot-acp": "GitHub Copilot ACP",
"copilot": "GitHub Copilot",
"anthropic": "Anthropic",
"gemini": "Google AI Studio",
"zai": "Z.AI / GLM",
"kimi-coding": "Kimi / Moonshot",
"minimax": "MiniMax",
@@ -941,21 +949,26 @@ def select_provider_and_model(args=None):
"huggingface": "Hugging Face",
"custom": "Custom endpoint",
}
active_label = provider_labels.get(active, active)
active_label = provider_labels.get(active, active) if active else "none"
print()
print(f" Current model: {current_model}")
print(f" Active provider: {active_label}")
print()
# Step 1: Provider selection — put active provider first with marker
providers = [
("openrouter", "OpenRouter (100+ models, pay-per-use)"),
# Step 1: Provider selection — top providers shown first, rest behind "More..."
top_providers = [
("nous", "Nous Portal (Nous Research subscription)"),
("openai-codex", "OpenAI Codex"),
("copilot-acp", "GitHub Copilot ACP (spawns `copilot --acp --stdio`)"),
("copilot", "GitHub Copilot (uses GITHUB_TOKEN or gh auth token)"),
("openrouter", "OpenRouter (100+ models, pay-per-use)"),
("anthropic", "Anthropic (Claude models — API key or Claude Code)"),
("openai-codex", "OpenAI Codex"),
("copilot", "GitHub Copilot (uses GITHUB_TOKEN or gh auth token)"),
("huggingface", "Hugging Face Inference Providers (20+ open models)"),
]
extended_providers = [
("copilot-acp", "GitHub Copilot ACP (spawns `copilot --acp --stdio`)"),
("gemini", "Google AI Studio (Gemini models — OpenAI-compatible endpoint)"),
("zai", "Z.AI / GLM (Zhipu AI direct API)"),
("kimi-coding", "Kimi / Moonshot (Moonshot AI direct API)"),
("minimax", "MiniMax (global direct API)"),
@@ -965,7 +978,6 @@ def select_provider_and_model(args=None):
("opencode-go", "OpenCode Go (open models, $10/month subscription)"),
("ai-gateway", "AI Gateway (Vercel — 200+ models, pay-per-use)"),
("alibaba", "Alibaba Cloud / DashScope Coding (Qwen + multi-provider)"),
("huggingface", "Hugging Face Inference Providers (20+ open models)"),
]
# Add user-defined custom providers from config.yaml
@@ -979,12 +991,11 @@ def select_provider_and_model(args=None):
base_url = (entry.get("base_url") or "").strip()
if not name or not base_url:
continue
# Generate a stable key from the name
key = "custom:" + name.lower().replace(" ", "-")
short_url = base_url.replace("https://", "").replace("http://", "").rstrip("/")
saved_model = entry.get("model", "")
model_hint = f" — {saved_model}" if saved_model else ""
providers.append((key, f"{name} ({short_url}){model_hint}"))
top_providers.append((key, f"{name} ({short_url}){model_hint}"))
_custom_provider_map[key] = {
"name": name,
"base_url": base_url,
@@ -992,31 +1003,54 @@ def select_provider_and_model(args=None):
"model": saved_model,
}
# Always add the manual custom endpoint option last
providers.append(("custom", "Custom endpoint (enter URL manually)"))
top_keys = {k for k, _ in top_providers}
extended_keys = {k for k, _ in extended_providers}
# Add removal option if there are saved custom providers
if _custom_provider_map:
providers.append(("remove-custom", "Remove a saved custom provider"))
# If the active provider is in the extended list, promote it into top
if active and active in extended_keys:
promoted = [(k, l) for k, l in extended_providers if k == active]
extended_providers = [(k, l) for k, l in extended_providers if k != active]
top_providers = promoted + top_providers
top_keys.add(active)
# Reorder so the active provider is at the top
known_keys = {k for k, _ in providers}
active_key = active if active in known_keys else "custom"
# Build the primary menu
ordered = []
for key, label in providers:
if key == active_key:
ordered.insert(0, (key, f"{label} ← currently active"))
default_idx = 0
for key, label in top_providers:
if active and key == active:
ordered.append((key, f"{label} ← currently active"))
default_idx = len(ordered) - 1
else:
ordered.append((key, label))
ordered.append(("more", "More providers..."))
ordered.append(("cancel", "Cancel"))
provider_idx = _prompt_provider_choice([label for _, label in ordered])
provider_idx = _prompt_provider_choice(
[label for _, label in ordered], default=default_idx,
)
if provider_idx is None or ordered[provider_idx][0] == "cancel":
print("No change.")
return
selected_provider = ordered[provider_idx][0]
# "More providers..." — show the extended list
if selected_provider == "more":
ext_ordered = list(extended_providers)
ext_ordered.append(("custom", "Custom endpoint (enter URL manually)"))
if _custom_provider_map:
ext_ordered.append(("remove-custom", "Remove a saved custom provider"))
ext_ordered.append(("cancel", "Cancel"))
ext_idx = _prompt_provider_choice(
[label for _, label in ext_ordered], default=0,
)
if ext_idx is None or ext_ordered[ext_idx][0] == "cancel":
print("No change.")
return
selected_provider = ext_ordered[ext_idx][0]
# Step 2: Provider-specific setup + model selection
if selected_provider == "openrouter":
_model_flow_openrouter(config, current_model)
@@ -1038,38 +1072,37 @@ def select_provider_and_model(args=None):
_model_flow_anthropic(config, current_model)
elif selected_provider == "kimi-coding":
_model_flow_kimi(config, current_model)
elif selected_provider in ("zai", "minimax", "minimax-cn", "kilocode", "opencode-zen", "opencode-go", "ai-gateway", "alibaba", "huggingface"):
elif selected_provider in ("gemini", "zai", "minimax", "minimax-cn", "kilocode", "opencode-zen", "opencode-go", "ai-gateway", "alibaba", "huggingface"):
_model_flow_api_key_provider(config, selected_provider, current_model)
def _prompt_provider_choice(choices):
"""Show provider selection menu. Returns index or None."""
def _prompt_provider_choice(choices, *, default=0):
"""Show provider selection menu with curses arrow-key navigation.
Falls back to a numbered list when curses is unavailable (e.g. piped
stdin, non-TTY environments). Returns the selected index, or None
if the user cancels.
"""
try:
from simple_term_menu import TerminalMenu
menu_items = [f" {c}" for c in choices]
menu = TerminalMenu(
menu_items, cursor_index=0,
menu_cursor="-> ", menu_cursor_style=("fg_green", "bold"),
menu_highlight_style=("fg_green",),
cycle_cursor=True, clear_screen=False,
title="Select provider:",
)
idx = menu.show()
print()
return idx
except (ImportError, NotImplementedError):
from hermes_cli.setup import _curses_prompt_choice
idx = _curses_prompt_choice("Select provider:", choices, default)
if idx >= 0:
print()
return idx
except Exception:
pass
# Fallback: numbered list
print("Select provider:")
for i, c in enumerate(choices, 1):
print(f" {i}. {c}")
marker = "→" if i - 1 == default else " "
print(f" {marker} {i}. {c}")
print()
while True:
try:
val = input(f"Choice [1-{len(choices)}]: ").strip()
val = input(f"Choice [1-{len(choices)}] ({default + 1}): ").strip()
if not val:
return None
return default
idx = int(val) - 1
if 0 <= idx < len(choices):
return idx
@@ -1092,7 +1125,8 @@ def _model_flow_openrouter(config, current_model=""):
print("Get one at: https://openrouter.ai/keys")
print()
try:
key = input("OpenRouter API key (or Enter to cancel): ").strip()
import getpass
key = getpass.getpass("OpenRouter API key (or Enter to cancel): ").strip()
except (KeyboardInterrupt, EOFError):
print()
return
@@ -1103,10 +1137,13 @@ def _model_flow_openrouter(config, current_model=""):
print("API key saved.")
print()
from hermes_cli.models import model_ids
from hermes_cli.models import model_ids, get_pricing_for_provider
openrouter_models = model_ids()
selected = _prompt_model_selection(openrouter_models, current_model=current_model)
# Fetch live pricing (non-blocking — returns empty dict on failure)
pricing = get_pricing_for_provider("openrouter")
selected = _prompt_model_selection(openrouter_models, current_model=current_model, pricing=pricing)
if selected:
_save_model_choice(selected)
@@ -1173,7 +1210,7 @@ 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
from hermes_cli.models import _PROVIDER_MODELS, get_pricing_for_provider
model_ids = _PROVIDER_MODELS.get("nous", [])
if not model_ids:
print("No curated models available for Nous Portal.")
@@ -1203,7 +1240,10 @@ def _model_flow_nous(config, current_model="", args=None):
print(f"Could not verify credentials: {msg}")
return
selected = _prompt_model_selection(model_ids, current_model=current_model)
# 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)
if selected:
_save_model_choice(selected)
# Reactivate Nous as the provider and update config
@@ -1309,7 +1349,8 @@ def _model_flow_custom(config):
try:
base_url = input(f"API base URL [{current_url or 'e.g. https://api.example.com/v1'}]: ").strip()
api_key = input(f"API key [{current_key[:8] + '...' if current_key else 'optional'}]: ").strip()
import getpass
api_key = getpass.getpass(f"API key [{current_key[:8] + '...' if current_key else 'optional'}]: ").strip()
except (KeyboardInterrupt, EOFError):
print("\nCancelled.")
return
@@ -1818,7 +1859,8 @@ def _model_flow_copilot(config, current_model=""):
return
elif choice == "2":
try:
new_key = input(" Token (COPILOT_GITHUB_TOKEN): ").strip()
import getpass
new_key = getpass.getpass(" Token (COPILOT_GITHUB_TOKEN): ").strip()
except (KeyboardInterrupt, EOFError):
print()
return
@@ -2059,7 +2101,8 @@ def _model_flow_kimi(config, current_model=""):
print(f"No {pconfig.name} API key configured.")
if key_env:
try:
new_key = input(f"{key_env} (or Enter to cancel): ").strip()
import getpass
new_key = getpass.getpass(f"{key_env} (or Enter to cancel): ").strip()
except (KeyboardInterrupt, EOFError):
print()
return
@@ -2153,7 +2196,8 @@ def _model_flow_api_key_provider(config, provider_id, current_model=""):
print(f"No {pconfig.name} API key configured.")
if key_env:
try:
new_key = input(f"{key_env} (or Enter to cancel): ").strip()
import getpass
new_key = getpass.getpass(f"{key_env} (or Enter to cancel): ").strip()
except (KeyboardInterrupt, EOFError):
print()
return
@@ -2182,24 +2226,37 @@ def _model_flow_api_key_provider(config, provider_id, current_model=""):
save_env_value(base_url_env, override)
effective_base = override
# Model selection — try live /models endpoint first, fall back to defaults.
# Providers with large live catalogs (100+ models) use a curated list instead
# so users see familiar model names rather than an overwhelming dump.
# Model selection — resolution order:
# 1. models.dev registry (cached, filtered for agentic/tool-capable models)
# 2. Curated static fallback list (offline insurance)
# 3. Live /models endpoint probe (small providers without models.dev data)
curated = _PROVIDER_MODELS.get(provider_id, [])
if curated and len(curated) >= 8:
# Try models.dev first — returns tool-capable models, filtered for noise
mdev_models: list = []
try:
from agent.models_dev import list_agentic_models
mdev_models = list_agentic_models(provider_id)
except Exception:
pass
if mdev_models:
model_list = mdev_models
print(f" Found {len(model_list)} model(s) from models.dev registry")
elif curated and len(curated) >= 8:
# Curated list is substantial — use it directly, skip live probe
live_models = None
model_list = curated
print(f" Showing {len(model_list)} curated models — use \"Enter custom model name\" for others.")
else:
api_key_for_probe = existing_key or (get_env_value(key_env) if key_env else "")
live_models = fetch_api_models(api_key_for_probe, effective_base)
if live_models and len(live_models) >= len(curated):
model_list = live_models
print(f" Found {len(model_list)} model(s) from {pconfig.name} API")
else:
model_list = curated
if model_list:
print(f" Showing {len(model_list)} curated models — use \"Enter custom model name\" for others.")
if live_models and len(live_models) >= len(curated):
model_list = live_models
print(f" Found {len(model_list)} model(s) from {pconfig.name} API")
else:
model_list = curated
if model_list:
print(f" Showing {len(model_list)} curated models — use \"Enter custom model name\" for others.")
# else: no defaults either, will fall through to raw input
if provider_id in {"opencode-zen", "opencode-go"}:
@@ -2287,7 +2344,8 @@ def _run_anthropic_oauth_flow(save_env_value):
print(" If the setup-token was displayed above, paste it here:")
print()
try:
manual_token = input(" Paste setup-token (or Enter to cancel): ").strip()
import getpass
manual_token = getpass.getpass(" Paste setup-token (or Enter to cancel): ").strip()
except (KeyboardInterrupt, EOFError):
print()
return False
@@ -2314,7 +2372,8 @@ def _run_anthropic_oauth_flow(save_env_value):
print(" Or paste an existing setup-token now (sk-ant-oat-...):")
print()
try:
token = input(" Setup-token (or Enter to cancel): ").strip()
import getpass
token = getpass.getpass(" Setup-token (or Enter to cancel): ").strip()
except (KeyboardInterrupt, EOFError):
print()
return False
@@ -2407,7 +2466,8 @@ def _model_flow_anthropic(config, current_model=""):
print(" Get an API key at: https://console.anthropic.com/settings/keys")
print()
try:
api_key = input(" API key (sk-ant-...): ").strip()
import getpass
api_key = getpass.getpass(" API key (sk-ant-...): ").strip()
except (KeyboardInterrupt, EOFError):
print()
return
@@ -3609,6 +3669,7 @@ def cmd_update(args):
from hermes_cli.gateway import (
is_macos, is_linux, _ensure_user_systemd_env,
get_systemd_linger_status, find_gateway_pids,
_get_service_pids,
)
import signal as _signal
@@ -3675,8 +3736,11 @@ def cmd_update(args):
pass
# --- Manual (non-service) gateways ---
# Kill any remaining gateway processes not managed by a service
manual_pids = find_gateway_pids()
# Kill any remaining gateway processes not managed by a service.
# Exclude PIDs that belong to just-restarted services so we don't
# immediately kill the process that systemd/launchd just spawned.
service_pids = _get_service_pids()
manual_pids = find_gateway_pids(exclude_pids=service_pids)
for pid in manual_pids:
try:
os.kill(pid, _signal.SIGTERM)
@@ -4012,6 +4076,26 @@ def cmd_completion(args):
print(generate_bash_completion())
def cmd_logs(args):
"""View and filter Hermes log files."""
from hermes_cli.logs import tail_log, list_logs
log_name = getattr(args, "log_name", "agent") or "agent"
if log_name == "list":
list_logs()
return
tail_log(
log_name,
num_lines=getattr(args, "lines", 50),
follow=getattr(args, "follow", False),
level=getattr(args, "level", None),
session=getattr(args, "session", None),
since=getattr(args, "since", None),
)
def main():
"""Main entry point for hermes CLI."""
parser = argparse.ArgumentParser(
@@ -4042,6 +4126,10 @@ Examples:
hermes sessions list List past sessions
hermes sessions browse Interactive session picker
hermes sessions rename ID T Rename/title a session
hermes logs View agent.log (last 50 lines)
hermes logs -f Follow agent.log in real time
hermes logs errors View errors.log
hermes logs --since 1h Lines from the last hour
hermes update Update to latest version
For more help on a command:
@@ -4130,7 +4218,7 @@ For more help on a command:
)
chat_parser.add_argument(
"--provider",
choices=["auto", "openrouter", "nous", "openai-codex", "copilot-acp", "copilot", "anthropic", "huggingface", "zai", "kimi-coding", "minimax", "minimax-cn", "kilocode"],
choices=["auto", "openrouter", "nous", "openai-codex", "copilot-acp", "copilot", "anthropic", "gemini", "huggingface", "zai", "kimi-coding", "minimax", "minimax-cn", "kilocode"],
default=None,
help="Inference provider (default: auto)"
)
@@ -5377,6 +5465,53 @@ For more help on a command:
)
completion_parser.set_defaults(func=cmd_completion)
# =========================================================================
# logs command
# =========================================================================
logs_parser = subparsers.add_parser(
"logs",
help="View and filter Hermes log files",
description="View, tail, and filter agent.log / errors.log / gateway.log",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""\
Examples:
hermes logs Show last 50 lines of agent.log
hermes logs -f Follow agent.log in real time
hermes logs errors Show last 50 lines of errors.log
hermes logs gateway -n 100 Show last 100 lines of gateway.log
hermes logs --level WARNING Only show WARNING and above
hermes logs --session abc123 Filter by session ID
hermes logs --since 1h Lines from the last hour
hermes logs --since 30m -f Follow, starting from 30 min ago
hermes logs list List available log files with sizes
""",
)
logs_parser.add_argument(
"log_name", nargs="?", default="agent",
help="Log to view: agent (default), errors, gateway, or 'list' to show available files",
)
logs_parser.add_argument(
"-n", "--lines", type=int, default=50,
help="Number of lines to show (default: 50)",
)
logs_parser.add_argument(
"-f", "--follow", action="store_true",
help="Follow the log in real time (like tail -f)",
)
logs_parser.add_argument(
"--level", metavar="LEVEL",
help="Minimum log level to show (DEBUG, INFO, WARNING, ERROR)",
)
logs_parser.add_argument(
"--session", metavar="ID",
help="Filter lines containing this session ID substring",
)
logs_parser.add_argument(
"--since", metavar="TIME",
help="Show lines since TIME ago (e.g. 1h, 30m, 2d)",
)
logs_parser.set_defaults(func=cmd_logs)
# =========================================================================
# Parse and execute
# =========================================================================

6
hermes_cli/model_normalize.py
View File

@@ -8,8 +8,9 @@ Different LLM providers expect model identifiers in different formats:
hyphens: ``claude-sonnet-4-6``.
- **Copilot** expects bare names *with* dots preserved:
``claude-sonnet-4.6``.
- **OpenCode** (Zen & Go) follows the same dot-to-hyphen convention as
- **OpenCode Zen** follows the same dot-to-hyphen convention as
Anthropic: ``claude-sonnet-4-6``.
- **OpenCode Go** preserves dots in model names: ``minimax-m2.7``.
- **DeepSeek** only accepts two model identifiers:
``deepseek-chat`` and ``deepseek-reasoner``.
- **Custom** and remaining providers pass the name through as-is.
@@ -41,6 +42,7 @@ _VENDOR_PREFIXES: dict[str, str] = {
"o3": "openai",
"o4": "openai",
"gemini": "google",
"gemma": "google",
"deepseek": "deepseek",
"glm": "z-ai",
"kimi": "moonshotai",
@@ -66,7 +68,6 @@ _AGGREGATOR_PROVIDERS: frozenset[str] = frozenset({
_DOT_TO_HYPHEN_PROVIDERS: frozenset[str] = frozenset({
"anthropic",
"opencode-zen",
"opencode-go",
})
# Providers that want bare names with dots preserved.
@@ -77,6 +78,7 @@ _STRIP_VENDOR_ONLY_PROVIDERS: frozenset[str] = frozenset({
# Providers whose own naming is authoritative -- pass through unchanged.
_PASSTHROUGH_PROVIDERS: frozenset[str] = frozenset({
"gemini",
"zai",
"kimi-coding",
"minimax",

87
hermes_cli/model_switch.py
View File

@@ -51,6 +51,25 @@ from agent.models_dev import (
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Non-agentic model warning
# ---------------------------------------------------------------------------
_HERMES_MODEL_WARNING = (
"Nous Research Hermes 3 & 4 models are NOT agentic and are not designed "
"for use with Hermes Agent. They lack the tool-calling capabilities "
"required for agent workflows. Consider using an agentic model instead "
"(Claude, GPT, Gemini, DeepSeek, etc.)."
)
def _check_hermes_model_warning(model_name: str) -> str:
"""Return a warning string if *model_name* looks like a Hermes LLM model."""
if "hermes" in model_name.lower():
return _HERMES_MODEL_WARNING
return ""
# ---------------------------------------------------------------------------
# Model aliases -- short names -> (vendor, family) with NO version numbers.
# Resolved dynamically against the live models.dev catalog.
@@ -320,12 +339,37 @@ def resolve_alias(
return None
def get_authenticated_provider_slugs(
current_provider: str = "",
user_providers: dict = None,
) -> list[str]:
"""Return slugs of providers that have credentials.
Uses ``list_authenticated_providers()`` which is backed by the models.dev
in-memory cache (1 hr TTL) — no extra network cost.
"""
try:
providers = list_authenticated_providers(
current_provider=current_provider,
user_providers=user_providers,
max_models=0,
)
return [p["slug"] for p in providers]
except Exception:
return []
def _resolve_alias_fallback(
raw_input: str,
fallback_providers: tuple[str, ...] = ("openrouter", "nous"),
authenticated_providers: list[str] = (),
) -> Optional[tuple[str, str, str]]:
"""Try to resolve an alias on fallback providers."""
for provider in fallback_providers:
"""Try to resolve an alias on the user's authenticated providers.
Falls back to ``("openrouter", "nous")`` only when no authenticated
providers are supplied (backwards compat for non-interactive callers).
"""
providers = authenticated_providers or ("openrouter", "nous")
for provider in providers:
result = resolve_alias(raw_input, provider)
if result is not None:
return result
@@ -400,14 +444,25 @@ def switch_model(
# Resolve the provider
pdef = resolve_provider_full(explicit_provider, user_providers)
if pdef is None:
_switch_err = (
f"Unknown provider '{explicit_provider}'. "
f"Check 'hermes model' for available providers, or define it "
f"in config.yaml under 'providers:'."
)
# Check for common config issues that cause provider resolution failures
try:
from hermes_cli.config import validate_config_structure
_cfg_issues = validate_config_structure()
if _cfg_issues:
_switch_err += "\n\nRun 'hermes doctor' — config issues detected:"
for _ci in _cfg_issues[:3]:
_switch_err += f"\n • {_ci.message}"
except Exception:
pass
return ModelSwitchResult(
success=False,
is_global=is_global,
error_message=(
f"Unknown provider '{explicit_provider}'. "
f"Check 'hermes model' for available providers, or define it "
f"in config.yaml under 'providers:'."
),
error_message=_switch_err,
)
target_provider = pdef.id
@@ -464,7 +519,11 @@ def switch_model(
# --- Step b: Alias exists but not on current provider -> fallback ---
key = raw_input.strip().lower()
if key in MODEL_ALIASES:
fallback_result = _resolve_alias_fallback(raw_input)
authed = get_authenticated_provider_slugs(
current_provider=current_provider,
user_providers=user_providers,
)
fallback_result = _resolve_alias_fallback(raw_input, authed)
if fallback_result is not None:
target_provider, new_model, resolved_alias = fallback_result
logger.debug(
@@ -619,6 +678,14 @@ def switch_model(
# --- Get full model info from models.dev ---
model_info = get_model_info(target_provider, new_model)
# --- Collect warnings ---
warnings: list[str] = []
if validation.get("message"):
warnings.append(validation["message"])
hermes_warn = _check_hermes_model_warning(new_model)
if hermes_warn:
warnings.append(hermes_warn)
# --- Build result ---
return ModelSwitchResult(
success=True,
@@ -628,7 +695,7 @@ def switch_model(
api_key=api_key,
base_url=base_url,
api_mode=api_mode,
warning_message=validation.get("message") or "",
warning_message=" | ".join(warnings) if warnings else "",
provider_label=provider_label,
resolved_via_alias=resolved_alias,
capabilities=capabilities,

226
hermes_cli/models.py
View File

@@ -60,7 +60,6 @@ _PROVIDER_MODELS: dict[str, list[str]] = {
"nous": [
"anthropic/claude-opus-4.6",
"anthropic/claude-sonnet-4.6",
"qwen/qwen3.6-plus:free",
"anthropic/claude-sonnet-4.5",
"anthropic/claude-haiku-4.5",
"openai/gpt-5.4",
@@ -112,6 +111,17 @@ _PROVIDER_MODELS: dict[str, list[str]] = {
"gemini-2.5-pro",
"grok-code-fast-1",
],
"gemini": [
"gemini-3.1-pro-preview",
"gemini-3-flash-preview",
"gemini-3.1-flash-lite-preview",
"gemini-2.5-pro",
"gemini-2.5-flash",
"gemini-2.5-flash-lite",
# Gemma open models (also served via AI Studio)
"gemma-4-31b-it",
"gemma-4-26b-it",
],
"zai": [
"glm-5",
"glm-5-turbo",
@@ -261,6 +271,7 @@ _PROVIDER_LABELS = {
"copilot-acp": "GitHub Copilot ACP",
"nous": "Nous Portal",
"copilot": "GitHub Copilot",
"gemini": "Google AI Studio",
"zai": "Z.AI / GLM",
"kimi-coding": "Kimi / Moonshot",
"minimax": "MiniMax",
@@ -287,6 +298,9 @@ _PROVIDER_ALIASES = {
"github-model": "copilot",
"github-copilot-acp": "copilot-acp",
"copilot-acp-agent": "copilot-acp",
"google": "gemini",
"google-gemini": "gemini",
"google-ai-studio": "gemini",
"kimi": "kimi-coding",
"moonshot": "kimi-coding",
"minimax-china": "minimax-cn",
@@ -327,6 +341,213 @@ def menu_labels() -> list[str]:
return labels
# ---------------------------------------------------------------------------
# Pricing helpers — fetch live pricing from OpenRouter-compatible /v1/models
# ---------------------------------------------------------------------------
# Cache: maps model_id → {"prompt": str, "completion": str} per endpoint
_pricing_cache: dict[str, dict[str, dict[str, str]]] = {}
def _format_price_per_mtok(per_token_str: str) -> str:
"""Convert a per-token price string to a human-friendly $/Mtok string.
Always uses 2 decimal places so that prices align vertically when
right-justified in a column (the decimal point stays in the same position).
Examples:
"0.000003" → "$3.00" (per million tokens)
"0.00003" → "$30.00"
"0.00000015" → "$0.15"
"0.0000001" → "$0.10"
"0.00018" → "$180.00"
"0" → "free"
"""
try:
val = float(per_token_str)
except (TypeError, ValueError):
return "?"
if val == 0:
return "free"
per_m = val * 1_000_000
return f"${per_m:.2f}"
def format_pricing_label(pricing: dict[str, str] | None) -> str:
"""Build a compact pricing label like 'in $3 · out $15 · cache $0.30/Mtok'.
Returns empty string when pricing is unavailable.
"""
if not pricing:
return ""
prompt_price = pricing.get("prompt", "")
completion_price = pricing.get("completion", "")
if not prompt_price and not completion_price:
return ""
inp = _format_price_per_mtok(prompt_price)
out = _format_price_per_mtok(completion_price)
if inp == "free" and out == "free":
return "free"
cache_read = pricing.get("input_cache_read", "")
cache_str = _format_price_per_mtok(cache_read) if cache_read else ""
if inp == out and not cache_str:
return f"{inp}/Mtok"
parts = [f"in {inp}", f"out {out}"]
if cache_str and cache_str != "?" and cache_str != inp:
parts.append(f"cache {cache_str}")
return " · ".join(parts) + "/Mtok"
def format_model_pricing_table(
models: list[tuple[str, str]],
pricing_map: dict[str, dict[str, str]],
current_model: str = "",
indent: str = " ",
) -> list[str]:
"""Build a column-aligned model+pricing table for terminal display.
Returns a list of pre-formatted lines ready to print.
*models* is ``[(model_id, description), ...]``.
"""
if not models:
return []
# Build rows: (model_id, input_price, output_price, cache_price, is_current)
rows: list[tuple[str, str, str, str, bool]] = []
has_cache = False
for mid, _desc in models:
is_cur = mid == current_model
p = pricing_map.get(mid)
if p:
inp = _format_price_per_mtok(p.get("prompt", ""))
out = _format_price_per_mtok(p.get("completion", ""))
cache_read = p.get("input_cache_read", "")
cache = _format_price_per_mtok(cache_read) if cache_read else ""
if cache:
has_cache = True
else:
inp, out, cache = "", "", ""
rows.append((mid, inp, out, cache, is_cur))
name_col = max(len(r[0]) for r in rows) + 2
# Compute price column widths from the actual data so decimals align
price_col = max(
max((len(r[1]) for r in rows if r[1]), default=4),
max((len(r[2]) for r in rows if r[2]), default=4),
3, # minimum: "In" / "Out" header
)
cache_col = max(
max((len(r[3]) for r in rows if r[3]), default=4),
5, # minimum: "Cache" header
) if has_cache else 0
lines: list[str] = []
# Header
if has_cache:
lines.append(f"{indent}{'Model':<{name_col}} {'In':>{price_col}} {'Out':>{price_col}} {'Cache':>{cache_col}} /Mtok")
lines.append(f"{indent}{'-' * name_col} {'-' * price_col} {'-' * price_col} {'-' * cache_col}")
else:
lines.append(f"{indent}{'Model':<{name_col}} {'In':>{price_col}} {'Out':>{price_col}} /Mtok")
lines.append(f"{indent}{'-' * name_col} {'-' * price_col} {'-' * price_col}")
for mid, inp, out, cache, is_cur in rows:
marker = " ← current" if is_cur else ""
if has_cache:
lines.append(f"{indent}{mid:<{name_col}} {inp:>{price_col}} {out:>{price_col}} {cache:>{cache_col}}{marker}")
else:
lines.append(f"{indent}{mid:<{name_col}} {inp:>{price_col}} {out:>{price_col}}{marker}")
return lines
def fetch_models_with_pricing(
api_key: str | None = None,
base_url: str = "https://openrouter.ai/api",
timeout: float = 8.0,
*,
force_refresh: bool = False,
) -> dict[str, dict[str, str]]:
"""Fetch ``/v1/models`` and return ``{model_id: {prompt, completion}}`` pricing.
Results are cached per *base_url* so repeated calls are free.
Works with any OpenRouter-compatible endpoint (OpenRouter, Nous Portal).
"""
cache_key = (base_url or "").rstrip("/")
if not force_refresh and cache_key in _pricing_cache:
return _pricing_cache[cache_key]
url = cache_key.rstrip("/") + "/v1/models"
headers: dict[str, str] = {"Accept": "application/json"}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
try:
req = urllib.request.Request(url, headers=headers)
with urllib.request.urlopen(req, timeout=timeout) as resp:
payload = json.loads(resp.read().decode())
except Exception:
_pricing_cache[cache_key] = {}
return {}
result: dict[str, dict[str, str]] = {}
for item in payload.get("data", []):
mid = item.get("id")
pricing = item.get("pricing")
if mid and isinstance(pricing, dict):
entry: dict[str, str] = {
"prompt": str(pricing.get("prompt", "")),
"completion": str(pricing.get("completion", "")),
}
if pricing.get("input_cache_read"):
entry["input_cache_read"] = str(pricing["input_cache_read"])
if pricing.get("input_cache_write"):
entry["input_cache_write"] = str(pricing["input_cache_write"])
result[mid] = entry
_pricing_cache[cache_key] = result
return result
def _resolve_openrouter_api_key() -> str:
"""Best-effort OpenRouter API key for pricing fetch."""
return os.getenv("OPENROUTER_API_KEY", "").strip()
def _resolve_nous_pricing_credentials() -> tuple[str, str]:
"""Return ``(api_key, base_url)`` for Nous Portal pricing, or empty strings."""
try:
from hermes_cli.auth import resolve_nous_runtime_credentials
creds = resolve_nous_runtime_credentials()
if creds:
return (creds.get("api_key", ""), creds.get("base_url", ""))
except Exception:
pass
return ("", "")
def get_pricing_for_provider(provider: str) -> dict[str, dict[str, str]]:
"""Return live pricing for providers that support it (openrouter, nous)."""
normalized = normalize_provider(provider)
if normalized == "openrouter":
return fetch_models_with_pricing(
api_key=_resolve_openrouter_api_key(),
base_url="https://openrouter.ai/api",
)
if normalized == "nous":
api_key, base_url = _resolve_nous_pricing_credentials()
if base_url:
# Nous base_url typically looks like https://inference-api.nousresearch.com/v1
# We need the part before /v1 for our fetch function
stripped = base_url.rstrip("/")
if stripped.endswith("/v1"):
stripped = stripped[:-3]
return fetch_models_with_pricing(
api_key=api_key,
base_url=stripped,
)
return {}
# All provider IDs and aliases that are valid for the provider:model syntax.
_KNOWN_PROVIDER_NAMES: set[str] = (
set(_PROVIDER_LABELS.keys())
@@ -344,7 +565,8 @@ def list_available_providers() -> list[dict[str, str]]:
# Canonical providers in display order
_PROVIDER_ORDER = [
"openrouter", "nous", "openai-codex", "copilot", "copilot-acp",
"huggingface", "zai", "kimi-coding", "minimax", "minimax-cn", "kilocode", "anthropic", "alibaba",
"gemini", "huggingface",
"zai", "kimi-coding", "minimax", "minimax-cn", "kilocode", "anthropic", "alibaba",
"opencode-zen", "opencode-go",
"ai-gateway", "deepseek", "custom",
]

7
hermes_cli/nous_subscription.py
View File

@@ -131,6 +131,7 @@ def _browser_label(current_provider: str) -> str:
mapping = {
"browserbase": "Browserbase",
"browser-use": "Browser Use",
"firecrawl": "Firecrawl",
"camofox": "Camofox",
"local": "Local browser",
}
@@ -156,6 +157,7 @@ def _resolve_browser_feature_state(
direct_camofox: bool,
direct_browserbase: bool,
direct_browser_use: bool,
direct_firecrawl: bool,
managed_browser_available: bool,
) -> tuple[str, bool, bool, bool]:
"""Resolve browser availability using the same precedence as runtime."""
@@ -179,6 +181,10 @@ def _resolve_browser_feature_state(
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)
return current_provider, available, active, False
if current_provider == "camofox":
return current_provider, False, False, False
@@ -315,6 +321,7 @@ def get_nous_subscription_features(
direct_camofox=direct_camofox,
direct_browserbase=direct_browserbase,
direct_browser_use=direct_browser_use,
direct_firecrawl=direct_firecrawl,
managed_browser_available=managed_browser_available,
)

2
hermes_cli/plugins.py
View File

@@ -56,6 +56,8 @@ VALID_HOOKS: Set[str] = {
"post_tool_call",
"pre_llm_call",
"post_llm_call",
"pre_api_request",
"post_api_request",
"on_session_start",
"on_session_end",
}

17
hermes_cli/plugins_cmd.py
View File

@@ -41,6 +41,11 @@ def _sanitize_plugin_name(name: str, plugins_dir: Path) -> Path:
if not name:
raise ValueError("Plugin name must not be empty.")
if name in (".", ".."):
raise ValueError(
f"Invalid plugin name '{name}': must not reference the plugins directory itself."
)
# Reject obvious traversal characters
for bad in ("/", "\\", ".."):
if bad in name:
@@ -49,10 +54,14 @@ def _sanitize_plugin_name(name: str, plugins_dir: Path) -> Path:
target = (plugins_dir / name).resolve()
plugins_resolved = plugins_dir.resolve()
if (
not str(target).startswith(str(plugins_resolved) + os.sep)
and target != plugins_resolved
):
if target == plugins_resolved:
raise ValueError(
f"Invalid plugin name '{name}': resolves to the plugins directory itself."
)
try:
target.relative_to(plugins_resolved)
except ValueError:
raise ValueError(
f"Invalid plugin name '{name}': resolves outside the plugins directory."
)

15
hermes_cli/runtime_provider.py
View File

@@ -2,10 +2,13 @@
from __future__ import annotations
import logging
import os
import re
from typing import Any, Dict, Optional
logger = logging.getLogger(__name__)
from hermes_cli import auth as auth_mod
from agent.credential_pool import CredentialPool, PooledCredential, get_custom_provider_pool_key, load_pool
from hermes_cli.auth import (
@@ -258,6 +261,12 @@ def _get_named_custom_provider(requested_provider: str) -> Optional[Dict[str, An
config = load_config()
custom_providers = config.get("custom_providers")
if not isinstance(custom_providers, list):
if isinstance(custom_providers, dict):
logger.warning(
"custom_providers in config.yaml is a dict, not a list. "
"Each entry must be prefixed with '-' in YAML. "
"Run 'hermes doctor' for details."
)
return None
for entry in custom_providers:
@@ -486,7 +495,11 @@ def _resolve_explicit_runtime(
explicit_base_url
or str(state.get("inference_base_url") or auth_mod.DEFAULT_NOUS_INFERENCE_URL).strip().rstrip("/")
)
api_key = explicit_api_key or str(state.get("agent_key") or state.get("access_token") or "").strip()
# Only use agent_key for inference — access_token is an OAuth token for the
# portal API (minting keys, refreshing tokens), not for the inference API.
# Falling back to access_token sends an OAuth bearer token to the inference
# endpoint, which returns 404 because it is not a valid inference credential.
api_key = explicit_api_key or str(state.get("agent_key") or "").strip()
expires_at = state.get("agent_key_expires_at") or state.get("expires_at")
if not api_key:
creds = resolve_nous_runtime_credentials(

1049
hermes_cli/setup.py
View File

File diff suppressed because it is too large Load Diff

9
hermes_cli/tools_config.py
View File

@@ -315,6 +315,15 @@ TOOL_CATEGORIES = {
"browser_provider": "browser-use",
"post_setup": "browserbase",
},
{
"name": "Firecrawl",
"tag": "Cloud browser with remote execution",
"env_vars": [
{"key": "FIRECRAWL_API_KEY", "prompt": "Firecrawl API key", "url": "https://firecrawl.dev"},
],
"browser_provider": "firecrawl",
"post_setup": "browserbase",
},
{
"name": "Camofox",
"tag": "Local anti-detection browser (Firefox/Camoufox)",