From 62124aa69357e0b80e14bcb19f1e993f1f8d3cb0 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:00:35 -0700 Subject: [PATCH] =?UTF-8?q?refactor(hermes=5Fcli):=20profiles=20cluster=20?= =?UTF-8?q?=E2=80=94=20contextlib.suppress=20for=20best-effort=20blocks,?= =?UTF-8?q?=20=5Fexisting=5Fprofile=5Fdir=20helper,=20drop=20intra-functio?= =?UTF-8?q?n=20blanks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hermes_cli/profile_cmd.py | 39 ----- hermes_cli/profile_describer.py | 12 -- hermes_cli/profile_distribution.py | 12 -- hermes_cli/profiles.py | 238 ++++++----------------------- 4 files changed, 45 insertions(+), 256 deletions(-) diff --git a/hermes_cli/profile_cmd.py b/hermes_cli/profile_cmd.py index 32d9626149..a037a66e56 100644 --- a/hermes_cli/profile_cmd.py +++ b/hermes_cli/profile_cmd.py @@ -96,15 +96,12 @@ def _profile_status(args): """Bare ``hermes profile`` — show current profile status.""" from hermes_constants import display_hermes_home from hermes_cli.profiles import format_profile_label, get_active_profile_name, list_profiles - profile_name = get_active_profile_name() dhh = display_hermes_home() - current = next((p for p in list_profiles() if _is_active(p, profile_name)), None) label = format_profile_label(profile_name, current.display_name if current else "") print(f"\nActive profile: {label}") print(f"Path: {dhh}") - if current is not None: p = current if p.model: @@ -118,21 +115,17 @@ def _profile_status(args): def _profile_list(args): from hermes_cli.profiles import format_profile_label, get_active_profile_name, list_profiles - profiles = list_profiles() active = get_active_profile_name() - if not profiles: print("No profiles found.") return - print( f"\n {'Profile':<16} {'Model':<28} {'Gateway':<12} {'Alias':<12} {'Distribution'}" ) print( f" {'─' * 15} {'─' * 27} {'─' * 11} {'─' * 11} {'─' * 20}" ) - for p in profiles: marker = " ◆" if _is_active(p, active) else " " name = format_profile_label(p.name, p.display_name) @@ -146,7 +139,6 @@ def _profile_list(args): def _profile_use(args): from hermes_cli.profiles import set_active_profile - name = args.profile_name try: set_active_profile(name) @@ -160,7 +152,6 @@ def _profile_create(args): _get_wrapper_dir, _is_wrapper_dir_in_path, check_alias_collision, create_profile, create_wrapper_script, get_active_profile_name, seed_profile_skills, ) - name = args.profile_name clone = getattr(args, "clone", False) clone_all = getattr(args, "clone_all", False) @@ -169,7 +160,6 @@ def _profile_create(args): clone_from = getattr(args, "clone_from", None) clone_config = clone or clone_from is not None cloned = clone_config or clone_all - try: profile_dir = create_profile( name=name, clone_from=clone_from, clone_all=clone_all, clone_config=clone_config, @@ -178,7 +168,6 @@ def _profile_create(args): except (ValueError, FileExistsError, FileNotFoundError) as e: _die(f"Error: {e}") print(f"\nProfile '{name}' created at {profile_dir}") - if cloned: source_label = clone_from or get_active_profile_name() if clone_all: @@ -188,7 +177,6 @@ def _profile_create(args): # Auto-clone Honcho config for the new profile (only with clone operations) try: from plugins.memory.honcho.cli import clone_honcho_for_profile - if clone_honcho_for_profile(name): print(f"Honcho config cloned (peer: {name})") except Exception: @@ -204,7 +192,6 @@ def _profile_create(args): print(f"{len(result.get('copied', []))} bundled skills synced.") else: print(f"⚠ Skills could not be seeded. Run `{name} update` to retry.") - if not no_alias: collision = check_alias_collision(name) if collision: @@ -219,12 +206,10 @@ def _profile_create(args): print(f"\n⚠ {_get_wrapper_dir()} is not in your PATH.") print(" Add to your shell config (~/.bashrc or ~/.zshrc):") print(' export PATH="$HOME/.local/bin:$PATH"') - try: profile_dir_display = "~/" + profile_dir.relative_to(Path.home()).as_posix() except ValueError: profile_dir_display = str(profile_dir) - print("\nNext steps:") print(f" {name} setup Configure API keys and model") print(f" {name} chat Start chatting") @@ -241,7 +226,6 @@ def _profile_create(args): def _profile_delete(args): from hermes_cli.profiles import delete_profile - try: delete_profile(args.profile_name, yes=getattr(args, "yes", False)) except (ValueError, FileNotFoundError) as e: @@ -252,7 +236,6 @@ def _describe_target_dir(name: str) -> Path: """Profile dir for ``describe``: ``default`` maps to the CURRENT home (get_hermes_home), everything else to its named directory.""" from hermes_cli import profiles as _profiles_mod - if _profiles_mod.normalize_profile_name(name) == "default": from hermes_constants import get_hermes_home as _hh return Path(_hh()) @@ -261,13 +244,11 @@ def _describe_target_dir(name: str) -> Path: def _profile_describe(args): from hermes_cli import profiles as _profiles_mod - all_flag = bool(getattr(args, "all_missing", False)) auto_flag = bool(getattr(args, "auto", False)) overwrite_flag = bool(getattr(args, "overwrite", False)) text_value = getattr(args, "text", None) name = getattr(args, "profile_name", None) - if all_flag and not auto_flag: _die("profile describe: --all requires --auto", 2, err=True) if all_flag and (text_value or name): @@ -307,14 +288,12 @@ def _profile_describe(args): # --auto path: invoke the LLM describer. from hermes_cli import profile_describer as _pd - if all_flag: targets = _pd.list_describable_profiles(missing_only=True) if not targets: _die("All profiles already have descriptions.", 0) else: targets = [name] - ok_count = 0 for tgt in targets: outcome = _pd.describe_profile(tgt, overwrite=overwrite_flag) @@ -333,7 +312,6 @@ def _profile_show(args): _served_by_running_multiplexer, _count_skills, _read_distribution_meta, _wrapper_path, find_alias_for_profile, format_profile_label, read_profile_meta, ) - if not profile_exists(name): _die(f"Error: Profile '{name}' does not exist.") profile_dir = get_profile_dir(name) @@ -342,7 +320,6 @@ def _profile_show(args): dist_name, dist_version, dist_source = _read_distribution_meta(profile_dir) alias_name = find_alias_for_profile(name) display = read_profile_meta(profile_dir).get("display_name", "") - print(f"\nProfile: {format_profile_label(name, display)}") print(f"Path: {profile_dir}") if model: @@ -366,21 +343,16 @@ def _profile_alias(args): _get_wrapper_dir, _is_wrapper_dir_in_path, check_alias_collision, create_wrapper_script, profile_exists, remove_wrapper_script, validate_alias_name, ) - name = args.profile_name remove = getattr(args, "remove", False) custom_name = getattr(args, "alias_name", None) - if not profile_exists(name): _die(f"Error: Profile '{name}' does not exist.") - alias_name = custom_name or name - try: validate_alias_name(alias_name) except ValueError as exc: _die(f"Error: {exc}") - if remove: if remove_wrapper_script(alias_name): print(f"✓ Removed alias '{alias_name}'") @@ -399,7 +371,6 @@ def _profile_alias(args): def _profile_rename(args): from hermes_cli.profiles import normalize_profile_name, rename_profile - try: new_dir = rename_profile(args.old_name, args.new_name) if normalize_profile_name(args.old_name) != "default": @@ -411,7 +382,6 @@ def _profile_rename(args): def _profile_export(args): from hermes_cli.profiles import export_profile, get_profile_export_path - name = args.profile_name try: output = args.output or str(get_profile_export_path(name)) @@ -423,12 +393,10 @@ def _profile_export(args): def _profile_import(args): from hermes_cli.profiles import check_alias_collision, create_wrapper_script, import_profile - try: profile_dir = import_profile(args.archive, name=getattr(args, "import_name", None)) name = profile_dir.name print(f"✓ Imported profile '{name}' at {profile_dir}") - if not check_alias_collision(name): wrapper_path = create_wrapper_script(name) if wrapper_path: @@ -441,18 +409,15 @@ def _profile_import(args): def _profile_install(args): import tempfile from hermes_cli.profile_distribution import DistributionError, install_distribution, plan_install - try: # Preview: stage into a scratch dir, show the manifest, then do the real install. # The double-stage avoids any side-effects if the user declines. with tempfile.TemporaryDirectory(prefix="hermes_dist_preview_") as tmp: plan = plan_install(args.source, Path(tmp), override_name=getattr(args, "install_name", None)) _render_distribution_plan(plan) - if not getattr(args, "yes", False) and not _confirm("\nProceed with install? [y/N] "): print("Install cancelled.") return - plan = install_distribution( args.source, name=getattr(args, "install_name", None), force=getattr(args, "force", False), create_alias=getattr(args, "alias", False), @@ -477,7 +442,6 @@ def _profile_install(args): def _profile_update(args): from hermes_cli.profile_distribution import DistributionError, read_manifest, update_distribution from hermes_cli.profiles import get_profile_dir, normalize_profile_name - try: canon = normalize_profile_name(args.profile_name) current = read_manifest(get_profile_dir(canon)) @@ -486,7 +450,6 @@ def _profile_update(args): f"Error: Profile '{canon}' is not a distribution (no distribution.yaml). " "Only profiles installed via `hermes profile install` can be updated." ) - force_config = getattr(args, "force_config", False) if not getattr(args, "yes", False): print(f"\nUpdate '{canon}' from: {current.source or '(no source)'}") @@ -499,7 +462,6 @@ def _profile_update(args): if not _confirm("\nProceed? [y/N] "): print("Update cancelled.") return - plan = update_distribution(canon, force_config=force_config) print(f"\n✓ Updated '{plan.manifest.name}' → v{plan.manifest.version}") if plan.has_cron: @@ -520,7 +482,6 @@ _INFO_FIELDS = ( def _profile_info(args): from hermes_cli.profile_distribution import describe_distribution, DistributionError - try: data = describe_distribution(args.profile_name) except (DistributionError, ValueError) as e: diff --git a/hermes_cli/profile_describer.py b/hermes_cli/profile_describer.py index f92b2813c0..3b5de354c4 100644 --- a/hermes_cli/profile_describer.py +++ b/hermes_cli/profile_describer.py @@ -106,7 +106,6 @@ def _sample_skills(names: list[str]) -> list[str]: def _extract_json_blob(raw: str) -> Optional[dict]: from hermes_cli.kanban_specify import _extract_json_blob as _extract - return _extract(raw, _FENCE_RE) @@ -121,7 +120,6 @@ def describe_profile( canon = profiles_mod.normalize_profile_name(profile_name) if not profiles_mod.profile_exists(canon): # handles the virtual "default" name return DescribeOutcome(canon, False, "profile not found") - try: if canon == "default": from hermes_constants import get_hermes_home # type: ignore @@ -130,32 +128,26 @@ def describe_profile( profile_dir = profiles_mod.get_profile_dir(canon) except Exception as exc: return DescribeOutcome(canon, False, f"cannot resolve profile dir: {exc}") - existing = profiles_mod.read_profile_meta(profile_dir) if existing.get("description") and not existing.get("description_auto") and not overwrite: return DescribeOutcome( canon, False, "profile already has a user-authored description (use --overwrite to replace)" ) - all_skills = _collect_skills(profile_dir) skill_list = "\n".join(f" - {n}" for n in _sample_skills(all_skills)) or " (no skills installed)" - try: model, provider = profiles_mod._read_config_model(profile_dir) except Exception: model, provider = None, None - try: from agent.auxiliary_client import call_llm # type: ignore except Exception as exc: logger.debug("describe: auxiliary client import failed: %s", exc) return DescribeOutcome(canon, False, "auxiliary client unavailable") - user_msg = _USER_TEMPLATE.format( name=canon, model=(model or "(unset)"), provider=(provider or "(unset)"), skill_count=len(all_skills), skill_cap=MAX_SKILLS_FOR_PROMPT, skill_list=skill_list, ) - try: # call_llm applies auxiliary.profile_describer.* config (provider/model/base_url, # extra_body, reasoning_effort, retries); the direct-create path dropped extra_body. @@ -171,12 +163,10 @@ def describe_profile( except Exception as exc: logger.info("describe: API call failed for %s (%s)", canon, exc) return DescribeOutcome(canon, False, f"LLM error: {type(exc).__name__}") - try: raw = resp.choices[0].message.content or "" except Exception: raw = "" - parsed = _extract_json_blob(raw) if parsed is None: # Fall back: raw text trimmed to one paragraph. @@ -191,14 +181,12 @@ def describe_profile( canon, False, "LLM response missing 'description' field" ) description = val.strip()[:280] - try: profiles_mod.write_profile_meta( profile_dir, description=description, description_auto=True ) except Exception as exc: return DescribeOutcome(canon, False, f"failed to write profile.yaml: {exc}") - return DescribeOutcome(canon, True, "described", description=description) diff --git a/hermes_cli/profile_distribution.py b/hermes_cli/profile_distribution.py index 408f64311c..af5bd528f8 100644 --- a/hermes_cli/profile_distribution.py +++ b/hermes_cli/profile_distribution.py @@ -260,7 +260,6 @@ def _stage_source(source: str, workdir: Path) -> Tuple[Path, str]: """Resolve *source* to ``(staged_dir, provenance)``: git URLs are shallow-cloned into *workdir* (``.git`` removed); a local directory is used in place.""" src_str = source.strip() - if _looks_like_git_url(src_str): staged, provenance = workdir / "clone", src_str _git_clone(src_str, staged) @@ -331,7 +330,6 @@ def plan_install( get_profile_dir, normalize_profile_name, validate_profile_name ) from hermes_cli import __version__ as hermes_version - staged, provenance = _stage_source(source, workdir) _reject_distribution_symlinks(staged) manifest = read_manifest(staged) @@ -339,9 +337,7 @@ def plan_install( raise DistributionError( f"No {MANIFEST_FILENAME} found at the distribution root — this source is not a Hermes distribution." ) - check_hermes_requires(manifest.hermes_requires, hermes_version) # fail fast - canon = normalize_profile_name(override_name or manifest.name) validate_profile_name(canon) if canon == "default": @@ -353,7 +349,6 @@ def plan_install( manifest.source = provenance # Stamped once here so both fresh install and update propagate a fresh timestamp. manifest.installed_at = datetime.now(timezone.utc).isoformat(timespec="seconds") - target_dir = get_profile_dir(canon) existing = target_dir.is_dir() return InstallPlan( @@ -430,7 +425,6 @@ def _copy_dist_payload( def _bootstrap_user_dirs(target: Path) -> None: """Create the bootstrap dirs a fresh profile expects (same set as ``create_profile``).""" from hermes_cli.profiles import _PROFILE_DIRS - for d in _PROFILE_DIRS: (target / d).mkdir(parents=True, exist_ok=True) @@ -443,10 +437,8 @@ def install_distribution( from hermes_cli.profiles import ( check_alias_collision, create_wrapper_script ) - with tempfile.TemporaryDirectory(prefix="hermes_dist_install_") as tmp: plan = plan_install(source, Path(tmp), override_name=name) - if plan.existing and not force: raise DistributionError( f"Profile '{plan.manifest.name}' already exists at {plan.target_dir}. " @@ -456,10 +448,8 @@ def install_distribution( # Fresh install: config.yaml comes from the distribution. _bootstrap_user_dirs(plan.target_dir) _copy_dist_payload(plan.staged_dir, plan.target_dir, plan.manifest, preserve_config=False) - if create_alias and check_alias_collision(plan.manifest.name) is None: create_wrapper_script(plan.manifest.name) - return plan @@ -468,7 +458,6 @@ def _existing_profile(profile_name: str) -> Tuple[str, Path]: from hermes_cli.profiles import ( get_profile_dir, normalize_profile_name, validate_profile_name ) - canon = normalize_profile_name(profile_name) validate_profile_name(canon) target = get_profile_dir(canon) @@ -494,7 +483,6 @@ def update_distribution( f"Profile '{canon}' has no recorded source. Re-install with " "`hermes profile install --name {canon} --force`." ) - with tempfile.TemporaryDirectory(prefix="hermes_dist_update_") as tmp: plan = plan_install(existing_manifest.source, Path(tmp), override_name=canon) plan.preserves_config = not force_config diff --git a/hermes_cli/profiles.py b/hermes_cli/profiles.py index 7469901a4d..80eebc148f 100644 --- a/hermes_cli/profiles.py +++ b/hermes_cli/profiles.py @@ -1,5 +1,6 @@ """Profile management for multiple isolated Hermes instances.""" +import contextlib import json import logging import os @@ -126,9 +127,7 @@ _HERMES_SUBCOMMANDS = frozenset({ }) -# --------------------------------------------------------------------------- # Path helpers -# --------------------------------------------------------------------------- def _get_profiles_root() -> Path: """Named-profiles root, anchored to the hermes root (NOT the current HERMES_HOME, which @@ -170,9 +169,7 @@ def _missing_profile_error(canon: str) -> FileNotFoundError: ) -# --------------------------------------------------------------------------- # Validation -# --------------------------------------------------------------------------- def normalize_profile_name(name: str) -> str: """Canonical profile id used on disk and in ``-p`` argv: lowercase, ``default`` matched @@ -221,6 +218,15 @@ def _canon_valid(name: str) -> str: return canon +def _existing_profile_dir(name: str) -> Tuple[str, Path]: + """``(canon, profile_dir)`` for an existing profile; FileNotFoundError otherwise.""" + canon = _canon_valid(name) + profile_dir = get_profile_dir(canon) + if not profile_dir.is_dir(): + raise FileNotFoundError(f"Profile '{canon}' does not exist.") + return canon, profile_dir + + def get_profile_dir(name: str) -> Path: """Resolve a profile name to its HERMES_HOME directory.""" canon = normalize_profile_name(name) @@ -248,7 +254,6 @@ def profile_matches_home(name: str, home: "Path | None" = None) -> bool: target = get_profile_dir(name) if home is None: from hermes_constants import get_hermes_home - home = get_hermes_home() return ( Path(target).expanduser().resolve(strict=False) == Path(home).expanduser().resolve(strict=False) @@ -275,16 +280,12 @@ def list_profile_names() -> List[str]: """Cheap name-only listing (``default`` + profile dirs). Unlike :func:`list_profiles` this reads NO per-profile config — safe for hot paths (cron target listings, create validation).""" names = ["default"] - try: + with contextlib.suppress(OSError): names.extend(entry.name for entry in _iter_named_profile_dirs(live_only=False)) - except OSError: - pass return names -# --------------------------------------------------------------------------- # Alias / wrapper script management -# --------------------------------------------------------------------------- def check_alias_collision(name: str) -> Optional[str]: """Return a human-readable collision message, or None if the name is safe.""" @@ -297,7 +298,6 @@ def check_alias_collision(name: str) -> Optional[str]: return f"'{canon}' is a reserved name" if canon in _HERMES_SUBCOMMANDS: return f"'{canon}' conflicts with a hermes subcommand" - try: result = subprocess.run( ["where" if sys.platform == "win32" else "which", canon], @@ -311,7 +311,6 @@ def check_alias_collision(name: str) -> Optional[str]: return f"'{canon}' conflicts with an existing command ({existing_path})" except (FileNotFoundError, subprocess.TimeoutExpired): pass - return None # safe @@ -331,7 +330,6 @@ def create_wrapper_script(name: str, target: Optional[str] = None) -> Optional[P except OSError as e: print(f"⚠ Could not create {wrapper_dir}: {e}") return None - wrapper_path = _wrapper_path(canon) try: if sys.platform == "win32": @@ -359,14 +357,11 @@ def remove_wrapper_script(name: str) -> bool: candidates = [_get_wrapper_dir() / canon] if sys.platform == "win32": candidates.insert(0, _get_wrapper_dir() / f"{canon}.bat") - for wrapper_path in candidates: if wrapper_path.exists() and _is_our_wrapper(wrapper_path): - try: + with contextlib.suppress(Exception): wrapper_path.unlink() return True - except Exception: - pass return False @@ -375,11 +370,11 @@ def _migrate_profile_config_if_outdated(profile_dir: Path) -> None: profile); otherwise the first desktop/doctor view shows a scary ``v0 -> latest`` warning.""" if not (profile_dir / "config.yaml").exists(): return - - try: + # Creation must not fail over an unmigratable old config; `hermes doctor --fix` surfaces + # the detailed error in the target profile. + with contextlib.suppress(Exception): from hermes_constants import reset_hermes_home_override, set_hermes_home_override from hermes_cli.config import check_config_version, migrate_config - token = set_hermes_home_override(str(profile_dir)) try: current_ver, latest_ver = check_config_version() @@ -387,10 +382,6 @@ def _migrate_profile_config_if_outdated(profile_dir: Path) -> None: migrate_config(interactive=False, quiet=True) finally: reset_hermes_home_override(token) - except Exception: - # Creation must not fail over an unmigratable old config; `hermes doctor --fix` - # surfaces the detailed error in the target profile. - pass def find_alias_for_profile(profile_name: str) -> Optional[str]: @@ -419,7 +410,6 @@ def build_alias_map() -> dict[str, str]: return result is_windows = sys.platform == "win32" prefix = "hermes -p " - for entry in sorted(wrapper_dir.iterdir()): if not entry.is_file(): continue @@ -450,9 +440,7 @@ def build_alias_map() -> dict[str, str]: return result -# --------------------------------------------------------------------------- # ProfileInfo -# --------------------------------------------------------------------------- @dataclass class ProfileInfo: @@ -489,8 +477,7 @@ def _load_yaml_dict(path: Path) -> Optional[dict]: return None try: import yaml - with open(path, "r", encoding="utf-8") as f: - data = yaml.safe_load(f) or {} + data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} except Exception: return None return data if isinstance(data, dict) else None @@ -512,15 +499,14 @@ def _read_config_model(profile_dir: Path) -> tuple: try: # load_config() targets the ACTIVE profile's home; read THIS profile's file raw. from hermes_cli.config import read_user_config_raw - cfg = read_user_config_raw(config_path) - model_cfg = cfg.get("model", {}) + model_cfg = read_user_config_raw(config_path).get("model", {}) if isinstance(model_cfg, str): return model_cfg, None if isinstance(model_cfg, dict): return model_cfg.get("default") or model_cfg.get("model"), model_cfg.get("provider") - return None, None except Exception: - return None, None + pass + return None, None def _seed_model_config(profile_dir: Path) -> None: @@ -529,22 +515,14 @@ def _seed_model_config(profile_dir: Path) -> None: config_path = profile_dir / "config.yaml" if config_path.exists(): return - try: + with contextlib.suppress(Exception): # creation must not fail over this; `hermes model` sets it later import yaml from hermes_constants import get_hermes_home from hermes_cli.config import read_user_config_raw - source = get_hermes_home() / "config.yaml" - if not source.is_file(): - return - model_cfg = read_user_config_raw(source).get("model") - if not model_cfg: - return - config_path.write_text( - yaml.safe_dump({"model": model_cfg}, sort_keys=False), encoding="utf-8" - ) - except Exception: - pass # creation must not fail over this; `hermes model` sets it later + model_cfg = read_user_config_raw(source).get("model") if source.is_file() else None + if model_cfg: + config_path.write_text(yaml.safe_dump({"model": model_cfg}, sort_keys=False), encoding="utf-8") def _check_gateway_running(profile_dir: Path) -> bool: @@ -574,7 +552,6 @@ def _served_by_running_multiplexer(profile_name: str) -> bool: gateway.pid of its own, so ``_check_gateway_running`` alone reports it stopped).""" try: from hermes_cli.gateway import named_profile_served_by_running_multiplexer - return named_profile_served_by_running_multiplexer(profile_name) except Exception: return False @@ -601,9 +578,7 @@ def _skills_dir_signature(skills_dir: Path) -> float: for entry in it: try: if entry.is_dir(follow_symlinks=False): - m = entry.stat(follow_symlinks=False).st_mtime - if m > sig: - sig = m + sig = max(sig, entry.stat(follow_symlinks=False).st_mtime) except OSError: continue except OSError: @@ -616,7 +591,6 @@ def _count_skills(profile_dir: Path) -> int: skills_dir = profile_dir / "skills" if not skills_dir.is_dir(): return 0 - key = str(skills_dir) signature = _skills_dir_signature(skills_dir) now = time.time() @@ -625,15 +599,12 @@ def _count_skills(profile_dir: Path) -> int: cached is not None and cached[0] == signature and (now - cached[1]) < _SKILL_COUNT_TTL_SECONDS ): return cached[2] - count = sum(1 for md in skills_dir.rglob("SKILL.md") if not is_excluded_skill_path(md)) _SKILL_COUNT_CACHE[key] = (signature, now, count) return count -# --------------------------------------------------------------------------- # profile.yaml — per-profile metadata (description, role, etc.) -# --------------------------------------------------------------------------- # Deliberately tiny and separate from ``config.yaml`` (user-facing Hermes config, ~5000 # lines of defaults): this is metadata ABOUT the profile. Missing file -> empty defaults, # never an error; the kanban decomposer falls back to the profile name. @@ -643,9 +614,7 @@ def read_profile_meta(profile_dir: Path) -> dict: """Read ``profile.yaml`` -> ``{description, description_auto, display_name}`` (empty defaults when missing/unreadable). Never raises — a corrupt file on one profile must not break ``hermes profile list``.""" - data = _load_yaml_dict(profile_dir / "profile.yaml") - if data is None: - return {"description": "", "description_auto": False, "display_name": ""} + data = _load_yaml_dict(profile_dir / "profile.yaml") or {} return { "description": str(data.get("description") or "").strip(), "description_auto": bool(data.get("description_auto", False)), @@ -676,7 +645,6 @@ def write_profile_meta( # Atomic write: bare open("w") truncates before the dump, and the read path swallows # parse errors as {}, so a crashed write would silently drop unspecified fields. from utils import atomic_yaml_write - atomic_yaml_write(path, existing, sort_keys=False) @@ -690,10 +658,7 @@ def format_profile_label(name: str, display_name: Optional[str]) -> str: def set_profile_display_name(profile_name: str, display_name: str) -> str: """Set (or clear, with ``""``) a presentation-only display name. Returns the stored value; raises ``ValueError`` over 64 chars.""" - canon = _canon_valid(profile_name) - profile_dir = get_profile_dir(canon) - if not profile_dir.is_dir(): - raise FileNotFoundError(f"Profile '{canon}' does not exist.") + canon, profile_dir = _existing_profile_dir(profile_name) cleaned = (display_name or "").strip() if len(cleaned) > 64: raise ValueError(f"Display name too long ({len(cleaned)} chars, max 64).") @@ -701,9 +666,7 @@ def set_profile_display_name(profile_name: str, display_name: str) -> str: return cleaned -# --------------------------------------------------------------------------- # CRUD operations -# --------------------------------------------------------------------------- def _profile_info(name: str, path: Path, *, is_default: bool, alias_name: Optional[str] = None) -> ProfileInfo: """Build one :class:`ProfileInfo` from a profile directory.""" @@ -721,8 +684,7 @@ def _profile_info(name: str, path: Path, *, is_default: bool, alias_name: Option provider=provider, has_env=(path / ".env").exists(), skill_count=_count_skills(path), alias_path=alias_path, alias_name=alias_name, distribution_name=dist_name, distribution_version=dist_version, distribution_source=dist_source, - description=meta.get("description", ""), description_auto=meta.get("description_auto", False), - display_name=meta.get("display_name", ""), + **meta, ) @@ -732,7 +694,6 @@ def list_profiles() -> List[ProfileInfo]: default_home = _get_default_hermes_home() if default_home.is_dir(): profiles.append(_profile_info("default", default_home, is_default=True)) - named = _iter_named_profile_dirs() if named: alias_map = build_alias_map() # ONCE, not per profile (was the dominant cost) @@ -755,7 +716,6 @@ def profiles_to_serve( active = get_active_profile_name() or "default" if not multiplex: return [(active, get_profile_dir(active))] - serve: List[Tuple[str, Path]] = [("default", _get_default_hermes_home())] allowed: Optional[set[str]] = None if profile_allowlist is not None: @@ -769,11 +729,9 @@ def profiles_to_serve( continue if name != "default": allowed.add(name) - for entry in _iter_named_profile_dirs(): if allowed is None or entry.name in allowed: serve.append((entry.name, entry)) - if allowed is not None: missing = tuple(sorted(allowed - {name for name, _ in serve})) if missing and missing not in _WARNED_MISSING_ALLOWLIST_ENTRIES: @@ -781,7 +739,6 @@ def profiles_to_serve( logger.warning( "Skipping missing gateway.multiplex_profile_allowlist profile(s): %s", ", ".join(missing) ) - return serve @@ -804,12 +761,10 @@ def _seed_file_if_missing(path: Path, text: str, mode: Optional[int] = None) -> """Best-effort: write *text* to *path* unless it already exists; never raises.""" if path.exists(): return - try: + with contextlib.suppress(OSError): path.write_text(text, encoding="utf-8") if mode is not None: os.chmod(str(path), mode) - except OSError: - pass def _clone_file(source_dir: Path, profile_dir: Path, relpath: str) -> None: @@ -822,10 +777,8 @@ def _clone_file(source_dir: Path, profile_dir: Path, relpath: str) -> None: dst.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(src, dst) if relpath == ".env": - try: + with contextlib.suppress(OSError): os.chmod(str(dst), 0o600) - except OSError: - pass def _clone_all_into(source_dir: Path, profile_dir: Path, canon: str) -> None: @@ -856,7 +809,6 @@ def _bootstrap_profile_dir(profile_dir: Path, source_dir: Optional[Path]) -> Non profile_dir.mkdir(parents=True, exist_ok=True) for subdir in _PROFILE_DIRS: (profile_dir / subdir).mkdir(parents=True, exist_ok=True) - if source_dir is None: _seed_model_config(profile_dir) return @@ -885,12 +837,10 @@ def create_profile( "(cloning explicitly copies skills from the source profile)." ) canon = _canon_valid(name) - if canon == "default": raise ValueError( "Cannot create a profile named 'default' — it is the built-in profile (~/.hermes)." ) - profile_dir = get_profile_dir(canon) if profile_dir.exists() and named_profile_is_deleted(profile_dir): # Empty shells left by post-delete mkdir may be replaced. Identity files mean the @@ -901,11 +851,9 @@ def create_profile( if profile_dir.exists(): raise FileExistsError(f"Profile '{canon}' already exists at {profile_dir}") clear_named_profile_deleted(profile_dir) - source_dir = None if clone_from is not None or clone_all or clone_config: source_dir = _resolve_clone_source(clone_from) - if clone_all and source_dir: _clone_all_into(source_dir, profile_dir, canon) else: @@ -918,11 +866,9 @@ def create_profile( _seed_file_if_missing(profile_dir / ".env", _PLACEHOLDER_ENV, 0o600) # Default SOUL.md to customize immediately (skipped when a clone already provided one). - try: + with contextlib.suppress(Exception): # best-effort — don't fail profile creation over this from hermes_cli.default_soul import DEFAULT_SOUL_MD _seed_file_if_missing(profile_dir / "SOUL.md", DEFAULT_SOUL_MD) - except Exception: - pass # best-effort — don't fail profile creation over this # Opt-out marker read by seed_profile_skills() and `hermes update`'s all-profile sync # (the feature still works via the empty skills/ dir if this fails). @@ -941,18 +887,13 @@ def create_profile( # Description last, so a partial-create failure doesn't strand a description file. if description and description.strip(): - try: - write_profile_meta( - profile_dir, description=description.strip(), description_auto=False - ) - except Exception: - pass # non-fatal — user can describe later with `hermes profile describe` + with contextlib.suppress(Exception): # non-fatal — `hermes profile describe` works later + write_profile_meta(profile_dir, description=description.strip(), description_auto=False) # Inside a container under s6, register the gateway as a runtime s6 service so # `hermes -p gateway start` supervises via `s6-svc -u` instead of a bare # process. No-op on host (systemd/launchd/windows unit generation handles lifecycle). _maybe_register_gateway_service(canon) - return profile_dir @@ -992,7 +933,6 @@ def backfill_profile_envs(quiet: bool = False) -> List[str]: the placeholder header). Never overwrites an existing profile ``.env``.""" backfilled: List[str] = [] default_env = _get_default_hermes_home() / ".env" - for entry in _iter_named_profile_dirs(): env_path = entry / ".env" if env_path.exists(): @@ -1007,7 +947,6 @@ def backfill_profile_envs(quiet: bool = False) -> List[str]: except OSError as e: if not quiet: print(f"⚠ Could not seed .env for profile '{entry.name}': {e}") - return backfilled @@ -1055,7 +994,6 @@ def _profile_bound_backend_pids(canon: str, profile_dir: Path) -> list[int]: import psutil # type: ignore except Exception: return [] - try: resolved_dir = profile_dir.resolve() except OSError: @@ -1064,21 +1002,16 @@ def _profile_bound_backend_pids(canon: str, profile_dir: Path) -> list[int]: # Never terminate ourselves or a parent (`hermes -p profile delete` runs under # the very profile it's deleting). skip: set[int] = {os.getpid()} - try: + with contextlib.suppress(Exception): parent = psutil.Process(os.getpid()).parent() while parent is not None: skip.add(parent.pid) parent = parent.parent() - except Exception: - pass - try: current_user = psutil.Process(os.getpid()).username() except Exception: current_user = None - pids: list[int] = [] - for proc in psutil.process_iter(["pid", "name", "username", "cmdline"]): try: info = proc.info @@ -1087,7 +1020,6 @@ def _profile_bound_backend_pids(canon: str, profile_dir: Path) -> list[int]: continue if current_user is not None and info.get("username") != current_user: continue - argv = info.get("cmdline") or [] if not argv or not _is_hermes_argv(argv): continue @@ -1099,18 +1031,13 @@ def _profile_bound_backend_pids(canon: str, profile_dir: Path) -> list[int]: normalize_profile_name(sel) == canon for sel in _argv_profile_selectors(argv) ) if not bound: - try: + with contextlib.suppress(Exception): # environ() can raise AccessDenied even same-user env_home = (proc.environ() or {}).get("HERMES_HOME", "") - if env_home and Path(env_home).resolve() == resolved_dir: - bound = True - except Exception: - pass # environ() can raise AccessDenied even same-user; argv signal only - + bound = bool(env_home) and Path(env_home).resolve() == resolved_dir if bound: pids.append(pid) except Exception: continue # NoSuchProcess / AccessDenied / ZombieProcess and anything else - return pids @@ -1119,19 +1046,14 @@ def _wait_then_force_kill(pids: List[int], start_times: dict, *, wait: float = 1 to exit, then force-kill stragglers. True when every pid exited gracefully. ``start_times`` pins each force kill to the same process incarnation (PID reuse guard).""" from gateway.status import _pid_exists, get_process_start_time, terminate_pid - for _ in range(int(wait / 0.5)): time.sleep(0.5) if not any(_pid_exists(pid) for pid in pids): return True for pid in pids: if _pid_exists(pid): - try: - terminate_pid( - pid, force=True, expected_start_time=start_times.get(pid, get_process_start_time(pid)) - ) - except (ProcessLookupError, PermissionError, OSError): - pass + with contextlib.suppress(ProcessLookupError, PermissionError, OSError): + terminate_pid(pid, force=True, expected_start_time=start_times.get(pid, get_process_start_time(pid))) return False @@ -1146,14 +1068,12 @@ def _stop_profile_backends(canon: str, profile_dir: Path) -> None: from gateway.status import terminate_pid except Exception: return - for pid in pids: try: terminate_pid(pid) # graceful first except (ProcessLookupError, PermissionError, OSError): continue _wait_then_force_kill(pids, {}) - print(f"✓ Stopped {len(pids)} profile backend process(es)") @@ -1167,10 +1087,8 @@ def _rmtree_make_writable(func, path, exc): raise for target in (path, os.path.dirname(path)): # parent needed for unlink/rmdir if target: - try: + with contextlib.suppress(OSError): os.chmod(target, os.stat(target).st_mode | stat.S_IWUSR) - except OSError: - pass func(path) @@ -1202,7 +1120,6 @@ def _print_delete_summary(canon: str, profile_dir: Path, gw_running: bool, wrapp model, provider = _read_config_model(profile_dir) skill_count = _count_skills(profile_dir) dist_name, dist_version, dist_source = _read_distribution_meta(profile_dir) - print(f"\nProfile: {canon}") print(f"Path: {profile_dir}") if model: @@ -1213,7 +1130,6 @@ def _print_delete_summary(canon: str, profile_dir: Path, gw_running: bool, wrapp print(f"Distribution: {dist_name}@{dist_version or '?'}") if dist_source: print(f"Installed from: {dist_source}") - print("\nThis will permanently delete:") print(" • All config, API keys, memories, sessions, skills, cron jobs") if wrapper_path is not None: @@ -1225,22 +1141,16 @@ def _print_delete_summary(canon: str, profile_dir: Path, gw_running: bool, wrapp def delete_profile(name: str, yes: bool = False) -> Path: """Delete a profile, its wrapper script, and its gateway service (service disabled first to prevent auto-restart, gateway stopped if running).""" - canon = _canon_valid(name) - + canon = normalize_profile_name(name) if canon == "default": raise ValueError( "Cannot delete the default profile (~/.hermes).\nTo remove everything, use: hermes uninstall" ) - - profile_dir = get_profile_dir(canon) - if not profile_dir.is_dir(): - raise FileNotFoundError(f"Profile '{canon}' does not exist.") - + canon, profile_dir = _existing_profile_dir(canon) gw_running = _check_gateway_running(profile_dir) wrapper_path = _get_wrapper_dir() / canon has_wrapper = wrapper_path.exists() _print_delete_summary(canon, profile_dir, gw_running, wrapper_path if has_wrapper else None) - if not yes: print() try: @@ -1270,14 +1180,11 @@ def delete_profile(name: str, yes: bool = False) -> Path: # Desktop's main serve process opens memory_store.db for every profile and is # deliberately not stopped above; on Windows its handles fail rmtree with WinError 32. # Inside serve (DELETE /api/profiles/) the handles live here; from the CLI no-op. - try: + with contextlib.suppress(Exception): # best-effort: never block the delete on the release path from plugins.memory.holographic.store import MemoryStore as _MemoryStore - _released = _MemoryStore.release_all_under(profile_dir) if _released: print(f"✓ Released {_released} memory-store connection(s) held by this process") - except Exception: - pass # best-effort: never block the delete on the release path # 3. Remove wrapper script if has_wrapper and remove_wrapper_script(canon): @@ -1294,10 +1201,8 @@ def delete_profile(name: str, yes: bool = False) -> Path: # 5. Clear active_profile if it pointed to this profile _retarget_active_profile(canon, "default", "✓ Active profile reset to default") - if remove_error is not None: raise RuntimeError(f"Could not remove profile directory {profile_dir}: {remove_error}") from remove_error - print(f"\nProfile '{canon}' deleted.") return profile_dir @@ -1372,10 +1277,9 @@ def _cleanup_gateway_service(name: str, profile_dir: Path) -> None: except Exception as e: print(f"⚠ Service cleanup: {e}") finally: + os.environ.pop("HERMES_HOME", None) if old_home is not None: os.environ["HERMES_HOME"] = old_home - else: - os.environ.pop("HERMES_HOME", None) def _stop_gateway_process(profile_dir: Path) -> None: @@ -1383,7 +1287,6 @@ def _stop_gateway_process(profile_dir: Path) -> None: pid_file = profile_dir / "gateway.pid" if not pid_file.exists(): return - try: raw = pid_file.read_text(encoding="utf-8").strip() data = json.loads(raw) if raw.startswith("{") else {"pid": int(raw)} @@ -1394,7 +1297,6 @@ def _stop_gateway_process(profile_dir: Path) -> None: from gateway.status import ( get_process_start_time, recorded_gateway_home_conflicts, terminate_pid ) - if recorded_gateway_home_conflicts(data, expected_home=profile_dir): print( f"✗ Refusing to stop PID {pid}: its recorded HERMES_HOME " @@ -1418,9 +1320,7 @@ def _stop_gateway_process(profile_dir: Path) -> None: print(f"⚠ Could not stop gateway: {e}") -# --------------------------------------------------------------------------- # Active profile (sticky default) -# --------------------------------------------------------------------------- def get_active_profile() -> str: """Read the sticky active profile name.""" @@ -1436,7 +1336,6 @@ def set_active_profile(name: str) -> None: canon = _canon_valid(name) if canon != "default" and not profile_exists(canon): raise _missing_profile_error(canon) - path = _get_active_profile_path() path.parent.mkdir(parents=True, exist_ok=True) if canon == "default": @@ -1449,12 +1348,10 @@ def set_active_profile(name: str) -> None: def _retarget_active_profile(old: str, new: str, message: str) -> None: """If the sticky active profile is *old*, point it at *new* and print *message*. Never raises.""" - try: + with contextlib.suppress(Exception): if get_active_profile() == old: set_active_profile(new) print(message) - except Exception: - pass def get_active_profile_name() -> str: @@ -1462,10 +1359,8 @@ def get_active_profile_name() -> str: name under ``~/.hermes/profiles/``, ``"custom"`` for any other path.""" from hermes_constants import get_hermes_home resolved = get_hermes_home().resolve() - if resolved == _get_default_hermes_home().resolve(): return "default" - profiles_root = _get_profiles_root().resolve() try: parts = resolved.relative_to(profiles_root).parts @@ -1473,13 +1368,10 @@ def get_active_profile_name() -> str: return parts[0] except ValueError: pass - return "custom" -# --------------------------------------------------------------------------- # Export / Import -# --------------------------------------------------------------------------- def _inside_git_checkout(path: Path) -> bool: """True when *path* lies inside a Git checkout. Walks the path's OWN resolved ancestry @@ -1497,7 +1389,6 @@ def _inside_git_checkout(path: Path) -> bool: def _profile_export_directory() -> Path: """Choose an export directory that cannot become source-tree input.""" import tempfile - export_dir = _get_default_hermes_home() / "profile-exports" if not _inside_git_checkout(export_dir): return export_dir @@ -1592,7 +1483,6 @@ def _scrub_export_secrets(staged: Path) -> None: sessions export --redact``). Runs on the staged copy only; symlinks to text files are materialized when content changes so redaction never follows a link back into the source.""" from agent.redact import redact_sensitive_text - for path in staged.rglob("*"): try: is_link = path.is_symlink() @@ -1600,19 +1490,15 @@ def _scrub_export_secrets(staged: Path) -> None: continue except OSError: continue - if not _should_redact_export_file(path): continue - try: text = path.read_text(encoding="utf-8") except (UnicodeDecodeError, OSError): continue - redacted = redact_sensitive_text(text, force=True) if redacted == text: continue - if is_link: path.unlink() path.write_text(redacted, encoding="utf-8") @@ -1622,12 +1508,7 @@ def export_profile(name: str, output_path: str, extra_files: Optional[Dict[str, """Export a profile to a tar.gz archive; credential files are excluded and staged text is force-redacted first. Returns the output file path.""" import tempfile - - canon = _canon_valid(name) - profile_dir = get_profile_dir(canon) - if not profile_dir.is_dir(): - raise FileNotFoundError(f"Profile '{canon}' does not exist.") - + canon, profile_dir = _existing_profile_dir(name) # Archive base name without extension (.tar.gz appended by the writer). base = str(Path(output_path)).removesuffix(".tar.gz").removesuffix(".tgz") @@ -1652,11 +1533,9 @@ def export_profile(name: str, output_path: str, extra_files: Optional[Dict[str, def import_profile(archive_path: str, name: Optional[str] = None) -> Path: """Import a profile from a tar.gz archive.""" import tempfile - archive = Path(archive_path) if not archive.exists(): raise FileNotFoundError(f"Archive not found: {archive}") - top_dirs = archive_root_dirs(archive) archive_root = top_dirs.pop() if len(top_dirs) == 1 else None inferred_name = name or archive_root @@ -1678,36 +1557,27 @@ def import_profile(archive_path: str, name: Optional[str] = None) -> Path: "Cannot import as 'default' — that is the built-in root profile (~/.hermes). " "Specify a different name: hermes profile import --name " ) - profile_dir = get_profile_dir(canon) if profile_dir.exists(): raise FileExistsError(f"Profile '{canon}' already exists at {profile_dir}") - _get_profiles_root().mkdir(parents=True, exist_ok=True) - with tempfile.TemporaryDirectory(prefix="hermes_profile_import_") as tmpdir: staging_root = Path(tmpdir) safe_extract_targz(archive, staging_root) - extracted = staging_root / archive_root if not extracted.is_dir(): raise ValueError( f"Profile archive root is missing or invalid: {archive_root}" ) - final_source = extracted if archive_root != canon: final_source = staging_root / canon extracted.rename(final_source) - shutil.move(str(final_source), str(profile_dir)) - return profile_dir -# --------------------------------------------------------------------------- # Rename -# --------------------------------------------------------------------------- def _atomic_write_json(path: Path, data: dict) -> bool: """Write *data* to *path* via a sibling ``.tmp`` + rename. Returns False (tmp cleaned) on OSError.""" @@ -1717,10 +1587,8 @@ def _atomic_write_json(path: Path, data: dict) -> bool: tmp.replace(path) return True except OSError: - try: + with contextlib.suppress(OSError): tmp.unlink(missing_ok=True) - except OSError: - pass return False @@ -1729,11 +1597,9 @@ def _migrate_honcho_profile_host(old_name: str, new_name: str, new_dir: Path) -> old_host = f"hermes_{old_name}" legacy_old_host = f"hermes.{old_name}" new_host = f"hermes_{new_name}" - candidates = [ new_dir / "honcho.json", _get_default_hermes_home() / "honcho.json", Path.home() / ".honcho" / "config.json" ] - seen: set[Path] = set() for path in candidates: try: @@ -1743,23 +1609,19 @@ def _migrate_honcho_profile_host(old_name: str, new_name: str, new_dir: Path) -> if resolved in seen or not path.is_file(): continue seen.add(resolved) - try: raw = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): continue - hosts = raw.get("hosts") if not isinstance(hosts, dict): continue source_host = old_host if old_host in hosts else legacy_old_host if source_host not in hosts: continue - if new_host in hosts: print(f"⚠ Honcho host block not migrated: {new_host} already exists in {path}") continue - block = hosts[source_host] if isinstance(block, dict) and "aiPeer" not in block: block["aiPeer"] = old_name # source_host is ``hermes_`` or legacy ``hermes.`` @@ -1773,22 +1635,17 @@ def rename_profile(old_name: str, new_name: str) -> Path: profile's home IS the installation root, so "renaming" it sets a presentation-only ``display_name`` instead — the canonical id stays ``default``.""" old_canon = _canon_valid(old_name) - if old_canon == "default": if not (new_name or "").strip(): raise ValueError("Display name cannot be empty.") cleaned = set_profile_display_name("default", new_name) print(f"✓ Display name set: {cleaned} (canonical id remains 'default')") return _get_default_hermes_home() - new_canon = _canon_valid(new_name) - if new_canon == "default": raise ValueError("Cannot rename to 'default' — it is reserved.") - old_dir = get_profile_dir(old_canon) new_dir = get_profile_dir(new_canon) - if not old_dir.is_dir(): raise FileNotFoundError(f"Profile '{old_canon}' does not exist.") if new_dir.exists(): @@ -1817,13 +1674,10 @@ def rename_profile(old_name: str, new_name: str) -> Path: # 5. Update active_profile if it pointed to old name _retarget_active_profile(old_canon, new_canon, f"✓ Active profile updated: {new_canon}") - return new_dir -# --------------------------------------------------------------------------- # Profile env resolution (called from _apply_profile_override) -# --------------------------------------------------------------------------- def resolve_profile_env(profile_name: str) -> str: """Resolve a profile name to a HERMES_HOME path string. Called early in the CLI entry @@ -1840,8 +1694,6 @@ def resolve_profile_env(profile_name: str) -> str: if canon == "default": return str(root) profile_dir = root / "profiles" / canon - if not profile_dir.is_dir() or named_profile_is_deleted(profile_dir): raise _missing_profile_error(canon) - return str(profile_dir)