Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions evalguard/cli/audit.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
)
from evalguard.rewardhack.mutations import ALL_MUTATION_STRATEGIES
from evalguard.rewardhack.tamper_detector import TamperDetector
from evalguard.sandbox.backends.docker import DockerBackend, DockerDiagnostic
from evalguard.sandbox.profiles import SandboxProfile

console = Console()
Expand Down Expand Up @@ -55,6 +56,26 @@ def run_audit(
workspace_base_dir=sb_profile.task_workspace_root,
)

# Pre-flight: emit actionable, color-coded guidance when an explicit isolation
# backend is requested but cannot come up on this host. This surfaces the
# structured DockerDiagnostic (probe reason + platform hints) instead of a
# bare warning, and hints at the `--backend auto` graceful fallback.
if backend == "docker":
diag: DockerDiagnostic = DockerBackend.diagnose()
if not diag.available:
console.print(
f"\n[bold red]Backend '{backend}' requested but unavailable on this host.[/bold red]"
)
console.print(f" [bold]Reason:[/bold] {diag.reason}")
if diag.hints:
console.print(" [bold yellow]Try one of the following:[/bold yellow]")
for hint in diag.hints:
console.print(f" - {hint}")
console.print(
"\n To run with automatic backend detection instead, retry with:\n"
" evalguard audit run --backend auto"
)

try:
audit_record: TaskAuditRecord = adapter.run_task_with_audit(
agent=command,
Expand Down
106 changes: 98 additions & 8 deletions evalguard/sandbox/backends/docker.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
import logging
import shutil
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path

from evalguard.sandbox.backends.base import CommandResult, SandboxBackend
Expand All @@ -13,6 +15,43 @@
logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class DockerDiagnostic:
"""Structured result of probing the local Docker installation.

``available`` is ``True`` only when both the CLI binary is on ``PATH``
and the daemon answers ``docker info`` successfully.
"""

available: bool
reason: str
hints: tuple[str, ...] = ()

@classmethod
def ok(cls) -> DockerDiagnostic:
"""Return a diagnostic representing a healthy Docker installation."""
return cls(available=True, reason="Docker daemon is reachable and healthy.")

@staticmethod
def platform_hints() -> tuple[str, ...]:
"""Return actionable instructions for the current platform."""
if sys.platform == "darwin":
return (
"Start Docker Desktop:\n open -a Docker",
"Wait for the whale icon to stop animating before re-running.",
)
if sys.platform.startswith("win"):
return (
'Start Docker Desktop from the Start menu or run:\n & "C:\\Program Files\\Docker\\Docker\\Docker Desktop.exe"',
"If Docker Desktop reports 'no daemon', relaunch it and retry.",
)
return (
"Start the Docker daemon (needs sudo unless you are in the docker group):\n sudo systemctl start docker",
"Enable auto-start on boot:\n sudo systemctl enable --now docker",
"After starting, verify with:\n docker info",
)


class DockerBackend(SandboxBackend):
"""Hermetic isolation backend using Docker containers."""

Expand All @@ -32,21 +71,72 @@ def backend_name(cls) -> str:
return "docker"

@classmethod
def is_available(cls) -> bool:
"""Check if docker binary is present and daemon is reachable."""
if not shutil.which("docker"):
return False
def diagnose(cls) -> DockerDiagnostic:
"""Probe the local Docker installation and return a structured reason.

This is the diagnostic core used by :meth:`is_available` and surfaced
by the CLI when a requested backend fails to come up. It distinguishes
three failure modes so callers can emit actionable guidance:
binary missing on ``PATH``, daemon unreachable, and probe timeout.
"""
docker_bin = shutil.which("docker")
if docker_bin is None:
hints = (
"Install the Docker CLI and add it to PATH:",
" https://docs.docker.com/get-docker/",
DockerDiagnostic.platform_hints()[0],
)
return DockerDiagnostic(
available=False,
reason=(
"Docker CLI not found on PATH. The 'docker' executable is required "
"before the daemon can be reached."
),
hints=hints,
)

try:
res = subprocess.run(
["docker", "info"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
stderr=subprocess.PIPE,
timeout=3,
check=False,
)
return res.returncode == 0
except Exception:
return False
except subprocess.TimeoutExpired:
return DockerDiagnostic(
available=False,
reason=(
"Docker daemon did not respond to 'docker info' within 3s. "
"The CLI is present but the daemon is likely not running."
),
hints=DockerDiagnostic.platform_hints(),
)
except Exception as exc: # pragma: no cover - defensive
return DockerDiagnostic(
available=False,
reason=f"Docker probe failed unexpectedly: {exc}",
hints=DockerDiagnostic.platform_hints(),
)

if res.returncode != 0:
stderr_tail = (res.stderr or b"").decode("utf-8", errors="replace").strip().splitlines()
detail = stderr_tail[-1] if stderr_tail else "docker info failed"
return DockerDiagnostic(
available=False,
reason=(
f"Docker daemon is unreachable ({detail}). The CLI is present "
"but the daemon is not answering 'docker info'."
),
hints=DockerDiagnostic.platform_hints(),
)

return DockerDiagnostic.ok()

@classmethod
def is_available(cls) -> bool:
"""Check if docker binary is present and daemon is reachable."""
return cls.diagnose().available

def setup(self) -> None:
"""Start an idle container with workspace volume mounts."""
Expand Down
42 changes: 42 additions & 0 deletions tests/unit/test_sandbox/test_backends_extended.py
Original file line number Diff line number Diff line change
Expand Up @@ -135,3 +135,45 @@ def test_gvisor_backend_mocked() -> None:

backend.teardown()
assert backend._is_running is False


def test_docker_diagnose_reports_missing_cli() -> None:
with patch("evalguard.sandbox.backends.docker.shutil.which", return_value=None):
diagnostic = DockerBackend.diagnose()
assert DockerBackend.is_available() is False

assert diagnostic.available is False
assert "CLI not found on PATH" in diagnostic.reason
assert any("docs.docker.com/get-docker" in hint for hint in diagnostic.hints)


def test_docker_diagnose_includes_daemon_error() -> None:
failed_result = MagicMock(
returncode=1, stdout="", stderr=b"permission denied\nCannot connect to the Docker daemon"
)
with (
patch("evalguard.sandbox.backends.docker.shutil.which", return_value="/usr/bin/docker"),
patch("evalguard.sandbox.backends.docker.subprocess.run", return_value=failed_result),
):
diagnostic = DockerBackend.diagnose()

assert diagnostic.available is False
assert "Cannot connect to the Docker daemon" in diagnostic.reason
assert diagnostic.hints


def test_docker_diagnose_handles_probe_timeout() -> None:
import subprocess

with (
patch("evalguard.sandbox.backends.docker.shutil.which", return_value="/usr/bin/docker"),
patch(
"evalguard.sandbox.backends.docker.subprocess.run",
side_effect=subprocess.TimeoutExpired(cmd=["docker", "info"], timeout=3),
),
):
diagnostic = DockerBackend.diagnose()

assert diagnostic.available is False
assert "within 3s" in diagnostic.reason
assert diagnostic.hints
Loading