diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 00000000..a2b4b2a3 --- /dev/null +++ b/.dockerignore @@ -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 diff --git a/pyproject.toml b/pyproject.toml index 2e1efc6f..7a8e3ac6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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] diff --git a/sandbox/.dockerignore.example b/sandbox/.dockerignore.example new file mode 100644 index 00000000..0c2317d6 --- /dev/null +++ b/sandbox/.dockerignore.example @@ -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 diff --git a/sandbox/Dockerfile.base b/sandbox/Dockerfile.base new file mode 100644 index 00000000..de7dd7a3 --- /dev/null +++ b/sandbox/Dockerfile.base @@ -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 diff --git a/sandbox/Dockerfile.candidate b/sandbox/Dockerfile.candidate new file mode 100644 index 00000000..776a47be --- /dev/null +++ b/sandbox/Dockerfile.candidate @@ -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: . +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 diff --git a/sandbox/README.md b/sandbox/README.md new file mode 100644 index 00000000..62507be8 --- /dev/null +++ b/sandbox/README.md @@ -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()`. diff --git a/sandbox/__init__.py b/sandbox/__init__.py new file mode 100644 index 00000000..d1d30d78 --- /dev/null +++ b/sandbox/__init__.py @@ -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", +] diff --git a/sandbox/candidate.py b/sandbox/candidate.py new file mode 100644 index 00000000..1581d1ff --- /dev/null +++ b/sandbox/candidate.py @@ -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 diff --git a/sandbox/direct_docker_workspace.py b/sandbox/direct_docker_workspace.py new file mode 100644 index 00000000..fdbd586e --- /dev/null +++ b/sandbox/direct_docker_workspace.py @@ -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 diff --git a/sandbox/docker_sandbox.py b/sandbox/docker_sandbox.py new file mode 100644 index 00000000..4a233c79 --- /dev/null +++ b/sandbox/docker_sandbox.py @@ -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() diff --git a/sandbox/models.py b/sandbox/models.py new file mode 100644 index 00000000..310e12e1 --- /dev/null +++ b/sandbox/models.py @@ -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 diff --git a/sandbox/worker.py b/sandbox/worker.py new file mode 100644 index 00000000..102c8b80 --- /dev/null +++ b/sandbox/worker.py @@ -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()) diff --git a/tests/unit/test_sandbox.py b/tests/unit/test_sandbox.py new file mode 100644 index 00000000..bec258dc --- /dev/null +++ b/tests/unit/test_sandbox.py @@ -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", + ]