diff --git a/verifiers/v1/harnesses/bash/harness.py b/verifiers/v1/harnesses/bash/harness.py index 4852f0d293..39248d5ca3 100644 --- a/verifiers/v1/harnesses/bash/harness.py +++ b/verifiers/v1/harnesses/bash/harness.py @@ -1,5 +1,8 @@ import os +from pydantic import PositiveInt +from pydantic_config import BaseConfig + from verifiers.v1.clients import ModelContext from verifiers.v1.configs.harness import HarnessConfig from verifiers.v1.harness import Harness @@ -24,6 +27,14 @@ ) +class CompactionConfig(BaseConfig): + """Context compaction policy for the bash agent loop.""" + + summarize_at_tokens: PositiveInt | None = None + """Compact at this token count. When unset, compact when 16k tokens remain below the + model context window when the provider advertises it.""" + + class BashHarnessConfig(HarnessConfig): edit: bool = True """Offer the local `edit` tool (single-occurrence string replacement in a file) alongside @@ -34,6 +45,9 @@ class BashHarnessConfig(HarnessConfig): eval environment; the key is handed to the program over argv (like the interception secret) so the agent's `bash` subprocesses don't inherit it.""" + compaction: CompactionConfig | None = None + """Context compaction policy. Set an empty config to use automatic thresholds.""" + class BashHarness(Harness[BashHarnessConfig]): APPENDS_SYSTEM_PROMPT = True @@ -69,6 +83,11 @@ async def launch( args = ["--bash"] if tool_interception_url: args.append(f"--tool-interception-url={tool_interception_url}") + if self.config.compaction is not None: + args.append("--compaction") + threshold = self.config.compaction.summarize_at_tokens + if threshold is not None: + args.append(f"--summarize-at-tokens={threshold}") if self.config.edit: args.append("--edit") if self.config.search: diff --git a/verifiers/v1/harnesses/minimal/program.py b/verifiers/v1/harnesses/minimal/program.py index 1ba59d3092..6535458f9a 100644 --- a/verifiers/v1/harnesses/minimal/program.py +++ b/verifiers/v1/harnesses/minimal/program.py @@ -12,7 +12,7 @@ from typing import TYPE_CHECKING import httpx -from openai import APIStatusError, AsyncOpenAI +from openai import APIError, APIStatusError, AsyncOpenAI if TYPE_CHECKING: # The harness bundles this module into the generated script before execution. @@ -40,6 +40,92 @@ def is_context_overflow(error: APIStatusError) -> bool: ) +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 + + BASH_TOOL = { "type": "function", "function": { @@ -197,12 +283,171 @@ def run_edit(path: str, old_str: str, new_str: str) -> str: async def chat( - client: AsyncOpenAI, model: str, messages: list[dict], tools: list[dict] + client: AsyncOpenAI, + model: str, + messages: list[dict], + tools: list[dict], + *, + tool_choice: str | None = None, ): - completion = await client.chat.completions.create( - model=model, messages=messages, tools=tools or None + kwargs = {"model": model, "messages": messages, "tools": tools or None} + if tools and tool_choice is not None: + kwargs["tool_choice"] = tool_choice + 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 completion.choices[0].message + 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( @@ -235,6 +480,8 @@ def parse_args() -> argparse.Namespace: 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="") @@ -284,17 +531,35 @@ async def main() -> None: 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) while True: try: - message = await chat(client, args.model, messages, tools) + 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 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): raise return + message = completion.choices[0].message messages.append(message.model_dump(exclude_none=True)) if not message.tool_calls: break + tool_result_tokens = 0 for call in message.tool_calls: name = call.function.name tool_message = { @@ -313,7 +578,11 @@ async def main() -> None: tool_message, ) if decision["action"] == "rewrite": - messages.append(decision["message"]) + rewritten = bound_tool_message(decision["message"]) + messages.append(rewritten) + tool_result_tokens += estimated_tokens( + str(rewritten.get("content", "")) + ) continue try: tool_args = json.loads(call.function.arguments or "{}") @@ -346,6 +615,7 @@ async def main() -> None: else: content = f"error: unknown tool {name!r}" tool_message["content"] = content + tool_message = bound_tool_message(tool_message) if args.tool_interception_url: assert tool_client is not None decision = await run_tool_hook( @@ -356,8 +626,14 @@ async def main() -> None: tool_message, ) if decision["action"] == "rewrite": - tool_message = decision["message"] + tool_message = bound_tool_message(decision["message"]) messages.append(tool_message) + tool_result_tokens += estimated_tokens(str(tool_message["content"])) + if compactor.reached(completion, tool_result_tokens) and compactable(messages): + try: + messages = await compactor.compact(messages) + except CompactionFailed: + break if tool_client is not None: await tool_client.aclose() diff --git a/verifiers/v1/interception/server.py b/verifiers/v1/interception/server.py index fcb5cc9b41..258d21170e 100644 --- a/verifiers/v1/interception/server.py +++ b/verifiers/v1/interception/server.py @@ -33,13 +33,15 @@ from tempfile import SpooledTemporaryFile from typing import Literal +import httpx from aiohttp import web from pydantic import ValidationError from pydantic_core import PydanticSerializationError, from_json, to_json from verifiers.v1 import graph from verifiers.v1.clients import Client, resolve_client -from verifiers.v1.configs.client import BaseClientConfig +from verifiers.v1.clients.base import join_url +from verifiers.v1.configs.client import BaseClientConfig, resolve_api_key from verifiers.v1.dialects import DIALECTS, Dialect from verifiers.v1.dialects.base import ( PROVIDER_CAPABILITY_POLICY_CODE, @@ -326,6 +328,7 @@ async def start(self) -> None: app.router.add_post(route, self._handler_for(dialect)) for aux in dialect.aux_routes: app.router.add_post(aux, self._aux_handler_for(dialect, aux)) + app.router.add_get("/v1/models", self.handle_models) # Tool servers use a state-only capability; the model bearer cannot reach these. app.router.add_get("/state", self.handle_state_get) app.router.add_put("/state", self.handle_state_put) @@ -1053,6 +1056,41 @@ async def handle_aux( return web.json_response(dialect.error_body(str(e)), status=502) return web.json_response(result) + async def handle_models(self, request: web.Request) -> web.Response: + """`GET /v1/models`: relay the upstream model listing so agent loops can read a + provider context-window extension (e.g. vLLM's `max_model_len`). The path is shared + by every dialect; only the auth carrier differs, so the bearer is tried per dialect. + A pure relay from the session's endpoint config — never recorded on the trace, and a + failure never fails the rollout.""" + for dialect in DIALECTS: + session = self.sessions.get(dialect.secret(request.headers)) + if session is not None: + break + else: + return web.json_response({"error": "unauthorized"}, status=401) + session.adopt(asyncio.current_task()) + logger.debug("intercept models: id=%s", session.trace.id) + config = session.ctx.client + headers = dict(config.headers or {}) + headers.update(dialect.auth_headers(resolve_api_key(config))) + try: + # Finite read timeout: a hung provider must not stall threshold discovery + # for the rollout's whole outer timeout - the loop falls back to no compaction. + async with httpx.AsyncClient( + timeout=httpx.Timeout(30.0, connect=5.0) + ) as client: + upstream = await client.get( + join_url(config.base_url, "/v1/models"), headers=headers + ) + except httpx.HTTPError as e: + logger.warning("models call failed: id=%s %s", session.trace.id, e) + return web.json_response(dialect.error_body(str(e)), status=502) + return web.Response( + body=upstream.content, + status=upstream.status_code, + content_type="application/json", + ) + def _session_for( self, request: web.Request, *, allow_service: bool = False ) -> RolloutSession | None: