"""Regression for #80449 — an oversized in-progress turn must stay compressible. When one turn (opening user message + many individually small tool groups) grows past the protected-tail soft ceiling, anchoring the cut back to the turn-opening request kept the whole turn verbatim: compaction re-fired with an empty summarizable window and the session sat over threshold. The cut must instead land on a tool-group-aligned mid-turn boundary, and the active request must survive the handoff. """ from __future__ import annotations from unittest.mock import patch import pytest from agent.context_compressor import ( COMPRESSED_SUMMARY_METADATA_KEY, ContextCompressor, _estimate_msg_budget_tokens, ) _ACTIVE_REQUEST = "Inspect every shard and preserve the active request exactly." _TOKEN_BUDGET = 250 def _make_compressor(**overrides) -> ContextCompressor: """Compressor with an explicit small tail budget so the ceiling is reachable in a short transcript.""" kwargs = { "model": "test/model", "threshold_percent": 0.85, "protect_first_n": 0, "protect_last_n": 3, "quiet_mode": True, } kwargs.update(overrides) with patch( "agent.context_compressor.get_model_context_length", return_value=100_000, ): instance = ContextCompressor(**kwargs) _ = instance.context_length instance.tail_token_budget = _TOKEN_BUDGET return instance @pytest.fixture() def compressor() -> ContextCompressor: return _make_compressor() def _tool_group(index: int) -> list[dict]: call_id = f"call_{index}" return [ { "role": "assistant", "content": "", "tool_calls": [ { "id": call_id, "type": "function", "function": { "name": "inspect_shard", # Large enough for the complete turn to exceed the # ceiling, but below the phase-1 argument-prune limit. "arguments": "x" * 440, }, } ], }, { "role": "tool", "tool_call_id": call_id, # Keep each result below the phase-1 result-prune floor. The bug is # aggregate turn size, not one individually oversized result. "content": f"result-{index}:" + "r" * 110, }, ] def _oversized_active_turn() -> list[dict]: messages = [ {"role": "system", "content": "system"}, {"role": "user", "content": "older request"}, {"role": "assistant", "content": "older request completed"}, {"role": "user", "content": _ACTIVE_REQUEST}, ] for index in range(10): messages.extend(_tool_group(index)) return messages def _assert_tool_pairs_are_complete(messages: list[dict]) -> None: call_ids = { call["id"] for message in messages for call in message.get("tool_calls") or [] } result_ids = { message["tool_call_id"] for message in messages if message.get("role") == "tool" } assert call_ids == result_ids def test_oversized_active_turn_uses_a_mid_turn_tool_boundary( compressor: ContextCompressor, ) -> None: messages = _oversized_active_turn() head_end = compressor._protect_head_size(messages) cut = compressor._find_tail_cut_by_tokens( messages, head_end, token_budget=_TOKEN_BUDGET, ) active_user_idx = next( index for index, message in enumerate(messages) if message.get("content") == _ACTIVE_REQUEST ) tail_tokens = sum(_estimate_msg_budget_tokens(msg) for msg in messages[cut:]) assert cut > active_user_idx # Tool-group alignment may retain one additional indivisible group beyond # the scalar ceiling. It must not retain the whole active turn. max_group_tokens = max( sum(_estimate_msg_budget_tokens(msg) for msg in _tool_group(index)) for index in range(10) ) assert tail_tokens <= int(_TOKEN_BUDGET * 1.5) + max_group_tokens assert tail_tokens < sum( _estimate_msg_budget_tokens(msg) for msg in messages[active_user_idx:] ) assert messages[cut]["role"] == "assistant" _assert_tool_pairs_are_complete(messages[head_end:cut]) _assert_tool_pairs_are_complete(messages[cut:]) def test_full_compaction_preserves_active_request_and_tool_pairs( compressor: ContextCompressor, ) -> None: messages = _oversized_active_turn() # Exercise the deterministic handoff too: even when the summary model is # unavailable, splitting the turn must not lose the opening request. with patch.object(compressor, "_generate_summary", return_value=None): compressed = compressor.compress(messages, current_tokens=90_000) summary_rows = [ message for message in compressed if message.get(COMPRESSED_SUMMARY_METADATA_KEY) ] assert len(summary_rows) == 1 assert _ACTIVE_REQUEST in str(summary_rows[0].get("content")) assert sum( _ACTIVE_REQUEST in str(message.get("content")) for message in compressed ) == 1 assert len(compressed) < len(messages) _assert_tool_pairs_are_complete(compressed) def test_n_user_tail_guarantee_outranks_the_split() -> None: """compression.min_tail_user_messages is a user-facing promise (#70250). The oversized-turn exception must not void it: with N > 1 the N-user tail anchor wins even when one turn alone exceeds the soft ceiling. """ compressor = _make_compressor(protect_first_n=1, min_tail_user_messages=3) user_turns = ["first request", "second request", _ACTIVE_REQUEST] messages = [ {"role": "system", "content": "system"}, {"role": "user", "content": "earlier request"}, {"role": "assistant", "content": "earlier request completed"}, ] for text in user_turns: messages.append({"role": "user", "content": text}) messages.append({"role": "assistant", "content": f"{text} completed"}) for index in range(10): messages.extend(_tool_group(index)) cut = compressor._find_tail_cut_by_tokens( messages, compressor._protect_head_size(messages), token_budget=_TOKEN_BUDGET, ) tail = messages[cut:] assert [m["content"] for m in tail if m.get("role") == "user"] == user_turns def test_a_tail_that_fits_the_budget_still_anchors_the_active_request() -> None: """The exception is for a turn that overflows the budget, not for one that fits. With the whole transcript inside the tail budget, keeping the active request verbatim costs nothing, so the anchor must still hold and the exception must not fire just because the turn happens to be built from tool groups. """ compressor = _make_compressor() compressor.tail_token_budget = 10_000 messages = _oversized_active_turn() cut = compressor._find_tail_cut_by_tokens(messages, compressor._protect_head_size(messages)) active_user_idx = next( index for index, message in enumerate(messages) if message.get("content") == _ACTIVE_REQUEST ) assert cut <= active_user_idx, "active request must stay inside the protected tail" assert any( m.get("content") == _ACTIVE_REQUEST for m in messages[cut:] )