Streaming Architecture

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 parse_finish_reason maps case-insensitively: stop|end_turn -> Stop, length|max_tokens -> Length, tool_calls|function_call|tool_use -> ToolCalls, content_filter|safety|recitation -> ContentFilter, stop_sequence -> StopSequence, refusal -> Refusal, pause_turn -> PauseTurn; unknown strings yield None.
• Built-in implementations: OpenAICompatibleTransformer, AnthropicTransformer, GeminiTransformer, CohereTransformer, DatabricksTransformer (see 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::from_utf8_lossy 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 process_stream_bytes path (stream mode): an end marker triggers transformer.finish_stream() instead, and data goes through transform_stream_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,
}

poll_next order: pop chunk_buffer, then take pending_error, then return None once finished, otherwise poll inner and feed bytes through process_stream_bytes.

• A read that yields zero complete chunks stores nothing; the stream returns Pending after cx.waker().wake_by_ref().
• If buffered plus new chunks would exceed MAX_CHUNK_BUFFER_SIZE (10_000), 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 create_provider_sse_stream(response, provider_name) boxes response.bytes_stream() behind an OpenAICompatibleTransformer.

References

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