diff --git a/crates/switchyard-components/src/backends/anthropic.rs b/crates/switchyard-components/src/backends/anthropic.rs index be76fd01b..26c3b7875 100644 --- a/crates/switchyard-components/src/backends/anthropic.rs +++ b/crates/switchyard-components/src/backends/anthropic.rs @@ -524,9 +524,10 @@ fn anthropic_sse_stream(response: reqwest::Response) -> BoxResponseStream { // Anthropic SSE has named events, but the payload we care about is // still the JSON `data:` line. while let Some(frame) = drain_next_sse_frame(&mut buffer, "Anthropic")? { - match parse_json_sse_frame(&frame, "Anthropic", None)? { + match parse_json_sse_frame(&frame, "Anthropic", Some("[DONE]"))? { ParsedSseFrame::Json(value) => yield StreamEvent::Json(value), - ParsedSseFrame::Done | ParsedSseFrame::Empty => {} + ParsedSseFrame::Done => return, + ParsedSseFrame::Empty => {} } } } @@ -535,7 +536,7 @@ fn anthropic_sse_stream(response: reqwest::Response) -> BoxResponseStream { // separator. if has_non_whitespace_bytes(&buffer) { let frame = decode_sse_frame(&buffer, "Anthropic")?; - match parse_json_sse_frame(&frame, "Anthropic", None)? { + match parse_json_sse_frame(&frame, "Anthropic", Some("[DONE]"))? { ParsedSseFrame::Json(value) => yield StreamEvent::Json(value), ParsedSseFrame::Done | ParsedSseFrame::Empty => {} } diff --git a/crates/switchyard-components/tests/adversarial_native_backends.rs b/crates/switchyard-components/tests/adversarial_native_backends.rs index f3e3b7ce5..f9bc85a83 100644 --- a/crates/switchyard-components/tests/adversarial_native_backends.rs +++ b/crates/switchyard-components/tests/adversarial_native_backends.rs @@ -1040,6 +1040,42 @@ async fn anthropic_streaming_returns_stream_events() -> Result<()> { Ok(()) } +#[tokio::test] +async fn anthropic_streaming_stops_at_done_marker() -> Result<()> { + let server = OneShotServer::sse( + "event: message_start\n\ + data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg-stream\"}}\n\n\ + data: [DONE]\n\n\ + event: message_stop\n\ + data: {\"type\":\"message_stop\"}\n\n", + )?; + let backend = AnthropicNativeBackend::new(anthropic_target(server.base_url().to_string())?)?; + let mut ctx = ProxyContext::new(); + + let response = backend + .call( + &mut ctx, + &ChatRequest::anthropic(json!({ + "model": "client-claude", + "max_tokens": 128, + "messages": [{"role": "user", "content": "stream"}], + "stream": true + })), + ) + .await?; + let events = collect_json_events(response).await?; + server.captured()?; + + assert_eq!( + events, + vec![json!({ + "type": "message_start", + "message": {"id": "msg-stream"} + })] + ); + Ok(()) +} + // Anthropic native must reject targets configured for OpenAI format. #[test] fn anthropic_backend_rejects_openai_targets() -> Result<()> {