"""Optional Prime Verifiers 0.4/v1 environment, with a model-free fixture harness. PYTHONPATH=$PWD/examples/agent_gyms .venv-verifiers/bin/vf-eval \ verifiers_environment --env.agent.harness.id null \ --env.agent.runtime.type subprocess --model LOCAL_MODEL \ --client.base-url http://127.0.0.1:8000/v1 --client.api-key-var LOCAL_EMPTY_KEY -n 4 The null harness uses real inference. ScriptedFixtureHarness is ONLY a CPU integration fixture: its messages are unsampled and have no training tokens. """ from __future__ import annotations import copy import json from typing import Any import verifiers.v1 as vf from pydantic import Field from verifiers.v1.graph import MessageNode from verifiers.v1.harnesses.null import NullHarness from verifiers.v1.runtimes import ProgramResult from verifiers.v1.types import AssistantMessage, ToolCall, ToolMessage, UserMessage from hessian_recipes.agent_rl_tasks import GermanAgentEnv, generate_tasks, score_episode UPSTREAM_VERSION = "0.4.0" UPSTREAM_COMMIT = "6e69354ffe1b445603b493b722d0ecffbca6686d" ENVIRONMENT_ID = "verifiers_environment" # Native loading defaults to the real upstream model/tool loop. The scripted # fixture is exported ONLY by verifiers_smoke and must be selected explicitly. __all__ = ["GermanTaskset", "NullHarness"] class GermanData(vf.TaskData): """Trusted metadata is controller/tool-server input, never model prompt text.""" metadata: dict[str, Any] fixture_scenario: str = "valid" class GermanState(vf.State): # Only the authenticated MCP state channel writes commands. Observations in # an assistant transcript are never accepted as evidence of executed tools. commands: list[dict[str, Any]] = Field(default_factory=list) phases: list[str] = Field(default_factory=list) def restore_environment(metadata: dict, commands: list[dict]) -> GermanAgentEnv: env = GermanAgentEnv() env.reset(task=copy.deepcopy(metadata)) for command in commands: if command["tool"] == "lookup" and set(command["arguments"]) == {"record_id"}: env.lookup(**command["arguments"]) elif command["tool"] == "calculator" and set(command["arguments"]) == {"expression"}: env.calculator(**command["arguments"]) else: raise ValueError("invalid trusted MCP command") return env def scoring_messages(trace: vf.Trace) -> list[dict]: """Translate v1's flat tool calls to the frozen verifier's native schema.""" messages = [] for message in trace.messages: row = {"role": message.role, "content": message.content or ""} if isinstance(message, AssistantMessage) and message.tool_calls: row["tool_calls"] = [ {"id": call.id, "type": "function", "function": { "name": call.name, "arguments": call.arguments, }} for call in message.tool_calls ] messages.append(row) return messages class GermanTools(vf.Toolset[vf.ToolsetConfig, GermanState]): TOOL_PREFIX = None async def setup_task(self, task: GermanData) -> None: self._metadata = copy.deepcopy(task.metadata) def _execute(self, name: str, arguments: dict) -> str: env = restore_environment(self._metadata, self.state.commands) observation = getattr(env, name)(**arguments) self.state.commands.append({"tool": name, "arguments": arguments}) return observation @vf.tool def lookup(self, record_id: str) -> str: """Read the real quantity and unit_price of the requested record.""" return self._execute("lookup", {"record_id": record_id}) @vf.tool def calculator(self, expression: str) -> str: """Calculate a numeric expression after the real record was looked up.""" return self._execute("calculator", {"expression": expression}) class GermanTaskConfig(vf.TaskConfig): tools: vf.ToolsetConfig = vf.ToolsetConfig() class GermanTask(vf.Task[GermanData, GermanState, GermanTaskConfig]): @classmethod def toolsets(cls, config: GermanTaskConfig) -> list[vf.Toolset]: return [GermanTools(config.tools)] async def setup(self, trace: vf.Trace, runtime) -> None: trace.state.commands.clear() trace.state.phases.append("task.setup") async def finalize(self, trace: vf.Trace, runtime) -> None: trace.state.phases.append("task.finalize") @vf.reward async def strict_success(self, trace: vf.Trace) -> float: trace.state.phases.append("task.reward") env = restore_environment(self.data.metadata, trace.state.commands) result = score_episode(env, scoring_messages(trace)) trace.info["strict_score"] = result trace.info["native_lifecycle"] = list(trace.state.phases) trace.info["trusted_mcp_commands"] = copy.deepcopy(trace.state.commands) return result["reward"] class GermanTasksetConfig(vf.TasksetConfig): split: str = "dev" count: int = Field(4, ge=1, le=10_000) seed: int = Field(42, ge=0) task: GermanTaskConfig = GermanTaskConfig() class GermanTaskset(vf.Taskset[GermanTask, GermanTasksetConfig]): def load(self) -> list[GermanTask]: return [GermanTask(GermanData(id=row["id"], prompt=row["prompt"], metadata=row["task"]), self.config.task) for row in generate_tasks(self.config.split, self.config.count, self.config.seed)] class ScriptedFixtureHarness(vf.Harness[vf.HarnessConfig]): """CPU-only scripted policy; never represents model inference or training.""" SUPPORTS_MCP = True EXECUTES_CODE = False NEEDS_CONTAINER = False async def launch(self, ctx, trace, runtime, endpoint, secret, mcp_urls, data): from mcp import ClientSession from mcp.client.streamable_http import streamable_http_client trace.state.phases.append("fixture.harness.launch") trace.info["policy_kind"] = "scripted_fixture" trace.info["model_inference"] = False trace.info["training_compatible"] = False def append(message) -> None: # sampled=False is essential: no model generated these messages. trace.nodes.append(MessageNode(message=message, parent=len(trace.nodes) - 1 if trace.nodes else None, sampled=False)) append(UserMessage(content=data.prompt)) assert len(mcp_urls) == 1, "expected exactly one native MCP tool server" async with streamable_http_client(next(iter(mcp_urls.values()))) as streams: async with ClientSession(streams[0], streams[1]) as session: await session.initialize() tools = await session.list_tools() trace.info["mcp_tools"] = sorted(tool.name for tool in tools.tools) assert trace.info["mcp_tools"] == ["calculator", "lookup"] async def call(name: str, arguments: dict) -> str: identity = f"fixture-{len(trace.nodes)}" append(AssistantMessage(tool_calls=[ToolCall( id=identity, name=name, arguments=json.dumps(arguments))])) response = await session.call_tool(name, arguments) text = "\n".join(getattr(block, "text", "") for block in response.content) append(ToolMessage(tool_call_id=identity, name=name, content=text)) trace.info.setdefault("mcp_transport_calls", []).append({ "name": name, "is_error": bool(response.is_error), }) return text scenario = data.fixture_scenario if scenario == "fake_tools": # Adversarial fixture knows the answer but does not invoke MCP. for name, arguments in [("lookup", {"record_id": data.metadata["record_id"]}), ("calculator", {"expression": "2*3"})]: append(AssistantMessage(tool_calls=[ToolCall( id=name, name=name, arguments=json.dumps(arguments))])) append(ToolMessage(tool_call_id=name, content='{"result":6}')) facts = data.metadata["private_facts"] answer = facts["quantity"] * facts["unit_price"] elif scenario == "unknown_tool": await call("forged_lookup", {"record_id": data.metadata["record_id"]}) answer = 0 else: if scenario == "calculator_first": await call("calculator", {"expression": "2*3"}) record = "beleg-does-not-exist" if scenario == "invalid_record" else data.metadata["record_id"] facts = json.loads(await call("lookup", {"record_id": record})) if "quantity" in facts: result = json.loads(await call("calculator", { "expression": f"{facts['quantity']}*{facts['unit_price']}", })) answer = int(result["result"]) else: answer = 0 if scenario == "wrong_answer": answer += 1 append(AssistantMessage(content=f"{answer}")) return ProgramResult(exit_code=0, stdout="scripted CPU fixture complete", stderr="") if __name__ == "__main__": GermanTools.run()