Standard Engines vs Custom Functions:

  Standard engines cover:            Custom functions needed for:
  +---------------------+           +---------------------+
  | Generic rule matching|           | Industry algorithms  |
  | Standard ML models  |           | Custom risk models   |
  | Common constraint   |           | Complex feature eng  |
  | Basic data transform|           | External API integ   |
  +---------------------+           +---------------------+
         |                                  |
         v                                  v
  Covers 80% of scenarios           Covers remaining 20%

FunctionRuntime Architecture:

  Function Definition (FunctionSpec)
       |
       v
  +--------------------------------------+
  |         FunctionRegistry              |
  |  Register / Discover / Version / ACL  |
  +-----------------+--------------------+
                    |
       +------------+------------+
       v            v            v
  +--------+  +--------+  +--------+
  | Python |  |   TS   |  | WASM   | ...
  |Runtime |  |Runtime |  |Runtime |
  +---+----+  +---+----+  +---+----+
      |           |           |
      v           v           v
  +--------------------------------+
  |        Sandbox Layer            |
  |  Quotas | Timeout | Net Isol   |
  +--------------------------------+

from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any


class FunctionLanguage(Enum):
    PYTHON = "python"
    TYPESCRIPT = "typescript"
    GROOVY = "groovy"
    WASM = "wasm"
    KOTLIN = "kotlin"


class FunctionStatus(Enum):
    DRAFT = "draft"
    ACTIVE = "active"
    DEPRECATED = "deprecated"
    DISABLED = "disabled"


@dataclass
class FunctionSpec:
    """Function specification"""
    function_id: str
    name: str
    language: FunctionLanguage
    version: str
    source_code: str
    entry_point: str
    description: str = ""
    input_schema: dict[str, Any] = field(default_factory=dict)
    output_schema: dict[str, Any] = field(default_factory=dict)
    dependencies: list[str] = field(default_factory=list)
    resource_limits: ResourceLimits = field(default_factory=lambda: ResourceLimits())
    tags: list[str] = field(default_factory=list)
    status: FunctionStatus = FunctionStatus.DRAFT
    created_at: datetime = field(default_factory=datetime.utcnow)
    updated_at: datetime = field(default_factory=datetime.utcnow)


@dataclass
class ResourceLimits:
    """Resource limits"""
    max_memory_mb: int = 256
    max_cpu_ms: int = 5000
    max_execution_time_ms: int = 30000
    max_output_size_kb: int = 1024
    network_enabled: bool = False
    filesystem_read_only: bool = True
    max_file_descriptors: int = 64


@dataclass
class FunctionInvocation:
    """Function invocation record"""
    invocation_id: str
    function_id: str
    inputs: dict[str, Any]
    outputs: dict[str, Any] | None = None
    status: str = "pending"
    error: str = ""
    duration_ms: float = 0.0
    memory_used_mb: float = 0.0
    started_at: datetime | None = None
    completed_at: datetime | None = None

from abc import ABC, abstractmethod


class LanguageRuntime(ABC):
    """Language runtime abstract base"""

    @abstractmethod
    async def initialize(self, spec: FunctionSpec) -> None:
        ...

    @abstractmethod
    async def execute(self, spec: FunctionSpec,
                       inputs: dict[str, Any]) -> dict[str, Any]:
        ...

    @abstractmethod
    async def cleanup(self, spec: FunctionSpec) -> None:
        ...

    @abstractmethod
    def supported_language(self) -> FunctionLanguage:
        ...

import subprocess
import json
import tempfile
import os


class PythonRuntime(LanguageRuntime):
    """Python function runtime"""

    def __init__(self, sandbox_type: str = "nsjail"):
        self._sandbox_type = sandbox_type
        self._venv_cache: dict[str, str] = {}

    def supported_language(self) -> FunctionLanguage:
        return FunctionLanguage.PYTHON

    async def initialize(self, spec: FunctionSpec) -> None:
        venv_dir = f"/tmp/fn-venv/{spec.function_id}/{spec.version}"
        os.makedirs(venv_dir, exist_ok=True)

        subprocess.run(
            ["python", "-m", "venv", venv_dir],
            check=True, timeout=30,
        )

        if spec.dependencies:
            pip = os.path.join(venv_dir, "bin", "pip")
            subprocess.run(
                [pip, "install", "--no-cache-dir"] + spec.dependencies,
                check=True, timeout=120,
            )

        self._venv_cache[spec.function_id] = venv_dir

    async def execute(self, spec: FunctionSpec,
                       inputs: dict[str, Any]) -> dict[str, Any]:
        wrapper = self._build_wrapper(spec, inputs)

        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".py", delete=False
        ) as f:
            f.write(wrapper)
            script_path = f.name

        try:
            venv_dir = self._venv_cache.get(spec.function_id, "")
            python_bin = os.path.join(venv_dir, "bin", "python") if venv_dir else "python"

            result = subprocess.run(
                [python_bin, script_path],
                capture_output=True, text=True,
                timeout=spec.resource_limits.max_execution_time_ms / 1000,
            )

            if result.returncode != 0:
                raise RuntimeError(f"Function failed: {result.stderr}")

            return json.loads(result.stdout)
        finally:
            os.unlink(script_path)

    def _build_wrapper(self, spec: FunctionSpec,
                        inputs: dict[str, Any]) -> str:
        return f"""
import json
import sys

{spec.source_code}

inputs = json.loads('''{json.dumps(inputs)}''')
result = {spec.entry_point}(**inputs)
print(json.dumps(result))
"""

    async def cleanup(self, spec: FunctionSpec) -> None:
        import shutil
        venv_dir = self._venv_cache.pop(spec.function_id, None)
        if venv_dir and os.path.exists(venv_dir):
            shutil.rmtree(venv_dir)

class TypeScriptRuntime(LanguageRuntime):
    """TypeScript function runtime"""

    def supported_language(self) -> FunctionLanguage:
        return FunctionLanguage.TYPESCRIPT

    async def initialize(self, spec: FunctionSpec) -> None:
        work_dir = f"/tmp/fn-ts/{spec.function_id}/{spec.version}"
        os.makedirs(work_dir, exist_ok=True)

        package = {
            "name": spec.function_id,
            "version": spec.version,
            "type": "module",
            "dependencies": {
                dep.split("@")[0]: dep.split("@")[1] if "@" in dep else "latest"
                for dep in spec.dependencies
            },
        }
        with open(os.path.join(work_dir, "package.json"), "w") as f:
            json.dump(package, f)

        subprocess.run(
            ["npm", "install", "--production"],
            cwd=work_dir, check=True, timeout=120,
        )

    async def execute(self, spec: FunctionSpec,
                       inputs: dict[str, Any]) -> dict[str, Any]:
        wrapper = f"""
const inputs = {json.dumps(inputs)};

{spec.source_code}

const result = {spec.entry_point}(inputs);
Promise.resolve(result).then(r => {{
    process.stdout.write(JSON.stringify(r));
}});
"""

        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".mjs", delete=False
        ) as f:
            f.write(wrapper)
            script_path = f.name

        try:
            result = subprocess.run(
                ["node", "--experimental-vm-modules", script_path],
                capture_output=True, text=True,
                timeout=spec.resource_limits.max_execution_time_ms / 1000,
            )

            if result.returncode != 0:
                raise RuntimeError(f"Function failed: {result.stderr}")

            return json.loads(result.stdout)
        finally:
            os.unlink(script_path)

    async def cleanup(self, spec: FunctionSpec) -> None:
        import shutil
        work_dir = f"/tmp/fn-ts/{spec.function_id}/{spec.version}"
        if os.path.exists(work_dir):
            shutil.rmtree(work_dir)

class FunctionRegistry:
    """Function registry"""

    def __init__(self, store):
        self._store = store
        self._runtimes: dict[FunctionLanguage, LanguageRuntime] = {}

    def register_runtime(self, runtime: LanguageRuntime) -> None:
        self._runtimes[runtime.supported_language()] = runtime

    async def register_function(self, spec: FunctionSpec) -> str:
        self._validate_spec(spec)

        runtime = self._runtimes.get(spec.language)
        if runtime is None:
            raise ValueError(f"Unsupported language: {spec.language}")

        await runtime.initialize(spec)
        spec.status = FunctionStatus.ACTIVE
        await self._store.save(spec)
        return spec.function_id

    async def invoke(self, function_id: str,
                      inputs: dict[str, Any]) -> FunctionInvocation:
        spec = await self._store.get(function_id)
        if spec is None:
            raise ValueError(f"Function not found: {function_id}")
        if spec.status != FunctionStatus.ACTIVE:
            raise ValueError(f"Function not active: {function_id}")

        runtime = self._runtimes.get(spec.language)
        invocation = FunctionInvocation(
            invocation_id=f"inv-{function_id}-{datetime.utcnow().timestamp()}",
            function_id=function_id,
            inputs=inputs,
            started_at=datetime.utcnow(),
        )

        try:
            invocation.status = "running"
            import time
            start = time.monotonic()
            outputs = await runtime.execute(spec, inputs)
            invocation.outputs = outputs
            invocation.status = "completed"
            invocation.duration_ms = (time.monotonic() - start) * 1000
        except TimeoutError:
            invocation.status = "timeout"
            invocation.error = "Execution timeout"
        except Exception as e:
            invocation.status = "failed"
            invocation.error = str(e)

        invocation.completed_at = datetime.utcnow()
        return invocation

    def _validate_spec(self, spec: FunctionSpec) -> None:
        if not spec.function_id:
            raise ValueError("function_id is required")
        if not spec.source_code:
            raise ValueError("source_code is required")
        if not spec.entry_point:
            raise ValueError("entry_point is required")
        if spec.resource_limits.max_memory_mb > 1024:
            raise ValueError("max_memory_mb cannot exceed 1024")

class TypeMapper:
    """Ontology type to language type mapping"""

    _mappings = {
        "string": {
            FunctionLanguage.PYTHON: "str",
            FunctionLanguage.TYPESCRIPT: "string",
            FunctionLanguage.KOTLIN: "String",
            FunctionLanguage.GROOVY: "String",
        },
        "integer": {
            FunctionLanguage.PYTHON: "int",
            FunctionLanguage.TYPESCRIPT: "number",
            FunctionLanguage.KOTLIN: "Int",
            FunctionLanguage.GROOVY: "Integer",
        },
        "float": {
            FunctionLanguage.PYTHON: "float",
            FunctionLanguage.TYPESCRIPT: "number",
            FunctionLanguage.KOTLIN: "Double",
            FunctionLanguage.GROOVY: "Double",
        },
        "boolean": {
            FunctionLanguage.PYTHON: "bool",
            FunctionLanguage.TYPESCRIPT: "boolean",
            FunctionLanguage.KOTLIN: "Boolean",
            FunctionLanguage.GROOVY: "Boolean",
        },
        "datetime": {
            FunctionLanguage.PYTHON: "datetime",
            FunctionLanguage.TYPESCRIPT: "Date",
            FunctionLanguage.KOTLIN: "Instant",
            FunctionLanguage.GROOVY: "Instant",
        },
        "object": {
            FunctionLanguage.PYTHON: "dict",
            FunctionLanguage.TYPESCRIPT: "Record<string, unknown>",
            FunctionLanguage.KOTLIN: "Map<String, Any>",
            FunctionLanguage.GROOVY: "Map<String, Object>",
        },
    }

    @classmethod
    def map_type(cls, ontology_type: str,
                  language: FunctionLanguage) -> str:
        type_map = cls._mappings.get(ontology_type, {})
        return type_map.get(language, "Any")

class FunctionDecisionPlugin:
    """Function as a decision engine plugin"""

    def __init__(self, registry: FunctionRegistry):
        self._registry = registry

    async def evaluate_with_function(self, function_id: str,
                                       context) -> dict:
        invocation = await self._registry.invoke(
            function_id, context.inputs
        )

        if invocation.status == "completed":
            return {
                "decision": invocation.outputs.get("decision", "unknown"),
                "confidence": invocation.outputs.get("confidence", 0.0),
                "source": "custom_function",
                "function_id": function_id,
                "duration_ms": invocation.duration_ms,
            }
        else:
            return {
                "decision": "error",
                "confidence": 0.0,
                "error": invocation.error,
            }

syntax = "proto3";
package onto.function.v1;

service FunctionService {
    rpc RegisterFunction(RegisterRequest) returns (RegisterResponse);
    rpc InvokeFunction(InvokeRequest) returns (InvokeResponse);
    rpc ListFunctions(ListRequest) returns (ListResponse);
    rpc GetFunction(GetRequest) returns (FunctionDetail);
    rpc UpdateFunction(UpdateRequest) returns (UpdateResponse);
    rpc DeprecateFunction(DeprecateRequest) returns (DeprecateResponse);
}

message RegisterRequest {
    string name = 1;
    string language = 2;
    string version = 3;
    string source_code = 4;
    string entry_point = 5;
    string input_schema_json = 6;
    string output_schema_json = 7;
    repeated string dependencies = 8;
    ResourceLimits limits = 9;
}

message InvokeRequest {
    string function_id = 1;
    string inputs_json = 2;
}

message InvokeResponse {
    string invocation_id = 1;
    string status = 2;
    string outputs_json = 3;
    double duration_ms = 4;
    string error = 5;
}

Function Runtime Security Layers:

  Layer 1: Code Review
  +-- Static analysis: ban os.system, subprocess
  +-- Dependency allowlist: only approved packages
  |
  Layer 2: Sandbox Isolation
  +-- nsjail / gVisor container sandbox
  +-- Read-only filesystem
  +-- Network disabled (default)
  |
  Layer 3: Resource Limits
  +-- CPU time cap
  +-- Memory cap
  +-- File descriptor cap
  |
  Layer 4: Output Validation
  +-- Output size limit
  +-- Output schema validation

# Register a custom credit scoring function

spec = FunctionSpec(
    function_id="fn-credit-custom-001",
    name="custom_credit_scorer",
    language=FunctionLanguage.PYTHON,
    version="1.0.0",
    entry_point="score",
    source_code="""
def score(credit_score: int, annual_income: float,
          debt_ratio: float, employment_years: int) -> dict:
    base_score = credit_score / 850 * 40
    income_score = min(annual_income / 500000, 1.0) * 25
    debt_penalty = max(0, (debt_ratio - 0.3) * 50)
    tenure_bonus = min(employment_years / 10, 1.0) * 10

    final_score = base_score + income_score - debt_penalty + tenure_bonus

    if final_score >= 60:
        decision = "approve"
    elif final_score >= 40:
        decision = "conditional_approve"
    else:
        decision = "reject"

    return {
        "decision": decision,
        "confidence": min(final_score / 100, 1.0),
        "score_breakdown": {
            "base": base_score,
            "income": income_score,
            "debt_penalty": debt_penalty,
            "tenure_bonus": tenure_bonus,
            "final": final_score,
        },
    }
""",
    dependencies=["numpy"],
    resource_limits=ResourceLimits(max_memory_mb=128, max_execution_time_ms=5000),
)

registry = FunctionRegistry(store)
await registry.register_function(spec)

result = await registry.invoke("fn-credit-custom-001", {
    "credit_score": 680,
    "annual_income": 250000,
    "debt_ratio": 0.35,
    "employment_years": 8,
})
# result.outputs = {"decision": "conditional_approve", "confidence": 0.57, ...}