From 10ced056763a61b3db2992c9ea880224b9322954 Mon Sep 17 00:00:00 2001 From: HexLab98 Date: Sun, 5 Jul 2026 16:58:55 +0700 Subject: [PATCH] test(compressor): pin the zero-user-turn compaction guard (#58753) Regression coverage for the kanban-worker crash where compression left a transcript with no user-role messages, triggering a non-retryable `400 No user query found in messages` from vLLM/Qwen. Exercises the real `compress()` path with the reporter's shape (no system prompt in the list, a re-compaction with the only user turn in the compressed middle) and asserts the output always keeps >=1 user turn, never introduces consecutive user roles, and leaves a surviving tail user message untouched. A source guardrail pins the guard so a future refactor cannot silently drop it. --- .../agent/test_compressor_zero_user_guard.py | 206 ++++++++++++++++++ 1 file changed, 206 insertions(+) create mode 100644 tests/agent/test_compressor_zero_user_guard.py diff --git a/tests/agent/test_compressor_zero_user_guard.py b/tests/agent/test_compressor_zero_user_guard.py new file mode 100644 index 000000000000..9e14a80925ac --- /dev/null +++ b/tests/agent/test_compressor_zero_user_guard.py @@ -0,0 +1,206 @@ +"""Regression coverage for #58753 — compression could drop the only +user-role message, leaving a transcript with ZERO user turns. + +The compressor already pins the handoff summary to ``role="user"`` when +the only protected head message is the system prompt (#52160). But that +guard keys off ``last_head_role == "system"``, which is only true when +the system prompt actually sits inside ``messages`` — the gateway +``/compress`` path. The main auto-compression path passes the transcript +WITHOUT the system prompt (it is prepended at request-build time, see +``conversation_loop`` — ``api_messages = [{"role": "system", ...}] + +api_messages``). There ``last_head_role`` defaults to ``"user"`` and the +summary is emitted as ``role="assistant"``. + +On a session whose only genuine user turn falls into the compressed +middle — the canonical shape being a ``hermes kanban`` worker seeded with +a single short ``"work kanban task "`` prompt followed by nothing but +assistant/tool turns — the compressed output then contains no user-role +message at all. OpenAI-compatible backends (vLLM/Qwen) reject such a +request with a non-retryable ``400 No user query found in messages``, +crashing the worker with no possible recovery (every resume replays the +same poisoned history). + +The fix generalises the #52160 guard: when NO user-role message survives +in the protected head or preserved tail, the summary MUST carry +``role="user"``. +""" + +from __future__ import annotations + +from unittest.mock import patch + +import pytest + + +@pytest.fixture() +def compressor(): + from agent.context_compressor import ContextCompressor + + with patch( + "agent.context_compressor.get_model_context_length", + return_value=100_000, + ): + c = ContextCompressor( + model="test/model", + threshold_percent=0.50, + protect_first_n=3, + protect_last_n=20, + quiet_mode=True, + ) + c.tail_token_budget = 40 + return c + + +def _tool_turns(start: int, n: int) -> list[dict]: + out: list[dict] = [] + for i in range(start, start + n): + out.append( + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": f"c{i}", + "function": {"name": "read_task", "arguments": "{}"}, + } + ], + } + ) + out.append({"role": "tool", "content": "x" * 300, "tool_call_id": f"c{i}"}) + return out + + +def _role_hist(messages: list[dict]) -> dict[str, int]: + hist: dict[str, int] = {} + for m in messages: + hist[m.get("role")] = hist.get(m.get("role"), 0) + 1 + return hist + + +class TestCompressAlwaysKeepsAUserTurn: + def test_kanban_worker_recompaction_keeps_user_turn(self, compressor): + """The exact #58753 shape: no system prompt in the list, a + re-compaction (``protect_first_n`` decayed to 0), and the only + user turn old enough to fall into the compressed middle. Before + the fix, the summary was emitted as ``assistant`` and the output + had zero user-role messages.""" + from agent.context_compressor import SUMMARY_PREFIX + + c = compressor + # A prior compaction has already happened → protect_first_n decays + # to 0 so compress_start lands at 0 (no protected head). + c.compression_count = 1 + # No system message: the main loop prepends it separately. + messages = [{"role": "user", "content": "work kanban task 42"}] + messages += _tool_turns(0, 12) + + mocked = f"{SUMMARY_PREFIX}\nrolled-up summary of the tool work" + with patch.object(c, "_generate_summary", return_value=mocked): + out = c.compress(messages, current_tokens=90_000) + + hist = _role_hist(out) + assert hist.get("user", 0) >= 1, ( + "REGRESSION (#58753): compression produced a transcript with " + f"zero user-role messages, which vLLM/Qwen reject with a " + f"non-retryable 400. Role histogram: {hist}" + ) + + def test_summary_pinned_to_user_when_no_user_survives(self, compressor): + """When the whole compressible region is assistant/tool and no + user message survives in head or tail, the inserted summary + itself must be the user turn.""" + from agent.context_compressor import ( + SUMMARY_PREFIX, + COMPRESSED_SUMMARY_METADATA_KEY, + ) + + c = compressor + c.compression_count = 1 + messages = [{"role": "user", "content": "work kanban task 7"}] + messages += _tool_turns(0, 12) + + mocked = f"{SUMMARY_PREFIX}\nsummary body" + with patch.object(c, "_generate_summary", return_value=mocked): + out = c.compress(messages, current_tokens=90_000) + + summary_rows = [m for m in out if m.get(COMPRESSED_SUMMARY_METADATA_KEY)] + assert len(summary_rows) == 1 + assert summary_rows[0].get("role") == "user", ( + "The handoff summary must carry role=user when it is the only " + "possible user turn in the compressed transcript (#58753)." + ) + + def test_no_consecutive_user_roles_introduced(self, compressor): + """Forcing the summary to role=user must not create two + consecutive user-role messages (strict alternation invariant). + When a user survives in the tail we do NOT force, so the pinned + summary can never collide with a user-role neighbour.""" + from agent.context_compressor import SUMMARY_PREFIX + + c = compressor + c.compression_count = 1 + messages = [{"role": "user", "content": "work kanban task 9"}] + messages += _tool_turns(0, 12) + + mocked = f"{SUMMARY_PREFIX}\nsummary body" + with patch.object(c, "_generate_summary", return_value=mocked): + out = c.compress(messages, current_tokens=90_000) + + for prev, cur in zip(out, out[1:]): + assert not ( + prev.get("role") == "user" and cur.get("role") == "user" + ), "compression introduced consecutive user-role messages" + + def test_preserved_tail_user_is_not_overridden(self, compressor): + """When a genuine user message survives in the tail, the guard + must NOT fire (the summary keeps its alternation-driven role) — + the request already has a user turn.""" + from agent.context_compressor import SUMMARY_PREFIX + + c = compressor + c.compression_count = 1 + c.tail_token_budget = 10 # tight tail so most turns compress + messages = [{"role": "user", "content": "old task"}] + messages += _tool_turns(0, 10) + # A recent, genuine user turn that will be preserved in the tail. + messages += [ + {"role": "user", "content": "the latest live user question"}, + {"role": "assistant", "content": "on it"}, + ] + + mocked = f"{SUMMARY_PREFIX}\nsummary body" + with patch.object(c, "_generate_summary", return_value=mocked): + out = c.compress(messages, current_tokens=90_000) + + hist = _role_hist(out) + assert hist.get("user", 0) >= 1 + joined = "\n".join( + m.get("content") for m in out if isinstance(m.get("content"), str) + ) + assert "the latest live user question" in joined + + +class TestSourceGuardrail: + @pytest.fixture + def source(self) -> str: + from pathlib import Path + + return ( + Path(__file__).resolve().parents[2] + / "agent" + / "context_compressor.py" + ).read_text(encoding="utf-8") + + def test_zero_user_guard_present(self, source): + assert "#58753" in source + # The guard must consider messages OUTSIDE the (system-only) head + # case — i.e. it inspects the preserved tail for a surviving user. + assert "_user_survives" in source + + def test_guard_wired_into_force_user_leading(self, source): + """The guard must feed ``_force_user_leading`` so the existing + flip logic cannot revert the summary back to assistant.""" + idx_guard = source.find("_user_survives") + idx_force = source.rfind("_force_user_leading = True") + assert idx_guard >= 0 and idx_force >= 0 + assert idx_force > idx_guard