diff --git a/agent/agent_init.py b/agent/agent_init.py index ec074c3324..72c1513925 100644 --- a/agent/agent_init.py +++ b/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 diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index ac272c43ea..1fdd2d153e 100644 --- a/agent/codex_runtime.py +++ b/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: diff --git a/agent/context_breakdown.py b/agent/context_breakdown.py index 628cbd038a..9f9c5da755 100644 --- a/agent/context_breakdown.py +++ b/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 diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index a4ab72f9c0..fd15799838 100644 --- a/agent/conversation_compression.py +++ b/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 diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 25dec15494..521be3a8c3 100644 --- a/agent/conversation_loop.py +++ b/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 diff --git a/agent/model_metadata.py b/agent/model_metadata.py index 4dc6de122f..4ed608fd24 100644 --- a/agent/model_metadata.py +++ b/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 diff --git a/cli.py b/cli.py index f60b2303e3..de464f1b41 100644 --- a/cli.py +++ b/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 diff --git a/run_agent.py b/run_agent.py index 18e0f070e3..eb904ccc71 100644 --- a/run_agent.py +++ b/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 diff --git a/tests/agent/test_turn_base_display_anchor.py b/tests/agent/test_turn_base_display_anchor.py new file mode 100644 index 0000000000..6c1a6f1183 --- /dev/null +++ b/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