tiphys.plugins

 41class PluginRegistry:
 42    """
 43    Central registry for all plugin extensions.
 44
 45    Manages channels, tools, and hooks registered by plugins.
 46    """
 47
 48    def __init__(self, hook_manager: HookManager | None = None) -> None:
 49        """
 50        Initialize the plugin registry.
 51
 52        Args:
 53            hook_manager: Optional hook manager. Uses global if not provided.
 54        """
 55        self._hook_manager = hook_manager or get_hook_manager()
 56        self._channels: dict[str, ChannelDock] = {}
 57        self._tools: dict[str, ToolDefinition] = {}
 58        self._plugins: dict[str, PluginInfo] = {}
 59
 60    @property
 61    def channels(self) -> dict[str, ChannelDock]:
 62        """Get all registered channels."""
 63        return self._channels
 64
 65    @property
 66    def tools(self) -> dict[str, ToolDefinition]:
 67        """Get all registered tools."""
 68        return self._tools
 69
 70    @property
 71    def hook_manager(self) -> HookManager:
 72        """Get the hook manager."""
 73        return self._hook_manager
 74
 75    def register_channel(self, name: str, channel: ChannelDock) -> None:
 76        """
 77        Register a channel adapter.
 78
 79        Args:
 80            name: Channel name.
 81            channel: The channel adapter.
 82
 83        Raises:
 84            ValueError: If channel name is already registered.
 85        """
 86        if name in self._channels:
 87            raise ValueError(f"Channel '{name}' is already registered")
 88
 89        self._channels[name] = channel
 90
 91    def unregister_channel(self, name: str) -> bool:
 92        """
 93        Unregister a channel.
 94
 95        Args:
 96            name: Channel name to unregister.
 97
 98        Returns:
 99            True if unregistered, False if not found.
100        """
101        if name in self._channels:
102            del self._channels[name]
103            return True
104        return False
105
106    def get_channel(self, name: str) -> ChannelDock | None:
107        """Get a channel by name."""
108        return self._channels.get(name)
109
110    def register_tool(self, tool: ToolDefinition) -> None:
111        """
112        Register a tool.
113
114        Args:
115            tool: The tool definition.
116
117        Raises:
118            ValueError: If tool name is already registered.
119        """
120        if tool.name in self._tools:
121            raise ValueError(f"Tool '{tool.name}' is already registered")
122
123        self._tools[tool.name] = tool
124
125    def unregister_tool(self, name: str) -> bool:
126        """
127        Unregister a tool.
128
129        Args:
130            name: Tool name to unregister.
131
132        Returns:
133            True if unregistered, False if not found.
134        """
135        if name in self._tools:
136            del self._tools[name]
137            return True
138        return False
139
140    def get_tool(self, name: str) -> ToolDefinition | None:
141        """Get a tool by name."""
142        return self._tools.get(name)
143
144    def register_hook(self, hook: Hook) -> None:
145        """
146        Register a hook.
147
148        Args:
149            hook: The hook to register.
150        """
151        self._hook_manager.register(hook)
152
153    def unregister_hook(self, name: str) -> bool:
154        """
155        Unregister a hook.
156
157        Args:
158            name: Hook name to unregister.
159
160        Returns:
161            True if unregistered, False if not found.
162        """
163        return self._hook_manager.unregister(name)
164
165    async def load_plugin(self, path: Path) -> PluginInfo:
166        """
167        Load a plugin from a directory.
168
169        The plugin directory should contain an __init__.py with a
170        `register(api: TiphysPluginApi)` function.
171
172        Args:
173            path: Path to the plugin directory.
174
175        Returns:
176            PluginInfo about the loaded plugin.
177
178        Raises:
179            ValueError: If plugin doesn't have a register function.
180            FileNotFoundError: If plugin path doesn't exist.
181        """
182        from tiphys.plugins.api import TiphysPluginApi
183
184        path = path.resolve()
185
186        if not path.exists():
187            raise FileNotFoundError(f"Plugin path not found: {path}")
188
189        init_file = path / "__init__.py"
190        if not init_file.exists():
191            raise ValueError(f"Plugin missing __init__.py: {path}")
192
193        # Load the module
194        spec = importlib.util.spec_from_file_location(f"chela_plugin_{path.name}", init_file)
195        if spec is None or spec.loader is None:
196            raise ValueError(f"Failed to create module spec for: {path}")
197
198        module = importlib.util.module_from_spec(spec)
199        spec.loader.exec_module(module)
200
201        # Check for register function
202        if not hasattr(module, "register"):
203            raise ValueError(f"Plugin missing register() function: {path}")
204
205        # Create plugin info
206        plugin_name = getattr(module, "__plugin_name__", path.name)
207        info = PluginInfo(
208            name=plugin_name,
209            path=path,
210            version=getattr(module, "__version__", "0.0.0"),
211            description=getattr(module, "__description__", ""),
212            module=module,
213        )
214
215        # Create API and call register
216        api = TiphysPluginApi(self, info)
217        module.register(api)
218
219        # Store plugin info
220        self._plugins[plugin_name] = info
221
222        # Emit plugin load event
223        await self._hook_manager.emit(
224            PluginLoadEvent(plugin_name=plugin_name, plugin_path=str(path))
225        )
226
227        await logger.ainfo(
228            "Plugin loaded",
229            plugin=plugin_name,
230            path=str(path),
231            channels=info.channels,
232            tools=info.tools,
233            hooks=info.hooks,
234        )
235
236        return info
237
238    async def unload_plugin(self, name: str) -> bool:
239        """
240        Unload a plugin.
241
242        Removes all channels, tools, and hooks registered by the plugin.
243
244        Args:
245            name: Plugin name to unload.
246
247        Returns:
248            True if unloaded, False if not found.
249        """
250        info = self._plugins.get(name)
251        if not info:
252            return False
253
254        # Remove channels
255        for channel_name in info.channels:
256            self.unregister_channel(channel_name)
257
258        # Remove tools
259        for tool_name in info.tools:
260            self.unregister_tool(tool_name)
261
262        # Remove hooks
263        for hook_name in info.hooks:
264            self.unregister_hook(hook_name)
265
266        # Remove plugin info
267        del self._plugins[name]
268
269        # Emit unload event
270        await self._hook_manager.emit(PluginUnloadEvent(plugin_name=name))
271
272        await logger.ainfo("Plugin unloaded", plugin=name)
273
274        return True
275
276    def get_plugin(self, name: str) -> PluginInfo | None:
277        """Get plugin info by name."""
278        return self._plugins.get(name)
279
280    def list_plugins(self) -> list[str]:
281        """List all loaded plugin names."""
282        return list(self._plugins.keys())
283
284    async def discover_plugins(self, directory: Path) -> list[PluginInfo]:
285        """
286        Discover and load all plugins in a directory.
287
288        Each subdirectory with an __init__.py is treated as a plugin.
289
290        Args:
291            directory: Directory to scan for plugins.
292
293        Returns:
294            List of loaded plugin info.
295        """
296        if not directory.exists():
297            return []
298
299        loaded: list[PluginInfo] = []
300
301        for item in directory.iterdir():
302            if item.is_dir() and (item / "__init__.py").exists():
303                try:
304                    info = await self.load_plugin(item)
305                    loaded.append(info)
306                except Exception as e:
307                    await logger.aerror(
308                        "Failed to load plugin",
309                        path=str(item),
310                        error=str(e),
311                    )
312
313        return loaded

27@dataclass
28class PluginInfo:
29    """Information about a loaded plugin."""
30
31    name: str
32    path: Path | None = None
33    version: str = "0.0.0"
34    description: str = ""
35    module: Any = None
36    channels: list[str] = field(default_factory=list)
37    tools: list[str] = field(default_factory=list)
38    hooks: list[str] = field(default_factory=list)

 17class TiphysPluginApi:
 18    """
 19    Public API exposed to plugins for registration.
 20
 21    Plugins receive an instance of this class in their `register()` function
 22    and use it to register channels, tools, and hooks.
 23    """
 24
 25    def __init__(
 26        self,
 27        registry: "PluginRegistry",
 28        plugin_info: "PluginInfo",
 29    ) -> None:
 30        """
 31        Initialize the plugin API.
 32
 33        Args:
 34            registry: The plugin registry.
 35            plugin_info: Info about the plugin being registered.
 36        """
 37        self._registry = registry
 38        self._plugin_info = plugin_info
 39
 40    @property
 41    def plugin_name(self) -> str:
 42        """Get the plugin name."""
 43        return self._plugin_info.name
 44
 45    def register_channel(self, name: str, channel: "ChannelDock") -> None:
 46        """
 47        Register a channel adapter.
 48
 49        Args:
 50            name: Channel name (e.g., "telegram", "discord").
 51            channel: The channel adapter instance.
 52
 53        Raises:
 54            ValueError: If channel name is already registered.
 55        """
 56        self._registry.register_channel(name, channel)
 57        self._plugin_info.channels.append(name)
 58
 59    def register_tool(self, tool: "ToolDefinition") -> None:
 60        """
 61        Register a tool.
 62
 63        Args:
 64            tool: The tool definition.
 65
 66        Raises:
 67            ValueError: If tool name is already registered.
 68        """
 69        self._registry.register_tool(tool)
 70        self._plugin_info.tools.append(tool.name)
 71
 72    def register_hook(self, hook: "Hook") -> None:
 73        """
 74        Register a hook for lifecycle events.
 75
 76        Args:
 77            hook: The hook to register.
 78
 79        Raises:
 80            ValueError: If hook name is already registered.
 81        """
 82        self._registry.register_hook(hook)
 83        self._plugin_info.hooks.append(hook.name)
 84
 85    def get_channel(self, name: str) -> "ChannelDock | None":
 86        """
 87        Get a registered channel by name.
 88
 89        Args:
 90            name: Channel name.
 91
 92        Returns:
 93            The channel adapter or None if not found.
 94        """
 95        return self._registry.get_channel(name)
 96
 97    def get_tool(self, name: str) -> "ToolDefinition | None":
 98        """
 99        Get a registered tool by name.
100
101        Args:
102            name: Tool name.
103
104        Returns:
105            The tool definition or None if not found.
106        """
107        return self._registry.get_tool(name)
108
109    def list_channels(self) -> list[str]:
110        """List all registered channel names."""
111        return list(self._registry.channels.keys())
112
113    def list_tools(self) -> list[str]:
114        """List all registered tool names."""
115        return list(self._registry.tools.keys())

320def get_plugin_registry() -> PluginRegistry:
321    """
322    Get the global plugin registry.
323
324    .. deprecated::
325        Use AppContext.plugin_registry instead. This function will be removed
326        in a future version.
327
328    Returns:
329        The global PluginRegistry instance.
330    """
331    import warnings
332
333    from tiphys.context import get_context, has_context
334
335    # If AppContext is available, use it (preferred path)
336    if has_context():
337        warnings.warn(
338            "get_plugin_registry() is deprecated. Use AppContext.plugin_registry instead.",
339            DeprecationWarning,
340            stacklevel=2,
341        )
342        return get_context().plugin_registry  # type: ignore[return-value]
343
344    # Fallback to global singleton for backwards compatibility
345    global _plugin_registry
346    if _plugin_registry is None:
347        _plugin_registry = PluginRegistry()
348    return _plugin_registry