Plugin Runtime¶

The VYRA plugin system allows modules to load and execute sandboxed WASM plugins at runtime. Python host functions are exposed to the plugin sandbox so that plugins can call back into the host application.

Overview ¶

Two backends are available:

Backend	When selected	Dependency
`WasmRuntime`	`wasmtime` installed and `.wasm` file exists	`pip install wasmtime`
`StubRuntime`	All other cases (pure Python fallback)	None

create_plugin_runtime() selects the backend automatically.

Quick Start ¶

from pathlib import Path
from vyra_base.plugin.runtime import create_plugin_runtime

runtime = create_plugin_runtime(
    plugin_id="my-plugin",
    wasm_path=Path("/opt/plugins/my-plugin/logic.wasm"),
    initial_state={"counter": 0},
)
await runtime.start()

result = await runtime.call("increment", {"step": 1})
await runtime.stop()

Plugin File Layout ¶

my_plugin/
├── logic.wasm          # Compiled WASM module
└── metadata.json       # Plugin manifest

metadata.json format:

{
    "id": "my-plugin",
    "version": "1.0.0",
    "description": "An example counter plugin",
    "exports": ["increment", "reset", "get_value"],
    "imports": ["host_log", "host_emit_event"],
    "initial_state": {
        "counter": {"type": "int", "default": 0},
        "step":    {"type": "int", "default": 1}
    }
}

Host Functions ¶

Host functions are Python callbacks that the plugin can invoke from within the WASM sandbox. Subclass BaseHostFunctions and prefix methods with host_:

from vyra_base.plugin.host_functions import BaseHostFunctions

class MyHostFunctions(BaseHostFunctions):
    def host_log(self, message: str) -> None:
        print(f"[plugin] {message}")

    def host_emit_event(self, event_type: str, payload: str) -> None:
        print(f"[event:{event_type}] {payload}")

runtime = create_plugin_runtime(
    plugin_id="my-plugin",
    wasm_path=...,
    host=MyHostFunctions(),
)

Use NullHostFunctions when no callbacks are needed.

API Reference ¶

vyra_base.plugin.runtime.create_plugin_runtime(plugin_id, wasm_path, host=None, initial_state=None, prefer_stub=False)[Quellcode]¶

Factory-Funktion: Gibt WasmRuntime zurück wenn wasmtime installiert ist und die .wasm-Datei existiert. Andernfalls StubRuntime.

Parameter:

plugin_id (str) – Plugin-ID (z.B. „my-plugin“)
wasm_path (str | Path) – Pfad zur logic.wasm Datei (neben ihr muss metadata.json liegen)
host (HostFunctions | None) – Host-Funktionen-Implementierung (optional, sonst NullHostFunctions)
initial_state (dict[str, Any] | None) – Startzustand — Keys müssen zu den args des ‚init‘-Exports passen
prefer_stub (bool) – Erzwinge StubRuntime auch wenn wasmtime verfügbar ist

Rückgabetyp:

PluginRuntime

Rückgabe:

WasmRuntime oder StubRuntime Instanz

Beispiel:

runtime = create_plugin_runtime(
    plugin_id     = "my-plugin",
    wasm_path     = "/opt/vyra/plugin_pool/my-plugin/1.0.0/logic.wasm",
    host          = my_host_functions,
    initial_state = {"initial_value": 0, "step": 1},
)
await runtime.start()
result = await runtime.call("increment", {"step": 2})

class vyra_base.plugin.runtime.WasmRuntime(plugin_id, wasm_path, host=None, initial_state=None)[Quellcode]¶

Bases: PluginRuntime

Echter WASM-Executor für VYRA-Plugins via wasmtime.

Lädt das kompilierte logic.wasm und liest die zur WASM-Datei gehörende metadata.json. Aus metadata.json["exports"] werden Funktionssignaturen dynamisch gecacht — kein hartkodiertes Wissen über einzelne Plugin-Funktionen.

Aufruf-Konvention (Metadata-driven i32):

metadata.json enthält exports[] mit Funktionsnamen und arg-Definitionen
call(function_name, data) mappt data-Keys in Reihenfolge der args auf i32
Fehlende Keys → 0; überschüssige Keys → ignoriert
Rückgabe: WASM-Funktionen mit i32-Rückgabe → {"result": <int>}
ping ist eingebaut (kein WASM-Export nötig)

Nur verfügbar wenn wasmtime installiert ist. Nutze create_plugin_runtime() für automatische Auswahl.

Parameter:

plugin_id (str)
wasm_path (str | Path)
host (HostFunctions | None)
initial_state (dict[str, Any] | None)

__init__(plugin_id, wasm_path, host=None, initial_state=None)[Quellcode]¶

Parameter:

plugin_id (str)
wasm_path (str | Path)
host (HostFunctions | None)
initial_state (dict[str, Any] | None)

Rückgabetyp:

None

async start()[Quellcode]¶

Lädt das WASM-Modul und initialisiert die Runtime.

Rückgabetyp:: None

async stop()[Quellcode]¶

Gibt Ressourcen der Runtime frei.

Rückgabetyp:: None

async call(function_name, data)[Quellcode]¶

Ruft eine exportierte Funktion des Plugins auf.

Parameter:

function_name (str) – Name der WASM-Funktion (z.B. „increment“, „get_state“)
data (dict[str, Any]) – Eingabe-Parameter als JSON-serialisierbares Dict

Rückgabetyp:

dict[str, Any]

Rückgabe:

Rückgabe-Wert als Dict

Verursacht:

PluginCallError – Bei Fehler im Plugin oder nicht unterstützter Funktion

async on_event(event_name, data)[Quellcode]¶

Leitet ein externes Event (Zenoh/Redis) an das Plugin weiter.

Parameter:

event_name (str) – Event-Name (z.B. „zenoh.message“, „module.state_changed“)
data (dict[str, Any]) – Event-Payload

Rückgabetyp:

None

is_running()¶

Gibt an, ob die Runtime aktiv ist.

Rückgabetyp:: bool

class vyra_base.plugin.runtime.StubRuntime(plugin_id, wasm_path, host=None, initial_state=None)[Quellcode]¶

Bases: PluginRuntime

Python-Stub-Implementierung der PluginRuntime.

Fallback für Umgebungen ohne wasmtime oder wenn die WASM-Datei fehlt. Eignet sich für Connectivity-Tests und Infrastruktur-Validierung ohne echte Plugin-Logik.

Unterstützte Funktionen (generisch für alle Plugins):

ping — Lebenszeichen zurückgeben
get_state — Internen State zurückgeben
set_state — Internen State setzen

Für alle anderen Funktionsnamen wird ein PluginCallError geworfen, da Plugin-spezifische Logik ausschließlich in der WasmRuntime ausgeführt wird. Unterklassen können _dispatch() erweitern um eigene Test-Stubs hinzuzufügen.

Parameter:

plugin_id (str)
wasm_path (str | Path)
host (HostFunctions | None)
initial_state (dict[str, Any] | None)

__init__(plugin_id, wasm_path, host=None, initial_state=None)[Quellcode]¶

Parameter:

plugin_id (str)
wasm_path (str | Path)
host (HostFunctions | None)
initial_state (dict[str, Any] | None)

Rückgabetyp:

None

async start()[Quellcode]¶

Lädt das WASM-Modul und initialisiert die Runtime.

Rückgabetyp:: None

async stop()[Quellcode]¶

Gibt Ressourcen der Runtime frei.

Rückgabetyp:: None

async call(function_name, data)[Quellcode]¶

Ruft eine exportierte Funktion des Plugins auf.

Parameter:

function_name (str) – Name der WASM-Funktion (z.B. „increment“, „get_state“)
data (dict[str, Any]) – Eingabe-Parameter als JSON-serialisierbares Dict

Rückgabetyp:

dict[str, Any]

Rückgabe:

Rückgabe-Wert als Dict

Verursacht:

PluginCallError – Bei Fehler im Plugin oder nicht unterstützter Funktion

async on_event(event_name, data)[Quellcode]¶

Leitet ein externes Event (Zenoh/Redis) an das Plugin weiter.

Parameter:

event_name (str) – Event-Name (z.B. „zenoh.message“, „module.state_changed“)
data (dict[str, Any]) – Event-Payload

Rückgabetyp:

None

is_running()¶

Gibt an, ob die Runtime aktiv ist.

Rückgabetyp:: bool

class vyra_base.plugin.host_functions.BaseHostFunctions[Quellcode]¶

Bases: ABC

Abstrakte Basisklasse für modul-spezifische HostFunctions-Implementierungen.

Implementiert log() konkret über Python-Logging. notify_ui, create_publisher, create_subscriber, create_server, create_client müssen von der Unterklasse implementiert werden.

Kommunikation läuft ausschließlich über die InterfaceFactory — kein direkter Zenoh-Zugriff. Der PluginFacade (vyra_base.plugin.plugin_facade) sitzt oberhalb und prüft Berechtigungen aus metadata.json vor jeder Operation.

Interface-Referenz:: Der plugin_event-Publisher wird in vyra_base/interfaces/config/vyra_plugin.meta.json beschrieben. Modulspezifische Services stehen in {module}_interfaces/config/{module}_plugin.meta.json.

Beispiel:

class MyModuleHostFunctions(BaseHostFunctions):
    def __init__(self, plugin_event_publisher):
        super().__init__()
        self._pub = plugin_event_publisher

    async def notify_ui(self, event_name: str, data: dict) -> None:
        await self._pub.publish({"event_name": event_name, "data": data})

    async def create_publisher(self, name, module_name, module_id=None, **kw):
        return await InterfaceFactory.create_publisher(
            name, module_id=module_id, module_name=module_name, **kw)

    async def create_subscriber(self, name, callback, module_name, module_id=None, **kw):
        return await InterfaceFactory.create_subscriber(
            name, subscriber_callback=callback,
            module_id=module_id, module_name=module_name, **kw)

    async def create_server(self, name, callback, **kw):
        return await InterfaceFactory.create_server(
            name, response_callback=callback, **kw)

    async def create_client(self, name, module_name, module_id=None, **kw):
        return await InterfaceFactory.create_client(
            name, module_id=module_id, module_name=module_name, **kw)