diff --git a/.gitignore b/.gitignore index e82b1c5f21..2cb74bc86b 100644 --- a/.gitignore +++ b/.gitignore @@ -14,7 +14,7 @@ env/ .python-version # artifacts -core.* +core.[0-9]* .coverage dist/ build/ diff --git a/verifiers/v1/harnesses/bash/harness.py b/verifiers/v1/harnesses/bash/harness.py index 39248d5ca3..733f039a46 100644 --- a/verifiers/v1/harnesses/bash/harness.py +++ b/verifiers/v1/harnesses/bash/harness.py @@ -6,8 +6,10 @@ from verifiers.v1.clients import ModelContext from verifiers.v1.configs.harness import HarnessConfig from verifiers.v1.harness import Harness -from verifiers.v1.harnesses.minimal import PROGRAM_SOURCE -from verifiers.v1.harnesses.standalone import launch_chat_program +from verifiers.v1.harnesses.utils.launch import ( + CHAT_PROGRAM_SOURCE, + launch_chat_program, +) from verifiers.v1.runtimes import ProgramResult, Runtime from verifiers.v1.task import TaskData from verifiers.v1.trace import Trace @@ -57,7 +59,7 @@ class BashHarness(Harness[BashHarnessConfig]): NEEDS_CONTAINER = False async def setup(self, runtime: Runtime) -> None: - await runtime.prepare_uv_script(PROGRAM_SOURCE, self.config.resolved_env) + await runtime.prepare_uv_script(CHAT_PROGRAM_SOURCE, self.config.resolved_env) async def launch( self, @@ -109,7 +111,7 @@ async def launch( ) args += ["--search", f"--serper-key={serper_key}"] return await launch_chat_program( - PROGRAM_SOURCE, + CHAT_PROGRAM_SOURCE, self.config, ctx, trace, diff --git a/verifiers/v1/harnesses/browser_use/harness.py b/verifiers/v1/harnesses/browser_use/harness.py index 8e5e48bd2b..8247087b2b 100644 --- a/verifiers/v1/harnesses/browser_use/harness.py +++ b/verifiers/v1/harnesses/browser_use/harness.py @@ -6,13 +6,14 @@ from verifiers.v1.clients import ModelContext from verifiers.v1.configs.harness import HarnessConfig from verifiers.v1.harness import Harness -from verifiers.v1.harnesses.standalone import inline_mcp_client, launch_chat_program +from verifiers.v1.harnesses.utils import mcp +from verifiers.v1.harnesses.utils.launch import bundle_program, launch_chat_program from verifiers.v1.runtimes import ProgramResult, Runtime from verifiers.v1.task import TaskData from verifiers.v1.trace import Trace -PROGRAM_SOURCE = inline_mcp_client( - (Path(__file__).resolve().parent / "program.py").read_text() +PROGRAM_SOURCE = bundle_program( + (Path(__file__).resolve().parent / "program.py").read_text(), mcp ) # The helper names and persistence rules the model needs to use the local tool. diff --git a/verifiers/v1/harnesses/browser_use/program.py b/verifiers/v1/harnesses/browser_use/program.py index 4c536b2d3f..6d7d6d9b93 100644 --- a/verifiers/v1/harnesses/browser_use/program.py +++ b/verifiers/v1/harnesses/browser_use/program.py @@ -35,7 +35,7 @@ if TYPE_CHECKING: # The harness bundles this module into the generated script before execution. - from verifiers.v1.mcp.client import call_mcp, connect_mcp # noqa: TC004 + from verifiers.v1.harnesses.utils.mcp import call_mcp, connect_mcp # noqa: TC004 BROWSER_TOOL_TIMEOUT = 3600 """Matches the bash harness's command timeout.""" diff --git a/verifiers/v1/harnesses/minimal/__init__.py b/verifiers/v1/harnesses/minimal/__init__.py deleted file mode 100644 index 86c84607ef..0000000000 --- a/verifiers/v1/harnesses/minimal/__init__.py +++ /dev/null @@ -1,9 +0,0 @@ -from pathlib import Path - -from verifiers.v1.harnesses.standalone import inline_mcp_client - -PROGRAM_SOURCE = inline_mcp_client( - (Path(__file__).resolve().parent / "program.py").read_text() -) - -__all__ = ["PROGRAM_SOURCE"] diff --git a/verifiers/v1/harnesses/null/harness.py b/verifiers/v1/harnesses/null/harness.py index 2c812ef676..ed4f2794b6 100644 --- a/verifiers/v1/harnesses/null/harness.py +++ b/verifiers/v1/harnesses/null/harness.py @@ -1,8 +1,10 @@ from verifiers.v1.clients import ModelContext from verifiers.v1.configs.harness import HarnessConfig from verifiers.v1.harness import Harness -from verifiers.v1.harnesses.minimal import PROGRAM_SOURCE -from verifiers.v1.harnesses.standalone import launch_chat_program +from verifiers.v1.harnesses.utils.launch import ( + CHAT_PROGRAM_SOURCE, + launch_chat_program, +) from verifiers.v1.runtimes import ProgramResult, Runtime from verifiers.v1.task import TaskData from verifiers.v1.trace import Trace @@ -20,7 +22,7 @@ class NullHarness(Harness[NullHarnessConfig]): NEEDS_CONTAINER = False async def setup(self, runtime: Runtime) -> None: - await runtime.prepare_uv_script(PROGRAM_SOURCE, self.config.resolved_env) + await runtime.prepare_uv_script(CHAT_PROGRAM_SOURCE, self.config.resolved_env) async def launch( self, @@ -34,7 +36,7 @@ async def launch( ) -> ProgramResult: system_prompt, prompt = self.resolve_prompt(data) return await launch_chat_program( - PROGRAM_SOURCE, + CHAT_PROGRAM_SOURCE, self.config, ctx, trace, diff --git a/verifiers/v1/harnesses/utils/__init__.py b/verifiers/v1/harnesses/utils/__init__.py new file mode 100644 index 0000000000..af099ed654 --- /dev/null +++ b/verifiers/v1/harnesses/utils/__init__.py @@ -0,0 +1,7 @@ +"""Shared building blocks for harness implementations. + +`mcp`, `compaction`, and `core` are written to run inside bundled PEP 723 +programs: `launch.bundle_program` splices their sources into a program script, +so they import only the packages a program declares and reference each other +through `TYPE_CHECKING` imports that the flat bundle resolves at runtime. +""" diff --git a/verifiers/v1/harnesses/utils/compaction.py b/verifiers/v1/harnesses/utils/compaction.py new file mode 100644 index 0000000000..c352c68473 --- /dev/null +++ b/verifiers/v1/harnesses/utils/compaction.py @@ -0,0 +1,256 @@ +"""Tool-output bounding and context compaction for bundled chat programs.""" + +from typing import TYPE_CHECKING + +from openai import APIError, APIStatusError, AsyncOpenAI + +if TYPE_CHECKING: + # The harness bundles this module into the generated script before execution. + from verifiers.v1.harnesses.utils.core import chat # noqa: TC004 + +RESERVE_TOKENS = 16_384 +"""Compact when this many tokens remain below the model context window.""" + +COMPACTION_ATTEMPTS = 3 +"""Checkpoint attempts before compaction fails: a rejected request falls back to the +last good snapshot; an empty or tool-calling reply is resampled.""" + +TOOL_OUTPUT_MAX_BYTES = 20_000 +"""Middle-out truncation budget for one tool result before it enters the conversation.""" + +CHECKPOINT_COMPACTION_PROMPT = """You are performing a CONTEXT CHECKPOINT COMPACTION. Create a handoff summary for another LLM that will resume the task. + +Include: +- Current progress and key decisions made +- Important context, constraints, or user preferences +- What remains to be done (clear next steps) +- Any critical data, examples, or references needed to continue + +Be concise, structured, and focused on helping the next LLM seamlessly continue the work. + +Reply with the summary as plain text. Do not call any tools - summarize from the conversation as it stands.""" + +POST_COMPACTION_FRAMING = """Another language model started to solve this problem and produced \ +a summary of its thinking process. You also have access to the state of the tools that \ +were used by that language model. Use this to build on the work \ +that has already been done and avoid duplicating work. Here is \ +the summary produced by the other language model, use the \ +information in this summary to assist with your own analysis:""" + +CONTEXT_OVERFLOW_MARKERS = ( + # OpenAI error code "context_length_exceeded"; OpenRouter relays the raw body. + "context_length_exceeded", + # OpenAI Responses/Completions: "Your input exceeds the context window of this model". + "exceeds the context window", + # OpenAI chat: "Input tokens exceed the configured limit of N tokens. Please reduce + # the length of the messages."; Groq words it the same way. + "reduce the length of the messages", + # vLLM: "This model's maximum context length is N tokens"; the renderers pre-flight: + # "Prompt length (N) exceeds maximum context length (M)"; Mistral uses the same words. + "maximum context length", + # Anthropic: "prompt is too long: N tokens > M maximum". + "prompt is too long", + # Anthropic byte-size overflow: HTTP 413 {"type": "request_too_large"}. + "request_too_large", + # HTTP proxies reject an oversized body with 413 "Request Entity Too Large". + "request entity too large", + # Google: "The input token count (N) exceeds the maximum number of tokens allowed (M)". + "exceeds the maximum number of tokens", + # xAI: "This model's maximum prompt length is N but the request contains M tokens". + "maximum prompt length is", +) + +CONTEXT_WINDOW_FIELDS = ( + "max_model_len", + "context_length", + "context_window", + "max_context_length", +) + + +class CompactionFailed(Exception): + """Every checkpoint attempt failed - the caller ends the run cleanly instead.""" + + +def is_context_overflow(error: APIStatusError) -> bool: + details = f"{error} {error.body or ''}" + # An overflow is deterministic: a 400, or a 413 for a byte-size cap. + return error.status_code in (400, 413) and any( + marker in details.casefold() for marker in CONTEXT_OVERFLOW_MARKERS + ) + + +def default_threshold(context_window: int) -> int: + """Leave a fixed reserve below the window; small windows keep at least half.""" + return max(context_window - RESERVE_TOKENS, context_window // 2) + + +async def discover_threshold(client: AsyncOpenAI, model: str) -> int | None: + """The compaction threshold, when the provider's model card advertises a context window. + + `models.list()` keeps provider extensions in each card's `model_extra`; a raw + `cast_to` parse breaks on one Python version or another.""" + try: + page = await client.models.list() + except APIError: + return None + for card in page.data: + if card.id != model: + continue + extra = card.model_extra or {} + for field in CONTEXT_WINDOW_FIELDS: + value = extra.get(field) + if isinstance(value, int) and not isinstance(value, bool) and value > 0: + return default_threshold(value) + break + return None + + +def compactable(messages: list[dict]) -> bool: + """Whether compaction can reclaim anything - some history beyond the task exists.""" + first_user = next( + (i for i, m in enumerate(messages) if m.get("role") == "user"), None + ) + return any( + m.get("role") != "system" and i != first_user for i, m in enumerate(messages) + ) + + +def bound_tool_message(message: dict) -> dict: + """Bound a tool message before it enters the conversation - rewrites included.""" + content = message.get("content") + if isinstance(content, str): + return {**message, "content": truncate_tool_output(content)} + # Multimodal results come back as content-part lists; only plain text is truncated. + return message + + +def truncate_tool_output(text: str) -> str: + """Keep the head and tail of an oversized tool result and say what was cut.""" + data = text.encode("utf-8") + if len(data) <= TOOL_OUTPUT_MAX_BYTES: + return text + keep = TOOL_OUTPUT_MAX_BYTES // 2 + head = data[:keep].decode("utf-8", errors="ignore") + tail = data[-keep:].decode("utf-8", errors="ignore") + return ( + f"Warning: truncated output (original token count: {estimated_tokens(text)})\n" + f"Total output lines: {text.count(chr(10)) + 1}\n\n" + f"{head}\n[... {len(data) - 2 * keep} bytes truncated ...]\n{tail}" + ) + + +def estimated_tokens(chars: str) -> int: + """Rough token count at four characters per token.""" + return (len(chars) + 3) // 4 + + +def context_tokens(completion) -> int: + usage = completion.usage + if usage is None: + return 0 + return (usage.prompt_tokens or 0) + (usage.completion_tokens or 0) + + +class Compactor: + """Compact once and retry once when a model turn exhausts its context.""" + + def __init__(self, client, model, tools, enabled, threshold): + self.client = client + self.model = model + self.tools = tools + self.enabled = enabled + self.threshold = threshold + self.compacted = False + self.last_good = 0 + """Message count of the newest state that passed a threshold check - by + definition a state with a full reserve of room, so a checkpoint over it fits.""" + + def reached(self, completion, extra_tokens: int = 0) -> bool: + return ( + self.enabled + and self.threshold is not None + and context_tokens(completion) + extra_tokens >= self.threshold + ) + + def note_good(self, messages: list[dict]) -> None: + self.last_good = len(messages) + + async def complete(self, messages: list[dict]): + try: + completion = await chat(self.client, self.model, messages, self.tools) + except APIStatusError as error: + if ( + not self.enabled + or self.threshold is None + or not is_context_overflow(error) + ): + raise + if not compactable(messages): + if self.compacted: + # The conversation is already a compaction floor and still + # overflows - out of moves, end cleanly. + raise CompactionFailed( + "the compacted conversation still overflows" + ) from error + raise + else: + choice = completion.choices[0] + if not self.reached(completion): + # Usage-verified: this exact prompt was accepted with a full + # reserve of room, so it is a safe checkpoint fallback. + self.note_good(messages) + return completion, messages + if choice.finish_reason != "length" or not compactable(messages): + return completion, messages + + messages = await self.compact(messages) + try: + completion = await chat(self.client, self.model, messages, self.tools) + except APIStatusError as error: + # The rebuilt conversation is sized to fit, so this is out of moves. + if is_context_overflow(error): + raise CompactionFailed( + "the rebuilt conversation still overflows" + ) from error + raise + return completion, messages + + async def compact(self, messages: list[dict]) -> list[dict]: + # A rejected checkpoint falls back to the last good snapshot (which has a + # full reserve of room, so it fits); an empty or tool-calling reply is + # resampled. Reasoning is never part of the summary. + system = [message for message in messages if message.get("role") == "system"] + base = messages + for _ in range(COMPACTION_ATTEMPTS): + checkpoint = [ + *base, + {"role": "user", "content": CHECKPOINT_COMPACTION_PROMPT}, + ] + try: + completion = await chat( + self.client, + self.model, + checkpoint, + self.tools, + tool_choice="none", + ) + except APIStatusError as error: + if not is_context_overflow(error): + raise + base = messages[: self.last_good] + continue + message = completion.choices[0].message + # Reasoning never enters the summary: only the reply's final text + # counts, so a reply that lives entirely in the reasoning channel + # is resampled like an empty one. + text = (message.content or "").strip() + if not message.tool_calls and text: + framed = POST_COMPACTION_FRAMING + "\n\n" + text + rebuilt = [*system, {"role": "user", "content": framed}] + self.note_good(rebuilt) + self.compacted = True + return rebuilt + raise CompactionFailed( + f"no usable summary after {COMPACTION_ATTEMPTS} attempts" + ) diff --git a/verifiers/v1/harnesses/minimal/program.py b/verifiers/v1/harnesses/utils/core.py similarity index 56% rename from verifiers/v1/harnesses/minimal/program.py rename to verifiers/v1/harnesses/utils/core.py index 6535458f9a..9f1d9fb902 100644 --- a/verifiers/v1/harnesses/minimal/program.py +++ b/verifiers/v1/harnesses/utils/core.py @@ -1,8 +1,4 @@ -# /// script -# requires-python = ">=3.10" -# dependencies = ["openai", "mcp==2.0.0", "httpx", "httpx2", "tenacity"] -# /// -"""Shared Null/Bash chat program; secrets use argv so tools do not inherit them.""" +"""Chat loop, local tools, and interception hook for bundled chat programs.""" import argparse import asyncio @@ -12,119 +8,22 @@ from typing import TYPE_CHECKING import httpx -from openai import APIError, APIStatusError, AsyncOpenAI +from openai import APIStatusError, AsyncOpenAI if TYPE_CHECKING: # The harness bundles this module into the generated script before execution. - from verifiers.v1.mcp.client import call_mcp, connect_mcp # noqa: TC004 - -SERPER_URL = "https://google.serper.dev/search" - -CONTEXT_OVERFLOW_MARKERS = ( - "context_length_exceeded", - "exceeds the context window", - "reduce the length of the messages", - "maximum context length", - "prompt is too long", - "request_too_large", - "request entity too large", - "exceeds the maximum number of tokens", - "maximum prompt length is", -) - - -def is_context_overflow(error: APIStatusError) -> bool: - details = f"{error} {error.body or ''}".casefold() - return error.status_code in (400, 413) and any( - marker in details for marker in CONTEXT_OVERFLOW_MARKERS + from verifiers.v1.harnesses.utils.compaction import ( # noqa: TC004 + CompactionFailed, + Compactor, + bound_tool_message, + compactable, + discover_threshold, + estimated_tokens, + is_context_overflow, ) + from verifiers.v1.harnesses.utils.mcp import call_mcp, connect_mcp # noqa: TC004 - -RESERVE_TOKENS = 16_384 -"""Compact when this many tokens remain below the model context window.""" - -COMPACTION_ATTEMPTS = 3 -"""Checkpoint attempts before compaction fails: a rejected request falls back to the -last good snapshot; an empty or tool-calling reply is resampled.""" - -TOOL_OUTPUT_MAX_BYTES = 20_000 -"""Middle-out truncation budget for one tool result before it enters the conversation.""" - -CHECKPOINT_COMPACTION_PROMPT = """You are performing a CONTEXT CHECKPOINT COMPACTION. Create a handoff summary for another LLM that will resume the task. - -Include: -- Current progress and key decisions made -- Important context, constraints, or user preferences -- What remains to be done (clear next steps) -- Any critical data, examples, or references needed to continue - -Be concise, structured, and focused on helping the next LLM seamlessly continue the work. - -Reply with the summary as plain text. Do not call any tools - summarize from the conversation as it stands.""" - -POST_COMPACTION_FRAMING = """Another language model started to solve this problem and produced \ -a summary of its thinking process. You also have access to the state of the tools that \ -were used by that language model. Use this to build on the work \ -that has already been done and avoid duplicating work. Here is \ -the summary produced by the other language model, use the \ -information in this summary to assist with your own analysis:""" - -CONTEXT_OVERFLOW_MARKERS = ( - # OpenAI error code "context_length_exceeded"; OpenRouter relays the raw body. - "context_length_exceeded", - # OpenAI Responses/Completions: "Your input exceeds the context window of this model". - "exceeds the context window", - # OpenAI chat: "Input tokens exceed the configured limit of N tokens. Please reduce - # the length of the messages."; Groq words it the same way. - "reduce the length of the messages", - # vLLM: "This model's maximum context length is N tokens"; the renderers pre-flight: - # "Prompt length (N) exceeds maximum context length (M)"; Mistral uses the same words. - "maximum context length", - # Anthropic: "prompt is too long: N tokens > M maximum". - "prompt is too long", - # Anthropic byte-size overflow: HTTP 413 {"type": "request_too_large"}. - "request_too_large", - # HTTP proxies reject an oversized body with 413 "Request Entity Too Large". - "request entity too large", - # Google: "The input token count (N) exceeds the maximum number of tokens allowed (M)". - "exceeds the maximum number of tokens", - # xAI: "This model's maximum prompt length is N but the request contains M tokens". - "maximum prompt length is", -) - -CONTEXT_WINDOW_FIELDS = ( - "max_model_len", - "context_length", - "context_window", - "max_context_length", -) - - -def default_threshold(context_window: int) -> int: - """Leave a fixed reserve below the window; small windows keep at least half.""" - return max(context_window - RESERVE_TOKENS, context_window // 2) - - -async def discover_threshold(client: AsyncOpenAI, model: str) -> int | None: - """The compaction threshold, when the provider's model card advertises a context window. - - `models.list()` keeps provider extensions in each card's `model_extra`; a raw - `cast_to` parse breaks on one Python version or another.""" - try: - page = await client.models.list() - except APIError: - return None - for card in page.data: - if card.id != model: - continue - extra = card.model_extra or {} - for field in CONTEXT_WINDOW_FIELDS: - value = extra.get(field) - if isinstance(value, int) and not isinstance(value, bool) and value > 0: - return default_threshold(value) - break - return None - +SERPER_URL = "https://google.serper.dev/search" BASH_TOOL = { "type": "function", @@ -166,7 +65,6 @@ async def discover_threshold(client: AsyncOpenAI, model: str) -> int | None: }, } - SEARCH_TOOL = { "type": "function", "function": { @@ -296,160 +194,6 @@ async def chat( return await client.chat.completions.create(**kwargs) -class CompactionFailed(Exception): - """Every checkpoint attempt failed - the caller ends the run cleanly instead.""" - - -def compactable(messages: list[dict]) -> bool: - """Whether compaction can reclaim anything - some history beyond the task exists.""" - first_user = next( - (i for i, m in enumerate(messages) if m.get("role") == "user"), None - ) - return any( - m.get("role") != "system" and i != first_user for i, m in enumerate(messages) - ) - - -def bound_tool_message(message: dict) -> dict: - """Bound a tool message before it enters the conversation - rewrites included.""" - content = message.get("content") - if isinstance(content, str): - return {**message, "content": truncate_tool_output(content)} - # Multimodal results come back as content-part lists; only plain text is truncated. - return message - - -def truncate_tool_output(text: str) -> str: - """Keep the head and tail of an oversized tool result and say what was cut.""" - data = text.encode("utf-8") - if len(data) <= TOOL_OUTPUT_MAX_BYTES: - return text - keep = TOOL_OUTPUT_MAX_BYTES // 2 - head = data[:keep].decode("utf-8", errors="ignore") - tail = data[-keep:].decode("utf-8", errors="ignore") - return ( - f"Warning: truncated output (original token count: {estimated_tokens(text)})\n" - f"Total output lines: {text.count(chr(10)) + 1}\n\n" - f"{head}\n[... {len(data) - 2 * keep} bytes truncated ...]\n{tail}" - ) - - -def estimated_tokens(chars: str) -> int: - """Rough token count at four characters per token.""" - return (len(chars) + 3) // 4 - - -def context_tokens(completion) -> int: - usage = completion.usage - if usage is None: - return 0 - return (usage.prompt_tokens or 0) + (usage.completion_tokens or 0) - - -class Compactor: - """Compact once and retry once when a model turn exhausts its context.""" - - def __init__(self, client, model, tools, enabled, threshold): - self.client = client - self.model = model - self.tools = tools - self.enabled = enabled - self.threshold = threshold - self.compacted = False - self.last_good = 0 - """Message count of the newest state that passed a threshold check - by - definition a state with a full reserve of room, so a checkpoint over it fits.""" - - def reached(self, completion, extra_tokens: int = 0) -> bool: - return ( - self.enabled - and self.threshold is not None - and context_tokens(completion) + extra_tokens >= self.threshold - ) - - def note_good(self, messages: list[dict]) -> None: - self.last_good = len(messages) - - async def complete(self, messages: list[dict]): - try: - completion = await chat(self.client, self.model, messages, self.tools) - except APIStatusError as error: - if ( - not self.enabled - or self.threshold is None - or not is_context_overflow(error) - ): - raise - if not compactable(messages): - if self.compacted: - # The conversation is already a compaction floor and still - # overflows - out of moves, end cleanly. - raise CompactionFailed( - "the compacted conversation still overflows" - ) from error - raise - else: - choice = completion.choices[0] - if not self.reached(completion): - # Usage-verified: this exact prompt was accepted with a full - # reserve of room, so it is a safe checkpoint fallback. - self.note_good(messages) - return completion, messages - if choice.finish_reason != "length" or not compactable(messages): - return completion, messages - - messages = await self.compact(messages) - try: - completion = await chat(self.client, self.model, messages, self.tools) - except APIStatusError as error: - # The rebuilt conversation is sized to fit, so this is out of moves. - if is_context_overflow(error): - raise CompactionFailed( - "the rebuilt conversation still overflows" - ) from error - raise - return completion, messages - - async def compact(self, messages: list[dict]) -> list[dict]: - # A rejected checkpoint falls back to the last good snapshot (which has a - # full reserve of room, so it fits); an empty or tool-calling reply is - # resampled. Reasoning is never part of the summary. - system = [message for message in messages if message.get("role") == "system"] - base = messages - for _ in range(COMPACTION_ATTEMPTS): - checkpoint = [ - *base, - {"role": "user", "content": CHECKPOINT_COMPACTION_PROMPT}, - ] - try: - completion = await chat( - self.client, - self.model, - checkpoint, - self.tools, - tool_choice="none", - ) - except APIStatusError as error: - if not is_context_overflow(error): - raise - base = messages[: self.last_good] - continue - message = completion.choices[0].message - # Reasoning never enters the summary: only the reply's final text - # counts, so a reply that lives entirely in the reasoning channel - # is resampled like an empty one. - text = (message.content or "").strip() - if not message.tool_calls and text: - framed = POST_COMPACTION_FRAMING + "\n\n" + text - rebuilt = [*system, {"role": "user", "content": framed}] - self.note_good(rebuilt) - self.compacted = True - return rebuilt - raise CompactionFailed( - f"no usable summary after {COMPACTION_ATTEMPTS} attempts" - ) - - async def run_tool_hook( client: httpx.AsyncClient, url: str, @@ -469,87 +213,22 @@ async def run_tool_hook( return decision -def parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser() - parser.add_argument("--base-url", required=True) - parser.add_argument("--api-key", required=True) - parser.add_argument("--model", required=True) - parser.add_argument("--system-prompt", default="") - parser.add_argument("--prompt", default="") - parser.add_argument("--initial-messages-file", default="") - parser.add_argument("--mcp-config", default="") - parser.add_argument("--tool-interception-url", default="") - parser.add_argument("--bash", action="store_true") - parser.add_argument("--compaction", action="store_true") - parser.add_argument("--summarize-at-tokens", type=int) - parser.add_argument("--edit", action="store_true") - parser.add_argument("--search", action="store_true") - parser.add_argument("--serper-key", default="") - return parser.parse_args() - - -async def main() -> None: - args = parse_args() - initial = [] - if args.initial_messages_file: - path = Path(args.initial_messages_file) - payload = path.read_bytes() - path.unlink() - initial = json.loads(payload) - client = AsyncOpenAI( - base_url=args.base_url, - api_key=args.api_key, - timeout=httpx.Timeout(600.0 if args.bash else None, connect=5.0), - ) - tool_client = ( - httpx.AsyncClient(timeout=httpx.Timeout(None, connect=5.0)) - if args.tool_interception_url - else None - ) - config = json.loads(args.mcp_config or "{}") - tools = [BASH_TOOL] if args.bash else [] - reserved = {"bash"} if args.bash else set() - if args.edit: - tools.append(EDIT_TOOL) - reserved.add("edit") - if args.search: - tools.append(SEARCH_TOOL) - reserved.add("search") - if config.get("mcpServers"): - mcp_tools, dispatch, servers = await asyncio.wait_for( - connect_mcp(config, reserved), timeout=None if args.bash else 60 - ) - else: - mcp_tools, dispatch, servers = [], {}, {} - tools += mcp_tools - messages = ( - [{"role": "system", "content": args.system_prompt}] - if args.system_prompt - else [] - ) - if initial: - messages.extend(initial) - elif args.prompt: - messages.append({"role": "user", "content": args.prompt}) - compactor = Compactor( - client, - args.model, - tools, - args.compaction, - args.summarize_at_tokens, - ) - if compactor.enabled and compactor.threshold is None: - compactor.threshold = await discover_threshold(client, args.model) - # The initial conversation is the floor for checkpoint fallbacks: a first-turn - # checkpoint must never retry from an empty base. - compactor.note_good(messages) +async def run_chat_loop( + args: argparse.Namespace, + compactor: "Compactor", + messages: list[dict], + dispatch: dict, + servers: dict, + tool_client: httpx.AsyncClient | None, +) -> None: + """Run the tool-calling conversation until a text-only reply or context exhaustion.""" while True: try: completion, messages = await compactor.complete(messages) except CompactionFailed: # The context is exhausted and could not be summarized: end the run # cleanly with what the conversation holds - still a trainable sample. - break + return except APIStatusError as error: # Null cannot compact, so context exhaustion ends it with the transcript so far. if args.bash or not is_context_overflow(error): @@ -558,7 +237,7 @@ async def main() -> None: message = completion.choices[0].message messages.append(message.model_dump(exclude_none=True)) if not message.tool_calls: - break + return tool_result_tokens = 0 for call in message.tool_calls: name = call.function.name @@ -633,10 +312,88 @@ async def main() -> None: try: messages = await compactor.compact(messages) except CompactionFailed: - break + return + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--base-url", required=True) + parser.add_argument("--api-key", required=True) + parser.add_argument("--model", required=True) + parser.add_argument("--system-prompt", default="") + parser.add_argument("--prompt", default="") + parser.add_argument("--initial-messages-file", default="") + parser.add_argument("--mcp-config", default="") + parser.add_argument("--tool-interception-url", default="") + parser.add_argument("--bash", action="store_true") + parser.add_argument("--compaction", action="store_true") + parser.add_argument("--summarize-at-tokens", type=int) + parser.add_argument("--edit", action="store_true") + parser.add_argument("--search", action="store_true") + parser.add_argument("--serper-key", default="") + return parser.parse_args() + + +async def main() -> None: + args = parse_args() + initial = [] + if args.initial_messages_file: + path = Path(args.initial_messages_file) + payload = path.read_bytes() + path.unlink() + initial = json.loads(payload) + client = AsyncOpenAI( + base_url=args.base_url, + api_key=args.api_key, + timeout=httpx.Timeout(600.0 if args.bash else None, connect=5.0), + ) + tool_client = ( + httpx.AsyncClient(timeout=httpx.Timeout(None, connect=5.0)) + if args.tool_interception_url + else None + ) + config = json.loads(args.mcp_config or "{}") + tools = [BASH_TOOL] if args.bash else [] + reserved = {"bash"} if args.bash else set() + if args.edit: + tools.append(EDIT_TOOL) + reserved.add("edit") + if args.search: + tools.append(SEARCH_TOOL) + reserved.add("search") + if config.get("mcpServers"): + mcp_tools, dispatch, servers = await asyncio.wait_for( + connect_mcp(config, reserved), timeout=None if args.bash else 60 + ) + else: + mcp_tools, dispatch, servers = [], {}, {} + tools += mcp_tools + messages = ( + [{"role": "system", "content": args.system_prompt}] + if args.system_prompt + else [] + ) + if initial: + messages.extend(initial) + elif args.prompt: + messages.append({"role": "user", "content": args.prompt}) + compactor = Compactor( + client, + args.model, + tools, + args.compaction, + args.summarize_at_tokens, + ) + if compactor.enabled and compactor.threshold is None: + compactor.threshold = await discover_threshold(client, args.model) + # The initial conversation is the floor for checkpoint fallbacks: a first-turn + # checkpoint must never retry from an empty base. + compactor.note_good(messages) + await run_chat_loop(args, compactor, messages, dispatch, servers, tool_client) if tool_client is not None: await tool_client.aclose() +# Inert on package import; the entry point once this module ends the bundled script. if __name__ == "__main__": asyncio.run(main()) diff --git a/verifiers/v1/harnesses/standalone.py b/verifiers/v1/harnesses/utils/launch.py similarity index 71% rename from verifiers/v1/harnesses/standalone.py rename to verifiers/v1/harnesses/utils/launch.py index 7546424735..0078c38dbb 100644 --- a/verifiers/v1/harnesses/standalone.py +++ b/verifiers/v1/harnesses/utils/launch.py @@ -1,23 +1,37 @@ import inspect import json from collections.abc import Sequence +from types import ModuleType from verifiers.v1.clients import ModelContext from verifiers.v1.configs.harness import HarnessConfig from verifiers.v1.dialects.chat import message_to_wire -from verifiers.v1.mcp import client as mcp_client +from verifiers.v1.harnesses.utils import compaction, core, mcp from verifiers.v1.runtimes import ProgramResult, Runtime from verifiers.v1.trace import Trace from verifiers.v1.types import Messages -MCP_CLIENT_SOURCE = inspect.getsource(mcp_client) PEP_723_END = "# ///\n" -def inline_mcp_client(program: str) -> str: - """Embed the public client so PEP 723 programs need only their declared packages.""" +def bundle_program(program: str, *modules: ModuleType) -> str: + """Embed utils modules so PEP 723 programs need only their declared packages.""" metadata, body = program.split(PEP_723_END, 1) - return f"{metadata}{PEP_723_END}{MCP_CLIENT_SOURCE}\n{body}" + sources = "\n".join(inspect.getsource(module) for module in modules) + return f"{metadata}{PEP_723_END}{sources}\n{body}" + + +# The shared Null/Bash chat program is the utils modules themselves: `core` ends with +# the `__main__` entry point, so the program text is only the script metadata. Secrets +# use argv so tools do not inherit them. +CHAT_PROGRAM_SOURCE = bundle_program( + '# /// script\n# requires-python = ">=3.10"\n' + '# dependencies = ["openai", "mcp==2.0.0", "httpx", "httpx2", "tenacity"]\n' + "# ///\n", + mcp, + compaction, + core, +) async def launch_chat_program( diff --git a/verifiers/v1/mcp/client.py b/verifiers/v1/harnesses/utils/mcp.py similarity index 100% rename from verifiers/v1/mcp/client.py rename to verifiers/v1/harnesses/utils/mcp.py diff --git a/verifiers/v1/mcp/__init__.py b/verifiers/v1/mcp/__init__.py index 70ebd0add2..5b98f22827 100644 --- a/verifiers/v1/mcp/__init__.py +++ b/verifiers/v1/mcp/__init__.py @@ -1,10 +1,3 @@ -from verifiers.v1.mcp.client import ( - call_mcp, - connect_mcp, - mcp_client, - mcp_content_to_chat_content, - with_retry, -) from verifiers.v1.mcp.launch import ( SharedToolServer, serve, @@ -20,12 +13,7 @@ "SharedToolsetConfig", "Toolset", "ToolsetConfig", - "call_mcp", - "connect_mcp", - "mcp_client", - "mcp_content_to_chat_content", "serve", "serve_shared", "serve_tools", - "with_retry", ] diff --git a/verifiers/v1/tasksets/nemo_gym/toolset.py b/verifiers/v1/tasksets/nemo_gym/toolset.py index 894a9afdbe..2c49f6eec0 100644 --- a/verifiers/v1/tasksets/nemo_gym/toolset.py +++ b/verifiers/v1/tasksets/nemo_gym/toolset.py @@ -7,7 +7,8 @@ from mcp.types import CallToolResult, TextContent, Tool from pydantic import Field -from verifiers.v1.mcp import SharedToolsetConfig, Toolset, mcp_client +from verifiers.v1.harnesses.utils.mcp import mcp_client +from verifiers.v1.mcp import SharedToolsetConfig, Toolset from verifiers.v1.state import State