Stop interrupted benchmark containers before collecting diagnostics

deploy/benchmark_docker/README.md

@@ -93,9 +93,17 @@ wrapper but no manifest are rejected; to reuse one, extract it and supply its
 `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`. 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.
+build, the pool also returns its text in `CaseResult.build_log`. `command_timeout`
+(default 1800 seconds) bounds the entire shell command, including worker startup,
+jobs, and Application shutdown. On timeout, cancellation, or a shell backend
+failure, the sandbox force-stops the container without candidate or plugin
+cooperation, attempts to cache the build log for up to 5 seconds, and removes the
+container before propagating the error. Docker stop and removal each have a
+30-second timeout; cleanup failures include the container ID in exception notes.
+Only the cached build log remains available after removal. The pool creates a
+replacement when more cases need a worker.
+Custom `workspace_builder` implementations must provide an async `stop()` that
+force-stops execution while retaining files for diagnostics, in addition to `close()`.
 Replacement requires successful cleanup of the old worker. Cleanup failures
 are reported in `CaseResult.error` and make `success` false for cases that used
 that worker; failures for an unused worker apply to the whole run. Cancellation

deploy/benchmark_docker/direct_docker_workspace.py

@@ -15,6 +15,8 @@ from typing import Any
 from aiodocker.exceptions import DockerError
 from agentscope.workspace import DockerWorkspace
 
+DOCKER_CLEANUP_TIMEOUT = 30.0
+
 
 class DirectDockerWorkspace(DockerWorkspace):
     """Use the supplied image directly and skip the unused MCP gateway."""
@@ -28,6 +30,7 @@ class DirectDockerWorkspace(DockerWorkspace):
     def __init__(self, **kwargs: Any) -> None:
         super().__init__(**kwargs)
         self._lifecycle_lock = asyncio.Lock()
+        self._stopped = False
 
     async def _build_or_reuse_image(self) -> None:
         """Select the caller-provided image instead of deriving a gateway image."""
@@ -56,13 +59,36 @@ class DirectDockerWorkspace(DockerWorkspace):
                     exc.add_note(f"workspace close failed: {type(close_exc).__name__}: {close_exc}")
                 raise
             self.is_alive = True
+            self._stopped = False
+
+    async def stop(self) -> None:
+        """Kill all container processes without asking the candidate to shut down."""
+        async with self._lifecycle_lock:
+            self.is_alive = False
+            if self._container is None or self._stopped:
+                return
+            async with asyncio.timeout(DOCKER_CLEANUP_TIMEOUT):
+                try:
+                    await self._container.kill(signal="SIGKILL")
+                    await self._container.wait(condition="not-running")
+                except DockerError as exc:
+                    if exc.status != 404:
+                        # Docker returns 409 for an already stopped container.
+                        if exc.status != 409 or (await self._container.show())["State"]["Running"]:
+                            raise
+                self._stopped = True
 
     async def _teardown_backend(self) -> None:
         """Remove the container, preserving handles until cleanup is confirmed."""
         self.is_alive = False
         if self._container is not None:
             # Preserve AgentScope's bind-mount ownership repair on native Linux.
-            if self.host_workdir is not None and sys.platform == "linux" and self._backend is not None:
+            if (
+                not self._stopped
+                and self.host_workdir is not None
+                and sys.platform == "linux"
+                and self._backend is not None
+            ):
                 try:
                     await self._backend.exec_shell(
                         ["chown", "-R", f"{os.getuid()}:{os.getgid()}", self.workdir],
@@ -71,16 +97,16 @@ class DirectDockerWorkspace(DockerWorkspace):
                 except Exception:
                     pass
             try:
-                # Force deletion also stops a running container; a separate kill
-                # would add a failure point without proving that removal succeeded.
-                await self._container.delete(force=True)
+                # Also stop processes if an earlier explicit stop failed, and
+                # retain handles unless Docker confirms removal.
+                await asyncio.wait_for(self._container.delete(force=True), timeout=DOCKER_CLEANUP_TIMEOUT)
             except DockerError as exc:
                 if exc.status != 404:
                     raise
             self._container = None
             self._backend = None
         if self._client is not None:
-            await self._client.close()
+            await asyncio.wait_for(self._client.close(), timeout=DOCKER_CLEANUP_TIMEOUT)
             self._client = None
         self._backend = None
 

deploy/benchmark_docker/docker_sandbox.py

@@ -30,6 +30,8 @@ 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}$")
 _CASE_ARCHIVE_FORMAT = "reme-case-v1"
+_CLEANUP_TIMEOUT = 30.0
+_LOG_TIMEOUT = 5.0
 _PRIVATE_EVALUATION_FIELDS = frozenset(
     {"query", "question", "golden_answer", "question_type", "rubric", "judge_arguments"},
 )
@@ -68,6 +70,10 @@ class Workspace(Protocol):
         """Close the workspace."""
         raise NotImplementedError
 
+    async def stop(self) -> None:
+        """Force-stop container processes while keeping files available."""
+        raise NotImplementedError
+
     def get_backend(self) -> Backend:
         """Return the execution backend."""
         raise NotImplementedError
@@ -81,7 +87,7 @@ class SandboxCommandError(RuntimeError):
 
 
 class SandboxCommandTimeoutError(SandboxCommandError):
-    """A command timed out and may still be running inside the container."""
+    """A command timed out; cleanup failures are attached as exception notes."""
 
 
 def _validate_case_id(case_id: str) -> None:
@@ -379,6 +385,7 @@ class DockerReMeSandbox:
         self._closed = False
         self._reusable = True
         self._operation_lock = asyncio.Lock()
+        self._build_log: bytes | None = None
 
     @property
     def reusable(self) -> bool:
@@ -524,13 +531,15 @@ class DockerReMeSandbox:
             raise SandboxCommandError("sandbox container is no longer reusable")
         started = _utc_now()
         try:
-            result = await self.backend.exec_shell(command, cwd=cwd, timeout=self.command_timeout)
+            async with asyncio.timeout(self.command_timeout):
+                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
-        except (Exception, asyncio.CancelledError):
+            error = SandboxCommandTimeoutError(f"sandbox action {name!r} timed out")
+            await cleanup_on_error([(f"container {self.container_id!r}", self._abort())], error)
+            raise error from exc
+        except (Exception, asyncio.CancelledError) as exc:
             # A backend failure may occur after Docker started the command.
-            self._reusable = False
+            await cleanup_on_error([(f"container {self.container_id!r}", self._abort())], exc)
             raise
         self.actions.append(
             ActionRecord(
@@ -543,10 +552,36 @@ class DockerReMeSandbox:
             ),
         )
         if result.exit_code == -1:
-            self._reusable = False
-            raise SandboxCommandTimeoutError(f"sandbox action {name!r} timed out; container must be destroyed")
+            error = SandboxCommandTimeoutError(f"sandbox action {name!r} timed out")
+            await cleanup_on_error([(f"container {self.container_id!r}", self._abort())], error)
+            raise error
         return result
 
+    async def _abort(self) -> None:
+        """Retire an uncertain command while its caller holds the operation lock."""
+        self._reusable = False
+        error = None
+        try:
+            await asyncio.wait_for(self.workspace.stop(), timeout=_CLEANUP_TIMEOUT)
+            # Read files only after the candidate has stopped. Diagnostics must
+            # never delay termination or prevent container removal.
+            try:
+                assert self.backend is not None
+                self._build_log = await asyncio.wait_for(
+                    self.backend.read_file(f"{CASE_ROOT}/build_log/build.log"),
+                    timeout=_LOG_TIMEOUT,
+                )
+            except Exception:
+                pass
+        except BaseException as exc:
+            error = exc
+            raise
+        finally:
+            if error is None:
+                await self._close_workspace()
+            else:
+                await cleanup_on_error([("container removal", self._close_workspace())], error)
+
     async def _exec_checked(self, name: str, command: list[str], *, cwd: str | None = None) -> Any:
         result = await self._exec(name, command, cwd=cwd)
         if not result.ok():
@@ -776,6 +811,10 @@ class DockerReMeSandbox:
             return (await self._read_build_log()).decode("utf-8", errors="replace")
 
     async def _read_build_log(self) -> bytes:
+        if self._build_log is not None:
+            return self._build_log
+        if not self.reusable:
+            raise SandboxCommandError("build log unavailable after interrupted command")
         return await self._require_artifact(f"{CASE_ROOT}/build_log/build.log", "build log")
 
     async def _require_artifact(self, path: str, description: str) -> bytes:
@@ -1078,10 +1117,14 @@ class DockerReMeSandbox:
     async def close(self) -> None:
         """Destroy the case container. Export before calling this method."""
         async with self._operation_lock:
-            if not self._closed:
-                self._reusable = False
-                await self.workspace.close()
-                self._closed = True
+            await self._close_workspace()
+
+    async def _close_workspace(self) -> None:
+        """Close without acquiring the operation lock a second time."""
+        if not self._closed:
+            self._reusable = False
+            await asyncio.wait_for(self.workspace.close(), timeout=_CLEANUP_TIMEOUT)
+            self._closed = True
 
     async def __aenter__(self) -> "DockerReMeSandbox":
         return self

deploy/benchmark_docker/pool.py

@@ -140,7 +140,7 @@ class BenchmarkDockerPool:
                             try:
                                 result.build = await worker.run_build(list(plan.build_jobs))
                             except Exception as exc:
-                                result.error = f"{type(exc).__name__}: {exc}"
+                                result.error = format_error(exc)
                                 await self._retain_failed_build_log(worker, plan, result)
                                 return result
                             if not result.build.get("success"):
@@ -162,14 +162,14 @@ class BenchmarkDockerPool:
                             await worker.export_evaluation(plan.export_evaluation_to)
                         return result
                     except Exception as exc:  # Keep other cases running and release this worker.
-                        result.error = f"{type(exc).__name__}: {exc}"
+                        result.error = format_error(exc)
                         return result
                     finally:
                         if worker.reusable:
                             available.put_nowait(worker)
                         else:
-                            # A timed-out command may still be running in the container.
-                            # Retire it before allowing another case into this slot.
+                            # Confirm cleanup (or retry a failed removal) before
+                            # allowing another case into this slot.
                             closed = False
                             try:
                                 await worker.close()

tests/integration/test_benchmark_docker.py

@@ -102,6 +102,45 @@ def _query(query_id: str) -> EvaluationQuery:
     )
 
 
+@pytest.mark.asyncio
+async def test_real_docker_timeout_kills_blocked_thread_pool(base_image):
+    """A host deadline must terminate a worker blocked in executor shutdown."""
+    from deploy.benchmark_docker.direct_docker_workspace import DirectDockerWorkspace
+    from deploy.benchmark_docker.docker_sandbox import DockerReMeSandbox, SandboxCommandTimeoutError
+
+    container_id = f"timeout-{uuid.uuid4().hex[:12]}"
+    workspace = DirectDockerWorkspace(workspace_id=container_id, base_image=base_image)
+    case = DockerReMeSandbox(
+        case_id="timeout",
+        container_id=container_id,
+        candidate=ImageCandidate(base_image),
+        workspace=workspace,
+        config="unused",
+        plugins=(),
+        command_timeout=2,
+        environment_names=[],
+    )
+    try:
+        await workspace.initialize()
+        case.backend = workspace.get_backend()
+        # Reproduce the shutdown condition without importing ReMe or a plugin.
+        script = (
+            "import threading; from pathlib import Path; from concurrent.futures import ThreadPoolExecutor; "
+            "pool = ThreadPoolExecutor(1); pool.submit(threading.Event().wait); "
+            "log = Path('/workspace/case/build_log/build.log'); log.parent.mkdir(parents=True); "
+            "log.write_text('waiting for thread pool shutdown'); pool.shutdown(wait=True)"
+        )
+        # pylint: disable=protected-access
+        async with case._operation_lock:
+            with pytest.raises(SandboxCommandTimeoutError):
+                await asyncio.wait_for(case._exec("blocked-close", ["python3", "-c", script]), timeout=75)
+        assert not _container_exists(f"as_ws_{container_id}")
+        assert not case.reusable
+        assert await case.read_build_log() == "waiting for thread pool shutdown"
+    finally:
+        await case.close()
+
+
 @pytest.mark.asyncio
 async def test_real_docker_source_pool_workspace_reuse(base_image, tmp_path):
     """Run build, query, workspace reuse, and query-only export in a real container."""

tests/unit/test_benchmark_docker.py

@@ -803,6 +803,7 @@ class FakeWorkspace:
     def __init__(self, backend: FakeBackend) -> None:
         self.backend = backend
         self.closed = False
+        self.stopped = False
 
     async def initialize(self) -> None:
         pass
@@ -813,6 +814,9 @@ class FakeWorkspace:
     async def close(self) -> None:
         self.closed = True
 
+    async def stop(self) -> None:
+        self.stopped = True
+
 
 def _fake_docker_factory(tmp_path, workspace_builder, *, install_projects=(".",)):
     (tmp_path / "reme").mkdir()
@@ -833,6 +837,7 @@ async def test_backend_timeout_prevents_direct_case_reset(tmp_path):
         with pytest.raises(SandboxCommandTimeoutError):
             await case.run_build([JobRequest("build")])
         assert not case.reusable
+        assert workspace.stopped and workspace.closed
         assert case.actions[-1].exit_code == -1
         assert await case.read_build_log() == "build still running"
         with pytest.raises(SandboxCommandError, match="no longer reusable"):
@@ -863,6 +868,7 @@ async def test_external_timeout_prevents_case_reset(tmp_path):
         with pytest.raises(TimeoutError):
             await asyncio.wait_for(build, timeout=0.01)
         assert not case.reusable
+        assert workspace.stopped and workspace.closed
         with pytest.raises(SandboxCommandError, match="no longer reusable"):
             await case.reset_case("next")
         inbox_count = sum("/inbox/" in path for path in workspace.backend.files)
@@ -874,6 +880,98 @@ async def test_external_timeout_prevents_case_reset(tmp_path):
     assert workspace.closed
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("failure", ["deadline", "backend-timeout", "cancel"])
+async def test_interrupted_worker_stops_before_reading_logs_and_removal(tmp_path, failure):
+    events = []
+    started = asyncio.Event()
+    stopping = asyncio.Event()
+    release_stop = asyncio.Event()
+
+    class BlockingBackend(FakeBackend):
+        async def exec_shell(self, command, *, cwd=None, timeout=None):
+            if "/workspace/harness/worker.py" in command:
+                self.files["/workspace/case/build_log/build.log"] = b"closing application"
+                started.set()
+                if failure == "backend-timeout":
+                    raise TimeoutError("backend timed out")
+                await asyncio.Event().wait()  # Includes a worker stuck in Application.close().
+            return await super().exec_shell(command, cwd=cwd, timeout=timeout)
+
+        async def read_file(self, path):
+            events.append("read")
+            assert workspace.stopped
+            return await super().read_file(path)
+
+    class OrderedWorkspace(FakeWorkspace):
+        async def stop(self):
+            stopping.set()
+            await release_stop.wait()
+            events.append("stop")
+            await super().stop()
+
+        async def close(self):
+            events.append("close")
+            await super().close()
+
+    workspace = OrderedWorkspace(BlockingBackend())
+    factory = _fake_docker_factory(tmp_path, lambda _id, _image, _env: workspace)
+    case = await factory.create_case("interrupted")
+    case.command_timeout = 0.01 if failure == "deadline" else 10
+    task = asyncio.create_task(case.run_build([JobRequest("build")]))
+    await asyncio.wait_for(started.wait(), timeout=2)
+    if failure == "cancel":
+        task.cancel()
+    await asyncio.wait_for(stopping.wait(), timeout=2)
+    if failure == "cancel":
+        task.cancel()  # Repeated cancellation must not interrupt container cleanup.
+    release_stop.set()
+    expected = asyncio.CancelledError if failure == "cancel" else SandboxCommandTimeoutError
+    with pytest.raises(expected):
+        await asyncio.wait_for(task, timeout=2)
+    assert events == ["stop", "read", "close"]
+    assert workspace.closed and not case.reusable
+    assert await case.read_build_log() == "closing application"
+    await case.close()
+    assert events == ["stop", "read", "close"]
+
+
+@pytest.mark.asyncio
+async def test_failed_stop_still_removes_container_and_preserves_timeout(tmp_path):
+    class BrokenStopWorkspace(FakeWorkspace):
+        async def stop(self):
+            raise RuntimeError("kill failed")
+
+    workspace = BrokenStopWorkspace(TimedOutBackend())
+    factory = _fake_docker_factory(tmp_path, lambda _id, _image, _env: workspace)
+    case = await factory.create_case("failed-stop")
+    with pytest.raises(SandboxCommandTimeoutError) as failure:
+        await case.run_build([JobRequest("build")])
+    assert workspace.closed
+    assert any("kill failed" in note for note in failure.value.__notes__)
+    with pytest.raises(SandboxCommandError, match="log unavailable"):
+        await case.read_build_log()
+
+
+@pytest.mark.asyncio
+async def test_blocked_diagnostics_cannot_prevent_removal(tmp_path, monkeypatch):
+    from deploy.benchmark_docker import docker_sandbox
+
+    monkeypatch.setattr(docker_sandbox, "_LOG_TIMEOUT", 0.01)
+
+    class BlockedLogBackend(TimedOutBackend):
+        async def read_file(self, path):
+            assert workspace.stopped
+            await asyncio.Event().wait()
+
+    workspace = FakeWorkspace(BlockedLogBackend())
+    factory = _fake_docker_factory(tmp_path, lambda _id, _image, _env: workspace)
+    case = await factory.create_case("blocked-log")
+    with pytest.raises(SandboxCommandTimeoutError):
+        await asyncio.wait_for(case.run_build([JobRequest("build")]), timeout=2)
+    assert workspace.closed
+
+
 @pytest.mark.asyncio
 async def test_cancelled_request_upload_prevents_case_reset(tmp_path):
     started = asyncio.Event()
@@ -1340,7 +1438,11 @@ async def test_pool_retries_strict_delete_without_replacing_uncertain_container(
     from deploy.benchmark_docker.direct_docker_workspace import DirectDockerWorkspace
 
     workspaces = []
-    container = SimpleNamespace(delete=AsyncMock(side_effect=[RuntimeError("delete failed"), None]))
+    container = SimpleNamespace(
+        kill=AsyncMock(),
+        wait=AsyncMock(),
+        delete=AsyncMock(side_effect=[RuntimeError("delete failed"), RuntimeError("delete failed"), None]),
+    )
     client = SimpleNamespace(close=AsyncMock())
 
     class StrictWorkspace(DirectDockerWorkspace):
@@ -1361,6 +1463,6 @@ async def test_pool_retries_strict_delete_without_replacing_uncertain_container(
     assert "worker close failed" in results[0].error
     assert "delete failed" in results[1].error
     assert len(workspaces) == 1
-    assert container.delete.await_count == 2
+    assert container.delete.await_count == 3
     assert workspaces[0]._container is workspaces[0]._client is None
     client.close.assert_awaited_once()

tests/unit/test_benchmark_docker_workspace.py

@@ -14,13 +14,49 @@ from deploy.benchmark_docker.direct_docker_workspace import DirectDockerWorkspac
 
 def make_workspace():
     workspace = DirectDockerWorkspace(workspace_id="cleanup-test")
-    workspace._container = SimpleNamespace(delete=AsyncMock())
+    workspace._container = SimpleNamespace(kill=AsyncMock(), wait=AsyncMock(), delete=AsyncMock())
     workspace._client = SimpleNamespace(close=AsyncMock())
     workspace._backend = SimpleNamespace(exec_shell=AsyncMock())
     workspace.is_alive = True
     return workspace
 
 
+@pytest.mark.asyncio
+async def test_stop_kills_and_confirms_exit_before_removal():
+    workspace = make_workspace()
+    container = workspace._container
+    await workspace.stop()
+    container.kill.assert_awaited_once_with(signal="SIGKILL")
+    container.wait.assert_awaited_once_with(condition="not-running")
+    container.delete.assert_not_awaited()
+    assert workspace._backend is not None
+    assert not workspace.is_alive
+    await workspace.stop()
+    assert container.kill.await_count == 1
+    await workspace.close()
+    container.delete.assert_awaited_once_with(force=True)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("operation", ["kill", "delete"])
+async def test_docker_cleanup_timeout_retains_handles(monkeypatch, operation):
+    from deploy.benchmark_docker import direct_docker_workspace
+
+    monkeypatch.setattr(direct_docker_workspace, "DOCKER_CLEANUP_TIMEOUT", 0.01)
+    workspace = make_workspace()
+    container = workspace._container
+
+    async def blocked(**_kwargs):
+        await asyncio.Event().wait()
+
+    getattr(container, operation).side_effect = blocked
+    with pytest.raises(TimeoutError):
+        await (workspace.stop() if operation == "kill" else workspace.close())
+    assert workspace._container is container
+    getattr(container, operation).side_effect = None
+    await workspace.close()
+
+
 @pytest.mark.asyncio
 async def test_delete_failure_retains_handles_and_can_be_retried():
     workspace = make_workspace()