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/docs/runtimes.md b/docs/runtimes.md index 455f0370c2..a4b78b76ff 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 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: 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/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 new file mode 100644 index 0000000000..3d0f7912f5 --- /dev/null +++ b/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") 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..a0c83e6bc1 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 @@ -22,6 +23,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 +72,7 @@ class CompactionConfig(BaseConfig): class RLMHarnessConfig(HarnessConfig): dialect: Literal["chat_completions"] = "chat_completions" version: str = Field( - default="e4d4407f1fe6f4bd82b2c1da26ccd737627a6543", 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.""" @@ -137,13 +139,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 +186,41 @@ 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, + [ + f"{self._install_dir()}/tools/rlm/bin/python", + "-m", + "rlm.execution", + "--listen", + "0.0.0.0", + "--port", + "0", + ], + {"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}, + "execution_cwd": workspace.config.workdir, + "kernel_env": dict(workspace.env), + } def _runtime_metadata( self, @@ -249,6 +291,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..4d31835778 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 @@ -303,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, @@ -311,6 +319,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 +367,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 +383,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 +402,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 +520,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 +534,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 +568,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 +681,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 +717,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..f5d96c71ab 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 re +import uuid +from collections.abc import AsyncIterator, Collection, 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,120 @@ class ServicesSpec: services: dict[str, PrimeConfig] +_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" +_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 + if [ "$entry" = "$1" ]; then kill -9 "${proc#/proc/}" 2>/dev/null; break; fi + done 2>/dev/null < "$proc/environ" +done +""" +"""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 +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` finds them wherever they were reparented.""" + + host: str + """The workspace's name on the private network.""" + _workers: list[tuple[Runtime, str, bool]] = field(default_factory=list) + + async def serve( + self, + runtime: Runtime, + 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. + + `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}", keep_detached)) + await runtime.run_background( + ["sh", "-c", _WRAPPER, "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") + # 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:]}") + + async def close(self) -> None: + """Stop every worker, with the processes their tools started that `serve` says + to stop.""" + workers, self._workers = self._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(argv, {}) + + class _DeploymentStrategy: """Placement, connectivity and lifecycle hooks for one deployment attempt.""" @@ -79,6 +197,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, @@ -88,6 +207,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 @@ -133,6 +254,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..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 ( @@ -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,14 @@ 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, + 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.""" + network of the rollout's own as it boots. A split deployment's `harness` VM joins + 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 " @@ -93,14 +100,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 +107,27 @@ 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}")) + # 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; @@ -152,13 +172,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..c4d9eaabf3 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,25 @@ 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), + harness_peers=d.tool_services, ) ) 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 +115,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 +127,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 +206,31 @@ 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 + if harness.creates_per_min is not None + else 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 +252,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