Fix benchmark worker reuse after Docker command timeout

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()`,

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:

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))

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()