feat(plugins): Stage 2.1 — plugin framework and AI tool-use
Introduces a standalone plugin system where every integration lives as an independent Python script in ~/.pyra/plugins/, not hardcoded in core. Plugin framework (src/pyra/plugins/): - base.py: Tool dataclass, PyraPlugin Protocol, BasePlugin helper - loader.py: importlib-based discovery; one bad plugin never crashes pyra - registry.py: singleton aggregating tools, slash commands, system prompts - executor.py: approval gate — scans args, prompts y/N, scans result, logs - install.py: copies bundled_plugins/ to ~/.pyra/plugins/ on install Chat integration: - AI tool-use loop (litellm function calling, up to 10 iterations) - Plugin system prompt additions injected per session - Plugin slash commands merged with static commands CLI additions: - pyra plugin list/install/enable/disable/setup - pyra daemon start/stop/status/restart/install/uninstall (stubs for 2.4) Config: PluginConfig + DaemonConfig added to PyraConfig (backwards-compatible) Bootstrap: ~/.pyra/plugins/ and ~/.pyra/logs/ created on startup Security: tool args and results always injection-scanned; plugin dirs validated with assert_safe_path() before loading (symlink protection) Tests: 37 new tests (loader, registry, executor, plugin isolation security) 161 total, all passing. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,151 @@
|
||||
"""Tests for PluginRegistry aggregation and singleton behavior."""
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from pyra.plugins.base import BasePlugin, Tool
|
||||
from pyra.plugins.registry import PluginRegistry
|
||||
|
||||
|
||||
def _make_plugin_dir(plugins_dir: Path, name: str, plugin_code: str) -> None:
|
||||
d = plugins_dir / name
|
||||
d.mkdir(parents=True)
|
||||
(d / "manifest.json").write_text(json.dumps({"name": name, "version": "1.0.0"}))
|
||||
(d / "plugin.py").write_text(plugin_code)
|
||||
|
||||
|
||||
_ALPHA_PLUGIN = """\
|
||||
from pyra.plugins.base import BasePlugin, Tool
|
||||
|
||||
class _P(BasePlugin):
|
||||
name = "alpha"
|
||||
description = "Alpha plugin"
|
||||
version = "1.0.0"
|
||||
|
||||
def tools(self):
|
||||
return [
|
||||
Tool("alpha_tool", "An alpha tool",
|
||||
{"type": "object", "properties": {}},
|
||||
lambda: "alpha result", requires_approval=False)
|
||||
]
|
||||
|
||||
def slash_commands(self):
|
||||
return {"/alpha": lambda: None}
|
||||
|
||||
def system_prompt_addition(self):
|
||||
return "Alpha is active."
|
||||
|
||||
def get_plugin():
|
||||
return _P()
|
||||
"""
|
||||
|
||||
_BETA_PLUGIN = """\
|
||||
from pyra.plugins.base import BasePlugin, Tool
|
||||
|
||||
class _P(BasePlugin):
|
||||
name = "beta"
|
||||
description = "Beta plugin"
|
||||
version = "1.0.0"
|
||||
|
||||
def tools(self):
|
||||
return [
|
||||
Tool("beta_tool", "A beta tool",
|
||||
{"type": "object", "properties": {}},
|
||||
lambda: "beta result", requires_approval=True)
|
||||
]
|
||||
|
||||
def system_prompt_addition(self):
|
||||
return "Beta is active."
|
||||
|
||||
def get_plugin():
|
||||
return _P()
|
||||
"""
|
||||
|
||||
|
||||
def test_singleton_returns_same_instance(tmp_pyra_home):
|
||||
r1 = PluginRegistry.instance()
|
||||
r2 = PluginRegistry.instance()
|
||||
assert r1 is r2
|
||||
|
||||
|
||||
def test_load_all_only_loads_enabled(tmp_pyra_home, tmp_path):
|
||||
plugins_dir = tmp_path / "plugins"
|
||||
plugins_dir.mkdir()
|
||||
_make_plugin_dir(plugins_dir, "alpha", _ALPHA_PLUGIN)
|
||||
_make_plugin_dir(plugins_dir, "beta", _BETA_PLUGIN)
|
||||
|
||||
registry = PluginRegistry.instance()
|
||||
registry.load_all(plugins_dir, enabled_names=["alpha"])
|
||||
|
||||
names = {p.name for p in registry.get_active_plugins()}
|
||||
assert "alpha" in names
|
||||
assert "beta" not in names
|
||||
|
||||
|
||||
def test_get_all_tools_aggregates(tmp_pyra_home, tmp_path):
|
||||
plugins_dir = tmp_path / "plugins"
|
||||
plugins_dir.mkdir()
|
||||
_make_plugin_dir(plugins_dir, "alpha", _ALPHA_PLUGIN)
|
||||
_make_plugin_dir(plugins_dir, "beta", _BETA_PLUGIN)
|
||||
|
||||
registry = PluginRegistry.instance()
|
||||
registry.load_all(plugins_dir, enabled_names=["alpha", "beta"])
|
||||
|
||||
tool_names = {t.name for t in registry.get_all_tools()}
|
||||
assert "alpha_tool" in tool_names
|
||||
assert "beta_tool" in tool_names
|
||||
|
||||
|
||||
def test_get_slash_commands_aggregates(tmp_pyra_home, tmp_path):
|
||||
plugins_dir = tmp_path / "plugins"
|
||||
plugins_dir.mkdir()
|
||||
_make_plugin_dir(plugins_dir, "alpha", _ALPHA_PLUGIN)
|
||||
|
||||
registry = PluginRegistry.instance()
|
||||
registry.load_all(plugins_dir, enabled_names=["alpha"])
|
||||
|
||||
cmds = registry.get_slash_commands()
|
||||
assert "/alpha" in cmds
|
||||
|
||||
|
||||
def test_get_system_prompt_additions(tmp_pyra_home, tmp_path):
|
||||
plugins_dir = tmp_path / "plugins"
|
||||
plugins_dir.mkdir()
|
||||
_make_plugin_dir(plugins_dir, "alpha", _ALPHA_PLUGIN)
|
||||
_make_plugin_dir(plugins_dir, "beta", _BETA_PLUGIN)
|
||||
|
||||
registry = PluginRegistry.instance()
|
||||
registry.load_all(plugins_dir, enabled_names=["alpha", "beta"])
|
||||
|
||||
additions = registry.get_system_prompt_additions()
|
||||
assert "Alpha is active." in additions
|
||||
assert "Beta is active." in additions
|
||||
|
||||
|
||||
def test_find_tool_returns_correct_tool(tmp_pyra_home, tmp_path):
|
||||
plugins_dir = tmp_path / "plugins"
|
||||
plugins_dir.mkdir()
|
||||
_make_plugin_dir(plugins_dir, "alpha", _ALPHA_PLUGIN)
|
||||
|
||||
registry = PluginRegistry.instance()
|
||||
registry.load_all(plugins_dir, enabled_names=["alpha"])
|
||||
|
||||
tool = registry.find_tool("alpha_tool")
|
||||
assert tool is not None
|
||||
assert tool.name == "alpha_tool"
|
||||
|
||||
|
||||
def test_find_tool_unknown_returns_none(tmp_pyra_home):
|
||||
registry = PluginRegistry.instance()
|
||||
registry.load_all(Path("/nonexistent"), enabled_names=[])
|
||||
assert registry.find_tool("no_such_tool") is None
|
||||
|
||||
|
||||
def test_empty_registry_returns_empty_collections(tmp_pyra_home):
|
||||
registry = PluginRegistry.instance()
|
||||
registry.load_all(Path("/nonexistent"), enabled_names=[])
|
||||
assert registry.get_all_tools() == []
|
||||
assert registry.get_slash_commands() == {}
|
||||
assert registry.get_system_prompt_additions() == ""
|
||||
assert registry.get_active_plugins() == []
|
||||
Reference in New Issue
Block a user