chore: initial sanitized public snapshot

crates/aria2-rust-pro-rpc/src/websocket/bridge.rs

@@ -0,0 +1,65 @@
+use std::sync::{Arc, Mutex};
+
+use aria2_rust_pro_core::{EventListener, RuntimeEvent, RuntimeEventKind};
+
+use crate::model::RpcMeta;
+
+use super::{
+    RpcNotificationKind, notification::RpcNotificationEvent, registry::WebSocketSessionRegistry,
+};
+
+#[derive(Debug, Clone)]
+/// Runtime event listener that forwards core download events into WebSocket session queues.
+pub(super) struct RuntimeEventWebSocketBridge {
+    /// Shared session registry updated whenever a compatible runtime event is observed.
+    sessions: Arc<Mutex<WebSocketSessionRegistry>>,
+}
+
+impl RuntimeEventWebSocketBridge {
+    /// Builds a bridge backed by the shared WebSocket session registry.
+    pub(super) const fn new(sessions: Arc<Mutex<WebSocketSessionRegistry>>) -> Self {
+        Self { sessions }
+    }
+}
+
+impl EventListener for RuntimeEventWebSocketBridge {
+    fn on_event(&mut self, event: &RuntimeEvent) {
+        let Some(kind) = runtime_event_to_rpc_notification_kind(event.kind) else {
+            return;
+        };
+        let gid = event.gid.map(|gid| gid.to_string());
+        let rpc_event = RpcNotificationEvent {
+            kind,
+            method: String::new(),
+            gid,
+            payload: None,
+            meta: RpcMeta::default(),
+        };
+        if let Ok(mut sessions) = self.sessions.lock() {
+            sessions.queue_broadcast_frame(rpc_event.to_websocket_frame());
+        }
+    }
+}
+
+/// Maps core runtime events onto the subset of WebSocket notifications exposed over RPC.
+const fn runtime_event_to_rpc_notification_kind(
+    kind: RuntimeEventKind,
+) -> Option<RpcNotificationKind> {
+    match kind {
+        RuntimeEventKind::DownloadAdded
+        | RuntimeEventKind::DownloadResumed
+        | RuntimeEventKind::OptionChanged
+        | RuntimeEventKind::SessionSaving
+        | RuntimeEventKind::SessionSaved
+        | RuntimeEventKind::ShutdownRequested
+        | RuntimeEventKind::ForceShutdownRequested
+        | RuntimeEventKind::SchedulerTick
+        | RuntimeEventKind::StatisticsUpdated
+        | RuntimeEventKind::PieceUpdated => None,
+        RuntimeEventKind::DownloadStarted => Some(RpcNotificationKind::DownloadStarted),
+        RuntimeEventKind::DownloadPaused => Some(RpcNotificationKind::DownloadPaused),
+        RuntimeEventKind::DownloadRemoved => Some(RpcNotificationKind::DownloadStopped),
+        RuntimeEventKind::DownloadCompleted => Some(RpcNotificationKind::DownloadComplete),
+        RuntimeEventKind::DownloadErrored => Some(RpcNotificationKind::DownloadError),
+    }
+}

crates/aria2-rust-pro-rpc/src/websocket/notification.rs

@@ -0,0 +1,189 @@
+use std::collections::BTreeMap;
+
+use crate::model::{RpcMeta, RpcValue};
+
+/// Maximum queued outbound WebSocket frames retained per connected session.
+pub(super) const MAX_PENDING_WEBSOCKET_FRAMES_PER_SESSION: usize = 256;
+/// Maximum bridged runtime-event broadcast frames retained before session fan-out.
+pub(super) const MAX_PENDING_WEBSOCKET_BROADCAST_FRAMES: usize = 1024;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+/// Notification kinds that can be bridged onto WebSocket sessions.
+pub enum RpcNotificationKind {
+    /// A download transitioned into the active state.
+    DownloadStarted,
+    /// A download was paused.
+    DownloadPaused,
+    /// A download was stopped or removed.
+    DownloadStopped,
+    /// A download completed successfully.
+    DownloadComplete,
+    /// A download ended in error.
+    DownloadError,
+    /// A download was removed.
+    DownloadRemoved,
+    /// A download entered the waiting queue.
+    DownloadWaiting,
+    /// A waiting download became active again.
+    DownloadActive,
+    /// A `BitTorrent` download completed.
+    DownloadBtDownloadComplete,
+    /// Synthetic notification for version polling.
+    SystemVersion,
+    /// Synthetic notification for method-list polling.
+    SystemListMethods,
+    /// Synthetic notification for notification-list polling.
+    SystemListNotifications,
+}
+
+#[derive(Debug, Clone, PartialEq)]
+/// Notification payload bridged to WebSocket clients.
+pub struct RpcNotificationEvent {
+    /// Logical notification kind.
+    pub kind: RpcNotificationKind,
+    /// Explicit method name override, if present.
+    pub method: String,
+    /// Download gid associated with the notification, if any.
+    pub gid: Option<String>,
+    /// Additional notification payload.
+    pub payload: Option<RpcValue>,
+    /// Supplemental metadata.
+    pub meta: RpcMeta,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+/// Minimal WebSocket frame shapes used by the RPC server.
+pub enum RpcWebSocketFrame {
+    /// UTF-8 text frame.
+    Text(String),
+    /// Binary frame carrying JSON-RPC bytes.
+    Binary(Vec<u8>),
+    /// Ping control frame.
+    Ping(Vec<u8>),
+    /// Pong control frame.
+    Pong(Vec<u8>),
+    /// Close control frame.
+    Close,
+}
+
+impl RpcNotificationKind {
+    /// Returns the canonical aria2-compatible method name for this notification.
+    #[must_use]
+    pub const fn method_name(self) -> &'static str {
+        match self {
+            Self::DownloadStarted | Self::DownloadActive => "aria2.onDownloadStart",
+            Self::DownloadPaused | Self::DownloadWaiting => "aria2.onDownloadPause",
+            Self::DownloadStopped | Self::DownloadRemoved => "aria2.onDownloadStop",
+            Self::DownloadComplete => "aria2.onDownloadComplete",
+            Self::DownloadError => "aria2.onDownloadError",
+            Self::DownloadBtDownloadComplete => "aria2.onBtDownloadComplete",
+            Self::SystemVersion => "aria2.getVersion",
+            Self::SystemListMethods => "system.listMethods",
+            Self::SystemListNotifications => "system.listNotifications",
+        }
+    }
+}
+
+impl RpcNotificationEvent {
+    #[must_use]
+    /// Returns the effective WebSocket method name for the event.
+    pub fn websocket_method_name(&self) -> &str {
+        if self.method.is_empty() {
+            self.kind.method_name()
+        } else {
+            &self.method
+        }
+    }
+
+    #[must_use]
+    /// Converts the event into a text WebSocket frame.
+    pub fn to_websocket_frame(&self) -> RpcWebSocketFrame {
+        RpcWebSocketFrame::Text(self.to_websocket_json())
+    }
+
+    #[must_use]
+    /// Renders the event into a JSON-RPC notification string.
+    pub fn to_websocket_json(&self) -> String {
+        let method = self.websocket_method_name();
+        let params = websocket_notification_params(self.gid.as_deref(), self.payload.as_ref());
+        let params = rpc_value_to_json(&RpcValue::Array(params));
+        format!(
+            "{{\"jsonrpc\":\"2.0\",\"method\":\"{}\",\"params\":{}}}",
+            escape_json(method),
+            params
+        )
+    }
+}
+
+/// Builds the JSON-RPC `params` array for a bridged WebSocket notification event.
+pub(super) fn websocket_notification_params(
+    gid: Option<&str>,
+    payload: Option<&RpcValue>,
+) -> Vec<RpcValue> {
+    if let Some(RpcValue::Array(items)) = payload {
+        return items.clone();
+    }
+
+    let mut event_spec = BTreeMap::new();
+    if let Some(gid) = gid {
+        event_spec.insert("gid".to_owned(), RpcValue::String(gid.to_owned()));
+    }
+
+    if let Some(RpcValue::Object(map)) = payload {
+        for (key, value) in map {
+            event_spec.insert(key.clone(), value.clone());
+        }
+    }
+
+    vec![RpcValue::Object(event_spec)]
+}
+
+/// Renders a transport-neutral RPC value into compact JSON text for WebSocket frames.
+pub(super) fn rpc_value_to_json(value: &RpcValue) -> String {
+    match value {
+        RpcValue::Null => "null".to_owned(),
+        RpcValue::Bool(value) => value.to_string(),
+        RpcValue::Number(value) => value.to_string(),
+        RpcValue::String(value) => format!("\"{}\"", escape_json(value)),
+        RpcValue::Array(values) => {
+            let items = values
+                .iter()
+                .map(rpc_value_to_json)
+                .collect::<Vec<_>>()
+                .join(",");
+            format!("[{items}]")
+        }
+        RpcValue::Object(map) => {
+            let members = map
+                .iter()
+                .map(|(key, value)| {
+                    format!("\"{}\":{}", escape_json(key), rpc_value_to_json(value))
+                })
+                .collect::<Vec<_>>()
+                .join(",");
+            format!("{{{members}}}")
+        }
+    }
+}
+
+/// Escapes a string for safe embedding in generated JSON text.
+pub(super) fn escape_json(value: &str) -> String {
+    let mut escaped = String::with_capacity(value.len());
+    for ch in value.chars() {
+        match ch {
+            '"' => escaped.push_str("\\\""),
+            '\\' => escaped.push_str("\\\\"),
+            '\n' => escaped.push_str("\\n"),
+            '\r' => escaped.push_str("\\r"),
+            '\t' => escaped.push_str("\\t"),
+            '\u{08}' => escaped.push_str("\\b"),
+            '\u{0C}' => escaped.push_str("\\f"),
+            ch if ch.is_control() => {
+                use std::fmt::Write as _;
+                let _ = write!(escaped, "\\u{:04x}", u32::from(ch));
+            }
+            ch => escaped.push(ch),
+        }
+    }
+    escaped
+}

crates/aria2-rust-pro-rpc/src/websocket/registry.rs

@@ -0,0 +1,219 @@
+use std::collections::{BTreeMap, BTreeSet, VecDeque};
+
+use super::notification::{
+    MAX_PENDING_WEBSOCKET_BROADCAST_FRAMES, MAX_PENDING_WEBSOCKET_FRAMES_PER_SESSION,
+    RpcNotificationEvent, RpcNotificationKind, RpcWebSocketFrame,
+};
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+/// Subscription descriptor for a WebSocket notification consumer.
+pub struct WebSocketSubscription {
+    /// Stable subscription identifier.
+    pub id: String,
+    /// Notification kind being subscribed to.
+    pub kind: RpcNotificationKind,
+    /// Client-visible topic string.
+    pub topic: String,
+}
+
+#[derive(Debug, Default)]
+/// Registry of active WebSocket notification subscriptions.
+pub struct WebSocketNotificationRegistry {
+    /// Subscription descriptors keyed by subscription identifier.
+    subscriptions: BTreeMap<String, WebSocketSubscription>,
+    /// Reverse index from notification kind to subscribed identifiers.
+    by_kind: BTreeMap<RpcNotificationKind, BTreeSet<String>>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+/// Per-session queued WebSocket state.
+pub struct WebSocketSessionState {
+    /// Stable session identifier.
+    pub id: String,
+    /// Frames waiting to be written to the socket for this session.
+    pending_frames: VecDeque<RpcWebSocketFrame>,
+}
+
+#[derive(Debug, Default)]
+/// Registry of connected WebSocket sessions.
+pub struct WebSocketSessionRegistry {
+    /// Session entries keyed by their stable session identifier.
+    sessions: BTreeMap<String, WebSocketSessionState>,
+    /// Bounded runtime-event broadcast ingress awaiting session fan-out.
+    pending_broadcast_frames: VecDeque<RpcWebSocketFrame>,
+}
+
+impl WebSocketNotificationRegistry {
+    /// Registers a new subscription.
+    pub fn subscribe(&mut self, subscription: WebSocketSubscription) {
+        self.by_kind
+            .entry(subscription.kind)
+            .or_default()
+            .insert(subscription.id.clone());
+        self.subscriptions
+            .insert(subscription.id.clone(), subscription);
+    }
+
+    /// Removes a subscription by identifier.
+    pub fn unsubscribe(&mut self, subscription_id: &str) {
+        if let Some(subscription) = self.subscriptions.remove(subscription_id)
+            && let Some(ids) = self.by_kind.get_mut(&subscription.kind)
+        {
+            ids.remove(subscription_id);
+        }
+    }
+
+    #[must_use]
+    /// Returns all subscriptions for the provided notification kind.
+    pub fn subscriptions_for(&self, kind: RpcNotificationKind) -> Vec<&WebSocketSubscription> {
+        self.by_kind
+            .get(&kind)
+            .into_iter()
+            .flat_map(|ids| ids.iter())
+            .filter_map(|id| self.subscriptions.get(id))
+            .collect()
+    }
+
+    #[must_use]
+    /// Returns the number of registered subscriptions.
+    pub fn len(&self) -> usize {
+        self.subscriptions.len()
+    }
+
+    #[must_use]
+    /// Returns whether the registry contains no subscriptions.
+    pub fn is_empty(&self) -> bool {
+        self.subscriptions.is_empty()
+    }
+
+    /// Builds a copy-on-write frame list for a single event fan-out.
+    #[must_use]
+    pub fn frames_for_event(
+        &self,
+        event: &RpcNotificationEvent,
+    ) -> Vec<(String, RpcWebSocketFrame)> {
+        let frame = event.to_websocket_frame();
+        self.subscriptions_for(event.kind)
+            .into_iter()
+            .map(|subscription| (subscription.id.clone(), frame.clone()))
+            .collect()
+    }
+}
+
+impl WebSocketSessionRegistry {
+    /// Connects or reuses a session entry for the provided identifier.
+    pub fn connect(&mut self, session_id: impl Into<String>) {
+        let session_id = session_id.into();
+        self.sessions
+            .entry(session_id.clone())
+            .or_insert_with(|| WebSocketSessionState {
+                id: session_id,
+                pending_frames: VecDeque::new(),
+            });
+    }
+
+    /// Disconnects a session and drops its pending queue.
+    pub fn disconnect(&mut self, session_id: &str) {
+        self.sessions.remove(session_id);
+    }
+
+    #[must_use]
+    /// Returns the number of connected sessions.
+    pub fn len(&self) -> usize {
+        self.sessions.len()
+    }
+
+    #[must_use]
+    /// Returns whether the registry contains no connected sessions.
+    pub fn is_empty(&self) -> bool {
+        self.sessions.is_empty()
+    }
+
+    #[must_use]
+    /// Returns whether the registry contains the provided session id.
+    pub fn contains(&self, session_id: &str) -> bool {
+        self.sessions.contains_key(session_id)
+    }
+
+    #[must_use]
+    /// Returns the connected session identifiers in key order.
+    pub fn session_ids(&self) -> Vec<String> {
+        self.sessions.keys().cloned().collect()
+    }
+
+    /// Queues a notification frame for every connected session.
+    pub fn queue_event_for_all(&mut self, event: &RpcNotificationEvent) {
+        let frame = event.to_websocket_frame();
+        for session in self.sessions.values_mut() {
+            enqueue_session_frame(session, frame.clone());
+        }
+    }
+
+    /// Queues one bridged broadcast frame for later session fan-out.
+    pub fn queue_broadcast_frame(&mut self, frame: RpcWebSocketFrame) {
+        if self.sessions.is_empty() {
+            return;
+        }
+        if self.pending_broadcast_frames.len() >= MAX_PENDING_WEBSOCKET_BROADCAST_FRAMES {
+            let _ = self.pending_broadcast_frames.pop_front();
+        }
+        self.pending_broadcast_frames.push_back(frame);
+    }
+
+    /// Queues a frame for a single session and reports whether it existed.
+    pub fn queue_frame_for_session(&mut self, session_id: &str, frame: RpcWebSocketFrame) -> bool {
+        let Some(session) = self.sessions.get_mut(session_id) else {
+            return false;
+        };
+        enqueue_session_frame(session, frame);
+        true
+    }
+
+    #[must_use]
+    /// Returns the pending frame count for a session, if it exists.
+    pub fn pending_count(&self, session_id: &str) -> Option<usize> {
+        self.sessions
+            .get(session_id)
+            .map(|session| session.pending_frames.len())
+    }
+
+    #[must_use]
+    /// Returns the pending bridged broadcast frame count.
+    pub fn pending_broadcast_count(&self) -> usize {
+        self.pending_broadcast_frames.len()
+    }
+
+    /// Fans out all pending bridged broadcast frames into the per-session queues.
+    pub fn drain_broadcast_frames_into_sessions(&mut self) {
+        while let Some(frame) = self.pending_broadcast_frames.pop_front() {
+            for session in self.sessions.values_mut() {
+                enqueue_session_frame(session, frame.clone());
+            }
+        }
+    }
+
+    /// Drains all queued frames for one session after applying pending broadcast fan-out.
+    pub fn drain_session_frames(&mut self, session_id: &str) -> Vec<RpcWebSocketFrame> {
+        self.drain_broadcast_frames_into_sessions();
+        self.sessions
+            .get_mut(session_id)
+            .map(|session| session.pending_frames.drain(..).collect())
+            .unwrap_or_default()
+    }
+
+    /// Pops the oldest pending frame for a session.
+    pub fn pop_frame(&mut self, session_id: &str) -> Option<RpcWebSocketFrame> {
+        self.drain_broadcast_frames_into_sessions();
+        self.sessions
+            .get_mut(session_id)
+            .and_then(|session| session.pending_frames.pop_front())
+    }
+}
+
+/// Adds a frame to the session queue while enforcing the bounded backpressure policy.
+fn enqueue_session_frame(session: &mut WebSocketSessionState, frame: RpcWebSocketFrame) {
+    if session.pending_frames.len() >= MAX_PENDING_WEBSOCKET_FRAMES_PER_SESSION {
+        let _ = session.pending_frames.pop_front();
+    }
+    session.pending_frames.push_back(frame);
+}

crates/aria2-rust-pro-rpc/src/websocket/tests.rs

@@ -0,0 +1,397 @@
+use std::{
+    collections::BTreeMap,
+    sync::{Arc, Mutex},
+};
+
+use aria2_rust_pro_core::{
+    DownloadEngine, DownloadId, EventListener, RuntimeEvent, RuntimeEventKind,
+};
+
+use super::notification::{
+    MAX_PENDING_WEBSOCKET_BROADCAST_FRAMES, MAX_PENDING_WEBSOCKET_FRAMES_PER_SESSION,
+};
+use super::*;
+use crate::{InProcessRpcDispatcher, JsonRpcRequest, RpcValue};
+
+#[test]
+/// Verifies that notification kinds keep the upstream aria2 WebSocket method names.
+fn notification_kind_uses_upstream_websocket_method_names() {
+    assert_eq!(
+        RpcNotificationKind::DownloadStarted.method_name(),
+        "aria2.onDownloadStart"
+    );
+    assert_eq!(
+        RpcNotificationKind::DownloadBtDownloadComplete.method_name(),
+        "aria2.onBtDownloadComplete"
+    );
+}
+
+#[test]
+/// Verifies that notification events render the upstream `JSON-RPC` notification shape.
+fn websocket_event_renders_upstream_jsonrpc_notification_shape() {
+    let event = RpcNotificationEvent {
+        kind: RpcNotificationKind::DownloadComplete,
+        method: String::new(),
+        gid: Some("a1b2c3".to_owned()),
+        payload: None,
+        meta: crate::model::RpcMeta::default(),
+    };
+
+    assert_eq!(
+        event.to_websocket_json(),
+        "{\"jsonrpc\":\"2.0\",\"method\":\"aria2.onDownloadComplete\",\"params\":[{\"gid\":\"a1b2c3\"}]}"
+    );
+    assert_eq!(
+        event.to_websocket_frame(),
+        RpcWebSocketFrame::Text(
+            "{\"jsonrpc\":\"2.0\",\"method\":\"aria2.onDownloadComplete\",\"params\":[{\"gid\":\"a1b2c3\"}]}".to_owned()
+        )
+    );
+}
+
+#[test]
+/// Verifies that object payloads merge with the gid field in notification output.
+fn websocket_event_merges_gid_with_object_payload() {
+    let event = RpcNotificationEvent {
+        kind: RpcNotificationKind::DownloadError,
+        method: String::new(),
+        gid: Some("deadbeef".to_owned()),
+        payload: Some(RpcValue::Object(BTreeMap::from([(
+            "status".to_owned(),
+            RpcValue::String("error".to_owned()),
+        )]))),
+        meta: crate::model::RpcMeta::default(),
+    };
+
+    assert_eq!(
+        event.to_websocket_json(),
+        "{\"jsonrpc\":\"2.0\",\"method\":\"aria2.onDownloadError\",\"params\":[{\"gid\":\"deadbeef\",\"status\":\"error\"}]}"
+    );
+}
+
+#[test]
+/// Verifies that one notification frame is emitted for each matching subscription.
+fn registry_emits_one_frame_per_matching_subscription() {
+    let mut registry = WebSocketNotificationRegistry::default();
+    registry.subscribe(WebSocketSubscription {
+        id: "sub-a".to_owned(),
+        kind: RpcNotificationKind::DownloadStarted,
+        topic: "aria2.onDownloadStart".to_owned(),
+    });
+    registry.subscribe(WebSocketSubscription {
+        id: "sub-b".to_owned(),
+        kind: RpcNotificationKind::DownloadStarted,
+        topic: "aria2.onDownloadStart".to_owned(),
+    });
+    registry.subscribe(WebSocketSubscription {
+        id: "sub-c".to_owned(),
+        kind: RpcNotificationKind::DownloadComplete,
+        topic: "aria2.onDownloadComplete".to_owned(),
+    });
+
+    let event = RpcNotificationEvent {
+        kind: RpcNotificationKind::DownloadStarted,
+        method: String::new(),
+        gid: Some("feedface".to_owned()),
+        payload: None,
+        meta: crate::model::RpcMeta::default(),
+    };
+
+    let frames = registry.frames_for_event(&event);
+    assert_eq!(frames.len(), 2);
+    assert!(frames.iter().any(|(id, _)| id == "sub-a"));
+    assert!(frames.iter().any(|(id, _)| id == "sub-b"));
+    assert!(
+        frames
+            .iter()
+            .all(|(_, frame)| matches!(frame, RpcWebSocketFrame::Text(_)))
+    );
+}
+
+#[test]
+/// Verifies that broadcast queueing fans notification frames out to every connected session.
+fn session_registry_broadcasts_notification_frames_to_all_sessions() {
+    let mut registry = WebSocketSessionRegistry::default();
+    registry.connect("sess-a");
+    registry.connect("sess-b");
+
+    let event = RpcNotificationEvent {
+        kind: RpcNotificationKind::DownloadComplete,
+        method: String::new(),
+        gid: Some("abc123".to_owned()),
+        payload: None,
+        meta: crate::model::RpcMeta::default(),
+    };
+
+    registry.queue_event_for_all(&event);
+
+    assert_eq!(registry.pending_count("sess-a"), Some(1));
+    assert_eq!(registry.pending_count("sess-b"), Some(1));
+    assert_eq!(
+        registry.pop_frame("sess-a"),
+        Some(RpcWebSocketFrame::Text(
+            "{\"jsonrpc\":\"2.0\",\"method\":\"aria2.onDownloadComplete\",\"params\":[{\"gid\":\"abc123\"}]}".to_owned()
+        ))
+    );
+    assert_eq!(
+        registry.pop_frame("sess-b"),
+        Some(RpcWebSocketFrame::Text(
+            "{\"jsonrpc\":\"2.0\",\"method\":\"aria2.onDownloadComplete\",\"params\":[{\"gid\":\"abc123\"}]}".to_owned()
+        ))
+    );
+}
+
+#[test]
+/// Verifies that bridged broadcast ingress stays bounded and drops the oldest frame first.
+fn session_registry_enforces_bounded_broadcast_ingress_backpressure() {
+    let mut registry = WebSocketSessionRegistry::default();
+    registry.connect("sess-a");
+
+    for index in 0..(MAX_PENDING_WEBSOCKET_BROADCAST_FRAMES + 8) {
+        registry.queue_broadcast_frame(RpcWebSocketFrame::Ping(vec![
+            u8::try_from(index % 256).expect("byte should fit"),
+        ]));
+    }
+
+    assert_eq!(
+        registry.pending_broadcast_count(),
+        MAX_PENDING_WEBSOCKET_BROADCAST_FRAMES
+    );
+    assert_eq!(
+        registry.pop_frame("sess-a"),
+        Some(RpcWebSocketFrame::Ping(vec![8])),
+        "oldest bridged broadcast frames should be dropped first once the ingress limit is reached"
+    );
+}
+
+#[test]
+/// Verifies that bridged broadcast ingress fans out only when a session drain occurs.
+fn session_registry_drains_broadcast_ingress_into_connected_sessions() {
+    let mut registry = WebSocketSessionRegistry::default();
+    registry.connect("sess-a");
+    registry.connect("sess-b");
+
+    registry.queue_broadcast_frame(RpcWebSocketFrame::Ping(vec![4, 2]));
+
+    assert_eq!(registry.pending_count("sess-a"), Some(0));
+    assert_eq!(registry.pending_count("sess-b"), Some(0));
+    assert_eq!(registry.pending_broadcast_count(), 1);
+
+    let sess_a_frames = registry.drain_session_frames("sess-a");
+    assert_eq!(registry.pending_broadcast_count(), 0);
+    assert_eq!(sess_a_frames, vec![RpcWebSocketFrame::Ping(vec![4, 2])]);
+    assert_eq!(
+        registry.pop_frame("sess-b"),
+        Some(RpcWebSocketFrame::Ping(vec![4, 2]))
+    );
+}
+
+#[test]
+/// Verifies targeted queueing, missing-session rejection, and disconnect behavior.
+fn session_registry_supports_targeted_queue_and_disconnect() {
+    let mut registry = WebSocketSessionRegistry::default();
+    registry.connect("sess-a");
+    registry.connect("sess-b");
+
+    assert!(registry.queue_frame_for_session("sess-a", RpcWebSocketFrame::Ping(vec![1, 2, 3]),));
+    assert!(!registry.queue_frame_for_session("sess-missing", RpcWebSocketFrame::Ping(vec![9]),));
+
+    assert_eq!(registry.pending_count("sess-a"), Some(1));
+    assert_eq!(registry.pending_count("sess-b"), Some(0));
+    assert_eq!(
+        registry.pop_frame("sess-a"),
+        Some(RpcWebSocketFrame::Ping(vec![1, 2, 3]))
+    );
+
+    registry.disconnect("sess-b");
+    assert!(!registry.contains("sess-b"));
+    assert_eq!(registry.len(), 1);
+}
+
+#[test]
+/// Verifies that the per-session queue is bounded and drops the oldest frame under pressure.
+fn session_registry_enforces_bounded_pending_frame_backpressure() {
+    let mut registry = WebSocketSessionRegistry::default();
+    registry.connect("sess-a");
+
+    for index in 0..(MAX_PENDING_WEBSOCKET_FRAMES_PER_SESSION + 8) {
+        assert!(registry.queue_frame_for_session(
+            "sess-a",
+            RpcWebSocketFrame::Ping(vec![u8::try_from(index % 256).expect("byte should fit")]),
+        ));
+    }
+
+    assert_eq!(
+        registry.pending_count("sess-a"),
+        Some(MAX_PENDING_WEBSOCKET_FRAMES_PER_SESSION)
+    );
+    assert_eq!(
+        registry.pop_frame("sess-a"),
+        Some(RpcWebSocketFrame::Ping(vec![8])),
+        "oldest queued frames should be dropped first once the per-session limit is reached"
+    );
+}
+
+#[test]
+/// Verifies that the runtime-event bridge queues real completion events into session frames.
+fn runtime_event_bridge_queues_real_download_events_into_sessions() {
+    let sessions = Arc::new(Mutex::new(WebSocketSessionRegistry::default()));
+    sessions
+        .lock()
+        .expect("sessions lock should succeed")
+        .connect("sess-a");
+    let mut bridge = RuntimeEventWebSocketBridge::new(Arc::clone(&sessions));
+
+    bridge.on_event(
+        &RuntimeEvent::new(RuntimeEventKind::DownloadCompleted).with_gid(DownloadId::new(0x2a)),
+    );
+
+    assert_eq!(
+        sessions
+            .lock()
+            .expect("sessions lock should succeed")
+            .pending_broadcast_count(),
+        1
+    );
+    let frame = sessions
+        .lock()
+        .expect("sessions lock should succeed")
+        .pop_frame("sess-a")
+        .expect("download completion should queue a notification frame");
+    match frame {
+        RpcWebSocketFrame::Text(text) => {
+            assert!(text.contains("aria2.onDownloadComplete"));
+            assert!(text.contains(r#""gid":"000000000000002a""#));
+        }
+        other => panic!("unexpected bridged notification frame: {other:?}"),
+    }
+}
+
+#[test]
+/// Verifies that `addUri` registration alone does not emit a start notification.
+fn runtime_event_bridge_does_not_treat_add_uri_as_download_start() {
+    let sessions = Arc::new(Mutex::new(WebSocketSessionRegistry::default()));
+    sessions
+        .lock()
+        .expect("sessions lock should succeed")
+        .connect("sess-a");
+    let bridge = RuntimeEventWebSocketBridge::new(Arc::clone(&sessions));
+    let mut dispatcher = InProcessRpcDispatcher::new();
+    dispatcher.register_runtime_listener(bridge);
+
+    let response = dispatcher.dispatch_json(JsonRpcRequest {
+        jsonrpc: Some("2.0".to_owned()),
+        id: Some(crate::jsonrpc::JsonRpcId::Number(1)),
+        method: "aria2.addUri".to_owned(),
+        params: vec![RpcValue::String("https://example.org/file.iso".to_owned())],
+        meta: crate::model::RpcMeta::default(),
+    });
+    assert!(response.error.is_none());
+
+    assert!(
+        sessions
+            .lock()
+            .expect("sessions lock should succeed")
+            .pop_frame("sess-a")
+            .is_none(),
+        "addUri should not emit aria2.onDownloadStart before the download actually starts"
+    );
+}
+
+#[test]
+/// Verifies that start notifications only appear after the scheduler activates the download.
+fn runtime_event_bridge_emits_start_only_when_scheduler_activates_download() {
+    let sessions = Arc::new(Mutex::new(WebSocketSessionRegistry::default()));
+    sessions
+        .lock()
+        .expect("sessions lock should succeed")
+        .connect("sess-a");
+
+    let mut engine = DownloadEngine::new();
+    engine.register_listener(RuntimeEventWebSocketBridge::new(Arc::clone(&sessions)));
+    let gid = engine.add_uri("https://example.org/file.iso").gid();
+
+    assert!(
+        sessions
+            .lock()
+            .expect("sessions lock should succeed")
+            .pop_frame("sess-a")
+            .is_none(),
+        "registration alone should not emit a start notification"
+    );
+
+    let _ = engine.schedule_once();
+
+    let frame = sessions
+        .lock()
+        .expect("sessions lock should succeed")
+        .pop_frame("sess-a")
+        .expect("scheduler activation should emit start notification");
+    match frame {
+        RpcWebSocketFrame::Text(text) => {
+            assert!(text.contains("aria2.onDownloadStart"));
+            assert!(text.contains(&format!(r#""gid":"{gid}""#)));
+        }
+        other => panic!("unexpected frame: {other:?}"),
+    }
+}
+
+#[test]
+/// Verifies that resumed downloads emit start only after they become active again.
+fn runtime_event_bridge_does_not_emit_start_until_resumed_download_is_active_again() {
+    let sessions = Arc::new(Mutex::new(WebSocketSessionRegistry::default()));
+    sessions
+        .lock()
+        .expect("sessions lock should succeed")
+        .connect("sess-a");
+
+    let mut engine = DownloadEngine::new();
+    engine.register_listener(RuntimeEventWebSocketBridge::new(Arc::clone(&sessions)));
+    let gid = engine.add_uri("https://example.org/file.iso").gid();
+    let _ = engine.schedule_once();
+
+    let _ = sessions
+        .lock()
+        .expect("sessions lock should succeed")
+        .pop_frame("sess-a");
+
+    engine.pause(gid).expect("pause should succeed");
+    let pause = sessions
+        .lock()
+        .expect("sessions lock should succeed")
+        .pop_frame("sess-a")
+        .expect("pause should emit pause notification");
+    match pause {
+        RpcWebSocketFrame::Text(text) => {
+            assert!(text.contains("aria2.onDownloadPause"));
+        }
+        other => panic!("unexpected pause frame: {other:?}"),
+    }
+
+    engine
+        .resume(gid)
+        .expect("resume should move download back to waiting");
+    assert!(
+        sessions
+            .lock()
+            .expect("sessions lock should succeed")
+            .pop_frame("sess-a")
+            .is_none(),
+        "resume should not emit start notification before reactivation"
+    );
+
+    let _ = engine.schedule_once();
+
+    let frame = sessions
+        .lock()
+        .expect("sessions lock should succeed")
+        .pop_frame("sess-a")
+        .expect("reactivation should emit start notification");
+    match frame {
+        RpcWebSocketFrame::Text(text) => {
+            assert!(text.contains("aria2.onDownloadStart"));
+            assert!(text.contains(&format!(r#""gid":"{gid}""#)));
+        }
+        other => panic!("unexpected reactivation frame: {other:?}"),
+    }
+}