Run split harnesses on Prime VMs beside the workspace

.github/workflows/tests.yml

@@ -109,6 +109,8 @@ jobs:
             tests: tests/integration/test_native_compose.py
           - behavior: Tool VMs
             tests: tests/integration/test_tool_vms.py
+          - behavior: Split Execution
+            tests: tests/integration/test_split_execution.py
           - behavior: Mounts
             tests: tests/integration/test_mounts.py
           - behavior: Sampling

docs/runtimes.md

@@ -191,6 +191,14 @@ A `Deployment` coordinates all execution resources for a single rollout attempt:
 
 The **workspace** (`deployment.workspace`) is the `Runtime` where task commands run and task files live; setting `agent.deployment.mode = "colocated"` places the harness and colocated tools inside that same workspace.
 
+With `agent.deployment.mode = "split"`, the harness runs on a Prime VM of its own (`agent.deployment.harness`, 1 CPU and 2 GB by default) beside a Prime workspace, and its tools execute in the workspace. The workspace is placed exactly as in colocated mode, including `ServicesSpec` services, Compose services and tool server VMs. The harness's VM joins the rollout's private LAN, where it may open connections to the workspace and to tool server VMs, and nothing may open one to it. It takes the task's network policy plus the model and tool routes; the workspace keeps no route to the model, and never holds its credentials. Tools run in a worker that the harness starts in the workspace through `ExecutionConnection.serve`. The worker listens on the LAN at a port the kernel picks and announces, and carries a `VF_EXEC_ID` marker in its environment that every process its tools start inherits. When the harness session closes, every process carrying the marker is killed, detached ones included, unless it emptied its environment. Each tool call crosses the LAN. Harnesses opt in with `SUPPORTS_SPLIT_EXECUTION` and `setup_execution()`; RLM supports its native tools and builtin skills. Borrowed runtimes and colocated task tools are rejected.
+
+```toml
+[env.agent.deployment]
+mode = "split"
+harness = { cpu = 2, memory = 4 }
+```
+
 On Prime, return a `ServicesSpec` from `deployment_spec()` to provision one native VM per named service beside the task's `main` workspace. Each service has its own image, resources, filesystem, and egress policy. The VMs join a private LAN for the rollout; they can call each other and the workspace through the names in `deployment.hosts`. Services bind to `0.0.0.0`, since each VM has its own loopback interface. No Docker daemon runs inside these VMs.
 
 Task setup starts processes with `deployment.target(name).run_background()` and waits for readiness before returning. Image entrypoints do not start services automatically. Pass process environment variables to `run_background`; the workspace's environment is not copied into service VMs. For example:

tests/integration/fixtures/split_execution_v1.py

@@ -0,0 +1,42 @@
+"""Check that a split harness's tools act on the workspace, not the harness's VM."""
+
+import json
+
+import verifiers.v1 as vf
+
+
+class SplitExecutionTask(vf.Task):
+    def runtime_env(self) -> dict[str, str]:
+        return {"VF_WORKSPACE_PROBE": "workspace-value"}
+
+    async def setup(self, runtime: vf.Runtime) -> None:
+        await runtime.write("input.txt", b"workspace-only-input")
+
+    @vf.reward
+    async def workspace_files(self, runtime: vf.Runtime, trace: vf.Trace) -> float:
+        report = json.loads((await runtime.read("answer.json")).decode())
+        patched = (await runtime.read("patched.txt")).decode().strip()
+        trace.info["workspace_report"] = report
+        return float(
+            report
+            == {
+                "cwd": runtime.config.workdir,
+                "input": "workspace-only-input",
+                "env": "workspace-value",
+            }
+            and patched == "patched-remotely"
+        )
+
+
+class SplitExecutionTaskset(vf.Taskset[SplitExecutionTask, vf.TasksetConfig]):
+    def load(self) -> list[SplitExecutionTask]:
+        prompt = (
+            "Use your shell tool to read input.txt, your current directory, and the "
+            "VF_WORKSPACE_PROBE environment variable. Write these actual values to "
+            "answer.json with keys input, cwd, env. Use your file editing tool to "
+            "create patched.txt containing patched-remotely. Then finish."
+        )
+        return [SplitExecutionTask(vf.TaskData(idx=0, prompt=prompt), self.config.task)]
+
+
+__all__ = ["SplitExecutionTaskset"]

tests/integration/grid.py

@@ -10,8 +10,8 @@
 the code.
 
 A harness without a capability the behavior needs is collected as a skip with its
-reason (`Harness.EXECUTES_CODE`, `SUPPORTS_MCP`, `SUPPORTS_SKILLS`, resume support via
-`SUPPORTS_RESUME` or a native session).
+reason (`Harness.EXECUTES_CODE`, `SUPPORTS_MCP`, `SUPPORTS_SKILLS`,
+`SUPPORTS_SPLIT_EXECUTION`, resume support via `SUPPORTS_RESUME` or a native session).
 
 A runtime without a feature the behavior needs (`RuntimeBackend.SUPPORTS_*`, or a
 `NetworkPolicyConfig` for a network policy) skips the cell the same way. That a runtime
@@ -65,6 +65,7 @@
     "expose": "port exposure",
     "compose": "Compose services",
     "services": "native service VMs",
+    "split": "split deployment",
 }
 
 # Harness dialects that cannot run against the CI model yet.
@@ -137,7 +138,7 @@ def supports(runtime: str, feature: str) -> bool:
     if feature == "network":
         config = TypeAdapter(RuntimeConfig).validate_python(runtime_config(runtime))
         return isinstance(config, NetworkPolicyConfig)
-    if feature == "services":
+    if feature in ("services", "split"):
         return issubclass(runtime_cls(runtime), PrimeBackend)
     flag = {
         "container": "SUPPORTS_TASK_IMAGES",
@@ -173,6 +174,8 @@ def unsupported(
         return f"{name} has no native skill support"
     if "resume" in needs and not resumes(cls):
         return f"{name} cannot resume an exchange"
+    if "split" in needs and not cls.SUPPORTS_SPLIT_EXECUTION:
+        return f"{name} does not support split deployment"
     runtime_needs = (*(("container",) if cls.NEEDS_CONTAINER else ()), *needs)
     for feature in runtime_needs:
         if feature in RUNTIME_FEATURES and not supports(runtime, feature):

tests/integration/test_split_execution.py

@@ -0,0 +1,59 @@
+"""Split deployment: the harness on a Prime VM of its own, its tools in the workspace,
+for every harness that declares `SUPPORTS_SPLIT_EXECUTION`."""
+
+import pytest
+from grid import grid
+
+import verifiers.v1 as vf
+from verifiers.v1.rollout import Rollout
+
+SPLIT = {
+    "deployment": {"mode": "split"},
+    "timeout": {"setup": 900, "rollout": 300, "scoring": 60},
+}
+
+
+@pytest.mark.parametrize("entrypoint", ["run", "interaction"])
+async def test_split_preflight_before_rollout(monkeypatch, entrypoint):
+    """Both public execution paths reject an unsupported harness before any boot."""
+
+    async def unexpected_open(self):
+        pytest.fail("unsupported split configuration reached rollout provisioning")
+
+    monkeypatch.setattr(Rollout, "open", unexpected_open)
+    agent = vf.Agent(
+        vf.AgentConfig(
+            model="unused",
+            harness={"id": "bash"},
+            runtime={"type": "prime"},
+            deployment={"mode": "split"},
+        )
+    )
+    task = vf.Task(vf.TaskData(prompt="unused"))
+    with pytest.raises(ValueError, match="does not support split deployment"):
+        if entrypoint == "run":
+            await agent.run(task)
+        else:
+            async with agent.interaction(task):
+                pytest.fail("unsupported interaction was opened")
+
+
+@pytest.mark.parametrize(
+    "harness,harness_runtime", grid(needs=("shell", "split")), indirect=True
+)
+async def test_split_execution(run, harness, harness_runtime):
+    """The agent loop runs on the harness's VM; its tools act on the workspace's files,
+    working directory and environment over the rollout's private network."""
+    (episode,) = await run(
+        "split-execution-v1",
+        harness=harness,
+        runtime={**harness_runtime, "workdir": "/workspace"},
+        env={"agent": SPLIT},
+        max_turns=10,
+        max_tokens=8192,
+    )
+    (trace,) = episode.traces
+    assert not trace.errors
+    assert trace.agent.runtime.type == trace.agent.harness_runtime.type == "prime"
+    assert trace.agent.runtime.id != trace.agent.harness_runtime.id
+    assert trace.reward == 1.0, trace.info.get("workspace_report")

verifiers/v1/__init__.py

@@ -110,7 +110,11 @@
         RuntimeProcess,
     )
     from verifiers.v1.runtimes.compose import ComposeSpec
-    from verifiers.v1.runtimes.deployment import Deployment, ServicesSpec
+    from verifiers.v1.runtimes.deployment import (
+        Deployment,
+        ExecutionConnection,
+        ServicesSpec,
+    )
     from verifiers.v1.semantic import (
         ACP_SEMANTIC_EDGES_METADATA_KEY,
         ACPInfo,
@@ -237,6 +241,7 @@
     "EvalClientConfig": "verifiers.v1.configs.client",
     "EvalRunInfo": "verifiers.v1.episode",
     "EvalWorkInfo": "verifiers.v1.episode",
+    "ExecutionConnection": "verifiers.v1.runtimes.deployment",
     "GroupInfo": "verifiers.v1.episode",
     "Harness": "verifiers.v1.harness",
     "HarnessConfig": "verifiers.v1.configs.harness",
@@ -581,6 +586,7 @@
     "ComposeSpec",
     "Deployment",
     "DeploymentConfig",
+    "ExecutionConnection",
     "ServicesSpec",
     "ScoringConfig",
 ]

verifiers/v1/agent.py

@@ -826,11 +826,25 @@ def _rollout_params(
         interception — shared by `run` and `interaction`."""
         harness = self.harness
         skills = [*task.data.skills, *harness.config.skills]
-        if skills:
-            # Skill installations can hold run-specific state, such as RLM's package environment.
+        split = self.config.deployment.mode == "split"
+        if skills or split:
+            # Skill installations and split execution workers hold run-specific state,
+            # such as RLM's package environment.
             harness = type(harness)(
                 harness.config.model_copy(update={"skills": skills})
             )
+        if split:
+            harness.validate_split_execution()
+            if runtime is not None:
+                raise ValueError(
+                    "split deployment provisions the harness's VM per rollout; it "
+                    "cannot run in a borrowed runtime"
+                )
+            if any(toolset.config.colocated for toolset in task.toolsets(task.config)):
+                raise ValueError(
+                    "split deployment does not support task tools colocated with the "
+                    "harness"
+                )
         if runtime is not None:
             _check_borrowed_placement(task, runtime, self.runtime_config)
             runtime_config = runtime.config
@@ -863,6 +877,11 @@ def _rollout_params(
     async def provision(self, task: Task | None = None) -> AsyncIterator[Runtime]:
         """Provision (and on exit tear down) a box from this agent's runtime
         policy, resolved for `task` when given; share it via `run(..., runtime=runtime)`."""
+        if self.config.deployment.mode == "split":
+            raise ValueError(
+                "split deployment provisions its VMs per rollout; provision() cannot "
+                "share them"
+            )
         config = (
             resolve_runtime_config(self.runtime_config, task, self._warned_resources)
             if task is not None

verifiers/v1/configs/agent.py

@@ -93,6 +93,14 @@ def _sampling_overrides(self, sampling: SamplingConfig | None) -> dict | None:
         # resolve_agent fills every field before the config is recorded on a trace.
         return sampling.model_dump(exclude_unset=True) if sampling is not None else None
 
+    @model_validator(mode="after")
+    def _validate_deployment(self):
+        if self.deployment.mode == "split" and not isinstance(
+            self.runtime, PrimeConfig
+        ):
+            raise ValueError("split deployment needs a Prime workspace runtime")
+        return self
+
     @model_validator(mode="before")
     @classmethod
     def _resolve_harness(cls, data):

verifiers/v1/configs/deployment.py

@@ -4,9 +4,16 @@
 
 from pydantic_config import BaseConfig
 
+from verifiers.v1.configs.runtime import PrimeConfig
+
 
 class DeploymentConfig(BaseConfig):
-    mode: Literal["colocated"] = "colocated"
-    """Harness and tools share the workspace."""
+    mode: Literal["colocated", "split"] = "colocated"
+    """`colocated`: the harness runs in the task workspace with its tools. `split`: the
+    harness runs on a Prime VM of its own on the rollout's private network, and its
+    tools execute in the workspace, which never holds the model credentials."""
+    harness: PrimeConfig = PrimeConfig()
+    """The harness's VM in split mode. It shares the workspace's labels, region and
+    creation rate unless set here, and always the task's network policy."""
     trust_compose: bool = False
     """Permit task-authored Compose privileges and bind mounts on local Docker."""

verifiers/v1/envs/agentic_judge/env.py

@@ -285,6 +285,11 @@ def _check_agents(self) -> None:
                 "that needs no execution is a plugged judge "
                 "(--env.taskset.task.judges), not an agent."
             )
+        if self.config.judge.deployment.mode != "colocated":
+            raise ValueError(
+                "the judge borrows the verifier workspace, so it needs colocated "
+                "deployment"
+            )
         if self.config.judge.scoring.model_fields_set:
             raise ValueError(
                 "configure scoring placement on solver.scoring, not judge.scoring"

verifiers/v1/harness.py

@@ -29,6 +29,7 @@
 if TYPE_CHECKING:
     # Annotation-only: `Trace` appears in signatures only, so this module stays
     # importable below the trace record.
+    from verifiers.v1.runtimes.deployment import ExecutionConnection
     from verifiers.v1.trace import Trace
 
 logger = logging.getLogger(__name__)
@@ -58,6 +59,9 @@ class Harness(ABC, Generic[ConfigT]):
     """Whether the program discovers SKILL.md skills — it calls
     `install_skills` with a run-scoped discovery location; configuring
     `skills` on a harness without support is rejected up front."""
+    SUPPORTS_SPLIT_EXECUTION: ClassVar[bool] = False
+    """Whether the program can run on a VM of its own while its tools execute in the
+    task workspace (`deployment.mode = "split"`), set up by `setup_execution`."""
     NEEDS_CONTAINER: ClassVar[bool] = True
     """Whether the program must run in a container runtime: True for every harness
     that installs and drives a third-party program — on the host (subprocess) it
@@ -114,6 +118,23 @@ def resolve_text_prompt(self, task: TaskData) -> tuple[str | None, str | None]:
     async def setup(self, runtime: Runtime) -> None:
         """Provision this harness through `runtime` before its execution timeout starts."""
 
+    def validate_split_execution(self) -> None:
+        """Reject a split deployment this harness cannot run before provisioning."""
+        if not self.SUPPORTS_SPLIT_EXECUTION:
+            raise ValueError(
+                f"harness {self.config.id!r} does not support split deployment"
+            )
+
+    async def setup_execution(
+        self, runtime: Runtime, workspace: Runtime, connection: ExecutionConnection
+    ) -> None:
+        """In a split deployment, concurrently with `setup(runtime)` on the harness's own
+        VM: make the program's tools execute in `workspace`, through workers
+        `connection` starts there."""
+        raise HarnessError(
+            f"harness {self.config.id!r} does not support split deployment"
+        )
+
     async def install_skills(self, runtime: Runtime, dest: str) -> None:
         """Install `config.skills` in the program's discovery directory.
         Runtime roots are copied in place; host skill folders are uploaded."""