gglib_core/sse/
parser.rs

1//! Parser for OpenAI-compatible SSE `data:` frames.
2//!
3//! Isolated so the frame-parsing logic and its tests are self-contained and
4//! do not require an HTTP client or async runtime.
5//!
6//! # Frame ordering for reasoning models
7//!
8//! When a single SSE frame carries both `reasoning_content` (chain-of-thought)
9//! **and** `content` (answer text), the [`ReasoningDelta`] event is emitted
10//! first.  This matches the temporal semantics of reasoning models such as
11//! `DeepSeek` R1 and `QwQ`, where the chain-of-thought is always produced before
12//! the answer — even if llama-server coalesces both into the same frame.
13//!
14//! [`ReasoningDelta`]: crate::LlmStreamEvent::ReasoningDelta
15
16use anyhow::{Result, anyhow};
17
18use crate::domain::agent::LlmStreamEvent;
19
20// =============================================================================
21// Public types
22// =============================================================================
23
24/// Result of parsing a single SSE `data:` payload.
25#[derive(Debug)]
26pub enum SseParseResult {
27    /// The value `[DONE]` — stream terminator, no events.
28    Done,
29    /// One or more events decoded from the JSON frame.
30    Events(Vec<LlmStreamEvent>),
31}
32
33// =============================================================================
34// Parser
35// =============================================================================
36
37/// Parse a top-level `usage` object into a [`LlmStreamEvent::Usage`] event.
38///
39/// Returns `None` when the frame carries no `usage` field. Deliberately
40/// returns just the event (not a full [`SseParseResult`]) — unlike
41/// [`parse_inline_error_frame`], a `usage` field does **not** imply the rest
42/// of the frame should be skipped. See the call site in [`parse_sse_frame`]
43/// for why.
44fn parse_usage_event(parsed: &serde_json::Value) -> Option<LlmStreamEvent> {
45    let usage = parsed.get("usage")?;
46    let prompt_tokens =
47        u32::try_from(usage["prompt_tokens"].as_u64().unwrap_or(0)).unwrap_or(u32::MAX);
48    let completion_tokens =
49        u32::try_from(usage["completion_tokens"].as_u64().unwrap_or(0)).unwrap_or(u32::MAX);
50    let total_tokens =
51        u32::try_from(usage["total_tokens"].as_u64().unwrap_or(0)).unwrap_or(u32::MAX);
52    // Unlike the counts above, a missing `cached_tokens` stays `None` rather
53    // than defaulting to 0 — see the field docs on `LlmStreamEvent::Usage`.
54    // llama.cpp nests it under `prompt_tokens_details`, matching OpenAI.
55    let cached_tokens = usage
56        .get("prompt_tokens_details")
57        .and_then(|d| d.get("cached_tokens"))
58        .and_then(serde_json::Value::as_u64)
59        .map(|v| u32::try_from(v).unwrap_or(u32::MAX));
60    Some(LlmStreamEvent::Usage {
61        prompt_tokens,
62        completion_tokens,
63        total_tokens,
64        cached_tokens,
65    })
66}
67
68/// Parse a bare top-level `error` object (with no `choices` key) into an
69/// [`LlmStreamEvent::UpstreamError`] event.
70///
71/// Returns `None` when the frame carries no `error` field, or when it also
72/// carries a `choices` key (even an empty one) — that shape doesn't match
73/// what downstream clients detect as an inline error, so it falls through
74/// to the remaining frame-shape checks instead.
75fn parse_inline_error_frame(parsed: &serde_json::Value) -> Option<SseParseResult> {
76    let err = parsed.get("error")?;
77    if parsed.get("choices").is_some() {
78        return None;
79    }
80    let (message, error_type, code) = match err {
81        serde_json::Value::String(s) => (
82            s.clone(),
83            "server_error".to_owned(),
84            "upstream_error".to_owned(),
85        ),
86        _ => (
87            err.get("message")
88                .and_then(serde_json::Value::as_str)
89                .unwrap_or("upstream returned an error")
90                .to_owned(),
91            err.get("type")
92                .and_then(serde_json::Value::as_str)
93                .unwrap_or("server_error")
94                .to_owned(),
95            err.get("code")
96                .and_then(serde_json::Value::as_str)
97                .unwrap_or("upstream_error")
98                .to_owned(),
99        ),
100    };
101    Some(SseParseResult::Events(vec![
102        LlmStreamEvent::UpstreamError {
103            message,
104            error_type,
105            code,
106        },
107    ]))
108}
109
110/// Parse a single SSE `data:` payload into zero or more [`LlmStreamEvent`]s.
111///
112/// Returns:
113/// - `Ok(SseParseResult::Done)` when `data == "[DONE]"`
114/// - `Ok(SseParseResult::Events(…))` for a valid JSON frame (may be empty
115///   when the frame carries no content or tool-call deltas)
116/// - `Err(…)` when the frame is not valid JSON
117///
118/// # Errors
119///
120/// Returns an error if the `data` payload is not valid JSON.
121pub fn parse_sse_frame(data: &str) -> Result<SseParseResult> {
122    if data == "[DONE]" {
123        return Ok(SseParseResult::Done);
124    }
125
126    let parsed: serde_json::Value = serde_json::from_str(data)
127        .map_err(|e| anyhow!("SSE frame JSON parse error: {e} — data: {data}"))?;
128
129    // ── Prompt-progress frames (llama-server `return_progress: true`) ────
130    // These arrive during the pre-fill phase and have no `choices` array.
131    // We check for them *before* the choices guard so they aren't silently
132    // dropped as "no choices" frames.
133    if let Some(pp) = parsed.get("prompt_progress") {
134        let processed = u32::try_from(pp["processed"].as_u64().unwrap_or(0)).unwrap_or(u32::MAX);
135        let total = u32::try_from(pp["total"].as_u64().unwrap_or(0)).unwrap_or(u32::MAX);
136        let cached = u32::try_from(pp["cache"].as_u64().unwrap_or(0)).unwrap_or(u32::MAX);
137        let time_ms = pp["time_ms"].as_u64().unwrap_or(0);
138        return Ok(SseParseResult::Events(vec![
139            LlmStreamEvent::PromptProgress {
140                processed,
141                total,
142                cached,
143                time_ms,
144            },
145        ]));
146    }
147
148    // ── Usage totals (`stream_options.include_usage: true`) ──────────────
149    // Extracted here but *not* an early return: strict OpenAI servers send
150    // this on a trailing chunk with empty `choices`, but llama-server
151    // attaches `usage` directly onto the same chunk that also carries a
152    // real `finish_reason` and non-empty `choices`
153    // (ggml-org/llama.cpp#12102, #15443). Treating `usage` presence as a
154    // reason to skip the rest of the frame would silently discard that
155    // finish_reason/delta. Decided below once we know whether `choices`
156    // is actually empty.
157    let usage_event = parse_usage_event(&parsed);
158
159    // ── Inline upstream error frame ───────────────────────────────────────
160    // Some OpenAI-compatible servers (including llama.cpp) can emit a bare
161    // `{"error": {...}}` frame mid-stream instead of a hard HTTP-level
162    // failure (e.g. a context-length overflow discovered only once
163    // generation is underway). Checked *before* the choices guard below for
164    // the same reason as `prompt_progress`/`usage`: this frame has no
165    // `choices` key at all, and clients such as the GitHub Copilot LLM
166    // Gateway extension specifically detect this shape via
167    // `'error' in obj && !('choices' in obj)` to surface a real error
168    // instead of hanging or seeing a silently truncated response.
169    if let Some(result) = parse_inline_error_frame(&parsed) {
170        return Ok(result);
171    }
172
173    // Guard against keepalive / error frames that carry no `choices` array.
174    // Without this check every field access falls through to `Value::Null`,
175    // events are silently dropped, and a `finish_reason: "stop"` in such a
176    // frame would mean the stream never emits `Done`.
177    let choices = &parsed["choices"];
178    if choices.as_array().is_none_or(Vec::is_empty) {
179        // Strict-OpenAI shape: no real choice, usage (if any) stands alone.
180        if let Some(usage_event) = usage_event {
181            return Ok(SseParseResult::Events(vec![usage_event]));
182        }
183        tracing::debug!(data = %data, "SSE frame has no 'choices' entries — skipping");
184        return Ok(SseParseResult::Events(vec![]));
185    }
186    let choice = &choices[0];
187    let delta = &choice["delta"];
188
189    let mut events: Vec<LlmStreamEvent> = Vec::new();
190
191    // ── Reasoning/CoT content delta (DeepSeek R1 / QwQ) ────────────────────
192    // Emitted FIRST: chain-of-thought semantically precedes answer text, so
193    // even when both fields appear in the same frame we preserve this order.
194    // llama-server emits `delta["reasoning_content"]` when started with
195    // `--reasoning-format deepseek`.
196    if let Some(reasoning) = delta["reasoning_content"].as_str()
197        && !reasoning.is_empty()
198    {
199        events.push(LlmStreamEvent::ReasoningDelta {
200            content: reasoning.to_owned(),
201        });
202    }
203
204    // ── Text content delta ──────────────────────────────────────────────────
205    if let Some(content) = delta["content"].as_str()
206        && !content.is_empty()
207    {
208        events.push(LlmStreamEvent::TextDelta {
209            content: content.to_owned(),
210        });
211    }
212
213    // ── Tool-call deltas ────────────────────────────────────────────────────
214    if let Some(tool_calls) = delta["tool_calls"].as_array() {
215        for (sequential, tc) in tool_calls.iter().enumerate() {
216            // Prefer the explicit `index` field; fall back to the element's
217            // position in the array when `index` is absent.  A server that
218            // omits `index` on every element is non-compliant with the OpenAI
219            // spec, but we handle it gracefully rather than silently collapsing
220            // all calls onto slot 0.
221            let index = tc["index"]
222                .as_u64()
223                .and_then(|i| usize::try_from(i).ok())
224                .unwrap_or(sequential);
225            let id = tc["id"].as_str().map(str::to_owned);
226            let name = tc["function"]["name"].as_str().map(str::to_owned);
227            let arguments = tc["function"]["arguments"].as_str().map(str::to_owned);
228            events.push(LlmStreamEvent::ToolCallDelta {
229                index,
230                id,
231                name,
232                arguments,
233            });
234        }
235    }
236
237    // ── Usage totals bundled with this finish chunk (llama.cpp shape) ────
238    // Pushed *before* the finish_reason/Done event below, never after: the
239    // encoder appends the `[DONE]` sentinel immediately after `Done`, and
240    // nothing may follow `[DONE]` on the wire.
241    if let Some(usage_event) = usage_event {
242        events.push(usage_event);
243    }
244
245    // ── Finish reason → Done ────────────────────────────────────────────────
246    if let Some(finish_reason) = choice["finish_reason"].as_str()
247        && !finish_reason.is_empty()
248    {
249        events.push(LlmStreamEvent::Done {
250            finish_reason: finish_reason.to_owned(),
251        });
252    }
253
254    Ok(SseParseResult::Events(events))
255}
256
257// =============================================================================
258// Tests
259// =============================================================================
260
261#[cfg(test)]
262mod tests {
263    use super::*;
264
265    // ── Frame builders ─────────────────────────────────────────────────────
266
267    fn text_frame(content: &str) -> String {
268        serde_json::json!({
269            "choices": [{ "delta": { "content": content }, "finish_reason": null }]
270        })
271        .to_string()
272    }
273
274    fn finish_frame(reason: &str) -> String {
275        serde_json::json!({
276            "choices": [{ "delta": {}, "finish_reason": reason }]
277        })
278        .to_string()
279    }
280
281    fn tool_frame(index: usize, id: &str, name: &str, args: &str) -> String {
282        serde_json::json!({
283            "choices": [{
284                "delta": {
285                    "tool_calls": [{
286                        "index": index,
287                        "id": id,
288                        "function": { "name": name, "arguments": args }
289                    }]
290                },
291                "finish_reason": null
292            }]
293        })
294        .to_string()
295    }
296
297    /// Build a frame whose `tool_calls` array elements intentionally omit the
298    /// `index` field, simulating a non-compliant but real-world server.
299    fn tool_frame_no_index(id: &str, name: &str, args: &str) -> String {
300        serde_json::json!({
301            "choices": [{
302                "delta": {
303                    "tool_calls": [
304                        { "id": id, "function": { "name": name, "arguments": args } }
305                    ]
306                },
307                "finish_reason": null
308            }]
309        })
310        .to_string()
311    }
312
313    /// Build a frame with two tool-call elements that both omit `index`.
314    fn two_tool_frames_no_index() -> String {
315        serde_json::json!({
316            "choices": [{
317                "delta": {
318                    "tool_calls": [
319                        { "id": "c1", "function": { "name": "search",  "arguments": "{}" } },
320                        { "id": "c2", "function": { "name": "read_file", "arguments": "{}" } }
321                    ]
322                },
323                "finish_reason": null
324            }]
325        })
326        .to_string()
327    }
328
329    // ── Tests ──────────────────────────────────────────────────────────────
330
331    #[test]
332    fn done_sentinel_returns_done_variant() {
333        assert!(matches!(
334            parse_sse_frame("[DONE]"),
335            Ok(SseParseResult::Done)
336        ));
337    }
338
339    #[test]
340    fn text_delta_frame_produces_text_event() {
341        let events = match parse_sse_frame(&text_frame("hello")) {
342            Ok(SseParseResult::Events(e)) => e,
343            other => panic!("unexpected: {other:?}"),
344        };
345        assert_eq!(events.len(), 1);
346        assert!(matches!(
347            &events[0],
348            LlmStreamEvent::TextDelta { content } if content == "hello"
349        ));
350    }
351
352    #[test]
353    fn empty_content_produces_no_text_event() {
354        let frame = serde_json::json!({
355            "choices": [{ "delta": { "content": "" }, "finish_reason": null }]
356        })
357        .to_string();
358        let events = match parse_sse_frame(&frame) {
359            Ok(SseParseResult::Events(e)) => e,
360            other => panic!("unexpected: {other:?}"),
361        };
362        assert!(
363            events.is_empty(),
364            "empty content should not produce TextDelta"
365        );
366    }
367
368    #[test]
369    fn finish_reason_produces_done_event() {
370        let events = match parse_sse_frame(&finish_frame("stop")) {
371            Ok(SseParseResult::Events(e)) => e,
372            other => panic!("unexpected: {other:?}"),
373        };
374        assert_eq!(events.len(), 1);
375        assert!(matches!(
376            &events[0],
377            LlmStreamEvent::Done { finish_reason } if finish_reason == "stop"
378        ));
379    }
380
381    #[test]
382    fn tool_call_delta_frame_is_parsed() {
383        let events = match parse_sse_frame(&tool_frame(0, "tc1", "search", r#"{"q":"rust"}"#)) {
384            Ok(SseParseResult::Events(e)) => e,
385            other => panic!("unexpected: {other:?}"),
386        };
387        assert_eq!(events.len(), 1);
388        assert!(matches!(
389            &events[0],
390            LlmStreamEvent::ToolCallDelta {
391                index: 0,
392                id: Some(id),
393                name: Some(n),
394                arguments: Some(a),
395            } if id == "tc1" && n == "search" && a == r#"{"q":"rust"}"#
396        ));
397    }
398
399    #[test]
400    fn tool_call_delta_with_no_index_defaults_to_sequential_position() {
401        let events = match parse_sse_frame(&tool_frame_no_index("tc1", "search", r#"{"q":"rust"}"#))
402        {
403            Ok(SseParseResult::Events(e)) => e,
404            other => panic!("unexpected: {other:?}"),
405        };
406        assert_eq!(events.len(), 1);
407        assert!(matches!(
408            &events[0],
409            LlmStreamEvent::ToolCallDelta { index: 0, id: Some(id), .. } if id == "tc1"
410        ));
411    }
412
413    #[test]
414    fn two_tool_calls_with_no_index_get_distinct_sequential_slots() {
415        let events = match parse_sse_frame(&two_tool_frames_no_index()) {
416            Ok(SseParseResult::Events(e)) => e,
417            other => panic!("unexpected: {other:?}"),
418        };
419        assert_eq!(events.len(), 2, "both tool-call deltas must be emitted");
420        assert!(matches!(
421            &events[0],
422            LlmStreamEvent::ToolCallDelta { index: 0, id: Some(id), .. } if id == "c1"
423        ));
424        assert!(matches!(
425            &events[1],
426            LlmStreamEvent::ToolCallDelta { index: 1, id: Some(id), .. } if id == "c2"
427        ));
428    }
429
430    #[test]
431    fn malformed_json_returns_error() {
432        assert!(
433            parse_sse_frame("{ broken json }").is_err(),
434            "malformed JSON should return Err"
435        );
436    }
437
438    #[test]
439    fn frame_with_text_and_finish_reason_produces_both_events() {
440        let frame = serde_json::json!({
441            "choices": [{ "delta": { "content": "hi" }, "finish_reason": "stop" }]
442        })
443        .to_string();
444        let events = match parse_sse_frame(&frame) {
445            Ok(SseParseResult::Events(e)) => e,
446            other => panic!("unexpected: {other:?}"),
447        };
448        assert_eq!(events.len(), 2);
449        assert!(matches!(&events[0], LlmStreamEvent::TextDelta { .. }));
450        assert!(matches!(&events[1], LlmStreamEvent::Done { .. }));
451    }
452
453    #[test]
454    fn reasoning_content_produces_reasoning_delta_event() {
455        let frame = serde_json::json!({
456            "choices": [{ "delta": { "reasoning_content": "I should check..." }, "finish_reason": null }]
457        })
458        .to_string();
459        let events = match parse_sse_frame(&frame) {
460            Ok(SseParseResult::Events(e)) => e,
461            other => panic!("unexpected: {other:?}"),
462        };
463        assert_eq!(events.len(), 1);
464        assert!(matches!(
465            &events[0],
466            LlmStreamEvent::ReasoningDelta { content } if content == "I should check..."
467        ));
468    }
469
470    #[test]
471    fn empty_reasoning_content_produces_no_event() {
472        let frame = serde_json::json!({
473            "choices": [{ "delta": { "reasoning_content": "" }, "finish_reason": null }]
474        })
475        .to_string();
476        let events = match parse_sse_frame(&frame) {
477            Ok(SseParseResult::Events(e)) => e,
478            other => panic!("unexpected: {other:?}"),
479        };
480        assert!(
481            events.is_empty(),
482            "empty reasoning_content should not produce ReasoningDelta"
483        );
484    }
485
486    #[test]
487    fn frame_with_reasoning_and_text_reasoning_emitted_first() {
488        let frame = serde_json::json!({
489            "choices": [{ "delta": { "content": "ok", "reasoning_content": "think" }, "finish_reason": null }]
490        })
491        .to_string();
492        let events = match parse_sse_frame(&frame) {
493            Ok(SseParseResult::Events(e)) => e,
494            other => panic!("unexpected: {other:?}"),
495        };
496        assert_eq!(events.len(), 2);
497        assert!(
498            matches!(&events[0], LlmStreamEvent::ReasoningDelta { content } if content == "think"),
499            "ReasoningDelta must come first"
500        );
501        assert!(
502            matches!(&events[1], LlmStreamEvent::TextDelta { content } if content == "ok"),
503            "TextDelta must come second"
504        );
505    }
506
507    #[test]
508    fn prompt_progress_frame_produces_progress_event() {
509        let frame = serde_json::json!({
510            "prompt_progress": {
511                "processed": 2048,
512                "total": 8192,
513                "cache": 512,
514                "time_ms": 1234
515            }
516        })
517        .to_string();
518        let events = match parse_sse_frame(&frame) {
519            Ok(SseParseResult::Events(e)) => e,
520            other => panic!("unexpected: {other:?}"),
521        };
522        assert_eq!(events.len(), 1);
523        assert!(matches!(
524            &events[0],
525            LlmStreamEvent::PromptProgress {
526                processed: 2048,
527                total: 8192,
528                cached: 512,
529                time_ms: 1234
530            }
531        ));
532    }
533
534    #[test]
535    fn prompt_progress_frame_not_confused_with_choices() {
536        let frame = serde_json::json!({
537            "prompt_progress": {
538                "processed": 100,
539                "total": 100,
540                "cache": 0,
541                "time_ms": 50
542            }
543        })
544        .to_string();
545        let events = match parse_sse_frame(&frame) {
546            Ok(SseParseResult::Events(e)) => e,
547            other => panic!("unexpected: {other:?}"),
548        };
549        assert!(
550            !events.is_empty(),
551            "prompt_progress frame must not be skipped"
552        );
553    }
554
555    #[test]
556    fn usage_frame_emits_usage_event() {
557        let frame = serde_json::json!({
558            "id": "chatcmpl-1",
559            "object": "chat.completion.chunk",
560            "created": 0,
561            "model": "test-model",
562            "choices": [],
563            "usage": {
564                "prompt_tokens": 123,
565                "completion_tokens": 45,
566                "total_tokens": 168
567            }
568        })
569        .to_string();
570        let events = match parse_sse_frame(&frame) {
571            Ok(SseParseResult::Events(e)) => e,
572            other => panic!("unexpected: {other:?}"),
573        };
574        assert_eq!(events.len(), 1);
575        assert!(matches!(
576            &events[0],
577            LlmStreamEvent::Usage {
578                prompt_tokens: 123,
579                completion_tokens: 45,
580                total_tokens: 168,
581                cached_tokens: None
582            }
583        ));
584    }
585
586    /// llama.cpp reports reused prompt tokens under the OpenAI-standard
587    /// `prompt_tokens_details` nesting (its `n_prompt_tokens_cache`).
588    #[test]
589    fn usage_frame_parses_nested_cached_token_count() {
590        let frame = serde_json::json!({
591            "id": "chatcmpl-1",
592            "object": "chat.completion.chunk",
593            "created": 0,
594            "model": "test-model",
595            "choices": [],
596            "usage": {
597                "prompt_tokens": 30342,
598                "completion_tokens": 893,
599                "total_tokens": 31235,
600                "prompt_tokens_details": { "cached_tokens": 892 }
601            }
602        })
603        .to_string();
604        let events = match parse_sse_frame(&frame) {
605            Ok(SseParseResult::Events(e)) => e,
606            other => panic!("unexpected: {other:?}"),
607        };
608        assert!(matches!(
609            &events[0],
610            LlmStreamEvent::Usage {
611                cached_tokens: Some(892),
612                ..
613            }
614        ));
615    }
616
617    /// Zero reused tokens is a genuine measurement — a full re-prefill — and
618    /// must not collapse into the same `None` used for "server didn't say".
619    #[test]
620    fn usage_frame_distinguishes_zero_cached_tokens_from_a_missing_field() {
621        let with_zero = serde_json::json!({
622            "choices": [],
623            "usage": {
624                "prompt_tokens": 10, "completion_tokens": 1, "total_tokens": 11,
625                "prompt_tokens_details": { "cached_tokens": 0 }
626            }
627        })
628        .to_string();
629        let events = match parse_sse_frame(&with_zero) {
630            Ok(SseParseResult::Events(e)) => e,
631            other => panic!("unexpected: {other:?}"),
632        };
633        assert!(matches!(
634            &events[0],
635            LlmStreamEvent::Usage {
636                cached_tokens: Some(0),
637                ..
638            }
639        ));
640    }
641
642    #[test]
643    fn usage_frame_not_confused_with_no_choices_guard() {
644        // Empty `choices` array — would be silently dropped by the "no
645        // choices" guard if the usage check didn't run first.
646        let frame = serde_json::json!({
647            "choices": [],
648            "usage": { "prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2 }
649        })
650        .to_string();
651        let events = match parse_sse_frame(&frame) {
652            Ok(SseParseResult::Events(e)) => e,
653            other => panic!("unexpected: {other:?}"),
654        };
655        assert!(!events.is_empty(), "usage frame must not be skipped");
656    }
657
658    #[test]
659    fn llama_cpp_combined_usage_and_finish_chunk_emits_both_events() {
660        // Real llama-server shape (ggml-org/llama.cpp#12102, #15443): usage
661        // is attached to the *same* chunk as a real finish_reason and a
662        // non-empty `choices` array, not a separate trailing chunk with
663        // empty choices. Must not silently drop the finish_reason.
664        let frame = serde_json::json!({
665            "choices": [{ "finish_reason": "tool_calls", "index": 0, "delta": {} }],
666            "created": 0,
667            "id": "chatcmpl-1",
668            "model": "test-model",
669            "object": "chat.completion.chunk",
670            "usage": { "prompt_tokens": 4181, "completion_tokens": 12, "total_tokens": 4193 }
671        })
672        .to_string();
673        let events = match parse_sse_frame(&frame) {
674            Ok(SseParseResult::Events(e)) => e,
675            other => panic!("unexpected: {other:?}"),
676        };
677        assert_eq!(events.len(), 2, "expected both a Usage and a Done event");
678        // Usage must come *before* Done -- the encoder appends `[DONE]`
679        // immediately after Done, so nothing may be emitted after it.
680        assert!(matches!(
681            &events[0],
682            LlmStreamEvent::Usage {
683                prompt_tokens: 4181,
684                completion_tokens: 12,
685                total_tokens: 4193,
686                cached_tokens: None
687            }
688        ));
689        assert!(matches!(
690            &events[1],
691            LlmStreamEvent::Done { finish_reason } if finish_reason == "tool_calls"
692        ));
693    }
694
695    #[test]
696    fn llama_cpp_combined_usage_and_stop_chunk_preserves_finish_reason() {
697        let frame = serde_json::json!({
698            "choices": [{ "finish_reason": "stop", "index": 0, "delta": {} }],
699            "usage": { "prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15 }
700        })
701        .to_string();
702        let events = match parse_sse_frame(&frame) {
703            Ok(SseParseResult::Events(e)) => e,
704            other => panic!("unexpected: {other:?}"),
705        };
706        assert_eq!(events.len(), 2);
707        assert!(matches!(&events[0], LlmStreamEvent::Usage { .. }));
708        assert!(matches!(
709            &events[1],
710            LlmStreamEvent::Done { finish_reason } if finish_reason == "stop"
711        ));
712    }
713
714    #[test]
715    fn inline_error_frame_object_form_extracts_all_fields() {
716        let frame = serde_json::json!({
717            "error": {
718                "message": "Context window limit reached.",
719                "type": "context_length_exceeded",
720                "code": "context_length_exceeded"
721            }
722        })
723        .to_string();
724        let events = match parse_sse_frame(&frame) {
725            Ok(SseParseResult::Events(e)) => e,
726            other => panic!("unexpected: {other:?}"),
727        };
728        assert_eq!(events.len(), 1);
729        assert!(matches!(
730            &events[0],
731            LlmStreamEvent::UpstreamError { message, error_type, code }
732                if message == "Context window limit reached."
733                    && error_type == "context_length_exceeded"
734                    && code == "context_length_exceeded"
735        ));
736    }
737
738    #[test]
739    fn inline_error_frame_string_form_uses_defaults() {
740        let frame = serde_json::json!({ "error": "boom" }).to_string();
741        let events = match parse_sse_frame(&frame) {
742            Ok(SseParseResult::Events(e)) => e,
743            other => panic!("unexpected: {other:?}"),
744        };
745        assert_eq!(events.len(), 1);
746        assert!(matches!(
747            &events[0],
748            LlmStreamEvent::UpstreamError { message, error_type, code }
749                if message == "boom" && error_type == "server_error" && code == "upstream_error"
750        ));
751    }
752
753    #[test]
754    fn inline_error_frame_not_dropped_by_no_choices_guard() {
755        // No `choices` key at all -- would be silently dropped by the "no
756        // choices" guard if the error check didn't run first.
757        let frame = serde_json::json!({ "error": { "message": "oops" } }).to_string();
758        let events = match parse_sse_frame(&frame) {
759            Ok(SseParseResult::Events(e)) => e,
760            other => panic!("unexpected: {other:?}"),
761        };
762        assert!(!events.is_empty(), "inline error frame must not be skipped");
763    }
764
765    #[test]
766    fn error_alongside_choices_key_is_not_treated_as_inline_error() {
767        // `choices` key present (even empty) means this isn't the bare
768        // inline-error shape the extension detects via `!('choices' in
769        // obj)` -- falls through to the normal "no choices" skip instead.
770        let frame =
771            serde_json::json!({ "error": { "message": "oops" }, "choices": [] }).to_string();
772        let events = match parse_sse_frame(&frame) {
773            Ok(SseParseResult::Events(e)) => e,
774            other => panic!("unexpected: {other:?}"),
775        };
776        assert!(
777            events.is_empty(),
778            "frame with a choices key should not be parsed as UpstreamError"
779        );
780    }
781}
gglib_core/sse/parser.rs

gglib_core/sse/
parser.rs
