systemprompt_wire/anthropic/
framing.rs

1//! Anthropic SSE byte framing: raw upstream bytes to canonical events.
2//!
3//! [`AnthropicStreamState`] maps one decoded `data:` payload; this module owns
4//! the step before it — buffering bytes until a frame is complete, splitting
5//! it into lines and decoding each payload. The gateway and the in-process AI
6//! service both read Anthropic streams (first-party and Vertex AI send the
7//! same frames), so the framing lives once, here.
8//!
9//! Copyright (c) systemprompt.io — Business Source License 1.1.
10//! See <https://systemprompt.io> for licensing details.
11
12use bytes::Bytes;
13use futures_util::stream::{self, BoxStream, Stream, StreamExt};
14// JSON: protocol boundary — each SSE payload is dynamic JSON keyed on `type`.
15use serde_json::Value;
16
17use super::sse::AnthropicStreamState;
18use crate::canonical::CanonicalEvent;
19use crate::error::WireStreamError;
20
21#[derive(Debug, Default)]
22pub struct SseFrameDecoder {
23    buf: Vec<u8>,
24    codec: AnthropicStreamState,
25}
26
27impl SseFrameDecoder {
28    pub fn push(&mut self, chunk: &[u8]) -> Vec<CanonicalEvent> {
29        self.buf.extend_from_slice(chunk);
30        let mut events = Vec::new();
31        while let Some(end) = crate::sse::frame_end(&self.buf) {
32            let frame: Vec<u8> = self.buf.drain(..end).collect();
33            let frame_str = String::from_utf8_lossy(&frame);
34            for line in frame_str.lines() {
35                let Some(data) = line.strip_prefix("data: ") else {
36                    continue;
37                };
38                if data.trim() == "[DONE]" {
39                    continue;
40                }
41                if let Ok(value) = serde_json::from_str::<Value>(data) {
42                    events.extend(self.codec.events_from_sse(&value));
43                }
44            }
45        }
46        events
47    }
48}
49
50pub fn sse_to_canonical_events<S, E>(
51    stream: S,
52) -> BoxStream<'static, Result<CanonicalEvent, WireStreamError>>
53where
54    S: Stream<Item = Result<Bytes, E>> + Send + 'static,
55    E: Into<Box<dyn std::error::Error + Send + Sync>> + 'static,
56{
57    stream
58        .scan(SseFrameDecoder::default(), |decoder, item| {
59            let out: Vec<Result<CanonicalEvent, WireStreamError>> = match item {
60                Ok(bytes) => decoder.push(&bytes).into_iter().map(Ok).collect(),
61                Err(e) => vec![Err(WireStreamError::transport(e))],
62            };
63            futures_util::future::ready(Some(out))
64        })
65        .flat_map(stream::iter)
66        .boxed()
67}
systemprompt_wire/anthropic/framing.rs

systemprompt_wire/anthropic/
framing.rs
