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

switchyard_protocol/
stream.rs
