eui_core/host/
decoders.rs

1//! The wire decoders every transport adapter shares (pure, incremental,
2//! fixture-locked in `fixtures/host-transport.json`): A2UI JSONL over a byte
3//! stream, Server-Sent Events, and the A2UI-over-MCP carrier.
4
5use serde::{Deserialize, Serialize};
6use serde_json::{json, Value};
7
8use super::contract::{MCP_ACTION_TOOL, MCP_MIME_TYPES, SSE_EVENTS};
9use super::js_trim;
10use crate::limits::MAX_MESSAGE_BYTES;
11
12#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
13pub struct DecodeIssue {
14    /// 1-based line (JSONL) or event (SSE) number.
15    pub at: u32,
16    pub message: String,
17}
18
19#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
20pub struct Decoded {
21    pub messages: Vec<Value>,
22    pub issues: Vec<DecodeIssue>,
23}
24
25impl Decoded {
26    pub fn extend(&mut self, other: Decoded) {
27        self.messages.extend(other.messages);
28        self.issues.extend(other.issues);
29    }
30
31    pub fn to_json(&self) -> Value {
32        serde_json::to_value(self).unwrap_or(Value::Null)
33    }
34}
35
36fn parse_line(line: &str, at: u32, out: &mut Decoded) {
37    let text = js_trim(line);
38    if text.is_empty() {
39        return;
40    }
41    match serde_json::from_str::<Value>(text) {
42        Ok(v) => out.messages.push(v),
43        Err(_) => out.issues.push(DecodeIssue { at, message: format!("line {at} is not JSON") }),
44    }
45}
46
47/// VAPP-103 rfix: one line of a byte stream, or a line past
48/// `maxMessageBytes` (its bytes are dropped up to the next newline).
49#[derive(Debug, Clone, PartialEq)]
50enum Line {
51    Text(String),
52    Oversized,
53}
54
55/// The line splitter both stream decoders share: the buffer holds only the
56/// unterminated tail (each push scans the new bytes once and drains once),
57/// never more than `maxMessageBytes`.
58#[derive(Debug, Clone, Default)]
59struct Lines {
60    buffer: String,
61    /// Bytes of `buffer` already scanned (no newline in them).
62    scanned: usize,
63    /// Inside an oversized line: drop until its newline.
64    skipping: bool,
65}
66
67impl Lines {
68    fn push(&mut self, chunk: &str) -> Vec<Line> {
69        let mut out = Vec::new();
70        let mut rest = chunk;
71        if self.skipping {
72            match rest.find('\n') {
73                None => return out,
74                Some(i) => {
75                    self.skipping = false;
76                    rest = &rest[i + 1..];
77                }
78            }
79        }
80        self.buffer.push_str(rest);
81        let mut start = 0;
82        let mut from = self.scanned;
83        while let Some(i) = self.buffer[from..].find('\n') {
84            let nl = from + i;
85            let line = &self.buffer[start..nl];
86            out.push(if line.len() > MAX_MESSAGE_BYTES { Line::Oversized } else { Line::Text(line.to_string()) });
87            start = nl + 1;
88            from = start;
89        }
90        if start > 0 {
91            self.buffer.drain(..start);
92        }
93        self.scanned = self.buffer.len();
94        if self.buffer.len() > MAX_MESSAGE_BYTES {
95            out.push(Line::Oversized);
96            self.buffer = String::new();
97            self.scanned = 0;
98            self.skipping = true;
99        }
100        out
101    }
102
103    /// The unterminated tail (`None` inside an oversized line).
104    fn end(&mut self) -> Option<String> {
105        let skipping = std::mem::take(&mut self.skipping);
106        self.scanned = 0;
107        let tail = std::mem::take(&mut self.buffer);
108        (!skipping).then_some(tail)
109    }
110}
111
112/// One message per line. `push` takes any chunking (a line may span chunks),
113/// `end` flushes a last line without a newline. A line past
114/// `maxMessageBytes` is an issue (`line N: message larger than … bytes`),
115/// never buffered.
116#[derive(Debug, Clone, Default)]
117pub struct JsonlDecoder {
118    lines: Lines,
119    line: u32,
120}
121
122impl JsonlDecoder {
123    pub fn new() -> Self {
124        Self::default()
125    }
126
127    pub fn push(&mut self, chunk: &str) -> Decoded {
128        let mut out = Decoded::default();
129        for line in self.lines.push(chunk) {
130            self.line += 1;
131            match line {
132                Line::Text(raw) => parse_line(&raw, self.line, &mut out),
133                Line::Oversized => out.issues.push(oversized(self.line, "line")),
134            }
135        }
136        out
137    }
138
139    pub fn end(&mut self) -> Decoded {
140        let mut out = Decoded::default();
141        if let Some(raw) = self.lines.end().filter(|raw| !js_trim(raw).is_empty()) {
142            self.line += 1;
143            parse_line(&raw, self.line, &mut out);
144        }
145        out
146    }
147}
148
149fn oversized(at: u32, what: &str) -> DecodeIssue {
150    DecodeIssue { at, message: format!("{what} {at}: {}", crate::limits::message_bytes_issue()) }
151}
152
153/// A whole JSONL document (or one JSON value / a JSON array of messages).
154pub fn decode_jsonl(text: &str) -> Decoded {
155    let trimmed = js_trim(text);
156    if trimmed.starts_with('[') {
157        if let Ok(Value::Array(messages)) = serde_json::from_str::<Value>(trimmed) {
158            return Decoded { messages, issues: vec![] };
159        }
160        // fall through: JSONL whose first line is an array is still JSONL
161    }
162    let mut d = JsonlDecoder::new();
163    let mut out = d.push(text);
164    out.extend(d.end());
165    out
166}
167
168/// Server-Sent Events: `data:` lines join with `\n` per event, a blank line
169/// dispatches, `:` comments and other fields are ignored (`retry:` is kept
170/// as [`SseDecoder::retry_ms`]), an `event:` name outside SSE_EVENTS drops
171/// the event. An event's data is one message or JSONL.
172#[derive(Debug, Clone, Default)]
173pub struct SseDecoder {
174    lines: Lines,
175    data: Vec<String>,
176    /// The event's data bytes so far (past `maxMessageBytes` it is dropped
177    /// and dispatches as an issue).
178    data_bytes: usize,
179    oversized: bool,
180    event: String,
181    count: u32,
182    retry_ms: Option<u64>,
183}
184
185impl SseDecoder {
186    pub fn new() -> Self {
187        Self::default()
188    }
189
190    /// The last `retry:` field (ms): the server resumes the stream after it
191    /// ends (a transport reconnects on a clean end only then).
192    pub fn retry_ms(&self) -> Option<u64> {
193        self.retry_ms
194    }
195
196    pub fn push(&mut self, chunk: &str) -> Decoded {
197        let mut out = Decoded::default();
198        for line in self.lines.push(chunk) {
199            match line {
200                Line::Text(mut line) => {
201                    if line.ends_with('\r') {
202                        line.pop();
203                    }
204                    self.line(&line, &mut out);
205                }
206                Line::Oversized => self.drop_data(),
207            }
208        }
209        out
210    }
211
212    pub fn end(&mut self) -> Decoded {
213        let mut out = Decoded::default();
214        match self.lines.end() {
215            Some(line) if !line.is_empty() => self.line(&line, &mut out),
216            Some(_) => {}
217            None => self.drop_data(),
218        }
219        self.dispatch(&mut out);
220        out
221    }
222
223    /// The event is past `maxMessageBytes`: forget its data (it dispatches
224    /// as an issue).
225    fn drop_data(&mut self) {
226        self.data.clear();
227        self.data_bytes = 0;
228        self.oversized = true;
229    }
230
231    fn line(&mut self, line: &str, out: &mut Decoded) {
232        if line.is_empty() {
233            return self.dispatch(out);
234        }
235        if line.starts_with(':') {
236            return;
237        }
238        let (field, value) = match line.find(':') {
239            Some(colon) => (&line[..colon], &line[colon + 1..]),
240            None => (line, ""),
241        };
242        let value = value.strip_prefix(' ').unwrap_or(value);
243        if field == "data" {
244            if self.oversized {
245                return;
246            }
247            self.data_bytes += value.len() + 1;
248            if self.data_bytes > MAX_MESSAGE_BYTES {
249                return self.drop_data();
250            }
251            self.data.push(value.to_string());
252        } else if field == "event" {
253            self.event = value.to_string();
254        } else if field == "retry" && !value.is_empty() && value.bytes().all(|b| b.is_ascii_digit()) {
255            if let Ok(ms) = value.parse() {
256                self.retry_ms = Some(ms);
257            }
258        }
259    }
260
261    fn dispatch(&mut self, out: &mut Decoded) {
262        self.data_bytes = 0;
263        if std::mem::take(&mut self.oversized) {
264            self.count += 1;
265            self.data.clear();
266            self.event.clear();
267            out.issues.push(oversized(self.count, "event"));
268            return;
269        }
270        if self.data.is_empty() {
271            self.event.clear();
272            return;
273        }
274        self.count += 1;
275        let name = if self.event.is_empty() { "message".to_string() } else { std::mem::take(&mut self.event) };
276        let data = std::mem::take(&mut self.data).join("\n");
277        self.event.clear();
278        if !SSE_EVENTS.contains(&name.as_str()) {
279            return;
280        }
281        let d = decode_jsonl(&data);
282        out.messages.extend(d.messages);
283        if !d.issues.is_empty() {
284            out.issues.push(DecodeIssue { at: self.count, message: format!("event {} is not JSON", self.count) });
285        }
286    }
287}
288
289/// The A2UI-over-MCP carrier: the messages inside an MCP tool result
290/// (`content[]` resources with an A2UI mime type, or
291/// `structuredContent.a2ui`).
292pub fn messages_from_mcp_result(result: &Value) -> Decoded {
293    let mut out = Decoded::default();
294    if !matches!(result, Value::Object(_) | Value::Array(_)) {
295        return out;
296    }
297    if let Some(Value::Array(structured)) = result.get("structuredContent").and_then(|s| s.get("a2ui")) {
298        out.messages.extend(structured.iter().cloned());
299    }
300    if let Some(Value::Array(content)) = result.get("content") {
301        for (i, item) in content.iter().enumerate() {
302            let Some(resource) = item.get("resource").filter(|r| super::js_truthy(r)) else { continue };
303            if item.get("type").and_then(Value::as_str) != Some("resource") {
304                continue;
305            }
306            let mime = resource.get("mimeType").and_then(Value::as_str).unwrap_or("");
307            if !MCP_MIME_TYPES.contains(&mime) {
308                continue;
309            }
310            let Some(Value::String(text)) = resource.get("text") else { continue };
311            let d = decode_jsonl(text);
312            out.messages.extend(d.messages);
313            for issue in d.issues {
314                out.issues.push(DecodeIssue { at: i as u32 + 1, message: format!("content[{i}]: {}", issue.message) });
315            }
316        }
317    }
318    out
319}
320
321/// A client message as the MCP tools/call that carries it back (`tool`
322/// defaults to `a2ui_event`).
323pub fn mcp_action_call(message: &Value, tool: Option<&str>) -> Value {
324    json!({"method": "tools/call", "params": {"name": tool.unwrap_or(MCP_ACTION_TOOL), "arguments": {"message": message}}})
325}
326
327#[cfg(test)]
328mod tests {
329    use super::{JsonlDecoder, SseDecoder, MAX_MESSAGE_BYTES};
330
331    /// VAPP-103 rfix: a line past `maxMessageBytes` is ONE issue and never
332    /// buffered; the lines around it decode; a stream fed byte by byte
333    /// scans linearly (the old find+drain per line was O(n²)).
334    #[test]
335    fn jsonl_caps_a_line_at_max_message_bytes() {
336        let mut d = JsonlDecoder::new();
337        let mut out = d.push("{\"a\":1}\n");
338        let big = "x".repeat(1 << 20);
339        for _ in 0..5 {
340            out.extend(d.push(&big));
341        }
342        assert!(d.lines.buffer.is_empty(), "the oversized line is not buffered");
343        out.extend(d.push("still the big line\n{\"b\":2}\n"));
344        out.extend(d.end());
345        assert_eq!(out.messages, vec![serde_json::json!({"a": 1}), serde_json::json!({"b": 2})]);
346        assert_eq!(out.issues.len(), 1);
347        assert_eq!(out.issues[0].at, 2);
348        assert_eq!(out.issues[0].message, format!("line 2: message larger than {MAX_MESSAGE_BYTES} bytes"));
349        // A one-chunk oversized line too.
350        let mut d = JsonlDecoder::new();
351        let mut out = d.push(&format!("{}\n{{\"c\":3}}\n", "y".repeat(MAX_MESSAGE_BYTES + 1)));
352        out.extend(d.end());
353        assert_eq!((out.messages.len(), out.issues.len()), (1, 1));
354    }
355
356    #[test]
357    fn jsonl_fed_byte_by_byte_is_linear() {
358        let line = format!("{{\"text\":\"{}\"}}\n", "z".repeat(200_000));
359        let mut d = JsonlDecoder::new();
360        let t = std::time::Instant::now();
361        let mut messages = 0;
362        for chunk in line.as_bytes().chunks(7) {
363            messages += d.push(std::str::from_utf8(chunk).unwrap()).messages.len();
364        }
365        assert_eq!(messages, 1);
366        assert!(t.elapsed() < std::time::Duration::from_secs(2), "{:?}", t.elapsed());
367    }
368
369    #[test]
370    fn sse_caps_an_event_at_max_message_bytes() {
371        let mut d = SseDecoder::new();
372        let mut out = d.push("data: {\"a\":1}\n\n");
373        let chunk = format!("data: {}\n", "x".repeat(1 << 20));
374        for _ in 0..5 {
375            out.extend(d.push(&chunk));
376        }
377        assert!(d.data.is_empty(), "the oversized event's data is dropped");
378        out.extend(d.push("\ndata: {\"b\":2}\n\n"));
379        out.extend(d.end());
380        assert_eq!(out.messages, vec![serde_json::json!({"a": 1}), serde_json::json!({"b": 2})]);
381        assert_eq!(out.issues.len(), 1);
382        assert_eq!(out.issues[0].message, format!("event 2: message larger than {MAX_MESSAGE_BYTES} bytes"));
383    }
384
385    #[test]
386    fn sse_keeps_the_last_numeric_retry_field_and_its_messages_are_unchanged() {
387        let mut d = SseDecoder::new();
388        assert_eq!(d.retry_ms(), None);
389        let mut out = d.push("retry: 1500\ndata: {\"a\":1}\n\nretry: soon\n");
390        out.extend(d.end());
391        assert_eq!(out.messages, vec![serde_json::json!({"a": 1})]);
392        assert_eq!(d.retry_ms(), Some(1500), "a non-numeric retry is ignored");
393        d.push("retry:250\n\n");
394        assert_eq!(d.retry_ms(), Some(250));
395    }
396}
eui_core/host/decoders.rs

eui_core/host/
decoders.rs
