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::wamp::actions::{action, parameter, payload_params};
8use crate::server::wamp::uri::uri_ok;
9use anyhow::{bail, ensure, Result};
10use serde_json::{json, Value};
11use std::sync::LazyLock;
12
13#[derive(Default)]
14pub struct WampClientProtocol;
15impl WampClientProtocol {
16 pub fn new() -> Self {
17 Self
18 }
19}
20
21fn with_payload(mut p: Vec<crate::llm::actions::Parameter>) -> Vec<crate::llm::actions::Parameter> {
22 p.extend(payload_params());
23 p
24}
25
26fn subscribe() -> ActionDefinition {
27 action(
28 "wamp_subscribe",
29 "Subscribe to a topic; events arrive as wamp_event",
30 vec![parameter("topic", "string", "Topic URI, or pattern for prefix / wildcard matching (wildcard: empty components match anything, e.g. com..update)", true), parameter("match", "string", "exact (default), prefix or wildcard", false)],
31 json!({"type": "wamp_subscribe", "topic": "com.example.tick"}),
32 )
33}
34fn unsubscribe() -> ActionDefinition {
35 action(
36 "wamp_unsubscribe",
37 "Remove this session's subscription to a topic",
38 vec![parameter(
39 "topic",
40 "string",
41 "The topic (or pattern) subscribed to",
42 true,
43 )],
44 json!({"type": "wamp_unsubscribe", "topic": "com.example.tick"}),
45 )
46}
47fn publish() -> ActionDefinition {
48 action(
49 "wamp_publish",
50 "Publish an event to a topic",
51 with_payload(vec![
52 parameter(
53 "topic",
54 "string",
55 "Topic URI, dot-separated, e.g. com.example.news",
56 true,
57 ),
58 parameter(
59 "acknowledge",
60 "boolean",
61 "Ask the router to confirm with PUBLISHED (default true)",
62 false,
63 ),
64 parameter(
65 "exclude_me",
66 "boolean",
67 "Leave this session out even if subscribed (default true)",
68 false,
69 ),
70 ]),
71 json!({"type": "wamp_publish", "topic": "com.example.news", "args": ["hello"]}),
72 )
73}
74fn call() -> ActionDefinition {
75 action(
76 "wamp_call",
77 "Call a procedure; the answer arrives as wamp_reply",
78 with_payload(vec![parameter(
79 "procedure",
80 "string",
81 "Procedure URI, dot-separated, e.g. com.example.add2",
82 true,
83 )]),
84 json!({"type": "wamp_call", "procedure": "com.example.add2", "args": [2, 3]}),
85 )
86}
87fn register() -> ActionDefinition {
88 action(
89 "wamp_register",
90 "Register a procedure this session answers; calls arrive as wamp_invocation",
91 vec![parameter(
92 "procedure",
93 "string",
94 "Procedure URI, dot-separated, e.g. com.example.add2",
95 true,
96 )],
97 json!({"type": "wamp_register", "procedure": "com.example.echo"}),
98 )
99}
100fn unregister() -> ActionDefinition {
101 action(
102 "wamp_unregister",
103 "Withdraw a procedure this session registered",
104 vec![parameter(
105 "procedure",
106 "string",
107 "Procedure URI, dot-separated, e.g. com.example.add2",
108 true,
109 )],
110 json!({"type": "wamp_unregister", "procedure": "com.example.echo"}),
111 )
112}
113fn yield_() -> ActionDefinition {
114 action(
115 "wamp_yield",
116 "Answer an invocation with its result",
117 with_payload(vec![parameter(
118 "invocation",
119 "number",
120 "The invocation's request ID from wamp_invocation (default: the oldest unanswered)",
121 false,
122 )]),
123 json!({"type": "wamp_yield", "args": ["echo"]}),
124 )
125}
126fn error() -> ActionDefinition {
127 action(
128 "wamp_error",
129 "Answer an invocation with an error",
130 with_payload(vec![
131 parameter(
132 "error",
133 "string",
134 "Error URI, e.g. com.example.error.invalid_input",
135 true,
136 ),
137 parameter(
138 "invocation",
139 "number",
140 "The invocation's request ID from wamp_invocation (default: the oldest unanswered)",
141 false,
142 ),
143 ]),
144 json!({"type": "wamp_error", "error": "com.example.error.invalid_input", "args": ["bad input"]}),
145 )
146}
147fn goodbye() -> ActionDefinition {
148 action(
149 "wamp_goodbye",
150 "Leave the realm with GOODBYE and close",
151 vec![],
152 json!({"type": "wamp_goodbye"}),
153 )
154}
155fn disconnect() -> ActionDefinition {
156 action(
157 "disconnect",
158 "Close the WebSocket without a GOODBYE",
159 vec![],
160 json!({"type": "disconnect"}),
161 )
162}
163fn actions() -> Vec<ActionDefinition> {
164 vec![
165 subscribe(),
166 unsubscribe(),
167 publish(),
168 call(),
169 register(),
170 unregister(),
171 yield_(),
172 error(),
173 goodbye(),
174 disconnect(),
175 ]
176}
177
178pub static WELCOME_EVENT: LazyLock<EventType> = LazyLock::new(|| {
179 EventType::new(
180 "wamp_welcome",
181 "The router admitted this session to the realm",
182 subscribe().example.clone(),
183 )
184 .with_parameters(vec![
185 parameter("session", "number", "This session's WAMP ID", true),
186 parameter("realm", "string", "The realm joined", true),
187 parameter(
188 "roles",
189 "array",
190 "Router roles announced: broker and/or dealer",
191 true,
192 ),
193 ])
194 .with_actions(actions())
195});
196pub static REPLY_EVENT: LazyLock<EventType> = LazyLock::new(|| {
197 EventType::new(
198 "wamp_reply",
199 "The router's answer to a subscribe, unsubscribe, publish, call, register or unregister",
200 json!({"type": "wamp_goodbye"}),
201 )
202 .with_parameters(vec![
203 parameter(
204 "operation",
205 "string",
206 "subscribe, unsubscribe, publish, call, register or unregister",
207 true,
208 ),
209 parameter("target", "string", "The topic or procedure", true),
210 parameter(
211 "ok",
212 "boolean",
213 "false when the router answered ERROR",
214 true,
215 ),
216 parameter("args", "array", "call: result or error arguments", false),
217 parameter(
218 "kwargs",
219 "object",
220 "call: result or error keyword arguments",
221 false,
222 ),
223 parameter("error", "string", "The error URI when ok is false", false),
224 ])
225 .with_actions(actions())
226});
227pub static EVENT_EVENT: LazyLock<EventType> = LazyLock::new(|| {
228 EventType::new(
229 "wamp_event",
230 "An event published to a topic this session subscribed to",
231 json!({"type": "wamp_publish", "topic": "com.example.seen", "args": ["got it"]}),
232 )
233 .with_parameters(vec![
234 parameter(
235 "topic",
236 "string",
237 "The topic (the subscription's, or the actual topic for pattern subscriptions)",
238 true,
239 ),
240 parameter("args", "array", "Positional arguments", true),
241 parameter("kwargs", "object", "Keyword arguments", true),
242 parameter(
243 "publication",
244 "number",
245 "The router's ID for this publication (the same for every subscriber that received it)",
246 true,
247 ),
248 ])
249 .with_actions(actions())
250});
251pub static INVOCATION_EVENT: LazyLock<EventType> = LazyLock::new(|| {
252 EventType::new("wamp_invocation", "A call to a procedure this session registered; answer with wamp_yield or wamp_error (no answer is wamp.error.unavailable)", yield_().example.clone())
253 .with_parameters(vec![
254 parameter("invocation", "number", "Request ID to answer", true),
255 parameter("procedure", "string", "The procedure called", true),
256 parameter("args", "array", "Positional arguments", true),
257 parameter("kwargs", "object", "Keyword arguments", true),
258 ])
259 .with_actions(vec![yield_(), error()])
260});
261pub static LEFT_EVENT: LazyLock<EventType> = LazyLock::new(|| {
262 EventType::new(
263 "wamp_left",
264 "The session ended",
265 disconnect().example.clone(),
266 )
267 .with_parameters(vec![parameter(
268 "reason",
269 "string",
270 "GOODBYE / ABORT reason URI or why the connection closed",
271 true,
272 )])
273 .with_actions(vec![disconnect()])
274});
275
276impl Protocol for WampClientProtocol {
277 fn protocol_name(&self) -> &'static str {
278 "WAMP"
279 }
280 fn stack_name(&self) -> &'static str {
281 "ETH>IP>TCP>HTTP>WebSocket>WAMP"
282 }
283 fn keywords(&self) -> Vec<&'static str> {
284 vec![
285 "wamp",
286 "web application messaging protocol",
287 "rpc",
288 "pubsub",
289 "caller",
290 "callee",
291 "subscriber",
292 ]
293 }
294 fn description(&self) -> &'static str {
295 "WAMP v2 client (JSON over WebSocket) in all four roles: caller, callee, publisher, subscriber"
296 }
297 fn get_async_actions(&self, _: &crate::state::app_state::AppState) -> Vec<ActionDefinition> {
298 actions()
299 }
300 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
301 vec![]
302 }
303 fn get_event_types(&self) -> Vec<EventType> {
304 vec![
305 WELCOME_EVENT.clone(),
306 REPLY_EVENT.clone(),
307 EVENT_EVENT.clone(),
308 INVOCATION_EVENT.clone(),
309 LEFT_EVENT.clone(),
310 ]
311 }
312 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
313 let p =
314 |name: &str, kind: &str, description: &str, example: Value, default: Option<Value>| {
315 ParameterDefinition {
316 name: name.into(),
317 type_hint: kind.into(),
318 description: description.into(),
319 required: false,
320 example,
321 default,
322 }
323 };
324 vec![
325 p(
326 "realm",
327 "string",
328 "Realm (routing domain) to join, as a URI, e.g. realm1",
329 json!("realm1"),
330 Some(json!(super::DEFAULT_REALM)),
331 ),
332 p(
333 "path",
334 "string",
335 "WebSocket path on the router",
336 json!("/ws"),
337 Some(json!(super::DEFAULT_PATH)),
338 ),
339 p(
340 "authid",
341 "string",
342 "authid to announce in HELLO (anonymous authentication)",
343 json!("netget"),
344 None,
345 ),
346 ]
347 }
348 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
349 use crate::protocol::metadata::*;
350 ProtocolMetadataV2::builder()
351 .state(DevelopmentState::Experimental)
352 .privilege_requirement(PrivilegeRequirement::None)
353 .implementation("tokio-tungstenite with the wamp.2.json subprotocol; hand-written WAMP v2 basic-profile session in all four roles")
354 .llm_control("What to call, publish, subscribe and register, and the answer to every invocation of a registered procedure")
355 .e2e_testing("tests/client/wamp: the nexus 3.3.0 router (Go; independent) with a nexus callee, publisher and subscriber: call, error, subscribe, publish, register and answer invocations")
356 .notes("JSON serializer only; anonymous authentication; ws:// only; no progressive results or cancellation.")
357 .max_inbound_bytes(crate::server::wamp::MAX_MESSAGE)
358 .build()
359 }
360 fn example_prompt(&self) -> &'static str {
361 "Join realm1 on the WAMP router at 127.0.0.1:8080 and call com.example.add2 with 2 and 3"
362 }
363 fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
364 let llm = json!({"type":"open_client","protocol":"wamp","remote_addr":"127.0.0.1:8080","instruction":"Call com.example.add2 with 2 and 3","startup_params":{"realm":"realm1"}});
365 let mut static_example = llm.clone();
366 static_example["event_handlers"] = json!([
367 {"event_pattern":"wamp_welcome","handler":{"type":"static","actions":[call().example]}},
368 {"event_pattern":"wamp_invocation","handler":{"type":"static","actions":[{"type":"wamp_error","error":"wamp.error.no_such_procedure"}]}},
369 {"event_pattern":"*","handler":{"type":"static","actions":[]}}
370 ]);
371 let mut scripted = static_example.clone();
372 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':'wamp_yield','invocation':e['invocation'],'args':e['args']}]}))"});
373 crate::llm::actions::StartupExamples::new(llm, scripted, static_example)
374 }
375 fn group_name(&self) -> &'static str {
376 "Application"
377 }
378}
379
380fn payload_ok(v: &Value) -> Result<()> {
381 if let Some(a) = v.get("args").filter(|a| !a.is_null()) {
382 ensure!(a.is_array(), "args is an array");
383 }
384 if let Some(k) = v.get("kwargs").filter(|k| !k.is_null()) {
385 ensure!(k.is_object(), "kwargs is an object");
386 }
387 Ok(())
388}
389
390impl Client for WampClientProtocol {
391 fn connect(
392 &self,
393 ctx: ConnectContext,
394 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
395 {
396 Box::pin(super::connect(ctx))
397 }
398 fn execute_action(&self, v: Value) -> Result<ClientActionResult> {
399 ensure!(
400 crate::utils::json_budget::within_budget(&v, 1024 * 1024, 100_000, 32),
401 "action exceeds the WAMP bounds"
402 );
403 let s = |k: &str, empty_ok: bool| v[k].as_str().is_some_and(|u| uri_ok(u, empty_ok));
404 match v["type"].as_str() {
405 Some("wamp_subscribe") => {
406 let policy = v["match"].as_str().unwrap_or("exact");
407 ensure!(
408 matches!(policy, "exact" | "prefix" | "wildcard"),
409 "match is exact, prefix or wildcard"
410 );
411 ensure!(s("topic", policy == "wildcard"), "topic is a URI");
412 }
413 Some("wamp_unsubscribe") => ensure!(
414 v["topic"].as_str().is_some_and(|t| !t.is_empty()),
415 "topic is required"
416 ),
417 Some("wamp_publish") => {
418 ensure!(s("topic", false), "topic is a URI");
419 payload_ok(&v)?;
420 }
421 Some("wamp_call") => {
422 ensure!(s("procedure", false), "procedure is a URI");
423 payload_ok(&v)?;
424 }
425 Some("wamp_register" | "wamp_unregister") => {
426 ensure!(s("procedure", false), "procedure is a URI")
427 }
428 Some("wamp_yield") => payload_ok(&v)?,
429 Some("wamp_error") => {
430 ensure!(s("error", false), "error is an error URI");
431 payload_ok(&v)?;
432 }
433 Some("wamp_goodbye") => {}
434 Some("disconnect") => return Ok(ClientActionResult::Disconnect),
435 _ => bail!("Unknown WAMP client action"),
436 }
437 Ok(ClientActionResult::Custom {
438 name: v["type"].as_str().unwrap_or_default().into(),
439 data: v,
440 })
441 }
442}