orchestrata_ultimate.plugin.manifest

Reference › orchestrata_ultimate › plugin › manifest

The plugin.yaml contract. PluginManifest is the validated description of a plugin: name, version, entry point, description, author, license, the capabilities it needs, plugin and Python dependencies, an optional configuration schema, the minimum orchestrata version (default 0.1.0), Prolog rule files, an optional KGSchemaExtension and the directory it came from. KGSchemaExtension lists the node types, relationship types, indexes, constraints and a migration version a plugin wants to add to the knowledge graph.

PluginManifest.from_yaml(path) reads a file and from_dict(data, source_path) validates a dictionary. Validation raises PluginManifestError when a required field is missing, the name is not kebab-case starting with a letter, the version is not semantic, or a capability is not in the known set (the set is the one declared in interfaces). The loader builds manifests from files, the registry stores them, and governance and the context judge a plugin by its declared capabilities.

At a glance

Class diagram

Dependency diagram

API

Classes

KGSchemaExtension (dataclass)

Decorators: traces(spec='015', story='US-006', reqs=['REQ-015-071', 'REQ-015-074']), dataclass · Line 38

Knowledge graph schema extension declared by a plugin.

PluginManifest (dataclass)

Decorators: traces(spec='015', story='US-013', reqs=['REQ-015-172']), dataclass · Line 58

Parsed and validated plugin manifest from plugin.yaml.

Source

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

"""PluginManifest — YAML manifest parser and validator.

Parses plugin.yaml files into a validated PluginManifest dataclass.
Validates required fields, capability declarations, and version
format before allowing a plugin to be loaded.
"""

import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

import yaml

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.plugin.exceptions import PluginManifestError

_SEMVER_PATTERN = re.compile(r"^\d+\.\d+\.\d+(?:-[\w.]+)?(?:\+[\w.]+)?$")
_NAME_PATTERN = re.compile(r"^[a-z][a-z0-9-]*$")

_VALID_CAPABILITIES = {
    "di_access",
    "redis",
    "neo4j",
    "network",
    "filesystem",
    "prolog",
    "event_bus",
    "provider",
    "gateway",
}

_REQUIRED_FIELDS = {"name", "version", "entry_point"}


@traces(spec="015", story="US-006", reqs=["REQ-015-071", "REQ-015-074"])
@dataclass
class KGSchemaExtension:
    """Knowledge graph schema extension declared by a plugin.

    Attributes:
        node_types: New node labels to create.
        relationship_types: New relationship types.
        indexes: Index definitions.
        constraints: Constraint definitions.
        migration_version: Schema migration version number.
    """

    node_types: list[str] = field(default_factory=list)
    relationship_types: list[str] = field(default_factory=list)
    indexes: list[str] = field(default_factory=list)
    constraints: list[str] = field(default_factory=list)
    migration_version: int = 1


@traces(spec="015", story="US-013", reqs=["REQ-015-172"])
@dataclass
class PluginManifest:
    """Parsed and validated plugin manifest from plugin.yaml.

    Attributes:
        name: Unique plugin name (kebab-case).
        version: Semantic version string.
        description: Human-readable description.
        author: Author name or organization.
        license: SPDX license identifier.
        entry_point: Python import path to IPlugin class.
        capabilities: Required system capabilities.
        dependencies: Other plugin names (load ordering).
        python_dependencies: Pip package specs.
        config_schema: Optional JSON Schema for plugin config.
        min_orchestrata_version: Minimum compatible version.
        prolog_rules: Relative paths to Prolog rule files.
        kg_schema: Optional knowledge graph schema extension.
        source_path: Filesystem path where manifest was found.
    """

    name: str
    version: str
    entry_point: str
    description: str = ""
    author: str = ""
    license: str = ""
    capabilities: list[str] = field(default_factory=list)
    dependencies: list[str] = field(default_factory=list)
    python_dependencies: list[str] = field(default_factory=list)
    config_schema: dict | None = None
    min_orchestrata_version: str = "0.1.0"
    prolog_rules: list[str] = field(default_factory=list)
    kg_schema: KGSchemaExtension | None = None
    source_path: Path | None = None

    @traces(spec="015", story="US-001", reqs=["REQ-015-004"])
    @staticmethod
    def from_yaml(path: Path) -> PluginManifest:
        """Parse and validate a plugin.yaml file.

        Args:
            path: Path to the plugin.yaml file.

        Returns:
            Validated PluginManifest instance.

        Raises:
            PluginManifestError: If the file is missing, malformed,
                or fails validation.
        """
        if not path.exists():
            raise PluginManifestError(
                f"Manifest file not found: {path}",
                context={"path": str(path)},
            )
        try:
            raw = yaml.safe_load(path.read_text(encoding="utf-8"))
        except yaml.YAMLError as exc:
            raise PluginManifestError(
                f"Malformed YAML in {path}: {exc}",
                context={"path": str(path)},
            ) from exc

        if not isinstance(raw, dict):
            raise PluginManifestError(
                f"Manifest must be a YAML mapping, got {type(raw).__name__}",
                context={"path": str(path)},
            )

        return PluginManifest.from_dict(raw, source_path=path.parent)

    @traces(spec="015", story="US-001", reqs=["REQ-015-004"])
    @staticmethod
    def from_dict(
        data: dict[str, Any],
        source_path: Path | None = None,
    ) -> PluginManifest:
        """Create a PluginManifest from a dictionary.

        Args:
            data: Dictionary with manifest fields.
            source_path: Directory containing the manifest.

        Returns:
            Validated PluginManifest instance.

        Raises:
            PluginManifestError: If required fields are missing
                or values are invalid.
        """
        missing = _REQUIRED_FIELDS - set(data.keys())
        if missing:
            raise PluginManifestError(
                f"Missing required fields: {sorted(missing)}",
                context={"missing": sorted(missing)},
            )

        name = data["name"]
        if not _NAME_PATTERN.match(name):
            raise PluginManifestError(
                f"Invalid plugin name '{name}': must be kebab-case "
                f"starting with a letter",
                context={"name": name},
            )

        version = data["version"]
        if not _SEMVER_PATTERN.match(version):
            raise PluginManifestError(
                f"Invalid version '{version}': must be semver",
                context={"version": version},
            )

        capabilities = data.get("capabilities", [])
        invalid_caps = set(capabilities) - _VALID_CAPABILITIES
        if invalid_caps:
            raise PluginManifestError(
                f"Unknown capabilities: {sorted(invalid_caps)}",
                context={"invalid": sorted(invalid_caps)},
            )

        kg_data = data.get("kg_schema")
        kg_schema = None
        if kg_data and isinstance(kg_data, dict):
            kg_schema = KGSchemaExtension(
                node_types=kg_data.get("node_types", []),
                relationship_types=kg_data.get("relationship_types", []),
                indexes=kg_data.get("indexes", []),
                constraints=kg_data.get("constraints", []),
                migration_version=kg_data.get("migration_version", 1),
            )

        return PluginManifest(
            name=name,
            version=version,
            entry_point=data["entry_point"],
            description=data.get("description", ""),
            author=data.get("author", ""),
            license=data.get("license", ""),
            capabilities=capabilities,
            dependencies=data.get("dependencies", []),
            python_dependencies=data.get("python_dependencies", []),
            config_schema=data.get("config_schema"),
            min_orchestrata_version=data.get("min_orchestrata_version", "0.1.0"),
            prolog_rules=data.get("prolog_rules", []),
            kg_schema=kg_schema,
            source_path=source_path,
        )