diff --git a/crates/switchyard-server/tests/server.rs b/crates/switchyard-server/tests/server.rs index 4c1d5d227..42aac1665 100644 --- a/crates/switchyard-server/tests/server.rs +++ b/crates/switchyard-server/tests/server.rs @@ -1488,6 +1488,109 @@ async fn streaming_error_records_error_without_usage_or_latency() -> TestResult Ok(()) } +#[tokio::test] +async fn responses_stream_error_does_not_emit_success_terminal_events() -> TestResult { + // A distinct target keeps this test's error-counter increments off the shared + // model/stream-error metric that streaming_error_records_... asserts an exact delta on. + let (_upstream, app) = test_app(&[(ROUTE_MODEL, &["model/responses-stream-error"])]).await?; + + let response = send( + &app, + "POST", + "/v1/responses", + Some(json!({ + "model": ROUTE_MODEL, + "input": "stream-error", + "stream": true + })), + ) + .await?; + + assert_eq!(response.status, StatusCode::OK); + let body = response.text()?; + assert_in_order(body, &["before", "upstream stream failed"]); + for event_type in [ + "response.content_part.done", + "response.output_item.done", + "response.completed", + ] { + assert!( + !body.contains(event_type), + "{event_type} followed an upstream stream error" + ); + } + Ok(()) +} + +#[tokio::test] +async fn chat_stream_error_does_not_emit_success_terminal_chunk() -> TestResult { + // A distinct target keeps this test's error-counter increments off the shared + // model/stream-error metric that streaming_error_records_... asserts an exact delta on. + let (_upstream, app) = test_app(&[(ROUTE_MODEL, &["model/chat-stream-error"])]).await?; + + let response = send( + &app, + "POST", + "/v1/chat/completions", + Some(json!({ + "model": ROUTE_MODEL, + "messages": [{"role": "user", "content": "stream-error"}], + "stream": true + })), + ) + .await?; + + assert_eq!(response.status, StatusCode::OK); + let body = response.text()?; + assert_in_order(body, &["before", "still here", "upstream stream failed"]); + // The finalizer must not synthesize a `finish_reason: stop` completion chunk after the error. + let after_error = body + .split_once("upstream stream failed") + .map(|(_, rest)| rest) + .unwrap_or_default(); + assert!( + !after_error.contains(r#""finish_reason":"stop""#), + "a finish_reason=stop chunk followed an upstream stream error:\n{body}" + ); + Ok(()) +} + +#[tokio::test] +async fn anthropic_stream_error_does_not_emit_success_terminal_events() -> TestResult { + // A distinct target keeps this test's error-counter increments off the shared + // model/stream-error metric that streaming_error_records_... asserts an exact delta on. + let (_upstream, app) = test_app(&[(ROUTE_MODEL, &["model/anthropic-stream-error"])]).await?; + + let response = send( + &app, + "POST", + "/v1/messages", + Some(json!({ + "model": ROUTE_MODEL, + "messages": [{"role": "user", "content": "stream-error"}], + "max_tokens": 16, + "stream": true + })), + ) + .await?; + + assert_eq!(response.status, StatusCode::OK); + let body = response.text()?; + assert_in_order(body, &["before", "upstream stream failed"]); + // The finalizer must not close the turn with message_delta/message_stop after the error. + let after_error = body + .split_once("upstream stream failed") + .map(|(_, rest)| rest) + .unwrap_or_default(); + for event_type in ["message_delta", "message_stop"] { + assert!( + !after_error.contains(event_type), + "{event_type} followed an upstream stream error:\n{body}" + ); + } + Ok(()) +} + #[tokio::test] async fn request_and_upstream_errors_use_the_inbound_wire_format() -> TestResult { let (_upstream, app) = test_app(&[(ROUTE_MODEL, &["model/a"])]).await?; diff --git a/crates/switchyard-translation/src/codecs/anthropic/stream.rs b/crates/switchyard-translation/src/codecs/anthropic/stream.rs index 6cb3d1868..450d796e1 100644 --- a/crates/switchyard-translation/src/codecs/anthropic/stream.rs +++ b/crates/switchyard-translation/src/codecs/anthropic/stream.rs @@ -147,6 +147,10 @@ fn encode_anthropic_stream( state: &mut StreamTranslationState, event: LlmResponseChunk, ) -> Vec { + // An in-band error is terminal: once the error is emitted, drop every later chunk. + if state.errored { + return Vec::new(); + } match event { LlmResponseChunk::MessageStart { id, model } => { record_source_identity(state, id, model); @@ -204,6 +208,9 @@ fn encode_anthropic_stream( Vec::new() } LlmResponseChunk::StreamError { message } | LlmResponseChunk::DecodeError { message } => { + // An in-band error is terminal: emit the error, then nothing further. + state.finished = true; // finish() adds no success events + state.errored = true; // the entry guard drops any later chunk vec![json!({"type": "error", "error": {"message": message}})] } } diff --git a/crates/switchyard-translation/src/codecs/openai_chat/stream.rs b/crates/switchyard-translation/src/codecs/openai_chat/stream.rs index 5ea2f10fd..b4c6273b1 100644 --- a/crates/switchyard-translation/src/codecs/openai_chat/stream.rs +++ b/crates/switchyard-translation/src/codecs/openai_chat/stream.rs @@ -153,6 +153,10 @@ fn encode_openai_chat_stream( state: &mut StreamTranslationState, event: LlmResponseChunk, ) -> Vec { + // An in-band error is terminal: once the error is emitted, drop every later chunk. + if state.errored { + return Vec::new(); + } match event { LlmResponseChunk::MessageStart { id, model } => { record_source_identity(state, id, model); @@ -224,6 +228,9 @@ fn encode_openai_chat_stream( )] } LlmResponseChunk::DecodeError { message } | LlmResponseChunk::StreamError { message } => { + // An in-band error is terminal: emit the error, then nothing further. + state.finished = true; // finish() adds no success events + state.errored = true; // the entry guard drops any later chunk vec![json!({"error": {"message": message}})] } } diff --git a/crates/switchyard-translation/src/codecs/responses/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index cd531cb60..8a7d324ec 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -157,6 +157,10 @@ fn encode_responses_stream( state: &mut StreamTranslationState, event: LlmResponseChunk, ) -> Vec { + // An in-band error is terminal: once the error is emitted, drop every later chunk. + if state.errored { + return Vec::new(); + } match event { LlmResponseChunk::MessageStart { id, model } => { record_source_identity(state, id, model); @@ -182,6 +186,9 @@ fn encode_responses_stream( Vec::new() } LlmResponseChunk::DecodeError { message } | LlmResponseChunk::StreamError { message } => { + // An in-band error is terminal: emit the error, then nothing further. + state.finished = true; // finish() adds no success events + state.errored = true; // the entry guard drops any later chunk vec![json!({"type": "error", "message": message})] } } diff --git a/crates/switchyard-translation/src/codecs/stream.rs b/crates/switchyard-translation/src/codecs/stream.rs index 79782907c..bcd586a27 100644 --- a/crates/switchyard-translation/src/codecs/stream.rs +++ b/crates/switchyard-translation/src/codecs/stream.rs @@ -35,6 +35,8 @@ pub struct StreamTranslationState { pub saw_message_start: bool, pub emitted_message_start: bool, pub finished: bool, + /// Set once an in-band error event was emitted; the encoder then emits nothing further. + pub errored: bool, pub usage: Usage, pub(crate) output_tokens_seen: u64, @@ -246,6 +248,9 @@ pub(crate) fn encode_response_stream_event( target: &FormatId, event: crate::LlmResponseStreamEvent, ) -> Vec { + if state.errored { + return Vec::new(); + } let (preservation, normalized) = event.into_parts(); if let Some(preservation) = preservation { let (source, raw) = preservation.into_parts(); diff --git a/crates/switchyard-translation/src/helpers.rs b/crates/switchyard-translation/src/helpers.rs index 5136825c8..f1fc7c321 100644 --- a/crates/switchyard-translation/src/helpers.rs +++ b/crates/switchyard-translation/src/helpers.rs @@ -117,6 +117,9 @@ pub fn encode_stream( ); yield value; } + if state.errored { + return; + } } for mut value in codec.finish(&mut state) { stamp_streamed_response_model( @@ -449,6 +452,123 @@ mod tests { Ok(()) } + // An in-band error is terminal for every target format: the encoder emits the pre-error + // content and the error, then drops any later chunk. Production truncates the source before + // the encoder, so this contract is only observable by driving encode_stream directly. + #[test] + fn encode_stream_stops_after_an_in_band_error() -> Result<(), BoxError> { + for message in [ + LlmResponseChunk::StreamError { + message: "boom".to_string(), + }, + LlmResponseChunk::DecodeError { + message: "boom".to_string(), + }, + ] { + for target in [ + WireFormat::OpenAiChat, + WireFormat::OpenAiResponses, + WireFormat::AnthropicMessages, + ] { + let chunks: LlmResponseStream = stream::iter(vec![ + Ok(LlmResponseChunk::TextDelta { + index: 0, + text: "before".to_string(), + } + .into()), + Ok(message.clone().into()), + Ok(LlmResponseChunk::TextDelta { + index: 0, + text: "after".to_string(), + } + .into()), + ]) + .boxed(); + let events = block_on(encode_stream(chunks, target, None)?.collect::>()) + .into_iter() + .collect::, BoxError>>()?; + let body = serde_json::to_string(&events)?; + assert!( + body.contains("before"), + "{target:?}: pre-error content missing:\n{body}" + ); + assert!( + body.contains("boom"), + "{target:?}: error event missing:\n{body}" + ); + assert!( + !body.contains("after"), + "{target:?}/{message:?}: content leaked after the error:\n{body}" + ); + } + } + Ok(()) + } + + // A replayed provider error ends the stream before the encoder polls the source again. + #[test] + fn encode_stream_stops_polling_after_a_replayed_error() -> Result<(), BoxError> { + let error = LlmResponseStreamEvent::preserved( + WireFormat::OpenAiResponses, + json!({"type": "error", "message": "boom"}), + vec![LlmResponseChunk::StreamError { + message: "boom".to_string(), + }], + ); + let chunks: LlmResponseStream = stream::iter([Ok(error)]) + .chain(stream::poll_fn(|_| { + panic!("encode_stream polled the source after an in-band error") + })) + .boxed(); + + let events = + block_on(encode_stream(chunks, WireFormat::OpenAiResponses, None)?.collect::>()) + .into_iter() + .collect::, BoxError>>()?; + + assert_eq!(events, vec![json!({"type": "error", "message": "boom"})]); + Ok(()) + } + + // The guard keys on `errored`, not `finished`, so a normal completion still emits the + // trailing usage chunk the OpenAI chat codec reports only after `finished` is set. + #[test] + fn encode_stream_keeps_trailing_usage_after_a_normal_stop() -> Result<(), BoxError> { + let chunks: LlmResponseStream = stream::iter(vec![ + Ok(LlmResponseChunk::TextDelta { + index: 0, + text: "hi".to_string(), + } + .into()), + Ok(LlmResponseChunk::MessageStop { + reason: Some("stop".to_string()), + } + .into()), + Ok(LlmResponseChunk::Usage(switchyard_protocol::llm::Usage { + output_tokens: Some(7), + ..Default::default() + }) + .into()), + ]) + .boxed(); + let events = + block_on(encode_stream(chunks, WireFormat::OpenAiChat, None)?.collect::>()) + .into_iter() + .collect::, BoxError>>()?; + let body = serde_json::to_string(&events)?; + assert!( + events + .iter() + .any(|event| event["choices"][0]["finish_reason"] == "stop"), + "missing stop terminal:\n{body}" + ); + assert!( + body.contains("\"usage\""), + "trailing usage dropped after a normal stop:\n{body}" + ); + Ok(()) + } + #[test] fn decode_stream_parses_sse_bytes_into_ir_chunks() -> Result<(), LlmClientError> { let sse = b"data: {\"choices\":[{\"delta\":{\"content\":\"Hello\"}}]}\n\n\ diff --git a/crates/switchyard-translation/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index 5ebd784b0..3aeed7159 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -63,6 +63,36 @@ fn preserved_same_format_events_replay_unknown_fields_exactly() -> TestResult { Ok(()) } +// A same-format error remains the last replayed event even if the source supplies more frames. +#[test] +fn preserved_same_format_replay_stops_after_an_error() -> TestResult { + let format = WireFormat::OpenAiResponses; + let events = [ + json!({"type": "response.output_text.delta", "delta": "before"}), + json!({"type": "error", "message": "boom"}), + json!({"type": "response.output_text.delta", "delta": "after"}), + json!({"type": "response.completed", "response": {"id": "resp_1"}}), + ]; + let engine = TranslationEngine::default(); + let mut decode_state = StreamTranslationState::new(format, format); + let mut encode_state = StreamTranslationState::new(format, format); + let mut replayed = Vec::new(); + + for event in events { + let preserved = engine.decode_stream_event(&mut decode_state, format, event)?; + replayed.extend(engine.encode_stream_event(&mut encode_state, format, preserved)?); + } + + assert_eq!( + replayed, + vec![ + json!({"type": "response.output_text.delta", "delta": "before"}), + json!({"type": "error", "message": "boom"}), + ] + ); + Ok(()) +} + // Replay emits the preserved event without running the encoder, so the encoder never sees the // stop it would normally record. Replay must still leave the stream marked finished or // `finish_stream` synthesizes a terminal the client already received.