scriptoriumadmin/ReMe
1
0
Fork 0
mirror of https://github.com/agentscope-ai/ReMe.git synced 2026-11-06 07:51:11 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 11

feat: add isolated Docker benchmark sandbox

Browse source
This commit is contained in:
酱牛肉 2026-08-03 01:56:38 +08:00
parent 3c4736e0aa
commit 5745d753d5
13 changed files with 1188 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
.dockerignore Normal file
View file

@ -0,0 +1,20 @@
.git
.reme
.env
.env.*
.venv
**/__pycache__
.pytest_cache
.mypy_cache
.ruff_cache
datasets
benchmark/datasets
memory_workspaces
logs
artifacts
build
dist
*.egg-info
*.pyc
*.pyo
*.log

7
pyproject.toml
View file

@ -61,9 +61,14 @@ dev = [
"pytest>=9.1.1",
"pytest-asyncio>=0.23",
]
sandbox = [
"agentscope==2.0.4.post1",
"aiodocker",
]
full = [
"reme-ai[core]",
"reme-ai[dev]",
"reme-ai[sandbox]",
]
[project.urls]
@ -75,7 +80,7 @@ Repository = "https://github.com/agentscope-ai/ReMe"
reme = "reme.reme:main"
[tool.setuptools]
packages = { find = { where = ["."], include = ["reme*"] } }
packages = { find = { where = ["."], include = ["reme*", "sandbox*"] } }
include-package-data = true
[tool.setuptools.package-data]

19
sandbox/.dockerignore.example Normal file
View file

@ -0,0 +1,19 @@
.git
.reme
.env
.env.*
.venv
__pycache__
.pytest_cache
.mypy_cache
.ruff_cache
datasets
benchmark/datasets
memory_workspaces
logs
artifacts
build
dist
*.egg-info
*.pyc
*.log

26
sandbox/Dockerfile.base Normal file
View file

@ -0,0 +1,26 @@
# Build from the repository root:
# docker build -f sandbox/Dockerfile.base \
# -t reme-sandbox-base:agentscope-2.0.4-post1 .
FROM python:3.11-slim
ENV PIP_DISABLE_PIP_VERSION_CHECK=1 \
PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1
RUN apt-get update \
&& apt-get install -y --no-install-recommends build-essential git tar \
&& rm -rf /var/lib/apt/lists/*
# Install the repository's exact dependency contract, then remove the dummy
# ReMe package. Source candidates are installed later with --no-deps, so no
# fixed ReMe implementation can leak into an experiment.
WORKDIR /tmp/reme-dependencies
COPY pyproject.toml README.md ./
RUN mkdir reme \
&& printf '__version__ = "0.0.0"\n' > reme/__init__.py \
&& python -m pip install '.[core]' \
&& python -m pip uninstall -y reme-ai \
&& test "$(python -c 'import agentscope; print(agentscope.__version__)')" = "2.0.4.post1" \
&& rm -rf /tmp/reme-dependencies
WORKDIR /workspace

12
sandbox/Dockerfile.candidate Normal file
View file

@ -0,0 +1,12 @@
# Build an immutable candidate from a source checkout:
# docker build -f sandbox/Dockerfile.candidate \
# --build-arg BASE_IMAGE=reme-sandbox-base:agentscope-2.0.4-post1 \
# -t reme-candidate:<candidate-id> .
ARG BASE_IMAGE=reme-sandbox-base:agentscope-2.0.4-post1
FROM ${BASE_IMAGE}
COPY . /opt/reme
RUN python -m pip install --no-deps /opt/reme \
&& python -c "import reme; assert reme.__file__.startswith('/opt/reme') or 'site-packages' in reme.__file__"
WORKDIR /workspace

131
sandbox/README.md Normal file
View file

@ -0,0 +1,131 @@
# ReMe Docker sandbox
This directory runs benchmark cases in AgentScope `DockerWorkspace` containers.
It never starts the ReMe HTTP service: the uploaded worker constructs a ReMe
`Application` and calls `run_job()` directly.
The harness uses a small `DockerWorkspace` subclass that retains AgentScope's
Docker lifecycle and `DockerBackend`, but skips its MCP gateway build/start.
The gateway is not used by direct ReMe jobs and would otherwise add a second
network-dependent image build and background process to every base image.
Install the host-side Docker dependency with:
```bash
pip install -e ".[sandbox]"
```
## Isolation model
- One candidate may create many case sandboxes.
- Every case is a separate Docker container.
- Every case writes to its own `/workspace/case/reme_workspace`.
- Source-candidate bytes are snapshotted once and reused unchanged across all
cases created by the same factory.
- Export happens before container shutdown.
## Build the dependency-only base image
Run from the repository root:
```bash
docker build \
-f sandbox/Dockerfile.base \
-t reme-sandbox-base:agentscope-2.0.4-post1 \
.
```
The image installs the dependencies from `pyproject.toml`, verifies AgentScope
`2.0.4.post1`, and then uninstalls the dummy ReMe package. Candidate source is
therefore the only importable ReMe implementation after case preparation.
## Mode 1: changing source candidate
```python
import asyncio
import os
from sandbox import DockerReMeSandboxFactory, SourceCandidate, SourceSnapshot
async def main():
# Do this once per candidate, not once per case.
snapshot = SourceSnapshot.from_directory(".")
factory = DockerReMeSandboxFactory(
SourceCandidate(snapshot),
env={
name: os.environ[name]
for name in ("LLM_API_KEY", "LLM_BASE_URL")
if name in os.environ
},
)
# Each ID gets a different container and runtime workspace, while both
# receive exactly the same snapshot bytes.
cases = await factory.create_cases(["session-001", "session-002"])
try:
for case in cases:
await case.ingest_session(
session_id=case.case_id,
messages=[{"role": "user", "content": "Remember this."}],
)
answer = await case.answer(query="What should you remember?")
await case.judge(
query="What should you remember?",
agent_answer=str(answer.answer),
golden_answer="Remember this.",
)
await case.export(f"artifacts/{case.case_id}.tar.gz")
finally:
await asyncio.gather(*(case.close() for case in cases))
asyncio.run(main())
```
Source archives exclude VCS state, `.reme`, virtual environments, caches,
credentials, benchmark datasets, existing memory workspaces, logs and build
output. Dataset inputs should be passed to cases explicitly rather than copied
into every candidate. Symbolic links are rejected so a snapshot cannot
accidentally capture files outside the candidate root.
## Mode 2: candidate preinstalled in an image
For a stable candidate, use the repository's `.dockerignore` (an equivalent
example is kept at `sandbox/.dockerignore.example`) and run:
```bash
docker build \
-f sandbox/Dockerfile.candidate \
--build-arg BASE_IMAGE=reme-sandbox-base:agentscope-2.0.4-post1 \
-t reme-candidate:my-candidate \
.
```
Use that candidate without uploading source per case:
```python
from sandbox import DockerReMeSandboxFactory, ImageCandidate
factory = DockerReMeSandboxFactory(
ImageCandidate("reme-candidate:my-candidate", candidate_id="git-or-content-hash"),
env={...},
)
case = await factory.create_case("session-001")
```
## Direct job and artifact contract
The high-level methods map to the existing `lme.yaml` jobs:
- `ingest_session()` → `auto_memory`, then optionally `index_update`
- `answer()` → `agentic_answer`
- `judge()` → `answer_judge`, with `yes=1`, `no=0`
- `run_job()` → any explicitly named ReMe job
`export()` downloads a gzip archive containing the complete ReMe runtime
workspace, ReMe logs, command audit log, per-job JSON results, canonical answer
and score files, and a manifest. Environment variable names are recorded, but
their secret values are never written to the manifest.
For an ephemeral Docker workspace, always call `export()` before `close()`.

18
sandbox/__init__.py Normal file
View file

@ -0,0 +1,18 @@
"""Docker sandbox infrastructure for isolated ReMe benchmark cases.
This package intentionally lives outside :mod:`reme`: benchmark isolation is
infrastructure, not part of ReMe's runtime contract.
"""
from .candidate import ImageCandidate, SourceCandidate, SourceSnapshot
from .docker_sandbox import DockerReMeSandbox, DockerReMeSandboxFactory
from .models import JobResult
__all__ = [
"DockerReMeSandbox",
"DockerReMeSandboxFactory",
"ImageCandidate",
"JobResult",
"SourceCandidate",
"SourceSnapshot",
]

133
sandbox/candidate.py Normal file
View file

@ -0,0 +1,133 @@
"""Candidate-code inputs shared by one or more isolated case sandboxes."""
from __future__ import annotations
from dataclasses import dataclass
import fnmatch
import gzip
import hashlib
import io
from pathlib import Path
import tarfile
DEFAULT_EXCLUDES = (
".git",
".reme",
".env",
".env.*",
".venv",
"__pycache__",
".pytest_cache",
".mypy_cache",
".ruff_cache",
"datasets",
"benchmark/datasets",
"memory_workspaces",
"logs",
"artifacts",
"build",
"dist",
"*.egg-info",
"*.pyc",
"*.pyo",
"*.log",
)
def _is_excluded(relative_path: Path, patterns: tuple[str, ...]) -> bool:
"""Return whether any path component or full POSIX path is excluded."""
posix = relative_path.as_posix()
return any(
fnmatch.fnmatch(posix, pattern) or any(fnmatch.fnmatch(part, pattern) for part in relative_path.parts)
for pattern in patterns
)
def _normalized_info(path: Path, arcname: str) -> tarfile.TarInfo:
"""Create deterministic tar metadata for a regular source file."""
stat = path.stat()
info = tarfile.TarInfo(arcname)
info.size = stat.st_size
info.mode = stat.st_mode & 0o777
info.mtime = 0
info.uid = 0
info.gid = 0
info.uname = ""
info.gname = ""
return info
@dataclass(frozen=True, slots=True)
class SourceSnapshot:
"""A deterministic source archive reusable across many case sandboxes."""
archive: bytes
sha256: str
file_count: int
@classmethod
def from_directory(
cls,
source_dir: str | Path,
*,
excludes: tuple[str, ...] = DEFAULT_EXCLUDES,
) -> "SourceSnapshot":
"""Snapshot a candidate directory without runtime or VCS state.
Symbolic links are rejected instead of followed. This prevents a
candidate archive from silently capturing files outside its root.
"""
root = Path(source_dir).resolve(strict=True)
if not root.is_dir():
raise ValueError(f"candidate source is not a directory: {root}")
files: list[tuple[Path, Path]] = []
for path in root.rglob("*"):
relative = path.relative_to(root)
if _is_excluded(relative, excludes):
continue
if path.is_symlink():
raise ValueError(f"candidate source contains a symbolic link: {relative}")
if path.is_file():
files.append((relative, path))
raw = io.BytesIO()
with gzip.GzipFile(fileobj=raw, mode="wb", mtime=0) as compressed:
with tarfile.open(fileobj=compressed, mode="w") as archive:
for relative, path in sorted(files, key=lambda item: item[0].as_posix()):
info = _normalized_info(path, relative.as_posix())
with path.open("rb") as source:
archive.addfile(info, source)
payload = raw.getvalue()
return cls(
archive=payload,
sha256=hashlib.sha256(payload).hexdigest(),
file_count=len(files),
)
@dataclass(frozen=True, slots=True)
class SourceCandidate:
"""A changing ReMe source snapshot installed after sandbox creation."""
snapshot: SourceSnapshot
base_image: str = "reme-sandbox-base:agentscope-2.0.4-post1"
@property
def candidate_id(self) -> str:
"""Stable identifier for the exact source snapshot."""
return self.snapshot.sha256
@dataclass(frozen=True, slots=True)
class ImageCandidate:
"""A ReMe candidate already installed in an immutable Docker image."""
image: str
candidate_id: str | None = None
@property
def resolved_candidate_id(self) -> str:
"""Return the caller-supplied digest/ID or the image reference."""
return self.candidate_id or self.image

40
sandbox/direct_docker_workspace.py Normal file
View file

@ -0,0 +1,40 @@
"""Minimal AgentScope 2.0.4.post1 DockerWorkspace for direct ReMe jobs.
The stock workspace builds and starts an MCP gateway. The benchmark harness
uses only DockerBackend's exec/read/write primitives, so building that gateway
would add an unrelated network-dependent image layer and process.
"""
from __future__ import annotations
from typing import Any
from agentscope.workspace import DockerWorkspace
class DirectDockerWorkspace(DockerWorkspace):
"""Use the supplied image directly and skip the unused MCP gateway."""
_image_tag: str
_backend: Any
is_alive: bool
async def _build_or_reuse_image(self) -> None:
"""Select the caller-provided image instead of deriving a gateway image."""
self._image_tag = self.base_image
async def initialize(self) -> None:
"""Start one Docker container and expose its filesystem/exec backend."""
if self.is_alive:
return
await self._provision_backend()
backend = self.get_backend()
result = await backend.exec_shell(["mkdir", "-p", self.workdir], cwd="/")
if not result.ok():
await self._teardown_backend()
self._backend = None
raise RuntimeError(
f"failed to create Docker workspace {self.workdir!r}: "
f"{result.stderr.decode('utf-8', errors='replace')}",
)
self.is_alive = True

434
sandbox/docker_sandbox.py Normal file
View file

@ -0,0 +1,434 @@
"""AgentScope DockerWorkspace orchestration for isolated ReMe cases."""
from __future__ import annotations
import asyncio
from dataclasses import asdict
from datetime import datetime, timezone
import json
from pathlib import Path
import re
from typing import Any, Callable, Protocol
import uuid
from .candidate import ImageCandidate, SourceCandidate
from .models import ActionRecord, JobResult
CASE_ROOT = "/workspace/case"
CANDIDATE_ROOT = "/workspace/candidate"
CANDIDATE_VENV = "/workspace/candidate-venv"
HARNESS_ROOT = "/workspace/harness"
SOURCE_ARCHIVE = "/workspace/candidate.tar.gz"
EXPORT_ARCHIVE = "/tmp/reme-case-export.tar.gz"
_CASE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$")
class Backend(Protocol):
"""Subset of AgentScope's backend API used by this module."""
async def exec_shell(
self,
command: list[str],
*,
cwd: str | None = None,
timeout: float | None = None,
) -> Any:
"""Execute an argv vector in the sandbox."""
raise NotImplementedError
async def read_file(self, path: str) -> bytes:
"""Read a sandbox file."""
raise NotImplementedError
async def write_file(self, path: str, data: bytes) -> None:
"""Write a sandbox file."""
raise NotImplementedError
class Workspace(Protocol):
"""Subset of AgentScope's DockerWorkspace used by this module."""
async def initialize(self) -> None:
"""Start the workspace."""
raise NotImplementedError
async def close(self) -> None:
"""Close the workspace."""
raise NotImplementedError
def get_backend(self) -> Backend:
"""Return the execution backend."""
raise NotImplementedError
WorkspaceBuilder = Callable[[str, str, dict[str, str]], Workspace]
class SandboxCommandError(RuntimeError):
"""A required command failed inside a case sandbox."""
def _utc_now() -> str:
"""Return an audit-friendly UTC timestamp."""
return datetime.now(timezone.utc).isoformat()
def _decode(value: bytes | str | None) -> str:
"""Decode backend process output without losing diagnostics."""
if isinstance(value, bytes):
return value.decode("utf-8", errors="replace")
return value or ""
def _default_workspace_builder(case_id: str, image: str, env: dict[str, str]) -> Workspace:
"""Create the exact AgentScope 2.0.4.post1 Docker backend lazily."""
import agentscope
from .direct_docker_workspace import DirectDockerWorkspace
if agentscope.__version__ != "2.0.4.post1":
raise RuntimeError(f"sandbox host requires agentscope 2.0.4.post1, got {agentscope.__version__}")
return DirectDockerWorkspace(
workspace_id=f"reme-benchmark-{case_id}",
base_image=image,
env=env,
)
class DockerReMeSandboxFactory:
"""Create independent case sandboxes that share one candidate definition.
A :class:`SourceCandidate` stores one immutable source archive in host
memory. The same bytes are uploaded to every case, while each case gets a
new Docker container and a fresh ReMe runtime workspace. An
:class:`ImageCandidate` skips source upload and uses an image in which ReMe
is already installed.
"""
def __init__(
self,
candidate: SourceCandidate | ImageCandidate,
*,
env: dict[str, str] | None = None,
config: str = "lme.yaml",
command_timeout: float = 1800.0,
workspace_builder: WorkspaceBuilder | None = None,
) -> None:
self.candidate = candidate
self.env = dict(env or {})
self.config = config
self.command_timeout = command_timeout
self._workspace_builder = workspace_builder or _default_workspace_builder
async def create_case(self, case_id: str) -> "DockerReMeSandbox":
"""Start and prepare a new container for one benchmark case."""
if not _CASE_ID_RE.fullmatch(case_id):
raise ValueError("case_id must contain only letters, numbers, '.', '_' or '-' and be <= 128 characters")
image = self.candidate.base_image if isinstance(self.candidate, SourceCandidate) else self.candidate.image
workspace = self._workspace_builder(case_id, image, self.env)
case = DockerReMeSandbox(
case_id=case_id,
candidate=self.candidate,
workspace=workspace,
config=self.config,
command_timeout=self.command_timeout,
environment_names=sorted(self.env),
)
try:
await case.initialize()
except Exception:
await workspace.close()
raise
return case
async def create_cases(self, case_ids: list[str], *, concurrency: int = 4) -> list["DockerReMeSandbox"]:
"""Create several independent cases while bounding Docker pressure."""
if concurrency < 1:
raise ValueError("concurrency must be at least 1")
if len(case_ids) != len(set(case_ids)):
raise ValueError("case_ids must be unique")
semaphore = asyncio.Semaphore(concurrency)
async def create(case_id: str) -> DockerReMeSandbox:
async with semaphore:
return await self.create_case(case_id)
results = await asyncio.gather(*(create(case_id) for case_id in case_ids), return_exceptions=True)
failures = [result for result in results if isinstance(result, BaseException)]
if failures:
await asyncio.gather(
*(result.close() for result in results if isinstance(result, DockerReMeSandbox)),
)
raise failures[0]
return [result for result in results if isinstance(result, DockerReMeSandbox)]
class DockerReMeSandbox:
"""One isolated Docker container and one fresh ReMe runtime workspace."""
def __init__(
self,
*,
case_id: str,
candidate: SourceCandidate | ImageCandidate,
workspace: Workspace,
config: str,
command_timeout: float,
environment_names: list[str],
) -> None:
self.case_id = case_id
self.candidate = candidate
self.workspace = workspace
self.config = config
self.command_timeout = command_timeout
self.environment_names = environment_names
self.backend: Backend | None = None
self.python = "python3"
self.actions: list[ActionRecord] = []
self._closed = False
@property
def runtime_workspace(self) -> str:
"""ReMe-owned runtime workspace inside this case container."""
return f"{CASE_ROOT}/reme_workspace"
@property
def logs_dir(self) -> str:
"""Directory containing ReMe and orchestration logs."""
return f"{CASE_ROOT}/logs"
@property
def results_dir(self) -> str:
"""Directory containing job, answer and score artifacts."""
return f"{CASE_ROOT}/results"
async def initialize(self) -> None:
"""Start Docker, upload the worker, and prepare the selected candidate."""
await self.workspace.initialize()
self.backend = self.workspace.get_backend()
await self._exec_checked(
"create_case_layout",
[
"mkdir",
"-p",
f"{CASE_ROOT}/inbox",
self.runtime_workspace,
self.logs_dir,
self.results_dir,
HARNESS_ROOT,
],
)
worker = Path(__file__).with_name("worker.py").read_bytes()
await self.backend.write_file(f"{HARNESS_ROOT}/worker.py", worker)
if isinstance(self.candidate, SourceCandidate):
await self._install_source_candidate()
else:
await self._validate_candidate(self.python)
async def _install_source_candidate(self) -> None:
assert self.backend is not None
await self.backend.write_file(SOURCE_ARCHIVE, self.candidate.snapshot.archive)
await self._exec_checked("create_candidate_dir", ["mkdir", "-p", CANDIDATE_ROOT])
await self._exec_checked(
"extract_candidate",
["tar", "-xzf", SOURCE_ARCHIVE, "-C", CANDIDATE_ROOT],
)
await self._exec_checked(
"create_candidate_venv",
["python3", "-m", "venv", "--system-site-packages", CANDIDATE_VENV],
)
self.python = f"{CANDIDATE_VENV}/bin/python"
await self._exec_checked(
"install_candidate",
[
self.python,
"-m",
"pip",
"install",
"--no-deps",
"--no-build-isolation",
"--editable",
CANDIDATE_ROOT,
],
)
await self._validate_candidate(self.python)
async def _validate_candidate(self, python: str) -> None:
script = (
"import agentscope, json, reme; "
"print(json.dumps({'reme': reme.__file__, 'agentscope': agentscope.__version__}))"
)
result = await self._exec_checked("validate_candidate", [python, "-c", script])
validation = json.loads(_decode(result.stdout))
if validation.get("agentscope") != "2.0.4.post1":
raise SandboxCommandError(
f"candidate environment requires agentscope 2.0.4.post1, got {validation.get('agentscope')!r}",
)
if isinstance(self.candidate, SourceCandidate) and CANDIDATE_ROOT not in str(validation.get("reme")):
raise SandboxCommandError(
f"candidate import did not resolve under {CANDIDATE_ROOT}: {validation.get('reme')}",
)
async def _exec(self, name: str, command: list[str], *, cwd: str | None = None) -> Any:
assert self.backend is not None, "sandbox is not initialized"
started = _utc_now()
result = await self.backend.exec_shell(command, cwd=cwd, timeout=self.command_timeout)
self.actions.append(
ActionRecord(
name=name,
exit_code=int(result.exit_code),
stdout=_decode(result.stdout),
stderr=_decode(result.stderr),
started_at=started,
finished_at=_utc_now(),
),
)
return result
async def _exec_checked(self, name: str, command: list[str], *, cwd: str | None = None) -> Any:
result = await self._exec(name, command, cwd=cwd)
if not result.ok():
raise SandboxCommandError(
f"sandbox action {name!r} failed with exit {result.exit_code}: {_decode(result.stderr)}",
)
return result
async def run_job(self, job: str, arguments: dict[str, Any] | None = None) -> JobResult:
"""Run one ReMe job directly in-process inside the container."""
assert self.backend is not None, "sandbox is not initialized"
request_id = uuid.uuid4().hex
request_path = f"{CASE_ROOT}/inbox/{request_id}.json"
response_path = f"{self.results_dir}/{request_id}.json"
request = {
"job": job,
"arguments": arguments or {},
"config": self.config,
"case_root": CASE_ROOT,
"workspace_dir": self.runtime_workspace,
}
await self.backend.write_file(request_path, json.dumps(request, ensure_ascii=False).encode("utf-8"))
await self._exec(
f"job:{job}",
[self.python, f"{HARNESS_ROOT}/worker.py", "--request", request_path, "--response", response_path],
cwd=CASE_ROOT,
)
try:
payload = json.loads((await self.backend.read_file(response_path)).decode("utf-8"))
except FileNotFoundError as exc:
raise SandboxCommandError(f"job {job!r} did not produce {response_path}") from exc
return JobResult.from_dict(payload)
async def ingest_session(
self,
*,
messages: list[dict[str, Any]],
session_id: str,
date: str = "",
memory_hint: str = "",
update_index: bool = True,
) -> JobResult:
"""Add one session, optionally making it searchable immediately."""
result = await self.run_job(
"auto_memory",
{"messages": messages, "session_id": session_id, "date": date, "memory_hint": memory_hint},
)
if result.success and update_index:
index_result = await self.run_job("index_update")
if not index_result.success:
return index_result
return result
async def answer(self, *, query: str, query_time: str = "") -> JobResult:
"""Generate an answer with the candidate's agentic_answer job."""
result = await self.run_job("agentic_answer", {"query": query, "query_time": query_time})
await self._write_named_result("answer.json", result)
return result
async def judge(
self,
*,
query: str,
agent_answer: str,
golden_answer: str,
question_type: str = "",
) -> JobResult:
"""Run LLM-as-Judge and persist a normalized numeric score."""
result = await self.run_job(
"answer_judge",
{
"query": query,
"agent_answer": agent_answer,
"golden_answer": golden_answer,
"question_type": question_type,
},
)
verdict = str(result.answer).strip().lower()
score = 1 if verdict == "yes" else 0 if verdict == "no" else None
assert self.backend is not None
await self.backend.write_file(
f"{self.results_dir}/score.json",
json.dumps(
{"success": result.success, "score": score, "verdict": verdict, "raw": asdict(result)},
ensure_ascii=False,
default=str,
).encode("utf-8"),
)
return result
async def _write_named_result(self, name: str, result: JobResult) -> None:
assert self.backend is not None
await self.backend.write_file(
f"{self.results_dir}/{name}",
json.dumps(asdict(result), ensure_ascii=False, default=str).encode("utf-8"),
)
async def export(self, destination: str | Path, *, include_candidate: bool = False) -> Path:
"""Download the complete runtime workspace, logs, results and manifest."""
assert self.backend is not None, "sandbox is not initialized"
if include_candidate and isinstance(self.candidate, ImageCandidate):
raise ValueError("an image candidate has no source directory to export")
candidate_id = (
self.candidate.candidate_id
if isinstance(self.candidate, SourceCandidate)
else self.candidate.resolved_candidate_id
)
manifest = {
"case_id": self.case_id,
"candidate_id": candidate_id,
"candidate_mode": "source" if isinstance(self.candidate, SourceCandidate) else "image",
"config": self.config,
"environment_names": self.environment_names,
"exported_at": _utc_now(),
}
await self.backend.write_file(
f"{CASE_ROOT}/manifest.json",
json.dumps(manifest, ensure_ascii=False, indent=2).encode("utf-8"),
)
await self.backend.write_file(
f"{self.logs_dir}/actions.jsonl",
b"".join(
(json.dumps(asdict(action), ensure_ascii=False) + "\n").encode("utf-8") for action in self.actions
),
)
command = ["tar", "-czf", EXPORT_ARCHIVE, "-C", CASE_ROOT, "."]
if include_candidate:
command.extend(["-C", "/workspace", "candidate"])
await self._exec_checked("export_case", command)
payload = await self.backend.read_file(EXPORT_ARCHIVE)
target = Path(destination).resolve()
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(payload)
return target
async def close(self) -> None:
"""Destroy the case container. Export before calling this method."""
if not self._closed:
self._closed = True
await self.workspace.close()
async def __aenter__(self) -> "DockerReMeSandbox":
return self
async def __aexit__(self, _exc_type: Any, _exc: Any, _tb: Any) -> None:
await self.close()

38
sandbox/models.py Normal file
View file

@ -0,0 +1,38 @@
"""Small serializable contracts shared by the host and sandbox worker."""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
@dataclass(frozen=True, slots=True)
class JobResult:
"""Structured result returned by a direct ReMe job invocation."""
success: bool
answer: Any = ""
metadata: dict[str, Any] = field(default_factory=dict)
error: str | None = None
@classmethod
def from_dict(cls, value: dict[str, Any]) -> "JobResult":
"""Build a result from the worker's JSON response."""
return cls(
success=bool(value.get("success")),
answer=value.get("answer", ""),
metadata=value.get("metadata") if isinstance(value.get("metadata"), dict) else {},
error=value.get("error") if isinstance(value.get("error"), str) else None,
)
@dataclass(frozen=True, slots=True)
class ActionRecord:
"""Audit record for one command executed inside a case sandbox."""
name: str
exit_code: int
stdout: str
stderr: str
started_at: str
finished_at: str

80
sandbox/worker.py Normal file
View file

@ -0,0 +1,80 @@
"""In-container worker that calls ReMe jobs directly, without HTTP.
The host uploads this file into every case sandbox. It deliberately imports
ReMe only after process startup so source candidates can control exactly which
ReMe package is loaded.
"""
from __future__ import annotations
import argparse
import asyncio
import json
import os
from pathlib import Path
import traceback
from typing import Any
async def _run(request: dict[str, Any]) -> dict[str, Any]:
"""Construct an Application, run one job, and close it deterministically."""
from reme.config import resolve_app_config
from reme.reme import ReMe
job = request["job"]
job_args = request.get("arguments") or {}
config = request.get("config") or "lme.yaml"
workspace_dir = request["workspace_dir"]
os.chdir(request["case_root"])
app_config = resolve_app_config(
config=config,
workspace_dir=workspace_dir,
log_to_console=False,
log_to_file=True,
)
app_config["environment"] = dict(os.environ)
app = ReMe(**app_config)
try:
await app.start()
response = await app.run_job(job, **job_args)
return {
"success": bool(response.success),
"answer": response.answer,
"metadata": response.metadata,
"error": None if response.success else str(response.answer),
}
finally:
await app.close()
def main() -> int:
"""Read one request file and always emit a structured response file."""
parser = argparse.ArgumentParser()
parser.add_argument("--request", required=True)
parser.add_argument("--response", required=True)
args = parser.parse_args()
request_path = Path(args.request)
response_path = Path(args.response)
response_path.parent.mkdir(parents=True, exist_ok=True)
try:
request = json.loads(request_path.read_text(encoding="utf-8"))
result = asyncio.run(_run(request))
exit_code = 0 if result["success"] else 1
except Exception as exc: # The host needs an artifact even for broken candidates.
result = {
"success": False,
"answer": "",
"metadata": {},
"error": f"{type(exc).__name__}: {exc}",
"traceback": traceback.format_exc(),
}
exit_code = 1
response_path.write_text(json.dumps(result, ensure_ascii=False, default=str), encoding="utf-8")
return exit_code
if __name__ == "__main__":
raise SystemExit(main())

231
tests/unit/test_sandbox.py Normal file
View file

@ -0,0 +1,231 @@
"""Tests for the standalone Docker benchmark sandbox infrastructure."""
from __future__ import annotations
import io
import json
import tarfile
from types import SimpleNamespace
import pytest
from sandbox import DockerReMeSandboxFactory, ImageCandidate, SourceCandidate, SourceSnapshot
from sandbox.direct_docker_workspace import DirectDockerWorkspace
from sandbox import worker
class FakeExecResult:
"""Minimal AgentScope ExecResult stand-in."""
def __init__(self, exit_code=0, stdout=b"", stderr=b""):
"""Store a fake process result."""
self.exit_code = exit_code
self.stdout = stdout
self.stderr = stderr
def ok(self):
"""Return whether the fake command succeeded."""
return self.exit_code == 0
class FakeBackend:
"""In-memory backend that emulates the worker and export boundaries."""
def __init__(self):
"""Initialize an empty in-memory filesystem and command log."""
self.files = {}
self.commands = []
async def write_file(self, path, data):
"""Write bytes into the fake filesystem."""
self.files[path] = data
async def read_file(self, path):
"""Read bytes from the fake filesystem."""
if path not in self.files:
raise FileNotFoundError(path)
return self.files[path]
async def exec_shell(self, command, *, cwd=None, timeout=None):
"""Record a command and emulate validation, worker, and tar output."""
self.commands.append((command, cwd, timeout))
if command[:2] == ["python3", "-c"] or (len(command) > 2 and command[1:3] == ["-c", command[2]]):
stdout = b'{"reme": "/workspace/candidate/reme/__init__.py", "agentscope": "2.0.4.post1"}\n'
return FakeExecResult(stdout=stdout)
if "/workspace/harness/worker.py" in command:
response_path = command[command.index("--response") + 1]
self.files[response_path] = json.dumps(
{"success": True, "answer": "yes", "metadata": {"job": "ok"}, "error": None},
).encode()
if command[:2] == ["tar", "-czf"]:
self.files[command[2]] = b"case-archive"
return FakeExecResult()
class FakeWorkspace:
"""A distinct workspace object for each factory call."""
def __init__(self, case_id, image, env):
"""Initialize one unique fake workspace."""
self.case_id = case_id
self.image = image
self.env = env
self.backend = FakeBackend()
self.initialized = False
self.closed = False
async def initialize(self):
"""Mark the fake workspace initialized."""
self.initialized = True
def get_backend(self):
"""Return the fake backend."""
return self.backend
async def close(self):
"""Mark the fake workspace closed."""
self.closed = True
@pytest.mark.asyncio
async def test_direct_workspace_uses_supplied_image_without_gateway_build(monkeypatch):
"""The direct backend starts Docker without deriving an MCP gateway image."""
workspace = DirectDockerWorkspace(base_image="candidate:exact")
backend = FakeBackend()
async def provision():
"""Bind a fake backend as Docker provisioning would."""
workspace._backend = backend # pylint: disable=protected-access
monkeypatch.setattr(workspace, "_provision_backend", provision)
await workspace._build_or_reuse_image() # pylint: disable=protected-access
await workspace.initialize()
assert workspace._image_tag == "candidate:exact" # pylint: disable=protected-access
assert workspace.is_alive is True
assert backend.commands == [(["mkdir", "-p", "/workspace"], "/", None)]
def test_source_snapshot_is_deterministic_and_excludes_runtime_state(tmp_path):
"""Equivalent source trees produce identical safe archives."""
(tmp_path / "reme").mkdir()
(tmp_path / "reme" / "main.py").write_text("print('candidate')\n")
(tmp_path / ".git").mkdir()
(tmp_path / ".git" / "config").write_text("secret-ish vcs state")
(tmp_path / ".env").write_text("TOKEN=secret")
first = SourceSnapshot.from_directory(tmp_path)
second = SourceSnapshot.from_directory(tmp_path)
assert first.archive == second.archive
assert first.sha256 == second.sha256
assert first.file_count == 1
with tarfile.open(fileobj=io.BytesIO(first.archive), mode="r:gz") as archive:
assert archive.getnames() == ["reme/main.py"]
def test_source_snapshot_rejects_symlinks(tmp_path):
"""A source snapshot never follows links outside the candidate root."""
target = tmp_path / "target.py"
target.write_text("pass\n")
(tmp_path / "link.py").symlink_to(target)
with pytest.raises(ValueError, match="symbolic link"):
SourceSnapshot.from_directory(tmp_path)
@pytest.mark.asyncio
async def test_one_source_candidate_is_reused_in_independent_case_workspaces(tmp_path):
"""The factory reuses code bytes but creates distinct containers."""
(tmp_path / "reme").mkdir()
(tmp_path / "reme" / "__init__.py").write_text("__version__ = 'candidate'\n")
(tmp_path / "pyproject.toml").write_text("[build-system]\nrequires=[]\nbuild-backend='x'\n")
candidate = SourceCandidate(SourceSnapshot.from_directory(tmp_path), base_image="base:test")
workspaces = []
def build(case_id, image, env):
"""Capture every separately constructed workspace."""
workspace = FakeWorkspace(case_id, image, env)
workspaces.append(workspace)
return workspace
factory = DockerReMeSandboxFactory(candidate, env={"LLM_API_KEY": "secret"}, workspace_builder=build)
first, second = await factory.create_cases(["case-1", "case-2"], concurrency=2)
assert first.workspace is not second.workspace
assert workspaces[0].image == workspaces[1].image == "base:test"
assert workspaces[0].backend.files["/workspace/candidate.tar.gz"] is candidate.snapshot.archive
assert workspaces[1].backend.files["/workspace/candidate.tar.gz"] is candidate.snapshot.archive
assert first.runtime_workspace == second.runtime_workspace == "/workspace/case/reme_workspace"
assert all(
not any("reme start" in part for part in command) for ws in workspaces for command, _, _ in ws.backend.commands
)
@pytest.mark.asyncio
async def test_image_candidate_skips_source_upload_and_runs_jobs(tmp_path):
"""An immutable image candidate needs no per-case source transfer."""
workspaces = []
def build(case_id, image, env):
"""Capture the workspace built for the image candidate."""
workspace = FakeWorkspace(case_id, image, env)
workspaces.append(workspace)
return workspace
factory = DockerReMeSandboxFactory(ImageCandidate("reme-candidate:test"), workspace_builder=build)
case = await factory.create_case("case-image")
result = await case.run_job("answer_judge", {"query": "q"})
exported = await case.export(tmp_path / "case.tar.gz")
assert result.success is True
assert result.answer == "yes"
assert "/workspace/candidate.tar.gz" not in workspaces[0].backend.files
assert exported.read_bytes() == b"case-archive"
assert json.loads(workspaces[0].backend.files["/workspace/case/manifest.json"])["candidate_mode"] == "image"
@pytest.mark.asyncio
async def test_worker_calls_application_directly_and_always_closes(monkeypatch, tmp_path):
"""The worker calls Application.run_job without any HTTP service."""
events = []
class FakeReMe:
"""Record the direct ReMe application lifecycle."""
def __init__(self, **config):
"""Record construction."""
events.append(("construct", config["workspace_dir"]))
async def start(self):
"""Record startup."""
events.append("start")
async def run_job(self, name, **arguments):
"""Return a successful direct job response."""
events.append(("run_job", name, arguments))
return SimpleNamespace(success=True, answer="answer", metadata={"direct": True})
async def close(self):
"""Record deterministic shutdown."""
events.append("close")
monkeypatch.setattr("reme.config.resolve_app_config", lambda **kwargs: kwargs)
monkeypatch.setattr("reme.reme.ReMe", FakeReMe)
result = await worker._run( # pylint: disable=protected-access
{
"job": "agentic_answer",
"arguments": {"query": "hello"},
"config": "lme.yaml",
"case_root": str(tmp_path),
"workspace_dir": str(tmp_path / "workspace"),
},
)
assert result == {"success": True, "answer": "answer", "metadata": {"direct": True}, "error": None}
assert events == [
("construct", str(tmp_path / "workspace")),
"start",
("run_job", "agentic_answer", {"query": "hello"}),
"close",
]