From d4ea61d0a94a6505e30aa4f68e1bc49b20f70983 Mon Sep 17 00:00:00 2001 From: hallerite Date: Thu, 29 Oct 2026 22:35:26 +0000 Subject: [PATCH 01/12] Run split harnesses on Prime VMs beside the workspace With deployment.mode = "split", the harness runs on a Prime VM of its own that joins the rollout's private LAN, and its tools execute in the workspace. The workspace is placed exactly as in colocated mode, so plain images, ServicesSpec services, native Compose services and tool server VMs all split the same way. - The harness VM may connect to the workspace and to tool server VMs; nothing may connect to it. It takes the task's network policy plus the model and tool routes, and the workspace keeps no route to the model. - ExecutionConnection.serve starts a worker in the workspace that listens on the LAN, in a cgroup of its own; closing the harness session kills the worker and every process its tools started. - RLM runs its native tools and builtin skills through nano-rlm's TCP worker. - Harness and worker setup run concurrently. Co-Authored-By: Claude Opus 5.5 --- docs/runtimes.md | 8 ++ .../fixtures/split_execution_v1.py | 42 +++++++ tests/integration/test_split_execution.py | 56 ++++++++++ verifiers/v1/__init__.py | 8 +- verifiers/v1/agent.py | 23 +++- verifiers/v1/configs/agent.py | 8 ++ verifiers/v1/configs/deployment.py | 11 +- verifiers/v1/envs/agentic_judge/env.py | 5 + verifiers/v1/harness.py | 21 ++++ verifiers/v1/harnesses/rlm/harness.py | 43 +++++++- verifiers/v1/rollout.py | 47 ++++++-- verifiers/v1/runtimes/deployment.py | 103 +++++++++++++++++- verifiers/v1/runtimes/private_network.py | 48 ++++++-- verifiers/v1/runtimes/strategies.py | 42 ++++++- verifiers/v1/trace.py | 2 + 15 files changed, 437 insertions(+), 30 deletions(-) create mode 100644 tests/integration/fixtures/split_execution_v1.py create mode 100644 tests/integration/test_split_execution.py diff --git a/docs/runtimes.md b/docs/runtimes.md index 455f0370c2..54ba45a714 100644 --- a/docs/runtimes.md +++ b/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 and runs in a cgroup of its own, which is killed with every process it started when the harness session closes. 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: diff --git a/tests/integration/fixtures/split_execution_v1.py b/tests/integration/fixtures/split_execution_v1.py new file mode 100644 index 0000000000..fd2faa188d --- /dev/null +++ b/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"] diff --git a/tests/integration/test_split_execution.py b/tests/integration/test_split_execution.py new file mode 100644 index 0000000000..8e81650b8c --- /dev/null +++ b/tests/integration/test_split_execution.py @@ -0,0 +1,56 @@ +"""Split deployment: the harness on a Prime VM of its own, its tools in the workspace.""" + +import pytest + +import verifiers.v1 as vf +from verifiers.v1.rollout import Rollout + + +@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.prime +@pytest.mark.rlm +async def test_rlm_split_execution(run): + """RLM's 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="rlm", + runtime={"type": "prime", "workdir": "/workspace"}, + env={ + "agent": { + "deployment": {"mode": "split"}, + "timeout": {"setup": 900, "rollout": 300, "scoring": 60}, + }, + }, + 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") diff --git a/verifiers/v1/__init__.py b/verifiers/v1/__init__.py index b2f3452fe5..02b8706834 100644 --- a/verifiers/v1/__init__.py +++ b/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", ] diff --git a/verifiers/v1/agent.py b/verifiers/v1/agent.py index 311746de92..ed36ca4b4f 100644 --- a/verifiers/v1/agent.py +++ b/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 diff --git a/verifiers/v1/configs/agent.py b/verifiers/v1/configs/agent.py index 5517c981ca..2828026965 100644 --- a/verifiers/v1/configs/agent.py +++ b/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): diff --git a/verifiers/v1/configs/deployment.py b/verifiers/v1/configs/deployment.py index a03ac9938f..b6c584dda0 100644 --- a/verifiers/v1/configs/deployment.py +++ b/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.""" diff --git a/verifiers/v1/envs/agentic_judge/env.py b/verifiers/v1/envs/agentic_judge/env.py index aeb3c67366..41024029af 100644 --- a/verifiers/v1/envs/agentic_judge/env.py +++ b/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" diff --git a/verifiers/v1/harness.py b/verifiers/v1/harness.py index a4329bbe61..2e6bb73d04 100644 --- a/verifiers/v1/harness.py +++ b/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.""" diff --git a/verifiers/v1/harnesses/rlm/harness.py b/verifiers/v1/harnesses/rlm/harness.py index f124963a62..befdf84278 100644 --- a/verifiers/v1/harnesses/rlm/harness.py +++ b/verifiers/v1/harnesses/rlm/harness.py @@ -22,6 +22,7 @@ from verifiers.v1.configs.runtime import NetworkPolicyConfig from verifiers.v1.harnesses.utils.install import ensure_installed, remove_dir from verifiers.v1.runtimes import Runtime +from verifiers.v1.runtimes.deployment import ExecutionConnection from verifiers.v1.task import TaskData from verifiers.v1.trace import Trace @@ -70,7 +71,7 @@ class CompactionConfig(BaseConfig): class RLMHarnessConfig(HarnessConfig): dialect: Literal["chat_completions"] = "chat_completions" version: str = Field( - default="e4d4407f1fe6f4bd82b2c1da26ccd737627a6543", min_length=1 + default="ff8767d3bdc311ca55a98fea2c6d4438c006741a", min_length=1 ) """Git ref (branch, tag, or commit) of nano-rlm to install. Must know every field this harness puts on the wire, i.e. be at least the default ref.""" @@ -137,13 +138,19 @@ class RLMHarness(ACPHarness[RLMHarnessConfig]): APPENDS_SYSTEM_PROMPT = True SUPPORTS_MCP = True SUPPORTS_SKILLS = True + SUPPORTS_SPLIT_EXECUTION = True _skills_install_dir: str | None = None + _execution: JsonObject | None = None async def setup(self, runtime: Runtime) -> None: if self.config.skills: # Editable skill packages and uv's tool environment must belong to this run. self._skills_install_dir = f"{RLM_CACHE_DIR}-skills-{secrets.token_hex(16)}" await self.install_skills(runtime, f"{self._skills_home()}/skills") + await self._install(runtime) + await super().setup(runtime) + + async def _install(self, runtime: Runtime) -> None: directory = self._install_dir() binary = f"{directory}/bin/rlm" checkout = f"{directory}/checkout" @@ -178,7 +185,37 @@ async def setup(self, runtime: Runtime) -> None: env=env, label="rlm", ) - await super().setup(runtime) + + def validate_split_execution(self) -> None: + super().validate_split_execution() + if self.config.skills: + raise ValueError("RLM split deployment supports builtin skills only") + + async def setup_execution( + self, runtime: Runtime, workspace: Runtime, connection: ExecutionConnection + ) -> None: + # The same pinned nano-rlm runs the tools in the workspace; the agent loop and + # the model credentials stay on the harness's VM. + await self._install(workspace) + token = secrets.token_urlsafe(32) + port = await connection.serve( + workspace, + lambda port: [ + f"{self._install_dir()}/tools/rlm/bin/python", + "-m", + "rlm.execution", + "--listen", + "0.0.0.0", + "--port", + str(port), + ], + {"RLM_EXECUTION_TOKEN": token}, + ) + self._execution = { + "execution_tcp": {"host": connection.host, "port": port, "token": token}, + "execution_cwd": workspace.config.workdir, + "kernel_env": dict(workspace.env), + } def _runtime_metadata( self, @@ -249,6 +286,8 @@ def _runtime_metadata( payload["network_block"] = runtime.config.block if self.config.builtin_tools is not None: payload["builtin_tools"] = list(self.config.builtin_tools) + if self._execution is not None: + payload.update(self._execution) return {RLM_RUNTIME_METADATA_KEY: payload} async def prepare_acp( diff --git a/verifiers/v1/rollout.py b/verifiers/v1/rollout.py index de855cc913..d43e804969 100644 --- a/verifiers/v1/rollout.py +++ b/verifiers/v1/rollout.py @@ -190,6 +190,9 @@ def __init__( self._urls: dict[str, str] = {} self._harness_session: HarnessSession | None = None self._agent_runtime: Runtime | None = None + self._split_runtime: Runtime | None = None + """A split deployment's harness VM, where the program runs apart from the + workspace; None when it runs in the workspace.""" self.deadline_at: float | None = None """The active harness segment's absolute deadline (event-loop clock), or None between segments / when unbounded. An interaction spends one cumulative @@ -311,6 +314,11 @@ async def open(self) -> bool: await self.deployment.start() self._deployment_started = True runtime = self.runtime = self.deployment.workspace + if self.deployment.split: + self._split_runtime = self.deployment.harness_backend.bind() + self.trace.agent.harness_runtime = ( + self._split_runtime.info.model_copy() + ) finally: # Providers may assign an ID before startup fails; retain it for diagnostics. self.trace.agent.runtime = runtime.info.model_copy() @@ -354,7 +362,11 @@ async def open(self) -> bool: self._agent_runtime = await resolve_agent_runtime( runtime, self.task.data.agent_user ) - await self.harness.setup(self._agent_runtime) + if self._split_runtime is None: + await self.harness.setup(self._agent_runtime) + else: + await self._setup_split_harness() + harness_host = self._split_runtime or runtime # `base_url` is the interception server's reachable URL for this rollout. # The harness reaches the model at `{base_url}/v1`; tool servers reach this # rollout's `/state` + `/task` at `base_url` — it's universally reachable @@ -366,13 +378,13 @@ async def open(self) -> bool: ) = await self._stack.enter_async_context( serve_interception( self._interception, - runtime, + harness_host, self._session, toolsets, self._shared_tools, ) ) - self._endpoint = runtime.host_url(f"{base_url.rstrip('/')}/v1") + self._endpoint = harness_host.host_url(f"{base_url.rstrip('/')}/v1") self._secret = model_secret services = ( { @@ -385,7 +397,7 @@ async def open(self) -> bool: self._urls = await self._stack.enter_async_context( serve_tools( toolsets, - self._agent_runtime, + self._split_runtime or self._agent_runtime, shared=self._shared_tools, state_secret=state_secret, proxy_mcp=self._session.proxy_mcp, @@ -503,7 +515,7 @@ async def open(self) -> bool: if not self._session.stopped: session_kwargs = ( { - "tool_interception_url": runtime.host_url( + "tool_interception_url": harness_host.host_url( f"{base_url.rstrip('/')}/tool" ) } @@ -517,7 +529,7 @@ async def open(self) -> bool: self._harness_session = await self.harness.session( self.ctx, self.trace, - self._agent_runtime, + self._split_runtime or self._agent_runtime, self._endpoint, self._secret, self._urls, @@ -551,6 +563,23 @@ async def open(self) -> bool: self.trace.notify() return not self._session.stopped + async def _setup_split_harness(self) -> None: + """Install the harness on its own VM while its tools' workers start in the + workspace. The workers stop with the harness session, before the tool servers + and interception that follow them onto the stack.""" + connection = self.deployment.connection() + self._stack.push_async_callback(connection.close) + try: + async with asyncio.TaskGroup() as group: + group.create_task(self.harness.setup(self._split_runtime)) + group.create_task( + self.harness.setup_execution( + self._split_runtime, self._agent_runtime, connection + ) + ) + except ExceptionGroup as failed: + raise failed.exceptions[0] # typed for retries + async def steer(self, message: str, *, message_id: str | None = None) -> dict: if self._closed or not self.ok: raise RuntimeError("this rollout is closed or stopped") @@ -647,7 +676,7 @@ async def _cleanup_harness(self, runtime: Runtime) -> None: if self._harness_cleaned: return try: - await self.harness.cleanup(self.trace, runtime) + await self.harness.cleanup(self.trace, self._split_runtime or runtime) except Exception: logger.warning( "harness cleanup failed (rollout %s)", self.trace.id, exc_info=True @@ -683,7 +712,9 @@ async def _score_harness(self, runtime: Runtime, deadline: float | None) -> None """Collect best-effort metrics without replacing a task failure.""" try: async with asyncio.timeout_at(deadline): - await self.harness.score(self.trace, self._agent_runtime or runtime) + await self.harness.score( + self.trace, self._split_runtime or self._agent_runtime or runtime + ) except Exception as error: self.trace.info["harness_scoring_error"] = { "type": type(error).__name__, diff --git a/verifiers/v1/runtimes/deployment.py b/verifiers/v1/runtimes/deployment.py index 919ce62a5e..ef463cb488 100644 --- a/verifiers/v1/runtimes/deployment.py +++ b/verifiers/v1/runtimes/deployment.py @@ -3,12 +3,16 @@ from __future__ import annotations import asyncio +import contextlib import logging -from collections.abc import AsyncIterator, Mapping +import random +import uuid +from collections.abc import AsyncIterator, Callable, Mapping from contextlib import AsyncExitStack, asynccontextmanager -from dataclasses import dataclass +from dataclasses import dataclass, field from verifiers.v1.configs.deployment import DeploymentConfig +from verifiers.v1.errors import SandboxError from verifiers.v1.runtimes import ( PrimeConfig, Runtime, @@ -37,6 +41,85 @@ class ServicesSpec: services: dict[str, PrimeConfig] +_WORKER_PORTS = range(20000, 32000) +"""Below Linux's ephemeral range, so a worker's port cannot collide with an outgoing +connection's.""" +_HOLD = 'echo $$ > "$1.pid"; while [ ! -e "$1.joined" ]; do sleep 0.05; done; shift; exec "$@"' +"""Wait, as the worker's user, until root has moved this process into its cgroup.""" +_JOIN = """ +set -e +group=/sys/fs/cgroup/$1 files=$2 port=$3 +mkdir "$group" +for _ in $(seq 300); do [ -s "$files.pid" ] && break; sleep 0.1; done +pid=$(cat "$files.pid") +echo "$pid" > "$group/cgroup.procs" +: > "$files.joined" +for _ in $(seq 600); do + (exec 3<>"/dev/tcp/127.0.0.1/$port") 2>/dev/null && exit 0 + kill -0 "$pid" 2>/dev/null || { echo "worker exited" >&2; exit 1; } + sleep 0.1 +done +echo "worker did not listen on port $port" >&2 +exit 1 +""" +_KILL = """ +group=/sys/fs/cgroup/$1 files=$2 +if [ -e "$files.joined" ]; then + echo 1 > "$group/cgroup.kill" + for _ in $(seq 50); do rmdir "$group" 2>/dev/null && break; sleep 0.1; done +elif [ -s "$files.pid" ]; then + kill -9 "$(cat "$files.pid")" +fi +rm -f "$files.pid" "$files.joined" +""" + + +@dataclass +class ExecutionConnection: + """A split deployment's way into the workspace for the harness's tools: workers + started there listen on the rollout's private network, where the harness reaches + the workspace at `host`. Each worker runs in a cgroup of its own, so `close` stops + every process its tools started, detached or not.""" + + host: str + """The workspace's name on the private network.""" + _workers: list[tuple[Runtime, str]] = field(default_factory=list) + + async def serve( + self, + runtime: Runtime, + argv: Callable[[int], list[str]], + env: dict[str, str], + ) -> int: + """Start `argv(port)` through `runtime`, a workspace view, and return the port + once the worker accepts connections on it.""" + port = random.choice(_WORKER_PORTS) + group = f"vf-exec-{uuid.uuid4().hex[:12]}" + files = f"/tmp/{group}" + root = runtime.with_user(None) + self._workers.append((root, group)) + await runtime.run_background( + ["sh", "-c", _HOLD, "sh", files, *argv(port)], env, f"{files}.log" + ) + result = await root.run( + ["bash", "-c", _JOIN, "bash", group, files, str(port)], {} + ) + if result.exit_code: + log = await root.run(["tail", "-c", "2000", f"{files}.log"], {}) + raise SandboxError( + f"execution worker failed to start: {result.stderr.strip()[-500:]}\n" + f"{log.stdout}" + ) + return port + + async def close(self) -> None: + """Stop every worker and the processes they started.""" + workers, self._workers = self._workers, [] + for root, group in workers: + with contextlib.suppress(Exception): + await root.run(["sh", "-c", _KILL, "sh", group, f"/tmp/{group}"], {}) + + class _DeploymentStrategy: """Placement, connectivity and lifecycle hooks for one deployment attempt.""" @@ -133,6 +216,22 @@ async def start(self) -> None: ) raise + @property + def split(self) -> bool: + """Whether the harness runs apart from the workspace, on a VM of its own.""" + return ( + self.harness_backend is not None + and self.harness_backend is not self.execution_backend + ) + + def connection(self) -> ExecutionConnection: + """A split harness's way into the workspace, over the rollout's private network.""" + from verifiers.v1.runtimes.private_network import lan_host + + if not self.split: + raise ValueError("only a split deployment connects its harness's tools") + return ExecutionConnection(lan_host(self.execution_backend)) + @property def execution_name(self) -> str: return next( diff --git a/verifiers/v1/runtimes/private_network.py b/verifiers/v1/runtimes/private_network.py index f7be0c36e5..4d7afbb424 100644 --- a/verifiers/v1/runtimes/private_network.py +++ b/verifiers/v1/runtimes/private_network.py @@ -28,6 +28,8 @@ logger = logging.getLogger(__name__) WORKSPACE_TAG = "vf-workspace" +HARNESS_TAG = "vf-harness" +TASK_TAG = "vf-task" def tool_vms( @@ -77,9 +79,13 @@ async def private_network( *, interconnect: bool = False, networks: Mapping[str, list[str]] | None = None, + harness: RuntimeBackend | None = None, + start_harness: Callable[[], Awaitable[object]] | None = None, ) -> AsyncIterator[dict[str, RuntimeBackend]]: """Boot the workspace's VM and one VM per member together, each joining a LAN - network of the rollout's own as it boots.""" + network of the rollout's own as it boots. A split deployment's `harness` VM joins + too: it may open connections to the workspace and the tool servers' VMs, and + nothing may open one to it.""" if not isinstance(workspace, PrimeBackend): raise TypeError( "tool servers on Prime runtimes of their own join the workspace over a " @@ -93,14 +99,6 @@ async def private_network( "main": [WORKSPACE_TAG], **{name: [tag] for name, tag in tags.items()}, } - rules = ( - [LanRule(from_="*", to="*")] - if interconnect - else [ - LanRule(from_=f"tag:{WORKSPACE_TAG}", to=f"tag:{tag}") - for tag in tags.values() - ] - ) if networks is not None: rules = [] for index, names in enumerate(networks.values()): @@ -108,6 +106,24 @@ async def private_network( rules.append(LanRule(from_=f"tag:{tag}", to=f"tag:{tag}")) for name in names: memberships[name].append(tag) + elif interconnect: + # Task services and the workspace, not the harness, reach each other. + rules = [LanRule(from_=f"tag:{TASK_TAG}", to=f"tag:{TASK_TAG}")] + for member_tags in memberships.values(): + member_tags.append(TASK_TAG) + else: + rules = [ + LanRule(from_=f"tag:{WORKSPACE_TAG}", to=f"tag:{tag}") + for tag in tags.values() + ] + if harness is not None: + rules.append(LanRule(from_=f"tag:{HARNESS_TAG}", to=f"tag:{WORKSPACE_TAG}")) + if networks is None and not interconnect: + # The harness consumes the tool servers. + rules += [ + LanRule(from_=f"tag:{HARNESS_TAG}", to=f"tag:{tag}") + for tag in tags.values() + ] async with AsyncExitStack() as stack: client = await stack.enter_async_context(leased_client()) # Install connectivity before boot: Compose peers must share a network; @@ -152,13 +168,27 @@ async def provision(name: str, config: PrimeConfig) -> None: ) workspace.lan = (network.id, memberships["main"]) + if harness is not None: + harness.lan = (network.id, [HARNESS_TAG]) + stack.push_async_callback(harness.stop) try: async with asyncio.TaskGroup() as group: group.create_task(start_workspace()) + if start_harness is not None: + group.create_task(start_harness()) for name, config in members.items(): group.create_task(provision(name, config)) except ExceptionGroup as failed: raise failed.exceptions[0] # one VM's failure, typed for retries + idle_timeout = workspace.config.idle_timeout + if harness is not None and idle_timeout is not None: + # The harness's tools reach the workspace over the LAN alone. + stack.push_async_callback( + stop, + asyncio.create_task( + _keepalive(workspace.bind(), max(10.0, idle_timeout / 3)) + ), + ) logger.info( "LAN network %s joins %s to %s", network.id, diff --git a/verifiers/v1/runtimes/strategies.py b/verifiers/v1/runtimes/strategies.py index 6284211c2f..e4fcd2d6c6 100644 --- a/verifiers/v1/runtimes/strategies.py +++ b/verifiers/v1/runtimes/strategies.py @@ -5,7 +5,7 @@ from contextlib import AsyncExitStack, asynccontextmanager from functools import partial -from verifiers.v1.runtimes import PrimeBackend, PrimeConfig +from verifiers.v1.runtimes import PrimeBackend, PrimeConfig, make_backend from verifiers.v1.runtimes.compose import compose_services from verifiers.v1.runtimes.deployment import ServicesSpec, _DeploymentStrategy from verifiers.v1.runtimes.private_network import lan_host, private_network @@ -62,6 +62,7 @@ async def prepare_verifier(self, deployment): class _PrivateNetworkDeployment(_DeploymentStrategy): _networks: dict[str, list[str]] | None = None _internal: frozenset[str] = frozenset() + _harness = None async def start(self, d): if d.borrowed_runtime is not None: @@ -72,6 +73,12 @@ async def start(self, d): d.backend.env = d.env native = isinstance(d.spec, ServicesSpec) members = d.spec.services if native else d.tool_vms + harness = self._harness = None + if d.config.mode == "split": + harness = self._harness = make_backend( + _harness_config(d.config.harness, d.backend.config), + f"{d.backend.name}-harness", + ) self._services = await d.enter(AsyncExitStack()) vms = await self._services.enter_async_context( private_network( @@ -80,20 +87,24 @@ async def start(self, d): partial(d.own, d.backend), interconnect=native, networks=self._networks, + harness=harness, + start_harness=harness and partial(d.own, harness), ) ) d.target_backends = {"main": d.backend, **vms} d.hosts = {name: lan_host(backend) for name, backend in vms.items()} if native: d.hosts["main"] = lan_host(d.backend) - d.execution_backend = d.harness_backend = d.backend + d.execution_backend = d.backend + d.harness_backend = harness or d.backend async def prepare_execution(self, d, routes, state_route=None): # Each VM enforces its own policy. A tool server's VM keeps only its state # channel, reached as the server reaches it; LAN traffic is unaffected by either. + # A split harness's VM keeps the routes; the workspace then keeps none. def own(name, backend): if name == "main": - return routes + return [] if self._harness else routes if name in self._internal: return [] return [backend.host_url(state_route)] if state_route else [] @@ -103,6 +114,7 @@ def own(name, backend): backend.prepare_execution(own(name, backend)) for name, backend in d.target_backends.items() ), + *([self._harness.prepare_execution(routes)] if self._harness else []), ) async def stop_execution(self, d): @@ -114,7 +126,9 @@ async def prepare_verifier(self, deployment): # Evidence is collected; release the solver's VMs before the verifier clock # starts, the workspace's even if the service VMs' or the network's release # fails. + harness = [(self._harness.stop, "harness VM")] if self._harness else [] for release, what in ( + *harness, (deployment.execution_backend.stop, "workspace VM"), (self._services.aclose, "service VMs"), ): @@ -191,7 +205,27 @@ async def close(self, deployment): ) +def _harness_config(harness: PrimeConfig, workspace: PrimeConfig) -> PrimeConfig: + """A split harness's VM: placed and labelled like the workspace unless set, and + under the task's network policy, which the harness would share colocated.""" + return harness.model_copy( + update={ + "labels": list(dict.fromkeys([*workspace.labels, *harness.labels])), + "region": harness.region or workspace.region, + "creates_per_min": harness.creates_per_min or workspace.creates_per_min, + "allow": workspace.allow, + "block": workspace.block, + } + ) + + def _deployment_strategy(config, backend, spec, tool_vms=None): + split = config is not None and config.mode == "split" + if split and not isinstance(backend, PrimeBackend): + raise ValueError( + "split deployment places the harness on a Prime VM beside a Prime " + f"workspace, not a {backend.type} one" + ) if isinstance(spec, ServicesSpec): if not isinstance(backend, PrimeBackend): raise TypeError("Native service VMs require a Prime workspace") @@ -213,6 +247,6 @@ def _deployment_strategy(config, backend, spec, tool_vms=None): "cannot run Compose services; use ServicesSpec on Prime or local Docker" ) return _ContainerDeployment() - if tool_vms: + if tool_vms or split: return _PrivateNetworkDeployment() return _ColocatedDeployment() diff --git a/verifiers/v1/trace.py b/verifiers/v1/trace.py index c0cbd6bbaa..67a617604a 100644 --- a/verifiers/v1/trace.py +++ b/verifiers/v1/trace.py @@ -149,6 +149,8 @@ class AgentInfo(BaseModel, Generic[AgentConfigT]): """The resolved config that rebuilds the agent (`Agent(trace.agent.config)`).""" runtime: RuntimeInfo | None = None """The box the rollout ran in; None until provisioning.""" + harness_runtime: RuntimeInfo | None = None + """A split deployment's harness VM, where the program ran apart from `runtime`.""" name: str = "agent" """The env agent that produced this trace (the config field name, e.g. `solver`).""" trainable: bool = True From 36eed72eff257a1a95bf140183cb1d1d07573e6c Mon Sep 17 00:00:00 2001 From: hallerite Date: Thu, 29 Oct 2026 23:36:35 +0000 Subject: [PATCH 02/12] Tie execution worker readiness to the worker's own socket A workspace service that already listened on the worker's random port made the readiness probe succeed for the wrong process. The worker is now ready only once a process in its cgroup listens on the port; a port another process holds before the worker is released is skipped for a fresh one, up to three times. Setting deployment.harness.creates_per_min to 0 no longer inherits the workspace's rate. Co-Authored-By: Claude Opus 5.5 --- verifiers/v1/runtimes/deployment.py | 69 ++++++++++++++++++++--------- verifiers/v1/runtimes/strategies.py | 6 ++- 2 files changed, 53 insertions(+), 22 deletions(-) diff --git a/verifiers/v1/runtimes/deployment.py b/verifiers/v1/runtimes/deployment.py index ef463cb488..38ca87138b 100644 --- a/verifiers/v1/runtimes/deployment.py +++ b/verifiers/v1/runtimes/deployment.py @@ -46,22 +46,39 @@ class ServicesSpec: connection's.""" _HOLD = 'echo $$ > "$1.pid"; while [ ! -e "$1.joined" ]; do sleep 0.05; done; shift; exec "$@"' """Wait, as the worker's user, until root has moved this process into its cgroup.""" +_PORT_TAKEN = 2 _JOIN = """ set -e group=/sys/fs/cgroup/$1 files=$2 port=$3 -mkdir "$group" +printf -v hex '%04X' "$port" +listening() { + cat /proc/net/tcp /proc/net/tcp6 2>/dev/null | while read -r _ address _ state _ _ _ _ _ inode _; do + if [ "$state" = 0A ] && [ "${address##*:}" = "$hex" ]; then echo "$inode"; fi + done +} for _ in $(seq 300); do [ -s "$files.pid" ] && break; sleep 0.1; done pid=$(cat "$files.pid") +[ -z "$(listening)" ] || exit 2 +mkdir "$group" echo "$pid" > "$group/cgroup.procs" : > "$files.joined" for _ in $(seq 600); do - (exec 3<>"/dev/tcp/127.0.0.1/$port") 2>/dev/null && exit 0 + for inode in $(listening); do + for proc in $(cat "$group/cgroup.procs"); do + for fd in /proc/"$proc"/fd/*; do + [ "$(readlink "$fd")" = "socket:[$inode]" ] && exit 0 + done + done + done kill -0 "$pid" 2>/dev/null || { echo "worker exited" >&2; exit 1; } sleep 0.1 done echo "worker did not listen on port $port" >&2 exit 1 """ +"""Move the held worker into its cgroup, then wait until a process in that cgroup +listens on the port. Exits `_PORT_TAKEN` before releasing the worker if another +process already listens there.""" _KILL = """ group=/sys/fs/cgroup/$1 files=$2 if [ -e "$files.joined" ]; then @@ -92,32 +109,42 @@ async def serve( env: dict[str, str], ) -> int: """Start `argv(port)` through `runtime`, a workspace view, and return the port - once the worker accepts connections on it.""" - port = random.choice(_WORKER_PORTS) - group = f"vf-exec-{uuid.uuid4().hex[:12]}" - files = f"/tmp/{group}" + once the worker itself listens on it.""" root = runtime.with_user(None) - self._workers.append((root, group)) - await runtime.run_background( - ["sh", "-c", _HOLD, "sh", files, *argv(port)], env, f"{files}.log" - ) - result = await root.run( - ["bash", "-c", _JOIN, "bash", group, files, str(port)], {} - ) - if result.exit_code: - log = await root.run(["tail", "-c", "2000", f"{files}.log"], {}) - raise SandboxError( - f"execution worker failed to start: {result.stderr.strip()[-500:]}\n" - f"{log.stdout}" + for _ in range(3): + port = random.choice(_WORKER_PORTS) + group = f"vf-exec-{uuid.uuid4().hex[:12]}" + files = f"/tmp/{group}" + self._workers.append((root, group)) + await runtime.run_background( + ["sh", "-c", _HOLD, "sh", files, *argv(port)], env, f"{files}.log" ) - return port + result = await root.run( + ["bash", "-c", _JOIN, "bash", group, files, str(port)], {} + ) + if not result.exit_code: + return port + if result.exit_code != _PORT_TAKEN: + log = await root.run(["tail", "-c", "2000", f"{files}.log"], {}) + raise SandboxError( + "execution worker failed to start: " + f"{result.stderr.strip()[-500:]}\n{log.stdout}" + ) + # A workspace service holds the port: release the held worker, try another. + self._workers.remove((root, group)) + await _stop(root, group) + raise SandboxError("execution worker found no free port in the workspace") async def close(self) -> None: """Stop every worker and the processes they started.""" workers, self._workers = self._workers, [] for root, group in workers: - with contextlib.suppress(Exception): - await root.run(["sh", "-c", _KILL, "sh", group, f"/tmp/{group}"], {}) + await _stop(root, group) + + +async def _stop(root: Runtime, group: str) -> None: + with contextlib.suppress(Exception): + await root.run(["sh", "-c", _KILL, "sh", group, f"/tmp/{group}"], {}) class _DeploymentStrategy: diff --git a/verifiers/v1/runtimes/strategies.py b/verifiers/v1/runtimes/strategies.py index e4fcd2d6c6..aa9ff0b548 100644 --- a/verifiers/v1/runtimes/strategies.py +++ b/verifiers/v1/runtimes/strategies.py @@ -212,7 +212,11 @@ def _harness_config(harness: PrimeConfig, workspace: PrimeConfig) -> PrimeConfig update={ "labels": list(dict.fromkeys([*workspace.labels, *harness.labels])), "region": harness.region or workspace.region, - "creates_per_min": harness.creates_per_min or workspace.creates_per_min, + "creates_per_min": ( + harness.creates_per_min + if harness.creates_per_min is not None + else workspace.creates_per_min + ), "allow": workspace.allow, "block": workspace.block, } From 3130462511b03935f6371ace9b6cb3616af1274c Mon Sep 17 00:00:00 2001 From: hallerite Date: Thu, 29 Oct 2026 23:58:59 +0000 Subject: [PATCH 03/12] Let a split harness reach tool servers in task services Split deployment accepted tool servers placed in a ServicesSpec or Compose service, but the LAN let the harness connect only to the workspace and to tool server VMs, so those tools failed after provisioning. The rollout now passes the services that host tool servers to the deployment, and the harness may connect to those services, not to the others. Pin nano-rlm to its split-execution head, which stops kernel commands with the kernel and keeps worker interrupts off the supervisor's event loop. Co-Authored-By: Claude Opus 5.5 --- verifiers/v1/rollout.py | 5 +++++ verifiers/v1/runtimes/deployment.py | 5 ++++- verifiers/v1/runtimes/private_network.py | 22 +++++++++++++--------- verifiers/v1/runtimes/strategies.py | 1 + 4 files changed, 23 insertions(+), 10 deletions(-) diff --git a/verifiers/v1/rollout.py b/verifiers/v1/rollout.py index d43e804969..4d31835778 100644 --- a/verifiers/v1/rollout.py +++ b/verifiers/v1/rollout.py @@ -306,6 +306,11 @@ async def open(self) -> bool: composed=spec is not None, borrowed=self._borrowed_runtime is not None, ), + tool_services={ + service + for toolset in toolsets + if (service := getattr(toolset.config, "service", None)) + }, env=runtime.env, config=self._deployment_config, borrowed_runtime=self._borrowed_runtime, diff --git a/verifiers/v1/runtimes/deployment.py b/verifiers/v1/runtimes/deployment.py index 38ca87138b..2279367e6e 100644 --- a/verifiers/v1/runtimes/deployment.py +++ b/verifiers/v1/runtimes/deployment.py @@ -7,7 +7,7 @@ import logging import random import uuid -from collections.abc import AsyncIterator, Callable, Mapping +from collections.abc import AsyncIterator, Callable, Collection, Mapping from contextlib import AsyncExitStack, asynccontextmanager from dataclasses import dataclass, field @@ -189,6 +189,7 @@ def __init__( *, spec: ComposeSpec | ServicesSpec | None = None, tool_vms: Mapping[str, PrimeConfig] | None = None, + tool_services: Collection[str] = (), env: dict[str, str] | None = None, config: DeploymentConfig | None = None, borrowed_runtime: Runtime | None = None, @@ -198,6 +199,8 @@ def __init__( self.tool_vms = dict(tool_vms or {}) """Tool servers on Prime runtimes of their own, by server name, which join the workspace over the rollout's private network.""" + self.tool_services = frozenset(tool_services) + """Task services that host tool servers, which a split harness connects to.""" self.env = dict(env or {}) self.config = config or DeploymentConfig() self.borrowed_runtime = borrowed_runtime diff --git a/verifiers/v1/runtimes/private_network.py b/verifiers/v1/runtimes/private_network.py index 4d7afbb424..fce0488bf6 100644 --- a/verifiers/v1/runtimes/private_network.py +++ b/verifiers/v1/runtimes/private_network.py @@ -13,7 +13,7 @@ import contextlib import logging import uuid -from collections.abc import AsyncIterator, Awaitable, Callable, Mapping +from collections.abc import AsyncIterator, Awaitable, Callable, Collection, Mapping from contextlib import AsyncExitStack, asynccontextmanager from verifiers.v1.runtimes import ( @@ -81,11 +81,12 @@ async def private_network( networks: Mapping[str, list[str]] | None = None, harness: RuntimeBackend | None = None, start_harness: Callable[[], Awaitable[object]] | None = None, + harness_peers: Collection[str] = (), ) -> AsyncIterator[dict[str, RuntimeBackend]]: """Boot the workspace's VM and one VM per member together, each joining a LAN network of the rollout's own as it boots. A split deployment's `harness` VM joins - too: it may open connections to the workspace and the tool servers' VMs, and - nothing may open one to it.""" + too: it may open connections to the workspace, to tool servers' VMs and to the + task services in `harness_peers`, and nothing may open one to it.""" if not isinstance(workspace, PrimeBackend): raise TypeError( "tool servers on Prime runtimes of their own join the workspace over a " @@ -118,12 +119,15 @@ async def private_network( ] if harness is not None: rules.append(LanRule(from_=f"tag:{HARNESS_TAG}", to=f"tag:{WORKSPACE_TAG}")) - if networks is None and not interconnect: - # The harness consumes the tool servers. - rules += [ - LanRule(from_=f"tag:{HARNESS_TAG}", to=f"tag:{tag}") - for tag in tags.values() - ] + # The harness consumes the tool servers, on VMs of their own or in services. + consumed = ( + tags.values() + if networks is None and not interconnect + else [tags[name] for name in harness_peers if name in tags] + ) + rules += [ + LanRule(from_=f"tag:{HARNESS_TAG}", to=f"tag:{tag}") for tag in consumed + ] async with AsyncExitStack() as stack: client = await stack.enter_async_context(leased_client()) # Install connectivity before boot: Compose peers must share a network; diff --git a/verifiers/v1/runtimes/strategies.py b/verifiers/v1/runtimes/strategies.py index aa9ff0b548..c4d9eaabf3 100644 --- a/verifiers/v1/runtimes/strategies.py +++ b/verifiers/v1/runtimes/strategies.py @@ -89,6 +89,7 @@ async def start(self, d): networks=self._networks, harness=harness, start_harness=harness and partial(d.own, harness), + harness_peers=d.tool_services, ) ) d.target_backends = {"main": d.backend, **vms} From 3550aa0732fd0a63ffe2363803607af911192708 Mon Sep 17 00:00:00 2001 From: hallerite Date: Fri, 30 Oct 2026 00:27:28 +0000 Subject: [PATCH 04/12] Act only on the launched worker when joining and stopping it Root read the worker's PID from a file the worker's user can write, moved that process into the worker's cgroup, and could kill it by that PID on cleanup, so a forged PID file could have aimed root at any process. Root now acts only on a PID it has checked to belong to the worker's user and to carry this launch's release flag; cleanup kills the cgroup, or that checked PID after checking it again, and never reads the file. The release flag lives in a root-owned directory, and the worker's log is read as the worker's user. Co-Authored-By: Claude Opus 5.5 --- verifiers/v1/runtimes/deployment.py | 136 ++++++++++++++++++++++------ 1 file changed, 108 insertions(+), 28 deletions(-) diff --git a/verifiers/v1/runtimes/deployment.py b/verifiers/v1/runtimes/deployment.py index 2279367e6e..1eb6c23ad9 100644 --- a/verifiers/v1/runtimes/deployment.py +++ b/verifiers/v1/runtimes/deployment.py @@ -44,12 +44,33 @@ class ServicesSpec: _WORKER_PORTS = range(20000, 32000) """Below Linux's ephemeral range, so a worker's port cannot collide with an outgoing connection's.""" -_HOLD = 'echo $$ > "$1.pid"; while [ ! -e "$1.joined" ]; do sleep 0.05; done; shift; exec "$@"' +_HOLD = ( + 'echo $$ > "$1.pid"; while [ ! -e "$2" ]; do sleep 0.05; done; shift 2; exec "$@"' +) """Wait, as the worker's user, until root has moved this process into its cgroup.""" +_FLAGS = "/run/vf-exec" +"""Root-owned, so the worker's user can neither forge nor redirect a release flag.""" +_HELD = """ +uid=$(id -u "${user:-root}") +held() { + # Whether $1 is this launch's held worker, running as the worker's user: root + # only acts on a process that user could signal itself. + local key real arg + [[ $1 =~ ^[0-9]+$ ]] || return 1 + while read -r key real _; do [ "$key" = Uid: ] && break; done < "/proc/$1/status" || return 1 + [ "$real" = "$uid" ] || return 1 + while IFS= read -r -d '' arg; do [ "$arg" = "$flag" ] && return 0; done < "/proc/$1/cmdline" + return 1 +} +""" _PORT_TAKEN = 2 -_JOIN = """ +_JOIN = ( + """ set -e -group=/sys/fs/cgroup/$1 files=$2 port=$3 +group=/sys/fs/cgroup/$1 files=$2 flag=$3 port=$4 user=$5 +""" + + _HELD + + """ printf -v hex '%04X' "$port" listening() { cat /proc/net/tcp /proc/net/tcp6 2>/dev/null | while read -r _ address _ state _ _ _ _ _ inode _; do @@ -58,10 +79,14 @@ class ServicesSpec: } for _ in $(seq 300); do [ -s "$files.pid" ] && break; sleep 0.1; done pid=$(cat "$files.pid") +held "$pid" || { echo "worker identity check failed" >&2; exit 1; } +echo "$pid" [ -z "$(listening)" ] || exit 2 mkdir "$group" echo "$pid" > "$group/cgroup.procs" -: > "$files.joined" +mkdir -p -m 755 "${flag%/*}" +[ "$(stat -c '%u %a' "${flag%/*}")" = "0 755" ] || { echo "unsafe flag directory" >&2; exit 1; } +: > "$flag" for _ in $(seq 600); do for inode in $(listening); do for proc in $(cat "$group/cgroup.procs"); do @@ -76,19 +101,42 @@ class ServicesSpec: echo "worker did not listen on port $port" >&2 exit 1 """ -"""Move the held worker into its cgroup, then wait until a process in that cgroup -listens on the port. Exits `_PORT_TAKEN` before releasing the worker if another -process already listens there.""" -_KILL = """ -group=/sys/fs/cgroup/$1 files=$2 -if [ -e "$files.joined" ]; then +) +"""Validate the held worker and print its PID, move it into its cgroup and release +it, then wait until a process in that cgroup listens on the port. Exits +`_PORT_TAKEN` before moving the worker if another process already listens there.""" +_KILL = ( + """ +group=/sys/fs/cgroup/$1 files=$2 flag=$3 pid=$4 user=$5 +""" + + _HELD + + """ +if [ -e "$group/cgroup.kill" ]; then echo 1 > "$group/cgroup.kill" for _ in $(seq 50); do rmdir "$group" 2>/dev/null && break; sleep 0.1; done -elif [ -s "$files.pid" ]; then - kill -9 "$(cat "$files.pid")" fi -rm -f "$files.pid" "$files.joined" +if held "$pid"; then kill -9 "$pid"; fi +rm -f "$flag" "$files.pid" """ +) +"""Kill the worker's cgroup, or a worker never released into it by the PID that +`_JOIN` validated, checked again.""" + + +@dataclass +class _Worker: + root: Runtime + group: str + user: str + pid: str = "" + + @property + def files(self) -> str: + return f"/tmp/{self.group}" + + @property + def flag(self) -> str: + return f"{_FLAGS}/{self.group}" @dataclass @@ -100,7 +148,7 @@ class ExecutionConnection: host: str """The workspace's name on the private network.""" - _workers: list[tuple[Runtime, str]] = field(default_factory=list) + _workers: list[_Worker] = field(default_factory=list) async def serve( self, @@ -110,41 +158,73 @@ async def serve( ) -> int: """Start `argv(port)` through `runtime`, a workspace view, and return the port once the worker itself listens on it.""" - root = runtime.with_user(None) for _ in range(3): port = random.choice(_WORKER_PORTS) - group = f"vf-exec-{uuid.uuid4().hex[:12]}" - files = f"/tmp/{group}" - self._workers.append((root, group)) + worker = _Worker( + runtime.with_user(None), + f"vf-exec-{uuid.uuid4().hex[:12]}", + runtime.user or "", + ) + self._workers.append(worker) await runtime.run_background( - ["sh", "-c", _HOLD, "sh", files, *argv(port)], env, f"{files}.log" + ["sh", "-c", _HOLD, "sh", worker.files, worker.flag, *argv(port)], + env, + f"{worker.files}.log", ) - result = await root.run( - ["bash", "-c", _JOIN, "bash", group, files, str(port)], {} + result = await worker.root.run( + [ + "bash", + "-c", + _JOIN, + "bash", + worker.group, + worker.files, + worker.flag, + str(port), + worker.user, + ], + {}, ) + worker.pid = result.stdout.strip() if not result.exit_code: return port if result.exit_code != _PORT_TAKEN: - log = await root.run(["tail", "-c", "2000", f"{files}.log"], {}) + # Read as the worker's user: the log lives where that user can write. + log = await runtime.run( + ["tail", "-c", "2000", f"{worker.files}.log"], {} + ) raise SandboxError( "execution worker failed to start: " f"{result.stderr.strip()[-500:]}\n{log.stdout}" ) # A workspace service holds the port: release the held worker, try another. - self._workers.remove((root, group)) - await _stop(root, group) + self._workers.remove(worker) + await _stop(worker) raise SandboxError("execution worker found no free port in the workspace") async def close(self) -> None: """Stop every worker and the processes they started.""" workers, self._workers = self._workers, [] - for root, group in workers: - await _stop(root, group) + for worker in workers: + await _stop(worker) -async def _stop(root: Runtime, group: str) -> None: +async def _stop(worker: _Worker) -> None: with contextlib.suppress(Exception): - await root.run(["sh", "-c", _KILL, "sh", group, f"/tmp/{group}"], {}) + await worker.root.run( + [ + "bash", + "-c", + _KILL, + "bash", + worker.group, + worker.files, + worker.flag, + worker.pid, + worker.user, + ], + {}, + ) class _DeploymentStrategy: From a8f8e630a3165c81d89831d7d08b8ee5b879c7d8 Mon Sep 17 00:00:00 2001 From: hallerite Date: Fri, 30 Oct 2026 00:28:57 +0000 Subject: [PATCH 05/12] Run split deployment in CI for every harness that supports it Split deployment is a grid capability: the Split Execution job runs one Prime cell per bundled harness, and a harness without SUPPORTS_SPLIT_EXECUTION is collected as a skip with its reason, so a harness that opts in is tested without a test change. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/tests.yml | 2 ++ tests/integration/grid.py | 9 ++++--- tests/integration/test_split_execution.py | 29 +++++++++++++---------- 3 files changed, 24 insertions(+), 16 deletions(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 067a4883e7..338089a0cf 100644 --- a/.github/workflows/tests.yml +++ b/.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 diff --git a/tests/integration/grid.py b/tests/integration/grid.py index 2d636ccd32..b3e8dd5dcb 100644 --- a/tests/integration/grid.py +++ b/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): diff --git a/tests/integration/test_split_execution.py b/tests/integration/test_split_execution.py index 8e81650b8c..3d0f7912f5 100644 --- a/tests/integration/test_split_execution.py +++ b/tests/integration/test_split_execution.py @@ -1,10 +1,17 @@ -"""Split deployment: the harness on a Prime VM of its own, its tools in the workspace.""" +"""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): @@ -31,21 +38,17 @@ async def unexpected_open(self): pytest.fail("unsupported interaction was opened") -@pytest.mark.prime -@pytest.mark.rlm -async def test_rlm_split_execution(run): - """RLM's loop runs on the harness's VM; its tools act on the workspace's files, +@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="rlm", - runtime={"type": "prime", "workdir": "/workspace"}, - env={ - "agent": { - "deployment": {"mode": "split"}, - "timeout": {"setup": 900, "rollout": 300, "scoring": 60}, - }, - }, + harness=harness, + runtime={**harness_runtime, "workdir": "/workspace"}, + env={"agent": SPLIT}, max_turns=10, max_tokens=8192, ) From 8792379ce7e5d6d817e4d26e1571c1764b6349e2 Mon Sep 17 00:00:00 2001 From: hallerite Date: Fri, 30 Oct 2026 01:13:05 +0000 Subject: [PATCH 06/12] Replace the execution worker's cgroup with an environment marker The worker's launch had grown into root-run shell that held the worker, moved it into a cgroup, matched its listening socket through /proc, and checked its identity against a PID file the worker's user could write. Two choices caused all of it: picking the worker's port, and cleaning up through a cgroup, which needs root. The worker now binds a port the kernel picks and announces it, so there is no port to choose or collide on. It runs with a VF_EXEC_ID marker in its environment, which every process its tools start inherits through detaching and reparenting; close() kills the processes carrying it as the worker's own user, so the cleanup reaches nothing that user could not signal itself. A process that empties its environment escapes it; the workspace VM still ends with the rollout. Pin nano-rlm to the head that announces its bound port. Co-Authored-By: Claude Opus 5.5 --- docs/runtimes.md | 2 +- verifiers/v1/harnesses/rlm/harness.py | 6 +- verifiers/v1/runtimes/deployment.py | 211 ++++++-------------------- 3 files changed, 50 insertions(+), 169 deletions(-) diff --git a/docs/runtimes.md b/docs/runtimes.md index 54ba45a714..a4b78b76ff 100644 --- a/docs/runtimes.md +++ b/docs/runtimes.md @@ -191,7 +191,7 @@ 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 and runs in a cgroup of its own, which is killed with every process it started when the harness session closes. 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. +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] diff --git a/verifiers/v1/harnesses/rlm/harness.py b/verifiers/v1/harnesses/rlm/harness.py index befdf84278..2cddd2d6aa 100644 --- a/verifiers/v1/harnesses/rlm/harness.py +++ b/verifiers/v1/harnesses/rlm/harness.py @@ -2,6 +2,7 @@ import hashlib import logging +import re import secrets import shlex from typing import Any, Literal @@ -200,16 +201,17 @@ async def setup_execution( token = secrets.token_urlsafe(32) port = await connection.serve( workspace, - lambda port: [ + [ f"{self._install_dir()}/tools/rlm/bin/python", "-m", "rlm.execution", "--listen", "0.0.0.0", "--port", - str(port), + "0", ], {"RLM_EXECUTION_TOKEN": token}, + re.compile(r"^listening on \S+:(\d+)$", re.MULTILINE), ) self._execution = { "execution_tcp": {"host": connection.host, "port": port, "token": token}, diff --git a/verifiers/v1/runtimes/deployment.py b/verifiers/v1/runtimes/deployment.py index 1eb6c23ad9..3864a2df3c 100644 --- a/verifiers/v1/runtimes/deployment.py +++ b/verifiers/v1/runtimes/deployment.py @@ -5,9 +5,9 @@ import asyncio import contextlib import logging -import random +import re import uuid -from collections.abc import AsyncIterator, Callable, Collection, Mapping +from collections.abc import AsyncIterator, Collection, Mapping from contextlib import AsyncExitStack, asynccontextmanager from dataclasses import dataclass, field @@ -41,190 +41,69 @@ class ServicesSpec: services: dict[str, PrimeConfig] -_WORKER_PORTS = range(20000, 32000) -"""Below Linux's ephemeral range, so a worker's port cannot collide with an outgoing -connection's.""" -_HOLD = ( - 'echo $$ > "$1.pid"; while [ ! -e "$2" ]; do sleep 0.05; done; shift 2; exec "$@"' -) -"""Wait, as the worker's user, until root has moved this process into its cgroup.""" -_FLAGS = "/run/vf-exec" -"""Root-owned, so the worker's user can neither forge nor redirect a release flag.""" -_HELD = """ -uid=$(id -u "${user:-root}") -held() { - # Whether $1 is this launch's held worker, running as the worker's user: root - # only acts on a process that user could signal itself. - local key real arg - [[ $1 =~ ^[0-9]+$ ]] || return 1 - while read -r key real _; do [ "$key" = Uid: ] && break; done < "/proc/$1/status" || return 1 - [ "$real" = "$uid" ] || return 1 - while IFS= read -r -d '' arg; do [ "$arg" = "$flag" ] && return 0; done < "/proc/$1/cmdline" - return 1 -} -""" -_PORT_TAKEN = 2 -_JOIN = ( - """ -set -e -group=/sys/fs/cgroup/$1 files=$2 flag=$3 port=$4 user=$5 -""" - + _HELD - + """ -printf -v hex '%04X' "$port" -listening() { - cat /proc/net/tcp /proc/net/tcp6 2>/dev/null | while read -r _ address _ state _ _ _ _ _ inode _; do - if [ "$state" = 0A ] && [ "${address##*:}" = "$hex" ]; then echo "$inode"; fi - done -} -for _ in $(seq 300); do [ -s "$files.pid" ] && break; sleep 0.1; done -pid=$(cat "$files.pid") -held "$pid" || { echo "worker identity check failed" >&2; exit 1; } -echo "$pid" -[ -z "$(listening)" ] || exit 2 -mkdir "$group" -echo "$pid" > "$group/cgroup.procs" -mkdir -p -m 755 "${flag%/*}" -[ "$(stat -c '%u %a' "${flag%/*}")" = "0 755" ] || { echo "unsafe flag directory" >&2; exit 1; } -: > "$flag" -for _ in $(seq 600); do - for inode in $(listening); do - for proc in $(cat "$group/cgroup.procs"); do - for fd in /proc/"$proc"/fd/*; do - [ "$(readlink "$fd")" = "socket:[$inode]" ] && exit 0 - done - done - done - kill -0 "$pid" 2>/dev/null || { echo "worker exited" >&2; exit 1; } - sleep 0.1 +_MARK = "VF_EXEC_ID" +"""Environment variable that marks a worker and every process its tools start, which +inherit it through detaching and reparenting.""" +_EXITED = "vf-exec: worker exited" +_STOP = """ +for proc in /proc/[0-9]*; do + while IFS= read -r -d '' entry; do + if [ "$entry" = "$1" ]; then kill -9 "${proc#/proc/}" 2>/dev/null; break; fi + done 2>/dev/null < "$proc/environ" done -echo "worker did not listen on port $port" >&2 -exit 1 -""" -) -"""Validate the held worker and print its PID, move it into its cgroup and release -it, then wait until a process in that cgroup listens on the port. Exits -`_PORT_TAKEN` before moving the worker if another process already listens there.""" -_KILL = ( - """ -group=/sys/fs/cgroup/$1 files=$2 flag=$3 pid=$4 user=$5 -""" - + _HELD - + """ -if [ -e "$group/cgroup.kill" ]; then - echo 1 > "$group/cgroup.kill" - for _ in $(seq 50); do rmdir "$group" 2>/dev/null && break; sleep 0.1; done -fi -if held "$pid"; then kill -9 "$pid"; fi -rm -f "$flag" "$files.pid" """ -) -"""Kill the worker's cgroup, or a worker never released into it by the PID that -`_JOIN` validated, checked again.""" - - -@dataclass -class _Worker: - root: Runtime - group: str - user: str - pid: str = "" - - @property - def files(self) -> str: - return f"/tmp/{self.group}" - - @property - def flag(self) -> str: - return f"{_FLAGS}/{self.group}" +"""Kill every process whose environment holds the marker `$1`, as the worker's user, +so it reaches nothing that user could not signal itself.""" @dataclass class ExecutionConnection: """A split deployment's way into the workspace for the harness's tools: workers started there listen on the rollout's private network, where the harness reaches - the workspace at `host`. Each worker runs in a cgroup of its own, so `close` stops - every process its tools started, detached or not.""" + the workspace at `host`. A worker and every process its tools start carry a marker + in their environment, so `close` stops them, detached or not.""" host: str """The workspace's name on the private network.""" - _workers: list[_Worker] = field(default_factory=list) + _workers: list[tuple[Runtime, str]] = field(default_factory=list) async def serve( self, runtime: Runtime, - argv: Callable[[int], list[str]], + argv: list[str], env: dict[str, str], + announce: re.Pattern[str], ) -> int: - """Start `argv(port)` through `runtime`, a workspace view, and return the port - once the worker itself listens on it.""" - for _ in range(3): - port = random.choice(_WORKER_PORTS) - worker = _Worker( - runtime.with_user(None), - f"vf-exec-{uuid.uuid4().hex[:12]}", - runtime.user or "", - ) - self._workers.append(worker) - await runtime.run_background( - ["sh", "-c", _HOLD, "sh", worker.files, worker.flag, *argv(port)], - env, - f"{worker.files}.log", - ) - result = await worker.root.run( - [ - "bash", - "-c", - _JOIN, - "bash", - worker.group, - worker.files, - worker.flag, - str(port), - worker.user, - ], - {}, - ) - worker.pid = result.stdout.strip() - if not result.exit_code: - return port - if result.exit_code != _PORT_TAKEN: - # Read as the worker's user: the log lives where that user can write. - log = await runtime.run( - ["tail", "-c", "2000", f"{worker.files}.log"], {} - ) - raise SandboxError( - "execution worker failed to start: " - f"{result.stderr.strip()[-500:]}\n{log.stdout}" - ) - # A workspace service holds the port: release the held worker, try another. - self._workers.remove(worker) - await _stop(worker) - raise SandboxError("execution worker found no free port in the workspace") + """Start `argv` through `runtime`, a workspace view, and return the port it + listens on. The worker binds a port the kernel picks and prints it in a line + that `announce` matches, with the port as its first group.""" + worker = uuid.uuid4().hex + log = f"/tmp/vf-exec-{worker}.log" + self._workers.append((runtime, f"{_MARK}={worker}")) + await runtime.run_background( + ["sh", "-c", f'"$@"; echo "{_EXITED} ($?)"', "sh", *argv], + {**env, _MARK: worker}, + log, + ) + output = "" + with contextlib.suppress(TimeoutError): + async with asyncio.timeout(60): + while True: + with contextlib.suppress(SandboxError): # not written yet + output = (await runtime.read(log)).decode(errors="replace") + if found := announce.search(output): + return int(found.group(1)) + if _EXITED in output: + break + await asyncio.sleep(0.2) + raise SandboxError(f"execution worker did not start: {output[-2000:]}") async def close(self) -> None: - """Stop every worker and the processes they started.""" + """Stop every worker and the processes their tools started.""" workers, self._workers = self._workers, [] - for worker in workers: - await _stop(worker) - - -async def _stop(worker: _Worker) -> None: - with contextlib.suppress(Exception): - await worker.root.run( - [ - "bash", - "-c", - _KILL, - "bash", - worker.group, - worker.files, - worker.flag, - worker.pid, - worker.user, - ], - {}, - ) + for runtime, marker in workers: + with contextlib.suppress(Exception): + await runtime.run(["bash", "-c", _STOP, "bash", marker], {}) class _DeploymentStrategy: From 8528fbfcbfedad6f79ce07352b55bf376004210c Mon Sep 17 00:00:00 2001 From: hallerite Date: Sat, 31 Oct 2026 01:23:21 +0000 Subject: [PATCH 07/12] Pin nano-rlm to its split-execution head The head stops native shell commands with their turn and session, and names the worker's history in compaction notices under split execution. Co-Authored-By: Claude Opus 5.5 --- verifiers/v1/harnesses/rlm/harness.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/verifiers/v1/harnesses/rlm/harness.py b/verifiers/v1/harnesses/rlm/harness.py index 2cddd2d6aa..024c226176 100644 --- a/verifiers/v1/harnesses/rlm/harness.py +++ b/verifiers/v1/harnesses/rlm/harness.py @@ -72,7 +72,7 @@ class CompactionConfig(BaseConfig): class RLMHarnessConfig(HarnessConfig): dialect: Literal["chat_completions"] = "chat_completions" version: str = Field( - default="ff8767d3bdc311ca55a98fea2c6d4438c006741a", min_length=1 + default="7cbe516e142c47ed0fde34834e7fbea47edbd94b", min_length=1 ) """Git ref (branch, tag, or commit) of nano-rlm to install. Must know every field this harness puts on the wire, i.e. be at least the default ref.""" From 1ad0c8c1ec6366f7cb71f09ab867c7648669d5b5 Mon Sep 17 00:00:00 2001 From: hallerite Date: Sat, 31 Oct 2026 01:38:26 +0000 Subject: [PATCH 08/12] Treat an execution worker that exited as not started A worker that crashes right after announcing its port leaves both its listen line and the exit marker in the log by the first read; serve returned the port. Check for the exit marker first. Co-Authored-By: Claude Opus 5.5 --- tests/unit/runtimes/test_private_network.py | 32 ++++++++++++++++++++- verifiers/v1/runtimes/deployment.py | 5 ++-- 2 files changed, 34 insertions(+), 3 deletions(-) diff --git a/tests/unit/runtimes/test_private_network.py b/tests/unit/runtimes/test_private_network.py index b774c5a516..f9cb510860 100644 --- a/tests/unit/runtimes/test_private_network.py +++ b/tests/unit/runtimes/test_private_network.py @@ -1,11 +1,12 @@ import asyncio +import re from contextlib import asynccontextmanager from types import SimpleNamespace import pytest import verifiers.v1 as vf -from verifiers.v1.errors import ToolsetError +from verifiers.v1.errors import SandboxError, ToolsetError from verifiers.v1.mcp import launch from verifiers.v1.runtimes import ( DockerBackend, @@ -15,6 +16,7 @@ private_network, ) from verifiers.v1.runtimes.backends import prime +from verifiers.v1.runtimes.deployment import ExecutionConnection from verifiers.v1.runtimes.strategies import ( _ColocatedDeployment, _ContainerDeployment, @@ -254,3 +256,31 @@ def test_an_explicit_service_needs_a_task_service_spec(monkeypatch, service): def test_a_restricted_runtime_without_a_private_network_is_refused(monkeypatch): with pytest.raises(ToolsetError, match="joins the rollout's private network"): served_url(monkeypatch, toolset("forth", runtime=FORTH)) + + +@pytest.mark.parametrize("exited", [False, True]) +async def test_an_execution_worker_is_ready_only_while_it_runs(exited): + """A worker that crashed right after announcing its port leaves both lines in its + log by the first read; it is not ready.""" + log = b"listening on 0.0.0.0:41234\n" + ( + b"vf-exec: worker exited (1)\n" if exited else b"" + ) + + class Workspace: + async def run_background(self, argv, env, path): + pass + + async def read(self, path): + return log + + serve = ExecutionConnection("10.0.0.2").serve( + Workspace(), + ["worker"], + {}, + re.compile(r"^listening on \S+:(\d+)$", re.MULTILINE), + ) + if exited: + with pytest.raises(SandboxError, match="did not start"): + await serve + else: + assert await serve == 41234 diff --git a/verifiers/v1/runtimes/deployment.py b/verifiers/v1/runtimes/deployment.py index 3864a2df3c..210aa2d6da 100644 --- a/verifiers/v1/runtimes/deployment.py +++ b/verifiers/v1/runtimes/deployment.py @@ -91,10 +91,11 @@ async def serve( while True: with contextlib.suppress(SandboxError): # not written yet output = (await runtime.read(log)).decode(errors="replace") - if found := announce.search(output): - return int(found.group(1)) + # A worker that crashed right after announcing has both lines. if _EXITED in output: break + if found := announce.search(output): + return int(found.group(1)) await asyncio.sleep(0.2) raise SandboxError(f"execution worker did not start: {output[-2000:]}") From 4c1fc8a50e964d4eca0002e04493a4b89e7effe3 Mon Sep 17 00:00:00 2001 From: hallerite Date: Sat, 31 Oct 2026 01:38:27 +0000 Subject: [PATCH 09/12] Pin nano-rlm to its split-execution head The head syncs the session ledger to the execution worker incrementally, so long split sessions no longer outgrow one 16 MiB frame. Co-Authored-By: Claude Opus 5.5 --- verifiers/v1/harnesses/rlm/harness.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/verifiers/v1/harnesses/rlm/harness.py b/verifiers/v1/harnesses/rlm/harness.py index 024c226176..baec9b756a 100644 --- a/verifiers/v1/harnesses/rlm/harness.py +++ b/verifiers/v1/harnesses/rlm/harness.py @@ -72,7 +72,7 @@ class CompactionConfig(BaseConfig): class RLMHarnessConfig(HarnessConfig): dialect: Literal["chat_completions"] = "chat_completions" version: str = Field( - default="7cbe516e142c47ed0fde34834e7fbea47edbd94b", min_length=1 + default="d98bc273ad3a8004f87fd205542c1f5dd8d0a6d2", min_length=1 ) """Git ref (branch, tag, or commit) of nano-rlm to install. Must know every field this harness puts on the wire, i.e. be at least the default ref.""" From 4a9c28a7866bc8f42269077c6cf5aa5728875c09 Mon Sep 17 00:00:00 2001 From: hallerite Date: Sat, 31 Oct 2026 02:10:41 +0000 Subject: [PATCH 10/12] Drop the execution worker's unit test Co-Authored-By: Claude Opus 5.5 --- tests/unit/runtimes/test_private_network.py | 32 +-------------------- 1 file changed, 1 insertion(+), 31 deletions(-) diff --git a/tests/unit/runtimes/test_private_network.py b/tests/unit/runtimes/test_private_network.py index f9cb510860..b774c5a516 100644 --- a/tests/unit/runtimes/test_private_network.py +++ b/tests/unit/runtimes/test_private_network.py @@ -1,12 +1,11 @@ import asyncio -import re from contextlib import asynccontextmanager from types import SimpleNamespace import pytest import verifiers.v1 as vf -from verifiers.v1.errors import SandboxError, ToolsetError +from verifiers.v1.errors import ToolsetError from verifiers.v1.mcp import launch from verifiers.v1.runtimes import ( DockerBackend, @@ -16,7 +15,6 @@ private_network, ) from verifiers.v1.runtimes.backends import prime -from verifiers.v1.runtimes.deployment import ExecutionConnection from verifiers.v1.runtimes.strategies import ( _ColocatedDeployment, _ContainerDeployment, @@ -256,31 +254,3 @@ def test_an_explicit_service_needs_a_task_service_spec(monkeypatch, service): def test_a_restricted_runtime_without_a_private_network_is_refused(monkeypatch): with pytest.raises(ToolsetError, match="joins the rollout's private network"): served_url(monkeypatch, toolset("forth", runtime=FORTH)) - - -@pytest.mark.parametrize("exited", [False, True]) -async def test_an_execution_worker_is_ready_only_while_it_runs(exited): - """A worker that crashed right after announcing its port leaves both lines in its - log by the first read; it is not ready.""" - log = b"listening on 0.0.0.0:41234\n" + ( - b"vf-exec: worker exited (1)\n" if exited else b"" - ) - - class Workspace: - async def run_background(self, argv, env, path): - pass - - async def read(self, path): - return log - - serve = ExecutionConnection("10.0.0.2").serve( - Workspace(), - ["worker"], - {}, - re.compile(r"^listening on \S+:(\d+)$", re.MULTILINE), - ) - if exited: - with pytest.raises(SandboxError, match="did not start"): - await serve - else: - assert await serve == 41234 From 66a38d97184f658ee6885d9f25518837fe5a57e8 Mon Sep 17 00:00:00 2001 From: hallerite Date: Sun, 1 Nov 2026 01:58:39 +0000 Subject: [PATCH 11/12] Leave daemons running when a split RLM worker closes Closing the execution connection killed every process that inherited the worker's marker, including daemons the agent started, while a colocated harness's exit leaves them running. The harness closes before grading when the agent did not finish, so a daemon the task needs was gone only in split mode. nano-rlm stops what is still attached to a session as the session closes, so an RLM worker's close now stops just the worker and its own process group. Codex's exec-server keeps the full stop. Co-Authored-By: Claude Opus 5.5 --- verifiers/v1/harnesses/rlm/harness.py | 3 ++ verifiers/v1/runtimes/deployment.py | 64 +++++++++++++++++++++++---- 2 files changed, 59 insertions(+), 8 deletions(-) diff --git a/verifiers/v1/harnesses/rlm/harness.py b/verifiers/v1/harnesses/rlm/harness.py index baec9b756a..0518248478 100644 --- a/verifiers/v1/harnesses/rlm/harness.py +++ b/verifiers/v1/harnesses/rlm/harness.py @@ -212,6 +212,9 @@ async def setup_execution( ], {"RLM_EXECUTION_TOKEN": token}, re.compile(r"^listening on \S+:(\d+)$", re.MULTILINE), + # A session that closes stops what is still attached to its kernel, as it + # does colocated; a daemon the agent started outlives the harness. + keep_detached=True, ) self._execution = { "execution_tcp": {"host": connection.host, "port": port, "token": token}, diff --git a/verifiers/v1/runtimes/deployment.py b/verifiers/v1/runtimes/deployment.py index 210aa2d6da..f5d96c71ab 100644 --- a/verifiers/v1/runtimes/deployment.py +++ b/verifiers/v1/runtimes/deployment.py @@ -45,6 +45,8 @@ class ServicesSpec: """Environment variable that marks a worker and every process its tools start, which inherit it through detaching and reparenting.""" _EXITED = "vf-exec: worker exited" +_WRAPPER = f'"$@"; echo "{_EXITED} ($?)"' +"""The script a worker runs under, which reports the worker's exit.""" _STOP = """ for proc in /proc/[0-9]*; do while IFS= read -r -d '' entry; do @@ -54,6 +56,39 @@ class ServicesSpec: """ """Kill every process whose environment holds the marker `$1`, as the worker's user, so it reaches nothing that user could not signal itself.""" +_STOP_GROUP = """ +marker=$1 wrapper=$2 group= +marked() { + while IFS= read -r -d '' entry; do + [ "$entry" = "$marker" ] && return 0 + done 2>/dev/null < "$1/environ" + return 1 +} +group_of() { + local stat + read -r stat 2>/dev/null < "$1/stat" || return + stat=${stat##*) } + set -- $stat + echo "$3" +} +for proc in /proc/[0-9]*; do + marked "$proc" || continue + while IFS= read -r -d '' arg; do + if [ "$arg" = "$wrapper" ]; then group=$(group_of "$proc"); break; fi + done 2>/dev/null < "$proc/cmdline" + [ -n "$group" ] && break +done +[ -n "$group" ] || exit 0 +for proc in /proc/[0-9]*; do + if marked "$proc" && [ "$(group_of "$proc")" = "$group" ]; then + kill -9 "${proc#/proc/}" 2>/dev/null + fi +done +exit 0 +""" +"""Kill the worker whose wrapper runs the script `$2`, with what runs in its process +group: the processes there whose environment holds the marker `$1`, as the worker's +user.""" @dataclass @@ -61,11 +96,11 @@ class ExecutionConnection: """A split deployment's way into the workspace for the harness's tools: workers started there listen on the rollout's private network, where the harness reaches the workspace at `host`. A worker and every process its tools start carry a marker - in their environment, so `close` stops them, detached or not.""" + in their environment, so `close` finds them wherever they were reparented.""" host: str """The workspace's name on the private network.""" - _workers: list[tuple[Runtime, str]] = field(default_factory=list) + _workers: list[tuple[Runtime, str, bool]] = field(default_factory=list) async def serve( self, @@ -73,15 +108,22 @@ async def serve( argv: list[str], env: dict[str, str], announce: re.Pattern[str], + *, + keep_detached: bool = False, ) -> int: """Start `argv` through `runtime`, a workspace view, and return the port it listens on. The worker binds a port the kernel picks and prints it in a line - that `announce` matches, with the port as its first group.""" + that `announce` matches, with the port as its first group. + + `close` stops the worker and every process its tools started, detached or not; + with `keep_detached`, for a worker that stops what is still attached to a + session as the session closes, only the worker and its own process group, so + a daemon its tools started keeps running, as after a colocated harness exits.""" worker = uuid.uuid4().hex log = f"/tmp/vf-exec-{worker}.log" - self._workers.append((runtime, f"{_MARK}={worker}")) + self._workers.append((runtime, f"{_MARK}={worker}", keep_detached)) await runtime.run_background( - ["sh", "-c", f'"$@"; echo "{_EXITED} ($?)"', "sh", *argv], + ["sh", "-c", _WRAPPER, "sh", *argv], {**env, _MARK: worker}, log, ) @@ -100,11 +142,17 @@ async def serve( raise SandboxError(f"execution worker did not start: {output[-2000:]}") async def close(self) -> None: - """Stop every worker and the processes their tools started.""" + """Stop every worker, with the processes their tools started that `serve` says + to stop.""" workers, self._workers = self._workers, [] - for runtime, marker in workers: + for runtime, marker, keep_detached in workers: + argv = ( + ["bash", "-c", _STOP_GROUP, "bash", marker, _WRAPPER] + if keep_detached + else ["bash", "-c", _STOP, "bash", marker] + ) with contextlib.suppress(Exception): - await runtime.run(["bash", "-c", _STOP, "bash", marker], {}) + await runtime.run(argv, {}) class _DeploymentStrategy: From e4b0ab1a412679ad7533f5e0b4996e899354339c Mon Sep 17 00:00:00 2001 From: hallerite Date: Sat, 31 Oct 2026 02:32:47 +0000 Subject: [PATCH 12/12] Pin nano-rlm to its split-execution merge nano-rlm#218 runs the supervisor's shell jobs, services and path watches in the workspace, so split execution refuses none of them and keeps the runtime prompt. Co-Authored-By: Claude Opus 5.5 --- verifiers/v1/harnesses/rlm/harness.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/verifiers/v1/harnesses/rlm/harness.py b/verifiers/v1/harnesses/rlm/harness.py index 0518248478..a0c83e6bc1 100644 --- a/verifiers/v1/harnesses/rlm/harness.py +++ b/verifiers/v1/harnesses/rlm/harness.py @@ -72,7 +72,7 @@ class CompactionConfig(BaseConfig): class RLMHarnessConfig(HarnessConfig): dialect: Literal["chat_completions"] = "chat_completions" version: str = Field( - default="d98bc273ad3a8004f87fd205542c1f5dd8d0a6d2", min_length=1 + default="208aea96cdd6e59fd8551ac543b837a76cda4b9f", min_length=1 ) """Git ref (branch, tag, or commit) of nano-rlm to install. Must know every field this harness puts on the wire, i.e. be at least the default ref."""