1use crate::llm::actions::{
2 client_trait::{Client, ClientActionResult},
3 protocol_trait::Protocol,
4 ActionDefinition, ParameterDefinition,
5};
6use crate::protocol::{ConnectContext, EventType};
7use crate::server::socketio::actions::{action, parameter};
8use crate::server::socketio::packet;
9use anyhow::{bail, ensure, Context, Result};
10use serde_json::{json, Value};
11use std::sync::LazyLock;
12
13#[derive(Default)]
14pub struct SocketIoClientProtocol;
15impl SocketIoClientProtocol {
16 pub fn new() -> Self {
17 Self
18 }
19}
20
21fn emit_action() -> ActionDefinition {
22 action(
23 "socketio_emit",
24 "Emit an event on a connected namespace; with ack the server's acknowledgement arrives as socketio_ack_received",
25 vec![
26 parameter("event", "string", "Event name, e.g. chat message (not a reserved name like connect)", true),
27 parameter("args", "array", "Event arguments as JSON values, e.g. [\"hello\"]", false),
28 parameter("namespace", "string", "Namespace to emit on (default /)", false),
29 parameter("ack", "boolean", "Ask the server to acknowledge this event", false),
30 ],
31 json!({"type":"socketio_emit","event":"chat message","args":["hello"],"ack":true}),
32 )
33}
34fn ack_action() -> ActionDefinition {
35 action(
36 "socketio_ack",
37 "Acknowledge an event the server sent with an acknowledgement id",
38 vec![
39 parameter("ack_id", "number", "The ack_id from socketio_event", true),
40 parameter(
41 "namespace",
42 "string",
43 "The event's namespace (default /)",
44 false,
45 ),
46 parameter(
47 "args",
48 "array",
49 "Arguments for the server's callback",
50 false,
51 ),
52 ],
53 json!({"type":"socketio_ack","ack_id":1,"args":["got it"]}),
54 )
55}
56fn join_action() -> ActionDefinition {
57 action(
58 "socketio_connect_namespace",
59 "Connect to another namespace on the same session",
60 vec![
61 parameter("namespace", "string", "Namespace, e.g. /admin", true),
62 parameter(
63 "auth",
64 "object",
65 "Optional auth payload for the server",
66 false,
67 ),
68 ],
69 json!({"type":"socketio_connect_namespace","namespace":"/admin"}),
70 )
71}
72fn leave_action() -> ActionDefinition {
73 action(
74 "socketio_disconnect_namespace",
75 "Leave a namespace (Socket.IO DISCONNECT); the session stays open",
76 vec![parameter("namespace", "string", "Namespace to leave", true)],
77 json!({"type":"socketio_disconnect_namespace","namespace":"/admin"}),
78 )
79}
80fn disconnect_action() -> ActionDefinition {
81 action(
82 "disconnect",
83 "Close the Engine.IO session and stop this client",
84 vec![],
85 json!({"type":"disconnect"}),
86 )
87}
88fn actions() -> Vec<ActionDefinition> {
89 vec![
90 emit_action(),
91 ack_action(),
92 join_action(),
93 leave_action(),
94 disconnect_action(),
95 ]
96}
97
98pub static CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
99 EventType::new(
100 "socketio_connected",
101 "A namespace accepted the connection",
102 emit_action().example.clone(),
103 )
104 .with_parameters(vec![
105 parameter("namespace", "string", "The connected namespace", true),
106 parameter("socket_id", "string", "Socket id the server assigned", true),
107 parameter("transport", "string", "websocket or polling", true),
108 ])
109 .with_actions(actions())
110});
111pub static CONNECT_ERROR_EVENT: LazyLock<EventType> = LazyLock::new(|| {
112 EventType::new(
113 "socketio_connect_error",
114 "A namespace refused the connection (CONNECT_ERROR)",
115 disconnect_action().example.clone(),
116 )
117 .with_parameters(vec![
118 parameter("namespace", "string", "The refused namespace", true),
119 parameter("message", "string", "The server's reason", true),
120 parameter("data", "object", "Extra details, when sent", false),
121 ])
122 .with_actions(actions())
123});
124pub static EVENT_EVENT: LazyLock<EventType> = LazyLock::new(|| {
125 EventType::new(
126 "socketio_event",
127 "The server emitted an event",
128 ack_action().example.clone(),
129 )
130 .with_parameters(vec![
131 parameter("namespace", "string", "Namespace it arrived on", true),
132 parameter(
133 "event",
134 "string",
135 "Name of the event the peer emitted, e.g. chat message",
136 true,
137 ),
138 parameter("args", "array", "Event arguments", true),
139 parameter(
140 "ack_id",
141 "number",
142 "Present when the server waits for socketio_ack",
143 false,
144 ),
145 ])
146 .with_actions(actions())
147});
148pub static ACK_EVENT: LazyLock<EventType> = LazyLock::new(|| {
149 EventType::new(
150 "socketio_ack_received",
151 "The server acknowledged an event emitted with ack",
152 emit_action().example.clone(),
153 )
154 .with_parameters(vec![
155 parameter("namespace", "string", "Namespace of the event", true),
156 parameter("event", "string", "The acknowledged event", true),
157 parameter(
158 "args",
159 "array",
160 "The server's acknowledgement arguments",
161 true,
162 ),
163 ])
164 .with_actions(actions())
165});
166pub static DISCONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
167 EventType::new(
168 "socketio_disconnected",
169 "The server disconnected a namespace (or the whole session)",
170 join_action().example.clone(),
171 )
172 .with_parameters(vec![
173 parameter(
174 "namespace",
175 "string",
176 "The namespace, or * for the whole session",
177 true,
178 ),
179 parameter(
180 "reason",
181 "string",
182 "io server disconnect, transport close or ping timeout",
183 true,
184 ),
185 ])
186 .with_actions(actions())
187});
188
189impl Protocol for SocketIoClientProtocol {
190 fn protocol_name(&self) -> &'static str {
191 "Socket.IO"
192 }
193 fn stack_name(&self) -> &'static str {
194 "ETH>IP>TCP>HTTP>Engine.IO>Socket.IO"
195 }
196 fn keywords(&self) -> Vec<&'static str> {
197 vec!["socket.io", "socketio", "engine.io", "socket.io client"]
198 }
199 fn description(&self) -> &'static str {
200 "Socket.IO v5 client over Engine.IO v4 (WebSocket or long-polling): namespaces, events and acknowledgements"
201 }
202 fn get_async_actions(&self, _: &crate::state::app_state::AppState) -> Vec<ActionDefinition> {
203 actions()
204 }
205 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
206 vec![]
207 }
208 fn get_event_types(&self) -> Vec<EventType> {
209 vec![
210 CONNECTED_EVENT.clone(),
211 CONNECT_ERROR_EVENT.clone(),
212 EVENT_EVENT.clone(),
213 ACK_EVENT.clone(),
214 DISCONNECTED_EVENT.clone(),
215 ]
216 }
217 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
218 let p =
219 |name: &str, kind: &str, description: &str, example: Value, default: Option<Value>| {
220 ParameterDefinition {
221 name: name.into(),
222 type_hint: kind.into(),
223 description: description.into(),
224 required: false,
225 example,
226 default,
227 }
228 };
229 vec![
230 p(
231 "path",
232 "string",
233 "HTTP path the server serves Engine.IO on",
234 json!("/realtime/"),
235 Some(json!(crate::server::socketio::DEFAULT_PATH)),
236 ),
237 p(
238 "transport",
239 "string",
240 "websocket (direct) or polling (HTTP long-polling only)",
241 json!("polling"),
242 Some(json!(super::DEFAULT_TRANSPORT)),
243 ),
244 p(
245 "namespaces",
246 "array",
247 "Namespaces to connect to on startup",
248 json!(["/", "/chat"]),
249 Some(json!(crate::server::socketio::DEFAULT_NAMESPACES)),
250 ),
251 p(
252 "auth",
253 "object",
254 "Auth payload sent with every namespace CONNECT",
255 json!({"token": "abc"}),
256 None,
257 ),
258 ]
259 }
260 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
261 use crate::protocol::metadata::*;
262 ProtocolMetadataV2::builder()
263 .state(DevelopmentState::Experimental)
264 .privilege_requirement(PrivilegeRequirement::None)
265 .implementation("Engine.IO v4 over a direct WebSocket (tokio-tungstenite) or HTTP long-polling (reqwest); answers server pings; Socket.IO v5 packets, namespaces and acknowledgement ids in Rust")
266 .llm_control("What to emit, which events to acknowledge and how, and which namespaces to join or leave")
267 .e2e_testing("tests/client/socketio: python-socketio 5.17.0's server (independent) over WebSocket and polling: connects, emits with acks, receives broadcasts and server acks, joins a second namespace and is refused a forbidden one")
268 .notes("No transport upgrade from polling to WebSocket, no binary attachments, no automatic reconnection. 1 MB payloads.")
269 .max_inbound_bytes(packet::MAX_PAYLOAD)
270 .build()
271 }
272 fn example_prompt(&self) -> &'static str {
273 "Join the Socket.IO chat at 127.0.0.1:3000 and say hello"
274 }
275 fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
276 let llm = json!({"type":"open_client","protocol":"socketio","remote_addr":"127.0.0.1:3000","instruction":"Say hello in the chat"});
277 let mut static_example = llm.clone();
278 static_example["event_handlers"] = json!([
279 {"event_pattern":"socketio_connected","handler":{"type":"static","actions":[emit_action().example]}},
280 {"event_pattern":"socketio_event","handler":{"type":"static","actions":[]}}
281 ]);
282 let mut scripted = static_example.clone();
283 scripted["event_handlers"][1]["handler"] = json!({"type":"script","language":"python","code":"import json\nprint(json.dumps({'actions':[]}))"});
284 crate::llm::actions::StartupExamples::new(llm, scripted, static_example)
285 }
286 fn group_name(&self) -> &'static str {
287 "Real-time"
288 }
289}
290
291impl Client for SocketIoClientProtocol {
292 fn connect(
293 &self,
294 ctx: ConnectContext,
295 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
296 {
297 Box::pin(super::connect(ctx))
298 }
299 fn execute_action(&self, v: Value) -> Result<ClientActionResult> {
300 ensure!(
301 crate::utils::json_budget::within_budget(&v, packet::MAX_PAYLOAD, 20_000, 32),
302 "action exceeds the Socket.IO bounds"
303 );
304 let nsp_ok = |k: &str| -> Result<()> {
305 if let Some(n) = v.get(k).filter(|n| !n.is_null()) {
306 ensure!(
307 n.as_str().is_some_and(packet::namespace_ok),
308 "{k} must look like /name"
309 );
310 }
311 Ok(())
312 };
313 match v["type"].as_str() {
314 Some("socketio_emit") => {
315 let e = v["event"].as_str().context("event is required")?;
316 ensure!(
317 !e.is_empty() && e.len() <= packet::MAX_EVENT_NAME && !packet::reserved(e),
318 "event must be 1..128 characters and not reserved"
319 );
320 if let Some(a) = v.get("args").filter(|a| !a.is_null()) {
321 ensure!(
322 a.as_array().is_some_and(|a| a.len() <= packet::MAX_ARGS),
323 "args must be an array of at most 16 values"
324 );
325 }
326 nsp_ok("namespace")?;
327 }
328 Some("socketio_ack") => {
329 ensure!(v["ack_id"].as_u64().is_some(), "ack_id is required");
330 nsp_ok("namespace")?;
331 }
332 Some("socketio_connect_namespace" | "socketio_disconnect_namespace") => {
333 ensure!(
334 v["namespace"].as_str().is_some_and(packet::namespace_ok),
335 "namespace must look like /name"
336 );
337 if let Some(a) = v.get("auth").filter(|a| !a.is_null()) {
338 ensure!(a.is_object(), "auth must be an object");
339 }
340 }
341 Some("disconnect") => return Ok(ClientActionResult::Disconnect),
342 _ => bail!("Unknown Socket.IO client action"),
343 }
344 Ok(ClientActionResult::Custom {
345 name: v["type"].as_str().unwrap_or_default().into(),
346 data: v,
347 })
348 }
349}