refactor(plugins): own dual-kind hook fallback in the ledger mixin

Move the general-vs-memory hook ownership logic out of the memory collector into
PluginLedgerMixin (_drop_fallback_hooks / _register_fallback_hook) so the collector
and the loader each call one manager method instead of reaching into manager privates.
Hoist hashlib to module scope. Trim the new suite to the three invariant cases
(run-once across load orders, distinct sources not suppressed, re-exported register).

hermes_cli/plugins_ledger.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 
 import logging
 from dataclasses import dataclass, field
+from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Set, Union
 
 from registration_lifecycle import replacement_coordinator
@@ -18,6 +19,12 @@ if TYPE_CHECKING:  # pragma: no cover
 logger = logging.getLogger("hermes_cli.plugins")
 
 
+def _hook_source_of(name: str, module: Any) -> Optional[tuple]:
+    """(plugin name, resolved ``__file__``) identity shared by the general and memory loaders."""
+    source = getattr(module, "__file__", None)
+    return (name, str(Path(source).resolve())) if source else None
+
+
 @dataclass
 class PluginRegistration:
     """One host-owned registration plus its inverse, so force reload unwinds registries in reverse order
@@ -169,6 +176,30 @@ class PluginLedgerMixin:
                 else:
                     ledger.pop(plugin_key, None)
 
+    # -- dual-kind hook ownership -------------------------------------------------------------
+    # A plugin dir loaded by general discovery AND as the configured memory provider calls
+    # ``register()`` twice against the same manager. General discovery owns the hook group; the
+    # memory loader's hooks are a fallback group keyed by (name, resolved source path) that is
+    # dropped once the same source loads through discovery. Groups, not callbacks, are replaced:
+    # one register() may deliberately create several closures for one hook.
+
+    def _drop_fallback_hooks(self, hook_source: Optional[tuple]) -> None:
+        if hook_source is not None:
+            for handle in self._memory_hook_registrations.pop(hook_source, []):
+                handle.dispose()
+
+    def _register_fallback_hook(self, context: Any, hook_source: Optional[tuple], hook_name: str,
+                                callback: Callable) -> PluginRegistration:
+        """Register ``callback`` unless the same source is already live through general discovery, in
+        which case return an inert handle so the provider keeps its disposable-handle contract."""
+        if hook_source is None:
+            return context.register_hook(hook_name, callback)
+        owned = any(loaded.enabled and _hook_source_of(loaded.manifest.name, loaded.module) == hook_source
+                    for loaded in self._plugins.values())
+        handle = context._track("hook", hook_name, lambda: None) if owned else context.register_hook(hook_name, callback)
+        self._memory_hook_registrations.setdefault(hook_source, []).append(handle)
+        return handle
+
     def _dispose_registrations(self, registrations: List[PluginRegistration]) -> None:
         """Dispose registrations in reverse acquisition order, best effort."""
         from hermes_cli.plugins import _PLUGINS_DEBUG

hermes_cli/plugins_loader.py

@@ -307,11 +307,9 @@ class PluginLoaderMixin:
                 register_fn(PluginContext(manifest, self))
                 self._attribute_registrations(loaded, plugin_key, registration_start)
                 loaded.enabled = True
-                source = getattr(module, "__file__", None)
-                if source:
-                    hook_source = (manifest.name, str(Path(source).resolve()))
-                    for handle in self._memory_hook_registrations.pop(hook_source, []):
-                        handle.dispose()
+                from hermes_cli.plugins_ledger import _hook_source_of
+
+                self._drop_fallback_hooks(_hook_source_of(manifest.name, module))
         except Exception as exc:
             owned = [r for r in self._registration_order if r.plugin_key == plugin_key]
             self._dispose_registrations(owned)

plugins/memory/__init__.py

@@ -9,11 +9,13 @@ must never shadow a shipped provider. Changing this order is a breaking change.
 
 from __future__ import annotations
 
+import hashlib
 import importlib.metadata
 import importlib.util
 import logging
 import sys
 from pathlib import Path
+from types import SimpleNamespace
 from typing import List, Optional, Tuple, TYPE_CHECKING
 
 from hermes_cli.config import cfg_get
@@ -70,8 +72,6 @@ def _module_name(provider_dir: Path, name: str) -> str:
     if _is_bundled(provider_dir):
         return f"plugins.memory.{name}"
     # Separate package trees, including relative imports, across homes/sources.
-    import hashlib
-
     digest = hashlib.sha256(str(provider_dir.resolve()).encode()).hexdigest()[:16]
     return f"{_USER_NAMESPACE}.{name}__source_{digest}"
 
@@ -297,40 +297,21 @@ class _ProviderCollector:
         self._hook_source = None
 
     def collect(self, register, *, source=None):
-        """General discovery owns hooks; memory-only activation supplies a fallback.
+        """Run ``register`` with this collector; hooks it registers form the fallback group that
+        general discovery of the same source replaces (see ``PluginLedgerMixin``)."""
+        from hermes_cli.plugins_ledger import _hook_source_of
 
-        Replace a whole registration group, not callbacks by name or identity:
-        a register function may deliberately create several distinct closures.
-        """
         module = sys.modules.get(getattr(register, "__module__", ""))
-        source = source or getattr(module, "__file__", None)
-        if source:
-            self._hook_source = (self.name, str(Path(source).resolve()))
+        self._hook_source = _hook_source_of(self.name, SimpleNamespace(__file__=source) if source else module)
         manager = self._plugin_context()._manager
         with manager._discovery_lock:
-            if self._hook_source is not None:
-                for handle in manager._memory_hook_registrations.pop(self._hook_source, []):
-                    handle.dispose()
+            manager._drop_fallback_hooks(self._hook_source)
             register(self)
 
     def register_hook(self, hook_name, callback):
         context = self._plugin_context()
-        manager = context._manager
-        with manager._discovery_lock:
-            if self._hook_source is not None:
-                for loaded in manager._plugins.values():
-                    source = getattr(loaded.module, "__file__", None)
-                    if (loaded.enabled and source and
-                            (loaded.manifest.name, str(Path(source).resolve())) == self._hook_source):
-                        # Preserve the disposable-handle contract without leasing
-                        # the general plugin's callback to this provider instance.
-                        handle = context._track("hook", hook_name, lambda: None)
-                        manager._memory_hook_registrations.setdefault(self._hook_source, []).append(handle)
-                        return handle
-            handle = context.register_hook(hook_name, callback)
-            if self._hook_source is not None:
-                manager._memory_hook_registrations.setdefault(self._hook_source, []).append(handle)
-            return handle
+        with context._manager._discovery_lock:
+            return context._manager._register_fallback_hook(context, self._hook_source, hook_name, callback)
 
     def register_memory_provider(self, provider):
         self.provider = provider

tests/plugins/test_memory_hook_registration.py

@@ -74,36 +74,6 @@ def test_dual_kind_plugin_hooks_run_once(tmp_path, monkeypatch, order):
         manager.unload()
 
 
-def test_equal_names_in_different_homes_keep_their_own_imports(tmp_path, monkeypatch):
-    managers = []
-    try:
-        for label in ("alpha", "beta"):
-            manager = _install(tmp_path / label, monkeypatch, label=label)
-            managers.append(manager)
-            assert load_memory_provider("dual") is not None
-            manager.discover_and_load()
-            assert _contexts(manager) == [{"context": label}, {"context": "second"}]
-        assert _contexts(managers[0]) == [{"context": "alpha"}, {"context": "second"}]
-        managers[1].unload()
-        assert _contexts(managers[0]) == [{"context": "alpha"}, {"context": "second"}]
-    finally:
-        for manager in managers:
-            manager.unload()
-
-
-def test_failed_general_registration_leaves_no_callable(tmp_path, monkeypatch):
-    manager = _install(tmp_path, monkeypatch)
-    try:
-        assert load_memory_provider("dual") is not None
-        source = tmp_path / "plugins" / "dual" / "__init__.py"
-        source.write_text(source.read_text() + '\n    raise RuntimeError("broken registration")\n')
-        manager.discover_and_load()
-        assert _contexts(manager) == []
-        assert not manager._plugins["dual"].enabled
-    finally:
-        manager.unload()
-
-
 def test_same_name_different_sources_are_not_suppressed(tmp_path, monkeypatch):
     import shutil
 
@@ -126,23 +96,6 @@ def test_same_name_different_sources_are_not_suppressed(tmp_path, monkeypatch):
         manager.unload()
 
 
-@pytest.mark.parametrize("entry_kind", ["module", "function"])
-def test_entry_point_registration_uses_same_hook_ownership(tmp_path, monkeypatch, entry_kind):
-    from types import SimpleNamespace
-    from plugins.memory import _load_provider_from_entry_point
-
-    manager = _install(tmp_path, monkeypatch)
-    try:
-        manager.discover_and_load()
-        module = manager._plugins["dual"].module
-        target = module if entry_kind == "module" else module.register
-        entry = SimpleNamespace(name="dual", load=lambda: target)
-        assert _load_provider_from_entry_point(entry) is not None
-        assert _contexts(manager) == [{"context": "first"}, {"context": "second"}]
-    finally:
-        manager.unload()
-
-
 @pytest.mark.parametrize("order", ["plugin-first", "memory-first"])
 def test_reexported_register_uses_plugin_source(tmp_path, monkeypatch, order):
     manager = _install(tmp_path, monkeypatch)
@@ -159,26 +112,3 @@ def test_reexported_register_uses_plugin_source(tmp_path, monkeypatch, order):
         assert _contexts(manager) == [{"context": "first"}, {"context": "second"}]
     finally:
         manager.unload()
-
-
-def test_suppressed_hook_still_returns_disposable_handle(tmp_path, monkeypatch):
-    manager = _install(tmp_path, monkeypatch)
-    source = tmp_path / "plugins" / "dual" / "__init__.py"
-    source.write_text(source.read_text().split("def register(ctx):")[0] + textwrap.dedent('''\
-        def register(ctx):
-            handle = ctx.register_hook("pre_llm_call", make_hook("discard"))
-            assert handle.active
-            handle.dispose()
-            assert not handle.active
-            provider = Provider()
-            provider.configured = True
-            ctx.register_memory_provider(provider)
-            ctx.register_hook("pre_llm_call", make_hook("kept"))
-    '''))
-    try:
-        manager.discover_and_load()
-        provider = load_memory_provider("dual")
-        assert provider is not None and provider.configured
-        assert _contexts(manager) == [{"context": "kept"}]
-    finally:
-        manager.unload()