zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-07-29 04:08:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: add Tenki sandbox provider (#17305)

Browse Source 
### Summary

Adds a `tenki` sandbox provider that runs each agent code execution in a
disposable Tenki (https://tenki.cloud) microVM (create → exec → destroy,
no volumes or snapshots).
Registration mirrors PR #15039, configure `api_key` and `project_id` in
Admin > Sandbox Settings.

Both runtimes are covered:
- Python: `agent/sandbox/providers/tenki.py` (structured results +
artifact collection).
- Go: `internal/agent/sandbox/tenki.go`, mirroring the e2b provider and
wired into the provider manager.

`tenki-sandbox` is an optional dependency (it requires `protobuf>=6.31`,
which differs from RAGFlow's pinned gRPC stack), lazily imported with a
clear error when missing; installation is documented in the sandbox
quickstart.

Unit tests cover execution, structured results, artifacts
(symlink/size/extension limits), non-zero exit, timeout, error mapping,
and idempotent destroy.

---------

Co-authored-by: yiming.wang <yiming.wang@luxor.com>
This commit is contained in:
yiming wang
2026-07-28 19:24:39 +08:00
committed by GitHub
parent 73860170ae
commit dcadd8d837
14 changed files with 1636 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
admin/server/services.py
View File

@@ -484,6 +484,11 @@ class SandboxMgr:
"description": "E2B Cloud - Code Execution Sandboxes",
"tags": ["saas", "fast", "global"],
},
"tenki": {
"name": "Tenki",
"description": "Tenki - Disposable microVM code sandboxes",
"tags": ["saas", "cloud", "microvm", "isolated"],
},
}
@staticmethod
@@ -503,6 +508,7 @@ class SandboxMgr:
SSHProvider,
AliyunCodeInterpreterProvider,
E2BProvider,
TenkiProvider,
)
schemas = {
@@ -511,6 +517,7 @@ class SandboxMgr:
"ssh": SSHProvider.get_config_schema(),
"aliyun_codeinterpreter": AliyunCodeInterpreterProvider.get_config_schema(),
"e2b": E2BProvider.get_config_schema(),
"tenki": TenkiProvider.get_config_schema(),
}
if provider_id not in schemas:
@@ -576,6 +583,7 @@ class SandboxMgr:
SSHProvider,
AliyunCodeInterpreterProvider,
E2BProvider,
TenkiProvider,
)
try:
@@ -620,6 +628,7 @@ class SandboxMgr:
"ssh": SSHProvider,
"aliyun_codeinterpreter": AliyunCodeInterpreterProvider,
"e2b": E2BProvider,
"tenki": TenkiProvider,
}
provider = provider_classes[provider_type]()
is_valid, error_msg = provider.validate_config(config)
@@ -667,6 +676,7 @@ class SandboxMgr:
SSHProvider,
AliyunCodeInterpreterProvider,
E2BProvider,
TenkiProvider,
)
# Instantiate provider based on type
@@ -676,6 +686,7 @@ class SandboxMgr:
"ssh": SSHProvider,
"aliyun_codeinterpreter": AliyunCodeInterpreterProvider,
"e2b": E2BProvider,
"tenki": TenkiProvider,
}
if provider_type not in provider_classes:

4
agent/sandbox/client.py
View File

@@ -77,6 +77,7 @@ def _load_provider_from_settings() -> None:
E2BProvider,
LocalProvider,
SSHProvider,
TenkiProvider,
)
provider_classes = {
@@ -85,6 +86,7 @@ def _load_provider_from_settings() -> None:
"e2b": E2BProvider,
"local": LocalProvider,
"ssh": SSHProvider,
"tenki": TenkiProvider,
}
if provider_type not in provider_classes:
@@ -97,7 +99,7 @@ def _load_provider_from_settings() -> None:
# Initialize the provider
if not provider.initialize(config):
message = f"Failed to initialize sandbox provider: {provider_type}. Config keys: {list(config.keys())}"
if provider_type in {"local", "ssh"}:
if provider_type in {"local", "ssh", "tenki"}:
raise SandboxProviderConfigError(message)
logger.error(message)
return

3
agent/sandbox/providers/__init__.py
View File

@@ -26,6 +26,7 @@ This package contains:
- e2b.py: E2B provider implementation
- local.py: Local process provider implementation
- ssh.py: Remote SSH provider implementation
- tenki.py: Tenki disposable microVM provider implementation
"""
from .base import SandboxProvider, SandboxInstance, ExecutionResult, SandboxProviderConfigError
@@ -35,6 +36,7 @@ from .aliyun_codeinterpreter import AliyunCodeInterpreterProvider
from .e2b import E2BProvider
from .local import LocalProvider
from .ssh import SSHProvider
from .tenki import TenkiProvider
__all__ = [
"SandboxProvider",
@@ -47,4 +49,5 @@ __all__ = [
"E2BProvider",
"LocalProvider",
"SSHProvider",
"TenkiProvider",
]

535
agent/sandbox/providers/tenki.py Normal file
View File

@@ -0,0 +1,535 @@
#
# Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from __future__ import annotations
import base64
import json
import logging
import mimetypes
import os
import posixpath
import stat
import time
import uuid
from typing import Any, Dict, List, Optional
from agent.sandbox.result_protocol import (
build_javascript_wrapper,
build_python_wrapper,
extract_structured_result,
)
from .base import (
ExecutionResult,
SandboxInstance,
SandboxProvider,
SandboxProviderConfigError,
)
logger = logging.getLogger(__name__)
ALLOWED_ARTIFACT_EXTENSIONS = {
".csv",
".html",
".jpeg",
".jpg",
".json",
".pdf",
".png",
".svg",
}
# Base directory inside the Tenki sandbox. The default image runs as the
# unprivileged "tenki" user whose home is writable.
SANDBOX_HOME = "/home/tenki"
# Maximum directory depth walked when collecting artifacts, guarding against
# deeply nested or symlink-looped trees produced by untrusted code.
MAX_ARTIFACT_DEPTH = 16
class TenkiProvider(SandboxProvider):
"""Execute code in a disposable Tenki microVM.
Each create_instance() provisions a fresh sandbox, execute_code() runs a
single script in it, and destroy_instance() terminates it. Only the stable
create/exec/destroy path is used; no volumes or snapshots.
"""
def __init__(self):
self.api_key = ""
self.project_id = ""
self.base_url = ""
self.image = ""
self.allow_outbound = False
self.timeout = 30
self.max_lifetime = 3600
self.cpu_cores = 0
self.memory_mb = 0
self.disk_size_gb = 0
self.max_output_bytes = 1024 * 1024
self.max_artifacts = 20
self.max_artifact_bytes = 10 * 1024 * 1024
self._initialized = False
self._client = None
self._instances: dict[str, dict[str, Any]] = {}
def initialize(self, config: Dict[str, Any]) -> bool:
self.api_key = str(config.get("api_key", "") or "").strip()
self.project_id = str(config.get("project_id", "") or "").strip()
self.base_url = str(config.get("base_url", "") or "").strip()
self.image = str(config.get("image", "") or "").strip()
self.allow_outbound = bool(config.get("allow_outbound", False))
self.timeout = int(config.get("timeout", 30) or 30)
self.max_lifetime = int(config.get("max_lifetime", 3600) or 3600)
self.cpu_cores = int(config.get("cpu_cores", 0) or 0)
self.memory_mb = int(config.get("memory_mb", 0) or 0)
self.disk_size_gb = int(config.get("disk_size_gb", 0) or 0)
self.max_output_bytes = int(config.get("max_output_bytes", 1024 * 1024) or 1024 * 1024)
self.max_artifacts = int(config.get("max_artifacts", 20) or 20)
self.max_artifact_bytes = int(config.get("max_artifact_bytes", 10 * 1024 * 1024) or 10 * 1024 * 1024)
is_valid, error_message = self.validate_config(
{
"api_key": self.api_key,
"project_id": self.project_id,
"timeout": self.timeout,
"max_lifetime": self.max_lifetime,
"max_output_bytes": self.max_output_bytes,
"max_artifacts": self.max_artifacts,
"max_artifact_bytes": self.max_artifact_bytes,
}
)
if not is_valid:
raise SandboxProviderConfigError(error_message or "Invalid Tenki provider configuration.")
self._client = self._create_client()
self._assert_connectivity()
self._initialized = True
logger.info("Tenki provider initialized")
return True
def create_instance(self, template: str = "python") -> SandboxInstance:
if not self._initialized:
raise RuntimeError("Provider not initialized. Call initialize() first.")
language = self._normalize_language(template)
errors = self._tenki_errors()
create_kwargs: dict[str, Any] = {
"project_id": self.project_id,
"allow_outbound": self.allow_outbound,
"max_duration": self.max_lifetime,
"metadata": {"source": "ragflow"},
}
if self.image:
create_kwargs["image"] = self.image
if self.cpu_cores > 0:
create_kwargs["cpu_cores"] = self.cpu_cores
if self.memory_mb > 0:
create_kwargs["memory_mb"] = self.memory_mb
if self.disk_size_gb > 0:
create_kwargs["disk_size_gb"] = self.disk_size_gb
try:
sandbox = self._client.create(**create_kwargs)
except errors.QuotaExceededError as exc:
raise RuntimeError(f"Tenki quota exceeded: {exc}") from exc
except errors.RateLimitedError as exc:
raise RuntimeError(f"Tenki rate limited, please retry: {exc}") from exc
except errors.UnauthorizedError as exc:
raise SandboxProviderConfigError("Tenki authentication failed: check the API key.") from exc
except Exception as exc:
# Satisfy the base contract: any other SDK failure becomes RuntimeError.
raise RuntimeError(f"Failed to create Tenki sandbox: {exc}") from exc
remote_work_dir = posixpath.join(SANDBOX_HOME, f"ragflow-codeexec-{uuid.uuid4().hex}")
try:
result = sandbox.exec(
"mkdir",
"-p",
posixpath.join(remote_work_dir, "artifacts"),
timeout=min(self.timeout, 10),
)
if result.exit_code != 0:
raise RuntimeError(f"Failed to create sandbox workspace: {result.stderr_text or 'unknown error'}")
except Exception:
self._safe_terminate(sandbox)
raise
instance_id = str(uuid.uuid4())
self._instances[instance_id] = {
"sandbox": sandbox,
"remote_work_dir": remote_work_dir,
"language": language,
}
logger.info("Tenki sandbox created (instance=%s, session=%s)", instance_id, sandbox.id)
return SandboxInstance(
instance_id=instance_id,
provider="tenki",
status="running",
metadata={"language": language, "remote_work_dir": remote_work_dir, "sandbox_id": sandbox.id},
)
def execute_code(
self,
instance_id: str,
code: str,
language: str,
timeout: int = 10,
arguments: Optional[Dict[str, Any]] = None,
) -> ExecutionResult:
if not self._initialized:
raise RuntimeError("Provider not initialized. Call initialize() first.")
if instance_id not in self._instances:
raise RuntimeError(f"Unknown Tenki sandbox instance: {instance_id}")
normalized_lang = self._normalize_language(language)
instance = self._instances[instance_id]
sandbox = instance["sandbox"]
remote_work_dir: str = instance["remote_work_dir"]
errors = self._tenki_errors()
args_json = json.dumps(arguments or {}, ensure_ascii=False)
script_path, argv = self._prepare_script(sandbox, remote_work_dir, normalized_lang, code, args_json)
requested_timeout = self.timeout if timeout is None else int(timeout)
if requested_timeout <= 0:
raise RuntimeError(f"Execution timeout must be greater than 0 seconds, got {requested_timeout}.")
exec_timeout = min(requested_timeout, self.timeout)
start_time = time.time()
try:
result = sandbox.exec(*argv, cwd=remote_work_dir, timeout=exec_timeout)
except (errors.CommandTimeoutError, errors.PrimitiveTimeoutError) as exc:
logger.warning("Tenki execution timed out (instance=%s, timeout=%ss)", instance_id, exec_timeout)
raise TimeoutError(f"Execution timed out after {exec_timeout} seconds") from exc
except (errors.SessionTerminatedError, errors.SessionNotFoundError) as exc:
# The sandbox was reclaimed (e.g. max_lifetime) or lost mid-run;
# surface it as RuntimeError per the base contract, not an SDK type.
raise RuntimeError(f"Tenki sandbox is no longer available: {exc}") from exc
except Exception as exc:
raise RuntimeError(f"Tenki execution failed: {exc}") from exc
execution_time = time.time() - start_time
stdout = result.stdout_text
stderr = result.stderr_text
exit_code = int(result.exit_code)
self._validate_output_size(stdout, stderr)
stdout, structured_result = extract_structured_result(stdout)
return ExecutionResult(
stdout=stdout,
stderr=stderr,
exit_code=exit_code,
execution_time=execution_time,
metadata={
"instance_id": instance_id,
"language": normalized_lang,
"script_path": script_path,
"remote_work_dir": remote_work_dir,
"status": "ok" if exit_code == 0 else "error",
"timeout": exec_timeout,
"artifacts": self._collect_artifacts(sandbox, posixpath.join(remote_work_dir, "artifacts")),
"result_present": structured_result.get("present", False),
"result_value": structured_result.get("value"),
"result_type": structured_result.get("type"),
},
)
def destroy_instance(self, instance_id: str) -> bool:
if not self._initialized:
raise RuntimeError("Provider not initialized. Call initialize() first.")
if instance_id not in self._instances:
return True
instance = self._instances.pop(instance_id)
self._safe_terminate(instance["sandbox"])
logger.info("Tenki sandbox destroyed (instance=%s)", instance_id)
return True
def health_check(self) -> bool:
try:
self._assert_connectivity()
return True
except Exception:
return False
def get_supported_languages(self) -> List[str]:
return ["python", "javascript"]
@staticmethod
def get_config_schema() -> Dict[str, Dict]:
return {
"api_key": {
"type": "string",
"required": True,
"label": "API Key",
"secret": True,
"placeholder": "tk_...",
"description": "Tenki API key. Create one at https://app.tenki.cloud under API Keys.",
},
"project_id": {
"type": "string",
"required": True,
"label": "Project ID",
"placeholder": "Tenki project UUID",
"description": "Tenki project that sandboxes are created under.",
},
"base_url": {
"type": "string",
"required": False,
"label": "API Endpoint",
"placeholder": "https://api.tenki.cloud",
"description": "Override the Tenki API endpoint. Leave empty for the default.",
},
"image": {
"type": "string",
"required": False,
"label": "Sandbox Image",
"description": "Base image for sandboxes. Empty uses the Tenki default image (includes python3 and node).",
},
"allow_outbound": {
"type": "boolean",
"required": False,
"label": "Allow Outbound Network",
"default": False,
"description": "Security-relevant. Disabled by default so sandboxed code has no network access. Enable it to let code make outbound connections (e.g. to install packages).",
},
"timeout": {
"type": "integer",
"required": False,
"label": "Timeout (seconds)",
"default": 30,
"min": 1,
"max": 600,
"description": "Maximum execution time for a single run.",
},
"max_lifetime": {
"type": "integer",
"required": False,
"label": "Max Sandbox Lifetime (seconds)",
"default": 3600,
"min": 60,
"max": 86400,
"description": "Tenki reclaims a sandbox after this, guarding against leaks if the run is interrupted.",
},
"cpu_cores": {
"type": "integer",
"required": False,
"label": "vCPU Cores",
"default": 0,
"min": 0,
"max": 16,
"description": "vCPU per sandbox. 0 uses the Tenki default.",
},
"memory_mb": {
"type": "integer",
"required": False,
"label": "Memory (MB)",
"default": 0,
"min": 0,
"max": 65536,
"description": "Memory per sandbox in MB. 0 uses the Tenki default.",
},
"disk_size_gb": {
"type": "integer",
"required": False,
"label": "Disk Size (GB)",
"default": 0,
"min": 0,
"max": 100,
"description": "Disk size per sandbox in GB. 0 uses the Tenki default.",
},
"max_output_bytes": {
"type": "integer",
"required": False,
"label": "Max Output Bytes",
"default": 1048576,
"min": 1024,
"max": 10485760,
"description": "Maximum combined stdout and stderr size.",
},
"max_artifacts": {
"type": "integer",
"required": False,
"label": "Max Artifacts",
"default": 20,
"min": 0,
"max": 100,
"description": "Maximum number of files collected from the sandbox artifacts directory.",
},
"max_artifact_bytes": {
"type": "integer",
"required": False,
"label": "Max Artifact Bytes",
"default": 10485760,
"min": 1024,
"max": 104857600,
"description": "Maximum size of a single artifact file in bytes.",
},
}
def validate_config(self, config: Dict[str, Any]) -> tuple[bool, Optional[str]]:
api_key = str(config.get("api_key", "") or "").strip()
project_id = str(config.get("project_id", "") or "").strip()
if not api_key:
return False, "Tenki API key is required"
if not project_id:
return False, "Tenki project_id is required"
for key in ("timeout", "max_lifetime", "max_output_bytes", "max_artifacts", "max_artifact_bytes"):
try:
value = int(config.get(key, 0) or 0)
except (TypeError, ValueError):
return False, f"{key} must be an integer"
if key == "max_artifacts":
if value < 0:
return False, "max_artifacts must be greater than or equal to 0"
elif value <= 0:
return False, f"{key} must be greater than 0"
return True, None
# -- internal helpers -------------------------------------------------
def _create_client(self):
tenki = _get_tenki_module()
# Bound control-plane requests (who_am_i, create) so a slow or
# unreachable API cannot block initialize()/create_instance() forever.
kwargs: dict[str, Any] = {"auth_token": self.api_key, "timeout": float(self.timeout)}
if self.base_url:
kwargs["base_url"] = self.base_url
return tenki.Client(**kwargs)
def _assert_connectivity(self) -> None:
client = self._client or self._create_client()
errors = self._tenki_errors()
try:
client.who_am_i()
except errors.UnauthorizedError as exc:
raise SandboxProviderConfigError("Tenki authentication failed: check the API key.") from exc
except Exception as exc:
raise SandboxProviderConfigError(f"Failed to reach Tenki API: {exc}") from exc
def _prepare_script(self, sandbox, remote_work_dir: str, language: str, code: str, args_json: str) -> tuple[str, list[str]]:
if language == "python":
script_name = "main.py"
script_content = build_python_wrapper(code, args_json)
executable = "python3"
elif language in {"javascript", "nodejs"}:
script_name = "main.js"
script_content = build_javascript_wrapper(code, args_json)
executable = "node"
else:
raise RuntimeError(f"Unsupported language for Tenki provider: {language}")
script_path = posixpath.join(remote_work_dir, script_name)
sandbox.fs.write_text(script_path, script_content)
return script_path, [executable, script_path]
def _validate_output_size(self, stdout: str, stderr: str) -> None:
output_size = len((stdout or "").encode("utf-8")) + len((stderr or "").encode("utf-8"))
if output_size > self.max_output_bytes:
raise RuntimeError(f"Tenki execution output exceeded {self.max_output_bytes} bytes.")
def _collect_artifacts(self, sandbox, artifacts_dir: str) -> list[dict[str, Any]]:
artifacts: list[dict[str, Any]] = []
self._collect_artifacts_recursive(sandbox, artifacts_dir, "", artifacts, depth=0)
return artifacts
def _collect_artifacts_recursive(self, sandbox, current_dir: str, relative_dir: str, artifacts: list[dict[str, Any]], depth: int) -> None:
if depth > MAX_ARTIFACT_DEPTH:
raise RuntimeError(f"Artifact directory nesting exceeds {MAX_ARTIFACT_DEPTH} levels: {relative_dir}")
errors = self._tenki_errors()
try:
entries = sandbox.fs.list(current_dir)
except errors.FileNotFoundError:
return
except FileNotFoundError:
return
# fs.list returns each entry's basename in `.path`, not an absolute
# path, so join it onto the directory being listed.
for entry in sorted(entries, key=lambda item: item.path):
name = posixpath.basename(entry.path)
remote_path = posixpath.join(current_dir, name)
relative_path = posixpath.join(relative_dir, name) if relative_dir else name
# Reject symlinks. `is_symlink` is not populated by every SDK
# release, so also inspect the stat mode bits as the reliable check.
if getattr(entry, "is_symlink", False) or stat.S_ISLNK(entry.mode or 0):
raise RuntimeError(f"Artifact symlinks are not allowed: {relative_path}")
if entry.is_dir:
self._collect_artifacts_recursive(sandbox, remote_path, relative_path, artifacts, depth + 1)
continue
if len(artifacts) >= self.max_artifacts:
raise RuntimeError(f"Tenki execution produced more than {self.max_artifacts} artifacts.")
size = int(entry.size or 0)
if size > self.max_artifact_bytes:
raise RuntimeError(f"Artifact exceeds {self.max_artifact_bytes} bytes: {relative_path}")
ext = os.path.splitext(name)[1].lower()
if ext not in ALLOWED_ARTIFACT_EXTENSIONS:
raise RuntimeError(f"Unsupported artifact type: {relative_path}")
content = sandbox.fs.read_bytes(remote_path)
artifacts.append(
{
"name": relative_path,
"content_b64": base64.b64encode(content).decode("ascii"),
"mime_type": mimetypes.guess_type(name)[0] or "application/octet-stream",
"size": size,
}
)
def _safe_terminate(self, sandbox) -> None:
# Best-effort: max_lifetime reclaims the sandbox if this fails.
try:
sandbox.terminate()
except Exception as exc:
logger.warning("Failed to terminate Tenki sandbox, relying on max_lifetime: %s", exc)
def _tenki_errors(self):
tenki = _get_tenki_module()
return tenki.errors
@staticmethod
def _normalize_language(language: str) -> str:
lang_lower = (language or "python").lower()
if lang_lower in {"python", "python3"}:
return "python"
if lang_lower in {"javascript", "nodejs"}:
return "nodejs"
return lang_lower
def _get_tenki_module():
try:
import tenki_sandbox
except ImportError as exc:
# tenki-sandbox is an optional dependency: it requires protobuf>=6.31,
# which conflicts with RAGFlow's pinned gRPC stack, so it is not a core
# dependency. Install it into the runtime to enable this provider.
raise SandboxProviderConfigError("tenki-sandbox is required for the Tenki sandbox provider. Install it with `pip install tenki-sandbox` (or `uv pip install tenki-sandbox`).") from exc
return tenki_sandbox

28
docs/guides/agent/agent_quickstarts/sandbox_quickstart.md
View File

@@ -24,7 +24,7 @@ Configure sandbox providers from the admin page:
- `self_managed`: Uses the executor manager service.
- `local`: Runs code on the current machine.
- `ssh`: Runs code on a remote machine over SSH.
- `aliyun_codeinterpreter` and `e2b`: Cloud providers.
- `aliyun_codeinterpreter`, `e2b`, and `tenki`: Cloud providers.
<img width="2547" height="1475" alt="admin-sandbox-settings" src="https://github.com/user-attachments/assets/59ab948e-b98a-45a8-9db4-f1afbf6c3685" />
@@ -38,6 +38,32 @@ Admin > Sandbox Settings after the services are up.
- `local`: Runs code as local Python or Node.js subprocesses. Use this only in trusted development environments.
- `ssh`: Runs code on a remote machine over SSH.
- `aliyun_codeinterpreter` and `e2b`: Cloud-hosted providers that remain available in the admin provider list.
- `tenki`: Cloud-hosted provider that runs each execution in a disposable [Tenki](https://tenki.cloud) microVM. See [Tenki](#tenki) below.
### Tenki
`tenki` runs each code execution in a fresh Tenki microVM and destroys it afterwards. It is cloud-hosted, so it needs no local sandbox services, gVisor, or Docker base images — only outbound network access and an API key.
The `tenki-sandbox` SDK is an optional dependency (it requires `protobuf>=6.31`, which differs from RAGFlow's default gRPC stack), so it is not installed by default. Install it into the RAGFlow runtime before selecting this provider:
```bash
pip install tenki-sandbox
```
Configure it in **Admin > Sandbox Settings**:
- `api_key` (required): Tenki API key. Create one at [app.tenki.cloud](https://app.tenki.cloud) under **API Keys**.
- `project_id` (required): the Tenki project that sandboxes are created under.
- `base_url` (optional): override the Tenki API endpoint.
- `image` (optional): sandbox base image. Leave empty to use the Tenki default image, which includes `python3` and `node`.
- `allow_outbound` (optional, security-relevant): whether the sandbox may make outbound network connections. Defaults to `false` so sandboxed code has no network access; set it to `true` when code needs the network (for example, to install packages).
- `timeout`, `max_lifetime`, `cpu_cores`, `memory_mb`, `disk_size_gb`, and the output/artifact limits have sensible defaults and can be tuned in the same page.
Notes:
- Supported languages are Python and JavaScript.
- Files written to the `artifacts/` directory of the working directory are returned as run artifacts.
- The provider uses only Tenki's create/exec/destroy operations; it does not use volumes or snapshots.
## Prerequisites

4
go.mod
View File

@@ -5,6 +5,7 @@ go 1.26.4
require (
cloud.google.com/go/storage v1.63.0
github.com/DATA-DOG/go-sqlmock v1.5.2
github.com/LuxorLabs/tenki-sdk-go/sandbox v0.5.2
github.com/alibabacloud-go/agentrun-20250910/v5 v5.8.4
github.com/alibabacloud-go/darabonba-openapi/v2 v2.2.1
github.com/alicebob/miniredis/v2 v2.38.0
@@ -70,6 +71,7 @@ require (
)
require (
buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.11-20260415201107-50325440f8f2.1 // indirect
cel.dev/expr v0.25.1 // indirect
cloud.google.com/go v0.123.0 // indirect
cloud.google.com/go/auth v0.21.0 // indirect
@@ -77,7 +79,7 @@ require (
cloud.google.com/go/compute/metadata v0.9.0 // indirect
cloud.google.com/go/iam v1.11.0 // indirect
cloud.google.com/go/monitoring v1.29.0 // indirect
connectrpc.com/connect v1.19.2 // indirect
connectrpc.com/connect v1.20.0 // indirect
github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.32.0 // indirect
github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.57.0 // indirect
github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.57.0 // indirect

8
go.sum
View File

@@ -1,3 +1,5 @@
buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.11-20260415201107-50325440f8f2.1 h1:s6hzCXtND/ICdGPTMGk7C+/BFlr2Jg5GyH0NKf4XGXg=
buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.11-20260415201107-50325440f8f2.1/go.mod h1:tvtbpgaVXZX4g6Pn+AnzFycuRK3MOz5HJfEGeEllXYM=
cel.dev/expr v0.25.1 h1:1KrZg61W6TWSxuNZ37Xy49ps13NUovb66QLprthtwi4=
cel.dev/expr v0.25.1/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4=
cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
@@ -21,8 +23,8 @@ cloud.google.com/go/storage v1.63.0 h1:hvXF2xfg9I32bjujggxgkEZn/Ej6sJ9pieFgeueBL
cloud.google.com/go/storage v1.63.0/go.mod h1:tirWVptrFNo5GEX2DQ47JooF7yaweJdAJ1hYAVMvKzE=
cloud.google.com/go/trace v1.16.0 h1:GmQovzFc5F0CNfl0VLgL64aoTtu7xsM0YajW2GlG9+E=
cloud.google.com/go/trace v1.16.0/go.mod h1:r+bdAn16dKLSV1G2D5v3e58IlQlizfxWrUfjx7kM7X0=
connectrpc.com/connect v1.19.2 h1:McQ83FGdzL+t60peksi0gXC7MQ/iLKgLduAnThbM0mo=
connectrpc.com/connect v1.19.2/go.mod h1:tN20fjdGlewnSFeZxLKb0xwIZ6ozc3OQs2hTXy4du9w=
connectrpc.com/connect v1.20.0 h1:6TNDAB+WeNd2uolWNlYczB5E0KNNaVMNUEx8JEUsPmQ=
connectrpc.com/connect v1.20.0/go.mod h1:A2ygJrukXwWy32vkCAAHNVguZrqZ+jeZ9rGRnGR4dN4=
github.com/Azure/azure-sdk-for-go/sdk/azcore v0.19.0/go.mod h1:h6H6c8enJmmocHUbLiiGY6sx7f9i+X3m1CHdd5c6Rdw=
github.com/Azure/azure-sdk-for-go/sdk/azidentity v0.11.0/go.mod h1:HcM1YX14R7CJcghJGOYCgdezslRSVzqwLf/q+4Y2r/0=
github.com/Azure/azure-sdk-for-go/sdk/internal v0.7.0/go.mod h1:yqy467j36fJxcRV2TzfVZ1pCb5vxm4BtZPUdYWe/Xo8=
@@ -37,6 +39,8 @@ github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0
github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.57.0/go.mod h1:dzcEjy1WJ0Q4u9twNR3LcLhNoYMRCrMCMafpxa0TjPQ=
github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.57.0 h1:RoO5+d7uCmDqovLrHCr2/BuViUXvdcrNxyNM1pN9dDQ=
github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.57.0/go.mod h1:YqwkQPrWSC7+byyc1VlKbWLBF5JsW5IoL6xUkemYSXk=
github.com/LuxorLabs/tenki-sdk-go/sandbox v0.5.2 h1:HN2UwfNhHYCT7qmJJ5KKf+erRoH5RTfGb7N8dEbE9FQ=
github.com/LuxorLabs/tenki-sdk-go/sandbox v0.5.2/go.mod h1:rBypesed6hSrj+OgpJwyMVBF76PaQ1iOqc2RGUL4RmI=
github.com/RaveNoX/go-jsoncommentstrip v1.0.0/go.mod h1:78ihd09MekBnJnxpICcwzCMzGrKSKYe4AqU6PDYYpjk=
github.com/airbrake/gobrake v3.6.1+incompatible/go.mod h1:wM4gu3Cn0W0K7GUuVWnlXZU11AGBXMILnrdOU8Kn00o=
github.com/alibabacloud-go/agentrun-20250910/v5 v5.8.4 h1:hiAsm9pz6aICOPLI1FC54vga10xwd/XxfNj06ow5jVM=

8
internal/agent/sandbox/manager.go
View File

@@ -299,8 +299,10 @@ func buildProvider(t ProviderType) (SandboxProvider, error) {
return newLocalProviderFromEnv(), nil
case ProviderSSH:
return newSSHProviderFromEnv(), nil
case ProviderTenki:
return newTenkiProviderFromEnv(), nil
default:
return nil, fmt.Errorf("unknown provider type %q (known: self_managed, aliyun_codeinterpreter, e2b, local, ssh)", t)
return nil, fmt.Errorf("unknown provider type %q (known: self_managed, aliyun_codeinterpreter, e2b, local, ssh, tenki)", t)
}
}
@@ -320,7 +322,9 @@ func buildProviderFromConfig(t ProviderType, cfg map[string]any) (SandboxProvide
return newLocalProviderFromConfig(cfg), nil
case ProviderSSH:
return newSSHProviderFromConfig(cfg), nil
case ProviderTenki:
return newTenkiProviderFromConfig(cfg), nil
default:
return nil, fmt.Errorf("unknown provider type %q (known: self_managed, aliyun_codeinterpreter, e2b, local, ssh)", t)
return nil, fmt.Errorf("unknown provider type %q (known: self_managed, aliyun_codeinterpreter, e2b, local, ssh, tenki)", t)
}
}

4
internal/agent/sandbox/provider.go
View File

@@ -65,6 +65,10 @@ const (
// ProviderSSH runs the user's code on a remote host via
// SSH. Matches Python's SSHProvider.
ProviderSSH ProviderType = "ssh"
// ProviderTenki runs each execution in a disposable Tenki
// microVM (TenkiCloud Go SDK). Matches Python's TenkiProvider.
ProviderTenki ProviderType = "tenki"
)
// ErrE2BProviderNotImplemented is returned when an operator configures

311
internal/agent/sandbox/tenki.go Normal file
View File

@@ -0,0 +1,311 @@
//
// Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// tenki.go is the Go port of
// `agent/sandbox/providers/tenki.py::TenkiProvider`. It runs each
// CodeExec call in a disposable Tenki microVM: CreateInstance
// provisions a fresh sandbox, ExecuteCode runs `python3 -c <wrapped>`
// or `node -e <wrapped>` inside it, and DestroyInstance terminates it.
// The provider uses only Tenki's create/exec/destroy operations; it
// does not use volumes or snapshots.
//
// The code-wrapping protocol is shared with SelfManaged, Aliyun and
// e2b, so the `__RAGFLOW_RESULT__:` marker extraction works uniformly.
// Like the Go e2b provider, this port does not collect run artifacts
// (artifact collection lives only in the local/ssh providers).
//
// SDK: github.com/LuxorLabs/tenki-sdk-go/sandbox (MIT). The auth
// token and API URL are read from env by Initialize; image and
// tunables come from the admin-panel config map. Sandbox scope is
// inferred from the API key by the SDK.
package sandbox
import (
"context"
"errors"
"fmt"
"ragflow/internal/common"
"sync"
"time"
tenkisdk "github.com/LuxorLabs/tenki-sdk-go/sandbox"
)
// tenkiDefaultSandboxTimeout bounds a single CodeExec call. A fresh
// sandbox is provisioned per CreateInstance and destroyed on
// DestroyInstance, so the lifetime need only outlast one execution.
const tenkiDefaultSandboxTimeout = 5 * time.Minute
// TenkiProvider is the Go port of the Python TenkiProvider.
type TenkiProvider struct {
client *tenkisdk.Client
apiKey string
apiURL string
image string
allowOutbound bool
sandboxTimeout time.Duration
mu sync.Mutex
initialized bool
}
// newTenkiProviderFromEnv reads TENKI_* env vars and returns a
// provider ready for Initialize.
func newTenkiProviderFromEnv() *TenkiProvider {
return newTenkiProviderFromConfig(tenkiConfigFromEnv())
}
// tenkiConfigFromEnv builds a config map from the TENKI_* env vars,
// mirroring the admin-panel settings JSON shape.
func tenkiConfigFromEnv() map[string]any {
return map[string]any{
"API_KEY": common.GetEnv(common.EnvTenkiApiKey),
"API_URL": common.GetEnv(common.EnvTenkiAPIURL),
"IMAGE": common.GetEnv(common.EnvTenkiImage),
"TIMEOUT": common.GetEnv(common.EnvTenkiTimeout),
"ALLOW_OUTBOUND": common.GetEnv(common.EnvTenkiAllowOutbound),
}
}
// newTenkiProviderFromConfig builds the provider from a JSON config
// map (admin-panel settings or the env-backed map above).
func newTenkiProviderFromConfig(cfg map[string]any) *TenkiProvider {
p := &TenkiProvider{
apiKey: configString(cfg, "API_KEY"),
apiURL: configString(cfg, "API_URL"),
image: configString(cfg, "IMAGE"),
// Outbound network is opt-in: sandboxed code has no egress
// unless ALLOW_OUTBOUND is explicitly "true". This matches
// the self_managed sandbox, which treats network access as an
// unauthorized-access event by default.
allowOutbound: configString(cfg, "ALLOW_OUTBOUND") == "true",
}
timeoutSec := configInt(cfg, "TIMEOUT", int(tenkiDefaultSandboxTimeout.Seconds()))
if timeoutSec > 0 {
p.sandboxTimeout = time.Duration(timeoutSec) * time.Second
} else {
p.sandboxTimeout = tenkiDefaultSandboxTimeout
}
return p
}
// ProviderType returns ProviderTenki.
func (p *TenkiProvider) ProviderType() ProviderType { return ProviderTenki }
// Initialize builds the Tenki SDK client. The auth token comes from
// the admin config or TENKI_API_KEY; we check explicitly so the
// manager does not register a broken provider.
func (p *TenkiProvider) Initialize(ctx context.Context) error {
apiKey := p.apiKey
if apiKey == "" {
apiKey = common.GetEnv(common.EnvTenkiApiKey)
}
if apiKey == "" {
return errors.New("tenki: API key is required (set it in Admin > Sandbox Settings or TENKI_API_KEY)")
}
apiURL := p.apiURL
if apiURL == "" {
apiURL = common.GetEnv(common.EnvTenkiAPIURL)
}
opts := []tenkisdk.Option{tenkisdk.WithAuthToken(apiKey)}
if apiURL != "" {
opts = append(opts, tenkisdk.WithBaseURL(apiURL))
}
c, err := tenkisdk.New(opts...)
if err != nil {
return fmt.Errorf("tenki: build client: %w", err)
}
p.client = c
p.mu.Lock()
p.initialized = true
p.mu.Unlock()
return nil
}
// SupportedLanguages returns the languages the default Tenki image can
// run. The default image ships with python3 and node.
func (p *TenkiProvider) SupportedLanguages() []string {
return []string{"python", "javascript"}
}
// CreateInstance provisions a fresh Tenki sandbox. As with the e2b
// provider, the template argument is treated as the language hint; the
// actual base image comes from the configured image (empty = Tenki's
// default image). InstanceID is the Tenki session id.
func (p *TenkiProvider) CreateInstance(ctx context.Context, template string) (*SandboxInstance, error) {
if !p.isInitialized() {
return nil, fmt.Errorf("tenki: provider not initialized")
}
lang := normalizeLanguage(template)
if lang == "" {
return nil, fmt.Errorf("tenki: unsupported language %q", template)
}
opts := []tenkisdk.CreateOption{
tenkisdk.WithAllowOutbound(p.allowOutbound),
tenkisdk.WithMaxDuration(p.sandboxTimeout),
tenkisdk.WithWaitTimeout(p.sandboxTimeout),
}
if p.image != "" {
opts = append(opts, tenkisdk.WithImage(p.image))
}
sess, err := p.client.Create(ctx, opts...)
if err != nil {
return nil, fmt.Errorf("tenki: Create: %w", err)
}
return &SandboxInstance{
InstanceID: sess.ID,
Provider: ProviderTenki,
Status: "running",
Metadata: map[string]any{
"language": lang,
"image": p.image,
"sandbox_timeout": p.sandboxTimeout.String(),
},
}, nil
}
// ExecuteCode runs the user's code inside the sandbox via
// `python3 -c <wrapped>` (Python) or `node -e <wrapped>` (JS). The
// wrapped code carries the `__RAGFLOW_RESULT__:` marker so the
// structured main() return value comes back as before.
func (p *TenkiProvider) ExecuteCode(
ctx context.Context,
inst *SandboxInstance,
code, language string,
timeoutSec int,
args map[string]any,
) (*ExecutionResult, error) {
if !p.isInitialized() {
return nil, fmt.Errorf("tenki: provider not initialized")
}
if inst == nil || inst.InstanceID == "" {
return nil, fmt.Errorf("tenki: instance id required")
}
lang := normalizeLanguage(language)
if lang == "" {
return nil, fmt.Errorf("tenki: unsupported language %q", language)
}
timeout, err := validateTimeout(timeoutSec)
if err != nil {
return nil, err
}
if timeout == 0 {
timeout = int(p.sandboxTimeout.Seconds())
}
argsJSON, err := argsToJSON(args)
if err != nil {
return nil, err
}
var wrapped, cmd string
var runArgs []string
if lang == "python" {
cmd = "python3"
wrapped = BuildPythonWrapper(code, argsJSON)
runArgs = []string{"-c", wrapped}
} else {
cmd = "node"
wrapped = BuildJavaScriptWrapper(code, argsJSON)
runArgs = []string{"-e", wrapped}
}
// Re-obtain a handle to the sandbox created in CreateInstance and
// wait for its data plane to be ready before exec.
sess, err := p.client.Session(inst.InstanceID)
if err != nil {
return nil, fmt.Errorf("tenki: Session(%s): %w", inst.InstanceID, err)
}
if err := sess.WaitReady(ctx, p.sandboxTimeout); err != nil {
return nil, fmt.Errorf("tenki: WaitReady(%s): %w", inst.InstanceID, err)
}
start := time.Now()
res, err := sess.Exec(ctx, cmd,
tenkisdk.WithArgs(runArgs...),
tenkisdk.WithTimeout(time.Duration(timeout)*time.Second),
)
if err != nil {
// A non-zero exit is reported as a *Result, not an error;
// a nil result here means a transport, timeout or session
// error that we cannot map to stdout/stderr. Do not include
// runArgs — they embed the user's code and arguments.
return nil, fmt.Errorf("tenki: exec %s: %w", cmd, err)
}
return buildTenkiExecutionResult(res, lang, start), nil
}
// buildTenkiExecutionResult maps the Tenki Result to our
// sandbox.ExecutionResult. Stdout is scanned (untrimmed) for the
// `__RAGFLOW_RESULT__:` marker so the model gets the structured
// main() return value.
func buildTenkiExecutionResult(r *tenkisdk.Result, lang string, start time.Time) *ExecutionResult {
stdout, structured := ExtractStructuredResult(string(r.Stdout))
return &ExecutionResult{
Stdout: stdout,
Stderr: string(r.Stderr),
ExitCode: int(r.ExitCode),
ExecutionTime: time.Since(start).Seconds(),
Metadata: map[string]any{
"language": lang,
"structured_result": structured,
"tenki_status": string(r.Status),
},
}
}
// DestroyInstance terminates the sandbox. A session that is already
// gone (not found, terminated, or expired past its max duration) is
// treated as success so the call is idempotent.
func (p *TenkiProvider) DestroyInstance(ctx context.Context, inst *SandboxInstance) error {
if !p.isInitialized() {
return fmt.Errorf("tenki: provider not initialized")
}
if inst == nil || inst.InstanceID == "" {
return fmt.Errorf("tenki: instance id required")
}
sess, err := p.client.Session(inst.InstanceID)
if err != nil {
return fmt.Errorf("tenki: Session(%s): %w", inst.InstanceID, err)
}
if err := sess.Close(ctx); err != nil {
if errors.Is(err, tenkisdk.ErrSessionNotFound) ||
errors.Is(err, tenkisdk.ErrSessionTerminated) ||
errors.Is(err, tenkisdk.ErrSessionExpired) {
return nil
}
return fmt.Errorf("tenki: Close(%s): %w", inst.InstanceID, err)
}
return nil
}
// HealthCheck probes the Tenki control plane. There is no dedicated
// ping endpoint, so a successful WhoAmI is our probe.
func (p *TenkiProvider) HealthCheck(ctx context.Context) error {
if !p.isInitialized() {
return errors.New("tenki: provider not initialized")
}
if _, err := p.client.WhoAmI(ctx); err != nil {
return fmt.Errorf("tenki: WhoAmI: %w", err)
}
return nil
}
func (p *TenkiProvider) isInitialized() bool {
p.mu.Lock()
defer p.mu.Unlock()
return p.initialized
}

292
internal/agent/sandbox/tenki_test.go Normal file
View File

@@ -0,0 +1,292 @@
//
// Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
package sandbox
import (
"context"
"ragflow/internal/common"
"strings"
"testing"
"time"
tenkisdk "github.com/LuxorLabs/tenki-sdk-go/sandbox"
)
func TestTenkiProvider_ProviderTypeAndLanguages(t *testing.T) {
t.Parallel()
p := newTenkiProviderFromEnv()
if p.ProviderType() != ProviderTenki {
t.Errorf("ProviderType = %q, want %q", p.ProviderType(), ProviderTenki)
}
langs := p.SupportedLanguages()
want := map[string]bool{"python": true, "javascript": true}
got := map[string]bool{}
for _, l := range langs {
if !want[l] {
t.Errorf("unexpected language: %q", l)
}
got[l] = true
}
for l := range want {
if !got[l] {
t.Errorf("missing required language: %q", l)
}
}
}
// TestTenkiProvider_Defaults exercises the env-var defaults. We clear
// TENKI_* so the test is independent of the host environment. Cannot
// use t.Parallel with t.Setenv.
func TestTenkiProvider_Defaults(t *testing.T) {
for _, k := range []string{"TENKI_IMAGE", "TENKI_TIMEOUT", "TENKI_ALLOW_OUTBOUND"} {
t.Setenv(k, "")
}
p := newTenkiProviderFromEnv()
if p.image != "" {
t.Errorf("image = %q, want empty (SDK default image)", p.image)
}
if p.sandboxTimeout != tenkiDefaultSandboxTimeout {
t.Errorf("sandboxTimeout = %v, want %v", p.sandboxTimeout, tenkiDefaultSandboxTimeout)
}
if p.allowOutbound {
t.Errorf("allowOutbound = true, want false by default (network is opt-in)")
}
}
func TestTenkiProvider_EnvOverride(t *testing.T) {
t.Setenv("TENKI_IMAGE", "custom-image")
t.Setenv("TENKI_TIMEOUT", "120")
t.Setenv("TENKI_ALLOW_OUTBOUND", "true")
p := newTenkiProviderFromEnv()
if p.image != "custom-image" {
t.Errorf("image = %q, want %q", p.image, "custom-image")
}
if p.sandboxTimeout != 120*time.Second {
t.Errorf("sandboxTimeout = %v, want 120s", p.sandboxTimeout)
}
if !p.allowOutbound {
t.Errorf("allowOutbound = false, want true when TENKI_ALLOW_OUTBOUND=true")
}
}
// TestTenkiProvider_Initialize_MissingCreds verifies the provider
// refuses to initialize when TENKI_API_KEY is unset.
func TestTenkiProvider_Initialize_MissingCreds(t *testing.T) {
t.Setenv("TENKI_API_KEY", "")
p := newTenkiProviderFromEnv()
err := p.Initialize(context.Background())
if err == nil {
t.Fatalf("Initialize with no creds: got nil error, want one")
}
if !strings.Contains(err.Error(), "TENKI_API_KEY") {
t.Errorf("err = %v, want to mention TENKI_API_KEY", err)
}
}
// TestTenkiProvider_AllOps_BeforeInit verifies the "not initialized"
// guard is in place for every operational method.
func TestTenkiProvider_AllOps_BeforeInit(t *testing.T) {
t.Parallel()
p := newTenkiProviderFromEnv()
// Do NOT call Initialize.
inst := &SandboxInstance{InstanceID: "x", Provider: ProviderTenki}
if _, err := p.CreateInstance(context.Background(), "python"); err == nil {
t.Errorf("CreateInstance before init: got nil error, want one")
}
if _, err := p.ExecuteCode(context.Background(), inst, "x", "python", 5, nil); err == nil {
t.Errorf("ExecuteCode before init: got nil error, want one")
}
if err := p.DestroyInstance(context.Background(), inst); err == nil {
t.Errorf("DestroyInstance before init: got nil error, want one")
}
if err := p.HealthCheck(context.Background()); err == nil {
t.Errorf("HealthCheck before init: got nil error, want one")
}
}
func TestTenkiProvider_ExecuteCode_RejectsBadInputs(t *testing.T) {
t.Parallel()
p := newTenkiProviderFromEnv()
// Force "initialized" without building the SDK client so we can
// test the input-validation paths without a network call.
p.initialized = true
cases := []struct {
name string
fn func() error
want string
}{
{
name: "empty instance id",
fn: func() error {
_, err := p.ExecuteCode(context.Background(),
&SandboxInstance{InstanceID: ""}, "x", "python", 5, nil)
return err
},
want: "instance id",
},
{
name: "nil instance",
fn: func() error {
_, err := p.ExecuteCode(context.Background(),
nil, "x", "python", 5, nil)
return err
},
want: "instance id",
},
{
name: "unsupported language",
fn: func() error {
_, err := p.ExecuteCode(context.Background(),
&SandboxInstance{InstanceID: "x"}, "x", "ruby", 5, nil)
return err
},
want: "unsupported language",
},
{
name: "timeout too small",
fn: func() error {
_, err := p.ExecuteCode(context.Background(),
&SandboxInstance{InstanceID: "x"}, "x", "python", 0, nil)
return err
},
want: "timeout",
},
{
name: "timeout too large",
fn: func() error {
_, err := p.ExecuteCode(context.Background(),
&SandboxInstance{InstanceID: "x"}, "x", "python", 1000, nil)
return err
},
want: "timeout",
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
err := tc.fn()
if err == nil {
t.Fatalf("got nil error, want one containing %q", tc.want)
}
if !strings.Contains(err.Error(), tc.want) {
t.Errorf("err = %v, want to contain %q", err, tc.want)
}
})
}
}
func TestTenkiProvider_CreateInstance_UnsupportedLanguage(t *testing.T) {
t.Parallel()
p := newTenkiProviderFromEnv()
p.initialized = true
if _, err := p.CreateInstance(context.Background(), "ruby"); err == nil {
t.Errorf("CreateInstance(ruby): got nil error, want one")
}
}
func TestTenkiProvider_DestroyInstance_EmptyID(t *testing.T) {
t.Parallel()
p := newTenkiProviderFromEnv()
p.initialized = true
if err := p.DestroyInstance(context.Background(), &SandboxInstance{InstanceID: ""}); err == nil {
t.Errorf("DestroyInstance(empty id): got nil error, want one")
}
if err := p.DestroyInstance(context.Background(), nil); err == nil {
t.Errorf("DestroyInstance(nil): got nil error, want one")
}
}
// TestTenkiProvider_BuildTenkiExecutionResult unit-tests the
// result-mapping helper, including the marker-extraction path, using
// a hand-built SDK Result (no network).
func TestTenkiProvider_BuildTenkiExecutionResult(t *testing.T) {
t.Parallel()
r := &tenkisdk.Result{
Stdout: []byte("hello\n"),
Stderr: []byte(""),
ExitCode: 0,
Status: tenkisdk.CommandStatusSucceeded,
}
res := buildTenkiExecutionResult(r, "python", time.Now())
if res == nil {
t.Fatalf("buildTenkiExecutionResult returned nil")
}
if res.ExitCode != 0 {
t.Errorf("ExitCode = %d, want 0", res.ExitCode)
}
if res.Stdout != "hello\n" {
t.Errorf("Stdout = %q, want %q", res.Stdout, "hello\n")
}
if v, _ := res.Metadata["language"].(string); v != "python" {
t.Errorf("Metadata[language] = %v, want python", res.Metadata["language"])
}
}
// TestTenkiProvider_FullE2E_SkipWithoutKey is the integration test
// path. The body is skipped unless TENKI_API_KEY is set, but the test
// always runs (so missing-secrets shows up in CI logs). When enabled,
// it creates a real sandbox, runs Python, and destroys it.
func TestTenkiProvider_FullE2E_SkipWithoutKey(t *testing.T) {
apiKey := common.GetEnv(common.EnvTenkiApiKey)
if apiKey == "" {
t.Skip("TENKI_API_KEY not set — skipping full E2E test (real network call)")
}
p := newTenkiProviderFromEnv()
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
defer cancel()
if err := p.Initialize(ctx); err != nil {
t.Fatalf("Initialize: %v", err)
}
inst, err := p.CreateInstance(ctx, "python")
if err != nil {
t.Fatalf("CreateInstance: %v", err)
}
defer func() {
cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cleanupCancel()
if err := p.DestroyInstance(cleanupCtx, inst); err != nil {
t.Errorf("DestroyInstance: %v", err)
}
}()
result, err := p.ExecuteCode(ctx, inst, "def main(): return 1+1", "python", 30, nil)
if err != nil {
t.Fatalf("ExecuteCode: %v", err)
}
if result.ExitCode != 0 {
t.Errorf("ExitCode = %d, want 0; stderr=%q", result.ExitCode, result.Stderr)
}
}
// TestTenkiProvider_ProviderType_StaysDistinct ensures the tenki
// provider does not collide on the wire with the other providers.
func TestTenkiProvider_ProviderType_StaysDistinct(t *testing.T) {
t.Parallel()
seen := map[ProviderType]bool{}
for _, p := range []SandboxProvider{
newSelfManagedProviderFromEnv(),
newAliyunProviderFromEnv(),
newE2BProviderFromEnv(),
newTenkiProviderFromEnv(),
} {
if seen[p.ProviderType()] {
t.Errorf("provider type %q seen twice", p.ProviderType())
}
seen[p.ProviderType()] = true
}
}

5
internal/common/environments.go
View File

@@ -62,6 +62,11 @@ const (
EnvE2BApiKey = "E2B_API_KEY"
EnvE2BAccessToken = "E2B_ACCESS_TOKEN"
EnvE2BDomain = "E2B_DOMAIN"
EnvTenkiApiKey = "TENKI_API_KEY"
EnvTenkiAPIURL = "TENKI_API_URL"
EnvTenkiImage = "TENKI_IMAGE"
EnvTenkiTimeout = "TENKI_TIMEOUT"
EnvTenkiAllowOutbound = "TENKI_ALLOW_OUTBOUND"
EnvLocalPythonBin = "LOCAL_PYTHON_BIN"
EnvLocalNodeBin = "LOCAL_NODE_BIN"
EnvLocalWorkDir = "LOCAL_WORK_DIR"

428
test/unit_test/agent/sandbox/test_tenki_provider.py Normal file
View File

@@ -0,0 +1,428 @@
import base64
import posixpath
import stat
from types import SimpleNamespace
import pytest
from agent.sandbox.providers.base import SandboxProviderConfigError
from agent.sandbox.providers.tenki import TenkiProvider
from agent.sandbox.result_protocol import RESULT_MARKER_PREFIX
pytestmark = pytest.mark.p3
class _FakeErrors:
"""Stand-in for tenki_sandbox.errors so tests need not install the SDK."""
class QuotaExceededError(Exception):
pass
class RateLimitedError(Exception):
pass
class UnauthorizedError(Exception):
pass
class CommandTimeoutError(Exception):
pass
class PrimitiveTimeoutError(Exception):
pass
class SessionTerminatedError(Exception):
pass
class SessionNotFoundError(Exception):
pass
class FileNotFoundError(Exception):
pass
class _FakeFS:
def __init__(self):
self.files: dict[str, bytes] = {}
def write_text(self, path: str, content: str):
self.files[path] = content.encode("utf-8")
def read_bytes(self, path: str) -> bytes:
return self.files[path]
def list(self, path: str, *, include_hidden: bool = False):
# Mirror the real SDK: entry.path is the basename within `path`,
# not an absolute path.
prefix = path.rstrip("/") + "/"
entries = []
seen_dirs: set[str] = set()
for file_path, payload in self.files.items():
if not file_path.startswith(prefix):
continue
relative = file_path[len(prefix) :]
head, _, tail = relative.partition("/")
if tail: # nested -> surface the intermediate directory once
if head not in seen_dirs:
seen_dirs.add(head)
entries.append(SimpleNamespace(path=head, size=4096, mode=0o40755, is_dir=True, is_symlink=False, symlink_target=""))
continue
entries.append(
SimpleNamespace(
path=head,
size=len(payload),
mode=0o100644,
is_dir=False,
is_symlink=False,
symlink_target="",
)
)
return entries
class _FakeCommandResult:
def __init__(self, exit_code: int, stdout: str = "", stderr: str = ""):
self.exit_code = exit_code
self.stdout_text = stdout
self.stderr_text = stderr
self.stdout = stdout.encode("utf-8")
self.stderr = stderr.encode("utf-8")
class _FakeSandbox:
def __init__(self, run_handler=None):
self.id = "sbx-fake-1"
self.state = "RUNNING"
self.fs = _FakeFS()
self.terminated = False
self._run_handler = run_handler
self.exec_calls: list[tuple] = []
def exec(self, *argv, cwd=None, timeout=None, env=None, input=None, check=False, privileged=False):
self.exec_calls.append((argv, cwd, timeout))
if argv and argv[0] == "mkdir":
return _FakeCommandResult(0)
if self._run_handler is not None:
return self._run_handler(self, argv, cwd)
return _FakeCommandResult(0)
def terminate(self):
self.terminated = True
class _FakeClient:
def __init__(self, sandbox: _FakeSandbox):
self._sandbox = sandbox
self.create_kwargs = None
def who_am_i(self):
return SimpleNamespace(owner_type="WORKSPACE")
def create(self, **kwargs):
self.create_kwargs = kwargs
return self._sandbox
def _build_provider(sandbox: _FakeSandbox, monkeypatch) -> tuple[TenkiProvider, _FakeClient]:
provider = TenkiProvider()
provider.api_key = "tk_test"
provider.project_id = "proj-1"
provider.timeout = 30
provider.max_output_bytes = 1024 * 1024
provider.max_artifacts = 20
provider.max_artifact_bytes = 1024 * 1024
provider._initialized = True
client = _FakeClient(sandbox)
provider._client = client
monkeypatch.setattr(provider, "_tenki_errors", lambda: _FakeErrors)
return provider, client
def test_tenki_provider_executes_python_main_and_collects_artifacts(monkeypatch):
def run_handler(sandbox, argv, cwd):
# Simulate the script run: emit a structured result + write an artifact.
sandbox.fs.files[posixpath.join(cwd, "artifacts", "chart.png")] = b"PNGDATA"
payload = base64.b64encode(b'{"present":true,"value":{"message":"hello tenki"},"type":"json"}').decode("ascii")
return _FakeCommandResult(0, stdout=f"debug line\n{RESULT_MARKER_PREFIX}{payload}\n")
sandbox = _FakeSandbox(run_handler)
provider, _ = _build_provider(sandbox, monkeypatch)
instance = provider.create_instance("python")
result = provider.execute_code(
instance.instance_id,
'def main() -> dict:\n return {"message": "hello tenki"}\n',
"python",
timeout=5,
)
provider.destroy_instance(instance.instance_id)
assert result.exit_code == 0
assert result.stdout == "debug line\n"
assert result.metadata["result_present"] is True
assert result.metadata["result_value"] == {"message": "hello tenki"}
assert result.metadata["artifacts"] == [
{
"name": "chart.png",
"content_b64": base64.b64encode(b"PNGDATA").decode("ascii"),
"mime_type": "image/png",
"size": 7,
}
]
assert sandbox.terminated is True
def test_tenki_provider_reports_nonzero_exit(monkeypatch):
def run_handler(sandbox, argv, cwd):
return _FakeCommandResult(7, stdout="", stderr="boom\n")
sandbox = _FakeSandbox(run_handler)
provider, _ = _build_provider(sandbox, monkeypatch)
instance = provider.create_instance("python")
result = provider.execute_code(instance.instance_id, "def main():\n raise SystemExit(7)\n", "python", timeout=5)
assert result.exit_code == 7
assert result.stderr == "boom\n"
assert result.metadata["status"] == "error"
assert result.metadata["result_present"] is False
def test_tenki_provider_propagates_timeout(monkeypatch):
def run_handler(sandbox, argv, cwd):
raise _FakeErrors.CommandTimeoutError("timed out")
sandbox = _FakeSandbox(run_handler)
provider, _ = _build_provider(sandbox, monkeypatch)
instance = provider.create_instance("python")
with pytest.raises(TimeoutError, match="timed out after"):
provider.execute_code(instance.instance_id, "def main():\n return 1\n", "python", timeout=5)
def test_tenki_provider_create_maps_quota_error(monkeypatch):
sandbox = _FakeSandbox()
provider, client = _build_provider(sandbox, monkeypatch)
def _raise_quota(**kwargs):
raise _FakeErrors.QuotaExceededError("no credits")
client.create = _raise_quota
with pytest.raises(RuntimeError, match="quota exceeded"):
provider.create_instance("python")
def test_tenki_provider_destroy_is_idempotent(monkeypatch):
sandbox = _FakeSandbox()
provider, _ = _build_provider(sandbox, monkeypatch)
# Destroying an unknown instance returns True without error.
assert provider.destroy_instance("nonexistent") is True
def test_tenki_provider_create_passes_project_and_outbound(monkeypatch):
sandbox = _FakeSandbox()
provider, client = _build_provider(sandbox, monkeypatch)
provider.image = "my-image"
provider.cpu_cores = 4
provider.allow_outbound = True
provider.create_instance("python")
assert client.create_kwargs["project_id"] == "proj-1"
assert client.create_kwargs["allow_outbound"] is True
assert client.create_kwargs["image"] == "my-image"
assert client.create_kwargs["cpu_cores"] == 4
def test_tenki_provider_config_schema_and_validation():
schema = TenkiProvider.get_config_schema()
assert schema["api_key"]["required"] is True
assert schema["api_key"]["secret"] is True
assert schema["project_id"]["required"] is True
provider = TenkiProvider()
ok, _ = provider.validate_config({"api_key": "tk", "project_id": "p", "timeout": 30, "max_lifetime": 3600, "max_output_bytes": 1024, "max_artifacts": 5, "max_artifact_bytes": 1024})
assert ok is True
bad, msg = provider.validate_config({"api_key": "", "project_id": "p"})
assert bad is False
assert "API key" in msg
def test_tenki_provider_supported_languages():
assert TenkiProvider().get_supported_languages() == ["python", "javascript"]
def test_tenki_provider_initialize_maps_auth_error(monkeypatch):
provider = TenkiProvider()
monkeypatch.setattr(provider, "_tenki_errors", lambda: _FakeErrors)
class _UnauthorizedClient:
def who_am_i(self):
raise _FakeErrors.UnauthorizedError("bad token")
monkeypatch.setattr(provider, "_create_client", lambda: _UnauthorizedClient())
with pytest.raises(SandboxProviderConfigError, match="authentication failed"):
provider.initialize({"api_key": "tk_bad", "project_id": "proj-1"})
def test_tenki_provider_instances_are_independent(monkeypatch):
sandboxes: list[_FakeSandbox] = []
class _MultiClient(_FakeClient):
def create(self, **kwargs):
sb = _FakeSandbox()
sb.id = f"sbx-{len(sandboxes)}"
sandboxes.append(sb)
return sb
provider = TenkiProvider()
provider.api_key = "tk_test"
provider.project_id = "proj-1"
provider.timeout = 30
provider._initialized = True
provider._client = _MultiClient(None)
monkeypatch.setattr(provider, "_tenki_errors", lambda: _FakeErrors)
a = provider.create_instance("python")
b = provider.create_instance("python")
assert a.instance_id != b.instance_id
assert a.metadata["sandbox_id"] != b.metadata["sandbox_id"]
provider.destroy_instance(a.instance_id)
assert sandboxes[0].terminated is True
assert sandboxes[1].terminated is False # destroying one leaves the other running
provider.destroy_instance(b.instance_id)
assert sandboxes[1].terminated is True
def test_tenki_provider_executes_javascript(monkeypatch):
def run_handler(sandbox, argv, cwd):
payload = base64.b64encode(b'{"present":true,"value":42,"type":"json"}').decode("ascii")
return _FakeCommandResult(0, stdout=f"{RESULT_MARKER_PREFIX}{payload}\n")
sandbox = _FakeSandbox(run_handler)
provider, _ = _build_provider(sandbox, monkeypatch)
inst = provider.create_instance("javascript")
assert inst.metadata["language"] == "nodejs"
result = provider.execute_code(inst.instance_id, "function main(args){return 42;}", "javascript", timeout=5)
# The script is written as main.js and run with node.
assert any(k.endswith("main.js") for k in sandbox.fs.files)
run_argv = [c[0] for c in sandbox.exec_calls if c[0] and c[0][0] == "node"]
assert run_argv, "expected a node invocation"
assert result.metadata["result_value"] == 42
def test_tenki_provider_rejects_symlink_artifacts(monkeypatch):
provider, _ = _build_provider(_FakeSandbox(), monkeypatch)
class _SymlinkFS:
def list(self, path, *, include_hidden=False):
# is_symlink False but mode marks a symlink -> must still be rejected.
return [SimpleNamespace(path="evil.json", size=10, mode=stat.S_IFLNK | 0o777, is_dir=False, is_symlink=False, symlink_target="/etc/passwd")]
fake_sb = SimpleNamespace(fs=_SymlinkFS())
with pytest.raises(RuntimeError, match="symlinks are not allowed"):
provider._collect_artifacts(fake_sb, "/home/tenki/wk/artifacts")
def test_tenki_provider_enforces_output_size_limit(monkeypatch):
def run_handler(sandbox, argv, cwd):
return _FakeCommandResult(0, stdout="x" * 5000)
sandbox = _FakeSandbox(run_handler)
provider, _ = _build_provider(sandbox, monkeypatch)
provider.max_output_bytes = 100
inst = provider.create_instance("python")
with pytest.raises(RuntimeError, match="output exceeded"):
provider.execute_code(inst.instance_id, "def main():\n return 1\n", "python", timeout=5)
def test_tenki_provider_rejects_disallowed_artifact_extension(monkeypatch):
def run_handler(sandbox, argv, cwd):
sandbox.fs.files[posixpath.join(cwd, "artifacts", "malware.exe")] = b"MZ"
return _FakeCommandResult(0, stdout="")
sandbox = _FakeSandbox(run_handler)
provider, _ = _build_provider(sandbox, monkeypatch)
inst = provider.create_instance("python")
with pytest.raises(RuntimeError, match="Unsupported artifact type"):
provider.execute_code(inst.instance_id, "def main():\n return 1\n", "python", timeout=5)
def test_tenki_provider_allow_outbound_configurable(monkeypatch):
sandbox = _FakeSandbox()
provider, client = _build_provider(sandbox, monkeypatch)
provider.allow_outbound = False
provider.create_instance("python")
assert client.create_kwargs["allow_outbound"] is False
def test_tenki_provider_maps_session_terminated(monkeypatch):
def run_handler(sandbox, argv, cwd):
raise _FakeErrors.SessionTerminatedError("reclaimed")
sandbox = _FakeSandbox(run_handler)
provider, _ = _build_provider(sandbox, monkeypatch)
inst = provider.create_instance("python")
with pytest.raises(RuntimeError, match="no longer available"):
provider.execute_code(inst.instance_id, "def main():\n return 1\n", "python", timeout=5)
def test_tenki_provider_create_maps_rate_limit(monkeypatch):
sandbox = _FakeSandbox()
provider, client = _build_provider(sandbox, monkeypatch)
def _raise_rate(**kwargs):
raise _FakeErrors.RateLimitedError("slow down")
client.create = _raise_rate
with pytest.raises(RuntimeError, match="rate limited"):
provider.create_instance("python")
def test_tenki_provider_create_wraps_unexpected_sdk_error(monkeypatch):
sandbox = _FakeSandbox()
provider, client = _build_provider(sandbox, monkeypatch)
def _raise_other(**kwargs):
raise ValueError("weird sdk failure")
client.create = _raise_other
with pytest.raises(RuntimeError, match="Failed to create Tenki sandbox"):
provider.create_instance("python")
def test_tenki_provider_enforces_max_artifacts(monkeypatch):
def run_handler(sandbox, argv, cwd):
for i in range(3):
sandbox.fs.files[posixpath.join(cwd, "artifacts", f"f{i}.json")] = b"{}"
return _FakeCommandResult(0, stdout="")
sandbox = _FakeSandbox(run_handler)
provider, _ = _build_provider(sandbox, monkeypatch)
provider.max_artifacts = 2
inst = provider.create_instance("python")
with pytest.raises(RuntimeError, match="more than 2 artifacts"):
provider.execute_code(inst.instance_id, "def main():\n return 1\n", "python", timeout=5)
def test_tenki_provider_enforces_max_artifact_bytes(monkeypatch):
def run_handler(sandbox, argv, cwd):
sandbox.fs.files[posixpath.join(cwd, "artifacts", "big.json")] = b"x" * 5000
return _FakeCommandResult(0, stdout="")
sandbox = _FakeSandbox(run_handler)
provider, _ = _build_provider(sandbox, monkeypatch)
provider.max_artifact_bytes = 100
inst = provider.create_instance("python")
with pytest.raises(RuntimeError, match="exceeds 100 bytes"):
provider.execute_code(inst.instance_id, "def main():\n return 1\n", "python", timeout=5)

2
web/src/pages/admin/sandbox-settings.tsx
View File

@@ -6,6 +6,7 @@ import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
import {
LucideChevronDown,
LucideCloud,
LucideCloudLightning,
LucideLink,
LucideLoader2,
LucideMonitor,
@@ -62,6 +63,7 @@ const PROVIDER_ICONS: Record<string, React.ElementType> = {
ssh: LucideTerminal,
aliyun_codeinterpreter: LucideCloud,
e2b: LucideZap,
tenki: LucideCloudLightning,
};
function AdminSandboxSettings() {