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

switchyard_protocol/
stream.rs
