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

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,17 +2818,25 @@ 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"
         # If the chosen role collides with the tail AND flipping wouldn't
         # 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=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
+        )