From e1b1c4d5306c201fb5a278442c0744bb2b35238d Mon Sep 17 00:00:00 2001 From: qbit-mirror-bot Date: Wed, 1 Jul 2026 10:40:13 +0000 Subject: [PATCH] fix(compressor): batch salvage \u2014 orphan tool_calls, cooldown abort flags, system-head summary role (#51225 #52056 #52167) --- agent/context_compressor.py | 146 +++++++-- tests/agent/test_context_compressor.py | 418 +++++++++++++++++++++++++ 2 files changed, 541 insertions(+), 23 deletions(-) diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 3b37af7b8ba6..0a5574e8e851 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -2092,8 +2092,16 @@ def _sanitize_tool_pairs(self, messages: List[Dict[str, Any]]) -> List[Dict[str, The API rejects this because every tool_call must be followed by a tool result with the matching call_id. - This method removes orphaned results and inserts stub results for - orphaned calls so the message list is always well-formed. + This method removes orphaned results and strips orphaned tool_calls + from assistant messages so the message list is always well-formed. + + Previous approach inserted stub ``role="tool"`` results for orphaned + tool_calls. That caused a secondary failure: the pre-API + ``repair_message_sequence()`` uses ``tc.get("id")`` to track known + call IDs while this sanitizer uses ``call_id || id``. When the two + disagree (Codex Responses API format: ``id != call_id``), stubs get + silently dropped by the repair pass, re-exposing the original orphans. + Stripping at the source avoids this entire class of mismatch. """ surviving_call_ids: set = set() for msg in messages: @@ -2120,24 +2128,34 @@ def _sanitize_tool_pairs(self, messages: List[Dict[str, Any]]) -> List[Dict[str, if not self.quiet_mode: logger.info("Compression sanitizer: removed %d orphaned tool result(s)", len(orphaned_results)) - # 2. Add stub results for assistant tool_calls whose results were dropped + # 2. Strip orphaned tool_calls from assistant messages whose results + # were dropped. Stripping is preferred over inserting stub results + # because stubs can be dropped by downstream repair_message_sequence + # when call_id != id (Codex Responses API format), re-exposing orphans. missing_results = surviving_call_ids - result_call_ids if missing_results: - patched: List[Dict[str, Any]] = [] for msg in messages: - patched.append(msg) - if msg.get("role") == "assistant": - for tc in msg.get("tool_calls") or []: - cid = self._get_tool_call_id(tc) - if cid in missing_results: - patched.append({ - "role": "tool", - "content": "[Result from earlier conversation — see context summary above]", - "tool_call_id": cid, - }) - messages = patched + if msg.get("role") != "assistant": + continue + tcs = msg.get("tool_calls") + if not tcs: + continue + kept = [tc for tc in tcs if self._get_tool_call_id(tc) not in missing_results] + if len(kept) != len(tcs): + if kept: + msg["tool_calls"] = kept + else: + msg.pop("tool_calls", None) + # Ensure the assistant message still has visible + # content so the API does not reject an empty turn. + content = msg.get("content") + if not content or (isinstance(content, str) and not content.strip()): + msg["content"] = "(tool call removed)" if not self.quiet_mode: - logger.info("Compression sanitizer: added %d stub tool result(s)", len(missing_results)) + logger.info( + "Compression sanitizer: stripped %d orphaned tool_call(s) from assistant messages", + len(missing_results), + ) return messages @@ -2224,9 +2242,21 @@ def _align_boundary_backward(self, messages: List[Dict[str, Any]], idx: int) -> def _find_last_user_message_idx( self, messages: List[Dict[str, Any]], head_end: int ) -> int: - """Return the index of the last user-role message at or after *head_end*, or -1.""" + """Return the index of the last user-role message at or after *head_end*, or -1. + + A context-compaction handoff banner can be inserted as a ``role="user"`` + message (see the summary-role selection in ``compress``). It is internal + continuity state, not a real user turn, so it must not be picked as the + tail anchor — otherwise ``_ensure_last_user_message_in_tail`` protects + the summary and rolls the genuine last user message into the next + compaction, re-triggering the active-task loss the anchor exists to + prevent. + """ for i in range(len(messages) - 1, head_end - 1, -1): - if messages[i].get("role") == "user": + msg = messages[i] + if msg.get("role") == "user" and not self._is_context_summary_content( + msg.get("content") + ): return i return -1 @@ -2350,6 +2380,17 @@ def _ensure_last_user_message_in_tail( (``messages[cut_idx:]``), walk ``cut_idx`` back to include it. We then re-align backward one more time to avoid splitting any tool_call/result group that immediately precedes the user message. + + Causal Coupling guard (#22523): the final ``max(last_user_idx, + head_end + 1)`` clamp can push the cut *past* the user message when + the user sits at ``head_end`` (the first compressible index) — the + only case where ``head_end + 1 > last_user_idx``. That splits the + turn-pair: the user lands in the compressed region without its + assistant reply, so the summariser records it as a pending ask and + the next session re-executes the already-completed task. When this + split is unavoidable, push the cut *forward* to ``pair_end`` so the + full pair (user + reply + tool results) is summarised together and + correctly marked as completed. """ last_user_idx = self._find_last_user_message_idx(messages, head_end) if last_user_idx < 0: @@ -2374,7 +2415,50 @@ def _ensure_last_user_message_in_tail( cut_idx, ) # Safety: never go back into the head region. - return max(last_user_idx, head_end + 1) + adjusted = max(last_user_idx, head_end + 1) + if adjusted > last_user_idx: + # The clamp would leave the user in the compressed region without + # its reply. Keep the pair intact by pushing the cut forward past + # the whole (user + assistant + tool results) turn-pair so it is + # summarised as a completed unit rather than a dangling ask. + pair_end = self._find_turn_pair_end(messages, last_user_idx) + if not self.quiet_mode: + logger.debug( + "Causal Coupling: cut would split turn-pair at user %d; " + "pushing cut forward to pair_end %d so the completed pair " + "is summarised together (#22523)", + last_user_idx, + pair_end, + ) + return max(pair_end, head_end + 1) + return adjusted + + def _find_turn_pair_end( + self, + messages: List[Dict[str, Any]], + user_idx: int, + ) -> int: + """Return the index *after* the complete turn-pair starting at *user_idx*. + + A turn-pair is: ``user`` -> ``assistant`` [-> zero-or-more ``tool`` + results]. Returns the index of the first message that does *not* + belong to the pair, i.e. the natural cut point that keeps the pair + intact on one side of the boundary. + + If *user_idx* is the last message (no assistant reply yet), returns + ``user_idx + 1`` so the user message itself is minimally covered. + """ + n = len(messages) + idx = user_idx + 1 + if idx >= n: + return idx # user is the very last message — no reply yet + if messages[idx].get("role") != "assistant": + return idx # no assistant reply immediately following + idx += 1 + # Include any tool results that belong to this assistant turn. + while idx < n and messages[idx].get("role") == "tool": + idx += 1 + return idx def _find_tail_cut_by_tokens( self, messages: List[Dict[str, Any]], head_end: int, @@ -2529,8 +2613,16 @@ def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None, f self._last_aux_model_failure_error = None self._last_aux_model_failure_model = None self._last_compress_aborted = False - self._last_summary_auth_failure = False - self._last_summary_network_failure = False + # NOTE: do NOT reset _last_summary_auth_failure or + # _last_summary_network_failure here. These flags are set by + # _generate_summary() on a terminal failure and are already cleared on + # a successful summary. Resetting them eagerly defeats the cooldown + # protection: _generate_summary() returns None from the cooldown + # early-return without re-asserting these flags, so the abort guard + # below would see False and fall through to the destructive + # static-fallback — the exact data-loss #29559 describes. Letting them + # persist across compress() calls is safe because a successful summary + # always clears both. # Manual /compress (force=True) bypasses the failure cooldown so the # user can retry immediately after an auto-compress abort. Without @@ -2726,9 +2818,17 @@ def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None, f _merge_summary_into_tail = False last_head_role = messages[compress_start - 1].get("role", "user") if compress_start > 0 else "user" first_tail_role = messages[compress_end].get("role", "user") if compress_end < n_messages else "user" + # When the only protected head message is the system prompt, the + # summary becomes the first *visible* message in the API request + # (most adapters — Anthropic, Bedrock — send the system prompt as + # a separate ``system`` parameter, not inside ``messages[]``). + # Anthropic unconditionally rejects requests whose first message + # is not role=user, so we must pin the summary to "user" and + # prevent the flip logic below from reverting it (#52160). + _force_user_leading = last_head_role == "system" # Pick a role that avoids consecutive same-role with both neighbors. # Priority: avoid colliding with head (already committed), then tail. - if last_head_role in {"assistant", "tool"}: + if last_head_role in {"assistant", "tool"} or _force_user_leading: summary_role = "user" else: summary_role = "assistant" @@ -2736,7 +2836,7 @@ def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None, f # collide with the head, flip it. if summary_role == first_tail_role: flipped = "assistant" if summary_role == "user" else "user" - if flipped != last_head_role: + if flipped != last_head_role and not _force_user_leading: summary_role = flipped else: # Both roles would create consecutive same-role messages diff --git a/tests/agent/test_context_compressor.py b/tests/agent/test_context_compressor.py index cd23d13480ce..be3fcdf5ab14 100644 --- a/tests/agent/test_context_compressor.py +++ b/tests/agent/test_context_compressor.py @@ -2782,3 +2782,421 @@ def test_real_value_not_revised_downward(self, compressor): compressor.last_prompt_tokens = 50_000 result = self._seed(compressor.last_prompt_tokens, 10_000) assert result == 50_000 + + +class TestTurnPairPreservation: + """Causal Coupling guard (#22523): compaction must never orphan a user turn. + + ``_ensure_last_user_message_in_tail`` pulls the cut back to keep the last + user message in the tail (fixes #10896). But its final + ``max(last_user_idx, head_end + 1)`` clamp pushes the cut *past* the user + when the user sits at ``head_end`` (the first compressible index) — the + only case where ``head_end + 1 > last_user_idx``. The user then lands in + the compressed region without its assistant reply; the summariser marks it + as a pending ask and the next session re-executes the completed task. + + The guard detects that split and pushes the cut forward to ``pair_end`` so + the complete (user -> assistant [-> tool results]) pair is summarised as a + finished unit. + """ + + @pytest.fixture + def compressor(self): + return ContextCompressor( + model="test/model", + threshold_percent=0.85, + protect_first_n=1, + protect_last_n=0, + quiet_mode=True, + ) + + # ------------------------------------------------------------------ + # _find_turn_pair_end unit tests + # ------------------------------------------------------------------ + + def test_pair_end_user_only(self, compressor): + """User at end of list — no reply yet — pair_end is user+1.""" + msgs = [{"role": "user", "content": "hello"}] + assert compressor._find_turn_pair_end(msgs, 0) == 1 + + def test_pair_end_user_with_assistant_reply(self, compressor): + """User + assistant — pair_end skips both.""" + msgs = [ + {"role": "user", "content": "do x"}, + {"role": "assistant", "content": "done"}, + ] + assert compressor._find_turn_pair_end(msgs, 0) == 2 + + def test_pair_end_user_assistant_with_tools(self, compressor): + """User + assistant + tool results — pair_end skips the whole group.""" + msgs = [ + {"role": "user", "content": "run it"}, + {"role": "assistant", "content": None, + "tool_calls": [{"function": {"name": "exec", "arguments": "{}"}}]}, + {"role": "tool", "tool_call_id": "c1", "content": "ok"}, + {"role": "tool", "tool_call_id": "c2", "content": "ok"}, + ] + assert compressor._find_turn_pair_end(msgs, 0) == 4 + + def test_pair_end_stops_at_next_user(self, compressor): + """pair_end must not cross into the next user turn.""" + msgs = [ + {"role": "user", "content": "first"}, + {"role": "assistant", "content": "reply"}, + {"role": "user", "content": "second"}, + ] + assert compressor._find_turn_pair_end(msgs, 0) == 2 + + # ------------------------------------------------------------------ + # _ensure_last_user_message_in_tail unit tests + # ------------------------------------------------------------------ + + def test_user_already_in_tail_unchanged(self, compressor): + """When the user message is already past cut_idx, nothing changes.""" + msgs = [ + {"role": "user", "content": "head"}, + {"role": "assistant", "content": "head reply"}, + {"role": "user", "content": "last user"}, + {"role": "assistant", "content": "last reply"}, + ] + result = compressor._ensure_last_user_message_in_tail(msgs, cut_idx=2, head_end=1) + assert result == 2 + + def test_user_in_compressed_region_pulled_back(self, compressor): + """User in the middle (not at head_end) is pulled into the tail (#10896).""" + msgs = [ + {"role": "user", "content": "head"}, # 0 + {"role": "assistant", "content": "hi"}, # 1 + {"role": "user", "content": "do thing"}, # 2 <- last user + {"role": "assistant", "content": "done"}, # 3 + ] + # head_end=0, so head_end+1=1 <= last_user_idx=2: the #10896 pullback + # applies and the user stays in the tail (no forward push). + result = compressor._ensure_last_user_message_in_tail(msgs, cut_idx=3, head_end=0) + assert result <= 2 + + def test_orphan_prevented_user_at_head_end(self, compressor): + """Causal Coupling: user at head_end pushes the WHOLE pair into the summary. + + This is the #22523 case: last_user_idx == head_end, so the clamp would + return head_end+1 and orphan the user. The guard instead pushes the + cut forward to pair_end so user + reply + tool results are summarised + together and the tail never starts with a dangling user ask. + """ + msgs = [ + {"role": "user", "content": "first exchange"}, # 0 head + {"role": "user", "content": "THE ACTIVE ASK"}, # 1 = head_end, last user + {"role": "assistant", "content": "done"}, # 2 reply + {"role": "tool", "tool_call_id": "c1", "content": "toolout"}, # 3 + {"role": "assistant", "content": "final reply"}, # 4 + ] + head_end = 1 + result = compressor._ensure_last_user_message_in_tail(msgs, cut_idx=3, head_end=head_end) + # Whole pair (indices 1..3) lands in the compressed region; tail starts at 4. + assert result == 4 + tail = msgs[result:] + assert tail and tail[0]["role"] == "assistant" + + def test_no_orphan_after_full_compaction_cycle(self, compressor): + """End-to-end: after _find_tail_cut_by_tokens, the tail never starts + with an unanswered user message.""" + msgs = [ + {"role": "user", "content": "initial"}, + {"role": "assistant", "content": "ok"}, + ] + for i in range(5): + msgs.append({"role": "user", "content": f"step {i}"}) + msgs.append({"role": "assistant", "content": f"done {i}"}) + msgs.append({"role": "user", "content": "lights off please"}) + msgs.append({"role": "assistant", "content": "lights are off"}) + + head_end = compressor.protect_first_n + cut = compressor._find_tail_cut_by_tokens(msgs, head_end) + tail = msgs[cut:] + + if tail and tail[0].get("role") == "user": + assert len(tail) >= 2 and tail[1].get("role") == "assistant", ( + f"Orphan user turn at tail start: {tail[0]['content']!r} — " + f"next role is {tail[1].get('role') if len(tail) > 1 else 'nothing'}" + ) + + +class TestSanitizerStripsOrphanedToolCalls: + """PR #51218 (salvaged from #51225): orphaned tool_calls are stripped from + assistant messages instead of having stub tool results inserted, avoiding + the call_id != id mismatch that let downstream repair_message_sequence drop + the stubs and re-expose orphans.""" + + def test_sanitizer_strips_orphaned_tool_calls(self, compressor): + """Orphaned tool_calls (no matching tool result) are stripped from + assistant messages instead of having stubs inserted. #51218""" + msgs = [ + { + "role": "assistant", + "content": "", + "tool_calls": [ + {"id": "tc_orphan", "function": {"name": "search", "arguments": "{}"}}, + ], + }, + {"role": "user", "content": "never mind"}, + ] + + sanitized = compressor._sanitize_tool_pairs(msgs) + + # Orphaned tool_call should be stripped, not stub-inserted + asst = next(m for m in sanitized if m.get("role") == "assistant") + assert not asst.get("tool_calls"), "orphaned tool_calls should be stripped" + # No stub tool messages should be added + assert not any(m.get("role") == "tool" for m in sanitized) + # Empty assistant should get placeholder content + assert asst.get("content") == "(tool call removed)" + + def test_sanitizer_strips_orphaned_keeps_valid(self, compressor): + """When an assistant has both valid and orphaned tool_calls, only + the orphans are stripped. #51218""" + msgs = [ + { + "role": "assistant", + "content": "", + "tool_calls": [ + {"id": "tc_valid", "function": {"name": "read_file", "arguments": "{}"}}, + {"id": "tc_orphan", "function": {"name": "search", "arguments": "{}"}}, + ], + }, + {"role": "tool", "tool_call_id": "tc_valid", "content": "file content"}, + ] + + sanitized = compressor._sanitize_tool_pairs(msgs) + + asst = next(m for m in sanitized if m.get("role") == "assistant") + assert len(asst["tool_calls"]) == 1 + assert asst["tool_calls"][0]["id"] == "tc_valid" + # Valid tool result preserved + tool_msgs = [m for m in sanitized if m.get("role") == "tool"] + assert len(tool_msgs) == 1 + assert tool_msgs[0]["tool_call_id"] == "tc_valid" + + def test_sanitizer_strips_orphaned_preserves_text_content(self, compressor): + """When an assistant has text content AND orphaned tool_calls, + the text is preserved and only tool_calls are stripped. #51218""" + msgs = [ + { + "role": "assistant", + "content": "Let me search for that.", + "tool_calls": [ + {"id": "tc_orphan", "function": {"name": "search", "arguments": "{}"}}, + ], + }, + {"role": "user", "content": "thanks"}, + ] + + sanitized = compressor._sanitize_tool_pairs(msgs) + + asst = next(m for m in sanitized if m.get("role") == "assistant") + assert asst["content"] == "Let me search for that." + assert not asst.get("tool_calls") + # The placeholder must NOT overwrite existing text content. + assert asst["content"] != "(tool call removed)" + + def test_sanitizer_strips_orphaned_with_call_id_mismatch(self, compressor): + """Stubs with call_id != id used to be dropped by downstream + repair_message_sequence, re-exposing orphans. Stripping avoids + this entirely. #51218""" + msgs = [ + { + "role": "assistant", + "content": "", + "tool_calls": [ + { + "id": "fc_abc", + "call_id": "call_abc", + "function": {"name": "search", "arguments": "{}"}, + }, + ], + }, + # No tool result for call_abc — orphaned + {"role": "user", "content": "next"}, + ] + + sanitized = compressor._sanitize_tool_pairs(msgs) + + asst = next(m for m in sanitized if m.get("role") == "assistant") + assert not asst.get("tool_calls") + # No stub tool messages (which would have call_id != id mismatch) + + +class TestCooldownReentryAbort: + """Regression: a second compress() call during the failure cooldown must + still abort when the original failure was a network/auth error. + + Before the fix, compress() unconditionally reset _last_summary_network_failure + and _last_summary_auth_failure at the top of every call. When + _generate_summary() returned None from the cooldown early-return (without + re-setting the flags), the abort guard saw False and fell through to the + destructive static-fallback path — reproducing the data-loss scenario from + #29559 / #25585 that PR #51881 originally fixed. + """ + + def _msgs(self, n=12): + return [ + {"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"} + for i in range(n) + ] + + def test_network_failure_cooldown_reentry_still_aborts(self): + """ConnectionError → first compress aborts (PR #51881). Second + compress within the 30s cooldown must ALSO abort — not drop the + middle window via the static-fallback path.""" + with patch("agent.context_compressor.get_model_context_length", return_value=100000): + c = ContextCompressor( + model="test", + quiet_mode=True, + protect_first_n=2, + protect_last_n=2, + abort_on_summary_failure=False, + ) + msgs = self._msgs(12) + + with patch( + "agent.context_compressor.call_llm", + side_effect=ConnectionError("Connection error."), + ): + first = c.compress(msgs, current_tokens=999999, force=True) + assert first == msgs + assert c._last_compress_aborted is True + assert c._last_summary_network_failure is True + + second = c.compress(msgs, current_tokens=999999) + assert second == msgs, ( + "Second compress during cooldown must abort (preserve messages), " + "not drop the middle window via static-fallback" + ) + assert c._last_compress_aborted is True + assert c._last_summary_fallback_used is False + + def test_auth_failure_cooldown_reentry_still_aborts(self): + """Same re-entry hole for auth failures: a 401 sets the flag, cooldown + returns None, second compress must still abort.""" + err = Exception("Error code: 401 - invalid api key") + err.status_code = 401 + with patch("agent.context_compressor.get_model_context_length", return_value=100000): + c = ContextCompressor( + model="test", + quiet_mode=True, + protect_first_n=2, + protect_last_n=2, + abort_on_summary_failure=False, + ) + msgs = self._msgs(12) + + with patch("agent.context_compressor.call_llm", side_effect=err): + first = c.compress(msgs, current_tokens=999999, force=True) + assert first == msgs + assert c._last_compress_aborted is True + assert c._last_summary_auth_failure is True + + second = c.compress(msgs, current_tokens=999999) + assert second == msgs, ( + "Second compress during cooldown must abort (preserve messages), " + "not drop the middle window via static-fallback" + ) + assert c._last_compress_aborted is True + assert c._last_summary_fallback_used is False + + +class TestDoubleCompactionSummaryRole: + """PR #52160 (salvaged from #52167): when only the system prompt is + protected, the summary must lead with role=user (Anthropic/Bedrock send + system as a separate param, so the summary is the first visible message).""" + + def test_double_compaction_summary_must_be_user_when_only_system_protected(self): + """After the first compression, protect_first_n decays to 0. + + On the second compression the only protected head message is the + system prompt (role=system). The summary becomes the first + *visible* message in the API request because adapters like + Anthropic and Bedrock send the system prompt as a separate + ``system`` parameter. The summary MUST be role=user or the + provider rejects with HTTP 400 (#52160). + """ + mock_response = MagicMock() + mock_response.choices = [MagicMock()] + mock_response.choices[0].message.content = "summary of earlier turns" + + with patch("agent.context_compressor.get_model_context_length", return_value=100000): + c = ContextCompressor( + model="test", quiet_mode=True, protect_first_n=2, protect_last_n=2, + ) + # Simulate second compression: protect_first_n decays to 0. + c.compression_count = 1 + + # compress_start will be 1 (system only), last_head_role = "system". + # Without the fix, summary_role would be "assistant". + msgs = [ + {"role": "system", "content": "You are a helpful assistant."}, + {"role": "user", "content": "msg 1"}, + {"role": "assistant", "content": "msg 2"}, + {"role": "user", "content": "msg 3"}, + {"role": "assistant", "content": "msg 4"}, + {"role": "user", "content": "msg 5"}, + {"role": "assistant", "content": "msg 6"}, + ] + with patch("agent.context_compressor.call_llm", return_value=mock_response): + result = c.compress(msgs) + + # The system message must still be at index 0. + assert result[0]["role"] == "system" + # The summary (first non-system message) must be role=user. + non_system = [m for m in result if m.get("role") != "system"] + assert non_system, "expected at least one non-system message" + assert non_system[0]["role"] == "user", ( + f"first non-system message must be role=user for Anthropic " + f"compatibility, got role={non_system[0]['role']!r}" + ) + + def test_double_compaction_user_tail_merges_into_tail(self): + """When the summary is forced to role=user (system-only head) and + the first tail message is also user, the summary must merge into + the tail rather than flipping back to assistant (#52160). + """ + mock_response = MagicMock() + mock_response.choices = [MagicMock()] + mock_response.choices[0].message.content = "summary of earlier turns" + + with patch("agent.context_compressor.get_model_context_length", return_value=100000): + c = ContextCompressor( + model="test", quiet_mode=True, protect_first_n=2, protect_last_n=2, + ) + c.compression_count = 1 # decay protect_first_n + + # tail starts with user → would collide with forced summary_role=user. + # The fix should merge into tail instead of flipping to assistant. + msgs = [ + {"role": "system", "content": "You are a helpful assistant."}, + {"role": "user", "content": "msg 1"}, + {"role": "assistant", "content": "msg 2"}, + {"role": "user", "content": "msg 3"}, + {"role": "assistant", "content": "msg 4"}, + {"role": "user", "content": "msg 5"}, # tail start (user) + {"role": "assistant", "content": "msg 6"}, + {"role": "user", "content": "msg 7"}, + ] + with patch("agent.context_compressor.call_llm", return_value=mock_response): + result = c.compress(msgs) + + # No standalone summary message should exist (merged into tail). + summary_msgs = [ + m for m in result + if m.get("_compressed_summary") and "msg 5" not in (m.get("content") or "") + ] + assert len(summary_msgs) == 0, ( + "summary should be merged into tail, not standalone" + ) + # The first non-system message must be role=user. + non_system = [m for m in result if m.get("role") != "system"] + assert non_system[0]["role"] == "user" + # The merged tail should contain the summary text. + assert any( + "summary of earlier turns" in (m.get("content") or "") + for m in result + )