"""Mount an AIMem bucket into an OpenSandbox sandbox via a host volume. OpenSandbox sandboxes expose no /dev/fuse and no CAP_SYS_ADMIN, so aimem cannot mount inside a sandbox. This script mounts aimem on the HOST (aimem-host-mount.sh) and binds the host mountpoint into the sandbox as a `host` volume at /workspace — the model OpenSandbox already uses for ossfs. Prereqs: an OpenSandbox server using Docker or Kata, with sandbox.toml [storage].allowed_host_paths including /mnt/aimem. See sandbox.toml.example. Env: AIMEM_BUCKET, AIMEM_ENDPOINT, AIMEM_ACCESS_KEY, AIMEM_SECRET_KEY (+ optional AIMEM_SESSION_TOKEN), OPENSANDBOX_DOMAIN, OPENSANDBOX_API_KEY. Run: python main.py Verified end-to-end against a real local `opensandbox-server` (Docker runtime, cgroupv2, `pip install opensandbox-server==0.2.1`) and the `opensandbox` client SDK 0.1.13: host-mount, sandbox creation with this exact Volume/Host shape, and a real read/write round-trip through the bind-mounted FUSE mount all work. Pin to these versions or newer. """ import asyncio import base64 import os import shlex import subprocess import sys import time import uuid from datetime import timedelta from pathlib import Path from opensandbox import Sandbox from opensandbox.models.sandboxes import Host, Volume from opensandbox.sandbox import ConnectionConfig CONTRACT_DIR = Path(__file__).resolve().parents[1] / "agent-worker-runtime" sys.path.insert(0, str(CONTRACT_DIR)) from contract import ( # noqa: E402 ToolEvent, create_symlink_argv, delayed_write_argv, emit_trace, env_probe_argv, path_exists_argv, read_text_base64_argv, redacted_failure, remove_tree_argv, run_async_contract, without_storage_credentials, write_text_argv, ) REQUIRED_ENV = ( "AIMEM_BUCKET", "AIMEM_ENDPOINT", "AIMEM_ACCESS_KEY", "AIMEM_SECRET_KEY", "OPENSANDBOX_DOMAIN", "OPENSANDBOX_API_KEY", ) def require_env() -> None: missing = [name for name in REQUIRED_ENV if not os.environ.get(name)] if missing: names = ", ".join(missing) raise SystemExit(f"missing required environment variables: {names}") def run_host_mount(script: Path, sandbox_id: str) -> str: result = subprocess.run( ["bash", str(script), "mount", sandbox_id], check=True, capture_output=True, text=True, ) return result.stdout.strip().splitlines()[-1] def cleanup_host_mount(script: Path, sandbox_id: str) -> None: subprocess.run(["bash", str(script), "cleanup", sandbox_id], check=False) class OpenSandboxWorkspaceTools: def __init__(self, sandbox: Sandbox, sandbox_id: str, timeout: int = 30) -> None: self.sandbox = sandbox self.sandbox_id = sandbox_id self.timeout = timeout self.trace: list[ToolEvent] = [] async def _exec(self, tool: str, argv: list[str], *, timeout: int | None = None) -> str: call_id = f"tool_{uuid.uuid4().hex}" command_timeout = timeout if timeout is not None else self.timeout command = shlex.join(["timeout", "--signal=KILL", f"{command_timeout}s", *argv]) started = time.monotonic() try: result = await self.sandbox.commands.run(command) except Exception as exc: stderr = f"runner raised {type(exc).__name__}" self.trace.append( ToolEvent( call_id=call_id, tool=tool, exit_code=-1, stdout="", stderr=stderr, duration_ms=round((time.monotonic() - started) * 1000), ) ) raise RuntimeError(redacted_failure(tool, self.sandbox_id, -1, stderr)) from exc stdout = "".join(message.text for message in result.logs.stdout) stderr = "".join(message.text for message in result.logs.stderr) exit_code = result.exit_code if result.exit_code is not None else -1 event = ToolEvent( call_id=call_id, tool=tool, exit_code=exit_code, stdout=stdout, stderr=stderr, duration_ms=round((time.monotonic() - started) * 1000), ) self.trace.append(event) if exit_code != 0: raise RuntimeError(redacted_failure(tool, self.sandbox_id, exit_code, stderr)) return stdout async def read_text(self, path: str) -> str: output = await self._exec( "read_text", read_text_base64_argv("/workspace", path), ) return base64.b64decode(output.strip(), validate=True).decode() async def write_text(self, path: str, content: str) -> None: await self._exec("write_text", write_text_argv("/workspace", path, content)) async def sandbox_env_var_is_set(self, name: str) -> bool: return (await self._exec("sandbox_env_probe", env_probe_argv(name))).strip() == "set" async def create_symlink(self, path: str, target: str) -> None: await self._exec("create_symlink", create_symlink_argv("/workspace", path, target)) async def path_exists(self, path: str) -> bool: output = await self._exec("path_exists", path_exists_argv("/workspace", path)) return output.strip() == "present" async def delayed_write(self, path: str, content: str, delay: int, timeout: int) -> None: await self._exec( "timeout_probe", delayed_write_argv("/workspace", path, content, delay), timeout=timeout, ) async def remove_tree(self, path: str) -> None: await self._exec("security_probe_cleanup", remove_tree_argv("/workspace", path)) async def main() -> None: require_env() sandbox_id = f"aimem-{uuid.uuid4().hex[:8]}" script = Path(__file__).with_name("aimem-host-mount.sh") host_mountpoint = "" try: host_mountpoint = run_host_mount(script, sandbox_id) config = ConnectionConfig( domain=os.environ["OPENSANDBOX_DOMAIN"], api_key=os.environ["OPENSANDBOX_API_KEY"], request_timeout=timedelta(seconds=45), ) sandbox = await Sandbox.create( "ubuntu", connection_config=config, volumes=[ Volume( name="aimem-workspace", host=Host(path=host_mountpoint), mountPath="/workspace", readOnly=False, ), ], ) async with sandbox: tools = OpenSandboxWorkspaceTools(sandbox, sandbox_id) try: with without_storage_credentials(): await run_async_contract( tools, "probe.txt", "hello-from-opensandbox\n", ) finally: emit_trace(tools) print("OPENSANDBOX EXTERNAL WORKER CONTRACT OK") finally: if host_mountpoint: cleanup_host_mount(script, sandbox_id) if __name__ == "__main__": asyncio.run(main())