ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-19-C' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:57:58 -07:00
parent 3db40dc962 f0b1d49d0f
commit 20cc091aba
6 changed files with 665 additions and 1405 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/env_loader.py
+128 -295
View File
@@ -15,73 +15,44 @@ from utils import atomic_replace, fast_safe_load
logger = logging.getLogger(__name__)
# Env var name suffixes that indicate credential values. These are the
# only env vars whose values we sanitize on load — we must not silently
# alter arbitrary user env vars, but credentials are known to require
# pure ASCII (they become HTTP header values).
# The ONLY env vars sanitized on load: we must not silently alter arbitrary user env vars, but
# credentials are known to require pure ASCII (they become HTTP header values).
_CREDENTIAL_SUFFIXES = ("_API_KEY", "_TOKEN", "_SECRET", "_KEY")
# Names we've already warned about during this process, so repeated
# load_hermes_dotenv() calls (user env + project env, gateway hot-reload,
# tests) don't spam the same warning multiple times.
_WARNED_KEYS: set[str] = set()
# Once-per-process guards: load_hermes_dotenv() is called repeatedly (user env + project env,
# gateway hot-reload, lazy imports mid-turn, tests) so warnings/logs fire once per key/path/home.
_WARNED_KEYS: set[str] = set() # credential names already given the non-ASCII warning
_WARNED_UTF32_PATHS: set[str] = set() # .env paths already given the UTF-32 refuse-to-mangle warning
_SCOPED_SKIP_LOGGED: set[str] = set() # routed profile homes whose multiplex dotenv skip was logged
# Paths we've already emitted a UTF-32 refuse-to-mangle warning for.
# load_hermes_dotenv can call _sanitize_env_file_if_needed multiple times
# for the same file (user env + project env + hot-reload); once per path
# is enough.
_WARNED_UTF32_PATHS: set[str] = set()
# Map of env-var name → source label ("bitwarden", etc.) for credentials
# that were injected by an external secret source during load_hermes_dotenv().
# Used by setup / `hermes model` flows to label detected credentials so
# users understand WHERE a key came from when their .env doesn't contain it
# directly (otherwise the "credentials detected ✓" line looks identical to
# the .env case and they don't know Bitwarden is wired up).
# env-var name → source label ("bitwarden", …) for credentials injected by an external secret
# source. Setup / `hermes model` use it so users know WHERE a key came from when .env lacks it.
_SECRET_SOURCES: dict[str, str] = {}
# Applied values are immutable per-home snapshots. ``os.environ`` is shared
# across profiles and may be overwritten by a later home's source apply.
# Applied values are immutable per-home snapshots: ``os.environ`` is shared across profiles and
# may be overwritten by a later home's source apply.
_SECRET_SOURCE_VALUES_BY_HOME: dict[str, dict[str, str]] = {}
# HERMES_HOME paths we've already pulled external secrets for during this
# process. ``load_hermes_dotenv()`` is called at module-import time from
# several hot modules (cli.py, hermes_cli/main.py, run_agent.py,
# trajectory_compressor.py, gateway/run.py, ...), so without this guard the
# Bitwarden status line gets printed 3-5x per startup. Bitwarden's own
# in-process cache prevents redundant network calls, but the print, the
# config re-parse, and the ASCII sanitization sweep still ran every time.
# HERMES_HOME paths already pulled external secrets for. load_hermes_dotenv() runs at import time
# from several hot modules (cli.py, hermes_cli/main.py, run_agent.py, gateway/run.py, …); without
# this the Bitwarden status line prints 3-5x per startup and the config re-parse + ASCII sweep
# re-run each time (Bitwarden's own cache only saves the network call).
_APPLIED_HOMES: set[str] = set()
_SECRET_SOURCE_CACHE_LOCK = threading.RLock()
# Routed profile homes whose dotenv load was skipped under multiplex, so the
# skip is logged once per home rather than on every lazy import mid-turn.
_SCOPED_SKIP_LOGGED: set[str] = set()
# Behavioral routing keys a parent Hermes process injects into child env and
# that silently redirect a profile onto the wrong provider path (ACP auth
# method, copilot-ACP endpoints). These — and ONLY these — are scrubbed from
# os.environ at startup when absent from the profile's .env. Credential keys
# (API keys/tokens) are excluded: shell exports are a legitimate,
# documented way to supply them, and read-time secret-scope checks
# Behavioral routing keys a parent Hermes process injects into child env that silently redirect a
# profile onto the wrong provider path. These — and ONLY these — are scrubbed from os.environ at
# startup when absent from the profile's .env. Credential keys are excluded: shell exports are a
# legitimate, documented way to supply them, and read-time secret-scope checks
# (agent/secret_scope.py) own cross-profile credential isolation.
_PROFILE_MANAGED_ENV_KEYS: frozenset[str] = frozenset({
"HERMES_ACP_AUTH_METHOD",
"HERMES_ACP_AUTO_APPROVE",
"HERMES_COPILOT_ACP_COMMAND",
"HERMES_COPILOT_ACP_ARGS",
"COPILOT_CLI_PATH",
"COPILOT_ACP_BASE_URL",
"HERMES_ACP_AUTH_METHOD", "HERMES_ACP_AUTO_APPROVE", "HERMES_COPILOT_ACP_COMMAND",
"HERMES_COPILOT_ACP_ARGS", "COPILOT_CLI_PATH", "COPILOT_ACP_BASE_URL",
})
def _env_keys_defined_in_dotenv(path: Path) -> set[str]:
"""Return KEY names assigned in a dotenv file (including empty ``KEY=``).
A fast line scanner rather than full dotenv parsing, so it works during early bootstrap
without importing python-dotenv. Non-ASCII decode errors fall back to ``latin-1``, matching
``_load_dotenv_with_fallback``.
"""
"""KEY names assigned in a dotenv file (including empty ``KEY=``). A fast line scanner, not full
dotenv parsing, so it works during early bootstrap without importing python-dotenv; decode
errors fall back to ``latin-1`` like ``_load_dotenv_with_fallback``."""
keys: set[str] = set()
try:
text = path.read_text(encoding="utf-8", errors="replace")
@@ -101,16 +72,8 @@ def _env_keys_defined_in_dotenv(path: Path) -> set[str]:
def _clear_known_keys_missing_from_dotenv(path: Path) -> None:
"""Remove inherited profile-managed Hermes keys absent from ``.env``.
After the profile's ``.env`` has been loaded with ``override=True``, scan the file for which
profile-managed keys it explicitly defines and delete any such key that exists in ``os.environ``
but is *not* present in the file.
Scope is deliberately NARROW: only ``_PROFILE_MANAGED_ENV_KEYS`` — behavioral routing keys (ACP
auth method, copilot-ACP endpoints) that a parent Hermes process injects and that silently
change *which provider path* a profile uses.
"""
"""After ``.env`` loaded with ``override=True``, delete inherited ``_PROFILE_MANAGED_ENV_KEYS``
it does not define. Deliberately NARROW: only keys that change *which provider path* is used."""
if not path.exists():
return
defined = _env_keys_defined_in_dotenv(path)
@@ -120,36 +83,25 @@ def _clear_known_keys_missing_from_dotenv(path: Path) -> None:
def get_secret_source(env_var: str) -> str | None:
"""Return the label of the secret source that supplied ``env_var``, if any.
``"bitwarden"`` for keys pulled during this process's ``load_hermes_dotenv()``; ``None`` for
``.env``/shell keys. The label is metadata only -- credential-pool persistence may store it to
explain a borrowed secret's origin but must never treat it as authorization to persist the
raw value.
"""
"""Source label that supplied ``env_var`` (``"bitwarden"`` …), None for .env/shell keys. Metadata
only -- credential-pool persistence may store it to explain a borrowed secret's origin but must
never treat it as authorization to persist the raw value."""
return _SECRET_SOURCES.get(env_var)
def get_secret_source_values(
hermes_home: str | os.PathLike,
) -> dict[str, str]:
def get_secret_source_values(hermes_home: str | os.PathLike) -> dict[str, str]:
"""Return the external-secret value snapshot for ``hermes_home``."""
home_key = str(Path(hermes_home).resolve())
return dict(_SECRET_SOURCE_VALUES_BY_HOME.get(home_key, {}))
return dict(_SECRET_SOURCE_VALUES_BY_HOME.get(str(Path(hermes_home).resolve()), {}))
def hydrate_profile_secret_sources(
hermes_home: str | os.PathLike,
) -> dict[str, str]:
def hydrate_profile_secret_sources(hermes_home: str | os.PathLike) -> dict[str, str]:
"""Resolve one profile's configured sources without mutating ``os.environ``.
Multiplex gateways can route a first turn to a secondary profile that has never run the process-
global dotenv startup path. Resolve that profile's sources against a private mapping seeded from
its own ``.env`` and record the usual per-home snapshot for ``build_profile_secret_scope()``.
Fail-open and once-per-home semantics intentionally mirror ``_apply_external_secret_sources``.
The returned mapping contains only values actually contributed by external sources, never the
profile's plaintext ``.env`` entries.
Multiplex gateways can route a first turn to a secondary profile that never ran the process-
global dotenv startup path: resolve against a private mapping seeded from its own ``.env`` and
record the per-home snapshot for ``build_profile_secret_scope()``. Fail-open and once-per-home
semantics mirror ``_apply_external_secret_sources``; the result holds only values contributed
by external sources, never plaintext ``.env`` entries.
"""
with _SECRET_SOURCE_CACHE_LOCK:
return _hydrate_profile_secret_sources(Path(hermes_home))
@@ -172,17 +124,10 @@ def _hydrate_profile_secret_sources(home: Path) -> dict[str, str]:
from agent.secret_scope import _is_global_env, load_env_file
from agent.secret_sources.registry import apply_all
local_env = {
name: value
for name, value in os.environ.items()
if _is_global_env(name)
}
local_env = {name: value for name, value in os.environ.items() if _is_global_env(name)}
local_env.update(load_env_file(home / ".env"))
# Mirror load_hermes_dotenv()'s .op.env bootstrap: the 1Password
# service-account token lives in <home>/.op.env (gitignored), not
# .env. Without seeding it here a cold profile configured for the
# supported .op.env flow fails 1Password hydration (sweeper review
# on #74549). .env values win — never override an existing key.
# Mirror load_hermes_dotenv()'s .op.env bootstrap (1Password service-account token lives in
# gitignored <home>/.op.env, not .env) or cold profiles fail 1Password hydration. .env wins.
op_env = home / ".op.env"
if op_env.exists():
for _name, _value in load_env_file(op_env).items():
@@ -209,30 +154,21 @@ def _hydrate_profile_secret_sources(home: Path) -> dict[str, str]:
def reset_secret_source_cache() -> None:
"""Forget which HERMES_HOME paths have already had external secrets applied.
``_apply_external_secret_sources(home_path)`` pulls once per process and then no-ops for that
home; call this to force a re-pull (tests, long-running processes after a config change).
"""
"""Forget applied homes so the next load re-pulls (tests, long-running processes after config edits)."""
_APPLIED_HOMES.clear()
_SECRET_SOURCES.clear()
_SECRET_SOURCE_VALUES_BY_HOME.clear()
def format_secret_source_suffix(env_var: str) -> str:
"""Return a human-readable suffix like ``" (from Bitwarden)"`` or ``""``.
Empty for ``.env``/shell credentials -- those are the implicit default cases users already
understand, so only external sources get called out.
"""
"""``" (from Bitwarden)"``-style suffix; ``""`` for .env/shell credentials (the implicit default
users already understand — only external sources get called out)."""
source = get_secret_source(env_var)
if not source:
return ""
if source == "bitwarden":
return " (from Bitwarden)"
# Ask the registry for the source's human label (e.g. "1Password").
# Fall back to the raw source name for labels the registry doesn't
# know (stale provenance from an uninstalled plugin, tests).
# Registry label (e.g. "1Password"); raw name for unknown sources (uninstalled plugin, tests).
try:
from agent.secret_sources.registry import get_source
@@ -245,7 +181,7 @@ def format_secret_source_suffix(env_var: str) -> str:
def _format_offending_chars(value: str, limit: int = 3) -> str:
"""Return a compact 'U+XXXX ('c'), ...' summary of non-ASCII codepoints."""
"""Compact ``U+XXXX ('c'), ...`` summary of non-ASCII codepoints."""
seen: list[str] = []
for ch in value:
if ord(ch) > 127:
@@ -260,12 +196,8 @@ def _format_offending_chars(value: str, limit: int = 3) -> str:
def _sanitize_loaded_credentials() -> None:
"""Strip non-ASCII characters from credential env vars in os.environ.
Called after dotenv loads so the rest of the codebase never sees non-ASCII API keys. Only
touches env vars whose names end with known credential suffixes (``_API_KEY``, ``_TOKEN``,
etc.).
"""
"""Strip non-ASCII from credential env vars (``_CREDENTIAL_SUFFIXES``) after dotenv loads, so the
rest of the codebase never sees non-ASCII API keys."""
for key, value in list(os.environ.items()):
if not any(key.endswith(suffix) for suffix in _CREDENTIAL_SUFFIXES):
continue
@@ -297,10 +229,8 @@ def _sanitize_loaded_credentials() -> None:
def _load_dotenv_with_fallback(path: Path, *, override: bool) -> None:
try:
# utf-8-sig strips a leading UTF-8 BOM if present (PowerShell 5.1
# Set-Content -Encoding UTF8 / Notepad) and is a no-op for BOM-less
# UTF-8. Plain "utf-8" would keep U+FEFF on the first key name and
# silently drop it from os.environ under its canonical name.
# utf-8-sig strips a leading UTF-8 BOM (PowerShell 5.1 / Notepad); plain "utf-8" would keep
# U+FEFF on the first key name and silently drop it from os.environ under its canonical name.
load_dotenv(dotenv_path=path, override=override, encoding="utf-8-sig")
except UnicodeDecodeError:
# utf-8-sig can't strip a BOM once we fall back to latin-1 decode.
@@ -308,24 +238,18 @@ def _load_dotenv_with_fallback(path: Path, *, override: bool) -> None:
if raw.startswith(codecs.BOM_UTF8):
raw = raw[len(codecs.BOM_UTF8) :]
load_dotenv(stream=io.StringIO(raw.decode("latin-1")), override=override)
# Strip non-ASCII characters from credential env vars that were just
# loaded. API keys must be pure ASCII since they're sent as HTTP
# header values (httpx encodes headers as ASCII). Non-ASCII chars
# typically come from copy-pasting keys from PDFs or rich-text editors
# that substitute Unicode lookalike glyphs (e.g. ʋ U+028B for v).
# API keys must be pure ASCII (httpx encodes headers as ASCII); non-ASCII typically comes from
# keys copy-pasted out of PDFs / rich-text editors that substitute lookalike glyphs.
_sanitize_loaded_credentials()
def _sanitize_env_file_if_needed(path: Path) -> None:
"""Pre-sanitize a .env file before python-dotenv reads it.
Encoding: sniffs a leading BOM *before* any text decode. UTF-16 (Notepad "Unicode") is decoded
correctly and rewritten as clean UTF-8. UTF-32 is refused (left untouched) so we never fall
through to the errors=replace corruption path. Order of BOM checks matters: UTF-32-LE's BOM
starts with UTF-16-LE's FF FE.
``hermes_cli.config._sanitize_env_lines`` normalizes line endings while treating content after
the first ``=`` as opaque for boundary discovery.
Sniffs a leading BOM *before* any text decode: UTF-16 (Notepad "Unicode") is decoded and
rewritten as clean UTF-8; UTF-32 is refused (left untouched) so we never fall through to the
errors=replace corruption path. ``hermes_cli.config._sanitize_env_lines`` normalizes line
endings while treating content after the first ``=`` as opaque for boundary discovery.
"""
if not path.exists():
return
@@ -339,14 +263,10 @@ def _sanitize_env_file_if_needed(path: Path) -> None:
except Exception:
return
# Sniff leading BOM bytes BEFORE decoding. ORDER MATTERS:
# codecs.BOM_UTF32_LE is FF FE 00 00, which startswith
# codecs.BOM_UTF16_LE (FF FE). Checking UTF-16 first would
# misdetect UTF-32-LE as UTF-16-LE and mangle the file.
# ORDER MATTERS: BOM_UTF32_LE (FF FE 00 00) startswith BOM_UTF16_LE (FF FE); checking UTF-16
# first would misdetect UTF-32-LE and mangle the file.
force_utf8_rewrite = False
if raw.startswith(codecs.BOM_UTF32_LE) or raw.startswith(codecs.BOM_UTF32_BE):
# Lazy import keeps the module import block identical to #65124's
# codecs/io additions so the two PRs auto-merge either order.
path_key = str(path.resolve())
if path_key not in _WARNED_UTF32_PATHS:
_WARNED_UTF32_PATHS.add(path_key)
@@ -357,49 +277,35 @@ def _sanitize_env_file_if_needed(path: Path) -> None:
)
return
if raw.startswith(codecs.BOM_UTF16_LE) or raw.startswith(codecs.BOM_UTF16_BE):
# "utf-16" uses the BOM to select endianness and strips it.
# TextIOWrapper + newline=None matches open()'s universal-newlines
# line splitting (\\n/\\r\\n/\\r only — not splitlines()'s extra
# Unicode boundaries like U+2028), so sanitize sees the same lines
# as the UTF-8 path.
# "utf-16" uses the BOM for endianness and strips it. newline=None matches open()'s
# universal-newlines splitting (\n/\r\n/\r only — not splitlines()'s extra Unicode
# boundaries like U+2028), so sanitize sees the same lines as the UTF-8 path.
try:
with io.TextIOWrapper(
io.BytesIO(raw), encoding="utf-16", newline=None
) as f:
with io.TextIOWrapper(io.BytesIO(raw), encoding="utf-16", newline=None) as f:
original = f.readlines()
except UnicodeDecodeError:
return
# Source is UTF-16 on disk; always rewrite as clean UTF-8 so
# the subsequent utf-8 dotenv load sees a canonical file.
force_utf8_rewrite = True
force_utf8_rewrite = True # always rewrite UTF-16 as UTF-8 so the dotenv load sees a canonical file
else:
# Default path: utf-8-sig (strips UTF-8 BOM if present) with
# errors=replace so embedded NULs can be stripped below.
# utf-8-sig strips a UTF-8 BOM; errors=replace so embedded NULs can be stripped below.
try:
with open(path, encoding="utf-8-sig", errors="replace") as f:
original = f.readlines()
except Exception:
return
# Defense-in-depth: errors=replace turns undecodable leading
# bytes into U+FFFD. Persisting that glues replacement chars
# onto the first key name and rewrites the file permanently
# (the UTF-16-with-BOM corruption path before BOM sniffing).
# Leave the file untouched rather than write the mangling.
# errors=replace turns undecodable leading bytes into U+FFFD; persisting that would glue
# replacement chars onto the first key name permanently. Leave the file untouched instead.
if original and original[0].startswith("\ufffd"):
return
try:
# Strip null bytes before _sanitize_env_lines so they never
# reach python-dotenv (which passes them to os.environ and
# crashes with ValueError). Also intentionally repairs
# BOM-less UTF-16 (NUL-padded ASCII) into clean UTF-8.
# Strip NULs so they never reach python-dotenv (os.environ raises ValueError on them); this
# also intentionally repairs BOM-less UTF-16 (NUL-padded ASCII) into clean UTF-8.
stripped = [line.replace("\x00", "") for line in original]
sanitized = _sanitize_env_lines(stripped)
if sanitized != original or force_utf8_rewrite:
import tempfile
fd, tmp = tempfile.mkstemp(
dir=str(path.parent), suffix=".tmp", prefix=".env_"
)
fd, tmp = tempfile.mkstemp(dir=str(path.parent), suffix=".tmp", prefix=".env_")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
f.writelines(sanitized)
@@ -422,22 +328,15 @@ def load_hermes_dotenv(
project_env: str | os.PathLike | None = None,
load_external_secrets: bool = True,
) -> list[Path]:
"""Load Hermes environment files with user config taking precedence.
Behavior: - `~/.hermes/.env` overrides stale shell-exported values when present. - project
`.env` acts as a dev fallback and only fills missing values when the user env exists. - if no
user env exists, the project `.env` also overrides stale shell vars.
"""
"""Load Hermes env files, user config first: ``~/.hermes/.env`` overrides stale shell exports;
project ``.env`` is a dev fallback that only fills gaps when the user env exists (and overrides
stale shell vars when it does not)."""
home_path = Path(hermes_home or os.getenv("HERMES_HOME", Path.home() / ".hermes"))
# A multiplex gateway hosts every profile in one process. While a routed
# profile-home override is active, copying that profile's .env into
# os.environ would expose its credentials to sibling turns and every
# subsequently spawned child. An unscoped startup load remains process
# configuration and must retain the normal loading path.
# External secret sources still need their normal refresh path, so resolve
# them against the existing profile-local mapping instead of simply
# returning before all hydration work.
# A multiplex gateway hosts every profile in one process: while a routed profile-home override
# is active, copying that profile's .env into os.environ would expose its credentials to sibling
# turns and every spawned child. An unscoped startup load is process configuration and keeps the
# normal path. External sources still refresh, against the profile-local mapping.
from agent.secret_scope import is_multiplex_active
from hermes_constants import get_hermes_home_override
@@ -470,20 +369,14 @@ def load_hermes_dotenv(
if user_env.exists():
_load_dotenv_with_fallback(user_env, override=True)
loaded.append(user_env)
# Mirror reload_env() known-key cleanup so inherited Hermes keys
# absent from this profile's .env do not leak into the runtime.
# Mirror reload_env() known-key cleanup so inherited Hermes keys absent from this
# profile's .env do not leak into the runtime.
_clear_known_keys_missing_from_dotenv(user_env)
# Load .op.env AFTER .env so that .env values win, but the bootstrap
# token (OP_SERVICE_ACCOUNT_TOKEN) becomes available for
# apply_onepassword_secrets() even in cron / subprocess environments
# that inherit no shell state (no systemd EnvironmentFile, no op run).
# .op.env is gitignored — the service-account token never enters the
# committed .env file.
# Users on systemd can alternatively use:
# EnvironmentFile=-/path/to/.hermes/.op.env
# in their gateway unit, which takes precedence (override=False below
# ensures .op.env never clobbers a token already in the environment).
# .op.env AFTER .env so .env wins, but the bootstrap OP_SERVICE_ACCOUNT_TOKEN is available to
# apply_onepassword_secrets() even in cron / subprocesses with no shell state. It is gitignored
# so the token never enters the committed .env; override=False lets a systemd
# `EnvironmentFile=-…/.op.env` token take precedence.
op_env = home_path / ".op.env"
if op_env.exists() and not os.environ.get("OP_SERVICE_ACCOUNT_TOKEN"):
_load_dotenv_with_fallback(op_env, override=False)
@@ -492,48 +385,32 @@ def load_hermes_dotenv(
_load_dotenv_with_fallback(project_env_path, override=not loaded)
loaded.append(project_env_path)
# External secret sources are skipped in two updater situations:
# 1. ``load_external_secrets=False`` — the caller is an ``update``
# invocation that must not import optional secret-manager libraries
# (Bitwarden → cryptography → ``_rust.pyd``) into the process that
# replaces that same environment on Windows (#73381, #86735).
# 2. A fresh ``hermes update`` retry just completed a deferred dependency
# install before importing this module. Do not remap native
# secret-source dependencies in that same updater process or the
# self-lock preflight will recreate the marker and exit 2 again.
# Dotenv and managed env still load in both cases; only external source
# resolution is unnecessary for the updater.
# External sources are skipped for the updater (dotenv + managed env still load):
# 1. ``load_external_secrets=False`` — ``update`` must not import optional secret-manager libs
# (Bitwarden → cryptography → ``_rust.pyd``) into the process replacing that env on Windows.
# 2. A fresh ``hermes update`` retry just completed a deferred dependency install; mapping native
# secret-source deps in that process would make the self-lock preflight exit 2 again.
from hermes_cli import _early_recovery
if load_external_secrets and not _early_recovery._should_skip_external_secret_sources():
_apply_external_secret_sources(home_path)
_apply_managed_env()
# config.yaml is the documented source of truth for terminal.* settings,
# but the dotenv loads above run with override=True — so a stale
# TERMINAL_ENV=docker left in ~/.hermes/.env (e.g. written by an older
# `hermes setup` before the user switched terminal.backend in config.yaml)
# silently wins again on every reload. Startup launchers bridge
# config→env once, but long-lived processes (gateway per-turn reload,
# cron standalone runs) call load_hermes_dotenv() repeatedly and used to
# flip the effective backend back to the stale .env value mid-session
# (#29186, #67323). Re-apply config.yaml's explicit terminal keys last so
# the documented config path always wins. Runs after _apply_managed_env()
# so the merged config (which already carries the managed overlay) is
# what lands in the env.
# config.yaml owns terminal.* settings, but the override=True loads above let a stale
# TERMINAL_ENV=docker in ~/.hermes/.env (older `hermes setup`) win on every reload; long-lived
# processes (gateway per-turn reload, cron) call this repeatedly and used to flip the backend
# back mid-session. Re-apply config.yaml's explicit terminal keys LAST, after
# _apply_managed_env(), so the merged config (with the managed overlay) lands in the env.
_reapply_terminal_config_bridge(home_path)
return loaded
def _reapply_terminal_config_bridge(home_path: Path) -> None:
"""Re-assert config.yaml's explicit ``terminal.*`` keys over reloaded .env.
Delegates to ``hermes_cli.config.apply_terminal_config_to_env`` — the single shared bridge (same
one terminal_tool's fallback and the TUI/ dashboard launchers use) — so key coverage, explicit-
keys-only override semantics, cwd placeholder handling, and the managed-scope overlay can't
drift from the other bridge sites.
"""
"""Re-assert config.yaml's explicit ``terminal.*`` keys over reloaded .env via the single shared
bridge ``hermes_cli.config.apply_terminal_config_to_env`` (also used by terminal_tool's fallback
and the TUI/dashboard launchers) so coverage, explicit-keys-only semantics, cwd placeholders and
the managed-scope overlay can't drift between sites."""
try:
if Path(home_path).resolve() != _process_hermes_home().resolve():
return
@@ -547,12 +424,9 @@ def _reapply_terminal_config_bridge(home_path: Path) -> None:
def _apply_managed_env() -> None:
"""Apply the managed-scope .env last, with override, so it beats user/shell.
This does NOT prevent the agent from later mutating ``os.environ`` in-process or ``export``-ing
in a subprocess shell; that hard boundary is a documented v2 item (design §8.1). v1 relies on
filesystem permissions only.
Fail-open: a missing managed dir or .env is the common case and a no-op; any error here is
swallowed so managed scope can never block startup.
Does NOT stop the agent from later mutating ``os.environ`` or ``export``-ing in a subprocess
shell — that hard boundary is a documented v2 item (design §8.1); v1 relies on filesystem
permissions. Fail-open: missing dir/.env is the common no-op; errors never block startup.
"""
try:
from hermes_cli import managed_scope
@@ -572,14 +446,11 @@ def _apply_managed_env() -> None:
def _apply_external_secret_sources(home_path: Path) -> None:
"""Pull secrets from every enabled external source into env.
Runs AFTER dotenv loads so .env values are visible (sources use them to locate bootstrap tokens)
but BEFORE the rest of Hermes reads ``os.environ`` for credentials. Any failure here is logged
and swallowed — external secret sources must never block startup.
The heavy lifting (source ordering, mapped-beats-bulk precedence, first-claim-wins conflict
handling, override semantics, provenance) lives in ``agent.secret_sources.registry.apply_all``;
this wrapper owns the once-per-HERMES_HOME guard, the post-apply ASCII sanitization sweep, the
``_SECRET_SOURCES`` provenance map that UI surfaces read, and the startup status lines.
Runs AFTER dotenv loads (sources use .env values to locate bootstrap tokens) but BEFORE Hermes
reads ``os.environ`` for credentials; failures are swallowed — sources must never block startup.
Ordering, mapped-beats-bulk precedence, first-claim-wins conflicts and provenance live in
``agent.secret_sources.registry.apply_all``; this wrapper owns the once-per-HERMES_HOME guard,
the post-apply ASCII sweep, the ``_SECRET_SOURCES`` map UI surfaces read, and status lines.
"""
home_key = str(Path(home_path).resolve())
if home_key in _APPLIED_HOMES:
@@ -588,31 +459,19 @@ def _apply_external_secret_sources(home_path: Path) -> None:
try:
cfg = _load_secrets_config(home_path)
except Exception: # noqa: BLE001 — config errors must not block startup
# Deliberately NOT marked applied: a malformed config.yaml would
# otherwise permanently disable secret loading for this process
# even after the user fixes the file (#40597).
# Deliberately NOT marked applied: a malformed config.yaml would otherwise permanently
# disable secret loading for this process even after the user fixes the file.
return
if not cfg:
# No secrets section (or everything disabled at parse level). Not
# marked applied either — the re-parse is a cheap fast_safe_load and
# leaving the home unmarked lets a process pick up a config change
# on its next load_hermes_dotenv() call instead of never.
# No secrets section. Not marked applied either — the re-parse is a cheap fast_safe_load and
# an unmarked home picks up a config change on the next load_hermes_dotenv() instead of never.
return
# Defer the registry import until we know a secrets source is enabled —
# agent.secret_sources.bitwarden eagerly loads cryptography._rust.pyd,
# which causes the Windows updater to self-lock before its preflight
# (the updater itself maps the .pyd before the dependency sync runs).
# A config with no enabled sources costs one dict scan; a config with
# enabled sources pays the crypto load exactly once, on demand.
# NOTE: only keys that smell like a real secret source trigger the import —
# a generic dict entry must not force crypto load on every hermes launch.
# We whitelist by *shape* (source dict with enabled flag) rather than
# hardcoding names, so plugin/test sources pass through unknown keys.
any_enabled = any(
isinstance(v, dict) and v.get("enabled") is True
for v in cfg.values()
)
# Defer the registry import until a source is enabled — agent.secret_sources.bitwarden eagerly
# loads cryptography._rust.pyd, which makes the Windows updater self-lock before its preflight.
# Detect by *shape* (dict with enabled flag), not hardcoded names, so plugin/test sources pass
# and a generic dict entry never forces the crypto load on every launch.
any_enabled = any(isinstance(v, dict) and v.get("enabled") is True for v in cfg.values())
if not any_enabled:
return
@@ -627,27 +486,18 @@ def _apply_external_secret_sources(home_path: Path) -> None:
return
if not report.sources:
# Config parsed but no source is enabled: keep retrying cheaply
# (no fetch happens for disabled sources) so flipping a source on
# mid-process takes effect on the next call.
# No source enabled: keep retrying cheaply so flipping one on mid-process takes effect.
return
# A real fetch attempt happened (success OR error). Mark the home now
# so the 3-5 import-time load_hermes_dotenv() calls per startup don't
# re-fetch / re-print — error retries within one process are opt-in via
# reset_secret_source_cache(). Marking AFTER the attempt (not before,
# see #40597) is what lets the earlier failure paths stay retryable.
# A real fetch attempt happened (success OR error). Mark the home now so the 3-5 import-time
# calls per startup don't re-fetch / re-print — error retries are opt-in via
# reset_secret_source_cache(). Marking AFTER the attempt keeps the earlier failure paths retryable.
_APPLIED_HOMES.add(home_key)
if report.applied_any:
# Re-run the ASCII sanitization pass: vault values are
# user-supplied and might have the same copy-paste corruption as
# a manually edited .env (see #6843).
# Vault values are user-supplied and may carry the same copy-paste corruption as .env.
_sanitize_loaded_credentials()
# Remember where each var came from so setup / `hermes model`
# flows can label detected credentials with "(from Bitwarden)" /
# "(from 1Password)" — otherwise users see "credentials ✓" with
# no hint the value came from a vault rather than .env.
# Provenance lets setup / `hermes model` label credentials "(from Bitwarden)" etc.
values: dict[str, str] = {}
for name, applied in report.provenance.items():
_SECRET_SOURCES[name] = applied.source
@@ -664,9 +514,7 @@ def _apply_external_secret_sources(home_path: Path) -> None:
)
if src.result.error:
print(f" {src.label}: {src.result.error}", file=sys.stderr)
hint = _remediation_hint(
src.name, src.result.error_kind, cfg, scope=home_key
)
hint = _remediation_hint(src.name, src.result.error_kind, cfg, scope=home_key)
if hint:
print(f" {src.label}: → {hint}", file=sys.stderr)
for warn in src.result.warnings:
@@ -675,18 +523,9 @@ def _apply_external_secret_sources(home_path: Path) -> None:
print(f" Secret sources: {conflict}", file=sys.stderr)
def _remediation_hint(
source_name: str,
error_kind,
secrets_cfg: dict,
*,
scope: str | None = None,
) -> str:
"""Ask the failed source for its one-line fix-it hint.
Defensive wrapper: remediation() is a pure mapping and shouldn't raise, but a plugin source
could — and startup must never break on a status line.
"""
def _remediation_hint(source_name: str, error_kind, secrets_cfg: dict, *, scope: str | None = None) -> str:
"""The failed source's one-line fix-it hint. remediation() is a pure mapping and shouldn't raise,
but a plugin source could — and startup must never break on a status line."""
try:
from agent.secret_sources.registry import get_source
@@ -701,21 +540,15 @@ def _remediation_hint(
def _load_secrets_config(home_path: Path) -> dict:
"""Read just the ``secrets:`` section out of config.yaml.
Imported lazily and isolated from the main config loader so a malformed config can't take down
dotenv loading entirely.
"""
"""Read just the ``secrets:`` section of config.yaml, isolated from the main config loader so a
malformed config can't take down dotenv loading entirely."""
config_path = home_path / "config.yaml"
if not config_path.exists():
return {}
# Prefer the shared (mtime, size)-keyed raw-config cache — this is the
# first config.yaml read in a normal `hermes` startup, so populating the
# shared cache here lets main.py's early bridge and hermes_logging reuse
# the same parse (one parse per process instead of 3-4). Falls back to a
# direct isolated parse if the shared reader is unavailable, preserving
# the "malformed config can't take down dotenv loading" property (the
# shared reader also swallows parse errors and returns {}).
# Prefer the shared (mtime, size)-keyed raw-config cache: this is the first config.yaml read in
# a normal startup, so populating it lets main.py's early bridge and hermes_logging reuse one
# parse instead of 3-4. Falls back to a direct isolated parse (the shared reader also swallows
# parse errors -> {}).
if home_path == _process_hermes_home():
try:
from hermes_cli.config import read_raw_config
hermes_cli/hooks.py
+50 -136
View File
@@ -10,12 +10,10 @@ from typing import Any, Dict, List
def hooks_command(args) -> None:
"""Entry point for ``hermes hooks`` — dispatches to the requested action."""
sub = getattr(args, "hooks_action", None)
if not sub:
print("Usage: hermes hooks {list|test|revoke|doctor}")
print("Run 'hermes hooks --help' for details.")
return
handler = _ACTIONS.get(sub)
if handler is None:
print(f"Unknown hooks subcommand: {sub}")
@@ -48,22 +46,18 @@ def _cmd_list(_args) -> None:
by_event: Dict[str, List] = {}
for spec in specs:
by_event.setdefault(spec.event, []).append(spec)
approved = {
(e.get("event"), e.get("command"))
for e in shell_hooks.load_allowlist().get("approvals", [])
if isinstance(e, dict)
}
print(f"Configured shell hooks ({len(specs)} total):\n")
for event in sorted(by_event):
print(f" [{event}]")
for spec in by_event[event]:
is_approved = (spec.event, spec.command) in approved
status = "✓ allowed" if is_approved else "✗ not allowlisted"
print(f" - {spec.command}{_matcher_part(spec)} (timeout={spec.timeout}s, {status})")
entry = shell_hooks.allowlist_entry_for(spec.event, spec.command) if is_approved else None
if entry and entry.get("approved_at"):
print(f" approved_at: {entry['approved_at']}")
@@ -91,8 +85,8 @@ def _matcher_part(obj) -> str:
def _mtime_drift(shell_hooks, spec, entry) -> tuple[bool | None, str, str]:
"""Return ``(drift, mtime_at, mtime_now)``: drift is True if the script changed since approval,
False if unchanged, None when either mtime is unknown."""
"""``(drift, mtime_at, mtime_now)``: True if the script changed since approval, False if
unchanged, None when either mtime is unknown."""
mtime_now = shell_hooks.script_mtime_iso(spec.command)
mtime_at = entry.get("script_mtime_at_approval")
if not (mtime_now and mtime_at):
@@ -106,120 +100,65 @@ def _mtime_drift(shell_hooks, spec, entry) -> tuple[bool | None, str, str]:
# test
# ---------------------------------------------------------------------------
# Synthetic kwargs matching the real invoke_hook() call sites — these are
# passed verbatim to agent.shell_hooks.run_once(), which routes them through
# the same _serialize_payload() that production firings use. That way the
# stdin a script sees under `hermes hooks test` and `hermes hooks doctor`
# is identical in shape to what it will see at runtime.
# Synthetic kwargs matching the real invoke_hook() call sites. They go verbatim to
# agent.shell_hooks.run_once() -> the same _serialize_payload() production uses, so the
# stdin a script sees under `hooks test` / `hooks doctor` has the runtime shape.
_DEFAULT_PAYLOADS = {
"pre_tool_call": {
"tool_name": "terminal",
"args": {"command": "echo hello"},
"session_id": "test-session",
"task_id": "test-task",
"tool_call_id": "test-call",
"tool_name": "terminal", "args": {"command": "echo hello"},
"session_id": "test-session", "task_id": "test-task", "tool_call_id": "test-call",
},
"post_tool_call": {
"tool_name": "terminal",
"args": {"command": "echo hello"},
"session_id": "test-session",
"task_id": "test-task",
"tool_call_id": "test-call",
"result": '{"output": "hello"}',
"duration_ms": 42,
"tool_name": "terminal", "args": {"command": "echo hello"},
"session_id": "test-session", "task_id": "test-task", "tool_call_id": "test-call",
"result": '{"output": "hello"}', "duration_ms": 42,
},
"pre_llm_call": {
"session_id": "test-session",
"user_message": "What is the weather?",
"conversation_history": [],
"is_first_turn": True,
"model": "gpt-4",
"platform": "cli",
},
"post_llm_call": {
"session_id": "test-session",
"model": "gpt-4",
"platform": "cli",
"session_id": "test-session", "user_message": "What is the weather?",
"conversation_history": [], "is_first_turn": True, "model": "gpt-4", "platform": "cli",
},
"post_llm_call": {"session_id": "test-session", "model": "gpt-4", "platform": "cli"},
"pre_verify": {
"session_id": "test-session",
"platform": "cli",
"model": "gpt-4",
"coding": True,
"attempt": 0,
"final_response": "All done — the change is applied.",
"session_id": "test-session", "platform": "cli", "model": "gpt-4", "coding": True,
"attempt": 0, "final_response": "All done — the change is applied.",
"changed_paths": ["src/app.tsx"],
},
"on_session_start": {"session_id": "test-session"},
"on_session_end": {
"session_id": "test-session",
"task_id": "test-task",
"turn_id": "test-turn",
"completed": True,
"failed": False,
"interrupted": False,
"turn_exit_reason": "text_response(stop)",
"model": "gpt-4",
"platform": "cli",
"session_id": "test-session", "task_id": "test-task", "turn_id": "test-turn",
"completed": True, "failed": False, "interrupted": False,
"turn_exit_reason": "text_response(stop)", "model": "gpt-4", "platform": "cli",
},
"on_session_finalize": {"session_id": "test-session"},
"on_session_reset": {"session_id": "test-session"},
"pre_api_request": {
"session_id": "test-session",
"task_id": "test-task",
"platform": "cli",
"model": "claude-sonnet-4-6",
"provider": "anthropic",
"base_url": "https://api.anthropic.com",
"api_mode": "anthropic_messages",
"api_call_count": 1,
"message_count": 4,
"tool_count": 12,
"approx_input_tokens": 2048,
"request_char_count": 8192,
"max_tokens": 4096,
"session_id": "test-session", "task_id": "test-task", "platform": "cli",
"model": "claude-sonnet-4-6", "provider": "anthropic",
"base_url": "https://api.anthropic.com", "api_mode": "anthropic_messages",
"api_call_count": 1, "message_count": 4, "tool_count": 12,
"approx_input_tokens": 2048, "request_char_count": 8192, "max_tokens": 4096,
},
"post_api_request": {
"session_id": "test-session",
"task_id": "test-task",
"platform": "cli",
"model": "claude-sonnet-4-6",
"provider": "anthropic",
"base_url": "https://api.anthropic.com",
"api_mode": "anthropic_messages",
"api_call_count": 1,
"api_duration": 1.234,
"started_at": 1756000000.0,
"ended_at": 1756000001.234,
"first_chunk_at": 1756000000.512,
"finish_reason": "stop",
"message_count": 4,
"response_model": "claude-sonnet-4-6",
"session_id": "test-session", "task_id": "test-task", "platform": "cli",
"model": "claude-sonnet-4-6", "provider": "anthropic",
"base_url": "https://api.anthropic.com", "api_mode": "anthropic_messages",
"api_call_count": 1, "api_duration": 1.234,
"started_at": 1756000000.0, "ended_at": 1756000001.234, "first_chunk_at": 1756000000.512,
"finish_reason": "stop", "message_count": 4, "response_model": "claude-sonnet-4-6",
"usage": {"input_tokens": 2048, "output_tokens": 512},
"assistant_content_chars": 1200,
"assistant_tool_call_count": 0,
# Per-advisor metrics on a MoA turn, None otherwise. MoA returns only
# the aggregator's response, so without this an observer cannot see the
# fan-out or price it at each advisor's own model.
"assistant_content_chars": 1200, "assistant_tool_call_count": 0,
# Per-advisor metrics on a MoA turn, None otherwise: MoA returns only the aggregator's
# response, so without this an observer cannot see or price the fan-out.
"moa_references": None,
},
"subagent_stop": {
"parent_session_id": "parent-sess",
"child_role": None,
"child_summary": "Synthetic summary for hooks test",
"child_status": "completed",
"tool_call_history": [
{
"tool_name": "write_file",
"tool_input": {
"argument_keys": ["content", "path"],
"targets": {"path": "/tmp/report.txt"},
},
"input_bytes": 128,
"output_bytes": 32,
"status": "ok",
}
],
"parent_session_id": "parent-sess", "child_role": None,
"child_summary": "Synthetic summary for hooks test", "child_status": "completed",
"tool_call_history": [{
"tool_name": "write_file",
"tool_input": {"argument_keys": ["content", "path"], "targets": {"path": "/tmp/report.txt"}},
"input_bytes": 128, "output_bytes": 32, "status": "ok",
}],
"duration_ms": 1234,
},
}
@@ -236,14 +175,11 @@ def _cmd_test(args) -> None:
print(f"Valid events: {', '.join(sorted(VALID_HOOKS))}")
return
# Synthetic kwargs in the same shape invoke_hook() would pass. Merged
# with --for-tool (overrides tool_name) and --payload-file (extra kwargs).
# Synthetic kwargs merged with --for-tool (overrides tool_name) and --payload-file (extra kwargs).
payload = dict(_DEFAULT_PAYLOADS.get(event, {"session_id": "test-session"}))
for_tool = getattr(args, "for_tool", None)
if for_tool:
payload["tool_name"] = for_tool
if getattr(args, "payload_file", None):
try:
custom = json.loads(Path(args.payload_file).read_text(encoding="utf-8"))
@@ -256,10 +192,8 @@ def _cmd_test(args) -> None:
return
specs = [s for s in shell_hooks.iter_configured_hooks(load_config()) if s.event == event]
if for_tool:
specs = [s for s in specs if s.event not in {"pre_tool_call", "post_tool_call"} or s.matches_tool(for_tool)]
if not specs:
print(f"No shell hooks configured for event: {event}")
if for_tool:
@@ -269,8 +203,7 @@ def _cmd_test(args) -> None:
print(f"Firing {len(specs)} hook(s) for event '{event}':\n")
for spec in specs:
print(f" → {spec.command}")
result = shell_hooks.run_once(spec, payload)
_print_run_result(result)
_print_run_result(shell_hooks.run_once(spec, payload))
print()
@@ -281,14 +214,11 @@ def _print_run_result(result: Dict[str, Any]) -> None:
if result.get("timed_out"):
print(f" ✗ timed out after {result['elapsed_seconds']}s")
return
print(f" exit={result.get('returncode')} elapsed={result.get('elapsed_seconds', 0)}s")
for stream in ("stdout", "stderr"):
text = (result.get(stream) or "").strip()
if text:
print(f" {stream}: {_truncate(text, 400)}")
parsed = result.get("parsed")
if parsed:
print(f" parsed (Hermes wire shape): {json.dumps(parsed)}")
@@ -301,7 +231,7 @@ def _truncate(s: str, n: int) -> str:
# ---------------------------------------------------------------------------
# revoke
# revoke / doctor
# ---------------------------------------------------------------------------
def _cmd_revoke(args) -> None:
@@ -318,34 +248,25 @@ def _cmd_revoke(args) -> None:
)
# ---------------------------------------------------------------------------
# doctor
# ---------------------------------------------------------------------------
def _cmd_doctor(_args) -> None:
from hermes_cli.config import load_config
from agent import shell_hooks
specs = shell_hooks.iter_configured_hooks(load_config())
if not specs:
print("No shell hooks configured — nothing to check.")
return
print(f"Checking {len(specs)} configured shell hook(s)...\n")
problems = 0
for spec in specs:
print(f" [{spec.event}] {spec.command}")
problems += _doctor_one(spec, shell_hooks)
print()
print(f"{problems} issue(s) found. Fix before relying on these hooks." if problems else "All shell hooks look healthy.")
def _doctor_one(spec, shell_hooks) -> int:
problems = 0
# 1. Script exists and is executable
if shell_hooks.script_is_executable(spec.command):
print(" ✓ script exists and is executable")
@@ -353,7 +274,6 @@ def _doctor_one(spec, shell_hooks) -> int:
problems += 1
print(" ✗ script missing or not executable "
"(chmod +x the file, or fix the path)")
# 2. Allowlist status
entry = shell_hooks.allowlist_entry_for(spec.event, spec.command)
if entry:
@@ -362,7 +282,6 @@ def _doctor_one(spec, shell_hooks) -> int:
problems += 1
print(" ✗ not allowlisted — hook will NOT fire at runtime "
"(run with --accept-hooks once, or confirm at the TTY prompt)")
# 3. Mtime drift
if entry and entry.get("script_mtime_at_approval"):
drift, mtime_at, mtime_now = _mtime_drift(shell_hooks, spec, entry)
@@ -373,19 +292,15 @@ def _doctor_one(spec, shell_hooks) -> int:
f"then `hermes hooks revoke` + re-approve to refresh")
elif drift is False:
print(" ✓ script unchanged since approval")
# 4. Produces valid JSON for a synthetic payload — only when the entry
# is already allowlisted. Otherwise `hermes hooks doctor` would execute
# every script listed in a freshly-pulled config before the user has
# reviewed them, which directly contradicts the documented workflow
# ("spot newly-added hooks *before they register*").
# 4. JSON smoke test on a synthetic payload — ONLY when already allowlisted. Otherwise doctor
# would execute every script in a freshly-pulled config before the user reviewed it, which
# contradicts the documented workflow ("spot newly-added hooks *before they register*").
if not entry:
print(" ℹ skipped JSON smoke test — not allowlisted yet. "
"Approve the hook first (via TTY prompt or --accept-hooks), "
"then re-run `hermes hooks doctor`.")
elif shell_hooks.script_is_executable(spec.command):
payload = _DEFAULT_PAYLOADS.get(spec.event, {"extra": {}})
result = shell_hooks.run_once(spec, payload)
result = shell_hooks.run_once(spec, _DEFAULT_PAYLOADS.get(spec.event, {"extra": {}}))
if result.get("timed_out"):
problems += 1
print(f" ✗ timed out after {result['elapsed_seconds']}s "
@@ -397,7 +312,10 @@ def _doctor_one(spec, shell_hooks) -> int:
rc = result.get("returncode")
elapsed = result.get("elapsed_seconds", 0)
stdout = (result.get("stdout") or "").strip()
if stdout:
if not stdout:
print(f" ✓ ran clean with empty stdout "
f"(exit={rc}, {elapsed}s) — hook is observer-only")
else:
try:
json.loads(stdout)
print(f" ✓ produced valid JSON on synthetic payload "
@@ -406,10 +324,6 @@ def _doctor_one(spec, shell_hooks) -> int:
problems += 1
print(f" ✗ stdout was not valid JSON (exit={rc}, "
f"{elapsed}s): {_truncate(stdout, 120)}")
else:
print(f" ✓ ran clean with empty stdout "
f"(exit={rc}, {elapsed}s) — hook is observer-only")
return problems
hermes_cli/inventory.py
+176 -358
View File
@@ -1,5 +1,5 @@
"""Provider/model inventory context — shared substrate for the dashboard ``/api/model/options``, the
TUI ``model.options``/``model.save_key`` JSON-RPC handlers, and the interactive picker.
"""Provider/model inventory context — shared substrate for the dashboard ``/api/model/options``,
the TUI ``model.options``/``model.save_key`` JSON-RPC handlers, and the interactive picker.
"""
from __future__ import annotations
@@ -8,14 +8,12 @@ from dataclasses import dataclass, replace
from typing import Any, Optional
# ─── Public types ───────────────────────────────────────────────────────
@dataclass(frozen=True)
class ConfigContext:
"""Snapshot of the model + provider config every inventory caller needs. Built once via
``load_picker_context()``; the TUI overlays live agent state via ``with_overrides()`` before
passing through.
"""Snapshot of the model + provider config every inventory caller needs.
Built once via ``load_picker_context()``; the TUI overlays live agent state via
``with_overrides()`` before passing through.
"""
current_provider: str
@@ -26,43 +24,30 @@ class ConfigContext:
excluded_providers: list = None
def with_overrides(
self,
*,
current_provider: Optional[str] = None,
current_model: Optional[str] = None,
self, *, current_provider: Optional[str] = None, current_model: Optional[str] = None,
current_base_url: Optional[str] = None,
) -> "ConfigContext":
"""Return a copy with truthy overrides applied.
Truthy-only because the TUI reads agent attributes that may be empty strings before an agent
is spawned — empties must NOT clobber the disk-config values.
Truthy-only: the TUI reads agent attributes that may be empty strings before an agent is
spawned — empties must NOT clobber the disk-config values.
"""
kw = {
k: v
for k, v in (
("current_provider", current_provider),
("current_model", current_model),
("current_base_url", current_base_url),
)
if v
}
overrides = (("current_provider", current_provider), ("current_model", current_model),
("current_base_url", current_base_url))
kw = {k: v for k, v in overrides if v}
return replace(self, **kw) if kw else self
def load_picker_context() -> ConfigContext:
"""Load the disk-config snapshot every consumer needs."""
from hermes_cli.config import (
coerce_provider_id,
get_compatible_custom_providers,
load_config,
stringify_provider_map,
coerce_provider_id, get_compatible_custom_providers, load_config, stringify_provider_map,
)
cfg = load_config()
model_cfg = cfg.get("model", {})
if isinstance(model_cfg, dict):
# PyYAML parses unquoted scalars as int (`provider: 2070`). Keep these
# as strings so picker/options paths never call `.strip()` on an int.
# PyYAML parses unquoted scalars as int (`provider: 2070`); keep strings so
# picker/options paths never call `.strip()` on an int.
current_model = str(model_cfg.get("default", model_cfg.get("name", "")) or "")
current_provider = coerce_provider_id(model_cfg.get("provider", ""))
current_base_url = str(model_cfg.get("base_url", "") or "")
@@ -73,9 +58,7 @@ def load_picker_context() -> ConfigContext:
current_base_url = ""
excluded = cfg.get("model_catalog", {}).get("excluded_providers") or []
return ConfigContext(
current_provider=current_provider,
current_model=current_model,
current_base_url=current_base_url,
current_provider=current_provider, current_model=current_model, current_base_url=current_base_url,
user_providers=stringify_provider_map(cfg.get("providers")),
custom_providers=get_compatible_custom_providers(cfg),
excluded_providers=excluded if isinstance(excluded, list) else [],
@@ -94,60 +77,40 @@ def _without_slug(rows: list[dict], slug: str) -> list[dict]:
def build_models_payload(
ctx: ConfigContext,
*,
explicit_only: bool = False,
include_unconfigured: bool = False,
picker_hints: bool = False,
canonical_order: bool = False,
pricing: bool = False,
capabilities: bool = False,
featured: bool = False,
force_fresh_nous_tier: bool = False,
refresh: bool = False,
probe_custom_providers: bool = True,
probe_current_custom_provider: bool = False,
for_picker: bool = False,
max_models: int | None = None,
ctx: ConfigContext, *, explicit_only: bool = False, include_unconfigured: bool = False,
picker_hints: bool = False, canonical_order: bool = False, pricing: bool = False,
capabilities: bool = False, featured: bool = False, force_fresh_nous_tier: bool = False,
refresh: bool = False, probe_custom_providers: bool = True, probe_current_custom_provider: bool = False,
for_picker: bool = False, max_models: int | None = None,
) -> dict:
"""Build the ``{providers, model, provider}`` shape every consumer needs from a single substrate call.
"""Build the ``{providers, model, provider}`` shape every consumer needs.
Flags: - ``explicit_only``: keep only providers the user explicitly configured (current
provider, providers from config, or providers backed by provider-specific env vars). This hides
ambient / auto-seeded credentials from desktop chat pickers.
``explicit_only`` keeps only providers the user explicitly configured (current provider,
config providers, provider-specific env vars) — hides ambient/auto-seeded credentials from
desktop chat pickers.
"""
from hermes_cli.model_switch import list_authenticated_providers
rows = list_authenticated_providers(
current_provider=ctx.current_provider,
current_base_url=ctx.current_base_url,
current_model=ctx.current_model,
user_providers=ctx.user_providers,
custom_providers=ctx.custom_providers,
force_fresh_nous_tier=force_fresh_nous_tier,
max_models=max_models,
refresh=refresh,
probe_custom_providers=probe_custom_providers,
probe_current_custom_provider=probe_current_custom_provider,
for_picker=for_picker,
current_provider=ctx.current_provider, current_base_url=ctx.current_base_url,
current_model=ctx.current_model, user_providers=ctx.user_providers,
custom_providers=ctx.custom_providers, force_fresh_nous_tier=force_fresh_nous_tier,
max_models=max_models, refresh=refresh, probe_custom_providers=probe_custom_providers,
probe_current_custom_provider=probe_current_custom_provider, for_picker=for_picker,
excluded_providers=ctx.excluded_providers or [],
)
# Managed local runtime: staged GGUFs are selectable like any provider's
# models. list_authenticated_providers can't know about them (no
# credential, no custom_providers entry — the credential is
# reachability), so inject the row here where every picker surface
# inherits it. Present whenever models are staged; picking one routes
# through the llamacpp alias -> managed/detected server resolution.
# Managed local runtime: staged GGUFs are selectable like any provider's models.
# list_authenticated_providers can't know about them (no credential, no custom_providers
# entry — the credential is reachability), so inject the row here where every picker
# surface inherits it. Picking one routes through the llamacpp alias -> managed server.
local_row = _local_runtime_row(ctx)
if local_row is not None:
rows = _without_slug(rows, "llamacpp") + [local_row]
# A live session on the managed server reports provider "custom"
# (the resolution seam's generic label for a raw base_url), which
# would otherwise materialize a duplicate "Custom endpoint" row
# carrying the same staged models and stealing the checkmark. The
# Local row owns the managed server's identity — drop custom rows
# that point at the managed endpoint.
# A live session on the managed server reports provider "custom" (the resolution seam's
# generic label for a raw base_url), which would materialize a duplicate "Custom endpoint"
# row carrying the same staged models and stealing the checkmark. The Local row owns the
# managed server's identity — drop custom rows that point at the managed endpoint.
if local_row.get("is_current"):
staged = set(local_row["models"])
@@ -163,28 +126,21 @@ def build_models_payload(
if explicit_only:
rows = _filter_explicit_provider_rows(rows, ctx)
# Desktop chat pickers request the explicit subset without the full
# unconfigured provider universe. If the configured current provider
# has lost its credential, list_authenticated_providers() omits it;
# keep that one row visible so the UI can show the saved selection and
# a re-auth affordance instead of appearing to jump to another provider.
# Exception: a "custom" current whose endpoint is the managed local
# server is already represented (with the checkmark) by the Local row
# — the skeleton would resurrect the duplicate the dedup above removed.
# Desktop chat pickers request the explicit subset without the full unconfigured
# universe. If the current provider lost its credential, list_authenticated_providers()
# omits it; keep that one row so the UI shows the saved selection and a re-auth
# affordance instead of appearing to jump to another provider. Exception: a "custom"
# current whose endpoint is the managed local server is already represented (with the
# checkmark) by the Local row — the skeleton would resurrect the duplicate removed above.
_local_owns_current = bool(local_row and local_row.get("is_current")
and (ctx.current_provider or "").lower() == "custom")
if not _local_owns_current:
rows = list(rows) + _append_unconfigured_rows(rows, ctx, current_only=True)
# --- Deduplicate: remove models from aggregators that overlap with
# user-defined providers. When a local proxy (e.g. litellm-proxy)
# serves a model whose name also appears in an aggregator's curated
# catalog, the picker would show the model under both providers.
# Selecting it from the aggregator row sets model.provider to the
# aggregator (e.g. openrouter) instead of the user's proxy — silently
# breaking the call. Filtering at the payload level keeps the
# aggregator rows honest: they only show models the user can't get
# from a more-specific provider. (#45954)
# Dedup aggregator models against user-defined providers: a local proxy serving a model that
# also appears in an aggregator's catalog would show under both, and selecting the aggregator
# row sets model.provider to the aggregator — silently breaking the call. Aggregator rows
# only show models the user can't get from a more-specific provider.
_strip_aggregator_overlaps(rows)
if include_unconfigured:
@@ -201,22 +157,18 @@ def build_models_payload(
_apply_featured(rows)
_apply_custom_aliases(rows)
return {
"providers": rows,
"model": ctx.current_model,
"provider": ctx.current_provider,
}
return {"providers": rows, "model": ctx.current_model, "provider": ctx.current_provider}
def _strip_aggregator_overlaps(rows: list[dict]) -> None:
"""Drop models from TRUE routing aggregators (OpenRouter, custom:* proxies) that a user-defined
provider also serves, so the picker never lists them under both (#45954).
"""Drop models from TRUE routing aggregators (OpenRouter, custom:* proxies) that a
user-defined provider also serves, so the picker never lists them under both.
A user's own configured provider is never an "aggregator duplicate" of itself: user_models is
built from these very rows, and is_routing_aggregator() reports True for every custom:* slug —
without that guard the dedup would empty a user-defined custom provider's row. Flat-namespace
resellers (opencode-go / opencode-zen) serve every listed model first-party and must keep models
that a user's proxy happens to share a name with (#47077).
built from these very rows and is_routing_aggregator() is True for every custom:* slug, so
without the is_user_defined guard the dedup would empty a user-defined custom row.
Flat-namespace resellers (opencode-go / opencode-zen) serve every model first-party and keep
models a user's proxy happens to share a name with.
"""
try:
from hermes_cli.providers import is_routing_aggregator
@@ -240,34 +192,19 @@ def _strip_aggregator_overlaps(rows: list[dict]) -> None:
def build_model_options_payload(
ctx: ConfigContext,
*,
explicit_only: bool = False,
include_unconfigured: bool = False,
ctx: ConfigContext, *, explicit_only: bool = False, include_unconfigured: bool = False,
refresh: bool = False,
) -> dict:
"""Build the shared API-server/dashboard/TUI model-options payload.
"""Shared API-server/dashboard/TUI model-options payload with the safe probe policy.
This wraps ``build_models_payload`` with the stable picker shape and the safe custom-provider
probe policy used for normal GUI/TUI opens:
- normal open: probe only the current custom provider so offline saved endpoints do not block
the picker - explicit refresh: probe every custom provider while busting the model cache so live
catalogs repopulate fully
Normal open: probe only the current custom provider so offline saved endpoints don't block
the picker. Explicit refresh: probe every custom provider and bust the model cache.
"""
refresh = bool(refresh)
return build_models_payload(
ctx,
explicit_only=bool(explicit_only),
include_unconfigured=bool(include_unconfigured),
picker_hints=True,
canonical_order=True,
pricing=True,
capabilities=True,
featured=True,
refresh=refresh,
probe_custom_providers=refresh,
probe_current_custom_provider=not refresh,
ctx, explicit_only=bool(explicit_only), include_unconfigured=bool(include_unconfigured),
picker_hints=True, canonical_order=True, pricing=True, capabilities=True, featured=True,
refresh=refresh, probe_custom_providers=refresh, probe_current_custom_provider=not refresh,
)
@@ -275,50 +212,34 @@ def build_model_options_payload(
def build_aux_picker_rows(
*,
current_provider: str = "",
current_model: str = "",
current_base_url: str = "",
*, current_provider: str = "", current_model: str = "", current_base_url: str = "",
max_models: int | None = None,
) -> list[dict]:
"""Provider rows for any auxiliary-task picker (vision, compression, …).
- user-defined ``providers:`` and saved ``custom_providers:`` entries -
``model_catalog.excluded_providers`` honoured, matching ``/model`` - exhausted-credential-pool
providers stay visible (``for_picker``) - the active custom endpoint is probed, offline saved
ones are not, so the picker never blocks on a dead local server
The virtual ``moa`` row is excluded: auxiliary tasks must not run the MoA reference fan-out, and
``auxiliary_client`` unwraps a ``moa`` provider to its aggregator slot anyway (see
``_resolve_auto``), so offering it here would be a choice silently rewritten behind the user's
back.
Honours ``model_catalog.excluded_providers`` like ``/model``; exhausted-credential-pool
providers stay visible (``for_picker``); only the active custom endpoint is probed so the
picker never blocks on a dead local server. The virtual ``moa`` row is excluded: auxiliary
tasks must not run the MoA fan-out, and ``auxiliary_client`` unwraps ``moa`` to its aggregator
anyway (``_resolve_auto``), so offering it would be a choice silently rewritten.
"""
ctx = load_picker_context().with_overrides(
current_provider=current_provider,
current_model=current_model,
current_base_url=current_base_url,
current_provider=current_provider, current_model=current_model, current_base_url=current_base_url,
)
rows = build_models_payload(
ctx,
for_picker=True,
probe_custom_providers=False,
probe_current_custom_provider=True,
ctx, for_picker=True, probe_custom_providers=False, probe_current_custom_provider=True,
max_models=max_models,
)["providers"]
return _without_slug(rows, "moa")
def format_aux_picker_entries(
rows: list[dict],
*,
current_provider: str = "",
current_base_url: str = "",
rows: list[dict], *, current_provider: str = "", current_base_url: str = "",
) -> list[tuple[str, str, list[str]]]:
"""Render aux-picker rows as ``(slug, label, models)`` menu entries.
A custom endpoint set via a raw ``base_url`` is "current" only through that URL — never through
a provider slug — so when ``current_base_url`` is set no provider row is marked, matching the
pre-existing behaviour of both call sites.
A custom endpoint set via a raw ``base_url`` is "current" only through that URL, never a
slug — so when ``current_base_url`` is set no provider row is marked.
"""
entries: list[tuple[str, str, list[str]]] = []
current_slug = str(current_provider or "").strip().lower()
@@ -328,11 +249,7 @@ def format_aux_picker_entries(
name = row.get("name") or slug
total = row.get("total_models") or len(row.get("models") or [])
model_hint = f" — {total} models" if total else ""
marker = (
" ← current"
if slug.lower() == current_slug and current_slug and not has_base_url
else ""
)
marker = " ← current" if slug.lower() == current_slug and current_slug and not has_base_url else ""
entries.append((slug, f"{name}{model_hint}{marker}", list(row.get("models") or [])))
return entries
@@ -340,16 +257,13 @@ def format_aux_picker_entries(
def _reasoning_catalog_reader(slug: str):
"""Per-model reasoning-capability reader for aggregators that publish one.
Cache-only — building the picker payload must never block on HTTP. A cold cache warms in the
background so the next open is accurate; until then the model reports no restriction and the UI
offers the full scale.
Cache-only — building the picker must never block on HTTP. A cold cache warms in the
background; until then the model reports no restriction and the UI offers the full scale.
"""
try:
from hermes_cli.models import (
nous_model_reasoning_capabilities,
openrouter_model_reasoning_capabilities,
warm_nous_reasoning_caps_async,
warm_openrouter_reasoning_caps_async,
nous_model_reasoning_capabilities, openrouter_model_reasoning_capabilities,
warm_nous_reasoning_caps_async, warm_openrouter_reasoning_caps_async,
)
except Exception:
return None
@@ -368,10 +282,10 @@ def _reasoning_catalog_reader(slug: str):
def _apply_capabilities(rows: list[dict]) -> None:
"""Attach a ``{model: {fast, reasoning, ...}}`` map to each provider row.
``fast`` mirrors the runtime gate. ``reasoning`` defaults True when the catalog is silent:
the dial is a no-op on models that ignore it, and hiding it from a capable model is worse.
A serving aggregator's per-model detail overrides models.dev (adds ``can_disable_reasoning``).
``supported_efforts`` is deliberately NOT forwarded -- it under-reports levels that work.
``fast`` mirrors the runtime gate. ``reasoning`` defaults True when the catalog is silent: the
dial is a no-op on models that ignore it, and hiding it from a capable model is worse. A serving
aggregator's per-model detail overrides models.dev (adds ``can_disable_reasoning``).
``supported_efforts`` is deliberately NOT forwarded — it under-reports levels that work.
"""
from hermes_cli.models import model_supports_fast_mode
@@ -395,10 +309,7 @@ def _apply_capabilities(rows: list[dict]) -> None:
except Exception:
reasoning = True
entry: dict[str, Any] = {
"fast": bool(model_supports_fast_mode(model)),
"reasoning": reasoning,
}
entry: dict[str, Any] = {"fast": bool(model_supports_fast_mode(model)), "reasoning": reasoning}
if reasoning and read_reasoning_catalog is not None:
try:
@@ -406,9 +317,8 @@ def _apply_capabilities(rows: list[dict]) -> None:
except Exception:
detail = None
if detail and not detail.get("supports_reasoning"):
# For a route it serves, the aggregator's own catalog beats
# models.dev: no reasoning parameter means no reasoning
# controls, so there is no disable to describe either.
# For a route it serves, the aggregator's catalog beats models.dev: no reasoning
# parameter means no reasoning controls, so no disable to describe either.
entry["reasoning"] = False
elif detail:
entry["can_disable_reasoning"] = not detail.get("mandatory")
@@ -418,20 +328,19 @@ def _apply_capabilities(rows: list[dict]) -> None:
row["capabilities"] = caps
# How many models per lab the picker features by default. Aggregator rows keep
# the newest N of each lab (by models.dev release_date) and hide the older tail
# behind search / show-all. 5 keeps a lab's current headliners without letting a
# prolific vendor (OpenAI's gpt-5.6-* family) flood the default view.
# Models per lab the picker features by default. Aggregator rows keep the newest N of each lab
# (by models.dev release_date) and hide the older tail behind search / show-all. 5 keeps a lab's
# current headliners without letting a prolific vendor flood the default view.
_FEATURED_PER_LAB = 5
def _apply_featured(rows: list[dict]) -> None:
"""Attach a ``featured_models`` shortlist to each aggregator provider row.
Aggregators serve many labs, so a flat top-N would drop whole labs; instead surface the
newest ``_FEATURED_PER_LAB`` per vendor, ranked by models.dev ``release_date`` within the
row's own models (never vs. today, so the choice is stable). Ties fall back to the curated
flagship-first order. Non-aggregators get an empty list and keep top-N behaviour.
Aggregators serve many labs, so a flat top-N would drop whole labs; instead surface the newest
``_FEATURED_PER_LAB`` per vendor, ranked by models.dev ``release_date`` within the row's own
models (never vs. today, so the choice is stable). Ties fall back to the curated flagship-first
order. Non-aggregators get an empty list and keep top-N behaviour.
"""
try:
from agent.models_dev import get_model_info
@@ -447,8 +356,7 @@ def _apply_featured(rows: list[dict]) -> None:
for pos, model in enumerate(models):
lab = model.split("/", 1)[0] if "/" in model else ""
if not lab:
# No vendor prefix → single-namespace provider, not an
# aggregator. Bail on the whole row (see below).
# No vendor prefix → single-namespace provider, not an aggregator. Bail on the row.
by_lab = {}
break
date = ""
@@ -457,16 +365,14 @@ def _apply_featured(rows: list[dict]) -> None:
date = getattr(info, "release_date", "") if info else ""
by_lab.setdefault(lab, []).append((pos, date, model))
# A shortlist only makes sense when the row spans several labs.
if len(by_lab) < 2:
row["featured_models"] = []
continue
featured: list[str] = []
for entries in by_lab.values():
# Newest release_date first; earlier list position breaks ties and
# is the sole key when a lab has no dated models (all ""). Keep the
# newest _FEATURED_PER_LAB of each lab.
# Newest release_date first; earlier list position breaks ties and is the sole key
# when a lab has no dated models (all "").
ranked = sorted(entries, key=lambda e: (e[1], -e[0]), reverse=True)
featured.extend(model for _pos, _date, model in ranked[:_FEATURED_PER_LAB])
# Preserve the row's model order for stable rendering.
@@ -477,10 +383,9 @@ def _apply_featured(rows: list[dict]) -> None:
def _apply_custom_aliases(rows: list[dict]) -> None:
"""Attach the accepted identity set to each user-defined provider row.
A session's ``model.options`` reports the canonical ``custom:<key>`` identity (via
``canonical_custom_identity``), while catalog rows carry the bare config key as ``slug``. GUI
pickers compare the two to decide which row is active; exact equality never matches for custom
providers (#87035).
A session's ``model.options`` reports the canonical ``custom:<key>`` identity while catalog rows
carry the bare config key as ``slug``; GUI pickers compare the two to find the active row, and
exact equality never matches for custom providers.
"""
from hermes_cli.providers import custom_provider_aliases
@@ -489,9 +394,7 @@ def _apply_custom_aliases(rows: list[dict]) -> None:
continue
try:
row["aliases"] = sorted(
custom_provider_aliases(
str(row.get("name", "")), str(row.get("slug", ""))
)
custom_provider_aliases(str(row.get("name", "")), str(row.get("slug", "")))
)
except Exception:
continue
@@ -514,26 +417,19 @@ def _canonical_row(entry, cur: str, **extra: Any) -> dict:
from hermes_cli.models import _PROVIDER_LABELS
return {
"slug": entry.slug,
"name": _PROVIDER_LABELS.get(entry.slug, entry.label),
"is_current": entry.slug.lower() == cur,
"is_user_defined": False,
**extra,
"slug": entry.slug, "name": _PROVIDER_LABELS.get(entry.slug, entry.label),
"is_current": entry.slug.lower() == cur, "is_user_defined": False, **extra,
}
def _append_unconfigured_rows(
rows: list[dict],
ctx: ConfigContext,
*,
current_only: bool = False,
rows: list[dict], ctx: ConfigContext, *, current_only: bool = False,
) -> list[dict]:
"""Build fallback rows for canonical providers missing from ``rows``.
Most missing canonical providers become empty setup skeletons. The one exception is the
*current* configured provider: if config.yaml still points at it but credentials are presently
unavailable, keep a visible row carrying the saved model so GUI pickers don't silently snap to
some other provider.
Missing providers become empty setup skeletons, except the *current* configured provider: if
config.yaml still points at it but credentials are unavailable, keep a visible row carrying the
saved model so GUI pickers don't silently snap to another provider.
"""
from hermes_cli.models import CANONICAL_PROVIDERS
@@ -556,13 +452,8 @@ def _append_unconfigured_rows(
"Showing the saved model only."
)
extras.append(_canonical_row(
entry, cur,
models=[cur_model] if cur_model else [],
total_models=1 if cur_model else 0,
source="configured-current",
authenticated=False,
auth_type=auth_type,
key_env=key_env,
entry, cur, models=[cur_model] if cur_model else [], total_models=1 if cur_model else 0,
source="configured-current", authenticated=False, auth_type=auth_type, key_env=key_env,
warning=warning,
))
continue
@@ -573,38 +464,28 @@ def _append_unconfigured_rows(
def _anthropic_oauth_credentials_present() -> bool:
"""True when the user explicitly authenticated Anthropic via OAuth.
Two deliberate flows leave no trace in active_provider / model.provider / API-key env vars:
Hermes' own Anthropic device flow (token in auth.json) and a Claude Code login
(~/.claude/.credentials.json).
Hermes' own device flow (auth.json token) and a Claude Code login (~/.claude/.credentials.json)
leave no trace in active_provider / model.provider / API-key env vars.
"""
try:
from agent.anthropic_adapter import (
read_claude_code_credentials,
read_hermes_oauth_credentials,
)
from agent.anthropic_adapter import read_claude_code_credentials, read_hermes_oauth_credentials
if any(
(read() or {}).get("accessToken")
for read in (read_hermes_oauth_credentials, read_claude_code_credentials)
):
readers = (read_hermes_oauth_credentials, read_claude_code_credentials)
if any((read() or {}).get("accessToken") for read in readers):
return True
except Exception:
return False
# Pool-only OAuth entries (auth.json credential_pool.anthropic) are the
# canonical location for wired tokens and equally deliberate — the
# discovery side accepts them via pool.has_credentials(), so the filter
# must too or those rows are built and then silently dropped. Read-only
# dict access (no load_pool) so a picker open never mutates auth.json.
# Pool-only OAuth entries (auth.json credential_pool.anthropic) are equally deliberate — the
# discovery side accepts them via pool.has_credentials(), so the filter must too or those rows
# are built and then silently dropped. Read-only access (no load_pool) so a picker open never
# mutates auth.json.
try:
from agent.credential_pool import AUTH_TYPE_OAUTH
from hermes_cli.auth import read_credential_pool
for entry in read_credential_pool("anthropic"):
if (
isinstance(entry, dict)
and entry.get("auth_type") == AUTH_TYPE_OAUTH
and str(entry.get("access_token") or "").strip()
):
if (isinstance(entry, dict) and entry.get("auth_type") == AUTH_TYPE_OAUTH
and str(entry.get("access_token") or "").strip()):
return True
except Exception:
pass
@@ -625,38 +506,28 @@ def _filter_explicit_provider_rows(rows: list[dict], ctx: ConfigContext) -> list
if (
row.get("is_user_defined")
or (current_slug and slug == current_slug)
# Managed local models are explicit configuration by existence:
# the user downloaded gigabytes into the machine-scoped models
# dir. There is deliberately no config credential to find
# (credential is reachability), so without this clause the row
# only survives on the profile where Use was last clicked —
# every other profile loses local models from its picker.
# Managed local models are explicit configuration by existence (gigabytes downloaded
# into the machine-scoped models dir). There is deliberately no config credential
# (credential is reachability), so without this clause the row only survives on the
# profile where Use was last clicked — every other profile loses local models.
or row.get("source") == "local-runtime"
):
return True
if slug == "moa":
# MoA is a virtual routing mode, not an independently configured
# provider. Hide it from explicit-only pickers unless it is the
# current provider (handled above) or the user explicitly wrote an
# enabled MoA preset into config.yaml. Use raw config so the
# DEFAULT_CONFIG preset does not make every desktop picker show MoA.
# MoA is a virtual routing mode, not a configured provider. Hide it unless current
# (handled above) or the user wrote an enabled preset into config.yaml. Raw config,
# so the DEFAULT_CONFIG preset does not make every desktop picker show MoA.
return _raw_config_has_enabled_moa_preset()
return (
# Keyless providers (opencode-free) require no configuration at
# all — there is nothing to "explicitly configure", and hiding
# them would defeat their purpose (zero-setup discoverability).
# Keyless providers need no configuration at all; hiding them would defeat their
# zero-setup discoverability.
_provider_is_keyless(slug)
# Anthropic OAuth logins (Hermes device flow / Claude Code) are
# deliberate sign-ins that leave no trace in active_provider,
# model.provider, or API-key env vars. The strict gate below
# would drop the row even though list_authenticated_providers
# just accepted those same credentials when building it.
# Anthropic OAuth logins (Hermes device flow / Claude Code) are deliberate sign-ins that
# leave no trace in active_provider, model.provider, or env; the strict gate below would
# drop the row list_authenticated_providers just accepted.
or (slug == "anthropic" and _anthropic_oauth_credentials_present())
# External-process providers (copilot-acp) authenticate through their
# own CLI (`copilot login`) — same class as Anthropic OAuth above:
# verified CLI credentials are a deliberate sign-in with no trace in
# active_provider/model.provider/env, so keep the row the
# picker-discovery side just accepted.
# External-process providers (copilot-acp) authenticate through their own CLI — same
# class as Anthropic OAuth: keep the row picker-discovery just accepted.
or _external_process_signed_in(slug)
or is_provider_explicitly_configured(slug)
)
@@ -672,10 +543,7 @@ def _filter_explicit_provider_rows(rows: list[dict], ctx: ConfigContext) -> list
def _external_process_signed_in(slug: str) -> bool:
"""True when an external-process provider has verified CLI credentials."""
try:
from hermes_cli.auth import (
PROVIDER_REGISTRY,
get_external_process_provider_status,
)
from hermes_cli.auth import PROVIDER_REGISTRY, get_external_process_provider_status
pconfig = PROVIDER_REGISTRY.get(slug)
if not pconfig or pconfig.auth_type != "external_process":
return False
@@ -695,12 +563,11 @@ def _provider_is_keyless(slug: str) -> bool:
def _raw_config_has_enabled_moa_preset() -> bool:
"""Return True when the user's raw config explicitly enables MoA.
"""True when the user's raw config explicitly enables MoA.
``load_config()`` includes ``DEFAULT_CONFIG["moa"].presets.default`` for everyone. Explicit-only
model pickers must not treat that default as a user choice, but they should keep MoA visible
once the user has saved at least one enabled preset (or an older flat MoA config) in their own
config.yaml.
``load_config()`` merges ``DEFAULT_CONFIG["moa"].presets.default`` for everyone; that default is
not a user choice. MoA stays visible once the user saved at least one enabled preset (or an
older flat MoA config) in their own config.yaml.
"""
try:
from hermes_cli.config import read_raw_config
@@ -709,9 +576,7 @@ def _raw_config_has_enabled_moa_preset() -> bool:
except Exception:
return False
if not isinstance(raw, dict):
return False
moa = raw.get("moa")
moa = raw.get("moa") if isinstance(raw, dict) else None
if not isinstance(moa, dict):
return False
@@ -719,18 +584,12 @@ def _raw_config_has_enabled_moa_preset() -> bool:
if isinstance(presets, dict):
return any(
not isinstance(preset, dict) or preset.get("enabled", True)
for name, preset in presets.items()
if str(name or "").strip()
for name, preset in presets.items() if str(name or "").strip()
)
legacy_keys = {
"reference_models",
"aggregator",
"reference_temperature",
"aggregator_temperature",
"max_tokens",
"reference_max_tokens",
"fanout",
"reference_models", "aggregator", "reference_temperature", "aggregator_temperature",
"max_tokens", "reference_max_tokens", "fanout",
}
return any(key in moa for key in legacy_keys) and bool(moa.get("enabled", True))
@@ -740,11 +599,8 @@ def _apply_picker_hints(rows: list[dict]) -> None:
for row in rows:
if "authenticated" in row:
continue
# Distinguish authenticated rows (returned by
# list_authenticated_providers) from skeleton rows (from
# _append_unconfigured_rows). The skeleton rows have empty
# `models` AND source="canonical"; authenticated rows have
# populated `models` OR a non-canonical source.
# Skeleton rows (from _append_unconfigured_rows) have empty `models` AND source="canonical";
# authenticated rows have populated `models` OR a non-canonical source.
is_skeleton = row.get("source") == "canonical" and not row.get("models")
row["authenticated"] = not is_skeleton
if not is_skeleton or row.get("is_user_defined"):
@@ -762,39 +618,26 @@ def _apply_picker_hints(rows: list[dict]) -> None:
def _reorder_canonical(rows: list[dict]) -> list[dict]:
"""Canonical slugs in ``CANONICAL_PROVIDERS`` declaration order; truly-custom rows last.
Keys on slug membership, NOT ``is_user_defined``: rows from the ``providers:`` config dict
carry that flag even for canonical slugs, so keying on it would demote canonical providers
configured via the keyed schema.
Keys on slug membership, NOT ``is_user_defined``: rows from the ``providers:`` config dict carry
that flag even for canonical slugs, so keying on it would demote canonical providers.
"""
from hermes_cli.models import CANONICAL_PROVIDERS
order = {e.slug: i for i, e in enumerate(CANONICAL_PROVIDERS)}
canon = sorted(
(r for r in rows if r["slug"] in order),
key=lambda r: order[r["slug"]],
)
canon = sorted((r for r in rows if r["slug"] in order), key=lambda r: order[r["slug"]])
extras = [r for r in rows if r["slug"] not in order]
return canon + extras
def _apply_pricing(
rows: list[dict],
*,
force_fresh_nous_tier: bool = False,
) -> None:
def _apply_pricing(rows: list[dict], *, force_fresh_nous_tier: bool = False) -> None:
"""Enrich each provider row with per-model pricing + Nous tier gating.
row["pricing"] = {model_id: {"input": "$3.00", "output": "$15.00", "cache": "$0.30" | None,
"free": bool}}
row["free_tier"] = bool # current account is free-tier row["unavailable_models"] = [...] # paid
models a free user can't pick
Sets ``row["pricing"] = {model_id: {input, output, cache | None, free}}``; for Nous also
``row["free_tier"]`` (account is free-tier) and ``row["unavailable_models"]`` (paid models a
free user can't pick).
"""
from hermes_cli.models import (
_format_price_per_mtok,
check_nous_free_tier,
compute_sale_discount,
get_pricing_for_provider,
_format_price_per_mtok, check_nous_free_tier, compute_sale_discount, get_pricing_for_provider,
partition_nous_models_by_tier,
)
@@ -824,21 +667,16 @@ def _apply_pricing(
inp = _format_price_per_mtok(inp_raw) if inp_raw != "" else ""
out = _format_price_per_mtok(out_raw) if out_raw != "" else ""
entry: dict = {
"input": inp,
"output": out,
"input": inp, "output": out,
"cache": _format_price_per_mtok(cache_raw) if cache_raw else None,
# A model is "free" when both input and output cost nothing.
# "free" when both input and output cost nothing.
"free": inp == "free" and out in ("free", ""),
}
# Sale chrome is Nous Portal-only. Other providers (OpenRouter,
# Novita, …) never get discount_percent / was_* even if a nested
# pricing.original somehow appeared in their catalog. Free / $0
# models get flat -100% chrome (was_* only when the gateway
# served an original).
# Sale chrome is Nous Portal-only: other providers never get discount_percent / was_*
# even if a nested pricing.original appears in their catalog. Free / $0 models get flat
# -100% chrome (was_* only when the gateway served an original).
if slug == "nous":
sale = compute_sale_discount(
inp_raw, out_raw, p.get("original")
)
sale = compute_sale_discount(inp_raw, out_raw, p.get("original"))
if sale is not None:
discount_percent, was_prompt_raw, was_out_raw = sale
entry["discount_percent"] = discount_percent
@@ -853,28 +691,23 @@ def _apply_pricing(
if slug == "nous":
try:
if nous_free_tier is None:
nous_free_tier = check_nous_free_tier(
force_fresh=force_fresh_nous_tier
)
nous_free_tier = check_nous_free_tier(force_fresh=force_fresh_nous_tier)
row["free_tier"] = bool(nous_free_tier)
row["unavailable_models"] = (
partition_nous_models_by_tier(list(models), raw_pricing, free_tier=True)[1]
if nous_free_tier
else []
if nous_free_tier else []
)
except Exception:
# Tier detection failed — fail open (no gating) so the user
# is never blocked from picking a model.
# Tier detection failed — fail open (no gating) so the user can always pick a model.
row["free_tier"] = False
row["unavailable_models"] = []
def _local_runtime_row(ctx: "ConfigContext") -> dict | None:
"""Build the ``llamacpp`` provider row from staged local models.
"""Build the ``llamacpp`` provider row from staged local models, or ``None`` when none staged.
Present whenever GGUFs are staged in the managed models directory — downloaded models must be
selectable even before the server is running (selection starts it via the runtime_provider seam
/ activate flow). Returns ``None`` when nothing is staged.
selectable before the server runs (selection starts it via the runtime_provider seam).
"""
try:
from hermes_cli.local_runtime.bootstrap import staged_model_ids
@@ -882,46 +715,36 @@ def _local_runtime_row(ctx: "ConfigContext") -> dict | None:
staged = staged_model_ids()
if not staged:
return None
current = (ctx.current_provider or "").strip().lower() in (
"llamacpp", "llama.cpp", "llama-cpp")
current = (ctx.current_provider or "").strip().lower() in ("llamacpp", "llama.cpp", "llama-cpp")
if not current:
# A LIVE session on the managed server reports provider "custom"
# (the resolution seam's label) with the managed base_url. Match
# on the endpoint so the picker still marks this row current —
# otherwise the session the user is chatting in shows no
# selection.
# A LIVE session on the managed server reports provider "custom" (the resolution seam's
# label) with the managed base_url. Match on the endpoint so the picker still marks
# this row current — otherwise the session being chatted in shows no selection.
try:
from hermes_cli.local_runtime.endpoint import _state_endpoint
managed = _state_endpoint()
current = bool(
managed
and (ctx.current_base_url or "").strip().rstrip("/")
== managed["base_url"].rstrip("/"))
and (ctx.current_base_url or "").strip().rstrip("/") == managed["base_url"].rstrip("/"))
except Exception:
current = False
return {
"slug": "llamacpp",
# Bare "Local" everywhere user-facing: the engine name is an
# implementation detail (the pane brands this "Local models").
# Bare "Local" everywhere user-facing: the engine name is an implementation detail.
"name": "Local",
"is_current": current,
"is_user_defined": False,
"models": staged,
"total_models": len(staged),
"is_current": current, "is_user_defined": False, "models": staged, "total_models": len(staged),
"source": "local-runtime",
"authenticated": True, # the credential is reachability
"auth_type": "local",
"warning": None,
"authenticated": True, # the credential is reachability
"auth_type": "local", "warning": None,
}
except Exception:
return None
def _moa_provider_row(current_provider: str = "") -> dict | None:
"""Build the virtual ``moa`` provider row for model pickers.
"""Build the virtual ``moa`` provider row shared by the CLI inventory and gateway picker.
Shared by the CLI inventory and the gateway picker path so the row shape lives in one place.
Returns ``None`` when no MoA presets exist.
"""
try:
@@ -933,15 +756,10 @@ def _moa_provider_row(current_provider: str = "") -> dict | None:
if not models:
return None
return {
"slug": "moa",
"name": "Mixture of Agents",
"is_current": (current_provider or "").lower() == "moa",
"is_user_defined": False,
"models": models,
"total_models": len(models),
"source": "virtual",
"authenticated": True,
"auth_type": "virtual",
"slug": "moa", "name": "Mixture of Agents",
"is_current": (current_provider or "").lower() == "moa", "is_user_defined": False,
"models": models, "total_models": len(models), "source": "virtual",
"authenticated": True, "auth_type": "virtual",
"warning": "Aggregator acts as the selected model; references provide analysis before each call.",
}
except Exception:
hermes_cli/journey.py
+12 -41
View File
@@ -55,40 +55,28 @@ def _resolve(style: str, alpha: float) -> Optional[str]:
def _relative_luminance(rgb: tuple[int, int, int]) -> float:
def channel(value: int) -> float:
normalized = value / 255
return (
normalized / 12.92
if normalized <= 0.03928
else ((normalized + 0.055) / 1.055) ** 2.4
)
n = value / 255
return n / 12.92 if n <= 0.03928 else ((n + 0.055) / 1.055) ** 2.4
red, green, blue = (channel(value) for value in rgb)
return 0.2126 * red + 0.7152 * green + 0.0722 * blue
def _contrast_ratio(
foreground: tuple[int, int, int], background: tuple[int, int, int]
) -> float:
foreground_luminance = _relative_luminance(foreground)
background_luminance = _relative_luminance(background)
high, low = sorted((foreground_luminance, background_luminance), reverse=True)
def _contrast_ratio(foreground: tuple[int, int, int], background: tuple[int, int, int]) -> float:
high, low = sorted((_relative_luminance(foreground), _relative_luminance(background)), reverse=True)
return (high + 0.05) / (low + 0.05)
def _ensure_contrast(
color: Optional[str], background: str, minimum: float
) -> Optional[str]:
def _ensure_contrast(color: Optional[str], background: str, minimum: float) -> Optional[str]:
"""Lift a foreground toward the readable pole until it clears ``minimum``."""
if not color:
return None
from agent.learning_graph_render import hex_to_rgb, mix_rgb, rgb_to_hex
foreground_rgb = hex_to_rgb(color)
background_rgb = hex_to_rgb(background)
if _contrast_ratio(foreground_rgb, background_rgb) >= minimum:
return color
pole = (0, 0, 0) if _relative_luminance(background_rgb) > 0.5 else (255, 255, 255)
for step in range(1, 21):
candidate = mix_rgb(foreground_rgb, pole, step * 0.05)
@@ -99,9 +87,7 @@ def _ensure_contrast(
def _resolve_charted_signal(style: str, alpha: float) -> Optional[str]:
"""Keep age tinting without allowing explanatory labels to disappear."""
return _ensure_contrast(
_resolve(style, alpha), _palette()["bg"], _CHARTED_SIGNAL_MIN_CONTRAST
)
return _ensure_contrast(_resolve(style, alpha), _palette()["bg"], _CHARTED_SIGNAL_MIN_CONTRAST)
def _row_to_text(row: list, color: bool):
@@ -109,8 +95,7 @@ def _row_to_text(row: list, color: bool):
text = Text()
for run in row:
chunk = run[0]
style = run[1]
chunk, style = run[0], run[1]
alpha = run[2] if len(run) > 2 else 1.0
override = run[3] if len(run) > 3 else None
if not color:
@@ -144,15 +129,13 @@ def _frame_renderable(payload, *, cols, rows, reveal, color):
frame = render.render_graph(payload, cols=inner, rows=field_rows, reveal=reveal)
count = len(payload.get("nodes", []))
parts: list[Any] = []
def st(style: Optional[str]) -> Optional[str]:
return style if color else None
title = Text()
title.append("✦ Journey ", style=st(f"bold {_TITLE_COLOR}"))
title.append("· learned skills & memories over time", style=st("grey62"))
parts.append(title)
parts: list[Any] = [title]
legend_line = Text(" ")
for i, item in enumerate(legend):
@@ -172,7 +155,6 @@ def _frame_renderable(payload, *, cols, rows, reveal, color):
parts.append(cat_line)
parts.append(Text(""))
for grow in frame["grid"]:
line = _row_to_text(grow, color)
line.pad_left(2)
@@ -181,8 +163,7 @@ def _frame_renderable(payload, *, cols, rows, reveal, color):
# Date axis under the field (oldest → now), with the playhead date centered.
axis_line = Text(" ")
axis_line.append(axis["start"], style=st("grey54"))
gap = max(1, inner - len(axis["start"]) - len(axis["end"]))
axis_line.append(" " * gap)
axis_line.append(" " * max(1, inner - len(axis["start"]) - len(axis["end"])))
axis_line.append(axis["end"], style=st("grey54"))
parts.append(axis_line)
@@ -197,14 +178,11 @@ def _frame_renderable(payload, *, cols, rows, reveal, color):
if labels:
parts.append(Text(""))
parts.append(Text(" charted signals", style=st("grey62")))
for item in labels[:6]:
row = Text(" ")
row.append(f"{item['key']} ", style=st("grey70"))
signal_style = (
_resolve_charted_signal(item["style"], float(item.get("alpha", 1.0)))
if color
else None
_resolve_charted_signal(item["style"], float(item.get("alpha", 1.0))) if color else None
)
row.append(f"{item['glyph']} ", style=signal_style)
row.append(str(item["label"]), style=signal_style)
@@ -214,7 +192,6 @@ def _frame_renderable(payload, *, cols, rows, reveal, color):
for line_text in summary:
parts.append(Text(" " + line_text, style=st("grey62")))
return Group(*parts)
@@ -249,10 +226,8 @@ def _cmd_show(args: argparse.Namespace) -> int:
"memories will start mapping out here.[/grey62]"
)
return 0
if getattr(args, "play", False):
return _play(console, payload, cols=cols, rows=rows, color=color, fps=getattr(args, "fps", 12))
reveal = _clamp(float(getattr(args, "reveal", 1.0) or 1.0), 0.0, 1.0)
console.print(_frame_renderable(payload, cols=cols, rows=rows, reveal=reveal, color=color))
return 0
@@ -334,8 +309,7 @@ def _cmd_edit(args: argparse.Namespace) -> int:
detail = _lookup_node(args.node)
if detail is None:
return 1
suffix = ".md" if detail["kind"] == "skill" else ".txt"
edited = _open_in_editor(detail["content"], suffix=suffix)
edited = _open_in_editor(detail["content"], suffix=".md" if detail["kind"] == "skill" else ".txt")
if edited is None or edited.strip() == detail["content"].strip():
print(" no changes")
return 0
@@ -369,10 +343,7 @@ def _open_in_editor(initial: str, *, suffix: str) -> Optional[str]:
def register_cli(parent: argparse.ArgumentParser) -> None:
parent.add_argument(
"--reveal",
type=float,
default=1.0,
metavar="0..1",
"--reveal", type=float, default=1.0, metavar="0..1",
help="Render the timeline built up to this point (0=oldest, 1=now).",
)
parent.add_argument("--play", action="store_true", help="Animate the build-up over time (Ctrl-C to stop).")
hermes_cli/kanban.py
+269 -520
View File
File diff suppressed because it is too large Load Diff
hermes_cli/kanban_boards.py
+30 -55
View File
@@ -1,6 +1,6 @@
"""``hermes kanban boards …`` — board directories, the ``current`` pointer and
``board.json``. Filesystem-only, so every action works before ``kanban init``
and must ignore the shared ``--board`` task-routing override.
"""``hermes kanban boards …`` — board directories, the ``current`` pointer and ``board.json``.
Filesystem-only, so every action works before ``kanban init`` and must ignore the shared
``--board`` task-routing override.
"""
from __future__ import annotations
@@ -12,10 +12,8 @@ from hermes_cli import kanban_db as kb
from hermes_cli.kanban_output import _err, _fmt_counts, _json_out
def _dispatch_boards(args: argparse.Namespace) -> int:
"""``hermes kanban boards <action>`` — filesystem-only (board dirs, the
``current`` pointer, ``board.json``), so it works before ``kanban init``."""
"""``hermes kanban boards <action>`` — filesystem-only, so it works before ``kanban init``."""
sub = getattr(args, "boards_action", None) or "list"
handler = _BOARD_HANDLERS.get(sub)
if handler is None:
@@ -24,15 +22,12 @@ def _dispatch_boards(args: argparse.Namespace) -> int:
def _board_task_counts(slug: str) -> dict[str, int]:
"""Return ``{status: count}`` for a board. Safe to call on an empty DB."""
"""``{status: count}`` for a board. Safe to call on an empty DB."""
try:
path = kb.kanban_db_path(board=slug)
if not path.exists():
if not kb.kanban_db_path(board=slug).exists():
return {}
with kb.connect_closing(board=slug) as conn:
rows = conn.execute(
"SELECT status, COUNT(*) AS n FROM tasks GROUP BY status"
).fetchall()
rows = conn.execute("SELECT status, COUNT(*) AS n FROM tasks GROUP BY status").fetchall()
return {r["status"]: int(r["n"]) for r in rows}
except Exception:
return {}
@@ -67,12 +62,9 @@ def _cmd_boards_list(args: argparse.Namespace) -> int:
print(f"{'':2s} {'SLUG':24s} {'NAME':28s} COUNTS")
for b in boards:
marker = "●" if b["is_current"] else " "
name = b.get("name") or ""
if b.get("archived"):
name += " [archived]"
name = (b.get("name") or "") + (" [archived]" if b.get("archived") else "")
print(f"{marker:2s} {b['slug']:24s} {name:28s} {_fmt_counts(b['counts'] or {}, '(empty)')}")
print()
print(f"Current board: {current}")
print(f"\nCurrent board: {current}")
if len(boards) > 1:
print("Switch boards with `hermes kanban boards switch <slug>`.")
return 0
@@ -84,17 +76,12 @@ def _cmd_boards_create(args: argparse.Namespace) -> int:
return rc
already = kb.board_exists(normed) and normed != kb.DEFAULT_BOARD
meta = kb.create_board(
normed,
name=args.name,
description=args.description,
icon=args.icon,
color=args.color,
normed, name=args.name, description=args.description, icon=args.icon, color=args.color,
default_workdir=args.default_workdir,
)
verb = "already exists" if already else "created"
print(f"Board {meta['slug']!r} {verb}.")
print(f" Display name: {meta.get('name', '')}")
print(f" DB path: {meta['db_path']}")
print(f"Board {meta['slug']!r} {'already exists' if already else 'created'}.\n"
f" Display name: {meta.get('name', '')}\n"
f" DB path: {meta['db_path']}")
if getattr(args, "switch", False):
kb.set_current_board(meta["slug"])
print(f" Switched to {meta['slug']!r}.")
@@ -104,17 +91,16 @@ def _cmd_boards_create(args: argparse.Namespace) -> int:
def _cmd_boards_rm(args: argparse.Namespace) -> int:
# `boards delete <slug>` (alias) never sets args.delete because --delete
# belongs to the 'rm' subparser only; treat the alias as `rm --delete`.
# `boards delete <slug>` (alias) never sets args.delete because --delete belongs to the 'rm'
# subparser only; treat the alias as `rm --delete`.
force_delete = getattr(args, "delete", False) or getattr(args, "boards_action", "") == "delete"
try:
res = kb.remove_board(args.slug, archive=not force_delete)
except ValueError as exc:
return _err(f"kanban boards rm: {exc}")
if res["action"] == "archived":
print(f"Board {res['slug']!r} archived → {res['new_path']}")
print("Recover by moving the directory back to "
"<root>/kanban/boards/<slug>/.")
print(f"Board {res['slug']!r} archived → {res['new_path']}\n"
"Recover by moving the directory back to <root>/kanban/boards/<slug>/.")
else:
print(f"Board {res['slug']!r} deleted.")
return 0
@@ -138,13 +124,11 @@ def _cmd_boards_show(args: argparse.Namespace) -> int:
current = kb.get_current_board()
meta = kb.read_board_metadata(current)
counts = _board_task_counts(current)
print(f"Current board: {current}")
print(f" Display name: {meta.get('name', '')}")
print(f"Current board: {current}\n Display name: {meta.get('name', '')}")
if meta.get("description"):
print(f" Description: {meta['description']}")
print(f" DB path: {meta['db_path']}")
print(f" Tasks: {sum(counts.values())} total"
+ (f" ({_fmt_counts(counts)})" if counts else ""))
print(f" DB path: {meta['db_path']}\n"
f" Tasks: {sum(counts.values())} total" + (f" ({_fmt_counts(counts)})" if counts else ""))
return 0
@@ -161,8 +145,7 @@ def _cmd_boards_set_default_workdir(args: argparse.Namespace) -> int:
normed, rc = _board_slug_arg(args, "set-default-workdir", must_exist=True)
if rc:
return rc
meta = kb.write_board_metadata(normed, default_workdir=args.path)
new_val = meta.get("default_workdir")
new_val = kb.write_board_metadata(normed, default_workdir=args.path).get("default_workdir")
if new_val:
print(f"Board {normed!r} default workdir set to {new_val!r}.")
else:
@@ -178,23 +161,19 @@ def _cmd_boards_export(args: argparse.Namespace) -> int:
output = args.output or f"{slug}.tar.gz"
try:
res = kanban_transfer.export_board(
slug,
output,
include_attachments=not args.no_attachments,
include_logs=args.include_logs,
slug, output, include_attachments=not args.no_attachments, include_logs=args.include_logs,
)
except (OSError, ValueError) as exc:
return _err(f"kanban boards export: {exc}")
if _json_out(args, res):
return 0
counts = res["counts"]
print(f"Exported board {res['board']!r} → {res['archive']}")
print(f" Size: {format_bytes(res['size'])}")
print(f" Tasks: {counts['tasks']}")
print(f" Comments: {counts['task_comments']}")
print(f" Attachments: {counts['attachment_files']}")
print("Import it with `hermes kanban boards import <archive>`.")
print(f"Exported board {res['board']!r} → {res['archive']}\n"
f" Size: {format_bytes(res['size'])}\n"
f" Tasks: {counts['tasks']}\n"
f" Comments: {counts['task_comments']}\n"
f" Attachments: {counts['attachment_files']}\n"
"Import it with `hermes kanban boards import <archive>`.")
return 0
@@ -202,19 +181,15 @@ def _cmd_boards_import(args: argparse.Namespace) -> int:
from hermes_cli import kanban_transfer
try:
res = kanban_transfer.import_board(
args.archive, args.as_slug, activate=args.switch
)
res = kanban_transfer.import_board(args.archive, args.as_slug, activate=args.switch)
except (OSError, ValueError) as exc:
return _err(f"kanban boards import: {exc}")
if _json_out(args, res):
return 0
print(f"Imported board {res['board']!r} ({res['name']}).")
if res["renamed"]:
print(f" Renamed from {res['requested_board']!r} — that slug was taken.")
print(f" Path: {res['path']}")
print(f" Tasks: {res['counts']['tasks']}")
print(f" Path: {res['path']}\n Tasks: {res['counts']['tasks']}")
for warning in res["warnings"]:
print(f" Note: {warning}")
if res["activated"]: