mirror/nanobot
1
0
Fork 0
mirror of https://github.com/HKUDS/nanobot.git synced 2026-08-10 22:38:40 +03:00
Code Issues Packages Projects Releases Wiki Activity

refactor(plugins): deepen lifecycle boundary

Browse Source
This commit is contained in:
Xubin Ren
2026-08-10 23:43:32 +09:00
parent 3700e1a9f7
commit 7c44bd17b9
8 changed files with 76 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nanobot/agent/plugins.py
+5 -18
View File
@@ -83,7 +83,7 @@ class AgentPluginState:
setup_required: bool
def discover_agent_plugins(workspace: Path) -> list[AgentPlugin]:
def _discover_agent_plugins(workspace: Path) -> list[AgentPlugin]:
"""Return installed packages found under ``<workspace>/plugins/*``."""
workspace = workspace.expanduser().resolve()
plugins_root = workspace / "plugins"
@@ -113,23 +113,10 @@ def discover_agent_plugins(workspace: Path) -> list[AgentPlugin]:
return plugins
def discover_agent_plugin_skills(workspace: Path) -> list[AgentPluginSkill]:
"""Return skills supplied by locally installed plugin packages.
The portable format does not prescribe acquisition or installation UX.
nanobot currently treats package presence in the workspace ``plugins``
directory as installed; activation remains a separate trust decision.
"""
skills: list[AgentPluginSkill] = []
for plugin in discover_agent_plugins(workspace):
skills.extend(_discover_plugin_skills(plugin.name, plugin.root))
return skills
def enabled_agent_plugin_skills(workspace: Path) -> list[AgentPluginSkill]:
"""Return skills from plugins the user has explicitly enabled."""
skills: list[AgentPluginSkill] = []
for plugin in discover_agent_plugins(workspace):
for plugin in _discover_agent_plugins(workspace):
if _enabled(workspace, plugin.name):
skills.extend(_discover_plugin_skills(plugin.name, plugin.root))
return skills
@@ -195,7 +182,7 @@ def agent_plugin_mcp_servers(
User configuration wins on the unlikely event of a namespaced collision.
"""
servers: dict[str, MCPServerConfig] = {}
for plugin in discover_agent_plugins(workspace):
for plugin in _discover_agent_plugins(workspace):
if not _enabled(workspace, plugin.name):
continue
plugin_servers = _plugin_mcp_servers(workspace, plugin)
@@ -212,7 +199,7 @@ def agent_plugin_mcp_servers(
def discover_agent_plugin_states(workspace: Path) -> list[AgentPluginState]:
"""Return component and lifecycle state for discovered plugins."""
states: list[AgentPluginState] = []
for plugin in discover_agent_plugins(workspace):
for plugin in _discover_agent_plugins(workspace):
states.append(
AgentPluginState(
plugin=plugin,
@@ -227,7 +214,7 @@ def discover_agent_plugin_states(workspace: Path) -> list[AgentPluginState]:
def set_agent_plugin_enabled(workspace: Path, name: str, enabled: bool) -> AgentPlugin:
"""Enable or disable one installed plugin."""
plugin = next((item for item in discover_agent_plugins(workspace) if item.name == name), None)
plugin = next((item for item in _discover_agent_plugins(workspace) if item.name == name), None)
if plugin is None:
raise ValueError(f"unknown Agent Plugin '{name}'")
data = _plugin_data_dir(workspace, plugin.name, create=True)
nanobot/apps/cli/service.py
+15 -22
View File
@@ -230,6 +230,15 @@ def _plugin_skill_relative_path(name: str) -> str:
return f"plugins/{skill_name}/skills/{skill_name}/SKILL.md"
def cli_app_skill_relative_path(workspace: Path, name: str) -> str:
"""Return a CLI App's skill path, including the legacy location."""
canonical = Path(_plugin_skill_relative_path(name))
legacy = Path("skills") / _legacy_skill_name(name) / "SKILL.md"
if not (workspace / canonical).is_file() and (workspace / legacy).is_file():
return legacy.as_posix()
return canonical.as_posix()
def _has_shell_meta(command: str) -> bool:
return any(char in command for char in _SHELL_META_CHARS)
@@ -628,7 +637,7 @@ class CliAppManager:
"name": installed_name,
"entry_point": entry_point,
"source": str(data.get("source") or ""),
"skill": self.skill_relative_path(installed_name),
"skill": cli_app_skill_relative_path(self.workspace, installed_name),
"tool": "run_cli_app",
}
)
@@ -654,22 +663,6 @@ class CliAppManager:
install_cmd = str(app.get("install_cmd") or "")
return not _has_shell_meta(install_cmd)
def _skill_path(self, name: str) -> Path:
skill_name = _safe_skill_name(name)
return self.workspace / "plugins" / skill_name / "skills" / skill_name / "SKILL.md"
def _legacy_skill_path(self, name: str) -> Path:
return self.workspace / "skills" / _legacy_skill_name(name) / "SKILL.md"
def _installed_skill_path(self, name: str) -> Path:
path = self._skill_path(name)
legacy_path = self._legacy_skill_path(name)
return legacy_path if not path.is_file() and legacy_path.is_file() else path
def skill_relative_path(self, name: str) -> str:
"""Return the existing skill path, falling back to the canonical plugin path."""
return self._installed_skill_path(name).relative_to(self.workspace).as_posix()
def _app_payload(
self,
app: dict[str, Any],
@@ -705,7 +698,7 @@ class CliAppManager:
"status": status,
"logo_url": logo_url,
"brand_color": brand_color,
"skill_installed": self._installed_skill_path(name).is_file(),
"skill_installed": (self.workspace / cli_app_skill_relative_path(self.workspace, name)).is_file(),
"manifest": self._manifest_payload(app, logo_url=logo_url, brand_color=brand_color),
}
@@ -1121,7 +1114,7 @@ Use the `run_cli_app` tool with `name="{name}"` for command execution. Do not in
return f"---\n{frontmatter.strip()}\n---\n\n{body}"
def install_skill(self, app: dict[str, Any]) -> Path:
path = self._skill_path(str(app["name"]))
path = self.workspace / _plugin_skill_relative_path(str(app["name"]))
path.parent.mkdir(parents=True, exist_ok=True)
content = self._fetch_skill_content(app) or self._fallback_skill(app)
content = self._normalise_skill(content, app)
@@ -1135,16 +1128,16 @@ Use the `run_cli_app` tool with `name="{name}"` for command execution. Do not in
"description": _catalog_description(app),
})
_write_json(plugin_root / "plugin.json", manifest)
legacy_dir = self._legacy_skill_path(str(app["name"])).parent
legacy_dir = self.workspace / "skills" / _legacy_skill_name(str(app["name"]))
if legacy_dir.is_dir():
shutil.rmtree(legacy_dir)
return path
def remove_skill(self, name: str) -> None:
plugin_root = self._skill_path(name).parents[2]
plugin_root = (self.workspace / _plugin_skill_relative_path(name)).parents[2]
if plugin_root.is_dir():
shutil.rmtree(plugin_root)
legacy_dir = self._legacy_skill_path(name).parent
legacy_dir = self.workspace / "skills" / _legacy_skill_name(name)
if legacy_dir.is_dir():
shutil.rmtree(legacy_dir)
nanobot/apps/cli/utils.py
+2 -3
View File
@@ -20,9 +20,8 @@ def runtime_lines_for_request(
"""Return CLI App annotations from an immutable request snapshot."""
structured = metadata.get("cli_apps") if isinstance(metadata, Mapping) else None
if isinstance(structured, list):
from nanobot.apps.cli import CliAppManager
from nanobot.apps.cli.service import cli_app_skill_relative_path
manager = CliAppManager(workspace=workspace)
structured_items = cast(list[Any], structured)
mentions = [
cast(Mapping[str, Any], item) for item in structured_items
@@ -35,7 +34,7 @@ def runtime_lines_for_request(
f"@{str(item['name']).strip().lower()} "
f"(installed; tool=run_cli_app; "
f"entry_point={str(item.get('entry_point') or 'unknown')}; "
f"skill={manager.skill_relative_path(str(item['name']))}). "
f"skill={cli_app_skill_relative_path(workspace, str(item['name']))}). "
"Read the skill when useful, then run this app with `run_cli_app`; do not bypass it with shell."
for item in mentions
if str(item.get("name") or "").strip()
nanobot/webui/mcp_presets_api.py
+21 -6
View File
@@ -1453,6 +1453,7 @@ async def mcp_presets_settings_action(
*,
reload_mcp: McpReload | None = None,
config: WebUISettingsConfig | None = None,
remote: bool = False,
) -> dict[str, Any]:
"""Run a WebUI MCP preset action and hot-reload the agent when config changes."""
config_path = config.path if config is not None else None
@@ -1462,14 +1463,28 @@ async def mcp_presets_settings_action(
if name.startswith("plugin-"):
plugin_config = load_config(config_path) if config_path is not None else load_config()
plugin_name = name.removeprefix("plugin-")
installed = {
state.plugin.name
for state in discover_agent_plugin_states(plugin_config.workspace_path)
if state.mcp_servers or state.plugin.install_command
}
if name not in plugin_config.tools.mcp_servers and plugin_name in installed:
plugin_state = next(
(
state
for state in discover_agent_plugin_states(plugin_config.workspace_path)
if state.plugin.name == plugin_name
and (state.mcp_servers or state.plugin.install_command)
),
None,
)
if name not in plugin_config.tools.mcp_servers and plugin_state is not None:
if action not in {"enable", "disable"}:
raise McpPresetError("Agent Plugins support enable and disable actions only")
if (
action == "enable"
and plugin_state.setup_required
and remote
and not plugin_config.tools.webui_allow_remote_package_install
):
raise McpPresetError(
"Agent Plugin setup is restricted to the local WebUI",
status=403,
)
plugin = await asyncio.to_thread(
set_agent_plugin_enabled,
plugin_config.workspace_path,
nanobot/webui/settings_routes.py
+1 -22
View File
@@ -17,7 +17,6 @@ from typing import Any, cast
from websockets.http11 import Request as WsRequest
from websockets.http11 import Response
from nanobot.agent.plugins import discover_agent_plugin_states
from nanobot.agent.tools.image_generation import request_image_generation_reload
from nanobot.agent.tools.mcp import request_mcp_reload
from nanobot.api.runtime import ApiRuntime, ApiStartOptions, api_runtime_paths
@@ -1218,32 +1217,12 @@ class WebUISettingsRouter:
return self._unauthorized()
try:
query = self._parse_mcp_settings_query(request)
name = (_query_first(query, "name") or "").strip()
if action == "enable" and name.startswith("plugin-"):
config = self.settings.config.load()
plugin_state = next(
(
state
for state in discover_agent_plugin_states(config.workspace_path)
if f"plugin-{state.plugin.name}" == name
),
None,
)
if (
name not in config.tools.mcp_servers
and plugin_state is not None
and plugin_state.setup_required
and not self._allow_feature_package_install(connection, request)
):
return self._error_response(
403,
"Agent Plugin setup is restricted to the local WebUI",
)
payload = await mcp_presets_settings_action(
action,
query,
reload_mcp=lambda: request_mcp_reload(self.bus),
config=self.settings.config,
remote=not _is_local_browser_request(connection, request.headers),
)
except Exception as e:
status = getattr(e, "status", 500)
tests/agent/test_agent_plugins.py
+16 -8
View File
@@ -14,8 +14,8 @@ from nanobot.agent.plugins import (
AGENT_PLUGIN_MCP_SCHEMA,
AGENT_PLUGIN_SCHEMA,
agent_plugin_mcp_servers,
discover_agent_plugin_skills,
discover_agent_plugin_states,
enabled_agent_plugin_skills,
set_agent_plugin_enabled,
)
from nanobot.agent.skills import SkillsLoader
@@ -57,6 +57,10 @@ def _write_plugin(
return root
def _loaded_plugin_skills(workspace: Path) -> list[str]:
return [skill.name for skill in enabled_agent_plugin_skills(workspace)]
def test_skills_loader_discovers_agent_plugin_skill(tmp_path: Path) -> None:
plugin = _write_plugin(tmp_path, "acme-tools")
_write_skill(plugin, "release-notes", description="Draft release notes from changes.")
@@ -103,8 +107,9 @@ def test_agent_plugin_skills_are_direct_children_only(tmp_path: Path) -> None:
"---\nname: nested\ndescription: Nested skill.\n---\n",
encoding="utf-8",
)
set_agent_plugin_enabled(tmp_path, "acme-tools", True)
assert [skill.name for skill in discover_agent_plugin_skills(tmp_path)] == ["direct"]
assert _loaded_plugin_skills(tmp_path) == ["direct"]
@pytest.mark.parametrize(
@@ -123,7 +128,7 @@ def test_invalid_agent_plugin_manifest_is_skipped(
plugin = _write_plugin(tmp_path, "demo", manifest=manifest)
_write_skill(plugin, "example")
assert discover_agent_plugin_skills(tmp_path) == []
assert discover_agent_plugin_states(tmp_path) == []
def test_unknown_manifest_fields_and_non_object_extensions_are_ignored(tmp_path: Path) -> None:
@@ -138,8 +143,9 @@ def test_unknown_manifest_fields_and_non_object_extensions_are_ignored(tmp_path:
},
)
_write_skill(plugin, "example")
set_agent_plugin_enabled(tmp_path, "demo", True)
assert [skill.name for skill in discover_agent_plugin_skills(tmp_path)] == ["example"]
assert _loaded_plugin_skills(tmp_path) == ["example"]
def test_agent_plugin_discovers_contained_raster_logo(tmp_path: Path) -> None:
@@ -155,7 +161,7 @@ def test_agent_plugin_discovers_contained_raster_logo(tmp_path: Path) -> None:
assets = plugin / "assets"
assets.mkdir()
(assets / "icon.png").write_bytes(b"\x89PNG\r\n\x1a\nlogo")
assert agent_plugins.discover_agent_plugins(tmp_path)[0].logo == assets / "icon.png"
assert discover_agent_plugin_states(tmp_path)[0].plugin.logo == assets / "icon.png"
def test_agent_plugin_logo_cannot_escape_package(tmp_path: Path) -> None:
@@ -177,7 +183,7 @@ def test_agent_plugin_logo_cannot_escape_package(tmp_path: Path) -> None:
except OSError as exc:
pytest.skip(f"file symlink unavailable: {exc}")
assert agent_plugins.discover_agent_plugins(tmp_path)[0].logo is None
assert discover_agent_plugin_states(tmp_path)[0].plugin.logo is None
@pytest.mark.parametrize(
@@ -197,8 +203,9 @@ def test_invalid_agent_skill_is_skipped(
skill = plugin / "skills" / skill_name
skill.mkdir(parents=True)
(skill / "SKILL.md").write_text(f"---\n{frontmatter}\n---\n", encoding="utf-8")
set_agent_plugin_enabled(tmp_path, "demo", True)
assert discover_agent_plugin_skills(tmp_path) == []
assert _loaded_plugin_skills(tmp_path) == []
def test_workspace_skill_overrides_plugin_skill(tmp_path: Path) -> None:
@@ -261,8 +268,9 @@ def test_plugin_skill_symlink_cannot_escape_plugin_root(tmp_path: Path) -> None:
)
except OSError as exc:
pytest.skip(f"directory symlink unavailable: {exc}")
set_agent_plugin_enabled(tmp_path, "demo", True)
assert discover_agent_plugin_skills(tmp_path) == []
assert _loaded_plugin_skills(tmp_path) == []
def test_plugin_mcp_requires_explicit_enable(tmp_path: Path) -> None:
tests/cli_apps/test_service.py
-4
View File
@@ -10,7 +10,6 @@ from types import SimpleNamespace
import pytest
from nanobot.agent import plugins as agent_plugins
from nanobot.agent.plugins import discover_agent_plugin_skills
from nanobot.agent.skills import SkillsLoader
from nanobot.apps.cli.service import CliAppError, CliAppManager, CliAppsRuntimeConfig
@@ -426,9 +425,6 @@ def test_install_dispatches_safe_pip_and_installs_skill(
}
assert "name: cli-app-gimp" in skill.read_text(encoding="utf-8")
assert 'run_cli_app` tool with `name="gimp"' in skill.read_text(encoding="utf-8")
assert [item.name for item in discover_agent_plugin_skills(manager.workspace)] == [
"cli-app-gimp"
]
assert [
item["name"]
for item in SkillsLoader(manager.workspace).list_skills()
tests/webui/test_mcp_presets_api.py
+16 -1
View File
@@ -33,6 +33,9 @@ def _write_agent_plugin(workspace: Path) -> None:
command = root / "bin" / "server"
command.parent.mkdir(parents=True, exist_ok=True)
command.write_text("#!/bin/sh\n", encoding="utf-8")
setup = root / "bin" / "install"
setup.write_text("#!/bin/sh\n", encoding="utf-8")
setup.chmod(0o755)
assets = root / "assets"
assets.mkdir()
(assets / "icon.png").write_bytes(b"\x89PNG\r\n\x1a\nlogo")
@@ -48,6 +51,7 @@ def _write_agent_plugin(workspace: Path) -> None:
"accentColor": "#ff7a1a",
"logo": "./assets/icon.png",
"permissions": ["screen-recording"],
"installCommand": ["./bin/install"],
}
},
}
@@ -116,13 +120,23 @@ def test_agent_plugin_reuses_mcp_catalog_and_runtime_action(
assert row["logo_url"] == "data:image/png;base64,iVBORw0KGgpsb2dv"
assert row["install_supported"] is False
assert row["installed"] is True
assert row["configured"] is True
assert row["configured"] is False
assert row["enabled"] is False
assert row["status"] == "disabled"
async def reload() -> dict[str, object]:
return {"ok": True, "message": "MCP reloaded.", "requires_restart": False}
with pytest.raises(McpPresetError, match="restricted") as restricted:
asyncio.run(
mcp_presets_settings_action(
"enable",
{"name": ["plugin-desktop"]},
remote=True,
)
)
assert restricted.value.status == 403
enabled = asyncio.run(
mcp_presets_settings_action(
"enable",
@@ -131,6 +145,7 @@ def test_agent_plugin_reuses_mcp_catalog_and_runtime_action(
)
)
enabled_row = next(item for item in enabled["presets"] if item["name"] == "plugin-desktop")
assert enabled_row["configured"] is True
assert enabled_row["enabled"] is True
assert enabled_row["status"] == "enabled"
assert enabled["requires_restart"] is False