Simplify benchmark worker requests and preserve artifacts

This commit is contained in:
酱牛肉 2026-10-13 16:44:08 +08:00
parent 132a0ee7cc
commit 082b8f8ccb
4 changed files with 307 additions and 190 deletions

View file

@ -63,11 +63,17 @@ and query-only cases; it includes `build_log` when that phase ran. Set
build, the pool also returns its text in `CaseResult.build_log`. A timed-out
AgentScope shell command makes its container unusable: the pool closes it after
retaining the log and creates a replacement when more cases need a worker.
The build log also records configuration and Application construction failures,
including their tracebacks.
For direct control, call `create_case()` or `create_cases()`, then use
`run_job()`, `run_build()`, `run_queries()`, `export_workspace()`,
`upload_workspace()`, `reset_case()`, and `close()`. Always close a case in a
`finally` block or use its async context manager. Export before closing.
The default `export("analysis.tar.gz")` also saves available construction and
query logs under `analysis.tar.gz.artifacts/build_log/` and
`analysis.tar.gz.artifacts/queries/`. These logs stay outside the archive;
choose a new destination if that artifacts directory already exists.
The candidate receives only query ID, question, answer Job, and answer
arguments. Golden answers, rubrics, and scoring remain on the host. Workspace

View file

@ -28,8 +28,6 @@ WORKSPACE_ARCHIVE = "/tmp/reme-workspace-upload.tar.gz"
WORKSPACE_EXPORT_ARCHIVE = "/tmp/reme-workspace-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}$")
_GIT_AUTHOR_NAME = "ReMe Sandbox"
_GIT_AUTHOR_EMAIL = "reme-sandbox@localhost"
_PRIVATE_EVALUATION_FIELDS = frozenset(
{"query", "question", "golden_answer", "question_type", "rubric", "judge_arguments"},
)
@ -500,29 +498,8 @@ class DockerReMeSandbox:
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"
response_path = f"{self.results_dir}/{request_id}.json"
request = {
"job": job,
"arguments": arguments or {},
"config": self.config,
"plugins": list(self.plugins),
"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)
response = await self._run_worker_request("job", {"job": job, "arguments": arguments or {}})
return JobResult.from_dict(response)
async def _run_worker_request(self, mode: str, payload: dict[str, Any]) -> dict[str, Any]:
"""Run one structured worker request and return its JSON response."""
@ -540,15 +517,17 @@ class DockerReMeSandbox:
**payload,
}
await self.backend.write_file(request_path, json.dumps(request, ensure_ascii=False).encode("utf-8"))
action_name = f"job:{payload['job']}" if mode == "job" else f"worker:{mode}"
await self._exec(
f"worker:{mode}",
action_name,
[self.python, f"{HARNESS_ROOT}/worker.py", "--request", request_path, "--response", response_path],
cwd=CASE_ROOT,
)
try:
result = json.loads((await self.backend.read_file(response_path)).decode("utf-8"))
except FileNotFoundError as exc:
raise SandboxCommandError(f"worker mode {mode!r} did not produce {response_path}") from exc
description = f"job {payload['job']!r}" if mode == "job" else f"worker mode {mode!r}"
raise SandboxCommandError(f"{description} did not produce {response_path}") from exc
if not isinstance(result, dict):
raise SandboxCommandError(f"worker mode {mode!r} returned a non-object response")
return result
@ -674,77 +653,9 @@ class DockerReMeSandbox:
if not isinstance(message, str) or not message.strip():
raise ValueError("memory history commit message must be a non-empty string")
async with self._operation_lock:
await self._commit_memory_history(message)
async def _daily_history_path(self) -> str:
"""Return the validated workspace-relative daily directory from the runtime layout."""
return (await self._memory_history_paths())[0]
async def _memory_history_paths(self) -> list[str]:
"""Return validated workspace-relative daily, digest, and session directories."""
assert self.backend is not None
try:
payload = json.loads((await self.backend.read_file(RUNTIME_LAYOUT)).decode("utf-8"))
except (FileNotFoundError, UnicodeDecodeError, json.JSONDecodeError) as exc:
raise SandboxCommandError(f"cannot resolve daily history path from runtime layout: {exc}") from exc
expected_root = PurePosixPath(self.runtime_workspace).relative_to(CASE_ROOT)
configured_paths = payload.get("configured_paths") if isinstance(payload, dict) else None
if not isinstance(configured_paths, dict):
raise SandboxCommandError("sandbox runtime layout has invalid configured_paths")
paths: list[str] = []
for field in ("daily_dir", "digest_dir", "session_dir"):
case_relative = self._validated_workspace_path(
configured_paths.get(field),
expected_root,
allow_root=False,
)
relative_path = PurePosixPath(case_relative).relative_to(expected_root).as_posix()
if relative_path not in paths:
paths.append(relative_path)
return paths
async def _commit_memory_history(self, message: str) -> None:
"""Commit one host-selected boundary across configured durable memory directories."""
memory_paths = await self._memory_history_paths()
await self._exec_checked(
"stage_memory_history",
["git", "add", "-A", "--", *memory_paths],
cwd=self.runtime_workspace,
)
changed_paths: list[str] = []
for path in memory_paths:
diff_result = await self._exec(
"check_memory_history",
["git", "diff", "--cached", "--quiet", "--", path],
cwd=self.runtime_workspace,
)
if diff_result.exit_code not in (0, 1):
raise SandboxCommandError(
f"cannot inspect staged memory changes: {_decode(diff_result.stderr)}",
)
if diff_result.exit_code == 1:
changed_paths.append(path)
commit_command = [
"git",
"-c",
f"user.name={_GIT_AUTHOR_NAME}",
"-c",
f"user.email={_GIT_AUTHOR_EMAIL}",
"commit",
"--quiet",
"--allow-empty",
"--only",
"-m",
message,
]
if changed_paths:
commit_command.extend(["--", *changed_paths])
await self._exec_checked(
"commit_memory_history",
commit_command,
cwd=self.runtime_workspace,
)
result = await self._run_worker_request("checkpoint", {"message": message})
if result.get("success") is not True:
raise SandboxCommandError(result.get("error") or "memory history checkpoint failed")
async def answer(self, *, query: str, query_time: str = "") -> JobResult:
"""Generate an answer with the candidate's agentic_answer job."""
@ -799,7 +710,11 @@ class DockerReMeSandbox:
profile: Literal["analysis", "full"] = "analysis",
include_candidate: bool = False,
) -> Path:
"""Download analysis artifacts, or the full case when requested."""
"""Download the case archive and, for analysis, separate build/query logs.
Analysis logs are saved under ``<destination>.artifacts`` without
changing the archive's contents.
"""
if profile not in ("analysis", "full"):
raise ValueError("export profile must be 'analysis' or 'full'")
async with self._operation_lock:
@ -1034,6 +949,10 @@ class DockerReMeSandbox:
) -> Path:
"""Export the active case while the caller holds the operation lock."""
assert self.backend is not None, "sandbox is not initialized"
target = Path(destination).expanduser().resolve()
artifacts = Path(f"{target}.artifacts")
if profile == "analysis" and artifacts.exists():
raise FileExistsError(f"analysis artifacts already exist: {artifacts}")
if include_candidate and isinstance(self.candidate, ImageCandidate):
raise ValueError("an image candidate has no source directory to export")
candidate_id = (
@ -1086,11 +1005,50 @@ class DockerReMeSandbox:
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)
if profile == "analysis":
await self._export_analysis_logs(artifacts)
return target
async def _export_analysis_logs(self, destination: Path) -> None:
"""Save available build and query artifacts alongside the analysis archive."""
assert self.backend is not None
try:
build_log = await self.backend.read_file(f"{CASE_ROOT}/build_log/build.log")
except FileNotFoundError:
build_log = None
if build_log is not None:
log_path = destination / "build_log" / "build.log"
log_path.parent.mkdir(parents=True, exist_ok=True)
log_path.write_bytes(build_log)
try:
await self.backend.read_file(f"{CASE_ROOT}/queries/summary.json")
except FileNotFoundError:
return
await self._exec_checked(
"export_analysis_queries",
["tar", "-czf", EXPORT_ARCHIVE, "-C", CASE_ROOT, "queries"],
)
archive = await self.backend.read_file(EXPORT_ARCHIVE)
with tarfile.open(fileobj=io.BytesIO(archive), mode="r:gz") as source:
for member in source:
path = _workspace_member_path(member.name)
if path is None or not path.parts or path.parts[0] != "queries":
raise SandboxCommandError(f"query export contains unsafe path: {member.name!r}")
output = destination.joinpath(*path.parts)
if member.isdir():
output.mkdir(parents=True, exist_ok=True)
elif member.isfile():
content = source.extractfile(member)
if content is None:
raise SandboxCommandError(f"query export cannot read: {member.name!r}")
output.parent.mkdir(parents=True, exist_ok=True)
output.write_bytes(content.read())
else:
raise SandboxCommandError(f"query export contains unsupported member: {member.name!r}")
async def reset_case(self, case_id: str) -> None:
"""Discard the active case and prepare this container for another.

View file

@ -211,18 +211,30 @@ async def _run(request: dict[str, Any]) -> dict[str, Any]:
async def _run_build(request: dict[str, Any]) -> dict[str, Any]:
"""Execute all construction jobs in one Application and one build log."""
case_root, app_config = _prepare_runtime(request)
from reme.reme import ReMe
from reme.utils import get_logger
jobs = request.get("jobs")
if not isinstance(jobs, list) or not jobs:
raise ValueError("build request requires a non-empty jobs list")
build_log = case_root / "build_log" / "build.log"
build_log = Path(request["case_root"]) / "build_log" / "build.log"
if build_log.exists():
raise FileExistsError(f"build artifact already exists: {build_log}")
build_log.parent.mkdir(parents=True, exist_ok=True)
build_log.touch()
app = ReMe(**app_config)
try:
case_root, app_config = _prepare_runtime(request)
from reme.reme import ReMe
from reme.utils import get_logger
jobs = request.get("jobs")
if not isinstance(jobs, list) or not jobs:
raise ValueError("build request requires a non-empty jobs list")
app = ReMe(**app_config)
except Exception:
# Config resolution and Application construction can fail before the
# ReMe logger is usable. Write the traceback to the durable artifact.
with build_log.open("a", encoding="utf-8") as log_file:
traceback.print_exc(file=log_file)
raise
# Application.__init__ force-initializes the global logger, removing any
# sink bound before construction. Attach the build artifact afterwards.
logger = _start_case_log(build_log)
logger.info("[sandbox] build phase started")
results: list[dict[str, Any]] = []
@ -313,6 +325,15 @@ def _commit_memory_checkpoint(case_root: Path, app_config: dict[str, Any], messa
}
async def _run_checkpoint(request: dict[str, Any]) -> dict[str, Any]:
"""Commit host-requested memory history using the build checkpoint rules."""
message = request.get("message")
if not isinstance(message, str) or not message.strip():
raise ValueError("memory history commit message must be a non-empty string")
case_root, app_config = _prepare_runtime(request)
return {"success": True, "memory_checkpoint": _commit_memory_checkpoint(case_root, app_config, message)}
async def _run_queries(request: dict[str, Any], *, write_summary: bool = True) -> dict[str, Any]:
"""Execute answer requests in one candidate Application with isolated logs.
@ -435,6 +456,8 @@ def main() -> int:
operation = _run_queries(request)
elif mode == "query":
operation = _run_queries(request, write_summary=False)
elif mode == "checkpoint":
operation = _run_checkpoint(request)
else:
raise ValueError(f"unknown sandbox worker mode: {mode!r}")
result = asyncio.run(operation)

View file

@ -1,11 +1,15 @@
"""Pool ordering, reuse, and cleanup without Docker or model calls."""
# pylint: disable=missing-class-docstring,missing-function-docstring,unused-argument
# pylint: disable=missing-class-docstring,missing-function-docstring,protected-access,unused-argument
from __future__ import annotations
import asyncio
import io
import json
from pathlib import Path
import subprocess
import tarfile
import pytest
@ -19,7 +23,7 @@ from deploy.benchmark_docker import (
SourceSnapshot,
)
from deploy.benchmark_docker.docker_sandbox import SandboxCommandError, SandboxCommandTimeoutError
from deploy.benchmark_docker.worker import _start_case_log
from deploy.benchmark_docker import worker as sandbox_worker
class FakeCase:
@ -131,10 +135,12 @@ async def test_pool_exports_query_only_case_from_reused_workspace(tmp_path):
@pytest.mark.asyncio
async def test_pool_keeps_build_log_on_failure(tmp_path):
@pytest.mark.parametrize("raises", [False, True], ids=["failed-result", "exception"])
async def test_pool_keeps_build_log_on_failure(tmp_path, raises):
class FailedBuildCase(FakeCase):
async def run_build(self, jobs):
self.events.append((self.case_id, "build_failed"))
if raises:
raise TimeoutError("build exceeded command timeout")
return {"success": False, "jobs": [{"error": "failed"}]}
class FailedBuildFactory(FakeFactory):
@ -148,36 +154,17 @@ async def test_pool_keeps_build_log_on_failure(tmp_path):
result = (await BenchmarkDockerPool(FailedBuildFactory(events), concurrency=1).run([plan]))[0]
assert not result.success
assert result.build == {"success": False, "jobs": [{"error": "failed"}]}
if raises:
assert result.error == "TimeoutError: build exceeded command timeout"
assert result.build is None
else:
assert result.error == "build failed"
assert result.build == {"success": False, "jobs": [{"error": "failed"}]}
assert result.build_log == "build failed"
assert log_path.read_text(encoding="utf-8") == "build failed"
assert events.index(("failed", "read_build_log")) < events.index(("pool-0", "close"))
@pytest.mark.asyncio
async def test_pool_keeps_build_log_when_build_raises(tmp_path):
class TimedOutBuildCase(FakeCase):
async def run_build(self, jobs):
self.events.append((self.case_id, "build_timed_out"))
raise TimeoutError("build exceeded command timeout")
class TimedOutBuildFactory(FakeFactory):
async def create_cases(self, case_ids, *, concurrency):
return [TimedOutBuildCase(name, self.events) for name in case_ids]
events: list[tuple] = []
log_path = tmp_path / "build.log"
plan = CasePlan("timed-out", build_jobs=(JobRequest("build"),), export_build_log_to=log_path)
result = (await BenchmarkDockerPool(TimedOutBuildFactory(events), concurrency=1).run([plan]))[0]
assert result.error == "TimeoutError: build exceeded command timeout"
assert result.build is None
assert result.build_log == "build failed"
assert log_path.read_text(encoding="utf-8") == "build failed"
assert events.index(("timed-out", "read_build_log")) < events.index(("pool-0", "close"))
@pytest.mark.asyncio
async def test_pool_cancels_case_before_closing_container():
started = asyncio.Event()
@ -215,23 +202,109 @@ def test_source_snapshot_excludes_unrelated_and_secret_files(tmp_path):
assert snapshot.file_count == 1
def test_worker_writes_log_to_requested_artifact_path(tmp_path):
"""The current ReMe logger accepts a directory, so bind the exact log file explicitly."""
from reme.utils import get_logger
@pytest.mark.asyncio
@pytest.mark.parametrize("failure_phase", ["config", "application"])
async def test_build_log_captures_initialization_failure(tmp_path, monkeypatch, failure_phase):
if failure_phase == "config":
path = tmp_path / "build_log" / "build.log"
logger = _start_case_log(path)
try:
logger.info("benchmark marker")
finally:
get_logger(log_to_console=False, log_to_file=False, force_init=True)
assert "benchmark marker" in path.read_text(encoding="utf-8")
def fail_config(_request):
raise ValueError("invalid benchmark config")
monkeypatch.setattr(sandbox_worker, "_prepare_runtime", fail_config)
expected = "ValueError: invalid benchmark config"
else:
config = {
"workspace_dir": str(tmp_path / "workspace"),
"service": {"backend": "cli"},
"jobs": {"broken": {"backend": "missing"}},
"enable_logo": False,
"log_to_console": False,
"log_to_file": False,
}
monkeypatch.setattr(sandbox_worker, "_prepare_runtime", lambda _request: (tmp_path, config))
expected = "ValueError: Unregistered backend 'missing' for Job 'broken'"
with pytest.raises(ValueError, match="invalid benchmark config|Unregistered backend"):
await sandbox_worker._run_build({"case_root": str(tmp_path), "jobs": [{"job": "build"}]})
log_text = (tmp_path / "build_log" / "build.log").read_text(encoding="utf-8")
assert "Traceback (most recent call last):" in log_text
assert expected in log_text
@pytest.mark.asyncio
async def test_build_log_remains_bound_after_application_initializes_logger(tmp_path, monkeypatch):
import reme.reme as reme_module
monkeypatch.setattr(sandbox_worker, "_prepare_runtime", lambda _request: (tmp_path, {}))
class FakeReMe:
def __init__(self, **_kwargs):
from reme.utils import get_logger
get_logger(log_to_console=False, log_to_file=False, force_init=True)
async def start(self):
pass
async def close(self):
pass
async def run_job(_app, _job, _arguments):
return {"success": True}
monkeypatch.setattr(reme_module, "ReMe", FakeReMe)
monkeypatch.setattr(sandbox_worker, "_run_job_on_app", run_job)
result = await sandbox_worker._run_build({"case_root": str(tmp_path), "jobs": [{"job": "build"}]})
assert result["success"]
log_text = (tmp_path / "build_log" / "build.log").read_text(encoding="utf-8")
assert "[sandbox] build phase started" in log_text
assert "[sandbox] build phase finished" in log_text
@pytest.mark.asyncio
async def test_worker_checkpoint_commits_only_memory_paths(tmp_path, monkeypatch):
workspace = tmp_path / "workspace"
workspace.mkdir()
subprocess.run(["git", "init", "--quiet", str(workspace)], check=True)
(workspace / "daily").mkdir()
(workspace / "daily" / "note.md").write_text("memory", encoding="utf-8")
(workspace / "digest").mkdir()
(workspace / "digest" / "note.md").write_text("digest", encoding="utf-8")
(workspace / "sessions").mkdir()
(workspace / "sessions" / "note.md").write_text("session", encoding="utf-8")
(workspace / "other.txt").write_text("unrelated", encoding="utf-8")
config = {"workspace_dir": str(workspace), "daily_dir": "daily", "digest_dir": "digest", "session_dir": "sessions"}
monkeypatch.setattr(sandbox_worker, "_prepare_runtime", lambda _request: (tmp_path, config))
result = await sandbox_worker._run_checkpoint({"message": "memory boundary"})
assert result["success"] is True
assert result["memory_checkpoint"]["message"] == "memory boundary"
committed = subprocess.run(
["git", "show", "--pretty=format:", "--name-only", "HEAD"],
cwd=workspace,
capture_output=True,
text=True,
check=True,
)
assert set(committed.stdout.splitlines()) == {"daily/note.md", "digest/note.md", "sessions/note.md"}
assert result["memory_checkpoint"]["paths"] == ["daily", "digest", "sessions"]
second = await sandbox_worker._run_checkpoint({"message": "empty boundary"})
assert second["memory_checkpoint"]["commit_sha"] != result["memory_checkpoint"]["commit_sha"]
config["daily_dir"] = "../outside"
with pytest.raises(ValueError, match="escapes the runtime workspace"):
await sandbox_worker._run_checkpoint({"message": "unsafe boundary"})
class FakeBackend:
def __init__(self) -> None:
self.files: dict[str, bytes] = {}
self.commands: list[list[str]] = []
self.worker_responses: dict[str, dict] = {}
async def write_file(self, path: str, data: bytes) -> None:
self.files[path] = data
@ -243,11 +316,11 @@ class FakeBackend:
raise FileNotFoundError(path) from exc
async def exec_shell(self, command: list[str], *, cwd=None, timeout=None):
import json
from types import SimpleNamespace
self.commands.append(command)
output = b""
exit_code = 0
if len(command) > 2 and command[1] == "-c":
output = json.dumps(
{
@ -258,11 +331,14 @@ class FakeBackend:
).encode()
if "/workspace/harness/worker.py" in command:
request = json.loads(self.files[command[command.index("--request") + 1]])
response = {"success": True, "jobs": []} if request.get("mode") == "build" else {"success": True}
mode = request.get("mode", "job")
default = {"success": True, "jobs": []} if mode == "build" else {"success": True}
response = self.worker_responses.get(mode, default)
self.files[command[command.index("--response") + 1]] = json.dumps(response).encode()
exit_code = 0 if response["success"] else 1
if command[:2] == ["tar", "-czf"]:
self.files[command[2]] = b"workspace archive"
return SimpleNamespace(exit_code=0, stdout=output, stderr=b"", ok=lambda: True)
return SimpleNamespace(exit_code=exit_code, stdout=output, stderr=b"", ok=lambda: exit_code == 0)
class TimedOutBackend(FakeBackend):
@ -291,16 +367,20 @@ class FakeWorkspace:
self.closed = True
@pytest.mark.asyncio
async def test_backend_timeout_prevents_direct_case_reset(tmp_path):
def _fake_docker_factory(tmp_path, workspace_builder, *, install_projects=(".",)):
(tmp_path / "reme").mkdir()
(tmp_path / "reme" / "__init__.py").write_text("ok", encoding="utf-8")
workspace = FakeWorkspace(TimedOutBackend())
factory = DockerReMeSandboxFactory(
SourceCandidate(SourceSnapshot.from_directory(tmp_path)),
return DockerReMeSandboxFactory(
SourceCandidate(SourceSnapshot.from_directory(tmp_path), install_projects=install_projects),
plugins=("lme",),
workspace_builder=lambda _id, _image, _env: workspace,
workspace_builder=workspace_builder,
)
@pytest.mark.asyncio
async def test_backend_timeout_prevents_direct_case_reset(tmp_path):
workspace = FakeWorkspace(TimedOutBackend())
factory = _fake_docker_factory(tmp_path, lambda _id, _image, _env: workspace)
case = await factory.create_case("timed-out")
try:
with pytest.raises(SandboxCommandTimeoutError):
@ -317,14 +397,12 @@ async def test_backend_timeout_prevents_direct_case_reset(tmp_path):
@pytest.mark.asyncio
async def test_source_candidate_installs_plugin_and_closes_container(tmp_path):
(tmp_path / "reme").mkdir()
(tmp_path / "reme" / "__init__.py").write_text("ok", encoding="utf-8")
backend = FakeBackend()
workspace = FakeWorkspace(backend)
factory = DockerReMeSandboxFactory(
SourceCandidate(SourceSnapshot.from_directory(tmp_path), install_projects=(".", "plugins/lme")),
plugins=("lme",),
workspace_builder=lambda _id, _image, _env: workspace,
factory = _fake_docker_factory(
tmp_path,
lambda _id, _image, _env: workspace,
install_projects=(".", "plugins/lme"),
)
case = await factory.create_case("example")
try:
@ -343,18 +421,43 @@ async def test_source_candidate_installs_plugin_and_closes_container(tmp_path):
assert workspace.closed
@pytest.mark.asyncio
async def test_job_and_checkpoint_share_worker_request_path(tmp_path):
backend = FakeBackend()
workspace = FakeWorkspace(backend)
factory = _fake_docker_factory(tmp_path, lambda _id, _image, _env: workspace)
case = await factory.create_case("example")
try:
result = await case.run_job("answer", {"query": "hello"})
assert result.success
assert case.actions[-1].name == "job:answer"
job_request = json.loads(backend.files[next(path for path in backend.files if "/inbox/" in path)])
assert job_request["mode"] == "job"
assert job_request["arguments"] == {"query": "hello"}
await case.commit_memory_history("memory boundary")
assert case.actions[-1].name == "worker:checkpoint"
requests = [json.loads(content) for path, content in backend.files.items() if "/inbox/" in path]
assert requests[-1]["mode"] == "checkpoint"
assert requests[-1]["message"] == "memory boundary"
assert not any(command[0] == "git" and "commit" in command for command in backend.commands)
with pytest.raises(ValueError, match="non-empty"):
await case.commit_memory_history(" ")
backend.worker_responses["checkpoint"] = {"success": False, "error": "git commit failed"}
with pytest.raises(SandboxCommandError, match="git commit failed"):
await case.commit_memory_history("next boundary")
assert case.actions[-1].exit_code == 1
finally:
await case.close()
@pytest.mark.asyncio
async def test_pool_replaces_worker_after_backend_timeout(tmp_path):
(tmp_path / "reme").mkdir()
(tmp_path / "reme" / "__init__.py").write_text("ok", encoding="utf-8")
timed_out_workspace = FakeWorkspace(TimedOutBackend())
replacement_workspace = FakeWorkspace(FakeBackend())
workspaces = iter((timed_out_workspace, replacement_workspace))
factory = DockerReMeSandboxFactory(
SourceCandidate(SourceSnapshot.from_directory(tmp_path)),
plugins=("lme",),
workspace_builder=lambda _id, _image, _env: next(workspaces),
)
factory = _fake_docker_factory(tmp_path, lambda _id, _image, _env: next(workspaces))
plans = [
CasePlan("timed-out", build_jobs=(JobRequest("build"),), export_build_log_to=tmp_path / "build.log"),
CasePlan("next", build_jobs=(JobRequest("build"),)),
@ -372,15 +475,9 @@ async def test_pool_replaces_worker_after_backend_timeout(tmp_path):
@pytest.mark.asyncio
async def test_query_only_evaluation_export_omits_missing_build_log(tmp_path):
(tmp_path / "reme").mkdir()
(tmp_path / "reme" / "__init__.py").write_text("ok", encoding="utf-8")
backend = FakeBackend()
workspace = FakeWorkspace(backend)
factory = DockerReMeSandboxFactory(
SourceCandidate(SourceSnapshot.from_directory(tmp_path)),
plugins=("lme",),
workspace_builder=lambda _id, _image, _env: workspace,
)
factory = _fake_docker_factory(tmp_path, lambda _id, _image, _env: workspace)
case = await factory.create_case("query-only")
try:
backend.files["/workspace/case/queries/summary.json"] = b"{}"
@ -395,10 +492,52 @@ async def test_query_only_evaluation_export_omits_missing_build_log(tmp_path):
await case.close()
@pytest.mark.asyncio
async def test_default_analysis_export_saves_logs_outside_archive(tmp_path):
class AnalysisBackend(FakeBackend):
async def exec_shell(self, command, *, cwd=None, timeout=None):
result = await super().exec_shell(command, cwd=cwd, timeout=timeout)
if command[:2] == ["tar", "-czf"] and command[-1] == "queries":
stream = io.BytesIO()
with tarfile.open(fileobj=stream, mode="w:gz") as archive:
for name, content in (
("queries/summary.json", b"{}"),
("queries/q1/answer.log", b"answer log"),
):
info = tarfile.TarInfo(name)
info.size = len(content)
archive.addfile(info, io.BytesIO(content))
self.files[command[2]] = stream.getvalue()
return result
backend = AnalysisBackend()
workspace = FakeWorkspace(backend)
factory = _fake_docker_factory(tmp_path, lambda _id, _image, _env: workspace)
case = await factory.create_case("example")
try:
backend.files["/workspace/case/runtime-layout.json"] = (
b'{"workspace_root":"reme_workspace","analysis_excludes":[]}'
)
backend.files["/workspace/case/build_log/build.log"] = b"build log"
backend.files["/workspace/case/queries/summary.json"] = b"{}"
archive = tmp_path / "analysis.tar.gz"
assert await case.export(archive) == archive
assert archive.read_bytes() == b"workspace archive"
artifacts = tmp_path / "analysis.tar.gz.artifacts"
assert (artifacts / "build_log" / "build.log").read_bytes() == b"build log"
assert (artifacts / "queries" / "summary.json").read_bytes() == b"{}"
assert (artifacts / "queries" / "q1" / "answer.log").read_bytes() == b"answer log"
archive_command = next(
command for command in backend.commands if command[:3] == ["tar", "-czf", "/tmp/reme-case-export.tar.gz"]
)
assert "build_log" not in archive_command
assert "queries" not in archive_command
finally:
await case.close()
@pytest.mark.asyncio
async def test_cancelled_case_initialization_closes_workspace(tmp_path):
(tmp_path / "reme").mkdir()
(tmp_path / "reme" / "__init__.py").write_text("ok", encoding="utf-8")
started = asyncio.Event()
class BlockingWorkspace(FakeWorkspace):
@ -407,10 +546,7 @@ async def test_cancelled_case_initialization_closes_workspace(tmp_path):
await asyncio.Event().wait()
workspace = BlockingWorkspace(FakeBackend())
factory = DockerReMeSandboxFactory(
SourceCandidate(SourceSnapshot.from_directory(tmp_path)),
workspace_builder=lambda _id, _image, _env: workspace,
)
factory = _fake_docker_factory(tmp_path, lambda _id, _image, _env: workspace)
task = asyncio.create_task(factory.create_case("cancelled"))
await asyncio.wait_for(started.wait(), timeout=2)
task.cancel()
@ -421,8 +557,6 @@ async def test_cancelled_case_initialization_closes_workspace(tmp_path):
@pytest.mark.asyncio
async def test_cancelled_batch_creation_closes_completed_workspace(tmp_path):
(tmp_path / "reme").mkdir()
(tmp_path / "reme" / "__init__.py").write_text("ok", encoding="utf-8")
second_started = asyncio.Event()
class BlockingWorkspace(FakeWorkspace):
@ -433,11 +567,7 @@ async def test_cancelled_batch_creation_closes_completed_workspace(tmp_path):
first = FakeWorkspace(FakeBackend())
second = BlockingWorkspace(FakeBackend())
workspaces = iter((first, second))
factory = DockerReMeSandboxFactory(
SourceCandidate(SourceSnapshot.from_directory(tmp_path)),
plugins=("lme",),
workspace_builder=lambda _id, _image, _env: next(workspaces),
)
factory = _fake_docker_factory(tmp_path, lambda _id, _image, _env: next(workspaces))
task = asyncio.create_task(factory.create_cases(["first", "second"], concurrency=2))
await asyncio.wait_for(second_started.wait(), timeout=2)
await asyncio.sleep(0)