CLAUDE-A follow-up: native Claude SSE in the proxy

[nuniesmith]

Summary

CLAUDE-A's open follow-up. The proxy's Claude streaming arm has been synthesising a single delta from a blocking send_message ever since CLAUDE-A landed — it worked but defeated the purpose of streaming (clients saw one giant chunk after the full response materialised). This switches it to real SSE via AnthropicClient::stream_message.

What changed

handle_streaming's Claude arm now drives stream.next_event() in a loop and translates each StreamEvent:

Anthropic event	Proxy action
MessageStart	capture resolved model slug + initial Usage
ContentBlockDelta::TextDelta	forward as StreamChunk::Delta(text)
ContentBlockDelta::{InputJson,Thinking,Signature}	drop (matches non-streaming extract_text)
MessageDelta	update latest_usage (cumulative output + cache totals)
MessageStop	flush StreamChunk::Done via send_claude_done
Stream exhausted with no MessageStop	still flush Done so the cache write tail fires
Mid-stream Err	emit StreamChunk::Error and exit

Plus:

• send_claude_done centralises the Usage → StreamChunk::Done translation so the two terminal paths agree. Keeps cache token counts honest by emitting None when Anthropic reported zero, rather than forcing Some(0) which would lie about cache participation downstream.
• Channel buffer 4 → 64. A real Claude stream produces dozens of small deltas; the prior 4-slot bound would back-pressure the reqwest chunk reader on every token.
• Receiver-dropped detection. tx.send().await.is_err() on a TextDelta returns early from the pump task, freeing the upstream socket without waiting for MessageStop — important for clients that disconnect mid-stream.
• The synthesised path is removed entirely. If we later need a blocking fallback (e.g. provider can't stream), it can return as a branch, but right now it's dead weight.

Tests

• send_claude_done_emits_cache_tokens_only_when_nonzero — verifies the Usage → Done translation, including the then_some guard that drops zero cache-read counts.
• send_claude_done_handles_missing_usage_gracefully — Done still fires with all-None token fields when the stream ended before any usage event arrived.

End-to-end SSE behaviour isn't unit-tested — the proxy module has no mock-server scaffolding yet. Manual smoke test:

curl --no-buffer -N \
  -H "Authorization: Bearer $RUSTCODE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model":"claude-sonnet-4-6","messages":[{"role":"user","content":"Count to ten"}],"stream":true}' \
  http://localhost:3500/v1/chat/completions

Expected: incremental SSE frames (one per text token batch), terminated by a final frame with finish_reason:"stop" carrying cache_creation_input_tokens / cache_read_input_tokens extension fields and the data: [DONE] sentinel.

Test plan

• cargo check --workspace --tests --exclude rustcode --exclude rag → clean
• cargo check -p rustcode --lib blocked locally by the known ort-sys CDN sandbox restriction; CI verifies
• New send_claude_done unit tests cover the two terminal paths
• Live curl --no-buffer smoke test against a deployed /v1/chat/completions with a configured ANTHROPIC_API_KEY

TODO

CLAUDE-A gains a "Native SSE follow-up done 2026-05-25" addendum.

https://claude.ai/code/session_01CWvyqRpgnpgZk2sXfMpzfL

---

Generated by Claude Code

Summary by CodeRabbit

• Improvements

  • Enhanced Claude streaming functionality with optimized buffer handling for improved performance during message bursts and more accurate token and cache counting

• Tests

  • Added test coverage for streaming behavior and edge cases in token and cache handling

[coderabbitai]

Caution

Review failed

The pull request is closed.

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 20853b74-30c1-4f8e-b78b-5d56cd6d65a0

📥 Commits

Reviewing files that changed from the base of the PR and between f2c93da and 80f5b66.

📒 Files selected for processing (2)

• TODO.md
• src/api/proxy.rs

---

📝 Walkthrough

Walkthrough

The Claude streaming path in the OpenAI /v1/chat/completions proxy is refactored to consume AnthropicClient::stream_message events directly, forwarding only text deltas as StreamChunk::Delta and accumulating model/usage metadata. A new send_claude_done helper centralizes mapping of Anthropic Usage (with optional cache tokens) to the terminal StreamChunk::Done, filtering zero values. Channel buffer increased to 64 for burst absorption.

Changes

Claude Native Streaming

Layer / File(s)	Summary
Streaming initialization, imports, and event loop src/api/proxy.rs	Imports StreamEvent, Usage, PromptCache, SystemBlock; increases mpsc channel buffer from 4 to 64; replaces prior synthetic delta logic with an event-driven loop consuming AnthropicClient::stream_message, forwarding text deltas only, recording model_used from MessageStart, accumulating usage from MessageDelta, and emitting StreamChunk::Done on MessageStop or stream exhaustion.
Terminal chunk helper and token filtering src/api/proxy.rs	Adds send_claude_done helper that maps Anthropic Usage (including optional cache creation/read tokens) to StreamChunk::Done, omitting cache token fields when their counts are zero. Unit tests verify correct omission of zero-valued fields and graceful handling of missing usage.
Documentation update TODO.md	Updates documentation to confirm native SSE implementation: notes direct forwarding of AnthropicClient::stream_message deltas, usage accumulation across events, conditional cache token emission, non-text delta dropping, and 64-element channel buffer for burst handling.

Sequence Diagram

sequenceDiagram
  participant Client
  participant ProxyHandler
  participant AnthropicClient
  participant StreamLoop
  participant SendClaudeDone
  
  Client->>ProxyHandler: POST /v1/chat/completions
  ProxyHandler->>AnthropicClient: stream_message(request)
  activate StreamLoop
  loop Process Stream Events
    AnthropicClient-->>StreamLoop: MessageStart {model}
    StreamLoop->>StreamLoop: record model_used
    AnthropicClient-->>StreamLoop: MessageDelta {text delta}
    StreamLoop->>Client: StreamChunk::Delta
    AnthropicClient-->>StreamLoop: MessageDelta {usage, cache}
    StreamLoop->>StreamLoop: accumulate Usage
  end
  AnthropicClient-->>StreamLoop: MessageStop or exhaustion
  deactivate StreamLoop
  StreamLoop->>SendClaudeDone: Usage {creation/read tokens}
  SendClaudeDone->>SendClaudeDone: omit zero cache token fields
  SendClaudeDone->>Client: StreamChunk::Done

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~45 minutes

Possibly related PRs

• nuniesmith/rustcode#1: Establishes the earlier Claude routing/dispatch and cache-token plumbing that this PR extends by refactoring the streaming path to native AnthropicClient::stream_message and send_claude_done mapping.
• nuniesmith/rustcode#47: Ensures the proxy sends SystemBlock/build_system_blocks with cache_control so Anthropic reports cache read tokens, which the new send_claude_done helper now properly emits in terminal chunks.

Poem

🐰 Streams of events flow like spring water clear,
Text deltas forward, cache counts precise,
No synthetic tricks, just native SSE,
Buffered for bursts at sixty-four strong,
A rabbit's delight: clean streaming done right! ✨

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch claude/proxy-claude-native-sse-2026-05-25

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}