feat: add bash context compaction

pyproject.toml

@@ -290,6 +290,7 @@ markers = [
     "bash: v1 e2e cases on the bash harness",
     "browser_use: v1 e2e cases on the browser_use harness",
     "rlm: v1 e2e cases on the rlm harness",
+    "compaction: v1 context-compaction E2E cases",
     "kimi_code: v1 e2e cases on the kimi-code harness",
     "pi: v1 e2e cases on the pi harness",
     "pool: v1 e2e cases on the pool harness",

tests/v1/conftest.py

@@ -96,7 +96,7 @@ def pytest_configure(config) -> None:
 def pytest_collection_modifyitems(config, items) -> None:
     """Skip the live-model tests (marked `e2e`) when no model endpoint is configured, so the
     rest of the suite (e.g. config parsing) still runs in a keyless environment."""
-    if os.environ.get("PRIME_API_KEY"):
+    if os.environ.get("PRIME_API_KEY") or os.environ.get("VF_COMPACTION_E2E_BASE_URL"):
         return
     skip = pytest.mark.skip(reason="needs PRIME_API_KEY")
     for item in items:
@@ -124,14 +124,16 @@ def _eval_config(
     harness: str | HarnessConfig | None = "null",
     n: int = 1,
     num_tasks: int = 1,
-    max_tokens: int = 2048,
+    max_tokens: int | None = 2048,
     max_turns: int | None = 4,
     rollout_timeout: float = 180,
     taskset_overrides: dict | None = None,
     runtime: dict | None = None,
     env: dict | None = None,
     pool: dict | None = None,
     reasoning_effort: str | None = None,
+    model: str = CI_MODEL,
+    client: dict | None = None,
     server: bool = False,
 ) -> EvalConfig:
     """Build the smallest `EvalConfig` that still exercises the path, shared by the in-process
@@ -187,7 +189,8 @@ def _eval_config(
         serve=({"pool": pool} if pool else {}) if server else None,
         output_dir=output_dir.parent,
         run={"dir": output_dir.name},
-        model=CI_MODEL,
+        model=model,
+        client=client or {"type": "eval"},
     )
 
 

tests/v1/fixtures/context_compaction_v1.py

@@ -0,0 +1,86 @@
+"""Context-compaction E2E scenarios for harness agent loops."""
+
+from typing import Literal
+
+from pydantic import Field
+
+import verifiers.v1 as vf
+
+
+class OverflowToolsetConfig(vf.ToolsetConfig):
+    payload_chars: int = Field(65_536, gt=0)
+
+
+class OverflowToolset(vf.Toolset[OverflowToolsetConfig]):
+    TOOL_PREFIX = "overflow"
+
+    def __init__(self, config: OverflowToolsetConfig):
+        super().__init__(config)
+        self.called = False
+
+    @vf.tool
+    def overflow_context(self) -> str:
+        """Return a payload that is intentionally larger than the model context."""
+        if self.called:
+            return "The overflow already occurred. Answer `recovered` now."
+        self.called = True
+        block = "0123456789abcdef "
+        repeats = self.config.payload_chars // len(block) + 1
+        return (block * repeats)[: self.config.payload_chars]
+
+
+class ContextCompactionTaskConfig(vf.TaskConfig):
+    scenario: Literal["decode", "tool_result"] = "decode"
+    payload_chars: int = Field(65_536, gt=0)
+    tools: OverflowToolsetConfig = OverflowToolsetConfig()
+
+
+class ContextCompactionTask(
+    vf.Task[vf.TaskData, vf.State, ContextCompactionTaskConfig]
+):
+    @classmethod
+    def toolsets(cls, config: ContextCompactionTaskConfig) -> list[vf.Toolset]:
+        if config.scenario != "tool_result":
+            return []
+        tool_config = config.tools.model_copy(
+            update={"payload_chars": config.payload_chars}
+        )
+        return [OverflowToolset(tool_config)]
+
+    @vf.reward
+    async def compacted(self, trace: vf.Trace) -> float:
+        return float(trace.num_branches > 1)
+
+
+class ContextCompactionConfig(vf.TasksetConfig):
+    task: ContextCompactionTaskConfig = ContextCompactionTaskConfig()
+
+
+class ContextCompactionTaskset(
+    vf.Taskset[ContextCompactionTask, ContextCompactionConfig]
+):
+    def load(self) -> list[ContextCompactionTask]:
+        if self.config.task.scenario == "decode":
+            prompt = (
+                "Write `x ` repeatedly. Do not use tools and do not stop. "
+                "Continue until the model context ends the decode."
+            )
+        else:
+            prompt = (
+                "Call the `overflow_context` tool exactly once, then answer `recovered`. "
+                "In an RLM IPython session, call it with "
+                "`result = await overflow_overflow_context(); print(result)`."
+            )
+        return [
+            ContextCompactionTask(
+                vf.TaskData(idx=0, prompt=prompt),
+                self.config.task,
+            )
+        ]
+
+
+__all__ = ["ContextCompactionTaskset"]
+
+
+if __name__ == "__main__":
+    OverflowToolset(OverflowToolsetConfig()).run()

tests/v1/test_e2e.py

@@ -5,8 +5,12 @@
 with distinct networking — instead of fanning the full cross product. prime/modal rows
 are local-only (their marks are excluded in CI)."""
 
+import os
+
 import pytest
 
+from verifiers.v1.utils.loaders import harness_config_type
+
 mark = pytest.mark
 
 
@@ -156,6 +160,66 @@ async def test_single_turn(run_v1, harness, harness_runtime, tmp_path):
         assert call.time.duration > 0
 
 
+@pytest.mark.e2e
+@pytest.mark.compaction
+@pytest.mark.docker
+@pytest.mark.parametrize("scenario", ["decode", "tool_result"])
+@pytest.mark.parametrize("harness_id", ["bash", "rlm"])
+async def test_context_compaction_matrix(run_v1, scenario, harness_id, tmp_path):
+    """Both in-house loops recover when decoding or tool output fills context."""
+    base_url = os.environ.get("VF_COMPACTION_E2E_BASE_URL")
+    model = os.environ.get("VF_COMPACTION_E2E_MODEL")
+    context_window = int(os.environ.get("VF_COMPACTION_E2E_CONTEXT_WINDOW", "4096"))
+    if not base_url or not model:
+        pytest.skip("needs a local compaction E2E model")
+
+    harness = {
+        "id": harness_id,
+        "summarize_at_tokens": context_window,
+        **({"compaction": True} if harness_id == "bash" else {}),
+    }
+    sampling = (
+        {"extra_body": {"ignore_eos": True}}
+        if scenario == "decode"
+        else {"temperature": 0.0}
+    )
+    (trace,) = await run_v1(
+        "context-compaction-v1",
+        harness=harness_config_type(harness_id).model_validate(harness),
+        runtime={"type": "docker"},
+        env={"agent": {"sampling": sampling, "max_output_tokens": None}},
+        client={
+            "type": "eval",
+            "base_url": base_url,
+            "api_key_var": "VF_COMPACTION_E2E_API_KEY",
+        },
+        model=model,
+        max_tokens=None if scenario == "decode" else 512,
+        max_turns=8,
+        rollout_timeout=600,
+        taskset_overrides={
+            "task": {
+                "scenario": scenario,
+                "payload_chars": context_window * 12,
+            }
+        },
+        output_dir=tmp_path / f"{scenario}-{harness_id}",
+    )
+
+    assert trace.ok, trace.errors
+    assert trace.num_branches > 1
+    assert trace.rewards["compacted"].score == 1.0
+    if scenario == "decode":
+        assert any(call.finish_reason == "length" for call in trace.calls)
+    else:
+        assert any(
+            call.error is not None and call.error.type == "OverlongPromptError"
+            for call in trace.calls
+        )
+    if harness_id == "rlm":
+        assert trace.metrics["num_compactions"] >= 1
+
+
 @pytest.mark.e2e
 @pytest.mark.browser_use
 @pytest.mark.docker

verifiers/v1/harnesses/bash/harness.py

@@ -2,6 +2,8 @@
 import os
 from pathlib import Path
 
+from pydantic import PositiveInt
+
 from verifiers.v1.clients import ModelContext
 from verifiers.v1.configs.harness import HarnessConfig
 from verifiers.v1.dialects.chat import message_to_wire
@@ -28,6 +30,13 @@
 
 
 class BashHarnessConfig(HarnessConfig):
+    compaction: bool = True
+    """Recover from context exhaustion with a handoff summary and a fresh branch."""
+
+    summarize_at_tokens: PositiveInt | None = None
+    """Compact proactively when the estimated active context reaches this threshold. When unset,
+    compaction still recovers from provider context errors."""
+
     edit: bool = True
     """Offer the local `edit` tool (single-occurrence string replacement in a file) alongside
     `bash`. On by default; set `--env.agent.harness.edit false` for a bash-only agent."""
@@ -77,6 +86,10 @@ async def launch(
         ]
         if tool_interception_url:
             args.append(f"--tool-interception-url={tool_interception_url}")
+        if self.config.compaction:
+            args.append("--compaction")
+            if self.config.summarize_at_tokens is not None:
+                args.append(f"--summarize-at-tokens={self.config.summarize_at_tokens}")
         if self.config.edit:
             args.append("--edit")
         if self.config.search: