1use super::uri;
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, Result};
10use serde_json::{json, Value};
11use std::sync::LazyLock;
12
13#[derive(Default)]
14pub struct WampProtocol;
15impl WampProtocol {
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!("-> WAMP {name}"))),
42 }
43}
44
45pub fn payload_params() -> Vec<Parameter> {
46 vec![
47 parameter(
48 "args",
49 "array",
50 "Positional arguments (any JSON values)",
51 false,
52 ),
53 parameter(
54 "kwargs",
55 "object",
56 "Keyword arguments (a JSON object)",
57 false,
58 ),
59 ]
60}
61
62fn welcome() -> ActionDefinition {
63 action(
64 "wamp_welcome",
65 "Admit the session to the realm; Rust sends WELCOME with a new session ID and the broker and dealer roles",
66 vec![parameter("authrole", "string", "Role to announce in WELCOME (default anonymous)", false)],
67 json!({"type": "wamp_welcome", "authrole": "user"}),
68 )
69}
70fn abort() -> ActionDefinition {
71 action(
72 "wamp_abort",
73 "Refuse the session with ABORT and close",
74 vec![
75 parameter(
76 "reason",
77 "string",
78 "Error URI, e.g. wamp.error.no_such_realm or wamp.error.not_authorized",
79 true,
80 ),
81 parameter(
82 "message",
83 "string",
84 "Human-readable message, up to 256 characters",
85 false,
86 ),
87 ],
88 json!({"type": "wamp_abort", "reason": "wamp.error.no_such_realm", "message": "this router serves realm1 only"}),
89 )
90}
91pub fn result() -> ActionDefinition {
92 action(
93 "wamp_result",
94 "Answer the call with RESULT carrying these arguments",
95 payload_params(),
96 json!({"type": "wamp_result", "args": [5], "kwargs": {}}),
97 )
98}
99pub fn error() -> ActionDefinition {
100 let mut p = vec![parameter(
101 "error",
102 "string",
103 "Error URI, e.g. com.example.error.invalid_input or wamp.error.no_such_procedure",
104 true,
105 )];
106 p.extend(payload_params());
107 action(
108 "wamp_error",
109 "Answer the call with ERROR",
110 p,
111 json!({"type": "wamp_error", "error": "com.example.error.invalid_input", "args": ["a and b must be numbers"]}),
112 )
113}
114pub fn publish() -> ActionDefinition {
115 let mut p = vec![parameter(
116 "topic",
117 "string",
118 "Topic URI to publish to in this session's realm",
119 true,
120 )];
121 p.extend(payload_params());
122 action("wamp_publish", "Publish an event from the router to every session subscribed to a matching topic in this session's realm", p, json!({"type": "wamp_publish", "topic": "com.example.news", "args": ["router says hello"]}))
123}
124
125pub static HELLO_EVENT: LazyLock<EventType> = LazyLock::new(|| {
126 EventType::new("wamp_hello", "A client asks to join a realm (HELLO). The WebSocket subprotocol and message shape were checked by Rust.", welcome().example.clone())
127 .with_parameters(vec![
128 parameter("realm", "string", "Realm URI requested", true),
129 parameter("roles", "array", "Roles the client announced: caller, callee, publisher, subscriber", true),
130 parameter("authid", "string", "authid the client offered, when any", false),
131 parameter("authmethods", "array", "Authentication methods offered (only anonymous is supported)", false),
132 ])
133 .with_actions(vec![welcome(), abort()])
134});
135pub static CALL_EVENT: LazyLock<EventType> = LazyLock::new(|| {
136 EventType::new("wamp_call", "A CALL to a procedure no session in the realm has registered; the router itself answers it", result().example.clone())
137 .with_parameters(vec![
138 parameter("realm", "string", "The caller's realm", true),
139 parameter("procedure", "string", "Procedure URI called", true),
140 parameter("args", "array", "Positional arguments", true),
141 parameter("kwargs", "object", "Keyword arguments", true),
142 parameter("caller_session", "number", "The caller's WAMP session ID", true),
143 ])
144 .with_actions(vec![result(), error()])
145});
146
147fn payload_ok(v: &Value) -> Result<()> {
148 if let Some(a) = v.get("args").filter(|a| !a.is_null()) {
149 ensure!(a.is_array(), "args is an array");
150 }
151 if let Some(k) = v.get("kwargs").filter(|k| !k.is_null()) {
152 ensure!(k.is_object(), "kwargs is an object");
153 }
154 Ok(())
155}
156
157pub fn check_answer(v: &Value) -> Result<()> {
158 ensure!(
159 crate::utils::json_budget::within_budget(v, 1024 * 1024, 100_000, 32),
160 "answer exceeds the WAMP bounds"
161 );
162 match v["type"].as_str() {
163 Some("wamp_welcome") => {
164 if let Some(r) = v.get("authrole").filter(|r| !r.is_null()) {
165 ensure!(
166 r.as_str().is_some_and(|r| uri::uri_ok(r, false)),
167 "authrole is a URI-like name"
168 );
169 }
170 }
171 Some("wamp_abort") => {
172 ensure!(
173 v["reason"].as_str().is_some_and(|r| uri::uri_ok(r, false)),
174 "reason is an error URI"
175 );
176 if let Some(m) = v.get("message").filter(|m| !m.is_null()) {
177 ensure!(
178 m.as_str().is_some_and(
179 |m| m.len() <= 256 && !crate::utils::sanitize::has_controls(&m)
180 ),
181 "message is up to 256 printable characters"
182 );
183 }
184 }
185 Some("wamp_result") => payload_ok(v)?,
186 Some("wamp_error") => {
187 ensure!(
188 v["error"].as_str().is_some_and(|e| uri::uri_ok(e, false)),
189 "error is an error URI"
190 );
191 payload_ok(v)?;
192 }
193 Some("wamp_publish") => {
194 ensure!(
195 v["topic"].as_str().is_some_and(|t| uri::uri_ok(t, false)),
196 "topic is a URI"
197 );
198 payload_ok(v)?;
199 }
200 _ => bail!("Unknown WAMP server action"),
201 }
202 Ok(())
203}
204
205fn startup(
206 name: &str,
207 kind: &str,
208 description: &str,
209 example: Value,
210 default: Option<Value>,
211) -> ParameterDefinition {
212 ParameterDefinition {
213 name: name.into(),
214 type_hint: kind.into(),
215 description: description.into(),
216 required: false,
217 example,
218 default,
219 }
220}
221
222impl Protocol for WampProtocol {
223 fn protocol_name(&self) -> &'static str {
224 "WAMP"
225 }
226 fn stack_name(&self) -> &'static str {
227 "ETH>IP>TCP>HTTP>WebSocket>WAMP"
228 }
229 fn description(&self) -> &'static str {
230 "WAMP v2 router (basic profile, JSON over WebSocket): realms, routed RPC and pub/sub; the handler admits sessions and answers calls nobody registered"
231 }
232 fn keywords(&self) -> Vec<&'static str> {
233 vec![
234 "wamp",
235 "web application messaging protocol",
236 "router",
237 "rpc",
238 "pubsub",
239 "autobahn",
240 "crossbar",
241 ]
242 }
243 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
244 vec![publish()]
245 }
246 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
247 vec![welcome(), abort(), result(), error()]
248 }
249 fn get_event_types(&self) -> Vec<EventType> {
250 vec![HELLO_EVENT.clone(), CALL_EVENT.clone()]
251 }
252 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
253 vec![
254 startup(
255 "hello_timeout_secs",
256 "integer",
257 "Seconds a new WebSocket has to send HELLO",
258 json!(5),
259 Some(json!(super::HELLO_TIMEOUT.as_secs())),
260 ),
261 startup(
262 "max_message_bytes",
263 "integer",
264 "Largest WAMP message accepted",
265 json!(65536),
266 Some(json!(super::MAX_MESSAGE)),
267 ),
268 ]
269 }
270 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
271 use crate::protocol::metadata::*;
272 ProtocolMetadataV2::builder()
273 .state(DevelopmentState::Experimental)
274 .privilege_requirement(PrivilegeRequirement::None)
275 .implementation("tokio-tungstenite WebSocket with the wamp.2.json subprotocol; hand-written WAMP v2 basic-profile router: realms, sessions, exact/prefix/wildcard subscriptions, publisher exclusion and black/white listing, single-callee registrations, call routing with INVOCATION/YIELD/ERROR")
276 .llm_control("Which sessions join which realm, and the answer to every call no client has registered")
277 .e2e_testing("tests/server/wamp: autobahn-python 24.4.2 and nexus 3.3.0 (Go; independent) join, register, subscribe, call through the router, publish with acknowledgement, call router procedures, get router errors and are refused a realm")
278 .notes("JSON serializer only (no MessagePack, CBOR or RawSocket); anonymous authentication only; no progressive results, call cancellation, shared registrations, meta API or event history.")
279 .answers_on_failure()
280 .max_inbound_bytes(super::MAX_MESSAGE)
281 .build()
282 }
283 fn example_prompt(&self) -> &'static str {
284 "WAMP router on port 8080 for realm1 whose com.example.time procedure returns the current time"
285 }
286 fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
287 let llm = json!({"type":"open_server","base_stack":"wamp","port":8080,"instruction":"Admit realm1 only; answer com.example.time with the current UTC time"});
288 let mut static_example = llm.clone();
289 static_example["event_handlers"] = json!([
290 {"event_pattern":"wamp_hello","handler":{"type":"static","actions":[{"type":"wamp_welcome"}]}},
291 {"event_pattern":"wamp_call","handler":{"type":"static","actions":[{"type":"wamp_error","error":"wamp.error.no_such_procedure"}]}}
292 ]);
293 let mut scripted = static_example.clone();
294 scripted["event_handlers"][1]["handler"] = json!({"type":"script","language":"python","code":"import json,sys\ne=json.load(sys.stdin)['event']\nif e['procedure']=='com.example.add':\n print(json.dumps({'actions':[{'type':'wamp_result','args':[sum(e['args'])]}]}))\nelse:\n print(json.dumps({'actions':[{'type':'wamp_error','error':'wamp.error.no_such_procedure'}]}))"});
295 crate::llm::actions::StartupExamples::new(llm, scripted, static_example)
296 }
297 fn group_name(&self) -> &'static str {
298 "Application"
299 }
300}
301
302impl Server for WampProtocol {
303 fn spawn(
304 &self,
305 ctx: SpawnContext,
306 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
307 {
308 Box::pin(super::spawn(ctx))
309 }
310 fn execute_action(&self, v: Value) -> Result<ActionResult> {
311 check_answer(&v)?;
312 Ok(ActionResult::Custom {
313 name: v["type"].as_str().unwrap_or_default().into(),
314 data: v,
315 })
316 }
317}