agentic-mobile-control/cloud/plugins.py

"""Plugin manifest and registry: discover and wire drivers/tools/skills.

Capability: ``plugin-system``.

Two discovery paths (Python ``importlib.metadata`` entry points in group
``device_agent_runtime.plugins``, and a local ``plugin.json`` file scan) feed
into one validation + registration path. Only ``driver``-kind plugins are
concretely wired (into ``driver/registry.py``'s extension point); ``tool``- and
``skill``-kind manifests are accepted and stored as ``wired=False``.
"""

from __future__ import annotations

import importlib
import importlib.metadata
import json
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal

if TYPE_CHECKING:
    from cloud.store import CloudStore


logger = logging.getLogger(__name__)

ENTRY_POINT_GROUP = "device_agent_runtime.plugins"
PLUGIN_MANIFEST_FILENAME = "plugin.json"
EntryPointKind = Literal["driver", "tool", "skill"]
_VALID_ENTRY_POINT_KINDS = ("driver", "tool", "skill")


@dataclass(frozen=True)
class PluginManifest:
    """A plugin's declaration: who it is, what kind it is, where it lives."""

    name: str
    version: str
    entry_point_kind: EntryPointKind
    target: str

    def __post_init__(self) -> None:
        if not self.name:
            raise PluginValidationError("plugin manifest 'name' is required")
        if not self.version:
            raise PluginValidationError("plugin manifest 'version' is required")
        if not self.target:
            raise PluginValidationError("plugin manifest 'target' is required")
        if self.entry_point_kind not in _VALID_ENTRY_POINT_KINDS:
            raise PluginValidationError(
                f"plugin manifest 'entry_point_kind' must be one of "
                f"{_VALID_ENTRY_POINT_KINDS}, got {self.entry_point_kind!r}"
            )

    def to_dict(self) -> dict[str, Any]:
        return {
            "name": self.name,
            "version": self.version,
            "entry_point_kind": self.entry_point_kind,
            "target": self.target,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "PluginManifest":
        return cls(
            name=str(data["name"]),
            version=str(data["version"]),
            entry_point_kind=str(data["entry_point_kind"]),  # type: ignore[arg-type]
            target=str(data["target"]),
        )


@dataclass
class DiscoveryResult:
    """The outcome of running both discovery scans."""

    registered: list[PluginManifest] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)


class PluginValidationError(ValueError):
    """Raised when a plugin manifest fails schema validation."""


class DuplicatePluginError(ValueError):
    """Raised when a plugin name is already registered."""


class DriverRegistryUnavailableError(RuntimeError):
    """Raised when ``driver.registry.register_driver_type`` is not importable."""


class PluginTargetResolutionError(RuntimeError):
    """Raised when a plugin target cannot be resolved to a callable."""


# Exception types `register()` can raise for a single bad manifest (validation
# failure, name conflict, or driver-wiring failure). `discover()` catches only
# these so one bad manifest is skipped/logged, per spec's "malformed manifest
# file is skipped, not fatal" resilience intent -- anything else propagates as
# a genuine programmer error.
_PLUGIN_REGISTRATION_FAILURES = (
    PluginValidationError,
    DuplicatePluginError,
    DriverRegistryUnavailableError,
    PluginTargetResolutionError,
)


class PluginRegistry:
    """Validates, persists, and wires plugins discovered or submitted directly."""

    def __init__(self, store: "CloudStore") -> None:
        self.store = store

    def register(self, manifest: PluginManifest) -> PluginManifest:
        """Validate, persist, and (for drivers) wire a plugin manifest."""
        if self.store.get_plugin(manifest.name) is not None:
            raise DuplicatePluginError(
                f"a plugin named {manifest.name!r} is already registered"
            )
        wired = False
        if manifest.entry_point_kind == "driver":
            wired = self._wire_driver(manifest)
        # tool/skill-kind: stored with wired=False, no other resolution attempted.
        self.store.save_plugin(manifest, wired=wired)
        if not wired:
            logger.info(
                "plugin %r registered but not wired to any execution path",
                manifest.name,
            )
        return manifest

    def list(self) -> list[tuple[PluginManifest, bool]]:
        return self.store.list_plugins()

    def discover(self, scan_path: str | Path | None = None) -> DiscoveryResult:
        result = DiscoveryResult()
        for manifest in self.discover_entry_points():
            try:
                self.register(manifest)
                result.registered.append(manifest)
            except _PLUGIN_REGISTRATION_FAILURES as exc:
                result.errors.append(
                    f"entry-point plugin {manifest.name!r}: {exc}"
                )
        if scan_path is not None:
            for manifest in self.discover_manifest_files(scan_path):
                try:
                    self.register(manifest)
                    result.registered.append(manifest)
                except _PLUGIN_REGISTRATION_FAILURES as exc:
                    result.errors.append(
                        f"manifest-file plugin {manifest.name!r}: {exc}"
                    )
        return result

    def discover_entry_points(self) -> list[PluginManifest]:
        manifests: list[PluginManifest] = []
        try:
            entry_points = importlib.metadata.entry_points(group=ENTRY_POINT_GROUP)
        except TypeError:
            # Older Pythons: entry_points() returns a dict-like view.
            all_eps = importlib.metadata.entry_points()
            entry_points = all_eps.get(ENTRY_POINT_GROUP, [])  # type: ignore[union-attr]
        for entry_point in entry_points:
            try:
                loaded = entry_point.load()
            except Exception as exc:  # pragma: no cover - exercised via fake eps in tests
                logger.warning("entry point %r failed to load: %s", entry_point.name, exc)
                continue
            manifest = _coerce_to_manifest(loaded)
            if manifest is not None:
                manifests.append(manifest)
        return manifests

    def discover_manifest_files(self, scan_path: str | Path) -> list[PluginManifest]:
        root = Path(scan_path)
        manifests: list[PluginManifest] = []
        if not root.exists():
            return manifests
        for path in sorted(root.rglob(PLUGIN_MANIFEST_FILENAME)):
            try:
                raw = path.read_text(encoding="utf-8")
                data = json.loads(raw)
                manifest = PluginManifest.from_dict(data)
            except (OSError, ValueError, KeyError, json.JSONDecodeError) as exc:
                logger.warning("malformed plugin manifest %s: %s", path, exc)
                continue
            manifests.append(manifest)
        return manifests

    def _wire_driver(self, manifest: PluginManifest) -> bool:
        register_fn = _resolve_driver_register_function()
        if register_fn is None:
            raise DriverRegistryUnavailableError(
                "driver.registry.register_driver_type is not importable; "
                "the driver-registry extension point must be available to "
                "register driver-kind plugins"
            )
        builder = _resolve_target(manifest.target)
        register_fn(manifest.name, builder)
        return True


def _resolve_driver_register_function():
    """Return ``driver.registry.register_driver_type`` if importable, else None."""
    try:
        module = importlib.import_module("driver.registry")
    except ImportError:
        return None
    fn = getattr(module, "register_driver_type", None)
    if not callable(fn):
        return None
    return fn


def _resolve_target(target: str):
    """Resolve ``module.sub:attribute`` to a callable."""
    if ":" not in target:
        raise PluginTargetResolutionError(
            f"plugin target {target!r} must be 'module.path:attribute'"
        )
    module_path, _, attribute = target.partition(":")
    if not module_path or not attribute:
        raise PluginTargetResolutionError(
            f"plugin target {target!r} must be 'module.path:attribute'"
        )
    try:
        module = importlib.import_module(module_path)
    except ImportError as exc:
        raise PluginTargetResolutionError(
            f"could not import plugin target module {module_path!r}: {exc}"
        ) from exc
    obj = getattr(module, attribute, None)
    if obj is None or not callable(obj):
        raise PluginTargetResolutionError(
            f"plugin target {target!r} did not resolve to a callable"
        )
    return obj


def _coerce_to_manifest(loaded: Any) -> PluginManifest | None:
    if isinstance(loaded, PluginManifest):
        return loaded
    if isinstance(loaded, dict):
        try:
            return PluginManifest.from_dict(loaded)
        except (KeyError, ValueError):
            return None
    if callable(loaded):
        try:
            result = loaded()
        except Exception:  # pragma: no cover - defensive
            return None
        return _coerce_to_manifest(result)
    return None