v0.2.7-pre.004

crates/ksp-onchain-transport-lib/src/error.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/src/error.rs
-// version: 3
+// version: 4
 
 /// Error code used when no logical endpoint can satisfy a request.
 pub const ERROR_CODE_ENDPOINT_SELECTION_FAILED: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("onchain_transport", "endpoint_selection_failed");
@@ -27,3 +27,11 @@ pub const ERROR_CODE_RATE_LIMITED: ksp_core_lib::ErrorCode = ksp_core_lib::Error
 pub const ERROR_CODE_RPC_APPLICATION_ERROR: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("onchain_transport", "rpc_application_error");
 /// Error code used when a transport deadline expires.
 pub const ERROR_CODE_TIMEOUT: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("onchain_transport", "timeout");
+/// Error code used when a bounded WebSocket runtime queue or pending-request capacity is exhausted.
+pub const ERROR_CODE_WS_BACKPRESSURE_OVERFLOW: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("onchain_transport", "ws_backpressure_overflow");
+/// Error code used when a physical WebSocket connection or handshake fails.
+pub const ERROR_CODE_WS_CONNECTION_FAILED: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("onchain_transport", "ws_connection_failed");
+/// Error code used when WebSocket wire data violates the KSP protocol contract.
+pub const ERROR_CODE_WS_PROTOCOL_ERROR: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("onchain_transport", "ws_protocol_error");
+/// Error code used when a WebSocket session is no longer available to a caller.
+pub const ERROR_CODE_WS_SESSION_CLOSED: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("onchain_transport", "ws_session_closed");

crates/ksp-onchain-transport-lib/src/lib.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/src/lib.rs
-// version: 22
+// version: 23
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -17,7 +17,8 @@
 //! modern/legacy `getBlock`, positional inflation rewards, runtime-provided economics values and the final `KSP-TRANSPORT-007` compliance target.
 //! The candidate surface therefore exposes typed wrappers for all 52 current audited Solana HTTP methods while retaining 14 removed historical descriptors.
 //! `0.2.7-pre.002` adds the provider-neutral WebSocket settings foundation, redacted endpoint URLs, explicit protocol-family discrimination, local session/subscription
-//! identities, observable lifecycle states and safe snapshots. Physical sockets and subscription execution are intentionally deferred to later prereleases.
+//! identities, observable lifecycle states and safe snapshots. `0.2.7-pre.004` adds the first physical WebSocket runtime: bounded handshake, one actor-owned socket,
+//! bounded command/pending JSON-RPC flow, safe session snapshots and deterministic local-server fixtures. Subscription registration remains deferred.
 
 mod client;
 mod constants;
@@ -37,6 +38,7 @@ mod rpc_tokens;
 mod rpc_transactions;
 mod settings;
 mod ws_lifecycle;
+mod ws_session;
 mod ws_settings;
 
 /// Passive runtime availability reported for one logical HTTP endpoint.
@@ -73,6 +75,14 @@ pub use self::error::ERROR_CODE_RATE_LIMITED;
 pub use self::error::ERROR_CODE_RPC_APPLICATION_ERROR;
 /// Error code used when a transport deadline expires.
 pub use self::error::ERROR_CODE_TIMEOUT;
+/// Error code used when bounded WebSocket runtime capacity is exhausted.
+pub use self::error::ERROR_CODE_WS_BACKPRESSURE_OVERFLOW;
+/// Error code used when a physical WebSocket connection or handshake fails.
+pub use self::error::ERROR_CODE_WS_CONNECTION_FAILED;
+/// Error code used when WebSocket wire data violates protocol invariants.
+pub use self::error::ERROR_CODE_WS_PROTOCOL_ERROR;
+/// Error code used when a WebSocket session is no longer available.
+pub use self::error::ERROR_CODE_WS_SESSION_CLOSED;
 /// JSON-RPC 2.0 error payload returned by a remote Solana endpoint.
 pub use self::json_rpc::JsonRpcErrorObject;
 /// Validated JSON-RPC 2.0 error response.
@@ -309,6 +319,8 @@ pub use self::ws_lifecycle::WsSubscriptionKind;
 pub use self::ws_lifecycle::WsSubscriptionSnapshot;
 /// Observable lifecycle state of one logical WebSocket subscription.
 pub use self::ws_lifecycle::WsSubscriptionState;
+/// Shareable handle for one explicitly created physical WebSocket session.
+pub use self::ws_session::WsSession;
 /// Open cluster or network descriptor used by WebSocket endpoint settings.
 pub use self::ws_settings::WsClusterName;
 /// Runtime settings for one named WebSocket endpoint.

crates/ksp-onchain-transport-lib/src/ws_lifecycle.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/src/ws_lifecycle.rs
-// version: 2
+// version: 3
 
 /// Stable local identity assigned to one physical WebSocket session.
 #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
@@ -183,7 +183,6 @@ pub struct WsSessionSnapshot {
 impl WsSessionSnapshot {
     /// Creates one safe session projection for Transport runtime internals.
     #[must_use]
-    #[cfg(test)]
     #[allow(clippy::too_many_arguments)]
     pub(crate) fn new(
         id: crate::WsSessionId,

crates/ksp-onchain-transport-lib/src/ws_session.rs

@@ -0,0 +1,574 @@
+// file: crates/ksp-onchain-transport-lib/src/ws_session.rs
+// version: 1
+
+use futures_util::SinkExt; // rust-rules: trait-import
+use futures_util::StreamExt; // rust-rules: trait-import
+
+static NEXT_WS_SESSION_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
+
+/// Shareable handle for one explicitly created physical WebSocket session.
+///
+/// The handle never exposes the sensitive endpoint URL or the underlying socket. All socket I/O is owned by one internal actor task and all caller
+/// interaction is serialized through a bounded command queue.
+#[derive(Clone)]
+pub struct WsSession {
+    id: crate::WsSessionId,
+    command_tx: tokio::sync::mpsc::Sender<WsSessionCommand>,
+    snapshot_rx: tokio::sync::watch::Receiver<crate::WsSessionSnapshot>,
+    command_timeout: std::time::Duration,
+}
+
+impl WsSession {
+    /// Opens one physical WebSocket connection for the supplied endpoint settings.
+    ///
+    /// Calling this function twice with the same endpoint creates two independent physical sessions. The function returns only after the WebSocket
+    /// handshake succeeds or the configured command timeout expires.
+    pub async fn connect(endpoint: crate::WsEndpointSettings) -> ksp_core_lib::Result<Self> {
+        let validation = crate::WsTransportSettings::new(std::vec![endpoint.clone()]).validate();
+        if let std::result::Result::Err(error) = validation {
+            return std::result::Result::Err(error);
+        }
+        let id_result = next_session_id();
+        let id = match id_result {
+            std::result::Result::Ok(id) => id,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        let initial_snapshot = crate::WsSessionSnapshot::new(
+            id,
+            endpoint.name(),
+            endpoint.provider().clone(),
+            endpoint.cluster().clone(),
+            endpoint.protocol(),
+            crate::WsSessionState::Connecting,
+            0,
+            0,
+            0,
+            std::vec::Vec::new(),
+        );
+        let (command_tx, command_rx) = tokio::sync::mpsc::channel(endpoint.session().command_queue_capacity());
+        let (snapshot_tx, snapshot_rx) = tokio::sync::watch::channel(initial_snapshot);
+        let (startup_tx, startup_rx) = tokio::sync::oneshot::channel();
+        let command_timeout = endpoint.session().command_timeout();
+        ksp_logging_lib::debug!(
+            target: crate::TRACING_TARGET,
+            session_id = id.get(),
+            endpoint_name = endpoint.name(),
+            provider = endpoint.provider().as_str(),
+            cluster = endpoint.cluster().as_str(),
+            protocol = endpoint.protocol().as_str(),
+            "starting physical WebSocket session actor"
+        );
+        let join_handle = tokio::spawn(run_ws_session_actor(id, endpoint, command_rx, snapshot_tx, startup_tx));
+        let startup_wait = tokio::time::timeout(command_timeout, startup_rx).await;
+        match startup_wait {
+            std::result::Result::Ok(std::result::Result::Ok(std::result::Result::Ok(()))) => {
+                return std::result::Result::Ok(Self { id, command_tx, snapshot_rx, command_timeout });
+            },
+            std::result::Result::Ok(std::result::Result::Ok(std::result::Result::Err(error))) => {
+                join_handle.abort();
+                return std::result::Result::Err(error);
+            },
+            std::result::Result::Ok(std::result::Result::Err(_)) => {
+                join_handle.abort();
+                return std::result::Result::Err(ws_session_closed_error(id, "WebSocket session actor ended during startup"));
+            },
+            std::result::Result::Err(_) => {
+                join_handle.abort();
+                return std::result::Result::Err(ws_timeout_error(id, "WebSocket handshake exceeded the configured command timeout"));
+            },
+        }
+    }
+
+    /// Returns the stable local session identity.
+    #[must_use]
+    pub const fn id(&self) -> crate::WsSessionId {
+        return self.id;
+    }
+
+    /// Returns the latest safe runtime snapshot published by the actor.
+    #[must_use]
+    pub fn snapshot(&self) -> crate::WsSessionSnapshot {
+        return self.snapshot_rx.borrow().clone();
+    }
+
+    /// Returns the latest observable physical-session state.
+    #[must_use]
+    pub fn state(&self) -> crate::WsSessionState {
+        return self.snapshot_rx.borrow().state();
+    }
+
+    /// Executes one internal JSON-RPC request through the actor-owned socket.
+    ///
+    /// This remains crate-private in `0.2.7`; standard subscriptions consume it without exposing a public raw provider-extension escape hatch.
+    #[allow(dead_code)] // Staged in pre.004 and consumed by the subscription engine starting in pre.006.
+    pub(crate) async fn execute_json_rpc(&self, method: &'static str, params: std::vec::Vec<serde_json::Value>) -> ksp_core_lib::Result<serde_json::Value> {
+        let (response_tx, response_rx) = tokio::sync::oneshot::channel();
+        let command = WsSessionCommand::ExecuteJsonRpc { method, params, response_tx };
+        let send_wait = tokio::time::timeout(self.command_timeout, self.command_tx.send(command)).await;
+        match send_wait {
+            std::result::Result::Ok(std::result::Result::Ok(())) => {},
+            std::result::Result::Ok(std::result::Result::Err(_)) => {
+                return std::result::Result::Err(ws_session_closed_error(self.id, "WebSocket session command channel is closed"));
+            },
+            std::result::Result::Err(_) => {
+                return std::result::Result::Err(ws_timeout_error(self.id, "WebSocket session command queue remained unavailable until timeout"));
+            },
+        }
+        let response_wait = tokio::time::timeout(self.command_timeout, response_rx).await;
+        return match response_wait {
+            std::result::Result::Ok(std::result::Result::Ok(result)) => result,
+            std::result::Result::Ok(std::result::Result::Err(_)) => {
+                std::result::Result::Err(ws_session_closed_error(self.id, "WebSocket session ended before the JSON-RPC response was delivered"))
+            },
+            std::result::Result::Err(_) => {
+                std::result::Result::Err(ws_timeout_error(self.id, "WebSocket JSON-RPC request exceeded the configured command timeout"))
+            },
+        };
+    }
+}
+
+impl std::fmt::Debug for WsSession {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        return formatter.debug_struct("WsSession").field("id", &self.id).field("snapshot", &self.snapshot()).finish();
+    }
+}
+
+enum WsSessionCommand {
+    ExecuteJsonRpc {
+        method: &'static str,
+        params: std::vec::Vec<serde_json::Value>,
+        response_tx: tokio::sync::oneshot::Sender<ksp_core_lib::Result<serde_json::Value>>,
+    },
+}
+
+struct PendingWsRequest {
+    method: &'static str,
+    deadline: tokio::time::Instant,
+    response_tx: tokio::sync::oneshot::Sender<ksp_core_lib::Result<serde_json::Value>>,
+}
+
+async fn run_ws_session_actor(
+    id: crate::WsSessionId,
+    endpoint: crate::WsEndpointSettings,
+    mut command_rx: tokio::sync::mpsc::Receiver<WsSessionCommand>,
+    snapshot_tx: tokio::sync::watch::Sender<crate::WsSessionSnapshot>,
+    startup_tx: tokio::sync::oneshot::Sender<ksp_core_lib::Result<()>>,
+) {
+    let websocket_config = tokio_tungstenite::tungstenite::protocol::WebSocketConfig::default()
+        .write_buffer_size(0)
+        .max_write_buffer_size(endpoint.session().max_write_buffer_size_bytes())
+        .max_message_size(std::option::Option::Some(endpoint.session().max_message_size_bytes()))
+        .max_frame_size(std::option::Option::Some(endpoint.session().max_frame_size_bytes()));
+    ksp_logging_lib::trace!(
+        target: crate::TRACING_TARGET,
+        session_id = id.get(),
+        endpoint_name = endpoint.name(),
+        provider = endpoint.provider().as_str(),
+        cluster = endpoint.cluster().as_str(),
+        "opening physical WebSocket connection"
+    );
+    let connect_wait = tokio::time::timeout(
+        endpoint.session().command_timeout(),
+        tokio_tungstenite::connect_async_with_config(endpoint.url().as_str(), std::option::Option::Some(websocket_config), false),
+    )
+    .await;
+    let (mut websocket, handshake_status) = match connect_wait {
+        std::result::Result::Ok(std::result::Result::Ok((websocket, response))) => (websocket, response.status().as_u16()),
+        std::result::Result::Ok(std::result::Result::Err(_)) => {
+            publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Failed, 0);
+            let error = ws_connection_error(id, &endpoint, "WebSocket connection or handshake failed");
+            let _ = startup_tx.send(std::result::Result::Err(error));
+            ksp_logging_lib::warn!(
+                target: crate::TRACING_TARGET,
+                session_id = id.get(),
+                endpoint_name = endpoint.name(),
+                provider = endpoint.provider().as_str(),
+                cluster = endpoint.cluster().as_str(),
+                "physical WebSocket connection failed"
+            );
+            return;
+        },
+        std::result::Result::Err(_) => {
+            publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Failed, 0);
+            let error = ws_timeout_error(id, "WebSocket connection handshake timed out");
+            let _ = startup_tx.send(std::result::Result::Err(error));
+            ksp_logging_lib::warn!(
+                target: crate::TRACING_TARGET,
+                session_id = id.get(),
+                endpoint_name = endpoint.name(),
+                "physical WebSocket connection handshake timed out"
+            );
+            return;
+        },
+    };
+    publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Active, 0);
+    let _ = startup_tx.send(std::result::Result::Ok(()));
+    ksp_logging_lib::debug!(
+        target: crate::TRACING_TARGET,
+        session_id = id.get(),
+        endpoint_name = endpoint.name(),
+        handshake_status,
+        "physical WebSocket session is active"
+    );
+    let mut next_request_id = 1_u64;
+    let mut pending = std::collections::BTreeMap::<u64, PendingWsRequest>::new();
+    loop {
+        let timeout_deadline = next_pending_deadline(&pending);
+        tokio::select! {
+            maybe_command = command_rx.recv() => {
+                let command = match maybe_command {
+                    std::option::Option::Some(command) => command,
+                    std::option::Option::None => {
+                        ksp_logging_lib::trace!(target: crate::TRACING_TARGET, session_id = id.get(), "all WebSocket session handles dropped; closing actor");
+                        let _ = websocket.send(tokio_tungstenite::tungstenite::Message::Close(std::option::Option::None)).await;
+                        publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Closed, pending.len());
+                        fail_all_pending(&mut pending, id, crate::ERROR_CODE_WS_SESSION_CLOSED, "WebSocket session closed before pending response delivery");
+                        return;
+                    },
+                };
+                let command_outcome = handle_session_command(id, &endpoint, &mut websocket, &mut pending, &mut next_request_id, command).await;
+                if !command_outcome {
+                    publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Failed, pending.len());
+                    fail_all_pending(&mut pending, id, crate::ERROR_CODE_WS_CONNECTION_FAILED, "WebSocket connection failed while writing a request");
+                    return;
+                }
+                publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Active, pending.len());
+            },
+            maybe_message = websocket.next() => {
+                let keep_running = handle_socket_message(id, &endpoint, maybe_message, &mut websocket, &mut pending).await;
+                if !keep_running {
+                    publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Failed, pending.len());
+                    fail_all_pending(&mut pending, id, crate::ERROR_CODE_WS_CONNECTION_FAILED, "WebSocket connection ended before pending response delivery");
+                    return;
+                }
+                publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Active, pending.len());
+            },
+            () = tokio::time::sleep_until(timeout_deadline) => {
+                expire_pending_requests(id, &mut pending);
+                publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Active, pending.len());
+            },
+        }
+    }
+}
+
+async fn handle_session_command<S>(
+    id: crate::WsSessionId,
+    endpoint: &crate::WsEndpointSettings,
+    websocket: &mut tokio_tungstenite::WebSocketStream<S>,
+    pending: &mut std::collections::BTreeMap<u64, PendingWsRequest>,
+    next_request_id: &mut u64,
+    command: WsSessionCommand,
+) -> bool
+where
+    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + std::marker::Unpin,
+{
+    return match command {
+        WsSessionCommand::ExecuteJsonRpc { method, params, response_tx } => {
+            if pending.len() >= endpoint.session().max_pending_requests() {
+                let error = ksp_core_lib::Error::new(crate::ERROR_CODE_WS_BACKPRESSURE_OVERFLOW, "WebSocket pending JSON-RPC request capacity is exhausted")
+                    .with_context("session_id", id.get().to_string())
+                    .with_context("method", method);
+                let _ = response_tx.send(std::result::Result::Err(error));
+                ksp_logging_lib::warn!(
+                    target: crate::TRACING_TARGET,
+                    session_id = id.get(),
+                    pending_request_count = pending.len(),
+                    max_pending_requests = endpoint.session().max_pending_requests(),
+                    "rejected WebSocket JSON-RPC request because pending capacity is exhausted"
+                );
+                return true;
+            }
+            let request_id = *next_request_id;
+            let incremented = request_id.checked_add(1);
+            *next_request_id = match incremented {
+                std::option::Option::Some(value) => value,
+                std::option::Option::None => {
+                    let error = ksp_core_lib::Error::new(crate::ERROR_CODE_WS_PROTOCOL_ERROR, "WebSocket JSON-RPC request identifier space is exhausted")
+                        .with_context("session_id", id.get().to_string());
+                    let _ = response_tx.send(std::result::Result::Err(error));
+                    return true;
+                },
+            };
+            let request_result = crate::JsonRpcRequest::new(request_id, method, params);
+            let request = match request_result {
+                std::result::Result::Ok(request) => request,
+                std::result::Result::Err(error) => {
+                    let _ = response_tx.send(std::result::Result::Err(error));
+                    return true;
+                },
+            };
+            let payload_result = request.to_json_string();
+            let payload = match payload_result {
+                std::result::Result::Ok(payload) => payload,
+                std::result::Result::Err(error) => {
+                    let _ = response_tx.send(std::result::Result::Err(error));
+                    return true;
+                },
+            };
+            let deadline = tokio::time::Instant::now() + endpoint.session().command_timeout();
+            pending.insert(request_id, PendingWsRequest { method, deadline, response_tx });
+            ksp_logging_lib::trace!(
+                target: crate::TRACING_TARGET,
+                session_id = id.get(),
+                request_id,
+                method,
+                pending_request_count = pending.len(),
+                "sending WebSocket JSON-RPC request"
+            );
+            let send_result = websocket.send(tokio_tungstenite::tungstenite::Message::Text(payload.into())).await;
+            if send_result.is_err() {
+                ksp_logging_lib::warn!(
+                    target: crate::TRACING_TARGET,
+                    session_id = id.get(),
+                    request_id,
+                    method,
+                    "WebSocket JSON-RPC request write failed"
+                );
+                return false;
+            }
+            return true;
+        },
+    };
+}
+
+async fn handle_socket_message<S>(
+    id: crate::WsSessionId,
+    endpoint: &crate::WsEndpointSettings,
+    maybe_message: std::option::Option<std::result::Result<tokio_tungstenite::tungstenite::Message, tokio_tungstenite::tungstenite::Error>>,
+    websocket: &mut tokio_tungstenite::WebSocketStream<S>,
+    pending: &mut std::collections::BTreeMap<u64, PendingWsRequest>,
+) -> bool
+where
+    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + std::marker::Unpin,
+{
+    let message = match maybe_message {
+        std::option::Option::Some(std::result::Result::Ok(message)) => message,
+        std::option::Option::Some(std::result::Result::Err(_)) => {
+            ksp_logging_lib::warn!(
+                target: crate::TRACING_TARGET,
+                session_id = id.get(),
+                endpoint_name = endpoint.name(),
+                "physical WebSocket read failed"
+            );
+            return false;
+        },
+        std::option::Option::None => {
+            ksp_logging_lib::warn!(
+                target: crate::TRACING_TARGET,
+                session_id = id.get(),
+                endpoint_name = endpoint.name(),
+                "physical WebSocket stream ended"
+            );
+            return false;
+        },
+    };
+    return match message {
+        tokio_tungstenite::tungstenite::Message::Text(text) => handle_text_message(id, text.as_str(), pending),
+        tokio_tungstenite::tungstenite::Message::Binary(_) => {
+            ksp_logging_lib::warn!(target: crate::TRACING_TARGET, session_id = id.get(), "received unexpected binary WebSocket message");
+            false
+        },
+        tokio_tungstenite::tungstenite::Message::Ping(payload) => {
+            ksp_logging_lib::trace!(target: crate::TRACING_TARGET, session_id = id.get(), "received WebSocket ping control frame");
+            return websocket.send(tokio_tungstenite::tungstenite::Message::Pong(payload)).await.is_ok();
+        },
+        tokio_tungstenite::tungstenite::Message::Pong(_) => {
+            ksp_logging_lib::trace!(target: crate::TRACING_TARGET, session_id = id.get(), "received WebSocket pong control frame");
+            true
+        },
+        tokio_tungstenite::tungstenite::Message::Close(_) => {
+            ksp_logging_lib::debug!(target: crate::TRACING_TARGET, session_id = id.get(), "remote peer closed physical WebSocket session");
+            false
+        },
+        tokio_tungstenite::tungstenite::Message::Frame(_) => {
+            ksp_logging_lib::trace!(target: crate::TRACING_TARGET, session_id = id.get(), "ignored internal WebSocket frame event");
+            true
+        },
+    };
+}
+
+fn handle_text_message(id: crate::WsSessionId, text: &str, pending: &mut std::collections::BTreeMap<u64, PendingWsRequest>) -> bool {
+    let decoded = serde_json::from_str::<serde_json::Value>(text);
+    let value = match decoded {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => {
+            ksp_logging_lib::warn!(target: crate::TRACING_TARGET, session_id = id.get(), "received malformed JSON WebSocket message");
+            return false;
+        },
+    };
+    let object = match value.as_object() {
+        std::option::Option::Some(object) => object,
+        std::option::Option::None => {
+            ksp_logging_lib::warn!(target: crate::TRACING_TARGET, session_id = id.get(), "received non-object JSON WebSocket message");
+            return false;
+        },
+    };
+    if !object.contains_key("id") {
+        if object.contains_key("method") && object.contains_key("params") {
+            ksp_logging_lib::trace!(
+                target: crate::TRACING_TARGET,
+                session_id = id.get(),
+                "received WebSocket notification before subscription registry activation; safely ignored"
+            );
+            return true;
+        }
+        ksp_logging_lib::warn!(target: crate::TRACING_TARGET, session_id = id.get(), "received structurally invalid WebSocket JSON-RPC message");
+        return false;
+    }
+    let response_id = match object.get("id").and_then(serde_json::Value::as_u64) {
+        std::option::Option::Some(response_id) => response_id,
+        std::option::Option::None => {
+            ksp_logging_lib::warn!(target: crate::TRACING_TARGET, session_id = id.get(), "received WebSocket JSON-RPC response with invalid id");
+            return false;
+        },
+    };
+    let pending_request = match pending.remove(&response_id) {
+        std::option::Option::Some(pending_request) => pending_request,
+        std::option::Option::None => {
+            ksp_logging_lib::debug!(
+                target: crate::TRACING_TARGET,
+                session_id = id.get(),
+                response_id,
+                "ignored unknown or stale WebSocket JSON-RPC response id"
+            );
+            return true;
+        },
+    };
+    let parsed = crate::parse_json_rpc_response_value(value, response_id);
+    let (result, keep_running) = match parsed {
+        std::result::Result::Ok(response) => {
+            let result = match response.into_result() {
+                std::result::Result::Ok(value) => std::result::Result::Ok(value),
+                std::result::Result::Err(error) => std::result::Result::Err(error.with_context("method", pending_request.method)),
+            };
+            (result, true)
+        },
+        std::result::Result::Err(_) => {
+            let error = ksp_core_lib::Error::new(crate::ERROR_CODE_WS_PROTOCOL_ERROR, "WebSocket JSON-RPC response violates protocol invariants")
+                .with_context("session_id", id.get().to_string())
+                .with_context("request_id", response_id.to_string())
+                .with_context("method", pending_request.method);
+            (std::result::Result::Err(error), false)
+        },
+    };
+    let _ = pending_request.response_tx.send(result);
+    ksp_logging_lib::trace!(
+        target: crate::TRACING_TARGET,
+        session_id = id.get(),
+        response_id,
+        method = pending_request.method,
+        pending_request_count = pending.len(),
+        "dispatched WebSocket JSON-RPC response to pending request"
+    );
+    return keep_running;
+}
+
+fn next_pending_deadline(pending: &std::collections::BTreeMap<u64, PendingWsRequest>) -> tokio::time::Instant {
+    let mut earliest = std::option::Option::None::<tokio::time::Instant>;
+    for request in pending.values() {
+        earliest = match earliest {
+            std::option::Option::Some(current) if current <= request.deadline => std::option::Option::Some(current),
+            _ => std::option::Option::Some(request.deadline),
+        };
+    }
+    return match earliest {
+        std::option::Option::Some(deadline) => deadline,
+        std::option::Option::None => tokio::time::Instant::now() + std::time::Duration::from_secs(86_400),
+    };
+}
+
+fn expire_pending_requests(id: crate::WsSessionId, pending: &mut std::collections::BTreeMap<u64, PendingWsRequest>) {
+    let now = tokio::time::Instant::now();
+    let expired_ids = pending
+        .iter()
+        .filter_map(|(request_id, request)| if request.deadline <= now { std::option::Option::Some(*request_id) } else { std::option::Option::None })
+        .collect::<std::vec::Vec<_>>();
+    for request_id in expired_ids {
+        if let std::option::Option::Some(request) = pending.remove(&request_id) {
+            let error = ws_timeout_error(id, "WebSocket JSON-RPC request timed out while awaiting the remote response").with_context("method", request.method);
+            let _ = request.response_tx.send(std::result::Result::Err(error));
+            ksp_logging_lib::debug!(
+                target: crate::TRACING_TARGET,
+                session_id = id.get(),
+                request_id,
+                method = request.method,
+                "expired pending WebSocket JSON-RPC request"
+            );
+        }
+    }
+}
+
+fn fail_all_pending(
+    pending: &mut std::collections::BTreeMap<u64, PendingWsRequest>,
+    id: crate::WsSessionId,
+    code: ksp_core_lib::ErrorCode,
+    message: &'static str,
+) {
+    let requests = std::mem::take(pending);
+    for (request_id, request) in requests {
+        let error = ksp_core_lib::Error::new(code, message)
+            .with_context("session_id", id.get().to_string())
+            .with_context("request_id", request_id.to_string())
+            .with_context("method", request.method);
+        let _ = request.response_tx.send(std::result::Result::Err(error));
+    }
+}
+
+fn publish_snapshot(
+    snapshot_tx: &tokio::sync::watch::Sender<crate::WsSessionSnapshot>,
+    id: crate::WsSessionId,
+    endpoint: &crate::WsEndpointSettings,
+    state: crate::WsSessionState,
+    pending_request_count: usize,
+) {
+    let snapshot = crate::WsSessionSnapshot::new(
+        id,
+        endpoint.name(),
+        endpoint.provider().clone(),
+        endpoint.cluster().clone(),
+        endpoint.protocol(),
+        state,
+        pending_request_count,
+        0,
+        0,
+        std::vec::Vec::new(),
+    );
+    snapshot_tx.send_replace(snapshot);
+}
+
+fn next_session_id() -> ksp_core_lib::Result<crate::WsSessionId> {
+    let update_result =
+        NEXT_WS_SESSION_ID.fetch_update(std::sync::atomic::Ordering::Relaxed, std::sync::atomic::Ordering::Relaxed, |current| current.checked_add(1));
+    let value = match update_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => {
+            return std::result::Result::Err(ksp_core_lib::Error::new(crate::ERROR_CODE_WS_PROTOCOL_ERROR, "WebSocket session identity space is exhausted"));
+        },
+    };
+    return match std::num::NonZeroU64::new(value) {
+        std::option::Option::Some(value) => std::result::Result::Ok(crate::WsSessionId::new(value)),
+        std::option::Option::None => {
+            std::result::Result::Err(ksp_core_lib::Error::new(crate::ERROR_CODE_WS_PROTOCOL_ERROR, "WebSocket session identity generator produced zero"))
+        },
+    };
+}
+
+fn ws_connection_error(id: crate::WsSessionId, endpoint: &crate::WsEndpointSettings, message: &'static str) -> ksp_core_lib::Error {
+    return ksp_core_lib::Error::new(crate::ERROR_CODE_WS_CONNECTION_FAILED, message)
+        .with_context("session_id", id.get().to_string())
+        .with_context("endpoint_name", endpoint.name())
+        .with_context("provider", endpoint.provider().as_str())
+        .with_context("cluster", endpoint.cluster().as_str());
+}
+
+fn ws_session_closed_error(id: crate::WsSessionId, message: &'static str) -> ksp_core_lib::Error {
+    return ksp_core_lib::Error::new(crate::ERROR_CODE_WS_SESSION_CLOSED, message).with_context("session_id", id.get().to_string());
+}
+
+fn ws_timeout_error(id: crate::WsSessionId, message: &'static str) -> ksp_core_lib::Error {
+    return ksp_core_lib::Error::new(crate::ERROR_CODE_TIMEOUT, message).with_context("session_id", id.get().to_string());
+}
+
+#[cfg(test)]
+#[path = "../unit_tests/ws_session.rs"]
+mod tests;