diff --git a/pyproject.toml b/pyproject.toml index 7a8e3ac6..89d65af4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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", diff --git a/sandbox/Dockerfile.base b/sandbox/Dockerfile.base index de7dd7a3..2e8d3f02 100644 --- a/sandbox/Dockerfile.base +++ b/sandbox/Dockerfile.base @@ -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 diff --git a/sandbox/README.md b/sandbox/README.md index 62507be8..6cd4ac7a 100644 --- a/sandbox/README.md +++ b/sandbox/README.md @@ -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. diff --git a/sandbox/docker_sandbox.py b/sandbox/docker_sandbox.py index 4a233c79..84fe3814 100644 --- a/sandbox/docker_sandbox.py +++ b/sandbox/docker_sandbox.py @@ -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 diff --git a/sandbox/worker.py b/sandbox/worker.py index 102c8b80..23d33654 100644 --- a/sandbox/worker.py +++ b/sandbox/worker.py @@ -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() diff --git a/tests/integration/test_sandbox_longmemeval.py b/tests/integration/test_sandbox_longmemeval.py new file mode 100644 index 00000000..8748645b --- /dev/null +++ b/tests/integration/test_sandbox_longmemeval.py @@ -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) diff --git a/tests/unit/test_sandbox.py b/tests/unit/test_sandbox.py index bec258dc..7ae92d62 100644 --- a/tests/unit/test_sandbox.py +++ b/tests/unit/test_sandbox.py @@ -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",