feat(default): auto-compaction via summarize_at_tokens

verifiers/v1/harnesses/default/harness.py

@@ -1,7 +1,10 @@
 import json
 import os
+import random
 from pathlib import Path
 
+from pydantic import model_validator
+
 from verifiers.v1.harness import Harness, HarnessConfig
 from verifiers.v1.clients import ModelContext
 from verifiers.v1.dialects.chat import message_to_wire
@@ -35,6 +38,30 @@ class DefaultHarnessConfig(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."""
 
+    summarize_at_tokens: int | tuple[int, int] | None = None
+    """Auto-compaction threshold: once the context grows past this many tokens, the program asks
+    the model to summarize its progress and restarts the message list from the initial prompt plus
+    that summary. An int is a fixed threshold; a `(lo, hi)` pair draws a per-group threshold
+    (seeded by the task index, so a task's rollouts share one draw and tasks vary). `None`
+    disables auto-compaction; ints must be positive."""
+
+    @model_validator(mode="after")
+    def validate_limits(self) -> "DefaultHarnessConfig":
+        value = self.summarize_at_tokens
+        if isinstance(value, tuple):
+            lo, hi = value
+            if lo <= 0 or hi <= 0:
+                raise ValueError("`summarize_at_tokens` range bounds must be positive.")
+            if lo > hi:
+                raise ValueError(
+                    "`summarize_at_tokens` range must be (lo, hi) with lo <= hi."
+                )
+        elif value is not None and value <= 0:
+            raise ValueError(
+                "`summarize_at_tokens` must be positive, or None to disable."
+            )
+        return self
+
 
 class DefaultHarness(Harness[DefaultHarnessConfig]):
     APPENDS_SYSTEM_PROMPT = True
@@ -45,6 +72,17 @@ class DefaultHarness(Harness[DefaultHarnessConfig]):
     async def setup(self, runtime: Runtime) -> None:
         await runtime.prepare_uv_script(PROGRAM_SOURCE, self.config.resolved_env)
 
+    def summarize_threshold(self, task_idx: int) -> int:
+        """The resolved auto-compaction threshold: a range draws per-group (seeded by task index,
+        so a task's rollouts share one threshold). 0 when disabled."""
+        value = self.config.summarize_at_tokens
+        if value is None:
+            return 0
+        if isinstance(value, tuple):
+            lo, hi = value
+            return random.Random(task_idx).randint(lo, hi)
+        return value
+
     async def launch(
         self,
         ctx: ModelContext,
@@ -72,6 +110,9 @@ async def launch(
         ]
         if self.config.edit:
             args.append("--edit")
+        threshold = self.summarize_threshold(trace.task.idx)
+        if threshold:
+            args.append(f"--summarize-at-tokens={threshold}")
         if self.config.search:
             # Resolve the key and keep it OUT of the program env: it's handed to the program over
             # argv (--serper-key), so popping it here stops the agent's `bash` subprocesses from

verifiers/v1/harnesses/default/program.py

@@ -17,6 +17,28 @@
 
 SERPER_URL = "https://google.serper.dev/search"
 
+# rlm-parity compaction prompts (mirrors rlm's engine.py).
+CHECKPOINT_COMPACTION_PROMPT = (
+    "You are performing a CONTEXT CHECKPOINT COMPACTION. "
+    "Create a handoff summary for another LLM that will resume the task.\n"
+    "\n"
+    "Include:\n"
+    "- Current progress and key decisions made\n"
+    "- Important context, constraints, or user preferences\n"
+    "- What remains to be done (clear next steps)\n"
+    "- Any critical data, examples, or references needed to continue\n"
+    "\n"
+    "Be concise, structured, and focused on helping the next LLM "
+    "seamlessly continue the work."
+)
+POST_COMPACTION_FRAMING = (
+    "Another language model started to solve this problem and produced "
+    "a summary of its thinking process. 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:"
+)
+
 BASH_TOOL = {
     "type": "function",
     "function": {
@@ -170,12 +192,20 @@ 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
-    )
-    return completion.choices[0].message
+    """One completion; returns (message, prompt tokens) — `usage.prompt_tokens` is the
+    context size this turn conditioned on (rlm's compaction trigger)."""
+    kwargs = {"model": model, "messages": messages, "tools": tools or None}
+    if tools and tool_choice is not None:
+        kwargs["tool_choice"] = tool_choice
+    completion = await client.chat.completions.create(**kwargs)
+    usage = completion.usage
+    return completion.choices[0].message, usage.prompt_tokens if usage else 0
 
 
 async def connect_mcp(stack: AsyncExitStack, config: dict) -> tuple[list[dict], dict]:
@@ -248,6 +278,7 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--edit", action="store_true")
     parser.add_argument("--search", action="store_true")
     parser.add_argument("--serper-key", default="")
+    parser.add_argument("--summarize-at-tokens", type=int, default=0)
     return parser.parse_args()
 
 
@@ -284,8 +315,11 @@ async def main() -> None:
             messages.extend(initial)
         elif args.prompt:
             messages.append({"role": "user", "content": args.prompt})
+        # Compaction keeps only the system message (rlm semantics: the original task
+        # prompt is dropped; the summary carries the goal).
+        system_messages = [m for m in messages if m.get("role") == "system"]
         while True:
-            message = await chat(client, args.model, messages, tools)
+            message, context_tokens = await chat(client, args.model, messages, tools)
             messages.append(message.model_dump(exclude_none=True))
             if not message.tool_calls:
                 break
@@ -338,6 +372,19 @@ async def main() -> None:
                 messages.append(
                     {"role": "tool", "tool_call_id": call.id, "content": content}
                 )
+            # Compact after the turn's tool results land, so the summary sees them (rlm
+            # semantics: tools stay in the request with tool_choice="none" so the prompt
+            # renders like a regular turn; tool calls in the reply are ignored; the rebuilt
+            # list is system + framed summary, dropping the original prompt).
+            if args.summarize_at_tokens and context_tokens >= args.summarize_at_tokens:
+                messages.append(
+                    {"role": "user", "content": CHECKPOINT_COMPACTION_PROMPT}
+                )
+                summary, _ = await chat(
+                    client, args.model, messages, tools, tool_choice="none"
+                )
+                framed = POST_COMPACTION_FRAMING + "\n\n" + (summary.content or "")
+                messages = system_messages + [{"role": "user", "content": framed}]
 
 
 if __name__ == "__main__":