perf: address greptile review for anthropic streaming hot path

- Bail to legacy in `_collapse_pure_text_chunks` when content_block_delta
  events from different block indexes are observed without an intervening
  flush. Anthropic sends blocks strictly sequentially, but defensive bail
  prevents silent text-merging if the protocol ever interleaves.
- Replace leaf-class `__dict__` check for `async_post_call_streaming_hook`
  in `_callback_capabilities` with a function-identity comparison that
  walks the MRO. A vendor base class can carry the override and the
  registered class can add nothing else; before this PR the hook was
  unconditionally invoked, so an inherited-override miss would silently
  drop the hook on the streaming path.
- Add unit tests for both behaviors.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py

@@ -390,10 +390,24 @@ class AnthropicPassthroughLoggingHandler:
                     return None
                 if dtype != "text_delta":
                     return None
-                if data.get("index") not in text_block_indexes:
+                cur_index = data.get("index")
+                if cur_index not in text_block_indexes:
+                    return None
+                # Defensive: Anthropic sends blocks strictly sequentially
+                # (start/deltas/stop, then next block), so pending_text from
+                # block N must be flushed by content_block_stop before block
+                # N+1's deltas arrive. If we ever see a delta whose index
+                # disagrees with the current pending buffer, the stream is
+                # interleaved -- fall back to legacy rather than risk merging
+                # text from different blocks under a single index.
+                if (
+                    pending_text
+                    and pending_index is not None
+                    and cur_index != pending_index
+                ):
                     return None
                 saw_any_text_delta = True
-                pending_index = data.get("index")
+                pending_index = cur_index
                 pending_text.append(delta.get("text") or "")
             elif etype == "ping":
                 # Interior no-op; legacy maps it to an empty chunk.

litellm/proxy/utils.py

@@ -1596,7 +1596,22 @@ class ProxyLogging:
                 iterator_overrides.append((resolved, "override"))
             elif "apply_guardrail" in cls_attrs:
                 iterator_overrides.append((resolved, "apply_guardrail"))
-            if "async_post_call_streaming_hook" in cls_attrs:
+            # Walk the MRO for ``async_post_call_streaming_hook`` rather than
+            # using the leaf-class ``__dict__`` check used by the other flags:
+            # before this PR the hook was unconditionally invoked, so a
+            # callback that inherits an override from an intermediate parent
+            # (e.g. a vendor base class providing the override, with the
+            # registered class adding nothing else) MUST still be detected.
+            # A leaf-class miss here would silently drop the inherited hook.
+            base_streaming_hook = CustomLogger.async_post_call_streaming_hook
+            cls_streaming_hook = getattr(
+                cls,
+                "async_post_call_streaming_hook",
+                base_streaming_hook,
+            )
+            if getattr(
+                cls_streaming_hook, "__func__", cls_streaming_hook
+            ) is not getattr(base_streaming_hook, "__func__", base_streaming_hook):
                 has_streaming_chunk_override = True
             if "async_pre_call_hook" in cls_attrs:
                 has_pre_call_override = True

tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_anthropic_passthrough_logging_handler.py

@@ -977,3 +977,69 @@ class TestPureTextFastPathParity:
         assert collapsed is not None
         # 50 text deltas + 50 event markers + 1 ping collapse to far fewer.
         assert len(collapsed) < len(all_chunks) / 2
+
+    def test_collapse_returns_none_for_interleaved_block_indexes(self):
+        """
+        Anthropic sends content blocks strictly sequentially (start/deltas/stop
+        for one, then the next). If a stream ever interleaves deltas across
+        block indexes, the fast path must bail to legacy rather than merge text
+        from different blocks under a single index.
+        """
+        frames = [
+            self._sse(
+                "message_start",
+                {
+                    "type": "message_start",
+                    "message": {
+                        "id": "msg_abc",
+                        "type": "message",
+                        "role": "assistant",
+                        "model": "claude-3-5-sonnet-20241022",
+                        "content": [],
+                        "stop_reason": None,
+                        "stop_sequence": None,
+                        "usage": {"input_tokens": 1, "output_tokens": 0},
+                    },
+                },
+            ),
+            self._sse(
+                "content_block_start",
+                {
+                    "type": "content_block_start",
+                    "index": 0,
+                    "content_block": {"type": "text", "text": ""},
+                },
+            ),
+            self._sse(
+                "content_block_start",
+                {
+                    "type": "content_block_start",
+                    "index": 1,
+                    "content_block": {"type": "text", "text": ""},
+                },
+            ),
+            # Interleave: delta for block 0, then delta for block 1, with no
+            # content_block_stop between them.
+            self._sse(
+                "content_block_delta",
+                {
+                    "type": "content_block_delta",
+                    "index": 0,
+                    "delta": {"type": "text_delta", "text": "hello "},
+                },
+            ),
+            self._sse(
+                "content_block_delta",
+                {
+                    "type": "content_block_delta",
+                    "index": 1,
+                    "delta": {"type": "text_delta", "text": "world"},
+                },
+            ),
+            self._sse("message_stop", {"type": "message_stop"}),
+        ]
+        all_chunks = list(self._to_all_chunks(frames))
+        assert (
+            AnthropicPassthroughLoggingHandler._collapse_pure_text_chunks(all_chunks)
+            is None
+        )

tests/test_litellm/proxy/test_proxy_logging_hook_detection.py

@@ -111,6 +111,28 @@ def test_callback_capabilities_captures_iterator_override(monkeypatch):
     assert kind == "override"
 
 
+def test_callback_capabilities_detects_inherited_streaming_chunk_override(monkeypatch):
+    """
+    ``async_post_call_streaming_hook`` must be detected even when the override
+    lives on an intermediate parent class — a vendor base class can carry the
+    override and the registered class can add nothing else. Before this PR the
+    hook was unconditionally invoked, so a leaf-class ``__dict__`` miss here
+    would silently drop the inherited hook.
+    """
+    ProxyLogging._callback_capabilities_cache.clear()
+
+    class _StreamingBase(CustomLogger):
+        async def async_post_call_streaming_hook(self, *args, **kwargs):  # type: ignore[override]
+            return kwargs.get("response")
+
+    class _LeafWithoutOverride(_StreamingBase):
+        pass
+
+    monkeypatch.setattr(litellm, "callbacks", [_LeafWithoutOverride()])
+    caps = ProxyLogging._callback_capabilities()
+    assert caps.has_streaming_chunk_override is True
+
+
 def test_callback_capabilities_cache_invalidates_on_list_change(monkeypatch):
     """The cache key includes (length, id-of-each-callback).  Mutating the
     callback list must produce a fresh capability snapshot."""