feat(examples): add sandbox session with env whitelist, timeout and output caps

Author: DNKYr

examples/skills_code_review_agent/review/sandbox.py

@@ -0,0 +1,136 @@
+# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
+#
+# Copyright (C) 2026 Tencent. All rights reserved.
+#
+# tRPC-Agent-Python is licensed under Apache-2.0.
+"""Sandbox execution of skill scripts via SDK workspace runtimes."""
+import os
+import shutil
+import sys
+from dataclasses import dataclass
+
+from trpc_agent_sdk.code_executors import (WorkspacePutFileInfo, WorkspaceRunProgramSpec,
+                                           WorkspaceStageOptions,
+                                           create_container_workspace_runtime,
+                                           create_local_workspace_runtime)
+
+SKILL_WS_DIR = "skills/code-review"
+DIFF_WS_PATH = "work/inputs/changes.diff"
+_PYTHON3_DIR = os.path.dirname(shutil.which("python3") or sys.executable)
+SANDBOX_ENV = {"PATH": f"{_PYTHON3_DIR}:/usr/local/bin:/usr/bin:/bin",
+               "HOME": "/tmp", "LANG": "C.UTF-8"}
+
+RUNTIME_LOCAL = "local"
+RUNTIME_CONTAINER = "container"
+RUNTIME_CUBE = "cube"
+
+
+async def create_runtime(kind: str):
+    """Create a workspace runtime. Container is the production default;
+    local is a development-only fallback; cube needs env credentials."""
+    if kind == RUNTIME_LOCAL:
+        return create_local_workspace_runtime()
+    if kind == RUNTIME_CONTAINER:
+        return create_container_workspace_runtime()
+    if kind == RUNTIME_CUBE:
+        from trpc_agent_sdk.code_executors.cube import (CubeClientConfig,
+                                                        CubeWorkspaceRuntimeConfig,
+                                                        create_cube_sandbox_client,
+                                                        create_cube_workspace_runtime)
+        if not os.getenv("CUBE_API_KEY") and not os.getenv("E2B_API_KEY"):
+            raise ValueError("cube runtime requires CUBE_API_KEY/E2B_API_KEY; "
+                             "see examples/skills_with_cube")
+        cfg = CubeClientConfig(
+            execute_timeout=float(os.getenv("CUBE_EXECUTE_TIMEOUT", "60")),
+            idle_timeout=int(os.getenv("CUBE_IDLE_TIMEOUT", "600")),
+            auto_recover=True)
+        client = await create_cube_sandbox_client(cfg)
+        return create_cube_workspace_runtime(sandbox_client=client,
+                                             execute_timeout=cfg.execute_timeout,
+                                             workspace_cfg=CubeWorkspaceRuntimeConfig())
+    raise ValueError(f"unknown runtime {kind!r}; expected local|container|cube")
+
+
+@dataclass
+class SandboxRunOutcome:
+    """Result of one sandboxed script execution."""
+
+    script: str
+    status: str  # ok | failed | timeout | error
+    exit_code: int
+    duration_ms: int
+    timed_out: bool
+    stdout: str
+    stderr: str
+    error_type: str = ""
+    truncated: bool = False
+
+
+class SandboxSession:
+    """One workspace: staged code-review skill + staged diff + script runs."""
+
+    def __init__(self, runtime, skill_root: str, timeout_sec: float = 60.0,
+                 max_output_bytes: int = 262144):
+        self._runtime = runtime
+        self._skill_root = skill_root
+        self._timeout = timeout_sec
+        self._max_output = max_output_bytes
+        self._ws = None
+        self._exec_id = ""
+
+    async def open(self, exec_id: str) -> None:
+        self._exec_id = exec_id
+        manager = self._runtime.manager(None)
+        self._ws = await manager.create_workspace(exec_id, None)
+        fs = self._runtime.fs(None)
+        await fs.stage_directory(self._ws, self._skill_root, SKILL_WS_DIR,
+                                 WorkspaceStageOptions(mode="copy"), None)
+
+    async def put_diff(self, diff_text: str) -> None:
+        fs = self._runtime.fs(None)
+        await fs.put_files(self._ws, [WorkspacePutFileInfo(
+            path=DIFF_WS_PATH, content=diff_text.encode("utf-8"), mode=0o644)], None)
+
+    def _truncate(self, text: str):
+        if len(text.encode("utf-8", errors="replace")) <= self._max_output:
+            return text, False
+        return text.encode("utf-8", errors="replace")[:self._max_output].decode(
+            "utf-8", errors="replace"), True
+
+    async def run_script(self, script_name: str,
+                         args: tuple = (DIFF_WS_PATH,)) -> SandboxRunOutcome:
+        """Run one skill script under `env -i` (environment whitelist)."""
+        env_args = [f"{k}={v}" for k, v in SANDBOX_ENV.items()]
+        argv = ["-i", *env_args, "python3",
+                f"{SKILL_WS_DIR}/scripts/{script_name}", *args]
+        spec = WorkspaceRunProgramSpec(cmd="env", args=argv, env={}, cwd=".",
+                                       timeout=self._timeout)
+        try:
+            ret = await self._runtime.runner(None).run_program(self._ws, spec, None)
+        except Exception as ex:  # noqa: BLE001 - sandbox failure must not crash the review
+            return SandboxRunOutcome(script=script_name, status="error", exit_code=-1,
+                                     duration_ms=0, timed_out=False, stdout="",
+                                     stderr=str(ex)[:1024], error_type=type(ex).__name__)
+        stdout, t1 = self._truncate(ret.stdout or "")
+        stderr, t2 = self._truncate(ret.stderr or "")
+        if ret.timed_out:
+            status = "timeout"
+        elif ret.exit_code == 0:
+            status = "ok"
+        else:
+            status = "failed"
+        return SandboxRunOutcome(script=script_name, status=status,
+                                 exit_code=ret.exit_code,
+                                 duration_ms=int(ret.duration * 1000),
+                                 timed_out=ret.timed_out, stdout=stdout,
+                                 stderr=stderr,
+                                 error_type="timeout" if ret.timed_out else "",
+                                 truncated=t1 or t2)
+
+    async def close(self) -> None:
+        if self._ws is None:
+            return
+        try:
+            await self._runtime.manager(None).cleanup(self._exec_id, None)
+        except Exception:  # noqa: BLE001 - best-effort cleanup
+            pass

examples/skills_code_review_agent/tests/test_sandbox.py

@@ -0,0 +1,90 @@
+# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
+#
+# Copyright (C) 2026 Tencent. All rights reserved.
+#
+# tRPC-Agent-Python is licensed under Apache-2.0.
+"""Tests for SandboxSession using the local workspace runtime."""
+import os
+from pathlib import Path
+
+from review.sandbox import DIFF_WS_PATH, SandboxSession, create_runtime  # noqa: F401
+
+EXAMPLE_ROOT = Path(__file__).resolve().parents[1]
+SKILL_ROOT = EXAMPLE_ROOT / "skills" / "code-review"
+FIXTURES = EXAMPLE_ROOT / "fixtures"
+
+
+def _make_tmp_skill(tmp_path, script_name, body):
+    skill = tmp_path / "code-review"
+    (skill / "scripts").mkdir(parents=True)
+    (skill / "SKILL.md").write_text("---\nname: code-review\ndescription: t\n---\nbody\n")
+    (skill / "scripts" / script_name).write_text(body)
+    return skill
+
+
+async def _open_session(skill_root, **kw):
+    runtime = await create_runtime("local")
+    session = SandboxSession(runtime, str(skill_root), **kw)
+    await session.open("cr_test_" + os.urandom(4).hex())
+    return session
+
+
+async def test_run_parse_diff_in_sandbox():
+    session = await _open_session(SKILL_ROOT)
+    try:
+        await session.put_diff((FIXTURES / "clean.diff").read_text())
+        outcome = await session.run_script("parse_diff.py")
+        assert outcome.status == "ok", outcome.stderr
+        assert '"files_changed": 2' in outcome.stdout
+    finally:
+        await session.close()
+
+
+async def test_env_whitelist_blocks_host_env(monkeypatch, tmp_path):
+    monkeypatch.setenv("CR_LEAKED_SECRET", "should-not-be-visible")
+    skill = _make_tmp_skill(tmp_path, "print_env.py",
+                            "import json, os\nprint(json.dumps(dict(os.environ)))\n")
+    session = await _open_session(skill)
+    try:
+        await session.put_diff("")
+        outcome = await session.run_script("print_env.py", args=())
+        assert outcome.status == "ok", outcome.stderr
+        assert "CR_LEAKED_SECRET" not in outcome.stdout
+    finally:
+        await session.close()
+
+
+async def test_timeout_is_enforced(tmp_path):
+    skill = _make_tmp_skill(tmp_path, "sleepy.py", "import time\ntime.sleep(30)\n")
+    session = await _open_session(skill, timeout_sec=2.0)
+    try:
+        await session.put_diff("")
+        outcome = await session.run_script("sleepy.py", args=())
+        assert outcome.timed_out is True
+        assert outcome.status == "timeout"
+    finally:
+        await session.close()
+
+
+async def test_output_size_cap(tmp_path):
+    skill = _make_tmp_skill(tmp_path, "noisy.py", "print('x' * 1000000)\n")
+    session = await _open_session(skill, max_output_bytes=1024)
+    try:
+        await session.put_diff("")
+        outcome = await session.run_script("noisy.py", args=())
+        assert len(outcome.stdout) <= 1024
+        assert outcome.truncated is True
+    finally:
+        await session.close()
+
+
+async def test_failing_script_reports_failed(tmp_path):
+    skill = _make_tmp_skill(tmp_path, "broken.py", "import sys\nsys.exit(3)\n")
+    session = await _open_session(skill)
+    try:
+        await session.put_diff("")
+        outcome = await session.run_script("broken.py", args=())
+        assert outcome.status == "failed"
+        assert outcome.exit_code == 3
+    finally:
+        await session.close()