Streaming Architecture Guide

Overview

Provider streaming lives in src/core/providers/base/sse.rs plus per-provider transformers under src/core/providers/base/sse/ (openai.rs, anthropic.rs, gemini.rs, cohere.rs, databricks.rs). The layer consumes a provider's raw SSE byte stream and yields Result<ChatChunk, ProviderError> items in an OpenAI-compatible shape, so the server routes never see provider-specific formats.

Streaming Flow

┌─────────────────────────────────────────────────────────────────┐
│                  Provider SSE byte stream                       │
│  reqwest::Response::bytes_stream()                              │
│  (OpenAI, Anthropic, Google, ...)                               │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                  UnifiedSSEStream<S, T>                         │
│  - polls upstream bytes, feeds UnifiedSSEParser                 │
│  - chunk_buffer: VecDeque<ChatChunk>, capped at 10_000          │
│  - Item = Result<ChatChunk, ProviderError>                      │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                  UnifiedSSEParser<T>                            │
│  - String line buffer (incomplete tail retained across reads)   │
│  - SSEEvent field parsing, multi-line data joining              │
│  - end-marker / finish_stream dispatch                          │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                  SSETransformer (per provider)                  │
│  - transform_chunk / transform_stream_chunk                     │
│  - normalizes wire format to ChatChunk                          │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                  Server route re-serialization                  │
│  ChatChunk -> SSE frames ("data: {...}\n\n") + final [DONE]     │
└─────────────────────────────────────────────────────────────────┘

The parser owns its transformer: UnifiedSSEParser<T: SSETransformer> calls back into T while parsing, so there is no separate processing stage between parser and transformer.

Core Components

SSEEvent

// src/core/providers/base/sse.rs
#[derive(Debug, Clone)]
pub struct SSEEvent {
    pub event_type: Option<String>,
    pub data: String,
    pub id: Option<String>,
    pub retry: Option<u64>,
}

SSEEvent::from_line(&str) -> Option<SSEEvent> parses one SSE field line:

• Empty lines and : comment lines return None.
• data, event, id, and retry set the matching field; whitespace after

the colon is trimmed.

• retry must parse as u64, otherwise None; unknown fields return None.

The parser accumulates multiple data lines of one event, joining them with \n, and dispatches on the blank line that terminates the event.

SSETransformer Trait

// src/core/providers/base/sse.rs
pub trait SSETransformer: Send + Sync {
    fn provider_name(&self) -> &'static str;

    fn is_end_marker(&self, data: &str) -> bool {
        data.trim() == "[DONE]"
    }

    fn transform_chunk(&self, data: &str) -> Result<Option<ChatChunk>, ProviderError>;

    fn transform_stream_chunk(&self, data: &str) -> Result<Option<ChatChunk>, ProviderError> {
        self.transform_chunk(data)
    }

    fn finish_stream(&self) -> Result<Option<ChatChunk>, ProviderError> {
        Ok(None)
    }

    fn parse_finish_reason(&self, reason: &str) -> Option<FinishReason> { ... }
}

• Errors are ProviderError

(crate::core::providers::unified_provider::ProviderError). There is no dedicated StreamError enum.

• The default parsefinishreason maps case-insensitively:

stop|endturn -> Stop, length|maxtokens -> Length, toolcalls|functioncall|tooluse -> ToolCalls, contentfilter|safety|recitation -> ContentFilter, stopsequence -> StopSequence, refusal -> Refusal, pauseturn -> PauseTurn; unknown strings yield None.

• Built-in implementations: OpenAICompatibleTransformer,

AnthropicTransformer, GeminiTransformer, CohereTransformer, DatabricksTransformer (see [reference/provider-transformers.md](reference/provider-transformers.md)).

UnifiedSSEParser\<T\>

// src/core/providers/base/sse.rs
pub struct UnifiedSSEParser<T: SSETransformer> {
    transformer: T,
    buffer: String,
    current_event: Option<SSEEvent>,
}

impl<T: SSETransformer> UnifiedSSEParser<T> {
    pub fn new(transformer: T) -> Self;
    pub fn process_bytes(&mut self, bytes: &[u8]) -> Result<Vec<ChatChunk>, ProviderError>;
}

• The buffer is a String, not a byte deque. Each incoming read is decoded

independently with String::fromutf8lossy and appended; only text up to the last \n is processed and the incomplete tail stays buffered for the next call. Line/event splits are retained, but a read boundary inside a multibyte UTF-8 code point is lossy because the undecoded bytes are not retained.

• process_bytes runs non-stream mode: an end marker produces nothing and

events go through transform_chunk.

• UnifiedSSEStream drives the private processstreambytes path (stream

mode): an end marker triggers transformer.finishstream() instead, and data goes through transformstream_chunk.

• No size cap applies to this buffer.
• The private finish_stream flushes any leftover partial line and pending

event, then appends transformer.finish_stream() output.

UnifiedSSEStream\<S, T\>

// src/core/providers/base/sse.rs
const MAX_CHUNK_BUFFER_SIZE: usize = 10_000;

pub struct UnifiedSSEStream<S, T>
where
    S: Stream<Item = Result<Bytes, reqwest::Error>> + Send + Unpin,
    T: SSETransformer + Clone,
{
    inner: S,
    parser: UnifiedSSEParser<T>,
    chunk_buffer: VecDeque<ChatChunk>,
    pending_error: Option<ProviderError>,
    finished: bool,
}

pollnext order: pop chunkbuffer, then take pendingerror, then return None once finished, otherwise poll inner and feed bytes through processstream_bytes.

• A read that yields zero complete chunks stores nothing; the stream returns

Pending after cx.waker().wakebyref().

• If buffered plus new chunks would exceed MAXCHUNKBUFFERSIZE (10000),

it yields Err(ProviderError::network(...)) instead of growing unboundedly.

• Transport errors are wrapped as

ProviderError::network(provider, format!("Stream error: {error}")); chunks drained from parser.finish_stream() are emitted before the error item.

• Upstream end-of-stream sets finished and drains parser.finish_stream()

before returning None.

Helper createproviderssestream(response, providername) boxes response.bytes_stream() behind an OpenAICompatibleTransformer.

References

• [reference/provider-transformers.md](reference/provider-transformers.md) — behavior of the OpenAI-compatible, Anthropic, Gemini, Cohere, and Databricks transformers
• [reference/stream-pipeline.md](reference/stream-pipeline.md) — UnifiedSSEStream pipeline internals, provider wiring pattern, and actix HTTP response streaming
• [reference/buffer-management.md](reference/buffer-management.md) — parser line buffer retention and chunk-buffer overflow guard
• [reference/configuration.md](reference/configuration.md) — server.streamidletimeout and fixed buffering constants
• [reference/best-practices.md](reference/best-practices.md) — incremental parsing, per-stream state, error mapping, usage handling