diff --git a/pyproject.toml b/pyproject.toml index a8602341d..b9ceda75f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,7 +35,7 @@ dependencies = [ "openai>=2.9.0", "openai-agents>=0.8.2", "prime-tunnel>=0.1.8", - "prime-sandboxes>=0.2.33", + "prime-sandboxes>=0.2.35", "pydantic>=2.12.3", "requests", "rich>=11.0.0", diff --git a/tests/v1/test_e2e.py b/tests/v1/test_e2e.py index 4f6a5f194..c50af639a 100644 --- a/tests/v1/test_e2e.py +++ b/tests/v1/test_e2e.py @@ -51,16 +51,18 @@ def _pair(a: str, b: str, id: str, *extra_marks): pytest.param("modal", marks=[_m.modal], id="harness-in-modal"), ] -# ACP-backed harnesses: each must preserve an exchange across process relaunches and +# ACP-backed harnesses: each must preserve an exchange across interaction segments and # retain MCP access after resuming. Cover every harness in the local container runtime, -# plus one remote placement for the sandbox/tunnel boundary. +# plus remote placements for the sandbox/tunnel and native-process boundaries. ACP_RESUME_PLACEMENTS = [ _pair("hermes-agent", "docker", "hermes-agent-acp-in-docker"), + _pair("rlm", "docker", "rlm-acp-in-docker"), _pair("kimi-code", "docker", "kimi-code-acp-in-docker"), _pair("pi", "docker", "pi-acp-in-docker"), _pair("pool", "docker", "pool-acp-in-docker"), _pair("openclaw", "docker", "openclaw-acp-in-docker"), _pair("pool", "prime", "pool-acp-in-prime"), + _pair("rlm", "prime", "rlm-acp-in-prime-vm"), ] # harness runtime x tool placement: every axis value once plus the two-container case @@ -215,7 +217,10 @@ async def test_acp_resume_with_tool(run_v1, harness, harness_runtime, tmp_path): (trace,) = await run_v1( "echo-acp-resume-v1", harness=harness, - runtime={"type": harness_runtime}, + runtime={ + "type": harness_runtime, + **({"vm": True} if harness_runtime == "prime" else {}), + }, output_dir=tmp_path, max_turns=8, max_tokens=8192, @@ -231,6 +236,8 @@ async def test_acp_resume_with_tool(run_v1, harness, harness_runtime, tmp_path): assert segments[1]["terminated"] is False assert "tool" in segments[1]["roles"] assert segments[1]["tool_outputs"] + if harness == "rlm": + assert "turns_since_last_compaction" in trace.metrics @pytest.mark.e2e diff --git a/uv.lock b/uv.lock index 38298ea39..d36ca612f 100644 --- a/uv.lock +++ b/uv.lock @@ -3136,7 +3136,7 @@ toml = [ [[package]] name = "prime-sandboxes" -version = "0.2.33" +version = "0.2.35" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "aiofiles" }, @@ -3146,9 +3146,9 @@ dependencies = [ { name = "pydantic" }, { name = "tenacity" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/15/a9/880685cefd503aa92d2b49da46b60ba807015f8839c344a83d8c5bc02f30/prime_sandboxes-0.2.33.tar.gz", hash = "sha256:4253a3c345ccf6c07b41a81790f8515763333f094d20bc405a257432461e81d8", size = 81228, upload-time = "2026-07-22T23:23:55.715Z" } +sdist = { url = "https://files.pythonhosted.org/packages/57/cc/287c0db192ddbe003bd5882d1aab5461cdb3912d0c92db7b39734a9e828e/prime_sandboxes-0.2.35.tar.gz", hash = "sha256:e716bfccdcb51aefd5e2d48fdfc3a79bea8c6825bce1871f49f095ced03efc94", size = 86226, upload-time = "2026-08-05T14:29:53.33Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/f8/1b/10b1044b8aaef561c32a34140f4a13c7310f73f92f9546337e2e8e6a5239/prime_sandboxes-0.2.33-py3-none-any.whl", hash = "sha256:92c9647557e76191ba926ac8ed01708de9ec4450142131b673cd5cf8d9d7ea56", size = 42773, upload-time = "2026-07-22T23:23:54.381Z" }, + { url = "https://files.pythonhosted.org/packages/25/07/c186395925f780b1a18b350f562ebc4e8a64a182c37bd1e3db3948f297e2/prime_sandboxes-0.2.35-py3-none-any.whl", hash = "sha256:29d7057532b21545be5938d3aee822ae3a54085a45934d5c3ac55190e06c5bbb", size = 47189, upload-time = "2026-08-05T14:29:52.142Z" }, ] [[package]] diff --git a/verifiers/v1/__init__.py b/verifiers/v1/__init__.py index f58e3dba2..b3e12616b 100644 --- a/verifiers/v1/__init__.py +++ b/verifiers/v1/__init__.py @@ -43,7 +43,7 @@ TunnelError, ) from verifiers.v1.graph import MessageNode -from verifiers.v1.harness import Harness +from verifiers.v1.harness import Harness, HarnessSession from verifiers.v1.judge import Judge, JudgeResponse, JudgeView from verifiers.v1.judges import ( Criterion, @@ -64,6 +64,7 @@ Runtime, RuntimeConfig, RuntimeInfo, + RuntimeProcess, SubprocessConfig, ) from verifiers.v1.state import State, StateT @@ -241,10 +242,12 @@ "TasksetConfig", "BaseConfig", "Harness", + "HarnessSession", "HarnessConfig", "ACP", "ModelContext", "Runtime", + "RuntimeProcess", "RuntimeConfig", "RuntimeInfo", "ProgramResult", diff --git a/verifiers/v1/acp/__init__.py b/verifiers/v1/acp/__init__.py index ef61209dc..2af8c6e4b 100644 --- a/verifiers/v1/acp/__init__.py +++ b/verifiers/v1/acp/__init__.py @@ -1,28 +1,68 @@ """Public Agent Client Protocol support for harness programs.""" +import asyncio +import contextlib import json import secrets +from collections.abc import AsyncIterator from pathlib import Path +from verifiers.v1.clients import ModelContext from verifiers.v1.dialects.chat import message_to_wire -from verifiers.v1.harness import Harness -from verifiers.v1.runtimes import ProgramResult, Runtime +from verifiers.v1.errors import HarnessError +from verifiers.v1.harness import Harness, HarnessSession +from verifiers.v1.runtimes import ProgramResult, Runtime, RuntimeProcess +from verifiers.v1.task import TaskData +from verifiers.v1.trace import Trace from verifiers.v1.types import Messages from verifiers.v1.utils.aio import run_shielded -ACP_SOURCE = (Path(__file__).resolve().parent / "_runner.py").read_text() +ACP_SOURCE = (Path(__file__).resolve().parent / "runner.py").read_text() +MAX_PACKET_BYTES = 128 * 1024 * 1024 __all__ = ["ACP"] class ACP: - """Run an ACP agent.""" + """Run one-shot ACP agents or create rollout-scoped ACP sessions.""" async def setup(self, harness: Harness, runtime: Runtime) -> None: await runtime.prepare_uv_script( ACP_SOURCE, {**harness.config.resolved_env, "UV_FROZEN": "false"} ) + def session( + self, + harness: Harness, + ctx: ModelContext, + trace: Trace, + runtime: Runtime, + endpoint: str, + secret: str, + mcp_urls: dict[str, str], + data: TaskData, + *, + env: dict[str, str], + command: list[str], + prompt: str | Messages | None, + system_prompt: str | None = None, + ) -> "ACPHarnessSession": + """Create a persistent ACP-backed handle owned by one rollout.""" + return ACPHarnessSession( + harness, + ctx, + trace, + runtime, + endpoint, + secret, + mcp_urls, + data, + env=env, + command=command, + prompt=prompt, + system_prompt=system_prompt, + ) + async def run( self, runtime: Runtime, @@ -34,6 +74,30 @@ async def run( system_prompt: str | None = None, session_path: str | None = None, allow_empty_tool_reply: bool = False, + ) -> ProgramResult: + """Run one ACP segment without retaining its process.""" + return await self._run( + runtime, + env, + command, + prompt, + mcp_urls=mcp_urls, + system_prompt=system_prompt, + session_path=session_path, + allow_empty_tool_reply=allow_empty_tool_reply, + ) + + async def _run( + self, + runtime: Runtime, + env: dict[str, str], + command: list[str], + prompt: str | Messages | None, + *, + mcp_urls: dict[str, str] | None = None, + system_prompt: str | None = None, + session_path: str | None = None, + allow_empty_tool_reply: bool = False, ) -> ProgramResult: if prompt is None: raise ValueError("ACP requires a prompt") @@ -60,7 +124,180 @@ async def run( path = f"{directory}/config.json" try: await runtime.write(path, json.dumps(config).encode()) - result = await runtime.run_program([*program, path], env) - return result + return await runtime.run_program([*program, "once", path], env) finally: await run_shielded(runtime.run(["rm", "-rf", directory], {})) + + +def _packet(value: dict) -> bytes: + data = json.dumps(value, ensure_ascii=False).encode() + if len(data) > MAX_PACKET_BYTES: + raise ValueError(f"ACP session packet is too large: {len(data)} bytes") + return len(data).to_bytes(8, "big") + data + + +class _PacketReader: + def __init__(self, source: AsyncIterator[bytes]) -> None: + self._source = source.__aiter__() + self._buffer = bytearray() + + async def _readexactly(self, size: int) -> bytes: + while len(self._buffer) < size: + try: + self._buffer.extend(await anext(self._source)) + except StopAsyncIteration as e: + raise EOFError("ACP process closed its stdout") from e + data = bytes(self._buffer[:size]) + del self._buffer[:size] + return data + + async def read(self) -> dict: + size = int.from_bytes(await self._readexactly(8), "big") + if size > MAX_PACKET_BYTES: + raise ValueError(f"ACP session packet is too large: {size} bytes") + return json.loads((await self._readexactly(size)).decode()) + + +class ACPHarnessSession(HarnessSession): + """A live ACP process, connection, and native session for one rollout.""" + + def __init__( + self, + harness: Harness, + ctx: ModelContext, + trace: Trace, + runtime: Runtime, + endpoint: str, + secret: str, + mcp_urls: dict[str, str], + data: TaskData, + env: dict[str, str], + command: list[str], + prompt: str | Messages | None, + system_prompt: str | None, + ) -> None: + super().__init__(harness, ctx, trace, runtime, endpoint, secret, mcp_urls, data) + self.env = env + self.command = command + self.prompt = prompt + self.system_prompt = system_prompt + self._process: RuntimeProcess | None = None + self._reader: _PacketReader | None = None + self._stderr_tail = bytearray() + self._stderr_task: asyncio.Task[None] | None = None + self._lock = asyncio.Lock() + + async def _start(self) -> None: + self._stderr_tail.clear() + program = await self.runtime.prepare_uv_script( + ACP_SOURCE, {**self.env, "UV_FROZEN": "false"} + ) + process = await self.runtime.open_process([*program, "stream"], self.env) + self._process = process + self._reader = _PacketReader(process.stdout) + self._stderr_task = asyncio.create_task(self._drain_stderr(process.stderr)) + + async def _drain_stderr(self, stream: AsyncIterator[bytes]) -> None: + async for chunk in stream: + self._stderr_tail.extend(chunk) + if len(self._stderr_tail) > 4000: + del self._stderr_tail[:-4000] + + def _stderr(self) -> str: + return self._stderr_tail.decode(errors="replace").strip() + + async def _run(self, messages: Messages | None) -> ProgramResult: + prompt = self.prompt if messages is None else messages + if prompt is None: + raise ValueError("ACP requires a prompt") + wire_messages = ( + [{"role": "user", "content": prompt}] + if isinstance(prompt, str) + else [message_to_wire(message) for message in prompt] + ) + config = { + "command": self.command, + "messages": wire_messages, + "mcp_urls": self.mcp_urls, + "system_prompt": self.system_prompt or "", + "session_path": None, + } + async with self._lock: + if self._closed: + raise HarnessError( + f"harness {self.harness.config.id!r} session is already closed" + ) + if self._process is None: + await self._start() + assert self._process is not None + assert self._reader is not None + try: + await self._process.write( + _packet({"operation": "prompt", "config": config}) + ) + response = await self._reader.read() + except BaseException: + await run_shielded(self._stop(graceful=False)) + raise + if not response.get("ok"): + detail = response.get("error") or "ACP session request failed" + if stderr := self._stderr(): + detail = f"{detail}\n\nACP process stderr:\n{stderr}" + raise RuntimeError(detail) + return ProgramResult(exit_code=0, stdout=response.get("reply", ""), stderr="") + + async def _stop(self, *, graceful: bool) -> None: + process, self._process = self._process, None + reader, self._reader = self._reader, None + stderr_task, self._stderr_task = self._stderr_task, None + if process is None: + return + failure: BaseException | None = None + try: + if graceful and reader is not None: + try: + await process.write(_packet({"operation": "shutdown"})) + response = await asyncio.wait_for(reader.read(), timeout=10) + if not response.get("ok"): + raise RuntimeError( + response.get("error") or "ACP session shutdown failed" + ) + except BaseException as error: # noqa: BLE001 - finish teardown if cancelled + failure = error + try: + await asyncio.wait_for(process.wait(), timeout=10 if graceful else 0.1) + except BaseException: # noqa: BLE001 - cancellation still requires termination + with contextlib.suppress(Exception): + await asyncio.wait_for(process.terminate(), timeout=5) + try: + await asyncio.wait_for(process.wait(), timeout=5) + except BaseException: # noqa: BLE001 - cancellation still requires a kill + with contextlib.suppress(Exception): + await asyncio.wait_for(process.kill(), timeout=5) + with contextlib.suppress(BaseException): + await asyncio.wait_for(process.wait(), timeout=5) + finally: + if stderr_task is not None: + if not stderr_task.done(): + stderr_task.cancel() + with contextlib.suppress(BaseException): + await stderr_task + if failure is not None: + detail = str(failure) + if stderr := self._stderr(): + detail = f"{detail}\n\nACP process stderr:\n{stderr}" + raise RuntimeError(detail) from failure + + async def close(self) -> None: + if self._closed: + return + # Publish closure before waiting for the process lock. A turn that + # already passed HarnessSession.turn()'s fast check rechecks under the + # same lock in _run(), so it cannot restart after teardown. + await super().close() + + async def close_process() -> None: + async with self._lock: + await self._stop(graceful=True) + + await run_shielded(close_process()) diff --git a/verifiers/v1/acp/_runner.py b/verifiers/v1/acp/_runner.py deleted file mode 100644 index 086b7e814..000000000 --- a/verifiers/v1/acp/_runner.py +++ /dev/null @@ -1,213 +0,0 @@ -# /// script -# requires-python = ">=3.10,<3.15" -# dependencies = ["agent-client-protocol==0.11.0"] -# /// -"""Run one harness segment through an ACP agent.""" - -import asyncio -import json -import os -import sys -from pathlib import Path -from typing import Any - -from acp import ( - PROTOCOL_VERSION, - Client, - RequestError, - image_block, - spawn_agent_process, - text_block, -) -from acp.schema import ( - AgentMessageChunk, - AllowedOutcome, - ClientCapabilities, - DeniedOutcome, - HttpMcpServer, - PermissionOption, - RequestPermissionResponse, - TextContentBlock, - ToolCall, - ToolCallUpdate, -) - - -class VerifiersClient(Client): - def __init__(self) -> None: - self.visible_reply = "" - self.message_id: str | None = None - self.tool_calls: dict[str, str] = {} - - async def session_update(self, session_id: str, update: Any, **kwargs: Any) -> None: - if isinstance(update, ToolCall): - self.tool_calls[update.tool_call_id] = update.status or "pending" - return - if isinstance(update, ToolCallUpdate): - if update.status: - self.tool_calls[update.tool_call_id] = update.status - return - if not isinstance(update, AgentMessageChunk) or not isinstance( - update.content, TextContentBlock - ): - return - message_id = getattr(update, "message_id", None) - if message_id is not None and message_id != self.message_id: - self.visible_reply = "" - self.message_id = message_id - self.visible_reply += update.content.text - - async def request_permission( - self, - session_id: str, - tool_call: Any, - options: list[PermissionOption], - **kwargs: Any, - ) -> RequestPermissionResponse: - option = next( - (item for item in options if item.kind in ("allow_once", "allow_always")), - None, - ) - outcome = ( - AllowedOutcome(outcome="selected", option_id=option.option_id) - if option - else DeniedOutcome(outcome="cancelled") - ) - return RequestPermissionResponse(outcome=outcome) - - -def content_blocks(messages: list[dict], supports_images: bool) -> list: - blocks = [] - transcript = len(messages) != 1 or messages[0].get("role") != "user" - for message in messages: - if transcript: - separator = "\n\n" if blocks else "" - blocks.append( - text_block(f"{separator}[{message.get('role', 'message')}]\n") - ) - content = message.get("content") or "" - parts = ( - [{"type": "text", "text": content}] if isinstance(content, str) else content - ) - for part in parts: - if part["type"] == "text": - blocks.append(text_block(part["text"])) - continue - if not supports_images: - raise ValueError("ACP agent does not support image prompts") - url = part["image_url"]["url"] - metadata, separator, data = url.partition(",") - media_type, *parameters = metadata.removeprefix("data:").split(";") - if ( - not separator - or not metadata.startswith("data:image/") - or not any(value.lower() == "base64" for value in parameters) - ): - raise ValueError("ACP image prompts require base64 data:image URLs") - blocks.append(image_block(data, media_type)) - metadata = { - key: value - for key, value in message.items() - if key not in ("role", "content") and value - } - if metadata: - blocks.append(text_block("\n" + json.dumps(metadata, ensure_ascii=False))) - return blocks - - -async def run_client(config: dict) -> None: - client = VerifiersClient() - command = config["command"] - async with spawn_agent_process( - client, - command[0], - *command[1:], - env=os.environ.copy(), - transport_kwargs={"stderr": None}, - ) as (connection, _process): - initialized = await connection.initialize( - protocol_version=PROTOCOL_VERSION, - client_capabilities=ClientCapabilities(), - ) - capabilities = initialized.agent_capabilities - prompt_capabilities = capabilities and capabilities.prompt_capabilities - supports_images = bool(prompt_capabilities and prompt_capabilities.image) - mcp_servers = [ - HttpMcpServer(type="http", name=name, url=url, headers=[]) - for name, url in config["mcp_urls"].items() - ] - session_path = Path(config["session_path"]) if config["session_path"] else None - is_new = session_path is None or not session_path.exists() - if is_new: - session = await connection.new_session( - cwd=os.getcwd(), mcp_servers=mcp_servers - ) - session_id = session.session_id - else: - session_id = session_path.read_text().strip() - session_capabilities = capabilities and capabilities.session_capabilities - if session_capabilities and session_capabilities.resume is not None: - await connection.resume_session( - cwd=os.getcwd(), session_id=session_id, mcp_servers=mcp_servers - ) - elif capabilities and capabilities.load_session: - await connection.load_session( - cwd=os.getcwd(), session_id=session_id, mcp_servers=mcp_servers - ) - else: - raise RuntimeError("ACP agent does not support resuming sessions") - - messages = config["messages"] - if not is_new: - last_assistant = max( - ( - index - for index, message in enumerate(messages) - if message.get("role") == "assistant" - ), - default=-1, - ) - messages = messages[last_assistant + 1 :] - if is_new and config["system_prompt"]: - messages = [ - {"role": "system", "content": config["system_prompt"]}, - *messages, - ] - prompt = content_blocks(messages, supports_images) - if not prompt: - raise ValueError("ACP prompt has no content") - client.visible_reply = "" - client.message_id = None - client.tool_calls = {} - try: - response = await connection.prompt(session_id=session_id, prompt=prompt) - except RequestError as error: - detail = error.data.get("details") if isinstance(error.data, dict) else None - raise RuntimeError(detail or str(error)) from error - tool_statuses = list(client.tool_calls.values()) - completed_tool_turn = ( - config["allow_empty_tool_reply"] - and response.stop_reason == "end_turn" - and bool(tool_statuses) - and all(status in ("completed", "failed") for status in tool_statuses) - ) - if not client.visible_reply.strip() and not completed_tool_turn: - raise RuntimeError( - "ACP agent produced no visible reply " - f"(stop_reason={response.stop_reason}, tool_statuses={tool_statuses})" - ) - sys.stdout.write(client.visible_reply) - if session_path and is_new: - session_path.parent.mkdir(parents=True, exist_ok=True) - session_path.write_text(session_id) - - -async def main() -> None: - path = Path(sys.argv[1]) - config = json.loads(path.read_text()) - path.unlink() - await run_client(config) - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/verifiers/v1/acp/runner.py b/verifiers/v1/acp/runner.py new file mode 100644 index 000000000..78d8a9a2c --- /dev/null +++ b/verifiers/v1/acp/runner.py @@ -0,0 +1,411 @@ +# /// script +# requires-python = ">=3.10,<3.15" +# dependencies = ["agent-client-protocol==0.11.0"] +# /// +"""Run harness segments through an ACP agent.""" + +import asyncio +import json +import os +import signal +import sys +import traceback +from contextlib import AsyncExitStack, suppress +from pathlib import Path +from typing import Any + +from acp import ( + PROTOCOL_VERSION, + Client, + RequestError, + image_block, + spawn_agent_process, + text_block, +) +from acp.schema import ( + AgentMessageChunk, + AllowedOutcome, + ClientCapabilities, + DeniedOutcome, + HttpMcpServer, + PermissionOption, + RequestPermissionResponse, + TextContentBlock, + ToolCall, + ToolCallUpdate, +) + +MAX_PACKET_BYTES = 128 * 1024 * 1024 + + +class VerifiersACPClient(Client): + def __init__(self) -> None: + self.visible_reply = "" + self.message_id: str | None = None + self.tool_calls: dict[str, str] = {} + + def reset(self) -> None: + self.visible_reply = "" + self.message_id = None + self.tool_calls = {} + + async def session_update(self, session_id: str, update: Any, **kwargs: Any) -> None: + if isinstance(update, ToolCall): + self.tool_calls[update.tool_call_id] = update.status or "pending" + return + if isinstance(update, ToolCallUpdate): + if update.status: + self.tool_calls[update.tool_call_id] = update.status + return + if not isinstance(update, AgentMessageChunk) or not isinstance( + update.content, TextContentBlock + ): + return + message_id = getattr(update, "message_id", None) + if message_id is not None and message_id != self.message_id: + self.visible_reply = "" + self.message_id = message_id + self.visible_reply += update.content.text + + async def request_permission( + self, + session_id: str, + tool_call: Any, + options: list[PermissionOption], + **kwargs: Any, + ) -> RequestPermissionResponse: + option = next( + (item for item in options if item.kind in ("allow_once", "allow_always")), + None, + ) + outcome = ( + AllowedOutcome(outcome="selected", option_id=option.option_id) + if option + else DeniedOutcome(outcome="cancelled") + ) + return RequestPermissionResponse(outcome=outcome) + + +def content_blocks(messages: list[dict], supports_images: bool) -> list: + blocks = [] + transcript = len(messages) != 1 or messages[0].get("role") != "user" + for message in messages: + if transcript: + separator = "\n\n" if blocks else "" + blocks.append( + text_block(f"{separator}[{message.get('role', 'message')}]\n") + ) + content = message.get("content") or "" + parts = ( + [{"type": "text", "text": content}] if isinstance(content, str) else content + ) + for part in parts: + if part["type"] == "text": + blocks.append(text_block(part["text"])) + continue + if not supports_images: + raise ValueError("ACP agent does not support image prompts") + url = part["image_url"]["url"] + metadata, separator, data = url.partition(",") + media_type, *parameters = metadata.removeprefix("data:").split(";") + if ( + not separator + or not metadata.startswith("data:image/") + or not any(value.lower() == "base64" for value in parameters) + ): + raise ValueError("ACP image prompts require base64 data:image URLs") + blocks.append(image_block(data, media_type)) + metadata = { + key: value + for key, value in message.items() + if key not in ("role", "content") and value + } + if metadata: + blocks.append(text_block("\n" + json.dumps(metadata, ensure_ascii=False))) + return blocks + + +def mcp_servers(config: dict) -> list[HttpMcpServer]: + return [ + HttpMcpServer(type="http", name=name, url=url, headers=[]) + for name, url in config["mcp_urls"].items() + ] + + +def segment_messages(config: dict, is_new: bool) -> list[dict]: + messages = config["messages"] + if not is_new: + last_assistant = max( + ( + index + for index, message in enumerate(messages) + if message.get("role") == "assistant" + ), + default=-1, + ) + messages = messages[last_assistant + 1 :] + if is_new and config["system_prompt"]: + messages = [ + {"role": "system", "content": config["system_prompt"]}, + *messages, + ] + return messages + + +async def prompt( + client: VerifiersACPClient, + connection: Any, + capabilities: Any, + session_id: str, + config: dict, + *, + is_new: bool, +) -> str: + prompt_capabilities = capabilities and capabilities.prompt_capabilities + supports_images = bool(prompt_capabilities and prompt_capabilities.image) + blocks = content_blocks(segment_messages(config, is_new), supports_images) + if not blocks: + raise ValueError("ACP prompt has no content") + client.reset() + try: + response = await connection.prompt(session_id=session_id, prompt=blocks) + except RequestError as error: + detail = error.data.get("details") if isinstance(error.data, dict) else None + raise RuntimeError(detail or str(error)) from error + tool_statuses = list(client.tool_calls.values()) + completed_tool_turn = ( + config.get("allow_empty_tool_reply", False) + and response.stop_reason == "end_turn" + and bool(tool_statuses) + and all(status in ("completed", "failed") for status in tool_statuses) + ) + if not client.visible_reply.strip() and not completed_tool_turn: + raise RuntimeError( + "ACP agent produced no visible reply " + f"(stop_reason={response.stop_reason}, tool_statuses={tool_statuses})" + ) + return client.visible_reply + + +async def run_once(config: dict) -> str: + client = VerifiersACPClient() + command = config["command"] + async with spawn_agent_process( + client, + command[0], + *command[1:], + env=os.environ.copy(), + transport_kwargs={"stderr": None}, + ) as (connection, _process): + initialized = await connection.initialize( + protocol_version=PROTOCOL_VERSION, + client_capabilities=ClientCapabilities(), + ) + capabilities = initialized.agent_capabilities + session_path = Path(config["session_path"]) if config["session_path"] else None + is_new = session_path is None or not session_path.exists() + servers = mcp_servers(config) + if is_new: + session = await connection.new_session(cwd=os.getcwd(), mcp_servers=servers) + session_id = session.session_id + else: + session_id = session_path.read_text().strip() + session_capabilities = capabilities and capabilities.session_capabilities + if session_capabilities and session_capabilities.resume is not None: + await connection.resume_session( + cwd=os.getcwd(), session_id=session_id, mcp_servers=servers + ) + elif capabilities and capabilities.load_session: + await connection.load_session( + cwd=os.getcwd(), session_id=session_id, mcp_servers=servers + ) + else: + raise RuntimeError("ACP agent does not support resuming sessions") + + reply = await prompt( + client, + connection, + capabilities, + session_id, + config, + is_new=is_new, + ) + if session_path and is_new: + session_path.parent.mkdir(parents=True, exist_ok=True) + session_path.write_text(session_id) + return reply + + +class LiveACPSession: + """One live ACP process, connection, and session shared by several turns.""" + + def __init__(self) -> None: + self.client = VerifiersACPClient() + self._reset() + + def _reset(self) -> None: + self.stack = AsyncExitStack() + self.connection: Any = None + self.capabilities: Any = None + self.session_id: str | None = None + self.command: list[str] | None = None + self.server_urls: dict[str, str] | None = None + self.system_prompt: str | None = None + self.is_new = True + + async def start(self, config: dict) -> None: + command = config["command"] + try: + self.connection, _process = await self.stack.enter_async_context( + spawn_agent_process( + self.client, + command[0], + *command[1:], + env=os.environ.copy(), + transport_kwargs={"stderr": None}, + ) + ) + initialized = await self.connection.initialize( + protocol_version=PROTOCOL_VERSION, + client_capabilities=ClientCapabilities(), + ) + self.capabilities = initialized.agent_capabilities + session = await self.connection.new_session( + cwd=os.getcwd(), mcp_servers=mcp_servers(config) + ) + except BaseException: + with suppress(BaseException): + await self.stack.aclose() + self._reset() + raise + self.session_id = session.session_id + self.command = command + self.server_urls = config["mcp_urls"] + self.system_prompt = config["system_prompt"] + self.is_new = True + + async def run(self, config: dict) -> str: + if self.connection is None: + await self.start(config) + elif ( + config["command"] != self.command + or config["mcp_urls"] != self.server_urls + or config["system_prompt"] != self.system_prompt + ): + raise RuntimeError("ACP session configuration changed") + assert self.session_id is not None + reply = await prompt( + self.client, + self.connection, + self.capabilities, + self.session_id, + config, + is_new=self.is_new, + ) + self.is_new = False + return reply + + async def close(self) -> None: + try: + if self.connection is not None and self.session_id is not None: + session_capabilities = ( + self.capabilities and self.capabilities.session_capabilities + ) + if session_capabilities and session_capabilities.close is not None: + with suppress(Exception): + await self.connection.close_session(session_id=self.session_id) + await self.stack.aclose() + finally: + self._reset() + + +async def read_packet(stream: asyncio.StreamReader) -> dict | None: + try: + header = await stream.readexactly(8) + except asyncio.IncompleteReadError as error: + if not error.partial: + return None + raise EOFError("ACP session packet ended early") from error + size = int.from_bytes(header, "big") + if size > MAX_PACKET_BYTES: + raise ValueError(f"ACP session packet is too large: {size} bytes") + try: + return json.loads((await stream.readexactly(size)).decode()) + except asyncio.IncompleteReadError as error: + raise EOFError("ACP session packet ended early") from error + + +def write_packet(stream: Any, value: dict) -> None: + data = json.dumps(value, ensure_ascii=False).encode() + if len(data) > MAX_PACKET_BYTES: + raise ValueError(f"ACP session packet is too large: {len(data)} bytes") + stream.write(len(data).to_bytes(8, "big")) + stream.write(data) + stream.flush() + + +async def serve_stream() -> None: + session = LiveACPSession() + closed = False + reader = asyncio.StreamReader() + protocol = asyncio.StreamReaderProtocol(reader) + await asyncio.get_running_loop().connect_read_pipe( + lambda: protocol, sys.stdin.buffer + ) + try: + while request := await read_packet(reader): + stop = False + try: + operation = request.get("operation") + if operation == "prompt": + response = { + "ok": True, + "reply": await session.run(request["config"]), + } + elif operation == "shutdown": + await session.close() + closed = True + stop = True + response = {"ok": True} + else: + raise ValueError(f"unknown ACP session operation: {operation!r}") + except Exception as error: # noqa: BLE001 - serialize protocol failures + traceback.print_exc() + response = { + "ok": False, + "error": f"{type(error).__name__}: {error}", + } + write_packet(sys.stdout.buffer, response) + if stop: + break + finally: + if not closed: + await session.close() + + +def read_config(path_value: str) -> dict: + path = Path(path_value) + config = json.loads(path.read_text()) + path.unlink() + return config + + +async def main() -> None: + operation = sys.argv[1] + if operation == "once": + sys.stdout.write(await run_once(read_config(sys.argv[2]))) + elif operation == "stream": + task = asyncio.current_task() + loop = asyncio.get_running_loop() + if task is not None: + for sig in (signal.SIGTERM, signal.SIGINT): + with suppress(NotImplementedError): + loop.add_signal_handler(sig, task.cancel) + with suppress(asyncio.CancelledError): + await serve_stream() + else: + raise ValueError(f"unknown ACP runner operation: {operation!r}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/verifiers/v1/harness.py b/verifiers/v1/harness.py index fcd54a438..f1e7fd2f4 100644 --- a/verifiers/v1/harness.py +++ b/verifiers/v1/harness.py @@ -127,10 +127,9 @@ async def run( data: TaskData, messages: Messages | None = None, ) -> None: - """Run ONE segment of the exchange: the program from launch (or, with - `messages`, the user's next turn(s) via `resume`) until it yields — a segment - ends when the program exits. The rollout loop owns the exchange across - segments (and stamps its end); a harness only ever sees one segment.""" + """Run ONE segment of the exchange without retaining a process: the program + from launch (or, with `messages`, the user's next turn(s) via `resume`) until + it yields. The rollout loop owns the exchange across segments.""" async with boundary(HarnessError, f"harness {self.config.id!r}"): if messages is None: result = await self.launch( @@ -140,19 +139,45 @@ async def run( result = await self.resume( ctx, trace, runtime, endpoint, secret, mcp_urls, data, messages ) + await self._check_result(trace, runtime, result) + + async def _check_result( + self, trace: Trace, runtime: Runtime, result: ProgramResult + ) -> None: if trace.stop_condition is not None: - return # a @stop refused a turn mid-rollout; the harness's exit is expected - if result.exit_code != 0: - # The real cause is at the END of a traceback, so keep the tail. - detail = (result.stderr or result.stdout).strip()[-2000:] or "" - if not await runtime.alive(): - raise SandboxError( - f"runtime died under harness {self.config.id!r} " - f"(exit {result.exit_code}): {detail}" - ) - raise HarnessError( - f"harness {self.config.id!r} exited {result.exit_code}: {detail}" + return # a @stop refused a turn mid-rollout; the exit is expected + if result.exit_code == 0: + return + # The real cause is at the END of a traceback, so keep the tail. + detail = (result.stderr or result.stdout).strip()[-2000:] or "" + if not await runtime.alive(): + raise SandboxError( + f"runtime died under harness {self.config.id!r} " + f"(exit {result.exit_code}): {detail}" ) + raise HarnessError( + f"harness {self.config.id!r} exited {result.exit_code}: {detail}" + ) + + async def session( + self, + ctx: ModelContext, + trace: Trace, + runtime: Runtime, + endpoint: str, + secret: str, + mcp_urls: dict[str, str], + data: TaskData, + ) -> HarnessSession: + """Create the rollout-scoped handle that drives this harness. + + The default adapts the existing launch/resume contract. Stateful harness + transports override this factory so one handle can own their live process, + connection, or native session for the rollout's full interaction. + """ + return HarnessSession( + self, ctx, trace, runtime, endpoint, secret, mcp_urls, data + ) async def score(self, trace: Trace, runtime: Runtime) -> None: """Run this harness's `@metric` methods over the finished trace, recording @@ -240,3 +265,69 @@ async def launch( loop in-process instead of launching a program, as long as every model call goes through `endpoint` + `secret` — it then returns a synthetic success `ProgramResult`, and the trace is the record of what ran.""" + + +class HarnessSession: + """One rollout's stateful handle onto one harness execution. + + The base adapter preserves the segment-oriented harness interface by invoking + `launch()` once and `resume()` for later caller turns. Specialized handles can + override `_run()` and `close()` to retain transport state across those turns. + """ + + def __init__( + self, + harness: Harness, + ctx: ModelContext, + trace: Trace, + runtime: Runtime, + endpoint: str, + secret: str, + mcp_urls: dict[str, str], + data: TaskData, + ) -> None: + self.harness = harness + self.ctx = ctx + self.trace = trace + self.runtime = runtime + self.endpoint = endpoint + self.secret = secret + self.mcp_urls = mcp_urls + self.data = data + self._closed = False + + async def turn(self, messages: Messages | None = None) -> None: + """Run one harness segment while retaining session state for the next.""" + if self._closed: + raise HarnessError( + f"harness {self.harness.config.id!r} session is already closed" + ) + async with boundary(HarnessError, f"harness {self.harness.config.id!r}"): + result = await self._run(messages) + await self.harness._check_result(self.trace, self.runtime, result) + + async def _run(self, messages: Messages | None) -> ProgramResult: + if messages is None: + return await self.harness.launch( + self.ctx, + self.trace, + self.runtime, + self.endpoint, + self.secret, + self.mcp_urls, + self.data, + ) + return await self.harness.resume( + self.ctx, + self.trace, + self.runtime, + self.endpoint, + self.secret, + self.mcp_urls, + self.data, + messages, + ) + + async def close(self) -> None: + """Close session-owned resources. Idempotent.""" + self._closed = True diff --git a/verifiers/v1/harnesses/rlm/harness.py b/verifiers/v1/harnesses/rlm/harness.py index a3d70653f..fcdf24441 100644 --- a/verifiers/v1/harnesses/rlm/harness.py +++ b/verifiers/v1/harnesses/rlm/harness.py @@ -1,4 +1,4 @@ -"""RLM exposes `RLM_MCP_CONFIG` tools as pre-imported IPython skills.""" +"""RLM over ACP, with MCP tools exposed as pre-imported IPython skills.""" import json import logging @@ -8,10 +8,10 @@ from pydantic import Field, PositiveInt, model_validator +from verifiers.v1.acp import ACP from verifiers.v1.clients import ModelContext from verifiers.v1.configs.harness import HarnessConfig -from verifiers.v1.dialects.chat import message_to_wire -from verifiers.v1.harness import Harness +from verifiers.v1.harness import Harness, HarnessSession from verifiers.v1.runtimes import ProgramResult, Runtime from verifiers.v1.task import TaskData from verifiers.v1.trace import Trace @@ -21,18 +21,17 @@ BuiltinSkill = Literal["edit", "search"] -RLM_REPO = "github.com/PrimeIntellect-ai/rlm.git" -# rlm writes its session under $RLM_HOME/sessions//; point it at a workdir- -# relative dir so it stays in the runtime (and is cleaned up with the workdir). -RLM_HOME = ".rlm" +RLM_REPO = "github.com/PrimeIntellect-ai/rlm-harness.git" RLM_DIR = "/tmp/vf-rlm" RLM_BIN = f"{RLM_DIR}/bin/rlm" SKILLS_DIR = "/task/rlm-skills" +RLM_STATE_DIR = ".vf-rlm" +RLM_ACP = ACP() class RLMHarnessConfig(HarnessConfig): version: str = "main" - """Git ref (branch, tag, or commit) of rlm to install.""" + """Git ref (branch, tag, or commit) of rlm-harness to install.""" max_depth: int = 0 """Recursion depth rlm may spawn sub-harnesses to (RLM_MAX_DEPTH).""" builtin_skills: list[BuiltinSkill] = Field(default_factory=list) @@ -85,12 +84,13 @@ async def setup(self, runtime: Runtime) -> None: logger.info("rlm: ensuring rlm is installed (version=%s)", self.config.version) ensure = shlex.quote(f"[ -x {RLM_BIN} ] || ({install})") guarded = f"mkdir -p {RLM_DIR} && flock {RLM_DIR}/install.lock sh -c {ensure}" - env = {**self.config.resolved_env, "RLM_HOME": RLM_HOME} + env = self.config.resolved_env.copy() extra_uv_args = env.get("RLM_EXTRA_UV_ARGS", "") env["RLM_EXTRA_UV_ARGS"] = f"{extra_uv_args} --with mcp~=1.28".strip() result = await runtime.run(["sh", "-c", guarded], env) if result.exit_code != 0: raise RuntimeError(f"rlm install failed: {result.stderr.strip()[-500:]}") + await RLM_ACP.setup(self, runtime) def summarize_threshold(self, task_idx: int | None) -> str: """The `RLM_SUMMARIZE_AT_TOKENS` value: a range draws per-group (seeded by task index — @@ -104,47 +104,85 @@ def summarize_threshold(self, task_idx: int | None) -> str: return str(random.Random(task_idx or 0).randint(lo, hi)) return str(value) - async def launch( + def _env( self, ctx: ModelContext, trace: Trace, - runtime: Runtime, endpoint: str, secret: str, - mcp_urls: dict[str, str], data: TaskData, - ) -> ProgramResult: - system_prompt, prompt = self.resolve_prompt(data) - if prompt is None: - raise ValueError("RLM requires a prompt") - if not isinstance(prompt, str): - prompt = json.dumps( - [message_to_wire(message) for message in prompt], ensure_ascii=False - ) + system_prompt: str | None, + ) -> dict[str, str]: env = { **self.config.resolved_env, "RLM_BASE_URL": endpoint, "RLM_API_KEY": secret, "RLM_MODEL": ctx.model, "RLM_MAX_DEPTH": str(self.config.max_depth), - "RLM_HOME": RLM_HOME, + "RLM_HOME": self._home(trace), "RLM_SUMMARIZE_AT_TOKENS": self.summarize_threshold(data.idx), } if system_prompt is not None: env["RLM_APPEND_TO_SYSTEM_PROMPT"] = system_prompt if self.config.builtin_skills: env["RLM_SKILLS"] = ",".join(self.config.builtin_skills) - if mcp_urls: - env["RLM_MCP_CONFIG"] = json.dumps( - {"mcpServers": {name: {"url": url} for name, url in mcp_urls.items()}} + return env + + async def session( + self, + ctx: ModelContext, + trace: Trace, + runtime: Runtime, + endpoint: str, + secret: str, + mcp_urls: dict[str, str], + data: TaskData, + ) -> HarnessSession: + if not runtime.supports_live_processes: + return await super().session( + ctx, trace, runtime, endpoint, secret, mcp_urls, data ) - # RLM has no interactive mode; resumed segments explicitly replay the transcript. - return await runtime.run_program([RLM_BIN, "--", prompt], env) + system_prompt, prompt = self.resolve_prompt(data) + return RLM_ACP.session( + self, + ctx, + trace, + runtime, + endpoint, + secret, + mcp_urls, + data, + env=self._env(ctx, trace, endpoint, secret, data, system_prompt), + command=[RLM_BIN, "--acp"], + prompt=prompt, + ) + + async def launch( + self, + ctx: ModelContext, + trace: Trace, + runtime: Runtime, + endpoint: str, + secret: str, + mcp_urls: dict[str, str], + data: TaskData, + ) -> ProgramResult: + """Run one standalone ACP segment through the default session adapter.""" + system_prompt, prompt = self.resolve_prompt(data) + return await RLM_ACP.run( + runtime, + self._env(ctx, trace, endpoint, secret, data, system_prompt), + [RLM_BIN, "--acp"], + prompt, + mcp_urls=mcp_urls, + ) @metric - async def rlm(self, runtime: Runtime) -> dict[str, float]: - # Stateless continuation creates one session per segment; report the latest. - latest = f'cat "$(ls -t {RLM_HOME}/sessions/*/meta.json | head -1)"' + async def rlm(self, trace: Trace, runtime: Runtime) -> dict[str, float]: + # RolloutRun closes the harness session before metrics, which finalizes + # RLM's meta.json while leaving the harness-owned state available here. + home = shlex.quote(self._home(trace)) + latest = f'cat "$(ls -t {home}/sessions/*/meta.json | head -1)"' result = await runtime.run(["sh", "-c", latest], {}) if result.exit_code != 0 or not result.stdout.strip(): return {} @@ -157,3 +195,10 @@ async def rlm(self, runtime: Runtime) -> dict[str, float]: for key, value in meta.get("metrics", {}).items() if isinstance(value, (int, float)) and not isinstance(value, bool) } + + async def cleanup(self, trace: Trace, runtime: Runtime) -> None: + await runtime.run(["rm", "-rf", f"{RLM_STATE_DIR}/{trace.id}"], {}) + + @staticmethod + def _home(trace: Trace) -> str: + return f"{RLM_STATE_DIR}/{trace.id}/home" diff --git a/verifiers/v1/interception/server.py b/verifiers/v1/interception/server.py index 9898a5fba..0c10bc190 100644 --- a/verifiers/v1/interception/server.py +++ b/verifiers/v1/interception/server.py @@ -397,7 +397,7 @@ def serve(response: Response) -> web.Response: ) if refused is not None: # Refuse the model call to halt the harness (it sees an HTTP error; - # `Harness.run` treats a stopped rollout as the clean exit it is). + # `HarnessSession.turn` treats a stopped rollout as the clean exit it is). return web.json_response( dialect.error_body(f"rollout stopped: {refused}"), status=400, diff --git a/verifiers/v1/mcp/launch.py b/verifiers/v1/mcp/launch.py index 85595cc95..920747bbb 100644 --- a/verifiers/v1/mcp/launch.py +++ b/verifiers/v1/mcp/launch.py @@ -13,7 +13,7 @@ from collections.abc import AsyncIterator from dataclasses import dataclass, field from functools import cache -from pathlib import Path +from pathlib import Path, PurePosixPath from typing import TYPE_CHECKING from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit @@ -112,25 +112,38 @@ async def _install_in_sandbox(server: ServerBase, runtime: Runtime) -> str: f"server {server.server_name!r} runs in a {runtime.type} runtime but its module is not " "a local package (no pyproject) — sandbox launch needs a local env package to upload" ) - root = "/tmp/vf-src" + # Prime VMs mount /tmp as a small tmpfs, while the runtime workdir lives on + # the VM's root disk. Keep source, build scratch space, and uv's cache on the + # durable runtime filesystem so ordinary dependency installs cannot exhaust + # the tmpfs. + workdir = str(PurePosixPath(runtime.config.workdir)) + root = str(PurePosixPath(workdir) / ".vf-src") + temp = str(PurePosixPath(workdir) / ".vf-tmp") + cache = str(PurePosixPath(workdir) / ".vf-uv-cache") vf, env = _verifiers_root(), Path(source_dir) await runtime.write(f"{root}/{vf.name}.tar.gz", _tar_source(vf, VF_BUILD_INPUTS)) await runtime.write(f"{root}/{env.name}.tar.gz", _tar_source(env)) - venv = "/tmp/vf-venv" + venv = str(PurePosixPath(workdir) / ".vf-venv") + root_q, temp_q, cache_q, venv_q = map(shlex.quote, (root, temp, cache, venv)) # The upload carries no .git, so hatch-vcs falls back to version 0.0.0 — an env # package's `verifiers>=...` floor would then resolve PyPI verifiers OVER the local # build, silently running the server against a released (older) API. Pretend the # local version so the floor is satisfied by the build we uploaded. vf_version = importlib.metadata.version("verifiers") extras = ",".join(type(server).EXTRAS) + vf_source = shlex.quote(str(PurePosixPath(root) / vf.name)) + env_source = shlex.quote( + str(PurePosixPath(root) / (env.name + (f"[{extras}]" if extras else ""))) + ) setup = ( - f"{_ENSURE_UV}; set -e; " - f'for t in {root}/*.tar.gz; do tar -xzf "$t" -C {root}; done && ' - f"uv venv {venv} && " + f"set -e; mkdir -p {root_q} {temp_q} {cache_q}; " + f"export TMPDIR={temp_q} UV_CACHE_DIR={cache_q}; " + f"{_ENSURE_UV}; " + f'for t in {root_q}/*.tar.gz; do tar -xzf "$t" -C {root_q}; done && ' + f"uv venv {venv_q} && " f"SETUPTOOLS_SCM_PRETEND_VERSION={shlex.quote(vf_version)} " - f"uv pip install --python {venv} {root}/{shlex.quote(vf.name)} && " - f"uv pip install --python {venv} " - f"{shlex.quote(f'{root}/{env.name}' + (f'[{extras}]' if extras else ''))}" + f"uv pip install --python {venv_q} {vf_source} && " + f"uv pip install --python {venv_q} {env_source}" ) result = await runtime.run(["sh", "-c", setup], {}) if result.exit_code != 0: diff --git a/verifiers/v1/rollout.py b/verifiers/v1/rollout.py index 98050f46c..285d3c6ca 100644 --- a/verifiers/v1/rollout.py +++ b/verifiers/v1/rollout.py @@ -17,7 +17,7 @@ ToolsetError, boundary, ) -from verifiers.v1.harness import Harness +from verifiers.v1.harness import Harness, HarnessSession from verifiers.v1.interception import Interception, serve_interception from verifiers.v1.mcp import SharedToolServer, serve_tools from verifiers.v1.runtimes import ( @@ -102,6 +102,7 @@ def __init__( self._closed = False self._endpoint: str | None = None self._urls: dict[str, str] = {} + self._harness_session: HarnessSession | None = None 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 @@ -232,6 +233,16 @@ async def open(self) -> bool: # Setup and service provisioning are complete. Apply the runtime's # execution policy while preserving the framework routes the agent uses. await runtime.prepare_execution([self._endpoint, *self._urls.values()]) + async with boundary(HarnessError, "opening harness session"): + self._harness_session = await self.harness.session( + self.ctx, + self.trace, + runtime, + self._endpoint, + self._secret, + self._urls, + self.trace.task.data, + ) except Exception as e: # noqa: BLE001 - setup boundary records every rollout failure self.fail(e) return False @@ -268,16 +279,8 @@ async def step(self, messages: Messages | None = None) -> bool: # Prefer an intercepted model/tool error to the harness exit it caused. try: async with asyncio.timeout_at(self.deadline_at): - await self.harness.run( - self.ctx, - trace, - self.runtime, - self._endpoint, - self._secret, - self._urls, - trace.task.data, - messages, - ) + assert self._harness_session is not None + await self._harness_session.turn(messages) except TimeoutError as e: # An expired rollout deadline is the agent breaking its time budget — # an agent failure, never a clean stop. A TimeoutError from the @@ -320,6 +323,9 @@ async def abort(self) -> None: (a cancellation mid-setup, a lifetime bug raised to the caller) means the driver will never reach `close()`. Safe after a partial `close()`.""" self._closed = True + if self._harness_session is not None: + with contextlib.suppress(Exception): + await self._harness_session.close() with contextlib.suppress(Exception): await self._stack.aclose() with contextlib.suppress(Exception): @@ -342,6 +348,17 @@ async def close(self) -> Trace: trace = self.trace runtime = self.runtime try: + if self._harness_session is not None: + try: + await self._harness_session.close() + except Exception: + # Generation already completed. A transport teardown failure + # must not discard its otherwise scoreable trajectory. + logger.warning( + "harness session close failed (rollout %s)", + trace.id, + exc_info=True, + ) try: await self._stack.aclose() finally: @@ -372,6 +389,11 @@ async def close(self) -> Trace: except Exception as e: # noqa: BLE001 - finalize boundary records every rollout failure self.fail(e) finally: + if self._harness_session is not None: + with contextlib.suppress(Exception): + await self._harness_session.close() + with contextlib.suppress(Exception): + await self._stack.aclose() trace.is_completed = True trace.ok = not self._failed now = time.time() diff --git a/verifiers/v1/runtimes/__init__.py b/verifiers/v1/runtimes/__init__.py index 7daa9c5c4..3a922933d 100644 --- a/verifiers/v1/runtimes/__init__.py +++ b/verifiers/v1/runtimes/__init__.py @@ -9,6 +9,7 @@ NetworkPolicyConfig, ProgramResult, Runtime, + RuntimeProcess, register, ) from verifiers.v1.runtimes.docker import DockerConfig, DockerRuntime, DockerRuntimeInfo @@ -85,6 +86,7 @@ def runtime_is_local(config: RuntimeConfig) -> bool: "Runtime", "RuntimeConfig", "RuntimeInfo", + "RuntimeProcess", "SubprocessConfig", "SubprocessRuntime", "SubprocessRuntimeInfo", diff --git a/verifiers/v1/runtimes/base.py b/verifiers/v1/runtimes/base.py index 7952b570d..883394b52 100644 --- a/verifiers/v1/runtimes/base.py +++ b/verifiers/v1/runtimes/base.py @@ -10,6 +10,7 @@ import uuid import weakref from abc import ABC, abstractmethod +from collections.abc import AsyncIterator from dataclasses import dataclass from pathlib import PurePosixPath from typing import ClassVar, Self @@ -54,6 +55,29 @@ class ProgramResult: stderr: str +class RuntimeProcess(ABC): + """A live process whose stdio crosses a runtime boundary.""" + + stdout: AsyncIterator[bytes] + stderr: AsyncIterator[bytes] + + @abstractmethod + async def write(self, data: bytes) -> None: + pass + + @abstractmethod + async def wait(self) -> int: + pass + + @abstractmethod + async def terminate(self) -> None: + pass + + @abstractmethod + async def kill(self) -> None: + pass + + def parse_gpu(gpu: str | None) -> tuple[str | None, int]: """A Modal-style GPU spec -> (type, count) for providers that want them split: "A100" -> ("A100", 1), "A100:2" -> ("A100", 2), "2" -> (None, 2) (count only, @@ -154,6 +178,11 @@ class Runtime(ABC): info: BaseRuntimeInfo + @property + def supports_live_processes(self) -> bool: + """Whether `open_process()` is implemented for this runtime instance.""" + return type(self).open_process is not Runtime.open_process + def __init__(self, name: str | None = None) -> None: self.name = name or f"vf-{uuid.uuid4().hex[:12]}" self._uv_interpreters: dict[str, str] = {} @@ -216,6 +245,14 @@ async def run_program(self, argv: list[str], env: dict[str, str]) -> ProgramResu still retry individual safe transport operations underneath `run`.""" return await self.run(argv, env) + async def open_process( + self, argv: list[str], env: dict[str, str] + ) -> RuntimeProcess: + """Start a live process for a rollout-scoped harness session.""" + raise NotImplementedError( + f"{type(self).__name__} does not support live processes" + ) + async def run_background( self, argv: list[str], env: dict[str, str], log: str ) -> None: diff --git a/verifiers/v1/runtimes/docker/__init__.py b/verifiers/v1/runtimes/docker/__init__.py index 9e714ddfe..c38691828 100644 --- a/verifiers/v1/runtimes/docker/__init__.py +++ b/verifiers/v1/runtimes/docker/__init__.py @@ -9,6 +9,8 @@ import subprocess import sys import tempfile +import uuid +from collections.abc import AsyncIterator from pathlib import PurePosixPath from typing import Literal from urllib.parse import urlsplit @@ -19,9 +21,11 @@ NetworkPolicyConfig, ProgramResult, Runtime, + RuntimeProcess, parse_gpu, ) from verifiers.v1.runtimes.docker.egress import HOST_ALIAS, EgressProxy, NetworkPolicy +from verifiers.v1.utils.aio import run_shielded logger = logging.getLogger(__name__) @@ -47,6 +51,60 @@ class DockerRuntimeInfo(DockerConfig, BaseRuntimeInfo): pass +async def _read_stream(reader: asyncio.StreamReader) -> AsyncIterator[bytes]: + while chunk := await reader.read(64 * 1024): + yield chunk + + +class DockerProcess(RuntimeProcess): + def __init__( + self, + process: asyncio.subprocess.Process, + container: str, + pid: int, + ) -> None: + self._process = process + self._container = container + self._pid = pid + assert process.stdin is not None + assert process.stdout is not None + assert process.stderr is not None + self._stdin = process.stdin + self.stdout = _read_stream(process.stdout) + self.stderr = _read_stream(process.stderr) + + async def write(self, data: bytes) -> None: + self._stdin.write(data) + await self._stdin.drain() + + async def wait(self) -> int: + return await self._process.wait() + + async def terminate(self) -> None: + await self._signal("TERM") + + async def kill(self) -> None: + await self._signal("KILL") + + async def _signal(self, signal: str) -> None: + if self._process.returncode is not None: + return + result = await docker( + "exec", + self._container, + "sh", + "-c", + 'kill -"$1" "-$2" 2>/dev/null || kill -"$1" "$2"', + "vf-signal", + signal, + str(self._pid), + ) + if result.exit_code != 0 and self._process.returncode is None: + raise SandboxError( + f"docker exec process signal failed: {result.stderr.strip()}" + ) + + async def docker(*args: str) -> ProgramResult: proc = await asyncio.create_subprocess_exec( "docker", @@ -54,7 +112,14 @@ async def docker(*args: str) -> ProgramResult: stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) - stdout, stderr = await proc.communicate() + try: + stdout, stderr = await proc.communicate() + except BaseException: + if proc.returncode is None: + with contextlib.suppress(ProcessLookupError): + proc.kill() + await run_shielded(proc.communicate()) + raise return ProgramResult( exit_code=proc.returncode or 0, stdout=stdout.decode(errors="replace"), @@ -62,6 +127,41 @@ async def docker(*args: str) -> ProgramResult: ) +async def _abort_process_startup( + proc: asyncio.subprocess.Process, container: str, pidfile: str +) -> str: + """Kill a partially opened container process and reap its local docker client.""" + # The target normally writes its PID immediately, but cancellation can win + # that race. Wait briefly for the file before signalling the process group. + cleanup = ( + 'i=0; while [ "$i" -lt 20 ]; do ' + 'if [ -s "$1" ]; then pid=$(cat "$1"); ' + 'kill -KILL "-$pid" 2>/dev/null || kill -KILL "$pid" 2>/dev/null || true; ' + 'rm -f "$1"; exit 0; fi; ' + "i=$((i + 1)); sleep 0.05; done; exit 1" + ) + try: + with contextlib.suppress(Exception): + await asyncio.wait_for( + docker( + "exec", + container, + "sh", + "-c", + cleanup, + "vf-process-cleanup", + pidfile, + ), + timeout=2, + ) + finally: + if proc.returncode is None: + with contextlib.suppress(ProcessLookupError): + proc.kill() + _, stderr = await proc.communicate() + return stderr.decode(errors="replace").strip() + + _PROXY_HOST = "host.docker.internal" _PASS_LISTENER = r""" import array, socket @@ -313,6 +413,67 @@ async def run(self, argv: list[str], env: dict[str, str]) -> ProgramResult: "exec", *env_args, "--workdir", self.config.workdir, self._container, *argv ) + async def open_process( + self, argv: list[str], env: dict[str, str] + ) -> RuntimeProcess: + assert self._container is not None + env = {**env, **(self._proxy_env() if self._cut else {})} + env_args = [ + arg for key, value in env.items() for arg in ("--env", f"{key}={value}") + ] + pidfile = f"/tmp/vf-process-{uuid.uuid4().hex}.pid" + # Give the target its own process group when `setsid -w` is available so + # terminate()/kill() reap its descendants while docker exec remains + # attached if setsid needs to fork. The inner shell records the + # post-setsid PID before exec preserves it as the target PID. + wrapper = ( + "if setsid -w true >/dev/null 2>&1; then " + 'exec setsid -w sh -c \'echo $$ > "$1"; shift; exec "$@"\' ' + 'vf-process "$@"; ' + 'fi; echo $$ > "$1"; shift; exec "$@"' + ) + proc = await asyncio.create_subprocess_exec( + "docker", + "exec", + "-i", + *env_args, + "--workdir", + self.config.workdir, + self._container, + "sh", + "-c", + wrapper, + "vf-process", + pidfile, + *argv, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + loop = asyncio.get_running_loop() + deadline = loop.time() + 5 + try: + while True: + ready = await docker("exec", self._container, "cat", pidfile) + if ready.exit_code == 0 and ready.stdout.strip().isdigit(): + return DockerProcess( + proc, self._container, int(ready.stdout.strip()) + ) + if proc.returncode is not None or loop.time() >= deadline: + break + await asyncio.sleep(0.05) + except BaseException: + await run_shielded(_abort_process_startup(proc, self._container, pidfile)) + raise + + stderr = await run_shielded( + _abort_process_startup(proc, self._container, pidfile) + ) + detail = stderr or ready.stderr.strip() + raise SandboxError( + f"docker live process failed to start: {detail or 'PID unavailable'}" + ) + async def run_background( self, argv: list[str], env: dict[str, str], log: str ) -> None: diff --git a/verifiers/v1/runtimes/modal.py b/verifiers/v1/runtimes/modal.py index 41baea523..4676bc14f 100644 --- a/verifiers/v1/runtimes/modal.py +++ b/verifiers/v1/runtimes/modal.py @@ -11,6 +11,8 @@ import contextlib import logging import shlex +import uuid +from collections.abc import AsyncIterator from pathlib import PurePosixPath from typing import ClassVar, Literal @@ -22,6 +24,7 @@ BaseRuntimeInfo, ProgramResult, Runtime, + RuntimeProcess, ) from verifiers.v1.runtimes.limiters import creation_limiter @@ -58,6 +61,46 @@ class ModalRuntimeInfo(ModalConfig, BaseRuntimeInfo): pass +class ModalProcess(RuntimeProcess): + def __init__(self, process, sandbox, pid: int, workdir: str) -> None: + self._process = process + self._sandbox = sandbox + self._pid = pid + self._workdir = workdir + self.stdout: AsyncIterator[bytes] = process.stdout + self.stderr: AsyncIterator[bytes] = process.stderr + + async def write(self, data: bytes) -> None: + await self._process.stdin.write.aio(data) + await self._process.stdin.drain.aio() + + async def wait(self) -> int: + return await self._process.wait.aio() + + async def terminate(self) -> None: + await self._signal("TERM") + + async def kill(self) -> None: + await self._signal("KILL") + + async def _signal(self, signal: str) -> None: + if await self._process.poll.aio() is not None: + return + proc = await self._sandbox.exec.aio( + "sh", + "-c", + 'kill -"$1" "$2"', + "vf-signal", + signal, + str(self._pid), + workdir=self._workdir, + ) + stderr = await proc.stderr.read.aio() + exit_code = await proc.wait.aio() + if exit_code != 0 and await self._process.poll.aio() is None: + raise SandboxError(f"modal exec process signal failed: {stderr.strip()}") + + class ModalRuntime(Runtime): is_local: ClassVar[bool] = False @@ -146,6 +189,62 @@ async def run(self, argv: list[str], env: dict[str, str]) -> ProgramResult: stderr=stderr or "", ) + async def open_process( + self, argv: list[str], env: dict[str, str] + ) -> RuntimeProcess: + pidfile = f".vf-process-{uuid.uuid4().hex}.pid" + wrapper = 'echo $$ > "$1"; shift; exec "$@"' + try: + proc = await self._sandbox.exec.aio( + "sh", + "-c", + wrapper, + "vf-process", + pidfile, + *argv, + workdir=self.config.workdir, + env=env, + text=False, + ) + loop = asyncio.get_running_loop() + deadline = loop.time() + 5 + while True: + ready = await self.run(["cat", pidfile], {}) + if ready.exit_code == 0 and ready.stdout.strip().isdigit(): + return ModalProcess( + proc, + self._sandbox, + int(ready.stdout.strip()), + self.config.workdir, + ) + returncode = await proc.poll.aio() + if returncode is not None or loop.time() >= deadline: + if returncode is None: + # Modal's ContainerProcess has no signal API. A live + # process without a readable PID cannot be targeted, so + # fail the sandbox closed instead of abandoning it. + sandbox = self._sandbox + try: + await sandbox.terminate.aio() + except Exception: + logger.warning( + "modal: failed to terminate sandbox %s after live " + "process startup timed out", + self.info.id, + exc_info=True, + ) + else: + self._sandbox = None + raise SandboxError( + "modal live process failed to start: " + f"{ready.stderr.strip() or 'PID unavailable'}" + ) + await asyncio.sleep(0.05) + except SandboxError: + raise + except Exception as e: + raise SandboxError(f"modal live process failed to start: {e}") from e + async def run_background( self, argv: list[str], env: dict[str, str], log: str ) -> None: diff --git a/verifiers/v1/runtimes/prime.py b/verifiers/v1/runtimes/prime.py index f0ae041df..409e842c7 100644 --- a/verifiers/v1/runtimes/prime.py +++ b/verifiers/v1/runtimes/prime.py @@ -12,6 +12,7 @@ import math import shlex import tempfile +from collections.abc import AsyncIterator from pathlib import Path, PurePosixPath from typing import ClassVar, Literal from urllib.parse import urlsplit @@ -26,6 +27,7 @@ NetworkPolicyConfig, ProgramResult, Runtime, + RuntimeProcess, parse_gpu, ) from verifiers.v1.runtimes.limiters import creation_limiter @@ -100,6 +102,25 @@ class PrimeRuntimeInfo(PrimeConfig, BaseRuntimeInfo): a first-use auto-build ran while this sandbox waited to start.""" +class PrimeProcess(RuntimeProcess): + def __init__(self, process) -> None: + self._process = process + self.stdout: AsyncIterator[bytes] = process.stdout + self.stderr: AsyncIterator[bytes] = process.stderr + + async def write(self, data: bytes) -> None: + await self._process.write_stdin(data) + + async def wait(self) -> int: + return await self._process.wait() + + async def terminate(self) -> None: + await self._process.terminate() + + async def kill(self) -> None: + await self._process.kill() + + class PrimeRuntime(Runtime): is_local: ClassVar[bool] = False @@ -109,6 +130,10 @@ def __init__(self, config: PrimeConfig, name: str | None = None) -> None: self.info = PrimeRuntimeInfo(**config.model_dump()) self._client = None + @property + def supports_live_processes(self) -> bool: + return self.config.vm + @property def published_port(self) -> int | None: return SERVICE_PORT @@ -241,6 +266,25 @@ async def run(self, argv: list[str], env: dict[str, str]) -> ProgramResult: stderr=result.stderr or "", ) + async def open_process( + self, argv: list[str], env: dict[str, str] + ) -> RuntimeProcess: + if not self.config.vm: + raise SandboxError( + "persistent harness sessions on Prime require a VM sandbox; " + "set runtime.prime.vm=true" + ) + try: + process = await self._client.open_process( + self.info.id, + shlex.join(argv), + working_dir=self.config.workdir, + env=env, + ) + except Exception as e: + raise SandboxError(f"prime live process failed to start: {e}") from e + return PrimeProcess(process) + async def expose(self, port: int) -> str | None: # Publish a server hosted IN the sandbox via the SDK's native port exposure → a public # HTTPS URL. Removed when the sandbox is deleted in stop(), so a tool in its own prime diff --git a/verifiers/v1/runtimes/subprocess.py b/verifiers/v1/runtimes/subprocess.py index a87e3ef71..b1237ac25 100644 --- a/verifiers/v1/runtimes/subprocess.py +++ b/verifiers/v1/runtimes/subprocess.py @@ -5,12 +5,18 @@ import os import shutil import signal +from collections.abc import AsyncIterator from pathlib import Path from typing import ClassVar, Literal from pydantic_config import BaseConfig -from verifiers.v1.runtimes.base import BaseRuntimeInfo, ProgramResult, Runtime +from verifiers.v1.runtimes.base import ( + BaseRuntimeInfo, + ProgramResult, + Runtime, + RuntimeProcess, +) _BACKGROUND_STOP_TIMEOUT = 5 @@ -28,6 +34,46 @@ class SubprocessRuntimeInfo(SubprocessConfig, BaseRuntimeInfo): pass +async def read_stream(reader: asyncio.StreamReader) -> AsyncIterator[bytes]: + while chunk := await reader.read(64 * 1024): + yield chunk + + +def signal_process( + process: asyncio.subprocess.Process, signal_: signal.Signals +) -> None: + if process.returncode is not None: + return + # open_process() creates a new session, so pgid == pid and signalling the + # group reaps the complete process tree. + with contextlib.suppress(ProcessLookupError, PermissionError): + os.killpg(os.getpgid(process.pid), signal_) + + +class SubprocessProcess(RuntimeProcess): + def __init__(self, process: asyncio.subprocess.Process) -> None: + self._process = process + assert process.stdin is not None + assert process.stdout is not None + assert process.stderr is not None + self._stdin = process.stdin + self.stdout = read_stream(process.stdout) + self.stderr = read_stream(process.stderr) + + async def write(self, data: bytes) -> None: + self._stdin.write(data) + await self._stdin.drain() + + async def wait(self) -> int: + return await self._process.wait() + + async def terminate(self) -> None: + signal_process(self._process, signal.SIGTERM) + + async def kill(self) -> None: + signal_process(self._process, signal.SIGKILL) + + class SubprocessRuntime(Runtime): # Share prepared script environments across the worker's per-rollout runtimes. _interpreters: ClassVar[dict[str, str]] = {} @@ -75,6 +121,23 @@ async def run(self, argv: list[str], env: dict[str, str]) -> ProgramResult: stderr=stderr.decode(errors="replace"), ) + async def open_process( + self, argv: list[str], env: dict[str, str] + ) -> RuntimeProcess: + full_env = {k: v for k, v in os.environ.items() if "API_KEY" not in k.upper()} + full_env.update(env) + proc = await asyncio.create_subprocess_exec( + *argv, + env=full_env, + cwd=self.workdir, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + start_new_session=True, + ) + self._background.append(proc) + return SubprocessProcess(proc) + async def run_background( self, argv: list[str], env: dict[str, str], log: str ) -> None: @@ -104,20 +167,11 @@ async def write(self, path: str, data: bytes) -> None: target.parent.mkdir(parents=True, exist_ok=True) await asyncio.to_thread(target.write_bytes, data) - @staticmethod - def _signal(proc: asyncio.subprocess.Process, sig: signal.Signals) -> None: - if proc.returncode is not None: - return - # Signal the whole group (start_new_session => pgid == pid), not just proc.pid, - # so a background server's children (sh -> uv -> python) stop with it. - with contextlib.suppress(ProcessLookupError, PermissionError): - os.killpg(os.getpgid(proc.pid), sig) - async def teardown(self) -> None: """Stop and reap background servers before their event loop closes.""" background = list(self._background) for proc in background: - self._signal(proc, signal.SIGTERM) + signal_process(proc, signal.SIGTERM) if background: try: await asyncio.wait_for( @@ -128,7 +182,7 @@ async def teardown(self) -> None: ) except TimeoutError: for proc in background: - self._signal(proc, signal.SIGKILL) + signal_process(proc, signal.SIGKILL) with contextlib.suppress(TimeoutError): await asyncio.wait_for( asyncio.gather( @@ -143,7 +197,7 @@ async def teardown(self) -> None: def cleanup(self) -> None: for proc in self._background: - self._signal(proc, signal.SIGTERM) + signal_process(proc, signal.SIGTERM) self._background = [] if self.workdir is not None: shutil.rmtree(self.workdir, ignore_errors=True) diff --git a/verifiers/v1/session.py b/verifiers/v1/session.py index 8e8611208..322804c72 100644 --- a/verifiers/v1/session.py +++ b/verifiers/v1/session.py @@ -126,7 +126,7 @@ def release(self) -> None: async def refused(self) -> str | None: """The framework's limits (turns / token budget) and `@stop` checks, run before each model call. Sets the stop condition and returns its name, else None. A refused first - call halts the harness (its model call errors out); Harness.run treats it as clean. A task + call halts the harness (its model call errors out); HarnessSession.turn treats it as clean. A task that ends a trajectory from `trace.state` does it with its own `@stop` (run here generically), so the interception server holds no opinion about the state's contents.""" if (limit := self.limits.reached(self.trace)) is not None: