feat(agent): context size anchors on provider-reported usage — estimation shrinks to the last turn

Every provider response carries usage.prompt_tokens — exact ground truth
for the full request (system prompt + tool schemas + history). Context-size
checks now anchor on the last main-loop response's usage and estimate only
the messages appended since, instead of re-estimating the whole history
with chars/4 heuristics and flat 1500-token image costs. The estimate error
window shrinks from the entire conversation to one turn and self-corrects
at every response.

- agent/model_metadata.py: capture_usage_anchor() / anchored_context_tokens()
  with a structural base-message identity check that fails closed on any
  transcript rewrite.
- agent/conversation_loop.py: anchor captured at the single main-loop usage
  site (MoA uses pre-fold aggregator usage; advisor/aux calls never anchor);
  pre-API pressure check prefers the anchor.
- agent/turn_context.py: preflight compression estimate prefers the anchor.
- agent/context_breakdown.py: /context display prefers the anchor.
- Invalidation: compaction rewrite (conversation_compression), codex native
  compaction (codex_runtime), session reset/switch (run_agent), plus the
  fail-closed structural check for splices/micro-compaction.
- Usage-less responses keep the previous anchor; no anchor -> pure
  estimation fallback (first request of a session).

agent/agent_init.py

@@ -2953,6 +2953,12 @@ def init_agent(
     # Copilot x-initiator flag: first API call of a user turn sends "user" (#3040).
     agent._is_user_initiated_turn = False
 
+    # Usage-anchored context accounting (agent/model_metadata.py): the last
+    # main-loop provider response's exact usage + transcript snapshot. None
+    # until the first response with usage; invalidated on compaction and
+    # session switches so stale anchors can never suppress compression.
+    agent._usage_anchor = None
+
     # Cumulative token usage for the session
     agent.session_prompt_tokens = 0
     agent.session_completion_tokens = 0

agent/codex_runtime.py

@@ -307,6 +307,10 @@ def _record_codex_app_server_compaction(
             compressor.last_completion_tokens = 0
             compressor.awaiting_real_usage_after_compression = True
 
+    # Native compaction rewrote the provider-side context; the usage anchor's
+    # transcript snapshot no longer matches what will be sent. Invalidate it.
+    agent._usage_anchor = None
+
     agent._last_compaction_in_place = False
     try:
         if getattr(agent, "event_callback", None):

agent/context_breakdown.py

@@ -129,8 +129,20 @@ def compute_session_context_breakdown(
 
     comp = getattr(agent, "context_compressor", None)
     context_max = int(getattr(comp, "context_length", 0) or 0) if comp else 0
+    # Prefer the usage-anchored figure: provider-exact prompt+completion of
+    # the last response plus a delta estimate of anything appended since —
+    # fresher than the raw last_prompt_tokens (which lags messages appended
+    # after the response) and far more accurate than the heuristic total.
+    from agent.model_metadata import anchored_context_tokens
+
+    anchored_used = anchored_context_tokens(
+        messages or [], getattr(agent, "_usage_anchor", None)
+    )
     measured_used = int(getattr(comp, "last_prompt_tokens", 0) or 0) if comp else 0
-    context_used = measured_used if measured_used > 0 else estimated_total
+    if anchored_used is not None:
+        context_used = anchored_used
+    else:
+        context_used = measured_used if measured_used > 0 else estimated_total
     context_percent = (
         max(0, min(100, round(context_used / context_max * 100)))
         if context_max

agent/conversation_compression.py

@@ -4626,6 +4626,12 @@ def compress_context(
         agent.context_compressor.last_prompt_tokens = -1
         agent.context_compressor.last_completion_tokens = 0
         agent.context_compressor.awaiting_real_usage_after_compression = True
+        # Compaction rewrote the transcript, so the usage anchor's base
+        # message-list snapshot no longer describes what will be sent —
+        # invalidate it. Context checks fall back to full estimation until
+        # the next response with usage re-anchors (its structural id/index
+        # check would also fail closed, but explicit is safer).
+        agent._usage_anchor = None
         # Arm the effectiveness verdict only after a completed rewrite crosses
         # the full compaction boundary. Exceptions, aborts, and no-op attempts
         # leave this false, so unrelated later usage cannot be charged to an

agent/conversation_loop.py

@@ -72,6 +72,8 @@ _STALE_MARKER_RE = re.compile(r"^\[[A-Za-z_][A-Za-z0-9_.-]*\]$")
 from agent.model_metadata import (
     MINIMUM_CONTEXT_LENGTH,
     _estimate_tools_tokens_rough,
+    anchored_context_tokens,
+    capture_usage_anchor,
     estimate_messages_tokens_rough,
     estimate_request_tokens_rough,
     get_context_length_from_provider_error,
@@ -2593,6 +2595,18 @@ def run_conversation(
         request_pressure_tokens = approx_tokens + (
             _estimate_tools_tokens_rough(agent.tools) if agent.tools else 0
         )
+        # Usage-anchored override: when the last provider response's exact
+        # usage is still valid for the durable transcript, replace the
+        # whole-history heuristic with anchor + delta-estimate. The anchor's
+        # prompt_tokens already includes system prompt AND tool schemas as
+        # the provider counted them, so no tools add-on is needed. Falls
+        # back to the rough figures above when the anchor is stale/missing
+        # (first request, post-compaction, usage-less providers).
+        _anchored_pressure = anchored_context_tokens(
+            messages, getattr(agent, "_usage_anchor", None)
+        )
+        if _anchored_pressure is not None:
+            request_pressure_tokens = _anchored_pressure
         total_chars = approx_tokens * 4
         # Stash this request's rough estimate so update_from_response() can
         # pair it with the provider's real prompt count — the (rough, real)
@@ -4185,6 +4199,25 @@ def run_conversation(
                         )
                     )
                     agent.context_compressor.update_from_response(usage_dict)
+                    # Usage-anchored context accounting: snapshot this
+                    # response's exact provider-reported usage against the
+                    # durable transcript. Later context-size checks anchor on
+                    # this and estimate only the messages appended since,
+                    # instead of re-estimating the whole history with
+                    # heuristics. Main-loop responses ONLY — MoA advisor and
+                    # auxiliary calls never reach this site, so they cannot
+                    # pollute the anchor. A usage-less response leaves the
+                    # previous anchor in place (still valid for its base).
+                    # MoA note: use the pre-fold aggregator usage — the folded
+                    # canonical figure adds advisor fan-out tokens that were
+                    # never part of THIS conversation's prompt.
+                    _new_anchor = capture_usage_anchor(
+                        aggregator_usage.prompt_tokens,
+                        aggregator_usage.output_tokens,
+                        messages,
+                    )
+                    if _new_anchor is not None:
+                        agent._usage_anchor = _new_anchor
                     _compression_threshold = int(
                         getattr(agent.context_compressor, "threshold_tokens", 0)
                         or 0

agent/model_metadata.py

@@ -3764,6 +3764,91 @@ def estimate_request_tokens_rough(
     return total
 
 
+# --- Usage-anchored context accounting ------------------------------------
+#
+# Provider responses carry ``usage.prompt_tokens`` — EXACT ground truth for
+# everything sent on that request (system prompt + tool schemas + full
+# history). Re-estimating the whole conversation with chars/4 heuristics on
+# every context-size check compounds error over the entire transcript (flat
+# 1500-token images, CJK density, provider replay blobs). Anchoring on the
+# last real usage shrinks the estimation window to the messages appended
+# since that response; the error self-corrects at every new response.
+#
+# The anchor is a plain dict so callers can store it anywhere:
+#   prompt_tokens / completion_tokens — provider-reported usage at capture.
+#   base_count — len(messages) at capture time (the assistant reply for the
+#       captured response is NOT yet appended at the capture site; when it
+#       appears at index base_count its cost is covered by completion_tokens,
+#       so the delta walk skips it).
+#   base_last_id / base_last_role — identity fingerprint of the last message
+#       at capture time. Compaction, splices, and history rewrites shift or
+#       replace that element, failing the check and falling back to full
+#       estimation. Belt-and-braces on top of explicit invalidation.
+
+
+def capture_usage_anchor(
+    prompt_tokens: Any,
+    completion_tokens: Any,
+    messages: List[Dict[str, Any]],
+) -> Optional[Dict[str, Any]]:
+    """Build a usage anchor from provider-reported usage, or None."""
+    try:
+        pt = int(prompt_tokens or 0)
+        ct = int(completion_tokens or 0)
+    except (TypeError, ValueError):
+        return None
+    if pt <= 0 or not isinstance(messages, list):
+        # No usable usage (some OpenAI-compatible endpoints omit it) — the
+        # caller keeps whatever anchor it had, or stays on pure estimation.
+        return None
+    base_count = len(messages)
+    last = messages[-1] if base_count else None
+    return {
+        "prompt_tokens": pt,
+        "completion_tokens": max(0, ct),
+        "base_count": base_count,
+        "base_last_id": id(last) if last is not None else None,
+        "base_last_role": last.get("role") if isinstance(last, dict) else None,
+    }
+
+
+def anchored_context_tokens(
+    messages: List[Dict[str, Any]],
+    anchor: Optional[Dict[str, Any]],
+) -> Optional[int]:
+    """Context size anchored on the last provider-reported usage.
+
+    Returns ``prompt_tokens + completion_tokens`` of the anchored response
+    plus a rough estimate of ONLY the messages appended since — or ``None``
+    when the anchor is missing or stale (caller falls back to full
+    estimation). The assistant reply produced by the anchored response
+    (first appended message after the base) is skipped: its cost is already
+    counted exactly by ``completion_tokens``.
+    """
+    if not isinstance(anchor, dict) or not isinstance(messages, list):
+        return None
+    base_count = anchor.get("base_count") or 0
+    if base_count <= 0 or len(messages) < base_count:
+        return None
+    base_msg = messages[base_count - 1]
+    if id(base_msg) != anchor.get("base_last_id"):
+        return None
+    base_role = base_msg.get("role") if isinstance(base_msg, dict) else None
+    if base_role != anchor.get("base_last_role"):
+        return None
+    total = int(anchor["prompt_tokens"]) + int(anchor.get("completion_tokens") or 0)
+    delta = messages[base_count:]
+    if delta:
+        first = delta[0]
+        if isinstance(first, dict) and first.get("role") == "assistant":
+            # The anchored response's own reply — already counted exactly by
+            # completion_tokens above.
+            delta = delta[1:]
+    if delta:
+        total += estimate_messages_tokens_rough(delta)
+    return total
+
+
 # NOTE: tool schemas can be large. Avoid repeated `str(tools)` conversions,
 # which are CPU-heavy and can stall GUI event loops under GIL pressure.
 #

agent/turn_context.py

@@ -44,6 +44,7 @@ from agent.memory_manager import build_memory_context_block
 from agent.memory_provider import is_trivial_prompt
 from agent.message_metadata import append_message, stamp_message_timestamp
 from agent.model_metadata import (
+    anchored_context_tokens,
     estimate_messages_tokens_rough,
     estimate_request_tokens_rough,
 )
@@ -62,7 +63,18 @@ def _preflight_request_tokens(
     count the checkpoint-pruned wire payload rather than the full durable
     transcript. Auxiliary compression still uses the generic estimator
     (``native_compaction_eligible=False``).
+
+    Usage-anchored fast path: when a provider-reported usage anchor is
+    valid for ``messages`` (see ``anchored_context_tokens``), it already
+    covers system prompt + tool schemas + full history EXACTLY as the
+    provider counted them, with estimation confined to the messages
+    appended since that response. Prefer it over every heuristic.
     """
+    anchored = anchored_context_tokens(
+        messages, getattr(agent, "_usage_anchor", None)
+    )
+    if anchored is not None:
+        return anchored
     tools = getattr(agent, "tools", None) or None
     try:
         from agent.codex_responses_adapter import (

run_agent.py

@@ -798,6 +798,11 @@ class AIAgent:
         self.session_estimated_cost_usd = 0.0
         self.session_cost_status = "unknown"
         self.session_cost_source = "none"
+
+        # Session boundary: the usage anchor describes the OLD session's
+        # transcript — a fresh/branched/resumed session must fall back to
+        # full estimation until its first provider response re-anchors.
+        self._usage_anchor = None
         
         # Turn counter (added after reset_session_state was first written — #2635)
         self._user_turn_count = 0

tests/agent/test_usage_anchor.py

@@ -0,0 +1,212 @@
+"""Usage-anchored context accounting (agent/model_metadata.py).
+
+Context-size checks anchor on the provider-reported ``usage.prompt_tokens``
+of the last main-loop response and estimate ONLY the messages appended
+since. These tests cover:
+
+  * anchor + delta arithmetic (exact base, small estimated delta);
+  * the image-heavy divergence the anchor eliminates (flat 1500/image
+    heuristic vs provider truth);
+  * fallback to full estimation when no anchor exists (first request,
+    usage-less providers);
+  * invalidation when compaction rewrites the transcript (structural
+    id/index check fails closed) and on explicit reset sites;
+  * the preflight consumer (_preflight_request_tokens) preferring the
+    anchor, plus a sabotage check proving the anchored path (not the
+    heuristic) produces the number.
+"""
+
+from types import SimpleNamespace
+
+import pytest
+
+from agent.model_metadata import (
+    anchored_context_tokens,
+    capture_usage_anchor,
+    estimate_messages_tokens_rough,
+)
+from agent.turn_context import _preflight_request_tokens
+
+
+def _msg(role, content):
+    return {"role": role, "content": content}
+
+
+def _image_msg():
+    # ~40KB of fake base64 — the rough estimator charges a flat 1500
+    # tokens per image part regardless of true provider accounting.
+    return {
+        "role": "user",
+        "content": [
+            {"type": "text", "text": "look at this"},
+            {
+                "type": "image_url",
+                "image_url": {"url": "data:image/png;base64," + "A" * 40000},
+            },
+        ],
+    }
+
+
+def _history_with_images(n_images=10):
+    msgs = [_msg("user", "start")]
+    for i in range(n_images):
+        msgs.append(_msg("assistant", f"taking screenshot {i}"))
+        msgs.append(_image_msg())
+    msgs.append(_msg("assistant", "done looking"))
+    return msgs
+
+
+class TestAnchorArithmetic:
+    def test_anchor_plus_small_delta(self):
+        messages = _history_with_images(10)
+        anchor = capture_usage_anchor(50_000, 250, messages)
+        assert anchor is not None
+        assert anchor["prompt_tokens"] == 50_000
+        assert anchor["base_count"] == len(messages)
+
+        # Main loop appends the response's own assistant reply, then a tool
+        # result / user follow-up.
+        messages.append(_msg("assistant", "the anchored reply itself"))
+        messages.append(_msg("user", "short follow-up"))
+
+        anchored = anchored_context_tokens(messages, anchor)
+        assert anchored is not None
+        # Exact base + completion; the assistant reply at base_count is
+        # covered by completion_tokens, so only the follow-up is estimated.
+        delta_est = estimate_messages_tokens_rough([messages[-1]])
+        assert anchored == 50_000 + 250 + delta_est
+        assert delta_est < 50  # the estimated window is one small message
+
+    def test_image_heavy_divergence_eliminated(self):
+        messages = _history_with_images(10)
+        # Provider ground truth: say the real prompt was 12,000 tokens
+        # (providers often charge far less than 1500/image, or the images
+        # were downscaled). The heuristic charges 10 * 1500 + text.
+        anchor = capture_usage_anchor(12_000, 100, messages)
+        messages.append(_msg("assistant", "reply"))
+        messages.append(_msg("user", "ok"))
+
+        rough = estimate_messages_tokens_rough(messages)
+        anchored = anchored_context_tokens(messages, anchor)
+        assert rough >= 15_000  # flat 1500 x 10 images dominates
+        assert anchored is not None
+        assert anchored < 12_200
+        # The whole-history heuristic diverges by thousands of tokens;
+        # the anchored figure is provider truth + a tiny delta.
+        assert rough - anchored > 2_800
+
+    def test_no_usage_returns_none(self):
+        messages = [_msg("user", "hi")]
+        assert capture_usage_anchor(0, 0, messages) is None
+        assert capture_usage_anchor(None, None, messages) is None
+        assert capture_usage_anchor("garbage", 1, messages) is None
+
+    def test_missing_anchor_falls_back(self):
+        messages = _history_with_images(2)
+        assert anchored_context_tokens(messages, None) is None
+
+
+class TestAnchorInvalidation:
+    def test_compaction_rewrite_fails_closed(self):
+        messages = _history_with_images(4)
+        anchor = capture_usage_anchor(30_000, 50, messages)
+        # Compaction: transcript rebuilt as a new, shorter list.
+        compacted = [
+            _msg("user", "summary handoff"),
+            _msg("assistant", "[compressed summary]"),
+        ]
+        assert anchored_context_tokens(compacted, anchor) is None
+
+    def test_middle_splice_shifts_base_and_fails_closed(self):
+        messages = _history_with_images(4)
+        anchor = capture_usage_anchor(30_000, 50, messages)
+        # Micro-compact style splice: middle window replaced by one marker.
+        spliced = messages[:1] + [_msg("assistant", "[marker]")] + messages[5:]
+        assert anchored_context_tokens(spliced, anchor) is None
+
+    def test_same_length_different_objects_fails_closed(self):
+        messages = _history_with_images(4)
+        anchor = capture_usage_anchor(30_000, 50, messages)
+        rebuilt = [dict(m) for m in messages]  # fresh dicts, same values
+        assert anchored_context_tokens(rebuilt, anchor) is None
+
+    def test_explicit_invalidation_sites(self):
+        """The compaction + session-reset sites null agent._usage_anchor."""
+        import inspect
+
+        import agent.conversation_compression as cc
+        import agent.codex_runtime as cr
+        import run_agent
+
+        assert "agent._usage_anchor = None" in inspect.getsource(cc)
+        assert "agent._usage_anchor = None" in inspect.getsource(cr)
+        assert "self._usage_anchor = None" in inspect.getsource(
+            run_agent.AIAgent.reset_session_state
+        )
+
+
+class TestPreflightConsumer:
+    def _agent(self, anchor):
+        return SimpleNamespace(
+            _usage_anchor=anchor,
+            tools=None,
+            api_mode="",
+            provider="openai",
+        )
+
+    def test_preflight_prefers_anchor(self):
+        messages = _history_with_images(10)
+        anchor = capture_usage_anchor(50_000, 250, messages)
+        messages.append(_msg("assistant", "reply"))
+        messages.append(_msg("user", "ok"))
+        agent = self._agent(anchor)
+
+        got = _preflight_request_tokens(agent, messages, "SYSTEM PROMPT " * 500)
+        expected = anchored_context_tokens(messages, anchor)
+        assert got == expected
+        # The anchored figure ignores the (already-counted) system prompt
+        # text passed in — provider usage includes the real one.
+        assert 50_000 < got < 50_500
+
+    def test_preflight_falls_back_without_anchor(self):
+        messages = _history_with_images(3)
+        agent = self._agent(None)
+        got = _preflight_request_tokens(agent, messages, "sys")
+        # Pure heuristic: flat image cost dominates.
+        assert got >= 4_500
+
+    def test_sabotage_disabling_anchor_changes_result(self):
+        """Prove the anchored path produced the number: with the anchor
+        removed (the sabotage), the same inputs yield the heuristic figure,
+        which diverges by thousands of tokens on an image-heavy history."""
+        messages = _history_with_images(10)
+        anchor = capture_usage_anchor(12_000, 100, messages)
+        messages.append(_msg("assistant", "reply"))
+        messages.append(_msg("user", "ok"))
+
+        anchored_result = _preflight_request_tokens(
+            self._agent(anchor), messages, ""
+        )
+        sabotaged_result = _preflight_request_tokens(
+            self._agent(None), messages, ""
+        )
+        assert sabotaged_result - anchored_result > 2_800
+
+
+class TestCompressionTriggerUsesAnchor:
+    def test_threshold_decision_flips_with_anchor(self):
+        """An image-heavy history the heuristic pushes over a 15K threshold
+        stays under it when the provider reports the real 12K prompt."""
+        messages = _history_with_images(10)
+        anchor = capture_usage_anchor(12_000, 100, messages)
+        messages.append(_msg("assistant", "reply"))
+
+        threshold = 15_000
+        heuristic = estimate_messages_tokens_rough(messages)
+        anchored = anchored_context_tokens(messages, anchor)
+        assert heuristic >= threshold  # old behavior: spurious compression
+        assert anchored is not None and anchored < threshold
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-v"]))

tests/run_agent/test_compression_budget_rearm.py

@@ -71,9 +71,9 @@ def _tool_definition():
 
 @pytest.mark.parametrize(
     ("prompt_tokens", "expected_compactions", "provider_recovery"),
-    [(50, 2, False), (150, 1, False), (50, 2, True)],
+    [(50, 1, False), (150, 1, False), (50, 2, True)],
     ids=[
-        "pressure-cleared-rearms",
+        "pressure-cleared-anchored-no-recompaction",
         "pressure-still-high-stays-capped",
         "pressure-cleared-rearms-after-provider-recovery",
     ],
@@ -83,7 +83,17 @@ def test_pre_api_compression_budget_rearms_only_after_pressure_clears(
     expected_compactions: int,
     provider_recovery: bool,
 ):
-    """Only provider-confirmed headroom starts a new pressure episode."""
+    """Only provider-confirmed headroom starts a new pressure episode.
+
+    Usage-anchored accounting update: once the provider reports
+    ``prompt_tokens=50`` for the full transcript, later pre-API checks anchor
+    on that real reading plus a delta estimate of the few appended messages —
+    the scripted whole-history rough estimate (200) no longer drives the
+    decision, so the pressure-cleared case performs exactly ONE compaction
+    (the pre-anchor one). The budget-rearm mechanics remain covered by the
+    provider-recovery variant, whose first response carries no usage (no
+    anchor) and therefore still compacts on the rough estimate.
+    """
     with (
         patch("run_agent.get_tool_definitions", return_value=[_tool_definition()]),
         patch("run_agent.check_toolset_requirements", return_value={}),