fix(translation): stop streams after in-band errors

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?;

crates/switchyard-translation/src/codecs/anthropic/stream.rs

@@ -147,6 +147,10 @@ fn encode_anthropic_stream(
     state: &mut StreamTranslationState,
     event: LlmResponseChunk,
 ) -> Vec<Value> {
+    // 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}})]
         }
     }

crates/switchyard-translation/src/codecs/openai_chat/stream.rs

@@ -153,6 +153,10 @@ fn encode_openai_chat_stream(
     state: &mut StreamTranslationState,
     event: LlmResponseChunk,
 ) -> Vec<Value> {
+    // 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}})]
         }
     }

crates/switchyard-translation/src/codecs/responses/stream.rs

@@ -157,6 +157,10 @@ fn encode_responses_stream(
     state: &mut StreamTranslationState,
     event: LlmResponseChunk,
 ) -> Vec<Value> {
+    // 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})]
         }
     }

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<Value> {
+    if state.errored {
+        return Vec::new();
+    }
     let (preservation, normalized) = event.into_parts();
     if let Some(preservation) = preservation {
         let (source, raw) = preservation.into_parts();

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::<Vec<_>>())
+                    .into_iter()
+                    .collect::<Result<Vec<Value>, 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::<Vec<_>>())
+                .into_iter()
+                .collect::<Result<Vec<Value>, 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::<Vec<_>>())
+                .into_iter()
+                .collect::<Result<Vec<Value>, 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\

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.