diff --git a/deploy/benchmark_docker/README.md b/deploy/benchmark_docker/README.md index d6f6259b..0ec0cb77 100644 --- a/deploy/benchmark_docker/README.md +++ b/deploy/benchmark_docker/README.md @@ -60,8 +60,9 @@ archive. A reused workspace may be queried without additional build jobs. `export_evaluation_to` saves the workspace and query artifacts for both build and query-only cases; it includes `build_log` when that phase ran. Set `export_build_log_to` to retain the construction log separately. On a failed -build, the pool also returns its text in `CaseResult.build_log` before reusing -or closing the container. +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. For direct control, call `create_case()` or `create_cases()`, then use `run_job()`, `run_build()`, `run_queries()`, `export_workspace()`, diff --git a/deploy/benchmark_docker/docker_sandbox.py b/deploy/benchmark_docker/docker_sandbox.py index 04519558..b1c75e93 100644 --- a/deploy/benchmark_docker/docker_sandbox.py +++ b/deploy/benchmark_docker/docker_sandbox.py @@ -80,6 +80,10 @@ class SandboxCommandError(RuntimeError): """A required command failed inside a case sandbox.""" +class SandboxCommandTimeoutError(SandboxCommandError): + """A command timed out and may still be running inside the container.""" + + 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): @@ -332,8 +336,14 @@ class DockerReMeSandbox: self.python = "python3" self.actions: list[ActionRecord] = [] self._closed = False + self._reusable = True self._operation_lock = asyncio.Lock() + @property + def reusable(self) -> bool: + """Whether another case can safely use this container.""" + return self._reusable and not self._closed + @property def runtime_workspace(self) -> str: """ReMe-owned runtime workspace inside this case container.""" @@ -457,8 +467,14 @@ class DockerReMeSandbox: async def _exec(self, name: str, command: list[str], *, cwd: str | None = None) -> Any: assert self.backend is not None, "sandbox is not initialized" + if not self.reusable: + raise SandboxCommandError("sandbox container is no longer reusable") started = _utc_now() - result = await self.backend.exec_shell(command, cwd=cwd, timeout=self.command_timeout) + try: + result = await self.backend.exec_shell(command, cwd=cwd, timeout=self.command_timeout) + except TimeoutError as exc: + self._reusable = False + raise SandboxCommandTimeoutError(f"sandbox action {name!r} timed out; container must be destroyed") from exc self.actions.append( ActionRecord( name=name, @@ -469,6 +485,9 @@ class DockerReMeSandbox: finished_at=_utc_now(), ), ) + if result.exit_code == -1: + self._reusable = False + raise SandboxCommandTimeoutError(f"sandbox action {name!r} timed out; container must be destroyed") return result async def _exec_checked(self, name: str, command: list[str], *, cwd: str | None = None) -> Any: diff --git a/deploy/benchmark_docker/pool.py b/deploy/benchmark_docker/pool.py index aea2161c..cdd41101 100644 --- a/deploy/benchmark_docker/pool.py +++ b/deploy/benchmark_docker/pool.py @@ -93,6 +93,7 @@ class BenchmarkDockerPool: available: asyncio.Queue[DockerReMeSandbox] = asyncio.Queue() for worker in workers: available.put_nowait(worker) + claimed = 0 try: with TemporaryDirectory(prefix="reme-benchmark-pool-") as temporary: @@ -102,6 +103,7 @@ class BenchmarkDockerPool: tasks: dict[str, asyncio.Task[CaseResult]] = {} async def execute(plan: CasePlan) -> CaseResult: + nonlocal claimed result = CaseResult(plan.case_id) source = plan.workspace_source if plan.reuse_workspace_from is not None: @@ -111,6 +113,7 @@ class BenchmarkDockerPool: source = parent.workspace_snapshot worker = await available.get() + claimed += 1 async def retain_failed_build_log() -> None: try: @@ -158,7 +161,17 @@ class BenchmarkDockerPool: result.error = f"{type(exc).__name__}: {exc}" return result finally: - available.put_nowait(worker) + if worker.reusable: + available.put_nowait(worker) + else: + # AgentScope reports a shell timeout while its process + # may still be running. Destroy the container before + # another case can reset and use this pool slot. + await worker.close() + if claimed < len(plans) and not asyncio.current_task().cancelling(): + replacement = await self.factory.create_case(f"pool-{len(workers)}") + workers.append(replacement) + available.put_nowait(replacement) for plan in plans: tasks[plan.case_id] = asyncio.create_task(execute(plan)) diff --git a/tests/unit/test_benchmark_docker.py b/tests/unit/test_benchmark_docker.py index 2a729974..aefe3839 100644 --- a/tests/unit/test_benchmark_docker.py +++ b/tests/unit/test_benchmark_docker.py @@ -18,6 +18,7 @@ from deploy.benchmark_docker import ( SourceCandidate, SourceSnapshot, ) +from deploy.benchmark_docker.docker_sandbox import SandboxCommandError, SandboxCommandTimeoutError from deploy.benchmark_docker.worker import _start_case_log @@ -26,6 +27,7 @@ class FakeCase: self.name = name self.events = events self.case_id = name + self.reusable = True async def reset_case(self, case_id: str) -> None: self.case_id = case_id @@ -263,6 +265,17 @@ class FakeBackend: return SimpleNamespace(exit_code=0, stdout=output, stderr=b"", ok=lambda: True) +class TimedOutBackend(FakeBackend): + async def exec_shell(self, command: list[str], *, cwd=None, timeout=None): + if "/workspace/harness/worker.py" in command: + from types import SimpleNamespace + + self.commands.append(command) + self.files["/workspace/case/build_log/build.log"] = b"build still running" + return SimpleNamespace(exit_code=-1, stdout=b"", stderr=b"timed out", ok=lambda: False) + return await super().exec_shell(command, cwd=cwd, timeout=timeout) + + class FakeWorkspace: def __init__(self, backend: FakeBackend) -> None: self.backend = backend @@ -278,6 +291,30 @@ class FakeWorkspace: self.closed = True +@pytest.mark.asyncio +async def test_backend_timeout_prevents_direct_case_reset(tmp_path): + (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)), + plugins=("lme",), + workspace_builder=lambda _id, _image, _env: workspace, + ) + case = await factory.create_case("timed-out") + try: + with pytest.raises(SandboxCommandTimeoutError): + await case.run_build([JobRequest("build")]) + assert not case.reusable + assert case.actions[-1].exit_code == -1 + assert await case.read_build_log() == "build still running" + with pytest.raises(SandboxCommandError, match="no longer reusable"): + await case.reset_case("next") + finally: + await case.close() + assert workspace.closed + + @pytest.mark.asyncio async def test_source_candidate_installs_plugin_and_closes_container(tmp_path): (tmp_path / "reme").mkdir() @@ -306,6 +343,33 @@ async def test_source_candidate_installs_plugin_and_closes_container(tmp_path): assert workspace.closed +@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), + ) + plans = [ + CasePlan("timed-out", build_jobs=(JobRequest("build"),), export_build_log_to=tmp_path / "build.log"), + CasePlan("next", build_jobs=(JobRequest("build"),)), + ] + + results = await BenchmarkDockerPool(factory, concurrency=1).run(plans) + + assert "SandboxCommandTimeoutError" in results[0].error + assert results[0].build_log == "build still running" + assert (tmp_path / "build.log").read_text(encoding="utf-8") == "build still running" + assert results[1].success + assert timed_out_workspace.closed and replacement_workspace.closed + assert sum(command[:2] == ["rm", "-rf"] for command in timed_out_workspace.backend.commands) == 1 + + @pytest.mark.asyncio async def test_query_only_evaluation_export_omits_missing_build_log(tmp_path): (tmp_path / "reme").mkdir()