Compaction summary role and leak handling in chat context

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #45153 fix(compression): append END OF CONTEXT SUMMARY marker to all standalone summaries (salvages #33346)

  merged to main

  files: agent/context_compressor.py, tests/agent/test_context_compressor.py

  Show the diff

  • agent/context_compressor.py

    @@ -69,6 +69,16 @@
     )
     LEGACY_SUMMARY_PREFIX = "[CONTEXT SUMMARY]:"
     
    +# Appended to every standalone summary message (and to the merged-into-tail
    +# prefix) so the model has an unambiguous "summary ends here" boundary.
    +# Without it, weak models read the verbatim "## Active Task" quote as fresh
    +# user input (#11475, #14521) or regurgitate an assistant-role summary as
    +# their own output (#33256).
    +_SUMMARY_END_MARKER = (
    +    "--- END OF CONTEXT SUMMARY — "
    +    "respond to the message below, not the summary above ---"
    +)
    +
     # Handoff prefixes that shipped in earlier releases. A summary persisted under
     # one of these can be inherited into a resumed lineage (#35344); when it is
     # re-normalized on re-compaction we must strip the OLD prefix too, otherwise the
    @@ -1616,7 +1626,13 @@ def _strip_summary_prefix(summary: str) -> str:
             text = (summary or "").strip()
             for prefix in (SUMMARY_PREFIX, LEGACY_SUMMARY_PREFIX, *_HISTORICAL_SUMMARY_PREFIXES):
                 if text.startswith(prefix):
    -                return text[len(prefix):].lstrip()
    +                text = text[len(prefix):].lstrip()
    +                break
    +        # Strip the trailing end marker too — a rehydrated handoff body that
    +        # keeps it would leak the boundary directive into the iterative-update
    +        # summarizer prompt (and the marker is re-appended on insertion anyway).
    +        if text.endswith(_SUMMARY_END_MARKER):
    +            text = text[: -len(_SUMMARY_END_MARKER)].rstrip()
             return text
     
         @classmethod
    @@ -2193,15 +2209,13 @@ def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None, f
     
             # When the summary lands as a standalone role="user" message,
             # weak models read the verbatim "## Active Task" quote of a past
    -        # user request as fresh input (#11475, #14521). Append the explicit
    -        # end marker — the same one used in the merge-into-tail path — so
    -        # the model has a clear "summary above, not new input" signal.
    -        if not _merge_summary_into_tail and summary_role == "user":
    -            summary = (
    -                summary
    -                + "\n\n--- END OF CONTEXT SUMMARY — "
    -                "respond to the message below, not the summary above ---"
    -            )
    +        # user request as fresh input (#11475, #14521).
    +        # When it lands as role="assistant", models may regurgitate the
    +        # summary text as their own output (#33256). In both cases, append
    +        # the explicit end marker so the model has a clear "summary ends
    +        # here, respond to the message below" signal.
    +        if not _merge_summary_into_tail:
    +            summary = summary + "\n\n" + _SUMMARY_END_MARKER
     
             if not _merge_summary_into_tail:
                 compressed.append({"role": summary_role, "content": summary})
    @@ -2209,11 +2223,7 @@ def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None, f
             for i in range(compress_end, n_messages):
                 msg = messages[i].copy()
                 if _merge_summary_into_tail and i == compress_end:
    -                merged_prefix = (
    -                    summary
    -                    + "\n\n--- END OF CONTEXT SUMMARY — "
    -                    "respond to the message below, not the summary above ---\n\n"
    -                )
    +                merged_prefix = summary + "\n\n" + _SUMMARY_END_MARKER + "\n\n"
                     msg["content"] = _append_text_to_content(
                         msg.get("content"),
                         merged_prefix,

  • tests/agent/test_context_compressor.py

    @@ -1252,6 +1252,48 @@ def test_user_role_summary_carries_end_marker(self):
                 "respond to the message below, not the summary above ---"
             )
     
    +    def test_assistant_role_summary_carries_end_marker(self):
    +        """When the summary lands as standalone role='assistant' (head ends
    +        with user), the message body must include the explicit
    +        '--- END OF CONTEXT SUMMARY ---' marker. Without it, models may
    +        regurgitate the summary text as their own output (#33256).
    +        """
    +        mock_client = MagicMock()
    +        mock_response = MagicMock()
    +        mock_response.choices = [MagicMock()]
    +        mock_response.choices[0].message.content = "[CONTEXT SUMMARY]: stuff happened"
    +        mock_client.chat.completions.create.return_value = mock_response
    +
    +        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)
    +
    +        # head_last=user → summary_role="assistant" (same setup as
    +        # test_summary_role_avoids_consecutive_user_when_head_ends_with_user).
    +        # With min_tail=3, tail = last 3 messages (indices 5-7).
    +        # head_last=user, tail_first=user → the assistant-role summary does
    +        # not collide with either neighbor and should be inserted standalone.
    +        msgs = [
    +            {"role": "system", "content": "system prompt"},
    +            {"role": "user", "content": "msg 1"},
    +            {"role": "user", "content": "msg 2"},  # last head — user
    +            {"role": "assistant", "content": "msg 3"},
    +            {"role": "user", "content": "msg 4"},
    +            {"role": "user", "content": "msg 5"},
    +            {"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)
    +
    +        summary_msg = next(
    +            m for m in result if (m.get("content") or "").startswith(SUMMARY_PREFIX)
    +        )
    +        assert summary_msg["role"] == "assistant"
    +        assert "END OF CONTEXT SUMMARY" in summary_msg["content"]
    +        assert summary_msg["content"].rstrip().endswith(
    +            "respond to the message below, not the summary above ---"
    +        )
    +
         def test_summary_role_avoids_consecutive_user_messages(self):
             """Summary role should alternate with the last head message to avoid consecutive same-role messages."""
             mock_client = MagicMock()

• #56197 fix(compressor): batch salvage \u2014 orphan tool_calls, cooldown abort flags, system-head summary role (#51225 #52056 #52167)

  merged to main

  files: agent/context_compressor.py, tests/agent/test_context_compressor.py

  Show the diff

  • 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
     
    @@ -2595,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
    @@ -2792,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"
    @@ -2802,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

  • tests/agent/test_context_compressor.py

    @@ -2919,3 +2919,284 @@ def test_no_orphan_after_full_compaction_cycle(self, compressor):
                     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=[redacted:secret-assignment] 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=[redacted:secret-assignment]
    +        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=[redacted:secret-assignment] 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=[redacted:secret-assignment]
    +        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
    +        )