Merge branch 'simp/r3-21-D' into simp/integration3

hermes_cli/main_agent_cmds.py

@@ -10,86 +10,80 @@ avoids an import cycle).
 import sys
 
 
+def _cmd_memory_off():
+    from hermes_cli.config import load_config, save_config
+    config = load_config()
+    if not isinstance(config.get("memory"), dict):
+        config["memory"] = {}
+    config["memory"]["provider"] = ""
+    save_config(config)
+    print("\n  ✓ Memory provider: built-in only")
+    print("  Saved to config.yaml\n")
+
+
+def _cmd_memory_reset(args):
+    from hermes_constants import get_hermes_home, display_hermes_home
+    mem_dir = get_hermes_home() / "memories"
+    target = getattr(args, "target", "all")
+    files_to_reset = []
+    if target in {"all", "memory"}:
+        files_to_reset.append(("MEMORY.md", "agent notes"))
+    if target in {"all", "user"}:
+        files_to_reset.append(("USER.md", "user profile"))
+
+    existing = [(f, desc) for f, desc in files_to_reset if (mem_dir / f).exists()]
+    if not existing:
+        print(f"\n  Nothing to reset — no memory files found in {display_hermes_home()}/memories/\n")
+        return
+
+    print("\n  This will permanently erase the following memory files:")
+    for f, desc in existing:
+        size = (mem_dir / f).stat().st_size
+        print(f"    ◆ {f} ({desc}) — {size:,} bytes")
+
+    if not getattr(args, "yes", False):
+        try:
+            answer = input("\n  Type 'yes' to confirm: ").strip().lower()
+        except (EOFError, KeyboardInterrupt):
+            print("\n  Cancelled.\n")
+            return
+        if answer != "yes":
+            print("  Cancelled.\n")
+            return
+
+    for f, desc in existing:
+        (mem_dir / f).unlink()
+        print(f"  ✓ Deleted {f} ({desc})")
+
+    print("\n  Memory reset complete. New sessions will start with a blank slate.")
+    print(f"  Files were in: {display_hermes_home()}/memories/\n")
+
+
 def cmd_memory(args):
     sub = getattr(args, "memory_command", None)
     if sub == "off":
-        from hermes_cli.config import load_config, save_config
-
-        config = load_config()
-        if not isinstance(config.get("memory"), dict):
-            config["memory"] = {}
-        config["memory"]["provider"] = ""
-        save_config(config)
-        print("\n  ✓ Memory provider: built-in only")
-        print("  Saved to config.yaml\n")
+        _cmd_memory_off()
     elif sub == "reset":
-        from hermes_constants import get_hermes_home, display_hermes_home
-
-        mem_dir = get_hermes_home() / "memories"
-        target = getattr(args, "target", "all")
-        files_to_reset = []
-        if target in {"all", "memory"}:
-            files_to_reset.append(("MEMORY.md", "agent notes"))
-        if target in {"all", "user"}:
-            files_to_reset.append(("USER.md", "user profile"))
-
-        # Check what exists
-        existing = [
-            (f, desc) for f, desc in files_to_reset if (mem_dir / f).exists()
-        ]
-        if not existing:
-            print(
-                f"\n  Nothing to reset — no memory files found in {display_hermes_home()}/memories/\n"
-            )
-            return
-
-        print("\n  This will permanently erase the following memory files:")
-        for f, desc in existing:
-            path = mem_dir / f
-            size = path.stat().st_size
-            print(f"    ◆ {f} ({desc}) — {size:,} bytes")
-
-        if not getattr(args, "yes", False):
-            try:
-                answer = input("\n  Type 'yes' to confirm: ").strip().lower()
-            except (EOFError, KeyboardInterrupt):
-                print("\n  Cancelled.\n")
-                return
-            if answer != "yes":
-                print("  Cancelled.\n")
-                return
-
-        for f, desc in existing:
-            (mem_dir / f).unlink()
-            print(f"  ✓ Deleted {f} ({desc})")
-
-        print(
-            "\n  Memory reset complete. New sessions will start with a blank slate."
-        )
-        print(f"  Files were in: {display_hermes_home()}/memories/\n")
+        _cmd_memory_reset(args)
     else:
         from hermes_cli.memory_setup import memory_command
-
         memory_command(args)
 
 
+# (args attribute, acp flag) — forwarded in this order.
+_ACP_FLAGS = (
+    ("acp_version", "--version"),
+    ("check", "--check"),
+    ("setup", "--setup"),
+    ("setup_browser", "--setup-browser"),
+    ("assume_yes", "--yes"))
+
+
 def cmd_acp(args):
     """Launch Hermes Agent as an ACP server."""
     try:
         from acp_adapter.entry import main as acp_main
-
-        acp_argv = []
-        if getattr(args, "acp_version", False):
-            acp_argv.append("--version")
-        if getattr(args, "check", False):
-            acp_argv.append("--check")
-        if getattr(args, "setup", False):
-            acp_argv.append("--setup")
-        if getattr(args, "setup_browser", False):
-            acp_argv.append("--setup-browser")
-        if getattr(args, "assume_yes", False):
-            acp_argv.append("--yes")
-        acp_main(acp_argv)
+        acp_main([flag for attr, flag in _ACP_FLAGS if getattr(args, attr, False)])
     except ImportError:
         print("ACP dependencies not installed.", file=sys.stderr)
         print("Install them with:  pip install -e '.[acp]'", file=sys.stderr)
@@ -101,16 +95,13 @@ def cmd_tools(args):
     action = getattr(args, "tools_action", None)
     if action in {"list", "disable", "enable"}:
         from hermes_cli.tools_config import tools_disable_enable_command
-
         tools_disable_enable_command(args)
     elif action == "post-setup":
         from hermes_cli.tools_config import run_post_setup_command
-
         sys.exit(run_post_setup_command(args))
     else:
         _require_tty("tools")
         from hermes_cli.tools_config import tools_command
-
         tools_command(args)
 
 
@@ -119,7 +110,6 @@ def cmd_insights(args):
     try:
         from hermes_state import SessionDB
         from agent.insights import InsightsEngine
-
         db = SessionDB()
         engine = InsightsEngine(db)
         report = engine.generate(days=args.days, source=args.source)
@@ -134,24 +124,20 @@ def cmd_insights(args):
                 pass
 
 
+def _dict_or_empty(value) -> dict:
+    return value if isinstance(value, dict) else {}
+
+
 def cmd_monitoring(args):
     """Gateway monitoring status: health & diagnostics export posture."""
     from hermes_cli.config import load_config
-
     action = getattr(args, "monitoring_action", None) or "status"
-    config = load_config()
-    mon_raw = config.get("monitoring")
-    mon: dict = mon_raw if isinstance(mon_raw, dict) else {}
+    mon = _dict_or_empty(load_config().get("monitoring"))
 
     if action == "status":
         from agent.monitoring import otlp_exporter
-
-        gh_raw = mon.get("gateway_health_export")
-        gh: dict = gh_raw if isinstance(gh_raw, dict) else {}
-        export_raw = mon.get("export")
-        export_cfg: dict = export_raw if isinstance(export_raw, dict) else {}
-        otlp_raw = export_cfg.get("otlp")
-        otlp: dict = otlp_raw if isinstance(otlp_raw, dict) else {}
+        gh = _dict_or_empty(mon.get("gateway_health_export"))
+        otlp = _dict_or_empty(_dict_or_empty(mon.get("export")).get("otlp"))
 
         print("Gateway monitoring")
         print(f"  Health export:  {'enabled' if gh.get('enabled') else 'disabled'} "
@@ -180,38 +166,32 @@ def cmd_monitoring(args):
 
 
 def cmd_skills(args):
-    # Route 'config' action to skills_config module
     from hermes_cli.main import _require_tty
-    if getattr(args, "skills_action", None) == "config":
+    action = getattr(args, "skills_action", None)
+    if action == "config":
         _require_tty("skills config")
         from hermes_cli.skills_config import skills_command as skills_config_command
-
         skills_config_command(args)
-    elif getattr(args, "skills_action", None) in ("trust", "untrust"):
+    elif action in ("trust", "untrust"):
         _cmd_skills_trust(args)
     else:
         from hermes_cli.skills_hub import skills_command
-
         skills_command(args)
 
 
 def _cmd_skills_trust(args):
-    """``hermes skills trust [path]`` / ``hermes skills untrust [path]``.
+    """``hermes skills trust|untrust [path]`` — manage ``skills.trusted_project_dirs``.
 
-    Manages ``skills.trusted_project_dirs`` in config.yaml. With no path,
-    operates on the project root enclosing the current directory (nearest
-    ancestor with ``.git``).
+    With no path, operates on the project root enclosing the current directory
+    (nearest ancestor with ``.git``).
     """
     from pathlib import Path
-
     from agent.skill_utils import (
         PROJECT_SKILLS_SUBDIRS,
         _candidate_project_skills_dirs,
         find_project_root,
-        iter_skill_index_files,
-    )
+        iter_skill_index_files)
     from hermes_cli.config import load_config, save_config
-
     action = args.skills_action
     raw_path = getattr(args, "path", None)
     if raw_path:
@@ -224,8 +204,7 @@ def _cmd_skills_trust(args):
         if root is None:
             print(
                 "Not inside a git checkout. Run from a project directory or "
-                "pass the project root path explicitly."
-            )
+                "pass the project root path explicitly.")
             return
 
     config = load_config()
@@ -236,8 +215,11 @@ def _cmd_skills_trust(args):
     trusted = [str(t) for t in trusted]
     root_str = str(root)
 
+    def _same(t: str) -> bool:
+        return str(Path(t).expanduser().resolve()) == root_str
+
     if action == "untrust":
-        kept = [t for t in trusted if str(Path(t).expanduser().resolve()) != root_str]
+        kept = [t for t in trusted if not _same(t)]
         if len(kept) == len(trusted):
             print(f"{root} was not trusted.")
             return
@@ -247,8 +229,7 @@ def _cmd_skills_trust(args):
         print("Project skills from this repo will no longer load.")
         return
 
-    # trust
-    if any(str(Path(t).expanduser().resolve()) == root_str for t in trusted):
+    if any(_same(t) for t in trusted):
         print(f"Already trusted: {root}")
     else:
         trusted.append(root_str)
@@ -257,14 +238,13 @@ def _cmd_skills_trust(args):
         print(f"Trusted: {root}")
 
     # Show what this unlocks
-    count = 0
-    for d in _candidate_project_skills_dirs(root):
-        count += sum(1 for _ in iter_skill_index_files(d, "SKILL.md"))
+    count = sum(
+        sum(1 for _ in iter_skill_index_files(d, "SKILL.md"))
+        for d in _candidate_project_skills_dirs(root))
     if count:
         print(
             f"{count} project skill(s) will load in sessions started inside "
-            "this repo (they take precedence over same-named profile skills)."
-        )
+            "this repo (they take precedence over same-named profile skills).")
     else:
         subdirs = " or ".join(PROJECT_SKILLS_SUBDIRS)
         print(f"No project skills found yet — add them under {subdirs}.")

hermes_cli/managed_scope.py

@@ -1,12 +1,10 @@
 """Managed scope — IT-pushed, user-immutable config & env layer.
 
-This is DISTINCT from ``hermes_cli.config.is_managed()`` / ``HERMES_MANAGED``, which is a coarse
-package-manager write-lock (declarative-distro / formula installs). That lock blocks all mutation;
-this layer injects specific immutable values. The two are independent and may coexist.
-
-v1 enforcement is filesystem permissions only — see ``docs/design/managed-scope.md`` §7. v1 is
-Linux/POSIX-first; ``get_managed_dir()`` is the single seam for adding macOS / Windows native
-locations later.
+DISTINCT from ``hermes_cli.config.is_managed()`` / ``HERMES_MANAGED`` (a coarse package-manager
+write-lock that blocks all mutation); this layer injects specific immutable values. The two are
+independent and may coexist. v1 enforcement is filesystem permissions only (see
+``docs/design/managed-scope.md`` §7); ``get_managed_dir()`` is the single seam for adding
+macOS / Windows native locations later.
 """
 from __future__ import annotations
 
@@ -21,8 +19,7 @@ import yaml
 
 logger = logging.getLogger(__name__)
 
-# POSIX default. Other-platform locations are a deliberate v2 item; when added,
-# they belong ONLY inside get_managed_dir().
+# POSIX default. Other-platform locations belong ONLY inside get_managed_dir().
 _DEFAULT_MANAGED_DIR = Path("/etc/hermes")
 
 _CACHE_LOCK = threading.Lock()
@@ -32,24 +29,17 @@ _ENV_CACHE: Dict[str, tuple] = {}
 
 
 def _under_pytest() -> bool:
-    """True when running inside the test suite.
-
-    Used to ignore the system default ``/etc/hermes`` under tests so a real managed scope on a
-    dev/CI box can't leak policy into the suite. An explicit ``HERMES_MANAGED_DIR`` is still
-    honored because that override runs before this guard.
-    """
+    """True inside the test suite: ignore the system ``/etc/hermes`` so a real managed scope on a
+    dev/CI box can't leak policy into the suite. An explicit ``HERMES_MANAGED_DIR`` still wins."""
     return "PYTEST_CURRENT_TEST" in os.environ
 
 
 def get_managed_dir() -> Optional[Path]:
     """Resolve the managed-scope directory, or None when no scope is present.
 
-    Resolution (highest priority first): 1. ``$HERMES_MANAGED_DIR`` — deployment/bootstrap path
-    override (IT-only; never persisted to any .env). Honored only when set to a non-empty value AND
-    the directory exists. 2. ``/etc/hermes`` — POSIX default, when it exists.
-
-    A non-existent directory at either tier resolves to None (no managed scope), which is the common
-    case and must be cheap + side-effect-free.
+    Priority: ``$HERMES_MANAGED_DIR`` (IT-only bootstrap override; never persisted to any .env;
+    honored only when non-empty AND the directory exists), then ``/etc/hermes`` when it exists.
+    A missing directory resolves to None — the common case, so it must be cheap + side-effect-free.
     """
     override = os.environ.get("HERMES_MANAGED_DIR", "").strip()
     if override:
@@ -69,11 +59,11 @@ def invalidate_managed_cache() -> None:
 
 
 def _cached_read(path: Path, cache: Dict[str, tuple], parse):
-    """Shared (mtime_ns, size)-keyed read. Returns a deepcopy of the parsed value.
+    """Shared (mtime_ns, size)-keyed read; returns a deepcopy of the parsed value.
 
-    Returns ``None`` when the file is absent or fails to parse (fail-open). A parse failure is
-    logged LOUDLY — the admin needs to know their policy isn't being applied — but never raises, so
-    a malformed managed file can't brick startup.
+    ``None`` when the file is absent or fails to parse (fail-open). A parse failure is logged
+    LOUDLY — the admin needs to know their policy isn't applied — but never raises, so a malformed
+    managed file can't brick startup.
     """
     try:
         st = path.stat()
@@ -92,8 +82,7 @@ def _cached_read(path: Path, cache: Dict[str, tuple], parse):
         logger.warning(
             "managed scope: failed to parse %s: %s — IGNORING this managed file. "
             "Admin policy from this file is NOT being applied. Fix and restart.",
-            path, exc,
-        )
+            path, exc)
         return None
     with _CACHE_LOCK:
         cache[path_key] = (*key, copy.deepcopy(parsed))
@@ -121,15 +110,11 @@ def load_managed_env() -> Dict[str, str]:
 def apply_managed_overlay(config: dict) -> dict:
     """Overlay administrator-pinned config values on top of an already-built dict.
 
-    * expand the managed config's ``${VAR}`` refs against the PROCESS env only (never user-config-
-    defined refs), so a user cannot shadow a managed literal via a ${VAR} they control; * normalize
-    the managed config's root ``model`` key (a bare ``model: x/y`` string is promoted to
-    ``model.default``) so it can't clobber the dict shape callers expect; * leaf-level deep-merge
-    managed ON TOP, so managed wins per-leaf while sibling keys stay user-controlled.
-
-    Fail-open: returns ``config`` unchanged if no managed scope is present or on any error — managed
-    scope must never break a caller's startup. Mutates and returns ``config`` (callers pass a dict
-    they own).
+    ``${VAR}`` refs in the managed config expand against the PROCESS env only, so a user cannot
+    shadow a managed literal via a ref they control; a bare root ``model: x/y`` string is promoted
+    to ``model.default`` so it can't clobber the dict shape callers expect; managed values
+    deep-merge ON TOP per leaf while sibling keys stay user-controlled. Fail-open: returns
+    ``config`` unchanged when no scope is present or on any error. Mutates and returns ``config``.
     """
     try:
         managed = load_managed_config()
@@ -137,14 +122,10 @@ def apply_managed_overlay(config: dict) -> dict:
             return config
         # Imported lazily to avoid an import cycle (config imports managed_scope).
         from hermes_cli.config import _deep_merge, _expand_env_vars, _normalize_root_model_keys
-
         managed_expanded = _normalize_root_model_keys(_expand_env_vars(managed))
-        # A bare ``model: x/y`` string in the managed file must merge as
-        # ``model.default`` — otherwise _deep_merge would replace the caller's
-        # ``model`` dict with a string and break every ``cfg["model"]["..."]``
-        # read. _normalize_root_model_keys only promotes the string when there
-        # are root provider/base_url keys to migrate, so handle the bare case
-        # here (matches cli.py's own string-model handling).
+        # _normalize_root_model_keys only promotes the string when root provider/base_url
+        # keys exist to migrate; handle the bare case here (matches cli.py) so _deep_merge
+        # never replaces the caller's ``model`` dict with a string.
         if isinstance(managed_expanded.get("model"), str):
             managed_expanded = dict(managed_expanded)
             managed_expanded["model"] = {"default": managed_expanded["model"]}

hermes_cli/mem_trim.py

@@ -23,6 +23,11 @@ logger = logging.getLogger(__name__)
 _DEFAULT_COOLDOWN_SECONDS = 60.0
 _DEFAULT_LOG_EVERY_N = 1
 _DEFAULT_INFO_LOG_MIN_DELTA_MB = 0.0
+# Even forced trims honor a short floor: AIAgent.close() forces a trim, and delegate
+# batches close N child subagents back-to-back in the SAME process — without a floor
+# that stacks N+1 uncooled full gc.collect() passes (50-500ms each in a large gateway
+# process). 5s coalesces the burst while keeping the parent's final close-trim effective.
+_FORCE_FLOOR_SECONDS = 5.0
 _trim_lock = threading.Lock()
 _last_trim_monotonic = 0.0
 _probe_done = False
@@ -31,17 +36,13 @@ _trim_call_count = 0
 
 
 def _config_settings() -> tuple[bool, float, int, float]:
-    """Return fail-open settings from the normal Hermes config path."""
-    enabled = True
+    """Return fail-open ``(enabled, cooldown, log_every_n, info_log_min_delta_mb)`` from config."""
     settings: Any = None
     try:
-        # Read-only access: settings are only .get()ed and coerced, never
-        # mutated — use the no-deepcopy variant. This runs on EVERY trim
-        # attempt (before the cooldown check), and generating a full-config
-        # deepcopy per attempt is exactly the allocator garbage this module
-        # exists to release.
+        # Read-only, no-deepcopy variant: this runs on EVERY trim attempt (before the
+        # cooldown check), and a full-config deepcopy per attempt is exactly the
+        # allocator garbage this module exists to release.
         from hermes_cli.config import load_config_readonly
-
         config = load_config_readonly() or {}
         context = config.get("context") if isinstance(config, dict) else None
         settings = context.get("memory_trim") if isinstance(context, dict) else None
@@ -49,14 +50,12 @@ def _config_settings() -> tuple[bool, float, int, float]:
         pass
     if not isinstance(settings, dict):
         settings = {}
-    if isinstance(settings.get("enabled"), bool):
-        enabled = settings["enabled"]
+    enabled = settings["enabled"] if isinstance(settings.get("enabled"), bool) else True
     return (
         enabled,
         _cooldown_seconds(settings.get("cooldown_seconds")),
         _coerce(settings.get("log_every_n"), _DEFAULT_LOG_EVERY_N, int, 1),
-        _coerce(settings.get("info_log_min_delta_mb"), _DEFAULT_INFO_LOG_MIN_DELTA_MB, float, 0.0),
-    )
+        _coerce(settings.get("info_log_min_delta_mb"), _DEFAULT_INFO_LOG_MIN_DELTA_MB, float, 0.0))
 
 
 def _coerce(value: Any, default, cast, floor):
@@ -84,16 +83,12 @@ def _read_proc_status() -> str | None:
 
 
 def collect_memory_snapshot(history_bytes: int | None = None) -> dict[str, int | None]:
-    """Return lightweight process-memory telemetry for trim logs and canaries.
+    """Lightweight process-memory telemetry for trim logs and canaries.
 
-    ``VmRSS`` and ``RssAnon`` are Linux-only best effort fields. The helper is intentionally
-    dependency-free so allocation recovery never requires psutil.
+    ``VmRSS`` / ``RssAnon`` are Linux-only best effort; deliberately psutil-free.
     """
     snapshot: dict[str, int | None] = {
-        "rss_kib": None,
-        "rss_anon_kib": None,
-        "thread_count": threading.active_count(),
-    }
+        "rss_kib": None, "rss_anon_kib": None, "thread_count": threading.active_count()}
     status = _read_proc_status()
     if status:
         for line in status.splitlines():
@@ -110,10 +105,9 @@ def collect_memory_snapshot(history_bytes: int | None = None) -> dict[str, int |
 
 def _should_log_trim(
     *, force: bool, log_every_n: int, call_count: int, before: dict[str, int | None],
-    after: dict[str, int | None], info_log_min_delta_mb: float,
-) -> bool:
-    # trim_memory calls this only after malloc_trim reported success. A forced
-    # successful trim is an explicit observability event, regardless of RSS.
+    after: dict[str, int | None], info_log_min_delta_mb: float) -> bool:
+    # Called only after malloc_trim reported success; a forced successful trim is an
+    # explicit observability event regardless of RSS.
     if force:
         return True
     if call_count % log_every_n:
@@ -136,8 +130,7 @@ def _probe_glibc_malloc_trim() -> Callable[[int], int] | None:
     try:
         if platform.libc_ver()[0].lower() != "glibc":
             return None
-        libc = ctypes.CDLL(None)
-        trim = libc.malloc_trim
+        trim = ctypes.CDLL(None).malloc_trim
         trim.argtypes = [ctypes.c_size_t]
         trim.restype = ctypes.c_int
         _malloc_trim = trim
@@ -147,11 +140,7 @@ def _probe_glibc_malloc_trim() -> Callable[[int], int] | None:
 
 
 def trim_memory(
-    *,
-    force: bool = False,
-    reason: str = "",
-    cooldown_seconds: float | None = None,
-) -> bool:
+    *, force: bool = False, reason: str = "", cooldown_seconds: float | None = None) -> bool:
     """Collect cycles and ask glibc to release free heap pages.
 
     Returns ``True`` only when ``malloc_trim(0)`` ran and reported success. Unsupported allocators,
@@ -169,15 +158,8 @@ def trim_memory(
             return False
         now = time.monotonic()
         cooldown = configured_cooldown if cooldown_seconds is None else _cooldown_seconds(cooldown_seconds)
-        if not force and _last_trim_monotonic and now - _last_trim_monotonic < cooldown:
-            return False
-        # Even forced trims honor a short floor: AIAgent.close() forces a trim,
-        # and delegate batches close N child subagents back-to-back in the SAME
-        # process — without a floor that stacks N+1 uncooled full gc.collect()
-        # passes (50-500ms each in a large gateway process). 5s coalesces the
-        # burst while keeping the parent's final close-trim effective.
-        _FORCE_FLOOR_SECONDS = 5.0
-        if force and _last_trim_monotonic and now - _last_trim_monotonic < _FORCE_FLOOR_SECONDS:
+        since_last = now - _last_trim_monotonic
+        if _last_trim_monotonic and since_last < (_FORCE_FLOOR_SECONDS if force else cooldown):
             return False
         # Record the attempt before calling into libc so repeated failures do not
         # turn every turn boundary into an expensive full collection.
@@ -193,19 +175,16 @@ def trim_memory(
             _trim_call_count += 1
             if released and _should_log_trim(
                 force=force, log_every_n=log_every_n, call_count=_trim_call_count,
-                before=before, after=after, info_log_min_delta_mb=info_log_min_delta_mb,
-            ):
+                before=before, after=after, info_log_min_delta_mb=info_log_min_delta_mb):
                 logger.info(
                     "memory trim: reason=%s malloc_trim=%s rss_kib=%s->%s "
                     "rss_anon_kib=%s->%s threads=%s duration_ms=%.1f",
                     reason or "cleanup", trim_result,
                     before.get("rss_kib"), after.get("rss_kib"),
                     before.get("rss_anon_kib"), after.get("rss_anon_kib"),
-                    after.get("thread_count"), duration_ms,
-                )
+                    after.get("thread_count"), duration_ms)
             return released
         except Exception as exc:
             logger.warning(
-                "memory trim failed after %s: %s: %s", reason or "cleanup", type(exc).__name__, exc,
-            )
+                "memory trim failed after %s: %s: %s", reason or "cleanup", type(exc).__name__, exc)
             return False

hermes_cli/migrate.py

@@ -15,35 +15,29 @@ from hermes_cli.config import load_config
 
 def cmd_migrate(args: Any) -> int:
     """Dispatcher for ``hermes migrate <subtype>``."""
-    sub = getattr(args, "migrate_type", None)
-    if sub == "xai":
+    if getattr(args, "migrate_type", None) == "xai":
         return cmd_migrate_xai(args)
 
     print("usage: hermes migrate xai [--apply] [--no-backup]", file=sys.stderr)
     return 2
 
 
+def _fail(message: str) -> int:
+    print(f"  {color('✗', Colors.RED)} {message}", file=sys.stderr)
+    return 1
+
+
 def cmd_migrate_xai(args: Any) -> int:
     """Run xAI May-15 model migration in dry-run or apply mode."""
     from hermes_cli.xai_retirement import (
-        MIGRATION_GUIDE_URL,
-        RETIREMENT_DATE,
-        apply_migration,
-        find_retired_xai_refs,
-        format_issue,
-    )
+        MIGRATION_GUIDE_URL, RETIREMENT_DATE, apply_migration, find_retired_xai_refs, format_issue)
 
     apply = bool(getattr(args, "apply", False))
     no_backup = bool(getattr(args, "no_backup", False))
-
-    config = load_config()
-    issues = find_retired_xai_refs(config)
+    issues = find_retired_xai_refs(load_config())
 
     print()
-    print(color(
-        f"◆ xAI Model Retirement Migration ({RETIREMENT_DATE})",
-        Colors.CYAN, Colors.BOLD,
-    ))
+    print(color(f"◆ xAI Model Retirement Migration ({RETIREMENT_DATE})", Colors.CYAN, Colors.BOLD))
     print()
 
     if not issues:
@@ -65,30 +59,16 @@ def cmd_migrate_xai(args: Any) -> int:
         print(color(
             "Re-run with `hermes migrate xai --apply` to rewrite "
             f"{config_path} in-place (backup created automatically).",
-            Colors.DIM,
-        ))
+            Colors.DIM))
         return 0
 
     if not config_path or not config_path.exists():
-        print(
-            f"  {color('✗', Colors.RED)} Could not locate config.yaml "
-            f"(looked at: {config_path})",
-            file=sys.stderr,
-        )
-        return 1
+        return _fail(f"Could not locate config.yaml (looked at: {config_path})")
 
     try:
-        result = apply_migration(
-            config_path=config_path,
-            issues=issues,
-            backup=not no_backup,
-        )
+        result = apply_migration(config_path=config_path, issues=issues, backup=not no_backup)
     except Exception as exc:
-        print(
-            f"  {color('✗', Colors.RED)} Migration failed: {exc}",
-            file=sys.stderr,
-        )
-        return 1
+        return _fail(f"Migration failed: {exc}")
 
     if not result.config_changed:
         print(f"  {color('⚠', Colors.YELLOW)} No changes written.")
@@ -98,18 +78,13 @@ def cmd_migrate_xai(args: Any) -> int:
         print(f"  {color('✓', Colors.GREEN)} Backup: {result.backup_path}")
     print(
         f"  {color('✓', Colors.GREEN)} Updated {len(result.issues_resolved)} "
-        f"slot(s) in {result.file_path}"
-    )
+        f"slot(s) in {result.file_path}")
     print()
-    print(color(
-        "Run `hermes doctor` to confirm no retired xAI models remain.",
-        Colors.DIM,
-    ))
+    print(color("Run `hermes doctor` to confirm no retired xAI models remain.", Colors.DIM))
     return 0
 
 
 def _resolve_config_path() -> Path:
     """Best-effort: locate the active config.yaml on disk."""
     from hermes_cli.config import get_hermes_home
-
     return get_hermes_home() / "config.yaml"