agentmux_srv/backend/
eventbus.rs

1// Copyright 2025, Command Line Inc.
2// Copyright 2025-2026, AgentMux Corp.
3// SPDX-License-Identifier: Apache-2.0
4
5//! Event bus: WebSocket event dispatching to connected clients.
6//! Port of Go's pkg/eventbus/eventbus.go.
7
8
9use std::collections::HashMap;
10use std::sync::{Arc, Mutex};
11
12use serde::{Deserialize, Serialize};
13
14use super::mps::{MuxEvent, WpsClient, EVENT_SYS_INFO, EVENT_BLOCK_STATS};
15
16// ---- Event type constants ----
17
18pub const WS_EVENT_RPC: &str = "rpc";
19
20/// One WS frame carrying an ARRAY of `MuxObjUpdate`s from a single atomic
21/// backend transition. Mirrored in `frontend/app/store/mps-events.ts`
22/// (`WpsEvent.MuxObjBatchedUpdates`) — the frontend applies the whole array
23/// in one Solid `batch()` flush. See `broadcast_mux_obj_updates` below.
24pub const WS_EVENT_MUX_OBJ_BATCHED_UPDATES: &str = "waveobj:batchedupdates";
25
26/// Egress priority lane for a server→client event.
27///
28/// `Background` is reserved for droppable perf telemetry (sysinfo + per-block
29/// stats) that must never delay interactive terminal I/O; everything else is
30/// `Priority`. The WebSocket egress loop drains the priority lane before the
31/// background lane (see `server/websocket.rs` and
32/// `docs/specs/SPEC_TERMINAL_INPUT_PRIORITY_OVER_SYSINFO_2026_06_16.md`).
33#[derive(Clone, Copy, Debug, PartialEq, Eq)]
34pub enum Lane {
35    Priority,
36    Background,
37}
38
39/// Egress channel capacities. Bounded (not unbounded) so a stalled/hidden-
40/// renderer WS consumer can't grow sidecar commit without limit — a healthy
41/// connection drains far below these depths; hitting the cap is itself the
42/// signal that the consumer isn't keeping up. See
43/// docs/specs/SPEC_MEMORY_COMMIT_ATTRIBUTION_CORRECTION_2026_07_02.md §B.2.
44///
45/// Priority carries terminal echo + interactive events, which must not be
46/// silently dropped in normal operation — sized generously so legitimate
47/// bursts (e.g. `cat` of a large file) never hit the cap.
48const PRIORITY_LANE_CAPACITY: usize = 8192;
49/// Background carries droppable perf telemetry (sysinfo ~1/s, per-block
50/// stats) already coalesced-to-latest by the receive loop — a healthy
51/// consumer never queues more than a couple of entries.
52const BACKGROUND_LANE_CAPACITY: usize = 256;
53
54/// The pair of receivers handed to a WebSocket connection on registration.
55/// Terminal echo + interactive events arrive on `priority`; perf telemetry on
56/// `background`.
57pub struct WsReceivers {
58    pub priority: tokio::sync::mpsc::Receiver<serde_json::Value>,
59    pub background: tokio::sync::mpsc::Receiver<serde_json::Value>,
60}
61
62// ---- Types ----
63
64#[derive(Debug, Clone, Serialize, Deserialize)]
65pub struct WSEventType {
66    pub eventtype: String,
67    #[serde(skip_serializing_if = "String::is_empty", default)]
68    pub oref: String,
69    #[serde(skip_serializing_if = "Option::is_none")]
70    pub data: Option<serde_json::Value>,
71}
72
73struct WindowWatchData {
74    /// Interactive lane: terminal echo, RPC-routed wave events, obj updates.
75    priority: tokio::sync::mpsc::Sender<serde_json::Value>,
76    /// Background lane: droppable perf telemetry (sysinfo, blockstats).
77    background: tokio::sync::mpsc::Sender<serde_json::Value>,
78    #[allow(dead_code)]
79    tab_id: String,
80}
81
82impl WindowWatchData {
83    fn sender(&self, lane: Lane) -> &tokio::sync::mpsc::Sender<serde_json::Value> {
84        match lane {
85            Lane::Priority => &self.priority,
86            Lane::Background => &self.background,
87        }
88    }
89}
90
91/// Global event bus for dispatching WebSocket events to connected clients.
92pub struct EventBus {
93    watches: Mutex<HashMap<String, WindowWatchData>>,
94}
95
96impl EventBus {
97    pub fn new() -> Self {
98        Self {
99            watches: Mutex::new(HashMap::new()),
100        }
101    }
102
103    /// Register a WebSocket connection for receiving events.
104    /// Returns the priority + background receiver pair for the connection.
105    pub fn register_ws(&self, conn_id: &str, tab_id: &str) -> WsReceivers {
106        let (priority_tx, priority_rx) = tokio::sync::mpsc::channel(PRIORITY_LANE_CAPACITY);
107        let (background_tx, background_rx) = tokio::sync::mpsc::channel(BACKGROUND_LANE_CAPACITY);
108        let mut watches = self.watches.lock().unwrap();
109        watches.insert(
110            conn_id.to_string(),
111            WindowWatchData {
112                priority: priority_tx,
113                background: background_tx,
114                tab_id: tab_id.to_string(),
115            },
116        );
117        WsReceivers {
118            priority: priority_rx,
119            background: background_rx,
120        }
121    }
122
123    /// Unregister a WebSocket connection.
124    pub fn unregister_ws(&self, conn_id: &str) {
125        let mut watches = self.watches.lock().unwrap();
126        watches.remove(conn_id);
127    }
128
129    /// Check if any connections exist for a given window/tab ID.
130    #[allow(dead_code)]
131    pub fn has_connections_for(&self, tab_id: &str) -> bool {
132        let watches = self.watches.lock().unwrap();
133        watches.values().any(|w| w.tab_id == tab_id)
134    }
135
136    /// Wait for a connection to appear for the given tab_id (with timeout).
137    #[allow(dead_code)]
138    pub async fn wait_for_connection(
139        &self,
140        tab_id: &str,
141        timeout: std::time::Duration,
142    ) -> bool {
143        let deadline = tokio::time::Instant::now() + timeout;
144        loop {
145            if self.has_connections_for(tab_id) {
146                return true;
147            }
148            if tokio::time::Instant::now() >= deadline {
149                return false;
150            }
151            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
152        }
153    }
154
155    /// Send an event to a single connection by conn_id on a specific lane.
156    /// No-op if not found.
157    #[cfg(test)]
158    pub fn send_to_conn_lane(&self, conn_id: &str, event: &WSEventType, lane: Lane) {
159        let data = match serde_json::to_value(event) {
160            Ok(v) => v,
161            Err(e) => {
162                tracing::error!("cannot marshal event: {}", e);
163                return;
164            }
165        };
166        self.send_value_to_conn_lane(conn_id, data, lane);
167    }
168
169    /// [`Self::send_to_conn_lane`] for an event already built as the wire
170    /// `Value` — for callers that would otherwise serialize a payload into a
171    /// `WSEventType` and have it serialized a second time here.
172    pub fn send_value_to_conn_lane(&self, conn_id: &str, data: serde_json::Value, lane: Lane) {
173        let watches = self.watches.lock().unwrap();
174        if let Some(watch) = watches.get(conn_id) {
175            Self::try_send_lane(conn_id, watch.sender(lane), data);
176        }
177    }
178
179    /// Broadcast an event to all connected WebSocket clients, on the priority lane.
180    pub fn broadcast_event(&self, event: &WSEventType) {
181        self.broadcast_event_lane(event, Lane::Priority);
182    }
183
184    /// A typed event as a raw event-bus frame: `eventtype` is its name and `oref` its scope, for
185    /// a caller that sends through `broadcast_event` or a lane rather than the broker.
186    /// `websocket.rs::forward_event` wraps it as the `eventrecv` frame the renderer reads.
187    pub fn typed_ws_event<E: super::mps_events::Event>(payload: &E::Payload, oref: impl Into<String>) -> WSEventType {
188        // typed-helper: builds the frame from a declared event and its payload type.
189        WSEventType {
190            eventtype: E::NAME.to_string(),
191            oref: oref.into(),
192            data: super::mps_events::payload_data(payload),
193        }
194    }
195
196    /// A typed event as the frame srv writes to one WebSocket itself (`srvinfo` and `config` on
197    /// connect): the same `rpc`/`eventrecv` frame a broadcast sends.
198    pub fn typed_frame<E: super::mps_events::Event>(payload: &E::Payload) -> serde_json::Value {
199        serde_json::json!({
200            "eventtype": WS_EVENT_RPC,
201            // typed-helper: builds the frame from a declared event and its payload type.
202            "data": { "command": "eventrecv", "data": { "event": E::NAME, "data": super::mps_events::payload_data(payload) } },
203        })
204    }
205
206    /// Broadcast a typed event (`mps_events`) to every connection directly, not through the
207    /// broker: no scopes, nothing persisted. The same `rpc`/`eventrecv` frame a broker event
208    /// gets (a serialized `MuxEvent` omits its empty scopes, sender and persist).
209    pub fn broadcast_typed<E: super::mps_events::Event>(&self, payload: &E::Payload) {
210        // typed-helper: builds the event from a declared event and its payload type.
211        let event = MuxEvent {
212            event: E::NAME.to_string(),
213            scopes: vec![],
214            sender: String::new(),
215            persist: 0,
216            data: super::mps_events::payload_data(payload),
217        };
218        self.broadcast_event(&WSEventType {
219            eventtype: WS_EVENT_RPC.to_string(),
220            oref: String::new(),
221            // typed-helper: the frame around the typed `event` above.
222            data: Some(serde_json::json!({ "command": "eventrecv", "data": event })),
223        });
224    }
225
226    /// Broadcast a set of related `MuxObjUpdate`s from ONE atomic backend
227    /// transition as ONE `waveobj:batchedupdates` WS frame, so the frontend
228    /// applies them in a single reactive flush (`updateMuxObjects`'s
229    /// `batch()`), preserving the array's order.
230    ///
231    /// Why one frame instead of N `waveobj:update` frames: each frame is
232    /// dispatched (and painted) independently by the renderer, so a
233    /// `[delete tab, update workspace]` pair sent as two frames blanks the
234    /// still-mounted tab a full paint before the workspace update unmounts
235    /// it — the same flash class
236    /// `SPEC_TAB_CLOSE_BUTTON_SELECT_FLASH_2026_08_25.md` §6 fixed at the
237    /// RPC-response layer, riding in unbatched through the WS push path
238    /// instead (§7). Single-update callers (blockcontroller, setmeta, the
239    /// wave-obj bridge) keep emitting plain `waveobj:update` — there is
240    /// nothing to batch there.
241    pub fn broadcast_mux_obj_updates(&self, updates: &[super::obj::MuxObjUpdate]) {
242        if updates.is_empty() {
243            return;
244        }
245        let batch = super::obj::MuxObjBatch(updates.to_vec());
246        self.broadcast_event(&Self::typed_ws_event::<super::mps_events::MuxObjBatchedUpdates>(&batch, String::new()));
247    }
248
249    /// Broadcast an event to all connected WebSocket clients on a specific lane.
250    pub fn broadcast_event_lane(&self, event: &WSEventType, lane: Lane) {
251        let data = match serde_json::to_value(event) {
252            Ok(v) => v,
253            Err(e) => {
254                tracing::error!("cannot marshal event: {}", e);
255                return;
256            }
257        };
258        self.broadcast_value_lane(data, lane);
259    }
260
261    /// [`Self::broadcast_event_lane`] for an already-built wire `Value`.
262    pub fn broadcast_value_lane(&self, data: serde_json::Value, lane: Lane) {
263        let watches = self.watches.lock().unwrap();
264        for (conn_id, watch) in watches.iter() {
265            Self::try_send_lane(conn_id, watch.sender(lane), data.clone());
266        }
267    }
268
269    /// Non-blocking send with distinct logging for a genuinely-full lane
270    /// (stalled/hidden-renderer consumer — see `PRIORITY_LANE_CAPACITY` /
271    /// `BACKGROUND_LANE_CAPACITY`) vs. a closed one (connection already
272    /// gone). Never blocks the caller — callers hold the `watches` lock.
273    fn try_send_lane(
274        conn_id: &str,
275        sender: &tokio::sync::mpsc::Sender<serde_json::Value>,
276        data: serde_json::Value,
277    ) {
278        match sender.try_send(data) {
279            Ok(()) => {}
280            Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
281                tracing::warn!(
282                    "ws egress lane full for conn {} — consumer stalled, dropping event",
283                    conn_id
284                );
285            }
286            Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
287                tracing::warn!("failed to send event to conn {}", conn_id);
288            }
289        }
290    }
291
292    /// Send an event to connections matching a specific tab_id.
293    #[allow(dead_code)]
294    pub fn send_to_tab(&self, tab_id: &str, event: &WSEventType) {
295        let data = match serde_json::to_value(event) {
296            Ok(v) => v,
297            Err(e) => {
298                tracing::error!("cannot marshal event: {}", e);
299                return;
300            }
301        };
302        let watches = self.watches.lock().unwrap();
303        for (conn_id, watch) in watches.iter() {
304            if watch.tab_id == tab_id {
305                Self::try_send_lane(conn_id, watch.sender(Lane::Priority), data.clone());
306            }
307        }
308    }
309
310    /// Get the number of active connections.
311    pub fn connection_count(&self) -> usize {
312        self.watches.lock().unwrap().len()
313    }
314}
315
316impl Default for EventBus {
317    fn default() -> Self {
318        Self::new()
319    }
320}
321
322/// Bridge from MPS Broker to EventBus.
323/// Wraps MuxEvents as RPC eventrecv messages and broadcasts them to all WS clients.
324pub struct EventBusBridge {
325    event_bus: Arc<EventBus>,
326}
327
328impl EventBusBridge {
329    pub fn new(event_bus: Arc<EventBus>) -> Self {
330        Self { event_bus }
331    }
332}
333
334impl WpsClient for EventBusBridge {
335    fn send_event(&self, route_id: &str, event: MuxEvent) {
336        // Perf telemetry is droppable and must never delay interactive terminal
337        // I/O, so route sysinfo + per-block stats to the background lane and
338        // everything else to the priority lane. This is the only place the raw
339        // MuxEvent type is visible before it's wrapped as an opaque RPC
340        // envelope. See SPEC_TERMINAL_INPUT_PRIORITY_OVER_SYSINFO_2026_06_16.
341        let lane = match event.event.as_str() {
342            EVENT_SYS_INFO | EVENT_BLOCK_STATS => Lane::Background,
343            _ => Lane::Priority,
344        };
345        // Wrap as RPC eventrecv message (format expected by frontend).
346        //
347        // Serialized exactly once. This runs under the MPS broker's lock, on
348        // every publish, and the payload can be a large transcript chunk;
349        // building a `WSEventType` and handing it to `send_to_conn_lane`
350        // serialized the payload into a `Value` and then deep-copied that
351        // `Value` again. Same wire shape as `WSEventType` (empty `oref`
352        // skipped) — pinned by `rpc_eventrecv_value_matches_wsevent_shape`.
353        let Some(data) = rpc_eventrecv_value(&event) else {
354            return;
355        };
356        // Route to the specific connection that subscribed. Broadcast is used
357        // only for legacy callers that pass "ws-main" (none remain after the
358        // per-conn-id fix), so this always takes the targeted path in practice.
359        if route_id == "ws-main" {
360            self.event_bus.broadcast_value_lane(data, lane);
361        } else {
362            self.event_bus.send_value_to_conn_lane(route_id, data, lane);
363        }
364    }
365}
366
367/// The `rpc`/`eventrecv` WS envelope for `event`, built with a single
368/// serialization of the payload. See `EventBusBridge::send_event`.
369fn rpc_eventrecv_value(event: &MuxEvent) -> Option<serde_json::Value> {
370    let payload = match serde_json::to_value(event) {
371        Ok(v) => v,
372        Err(e) => {
373            tracing::error!("cannot marshal event: {}", e);
374            return None;
375        }
376    };
377    let mut inner = serde_json::Map::new();
378    inner.insert("command".to_string(), serde_json::Value::String("eventrecv".to_string()));
379    inner.insert("data".to_string(), payload);
380    let mut outer = serde_json::Map::new();
381    outer.insert("eventtype".to_string(), serde_json::Value::String(WS_EVENT_RPC.to_string()));
382    outer.insert("data".to_string(), serde_json::Value::Object(inner));
383    Some(serde_json::Value::Object(outer))
384}
385
386// ====================================================================
387// Tests
388// ====================================================================
389
390#[cfg(test)]
391mod tests {
392    use super::*;
393    use crate::backend::mps::EVENT_BLOCK_FILE;
394
395    #[test]
396    fn test_register_unregister() {
397        let bus = EventBus::new();
398        let _rx = bus.register_ws("conn-1", "tab-1");
399        assert_eq!(bus.connection_count(), 1);
400        assert!(bus.has_connections_for("tab-1"));
401        assert!(!bus.has_connections_for("tab-2"));
402
403        bus.unregister_ws("conn-1");
404        assert_eq!(bus.connection_count(), 0);
405        assert!(!bus.has_connections_for("tab-1"));
406    }
407
408    #[test]
409    fn test_broadcast_event() {
410        let bus = EventBus::new();
411        let mut rx1 = bus.register_ws("conn-1", "tab-1");
412        let mut rx2 = bus.register_ws("conn-2", "tab-2");
413
414        let event = WSEventType {
415            eventtype: WS_EVENT_RPC.to_string(),
416            oref: String::new(),
417            data: Some(serde_json::json!({"test": true})),
418        };
419        bus.broadcast_event(&event);
420
421        // Default broadcast lands on the priority lane.
422        assert!(rx1.priority.try_recv().is_ok());
423        assert!(rx2.priority.try_recv().is_ok());
424        assert!(rx1.background.try_recv().is_err());
425    }
426
427    #[test]
428    fn test_send_to_tab() {
429        let bus = EventBus::new();
430        let mut rx1 = bus.register_ws("conn-1", "tab-1");
431        let mut rx2 = bus.register_ws("conn-2", "tab-2");
432
433        let event = WSEventType {
434            eventtype: WS_EVENT_RPC.to_string(),
435            oref: String::new(),
436            data: None,
437        };
438        bus.send_to_tab("tab-1", &event);
439
440        assert!(rx1.priority.try_recv().is_ok());
441        assert!(rx2.priority.try_recv().is_err()); // tab-2 should not receive
442    }
443
444    /// A multi-object atomic transition must go out as ONE
445    /// `waveobj:batchedupdates` frame (array payload, order preserved), not
446    /// N frames — N frames repaint the renderer in N unbatched steps (the
447    /// SPEC_TAB_CLOSE_BUTTON_SELECT_FLASH_2026_08_25.md §7 flash).
448    #[test]
449    fn test_broadcast_mux_obj_updates_single_batched_frame() {
450        use crate::backend::obj::MuxObjUpdate;
451
452        let bus = EventBus::new();
453        let mut rx = bus.register_ws("conn-1", "tab-1");
454
455        let updates = vec![
456            MuxObjUpdate {
457                updatetype: "update".into(),
458                otype: "workspace".into(),
459                oid: "ws-1".into(),
460                obj: Some(serde_json::json!({"oid": "ws-1"})),
461            },
462            MuxObjUpdate {
463                updatetype: "delete".into(),
464                otype: "tab".into(),
465                oid: "tab-9".into(),
466                obj: None,
467            },
468        ];
469        bus.broadcast_mux_obj_updates(&updates);
470
471        let msg = rx.priority.try_recv().expect("one frame expected");
472        assert_eq!(
473            msg.get("eventtype").and_then(|v| v.as_str()),
474            Some(WS_EVENT_MUX_OBJ_BATCHED_UPDATES),
475        );
476        let arr = msg.get("data").and_then(|d| d.as_array()).expect("array payload");
477        assert_eq!(arr.len(), 2);
478        assert_eq!(arr[0].get("otype").and_then(|v| v.as_str()), Some("workspace"));
479        assert_eq!(arr[1].get("updatetype").and_then(|v| v.as_str()), Some("delete"));
480        // Exactly one frame — nothing else queued.
481        assert!(rx.priority.try_recv().is_err());
482
483        // Empty slice → no frame at all.
484        bus.broadcast_mux_obj_updates(&[]);
485        assert!(rx.priority.try_recv().is_err());
486    }
487
488    #[test]
489    fn test_lane_separation() {
490        // A background-lane send must not land on the priority lane, and vice
491        // versa — this is the core of "terminal typing has complete priority
492        // over perf monitoring".
493        let bus = EventBus::new();
494        let mut rx = bus.register_ws("conn-1", "tab-1");
495
496        let event = WSEventType {
497            eventtype: WS_EVENT_RPC.to_string(),
498            oref: String::new(),
499            data: None,
500        };
501        bus.send_to_conn_lane("conn-1", &event, Lane::Background);
502        assert!(rx.priority.try_recv().is_err()); // nothing on priority
503        assert!(rx.background.try_recv().is_ok()); // telemetry on background
504
505        bus.send_to_conn_lane("conn-1", &event, Lane::Priority);
506        assert!(rx.background.try_recv().is_err()); // nothing on background
507        assert!(rx.priority.try_recv().is_ok()); // interactive on priority
508    }
509
510    #[test]
511    fn test_stalled_consumer_drops_instead_of_growing_unbounded() {
512        // A consumer that never drains must not let the channel grow past its
513        // bound — this is the memory-safety property B.2 exists for. Fill the
514        // background lane (small capacity) past its cap without ever calling
515        // recv(), then drain and count: the received count must be bounded
516        // by the channel capacity, proving excess sends were dropped
517        // (try_send Full) rather than queued without limit.
518        let bus = EventBus::new();
519        let mut rx = bus.register_ws("conn-1", "tab-1"); // never drained during the flood
520        let event = WSEventType {
521            eventtype: WS_EVENT_RPC.to_string(),
522            oref: String::new(),
523            data: None,
524        };
525        let sent = BACKGROUND_LANE_CAPACITY * 4;
526        for _ in 0..sent {
527            bus.send_to_conn_lane("conn-1", &event, Lane::Background);
528        }
529        let mut received = 0;
530        while rx.background.try_recv().is_ok() {
531            received += 1;
532        }
533        assert!(
534            received <= BACKGROUND_LANE_CAPACITY,
535            "received {received} events from a lane capped at {BACKGROUND_LANE_CAPACITY} — sends were not bounded"
536        );
537        assert!(received < sent, "no events were dropped despite flooding well past capacity");
538    }
539
540    #[test]
541    fn test_bridge_routes_telemetry_to_background() {
542        // EventBusBridge must demote sysinfo + blockstats to the background lane
543        // and keep everything else (e.g. terminal blockfile output) on priority.
544        let bus = Arc::new(EventBus::new());
545        let mut rx = bus.register_ws("conn-1", "tab-1");
546        let bridge = EventBusBridge::new(bus.clone());
547
548        let telemetry = |event: &str| MuxEvent {
549            event: event.to_string(),
550            scopes: vec![],
551            sender: String::new(),
552            persist: 0,
553            data: None,
554        };
555
556        bridge.send_event("conn-1", telemetry(EVENT_SYS_INFO));
557        bridge.send_event("conn-1", telemetry(EVENT_BLOCK_STATS));
558        assert!(rx.priority.try_recv().is_err()); // no telemetry on priority
559        assert!(rx.background.try_recv().is_ok()); // sysinfo
560        assert!(rx.background.try_recv().is_ok()); // blockstats
561
562        // Terminal output (blockfile) stays on the interactive priority lane.
563        bridge.send_event("conn-1", telemetry(EVENT_BLOCK_FILE));
564        assert!(rx.priority.try_recv().is_ok());
565        assert!(rx.background.try_recv().is_err());
566    }
567
568    #[test]
569    fn test_ws_event_serialization() {
570        let event = WSEventType {
571            eventtype: "test".to_string(),
572            oref: String::new(),
573            data: Some(serde_json::json!(42)),
574        };
575        let json = serde_json::to_string(&event).unwrap();
576        assert!(json.contains("\"eventtype\":\"test\""));
577        // Empty oref should be omitted
578        assert!(!json.contains("\"oref\""));
579    }
580
581    /// `send_event` now builds the wire value directly instead of going through
582    /// `WSEventType`; the frontend must see byte-for-byte the same shape.
583    #[test]
584    fn rpc_eventrecv_value_matches_wsevent_shape() {
585        let event = MuxEvent {
586            event: "blockfile".to_string(),
587            scopes: vec!["block:abc".to_string()],
588            sender: String::new(),
589            persist: 0,
590            data: Some(serde_json::json!({ "data64": "aGVsbG8=", "n": 3 })),
591        };
592        let old = serde_json::to_value(WSEventType {
593            eventtype: WS_EVENT_RPC.to_string(),
594            oref: String::new(),
595            data: Some(serde_json::json!({ "command": "eventrecv", "data": event })),
596        })
597        .unwrap();
598        assert_eq!(rpc_eventrecv_value(&event).unwrap(), old);
599    }
600
601    #[test]
602    fn send_event_delivers_the_eventrecv_envelope_to_the_subscribed_conn() {
603        let bus = Arc::new(EventBus::new());
604        let mut rx = bus.register_ws("conn-1", "tab-1");
605        let bridge = EventBusBridge::new(Arc::clone(&bus));
606        let event = MuxEvent {
607            event: "blockfile".to_string(),
608            scopes: vec![],
609            sender: String::new(),
610            persist: 0,
611            data: Some(serde_json::json!({ "k": "v" })),
612        };
613        bridge.send_event("conn-1", event.clone());
614        let got = rx.priority.try_recv().expect("delivered on the priority lane");
615        assert_eq!(got, rpc_eventrecv_value(&event).unwrap());
616        assert_eq!(got["eventtype"], "rpc");
617        assert_eq!(got["data"]["command"], "eventrecv");
618        assert_eq!(got["data"]["data"]["data"]["k"], "v");
619    }
620}
agentmux_srv/backend/eventbus.rs

agentmux_srv/backend/
eventbus.rs
