feat(messages): emit ping and error stream events for Anthropic SDK compatibility (#6007)

## Summary
- Add `PingEvent` and `ErrorStreamEvent` models to
`AnthropicStreamEvent` union
- Emit ping heartbeats after `message_start` and each
`content_block_stop` in translation-mode streams to prevent
load-balancer idle-timeout disconnects
- Wrap `_stream_openai_to_anthropic` and `_passthrough_stream` in error
handling that yields `ErrorStreamEvent` before closing, so clients
receive typed errors instead of truncated streams
- Handle `ping` and `error` event types in `_parse_sse_event` for
passthrough mode

## Test plan
- Unit tests for ping event placement in text and tool streams
- Unit tests for error event emission on mid-stream exceptions
- Unit tests for ping/error parsing in `_parse_sse_event`
- Full unit test suite passes with no regressions

🤖 Generated with [Claude Code](https://claude.com/claude-code)
<!-- devin-review-badge-begin -->

---

<a href="https://app.devin.ai/review/ogx-ai/ogx/pull/6007"
target="_blank">
  <picture>
<source media="(prefers-color-scheme: dark)"
srcset="https://static.devin.ai/assets/gh-open-in-devin-review-dark.svg?v=1">
<img
src="https://static.devin.ai/assets/gh-open-in-devin-review-light.svg?v=1"
alt="Open in Devin Review">
  </picture>
</a>
<!-- devin-review-badge-end -->

---------

Signed-off-by: Eleanor Hu <ehu@redhat.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: Charlie Doern <cdoern@redhat.com>

Author: 3 people

src/ogx/providers/inline/messages/impl.py

@@ -60,6 +60,7 @@
     ContentBlockStartEvent,
     ContentBlockStopEvent,
     CreateMessageBatchRequest,
+    ErrorStreamEvent,
     ListMessageBatchesRequest,
     ListMessageBatchesResponse,
     MessageBatch,
@@ -71,6 +72,7 @@
     MessageDeltaEvent,
     MessageStartEvent,
     MessageStopEvent,
+    PingEvent,
     RetrieveMessageBatchRequest,
     RetrieveMessageBatchResultsRequest,
     _AnthropicErrorDetail,
@@ -515,19 +517,26 @@ async def _passthrough_stream(
         body: dict[str, Any],
     ) -> AsyncIterator[AnthropicStreamEvent]:
         """Stream SSE events directly from the provider."""
-        async with self._client.stream("POST", url, json=body, headers=headers, timeout=300) as resp:
-            resp.raise_for_status()
-            event_type = None
-            async for line in resp.aiter_lines():
-                line = line.strip()
-                if line.startswith("event: "):
-                    event_type = line[7:]
-                elif line.startswith("data: ") and event_type:
-                    data = json.loads(line[6:])
-                    event = self._parse_sse_event(event_type, data)
-                    if event:
-                        yield event
-                    event_type = None
+        try:
+            async with self._client.stream("POST", url, json=body, headers=headers, timeout=300) as resp:
+                resp.raise_for_status()
+                event_type = None
+                async for line in resp.aiter_lines():
+                    line = line.strip()
+                    if line.startswith("event: "):
+                        event_type = line[7:]
+                    elif line.startswith("data: ") and event_type:
+                        data = json.loads(line[6:])
+                        event = self._parse_sse_event(event_type, data)
+                        if event:
+                            yield event
+                        event_type = None
+        except Exception:
+            logger.exception("Failed to stream passthrough response")
+            yield ErrorStreamEvent(
+                error=_AnthropicErrorDetail(type="api_error", message="Internal server error"),
+            )
+            return
 
     def _parse_sse_event(self, event_type: str, data: dict[str, Any]) -> AnthropicStreamEvent | None:
         """Parse an Anthropic SSE event from its type and data."""
@@ -572,6 +581,10 @@ def _parse_sse_event(self, event_type: str, data: dict[str, Any]) -> AnthropicSt
             )
         if event_type == "message_stop":
             return MessageStopEvent()
+        if event_type == "ping":
+            return PingEvent()
+        if event_type == "error":
+            return ErrorStreamEvent(error=_AnthropicErrorDetail(**data["error"]))
         return None
 
     async def _passthrough_count_tokens(
@@ -874,6 +887,7 @@ async def _stream_openai_to_anthropic(
                 usage=AnthropicUsage(input_tokens=0, output_tokens=0),
             ),
         )
+        yield PingEvent()
 
         content_block_index = 0
         in_text_block = False
@@ -884,81 +898,93 @@ async def _stream_openai_to_anthropic(
         cache_read_tokens: int | None = None
         stop_reason = "end_turn"
 
-        async for chunk in openai_stream:
-            if not chunk.choices:
-                # Usage-only chunk
-                if chunk.usage:
-                    input_tokens = chunk.usage.prompt_tokens or 0
-                    output_tokens = chunk.usage.completion_tokens or 0
-                    if chunk.usage.prompt_tokens_details and hasattr(
-                        chunk.usage.prompt_tokens_details, "cached_tokens"
-                    ):
-                        cache_read_tokens = chunk.usage.prompt_tokens_details.cached_tokens
-                continue
-
-            choice = chunk.choices[0]
-            delta = choice.delta
+        try:
+            async for chunk in openai_stream:
+                if not chunk.choices:
+                    # Usage-only chunk
+                    if chunk.usage:
+                        input_tokens = chunk.usage.prompt_tokens or 0
+                        output_tokens = chunk.usage.completion_tokens or 0
+                        if chunk.usage.prompt_tokens_details and hasattr(
+                            chunk.usage.prompt_tokens_details, "cached_tokens"
+                        ):
+                            cache_read_tokens = chunk.usage.prompt_tokens_details.cached_tokens
+                    continue
+
+                choice = chunk.choices[0]
+                delta = choice.delta
+
+                if delta and delta.content:
+                    if not in_text_block:
+                        yield ContentBlockStartEvent(
+                            index=content_block_index,
+                            content_block=AnthropicTextBlock(text=""),
+                        )
+                        in_text_block = True
 
-            if delta and delta.content:
-                if not in_text_block:
-                    yield ContentBlockStartEvent(
+                    yield ContentBlockDeltaEvent(
                         index=content_block_index,
-                        content_block=AnthropicTextBlock(text=""),
+                        delta=_TextDelta(text=delta.content),
                     )
-                    in_text_block = True
-
-                yield ContentBlockDeltaEvent(
-                    index=content_block_index,
-                    delta=_TextDelta(text=delta.content),
-                )
-
-            if delta and delta.tool_calls:
-                for tc_delta in delta.tool_calls:
-                    tc_idx = tc_delta.index if tc_delta.index is not None else 0
 
-                    if tc_idx not in in_tool_blocks:
-                        # Close text block if open
-                        if in_text_block:
-                            yield ContentBlockStopEvent(index=content_block_index)
+                if delta and delta.tool_calls:
+                    for tc_delta in delta.tool_calls:
+                        tc_idx = tc_delta.index if tc_delta.index is not None else 0
+
+                        if tc_idx not in in_tool_blocks:
+                            # Close text block if open
+                            if in_text_block:
+                                yield ContentBlockStopEvent(index=content_block_index)
+                                yield PingEvent()
+                                content_block_index += 1
+                                in_text_block = False
+
+                            # Start new tool_use block
+                            in_tool_blocks[tc_idx] = True
+                            tool_call_index_to_block_index[tc_idx] = content_block_index
+
+                            yield ContentBlockStartEvent(
+                                index=content_block_index,
+                                content_block=AnthropicToolUseBlock(
+                                    id=tc_delta.id or f"toolu_{uuid.uuid4().hex[:24]}",
+                                    name=tc_delta.function.name if tc_delta.function and tc_delta.function.name else "",
+                                    input={},
+                                ),
+                            )
                             content_block_index += 1
-                            in_text_block = False
-
-                        # Start new tool_use block
-                        in_tool_blocks[tc_idx] = True
-                        tool_call_index_to_block_index[tc_idx] = content_block_index
 
-                        yield ContentBlockStartEvent(
-                            index=content_block_index,
-                            content_block=AnthropicToolUseBlock(
-                                id=tc_delta.id or f"toolu_{uuid.uuid4().hex[:24]}",
-                                name=tc_delta.function.name if tc_delta.function and tc_delta.function.name else "",
-                                input={},
-                            ),
-                        )
-                        content_block_index += 1
-
-                    if tc_delta.function and tc_delta.function.arguments:
-                        block_idx = tool_call_index_to_block_index[tc_idx]
-                        yield ContentBlockDeltaEvent(
-                            index=block_idx,
-                            delta=_InputJsonDelta(partial_json=tc_delta.function.arguments),
-                        )
+                        if tc_delta.function and tc_delta.function.arguments:
+                            block_idx = tool_call_index_to_block_index[tc_idx]
+                            yield ContentBlockDeltaEvent(
+                                index=block_idx,
+                                delta=_InputJsonDelta(partial_json=tc_delta.function.arguments),
+                            )
 
-            if choice.finish_reason:
-                stop_reason = _FINISH_TO_STOP_REASON.get(choice.finish_reason, "end_turn")
+                if choice.finish_reason:
+                    stop_reason = _FINISH_TO_STOP_REASON.get(choice.finish_reason, "end_turn")
 
-            if chunk.usage:
-                input_tokens = chunk.usage.prompt_tokens or 0
-                output_tokens = chunk.usage.completion_tokens or 0
-                if chunk.usage.prompt_tokens_details and hasattr(chunk.usage.prompt_tokens_details, "cached_tokens"):
-                    cache_read_tokens = chunk.usage.prompt_tokens_details.cached_tokens
+                if chunk.usage:
+                    input_tokens = chunk.usage.prompt_tokens or 0
+                    output_tokens = chunk.usage.completion_tokens or 0
+                    if chunk.usage.prompt_tokens_details and hasattr(
+                        chunk.usage.prompt_tokens_details, "cached_tokens"
+                    ):
+                        cache_read_tokens = chunk.usage.prompt_tokens_details.cached_tokens
+        except Exception:
+            logger.exception("Failed to stream translation response")
+            yield ErrorStreamEvent(
+                error=_AnthropicErrorDetail(type="api_error", message="Internal server error"),
+            )
+            return
 
         # Close any open blocks
         if in_text_block:
             yield ContentBlockStopEvent(index=content_block_index)
+            yield PingEvent()
 
         for _tc_idx, block_idx in tool_call_index_to_block_index.items():
             yield ContentBlockStopEvent(index=block_idx)
+            yield PingEvent()
 
         # Final events
         yield MessageDeltaEvent(

src/ogx_api/messages/models.py

@@ -434,16 +434,6 @@ class MessageStopEvent(BaseModel):
     type: Literal["message_stop"] = "message_stop"
 
 
-AnthropicStreamEvent = (
-    MessageStartEvent
-    | ContentBlockStartEvent
-    | ContentBlockDeltaEvent
-    | ContentBlockStopEvent
-    | MessageDeltaEvent
-    | MessageStopEvent
-)
-
-
 # -- Error response --
 
 
@@ -459,6 +449,31 @@ class AnthropicErrorResponse(BaseModel):
     error: _AnthropicErrorDetail
 
 
+class PingEvent(BaseModel):
+    """Keep-alive heartbeat sent between content blocks."""
+
+    type: Literal["ping"] = "ping"
+
+
+class ErrorStreamEvent(BaseModel):
+    """Mid-stream error event sent before the stream closes."""
+
+    type: Literal["error"] = "error"
+    error: _AnthropicErrorDetail
+
+
+AnthropicStreamEvent = (
+    MessageStartEvent
+    | ContentBlockStartEvent
+    | ContentBlockDeltaEvent
+    | ContentBlockStopEvent
+    | MessageDeltaEvent
+    | MessageStopEvent
+    | PingEvent
+    | ErrorStreamEvent
+)
+
+
 # -- Message Batches --