core/cli: plugins list/enable/disable commands and /plugins slash

This commit is contained in:
2026-07-24 16:44:44 +08:00
parent 1d718582ac
commit 03e04f9a38
9 changed files with 588 additions and 9 deletions
+159 -2
View File
@@ -35,9 +35,18 @@ if TYPE_CHECKING:
_DEFAULT_DB_FILENAME = "chat.db"
def _load_config(config_path: Path | None) -> ConfigStore:
def _load_config(config_path: Path | None, *, require_providers: bool = True) -> ConfigStore:
"""Load config; optionally skip the interactive “no providers” recovery path.
Plugin management only needs a valid TOML file (and will create one if
missing via :func:`load_config_with_optional_edit` when providers are required
for chat). For plugins, use ``require_providers=False``.
"""
try:
return load_config_with_optional_edit(config_path)
if require_providers:
return load_config_with_optional_edit(config_path)
path = resolve_config_path(config_path)
return config_mod.load(path)
except config_mod.ConfigFormatError as exc:
path = resolve_config_path(config_path)
msg = f"invalid config TOML ({path}): {exc}"
@@ -501,6 +510,154 @@ def providers_cmd(config_path: Path | None) -> None:
_warn_bad_providers(store.bad_providers)
def print_plugins_table(store: ConfigStore) -> None:
"""Print discovered plugins with allowlist status (CLI + slash)."""
from plyngent.tools.plugins import list_plugin_statuses
cfg = store.plugins_config
click.echo(f"config={store.path}")
enable_s = ", ".join(cfg.enable) if cfg.enable else "(none)"
disable_s = ", ".join(cfg.disable) if cfg.disable else "(none)"
click.echo(f"enable=[{enable_s}] disable=[{disable_s}]")
rows = list_plugin_statuses(enable=cfg.enable, disable=cfg.disable)
if not rows:
click.echo("(no plyngent.tools entry points installed)")
return
click.echo("id\tstatus\tpackage\tvalue")
for row in rows:
if row.disabled:
status = "disabled"
elif row.enabled:
status = "enabled"
else:
status = "off"
pkg = row.plugin.package or "-"
if row.plugin.version:
pkg = f"{pkg}@{row.plugin.version}"
click.echo(f"{row.plugin.id}\t{status}\t{pkg}\t{row.plugin.value}")
@main.group("plugins")
def plugins_group() -> None:
"""Manage third-party plugins (``[plugins]`` allowlist in config)."""
@plugins_group.command("list")
@click.option(
"--config",
"config_path",
type=click.Path(path_type=Path, dir_okay=False),
default=None,
help="Path to plyngent.toml.",
)
def plugins_list_cmd(config_path: Path | None) -> None:
"""List installed ``plyngent.tools`` entry points and enable/disable status."""
store = _load_config(config_path, require_providers=False)
print_plugins_table(store)
@plugins_group.command("enable")
@click.argument("name")
@click.option(
"--config",
"config_path",
type=click.Path(path_type=Path, dir_okay=False),
default=None,
help="Path to plyngent.toml.",
)
def plugins_enable_cmd(name: str, config_path: Path | None) -> None:
"""Allowlist a plugin entry-point name (or ``*`` for all) and write config.
Removes the name from ``[plugins].disable`` if present. Takes effect for
new chat sessions (registry is built at chat start).
"""
store = _load_config(config_path, require_providers=False)
try:
_ = store.enable_plugin(name)
except ValueError as exc:
raise click.ClickException(str(exc)) from exc
store.write()
click.echo(f"enabled {name.strip()!r} in {store.path}")
enable_s = ", ".join(store.plugins_config.enable) or "(none)"
disable_s = ", ".join(store.plugins_config.disable) or "(none)"
click.echo(f"enable=[{enable_s}] disable=[{disable_s}]")
@plugins_group.command("disable")
@click.argument("name")
@click.option(
"--config",
"config_path",
type=click.Path(path_type=Path, dir_okay=False),
default=None,
help="Path to plyngent.toml.",
)
def plugins_disable_cmd(name: str, config_path: Path | None) -> None:
"""Block a plugin via ``[plugins].disable`` and write config.
Disable always wins over enable / ``*``. Use ``plugins undeny`` to drop the
block without adding the plugin to enable.
"""
store = _load_config(config_path, require_providers=False)
try:
_ = store.disable_plugin(name)
except ValueError as exc:
raise click.ClickException(str(exc)) from exc
store.write()
click.echo(f"disabled {name.strip()!r} in {store.path}")
enable_s = ", ".join(store.plugins_config.enable) or "(none)"
disable_s = ", ".join(store.plugins_config.disable) or "(none)"
click.echo(f"enable=[{enable_s}] disable=[{disable_s}]")
@plugins_group.command("undeny")
@click.argument("name")
@click.option(
"--config",
"config_path",
type=click.Path(path_type=Path, dir_okay=False),
default=None,
help="Path to plyngent.toml.",
)
def plugins_undeny_cmd(name: str, config_path: Path | None) -> None:
"""Remove *name* from ``[plugins].disable`` only (does not enable it)."""
store = _load_config(config_path, require_providers=False)
try:
_ = store.undeny_plugin(name)
except ValueError as exc:
raise click.ClickException(str(exc)) from exc
store.write()
click.echo(f"undenied {name.strip()!r} in {store.path}")
disable_s = ", ".join(store.plugins_config.disable) or "(none)"
click.echo(f"disable=[{disable_s}]")
@plugins_group.command("clear")
@click.option(
"--config",
"config_path",
type=click.Path(path_type=Path, dir_okay=False),
default=None,
help="Path to plyngent.toml.",
)
@click.option(
"--yes",
"confirm_yes",
is_flag=True,
default=False,
help="Skip confirmation prompt.",
)
def plugins_clear_cmd(config_path: Path | None, *, confirm_yes: bool) -> None:
"""Clear ``[plugins].enable`` and ``disable`` (load no plugins)."""
store = _load_config(config_path, require_providers=False)
needs_confirm = not confirm_yes and (store.plugins_config.enable or store.plugins_config.disable)
if needs_confirm and not click.confirm("Clear all plugin enable/disable entries?", default=False):
raise click.Abort
_ = store.clear_plugins()
store.write()
click.echo(f"cleared plugins lists in {store.path}")
@main.group("config")
def config_group() -> None:
"""Manage plyngent configuration."""
+71
View File
@@ -972,6 +972,77 @@ def tools_cmd(
click.echo(f"tools={'on' if enabled else 'off'}")
def _slash_rebuild_tools_if_on(state: ReplState) -> None:
if state.tools_enabled:
state.rebuild_client()
def _slash_apply_plugin_action(state: ReplState, act: str, token: str) -> None:
if act == "enable":
_ = state.config.enable_plugin(token)
elif act == "disable":
_ = state.config.disable_plugin(token)
elif act == "undeny":
_ = state.config.undeny_plugin(token)
else:
msg = f"unknown plugins action {act!r}"
raise click.UsageError(msg)
state.config.write()
_slash_rebuild_tools_if_on(state)
cfg = state.config.plugins_config
click.echo(f"plugins {act} {token!r} enable={list(cfg.enable)!r} disable={list(cfg.disable)!r}")
@slash.command("plugins")
@click.argument(
"action",
required=False,
type=click.Choice(["list", "enable", "disable", "undeny", "reload", "clear"]),
)
@click.argument("name", required=False)
@click.pass_obj
def plugins_slash_cmd(state: ReplState, action: str | None, name: str | None) -> None:
"""List or change plugin allowlist; reload tools into this session.
``/plugins`` or ``/plugins list`` — installed entry points + config status.
``/plugins enable NAME`` / ``disable NAME`` / ``undeny NAME`` — update
config on disk, then rebuild the tool registry for this session.
``/plugins reload`` — re-read config and rebuild tools without list changes.
``/plugins clear`` — empty enable/disable lists and rebuild.
"""
act = (action or "list").lower()
if act == "list":
from plyngent.cli.app import print_plugins_table
print_plugins_table(state.config)
return
if act == "reload":
state.config.reload()
_slash_rebuild_tools_if_on(state)
cfg = state.config.plugins_config
click.echo(
f"reloaded plugins enable={list(cfg.enable)!r} disable={list(cfg.disable)!r} "
f"tools={'on' if state.tools_enabled else 'off'}"
)
return
if act == "clear":
_ = state.config.clear_plugins()
state.config.write()
_slash_rebuild_tools_if_on(state)
if state.tools_enabled:
click.echo("cleared plugins enable/disable; tools registry rebuilt")
else:
click.echo("cleared plugins enable/disable")
return
if name is None or not name.strip():
msg = f"/plugins {act} requires a plugin entry-point name"
raise click.UsageError(msg)
try:
_slash_apply_plugin_action(state, act, name.strip())
except ValueError as exc:
raise click.UsageError(str(exc)) from exc
@slash.command("yolo")
@click.argument("mode", required=False, type=YOLO_MODE, metavar="[on|off|once]")
@click.pass_obj
+68
View File
@@ -158,6 +158,74 @@ class ConfigStore:
"""Typed plugin allowlist (enable / disable entry-point names)."""
return self._plugins
def set_plugins_enable(self, names: Sequence[str]) -> PluginsConfig:
"""Replace the plugin enable list (in-memory; call :meth:`write` to persist)."""
cleaned = [name.strip() for name in names if name.strip()]
self._plugins = msgspec.structs.replace(self._plugins, enable=cleaned)
return self._plugins
def set_plugins_disable(self, names: Sequence[str]) -> PluginsConfig:
"""Replace the plugin disable list (in-memory; call :meth:`write` to persist)."""
cleaned = [name.strip() for name in names if name.strip()]
self._plugins = msgspec.structs.replace(self._plugins, disable=cleaned)
return self._plugins
def enable_plugin(self, name: str) -> PluginsConfig:
"""Allowlist *name* and remove it from the disable list.
Special case: ``name == "*"`` sets enable to ``["*"]`` (load all).
"""
token = name.strip()
if not token:
msg = "plugin name must not be empty"
raise ValueError(msg)
disable = [n for n in self._plugins.disable if n.strip() and n.strip() != token]
if token == "*":
self._plugins = msgspec.structs.replace(self._plugins, enable=["*"], disable=disable)
return self._plugins
enable = list(self._plugins.enable)
if any(item.strip() == "*" for item in enable):
# Already load-all; only ensure not disabled.
self._plugins = msgspec.structs.replace(self._plugins, disable=disable)
return self._plugins
if token not in enable:
enable.append(token)
self._plugins = msgspec.structs.replace(self._plugins, enable=enable, disable=disable)
return self._plugins
def disable_plugin(self, name: str) -> PluginsConfig:
"""Block *name* via the disable list (wins over enable / ``*``).
Also removes *name* from a non-``*`` enable list so config stays tidy.
"""
token = name.strip()
if not token or token == "*":
msg = "disable requires a concrete plugin entry-point name"
raise ValueError(msg)
enable = list(self._plugins.enable)
if not any(item.strip() == "*" for item in enable) and token in enable:
enable = [n for n in enable if n.strip() != token]
disable = list(self._plugins.disable)
if token not in disable:
disable.append(token)
self._plugins = msgspec.structs.replace(self._plugins, enable=enable, disable=disable)
return self._plugins
def undeny_plugin(self, name: str) -> PluginsConfig:
"""Remove *name* from the disable list only (does not add to enable)."""
token = name.strip()
if not token:
msg = "plugin name must not be empty"
raise ValueError(msg)
disable = [n for n in self._plugins.disable if n.strip() != token]
self._plugins = msgspec.structs.replace(self._plugins, disable=disable)
return self._plugins
def clear_plugins(self) -> PluginsConfig:
"""Clear enable and disable lists (load no plugins)."""
self._plugins = PluginsConfig()
return self._plugins
# -- providers (read/write) --
@property
+85
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import importlib.metadata
from dataclasses import dataclass
from typing import TYPE_CHECKING
from plyngent.tools.catalog import ToolSource, get_catalog, registration_source
@@ -22,6 +23,90 @@ def _iter_entry_points() -> list[importlib.metadata.EntryPoint]:
return list(selected)
@dataclass(frozen=True, slots=True)
class DiscoveredPlugin:
"""One installed ``plyngent.tools`` entry point (not necessarily enabled)."""
id: str
value: str
package: str | None = None
version: str | None = None
@property
def module(self) -> str:
return self.value.split(":", 1)[0] if ":" in self.value else self.value
@dataclass(frozen=True, slots=True)
class PluginStatus:
"""Discovery + allowlist status for CLI listing."""
plugin: DiscoveredPlugin
enabled: bool
disabled: bool
@property
def will_load(self) -> bool:
"""True if current config would load this plugin."""
return self.enabled and not self.disabled
def list_discovered_plugins() -> list[DiscoveredPlugin]:
"""Return installed entry points for :data:`ENTRY_POINT_GROUP`, sorted by id."""
found: list[DiscoveredPlugin] = []
for entry in _iter_entry_points():
dist = entry.dist
package = dist.name if dist is not None else None
version = dist.version if dist is not None else None
found.append(
DiscoveredPlugin(
id=entry.name,
value=entry.value,
package=package,
version=version,
)
)
return sorted(found, key=lambda p: p.id)
def plugin_would_load(
plugin_id: str,
*,
enable: Sequence[str] | None,
disable: Iterable[str] | None = None,
) -> bool:
"""Whether *plugin_id* would be loaded under the given allowlist."""
disabled = {name.strip() for name in (disable or ()) if name.strip()}
if plugin_id in disabled:
return False
allow = resolve_plugin_allowlist(enable)
if allow is None:
return True
return plugin_id in allow
def list_plugin_statuses(
*,
enable: Sequence[str] | None,
disable: Iterable[str] | None = None,
) -> list[PluginStatus]:
"""Discovered plugins with enable/disable/will-load flags from config lists."""
disabled = {name.strip() for name in (disable or ()) if name.strip()}
allow = resolve_plugin_allowlist(enable)
rows: list[PluginStatus] = []
for plugin in list_discovered_plugins():
is_disabled = plugin.id in disabled
is_enabled = True if allow is None else plugin.id in allow
rows.append(
PluginStatus(
plugin=plugin,
enabled=is_enabled and not is_disabled,
disabled=is_disabled,
)
)
return rows
def resolve_plugin_allowlist(plugins: Sequence[str] | None) -> set[str] | None:
"""Return the set of plugin ids to load, or ``None`` meaning all.