fix(watch-together): retry guest media switches until they commit

Guest switch dispatch pre-marked its dedup key and fired-and-forgot, so
any failure (fetch error, reload busy with an auto-advance, navigation
race with the host exiting) silently stranded the guest on the old media.
A CurrentPlaybackDispatcher now marks a key handled only after the sink
reports success against the committed identity, with a serialized
in-flight slot, timeout, and generation reset; the reconciler re-offers
unattached media on every host heartbeat, making the heartbeat the retry
channel. Fetches that outlive their dispatch are re-validated against the
current snapshot so a stale switch can't override the live one.
hostExitedPlayer now rides the controller's ordered message queue with
host authentication instead of racing state handling in the provider.

test/watch_together/current_playback_dispatcher_test.dart

@@ -0,0 +1,103 @@
+import 'dart:async';
+
+import 'package:fake_async/fake_async.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:plezy/watch_together/services/current_playback_dispatcher.dart';
+
+void main() {
+  group('CurrentPlaybackDispatcher', () {
+    test('success marks the key handled and suppresses re-dispatch', () async {
+      final d = CurrentPlaybackDispatcher();
+      expect(d.shouldDispatch('a'), isTrue);
+
+      await d.dispatch('a', () async => true);
+      expect(d.shouldDispatch('a'), isFalse); // Handled.
+      expect(d.shouldDispatch('b'), isTrue); // Other keys unaffected.
+    });
+
+    test('failure frees the slot without marking handled (heartbeat retry)', () async {
+      final d = CurrentPlaybackDispatcher();
+      await d.dispatch('a', () async => false);
+      expect(d.inFlightKey, isNull);
+      expect(d.shouldDispatch('a'), isTrue); // Retryable.
+    });
+
+    test('a throwing callback is a failure, not an unhandled error', () async {
+      final d = CurrentPlaybackDispatcher();
+      await expectLater(d.dispatch('a', () async => throw StateError('boom')), completes);
+      expect(d.shouldDispatch('a'), isTrue);
+    });
+
+    test('serializes: nothing dispatches while a key is in flight', () async {
+      final d = CurrentPlaybackDispatcher();
+      final gate = Completer<bool>();
+      final dispatch = d.dispatch('a', () => gate.future);
+
+      // The slot is claimed synchronously, before the first await.
+      expect(d.inFlightKey, 'a');
+      expect(d.shouldDispatch('a'), isFalse);
+      expect(d.shouldDispatch('b'), isFalse);
+
+      gate.complete(true);
+      await dispatch;
+      expect(d.inFlightKey, isNull);
+      expect(d.shouldDispatch('b'), isTrue);
+    });
+
+    test('a hung callback times out as a failure and frees the slot', () {
+      fakeAsync((async) {
+        final d = CurrentPlaybackDispatcher();
+        final never = Completer<bool>();
+        unawaited(d.dispatch('a', () => never.future));
+
+        async.elapse(CurrentPlaybackDispatcher.dispatchTimeout - const Duration(seconds: 1));
+        expect(d.inFlightKey, 'a');
+
+        async.elapse(const Duration(seconds: 2));
+        expect(d.inFlightKey, isNull);
+        expect(d.shouldDispatch('a'), isTrue); // Timed out ⇒ unhandled.
+
+        // A late success from the original callback changes nothing.
+        never.complete(true);
+        async.flushMicrotasks();
+        expect(d.shouldDispatch('a'), isTrue);
+      });
+    });
+
+    test('reset() mid-flight discards the stale completion', () async {
+      final d = CurrentPlaybackDispatcher();
+      final gateA = Completer<bool>();
+      final dispatchA = d.dispatch('a', () => gateA.future);
+
+      d.reset(); // Host exited / session left.
+      expect(d.inFlightKey, isNull);
+
+      // A newer dispatch claims the slot under the new generation.
+      final gateB = Completer<bool>();
+      final dispatchB = d.dispatch('b', () => gateB.future);
+
+      // The stale completion must neither mark 'a' handled nor free 'b'.
+      gateA.complete(true);
+      await dispatchA;
+      expect(d.shouldDispatch('a'), isFalse); // 'b' still occupies the slot...
+      expect(d.inFlightKey, 'b'); // ...untouched by the stale completion.
+
+      gateB.complete(true);
+      await dispatchB;
+      expect(d.shouldDispatch('a'), isTrue); // 'a' was never marked handled.
+      expect(d.shouldDispatch('b'), isFalse);
+    });
+
+    test('markHandled suppresses a key without a dispatch (user-initiated join)', () {
+      final d = CurrentPlaybackDispatcher();
+      d.markHandled('a');
+      expect(d.shouldDispatch('a'), isFalse);
+      expect(d.shouldDispatch('b'), isTrue);
+    });
+
+    test('null keys never dispatch', () {
+      final d = CurrentPlaybackDispatcher();
+      expect(d.shouldDispatch(null), isFalse);
+    });
+  });
+}

test/watch_together/guest_playback_reconciler_test.dart

@@ -364,6 +364,44 @@ void main() {
       });
     });
 
+    test('detached guest is re-notified on every state (heartbeat retry channel)', () {
+      fakeAsync((async) {
+        final switches = <String>[];
+        final h = _Harness(
+          async,
+          callbacks: GuestReconcilerCallbacks(onMediaSwitchNeeded: (rk, sid, title) => switches.add(rk)),
+        );
+
+        // Never attached: every heartbeat re-offers the switch so a failed
+        // navigation can retry (the provider's dispatcher dedups).
+        h.reconciler.onState(h.state());
+        h.reconciler.onState(h.state());
+        h.reconciler.onState(h.state());
+        async.flushMicrotasks();
+
+        expect(switches, ['rk1', 'rk1', 'rk1']);
+        expect(h.player.commandLog, isEmpty);
+        h.dispose();
+      });
+    });
+
+    test('attached to matching media never fires the switch callback', () {
+      fakeAsync((async) {
+        final switches = <String>[];
+        final h = _Harness(
+          async,
+          callbacks: GuestReconcilerCallbacks(onMediaSwitchNeeded: (rk, sid, title) => switches.add(rk)),
+        );
+        h.attachReady();
+
+        h.reconciler.onState(h.state());
+        async.elapse(const Duration(seconds: 2));
+
+        expect(switches, isEmpty);
+        h.dispose();
+      });
+    });
+
     test('attach reconciles to the latest state received while detached', () {
       fakeAsync((async) {
         final h = _Harness(async);

test/watch_together/watch_together_controller_test.dart

@@ -273,4 +273,56 @@ void main() {
       room.dispose();
     });
   });
+
+  group('hostExitedPlayer routing', () {
+    test('rides the ordered queue: never overtakes states sent before it', () {
+      fakeAsync((async) {
+        final room = _Room(async);
+        final log = <String>[];
+        room.guest.onMediaStateReceived = (rk, sid, title) => log.add('state:$rk');
+        room.guest.onHostExitedPlayer = () => log.add('hostExit');
+
+        // Host starts media, then exits the player — wire order matters.
+        room.hostStartsMedia();
+        room.hostService.broadcast(SyncMessage.hostExitedPlayer(peerId: 'host'));
+        async.flushMicrotasks();
+
+        expect(log, isNotEmpty);
+        expect(log.first, 'state:rk1');
+        expect(log.last, 'hostExit');
+        room.dispose();
+      });
+    });
+
+    test('is ignored when forged by a non-host peer', () {
+      fakeAsync((async) {
+        final room = _Room(async);
+        var hostExits = 0;
+        room.guest.onHostExitedPlayer = () => hostExits++;
+
+        final evil = room.hub.register('evil');
+        evil.broadcast(SyncMessage.hostExitedPlayer(peerId: 'evil'));
+        async.flushMicrotasks();
+
+        expect(hostExits, 0);
+        room.dispose();
+      });
+    });
+
+    test('the host itself never reacts to a hostExitedPlayer echo', () {
+      fakeAsync((async) {
+        final room = _Room(async);
+        var hostExits = 0;
+        room.host.onHostExitedPlayer = () => hostExits++;
+
+        // A confused/malicious guest sends the message; the host must not
+        // tear down its own epoch.
+        room.guestService.broadcast(SyncMessage.hostExitedPlayer(peerId: 'guest'));
+        async.flushMicrotasks();
+
+        expect(hostExits, 0);
+        room.dispose();
+      });
+    });
+  });
 }

test/watch_together/watch_together_provider_test.dart

@@ -1,3 +1,5 @@
+import 'dart:async';
+
 import 'package:flutter_test/flutter_test.dart';
 import 'package:plezy/media/ids.dart';
 import 'package:plezy/watch_together/models/watch_session.dart';
@@ -142,6 +144,133 @@ void main() {
     });
   });
 
+  group('WatchTogetherProvider — media switch dispatch', () {
+    test('dispatches once with typed args and suppresses the key after success', () async {
+      final p = WatchTogetherProvider();
+      final calls = <(String, String, String)>[];
+      p.onMediaSwitched = (ratingKey, serverId, mediaTitle) async {
+        calls.add((ratingKey, serverId, mediaTitle));
+        return true;
+      };
+
+      p.debugHandleMediaState('rk1', 's1', 'Ep 1');
+      await Future<void>.delayed(Duration.zero);
+      expect(calls, [('rk1', 's1', 'Ep 1')]);
+
+      // Heartbeat repeat of the handled key: no re-dispatch.
+      p.debugHandleMediaState('rk1', 's1', 'Ep 1');
+      await Future<void>.delayed(Duration.zero);
+      expect(calls.length, 1);
+      p.dispose();
+    });
+
+    test('a false result is retried on the next heartbeat state', () async {
+      final p = WatchTogetherProvider();
+      var calls = 0;
+      p.onMediaSwitched = (ratingKey, serverId, mediaTitle) async {
+        calls++;
+        return calls > 1; // Fail once, then succeed.
+      };
+
+      p.debugHandleMediaState('rk1', 's1', null);
+      await Future<void>.delayed(Duration.zero);
+      p.debugHandleMediaState('rk1', 's1', null);
+      await Future<void>.delayed(Duration.zero);
+      expect(calls, 2);
+
+      p.debugHandleMediaState('rk1', 's1', null);
+      await Future<void>.delayed(Duration.zero);
+      expect(calls, 2); // Second attempt succeeded; key now handled.
+      p.dispose();
+    });
+
+    test('a throwing callback is contained and retried', () async {
+      final p = WatchTogetherProvider();
+      var calls = 0;
+      p.onMediaSwitched = (ratingKey, serverId, mediaTitle) async {
+        calls++;
+        throw StateError('network down');
+      };
+
+      expect(() => p.debugHandleMediaState('rk1', 's1', null), returnsNormally);
+      await Future<void>.delayed(Duration.zero);
+      p.debugHandleMediaState('rk1', 's1', null);
+      await Future<void>.delayed(Duration.zero);
+      expect(calls, 2);
+      p.dispose();
+    });
+
+    test('no double dispatch while a switch is pending, even for another key', () async {
+      final p = WatchTogetherProvider();
+      final pending = Completer<bool>();
+      final calls = <String>[];
+      p.onMediaSwitched = (ratingKey, serverId, mediaTitle) {
+        calls.add(ratingKey);
+        return pending.future;
+      };
+
+      p.debugHandleMediaState('rk1', 's1', null);
+      p.debugHandleMediaState('rk1', 's1', null);
+      p.debugHandleMediaState('rk2', 's1', null); // Serialized behind rk1.
+      await Future<void>.delayed(Duration.zero);
+      expect(calls, ['rk1']);
+
+      pending.complete(false);
+      await Future<void>.delayed(Duration.zero);
+      // The slot is free again; the next heartbeat re-dispatches.
+      p.debugHandleMediaState('rk2', 's1', null);
+      await Future<void>.delayed(Duration.zero);
+      expect(calls, ['rk1', 'rk2']);
+      p.dispose();
+    });
+
+    test('onPlayerMediaSwitched takes priority over onMediaSwitched', () async {
+      final p = WatchTogetherProvider();
+      final calls = <String>[];
+      p.onMediaSwitched = (ratingKey, serverId, mediaTitle) async {
+        calls.add('main');
+        return true;
+      };
+      p.onPlayerMediaSwitched = (ratingKey, serverId, mediaTitle) async {
+        calls.add('player');
+        return true;
+      };
+
+      p.debugHandleMediaState('rk1', 's1', null);
+      await Future<void>.delayed(Duration.zero);
+      expect(calls, ['player']);
+      p.dispose();
+    });
+
+    test('markCurrentPlaybackHandled suppresses the marked key', () async {
+      final p = WatchTogetherProvider();
+      var calls = 0;
+      p.onMediaSwitched = (ratingKey, serverId, mediaTitle) async {
+        calls++;
+        return true;
+      };
+
+      p.markCurrentPlaybackHandled(ratingKey: 'rk1', serverId: ServerId('s1'));
+      p.debugHandleMediaState('rk1', 's1', null);
+      await Future<void>.delayed(Duration.zero);
+      expect(calls, 0);
+      p.dispose();
+    });
+
+    test('a blank serverId is ignored without throwing', () {
+      final p = WatchTogetherProvider();
+      var calls = 0;
+      p.onMediaSwitched = (ratingKey, serverId, mediaTitle) async {
+        calls++;
+        return true;
+      };
+
+      expect(() => p.debugHandleMediaState('rk1', '', null), returnsNormally);
+      expect(calls, 0);
+      p.dispose();
+    });
+  });
+
   group('WatchTogetherProvider — leaveSession safety', () {
     test('leaveSession on a fresh provider is a no-op (no notify)', () async {
       final p = WatchTogetherProvider();