From 7147ff0335233f7a4ffae0eedb9b51260bfad221 Mon Sep 17 00:00:00 2001 From: hallerite Date: Sun, 26 Jul 2026 15:11:02 -0700 Subject: [PATCH 1/5] feat(v1): add rollout-scoped harness sessions --- .../references/REFERENCE.md | 7 +- tests/v1/test_e2e.py | 3 + verifiers/v1/acp/__init__.py | 239 ++++++++++- verifiers/v1/acp/_runner.py | 397 ++++++++++++++++-- verifiers/v1/harness.py | 126 +++++- verifiers/v1/harnesses/rlm/harness.py | 113 +++-- verifiers/v1/rollout.py | 58 ++- 7 files changed, 818 insertions(+), 125 deletions(-) diff --git a/skills/evaluate-environments/references/REFERENCE.md b/skills/evaluate-environments/references/REFERENCE.md index 8c4572791..bbb67f122 100644 --- a/skills/evaluate-environments/references/REFERENCE.md +++ b/skills/evaluate-environments/references/REFERENCE.md @@ -286,11 +286,14 @@ Installs the Codex CLI into the runtime and runs `codex exec`. #### `RLMHarnessConfig` — `id: "rlm"` -Installs the rlm CLI and runs it. Knobs map onto `RLM_*` env vars; base `HarnessConfig.env` passes any other `RLM_*` var through verbatim. +Installs rlm-harness and runs its ACP agent. MCP tools become pre-imported +IPython skills while the model-facing tool surface remains `ipython`. Knobs map +onto `RLM_*` env vars; base `HarnessConfig.env` passes any other `RLM_*` var +through verbatim. | Field | Type | Default | Notes | | --- | --- | --- | --- | -| `version` | `str` | `"main"` | Git ref (branch/tag/commit) of rlm to install. | +| `version` | `str` | `"56218f33796ecbe465445bc43948886354fde196"` | Git ref (branch/tag/commit) of rlm-harness to install. | | `max_depth` | `int` | `0` | Recursion depth rlm may spawn sub-harnesses to (`RLM_MAX_DEPTH`). | | `skills` | `list["edit" \| "search"]` | `[]` | Built-in rlm skills to enable (`RLM_SKILLS`). Empty enables none. | | `summarize_at_tokens` | `int \| (int, int) \| None` | `None` | Auto-compaction threshold (`RLM_SUMMARIZE_AT_TOKENS`): compact once context grows past this many tokens. An int is fixed; a `(lo, hi)` pair draws a per-group threshold (seeded by task index). `None` disables. Ints must be positive. | diff --git a/tests/v1/test_e2e.py b/tests/v1/test_e2e.py index 0c77546ae..98d396290 100644 --- a/tests/v1/test_e2e.py +++ b/tests/v1/test_e2e.py @@ -54,6 +54,7 @@ def _pair(a: str, b: str, id: str, *extra_marks): # retain MCP access after resuming. Cover every harness in the local container runtime, # plus one remote placement for the sandbox/tunnel boundary. ACP_RESUME_PLACEMENTS = [ + _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"), @@ -205,6 +206,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 0c17b7e21..487d8e38d 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, @@ -33,6 +88,29 @@ async def run( mcp_urls: dict[str, str] | None = None, system_prompt: str | None = None, session_path: str | None = None, + ) -> 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, + ) + + 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, ) -> ProgramResult: if prompt is None: raise ValueError("ACP requires a prompt") @@ -51,6 +129,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: @@ -58,7 +183,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 5b794e610..bde9904da 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 @@ -30,12 +32,19 @@ TextContentBlock, ) +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 + def reset(self) -> None: + self.visible_reply = "" + self.message_id = None + async def session_update(self, session_id: str, update: Any, **kwargs: Any) -> None: if not isinstance(update, AgentMessageChunk) or not isinstance( update.content, TextContentBlock @@ -105,7 +114,62 @@ 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 + if not client.visible_reply.strip(): + raise RuntimeError( + "ACP agent produced no visible reply " + f"(stop_reason={response.stop_reason!r})" + ) + return client.visible_reply + + +async def run_once(config: dict) -> str: client = VerifiersClient() command = config["command"] async with spawn_agent_process( @@ -120,72 +184,315 @@ 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 - try: - 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 - if not client.visible_reply.strip(): - raise RuntimeError("ACP agent produced no visible reply") - sys.stdout.write(client.visible_reply) + reply = await prompt( + client, + connection, + capabilities, + session_id, + config, + is_new=is_new, + ) if session_path and is_new: session_path.parent.mkdir(parents=True, exist_ok=True) session_path.write_text(session_id) + return reply -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: + try: + await self.stack.aclose() + except BaseException: + pass + 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: + 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 8f8eefe69..6b9d83d1d 100644 --- a/verifiers/v1/harness.py +++ b/verifiers/v1/harness.py @@ -7,10 +7,10 @@ from typing import TYPE_CHECKING, ClassVar, Generic, TypeVar +from verifiers.v1.configs.harness import HarnessConfig from verifiers.v1.clients import ModelContext from verifiers.v1.decorators import discover_decorated, invoke_all from verifiers.v1.errors import HarnessError, boundary -from verifiers.v1.configs.harness import HarnessConfig from verifiers.v1.runtimes import ProgramResult, Runtime from verifiers.v1.task import TaskData from verifiers.v1.types import Messages @@ -123,27 +123,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 "" - 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 @@ -229,3 +237,77 @@ 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 "" + 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 3884dd356..0696c107b 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,31 +8,31 @@ from pydantic import model_validator -from verifiers.v1.configs.harness import HarnessConfig -from verifiers.v1.harness import Harness +from verifiers.v1.acp import ACP from verifiers.v1.clients import ModelContext +from verifiers.v1.configs.harness import HarnessConfig from verifiers.v1.decorators import metric -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.trace import Trace from verifiers.v1.task import TaskData +from verifiers.v1.trace import Trace logger = logging.getLogger(__name__) 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] = [] @@ -93,10 +93,10 @@ 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} - result = await runtime.run(["sh", "-c", guarded], env) + result = await runtime.run(["sh", "-c", guarded], self.config.resolved_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 — @@ -110,47 +110,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 {} @@ -163,3 +197,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 3df863234..709750275 100644 --- a/verifiers/v1/rollout.py +++ b/verifiers/v1/rollout.py @@ -1,12 +1,12 @@ """A rollout: one trajectory — drive a harness segment by segment and score its trace. -A rollout's exchange is a sequence of SEGMENTS: the harness program runs until it -yields (= exits), the run's user answers its final message, and the next segment -resumes the exchange with that answer (`Harness.resume` — a relaunch on the accreted -conversation by default, a native continuation for harnesses with their own session -state). The user loop lives between segments, at the exchange's natural turn -granularity — never inside the model boundary, so a harness's own tool loop can -never race or amputate it. +A rollout's exchange is a sequence of SEGMENTS: its rollout-scoped `HarnessSession` +runs until it yields, the run's user answers its final message, and the next segment +continues the same handle with that answer. The default handle adapts `Harness.resume` +by relaunching on the accreted conversation; stateful transports can keep a live +process and native session behind the same interface. The user loop lives between +segments, at the exchange's natural turn granularity — never inside the model +boundary, so a harness's own tool loop can never race or amputate it. `RolloutRun` is the engine, a staged lifecycle: `open()` boots the world, each `step()` runs one segment, `close()` finalizes, scores, and tears the world down — @@ -26,7 +26,7 @@ from verifiers import __version__ from verifiers.v1.configs.agent import AgentConfig -from verifiers.v1.harness import Harness +from verifiers.v1.harness import Harness, HarnessSession from verifiers.v1.clients import ModelContext from verifiers.v1.decorators import discover_decorated, invoke from verifiers.v1.dialects import parse_message @@ -164,6 +164,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 @@ -289,6 +290,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: self.fail(e) return False @@ -325,16 +336,8 @@ async def step(self, messages: Messages | None = None) -> bool: # A timeout still scores the partial trajectory. 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: # Only the rollout deadline reads as a clean truncation; a TimeoutError # from the harness's own I/O with no expired deadline is a failure — @@ -372,6 +375,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() if self.runtime is not None: @@ -392,6 +398,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: @@ -422,6 +439,11 @@ async def close(self) -> Trace: except Exception as e: 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 a188af8a936895a1d2f87f1bd8b943c981fd30b0 Mon Sep 17 00:00:00 2001 From: hallerite Date: Mon, 27 Jul 2026 02:43:55 +0200 Subject: [PATCH 2/5] fix(v1): harden harness session lifecycle --- verifiers/v1/acp/__init__.py | 8 ++++++++ verifiers/v1/acp/_runner.py | 39 +++++++++++++++++++++++++++--------- verifiers/v1/harness.py | 12 ++++++++++- 3 files changed, 49 insertions(+), 10 deletions(-) diff --git a/verifiers/v1/acp/__init__.py b/verifiers/v1/acp/__init__.py index 487d8e38d..5a41fa201 100644 --- a/verifiers/v1/acp/__init__.py +++ b/verifiers/v1/acp/__init__.py @@ -111,6 +111,7 @@ async def _run( system_prompt: str | None = None, session_path: str | None = None, sidecar_path: str | None = None, + allow_sidecar_start: bool = False, ) -> ProgramResult: if prompt is None: raise ValueError("ACP requires a prompt") @@ -136,6 +137,11 @@ async def _run( async with self._sidecar_lock(runtime, sidecar_path): probe = await runtime.run([*program, "probe", sidecar_path], {}) if probe.exit_code == PROBE_UNAVAILABLE_EXIT_CODE: + if not allow_sidecar_start: + raise RuntimeError( + "ACP session disappeared between turns; refusing to " + "restart without its conversation and process state" + ) removed = await runtime.run(["rm", "-f", sidecar_path], {}) if removed.exit_code != 0: raise RuntimeError( @@ -274,6 +280,7 @@ def __init__( self._started = False async def _run(self, messages: Messages | None) -> ProgramResult: + first_turn = not self._started self._started = True return await self.acp._run( self.runtime, @@ -283,6 +290,7 @@ async def _run(self, messages: Messages | None) -> ProgramResult: mcp_urls=self.mcp_urls, system_prompt=self.system_prompt, sidecar_path=self.sidecar_path, + allow_sidecar_start=first_turn, ) async def close(self) -> None: diff --git a/verifiers/v1/acp/_runner.py b/verifiers/v1/acp/_runner.py index bde9904da..9785fe9fe 100644 --- a/verifiers/v1/acp/_runner.py +++ b/verifiers/v1/acp/_runner.py @@ -161,7 +161,7 @@ async def 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 - if not client.visible_reply.strip(): + if not client.visible_reply.strip() and response.stop_reason != "end_turn": raise RuntimeError( "ACP agent produced no visible reply " f"(stop_reason={response.stop_reason!r})" @@ -326,12 +326,15 @@ async def serve_session(socket_path: str) -> None: shutdown = asyncio.Event() active_prompt: asyncio.Task[str] | None = None - async def run_prompt(config: dict) -> str: + async def run_prompt(config: dict, owns_session: asyncio.Event) -> 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)) + # Set before the first await while holding the lock: a disconnect + # handler can now distinguish this lock owner from a queued request. + owns_session.set() try: return await active_prompt finally: @@ -362,7 +365,10 @@ async def handle( await stop_session() response = {"ok": True} elif operation == "prompt": - prompt_task = asyncio.create_task(run_prompt(request["config"])) + owns_session = asyncio.Event() + prompt_task = asyncio.create_task( + run_prompt(request["config"], owns_session) + ) disconnect_task = asyncio.create_task(reader.read()) done, _ = await asyncio.wait( (prompt_task, disconnect_task), @@ -373,12 +379,19 @@ async def handle( 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) + if owns_session.is_set(): + # The disconnected request owns the active ACP prompt. Stop + # it and its agent so neither consumes tokens unattended. + await stop_session() + # `stop_session` awaits the tracked ACP prompt; now let its + # lock-owning wrapper finish and clear bookkeeping. + await asyncio.gather(prompt_task, return_exceptions=True) + else: + # This request was still queued behind another client. It + # has no session state to tear down, so cancel only its + # wrapper and leave the active owner untouched. + prompt_task.cancel() + await asyncio.gather(prompt_task, return_exceptions=True) return else: raise ValueError(f"unknown ACP session operation: {operation!r}") @@ -491,6 +504,14 @@ async def main() -> None: if str(error) == "timed out waiting for ACP session": raise SystemExit(PROBE_UNAVAILABLE_EXIT_CODE) from None raise + except TimeoutError as error: + # The socket accepted our connection but did not answer. It may still + # own a live agent, so unlinking it and starting another would orphan + # that process and violate single-session ownership. + raise RuntimeError( + "ACP session accepted the probe but did not respond; refusing " + "to restart a potentially live session" + ) from error else: raise ValueError(f"unknown ACP runner operation: {operation!r}") diff --git a/verifiers/v1/harness.py b/verifiers/v1/harness.py index 6b9d83d1d..c6a01ebc3 100644 --- a/verifiers/v1/harness.py +++ b/verifiers/v1/harness.py @@ -131,7 +131,17 @@ async def run( try: await session.turn(messages) finally: - await session.close() + try: + await session.close() + except Exception: + # Generation already completed or raised. Teardown must neither + # turn a successful compatibility run into a failure nor replace + # the generation error the caller needs to see. + logger.warning( + "harness session close failed (harness %s)", + self.config.id, + exc_info=True, + ) async def open_session( self, From 3bbda6ce4c087a47dcac3b97a8fb4df45e925c5d Mon Sep 17 00:00:00 2001 From: hallerite Date: Mon, 27 Jul 2026 02:59:35 +0200 Subject: [PATCH 3/5] fix(v1): keep session cleanup retryable --- verifiers/v1/acp/__init__.py | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/verifiers/v1/acp/__init__.py b/verifiers/v1/acp/__init__.py index 5a41fa201..656e1aea4 100644 --- a/verifiers/v1/acp/__init__.py +++ b/verifiers/v1/acp/__init__.py @@ -296,8 +296,9 @@ async def _run(self, messages: Messages | None) -> ProgramResult: 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() + if self._started: + await self.acp._close(self.runtime, self.sidecar_path) + # A failed transport teardown remains retryable. RolloutRun deliberately + # calls close again from its final cleanup path; only a successful teardown + # may make that retry an idempotent no-op. + await super().close() From 4e5526566abadb5791ba12bad06ae82be369bf4d Mon Sep 17 00:00:00 2001 From: hallerite Date: Mon, 27 Jul 2026 03:27:42 +0200 Subject: [PATCH 4/5] fix(v1): make session teardown cancellation-safe --- verifiers/v1/acp/__init__.py | 43 ++++++++++++++++++------------------ verifiers/v1/rollout.py | 13 +++++++---- 2 files changed, 30 insertions(+), 26 deletions(-) diff --git a/verifiers/v1/acp/__init__.py b/verifiers/v1/acp/__init__.py index 656e1aea4..b5c664d6f 100644 --- a/verifiers/v1/acp/__init__.py +++ b/verifiers/v1/acp/__init__.py @@ -217,31 +217,30 @@ async def _close( ) -> 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: + if exists.exit_code != 0: await run_shielded(runtime.run(["rm", "-rf", sidecar_dir], {})) - if failure: + return + + 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()}" + # Preserve the socket and its private directory so the rollout's + # final cleanup pass can retry shutdown instead of orphaning a live + # sidecar that is no longer addressable. raise RuntimeError(failure) + await run_shielded(runtime.run(["rm", "-rf", sidecar_dir], {})) + @staticmethod def _sidecar_dir(sidecar_path: str) -> str: path = PurePosixPath(sidecar_path) diff --git a/verifiers/v1/rollout.py b/verifiers/v1/rollout.py index 709750275..34f792ca6 100644 --- a/verifiers/v1/rollout.py +++ b/verifiers/v1/rollout.py @@ -54,6 +54,7 @@ from verifiers.v1.task import Task, TaskData from verifiers.v1.trace import AgentInfo, Trace, TraceTask, VersionInfo from verifiers.v1.types import Messages +from verifiers.v1.utils.aio import run_shielded from verifiers.v1.utils.version import verifiers_commit logger = logging.getLogger(__name__) @@ -375,16 +376,20 @@ 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 + await run_shielded(self._abort_cleanup()) + + async def _abort_cleanup(self) -> None: + """Complete best-effort teardown even while the driver is being cancelled.""" if self._harness_session is not None: - with contextlib.suppress(Exception): + with contextlib.suppress(asyncio.CancelledError, Exception): await self._harness_session.close() - with contextlib.suppress(Exception): + with contextlib.suppress(asyncio.CancelledError, Exception): await self._stack.aclose() if self.runtime is not None: - with contextlib.suppress(Exception): + with contextlib.suppress(asyncio.CancelledError, Exception): await self.harness.cleanup(self.trace, self.runtime) if self._owns_runtime and self.runtime is not None: - with contextlib.suppress(Exception): + with contextlib.suppress(asyncio.CancelledError, Exception): await self.runtime.stop() async def close(self) -> Trace: From b8e442cdcc20c167c55648eed22111bbfd95dcd8 Mon Sep 17 00:00:00 2001 From: hallerite Date: Mon, 27 Jul 2026 03:40:06 +0200 Subject: [PATCH 5/5] fix(v1): preserve ACP ownership during cancellation --- verifiers/v1/acp/_runner.py | 34 +++++++++++++++++++++++++++++++--- 1 file changed, 31 insertions(+), 3 deletions(-) diff --git a/verifiers/v1/acp/_runner.py b/verifiers/v1/acp/_runner.py index 9785fe9fe..917be6430 100644 --- a/verifiers/v1/acp/_runner.py +++ b/verifiers/v1/acp/_runner.py @@ -36,6 +36,22 @@ PROBE_UNAVAILABLE_EXIT_CODE = 75 +async def run_shielded(coro: Any) -> Any: + """Await cleanup to completion before propagating caller cancellation.""" + task = asyncio.ensure_future(coro) + cancelled: asyncio.CancelledError | None = None + while not task.done(): + try: + await asyncio.shield(task) + except asyncio.CancelledError as error: + cancelled = error + except BaseException: + pass + if cancelled is not None: + raise cancelled from (None if task.cancelled() else task.exception()) + return task.result() + + class VerifiersClient(Client): def __init__(self) -> None: self.visible_reply = "" @@ -290,17 +306,29 @@ async def run(self, config: dict) -> str: return reply async def close(self) -> None: - try: + completed = False + + async def close_owned_resources() -> None: + nonlocal completed 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): + # A protocol-level close is courteous, but losing it must not + # prevent the process-owning exit stack from being closed. + with suppress(asyncio.CancelledError, Exception): await self.connection.close_session(session_id=self.session_id) await self.stack.aclose() + completed = True + + try: + await run_shielded(close_owned_resources()) finally: - self._reset() + # `run_shielded` delays caller cancellation until owned resources are + # closed. Only then is it safe to discard the stack that owns them. + if completed: + self._reset() async def read_packet(reader: asyncio.StreamReader) -> dict: