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(sandbox): add container reuse, selective export, and serialized operations

Browse source 
Enhance the Docker benchmark sandbox so one container can process
multiple cases sequentially and export only analysis-relevant artifacts:

- Add reset_case() to discard the active case while keeping the
  installed candidate, venv, and worker for the next case.
- Add export(profile="analysis") (default) and export_full() to choose
  between a selective analysis archive and the complete disposable case
  tree; analysis export omits rebuildable indexes, caches, temp files,
  and source resources.
- Introduce runtime-layout.json written by the worker with validated,
  non-secret, case-relative paths so the host can build safe tar
  excludes.
- Serialize job, export, and reset operations with a per-case lock.
- Pin a sandbox-runtime extra in pyproject.toml that omits cookbook-only
  tushare, dingtalk-stream, pypdf, and polars, and install it in
  Dockerfile.base.
- Set TMPDIR to a case-local directory so subprocess temp files stay
  inside the disposable case root.
- Add unit tests for the new behaviors and a real-Docker LongMemEval
  integration smoke test.
This commit is contained in:
酱牛肉 2026-08-03 14:57:44 +08:00
parent 5745d753d5
commit 7f3ac6a762
7 changed files with 641 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
pyproject.toml
View file

@ -56,6 +56,17 @@ core = [
"polars>=1.43.0",
"tushare>=1.4.29",
]
sandbox-runtime = [
"agentscope==2.0.4.post1",
"claude-agent-sdk>=0.2.126",
"openai-codex>=0.144.4",
"pproxy>=2.7.9,<3",
"faiss-cpu>=1.13.2",
"jieba>=0.42.1",
"rjieba>=0.2.1",
"neo4j>=6.2.0",
"networkx>=3.4.2",
]
dev = [
"pre-commit>=4.6.1",
"pytest>=9.1.1",

10
sandbox/Dockerfile.base
View file

@ -11,14 +11,16 @@ 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.
# Install the benchmark sandbox dependency contract, then remove the dummy
# ReMe package. The sandbox-runtime extra deliberately omits cookbook-only
# tushare, dingtalk-stream, pypdf, and polars. 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 install '.[sandbox-runtime]' \
&& python -m pip uninstall -y reme-ai \
&& test "$(python -c 'import agentscope; print(agentscope.__version__)')" = "2.0.4.post1" \
&& rm -rf /tmp/reme-dependencies

53
sandbox/README.md
View file

@ -17,9 +17,11 @@ 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`.
- One candidate may create many container sandboxes.
- `create_cases()` gives every case a separate container; `reset_case()` can
instead reuse one container for sequential cases.
- The active case writes to `/workspace/case/reme_workspace`, which is deleted
before the container is assigned to the next case.
- Source-candidate bytes are snapshotted once and reused unchanged across all
cases created by the same factory.
- Export happens before container shutdown.
@ -123,9 +125,44 @@ The high-level methods map to the existing `lme.yaml` jobs:
- `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.
`export()` defaults to an analysis-focused gzip archive containing all ReMe log
files, the command audit log, per-job JSON results, canonical answer and score
files, persisted agent sessions, and the user-owned workspace files, including
files written outside the expected daily/digest paths. It also includes a
validated runtime layout and a manifest. It omits request copies, temporary
files, source resources, and rebuildable indexes/caches. Raw dialog sessions
under the configured `session_dir`/`dialog_dir` are included. Use
`export(profile="full")` or `export_full()` to download the complete disposable
case tree. 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()`.
For an ephemeral Docker workspace, always call `export()` or `export_full()`
before `close()`.
## Reuse one container for sequential cases
When container startup and source installation dominate the benchmark, one
container can process multiple cases sequentially. Finish and optionally
export the active case, then reset the disposable case directory before
uploading the next case's sessions:
```python
case = await factory.create_case("session-001")
try:
await case.ingest_session(session_id="session-001", messages=[...])
first = await case.answer(query="...")
await case.export("artifacts/session-001.tar.gz")
await case.reset_case("session-002")
await case.ingest_session(session_id="session-002", messages=[...])
second = await case.answer(query="...")
finally:
await case.close()
```
`reset_case()` deletes the ReMe runtime workspace, requests, logs, results,
manifest, case-scoped temporary files, and temporary export archive. It retains
the installed candidate, candidate virtual environment, and benchmark worker.
Jobs, exports, and resets share one lock, so they never overlap within a
container. Do not retain paths or state from the previous case after reset
returns.

239
sandbox/docker_sandbox.py
View file

@ -7,8 +7,9 @@ from dataclasses import asdict
from datetime import datetime, timezone
import json
from pathlib import Path
from pathlib import PurePosixPath
import re
from typing import Any, Callable, Protocol
from typing import Any, Callable, Literal, Protocol
import uuid
from .candidate import ImageCandidate, SourceCandidate
@ -20,6 +21,7 @@ CANDIDATE_VENV = "/workspace/candidate-venv"
HARNESS_ROOT = "/workspace/harness"
SOURCE_ARCHIVE = "/workspace/candidate.tar.gz"
EXPORT_ARCHIVE = "/tmp/reme-case-export.tar.gz"
RUNTIME_LAYOUT = f"{CASE_ROOT}/runtime-layout.json"
_CASE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$")
@ -68,6 +70,12 @@ class SandboxCommandError(RuntimeError):
"""A required command failed inside a case sandbox."""
def _validate_case_id(case_id: str) -> None:
"""Reject identifiers that are unsafe for Docker names and manifests."""
if not _CASE_ID_RE.fullmatch(case_id):
raise ValueError("case_id must contain only letters, numbers, '.', '_' or '-' and be <= 128 characters")
def _utc_now() -> str:
"""Return an audit-friendly UTC timestamp."""
return datetime.now(timezone.utc).isoformat()
@ -122,8 +130,7 @@ class DockerReMeSandboxFactory:
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")
_validate_case_id(case_id)
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(
@ -186,6 +193,7 @@ class DockerReMeSandbox:
self.python = "python3"
self.actions: list[ActionRecord] = []
self._closed = False
self._operation_lock = asyncio.Lock()
@property
def runtime_workspace(self) -> str:
@ -206,18 +214,7 @@ class DockerReMeSandbox:
"""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,
],
)
await self._create_case_layout()
worker = Path(__file__).with_name("worker.py").read_bytes()
await self.backend.write_file(f"{HARNESS_ROOT}/worker.py", worker)
@ -226,6 +223,22 @@ class DockerReMeSandbox:
else:
await self._validate_candidate(self.python)
async def _create_case_layout(self) -> None:
"""Create the disposable directories owned by the active case."""
await self._exec_checked(
"create_case_layout",
[
"mkdir",
"-p",
f"{CASE_ROOT}/inbox",
f"{CASE_ROOT}/tmp",
self.runtime_workspace,
self.logs_dir,
self.results_dir,
HARNESS_ROOT,
],
)
async def _install_source_candidate(self) -> None:
assert self.backend is not None
await self.backend.write_file(SOURCE_ARCHIVE, self.candidate.snapshot.archive)
@ -294,8 +307,8 @@ class DockerReMeSandbox:
)
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."""
async def _run_job(self, job: str, arguments: dict[str, Any] | None = None) -> JobResult:
"""Run one ReMe job while the caller holds the operation lock."""
assert self.backend is not None, "sandbox is not initialized"
request_id = uuid.uuid4().hex
request_path = f"{CASE_ROOT}/inbox/{request_id}.json"
@ -319,6 +332,11 @@ class DockerReMeSandbox:
raise SandboxCommandError(f"job {job!r} did not produce {response_path}") from exc
return JobResult.from_dict(payload)
async def run_job(self, job: str, arguments: dict[str, Any] | None = None) -> JobResult:
"""Run one ReMe job without overlapping another operation."""
async with self._operation_lock:
return await self._run_job(job, arguments)
async def ingest_session(
self,
*,
@ -329,21 +347,23 @@ class DockerReMeSandbox:
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 with self._operation_lock:
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 with self._operation_lock:
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,
@ -354,27 +374,28 @@ class DockerReMeSandbox:
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 with self._operation_lock:
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
@ -383,8 +404,77 @@ class DockerReMeSandbox:
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."""
async def export(
self,
destination: str | Path,
*,
profile: Literal["analysis", "full"] = "analysis",
include_candidate: bool = False,
) -> Path:
"""Download analysis artifacts, or the full case when requested."""
if profile not in ("analysis", "full"):
raise ValueError("export profile must be 'analysis' or 'full'")
async with self._operation_lock:
return await self._export(destination, profile=profile, include_candidate=include_candidate)
async def export_full(self, destination: str | Path, *, include_candidate: bool = False) -> Path:
"""Download every file under the disposable case root."""
return await self.export(destination, profile="full", include_candidate=include_candidate)
@staticmethod
def _validated_workspace_path(value: Any, workspace_root: PurePosixPath, *, allow_root: bool) -> str:
"""Validate one case-relative path received from candidate-controlled state."""
if not isinstance(value, str):
raise SandboxCommandError("sandbox runtime layout contains a non-string path")
path = PurePosixPath(value)
is_root = path == workspace_root
is_descendant = workspace_root in path.parents
is_unsafe = path.is_absolute() or ".." in path.parts
if is_unsafe or (is_root and not allow_root) or (not is_root and not is_descendant):
raise SandboxCommandError(f"sandbox runtime layout contains unsafe path: {value!r}")
return path.as_posix()
async def _analysis_archive_spec(self) -> tuple[str, list[str]]:
"""Read and validate the worker-resolved workspace export rules."""
assert self.backend is not None
try:
payload = json.loads((await self.backend.read_file(RUNTIME_LAYOUT)).decode("utf-8"))
except FileNotFoundError as exc:
raise SandboxCommandError("analysis export requires at least one completed job") from exc
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise SandboxCommandError(f"invalid sandbox runtime layout: {exc}") from exc
if not isinstance(payload, dict):
raise SandboxCommandError("sandbox runtime layout must be an object")
expected_root = PurePosixPath(self.runtime_workspace).relative_to(CASE_ROOT)
workspace_root_value = self._validated_workspace_path(
payload.get("workspace_root"),
expected_root,
allow_root=True,
)
if PurePosixPath(workspace_root_value) != expected_root:
raise SandboxCommandError(
f"sandbox runtime layout changed workspace root: {workspace_root_value!r}",
)
values = payload.get("analysis_excludes")
if not isinstance(values, list):
raise SandboxCommandError("sandbox runtime layout has invalid analysis_excludes")
excludes: list[str] = []
for value in values:
normalized = self._validated_workspace_path(value, expected_root, allow_root=False)
if normalized not in excludes:
excludes.append(normalized)
return workspace_root_value, excludes
async def _export(
self,
destination: str | Path,
*,
profile: Literal["analysis", "full"],
include_candidate: bool,
) -> Path:
"""Export the active case while the caller holds the operation lock."""
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")
@ -399,6 +489,7 @@ class DockerReMeSandbox:
"candidate_mode": "source" if isinstance(self.candidate, SourceCandidate) else "image",
"config": self.config,
"environment_names": self.environment_names,
"export_profile": profile,
"exported_at": _utc_now(),
}
await self.backend.write_file(
@ -411,7 +502,23 @@ class DockerReMeSandbox:
(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 profile == "analysis":
workspace_root, analysis_excludes = await self._analysis_archive_spec()
command = [
"tar",
"-czf",
EXPORT_ARCHIVE,
"-C",
CASE_ROOT,
*(f"--exclude={path}" for path in analysis_excludes),
"manifest.json",
"runtime-layout.json",
"logs",
"results",
workspace_root,
]
else:
command = ["tar", "-czf", EXPORT_ARCHIVE, "-C", CASE_ROOT, "."]
if include_candidate:
command.extend(["-C", "/workspace", "candidate"])
await self._exec_checked("export_case", command)
@ -421,11 +528,31 @@ class DockerReMeSandbox:
target.write_bytes(payload)
return target
async def reset_case(self, case_id: str) -> None:
"""Discard the active case and prepare this container for another.
Candidate code, its virtual environment, and the uploaded worker stay
intact. The runtime workspace, requests, logs, results, manifest, and
previous export archive are removed. Operations are serialized so a
reset cannot overlap a job or export.
"""
_validate_case_id(case_id)
async with self._operation_lock:
assert self.backend is not None, "sandbox is not initialized"
await self._exec_checked(
"clear_case",
["rm", "-rf", CASE_ROOT, EXPORT_ARCHIVE],
)
await self._create_case_layout()
self.case_id = case_id
self.actions = []
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 with self._operation_lock:
if not self._closed:
self._closed = True
await self.workspace.close()
async def __aenter__(self) -> "DockerReMeSandbox":
return self

66
sandbox/worker.py
View file

@ -15,17 +15,73 @@ from pathlib import Path
import traceback
from typing import Any
_WORKSPACE_PATH_FIELDS = (
"metadata_dir",
"session_dir",
"mem_session_dir",
"resource_dir",
"daily_dir",
"digest_dir",
"dialog_dir",
)
_ANALYSIS_EXCLUDE_FIELDS = ("metadata_dir", "resource_dir")
def _write_runtime_layout(case_root: Path, app_config: Any) -> None:
"""Persist validated, non-secret paths needed by selective export."""
resolved_case_root = case_root.resolve()
workspace_root = Path(app_config.workspace_dir).resolve()
try:
workspace_relative = workspace_root.relative_to(resolved_case_root)
except ValueError as exc:
raise ValueError(f"sandbox workspace must stay under {resolved_case_root}: {workspace_root}") from exc
resolved_paths: dict[str, str | None] = {}
for field in _WORKSPACE_PATH_FIELDS:
configured = str(getattr(app_config, field)).strip()
if not configured:
resolved_paths[field] = None
continue
path = (workspace_root / configured).resolve()
try:
relative = path.relative_to(resolved_case_root)
except ValueError as exc:
raise ValueError(f"sandbox config path {field} must stay under {workspace_root}: {path}") from exc
if path != workspace_root and workspace_root not in path.parents:
raise ValueError(f"sandbox config path {field} must stay under {workspace_root}: {path}")
resolved_paths[field] = relative.as_posix()
analysis_excludes = list(
dict.fromkeys(path for field in _ANALYSIS_EXCLUDE_FIELDS if (path := resolved_paths[field]) is not None),
)
layout = {
"workspace_root": workspace_relative.as_posix(),
"configured_paths": resolved_paths,
"analysis_excludes": analysis_excludes,
}
(resolved_case_root / "runtime-layout.json").write_text(
json.dumps(layout, ensure_ascii=False, indent=2),
encoding="utf-8",
)
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"])
case_root = Path(request["case_root"])
case_tmp = case_root / "tmp"
case_tmp.mkdir(parents=True, exist_ok=True)
os.environ["TMPDIR"] = str(case_tmp)
os.chdir(case_root)
# Import after setting TMPDIR so ReMe and its subprocesses cannot leave
# case-specific temporary files outside the disposable case root.
from reme.config import resolve_app_config
from reme.reme import ReMe
from reme.schema import ApplicationConfig
app_config = resolve_app_config(
config=config,
@ -34,6 +90,8 @@ async def _run(request: dict[str, Any]) -> dict[str, Any]:
log_to_file=True,
)
app_config["environment"] = dict(os.environ)
resolved_config = ApplicationConfig.model_validate(app_config)
_write_runtime_layout(case_root, resolved_config)
app = ReMe(**app_config)
try:
await app.start()

136
tests/integration/test_sandbox_longmemeval.py Normal file
View file

@ -0,0 +1,136 @@
"""Real Docker smoke test for the reusable LongMemEval sandbox.
This test uses the first LongMemEval item but intentionally ingests only its
first two chronological sessions. Answer quality is outside the contract: the
test verifies the end-to-end container workflow and the analysis archive.
Prerequisites:
* a running Docker daemon;
* ``reme-sandbox-base:agentscope-2.0.4-post1`` built from ``Dockerfile.base``;
* LLM and embedding credentials in the environment or repository ``.env``.
"""
from __future__ import annotations
import io
import json
import os
from pathlib import Path, PurePosixPath
import tarfile
from dotenv import load_dotenv
import pytest
from benchmark.longmemeval.run import format_messages_for_reme, parse_haystack_date, sessions_sorted_by_time, to_iso
from sandbox import DockerReMeSandboxFactory, SourceCandidate, SourceSnapshot
PROJECT_ROOT = Path(__file__).resolve().parents[2]
DATASET_PATH = PROJECT_ROOT / "benchmark/datasets/longmemeval/longmemeval_s_reme_cleaned.json"
SANDBOX_ENV_NAMES = (
"LLM_API_KEY",
"LLM_BASE_URL",
"LLM_BACKEND",
"LLM_MODEL_NAME",
"BENCH_MODEL_NAME",
"EMBEDDING_API_KEY",
"EMBEDDING_BASE_URL",
"EMBEDDING_BACKEND",
"EMBEDDING_MODEL_NAME",
)
def _archive_members(payload: bytes) -> tuple[set[str], dict[str, bytes]]:
"""Return normalized member names and regular-file payloads."""
names: set[str] = set()
files: dict[str, bytes] = {}
with tarfile.open(fileobj=io.BytesIO(payload), mode="r:gz") as archive:
for member in archive.getmembers():
normalized = PurePosixPath(member.name).as_posix().removeprefix("./")
names.add(normalized.rstrip("/"))
if member.isfile():
extracted = archive.extractfile(member)
assert extracted is not None
files[normalized] = extracted.read()
return names, files
@pytest.mark.asyncio
async def test_first_longmemeval_case_analysis_export(tmp_path):
"""Ingest two sessions, answer once, and verify selective download paths."""
load_dotenv(PROJECT_ROOT / ".env", override=False)
missing = [name for name in ("LLM_API_KEY", "EMBEDDING_API_KEY") if not os.environ.get(name)]
if missing:
pytest.skip(f"missing sandbox integration credentials: {', '.join(missing)}")
dataset = json.loads(DATASET_PATH.read_text(encoding="utf-8"))
item = dataset[0]
sessions = sessions_sorted_by_time(item)[:2]
assert len(sessions) == 2
env = {name: os.environ[name] for name in SANDBOX_ENV_NAMES if os.environ.get(name)}
candidate = SourceCandidate(SourceSnapshot.from_directory(PROJECT_ROOT))
factory = DockerReMeSandboxFactory(candidate, env=env, command_timeout=1800.0)
case = await factory.create_case(f"longmemeval-{item['question_id']}-smoke")
archive_path = tmp_path / "analysis.tar.gz"
try:
assert case.backend is not None
dependency_check = await case.backend.exec_shell(
[
case.python,
"-c",
"import importlib.util, json; "
"print(json.dumps({name: importlib.util.find_spec(name) is not None "
"for name in ('tushare', 'dingtalk_stream', 'pypdf', 'polars')}))",
],
)
assert dependency_check.ok(), dependency_check.stderr
assert json.loads(dependency_check.stdout) == {
"tushare": False,
"dingtalk_stream": False,
"pypdf": False,
"polars": False,
}
for _, session_dt, session_id, messages in sessions:
result = await case.ingest_session(
session_id=session_id,
messages=format_messages_for_reme(messages, session_dt),
date=session_dt.strftime("%Y-%m-%d"),
)
assert result.success, result.error or result.answer
answer = await case.answer(
query=item["question"],
query_time=to_iso(parse_haystack_date(item["question_date"])),
)
assert answer.success, answer.error or answer.answer
await case.export(archive_path)
finally:
await case.close()
names, files = _archive_members(archive_path.read_bytes())
assert {"manifest.json", "runtime-layout.json", "logs", "results", "reme_workspace"} <= names
assert "results/answer.json" in names
assert "logs/actions.jsonl" in names
assert any(name.startswith("logs/") and name.endswith(".log") for name in names)
manifest = json.loads(files["manifest.json"])
layout = json.loads(files["runtime-layout.json"])
assert manifest["case_id"] == f"longmemeval-{item['question_id']}-smoke"
assert manifest["export_profile"] == "analysis"
assert layout["workspace_root"] == "reme_workspace"
assert layout["configured_paths"]["daily_dir"] == "reme_workspace/daily"
assert layout["configured_paths"]["mem_session_dir"] == "reme_workspace/mem_session"
assert any(name.startswith("reme_workspace/daily/") and name.endswith(".md") for name in names)
assert any(name.startswith("reme_workspace/mem_session/") for name in names)
assert any(name.startswith("reme_workspace/session/dialog/") and name.endswith(".jsonl") for name in names)
excluded_prefixes = (
"inbox",
"tmp",
"reme_workspace/metadata",
"reme_workspace/resource",
)
assert not any(name == prefix or name.startswith(f"{prefix}/") for prefix in excluded_prefixes for name in names)
assert not any(".reme" in PurePosixPath(name).parts for name in names)

200
tests/unit/test_sandbox.py
View file

@ -2,6 +2,7 @@
from __future__ import annotations
import asyncio
import io
import json
import tarfile
@ -9,6 +10,7 @@ from types import SimpleNamespace
import pytest
from reme.schema import ApplicationConfig
from sandbox import DockerReMeSandboxFactory, ImageCandidate, SourceCandidate, SourceSnapshot
from sandbox.direct_docker_workspace import DirectDockerWorkspace
from sandbox import worker
@ -49,11 +51,31 @@ class FakeBackend:
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] == ["rm", "-rf"]:
roots = tuple(f"{path}/" for path in command[2:])
for path in list(self.files):
if path in command[2:] or path.startswith(roots):
del self.files[path]
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["/workspace/case/runtime-layout.json"] = json.dumps(
{
"workspace_root": "reme_workspace",
"configured_paths": {
"metadata_dir": "reme_workspace/metadata",
"session_dir": "reme_workspace/session",
"resource_dir": "reme_workspace/resource",
"dialog_dir": "reme_workspace/session/dialog",
"daily_dir": "reme_workspace/daily",
"digest_dir": "reme_workspace/digest",
"mem_session_dir": "reme_workspace/mem_session",
},
"analysis_excludes": ["reme_workspace/metadata", "reme_workspace/resource"],
},
).encode()
self.files[response_path] = json.dumps(
{"success": True, "answer": "yes", "metadata": {"job": "ok"}, "error": None},
).encode()
@ -65,12 +87,12 @@ class FakeBackend:
class FakeWorkspace:
"""A distinct workspace object for each factory call."""
def __init__(self, case_id, image, env):
def __init__(self, case_id, image, env, backend=None):
"""Initialize one unique fake workspace."""
self.case_id = case_id
self.image = image
self.env = env
self.backend = FakeBackend()
self.backend = backend or FakeBackend()
self.initialized = False
self.closed = False
@ -183,12 +205,179 @@ async def test_image_candidate_skips_source_upload_and_runs_jobs(tmp_path):
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"
export_command = next(command for command, _, _ in workspaces[0].backend.commands if command[:2] == ["tar", "-czf"])
assert "logs" in export_command
assert "results" in export_command
assert "reme_workspace" in export_command
assert "inbox" not in export_command
assert "--exclude=reme_workspace/metadata" in export_command
assert "--exclude=reme_workspace/resource" in export_command
assert "--exclude=reme_workspace/session" not in export_command
@pytest.mark.asyncio
async def test_one_container_can_reset_and_run_another_case(tmp_path):
"""Reset removes case state while retaining the installed candidate and worker."""
workspaces = []
def build(case_id, image, env):
"""Capture the single workspace reused across cases."""
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-1")
await case.run_job("auto_memory", {"session_id": "first"})
await case.export(tmp_path / "case-1.tar.gz")
backend = workspaces[0].backend
worker_bytes = backend.files["/workspace/harness/worker.py"]
backend.files["/workspace/candidate/reme/__init__.py"] = b"candidate-code"
assert any(path.startswith("/workspace/case/") for path in backend.files)
await case.reset_case("case-2")
assert case.case_id == "case-2"
assert len(workspaces) == 1
assert backend.files["/workspace/harness/worker.py"] == worker_bytes
assert backend.files["/workspace/candidate/reme/__init__.py"] == b"candidate-code"
assert not any(path.startswith("/workspace/case/") for path in backend.files)
assert "/tmp/reme-case-export.tar.gz" not in backend.files
result = await case.run_job("agentic_answer", {"query": "second"})
await case.export(tmp_path / "case-2.tar.gz")
assert result.success is True
manifest = json.loads(backend.files["/workspace/case/manifest.json"])
assert manifest["case_id"] == "case-2"
@pytest.mark.asyncio
async def test_full_export_keeps_complete_case_tree(tmp_path):
"""The explicit full profile preserves the previous export contract."""
workspace = FakeWorkspace("case-1", "candidate:test", {})
factory = DockerReMeSandboxFactory(
ImageCandidate("candidate:test"),
workspace_builder=lambda *_args: workspace,
)
case = await factory.create_case("case-1")
await case.export_full(tmp_path / "full.tar.gz")
export_command = next(command for command, _, _ in workspace.backend.commands if command[:2] == ["tar", "-czf"])
assert export_command == [
"tar",
"-czf",
"/tmp/reme-case-export.tar.gz",
"-C",
"/workspace/case",
".",
]
manifest = json.loads(workspace.backend.files["/workspace/case/manifest.json"])
assert manifest["export_profile"] == "full"
@pytest.mark.asyncio
async def test_analysis_export_requires_runtime_layout(tmp_path):
"""Selective export fails clearly before any job resolves configured paths."""
workspace = FakeWorkspace("case-1", "candidate:test", {})
factory = DockerReMeSandboxFactory(
ImageCandidate("candidate:test"),
workspace_builder=lambda *_args: workspace,
)
case = await factory.create_case("case-1")
with pytest.raises(RuntimeError, match="completed job"):
await case.export(tmp_path / "analysis.tar.gz")
def test_runtime_layout_uses_resolved_custom_memory_directories(tmp_path):
"""Selective export paths come from config instead of fixed directory names."""
workspace = tmp_path / "runtime"
config = ApplicationConfig(
workspace_dir=str(workspace),
daily_dir="memories/daily-notes",
digest_dir="memories/digests",
mem_session_dir="traces/agent-sessions",
)
worker._write_runtime_layout(tmp_path, config) # pylint: disable=protected-access
layout = json.loads((tmp_path / "runtime-layout.json").read_text())
assert layout["configured_paths"]["daily_dir"] == "runtime/memories/daily-notes"
assert layout["configured_paths"]["digest_dir"] == "runtime/memories/digests"
assert layout["configured_paths"]["mem_session_dir"] == "runtime/traces/agent-sessions"
assert layout["analysis_excludes"] == [
"runtime/metadata",
"runtime/resource",
]
@pytest.mark.asyncio
async def test_reset_case_rejects_unsafe_id_without_deleting_state():
"""An invalid next case ID cannot trigger cleanup."""
workspace = FakeWorkspace("case-1", "candidate:test", {})
factory = DockerReMeSandboxFactory(
ImageCandidate("candidate:test"),
workspace_builder=lambda *_args: workspace,
)
case = await factory.create_case("case-1")
before = list(workspace.backend.commands)
with pytest.raises(ValueError, match="case_id"):
await case.reset_case("../case-2")
assert case.case_id == "case-1"
assert workspace.backend.commands == before
@pytest.mark.asyncio
async def test_jobs_in_one_container_are_serialized():
"""Concurrent host calls cannot execute two case jobs at once."""
class TrackingBackend(FakeBackend):
"""Measure concurrent worker processes in one fake container."""
def __init__(self):
super().__init__()
self.active_jobs = 0
self.max_active_jobs = 0
async def exec_shell(self, command, *, cwd=None, timeout=None):
is_job = "/workspace/harness/worker.py" in command
if is_job:
self.active_jobs += 1
self.max_active_jobs = max(self.max_active_jobs, self.active_jobs)
await asyncio.sleep(0.01)
try:
return await super().exec_shell(command, cwd=cwd, timeout=timeout)
finally:
if is_job:
self.active_jobs -= 1
backend = TrackingBackend()
workspace = FakeWorkspace("case-1", "candidate:test", {}, backend=backend)
factory = DockerReMeSandboxFactory(
ImageCandidate("candidate:test"),
workspace_builder=lambda *_args: workspace,
)
case = await factory.create_case("case-1")
first, second = await asyncio.gather(
case.run_job("agentic_answer", {"query": "first"}),
case.run_job("agentic_answer", {"query": "second"}),
)
assert first.success is second.success is True
assert backend.max_active_jobs == 1
@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 = []
monkeypatch.setenv("TMPDIR", str(tmp_path / "original-tmp"))
class FakeReMe:
"""Record the direct ReMe application lifecycle."""
@ -223,6 +412,13 @@ async def test_worker_calls_application_directly_and_always_closes(monkeypatch,
)
assert result == {"success": True, "answer": "answer", "metadata": {"direct": True}, "error": None}
assert worker.os.environ["TMPDIR"] == str(tmp_path / "tmp")
layout = json.loads((tmp_path / "runtime-layout.json").read_text())
assert layout["workspace_root"] == "workspace"
assert layout["analysis_excludes"] == [
"workspace/metadata",
"workspace/resource",
]
assert events == [
("construct", str(tmp_path / "workspace")),
"start",