Struct NormalizingStream 

pub struct NormalizingStream {
    inner: Pin<Box<dyn Stream<Item = Result<LlmStreamEvent>> + Send>>,
    parser: Box<dyn ToolCallParser>,
    queued: VecDeque<LlmStreamEvent>,
    next_index: usize,
    terminated: bool,
    done_forwarded: bool,
}

Stream adapter that runs every event through a ToolCallParser before re-emitting the normalized result. See module docs.

Fields

inner: Pin<Box<dyn Stream<Item = Result<LlmStreamEvent>> + Send>>

parser: Box<dyn ToolCallParser>

queued: VecDeque<LlmStreamEvent>

Events ready to emit on the next poll. A single upstream event can expand to many downstream events (e.g. Done flushes parser state before propagating).

next_index: usize

Lowest tool-call index that is safe to use for a synthesised delta. Bumped past every upstream index we observe so downstream collectors can use indices as keys without collision.

terminated: bool

true once the upstream inner stream is fully exhausted (ended or errored). Subsequent polls return None. Note this is not set merely because a Done event was seen — see done_forwarded.

done_forwarded: bool

true once we’ve forwarded the upstream Done event.

Done is not treated as an absolute end-of-stream signal: per the OpenAI stream_options.include_usage convention (and llama.cpp’s wire behaviour), a trailing Usage event legitimately arrives after the finish_reason/Done chunk, before the underlying byte stream actually closes. Once this flag is set, only trailer-safe events (Usage, PromptProgress, NormalizationError, UpstreamError) are still queued; any further content-bearing event (TextDelta, ReasoningDelta, ToolCallDelta, a second Done) is dropped defensively — a well-formed stream never sends these after Done, and the parser has already been finalised.

Implementations

impl NormalizingStream

pub fn new( inner: Pin<Box<dyn Stream<Item = Result<LlmStreamEvent>> + Send>>, parser: Box<dyn ToolCallParser>, ) -> Self

Wrap inner so every event is normalized through parser.

fn enqueue_parser_output(&mut self, out: ParserOutput)

Translate one parser output batch into the queued event sequence.

fn handle_upstream(&mut self, event: LlmStreamEvent)

Process one upstream event and queue the resulting downstream events.

Trait Implementations

impl Stream for NormalizingStream

type Item = Result<LlmStreamEvent, Error>

Values yielded by the stream.

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.