feat(remote_device_bridge): RemotePluginSession over ITransport

src/plugins/remote_device_bridge/AGENTS.md

@@ -9,4 +9,7 @@ SPDX-License-Identifier: Apache-2.0
 
 - `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.
+- Never add `OpenXR::openxr_loader`, `Teleop::plugin_utils` or DeviceIO session libraries to `Teleop::remote_plugin_session`: a plugin using `--remote` must run without an OpenXR runtime.
+- ERROR on channel 0 ends the session; ERROR on a channel id refuses only that OPEN and the session and its other channels continue. Refuse per channel whenever the session itself is still sound.
+- `RemotePluginSession` never throws on the sample path for transport failures (`push`, `push_buffer`, `query`, `update`): after one they become no-ops; only channel creation and the constructor throw. Caller bugs (null joints or buffer) still throw `std::invalid_argument`.
+- Timestamps cross the wire in the sender's `CLOCK_MONOTONIC`; the receiver maps them with `ClockOffsetEstimator`, which prefers PING/PONG exchanges (minimum RTT) over the one-way minimum of `local_recv - remote_send`; the one-way minimum hides the minimum network delay. Do not push raw remote timestamps into OpenXR.

src/plugins/remote_device_bridge/session/CMakeLists.txt

@@ -3,14 +3,20 @@
 
 cmake_minimum_required(VERSION 3.20)
 
+find_package(Threads REQUIRED)
+
 add_library(remote_plugin_session STATIC
+    clock_offset_estimator.cpp
+    inc/remote_plugin_session/clock_offset_estimator.hpp
     protocol_frame.cpp
     protocol_payloads.cpp
     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
+    remote_plugin_session.cpp
+    inc/remote_plugin_session/remote_plugin_session.hpp
     tcp_socket.cpp
     inc/remote_plugin_session/tcp_socket.hpp
     tcp_transport.cpp
@@ -28,6 +34,9 @@ target_include_directories(remote_plugin_session
 target_link_libraries(remote_plugin_session
     PUBLIC
         pusherio::pusherio
+    PRIVATE
+        log_bridge::log_bridge_core
+        Threads::Threads
 )
 
 add_library(Teleop::remote_plugin_session ALIAS remote_plugin_session)

src/plugins/remote_device_bridge/session/clock_offset_estimator.cpp

@@ -0,0 +1,123 @@
+// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+#include "inc/remote_plugin_session/clock_offset_estimator.hpp"
+
+#include <algorithm>
+#include <stdexcept>
+
+namespace remote_plugin_session
+{
+
+ClockOffsetEstimator::ClockOffsetEstimator(size_t max_samples, int64_t window_ns, size_t two_way_window)
+    : m_max_samples(max_samples), m_window_ns(window_ns), m_two_way_window(two_way_window)
+{
+    if (max_samples == 0 || window_ns <= 0 || two_way_window == 0)
+    {
+        throw std::invalid_argument("ClockOffsetEstimator requires a non-empty window");
+    }
+}
+
+void ClockOffsetEstimator::add_sample(int64_t remote_t_send_ns, int64_t local_t_recv_ns)
+{
+    m_samples.push_back(Sample{ local_t_recv_ns, local_t_recv_ns - remote_t_send_ns });
+
+    while (m_samples.size() > m_max_samples)
+    {
+        m_samples.pop_front();
+    }
+    while (m_samples.front().local_t_recv_ns < local_t_recv_ns - m_window_ns)
+    {
+        m_samples.pop_front();
+    }
+
+    m_offset_ns = std::min_element(m_samples.begin(), m_samples.end(),
+                                   [](const Sample& a, const Sample& b) { return a.offset_ns < b.offset_ns; })
+                      ->offset_ns;
+}
+
+void ClockOffsetEstimator::add_two_way_sample(int64_t t1_ns, int64_t t2_ns, int64_t t3_ns, int64_t t4_ns)
+{
+    const int64_t rtt_ns = (t4_ns - t1_ns) - (t3_ns - t2_ns);
+    if (rtt_ns < 0)
+    {
+        return;
+    }
+    // local - remote, the same sign as the one-way offset.
+    const int64_t offset_ns = ((t1_ns - t2_ns) + (t4_ns - t3_ns)) / 2;
+    m_exchanges.push_back(Exchange{ rtt_ns, offset_ns });
+    if (m_exchanges.size() > m_two_way_window)
+    {
+        m_exchanges.pop_front();
+    }
+}
+
+const ClockOffsetEstimator::Exchange& ClockOffsetEstimator::best_exchange() const
+{
+    return *std::min_element(m_exchanges.begin(), m_exchanges.end(),
+                             [](const Exchange& a, const Exchange& b) { return a.rtt_ns < b.rtt_ns; });
+}
+
+int64_t ClockOffsetEstimator::offset_ns() const
+{
+    return m_exchanges.empty() ? m_offset_ns : best_exchange().offset_ns;
+}
+
+int64_t ClockOffsetEstimator::remote_to_local(int64_t remote_ns) const
+{
+    return remote_ns + offset_ns();
+}
+
+int64_t ClockOffsetEstimator::rtt_min_ns() const
+{
+    return m_exchanges.empty() ? 0 : best_exchange().rtt_ns;
+}
+
+int64_t ClockOffsetEstimator::rtt_last_ns() const
+{
+    return m_exchanges.empty() ? 0 : m_exchanges.back().rtt_ns;
+}
+
+ClockOffsetEstimator::Source ClockOffsetEstimator::offset_source() const
+{
+    return m_exchanges.empty() ? Source::OneWay : Source::TwoWay;
+}
+
+PingScheduler::PingScheduler(int64_t start_ns) : m_start_ns(start_ns), m_next_ns(start_ns)
+{
+}
+
+PingScheduler::Action PingScheduler::poll(int64_t now_ns)
+{
+    if (m_state == State::GaveUp)
+    {
+        return Action::None;
+    }
+    if (m_state == State::Waiting && now_ns - m_start_ns >= kPongTimeoutNs)
+    {
+        m_state = State::GaveUp;
+        return Action::GiveUp;
+    }
+    if (now_ns < m_next_ns)
+    {
+        return Action::None;
+    }
+    ++m_sent;
+    m_next_ns = now_ns + (m_sent < kBurstCount ? kBurstPeriodNs : kSteadyPeriodNs);
+    return Action::Ping;
+}
+
+void PingScheduler::on_pong()
+{
+    if (m_state == State::Waiting)
+    {
+        m_state = State::GotPong;
+    }
+}
+
+bool PingScheduler::active() const
+{
+    return m_state != State::GaveUp;
+}
+
+} // namespace remote_plugin_session

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

@@ -0,0 +1,118 @@
+// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <deque>
+
+namespace remote_plugin_session
+{
+
+/*!
+ * @brief Maps a remote monotonic clock onto the local one (local ~= remote + offset).
+ *
+ * One-way: fed with (remote send time, local receive time) of every received
+ * frame; the offset is the running minimum of (local - remote) over a sliding
+ * window, which hides one minimum one-way delay. Two-way: once a PING/PONG
+ * exchange arrives the offset comes from the minimum-RTT exchange of the
+ * two-way window instead (NTP filter; error = path asymmetry / 2) and one-way
+ * samples no longer move it. Not thread-safe; the owner serializes access.
+ */
+class ClockOffsetEstimator
+{
+public:
+    enum class Source
+    {
+        OneWay,
+        TwoWay, //!< once an exchange was added
+    };
+
+    //! One-way: at most `max_samples` within `window_ns`; two-way: the last `two_way_window` exchanges.
+    explicit ClockOffsetEstimator(size_t max_samples = 200, int64_t window_ns = 5'000'000'000, size_t two_way_window = 32);
+
+    void add_sample(int64_t remote_t_send_ns, int64_t local_t_recv_ns);
+    //! t1/t4: local PING send / PONG receive; t2/t3: remote PING receive / PONG send. Negative RTTs are ignored.
+    void add_two_way_sample(int64_t t1_ns, int64_t t2_ns, int64_t t3_ns, int64_t t4_ns);
+
+    //! Zero until the first sample arrives.
+    int64_t offset_ns() const;
+    int64_t remote_to_local(int64_t remote_ns) const;
+    //! Zero until the first two-way exchange.
+    int64_t rtt_min_ns() const;
+    int64_t rtt_last_ns() const;
+    Source offset_source() const;
+
+private:
+    struct Sample
+    {
+        int64_t local_t_recv_ns;
+        int64_t offset_ns;
+    };
+    struct Exchange
+    {
+        int64_t rtt_ns;
+        int64_t offset_ns;
+    };
+
+    const Exchange& best_exchange() const;
+
+    size_t m_max_samples;
+    int64_t m_window_ns;
+    size_t m_two_way_window;
+    std::deque<Sample> m_samples;
+    int64_t m_offset_ns = 0;
+    std::deque<Exchange> m_exchanges;
+};
+
+/*!
+ * @brief When the bridge sends PINGs: a burst right after HELLO_ACK, then 1 Hz.
+ *
+ * Without any PONG within kPongTimeout of the start, poll() returns GiveUp once
+ * and nothing after that: the peer stays on the one-way estimate.
+ */
+class PingScheduler
+{
+public:
+    enum class Action
+    {
+        None,
+        Ping,
+        GiveUp,
+    };
+
+    static constexpr int kBurstCount = 10;
+    static constexpr int64_t kBurstPeriodNs = 50'000'000;
+    static constexpr int64_t kSteadyPeriodNs = 1'000'000'000;
+    static constexpr int64_t kPongTimeoutNs = 2'000'000'000;
+
+    explicit PingScheduler(int64_t start_ns);
+
+    Action poll(int64_t now_ns);
+    void on_pong();
+    //! False after GiveUp; late PONGs are then ignored.
+    bool active() const;
+
+private:
+    enum class State
+    {
+        Waiting,
+        GotPong,
+        GaveUp,
+    };
+
+    int64_t m_start_ns;
+    int64_t m_next_ns;
+    int m_sent = 0;
+    State m_state = State::Waiting;
+};
+
+//! True when a mapped sample time is older than now - max_age; max_age <= 0 disables the guard. After a link
+//! stall the backlog arrives in a burst; pushing it into OpenXR would replay old motion instead of skipping it.
+inline bool should_drop_stale(int64_t mapped_sample_ns, int64_t now_ns, int64_t max_age_ns)
+{
+    return max_age_ns > 0 && now_ns - mapped_sample_ns > max_age_ns;
+}
+
+} // namespace remote_plugin_session

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

@@ -0,0 +1,79 @@
+// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include "remote_address.hpp"
+#include "transport.hpp"
+
+#include <pusherio/plugin_session.hpp>
+
+#include <cstdint>
+#include <map>
+#include <memory>
+#include <string>
+
+namespace remote_plugin_session
+{
+
+namespace detail
+{
+class RemotePluginSessionImpl;
+}
+
+/*!
+ * @brief Plugin session that forwards every channel over an ITransport to a remote_device_bridge.
+ *
+ * Same contract as the OpenXR plugin session: the session owns its pull channel
+ * (opened with the session, with the wrist source HELLO declared) and the owner
+ * destroys all push channels before the session. After the transport
+ * drops, pushes become silent no-ops and channels stay safe to destroy.
+ * Each push channel has a small latest-wins queue; on a stalled link the oldest
+ * samples of that channel are dropped, other channels are unaffected. A channel
+ * the bridge refuses (ERROR on its channel id) turns into a no-op; the session
+ * and its other channels continue. The pull channel exposes no trackers; wrist
+ * poses are the latest bridge samples (no interpolation).
+ */
+class RemotePluginSession final : public core::IPluginSession
+{
+public:
+    struct Stats
+    {
+        uint64_t sent_hand = 0;
+        uint64_t sent_schema = 0;
+        uint64_t wrist_rx = 0;
+        uint64_t heartbeats = 0;
+        uint64_t dropped_queue = 0; //!< samples evicted by the per-channel latest-wins queues
+        uint64_t tx_bytes = 0;
+        std::map<uint16_t, uint64_t> dropped_by_channel; //!< open push channels only
+        std::map<uint16_t, std::string> refused_channels; //!< channel id -> the bridge's reason
+    };
+
+    //! Sends HELLO over a connected transport (see connect_transport()) and waits for HELLO_ACK; throws
+    //! std::runtime_error on any failure.
+    RemotePluginSession(std::unique_ptr<ITransport> transport,
+                        std::string app_name,
+                        core::PluginSessionRequirements requirements);
+    ~RemotePluginSession() override;
+
+    core::IPluginPullChannel& pull_channel() override;
+    std::unique_ptr<core::ISchemaPushChannel> create_schema_push_channel(const core::SchemaPusherConfig& config) override;
+    std::unique_ptr<core::IHandTrackingPushChannel> create_hand_tracking_push_channel(XrHandEXT hand) override;
+
+    //! False once the transport failed; channel creation then throws and pushes are dropped.
+    bool connected() const;
+    //! Estimated bridge-to-local clock offset in ns (local ~= bridge + offset), measured from HELLO_ACK onward.
+    int64_t clock_offset_ns() const;
+    //! Cumulative counters; their deltas are also logged periodically.
+    Stats stats() const;
+
+private:
+    std::unique_ptr<detail::RemotePluginSessionImpl> m_impl;
+    std::unique_ptr<core::IPluginPullChannel> m_pull_channel;
+};
+
+//! Connects each session it builds to `remote`. The remote session exposes no trackers, so the factory throws
+//! std::logic_error when handed any.
+core::PluginSessionFactory make_remote_plugin_session_factory(std::string app_name, RemoteAddress remote);
+
+} // namespace remote_plugin_session