refactor(plugins): consolidate validation boundaries

docs/configuration.md

@@ -2308,10 +2308,9 @@ Disabled skills are excluded from the main agent's skill summary, from always-on
 
 ### Agent Plugins v1
 
-nanobot also loads locally installed [Agent Plugins](https://agent-plugins.org/) from
+nanobot discovers [Agent Plugins](https://agent-plugins.org/) in
-`<workspace>/plugins/<plugin>/`. Package presence in this directory is the installation state;
+`<workspace>/plugins/<plugin>/`. A v1 package has `plugin.json` and may provide skills, MCP
-enabling it is a separate trust decision. A supported package has a root `plugin.json` that
+servers, or both:
-targets Agent Plugins v1 and may provide skills, MCP servers, or both:
 
 ```text
 plugins/
@@ -2323,38 +2322,20 @@ plugins/
             └── SKILL.md
 ```
 
-Plugin skills use the same progressive loading and `$skill-name` invocation as workspace
+Directory presence means installed; activation is an explicit trust decision in **Apps**.
-skills after the plugin is explicitly enabled. Disabling a plugin removes both its skills and
+Enabled skills use normal progressive loading and `$skill-name` invocation. Workspace skills
-MCP servers from the agent. A workspace skill wins when it has the same name as an enabled
+override plugin skills, which override built-ins. Enabled `stdio` servers from `mcp.json` receive
-plugin skill; plugin skills win over built-in skills. Invalid manifests, invalid Agent Skills,
+contained `PLUGIN_ROOT` and isolated `PLUGIN_DATA` paths; explicit `tools.mcpServers` entries win
-nested skill directories, and paths that resolve outside the plugin root are ignored.
+name collisions. Invalid manifests, components, nested skills, and escaping paths are ignored.
 
-Portable MCP servers declared in `mcp.json` appear in **Apps**, but are never started merely
+Enabled plugins run as the nanobot user; declared permissions are descriptive, not an OS sandbox.
-because a package exists. Enabling a plugin there is the explicit trust decision that activates
+The optional `extensions.dev.nanobot.installCommand` is a shell-free argv run once per version
-its executable components. The host expands `PLUGIN_ROOT` and an isolated `PLUGIN_DATA`, checks
+before local enable. Remote setup requires `tools.webuiAllowRemotePackageInstall`. The optional
-package paths before launch, and hot-reloads MCP connections. Explicit `tools.mcpServers`
+`extensions.dev.nanobot.logo` accepts a contained PNG, JPEG, or WebP up to 256 KiB.
-configuration wins over a plugin server if their host names collide. The v1 host currently
-supports plugin `stdio` servers; unsupported remote transports are skipped independently.
 
-Treat enabled plugins as local code running with the nanobot user's privileges. Manifest
+WebUI-installed CLI Apps use the same package layout as skills-only plugins. Their external
-permissions are descriptive; nanobot does not currently enforce them with an OS sandbox.
+executables remain managed by the CLI Apps installer; update refreshes the package and uninstall
-
+removes it. Future catalogs can acquire and place packages before using this same activation path.
-Plugins may optionally declare a shell-free `extensions.dev.nanobot.installCommand` array. The
-local WebUI runs it once per plugin version before first enable; remote WebUI clients cannot run
-plugin setup unless remote package installation was explicitly allowed. Enabling an already set
-up package does not install it and is allowed remotely. Agent Plugins v1 deliberately leaves
-distribution and installation UX to each host. A future catalog can therefore acquire, verify,
-and place a package atomically before handing it to this same runtime; users should still see one
-Install action, not separate download and installation steps.
-
-The optional `extensions.dev.nanobot.logo` field points to a packaged PNG, JPEG, or WebP asset
-such as `./assets/icon.png`. nanobot only reads contained raster files up to 256 KiB and embeds
-them locally in the Apps catalog; invalid or missing assets fall back to the plugin initials.
-
-CLI Apps installed from the WebUI use the same package layout. nanobot installs the CLI through
-its catalog adapter, then writes and enables a skills-only Agent Plugin under
-`<workspace>/plugins/`; updates refresh that package and uninstall removes it. The external
-executable remains managed by the CLI Apps installer rather than by the Agent Plugins manifest.
 
 ## Tool Hint Max Length
 

nanobot/agent/plugins.py

@@ -9,12 +9,13 @@ import subprocess
 from dataclasses import dataclass
 from hashlib import sha256
 from pathlib import Path
-from typing import Any, cast
+from typing import cast
 
-import yaml
 from filelock import FileLock
 from loguru import logger
+from pydantic import ValidationError
 
+from nanobot.agent.skills import parse_skill_metadata, valid_skill_metadata
 from nanobot.config.loader import get_config_path
 from nanobot.config.schema import MCPServerConfig
 
@@ -22,12 +23,9 @@ AGENT_PLUGIN_SCHEMA = "https://agent-plugins.org/schemas/1.0.0/plugin.schema.jso
 AGENT_PLUGIN_MCP_SCHEMA = "https://agent-plugins.org/schemas/1.0.0/mcp.schema.json"
 
 _PLUGIN_NAME = re.compile(r"^(?!.*(?:--|\.\.))[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?$")
-_SKILL_NAME = re.compile(r"^(?!.*--)[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$")
-_SKILL_FRONTMATTER = re.compile(r"^---\s*\r?\n(.*?)\r?\n---\s*\r?\n?", re.DOTALL)
 _MCP_SERVER_FIELDS = {"type", "command", "args", "env", "cwd"}
 _SETUP_ENV = {"HOME", "LANG", "LC_ALL", "LOGNAME", "PATH", "SHELL", "TMPDIR", "USER"}
 _SETUP_TIMEOUT_SECONDS = 600
-_LOGO_SUFFIXES = {".jpeg", ".jpg", ".png", ".webp"}
 _MAX_LOGO_BYTES = 256 * 1024
 
 
@@ -70,18 +68,11 @@ class AgentPluginState:
 def _discover_agent_plugins(workspace: Path) -> list[AgentPlugin]:
     """Return installed packages found under ``<workspace>/plugins/*``."""
     workspace = workspace.expanduser().resolve()
-    plugins_root = workspace / "plugins"
+    root = _contained_directory(workspace / "plugins", workspace)
-    if not plugins_root.is_dir():
+    if root is None:
         return []
     try:
-        root = plugins_root.resolve(strict=True)
+        candidates = sorted(root.iterdir(), key=lambda path: path.name)
-    except OSError:
-        return []
-    if not root.is_relative_to(workspace):
-        logger.warning("Ignoring Agent Plugins directory outside the workspace")
-        return []
-    try:
-        candidates = sorted(plugins_root.iterdir(), key=lambda path: path.name)
     except OSError as exc:
         logger.warning("Could not inspect Agent Plugins directory: {}", exc)
         return []
@@ -107,19 +98,9 @@ def enabled_agent_plugin_skills(workspace: Path) -> list[AgentPluginSkill]:
 
 
 def _load_manifest(plugin_root: Path) -> AgentPlugin | None:
-    manifest = _contained_file(plugin_root / "plugin.json", plugin_root)
+    payload = _read_object(plugin_root / "plugin.json", plugin_root)
-    if manifest is None:
+    if payload is None:
         return None
-    try:
-        value = cast(object, json.loads(manifest.read_text(encoding="utf-8")))
-    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
-        logger.warning("Ignoring invalid Agent Plugin manifest '{}': {}", manifest, exc)
-        return None
-    if not isinstance(value, dict):
-        logger.warning("Ignoring Agent Plugin manifest '{}': expected a JSON object", manifest)
-        return None
-
-    payload = cast(dict[str, Any], value)
     if payload.get("$schema") != AGENT_PLUGIN_SCHEMA:
         return None
     name = payload.get("name")
@@ -128,7 +109,7 @@ def _load_manifest(plugin_root: Path) -> AgentPlugin | None:
         or len(name) > 64
         or _PLUGIN_NAME.fullmatch(name) is None
     ):
-        logger.warning("Ignoring Agent Plugin manifest '{}': invalid name", manifest)
+        logger.warning("Ignoring Agent Plugin manifest in '{}': invalid name", plugin_root)
         return None
     extension = payload.get("extensions")
     extension_payload = cast(dict[str, object], extension) if isinstance(extension, dict) else {}
@@ -165,18 +146,15 @@ def agent_plugin_mcp_servers(
         for name, server in plugin_servers.items():
             host_name = plugin.name if len(plugin_servers) == 1 else f"{plugin.name}-{name}"
             servers[host_name] = server
-    for name, server in (configured or {}).items():
+    configured = configured or {}
-        if name in servers:
+    if collisions := servers.keys() & configured.keys():
-            logger.warning("Configured MCP server '{}' overrides an Agent Plugin server", name)
+        logger.warning("Configured MCP servers override Agent Plugins: {}", ", ".join(sorted(collisions)))
-        servers[name] = server
+    return servers | configured
-    return servers
 
 
 def discover_agent_plugin_states(workspace: Path) -> list[AgentPluginState]:
     """Return component and lifecycle state for discovered plugins."""
-    states: list[AgentPluginState] = []
+    return [
-    for plugin in _discover_agent_plugins(workspace):
-        states.append(
         AgentPluginState(
             plugin=plugin,
             mcp_servers=tuple(sorted(_plugin_mcp_servers(workspace, plugin))),
@@ -184,8 +162,8 @@ def discover_agent_plugin_states(workspace: Path) -> list[AgentPluginState]:
             setup_required=bool(plugin.install_command)
             and _setup_version(workspace, plugin.name) != (plugin.version or "unknown"),
         )
-        )
+        for plugin in _discover_agent_plugins(workspace)
-    return states
+    ]
 
 
 def set_agent_plugin_enabled(workspace: Path, name: str, enabled: bool) -> AgentPlugin:
@@ -212,9 +190,7 @@ def _string(value: object) -> str:
 
 
 def _string_tuple(value: object) -> tuple[str, ...]:
-    if not isinstance(value, list):
+    items = cast(list[object], value) if isinstance(value, list) else []
-        return ()
-    items = cast(list[object], value)
     return tuple(item.strip() for item in items if isinstance(item, str) and item.strip())
 
 
@@ -230,16 +206,20 @@ def _plugin_logo(value: object, plugin_root: Path) -> Path | None:
         logger.warning("Ignoring invalid Agent Plugin logo in '{}'", plugin_root)
         return None
     logo = _contained_file(plugin_root / value[2:], plugin_root)
-    if logo is None or logo.suffix.lower() not in _LOGO_SUFFIXES:
+    try:
+        data = logo.read_bytes() if logo is not None else b""
+        suffix = logo.suffix.lower() if logo is not None else ""
+        valid = (
+            suffix == ".png" and data.startswith(b"\x89PNG\r\n\x1a\n")
+            or suffix in {".jpg", ".jpeg"} and data.startswith(b"\xff\xd8\xff")
+            or suffix == ".webp" and data.startswith(b"RIFF") and data[8:12] == b"WEBP"
+        )
+        if valid and len(data) <= _MAX_LOGO_BYTES:
+            return logo
+    except OSError:
+        pass
     logger.warning("Ignoring invalid Agent Plugin logo in '{}'", plugin_root)
     return None
-    try:
-        if logo.stat().st_size > _MAX_LOGO_BYTES:
-            logger.warning("Ignoring oversized Agent Plugin logo in '{}'", plugin_root)
-            return None
-    except OSError:
-        return None
-    return logo
 
 
 def _install_command(value: object, plugin_root: Path) -> tuple[str, ...]:
@@ -263,17 +243,9 @@ def _install_command(value: object, plugin_root: Path) -> tuple[str, ...]:
 
 
 def _plugin_mcp_servers(workspace: Path, plugin: AgentPlugin) -> dict[str, MCPServerConfig]:
-    path = _contained_file(plugin.root / "mcp.json", plugin.root)
+    payload = _read_object(plugin.root / "mcp.json", plugin.root)
-    if path is None:
+    if payload is None:
         return {}
-    try:
-        value = cast(object, json.loads(path.read_text(encoding="utf-8")))
-    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
-        logger.warning("Ignoring invalid MCP component for Agent Plugin '{}': {}", plugin.name, exc)
-        return {}
-    if not isinstance(value, dict):
-        return {}
-    payload = cast(dict[str, Any], value)
     raw_servers = payload.get("mcpServers")
     if payload.get("$schema") != AGENT_PLUGIN_MCP_SCHEMA or not isinstance(raw_servers, dict):
         logger.warning("Ignoring invalid MCP component for Agent Plugin '{}'", plugin.name)
@@ -296,41 +268,30 @@ def _plugin_mcp_servers(workspace: Path, plugin: AgentPlugin) -> dict[str, MCPSe
 def _plugin_mcp_server(raw: object, root: Path, data: Path) -> MCPServerConfig | None:
     if not isinstance(raw, dict):
         return None
-    payload = cast(dict[str, Any], raw)
+    payload = cast(dict[str, object], raw)
-    if payload.get("type") != "stdio" or payload.keys() - _MCP_SERVER_FIELDS:
+    if payload.keys() - _MCP_SERVER_FIELDS:
         return None
-    command = _stdio_command(payload.get("command"), root)
+    try:
-    args = payload.get("args", [])
+        server = MCPServerConfig.model_validate(payload)
-    env = payload.get("env", {})
+    except ValidationError:
+        return None
+    command = _stdio_command(server.command, root)
     cwd = _stdio_cwd(payload.get("cwd"), root, data)
-    if (
+    if server.type != "stdio" or command is None or cwd is None:
-        command is None
-        or not isinstance(args, list)
-        or not all(isinstance(item, str) for item in cast(list[object], args))
-        or not isinstance(env, dict)
-        or cwd is None
-    ):
         return None
-    env_payload = cast(dict[object, object], env)
+    if {"PLUGIN_ROOT", "PLUGIN_DATA"} & server.env.keys():
-    if any(
-        not isinstance(key, str)
-        or key in {"PLUGIN_ROOT", "PLUGIN_DATA"}
-        or not isinstance(value, str)
-        for key, value in env_payload.items()
-    ):
         return None
-    string_env = cast(dict[str, str], env)
+    return server.model_copy(
-    replacements = {"${PLUGIN_ROOT}": str(root), "${PLUGIN_DATA}": str(data)}
+        update={
-    return MCPServerConfig(
+            "command": command,
-        type="stdio",
+            "args": [_expand(item, root, data) for item in server.args],
-        command=command,
+            "env": {
-        args=[_expand(item, replacements) for item in cast(list[str], args)],
+                **{key: _expand(value, root, data) for key, value in server.env.items()},
-        env={
-            **{key: _expand(value, replacements) for key, value in string_env.items()},
                 "PLUGIN_ROOT": str(root),
                 "PLUGIN_DATA": str(data),
             },
-        cwd=str(cwd),
+            "cwd": str(cwd),
+        }
     )
 
 
@@ -365,10 +326,8 @@ def _stdio_cwd(value: object, root: Path, data: Path) -> Path | None:
     return None
 
 
-def _expand(value: str, replacements: dict[str, str]) -> str:
+def _expand(value: str, root: Path, data: Path) -> str:
-    for token, replacement in replacements.items():
+    return value.replace("${PLUGIN_ROOT}", str(root)).replace("${PLUGIN_DATA}", str(data))
-        value = value.replace(token, replacement)
-    return value
 
 
 def _plugin_data_dir(workspace: Path, name: str, *, create: bool) -> Path:
@@ -408,12 +367,7 @@ def _setup_version(workspace: Path, name: str) -> str:
 
 
 def _write_state(path: Path, value: str) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(value, encoding="utf-8")
-    path.parent.chmod(0o700)
-    temporary = path.with_name(f".{path.name}.tmp")
-    temporary.write_text(value, encoding="utf-8")
-    temporary.chmod(0o600)
-    temporary.replace(path)
     path.chmod(0o600)
 
 
@@ -441,12 +395,8 @@ def _run_install(plugin: AgentPlugin, data: Path) -> None:
 
 
 def _discover_plugin_skills(plugin_name: str, plugin_root: Path) -> list[AgentPluginSkill]:
-    skills_root = plugin_root / "skills"
+    skills_root = _contained_directory(plugin_root / "skills", plugin_root)
-    if not skills_root.exists():
+    if skills_root is None:
-        return []
-    resolved_skills_root = _contained_directory(skills_root, plugin_root)
-    if resolved_skills_root is None:
-        logger.warning("Ignoring invalid skills component in Agent Plugin '{}'", plugin_name)
         return []
 
     try:
@@ -457,7 +407,7 @@ def _discover_plugin_skills(plugin_name: str, plugin_root: Path) -> list[AgentPl
 
     skills: list[AgentPluginSkill] = []
     for candidate in candidates:
-        skill_root = _contained_directory(candidate, resolved_skills_root)
+        skill_root = _contained_directory(candidate, skills_root)
         if skill_root is None:
             continue
         skill_file = _contained_file(skill_root / "SKILL.md", plugin_root)
@@ -474,31 +424,14 @@ def _valid_skill(path: Path, directory_name: str, plugin_name: str) -> bool:
         content = path.read_text(encoding="utf-8")
     except (OSError, UnicodeError):
         return False
-    match = _SKILL_FRONTMATTER.match(content)
+    metadata = parse_skill_metadata(content)
-    if match is None:
+    if metadata is None:
         logger.warning("Ignoring Agent Plugin '{}' skill '{}': invalid frontmatter", plugin_name, directory_name)
         return False
-    try:
+    if not valid_skill_metadata(metadata, directory_name):
-        metadata = cast(object, yaml.safe_load(match.group(1)))
-    except yaml.YAMLError:
-        metadata = None
-    if not isinstance(metadata, dict):
-        logger.warning("Ignoring Agent Plugin '{}' skill '{}': invalid frontmatter", plugin_name, directory_name)
-        return False
-    payload = cast(dict[object, object], metadata)
-    name = payload.get("name")
-    description = payload.get("description")
-    valid = (
-        name == directory_name
-        and isinstance(name, str)
-        and len(name) <= 64
-        and _SKILL_NAME.fullmatch(name) is not None
-        and isinstance(description, str)
-        and 1 <= len(description.strip()) <= 1024
-    )
-    if not valid:
         logger.warning("Ignoring Agent Plugin '{}' skill '{}': invalid metadata", plugin_name, directory_name)
-    return valid
+        return False
+    return True
 
 
 def _contained_directory(path: Path, root: Path) -> Path | None:
@@ -509,6 +442,18 @@ def _contained_directory(path: Path, root: Path) -> Path | None:
     return resolved if resolved.is_dir() and resolved.is_relative_to(root) else None
 
 
+def _read_object(path: Path, root: Path) -> dict[str, object] | None:
+    contained = _contained_file(path, root)
+    if contained is None:
+        return None
+    try:
+        value = cast(object, json.loads(contained.read_text(encoding="utf-8")))
+    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
+        logger.warning("Ignoring invalid Agent Plugin component '{}': {}", contained, exc)
+        return None
+    return cast(dict[str, object], value) if isinstance(value, dict) else None
+
+
 def _contained_file(path: Path, root: Path) -> Path | None:
     try:
         resolved = path.resolve(strict=True)

nanobot/agent/skills.py

@@ -17,9 +17,48 @@ _STRIP_SKILL_FRONTMATTER = re.compile(
     r"^---\s*\r?\n(.*?)\r?\n---\s*\r?\n?",
     re.DOTALL,
 )
+_SKILL_NAME = re.compile(r"^(?!.*--)[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$")
+_SKILL_NAME_LINE = re.compile(r"^name\s*:.*$", re.MULTILINE)
 _SKILL_REFERENCE = re.compile(r"(?<![\w$])\$([A-Za-z0-9_-]+)")
 
 
+def parse_skill_metadata(content: str) -> dict[str, object] | None:
+    """Parse a skill document's YAML frontmatter."""
+    if not (match := _STRIP_SKILL_FRONTMATTER.match(content)):
+        return None
+    try:
+        parsed = yaml.safe_load(match.group(1))
+    except yaml.YAMLError:
+        return None
+    if not isinstance(parsed, dict):
+        return None
+    return {str(key): value for key, value in cast(dict[object, object], parsed).items()}
+
+
+def valid_skill_metadata(metadata: dict[str, object], name: str) -> bool:
+    """Return whether metadata satisfies the Agent Skills identity contract."""
+    description = metadata.get("description")
+    return (
+        metadata.get("name") == name
+        and len(name) <= 64
+        and _SKILL_NAME.fullmatch(name) is not None
+        and isinstance(description, str)
+        and 1 <= len(description.strip()) <= 1024
+    )
+
+
+def normalize_skill_document(content: str, name: str) -> str | None:
+    """Return a valid skill document with a canonical name."""
+    match = _STRIP_SKILL_FRONTMATTER.match(content)
+    metadata = parse_skill_metadata(content)
+    if match is None or metadata is None or not valid_skill_metadata(metadata | {"name": name}, name):
+        return None
+    frontmatter, replaced = _SKILL_NAME_LINE.subn(f"name: {name}", match.group(1), count=1)
+    if not replaced:
+        frontmatter = f"name: {name}\n{frontmatter}"
+    return f"---\n{frontmatter.strip()}\n---\n\n{content[match.end():].lstrip()}"
+
+
 class SkillsLoader:
     """
     Loader for agent skills.
@@ -291,21 +330,4 @@ class SkillsLoader:
         Returns:
             Metadata dict or None.
         """
-        content = self.load_skill(name)
+        return parse_skill_metadata(self.load_skill(name) or "")
-        if not content or not content.startswith("---"):
-            return None
-        match = _STRIP_SKILL_FRONTMATTER.match(content)
-        if not match:
-            return None
-        try:
-            parsed = yaml.safe_load(match.group(1))
-        except yaml.YAMLError:
-            return None
-        if not isinstance(parsed, dict):
-            return None
-        # yaml.safe_load returns native types (int, bool, list, etc.);
-        # keep values as-is so downstream consumers get correct types.
-        metadata: dict[str, object] = {}
-        for key, value in cast(dict[object, object], parsed).items():
-            metadata[str(key)] = value
-        return metadata

nanobot/apps/cli/service.py

@@ -18,9 +18,9 @@ from typing import Any, cast
 from urllib.parse import urlparse
 
 import httpx
-import yaml
 from loguru import logger
 
+from nanobot.agent.skills import normalize_skill_document
 from nanobot.apps.protocol import app_manifest, compact_dict
 from nanobot.config.paths import get_runtime_subdir
 from nanobot.security.workspace_policy import is_path_within
@@ -43,8 +43,6 @@ _MAX_ARTIFACT_REPORT = 12
 _SAFE_NAME_RE = re.compile(r"[^a-z0-9_-]+")
 _SAFE_NPM_DIR_RE = re.compile(r"^[a-z0-9._-]+$", re.IGNORECASE)
 _MENTION_RE = re.compile(r"(^|[\s([{])@([a-z0-9_-]+)\b", re.IGNORECASE)
-_SKILL_FRONTMATTER_RE = re.compile(r"^---\s*\r?\n(.*?)\r?\n---\s*\r?\n?", re.DOTALL)
-_SKILL_NAME_LINE_RE = re.compile(r"^name\s*:.*$", re.MULTILINE)
 _SHELL_META_CHARS = ("|", "&&", "||", ";", "$(", "`", ">", "<")
 _ENDORSEMENT_WORD_RE = re.compile(r"\bofficial\s+", re.IGNORECASE)
 _ARTIFACT_EXTENSIONS = frozenset({
@@ -232,11 +230,11 @@ def _plugin_skill_relative_path(name: str) -> str:
 
 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))
+    canonical = _plugin_skill_relative_path(name)
-    legacy = Path("skills") / _legacy_skill_name(name) / "SKILL.md"
+    legacy = f"skills/{_legacy_skill_name(name)}/SKILL.md"
     if not (workspace / canonical).is_file() and (workspace / legacy).is_file():
-        return legacy.as_posix()
+        return legacy
-    return canonical.as_posix()
+    return canonical
 
 
 def _has_shell_meta(command: str) -> bool:
@@ -1093,31 +1091,12 @@ Use the `run_cli_app` tool with `name="{name}"` for command execution. Do not in
                     return "".join(lines[: index + 1]) + "\n" + note + "\n" + "".join(lines[index + 1 :])
         return note + "\n" + content
 
-    def _normalise_skill(self, content: str, app: dict[str, Any]) -> str:
-        """Give a catalog skill the identity required by its plugin directory."""
-        match = _SKILL_FRONTMATTER_RE.match(content)
-        if match is None:
-            return self._fallback_skill(app)
-        try:
-            metadata = _as_object_dict(cast(object, yaml.safe_load(match.group(1))))
-        except yaml.YAMLError:
-            return self._fallback_skill(app)
-        description = metadata.get("description") if metadata is not None else None
-        if not isinstance(description, str) or not 1 <= len(description.strip()) <= 1024:
-            return self._fallback_skill(app)
-
-        name = _safe_skill_name(str(app["name"]))
-        frontmatter, replaced = _SKILL_NAME_LINE_RE.subn(f"name: {name}", match.group(1), count=1)
-        if not replaced:
-            frontmatter = f"name: {name}\n{frontmatter}"
-        body = content[match.end():].lstrip()
-        return f"---\n{frontmatter.strip()}\n---\n\n{body}"
-
     def install_skill(self, app: dict[str, Any]) -> Path:
-        path = self.workspace / _plugin_skill_relative_path(str(app["name"]))
+        name = str(app["name"])
+        path = self.workspace / _plugin_skill_relative_path(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)
+        content = normalize_skill_document(content, _safe_skill_name(name)) or self._fallback_skill(app)
         content = self._with_nanobot_skill_note(content, app)
         path.write_text(content, encoding="utf-8")
         plugin_root = path.parents[2]

nanobot/webui/mcp_presets_api.py

@@ -57,13 +57,6 @@ _MAX_TEST_TOOLS = 16
 _DEFAULT_TEST_TIMEOUT = 20
 _DEFAULT_CUSTOM_TIMEOUT = 30
 _CUSTOM_ACTIONS = {"custom", "import", "import-cursor", "tools"}
-_PLUGIN_LOGO_MIME_TYPES = {
-    ".jpeg": "image/jpeg",
-    ".jpg": "image/jpeg",
-    ".png": "image/png",
-    ".webp": "image/webp",
-}
-
 McpReload = Callable[[], Awaitable[dict[str, Any]]]
 
 
@@ -859,16 +852,9 @@ def _plugin_logo_data_url(path: Path | None) -> str | None:
         data = path.read_bytes()
     except OSError:
         return None
-    suffix = path.suffix.lower()
-    valid = (
-        suffix == ".png" and data.startswith(b"\x89PNG\r\n\x1a\n")
-        or suffix in {".jpg", ".jpeg"} and data.startswith(b"\xff\xd8\xff")
-        or suffix == ".webp" and data.startswith(b"RIFF") and data[8:12] == b"WEBP"
-    )
-    if not valid:
-        return None
     encoded = base64.b64encode(data).decode("ascii")
-    return f"data:{_PLUGIN_LOGO_MIME_TYPES[suffix]};base64,{encoded}"
+    image_format = path.suffix.lower().lstrip(".").replace("jpg", "jpeg")
+    return f"data:image/{image_format};base64,{encoded}"
 
 
 def _agent_plugin_payload(state: AgentPluginState) -> dict[str, Any]:

tests/agent/test_agent_plugins.py

@@ -40,6 +40,10 @@ def _write_skill(root: Path, name: str, *, description: str = "Plugin skill.") -
     return skill
 
 
+def _manifest(name: str, **fields: object) -> dict[str, object]:
+    return {"$schema": AGENT_PLUGIN_SCHEMA, "name": name, **fields}
+
+
 def _write_plugin(
     workspace: Path,
     directory: str,
@@ -49,24 +53,25 @@ def _write_plugin(
 ) -> Path:
     root = workspace / "plugins" / directory
     root.mkdir(parents=True)
-    payload = manifest or {
+    payload = manifest or _manifest(name or directory)
-        "$schema": AGENT_PLUGIN_SCHEMA,
-        "name": name or directory,
-    }
     (root / "plugin.json").write_text(json.dumps(payload), encoding="utf-8")
     return root
 
 
+def _write_mcp(root: Path, servers: dict[str, object], **fields: object) -> None:
+    payload = {"$schema": AGENT_PLUGIN_MCP_SCHEMA, "mcpServers": servers, **fields}
+    (root / "mcp.json").write_text(json.dumps(payload), encoding="utf-8")
+
+
 def _write_setup_plugin(workspace: Path) -> tuple[Path, Path]:
     plugin = _write_plugin(
         workspace,
         "desktop",
-        manifest={
+        manifest=_manifest(
-            "$schema": AGENT_PLUGIN_SCHEMA,
+            "desktop",
-            "name": "desktop",
+            version="1.2.3",
-            "version": "1.2.3",
+            extensions={"dev.nanobot": {"installCommand": ["./bin/install"]}},
-            "extensions": {"dev.nanobot": {"installCommand": ["./bin/install"]}},
+        ),
-        },
     )
     executable = plugin / "bin" / "install"
     executable.parent.mkdir()
@@ -150,14 +155,13 @@ def test_unknown_manifest_fields_and_non_object_extensions_are_ignored(tmp_path:
     plugin = _write_plugin(
         tmp_path,
         "demo",
-        manifest={
+        manifest=_manifest(
-            "$schema": AGENT_PLUGIN_SCHEMA,
+            "demo",
-            "name": "demo",
+            futureField=True,
-            "futureField": True,
+            author=None,
-            "author": None,
+            keywords=None,
-            "keywords": None,
+            extensions="invalid but non-fatal",
-            "extensions": "invalid but non-fatal",
+        ),
-        },
     )
     _write_skill(plugin, "example")
     set_agent_plugin_enabled(tmp_path, "demo", True)
@@ -169,11 +173,10 @@ def test_agent_plugin_discovers_contained_raster_logo(tmp_path: Path) -> None:
     plugin = _write_plugin(
         tmp_path,
         "demo",
-        manifest={
+        manifest=_manifest(
-            "$schema": AGENT_PLUGIN_SCHEMA,
+            "demo",
-            "name": "demo",
+            extensions={"dev.nanobot": {"logo": "./assets/icon.png"}},
-            "extensions": {"dev.nanobot": {"logo": "./assets/icon.png"}},
+        ),
-        },
     )
     assets = plugin / "assets"
     assets.mkdir()
@@ -187,11 +190,10 @@ def test_agent_plugin_logo_cannot_escape_package(tmp_path: Path) -> None:
     plugin = _write_plugin(
         tmp_path,
         "demo",
-        manifest={
+        manifest=_manifest(
-            "$schema": AGENT_PLUGIN_SCHEMA,
+            "demo",
-            "name": "demo",
+            extensions={"dev.nanobot": {"logo": "./assets/icon.png"}},
-            "extensions": {"dev.nanobot": {"logo": "./assets/icon.png"}},
+        ),
-        },
     )
     assets = plugin / "assets"
     assets.mkdir()
@@ -295,12 +297,9 @@ def test_plugin_mcp_requires_explicit_enable(tmp_path: Path) -> None:
     executable = plugin / "bin" / "server"
     executable.parent.mkdir()
     executable.write_text("#!/bin/sh\n", encoding="utf-8")
-    (plugin / "mcp.json").write_text(
+    _write_mcp(
-        json.dumps(
+        plugin,
         {
-                "$schema": AGENT_PLUGIN_MCP_SCHEMA,
-                "futureField": True,
-                "mcpServers": {
             "desktop": {
                 "type": "stdio",
                 "command": "./bin/server",
@@ -308,9 +307,7 @@ def test_plugin_mcp_requires_explicit_enable(tmp_path: Path) -> None:
                 "cwd": "${PLUGIN_ROOT}",
             }
         },
-            }
+        futureField=True,
-        ),
-        encoding="utf-8",
     )
 
     assert agent_plugin_mcp_servers(tmp_path) == {}
@@ -385,18 +382,13 @@ def test_invalid_plugin_mcp_entries_do_not_block_valid_servers(tmp_path: Path) -
     executable = plugin / "bin" / "server"
     executable.parent.mkdir()
     executable.write_text("#!/bin/sh\n", encoding="utf-8")
-    (plugin / "mcp.json").write_text(
+    _write_mcp(
-        json.dumps(
+        plugin,
         {
-                "$schema": AGENT_PLUGIN_MCP_SCHEMA,
-                "mcpServers": {
             "public-http": {"type": "streamable-http", "url": "http://example.com/mcp"},
             "local": {"type": "stdio", "command": "./bin/server"},
             "escape": {"type": "stdio", "command": "../outside"},
         },
-            }
-        ),
-        encoding="utf-8",
     )
     set_agent_plugin_enabled(tmp_path, "network", True)
 

tests/cli_apps/test_service.py

@@ -515,14 +515,7 @@ def test_install_records_available_cli_without_reinstalling(
     assert "entry_point_available" in payload["last_action"]["verification"]
     installed = json.loads(manager.installed_path.read_text(encoding="utf-8"))["apps"]
     assert installed["feishu"]["entry_point_path"] == str(resolved)
-    skill = (
+    skill = manager.workspace / "plugins/cli-app-feishu/skills/cli-app-feishu/SKILL.md"
-        manager.workspace
-        / "plugins"
-        / "cli-app-feishu"
-        / "skills"
-        / "cli-app-feishu"
-        / "SKILL.md"
-    )
     assert skill.is_file()
     assert 'run_cli_app` tool with `name="feishu"' in skill.read_text(encoding="utf-8")
 

tests/webui/test_mcp_presets_api.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import asyncio
 import json
 import subprocess
+from functools import partial
 from pathlib import Path
 
 import pytest
@@ -132,36 +133,22 @@ def test_agent_plugin_reuses_mcp_catalog_and_runtime_action(
     async def reload() -> dict[str, object]:
         return {"ok": True, "message": "MCP reloaded.", "requires_restart": False}
 
+    plugin_action = partial(
+        mcp_presets_settings_action,
+        query={"name": ["plugin-desktop"]},
+    )
     with pytest.raises(McpPresetError, match="restricted") as restricted:
-        asyncio.run(
+        asyncio.run(plugin_action("enable", remote=True))
-            mcp_presets_settings_action(
-                "enable",
-                {"name": ["plugin-desktop"]},
-                remote=True,
-            )
-        )
     assert restricted.value.status == 403
 
-    enabled = asyncio.run(
+    enabled = asyncio.run(plugin_action("enable", reload_mcp=reload))
-        mcp_presets_settings_action(
-            "enable",
-            {"name": ["plugin-desktop"]},
-            reload_mcp=reload,
-        )
-    )
     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
 
-    disabled = asyncio.run(
+    disabled = asyncio.run(plugin_action("disable", reload_mcp=reload))
-        mcp_presets_settings_action(
-            "disable",
-            {"name": ["plugin-desktop"]},
-            reload_mcp=reload,
-        )
-    )
     disabled_row = next(item for item in disabled["presets"] if item["name"] == "plugin-desktop")
     assert disabled_row["installed"] is True
     assert disabled_row["configured"] is True
@@ -169,12 +156,7 @@ def test_agent_plugin_reuses_mcp_catalog_and_runtime_action(
     assert disabled_row["status"] == "disabled"
 
     with pytest.raises(McpPresetError, match="enable and disable"):
-        asyncio.run(
+        asyncio.run(plugin_action("remove"))
-            mcp_presets_settings_action(
-                "remove",
-                {"name": ["plugin-desktop"]},
-            )
-        )
 
 
 def test_explicit_mcp_config_wins_over_plugin_catalog_name(