fix: isolate Python example evaluation in Docker

examples/01_code_pytest.py

@@ -17,19 +17,19 @@
 Loop type:      verify_revise.
 
 Run:
+    # First prepare the trusted pytest image described in examples/README.md.
     pip install 'loopgain[examples]'
     python examples/01_code_pytest.py
 """
 
 from __future__ import annotations
 
 import re
-import subprocess
-import sys
 import tempfile
 from pathlib import Path
 
 from loopgain import LoopGain
+from _sandbox import SandboxExecutionError, ensure_available, execute_pytest
 
 from _common import (
     call_claude,
@@ -99,14 +99,16 @@ def strip_code_fences(text: str) -> str:
 
 
 def run_pytest(workdir: Path) -> tuple[int, str]:
-    proc = subprocess.run(
-        [sys.executable, "-m", "pytest", "-q", "--tb=short", "--no-header"],
-        cwd=workdir, capture_output=True, text=True, timeout=30,
-    )
-    out = proc.stdout + proc.stderr
+    try:
+        returncode, out = execute_pytest(
+            (workdir / "solution.py").read_text(),
+            (workdir / "test_solution.py").read_text(),
+        )
+    except SandboxExecutionError as exc:
+        return 15, f"Sandbox candidate failure: {exc}"
     m = re.search(r"(\d+)\s+failed", out)
     failures = int(m.group(1)) if m else 0
-    if proc.returncode != 0 and failures == 0:
+    if returncode != 0 and failures == 0:
         failures = 15  # collection/import error → worst-case signal
     return failures, out
 
@@ -121,6 +123,7 @@ def one_iteration(client, workdir: Path, prev_code: str, prev_failures: str):
             f"\n\npytest reported these failures:\n```\n{prev_failures[-1500:]}\n```"
             f"\n\nReturn a fully corrected `solution.py`. Code only, no prose."
         )
+    ensure_available()  # Fail before spending on a candidate that cannot be isolated.
     code = strip_code_fences(call_claude(client, prompt))
     if not code:
         return 15, "", ""
@@ -154,17 +157,19 @@ def loopgain_run(client, workdir: Path):
 
 
 def main() -> None:
+    ensure_available()
     client = get_client()
-    workdir = Path(tempfile.mkdtemp(prefix="loopgain-ex01-"))
-    (workdir / "test_solution.py").write_text(TESTS)
-    print(f"Workdir: {workdir}")
-    print(f"Spec:    format_duration — 15 parametrized test cases.\n")
-
-    baseline_err, baseline_iters = baseline_run(client, workdir)
+    with tempfile.TemporaryDirectory(prefix="loopgain-ex01-") as temporary:
+        workdir = Path(temporary)
+        (workdir / "test_solution.py").write_text(TESTS)
+        print(f"Workdir: {workdir}")
+        print("Spec:    format_duration — 15 parametrized test cases.\n")
+        baseline_err, baseline_iters = baseline_run(client, workdir)
     # Fresh workdir for the LoopGain run so the comparison is apples-to-apples.
-    workdir = Path(tempfile.mkdtemp(prefix="loopgain-ex01-lg-"))
-    (workdir / "test_solution.py").write_text(TESTS)
-    lg = loopgain_run(client, workdir)
+    with tempfile.TemporaryDirectory(prefix="loopgain-ex01-lg-") as temporary:
+        workdir = Path(temporary)
+        (workdir / "test_solution.py").write_text(TESTS)
+        lg = loopgain_run(client, workdir)
 
     print_comparison(baseline_iters, baseline_err, lg)
     send_telemetry(lg, workload_id=WORKLOAD_ID, loop_type="verify_revise")

examples/README.md

@@ -42,6 +42,35 @@ Override the model via `LOOPGAIN_EXAMPLE_MODEL` (default `claude-haiku-4-5`).
 
 ---
 
+## Example 01 sandbox prerequisite
+
+Example 01 requires Docker on a POSIX host and a trusted local Python 3.12 image
+containing pytest. The default image is `loopgain-python-pytest:local`; select a
+different trusted image with `LOOPGAIN_SANDBOX_IMAGE`. Image preparation is a
+separate operator action. For example, use a Dockerfile with a trusted Python
+base and installed pytest, then tag it with the default name. Prepare or update
+that image before starting the example; the runtime never builds or pulls one.
+
+The example resolves the image to an immutable local ID and probes pytest with
+the actual isolation settings before creating a model client and before every
+model request. Missing Docker, image, pytest or failed cleanup stops the loop;
+there is no host execution fallback. Candidate/test files are staged read-only
+in a disposable non-root container with networking disabled, a read-only root,
+no host credentials or Docker socket, dropped capabilities, no-new-privileges,
+128 MiB memory, one CPU, 32 PIDs, 16 MiB `/tmp`, and bounded input/output and
+wall time. Containers and descendants are forcibly removed on every exit.
+External pytest plugins are disabled. Keep the image and Docker/kernel patched;
+containers are not a separate VM boundary. Python/runtime limits differ from
+older host execution, so do not compare historical measurements unchanged.
+
+Verify only the synthetic sandbox tests without provider/model calls:
+
+```bash
+LOOPGAIN_SANDBOX_INTEGRATION=1 python -m pytest tests/test_example_sandbox.py
+```
+
+The core `loopgain` package gains no runtime dependencies from this helper.
+
 ## Run
 
 ```bash

examples/_sandbox.py

@@ -0,0 +1,261 @@
+"""Fail-closed Docker execution for untrusted candidates (stdlib only).
+
+Images must already exist locally; this module never pulls or builds one.
+The image ID is resolved before execution so a mutable tag cannot change mid-run.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import selectors
+import shutil
+import signal
+import subprocess
+import tempfile
+import time
+import uuid
+from pathlib import Path
+
+MAX_INPUT_BYTES = 2 * 1024 * 1024
+MAX_OUTPUT_BYTES = 64 * 1024
+
+
+class SandboxUnavailable(RuntimeError):
+    """Infrastructure failure: stop rather than spend on an unexecutable trial."""
+
+
+class SandboxExecutionError(RuntimeError):
+    """Candidate exceeded a limit, crashed, or returned invalid output."""
+
+
+def _docker() -> str:
+    executable = shutil.which("docker")
+    if os.name != "posix" or not executable:
+        raise SandboxUnavailable(
+            "Candidate execution requires Docker on a POSIX host; no host fallback."
+        )
+    return executable
+
+
+def _docker_env() -> dict[str, str]:
+    # These configure the CLI only; the candidate environment is cleared by env -i.
+    return {
+        key: os.environ[key]
+        for key in ("PATH", "HOME", "DOCKER_CONFIG")
+        if key in os.environ
+    }
+
+
+def _local_image(docker: str) -> str:
+    image = os.environ.get("LOOPGAIN_SANDBOX_IMAGE", "loopgain-python-pytest:local")
+    try:
+        result = subprocess.run(
+            [docker, "image", "inspect", "--format", "{{.Id}}", image],
+            capture_output=True,
+            text=True,
+            timeout=10,
+            env=_docker_env(),
+            check=True,
+        )
+    except (OSError, subprocess.SubprocessError) as exc:
+        raise SandboxUnavailable(
+            "Docker or local Python sandbox image unavailable; no automatic pull."
+        ) from exc
+    image_id = result.stdout.strip()
+    if not re.fullmatch(r"sha256:[0-9a-f]{64}", image_id):
+        raise SandboxUnavailable("Docker did not return an immutable local image ID.")
+    return image_id
+
+
+def _command(docker: str, image: str, name: str, directory: Path) -> list[str]:
+    return [
+        docker,
+        "create",
+        "--pull=never",
+        "--name",
+        name,
+        "--network=none",
+        "--read-only",
+        "--user=65534:65534",
+        "--cap-drop=ALL",
+        "--security-opt=no-new-privileges",
+        "--pids-limit=32",
+        "--memory=128m",
+        "--memory-swap=128m",
+        "--cpus=1",
+        "--ulimit=cpu=20:20",
+        "--ulimit=nofile=64:64",
+        "--ulimit=core=0:0",
+        "--tmpfs=/tmp:rw,noexec,nosuid,nodev,size=16m,mode=1777",
+        "--mount",
+        f"type=bind,src={directory},dst=/input,readonly",
+        "--workdir=/tmp",
+        "--log-driver=none",
+        "--entrypoint=/usr/bin/env",
+        image,
+        "-i",
+        "PATH=/usr/local/bin:/usr/bin:/bin",
+        "HOME=/tmp",
+        "PYTEST_DISABLE_PLUGIN_AUTOLOAD=1",
+        "python3",
+        "-I",
+        "-B",
+        "-c",
+        (
+            "import runpy,sys; sys.stdin=open('/input/request.json'); "
+            "runpy.run_path('/input/worker.py',run_name='__main__')"
+        ),
+    ]
+
+
+def _run(image: str, worker: str, payload: dict, timeout: float, files=None) -> str:
+    encoded = json.dumps(payload).encode()
+    files = files or {}
+    if any(name not in {"solution.py", "test_solution.py"} for name in files):
+        raise ValueError("Unexpected sandbox fixture name")
+    if (
+        len(encoded) + sum(len(text.encode()) for text in files.values())
+        > MAX_INPUT_BYTES
+    ):
+        raise SandboxExecutionError("Sandbox input exceeds 2 MiB")
+    docker = _docker()
+    name = f"loopgain-example-candidate-{uuid.uuid4().hex}"
+    with tempfile.TemporaryDirectory(prefix="loopgain-example-candidate-") as temporary:
+        directory = Path(temporary)
+        # Only staged worker/request/fixtures are mounted; never repo, home or Docker socket.
+        directory.chmod(0o755)
+        for filename, contents in (
+            ("worker.py", worker.encode()),
+            ("request.json", encoded),
+        ):
+            path = directory / filename
+            path.write_bytes(contents)
+            path.chmod(0o444)
+        for filename, contents in files.items():
+            path = directory / filename
+            path.write_text(contents)
+            path.chmod(0o444)
+        proc = None
+        try:
+            subprocess.run(
+                _command(docker, image, name, directory),
+                capture_output=True,
+                timeout=15,
+                env=_docker_env(),
+                check=True,
+            )
+            proc = subprocess.Popen(
+                [docker, "start", "--attach", name],
+                stdin=subprocess.DEVNULL,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                env=_docker_env(),
+                start_new_session=True,
+            )
+            output = bytearray()
+            error = bytearray()
+            deadline = time.monotonic() + timeout
+            with selectors.DefaultSelector() as selector:
+                selector.register(proc.stdout, selectors.EVENT_READ, output)
+                selector.register(proc.stderr, selectors.EVENT_READ, error)
+                while selector.get_map():
+                    remaining = deadline - time.monotonic()
+                    if remaining <= 0:
+                        raise SandboxExecutionError("Sandbox wall-clock timeout")
+                    for key, _ in selector.select(min(remaining, 0.1)):
+                        chunk = os.read(key.fileobj.fileno(), 4096)
+                        if not chunk:
+                            selector.unregister(key.fileobj)
+                            continue
+                        key.data.extend(chunk)
+                        if len(output) + len(error) > MAX_OUTPUT_BYTES:
+                            raise SandboxExecutionError("Sandbox output limit exceeded")
+            proc.wait(timeout=max(0.01, deadline - time.monotonic()))
+            if proc.returncode:
+                raise SandboxExecutionError(
+                    f"Sandbox worker exited {proc.returncode}: {error[:200]!r}"
+                )
+            try:
+                return output.decode("utf8")
+            except UnicodeError as exc:
+                raise SandboxExecutionError("Invalid sandbox output encoding") from exc
+        except (OSError, subprocess.SubprocessError) as exc:
+            raise SandboxUnavailable("Docker sandbox process failed") from exc
+        finally:
+            # Remove the container explicitly, including every descendant process.
+            # Do not rely on terminating the attached docker CLI to stop a container.
+            if proc is not None and proc.poll() is None:
+                try:
+                    os.killpg(proc.pid, signal.SIGKILL)
+                except ProcessLookupError:
+                    pass
+            try:
+                cleanup = subprocess.run(
+                    [docker, "rm", "--force", name],
+                    capture_output=True,
+                    timeout=10,
+                    env=_docker_env(),
+                    check=False,
+                )
+                if cleanup.returncode and b"No such container" not in cleanup.stderr:
+                    raise SandboxUnavailable(
+                        "Could not confirm sandbox container cleanup; stop the run."
+                    )
+            except (OSError, subprocess.TimeoutExpired) as exc:
+                raise SandboxUnavailable(
+                    "Sandbox cleanup unavailable; stop the run."
+                ) from exc
+            finally:
+                if proc is not None:
+                    if proc.poll() is None:
+                        os.killpg(proc.pid, signal.SIGKILL)
+                    proc.wait(timeout=5)
+                    proc.stdout.close()
+                    proc.stderr.close()
+
+
+def ensure_available() -> str:
+    """Exercise the actual isolation flags before any model request is made."""
+    image = _local_image(_docker())
+    try:
+        result = _run(image, 'import pytest; print("sandbox-ready")', {}, 15)
+    except SandboxExecutionError as exc:
+        raise SandboxUnavailable(
+            "Sandbox preflight failed; do not start model requests."
+        ) from exc
+    if result.strip() != "sandbox-ready":
+        raise SandboxUnavailable("Unexpected sandbox preflight result")
+    return image
+
+
+_PYTEST_WORKER = """import contextlib,io,json,pytest
+output=io.StringIO()
+with contextlib.redirect_stdout(output), contextlib.redirect_stderr(output):
+    result=pytest.main(['/input/test_solution.py','-q','--tb=short','--no-header','-p','no:cacheprovider'])
+print(json.dumps({'returncode':int(result),'output':output.getvalue()}))
+"""
+
+
+def execute_pytest(solution: str, tests: str) -> tuple[int, str]:
+    """Run the fixed pytest fixture and candidate exclusively inside Docker."""
+    image = _local_image(_docker())
+    raw = _run(
+        image,
+        _PYTEST_WORKER,
+        {},
+        30,
+        {"solution.py": solution, "test_solution.py": tests},
+    )
+    try:
+        result = json.loads(raw)
+        if (
+            not isinstance(result, dict)
+            or not isinstance(result.get("returncode"), int)
+            or not isinstance(result.get("output"), str)
+        ):
+            raise ValueError("Invalid pytest result")
+        return result["returncode"], result["output"]
+    except (TypeError, ValueError) as exc:
+        raise SandboxExecutionError("Invalid sandbox result") from exc