Support restricted shared agent runtimes

tests/v1/test_e2e.py

@@ -406,8 +406,12 @@ async def test_env_id_agentic_judge(run_v1, tmp_path):
         harness=None,  # seats pin their own harness; there is no run-level one
         env={
             "id": "agentic-judge",
-            # The solver owns the shared box, so the container is pinned here.
-            "solver": {"harness": {"id": "bash"}, "runtime": {"type": "docker"}},
+            # The solver owns the shared box. The blocklist makes it a restricted
+            # runtime; each sequential agent gets trusted setup before its own cut.
+            "solver": {
+                "harness": {"id": "bash"},
+                "runtime": {"type": "docker", "block": ["example.com"]},
+            },
             # The judge reads the trace and reasons before it writes the
             # verdict file; the shared 2048-token run cap truncates it mid-audit.
             "judge": {

verifiers/v1/envs/agentic_judge/env.py

@@ -150,6 +150,8 @@ def from_trace(cls, solution: vf.Trace, config: "JudgeTaskConfig") -> "JudgeTask
                 image=solved.image,
                 workdir=solved.workdir,
                 resources=solved.resources,
+                network_allow=solved.network_allow,
+                network_block=solved.network_block,
             ),
             files=files,
         )
@@ -300,6 +302,10 @@ async def setup(self, agents: vf.Agents) -> None:
     async def run(self, task: vf.Task, agents: vf.Agents) -> None:
         async with agents.solver.provision(task) as box:
             solution = await agents.solver.run(task, runtime=box)
+            if box.network_restricted and not box.execution_prepared:
+                raise RuntimeError(
+                    "solver failed before the shared runtime's network policy activated"
+                )
             judge_task = JudgeTask.from_trace(solution, self.config.task)
             await agents.judge.run(judge_task, runtime=box)
 

verifiers/v1/rollout.py

@@ -157,6 +157,7 @@ def __init__(
             ctx, self.trace, discover_decorated(task, "stop"), limits or RolloutLimits()
         )
         self._stack = AsyncExitStack()
+        self._runtime_stack = AsyncExitStack()
         self._failed = False
         self._failure: Exception | None = None
         self._opened = False
@@ -239,7 +240,7 @@ async def open(self) -> bool:
                 )
             if self._owns_runtime:
                 await runtime.start()
-            await runtime.prepare_setup()
+            await self._runtime_stack.enter_async_context(runtime.rollout())
             now = time.time()
             self.trace.timing.boot.end = now
             self.trace.timing.setup.start = now
@@ -376,6 +377,8 @@ async def abort(self) -> None:
         if self.runtime is not None:
             with contextlib.suppress(Exception):
                 await self.harness.cleanup(self.trace, self.runtime)
+        with contextlib.suppress(Exception):
+            await self._runtime_stack.aclose()
         if self._owns_runtime and self.runtime is not None:
             with contextlib.suppress(Exception):
                 await self.runtime.stop()
@@ -441,6 +444,7 @@ async def close(self) -> Trace:
                     logger.warning(
                         "harness cleanup failed (rollout %s)", trace.id, exc_info=True
                     )
+            await self._runtime_stack.aclose()
             # Tear down here — the env's `score()` (later) needs only the traces,
             # not a live runtime. A borrowed runtime is its creator's to tear down,
             # not this rollout's.

verifiers/v1/runtimes/base.py

@@ -9,6 +9,7 @@
 import uuid
 import weakref
 from abc import ABC, abstractmethod
+from collections.abc import AsyncIterator
 from dataclasses import dataclass
 from pathlib import PurePosixPath
 from typing import ClassVar, Self
@@ -40,6 +41,57 @@
     f"|| {{ {_INSTALL_CURL}; {_DOWNLOAD_UV}; }}"
 )
 
+# A reused restricted box must stop processes and discard framework setup caches
+# left by the prior untrusted agent before trusted setup reopens egress.
+# PID + kernel start time survives PID reuse.
+_PROCESS_SNAPSHOT = r"""
+for proc in /proc/[0-9]*; do
+    pid=${proc##*/}
+    stat=$(cat "$proc/stat" 2>/dev/null) || continue
+    rest=${stat##*) }
+    set -- $rest
+    printf '%s:%s\n' "$pid" "${20}"
+done
+"""
+_KILL_AFTER_SNAPSHOT = r"""
+baseline="
+$VF_PROCESS_BASELINE
+"
+protected=" "
+pid=$$
+while [ "$pid" -gt 1 ] 2>/dev/null; do
+    protected="$protected$pid "
+    stat=$(cat "/proc/$pid/stat" 2>/dev/null) || break
+    rest=${stat##*) }
+    set -- $rest
+    pid=$2
+done
+while :; do
+    targeted=
+    for proc in /proc/[0-9]*; do
+        pid=${proc##*/}
+        case "$protected" in *" $pid "*) continue ;; esac
+        stat=$(cat "$proc/stat" 2>/dev/null) || continue
+        rest=${stat##*) }
+        set -- $rest
+        case "$1" in Z|X) continue ;; esac
+        if [ "$1" = D ]; then
+            echo "process $pid is stuck in uninterruptible sleep" >&2
+            exit 1
+        fi
+        identity="$pid:${20}"
+        case "$baseline" in *"
+$identity
+"*) continue ;; esac
+        if kill -STOP "$pid" 2>/dev/null; then
+            kill -KILL "$pid" 2>/dev/null || true
+            targeted=1
+        fi
+    done
+    [ -z "$targeted" ] && break
+done
+"""
+
 # The single port a self-publishing runtime (modal/prime) forwards to a public URL for a server
 # hosted in its sandbox. A server placed in such a runtime binds this (on 0.0.0.0) and is reached
 # at the runtime's public URL.
@@ -143,6 +195,9 @@ def __init__(self, name: str | None = None) -> None:
         self._uv_interpreters: dict[str, str] = {}
         self._uv_script_locks: dict[str, asyncio.Lock] = {}
         self._setup_claimed = False
+        self._process_baseline: str | None = None
+        self.execution_prepared = False
+        """Whether a rollout successfully activated this runtime's execution policy."""
         self.stopped = False
         """Whether teardown has begun (set by `stop`). A stopped runtime is dead: a rollout
         refuses to borrow one — the owner tore it down, so any use is a lifetime bug in the
@@ -272,20 +327,70 @@ def host_url(self, url: str) -> str:
         """The URL a program inside this runtime uses to reach a host-bound `url`."""
         return url
 
-    async def prepare_setup(self) -> None:
-        """Claim the runtime for trusted setup; restricted runtimes may reject reuse."""
+    @contextlib.asynccontextmanager
+    async def rollout(self) -> AsyncIterator[None]:
+        """Claim one restricted rollout, reopening trusted setup only between runs."""
         if not self.network_restricted:
+            yield
             return
         if self._setup_claimed:
             raise SandboxError(
-                f"network-filtered {self.type} runtimes are single-rollout; "
-                "provision a fresh runtime instead of reusing this one"
+                f"network-filtered {self.type} runtime {self.name!r} is already in "
+                "use; wait for its rollout to finish before reusing it"
             )
         self._setup_claimed = True
+        try:
+            if self._process_baseline is not None:
+                result = await self.run(
+                    ["sh", "-c", _KILL_AFTER_SNAPSHOT],
+                    {"VF_PROCESS_BASELINE": self._process_baseline},
+                )
+                if result.exit_code != 0:
+                    raise SandboxError(
+                        f"failed to isolate reused {self.type} runtime processes: "
+                        f"{result.stderr.strip()[-500:]}"
+                    )
+                uv_envs = [
+                    str(PurePosixPath(path).parent.parent)
+                    for path in self._uv_interpreters.values()
+                ]
+                reset = await self.run(
+                    ["sh", "-c", 'rm -rf "$@" /tmp/vf-*', "reset", *uv_envs],
+                    {},
+                )
+                if reset.exit_code != 0:
+                    raise SandboxError(
+                        f"failed to reset reused {self.type} runtime setup caches: "
+                        f"{reset.stderr.strip()[-500:]}"
+                    )
+                self._uv_interpreters.clear()
+            snapshot = await self.run(["sh", "-c", _PROCESS_SNAPSHOT], {})
+            if snapshot.exit_code != 0:
+                raise SandboxError(
+                    f"failed to snapshot {self.type} runtime processes: "
+                    f"{snapshot.stderr.strip()[-500:]}"
+                )
+            self._process_baseline = snapshot.stdout
+            if self.execution_prepared:
+                await self._apply_network_policy(None)
+                self.execution_prepared = False
+            yield
+        finally:
+            self._setup_claimed = False
 
     async def prepare_execution(self, routes: list[str]) -> None:
         """Last setup step, right before the agent starts. Restricted runtimes enforce
         their policy here while keeping the interception and MCP `routes` reachable."""
+        if not self.network_restricted:
+            return
+        await self._apply_network_policy(routes)
+        self.execution_prepared = True
+
+    async def _apply_network_policy(self, routes: list[str] | None) -> None:
+        """Apply execution routes, or restore unrestricted trusted setup for None."""
+        raise NotImplementedError(
+            f"{type(self).__name__} does not support restricted networking"
+        )
 
     @property
     def network_restricted(self) -> bool:

verifiers/v1/runtimes/docker/__init__.py

@@ -233,11 +233,14 @@ def host_url(self, url: str) -> str:
             return url.replace(host, "host.docker.internal", 1)
         return url
 
-    async def prepare_execution(self, routes: list[str]) -> None:
+    async def _apply_network_policy(self, routes: list[str] | None) -> None:
         """Allow the declared framework routes, then leave the proxy as the only route."""
-        if not self.network_restricted:
-            return
         assert self._proxy is not None
+        if routes is None:
+            self._proxy.policy = NetworkPolicy(
+                ["*"], [], [HOST_ALIAS], allow_non_global=True
+            )
+            return
         framework = [
             urlsplit(url)._replace(path="", query="", fragment="").geturl()
             for url in routes
@@ -247,6 +250,8 @@ async def prepare_execution(self, routes: list[str]) -> None:
             self.config.block,
             framework,
         )
+        if self._cut:
+            return
         script = (
             "set -eu; HOST=$1; "
             "PORT=$2; "

verifiers/v1/runtimes/prime.py

@@ -180,12 +180,12 @@ async def start(self) -> None:
         ) as e:  # provisioning failure is one rollout's problem, not the eval's
             raise SandboxError(f"prime sandbox provisioning failed: {e}") from e
 
-    async def prepare_execution(self, routes: list[str]) -> None:
+    async def _apply_network_policy(self, routes: list[str] | None) -> None:
         """Apply the host policy after setup and wait until the platform enforces it."""
-        if not self.network_restricted:
-            return
         try:
-            if self.config.allow == ["*"]:
+            if routes is None:
+                policy = {"allow": ["*"]}
+            elif self.config.allow == ["*"]:
                 policy = {"deny": self.config.block}
             else:
                 hosts = [h for h in (urlsplit(route).hostname for route in routes) if h]
@@ -214,8 +214,8 @@ async def prepare_execution(self, routes: list[str]) -> None:
         logger.info(
             "prime: egress policy applied on sandbox %s (allow=%s block=%s)",
             self.info.id,
-            self.config.allow,
-            self.config.block,
+            policy.get("allow"),
+            policy.get("deny"),
         )
 
     async def run(self, argv: list[str], env: dict[str, str]) -> ProgramResult: