feat(v1): run RLM over persistent ACP

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. |

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

verifiers/v1/acp/__init__.py

@@ -1,8 +1,10 @@
 """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.dialects.chat import message_to_wire
 from verifiers.v1.harness import Harness
@@ -11,13 +13,30 @@
 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."""
 
+    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"}
@@ -33,6 +52,7 @@ 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,
     ) -> ProgramResult:
         if prompt is None:
             raise ValueError("ACP requires a prompt")
@@ -51,14 +71,117 @@ 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 sidecar 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 sidecar 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 "sidecar did not become ready"
+                        )
+                        if log.exit_code == 0 and log.stdout:
+                            detail = (
+                                f"{detail}\n\nACP sidecar log:\n{log.stdout.rstrip()}"
+                            )
+                        raise RuntimeError(f"ACP sidecar failed to start: {detail}")
+                elif probe.exit_code != 0:
+                    detail = (
+                        probe.stderr.strip()
+                        or probe.stdout.strip()
+                        or "sidecar did not respond"
+                    )
+                    raise RuntimeError(f"ACP sidecar probe failed: {detail}")
         directory = f".vf-acp-{secrets.token_hex(8)}"
         created = await runtime.run(["mkdir", "-m", "700", directory], {})
         if created.exit_code != 0:
             raise RuntimeError(f"ACP config directory failed: {created.stderr.strip()}")
         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 sidecar 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, *, remove: bool = True
+    ) -> None:
+        """Stop a persistent ACP session, optionally keeping its artifacts."""
+        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 sidecar shutdown failed"
+                    )
+                    if log.exit_code == 0 and log.stdout:
+                        failure = (
+                            f"{failure}\n\nACP sidecar log:\n{log.stdout.rstrip()}"
+                        )
+        finally:
+            if remove:
+                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 sidecar must live in a private subdirectory")
+        return parent