ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

feat(plugins): add ownership ledger unload lifecycle

Browse Source
This commit is contained in:
doncazper
2026-08-01 17:53:10 -07:00
committed by Teknium
parent 03d2c0e144
commit 22af80bcfd
12 changed files with 1000 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/browser_registry.py
+16
View File
@@ -94,6 +94,22 @@ def get_provider(name: str) -> Optional[BrowserProvider]:
return _providers.get(name.strip())
def restore_registration(
name: str,
current: BrowserProvider,
previous: Optional[BrowserProvider],
) -> bool:
"""Restore a plugin registration only when *current* is still installed."""
with _lock:
if _providers.get(name) is not current:
return False
if previous is None:
_providers.pop(name, None)
else:
_providers[name] = previous
return True
# ---------------------------------------------------------------------------
# Active-provider resolution
# ---------------------------------------------------------------------------
agent/image_gen_registry.py
+16
View File
@@ -72,6 +72,22 @@ def get_provider(name: str) -> Optional[ImageGenProvider]:
return _providers.get(name.strip())
def restore_registration(
name: str,
current: ImageGenProvider,
previous: Optional[ImageGenProvider],
) -> bool:
"""Restore a plugin registration only when *current* is still installed."""
with _lock:
if _providers.get(name) is not current:
return False
if previous is None:
_providers.pop(name, None)
else:
_providers[name] = previous
return True
def get_active_provider() -> Optional[ImageGenProvider]:
"""Resolve the currently-active provider.
agent/secret_sources/registry.py
+16
View File
@@ -156,6 +156,22 @@ def get_source(name: str) -> Optional[SecretSource]:
return _SOURCES.get(name)
def restore_registration(
name: str,
current: SecretSource,
previous: Optional[SecretSource],
) -> bool:
"""Restore a host-owned source registration if it is still current."""
_ensure_builtin_sources()
if _SOURCES.get(name) is not current:
return False
if previous is None:
_SOURCES.pop(name, None)
else:
_SOURCES[name] = previous
return True
def list_sources() -> List[SecretSource]:
_ensure_builtin_sources()
return list(_SOURCES.values())
agent/transcription_registry.py
+17
View File
@@ -118,6 +118,23 @@ def get_provider(name: str) -> Optional[TranscriptionProvider]:
return _providers.get(name.strip().lower())
def restore_registration(
name: str,
current: TranscriptionProvider,
previous: Optional[TranscriptionProvider],
) -> bool:
"""Restore a plugin registration only when *current* is still installed."""
key = name.strip().lower()
with _lock:
if _providers.get(key) is not current:
return False
if previous is None:
_providers.pop(key, None)
else:
_providers[key] = previous
return True
def _reset_for_tests() -> None:
"""Clear the registry. **Test-only.**"""
with _lock:
agent/tts_registry.py
+17
View File
@@ -128,6 +128,23 @@ def get_provider(name: str) -> Optional[TTSProvider]:
return _providers.get(name.strip().lower())
def restore_registration(
name: str,
current: TTSProvider,
previous: Optional[TTSProvider],
) -> bool:
"""Restore a plugin registration only when *current* is still installed."""
key = name.strip().lower()
with _lock:
if _providers.get(key) is not current:
return False
if previous is None:
_providers.pop(key, None)
else:
_providers[key] = previous
return True
def _reset_for_tests() -> None:
"""Clear the registry. **Test-only.**"""
with _lock:
agent/video_gen_registry.py
+16
View File
@@ -76,6 +76,22 @@ def get_provider(name: str) -> Optional[VideoGenProvider]:
return _providers.get(name.strip())
def restore_registration(
name: str,
current: VideoGenProvider,
previous: Optional[VideoGenProvider],
) -> bool:
"""Restore a plugin registration only when *current* is still installed."""
with _lock:
if _providers.get(name) is not current:
return False
if previous is None:
_providers.pop(name, None)
else:
_providers[name] = previous
return True
def get_active_provider() -> Optional[VideoGenProvider]:
"""Resolve the currently-active provider.
agent/web_search_registry.py
+16
View File
@@ -90,6 +90,22 @@ def get_provider(name: str) -> Optional[WebSearchProvider]:
return _providers.get(name.strip())
def restore_registration(
name: str,
current: WebSearchProvider,
previous: Optional[WebSearchProvider],
) -> bool:
"""Restore a plugin registration only when *current* is still installed."""
with _lock:
if _providers.get(name) is not current:
return False
if previous is None:
_providers.pop(name, None)
else:
_providers[name] = previous
return True
# ---------------------------------------------------------------------------
# Active-provider resolution
# ---------------------------------------------------------------------------
gateway/platform_registry.py
+44
View File
@@ -245,6 +245,50 @@ class PlatformRegistry:
return
self._deferred[name] = loader
def snapshot_registration(
self,
name: str,
) -> tuple[Optional[PlatformEntry], Optional[Callable[[], None]]]:
"""Return the concrete and deferred state for *name* without resolving it.
This host-facing snapshot lets the plugin ledger restore a deferred
platform loader that a concrete registration displaced, without
importing the displaced adapter as a side effect of taking the
snapshot.
"""
return self._entries.get(name), self._deferred.get(name)
def restore_registration(
self,
name: str,
current: tuple[Optional[PlatformEntry], Optional[Callable[[], None]]],
previous: tuple[Optional[PlatformEntry], Optional[Callable[[], None]]],
) -> bool:
"""Restore a platform registration if its full state is still current.
The identity checks protect a later registration from being removed
while still allowing an unloaded override to reveal the registration
it displaced. Both concrete entries and deferred loaders are part of
the state because bundled platform plugins load lazily.
"""
current_state = self.snapshot_registration(name)
if (
current_state[0] is not current[0]
or current_state[1] is not current[1]
):
return False
entry, loader = previous
if entry is None:
self._entries.pop(name, None)
else:
self._entries[name] = entry
if loader is None:
self._deferred.pop(name, None)
else:
self._deferred[name] = loader
return True
def _resolve(self, name: str) -> None:
"""Run the deferred loader for *name* if one is pending."""
loader = self._deferred.pop(name, None)
hermes_cli/dashboard_auth/registry.py
+16
View File
@@ -46,6 +46,22 @@ def get_provider(name: str) -> Optional[DashboardAuthProvider]:
return _providers.get(name)
def restore_registration(
name: str,
current: DashboardAuthProvider,
previous: Optional[DashboardAuthProvider],
) -> bool:
"""Restore a host-owned provider registration if it is still current."""
with _lock:
if _providers.get(name) is not current:
return False
if previous is None:
_providers.pop(name, None)
else:
_providers[name] = previous
return True
def list_providers() -> List[DashboardAuthProvider]:
"""All registered providers, in registration order."""
with _lock:
hermes_cli/plugins.py
+582 -82
View File
File diff suppressed because it is too large Load Diff
tests/hermes_cli/test_plugin_ownership_ledger.py
+191
View File
@@ -0,0 +1,191 @@
"""End-to-end coverage for plugin registration ownership and reload cleanup."""
from __future__ import annotations
from pathlib import Path
import yaml
def _write_plugin(hermes_home: Path) -> None:
plugin_dir = hermes_home / "plugins" / "ledger_probe"
plugin_dir.mkdir(parents=True)
(plugin_dir / "plugin.yaml").write_text(
yaml.safe_dump(
{
"name": "ledger_probe",
"version": "0.1.0",
"description": "ownership ledger probe",
}
)
)
(plugin_dir / "SKILL.md").write_text("# Ledger probe\n")
(plugin_dir / "__init__.py").write_text(
"from pathlib import Path\n"
"\n"
"def _hook(**kwargs):\n"
" return {'hook': 'ledger'}\n"
"\n"
"def _middleware(**kwargs):\n"
" return {'middleware': 'ledger'}\n"
"\n"
"def register(ctx):\n"
" ctx.register_tool(\n"
" name='ledger_probe_tool',\n"
" toolset='plugin_ledger_probe',\n"
" schema={'name': 'ledger_probe_tool', 'parameters': {'type': 'object', 'properties': {}}},\n"
" handler=lambda args, **kwargs: 'ledger',\n"
" )\n"
" ctx.register_platform(\n"
" name='ledger_probe_platform',\n"
" label='Ledger probe',\n"
" adapter_factory=lambda config: object(),\n"
" check_fn=lambda: True,\n"
" )\n"
" ctx.register_cli_command(\n"
" 'ledger-probe-cli', 'Ledger CLI', lambda parser: None,\n"
" handler_fn=lambda args: None,\n"
" )\n"
" ctx.register_command(\n"
" 'ledger-probe-command', lambda args: args,\n"
" description='Ledger command',\n"
" )\n"
" ctx.register_hook('pre_tool_call', _hook)\n"
" ctx.register_middleware('tool_request', _middleware)\n"
" ctx.register_auxiliary_task(\n"
" key='ledger_probe_task',\n"
" display_name='Ledger probe task',\n"
" description='Ledger task',\n"
" )\n"
" ctx.register_skill(\n"
" 'ledger-probe', Path(__file__).with_name('SKILL.md'),\n"
" 'Ledger skill',\n"
" )\n"
)
(hermes_home / "config.yaml").write_text(
yaml.safe_dump({"plugins": {"enabled": ["ledger_probe"]}})
)
def test_load_force_reload_and_unload_remove_every_manager_registration(
tmp_path,
monkeypatch,
):
"""A real temporary plugin has one live registration after each reload."""
import hermes_cli.plugins as plugins_mod
from gateway.platform_registry import platform_registry
from hermes_cli.plugins import PluginManager
from tools.registry import registry
hermes_home = tmp_path / "hermes"
_write_plugin(hermes_home)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(
plugins_mod,
"get_bundled_plugins_dir",
lambda: tmp_path / "empty-bundled",
)
monkeypatch.setattr(PluginManager, "_scan_entry_points", lambda self: [])
manager = PluginManager()
manager.discover_and_load()
first_tool = registry.get_entry("ledger_probe_tool")
first_platform = platform_registry.get("ledger_probe_platform")
first_hook = manager._hooks["pre_tool_call"][0]
first_middleware = manager._middleware["tool_request"][0]
first_command = manager._plugin_commands["ledger-probe-command"]
first_cli_command = manager._cli_commands["ledger-probe-cli"]
first_skill = manager._plugin_skills["ledger_probe:ledger-probe"]
assert first_tool is not None
assert first_platform is not None
assert set(registration.kind for registration in manager._ownership_ledger["ledger_probe"]) == {
"tool",
"platform",
"cli_command",
"command",
"hook",
"middleware",
"auxiliary_task",
"skill",
}
manager.discover_and_load(force=True)
second_tool = registry.get_entry("ledger_probe_tool")
second_platform = platform_registry.get("ledger_probe_platform")
assert second_tool is not None and second_tool is not first_tool
assert second_platform is not None and second_platform is not first_platform
assert second_tool.handler is not first_tool.handler
assert first_hook not in manager._hooks["pre_tool_call"]
assert first_middleware not in manager._middleware["tool_request"]
assert len(manager._hooks["pre_tool_call"]) == 1
assert len(manager._middleware["tool_request"]) == 1
assert manager._plugin_commands["ledger-probe-command"] is not first_command
assert manager._cli_commands["ledger-probe-cli"] is not first_cli_command
assert manager._plugin_skills["ledger_probe:ledger-probe"] is not first_skill
assert len(manager._aux_tasks) == 1
assert [
entry
for entry in platform_registry.plugin_entries()
if entry.name == "ledger_probe_platform"
] == [second_platform]
assert manager.unload("ledger_probe") is True
assert registry.get_entry("ledger_probe_tool") is None
assert not platform_registry.is_registered("ledger_probe_platform")
assert "pre_tool_call" not in manager._hooks
assert "tool_request" not in manager._middleware
assert "ledger-probe-command" not in manager._plugin_commands
assert "ledger-probe-cli" not in manager._cli_commands
assert "ledger_probe:ledger-probe" not in manager._plugin_skills
assert manager._aux_tasks == {}
assert manager._ownership_ledger == {}
def test_reverse_unload_restores_an_overridden_platform_registration():
"""Reverse teardown reveals an older entry before removing it."""
from gateway.platform_registry import platform_registry
from hermes_cli.plugins import PluginContext, PluginManager, PluginManifest
name = "ledger_override_platform"
previous = platform_registry.snapshot_registration(name)
manager_a = PluginManager()
manager_b = PluginManager()
context_a = PluginContext(
PluginManifest(name="ledger_owner_a", key="ledger_owner_a"), manager_a
)
context_b = PluginContext(
PluginManifest(name="ledger_owner_b", key="ledger_owner_b"), manager_b
)
try:
handle_a = context_a.register_platform(
name=name,
label="Ledger A",
adapter_factory=lambda config: "a",
check_fn=lambda: True,
)
entry_a = platform_registry.get(name)
handle_b = context_b.register_platform(
name=name,
label="Ledger B",
adapter_factory=lambda config: "b",
check_fn=lambda: True,
)
entry_b = platform_registry.get(name)
assert handle_a is not None and handle_b is not None
assert entry_a is not None and entry_b is not None
assert entry_a is not entry_b
handle_b.dispose()
assert platform_registry.get(name) is entry_a
handle_a.dispose()
assert platform_registry.snapshot_registration(name) == previous
finally:
# The test uses a deliberately unique name, but restore any state that
# a surrounding test may have installed under it.
current = platform_registry.snapshot_registration(name)
platform_registry.restore_registration(name, current, previous)
tools/registry.py
+53
View File
@@ -710,6 +710,59 @@ class ToolRegistry:
self._generation += 1
logger.debug("Deregistered tool: %s", name)
def restore_registration(
self,
name: str,
current: ToolEntry,
previous: Optional[ToolEntry],
) -> bool:
"""Restore a host-owned registration if it is still current.
This is the narrow inverse used by the plugin ownership ledger. The
identity check is deliberate: another plugin (or another
``PluginManager`` in a multi-profile process) may have registered a
newer entry under the same name, in which case unloading this entry
must leave the newer entry untouched.
"""
with self._lock:
if self._tools.get(name) is not current:
return False
if previous is None:
self._tools.pop(name, None)
else:
self._tools[name] = previous
# Rebuild the affected toolset checks from the surviving entries.
# A plugin may have replaced an entry in the same toolset, so
# simply leaving the current check_fn behind would retain stale
# plugin state after restoration.
affected_toolsets = {current.toolset}
if previous is not None:
affected_toolsets.add(previous.toolset)
for toolset in affected_toolsets:
surviving = [
entry for entry in self._tools.values()
if entry.toolset == toolset
]
check_fn = next(
(entry.check_fn for entry in surviving if entry.check_fn),
None,
)
if check_fn is None:
self._toolset_checks.pop(toolset, None)
else:
self._toolset_checks[toolset] = check_fn
if not surviving:
self._toolset_aliases = {
alias: target
for alias, target in self._toolset_aliases.items()
if target != toolset
}
self._generation += 1
logger.debug("Restored tool registration: %s", name)
return True
# ------------------------------------------------------------------
# Schema retrieval
# ------------------------------------------------------------------