fix(plugins): a reload publishes its own staged context reference

The staged context references were found by slicing the manager's
registration list from the offset saved at load start. An earlier
registration disposed while the plugin loaded shortens that list, so the
reload's entries slid below the offset and the old provider kept answering
while the plugin was reported as reloaded.

Each load's PluginContext now keeps the registrations it staged, and a
successful load publishes exactly those, then clears the list so a
provider that holds its ctx does not keep the provider it replaced.

hermes_cli/plugins.py

@@ -237,6 +237,7 @@ class PluginContext(PluginCapabilityCtxMixin):
         # Set when this context's load overran ``plugins.load_timeout_seconds``: the abandoned worker may
         # still be running register(), and nothing it registers from then on may reach a registry.
         self._load_abandoned = False
+        self._staged: list[PluginRegistration] = []  # this load's staged registrations, published if it succeeds
 
     def _abandon_load(self) -> None:
         """Mark this load as timed out; every later ``register_*``/``subscribe``/``on_unload`` is ignored."""
@@ -707,16 +708,15 @@ class PluginContext(PluginCapabilityCtxMixin):
         return handle
 
     def register_context_reference(self, provider) -> None:
-        """Register a :class:`agent.context_references.ContextReferenceProvider`; ``provider.prefix``
-        defines ``@<prefix>:``. Built-in prefixes (diff, staged, file, folder, git, url) are
-        rejected."""
+        """Register a :class:`agent.context_references.ContextReferenceProvider`; ``provider.prefix`` defines
+        ``@<prefix>:``. Built-in prefixes (diff, staged, file, folder, git, url) are rejected."""
         from agent.context_references import (
             ContextReferenceProvider as _CRP, register_context_reference_provider as _register,
         )
         if self._wrong_type(provider, _CRP, "context reference provider"):
             return
         try:
-            self._manager._track_context_reference(self.manifest, provider, _register)
+            self._staged.append(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

@@ -100,7 +100,9 @@ class PluginLedgerMixin:
         )
         return self._track_registration(manifest, kind, name, lease.dispose)
 
-    def _track_context_reference(self, manifest: PluginManifest, provider: Any, register: Callable[..., Any]) -> None:
+    def _track_context_reference(
+        self, manifest: PluginManifest, provider: Any, register: Callable[..., Any],
+    ) -> PluginRegistration:
         """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.
@@ -109,13 +111,19 @@ class PluginLedgerMixin:
         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
+        registration = self._track_registration(manifest, "context_reference", prefix, lambda: None)
+        registration.publish = publish
+        return registration
 
-    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:
+    @staticmethod
+    def _publish_staged_registrations(ctx: Any) -> None:
+        """Apply the staged changes of a plugin load that succeeded: only the ones this load's ``ctx`` made.
+        Neither an offset into ``_registration_order`` (an earlier registration disposed mid-load shortens it)
+        nor the plugin's whole ledger (a timed-out load's worker can stage into it late) is safe."""
+        for registration in ctx._staged:
+            if registration.active and registration.publish is not None:
                 registration.publish()
+        ctx._staged.clear()  # a provider that keeps its ctx must not keep the provider it replaced
 
     def _active_persistent(self) -> list[PluginRegistration]:
         """Live persistent registrations across every plugin in the ownership ledger."""

hermes_cli/plugins_loader.py

@@ -623,7 +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)
+                self._publish_staged_registrations(ctx)
                 loaded.enabled = True
                 from hermes_cli.plugins_ledger import _hook_source_of
 

tests/hermes_cli/test_plugin_context_reference_reload.py

@@ -6,6 +6,8 @@ from __future__ import annotations
 import asyncio
 import logging
 import os
+import sys
+import types
 from pathlib import Path
 
 import hermes_yaml as yaml
@@ -79,6 +81,44 @@ def test_successful_reload_replaces_the_plugins_provider(home):
         manager.unload()
 
 
+def test_reload_publishes_when_earlier_registrations_are_disposed_during_it(home, monkeypatch):
+    # Registrations made before this plugin's load can be disposed while it loads (a supervised task
+    # finishing, a handle released from a callback); the manager's registration list gets shorter then.
+    handles: list = []
+    shared = types.ModuleType("ctxref_shared")
+    shared.__dict__["handles"] = handles
+    monkeypatch.setitem(sys.modules, "ctxref_shared", shared)
+    holder = home / "plugins" / "aaa-holder"
+    holder.mkdir(parents=True)
+    (holder / "plugin.yaml").write_text(yaml.safe_dump({"name": "aaa-holder", "version": "1.0"}), encoding="utf-8")
+    (holder / "__init__.py").write_text(
+        "import ctxref_shared\n\ndef register(ctx):\n"
+        "    ctxref_shared.handles[:] = [ctx.on_unload(lambda: None), ctx.on_unload(lambda: None)]\n",
+        encoding="utf-8")
+    _write_plugin(home, "tickets", "v1")
+    _write_config(home, "aaa-holder", "tickets")
+    manager = PluginManager()
+    try:
+        manager.discover_and_load()
+        assert _expand("7") == "v1 7"
+
+        _write_plugin(home, "tickets", "v2")
+        source = home / "plugins" / "tickets" / "__init__.py"
+        source.write_text(source.read_text(encoding="utf-8").replace(
+            "def register(ctx):\n",
+            "def register(ctx):\n"
+            "    import ctxref_shared\n"
+            "    for handle in ctxref_shared.handles:\n"
+            "        handle.dispose()\n",
+        ), encoding="utf-8")
+        manager.discover_and_load(force=True)
+        assert handles and not any(handle.active for handle in handles)  # disposed mid-load
+        assert manager._plugins["tickets"].error is None
+        assert _expand("7") == "v2 7"
+    finally:
+        manager.unload()
+
+
 def test_failed_reload_keeps_the_working_provider(home):
     _write_plugin(home, "tickets", "working")
     _write_config(home, "tickets")