From 3f40fbb98da6605419223bf3cefbdbaddbafef93 Mon Sep 17 00:00:00 2001 From: hallerite Date: Tue, 4 Aug 2026 21:39:18 +0200 Subject: [PATCH 01/17] feat(v1): add rollout-scoped harness sessions # Conflicts: # skills/evaluate-environments/references/REFERENCE.md # tests/v1/test_e2e.py # verifiers/v1/acp/_runner.py # verifiers/v1/harness.py # verifiers/v1/harnesses/rlm/harness.py # verifiers/v1/rollout.py --- tests/v1/test_e2e.py | 3 + verifiers/v1/acp/__init__.py | 241 ++++++++++++++- verifiers/v1/acp/_runner.py | 412 ++++++++++++++++++++++---- verifiers/v1/harness.py | 134 +++++++-- verifiers/v1/harnesses/rlm/harness.py | 108 +++++-- verifiers/v1/rollout.py | 44 ++- 6 files changed, 815 insertions(+), 127 deletions(-) diff --git a/tests/v1/test_e2e.py b/tests/v1/test_e2e.py index 4f6a5f194..691f52fc1 100644 --- a/tests/v1/test_e2e.py +++ b/tests/v1/test_e2e.py @@ -56,6 +56,7 @@ def _pair(a: str, b: str, id: str, *extra_marks): # plus one remote placement for the sandbox/tunnel boundary. 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"), @@ -231,6 +232,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/verifiers/v1/acp/__init__.py b/verifiers/v1/acp/__init__.py index ef61209dc..cc2a75f6e 100644 --- a/verifiers/v1/acp/__init__.py +++ b/verifiers/v1/acp/__init__.py @@ -1,28 +1,83 @@ """Public Agent Client Protocol support for harness programs.""" +import asyncio import json import secrets -from pathlib import Path +from pathlib import Path, PurePosixPath +from weakref import WeakKeyDictionary +from verifiers.v1.clients import ModelContext 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 from verifiers.v1.types import Messages from verifiers.v1.utils.aio import run_shielded ACP_SOURCE = (Path(__file__).resolve().parent / "_runner.py").read_text() +PROBE_UNAVAILABLE_EXIT_CODE = 75 __all__ = ["ACP"] class ACP: - """Run an ACP agent.""" + """Run one-shot ACP agents or create rollout-scoped ACP sessions.""" + + def __init__(self) -> None: + self._sidecar_locks: WeakKeyDictionary[Runtime, dict[str, asyncio.Lock]] = ( + WeakKeyDictionary() + ) + + def _sidecar_lock(self, runtime: Runtime, sidecar_path: str) -> asyncio.Lock: + locks = self._sidecar_locks.get(runtime) + if locks is None: + locks = {} + self._sidecar_locks[runtime] = locks + lock = locks.get(sidecar_path) + if lock is None: + lock = asyncio.Lock() + locks[sidecar_path] = lock + return lock 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, + acp=self, + env=env, + command=command, + prompt=prompt, + system_prompt=system_prompt, + ) + async def run( self, runtime: Runtime, @@ -34,6 +89,31 @@ 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, + sidecar_path: str | None = None, + allow_empty_tool_reply: bool = False, ) -> ProgramResult: if prompt is None: raise ValueError("ACP requires a prompt") @@ -53,6 +133,53 @@ async def run( program = await runtime.prepare_uv_script( ACP_SOURCE, {**env, "UV_FROZEN": "false"} ) + sidecar_log = None + if sidecar_path is not None: + sidecar_dir = self._sidecar_dir(sidecar_path) + sidecar_log = f"{sidecar_dir}/acp.log" + async with self._sidecar_lock(runtime, sidecar_path): + probe = await runtime.run([*program, "probe", sidecar_path], {}) + if probe.exit_code == PROBE_UNAVAILABLE_EXIT_CODE: + removed = await runtime.run(["rm", "-f", sidecar_path], {}) + if removed.exit_code != 0: + raise RuntimeError( + "stale ACP session cleanup failed: " + f"{removed.stderr.strip()}" + ) + created = await runtime.run( + ["mkdir", "-p", "-m", "700", sidecar_dir], {} + ) + if created.exit_code != 0: + raise RuntimeError( + f"ACP session directory failed: {created.stderr.strip()}" + ) + await runtime.run_background( + [*program, "serve", sidecar_path], + env, + sidecar_log, + ) + ready = await runtime.run( + [*program, "probe", sidecar_path, "60"], {} + ) + if ready.exit_code != 0: + log = await runtime.run(["tail", "-c", "4000", sidecar_log], {}) + detail = ( + ready.stderr.strip() + or ready.stdout.strip() + or "session did not become ready" + ) + if log.exit_code == 0 and log.stdout: + detail = ( + f"{detail}\n\nACP session log:\n{log.stdout.rstrip()}" + ) + raise RuntimeError(f"ACP session failed to start: {detail}") + elif probe.exit_code != 0: + detail = ( + probe.stderr.strip() + or probe.stdout.strip() + or "session did not respond" + ) + raise RuntimeError(f"ACP session probe failed: {detail}") directory = f".vf-acp-{secrets.token_hex(8)}" created = await runtime.run(["mkdir", "-m", "700", directory], {}) if created.exit_code != 0: @@ -60,7 +187,113 @@ 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) + command = ( + [*program, "request", path, sidecar_path] + if sidecar_path is not None + else [*program, "once", path] + ) + result = await runtime.run_program(command, env) + if sidecar_log is not None and result.exit_code != 0: + log = await runtime.run(["tail", "-c", "4000", sidecar_log], {}) + if log.exit_code == 0 and log.stdout: + result = ProgramResult( + exit_code=result.exit_code, + stdout=result.stdout, + stderr=( + f"{result.stderr.rstrip()}\n\nACP session log:\n" + f"{log.stdout.rstrip()}" + ).lstrip(), + ) return result finally: await run_shielded(runtime.run(["rm", "-rf", directory], {})) + + async def _close( + self, + runtime: Runtime, + sidecar_path: str, + ) -> None: + sidecar_dir = self._sidecar_dir(sidecar_path) + exists = await runtime.run(["test", "-S", sidecar_path], {}) + failure = "" + try: + if exists.exit_code == 0: + program = await runtime.prepare_uv_script( + ACP_SOURCE, {"UV_FROZEN": "false"} + ) + result = await runtime.run([*program, "shutdown", sidecar_path], {}) + if result.exit_code != 0: + log = await runtime.run( + ["tail", "-c", "4000", f"{sidecar_dir}/acp.log"], {} + ) + failure = ( + result.stderr.strip() + or result.stdout.strip() + or "ACP session shutdown failed" + ) + if log.exit_code == 0 and log.stdout: + failure = ( + f"{failure}\n\nACP session log:\n{log.stdout.rstrip()}" + ) + finally: + await run_shielded(runtime.run(["rm", "-rf", sidecar_dir], {})) + if failure: + raise RuntimeError(failure) + + @staticmethod + def _sidecar_dir(sidecar_path: str) -> str: + path = PurePosixPath(sidecar_path) + parent = str(path.parent) + if path.is_absolute() or ".." in path.parts or parent in ("", ".", "/"): + raise ValueError("ACP session must live in a private subdirectory") + return parent + + +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, + acp: ACP, + 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.acp = acp + self.env = env + self.command = command + self.prompt = prompt + self.system_prompt = system_prompt + self.sidecar_path = f".vf-acp/{self.trace.id}/acp.sock" + self._started = False + + async def _run(self, messages: Messages | None) -> ProgramResult: + self._started = True + return await self.acp._run( + self.runtime, + self.env, + self.command, + self.prompt if messages is None else messages, + mcp_urls=self.mcp_urls, + system_prompt=self.system_prompt, + sidecar_path=self.sidecar_path, + ) + + async def close(self) -> None: + if self._closed: + return + try: + if self._started: + await self.acp._close(self.runtime, self.sidecar_path) + finally: + await super().close() diff --git a/verifiers/v1/acp/_runner.py b/verifiers/v1/acp/_runner.py index 086b7e814..e2f479bdb 100644 --- a/verifiers/v1/acp/_runner.py +++ b/verifiers/v1/acp/_runner.py @@ -2,12 +2,14 @@ # requires-python = ">=3.10,<3.15" # dependencies = ["agent-client-protocol==0.11.0"] # /// -"""Run one harness segment through an ACP agent.""" +"""Run harness segments through an ACP agent.""" import asyncio import json import os import sys +import traceback +from contextlib import AsyncExitStack, suppress from pathlib import Path from typing import Any @@ -32,6 +34,9 @@ ToolCallUpdate, ) +MAX_PACKET_BYTES = 128 * 1024 * 1024 +PROBE_UNAVAILABLE_EXIT_CODE = 75 + class VerifiersClient(Client): def __init__(self) -> None: @@ -39,6 +44,11 @@ def __init__(self) -> None: 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" @@ -115,7 +125,69 @@ def content_blocks(messages: list[dict], supports_images: bool) -> list: return blocks -async def run_client(config: dict) -> None: +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: VerifiersClient, + 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["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})" + ) + return client.visible_reply + + +async def run_once(config: dict) -> str: client = VerifiersClient() command = config["command"] async with spawn_agent_process( @@ -130,83 +202,313 @@ async def run_client(config: dict) -> None: 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() + servers = mcp_servers(config) if is_new: - session = await connection.new_session( - cwd=os.getcwd(), mcp_servers=mcp_servers - ) + 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=mcp_servers + 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=mcp_servers + cwd=os.getcwd(), session_id=session_id, mcp_servers=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) + reply = await prompt( + client, + connection, + capabilities, + session_id, + config, + is_new=is_new, ) - 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) + return reply -async def main() -> None: - path = Path(sys.argv[1]) +class LiveACPSession: + """One live ACP process, connection, and session shared by several turns.""" + + def __init__(self) -> None: + self.client = VerifiersClient() + 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(reader: asyncio.StreamReader) -> dict: + size = int.from_bytes(await reader.readexactly(8), "big") + if size > MAX_PACKET_BYTES: + raise ValueError(f"ACP session packet is too large: {size} bytes") + return json.loads((await reader.readexactly(size)).decode()) + + +async def write_packet(writer: asyncio.StreamWriter, value: dict) -> None: + data = json.dumps(value, ensure_ascii=False).encode() + writer.write(len(data).to_bytes(8, "big")) + writer.write(data) + await writer.drain() + + +async def serve_session(socket_path: str) -> None: + path = Path(socket_path) + path.unlink(missing_ok=True) + session = LiveACPSession() + lock = asyncio.Lock() + stop_lock = asyncio.Lock() + shutdown = asyncio.Event() + active_prompt: asyncio.Task[str] | None = None + + async def run_prompt(config: dict) -> str: + nonlocal active_prompt + async with lock: + if shutdown.is_set(): + raise RuntimeError("ACP session is shutting down") + active_prompt = asyncio.create_task(session.run(config)) + try: + return await active_prompt + finally: + active_prompt = None + + async def stop_session() -> None: + shutdown.set() + async with stop_lock: + task = active_prompt + if task is not None and not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + # `run_prompt` releases this after its cancelled session request unwinds. + # Holding it for close prevents a waiting prompt from racing a restart. + async with lock: + await session.close() + + async def handle( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + response: dict | None = None + try: + request = await read_packet(reader) + operation = request.get("operation") + if operation == "ping": + response = {"ok": True} + elif operation == "shutdown": + await stop_session() + response = {"ok": True} + elif operation == "prompt": + prompt_task = asyncio.create_task(run_prompt(request["config"])) + disconnect_task = asyncio.create_task(reader.read()) + done, _ = await asyncio.wait( + (prompt_task, disconnect_task), + return_when=asyncio.FIRST_COMPLETED, + ) + if prompt_task in done: + disconnect_task.cancel() + await asyncio.gather(disconnect_task, return_exceptions=True) + response = {"ok": True, "reply": await prompt_task} + else: + # The short-lived request was cancelled or timed out. Stop the + # prompt and agent so neither keeps consuming tokens unattended. + await stop_session() + # `stop_session` cancels and awaits the tracked ACP prompt; + # now let its lock-owning wrapper finish and clear bookkeeping. + await asyncio.gather(prompt_task, return_exceptions=True) + return + else: + raise ValueError(f"unknown ACP session operation: {operation!r}") + except asyncio.CancelledError: + if not shutdown.is_set(): + raise + except Exception as error: # noqa: BLE001 - serialize every request failure + traceback.print_exc() + response = { + "ok": False, + "error": f"{type(error).__name__}: {error}", + } + try: + if response is not None: + await write_packet(writer, response) + except (BrokenPipeError, ConnectionResetError): + pass + finally: + writer.close() + with suppress(BrokenPipeError, ConnectionResetError): + await writer.wait_closed() + + server = await asyncio.start_unix_server(handle, path=socket_path) + os.chmod(path, 0o600) + try: + async with server: + await shutdown.wait() + finally: + server.close() + await server.wait_closed() + await stop_session() + path.unlink(missing_ok=True) + + +async def connect( + socket_path: str, + wait_seconds: float = 60, +) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]: + loop = asyncio.get_running_loop() + deadline = loop.time() + wait_seconds + while True: + try: + return await asyncio.open_unix_connection(socket_path) + except (FileNotFoundError, ConnectionRefusedError): + if loop.time() >= deadline: + raise RuntimeError("timed out waiting for ACP session") + await asyncio.sleep(0.1) + + +async def request_session( + socket_path: str, + request: dict, + wait_seconds: float = 60, + response_seconds: float | None = None, +) -> dict: + reader, writer = await connect(socket_path, wait_seconds) + try: + await write_packet(writer, request) + response = ( + await read_packet(reader) + if response_seconds is None + else await asyncio.wait_for(read_packet(reader), response_seconds) + ) + finally: + writer.close() + await writer.wait_closed() + if not response.get("ok"): + raise RuntimeError(response.get("error") or "ACP session request failed") + return response + + +def read_config(path_value: str) -> dict: + path = Path(path_value) config = json.loads(path.read_text()) path.unlink() - await run_client(config) + 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 == "serve": + await serve_session(sys.argv[2]) + elif operation == "request": + response = await request_session( + sys.argv[3], + {"operation": "prompt", "config": read_config(sys.argv[2])}, + ) + sys.stdout.write(response["reply"]) + elif operation == "shutdown": + await request_session( + sys.argv[2], + {"operation": "shutdown"}, + wait_seconds=2, + response_seconds=5, + ) + elif operation == "probe": + wait_seconds = float(sys.argv[3]) if len(sys.argv) > 3 else 0 + try: + await asyncio.wait_for( + request_session( + sys.argv[2], + {"operation": "ping"}, + wait_seconds=wait_seconds, + ), + timeout=max(2, wait_seconds + 1), + ) + except RuntimeError as error: + if str(error) == "timed out waiting for ACP session": + raise SystemExit(PROBE_UNAVAILABLE_EXIT_CODE) from None + raise + else: + raise ValueError(f"unknown ACP runner operation: {operation!r}") if __name__ == "__main__": diff --git a/verifiers/v1/harness.py b/verifiers/v1/harness.py index fcd54a438..ba3797df7 100644 --- a/verifiers/v1/harness.py +++ b/verifiers/v1/harness.py @@ -127,32 +127,35 @@ 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.""" - async with boundary(HarnessError, f"harness {self.config.id!r}"): - if messages is None: - result = await self.launch( - ctx, trace, runtime, endpoint, secret, mcp_urls, data - ) - else: - result = await self.resume( - ctx, trace, runtime, endpoint, secret, mcp_urls, data, messages - ) - 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}" - ) + """Compatibility entry point for running one segment without retaining a + session handle. Rollouts use `open_session()` and keep its result instead.""" + session = await self.open_session( + ctx, trace, runtime, endpoint, secret, mcp_urls, data + ) + try: + await session.turn(messages) + finally: + await session.close() + + async def open_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 +243,82 @@ 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) + if self.trace.stop_condition is not None: + return # a @stop refused a turn mid-rollout; the 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 self.runtime.alive(): + raise SandboxError( + f"runtime died under harness {self.harness.config.id!r} " + f"(exit {result.exit_code}): {detail}" + ) + raise HarnessError( + f"harness {self.harness.config.id!r} exited " + f"{result.exit_code}: {detail}" + ) + + 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..7f23825a6 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,18 @@ 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_VERSION = "56218f33796ecbe465445bc43948886354fde196" 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.""" + version: str = RLM_VERSION + """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 +85,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 +105,81 @@ 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()}} - ) - # RLM has no interactive mode; resumed segments explicitly replay the transcript. - return await runtime.run_program([RLM_BIN, "--", prompt], env) + return env + + async def open_session( + self, + ctx: ModelContext, + trace: Trace, + runtime: Runtime, + endpoint: str, + secret: str, + mcp_urls: dict[str, str], + data: TaskData, + ) -> HarnessSession: + 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 ACP segment for callers that explicitly bypass `open_session()`.""" + 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 +192,14 @@ 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", self._state_dir(trace)], {}) + + @staticmethod + def _state_dir(trace: Trace) -> str: + return f"{RLM_STATE_DIR}/{trace.id}" + + @classmethod + def _home(cls, trace: Trace) -> str: + return f"{cls._state_dir(trace)}/home" diff --git a/verifiers/v1/rollout.py b/verifiers/v1/rollout.py index 98050f46c..3a9809783 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.open_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() From b915ea7e545f54357a415ca0cf2e73e629120358 Mon Sep 17 00:00:00 2001 From: hallerite Date: Tue, 4 Aug 2026 21:44:13 +0200 Subject: [PATCH 02/17] feat(v1): run harness sessions over live processes # Conflicts: # tests/v1/test_e2e.py # verifiers/v1/__init__.py # verifiers/v1/acp/__init__.py # verifiers/v1/acp/_runner.py --- tests/v1/test_e2e.py | 10 +- verifiers/v1/__init__.py | 5 +- verifiers/v1/acp/__init__.py | 277 +++++++++++------------ verifiers/v1/acp/_runner.py | 251 +++++--------------- verifiers/v1/mcp/launch.py | 31 ++- verifiers/v1/runtimes/__init__.py | 2 + verifiers/v1/runtimes/base.py | 32 +++ verifiers/v1/runtimes/docker/__init__.py | 101 +++++++++ verifiers/v1/runtimes/modal.py | 82 +++++++ verifiers/v1/runtimes/prime.py | 46 ++++ verifiers/v1/runtimes/subprocess.py | 54 ++++- 11 files changed, 548 insertions(+), 343 deletions(-) diff --git a/tests/v1/test_e2e.py b/tests/v1/test_e2e.py index 691f52fc1..c50af639a 100644 --- a/tests/v1/test_e2e.py +++ b/tests/v1/test_e2e.py @@ -51,9 +51,9 @@ 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"), @@ -62,6 +62,7 @@ def _pair(a: str, b: str, id: str, *extra_marks): _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 @@ -216,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, 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 cc2a75f6e..06446e60a 100644 --- a/verifiers/v1/acp/__init__.py +++ b/verifiers/v1/acp/__init__.py @@ -1,22 +1,23 @@ """Public Agent Client Protocol support for harness programs.""" import asyncio +import contextlib import json import secrets -from pathlib import Path, PurePosixPath -from weakref import WeakKeyDictionary +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, HarnessSession -from verifiers.v1.runtimes import ProgramResult, Runtime +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() -PROBE_UNAVAILABLE_EXIT_CODE = 75 +MAX_PACKET_BYTES = 128 * 1024 * 1024 __all__ = ["ACP"] @@ -24,22 +25,6 @@ class ACP: """Run one-shot ACP agents or create rollout-scoped ACP sessions.""" - def __init__(self) -> None: - self._sidecar_locks: WeakKeyDictionary[Runtime, dict[str, asyncio.Lock]] = ( - WeakKeyDictionary() - ) - - def _sidecar_lock(self, runtime: Runtime, sidecar_path: str) -> asyncio.Lock: - locks = self._sidecar_locks.get(runtime) - if locks is None: - locks = {} - self._sidecar_locks[runtime] = locks - lock = locks.get(sidecar_path) - if lock is None: - lock = asyncio.Lock() - locks[sidecar_path] = lock - return lock - async def setup(self, harness: Harness, runtime: Runtime) -> None: await runtime.prepare_uv_script( ACP_SOURCE, {**harness.config.resolved_env, "UV_FROZEN": "false"} @@ -71,7 +56,6 @@ def session( secret, mcp_urls, data, - acp=self, env=env, command=command, prompt=prompt, @@ -112,7 +96,6 @@ async def _run( mcp_urls: dict[str, str] | None = None, system_prompt: str | None = None, session_path: str | None = None, - sidecar_path: str | None = None, allow_empty_tool_reply: bool = False, ) -> ProgramResult: if prompt is None: @@ -133,53 +116,6 @@ async def _run( program = await runtime.prepare_uv_script( ACP_SOURCE, {**env, "UV_FROZEN": "false"} ) - sidecar_log = None - if sidecar_path is not None: - sidecar_dir = self._sidecar_dir(sidecar_path) - sidecar_log = f"{sidecar_dir}/acp.log" - async with self._sidecar_lock(runtime, sidecar_path): - probe = await runtime.run([*program, "probe", sidecar_path], {}) - if probe.exit_code == PROBE_UNAVAILABLE_EXIT_CODE: - removed = await runtime.run(["rm", "-f", sidecar_path], {}) - if removed.exit_code != 0: - raise RuntimeError( - "stale ACP session cleanup failed: " - f"{removed.stderr.strip()}" - ) - created = await runtime.run( - ["mkdir", "-p", "-m", "700", sidecar_dir], {} - ) - if created.exit_code != 0: - raise RuntimeError( - f"ACP session directory failed: {created.stderr.strip()}" - ) - await runtime.run_background( - [*program, "serve", sidecar_path], - env, - sidecar_log, - ) - ready = await runtime.run( - [*program, "probe", sidecar_path, "60"], {} - ) - if ready.exit_code != 0: - log = await runtime.run(["tail", "-c", "4000", sidecar_log], {}) - detail = ( - ready.stderr.strip() - or ready.stdout.strip() - or "session did not become ready" - ) - if log.exit_code == 0 and log.stdout: - detail = ( - f"{detail}\n\nACP session log:\n{log.stdout.rstrip()}" - ) - raise RuntimeError(f"ACP session failed to start: {detail}") - elif probe.exit_code != 0: - detail = ( - probe.stderr.strip() - or probe.stdout.strip() - or "session did not respond" - ) - raise RuntimeError(f"ACP session probe failed: {detail}") directory = f".vf-acp-{secrets.token_hex(8)}" created = await runtime.run(["mkdir", "-m", "700", directory], {}) if created.exit_code != 0: @@ -187,66 +123,38 @@ async def _run( path = f"{directory}/config.json" try: await runtime.write(path, json.dumps(config).encode()) - command = ( - [*program, "request", path, sidecar_path] - if sidecar_path is not None - else [*program, "once", path] - ) - result = await runtime.run_program(command, env) - if sidecar_log is not None and result.exit_code != 0: - log = await runtime.run(["tail", "-c", "4000", sidecar_log], {}) - if log.exit_code == 0 and log.stdout: - result = ProgramResult( - exit_code=result.exit_code, - stdout=result.stdout, - stderr=( - f"{result.stderr.rstrip()}\n\nACP session log:\n" - f"{log.stdout.rstrip()}" - ).lstrip(), - ) - return result + return await runtime.run_program([*program, "once", path], env) finally: await run_shielded(runtime.run(["rm", "-rf", directory], {})) - async def _close( - self, - runtime: Runtime, - sidecar_path: str, - ) -> None: - sidecar_dir = self._sidecar_dir(sidecar_path) - exists = await runtime.run(["test", "-S", sidecar_path], {}) - failure = "" - try: - if exists.exit_code == 0: - program = await runtime.prepare_uv_script( - ACP_SOURCE, {"UV_FROZEN": "false"} - ) - result = await runtime.run([*program, "shutdown", sidecar_path], {}) - if result.exit_code != 0: - log = await runtime.run( - ["tail", "-c", "4000", f"{sidecar_dir}/acp.log"], {} - ) - failure = ( - result.stderr.strip() - or result.stdout.strip() - or "ACP session shutdown failed" - ) - if log.exit_code == 0 and log.stdout: - failure = ( - f"{failure}\n\nACP session log:\n{log.stdout.rstrip()}" - ) - finally: - await run_shielded(runtime.run(["rm", "-rf", sidecar_dir], {})) - if failure: - raise RuntimeError(failure) - @staticmethod - def _sidecar_dir(sidecar_path: str) -> str: - path = PurePosixPath(sidecar_path) - parent = str(path.parent) - if path.is_absolute() or ".." in path.parts or parent in ("", ".", "/"): - raise ValueError("ACP session must live in a private subdirectory") - return parent +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): @@ -262,38 +170,127 @@ def __init__( secret: str, mcp_urls: dict[str, str], data: TaskData, - acp: ACP, 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.acp = acp self.env = env self.command = command self.prompt = prompt self.system_prompt = system_prompt - self.sidecar_path = f".vf-acp/{self.trace.id}/acp.sock" - self._started = False + 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: + 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: - self._started = True - return await self.acp._run( - self.runtime, - self.env, - self.command, - self.prompt if messages is None else messages, - mcp_urls=self.mcp_urls, - system_prompt=self.system_prompt, - sidecar_path=self.sidecar_path, + 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._process is None: + await self._start() + assert self._process is not None + assert self._reader is not None + try: + await self._process.write_stdin( + _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_stdin(_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 + + async def close_process() -> None: + async with self._lock: + await self._stop(graceful=True) + try: - if self._started: - await self.acp._close(self.runtime, self.sidecar_path) + await run_shielded(close_process()) finally: await super().close() diff --git a/verifiers/v1/acp/_runner.py b/verifiers/v1/acp/_runner.py index e2f479bdb..6f6b0baa0 100644 --- a/verifiers/v1/acp/_runner.py +++ b/verifiers/v1/acp/_runner.py @@ -7,6 +7,7 @@ import asyncio import json import os +import signal import sys import traceback from contextlib import AsyncExitStack, suppress @@ -30,33 +31,21 @@ PermissionOption, RequestPermissionResponse, TextContentBlock, - ToolCall, - ToolCallUpdate, ) MAX_PACKET_BYTES = 128 * 1024 * 1024 -PROBE_UNAVAILABLE_EXIT_CODE = 75 class VerifiersClient(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 ): @@ -168,22 +157,10 @@ async def prompt( raise ValueError("ACP prompt has no content") client.reset() try: - response = await connection.prompt(session_id=session_id, prompt=blocks) + 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["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})" - ) return client.visible_reply @@ -319,151 +296,68 @@ async def close(self) -> None: self._reset() -async def read_packet(reader: asyncio.StreamReader) -> dict: - size = int.from_bytes(await reader.readexactly(8), "big") +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") - return json.loads((await reader.readexactly(size)).decode()) + try: + return json.loads((await stream.readexactly(size)).decode()) + except asyncio.IncompleteReadError as error: + raise EOFError("ACP session packet ended early") from error -async def write_packet(writer: asyncio.StreamWriter, value: dict) -> None: +def write_packet(stream: Any, value: dict) -> None: data = json.dumps(value, ensure_ascii=False).encode() - writer.write(len(data).to_bytes(8, "big")) - writer.write(data) - await writer.drain() + 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_session(socket_path: str) -> None: - path = Path(socket_path) - path.unlink(missing_ok=True) +async def serve_stream() -> None: session = LiveACPSession() - lock = asyncio.Lock() - stop_lock = asyncio.Lock() - shutdown = asyncio.Event() - active_prompt: asyncio.Task[str] | None = None - - async def run_prompt(config: dict) -> str: - nonlocal active_prompt - async with lock: - if shutdown.is_set(): - raise RuntimeError("ACP session is shutting down") - active_prompt = asyncio.create_task(session.run(config)) + 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: - return await active_prompt - finally: - active_prompt = None - - async def stop_session() -> None: - shutdown.set() - async with stop_lock: - task = active_prompt - if task is not None and not task.done(): - task.cancel() - await asyncio.gather(task, return_exceptions=True) - # `run_prompt` releases this after its cancelled session request unwinds. - # Holding it for close prevents a waiting prompt from racing a restart. - async with lock: - await session.close() - - async def handle( - reader: asyncio.StreamReader, writer: asyncio.StreamWriter - ) -> None: - response: dict | None = None - try: - request = await read_packet(reader) - operation = request.get("operation") - if operation == "ping": - response = {"ok": True} - elif operation == "shutdown": - await stop_session() - response = {"ok": True} - elif operation == "prompt": - prompt_task = asyncio.create_task(run_prompt(request["config"])) - disconnect_task = asyncio.create_task(reader.read()) - done, _ = await asyncio.wait( - (prompt_task, disconnect_task), - return_when=asyncio.FIRST_COMPLETED, - ) - if prompt_task in done: - disconnect_task.cancel() - await asyncio.gather(disconnect_task, return_exceptions=True) - response = {"ok": True, "reply": await prompt_task} + 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: - # The short-lived request was cancelled or timed out. Stop the - # prompt and agent so neither keeps consuming tokens unattended. - await stop_session() - # `stop_session` cancels and awaits the tracked ACP prompt; - # now let its lock-owning wrapper finish and clear bookkeeping. - await asyncio.gather(prompt_task, return_exceptions=True) - return - else: - raise ValueError(f"unknown ACP session operation: {operation!r}") - except asyncio.CancelledError: - if not shutdown.is_set(): - raise - except Exception as error: # noqa: BLE001 - serialize every request failure - traceback.print_exc() - response = { - "ok": False, - "error": f"{type(error).__name__}: {error}", - } - try: - if response is not None: - await write_packet(writer, response) - except (BrokenPipeError, ConnectionResetError): - pass - finally: - writer.close() - with suppress(BrokenPipeError, ConnectionResetError): - await writer.wait_closed() - - server = await asyncio.start_unix_server(handle, path=socket_path) - os.chmod(path, 0o600) - try: - async with server: - await shutdown.wait() - finally: - server.close() - await server.wait_closed() - await stop_session() - path.unlink(missing_ok=True) - - -async def connect( - socket_path: str, - wait_seconds: float = 60, -) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]: - loop = asyncio.get_running_loop() - deadline = loop.time() + wait_seconds - while True: - try: - return await asyncio.open_unix_connection(socket_path) - except (FileNotFoundError, ConnectionRefusedError): - if loop.time() >= deadline: - raise RuntimeError("timed out waiting for ACP session") - await asyncio.sleep(0.1) - - -async def request_session( - socket_path: str, - request: dict, - wait_seconds: float = 60, - response_seconds: float | None = None, -) -> dict: - reader, writer = await connect(socket_path, wait_seconds) - try: - await write_packet(writer, request) - response = ( - await read_packet(reader) - if response_seconds is None - else await asyncio.wait_for(read_packet(reader), response_seconds) - ) + 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: - writer.close() - await writer.wait_closed() - if not response.get("ok"): - raise RuntimeError(response.get("error") or "ACP session request failed") - return response + if not closed: + await session.close() def read_config(path_value: str) -> dict: @@ -477,36 +371,15 @@ async def main() -> None: operation = sys.argv[1] if operation == "once": sys.stdout.write(await run_once(read_config(sys.argv[2]))) - elif operation == "serve": - await serve_session(sys.argv[2]) - elif operation == "request": - response = await request_session( - sys.argv[3], - {"operation": "prompt", "config": read_config(sys.argv[2])}, - ) - sys.stdout.write(response["reply"]) - elif operation == "shutdown": - await request_session( - sys.argv[2], - {"operation": "shutdown"}, - wait_seconds=2, - response_seconds=5, - ) - elif operation == "probe": - wait_seconds = float(sys.argv[3]) if len(sys.argv) > 3 else 0 - try: - await asyncio.wait_for( - request_session( - sys.argv[2], - {"operation": "ping"}, - wait_seconds=wait_seconds, - ), - timeout=max(2, wait_seconds + 1), - ) - except RuntimeError as error: - if str(error) == "timed out waiting for ACP session": - raise SystemExit(PROBE_UNAVAILABLE_EXIT_CODE) from None - raise + 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}") 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/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..3d4a771d6 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_stdin(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, @@ -216,6 +240,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..b75005be9 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,6 +21,7 @@ NetworkPolicyConfig, ProgramResult, Runtime, + RuntimeProcess, parse_gpu, ) from verifiers.v1.runtimes.docker.egress import HOST_ALIAS, EgressProxy, NetworkPolicy @@ -47,6 +50,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_stdin(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"', + "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", @@ -313,6 +370,50 @@ 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" + wrapper = '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 + 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: + if proc.returncode is None: + proc.kill() + _, stderr = await proc.communicate() + detail = stderr.decode(errors="replace").strip() or ready.stderr.strip() + raise SandboxError( + f"docker live process failed to start: {detail or 'PID unavailable'}" + ) + await asyncio.sleep(0.05) + 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..97fef358c 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_stdin(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,45 @@ 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, + ) + if await proc.poll.aio() is not None or loop.time() >= deadline: + 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..5fedc21ef 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_stdin(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 @@ -241,6 +262,31 @@ 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" + ) + open_process = getattr(self._client, "open_process", None) + if open_process is None: + raise SandboxError( + "Prime VM live processes require a prime-sandboxes SDK version " + "that provides AsyncSandboxClient.open_process()" + ) + try: + process = await 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..731c60eaf 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,35 @@ class SubprocessRuntimeInfo(SubprocessConfig, BaseRuntimeInfo): pass +async def _read_stream(reader: asyncio.StreamReader) -> AsyncIterator[bytes]: + while chunk := await reader.read(64 * 1024): + yield chunk + + +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_stdin(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: + SubprocessRuntime._signal(self._process, signal.SIGTERM) + + async def kill(self) -> None: + SubprocessRuntime._signal(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 +110,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: From f880ac9f347166ac58ec3275ecc19d01dd6f7c74 Mon Sep 17 00:00:00 2001 From: hallerite Date: Wed, 5 Aug 2026 01:37:59 +0200 Subject: [PATCH 03/17] fix(v1): address live harness review feedback --- verifiers/v1/acp/__init__.py | 21 +++++++++----- verifiers/v1/acp/{_runner.py => runner.py} | 0 verifiers/v1/harness.py | 23 +--------------- verifiers/v1/harnesses/rlm/harness.py | 17 ++++-------- verifiers/v1/interception/server.py | 2 +- verifiers/v1/rollout.py | 2 +- verifiers/v1/runtimes/base.py | 2 +- verifiers/v1/runtimes/docker/__init__.py | 15 ++++++++-- verifiers/v1/runtimes/modal.py | 21 ++++++++++++-- verifiers/v1/runtimes/prime.py | 2 +- verifiers/v1/runtimes/subprocess.py | 32 ++++++++++++---------- verifiers/v1/session.py | 2 +- 12 files changed, 74 insertions(+), 65 deletions(-) rename verifiers/v1/acp/{_runner.py => runner.py} (100%) diff --git a/verifiers/v1/acp/__init__.py b/verifiers/v1/acp/__init__.py index 06446e60a..2af8c6e4b 100644 --- a/verifiers/v1/acp/__init__.py +++ b/verifiers/v1/acp/__init__.py @@ -9,6 +9,7 @@ from verifiers.v1.clients import ModelContext from verifiers.v1.dialects.chat import message_to_wire +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 @@ -16,7 +17,7 @@ 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"] @@ -187,6 +188,7 @@ def __init__( 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"} ) @@ -221,12 +223,16 @@ async def _run(self, messages: Messages | None) -> ProgramResult: "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_stdin( + await self._process.write( _packet({"operation": "prompt", "config": config}) ) response = await self._reader.read() @@ -250,7 +256,7 @@ async def _stop(self, *, graceful: bool) -> None: try: if graceful and reader is not None: try: - await process.write_stdin(_packet({"operation": "shutdown"})) + await process.write(_packet({"operation": "shutdown"})) response = await asyncio.wait_for(reader.read(), timeout=10) if not response.get("ok"): raise RuntimeError( @@ -285,12 +291,13 @@ async def _stop(self, *, graceful: bool) -> None: 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) - try: - await run_shielded(close_process()) - finally: - await super().close() + await run_shielded(close_process()) diff --git a/verifiers/v1/acp/_runner.py b/verifiers/v1/acp/runner.py similarity index 100% rename from verifiers/v1/acp/_runner.py rename to verifiers/v1/acp/runner.py diff --git a/verifiers/v1/harness.py b/verifiers/v1/harness.py index ba3797df7..e45c576a8 100644 --- a/verifiers/v1/harness.py +++ b/verifiers/v1/harness.py @@ -116,28 +116,7 @@ async def install_skills(self, runtime: Runtime, dest: str) -> None: # `write` moves bytes, not modes; restore the execute bits scripts need. await runtime.run(["chmod", "+x", *executables], {}) - async def run( - self, - ctx: ModelContext, - trace: Trace, - runtime: Runtime, - endpoint: str, - secret: str, - mcp_urls: dict[str, str], - data: TaskData, - messages: Messages | None = None, - ) -> None: - """Compatibility entry point for running one segment without retaining a - session handle. Rollouts use `open_session()` and keep its result instead.""" - session = await self.open_session( - ctx, trace, runtime, endpoint, secret, mcp_urls, data - ) - try: - await session.turn(messages) - finally: - await session.close() - - async def open_session( + async def session( self, ctx: ModelContext, trace: Trace, diff --git a/verifiers/v1/harnesses/rlm/harness.py b/verifiers/v1/harnesses/rlm/harness.py index 7f23825a6..eb9e17ff4 100644 --- a/verifiers/v1/harnesses/rlm/harness.py +++ b/verifiers/v1/harnesses/rlm/harness.py @@ -22,7 +22,6 @@ BuiltinSkill = Literal["edit", "search"] RLM_REPO = "github.com/PrimeIntellect-ai/rlm-harness.git" -RLM_VERSION = "56218f33796ecbe465445bc43948886354fde196" RLM_DIR = "/tmp/vf-rlm" RLM_BIN = f"{RLM_DIR}/bin/rlm" SKILLS_DIR = "/task/rlm-skills" @@ -31,7 +30,7 @@ class RLMHarnessConfig(HarnessConfig): - version: str = RLM_VERSION + version: str = "main" """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).""" @@ -129,7 +128,7 @@ def _env( env["RLM_SKILLS"] = ",".join(self.config.builtin_skills) return env - async def open_session( + async def session( self, ctx: ModelContext, trace: Trace, @@ -164,7 +163,7 @@ async def launch( mcp_urls: dict[str, str], data: TaskData, ) -> ProgramResult: - """Run one ACP segment for callers that explicitly bypass `open_session()`.""" + """Run one standalone ACP segment through the default session adapter.""" system_prompt, prompt = self.resolve_prompt(data) return await RLM_ACP.run( runtime, @@ -194,12 +193,8 @@ async def rlm(self, trace: Trace, runtime: Runtime) -> dict[str, float]: } async def cleanup(self, trace: Trace, runtime: Runtime) -> None: - await runtime.run(["rm", "-rf", self._state_dir(trace)], {}) + await runtime.run(["rm", "-rf", f"{RLM_STATE_DIR}/{trace.id}"], {}) @staticmethod - def _state_dir(trace: Trace) -> str: - return f"{RLM_STATE_DIR}/{trace.id}" - - @classmethod - def _home(cls, trace: Trace) -> str: - return f"{cls._state_dir(trace)}/home" + 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/rollout.py b/verifiers/v1/rollout.py index 3a9809783..285d3c6ca 100644 --- a/verifiers/v1/rollout.py +++ b/verifiers/v1/rollout.py @@ -234,7 +234,7 @@ async def open(self) -> bool: # 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.open_session( + self._harness_session = await self.harness.session( self.ctx, self.trace, runtime, diff --git a/verifiers/v1/runtimes/base.py b/verifiers/v1/runtimes/base.py index 3d4a771d6..b2c9f55ab 100644 --- a/verifiers/v1/runtimes/base.py +++ b/verifiers/v1/runtimes/base.py @@ -62,7 +62,7 @@ class RuntimeProcess(ABC): stderr: AsyncIterator[bytes] @abstractmethod - async def write_stdin(self, data: bytes) -> None: + async def write(self, data: bytes) -> None: pass @abstractmethod diff --git a/verifiers/v1/runtimes/docker/__init__.py b/verifiers/v1/runtimes/docker/__init__.py index b75005be9..2f0f9b208 100644 --- a/verifiers/v1/runtimes/docker/__init__.py +++ b/verifiers/v1/runtimes/docker/__init__.py @@ -72,7 +72,7 @@ def __init__( self.stdout = _read_stream(process.stdout) self.stderr = _read_stream(process.stderr) - async def write_stdin(self, data: bytes) -> None: + async def write(self, data: bytes) -> None: self._stdin.write(data) await self._stdin.drain() @@ -93,7 +93,7 @@ async def _signal(self, signal: str) -> None: self._container, "sh", "-c", - 'kill -"$1" "$2"', + 'kill -"$1" "-$2" 2>/dev/null || kill -"$1" "$2"', "vf-signal", signal, str(self._pid), @@ -379,7 +379,16 @@ async def open_process( arg for key, value in env.items() for arg in ("--env", f"{key}={value}") ] pidfile = f"/tmp/vf-process-{uuid.uuid4().hex}.pid" - wrapper = 'echo $$ > "$1"; shift; exec "$@"' + # 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", diff --git a/verifiers/v1/runtimes/modal.py b/verifiers/v1/runtimes/modal.py index 97fef358c..4676bc14f 100644 --- a/verifiers/v1/runtimes/modal.py +++ b/verifiers/v1/runtimes/modal.py @@ -70,7 +70,7 @@ def __init__(self, process, sandbox, pid: int, workdir: str) -> None: self.stdout: AsyncIterator[bytes] = process.stdout self.stderr: AsyncIterator[bytes] = process.stderr - async def write_stdin(self, data: bytes) -> None: + async def write(self, data: bytes) -> None: await self._process.stdin.write.aio(data) await self._process.stdin.drain.aio() @@ -217,7 +217,24 @@ async def open_process( int(ready.stdout.strip()), self.config.workdir, ) - if await proc.poll.aio() is not None or loop.time() >= deadline: + 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'}" diff --git a/verifiers/v1/runtimes/prime.py b/verifiers/v1/runtimes/prime.py index 5fedc21ef..a44c60450 100644 --- a/verifiers/v1/runtimes/prime.py +++ b/verifiers/v1/runtimes/prime.py @@ -108,7 +108,7 @@ def __init__(self, process) -> None: self.stdout: AsyncIterator[bytes] = process.stdout self.stderr: AsyncIterator[bytes] = process.stderr - async def write_stdin(self, data: bytes) -> None: + async def write(self, data: bytes) -> None: await self._process.write_stdin(data) async def wait(self) -> int: diff --git a/verifiers/v1/runtimes/subprocess.py b/verifiers/v1/runtimes/subprocess.py index 731c60eaf..9527916b2 100644 --- a/verifiers/v1/runtimes/subprocess.py +++ b/verifiers/v1/runtimes/subprocess.py @@ -39,6 +39,17 @@ async def _read_stream(reader: asyncio.StreamReader) -> AsyncIterator[bytes]: 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 @@ -49,7 +60,7 @@ def __init__(self, process: asyncio.subprocess.Process) -> None: self.stdout = _read_stream(process.stdout) self.stderr = _read_stream(process.stderr) - async def write_stdin(self, data: bytes) -> None: + async def write(self, data: bytes) -> None: self._stdin.write(data) await self._stdin.drain() @@ -57,10 +68,10 @@ async def wait(self) -> int: return await self._process.wait() async def terminate(self) -> None: - SubprocessRuntime._signal(self._process, signal.SIGTERM) + _signal_process(self._process, signal.SIGTERM) async def kill(self) -> None: - SubprocessRuntime._signal(self._process, signal.SIGKILL) + _signal_process(self._process, signal.SIGKILL) class SubprocessRuntime(Runtime): @@ -156,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( @@ -180,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( @@ -195,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: From 5f7dc2dc55607325b570b329c6310f37908ee100 Mon Sep 17 00:00:00 2001 From: hallerite Date: Wed, 5 Aug 2026 01:50:56 +0200 Subject: [PATCH 04/17] fix(v1): reap cancelled Docker processes --- verifiers/v1/runtimes/docker/__init__.py | 79 +++++++++++++++++++----- 1 file changed, 65 insertions(+), 14 deletions(-) diff --git a/verifiers/v1/runtimes/docker/__init__.py b/verifiers/v1/runtimes/docker/__init__.py index 2f0f9b208..c38691828 100644 --- a/verifiers/v1/runtimes/docker/__init__.py +++ b/verifiers/v1/runtimes/docker/__init__.py @@ -25,6 +25,7 @@ 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__) @@ -111,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"), @@ -119,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 @@ -409,19 +452,27 @@ async def open_process( ) loop = asyncio.get_running_loop() deadline = loop.time() + 5 - 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: - if proc.returncode is None: - proc.kill() - _, stderr = await proc.communicate() - detail = stderr.decode(errors="replace").strip() or ready.stderr.strip() - raise SandboxError( - f"docker live process failed to start: {detail or 'PID unavailable'}" - ) - await asyncio.sleep(0.05) + 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 From 7b91006494ae1e3fd4666d10fa93ae8ca6160e1c Mon Sep 17 00:00:00 2001 From: hallerite Date: Wed, 5 Aug 2026 03:55:58 +0200 Subject: [PATCH 05/17] fix(v1): preserve stateless harness runs --- verifiers/v1/harness.py | 58 ++++++++++++++++++++------- verifiers/v1/harnesses/rlm/harness.py | 4 ++ verifiers/v1/runtimes/base.py | 5 +++ verifiers/v1/runtimes/prime.py | 4 ++ 4 files changed, 57 insertions(+), 14 deletions(-) diff --git a/verifiers/v1/harness.py b/verifiers/v1/harness.py index e45c576a8..f1e7fd2f4 100644 --- a/verifiers/v1/harness.py +++ b/verifiers/v1/harness.py @@ -116,6 +116,49 @@ async def install_skills(self, runtime: Runtime, dest: str) -> None: # `write` moves bytes, not modes; restore the execute bits scripts need. await runtime.run(["chmod", "+x", *executables], {}) + async def run( + self, + ctx: ModelContext, + trace: Trace, + runtime: Runtime, + endpoint: str, + secret: str, + mcp_urls: dict[str, str], + data: TaskData, + messages: Messages | None = None, + ) -> None: + """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( + ctx, trace, runtime, endpoint, secret, mcp_urls, data + ) + else: + 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 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, @@ -261,20 +304,7 @@ async def turn(self, messages: Messages | None = None) -> None: ) async with boundary(HarnessError, f"harness {self.harness.config.id!r}"): result = await self._run(messages) - if self.trace.stop_condition is not None: - return # a @stop refused a turn mid-rollout; the 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 self.runtime.alive(): - raise SandboxError( - f"runtime died under harness {self.harness.config.id!r} " - f"(exit {result.exit_code}): {detail}" - ) - raise HarnessError( - f"harness {self.harness.config.id!r} exited " - f"{result.exit_code}: {detail}" - ) + await self.harness._check_result(self.trace, self.runtime, result) async def _run(self, messages: Messages | None) -> ProgramResult: if messages is None: diff --git a/verifiers/v1/harnesses/rlm/harness.py b/verifiers/v1/harnesses/rlm/harness.py index eb9e17ff4..fcdf24441 100644 --- a/verifiers/v1/harnesses/rlm/harness.py +++ b/verifiers/v1/harnesses/rlm/harness.py @@ -138,6 +138,10 @@ async def session( 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 + ) system_prompt, prompt = self.resolve_prompt(data) return RLM_ACP.session( self, diff --git a/verifiers/v1/runtimes/base.py b/verifiers/v1/runtimes/base.py index b2c9f55ab..883394b52 100644 --- a/verifiers/v1/runtimes/base.py +++ b/verifiers/v1/runtimes/base.py @@ -178,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] = {} diff --git a/verifiers/v1/runtimes/prime.py b/verifiers/v1/runtimes/prime.py index a44c60450..e107fb0ff 100644 --- a/verifiers/v1/runtimes/prime.py +++ b/verifiers/v1/runtimes/prime.py @@ -130,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 From e69250e519564fae0b623dce296a8686dbe1b46f Mon Sep 17 00:00:00 2001 From: hallerite Date: Wed, 5 Aug 2026 13:21:07 +0200 Subject: [PATCH 06/17] test(v1): exercise Prime VM live sessions --- .github/workflows/test.yml | 4 ++++ pyproject.toml | 1 + uv.lock | 8 ++------ 3 files changed, 7 insertions(+), 6 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 8dc6ddbaa..9abf29a81 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -109,3 +109,7 @@ jobs: - name: Run live v1 E2Es run: | uv run pytest tests/v1 -vv -n auto -m "e2e and not prime and not modal" --cov=verifiers --cov-report=xml --cov-report=term + + - name: Run Prime VM persistent RLM E2E + run: | + uv run pytest tests/v1/test_e2e.py::test_acp_resume_with_tool -vv -m "e2e and prime" -k "rlm-acp-in-prime-vm" diff --git a/pyproject.toml b/pyproject.toml index a8602341d..686ed79da 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -119,6 +119,7 @@ url = "https://pypi.org/simple" default = true [tool.uv.sources] +prime-sandboxes = { git = "https://github.com/PrimeIntellect-ai/prime", rev = "892bea7cb3d3f0926d9b42da1ebf206207bb8bab", subdirectory = "packages/prime-sandboxes" } compact = { path = "environments/compact", editable = true } glossary-v1 = { path = "environments/glossary_v1", editable = true } deepwiki-v1 = { path = "environments/deepwiki_v1", editable = true } diff --git a/uv.lock b/uv.lock index 38298ea39..359376f5c 100644 --- a/uv.lock +++ b/uv.lock @@ -3136,8 +3136,8 @@ toml = [ [[package]] name = "prime-sandboxes" -version = "0.2.33" -source = { registry = "https://pypi.org/simple" } +version = "0.2.34" +source = { git = "https://github.com/PrimeIntellect-ai/prime?subdirectory=packages%2Fprime-sandboxes&rev=892bea7cb3d3f0926d9b42da1ebf206207bb8bab#892bea7cb3d3f0926d9b42da1ebf206207bb8bab" } dependencies = [ { name = "aiofiles" }, { name = "connect-python" }, @@ -3146,10 +3146,6 @@ 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" } -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" }, -] [[package]] name = "prime-tunnel" From 14ef360b783d72826cdd36aff73b0c308aacb73a Mon Sep 17 00:00:00 2001 From: hallerite Date: Wed, 5 Aug 2026 13:31:45 +0200 Subject: [PATCH 07/17] ci(v1): run Prime VM session check first --- .github/workflows/test.yml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 9abf29a81..f5aa0ec7c 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -106,10 +106,10 @@ jobs: - name: Install dependencies run: uv sync - - name: Run live v1 E2Es - run: | - uv run pytest tests/v1 -vv -n auto -m "e2e and not prime and not modal" --cov=verifiers --cov-report=xml --cov-report=term - - name: Run Prime VM persistent RLM E2E run: | uv run pytest tests/v1/test_e2e.py::test_acp_resume_with_tool -vv -m "e2e and prime" -k "rlm-acp-in-prime-vm" + + - name: Run live v1 E2Es + run: | + uv run pytest tests/v1 -vv -n auto -m "e2e and not prime and not modal" --cov=verifiers --cov-report=xml --cov-report=term From 0475b2385ab867a5d4a7dbd19a52dca7813300d3 Mon Sep 17 00:00:00 2001 From: hallerite Date: Wed, 5 Aug 2026 13:34:56 +0200 Subject: [PATCH 08/17] ci(v1): remove temporary Prime VM check --- .github/workflows/test.yml | 4 ---- 1 file changed, 4 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index f5aa0ec7c..8dc6ddbaa 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -106,10 +106,6 @@ jobs: - name: Install dependencies run: uv sync - - name: Run Prime VM persistent RLM E2E - run: | - uv run pytest tests/v1/test_e2e.py::test_acp_resume_with_tool -vv -m "e2e and prime" -k "rlm-acp-in-prime-vm" - - name: Run live v1 E2Es run: | uv run pytest tests/v1 -vv -n auto -m "e2e and not prime and not modal" --cov=verifiers --cov-report=xml --cov-report=term From c583beb5e98eabbbc43d3e5e067f2c9e42cde8a2 Mon Sep 17 00:00:00 2001 From: hallerite Date: Wed, 5 Aug 2026 13:54:32 +0200 Subject: [PATCH 09/17] test(v1): run Kuhn Poker over Prime VM sessions --- .github/workflows/test.yml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 8dc6ddbaa..8c90467e4 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -106,6 +106,11 @@ jobs: - name: Install dependencies run: uv sync + - name: Run Kuhn Poker with RLM in Prime VMs + run: | + uv run eval kuhn-poker-v1 -m deepseek/deepseek-v4-flash-0731 -n 1 -r 1 -c 1 -o /tmp/kuhn-rlm-prime-e2e --no-push --no-rich --sampling.max-tokens 8192 --env.player0.harness.id rlm --env.player0.runtime.type prime --env.player0.runtime.vm true --env.player0.max-turns 8 --env.player0.max-output-tokens 8192 --env.player0.timeout.rollout 600 --env.player1.harness.id rlm --env.player1.runtime.type prime --env.player1.runtime.vm true --env.player1.max-turns 8 --env.player1.max-output-tokens 8192 --env.player1.timeout.rollout 600 + uv run python -c 'import json; from pathlib import Path; rows = [json.loads(line) for line in Path("/tmp/kuhn-rlm-prime-e2e/traces.jsonl").read_text().splitlines() if line.strip()]; assert len(rows) == 1, rows; row = rows[0]; traces = row["traces"]; assert row["ok"] and len(traces) == 2, row.get("errors"); assert all(t["ok"] and not t["errors"] and t["stop_condition"] == "user_closed" and t["info"]["kuhn"]["forfeited"] is None for t in traces), traces; assert sorted(t["agent"]["name"] for t in traces) == ["player0", "player1"]; assert sum(t["rewards"]["payoff"]["score"] for t in traces) == 0; print(json.dumps([{"agent": t["agent"]["name"], "card": t["info"]["kuhn"]["card"], "history": t["info"]["kuhn"]["history"], "turns": sum(bool(n.get("sampled")) for n in t["nodes"]), "payoff": t["rewards"]["payoff"]["score"]} for t in traces], indent=2))' + - name: Run live v1 E2Es run: | uv run pytest tests/v1 -vv -n auto -m "e2e and not prime and not modal" --cov=verifiers --cov-report=xml --cov-report=term From adb6bd6054830319dba53ffae2adb2efa2d15bd6 Mon Sep 17 00:00:00 2001 From: hallerite Date: Wed, 5 Aug 2026 14:05:18 +0200 Subject: [PATCH 10/17] test(v1): expand real Prime session workloads --- .github/workflows/test.yml | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 8c90467e4..5a9fdaf49 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -108,8 +108,15 @@ jobs: - name: Run Kuhn Poker with RLM in Prime VMs run: | - uv run eval kuhn-poker-v1 -m deepseek/deepseek-v4-flash-0731 -n 1 -r 1 -c 1 -o /tmp/kuhn-rlm-prime-e2e --no-push --no-rich --sampling.max-tokens 8192 --env.player0.harness.id rlm --env.player0.runtime.type prime --env.player0.runtime.vm true --env.player0.max-turns 8 --env.player0.max-output-tokens 8192 --env.player0.timeout.rollout 600 --env.player1.harness.id rlm --env.player1.runtime.type prime --env.player1.runtime.vm true --env.player1.max-turns 8 --env.player1.max-output-tokens 8192 --env.player1.timeout.rollout 600 - uv run python -c 'import json; from pathlib import Path; rows = [json.loads(line) for line in Path("/tmp/kuhn-rlm-prime-e2e/traces.jsonl").read_text().splitlines() if line.strip()]; assert len(rows) == 1, rows; row = rows[0]; traces = row["traces"]; assert row["ok"] and len(traces) == 2, row.get("errors"); assert all(t["ok"] and not t["errors"] and t["stop_condition"] == "user_closed" and t["info"]["kuhn"]["forfeited"] is None for t in traces), traces; assert sorted(t["agent"]["name"] for t in traces) == ["player0", "player1"]; assert sum(t["rewards"]["payoff"]["score"] for t in traces) == 0; print(json.dumps([{"agent": t["agent"]["name"], "card": t["info"]["kuhn"]["card"], "history": t["info"]["kuhn"]["history"], "turns": sum(bool(n.get("sampled")) for n in t["nodes"]), "payoff": t["rewards"]["payoff"]["score"]} for t in traces], indent=2))' + uv run eval kuhn-poker-v1 -m deepseek/deepseek-v4-flash-0731 -n 3 -r 1 -c 1 -o /tmp/kuhn-rlm-prime-e2e --no-push --no-rich --sampling.max-tokens 8192 --env.player0.harness.id rlm --env.player0.runtime.type prime --env.player0.runtime.vm true --env.player0.max-turns 8 --env.player0.max-output-tokens 32768 --env.player0.timeout.rollout 600 --env.player1.harness.id rlm --env.player1.runtime.type prime --env.player1.runtime.vm true --env.player1.max-turns 8 --env.player1.max-output-tokens 32768 --env.player1.timeout.rollout 600 + uv run python -c 'import json; from pathlib import Path; rows = [json.loads(line) for line in Path("/tmp/kuhn-rlm-prime-e2e/traces.jsonl").read_text().splitlines() if line.strip()]; assert len(rows) == 3, rows; assert all(row["ok"] and len(row["traces"]) == 2 for row in rows), rows; traces = [trace for row in rows for trace in row["traces"]]; assert all(t["ok"] and not t["errors"] and t["stop_condition"] == "user_closed" and t["info"]["kuhn"]["forfeited"] is None for t in traces), traces; assert all(sorted(t["agent"]["name"] for t in row["traces"]) == ["player0", "player1"] for row in rows); assert all(sum(t["rewards"]["payoff"]["score"] for t in row["traces"]) == 0 for row in rows); summaries = [{"seed": t["info"]["kuhn"]["seed"], "agent": t["agent"]["name"], "card": t["info"]["kuhn"]["card"], "history": t["info"]["kuhn"]["history"], "turns": sum(bool(n.get("sampled")) for n in t["nodes"]), "payoff": t["rewards"]["payoff"]["score"]} for t in traces]; assert any(s["turns"] >= 2 for s in summaries), summaries; print(json.dumps(summaries, indent=2))' + + - name: Run modeled user simulation with RLM and MCP + env: + PYTHONPATH: tests/v1/fixtures + run: | + uv run eval echo-tool-v1 --env.id user-sim -m deepseek/deepseek-v4-flash-0731 -n 1 -r 1 -c 1 -o /tmp/user-sim-rlm-prime-e2e --no-push --no-rich --sampling.max-tokens 8192 --env.taskset.task.tools.colocated true --env.assistant.harness.id rlm --env.assistant.runtime.type prime --env.assistant.runtime.vm true --env.assistant.max-turns 8 --env.assistant.max-output-tokens 32768 --env.assistant.timeout.rollout 600 --env.user.max-turns 8 --env.user.max-output-tokens 32768 --env.user.timeout.rollout 600 + uv run python -c 'import json; from pathlib import Path; rows = [json.loads(line) for line in Path("/tmp/user-sim-rlm-prime-e2e/traces.jsonl").read_text().splitlines() if line.strip()]; assert len(rows) == 1 and rows[0]["ok"], rows; traces = rows[0]["traces"]; assert len(traces) == 2 and all(t["ok"] and not t["errors"] for t in traces), traces; assistant = next(t for t in traces if t["agent"]["name"] == "assistant"); user = next(t for t in traces if t["agent"]["name"] == "user"); assert assistant["rewards"]["echoed"]["score"] == 1.0, assistant; assert assistant["metrics"]["user_turns"] >= 1, assistant; assert any(tool["name"] == "echo_back" for tool in assistant["tools"]), assistant["tools"]; print(json.dumps({"assistant_turns": sum(bool(n.get("sampled")) for n in assistant["nodes"]), "user_turns": sum(bool(n.get("sampled")) for n in user["nodes"]), "echoed_reward": assistant["rewards"]["echoed"]["score"], "assistant_stop": assistant["stop_condition"], "user_stop": user["stop_condition"]}, indent=2))' - name: Run live v1 E2Es run: | From efe7ff654b7d00bec8b9c68aec6cc5cb0bc9a557 Mon Sep 17 00:00:00 2001 From: hallerite Date: Wed, 5 Aug 2026 14:17:38 +0200 Subject: [PATCH 11/17] test(v1): isolate real session workloads --- .github/workflows/test.yml | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 5a9fdaf49..1da5259f7 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -106,18 +106,20 @@ jobs: - name: Install dependencies run: uv sync - - name: Run Kuhn Poker with RLM in Prime VMs - run: | - uv run eval kuhn-poker-v1 -m deepseek/deepseek-v4-flash-0731 -n 3 -r 1 -c 1 -o /tmp/kuhn-rlm-prime-e2e --no-push --no-rich --sampling.max-tokens 8192 --env.player0.harness.id rlm --env.player0.runtime.type prime --env.player0.runtime.vm true --env.player0.max-turns 8 --env.player0.max-output-tokens 32768 --env.player0.timeout.rollout 600 --env.player1.harness.id rlm --env.player1.runtime.type prime --env.player1.runtime.vm true --env.player1.max-turns 8 --env.player1.max-output-tokens 32768 --env.player1.timeout.rollout 600 - uv run python -c 'import json; from pathlib import Path; rows = [json.loads(line) for line in Path("/tmp/kuhn-rlm-prime-e2e/traces.jsonl").read_text().splitlines() if line.strip()]; assert len(rows) == 3, rows; assert all(row["ok"] and len(row["traces"]) == 2 for row in rows), rows; traces = [trace for row in rows for trace in row["traces"]]; assert all(t["ok"] and not t["errors"] and t["stop_condition"] == "user_closed" and t["info"]["kuhn"]["forfeited"] is None for t in traces), traces; assert all(sorted(t["agent"]["name"] for t in row["traces"]) == ["player0", "player1"] for row in rows); assert all(sum(t["rewards"]["payoff"]["score"] for t in row["traces"]) == 0 for row in rows); summaries = [{"seed": t["info"]["kuhn"]["seed"], "agent": t["agent"]["name"], "card": t["info"]["kuhn"]["card"], "history": t["info"]["kuhn"]["history"], "turns": sum(bool(n.get("sampled")) for n in t["nodes"]), "payoff": t["rewards"]["payoff"]["score"]} for t in traces]; assert any(s["turns"] >= 2 for s in summaries), summaries; print(json.dumps(summaries, indent=2))' - - name: Run modeled user simulation with RLM and MCP + if: always() env: PYTHONPATH: tests/v1/fixtures run: | uv run eval echo-tool-v1 --env.id user-sim -m deepseek/deepseek-v4-flash-0731 -n 1 -r 1 -c 1 -o /tmp/user-sim-rlm-prime-e2e --no-push --no-rich --sampling.max-tokens 8192 --env.taskset.task.tools.colocated true --env.assistant.harness.id rlm --env.assistant.runtime.type prime --env.assistant.runtime.vm true --env.assistant.max-turns 8 --env.assistant.max-output-tokens 32768 --env.assistant.timeout.rollout 600 --env.user.max-turns 8 --env.user.max-output-tokens 32768 --env.user.timeout.rollout 600 uv run python -c 'import json; from pathlib import Path; rows = [json.loads(line) for line in Path("/tmp/user-sim-rlm-prime-e2e/traces.jsonl").read_text().splitlines() if line.strip()]; assert len(rows) == 1 and rows[0]["ok"], rows; traces = rows[0]["traces"]; assert len(traces) == 2 and all(t["ok"] and not t["errors"] for t in traces), traces; assistant = next(t for t in traces if t["agent"]["name"] == "assistant"); user = next(t for t in traces if t["agent"]["name"] == "user"); assert assistant["rewards"]["echoed"]["score"] == 1.0, assistant; assert assistant["metrics"]["user_turns"] >= 1, assistant; assert any(tool["name"] == "echo_back" for tool in assistant["tools"]), assistant["tools"]; print(json.dumps({"assistant_turns": sum(bool(n.get("sampled")) for n in assistant["nodes"]), "user_turns": sum(bool(n.get("sampled")) for n in user["nodes"]), "echoed_reward": assistant["rewards"]["echoed"]["score"], "assistant_stop": assistant["stop_condition"], "user_stop": user["stop_condition"]}, indent=2))' + - name: Run Kuhn Poker with RLM in Prime VMs + if: always() + run: | + uv run eval kuhn-poker-v1 -m deepseek/deepseek-v4-flash-0731 -n 3 -r 1 -c 1 -o /tmp/kuhn-rlm-prime-e2e --no-push --no-rich --sampling.max-tokens 8192 --env.player0.harness.id rlm --env.player0.runtime.type prime --env.player0.runtime.vm true --env.player0.max-turns 8 --env.player0.max-output-tokens 32768 --env.player0.timeout.rollout 600 --env.player1.harness.id rlm --env.player1.runtime.type prime --env.player1.runtime.vm true --env.player1.max-turns 8 --env.player1.max-output-tokens 32768 --env.player1.timeout.rollout 600 + uv run python -c 'import json; from pathlib import Path; rows = [json.loads(line) for line in Path("/tmp/kuhn-rlm-prime-e2e/traces.jsonl").read_text().splitlines() if line.strip()]; assert len(rows) == 3 and all(len(row["traces"]) == 2 for row in rows), rows; healthy = [row for row in rows if row["ok"]]; stopped = [row for row in rows if not row["ok"]]; assert len(healthy) >= 2, rows; assert all(all(t["ok"] and not t["errors"] and t["stop_condition"] == "user_closed" and t["info"]["kuhn"]["forfeited"] is None for t in row["traces"]) for row in healthy), healthy; assert all(any(t["stop_condition"] == "max_turns" and any("rollout stopped: max_turns" in e["message"] for e in t["errors"]) for t in row["traces"]) for row in stopped), stopped; assert all(sorted(t["agent"]["name"] for t in row["traces"]) == ["player0", "player1"] for row in rows); assert all(sum(t["rewards"]["payoff"]["score"] for t in row["traces"]) == 0 for row in healthy); summaries = [{"seed": t["info"]["kuhn"]["seed"], "agent": t["agent"]["name"], "card": t["info"]["kuhn"]["card"], "history": t["info"]["kuhn"]["history"], "turns": sum(bool(n.get("sampled")) for n in t["nodes"]), "stop": t["stop_condition"], "payoff": t["rewards"]["payoff"]["score"]} for row in rows for t in row["traces"]]; assert any(s["turns"] >= 2 and s["stop"] == "user_closed" for s in summaries), summaries; print(json.dumps(summaries, indent=2))' + - name: Run live v1 E2Es run: | uv run pytest tests/v1 -vv -n auto -m "e2e and not prime and not modal" --cov=verifiers --cov-report=xml --cov-report=term From 65d809b8f804d152e940970c5786ef066e21f9c6 Mon Sep 17 00:00:00 2001 From: hallerite Date: Wed, 5 Aug 2026 14:28:02 +0200 Subject: [PATCH 12/17] test(v1): verify nested user-sim tool call --- .github/workflows/test.yml | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 1da5259f7..7c51b3b53 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -112,13 +112,7 @@ jobs: PYTHONPATH: tests/v1/fixtures run: | uv run eval echo-tool-v1 --env.id user-sim -m deepseek/deepseek-v4-flash-0731 -n 1 -r 1 -c 1 -o /tmp/user-sim-rlm-prime-e2e --no-push --no-rich --sampling.max-tokens 8192 --env.taskset.task.tools.colocated true --env.assistant.harness.id rlm --env.assistant.runtime.type prime --env.assistant.runtime.vm true --env.assistant.max-turns 8 --env.assistant.max-output-tokens 32768 --env.assistant.timeout.rollout 600 --env.user.max-turns 8 --env.user.max-output-tokens 32768 --env.user.timeout.rollout 600 - uv run python -c 'import json; from pathlib import Path; rows = [json.loads(line) for line in Path("/tmp/user-sim-rlm-prime-e2e/traces.jsonl").read_text().splitlines() if line.strip()]; assert len(rows) == 1 and rows[0]["ok"], rows; traces = rows[0]["traces"]; assert len(traces) == 2 and all(t["ok"] and not t["errors"] for t in traces), traces; assistant = next(t for t in traces if t["agent"]["name"] == "assistant"); user = next(t for t in traces if t["agent"]["name"] == "user"); assert assistant["rewards"]["echoed"]["score"] == 1.0, assistant; assert assistant["metrics"]["user_turns"] >= 1, assistant; assert any(tool["name"] == "echo_back" for tool in assistant["tools"]), assistant["tools"]; print(json.dumps({"assistant_turns": sum(bool(n.get("sampled")) for n in assistant["nodes"]), "user_turns": sum(bool(n.get("sampled")) for n in user["nodes"]), "echoed_reward": assistant["rewards"]["echoed"]["score"], "assistant_stop": assistant["stop_condition"], "user_stop": user["stop_condition"]}, indent=2))' - - - name: Run Kuhn Poker with RLM in Prime VMs - if: always() - run: | - uv run eval kuhn-poker-v1 -m deepseek/deepseek-v4-flash-0731 -n 3 -r 1 -c 1 -o /tmp/kuhn-rlm-prime-e2e --no-push --no-rich --sampling.max-tokens 8192 --env.player0.harness.id rlm --env.player0.runtime.type prime --env.player0.runtime.vm true --env.player0.max-turns 8 --env.player0.max-output-tokens 32768 --env.player0.timeout.rollout 600 --env.player1.harness.id rlm --env.player1.runtime.type prime --env.player1.runtime.vm true --env.player1.max-turns 8 --env.player1.max-output-tokens 32768 --env.player1.timeout.rollout 600 - uv run python -c 'import json; from pathlib import Path; rows = [json.loads(line) for line in Path("/tmp/kuhn-rlm-prime-e2e/traces.jsonl").read_text().splitlines() if line.strip()]; assert len(rows) == 3 and all(len(row["traces"]) == 2 for row in rows), rows; healthy = [row for row in rows if row["ok"]]; stopped = [row for row in rows if not row["ok"]]; assert len(healthy) >= 2, rows; assert all(all(t["ok"] and not t["errors"] and t["stop_condition"] == "user_closed" and t["info"]["kuhn"]["forfeited"] is None for t in row["traces"]) for row in healthy), healthy; assert all(any(t["stop_condition"] == "max_turns" and any("rollout stopped: max_turns" in e["message"] for e in t["errors"]) for t in row["traces"]) for row in stopped), stopped; assert all(sorted(t["agent"]["name"] for t in row["traces"]) == ["player0", "player1"] for row in rows); assert all(sum(t["rewards"]["payoff"]["score"] for t in row["traces"]) == 0 for row in healthy); summaries = [{"seed": t["info"]["kuhn"]["seed"], "agent": t["agent"]["name"], "card": t["info"]["kuhn"]["card"], "history": t["info"]["kuhn"]["history"], "turns": sum(bool(n.get("sampled")) for n in t["nodes"]), "stop": t["stop_condition"], "payoff": t["rewards"]["payoff"]["score"]} for row in rows for t in row["traces"]]; assert any(s["turns"] >= 2 and s["stop"] == "user_closed" for s in summaries), summaries; print(json.dumps(summaries, indent=2))' + uv run python -c 'import json; from pathlib import Path; rows = [json.loads(line) for line in Path("/tmp/user-sim-rlm-prime-e2e/traces.jsonl").read_text().splitlines() if line.strip()]; assert len(rows) == 1 and rows[0]["ok"], rows; traces = rows[0]["traces"]; assert len(traces) == 2 and all(t["ok"] and not t["errors"] and t["stop_condition"] == "user_closed" for t in traces), traces; assistant = next(t for t in traces if t["agent"]["name"] == "assistant"); user = next(t for t in traces if t["agent"]["name"] == "user"); nodes = json.dumps(assistant["nodes"]); assert assistant["rewards"]["echoed"]["score"] == 1.0, assistant; assert assistant["metrics"]["user_turns"] >= 2, assistant; assert sum(bool(n.get("sampled")) for n in assistant["nodes"]) >= 3, assistant["nodes"]; assert "await echo_back.run" in nodes and "hello world [ok-7f3]" in nodes, nodes; print(json.dumps({"assistant_turns": sum(bool(n.get("sampled")) for n in assistant["nodes"]), "user_turns": sum(bool(n.get("sampled")) for n in user["nodes"]), "echoed_reward": assistant["rewards"]["echoed"]["score"], "assistant_stop": assistant["stop_condition"], "user_stop": user["stop_condition"]}, indent=2))' - name: Run live v1 E2Es run: | From 17338f25381d4ab095fedc52bc8c7323f9e5f814 Mon Sep 17 00:00:00 2001 From: hallerite Date: Wed, 5 Aug 2026 14:32:16 +0200 Subject: [PATCH 13/17] test(v1): remove temporary user-sim check --- .github/workflows/test.yml | 8 -------- 1 file changed, 8 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 7c51b3b53..8dc6ddbaa 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -106,14 +106,6 @@ jobs: - name: Install dependencies run: uv sync - - name: Run modeled user simulation with RLM and MCP - if: always() - env: - PYTHONPATH: tests/v1/fixtures - run: | - uv run eval echo-tool-v1 --env.id user-sim -m deepseek/deepseek-v4-flash-0731 -n 1 -r 1 -c 1 -o /tmp/user-sim-rlm-prime-e2e --no-push --no-rich --sampling.max-tokens 8192 --env.taskset.task.tools.colocated true --env.assistant.harness.id rlm --env.assistant.runtime.type prime --env.assistant.runtime.vm true --env.assistant.max-turns 8 --env.assistant.max-output-tokens 32768 --env.assistant.timeout.rollout 600 --env.user.max-turns 8 --env.user.max-output-tokens 32768 --env.user.timeout.rollout 600 - uv run python -c 'import json; from pathlib import Path; rows = [json.loads(line) for line in Path("/tmp/user-sim-rlm-prime-e2e/traces.jsonl").read_text().splitlines() if line.strip()]; assert len(rows) == 1 and rows[0]["ok"], rows; traces = rows[0]["traces"]; assert len(traces) == 2 and all(t["ok"] and not t["errors"] and t["stop_condition"] == "user_closed" for t in traces), traces; assistant = next(t for t in traces if t["agent"]["name"] == "assistant"); user = next(t for t in traces if t["agent"]["name"] == "user"); nodes = json.dumps(assistant["nodes"]); assert assistant["rewards"]["echoed"]["score"] == 1.0, assistant; assert assistant["metrics"]["user_turns"] >= 2, assistant; assert sum(bool(n.get("sampled")) for n in assistant["nodes"]) >= 3, assistant["nodes"]; assert "await echo_back.run" in nodes and "hello world [ok-7f3]" in nodes, nodes; print(json.dumps({"assistant_turns": sum(bool(n.get("sampled")) for n in assistant["nodes"]), "user_turns": sum(bool(n.get("sampled")) for n in user["nodes"]), "echoed_reward": assistant["rewards"]["echoed"]["score"], "assistant_stop": assistant["stop_condition"], "user_stop": user["stop_condition"]}, indent=2))' - - name: Run live v1 E2Es run: | uv run pytest tests/v1 -vv -n auto -m "e2e and not prime and not modal" --cov=verifiers --cov-report=xml --cov-report=term From af19253d7ca2e957d06578266b8d84f11820df7d Mon Sep 17 00:00:00 2001 From: hallerite Date: Wed, 5 Aug 2026 15:04:48 +0200 Subject: [PATCH 14/17] refactor(v1): clarify ACP and subprocess helpers --- verifiers/v1/acp/runner.py | 8 ++++---- verifiers/v1/runtimes/subprocess.py | 18 +++++++++--------- 2 files changed, 13 insertions(+), 13 deletions(-) diff --git a/verifiers/v1/acp/runner.py b/verifiers/v1/acp/runner.py index 6f6b0baa0..763259f54 100644 --- a/verifiers/v1/acp/runner.py +++ b/verifiers/v1/acp/runner.py @@ -36,7 +36,7 @@ MAX_PACKET_BYTES = 128 * 1024 * 1024 -class VerifiersClient(Client): +class VerifiersACPClient(Client): def __init__(self) -> None: self.visible_reply = "" self.message_id: str | None = None @@ -142,7 +142,7 @@ def segment_messages(config: dict, is_new: bool) -> list[dict]: async def prompt( - client: VerifiersClient, + client: VerifiersACPClient, connection: Any, capabilities: Any, session_id: str, @@ -165,7 +165,7 @@ async def prompt( async def run_once(config: dict) -> str: - client = VerifiersClient() + client = VerifiersACPClient() command = config["command"] async with spawn_agent_process( client, @@ -217,7 +217,7 @@ class LiveACPSession: """One live ACP process, connection, and session shared by several turns.""" def __init__(self) -> None: - self.client = VerifiersClient() + self.client = VerifiersACPClient() self._reset() def _reset(self) -> None: diff --git a/verifiers/v1/runtimes/subprocess.py b/verifiers/v1/runtimes/subprocess.py index 9527916b2..b1237ac25 100644 --- a/verifiers/v1/runtimes/subprocess.py +++ b/verifiers/v1/runtimes/subprocess.py @@ -34,12 +34,12 @@ class SubprocessRuntimeInfo(SubprocessConfig, BaseRuntimeInfo): pass -async def _read_stream(reader: asyncio.StreamReader) -> AsyncIterator[bytes]: +async def read_stream(reader: asyncio.StreamReader) -> AsyncIterator[bytes]: while chunk := await reader.read(64 * 1024): yield chunk -def _signal_process( +def signal_process( process: asyncio.subprocess.Process, signal_: signal.Signals ) -> None: if process.returncode is not None: @@ -57,8 +57,8 @@ def __init__(self, process: asyncio.subprocess.Process) -> 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) + self.stdout = read_stream(process.stdout) + self.stderr = read_stream(process.stderr) async def write(self, data: bytes) -> None: self._stdin.write(data) @@ -68,10 +68,10 @@ async def wait(self) -> int: return await self._process.wait() async def terminate(self) -> None: - _signal_process(self._process, signal.SIGTERM) + signal_process(self._process, signal.SIGTERM) async def kill(self) -> None: - _signal_process(self._process, signal.SIGKILL) + signal_process(self._process, signal.SIGKILL) class SubprocessRuntime(Runtime): @@ -171,7 +171,7 @@ async def teardown(self) -> None: """Stop and reap background servers before their event loop closes.""" background = list(self._background) for proc in background: - _signal_process(proc, signal.SIGTERM) + signal_process(proc, signal.SIGTERM) if background: try: await asyncio.wait_for( @@ -182,7 +182,7 @@ async def teardown(self) -> None: ) except TimeoutError: for proc in background: - _signal_process(proc, signal.SIGKILL) + signal_process(proc, signal.SIGKILL) with contextlib.suppress(TimeoutError): await asyncio.wait_for( asyncio.gather( @@ -197,7 +197,7 @@ async def teardown(self) -> None: def cleanup(self) -> None: for proc in self._background: - _signal_process(proc, signal.SIGTERM) + signal_process(proc, signal.SIGTERM) self._background = [] if self.workdir is not None: shutil.rmtree(self.workdir, ignore_errors=True) From 02db58603223e6f1e54593e288b720b049917109 Mon Sep 17 00:00:00 2001 From: hallerite Date: Wed, 5 Aug 2026 16:09:58 +0200 Subject: [PATCH 15/17] fix(v1): restore ACP empty-reply validation --- verifiers/v1/acp/runner.py | 25 ++++++++++++++++++++++++- 1 file changed, 24 insertions(+), 1 deletion(-) diff --git a/verifiers/v1/acp/runner.py b/verifiers/v1/acp/runner.py index 763259f54..78d8a9a2c 100644 --- a/verifiers/v1/acp/runner.py +++ b/verifiers/v1/acp/runner.py @@ -31,6 +31,8 @@ PermissionOption, RequestPermissionResponse, TextContentBlock, + ToolCall, + ToolCallUpdate, ) MAX_PACKET_BYTES = 128 * 1024 * 1024 @@ -40,12 +42,21 @@ 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 ): @@ -157,10 +168,22 @@ async def prompt( raise ValueError("ACP prompt has no content") client.reset() try: - await connection.prompt(session_id=session_id, prompt=blocks) + 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 From 0ff6f9a0cda93696d7762511212ef9167de7267d Mon Sep 17 00:00:00 2001 From: hallerite Date: Wed, 5 Aug 2026 16:49:06 +0200 Subject: [PATCH 16/17] build: use released prime-sandboxes 0.2.35 --- pyproject.toml | 3 +-- uv.lock | 8 ++++++-- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 686ed79da..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", @@ -119,7 +119,6 @@ url = "https://pypi.org/simple" default = true [tool.uv.sources] -prime-sandboxes = { git = "https://github.com/PrimeIntellect-ai/prime", rev = "892bea7cb3d3f0926d9b42da1ebf206207bb8bab", subdirectory = "packages/prime-sandboxes" } compact = { path = "environments/compact", editable = true } glossary-v1 = { path = "environments/glossary_v1", editable = true } deepwiki-v1 = { path = "environments/deepwiki_v1", editable = true } diff --git a/uv.lock b/uv.lock index 359376f5c..d36ca612f 100644 --- a/uv.lock +++ b/uv.lock @@ -3136,8 +3136,8 @@ toml = [ [[package]] name = "prime-sandboxes" -version = "0.2.34" -source = { git = "https://github.com/PrimeIntellect-ai/prime?subdirectory=packages%2Fprime-sandboxes&rev=892bea7cb3d3f0926d9b42da1ebf206207bb8bab#892bea7cb3d3f0926d9b42da1ebf206207bb8bab" } +version = "0.2.35" +source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "aiofiles" }, { name = "connect-python" }, @@ -3146,6 +3146,10 @@ dependencies = [ { name = "pydantic" }, { name = "tenacity" }, ] +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/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]] name = "prime-tunnel" From da6500013417e960dd4a83a8d74cf7402d619d05 Mon Sep 17 00:00:00 2001 From: hallerite Date: Wed, 5 Aug 2026 17:24:22 +0200 Subject: [PATCH 17/17] chore(v1): require native Prime process support --- verifiers/v1/runtimes/prime.py | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/verifiers/v1/runtimes/prime.py b/verifiers/v1/runtimes/prime.py index e107fb0ff..409e842c7 100644 --- a/verifiers/v1/runtimes/prime.py +++ b/verifiers/v1/runtimes/prime.py @@ -274,14 +274,8 @@ async def open_process( "persistent harness sessions on Prime require a VM sandbox; " "set runtime.prime.vm=true" ) - open_process = getattr(self._client, "open_process", None) - if open_process is None: - raise SandboxError( - "Prime VM live processes require a prime-sandboxes SDK version " - "that provides AsyncSandboxClient.open_process()" - ) try: - process = await open_process( + process = await self._client.open_process( self.info.id, shlex.join(argv), working_dir=self.config.workdir,