fix(plugins): a plugin reload replaces its context reference once it loads

A forced plugin re-discovery (reload) re-runs register(), but the plugin's
@-prefix was still held from its first load, so the new provider was refused
with "prefix 'x' is already registered" and the stale one kept answering.

Context-reference registrations now go through the plugin ownership ledger.
The prefix table remembers which plugin on which manager claimed a prefix.
Unloading keeps the provider registered, as before, and only releases the
claim. When the same plugin on the same manager registers that prefix again,
the new provider is staged on a ledger entry and published only after the
plugin's load succeeds. A load that fails or times out disposes the entry
first, so the working provider and its owner stay exactly as they were.

Everything else matches main: providers stay in one process-wide table shared
by every profile; a second registration of a prefix in the same load, another
plugin, another profile's manager or a second manager are refused with the
same warning; built-in prefixes are refused; both registration calls still
return None when called as before.

agent/context_references.py

@@ -28,6 +28,8 @@ from hermes_cli.sizefmt import format_bytes
 BUILTIN_PREFIXES = frozenset({"diff", "staged", "file", "folder", "git", "url"})
 
 _context_reference_providers: dict[str, ContextReferenceProvider] = {}
+_context_reference_owners: dict[str, object] = {}  # prefix -> owner given at registration (None: none given)
+_context_reference_lock = threading.Lock()
 
 
 @dataclass(slots=True)
@@ -58,8 +60,13 @@ class ContextReferenceProvider(ABC):
         """Expand *target* to prompt content.  Return ``None`` to skip."""
 
 
-def register_context_reference_provider(provider: ContextReferenceProvider) -> None:
-    """Register a plugin context reference provider."""
+def register_context_reference_provider(
+    provider: ContextReferenceProvider, *, owner: object = None, reclaim: bool = False,
+) -> Callable[[], bool] | None:
+    """Register a plugin context reference provider; a taken prefix is refused. ``owner`` records who holds
+    the prefix. With ``reclaim``, a prefix the same ``owner`` holds is not refused but staged: nothing changes
+    until the returned callable publishes ``provider`` (a reload that loaded), which it skips if the prefix has
+    changed hands meanwhile."""
     if not isinstance(provider, ContextReferenceProvider):
         raise TypeError("provider must be a ContextReferenceProvider instance")
     prefix = provider.prefix.lower().strip()
@@ -67,9 +74,23 @@ def register_context_reference_provider(provider: ContextReferenceProvider) -> N
         raise ValueError("prefix must be a non-empty string")
     if prefix in BUILTIN_PREFIXES:
         raise ValueError(f"prefix '{prefix}' is reserved for built-in references")
-    if prefix in _context_reference_providers:
-        raise ValueError(f"prefix '{prefix}' is already registered")
-    _context_reference_providers[prefix] = provider
+    with _context_reference_lock:
+        held = _context_reference_providers.get(prefix)
+        if held is None:
+            _context_reference_providers[prefix] = provider
+            _context_reference_owners[prefix] = owner
+            return None
+        if not reclaim or owner is None or _context_reference_owners.get(prefix) != owner:
+            raise ValueError(f"prefix '{prefix}' is already registered")
+
+    def publish() -> bool:
+        with _context_reference_lock:
+            if _context_reference_providers.get(prefix) is not held or _context_reference_owners.get(prefix) != owner:
+                return False
+            _context_reference_providers[prefix] = provider
+            return True
+
+    return publish
 
 
 def get_context_reference_providers() -> dict[str, ContextReferenceProvider]:

hermes_cli/plugins.py

@@ -716,7 +716,7 @@ class PluginContext(PluginCapabilityCtxMixin):
         if self._wrong_type(provider, _CRP, "context reference provider"):
             return
         try:
-            _register(provider)
+            self._manager._track_context_reference(self.manifest, provider, _register)
         except ValueError as exc:
             logger.warning("Plugin '%s' context reference registration failed: %s", self.manifest.name, exc)
             return

hermes_cli/plugins_ledger.py

@@ -5,6 +5,7 @@ reload / targeted unload unwind registries in reverse order. Mixed into :class:`
 from __future__ import annotations
 
 import logging
+import weakref
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Set, Union
@@ -42,6 +43,9 @@ class PluginRegistration:
     # A ``transform_llm_output`` registration that must see the reply before the user does
     # (``register_hook(..., hold_stream=True)``; read by ``PluginDispatchMixin.holds_stream``).
     hold_stream: bool = False
+    # Staged change applied only once the plugin's load succeeded (``_publish_staged_registrations``); a failed
+    # or timed-out load disposes the handle first, so the change is never applied.
+    publish: Optional[Callable[[], Any]] = None
     _disposed: bool = field(default=False, init=False, repr=False)
     _on_dispose: Optional[Callable[[PluginRegistration], None]] = field(default=None, init=False, repr=False)
 
@@ -96,6 +100,23 @@ class PluginLedgerMixin:
         )
         return self._track_registration(manifest, kind, name, lease.dispose)
 
+    def _track_context_reference(self, manifest: PluginManifest, provider: Any, register: Callable[..., Any]) -> None:
+        """Claim ``provider.prefix`` in the process-wide context-reference table and record the claim. The table
+        keeps the provider when the plugin unloads, so the same plugin on the same manager loading again (a
+        reload) holds the prefix with no live claim: its new provider is staged and published on load success.
+        A live claim (a second registration in the same load), another plugin or another manager is refused."""
+        prefix, plugin_key = provider.prefix.lower().strip(), manifest_key(manifest)
+        reclaim = not any(r.active and r.kind == "context_reference" and r.key == prefix
+                          for r in self._ownership_ledger.get(plugin_key, ()))
+        publish = register(provider, owner=(plugin_key, weakref.ref(self)), reclaim=reclaim)
+        self._track_registration(manifest, "context_reference", prefix, lambda: None).publish = publish
+
+    def _publish_staged_registrations(self, plugin_key: str, start: int) -> None:
+        """Apply the staged changes of a plugin load that succeeded."""
+        for registration in self._registration_order[start:]:
+            if registration.plugin_key == plugin_key and registration.active and registration.publish is not None:
+                registration.publish()
+
     def _active_persistent(self) -> list[PluginRegistration]:
         """Live persistent registrations across every plugin in the ownership ledger."""
         return [r for owned in self._ownership_ledger.values() for r in owned if r.persistent and r.active]

hermes_cli/plugins_loader.py

@@ -623,6 +623,7 @@ class PluginLoaderMixin:
         try:
             if run_with_load_deadline(plugin_key, ctx, _import_and_register):
                 self._attribute_registrations(loaded, plugin_key, registration_start)
+                self._publish_staged_registrations(plugin_key, registration_start)
                 loaded.enabled = True
                 from hermes_cli.plugins_ledger import _hook_source_of
 

tests/hermes_cli/test_plugin_context_reference_reload.py

@@ -0,0 +1,161 @@
+"""Plugin context-reference providers across a plugin reload (force re-discovery), within one load and across
+profiles. Only the successful-reload test is new behavior; the rest pin what already holds on main."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+from pathlib import Path
+
+import hermes_yaml as yaml
+import pytest
+
+from agent.context_references import _context_reference_providers, get_context_reference_providers
+from hermes_cli import plugins as plugins_mod
+from hermes_cli.plugins import PluginManager, _plugin_home_scope
+
+_PROVIDER = '''
+from agent.context_references import ContextReferenceProvider
+
+class Tickets(ContextReferenceProvider):
+    prefix = "tkt"
+
+    def __init__(self, label="{label}"):
+        self.label = label
+
+    async def autocomplete(self, query, *, limit=10):
+        return []
+
+    async def expand(self, target):
+        return self.label + " " + target
+
+def register(ctx):
+    ctx.register_context_reference(Tickets())
+'''
+_REFUSED = "prefix 'tkt' is already registered"
+
+
+def _write_plugin(home: Path, name: str, label: str, extra: str = "") -> None:
+    plugin_dir = home / "plugins" / name
+    plugin_dir.mkdir(parents=True, exist_ok=True)
+    (plugin_dir / "plugin.yaml").write_text(yaml.safe_dump({"name": name, "version": "1.0"}), encoding="utf-8")
+    (plugin_dir / "__init__.py").write_text(_PROVIDER.replace("{label}", label) + extra, encoding="utf-8")
+
+
+def _write_config(home: Path, *names: str) -> None:
+    (home / "config.yaml").write_text(yaml.safe_dump({"plugins": {"enabled": list(names)}}), encoding="utf-8")
+
+
+@pytest.fixture
+def home(tmp_path, monkeypatch):
+    home = Path(os.environ["HERMES_HOME"])
+    empty_bundled = tmp_path / "bundled"
+    empty_bundled.mkdir()
+    monkeypatch.setattr(plugins_mod, "get_bundled_plugins_dir", lambda: empty_bundled)
+    _context_reference_providers.clear()
+    yield home
+    _context_reference_providers.clear()
+
+
+def _expand(target: str) -> str | None:
+    provider = get_context_reference_providers().get("tkt")
+    return None if provider is None else asyncio.run(provider.expand(target))
+
+
+def test_successful_reload_replaces_the_plugins_provider(home):
+    _write_plugin(home, "tickets", "v1")
+    _write_config(home, "tickets")
+    manager = PluginManager()
+    try:
+        manager.discover_and_load()
+        assert _expand("7") == "v1 7"
+
+        _write_plugin(home, "tickets", "v2-reloaded")
+        manager.discover_and_load(force=True)
+        assert manager._plugins["tickets"].error is None
+        assert _expand("7") == "v2-reloaded 7"
+    finally:
+        manager.unload()
+
+
+def test_failed_reload_keeps_the_working_provider(home):
+    _write_plugin(home, "tickets", "working")
+    _write_config(home, "tickets")
+    manager = PluginManager()
+    try:
+        manager.discover_and_load()
+        working = get_context_reference_providers()["tkt"]
+
+        _write_plugin(home, "tickets", "half-loaded", extra="    raise RuntimeError('configuration unavailable')\n")
+        manager.discover_and_load(force=True)
+        assert "configuration unavailable" in manager._plugins["tickets"].error
+        assert get_context_reference_providers()["tkt"] is working
+    finally:
+        manager.unload()
+
+
+def test_duplicate_prefix_in_one_load_keeps_the_first_provider(home, caplog):
+    _write_plugin(home, "tickets", "first", extra="    ctx.register_context_reference(Tickets('second'))\n")
+    _write_config(home, "tickets")
+    manager = PluginManager()
+    try:
+        with caplog.at_level(logging.WARNING):
+            manager.discover_and_load()
+        assert manager._plugins["tickets"].error is None
+        assert _expand("7") == "first 7"
+        assert _REFUSED in caplog.text
+
+        manager.unload("tickets")  # disabling a plugin leaves its provider registered
+        assert _expand("7") == "first 7"
+    finally:
+        manager.unload()
+
+
+def test_another_plugin_cannot_take_a_registered_prefix(home, caplog):
+    _write_plugin(home, "tickets-a", "a")
+    _write_plugin(home, "tickets-b", "b")
+    _write_config(home, "tickets-a", "tickets-b")
+    manager = PluginManager()
+    try:
+        with caplog.at_level(logging.WARNING):
+            manager.discover_and_load()
+        first = get_context_reference_providers()["tkt"]
+        assert _REFUSED in caplog.text
+
+        caplog.clear()
+        with caplog.at_level(logging.WARNING):
+            manager.discover_and_load(force=True)
+        assert _REFUSED in caplog.text
+        assert _expand("1") == asyncio.run(first.expand("1"))
+    finally:
+        manager.unload()
+
+
+def test_profiles_share_the_first_profiles_provider(home, tmp_path, caplog):
+    # One process, one manager per profile (multiplexed gateway): A and C enable the plugin, B does not.
+    # The prefix is process-wide: every profile sees A's provider and C's registration is refused.
+    homes = {}
+    for name in ("A", "B", "C"):
+        homes[name] = tmp_path / f"profile-{name}"
+        _write_plugin(homes[name], "xref", name)
+        _write_config(homes[name], *(("xref",) if name != "B" else ()))
+    managers = {name: PluginManager(scope_key=str(path)) for name, path in homes.items()}
+    try:
+        for name, manager in managers.items():
+            caplog.clear()
+            with caplog.at_level(logging.WARNING):
+                manager.discover_and_load()
+            assert (_REFUSED in caplog.text) is (name == "C")
+
+        caplog.clear()
+        with caplog.at_level(logging.WARNING):
+            managers["C"].discover_and_load(force=True)
+        assert _REFUSED in caplog.text
+        managers["A"].discover_and_load(force=True)
+        for manager in managers.values():
+            with _plugin_home_scope(manager.home_path):
+                assert _expand("1") == "A 1"
+    finally:
+        for manager in managers.values():
+            manager.unload()