fix: re-inject Anthropic keepalive pings on the streaming path so long generations don't idle out

[sjawhar]

Problem

When proxying a streaming request (messages.create(stream=True)), the gateway can go completely silent on the client connection for the entire duration of a long generation. Any intermediary with an idle timeout (load balancer, reverse proxy, API gateway) then cuts the connection mid-stream even though the request is healthy — the client sees a truncated stream / RemoteProtocolError / 504 while the gateway and Anthropic both consider the request successful.

Root cause

Anthropic's wire protocol emits periodic event: ping keepalives during generation (notably when a model produces no output tokens for a while — extended thinking, or a slow/gated response). The Anthropic Python SDK's typed event stream drops these ping events (they aren't part of RawMessageStreamEvent), so async for event in client.messages.create(..., stream=True) never yields them. The gateway therefore forwards nothing to the client between message_start and the first content event. A direct connection to api.anthropic.com survives the same long generation precisely because those pings keep the socket active; routed through the gateway, they're gone.

Reproduction (observed)

A model that delivers its content in a burst near the end of a ~4-minute generation (first content_block_delta at ~235s), behind an intermediary with a 120s idle timeout:

• Direct to Anthropic: completes; max gap between received SSE events ~30s (the ping cadence).
• Through the gateway (before): message_start arrives, then silence; connection cut at ~120s (RemoteProtocolError: peer closed connection without sending complete message body).
• Through the gateway (after): max inter-event gap drops to the keepalive interval (~15s) and the stream completes normally.

Fix

_stream_with_keepalive in pipeline/anthropic_processor.py wraps the emission stream and, whenever the upstream produces no event within STREAM_KEEPALIVE_SECONDS (15s), emits an Anthropic-style event: ping — only after message_start, to preserve wire event ordering. The in-flight __anext__ is shielded from the per-wait timeout so no real event is ever dropped, and the pending read is cancelled on generator close.

Tests

tests/luthien_proxy/unit_tests/pipeline/test_anthropic_processor.py::TestStreamWithKeepalive — 4 cases: keepalives injected during a gap without dropping/reordering real events; no keepalives when fast; empty source terminates; generator close cancels the pending read. Full processor suite passes.