lago_api/sse/
lago.rs

1use lago_core::EventEnvelope;
2
3use super::format::{SseFormat, SseFrame};
4
5/// Native Lago SSE format. Events are sent as-is in their JSON envelope
6/// representation, preserving the full event structure.
7pub struct LagoFormat;
8
9impl SseFormat for LagoFormat {
10    fn format(&self, event: &EventEnvelope) -> Vec<SseFrame> {
11        let data = match serde_json::to_string(event) {
12            Ok(json) => json,
13            Err(e) => {
14                tracing::warn!(error = %e, "failed to serialize event envelope");
15                return Vec::new();
16            }
17        };
18
19        vec![SseFrame {
20            event: Some("event".to_string()),
21            data,
22            id: Some(event.seq.to_string()),
23        }]
24    }
25
26    fn done_frame(&self) -> Option<SseFrame> {
27        Some(SseFrame {
28            event: Some("done".to_string()),
29            data: r#"{"type":"done"}"#.to_string(),
30            id: None,
31        })
32    }
33
34    fn extra_headers(&self) -> Vec<(String, String)> {
35        Vec::new()
36    }
37
38    fn name(&self) -> &str {
39        "lago"
40    }
41}
42
43#[cfg(test)]
44mod tests {
45    use super::*;
46    use lago_core::event::EventPayload;
47    use lago_core::id::*;
48    use std::collections::HashMap;
49
50    fn make_envelope(payload: EventPayload, seq: u64) -> EventEnvelope {
51        EventEnvelope {
52            event_id: EventId::from_string("EVT001"),
53            session_id: SessionId::from_string("SESS001"),
54            branch_id: BranchId::from_string("main"),
55            run_id: None,
56            seq,
57            timestamp: 1_700_000_000_000_000,
58            parent_id: None,
59            payload,
60            metadata: HashMap::new(),
61            schema_version: 1,
62        }
63    }
64
65    #[test]
66    fn lago_format_passes_through_all_events() {
67        let fmt = LagoFormat;
68
69        // Message events
70        let event = make_envelope(
71            EventPayload::Message {
72                role: "user".into(),
73                content: "hi".into(),
74                model: None,
75                token_usage: None,
76            },
77            1,
78        );
79        let frames = fmt.format(&event);
80        assert_eq!(frames.len(), 1);
81        assert_eq!(frames[0].event.as_deref(), Some("event"));
82        assert_eq!(frames[0].id.as_deref(), Some("1"));
83
84        // The data should be a valid JSON EventEnvelope
85        let back: EventEnvelope = serde_json::from_str(&frames[0].data).unwrap();
86        assert_eq!(back.event_id.as_str(), "EVT001");
87    }
88
89    #[test]
90    fn lago_format_includes_file_events() {
91        let fmt = LagoFormat;
92        let event = make_envelope(
93            EventPayload::FileDelete {
94                path: "/tmp/x".into(),
95            },
96            5,
97        );
98        let frames = fmt.format(&event);
99        assert_eq!(frames.len(), 1);
100        // Verify the full envelope is in the data
101        let back: EventEnvelope = serde_json::from_str(&frames[0].data).unwrap();
102        if let EventPayload::FileDelete { path } = &back.payload {
103            assert_eq!(path, "/tmp/x");
104        } else {
105            panic!("wrong payload");
106        }
107    }
108
109    #[test]
110    fn lago_done_frame() {
111        let fmt = LagoFormat;
112        let done = fmt.done_frame().unwrap();
113        assert_eq!(done.event.as_deref(), Some("done"));
114        assert_eq!(done.data, r#"{"type":"done"}"#);
115    }
116
117    #[test]
118    fn lago_no_extra_headers() {
119        assert!(LagoFormat.extra_headers().is_empty());
120    }
121
122    #[test]
123    fn name_is_lago() {
124        assert_eq!(LagoFormat.name(), "lago");
125    }
126}
lago_api/sse/lago.rs

lago_api/sse/
lago.rs
