ContextCompressor orphan tool_call handling and related compressor fixes

Fixes on record

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

• #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
    +        )