fix(session): recover stalled video subscriptions with a keyframe-forcing watchdog

Camera sources in the OBS plugin were dropping out at random and never
recovering, while the same players stayed healthy in browser talkback.
Measured on the live server: every OBS plugin subscriber racked up
862-1099 nackMisses (retransmit requests for packets the SFU had already
aged out of its send buffer -- unrecoverable loss) and sat at plis == 2
for a multi-hour session, versus 0/adaptive-PLI for a browser subscriber
in the same room. The pinned client-sdk-cpp (1.10.1) exposes no
PLI/keyframe-request API at all, so a decoder that lost a frame that way
had no way to resync for the rest of the show.

Add a stall-recovery watchdog: StallWatchdog (session_types.h/.cpp) is a
pure, fake-clock-testable class that decides when a video subscription
has gone too long (2000ms, kStallRecoveryTimeout) without a decoded
frame reaching OBS. LiveKitSession::Impl polls it from a dedicated
thread and, through the existing command queue (never touching the SDK
off the worker thread), toggles the publication's setEnabled(false)/
setEnabled(true) -- the one lever this SDK exposes that makes the SFU
restart delivery, and a restart always begins with a keyframe.

Repeated attempts against the same stall back off exponentially
(2s/4s/8s/16s/30s-capped, mirroring the shape of the adapter's own
reconnect backoff) so a genuinely gone publisher is retried on a bounded
cadence instead of hammered every 2 seconds. A muted, disabled or
unsubscribed track, an audio-only source, or a disconnected session
never arms the watchdog: new onTrackMuted/onTrackUnmuted handlers
suspend and resume its clock, always re-baselining from "now" rather
than a stale timestamp, so un-muting after a long legitimate camera-off
period cannot read as a multi-minute stall. Each attempt, and eventual
recovery, is logged through a new DiagnosticHandler that the OBS adapter
maps onto obs_log at the same severities the file already uses for other
notable events.

Also investigated (not changed): the setVideoQuality(HIGH) pin added for
an earlier simulcast-resize bug. The SDK's own docs and the FFI binary's
wire-protocol strings show setVideoQuality only bounds spatial/simulcast
layer selection (UpdateTrackSettings.quality), never temporal layers or
frame rate, which the SFU's own congestion control governs independently
-- so this pin is unlikely to be the cause of the packet-loss symptom,
and is probably an inert no-op now that publishing is pinned server-side
to a single spatial layer (L1T3). Left in place since that is not fully
unambiguous from the SDK alone. Full writeup in
.stall-recovery-report.md (untracked, not part of this commit).

Adds 3 new headless StallWatchdog tests (fires-after-threshold, does-
not-fire-when-muted/disabled, backs-off-rather-than-loops) to
test_session.cpp; caught a real bug during development where
onFrameDelivered() reset the backoff duration but not the
next-attempt-allowed timestamp, throttling a just-recovered stream
against its own stale backoff. Built and all 6 CTest suites pass
(124/124 checks in test_session); also verified clean under
ThreadSanitizer with warning counts unchanged from the pre-change
baseline.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

core/include/stplugin/session.h

@@ -11,6 +11,7 @@ You may obtain a copy of the License at
 
 #pragma once
 
+#include <chrono>
 #include <memory>
 #include <string>
 
@@ -54,6 +55,22 @@ struct SessionConfig {
     int connect_timeout_ms = 15000;
 };
 
+/// How long the stall-recovery watchdog waits for a decoded video frame
+/// before treating the subscription as stalled and forcing a fresh
+/// keyframe (see StallWatchdog's comment in session_types.h for why this
+/// exists -- unrecoverable packet loss with no PLI/keyframe-request API in
+/// the pinned SDK). Long enough that ordinary jitter never trips it (a
+/// healthy 30fps subscription delivers a frame at least every ~33ms);
+/// short enough a director barely has time to notice before it recovers.
+constexpr std::chrono::milliseconds kStallRecoveryTimeout{2000};
+
+/// Ceiling for the backoff between repeated recovery attempts against the
+/// SAME stall. Starts at kStallRecoveryTimeout and doubles each attempt, so
+/// a genuinely gone publisher (crashed encoder, dead upstream network) is
+/// retried every 2s, 4s, 8s, ... 30s rather than hammered every 2 seconds
+/// for the rest of the show.
+constexpr std::chrono::milliseconds kStallRecoveryMaxBackoff{30000};
+
 /// Wraps livekit::Room for exactly one subscribed slot.
 ///
 /// Threading contract, which the OBS adapter depends on:
@@ -67,6 +84,11 @@ struct SessionConfig {
 ///     call.
 ///   - The state handler is invoked from whichever thread observed the
 ///     change. It must not block and must not call back into this object.
+///   - The diagnostic handler (currently just the stall-recovery watchdog,
+///     see kStallRecoveryTimeout below) may be invoked from the internal
+///     command-queue worker thread or a video reader thread. Same rules as
+///     the state handler: must not block, must not call back into this
+///     object.
 ///   - All handlers must be installed before connect(); they are not
 ///     synchronised against a running session.
 class LiveKitSession {
@@ -80,6 +102,7 @@ public:
     void setVideoHandler(VideoFrameHandler handler);
     void setAudioHandler(AudioFrameHandler handler);
     void setStateHandler(SessionStateHandler handler);
+    void setDiagnosticHandler(DiagnosticHandler handler);
 
     /// Connect and start subscribing. Returns true once the room is up; the
     /// selected slot's tracks may still arrive later (or not at all, if the

core/include/stplugin/session_types.h

@@ -19,6 +19,7 @@ You may obtain a copy of the License at
 // self-consistent -- all live here, and LiveKitSession is the (much thinner)
 // piece that wires real SDK callbacks into them.
 
+#include <chrono>
 #include <cstddef>
 #include <cstdint>
 #include <functional>
@@ -137,6 +138,82 @@ private:
     bool has_audio_ = false;
 };
 
+// ---------------------------------------------------------------------------
+// Stall-recovery watchdog (pure timing/decision logic)
+// ---------------------------------------------------------------------------
+
+/// Decides WHEN to force a fresh keyframe on an already-subscribed video
+/// track. It does not touch LiveKit or OBS at all -- LiveKitSession::Impl
+/// (session.cpp) is what actually carries the decision out
+/// (RemoteTrackPublication::setEnabled(false) then setEnabled(true)), and
+/// only from its SDK command-queue thread, the same rule every other SDK
+/// interaction in that file already follows.
+///
+/// Why this exists: measured on the live server, comparing subscribers in
+/// the same LiveKit room over the same 30-minute window, every OBS plugin
+/// connection racked up ~862-1099 nackMisses and ~2200-3100 nackRepeated --
+/// nackMisses means the subscriber asked the SFU to retransmit a packet
+/// that had already aged out of its send buffer, i.e. unrecoverable loss --
+/// while a browser subscriber in the same room saw 0 and 0. A decoder that
+/// loses a frame that way cannot resync without a fresh keyframe. Every OBS
+/// plugin connection also sat at `plis` == 2 for a multi-hour session (a
+/// browser adapts and asks for keyframes normally), and grepping the pinned
+/// client-sdk-cpp (1.10.1) headers turns up no PLI/keyframe-request API at
+/// all. So today, once that happens, the source just stays broken for the
+/// rest of the show -- "drops at random and never recovers". Toggling the
+/// subscription off and back on is the one lever this SDK exposes that
+/// forces the SFU to stop and restart delivery of the track, and a restart
+/// always begins with a keyframe. This class is the "have we gone too long
+/// without a frame, and is it still worth trying again" clock behind that
+/// lever; see LiveKitSession::Impl::recoverVideo() for where it is pulled.
+///
+/// Not thread-safe on its own, deliberately -- same contract as
+/// SessionStateMachine above: LiveKitSession::Impl owns the lock (in
+/// practice the same state_mutex that guards `machine`).
+class StallWatchdog {
+public:
+    StallWatchdog(std::chrono::milliseconds timeout, std::chrono::milliseconds max_backoff);
+
+    /// Call whenever whether a frame could legitimately arrive right now
+    /// changes: true once a video track is subscribed (and unmuted), false
+    /// on detach/unsubscribe/mute/disconnect. Flipping to false always
+    /// clears all timing state; flipping back to true always starts a
+    /// brand-new grace period rather than measuring from a stale timestamp.
+    /// That is what stops an unmute -- or an ordinary publisher swap --
+    /// from firing the INSTANT it resumes, off a "last frame" that might
+    /// actually be minutes old: a muted, disabled or unsubscribed track, an
+    /// audio-only source, or a disconnected session must never trip this.
+    void setExpectingFrames(bool expecting, std::chrono::steady_clock::time_point now);
+
+    /// Call every time a decoded video frame is actually delivered.
+    void onFrameDelivered(std::chrono::steady_clock::time_point now);
+
+    /// Call periodically (finer-grained than the configured timeout).
+    /// Returns true exactly when a recovery attempt should be made right
+    /// now; each true also arms the backoff before the next one is even
+    /// considered, so a caller polling in a tight loop still cannot fire
+    /// back-to-back attempts against a publisher that never comes back --
+    /// see the class comment: hammering every 2 seconds forever against a
+    /// genuinely gone publisher is worse than a frozen source.
+    bool poll(std::chrono::steady_clock::time_point now);
+
+    /// Attempts made since the current stall started (since the last frame,
+    /// or since expecting-frames most recently became true). Reset by
+    /// onFrameDelivered and by setExpectingFrames.
+    int attemptsThisStall() const { return attempts_; }
+
+private:
+    std::chrono::milliseconds timeout_;
+    std::chrono::milliseconds max_backoff_;
+
+    bool expecting_ = false;
+    bool have_baseline_ = false;
+    std::chrono::steady_clock::time_point baseline_{};
+    std::chrono::milliseconds backoff_{};
+    std::chrono::steady_clock::time_point next_attempt_allowed_{};
+    int attempts_ = 0;
+};
+
 // ---------------------------------------------------------------------------
 // Frames handed to the OBS adapter
 // ---------------------------------------------------------------------------
@@ -177,4 +254,13 @@ using VideoFrameHandler = std::function<void(const VideoFrameData &)>;
 using AudioFrameHandler = std::function<void(const AudioFrameData &)>;
 using SessionStateHandler = std::function<void(SessionState state, const std::string &detail)>;
 
+/// Severity for LiveKitSession's own diagnostic log lines (currently just
+/// the stall-recovery watchdog). Kept separate from OBS's LOG_* levels and
+/// from the SDK's own livekit::LogLevel so core/ stays free of any OBS
+/// dependency -- the adapter maps this onto obs_log the same way it already
+/// maps livekit::LogLevel (see plugin-main.cpp's livekit log bridge).
+enum class DiagnosticLevel { Info, Warning };
+
+using DiagnosticHandler = std::function<void(DiagnosticLevel level, const std::string &message)>;
+
 } // namespace stplugin

core/src/session.cpp

@@ -17,6 +17,7 @@ You may obtain a copy of the License at
 #include <deque>
 #include <exception>
 #include <mutex>
+#include <string>
 #include <thread>
 #include <utility>
 #include <vector>
@@ -30,6 +31,7 @@ You may obtain a copy of the License at
 #include <livekit/room_delegate.h>
 #include <livekit/room_event_types.h>
 #include <livekit/track.h>
+#include <livekit/track_publication.h>
 #include <livekit/video_frame.h>
 #include <livekit/video_stream.h>
 
@@ -138,6 +140,13 @@ int &globalRefCount()
     return n;
 }
 
+// --- Stall-recovery watchdog tuning -----------------------------------------
+
+/// How often the watchdog thread checks StallWatchdog's clock. Deliberately
+/// finer than kStallRecoveryTimeout (session.h) so detection latency tracks
+/// the threshold itself, not the threshold plus a whole polling period.
+constexpr std::chrono::milliseconds kWatchdogPollInterval{250};
+
 } // namespace
 
 // ---------------------------------------------------------------------------
@@ -145,11 +154,12 @@ int &globalRefCount()
 // ---------------------------------------------------------------------------
 
 struct LiveKitSession::Impl : public livekit::RoomDelegate {
-    enum class CommandType { AttachVideo, DetachVideo, AttachAudio, DetachAudio, Stop };
+    enum class CommandType { AttachVideo, DetachVideo, AttachAudio, DetachAudio, RecoverVideo, Stop };
 
     struct Command {
         CommandType type;
         std::shared_ptr<livekit::Track> track;
+        std::shared_ptr<livekit::RemoteTrackPublication> publication;
     };
 
     livekit::Room room;
@@ -157,10 +167,14 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
 
     mutable std::mutex state_mutex;
     SessionStateMachine machine;
+    // Guarded by state_mutex too, same contract as `machine` -- see
+    // StallWatchdog's own comment for what it decides and why it exists.
+    StallWatchdog stall_watchdog{kStallRecoveryTimeout, kStallRecoveryMaxBackoff};
 
     VideoFrameHandler on_video;
     AudioFrameHandler on_audio;
     SessionStateHandler on_state;
+    DiagnosticHandler on_diagnostic;
 
     std::atomic<std::uint64_t> video_frames{0};
     std::atomic<std::uint64_t> audio_frames{0};
@@ -170,7 +184,10 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
     // livekit::AudioStream happens on `worker`, never on a room event thread:
     // the SDK's room callbacks run on its own event thread and blocking or
     // re-entering there stalls every other event (and Room::disconnect() from
-    // inside one is documented to deadlock outright).
+    // inside one is documented to deadlock outright). The stall-recovery
+    // watchdog thread follows the same rule: it never touches
+    // RemoteTrackPublication itself, only posts CommandType::RecoverVideo
+    // and lets the worker thread do it (see recoverVideo() below).
     std::mutex queue_mutex;
     std::condition_variable queue_cv;
     std::deque<Command> queue;
@@ -183,6 +200,26 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
     std::shared_ptr<livekit::AudioStream> audio_stream;
     std::thread audio_thread;
 
+    // The publication/track backing the CURRENT video subscription, kept
+    // around purely so recoverVideo() and the mute-change handlers have
+    // something to act on without reaching into `video_stream` (which is
+    // worker-thread-exclusive, per the comment above). Set together in
+    // attachVideo(), cleared together in detachVideo(), both on the worker
+    // thread; read from the room event thread (handleMuteChange) and the
+    // worker thread (recoverVideo()) under this mutex.
+    std::mutex video_track_mutex;
+    std::shared_ptr<livekit::Track> current_video_track;
+    std::shared_ptr<livekit::RemoteTrackPublication> current_video_publication;
+
+    // The watchdog's own timer thread. It owns no SDK state and calls no SDK
+    // method directly -- see the queue comment above. `watchdog_mutex` only
+    // ever guards the shutdown flag/condvar pair, never `stall_watchdog`
+    // (that is guarded by `state_mutex`, alongside `machine`).
+    std::mutex watchdog_mutex;
+    std::condition_variable watchdog_cv;
+    bool watchdog_running = false;
+    std::thread watchdog_thread;
+
     bool connected = false;
 
     ~Impl() override = default;
@@ -207,17 +244,29 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
             handler(state, detail);
     }
 
-    void post(CommandType type, std::shared_ptr<livekit::Track> track = nullptr)
+    void post(CommandType type, std::shared_ptr<livekit::Track> track = nullptr,
+              std::shared_ptr<livekit::RemoteTrackPublication> publication = nullptr)
     {
         {
             std::lock_guard<std::mutex> guard(queue_mutex);
             if (!worker_running)
                 return;
-            queue.push_back(Command{type, std::move(track)});
+            queue.push_back(Command{type, std::move(track), std::move(publication)});
         }
         queue_cv.notify_one();
     }
 
+    void logDiagnostic(DiagnosticLevel level, const std::string &message)
+    {
+        DiagnosticHandler handler;
+        {
+            std::lock_guard<std::mutex> guard(state_mutex);
+            handler = on_diagnostic;
+        }
+        if (handler)
+            handler(level, message);
+    }
+
     // Handles the wanted video track once matched, shared by onTrackSubscribed
     // (a fresh subscription) and attachExistingTracks (one already up when
     // this session started watching). Two responsibilities that only make
@@ -261,7 +310,42 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
                 // it already had, which is the pre-existing behaviour.
             }
         }
-        post(CommandType::AttachVideo, track);
+        // `publication` rides along so attachVideo() can remember it: it is
+        // the handle the stall-recovery watchdog later toggles
+        // (setEnabled(false)/(true)) to force a fresh keyframe. See
+        // recoverVideo() and StallWatchdog's comment in session_types.h.
+        post(CommandType::AttachVideo, track, publication);
+    }
+
+    // Fired for ANY track (any participant, any kind) muting or unmuting.
+    // Filtered down to "is this the video publication we are currently
+    // watching" by SID, which also naturally excludes every audio mute and
+    // every other participant's tracks without a separate identity/kind
+    // check.
+    //
+    // Why this exists: the stall-recovery watchdog (see StallWatchdog's
+    // comment) treats "no decoded frame for kStallRecoveryTimeout" as a
+    // stall worth toggling the subscription over. A publisher who
+    // legitimately turned their camera off produces exactly that symptom on
+    // purpose, and toggling their subscription every couple of seconds for
+    // as long as they stay off would be an endless, pointless loop against
+    // healthy behaviour. Muting suspends the watchdog's clock entirely;
+    // unmuting starts a brand-new grace period rather than reading "muted
+    // for twenty minutes" as "stalled for twenty minutes".
+    void handleMuteChange(const std::shared_ptr<livekit::TrackPublication> &publication, bool unmuted)
+    {
+        if (!publication || publication->kind() != livekit::TrackKind::KIND_VIDEO)
+            return;
+        std::string current_sid;
+        {
+            std::lock_guard<std::mutex> guard(video_track_mutex);
+            if (current_video_publication)
+                current_sid = current_video_publication->sid();
+        }
+        if (current_sid.empty() || publication->sid() != current_sid)
+            return;
+        std::lock_guard<std::mutex> guard(state_mutex);
+        stall_watchdog.setExpectingFrames(unmuted, std::chrono::steady_clock::now());
     }
 
     // --- RoomDelegate ------------------------------------------------------
@@ -304,6 +388,16 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
         post(CommandType::DetachAudio);
     }
 
+    void onTrackMuted(livekit::Room &, const livekit::TrackMutedEvent &event) override
+    {
+        handleMuteChange(event.publication, false);
+    }
+
+    void onTrackUnmuted(livekit::Room &, const livekit::TrackUnmutedEvent &event) override
+    {
+        handleMuteChange(event.publication, true);
+    }
+
     void onReconnecting(livekit::Room &, const livekit::ReconnectingEvent &) override
     {
         mutateState([](SessionStateMachine &m) { m.onReconnecting(); });
@@ -359,7 +453,7 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
     void workerLoop()
     {
         for (;;) {
-            Command command{CommandType::Stop, nullptr};
+            Command command{CommandType::Stop, nullptr, nullptr};
             {
                 std::unique_lock<std::mutex> lock(queue_mutex);
                 queue_cv.wait(lock, [this] { return !queue.empty(); });
@@ -369,7 +463,7 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
 
             switch (command.type) {
             case CommandType::AttachVideo:
-                attachVideo(command.track);
+                attachVideo(command.track, command.publication);
                 break;
             case CommandType::DetachVideo:
                 detachVideo();
@@ -380,6 +474,9 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
             case CommandType::DetachAudio:
                 detachAudio();
                 break;
+            case CommandType::RecoverVideo:
+                recoverVideo();
+                break;
             case CommandType::Stop:
                 detachVideo();
                 detachAudio();
@@ -388,7 +485,103 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
         }
     }
 
-    void attachVideo(const std::shared_ptr<livekit::Track> &track)
+    // --- stall-recovery watchdog --------------------------------------------
+    //
+    // See StallWatchdog's comment (session_types.h) for the measured
+    // evidence and why toggling the publication is the only lever
+    // available. This thread does exactly one thing: tick StallWatchdog's
+    // clock and, when it says to, post CommandType::RecoverVideo so the
+    // worker thread does the actual SDK call. It never touches
+    // livekit::VideoStream, livekit::Room or RemoteTrackPublication itself.
+
+    void startWatchdog()
+    {
+        {
+            std::lock_guard<std::mutex> guard(watchdog_mutex);
+            watchdog_running = true;
+        }
+        watchdog_thread = std::thread([this] { watchdogLoop(); });
+    }
+
+    void stopWatchdog()
+    {
+        {
+            std::lock_guard<std::mutex> guard(watchdog_mutex);
+            if (!watchdog_running)
+                return;
+            watchdog_running = false;
+        }
+        watchdog_cv.notify_all();
+        if (watchdog_thread.joinable())
+            watchdog_thread.join();
+    }
+
+    void watchdogLoop()
+    {
+        std::unique_lock<std::mutex> lock(watchdog_mutex);
+        while (watchdog_running) {
+            // kWatchdogPollInterval is finer than kStallRecoveryTimeout so
+            // detection latency tracks the threshold itself rather than the
+            // threshold plus a whole polling period; woken early on
+            // shutdown by stopWatchdog()'s notify_all().
+            watchdog_cv.wait_for(lock, kWatchdogPollInterval);
+            if (!watchdog_running)
+                break;
+            lock.unlock();
+
+            bool should_fire;
+            {
+                std::lock_guard<std::mutex> guard(state_mutex);
+                should_fire = stall_watchdog.poll(std::chrono::steady_clock::now());
+            }
+            if (should_fire)
+                post(CommandType::RecoverVideo);
+
+            lock.lock();
+        }
+    }
+
+    // Actually pulls the lever: toggles the video publication off and back
+    // on, which makes the SFU stop and restart delivery of that track --
+    // and a restart always begins with a keyframe. Only ever called from
+    // the worker thread (via CommandType::RecoverVideo), same as every
+    // other RemoteTrackPublication/VideoStream call in this file.
+    void recoverVideo()
+    {
+        std::shared_ptr<livekit::RemoteTrackPublication> publication;
+        std::shared_ptr<livekit::Track> track;
+        {
+            std::lock_guard<std::mutex> guard(video_track_mutex);
+            publication = current_video_publication;
+            track = current_video_track;
+        }
+        // Detached, replaced, or muted between the watchdog deciding to
+        // fire and the worker getting to this command -- nothing to do, and
+        // silently: this is the expected shape of the race, not a failure
+        // worth logging.
+        if (!publication || !track || track->muted())
+            return;
+
+        int attempt = 0;
+        {
+            std::lock_guard<std::mutex> guard(state_mutex);
+            attempt = stall_watchdog.attemptsThisStall();
+        }
+
+        logDiagnostic(DiagnosticLevel::Warning,
+                      "no decoded video frame for >= " + std::to_string(kStallRecoveryTimeout.count()) +
+                          "ms; toggling the subscription to force a fresh keyframe (attempt " +
+                          std::to_string(attempt) + ")");
+        try {
+            publication->setEnabled(false);
+            publication->setEnabled(true);
+        } catch (const std::exception &e) {
+            logDiagnostic(DiagnosticLevel::Warning, std::string("stall-recovery toggle failed: ") + e.what());
+        }
+    }
+
+    void attachVideo(const std::shared_ptr<livekit::Track> &track,
+                      const std::shared_ptr<livekit::RemoteTrackPublication> &publication)
     {
         if (!track)
             return;
@@ -413,6 +606,18 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
         if (!stream)
             return;
 
+        {
+            std::lock_guard<std::mutex> guard(video_track_mutex);
+            current_video_track = track;
+            current_video_publication = publication;
+        }
+        {
+            // A fresh subscription (or a publisher swap) starts a brand-new
+            // grace period -- see StallWatchdog::setExpectingFrames.
+            std::lock_guard<std::mutex> guard(state_mutex);
+            stall_watchdog.setExpectingFrames(true, std::chrono::steady_clock::now());
+        }
+
         video_stream = stream;
         video_thread = std::thread([this, stream] { videoReaderLoop(stream); });
         mutateState([](SessionStateMachine &m) { m.onVideoAttached(); });
@@ -426,6 +631,20 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
             video_thread.join();
         const bool had = static_cast<bool>(video_stream);
         video_stream.reset();
+        {
+            std::lock_guard<std::mutex> guard(video_track_mutex);
+            current_video_track.reset();
+            current_video_publication.reset();
+        }
+        {
+            // Nothing subscribed means nothing expected -- see
+            // StallWatchdog::setExpectingFrames. Unconditional, not gated on
+            // `had`: this also covers the detachVideo() at the top of
+            // attachVideo() above, which is exactly the publisher-swap
+            // moment the grace period needs to restart from.
+            std::lock_guard<std::mutex> guard(state_mutex);
+            stall_watchdog.setExpectingFrames(false, std::chrono::steady_clock::now());
+        }
         if (had)
             mutateState([](SessionStateMachine &m) { m.onVideoDetached(); });
     }
@@ -548,6 +767,23 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
         out.plane_count = count;
 
         video_frames.fetch_add(1);
+
+        // Tell the stall-recovery watchdog a frame actually made it all the
+        // way to "about to hand to OBS" -- not merely that the SDK's queue
+        // produced an event, which the drop paths above also see. See
+        // StallWatchdog's comment for why this exists.
+        int attempts_before_this_frame = 0;
+        {
+            std::lock_guard<std::mutex> guard(state_mutex);
+            attempts_before_this_frame = stall_watchdog.attemptsThisStall();
+            stall_watchdog.onFrameDelivered(std::chrono::steady_clock::now());
+        }
+        if (attempts_before_this_frame > 0) {
+            logDiagnostic(DiagnosticLevel::Info, "video resumed after " +
+                                                      std::to_string(attempts_before_this_frame) +
+                                                      " stall-recovery attempt(s)");
+        }
+
         handler(out);
     }
 
@@ -633,6 +869,12 @@ void LiveKitSession::setStateHandler(SessionStateHandler handler)
     impl_->on_state = std::move(handler);
 }
 
+void LiveKitSession::setDiagnosticHandler(DiagnosticHandler handler)
+{
+    std::lock_guard<std::mutex> guard(impl_->state_mutex);
+    impl_->on_diagnostic = std::move(handler);
+}
+
 bool LiveKitSession::connect(const SessionConfig &config)
 {
     if (impl_->connected)
@@ -652,6 +894,10 @@ bool LiveKitSession::connect(const SessionConfig &config)
     }
 
     impl_->startWorker();
+    // Runs for the lifetime of the worker: connected-but-nothing-subscribed
+    // is a no-op for StallWatchdog (see setExpectingFrames), so there is no
+    // reason to start/stop it separately from the worker it posts to.
+    impl_->startWatchdog();
 
     livekit::RoomOptions options;
     // auto_subscribe is what makes track_subscribed events (and therefore any
@@ -680,6 +926,7 @@ bool LiveKitSession::connect(const SessionConfig &config)
     } catch (const std::exception &e) {
         ok = false;
         impl_->mutateState([&](SessionStateMachine &m) { m.onConnectFailed(e.what()); });
+        impl_->stopWatchdog();
         impl_->stopWorker();
         impl_->room.setDelegate(nullptr);
         return false;
@@ -689,6 +936,7 @@ bool LiveKitSession::connect(const SessionConfig &config)
         impl_->mutateState([](SessionStateMachine &m) {
             m.onConnectFailed("could not connect to LiveKit (check the server URL, or the token may have expired)");
         });
+        impl_->stopWatchdog();
         impl_->stopWorker();
         impl_->room.setDelegate(nullptr);
         return false;
@@ -705,9 +953,11 @@ void LiveKitSession::disconnect()
     if (!impl_)
         return;
 
-    // Order matters: stop the readers first so nothing is mid-read on a
-    // stream the room is about to tear down, then disconnect the room, then
-    // drop the delegate so no event can arrive at a half-destroyed object.
+    // Order matters: stop the watchdog and the readers first so nothing is
+    // mid-read (or about to post a recovery command) on a stream the room
+    // is about to tear down, then disconnect the room, then drop the
+    // delegate so no event can arrive at a half-destroyed object.
+    impl_->stopWatchdog();
     impl_->stopWorker();
 
     if (impl_->connected) {

core/src/session_types.cpp

@@ -11,6 +11,8 @@ You may obtain a copy of the License at
 
 #include "stplugin/session_types.h"
 
+#include <algorithm>
+
 namespace stplugin {
 
 const char *describePixelFormat(PixelFormat format)
@@ -175,4 +177,62 @@ void SessionStateMachine::onAudioDetached()
     has_audio_ = false;
 }
 
+// ---------------------------------------------------------------------------
+// StallWatchdog
+// ---------------------------------------------------------------------------
+
+StallWatchdog::StallWatchdog(std::chrono::milliseconds timeout, std::chrono::milliseconds max_backoff)
+    : timeout_(timeout), max_backoff_(max_backoff), backoff_(timeout)
+{
+}
+
+void StallWatchdog::setExpectingFrames(bool expecting, std::chrono::steady_clock::time_point now)
+{
+    expecting_ = expecting;
+    // Always re-baseline from `now`, whichever direction this flips.
+    // Losing the baseline (rather than, say, keeping the old one around for
+    // when expecting_ next becomes true) is what stops a track that was
+    // muted for the last twenty minutes from reading as "twenty minutes
+    // stalled" the instant it unmutes.
+    have_baseline_ = expecting;
+    baseline_ = now;
+    backoff_ = timeout_;
+    next_attempt_allowed_ = now;
+    attempts_ = 0;
+}
+
+void StallWatchdog::onFrameDelivered(std::chrono::steady_clock::time_point now)
+{
+    have_baseline_ = true;
+    baseline_ = now;
+    backoff_ = timeout_;
+    // A recovered stream must be able to fire again the moment a FRESH
+    // stall clears the (now-reset) timeout, not sit throttled by whatever
+    // backoff a previous, unrelated stall had climbed to -- next_attempt_
+    // allowed_ belongs to that old stall and is meaningless once frames are
+    // flowing again.
+    next_attempt_allowed_ = now;
+    attempts_ = 0;
+}
+
+bool StallWatchdog::poll(std::chrono::steady_clock::time_point now)
+{
+    if (!expecting_ || !have_baseline_)
+        return false;
+    if (now - baseline_ < timeout_)
+        return false;
+    if (now < next_attempt_allowed_)
+        return false;
+
+    ++attempts_;
+    // Next attempt against this SAME stall is not allowed until the backoff
+    // elapses, and the backoff itself doubles (capped) each time -- 2s, 4s,
+    // 8s, ... up to max_backoff_ -- so a publisher that is genuinely gone
+    // gets progressively less frequent toggles instead of one every 2
+    // seconds for the rest of the show.
+    next_attempt_allowed_ = now + backoff_;
+    backoff_ = std::min(backoff_ * 2, max_backoff_);
+    return true;
+}
+
 } // namespace stplugin

core/tests/test_session.cpp

@@ -14,8 +14,10 @@ You may obtain a copy of the License at
 //   COVERED headlessly -- the session's own decision-making: the state
 //   machine's transitions (including the publisher-swap and reconnect paths
 //   that motivated this plugin), track selection, frame geometry validation,
-//   and the real connect() failure paths against the real SDK (bad URL,
-//   unreachable host, garbage token).
+//   the stall-recovery watchdog's timing/backoff decisions (StallWatchdog,
+//   driven with a fake clock -- see its own section below), and the real
+//   connect() failure paths against the real SDK (bad URL, unreachable
+//   host, garbage token).
 //
 //   NOT COVERED here -- anything that needs a LiveKit server to answer:
 //   a successful connect, actual subscription, and actual decoded frames
@@ -209,6 +211,134 @@ void testFailureAndRecovery()
     ST_ASSERT(idle.state() == SessionState::Idle);
 }
 
+// ---------------------------------------------------------------------------
+// StallWatchdog -- the stall-recovery watchdog's pure timing/decision logic.
+//
+// This is deliberately driven with an explicit, fake clock (arbitrary
+// steady_clock::time_points built by hand, never std::chrono::...::now())
+// rather than real sleeps: every case below needs to be exact about
+// "1999ms in" vs "2001ms in" and about backoff boundaries, and a test that
+// actually slept for 30+ seconds to exercise the backoff ceiling would be
+// exactly the kind of slow, flaky test this project's whole headless-test
+// philosophy exists to avoid. See StallWatchdog's own comment
+// (session_types.h) for the measured server evidence this exists to fix.
+// ---------------------------------------------------------------------------
+
+void testStallWatchdogFiresAfterThreshold()
+{
+    const auto t0 = std::chrono::steady_clock::now();
+    StallWatchdog w(std::chrono::milliseconds(2000), std::chrono::milliseconds(30000));
+
+    // Nothing subscribed yet: polling is a no-op, no matter how much time
+    // has "passed" -- an audio-only source or a disconnected session must
+    // never fire.
+    ST_ASSERT(!w.poll(t0 + std::chrono::milliseconds(10000)));
+
+    // A track becomes subscribed. Still well under the threshold: quiet.
+    w.setExpectingFrames(true, t0);
+    ST_ASSERT(!w.poll(t0 + std::chrono::milliseconds(500)));
+    ST_ASSERT(!w.poll(t0 + std::chrono::milliseconds(1999)));
+
+    // No frame ever arrived, and the threshold has now elapsed: fires
+    // exactly once when asked right at/after the boundary.
+    ST_ASSERT(w.poll(t0 + std::chrono::milliseconds(2000)));
+    ST_ASSERT_EQ(w.attemptsThisStall(), 1);
+
+    // A frame arriving resets the clock -- the far more common case in a
+    // healthy stream, where onFrameDelivered() is called every ~33ms and
+    // poll() (every kWatchdogPollInterval) never sees 2000ms of silence.
+    StallWatchdog healthy(std::chrono::milliseconds(2000), std::chrono::milliseconds(30000));
+    healthy.setExpectingFrames(true, t0);
+    for (int ms = 0; ms <= 5000; ms += 33)
+        healthy.onFrameDelivered(t0 + std::chrono::milliseconds(ms));
+    ST_ASSERT(!healthy.poll(t0 + std::chrono::milliseconds(5010)));
+    ST_ASSERT_EQ(healthy.attemptsThisStall(), 0);
+}
+
+void testStallWatchdogDoesNotFireWhenNotExpectingFrames()
+{
+    const auto t0 = std::chrono::steady_clock::now();
+    StallWatchdog w(std::chrono::milliseconds(2000), std::chrono::milliseconds(30000));
+
+    // A muted (or disabled/unsubscribed) track is expected silence, not a
+    // stall -- setExpectingFrames(false, ...) is exactly what
+    // LiveKitSession::Impl::handleMuteChange (and detachVideo()) call in
+    // that case. It must not fire no matter how long it stays that way.
+    w.setExpectingFrames(false, t0);
+    ST_ASSERT(!w.poll(t0 + std::chrono::milliseconds(2000)));
+    ST_ASSERT(!w.poll(t0 + std::chrono::milliseconds(60000)));
+    ST_ASSERT(!w.poll(t0 + std::chrono::milliseconds(600000)));
+
+    // Un-muting (setExpectingFrames(true, ...)) starts a BRAND NEW grace
+    // period from that moment -- it must not read "was silent for ten
+    // minutes" as "stalled for ten minutes" and fire immediately.
+    const auto unmuted_at = t0 + std::chrono::milliseconds(600000);
+    w.setExpectingFrames(true, unmuted_at);
+    ST_ASSERT(!w.poll(unmuted_at + std::chrono::milliseconds(1999)));
+    ST_ASSERT(w.poll(unmuted_at + std::chrono::milliseconds(2000)));
+}
+
+void testStallWatchdogBacksOffRatherThanLooping()
+{
+    const auto t0 = std::chrono::steady_clock::now();
+    StallWatchdog w(std::chrono::milliseconds(2000), std::chrono::milliseconds(30000));
+    w.setExpectingFrames(true, t0);
+
+    // First attempt at the threshold.
+    auto now = t0 + std::chrono::milliseconds(2000);
+    ST_ASSERT(w.poll(now));
+    ST_ASSERT_EQ(w.attemptsThisStall(), 1);
+
+    // A genuinely gone publisher: no frame ever comes back. Immediately
+    // asking again (the naive "retry every poll interval forever" a
+    // watchdog without backoff would do) must NOT fire -- that is precisely
+    // the "hammered every 2 seconds forever" this backoff exists to avoid.
+    ST_ASSERT(!w.poll(now + std::chrono::milliseconds(250)));
+    ST_ASSERT(!w.poll(now + std::chrono::milliseconds(1999)));
+
+    // The backoff after attempt 1 is the base timeout (2000ms): the second
+    // attempt is allowed at +2000ms from the first, not before.
+    now += std::chrono::milliseconds(2000);
+    ST_ASSERT(w.poll(now));
+    ST_ASSERT_EQ(w.attemptsThisStall(), 2);
+
+    // Backoff doubles: the third attempt needs a 4000ms gap, not 2000ms.
+    ST_ASSERT(!w.poll(now + std::chrono::milliseconds(3999)));
+    now += std::chrono::milliseconds(4000);
+    ST_ASSERT(w.poll(now));
+    ST_ASSERT_EQ(w.attemptsThisStall(), 3);
+
+    // ... and again to 8000ms, and again to 16000ms.
+    ST_ASSERT(!w.poll(now + std::chrono::milliseconds(7999)));
+    now += std::chrono::milliseconds(8000);
+    ST_ASSERT(w.poll(now));
+    ST_ASSERT_EQ(w.attemptsThisStall(), 4);
+
+    ST_ASSERT(!w.poll(now + std::chrono::milliseconds(15999)));
+    now += std::chrono::milliseconds(16000);
+    ST_ASSERT(w.poll(now));
+    ST_ASSERT_EQ(w.attemptsThisStall(), 5);
+
+    // The backoff is capped: doubling 16000ms would be 32000ms, but it
+    // never exceeds max_backoff (30000ms) no matter how many attempts have
+    // failed, so a publisher that comes back after an hour is still
+    // retried at a bounded cadence, not abandoned.
+    for (int i = 0; i < 6; ++i) {
+        ST_ASSERT(!w.poll(now + std::chrono::milliseconds(29999)));
+        now += std::chrono::milliseconds(30000);
+        ST_ASSERT(w.poll(now));
+    }
+
+    // A frame finally arrives: the stall is over, and the NEXT one (a fresh
+    // stall, not a continuation) starts back at the base cadence rather
+    // than staying parked at the 30s ceiling forever.
+    w.onFrameDelivered(now);
+    ST_ASSERT_EQ(w.attemptsThisStall(), 0);
+    ST_ASSERT(!w.poll(now + std::chrono::milliseconds(1999)));
+    ST_ASSERT(w.poll(now + std::chrono::milliseconds(2000)));
+    ST_ASSERT_EQ(w.attemptsThisStall(), 1);
+}
+
 // ---------------------------------------------------------------------------
 // Real SDK, failure paths only (no LiveKit server available headlessly)
 // ---------------------------------------------------------------------------
@@ -335,6 +465,10 @@ int main()
     testReconnect();
     testFailureAndRecovery();
 
+    testStallWatchdogFiresAfterThreshold();
+    testStallWatchdogDoesNotFireWhenNotExpectingFrames();
+    testStallWatchdogBacksOffRatherThanLooping();
+
     LiveKitSession::globalInitialize();
     testConnectRejectsIncompleteConfig();
     testConnectToUnreachableServerFailsCleanly();

obs-adapter/src/plugin-main.cpp

@@ -375,6 +375,13 @@ void *sourceCreate(obs_data_t *settings, obs_source_t *source)
 
     self->session->setVideoHandler([self](const VideoFrameData &frame) { outputVideoFrame(self, frame); });
     self->session->setAudioHandler([self](const AudioFrameData &frame) { outputAudioFrame(self, frame); });
+    // The stall-recovery watchdog (core/src/session.cpp) is the only thing
+    // that currently uses this: it logs each toggle-the-subscription
+    // recovery attempt, and its eventual success, so a stalled-and-fixed
+    // camera is diagnosable from an OBS log afterward instead of invisible.
+    self->session->setDiagnosticHandler([](DiagnosticLevel level, const std::string &message) {
+        obs_log(level == DiagnosticLevel::Warning ? LOG_WARNING : LOG_INFO, "%s", message.c_str());
+    });
     self->session->setStateHandler([self](SessionState state, const std::string &detail) {
         self->setStatus(detail.empty() ? describeSessionState(state) : detail);
         self->status_is_error.store(state == SessionState::Failed);