"""Sandbox implementation for isolated code execution."""
import argparse
import logging
import subprocess
import sys
import tempfile
import os
import json
from dataclasses import dataclass
from typing import Optional, Dict, Any, List, Union, Callable
import yr
from yr.config import InvokeOptions, PortForwarding
from yr.runtime_holder import global_runtime
from yr.config_manager import ConfigManager
from yr.sandbox.filesystem import CpDirection, SandboxFilesystem, _get_gateway_host
logger = logging.getLogger(__name__)
def _sanitize_instance_id(instance_id: str) -> str:
"""Sanitize instance ID to match TraefikRegistry::SanitizeID (C++).
Rules: @ -> -at-, / . _ -> -, truncate to 200 chars.
"""
result = instance_id
pos = 0
while True:
pos = result.find("@", pos)
if pos == -1:
break
result = result[:pos] + "-at-" + result[pos + 1:]
pos += 4
result = result.replace("/", "-").replace(".", "-").replace("_", "-")
if len(result) > 200:
result = result[:200]
return result
def _build_gateway_url(instance_id: str, sandbox_port: int, gateway_host: str, path: str = "") -> str:
"""Build Gateway HTTP path URL: http://{gateway_host}/{safeID}/{sandbox_port}{path}.
URL format must match TraefikRegistry::RegisterInstance in function_proxy:
- {safeID} is sanitized instance ID (SanitizeID logic)
- {sandbox_port} is the original sandbox port
- Full path format: /{safeID}/{sandbox_port}
See: functionsystem/src/function_proxy/local_scheduler/traefik_registry/traefik_registry.cpp
"""
safe_id = _sanitize_instance_id(instance_id)
base = f"http://{gateway_host}/{safe_id}/{sandbox_port}"
if path:
path = path if path.startswith("/") else f"/{path}"
return f"{base}{path}"
return base
def _print_gateway_urls(instance_id: str, port_forwardings: List["PortForwarding"]) -> None:
"""Print Gateway URLs for port forwardings after sandbox creation."""
if not port_forwardings:
return
gateway_host = _get_gateway_host()
if not gateway_host:
print("Warning: cannot print port forwarding URLs: YR_GATEWAY_ADDRESS or YR_SERVER_ADDRESS not set")
return
print("sandbox created, port forwarding URLs:")
for pf in port_forwardings:
url = _build_gateway_url(instance_id, pf.port, gateway_host)
print(f" port {pf.port}: {url}")
@yr.instance
class SandboxInstance:
"""
SandboxInstance class provides isolated environment for code execution.
This class creates a sandboxed environment where code can be executed
with limited permissions and resource constraints.
This is the underlying instance class decorated with @yr.instance.
Users should typically use the Sandbox wrapper class instead.
"""
def __init__(
self, working_dir: Optional[str] = None, env: Optional[Dict[str, str]] = None
):
"""
Initialize the Sandbox instance.
Args:
working_dir (Optional[str]): The working directory for sandbox execution.
If None, a temporary directory will be created.
env (Optional[Dict[str, str]]): Environment variables for the sandbox.
If None, inherits from parent process.
"""
if working_dir is None:
self._temp_dir = tempfile.TemporaryDirectory(prefix="yr_sandbox_")
self.working_dir = self._temp_dir.name
self._temp_dir_created = True
else:
self._temp_dir = None
self.working_dir = working_dir
self._temp_dir_created = False
self.env = env if env is not None else os.environ.copy()
self._initialized = True
self.__yr_before_snapshot__ = None
self.__yr_after_snapstart__ = None
def __del__(self):
"""Destructor to ensure cleanup on object deletion."""
self.cleanup()
@staticmethod
def get_internal_urls() -> Dict[int, str]:
"""Return internal cluster URLs for port-forwarded services.
Reads environment variables injected by Runtime Manager to build
direct-access URLs that bypass the Traefik gateway. Other sandbox
instances can call this method via RPC to discover how to reach
this sandbox's forwarded ports on the internal network.
Returns:
Dict[int, str]: Mapping from container port to internal URL.
e.g. {8080: "https://192.0.2.1:40001", 9090: "https://192.0.2.1:40002"}.
Returns an empty dict if no port forwarding is configured.
"""
host_ip = os.environ.get("YR_INTERNAL_HOST_IP", "")
pf_str = os.environ.get("YR_PORT_FORWARDINGS", "")
if not host_ip or not pf_str:
return {}
result = {}
for mapping in pf_str.split(";"):
parts = mapping.split(":")
if len(parts) >= 3:
protocol = parts[0].lower()
host_port = parts[1]
container_port = int(parts[2])
scheme = "https" if protocol == "https" else "http"
result[container_port] = f"{scheme}://{host_ip}:{host_port}"
return result
@staticmethod
def start_tunnel_server(ws_port: int = 8765, http_port: int = 8766) -> None:
"""Start TunnelServer in a background thread within this sandbox instance."""
import asyncio
import threading
from yr.sandbox.tunnel_server import TunnelServer
ready = threading.Event()
error = [None]
def _run():
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
server = TunnelServer(ws_port=ws_port, http_port=http_port)
try:
loop.run_until_complete(server.start())
except Exception as e:
error[0] = e
ready.set()
return
ready.set()
loop.run_forever()
t = threading.Thread(target=_run, name="tunnel-server", daemon=True)
t.start()
if not ready.wait(timeout=5.0):
logger.warning("TunnelServer not ready within 5s, continuing anyway")
if error[0] is not None:
raise RuntimeError(f"TunnelServer failed to start: {error[0]}") from error[0]
@staticmethod
def get_name():
"""
Get the name of the sandbox instance.
Returns:
str: The name of the sandbox instance.
"""
return os.environ.get("INSTANCE_ID", "")
def execute(
self,
command: Union[str, List[str]],
working_dir: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
timeout: Optional[int] = None,
) -> Dict[str, Any]:
"""
Execute a command in the sandbox environment.
Args:
command (Union[str, List[str]]): The command to execute.
working_dir (Optional[str]): Working dir for command execution.
env (Optional[Dict[str, str]]): Environment variables for command execution.
timeout (Optional[int]): Timeout in seconds for command execution.
If None, no timeout is set.
Returns:
Dict[str, Any]: A dictionary containing:
- returncode (int): The return code of the command.
- stdout (str): Standard output of the command.
- stderr (str): Standard error of the command.
Raises:
RuntimeError: If the sandbox is not initialized.
subprocess.TimeoutExpired: If the command execution times out.
Examples:
>>> sandbox = yr.sandbox.Sandbox.invoke()
>>> result = yr.get(sandbox.execute.invoke("ls -la"))
>>> print(result['stdout'])
"""
if not self._initialized:
raise RuntimeError("Sandbox is not initialized")
if isinstance(command, str):
cmd_args = ["/bin/sh", "-c", command]
elif isinstance(command, list):
if len(command) == 0:
return {
"returncode": -1,
"stdout": "",
"stderr": "Error: cmd list cannot be empty",
}
if not all(isinstance(arg, str) for arg in command):
return {
"returncode": -1,
"stdout": "",
"stderr": "Error: All elements in command list must be strings",
}
cmd_args = command
else:
return {
"returncode": -1,
"stdout": "",
"stderr": f"Error: cmd must be a string or a list of strings, got {type(command).__name__}",
}
if not cmd_args:
return {
"returncode": -1,
"stdout": "",
"stderr": "Error: cmd list cannot be empty",
}
try:
subprocess_cwd = self.working_dir
if working_dir is not None:
subprocess_cwd = working_dir
subprocess_env = self.env
if env is not None:
subprocess_env = env
result = subprocess.run(
args=cmd_args,
shell=False,
cwd=subprocess_cwd,
env=subprocess_env,
capture_output=True,
text=True,
timeout=timeout,
)
return {
"returncode": result.returncode,
"stdout": sys.version + '\n' + result.stdout,
"stderr": result.stderr,
}
except subprocess.TimeoutExpired as e:
return {
"returncode": -1,
"stdout": e.stdout if e.stdout else "",
"stderr": f"Command timed out after {timeout} seconds",
}
except Exception as e:
return {"returncode": -1, "stdout": "", "stderr": str(e)}
def get_working_dir(self) -> str:
"""
Get the working directory of the sandbox.
Returns:
str: The path to the working directory.
"""
return self.working_dir
def cleanup(self) -> None:
"""
Cleanup the sandbox environment.
This method removes temporary files and directories created by the sandbox.
"""
temp_dir_created = getattr(self, "_temp_dir_created", False)
working_dir = getattr(self, "working_dir", None)
temp_dir = getattr(self, "_temp_dir", None)
if temp_dir_created and working_dir and os.path.exists(working_dir):
import shutil
try:
if temp_dir is not None:
temp_dir.cleanup()
else:
shutil.rmtree(working_dir)
except Exception as e:
print(
f"Warning: Failed to cleanup sandbox directory {working_dir}: {e}"
)
def register_before_snapshot_hook(self, hook_func: Callable[..., Any]):
"""Register a hook to be called before snapshot."""
self.__yr_before_snapshot__ = hook_func
def register_after_snapstart_hook(self, hook_func: Callable[..., Any]):
"""Register a hook to be called after snapstart."""
self.__yr_after_snapstart__ = hook_func
@dataclass
class SandboxCreateOptions:
name: Optional[str] = None
rootfs: Optional[str] = None
env: Optional[Dict[str, str]] = None
idle_timeout: int = 300
working_dir: Optional[str] = None
cpu: Optional[int] = None
memory: Optional[int] = None
extra_config: Optional[Dict[str, Any]] = None
ports: Optional[List[str]] = None
upstream: Optional[str] = None
proxy_port: int = 8766
sandbox_type: str = ""
before_checkpoint_func: Optional[Callable[..., Any]] = None
after_restore_func: Optional[Callable[..., Any]] = None
def create(
*args: str,
name: Optional[str] = None,
rootfs: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
idle_timeout: int = 300,
working_dir: Optional[str] = None,
cpu: Optional[int] = None,
memory: Optional[int] = None,
extra_config: Optional[Dict[str, Any]] = None,
ports: Optional[str] = None,
upstream: Optional[str] = None,
proxy_port: int = 8766,
sandbox_type: str = "",
before_checkpoint_func: Optional[Callable[..., Any]] = None,
after_restore_func: Optional[Callable[..., Any]] = None,
):
"""
Create a new Sandbox instance.
Args:
working_dir: Working directory for sandbox execution.
env: Environment variables for the sandbox.
port: Additional port forwarding rules.
upstream: Local service address to tunnel to, e.g. "upstream.example.com:8000".
When set, starts a reverse tunnel so sandbox code can reach the
local service via http://127.0.0.1:{proxy_port}.
proxy_port: Port B — the HTTP proxy port inside the sandbox
(default 8766). Port A (WS tunnel) = proxy_port - 1.
Returns:
Sandbox wrapper instance.
Examples:
>>> import yr
>>> yr.init()
>>>
>>> sandbox = yr.sandbox.create()
>>> # Create sandbox with supervisor executor
>>> sandbox = yr.sandbox.create(sandbox_type="supervisor")
>>> result = yr.get(sandbox.exec("pwd"))
>>> print(result['stdout'])
>>>
>>> sandbox.terminate()
>>> yr.finalize()
"""
try:
return Sandbox(
name=name,
rootfs=rootfs,
cpu=cpu,
memory=memory,
idle_timeout=idle_timeout,
working_dir=working_dir,
env=env,
extra_config=extra_config,
ports=ports,
upstream=upstream,
proxy_port=proxy_port,
sandbox_type=sandbox_type,
before_checkpoint_func=before_checkpoint_func,
after_restore_func=after_restore_func,
)
except Exception as e:
print(f"failed to create, exception: {e}")
return None
def restore(
checkpoint_id: str,
before_checkpoint_func: Optional[Callable[..., Any]] = None,
after_restore_func: Optional[Callable[..., Any]] = None,
):
"""
Restore a Sandbox from a previously created checkpoint.
Args:
checkpoint_id (str): The checkpoint ID returned by a previous checkpoint() call.
before_checkpoint_func (Optional[Callable]): Hook called before future checkpoints.
after_restore_func (Optional[Callable]): Hook called after restore completes.
Returns:
Sandbox wrapper instance restored from the checkpoint.
Examples:
>>> import yr
>>> yr.init()
>>>
>>> sandbox = yr.sandbox.create()
>>> checkpoint_id = sandbox.checkpoint()
>>> restored = yr.sandbox.restore(checkpoint_id)
>>> result = restored.exec("echo hello")
>>> restored.terminate()
>>> yr.finalize()
"""
return Sandbox(
checkpoint_id=checkpoint_id,
before_checkpoint_func=before_checkpoint_func,
after_restore_func=after_restore_func,
)
class Sandbox:
"""
Sandbox wrapper class for convenient sandbox operations.
When upstream is provided, starts a reverse tunnel:
- Port B (proxy_port, loopback): sandbox code calls http://127.0.0.1:{proxy_port}
- Port A (proxy_port-1, 0.0.0.0): WebSocket tunnel endpoint registered with Traefik
Examples:
>>> import yr
>>> yr.init()
>>>
>>> # Basic sandbox
>>> sb = yr.sandbox.Sandbox()
>>> result = yr.get(sb.exec("echo hello"))
>>> print(result['stdout'])
>>>
>>> # Sandbox with reverse tunnel to local service
>>> sb = yr.sandbox.create(upstream="upstream.example.com:8000")
>>> url = sb.get_tunnel_url() # "http://127.0.0.1:8766"
>>> result = yr.get(sb.exec(f"curl {url}/api/data"))
>>>
>>> sb.terminate()
>>> yr.finalize()
"""
def __init__(
self,
checkpoint_id: Optional[str] = None,
name: Optional[str] = None,
rootfs: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
idle_timeout: int = 300,
working_dir: Optional[str] = None,
cpu: Optional[int] = None,
memory: Optional[int] = None,
extra_config: Optional[Dict[str, Any]] = None,
ports: Optional[List[str]] = None,
upstream: Optional[str] = None,
proxy_port: int = 8766,
sandbox_type: str = "",
before_checkpoint_func: Optional[Callable[..., Any]] = None,
after_restore_func: Optional[Callable[..., Any]] = None,
):
"""
Initialize the Sandbox wrapper.
Args:
checkpoint_id (Optional[str]): If provided, restore from this checkpoint
instead of creating a new instance.
name (Optional[str]): Name of the sandbox instance.
rootfs (Optional[str]): Root filesystem image for the sandbox.
env (Optional[Dict[str, str]]): Environment variables for the sandbox.
idle_timeout (int): Idle timeout in seconds. Default 300.
working_dir (Optional[str]): Working directory inside the sandbox.
cpu (Optional[int]): CPU resource limit in millicores.
memory (Optional[int]): Memory resource limit in MB.
extra_config (Optional[Dict[str, Any]]): Additional configuration.
ports (Optional[List[str]]): Port forwarding configurations.
upstream (Optional[str]): Local service address to tunnel to.
proxy_port (int): HTTP proxy port inside the sandbox. Default 8766.
before_checkpoint_func (Optional[Callable]): Hook called before checkpoint.
after_restore_func (Optional[Callable]): Hook called after restore.
"""
self._forwarded_ports = set()
self._tunnel_client = None
self._proxy_port = proxy_port
self._filesystem: Optional["SandboxFilesystem"] = None
if checkpoint_id is None:
self.create_new_instance(SandboxCreateOptions(
name=name,
rootfs=rootfs,
env=env,
idle_timeout=idle_timeout,
working_dir=working_dir,
cpu=cpu,
memory=memory,
extra_config=extra_config,
ports=ports,
upstream=upstream,
proxy_port=proxy_port,
sandbox_type=sandbox_type,
before_checkpoint_func=before_checkpoint_func,
after_restore_func=after_restore_func,
))
else:
self.restore_instance(checkpoint_id=checkpoint_id)
def __del__(self):
"""
Destructor to ensure cleanup and termination on object deletion.
Automatically calls cleanup() and terminate() when the Sandbox object is deleted.
"""
try:
if hasattr(self, "_instance") and self._instance is not None:
yr.get(self.cleanup())
self.terminate()
except Exception:
logger.debug("ignore sandbox cleanup failure during destructor", exc_info=True)
@property
def filesystem(self) -> "SandboxFilesystem":
"""Namespace for sandbox filesystem copy operations.
Example::
sb.filesystem.copy_from_local("/local/data.csv", "/sandbox/data.csv")
sb.filesystem.copy_to_local("/sandbox/output.txt", "/local/output.txt")
"""
if self._filesystem is None:
self._filesystem = SandboxFilesystem(self)
return self._filesystem
@staticmethod
def get_exec_result(exec_ref):
return yr.get(exec_ref)
def create_new_instance(self, options: SandboxCreateOptions):
"""
Initialize the Sandbox wrapper.
Args:
working_dir (Optional[str]): The working directory for sandbox execution.
If None, a temporary directory will be created.
env (Optional[Dict[str, str]]): Environment variables for the sandbox.
If None, inherits from parent process.
ports (Optional[str]): List of port forwarding
configurations. Each string specifies a port to be forwarded
inside the sandbox environment with format protocol:port.
sandbox_type (str): Type of sandbox executor.
Supported values:
- "supervisor": Uses SUPERVISOR executor
- "": Use default executor (RUNTIME)
"""
name = options.name
rootfs = options.rootfs
env = options.env
idle_timeout = options.idle_timeout
working_dir = options.working_dir
cpu = options.cpu
memory = options.memory
extra_config = options.extra_config
ports = options.ports
upstream = options.upstream
proxy_port = options.proxy_port
sandbox_type = options.sandbox_type
before_checkpoint_func = options.before_checkpoint_func
after_restore_func = options.after_restore_func
self._proxy_port = proxy_port
self._tunnel_client = None
opt = yr.InvokeOptions()
opt.skip_serialize = True
opt.idle_timeout = idle_timeout
if sandbox_type:
opt.custom_extensions["sandbox_type"] = sandbox_type
if name is not None:
opt.name = name
if cpu is not None:
opt.cpu = cpu
if memory is not None:
opt.memory = memory
if env is not None:
opt.env_vars = env
if working_dir is not None:
opt.runtime_env["working_dir"] = working_dir
if extra_config is not None:
opt.custom_extensions["extra_config"] = json.dumps(extra_config)
if rootfs is not None:
opt.custom_extensions["rootfs"] = rootfs
if ports is not None:
yr_port_forwardings = []
for port_forward in ports:
parts = port_forward.split(":")
if len(parts) == 2:
protocol, port_str = parts
try:
port = int(port_str)
except ValueError as e:
raise ValueError(
f"Invalid port number: '{port_str}' in '{port_forward}'. "
"Port must be a valid integer."
) from e
yr_port_forwardings.append(PortForwarding(port=port, protocol=protocol.upper()))
elif len(parts) == 1:
try:
port = int(parts[0])
except ValueError as e:
raise ValueError(
f"Invalid port number: '{parts[0]}'. "
"Port must be a valid integer."
) from e
yr_port_forwardings.append(PortForwarding(port=port, protocol="TCP"))
else:
raise ValueError(
f"Invalid port_forwarding format: {port_forward}. "
"Expected format: protocol:port (e.g., tcp:8080)"
)
opt.port_forwardings = yr_port_forwardings
self._forwarded_ports = set()
if ports is not None:
for port_forward in ports:
parts = port_forward.split(":")
if len(parts) >= 1:
port_str = parts[-1]
try:
port = int(port_str)
self._forwarded_ports.add(port)
except ValueError:
pass
if upstream is not None:
tunnel_port = proxy_port - 1
tunnel_pf = yr.PortForwarding(port=tunnel_port)
opt.port_forwardings = (list(opt.port_forwardings) if opt.port_forwardings else []) + [tunnel_pf]
self._instance = SandboxInstance.options(opt).invoke(working_dir, env)
if opt.port_forwardings:
instance_id = yr.get(self._instance.get_name.invoke())
_print_gateway_urls(instance_id, opt.port_forwardings)
yr.get(self._instance.start_tunnel_server.invoke(tunnel_port, proxy_port))
instance_id = yr.get(self._instance.get_name.invoke())
gateway_host = _get_gateway_host()
tunnel_url = _build_gateway_url(instance_id, tunnel_port, gateway_host)
tunnel_ws_url = tunnel_url.replace("https://", "wss://").replace("http://", "ws://")
from yr.sandbox.tunnel_client import TunnelClient
self._tunnel_client = TunnelClient(upstream)
print(f"[INFO] Connecting to tunnel: {tunnel_ws_url}")
if self._tunnel_client.start(tunnel_ws_url, timeout=10.0):
print("[OK] TunnelClient connected successfully")
else:
print("[WARN] TunnelClient connection timeout, will retry in background")
return
self._instance = SandboxInstance.options(opt).invoke(working_dir, env)
try:
instance_id = yr.get(self._instance.get_name.invoke())
if opt.port_forwardings:
_print_gateway_urls(instance_id, opt.port_forwardings)
except Exception as e:
raise RuntimeError(f"Failed to initialize sandbox instance: {e}") from e
if before_checkpoint_func is not None:
yr.get(self._instance.register_before_snapshot_hook.invoke(before_checkpoint_func))
if after_restore_func is not None:
yr.get(self._instance.register_after_snapstart_hook.invoke(after_restore_func))
def exec(
self,
command: str,
working_dir: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
timeout: Optional[int] = None,
):
"""
Execute a command in the sandbox environment.
Args:
*args (str): The command to execute.
working_dir (Optional[str]): Working dir for command execution.
env (Optional[Dict[str, str]]): Environment variables for command execution.
timeout (Optional[int]): Timeout in seconds for command execution.
If None, no timeout is set.
Returns:
A dictionary containing:
- returncode (int): The return code of the command.
- stdout (str): Standard output of the command.
- stderr (str): Standard error of the command.
"""
exec_ref = self._exec(
command=command,
working_dir=working_dir,
env=env,
timeout=timeout,
)
return self.get_exec_result(exec_ref)
def exec_async(
self,
command: str,
working_dir: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
timeout: Optional[int] = None,
):
"""
Execute a command in the sandbox environment asynchronously.
Args:
command (str): The command to execute.
working_dir (Optional[str]): Working dir for command execution.
env (Optional[Dict[str, str]]): Environment variables for command execution.
timeout (Optional[int]): Timeout in seconds for command execution.
If None, no timeout is set.
Returns:
ObjectRef: Reference to the execution result that can be used with self.get_exec_result().
"""
return self._exec(
command=command,
working_dir=working_dir,
env=env,
timeout=timeout,
)
def get_working_dir(self):
"""Get the working directory of the sandbox."""
return self._instance.get_working_dir.invoke()
def cleanup(self):
"""Cleanup temp files in the sandbox."""
return self._instance.cleanup.invoke()
def terminate(self):
"""
Terminate the sandbox instance.
Stop tunnel client (if any) and terminate the sandbox instance.
"""
if self._tunnel_client is not None:
self._tunnel_client.stop()
self._tunnel_client = None
self._instance.terminate()
def get_tunnel_url(self) -> str:
"""Return the internal HTTP proxy URL for sandbox code to call.
Returns:
str: e.g. "http://127.0.0.1:8766"
Raises:
RuntimeError: if no upstream was configured.
"""
if self._tunnel_client is None:
raise RuntimeError("No upstream configured. Pass upstream= to create().")
return f"http://127.0.0.1:{self._proxy_port}"
def get_internal_urls(self) -> Dict[int, str]:
"""Return internal cluster URLs for port-forwarded services.
Other sandbox instances can use these URLs to reach this sandbox's
forwarded ports on the internal network.
Returns:
Dict[int, str]: Mapping from container port to internal URL.
e.g. {8080: "https://192.0.2.1:40001", 9090: "https://192.0.2.1:40002"}
Returns an empty dict if no port forwarding is configured.
"""
return yr.get(self._instance.get_internal_urls.invoke())
def get_tunnel(self, port: int) -> str:
"""
Get the tunnel URL for a forwarded port.
Args:
port (int): The port number to get the tunnel for.
Returns:
str: The tunnel URL in format http://{TraefikAddress}/{real_id}/{port}
Raises:
ValueError: If the port is not in the list of forwarded ports.
"""
if port not in self._forwarded_ports:
raise ValueError(f"Invalid port: {port}. Port is not in the list of forwarded ports.")
logical_id = self._instance.instance_id
real_id = global_runtime.get_runtime().get_real_instance_id(logical_id)
traefik_address = os.environ.get("YR_GATEWAY_ADDRESS", "")
if traefik_address == "":
raise ValueError(f"YR_GATEWAY_ADDRESS is not set.")
return f"http://{traefik_address}/{real_id}/{port}"
def checkpoint(
self,
ttl: int = -1,
leave_running: bool = False,
) -> str:
"""
Create a checkpoint of the current sandbox state.
This triggers a snapshot of the underlying instance. The returned checkpoint ID
can be used with ``yr.sandbox.restore()`` or ``Sandbox(checkpoint_id=...)`` to
create a new sandbox with the same state.
Args:
ttl (int): Time-to-live for the checkpoint in seconds. -1 means no expiration.
Default -1.
leave_running (bool): If True, the sandbox continues running after
checkpointing. If False, the sandbox is terminated after checkpoint.
Default False.
Returns:
str: The checkpoint ID that uniquely identifies this snapshot.
Raises:
RuntimeError: If the sandbox instance is not active.
Examples:
>>> sandbox = yr.sandbox.create(rootfs="python:3.12-slim")
>>> sandbox.exec("echo setup done")
>>> checkpoint_id = sandbox.checkpoint(leave_running=True)
>>> print(f"Checkpoint: {checkpoint_id}")
"""
checkpoint_id = self._instance.snapshot(
ttl=ttl,
leave_running=leave_running,
)
return checkpoint_id
def restore_instance(self, checkpoint_id: str):
"""
Restore the sandbox from a checkpoint.
Uses ``InstanceCreator.snapstart()`` to create a new instance from the
checkpoint, then verifies the instance is ready.
Args:
checkpoint_id (str): The checkpoint ID returned by a previous
``checkpoint()`` call.
Raises:
RuntimeError: If the restore or readiness check fails.
"""
self._instance = SandboxInstance.snapstart(checkpoint_id=checkpoint_id)
try:
ref = self._instance.get_working_dir.invoke()
yr.get(ref)
except Exception as e:
raise RuntimeError(
f"Failed to restore sandbox from checkpoint {checkpoint_id}: {e}"
) from e
def cp(
self,
src: str,
dst: str,
direction: CpDirection = CpDirection.UPLOAD,
streaming: Optional[bool] = None,
) -> None:
"""Copy a file or directory to or from the sandbox.
The *direction* parameter controls which side is local and which is the
sandbox:
* ``CpDirection.UPLOAD`` *(default)* – ``src`` is a **local** path,
``dst`` is the destination path **inside the sandbox**.
* ``CpDirection.DOWNLOAD`` – ``src`` is a path **inside the sandbox**,
``dst`` is the **local** destination path.
Prefer :attr:`filesystem` for explicit, self-documenting code::
# Equivalent to sb.cp("/local/file", "/remote/file")
sb.filesystem.copy_from_local("/local/file", "/remote/file")
# Equivalent to sb.cp("/remote/file", "/local/file", direction=CpDirection.DOWNLOAD)
sb.filesystem.copy_to_local("/remote/file", "/local/file")
Args:
src: Source path (local when uploading, sandbox path when downloading).
dst: Destination path (sandbox path when uploading, local when downloading).
direction: Transfer direction. Defaults to :attr:`CpDirection.UPLOAD`.
streaming: Transfer mode override.
* ``True`` – always use gzip streaming (best for large compressible data).
* ``False`` – always use non-streaming (best for small or binary data).
* ``None`` – auto-select based on file size and compressibility (default).
Raises:
FileNotFoundError: *src* does not exist on the local machine (upload only).
RuntimeError: Server address is not configured.
Examples:
>>> sb = yr.sandbox.create()
>>> # Upload a local file to the sandbox (default direction)
>>> sb.cp("/local/data.csv", "/sandbox/data.csv")
>>> # Download a file from the sandbox
>>> from yr.sandbox.filesystem import CpDirection
>>> sb.cp("/sandbox/output.txt", "/local/output.txt", direction=CpDirection.DOWNLOAD)
>>> # Upload a directory
>>> sb.cp("/local/project/", "/sandbox/project/")
"""
getattr(self.filesystem, "_cp")(src, dst, direction=direction, streaming=streaming)
def _exec(
self,
command: str,
working_dir: Optional[str] = None,
env: Optional[Dict[str, str]] = None,
timeout: Optional[int] = None,
):
return self._instance.execute.invoke(
command=command,
working_dir=working_dir,
env=env,
timeout=timeout,
)
def main():
parser = argparse.ArgumentParser(description="Create a detached sandbox instance")
parser.add_argument(
"--name", type=str, default=None, help="Name of the sandbox instance"
)
parser.add_argument(
"--namespace",
type=str,
default="detached.sandbox",
help="Namespace for the sandbox instance",
)
args = parser.parse_args()
os.environ.pop("YR_WORKING_DIR", None)
cfg = yr.Config()
cfg.in_cluster = True
yr.init(cfg)
try:
opt = yr.InvokeOptions()
opt.custom_extensions["lifecycle"] = "detached"
opt.idle_timeout = 60 * 60 * 24 * 7
opt.cpu = 1000
opt.memory = 2048
opt.name = args.name
opt.namespace = args.namespace
opt.skip_serialize = True
if not opt.name:
import uuid
opt.name = str(uuid.uuid4())
sandbox = SandboxInstance.options(opt).invoke()
try:
name = yr.get(sandbox.get_name.invoke())
print(f"sandbox created, instance_name={name}")
except Exception as e:
print(f"sandbox create failed, name={opt.name}, error={e}")
finally:
yr.finalize()
if __name__ == "__main__":
main()
