From 822be391bf79fb852ef00a32c74b5700067d6ff6 Mon Sep 17 00:00:00 2001 From: hallerite Date: Tue, 25 Aug 2026 17:29:04 +0200 Subject: [PATCH 1/4] feat(v1): integrate nano-RLM training over ACP --- tests/v1/test_rlm_prime_contract.py | 211 ++++++++++++++++++++++++++ verifiers/v1/harnesses/rlm/harness.py | 191 ++++++++++++++++------- 2 files changed, 351 insertions(+), 51 deletions(-) create mode 100644 tests/v1/test_rlm_prime_contract.py diff --git a/tests/v1/test_rlm_prime_contract.py b/tests/v1/test_rlm_prime_contract.py new file mode 100644 index 0000000000..27abcda64a --- /dev/null +++ b/tests/v1/test_rlm_prime_contract.py @@ -0,0 +1,211 @@ +"""Deterministic RLM training-contract E2E in a Prime VM.""" + +import asyncio +import json +import time +from typing import Any + +import pytest + +from verifiers.v1.harnesses.rlm import RLMHarnessConfig + +CODEWORD = "violet-cascade-731" +TOOL_STAMP = "resume-ok-9d2" +FAKE_API_KEY_VAR = "RLM_CONTRACT_E2E_API_KEY" + + +def _message_text(message: dict[str, Any]) -> str: + content = message.get("content", "") + if isinstance(content, str): + return content + if isinstance(content, list): + return "\n".join( + str(item.get("text", "")) if isinstance(item, dict) else str(item) + for item in content + ) + return str(content) + + +def _branch_reply(branch: Any) -> str: + for node in reversed(branch.nodes): + if node.sampled and node.message.role == "assistant": + return str(node.message.content or "") + return "" + + +def _completion(body: dict[str, Any], sequence: int) -> tuple[dict[str, Any], str]: + messages = body["messages"] + users = [ + _message_text(message) for message in messages if message["role"] == "user" + ] + last_role = messages[-1]["role"] + + if any(text == "child-one" for text in users): + message, label = {"role": "assistant", "content": "CHILD-ONE"}, "child-one" + elif any(text == "child-two" for text in users): + message, label = {"role": "assistant", "content": "CHILD-TWO"}, "child-two" + elif last_role == "tool": + message, label = ( + {"role": "assistant", "content": f"{CODEWORD} [{TOOL_STAMP}]"}, + "root-final", + ) + elif any("Call the `recall` tool" in text for text in users): + code = ( + "import asyncio\n" + "import os\n" + "import subprocess\n" + "assert os.environ.get('TASK_VISIBLE') == 'yes'\n" + "assert os.environ.get('EXPLICIT_TASK') == 'also-yes'\n" + "private = ('RLM_API_KEY', 'SERPER_API_KEY')\n" + "assert all(name not in os.environ for name in private)\n" + "child_env = subprocess.check_output(['env'], text=True)\n" + "assert all(f'{name}=' not in child_env for name in private)\n" + "tool_result, children = await asyncio.gather(\n" + f" resume_recall.run(codeword={CODEWORD!r}),\n" + " asyncio.gather(rlm('child-one'), rlm('child-two')),\n" + ")\n" + "print(tool_result)\n" + "print([child.answer for child in children])" + ) + message, label = ( + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "call-contract-ipython", + "type": "function", + "function": { + "name": "ipython", + "arguments": json.dumps({"code": code, "timeout": 120}), + }, + } + ], + }, + "root-tool", + ) + else: + message, label = {"role": "assistant", "content": "READY"}, "root-ready" + + finish_reason = "tool_calls" if "tool_calls" in message else "stop" + return ( + { + "id": f"chatcmpl-contract-{sequence}", + "object": "chat.completion", + "created": int(time.time()), + "model": body.get("model", "contract-model"), + "choices": [ + { + "index": 0, + "message": message, + "finish_reason": finish_reason, + } + ], + "usage": { + "prompt_tokens": 10, + "completion_tokens": 5, + "total_tokens": 15, + }, + }, + label, + ) + + +@pytest.mark.e2e +@pytest.mark.prime +async def test_rlm_training_contract_in_prime_vm(run_v1, tmp_path, monkeypatch): + calls: list[str] = [] + children_started = 0 + both_children_started = asyncio.Event() + + async def handle( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + nonlocal children_started + try: + head = await reader.readuntil(b"\r\n\r\n") + headers = { + name.strip().lower(): value.strip() + for line in head.decode("latin-1").split("\r\n")[1:] + if line and (name_value := line.split(":", 1)) + for name, value in [name_value] + } + body = json.loads( + await reader.readexactly(int(headers.get("content-length", "0"))) + ) + response, label = _completion(body, len(calls)) + calls.append(label) + if label in {"child-one", "child-two"}: + children_started += 1 + if children_started == 2: + both_children_started.set() + await asyncio.wait_for(both_children_started.wait(), timeout=30) + payload = json.dumps(response, separators=(",", ":")).encode() + writer.write( + b"HTTP/1.1 200 OK\r\n" + b"Content-Type: application/json\r\n" + + f"Content-Length: {len(payload)}\r\n".encode() + + b"Connection: close\r\n\r\n" + + payload + ) + await writer.drain() + finally: + writer.close() + await writer.wait_closed() + + server = await asyncio.start_server(handle, "127.0.0.1", 0) + port = server.sockets[0].getsockname()[1] + monkeypatch.setenv(FAKE_API_KEY_VAR, "contract-secret") + harness = RLMHarnessConfig( + id="rlm", + max_depth=1, + max_concurrent_subagents=2, + max_subagent_calls=2, + kernel_env={"TASK_VISIBLE": "yes", "EXPLICIT_TASK": "also-yes"}, + env={ + "RLM_API_KEY": "ambient-provider-secret", + "SERPER_API_KEY": "search-secret", + }, + ) + + try: + (trace,) = await run_v1( + "echo-acp-resume-v1", + harness=harness, + runtime={"type": "prime", "vm": True}, + env={ + "agent": { + "client": { + "type": "eval", + "base_url": f"http://127.0.0.1:{port}/v1", + "api_key_var": FAKE_API_KEY_VAR, + } + } + }, + output_dir=tmp_path, + max_turns=8, + max_tokens=8192, + rollout_timeout=600, + ) + finally: + server.close() + await server.wait_closed() + + assert trace.ok, trace.errors + assert trace.rewards["resumed"].score == 1.0 + assert trace.primary_reply == f"{CODEWORD} [{TOOL_STAMP}]" + assert trace.num_branches == 3 + assert {_branch_reply(branch) for branch in trace.branches} == { + "CHILD-ONE", + "CHILD-TWO", + f"{CODEWORD} [{TOOL_STAMP}]", + } + assert set(calls) == { + "root-ready", + "root-tool", + "root-final", + "child-one", + "child-two", + } + assert trace.metrics["sub_rlm_num_calls"] == 2 + assert trace.metrics["has_sub_rlm"] == 1 diff --git a/verifiers/v1/harnesses/rlm/harness.py b/verifiers/v1/harnesses/rlm/harness.py index 6be4fb0412..0db2e3fbd3 100644 --- a/verifiers/v1/harnesses/rlm/harness.py +++ b/verifiers/v1/harnesses/rlm/harness.py @@ -1,20 +1,26 @@ """RLM over ACP, with MCP tools exposed as pre-imported IPython skills.""" -import json import logging import random import shlex from typing import Literal -from pydantic import Field, PositiveInt, model_validator +from pydantic import ( + BaseModel, + ConfigDict, + Field, + NonNegativeInt, + PositiveInt, + field_validator, + model_validator, +) -from verifiers.v1.acp import ACPConfig, ACPHarness +from verifiers.v1.acp import ACPCloseResult, ACPConfig, ACPHarness, JsonObject from verifiers.v1.clients import ModelContext from verifiers.v1.configs.harness import HarnessConfig from verifiers.v1.runtimes import Runtime from verifiers.v1.task import TaskData from verifiers.v1.trace import Trace -from verifiers.v1.utils.decorators import metric logger = logging.getLogger(__name__) @@ -25,15 +31,73 @@ RLM_BIN = f"{RLM_DIR}/bin/rlm" SKILLS_DIR = "/task/rlm-skills" RLM_STATE_DIR = ".vf-rlm" +RLM_CONTRACT_METADATA_KEY = "ai.prime.rlm/contract-v1" +RLM_RUNTIME_METADATA_KEY = "ai.prime.rlm/runtime-v1" +RLM_SESSION_METADATA_KEY = "ai.prime.rlm/session-v1" + + +class _ContractModel(BaseModel): + model_config = ConfigDict(extra="forbid", strict=True) + + +class _UsageSnapshot(_ContractModel): + prompt_tokens: int = Field(ge=0) + completion_tokens: int = Field(ge=0) + total_tokens: int = Field(ge=0) + + +class _ProgrammaticToolCallSnapshot(_ContractModel): + python_total: int = Field(ge=0) + bash_total: int = Field(ge=0) + by_tool_python: dict[str, int] + by_tool_bash: dict[str, int] + + +class _SupervisorSnapshot(_ContractModel): + subagent_calls: int = Field(ge=0) + active_subagent_calls: int = Field(ge=0) + + +class _LimitsSnapshot(_ContractModel): + max_depth: int = Field(ge=0) + max_concurrent_subagents: int = Field(gt=0) + max_subagent_calls: int = Field(gt=0) + max_tokens: int | None = Field(default=None, gt=0) + summarize_at_tokens: int | None = Field(default=None, gt=0) + max_compactions: int | None = Field(default=None, gt=0) + max_tool_output_chars: int | None = Field(default=None, gt=0) + allow_git: bool + + +class _SessionSnapshot(_ContractModel): + session_id: str = Field(pattern=r"^[A-Za-z0-9._:-]{1,128}$") + last_stop_reason: str | None + model: str = Field(min_length=1) + turns: int = Field(ge=0) + usage: _UsageSnapshot + metrics: dict[str, int | float] + programmatic_tool_call_stats: _ProgrammaticToolCallSnapshot + supervisor: _SupervisorSnapshot + limits: _LimitsSnapshot class RLMHarnessConfig(HarnessConfig): - version: str = Field( - default="83ef01f7a6c97328919387343bd30cf4edaac20d", min_length=1 - ) + version: str = Field(default="main", min_length=1) """Git ref (branch, tag, or commit) of nano-rlm to install.""" - max_depth: int = 0 - """Recursion depth rlm may spawn sub-harnesses to (RLM_MAX_DEPTH).""" + max_depth: NonNegativeInt = 0 + """Recursion depth RLM may spawn sub-harnesses to.""" + exec_timeout: PositiveInt = 300 + max_output: int = -1 + max_tokens: PositiveInt | None = None + max_compactions: PositiveInt | None = None + max_concurrent_subagents: PositiveInt | None = None + max_subagent_calls: PositiveInt = 64 + max_tool_output_chars: PositiveInt | None = None + allow_git: bool = False + sdk_max_retries: NonNegativeInt = 5 + system_prompt_path: str | None = None + kernel_env: dict[str, str] = Field(default_factory=dict) + """Task variables intentionally visible to model-controlled kernel code.""" builtin_skills: list[BuiltinSkill] = Field(default_factory=list) """Built-in rlm skills to enable (RLM_SKILLS), e.g. `["edit"]`; empty enables none. The tool set is fixed (ipython); the base `skills` field takes SKILL.md paths.""" @@ -52,9 +116,24 @@ def validate_range(self) -> "RLMHarnessConfig": ) return self + @field_validator("max_output") + @classmethod + def validate_max_output(cls, value: int) -> int: + if value == 0 or value < -1: + raise ValueError("must be positive, or -1 to disable truncation") + return value + + @model_validator(mode="after") + def validate_concurrency(self) -> "RLMHarnessConfig": + if ( + self.max_concurrent_subagents is not None + and self.max_concurrent_subagents < self.max_depth + ): + raise ValueError("max_concurrent_subagents must be at least max_depth") + return self + @model_validator(mode="after") def reject_disabled_tools(self) -> "RLMHarnessConfig": - # rlm's only tool is ipython, which must stay enabled, so there's nothing to disable. if self.disabled_tools: raise ValueError( "the rlm harness has a fixed tool set (ipython) and does not support " @@ -91,19 +170,17 @@ async def setup(self, runtime: Runtime) -> None: raise RuntimeError(f"rlm install failed: {result.stderr.strip()[-500:]}") await super().setup(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 — - 0 when unset — so a task's rollouts share one threshold). Always set — "" when disabled — - so the typed field, not a host var the subprocess runtime would inherit, wins.""" + def summarize_threshold(self, task_idx: int | None) -> int | None: + """Resolve a fixed or per-task compaction threshold.""" value = self.config.summarize_at_tokens if value is None: - return "" + return None if isinstance(value, tuple): lo, hi = value - return str(random.Random(task_idx or 0).randint(lo, hi)) - return str(value) + return random.Random(task_idx or 0).randint(lo, hi) + return value - def _env( + def _runtime_metadata( self, ctx: ModelContext, trace: Trace, @@ -111,21 +188,38 @@ def _env( secret: str, data: TaskData, 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": self._home(trace), - "RLM_SUMMARIZE_AT_TOKENS": self.summarize_threshold(data.idx), + ) -> JsonObject: + max_concurrent = self.config.max_concurrent_subagents or max( + 4, self.config.max_depth + ) + payload = { + "session_id": trace.id, + "model": ctx.model, + "provider": { + "base_url": endpoint, + "api_key": secret, + "headers": {}, + "max_retries": self.config.sdk_max_retries, + }, + "policy": { + "max_depth": self.config.max_depth, + "exec_timeout": self.config.exec_timeout, + "max_output": self.config.max_output, + "max_tokens": self.config.max_tokens, + "summarize_at_tokens": self.summarize_threshold(data.idx), + "max_compactions": self.config.max_compactions, + "max_concurrent_subagents": max_concurrent, + "max_subagent_calls": self.config.max_subagent_calls, + "max_tool_output_chars": self.config.max_tool_output_chars, + "allow_git": self.config.allow_git, + }, + "system_prompt_path": self.config.system_prompt_path, + "append_to_system_prompt": system_prompt, + "skills": list(self.config.builtin_skills), + "kernel_env": self.config.kernel_env, + "search_api_key": self.config.resolved_env.get("SERPER_API_KEY"), } - 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) - return env + return {RLM_RUNTIME_METADATA_KEY: payload} async def prepare_acp( self, @@ -139,29 +233,24 @@ async def prepare_acp( ) -> ACPConfig: system_prompt, prompt = self.resolve_prompt(data) return ACPConfig( - env=self._env(ctx, trace, endpoint, secret, data, system_prompt), + env={**self.config.resolved_env, "RLM_HOME": self._home(trace)}, command=[RLM_BIN, "--acp"], prompt=prompt, + session_meta=self._runtime_metadata( + ctx, trace, endpoint, secret, data, system_prompt + ), + required_agent_meta={RLM_CONTRACT_METADATA_KEY: True}, ) - @metric - 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 {} - try: - meta = json.loads(result.stdout) - except json.JSONDecodeError: - return {} - return { - key: float(value) - for key, value in meta.get("metrics", {}).items() - if isinstance(value, (int, float)) and not isinstance(value, bool) - } + def acp_close_metrics( + self, trace: Trace, result: ACPCloseResult + ) -> dict[str, float]: + snapshot = _SessionSnapshot.model_validate( + result.metadata.get(RLM_SESSION_METADATA_KEY) + ) + if snapshot.session_id != trace.id: + raise ValueError("RLM session snapshot does not match the rollout") + return {name: float(value) for name, value in snapshot.metrics.items()} async def cleanup(self, trace: Trace, runtime: Runtime) -> None: await runtime.run(["rm", "-rf", f"{RLM_STATE_DIR}/{trace.id}"], {}) From e159495cce6ea243a11784fba23cb4e3464624b0 Mon Sep 17 00:00:00 2001 From: hallerite Date: Tue, 25 Aug 2026 19:00:52 +0200 Subject: [PATCH 2/4] refactor(v1): read RLM metrics from ACP prompts --- verifiers/v1/harnesses/rlm/harness.py | 58 +++++---------------------- 1 file changed, 10 insertions(+), 48 deletions(-) diff --git a/verifiers/v1/harnesses/rlm/harness.py b/verifiers/v1/harnesses/rlm/harness.py index 0db2e3fbd3..5c1705d6e1 100644 --- a/verifiers/v1/harnesses/rlm/harness.py +++ b/verifiers/v1/harnesses/rlm/harness.py @@ -15,7 +15,7 @@ model_validator, ) -from verifiers.v1.acp import ACPCloseResult, ACPConfig, ACPHarness, JsonObject +from verifiers.v1.acp import ACPConfig, ACPHarness, ACPTurnResult, JsonObject from verifiers.v1.clients import ModelContext from verifiers.v1.configs.harness import HarnessConfig from verifiers.v1.runtimes import Runtime @@ -36,53 +36,17 @@ RLM_SESSION_METADATA_KEY = "ai.prime.rlm/session-v1" -class _ContractModel(BaseModel): - model_config = ConfigDict(extra="forbid", strict=True) +class _SessionSnapshot(BaseModel): + model_config = ConfigDict(extra="ignore", strict=True) - -class _UsageSnapshot(_ContractModel): - prompt_tokens: int = Field(ge=0) - completion_tokens: int = Field(ge=0) - total_tokens: int = Field(ge=0) - - -class _ProgrammaticToolCallSnapshot(_ContractModel): - python_total: int = Field(ge=0) - bash_total: int = Field(ge=0) - by_tool_python: dict[str, int] - by_tool_bash: dict[str, int] - - -class _SupervisorSnapshot(_ContractModel): - subagent_calls: int = Field(ge=0) - active_subagent_calls: int = Field(ge=0) - - -class _LimitsSnapshot(_ContractModel): - max_depth: int = Field(ge=0) - max_concurrent_subagents: int = Field(gt=0) - max_subagent_calls: int = Field(gt=0) - max_tokens: int | None = Field(default=None, gt=0) - summarize_at_tokens: int | None = Field(default=None, gt=0) - max_compactions: int | None = Field(default=None, gt=0) - max_tool_output_chars: int | None = Field(default=None, gt=0) - allow_git: bool - - -class _SessionSnapshot(_ContractModel): session_id: str = Field(pattern=r"^[A-Za-z0-9._:-]{1,128}$") - last_stop_reason: str | None - model: str = Field(min_length=1) - turns: int = Field(ge=0) - usage: _UsageSnapshot metrics: dict[str, int | float] - programmatic_tool_call_stats: _ProgrammaticToolCallSnapshot - supervisor: _SupervisorSnapshot - limits: _LimitsSnapshot class RLMHarnessConfig(HarnessConfig): - version: str = Field(default="main", min_length=1) + version: str = Field( + default="11dcb9c353f1f7c89c8c6f1bc0ddea33b3cffa19", min_length=1 + ) """Git ref (branch, tag, or commit) of nano-rlm to install.""" max_depth: NonNegativeInt = 0 """Recursion depth RLM may spawn sub-harnesses to.""" @@ -239,18 +203,16 @@ async def prepare_acp( session_meta=self._runtime_metadata( ctx, trace, endpoint, secret, data, system_prompt ), - required_agent_meta={RLM_CONTRACT_METADATA_KEY: True}, + required_agent_meta=(RLM_CONTRACT_METADATA_KEY,), ) - def acp_close_metrics( - self, trace: Trace, result: ACPCloseResult - ) -> dict[str, float]: + def acp_turn_result(self, trace: Trace, result: ACPTurnResult) -> None: snapshot = _SessionSnapshot.model_validate( - result.metadata.get(RLM_SESSION_METADATA_KEY) + result.response_metadata.get(RLM_SESSION_METADATA_KEY) ) if snapshot.session_id != trace.id: raise ValueError("RLM session snapshot does not match the rollout") - return {name: float(value) for name, value in snapshot.metrics.items()} + trace.record_metrics(snapshot.metrics) async def cleanup(self, trace: Trace, runtime: Runtime) -> None: await runtime.run(["rm", "-rf", f"{RLM_STATE_DIR}/{trace.id}"], {}) From 8108f630e41423e2550cab3a3b70cc3abbe68469 Mon Sep 17 00:00:00 2001 From: hallerite Date: Tue, 25 Aug 2026 21:59:05 +0200 Subject: [PATCH 3/4] refactor(v1): narrow RLM ACP configuration --- tests/v1/test_rlm_prime_contract.py | 211 -------------------------- verifiers/v1/harnesses/rlm/harness.py | 63 +------- 2 files changed, 8 insertions(+), 266 deletions(-) delete mode 100644 tests/v1/test_rlm_prime_contract.py diff --git a/tests/v1/test_rlm_prime_contract.py b/tests/v1/test_rlm_prime_contract.py deleted file mode 100644 index 27abcda64a..0000000000 --- a/tests/v1/test_rlm_prime_contract.py +++ /dev/null @@ -1,211 +0,0 @@ -"""Deterministic RLM training-contract E2E in a Prime VM.""" - -import asyncio -import json -import time -from typing import Any - -import pytest - -from verifiers.v1.harnesses.rlm import RLMHarnessConfig - -CODEWORD = "violet-cascade-731" -TOOL_STAMP = "resume-ok-9d2" -FAKE_API_KEY_VAR = "RLM_CONTRACT_E2E_API_KEY" - - -def _message_text(message: dict[str, Any]) -> str: - content = message.get("content", "") - if isinstance(content, str): - return content - if isinstance(content, list): - return "\n".join( - str(item.get("text", "")) if isinstance(item, dict) else str(item) - for item in content - ) - return str(content) - - -def _branch_reply(branch: Any) -> str: - for node in reversed(branch.nodes): - if node.sampled and node.message.role == "assistant": - return str(node.message.content or "") - return "" - - -def _completion(body: dict[str, Any], sequence: int) -> tuple[dict[str, Any], str]: - messages = body["messages"] - users = [ - _message_text(message) for message in messages if message["role"] == "user" - ] - last_role = messages[-1]["role"] - - if any(text == "child-one" for text in users): - message, label = {"role": "assistant", "content": "CHILD-ONE"}, "child-one" - elif any(text == "child-two" for text in users): - message, label = {"role": "assistant", "content": "CHILD-TWO"}, "child-two" - elif last_role == "tool": - message, label = ( - {"role": "assistant", "content": f"{CODEWORD} [{TOOL_STAMP}]"}, - "root-final", - ) - elif any("Call the `recall` tool" in text for text in users): - code = ( - "import asyncio\n" - "import os\n" - "import subprocess\n" - "assert os.environ.get('TASK_VISIBLE') == 'yes'\n" - "assert os.environ.get('EXPLICIT_TASK') == 'also-yes'\n" - "private = ('RLM_API_KEY', 'SERPER_API_KEY')\n" - "assert all(name not in os.environ for name in private)\n" - "child_env = subprocess.check_output(['env'], text=True)\n" - "assert all(f'{name}=' not in child_env for name in private)\n" - "tool_result, children = await asyncio.gather(\n" - f" resume_recall.run(codeword={CODEWORD!r}),\n" - " asyncio.gather(rlm('child-one'), rlm('child-two')),\n" - ")\n" - "print(tool_result)\n" - "print([child.answer for child in children])" - ) - message, label = ( - { - "role": "assistant", - "content": None, - "tool_calls": [ - { - "id": "call-contract-ipython", - "type": "function", - "function": { - "name": "ipython", - "arguments": json.dumps({"code": code, "timeout": 120}), - }, - } - ], - }, - "root-tool", - ) - else: - message, label = {"role": "assistant", "content": "READY"}, "root-ready" - - finish_reason = "tool_calls" if "tool_calls" in message else "stop" - return ( - { - "id": f"chatcmpl-contract-{sequence}", - "object": "chat.completion", - "created": int(time.time()), - "model": body.get("model", "contract-model"), - "choices": [ - { - "index": 0, - "message": message, - "finish_reason": finish_reason, - } - ], - "usage": { - "prompt_tokens": 10, - "completion_tokens": 5, - "total_tokens": 15, - }, - }, - label, - ) - - -@pytest.mark.e2e -@pytest.mark.prime -async def test_rlm_training_contract_in_prime_vm(run_v1, tmp_path, monkeypatch): - calls: list[str] = [] - children_started = 0 - both_children_started = asyncio.Event() - - async def handle( - reader: asyncio.StreamReader, writer: asyncio.StreamWriter - ) -> None: - nonlocal children_started - try: - head = await reader.readuntil(b"\r\n\r\n") - headers = { - name.strip().lower(): value.strip() - for line in head.decode("latin-1").split("\r\n")[1:] - if line and (name_value := line.split(":", 1)) - for name, value in [name_value] - } - body = json.loads( - await reader.readexactly(int(headers.get("content-length", "0"))) - ) - response, label = _completion(body, len(calls)) - calls.append(label) - if label in {"child-one", "child-two"}: - children_started += 1 - if children_started == 2: - both_children_started.set() - await asyncio.wait_for(both_children_started.wait(), timeout=30) - payload = json.dumps(response, separators=(",", ":")).encode() - writer.write( - b"HTTP/1.1 200 OK\r\n" - b"Content-Type: application/json\r\n" - + f"Content-Length: {len(payload)}\r\n".encode() - + b"Connection: close\r\n\r\n" - + payload - ) - await writer.drain() - finally: - writer.close() - await writer.wait_closed() - - server = await asyncio.start_server(handle, "127.0.0.1", 0) - port = server.sockets[0].getsockname()[1] - monkeypatch.setenv(FAKE_API_KEY_VAR, "contract-secret") - harness = RLMHarnessConfig( - id="rlm", - max_depth=1, - max_concurrent_subagents=2, - max_subagent_calls=2, - kernel_env={"TASK_VISIBLE": "yes", "EXPLICIT_TASK": "also-yes"}, - env={ - "RLM_API_KEY": "ambient-provider-secret", - "SERPER_API_KEY": "search-secret", - }, - ) - - try: - (trace,) = await run_v1( - "echo-acp-resume-v1", - harness=harness, - runtime={"type": "prime", "vm": True}, - env={ - "agent": { - "client": { - "type": "eval", - "base_url": f"http://127.0.0.1:{port}/v1", - "api_key_var": FAKE_API_KEY_VAR, - } - } - }, - output_dir=tmp_path, - max_turns=8, - max_tokens=8192, - rollout_timeout=600, - ) - finally: - server.close() - await server.wait_closed() - - assert trace.ok, trace.errors - assert trace.rewards["resumed"].score == 1.0 - assert trace.primary_reply == f"{CODEWORD} [{TOOL_STAMP}]" - assert trace.num_branches == 3 - assert {_branch_reply(branch) for branch in trace.branches} == { - "CHILD-ONE", - "CHILD-TWO", - f"{CODEWORD} [{TOOL_STAMP}]", - } - assert set(calls) == { - "root-ready", - "root-tool", - "root-final", - "child-one", - "child-two", - } - assert trace.metrics["sub_rlm_num_calls"] == 2 - assert trace.metrics["has_sub_rlm"] == 1 diff --git a/verifiers/v1/harnesses/rlm/harness.py b/verifiers/v1/harnesses/rlm/harness.py index 5c1705d6e1..13ba3b2c7e 100644 --- a/verifiers/v1/harnesses/rlm/harness.py +++ b/verifiers/v1/harnesses/rlm/harness.py @@ -5,15 +5,7 @@ import shlex from typing import Literal -from pydantic import ( - BaseModel, - ConfigDict, - Field, - NonNegativeInt, - PositiveInt, - field_validator, - model_validator, -) +from pydantic import BaseModel, ConfigDict, Field, PositiveInt, model_validator from verifiers.v1.acp import ACPConfig, ACPHarness, ACPTurnResult, JsonObject from verifiers.v1.clients import ModelContext @@ -45,23 +37,11 @@ class _SessionSnapshot(BaseModel): class RLMHarnessConfig(HarnessConfig): version: str = Field( - default="11dcb9c353f1f7c89c8c6f1bc0ddea33b3cffa19", min_length=1 + default="5ee1c34024a183bbbd3a38a6129995f5b982631d", min_length=1 ) """Git ref (branch, tag, or commit) of nano-rlm to install.""" - max_depth: NonNegativeInt = 0 + max_depth: int = 0 """Recursion depth RLM may spawn sub-harnesses to.""" - exec_timeout: PositiveInt = 300 - max_output: int = -1 - max_tokens: PositiveInt | None = None - max_compactions: PositiveInt | None = None - max_concurrent_subagents: PositiveInt | None = None - max_subagent_calls: PositiveInt = 64 - max_tool_output_chars: PositiveInt | None = None - allow_git: bool = False - sdk_max_retries: NonNegativeInt = 5 - system_prompt_path: str | None = None - kernel_env: dict[str, str] = Field(default_factory=dict) - """Task variables intentionally visible to model-controlled kernel code.""" builtin_skills: list[BuiltinSkill] = Field(default_factory=list) """Built-in rlm skills to enable (RLM_SKILLS), e.g. `["edit"]`; empty enables none. The tool set is fixed (ipython); the base `skills` field takes SKILL.md paths.""" @@ -80,22 +60,6 @@ def validate_range(self) -> "RLMHarnessConfig": ) return self - @field_validator("max_output") - @classmethod - def validate_max_output(cls, value: int) -> int: - if value == 0 or value < -1: - raise ValueError("must be positive, or -1 to disable truncation") - return value - - @model_validator(mode="after") - def validate_concurrency(self) -> "RLMHarnessConfig": - if ( - self.max_concurrent_subagents is not None - and self.max_concurrent_subagents < self.max_depth - ): - raise ValueError("max_concurrent_subagents must be at least max_depth") - return self - @model_validator(mode="after") def reject_disabled_tools(self) -> "RLMHarnessConfig": if self.disabled_tools: @@ -148,39 +112,28 @@ def _runtime_metadata( self, ctx: ModelContext, trace: Trace, + runtime: Runtime, endpoint: str, secret: str, data: TaskData, system_prompt: str | None, ) -> JsonObject: - max_concurrent = self.config.max_concurrent_subagents or max( - 4, self.config.max_depth - ) payload = { "session_id": trace.id, "model": ctx.model, "provider": { "base_url": endpoint, "api_key": secret, - "headers": {}, - "max_retries": self.config.sdk_max_retries, }, "policy": { "max_depth": self.config.max_depth, - "exec_timeout": self.config.exec_timeout, - "max_output": self.config.max_output, - "max_tokens": self.config.max_tokens, "summarize_at_tokens": self.summarize_threshold(data.idx), - "max_compactions": self.config.max_compactions, - "max_concurrent_subagents": max_concurrent, - "max_subagent_calls": self.config.max_subagent_calls, - "max_tool_output_chars": self.config.max_tool_output_chars, - "allow_git": self.config.allow_git, + "max_concurrent_subagents": max(4, self.config.max_depth), }, - "system_prompt_path": self.config.system_prompt_path, + "system_prompt_path": None, "append_to_system_prompt": system_prompt, "skills": list(self.config.builtin_skills), - "kernel_env": self.config.kernel_env, + "kernel_env": runtime.env, "search_api_key": self.config.resolved_env.get("SERPER_API_KEY"), } return {RLM_RUNTIME_METADATA_KEY: payload} @@ -201,7 +154,7 @@ async def prepare_acp( command=[RLM_BIN, "--acp"], prompt=prompt, session_meta=self._runtime_metadata( - ctx, trace, endpoint, secret, data, system_prompt + ctx, trace, runtime, endpoint, secret, data, system_prompt ), required_agent_meta=(RLM_CONTRACT_METADATA_KEY,), ) From a15fcc6bcf5982d77b8bbc200f220a8c0b2dd94e Mon Sep 17 00:00:00 2001 From: hallerite Date: Tue, 25 Aug 2026 22:19:41 +0200 Subject: [PATCH 4/4] refactor(v1): trust pinned RLM ACP contract --- verifiers/v1/harnesses/rlm/harness.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/verifiers/v1/harnesses/rlm/harness.py b/verifiers/v1/harnesses/rlm/harness.py index 13ba3b2c7e..83712dbaf1 100644 --- a/verifiers/v1/harnesses/rlm/harness.py +++ b/verifiers/v1/harnesses/rlm/harness.py @@ -23,7 +23,6 @@ RLM_BIN = f"{RLM_DIR}/bin/rlm" SKILLS_DIR = "/task/rlm-skills" RLM_STATE_DIR = ".vf-rlm" -RLM_CONTRACT_METADATA_KEY = "ai.prime.rlm/contract-v1" RLM_RUNTIME_METADATA_KEY = "ai.prime.rlm/runtime-v1" RLM_SESSION_METADATA_KEY = "ai.prime.rlm/session-v1" @@ -156,7 +155,6 @@ async def prepare_acp( session_meta=self._runtime_metadata( ctx, trace, runtime, endpoint, secret, data, system_prompt ), - required_agent_meta=(RLM_CONTRACT_METADATA_KEY,), ) def acp_turn_result(self, trace: Trace, result: ACPTurnResult) -> None: