arena/docker_backend.py

"""Docker-based sandbox backend for isolated execution.

Uses ``subprocess`` to manage a single container, avoiding extra dependencies.
"""

from __future__ import annotations

import json
import shlex
import subprocess
import tempfile
from pathlib import Path
from typing import Any


class DockerSandboxBackend:
    """A sandbox backend backed by a Docker container for process isolation."""

    def __init__(
        self,
        image: str = "python:3.11-slim",
        timeout_seconds: int = 20,
    ) -> None:
        self._image = image
        self._timeout = timeout_seconds
        self._container_id: str | None = None
        self._workspace_dir = "/workspace"
        self._start_container()

    def _start_container(self) -> None:
        result = subprocess.run(
            [
                "docker", "run", "-d", "--rm",
                "-w", "/workspace",
                self._image,
                "tail", "-f", "/dev/null",
            ],
            capture_output=True, text=True, timeout=30,
            check=False,
        )
        if result.returncode != 0:
            stderr = result.stderr.strip()
            if "docker" in stderr.lower() and ("not found" in stderr.lower() or "daemon" in stderr.lower()):
                raise RuntimeError(
                    "Docker is not available. Install Docker or set DEEPAGENT_SANDBOX_PROVIDER=local."
                )
            raise RuntimeError(f"Failed to start Docker container: {stderr}")
        self._container_id = result.stdout.strip()
        self._exec("mkdir", "-p", self._workspace_dir)

    def write(self, path: str, content: str) -> Any:
        return self._copy_to(path, content.encode("utf-8"))

    def read(self, path: str) -> Any:
        data = self._copy_from(path)
        content = data.decode("utf-8") if isinstance(data, bytes) else str(data)
        return type("ReadResult", (), {"error": None, "file_data": {"content": content}})()

    def ls(self, path: str) -> Any:
        output = self._exec_output("find", path.rstrip("/"), "-maxdepth", "1", "-printf", "%p\t%y\n")
        entries = []
        for line in output.splitlines():
            if not line.strip():
                continue
            parts = line.split("\t")
            if len(parts) < 2:
                continue
            filepath, kind = parts[0], parts[1]
            if filepath == path.rstrip("/"):
                continue
            entries.append({"path": filepath, "is_dir": kind == "d"})
        return type("LsResult", (), {"error": None, "entries": entries})()

    def execute(self, command: str, timeout: int | None = None) -> Any:
        t = timeout or self._timeout
        result = subprocess.run(
            ["docker", "exec", self._container_id, "sh", "-c", command],
            capture_output=True, text=True, timeout=t,
            check=False,
        )
        output = result.stdout
        if result.stderr:
            output = f"{output}\n{result.stderr}"
        return type("ExecResult", (), {"error": None, "output": output, "exit_code": result.returncode})()

    def glob(self, pattern: str) -> Any:
        find_pattern = "/".join(
            part.replace("**", ".").replace("*", "[^/]*") if part != "**" else part
            for part in pattern.split("/")
        )
        output = self._exec_output("find", "/", "-path", pattern, "-not", "-type", "d")
        matches = []
        for line in output.splitlines():
            line = line.strip()
            if line:
                matches.append({"path": line, "is_dir": False})
        return type("GlobResult", (), {"error": None, "matches": matches})()

    def download_files(self, paths: list[str]) -> list[Any]:
        results = []
        for path in paths:
            try:
                data = self._copy_from(path)
                results.append(type("DownloadResult", (), {"error": None, "path": path, "content": data})())
            except Exception as exc:
                results.append(type("DownloadResult", (), {"error": str(exc), "path": path, "content": b""})())
        return results

    def upload_files(self, files: list[tuple[str, bytes]]) -> list[Any]:
        results = []
        for path, content in files:
            try:
                self._copy_to(path, content)
                results.append(type("UploadResult", (), {"error": None, "path": path})())
            except Exception as exc:
                results.append(type("UploadResult", (), {"error": str(exc), "path": path})())
        return results

    def stop(self) -> None:
        if self._container_id:
            subprocess.run(
                ["docker", "rm", "-f", self._container_id],
                capture_output=True, timeout=10, check=False,
            )
            self._container_id = None

    def delete(self) -> None:
        self.stop()

    def _exec_output(self, *cmd: str) -> str:
        result = subprocess.run(
            ["docker", "exec", self._container_id, *cmd],
            capture_output=True, text=True, timeout=self._timeout,
            check=False,
        )
        return result.stdout

    def _exec(self, *cmd: str) -> None:
        result = subprocess.run(
            ["docker", "exec", self._container_id, *cmd],
            capture_output=True, text=True, timeout=self._timeout,
            check=False,
        )
        if result.returncode != 0:
            raise RuntimeError(f"Docker exec failed: {result.stderr.strip()}")

    def _copy_to(self, container_path: str, content: bytes) -> Any:
        with tempfile.NamedTemporaryFile(delete=False) as tmp:
            tmp.write(content)
            tmp_path = tmp.name
        try:
            result = subprocess.run(
                ["docker", "cp", tmp_path, f"{self._container_id}:{shlex.quote(container_path)}"],
                capture_output=True, text=True, timeout=self._timeout,
                check=False,
            )
        finally:
            Path(tmp_path).unlink(missing_ok=True)
        if result.returncode != 0:
            raise RuntimeError(f"docker cp failed: {result.stderr.strip()}")
        return type("WriteResult", (), {"error": None})()

    def _copy_from(self, container_path: str) -> bytes:
        with tempfile.NamedTemporaryFile(delete=False) as tmp:
            tmp_path = tmp.name
        try:
            result = subprocess.run(
                ["docker", "cp", f"{self._container_id}:{shlex.quote(container_path)}", tmp_path],
                capture_output=True, text=True, timeout=self._timeout,
                check=False,
            )
            if result.returncode != 0:
                raise RuntimeError(f"docker cp failed: {result.stderr.strip()}")
            return Path(tmp_path).read_bytes()
        finally:
            Path(tmp_path).unlink(missing_ok=True)

    def _workspace_dir(self) -> str:
        return self._workspace_dir