From b6c9c9a99f0bc65a0bb5b561d85d48aecfc8eede Mon Sep 17 00:00:00 2001
From: merlerm <matteo.merler@gmail.com>
Date: Sat, 19 Sep 2026 16:45:31 +0200
Subject: [PATCH 1/3] Add a credential-isolating broker for validated model
 inference

---
 src/robocode/utils/backends/__init__.py |  17 +-
 src/robocode/utils/model_broker.py      | 494 ++++++++++++++++++++++++
 tests/utils/test_backends.py            |   5 +
 tests/utils/test_model_broker.py        | 331 ++++++++++++++++
 4 files changed, 843 insertions(+), 4 deletions(-)
 create mode 100644 src/robocode/utils/model_broker.py
 create mode 100644 tests/utils/test_model_broker.py

diff --git a/src/robocode/utils/backends/__init__.py b/src/robocode/utils/backends/__init__.py
index 2c884e40..b2683acc 100644
--- a/src/robocode/utils/backends/__init__.py
+++ b/src/robocode/utils/backends/__init__.py
@@ -6,7 +6,9 @@
 To add a new provider, add an entry to :data:`PROVIDERS` below. The
 ``domains`` list is used by the Docker firewall whitelist, and
 ``api_key_env`` is the environment variable forwarded into Docker
-containers for authentication.
+containers for authentication. Registering a provider here does not enable it
+in the Apptainer broker: that transport requires an explicitly validated API
+protocol in ``model_broker.py``.
 """
 
 from dataclasses import dataclass, field
@@ -75,19 +77,26 @@ class ProviderInfo:
     api_key_env: str = ""
 
 
+# Fixed provider hosts shared by Docker's domain list and the inference broker.
+# These constants are destinations, not interchangeable network policies.
+OPENAI_API_HOST = "api.openai.com"
+ANTHROPIC_API_HOST = "api.anthropic.com"
+CODEX_CHATGPT_HOST = "chatgpt.com"
+
+
 # ---- Provider registry ----
 # Add new providers here. The key is the provider prefix used in model
 # strings (e.g. "openai" in "openai/gpt-4o").
 PROVIDERS: dict[str, ProviderInfo] = {
     "openai": ProviderInfo(
-        domains=["api.openai.com"],
+        domains=[OPENAI_API_HOST],
         api_key_env="OPENAI_API_KEY",
     ),
     "codex": ProviderInfo(
-        domains=["api.openai.com", "chatgpt.com", "ab.chatgpt.com"],
+        domains=[OPENAI_API_HOST, CODEX_CHATGPT_HOST, "ab.chatgpt.com"],
     ),
     "anthropic": ProviderInfo(
-        domains=["api.anthropic.com"],
+        domains=[ANTHROPIC_API_HOST],
         api_key_env="ANTHROPIC_API_KEY",
     ),
     "google": ProviderInfo(
diff --git a/src/robocode/utils/model_broker.py b/src/robocode/utils/model_broker.py
new file mode 100644
index 00000000..39f411c9
--- /dev/null
+++ b/src/robocode/utils/model_broker.py
@@ -0,0 +1,494 @@
+"""Host-owned inference broker for network-disconnected Apptainer agents.
+
+Only this process holds provider credentials. The Unix socket exposes a small
+HTTP API, not CONNECT or an arbitrary destination proxy. Request bodies are
+validated before forwarding to fixed HTTPS endpoints; redirects are never followed.
+"""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import http.client
+import json
+import os
+import socketserver
+import ssl
+import threading
+from collections.abc import Iterator
+from contextlib import contextmanager
+from dataclasses import dataclass, field
+from http.server import BaseHTTPRequestHandler
+from pathlib import Path
+from typing import Any
+
+from robocode.utils.backends import (
+    ANTHROPIC_API_HOST,
+    CODEX_CHATGPT_HOST,
+    OPENAI_API_HOST,
+)
+from robocode.utils.claude_auth import host_claude_config_dir
+from robocode.utils.codex_auth import host_codex_home
+
+MAX_BODY = 32 * 1024 * 1024
+MODEL_PORT = 18080
+BROKER_DIR = "/run/robocode-broker"
+
+
+class BrokerPolicyError(ValueError):
+    """A request is outside the explicitly supported inference protocol."""
+
+
+def _no_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+    result: dict[str, Any] = {}
+    for key, value in pairs:
+        if key in result:
+            raise BrokerPolicyError("duplicate JSON key")
+        result[key] = value
+    return result
+
+
+def _local_content(value: Any) -> None:
+    """Reject provider-side URL/file retrieval in actual content (not prose)."""
+    if isinstance(value, list):
+        for child in value:
+            _local_content(child)
+    elif isinstance(value, dict):
+        kind = value.get("type", "")
+        if isinstance(kind, str) and (
+            kind.startswith(
+                (
+                    "web_",
+                    "mcp_",
+                    "server_",
+                    "computer_",
+                    "code_interpreter",
+                    "file_search",
+                )
+            )
+            or kind
+            in {
+                "tool_search_call",
+                "tool_search_output",
+                "item_reference",
+                "input_file",
+                "document",
+            }
+        ):
+            raise BrokerPolicyError("server-side content operation")
+        for key, child in value.items():
+            if key in {"url", "image_url"}:
+                if not isinstance(child, str):
+                    raise BrokerPolicyError("remote content URL")
+                prefix, sep, encoded = child.partition(",")
+                if not sep or prefix not in {
+                    "data:image/png;base64",
+                    "data:image/jpeg;base64",
+                    "data:image/webp;base64",
+                    "data:image/gif;base64",
+                }:
+                    raise BrokerPolicyError("only inline raster images are supported")
+                try:
+                    base64.b64decode(encoded, validate=True)
+                except (ValueError, binascii.Error) as exc:
+                    raise BrokerPolicyError("invalid inline image") from exc
+            if key in {"file_url", "file_id", "container_id", "server_url"}:
+                raise BrokerPolicyError("remote content reference")
+            if key == "source" and isinstance(child, dict):
+                if child.get("type") not in {"base64", "text"}:
+                    raise BrokerPolicyError("remote content source")
+                if child.get("type") == "base64" and child.get("media_type") not in {
+                    "image/png",
+                    "image/jpeg",
+                    "image/webp",
+                    "image/gif",
+                }:
+                    raise BrokerPolicyError("only inline raster images are supported")
+            _local_content(child)
+
+
+def _schemas(value: Any) -> None:
+    if isinstance(value, dict):
+        for key, child in value.items():
+            if key == "$ref" and (
+                not isinstance(child, str) or not child.startswith("#")
+            ):
+                raise BrokerPolicyError("remote schema reference")
+            _schemas(child)
+    elif isinstance(value, list):
+        for child in value:
+            _schemas(child)
+
+
+def _openai_tools(tools: Any) -> None:
+    if not isinstance(tools, list):
+        raise BrokerPolicyError("tools must be a list")
+    for tool in tools:
+        if not isinstance(tool, dict):
+            raise BrokerPolicyError("invalid tool")
+        kind = tool.get("type")
+        if kind == "namespace":
+            if set(tool) - {"type", "name", "description", "tools"}:
+                raise BrokerPolicyError("unsupported namespace fields")
+            _openai_tools(tool.get("tools"))
+        elif kind == "function":
+            if set(tool) - {
+                "type",
+                "name",
+                "description",
+                "parameters",
+                "strict",
+                "defer_loading",
+            }:
+                raise BrokerPolicyError("unsupported function fields")
+            _schemas(tool)
+        elif kind == "custom":
+            if set(tool) - {"type", "name", "description", "format", "defer_loading"}:
+                raise BrokerPolicyError("unsupported custom tool fields")
+        else:
+            raise BrokerPolicyError("server-side tools are forbidden")
+
+
+def validate_request(protocol: str, path: str, raw: bytes) -> dict[str, Any]:
+    """Parse a bounded request and fail closed on unsupported API operations."""
+    if len(raw) > MAX_BODY:
+        raise BrokerPolicyError("body too large")
+    try:
+        data = json.loads(raw, object_pairs_hook=_no_duplicates)
+    except (ValueError, RecursionError) as exc:
+        raise BrokerPolicyError("invalid JSON") from exc
+    if not isinstance(data, dict):
+        raise BrokerPolicyError("body must be an object")
+    if protocol == "responses":
+        if path not in {"/v1/responses", "/v1/responses/compact"}:
+            raise BrokerPolicyError("endpoint forbidden")
+        allowed = {
+            "model",
+            "instructions",
+            "input",
+            "tools",
+            "tool_choice",
+            "parallel_tool_calls",
+            "stream",
+            "store",
+            "reasoning",
+            "text",
+            "include",
+            "prompt_cache_key",
+            "service_tier",
+            "max_output_tokens",
+            "temperature",
+            "top_p",
+            "metadata",
+            "truncation",
+            "prompt_cache_retention",
+            "safety_identifier",
+            "client_metadata",
+        }
+        if set(data) - allowed:
+            raise BrokerPolicyError(
+                "unsupported fields: " + ",".join(sorted(set(data) - allowed))
+            )
+        data.pop("client_metadata", None)  # do not grant authority via client hints
+        _openai_tools(data.get("tools", []))
+        choice = data.get("tool_choice", "auto")
+        if not (
+            choice in ("auto", "none", "required")
+            if isinstance(choice, str)
+            else isinstance(choice, dict)
+            and choice.get("type") in {"function", "custom"}
+        ):
+            raise BrokerPolicyError("unsupported tool choice")
+        if any(
+            item != "reasoning.encrypted_content" for item in data.get("include", [])
+        ):
+            raise BrokerPolicyError("unsupported include")
+        _local_content(data.get("input"))
+        _schemas(data.get("text"))
+    elif protocol == "messages":
+        if path not in {
+            "/v1/messages",
+            "/v1/messages?beta=true",
+            "/v1/messages/count_tokens",
+            "/v1/messages/count_tokens?beta=true",
+        }:
+            raise BrokerPolicyError("endpoint forbidden")
+        allowed = {
+            "model",
+            "messages",
+            "system",
+            "tools",
+            "tool_choice",
+            "max_tokens",
+            "stream",
+            "temperature",
+            "top_p",
+            "top_k",
+            "thinking",
+            "output_config",
+            "metadata",
+            "stop_sequences",
+            "service_tier",
+            "context_management",
+        }
+        if set(data) - allowed:
+            raise BrokerPolicyError(
+                "unsupported fields: " + ",".join(sorted(set(data) - allowed))
+            )
+        tools = data.get("tools", [])
+        if not isinstance(tools, list):
+            raise BrokerPolicyError("tools must be a list")
+        for tool in tools:
+            if not isinstance(tool, dict) or tool.get("type", "custom") != "custom":
+                raise BrokerPolicyError("server-side tools are forbidden")
+            if set(tool) - {
+                "type",
+                "name",
+                "description",
+                "input_schema",
+                "cache_control",
+                "defer_loading",
+                "strict",
+                "input_examples",
+            }:
+                raise BrokerPolicyError("unsupported custom tool fields")
+            _schemas(tool)
+        _local_content(data.get("messages"))
+        _local_content(data.get("system"))
+        context = data.get("context_management", {})
+        if not isinstance(context, dict) or set(context) - {"edits"}:
+            raise BrokerPolicyError("unsupported context management")
+        for edit in context.get("edits", []):
+            if not isinstance(edit, dict) or edit.get("type") not in {
+                "clear_thinking_20251015",
+                "clear_tool_uses_20250919",
+            }:
+                raise BrokerPolicyError("unsupported context operation")
+    else:
+        raise BrokerPolicyError("unsupported protocol")
+    if not isinstance(data.get("model"), str) or not data["model"]:
+        raise BrokerPolicyError("model required")
+    return data
+
+
+@dataclass(frozen=True)
+class BrokerUpstream:
+    """Trusted upstream selection; never populated from a container request."""
+
+    protocol: str
+    host: str
+    base_path: str
+    headers: dict[str, str] = field(repr=False)
+    chatgpt: bool = False
+
+
+def load_broker_upstream(backend: str) -> BrokerUpstream:
+    """Load credentials on the host without copying them into the container."""
+    if backend == "codex":
+        key = os.environ.get("CODEX_API_KEY")
+        if key:
+            return BrokerUpstream(
+                "responses", OPENAI_API_HOST, "/v1", {"Authorization": "Bearer " + key}
+            )
+        auth = json.loads((host_codex_home() / "auth.json").read_text(encoding="utf-8"))
+        if auth.get("auth_mode") == "chatgpt":
+            tokens = auth["tokens"]
+            return BrokerUpstream(
+                "responses",
+                CODEX_CHATGPT_HOST,
+                "/backend-api/codex",
+                {
+                    "Authorization": "Bearer " + tokens["access_token"],
+                    "ChatGPT-Account-ID": tokens["account_id"],
+                    "OpenAI-Beta": "responses=experimental",
+                    "originator": "codex_cli_rs",
+                },
+                chatgpt=True,
+            )
+        key = auth.get("OPENAI_API_KEY") or os.environ.get("OPENAI_API_KEY")
+        if key:
+            return BrokerUpstream(
+                "responses", OPENAI_API_HOST, "/v1", {"Authorization": "Bearer " + key}
+            )
+        raise RuntimeError("No supported Codex credentials for the isolated broker")
+    if backend == "claude":
+        token = os.environ.get("CLAUDE_CODE_OAUTH_TOKEN")
+        key = os.environ.get("ANTHROPIC_API_KEY")
+        if not token and not key:
+            creds = json.loads(
+                (host_claude_config_dir() / ".credentials.json").read_text(
+                    encoding="utf-8"
+                )
+            )
+            token = creds.get("claudeAiOauth", {}).get("accessToken")
+        headers = {"anthropic-version": "2023-06-01"}
+        if token:
+            headers.update(
+                {
+                    "Authorization": "Bearer " + token,
+                    "anthropic-beta": "oauth-2025-04-20,context-management-2025-06-27",
+                }
+            )
+        elif key:
+            headers["x-api-key"] = key
+        else:
+            raise RuntimeError("No Claude credentials for the isolated broker")
+        return BrokerUpstream("messages", ANTHROPIC_API_HOST, "/v1", headers)
+    raise RuntimeError(
+        f"Isolated Apptainer model transport does not support {backend!r}; "
+        "refusing host networking"
+    )
+
+
+class _Server(socketserver.ThreadingMixIn, socketserver.UnixStreamServer):
+    daemon_threads = True
+    block_on_close = False
+
+    def __init__(self, path: Path, provider: BrokerUpstream, log_path: Path):
+        self.provider = provider
+        self.log_path = log_path
+        self.log_lock = threading.Lock()
+        super().__init__(str(path), _Handler)
+
+    def record(self, path: str, status: int, reason: str) -> None:
+        """Retain decisions, never credentials or request/response bodies."""
+        with self.log_lock, self.log_path.open("a", encoding="utf-8") as log:
+            log.write(
+                json.dumps({"path": path[:200], "status": status, "reason": reason})
+                + "\n"
+            )
+
+
+class _Handler(BaseHTTPRequestHandler):
+    server: _Server
+    close_connection: bool
+    protocol_version = "HTTP/1.0"  # one framed request per connection
+
+    def setup(self) -> None:
+        self.request.settimeout(120)
+        super().setup()
+
+    def log_message(  # pylint: disable=redefined-builtin
+        self, format: str, *args: Any
+    ) -> None:
+        """Suppress the standard HTTP logger; use body-free policy audit records."""
+
+    def _reject(self, status: int, reason: str) -> None:
+        self.server.record(self.path, status, reason)
+        body = json.dumps(
+            {
+                "error": {
+                    "message": "Robocode broker: " + reason,
+                    "type": "broker_policy",
+                }
+            }
+        ).encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+        self.close_connection = True
+
+    def do_CONNECT(self) -> None:  # pylint: disable=invalid-name
+        """Never expose a TCP tunnel."""
+        self._reject(403, "CONNECT forbidden")
+
+    def do_GET(self) -> None:  # pylint: disable=invalid-name
+        """Do not expose discovery, search, retrieval, or websocket upgrades."""
+        self._reject(403, "GET and websocket upgrades forbidden")
+
+    def do_POST(self) -> None:  # pylint: disable=invalid-name
+        """Validate, authenticate on the host, and stream a fixed upstream."""
+        try:
+            lengths = self.headers.get_all("Content-Length", [])
+            if (
+                len(lengths) != 1
+                or not lengths[0].isascii()
+                or not lengths[0].isdigit()
+            ):
+                raise BrokerPolicyError("single Content-Length required")
+            length = int(lengths[0])
+            if not 0 < length <= MAX_BODY:
+                raise BrokerPolicyError("invalid body size")
+            if self.headers.get("Transfer-Encoding") or self.headers.get("Upgrade"):
+                raise BrokerPolicyError("transfer encoding and upgrades forbidden")
+            if self.headers.get("Content-Encoding", "identity") != "identity":
+                raise BrokerPolicyError("compressed requests unsupported")
+            raw = self.rfile.read(length)
+            if len(raw) != length:
+                raise BrokerPolicyError("incomplete body")
+            provider = self.server.provider
+            data = validate_request(provider.protocol, self.path, raw)
+            # The ChatGPT Codex endpoint only accepts streaming, unstored inference.
+            if provider.protocol == "responses":
+                data["store"] = False
+            if provider.chatgpt:
+                if self.path == "/v1/responses":
+                    data["stream"] = True
+            raw = json.dumps(data, allow_nan=False).encode()
+        except (
+            BrokerPolicyError,
+            ValueError,
+            RecursionError,
+            TypeError,
+            AttributeError,
+        ) as exc:
+            self._reject(403, str(exc))
+            return
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "text/event-stream",
+            **provider.headers,
+        }
+        conn = http.client.HTTPSConnection(
+            provider.host, timeout=120, context=ssl.create_default_context()
+        )
+        started = False
+        try:
+            conn.request(
+                "POST",
+                provider.base_path + self.path[len("/v1") :],
+                body=raw,
+                headers=headers,
+            )
+            response = conn.getresponse()
+            if 300 <= response.status < 400:
+                self._reject(502, "upstream redirect forbidden")
+                return
+            self.server.record(self.path, response.status, "forwarded")
+            self.send_response(response.status)
+            self.send_header(
+                "Content-Type", response.getheader("Content-Type", "application/json")
+            )
+            self.send_header("Connection", "close")
+            self.end_headers()
+            started = True
+            while chunk := response.read1(65536):
+                self.wfile.write(chunk)
+                self.wfile.flush()
+        except (OSError, http.client.HTTPException):
+            if not started:
+                self._reject(502, "upstream unavailable")
+        finally:
+            conn.close()
+            self.close_connection = True
+
+
+@contextmanager
+def model_broker(
+    directory: Path, provider: BrokerUpstream, log_path: Path
+) -> Iterator[Path]:
+    """Expose only the per-run Unix endpoint; the private log stays outside binds."""
+    log_path.parent.mkdir(parents=True, exist_ok=True)
+    path = directory / "model.sock"
+    with _Server(path, provider, log_path) as server:
+        path.chmod(0o600)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            yield path
+        finally:
+            server.shutdown()
+            thread.join(timeout=5)
+            path.unlink(missing_ok=True)
diff --git a/tests/utils/test_backends.py b/tests/utils/test_backends.py
index 3338e8d5..30d16979 100644
--- a/tests/utils/test_backends.py
+++ b/tests/utils/test_backends.py
@@ -818,6 +818,11 @@ def test_provider_from_model_no_slash(self) -> None:
 
     def test_firewall_domains_for_known_providers(self) -> None:
         """Known providers return their API domains."""
+        assert firewall_domains_for_provider("codex") == [
+            "api.openai.com",
+            "chatgpt.com",
+            "ab.chatgpt.com",
+        ]
         assert firewall_domains_for_provider("openai") == ["api.openai.com"]
         assert firewall_domains_for_provider("anthropic") == ["api.anthropic.com"]
         assert firewall_domains_for_provider("google") == [
diff --git a/tests/utils/test_model_broker.py b/tests/utils/test_model_broker.py
new file mode 100644
index 00000000..b776a288
--- /dev/null
+++ b/tests/utils/test_model_broker.py
@@ -0,0 +1,331 @@
+"""Adversarial policy and HTTP framing checks for the trusted inference broker."""
+
+# pylint: disable=redefined-outer-name
+
+import http.client
+import io
+import json
+import socket
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from robocode.utils.model_broker import (
+    BrokerPolicyError,
+    BrokerUpstream,
+    load_broker_upstream,
+    model_broker,
+    validate_request,
+)
+
+
+def _body(**extra: Any) -> bytes:
+    return json.dumps({"model": "test-model", "input": "hello", **extra}).encode()
+
+
+@pytest.mark.parametrize(
+    "tool",
+    [
+        "web_search",
+        "web_search_preview",
+        "file_search",
+        "mcp",
+        "code_interpreter",
+        "computer_use_preview",
+        "image_generation",
+        "tool_search",
+        "future_server_tool",
+    ],
+)
+def test_hosted_tools_rejected(tool):
+    """New or known provider-executed tools never reach the upstream API."""
+    with pytest.raises(BrokerPolicyError):
+        validate_request("responses", "/v1/responses", _body(tools=[{"type": tool}]))
+
+
+def test_nested_namespace_cannot_hide_hosted_tool():
+    """Namespaces contain only client-executed function/custom declarations."""
+    tools = [{"type": "namespace", "name": "a", "tools": [{"type": "web_search"}]}]
+    with pytest.raises(BrokerPolicyError):
+        validate_request("responses", "/v1/responses", _body(tools=tools))
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "https://example.com/v1/responses",
+        "//example.com/v1/responses",
+        "/v1/models",
+        "/v1/responses?url=https://example.com",
+        "/v1/responses/../search",
+        "/v1/responses%2f..%2fsearch",
+        "/v1/files",
+        "/v1/responses/123",
+    ],
+)
+def test_only_exact_inference_paths_allowed(path):
+    """Absolute URLs, redirects, retrieval endpoints and encoded paths fail."""
+    with pytest.raises(BrokerPolicyError):
+        validate_request("responses", path, _body())
+
+
+@pytest.mark.parametrize(
+    "content",
+    [
+        {"type": "input_image", "image_url": "https://example.com/image.png"},
+        {"type": "input_file", "file_url": "https://example.com/f"},
+        {"type": "input_file", "file_id": "file-123"},
+        {"source": {"type": "url", "url": "https://example.com"}},
+    ],
+)
+def test_remote_content_retrieval_rejected(content):
+    """An inference endpoint cannot be used as a URL fetcher."""
+    with pytest.raises(BrokerPolicyError):
+        validate_request("responses", "/v1/responses", _body(input=[content]))
+
+
+def test_inline_image_and_literal_url_text_allowed():
+    """Locally supplied pixels and ordinary URL mentions are not network fetches."""
+    data = _body(
+        input=[
+            {
+                "role": "user",
+                "content": [
+                    {"type": "input_text", "text": "https://example.com"},
+                    {"type": "input_image", "image_url": "data:image/png;base64,AAAA"},
+                ],
+            }
+        ],
+        tools=[{"type": "function", "name": "shell", "parameters": {"type": "object"}}],
+    )
+    assert validate_request("responses", "/v1/responses", data)["model"] == "test-model"
+
+
+@pytest.mark.parametrize(
+    "extra",
+    [
+        {"tools": [{"type": "web_search_20250305", "name": "web_search"}]},
+        {"tools": [{"type": "web_fetch_20250910", "name": "web_fetch"}]},
+        {"mcp_servers": [{"url": "https://example.com"}]},
+        {"container": {"skills": []}},
+        {
+            "messages": [
+                {"content": [{"source": {"type": "url", "url": "https://example.com"}}]}
+            ]
+        },
+    ],
+)
+def test_claude_server_capabilities_rejected(extra):
+    """Only client-executed Claude tools and inline message content are allowed."""
+    data = json.dumps({"model": "claude", "messages": [], **extra}).encode()
+    with pytest.raises(BrokerPolicyError):
+        validate_request("messages", "/v1/messages?beta=true", data)
+
+
+def test_duplicate_keys_and_unknown_fields_fail_closed():
+    """Neither ambiguous JSON nor future API switches can silently expand access."""
+    for data in (
+        b'{"model":"a","tools":[],"tools":[{"type":"web_search"}]}',
+        _body(new_network_feature=True),
+    ):
+        with pytest.raises(BrokerPolicyError):
+            validate_request("responses", "/v1/responses", data)
+
+
+class _UnixHTTP(http.client.HTTPConnection):
+    def connect(self):
+        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        self.sock.settimeout(5)
+        self.sock.connect(self.host)
+
+
+class _Response(io.BytesIO):
+    status = 200
+
+    def getheader(self, name, default=None):
+        """Return the minimal streaming response headers."""
+        return "text/event-stream" if name == "Content-Type" else default
+
+
+@pytest.fixture
+def broker(tmp_path: Path, monkeypatch):
+    """A real Unix HTTP server with a fake, observable HTTPS upstream."""
+    requests = []
+    response = _Response(b'data: {"ok":true}\n\n')
+
+    class Connection:
+        """Record upstream operations without performing network I/O."""
+
+        def __init__(self, host, **kwargs):
+            assert host == "api.openai.com"
+            assert kwargs["context"].check_hostname
+
+        def request(self, method, path, body, headers):
+            """Record exactly what would be sent to the provider."""
+            requests.append((method, path, body, headers))
+
+        def getresponse(self):
+            """Return the fixture response."""
+            return response
+
+        def close(self):
+            """No real upstream connection needs closing."""
+
+    monkeypatch.setattr(
+        "robocode.utils.model_broker.http.client.HTTPSConnection", Connection
+    )
+    provider = BrokerUpstream(
+        "responses", "api.openai.com", "/v1", {"Authorization": "Bearer host-secret"}
+    )
+    with model_broker(tmp_path, provider, tmp_path / "audit.jsonl") as path:
+        yield path, requests, response
+
+
+@pytest.mark.parametrize(
+    "method,path,body,headers",
+    [
+        ("CONNECT", "example.com:443", None, {}),
+        ("GET", "/v1/responses", None, {"Upgrade": "websocket"}),
+        ("POST", "/v1/responses", _body(), {"Content-Encoding": "gzip"}),
+        ("POST", "/v1/responses", _body(), {"Transfer-Encoding": "chunked"}),
+        ("POST", "/v1/responses", _body(tools=[{"type": "web_search"}]), {}),
+    ],
+)
+def test_http_denials_never_open_upstream(broker, method, path, body, headers):
+    """Framing, tunneling and tool bypasses are rejected before HTTPS starts."""
+    address, requests, _ = broker
+    conn = _UnixHTTP(str(address))
+    conn.request(method, path, body=body, headers=headers)
+    assert conn.getresponse().status == 403
+    conn.close()
+    assert not requests
+
+
+def test_valid_stream_uses_fixed_host_path_and_host_credentials(broker):
+    """An agent's Host, Authorization and forwarding headers carry no authority."""
+    address, requests, _ = broker
+    conn = _UnixHTTP(str(address))
+    conn.request(
+        "POST",
+        "/v1/responses",
+        body=_body(),
+        headers={
+            "Host": "evil.invalid",
+            "Authorization": "Bearer attacker",
+            "X-Forwarded-Host": "evil.invalid",
+        },
+    )
+    response = conn.getresponse()
+    assert response.status == 200
+    assert response.read() == b'data: {"ok":true}\n\n'
+    conn.close()
+    assert requests[0][0:2] == ("POST", "/v1/responses")
+    assert requests[0][3]["Authorization"] == "Bearer host-secret"
+    assert "X-Forwarded-Host" not in requests[0][3]
+    assert "host-secret" not in (address.parent / "audit.jsonl").read_text(
+        encoding="utf-8"
+    )
+
+
+def test_upstream_redirect_is_never_followed(broker):
+    """Even a redirect from the approved provider cannot change destination."""
+    address, requests, upstream = broker
+    upstream.status = 302
+    conn = _UnixHTTP(str(address))
+    conn.request("POST", "/v1/responses", body=_body())
+    assert conn.getresponse().status == 502
+    conn.close()
+    assert len(requests) == 1
+
+
+def test_duplicate_content_length_rejected(broker):
+    """Conflicting framing cannot smuggle a second request to the provider."""
+    address, requests, _ = broker
+    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
+        sock.connect(str(address))
+        sock.sendall(
+            b"POST /v1/responses HTTP/1.1\r\nHost: local\r\n"
+            b"Content-Length: 2\r\nContent-Length: 3\r\n\r\n{}"
+        )
+        assert b"403" in sock.recv(4096)
+    assert not requests
+
+
+@pytest.mark.parametrize(
+    "content",
+    [
+        {"type": "input_image", "image_url": "data:image/svg+xml;base64,AAAA"},
+        {
+            "type": "image",
+            "source": {
+                "type": "base64",
+                "media_type": "application/pdf",
+                "data": "AAAA",
+            },
+        },
+        {"type": "input_file", "file_data": "AAAA"},
+        {"type": "mcp_approval_response", "approval_request_id": "x", "approve": True},
+    ],
+)
+def test_indirect_document_and_server_operation_channels_rejected(content):
+    """Opaque document formats and inherited hosted operations stay unsupported."""
+    with pytest.raises(BrokerPolicyError):
+        validate_request("responses", "/v1/responses", _body(input=[content]))
+
+
+def test_cannot_inherit_tools_from_stored_response():
+    """A caller cannot continue an unrelated stored response with hosted tools."""
+    with pytest.raises(BrokerPolicyError):
+        validate_request(
+            "responses", "/v1/responses", _body(previous_response_id="resp_other")
+        )
+
+
+@pytest.mark.parametrize(
+    "backend,env_name,expected_host,header,prefix",
+    [
+        ("codex", "CODEX_API_KEY", "api.openai.com", "Authorization", "Bearer "),
+        (
+            "claude",
+            "CLAUDE_CODE_OAUTH_TOKEN",
+            "api.anthropic.com",
+            "Authorization",
+            "Bearer ",
+        ),
+        ("claude", "ANTHROPIC_API_KEY", "api.anthropic.com", "x-api-key", ""),
+    ],
+)
+def test_credentials_belong_to_host_upstream(
+    monkeypatch, backend, env_name, expected_host, header, prefix
+):
+    """The broker resolves host auth without manufacturing a container auth mount."""
+    for key in ("CODEX_API_KEY", "CLAUDE_CODE_OAUTH_TOKEN", "ANTHROPIC_API_KEY"):
+        monkeypatch.delenv(key, raising=False)
+    monkeypatch.setenv(env_name, "host-only-test-secret")
+    upstream = load_broker_upstream(backend)
+    assert upstream.host == expected_host
+    assert upstream.headers[header] == prefix + "host-only-test-secret"
+    assert "host-only-test-secret" not in repr(upstream)
+
+
+def test_codex_session_auth_stays_on_host(tmp_path, monkeypatch):
+    """ChatGPT account routing is loaded from host auth, not a container hint."""
+    monkeypatch.delenv("CODEX_API_KEY", raising=False)
+    monkeypatch.setattr("robocode.utils.model_broker.host_codex_home", lambda: tmp_path)
+    (tmp_path / "auth.json").write_text(
+        json.dumps(
+            {
+                "auth_mode": "chatgpt",
+                "tokens": {
+                    "access_token": "host-session-secret",
+                    "account_id": "trusted-account",
+                },
+            }
+        ),
+        encoding="utf-8",
+    )
+    upstream = load_broker_upstream("codex")
+    assert (upstream.host, upstream.base_path) == ("chatgpt.com", "/backend-api/codex")
+    assert upstream.headers["ChatGPT-Account-ID"] == "trusted-account"
+    assert "host-session-secret" not in repr(upstream)

From fc5f8efe06d991e4960b2e50ecbd8106ee736015 Mon Sep 17 00:00:00 2001
From: merlerm <matteo.merler@gmail.com>
Date: Sat, 19 Sep 2026 16:46:17 +0200
Subject: [PATCH 2/3] Remove the importable MCP environment from strict
 blackbox images

---
 docker/Dockerfile.strict-blackbox        |  37 ++--
 docs/blackbox.md                         |   8 +-
 src/robocode/mcp/__init__.py             |  10 +-
 src/robocode/mcp/strict_server.py        | 224 +++++++++++++++++++++++
 src/robocode/utils/apptainer_sandbox.py  |   3 +-
 src/robocode/utils/backends/claude.py    |   1 +
 src/robocode/utils/backends/codex.py     |   1 +
 src/robocode/utils/backends/opencode.py  |   1 +
 src/robocode/utils/docker_sandbox.py     |   3 +-
 src/robocode/utils/strict_blackbox.py    |   1 -
 tests/utils/test_apptainer_sandbox.py    |   8 +-
 tests/utils/test_strict_blackbox.py      |  25 ++-
 tests/utils/test_strict_render_server.py |  93 ++++++++++
 13 files changed, 370 insertions(+), 45 deletions(-)
 create mode 100644 src/robocode/mcp/strict_server.py
 create mode 100644 tests/utils/test_strict_render_server.py

diff --git a/docker/Dockerfile.strict-blackbox b/docker/Dockerfile.strict-blackbox
index 3ab8b5e4..578280bd 100644
--- a/docker/Dockerfile.strict-blackbox
+++ b/docker/Dockerfile.strict-blackbox
@@ -3,8 +3,8 @@
 # Deliberately never copies project metadata, environment/KinDER/simulator code,
 # or robotics/geometry packages. Generated programs use only Python's standard
 # library plus pinned NumPy/SciPy; the frozen program is later checked on the host
-# against exactly that allowlist (src/robocode/utils/strict_blackbox.py). A separate
-# Python environment contains only the generic MCP-to-env-server render proxy.
+# against exactly that allowlist (src/robocode/utils/strict_blackbox.py). Rendering
+# uses the same interpreter, without installing any project or MCP package.
 FROM node:22
 
 ARG CLAUDE_CODE_VERSION=latest
@@ -40,6 +40,12 @@ RUN python3.11 -m venv /opt/robocode-strict \
     && /opt/robocode-strict/bin/pip install --no-cache-dir \
         numpy==1.26.4 scipy==1.14.0
 
+# Remove installer and base-image Python packages after the numerical wheels
+# are installed. An agent can add any readable package directory to sys.path.
+RUN /opt/robocode-strict/bin/python -m pip uninstall -y pip setuptools \
+    && rm -rf /usr/lib/python3/dist-packages/* \
+        /usr/local/lib/python3.11/dist-packages/* /usr/share/python-wheels/*
+
 RUN mkdir -p /usr/local/share/npm-global \
     && chown -R node:node /usr/local/share
 ENV NPM_CONFIG_PREFIX=/usr/local/share/npm-global
@@ -48,24 +54,17 @@ RUN npm install -g @anthropic-ai/claude-code@${CLAUDE_CODE_VERSION} \
     && npm install -g opencode-ai@${OPENCODE_VERSION} \
     && npm install -g @openai/codex@${CODEX_VERSION}
 
-# Keep MCP infrastructure out of the Python environment used by generated
-# programs. This separate interpreter hosts only the generic blackbox render
-# proxy; a .pth shares the strict environment's packages so render_policy
-# can execute any approach that obeys the scoring import allowlist.
-RUN python3.11 -m venv /opt/robocode-mcp \
-    && /opt/robocode-mcp/bin/pip install --no-cache-dir "mcp==1.29.0" \
-    && echo "/opt/robocode-strict/lib/python3.11/site-packages" \
-        > /opt/robocode-mcp/lib/python3.11/site-packages/strict-blackbox.pth
+# node:22 also carries Python build/debug helpers outside site-packages. They
+# are not used by the installed agent CLIs and must not become import backdoors.
+RUN rm -rf /usr/local/lib/node_modules/npm/node_modules/node-gyp/gyp/pylib \
+    /usr/share/glib-2.0/codegen /usr/share/gcc/python \
+    /usr/share/doc/subversion/examples /usr/share/doc/libsvn1/examples \
+    /usr/share/python3/debpython
 
-# Install only the generic, environment-independent blackbox MCP proxy. No
-# environment, simulator, primitive, rendering, or approach source enters the
-# image; actual pixels are produced by the host env server.
-COPY src/robocode/__init__.py \
-    /opt/robocode-mcp/lib/python3.11/site-packages/robocode/__init__.py
-COPY src/robocode/mcp/__init__.py src/robocode/mcp/server.py \
-    /opt/robocode-mcp/lib/python3.11/site-packages/robocode/mcp/
-COPY src/robocode/utils/__init__.py src/robocode/utils/env_client.py \
-    /opt/robocode-mcp/lib/python3.11/site-packages/robocode/utils/
+# No second interpreter or project package: virtualenvs are not access controls.
+# The render protocol uses stdlib plus the same NumPy client as agent scripts.
+COPY src/robocode/mcp/strict_server.py src/robocode/utils/env_client.py \
+    /opt/robocode-render/
 
 COPY docker/init-firewall.sh /usr/local/bin/init-firewall.sh
 COPY docker/strict-blackbox-entrypoint.sh /usr/local/bin/entrypoint.sh
diff --git a/docs/blackbox.md b/docs/blackbox.md
index 55ff213b..c3995085 100644
--- a/docs/blackbox.md
+++ b/docs/blackbox.md
@@ -289,9 +289,11 @@ named `robocode-tools`) have two implementations, selected at MCP-config time by
   and steps the env over the protocol), then renders each visited state via
   `render_state`. The host therefore never executes `approach.py`.
 
-Strict blackbox uses that same proxy protocol, but runs the MCP server with a
-separate `/opt/robocode-mcp/bin/python`; the generated-code interpreter remains
-dependency-clean. Its host connection permits `render_state` but still rejects raw
+Strict blackbox uses that same proxy protocol, with a standalone stdlib MCP
+HTTP server at `/opt/robocode-render/strict_server.py`. Both rendering and agent
+scripts use `/opt/robocode-strict/bin/python`; no project or MCP framework package
+is installed. A second virtualenv would not prevent agents from importing its
+packages by changing `sys.path`. Its host connection permits `render_state` but still rejects raw
 `get_state` snapshots and all other helpers. Consequently, strict `render_policy`
 renders the observations returned by `reset`/`step` rather than requesting hidden
 state snapshots.
diff --git a/src/robocode/mcp/__init__.py b/src/robocode/mcp/__init__.py
index 389047e5..fb30c9bc 100644
--- a/src/robocode/mcp/__init__.py
+++ b/src/robocode/mcp/__init__.py
@@ -341,6 +341,7 @@ def setup_mcp_config(
     blackbox: bool = False,
     transport: str = "stdio",
     port: int = MCP_HTTP_PORT,
+    strict_blackbox: bool = False,
 ) -> Path:
     """Write MCP server config into ``sandbox_dir/.mcp/``.
 
@@ -375,8 +376,15 @@ def setup_mcp_config(
         # <container_sandbox>/.mcp/env_config.json; the env_spaces.json the
         # approach wrote sits at the sandbox root.
         env_spaces_path = Path(env_config_path).parent.parent / "env_spaces.json"
+        entrypoint = (
+            "/opt/robocode-render/strict_server.py"
+            if strict_blackbox
+            else "-m robocode.mcp.server"
+        )
+        if strict_blackbox and transport != "http":
+            raise ValueError("Strict rendering requires HTTP transport")
         server_cmd = (
-            f"{python_cmd} -m robocode.mcp.server"
+            f"{python_cmd} {entrypoint}"
             f" --env-spaces {env_spaces_path}"
             f" --tools {','.join(tool_names)}"
             f" --log-file {log_file_path}"
diff --git a/src/robocode/mcp/strict_server.py b/src/robocode/mcp/strict_server.py
new file mode 100644
index 00000000..6fdfd616
--- /dev/null
+++ b/src/robocode/mcp/strict_server.py
@@ -0,0 +1,224 @@
+"""Standalone strict render tools using only stdlib and the numerical env client.
+
+Copied into the strict image as a plain script, never as a robocode package.
+The small stateless MCP HTTP surface deliberately has no framework environment
+that an agent or rendered policy could import. Policies execute in this same
+isolated container. The host receives only the existing environment protocol.
+
+Transport: https://modelcontextprotocol.io/specification/2025-06-18/basic/transports
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlsplit
+
+VERSIONS = ("2025-03-26", "2025-06-18", "2025-11-25")
+PROPERTIES: dict[str, dict[str, Any]] = {
+    "seed": {"type": "integer", "default": 42},
+    "object_count": {"anyOf": [{"type": "integer"}, {"type": "null"}], "default": None},
+    "state": {
+        "anyOf": [{"type": "array", "items": {"type": "number"}}, {"type": "null"}],
+        "default": None,
+    },
+    "label": {"type": "string", "default": ""},
+    "approach_dir": {"type": "string", "default": "."},
+    "max_steps": {"type": "integer", "default": 1000},
+    "max_frames": {"type": "integer", "default": 100},
+}
+TOOL_ARGUMENTS = {
+    "render_state": ("seed", "state", "label", "object_count"),
+    "render_policy": (
+        "approach_dir",
+        "seed",
+        "max_steps",
+        "max_frames",
+        "object_count",
+    ),
+}
+TOOL_DESCRIPTIONS = {
+    "render_state": (
+        "Render a reset state (seed) or observation vector (state) "
+        "as a PNG. Returns the saved image path."
+    ),
+    "render_policy": (
+        "Run approach_dir/approach.py inside the isolated container "
+        "and save episode frames as PNGs. Returns saved image paths."
+    ),
+}
+
+
+class RenderTools:
+    """The strict environment client is the only dependency beyond stdlib."""
+
+    def __init__(self, metadata: Path, tools: list[str]):
+        if not set(tools) <= TOOL_ARGUMENTS.keys():
+            raise ValueError("Unknown strict render tool")
+        self.metadata = metadata.resolve()
+        self.tools = tools
+
+    def list_tools(self) -> list[dict[str, Any]]:
+        """Describe the two fixed tools without a schema-generation dependency."""
+        return [
+            {
+                "name": name,
+                "description": TOOL_DESCRIPTIONS[name],
+                "inputSchema": {
+                    "type": "object",
+                    "properties": {
+                        key: PROPERTIES[key] for key in TOOL_ARGUMENTS[name]
+                    },
+                    "additionalProperties": False,
+                },
+            }
+            for name in self.tools
+        ]
+
+    def call(self, name: str, arguments: dict[str, Any]) -> str | list[str]:
+        """Use a fresh connection per call; policies never share a host process."""
+        # env_client is installed beside this standalone script, not in a project
+        # package. It contains only generic protocol/observation handling.
+        # pylint: disable=import-outside-toplevel,import-error
+        from env_client import BlackboxEnv  # type: ignore[import-not-found]
+
+        # pylint: enable=import-outside-toplevel,import-error
+
+        if name not in self.tools or not set(arguments) <= set(TOOL_ARGUMENTS[name]):
+            raise ValueError("Unknown tool or arguments")
+        root = self.metadata.parent
+        meta = json.loads(self.metadata.read_text(encoding="utf-8"))
+        if meta.get("strict") is not True:
+            raise ValueError("Strict rendering requires strict environment metadata")
+        with BlackboxEnv(meta, sandbox_root=root) as client:
+            if name == "render_state":
+                return str(root / client.render_state(**arguments))
+            kwargs = dict(arguments)
+            approach_dir = kwargs.pop("approach_dir", ".")
+            paths = client.render_policy(
+                approach_path=root / approach_dir / "approach.py", **kwargs
+            )
+            return [str(root / path) for path in paths]
+
+    def dispatch(self, request: dict[str, Any]) -> dict[str, Any] | None:
+        """Handle MCP lifecycle and tool calls; no resources, prompts, or proxies."""
+        if "id" not in request:
+            return None
+        response: dict[str, Any] = {"jsonrpc": "2.0", "id": request["id"]}
+        method, params = request.get("method"), request.get("params", {})
+        if method == "initialize":
+            version = params.get("protocolVersion")
+            response["result"] = {
+                "protocolVersion": version if version in VERSIONS else VERSIONS[-1],
+                "capabilities": {"tools": {"listChanged": False}},
+                "serverInfo": {"name": "robocode-tools", "version": "1.0"},
+            }
+        elif method == "ping":
+            response["result"] = {}
+        elif method == "tools/list":
+            response["result"] = {"tools": self.list_tools()}
+        elif method == "tools/call":
+            try:
+                value = self.call(params["name"], params.get("arguments", {}))
+                response["result"] = {
+                    "content": [
+                        {
+                            "type": "text",
+                            "text": (
+                                value if isinstance(value, str) else json.dumps(value)
+                            ),
+                        }
+                    ],
+                    "isError": False,
+                }
+            except Exception as exc:  # pylint: disable=broad-exception-caught
+                logging.exception("Render tool failed")
+                response["result"] = {
+                    "content": [{"type": "text", "text": str(exc)}],
+                    "isError": True,
+                }
+        else:
+            response["error"] = {"code": -32601, "message": "Method not found"}
+        return response
+
+
+def serve(tools: RenderTools, host: str, port: int) -> None:
+    """Serve JSON responses on the MCP HTTP endpoint; optional SSE is unsupported."""
+    if host != "127.0.0.1":
+        raise ValueError("Strict MCP must bind only to loopback")
+
+    class Handler(BaseHTTPRequestHandler):
+        """No files, uploads, URL fetching, or arbitrary RPC dispatch."""
+
+        def reply(self, status: int, value: Any = None) -> None:
+            """Write one JSON response with an explicit length."""
+            payload = b"" if value is None else json.dumps(value).encode()
+            self.send_response(status)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(payload)))
+            self.end_headers()
+            self.wfile.write(payload)
+
+        def valid(self) -> bool:
+            """Reject unrelated origins, paths, and protocol versions."""
+            origin = self.headers.get("Origin")
+            if origin and origin != f"http://127.0.0.1:{port}":
+                self.reply(403)
+                return False
+            if urlsplit(self.path).path != "/mcp":
+                self.reply(404)
+                return False
+            version = self.headers.get("MCP-Protocol-Version")
+            if version and version not in VERSIONS:
+                self.reply(400)
+                return False
+            return True
+
+        def do_POST(self) -> None:  # pylint: disable=invalid-name
+            """Handle one bounded JSON-RPC message."""
+            if not self.valid():
+                return
+            try:
+                size = int(self.headers.get("Content-Length", "0"))
+                if not 0 < size <= 1024 * 1024 or self.headers.get("Transfer-Encoding"):
+                    raise ValueError("Invalid request size")
+                request = json.loads(self.rfile.read(size))
+                if not isinstance(request, dict) or request.get("jsonrpc") != "2.0":
+                    raise ValueError("Expected JSON-RPC object")
+                response = tools.dispatch(request)
+            except (ValueError, TypeError, KeyError):
+                self.reply(400)
+                return
+            self.reply(202 if response is None else 200, response)
+
+        def do_GET(self) -> None:  # pylint: disable=invalid-name
+            """This stateless server has no optional server-to-client stream."""
+            if self.valid():
+                self.reply(405)
+
+        do_DELETE = do_GET
+
+    with ThreadingHTTPServer((host, port), Handler) as server:
+        server.serve_forever()
+
+
+def main() -> None:
+    """Start the only strict render server, under the strict Python interpreter."""
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--env-spaces", type=Path, required=True)
+    parser.add_argument("--tools", required=True)
+    parser.add_argument("--log-file", required=True)
+    parser.add_argument("--transport", choices=["http"], required=True)
+    parser.add_argument("--host", default="127.0.0.1")
+    parser.add_argument("--port", type=int, required=True)
+    args = parser.parse_args()
+    logging.basicConfig(filename=args.log_file, level=logging.INFO)
+    serve(RenderTools(args.env_spaces, args.tools.split(",")), args.host, args.port)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/src/robocode/utils/apptainer_sandbox.py b/src/robocode/utils/apptainer_sandbox.py
index 15235f37..54b89db5 100644
--- a/src/robocode/utils/apptainer_sandbox.py
+++ b/src/robocode/utils/apptainer_sandbox.py
@@ -90,7 +90,6 @@
     _stream_result_to_sandbox_result,
     agent_stdin,
 )
-from robocode.utils.strict_blackbox import STRICT_BLACKBOX_MCP_PYTHON
 from robocode.utils.telemetry import container_launch
 
 logger = logging.getLogger(__name__)
@@ -362,7 +361,7 @@ async def run_agent_in_apptainer_sandbox(
         # Under strict the agent's scripts run in the dependency-clean venv and
         # the render proxy in its own, so MCP packages never reach the former.
         agent_python = container_python(strict_blackbox)
-        mcp_python = STRICT_BLACKBOX_MCP_PYTHON if strict_blackbox else agent_python
+        mcp_python = agent_python
         agent_cmd = backend.build_cli_cmd(
             config,
             mcp_python_cmd=mcp_python,
diff --git a/src/robocode/utils/backends/claude.py b/src/robocode/utils/backends/claude.py
index eb07ec91..ab755649 100644
--- a/src/robocode/utils/backends/claude.py
+++ b/src/robocode/utils/backends/claude.py
@@ -229,6 +229,7 @@ def build_cli_cmd(
                 mcp_env_config_path,
                 log_path,
                 blackbox=config.blackbox,
+                strict_blackbox=getattr(config, "blackbox_strict", False),
                 transport=mcp_transport,
                 port=mcp_port,
             )
diff --git a/src/robocode/utils/backends/codex.py b/src/robocode/utils/backends/codex.py
index 445f357f..f71a650a 100644
--- a/src/robocode/utils/backends/codex.py
+++ b/src/robocode/utils/backends/codex.py
@@ -114,6 +114,7 @@ def build_cli_cmd(
                 mcp_env_config_path,
                 log_path,
                 blackbox=config.blackbox,
+                strict_blackbox=getattr(config, "blackbox_strict", False),
                 transport=mcp_transport,
                 port=mcp_port,
             )
diff --git a/src/robocode/utils/backends/opencode.py b/src/robocode/utils/backends/opencode.py
index 8bfe990f..bc0725b5 100644
--- a/src/robocode/utils/backends/opencode.py
+++ b/src/robocode/utils/backends/opencode.py
@@ -112,6 +112,7 @@ def build_cli_cmd(
                 mcp_env_config_path,
                 log_path,
                 blackbox=config.blackbox,
+                strict_blackbox=getattr(config, "blackbox_strict", False),
                 transport=mcp_transport,
                 port=mcp_port,
             )
diff --git a/src/robocode/utils/docker_sandbox.py b/src/robocode/utils/docker_sandbox.py
index cb99b4d6..e2c99cef 100644
--- a/src/robocode/utils/docker_sandbox.py
+++ b/src/robocode/utils/docker_sandbox.py
@@ -80,7 +80,6 @@
 )
 from robocode.utils.strict_blackbox import (
     STRICT_BLACKBOX_IMAGE,
-    STRICT_BLACKBOX_MCP_PYTHON,
     STRICT_BLACKBOX_PYTHON,
 )
 from robocode.utils.telemetry import container_launch
@@ -727,7 +726,7 @@ async def run_agent_in_docker_sandbox(
             env_server_port = int(metadata["port"])
         docker_image = STRICT_BLACKBOX_IMAGE if strict_blackbox else config.docker_image
         docker_python = container_python(strict_blackbox)
-        mcp_python = STRICT_BLACKBOX_MCP_PYTHON if strict_blackbox else docker_python
+        mcp_python = docker_python
         docker_cmd = _docker_run_prefix(
             container_name,
             docker_image,
diff --git a/src/robocode/utils/strict_blackbox.py b/src/robocode/utils/strict_blackbox.py
index 1e0aa2c4..514c7e2a 100644
--- a/src/robocode/utils/strict_blackbox.py
+++ b/src/robocode/utils/strict_blackbox.py
@@ -37,7 +37,6 @@
 
 STRICT_BLACKBOX_IMAGE = "robocode-strict-blackbox"
 STRICT_BLACKBOX_PYTHON = "/opt/robocode-strict/bin/python"
-STRICT_BLACKBOX_MCP_PYTHON = "/opt/robocode-mcp/bin/python"
 
 
 class StrictImportError(ValueError):
diff --git a/tests/utils/test_apptainer_sandbox.py b/tests/utils/test_apptainer_sandbox.py
index aed02ff5..d079881d 100644
--- a/tests/utils/test_apptainer_sandbox.py
+++ b/tests/utils/test_apptainer_sandbox.py
@@ -31,7 +31,6 @@
     _find_repo_root,
 )
 from robocode.utils.strict_blackbox import (
-    STRICT_BLACKBOX_MCP_PYTHON,
     STRICT_BLACKBOX_PYTHON,
 )
 
@@ -102,7 +101,7 @@ class _Launched(Exception):
     """Raised by the fake launcher once the command line has been captured."""
 
 
-def test_strict_run_wires_separate_interpreters(  # type: ignore
+def test_strict_run_uses_only_clean_interpreter(  # type: ignore
     tmp_path: Path, monkeypatch
 ) -> None:
     """The agent's scripts use the strict venv and the render proxy its own."""
@@ -147,7 +146,10 @@ def fake_popen(cmd: list[str], **kwargs):  # type: ignore
     assert f"{STRICT_BLACKBOX_PYTHON} -c" in joined
     assert STRICT_BLACKBOX_PYTHON in (sandbox_dir / "CLAUDE.md").read_text()
     start_script = (sandbox_dir / ".mcp" / MCP_START_SCRIPT).read_text()
-    assert f"{STRICT_BLACKBOX_MCP_PYTHON} -m robocode.mcp.server" in start_script
+    assert (
+        f"{STRICT_BLACKBOX_PYTHON} /opt/robocode-render/strict_server.py"
+        in start_script
+    )
     assert APPTAINER_PYTHON not in start_script
 
 
diff --git a/tests/utils/test_strict_blackbox.py b/tests/utils/test_strict_blackbox.py
index ef5efab8..e9a1965d 100644
--- a/tests/utils/test_strict_blackbox.py
+++ b/tests/utils/test_strict_blackbox.py
@@ -26,7 +26,6 @@
 from robocode.utils.episode import load_generated_approach
 from robocode.utils.strict_blackbox import (
     STRICT_ALLOWED_PACKAGES,
-    STRICT_BLACKBOX_MCP_PYTHON,
     STRICT_BLACKBOX_PYTHON,
     StrictImportError,
     check_strict_imports,
@@ -310,9 +309,8 @@ def test_strict_docker_launch_has_no_project_mounts(tmp_path: Path) -> None:
     assert "ss-pybullet" not in joined
 
 
-def test_strict_mcp_uses_separate_python_environment() -> None:
+def test_strict_mcp_uses_clean_python_environment() -> None:
     """MCP startup must not add its dependencies to the generated-code Python."""
-    assert STRICT_BLACKBOX_MCP_PYTHON != STRICT_BLACKBOX_PYTHON
     command = " ".join(
         _mcp_prestart_wrapper(["agent"], python_cmd=STRICT_BLACKBOX_PYTHON)
     )
@@ -336,7 +334,7 @@ def test_strict_container_keeps_generated_python_dependency_clean(
 def test_strict_container_mcp_renders_state_and_policy_through_host(
     container_backend: str, tmp_path: Path
 ) -> None:
-    """The isolated MCP interpreter can proxy strict renders to the host."""
+    """The clean interpreter can render states and policies through the host."""
     sandbox = tmp_path / "sandbox"
     sandbox.mkdir()
     (sandbox / "approach.py").write_text(
@@ -367,18 +365,17 @@ def test_strict_container_mcp_renders_state_and_policy_through_host(
                 strict=True,
             )
             code = (
-                "import asyncio, json; from pathlib import Path; "
-                "from robocode.mcp.server import build_blackbox_server; "
-                "srv=build_blackbox_server(['render_state','render_policy'], "
-                "Path('/sandbox/env_spaces.json')); "
-                "_,state=asyncio.run(srv.call_tool('render_state', {'seed': 3})); "
-                "_,policy=asyncio.run(srv.call_tool('render_policy', "
-                "{'seed': 3, 'max_steps': 2})); "
-                "print(json.dumps({'state': state['result'], "
-                "'policy': policy['result']}))"
+                "import sys,json; from pathlib import Path; "
+                "sys.path.insert(0, '/opt/robocode-render'); "
+                "from strict_server import RenderTools; "
+                "srv=RenderTools(Path('/sandbox/env_spaces.json'), "
+                "['render_state','render_policy']); "
+                "state=srv.call('render_state', {'seed': 3}); "
+                "policy=srv.call('render_policy', {'seed': 3, 'max_steps': 2}); "
+                "print(json.dumps({'state': state, 'policy': policy}))"
             )
             result = _strict_container_run(
-                container_backend, STRICT_BLACKBOX_MCP_PYTHON, code, sandbox=sandbox
+                container_backend, STRICT_BLACKBOX_PYTHON, code, sandbox=sandbox
             )
     finally:
         env.close()
diff --git a/tests/utils/test_strict_render_server.py b/tests/utils/test_strict_render_server.py
new file mode 100644
index 00000000..450112ba
--- /dev/null
+++ b/tests/utils/test_strict_render_server.py
@@ -0,0 +1,93 @@
+"""Strict rendering must work without importing project or MCP dependencies."""
+
+import asyncio
+import json
+import socket
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+from mcp import ClientSession
+from mcp.client.streamable_http import streamablehttp_client
+
+from robocode.mcp.strict_server import RenderTools
+
+
+def test_only_render_capabilities(tmp_path):
+    """Unknown RPCs cannot reach arbitrary client/server attributes."""
+    tools = RenderTools(tmp_path / "env_spaces.json", ["render_state", "render_policy"])
+    assert {t["name"] for t in tools.list_tools()} == {"render_state", "render_policy"}
+    assert tools.dispatch({"jsonrpc": "2.0", "id": 1, "method": "initialize"})[
+        "result"
+    ]["capabilities"] == {"tools": {"listChanged": False}}
+    assert tools.dispatch({"id": 2, "method": "getattr"})["error"]["code"] == -32601
+    assert tools.dispatch({"method": "notifications/initialized"}) is None
+    with pytest.raises(ValueError):
+        RenderTools(tmp_path / "meta", ["execute_python"])
+
+
+def test_official_mcp_client_interoperability(tmp_path):
+    """Exercise initialize, tools/list, success/error calls through the SDK client."""
+    source = Path(__file__).resolve().parents[2] / "src/robocode/mcp/strict_server.py"
+    (tmp_path / "strict_server.py").write_bytes(source.read_bytes())
+    (tmp_path / "env_spaces.json").write_text(json.dumps({"strict": True}))
+    (tmp_path / "env_client.py").write_text("""class BlackboxEnv:
+ def __init__(self, *args, **kwargs): pass
+ def __enter__(self): return self
+ def __exit__(self, *args): pass
+ def render_state(self, **kwargs): return 'state.png'
+ def render_policy(self, **kwargs): return ['frame.png']
+""")
+    with socket.socket() as listener:
+        listener.bind(("127.0.0.1", 0))
+        port = listener.getsockname()[1]
+    with subprocess.Popen(
+        [
+            sys.executable,
+            str(tmp_path / "strict_server.py"),
+            "--env-spaces",
+            str(tmp_path / "env_spaces.json"),
+            "--tools",
+            "render_state,render_policy",
+            "--transport",
+            "http",
+            "--port",
+            str(port),
+            "--log-file",
+            str(tmp_path / "server.log"),
+        ],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    ) as process:
+
+        async def check():
+            for _ in range(100):
+                try:
+                    reader, writer = await asyncio.open_connection("127.0.0.1", port)
+                    del reader
+                    writer.close()
+                    await writer.wait_closed()
+                    break
+                except OSError:
+                    await asyncio.sleep(0.05)
+            async with streamablehttp_client(f"http://127.0.0.1:{port}/mcp") as (
+                read,
+                write,
+                _,
+            ):
+                async with ClientSession(read, write) as session:
+                    await session.initialize()
+                    assert len((await session.list_tools()).tools) == 2
+                    state = await session.call_tool("render_state", {"seed": 3})
+                    assert not state.isError and "state.png" in state.content[0].text
+                    policy = await session.call_tool("render_policy", {"max_steps": 2})
+                    assert not policy.isError and "frame.png" in policy.content[0].text
+                    bad = await session.call_tool("arbitrary_command", {})
+                    assert bad.isError
+
+        try:
+            asyncio.run(check())
+        finally:
+            process.terminate()
+            process.wait(timeout=5)

From 8bc126a29f2722fc3471f37e5541d54e26f05c24 Mon Sep 17 00:00:00 2001
From: merlerm <matteo.merler@gmail.com>
Date: Sat, 19 Sep 2026 16:47:03 +0200
Subject: [PATCH 3/3] Run Apptainer agents in disconnected namespaces with
 fixed relays

---
 .gitignore                                    |   2 +
 README.md                                     |   4 +-
 docker/entrypoint.sh                          |   6 +-
 docker/strict-blackbox-entrypoint.sh          |   6 +-
 docs/apptainer-network-isolation.md           | 240 ++++++++++
 docs/blackbox.md                              |  63 ++-
 experiments/conf/approach/best_of_k.yaml      |   7 +-
 experiments/conf/approach/llm_genplan.yaml    |   7 +-
 src/robocode/approaches/agentic_base.py       |   3 +
 .../approaches/agentic_cdl_approach.py        |   3 +
 .../approaches/llm_genplan_approach.py        |  33 +-
 src/robocode/utils/apptainer_environment.py   | 116 +++++
 src/robocode/utils/apptainer_sandbox.py       | 441 +++++++-----------
 src/robocode/utils/env_server.py              |   6 +-
 src/robocode/utils/isolated_transport.py      | 155 ++++++
 tests/approaches/test_llm_genplan_approach.py |  16 +
 tests/utils/test_apptainer_sandbox.py         | 233 +++------
 tests/utils/test_isolated_transport.py        |  91 ++++
 18 files changed, 938 insertions(+), 494 deletions(-)
 create mode 100644 docs/apptainer-network-isolation.md
 create mode 100644 src/robocode/utils/apptainer_environment.py
 create mode 100644 src/robocode/utils/isolated_transport.py
 create mode 100644 tests/utils/test_isolated_transport.py

diff --git a/.gitignore b/.gitignore
index 2c7debdd..f46aab2d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -180,3 +180,5 @@ ref/
 # pddlstream writes its FastDownward scratch files into the working directory.
 temp/
 statistics/
+
+.apptainer-env-cache/
diff --git a/README.md b/README.md
index 6919b288..b1249b09 100644
--- a/README.md
+++ b/README.md
@@ -84,7 +84,7 @@ python experiments/run_experiment.py approach=agentic \
     replicate_seed=0 eval_seed="$EVAL_SEED"
 ```
 
-With login-file authentication, Docker and Apptainer copy only `auth.json` into a throwaway Codex home. Host `config.toml`, `AGENTS.md`, skills, and session history are not mounted. Each fresh experiment starts with an empty sandbox-local session directory; only an automatic retry of that same experiment can resume it.
+With login-file authentication, Docker copies only `auth.json` into a throwaway Codex home. Apptainer keeps authentication on the host in its inference broker and mounts no real credentials. Neither mounts host `config.toml`, `AGENTS.md`, skills, or session history. Each fresh experiment starts with an empty sandbox-local session directory; only an automatic retry of that same experiment can resume it.
 
 #### OpenCode (multi-provider)
 
@@ -379,7 +379,7 @@ The agent runs inside a Docker container (`robocode-sandbox`) that provides full
 | Network | `init-firewall.sh` whitelists API endpoints for the configured provider (Anthropic, OpenAI, Google, etc.), GitHub IPs, and telemetry; blocks everything else via iptables. Extra domains are passed via `ROBOCODE_FIREWALL_EXTRA_DOMAINS`. |
 | Write hook | Claude backend: `PreToolUse` hook in `.claude/settings.json` double-checks Write/Edit paths stay inside `/sandbox`. Codex and OpenCode rely on the enclosing Docker filesystem boundary. |
 
-The Apptainer backend (`container_backend=apptainer`, for HPC clusters with no Docker daemon) keeps the same filesystem isolation but has **no network firewall**: unprivileged Apptainer cannot grant `CAP_NET_ADMIN`, so `init-firewall.sh` is skipped and generated code runs with unrestricted network egress. Use Docker where the iptables allowlist matters.
+The Apptainer backend (`container_backend=apptainer`, for HPC clusters without Docker) now runs Codex and Claude in a disconnected network namespace (`--userns --net --network none`). A host broker permits validated model inference, and a separate relay reaches only the experiment environment server. Agent processes cannot use general internet access, and provider credentials stay outside the container. See [implementation, test evidence, and limitations](docs/apptainer-network-isolation.md). Unsupported Apptainer backends and GenPlan fail closed.
 
 ### What the agent sees
 
diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh
index c2288bd4..752bc3c4 100644
--- a/docker/entrypoint.sh
+++ b/docker/entrypoint.sh
@@ -30,13 +30,13 @@ if [ "$(id -u)" -eq 0 ]; then
     HOME=/home/node USER=node LOGNAME=node \
         "${run_as_node[@]}" uv sync --frozen --python python3.11 "${uv_extra_args[@]}"
 else
-    # Unprivileged Apptainer runs preserve the host UID.
+    # Preserve non-root invocation behavior when firewall setup is explicitly skipped.
     uv sync --frozen --python python3.11 "${uv_extra_args[@]}"
 fi
 cd /sandbox
 
-# Skipped under unprivileged Apptainer, which cannot grant CAP_NET_ADMIN;
-# ROBOCODE_SKIP_FIREWALL=1 is set by apptainer_sandbox.py.
+# Docker firewall setup. Preserve the existing explicit skip override.
+# Apptainer does not invoke this entrypoint; it uses a disconnected namespace.
 if [ "${ROBOCODE_SKIP_FIREWALL:-0}" = "1" ]; then
     echo "entrypoint: ROBOCODE_SKIP_FIREWALL=1, skipping firewall init" >&2
 else
diff --git a/docker/strict-blackbox-entrypoint.sh b/docker/strict-blackbox-entrypoint.sh
index d5d4af5f..635dbbf2 100644
--- a/docker/strict-blackbox-entrypoint.sh
+++ b/docker/strict-blackbox-entrypoint.sh
@@ -6,8 +6,8 @@
 set -euo pipefail
 IFS=$'\n\t'
 
-# Skipped under unprivileged Apptainer, which cannot grant CAP_NET_ADMIN;
-# ROBOCODE_SKIP_FIREWALL=1 is set by apptainer_sandbox.py.
+# Docker firewall setup. Preserve the existing explicit skip override.
+# Apptainer does not invoke this entrypoint; it uses a disconnected namespace.
 if [ "${ROBOCODE_SKIP_FIREWALL:-0}" = "1" ]; then
     echo "entrypoint: ROBOCODE_SKIP_FIREWALL=1, skipping firewall init" >&2
 else
@@ -34,5 +34,5 @@ if [ "$(id -u)" -eq 0 ]; then
         -- "$@"
 fi
 
-# Unprivileged Apptainer runs preserve the host UID.
+# Preserve non-root invocation behavior when firewall setup is explicitly skipped.
 exec "$@"
diff --git a/docs/apptainer-network-isolation.md b/docs/apptainer-network-isolation.md
new file mode 100644
index 00000000..41e6a3a2
--- /dev/null
+++ b/docs/apptainer-network-isolation.md
@@ -0,0 +1,240 @@
+# Apptainer isolation — implementation and audit, 2026-09-19
+
+The updated launcher supports usable, unprivileged network isolation for **Codex
+and Claude** on the tested cluster node `rk019192` (Apptainer 1.4.3, UID 1013).
+This proves the agent-runtime boundary, not end-to-end experiment isolation:
+`GeneratedProgramApproach._load_generated` still calls the host-side
+`load_generated_approach`, which executes the generated policy during scoring.
+That separate execution path is not isolated by the broker and must be addressed
+before treating entire experiments as internet-blocked.
+
+No Docker, sudo, or host firewall changes were needed. The previous launcher
+shared host networking; disabling Codex web tools alone did not prevent fetching
+websites or installing packages.
+
+## Why there is a broker
+
+A completely disconnected container cannot contact a model API. The agent now
+runs with `--userns --net --network none`: its only network interface is loopback.
+A small trusted program on the host, the **model broker**, accepts requests through
+an explicitly mounted Unix socket. Unix sockets are local IPC and can cross this
+boundary when their filesystem path is deliberately mounted.
+
+The broker forwards only validated inference requests to a fixed OpenAI or
+Anthropic HTTPS endpoint. It holds the real credentials outside the container.
+It rejects arbitrary destinations, CONNECT, redirects, client-supplied routing
+and authentication headers, provider-hosted web search, remote MCP, remote
+image/file retrieval, and unsupported API operations. It accepts client-side tool
+definitions so the agent can still run its local shell and experiment tools.
+Its logs contain endpoint/status/rejection information, not prompts or tokens.
+
+This is an explicit exception for model inference, not a general internet proxy.
+An agent can submit permitted inference requests itself; the broker is not an
+anti-abuse or spending boundary for that authorized API channel. Prompts and
+inline raster images are intentionally sent to the model provider.
+
+A **separate pinned relay** connects to one host experiment environment-server
+port, selected by trusted launcher configuration. The agent can reset, step, and
+render through the existing environment protocol. Editing its metadata cannot
+select a different host service. This server remains part of the trusted surface.
+
+## Code structure and Docker compatibility
+
+There is one supported Apptainer transport, with separate responsibilities:
+
+| Module | Responsibility |
+| --- | --- |
+| `apptainer_sandbox.py` | Agent launch; broker/relay lifetime; local CLI configuration |
+| `model_broker.py` | Host credentials, fixed upstream selection, HTTP request policy |
+| `isolated_transport.py` | Container namespace checks and fixed-destination byte relays |
+| `apptainer_environment.py` | Trusted dependency preparation and clean child environment |
+
+The relays do not implement another model policy; they deliver bytes to the broker
+or one environment server. Provider hostname constants live in `backends/__init__.py`
+and are shared with Docker's existing domain registry. The broker's `BrokerUpstream`
+is a resolved, credential-bearing host connection, not a second provider registry.
+
+The old Apptainer credential-forwarding helper, credential mounts, firewall-domain
+arguments, and image-entrypoint dependency flags have been removed. There is no
+legacy transport switch or fallback. Environment relays require an explicit trusted
+`env_server_port`; `env_spaces.json` cannot select a host destination. GenPlan and
+Best-of-K reject Apptainer at configuration time instead of exposing a dummy runner.
+
+Docker keeps its existing credential mounts/environment, firewall domain settings,
+and entrypoint execution. Its scripts' behavior is unchanged; only stale comments
+about Apptainer were corrected. Docker's domain firewall and Apptainer's broker
+provide different policies and are not selectable alternatives within Apptainer.
+`ROBOCODE_FIREWALL_EXTRA_DOMAINS` remains a Docker setting, never a broker override.
+Docker runtime tests require a collaborator's Docker-capable machine; unit tests
+cover its command construction, auth, shared callers, and GenPlan dispatch here.
+
+## Production behavior
+
+- Every supported agent launch and resumed session uses the disconnected namespace.
+  A supervisor checks non-root UID, loopback-only interfaces, no IPv4 routes,
+  zero capabilities (including the bounding set), and `NoNewPrivs: 1` before
+  starting the agent. Failure aborts; there is no host-network fallback.
+- Filtered mounts, `--containall`, `--no-home`, `--cleanenv`, and PID isolation
+  prevent default host-home mounts and inherited secrets. Real provider auth
+  files are not mounted. Container API tokens are inert local placeholders.
+- Regular Python environments are prepared and cached by a trusted installer
+  **before** agent execution. That phase has network access but no agent files,
+  sessions, or credentials. The completed environment is mounted read-only.
+  The agent phase skips the image's online entrypoint, sets `UV_OFFLINE=1` and
+  `PIP_NO_INDEX=1`, and is additionally blocked by the actual network namespace.
+- Strict runs use the dependency-clean strict image. Codex web tools remain
+  disabled by the pulled upstream configuration; broker enforcement also rejects
+  attempts to enable hosted tools through raw API requests.
+- OpenCode, custom upstreams, and the separate GenPlan/Best-of-K Apptainer path
+  are currently unsupported and fail closed. They require a separate integration.
+- Host credentials must be valid when a run starts; the broker does not implement
+  OAuth refresh. Refresh/login happens on the trusted host.
+
+Use this updated checkout. Old experiment worktrees retain their own old launcher
+code and are **not** protected merely by reusing an updated SIF. The default
+root-level SIFs were replaced with the tested Codex-capable images from
+`outputs/codex-ready`; the previous images are retained with
+`.pre-network-broker-20260919.bak` suffixes.
+
+## Real test evidence
+
+The audit pairs each isolated launch with an explicitly unrestricted **test-only
+positive control**, using both regular and strict images. It checks namespace
+identity, interfaces, routes, capabilities, and these actual communication paths:
+
+- Python HTTPS and direct IPv4 TCP; curl, curl with a pinned IP, wget, Git HTTPS,
+  Node HTTP, Bash `/dev/tcp`, and strict-image `pip download six==1.17.0`.
+- DNS over UDP and TCP; controlled host IPv4/IPv6 loopback TCP, UDP, HTTP, and an
+  explicit HTTP proxy; host abstract Unix sockets and an unmounted pathname socket.
+- Raw sockets, route modification, namespace joining, nested namespaces, and a
+  positive listener inside the container's own loopback namespace.
+
+The live Codex and Claude runs execute the supplied probe unchanged through the
+production launcher. Both passed: model calls worked while the tested forbidden
+communication paths failed. All **14 broker attacks** per backend were rejected
+with HTTP 403, testing both the loopback API endpoint and the mounted Unix socket:
+CONNECT, GET, absolute URLs, hosted OpenAI/Claude web search, remote images, and
+remote MCP. Prompts also ask agents to try their built-in web tools.
+
+Further live checks passed:
+
+- Codex and Claude resumed sessions, including absence of real credential files
+  and secret environment variables, loopback-only networking, and blocked direct TCP.
+- Claude strict environment reset/step plus MCP rendering to a PNG.
+- Codex regular environment imports, and a real Motion2D reset/step rollout using
+  the default regular SIF and the optional bilevel dependency environment.
+
+**All successful Claude checks used the `_tom` token**, loaded on the host without
+printing or mounting it. An initial Claude broker attempt returned HTTP 400 for
+its context-management payload; the fixed, host-selected context-management beta
+header resolved it and the complete live audit passed on rerun.
+
+Artifacts are local and ignored by Git:
+
+| Directory under `network_audit_results/` | Evidence |
+| --- | --- |
+| `cleanup-live-codex/` | Post-cleanup paired controls and live Codex broker attacks |
+| `cleanup-live-claude/` | Post-cleanup paired controls and live Claude `_tom` broker attacks |
+| `cleanup-strict-env-smoke/` | Post-cleanup pinned relay and MCP render |
+| `cleanup-whitebox-rollout/` | Post-cleanup prepared bilevel environment and Motion2D rollout |
+| `final-default-isolation/` | Final paired network controls using both default SIFs |
+| `isolated-live-codex/` | Paired controls and successful live Codex attack suite |
+| `isolated-live-v2-claude/` | Paired controls and successful live Claude attack suite |
+| `strict-env-smoke/` | Strict environment and rendered PNG |
+| `regular-smoke/` | Regular dependency imports |
+| `default-whitebox-rollout/` | Default-image Motion2D rollout |
+| `live-methods/` | Historical successful internet access through the old launcher |
+
+Resume evidence is stored in the live run work directories. Summaries include
+host/kernel/runtime and image fingerprints. Earlier artifacts call the unrestricted
+control `production` and live results `live_baseline`; current code uses
+`unrestricted_control` and `live_run` to avoid confusion.
+
+External IPv6 had no working host-network control, so that particular test is
+**inconclusive**, not a pass. Host IPv6 loopback isolation was positively tested.
+The original strict image provided the positive `pip download` control and its
+isolated download was blocked. The corrected strict image removes pip/setuptools
+and the old MCP environment; both final images report pip as unavailable, not as
+a successful blocked-download test. Missing reports, refused prompts, changed probe scripts,
+failed model calls, and missing executables are never isolation successes.
+
+## Reproduce
+
+From the updated repository root, using fresh results directories:
+
+```sh
+.venv/bin/python -m integration_tests.red_team_sandbox \
+  --network-isolation-only apptainer \
+  --network-results-dir network_audit_results/new-audit
+```
+
+Add `--network-live-backend codex` or `--network-live-backend claude` for a paid
+live agent audit (configured budget $2). Host credentials are required. Select a
+specific pair of images with `--network-image-dir outputs/codex-ready` if needed.
+The full `--apptainer-strict-blackbox` suite now includes its network and
+package-install attacks as well as import, filesystem, and environment-protocol
+attacks. The deterministic audit is required for strong network evidence; the
+older webpage-only script reports a `BLOCKED` self-report as inconclusive (exit 2).
+
+Run actual socket/container tests outside additional execution sandboxes that
+forbid all sockets: such a sandbox can prevent Apptainer itself from starting and
+would invalidate the test. The audit's in-namespace listener is a positive control
+against this false pass.
+
+## Code verification
+
+The post-cleanup core regression run passed **259 tests**, with **19 Docker runtime
+checks skipped** because Docker is unavailable. A further suite passed 60 checks
+covering Best-of-K, retry routing, environment-server behavior, and provider domain
+lists (the provider-list checks also appear in the core suite). Broker tests now
+verify host-only credential loading instead of the removed credential-forwarding
+helper. Namespace, request-policy, pinned-port, and secret-exclusion checks remain.
+
+Mypy passed for eleven checked modules. Pylint and whitespace checks passed.
+
+Docker's network launch, firewall, and credential behavior are unchanged.
+The later strict-image fix switches strict Docker rendering to the same numerical
+interpreter and standalone server, so strict Docker users must rebuild their image.
+The shared source filter also now removes bytecode. Actual Docker execution still
+needs verification on a Docker host.
+
+## Scope of the conclusion
+
+This cluster can run usable isolated Apptainer agents with this implementation.
+A different cluster is not required by a fundamental rootless-Apptainer limit.
+The trusted host, kernel, Apptainer, broker, dependency preparation, and environment
+server remain part of the security boundary. Finite tests cannot prove the absence
+of every kernel or protocol vulnerability. Repeat the audit on every execution
+node/image and after runtime, broker, provider-protocol, or launcher changes.
+Disabling tools alone, or switching container runtimes alone, is insufficient.
+
+Primary references:
+
+- [Apptainer 1.4 network virtualization](https://apptainer.org/docs/user/1.4/networking.html)
+  documents the unprivileged `none` network.
+- [OpenAI configuration reference](https://learn.chatgpt.com/docs/config-file/config-reference)
+  documents custom providers and transport settings.
+- [Claude context editing](https://platform.claude.com/docs/en/build-with-claude/context-editing)
+  documents the context-management beta used by the fixed upstream headers.
+
+## Full red-team follow-up and strict package correction
+
+Both backends completed the full 44-case catalog plus the network/broker audit.
+The original summaries include failures/inconclusives and are retained unchanged.
+See `network_audit_results/full-redteam-review.md` for raw results and adjudication.
+
+A real strict-image defect was found: changing interpreters or `sys.path` exposed
+the reduced project/MCP packages. The rebuilt default image removes that entire
+environment; both render tools use stdlib and the allowed numerical client.
+The final import audit also inventories readable package sources and tests system
+Python, the strict interpreter, and manually injected paths. Both agents tested
+imports from inside a rendered policy. The supervisor refuses the original image.
+A separate real bytecode exposure in models-off mounts was fixed by excluding
+caches and compiled files. The generated-code host-scoring scope is unchanged.
+
+Final artifacts: `final-network-codex/`, `final-network-claude/`,
+`strict-import-final/`, and `strict-startup-guard/`, all under
+`network_audit_results/`. Final network tests passed for both live agents and
+rejected every tested broker bypass. External IPv6 and missing pip controls remain
+explicitly inconclusive. Focused regressions passed 141 tests (2 Docker tests
+skipped), plus 8 checker/source-filter tests; mypy and pylint passed.
diff --git a/docs/blackbox.md b/docs/blackbox.md
index c3995085..1dc61a8b 100644
--- a/docs/blackbox.md
+++ b/docs/blackbox.md
@@ -195,7 +195,8 @@ The approach, in `train()`:
    `{host, port, token, observation_space, action_space, max_steps}`.
    `host` is `host.docker.internal` for Docker (mapped via
    `--add-host host.docker.internal:host-gateway`) or `127.0.0.1` for the
-   apptainer and local backends.
+   Apptainer and local backends. Apptainer then rewrites the port to a private
+   loopback relay pinned to the separately configured host `env_server_port`.
 4. Mounts a filtered copy of the repo that strips `environments/`, the kinder
    `envs/` and `demos/`, plus the always-excluded `oracles/`, `primitives/`,
    `tests/`, and `docs/`.
@@ -230,10 +231,10 @@ makes blackbox meaningful differs:
   many `apptainer.conf` setups still bind the host `/home`, so the agent could
   read the real source straight off `/home/<user>/.../environments`.
   `--containall` is what fixes this, dropping all default binds (home, tmp, cwd)
-  so the stripped source is the only source present. It reaches the env server
-  over `127.0.0.1` (apptainer shares the host network namespace, so no
-  `--add-host`/firewall is needed; note this also means apptainer does not apply
-  the default-deny network firewall). Run
+  so the stripped source is the only source present. Its network namespace has
+  only loopback. A Unix-socket relay reaches one pinned host env-server port, and
+  a separate broker permits validated model inference. See
+  [network isolation](apptainer-network-isolation.md). Run
   `python integration_tests/red_team_sandbox.py --apptainer-blackbox` (needs
   `robocode-sandbox.sif` built) to verify env source stays unreachable.
 - **local**: best-effort ONLY, isolation is NOT enforced. The OS-level sandbox
@@ -251,15 +252,16 @@ makes blackbox meaningful differs:
   only for quick local iteration, not for results that depend on the agent not
   having read the source. Use `docker` or `apptainer` for enforced isolation.
 
-Strict mode runs under Docker or Apptainer during synthesis. The strict image gets
-one writable mount (`/sandbox`). Under Docker its firewall allows the model provider
-plus only the exact host TCP port of the environment server; GitHub, SSH, package
-registries, and other host ports are not allowed. Under Apptainer the same image runs
-as `robocode-strict-blackbox.sif` with the sandbox as its only mount, but, as in
-legacy blackbox, unprivileged Apptainer cannot install the firewall, so that network
-restriction is not enforced there: the strict ablation then rests on the
-dependency-clean image, the strict env server, and the scoring-time import allowlist.
-Scoring needs no container: the import allowlist check
+Strict mode runs under Docker or Apptainer during synthesis. Under Docker its
+firewall allows the model provider plus only the exact host TCP port of the
+environment server; GitHub, SSH, package registries, and other host ports are not
+allowed. Under Apptainer the dependency-clean `robocode-strict-blackbox.sif` runs
+in a disconnected namespace with the sandbox, session directory, and read-only
+broker socket directory mounted. Only validated model inference and the pinned
+environment-server relay cross that network boundary.
+
+Final scoring currently runs on the host. Its import allowlist check is a
+methodological guardrail, not a network or hostile-code sandbox. The check
 runs before the program is loaded, so an approach that imports `pybullet_helpers`,
 `tomsgeoms2d`, `robocode`, `kinder`, or any other undeclared dependency fails the
 run with a message naming the import instead of silently succeeding from the host
@@ -324,7 +326,7 @@ state snapshots.
   |                   |                       sandbox_dir/mcp_renders/*.png
   +-------------------+---------------------------------------^---------+
                       |  host.docker.internal:port (docker)   | bind mount
-                      |  127.0.0.1:port (apptainer/local)      | (rw)
+                      |  127.0.0.1:port (local only)          | (rw)
   ====================+======= container boundary (firewall:   | =========
                       |         default-DROP + allow host /24) |
                       |                                        |
@@ -351,6 +353,11 @@ state snapshots.
   Withheld:  environments/ , kinder envs+demos , oracles/ , primitives/ , tests/ , docs/
 ```
 
+Apptainer uses the same JSON protocol through a different transport: private
+container loopback → mounted Unix socket → one pinned host environment-server
+port. Its namespace has no external interface. The diagram's Docker firewall
+and direct local connection do not describe Apptainer's network boundary.
+
 ## Key takeaways
 
 - **One protocol, two consumers inside the sandbox.** The agent's own test
@@ -361,7 +368,8 @@ state snapshots.
   host into the shared `mcp_renders/` mount; only relative paths travel back over
   the socket.
 - **Isolation is layered.** Withheld source mounts, JSON-only codec, per-run
-  token, per-connection fresh env, and a default-deny firewall. The `local`
+  token, per-connection fresh env, and Docker's firewall or Apptainer's
+  disconnected namespace with restricted relays. The `local`
   backend is best-effort only.
 
 ## Red-teaming strict blackbox
@@ -371,21 +379,21 @@ Build the strict image, then run its dedicated live adversarial suite:
 ```bash
 bash docker/build_strict_blackbox.sh
 python integration_tests/red_team_sandbox.py --strict-blackbox
-# Apptainer: the same suite minus the network probe (no firewall there).
+# Apptainer: the same suite, including network and package-install probes.
 bash docker/build_strict_blackbox_sif.sh
 python integration_tests/red_team_sandbox.py --apptainer-strict-blackbox
 ```
 
 The suite first proves the allowed surface works: the generated-program
 interpreter can import NumPy and SciPy and can reset and step the environment, and
-the separate MCP interpreter renders a state through the host.
+the same dependency-clean interpreter renders a state through the host.
 It then asks an agent to attack the same configuration used in production and
 fails if any of these boundaries break:
 
 - the generated-program interpreter imports RoboCode, KinDER, MCP, Gymnasium,
   geometry, or simulator packages, or installs a package from the network;
-- the separate MCP interpreter exposes environment, primitive, simulator, or
-  robotics modules;
+- changing interpreters/package paths, or executing a policy through MCP,
+  exposes withheld project, framework, environment, or simulator packages;
 - environment source, the host-side canary, or arbitrary Internet content is
   reachable through filesystem or network probing;
 - withheld client methods or raw protocol commands such as `get_state`,
@@ -406,14 +414,19 @@ after scoring begins.
 Blackbox mode is a *methodological* constraint first: it stops the agent from
 reading environment source so it must discover the dynamics empirically. The
 isolation behind it (withheld mounts, JSON-only codec, per-run token,
-per-connection env, default-deny firewall) is real. The host never executes
-agent code: the only things it runs are env stepping and `render_state`, both
-trusted. `render_policy` deliberately runs in the container, so an agent that
+per-connection env, and backend-specific network restrictions) applies during
+agent execution. The environment server runs trusted env stepping and
+`render_state`, never agent code. `render_policy` deliberately runs in the container, so an agent that
 writes a malicious `approach.py` cannot reach the env source through rendering
-(there is no env source in the container, and the host never execs the file).
+(there is no env source in the container, and the environment server never
+executes the file).
 The `blackbox_render_*` red-team tests exercise exactly this path.
 
-One limit is worth stating plainly:
+Final policy scoring is a separate host-side execution path and is not contained
+by these agent-runtime protections. Full experiment network isolation requires
+addressing that path too.
+
+Another limit:
 
 - **The env server listens on all interfaces.** It binds
   `0.0.0.0:<ephemeral>` and the container firewall opens the host's `/24`, so
diff --git a/experiments/conf/approach/best_of_k.yaml b/experiments/conf/approach/best_of_k.yaml
index 01481c39..7f95f1d5 100644
--- a/experiments/conf/approach/best_of_k.yaml
+++ b/experiments/conf/approach/best_of_k.yaml
@@ -16,8 +16,7 @@ max_budget_usd: 20.0  # dollar cap; only bounds cost-reporting backends (null ->
 # true: each candidate repeats GenPlan's summary -> strategy -> code flow, no debug loop.
 chain_of_thought: false
 # The per-episode validation timeout is the shared top-level eval_timeout.
-# Run the whole loop inside one sandbox container (like the agentic approach),
-# so generated code never executes on the host. local runs in-process.
-container_backend: docker  # docker | apptainer | local
+# Run candidate generation and validation in Docker. Final scoring still loads
+# the policy on the host; local also runs generation in-process.
+container_backend: docker  # docker | local; Apptainer transport is unsupported
 docker_image: robocode-sandbox
-sif_path: null  # null -> <repo_root>/robocode-sandbox.sif
diff --git a/experiments/conf/approach/llm_genplan.yaml b/experiments/conf/approach/llm_genplan.yaml
index 9dd680b6..230a0a5a 100644
--- a/experiments/conf/approach/llm_genplan.yaml
+++ b/experiments/conf/approach/llm_genplan.yaml
@@ -13,8 +13,7 @@ max_debug_attempts: 4  # step cap: 1 initial attempt + this many debug attempts
 max_budget_usd: 20.0  # dollar cap; only bounds cost-reporting backends (null -> step cap only)
 chain_of_thought: true  # summary -> strategy -> code; false: single prompt -> code
 # The per-episode validation timeout is the shared top-level eval_timeout.
-# Run the whole genplan loop inside one sandbox container (like the agentic
-# approach), so generated code never executes on the host. local runs in-process.
-container_backend: docker  # docker | apptainer | local
+# Run generation and debugging in Docker. Final scoring still loads the policy
+# on the host; local also runs generation in-process.
+container_backend: docker  # docker | local; Apptainer transport is unsupported
 docker_image: robocode-sandbox
-sif_path: null  # null -> <repo_root>/robocode-sandbox.sif
diff --git a/src/robocode/approaches/agentic_base.py b/src/robocode/approaches/agentic_base.py
index dd2aa64e..a5ddc292 100644
--- a/src/robocode/approaches/agentic_base.py
+++ b/src/robocode/approaches/agentic_base.py
@@ -18,6 +18,7 @@ class that does both) is deliberate, so the generalized baseline cannot be broke
 import sys
 from collections.abc import Callable
 from contextlib import ExitStack
+from dataclasses import replace
 from pathlib import Path
 from typing import Any, TypeVar
 
@@ -334,6 +335,8 @@ def _run_sandbox(
                         ),
                         strict=self._blackbox_strict,
                     )
+                if self._blackbox and apptainer_config is not None:
+                    apptainer_config = replace(apptainer_config, env_server_port=port)
                 result = run_with_rate_limit_retry(
                     docker_config,
                     config,
diff --git a/src/robocode/approaches/agentic_cdl_approach.py b/src/robocode/approaches/agentic_cdl_approach.py
index 0180c18a..e4958ea7 100644
--- a/src/robocode/approaches/agentic_cdl_approach.py
+++ b/src/robocode/approaches/agentic_cdl_approach.py
@@ -17,6 +17,7 @@
 import sys
 from collections.abc import Callable
 from contextlib import ExitStack
+from dataclasses import replace
 from pathlib import Path
 from typing import Any, TypeVar
 
@@ -302,6 +303,8 @@ def train(self) -> None:  # noqa: C901 — mirrors AgenticApproach.train
                             list(self._primitives)
                         ),
                     )
+                if self._blackbox and apptainer_config is not None:
+                    apptainer_config = replace(apptainer_config, env_server_port=port)
                 result = run_with_rate_limit_retry(
                     docker_config,
                     config,
diff --git a/src/robocode/approaches/llm_genplan_approach.py b/src/robocode/approaches/llm_genplan_approach.py
index 2c3551f1..5c6ae455 100644
--- a/src/robocode/approaches/llm_genplan_approach.py
+++ b/src/robocode/approaches/llm_genplan_approach.py
@@ -28,7 +28,6 @@
 from robocode import prompts
 from robocode.approaches.base_approach import BaseApproach
 from robocode.primitive_descriptions import format_primitives_description
-from robocode.utils.apptainer_sandbox import _DEFAULT_SIF, run_genplan_in_apptainer
 from robocode.utils.docker_sandbox import run_genplan_in_docker
 from robocode.utils.episode import load_generated_approach
 from robocode.utils.genplan_validate import (
@@ -88,7 +87,6 @@ def __init__(
         use_docker: bool = True,
         container_backend: str | None = None,
         docker_image: str = "robocode-sandbox",
-        sif_path: str | None = None,
         **kwargs: Any,
     ) -> None:
         super().__init__(
@@ -102,6 +100,12 @@ def __init__(
         self._container_backend = resolve_container_backend(
             container_backend, use_docker
         )
+        if self._container_backend == "apptainer":
+            raise ValueError(
+                "GenPlan/Best-of-K does not support the isolated Apptainer transport. "
+                "Use an agentic Codex/Claude approach on Apptainer. "
+                "GenPlan's existing Docker backend remains supported."
+            )
         # Sandboxed runs build the client inside the container, so the host
         # needs no client/key.
         self._client: LLMClient | None = (
@@ -119,7 +123,6 @@ def __init__(
         self._chain_of_thought = chain_of_thought
         self._eval_timeout = eval_timeout
         self._docker_image = docker_image
-        self._sif_path = Path(sif_path) if sif_path is not None else _DEFAULT_SIF
         self._generated: Any = None
         self.total_cost_usd: float | None = None
         # Number of LLM generations made (debug attempts for genplan, candidates
@@ -136,9 +139,9 @@ def train(self) -> None:
             "(max_debug_attempts / max_generation_steps)"
         )
 
-        # Sandboxed: run the whole loop inside one container (docker/apptainer)
+        # Sandboxed: run the whole loop inside one Docker container
         # via the genplan driver; the driver reruns train() locally inside.
-        if self._container_backend in ("docker", "apptainer"):
+        if self._container_backend == "docker":
             self._train_in_container()
             self._load_generated(self._output_dir / "sandbox" / "approach.py")
             return
@@ -222,20 +225,12 @@ def _train_in_container(self) -> None:
         config = self._driver_config(completion)
         (sandbox_dir / "genplan_config.json").write_text(json.dumps(config))
         include_bilevel = "bilevel_models" in self._primitives
-        if self._container_backend == "apptainer":
-            run_genplan_in_apptainer(
-                sandbox_dir,
-                completion,
-                sif_path=self._sif_path,
-                include_bilevel=include_bilevel,
-            )
-        else:
-            run_genplan_in_docker(
-                sandbox_dir,
-                completion,
-                image=self._docker_image,
-                include_bilevel=include_bilevel,
-            )
+        run_genplan_in_docker(
+            sandbox_dir,
+            completion,
+            image=self._docker_image,
+            include_bilevel=include_bilevel,
+        )
         cost = json.loads((sandbox_dir / "cost.json").read_text(encoding="utf-8"))
         self.total_cost_usd = cost["total_cost_usd"]
         self.num_generations = cost.get("num_generations")
diff --git a/src/robocode/utils/apptainer_environment.py b/src/robocode/utils/apptainer_environment.py
new file mode 100644
index 00000000..96fd88ce
--- /dev/null
+++ b/src/robocode/utils/apptainer_environment.py
@@ -0,0 +1,116 @@
+"""Prepare trusted Python dependencies before starting isolated agent execution."""
+
+from __future__ import annotations
+
+import fcntl
+import hashlib
+import os
+import shutil
+import subprocess
+import tempfile
+from pathlib import Path
+
+
+def clean_apptainer_env() -> dict[str, str]:
+    """Never pass provider secrets or Apptainer override variables to children."""
+    return {
+        key: os.environ[key]
+        for key in (
+            "HOME",
+            "PATH",
+            "USER",
+            "LOGNAME",
+            "LANG",
+            "LC_ALL",
+            "TERM",
+            "TMPDIR",
+        )
+        if key in os.environ
+    }
+
+
+def prepared_environment(sif: Path, binds: list[str], *, include_bilevel: bool) -> Path:
+    """Cache a venv prepared without agent files, credentials, or session mounts.
+
+    The preparation container is intentionally network-capable, but runs only the
+    trusted locked installer. Its completed venv is mounted read-only for agents.
+    Cache entries are never populated from an agent's writable container overlay.
+    """
+    from robocode.utils.docker_sandbox import (  # pylint: disable=import-outside-toplevel
+        _find_repo_root,
+    )
+
+    root = _find_repo_root()
+    cache = root / ".apptainer-env-cache"
+    cache.mkdir(mode=0o700, exist_ok=True)
+    digest = hashlib.sha256()
+    digest.update(str(sif.resolve()).encode())
+    digest.update(str((sif.stat().st_size, sif.stat().st_mtime_ns)).encode())
+    digest.update(str(include_bilevel).encode())
+    for path in (
+        root / "pyproject.toml",
+        root / "uv.lock",
+        root / "third-party/kindergarden/pyproject.toml",
+    ):
+        digest.update(path.read_bytes())
+    for bind in binds:
+        directory = Path(bind.split(":", 1)[0])
+        if directory.is_dir():
+            for metadata in sorted(directory.rglob("pyproject.toml")):
+                digest.update(str(metadata.relative_to(directory)).encode())
+                digest.update(metadata.read_bytes())
+    key = digest.hexdigest()
+    destination = cache / key
+    with (cache / (key + ".lock")).open("w", encoding="utf-8") as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX)
+        if (destination / "READY").is_file():
+            return destination / "venv"
+        if destination.exists():
+            shutil.rmtree(destination)
+        temporary = Path(tempfile.mkdtemp(prefix="prepare-", dir=cache))
+        cmd = [
+            "apptainer",
+            "exec",
+            "--userns",
+            "--containall",
+            "--cleanenv",
+            "--no-home",
+            "--writable-tmpfs",
+            "--pwd",
+            "/robocode",
+            "--bind",
+            f"{temporary}:/prepared",
+        ]
+        for bind in binds:
+            cmd += ["--bind", bind]
+        # Seed from the image to avoid redownloading its heavy runtime packages.
+        script = (
+            "cp -a /robocode/.venv /prepared/venv && "
+            "UV_PROJECT_ENVIRONMENT=/prepared/venv "
+            "uv sync --frozen --python /usr/bin/python3.11"
+        )
+        if include_bilevel:
+            script += " --extra bilevel"
+        cmd += [str(sif.resolve()), "/bin/sh", "-ec", script]
+        proc = subprocess.run(
+            cmd,
+            env=clean_apptainer_env(),
+            capture_output=True,
+            text=True,
+            timeout=600,
+            check=False,
+        )
+        (temporary / "prepare.log").write_text(
+            proc.stdout + proc.stderr, encoding="utf-8"
+        )
+        if proc.returncode:
+            raise RuntimeError(
+                "Trusted dependency preparation failed; inspect "
+                f"{temporary / 'prepare.log'}"
+            )
+        # uv created console scripts for /prepared/venv; the runtime bind uses
+        # that same path. The agent interpreter remains /robocode/.venv/python
+        # via a second read-only bind for existing experiment configuration.
+        (temporary / "READY").write_text("1\n", encoding="utf-8")
+        temporary.rename(destination)
+        return destination / "venv"
diff --git a/src/robocode/utils/apptainer_sandbox.py b/src/robocode/utils/apptainer_sandbox.py
index 54b89db5..b5b0d26b 100644
--- a/src/robocode/utils/apptainer_sandbox.py
+++ b/src/robocode/utils/apptainer_sandbox.py
@@ -1,90 +1,60 @@
-"""Apptainer/Singularity-based sandboxed agent runner.
-
-Mirror of :mod:`robocode.utils.docker_sandbox` for environments where the
-Docker daemon is unavailable (typical on HPC clusters). The SIF image is
-built from the existing ``docker/Dockerfile`` via ``docker/build_sif.sh``
-(podman build + apptainer build) -- no separate definition file.
-
-The container interior (entrypoint, firewall script, /robocode/.venv,
-bind-mount layout) is byte-for-byte identical to the Docker image. The
-only differences are at the host invocation layer:
-
-* ``--bind`` instead of ``-v``
-* ``--env KEY=val`` instead of ``-e KEY=val``
-* ``--pwd`` instead of ``-w``
-* ``--writable-tmpfs`` so the entrypoint's ``uv sync`` can write to
-  ``/robocode/.venv`` (the SIF rootfs is read-only)
-* ``--containall`` so administrator-configured home, tmp, and cwd binds do not
-  expose host files beyond the explicit filtered mounts
-* ``--no-home`` so the host home doesn't shadow ``/home/node``
-* ``--cleanenv`` so the host env doesn't leak in
-* ``--pid`` so the container gets its own PID namespace (Docker does this by
-  default; apptainer shares the host's unless asked)
-
-Namespaces: the filesystem, PID, and IPC namespaces are the container's own.
-The NETWORK namespace is still the host's: ``--net`` needs
-privileges the unprivileged cluster install does not have, which is also why the
-firewall is skipped. So host loopback services stay reachable from the sandbox,
-and the render http server must pick a free host port (see ``_free_port``).
-
-``init-firewall.sh`` is skipped via ``ROBOCODE_SKIP_FIREWALL=1``: the
-unprivileged apptainer install on the target cluster can't grant real
-``CAP_NET_ADMIN``, so iptables would fail.
-
-The image ENTRYPOINT is invoked explicitly rather than via
-``apptainer run`` so behaviour does not depend on Apptainer's runscript
-translation of Docker images.
-
-Strict blackbox runs (``blackbox_strict=True``) execute in
-``robocode-strict-blackbox.sif`` instead, built from
-``docker/Dockerfile.strict-blackbox`` via ``docker/build_strict_blackbox_sif.sh``.
-No project code is bound into it: the sandbox is the only mount. The strict
-firewall (model provider plus the env server's port) cannot be installed here
-for the same reason the regular one is skipped, so under Apptainer the strict
-ablation rests on the dependency-clean image, the strict env server, and the
-host-side import allowlist at scoring time.
+"""Apptainer agent runner with an isolated network and host inference broker.
+
+Agents run without root in a fresh user/network/PID/IPC namespace using
+``--net --network none``, filtered mounts, a clean environment, and no-new-privileges.
+A supervisor verifies the network and capability boundary before starting the CLI.
+The host broker accepts only validated model inference over a mounted Unix socket;
+provider credentials remain on the host. A separate socket relays to one pinned
+experiment environment server. Neither relay provides general internet access.
+
+Regular Python dependencies are prepared in a trusted installer phase and mounted
+read-only; the agent phase never runs the network-dependent image entrypoint.
+Strict runs use the dependency-clean strict SIF. Unsupported backend/GenPlan paths
+fail closed. See ``docs/apptainer-network-isolation.md`` for evidence and limits.
 """
 
 from __future__ import annotations
 
+import json
 import logging
-import os
+import shutil
 import subprocess
 import tempfile
+import threading
 import time
 import uuid
 from collections.abc import Iterator
 from contextlib import ExitStack, contextmanager, nullcontext
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any
 
 from robocode.mcp import MCP_STARTUP_TIMEOUT_MS
-from robocode.utils.backends import (
-    PROVIDERS,
-    AgentBackend,
-    firewall_domains_for_provider,
-    provider_from_model,
+from robocode.utils.apptainer_environment import (
+    clean_apptainer_env,
+    prepared_environment,
 )
-from robocode.utils.claude_auth import (
-    sandbox_claude_session_store,
-    throwaway_claude_config,
-)
-from robocode.utils.codex_auth import sandbox_codex_sessions, throwaway_codex_home
+from robocode.utils.backends import AgentBackend
+from robocode.utils.claude_auth import sandbox_claude_session_store
+from robocode.utils.codex_auth import sandbox_codex_sessions
 from robocode.utils.docker_sandbox import (
     DOCKER_PYTHON,
-    GENPLAN_CONTAINER_TIMEOUT_S,
     _filtered_repo_mounts,
     _find_repo_root,
-    _get_claude_oauth_token,
     _mcp_prestart_wrapper,
     container_python,
 )
+from robocode.utils.isolated_transport import UnixRelay
+from robocode.utils.model_broker import (
+    BROKER_DIR,
+    MODEL_PORT,
+    BrokerUpstream,
+    load_broker_upstream,
+    model_broker,
+)
 from robocode.utils.sandbox import (
     SandboxConfig,
     SandboxResult,
     _final_commit,
-    _free_port,
     _initial_commit,
     _setup_sandbox_dir,
     _stream_result_to_sandbox_result,
@@ -133,6 +103,8 @@ class ApptainerSandboxConfig(SandboxConfig):
     sif_path: Path = _DEFAULT_SIF
     blackbox_strict: bool = False
     strict_sif_path: Path = _DEFAULT_STRICT_SIF
+    # Trusted host destination. Never inferred from agent-writable metadata.
+    env_server_port: int | None = None
 
 
 def sif_path_for(config: ApptainerSandboxConfig) -> Path:
@@ -140,67 +112,6 @@ def sif_path_for(config: ApptainerSandboxConfig) -> Path:
     return config.strict_sif_path if config.blackbox_strict else config.sif_path
 
 
-@contextmanager
-def _build_apptainer_auth_args(
-    backend_name: str,
-) -> Iterator[tuple[list[str], dict[str, str]]]:
-    """Yield Apptainer CLI args and env vars for backend authentication.
-
-    Mirrors :func:`docker_sandbox._build_docker_auth_args`. Secrets (the
-    Claude OAuth token, provider API keys) are returned as host env vars
-    with Apptainer's ``APPTAINERENV_`` prefix rather than inline ``--env``
-    flags: Apptainer injects ``APPTAINERENV_*`` into the container even
-    under ``--cleanenv``, and the value never reaches argv (world-readable
-    via ``ps`` / ``/proc/<pid>/cmdline`` on shared nodes). Only non-secret
-    bind mounts are returned as CLI args.
-
-    The credentials fallback uses a writable throwaway copy, never the live
-    host config, so experiment reads and writes cannot leak across runs or into
-    the operator's Claude history.
-    """
-    apptainer_args: list[str] = []
-    extra_env: dict[str, str] = {}
-
-    with ExitStack() as stack:
-        if backend_name == "claude":
-            oauth_token = _get_claude_oauth_token()
-            if oauth_token:
-                # APPTAINERENV_ prefix, not an inline --env flag, so the secret is
-                # injected into the container (surviving --cleanenv) without ever
-                # appearing on the command line.
-                extra_env["APPTAINERENV_CLAUDE_CODE_OAUTH_TOKEN"] = oauth_token
-            else:
-                logger.warning(
-                    "No Claude OAuth token found; falling back to a throwaway "
-                    "credentials-only config. Run `claude login` on the host "
-                    "if the container cannot authenticate."
-                )
-                claude_copy = stack.enter_context(throwaway_claude_config())
-                apptainer_args += ["--bind", f"{claude_copy}:/home/node/.claude"]
-        elif backend_name == "codex":
-            if os.environ.get("CODEX_API_KEY"):
-                extra_env["APPTAINERENV_CODEX_API_KEY"] = os.environ["CODEX_API_KEY"]
-            else:
-                codex_home = stack.enter_context(throwaway_codex_home())
-                apptainer_args += ["--bind", f"{codex_home}:/home/node/.codex"]
-        else:
-            opencode_data = Path.home() / ".local" / "share" / "opencode"
-            if opencode_data.exists():
-                apptainer_args += [
-                    "--bind",
-                    f"{opencode_data}:/home/node/.local/share/opencode",
-                ]
-
-            for info in PROVIDERS.values():
-                if info.api_key_env:
-                    val = os.environ.get(info.api_key_env)
-                    if val:
-                        # APPTAINERENV_ keeps the key off argv (see above).
-                        extra_env[f"APPTAINERENV_{info.api_key_env}"] = val
-
-        yield apptainer_args, extra_env
-
-
 def _apptainer_exec_prefix() -> list[str]:
     """Return the filesystem/process isolation shared by all Apptainer runs."""
     # --no-home alone does not reliably suppress administrator-configured host
@@ -209,6 +120,10 @@ def _apptainer_exec_prefix() -> list[str]:
     return [
         "apptainer",
         "exec",
+        "--userns",
+        "--net",
+        "--network",
+        "none",
         "--containall",
         # Apptainer shares the host PID namespace by default, so a `pkill -f`
         # inside the container could otherwise reach the harness, concurrent
@@ -229,8 +144,6 @@ def _build_apptainer_cmd(
     src_abs: str | None,
     kindergarden_abs: str | None,
     kinder_baselines_abs: str | None,
-    auth_args: list[str],
-    firewall_domains: list[str],
     agent_cmd: list[str],
     extra_binds: list[str] | None = None,
     ss_pybullet_abs: str | None = None,
@@ -240,8 +153,9 @@ def _build_apptainer_cmd(
     Split out from :func:`run_agent_in_apptainer_sandbox` so unit tests
     can inspect the constructed command without running anything.
 
-    A strict blackbox launch passes ``None`` for the repo mounts: its image holds
-    no project code, so the sandbox is the only mount.
+    Strict blackbox launches omit project source mounts. The high-level runner
+    adds the broker and session mounts. This builder never installs dependencies
+    or forwards provider credentials and firewall settings.
     """
     cmd = _apptainer_exec_prefix()
     cmd += [
@@ -253,8 +167,6 @@ def _build_apptainer_cmd(
         # tools (--containall drops the host env, so this must be explicit).
         "--env",
         f"MCP_TIMEOUT={MCP_STARTUP_TIMEOUT_MS}",
-        "--env",
-        "ROBOCODE_SKIP_FIREWALL=1",
         # Headless container has no GPU, so mujoco's Dynamic3D offscreen renderer
         # must use OSMesa (software); EGL device displays fail without a GPU.
         "--env",
@@ -263,19 +175,6 @@ def _build_apptainer_cmd(
         "PYOPENGL_PLATFORM=osmesa",
     ]
 
-    if firewall_domains:
-        cmd += [
-            "--env",
-            f"ROBOCODE_FIREWALL_EXTRA_DOMAINS={','.join(firewall_domains)}",
-        ]
-
-    # Only when the bilevel_models primitive is in play: sync the bilevel extra
-    # (the bind is added below). Otherwise no bilevel source/deps enter the sandbox.
-    if kinder_baselines_abs is not None:
-        cmd += ["--env", "ROBOCODE_UV_EXTRA_ARGS=--extra bilevel"]
-
-    cmd += auth_args
-
     cmd += ["--bind", f"{sandbox_abs}:/sandbox"]
     if src_abs is not None:
         cmd += ["--bind", f"{src_abs}:/robocode/src"]
@@ -292,23 +191,116 @@ def _build_apptainer_cmd(
         cmd += ["--bind", bind]
     cmd += [
         str(sif_path_for(config)),
-        "/usr/local/bin/entrypoint.sh",
+        "/usr/bin/setpriv",
+        "--no-new-privs",
+        "--",
     ]
     cmd += agent_cmd
     return cmd
 
 
+@contextmanager
+def _isolated_transport(
+    config: ApptainerSandboxConfig, provider: BrokerUpstream
+) -> Iterator[Path]:
+    """Own the broker and optional pinned environment relay for one agent run."""
+    with ExitStack() as isolation:
+        bridge = Path(
+            isolation.enter_context(tempfile.TemporaryDirectory(prefix="robocode-net-"))
+        )
+        isolation.enter_context(
+            model_broker(bridge, provider, config.sandbox_dir.parent / "broker.jsonl")
+        )
+        shutil.copyfile(
+            Path(__file__).with_name("isolated_transport.py"), bridge / "transport.py"
+        )
+        listeners = [{"port": MODEL_PORT, "socket": f"{BROKER_DIR}/model.sock"}]
+        # Only immutable host configuration selects a destination. Sandbox metadata
+        # describes the client view and never authorizes a host connection.
+        metadata_path = config.sandbox_dir / "env_spaces.json"
+        port = config.env_server_port
+        if metadata_path.exists() and port is None:
+            raise RuntimeError(
+                "env_spaces.json requires an explicit trusted env_server_port"
+            )
+        if port is not None:
+            if (
+                isinstance(port, bool)
+                or not isinstance(port, int)
+                or not 1 <= port <= 65535
+            ):
+                raise RuntimeError("Invalid trusted environment server port")
+            metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
+            relay = isolation.enter_context(
+                UnixRelay(str(bridge / "environment.sock"), ("127.0.0.1", port))
+            )
+            thread = threading.Thread(target=relay.serve_forever, daemon=True)
+            thread.start()
+            isolation.callback(thread.join, 5)
+            isolation.callback(relay.shutdown)
+            listeners.append(
+                {"port": MODEL_PORT + 1, "socket": f"{BROKER_DIR}/environment.sock"}
+            )
+            metadata.update(host="127.0.0.1", port=MODEL_PORT + 1)
+            (config.sandbox_dir / "env_spaces.json").write_text(
+                json.dumps(metadata), encoding="utf-8"
+            )
+        (bridge / "transport.json").write_text(
+            json.dumps(
+                {"listeners": listeners, "strict_blackbox": config.blackbox_strict}
+            ),
+            encoding="utf-8",
+        )
+        yield bridge
+
+
+def _model_client(
+    backend_name: str, agent_cmd: list[str]
+) -> tuple[list[str], dict[str, str]]:
+    """Point a CLI at the local broker using inert tokens; never load real auth."""
+    agent_cmd = list(agent_cmd)
+    local_token = "local-broker-no-provider-secret"
+    client_env = {
+        "APPTAINERENV_ROBOCODE_MODEL_TOKEN": local_token,
+        "APPTAINERENV_UV_OFFLINE": "1",
+        "APPTAINERENV_PIP_NO_INDEX": "1",
+    }
+    if backend_name == "claude":
+        client_env.update(
+            {
+                "APPTAINERENV_ANTHROPIC_BASE_URL": f"http://127.0.0.1:{MODEL_PORT}",
+                "APPTAINERENV_ANTHROPIC_AUTH_TOKEN": local_token,
+                "APPTAINERENV_CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1",
+            }
+        )
+    if backend_name == "codex":
+        # Explicit custom provider avoids giving the CLI any real credential.
+        # Transport is SSE; websocket upgrades are rejected at the broker.
+        overrides = {
+            "model_provider": "robocode",
+            "model_providers.robocode.name": "Robocode isolated broker",
+            "model_providers.robocode.base_url": f"http://127.0.0.1:{MODEL_PORT}/v1",
+            "model_providers.robocode.wire_api": "responses",
+            "model_providers.robocode.env_key": "ROBOCODE_MODEL_TOKEN",
+            "model_providers.robocode.supports_websockets": False,
+            "features.responses_websockets": False,
+            "features.responses_websockets_v2": False,
+        }
+        for key, value in overrides.items():
+            agent_cmd[-1:-1] = ["--config", f"{key}={json.dumps(value)}"]
+    return agent_cmd, client_env
+
+
 async def run_agent_in_apptainer_sandbox(
     config: ApptainerSandboxConfig,
     backend: AgentBackend,
 ) -> SandboxResult:
-    """Run an agent inside the ``robocode-sandbox`` SIF via apptainer.
-
-    Step-for-step parallel of
-    :func:`~robocode.utils.docker_sandbox.run_agent_in_docker_sandbox`.
-    See the module docstring for the docker -> apptainer flag mapping.
-    """
+    """Run a supported agent with isolated networking and validated inference."""
     backend_name = backend.name
+    if getattr(backend, "_base_url", ""):
+        raise RuntimeError(
+            "Custom model endpoints are not supported by the isolated broker"
+        )
     strict_blackbox = config.blackbox_strict
 
     sif_path = sif_path_for(config)
@@ -322,13 +314,13 @@ async def run_agent_in_apptainer_sandbox(
             f"SIF image not found at {sif_path}; build it with: bash {build_script}"
         )
 
+    provider = load_broker_upstream(backend_name)
     _setup_sandbox_dir(config)
 
     sandbox_abs = str(config.sandbox_dir.resolve())
     run_id = f"apptainer-sandbox-{uuid.uuid4().hex[:8]}"
 
-    # The strict image holds no project code, so nothing is mounted beside the
-    # sandbox.
+    # The strict image needs no project source mounts.
     mounts = (
         nullcontext((None, None, None, None))
         if strict_blackbox
@@ -344,22 +336,31 @@ async def run_agent_in_apptainer_sandbox(
             filtered_kinder_baselines,
             ss_pybullet,
         ),
-        _build_apptainer_auth_args(backend_name) as (auth_args, auth_env),
+        _isolated_transport(config, provider) as bridge,
     ):
-        firewall_domains: list[str] = []
-        if backend_name in {"opencode", "codex"}:
-            firewall_domains = firewall_domains_for_provider(
-                "codex"
-                if backend_name == "codex"
-                else provider_from_model(config.model)
+        transport_binds = [f"{bridge}:{BROKER_DIR}:ro"]
+        if not strict_blackbox:
+            preparation_binds = [
+                f"{filtered_src}:/robocode/src",
+                f"{filtered_kindergarden}:/robocode/third-party/kindergarden",
+                f"{_find_repo_root() / 'pyproject.toml'}:/robocode/pyproject.toml:ro",
+                f"{_find_repo_root() / 'uv.lock'}:/robocode/uv.lock:ro",
+            ]
+            if filtered_kinder_baselines is not None:
+                preparation_binds.append(
+                    f"{filtered_kinder_baselines}:/robocode/third-party/kinder-baselines"
+                )
+            venv = prepared_environment(
+                sif_path,
+                preparation_binds,
+                include_bilevel=filtered_kinder_baselines is not None,
             )
-
-        # Apptainer shares the host network namespace (even with --containall and
-        # --pid), so use a free loopback port for the render http server to avoid
-        # colliding with the host or a concurrent run.
-        mcp_port = _free_port()
-        # Under strict the agent's scripts run in the dependency-clean venv and
-        # the render proxy in its own, so MCP packages never reach the former.
+            transport_binds += [
+                f"{venv}:/robocode/.venv:ro",
+                f"{venv}:/prepared/venv:ro",
+            ]
+        mcp_port = MODEL_PORT + 2
+        # Strict rendering uses the same dependency-clean interpreter as agents.
         agent_python = container_python(strict_blackbox)
         mcp_python = agent_python
         agent_cmd = backend.build_cli_cmd(
@@ -371,6 +372,7 @@ async def run_agent_in_apptainer_sandbox(
             mcp_transport="http",
             mcp_port=mcp_port,
         )
+        agent_cmd, client_env = _model_client(backend_name, agent_cmd)
         # Start and health-check the render server before the CLI (same wrapper
         # as docker) so its tools are connected on the agent's first turn.
         if config.mcp_tools:
@@ -380,7 +382,7 @@ async def run_agent_in_apptainer_sandbox(
 
         # Persist the CLI session store under the sandbox dir (survives the
         # ephemeral container) so a rate-limited run can be resumed via
-        # --continue in a fresh retry container. Claude only.
+        # the backend resume command in a fresh retry container.
         session_binds: list[str] = []
         if backend_name == "claude":
             sessions_dir = sandbox_claude_session_store(config.sandbox_dir)
@@ -407,10 +409,13 @@ async def run_agent_in_apptainer_sandbox(
                 if filtered_kinder_baselines is not None
                 else None
             ),
-            auth_args=auth_args,
-            firewall_domains=firewall_domains,
-            agent_cmd=agent_cmd,
-            extra_binds=session_binds + tel_binds,
+            agent_cmd=[
+                agent_python,
+                f"{BROKER_DIR}/transport.py",
+                f"{BROKER_DIR}/transport.json",
+                *agent_cmd,
+            ],
+            extra_binds=session_binds + tel_binds + transport_binds,
         )
 
         backend.setup_sandbox_files(
@@ -420,7 +425,8 @@ async def run_agent_in_apptainer_sandbox(
         )
         _initial_commit(config.sandbox_dir)
 
-        env = backend.build_env(config, auth_env if auth_env else None)
+        env = clean_apptainer_env()
+        env.update(client_env)
         env.update(tel_env)
 
         logger.info(
@@ -444,6 +450,8 @@ async def run_agent_in_apptainer_sandbox(
                 stdout=subprocess.PIPE,
                 stderr=stderr_file,
                 text=True,
+                # Claude stops capped runs via killpg(proc.pid); own the group.
+                start_new_session=True,
             )
 
             stream = backend.parse_stream(
@@ -451,6 +459,10 @@ async def run_agent_in_apptainer_sandbox(
                 stream_log_path=config.sandbox_dir.parent / "stream.jsonl",
                 stderr_file=stderr_file,
             )
+            stderr_file.seek(0)
+            (config.sandbox_dir.parent / "container.stderr").write_text(
+                stderr_file.read(), encoding="utf-8"
+            )
         wall_time_s = time.monotonic() - wall_start
 
         logger.info(
@@ -469,94 +481,3 @@ async def run_agent_in_apptainer_sandbox(
             config.output_filename,
             wall_time_s=wall_time_s,
         )
-
-
-def run_genplan_in_apptainer(
-    sandbox_dir: Path,
-    completion_cfg: dict[str, Any],
-    sif_path: Path = _DEFAULT_SIF,
-    timeout: float = GENPLAN_CONTAINER_TIMEOUT_S,
-    include_bilevel: bool = False,
-) -> None:
-    """Apptainer analog of :func:`docker_sandbox.run_genplan_in_docker`.
-
-    Mirrors the docker function: runs the whole LLM-GenPlan loop inside one
-    sandbox container via the genplan driver, which reads
-    ``sandbox_dir/genplan_config.json`` and writes ``sandbox_dir/approach.py``
-    and ``sandbox_dir/cost.json``. Keeps ``primitives`` in the source mount so
-    the policy can build/use them as eval does on the host. With *include_bilevel*
-    (the genplan config requested ``bilevel_models``), the kinder-baselines source
-    is mounted and ``uv sync --extra bilevel`` runs so the models are importable.
-    """
-    if not sif_path.exists():
-        raise RuntimeError(
-            f"SIF image not found at {sif_path}; build it with: bash docker/build_sif.sh"
-        )
-    run_id = f"apptainer-genplan-{uuid.uuid4().hex[:8]}"
-    auth_backend = "claude" if completion_cfg["provider"] == "cli" else "opencode"
-    with (
-        _filtered_repo_mounts(
-            keep_primitives=True, include_bilevel=include_bilevel
-        ) as (
-            filtered_src,
-            filtered_kindergarden,
-            filtered_kinder_baselines,
-            ss_pybullet,
-        ),
-        _build_apptainer_auth_args(auth_backend) as (auth_args, auth_env),
-    ):
-        firewall_domains = firewall_domains_for_provider(
-            completion_cfg["provider"], completion_cfg.get("base_url", "")
-        )
-        firewall_env: list[str] = []
-        if firewall_domains:
-            firewall_env = [
-                "--env",
-                f"ROBOCODE_FIREWALL_EXTRA_DOMAINS={','.join(firewall_domains)}",
-            ]
-        # With bilevel_models, mount the kinder-baselines path deps and tell the
-        # entrypoint to `uv sync --extra bilevel` (mirrors _docker_run_prefix).
-        bilevel_env: list[str] = []
-        bilevel_bind: list[str] = []
-        ss_pybullet_bind: list[str] = []
-        if ss_pybullet is not None:
-            ss_pybullet_bind = [
-                "--bind",
-                f"{ss_pybullet.resolve()}:/robocode/third-party/ss-pybullet:ro",
-            ]
-        if filtered_kinder_baselines is not None:
-            bilevel_env = ["--env", "ROBOCODE_UV_EXTRA_ARGS=--extra bilevel"]
-            bilevel_bind = [
-                "--bind",
-                f"{filtered_kinder_baselines.resolve()}"
-                ":/robocode/third-party/kinder-baselines",
-            ]
-        apptainer_cmd = [
-            *_apptainer_exec_prefix(),
-            "--env",
-            "ROBOCODE_SKIP_FIREWALL=1",
-            *firewall_env,
-            *bilevel_env,
-            *auth_args,
-            "--bind",
-            f"{sandbox_dir.resolve()}:/sandbox",
-            "--bind",
-            f"{filtered_src.resolve()}:/robocode/src",
-            "--bind",
-            f"{filtered_kindergarden.resolve()}:/robocode/third-party/kindergarden",
-            *ss_pybullet_bind,
-            *bilevel_bind,
-            str(sif_path),
-            "/usr/local/bin/entrypoint.sh",
-            APPTAINER_PYTHON,
-            "-m",
-            "robocode.approaches.genplan_driver",
-        ]
-        logger.info("Starting genplan Apptainer run %s sif=%s", run_id, sif_path)
-        subprocess.run(
-            apptainer_cmd,
-            env={**os.environ, **auth_env},
-            stdin=subprocess.DEVNULL,
-            check=True,
-            timeout=timeout,
-        )
diff --git a/src/robocode/utils/env_server.py b/src/robocode/utils/env_server.py
index 126a2055..f4b97db5 100644
--- a/src/robocode/utils/env_server.py
+++ b/src/robocode/utils/env_server.py
@@ -234,8 +234,10 @@ def write_env_spaces(
     """Write ``env_spaces.json``, the metadata the sandbox's env_client reads.
 
     The host is ``host.docker.internal`` for Docker (mapped to the host
-    gateway via ``--add-host``) and ``127.0.0.1`` for the apptainer and local
-    backends, which share the host's loopback. *primitives_manifest* (from
+    gateway via ``--add-host``) and ``127.0.0.1`` for local and Apptainer. Local
+    uses host loopback directly; Apptainer rewrites the port to its private
+    loopback relay using the explicit trusted ``env_server_port`` config.
+    *primitives_manifest* (from
     :func:`robocode.primitives.blackbox_primitive_manifest`) tells the sandbox
     how to rebuild the eval-time primitives; the caller passes it rather than
     this module importing the primitives package, keeping the host process
diff --git a/src/robocode/utils/isolated_transport.py b/src/robocode/utils/isolated_transport.py
new file mode 100644
index 00000000..7e37a63f
--- /dev/null
+++ b/src/robocode/utils/isolated_transport.py
@@ -0,0 +1,155 @@
+"""Small fixed-destination stream relays; runnable with the container stdlib.
+
+The container listeners connect ONLY to named Unix sockets. There is no network
+bridge, DNS forwarding, SOCKS negotiation, CONNECT support, or destination field.
+Host-side environment relays have one destination selected by trusted code.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import pkgutil
+import select
+import signal
+import socket
+import socketserver
+import subprocess
+import sys
+import threading
+from contextlib import ExitStack
+from pathlib import Path
+from typing import Any
+
+
+def copy_streams(left: socket.socket, right: socket.socket) -> None:
+    """Copy duplex streams while preserving half-close semantics."""
+    readable = [left, right]
+    while readable:
+        ready, _, _ = select.select(readable, [], [], 120)
+        if not ready:
+            return
+        for source in ready:
+            target = right if source is left else left
+            data = source.recv(65536)
+            if data:
+                target.sendall(data)
+            else:
+                readable.remove(source)
+                target.shutdown(socket.SHUT_WR)
+
+
+class RelayHandler(socketserver.BaseRequestHandler):
+    """Relay bytes to the one address configured by the trusted parent."""
+
+    def handle(self) -> None:
+        try:
+            target = self.server.target  # type: ignore[attr-defined]
+            if isinstance(target, str):
+                remote = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+                remote.settimeout(120)
+                remote.connect(target)
+            else:
+                remote = socket.create_connection(target, timeout=120)
+            with remote:
+                copy_streams(self.request, remote)
+        except OSError:
+            pass
+
+
+class UnixRelay(socketserver.ThreadingMixIn, socketserver.UnixStreamServer):
+    """Host endpoint pinned to a single loopback environment server."""
+
+    daemon_threads = True
+    block_on_close = False
+
+    def __init__(self, path: str, target: tuple[str, int]):
+        self.target = target
+        super().__init__(path, RelayHandler)
+
+
+class TCPRelay(socketserver.ThreadingMixIn, socketserver.TCPServer):
+    """Container loopback endpoint pinned to a mounted Unix socket."""
+
+    daemon_threads = True
+    block_on_close = False
+    allow_reuse_address = True
+
+    def __init__(self, port: int, target: str):
+        self.target = target
+        super().__init__(("127.0.0.1", port), RelayHandler)
+
+
+def verify_namespace() -> None:
+    """Refuse to execute any agent unless the kernel boundary is established."""
+    if os.getuid() == 0 or {name for _, name in socket.if_nameindex()} != {"lo"}:
+        raise RuntimeError(
+            "Apptainer isolation requires a non-root, loopback-only namespace"
+        )
+    if len(Path("/proc/net/route").read_text(encoding="utf-8").splitlines()) != 1:
+        raise RuntimeError("Unexpected route in isolated namespace")
+    status = dict(
+        line.split(":", 1)
+        for line in Path("/proc/self/status").read_text(encoding="utf-8").splitlines()
+    )
+    for key in ("CapEff", "CapPrm", "CapBnd", "CapInh", "CapAmb"):
+        if int(status[key].strip(), 16):
+            raise RuntimeError("Agent retains capabilities")
+    if status["NoNewPrivs"].strip() != "1":
+        raise RuntimeError("NoNewPrivs is required")
+
+
+def verify_strict_runtime() -> None:
+    """Reject old images and readable third-party Python package environments.
+
+    This startup guard supplements the image audit; a virtualenv alone does not
+    stop an agent from reading another interpreter's packages.
+    """
+    if (
+        Path("/opt/robocode-mcp").exists()
+        or not Path("/opt/robocode-render/strict_server.py").is_file()
+    ):
+        raise RuntimeError("Rebuild the strict image: legacy MCP environment is unsafe")
+    roots = [Path("/opt"), Path("/usr/lib"), Path("/usr/local/lib")]
+    package_dirs = [
+        directory
+        for root in roots
+        for pattern in ("**/site-packages", "**/dist-packages")
+        for directory in root.glob(pattern)
+        if directory.is_dir()
+    ]
+    unexpected = {
+        module.name
+        for module in pkgutil.iter_modules([str(path) for path in package_dirs])
+        if module.name not in {"numpy", "scipy"}
+    }
+    if unexpected:
+        raise RuntimeError(f"Unexpected strict-image packages: {sorted(unexpected)}")
+
+
+def main() -> None:
+    """Start local relays only after verifying isolation, then supervise the CLI."""
+    verify_namespace()
+    config: dict[str, Any] = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))
+    if config.get("strict_blackbox"):
+        verify_strict_runtime()
+    with ExitStack() as stack:
+        for listener in config["listeners"]:
+            server = stack.enter_context(TCPRelay(listener["port"], listener["socket"]))
+            thread = threading.Thread(target=server.serve_forever, daemon=True)
+            thread.start()
+            stack.callback(thread.join, 5)
+            stack.callback(server.shutdown)
+        child = subprocess.Popen(sys.argv[2:])  # pylint: disable=consider-using-with
+        signal.signal(signal.SIGTERM, lambda *_: child.terminate())
+        try:
+            code = child.wait()
+        finally:
+            if child.poll() is None:
+                child.kill()
+                child.wait()
+        raise SystemExit(code)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/approaches/test_llm_genplan_approach.py b/tests/approaches/test_llm_genplan_approach.py
index fb4d1573..50de68ca 100644
--- a/tests/approaches/test_llm_genplan_approach.py
+++ b/tests/approaches/test_llm_genplan_approach.py
@@ -600,3 +600,19 @@ def test_generalized_source_includes_underlying_mechanics(module_name, class_nam
             assert Path(path).read_text(encoding="utf-8") in source
     finally:
         env.close()
+
+
+def test_apptainer_rejected_before_generation(tmp_path):
+    """An unsupported transport is a configuration error, never a host fallback."""
+    env = _ToyEnv()
+    with pytest.raises(ValueError, match="does not support the isolated Apptainer"):
+        LLMGenPlanApproach(
+            action_space=env.action_space,
+            observation_space=env.observation_space,
+            seed=0,
+            primitives={},
+            completion=DictConfig({"provider": "cli"}),
+            container_backend="apptainer",
+            output_dir=str(tmp_path),
+        )
+    assert not list(tmp_path.iterdir())
diff --git a/tests/utils/test_apptainer_sandbox.py b/tests/utils/test_apptainer_sandbox.py
index d079881d..afa827f9 100644
--- a/tests/utils/test_apptainer_sandbox.py
+++ b/tests/utils/test_apptainer_sandbox.py
@@ -7,8 +7,8 @@
 """
 
 import asyncio
+import json
 import subprocess
-from contextlib import nullcontext
 from pathlib import Path
 
 import pytest
@@ -18,18 +18,18 @@
 from robocode.utils.apptainer_sandbox import (
     APPTAINER_PYTHON,
     ApptainerSandboxConfig,
-    _build_apptainer_auth_args,
     _build_apptainer_cmd,
+    _isolated_transport,
     run_agent_in_apptainer_sandbox,
-    run_genplan_in_apptainer,
     sif_path_for,
 )
 from robocode.utils.backends import create_backend
 from robocode.utils.docker_sandbox import (
     DOCKER_PYTHON,
-    GENPLAN_CONTAINER_TIMEOUT_S,
     _find_repo_root,
 )
+from robocode.utils.isolated_transport import UnixRelay
+from robocode.utils.model_broker import BrokerUpstream
 from robocode.utils.strict_blackbox import (
     STRICT_BLACKBOX_PYTHON,
 )
@@ -81,8 +81,6 @@ def test_build_cmd_strict_has_no_project_mounts(tmp_path: Path) -> None:
         src_abs=None,
         kindergarden_abs=None,
         kinder_baselines_abs=None,
-        auth_args=[],
-        firewall_domains=[],
         agent_cmd=["claude"],
     )
     joined = " ".join(cmd)
@@ -90,7 +88,7 @@ def test_build_cmd_strict_has_no_project_mounts(tmp_path: Path) -> None:
     assert str(config.sif_path) not in cmd
     assert "/host/sandbox:/sandbox" in cmd
     assert "--containall" in cmd
-    assert "ROBOCODE_SKIP_FIREWALL=1" in cmd
+    assert not any("ROBOCODE_SKIP_FIREWALL" in arg for arg in cmd)
     assert "/robocode/src" not in joined
     assert "kindergarden" not in joined
     assert "ss-pybullet" not in joined
@@ -104,10 +102,14 @@ class _Launched(Exception):
 def test_strict_run_uses_only_clean_interpreter(  # type: ignore
     tmp_path: Path, monkeypatch
 ) -> None:
-    """The agent's scripts use the strict venv and the render proxy its own."""
+    """Agent scripts and render tools share only the strict numerical dependencies."""
     strict_sif_path = tmp_path / "robocode-strict-blackbox.sif"
     strict_sif_path.touch()
     sandbox_dir = tmp_path / "run" / "sandbox"
+    metadata_path = tmp_path / "env_spaces.json"
+    metadata_path.write_text(
+        json.dumps({"host": "attacker.invalid", "port": 9999}), encoding="utf-8"
+    )
     config = ApptainerSandboxConfig(
         sandbox_dir=sandbox_dir,
         sif_path=tmp_path / "robocode-sandbox.sif",
@@ -117,17 +119,30 @@ def test_strict_run_uses_only_clean_interpreter(  # type: ignore
         mcp_tools=("render_state",),
         prompt="hello",
         output_filename="approach.py",
+        env_server_port=12345,
+        init_files={"env_spaces.json": metadata_path},
     )
     monkeypatch.setattr(
-        "robocode.utils.apptainer_sandbox._build_apptainer_auth_args",
-        lambda _backend: nullcontext(([], {})),
+        "robocode.utils.apptainer_sandbox.load_broker_upstream",
+        lambda _: BrokerUpstream("messages", "api.anthropic.com", "/v1", {}),
     )
+    targets: list[tuple[str, int]] = []
+
+    def capture_relay(path: str, target: tuple[str, int]) -> UnixRelay:
+        targets.append(target)
+        return UnixRelay(path, target)
+
+    monkeypatch.setattr("robocode.utils.apptainer_sandbox.UnixRelay", capture_relay)
+    monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN", "host-only-test-secret")
     launched: list[list[str]] = []
     real_popen = subprocess.Popen
 
     def fake_popen(cmd: list[str], **kwargs):  # type: ignore
         if cmd[0] != "apptainer":  # the sandbox's own git commands
             return real_popen(cmd, **kwargs)
+        assert kwargs["start_new_session"] is True
+        assert "host-only-test-secret" not in str(kwargs)
+        assert ".credentials.json" not in " ".join(cmd)
         launched.append(cmd)
         raise _Launched
 
@@ -136,13 +151,16 @@ def fake_popen(cmd: list[str], **kwargs):  # type: ignore
     with pytest.raises(_Launched):
         asyncio.run(run_agent_in_apptainer_sandbox(config, backend))
 
+    assert targets == [("127.0.0.1", 12345)]
+    metadata = json.loads((sandbox_dir / "env_spaces.json").read_text(encoding="utf-8"))
+    assert (metadata["host"], metadata["port"]) == ("127.0.0.1", 18081)
     assert len(launched) == 1
     cmd = launched[0]
     joined = " ".join(cmd)
     assert str(strict_sif_path) in cmd
     assert "/robocode/src" not in joined
-    # The render-server probe and CLAUDE.md name the strict interpreter; the MCP
-    # start script the render proxy's separate one.
+    # Agent scripts, the render server, and the startup probe use the same
+    # dependency-clean interpreter.
     assert f"{STRICT_BLACKBOX_PYTHON} -c" in joined
     assert STRICT_BLACKBOX_PYTHON in (sandbox_dir / "CLAUDE.md").read_text()
     start_script = (sandbox_dir / ".mcp" / MCP_START_SCRIPT).read_text()
@@ -167,8 +185,6 @@ def test_build_cmd_basic_shape(tmp_path: Path) -> None:
         src_abs="/host/src",
         kindergarden_abs="/host/kindergarden",
         kinder_baselines_abs=None,
-        auth_args=[],
-        firewall_domains=[],
         agent_cmd=["claude", "--print", "hello"],
     )
 
@@ -188,8 +204,8 @@ def test_build_cmd_basic_shape(tmp_path: Path) -> None:
     # Env vars are passed as `--env KEY=val` pairs.
     assert "CLAUDE_CODE_MAX_OUTPUT_TOKENS=8192" in cmd
     assert "CLAUDE_AUTOCOMPACT_PCT_OVERRIDE=70" in cmd
-    # init-firewall.sh is skipped (apptainer can't grant CAP_NET_ADMIN).
-    assert "ROBOCODE_SKIP_FIREWALL=1" in cmd
+    # Apptainer uses the isolated namespace, not the Docker firewall entrypoint.
+    assert not any("ROBOCODE_SKIP_FIREWALL" in arg for arg in cmd)
     # Headless container has no GPU: mujoco's Dynamic3D renderer must use OSMesa
     # (software), so the sandbox forces it; EGL device displays would crash.
     assert "MUJOCO_GL=osmesa" in cmd
@@ -202,7 +218,11 @@ def test_build_cmd_basic_shape(tmp_path: Path) -> None:
 
     # SIF path appears before the entrypoint invocation.
     sif_idx = cmd.index(str(config.sif_path))
-    entrypoint_idx = cmd.index("/usr/local/bin/entrypoint.sh")
+    entrypoint_idx = cmd.index("/usr/bin/setpriv")
+    assert "--net" in cmd
+    assert cmd[cmd.index("--network") + 1] == "none"
+    assert "--userns" in cmd
+    assert "/usr/local/bin/entrypoint.sh" not in cmd
     assert sif_idx < entrypoint_idx
 
     # Agent command is appended at the end.
@@ -210,7 +230,7 @@ def test_build_cmd_basic_shape(tmp_path: Path) -> None:
 
 
 def test_build_cmd_bilevel_conditional(tmp_path: Path) -> None:
-    """The kinder-baselines bind and --extra bilevel sync appear only when requested."""
+    """Bilevel source is conditional; dependency installation is a separate phase."""
 
     def build(kinder_baselines_abs: str | None) -> list[str]:
         return _build_apptainer_cmd(
@@ -219,8 +239,6 @@ def build(kinder_baselines_abs: str | None) -> list[str]:
             src_abs="/host/src",
             kindergarden_abs="/host/kindergarden",
             kinder_baselines_abs=kinder_baselines_abs,
-            auth_args=[],
-            firewall_domains=[],
             agent_cmd=["claude"],
         )
 
@@ -230,7 +248,7 @@ def build(kinder_baselines_abs: str | None) -> list[str]:
 
     on = build("/host/kinder-baselines")
     assert "/host/kinder-baselines:/robocode/third-party/kinder-baselines" in on
-    assert "ROBOCODE_UV_EXTRA_ARGS=--extra bilevel" in on
+    assert not any("ROBOCODE_UV_EXTRA_ARGS" in arg for arg in on)
 
 
 def test_build_cmd_always_adds_containall(tmp_path: Path) -> None:
@@ -247,8 +265,6 @@ def test_build_cmd_always_adds_containall(tmp_path: Path) -> None:
         src_abs="/host/src",
         kindergarden_abs="/host/kindergarden",
         kinder_baselines_abs=None,
-        auth_args=[],
-        firewall_domains=[],
         agent_cmd=["claude"],
     )
     default_cmd = _build_apptainer_cmd(
@@ -257,8 +273,6 @@ def test_build_cmd_always_adds_containall(tmp_path: Path) -> None:
         src_abs="/host/src",
         kindergarden_abs="/host/kindergarden",
         kinder_baselines_abs=None,
-        auth_args=[],
-        firewall_domains=[],
         agent_cmd=["claude"],
     )
     assert "--containall" in blackbox_cmd
@@ -268,154 +282,29 @@ def test_build_cmd_always_adds_containall(tmp_path: Path) -> None:
     assert "--pid" in default_cmd
 
 
-def test_genplan_cmd_adds_containall(
-    tmp_path: Path, monkeypatch  # type: ignore
-) -> None:
-    """GenPlan gets the same default-bind isolation as the agentic path."""
-    sandbox_dir = tmp_path / "sandbox"
-    sandbox_dir.mkdir()
-    sif_path = tmp_path / "robocode-sandbox.sif"
-    sif_path.touch()
-    filtered_src = tmp_path / "src"
-    filtered_kindergarden = tmp_path / "kindergarden"
-    filtered_src.mkdir()
-    filtered_kindergarden.mkdir()
-
-    monkeypatch.setattr(
-        "robocode.utils.apptainer_sandbox._filtered_repo_mounts",
-        lambda **_kwargs: nullcontext(
-            (filtered_src, filtered_kindergarden, None, None)
-        ),
+def test_metadata_cannot_select_an_environment_destination(tmp_path: Path) -> None:
+    """Only the explicit host config can authorize a relay, including on resume."""
+    (tmp_path / "env_spaces.json").write_text(
+        json.dumps({"host": "127.0.0.1", "port": 9999}), encoding="utf-8"
     )
-    monkeypatch.setattr(
-        "robocode.utils.apptainer_sandbox._build_apptainer_auth_args",
-        lambda _backend: nullcontext(([], {})),
+    config = ApptainerSandboxConfig(sandbox_dir=tmp_path)
+    upstream = BrokerUpstream("messages", "api.anthropic.com", "/v1", {})
+    with pytest.raises(RuntimeError, match="explicit trusted env_server_port"):
+        with _isolated_transport(config, upstream):
+            pytest.fail("Untrusted metadata enabled a host relay")
+
+
+def test_unsupported_backend_never_sets_up_an_agent(tmp_path, monkeypatch) -> None:
+    """Removing the old OpenCode auth path cannot cause an unbrokered fallback."""
+    image = tmp_path / "image.sif"
+    image.touch()
+    config = ApptainerSandboxConfig(sandbox_dir=tmp_path / "sandbox", sif_path=image)
+    backend = create_backend(
+        DictConfig({"backend": "opencode", "model": "openai/test"})
     )
     monkeypatch.setattr(
-        "robocode.utils.apptainer_sandbox.firewall_domains_for_provider",
-        lambda *_args: [],
+        "robocode.utils.apptainer_sandbox._setup_sandbox_dir",
+        lambda _: pytest.fail("Unsupported backend reached agent setup"),
     )
-    calls: list[list[str]] = []
-
-    timeouts: list[float] = []
-
-    def fake_run(cmd: list[str], **kwargs) -> None:
-        calls.append(cmd)
-        timeouts.append(kwargs["timeout"])
-
-    monkeypatch.setattr("robocode.utils.apptainer_sandbox.subprocess.run", fake_run)
-
-    run_genplan_in_apptainer(
-        sandbox_dir,
-        {"provider": "cli"},
-        sif_path=sif_path,
-    )
-
-    assert len(calls) == 1
-    assert timeouts == [GENPLAN_CONTAINER_TIMEOUT_S]
-    assert calls[0][:3] == ["apptainer", "exec", "--containall"]
-    assert "--pid" in calls[0]
-
-
-def test_build_cmd_firewall_domains(tmp_path: Path) -> None:
-    """Firewall domains, when present, are forwarded via --env."""
-    config = ApptainerSandboxConfig(sandbox_dir=tmp_path / "sandbox")
-    cmd = _build_apptainer_cmd(
-        config,
-        sandbox_abs="/host/sandbox",
-        src_abs="/host/src",
-        kindergarden_abs="/host/kindergarden",
-        kinder_baselines_abs=None,
-        auth_args=[],
-        firewall_domains=["api.example.com", "cdn.example.com"],
-        agent_cmd=["claude"],
-    )
-    assert "ROBOCODE_FIREWALL_EXTRA_DOMAINS=api.example.com,cdn.example.com" in cmd
-
-
-def test_build_cmd_no_firewall_when_empty(tmp_path: Path) -> None:
-    """When no extra domains are requested, the env var is not added."""
-    config = ApptainerSandboxConfig(sandbox_dir=tmp_path / "sandbox")
-    cmd = _build_apptainer_cmd(
-        config,
-        sandbox_abs="/host/sandbox",
-        src_abs="/host/src",
-        kindergarden_abs="/host/kindergarden",
-        kinder_baselines_abs=None,
-        auth_args=[],
-        firewall_domains=[],
-        agent_cmd=["claude"],
-    )
-    assert not any("ROBOCODE_FIREWALL_EXTRA_DOMAINS" in arg for arg in cmd)
-
-
-def test_build_cmd_auth_args_inserted(tmp_path: Path) -> None:
-    """Caller-supplied auth args (e.g. a --bind) appear in the cmd."""
-    config = ApptainerSandboxConfig(sandbox_dir=tmp_path / "sandbox")
-    auth_args = ["--bind", "/home/u/.claude:/home/node/.claude"]
-    cmd = _build_apptainer_cmd(
-        config,
-        sandbox_abs="/host/sandbox",
-        src_abs="/host/src",
-        kindergarden_abs="/host/kindergarden",
-        kinder_baselines_abs=None,
-        auth_args=auth_args,
-        firewall_domains=[],
-        agent_cmd=["claude"],
-    )
-    assert "/home/u/.claude:/home/node/.claude" in cmd
-
-
-def test_opencode_auth_passes_api_keys(monkeypatch) -> None:  # type: ignore
-    """Provider API keys are forwarded via APPTAINERENV_ env vars, not argv."""
-    monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test-value")
-    with _build_apptainer_auth_args("opencode") as (args, env):
-        assert env.get("APPTAINERENV_ANTHROPIC_API_KEY") == "sk-test-value"
-        # The secret must not appear on the command line.
-        assert not any("sk-test-value" in a for a in args)
-
-
-def test_codex_auth_passes_codex_api_key(monkeypatch) -> None:  # type: ignore
-    """Forward the Codex key through the container environment."""
-    monkeypatch.setenv("CODEX_API_KEY", "sk-test-value")
-
-    with _build_apptainer_auth_args("codex") as (args, env):
-        assert not args
-        assert env == {"APPTAINERENV_CODEX_API_KEY": "sk-test-value"}
-
-
-def test_claude_auth_uses_env_token(monkeypatch) -> None:  # type: ignore
-    """CLAUDE_CODE_OAUTH_TOKEN is forwarded via APPTAINERENV_, never on argv."""
-    monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN", "sk-ant-oat01-test")
-    with _build_apptainer_auth_args("claude") as (args, env):
-        assert env.get("APPTAINERENV_CLAUDE_CODE_OAUTH_TOKEN") == "sk-ant-oat01-test"
-        # The token must not appear on the command line (visible via `ps`).
-        assert not any("sk-ant-oat01-test" in a for a in args)
-        assert not any("--bind" in a for a in args)
-
-
-def test_claude_auth_binds_credentials_only(  # type: ignore
-    tmp_path: Path, monkeypatch
-) -> None:
-    """The fallback mount is a throwaway credentials-only copy."""
-    monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False)
-    # Force the resolver to report no token (avoid Keychain hit on dev macOS).
-    monkeypatch.setattr(
-        "robocode.utils.apptainer_sandbox._get_claude_oauth_token",
-        lambda: None,
-    )
-    host = tmp_path / ".claude"
-    (host / "projects").mkdir(parents=True)
-    (host / "projects" / "past.jsonl").write_text("past")
-    (host / ".credentials.json").write_text("credentials")
-    monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(host))
-
-    with _build_apptainer_auth_args("claude") as (args, env):
-        assert not env
-        bind = next(arg for arg in args if arg.endswith(":/home/node/.claude"))
-        mounted = Path(bind.split(":", 1)[0])
-        assert mounted != host
-        assert [path.name for path in mounted.iterdir()] == [".credentials.json"]
-        copied = mounted
-
-    assert not copied.exists()
+    with pytest.raises(RuntimeError, match="does not support 'opencode'"):
+        asyncio.run(run_agent_in_apptainer_sandbox(config, backend))
diff --git a/tests/utils/test_isolated_transport.py b/tests/utils/test_isolated_transport.py
new file mode 100644
index 00000000..4046960c
--- /dev/null
+++ b/tests/utils/test_isolated_transport.py
@@ -0,0 +1,91 @@
+"""The agent must never start when namespace setup or privilege dropping fails."""
+
+# pylint: disable=redefined-outer-name
+
+import pytest
+
+from robocode.utils.apptainer_environment import clean_apptainer_env
+from robocode.utils.isolated_transport import verify_namespace, verify_strict_runtime
+
+
+@pytest.fixture
+def namespace(monkeypatch):
+    """Provide a kernel snapshot that models the required private namespace."""
+    state = {
+        "uid": 1013,
+        "interfaces": [(1, "lo")],
+        "route": "header\n",
+        "caps": "0",
+        "nnp": "1",
+    }
+    monkeypatch.setattr(
+        "robocode.utils.isolated_transport.os.getuid", lambda: state["uid"]
+    )
+    monkeypatch.setattr(
+        "robocode.utils.isolated_transport.socket.if_nameindex",
+        lambda: state["interfaces"],
+    )
+
+    def read(path, **_kwargs):
+        if str(path) == "/proc/net/route":
+            return state["route"]
+        return (
+            "\n".join(
+                f"{key}: {state['caps']}"
+                for key in ("CapEff", "CapPrm", "CapBnd", "CapInh", "CapAmb")
+            )
+            + f"\nNoNewPrivs: {state['nnp']}\n"
+        )
+
+    monkeypatch.setattr("robocode.utils.isolated_transport.Path.read_text", read)
+    return state
+
+
+@pytest.mark.usefixtures("namespace")
+def test_private_namespace_passes():
+    """The required kernel state allows the supervisor to proceed."""
+    verify_namespace()
+
+
+@pytest.mark.parametrize(
+    "key,value",
+    [
+        ("uid", 0),
+        ("interfaces", [(1, "lo"), (2, "eth0")]),
+        ("route", "header\nroute\n"),
+        ("caps", "1000"),
+        ("nnp", "0"),
+    ],
+)
+def test_bad_namespace_cannot_fall_back(namespace, key, value):
+    """A failed invariant aborts before any agent or relay is started."""
+    namespace[key] = value
+    with pytest.raises(RuntimeError):
+        verify_namespace()
+
+
+def test_child_environment_excludes_credentials_and_override_flags(monkeypatch):
+    """Host auth and Apptainer special variables cannot leak to the child."""
+    for key in (
+        "OPENAI_API_KEY",
+        "CODEX_API_KEY",
+        "CLAUDE_CODE_OAUTH_TOKEN",
+        "ANTHROPIC_API_KEY",
+        "APPTAINERENV_OPENAI_API_KEY",
+        "APPTAINER_BINDPATH",
+        "SINGULARITY_BINDPATH",
+        "HTTPS_PROXY",
+        "LD_PRELOAD",
+    ):
+        monkeypatch.setenv(key, "secret-or-override")
+    env = clean_apptainer_env()
+    assert "secret-or-override" not in env.values()
+
+
+def test_strict_runtime_rejects_old_image(monkeypatch):
+    """The previous image cannot silently remain in use after upgrading the runner."""
+    monkeypatch.setattr(
+        "pathlib.Path.exists", lambda self: str(self) == "/opt/robocode-mcp"
+    )
+    with pytest.raises(RuntimeError, match="legacy MCP"):
+        verify_strict_runtime()