hms-mediaengine/_entwicklung/packages/plugin_sdk/hms_plugin_sdk/lifecycle.py

"""Plugin-Lifecycle-Manager (PLAN.md §14.5, §14.6, §26.3).

Lebenszyklus:

discovered → validated → installed → enabled → compiled → active
                                     ↘ quarantined / incompatible

Regeln:
- Übergänge nur entlang definierter Kanten; Sprünge sind Fehler
- Validierung umfasst Manifest-Schema, API-Kompatibilität,
  Pfadsicherheit, Backend-/Shader-Prüfung (§14.5)
- ein fehlerhaftes Plugin wird quarantiniert, ohne ein Projekt unbrauchbar
  zu machen (§3.5); der Effekt wird überbrückt (bypass_on_error)
- Show-Lock (§26.3): Installieren/Updaten ist gesperrt; Enable/Disable
  und Parameter bleiben erlaubt (§17.7 Live-Modus)
- Doppelte Plugin-IDs sind Fehler, keine stillen Überschreibungen
"""

from __future__ import annotations

from dataclasses import dataclass, field
from enum import StrEnum
from pathlib import Path

from hms_plugin_sdk.manifest import load_manifest


class LifecycleState(StrEnum):
    """Zustände gemäß §14.5."""

    DISCOVERED = "discovered"
    VALIDATED = "validated"
    INSTALLED = "installed"
    ENABLED = "enabled"
    COMPILED = "compiled"
    ACTIVE = "active"
    QUARANTINED = "quarantined"
    INCOMPATIBLE = "incompatible"
    DISABLED = "disabled"


# Erlaubte Übergänge (§14.5-Lebenszyklusgraph)
_TRANSITIONS: dict[LifecycleState, frozenset[LifecycleState]] = {
    LifecycleState.DISCOVERED: frozenset(
        {LifecycleState.VALIDATED, LifecycleState.INCOMPATIBLE, LifecycleState.QUARANTINED}
    ),
    LifecycleState.VALIDATED: frozenset(
        {LifecycleState.INSTALLED, LifecycleState.INCOMPATIBLE, LifecycleState.QUARANTINED}
    ),
    LifecycleState.INSTALLED: frozenset(
        {LifecycleState.ENABLED, LifecycleState.DISABLED, LifecycleState.QUARANTINED}
    ),
    LifecycleState.ENABLED: frozenset(
        {LifecycleState.COMPILED, LifecycleState.DISABLED, LifecycleState.QUARANTINED}
    ),
    LifecycleState.COMPILED: frozenset(
        {LifecycleState.ACTIVE, LifecycleState.QUARANTINED, LifecycleState.DISABLED}
    ),
    LifecycleState.ACTIVE: frozenset(
        {LifecycleState.DISABLED, LifecycleState.QUARANTINED}
    ),
    LifecycleState.DISABLED: frozenset(
        {LifecycleState.ENABLED, LifecycleState.QUARANTINED}
    ),
    LifecycleState.QUARANTINED: frozenset(),  # manuelle Entfernung/Neuinstallation
    LifecycleState.INCOMPATIBLE: frozenset(),
}


class InvalidTransitionError(Exception):
    """Unerlaubter Zustandsübergang im Lebenszyklus."""


@dataclass
class PluginRecord:
    """Ein Plugin im Lifecycle-Manager."""

    plugin_id: str
    version: str
    state: LifecycleState = LifecycleState.DISCOVERED
    package_hash: str | None = None
    last_error: str | None = None
    manifest: dict = field(default_factory=dict)


class PluginLifecycleManager:
    """Verwaltet den Lebenszyklus aller installierten Plugins (§14.5).

    - discover(): Verzeichnis scannen, Manifest laden, Manifest-Validierung
    - advance(): zustandsgeprüfter Übergang
    - quarantine(): Fehlerfall mit Ursache (§3.5, §14.5)
    - Show-Lock: install_validate/enable_new blockiert Strukturänderungen
      (§26.3); Aktivieren bereits installierter Plugins bleibt erlaubt
    """

    def __init__(self, show_lock: bool = False) -> None:
        self._plugins: dict[str, PluginRecord] = {}
        self._show_lock = show_lock

    @property
    def show_lock(self) -> bool:
        return self._show_lock

    def set_show_lock(self, enabled: bool) -> None:
        """§26.3: Show-Lock verhindert Plugininstallation/-update."""
        self._show_lock = enabled

    # ---------- Discovery & Validierung (§14.5) ----------

    def discover(self, plugin_dir: Path) -> list[str]:
        """Scannt ein Verzeichnis mit Plugin-Ordnern.

        Lädt plugin.json, validiert es (inkl. Shader-Existenz) und
        überführt jedes Plugin in VALIDATED oder INCOMPATIBLE/QUARANTINED.
        Rückgabe: Liste diagnostizierter Fehler (leer = alles sauber).
        """
        errors: list[str] = []
        if not plugin_dir.is_dir():
            return [f"Plugin-Verzeichnis fehlt: {plugin_dir}"]
        for child in sorted(plugin_dir.iterdir()):
            if not child.is_dir():
                continue
            manifest_path = child / "plugin.json"
            if not manifest_path.is_file():
                continue  # kein Plugin-Ordner (z. B. .git)
            manifest, validation_errors = load_manifest(child)
            if validation_errors:
                pid = manifest.get("id") or child.name
                errors.extend(f"{pid}: {e}" for e in validation_errors)
                record = PluginRecord(
                    plugin_id=pid,
                    version=str(manifest.get("version", "0.0.0")),
                    state=LifecycleState.QUARANTINED,
                    last_error="; ".join(validation_errors),
                    manifest=manifest,
                )
                self._upsert(record)
                continue
            pid = manifest["id"]
            record = PluginRecord(
                plugin_id=pid,
                version=manifest["version"],
                state=LifecycleState.VALIDATED,
                manifest=manifest,
            )
            self._upsert(record)
        return errors

    # ---------- Übergänge (§14.5) ----------

    def advance(self, plugin_id: str, new_state: LifecycleState) -> LifecycleState:
        """Führt einen Lifecycle-Übergang aus; wirft bei illegaler Kante.

        Show-Lock (§26.3): Übergänge, die Installation/Update bedeuten
        (INSTALLED von DISCOVERED/VALIDATED), sind gesperrt. Aktivieren
        bereits installierter Plugins (ENABLED/COMPILED/ACTIVE) bleibt
        erlaubt (§17.7 Live: Layerparameter und Livefunktionen nutzbar).
        """
        record = self._plugins.get(plugin_id)
        if record is None:
            raise KeyError(f"unbekanntes Plugin {plugin_id!r}")
        current = record.state
        if new_state not in _TRANSITIONS[current]:
            raise InvalidTransitionError(
                f"{plugin_id}: Übergang {current.value} → {new_state.value} nicht erlaubt"
            )
        if (
            self._show_lock
            and new_state is LifecycleState.INSTALLED
            and current in (LifecycleState.DISCOVERED, LifecycleState.VALIDATED)
        ):
            raise InvalidTransitionError(
                f"{plugin_id}: Installation im Show-Lock gesperrt (§26.3)"
            )
        record.state = new_state
        record.last_error = None
        return record.state

    def quarantine(self, plugin_id: str, reason: str) -> None:
        """Fehlerfall: Plugin überbrücken, Projekt bleibt nutzbar (§3.5)."""
        record = self._plugins.get(plugin_id)
        if record is None:
            raise KeyError(f"unbekanntes Plugin {plugin_id!r}")
        record.state = LifecycleState.QUARANTINED
        record.last_error = reason

    # ---------- Abfragen ----------

    def get(self, plugin_id: str) -> PluginRecord | None:
        return self._plugins.get(plugin_id)

    def by_state(self, state: LifecycleState) -> list[PluginRecord]:
        return [r for r in self._plugins.values() if r.state is state]

    def active_plugins(self) -> list[PluginRecord]:
        return self.by_state(LifecycleState.ACTIVE)

    def all(self) -> list[PluginRecord]:
        return list(self._plugins.values())

    def compatible_backends(self, plugin_id: str) -> frozenset[str]:
        """Gemeinsame Backends: deklariert im Manifest und lokal verfügbar."""
        record = self._plugins.get(plugin_id)
        if record is None:
            return frozenset()
        declared = set(record.manifest.get("entrypoints", {}))
        supported = set(
            record.manifest.get("capabilities", {}).get("supported_backends", [])
        )
        return frozenset(declared & supported)

    # ---------- Interna ----------

    def _upsert(self, record: PluginRecord) -> None:
        existing = self._plugins.get(record.plugin_id)
        if existing is not None and existing.version != record.version:
            # Versionskonflikt: neuer Stand gewinnt nur ohne Show-Lock.
            # Ein Update startet einen neuen Zyklus: discover() hat das
            # neue Manifest bereits validiert (§14.5), also ist VALIDATED
            # der korrekte Zustand – eine erneute Installation ist nötig,
            # der alte INSTALLED/ACTIVE-Status gilt nicht mehr.
            if self._show_lock:
                raise InvalidTransitionError(
                    f"{record.plugin_id}: Versionswechsel {existing.version} → "
                    f"{record.version} im Show-Lock gesperrt (§26.3)"
                )
        self._plugins[record.plugin_id] = record