| """ |
| Profile management for multiple isolated Hermes instances. |
| |
| Each profile is a fully independent HERMES_HOME directory with its own |
| config.yaml, .env, memory, sessions, skills, gateway, cron, and logs. |
| Profiles live under ``~/.hermes/profiles/<name>/`` by default. |
| |
| The "default" profile is ``~/.hermes`` itself — backward compatible, |
| zero migration needed. |
| |
| Usage:: |
| |
| hermes profile create coder # fresh profile + bundled skills |
| hermes profile create coder --clone # also copy config, .env, SOUL.md |
| hermes profile create coder --clone-all # full copy of source profile |
| coder chat # use via wrapper alias |
| hermes -p coder chat # or via flag |
| hermes profile use coder # set as sticky default |
| hermes profile delete coder # remove profile + alias + service |
| """ |
|
|
| import json |
| import os |
| import re |
| import shutil |
| import stat |
| import subprocess |
| import sys |
| from dataclasses import dataclass |
| from pathlib import Path, PurePosixPath, PureWindowsPath |
| from typing import List, Optional |
|
|
| _PROFILE_ID_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") |
|
|
| |
| _PROFILE_DIRS = [ |
| "memories", |
| "sessions", |
| "skills", |
| "skins", |
| "logs", |
| "plans", |
| "workspace", |
| "cron", |
| |
| |
| |
| |
| "home", |
| ] |
|
|
| |
| _CLONE_CONFIG_FILES = [ |
| "config.yaml", |
| ".env", |
| "SOUL.md", |
| ] |
|
|
| |
| |
| |
| _CLONE_SUBDIR_FILES = [ |
| "memories/MEMORY.md", |
| "memories/USER.md", |
| ] |
|
|
| |
| _CLONE_ALL_STRIP = [ |
| "gateway.pid", |
| "gateway_state.json", |
| "processes.json", |
| ] |
|
|
| |
| |
| |
| |
| _DEFAULT_EXPORT_EXCLUDE_ROOT = frozenset({ |
| |
| "hermes-agent", |
| ".worktrees", |
| "profiles", |
| "bin", |
| "node_modules", |
| |
| "state.db", "state.db-shm", "state.db-wal", |
| "hermes_state.db", |
| "response_store.db", "response_store.db-shm", "response_store.db-wal", |
| "gateway.pid", "gateway_state.json", "processes.json", |
| "auth.json", |
| ".env", |
| "auth.lock", "active_profile", ".update_check", |
| "errors.log", |
| ".hermes_history", |
| |
| "image_cache", "audio_cache", "document_cache", |
| "browser_screenshots", "checkpoints", |
| "sandboxes", |
| "logs", |
| }) |
|
|
| |
| _RESERVED_NAMES = frozenset({ |
| "hermes", "default", "test", "tmp", "root", "sudo", |
| }) |
|
|
| |
| _HERMES_SUBCOMMANDS = frozenset({ |
| "chat", "model", "gateway", "setup", "whatsapp", "login", "logout", |
| "status", "cron", "doctor", "dump", "config", "pairing", "skills", "tools", |
| "mcp", "sessions", "insights", "version", "update", "uninstall", |
| "profile", "plugins", "honcho", "acp", |
| }) |
|
|
|
|
| |
| |
| |
|
|
| def _get_profiles_root() -> Path: |
| """Return the directory where named profiles are stored. |
| |
| Anchored to the hermes root, NOT to the current HERMES_HOME |
| (which may itself be a profile). This ensures ``coder profile list`` |
| can see all profiles. |
| |
| In Docker/custom deployments where HERMES_HOME points outside |
| ``~/.hermes``, profiles live under ``HERMES_HOME/profiles/`` so |
| they persist on the mounted volume. |
| """ |
| return _get_default_hermes_home() / "profiles" |
|
|
|
|
| def _get_default_hermes_home() -> Path: |
| """Return the default (pre-profile) HERMES_HOME path. |
| |
| In standard deployments this is ``~/.hermes``. |
| In Docker/custom deployments where HERMES_HOME is outside ``~/.hermes`` |
| (e.g. ``/opt/data``), returns HERMES_HOME directly. |
| """ |
| from hermes_constants import get_default_hermes_root |
| return get_default_hermes_root() |
|
|
|
|
| def _get_active_profile_path() -> Path: |
| """Return the path to the sticky active_profile file.""" |
| return _get_default_hermes_home() / "active_profile" |
|
|
|
|
| def _get_wrapper_dir() -> Path: |
| """Return the directory for wrapper scripts.""" |
| return Path.home() / ".local" / "bin" |
|
|
|
|
| |
| |
| |
|
|
| def validate_profile_name(name: str) -> None: |
| """Raise ``ValueError`` if *name* is not a valid profile identifier.""" |
| if name == "default": |
| return |
| if not _PROFILE_ID_RE.match(name): |
| raise ValueError( |
| f"Invalid profile name {name!r}. Must match " |
| f"[a-z0-9][a-z0-9_-]{{0,63}}" |
| ) |
|
|
|
|
| def get_profile_dir(name: str) -> Path: |
| """Resolve a profile name to its HERMES_HOME directory.""" |
| if name == "default": |
| return _get_default_hermes_home() |
| return _get_profiles_root() / name |
|
|
|
|
| def profile_exists(name: str) -> bool: |
| """Check whether a profile directory exists.""" |
| if name == "default": |
| return True |
| return get_profile_dir(name).is_dir() |
|
|
|
|
| |
| |
| |
|
|
| def check_alias_collision(name: str) -> Optional[str]: |
| """Return a human-readable collision message, or None if the name is safe. |
| |
| Checks: reserved names, hermes subcommands, existing binaries in PATH. |
| """ |
| if name in _RESERVED_NAMES: |
| return f"'{name}' is a reserved name" |
| if name in _HERMES_SUBCOMMANDS: |
| return f"'{name}' conflicts with a hermes subcommand" |
|
|
| |
| wrapper_dir = _get_wrapper_dir() |
| try: |
| result = subprocess.run( |
| ["which", name], capture_output=True, text=True, timeout=5, |
| ) |
| if result.returncode == 0: |
| existing_path = result.stdout.strip() |
| |
| if existing_path == str(wrapper_dir / name): |
| try: |
| content = (wrapper_dir / name).read_text() |
| if "hermes -p" in content: |
| return None |
| except Exception: |
| pass |
| return f"'{name}' conflicts with an existing command ({existing_path})" |
| except (FileNotFoundError, subprocess.TimeoutExpired): |
| pass |
|
|
| return None |
|
|
|
|
| def _is_wrapper_dir_in_path() -> bool: |
| """Check if ~/.local/bin is in PATH.""" |
| wrapper_dir = str(_get_wrapper_dir()) |
| return wrapper_dir in os.environ.get("PATH", "").split(os.pathsep) |
|
|
|
|
| def create_wrapper_script(name: str) -> Optional[Path]: |
| """Create a shell wrapper script at ~/.local/bin/<name>. |
| |
| Returns the path to the created wrapper, or None if creation failed. |
| """ |
| wrapper_dir = _get_wrapper_dir() |
| try: |
| wrapper_dir.mkdir(parents=True, exist_ok=True) |
| except OSError as e: |
| print(f"⚠ Could not create {wrapper_dir}: {e}") |
| return None |
|
|
| wrapper_path = wrapper_dir / name |
| try: |
| wrapper_path.write_text(f'#!/bin/sh\nexec hermes -p {name} "$@"\n') |
| wrapper_path.chmod(wrapper_path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH) |
| return wrapper_path |
| except OSError as e: |
| print(f"⚠ Could not create wrapper at {wrapper_path}: {e}") |
| return None |
|
|
|
|
| def remove_wrapper_script(name: str) -> bool: |
| """Remove the wrapper script for a profile. Returns True if removed.""" |
| wrapper_path = _get_wrapper_dir() / name |
| if wrapper_path.exists(): |
| try: |
| |
| content = wrapper_path.read_text() |
| if "hermes -p" in content: |
| wrapper_path.unlink() |
| return True |
| except Exception: |
| pass |
| return False |
|
|
|
|
| |
| |
| |
|
|
| @dataclass |
| class ProfileInfo: |
| """Summary information about a profile.""" |
| name: str |
| path: Path |
| is_default: bool |
| gateway_running: bool |
| model: Optional[str] = None |
| provider: Optional[str] = None |
| has_env: bool = False |
| skill_count: int = 0 |
| alias_path: Optional[Path] = None |
|
|
|
|
| def _read_config_model(profile_dir: Path) -> tuple: |
| """Read model/provider from a profile's config.yaml. Returns (model, provider).""" |
| config_path = profile_dir / "config.yaml" |
| if not config_path.exists(): |
| return None, None |
| try: |
| import yaml |
| with open(config_path, "r") as f: |
| cfg = yaml.safe_load(f) or {} |
| model_cfg = cfg.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 |
|
|
|
|
| def _check_gateway_running(profile_dir: Path) -> bool: |
| """Check if a gateway is running for a given profile directory.""" |
| try: |
| from gateway.status import get_running_pid |
| return get_running_pid(profile_dir / "gateway.pid", cleanup_stale=False) is not None |
| except Exception: |
| return False |
|
|
|
|
| def _count_skills(profile_dir: Path) -> int: |
| """Count installed skills in a profile.""" |
| skills_dir = profile_dir / "skills" |
| if not skills_dir.is_dir(): |
| return 0 |
| count = 0 |
| for md in skills_dir.rglob("SKILL.md"): |
| if "/.hub/" not in str(md) and "/.git/" not in str(md): |
| count += 1 |
| return count |
|
|
|
|
| |
| |
| |
|
|
| def list_profiles() -> List[ProfileInfo]: |
| """Return info for all profiles, including the default.""" |
| profiles = [] |
| wrapper_dir = _get_wrapper_dir() |
|
|
| |
| default_home = _get_default_hermes_home() |
| if default_home.is_dir(): |
| model, provider = _read_config_model(default_home) |
| profiles.append(ProfileInfo( |
| name="default", |
| path=default_home, |
| is_default=True, |
| gateway_running=_check_gateway_running(default_home), |
| model=model, |
| provider=provider, |
| has_env=(default_home / ".env").exists(), |
| skill_count=_count_skills(default_home), |
| )) |
|
|
| |
| profiles_root = _get_profiles_root() |
| if profiles_root.is_dir(): |
| for entry in sorted(profiles_root.iterdir()): |
| if not entry.is_dir(): |
| continue |
| name = entry.name |
| if not _PROFILE_ID_RE.match(name): |
| continue |
| model, provider = _read_config_model(entry) |
| alias_path = wrapper_dir / name |
| profiles.append(ProfileInfo( |
| name=name, |
| path=entry, |
| is_default=False, |
| gateway_running=_check_gateway_running(entry), |
| model=model, |
| provider=provider, |
| has_env=(entry / ".env").exists(), |
| skill_count=_count_skills(entry), |
| alias_path=alias_path if alias_path.exists() else None, |
| )) |
|
|
| return profiles |
|
|
|
|
| def create_profile( |
| name: str, |
| clone_from: Optional[str] = None, |
| clone_all: bool = False, |
| clone_config: bool = False, |
| no_alias: bool = False, |
| ) -> Path: |
| """Create a new profile directory. |
| |
| Parameters |
| ---------- |
| name: |
| Profile identifier (lowercase, alphanumeric, hyphens, underscores). |
| clone_from: |
| Source profile to clone from. If ``None`` and clone_config/clone_all |
| is True, defaults to the currently active profile. |
| clone_all: |
| If True, do a full copytree of the source (all state). |
| clone_config: |
| If True, copy only config files (config.yaml, .env, SOUL.md). |
| no_alias: |
| If True, skip wrapper script creation. |
| |
| Returns |
| ------- |
| Path |
| The newly created profile directory. |
| """ |
| validate_profile_name(name) |
|
|
| if name == "default": |
| raise ValueError( |
| "Cannot create a profile named 'default' — it is the built-in profile (~/.hermes)." |
| ) |
|
|
| profile_dir = get_profile_dir(name) |
| if profile_dir.exists(): |
| raise FileExistsError(f"Profile '{name}' already exists at {profile_dir}") |
|
|
| |
| source_dir = None |
| if clone_from is not None or clone_all or clone_config: |
| if clone_from is None: |
| |
| from hermes_constants import get_hermes_home |
| source_dir = get_hermes_home() |
| else: |
| validate_profile_name(clone_from) |
| source_dir = get_profile_dir(clone_from) |
| if not source_dir.is_dir(): |
| raise FileNotFoundError( |
| f"Source profile '{clone_from or 'active'}' does not exist at {source_dir}" |
| ) |
|
|
| if clone_all and source_dir: |
| |
| shutil.copytree(source_dir, profile_dir) |
| |
| for stale in _CLONE_ALL_STRIP: |
| (profile_dir / stale).unlink(missing_ok=True) |
| else: |
| |
| 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 not None: |
| for filename in _CLONE_CONFIG_FILES: |
| src = source_dir / filename |
| if src.exists(): |
| shutil.copy2(src, profile_dir / filename) |
|
|
| |
| for relpath in _CLONE_SUBDIR_FILES: |
| src = source_dir / relpath |
| if src.exists(): |
| dst = profile_dir / relpath |
| dst.parent.mkdir(parents=True, exist_ok=True) |
| shutil.copy2(src, dst) |
|
|
| |
| |
| soul_path = profile_dir / "SOUL.md" |
| if not soul_path.exists(): |
| try: |
| from hermes_cli.default_soul import DEFAULT_SOUL_MD |
| soul_path.write_text(DEFAULT_SOUL_MD, encoding="utf-8") |
| except Exception: |
| pass |
|
|
| return profile_dir |
|
|
|
|
| def seed_profile_skills(profile_dir: Path, quiet: bool = False) -> Optional[dict]: |
| """Seed bundled skills into a profile via subprocess. |
| |
| Uses subprocess because sync_skills() caches HERMES_HOME at module level. |
| Returns the sync result dict, or None on failure. |
| """ |
| project_root = Path(__file__).parent.parent.resolve() |
| try: |
| result = subprocess.run( |
| [sys.executable, "-c", |
| "import json; from tools.skills_sync import sync_skills; " |
| "r = sync_skills(quiet=True); print(json.dumps(r))"], |
| env={**os.environ, "HERMES_HOME": str(profile_dir)}, |
| cwd=str(project_root), |
| capture_output=True, text=True, timeout=60, |
| ) |
| if result.returncode == 0 and result.stdout.strip(): |
| return json.loads(result.stdout.strip()) |
| if not quiet: |
| print(f"⚠ Skill seeding returned exit code {result.returncode}") |
| if result.stderr.strip(): |
| print(f" {result.stderr.strip()[:200]}") |
| return None |
| except subprocess.TimeoutExpired: |
| if not quiet: |
| print("⚠ Skill seeding timed out (60s)") |
| return None |
| except Exception as e: |
| if not quiet: |
| print(f"⚠ Skill seeding failed: {e}") |
| return None |
|
|
|
|
| def delete_profile(name: str, yes: bool = False) -> Path: |
| """Delete a profile, its wrapper script, and its gateway service. |
| |
| Stops the gateway if running. Disables systemd/launchd service first |
| to prevent auto-restart. |
| |
| Returns the path that was removed. |
| """ |
| validate_profile_name(name) |
|
|
| if name == "default": |
| raise ValueError( |
| "Cannot delete the default profile (~/.hermes).\n" |
| "To remove everything, use: hermes uninstall" |
| ) |
|
|
| profile_dir = get_profile_dir(name) |
| if not profile_dir.is_dir(): |
| raise FileNotFoundError(f"Profile '{name}' does not exist.") |
|
|
| |
| model, provider = _read_config_model(profile_dir) |
| gw_running = _check_gateway_running(profile_dir) |
| skill_count = _count_skills(profile_dir) |
|
|
| print(f"\nProfile: {name}") |
| print(f"Path: {profile_dir}") |
| if model: |
| print(f"Model: {model}" + (f" ({provider})" if provider else "")) |
| if skill_count: |
| print(f"Skills: {skill_count}") |
|
|
| items = [ |
| "All config, API keys, memories, sessions, skills, cron jobs", |
| ] |
|
|
| |
| wrapper_path = _get_wrapper_dir() / name |
| has_wrapper = wrapper_path.exists() |
| if has_wrapper: |
| items.append(f"Command alias ({wrapper_path})") |
|
|
| print(f"\nThis will permanently delete:") |
| for item in items: |
| print(f" • {item}") |
| if gw_running: |
| print(f" ⚠ Gateway is running — it will be stopped.") |
|
|
| |
| if not yes: |
| print() |
| try: |
| confirm = input(f"Type '{name}' to confirm: ").strip() |
| except (KeyboardInterrupt, EOFError): |
| print("\nCancelled.") |
| return profile_dir |
| if confirm != name: |
| print("Cancelled.") |
| return profile_dir |
|
|
| |
| _cleanup_gateway_service(name, profile_dir) |
|
|
| |
| if gw_running: |
| _stop_gateway_process(profile_dir) |
|
|
| |
| if has_wrapper: |
| if remove_wrapper_script(name): |
| print(f"✓ Removed {wrapper_path}") |
|
|
| |
| try: |
| shutil.rmtree(profile_dir) |
| print(f"✓ Removed {profile_dir}") |
| except Exception as e: |
| print(f"⚠ Could not remove {profile_dir}: {e}") |
|
|
| |
| try: |
| active = get_active_profile() |
| if active == name: |
| set_active_profile("default") |
| print("✓ Active profile reset to default") |
| except Exception: |
| pass |
|
|
| print(f"\nProfile '{name}' deleted.") |
| return profile_dir |
|
|
|
|
| def _cleanup_gateway_service(name: str, profile_dir: Path) -> None: |
| """Disable and remove systemd/launchd service for a profile.""" |
| import platform as _platform |
|
|
| |
| |
| old_home = os.environ.get("HERMES_HOME") |
| try: |
| os.environ["HERMES_HOME"] = str(profile_dir) |
| from hermes_cli.gateway import get_service_name, get_launchd_plist_path |
|
|
| if _platform.system() == "Linux": |
| svc_name = get_service_name() |
| svc_file = Path.home() / ".config" / "systemd" / "user" / f"{svc_name}.service" |
| if svc_file.exists(): |
| subprocess.run( |
| ["systemctl", "--user", "disable", svc_name], |
| capture_output=True, check=False, timeout=10, |
| ) |
| subprocess.run( |
| ["systemctl", "--user", "stop", svc_name], |
| capture_output=True, check=False, timeout=10, |
| ) |
| svc_file.unlink(missing_ok=True) |
| subprocess.run( |
| ["systemctl", "--user", "daemon-reload"], |
| capture_output=True, check=False, timeout=10, |
| ) |
| print(f"✓ Service {svc_name} removed") |
|
|
| elif _platform.system() == "Darwin": |
| plist_path = get_launchd_plist_path() |
| if plist_path.exists(): |
| subprocess.run( |
| ["launchctl", "unload", str(plist_path)], |
| capture_output=True, check=False, timeout=10, |
| ) |
| plist_path.unlink(missing_ok=True) |
| print(f"✓ Launchd service removed") |
| except Exception as e: |
| print(f"⚠ Service cleanup: {e}") |
| finally: |
| if old_home is not None: |
| os.environ["HERMES_HOME"] = old_home |
| elif "HERMES_HOME" in os.environ: |
| del os.environ["HERMES_HOME"] |
|
|
|
|
| def _stop_gateway_process(profile_dir: Path) -> None: |
| """Stop a running gateway process via its PID file.""" |
| import signal as _signal |
| import time as _time |
|
|
| pid_file = profile_dir / "gateway.pid" |
| if not pid_file.exists(): |
| return |
|
|
| try: |
| raw = pid_file.read_text().strip() |
| data = json.loads(raw) if raw.startswith("{") else {"pid": int(raw)} |
| pid = int(data["pid"]) |
| os.kill(pid, _signal.SIGTERM) |
| |
| for _ in range(20): |
| _time.sleep(0.5) |
| try: |
| os.kill(pid, 0) |
| except ProcessLookupError: |
| print(f"✓ Gateway stopped (PID {pid})") |
| return |
| |
| try: |
| os.kill(pid, _signal.SIGKILL) |
| except ProcessLookupError: |
| pass |
| print(f"✓ Gateway force-stopped (PID {pid})") |
| except (ProcessLookupError, PermissionError): |
| print("✓ Gateway already stopped") |
| except Exception as e: |
| print(f"⚠ Could not stop gateway: {e}") |
|
|
|
|
| |
| |
| |
|
|
| def get_active_profile() -> str: |
| """Read the sticky active profile name. |
| |
| Returns ``"default"`` if no active_profile file exists or it's empty. |
| """ |
| path = _get_active_profile_path() |
| try: |
| name = path.read_text().strip() |
| if not name: |
| return "default" |
| return name |
| except (FileNotFoundError, UnicodeDecodeError, OSError): |
| return "default" |
|
|
|
|
| def set_active_profile(name: str) -> None: |
| """Set the sticky active profile. |
| |
| Writes to ``~/.hermes/active_profile``. Use ``"default"`` to clear. |
| """ |
| validate_profile_name(name) |
| if name != "default" and not profile_exists(name): |
| raise FileNotFoundError( |
| f"Profile '{name}' does not exist. " |
| f"Create it with: hermes profile create {name}" |
| ) |
|
|
| path = _get_active_profile_path() |
| path.parent.mkdir(parents=True, exist_ok=True) |
| if name == "default": |
| |
| path.unlink(missing_ok=True) |
| else: |
| |
| tmp = path.with_suffix(".tmp") |
| tmp.write_text(name + "\n") |
| tmp.replace(path) |
|
|
|
|
| def get_active_profile_name() -> str: |
| """Infer the current profile name from HERMES_HOME. |
| |
| Returns ``"default"`` if HERMES_HOME is not set or points to ``~/.hermes``. |
| Returns the profile name if HERMES_HOME points into ``~/.hermes/profiles/<name>``. |
| Returns ``"custom"`` if HERMES_HOME is set to an unrecognized path. |
| """ |
| from hermes_constants import get_hermes_home |
| hermes_home = get_hermes_home() |
| resolved = hermes_home.resolve() |
|
|
| default_resolved = _get_default_hermes_home().resolve() |
| if resolved == default_resolved: |
| return "default" |
|
|
| profiles_root = _get_profiles_root().resolve() |
| try: |
| rel = resolved.relative_to(profiles_root) |
| parts = rel.parts |
| if len(parts) == 1 and _PROFILE_ID_RE.match(parts[0]): |
| return parts[0] |
| except ValueError: |
| pass |
|
|
| return "custom" |
|
|
|
|
| |
| |
| |
|
|
| def _default_export_ignore(root_dir: Path): |
| """Return an *ignore* callable for :func:`shutil.copytree`. |
| |
| At the root level it excludes everything in ``_DEFAULT_EXPORT_EXCLUDE_ROOT``. |
| At all levels it excludes ``__pycache__``, sockets, and temp files. |
| """ |
|
|
| def _ignore(directory: str, contents: list) -> set: |
| ignored: set = set() |
| for entry in contents: |
| |
| if entry == "__pycache__" or entry.endswith((".sock", ".tmp")): |
| ignored.add(entry) |
| |
| elif entry in ("package.json", "package-lock.json"): |
| ignored.add(entry) |
| |
| if Path(directory) == root_dir: |
| ignored.update(c for c in contents if c in _DEFAULT_EXPORT_EXCLUDE_ROOT) |
| return ignored |
|
|
| return _ignore |
|
|
|
|
| def export_profile(name: str, output_path: str) -> Path: |
| """Export a profile to a tar.gz archive. |
| |
| Returns the output file path. |
| """ |
| import tempfile |
|
|
| validate_profile_name(name) |
| profile_dir = get_profile_dir(name) |
| if not profile_dir.is_dir(): |
| raise FileNotFoundError(f"Profile '{name}' does not exist.") |
|
|
| output = Path(output_path) |
| |
| base = str(output).removesuffix(".tar.gz").removesuffix(".tgz") |
|
|
| if name == "default": |
| |
| |
| |
| with tempfile.TemporaryDirectory() as tmpdir: |
| staged = Path(tmpdir) / "default" |
| shutil.copytree( |
| profile_dir, |
| staged, |
| ignore=_default_export_ignore(profile_dir), |
| ) |
| result = shutil.make_archive(base, "gztar", tmpdir, "default") |
| return Path(result) |
|
|
| |
| with tempfile.TemporaryDirectory() as tmpdir: |
| staged = Path(tmpdir) / name |
| _CREDENTIAL_FILES = {"auth.json", ".env"} |
| shutil.copytree( |
| profile_dir, |
| staged, |
| ignore=lambda d, contents: _CREDENTIAL_FILES & set(contents), |
| ) |
| result = shutil.make_archive(base, "gztar", tmpdir, name) |
| return Path(result) |
|
|
|
|
| def _normalize_profile_archive_parts(member_name: str) -> List[str]: |
| """Return safe path parts for a profile archive member.""" |
| normalized_name = member_name.replace("\\", "/") |
| posix_path = PurePosixPath(normalized_name) |
| windows_path = PureWindowsPath(member_name) |
|
|
| if ( |
| not normalized_name |
| or posix_path.is_absolute() |
| or windows_path.is_absolute() |
| or windows_path.drive |
| ): |
| raise ValueError(f"Unsafe archive member path: {member_name}") |
|
|
| parts = [part for part in posix_path.parts if part not in ("", ".")] |
| if not parts or any(part == ".." for part in parts): |
| raise ValueError(f"Unsafe archive member path: {member_name}") |
| return parts |
|
|
|
|
| def _safe_extract_profile_archive(archive: Path, destination: Path) -> None: |
| """Extract a profile archive without allowing path escapes or links.""" |
| import tarfile |
|
|
| with tarfile.open(archive, "r:gz") as tf: |
| for member in tf.getmembers(): |
| parts = _normalize_profile_archive_parts(member.name) |
| target = destination.joinpath(*parts) |
|
|
| if member.isdir(): |
| target.mkdir(parents=True, exist_ok=True) |
| continue |
|
|
| if not member.isfile(): |
| raise ValueError( |
| f"Unsupported archive member type: {member.name}" |
| ) |
|
|
| target.parent.mkdir(parents=True, exist_ok=True) |
| extracted = tf.extractfile(member) |
| if extracted is None: |
| raise ValueError(f"Cannot read archive member: {member.name}") |
|
|
| with extracted, open(target, "wb") as dst: |
| shutil.copyfileobj(extracted, dst) |
|
|
| try: |
| os.chmod(target, member.mode & 0o777) |
| except OSError: |
| pass |
|
|
|
|
| def _inspect_profile_archive_roots(archive: Path) -> set[str]: |
| """Return the archive's top-level directory names. |
| |
| Profile imports expect exactly one root directory. Inspecting the archive |
| before extraction lets us stage the import safely instead of mutating a |
| live profile tree first and reconciling names later. |
| """ |
| import tarfile |
|
|
| with tarfile.open(archive, "r:gz") as tf: |
| top_dirs = { |
| parts[0] |
| for member in tf.getmembers() |
| for parts in [_normalize_profile_archive_parts(member.name)] |
| if len(parts) > 1 or member.isdir() |
| } |
| if not top_dirs: |
| top_dirs = { |
| _normalize_profile_archive_parts(member.name)[0] |
| for member in tf.getmembers() |
| if member.isdir() |
| } |
| return top_dirs |
|
|
|
|
| def import_profile(archive_path: str, name: Optional[str] = None) -> Path: |
| """Import a profile from a tar.gz archive. |
| |
| If *name* is not given, infers it from the archive's top-level directory. |
| Returns the imported profile directory. |
| """ |
| import tempfile |
|
|
| archive = Path(archive_path) |
| if not archive.exists(): |
| raise FileNotFoundError(f"Archive not found: {archive}") |
|
|
| top_dirs = _inspect_profile_archive_roots(archive) |
| archive_root = top_dirs.pop() if len(top_dirs) == 1 else None |
| inferred_name = name or archive_root |
| if not inferred_name: |
| raise ValueError( |
| "Cannot determine profile name from archive. " |
| "Specify it explicitly: hermes profile import <archive> --name <name>" |
| ) |
| if archive_root is None: |
| raise ValueError( |
| "Profile archive must contain exactly one top-level directory." |
| ) |
|
|
| |
| |
| |
| if inferred_name == "default": |
| raise ValueError( |
| "Cannot import as 'default' — that is the built-in root profile (~/.hermes). " |
| "Specify a different name: hermes profile import <archive> --name <name>" |
| ) |
|
|
| validate_profile_name(inferred_name) |
| profile_dir = get_profile_dir(inferred_name) |
| if profile_dir.exists(): |
| raise FileExistsError(f"Profile '{inferred_name}' already exists at {profile_dir}") |
|
|
| profiles_root = _get_profiles_root() |
| profiles_root.mkdir(parents=True, exist_ok=True) |
|
|
| with tempfile.TemporaryDirectory(prefix="hermes_profile_import_") as tmpdir: |
| staging_root = Path(tmpdir) |
| _safe_extract_profile_archive(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 != inferred_name: |
| final_source = staging_root / inferred_name |
| extracted.rename(final_source) |
|
|
| shutil.move(str(final_source), str(profile_dir)) |
|
|
| return profile_dir |
|
|
|
|
| |
| |
| |
|
|
| def rename_profile(old_name: str, new_name: str) -> Path: |
| """Rename a profile: directory, wrapper script, service, active_profile. |
| |
| Returns the new profile directory. |
| """ |
| validate_profile_name(old_name) |
| validate_profile_name(new_name) |
|
|
| if old_name == "default": |
| raise ValueError("Cannot rename the default profile.") |
| if new_name == "default": |
| raise ValueError("Cannot rename to 'default' — it is reserved.") |
|
|
| old_dir = get_profile_dir(old_name) |
| new_dir = get_profile_dir(new_name) |
|
|
| if not old_dir.is_dir(): |
| raise FileNotFoundError(f"Profile '{old_name}' does not exist.") |
| if new_dir.exists(): |
| raise FileExistsError(f"Profile '{new_name}' already exists.") |
|
|
| |
| if _check_gateway_running(old_dir): |
| _cleanup_gateway_service(old_name, old_dir) |
| _stop_gateway_process(old_dir) |
|
|
| |
| old_dir.rename(new_dir) |
| print(f"✓ Renamed {old_dir.name} → {new_dir.name}") |
|
|
| |
| remove_wrapper_script(old_name) |
| collision = check_alias_collision(new_name) |
| if not collision: |
| create_wrapper_script(new_name) |
| print(f"✓ Alias updated: {new_name}") |
| else: |
| print(f"⚠ Cannot create alias '{new_name}' — {collision}") |
|
|
| |
| try: |
| if get_active_profile() == old_name: |
| set_active_profile(new_name) |
| print(f"✓ Active profile updated: {new_name}") |
| except Exception: |
| pass |
|
|
| return new_dir |
|
|
|
|
| |
| |
| |
|
|
| def generate_bash_completion() -> str: |
| """Generate a bash completion script for hermes profile names.""" |
| return '''# Hermes Agent profile completion |
| # Add to ~/.bashrc: eval "$(hermes completion bash)" |
| |
| _hermes_profiles() { |
| local profiles_dir="$HOME/.hermes/profiles" |
| local profiles="default" |
| if [ -d "$profiles_dir" ]; then |
| profiles="$profiles $(ls "$profiles_dir" 2>/dev/null)" |
| fi |
| echo "$profiles" |
| } |
| |
| _hermes_completion() { |
| local cur prev |
| cur="${COMP_WORDS[COMP_CWORD]}" |
| prev="${COMP_WORDS[COMP_CWORD-1]}" |
| |
| # Complete profile names after -p / --profile |
| if [[ "$prev" == "-p" || "$prev" == "--profile" ]]; then |
| COMPREPLY=($(compgen -W "$(_hermes_profiles)" -- "$cur")) |
| return |
| fi |
| |
| # Complete profile subcommands |
| if [[ "${COMP_WORDS[1]}" == "profile" ]]; then |
| case "$prev" in |
| profile) |
| COMPREPLY=($(compgen -W "list use create delete show alias rename export import" -- "$cur")) |
| return |
| ;; |
| use|delete|show|alias|rename|export) |
| COMPREPLY=($(compgen -W "$(_hermes_profiles)" -- "$cur")) |
| return |
| ;; |
| esac |
| fi |
| |
| # Top-level subcommands |
| if [[ "$COMP_CWORD" == 1 ]]; then |
| local commands="chat model gateway setup status cron doctor dump config skills tools mcp sessions profile update version" |
| COMPREPLY=($(compgen -W "$commands" -- "$cur")) |
| fi |
| } |
| |
| complete -F _hermes_completion hermes |
| ''' |
|
|
|
|
| def generate_zsh_completion() -> str: |
| """Generate a zsh completion script for hermes profile names.""" |
| return '''#compdef hermes |
| # Hermes Agent profile completion |
| # Add to ~/.zshrc: eval "$(hermes completion zsh)" |
| |
| _hermes() { |
| local -a profiles |
| profiles=(default) |
| if [[ -d "$HOME/.hermes/profiles" ]]; then |
| profiles+=("${(@f)$(ls $HOME/.hermes/profiles 2>/dev/null)}") |
| fi |
| |
| _arguments \\ |
| '-p[Profile name]:profile:($profiles)' \\ |
| '--profile[Profile name]:profile:($profiles)' \\ |
| '1:command:(chat model gateway setup status cron doctor dump config skills tools mcp sessions profile update version)' \\ |
| '*::arg:->args' |
| |
| case $words[1] in |
| profile) |
| _arguments '1:action:(list use create delete show alias rename export import)' \\ |
| '2:profile:($profiles)' |
| ;; |
| esac |
| } |
| |
| _hermes "$@" |
| ''' |
|
|
|
|
| |
| |
| |
|
|
| def resolve_profile_env(profile_name: str) -> str: |
| """Resolve a profile name to a HERMES_HOME path string. |
| |
| Called early in the CLI entry point, before any hermes modules |
| are imported, to set the HERMES_HOME environment variable. |
| """ |
| validate_profile_name(profile_name) |
| profile_dir = get_profile_dir(profile_name) |
|
|
| if profile_name != "default" and not profile_dir.is_dir(): |
| raise FileNotFoundError( |
| f"Profile '{profile_name}' does not exist. " |
| f"Create it with: hermes profile create {profile_name}" |
| ) |
|
|
| return str(profile_dir) |
|
|