hermes-agent/tests/agent/test_estimator_parity.py

"""Regression tests for #84371 — compaction dead-loop on reasoning-heavy
codex_responses sessions.

Root cause: the compaction TRIGGER (``estimate_messages_tokens_rough`` /
``estimate_request_tokens_rough``) charged stale ``reasoning`` /
``reasoning_content`` on EVERY assistant message, while the tail-protection
walk (``_find_tail_cut_by_tokens``) charged them on the newest turn only
(#73624).  On a session where most tokens live in stale reasoning replay the
trigger fires above threshold while the walk protects everything —
``middle_window_tokens == 0`` / "insufficient progress" — and the same
compaction re-fires every turn, each attempt burning a full aux
summarization.

Wire truth (``_chat_messages_to_responses_input``): the codex_responses
input builder never reads the text thinking keys; reasoning continuity rides
the encrypted ``codex_reasoning_items`` sidecar, which both estimators charge
unconditionally.  So the TRIGGER overcounted reality and the fix makes the
trigger route-aware (charge stale thinking only when the route echoes it),
while echo-back chat-completions routes (DeepSeek/Kimi/MiMo thinking mode)
now charge it in the walk too — one policy per session shape, chosen by
``message_sanitization.stale_thinking_reaches_wire``.
"""

from agent.context_compressor import (
    ContextCompressor,
    _estimate_msg_budget_tokens,
)
from agent.message_sanitization import stale_thinking_reaches_wire
from agent.model_metadata import (
    estimate_messages_tokens_rough,
)

STALE_THINKING = "considering the next move carefully... " * 200  # ~2K tok

def _reasoning_heavy_session(n_turns: int = 40) -> list:
    """Transcript whose bulk is stale reasoning replay (the #84371 shape)."""
    msgs = [{"role": "system", "content": "You are Hermes."}]
    msgs.append({"role": "user", "content": "do the big task"})
    for i in range(n_turns):
        msgs.append(
            {
                "role": "assistant",
                "content": f"step {i}",
                "reasoning_content": STALE_THINKING,
                "tool_calls": [
                    {
                        "id": f"c{i}",
                        "type": "function",
                        "function": {"name": "t", "arguments": "{}"},
                    }
                ],
            }
        )
        msgs.append({"role": "tool", "tool_call_id": f"c{i}", "content": f"r{i}"})
    return msgs

class TestWireTruthPredicate:
    def test_codex_responses_never_ships_stale_thinking_text(self):
        assert stale_thinking_reaches_wire(
            "codex_responses", "deepseek", "deepseek-v4-flash", ""
        ) is False

    def test_chat_completions_echo_family_ships_it(self):
        # DeepSeek thinking mode over chat_completions echoes stored
        # reasoning_content back on every assistant turn.
        assert stale_thinking_reaches_wire(
            "", "deepseek", "deepseek-reasoner", "https://api.deepseek.com"
        ) is True

    def test_strict_chat_completions_strips_it(self):
        assert stale_thinking_reaches_wire(
            "", "mistral", "mistral-large", "https://api.mistral.ai"
        ) is False

class TestEstimatorParity:
    """Trigger-fires must imply the walk finds a compactable middle."""

    def test_trigger_fires_implies_walk_finds_middle(self):
        msgs = _reasoning_heavy_session()
        wire = stale_thinking_reaches_wire(
            "codex_responses", "deepseek", "deepseek-v4-flash", ""
        )
        trigger = estimate_messages_tokens_rough(
            msgs, charge_stale_thinking=wire
        )

        cc = ContextCompressor(
            model="deepseek-v4-flash",
            provider="deepseek",
            api_mode="codex_responses",
            quiet_mode=True,
            config_context_length=200_000,
        )
        # THE RELATION, not literals: whenever the route-aware trigger says
        # the session is over threshold, the tail walk must leave a real
        # middle region so the fired compaction can actually make progress.
        if trigger >= cc.threshold_tokens:
            start = cc._protect_head_size(msgs)
            end = cc._find_tail_cut_by_tokens(msgs, start)
            assert end > start, (
                "trigger fired but the tail walk protected everything — "
                "the #84371 dead-loop shape"
            )
            middle_tokens = estimate_messages_tokens_rough(msgs[start:end])
            assert middle_tokens > 0

    def test_route_aware_trigger_matches_walk_size_class(self):
        """On codex_responses the trigger no longer counts stale thinking
        the walk excludes: both figures land in the same size class."""
        msgs = _reasoning_heavy_session()
        legacy = estimate_messages_tokens_rough(msgs)
        route_aware = estimate_messages_tokens_rough(
            msgs, charge_stale_thinking=False
        )
        from agent.context_compressor import _last_assistant_index

        newest = _last_assistant_index(msgs)
        walk = sum(
            _estimate_msg_budget_tokens(m, charge_stale_thinking=(i == newest))
            for i, m in enumerate(msgs)
        )
        # Stale thinking dominates this transcript, so the legacy figure is
        # several times the walk's; the route-aware figure must not be.
        assert legacy > 3 * walk
        assert route_aware < 2 * walk

    def test_newest_turn_thinking_still_charged(self):
        msgs = _reasoning_heavy_session(n_turns=2)
        stripped = estimate_messages_tokens_rough(
            msgs, charge_stale_thinking=False
        )
        no_thinking = estimate_messages_tokens_rough(
            [
                {k: v for k, v in m.items()
                 if k not in ("reasoning", "reasoning_content")}
                for m in msgs
            ]
        )
        # The newest assistant turn's thinking survives the stale strip.
        assert stripped > no_thinking

    def test_walk_charges_stale_thinking_on_echo_route(self):
        """Echo-back chat_completions route: the walk now charges stale
        thinking on every turn, matching the trigger's full charge; the
        codex_responses route keeps newest-turn-only.  Assert the per-route
        charge policy directly — for each route, the walk's per-message sum
        must land in the same size class as that route's trigger estimate."""
        msgs = _reasoning_heavy_session()
        from agent.context_compressor import _last_assistant_index

        cc_echo = ContextCompressor(
            model="deepseek-reasoner",
            provider="deepseek",
            api_mode="",
            base_url="https://api.deepseek.com",
            quiet_mode=True,
            config_context_length=200_000,
        )
        cc_codex = ContextCompressor(
            model="deepseek-v4-flash",
            provider="deepseek",
            api_mode="codex_responses",
            quiet_mode=True,
            config_context_length=200_000,
        )
        assert cc_echo._stale_thinking_on_wire() is True
        assert cc_codex._stale_thinking_on_wire() is False

        newest = _last_assistant_index(msgs)

        def walk_sum(cc):
            charge_all = cc._stale_thinking_on_wire()
            return sum(
                _estimate_msg_budget_tokens(
                    m, charge_stale_thinking=(charge_all or i == newest)
                )
                for i, m in enumerate(msgs)
            )

        trigger_echo = estimate_messages_tokens_rough(
            msgs, charge_stale_thinking=True
        )
        trigger_codex = estimate_messages_tokens_rough(
            msgs, charge_stale_thinking=False
        )
        walk_echo = walk_sum(cc_echo)
        walk_codex = walk_sum(cc_codex)

        # Per route, trigger and walk agree within a small rough-estimator
        # factor — the pre-fix codex disagreement was >3x.
        assert walk_echo <= trigger_echo * 2 and trigger_echo <= walk_echo * 2
        assert walk_codex <= trigger_codex * 2 and trigger_codex <= walk_codex * 2
        # And the echo route genuinely charges the stale thinking bulk.
        assert walk_echo > 3 * walk_codex

class TestReasoningDoubleCount:
    """``reasoning`` and ``reasoning_content`` carrying the same text must be
    charged once — the wire ships at most one of them."""

    def test_trigger_counts_identical_pair_once(self):
        base = [{"role": "assistant", "content": "x"}]
        rc_only = [{"role": "assistant", "content": "x",
                    "reasoning_content": "y" * 4000}]
        both = [{"role": "assistant", "content": "x",
                 "reasoning": "y" * 4000, "reasoning_content": "y" * 4000}]
        e0 = estimate_messages_tokens_rough(base)
        e1 = estimate_messages_tokens_rough(rc_only)
        e2 = estimate_messages_tokens_rough(both)
        # Adding a duplicate `reasoning` key must not (materially) grow the
        # estimate: the increment over rc_only stays far below a second copy.
        assert (e2 - e0) < 1.5 * (e1 - e0)

    def test_walk_counts_identical_pair_once(self):
        base = {"role": "assistant", "content": "x"}
        rc_only = {"role": "assistant", "content": "x",
                   "reasoning_content": "y" * 4000}
        both = {"role": "assistant", "content": "x",
                "reasoning": "y" * 4000, "reasoning_content": "y" * 4000}
        w0 = _estimate_msg_budget_tokens(base, charge_stale_thinking=True)
        w1 = _estimate_msg_budget_tokens(rc_only, charge_stale_thinking=True)
        w2 = _estimate_msg_budget_tokens(both, charge_stale_thinking=True)
        assert (w2 - w0) < 1.5 * (w1 - w0)

    def test_reasoning_alone_still_charged(self):
        """No reasoning_content to displace it → `reasoning` is the
        promotion proxy and must stay counted."""
        base = [{"role": "assistant", "content": "x"}]
        r_only = [{"role": "assistant", "content": "x",
                   "reasoning": "y" * 4000}]
        assert (
            estimate_messages_tokens_rough(r_only)
            > estimate_messages_tokens_rough(base) + 500
        )
        w_base = _estimate_msg_budget_tokens(
            {"role": "assistant", "content": "x"}, charge_stale_thinking=True
        )
        w_r = _estimate_msg_budget_tokens(
            {"role": "assistant", "content": "x", "reasoning": "y" * 4000},
            charge_stale_thinking=True,
        )
        assert w_r > w_base + 500

    def test_pad_reasoning_content_does_not_displace(self):
        """A one-space echo pad is not real content; `reasoning` still
        carries the chargeable text."""
        msg = {"role": "assistant", "content": "x",
               "reasoning": "y" * 4000, "reasoning_content": " "}
        w = _estimate_msg_budget_tokens(msg, charge_stale_thinking=True)
        w_base = _estimate_msg_budget_tokens(
            {"role": "assistant", "content": "x", "reasoning_content": " "},
            charge_stale_thinking=True,
        )
        assert w > w_base + 500

class TestNoProgressDeadLoopBreaker:
    """A fired compaction that returns the transcript unchanged must arm the
    structural backoff so it cannot re-fire (and re-summarize) every turn."""

    def test_no_progress_arms_structural_backoff(self):
        import time

        cc = ContextCompressor(
            model="deepseek-v4-flash",
            provider="deepseek",
            api_mode="codex_responses",
            quiet_mode=True,
            config_context_length=200_000,
        )
        assert cc._structural_no_op_backoff_until <= time.monotonic()
        cc._record_structural_no_op("compaction returned unchanged")
        assert cc._structural_no_op_backoff_until > time.monotonic()
        # Over-threshold but blocked: no second aux summarization this turn.
        over = cc.threshold_tokens + 1
        assert cc.should_compress(over) is False
        reason = cc._compression_block_reason() or ""
        assert reason.startswith("structural_backoff")