orchestrata_ultimate.plugin.discovery

Reference › orchestrata_ultimate › plugin › discovery

The function-level lifecycle operations behind the PluginManager, written as plain functions over a PluginRegistry so the manager stays small.

discover_plugins(loader, registry) asks the loader for every manifest and registers each as DISCOVERED, with a source of local:<dir> or pip. install_plugin(source, registry, install_dir) accepts an existing local directory, which it reads and registers, or a git+<url> or <url>.git source; anything else raises PluginLoadError with a hint that pip plugins are added with uv add and found through their entry point. _install_from_git derives the directory name from the URL, takes the base directory from install_dir, the ORCHESTRATA_PLUGIN_DIR variable or plugins, makes a shallow git clone with a 120-second limit unless the target already holds a plugin.yaml, and refuses an existing target without a manifest.

uninstall_plugin stops the plugin and removes it from the registry, which also cleans up its context. disable_plugin stops it and marks it DISABLED. enable_plugin re-registers a disabled plugin as discovered, loads it through the supplied load_fn and starts it. unload_plugin stops and unregisters it. reload_plugin stops and unregisters, registers the manifest again, loads and starts. Unknown names raise PluginNotFoundError, and an error while stopping is logged and does not stop the operation.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/plugin/discovery.py

"""Plugin discovery and installation helpers.

Extracted from manager.py to keep PluginManager under 300 lines
and under 15 public methods. Contains discover, install, and
uninstall logic that delegates to PluginLoader and PluginRegistry.
"""

import logging
import os
import subprocess
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.plugin.exceptions import (
    PluginLoadError,
    PluginNotFoundError,
)
from orchestrata_ultimate.plugin.interfaces import PluginStatus
from orchestrata_ultimate.plugin.loader import PluginLoader
from orchestrata_ultimate.plugin.manifest import PluginManifest
from orchestrata_ultimate.plugin.registry import PluginRegistry

_LOG = logging.getLogger(__name__)


@traces(
    spec="015",
    story="US-001",
    reqs=["REQ-015-006"],
)
def discover_plugins(
    loader: PluginLoader,
    registry: PluginRegistry,
) -> list[PluginManifest]:
    """Discover available plugins from all sources.

    Scans local directories and pip-installed packages.
    Discovered manifests are registered with DISCOVERED status.

    Args:
        loader: Plugin loader for filesystem/pip discovery.
        registry: Plugin registry for tracking.

    Returns:
        List of discovered plugin manifests.
    """
    manifests = loader.discover_all()
    for manifest in manifests:
        source = f"local:{manifest.source_path}" if manifest.source_path else "pip"
        registry.register_discovered(manifest, source=source)
    _LOG.info("Discovered %d plugins", len(manifests))
    return manifests


@traces(
    spec="015",
    story="US-001",
    reqs=["REQ-015-002", "REQ-015-003"],
)
def install_plugin(
    source: str,
    registry: PluginRegistry,
    *,
    install_dir: str | None = None,
) -> PluginManifest:
    """Install a plugin from a local directory or a git repository.

    Reads ``plugin.yaml`` from the resolved plugin directory and registers the
    plugin as discovered. Two source forms are supported:

    - **Local directory** — a filesystem path to a directory containing a
      ``plugin.yaml`` manifest.
    - **Git repository** — a ``git+<url>`` source (or a ``<url>.git``), which is
      cloned (shallow) into ``install_dir`` and then read like a local plugin.

    Pip-distributed plugins are not installed here: install them with
    ``uv add <package>`` and they are picked up automatically by
    :meth:`PluginLoader.discover_installed` via their ``orchestrata_plugin``
    entry point.

    Args:
        source: A local plugin directory path, or a ``git+<url>`` / ``<url>.git``
            repository URL.
        registry: Plugin registry for tracking.
        install_dir: Directory git clones are placed under. Defaults to the
            ``ORCHESTRATA_PLUGIN_DIR`` environment variable, or ``"plugins"``
            (matching :class:`PluginLoader`'s default search dir). Ignored for
            local-directory sources.

    Returns:
        The installed plugin's manifest.

    Raises:
        PluginLoadError: If the source form is unsupported, the git clone
            fails, or the resolved directory lacks a valid ``plugin.yaml``.
    """
    path = Path(source).expanduser()
    if path.exists() and path.is_dir():
        manifest_path = path / "plugin.yaml"
        manifest = PluginManifest.from_yaml(manifest_path)
        registry.register_discovered(
            manifest,
            source=f"local:{path}",
        )
        _LOG.info(
            "Installed plugin '%s' from %s",
            manifest.name,
            source,
        )
        return manifest
    if source.startswith("git+") or source.endswith(".git"):
        return _install_from_git(source, registry, install_dir)
    raise PluginLoadError(
        f"Unsupported plugin source: {source!r}. Supported sources are a local "
        f"directory path or a 'git+<url>' / '<url>.git' repository. "
        f"Pip-distributed plugins are installed with `uv add <package>` and "
        f"discovered via their entry point (discover_installed).",
        context={"source": source},
    )


@traces(spec="015", story="US-001", reqs=["REQ-015-003"])
def _install_from_git(
    source: str,
    registry: PluginRegistry,
    install_dir: str | None,
) -> PluginManifest:
    """Clone a git plugin repo into ``install_dir`` and register its manifest.

    Idempotent: if the target already holds a ``plugin.yaml`` (already cloned),
    it is re-read and re-registered rather than re-cloned.
    """
    url = source[4:] if source.startswith("git+") else source
    name = url.rstrip("/").rsplit("/", 1)[-1]
    if name.endswith(".git"):
        name = name[:-4]
    if not name:
        raise PluginLoadError(
            f"Cannot derive a plugin directory name from git source: {source!r}",
            context={"source": source},
        )
    base = Path(
        install_dir or os.environ.get("ORCHESTRATA_PLUGIN_DIR", "plugins")
    ).expanduser()
    target = base / name

    if not (target / "plugin.yaml").exists():
        if target.exists():
            raise PluginLoadError(
                f"Git install target already exists without a plugin.yaml: {target}",
                context={"target": str(target)},
            )
        base.mkdir(parents=True, exist_ok=True)
        result = subprocess.run(
            ["git", "clone", "--depth", "1", url, str(target)],
            capture_output=True,
            text=True,
            timeout=120,
        )
        if result.returncode != 0:
            raise PluginLoadError(
                f"git clone failed for {url}: {result.stderr.strip()}",
                context={"url": url, "returncode": result.returncode},
            )

    manifest = PluginManifest.from_yaml(target / "plugin.yaml")
    manifest.source_path = target
    registry.register_discovered(manifest, source=f"git:{url}")
    _LOG.info("Installed plugin '%s' from %s", manifest.name, source)
    return manifest


@traces(
    spec="015",
    story="US-007",
    reqs=["REQ-015-088"],
)
async def uninstall_plugin(
    name: str,
    registry: PluginRegistry,
) -> None:
    """Uninstall a plugin completely.

    Stops, unloads, and removes from registry.

    Args:
        name: Plugin name to uninstall.
        registry: Plugin registry for tracking.

    Raises:
        PluginNotFoundError: If plugin is not known.
    """
    info = registry.get_info(name)
    if info is None:
        raise PluginNotFoundError(
            f"Plugin '{name}' not found",
            context={"name": name},
        )
    plugin = registry.get_plugin(name)
    if plugin is not None:
        try:
            await plugin.stop()
        except Exception as exc:
            _LOG.error(
                "Error stopping '%s' during uninstall: %s",
                name,
                exc,
            )
    registry.unregister(name)
    _LOG.info("Uninstalled plugin '%s'", name)


@traces(
    spec="015",
    story="US-007",
    reqs=["REQ-015-085"],
)
async def enable_plugin(
    name: str,
    registry: PluginRegistry,
    load_fn: Any,
) -> None:
    """Enable a disabled plugin.

    Loads and starts the plugin if it was previously disabled.

    Args:
        name: Plugin name to enable.
        registry: Plugin registry for tracking.
        load_fn: Async callable to load a plugin by name.

    Raises:
        PluginNotFoundError: If plugin is not known.
    """
    info = registry.get_info(name)
    if info is None:
        raise PluginNotFoundError(
            f"Plugin '{name}' not found",
            context={"name": name},
        )
    if info.status == PluginStatus.DISABLED:
        manifest = registry.get_manifest(name)
        if manifest is None:
            raise PluginNotFoundError(
                f"Manifest lost for '{name}'",
                context={"name": name},
            )
        source = info.source
        registry.unregister(name)
        registry.register_discovered(
            manifest,
            source=source,
        )
        await load_fn(name)
        plugin = registry.get_plugin(name)
        if plugin is not None:
            await plugin.start()
        _LOG.info("Enabled plugin '%s'", name)


@traces(
    spec="015",
    story="US-007",
    reqs=["REQ-015-086", "REQ-015-087"],
)
async def disable_plugin(
    name: str,
    registry: PluginRegistry,
) -> None:
    """Disable a plugin without uninstalling.

    Stops the plugin and marks it as DISABLED.

    Args:
        name: Plugin name to disable.
        registry: Plugin registry for tracking.

    Raises:
        PluginNotFoundError: If plugin is not known.
    """
    info = registry.get_info(name)
    if info is None:
        raise PluginNotFoundError(
            f"Plugin '{name}' not found",
            context={"name": name},
        )
    plugin = registry.get_plugin(name)
    if plugin is not None:
        try:
            await plugin.stop()
        except Exception as exc:
            _LOG.error(
                "Error stopping '%s' during disable: %s",
                name,
                exc,
            )
    registry.set_status(name, PluginStatus.DISABLED)
    _LOG.info("Disabled plugin '%s'", name)


@traces(
    spec="015",
    story="US-010",
    reqs=["REQ-015-127"],
)
async def unload_plugin(
    name: str,
    registry: PluginRegistry,
) -> None:
    """Stop and unload a plugin.

    Args:
        name: Plugin name to unload.
        registry: Plugin registry for tracking.

    Raises:
        PluginNotFoundError: If plugin is not loaded.
    """
    plugin = registry.get_plugin(name)
    if plugin is None:
        raise PluginNotFoundError(
            f"Plugin '{name}' is not loaded",
            context={"name": name},
        )
    try:
        await plugin.stop()
    except Exception as exc:
        _LOG.error(
            "Error stopping plugin '%s': %s",
            name,
            exc,
        )
    registry.unregister(name)
    _LOG.info("Unloaded plugin '%s'", name)


@traces(
    spec="015",
    story="US-010",
    reqs=["REQ-015-136"],
)
async def reload_plugin(
    name: str,
    registry: PluginRegistry,
    load_fn: Any,
) -> None:
    """Hot-reload a plugin (stop, unload, load, start).

    Args:
        name: Plugin name to reload.
        registry: Plugin registry for tracking.
        load_fn: Async callable to load a plugin by name.

    Raises:
        PluginNotFoundError: If plugin not known.
    """
    manifest = registry.get_manifest(name)
    if manifest is None:
        raise PluginNotFoundError(
            f"Plugin '{name}' not found for reload",
            context={"name": name},
        )
    plugin = registry.get_plugin(name)
    if plugin is not None:
        try:
            await plugin.stop()
        except Exception as exc:
            _LOG.error(
                "Error stopping '%s' during reload: %s",
                name,
                exc,
            )
        registry.unregister(name)
    registry.register_discovered(
        manifest,
        source=(f"local:{manifest.source_path}" if manifest.source_path else "pip"),
    )
    await load_fn(name)
    new_plugin = registry.get_plugin(name)
    if new_plugin is not None:
        await new_plugin.start()
    _LOG.info("Reloaded plugin '%s'", name)