1use super::packet;
2use crate::llm::actions::{
3 protocol_trait::{ActionResult, Protocol, Server},
4 ActionDefinition, Parameter, ParameterDefinition,
5};
6use crate::protocol::log_template::LogTemplate;
7use crate::protocol::{EventType, SpawnContext};
8use crate::state::app_state::AppState;
9use anyhow::{bail, ensure, Context, Result};
10use serde_json::{json, Value};
11use std::sync::LazyLock;
12
13#[derive(Default)]
14pub struct SocketIoProtocol;
15impl SocketIoProtocol {
16 pub fn new() -> Self {
17 Self
18 }
19}
20
21pub fn parameter(name: &str, kind: &str, description: &str, required: bool) -> Parameter {
22 Parameter {
23 name: name.into(),
24 type_hint: kind.into(),
25 description: description.into(),
26 required,
27 }
28}
29
30pub fn action(
31 name: &str,
32 description: &str,
33 parameters: Vec<Parameter>,
34 example: Value,
35) -> ActionDefinition {
36 ActionDefinition {
37 name: name.into(),
38 description: description.into(),
39 parameters,
40 example,
41 log_template: Some(LogTemplate::new().with_info(format!("-> Socket.IO {name}"))),
42 }
43}
44
45pub fn emit_action() -> ActionDefinition {
46 action(
47 "socketio_emit",
48 "Emit an event. Rust encodes the Socket.IO packet and delivers it to the chosen sockets over their Engine.IO transport (polling or WebSocket).",
49 vec![
50 parameter("event", "string", "Event name, e.g. chat message (not a reserved name like connect or disconnect)", true),
51 parameter("args", "array", "Event arguments as JSON values, e.g. [\"hello\", {\"from\": \"bot\"}]", false),
52 parameter("to", "string", "Recipients: sender (default, the socket that triggered this), namespace (every socket in it), room:<name>, or socket:<id>", false),
53 parameter("namespace", "string", "Namespace to emit in; defaults to the one the event came from, else /", false),
54 parameter("include_sender", "boolean", "For namespace and room broadcasts: also deliver to the triggering socket (default true)", false),
55 parameter("ack", "boolean", "Ask the single recipient to acknowledge; its answer arrives as socketio_ack_received", false),
56 ],
57 json!({"type":"socketio_emit","event":"chat message","args":["welcome!"]}),
58 )
59}
60fn ack_action() -> ActionDefinition {
61 action(
62 "socketio_ack",
63 "Acknowledge the event the client sent with an acknowledgement id (its callback receives these arguments)",
64 vec![parameter("args", "array", "Arguments passed to the client's callback, e.g. [{\"ok\": true}]", false)],
65 json!({"type":"socketio_ack","args":[{"ok":true}]}),
66 )
67}
68fn join_action() -> ActionDefinition {
69 action(
70 "socketio_join",
71 "Add the triggering socket to a room in its namespace",
72 vec![parameter(
73 "room",
74 "string",
75 "Room name, 1 to 128 characters",
76 true,
77 )],
78 json!({"type":"socketio_join","room":"lobby"}),
79 )
80}
81fn leave_action() -> ActionDefinition {
82 action(
83 "socketio_leave",
84 "Remove the triggering socket from a room",
85 vec![parameter("room", "string", "Room name to leave", true)],
86 json!({"type":"socketio_leave","room":"lobby"}),
87 )
88}
89pub fn disconnect_socket_action() -> ActionDefinition {
90 action("socketio_disconnect_socket", "Disconnect the socket from its namespace (Socket.IO DISCONNECT); the Engine.IO session stays open", vec![], json!({"type":"socketio_disconnect_socket"}))
91}
92fn accept_action() -> ActionDefinition {
93 action(
94 "socketio_accept",
95 "Accept the namespace connection; Rust assigns the socket id and answers CONNECT",
96 vec![],
97 json!({"type":"socketio_accept"}),
98 )
99}
100fn reject_action() -> ActionDefinition {
101 action(
102 "socketio_reject",
103 "Refuse the namespace connection with CONNECT_ERROR",
104 vec![
105 parameter(
106 "message",
107 "string",
108 "Reason the client's connect_error handler receives, e.g. not authorized",
109 true,
110 ),
111 parameter(
112 "data",
113 "object",
114 "Optional extra details sent with the error",
115 false,
116 ),
117 ],
118 json!({"type":"socketio_reject","message":"not authorized"}),
119 )
120}
121fn close_action() -> ActionDefinition {
122 action(
123 "disconnect",
124 "Close the whole Engine.IO session (every namespace on it)",
125 vec![],
126 json!({"type":"disconnect"}),
127 )
128}
129
130fn reacting() -> Vec<ActionDefinition> {
131 vec![
132 emit_action(),
133 ack_action(),
134 join_action(),
135 leave_action(),
136 disconnect_socket_action(),
137 ]
138}
139
140pub static CONNECT_EVENT: LazyLock<EventType> = LazyLock::new(|| {
141 EventType::new("socketio_connect", "A client asks to connect to a namespace (Socket.IO CONNECT). Accept or reject; emits after an accept go to the new socket.", accept_action().example.clone())
142 .with_parameters(vec![
143 parameter("namespace", "string", "Namespace requested, e.g. / or /admin", true),
144 parameter("auth", "object", "The client's auth payload, when it sent one", false),
145 parameter("session_id", "string", "Engine.IO session id", true),
146 parameter("transport", "string", "polling or websocket", true),
147 ])
148 .with_actions(vec![accept_action(), reject_action(), emit_action(), join_action()])
149});
150pub static EVENT_EVENT: LazyLock<EventType> = LazyLock::new(|| {
151 EventType::new(
152 "socketio_event",
153 "A client emitted an event",
154 emit_action().example.clone(),
155 )
156 .with_parameters(vec![
157 parameter("socket_id", "string", "The emitting socket", true),
158 parameter(
159 "namespace",
160 "string",
161 "Namespace of the socket, e.g. / or /admin",
162 true,
163 ),
164 parameter(
165 "event",
166 "string",
167 "Name of the event the peer emitted, e.g. chat message",
168 true,
169 ),
170 parameter("args", "array", "Event arguments", true),
171 parameter(
172 "ack_requested",
173 "boolean",
174 "True when the client waits for socketio_ack",
175 true,
176 ),
177 parameter("rooms", "array", "Rooms the socket has joined", true),
178 ])
179 .with_actions(reacting())
180});
181pub static ACK_EVENT: LazyLock<EventType> = LazyLock::new(|| {
182 EventType::new(
183 "socketio_ack_received",
184 "A client acknowledged an event emitted with ack: true",
185 emit_action().example.clone(),
186 )
187 .with_parameters(vec![
188 parameter("socket_id", "string", "The acknowledging socket", true),
189 parameter(
190 "namespace",
191 "string",
192 "Namespace of the socket, e.g. / or /admin",
193 true,
194 ),
195 parameter(
196 "event",
197 "string",
198 "The emitted event being acknowledged",
199 true,
200 ),
201 parameter(
202 "args",
203 "array",
204 "The arguments the client's callback was given",
205 true,
206 ),
207 ])
208 .with_actions(reacting())
209});
210pub static DISCONNECT_EVENT: LazyLock<EventType> = LazyLock::new(|| {
211 EventType::new("socketio_disconnect", "A socket left its namespace (client disconnect, transport close or ping timeout); emit to others if they should know", emit_action().example.clone())
212 .with_parameters(vec![
213 parameter("socket_id", "string", "The socket that left", true),
214 parameter("namespace", "string", "Namespace of the socket, e.g. / or /admin", true),
215 parameter("reason", "string", "client namespace disconnect, client disconnect, transport close, ping timeout or server namespace disconnect", true),
216 ])
217 .with_actions(vec![emit_action()])
218});
219
220fn startup(
221 name: &str,
222 kind: &str,
223 description: &str,
224 example: Value,
225 default: Value,
226) -> ParameterDefinition {
227 ParameterDefinition {
228 name: name.into(),
229 type_hint: kind.into(),
230 description: description.into(),
231 required: false,
232 example,
233 default: Some(default),
234 }
235}
236
237impl Protocol for SocketIoProtocol {
238 fn protocol_name(&self) -> &'static str {
239 "Socket.IO"
240 }
241 fn stack_name(&self) -> &'static str {
242 "ETH>IP>TCP>HTTP>Engine.IO>Socket.IO"
243 }
244 fn description(&self) -> &'static str {
245 "Socket.IO v5 server over Engine.IO v4 (long-polling and WebSocket): namespaces, events, acknowledgements, rooms"
246 }
247 fn keywords(&self) -> Vec<&'static str> {
248 vec![
249 "socket.io",
250 "socketio",
251 "engine.io",
252 "realtime",
253 "websocket events",
254 ]
255 }
256 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
257 vec![emit_action(), disconnect_socket_action(), close_action()]
258 }
259 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
260 vec![
261 accept_action(),
262 reject_action(),
263 emit_action(),
264 ack_action(),
265 join_action(),
266 leave_action(),
267 disconnect_socket_action(),
268 ]
269 }
270 fn get_event_types(&self) -> Vec<EventType> {
271 vec![
272 CONNECT_EVENT.clone(),
273 EVENT_EVENT.clone(),
274 ACK_EVENT.clone(),
275 DISCONNECT_EVENT.clone(),
276 ]
277 }
278 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
279 vec![
280 startup("path", "string", "HTTP path Engine.IO is served on", json!("/realtime/"), json!(super::DEFAULT_PATH)),
281 startup("namespaces", "array", "Namespaces clients may connect to; others get CONNECT_ERROR Invalid namespace", json!(["/", "/admin"]), json!(super::DEFAULT_NAMESPACES)),
282 startup("ping_interval_ms", "integer", "Engine.IO pingInterval: how often the server pings each session (1000 to 120000)", json!(10000), json!(super::DEFAULT_PING_INTERVAL.as_millis() as u64)),
283 startup("ping_timeout_ms", "integer", "Engine.IO pingTimeout: how long a pong may take before the session is closed (1000 to 120000)", json!(5000), json!(super::DEFAULT_PING_TIMEOUT.as_millis() as u64)),
284 ]
285 }
286 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
287 use crate::protocol::metadata::*;
288 ProtocolMetadataV2::builder()
289 .state(DevelopmentState::Experimental)
290 .privilege_requirement(PrivilegeRequirement::None)
291 .implementation("hyper HTTP/1.1 Engine.IO v4: long-polling with record-separated payloads, WebSocket (direct or by probe/upgrade, tokio-tungstenite), server heartbeats; Socket.IO v5 packets, namespaces, acknowledgements both ways, rooms and broadcasts in Rust")
292 .llm_control("Which namespace connections to accept and what to emit, acknowledge, join, leave or disconnect for each event")
293 .e2e_testing("tests/server/socketio: python-socketio 5.17.0 (polling with upgrade, and WebSocket only) and the reference socket.io-client 4.8.4 (JavaScript) connect, emit with acks, receive broadcasts and server acks, use a second namespace and are refused a forbidden one")
294 .notes("Text packets only: binary events and attachments are refused. No CORS headers, no Engine.IO v3 / Socket.IO v4 protocol revisions, no adapter for several processes. 256 sessions, 1 MB payloads, 256 queued packets per session.")
295 .answers_on_failure()
296 .max_inbound_bytes(packet::MAX_PAYLOAD)
297 .build()
298 }
299 fn example_prompt(&self) -> &'static str {
300 "Socket.IO chat server that echoes every chat message to the whole room"
301 }
302 fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
303 let llm = json!({"type":"open_server","base_stack":"socketio","port":3000,"instruction":"A chat server: accept everyone and broadcast every chat message"});
304 let mut static_example = llm.clone();
305 static_example["event_handlers"] = json!([
306 {"event_pattern":"socketio_connect","handler":{"type":"static","actions":[{"type":"socketio_accept"}]}},
307 {"event_pattern":"socketio_event","handler":{"type":"static","actions":[{"type":"socketio_emit","event":"chat message","args":["ok"],"to":"namespace"}]}}
308 ]);
309 let mut scripted = static_example.clone();
310 scripted["event_handlers"][1]["handler"] = json!({"type":"script","language":"python","code":"import json,sys\ne=json.load(sys.stdin)['event']\nprint(json.dumps({'actions':[{'type':'socketio_emit','event':e['event'],'args':e['args'],'to':'namespace'}]}))"});
311 crate::llm::actions::StartupExamples::new(llm, scripted, static_example)
312 }
313 fn group_name(&self) -> &'static str {
314 "Real-time"
315 }
316}
317
318impl Server for SocketIoProtocol {
319 fn spawn(
320 &self,
321 ctx: SpawnContext,
322 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
323 {
324 Box::pin(super::spawn(ctx))
325 }
326 fn execute_action(&self, v: Value) -> Result<ActionResult> {
327 check_action(&v)?;
328 if v["type"] == "disconnect" {
329 return Ok(ActionResult::CloseConnection);
330 }
331 Ok(ActionResult::Custom {
332 name: v["type"].as_str().unwrap_or_default().into(),
333 data: v,
334 })
335 }
336}
337
338pub fn check_action(v: &Value) -> Result<()> {
339 ensure!(
340 crate::utils::json_budget::within_budget(v, packet::MAX_PAYLOAD, 20_000, 32),
341 "action exceeds the Socket.IO bounds"
342 );
343 let room = |k: &str| -> Result<()> {
344 let r = v[k].as_str().with_context(|| format!("{k} is required"))?;
345 ensure!(
346 !r.is_empty() && r.len() <= 128,
347 "{k} must be 1..128 characters"
348 );
349 Ok(())
350 };
351 match v["type"].as_str() {
352 Some("socketio_emit") => {
353 let e = v["event"].as_str().context("event is required")?;
354 ensure!(
355 !e.is_empty() && e.len() <= packet::MAX_EVENT_NAME && !packet::reserved(e),
356 "event must be 1..128 characters and not reserved"
357 );
358 if let Some(a) = v.get("args").filter(|a| !a.is_null()) {
359 ensure!(
360 a.as_array().is_some_and(|a| a.len() <= packet::MAX_ARGS),
361 "args must be an array of at most 16 values"
362 );
363 }
364 if let Some(t) = v.get("to").filter(|t| !t.is_null()) {
365 let t = t.as_str().context("to must be a string")?;
366 ensure!(
367 t == "sender"
368 || t == "namespace"
369 || t.strip_prefix("room:")
370 .is_some_and(|r| !r.is_empty() && r.len() <= 128)
371 || t.strip_prefix("socket:")
372 .is_some_and(|r| !r.is_empty() && r.len() <= 64),
373 "to is sender, namespace, room:<name> or socket:<id>"
374 );
375 }
376 if let Some(n) = v.get("namespace").filter(|n| !n.is_null()) {
377 ensure!(
378 n.as_str().is_some_and(packet::namespace_ok),
379 "namespace must look like /name"
380 );
381 }
382 }
383 Some("socketio_ack") => {
384 if let Some(a) = v.get("args").filter(|a| !a.is_null()) {
385 ensure!(
386 a.as_array().is_some_and(|a| a.len() <= packet::MAX_ARGS),
387 "args must be an array of at most 16 values"
388 );
389 }
390 }
391 Some("socketio_join" | "socketio_leave") => room("room")?,
392 Some("socketio_reject") => {
393 ensure!(
394 v["message"]
395 .as_str()
396 .is_some_and(|m| !m.is_empty() && m.len() <= 1024),
397 "message must be 1..1024 bytes"
398 );
399 }
400 Some("socketio_accept" | "socketio_disconnect_socket" | "disconnect") => {}
401 _ => bail!("Unknown Socket.IO server action"),
402 }
403 Ok(())
404}