fix(cli): context meter no longer sawtooths on reasoning models — show durable transcript, not last-request replay

On reasoning models a long tool loop replays the current turn's thinking +
scaffolding on every request, so the LAST request's prompt_tokens can exceed
the durable transcript by hundreds of K — all of which evaporates at the turn
boundary. The status bar and /context breakdown rendered that raw figure, so
users watched 'context' jump (e.g.) 850K -> 600K across a turn boundary and
read it as a broken compaction.

- conversation_loop: capture a turn-base usage anchor from the turn's FIRST
  provider response (api_call_count == 1), where replay is minimal.
- anchored_context_tokens: new charge_stale_thinking kwarg forwarded to the
  delta estimate (stale reasoning excluded on all but the newest assistant
  message).
- cli status snapshot + context_breakdown: prefer the turn-base anchored
  figure; fall back to last-response anchor / raw last_prompt_tokens.
- All _usage_anchor invalidation sites also clear _turn_base_usage_anchor.

Display-only: compression trigger math keeps using real last-request usage
(the inflated request is what actually risks the window mid-loop).

agent/agent_init.py

@@ -3011,6 +3011,7 @@ def init_agent(
     # until the first response with usage; invalidated on compaction and
     # session switches so stale anchors can never suppress compression.
     agent._usage_anchor = None
+    agent._turn_base_usage_anchor = None
 
     # Cumulative token usage for the session
     agent.session_prompt_tokens = 0

agent/codex_runtime.py

@@ -310,6 +310,7 @@ def _record_codex_app_server_compaction(
     # 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._turn_base_usage_anchor = None
 
     agent._last_compaction_in_place = False
     try:

agent/context_breakdown.py

@@ -135,9 +135,20 @@ def compute_session_context_breakdown(
     # after the response) and far more accurate than the heuristic total.
     from agent.model_metadata import anchored_context_tokens
 
+    # Prefer the turn-base anchor (first response of the current turn): on
+    # reasoning models, later same-turn responses inflate prompt_tokens with
+    # replayed thinking that evaporates at the turn boundary, so anchoring on
+    # the LAST response makes the meter sawtooth. Fall back to the last-
+    # response anchor, then to measured/estimated figures.
     anchored_used = anchored_context_tokens(
-        messages or [], getattr(agent, "_usage_anchor", None)
+        messages or [],
+        getattr(agent, "_turn_base_usage_anchor", None),
+        charge_stale_thinking=False,
     )
+    if anchored_used is None:
+        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
     if anchored_used is not None:
         context_used = anchored_used

agent/conversation_compression.py

@@ -5355,6 +5355,7 @@ def compress_context(
         # the next response with usage re-anchors (its structural id/index
         # check would also fail closed, but explicit is safer).
         agent._usage_anchor = None
+        agent._turn_base_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

@@ -4486,6 +4486,20 @@ def run_conversation(
                     )
                     if _new_anchor is not None:
                         agent._usage_anchor = _new_anchor
+                        # Turn-base anchor for display surfaces: the FIRST
+                        # response of a turn carries minimal current-turn
+                        # reasoning replay, so its prompt_tokens approximate
+                        # the durable transcript cost (what the next turn
+                        # inherits). Later same-turn responses inflate
+                        # prompt_tokens with replayed thinking + tool
+                        # scaffolding that evaporates at the turn boundary —
+                        # anchoring the context meter here instead of on the
+                        # last response removes the end-of-turn sawtooth
+                        # (850K mid-loop -> 600K next turn) that users read
+                        # as a broken compaction. Display-only: compression
+                        # trigger math keeps using real last-request usage.
+                        if api_call_count == 1:
+                            agent._turn_base_usage_anchor = _new_anchor
                     _compression_threshold = int(
                         getattr(agent.context_compressor, "threshold_tokens", 0)
                         or 0

agent/model_metadata.py

@@ -3958,6 +3958,8 @@ def capture_usage_anchor(
 def anchored_context_tokens(
     messages: List[Dict[str, Any]],
     anchor: Optional[Dict[str, Any]],
+    *,
+    charge_stale_thinking: bool = True,
 ) -> Optional[int]:
     """Context size anchored on the last provider-reported usage.
 
@@ -3967,6 +3969,13 @@ def anchored_context_tokens(
     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``.
+
+    ``charge_stale_thinking`` is forwarded to the delta estimate — pass
+    ``False`` to exclude transient ``reasoning``/``reasoning_content`` text
+    on all but the newest assistant message in the delta (the durable-
+    transcript view used by display surfaces; see the turn-base anchor in
+    ``agent/conversation_loop.py``). Default ``True`` preserves the
+    conservative full charge for request-size callers.
     """
     if not isinstance(anchor, dict) or not isinstance(messages, list):
         return None
@@ -3988,7 +3997,9 @@ def anchored_context_tokens(
             # completion_tokens above.
             delta = delta[1:]
     if delta:
-        total += estimate_messages_tokens_rough(delta)
+        total += estimate_messages_tokens_rough(
+            delta, charge_stale_thinking=charge_stale_thinking
+        )
     return total
 
 

cli.py

@@ -6667,6 +6667,29 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
             context_tokens = getattr(compressor, "last_prompt_tokens", 0) or 0
             if context_tokens < 0:
                 context_tokens = 0
+            # Durable-transcript view: on reasoning models a long tool loop
+            # replays the current turn's thinking + scaffolding on every
+            # request, so the LAST request's prompt_tokens can exceed the
+            # durable transcript by hundreds of K — all of which evaporates
+            # at the turn boundary. Rendering that raw figure makes the bar
+            # sawtooth (e.g. 850K mid-turn -> 600K next turn) and reads as a
+            # broken compaction. Anchor the display on the turn's FIRST
+            # response (minimal replay) plus a delta estimate of messages
+            # appended since, excluding stale thinking. Display-only: the
+            # compression trigger keeps using real last-request usage.
+            try:
+                from agent.model_metadata import anchored_context_tokens
+
+                _msgs = getattr(agent, "_session_messages", None)
+                _anchored = anchored_context_tokens(
+                    _msgs if isinstance(_msgs, list) else [],
+                    getattr(agent, "_turn_base_usage_anchor", None),
+                    charge_stale_thinking=False,
+                )
+                if _anchored is not None and _anchored > 0:
+                    context_tokens = _anchored
+            except Exception:
+                pass
             context_length = getattr(compressor, "context_length", 0) or 0
             if context_length < 0:
                 context_length = 0

run_agent.py

@@ -876,6 +876,7 @@ class AIAgent:
         # transcript — a fresh/branched/resumed session must fall back to
         # full estimation until its first provider response re-anchors.
         self._usage_anchor = None
+        self._turn_base_usage_anchor = None
         
         # Turn counter (added after reset_session_state was first written — #2635)
         self._user_turn_count = 0

tests/agent/test_turn_base_display_anchor.py

@@ -0,0 +1,201 @@
+"""Turn-base display anchor: the context meter shows durable-transcript cost.
+
+On reasoning models a long tool loop replays the current turn's thinking +
+scaffolding on every request, so the LAST request's ``prompt_tokens`` can
+exceed the durable transcript by hundreds of K — all of which evaporates at
+the turn boundary. Display surfaces (CLI status bar, /context breakdown)
+therefore anchor on the turn's FIRST response (``_turn_base_usage_anchor``)
+plus a stale-thinking-free delta estimate, instead of the raw last-request
+figure. Compression trigger math is unchanged (real last-request usage).
+
+Covers:
+  * anchored_context_tokens(charge_stale_thinking=False) excludes stale
+    reasoning text in the delta while keeping the newest assistant turn;
+  * the CLI status snapshot prefers the turn-base anchored figure over
+    compressor.last_prompt_tokens and falls back cleanly without an anchor;
+  * compute_session_context_breakdown prefers the turn-base anchor over the
+    last-response anchor;
+  * invalidation sites clear _turn_base_usage_anchor alongside _usage_anchor.
+"""
+
+from types import SimpleNamespace
+
+from agent.model_metadata import (
+    anchored_context_tokens,
+    capture_usage_anchor,
+    estimate_messages_tokens_rough,
+)
+
+
+def _msg(role, content, **extra):
+    m = {"role": role, "content": content}
+    m.update(extra)
+    return m
+
+
+class TestChargeStaleThinkingKwarg:
+    def test_delta_excludes_stale_reasoning(self):
+        messages = [_msg("user", "start"), _msg("assistant", "base reply")]
+        anchor = capture_usage_anchor(10_000, 100, messages)
+        assert anchor is not None
+
+        # Simulate a tool loop appending reasoning-heavy assistant turns.
+        big_thinking = "deliberation " * 5_000  # ~65K chars ≈ 16K tokens
+        messages.append(_msg("assistant", "the anchored reply itself"))
+        messages.append(
+            _msg("assistant", "step one", reasoning_content=big_thinking)
+        )
+        messages.append(_msg("tool", "tool output", tool_call_id="c1"))
+        messages.append(
+            _msg("assistant", "step two", reasoning_content=big_thinking)
+        )
+
+        charged = anchored_context_tokens(messages, anchor)
+        uncharged = anchored_context_tokens(
+            messages, anchor, charge_stale_thinking=False
+        )
+        assert charged is not None and uncharged is not None
+        # Stale thinking on the non-newest assistant message is excluded;
+        # the newest assistant message keeps its reasoning charge.
+        one_thinking_tokens = estimate_messages_tokens_rough(
+            [_msg("assistant", "", reasoning_content=big_thinking)]
+        )
+        assert charged - uncharged >= one_thinking_tokens * 0.9
+        assert uncharged >= 10_000 + 100  # anchor base still counted exactly
+
+    def test_default_remains_full_charge(self):
+        messages = [_msg("user", "s"), _msg("assistant", "r")]
+        anchor = capture_usage_anchor(1_000, 10, messages)
+        messages.append(_msg("assistant", "reply"))
+        assert anchored_context_tokens(messages, anchor) == anchored_context_tokens(
+            messages, anchor, charge_stale_thinking=True
+        )
+
+
+class TestCliStatusSnapshotPrefersTurnBaseAnchor:
+    def _agent_with(self, last_prompt_tokens, messages, anchor):
+        compressor = SimpleNamespace(
+            last_prompt_tokens=last_prompt_tokens,
+            context_length=1_000_000,
+            compression_count=0,
+        )
+        return SimpleNamespace(
+            context_compressor=compressor,
+            _session_messages=messages,
+            _turn_base_usage_anchor=anchor,
+        )
+
+    def _snapshot_context_tokens(self, agent):
+        """Mirror the cli.py snapshot block's context_tokens resolution."""
+        compressor = agent.context_compressor
+        context_tokens = getattr(compressor, "last_prompt_tokens", 0) or 0
+        if context_tokens < 0:
+            context_tokens = 0
+        msgs = getattr(agent, "_session_messages", None)
+        anchored = anchored_context_tokens(
+            msgs if isinstance(msgs, list) else [],
+            getattr(agent, "_turn_base_usage_anchor", None),
+            charge_stale_thinking=False,
+        )
+        if anchored is not None and anchored > 0:
+            context_tokens = anchored
+        return context_tokens
+
+    def test_turn_base_anchor_wins_over_inflated_last_request(self):
+        messages = [_msg("user", "start"), _msg("assistant", "reply")]
+        anchor = capture_usage_anchor(600_000, 500, messages)
+        messages.append(_msg("assistant", "anchored reply"))
+        agent = self._agent_with(850_000, messages, anchor)
+        # Bar shows the durable figure, not the inflated last request.
+        tokens = self._snapshot_context_tokens(agent)
+        assert 600_000 <= tokens < 650_000
+
+    def test_fallback_without_anchor(self):
+        agent = self._agent_with(123_456, [_msg("user", "x")], None)
+        assert self._snapshot_context_tokens(agent) == 123_456
+
+    def test_stale_anchor_falls_back(self):
+        messages = [_msg("user", "start"), _msg("assistant", "reply")]
+        anchor = capture_usage_anchor(50_000, 10, messages)
+        agent = self._agent_with(77_000, [_msg("user", "rebuilt")], anchor)
+        # Compaction rebuilt the list: structural check fails, raw fallback.
+        assert self._snapshot_context_tokens(agent) == 77_000
+
+    def test_negative_sentinel_still_clamped(self):
+        agent = self._agent_with(-1, [], None)
+        assert self._snapshot_context_tokens(agent) == 0
+
+
+class TestContextBreakdownPrefersTurnBaseAnchor:
+    def test_breakdown_uses_turn_base_over_last_response(self, monkeypatch):
+        from agent import context_breakdown as cb
+
+        messages = [_msg("user", "start"), _msg("assistant", "reply")]
+        turn_base = capture_usage_anchor(400_000, 200, messages)
+        messages.append(_msg("assistant", "anchored reply"))
+        last_anchor = capture_usage_anchor(900_000, 50, messages)
+
+        agent = SimpleNamespace(
+            _usage_anchor=last_anchor,
+            _turn_base_usage_anchor=turn_base,
+            _memory_store=None,
+            tools=[],
+            model="test/model",
+            context_compressor=SimpleNamespace(
+                context_length=1_000_000, last_prompt_tokens=900_000
+            ),
+        )
+        monkeypatch.setattr(
+            "agent.system_prompt.build_system_prompt_parts",
+            lambda a: {"stable": "sys", "context": "", "volatile": ""},
+        )
+        payload = cb.compute_session_context_breakdown(agent, messages)
+        assert 400_000 <= payload["context_used"] < 450_000
+
+    def test_breakdown_falls_back_to_last_response_anchor(self, monkeypatch):
+        from agent import context_breakdown as cb
+
+        messages = [_msg("user", "start"), _msg("assistant", "reply")]
+        last_anchor = capture_usage_anchor(300_000, 50, messages)
+
+        agent = SimpleNamespace(
+            _usage_anchor=last_anchor,
+            _turn_base_usage_anchor=None,
+            _memory_store=None,
+            tools=[],
+            model="test/model",
+            context_compressor=SimpleNamespace(
+                context_length=1_000_000, last_prompt_tokens=1
+            ),
+        )
+        monkeypatch.setattr(
+            "agent.system_prompt.build_system_prompt_parts",
+            lambda a: {"stable": "sys", "context": "", "volatile": ""},
+        )
+        payload = cb.compute_session_context_breakdown(agent, messages)
+        assert payload["context_used"] >= 300_000
+
+
+class TestInvalidationSitesClearTurnBaseAnchor:
+    def test_compression_invalidation_clears_both(self):
+        import inspect
+        from agent import conversation_compression
+
+        src = inspect.getsource(conversation_compression)
+        block = src.split("agent._usage_anchor = None", 1)[1][:200]
+        assert "_turn_base_usage_anchor = None" in block
+
+    def test_codex_native_invalidation_clears_both(self):
+        import inspect
+        from agent import codex_runtime
+
+        src = inspect.getsource(codex_runtime)
+        block = src.split("agent._usage_anchor = None", 1)[1][:200]
+        assert "_turn_base_usage_anchor = None" in block
+
+    def test_agent_init_defines_turn_base_anchor(self):
+        import inspect
+        from agent import agent_init
+
+        src = inspect.getsource(agent_init)
+        assert "_turn_base_usage_anchor = None" in src