fix: let provider evidence adjudicate past-window preflight estimates

agent/context_compressor.py

@@ -2162,9 +2162,8 @@ class ContextCompressor(SummaryDispatchMixin, MicroCompactionMixin, ContextEngin
         self._tail_token_budget = None
         _ = self.tail_token_budget  # eager recompute, same timing as before
         self.max_summary_tokens = min(int(context_length * 0.05), _SUMMARY_TOKENS_CEILING)
-        # Calibration state is only valid for the model that produced it: carried to a smaller window it would let
-        # should_defer_preflight_to_real_usage() suppress a compaction the new model needs. 0 (not the -1 sentinel)
-        # means "no real usage yet -> use the rough estimate" so post-response should_compress still fires.
+        # Old usage cannot price a new model. Clear it without arming the post-compaction
+        # latch: the next response supplies usage or enables the usage-less fallback.
         self.last_prompt_tokens = self.last_completion_tokens = self.last_total_tokens = 0
         self._reset_real_usage_pairing()
         # Strikes were judged against the previous threshold; void them durably too.
@@ -2441,9 +2440,9 @@ class ContextCompressor(SummaryDispatchMixin, MicroCompactionMixin, ContextEngin
             return False
         if self.awaiting_real_usage_after_compression:
             return True
-        # A real reading already at/over threshold needs no second opinion, and a rough figure past
-        # the whole window describes a request certain to fail — sending it only buys an overflow error.
-        if self.last_real_prompt_tokens >= self.threshold_tokens or rough_tokens >= self.context_length:
+        # Estimate magnitude is not evidence of overflow, even past the full window.
+        # Let the provider adjudicate; its overflow error still triggers reactive recovery.
+        if self.last_real_prompt_tokens >= self.threshold_tokens:
             return False
         return not self._provider_omits_usage
 

evals/token_accounting/replay_gates.py

@@ -3,8 +3,8 @@ local ``bytes/4`` estimate? (#104462)
 
 Runs the REAL ``AIAgent`` turn loop against a local fake OpenAI chat-completions server whose
 ``usage.prompt_tokens`` is scripted per request and deliberately disagrees with the transcript
-size. The only patch on the agent path is a counting wrapper around ``_compress_context``
-(the question is *whether* a gate fires, not what a summarizer writes).
+size. Gate scenarios replace ``_compress_context`` with a counting hook; recovery
+scenarios run real compression with only summary generation replaced by fixed local text.
 
 Shapes (each runs several turns):
 
@@ -18,6 +18,9 @@ Scenarios per shape:
 
 * ``inflated``  — transcript ~3x the threshold by bytes/4, provider reports real usage well
                   UNDER threshold. No gate may fire (positive arm).
+* ``past_window_inflated`` — whole-history estimate exceeds the entire model window.
+* ``overflow_recovery`` / ``usage_less_fallback`` — real compression and local HTTP retry,
+  with fixed local summary text (no inference); provider capability probes do not consume usage.
 * ``deflated``  — transcript tiny by bytes/4, provider reports real usage OVER threshold.
                   Local compression MUST fire once real usage is known (negative arm).
 
@@ -29,6 +32,7 @@ Usage (from a checkout root, venv python)::
 from __future__ import annotations
 
 import argparse
+import hashlib
 import json
 import os
 import subprocess
@@ -55,7 +59,7 @@ class _FakeChat:
 
     def __init__(self) -> None:
         self.requests: list[dict] = []
-        self.usage_script: list[int] = []
+        self.usage_script: list[int | None | str] = []
         self.lock = threading.Lock()
         server = self
 
@@ -66,10 +70,25 @@ class _FakeChat:
             def do_POST(self):
                 n = int(self.headers.get("content-length", 0))
                 body = json.loads(self.rfile.read(n) or b"{}")
+                # Runtime capability probes are not model requests and must not consume usage.
+                if "messages" not in body:
+                    self.send_error(404, "Only chat completions are supported by this fixture")
+                    return
                 with server.lock:
                     server.requests.append(body)
                     prompt = server.usage_script.pop(0) if server.usage_script else REAL_UNDER
-                usage = {"prompt_tokens": prompt, "completion_tokens": 5, "total_tokens": prompt + 5}
+                if prompt == "overflow":
+                    data = json.dumps({"error": {"message": "maximum context length exceeded",
+                                                "type": "invalid_request_error",
+                                                "code": "context_length_exceeded"}}).encode()
+                    self.send_response(400)
+                    self.send_header("content-type", "application/json")
+                    self.send_header("content-length", str(len(data)))
+                    self.end_headers()
+                    self.wfile.write(data)
+                    return
+                usage = ({"prompt_tokens": prompt, "completion_tokens": 5, "total_tokens": prompt + 5}
+                         if isinstance(prompt, int) else None)
                 msg = {"role": "assistant", "content": "ok"}
                 if body.get("stream") is True:
                     self.send_response(200)
@@ -147,6 +166,7 @@ def _run_turns(wire: _FakeChat, agent_factory, history: list[dict], *, reload: b
             history = json.loads(json.dumps(history))
         before = len(agent._ab_compress_calls)
         r = agent.run_conversation(f"turn {t + 1}", conversation_history=history)
+        assert r.get("completed"), r.get("error")
         compress_by_turn.append(len(agent._ab_compress_calls) - before)
         history = r["messages"]
         approx = list(agent._ab_compress_calls)
@@ -176,7 +196,7 @@ def scenario(wire: _FakeChat, *, shape: str, real: int, chars: int, tmp: Path) -
 
     wire.requests.clear()
     wire.usage_script[:] = [real] * (TURNS + 2)
-    sid = f"replay-{shape}-{real}"
+    sid = f"replay-{shape}-{real}-{chars}"
     db_path = tmp / f"state-{sid}.db"
     db = SessionDB(db_path=db_path)
     # The prior transcript is already durable (run_conversation treats ``conversation_history``
@@ -187,6 +207,7 @@ def scenario(wire: _FakeChat, *, shape: str, real: int, chars: int, tmp: Path) -
     history = db.get_messages_as_conversation(sid)
     first = _make_agent(wire.base_url, session_id=sid, session_db=db)
     r1 = first.run_conversation("turn 1", conversation_history=history)
+    assert r1.get("completed"), r1.get("error")
     calls_t1 = len(first._ab_compress_calls)
     db.close()
     reopened = SessionDB(db_path=db_path)
@@ -202,6 +223,7 @@ def scenario(wire: _FakeChat, *, shape: str, real: int, chars: int, tmp: Path) -
     for t in range(1, TURNS):
         before = len(fresh._ab_compress_calls)
         r = fresh.run_conversation(f"turn {t + 1}", conversation_history=hist)
+        assert r.get("completed"), r.get("error")
         compress_by_turn.append(len(fresh._ab_compress_calls) - before)
         hist = r["messages"]
     reopened.close()
@@ -209,13 +231,44 @@ def scenario(wire: _FakeChat, *, shape: str, real: int, chars: int, tmp: Path) -
         "threshold": THRESHOLD, "real_prompt_tokens": real,
         "rough_estimate_turn1_messages": _rough(r1["messages"]),
         "persisted_messages": len(persisted),
-        "anchor_restored_in_fresh_process": anchor_restored,
+        "anchor_restored_in_fresh_agent": anchor_restored,
         "local_compress_calls_by_turn": compress_by_turn,
         "local_compress_approx_tokens": list(first._ab_compress_calls) + list(fresh._ab_compress_calls),
         "provider_requests": len(wire.requests),
     }
 
 
+def recovery_scenario(wire: _FakeChat, *, usage_less: bool) -> dict:
+    """Real compression/retry, with only summary generation replaced by fixed local text."""
+    from run_agent import AIAgent
+
+    agent = _make_agent(wire.base_url)
+    wire.requests.clear()
+    wire.usage_script[:] = [None] * 8 if usage_less else ["overflow", REAL_UNDER]
+    calls = []
+    original = AIAgent._compress_context.__get__(agent, AIAgent)
+
+    def compress(*args, **kwargs):
+        calls.append(len(wire.requests))
+        return original(*args, **kwargs)
+
+    agent._compress_context = compress
+    agent.context_compressor._generate_summary = lambda *a, **kw: "Local fixture summary of prior work."
+    history = [{"role": "user" if i % 2 == 0 else "assistant",
+                "content": f"record {i} " + "x" * 32_000} for i in range(20)]
+    first = agent.run_conversation("continue", conversation_history=history)
+    first_calls = len(calls)
+    result = (agent.run_conversation("continue again", conversation_history=first["messages"])
+              if usage_less else first)
+    sizes = [len(json.dumps(r["messages"])) for r in wire.requests]
+    return {"completed": result.get("completed"), "response": result.get("final_response"),
+            "compression_after_provider_requests": calls, "first_turn_compressions": first_calls,
+            "provider_request_sizes": sizes, "provider_omits_usage": agent.context_compressor._provider_omits_usage,
+            "pass": bool(result.get("completed") and calls and calls[0] >= 1
+                         and min(sizes[1:]) < sizes[0]
+                         and (not usage_less or first_calls == 0))}
+
+
 def main() -> int:
     ap = argparse.ArgumentParser()
     ap.add_argument("--out", required=True)
@@ -225,17 +278,26 @@ def main() -> int:
     (tmp / "home").mkdir(parents=True)
     head = subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT, capture_output=True, text=True).stdout.strip()
     wire = _FakeChat()
-    result: dict = {"checkout": str(ROOT), "head": head}
+    result: dict = {"checkout": str(ROOT), "head": head,
+                    "compressor_sha256": hashlib.sha256(
+                        (ROOT / "agent/context_compressor.py").read_bytes()).hexdigest()}
     try:
         for shape in ("cli", "gateway", "restore"):
+            result[f"{shape}_past_window_inflated"] = scenario(
+                wire, shape=shape, real=REAL_UNDER, chars=640_000, tmp=tmp)
             result[f"{shape}_inflated"] = scenario(wire, shape=shape, real=REAL_UNDER, chars=INFLATED_CHARS, tmp=tmp)
             result[f"{shape}_deflated"] = scenario(wire, shape=shape, real=REAL_OVER, chars=400, tmp=tmp)
+        result["overflow_recovery"] = recovery_scenario(wire, usage_less=False)
+        result["usage_less_fallback"] = recovery_scenario(wire, usage_less=True)
     finally:
         wire.close()
     verdict = {}
     for k, v in result.items():
         if not isinstance(v, dict):
             continue
+        if "pass" in v:
+            verdict[k] = "PASS" if v["pass"] else "FAIL"
+            continue
         fired = sum(v["local_compress_calls_by_turn"])
         # inflated: real under threshold → no gate may fire. deflated: real over → must fire ≥1 after turn 1.
         verdict[k] = ("PASS" if fired == 0 else "FAIL") if k.endswith("inflated") else (

tests/agent/test_context_compressor.py

@@ -282,14 +282,14 @@ class TestPreflightDeferral:
         assert compressor.should_defer_preflight_to_real_usage(80_000) is False
 
     def test_never_defers_when_real_usage_cannot_arrive_or_is_already_over(self, compressor):
-        """Deferral is one request, never a disable: a provider that omits usage, a real reading
-        already over threshold, or a rough figure past the whole window all compress now."""
+        """Only provider evidence ends deferral, not the magnitude of a rough estimate."""
         compressor.context_length = 100_000
         compressor.threshold_tokens = 85_000
         compressor.last_real_prompt_tokens = 90_000
         assert compressor.should_defer_preflight_to_real_usage(95_000) is False
         compressor.last_real_prompt_tokens = 50_000
-        assert compressor.should_defer_preflight_to_real_usage(150_000) is False
+        for rough in (compressor.context_length, 150_000, 10_000_000):
+            assert compressor.should_defer_preflight_to_real_usage(rough) is True
         compressor.note_usage_less_response()
         assert compressor.should_defer_preflight_to_real_usage(95_000) is False
         compressor.update_from_response({"prompt_tokens": 50_000})
@@ -2151,18 +2151,18 @@ class TestUpdateModelResetsCalibration:
         assert comp.awaiting_real_usage_after_compression is False
         assert comp._ineffective_compression_count == 0
 
-    def test_defer_no_longer_suppresses_after_switch(self):
-        """The exact #23767 failure: old model's 'it fit' must not defer
-        preflight on the new smaller model."""
+    def test_switch_waits_for_new_provider_evidence(self):
+        """A model switch clears old evidence; the new provider adjudicates pressure."""
         comp = self._comp()
         comp.last_real_prompt_tokens = 50_000
         # Before switch, a modest rough growth would defer.
         comp.threshold_tokens = 85_000
         assert comp.should_defer_preflight_to_real_usage(93_000) is True
 
-        # After switching to a 65K model the stale reading is gone; a rough estimate past the
-        # NEW window is a request certain to fail and is never deferred — preflight will run.
         comp.update_model("small-model", context_length=65_536)
+        assert comp.last_real_prompt_tokens == 0
+        assert comp.should_defer_preflight_to_real_usage(comp.context_length + 5_000) is True
+        comp.update_from_response({"prompt_tokens": comp.threshold_tokens + 1})
         assert comp.should_defer_preflight_to_real_usage(comp.context_length + 5_000) is False
 
 

website/docs/developer-guide/context-compression-and-caching.md

@@ -75,23 +75,58 @@ Located in `agent/context_compressor.py`. This is the **primary compression
 system** that runs inside the agent's tool loop with access to accurate,
 API-reported token counts.
 
-#### Token accounting: real usage decides, the estimate only decides whether to wait
+#### Token accounting: provider anchors and explicit heuristic fallbacks
 
 Every compaction gate (turn-start preflight, idle, pre-API pressure, post-tool)
 asks the **usage anchor** first (`agent/usage_anchor.py`): the provider's last
-`usage.prompt_tokens` plus a rough estimate of ONLY the messages appended since
+prompt and completion token counts plus a rough estimate of ONLY the messages appended since
 that response. The anchor identifies the priced transcript by a content
 fingerprint, so it survives the gateway re-reading history from the DB every
 turn, and it is persisted on the session row so a fresh process (`--resume`,
 desktop per-turn `serve`) restores it while the durable transcript still
 matches. Compaction, session reset and codex-native compaction clear it.
 
-Without an anchor (first request, rewind/edit-resend) a whole-context rough
-estimate over threshold **waits one request** for the provider's real count
-instead of compressing on a guess (`should_defer_preflight_to_real_usage`).
-The wait is one request, never a disable: a provider that omits usage, a real
-reading already over threshold, a rough figure past the whole window, or a
-provider-proven overflow all compress immediately.
+For the built-in engine's **turn-start and pre-API threshold gates**, without an
+anchor (first request, rewind/edit-resend), a whole-context rough estimate over
+threshold **waits one request** for provider evidence
+(`should_defer_preflight_to_real_usage`). This includes estimates at or above the
+entire context window: estimate magnitude does not prove that a request will fail.
+After a model switch, old usage is cleared and the new provider adjudicates the
+first request too; a genuinely oversized request can incur one rejected request
+before reactive recovery.
+
+The wait is not a disable. Once a response omits usage, the existing heuristic
+fallback remains available; real usage already over threshold and provider-proven
+overflow still allow compression. A post-compaction latch waits for one response
+and is consumed even if that response omits usage. Recovery remains bounded by the
+compression attempt budget and no-progress guards, not an indefinite resend loop.
+
+This is **not an exact-count-only policy**, nor closure of #104462's literal
+never-estimate acceptance. The following policies remain unchanged:
+
+- An anchor includes the provider's prompt and completion tokens plus a **rough
+  appended-message delta** (the first appended assistant is already covered by
+  completion usage). A large new tool result can therefore still cross a threshold
+  on an estimated delta. Boundary fingerprint matching does not fingerprint the
+  whole prefix, model, tools, or system prompt.
+- Opt-in idle compaction uses its own floor/cooldown and can act on unanchored
+  pressure; it does not share the threshold gate's one-request wait.
+- Pre-agent gateway hygiene retains its rough-history fallback and hard-message
+  safety valve. The replay harness's `gateway` shape reloads transcript dictionaries;
+  it does **not** exercise that separate hygiene policy.
+- Post-tool usage-less fallback, micro-compaction, summary/tail sizing, pruning and
+  overflow progress checks still use local estimates. Native compaction keeps its
+  provider-specific ownership and checkpoint latch.
+
+Provider count endpoints remain deferred. Eliminating these remaining estimates
+requires an explicit policy decision: accept the documented liveness fallbacks,
+or replace them with provider evidence while defining behavior for providers that
+never return usage. Simply disabling all unanchored maintenance is not equivalent.
+
+`evals/token_accounting/replay_gates.py` covers below-window and past-window
+inflation, real-over-threshold controls, reload/restore anchors, and local HTTP
+overflow/usage-less recovery with real compression but fixed local summary text.
+These are scripted control-flow checks, not vendor tokenizer or billing evidence.
 
 Opaque provider blobs (`encrypted_content` on Codex reasoning / compaction
 items) contribute 0 to every local estimate; only real usage ever prices them.