refactor(hermes_cli): compact plugin manifest/loader/discovery/ledger/state helpers (-100 LOC)

hermes_cli/plugins_discovery.py

@@ -38,13 +38,10 @@ def _select_entry_point_group(entry_points: Any, group: str) -> list:
 
 
 def discover_entrypoint_manifests() -> List["PluginManifest"]:
-    """Return metadata-only manifests for installed entry-point plugins.
-
-    Kind comes from an import-free source scan (memory/model providers route to their own
-    discovery). Capabilities come from the companion ``hermes_agent.plugin_capabilities`` group
-    (``<plugin-id>.<capability-id>`` entries pointing at the same object), so consent works without
-    importing plugin code. Failures are isolated per entry point.
-    """
+    """Return metadata-only manifests for installed entry-point plugins. Kind comes from an import-free source
+    scan (memory/model providers route to their own discovery). Capabilities come from the companion
+    ``hermes_agent.plugin_capabilities`` group (``<plugin-id>.<capability-id>`` entries pointing at the same
+    object), so consent works without importing plugin code. Failures are isolated per entry point."""
     manifests: List[PluginManifest] = []
     try:
         eps = importlib.metadata.entry_points()
@@ -55,13 +52,8 @@ def discover_entrypoint_manifests() -> List["PluginManifest"]:
         return manifests
     for ep in group_eps:
         try:
-            capabilities = [
-                capability for capability in VALID_CAPABILITY_IDS
-                if any(
-                    declaration.name == f"{ep.name}.{capability}" and declaration.value == ep.value
-                    for declaration in capability_eps
-                )
-            ]
+            declared = {d.name for d in capability_eps if d.value == ep.value}
+            capabilities = [c for c in VALID_CAPABILITY_IDS if f"{ep.name}.{c}" in declared]
             dist = getattr(ep, "dist", None)
             metadata = getattr(dist, "metadata", None)
             manifests.append(PluginManifest(
@@ -80,9 +72,7 @@ def _classify_entrypoint_value_kind(value: str) -> str:
     """Classify an entry-point target by import-free source scan (unresolvable -> standalone)."""
     try:
         module_name = str(value).split(":", 1)[0].strip()
-        if not module_name:
-            return "standalone"
-        return _detect_kind_from_source(_resolve_module_source(module_name)) or "standalone"
+        return (_detect_kind_from_source(_resolve_module_source(module_name)) if module_name else None) or "standalone"
     except Exception:
         return "standalone"
 
@@ -98,11 +88,9 @@ def _get_disabled_plugins() -> set:
 
 
 def _get_enabled_plugins() -> Optional[set]:
-    """Read the ``plugins.enabled`` allow-list (plugins are opt-in).
-
-    ``None`` = key missing/malformed ("nothing enabled yet"; the first ``migrate_config`` run
-    grandfathers installed user plugins); ``set()`` = explicitly empty; else the allow-list.
-    """
+    """Read the ``plugins.enabled`` allow-list (plugins are opt-in). ``None`` = key missing/malformed ("nothing
+    enabled yet"; the first ``migrate_config`` run grandfathers installed user plugins); ``set()`` = explicitly
+    empty; else the allow-list."""
     try:
         from hermes_cli.config import load_config
         enabled = cfg_get(load_config(), "plugins", "enabled")
@@ -112,46 +100,41 @@ def _get_enabled_plugins() -> Optional[set]:
 
 
 def scan_directory(
-    path: Path, source: str, *, skip_names: Optional[Set[str]] = None, prefix: str = "",
-    depth: int = 0,
+    path: Path, source: str, *, skip_names: Optional[Set[str]] = None, prefix: str = "", depth: int = 0
 ) -> List[PluginManifest]:
     """Read manifests under *path*: flat ``<root>/<name>/plugin.yaml`` (key ``name``) or category
-    ``<root>/<cat>/<name>/plugin.yaml`` (key ``cat/name``; a manifest-less directory recurses one
-    level, depth capped at two). *skip_names* ignores top-level names; portable ``plugin.json``
-    packages are accepted alongside YAML manifests."""
+    ``<root>/<cat>/<name>/plugin.yaml`` (key ``cat/name``; a manifest-less directory recurses one level, depth
+    capped at two). *skip_names* ignores top-level names; portable ``plugin.json`` packages are accepted
+    alongside YAML manifests."""
     manifests: List[PluginManifest] = []
     if not path.is_dir():
         return manifests
     for child in sorted(path.iterdir()):
         if not child.is_dir() or (depth == 0 and skip_names and child.name in skip_names):
             continue
-        manifest_file = child / "plugin.yaml"
-        if not manifest_file.exists():
-            manifest_file = child / "plugin.yml"
-        if manifest_file.exists():
+        manifest_file = next((f for f in (child / "plugin.yaml", child / "plugin.yml") if f.exists()), None)
+        portable_file = child / "plugin.json"
+        if manifest_file is not None:
             manifest = parse_manifest_file(manifest_file, child, source, prefix)
             if manifest is not None:
                 manifests.append(manifest)
-            continue
-        portable_file = child / "plugin.json"
-        if portable_file.exists() or portable_file.is_symlink():
+        elif portable_file.exists() or portable_file.is_symlink():
             try:
                 manifests.append(portable_plugin_manifest(child, source, prefix))
             except Exception as exc:
                 logger.warning("Failed to parse %s: %s", portable_file, exc)
-            continue
-        if depth >= 1:
+        elif depth >= 1:
             logger.debug("Skipping %s (no plugin.yaml, depth cap reached)", child)
-            continue
-        sub_prefix = f"{prefix}/{child.name}" if prefix else child.name
-        manifests.extend(scan_directory(child, source, prefix=sub_prefix, depth=depth + 1))
+        else:
+            sub_prefix = f"{prefix}/{child.name}" if prefix else child.name
+            manifests.extend(scan_directory(child, source, prefix=sub_prefix, depth=depth + 1))
     return manifests
 
 
 def collect_directory_manifests() -> List[PluginManifest]:
-    """Read directory manifests in full-discovery order (bundled top-level, bundled/platforms, user,
-    opt-in project) without loading or mutating anything, so startup probes share the exact
-    precedence/containment rules of the real discovery sweep."""
+    """Read directory manifests in full-discovery order (bundled top-level, bundled/platforms, user, opt-in
+    project) without loading or mutating anything, so startup probes share the exact precedence/containment
+    rules of the real discovery sweep."""
     from hermes_cli import plugins as _origin  # patched names resolve through the origin
     manifests: List[PluginManifest] = []
 
@@ -163,8 +146,7 @@ def collect_directory_manifests() -> List[PluginManifest]:
     # Excluded bundled top-level categories have their own discovery; platforms scan separately.
     repo_plugins = _origin.get_bundled_plugins_dir()
     logger.debug("Scanning bundled plugins: %s", repo_plugins)
-    _scan("bundled (top-level)", repo_plugins, "bundled",
-          {"memory", "context_engine", "platforms", "model-providers"})
+    _scan("bundled (top-level)", repo_plugins, "bundled", {"memory", "context_engine", "platforms", "model-providers"})
     _scan("bundled/platforms", repo_plugins / "platforms", "bundled")
     user_dir = get_hermes_home() / "plugins"
     logger.debug("Scanning user plugins: %s", user_dir)
@@ -191,51 +173,46 @@ class ManifestGate:
 def gate_manifest(
     manifest: PluginManifest, disabled: Set[str], enabled: Optional[Set[str]]
 ) -> ManifestGate:
-    """Decide how one winning manifest is handled. Gate order matters: legacy relay refusal, explicit
-    disable, category-owned kinds (exclusive / model-provider), bundled auto-loads (backend now,
-    platform deferred), then ``plugins.enabled`` opt-in (path-derived key or legacy bare name)."""
+    """Decide how one winning manifest is handled. Gate order matters: legacy relay refusal, explicit disable,
+    category-owned kinds (exclusive / model-provider), bundled auto-loads (backend now, platform deferred),
+    then ``plugins.enabled`` opt-in (path-derived key or legacy bare name)."""
     lookup_key = manifest_key(manifest)
     names = {lookup_key, manifest.name}
+
+    def _placeholder(error: Optional[str], level: int, message: str, *args, enabled: bool = False) -> ManifestGate:
+        return ManifestGate("placeholder", enabled=enabled, error=error, log=(level, message, lookup_key, *args))
+
     # Relay lifecycle is core-owned; an old plugin copy would compete for its registries.
     if names & LEGACY_RELAY_PLUGIN_KEYS:
         error = (
             "removed — Relay lifecycle is owned by Hermes core; configure "
             f"{RELAY_PLUGINS_CONFIG_ENV} instead"
         )
-        return ManifestGate(
-            "placeholder", error=error,
-            log=(logging.WARNING, "Refusing to load removed Hermes Relay plugin '%s'; %s", lookup_key, error),
-        )
+        return _placeholder(error, logging.WARNING, "Refusing to load removed Hermes Relay plugin '%s'; %s", error)
     if names & disabled:
-        return ManifestGate(
-            "placeholder", error="disabled via config",
-            log=(logging.DEBUG, "Skipping disabled plugin '%s'", lookup_key),
-        )
+        return _placeholder("disabled via config", logging.DEBUG, "Skipping disabled plugin '%s'")
     # Exclusive plugins (memory providers) have their own activation path; record only.
     if manifest.kind == "exclusive":
-        return ManifestGate(
-            "placeholder", error="exclusive plugin — activate via <category>.provider config",
-            log=(logging.DEBUG, "Skipping '%s' (exclusive, handled by category discovery)", lookup_key),
+        return _placeholder(
+            "exclusive plugin — activate via <category>.provider config", logging.DEBUG,
+            "Skipping '%s' (exclusive, handled by category discovery)",
         )
-    # Model providers load via providers/__init__.py; a second import here would create two
-    # ProviderProfile instances and break the bundled-vs-user "last writer wins" override.
+    # Model providers load via providers/__init__.py; a second import here would create two ProviderProfile
+    # instances and break the bundled-vs-user "last writer wins" override.
     if manifest.kind == "model-provider":
-        return ManifestGate(
-            "placeholder", enabled=True,
-            log=(logging.DEBUG, "Skipping '%s' (model-provider, handled by providers/ discovery)", lookup_key),
-        )
+        return _placeholder(
+            None, logging.DEBUG, "Skipping '%s' (model-provider, handled by providers/ discovery)", enabled=True)
     if manifest.source == "bundled":
         # Bundled backends auto-load; selection among them is ``<category>.provider`` config.
         if manifest.kind == "backend":
             return ManifestGate("load_now")
-        # Bundled platforms register LAZILY: eagerly importing ~20 heavy SDKs added seconds to every
-        # `hermes` invocation. A deferred loader keeps every platform available on first use.
+        # Bundled platforms register LAZILY: eagerly importing ~20 heavy SDKs added seconds to every `hermes`
+        # invocation. A deferred loader keeps every platform available on first use.
         if manifest.kind == "platform":
             return ManifestGate("defer")
     if enabled is None or not names & enabled:
-        return ManifestGate(
-            "placeholder",
-            error=f"not enabled in config (run `hermes plugins enable {lookup_key}` to activate)",
-            log=(logging.DEBUG, "Skipping '%s' (not in plugins.enabled)", lookup_key),
+        return _placeholder(
+            f"not enabled in config (run `hermes plugins enable {lookup_key}` to activate)", logging.DEBUG,
+            "Skipping '%s' (not in plugins.enabled)",
         )
     return ManifestGate("load")

hermes_cli/plugins_ledger.py

@@ -1,6 +1,5 @@
-"""Registration ownership ledger and unload: every plugin registration is recorded with its inverse
-so force reload / targeted unload unwind registries in reverse order. Mixed into
-:class:`hermes_cli.plugins.PluginManager`.
+"""Registration ownership ledger and unload: every plugin registration is recorded with its inverse so force
+reload / targeted unload unwind registries in reverse order. Mixed into :class:`hermes_cli.plugins.PluginManager`.
 """
 
 from __future__ import annotations
@@ -21,21 +20,19 @@ logger = logging.getLogger("hermes_cli.plugins")
 
 @dataclass
 class PluginRegistration:
-    """One host-owned registration plus its inverse, so force reload unwinds registries in reverse
-    order (including restoring the entry an override replaced)."""
+    """One host-owned registration plus its inverse, so force reload unwinds registries in reverse order
+    (including restoring the entry an override replaced)."""
 
     kind: str
     key: str
     release: Callable[[], None]
     plugin_key: str = ""
-    # Process-global host infrastructure (e.g. dashboard-auth providers): kept out of
-    # ``_registration_order`` so unload-all cannot dispose it, but still disposed by a *targeted*
-    # unload and evicted on force re-discovery when the plugin no longer re-registers it.
+    # Process-global host infrastructure (e.g. dashboard-auth providers): kept out of ``_registration_order``
+    # so unload-all cannot dispose it, but still disposed by a *targeted* unload and evicted on force
+    # re-discovery when the plugin no longer re-registers it.
     persistent: bool = False
     _disposed: bool = field(default=False, init=False, repr=False)
-    _on_dispose: Optional[Callable[["PluginRegistration"], None]] = field(
-        default=None, init=False, repr=False
-    )
+    _on_dispose: Optional[Callable[["PluginRegistration"], None]] = field(default=None, init=False, repr=False)
 
     @property
     def active(self) -> bool:
@@ -59,14 +56,11 @@ class PluginLedgerMixin:
         self, manifest: PluginManifest, kind: str, key: str, release: Callable[[], None], *,
         persistent: bool = False,
     ) -> PluginRegistration:
-        """Record one registration under its canonical plugin key. ``persistent`` ones
-        (process-global host infrastructure) stay in the ownership ledger for attribution but NOT in
-        ``_registration_order``, so a routine unload cannot dispose them; the handle still releases
-        on explicit ``dispose()``."""
+        """Record one registration under its canonical plugin key. ``persistent`` ones (process-global host
+        infrastructure) stay in the ownership ledger for attribution but NOT in ``_registration_order``, so a
+        routine unload cannot dispose them; the handle still releases on explicit ``dispose()``."""
         registration = PluginRegistration(
-            kind=kind, key=key, release=release, plugin_key=manifest_key(manifest),
-            persistent=persistent,
-        )
+            kind=kind, key=key, release=release, plugin_key=manifest_key(manifest), persistent=persistent)
         registration._on_dispose = lambda disposed: self._forget_registrations([disposed])
         self._ownership_ledger.setdefault(registration.plugin_key, []).append(registration)
         if not persistent:
@@ -77,29 +71,25 @@ class PluginLedgerMixin:
         self, manifest: PluginManifest, kind: str, name: str, registry: Any, current: Any,
         previous: Any, *, finalize: Optional[Callable[[], None]] = None,
     ) -> PluginRegistration:
-        """Lease one ``(kind, scope, name)`` slot of a scope-keyed process-global registry. Unload
-        calls ``registry.restore_registration(name, current, replacement, scope=...)`` —
-        identity-conditional, so a later generation is never removed by an earlier owner."""
+        """Lease one ``(kind, scope, name)`` slot of a scope-keyed process-global registry. Unload calls
+        ``registry.restore_registration(name, current, replacement, scope=...)`` — identity-conditional, so a
+        later generation is never removed by an earlier owner."""
         scope = self.scope_key
         lease = replacement_coordinator.acquire(
             (kind, scope, name), current=current, previous=previous,
-            restore=lambda replacement: registry.restore_registration(
-                name, current, replacement, scope=scope
-            ), finalize=finalize,
+            restore=lambda replacement: registry.restore_registration(name, current, replacement, scope=scope),
+            finalize=finalize,
         )
         return self._track_registration(manifest, kind, name, lease.dispose)
 
     def _active_persistent(self) -> List[PluginRegistration]:
         """Live persistent registrations across every plugin in the ownership ledger."""
-        return [
-            registration for owned in self._ownership_ledger.values()
-            for registration in owned if registration.persistent and registration.active
-        ]
+        return [r for owned in self._ownership_ledger.values() for r in owned if r.persistent and r.active]
 
     def _evict_stale_persistent_registrations(self) -> None:
-        """After re-discovery, dispose parked persistent handles whose plugin did not re-register
-        the same ``(kind, key)``. Re-registered ones are dropped WITHOUT disposing — the same object
-        re-registered would pass the identity check and evict the live entry."""
+        """After re-discovery, dispose parked persistent handles whose plugin did not re-register the same
+        ``(kind, key)``. Re-registered ones are dropped WITHOUT disposing — the same object re-registered would
+        pass the identity check and evict the live entry."""
         if not self._persistent_carryover:
             return
         parked, self._persistent_carryover = self._persistent_carryover, []
@@ -108,23 +98,20 @@ class PluginLedgerMixin:
         for registration in stale:
             logger.info(
                 "Evicting persistent registration %s/%s: plugin '%s' no "
-                "longer supplies it after re-discovery", registration.kind, registration.key,
-                registration.plugin_key,
+                "longer supplies it after re-discovery", registration.kind, registration.key, registration.plugin_key,
             )
         self._dispose_registrations(stale)
 
     @staticmethod
     def _remove_identity(values: list, target: Any) -> bool:
         """Remove the last exact object match from a registration list."""
-        for index in range(len(values) - 1, -1, -1):
-            if values[index] is target:
-                del values[index]
-                return True
-        return False
+        index = next((i for i in range(len(values) - 1, -1, -1) if values[i] is target), None)
+        if index is None:
+            return False
+        del values[index]
+        return True
 
-    def _remove_callback(
-        self, mapping: Dict[str, List[Callable]], key: str, callback: Callable,
-    ) -> None:
+    def _remove_callback(self, mapping: Dict[str, List[Callable]], key: str, callback: Callable) -> None:
         callbacks = mapping.get(key)
         if callbacks is None:
             return
@@ -132,9 +119,7 @@ class PluginLedgerMixin:
         if not callbacks:
             mapping.pop(key, None)
 
-    def _restore_mapping(
-        self, mapping: Dict[str, Any], key: str, current: Any, previous: Optional[Any],
-    ) -> bool:
+    def _restore_mapping(self, mapping: Dict[str, Any], key: str, current: Any, previous: Optional[Any]) -> bool:
         """Restore a manager-local mapping only when *current* is still present."""
         if mapping.get(key) is not current:
             return False
@@ -153,9 +138,7 @@ class PluginLedgerMixin:
 
     def _remove_name_if_unowned(self, kind: str, names: Set[str], name: str) -> None:
         """Drop *name* from the manager-local name set once no active ledger entry owns it."""
-        if not any(
-            r.active and r.kind == kind and r.key == name for r in self._registration_order
-        ):
+        if not any(r.active and r.kind == kind and r.key == name for r in self._registration_order):
             names.discard(name)
 
     def _remove_tool_name_if_unowned(self, name: str) -> None:
@@ -193,9 +176,7 @@ class PluginLedgerMixin:
         from hermes_cli.plugins import LoadedPlugin
         if isinstance(plugin, LoadedPlugin):
             return manifest_key(plugin.manifest)
-        if isinstance(plugin, PluginManifest):
-            return manifest_key(plugin)
-        return str(plugin)
+        return manifest_key(plugin) if isinstance(plugin, PluginManifest) else str(plugin)
 
     def unload(self, plugin: Union[str, PluginManifest, LoadedPlugin, None] = None) -> bool:
         """Unload registrations while excluding discovery/deferred loading."""
@@ -203,10 +184,9 @@ class PluginLedgerMixin:
             return self._unload_scoped(plugin)
 
     def _unload_scoped(self, plugin: Union[str, PluginManifest, LoadedPlugin, None] = None) -> bool:
-        """Unload one plugin (or all when ``plugin=None``, as force rediscovery does). Every ledger
-        registration — including on_unload callbacks and supervised tasks — is disposed in reverse
-        acquisition order with identity-conditional inverses. Returns ``True`` when anything was
-        found."""
+        """Unload one plugin (or all when ``plugin=None``, as force rediscovery does). Every ledger registration
+        — including on_unload callbacks and supervised tasks — is disposed in reverse acquisition order with
+        identity-conditional inverses. Returns ``True`` when anything was found."""
         unload_all = plugin is None
         if unload_all:
             target_keys = set(self._ownership_ledger) | set(self._plugins)
@@ -214,12 +194,11 @@ class PluginLedgerMixin:
         else:
             target_keys = self._unload_target_keys(self._resolve_plugin_key(plugin))
             registrations = [r for r in self._registration_order if r.plugin_key in target_keys]
-            # Persistent registrations are absent from _registration_order (unload-all keeps them),
-            # but a *targeted* unload is the disable/uninstall path: a disabled auth plugin's
-            # provider must NOT stay live process-wide.
+            # Persistent registrations are absent from _registration_order (unload-all keeps them), but a
+            # *targeted* unload is the disable/uninstall path: a disabled auth plugin's provider must NOT stay
+            # live process-wide.
             registrations.extend(
-                r for key in target_keys for r in self._ownership_ledger.get(key, [])
-                if r.persistent and r.active
+                r for key in target_keys for r in self._ownership_ledger.get(key, []) if r.persistent and r.active
             )
         found = bool(target_keys or registrations)
         self._dispose_registrations(registrations)
@@ -239,14 +218,14 @@ class PluginLedgerMixin:
 
     def _reset_after_unload_all(self, registrations: List[PluginRegistration]) -> None:
         """Sweep pre-ledger global state and clear every manager-local container."""
-        # Handles are authoritative for global registries; names present in the manager-local sets
-        # without a ledger entry (pre-ledger or manually set state) are swept here so they do not
-        # survive a force reload as zombies.
+        # Handles are authoritative for global registries; names present in the manager-local sets without a
+        # ledger entry (pre-ledger or manually set state) are swept here so they do not survive a force reload
+        # as zombies.
         from gateway.platform_registry import platform_registry
         for platform_name in tuple(self._plugin_platform_names):
             platform_registry.unregister(platform_name)
-        # Ledger-owned tool names are excluded: their handles already restored the previous entry,
-        # and blanket deregistration would remove what the ledger just restored.
+        # Ledger-owned tool names are excluded: their handles already restored the previous entry, and blanket
+        # deregistration would remove what the ledger just restored.
         ledger_tool_names = {r.key for r in registrations if r.kind == "tool"}
         preledger_tools = tuple(n for n in self._plugin_tool_names if n not in ledger_tool_names)
         if preledger_tools:
@@ -260,12 +239,10 @@ class PluginLedgerMixin:
                         tool_registry.deregister(tool_name)
                     except Exception as exc:
                         logger.debug("unload: tool deregister %s failed: %s", tool_name, exc)
-        # Persistent registrations survive unload-all but must not be orphaned by the ledger clear:
-        # carry them over so force re-discovery can evict the ones whose plugin does not come back.
+        # Persistent registrations survive unload-all but must not be orphaned by the ledger clear: carry them
+        # over so force re-discovery can evict the ones whose plugin does not come back.
         carryover_ids = {id(r) for r in self._persistent_carryover}
-        self._persistent_carryover.extend(
-            r for r in self._active_persistent() if id(r) not in carryover_ids
-        )
+        self._persistent_carryover.extend(r for r in self._active_persistent() if id(r) not in carryover_ids)
         for container in (
             self._ownership_ledger, self._plugins, self._hooks, self._middleware,
             self._plugin_tool_names, self._plugin_platform_names, self._cli_commands,

hermes_cli/plugins_loader.py

@@ -1,5 +1,5 @@
-"""Plugin loading: directory/entry-point module import, deferred bundled platforms, portable
-packages, dependency/config-schema warnings. Mixed into :class:`hermes_cli.plugins.PluginManager`.
+"""Plugin loading: directory/entry-point module import, deferred bundled platforms, portable packages,
+dependency/config-schema warnings. Mixed into :class:`hermes_cli.plugins.PluginManager`.
 
 Origin-internal names (``PluginContext``, ``LoadedPlugin``, ``_PLUGINS_DEBUG`` …) are imported lazily
 through ``hermes_cli.plugins`` so tests that patch them on the origin keep working.
@@ -64,11 +64,25 @@ def _plugin_home_scope(home: Path):
         reset_hermes_home_override(token)
 
 
+def _dist_installed(req: str) -> Optional[bool]:
+    """Best-effort presence probe on a requirement's distribution name; ``None`` when unprobeable."""
+    dist = re.split(r"[<>=!~\[;\s]", req, maxsplit=1)[0].strip()
+    if not dist:
+        return None
+    try:
+        importlib.metadata.version(dist)
+        return True
+    except importlib.metadata.PackageNotFoundError:
+        return False
+    except Exception:
+        return None
+
+
 class PluginLoaderMixin:
     @staticmethod
     def _platform_name_from_manifest(manifest: PluginManifest) -> str:
-        """Derive the platform name without importing the adapter: strip a trailing ``-platform``
-        from the manifest name, else the directory basename (the bundled convention)."""
+        """Derive the platform name without importing the adapter: strip a trailing ``-platform`` from the
+        manifest name, else the directory basename (the bundled convention)."""
         name = manifest.name or ""
         if name.endswith("-platform"):
             return name[: -len("-platform")]
@@ -77,8 +91,8 @@ class PluginLoaderMixin:
     @_serialized_replacement
     def _register_deferred_platform(self, manifest: PluginManifest) -> None:
         """Register a lazy loader for a bundled platform: the adapter imports only when the
-        ``platform_registry`` is first asked for it; a placeholder ``LoadedPlugin`` keeps it visible
-        in ``hermes plugins list`` until then."""
+        ``platform_registry`` is first asked for it; a placeholder ``LoadedPlugin`` keeps it visible in
+        ``hermes plugins list`` until then."""
         from hermes_cli.plugins import LoadedPlugin
         lookup_key = manifest_key(manifest)
         platform_name = self._platform_name_from_manifest(manifest)
@@ -89,9 +103,8 @@ class PluginLoaderMixin:
             scope = self.scope_key
 
             def _loader(_manifest: PluginManifest = manifest) -> None:
-                # Lock before checking cancellation: if an unload won the race it restored the
-                # predecessor and this loader must publish nothing; if loading won, unload waits
-                # and disposes the completed set.
+                # Lock before checking cancellation: if an unload won the race it restored the predecessor
+                # and this loader must publish nothing; if loading won, unload waits and disposes the set.
                 with self._discovery_lock, _plugin_home_scope(self.home_path):
                     if platform_registry.is_deferred_load_cancelled(platform_name, scope=scope):
                         return
@@ -106,26 +119,20 @@ class PluginLoaderMixin:
                     manifest, "platform", platform_name, platform_registry, current, previous,
                     finalize=lambda: self._remove_platform_name_if_unowned(platform_name),
                 )
-            logger.debug(
-                "Registered deferred platform loader: %s (plugin=%s)", platform_name, lookup_key,
-            )
+            logger.debug("Registered deferred platform loader: %s (plugin=%s)", platform_name, lookup_key)
         except Exception:
             # Fall back to eager loading so the platform is never silently lost.
             logger.debug(
-                "Deferred platform registration failed for '%s'; eager-loading", lookup_key,
-                exc_info=True,
-            )
+                "Deferred platform registration failed for '%s'; eager-loading", lookup_key, exc_info=True)
             self._load_plugin(manifest)
             return
         self._register_deferred_platform_tools(manifest, loaded)
 
-    def _register_deferred_platform_tools(
-        self, manifest: PluginManifest, loaded: LoadedPlugin
-    ) -> None:
-        """Register a deferred platform's *client* tools without its adapter. Deferring the plugin
-        would otherwise defer its outbound tools too, so CLI/TUI processes (which never materialize
-        platforms) would miss them in ``hermes tools`` / ``platform_toolsets``. Opt-in is explicit via
-        ``provides_tools``; tools live in a ``tools`` submodule so ``__init__`` stays import-light."""
+    def _register_deferred_platform_tools(self, manifest: PluginManifest, loaded: LoadedPlugin) -> None:
+        """Register a deferred platform's *client* tools without its adapter. Deferring the plugin would
+        otherwise defer its outbound tools too, so CLI/TUI processes (which never materialize platforms)
+        would miss them in ``hermes tools`` / ``platform_toolsets``. Opt-in is explicit via ``provides_tools``;
+        tools live in a ``tools`` submodule so ``__init__`` stays import-light."""
         from hermes_cli.plugins import PluginContext, _PLUGINS_DEBUG
         if not manifest.provides_tools:
             return
@@ -151,12 +158,12 @@ class PluginLoaderMixin:
 
         try:
             module = self._load_directory_module(manifest)
-            # Record the module even if nothing registers: the package body has run, so
-            # materializing the adapter later must reuse it rather than execute it twice.
+            # Record the module even if nothing registers: the package body has run, so materializing the
+            # adapter later must reuse it rather than execute it twice.
             loaded.module = module
             self._predeclared_modules[lookup_key] = module
-            register_tools = getattr(importlib.import_module(f"{module.__name__}.tools"),
-                                     "register_tools", None)
+            tools_module = importlib.import_module(f"{module.__name__}.tools")
+            register_tools = getattr(tools_module, "register_tools", None)
             if register_tools is None:
                 logger.warning(
                     "Plugin '%s' declares provides_tools %s but its tools.py "
@@ -167,26 +174,24 @@ class PluginLoaderMixin:
             register_tools(PluginContext(manifest, self))
             registered = _credit()
             logger.debug(
-                "Deferred platform '%s': pre-registered %d client tool(s) %s", lookup_key,
-                len(registered), registered,
+                "Deferred platform '%s': pre-registered %d client tool(s) %s", lookup_key, len(registered),
+                registered,
             )
         except Exception as exc:
-            # Tools registered before the raise are live: credit them or `hermes plugins list`
-            # under-reports (and _load_plugin's later diff would miss them too). Never break
-            # discovery (the platform stays deferred), but a broken tools.py IS the symptom, so warn
-            # — and say where it failed, which is what the operator needs first.
-            partial = _credit()
-            total = len(declared)
-            if not partial:
-                scope = f"before registering any of its {total} declared tool(s)"
-            elif len(partial) >= total:
-                scope = f"after registering all {total} declared tool(s)"
-            else:
-                scope = f"after registering {len(partial)} of {total} declared tool(s)"
+            # Tools registered before the raise are live: credit them or `hermes plugins list` under-reports
+            # (and _load_plugin's later diff would miss them too). Never break discovery (the platform stays
+            # deferred), but a broken tools.py IS the symptom, so warn — and say where it failed first.
+            partial, total = _credit(), len(declared)
+            complete = len(partial) >= total
+            scope = (
+                f"before registering any of its {total} declared tool(s)" if not partial
+                else f"after registering all {total} declared tool(s)" if complete
+                else f"after registering {len(partial)} of {total} declared tool(s)"
+            )
             logger.warning(
-                "Plugin '%s': client-tool pre-registration failed %s (%s).%s", lookup_key, scope,
-                exc, "" if len(partial) >= total else
-                " The remainder will be missing from CLI/TUI sessions.", exc_info=_PLUGINS_DEBUG,
+                "Plugin '%s': client-tool pre-registration failed %s (%s).%s", lookup_key, scope, exc,
+                "" if complete else " The remainder will be missing from CLI/TUI sessions.",
+                exc_info=_PLUGINS_DEBUG,
             )
 
     def _warn_python_dependencies(self, manifest: PluginManifest) -> None:
@@ -195,18 +200,7 @@ class PluginLoaderMixin:
         if not deps:
             return
         key = manifest_key(manifest)
-        missing: List[str] = []
-        for req in deps:
-            # Best-effort presence probe on the distribution name.
-            dist = re.split(r"[<>=!~\[;\s]", req, maxsplit=1)[0].strip()
-            if not dist:
-                continue
-            try:
-                importlib.metadata.version(dist)
-            except importlib.metadata.PackageNotFoundError:
-                missing.append(req)
-            except Exception:
-                continue
+        missing = [req for req in deps if _dist_installed(req) is False]
         if missing:
             logger.warning(
                 "Plugin %s declares Python dependencies that are not "
@@ -258,11 +252,10 @@ class PluginLoaderMixin:
         try:
             # Reuse a deferred platform's already-imported package so its body doesn't run twice.
             module = self._predeclared_modules.pop(plugin_key, None)
-            if module is None:
-                if manifest.source in {"user", "project", "bundled"}:
-                    module = self._load_directory_module(manifest, module_name=module_name)
-                else:
-                    module = self._load_entrypoint_module(manifest)
+            if module is None and manifest.source in {"user", "project", "bundled"}:
+                module = self._load_directory_module(manifest, module_name=module_name)
+            elif module is None:
+                module = self._load_entrypoint_module(manifest)
             loaded.module = module
             register_fn = getattr(module, "register", None)
             if register_fn is None:
@@ -277,14 +270,12 @@ class PluginLoaderMixin:
             self._dispose_registrations(owned)
             self._forget_registrations(owned)
             loaded.error = str(exc)
-            # register() may have subscribed before raising; a failed plugin must leave no callable
-            # reachable from later event dispatch.
+            # register() may have subscribed before raising; a failed plugin must leave no callable reachable
+            # from later event dispatch.
             self._remove_plugin_subscriptions(plugin_key)
-            logger.warning(
-                "Failed to load plugin '%s': %s", manifest.name, exc, exc_info=_PLUGINS_DEBUG,
-            )
-        # The failure path swept this plugin's whole ledger (not just the registration_start slice),
-        # so discovery-time pre-registrations are gone too.
+            logger.warning("Failed to load plugin '%s': %s", manifest.name, exc, exc_info=_PLUGINS_DEBUG)
+        # The failure path swept this plugin's whole ledger (not just the registration_start slice), so
+        # discovery-time pre-registrations are gone too.
         if not loaded.enabled:
             self._predeclared_tools.pop(plugin_key, None)
         self._plugins[plugin_key] = loaded
@@ -306,9 +297,7 @@ class PluginLoaderMixin:
                     module_name, current_policy, replacement, scope=scope,
                 ),
             )
-            self._track_registration(
-                manifest, "tool_override_policy", module_name, policy_lease.dispose,
-            )
+            self._track_registration(manifest, "tool_override_policy", module_name, policy_lease.dispose)
 
     def _attribute_registrations(
         self, loaded: LoadedPlugin, plugin_key: str, registration_start: int
@@ -322,11 +311,9 @@ class PluginLoaderMixin:
         def _keys(kind: str) -> List[str]:
             return [r.key for r in registrations if r.kind == kind]
 
-        # Discovery-time tools predate registration_start; credit them back or `hermes plugins
-        # list` under-reports once the deferred adapter materializes.
-        predeclared = [
-            t for t in self._predeclared_tools.pop(plugin_key, []) if t in self._plugin_tool_names
-        ]
+        # Discovery-time tools predate registration_start; credit them back or `hermes plugins list`
+        # under-reports once the deferred adapter materializes.
+        predeclared = [t for t in self._predeclared_tools.pop(plugin_key, []) if t in self._plugin_tool_names]
         loaded.tools_registered = predeclared + [k for k in _keys("tool") if k not in predeclared]
         loaded.hooks_registered = _keys("hook")
         loaded.middleware_registered = _keys("middleware")
@@ -345,28 +332,19 @@ class PluginLoaderMixin:
         try:
             from hermes_cli.agent_plugins import load_agent_plugin
             package = load_agent_plugin(
-                Path(manifest.path), get_hermes_home() / "plugin-data" / manifest.skill_namespace,
-            )
+                Path(manifest.path), get_hermes_home() / "plugin-data" / manifest.skill_namespace)
             ctx = PluginContext(manifest, self)
             for diagnostic in package.diagnostics:
-                logger.warning(
-                    "Agent Plugin '%s' [%s]: %s", lookup_key, diagnostic.scope, diagnostic.message,
-                )
+                logger.warning("Agent Plugin '%s' [%s]: %s", lookup_key, diagnostic.scope, diagnostic.message)
             for skill in package.skills:
                 try:
-                    ctx.register_skill(
-                        skill.name, skill.skill_md, skill.description, skill.frontmatter,
-                    )
+                    ctx.register_skill(skill.name, skill.skill_md, skill.description, skill.frontmatter)
                 except Exception as exc:
-                    logger.warning(
-                        "Agent Plugin '%s' skill '%s' skipped: %s", lookup_key, skill.name, exc,
-                    )
+                    logger.warning("Agent Plugin '%s' skill '%s' skipped: %s", lookup_key, skill.name, exc)
             for server_name, config in package.mcp_servers.items():
                 internal_name = f"{manifest.skill_namespace}__{server_name}"
                 if internal_name in self._portable_mcp_servers:
-                    logger.warning(
-                        "Agent Plugin '%s' MCP server collision: %s", lookup_key, internal_name,
-                    )
+                    logger.warning("Agent Plugin '%s' MCP server collision: %s", lookup_key, internal_name)
                     continue
                 self._portable_mcp_servers[internal_name] = dict(config)
             loaded.enabled = True
@@ -376,13 +354,12 @@ class PluginLoaderMixin:
         self._plugins[lookup_key] = loaded
 
     def _directory_module_name(self, manifest: PluginManifest) -> str:
-        """Profile-safe import namespace for a directory plugin: the bare ``hermes_plugins.<slug>``
-        for the first scope that claims it, a ``__home_<digest>`` suffix for any other scope."""
+        """Profile-safe import namespace for a directory plugin: the bare ``hermes_plugins.<slug>`` for the
+        first scope that claims it, a ``__home_<digest>`` suffix for any other scope."""
         slug = manifest_key(manifest).replace("/", "__").replace("-", "_")
         bare_name = f"{_NS_PARENT}.{slug}"
         with _MODULE_NAMESPACE_LOCK:
-            owner = _BARE_MODULE_SCOPE.setdefault(bare_name, self.scope_key)
-            if owner == self.scope_key:
+            if _BARE_MODULE_SCOPE.setdefault(bare_name, self.scope_key) == self.scope_key:
                 return bare_name
             digest = hashlib.sha256(self.scope_key.encode("utf-8")).hexdigest()[:12]
             return f"{bare_name}__home_{digest}"
@@ -410,14 +387,13 @@ class PluginLoaderMixin:
             ns_pkg.__package__ = _NS_PARENT
             sys.modules[_NS_PARENT] = ns_pkg
         module_name = module_name or self._directory_module_name(manifest)
-        # Evict stale entries for this slug (same slug cached from another Hermes home, or an
-        # earlier force reload). Replacing only sys.modules[module_name] is not enough: the plugin's
-        # relative imports are cached as "module_name.sub" and resolve from sys.modules first, so a
-        # stale submodule would keep serving the previous load's code/state.
+        # Evict stale entries for this slug (same slug cached from another Hermes home, or an earlier force
+        # reload). Replacing only sys.modules[module_name] is not enough: the plugin's relative imports are
+        # cached as "module_name.sub" and resolve from sys.modules first, so a stale submodule would keep
+        # serving the previous load's code/state.
         _evict_modules(module_name)
         spec = importlib.util.spec_from_file_location(
-            module_name, init_file, submodule_search_locations=[str(plugin_dir)],
-        )
+            module_name, init_file, submodule_search_locations=[str(plugin_dir)])
         if spec is None or spec.loader is None:
             raise ImportError(f"Cannot create module spec for {init_file}")
         module = importlib.util.module_from_spec(spec)
@@ -427,8 +403,8 @@ class PluginLoaderMixin:
         try:
             spec.loader.exec_module(module)
         except BaseException:
-            # Don't leave a half-initialized module (or its partially imported relative submodules)
-            # cached — a retry or a same-slug plugin in another profile would inherit broken state.
+            # Don't leave a half-initialized module (or its partially imported relative submodules) cached — a
+            # retry or a same-slug plugin in another profile would inherit broken state.
             _evict_modules(module_name)
             raise
         return module
@@ -438,6 +414,4 @@ class PluginLoaderMixin:
         for ep in _select_entry_point_group(importlib.metadata.entry_points(), ENTRY_POINTS_GROUP):
             if ep.name == manifest.name:
                 return ep.load()
-        raise ImportError(
-            f"Entry point '{manifest.name}' not found in group '{ENTRY_POINTS_GROUP}'"
-        )
+        raise ImportError(f"Entry point '{manifest.name}' not found in group '{ENTRY_POINTS_GROUP}'")

hermes_cli/plugins_manifest.py

@@ -11,7 +11,7 @@ import logging
 from contextlib import suppress
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any, Dict, List, Mapping, Optional, Set, Union
+from typing import Any, Callable, Dict, List, Mapping, Optional, Set, Union
 
 from utils import fast_safe_load
 from hermes_cli.plugin_capabilities import parse_declared_capabilities as _parse_declared_capabilities
@@ -52,9 +52,8 @@ def _plugins_debug() -> bool:
 
 def _portable_skill_namespace(key: str) -> str:
     """Return a readable, collision-resistant namespace for a portable plugin."""
-    slug = "".join(
-        ch if ch.isascii() and (ch.isalnum() or ch in "_-") else "-" for ch in key.lower()
-    ).strip("-_") or "plugin"
+    slug = "".join(ch if ch.isascii() and (ch.isalnum() or ch in "_-") else "-" for ch in key.lower())
+    slug = slug.strip("-_") or "plugin"
     digest = hashlib.sha256(key.encode("utf-8")).hexdigest()[:8]
     return f"agent-plugin-{slug}-{digest}"
 
@@ -75,70 +74,68 @@ def _manifest_field_of_type(data: Mapping, key: str, field_name: str, typ, what:
     return raw
 
 
+def _manifest_list(data: Mapping, key: str, field_name: str, what: str, coerce: Callable, warn: str) -> list:
+    """Coerce each item of list field ``field_name``; ``coerce`` returning None warns ``warn`` and skips."""
+    out = []
+    for item in _manifest_field_of_type(data, key, field_name, list, what) or []:
+        coerced = coerce(item)
+        if coerced is None:
+            logger.warning(warn, key, item)
+        else:
+            out.append(coerced)
+    return out
+
+
+def _dependency_entry(item: object) -> Optional[Dict[str, Any]]:
+    """``{id, version_range}`` from a requires_plugins item (str shorthand ok); None when malformed."""
+    if isinstance(item, str):
+        return {"id": item, "version_range": None}
+    if isinstance(item, Mapping) and isinstance(item.get("id"), str) and item["id"]:
+        vr = item.get("version_range")
+        return {"id": item["id"], "version_range": str(vr) if vr is not None else None}
+    return None
+
+
+def _manifest_int(raw: object, key: str, warn: str, fallback: Optional[int]) -> Optional[int]:
+    """``int(raw)``; a non-integer warns ``warn`` (formatted with key, raw) and yields ``fallback``."""
+    try:
+        return int(raw)  # type: ignore[call-overload]
+    except (TypeError, ValueError):
+        logger.warning(warn, key, raw)
+        return fallback
+
+
 def _parse_manifest_v2_fields(data: Mapping, key: str) -> Dict[str, Any]:
     """Validate/normalize manifest v2 fields into PluginManifest kwargs (warnings, never failures)."""
-    out: Dict[str, Any] = {}
-    # manifest_version — absent means v1 (supported forever).
-    raw_mv = data.get("manifest_version", 1)
-    try:
-        mv = int(raw_mv)
-    except (TypeError, ValueError):
-        logger.warning(
-            "Plugin %s: manifest_version %r is not an integer; treating as 1", key, raw_mv,
-        )
-        mv = 1
+    # manifest_version — absent means v1 (supported forever); api_version is the independent API generation.
+    mv = _manifest_int(data.get("manifest_version", 1), key,
+                       "Plugin %s: manifest_version %r is not an integer; treating as 1", 1)
     if mv > SUPPORTED_MANIFEST_VERSION:
         logger.warning(
             "Plugin %s: manifest_version %d is newer than this Hermes "
-            "supports (%d); loading anyway and ignoring unknown fields",
-            key, mv, SUPPORTED_MANIFEST_VERSION,
+            "supports (%d); loading anyway and ignoring unknown fields", key, mv, SUPPORTED_MANIFEST_VERSION,
         )
-    out["manifest_version"] = mv
-    # api_version — plugin API generation (independent of manifest_version).
     raw_api = data.get("api_version")
-    out["api_version"] = None
-    if raw_api is not None:
-        try:
-            out["api_version"] = int(raw_api)
-        except (TypeError, ValueError):
-            logger.warning("Plugin %s: api_version %r is not an integer; ignoring", key, raw_api)
-    # requires_plugins — list of {id, version_range?} (str shorthand ok).
-    deps: List[Dict[str, Any]] = []
-    for item in _manifest_field_of_type(data, key, "requires_plugins", list, "a list") or []:
-        if isinstance(item, str):
-            deps.append({"id": item, "version_range": None})
-        elif isinstance(item, Mapping) and isinstance(item.get("id"), str) and item["id"]:
-            vr = item.get("version_range")
-            deps.append({"id": item["id"], "version_range": str(vr) if vr is not None else None})
-        else:
-            logger.warning(
-                "Plugin %s: requires_plugins entry %r must be a plugin id "
-                "string or a {id, version_range} mapping; skipping", key, item,
-            )
-    out["requires_plugins"] = deps
-    # python_dependencies — validated and surfaced ONLY; never auto-installed.
-    pydeps: List[str] = []
-    raw_pydeps = _manifest_field_of_type(
-        data, key, "python_dependencies", list, "a list of requirement strings"
+    api = None if raw_api is None else _manifest_int(
+        raw_api, key, "Plugin %s: api_version %r is not an integer; ignoring", None)
+    deps = _manifest_list(
+        data, key, "requires_plugins", "a list", _dependency_entry,
+        "Plugin %s: requires_plugins entry %r must be a plugin id "
+        "string or a {id, version_range} mapping; skipping",
+    )
+    # python_dependencies — validated and surfaced ONLY; never auto-installed.
+    pydeps = _manifest_list(
+        data, key, "python_dependencies", "a list of requirement strings",
+        lambda item: item.strip() if isinstance(item, str) and item.strip() else None,
+        "Plugin %s: python_dependencies entry %r must be a non-empty requirement string; skipping",
     )
-    for item in raw_pydeps or []:
-        if isinstance(item, str) and item.strip():
-            pydeps.append(item.strip())
-        else:
-            logger.warning(
-                "Plugin %s: python_dependencies entry %r must be a non-empty "
-                "requirement string; skipping", key, item,
-            )
-    out["python_dependencies"] = pydeps
     # config_schema — mapping of key -> {type?, default?, description?, required?}.
     schema: Dict[str, Any] = {}
     raw_schema = _manifest_field_of_type(data, key, "config_schema", Mapping, "a mapping")
     for skey, spec in (raw_schema or {}).items():
         if not isinstance(spec, Mapping):
             logger.warning(
-                "Plugin %s: config_schema entry %r must be a mapping "
-                "(e.g. {type: str}); skipping", key, skey,
-            )
+                "Plugin %s: config_schema entry %r must be a mapping (e.g. {type: str}); skipping", key, skey)
             continue
         stype = spec.get("type")
         if stype is not None and str(stype).lower() not in _CONFIG_SCHEMA_TYPES:
@@ -148,20 +145,19 @@ def _parse_manifest_v2_fields(data: Mapping, key: str) -> Dict[str, Any]:
                 key, skey, stype, ", ".join(sorted(_CONFIG_SCHEMA_TYPES)),
             )
         schema[str(skey)] = dict(spec)
-    out["config_schema"] = schema
-    out["license"] = str(data.get("license") or "")
-    out["homepage"] = str(data.get("homepage") or "")
-    raw_tags = _manifest_field_of_type(data, key, "tags", list, "a list")
-    out["tags"] = [str(t) for t in (raw_tags or [])]
+    tags = [str(t) for t in (_manifest_field_of_type(data, key, "tags", list, "a list") or [])]
     # Forward compat: unknown fields warn (never fail); v1 manifests only at debug.
     unknown = sorted(set(data.keys()) - _KNOWN_MANIFEST_FIELDS)
     if unknown:
-        log = logger.warning if mv >= 2 else logger.debug
-        log(
+        (logger.warning if mv >= 2 else logger.debug)(
             "Plugin %s: unknown manifest field(s) ignored: %s "
             "(newer manifest schema or typo; plugin still loads)", key, ", ".join(unknown),
         )
-    return out
+    return {
+        "manifest_version": mv, "api_version": api, "requires_plugins": deps, "python_dependencies": pydeps,
+        "config_schema": schema, "license": str(data.get("license") or ""),
+        "homepage": str(data.get("homepage") or ""), "tags": tags,
+    }
 
 
 def validate_config_schema(plugin_id: str, schema: Mapping, settings: Mapping) -> List[str]:
@@ -194,11 +190,9 @@ def validate_config_schema(plugin_id: str, schema: Mapping, settings: Mapping) -
 
 
 def resolve_plugin_load_order(manifests: Mapping[str, "PluginManifest"]) -> List[str]:
-    """Return plugin keys in dependency order: B before A when A requires B; alphabetical ties.
-
-    A cycle warns and falls back to alphabetical order for all; a missing dependency warns once
-    but never removes the dependent plugin (loads never hard-fail on advisory deps).
-    """
+    """Return plugin keys in dependency order: B before A when A requires B; alphabetical ties. A cycle warns
+    and falls back to alphabetical order for all; a missing dependency warns once but never removes the
+    dependent plugin (loads never hard-fail on advisory deps)."""
     import graphlib
     keys = sorted(manifests.keys())
     by_name: Dict[str, str] = {}
@@ -242,9 +236,9 @@ def resolve_plugin_load_order(manifests: Mapping[str, "PluginManifest"]) -> List
 
 
 def _detect_kind_from_source(source_text: str) -> Optional[str]:
-    """Kind implied by source markers (mirrors plugins/memory ``_is_memory_provider_dir``):
-    memory-provider markers -> ``exclusive``; ``register_provider`` + ``ProviderProfile`` ->
-    ``model-provider``; else ``None``. Keeps both kinds out of the general manager's eager import."""
+    """Kind implied by source markers (mirrors plugins/memory ``_is_memory_provider_dir``): memory-provider
+    markers -> ``exclusive``; ``register_provider`` + ``ProviderProfile`` -> ``model-provider``; else
+    ``None``. Keeps both kinds out of the general manager's eager import."""
     if "register_memory_provider" in source_text or "MemoryProvider" in source_text:
         return "exclusive"
     if "register_provider" in source_text and "ProviderProfile" in source_text:
@@ -254,12 +248,10 @@ def _detect_kind_from_source(source_text: str) -> Optional[str]:
 
 def _read_source_from_origin(origin: Optional[str], limit: int = 8192) -> str:
     """First ``limit`` chars of a module's source (``.pyc`` mapped back to ``.py``); "" on failure."""
-    if not origin:
-        return ""
     try:
-        if origin.endswith((".pyc", ".pyo")):
+        if origin and origin.endswith((".pyc", ".pyo")):
             origin = importlib.util.source_from_cache(origin)
-        if not origin.endswith(".py"):
+        if not origin or not origin.endswith(".py"):
             return ""
         return Path(origin).read_text(encoding="utf-8", errors="replace")[:limit]
     except Exception:
@@ -267,12 +259,10 @@ def _read_source_from_origin(origin: Optional[str], limit: int = 8192) -> str:
 
 
 def resolve_module_origin(module_name: str) -> Optional[str]:
-    """Return a module's source path WITHOUT importing it, or ``None``.
-
-    ``find_spec`` on a dotted name imports the parent package, so only the top-level name uses it;
-    remaining segments are walked through ``submodule_search_locations`` by hand. Namespace/zipped/
-    extension modules return ``None``. Shared with ``plugins/memory/__init__.py``.
-    """
+    """Return a module's source path WITHOUT importing it, or ``None``. ``find_spec`` on a dotted name imports
+    the parent package, so only the top-level name uses it; remaining segments are walked through
+    ``submodule_search_locations`` by hand. Namespace/zipped/extension modules return ``None``. Shared with
+    ``plugins/memory/__init__.py``."""
     parts = [p for p in module_name.split(".") if p]
     if not parts:
         return None
@@ -286,16 +276,12 @@ def resolve_module_origin(module_name: str) -> Optional[str]:
         if not search_paths:
             return None
         for i, part in enumerate(parts[1:], start=2):
-            found_origin = None
-            next_paths = None
-            for base in search_paths:
-                base = Path(base)
-                pkg_init = base / part / "__init__.py"
+            found_origin = next_paths = None
+            for base in map(Path, search_paths):
+                pkg_init, mod_file = base / part / "__init__.py", base / (part + ".py")
                 if pkg_init.is_file():
-                    found_origin = str(pkg_init)
-                    next_paths = [base / part]
+                    found_origin, next_paths = str(pkg_init), [base / part]
                     break
-                mod_file = base / (part + ".py")
                 if mod_file.is_file():
                     found_origin = str(mod_file)
                     break
@@ -332,21 +318,21 @@ class PluginManifest:
     provides_hooks: List[str] = field(default_factory=list)
     source: str = ""        # "bundled", "user", "project", or "entrypoint"
     path: Optional[str] = None
-    # ``standalone`` (default; opt-in via plugins.enabled) | ``backend`` (pluggable backend for a
-    # core tool; bundled auto-load, user-installed gated) | ``exclusive`` (one active provider,
-    # selected via <category>.provider; own discovery, general scanner skips) | ``platform``
-    # (gateway adapter; bundled auto-load, user-installed gated as untrusted code).
+    # ``standalone`` (default; opt-in via plugins.enabled) | ``backend`` (pluggable backend for a core tool;
+    # bundled auto-load, user-installed gated) | ``exclusive`` (one active provider, selected via
+    # <category>.provider; own discovery, general scanner skips) | ``platform`` (gateway adapter; bundled
+    # auto-load, user-installed gated as untrusted code).
     kind: str = "standalone"
-    # Path-derived registry key used by plugins.enabled/disabled and `hermes plugins list`:
-    # ``disk-cleanup`` for a flat plugin, ``image_gen/openai`` for a category plugin. Empty -> name.
+    # Path-derived registry key used by plugins.enabled/disabled and `hermes plugins list`: ``disk-cleanup``
+    # for a flat plugin, ``image_gen/openai`` for a category plugin. Empty -> name.
     key: str = ""
     portable: bool = False
     skill_namespace: str = ""
-    # Declared capability ids, normalized to KNOWN ids. Declaration is consent metadata, NOT a
-    # grant: live only via plugins.entries.<id>.granted_capabilities or the legacy allow_* key.
+    # Declared capability ids, normalized to KNOWN ids. Declaration is consent metadata, NOT a grant: live
+    # only via plugins.entries.<id>.granted_capabilities or the legacy allow_* key.
     capabilities: List[str] = field(default_factory=list)
-    # Manifest v2 fields — all optional and additive. manifest_version versions the FILE FORMAT
-    # (v1 supported forever); api_version is the runtime plugin API generation (None = current).
+    # Manifest v2 fields — all optional and additive. manifest_version versions the FILE FORMAT (v1 supported
+    # forever); api_version is the runtime plugin API generation (None = current).
     manifest_version: int = 1
     api_version: Optional[int] = None
     # Advisory deps [{"id", "version_range"}]: missing ones warn but load; they order the load.
@@ -379,28 +365,24 @@ def portable_plugin_manifest(child: Path, source: str, prefix: str) -> PluginMan
 
 
 def _manifest_kind(data: Mapping, key: str, plugin_dir: Path) -> str:
-    """Normalize ``kind``; undeclared memory/model providers are auto-detected from ``__init__.py``
-    so they route to their own discovery instead of the general manager."""
+    """Normalize ``kind``; undeclared memory/model providers are auto-detected from ``__init__.py`` so they
+    route to their own discovery instead of the general manager."""
     raw_kind = data.get("kind", "standalone")
-    if not isinstance(raw_kind, str):
-        raw_kind = "standalone"
-    kind = raw_kind.strip().lower()
+    kind = raw_kind.strip().lower() if isinstance(raw_kind, str) else "standalone"
     if kind not in _VALID_PLUGIN_KINDS:
         logger.warning(
             "Plugin %s: unknown kind '%s' (valid: %s); treating as 'standalone'",
             key, raw_kind, ", ".join(sorted(_VALID_PLUGIN_KINDS)),
         )
         kind = "standalone"
-    if kind == "standalone" and "kind" not in data:
-        init_file = plugin_dir / "__init__.py"
-        if init_file.exists():
-            with suppress(Exception):
-                detected = _detect_kind_from_source(
-                    init_file.read_text(errors="replace", encoding="utf-8")[:8192]
-                )
-                if detected:
-                    kind = detected
-                    logger.debug("Plugin %s: detected %s, treating as kind='%s'", key, detected, detected)
+    init_file = plugin_dir / "__init__.py"
+    if kind == "standalone" and "kind" not in data and init_file.exists():
+        with suppress(Exception):
+            source_text = init_file.read_text(errors="replace", encoding="utf-8")[:8192]
+            detected = _detect_kind_from_source(source_text)
+            if detected:
+                kind = detected
+                logger.debug("Plugin %s: detected %s, treating as kind='%s'", key, detected, detected)
     return kind
 
 
@@ -417,9 +399,7 @@ def parse_manifest_file(
         key = f"{prefix}/{plugin_dir.name}" if prefix else name
         kind = _manifest_kind(data, key, plugin_dir)
         logger.debug(
-            "Parsed manifest: key=%s name=%s kind=%s source=%s path=%s",
-            key, name, kind, source, plugin_dir,
-        )
+            "Parsed manifest: key=%s name=%s kind=%s source=%s path=%s", key, name, kind, source, plugin_dir)
         return PluginManifest(
             name=name, version=str(data.get("version", "")),
             description=data.get("description", ""), author=_display_author(data.get("author", "")),

hermes_cli/plugins_state.py

@@ -22,16 +22,13 @@ _PLUGIN_STATE_LOCKS_GUARD = threading.Lock()
 
 
 def _plugin_relative_segments(key: str) -> tuple[str, ...]:
-    """Validate/split a plugin-relative settings key; global paths, traversal, and core roots are
-    rejected before any config read."""
+    """Validate/split a plugin-relative settings key; global paths, traversal, and core roots are rejected
+    before any config read."""
     if not isinstance(key, str):
         raise ValueError("Expected a plugin-relative config key string")
     segments = tuple(key.split("."))
-    if (
-        not key or "/" in key or "\\" in key
-        or not all(_PLUGIN_SETTING_SEGMENT_RE.fullmatch(segment) for segment in segments)
-        or segments[0].lower() in _PLUGIN_SETTING_RESERVED_ROOTS
-    ):
+    invalid = not key or "/" in key or "\\" in key or segments[0].lower() in _PLUGIN_SETTING_RESERVED_ROOTS
+    if invalid or not all(_PLUGIN_SETTING_SEGMENT_RE.fullmatch(segment) for segment in segments):
         raise ValueError(
             "Expected a plugin-relative config key such as 'endpoint' or "
             "'retry.policy'; global, cross-plugin, and traversal paths are forbidden"
@@ -64,27 +61,23 @@ def _plugin_settings_entry(config: object, plugin_id: str) -> Mapping[str, Any]
 
 
 def _plugin_data_namespace(plugin_id: str, skill_namespace: str) -> str:
-    """Return one Windows-safe directory component for plugin-owned data."""
+    """Return one Windows-safe directory component for plugin-owned data. Portable Agent Plugins already
+    receive this exact PLUGIN_DATA path; otherwise the fixed prefix avoids Windows reserved device names and
+    the digest prevents fold collisions."""
     candidate = skill_namespace or plugin_id
-    if (
-        skill_namespace and candidate.startswith("agent-plugin-")
-        and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,191}", candidate)
-    ):
-        # Portable Agent Plugins already receive this exact PLUGIN_DATA path.
+    portable = skill_namespace and candidate.startswith("agent-plugin-")
+    if portable and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,191}", candidate):
         return candidate
-    # Fixed prefix avoids Windows reserved device names; digest prevents fold collisions.
     return _portable_skill_namespace(candidate)
 
 
 @contextmanager
 def _locked_plugin_state(path: Path):
-    """Serialize state read-modify-write across threads/processes (fcntl / msvcrt). The lock lives
-    in a sibling file because atomic replacement changes the target's inode."""
+    """Serialize state read-modify-write across threads/processes (fcntl / msvcrt). The lock lives in a
+    sibling file because atomic replacement changes the target's inode."""
     lock_path = path.with_name(f".{path.name}.lock")
     with _PLUGIN_STATE_LOCKS_GUARD:
-        thread_lock = _PLUGIN_STATE_LOCKS.setdefault(
-            str(lock_path.resolve(strict=False)), threading.RLock()
-        )
+        thread_lock = _PLUGIN_STATE_LOCKS.setdefault(str(lock_path.resolve(strict=False)), threading.RLock())
     with thread_lock:
         lock_path.parent.mkdir(parents=True, exist_ok=True)
         with open(lock_path, "a+b") as handle:
@@ -162,9 +155,7 @@ class PluginState:
             try:
                 encoded = json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8")
             except (TypeError, ValueError) as exc:
-                raise ValueError(
-                    f"Plugin state value for {key!r} is not JSON-serializable"
-                ) from exc
+                raise ValueError(f"Plugin state value for {key!r} is not JSON-serializable") from exc
             if len(encoded) > self.quota_bytes:
                 raise ValueError(
                     f"Plugin state quota exceeded: {len(encoded)} bytes is greater "