openrouter_rs/types/
stream.rs

1//! # Tool-Aware Streaming
2//!
3//! This module provides [`ToolAwareStream`], a wrapper around the raw SSE
4//! stream that automatically accumulates partial tool call fragments into
5//! complete [`ToolCall`] objects while still yielding text and reasoning
6//! content deltas in real time.
7//!
8//! ## Problem
9//!
10//! When the OpenRouter API streams a response that includes tool calls,
11//! the tool call data arrives incrementally across many SSE chunks:
12//!
13//! - Chunk 1: `{index: 0, id: "call_abc", type: "function", function: {name: "get_weather", arguments: ""}}`
14//! - Chunk 2: `{index: 0, function: {arguments: "{\"loc"}}`
15//! - Chunk 3: `{index: 0, function: {arguments: "ation\":"}}`
16//! - Chunk N: `{index: 0, function: {arguments: " \"NYC\"}"}}`
17//!
18//! The raw stream yields these as [`PartialToolCall`] fragments that cannot
19//! be used directly. `ToolAwareStream` handles merging them by `index`.
20//!
21//! ## Solution
22//!
23//! Wrap the raw stream in a `ToolAwareStream` to get a stream of
24//! [`StreamEvent`] values:
25//!
26//! ```rust,no_run
27//! use futures_util::StreamExt;
28//! use openrouter_rs::types::stream::{ToolAwareStream, StreamEvent};
29//!
30//! # async fn example(client: openrouter_rs::OpenRouterClient, request: openrouter_rs::api::chat::ChatCompletionRequest) -> Result<(), Box<dyn std::error::Error>> {
31//! let raw_stream = client.chat().stream(&request).await?;
32//! let mut stream = ToolAwareStream::new(raw_stream);
33//!
34//! while let Some(event) = stream.next().await {
35//!     match event {
36//!         StreamEvent::ContentDelta(text) => print!("{}", text),
37//!         StreamEvent::ReasoningDelta(text) => { /* reasoning content */ },
38//!         StreamEvent::Done { tool_calls, .. } => {
39//!             for tc in &tool_calls {
40//!                 println!("Tool: {} args: {}", tc.name(), tc.arguments_json());
41//!             }
42//!         },
43//!         StreamEvent::Error(e) => eprintln!("Error: {}", e),
44//!         _ => {}
45//!     }
46//! }
47//! # Ok(())
48//! # }
49//! ```
50
51use std::collections::{BTreeMap, VecDeque};
52use std::pin::Pin;
53use std::task::{Context, Poll};
54
55use futures_util::stream::BoxStream;
56use futures_util::{Stream, StreamExt};
57use serde_json::Value;
58
59use crate::error::OpenRouterError;
60use crate::types::completion::{
61    CompletionsResponse, FunctionCall, PartialToolCall, ReasoningDetail, ResponseUsage, ToolCall,
62};
63use crate::{
64    api::{
65        messages::{AnthropicContentPart, AnthropicMessagesSseEvent, AnthropicMessagesStreamEvent},
66        responses::ResponsesStreamEvent,
67    },
68    types::completion::FinishReason,
69};
70
71/// Events emitted by [`ToolAwareStream`].
72///
73/// Content and reasoning deltas are yielded immediately as they arrive.
74/// Tool calls are accumulated internally and emitted as complete objects
75/// only once in the final [`StreamEvent::Done`] event.
76#[derive(Debug)]
77#[non_exhaustive]
78pub enum StreamEvent {
79    /// A fragment of text content from the assistant's response.
80    ContentDelta(String),
81
82    /// A fragment of reasoning/chain-of-thought content.
83    ReasoningDelta(String),
84
85    /// Structured reasoning detail blocks (e.g., encrypted reasoning).
86    ReasoningDetailsDelta(Vec<ReasoningDetail>),
87
88    /// The stream has finished. Contains all accumulated data.
89    ///
90    /// `tool_calls` will be empty if the model did not invoke any tools.
91    /// `usage` is typically only present in the final SSE chunk.
92    Done {
93        /// Fully assembled tool calls (empty if none were requested).
94        tool_calls: Vec<ToolCall>,
95        /// The reason the model stopped generating.
96        finish_reason: Option<FinishReason>,
97        /// Token usage statistics (if provided by the API).
98        usage: Option<ResponseUsage>,
99        /// The response ID from the API.
100        id: String,
101        /// The model that generated the response.
102        model: String,
103    },
104
105    /// An error occurred while processing the stream.
106    Error(OpenRouterError),
107}
108
109/// Internal accumulator for a single tool call being assembled from
110/// streaming fragments.
111#[derive(Debug, Clone, Default)]
112struct ToolCallAccumulator {
113    index: Option<u32>,
114    id: Option<String>,
115    type_: Option<String>,
116    name: Option<String>,
117    arguments: String,
118}
119
120impl ToolCallAccumulator {
121    /// Merge a partial tool call fragment into this accumulator.
122    fn merge(&mut self, partial: &PartialToolCall) {
123        if partial.index.is_some() {
124            self.index = partial.index;
125        }
126        if let Some(id) = &partial.id {
127            self.id = Some(id.clone());
128        }
129        if let Some(type_) = &partial.type_ {
130            self.type_ = Some(type_.clone());
131        }
132        if let Some(func) = &partial.function {
133            if let Some(name) = &func.name {
134                self.name = Some(name.clone());
135            }
136            if let Some(args) = &func.arguments {
137                self.arguments.push_str(args);
138            }
139        }
140    }
141
142    /// Try to convert this accumulator into a complete [`ToolCall`].
143    ///
144    /// Returns `None` if required fields (`id`, `name`) are still missing,
145    /// which would indicate an incomplete stream.
146    fn into_tool_call(self) -> Option<ToolCall> {
147        Some(ToolCall {
148            id: self.id?,
149            type_: self.type_.unwrap_or_else(|| "function".to_string()),
150            function: FunctionCall {
151                name: self.name?,
152                arguments: self.arguments,
153            },
154            index: self.index,
155        })
156    }
157}
158
159/// A stream wrapper that accumulates partial tool call fragments and
160/// yields [`StreamEvent`] values.
161///
162/// Text content and reasoning deltas are forwarded immediately. Tool call
163/// chunks are buffered internally and assembled into complete [`ToolCall`]
164/// objects, which are emitted in the final [`StreamEvent::Done`] event.
165///
166/// # Construction
167///
168/// Wrap any raw streaming response from
169/// [`stream_chat_completion`](crate::api::chat::stream_chat_completion):
170///
171/// ```rust,no_run
172/// # async fn example(client: openrouter_rs::OpenRouterClient, request: openrouter_rs::api::chat::ChatCompletionRequest) -> Result<(), Box<dyn std::error::Error>> {
173/// use openrouter_rs::types::stream::ToolAwareStream;
174///
175/// let raw = client.chat().stream(&request).await?;
176/// let stream = ToolAwareStream::new(raw);
177/// # Ok(())
178/// # }
179/// ```
180///
181/// Or use the convenience method on the client:
182///
183/// ```rust,no_run
184/// # async fn example(client: openrouter_rs::OpenRouterClient, request: openrouter_rs::api::chat::ChatCompletionRequest) -> Result<(), Box<dyn std::error::Error>> {
185/// let stream = client.stream_chat_completion_tool_aware(&request).await?;
186/// # Ok(())
187/// # }
188/// ```
189pub struct ToolAwareStream {
190    inner: BoxStream<'static, Result<CompletionsResponse, OpenRouterError>>,
191    /// Tool call fragments accumulated by tool call index.
192    tool_accumulators: BTreeMap<u32, ToolCallAccumulator>,
193    /// Buffered events ready to be yielded.
194    pending_events: VecDeque<StreamEvent>,
195    /// Last seen response ID.
196    last_id: String,
197    /// Last seen model name.
198    last_model: String,
199    /// Last seen usage stats.
200    last_usage: Option<ResponseUsage>,
201    /// Last seen finish reason.
202    last_finish_reason: Option<FinishReason>,
203    /// Whether the stream has completed.
204    finished: bool,
205}
206
207impl ToolAwareStream {
208    /// Create a new `ToolAwareStream` wrapping a raw SSE stream.
209    pub fn new(inner: BoxStream<'static, Result<CompletionsResponse, OpenRouterError>>) -> Self {
210        Self {
211            inner,
212            tool_accumulators: BTreeMap::new(),
213            pending_events: VecDeque::new(),
214            last_id: String::new(),
215            last_model: String::new(),
216            last_usage: None,
217            last_finish_reason: None,
218            finished: false,
219        }
220    }
221
222    /// Process a single `CompletionsResponse` chunk, extracting events
223    /// and accumulating tool call fragments.
224    fn process_chunk(&mut self, response: CompletionsResponse) {
225        // Track metadata from every chunk
226        self.last_id.clone_from(&response.id);
227        self.last_model.clone_from(&response.model);
228        if response.usage.is_some() {
229            self.last_usage = response.usage;
230        }
231
232        for choice in &response.choices {
233            // Track finish reason
234            if let Some(reason) = choice.finish_reason() {
235                self.last_finish_reason = Some(reason.clone());
236            }
237
238            // Extract content delta
239            if let Some(content) = choice.content() {
240                if !content.is_empty() {
241                    self.pending_events
242                        .push_back(StreamEvent::ContentDelta(content.to_string()));
243                }
244            }
245
246            // Extract reasoning delta
247            if let Some(reasoning) = choice.reasoning() {
248                if !reasoning.is_empty() {
249                    self.pending_events
250                        .push_back(StreamEvent::ReasoningDelta(reasoning.to_string()));
251                }
252            }
253
254            // Extract reasoning details
255            if let Some(details) = choice.reasoning_details() {
256                if !details.is_empty() {
257                    self.pending_events
258                        .push_back(StreamEvent::ReasoningDetailsDelta(details.to_vec()));
259                }
260            }
261
262            // Accumulate partial tool calls
263            if let Some(partial_tool_calls) = choice.partial_tool_calls() {
264                for partial in partial_tool_calls {
265                    // Use the index field to identify which tool call this
266                    // fragment belongs to. Default to 0 if not specified.
267                    let idx = partial.index.unwrap_or(0);
268                    let acc = self.tool_accumulators.entry(idx).or_default();
269                    acc.merge(partial);
270                }
271            }
272        }
273    }
274
275    /// Finalize the stream: assemble complete tool calls and emit `Done`.
276    fn finalize(&mut self) {
277        let tool_calls: Vec<ToolCall> = std::mem::take(&mut self.tool_accumulators)
278            .into_values()
279            .filter_map(ToolCallAccumulator::into_tool_call)
280            .collect();
281
282        self.pending_events.push_back(StreamEvent::Done {
283            tool_calls,
284            finish_reason: self.last_finish_reason.take(),
285            usage: self.last_usage.take(),
286            id: std::mem::take(&mut self.last_id),
287            model: std::mem::take(&mut self.last_model),
288        });
289
290        self.finished = true;
291    }
292}
293
294impl Stream for ToolAwareStream {
295    type Item = StreamEvent;
296
297    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
298        // Drain any buffered events first
299        if !self.pending_events.is_empty() {
300            return Poll::Ready(self.pending_events.pop_front());
301        }
302
303        if self.finished {
304            return Poll::Ready(None);
305        }
306
307        // Poll the inner stream for the next chunk
308        match self.inner.poll_next_unpin(cx) {
309            Poll::Ready(Some(Ok(response))) => {
310                self.process_chunk(response);
311
312                // Return the first pending event if any
313                if !self.pending_events.is_empty() {
314                    Poll::Ready(self.pending_events.pop_front())
315                } else {
316                    // No events from this chunk (e.g., empty delta), poll again
317                    cx.waker().wake_by_ref();
318                    Poll::Pending
319                }
320            }
321            Poll::Ready(Some(Err(e))) => Poll::Ready(Some(StreamEvent::Error(e))),
322            Poll::Ready(None) => {
323                // Inner stream ended -- emit Done with accumulated tool calls
324                if !self.finished {
325                    self.finalize();
326                    // Return the Done event
327                    if !self.pending_events.is_empty() {
328                        Poll::Ready(self.pending_events.pop_front())
329                    } else {
330                        Poll::Ready(None)
331                    }
332                } else {
333                    Poll::Ready(None)
334                }
335            }
336            Poll::Pending => Poll::Pending,
337        }
338    }
339}
340
341/// Source stream family for a [`UnifiedStreamEvent`].
342#[derive(Debug, Clone, Copy, PartialEq, Eq)]
343#[non_exhaustive]
344pub enum UnifiedStreamSource {
345    Chat,
346    Responses,
347    Messages,
348}
349
350/// Unified stream event model across chat/responses/messages APIs.
351#[derive(Debug)]
352#[non_exhaustive]
353pub enum UnifiedStreamEvent {
354    /// Text content delta from the model.
355    ContentDelta(String),
356    /// Reasoning/thinking delta.
357    ReasoningDelta(String),
358    /// Structured reasoning detail blocks (chat stream only).
359    ReasoningDetailsDelta(Vec<ReasoningDetail>),
360    /// Tool-related delta payload (format depends on source API).
361    ToolDelta(Value),
362    /// Source-specific event payload when no common projection applies.
363    Raw {
364        source: UnifiedStreamSource,
365        event_type: String,
366        data: Value,
367    },
368    /// Terminal event for a stream source.
369    Done {
370        source: UnifiedStreamSource,
371        id: Option<String>,
372        model: Option<String>,
373        finish_reason: Option<String>,
374        usage: Option<Value>,
375    },
376    /// Transport/parsing/runtime error from the underlying stream.
377    Error(OpenRouterError),
378}
379
380/// A unified stream type across all streaming APIs.
381pub type UnifiedStream = BoxStream<'static, UnifiedStreamEvent>;
382
383#[derive(Debug, Default)]
384struct StreamMeta {
385    id: Option<String>,
386    model: Option<String>,
387    finish_reason: Option<String>,
388    usage: Option<Value>,
389}
390
391fn finish_reason_to_string(reason: &FinishReason) -> String {
392    match reason {
393        FinishReason::ToolCalls => "tool_calls".to_string(),
394        FinishReason::Stop => "stop".to_string(),
395        FinishReason::Length => "length".to_string(),
396        FinishReason::ContentFilter => "content_filter".to_string(),
397        FinishReason::Error => "error".to_string(),
398        FinishReason::Other(value) => value.clone(),
399    }
400}
401
402/// Adapt a chat-completions SSE stream to [`UnifiedStreamEvent`].
403pub fn adapt_chat_stream(
404    inner: BoxStream<'static, Result<CompletionsResponse, OpenRouterError>>,
405) -> UnifiedStream {
406    struct State {
407        inner: BoxStream<'static, Result<CompletionsResponse, OpenRouterError>>,
408        pending: VecDeque<UnifiedStreamEvent>,
409        done_emitted: bool,
410        meta: StreamMeta,
411    }
412
413    let state = State {
414        inner,
415        pending: VecDeque::new(),
416        done_emitted: false,
417        meta: StreamMeta::default(),
418    };
419
420    futures_util::stream::unfold(state, |mut state| async move {
421        loop {
422            if let Some(event) = state.pending.pop_front() {
423                return Some((event, state));
424            }
425
426            if state.done_emitted {
427                return None;
428            }
429
430            match state.inner.next().await {
431                Some(Ok(response)) => {
432                    state.meta.id = Some(response.id.clone());
433                    state.meta.model = Some(response.model.clone());
434                    if let Some(usage) = response.usage {
435                        state.meta.usage = serde_json::to_value(usage).ok();
436                    }
437
438                    for choice in &response.choices {
439                        if let Some(content) = choice.content() {
440                            if !content.is_empty() {
441                                state.pending.push_back(UnifiedStreamEvent::ContentDelta(
442                                    content.to_string(),
443                                ));
444                            }
445                        }
446
447                        if let Some(reasoning) = choice.reasoning() {
448                            if !reasoning.is_empty() {
449                                state.pending.push_back(UnifiedStreamEvent::ReasoningDelta(
450                                    reasoning.to_string(),
451                                ));
452                            }
453                        }
454
455                        if let Some(reasoning_details) = choice.reasoning_details() {
456                            if !reasoning_details.is_empty() {
457                                state
458                                    .pending
459                                    .push_back(UnifiedStreamEvent::ReasoningDetailsDelta(
460                                        reasoning_details.to_vec(),
461                                    ));
462                            }
463                        }
464
465                        if let Some(partials) = choice.partial_tool_calls() {
466                            for partial in partials {
467                                state.pending.push_back(UnifiedStreamEvent::ToolDelta(
468                                    serde_json::to_value(partial).unwrap_or(Value::Null),
469                                ));
470                            }
471                        }
472
473                        if let Some(reason) = choice.finish_reason() {
474                            state.meta.finish_reason = Some(finish_reason_to_string(reason));
475                        }
476                    }
477                }
478                Some(Err(error)) => {
479                    state.pending.push_back(UnifiedStreamEvent::Error(error));
480                }
481                None => {
482                    state.done_emitted = true;
483                    state.pending.push_back(UnifiedStreamEvent::Done {
484                        source: UnifiedStreamSource::Chat,
485                        id: state.meta.id.take(),
486                        model: state.meta.model.take(),
487                        finish_reason: state.meta.finish_reason.take(),
488                        usage: state.meta.usage.take(),
489                    });
490                }
491            }
492        }
493    })
494    .boxed()
495}
496
497/// Adapt a Responses API SSE stream to [`UnifiedStreamEvent`].
498pub fn adapt_responses_stream(
499    inner: BoxStream<'static, Result<ResponsesStreamEvent, OpenRouterError>>,
500) -> UnifiedStream {
501    struct State {
502        inner: BoxStream<'static, Result<ResponsesStreamEvent, OpenRouterError>>,
503        pending: VecDeque<UnifiedStreamEvent>,
504        done_emitted: bool,
505        meta: StreamMeta,
506    }
507
508    let state = State {
509        inner,
510        pending: VecDeque::new(),
511        done_emitted: false,
512        meta: StreamMeta::default(),
513    };
514
515    futures_util::stream::unfold(state, |mut state| async move {
516        loop {
517            if let Some(event) = state.pending.pop_front() {
518                return Some((event, state));
519            }
520
521            if state.done_emitted {
522                return None;
523            }
524
525            match state.inner.next().await {
526                Some(Ok(event)) => {
527                    let event_type = event.event_type.clone();
528                    let data_value = serde_json::to_value(&event.data).unwrap_or(Value::Null);
529                    let mut emitted = false;
530
531                    if let Some(response) = event.data.get("response") {
532                        if let Some(id) = response.get("id").and_then(Value::as_str) {
533                            state.meta.id = Some(id.to_string());
534                        }
535                        if let Some(model) = response.get("model").and_then(Value::as_str) {
536                            state.meta.model = Some(model.to_string());
537                        }
538                        if let Some(status) = response.get("status").and_then(Value::as_str) {
539                            state.meta.finish_reason = Some(status.to_string());
540                        }
541                        if let Some(usage) = response.get("usage") {
542                            state.meta.usage = Some(usage.clone());
543                        }
544                    }
545
546                    if event_type.contains("output_text.delta") {
547                        if let Some(delta) = event.data.get("delta").and_then(Value::as_str) {
548                            state
549                                .pending
550                                .push_back(UnifiedStreamEvent::ContentDelta(delta.to_string()));
551                            emitted = true;
552                        }
553                    }
554
555                    if !emitted && event_type.contains("reasoning") {
556                        let reasoning = event
557                            .data
558                            .get("delta")
559                            .and_then(Value::as_str)
560                            .or_else(|| event.data.get("text").and_then(Value::as_str))
561                            .or_else(|| event.data.get("reasoning").and_then(Value::as_str));
562                        if let Some(reasoning) = reasoning {
563                            state.pending.push_back(UnifiedStreamEvent::ReasoningDelta(
564                                reasoning.to_string(),
565                            ));
566                            emitted = true;
567                        }
568                    }
569
570                    if !emitted && event_type.contains("tool") {
571                        state
572                            .pending
573                            .push_back(UnifiedStreamEvent::ToolDelta(data_value.clone()));
574                        emitted = true;
575                    }
576
577                    if event_type == "response.completed" {
578                        state.done_emitted = true;
579                        state.pending.push_back(UnifiedStreamEvent::Done {
580                            source: UnifiedStreamSource::Responses,
581                            id: state.meta.id.take(),
582                            model: state.meta.model.take(),
583                            finish_reason: state.meta.finish_reason.take(),
584                            usage: state.meta.usage.take(),
585                        });
586                        continue;
587                    }
588
589                    if !emitted {
590                        state.pending.push_back(UnifiedStreamEvent::Raw {
591                            source: UnifiedStreamSource::Responses,
592                            event_type,
593                            data: data_value,
594                        });
595                    }
596                }
597                Some(Err(error)) => {
598                    state.pending.push_back(UnifiedStreamEvent::Error(error));
599                }
600                None => {
601                    state.done_emitted = true;
602                    state.pending.push_back(UnifiedStreamEvent::Done {
603                        source: UnifiedStreamSource::Responses,
604                        id: state.meta.id.take(),
605                        model: state.meta.model.take(),
606                        finish_reason: state.meta.finish_reason.take(),
607                        usage: state.meta.usage.take(),
608                    });
609                }
610            }
611        }
612    })
613    .boxed()
614}
615
616/// Adapt a Messages API SSE stream to [`UnifiedStreamEvent`].
617pub fn adapt_messages_stream(
618    inner: BoxStream<'static, Result<AnthropicMessagesSseEvent, OpenRouterError>>,
619) -> UnifiedStream {
620    struct State {
621        inner: BoxStream<'static, Result<AnthropicMessagesSseEvent, OpenRouterError>>,
622        pending: VecDeque<UnifiedStreamEvent>,
623        done_emitted: bool,
624        meta: StreamMeta,
625    }
626
627    let state = State {
628        inner,
629        pending: VecDeque::new(),
630        done_emitted: false,
631        meta: StreamMeta::default(),
632    };
633
634    futures_util::stream::unfold(state, |mut state| async move {
635        loop {
636            if let Some(event) = state.pending.pop_front() {
637                return Some((event, state));
638            }
639
640            if state.done_emitted {
641                return None;
642            }
643
644            match state.inner.next().await {
645                Some(Ok(event)) => {
646                    let event_name = event.event.clone();
647                    match event.data {
648                        AnthropicMessagesStreamEvent::MessageStart { message } => {
649                            state.meta.id = message.id.clone();
650                            state.meta.model = message.model.clone();
651                            if let Some(usage) = message.usage {
652                                state.meta.usage = serde_json::to_value(usage).ok();
653                            }
654                        }
655                        AnthropicMessagesStreamEvent::MessageDelta { delta, usage } => {
656                            state.meta.usage = Some(usage);
657                            if let Some(reason) = delta.get("stop_reason").and_then(Value::as_str) {
658                                state.meta.finish_reason = Some(reason.to_string());
659                            }
660                            let text = delta
661                                .get("text")
662                                .and_then(Value::as_str)
663                                .or_else(|| delta.get("output_text").and_then(Value::as_str));
664                            if let Some(text) = text {
665                                state
666                                    .pending
667                                    .push_back(UnifiedStreamEvent::ContentDelta(text.to_string()));
668                            }
669                        }
670                        AnthropicMessagesStreamEvent::ContentBlockStart {
671                            index,
672                            content_block,
673                        } => match *content_block {
674                            AnthropicContentPart::Thinking { thinking, .. } => {
675                                state
676                                    .pending
677                                    .push_back(UnifiedStreamEvent::ReasoningDelta(thinking));
678                            }
679                            AnthropicContentPart::ToolUse { .. }
680                            | AnthropicContentPart::ServerToolUse { .. } => {
681                                let content_block_value =
682                                    serde_json::to_value(content_block).unwrap_or(Value::Null);
683                                state.pending.push_back(UnifiedStreamEvent::ToolDelta(
684                                    serde_json::json!({
685                                        "index": index,
686                                        "content_block": content_block_value,
687                                    }),
688                                ));
689                            }
690                            _ => {}
691                        },
692                        AnthropicMessagesStreamEvent::ContentBlockDelta { index, delta } => {
693                            let delta_type = delta
694                                .get("type")
695                                .and_then(Value::as_str)
696                                .unwrap_or_default();
697                            if delta_type.contains("text_delta") {
698                                if let Some(text) = delta.get("text").and_then(Value::as_str) {
699                                    state.pending.push_back(UnifiedStreamEvent::ContentDelta(
700                                        text.to_string(),
701                                    ));
702                                }
703                            } else if delta_type.contains("thinking") {
704                                let reasoning = delta
705                                    .get("thinking")
706                                    .and_then(Value::as_str)
707                                    .or_else(|| delta.get("text").and_then(Value::as_str));
708                                if let Some(reasoning) = reasoning {
709                                    state.pending.push_back(UnifiedStreamEvent::ReasoningDelta(
710                                        reasoning.to_string(),
711                                    ));
712                                }
713                            } else if delta_type.contains("tool")
714                                || delta_type.contains("json")
715                                || delta.get("partial_json").is_some()
716                            {
717                                state.pending.push_back(UnifiedStreamEvent::ToolDelta(
718                                    serde_json::json!({
719                                        "index": index,
720                                        "delta": delta
721                                    }),
722                                ));
723                            } else {
724                                state.pending.push_back(UnifiedStreamEvent::Raw {
725                                    source: UnifiedStreamSource::Messages,
726                                    event_type: event_name,
727                                    data: delta,
728                                });
729                            }
730                        }
731                        AnthropicMessagesStreamEvent::MessageStop { .. } => {
732                            state.done_emitted = true;
733                            state.pending.push_back(UnifiedStreamEvent::Done {
734                                source: UnifiedStreamSource::Messages,
735                                id: state.meta.id.take(),
736                                model: state.meta.model.take(),
737                                finish_reason: state.meta.finish_reason.take(),
738                                usage: state.meta.usage.take(),
739                            });
740                        }
741                        AnthropicMessagesStreamEvent::Error { error } => {
742                            let message = error
743                                .get("message")
744                                .and_then(Value::as_str)
745                                .map(ToOwned::to_owned)
746                                .unwrap_or_else(|| error.to_string());
747                            state.pending.push_back(UnifiedStreamEvent::Error(
748                                OpenRouterError::Unknown(format!(
749                                    "messages stream error event: {message}"
750                                )),
751                            ));
752                        }
753                        AnthropicMessagesStreamEvent::ContentBlockStop { .. }
754                        | AnthropicMessagesStreamEvent::Ping => {}
755                    }
756                }
757                Some(Err(error)) => {
758                    state.pending.push_back(UnifiedStreamEvent::Error(error));
759                }
760                None => {
761                    state.done_emitted = true;
762                    state.pending.push_back(UnifiedStreamEvent::Done {
763                        source: UnifiedStreamSource::Messages,
764                        id: state.meta.id.take(),
765                        model: state.meta.model.take(),
766                        finish_reason: state.meta.finish_reason.take(),
767                        usage: state.meta.usage.take(),
768                    });
769                }
770            }
771        }
772    })
773    .boxed()
774}
openrouter_rs/types/stream.rs

openrouter_rs/types/
stream.rs
