Skip to content

wasm_runner

wasm_runner

WASM sandbox — lightweight isolation via Wasmtime.

Classes

WasmResult dataclass 

WasmResult(success: bool = True, output: str = '', duration_seconds: float = 0.0, fuel_consumed: int = 0, memory_used_bytes: int = 0)

Result from a WASM execution.

WasmRunner 

WasmRunner(*, fuel_limit: int = 1000000, memory_limit_mb: int = 256, timeout: float = 30.0)

Execute WASM modules with resource limits.

Uses wasmtime-py for sub-100ms isolation. Supplements Docker-based ContainerRunner for lightweight, fast sandboxing.

Requires: uv sync --extra sandbox-wasm

Source code in src/openjarvis/sandbox/wasm_runner.py 
29
30
31
32
33
34
35
36
37
38
def __init__(
    self,
    *,
    fuel_limit: int = 1_000_000,
    memory_limit_mb: int = 256,
    timeout: float = 30.0,
) -> None:
    self._fuel_limit = fuel_limit
    self._memory_limit_mb = memory_limit_mb
    self._timeout = timeout
Functions
available staticmethod 
available() -> bool

Check if wasmtime is available.

Source code in src/openjarvis/sandbox/wasm_runner.py 
40
41
42
43
44
45
46
47
@staticmethod
def available() -> bool:
    """Check if wasmtime is available."""
    try:
        import wasmtime  # noqa: F401
        return True
    except ImportError:
        return False
run 
run(wasm_bytes: bytes, input_data: Optional[Dict[str, Any]] = None) -> WasmResult

Execute a WASM module with input data.

The module is expected to export a run function that takes a pointer and length and returns a pointer and length. For simpler modules, we attempt to call _start (WASI).

Source code in src/openjarvis/sandbox/wasm_runner.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def run(
    self,
    wasm_bytes: bytes,
    input_data: Optional[Dict[str, Any]] = None,
) -> WasmResult:
    """Execute a WASM module with input data.

    The module is expected to export a ``run`` function that takes
    a pointer and length and returns a pointer and length.
    For simpler modules, we attempt to call ``_start`` (WASI).
    """
    try:
        import wasmtime
    except ImportError:
        return WasmResult(
            success=False,
            output=(
                "wasmtime not installed. Install with: "
                "uv sync --extra sandbox-wasm"
            ),
        )

    t0 = time.time()
    try:
        # Configure engine with fuel metering
        config = wasmtime.Config()
        config.consume_fuel = True
        engine = wasmtime.Engine(config)

        # Create store with fuel limit
        store = wasmtime.Store(engine)
        store.set_fuel(self._fuel_limit)

        # Compile module
        module = wasmtime.Module(engine, wasm_bytes)

        # Set up WASI if needed
        wasi_config = wasmtime.WasiConfig()
        wasi_config.inherit_stdout()
        wasi_config.inherit_stderr()
        store.set_wasi(wasi_config)

        # Create linker and link WASI
        linker = wasmtime.Linker(engine)
        linker.define_wasi()

        # Instantiate
        instance = linker.instantiate(store, module)

        # Try to call _start (WASI entry point)
        start_func = instance.exports(store).get("_start")
        if start_func and isinstance(start_func, wasmtime.Func):
            start_func(store)

        fuel_remaining = store.get_fuel()
        fuel_consumed = self._fuel_limit - fuel_remaining

        duration = time.time() - t0
        return WasmResult(
            success=True,
            output="WASM module executed successfully.",
            duration_seconds=duration,
            fuel_consumed=fuel_consumed,
        )

    except Exception as exc:
        duration = time.time() - t0
        return WasmResult(
            success=False,
            output=f"WASM execution error: {exc}",
            duration_seconds=duration,
        )
validate 
validate(wasm_bytes: bytes) -> bool

Validate that bytes represent a valid WASM module.

Source code in src/openjarvis/sandbox/wasm_runner.py 
122
123
124
125
126
127
128
129
130
131
def validate(self, wasm_bytes: bytes) -> bool:
    """Validate that bytes represent a valid WASM module."""
    try:
        import wasmtime
        config = wasmtime.Config()
        engine = wasmtime.Engine(config)
        wasmtime.Module.validate(engine, wasm_bytes)
        return True
    except Exception:
        return False

Functions

create_sandbox_runner 

create_sandbox_runner(config: Any = None) -> Any

Factory: select Docker or WASM runner based on config/availability.

Source code in src/openjarvis/sandbox/wasm_runner.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def create_sandbox_runner(config: Any = None) -> Any:
    """Factory: select Docker or WASM runner based on config/availability."""
    if config and getattr(config, "runtime", "") == "wasm":
        runner = WasmRunner(
            fuel_limit=getattr(config, "wasm_fuel_limit", 1_000_000),
            memory_limit_mb=getattr(config, "wasm_memory_limit_mb", 256),
            timeout=getattr(config, "timeout", 30),
        )
        if runner.available():
            return runner

    # Fall back to Docker ContainerRunner
    try:
        from openjarvis.sandbox.runner import ContainerRunner
        return ContainerRunner(
            image=getattr(config, "image", "openjarvis-sandbox:latest"),
            timeout=getattr(config, "timeout", 300),
        )
    except ImportError:
        return None