feat(remote_device_bridge): ITransport with TCP and WebSocket

src/plugins/remote_device_bridge/AGENTS.md

@@ -7,5 +7,6 @@ SPDX-License-Identifier: Apache-2.0
 
 **CRITICAL:** Complete the mandatory `AGENTS.md` preflight in [`../../../AGENTS.md`](../../../AGENTS.md) before editing here.
 
-- `session/protocol_frame.*` and `session/protocol_payloads.*` are the only serialization points. Every frame is encoded and decoded there and nowhere else; changing the wire format means changing them and `tests/cpp/plugins/remote_device_bridge/test_wire_codec.cpp` together.
+- `session/protocol_frame.*` and `session/protocol_payloads.*` are the only serialization points. Every frame is encoded and decoded there and nowhere else; changing the wire format means changing them and `tests/cpp/plugins/remote_device_bridge/test_wire_codec.cpp` together. Sessions hold an `ITransport` (`session/inc/remote_plugin_session/transport.hpp`) and only call `send/receive/shutdown` on complete frames; TCP framing lives in `TcpTransport`.
+- WebSocket is a second transport, not a second protocol: `WebSocketTransport` carries exactly one wire frame per binary message (no TLS; terminate `wss://` in a proxy). Clients pick it with `--remote=ws://host:port[/path]` via `parse_remote`/`connect_transport`; the bridge serves it on a separate `--ws-listen` port sharing the fork-per-client loop and `--max-clients`. Never sniff the protocol on the TCP port. The server-side HTTP upgrade runs in the first `receive()` so the accept loop never blocks on a client.
 - `Teleop::remote_plugin_session` links only `pusherio::pusherio` (OpenXR value types via headers). Never add `OpenXR::openxr_loader`, `Teleop::plugin_utils` or DeviceIO session libraries to it: a plugin using `--remote` must run without an OpenXR runtime.

src/plugins/remote_device_bridge/session/CMakeLists.txt

@@ -9,6 +9,15 @@ add_library(remote_plugin_session STATIC
     protocol_io.hpp
     inc/remote_plugin_session/protocol_frame.hpp
     inc/remote_plugin_session/protocol_payloads.hpp
+    remote_address.cpp
+    inc/remote_plugin_session/remote_address.hpp
+    tcp_socket.cpp
+    inc/remote_plugin_session/tcp_socket.hpp
+    tcp_transport.cpp
+    inc/remote_plugin_session/tcp_transport.hpp
+    inc/remote_plugin_session/transport.hpp
+    websocket_transport.cpp
+    inc/remote_plugin_session/websocket_transport.hpp
 )
 
 target_include_directories(remote_plugin_session

src/plugins/remote_device_bridge/session/inc/remote_plugin_session/remote_address.hpp

@@ -0,0 +1,35 @@
+// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include "tcp_socket.hpp"
+#include "transport.hpp"
+
+#include <chrono>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+
+namespace remote_plugin_session
+{
+
+//! Parses "host[:port]" or "[v6-host][:port]"; a missing port falls back to default_port. Throws
+//! std::invalid_argument.
+Endpoint parse_endpoint(const std::string& text, uint16_t default_port);
+
+//! Where a client connects: plain "host[:port]" (TCP) or "ws://host[:port][/path]" (WebSocket).
+struct RemoteAddress
+{
+    Endpoint endpoint;
+    std::optional<std::string> ws_path; //!< set for ws://
+};
+
+//! Throws std::invalid_argument; wss:// is refused (terminate TLS in a proxy in front of the bridge).
+RemoteAddress parse_remote(const std::string& text, uint16_t default_port);
+//! Connects (and for ws:// completes the upgrade) within `timeout`; throws std::runtime_error.
+std::unique_ptr<ITransport> connect_transport(const RemoteAddress& remote,
+                                              std::chrono::milliseconds timeout = kConnectTimeout);
+
+} // namespace remote_plugin_session

src/plugins/remote_device_bridge/session/inc/remote_plugin_session/tcp_socket.hpp

@@ -0,0 +1,76 @@
+// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include "transport.hpp"
+
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+namespace remote_plugin_session
+{
+
+struct Endpoint
+{
+    std::string host;
+    uint16_t port = 0;
+};
+
+//! Minimal RAII wrapper over a blocking POSIX TCP socket (IPv4/IPv6 via getaddrinfo).
+class TcpSocket
+{
+public:
+    TcpSocket() = default;
+    explicit TcpSocket(int fd);
+    ~TcpSocket();
+
+    TcpSocket(TcpSocket&& other) noexcept;
+    TcpSocket& operator=(TcpSocket&& other) noexcept;
+    TcpSocket(const TcpSocket&) = delete;
+    TcpSocket& operator=(const TcpSocket&) = delete;
+
+    //! Throws std::runtime_error when no address connects within the timeout.
+    static TcpSocket connect(const Endpoint& endpoint, std::chrono::milliseconds timeout);
+    //! Throws std::runtime_error when bind/listen fails. Port 0 picks an ephemeral port.
+    static TcpSocket listen(const Endpoint& endpoint, int backlog);
+    //! Returns an invalid socket when the timeout expires; throws on a hard error. `peer`, when given, receives
+    //! the client's numeric host:port.
+    TcpSocket accept(std::chrono::milliseconds timeout, std::string* peer = nullptr);
+
+    bool valid() const;
+    int fd() const;
+    uint16_t local_port() const;
+
+    //! False when the peer closed or the send stalled past the socket send timeout.
+    bool send_all(const uint8_t* data, size_t size);
+    //! Timeout applies to the wait for the first byte; a stall mid-message reports Closed.
+    //! Ok, Timeout or Closed.
+    ReceiveResult recv_all(uint8_t* data, size_t size, std::chrono::milliseconds timeout);
+
+    //! Unblocks recv/send on other threads; the descriptor stays open until close().
+    void shutdown();
+    void close();
+
+private:
+    //! Numeric "host:port" of the connected peer; "?" when unavailable.
+    std::string peer_name() const;
+
+    int m_fd = -1;
+};
+
+//! Accepts one connection on `listener` and wraps it in `Transport`; null when the poll expires. `peer`, when
+//! given, receives the client's numeric host:port.
+template <typename Transport>
+std::unique_ptr<ITransport> accept_transport(TcpSocket& listener, std::chrono::milliseconds poll, std::string* peer)
+{
+    TcpSocket client = listener.accept(poll, peer);
+    return client.valid() ? std::make_unique<Transport>(std::move(client)) : nullptr;
+}
+
+} // namespace remote_plugin_session

src/plugins/remote_device_bridge/session/inc/remote_plugin_session/tcp_transport.hpp

@@ -0,0 +1,39 @@
+// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include "tcp_socket.hpp"
+#include "transport.hpp"
+
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <vector>
+
+namespace remote_plugin_session
+{
+
+//! ITransport over one TCP stream; messages are delimited by the wire header's payload_size.
+class TcpTransport final : public ITransport
+{
+public:
+    explicit TcpTransport(TcpSocket socket);
+
+    //! One transport per connection accepted on `listener`; null when the poll expires. `peer`, when given,
+    //! receives the client's numeric host:port.
+    static std::unique_ptr<ITransport> accept(TcpSocket& listener,
+                                              std::chrono::milliseconds poll,
+                                              std::string* peer = nullptr);
+
+    bool send(const uint8_t* frame, size_t size) override;
+    ReceiveResult receive(std::vector<uint8_t>& frame, std::chrono::milliseconds timeout) override;
+    void shutdown() override;
+
+private:
+    TcpSocket m_socket;
+};
+
+} // namespace remote_plugin_session

src/plugins/remote_device_bridge/session/inc/remote_plugin_session/transport.hpp

@@ -0,0 +1,53 @@
+// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <vector>
+
+namespace remote_plugin_session
+{
+
+//! Default connect/upgrade timeout for client transports.
+constexpr std::chrono::milliseconds kConnectTimeout{ 3000 };
+//! A frame that has started arriving must finish within this window, or the peer counts as gone.
+constexpr std::chrono::milliseconds kMidMessageStall{ 3000 };
+
+enum class ReceiveResult
+{
+    Ok,
+    Timeout,
+    Closed,
+    Malformed,
+};
+
+/*!
+ * @brief Message transport between a RemotePluginSession and the bridge.
+ *
+ * A frame is one complete wire message (protocol kHeaderSize header + payload)
+ * as built by protocol::encode_frame. send() and receive() may run concurrently on two
+ * threads; shutdown() unblocks both. Framing (TCP is a stream) lives in the
+ * implementation, not in the callers.
+ */
+class ITransport
+{
+public:
+    virtual ~ITransport() = default;
+
+    //! False once the peer is gone or a send stalled past the transport's timeout.
+    virtual bool send(const uint8_t* frame, size_t size) = 0;
+    //! Timeout covers the wait for the first byte; Ok means `frame` holds exactly one complete message.
+    virtual ReceiveResult receive(std::vector<uint8_t>& frame, std::chrono::milliseconds timeout) = 0;
+    virtual void shutdown() = 0;
+    //! Server side: finishes the transport's own handshake so send() works without waiting for a frame. Ok at once
+    //! when there is none (TCP).
+    virtual ReceiveResult wait_ready(std::chrono::milliseconds /*timeout*/)
+    {
+        return ReceiveResult::Ok;
+    }
+};
+
+} // namespace remote_plugin_session

src/plugins/remote_device_bridge/session/inc/remote_plugin_session/websocket_transport.hpp

@@ -0,0 +1,72 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include "tcp_socket.hpp"
+#include "transport.hpp"
+
+#include <atomic>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <random>
+#include <string>
+#include <vector>
+
+namespace remote_plugin_session
+{
+
+/*!
+ * @brief RFC 6455 transport: each wire frame travels as one binary WebSocket message. No TLS, no extensions.
+ *
+ * The server role answers the HTTP upgrade inside its first receive(), so an accept loop never waits on a
+ * slow client; send() fails until the upgrade is done. Incoming fragments are reassembled, pings answered.
+ */
+class WebSocketTransport final : public ITransport
+{
+public:
+    //! Server role over an accepted socket; no I/O until the first receive() or wait_ready().
+    explicit WebSocketTransport(TcpSocket socket);
+
+    //! Client role: sends the upgrade for `path` with `host_header` ("host:port") and throws
+    //! std::runtime_error unless the server answers 101.
+    WebSocketTransport(TcpSocket socket,
+                       const std::string& host_header,
+                       const std::string& path,
+                       std::chrono::milliseconds timeout = kConnectTimeout);
+
+    //! Like TcpTransport::accept; the upgrade happens on the first receive().
+    static std::unique_ptr<ITransport> accept(TcpSocket& listener,
+                                              std::chrono::milliseconds poll,
+                                              std::string* peer = nullptr);
+
+    bool send(const uint8_t* frame, size_t size) override;
+    ReceiveResult receive(std::vector<uint8_t>& frame, std::chrono::milliseconds timeout) override;
+    void shutdown() override;
+    //! Server: answers the HTTP upgrade if receive() has not yet.
+    ReceiveResult wait_ready(std::chrono::milliseconds timeout) override;
+
+private:
+    enum class Role
+    {
+        Client,
+        Server,
+    };
+
+    WebSocketTransport(TcpSocket socket, Role role);
+    void client_upgrade(const std::string& host_header, const std::string& path, std::chrono::milliseconds timeout);
+    bool send_message(uint8_t opcode, const uint8_t* data, size_t size);
+    ReceiveResult accept_upgrade(std::chrono::milliseconds timeout);
+
+    TcpSocket m_socket;
+    Role m_role;
+    std::atomic<bool> m_open{ false };
+    std::mutex m_send_mutex; // receive() answers pings and closes concurrently with send()
+    std::vector<uint8_t> m_send_buffer;
+    std::mt19937 m_mask_rng;
+};
+
+} // namespace remote_plugin_session