DataOps-platform/app/core/plugins/runtime.py

"""Fixed, local-only plugin handlers and execution guard.

No submitted artifact is imported or evaluated.  The mapping below is the
complete executable surface for the engineering fixture baseline.
"""

from __future__ import annotations

import hashlib
import time
from typing import Any


class PluginExecutionError(ValueError):
    """A requested invocation cannot pass the fixed execution boundary."""


_HANDLERS = {
    "connector": frozenset({"discover", "snapshot", "evidence"}),
    "parser": frozenset({"parse", "evidence"}),
    "quality": frozenset({"evaluate", "evidence"}),
    "notification": frozenset({"prepare", "evidence"}),
    "approval": frozenset({"evaluate", "evidence"}),
    "agent_mcp": frozenset({"read", "suggest", "evidence"}),
}


class BuiltinFixtureDispatcher:
    """Deterministic hash-only handler for the six fixed plugin categories."""

    def execute(self, manifest: dict[str, Any], *, operation: str, input_digest: str) -> dict[str, Any]:
        if not isinstance(manifest, dict) or manifest.get("distribution", {}).get("kind") != "builtin_fixture" or manifest.get("distribution", {}).get("fixture_id") != "ENGINEERING_EVIDENCE_ONLY":
            raise PluginExecutionError("execution_fixture_denied")
        permissions = manifest.get("permissions")
        if permissions != {"child_process": False, "file": False, "network": False, "secret": False}:
            raise PluginExecutionError("execution_permissions_denied")
        plugin_type = manifest.get("type")
        if plugin_type not in _HANDLERS or operation not in _HANDLERS[plugin_type] or operation not in manifest.get("capabilities", []):
            raise PluginExecutionError("execution_capability_denied")
        if not isinstance(input_digest, str) or len(input_digest) != 64 or set(input_digest) - set("0123456789abcdef"):
            raise PluginExecutionError("execution_input_digest_invalid")
        resource = manifest.get("resource")
        if not isinstance(resource, dict) or set(resource) != {"timeout_ms", "max_output_bytes", "max_concurrency", "max_retries"}:
            raise PluginExecutionError("execution_resource_denied")
        if any(isinstance(value, bool) or not isinstance(value, int) for value in resource.values()) or not 1 <= resource["timeout_ms"] <= 5000 or not 64 <= resource["max_output_bytes"] <= 65536 or not 1 <= resource["max_concurrency"] <= 1 or not 1 <= resource["max_retries"] <= 3:
            raise PluginExecutionError("execution_resource_denied")
        started = time.monotonic_ns()
        output_digest = hashlib.sha256(
            f"ENGINEERING_EVIDENCE_ONLY|{plugin_type}|{operation}|{manifest.get('manifest_digest')}|{input_digest}".encode("ascii")
        ).hexdigest()
        if len(output_digest.encode("ascii")) > resource["max_output_bytes"] or (time.monotonic_ns() - started) > resource["timeout_ms"] * 1_000_000:
            raise PluginExecutionError("execution_resource_exceeded")
        return {"output_digest": output_digest, "fixture_id": "ENGINEERING_EVIDENCE_ONLY", "provider_enabled": False}


__all__ = ["BuiltinFixtureDispatcher", "PluginExecutionError"]