| """
|
| Plugin discovery, validation, collision detection, and dispatch.
|
|
|
| Discovery runs once (JarvisLive.__init__ calls discover_plugins()); the resulting
|
| PluginRegistry is cached for the process lifetime. Enable/disable state is re-read
|
| from config on every call to get_tool_declarations() / run() / list_for_ui(), so
|
| toggling a plugin does not require restarting the app or re-importing anything.
|
| """
|
| from __future__ import annotations
|
|
|
| import importlib.util
|
| import inspect
|
| import re
|
| import sys
|
| import traceback
|
| from dataclasses import dataclass, field
|
| from pathlib import Path
|
| from typing import Callable, Optional
|
|
|
| from memory.config_manager import get_plugin_enabled, get_plugin_config
|
|
|
| _NAME_RE = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]{0,63}$")
|
| _DEFAULT_PARAMS = {"type": "OBJECT", "properties": {}}
|
|
|
|
|
|
|
|
|
| _BEHAVIORS = ("BLOCKING", "NON_BLOCKING")
|
| _SCHEDULING = ("WHEN_IDLE", "SILENT", "INTERRUPT")
|
|
|
|
|
| def _opt_upper(value, allowed: tuple[str, ...]) -> Optional[str]:
|
| v = str(value or "").strip().upper()
|
| return v if v in allowed else None
|
|
|
|
|
| @dataclass
|
| class PluginRecord:
|
| name: str
|
| description: str = ""
|
| parameters: dict = field(default_factory=lambda: dict(_DEFAULT_PARAMS))
|
| run: Optional[Callable] = None
|
| file: str = ""
|
| valid: bool = False
|
| error: str = ""
|
| settings: Optional[dict] = None
|
| behavior: Optional[str] = None
|
| scheduling: Optional[str] = None
|
|
|
|
|
| class PluginRegistry:
|
| def __init__(self, plugins: dict[str, PluginRecord], logger: Callable[[str], None],
|
| notify: Callable[[str], None] | None = None):
|
| self._plugins = plugins
|
| self._all_records: list[PluginRecord] = []
|
| self._logger = logger
|
|
|
|
|
|
|
|
|
| self._notify = notify or (lambda _msg: None)
|
|
|
|
|
| def get_tool_declarations(self) -> list[dict]:
|
| decls = []
|
| for name, rec in self._plugins.items():
|
| if get_plugin_enabled(name):
|
| decl = {
|
| "name": rec.name,
|
| "description": rec.description,
|
| "parameters": rec.parameters,
|
| }
|
| if rec.behavior:
|
| decl["behavior"] = rec.behavior
|
| decls.append(decl)
|
| return decls
|
|
|
| def has(self, name: str) -> bool:
|
| return name in self._plugins
|
|
|
| def scheduling(self, name: str) -> Optional[str]:
|
| """How this plugin's result should re-enter the conversation, if it said."""
|
| rec = self._plugins.get(name)
|
| return rec.scheduling if rec else None
|
|
|
|
|
| def run(self, name: str, parameters: dict, player=None, session_memory=None) -> str:
|
| rec = self._plugins.get(name)
|
| if rec is None or not rec.valid:
|
| return f"Plugin '{name}' is not available."
|
| if not get_plugin_enabled(name):
|
| return f"The '{name}' plugin is currently disabled."
|
| try:
|
| return _call_run(rec.run, parameters, player, session_memory) or "Done."
|
| except Exception as e:
|
| self._logger(f"Plugin '{name}' crashed during run(): {e}")
|
| self._notify(f"Plugin '{name}' failed — see the console for details.")
|
| traceback.print_exc()
|
| return f"Sir, the '{name}' plugin failed: {e}"
|
|
|
|
|
| def settings_schemas(self) -> list[dict]:
|
| """One entry per settings SECTION, for enabled plugins that declare a
|
| PLUGIN_SETTINGS schema. Sections are deduped by namespace so a suite of
|
| plugins sharing one namespace (e.g. the printer trio) shows a single
|
| form. Current stored values are merged in so the UI can pre-fill fields.
|
| """
|
| seen: set[str] = set()
|
| out: list[dict] = []
|
| for name, rec in self._plugins.items():
|
| if not rec.settings or not get_plugin_enabled(name):
|
| continue
|
| ns = rec.settings.get("namespace") or rec.name
|
| if ns in seen:
|
| continue
|
| seen.add(ns)
|
| out.append({
|
| "plugin": rec.name,
|
| "namespace": ns,
|
| "title": rec.settings.get("title") or rec.name,
|
| "fields": rec.settings.get("fields", []),
|
| "values": get_plugin_config(ns),
|
| "action": rec.settings.get("action"),
|
| })
|
| return out
|
|
|
|
|
| def list_for_ui(self) -> list[dict]:
|
| out = []
|
| for rec in self._all_records:
|
| out.append({
|
| "name": rec.name,
|
| "description": rec.description,
|
| "file": rec.file,
|
| "valid": rec.valid,
|
| "error": rec.error,
|
| "enabled": get_plugin_enabled(rec.name) if rec.valid else False,
|
| })
|
| return out
|
|
|
|
|
| def _call_run(run_fn, parameters, player, session_memory):
|
| """Invoke run() passing only the kwargs it actually declares (or all of them
|
| if it has **kwargs), so a minimal `def run(parameters):` plugin still works."""
|
| sig = inspect.signature(run_fn)
|
| has_var_kw = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values())
|
| kwargs = {}
|
| if has_var_kw or "player" in sig.parameters:
|
| kwargs["player"] = player
|
| if has_var_kw or "session_memory" in sig.parameters:
|
| kwargs["session_memory"] = session_memory
|
| return run_fn(parameters, **kwargs)
|
|
|
|
|
| def _validate(module, filename: str) -> PluginRecord:
|
| """Returns a PluginRecord; .valid=False + .error set on any problem. Never raises."""
|
| plugin_meta = getattr(module, "PLUGIN", None)
|
| if not isinstance(plugin_meta, dict):
|
| return PluginRecord(name=Path(filename).stem, file=filename,
|
| error="Missing PLUGIN dict constant.")
|
|
|
| name = plugin_meta.get("name")
|
| if not isinstance(name, str) or not _NAME_RE.match(name):
|
| return PluginRecord(name=str(name or Path(filename).stem), file=filename,
|
| error="PLUGIN['name'] missing or not a valid identifier "
|
| "(letters/digits/underscore, must start with letter/underscore).")
|
|
|
| description = plugin_meta.get("description")
|
| if not isinstance(description, str) or not description.strip():
|
| return PluginRecord(name=name, file=filename,
|
| error="PLUGIN['description'] missing or empty.")
|
|
|
| parameters = plugin_meta.get("parameters", _DEFAULT_PARAMS)
|
| if not isinstance(parameters, dict) or parameters.get("type") != "OBJECT":
|
| return PluginRecord(name=name, file=filename,
|
| error="PLUGIN['parameters'] must be a dict with \"type\": \"OBJECT\".")
|
|
|
| run_fn = getattr(module, "run", None)
|
| if not callable(run_fn):
|
| return PluginRecord(name=name, file=filename,
|
| error="Missing callable run(parameters, ...) function.")
|
|
|
|
|
|
|
| settings = getattr(module, "PLUGIN_SETTINGS", None)
|
| if not (isinstance(settings, dict) and isinstance(settings.get("fields"), list)):
|
| settings = None
|
|
|
| return PluginRecord(name=name, description=description.strip(), parameters=parameters,
|
| run=run_fn, file=filename, valid=True, error="", settings=settings,
|
| behavior=_opt_upper(plugin_meta.get("behavior"), _BEHAVIORS),
|
| scheduling=_opt_upper(plugin_meta.get("scheduling"), _SCHEDULING))
|
|
|
|
|
| def _load_error(path: Path, plugins_dir: Path, exc: Exception) -> str:
|
| """Turn an import failure into something the person who downloaded the file
|
| can act on.
|
|
|
| Plugins are shared one file at a time, but some of them sit on a helper —
|
| anything named with a leading underscore, which this loader deliberately
|
| skips so it is never treated as a plugin of its own. Download the plugin
|
| without its helper and Python reports `No module named 'plugins._x'`, which
|
| is accurate and tells a non-programmer nothing. Naming the missing file, and
|
| saying it belongs next to this one, turns a support question into a
|
| thirty-second fix. Nothing here is specific to any plugin: the helper's name
|
| comes from the exception itself.
|
| """
|
| if isinstance(exc, ModuleNotFoundError):
|
| missing = (getattr(exc, "name", "") or "").split(".")
|
| if len(missing) == 2 and missing[0] == "plugins" and missing[1].startswith("_"):
|
| helper = missing[1] + ".py"
|
| return (f"Needs the shared file '{helper}', which is not in "
|
| f"{plugins_dir.name}/. It comes with this plugin — download "
|
| f"'{helper}' into the same folder as {path.name} and restart.")
|
| if missing and missing[0] not in ("plugins",):
|
| return (f"Needs a package that is not installed: "
|
| f"pip install {missing[0]}")
|
| return f"Failed to load: {exc}"
|
|
|
|
|
| def discover_plugins(plugins_dir: Path, core_tool_names: set[str],
|
| logger: Callable[[str], None] = print,
|
| notify: Callable[[str], None] | None = None,
|
| extra_dirs: Optional[list] = None) -> PluginRegistry:
|
| """
|
| Scans plugins_dir — and any `extra_dirs`, in order — for *.py files (skips
|
| files starting with '_', e.g. __init__.py, _template.py, and any
|
| shared-helper modules an author prefixes with '_').
|
|
|
| Import errors, validation errors, and name collisions are logged and the
|
| offending file is skipped — they NEVER raise out of this function and never
|
| abort the scan of remaining files.
|
|
|
| WHY A SECOND DIRECTORY
|
| The plugin system was already the right idea being used for the wrong
|
| thing. Discovery scanned one directory, once, at startup — so the only
|
| way to add a tool was to commit it and wait out a restart. On a hosted
|
| Space that is ten minutes for a ten-line plugin, which makes editing
|
| core look like the faster path. It is backwards: a plugin exists
|
| precisely so that small things do not have to go through a release.
|
|
|
| So there are two directories. `plugins_dir` is the repository — in
|
| version control, reviewed, the shipped ones. `extra_dirs` is the
|
| persistent volume, written at runtime, and it survives rebuilds
|
| because it outlives the image. Order is deliberate: repository first,
|
| so a runtime plugin can never quietly shadow a reviewed one, and the
|
| loser of a name clash is rejected with both filenames named.
|
| """
|
| plugins_dir.mkdir(parents=True, exist_ok=True)
|
| scan_dirs = [plugins_dir]
|
| for extra in (extra_dirs or []):
|
| try:
|
| Path(extra).mkdir(parents=True, exist_ok=True)
|
| scan_dirs.append(Path(extra))
|
| except Exception as e:
|
| logger(f"[Plugins] cannot use {extra}: {type(e).__name__}: {e}")
|
|
|
| valid: dict[str, PluginRecord] = {}
|
| all_records: list[PluginRecord] = []
|
|
|
| files: list = []
|
| for d in scan_dirs:
|
| files.extend(sorted(d.glob("*.py"), key=lambda p: p.name))
|
| for path in files:
|
| if path.name.startswith("_"):
|
| continue
|
| try:
|
| module_name = f"plugins.{path.stem}"
|
| spec = importlib.util.spec_from_file_location(module_name, path)
|
| if spec is None or spec.loader is None:
|
| raise ImportError("could not build import spec")
|
| module = importlib.util.module_from_spec(spec)
|
| sys.modules[module_name] = module
|
| try:
|
| spec.loader.exec_module(module)
|
| except Exception:
|
| sys.modules.pop(module_name, None)
|
| raise
|
|
|
| rec = _validate(module, path.name)
|
|
|
| if rec.valid and rec.name in core_tool_names:
|
| rec = PluginRecord(name=rec.name, file=path.name,
|
| error=f"Name '{rec.name}' collides with a core tool — rejected.")
|
| elif rec.valid and rec.name in valid:
|
| other = valid[rec.name].file
|
| rec = PluginRecord(name=rec.name, file=path.name,
|
| error=f"Name '{rec.name}' already used by plugin '{other}' — rejected.")
|
|
|
| except Exception as e:
|
| rec = PluginRecord(name=path.stem, file=path.name,
|
| error=_load_error(path, plugins_dir, e))
|
| traceback.print_exc()
|
|
|
| all_records.append(rec)
|
| if rec.valid:
|
| valid[rec.name] = rec
|
| logger(f"Plugin loaded: {rec.name} ({path.name})")
|
| else:
|
| logger(f"Plugin rejected: {path.name} — {rec.error}")
|
|
|
| notify = notify or (lambda _msg: None)
|
| registry = PluginRegistry(valid, logger, notify)
|
| registry._all_records = all_records
|
| rejected = len(all_records) - len(valid)
|
| logger(f"Plugin discovery complete: {len(valid)} active, "
|
| f"{rejected} rejected, {len(all_records)} total.")
|
|
|
|
|
|
|
| if rejected:
|
| notify(f"{rejected} plugin(s) could not be loaded — see the console.")
|
| return registry
|
|
|
