perf(plugins): cache verified skill roots

nanobot/agent/plugins.py

@@ -23,6 +23,7 @@ AGENT_PLUGIN_MCP_SCHEMA = "https://agent-plugins.org/schemas/1.0.0/mcp.schema.js
 _PLUGIN_NAME = re.compile(r"^(?!.*(?:--|\.\.))[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?$")
 _MCP_SERVER_FIELDS = {"type", "command", "args", "env", "cwd"}
 _MAX_LOGO_BYTES = 256 * 1024
+_SKILL_CACHE: dict[tuple[Path, Path], tuple[tuple[str, Path], ...]] = {}
 
 
 @dataclass(frozen=True)
@@ -64,13 +65,35 @@ def _installed_plugins(workspace: Path) -> list[AgentPlugin]:
 
 
 def enabled_agent_plugin_skills(workspace: Path) -> list[tuple[str, Path]]:
-    """Return skills from plugins the user has explicitly enabled."""
-    return [
+    """Verify and return skills from plugins the user has explicitly enabled."""
+    skills = [
         skill
         for plugin in _installed_plugins(workspace)
         if _enabled(workspace, plugin)
         for skill in _discover_plugin_skills(plugin.name, plugin.root)
     ]
+    _SKILL_CACHE[_skill_cache_key(workspace)] = tuple(skills)
+    return skills
+
+
+def enabled_agent_plugin_skill_dirs(workspace: Path) -> tuple[Path, ...]:
+    """Return the last verified skill roots, verifying once on a cache miss."""
+    key = _skill_cache_key(workspace)
+    skills = _SKILL_CACHE.get(key)
+    if skills is None:
+        skills = tuple(enabled_agent_plugin_skills(workspace))
+    return tuple(path.parent for _name, path in skills)
+
+
+def _skill_cache_key(workspace: Path) -> tuple[Path, Path]:
+    return (
+        workspace.expanduser().resolve(),
+        get_config_path().expanduser().resolve(),
+    )
+
+
+def _invalidate_skill_cache(workspace: Path) -> None:
+    _SKILL_CACHE.pop(_skill_cache_key(workspace), None)
 
 
 def _load_manifest(plugin_root: Path) -> AgentPlugin | None:
@@ -155,6 +178,7 @@ def set_agent_plugin_enabled(workspace: Path, name: str, enabled: bool) -> None:
         marker.chmod(0o600)
     else:
         marker.unlink(missing_ok=True)
+    _invalidate_skill_cache(workspace)
 
 
 def _string(value: object) -> str:
@@ -314,6 +338,8 @@ def _enabled(workspace: Path, plugin: AgentPlugin) -> bool:
         current = marker.read_text(encoding="utf-8")
         activation = _activation_marker(plugin)
         if activation is None:
+            marker.unlink(missing_ok=True)
+            _invalidate_skill_cache(workspace)
             return False
         if current == activation:
             return True
@@ -321,8 +347,11 @@ def _enabled(workspace: Path, plugin: AgentPlugin) -> bool:
             marker.write_text(activation, encoding="utf-8")
             marker.chmod(0o600)
             return True
+        marker.unlink(missing_ok=True)
+        _invalidate_skill_cache(workspace)
         return False
     except OSError:
+        _invalidate_skill_cache(workspace)
         return False
 
 

nanobot/agent/tools/filesystem.py

@@ -150,13 +150,12 @@ class _FsTool(Tool):
     def _resolve_read(self, path: str) -> Path:
         plugin_skill_dirs: list[Path] = []
         if self._workspace is not None:
-            from nanobot.agent.plugins import enabled_agent_plugin_skills
+            from nanobot.agent.plugins import enabled_agent_plugin_skill_dirs
 
             try:
-                plugin_skill_dirs = [
-                    skill.parent
-                    for _name, skill in enabled_agent_plugin_skills(Path(self._workspace))
-                ]
+                plugin_skill_dirs = list(
+                    enabled_agent_plugin_skill_dirs(Path(self._workspace))
+                )
             except (OSError, RuntimeError):
                 pass
         return self._resolve_with_extra(

tests/agent/test_agent_plugins.py

@@ -10,6 +10,7 @@ from nanobot.agent.plugins import (
     AGENT_PLUGIN_SCHEMA,
     agent_plugin_mcp_servers,
     discover_agent_plugins,
+    enabled_agent_plugin_skill_dirs,
     enabled_agent_plugin_skills,
     set_agent_plugin_enabled,
 )
@@ -206,7 +207,7 @@ def test_plugin_mcp_namespaces_cannot_shadow_plugin_identities(tmp_path: Path) -
 
 @pytest.mark.asyncio
 async def test_restricted_project_can_read_only_enabled_plugin_skill(
-    tmp_path: Path,
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     agent_workspace = tmp_path / "agent"
     project = tmp_path / "project"
@@ -222,6 +223,15 @@ async def test_restricted_project_can_read_only_enabled_plugin_skill(
     read_tool = ReadFileTool.create(ctx)
     write_tool = WriteFileTool.create(ctx)
     set_agent_plugin_enabled(agent_workspace, "demo", True)
+    activation_checks = 0
+    activation_marker = agent_plugins._activation_marker
+
+    def count_activation_checks(plugin: agent_plugins.AgentPlugin) -> str | None:
+        nonlocal activation_checks
+        activation_checks += 1
+        return activation_marker(plugin)
+
+    monkeypatch.setattr(agent_plugins, "_activation_marker", count_activation_checks)
     scope = validate_workspace_scope_payload(
         {"project_path": str(project), "access_mode": "restricted"},
         default_workspace=agent_workspace,
@@ -231,6 +241,7 @@ async def test_restricted_project_can_read_only_enabled_plugin_skill(
     token = bind_workspace_scope(scope)
     try:
         read_result = await read_tool.execute(path=str(resource))
+        repeated_read_result = await read_tool.execute(path=str(resource))
         write_result = await write_tool.execute(path=str(resource), content="changed")
         set_agent_plugin_enabled(agent_workspace, "demo", False)
         disabled_result = await read_tool.execute(path=str(resource))
@@ -238,6 +249,8 @@ async def test_restricted_project_can_read_only_enabled_plugin_skill(
         reset_workspace_scope(token)
 
     assert "plugin reference" in read_result
+    assert "File unchanged since last read" in repeated_read_result
+    assert activation_checks == 1
     assert "outside allowed directory" in write_result
     assert "outside allowed directory" in disabled_result
     assert resource.read_text(encoding="utf-8") == "plugin reference"
@@ -330,6 +343,7 @@ def test_plugin_activation_does_not_survive_in_place_code_replacement(
     tmp_path: Path,
 ) -> None:
     plugin = _plugin(tmp_path, "desktop")
+    _skill(plugin / "skills", "demo")
     executable = plugin / "server.py"
     executable.write_text("print('trusted')\n", encoding="utf-8")
     _write_json(
@@ -347,8 +361,10 @@ def test_plugin_activation_does_not_survive_in_place_code_replacement(
     )
     set_agent_plugin_enabled(tmp_path, "desktop", True)
     assert discover_agent_plugins(tmp_path)[0].enabled is True
+    assert enabled_agent_plugin_skill_dirs(tmp_path) == (plugin / "skills" / "demo",)
 
     executable.write_text("print('replacement')\n", encoding="utf-8")
 
     assert discover_agent_plugins(tmp_path)[0].enabled is False
+    assert enabled_agent_plugin_skill_dirs(tmp_path) == ()
     assert agent_plugin_mcp_servers(tmp_path) == {}