feat(runtime): project process connections through providers

hud/environment/process_guard.py

@@ -191,11 +191,11 @@ def _destination(raw: bytes) -> tuple[str, int] | None:
     return None
 
 
-def _emulate_connect(pid: int, descriptor: int, address: bytes) -> int:
+def _emulate_connect(tgid: int, descriptor: int, address: bytes) -> int:
     pidfd_open = getattr(os, "pidfd_open", None)
     if pidfd_open is None:
         return -errno.ENOSYS
-    pidfd = int(pidfd_open(pid))
+    pidfd = int(pidfd_open(tgid))
     try:
         duplicate = _LIBC.syscall(_PIDFD_GETFD_SYSCALL, pidfd, descriptor, 0)
         if duplicate < 0:
@@ -338,7 +338,7 @@ def _broker(self) -> None:
                         response.error = -errno.EPERM
                     else:
                         result = _emulate_connect(
-                            notification.pid,
+                            process_tgid,
                             notification.data.args[0],
                             address,
                         )

hud/environment/tests/test_process_connection.py

@@ -64,6 +64,17 @@ def _fetch_script(url: str) -> str:
     return f"exec {shlex.quote(sys.executable)} -c {shlex.quote(_fetch_source(url))}"
 
 
+def _threaded_fetch_source(url: str) -> str:
+    return (
+        "import threading,urllib.request;"
+        "result=[];"
+        f"request=urllib.request.Request({url!r});"
+        "thread=threading.Thread(target=lambda:result.append("
+        "urllib.request.urlopen(request,timeout=5).read().decode()));"
+        "thread.start();thread.join();print(result[0])"
+    )
+
+
 def _proxy_fetch_source(url: str) -> str:
     return (
         "import http.client,sys;"
@@ -126,6 +137,15 @@ async def test_only_bound_process_reaches_controller_connection(tmp_path: Path)
             assert completed.returncode == 0
             assert completed.stdout == b"ok\n"
 
+            threaded = await ssh.create_process(
+                f"exec {shlex.quote(sys.executable)} -c "
+                f"{shlex.quote(_threaded_fetch_source(connection.client_url))}",
+                connections=(connection,),
+            )
+            threaded_result = await threaded.wait()
+            assert threaded_result.returncode == 0
+            assert threaded_result.stdout == b"ok\n"
+
             child_source = (
                 "import subprocess,sys;"
                 f"result=subprocess.run([sys.executable,'-c',{_fetch_source(connection.client_url)!r}]);"

hud/environment/tests/test_workspace.py

@@ -929,6 +929,28 @@ async def test_namespace_management_does_not_share_process_connections(
     management.wait_closed.assert_awaited_once_with()
 
 
+def test_process_guard_executes_packaged_file_without_module_reentry() -> None:
+    argv = workspace_mod._guarded_process_argv("/tmp/guard.sock", ["bash", "-lc", "true"])
+
+    assert argv == [
+        sys.executable,
+        str(Path(workspace_mod.__file__).with_name("process_guard.py")),
+        "/tmp/guard.sock",
+        "--",
+        "bash",
+        "-lc",
+        "true",
+    ]
+    result = subprocess.run(
+        [*argv[:2], "--help"],
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    assert result.returncode == 0
+    assert result.stderr == ""
+
+
 @pytest.mark.asyncio
 async def test_namespace_host_only_terminates_a_used_session_holder(
     tmp_path: Path,

hud/environment/workspace.py

@@ -57,6 +57,11 @@
 _INVALID_DWORD = 0xFFFFFFFF
 
 
+def _guarded_process_argv(socket_path: str, argv: Sequence[str]) -> list[str]:
+    guard_path = Path(__file__).with_name("process_guard.py")
+    return [sys.executable, str(guard_path), socket_path, "--", *argv]
+
+
 class _WindowsJob:
     """Windows Job Object which owns a subprocess and all of its descendants."""
 
@@ -1696,15 +1701,13 @@ async def _handle_process(self, process: asyncssh.SSHServerProcess[bytes]) -> No
                     if process.command is not None
                     else ["bash", "-l"]
                 )
-                guarded_command = [
-                    sys.executable,
-                    "-m",
-                    "hud.environment.process_guard",
+                guarded_command = _guarded_process_argv(
                     guard.sandbox_socket,
-                    "--",
-                    *self._drop_argv(),
-                    *shell_command,
-                ]
+                    [
+                        *self._drop_argv(),
+                        *shell_command,
+                    ],
+                )
                 argv = self.bwrap_argv(
                     guarded_command,
                     env=session_env,

hud/eval/runtime/__init__.py

@@ -1,6 +1,6 @@
 """Runtime placement and provider configuration."""
 
-from .compose import ComposeProject
+from .compose import ComposeProject, DockerBindMount
 from .core import (
     Provider,
     Runtime,
@@ -21,6 +21,7 @@
 __all__ = [
     "ComposeProject",
     "DaytonaRuntime",
+    "DockerBindMount",
     "DockerRuntime",
     "HUDRuntime",
     "HostedRuntime",

hud/eval/runtime/compose.py

@@ -11,7 +11,7 @@
 import tarfile
 import tempfile
 from dataclasses import dataclass
-from pathlib import Path
+from pathlib import Path, PurePosixPath
 from typing import TYPE_CHECKING, Any
 
 import yaml
@@ -29,7 +29,7 @@
 from yaml.nodes import MappingNode, Node, ScalarNode, SequenceNode
 
 if TYPE_CHECKING:
-    from collections.abc import Iterator, Mapping
+    from collections.abc import Iterator, Mapping, Sequence
 
 
 _COMPOSE_VARIABLE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
@@ -443,6 +443,47 @@ class ComposeLaunchFiles:
     archive: Path | None
 
 
+@dataclass(frozen=True, slots=True)
+class DockerBindMount:
+    """A provider-owned bind mount injected into a Docker environment."""
+
+    source: Path
+    target: PurePosixPath
+    read_only: bool = True
+
+    def __init__(
+        self,
+        source: str | Path,
+        target: str | PurePosixPath,
+        read_only: bool = True,
+    ) -> None:
+        resolved_source = Path(source)
+        resolved_target = PurePosixPath(target)
+        if not resolved_source.is_absolute():
+            raise ValueError("Docker bind mount source must be absolute")
+        if not resolved_target.is_absolute():
+            raise ValueError("Docker bind mount target must be absolute")
+        if "," in str(resolved_source) or "," in str(resolved_target):
+            raise ValueError("Docker bind mount paths cannot contain commas")
+        object.__setattr__(self, "source", resolved_source)
+        object.__setattr__(self, "target", resolved_target)
+        object.__setattr__(self, "read_only", read_only)
+
+    def docker_argument(self) -> str:
+        argument = f"type=bind,source={self.source},target={self.target}"
+        return f"{argument},readonly" if self.read_only else argument
+
+    def compose_volume(self) -> dict[str, str | bool]:
+        volume: dict[str, str | bool] = {
+            "type": "bind",
+            "source": str(self.source),
+            "target": str(self.target),
+        }
+        if self.read_only:
+            volume["read_only"] = True
+        return volume
+
+
 class ComposeProject(BaseModel):
     """A Compose recipe and the project data it may need at runtime."""
 
@@ -512,6 +553,7 @@ def stage(
         port_service: str = "main",
         seccomp: str | Path,
         service_socket: str | None = None,
+        bind_mounts: Sequence[DockerBindMount] = (),
         env_vars: Mapping[str, str] | None = None,
         cpu: float | None = None,
         memory_mb: int | None = None,
@@ -528,14 +570,17 @@ def stage(
                 "apparmor=unconfined",
             ],
         }
+        volumes = [mount.compose_volume() for mount in bind_mounts]
         if service_socket is not None:
-            main["volumes"] = [
+            volumes.append(
                 {
                     "type": "bind",
                     "source": service_socket,
                     "target": "/media/hud/docker.sock",
                 }
-            ]
+            )
+        if volumes:
+            main["volumes"] = volumes
         if env_vars:
             main["environment"] = dict(env_vars)
         if cpu is not None:

hud/eval/runtime/docker.py

@@ -18,7 +18,7 @@
 from hud.utils.docker import docker as _docker
 from hud.utils.process import create_process_group_exec, finish_output, stream_output
 
-from .compose import ComposeConfig
+from .compose import ComposeConfig, DockerBindMount
 from .core import Runtime, RuntimeConfig, validate_session_id
 
 if TYPE_CHECKING:
@@ -134,12 +134,14 @@ def __init__(
         *,
         port: int = 8765,
         run_args: Sequence[str] = (),
+        bind_mounts: Sequence[DockerBindMount] = (),
         compose_service_socket: str | Path | None = None,
         runtime_config: RuntimeConfig | dict[str, Any] | None = None,
         env_vars: Mapping[str, str] | None = None,
     ) -> None:
         self.port = port
         self.run_args = tuple(run_args)
+        self.bind_mounts = tuple(bind_mounts)
         self.env_vars = dict(env_vars or {})
         self.compose_service_socket = (
             str(Path(compose_service_socket)) if compose_service_socket is not None else None
@@ -204,6 +206,7 @@ async def __call__(self, task: Task) -> AsyncIterator[Runtime]:
                 port_service=port_service,
                 seccomp=_DOCKER_SECCOMP_PROFILE,
                 service_socket=service_socket,
+                bind_mounts=self.bind_mounts,
                 env_vars=self.env_vars,
                 cpu=resources.cpu if resources is not None else None,
                 memory_mb=resources.memory_mb if resources is not None else None,
@@ -296,12 +299,16 @@ async def __call__(self, task: Task) -> AsyncIterator[Runtime]:
         env_args: list[str] = []
         for key, value in self.env_vars.items():
             env_args.extend(("--env", f"{key}={value}"))
+        mount_args: list[str] = []
+        for mount in self.bind_mounts:
+            mount_args.extend(("--mount", mount.docker_argument()))
         out, _ = await _docker(
             "run",
             "--detach",
             *self.run_args,
             *env_args,
             *resource_args,
+            *mount_args,
             *_DOCKER_SECURITY_ARGS,
             "--publish",
             f"127.0.0.1::{self.port}",

hud/eval/tests/test_docker_provider.py

@@ -26,6 +26,7 @@
 import hud.utils.process as process_module
 from hud.eval.runtime import (
     DaytonaRuntime,
+    DockerBindMount,
     DockerRuntime,
     ModalRuntime,
     RuntimeConfig,
@@ -602,6 +603,46 @@ async def test_acquisition_publishes_ephemeral_port_and_removes_container(
     assert capsys.readouterr().out == "ImportError: boom\n"
 
 
+async def test_docker_runtime_injects_provider_bind_mount(
+    tmp_path: Path,
+    docker_log: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _install_fake_docker(tmp_path, port_behavior="echo 127.0.0.1:43210", monkeypatch=monkeypatch)
+    bundle = tmp_path / "agents" / "codex"
+    bundle.mkdir(parents=True)
+
+    provider = DockerRuntime(
+        "img:tag",
+        bind_mounts=(DockerBindMount(bundle, "/usr/local/lib/agents/codex"),),
+    )
+    async with provider(_row()):
+        pass
+
+    assert (await _docker_calls(docker_log))[0] == (
+        f"run --detach --mount type=bind,source={bundle},"
+        "target=/usr/local/lib/agents/codex,readonly "
+        f"{_docker_security_args()} --publish 127.0.0.1::8765 img:tag"
+    )
+
+
+@pytest.mark.parametrize(
+    ("source", "target", "message"),
+    [
+        ("relative", "/opt/agents", "source must be absolute"),
+        ("/opt/agents", "relative", "target must be absolute"),
+        ("/opt/agents,old", "/opt/agents", "paths cannot contain commas"),
+    ],
+)
+def test_docker_bind_mount_rejects_ambiguous_paths(
+    source: str,
+    target: str,
+    message: str,
+) -> None:
+    with pytest.raises(ValueError, match=message):
+        DockerBindMount(source, target)
+
+
 async def test_docker_session_archives_inside_the_container(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -994,6 +1035,53 @@ async def fake_docker(*args: str, **_kwargs: Any) -> tuple[str, str]:
     ]
 
 
+async def test_docker_runtime_stages_provider_mount_with_service_socket(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    compose = tmp_path / "compose.yaml"
+    compose.write_text("services:\n  main:\n    image: hud-env:one\n", encoding="utf-8")
+    bundle = tmp_path / "agents" / "claude"
+    bundle.mkdir(parents=True)
+    rendered: dict[str, Any] = {}
+
+    async def fake_docker(*args: str, **_kwargs: Any) -> tuple[str, str]:
+        if args[-4:] == ("up", "--detach", "--build", "--remove-orphans"):
+            files = [Path(args[index + 1]) for index, value in enumerate(args) if value == "--file"]
+            rendered.update(json.loads(files[1].read_text("utf-8")))
+        if args[-3:] == ("port", "main", "8765"):
+            return "127.0.0.1:43210\n", ""
+        return "", ""
+
+    monkeypatch.setattr(runtime_module, "_docker", fake_docker)
+    task = Task(
+        env="any-env",
+        id="t",
+        runtime_config=RuntimeConfig(compose=ComposeProject(document=compose, service_access=True)),
+    )
+    provider = DockerRuntime(
+        compose_service_socket="/vm/run/docker.sock",
+        bind_mounts=(DockerBindMount(bundle, "/usr/local/lib/agents/claude"),),
+    )
+
+    async with provider(task):
+        pass
+
+    assert rendered["services"]["main"]["volumes"] == [
+        {
+            "type": "bind",
+            "source": str(bundle),
+            "target": "/usr/local/lib/agents/claude",
+            "read_only": True,
+        },
+        {
+            "type": "bind",
+            "source": "/vm/run/docker.sock",
+            "target": "/media/hud/docker.sock",
+        },
+    ]
+
+
 def test_docker_runtime_accepts_only_one_environment_definition(tmp_path: Path) -> None:
     with pytest.raises(ValueError, match="either image or compose"):
         RuntimeConfig(