netget/server/sse/
actions.rs

1//! What the model decides as a server-sent-events server: which requests get a stream, what
2//! each stream carries, and what is broadcast to every subscriber. Rust owns the HTTP framing,
3//! the event-stream encoding, keep-alive comments and the subscriber table.
4use crate::llm::actions::{
5    protocol_trait::{ActionResult, Protocol, Server},
6    ActionDefinition, Parameter, ParameterDefinition, StartupExamples,
7};
8use crate::protocol::log_template::LogTemplate;
9use crate::protocol::{EventType, SpawnContext};
10use crate::state::app_state::AppState;
11use anyhow::{bail, ensure, Context, Result};
12use serde_json::{json, Value};
13use std::sync::LazyLock;
14
15pub const ACCEPT: &str = "sse_accept";
16pub const EVENT: &str = "sse_event";
17pub const COMMENT: &str = "sse_comment";
18pub const CLOSE: &str = "sse_close";
19pub const RESPOND: &str = "sse_respond";
20pub const BROADCAST: &str = "sse_broadcast";
21/// How often an idle stream gets a keep-alive comment.
22pub const KEEPALIVE_SECS: u64 = 15;
23/// A `retry:` hint at most (one hour), in milliseconds.
24pub const MAX_RETRY_MS: u64 = 3_600_000;
25
26#[derive(Default)]
27pub struct SseProtocol;
28impl SseProtocol {
29    pub fn new() -> Self {
30        Self
31    }
32}
33
34pub fn p(name: &str, kind: &str, description: &str, required: bool) -> Parameter {
35    Parameter {
36        name: name.into(),
37        type_hint: kind.into(),
38        description: description.into(),
39        required,
40    }
41}
42
43pub fn action(
44    name: &str,
45    description: &str,
46    parameters: Vec<Parameter>,
47    example: Value,
48) -> ActionDefinition {
49    ActionDefinition {
50        name: name.into(),
51        description: description.into(),
52        parameters,
53        example,
54        log_template: Some(LogTemplate::new().with_info(format!("-> SSE {name}"))),
55    }
56}
57
58fn event_params() -> Vec<Parameter> {
59    vec![
60        p(
61            "data",
62            "string",
63            "The event's data; each line of it becomes one data: line",
64            true,
65        ),
66        p(
67            "event",
68            "string",
69            "The event type (the client's addEventListener name); omitted means message",
70            false,
71        ),
72        p(
73            "id",
74            "string",
75            "The event ID; a reconnecting client sends the last one back as Last-Event-ID",
76            false,
77        ),
78    ]
79}
80
81pub fn actions() -> Vec<ActionDefinition> {
82    vec![
83        action(
84            ACCEPT,
85            "Answer the request with an event stream (200 text/event-stream) and keep it open. Events in the same answer follow at once",
86            vec![p("retry_ms", "number", "Tell the client how long to wait before reconnecting, in milliseconds", false)],
87            json!({"type": ACCEPT, "retry_ms": 3000}),
88        ),
89        action(
90            EVENT,
91            "Send one event on this stream (accepting the request first if it is not a stream yet)",
92            event_params(),
93            json!({"type": EVENT, "event": "update", "id": "1", "data": "hello"}),
94        ),
95        action(
96            COMMENT,
97            "Send a comment line, which clients ignore",
98            vec![p("text", "string", "The comment, one line", true)],
99            json!({"type": COMMENT, "text": "still here"}),
100        ),
101        action(
102            CLOSE,
103            "End this stream. A browser EventSource reconnects after its retry delay with Last-Event-ID",
104            vec![],
105            json!({"type": CLOSE}),
106        ),
107        action(
108            RESPOND,
109            "Answer with an ordinary HTTP response instead of a stream: 204 tells an EventSource to stop reconnecting, 401/404 refuse",
110            vec![
111                p("status", "number", "HTTP status, 200-599", true),
112                p("body", "string", "Response body text (default empty)", false),
113                p("content_type", "string", "Content-Type of the body (default text/plain; charset=utf-8)", false),
114            ],
115            json!({"type": RESPOND, "status": 404, "body": "no such stream"}),
116        ),
117        action(
118            BROADCAST,
119            "Send one event to every open stream (or only those on one path)",
120            {
121                let mut v = event_params();
122                v.push(p("path", "string", "Only streams opened on this path, e.g. /news (default every stream)", false));
123                v
124            },
125            json!({"type": BROADCAST, "path": "/news", "event": "post", "id": "7", "data": "new article"}),
126        ),
127    ]
128}
129
130pub static REQUEST_EVENT: LazyLock<EventType> = LazyLock::new(|| {
131    EventType::new(
132        "sse_request",
133        "An HTTP request arrived: answer it with a stream (sse_accept and events), an ordinary response, or a broadcast",
134        json!({"type": ACCEPT}),
135    )
136    .with_parameters(vec![
137        p("method", "string", "The HTTP method (GET or HEAD)", true),
138        p("path", "string", "The request path without the query, e.g. /news", true),
139        p("query", "object", "The query string's parameters", true),
140        p("last_event_id", "string", "Last-Event-ID a reconnecting client sent (the last event ID it saw), empty on a first connection", true),
141        p("accept", "string", "The Accept header (an EventSource sends text/event-stream)", false),
142        p("user_agent", "string", "The User-Agent header", false),
143        p("client_ip", "string", "Where the request came from", true),
144    ])
145    .with_actions(actions())
146});
147
148pub fn check(v: &Value) -> Result<()> {
149    let text = |k: &str, required: bool| -> Result<Option<String>> {
150        match v.get(k).filter(|x| !x.is_null()) {
151            Some(Value::String(s)) => Ok(Some(s.clone())),
152            Some(other) => Ok(Some(other.to_string())),
153            None => {
154                ensure!(!required, "{k} is required");
155                Ok(None)
156            }
157        }
158    };
159    match v["type"].as_str().unwrap_or_default() {
160        ACCEPT => {
161            if let Some(r) = v.get("retry_ms").filter(|x| !x.is_null()) {
162                ensure!(
163                    r.as_u64().is_some_and(|r| r <= MAX_RETRY_MS),
164                    "retry_ms is 0..={MAX_RETRY_MS}"
165                );
166            }
167        }
168        EVENT | BROADCAST => {
169            let data = text("data", true)?.unwrap_or_default();
170            super::wire::encode_event(
171                text("event", false)?.as_deref(),
172                text("id", false)?.as_deref(),
173                &data,
174                None,
175            )?;
176            if let Some(path) = text("path", false)? {
177                ensure!(
178                    path.starts_with('/') && path.len() <= 2048,
179                    "path starts with /"
180                );
181            }
182        }
183        COMMENT => {
184            super::wire::encode_comment(&text("text", true)?.unwrap_or_default())?;
185        }
186        CLOSE => {}
187        RESPOND => {
188            let status = v["status"].as_u64().context("status is a number")?;
189            ensure!((200..=599).contains(&status), "status is 200..=599");
190            if let Some(ct) = text("content_type", false)? {
191                ensure!(!ct.contains(['\r', '\n']), "content_type is one line");
192            }
193            ensure!(
194                text("body", false)?.is_none_or(|b| b.len() <= 1024 * 1024),
195                "body is at most 1 MiB"
196            );
197        }
198        other => bail!("Unknown SSE action {other:?}"),
199    }
200    Ok(())
201}
202
203/// The encoded event for `sse_event`/`sse_broadcast`.
204pub fn encode(v: &Value) -> Result<Vec<u8>> {
205    let s = |k: &str| match v.get(k) {
206        Some(Value::String(s)) => Some(s.clone()),
207        Some(Value::Null) | None => None,
208        Some(other) => Some(other.to_string()),
209    };
210    super::wire::encode_event(
211        s("event").as_deref(),
212        s("id").as_deref(),
213        &s("data").unwrap_or_default(),
214        None,
215    )
216}
217
218impl Protocol for SseProtocol {
219    fn protocol_name(&self) -> &'static str {
220        "SSE"
221    }
222    fn stack_name(&self) -> &'static str {
223        "ETH>IP>TCP>HTTP>SSE"
224    }
225    fn keywords(&self) -> Vec<&'static str> {
226        vec![
227            "sse",
228            "server-sent events",
229            "eventsource",
230            "event stream",
231            "text/event-stream",
232        ]
233    }
234    fn description(&self) -> &'static str {
235        "Server-sent events server: streams events to EventSource clients, per stream and by broadcast"
236    }
237    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
238        actions()
239            .into_iter()
240            .filter(|a| a.name == BROADCAST)
241            .collect()
242    }
243    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
244        actions()
245    }
246    fn get_event_types(&self) -> Vec<EventType> {
247        vec![REQUEST_EVENT.clone()]
248    }
249    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
250        vec![ParameterDefinition {
251            name: "keepalive_secs".into(),
252            type_hint: "number".into(),
253            description: "Seconds (1..=3600) an open stream may be silent before a keep-alive comment is sent".into(),
254            required: false,
255            example: json!(15),
256            default: Some(json!(KEEPALIVE_SECS)),
257        }]
258    }
259    fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
260        use crate::protocol::metadata::*;
261        ProtocolMetadataV2::builder()
262            .state(DevelopmentState::Experimental)
263            .privilege_requirement(PrivilegeRequirement::None)
264            .implementation("Hand-rolled HTTP/1.1 over Tokio TCP: one request per connection, answered with a text/event-stream kept open (keep-alive comments, retry hints) or an ordinary response; WHATWG event-stream encoding; a per-server subscriber table for broadcasts; each stream registered as a peer for injected events")
265            .llm_control("Which requests get a stream, what each stream says and when it ends, and what is broadcast to every subscriber")
266            .e2e_testing("tests/server/sse: the eventsource npm package (WHATWG EventSource, reconnecting with Last-Event-ID) and Python's httpx-sse read NetGet's streams, broadcasts and injected events")
267            .notes("HTTP/1.1 only, no TLS, no request bodies (GET/HEAD); one request per connection. Request heads are capped at 16 KiB and 10 s; event data at 1 MiB.")
268            .answers_on_failure()
269            .max_inbound_bytes(super::MAX_REQUEST_HEAD_BYTES)
270            .build()
271    }
272    fn example_prompt(&self) -> &'static str {
273        "Event stream on port 8080 that greets each subscriber of /news and broadcasts whatever is posted to /post?msg="
274    }
275    fn get_startup_examples(&self) -> StartupExamples {
276        let llm = json!({"type":"open_server","base_stack":"sse","port":8080,
277            "instruction":"Stream /news: greet each subscriber with an event. A request to /post?msg=X broadcasts X to /news and answers 202"});
278        let mut static_example = llm.clone();
279        static_example["event_handlers"] = json!([{"event_pattern":"sse_request","handler":{"type":"static","actions":[
280            {"type":ACCEPT,"retry_ms":3000},{"type":EVENT,"event":"welcome","id":"1","data":"hello"}]}}]);
281        let mut scripted = llm.clone();
282        scripted["event_handlers"] = json!([{"event_pattern":"*","handler":{"type":"script","language":"python",
283            "code":"import json,sys\ne=json.load(sys.stdin)['event']\nif e['path']=='/post':\n  a=[{'type':'sse_broadcast','path':'/news','data':e['query'].get('msg','')},{'type':'sse_respond','status':202}]\nelse:\n  a=[{'type':'sse_accept'},{'type':'sse_event','data':'welcome'}]\nprint(json.dumps({'actions':a}))"}}]);
284        StartupExamples::new(llm, scripted, static_example)
285    }
286    fn group_name(&self) -> &'static str {
287        "Web & File"
288    }
289}
290
291impl Server for SseProtocol {
292    fn spawn(
293        &self,
294        ctx: SpawnContext,
295    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
296    {
297        Box::pin(super::spawn(ctx))
298    }
299
300    fn execute_action_with_state<'a>(
301        &'a self,
302        action: Value,
303        state: AppState,
304        server_id: Option<crate::state::ServerId>,
305    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<ActionResult>> + Send + 'a>>
306    {
307        if action["type"].as_str() != Some(BROADCAST) {
308            let result = self.execute_action(action);
309            return Box::pin(async move { result });
310        }
311        Box::pin(async move {
312            check(&action)?;
313            let id = server_id.context("sse_broadcast needs the running SSE server")?;
314            let bytes = encode(&action)?;
315            let delivered = super::broadcast(&state, id, action["path"].as_str(), &bytes).await;
316            Ok(ActionResult::Custom {
317                name: BROADCAST.into(),
318                data: json!({"delivered": delivered}),
319            })
320        })
321    }
322
323    fn execute_action(&self, action: Value) -> Result<ActionResult> {
324        check(&action)?;
325        Ok(match action["type"].as_str().unwrap_or_default() {
326            EVENT => ActionResult::Output(encode(&action)?),
327            COMMENT => ActionResult::Output(super::wire::encode_comment(
328                action["text"].as_str().unwrap_or_default(),
329            )?),
330            CLOSE => ActionResult::CloseConnection,
331            name => ActionResult::Custom {
332                name: name.into(),
333                data: action,
334            },
335        })
336    }
337}
netget/server/sse/actions.rs

netget/server/sse/
actions.rs
