Struct StreamAccumulator 

pub struct StreamAccumulator { /* private fields */ }

Rebuilds one Response from a stream of WireStreamEvents.

This is the receiving half of the proxy pattern: the client-side equivalent of RequestBuilder::stream_accumulated, operating on events that arrived over a network rather than on a live provider connection.

Unlike stream_accumulated, this does reassemble tool calls: fragments from ToolCallStart / ToolCallDelta / ToolCallEnd are concatenated and attached to the assistant message.

Truncation is an error

finish fails unless the stream was well-formed: it must have opened with MessageStart and ended with a terminal event. A stream that just stops — the client’s socket died mid-generation — yields WireErrorKind::Stream, which is what distinguishes “the network died” from “the provider refused” (the latter arrives as WireStreamEvent::Error and is returned as its own WireError).

Examples

use rai_sdk::ProviderKind;
use rai_sdk::wire::{StreamAccumulator, WireStreamEvent};

let mut accumulator = StreamAccumulator::new();
accumulator.push(WireStreamEvent::message_start("gpt-4o-mini", ProviderKind::OpenAI))?;
accumulator.push(WireStreamEvent::TextDelta { text: "done".into() })?;
accumulator.push(WireStreamEvent::MessageStop { finish_reason: Some("stop".into()) })?;

let response = accumulator.finish()?;
assert_eq!(response.text(), "done");
assert_eq!(response.finish_reason.as_deref(), Some("stop"));

Implementations

impl StreamAccumulator

pub fn new() -> Self

Start with an empty accumulator.

pub fn protocol_version(&self) -> Option<u32>

The protocol version the sender advertised, once MessageStart has been seen.

pub fn text(&self) -> &str

The assistant text accumulated so far.

pub fn usage(&self) -> Option<&Usage>

Token usage, once a Usage event has arrived.

pub fn is_complete(&self) -> bool

Whether a terminal event has been seen.

A stream whose events are exhausted while this is false was truncated.

pub fn push(&mut self, event: WireStreamEvent) -> Result<(), WireError>

Absorb one event.

Errors

Returns the carried WireError when event is WireStreamEvent::Error, so a caller driving this in a loop can stop with ?. The error is remembered, so a later finish reports it too.

pub fn finish(self) -> Result<Response, WireError>

Assemble everything pushed so far into one Response.

Errors

• The WireError from a WireStreamEvent::Error, if one arrived.
• WireErrorKind::Stream if no MessageStart was seen, since the model and provider a Response requires are unknown.
• WireErrorKind::Stream if no terminal event was seen, meaning the stream was cut short rather than finished.

pub async fn accumulate<S>(stream: S) -> Result<Response, WireError>

where S: Stream<Item = WireStreamEvent>,

Drain a stream of wire events into one Response.

The convenience form of new() + push() in a loop + finish(), for the common client-side case where the whole stream is consumed at once.

Errors

Same as push and finish.

Examples

use rai_sdk::ProviderKind;
use rai_sdk::wire::{StreamAccumulator, WireStreamEvent};

let events = futures::stream::iter(vec![
    WireStreamEvent::message_start("claude-sonnet-4-6", ProviderKind::Anthropic),
    WireStreamEvent::TextDelta { text: "hi".into() },
    WireStreamEvent::MessageStop { finish_reason: Some("stop".into()) },
]);

let response = StreamAccumulator::accumulate(events).await?;
assert_eq!(response.text(), "hi");

Trait Implementations

impl Clone for StreamAccumulator

fn clone(&self) -> StreamAccumulator

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for StreamAccumulator

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for StreamAccumulator

fn default() -> StreamAccumulator

Returns the “default value” for a type.