"""Move a home onto the catalog plugin of a feature that left core. A feature (gateway platform, toolset, bundled plugin) that moves from core into a standalone catalog plugin keeps its names, config keys and env vars, so migrating is only "install the plugin". Each row of :data:`LEFT_CORE` names the catalog plugin and a read-only predicate "does this home use it". The contract is the one memory providers established (``memory_provider_migration``, whose helpers this reuses; memory is not a row because its plugin name comes from ``memory.provider``, not a table): * ``hermes update`` installs the plugin for every profile home sharing the venv that uses the feature. * Agent start and gateway start retry once per process for the active home (Desktop users update through the app and never run ``hermes update``), honouring ``security.allow_lazy_installs``. * After a failed attempt (catalog miss or unreachable, failed install) starts skip the row for an hour, doubling with each further failure up to a week, so an offline machine pays the network round trip once, not on every ``hermes chat``, and a home whose install cannot succeed stops failing every hour; ``hermes update`` always retries. * A home gets the plugin automatically at most once (``_left_core_installed`` in its config.yaml): ``hermes plugins remove`` afterwards is the user's choice and sticks. Installs go through the normal catalog install path at the reviewed pin (kill list, dependency constraints, enable). Unattended dependency consent covers only these rows: the feature shipped in core, so its users already accepted its dependencies. Agent/gateway start always installs with it (nothing can answer a prompt there); ``hermes update`` in a terminal still asks. Every outcome reaches the user (terminal, Desktop, chat); a gateway-start outcome waits for the home's first agent to deliver it. """ from __future__ import annotations import logging import os import sys from dataclasses import dataclass from pathlib import Path from typing import Callable, Optional from hermes_cli.memory_provider_migration import ( _failed_recently, _home_consent, _home_label, _install_command, _interactive, _note_failure, _unattended_consent, ) logger = logging.getLogger(__name__) # ``terminal.*`` keys a left-core plugin still reads -> the ``.env`` copy core's ``hermes config set`` also # wrote. Not in DEFAULT_CONFIG (a seeded default would hide the plugin's env fallback), so ``config # set/get`` recognise them here, and ``config unset`` removes that copy too: the plugin falls back to it, # and the old value would return instead of the default. LEFT_CORE_TERMINAL_KEYS = {"terminal.singularity_image": "TERMINAL_SINGULARITY_IMAGE"} def _read_config(home: Path) -> dict: import utils try: data = utils.fast_safe_load((home / "config.yaml").read_text(encoding="utf-8-sig")) except Exception: # missing/unreadable/invalid YAML: nothing we can see is in use return {} return data if isinstance(data, dict) else {} def platform_configured_on(home: Path, name: str) -> bool: """Platform *name* is on in *home*'s config files as the gateway loader reads them: legacy ``gateway.json`` base layer, then config.yaml's ``gateway.platforms`` / ``platforms`` / ``gateway.`` blocks merged in the loader's order (managed overlay and ``${VAR}`` applied), so the last explicit ``enabled`` wins. A stored token with no ``enabled`` anywhere counts too. Env credentials are the caller's to check. Read-only; works before the platform's plugin loads.""" from gateway import config_loader from gateway.config import PlatformConfig gw_data = config_loader.load_legacy_gateway_json(home) gw_data = gw_data if isinstance(gw_data, dict) else {} try: yaml_cfg = config_loader.read_yaml_layers(home) except Exception: # malformed config.yaml: the gateway falls back to gateway.json alone too yaml_cfg = {} yaml_cfg = yaml_cfg if isinstance(yaml_cfg, dict) else {} block = config_loader.merge_platform_sections( yaml_cfg, yaml_cfg.get("gateway"), gw_data, also=frozenset({name})).get(name) if not isinstance(block, dict): return False platform = PlatformConfig.from_dict(block) return platform.enabled or ("enabled" not in block and bool(str(platform.token or "").strip())) def _toolset_listed(config: dict, names: frozenset[str], *, own_platform: str = "") -> bool: """A toolset in *names* is selected in ``platform_toolsets`` or the top-level ``toolsets`` list. *own_platform*'s row is skipped: it scopes that platform's own sessions, which only exist once the platform is on (:func:`platform_configured_on`). The installers copy cli-config.yaml.example into config.yaml, and it carried ``homeassistant: [hermes-homeassistant]`` until Home Assistant left core, so that row alone is not use.""" rows = config.get("platform_toolsets") selections = [sel for platform, sel in rows.items() if platform != own_platform] if isinstance(rows, dict) else [] selections.append(config.get("toolsets")) return any(isinstance(sel, list) and any(str(item) in names for item in sel) for sel in selections) def homeassistant_in_use(home: Path, *, process_env: bool = False) -> bool: """What made core run Home Assistant for *home*: ``HASS_TOKEN`` in its ``.env`` (it enabled both the gateway platform and the tools), the platform on in its gateway config (:func:`platform_configured_on`), or the ``homeassistant`` / ``hermes-homeassistant`` toolset selected for another platform or in the top-level ``toolsets`` list. *process_env* (the active home at startup only) also counts a ``HASS_TOKEN`` the process received from its environment (systemd unit, Docker, shell export).""" from agent.secret_scope import load_env_file if (load_env_file(home / ".env").get("HASS_TOKEN") or "").strip(): return True if process_env: try: from agent.secret_scope import get_secret if (get_secret("HASS_TOKEN", "") or "").strip(): return True except Exception: pass if platform_configured_on(home, "homeassistant"): return True return _toolset_listed( _read_config(home), frozenset({"homeassistant", "hermes-homeassistant"}), own_platform="homeassistant") def spotify_in_use(home: Path, *, process_env: bool = False) -> bool: """What made core's Spotify tools usable for *home*: a login stored in its ``auth.json`` (``providers.spotify``, written by ``hermes auth spotify``) or the ``spotify`` toolset selected for a platform. The tools were opt-in and login-gated, so a client id alone is not use.""" import json try: store = json.loads((home / "auth.json").read_text(encoding="utf-8-sig")) except (OSError, ValueError): # missing/unreadable/invalid JSON: no login we can see store = {} providers = store.get("providers") if isinstance(store, dict) else None state = providers.get("spotify") if isinstance(providers, dict) else None if isinstance(state, dict) and (state.get("access_token") or state.get("refresh_token")): return True return _toolset_listed(_read_config(home), frozenset({"spotify"})) def _bundled_plugin_selected(home: Path, plugin: str) -> bool: """*plugin*'s bundled copy was on for *home*: its key or name in ``plugins.enabled`` and neither in ``plugins.disabled`` (which wins, as in the loader), read with the managed overlay applied like the loader's config, so an administrator's pinned list counts.""" from gateway import config_loader feature = next(f for f in LEFT_CORE if f.plugin == plugin) try: plugins = config_loader.read_yaml_layers(home).get("plugins") except Exception: # malformed config.yaml: the loader enables nothing from it either logger.debug("left-core: unreadable config for %s", home, exc_info=True) return False plugins = plugins if isinstance(plugins, dict) else {} def listed(key: str) -> bool: names = plugins.get(key) return isinstance(names, list) and any(str(n) in (feature.plugin_key, feature.plugin) for n in names) return listed("enabled") and not listed("disabled") def langfuse_in_use(home: Path, *, process_env: bool = False) -> bool: """What made core trace to Langfuse for *home*: the bundled ``observability/langfuse`` selected (:func:`_bundled_plugin_selected`). The plugin was opt-in, so keys in ``.env`` alone are not use.""" return _bundled_plugin_selected(home, "langfuse") def singularity_in_use(home: Path, *, process_env: bool = False) -> bool: """What made core run its Singularity backend for *home*: ``terminal.backend`` (or the legacy ``terminal.env_type``) set to ``singularity`` in its config (managed overlay applied, as the bridge reads it), else ``TERMINAL_ENV=singularity`` from its dotenv layers as ``load_hermes_dotenv`` resolves them (the home's ``.env``, then the administrator-managed ``.env`` with override, ``${VAR}`` references expanded: :func:`~hermes_cli.env_loader.resolve_home_dotenv`). config.yaml wins over both, as in the bridge. *process_env* (startup) also counts a ``TERMINAL_ENV`` the process received from its environment (systemd unit, Docker, shell export), for the launch home only: ``TERMINAL_*`` is process-global (``get_secret`` never scopes it), so for any other served home it is the launch profile's backend, not this one's.""" from gateway import config_loader try: terminal = config_loader.read_yaml_layers(home).get("terminal") except Exception: # malformed config.yaml: the bridge reads nothing from it either logger.debug("left-core: unreadable config for %s", home, exc_info=True) terminal = None terminal = terminal if isinstance(terminal, dict) else {} configured = str(terminal.get("backend") or terminal.get("env_type") or "").strip() if not configured: from hermes_cli.env_loader import resolve_home_dotenv configured = resolve_home_dotenv(home).get("TERMINAL_ENV", "").strip() if not configured and process_env: from hermes_constants import get_routing_process_hermes_home, hermes_home_key if hermes_home_key(home) == hermes_home_key(get_routing_process_hermes_home()): configured = os.environ.get("TERMINAL_ENV", "").strip() return configured.lower() == "singularity" def agent_imports_in_use(home: Path, *, process_env: bool = False) -> bool: """*home* registered a ``hermes import-agent`` source for ``--sync`` (its ``import-sync.json``, which scheduled syncs read). A one-shot import leaves no other trace, and installing the plugin for every home would add a command nobody asked for; the old spelling names the install.""" return (home / "import-sync.json").is_file() @dataclass(frozen=True) class LeftCoreFeature: plugin: str # catalog entry name == installed plugin name label: str # user-facing feature name in_use: Callable[..., bool] # (home, *, process_env=False) -> bool, read-only unchanged: str # what migrated users keep, shown on success # Credentials core stripped from every child process while it shipped the feature. Core keeps # stripping them (tools/environments/local_env_policy.py): with the plugin absent (migration # pending, declined or failed) no manifest declares them, and they would reach every child. secret_env: tuple[str, ...] = () # Non-secret settings core kept out of children by default (provider blocklist, Tier 2). private_env: tuple[str, ...] = () # Toolsets the plugin registers. Core resolved them like any built-in toolset (on only where the # platform's saved list or default composite carried them); a plugin toolset is on for every # platform whose known_plugin_toolsets does not name it, so :func:`_record_migration` records # them there wherever core had them off. toolsets: tuple[str, ...] = () # Platforms whose core default composite (``hermes-``) never carried those toolsets. off_platforms: tuple[str, ...] = () # Gateway platform the plugin ships: a running gateway serves it only after a restart. platform: str = "" # Channel ownership core declared for that platform, kept while the plugin is absent (migration # pending, declined, failed) so a channel-less profile clone still strips the source's identity # (hermes_cli/profile_channels.py): the env whose presence enabled the adapter, and the env # prefixes it shared with the feature's tools (stripped only when the source runs the adapter). enable_env: tuple[str, ...] = () channel_env_prefixes: tuple[str, ...] = () # Top-level command the plugin registers in place of core's ``hermes auth `` login # (``hermes login|status|logout``); :func:`moved_command_hint` points the old spelling at it. cli: str = "" # Top-level command the plugin registers: core's own command kept under its name (``hermes # import-agent``), or in place of core's nested ``old_command`` (``hermes mcp serve`` -> ``hermes # mcp-serve``); :func:`moved_command_hint` points the old spelling at it. command: str = "" old_command: str = "" # Other top-level commands core shipped that the same plugin now registers (``hermes claw``, the # OpenClaw migration's spelling before ``hermes import-agent openclaw``); hinted like *command*. old_cli: tuple[str, ...] = () # Key of the bundled plugin core shipped (``observability/langfuse``). It stays an alias of the # catalog plugin in ``plugins.enabled`` / ``plugins.disabled`` (the loader's gate, an # administrator's pinned list) and for every `hermes plugins ` (:func:`legacy_plugin_alias`). plugin_key: str = "" # ``terminal.backend`` value the plugin registers: unknown-backend errors name the install command. terminal_backend: str = "" LEFT_CORE: tuple[LeftCoreFeature, ...] = ( LeftCoreFeature( plugin="homeassistant", label="Home Assistant", in_use=homeassistant_in_use, unchanged="HASS_TOKEN/HASS_URL, platforms.homeassistant and the ha_* tool names are unchanged", secret_env=("HASS_TOKEN",), private_env=("HASS_URL",), toolsets=("homeassistant",), off_platforms=("acp", "webhook"), platform="homeassistant", enable_env=("HASS_TOKEN",), channel_env_prefixes=("HASS_",), ), LeftCoreFeature( plugin="spotify", label="Spotify", in_use=spotify_in_use, unchanged="your Spotify login, the spotify toolset and the spotify_* tools are unchanged; " "`hermes auth spotify` is now `hermes spotify login`", cli="spotify", ), LeftCoreFeature( plugin="langfuse", label="Langfuse", in_use=langfuse_in_use, unchanged="your Langfuse keys and HERMES_LANGFUSE_* settings are unchanged; the plugin is now " "`langfuse` (was observability/langfuse)", secret_env=("HERMES_LANGFUSE_SECRET_KEY",), private_env=("HERMES_LANGFUSE_PUBLIC_KEY", "HERMES_LANGFUSE_BASE_URL"), plugin_key="observability/langfuse", ), LeftCoreFeature( # MCP clients that launch `hermes mcp serve` are configured outside Hermes, so no home shows the # feature in use: nothing installs automatically, and the old spelling prints where it went. plugin="mcp-serve", label="Hermes as an MCP server", in_use=lambda home, **_: False, unchanged="the conversation tools keep their names; MCP clients change their args to [\"mcp-serve\"]", command="mcp-serve", old_command="hermes mcp serve", ), LeftCoreFeature( plugin="singularity", label="Singularity/Apptainer terminal backend", in_use=singularity_in_use, unchanged="terminal.backend: singularity, terminal.singularity_image (TERMINAL_SINGULARITY_IMAGE) " "and the container_* limits are unchanged", terminal_backend="singularity", ), LeftCoreFeature( plugin="agent-importers", label="`hermes import-agent`", in_use=agent_imports_in_use, unchanged="`hermes import-agent`, its flags and your import-sync.json sync sources are unchanged", command="import-agent", old_cli=("claw",), ), ) def _install_sentence(feature: LeftCoreFeature) -> str: from hermes_constants import get_hermes_home return (f"{feature.label} moved out of core into the '{feature.plugin}' plugin: install it with " f"`{_install_command(feature.plugin, Path(get_hermes_home()))}`") def platform_install_hint(platform: str) -> str: """For an unknown platform name that left core: the sentence that names its install command for the active home, else ``""``. Appended to "unknown platform" errors (``send_message``).""" feature = next((f for f in LEFT_CORE if f.platform and f.platform == platform), None) return f". {_install_sentence(feature)}" if feature else "" def terminal_backend_install_hint(backend: str) -> str: """For a ``terminal.backend`` value that left core: the sentence that names its install command for the active home, else ``""``. Appended to "unknown terminal backend" errors.""" feature = next((f for f in LEFT_CORE if f.terminal_backend and f.terminal_backend == backend), None) return f". {_install_sentence(feature)}" if feature else "" def legacy_plugin_alias(name: str) -> str: """The catalog plugin a bundled key that left core now names (``observability/langfuse`` -> ``langfuse``), else ``""``.""" return next((f.plugin for f in LEFT_CORE if f.plugin_key and f.plugin_key == name), "") def legacy_plugin_keys(plugin: str) -> set[str]: """Bundled keys that now name catalog *plugin* (inverse of :func:`legacy_plugin_alias`).""" return {f.plugin_key for f in LEFT_CORE if f.plugin_key and f.plugin == plugin} def moved_command_hint(prog: str, value: str, action: str = "") -> str: """For a command a left-core plugin took over: ``hermes auth [] `` (core's old login spelling), core's old nested spelling (``hermes mcp serve``), or the plugin's own ``hermes `` / ``hermes `` / ``hermes `` while the plugin is not loaded, the line saying where it went and how to get it for the active home, else ``""``.""" from hermes_constants import get_hermes_home home = Path(get_hermes_home()) moved = next((f for f in LEFT_CORE if f.old_command and f.old_command == f"{prog} {value}"), None) if moved is not None: present = plugin_present(moved.plugin, home) get = "" if present else f" Install it with: `{_install_command(moved.plugin, home)}`." return (f"{moved.label} moved out of core into the '{moved.plugin}' plugin: `{moved.old_command}` is now " f"`hermes {moved.command}`.{get}") if prog == "hermes auth": feature = next((f for f in LEFT_CORE if f.cli and f.cli == value), None) elif prog == "hermes": feature = next((f for f in LEFT_CORE if value and value in (f.cli, f.command, *f.old_cli)), None) else: feature = None if feature is None: return "" install = _install_command(feature.plugin, home) if not plugin_present(feature.plugin, home): get = f" Install it with: `{install}`." else: get = f" Enable it with: `{install.replace(' install ', ' enable ')}`." if prog == "hermes" else "" if prog == "hermes auth": old = " ".join(p for p in ("hermes auth", action, value) if p) return (f"{feature.label} moved out of core into the '{feature.plugin}' plugin: `{old}` is now " f"`hermes {value} {action or 'login'}`.{get}") label = f"`hermes {value}`" if value in feature.old_cli else feature.label return f"{label} moved out of core into the '{feature.plugin}' plugin, which is not loaded.{get}" def moved_plugin_hint(name: str) -> str: """For a plugin name nothing resolves to (so the plugin is not installed): the bundled key of a plugin that left core (``observability/langfuse``) or its catalog name, the line naming the install command for the active home, else ``""``. Appended to "No plugin named ..." errors.""" feature = next((f for f in LEFT_CORE if f.plugin_key and name in (f.plugin_key, f.plugin)), None) return f"{_install_sentence(feature)}." if feature else "" _attempted: set[str] = set() _undelivered: dict[str, list[str]] = {} def plugin_present(plugin: str, home: Path) -> bool: """Installed in *home* (enabled or not: a user who disabled it chose to). Read-only.""" return (home / "plugins" / plugin).is_dir() def _core_carried(feature: LeftCoreFeature, selection: list) -> bool: """Whether core resolved *feature*'s toolsets on for a platform selecting *selection*: named directly, or through a ``hermes`` / ``hermes-`` composite that included them.""" def carries(name: str) -> bool: if name in feature.toolsets or name == "hermes": return True return name.startswith("hermes-") and name[len("hermes-"):].replace("-", "_") not in feature.off_platforms return any(carries(str(name)) for name in selection) _SCOPED_KEY = "_left_core_scoped" # config.yaml: rows whose toolset scope this home already converted # config.yaml: rows this home has had installed (by the migration, or by the user before it ran). # Never written before the plugin dir exists, so a home whose install never succeeded keeps retrying. _INSTALLED_KEY = "_left_core_installed" def _marked(config: dict, key: str, feature: LeftCoreFeature) -> bool: done = config.get(key) return isinstance(done, list) and feature.plugin in map(str, done) def _scope_kept(config: dict, feature: LeftCoreFeature) -> bool: return not feature.toolsets or _marked(config, _SCOPED_KEY, feature) def _mark(config: dict, key: str, feature: LeftCoreFeature) -> None: done = config.get(key) config[key] = sorted({*(map(str, done) if isinstance(done, list) else ()), feature.plugin}) def _record_migration(home: Path, feature: LeftCoreFeature) -> None: """Keep *feature*'s core-era toolset scope, and mark it installed once its plugin is in *home*, in one config write. Scope: record the toolsets in ``known_plugin_toolsets[platform]`` (= off) for every platform where core had them off: a saved ``platform_toolsets`` list that does not carry them, or no list on an ``off_platforms`` platform. Platforms core resolved them on for are left alone. A one-time conversion per home and row, independent of installation: a plugin installed before the update (it stays inert on a core that still ships the feature) still needs it. Completion is marked under ``_left_core_scoped`` so a later run never undoes a choice the user made since (``hermes tools``). Raises when config.yaml cannot be read or written.""" from hermes_cli.config import atomic_config_write, read_user_config_raw from hermes_cli.toolset_validation import parse_platform_toolsets_value path = home / "config.yaml" config = read_user_config_raw(path) installed = plugin_present(feature.plugin, home) and not _marked(config, _INSTALLED_KEY, feature) if _scope_kept(config, feature) and not installed: return if installed: _mark(config, _INSTALLED_KEY, feature) if not _scope_kept(config, feature): saved = config.get("platform_toolsets") saved = saved if isinstance(saved, dict) else {} selections = {str(p): parse_platform_toolsets_value(v) for p, v in saved.items()} for platform in feature.off_platforms: selections.setdefault(platform, None) known = config.get("known_plugin_toolsets") known = known if isinstance(known, dict) else {} for platform, selection in selections.items(): if selection is None and platform not in feature.off_platforms: continue # no (valid) saved list: core's default composite carried the toolsets if selection is not None and _core_carried(feature, selection): continue current = known.get(platform) if isinstance(known.get(platform), list) else [] if missing := [ts for ts in feature.toolsets if ts not in current]: known[platform] = sorted({*map(str, current), *missing}) if known: config["known_plugin_toolsets"] = known _mark(config, _SCOPED_KEY, feature) atomic_config_write(path, config) def _pending(home: Path, *, say: Callable[[str], None], process_env: bool = False, backoff: bool = False) -> list[LeftCoreFeature]: """Rows *home* uses whose plugin it never had and that the catalog ships (a catalog miss is reported through *say*). Converts the toolset scope of every row *home* uses once (:func:`_record_migration`), installed or not; a row whose scope cannot be recorded is reported and skipped, never installed unscoped. A row marked installed is done for good. *backoff* (startup) skips a row whose last attempt failed recently, before any network.""" from hermes_cli.memory_provider_migration import catalog_source out = [] for feature in LEFT_CORE: if _marked(_read_config(home), _INSTALLED_KEY, feature): continue if not feature.in_use(home, process_env=process_env): continue present = plugin_present(feature.plugin, home) try: _record_migration(home, feature) except Exception as exc: say(f" ⚠ {feature.label} moved out of core into the '{feature.plugin}' plugin: its per-platform " f"toolset selection could not be kept ({exc}). Check `hermes tools`" + ("." if present else f" and run `{_install_command(feature.plugin, home)}`.")) continue if present: continue if backoff and _failed_recently(home, feature.plugin): logger.info("%s plugin install failed recently for %s; retrying later or on `hermes update`", feature.label, home) continue if catalog_source(feature.plugin) is None: _note_failure(home, feature.plugin) say(f" ⚠ {feature.label} moved out of core into the '{feature.plugin}' plugin, which this " f"Hermes cannot find in the plugin catalog yet. Run `{_install_command(feature.plugin, home)}` " f"once it is listed.") continue out.append(feature) return out def _install_into(home: Path, *, consent: Optional[bool] = None) -> Callable[[str], dict]: def _install(name: str) -> dict: from hermes_cli.plugins_cmd import dashboard_install_plugin from hermes_constants import reset_hermes_home_override, set_hermes_home_override token = set_hermes_home_override(home) try: return dashboard_install_plugin("", force=False, enable=True, catalog_name=name, assume_deps_consent=_unattended_consent() if consent is None else consent) finally: reset_hermes_home_override(token) return _install def _first_cause(error: str) -> str: """One line for a notice: git's own ``fatal:`` / ``error:`` line when *error* ends with raw git stderr (a failed clone), else its first non-empty line.""" lines = [line.strip() for line in error.splitlines() if line.strip()] cause = next((line.split(":", 1)[1].strip() for line in lines if line.startswith(("fatal:", "error:"))), None) return (cause or (lines[0] if lines else "unknown error")).rstrip(". ") def _install_one(home: Path, feature: LeftCoreFeature, *, install: Callable[[str], dict], say: Callable[[str], None]) -> bool: try: result = install(feature.plugin) except Exception as exc: # network, uv, kill list — report, do not raise result = {"ok": False, "error": str(exc)} if result.get("ok"): try: _record_migration(home, feature) except Exception as exc: # the next run sees the plugin dir and records it then logger.debug("left-core install marker not written for %s: %s", home, exc) say(f" ✓ {feature.label} moved out of core — installed the '{feature.plugin}' plugin from the " f"catalog ({feature.unchanged}).") return True _note_failure(home, feature.plugin) error = _first_cause(str(result.get("error") or "")) say(f" ⚠ {feature.label} moved out of core and its '{feature.plugin}' plugin could not be installed " f"automatically: {error}. Run `{_install_command(feature.plugin, home)}`.") return False def migrate_home(home: Path, *, install: Callable[[str], dict], say: Callable[[str], None] = print, process_env: bool = False) -> list[str]: """Install every left-core plugin *home* uses and lacks. Returns installed plugin names; never raises.""" installed = [] for feature in _pending(home, say=say, process_env=process_env): if _install_one(home, feature, install=install, say=say): installed.append(feature.plugin) return installed def migrate_all_homes(*, say: Callable[[str], None] = print) -> list[str]: """``hermes update`` hook: every profile home sharing this venv. Grouped by plugin and each home's unattended consent like the memory migration: homes of one group share dependency answers, and a failure names the rest of its group in one line instead of failing them one by one.""" from hermes_cli.plugins_cmd_install import shared_dependency_answers from pm.plugins_state import dependency_homes def labelled(home: Path) -> Callable[[str], None]: return lambda message: say(f" [{_home_label(home)}] {message.lstrip()}") pending: dict[tuple[str, bool], tuple[LeftCoreFeature, list[Path]]] = {} for home in dependency_homes(): try: features = _pending(home, say=labelled(home)) consent = bool(features) and _home_consent(home) except Exception as exc: logger.debug("left-core migration skipped for %s: %s", home, exc) continue for feature in features: pending.setdefault((feature.plugin, consent), (feature, []))[1].append(home) installed: list[str] = [] try: for feature, homes in pending.values(): if len(homes) > 1 and _interactive(): say(f" {feature.label} is used in {len(homes)} profiles " f"({', '.join(_home_label(h) for h in homes)}); your answers to its dependency " f"questions apply to all of them.") with shared_dependency_answers(): for index, home in enumerate(homes): if _install_one(home, feature, install=_install_into(home), say=labelled(home)): installed.append(feature.plugin) continue rest = homes[index + 1:] if rest: say(f" ⚠ The '{feature.plugin}' plugin was not installed for " f"{', '.join(_home_label(h) for h in rest)} either. Run " + ", ".join(f"`{_install_command(feature.plugin, h)}`" for h in rest) + ".") break except KeyboardInterrupt: say(" ⚠ Plugin migration cancelled. Profiles already migrated keep their plugin; run " "`hermes plugins install ` (with `-p `) for the rest.") return installed def _gateway_serves(home: Path) -> bool: """A live gateway serves *home*: it loaded its platforms before this agent-start install.""" try: from gateway.status import resolve_gateway_liveness return resolve_gateway_liveness(profile_dir=home, use_cache=False).running except Exception: return False def recover_at_startup(*, say: Optional[Callable[[str], None]] = None) -> list[str]: """Agent/gateway start hook for the active home: one attempt per process per home. With *say* (an agent's startup-warning sink) outcomes are delivered now, together with any a gateway-start attempt queued for this home; without it they are logged and queued for the home's first agent. Returns installed plugin names.""" from agent.safe_worker_policy import worker_config_snapshot if worker_config_snapshot() is not None: # A frozen safe/ignore-user-config worker never reads the profile's config.yaml and never # installs customizations; the owner process runs this migration for the home. return [] from hermes_constants import get_hermes_home, hermes_home_key home = Path(get_hermes_home()) key = hermes_home_key(home) def deliver(message: str) -> None: if say is None: _undelivered.setdefault(key, []).append(message) return try: say(message) except Exception: logger.debug("left-core migration notification failed", exc_info=True) if say is not None: for message in _undelivered.pop(key, []): deliver(message) if key in _attempted: return [] _attempted.add(key) def report(message: str) -> None: message = message.strip() logger.warning(message) deliver(message) try: features = _pending(home, say=report, process_env=True, backoff=True) if not features: return [] from pm.install import lazy_installs_allowed if not lazy_installs_allowed(): for feature in features: report(f"⚠ {feature.label} moved out of core and its '{feature.plugin}' plugin is not " f"installed, so it is off. security.allow_lazy_installs is off, so Hermes did not " f"fetch it: run `{_install_command(feature.plugin, home)}`.") return [] installed = [] # Agent init cannot answer a dependency prompt: under the CLI the chat input owns the terminal, # so the question hangs at "Initializing agent...". Every row shipped in core, so its users # already accepted its dependencies (as memory_provider_migration.recover_at_startup does). for feature in features: if _install_one(home, feature, install=_install_into(home, consent=True), say=report): installed.append(feature.plugin) if say is not None and feature.platform and _gateway_serves(home): report(f"Restart the gateway (`hermes gateway restart`) so it serves {feature.label}.") return installed except Exception as exc: # never take agent/gateway start down logger.warning("left-core plugin migration failed: %s", exc, exc_info=True) return [] def recover_for_command(name: Optional[str]) -> None: """``hermes `` names a command a left-core plugin registers, and the active home uses that feature but lacks the plugin: run :func:`recover_at_startup` now, before plugin commands are discovered, so the command it installs runs in this same invocation. A home nothing migrated yet (a Docker image swap runs no ``hermes update``) whose only Hermes process is a scheduled ``hermes import-agent --sync`` never starts an agent or a gateway to do it.""" from hermes_constants import get_hermes_home home = Path(get_hermes_home()) if any(name and name in (f.cli, f.command, *f.old_cli) and not plugin_present(f.plugin, home) and f.in_use(home) for f in LEFT_CORE): recover_at_startup(say=lambda message: print(message, file=sys.stderr))