shopdb-flask/shopdb/plugins/__init__.py

"""Plugin manager - main entry point for plugin system."""

from pathlib import Path
from typing import Dict, List, Optional
from flask import Flask
import logging
import re

from .base import BasePlugin, PluginMeta, renderable_settings_routes
from .registry import PluginRegistry, PluginState
from .loader import PluginLoader
from .migrations import PluginMigrationManager
from ..exceptions import PluginError

logger = logging.getLogger(__name__)

__all__ = [
    'PluginManager',
    'BasePlugin',
    'PluginMeta',
    'PluginRegistry',
    'PluginState',
    'plugin_manager'
]


class PluginManager:
    """
    Central manager for all plugin operations.

    Usage:
        plugin_manager = PluginManager()
        plugin_manager.init_app(app, db)

        # In CLI:
        plugin_manager.install_plugin('printers')
    """

    def __init__(self):
        self.registry: Optional[PluginRegistry] = None
        self.loader: Optional[PluginLoader] = None
        # Plugins installed while the app was already serving, whose blueprint
        # and CLI registration had to wait for a restart. The loader counts them
        # as loaded (it instantiated them), so this is the only record that
        # their ROUTES are not live yet.
        self._deferred_registration: set = set()
        # Plugins whose routes this process registered at startup. See
        # _load_enabled_plugins.
        self._booted_with: set = set()
        self.migration_manager: Optional[PluginMigrationManager] = None
        self.verifier = None
        self._app: Optional[Flask] = None
        self._db = None
        # API prefixes already claimed by a registered plugin blueprint, to
        # detect two plugins overlapping on the same /api/... namespace.
        self._registered_prefixes: set = set()

    def init_app(self, app: Flask, db) -> None:
        """Initialize plugin manager with Flask app."""
        self._app = app
        self._db = db
        # Reset per-app so the prefix-uniqueness guard tracks only this app's
        # registrations (the manager is a process-wide singleton; tests build
        # multiple apps from it).
        self._registered_prefixes = set()

        # Setup paths
        instance_path = Path(app.instance_path)
        plugins_dir = Path(app.root_path).parent / 'plugins'

        # Initialize components
        self.registry = PluginRegistry(instance_path / 'plugins.json')
        self.loader = PluginLoader(plugins_dir, self.registry)
        self.migration_manager = PluginMigrationManager(
            plugins_dir,
            app.config.get('SQLALCHEMY_DATABASE_URI')
        )

        # One trust policy, shared by verify-at-load and verify-at-migrate.
        from .verification import PluginVerifier
        verifier = PluginVerifier(
            plugins_dir,
            require_signed=app.config.get('PLUGIN_REQUIRE_SIGNED', False),
            trusted_key_paths=app.config.get('PLUGIN_TRUSTED_KEYS', []),
            dev_trust_dirs=app.config.get('PLUGIN_DEV_TRUST_DIRS', []),
            is_dev=bool(app.config.get('DEBUG') or app.config.get('TESTING')),
        )
        self.verifier = verifier
        self.loader.verifier = verifier
        self.migration_manager.verifier = verifier

        # Install (or clear) the import guard that verifies every `plugins.*`
        # import against a trusted signature. This is what covers the core
        # request handlers that do `from plugins.<name>.models import ...`, which
        # never pass through the loader (findings #1, #2).
        from . import importguard
        if verifier.require_signed:
            importguard.install(plugins_dir, verifier)
        else:
            importguard.uninstall()

        # Load enabled plugins
        self._load_enabled_plugins()

        # Store on app for access
        app.extensions['plugin_manager'] = self

    def _load_enabled_plugins(self) -> None:
        """Load and register all enabled plugins."""
        plugins = self.loader.load_enabled_plugins(self._app, self._db)
        # What THIS process actually registered routes for, recorded at boot.
        # Everything else is per-process and cannot answer "does the fleet need
        # a restart": under gunicorn each worker has its own manager, its own
        # loader and its own copy of the registry, so a plugin enabled through
        # one worker is invisible to the other three. Comparing this snapshot
        # against plugins.json re-read from disk is correct in every worker,
        # because a plugin enabled after boot is absent from every worker's
        # snapshot no matter which one served the enable.
        self._booted_with = set(plugins.keys())

        for name, plugin in plugins.items():
            self._register_plugin_components(plugin)
            self._check_settings_routes(name, plugin)

    def _check_settings_routes(self, name: str, plugin: BasePlugin) -> None:
        """Say at boot, once per plugin, if a declared settings tab is dead.

        Only some settings hubs draw plugin tabs (see
        SETTINGS_HUBS_WITH_PLUGIN_TABS). A route for any other hub is accepted
        by the hook and returned by the API and then rendered by nobody, which
        looks from every side like the plugin is at fault. The API drops such a
        route; this makes the same finding at startup, so it is in the log of a
        site that never opens the settings page."""
        try:
            renderable_settings_routes(name, plugin.get_settings_routes(),
                                       logger)
        except Exception:
            logger.exception("get_settings_routes failed for %s", name)

    def upgrade_all_plugins(self) -> Dict[str, str]:
        """Run pending Alembic migrations for every discovered plugin.

        Returns {plugin_name: 'ok'|'no-migrations'|<error str>}. Skips
        plugins with no migrations/ directory. Driven by the CLI
        (`flask plugin upgrade-all`), which every deploy runs after
        `flask db upgrade`. Each bundled plugin's chain begins with a
        stamp-only anchor (the core chain already built its tables); later
        per-plugin migrations extend that chain. See ADR-008. Idempotent.
        """
        results: Dict[str, str] = {}
        # Settings declared by a later plugin version reach an already-installed
        # site here. on_install/on_enable fire only on a state transition, so an
        # upgrade is the only moment left to seed them. Runs even with no
        # migration manager - it is independent of the Alembic chains.
        self.sync_all_plugin_settings()
        # Permissions need the same treatment for the same reason. A permission
        # added in a later plugin version would otherwise exist only after a
        # manual `flask seed permissions`, and until someone ran it the route
        # guarding on that name would refuse every caller - including admins on
        # sites that gate by permission rather than role. Deploys run
        # upgrade-all, so this is the moment the new name has to appear.
        self.sync_all_plugin_permissions()
        if not self.migration_manager:
            return results
        # Only ADOPTED plugins (those in the registry) get migrated. A plugin
        # folder that merely sits on disk unadopted must not have its DDL run as
        # a side effect of a deploy - migrations execute with full DB rights, so
        # "on disk" is not "trusted to run". (The old code probed a
        # list_installed() method the registry never had, so it always fell
        # through to discover_plugins() and migrated every folder.)
        plugin_names = list(self.registry.get_all().keys())
        for name in plugin_names:
            if not self.migration_manager.has_pending_migrations(name):
                results[name] = 'no-migrations'
                continue
            try:
                self._release_session_locks()
                ok = self.migration_manager.run_plugin_migrations(name)
                results[name] = 'ok' if ok else 'failed'
            except Exception as ex:
                results[name] = f'error: {ex}'
        return results

    def sync_all_plugin_settings(self) -> int:
        """Seed every enabled plugin's declared settings. Idempotent.

        Returns the number of plugins whose settings were touched. Best-effort
        per plugin: one bad plugin must not abort a deploy's upgrade pass.
        """
        touched = 0
        for name in list(self.registry.get_all().keys()):
            if not self.registry.is_enabled(name):
                continue
            try:
                plugin = self.loader.load_plugin(name, self._app, self._db)
            except Exception:
                logger.exception("Could not load %s to sync its settings", name)
                continue
            if plugin and self._seed_plugin_settings(plugin):
                touched += 1
        return touched

    def sync_all_plugin_permissions(self) -> int:
        """Seed every enabled plugin's declared permissions, then relink scopes.

        Returns the number of plugins whose permissions were seeded. Best-effort
        per plugin: one bad plugin must not abort a deploy's upgrade pass.

        The scope relink is the load-bearing half. Creating the Permission row
        only makes the name exist; the prebuilt scopes derive their grants from
        the catalog, so a permission seeded without a following
        Role.seedscopes() belongs to nobody but admin. The symptom is a plugin
        that installs cleanly and whose UI is invisible to every non-admin until
        someone re-runs the seed by hand. Ordering matters for the same reason
        it does in `flask seed permissions`: seedscopes only links rows that
        already exist, so it runs last.
        """
        touched = 0
        for name in list(self.registry.get_all().keys()):
            if not self.registry.is_enabled(name):
                continue
            try:
                plugin = self.loader.load_plugin(name, self._app, self._db)
            except Exception:
                logger.exception(
                    "Could not load %s to sync its permissions", name)
                continue
            if plugin:
                self._seed_plugin_permissions(plugin)
                touched += 1

        # Additive: this can hand a scope a newly declared permission, never
        # take one away, so a site's deliberate role edits survive the deploy.
        from shopdb.core.models import Role
        try:
            with self._app.app_context():
                result = Role.seedscopes()
                self._db.session.commit()
                granted = sum((result or {}).get('granted', {}).values())
                if granted:
                    logger.info(
                        "Linked %d new scope grant(s) after plugin upgrade",
                        granted)
        except Exception:
            logger.exception("Could not relink prebuilt scopes after upgrade")
        return touched

    def get_declared_plugin_settings(self) -> Dict[str, dict]:
        """Return {key: declared entry} across all loaded plugins.

        The settings API uses this to file a first-time write under the owning
        plugin's category instead of a placeholder, and to decide which keys an
        unauthenticated caller may read.
        """
        declared: Dict[str, dict] = {}
        for name, plugin in self.loader.get_all_loaded().items():
            try:
                entries = plugin.get_settings_defaults() or []
            except Exception:
                logger.exception("get_settings_defaults failed for %s", name)
                continue
            for entry in entries:
                key = entry.get('key')
                if key:
                    declared[key] = entry
        return declared

    def get_public_setting_keys(self) -> set:
        """Setting keys plugins declare readable without authentication."""
        return {key for key, entry in self.get_declared_plugin_settings().items()
                if entry.get('public')}

    def _release_session_locks(self) -> None:
        """Close this request's transaction before any plugin DDL runs.

        SELF-DEADLOCK, AND WHY IT TAKES THE WHOLE SITE DOWN. Enabling a plugin
        over HTTP arrives inside a request that has already authenticated, and
        @jwt_required does a SELECT on users - so the request's session is
        sitting in an open transaction holding a SHARED_READ metadata lock on
        users. The plugin's migrations then run through Alembic, which builds
        its OWN connection; a plugin table with a foreign key to users (
        endpointlocks, policyactionholds, endpointlogcollections and others)
        needs an EXCLUSIVE metadata lock on it. That second connection waits on
        the first, the first cannot commit because the request is blocked on the
        second, and it never resolves.

        It does not stop at the one request. MySQL grants metadata locks in
        request order, so every later SELECT on users queues BEHIND the pending
        exclusive even though those reads are compatible with the shared lock
        already held. Every login on the site then hangs: observed as a total
        outage from one plugin toggle.

        Committing first ends the transaction and drops the lock, so the DDL
        takes its exclusive lock immediately. The session stays usable - the
        next statement simply begins a new transaction.
        """
        if self._db is None:
            return
        try:
            self._db.session.commit()
        except Exception:
            try:
                self._db.session.rollback()
            except Exception:
                pass

    def _register_plugin_components(self, plugin: BasePlugin) -> None:
        """Register plugin's blueprint, models, CLI commands, etc.

        NOT POSSIBLE ON A LIVE APP, and that is fine. Flask refuses
        register_blueprint once the app has served its first request, raising
        AssertionError. Installing a plugin at runtime - which is exactly what
        enabling one from the setup wizard or the plugins page does - therefore
        used to run the plugin's migrations, update the registry, and THEN blow
        up with a 500, leaving the operator with an error over a change that had
        in fact been recorded.

        Routes only ever take effect on restart anyway; the UI has always said
        so. So on a live app, skip the parts that mutate app setup and let the
        restart register them properly. The registry entry, the migrations, the
        permissions and the settings - everything that belongs in the database -
        still happen here.
        """
        # Flask renamed this: 2.x exposed `got_first_request` publicly, 3.x
        # keeps it as `_got_first_request`. Check both, because defaulting to
        # False on the wrong name silently disables this guard - which is how
        # the 500 survived the first attempt at this fix.
        served = getattr(self._app, 'got_first_request', None)
        if served is None:
            served = getattr(self._app, '_got_first_request', False)
        if served:
            self._deferred_registration.add(plugin.meta.name)
            logger.info(
                "%s installed while the app is serving; its blueprint and CLI "
                "commands register on the next restart (Flask does not allow "
                "either after the first request).", plugin.meta.name)
            return

        # Register blueprint
        blueprint = plugin.get_blueprint()
        if blueprint:
            prefix = plugin.meta.api_prefix
            # Guard against two plugins claiming the same API prefix; Flask only
            # rejects duplicate blueprint names, not overlapping url_prefixes, so
            # an overlap would silently shadow routes.
            if prefix in self._registered_prefixes:
                raise ValueError(
                    f"Plugin {plugin.meta.name} api_prefix '{prefix}' is already "
                    f"claimed by another blueprint"
                )
            self._app.register_blueprint(blueprint, url_prefix=prefix)
            self._registered_prefixes.add(prefix)
            logger.debug(f"Registered blueprint: {prefix}")

        # Register CLI commands
        for cmd in plugin.get_cli_commands():
            self._app.cli.add_command(cmd)

    def get_booted_with(self) -> set:
        """Plugins this process registered routes for at startup."""
        return set(self._booted_with)

    def enabled_on_disk(self) -> set:
        """Enabled plugins per plugins.json, re-read now.

        The in-memory registry is per-worker and goes stale the moment another
        worker enables something, so a restart check must go back to the file.
        """
        import json
        try:
            path = self.registry.state_file
            with open(path) as handle:
                data = json.load(handle)
            return {name for name, state in (data.get('plugins') or {}).items()
                    if state.get('enabled')}
        except FileNotFoundError:
            return set()
        except Exception:
            # Unreadable or corrupt: fall back to this worker's view rather
            # than claiming nothing is enabled.
            return set(self.registry.get_enabled_plugins())

    def get_deferred_registrations(self) -> set:
        """Plugins enabled in this process whose routes await a restart.

        get_all_loaded() is not a substitute: install_plugin instantiates the
        plugin through the loader, so a plugin whose blueprint was skipped still
        reports as loaded. Without this, a health check says no restart is
        needed while every one of that plugin's routes answers 404.
        """
        return set(self._deferred_registration)

    def discover_available(self) -> List[Dict]:
        """
        Get list of all available plugins (installed or not).
        Returns list of plugin info dicts.
        """
        available = []

        for name in self.loader.discover_plugins():
            # Under enforcement load_plugin_class refuses an unverified plugin
            # (it will not import it). Skip such a plugin from the listing
            # instead of failing the whole call.
            try:
                plugin_class = self.loader.load_plugin_class(name)
            except PluginError as e:
                logger.warning(f"Skipping plugin {name} in listing: {e}")
                continue
            if plugin_class:
                try:
                    temp = plugin_class()
                    meta = temp.meta
                    state = self.registry.get(name)

                    try:
                        config_schema = temp.get_config_schema()
                    except Exception:
                        config_schema = []
                    try:
                        provisioning_note = temp.get_provisioning_note()
                    except Exception:
                        provisioning_note = None
                    manifest = self.loader.load_manifest(name)
                    available.append({
                        'name': meta.name,
                        # Human label for UIs. Manifest display_name wins; else
                        # title-case the machine name (right for most, e.g.
                        # "computers" -> "Computers").
                        'displayname': (manifest.get('display_name')
                                        or meta.name.replace('_', ' ').title()),
                        'version': meta.version,
                        'description': meta.description,
                        'author': meta.author,
                        'dependencies': meta.dependencies,
                        'installed': state is not None,
                        'enabled': state.enabled if state else False,
                        'installedat': state.installed_at if state else None,
                        'config_schema': config_schema,
                        'provisioning_note': provisioning_note,
                        'default_enabled': manifest.get('default_enabled', True),
                        # Declared per plugin in its manifest, not held as a
                        # list in the wizard, so adding a plugin cannot leave
                        # the recommendation out of date in another file.
                        'recommended': bool(manifest.get('recommended', False)),
                    })
                except Exception as e:
                    logger.warning(f"Error inspecting plugin {name}: {e}")

        return available

    def install_plugin(self, name: str, run_migrations: bool = True) -> bool:
        """
        Install a plugin.

        Steps:
        1. Verify plugin exists
        2. Check dependencies
        3. Run database migrations
        4. Register in registry
        5. Call plugin's on_install hook
        """
        # Check if already installed
        if self.registry.is_installed(name):
            logger.warning(f"Plugin {name} is already installed")
            return False

        # Read metadata from the manifest (single source of truth) instead of
        # instantiating the plugin class just to inspect deps/version.
        manifest = self.loader.load_manifest(name)
        if not manifest:
            logger.error(f"Plugin {name} not found")
            return False
        manifest_version = manifest.get('version')

        # Check dependencies
        for dep in manifest.get('dependencies', []):
            if not self.registry.is_installed(dep):
                logger.error(
                    f"Plugin {name} requires {dep} to be installed first"
                )
                return False

        # Run migrations
        if run_migrations:
            self._release_session_locks()
            success = self.migration_manager.run_plugin_migrations(name)
            if not success:
                logger.error(f"Failed to run migrations for {name}")
                return False

        # Register plugin
        # Plugins that provision extra tables install disabled until a site
        # opts in (manifest default_enabled=false).
        self.registry.register(name, manifest_version,
                               enabled=manifest.get('default_enabled', True))

        # Load the plugin
        plugin = self.loader.load_plugin(name, self._app, self._db)
        if plugin:
            self._register_plugin_components(plugin)
            self._seed_plugin_permissions(plugin)
            self._seed_plugin_settings(plugin)
            plugin.on_install(self._app)

        logger.info(f"Installed plugin: {name} v{manifest_version}")
        return True

    def _seed_plugin_permissions(self, plugin: BasePlugin) -> None:
        """Idempotently create Permission rows for a plugin's declared perms.

        Runs at install and enable so a fresh plugin's RBAC permissions exist
        without a separate `flask seed permissions` pass. Best-effort: a plugin
        that raises must not abort the lifecycle."""
        try:
            entries = plugin.get_permissions() or []
        except Exception:
            logger.exception(
                "get_permissions failed for %s", plugin.meta.name)
            return
        if not entries:
            return
        from shopdb.core.models import Permission
        with self._app.app_context():
            created = Permission.seed_entries(entries)
            if created:
                self._db.session.commit()
                logger.info(
                    "Seeded %d permission(s) for plugin %s",
                    created, plugin.meta.name)

    def _seed_plugin_settings(self, plugin: BasePlugin) -> bool:
        """Idempotently create Setting rows for a plugin's declared settings.

        Runs at install, at enable, and on `flask plugin upgrade-all`. Existing
        values are never overwritten - only missing rows are created, and a row
        whose category does not match the declaration is re-homed. That repair
        matters: before a key was declared, the first save created it under the
        settings API's placeholder category, where the owning plugin's settings
        page (which filters by category) could not see it again, so the value
        looked like it never saved.

        Returns True if anything was created or repaired. Best-effort: a plugin
        that raises must not abort the lifecycle.
        """
        try:
            entries = plugin.get_settings_defaults() or []
        except Exception:
            logger.exception(
                "get_settings_defaults failed for %s", plugin.meta.name)
            return False
        if not entries:
            return False
        from shopdb.core.models import Setting
        created = 0
        repaired = 0
        with self._app.app_context():
            for entry in entries:
                key = entry.get('key')
                if not key:
                    continue
                setting = Setting.query.filter_by(key=key).first()
                if setting is None:
                    self._db.session.add(Setting(
                        key=key,
                        value=entry.get('value', ''),
                        valuetype=entry.get('valuetype', 'string'),
                        category=entry.get('category', 'plugin'),
                        description=entry.get('description'),
                    ))
                    created += 1
                    continue
                category = entry.get('category')
                if category and setting.category != category:
                    setting.category = category
                    repaired += 1
            if created or repaired:
                self._db.session.commit()
                logger.info(
                    "Plugin %s settings: %d created, %d re-homed",
                    plugin.meta.name, created, repaired)
        return bool(created or repaired)

    def _is_core_tier(self, name: str) -> bool:
        """True when the plugin's manifest marks it tier=core (mandatory).

        A core-tier plugin refuses uninstall/disable so a site cannot remove a
        plugin its deployment depends on. No plugin ships tier=core today; the
        guard makes that a manifest edit, not a code change (ADR-013).
        """
        try:
            return self.loader.load_manifest(name).get('tier') == 'core'
        except PluginError:
            return False

    def _installed_dependents(self, name: str,
                              enabled_only: bool = False) -> List[str]:
        """Installed plugins that declare `name` as a hard dependency.

        Reads each dependency list from the manifest on DISK, not from a loaded
        plugin instance, so a dependent that is installed-but-not-loaded (e.g.
        disabled, or failed to load) is still counted. enabled_only restricts to
        currently-enabled dependents (what matters when disabling, since a
        disabled dependent is not running).
        """
        dependents = []
        for other_name, state in self.registry.get_all().items():
            if other_name == name:
                continue
            if enabled_only and not state.enabled:
                continue
            try:
                manifest = self.loader.load_manifest(other_name)
            except PluginError:
                continue
            if name in manifest.get('dependencies', []):
                dependents.append(other_name)
        return dependents

    def uninstall_plugin(self, name: str, remove_data: bool = False) -> bool:
        """
        Uninstall a plugin.

        Args:
            name: Plugin name
            remove_data: If True, run downgrade migrations to remove tables
        """
        if not self.registry.is_installed(name):
            logger.warning(f"Plugin {name} is not installed")
            return False

        if self._is_core_tier(name):
            logger.error(f"Cannot uninstall {name}: it is a core-tier plugin")
            return False

        # Refuse if any INSTALLED plugin (enabled or not) depends on this one.
        dependents = self._installed_dependents(name)
        if dependents:
            logger.error(
                f"Cannot uninstall {name}: {', '.join(dependents)} depend(s) on it"
            )
            return False

        # Get plugin instance
        plugin = self.loader.get_loaded_plugin(name)

        # Call on_uninstall hook
        if plugin:
            plugin.on_uninstall(self._app)

        # Optionally remove data
        if remove_data:
            self.migration_manager.downgrade_plugin(name)

        # Unregister
        self.registry.unregister(name)

        logger.info(f"Uninstalled plugin: {name}")
        return True

    def enable_plugin(self, name: str) -> bool:
        """Enable a disabled plugin. Auto-installs if available but not installed."""
        # Security: Validate plugin name to prevent path traversal
        # Only allow alphanumeric, underscore, and hyphen (matches existing plugins)
        if not re.match(r'^[a-zA-Z0-9_-]+$', name):
            logger.error(f"Invalid plugin name format: {name}")
            return False

        # Reject reserved names that could be used for directory traversal
        if name in ('.', '..', ''):
            logger.error(f"Invalid plugin name: {name}")
            return False

        # Decisions below read the registry, so start from what is on disk
        # rather than this worker's boot snapshot.
        self.registry.refresh()

        # Auto-install if available but not installed (UX: enabling should just work)
        if not self.registry.is_installed(name):
            logger.info(f"Plugin {name} not installed, installing first")
            if not self.install_plugin(name):
                logger.error(f"Failed to install plugin {name}")
                return False

        if self.registry.is_enabled(name):
            logger.info(f"Plugin {name} is already enabled")
            return True

        # Check dependencies are enabled. Read deps from the manifest, not by
        # instantiating the plugin class (manifest is the single source of
        # truth; instantiating fires __init__ side effects unnecessarily).
        manifest = self.loader.load_manifest(name)
        for dep in manifest.get('dependencies', []):
            if not self.registry.is_enabled(dep):
                logger.error(f"Cannot enable {name}: {dep} is not enabled")
                return False

        self.registry.enable(name)

        # Surface (do not auto-apply) a plugin chain that is ahead of the DB, so
        # an operator enabling a plugin knows to run `flask plugin upgrade-all`.
        try:
            if self.migration_manager and \
                    self.migration_manager.has_unapplied_migrations(name):
                logger.warning(
                    f"Plugin {name} has unapplied migrations; "
                    f"run 'flask plugin upgrade-all'"
                )
        except Exception:
            logger.debug(f"Could not check migration state for {name}")

        # Fire the on_enable hook best-effort. Do NOT register the blueprint
        # here: Flask forbids register_blueprint after the first request, so
        # routes/nav for a re-enabled plugin take effect on the next restart
        # (symmetric with disable).
        try:
            plugin = self.loader.load_plugin(name, self._app, self._db)
            if plugin:
                self._seed_plugin_permissions(plugin)
                self._seed_plugin_settings(plugin)
                plugin.on_enable(self._app)
        except Exception:
            logger.exception(f"on_enable hook failed for plugin {name}")

        logger.info(f"Enabled plugin: {name}")
        return True

    def disable_plugin(self, name: str) -> bool:
        """Disable an enabled plugin."""
        self.registry.refresh()
        if not self.registry.is_enabled(name):
            logger.info(f"Plugin {name} is already disabled")
            return True

        if self._is_core_tier(name):
            logger.error(f"Cannot disable {name}: it is a core-tier plugin")
            return False

        # Refuse if any ENABLED plugin depends on this one (a disabled dependent
        # is not running, so it does not block). Manifests read from disk.
        dependents = self._installed_dependents(name, enabled_only=True)
        if dependents:
            logger.error(
                f"Cannot disable {name}: {', '.join(dependents)} depend(s) on it"
            )
            return False

        plugin = self.loader.get_loaded_plugin(name)
        if plugin:
            plugin.on_disable(self._app)

        self.registry.disable(name)
        logger.info(f"Disabled plugin: {name}")
        return True

    def _dependency_closure(self, names: List[str]) -> List[str]:
        """Chosen plugins plus their hard-dependency closure, dependencies first.

        Post-order DFS over manifest `dependencies`, so the returned order is
        safe to install/enable top to bottom (a dependency always precedes the
        plugin that needs it). Deduplicated.
        """
        closure: List[str] = []
        seen = set()

        def add(name):
            if name in seen:
                return
            seen.add(name)
            try:
                deps = self.loader.load_manifest(name).get('dependencies', [])
            except PluginError:
                deps = []
            for dep in deps:
                add(dep)
            closure.append(name)

        for name in names:
            add(name)
        return closure

    def apply_profile(self, plugins: List[str],
                      locked: Optional[List[str]] = None) -> Dict[str, List[str]]:
        """Install AND enable exactly the chosen plugins plus their dependency
        closure, in dependency order.

        Declarative replacement for the hand-ordered install/enable sequences in
        the deploy runbooks (which could enable a plugin that was never
        installed). Idempotent: already-installed/enabled plugins are left as-is.
        This does NOT remove anything absent from the list - removal stays an
        explicit, separate operation. `locked` is accepted for forward
        compatibility (a future guard against removing a site-mandated plugin)
        and is currently informational.

        Raises PluginError if the profile names a plugin (or pulls in a
        dependency) that does not exist on disk. Returns
        {installed: [...], enabled: [...], already: [...]}.
        """
        available = set(self.loader.discover_plugins())
        unknown = [p for p in plugins if p not in available]
        if unknown:
            raise PluginError(
                f"Profile names unknown plugins: {', '.join(sorted(unknown))}")

        order = self._dependency_closure(plugins)
        missing_deps = [p for p in order if p not in available]
        if missing_deps:
            raise PluginError(
                f"Missing dependency plugins: {', '.join(sorted(missing_deps))}")

        result: Dict[str, List[str]] = {
            'installed': [], 'enabled': [], 'already': []}
        for name in order:
            changed = False
            if not self.registry.is_installed(name):
                if self.install_plugin(name):
                    result['installed'].append(name)
                    changed = True
            if not self.registry.is_enabled(name):
                if self.enable_plugin(name):
                    result['enabled'].append(name)
                    changed = True
            if not changed:
                result['already'].append(name)
        return result

    def get_plugin(self, name: str) -> Optional[BasePlugin]:
        """Get a loaded plugin instance."""
        return self.loader.get_loaded_plugin(name)

    def get_all_plugins(self) -> Dict[str, BasePlugin]:
        """Get all loaded plugins."""
        return self.loader.get_all_loaded()

    def get_service(self, name: str):
        """Resolve a service exposed by an enabled plugin via get_services().

        Consumer for the BasePlugin.get_services hook: searches enabled plugins
        for one that registers `name` and returns the registered value (a service
        class or factory). Returns None if no enabled plugin provides it. This is
        how one plugin obtains another's service (e.g. the Zabbix service).
        """
        for plugin_name, plugin in self.get_all_plugins().items():
            if not self.registry.is_enabled(plugin_name):
                continue
            try:
                services = plugin.get_services() or {}
            except Exception:
                continue
            if name in services:
                return services[name]
        return None


# Global plugin manager instance
plugin_manager = PluginManager()