switchyard_protocol/
stream.rs

1// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Streaming half of the neutral IR: incremental response chunks ([`LlmResponseChunk`]),
5//! their stream envelope ([`LlmResponseStreamEvent`]), and the streamed response
6//! ([`LlmResponse`]) that carries either a live stream or the terminal [`AggLlmResponse`].
7
8use std::collections::BTreeMap;
9use std::pin::Pin;
10
11use futures::{Stream, StreamExt};
12use http::StatusCode;
13use serde::{Deserialize, Serialize};
14use serde_json::Value;
15
16use crate::{
17    LlmClientError,
18    format::FormatId,
19    llm::{AggLlmResponse, ContentBlock, ResponseOutput, Role, StopReason, ToolCall, Usage},
20};
21
22/// Status reported for an upstream error delivered inside a streaming body. The
23/// upstream already sent a success status line before failing, so there is no real
24/// code to propagate; 502 matches how a failed upstream call surfaces elsewhere.
25const MID_STREAM_UPSTREAM_STATUS: StatusCode = StatusCode::BAD_GATEWAY;
26
27/// A boxed, `Send` stream of response events. Each item may fail independently mid-stream.
28pub type LlmResponseStream =
29    Pin<Box<dyn Stream<Item = Result<LlmResponseStreamEvent, LlmClientError>> + Send>>;
30
31/// Parsed provider event retained for same-format replay.
32#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
33pub struct ProviderStreamEvent {
34    source: FormatId,
35    raw: Value,
36}
37
38impl ProviderStreamEvent {
39    /// Source wire format of the retained event.
40    pub fn source(&self) -> &FormatId {
41        &self.source
42    }
43
44    /// Parsed provider JSON retained for replay.
45    pub fn raw(&self) -> &Value {
46        &self.raw
47    }
48
49    /// Consumes the preservation value into its source format and parsed JSON.
50    pub fn into_parts(self) -> (FormatId, Value) {
51        (self.source, self.raw)
52    }
53}
54
55/// One streaming item crossing the host/algorithm boundary.
56///
57/// `normalized` contains only provider-neutral chunks. `preservation` is opaque to
58/// algorithms and is interpreted by `switchyard-translation` for same-format replay.
59#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
60pub struct LlmResponseStreamEvent {
61    preservation: Option<ProviderStreamEvent>,
62    normalized: Vec<LlmResponseChunk>,
63}
64
65impl LlmResponseStreamEvent {
66    /// Creates an event containing only normalized chunks.
67    pub fn new(normalized: Vec<LlmResponseChunk>) -> Self {
68        Self {
69            preservation: None,
70            normalized,
71        }
72    }
73
74    /// Creates an event with parsed provider JSON retained for replay.
75    pub fn preserved(
76        source: impl Into<FormatId>,
77        raw: Value,
78        normalized: Vec<LlmResponseChunk>,
79    ) -> Self {
80        Self {
81            preservation: Some(ProviderStreamEvent {
82                source: source.into(),
83                raw,
84            }),
85            normalized,
86        }
87    }
88
89    /// Retained provider event, when this event came directly from a provider.
90    pub fn preservation(&self) -> Option<&ProviderStreamEvent> {
91        self.preservation.as_ref()
92    }
93
94    /// Provider-neutral chunks carried by this event.
95    pub fn normalized(&self) -> &[LlmResponseChunk] {
96        &self.normalized
97    }
98
99    /// Consumes the event into its preservation and normalized content.
100    pub fn into_parts(self) -> (Option<ProviderStreamEvent>, Vec<LlmResponseChunk>) {
101        (self.preservation, self.normalized)
102    }
103
104    /// Replaces semantic content and drops raw replay data that no longer describes it.
105    pub fn replace_normalized(self, normalized: Vec<LlmResponseChunk>) -> Self {
106        Self::new(normalized)
107    }
108}
109
110impl From<LlmResponseChunk> for LlmResponseStreamEvent {
111    fn from(chunk: LlmResponseChunk) -> Self {
112        Self::new(vec![chunk])
113    }
114}
115
116/// A model response: either a live [`Stream`](LlmResponse::Stream) of events or a
117/// terminal buffered [`LlmResponse::Agg`] response.
118///
119/// Not `Clone` — the `Stream` variant owns a single-consumption stream. A buffered
120/// backend returns `Agg` directly; a streaming one returns `Stream` and the consumer
121/// drives it, folding to an [`AggLlmResponse`] when it needs the whole response.
122pub enum LlmResponse {
123    /// Live, single-consumption response stream.
124    Stream(LlmResponseStream),
125    /// Fully buffered terminal response.
126    Agg(AggLlmResponse),
127}
128
129impl LlmResponse {
130    /// Borrow the aggregate; `None` while this is still a stream.
131    pub fn as_agg(&self) -> Option<&AggLlmResponse> {
132        match self {
133            LlmResponse::Agg(agg) => Some(agg),
134            LlmResponse::Stream(_) => None,
135        }
136    }
137
138    /// Reduce to the buffered aggregate: return an `Agg` unchanged, or drive a `Stream`
139    /// to completion, folding its normalized chunks into an [`AggLlmResponse`] via
140    /// [`ResponseAccumulator`]. A stream item error aborts with `Err`, as does an
141    /// in-band [`LlmResponseChunk::DecodeError`] (as `ResponseTranslation`) or
142    /// [`LlmResponseChunk::StreamError`] (as `UpstreamHttp`).
143    pub async fn into_agg(self) -> Result<AggLlmResponse, LlmClientError> {
144        match self {
145            LlmResponse::Agg(agg) => Ok(agg),
146            LlmResponse::Stream(mut stream) => {
147                let mut accumulator = ResponseAccumulator::new();
148                while let Some(item) = stream.next().await {
149                    for chunk in item?.normalized {
150                        push_checked_chunk(&mut accumulator, chunk)?;
151                    }
152                }
153                Ok(accumulator.finish())
154            }
155        }
156    }
157
158    /// Returns the model recorded by a buffered response.
159    ///
160    /// Returns `None` for a live stream because reading its model-bearing
161    /// [`LlmResponseChunk::MessageStart`] event would consume the stream.
162    pub fn selected_model(&self) -> Option<&str> {
163        match self {
164            LlmResponse::Agg(agg) => agg.model.as_deref(),
165            // TODO: How do we get the model name on a stream?
166            LlmResponse::Stream(_) => None,
167        }
168    }
169}
170
171impl AggLlmResponse {
172    /// Converts a fully-buffered response into a synthetic chunk stream.
173    ///
174    /// Useful when a caller had to buffer the response (e.g. to judge it) but the
175    /// downstream expects `LlmResponse::Stream` — for instance when `stream: true` was
176    /// requested and the algorithm had to aggregate before it could return.
177    ///
178    /// This conversion is lossy: only text, reasoning, and tool-call content has
179    /// a synthetic chunk representation. Refusals, tool results, media, files,
180    /// unknown blocks, response extensions, and preservation metadata are omitted.
181    pub fn into_stream(self) -> LlmResponseStream {
182        let mut chunks: Vec<LlmResponseChunk> = Vec::new();
183        chunks.push(LlmResponseChunk::MessageStart {
184            id: self.id,
185            model: self.model,
186        });
187        let mut tool_call_index = 0usize;
188        for (output_index, output) in self.outputs.into_iter().enumerate() {
189            for block in output.content {
190                match block {
191                    ContentBlock::Text { text } => {
192                        chunks.push(LlmResponseChunk::TextDelta {
193                            index: output_index,
194                            text,
195                        });
196                    }
197                    ContentBlock::Reasoning { text, details, .. } => {
198                        if !details.is_empty() {
199                            chunks.push(LlmResponseChunk::ReasoningDetailsDelta {
200                                index: output_index,
201                                details,
202                                text,
203                            });
204                        } else {
205                            chunks.push(LlmResponseChunk::ReasoningDelta {
206                                index: output_index,
207                                text,
208                            });
209                        }
210                    }
211                    ContentBlock::ToolCall(tool) => {
212                        let args = serde_json::to_string(&tool.arguments).unwrap_or_default();
213                        chunks.push(LlmResponseChunk::ToolCallDelta {
214                            index: tool_call_index,
215                            id: Some(tool.id),
216                            name: Some(tool.name),
217                            arguments_delta: Some(args),
218                        });
219                        tool_call_index += 1;
220                    }
221                    // Other variants (Image, ToolResult, Refusal, etc.) don't have
222                    // a streaming chunk representation and don't appear in assistant outputs.
223                    _ => {}
224                }
225            }
226            chunks.push(LlmResponseChunk::MessageStop {
227                reason: output.stop_reason.and_then(|r| {
228                    serde_json::to_value(r)
229                        .ok()
230                        .and_then(|v| v.as_str().map(String::from))
231                }),
232            });
233        }
234        chunks.push(LlmResponseChunk::Usage(self.usage));
235        Box::pin(futures::stream::iter(
236            chunks.into_iter().map(|chunk| Ok(chunk.into())),
237        ))
238    }
239}
240
241fn push_checked_chunk(
242    accumulator: &mut ResponseAccumulator,
243    chunk: LlmResponseChunk,
244) -> Result<(), LlmClientError> {
245    match chunk {
246        LlmResponseChunk::DecodeError { message } => {
247            Err(LlmClientError::ResponseTranslation(message))
248        }
249        LlmResponseChunk::StreamError { message } => Err(LlmClientError::UpstreamHttp {
250            status: MID_STREAM_UPSTREAM_STATUS,
251            body: message,
252        }),
253        chunk => {
254            accumulator.push(chunk);
255            Ok(())
256        }
257    }
258}
259
260/// One provider-neutral streaming response chunk.
261#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
262pub enum LlmResponseChunk {
263    /// Starts a response message.
264    MessageStart {
265        /// Provider response identifier.
266        id: Option<String>,
267        /// Model reported by the provider.
268        model: Option<String>,
269    },
270    /// Adds text to one output index.
271    TextDelta {
272        /// Provider output index.
273        index: usize,
274        /// Text fragment.
275        text: String,
276    },
277    /// Adds reasoning text to one output index.
278    ReasoningDelta {
279        /// Provider output index.
280        index: usize,
281        /// Reasoning fragment.
282        text: String,
283    },
284    /// Adds structured reasoning details to one output index.
285    ReasoningDetailsDelta {
286        /// Provider output index.
287        index: usize,
288        /// Reasoning detail objects in provider order.
289        details: Vec<Value>,
290        /// Normalized reasoning text represented by or accompanying the details.
291        text: String,
292    },
293    /// Adds or updates a tool call at one index.
294    ToolCallDelta {
295        /// Tool-call index within the response.
296        index: usize,
297        /// Tool-call identifier, normally supplied by the first delta.
298        id: Option<String>,
299        /// Tool name, normally supplied by the first delta.
300        name: Option<String>,
301        /// Fragment of the serialized tool arguments.
302        arguments_delta: Option<String>,
303    },
304    /// Reports token usage, normally near the end of the stream.
305    Usage(Usage),
306    /// Ends a response message.
307    MessageStop {
308        /// Provider stop-reason string before normalization.
309        reason: Option<String>,
310    },
311    /// Reports that an inbound stream event could not be decoded.
312    DecodeError {
313        /// Human-readable decoding failure.
314        message: String,
315    },
316    /// Reports an upstream failure delivered inside an otherwise successful stream.
317    StreamError {
318        /// Human-readable upstream failure.
319        message: String,
320    },
321}
322
323/// Folds a sequence of [`LlmResponseChunk`]s into the terminal [`AggLlmResponse`].
324///
325/// Text and reasoning deltas concatenate; tool-call deltas assemble by index (name,
326/// id, and a growing arguments string parsed as JSON at the end); `MessageStart`,
327/// `Usage`, and `MessageStop` set the corresponding fields.
328/// [`DecodeError`](LlmResponseChunk::DecodeError) and
329/// [`StreamError`](LlmResponseChunk::StreamError) chunks are ignored here; use
330/// [`LlmResponse::into_agg`] when they must become errors.
331///
332/// Folding is lossy for multiple outputs: text and reasoning indices are ignored,
333/// and [`finish`](Self::finish) produces one assistant output. Prefer
334/// [`LlmResponse::into_agg`] when stream errors must be surfaced.
335///
336/// Drive it by `push`-ing each chunk in order, then call [`finish`](Self::finish).
337#[derive(Default)]
338pub struct ResponseAccumulator {
339    id: Option<String>,
340    model: Option<String>,
341    text: String,
342    reasoning: Option<String>,
343    reasoning_details: Vec<Value>,
344    tool_calls: BTreeMap<usize, PartialToolCall>,
345    usage: Usage,
346    stop_reason: Option<StopReason>,
347}
348
349/// A tool call being assembled from streamed [`LlmResponseChunk::ToolCallDelta`]s.
350#[derive(Default)]
351struct PartialToolCall {
352    id: Option<String>,
353    name: Option<String>,
354    arguments: String,
355}
356
357impl ResponseAccumulator {
358    /// A fresh accumulator with no chunks applied.
359    pub fn new() -> Self {
360        Self::default()
361    }
362
363    /// Apply one chunk. Later `MessageStart`/`Usage`/`MessageStop` fields overwrite
364    /// earlier ones; text, reasoning, and tool-call arguments append.
365    pub fn push(&mut self, chunk: LlmResponseChunk) {
366        match chunk {
367            LlmResponseChunk::MessageStart { id, model } => {
368                if id.is_some() {
369                    self.id = id;
370                }
371                if model.is_some() {
372                    self.model = model;
373                }
374            }
375            LlmResponseChunk::TextDelta { text, .. } => self.text.push_str(&text),
376            LlmResponseChunk::ReasoningDelta { text, .. } => {
377                self.reasoning
378                    .get_or_insert_with(String::new)
379                    .push_str(&text);
380            }
381            LlmResponseChunk::ReasoningDetailsDelta { details, text, .. } => {
382                self.reasoning_details.extend(details);
383                if !text.is_empty() {
384                    self.reasoning
385                        .get_or_insert_with(String::new)
386                        .push_str(&text);
387                }
388            }
389            LlmResponseChunk::ToolCallDelta {
390                index,
391                id,
392                name,
393                arguments_delta,
394            } => {
395                let call = self.tool_calls.entry(index).or_default();
396                if id.is_some() {
397                    call.id = id;
398                }
399                if name.is_some() {
400                    call.name = name;
401                }
402                if let Some(delta) = arguments_delta {
403                    call.arguments.push_str(&delta);
404                }
405            }
406            LlmResponseChunk::Usage(usage) => self.usage = usage,
407            LlmResponseChunk::MessageStop { reason } => {
408                self.stop_reason = Some(stop_reason_from_str(reason.as_deref()));
409            }
410            LlmResponseChunk::DecodeError { .. } | LlmResponseChunk::StreamError { .. } => {}
411        }
412    }
413
414    /// Build the buffered response. Content is ordered reasoning, then text, then
415    /// tool calls (by ascending delta index) — a single assistant output.
416    pub fn finish(self) -> AggLlmResponse {
417        let mut content = Vec::new();
418        if self.reasoning.is_some() || !self.reasoning_details.is_empty() {
419            content.push(ContentBlock::Reasoning {
420                text: self.reasoning.unwrap_or_default(),
421                signature: None,
422                details: self.reasoning_details,
423            });
424        }
425        if !self.text.is_empty() {
426            content.push(ContentBlock::Text { text: self.text });
427        }
428        for call in self.tool_calls.into_values() {
429            content.push(ContentBlock::ToolCall(ToolCall {
430                id: call.id.unwrap_or_default(),
431                name: call.name.unwrap_or_default(),
432                arguments: parse_tool_arguments(&call.arguments),
433            }));
434        }
435        AggLlmResponse {
436            id: self.id,
437            model: self.model,
438            outputs: vec![ResponseOutput {
439                role: Role::Assistant,
440                content,
441                stop_reason: self.stop_reason,
442            }],
443            usage: self.usage,
444            ..AggLlmResponse::default()
445        }
446    }
447}
448
449/// Parse an assembled tool-call arguments string as JSON, falling back to a JSON
450/// string when it is not valid JSON and to an empty object when it is empty.
451fn parse_tool_arguments(arguments: &str) -> Value {
452    if arguments.is_empty() {
453        return Value::Object(serde_json::Map::new());
454    }
455    serde_json::from_str(arguments).unwrap_or_else(|_| Value::String(arguments.to_string()))
456}
457
458/// Map a provider stop-reason string (as carried by [`LlmResponseChunk::MessageStop`])
459/// to a normalized [`StopReason`], covering the common OpenAI and Anthropic spellings.
460fn stop_reason_from_str(reason: Option<&str>) -> StopReason {
461    match reason {
462        Some("length" | "max_tokens") => StopReason::MaxTokens,
463        Some("tool_calls" | "function_call" | "tool_use") => StopReason::ToolUse,
464        Some("content_filter") => StopReason::ContentFilter,
465        Some("stop" | "end_turn" | "stop_sequence") | None => StopReason::EndTurn,
466        Some(_) => StopReason::Unknown,
467    }
468}
469
470#[cfg(test)]
471mod tests {
472    use futures::executor::block_on;
473    use futures::stream;
474
475    use super::*;
476    use serde_json::json;
477
478    fn fold(chunks: Vec<LlmResponseChunk>) -> AggLlmResponse {
479        let mut accumulator = ResponseAccumulator::new();
480        for chunk in chunks {
481            accumulator.push(chunk);
482        }
483        accumulator.finish()
484    }
485
486    #[test]
487    fn folds_text_usage_and_stop_reason() {
488        let agg = fold(vec![
489            LlmResponseChunk::MessageStart {
490                id: Some("id1".to_string()),
491                model: Some("m".to_string()),
492            },
493            LlmResponseChunk::TextDelta {
494                index: 0,
495                text: "Hel".to_string(),
496            },
497            LlmResponseChunk::TextDelta {
498                index: 0,
499                text: "lo".to_string(),
500            },
501            LlmResponseChunk::Usage(Usage {
502                output_tokens: Some(2),
503                ..Usage::default()
504            }),
505            LlmResponseChunk::MessageStop {
506                reason: Some("length".to_string()),
507            },
508        ]);
509        assert_eq!(agg.id.as_deref(), Some("id1"));
510        assert_eq!(agg.model.as_deref(), Some("m"));
511        assert_eq!(agg.usage.output_tokens, Some(2));
512        assert_eq!(agg.outputs[0].stop_reason, Some(StopReason::MaxTokens));
513        assert_eq!(
514            agg.outputs[0].content,
515            vec![ContentBlock::Text {
516                text: "Hello".to_string()
517            }]
518        );
519    }
520
521    #[test]
522    fn aggregates_normalized_chunks_inside_stream_event() {
523        let response = LlmResponse::Stream(Box::pin(stream::iter([Ok(
524            LlmResponseStreamEvent::preserved(
525                crate::WireFormat::OpenAiChat,
526                json!({
527                "choices": [{"delta": {"content": "hello"}}],
528                "system_fingerprint": "fp_exact"
529                }),
530                vec![LlmResponseChunk::TextDelta {
531                    index: 0,
532                    text: "hello".to_string(),
533                }],
534            ),
535        )])));
536        let aggregate = block_on(response.into_agg()).expect("stream event should aggregate");
537
538        assert_eq!(
539            aggregate.outputs[0].content,
540            vec![ContentBlock::Text {
541                text: "hello".to_string()
542            }]
543        );
544    }
545
546    #[test]
547    fn replacing_normalized_content_drops_preservation() {
548        let event = LlmResponseStreamEvent::preserved(
549            crate::WireFormat::OpenAiChat,
550            json!({"choices": [{"delta": {"content": "old"}}]}),
551            vec![LlmResponseChunk::TextDelta {
552                index: 0,
553                text: "old".to_string(),
554            }],
555        )
556        .replace_normalized(vec![LlmResponseChunk::TextDelta {
557            index: 0,
558            text: "new".to_string(),
559        }]);
560
561        assert!(event.preservation().is_none());
562        assert_eq!(
563            event.normalized(),
564            &[LlmResponseChunk::TextDelta {
565                index: 0,
566                text: "new".to_string(),
567            }]
568        );
569    }
570
571    #[test]
572    fn stream_errors_inside_preserved_events_remain_typed() {
573        let response = LlmResponse::Stream(Box::pin(stream::iter([Ok(
574            LlmResponseStreamEvent::preserved(
575                crate::WireFormat::OpenAiChat,
576                json!({"error": {"message": "provider failed"}}),
577                vec![LlmResponseChunk::StreamError {
578                    message: "provider failed".to_string(),
579                }],
580            ),
581        )])));
582
583        let error = block_on(response.into_agg()).err();
584        assert!(matches!(
585            error,
586            Some(LlmClientError::UpstreamHttp {
587                status: MID_STREAM_UPSTREAM_STATUS,
588                ..
589            })
590        ));
591    }
592
593    #[test]
594    fn assembles_tool_calls_by_index() {
595        // id/name arrive once, arguments stream across deltas and parse as JSON.
596        let agg = fold(vec![
597            LlmResponseChunk::ToolCallDelta {
598                index: 0,
599                id: Some("call_1".to_string()),
600                name: Some("lookup".to_string()),
601                arguments_delta: Some("{\"q\":".to_string()),
602            },
603            LlmResponseChunk::ToolCallDelta {
604                index: 0,
605                id: None,
606                name: None,
607                arguments_delta: Some("\"rust\"}".to_string()),
608            },
609            LlmResponseChunk::MessageStop {
610                reason: Some("tool_calls".to_string()),
611            },
612        ]);
613        assert_eq!(agg.outputs[0].stop_reason, Some(StopReason::ToolUse));
614        assert_eq!(
615            agg.outputs[0].content,
616            vec![ContentBlock::ToolCall(ToolCall {
617                id: "call_1".to_string(),
618                name: "lookup".to_string(),
619                arguments: json!({"q": "rust"}),
620            })]
621        );
622    }
623
624    #[test]
625    fn reasoning_precedes_text_in_content() {
626        let agg = fold(vec![
627            LlmResponseChunk::ReasoningDelta {
628                index: 0,
629                text: "think".to_string(),
630            },
631            LlmResponseChunk::TextDelta {
632                index: 0,
633                text: "answer".to_string(),
634            },
635        ]);
636        assert_eq!(
637            agg.outputs[0].content,
638            vec![
639                ContentBlock::Reasoning {
640                    text: "think".to_string(),
641                    signature: None,
642                    details: Vec::new(),
643                },
644                ContentBlock::Text {
645                    text: "answer".to_string(),
646                },
647            ]
648        );
649    }
650
651    #[test]
652    fn into_stream_round_trips_through_into_agg() {
653        let original = AggLlmResponse {
654            id: Some("id1".to_string()),
655            model: Some("m".to_string()),
656            outputs: vec![ResponseOutput {
657                role: Role::Assistant,
658                content: vec![ContentBlock::Text {
659                    text: "hello".to_string(),
660                }],
661                stop_reason: Some(StopReason::EndTurn),
662            }],
663            usage: Usage {
664                output_tokens: Some(3),
665                ..Usage::default()
666            },
667            ..AggLlmResponse::default()
668        };
669        let stream = LlmResponse::Stream(original.clone().into_stream());
670        let recovered = block_on(stream.into_agg()).expect("into_agg failed");
671        assert_eq!(recovered.id, original.id);
672        assert_eq!(recovered.model, original.model);
673        assert_eq!(recovered.usage.output_tokens, original.usage.output_tokens);
674        assert_eq!(
675            recovered.outputs[0].stop_reason,
676            original.outputs[0].stop_reason
677        );
678        assert_eq!(recovered.outputs[0].content, original.outputs[0].content);
679    }
680
681    #[test]
682    fn into_stream_retains_encrypted_reasoning_and_text() {
683        let details = vec![json!({
684            "type": "reasoning.encrypted",
685            "data": "opaque-encrypted-reasoning"
686        })];
687        let original = AggLlmResponse {
688            outputs: vec![ResponseOutput {
689                role: Role::Assistant,
690                content: vec![ContentBlock::Reasoning {
691                    text: "fallback reasoning".to_string(),
692                    signature: None,
693                    details: details.clone(),
694                }],
695                stop_reason: Some(StopReason::EndTurn),
696            }],
697            ..AggLlmResponse::default()
698        };
699
700        let recovered = block_on(LlmResponse::Stream(original.into_stream()).into_agg())
701            .expect("into_agg failed");
702        let ContentBlock::Reasoning {
703            text,
704            details: recovered_details,
705            ..
706        } = &recovered.outputs[0].content[0]
707        else {
708            panic!("expected reasoning block");
709        };
710        assert_eq!(text, "fallback reasoning");
711        assert_eq!(recovered_details, &details);
712    }
713
714    #[test]
715    fn into_agg_preserves_stream_item_error() {
716        let response = LlmResponse::Stream(Box::pin(stream::once(async {
717            Err(LlmClientError::Timeout {
718                source: Box::new(std::io::Error::other("timed out")),
719            })
720        })));
721
722        let Err(error) = block_on(response.into_agg()) else {
723            panic!("expected stream aggregation to fail");
724        };
725        assert!(matches!(error, LlmClientError::Timeout { .. }));
726    }
727}
switchyard_protocol/stream.rs

switchyard_protocol/
stream.rs
