tea_model/
stream.rs

1use std::collections::{BTreeMap, BTreeSet};
2use std::fmt;
3use std::str::FromStr;
4
5use serde_json::Value;
6use tea_protocol::{
7    ExactCost, ExternalSource, HostedToolOutcome, MAX_HOSTED_TOOL_SOURCES, ModelId,
8    ProtocolMetadata, ProviderContinuation, SourceCitation, StopReason, Usage,
9};
10use thiserror::Error;
11
12use crate::ModelFailure;
13
14/// Maximum UTF-8 bytes in one normalized text, thinking, or argument delta.
15pub const MAX_MODEL_DELTA_BYTES: usize = 64 * 1024;
16/// Maximum UTF-8 bytes in one opaque provider response or tool-call ID.
17pub const MAX_PROVIDER_OPAQUE_ID_BYTES: usize = 256;
18/// Largest concurrent tool-call index supported in one response.
19pub const MAX_MODEL_STREAM_INDEX: u16 = 1023;
20const MAX_COMPLETED_TOOL_ARGUMENT_BYTES: usize = 256 * 1024;
21const MAX_COMPLETED_TOOL_ARGUMENT_DEPTH: usize = 32;
22
23macro_rules! opaque_id {
24    ($name:ident, $doc:literal) => {
25        #[doc = $doc]
26        #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
27        pub struct $name(String);
28
29        impl $name {
30            /// Returns the bounded opaque provider value.
31            #[must_use]
32            pub fn as_str(&self) -> &str {
33                &self.0
34            }
35        }
36
37        impl FromStr for $name {
38            type Err = ModelStreamValueError;
39
40            fn from_str(value: &str) -> Result<Self, Self::Err> {
41                if value.is_empty()
42                    || value.len() > MAX_PROVIDER_OPAQUE_ID_BYTES
43                    || value.chars().any(char::is_control)
44                {
45                    return Err(ModelStreamValueError::InvalidProviderOpaqueId);
46                }
47                Ok(Self(value.to_owned()))
48            }
49        }
50
51        impl fmt::Display for $name {
52            fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
53                formatter.write_str(&self.0)
54            }
55        }
56    };
57}
58
59opaque_id!(
60    ProviderResponseId,
61    "Bounded opaque response identifier returned by a provider."
62);
63opaque_id!(
64    ProviderToolCallId,
65    "Bounded provider-scoped identifier joining streamed tool-call fragments."
66);
67
68/// Bounded zero-based content/tool index in one provider response.
69#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
70pub struct ModelStreamIndex(u16);
71
72impl ModelStreamIndex {
73    /// Creates a bounded stream index.
74    ///
75    /// # Errors
76    ///
77    /// Returns an error above [`MAX_MODEL_STREAM_INDEX`].
78    pub const fn new(value: u16) -> Result<Self, ModelStreamValueError> {
79        if value > MAX_MODEL_STREAM_INDEX {
80            Err(ModelStreamValueError::InvalidStreamIndex)
81        } else {
82            Ok(Self(value))
83        }
84    }
85
86    /// Returns the numeric index.
87    #[must_use]
88    pub const fn get(self) -> u16 {
89        self.0
90    }
91}
92
93/// Bounded non-empty UTF-8 text or thinking delta.
94#[derive(Debug, Clone, PartialEq, Eq)]
95pub struct Utf8Delta(String);
96
97impl Utf8Delta {
98    /// Creates a bounded delta.
99    ///
100    /// # Errors
101    ///
102    /// Returns an error when empty, oversized, or containing a null character.
103    pub fn new(value: impl Into<String>) -> Result<Self, ModelStreamValueError> {
104        let value = value.into();
105        validate_delta(&value)?;
106        Ok(Self(value))
107    }
108
109    /// Returns delta text.
110    #[must_use]
111    pub fn as_str(&self) -> &str {
112        &self.0
113    }
114}
115
116/// Metadata reported when a normalized provider response starts.
117#[derive(Debug, Clone, PartialEq, Default)]
118pub struct ModelResponseInfo {
119    response_id: Option<ProviderResponseId>,
120    response_model: Option<ModelId>,
121    metadata: ProtocolMetadata,
122}
123
124impl ModelResponseInfo {
125    /// Creates empty response metadata.
126    #[must_use]
127    pub fn new() -> Self {
128        Self::default()
129    }
130
131    /// Adds an opaque provider response ID.
132    #[must_use]
133    pub fn with_response_id(mut self, response_id: ProviderResponseId) -> Self {
134        self.response_id = Some(response_id);
135        self
136    }
137
138    /// Adds the concrete response model when it differs or is informative.
139    #[must_use]
140    pub fn with_response_model(mut self, response_model: ModelId) -> Self {
141        self.response_model = Some(response_model);
142        self
143    }
144
145    /// Adds bounded namespaced response metadata.
146    #[must_use]
147    pub fn with_metadata(mut self, metadata: ProtocolMetadata) -> Self {
148        self.metadata = metadata;
149        self
150    }
151
152    /// Returns the provider response ID.
153    #[must_use]
154    pub const fn response_id(&self) -> Option<&ProviderResponseId> {
155        self.response_id.as_ref()
156    }
157
158    /// Returns the concrete response model.
159    #[must_use]
160    pub const fn response_model(&self) -> Option<&ModelId> {
161        self.response_model.as_ref()
162    }
163
164    /// Returns bounded response metadata.
165    #[must_use]
166    pub const fn metadata(&self) -> &ProtocolMetadata {
167        &self.metadata
168    }
169}
170
171/// Start of one provider-streamed tool call.
172#[derive(Debug, Clone, PartialEq, Eq)]
173pub struct ToolCallStarted {
174    index: ModelStreamIndex,
175    provider_call_id: ProviderToolCallId,
176    tool_name: String,
177}
178
179impl ToolCallStarted {
180    /// Creates a validated tool-call start.
181    ///
182    /// # Errors
183    ///
184    /// Returns an error when the tool name is not canonical.
185    pub fn new(
186        index: ModelStreamIndex,
187        provider_call_id: ProviderToolCallId,
188        tool_name: impl Into<String>,
189    ) -> Result<Self, ModelStreamValueError> {
190        let tool_name = tool_name.into();
191        validate_tool_name(&tool_name)?;
192        Ok(Self {
193            index,
194            provider_call_id,
195            tool_name,
196        })
197    }
198
199    /// Returns the response-local stream index.
200    #[must_use]
201    pub const fn index(&self) -> ModelStreamIndex {
202        self.index
203    }
204
205    /// Returns the provider call ID.
206    #[must_use]
207    pub const fn provider_call_id(&self) -> &ProviderToolCallId {
208        &self.provider_call_id
209    }
210
211    /// Returns the canonical tool name.
212    #[must_use]
213    pub fn tool_name(&self) -> &str {
214        &self.tool_name
215    }
216}
217
218/// Incomplete provider tool-arguments fragment.
219#[derive(Debug, Clone, PartialEq, Eq)]
220pub struct ToolArgumentsDelta {
221    index: ModelStreamIndex,
222    provider_call_id: ProviderToolCallId,
223    delta: String,
224}
225
226impl ToolArgumentsDelta {
227    /// Creates a bounded incomplete argument fragment.
228    ///
229    /// # Errors
230    ///
231    /// Returns an error when the fragment is empty, oversized, or contains a
232    /// null character.
233    pub fn new(
234        index: ModelStreamIndex,
235        provider_call_id: ProviderToolCallId,
236        delta: impl Into<String>,
237    ) -> Result<Self, ModelStreamValueError> {
238        let delta = delta.into();
239        validate_delta(&delta)?;
240        Ok(Self {
241            index,
242            provider_call_id,
243            delta,
244        })
245    }
246
247    /// Returns the response-local stream index.
248    #[must_use]
249    pub const fn index(&self) -> ModelStreamIndex {
250        self.index
251    }
252
253    /// Returns the provider call ID.
254    #[must_use]
255    pub const fn provider_call_id(&self) -> &ProviderToolCallId {
256        &self.provider_call_id
257    }
258
259    /// Returns the incomplete UTF-8 argument fragment.
260    #[must_use]
261    pub fn delta(&self) -> &str {
262        &self.delta
263    }
264}
265
266/// Completed, parsed provider tool call safe for later canonical projection.
267#[derive(Debug, Clone, PartialEq)]
268pub struct ToolCallCompleted {
269    index: ModelStreamIndex,
270    provider_call_id: ProviderToolCallId,
271    tool_name: String,
272    arguments: Value,
273}
274
275impl ToolCallCompleted {
276    /// Creates a validated completed call with object arguments.
277    ///
278    /// # Errors
279    ///
280    /// Returns an error for an invalid name, non-object arguments, or bounded
281    /// JSON violations.
282    pub fn new(
283        index: ModelStreamIndex,
284        provider_call_id: ProviderToolCallId,
285        tool_name: impl Into<String>,
286        arguments: Value,
287    ) -> Result<Self, ModelStreamValueError> {
288        let tool_name = tool_name.into();
289        validate_tool_name(&tool_name)?;
290        validate_completed_arguments(&arguments)?;
291        Ok(Self {
292            index,
293            provider_call_id,
294            tool_name,
295            arguments,
296        })
297    }
298
299    /// Returns the response-local stream index.
300    #[must_use]
301    pub const fn index(&self) -> ModelStreamIndex {
302        self.index
303    }
304
305    /// Returns the provider call ID.
306    #[must_use]
307    pub const fn provider_call_id(&self) -> &ProviderToolCallId {
308        &self.provider_call_id
309    }
310
311    /// Returns the canonical tool name.
312    #[must_use]
313    pub fn tool_name(&self) -> &str {
314        &self.tool_name
315    }
316
317    /// Returns complete parsed object arguments.
318    #[must_use]
319    pub const fn arguments(&self) -> &Value {
320        &self.arguments
321    }
322}
323
324/// Start of one provider-hosted tool activity.
325#[derive(Debug, Clone, PartialEq, Eq)]
326pub struct HostedToolStarted {
327    index: ModelStreamIndex,
328    provider_call_id: ProviderToolCallId,
329    tool_name: String,
330}
331
332impl HostedToolStarted {
333    /// Creates a validated hosted tool start.
334    ///
335    /// # Errors
336    ///
337    /// Returns an error when the tool name is not canonical.
338    pub fn new(
339        index: ModelStreamIndex,
340        provider_call_id: ProviderToolCallId,
341        tool_name: impl Into<String>,
342    ) -> Result<Self, ModelStreamValueError> {
343        let tool_name = tool_name.into();
344        validate_tool_name(&tool_name)?;
345        Ok(Self {
346            index,
347            provider_call_id,
348            tool_name,
349        })
350    }
351
352    /// Returns the response-local stream index.
353    #[must_use]
354    pub const fn index(&self) -> ModelStreamIndex {
355        self.index
356    }
357
358    /// Returns the provider activity identifier.
359    #[must_use]
360    pub const fn provider_call_id(&self) -> &ProviderToolCallId {
361        &self.provider_call_id
362    }
363
364    /// Returns the canonical hosted tool name.
365    #[must_use]
366    pub fn tool_name(&self) -> &str {
367        &self.tool_name
368    }
369}
370
371/// Complete provider-hosted tool activity safe for canonical projection.
372#[derive(Debug, Clone, PartialEq, Eq)]
373pub struct HostedToolCompleted {
374    index: ModelStreamIndex,
375    provider_call_id: ProviderToolCallId,
376    tool_name: String,
377    arguments: Value,
378    outcome: HostedToolOutcome,
379    sources: Vec<ExternalSource>,
380    continuation: Option<ProviderContinuation>,
381}
382
383impl HostedToolCompleted {
384    /// Creates one validated complete hosted activity.
385    ///
386    /// # Errors
387    ///
388    /// Returns an error for invalid identity, arguments, or source bounds.
389    #[allow(clippy::too_many_arguments)]
390    pub fn new(
391        index: ModelStreamIndex,
392        provider_call_id: ProviderToolCallId,
393        tool_name: impl Into<String>,
394        arguments: Value,
395        outcome: HostedToolOutcome,
396        sources: Vec<ExternalSource>,
397        continuation: Option<ProviderContinuation>,
398    ) -> Result<Self, ModelStreamValueError> {
399        let tool_name = tool_name.into();
400        validate_tool_name(&tool_name)?;
401        validate_completed_arguments(&arguments)?;
402        if sources.len() > MAX_HOSTED_TOOL_SOURCES {
403            return Err(ModelStreamValueError::TooManyHostedToolSources);
404        }
405        Ok(Self {
406            index,
407            provider_call_id,
408            tool_name,
409            arguments,
410            outcome,
411            sources,
412            continuation,
413        })
414    }
415
416    /// Returns the response-local stream index.
417    #[must_use]
418    pub const fn index(&self) -> ModelStreamIndex {
419        self.index
420    }
421
422    /// Returns the provider activity identifier.
423    #[must_use]
424    pub const fn provider_call_id(&self) -> &ProviderToolCallId {
425        &self.provider_call_id
426    }
427
428    /// Returns the canonical hosted tool name.
429    #[must_use]
430    pub fn tool_name(&self) -> &str {
431        &self.tool_name
432    }
433
434    /// Returns normalized object arguments.
435    #[must_use]
436    pub const fn arguments(&self) -> &Value {
437        &self.arguments
438    }
439
440    /// Returns the provider-reported terminal outcome.
441    #[must_use]
442    pub const fn outcome(&self) -> &HostedToolOutcome {
443        &self.outcome
444    }
445
446    /// Returns normalized sources in provider order.
447    #[must_use]
448    pub fn sources(&self) -> &[ExternalSource] {
449        &self.sources
450    }
451
452    /// Returns opaque same-provider continuation data.
453    #[must_use]
454    pub const fn continuation(&self) -> Option<&ProviderContinuation> {
455        self.continuation.as_ref()
456    }
457}
458
459/// Provider-streamed citation awaiting canonical hosted call identity mapping.
460#[derive(Debug, Clone, PartialEq, Eq)]
461pub struct ModelSourceCitation {
462    provider_call_id: Option<ProviderToolCallId>,
463    citation: SourceCitation,
464}
465
466impl ModelSourceCitation {
467    /// Creates a normalized citation without a canonical tool-call identifier.
468    ///
469    /// # Errors
470    ///
471    /// Returns an error when an adapter attempts to assign kernel-owned identity.
472    pub fn new(
473        provider_call_id: Option<ProviderToolCallId>,
474        citation: SourceCitation,
475    ) -> Result<Self, ModelStreamValueError> {
476        if citation.tool_call_id().is_some() {
477            return Err(ModelStreamValueError::CitationAlreadyCanonical);
478        }
479        Ok(Self {
480            provider_call_id,
481            citation,
482        })
483    }
484
485    /// Returns the related provider activity identifier, when available.
486    #[must_use]
487    pub const fn provider_call_id(&self) -> Option<&ProviderToolCallId> {
488        self.provider_call_id.as_ref()
489    }
490
491    /// Returns the normalized provider-neutral citation.
492    #[must_use]
493    pub const fn citation(&self) -> &SourceCitation {
494        &self.citation
495    }
496}
497
498/// Successful terminal model completion data.
499#[derive(Debug, Clone, PartialEq)]
500pub struct ModelCompletion {
501    stop_reason: StopReason,
502    usage: Option<Usage>,
503    cost: Option<ExactCost>,
504    metadata: ProtocolMetadata,
505}
506
507impl ModelCompletion {
508    /// Creates a normal successful completion.
509    #[must_use]
510    pub fn completed() -> Self {
511        Self {
512            stop_reason: StopReason::Completed,
513            usage: None,
514            cost: None,
515            metadata: ProtocolMetadata::default(),
516        }
517    }
518
519    /// Creates successful terminal completion data.
520    ///
521    /// # Errors
522    ///
523    /// Returns an error for cancelled, error, or unknown stop reasons. Those
524    /// outcomes must use [`ModelEvent::Failed`].
525    pub fn new(stop_reason: StopReason) -> Result<Self, ModelStreamValueError> {
526        if !matches!(
527            stop_reason,
528            StopReason::Completed
529                | StopReason::Length
530                | StopReason::ToolUse
531                | StopReason::PauseTurn
532        ) {
533            return Err(ModelStreamValueError::InvalidCompletionReason);
534        }
535        Ok(Self {
536            stop_reason,
537            usage: None,
538            cost: None,
539            metadata: ProtocolMetadata::default(),
540        })
541    }
542
543    /// Adds normalized token usage.
544    #[must_use]
545    pub fn with_usage(mut self, usage: Usage) -> Self {
546        self.usage = Some(usage);
547        self
548    }
549
550    /// Adds exact provider-reported cost.
551    #[must_use]
552    pub fn with_cost(mut self, cost: ExactCost) -> Self {
553        self.cost = Some(cost);
554        self
555    }
556
557    /// Adds bounded terminal metadata.
558    #[must_use]
559    pub fn with_metadata(mut self, metadata: ProtocolMetadata) -> Self {
560        self.metadata = metadata;
561        self
562    }
563
564    /// Returns the normalized stop reason.
565    #[must_use]
566    pub const fn stop_reason(&self) -> &StopReason {
567        &self.stop_reason
568    }
569
570    /// Returns normalized token usage when reported.
571    #[must_use]
572    pub const fn usage(&self) -> Option<&Usage> {
573        self.usage.as_ref()
574    }
575
576    /// Returns exact provider-reported cost when available.
577    #[must_use]
578    pub const fn cost(&self) -> Option<&ExactCost> {
579        self.cost.as_ref()
580    }
581
582    /// Returns bounded terminal metadata.
583    #[must_use]
584    pub const fn metadata(&self) -> &ProtocolMetadata {
585        &self.metadata
586    }
587}
588
589/// One normalized provider stream event.
590#[derive(Debug, Clone, PartialEq)]
591pub enum ModelEvent {
592    /// Provider accepted the request and began one response.
593    Started(ModelResponseInfo),
594    /// Visible assistant text fragment.
595    TextDelta(Utf8Delta),
596    /// Assistant reasoning fragment.
597    ThinkingDelta(Utf8Delta),
598    /// Start of a provider tool call.
599    ToolCallStarted(ToolCallStarted),
600    /// Incomplete tool argument fragment, never executable by itself.
601    ToolArgumentsDelta(ToolArgumentsDelta),
602    /// Completed tool call with parsed object arguments.
603    ToolCallCompleted(ToolCallCompleted),
604    /// Start of a provider-hosted tool activity.
605    HostedToolStarted(HostedToolStarted),
606    /// Completed provider-hosted tool activity with normalized sources.
607    HostedToolCompleted(HostedToolCompleted),
608    /// Citation emitted for assistant text and an external source.
609    SourceCitation(ModelSourceCitation),
610    /// Successful terminal event.
611    Completed(ModelCompletion),
612    /// Failed or cancelled terminal event.
613    Failed(ModelFailure),
614}
615
616impl ModelEvent {
617    /// Returns visible text for a text-delta event.
618    #[must_use]
619    pub fn as_text_delta(&self) -> Option<&str> {
620        match self {
621            Self::TextDelta(delta) => Some(delta.as_str()),
622            _ => None,
623        }
624    }
625
626    /// Returns reasoning text for a thinking-delta event.
627    #[must_use]
628    pub fn as_thinking_delta(&self) -> Option<&str> {
629        match self {
630            Self::ThinkingDelta(delta) => Some(delta.as_str()),
631            _ => None,
632        }
633    }
634
635    /// Returns the tool-call start payload.
636    #[must_use]
637    pub const fn as_tool_call_started(&self) -> Option<&ToolCallStarted> {
638        match self {
639            Self::ToolCallStarted(call) => Some(call),
640            _ => None,
641        }
642    }
643
644    /// Returns the incomplete tool-arguments payload.
645    #[must_use]
646    pub const fn as_tool_arguments_delta(&self) -> Option<&ToolArgumentsDelta> {
647        match self {
648            Self::ToolArgumentsDelta(delta) => Some(delta),
649            _ => None,
650        }
651    }
652
653    /// Returns the completed tool-call payload.
654    #[must_use]
655    pub const fn as_tool_call_completed(&self) -> Option<&ToolCallCompleted> {
656        match self {
657            Self::ToolCallCompleted(call) => Some(call),
658            _ => None,
659        }
660    }
661
662    /// Returns the hosted tool start payload.
663    #[must_use]
664    pub const fn as_hosted_tool_started(&self) -> Option<&HostedToolStarted> {
665        match self {
666            Self::HostedToolStarted(call) => Some(call),
667            _ => None,
668        }
669    }
670
671    /// Returns the hosted tool completion payload.
672    #[must_use]
673    pub const fn as_hosted_tool_completed(&self) -> Option<&HostedToolCompleted> {
674        match self {
675            Self::HostedToolCompleted(call) => Some(call),
676            _ => None,
677        }
678    }
679
680    /// Returns the normalized source citation payload.
681    #[must_use]
682    pub const fn as_source_citation(&self) -> Option<&ModelSourceCitation> {
683        match self {
684            Self::SourceCitation(citation) => Some(citation),
685            _ => None,
686        }
687    }
688}
689
690/// Deterministic validator for normalized provider stream grammar.
691#[derive(Debug, Default)]
692pub struct ModelStreamValidator {
693    started: bool,
694    terminal: Option<bool>,
695    event_count: usize,
696    completed_tool_calls: usize,
697    completed_hosted_tools: usize,
698    active_tools: BTreeMap<ModelStreamIndex, (ProviderToolCallId, String)>,
699    active_hosted_tools: BTreeMap<ModelStreamIndex, (ProviderToolCallId, String)>,
700    completed_hosted_ids: BTreeSet<ProviderToolCallId>,
701    seen_tool_indexes: BTreeSet<ModelStreamIndex>,
702}
703
704impl ModelStreamValidator {
705    /// Creates an empty stream grammar validator.
706    #[must_use]
707    pub fn new() -> Self {
708        Self::default()
709    }
710
711    /// Observes one event in source order.
712    ///
713    /// # Errors
714    ///
715    /// Returns a typed violation when the event does not follow normalized
716    /// stream grammar. Rejected events do not advance validator state.
717    pub fn observe(&mut self, event: &ModelEvent) -> Result<(), ModelStreamViolation> {
718        if self.terminal.is_some() {
719            return Err(ModelStreamViolation::EventAfterTerminal);
720        }
721        if !self.started {
722            if matches!(event, ModelEvent::Started(_)) {
723                self.started = true;
724                self.event_count += 1;
725                return Ok(());
726            }
727            return Err(ModelStreamViolation::EventBeforeStart);
728        }
729
730        match event {
731            ModelEvent::Started(_) => return Err(ModelStreamViolation::DuplicateStart),
732            ModelEvent::ToolCallStarted(call) => {
733                if self.seen_tool_indexes.contains(&call.index()) {
734                    return Err(ModelStreamViolation::DuplicateToolIndex);
735                }
736                self.seen_tool_indexes.insert(call.index());
737                self.active_tools.insert(
738                    call.index(),
739                    (call.provider_call_id().clone(), call.tool_name().to_owned()),
740                );
741            }
742            ModelEvent::HostedToolStarted(call) => {
743                if self.seen_tool_indexes.contains(&call.index()) {
744                    return Err(ModelStreamViolation::DuplicateToolIndex);
745                }
746                self.seen_tool_indexes.insert(call.index());
747                self.active_hosted_tools.insert(
748                    call.index(),
749                    (call.provider_call_id().clone(), call.tool_name().to_owned()),
750                );
751            }
752            ModelEvent::ToolArgumentsDelta(delta) => {
753                let Some((call_id, _)) = self.active_tools.get(&delta.index()) else {
754                    return Err(ModelStreamViolation::UnknownToolIndex);
755                };
756                if call_id != delta.provider_call_id() {
757                    return Err(ModelStreamViolation::ToolIdentityMismatch);
758                }
759            }
760            ModelEvent::ToolCallCompleted(call) => {
761                let Some((call_id, tool_name)) = self.active_tools.get(&call.index()) else {
762                    return Err(ModelStreamViolation::UnknownToolIndex);
763                };
764                if call_id != call.provider_call_id() || tool_name != call.tool_name() {
765                    return Err(ModelStreamViolation::ToolIdentityMismatch);
766                }
767                self.active_tools.remove(&call.index());
768                self.completed_tool_calls += 1;
769            }
770            ModelEvent::HostedToolCompleted(call) => {
771                let Some((call_id, tool_name)) = self.active_hosted_tools.get(&call.index()) else {
772                    return Err(ModelStreamViolation::UnknownToolIndex);
773                };
774                if call_id != call.provider_call_id() || tool_name != call.tool_name() {
775                    return Err(ModelStreamViolation::ToolIdentityMismatch);
776                }
777                self.active_hosted_tools.remove(&call.index());
778                self.completed_hosted_ids
779                    .insert(call.provider_call_id().clone());
780                self.completed_hosted_tools += 1;
781            }
782            ModelEvent::SourceCitation(citation) => {
783                if citation
784                    .provider_call_id()
785                    .is_some_and(|call_id| !self.completed_hosted_ids.contains(call_id))
786                {
787                    return Err(ModelStreamViolation::UnknownHostedCitation);
788                }
789            }
790            ModelEvent::Completed(_) => {
791                if !self.active_tools.is_empty() || !self.active_hosted_tools.is_empty() {
792                    return Err(ModelStreamViolation::IncompleteToolCalls);
793                }
794                self.terminal = Some(true);
795            }
796            ModelEvent::Failed(_) => {
797                self.terminal = Some(false);
798            }
799            ModelEvent::TextDelta(_) | ModelEvent::ThinkingDelta(_) => {}
800        }
801        self.event_count += 1;
802        Ok(())
803    }
804
805    /// Finishes validation after the stream returns `None`.
806    ///
807    /// # Errors
808    ///
809    /// Returns an error when start or terminal events are missing.
810    pub fn finish(self) -> Result<ModelStreamSummary, ModelStreamViolation> {
811        if !self.started {
812            return Err(ModelStreamViolation::MissingStart);
813        }
814        let succeeded = self.terminal.ok_or(ModelStreamViolation::MissingTerminal)?;
815        Ok(ModelStreamSummary {
816            event_count: self.event_count,
817            completed_tool_calls: self.completed_tool_calls,
818            completed_hosted_tools: self.completed_hosted_tools,
819            succeeded,
820        })
821    }
822}
823
824/// Summary of one fully validated normalized stream.
825#[derive(Debug, Clone, Copy, PartialEq, Eq)]
826pub struct ModelStreamSummary {
827    event_count: usize,
828    completed_tool_calls: usize,
829    completed_hosted_tools: usize,
830    succeeded: bool,
831}
832
833impl ModelStreamSummary {
834    /// Returns accepted event count including start and terminal events.
835    #[must_use]
836    pub const fn event_count(self) -> usize {
837        self.event_count
838    }
839
840    /// Returns the number of completed tool calls.
841    #[must_use]
842    pub const fn completed_tool_calls(self) -> usize {
843        self.completed_tool_calls
844    }
845
846    /// Returns the number of completed provider-hosted activities.
847    #[must_use]
848    pub const fn completed_hosted_tools(self) -> usize {
849        self.completed_hosted_tools
850    }
851
852    /// Returns whether termination was successful rather than failed.
853    #[must_use]
854    pub const fn succeeded(self) -> bool {
855        self.succeeded
856    }
857}
858
859/// Normalized stream grammar violation.
860#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
861pub enum ModelStreamViolation {
862    /// A non-start event appeared before `Started`.
863    #[error("model stream event appeared before start")]
864    EventBeforeStart,
865    /// No start event was observed.
866    #[error("model stream is missing start")]
867    MissingStart,
868    /// More than one start event appeared.
869    #[error("model stream contains duplicate start")]
870    DuplicateStart,
871    /// Stream ended without success or failure.
872    #[error("model stream is missing terminal event")]
873    MissingTerminal,
874    /// An event appeared after success or failure.
875    #[error("model stream event appeared after terminal event")]
876    EventAfterTerminal,
877    /// A response-local tool index was started more than once concurrently.
878    #[error("model stream contains duplicate active tool index")]
879    DuplicateToolIndex,
880    /// A tool fragment or completion references no active call.
881    #[error("model stream references an unknown tool index")]
882    UnknownToolIndex,
883    /// Tool index, provider ID, or name changed within one call.
884    #[error("model stream tool identity changed")]
885    ToolIdentityMismatch,
886    /// Successful termination left one or more calls incomplete.
887    #[error("model stream completed with incomplete tool calls")]
888    IncompleteToolCalls,
889    /// A citation references no completed provider-hosted activity.
890    #[error("model stream citation references an unknown hosted tool")]
891    UnknownHostedCitation,
892}
893
894/// Error returned by normalized stream value constructors.
895#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
896pub enum ModelStreamValueError {
897    /// Provider opaque identifier is empty, oversized, or contains controls.
898    #[error("provider opaque identifier is invalid")]
899    InvalidProviderOpaqueId,
900    /// Stream index exceeds the supported response-local range.
901    #[error("model stream index is invalid")]
902    InvalidStreamIndex,
903    /// Delta is empty, oversized, or contains a null character.
904    #[error("model stream delta is invalid")]
905    InvalidDelta,
906    /// Tool name is not canonical.
907    #[error("model tool name is invalid")]
908    InvalidToolName,
909    /// Completed tool arguments must be an object.
910    #[error("completed tool arguments must be a JSON object")]
911    ToolArgumentsMustBeObject,
912    /// Completed tool arguments exceed JSON bounds.
913    #[error("completed tool arguments exceed supported bounds")]
914    ToolArgumentsOutOfBounds,
915    /// Hosted tool returned too many normalized sources.
916    #[error("hosted tool returned too many sources")]
917    TooManyHostedToolSources,
918    /// Provider adapters cannot assign kernel-owned citation identity.
919    #[error("model citation already contains a canonical tool-call identifier")]
920    CitationAlreadyCanonical,
921    /// Successful completion used a failure-only or unknown stop reason.
922    #[error("successful completion stop reason is invalid")]
923    InvalidCompletionReason,
924    /// Failure message is empty, oversized, or contains a null character.
925    #[error("model failure message is invalid")]
926    InvalidFailureMessage,
927}
928
929fn validate_delta(value: &str) -> Result<(), ModelStreamValueError> {
930    if value.is_empty() || value.len() > MAX_MODEL_DELTA_BYTES || value.contains('\0') {
931        Err(ModelStreamValueError::InvalidDelta)
932    } else {
933        Ok(())
934    }
935}
936
937fn validate_tool_name(value: &str) -> Result<(), ModelStreamValueError> {
938    let mut bytes = value.bytes();
939    if value.len() > 128
940        || !bytes.next().is_some_and(|byte| byte.is_ascii_lowercase())
941        || !bytes.all(|byte| {
942            byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'_' | b'-' | b'.')
943        })
944    {
945        Err(ModelStreamValueError::InvalidToolName)
946    } else {
947        Ok(())
948    }
949}
950
951fn validate_completed_arguments(arguments: &Value) -> Result<(), ModelStreamValueError> {
952    if !arguments.is_object() {
953        return Err(ModelStreamValueError::ToolArgumentsMustBeObject);
954    }
955    if serde_json::to_vec(arguments)
956        .map_err(|_| ModelStreamValueError::ToolArgumentsOutOfBounds)?
957        .len()
958        > MAX_COMPLETED_TOOL_ARGUMENT_BYTES
959        || json_depth(arguments) > MAX_COMPLETED_TOOL_ARGUMENT_DEPTH
960    {
961        return Err(ModelStreamValueError::ToolArgumentsOutOfBounds);
962    }
963    Ok(())
964}
965
966fn json_depth(value: &Value) -> usize {
967    match value {
968        Value::Array(values) => 1 + values.iter().map(json_depth).max().unwrap_or(0),
969        Value::Object(values) => 1 + values.values().map(json_depth).max().unwrap_or(0),
970        _ => 1,
971    }
972}
tea_model/stream.rs

tea_model/
stream.rs
