fix(route): fix verified routing lifecycle invariants (#332)

* Fix verified routing lifecycle invariants

* Compare delivered candidates against stalled routes

docs/adr/0005-automatic-hybrid-media-routing.md

@@ -181,6 +181,12 @@ the LiveKit namespace to Screener and complete startup ownership/cleanup before
 accepting traffic; [self-hosting operations](../operations/self-hosting.md) owns
 that procedure.
 
+Retiring a publication generation releases every physical SFU handle. An old
+SFU root that still carries Peer descendants remains only as an inactive graph
+anchor until those descendants are reassigned; a childless old root is removed
+immediately. This preserves one source-reachable transition graph without
+keeping old media active.
+
 Only confirmed LiveKit media-participant absence retires an abandoned Host
 publication. A Host signaling disconnect keeps the generation charged while the
 exact LiveKit Host participant remains; one bounded control-plane check releases
@@ -240,6 +246,12 @@ time and the old route keeps playing. The same Viewer compares fresh,
 overlapping old and candidate receive windows. A P2P candidate commits only after first decoded
 frame and three consecutive windows with no freeze/pause, no lower pixel area or
 rounded FPS, and a strict improvement in at least one of those dimensions.
+An otherwise complete current-route window with zero decoded frames is compared
+as zero delivered pixels, FPS, and bitrate. It can approve only a temporally
+overlapping candidate window that actually decoded video and satisfies the same
+freeze, pause, and P2P/SFU partial-order checks. If both routes decode nothing,
+the source may be stalled, so the window remains unknown and cannot justify a
+move.
 Bitrate does not rank P2P candidates because codec and content phase make it
 non-monotonic. Three consecutive comparable non-improving windows reject the
 candidate and advance the existing cursor without resetting the deadline. A

docs/product/routing-transport.md

@@ -113,9 +113,11 @@ per-share convergence gate is enabled, Screener uses an exact persistent native
 sender limitation only to trigger one measured experiment through the existing
 serial operation. The same Viewer must then prove a P2P candidate strictly
 improves delivered pixel area or rounded FPS without regressing either dimension
-or producing a freeze/pause across three fresh overlapping windows. A persistently
-limited exact candidate sender cannot commit. Ordinary quality moves remain
-Peer-to-Peer. SFU may
+or producing a freeze/pause across three fresh overlapping windows. A complete
+zero-frame current window counts as zero delivery only when an overlapping
+candidate actually decodes clean video; two zero-frame paths remain unknown. A
+persistently limited exact candidate sender cannot commit. Ordinary quality
+moves remain Peer-to-Peer. SFU may
 join quality work only as Host fanout relief when every current Host-origin Peer
 edge, with at least two such edges, is persistently degraded; one Viewer moves
 before the controller observes the new topology again. During that overlap, the

src/client/lib/signaling.ts

@@ -89,6 +89,7 @@ export class SignalingClient {
   private watchdogTimer: number | null = null;
   private watchdogDeadlineMs = 0;
   private pendingChallenge: PendingSignalingChallenge | null = null;
+  private previousChallenge: PendingSignalingChallenge | null = null;
   private visibilityListenerAttached = false;
   private hostQualityIntent: HostQualityIntent | null = null;
 
@@ -507,6 +508,9 @@ export class SignalingClient {
         : this.challengeSequence + 1;
     const sequence = this.challengeSequence;
     socket.send(JSON.stringify({ type: "signaling-challenge", sequence }));
+    this.previousChallenge = this.pendingChallenge?.generation === generation
+      ? this.pendingChallenge
+      : null;
     this.pendingChallenge = { generation, sequence, confirm };
     this.armSignalingWatchdog(SIGNALING_CHALLENGE_TIMEOUT_MS);
   }
@@ -516,10 +520,14 @@ export class SignalingClient {
     generation: number,
   ): void {
     const pending = this.pendingChallenge;
+    const previous = this.previousChallenge;
     if (
-      !pending ||
-      pending.generation !== generation ||
-      pending.sequence !== sequence ||
+      ![pending, previous].some(
+        (challenge) =>
+          challenge !== null &&
+          challenge.generation === generation &&
+          challenge.sequence === sequence,
+      ) ||
       generation !== this.socketGeneration
     ) {
       return;
@@ -553,6 +561,7 @@ export class SignalingClient {
 
   private clearSignalingWatchdog(): void {
     this.pendingChallenge = null;
+    this.previousChallenge = null;
     if (this.watchdogTimer !== null) {
       window.clearTimeout(this.watchdogTimer);
       this.watchdogTimer = null;

src/client/media/candidate-quality-probe.ts

@@ -9,6 +9,30 @@ interface DeliveredVideoSample {
   bitrateKbps: number | null;
 }
 
+function deliveredNothingSample(
+  metrics: ConnectionMetrics,
+): DeliveredVideoSample | null {
+  const { sampleTimestampMs, sampleWindowMs, intervalFramesDecoded } = metrics;
+  if (
+    sampleTimestampMs === null ||
+    sampleWindowMs === null ||
+    !Number.isFinite(sampleTimestampMs) ||
+    !Number.isFinite(sampleWindowMs) ||
+    sampleWindowMs < 1_000 ||
+    sampleWindowMs > 5_000 ||
+    intervalFramesDecoded !== 0
+  ) {
+    return null;
+  }
+  return {
+    timestampMs: sampleTimestampMs,
+    windowMs: sampleWindowMs,
+    pixels: 0,
+    framesPerSecond: 0,
+    bitrateKbps: 0,
+  };
+}
+
 function deliveredVideoSample(
   metrics: ConnectionMetrics,
 ): DeliveredVideoSample | null {
@@ -60,7 +84,9 @@ function comparableSamples(
   current: ConnectionMetrics,
   candidate: ConnectionMetrics,
 ): [DeliveredVideoSample, DeliveredVideoSample] | null {
-  const currentVideo = deliveredVideoSample(current);
+  // A dead incumbent is comparable only when the candidate actually delivers.
+  const currentVideo =
+    deliveredVideoSample(current) ?? deliveredNothingSample(current);
   const candidateVideo = deliveredVideoSample(candidate);
   if (
     !currentVideo ||
@@ -139,16 +165,29 @@ class ConsecutiveCandidateQualityProbe {
     current: ConnectionMetrics | null,
     candidate: ConnectionMetrics,
   ): CandidateQualityProbeResult {
+    const candidateTimestampMs = candidate.sampleTimestampMs;
     if (
-      candidate.sampleTimestampMs === null ||
-      candidate.sampleTimestampMs === this.lastCandidateTimestampMs
+      candidateTimestampMs === null ||
+      !Number.isFinite(candidateTimestampMs)
     ) {
+      this.resetRuns();
       return "pending";
     }
-    this.lastCandidateTimestampMs = candidate.sampleTimestampMs;
+    if (candidateTimestampMs === this.lastCandidateTimestampMs) {
+      return "pending";
+    }
+    if (
+      this.lastCandidateTimestampMs !== null &&
+      candidateTimestampMs < this.lastCandidateTimestampMs
+    ) {
+      this.resetRuns();
+      return "pending";
+    }
+    this.lastCandidateTimestampMs = candidateTimestampMs;
     if (
       !current ||
       current.sampleTimestampMs === null ||
+      !Number.isFinite(current.sampleTimestampMs) ||
       (this.lastCurrentTimestampMs !== null &&
         current.sampleTimestampMs <= this.lastCurrentTimestampMs)
     ) {

src/client/media/host-sfu-route.ts

@@ -234,11 +234,11 @@ export class HostSfuRoute {
     let slot: HostPublisherSlot;
     const publisher =
       this.events.createPublisher?.(
-        () => this.handleFailure(slot),
+        () => this.failPublisherSlot(slot),
         (metrics) => this.handlePublisherStats(slot, metrics),
       ) ??
       new SfuPublisher({
-        onDisconnected: () => this.handleFailure(slot),
+        onDisconnected: () => this.failPublisherSlot(slot),
         onStats: (metrics) => this.handlePublisherStats(slot, metrics),
       });
     slot = {
@@ -258,11 +258,20 @@ export class HostSfuRoute {
       if (!connected) {
         if (this.pending === slot && this.route.owns(token)) {
           this.handleFailure(slot);
+        } else {
+          if (this.pending === slot) {
+            this.pending = null;
+          }
+          slot.failed = true;
         }
         await disconnectPublisher(publisher);
         return;
       }
       if (this.pending !== slot || !this.route.owns(token)) {
+        if (this.pending === slot) {
+          this.pending = null;
+        }
+        slot.failed = true;
         await disconnectPublisher(publisher);
         return;
       }
@@ -273,6 +282,7 @@ export class HostSfuRoute {
           if (this.pending === slot) {
             this.pending = null;
           }
+          slot.failed = true;
           await disconnectPublisher(publisher);
           return;
         }
@@ -285,8 +295,8 @@ export class HostSfuRoute {
         await this.queueActivation(token);
       }
     } catch {
+      this.failPublisherSlot(slot);
       await disconnectPublisher(publisher);
-      this.handleFailure(slot);
     }
   }
 
@@ -319,15 +329,39 @@ export class HostSfuRoute {
       : null;
   }
 
-  replaceStream(stream: MediaStream): Promise<boolean> {
+  async replaceStream(stream: MediaStream): Promise<boolean> {
     const slots = this.publishingSlots();
-    return slots.length === 0
-      ? Promise.resolve(true)
-      : Promise.all(
-          slots.map((slot) =>
-            slot.publisher.replaceStream(stream).catch(() => false),
-          ),
-        ).then((results) => results.every(Boolean));
+    const initialActive = this.active;
+    if (slots.length === 0) {
+      return true;
+    }
+
+    const results = await Promise.all(
+      slots.map((slot) => slot.publisher.replaceStream(stream).catch(() => false)),
+    );
+    let replaced = true;
+    await this.queueTransition(async () => {
+      const currentActive = this.active;
+      if (currentActive) {
+        const activeIndex = slots.indexOf(currentActive);
+        replaced = activeIndex < 0 || results[activeIndex] === true;
+      } else if (initialActive) {
+        const activeIndex = slots.indexOf(initialActive);
+        const stillPlanned =
+          this.route.getPlannedAssignment()?.sfuPublicationGeneration ===
+          initialActive.publicationGeneration;
+        replaced =
+          activeIndex < 0 || results[activeIndex] === true || !stillPlanned;
+      }
+
+      for (const [index, slot] of slots.entries()) {
+        if (results[index] === false) {
+          this.failPublisherSlot(slot);
+          await disconnectPublisher(slot.publisher);
+        }
+      }
+    });
+    return replaced;
   }
 
   private publishingSlots(): HostPublisherSlot[] {
@@ -337,18 +371,6 @@ export class HostSfuRoute {
     );
   }
 
-  async failActivePublisher(): Promise<void> {
-    await this.queueTransition(async () => {
-      const slot = this.active;
-      if (!slot) {
-        return;
-      }
-      this.active = null;
-      await disconnectPublisher(slot.publisher);
-      this.handleFailure(slot);
-    });
-  }
-
   async disconnect(): Promise<void> {
     if (this.closed) {
       return;
@@ -363,6 +385,12 @@ export class HostSfuRoute {
       const active = this.active;
       this.pending = null;
       this.active = null;
+      if (pending) {
+        pending.failed = true;
+      }
+      if (active) {
+        active.failed = true;
+      }
       if (active?.active) {
         await active.publisher.deactivate().catch(() => false);
       }
@@ -455,12 +483,14 @@ export class HostSfuRoute {
     this.active = pending;
     if (!this.route.markMediaActive(token)) {
       this.active = previous;
+      pending.failed = true;
       await disconnectPublisher(pending.publisher);
       return;
     }
     this.lastFailureStage = null;
     this.recovery = null;
     if (previous && previous !== pending) {
+      previous.failed = true;
       if (previous.active) {
         await previous.publisher.deactivate().catch(() => false);
       }
@@ -473,12 +503,12 @@ export class HostSfuRoute {
     const profile = this.events.getProfile();
     const videoCodec = this.events.getVideoCodec();
     if (!stream) {
-      this.handleFailure(slot);
+      this.failPublisherSlot(slot);
       return false;
     }
     try {
       if (!(await slot.publisher.activate(stream, profile, videoCodec))) {
-        this.handleFailure(slot);
+        this.failPublisherSlot(slot);
         await disconnectPublisher(slot.publisher);
         return false;
       }
@@ -493,7 +523,7 @@ export class HostSfuRoute {
         latestStream !== stream &&
         !(await slot.publisher.replaceStream(latestStream))
       ) {
-        this.handleFailure(slot);
+        this.failPublisherSlot(slot);
         await disconnectPublisher(slot.publisher);
         return false;
       }
@@ -502,13 +532,13 @@ export class HostSfuRoute {
         !qualitySettingsEqual(latestProfile, profile) &&
         !(await slot.publisher.updateProfile(latestProfile))
       ) {
-        this.handleFailure(slot);
+        this.failPublisherSlot(slot);
         await disconnectPublisher(slot.publisher);
         return false;
       }
       return this.pending === slot && this.ownsPublisherSlot(slot);
     } catch {
-      this.handleFailure(slot);
+      this.failPublisherSlot(slot);
       await disconnectPublisher(slot.publisher);
       return false;
     }
@@ -536,6 +566,7 @@ export class HostSfuRoute {
     }
     const pending = this.pending;
     this.pending = null;
+    pending.failed = true;
     void disconnectPublisher(pending.publisher);
   }
 
@@ -545,6 +576,7 @@ export class HostSfuRoute {
     }
     const active = this.active;
     this.active = null;
+    active.failed = true;
     await this.disconnectRetiredPublisher(active);
   }
 
@@ -559,6 +591,7 @@ export class HostSfuRoute {
     }
     const active = this.active;
     this.active = null;
+    active.failed = true;
     return active;
   }
 
@@ -577,15 +610,16 @@ export class HostSfuRoute {
     }
     const wasActive = this.active === slot;
     const wasPending = this.pending === slot;
+    if (!wasActive && !wasPending) {
+      slot.failed = true;
+      return;
+    }
     const assignment = this.route.getPlannedAssignment();
     const revision = this.route.getRevision();
     const phase = this.route.getPhase();
     const matchesPlannedRoute =
       revision === slot.revision &&
       assignment?.sfuPublicationGeneration === slot.publicationGeneration;
-    if (!wasActive && !wasPending && !matchesPlannedRoute) {
-      return;
-    }
     this.lastFailureStage = slot.publisher.getFailureStage?.() ?? "transport";
     slot.failed = true;
     if (wasPending) {
@@ -610,6 +644,14 @@ export class HostSfuRoute {
     this.requestRecovery(revision);
   }
 
+  private failPublisherSlot(slot: HostPublisherSlot): void {
+    if (this.pending !== slot && this.active !== slot) {
+      slot.failed = true;
+      return;
+    }
+    this.handleFailure(slot);
+  }
+
   private requestRecovery(revision: number): void {
     if (!this.recovery || this.recovery.revision !== revision) {
       this.recovery = { revision, refreshed: false };

src/client/media/viewer-sfu-route.ts

@@ -502,7 +502,7 @@ export class ViewerSfuRoute {
           this.events.onSfuState?.(state, slot.revision);
         }
       },
-      onDisconnected: () => this.handleFailure(slot),
+      onDisconnected: () => this.failSubscriberSlot(slot),
     };
     const subscriber =
       this.events.createSubscriber?.(subscriberEvents) ??
@@ -540,19 +540,28 @@ export class ViewerSfuRoute {
         if (this.pending === slot && currentToken) {
           slot.revision = currentToken.revision;
           this.handleFailure(slot);
+        } else {
+          if (this.pending === slot) {
+            this.pending = null;
+          }
+          slot.failed = true;
         }
         await disconnectSubscriber(subscriber);
         return;
       }
       const currentToken = this.currentPendingToken(slot, token);
       if (this.pending !== slot || !currentToken) {
+        if (this.pending === slot) {
+          this.pending = null;
+        }
+        slot.failed = true;
         await disconnectSubscriber(subscriber);
         return;
       }
       slot.revision = currentToken.revision;
       slot.connected = true;
       if (!subscriber.activate()) {
-        this.handleFailure(slot);
+        this.failSubscriberSlot(slot);
         await disconnectSubscriber(subscriber);
         return;
       }
@@ -564,8 +573,8 @@ export class ViewerSfuRoute {
         await this.queueActiveRoute(currentToken, true);
       }
     } catch {
+      this.failSubscriberSlot(slot);
       await disconnectSubscriber(subscriber);
-      this.handleFailure(slot);
     }
   }
 
@@ -596,6 +605,12 @@ export class ViewerSfuRoute {
       const active = this.active;
       this.pending = null;
       this.active = null;
+      if (pending) {
+        pending.failed = true;
+      }
+      if (active) {
+        active.failed = true;
+      }
       this.events.onSfuUpdate?.(
         null,
         active?.revision ?? pending?.revision ?? this.route.getRevision() ?? 0,
@@ -826,6 +841,7 @@ export class ViewerSfuRoute {
     this.active = slot;
     if (!this.route.markMediaActive(token)) {
       this.active = previous;
+      slot.failed = true;
       await disconnectSubscriber(slot.subscriber);
       return;
     }
@@ -836,6 +852,7 @@ export class ViewerSfuRoute {
     slot.mediaAvailable = true;
     this.events.onSfuStream(slot.stream, assignment, true, slot.revision);
     if (previous && previous !== slot) {
+      previous.failed = true;
       try {
         previous.subscriber.deactivate();
       } catch {
@@ -880,6 +897,7 @@ export class ViewerSfuRoute {
     }
     const pending = this.pending;
     this.pending = null;
+    pending.failed = true;
     void disconnectSubscriber(pending.subscriber);
   }
 
@@ -889,6 +907,7 @@ export class ViewerSfuRoute {
     }
     const active = this.active;
     this.active = null;
+    active.failed = true;
     this.events.onSfuUpdate?.(null, active.revision);
     await this.disconnectRetiredSubscriber(active);
   }
@@ -908,6 +927,7 @@ export class ViewerSfuRoute {
     }
     const active = this.active;
     this.active = null;
+    active.failed = true;
     this.events.onSfuUpdate?.(null, active.revision);
     return active;
   }
@@ -930,8 +950,13 @@ export class ViewerSfuRoute {
       return;
     }
     const wasActive = this.active === slot;
+    const wasPending = this.pending === slot;
+    if (!wasActive && !wasPending) {
+      slot.failed = true;
+      return;
+    }
     const failedManualReconnect =
-      this.pending === slot &&
+      wasPending &&
       this.manualReconnectRevision === slot.revision &&
       this.active?.publicationGeneration === slot.publicationGeneration &&
       this.active.activated &&
@@ -980,6 +1005,14 @@ export class ViewerSfuRoute {
     this.requestRecovery(revision);
   }
 
+  private failSubscriberSlot(slot: ViewerSubscriberSlot): void {
+    if (this.pending !== slot && this.active !== slot) {
+      slot.failed = true;
+      return;
+    }
+    this.handleFailure(slot);
+  }
+
   private requestRecovery(revision: number): void {
     if (!this.recovery || this.recovery.revision !== revision) {
       this.recovery = { revision, refreshed: false };

src/client/pages/HostPage.tsx

@@ -2146,14 +2146,6 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
         activeSfuRoute && hostSfuRouteRef.current === activeSfuRoute
           ? activeSfuRoute.getQualityWarning()
           : null;
-      if (
-        !sfuReplaced &&
-        activeSfuRoute &&
-        hostSfuRouteRef.current === activeSfuRoute
-      ) {
-        await activeSfuRoute.failActivePublisher();
-      }
-
       previousStream.getTracks().forEach((track) => track.stop());
       if (retiringStreamRef.current === previousStream) {
         retiringStreamRef.current = null;

src/client/pages/ViewerPage.tsx

@@ -94,10 +94,10 @@ import type {
   PeerSnapshot,
 } from "../types";
 import {
+  createOwnedViewerRestartSender,
   limitMediaAssignment,
   MAX_ENDPOINT_MEDIA_CHILDREN,
   type MediaAssignment,
-  viewerRestartMessage,
   viewerSignalMessage,
 } from "../webrtc/media-assignment";
 import { ViewerPeer } from "../webrtc/viewer-peer";
@@ -1243,7 +1243,20 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
         {
           sendSignal: (targetPeerId, payload) =>
             signal.send(viewerSignalMessage(peerAssisted, targetPeerId, payload)),
-          sendRestartRequest: () => false,
+          sendRestartRequest: createOwnedViewerRestartSender(
+            peerAssisted,
+            (targetPeerId, connectionId) => {
+              const identity = peer.getConnectionIdentity();
+              return (
+                active &&
+                pendingPeer !== probe &&
+                peerRef.current === peer &&
+                identity?.parentPeerId === targetPeerId &&
+                identity.connectionId === connectionId
+              );
+            },
+            (message) => signal.send(message),
+          ),
           onStream: (stream) => {
             if (pendingPeer === probe) {
               probe.stream = stream;
@@ -1357,15 +1370,19 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
             signal.send(
               viewerSignalMessage(peerAssisted, targetPeerId, payload),
             ),
-          sendRestartRequest: (targetPeerId, connectionId, rebuild) =>
-            signal.send(
-              viewerRestartMessage(
-                peerAssisted,
-                targetPeerId,
-                connectionId,
-                rebuild,
-              ),
-            ),
+          sendRestartRequest: createOwnedViewerRestartSender(
+            peerAssisted,
+            (targetPeerId, connectionId) => {
+              const identity = peer.getConnectionIdentity();
+              return (
+                active &&
+                peerRef.current === peer &&
+                identity?.parentPeerId === targetPeerId &&
+                identity.connectionId === connectionId
+              );
+            },
+            (message) => signal.send(message),
+          ),
           onStream: (nextStream) => {
             if (active) {
               bindRemoteStream(nextStream, currentRouteRevision);

src/client/sfu/publisher.ts

@@ -136,6 +136,7 @@ export class SfuPublisher {
       }
 
       const room = new sdk.Room({
+        disconnectOnPageLeave: false,
         dynacast: true,
         stopLocalTrackOnUnpublish: false,
       });
@@ -441,6 +442,7 @@ export class SfuPublisher {
         if (!this.owns(room, generation)) {
           return false;
         }
+        this.failureStage = "source";
         const failureWarning = say("host.fail.sfuSwitch");
 
         // A rejected publish/unpublish may have changed server state without

src/client/sfu/subscriber.ts

@@ -278,7 +278,10 @@ export class SfuSubscriber {
             }
           }
           this.desiredTrackSids.clear();
-          this.clearMedia(true);
+          // LiveKit full reconnect removes participants before it emits
+          // Reconnecting. Keep any route-owned proof intent until the route is
+          // deactivated or the subscriber terminates.
+          this.clearMedia(true, true);
         }
       },
     );
@@ -290,6 +293,7 @@ export class SfuSubscriber {
     room.on(sdk.RoomEvent.Reconnected, () => {
       if (this.owns(room, generation) && this.state === "active") {
         this.reconcileHostSubscriptions(room, sdk, generation);
+        this.startDecodedFrameProof();
         this.events.onState?.("connected");
         void this.updateStats();
       }
@@ -625,10 +629,14 @@ export class SfuSubscriber {
     this.audioStatsInFlight = null;
   }
 
-  private clearMedia(notify: boolean): void {
+  private clearMedia(notify: boolean, preserveDecodedFrameProof = false): void {
     const hadStream = this.streamEmitted;
     const hadVideo = this.video !== null;
-    this.stopDecodedFrameProof();
+    if (preserveDecodedFrameProof) {
+      this.cancelDecodedFrameObserver();
+    } else {
+      this.stopDecodedFrameProof();
+    }
     if (this.state !== "active") {
       this.stopStats();
     }

src/client/webrtc/host-peer.ts

@@ -74,6 +74,8 @@ export class HostPeer {
   private senderMutationTail: Promise<void> = Promise.resolve();
   private negotiationTail: Promise<void> = Promise.resolve();
   private startupVideoProfilePending: boolean;
+  private connectedOnce = false;
+  private awaitingReconnect = false;
   private snapshot: PeerSnapshot;
 
   constructor(
@@ -360,6 +362,20 @@ export class HostPeer {
       });
     });
     this.connection.addEventListener("connectionstatechange", () => {
+      const state = this.connection.connectionState;
+      if (state === "connected") {
+        const replayProfile = this.awaitingReconnect;
+        this.connectedOnce = true;
+        this.awaitingReconnect = false;
+        if (replayProfile) {
+          this.appliedVideoProfile = null;
+          this.appliedAudioQuality = null;
+          this.appliedAudioSenderParameters = null;
+          void this.updateProfile(this.desiredProfile);
+        }
+      } else if (this.connectedOnce && state !== "closed") {
+        this.awaitingReconnect = true;
+      }
       this.emit();
     });
     this.connection.addEventListener("iceconnectionstatechange", () => this.emit());

src/client/webrtc/media-assignment.ts

@@ -79,3 +79,26 @@ export function viewerRestartMessage(
       }
     : { type: "restart-request", connectionId, rebuild };
 }
+
+export function createOwnedViewerRestartSender(
+  peerAssisted: boolean,
+  ownsActivePeer: (targetPeerId: string, connectionId: string) => boolean,
+  send: (
+    message: Extract<ClientMessage, { type: "restart-request" }>,
+  ) => boolean,
+): (
+  targetPeerId: string,
+  connectionId: string,
+  rebuild: boolean,
+) => boolean {
+  return (targetPeerId, connectionId, rebuild) =>
+    ownsActivePeer(targetPeerId, connectionId) &&
+    send(
+      viewerRestartMessage(
+        peerAssisted,
+        targetPeerId,
+        connectionId,
+        rebuild,
+      ),
+    );
+}

src/server/hybrid-media-router.ts

@@ -77,6 +77,8 @@ type SfuDrainTask = SfuDrainTarget & {
 interface HostOfflineCheck {
   hostPeerId: string;
   fence: SfuResourceFence;
+  hostSessionId: string;
+  connectionId: string;
   timer: NodeJS.Timeout;
 }
 
@@ -216,6 +218,21 @@ export class HybridMediaRouter {
   connectParticipant(input: AuthenticatedRouteParticipant): HybridAuthenticationState {
     const room = this.room(input.roomId);
     if (input.role === "host") {
+      if (room.controller && room.hostPeerId !== input.peerId) {
+        this.clearDeadline(room);
+        for (const viewerPeerId of this.options.roomStore.getViewerPeerIds(
+          input.roomId,
+        )) {
+          this.options.deleteConnectionId(input.roomId, viewerPeerId);
+        }
+        this.releaseResources(
+          room.controller.rebindHostIdentity(
+            input.peerId,
+            input.sessionId,
+            this.now(),
+          ),
+        );
+      }
       room.hostPeerId = input.peerId;
       this.cancelHostOfflineCheck(input.roomId);
       if (!room.controller) this.createController(input.roomId, room, input);
@@ -1181,6 +1198,7 @@ export class HybridMediaRouter {
         route: this.debugTuple(roomId, plan.tuple),
         cursor: operation.cursor,
       });
+      let beginGuard = guard;
       let currentOperation = controller.snapshot().operation;
       if (plan.endpointTransition.kind === "bounded-gap") {
         const beforeGap = controller.snapshot().revision;
@@ -1197,14 +1215,15 @@ export class HybridMediaRouter {
         }
         if (controller.snapshot().revision !== beforeGap) this.broadcastActive(roomId, room);
         currentOperation = controller.snapshot().operation;
-      }
-      const currentPlan = currentOperation?.candidates[currentOperation.cursor];
-      if (!currentOperation || !currentPlan) {
-        this.releaseReservation(preparation.prepared.reservation);
-        continue;
+        const currentPlan = currentOperation?.candidates[currentOperation.cursor];
+        if (!currentOperation || !currentPlan) {
+          this.releaseReservation(preparation.prepared.reservation);
+          continue;
+        }
+        beginGuard = cursorGuard(currentOperation, currentPlan);
       }
       const begun = controller.beginCurrentCandidate({
-        guard: cursorGuard(currentOperation, currentPlan),
+        guard: beginGuard,
         nowMs: this.now(),
         connectionId: preparation.prepared.connectionId,
         reservation: preparation.prepared.reservation,
@@ -1841,23 +1860,34 @@ export class HybridMediaRouter {
     });
   }
 
-  private activeSfuFence(roomId: string): SfuResourceFence | null {
+  private activeHostPublication(
+    roomId: string,
+  ): Omit<HostOfflineCheck, "hostPeerId" | "timer"> | null {
     const publication = this.rooms.get(roomId)?.controller?.snapshot().hostPublication;
-    return publication?.physicalActive && publication.resource.kind === "sfu-publication"
-      ? publication.resource.fence
+    return publication?.physicalActive &&
+      publication.resource.kind === "sfu-publication"
+      ? {
+          fence: publication.resource.fence,
+          hostSessionId: publication.hostSessionId,
+          connectionId: publication.connectionId,
+        }
       : null;
   }
 
   private scheduleHostOfflineCheck(roomId: string, hostPeerId: string): void {
     this.cancelHostOfflineCheck(roomId);
     const fallback = this.options.sfuFallback;
-    const fence = this.activeSfuFence(roomId);
-    if (!fallback || !fence || this.closing) return;
+    const publication = this.activeHostPublication(roomId);
+    if (!fallback || !publication || this.closing) return;
     const timer = setTimeout(() => {
       void this.checkHostOffline(roomId).catch(() => undefined);
     }, fallback.hostOfflineCheckMs ?? DEFAULT_HOST_OFFLINE_CHECK_MS);
     timer.unref();
-    this.hostOfflineChecks.set(roomId, { hostPeerId, fence, timer });
+    this.hostOfflineChecks.set(roomId, {
+      hostPeerId,
+      ...publication,
+      timer,
+    });
   }
 
   private async checkHostOffline(roomId: string): Promise<void> {
@@ -1872,6 +1902,7 @@ export class HybridMediaRouter {
     try {
       exists = await fallback.roomControl.hostParticipantExists(check.fence);
     } catch {
+      if (this.hostOfflineChecks.get(roomId) !== check) return;
       this.hostOfflineChecks.delete(roomId);
       this.scheduleHostOfflineCheck(roomId, check.hostPeerId);
       return;
@@ -1887,6 +1918,22 @@ export class HybridMediaRouter {
     const snapshot = controller?.snapshot();
     const publication = snapshot?.hostPublication;
     if (!room || !controller || !snapshot || !publication) return;
+    if (
+      publication.resource.kind !== "sfu-publication" ||
+      publication.resource.fence.roomId !== check.fence.roomId ||
+      publication.resource.fence.shareGeneration !==
+        check.fence.shareGeneration ||
+      publication.resource.fence.publicationGeneration !==
+        check.fence.publicationGeneration ||
+      publication.generation !== check.fence.publicationGeneration ||
+      publication.hostSessionId !== check.hostSessionId ||
+      publication.connectionId !== check.connectionId ||
+      !publication.physicalActive
+    ) {
+      this.scheduleHostOfflineCheck(roomId, check.hostPeerId);
+      return;
+    }
+    const beforeRevision = snapshot.revision;
     controller.invalidateHostPublication({
       hostSessionId: publication.hostSessionId,
       routeRevision: snapshot.revision,
@@ -1902,6 +1949,19 @@ export class HybridMediaRouter {
         connectionId: publication.connectionId,
       }),
     );
+    const afterRetirement = controller.snapshot().hostPublication;
+    if (
+      afterRetirement?.generation === publication.generation &&
+      afterRetirement.connectionId === publication.connectionId &&
+      afterRetirement.physicalActive
+    ) {
+      if (controller.snapshot().revision !== beforeRevision) {
+        this.broadcastActive(roomId, room);
+      }
+      this.scheduleHostOfflineCheck(roomId, check.hostPeerId);
+      this.requestPump(roomId);
+      return;
+    }
     this.broadcastActive(roomId, room);
     this.requestPump(roomId);
   }

src/server/room-route-controller.ts

@@ -355,6 +355,7 @@ export class RoomRouteController<Resource = unknown> {
   private sfuPublisherQualityObservation?: SfuPublisherQualityObservation;
   private readonly qualityBaselinesPending = new Set<string>();
   private readonly directContinuations = new Map<string, DirectContinuation>();
+  private readonly retiringPublicationGenerations = new Set<string>();
   private sfuBootstrapIntent?: SfuBootstrapIntent;
   private readonly consumedSfuBootstrapOpportunities = new Map<
     string,
@@ -546,6 +547,78 @@ export class RoomRouteController<Resource = unknown> {
     return released;
   }
 
+  rebindHostIdentity(
+    peerId: string,
+    sessionId: string,
+    nowMs?: number,
+  ): readonly Resource[] {
+    const previousPeerId = this.options.hostPeerId;
+    if (peerId === previousPeerId) {
+      return this.upsertParticipant({
+        peerId,
+        role: "host",
+        sessionId,
+        effectiveDownstreamCapacity: this.options.endpointMediaCopyCapacity,
+      }, nowMs);
+    }
+    if (this.participants.has(peerId)) {
+      throw new Error("Route Host identity collides with another participant");
+    }
+    const host = this.participants.get(previousPeerId);
+    if (!host || host.role !== "host") {
+      throw new Error("Route Host identity is unavailable");
+    }
+
+    const released = new Set(
+      this.operation ? this.abortOperation(nowMs, "stale") : [],
+    );
+    for (const edge of this.upstreamByViewer.values()) {
+      if (edge.kind === "sfu" && edge.physicalActive) {
+        released.add(edge.resource);
+      }
+    }
+    if (this.hostPublication?.physicalActive) {
+      released.add(this.hostPublication.resource);
+    }
+    this.upstreamByViewer.clear();
+    this.hostPublication = null;
+    this.routeTimings.clear();
+    this.qualityObservations.clear();
+    this.senderQualityObservations.clear();
+    this.senderQualityBaselines.clear();
+    this.qualityBaselinesPending.clear();
+    this.sfuPublisherQualityObservation = undefined;
+    this.directContinuations.clear();
+    this.retiringPublicationGenerations.clear();
+    this.sfuBootstrapIntent = undefined;
+    this.consumedSfuBootstrapOpportunities.clear();
+    this.rootConvergenceRootPeerId = undefined;
+    for (const participant of this.participants.values()) {
+      participant.availabilityExhausted = false;
+      participant.bootstrapFailureReported = undefined;
+      participant.sfuFirstAtNextRoute = undefined;
+      participant.consumedCandidateOpportunities.clear();
+      if (
+        participant.role === "viewer" &&
+        participant.sessionId !== null &&
+        nowMs !== undefined
+      ) {
+        this.recordDemand(participant.peerId, nowMs, "edge-unavailable");
+      }
+    }
+
+    this.participants.delete(previousPeerId);
+    host.peerId = peerId;
+    host.sessionId = sessionId;
+    host.departureConfirmed = false;
+    this.participants.set(peerId, host);
+    this.options.hostPeerId = peerId;
+    this.revision = this.allocateRevision();
+    this.touchFacts();
+    this.assertGraph();
+    return [...released];
+  }
+
   disconnectSession(peerId: string, sessionId: string): boolean {
     const participant = this.participants.get(peerId);
     if (!participant || participant.sessionId !== sessionId) return false;
@@ -771,14 +844,20 @@ export class RoomRouteController<Resource = unknown> {
         (edge.kind === "peer" && edge.parentSessionId !== guard.parentSessionId)) {
       return [];
     }
-    edge.physicalActive = false;
-    edge.usable = false;
+    const released: Resource[] = [];
+    if (edge.kind === "sfu") {
+      this.retireSfuEdge(guard.childPeerId, edge, released);
+      this.pruneRetiringSfuAnchors();
+    } else {
+      edge.physicalActive = false;
+      edge.usable = false;
+    }
     this.clearQualityForParticipant(guard.childPeerId);
     child.availabilityExhausted = false;
     this.revision = this.allocateRevision();
     if (this.operation) this.operation.baseRevision = this.revision;
     this.touchFacts();
-    return edge.transport === "direct" ? [] : [edge.resource];
+    return released;
   }
 
   retireHostPublication(guard: {
@@ -793,13 +872,24 @@ export class RoomRouteController<Resource = unknown> {
         publication.hostSessionId !== guard.hostSessionId ||
         publication.generation !== guard.generation ||
         publication.connectionId !== guard.connectionId) return [];
+    const released: Resource[] = [];
+    for (const [viewerPeerId, edge] of this.upstreamByViewer) {
+      if (
+        edge.kind === "sfu" &&
+        edge.publicationGeneration === publication.generation
+      ) {
+        this.retireSfuEdge(viewerPeerId, edge, released);
+      }
+    }
     publication.physicalActive = false;
     publication.usable = false;
     this.clearSfuQuality();
     this.revision = this.allocateRevision();
     if (this.operation) this.operation.baseRevision = this.revision;
     this.touchFacts();
-    return [publication.resource];
+    released.push(publication.resource);
+    this.pruneRetiringSfuAnchors();
+    return released;
   }
 
   setPaused(paused: boolean, nowMs?: number): readonly Resource[] {
@@ -1020,10 +1110,6 @@ export class RoomRouteController<Resource = unknown> {
         });
       }
       this.senderQualityObservations.delete(input.childPeerId);
-      this.senderQualityBaselines.set(
-        input.childPeerId,
-        this.senderQualityBaselines.get(input.childPeerId) ?? true,
-      );
       const released =
         operation?.reason === "quality-convergence" &&
         operation.childPeerId === input.childPeerId
@@ -1968,7 +2054,7 @@ export class RoomRouteController<Resource = unknown> {
           attempt.tuple.publication === "reuse" ||
           (attempt.senderQualityConsecutiveHealthyWindows ?? 0) >=
             PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS);
-      if (!candidateQualityReady && candidateQualityState === "unknown") {
+      if (!candidateQualityReady && candidateQualityState !== "degraded") {
         return {
           accepted: true,
           committed: false,
@@ -2032,7 +2118,9 @@ export class RoomRouteController<Resource = unknown> {
         ? [...this.upstreamByViewer]
             .filter(
               ([childPeerId, edge]) =>
-                childPeerId !== operation.childPeerId && edge.kind === "sfu",
+                childPeerId !== operation.childPeerId &&
+                edge.kind === "sfu" &&
+                edge.physicalActive,
             )
             .map(([childPeerId]) => childPeerId)
         : [];
@@ -2167,6 +2255,7 @@ export class RoomRouteController<Resource = unknown> {
     this.participants.clear();
     this.routeTimings.clear();
     this.directContinuations.clear();
+    this.retiringPublicationGenerations.clear();
     this.sfuBootstrapIntent = undefined;
     this.consumedSfuBootstrapOpportunities.clear();
     this.rootConvergenceRootPeerId = undefined;
@@ -2536,6 +2625,7 @@ export class RoomRouteController<Resource = unknown> {
     this.senderQualityBaselines.set(operation.childPeerId, false);
     this.revision = attempt.revision;
     this.operation = undefined;
+    this.pruneRetiringSfuAnchors();
     this.touchFacts();
     if (createsHostRoot) {
       this.stageRootConvergence(operation.childPeerId);
@@ -3595,8 +3685,11 @@ export class RoomRouteController<Resource = unknown> {
       for (const participant of this.participants.values()) {
         if (participant.role !== "viewer" || !participant.departureConfirmed || this.childrenOf(participant.peerId).length > 0) continue;
         const edge = this.upstreamByViewer.get(participant.peerId);
-        if (edge && edge.transport !== "direct" && edge.physicalActive) released.push(edge.resource);
-        this.upstreamByViewer.delete(participant.peerId);
+        if (edge?.kind === "sfu") {
+          this.retireSfuEdge(participant.peerId, edge, released, false);
+        } else {
+          this.upstreamByViewer.delete(participant.peerId);
+        }
         this.clearQualityForParticipant(participant.peerId);
         this.participants.delete(participant.peerId);
         this.routeTimings.delete(participant.peerId);
@@ -3614,6 +3707,7 @@ export class RoomRouteController<Resource = unknown> {
       }
       this.revision = this.allocateRevision();
       this.touchFacts();
+      this.pruneRetiringSfuAnchors();
     }
     return removed;
   }
@@ -3695,9 +3789,13 @@ export class RoomRouteController<Resource = unknown> {
   private retireInvalidOperationEdge(childPeerId: string, released: Resource[]): boolean {
     const edge = this.upstreamByViewer.get(childPeerId);
     if (!edge || !this.edgeRequiresMove(childPeerId, edge)) return false;
-    if (edge.kind === "sfu" && edge.physicalActive) released.push(edge.resource);
-    edge.physicalActive = false;
-    edge.usable = false;
+    if (edge.kind === "sfu") {
+      this.retireSfuEdge(childPeerId, edge, released);
+      this.pruneRetiringSfuAnchors();
+    } else {
+      edge.physicalActive = false;
+      edge.usable = false;
+    }
     if (edge.kind === "sfu" && !this.hasSfuSubscribers() && this.hostPublication) {
       if (this.hostPublication.physicalActive) released.push(this.hostPublication.resource);
       this.hostPublication.physicalActive = false;
@@ -4313,24 +4411,83 @@ export class RoomRouteController<Resource = unknown> {
     const released: Resource[] = [];
     for (const [viewerPeerId, edge] of this.upstreamByViewer) {
       if (edge.kind !== "sfu" || edge.publicationGeneration !== generation) continue;
-      if (edge.physicalActive) released.push(edge.resource);
-      this.clearQualityForParticipant(viewerPeerId);
-      this.upstreamByViewer.delete(viewerPeerId);
+      this.retireSfuEdge(viewerPeerId, edge, released);
     }
     if (this.hostPublication?.generation === generation) {
       if (this.hostPublication.physicalActive) released.push(this.hostPublication.resource);
       this.hostPublication = null;
     }
+    this.pruneRetiringSfuAnchors();
     return released;
   }
 
+  private retireSfuEdge(
+    viewerPeerId: string,
+    edge: Extract<CommittedEdge<Resource>, { kind: "sfu" }>,
+    released: Resource[],
+    retainAnchor = this.childrenOf(viewerPeerId).length > 0,
+  ): void {
+    if (edge.physicalActive) released.push(edge.resource);
+    this.clearQualityForParticipant(viewerPeerId);
+    if (retainAnchor) {
+      edge.usable = false;
+      edge.physicalActive = false;
+      this.retiringPublicationGenerations.add(edge.publicationGeneration);
+    } else {
+      this.upstreamByViewer.delete(viewerPeerId);
+    }
+  }
+
+  private pruneRetiringSfuAnchors(): void {
+    let removed = true;
+    while (removed) {
+      removed = false;
+      for (const [viewerPeerId, edge] of this.upstreamByViewer) {
+        if (
+          edge.kind === "sfu" &&
+          !edge.physicalActive &&
+          this.retiringPublicationGenerations.has(
+            edge.publicationGeneration,
+          ) &&
+          this.childrenOf(viewerPeerId).length === 0
+        ) {
+          this.upstreamByViewer.delete(viewerPeerId);
+          removed = true;
+        }
+      }
+    }
+    for (const generation of this.retiringPublicationGenerations) {
+      if (![...this.upstreamByViewer.values()].some(
+        (edge) =>
+          edge.kind === "sfu" &&
+          !edge.physicalActive &&
+          edge.publicationGeneration === generation,
+      )) {
+        this.retiringPublicationGenerations.delete(generation);
+      }
+    }
+  }
+
   private assertGraph(): void {
     const host = this.participants.get(this.options.hostPeerId);
     if (!host || host.role !== "host") throw new Error("Route Host is missing");
     for (const [child, edge] of this.upstreamByViewer) {
       this.assertViewer(child);
       if (edge.kind === "peer" && !this.participants.has(edge.parentPeerId)) throw new Error("Route parent is missing");
-      if (edge.kind === "sfu" && this.hostPublication?.generation !== edge.publicationGeneration) throw new Error("SFU publication is stale");
+      if (
+        edge.kind === "sfu" &&
+        edge.physicalActive &&
+        this.hostPublication?.generation !== edge.publicationGeneration
+      ) {
+        throw new Error("SFU publication is stale");
+      }
+      if (
+        edge.kind === "sfu" &&
+        !edge.physicalActive &&
+        !this.retiringPublicationGenerations.has(edge.publicationGeneration)
+      ) {
+        throw new Error("Retiring SFU publication is unknown");
+      }
       const seen = new Set<string>();
       let current = child;
       while (current !== this.options.hostPeerId) {
@@ -4338,13 +4495,34 @@ export class RoomRouteController<Resource = unknown> {
         seen.add(current);
         const currentEdge = this.upstreamByViewer.get(current);
         if (!currentEdge) throw new Error("Route is not source-reachable");
-        if (currentEdge.kind === "sfu") break;
+        if (currentEdge.kind === "sfu") {
+          if (
+            currentEdge.physicalActive
+              ? this.hostPublication?.generation !== currentEdge.publicationGeneration
+              : !this.retiringPublicationGenerations.has(
+                  currentEdge.publicationGeneration,
+                )
+          ) {
+            throw new Error("Route terminates at an unknown SFU publication");
+          }
+          break;
+        }
         current = currentEdge.parentPeerId;
       }
     }
     for (const participant of this.participants.values()) if (this.usedSlots(participant.peerId) > this.options.endpointMediaCopyCapacity) {
       throw new Error("Committed route exceeds endpoint capacity");
     }
+    for (const generation of this.retiringPublicationGenerations) {
+      if (![...this.upstreamByViewer.values()].some(
+        (edge) =>
+          edge.kind === "sfu" &&
+          !edge.physicalActive &&
+          edge.publicationGeneration === generation,
+      )) {
+        throw new Error("Retiring SFU publication has no anchor");
+      }
+    }
   }
 
   private assertViewer(peerId: string): void {

tests/candidate-quality-probe.test.ts

@@ -233,3 +233,107 @@ describe("SFU quality probe", () => {
     expect(probe.observe(metrics(17_000), metrics(17_100))).toBe("pending");
   });
 });
+
+describe("zero-frame current route", () => {
+  const zeroFrames = (timestampMs: number) =>
+    metrics(timestampMs, { intervalFramesDecoded: 0 });
+  const degraded = (timestampMs: number) =>
+    metrics(timestampMs, {
+      frameWidth: 320,
+      frameHeight: 180,
+      framesPerSecond: 10,
+    });
+
+  it("approves clean delivery on both candidate route types", () => {
+    for (const probe of [new P2pQualityProbe(), new SfuQualityProbe()]) {
+      expect(probe.observe(zeroFrames(1_000), metrics(1_100))).toBe("pending");
+      expect(probe.observe(zeroFrames(3_000), metrics(3_100))).toBe("pending");
+      expect(probe.observe(zeroFrames(5_000), metrics(5_100))).toBe("approved");
+    }
+  });
+
+  it("continues a real improvement when the current route then stops decoding", () => {
+    const probe = new P2pQualityProbe();
+    expect(probe.observe(degraded(1_000), metrics(1_100))).toBe("pending");
+    expect(probe.observe(zeroFrames(3_000), metrics(3_100))).toBe("pending");
+    expect(probe.observe(zeroFrames(5_000), metrics(5_100))).toBe("approved");
+  });
+
+  it("never approves when both routes stop decoding", () => {
+    for (const probe of [new P2pQualityProbe(), new SfuQualityProbe()]) {
+      for (let index = 0; index < 15; index += 1) {
+        const timestampMs = 1_000 + index * 2_000;
+        expect(
+          probe.observe(zeroFrames(timestampMs), zeroFrames(timestampMs + 100)),
+        ).toBe("pending");
+      }
+    }
+  });
+
+  it("rejects a candidate that decodes only with freezes", () => {
+    for (const probe of [new P2pQualityProbe(), new SfuQualityProbe()]) {
+      for (const timestampMs of [1_000, 3_000]) {
+        expect(
+          probe.observe(
+            zeroFrames(timestampMs),
+            metrics(timestampMs + 100, { intervalFreezeCount: 1 }),
+          ),
+        ).toBe("pending");
+      }
+      expect(
+        probe.observe(
+          zeroFrames(5_000),
+          metrics(5_100, { intervalFreezeCount: 1 }),
+        ),
+      ).toBe("rejected");
+    }
+  });
+
+  it("keeps incomplete and non-overlapping candidates unknown", () => {
+    const incomplete = new SfuQualityProbe();
+    for (const timestampMs of [1_000, 3_000, 5_000]) {
+      expect(
+        incomplete.observe(
+          zeroFrames(timestampMs),
+          metrics(timestampMs + 100, { bitrateKbps: null }),
+        ),
+      ).toBe("pending");
+    }
+
+    const nonOverlapping = new P2pQualityProbe();
+    for (const timestampMs of [1_000, 4_000, 7_000]) {
+      expect(
+        nonOverlapping.observe(
+          zeroFrames(timestampMs),
+          metrics(timestampMs + 2_100),
+        ),
+      ).toBe("pending");
+    }
+  });
+
+  it("clears accumulated evidence when the current window is missing", () => {
+    const probe = new P2pQualityProbe();
+    expect(probe.observe(degraded(1_000), metrics(1_100))).toBe("pending");
+    expect(probe.observe(degraded(3_000), metrics(3_100))).toBe("pending");
+    expect(probe.observe(null, metrics(5_100))).toBe("pending");
+    expect(probe.observe(degraded(7_000), metrics(7_100))).toBe("pending");
+    expect(probe.observe(degraded(9_000), metrics(9_100))).toBe("pending");
+    expect(probe.observe(degraded(11_000), metrics(11_100))).toBe("approved");
+  });
+
+  it("ignores a duplicate timestamp but clears a late candidate window", () => {
+    const probe = new P2pQualityProbe();
+    expect(probe.observe(degraded(1_000), metrics(1_100))).toBe("pending");
+    expect(probe.observe(degraded(3_000), metrics(3_100))).toBe("pending");
+    expect(probe.observe(degraded(5_000), metrics(3_100))).toBe("pending");
+    expect(probe.observe(degraded(7_000), metrics(7_100))).toBe("approved");
+
+    probe.reset();
+    expect(probe.observe(degraded(9_000), metrics(9_100))).toBe("pending");
+    expect(probe.observe(degraded(11_000), metrics(11_100))).toBe("pending");
+    expect(probe.observe(degraded(13_000), metrics(10_100))).toBe("pending");
+    expect(probe.observe(degraded(15_000), metrics(15_100))).toBe("pending");
+    expect(probe.observe(degraded(17_000), metrics(17_100))).toBe("pending");
+    expect(probe.observe(degraded(19_000), metrics(19_100))).toBe("approved");
+  });
+});

tests/client-state.test.ts

@@ -1208,9 +1208,9 @@ describe("client signaling recovery policy", () => {
     });
     advance(2_000);
 
-    expect(sockets).toHaveLength(2);
-    expect(sockets[0]!.close).toHaveBeenCalledWith(4002, "signaling timeout");
-    expect(statuses.at(-1)).toBe("reconnecting");
+    expect(sockets).toHaveLength(1);
+    expect(sockets[0]!.close).not.toHaveBeenCalled();
+    expect(statuses.at(-1)).toBe("connected");
     expect(onTerminated).not.toHaveBeenCalled();
     expect(
       sockets[0]!.send.mock.calls
@@ -1218,27 +1218,18 @@ describe("client signaling recovery policy", () => {
         .includes("route-failed"),
     ).toBe(false);
 
+    advance(3_000);
+    const next = JSON.parse(String(sockets[0]!.send.mock.calls.at(-1)![0]));
+    expect(next).toEqual({ type: "signaling-challenge", sequence: 3 });
     receive(sockets[0]!, {
       type: "signaling-challenge-response",
-      sequence: confirm.sequence,
-    });
-    const oldClose = new Event("close");
-    Object.defineProperties(oldClose, {
-      code: { value: 4002 },
-      reason: { value: "signaling timeout" },
-    });
-    sockets[0]!.dispatchEvent(oldClose);
-    authenticate(sockets[1]!);
-    advance(5_000);
-    const healthy = JSON.parse(String(sockets[1]!.send.mock.calls.at(-1)![0]));
-    receive(sockets[1]!, {
-      type: "signaling-challenge-response",
-      sequence: healthy.sequence,
+      sequence: next.sequence,
     });
     advance(2_000);
 
-    expect(sockets).toHaveLength(2);
-    expect(onMessage).toHaveBeenCalledTimes(2);
+    expect(sockets).toHaveLength(1);
+    expect(confirm.sequence).toBe(2);
+    expect(onMessage).toHaveBeenCalledTimes(1);
     signal.stop();
   });
 

tests/host-peer-tracks.test.ts

@@ -958,6 +958,38 @@ describe("HostPeer source replacement", () => {
     expect(videoSender.appliedMaxBitrates).toEqual([3_000_000, 8_000_000]);
   });
 
+  it("replays the desired profile only after a connected peer reconnects", async () => {
+    const peer = createPeer(
+      createStream(
+        createTrack("video", "video"),
+        createTrack("audio", "audio"),
+      ),
+    );
+
+    await expect(peer.start()).resolves.toBe(true);
+    await acceptPeerAnswer(peer);
+    const connection = FakePeerConnection.latest!;
+    const videoSender = connection.senders[0]!;
+    const audioSender = connection.senders[1]!;
+    const videoCalls = videoSender.setParameters.mock.calls.length;
+    const audioCalls = audioSender.setParameters.mock.calls.length;
+
+    connection.dispatchEvent(new Event("connectionstatechange"));
+    await Promise.resolve();
+    expect(videoSender.setParameters).toHaveBeenCalledTimes(videoCalls);
+    expect(audioSender.setParameters).toHaveBeenCalledTimes(audioCalls);
+
+    connection.connectionState = "disconnected";
+    connection.dispatchEvent(new Event("connectionstatechange"));
+    connection.connectionState = "connected";
+    connection.dispatchEvent(new Event("connectionstatechange"));
+
+    await vi.waitFor(() =>
+      expect(videoSender.setParameters).toHaveBeenCalledTimes(videoCalls + 1),
+    );
+    expect(audioSender.setParameters).toHaveBeenCalledTimes(audioCalls + 1);
+  });
+
   it("continues queued profile updates after initial configuration rejects", async () => {
     const updates: PeerSnapshot[] = [];
     const peer = createPeer(

tests/hybrid-media-router.test.ts

@@ -33,19 +33,21 @@ function connectHost(
   store: RoomStore,
   room: CreatedRoom,
   routePolicy?: RoutePolicy,
+  clientId = "host_client_12345678",
+  sessionId = "host_session_12345678",
 ) {
   const connected = store.connectParticipant({
     roomId: room.roomId,
     role: "host",
     token: room.hostToken,
-    clientId: "host_client_12345678",
-    sessionId: "host_session_12345678",
+    clientId,
+    sessionId,
   });
   return {
     roomId: room.roomId,
     role: "host" as const,
     peerId: connected.peerId,
-    sessionId: "host_session_12345678",
+    sessionId,
     ...(routePolicy ? { routePolicy } : {}),
   };
 }
@@ -120,12 +122,24 @@ function deferred(): {
   return { promise, resolve, reject };
 }
 
+function deferredValue<T>(): {
+  promise: Promise<T>;
+  resolve: (value: T) => void;
+} {
+  let resolve!: (value: T) => void;
+  const promise = new Promise<T>((settle) => {
+    resolve = settle;
+  });
+  return { promise, resolve };
+}
+
 function harness(
   endpointMediaCopyCapacity: 1 | 2 | 3,
   withSfu = false,
   prepareTimeoutMs?: number,
   sfuIngressCapacity = 2,
   drainRetryMs?: number,
+  hostOfflineCheckMs?: number,
 ) {
   const store = createStore();
   const sent = new Map<string, ServerMessage[]>();
@@ -150,6 +164,7 @@ function harness(
             roomControl,
             ...(prepareTimeoutMs ? { prepareTimeoutMs } : {}),
             ...(drainRetryMs ? { drainRetryMs } : {}),
+            ...(hostOfflineCheckMs ? { hostOfflineCheckMs } : {}),
             tokenIssuer: {
               async issueToken({ peerId }) {
                 if (nextTokenIssueError) {
@@ -183,6 +198,7 @@ function harness(
   return {
     store,
     sent,
+    connections,
     admission,
     roomControl,
     onRoutesChanged,
@@ -513,6 +529,70 @@ describe("HybridMediaRouter v9 runtime", () => {
     }
   });
 
+  it("rebinds a replacement Host and clears offline Viewer media ownership", async () => {
+    const { store, sent, connections, router } = harness(2);
+    try {
+      const room = await store.createRoom();
+      const previousHost = connectHost(store, room);
+      complete(router, previousHost);
+      const viewer = connectViewer(store, room, "host-rebind-offline-viewer");
+      complete(router, viewer);
+      const prepared = await vi.waitFor(() => {
+        const message = preparedFor(sent, viewer.sessionId);
+        expect(message).toBeDefined();
+        return message!;
+      });
+      router.handleRouteReady(viewer, {
+        type: "route-ready",
+        revision: prepared.revision,
+        phase: "prepare",
+      });
+      router.setViewerRelayCapacity(viewer, 1);
+      await vi.waitFor(() =>
+        expect(connections.has(`${room.roomId}:${viewer.peerId}`)).toBe(true),
+      );
+
+      store.disconnectParticipant(room.roomId, viewer.peerId, viewer.sessionId);
+      router.disconnectParticipant(room.roomId, viewer.peerId, viewer.sessionId);
+      store.disconnectParticipant(
+        room.roomId,
+        previousHost.peerId,
+        previousHost.sessionId,
+      );
+      router.disconnectParticipant(
+        room.roomId,
+        previousHost.peerId,
+        previousHost.sessionId,
+      );
+
+      const resumedHost = connectHost(
+        store,
+        room,
+        undefined,
+        "replacement-host-client",
+        "replacement-host-session",
+      );
+      expect(resumedHost.peerId).not.toBe(previousHost.peerId);
+      complete(router, resumedHost);
+      expect(connections.has(`${room.roomId}:${viewer.peerId}`)).toBe(false);
+
+      const reconnectedViewer = connectViewer(
+        store,
+        room,
+        "host-rebind-offline-viewer",
+      );
+      expect(reconnectedViewer.peerId).toBe(viewer.peerId);
+      router.connectParticipant(reconnectedViewer);
+      expect(
+        router
+          .routeDiagnosticSnapshot(room.roomId)
+          .children.find((child) => child.effectiveCapacity === 1),
+      ).toBeDefined();
+    } finally {
+      await router.close();
+    }
+  });
+
   it("accepts active failure for a direct connection adopted during rebuild", async () => {
     const { store, sent, router } = harness(2);
     try {
@@ -1704,6 +1784,206 @@ describe("HybridMediaRouter v9 runtime", () => {
     await router.close();
   });
 
+  it("discards an SFU preparation whose cursor becomes stale while awaiting LiveKit", async () => {
+    const { store, sent, admission, roomControl, router } = harness(1, true);
+    try {
+      const room = await store.createRoom();
+      const host = connectHost(store, room);
+      complete(router, host);
+      const root = connectViewer(store, room, "prepare-stale-root");
+      complete(router, root);
+      const direct = await vi.waitFor(() => {
+        const prepared = preparedFor(sent, root.sessionId);
+        expect(prepared?.candidate.transport).toBe("direct");
+        return prepared!;
+      });
+      router.handleRouteReady(root, {
+        type: "route-ready",
+        revision: direct.revision,
+        phase: "prepare",
+      });
+      const gate = deferred();
+      roomControl!.createBarrier = gate.promise;
+      router.setViewerRelayCapacity(root, 0);
+      const waiting = connectViewer(store, room, "prepare-stale-waiting");
+      complete(router, waiting);
+      await vi.waitFor(() =>
+        expect(admission?.usage().ingress).toBe(1),
+      );
+      store.disconnectParticipant(room.roomId, root.peerId, root.sessionId);
+      router.disconnectParticipant(room.roomId, root.peerId, root.sessionId);
+      gate.resolve();
+      await vi.waitFor(() =>
+        expect(admission?.usage()).toEqual({ ingress: 0, egress: 0 }),
+      );
+      expect(router.routeDiagnosticSnapshot(room.roomId).operation).toBeNull();
+    } finally {
+      await router.close();
+    }
+  });
+
+  it("reschedules the regular Host-offline check after a read error", async () => {
+    const { store, sent, admission, roomControl, router } = harness(
+      1,
+      true,
+      undefined,
+      2,
+      undefined,
+      10,
+    );
+    try {
+      const room = await store.createRoom();
+      const { host } = await establishSfuRoom(store, sent, router, room);
+      const retry = deferredValue<boolean>();
+      const hostCheck = vi
+        .spyOn(roomControl!, "hostParticipantExists")
+        .mockRejectedValueOnce(new Error("temporary LiveKit read failure"))
+        .mockReturnValueOnce(retry.promise);
+      store.disconnectParticipant(room.roomId, host.peerId, host.sessionId);
+      router.disconnectParticipant(room.roomId, host.peerId, host.sessionId);
+
+      await vi.waitFor(() => expect(hostCheck).toHaveBeenCalledTimes(2));
+      expect(admission?.usage()).toEqual({ ingress: 1, egress: 2 });
+    } finally {
+      await router.close();
+    }
+  });
+
+  it("rechecks Host-offline retirement after a live route operation", async () => {
+    const { store, sent, admission, roomControl, router } = harness(
+      1,
+      true,
+      undefined,
+      2,
+      undefined,
+      10,
+    );
+    try {
+      const room = await store.createRoom();
+      const { host, first, second } = await establishSfuRoom(
+        store,
+        sent,
+        router,
+        room,
+      );
+      router.setViewerRelayCapacity(second, 1);
+      const active = router.resolveActiveViewerMediaEdge(
+        room.roomId,
+        first.peerId,
+      )!;
+      router.handleRouteFailed(first, {
+        type: "route-failed",
+        revision: active.revision,
+        phase: "active",
+        connectionId: active.connectionId,
+      });
+      await vi.waitFor(() => {
+        const prepared = preparedFor(sent, first.sessionId);
+        expect(prepared?.revision).toBeGreaterThan(active.revision);
+        expect(prepared?.candidate.transport).toBe("direct");
+        return prepared!;
+      });
+      const retry = deferredValue<boolean>();
+      const hostCheck = vi
+        .spyOn(roomControl!, "hostParticipantExists")
+        .mockResolvedValueOnce(false)
+        .mockReturnValueOnce(retry.promise);
+      store.disconnectParticipant(room.roomId, host.peerId, host.sessionId);
+      router.disconnectParticipant(room.roomId, host.peerId, host.sessionId);
+
+      await vi.waitFor(() => expect(hostCheck).toHaveBeenCalledTimes(2));
+      expect(
+        router.resolveActiveViewerMediaEdge(room.roomId, first.peerId),
+      ).toBeUndefined();
+      retry.resolve(false);
+      await vi.waitFor(() =>
+        expect(admission?.usage()).toEqual({ ingress: 0, egress: 0 }),
+      );
+      expect(hostCheck).toHaveBeenCalledTimes(2);
+    } finally {
+      await router.close();
+    }
+  });
+
+  it("ignores an old Host-offline result after a new SFU generation commits", async () => {
+    const { store, sent, admission, roomControl, router } = harness(
+      1,
+      true,
+      undefined,
+      2,
+      undefined,
+      10,
+    );
+    try {
+      const room = await store.createRoom();
+      const { host, first, second } = await establishSfuRoom(
+        store,
+        sent,
+        router,
+        room,
+      );
+      const oldRevision = Math.max(
+        router.resolveActiveViewerMediaEdge(room.roomId, first.peerId)!.revision,
+        router.resolveActiveViewerMediaEdge(room.roomId, second.peerId)!.revision,
+      );
+      const delayed = deferredValue<boolean>();
+      const hostCheck = vi
+        .spyOn(roomControl!, "hostParticipantExists")
+        .mockReturnValueOnce(delayed.promise);
+      store.disconnectParticipant(room.roomId, host.peerId, host.sessionId);
+      router.disconnectParticipant(room.roomId, host.peerId, host.sessionId);
+      await vi.waitFor(() => expect(hostCheck).toHaveBeenCalledTimes(1));
+
+      const replacementHost = connectHost(
+        store,
+        room,
+        undefined,
+        "new-physical-host-client",
+        "new-physical-host-session",
+      );
+      complete(router, replacementHost);
+      const handled = new Set<string>();
+      for (let index = 0; index < 3; index += 1) {
+        const next = await vi.waitFor(() => {
+          const prepared = [first, second]
+            .map((viewer) => ({
+              viewer,
+              prepared: preparedFor(sent, viewer.sessionId),
+            }))
+            .find(
+              ({ prepared }) =>
+                prepared &&
+                prepared.revision > oldRevision &&
+                !handled.has(prepared.candidate.connectionId),
+            );
+          expect(prepared).toBeDefined();
+          return prepared!;
+        });
+        handled.add(next.prepared!.candidate.connectionId);
+        router.handleRouteReady(next.viewer, {
+          type: "route-ready",
+          revision: next.prepared!.revision,
+          phase: "prepare",
+        });
+      }
+      await vi.waitFor(() =>
+        expect(admission?.usage()).toEqual({ ingress: 1, egress: 2 }),
+      );
+      const currentFence = roomControl!.created.at(-1)!;
+      expect(currentFence.publicationGeneration).not.toBe(
+        roomControl!.created[0]!.publicationGeneration,
+      );
+
+      delayed.resolve(false);
+      await Promise.resolve();
+      await Promise.resolve();
+      expect(admission?.usage()).toEqual({ ingress: 1, egress: 2 });
+      expect(roomControl!.deleted).not.toContainEqual(currentFence);
+    } finally {
+      await router.close();
+    }
+  });
+
   it("resolves exact direct and peer-relayed Viewer evidence sources", async () => {
     const { store, sent, router } = harness(1);
     try {

tests/media-assignment.test.ts

@@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest";
 
 import type { SignalPayload } from "../src/shared/protocol.ts";
 import {
+  createOwnedViewerRestartSender,
   limitMediaAssignment,
   MAX_ENDPOINT_MEDIA_CHILDREN,
   reconcileBoundedMediaChildren,
@@ -116,4 +117,41 @@ describe("peer-assisted client assignment", () => {
       rebuild: false,
     });
   });
+
+  it("allows a pending peer restart closure only after exact promotion", () => {
+    let active = false;
+    const messages: unknown[] = [];
+    const sendRestart = createOwnedViewerRestartSender(
+      true,
+      (targetPeerId, connectionId) =>
+        active &&
+        targetPeerId === "parent_12345678" &&
+        connectionId === "connection_12345678",
+      (message) => {
+        messages.push(message);
+        return true;
+      },
+    );
+
+    expect(
+      sendRestart("parent_12345678", "connection_12345678", false),
+    ).toBe(false);
+    expect(messages).toEqual([]);
+
+    active = true;
+    expect(
+      sendRestart("parent_12345678", "connection_12345678", true),
+    ).toBe(true);
+    expect(messages).toEqual([
+      {
+        type: "restart-request",
+        targetPeerId: "parent_12345678",
+        connectionId: "connection_12345678",
+        rebuild: true,
+      },
+    ]);
+    expect(sendRestart("other-parent", "connection_12345678", true)).toBe(
+      false,
+    );
+  });
 });

tests/room-route-controller.test.ts

@@ -388,6 +388,60 @@ describe("RoomRouteController", () => {
     expect(routes.diagnosticParticipantLabel(C)).not.toBe(label);
   });
 
+  it("clears an unknown sender run without rearming its healthy baseline", () => {
+    const routes = controller(2, { qualityConvergenceEnabled: true });
+    addViewer(routes, A, 2);
+    addViewer(routes, B, 2);
+    routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"));
+    routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"));
+    for (const childPeerId of [A, B]) {
+      expect(
+        observeSenderState(routes, {
+          childPeerId,
+          connectionId: `${childPeerId === A ? "a" : "b"}_from_host`,
+          state: "healthy",
+          acceptedAtMs: 0,
+        }).accepted,
+      ).toBe(true);
+    }
+    expect(
+      observeSenderState(routes, {
+        childPeerId: A,
+        connectionId: "a_from_host",
+        state: "degraded",
+        acceptedAtMs: 1,
+      }).accepted,
+    ).toBe(true);
+    expect(
+      routes.observeSenderQualityEvidence({
+        parentPeerId: HOST,
+        parentSessionId: "host_session",
+        childPeerId: A,
+        routeRevision: routes.snapshot().revision,
+        connectionId: "a_from_host",
+        senderIdentity: null,
+        sampleTimestampMs: null,
+        state: "unknown",
+        acceptedAtMs: 2,
+      }).accepted,
+    ).toBe(true);
+
+    for (const acceptedAtMs of [3, 4, 5]) {
+      expect(
+        observeSenderState(routes, {
+          childPeerId: A,
+          connectionId: "a_from_host",
+          state: "degraded",
+          acceptedAtMs,
+        }).accepted,
+      ).toBe(true);
+    }
+    expect(routes.reconcile(6).operation).toMatchObject({
+      childPeerId: A,
+      reason: "quality-convergence",
+    });
+  });
+
   it("keeps isolated limitation windows diagnostic", () => {
     const routes = controller(2, { qualityConvergenceEnabled: true });
     addViewer(routes, A, 2);
@@ -1307,28 +1361,119 @@ describe("RoomRouteController", () => {
       hostSessionId: "host_session",
     });
     const current = begin.operation!.current!;
+    const guard: CandidateGuard = {
+      childPeerId,
+      childSessionId: `${childPeerId}_session`,
+      revision: current.revision,
+      connectionId: "quality_sfu_candidate",
+    };
+    expect(
+      routes.candidateReady(guard, 112),
+    ).toMatchObject({ accepted: true, committed: false });
+    for (const acceptedAtMs of [113, 114]) {
+      expect(
+        routes.observeSfuPublisherQualityEvidence({
+          hostPeerId: HOST,
+          hostSessionId: "host_session",
+          publicationGeneration: "publication_generation_12345678",
+          routeRevision: current.revision,
+          state: "healthy",
+          sampleTimestampMs: acceptedAtMs,
+          acceptedAtMs,
+        }),
+      ).toMatchObject({ accepted: true, committed: false });
+      expect(routes.candidateReady(guard, acceptedAtMs)).toMatchObject({
+        accepted: true,
+        committed: false,
+      });
+    }
+    expect(
+      routes.observeSfuPublisherQualityEvidence({
+        hostPeerId: HOST,
+        hostSessionId: "host_session",
+        publicationGeneration: "publication_generation_12345678",
+        routeRevision: current.revision,
+        state: "healthy",
+        sampleTimestampMs: 115,
+        acceptedAtMs: 115,
+      }).committed,
+    ).toBe(true);
+    expect(routes.snapshot().upstreamByViewer.get(childPeerId)).toMatchObject({
+      kind: "sfu",
+      connectionId: "quality_sfu_candidate",
+    });
+  });
+
+  it("rejects an SFU quality candidate on degraded publisher evidence", () => {
+    const routes = controller(2, {
+      sfuEnabled: true,
+      qualityConvergenceEnabled: true,
+    });
+    addViewer(routes, A, 1);
+    addViewer(routes, B, 0);
+    routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"));
+    routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"));
+    observePersistentDegraded(routes, A, "a_from_host", 99);
+    expect(routes.reconcile(103).operation).toBeUndefined();
+    observePersistentDegraded(routes, B, "b_from_host", 104);
+    const operation = routes.reconcile(108).operation!;
+    const childPeerId = operation.childPeerId;
+    const peerAttempt = beginCandidate(routes, {
+      nowMs: 109,
+      connectionId: "quality_peer_candidate",
+      reservation: { kind: "direct" },
+    }).operation!;
+    expect(
+      routes.candidateFailed(
+        {
+          childPeerId,
+          childSessionId: `${childPeerId}_session`,
+          revision: peerAttempt.current!.revision,
+          connectionId: "quality_peer_candidate",
+        },
+        110,
+      ).accepted,
+    ).toBe(true);
+    const sfuAttempt = beginCandidate(routes, {
+      nowMs: 111,
+      connectionId: "quality_sfu_degraded",
+      reservation: {
+        kind: "sfu-create",
+        edge: "sfu-edge",
+        publication: "sfu-publication",
+        overlap: "host-overlap",
+      },
+      publicationGeneration: "publication_generation_degraded",
+      publicationConnectionId: "publication_connection_degraded",
+      hostSessionId: "host_session",
+    }).operation!;
+    const current = sfuAttempt.current!;
     expect(
       routes.candidateReady(
         {
           childPeerId,
           childSessionId: `${childPeerId}_session`,
           revision: current.revision,
-          connectionId: "quality_sfu_candidate",
+          connectionId: current.connectionId,
         },
         112,
       ),
     ).toMatchObject({ accepted: true, committed: false });
     expect(
-      observePersistentSfuHealthy(
-        routes,
-        "publication_generation_12345678",
-        current.revision,
-        113,
-      ).committed,
-    ).toBe(true);
+      routes.observeSfuPublisherQualityEvidence({
+        hostPeerId: HOST,
+        hostSessionId: "host_session",
+        publicationGeneration: "publication_generation_degraded",
+        routeRevision: current.revision,
+        state: "degraded",
+        sampleTimestampMs: 113,
+        acceptedAtMs: 113,
+      }),
+    ).toMatchObject({ accepted: true, committed: false });
+    expect(routes.snapshot().operation).toBeUndefined();
     expect(routes.snapshot().upstreamByViewer.get(childPeerId)).toMatchObject({
-      kind: "sfu",
-      connectionId: "quality_sfu_candidate",
+      kind: "peer",
+      parentPeerId: HOST,
     });
   });
 
@@ -3229,12 +3374,18 @@ describe("RoomRouteController", () => {
   });
 
 
-  it("replaces one publication and makes every old subscriber waiting", () => {
+  it("keeps only old SFU roots that still anchor peer descendants", () => {
     const routes = controller(1, { sfuEnabled: true });
     addViewer(routes, A, 0);
-    addViewer(routes, B, 0);
+    addViewer(routes, B, 1);
+    addViewer(routes, C, 0);
+    addViewer(routes, D, 0);
     routes.hydrateHostPublication("publication_1", "publication_resource_1");
-    for (const [peerId, resource] of [[A, "subscription_a"], [B, "subscription_b"]] as const) {
+    for (const [peerId, resource] of [
+      [A, "subscription_a"],
+      [B, "subscription_b"],
+      [D, "subscription_d"],
+    ] as const) {
       routes.hydrateEdge(peerId, {
         kind: "sfu",
         publicationGeneration: "publication_1",
@@ -3245,6 +3396,7 @@ describe("RoomRouteController", () => {
         resource,
       });
     }
+    routes.hydrateEdge(C, peerEdge(B, "c_from_b"));
     expect(routes.invalidateHostPublication({
       hostSessionId: "host_session",
       routeRevision: 0,
@@ -3276,17 +3428,27 @@ describe("RoomRouteController", () => {
       connectionId: "a_sfu_2",
     }, 22);
     expect(settled.accepted).toBe(true);
-    expect(settled.released).toHaveLength(4);
     expect(settled.released).toEqual(expect.arrayContaining([
       "publication_overlap",
       "subscription_a",
       "subscription_b",
+      "subscription_d",
       "publication_resource_1",
     ]));
     expect(routes.snapshot().upstreamByViewer.get(A)).toMatchObject({
       publicationGeneration: "publication_2",
     });
-    expect(routes.snapshot().upstreamByViewer.has(B)).toBe(false);
+    expect(routes.snapshot().upstreamByViewer.get(B)).toMatchObject({
+      kind: "sfu",
+      publicationGeneration: "publication_1",
+      usable: false,
+      physicalActive: false,
+    });
+    expect(routes.snapshot().upstreamByViewer.get(C)).toMatchObject({
+      kind: "peer",
+      parentPeerId: B,
+    });
+    expect(routes.snapshot().upstreamByViewer.has(D)).toBe(false);
     expect(routes.routeDiagnosticSnapshot(22).children[1]).toMatchObject({
       demandAgeMs: 12,
       finalRoute: "waiting",

tests/route-transition.test.ts

@@ -393,6 +393,144 @@ describe("minimal route transition contracts", () => {
     expect(publishers[1]?.disconnect).not.toHaveBeenCalled();
   });
 
+  it("reports and retires an exact active SFU source replacement failure", async () => {
+    const messages: ClientMessage[] = [];
+    const publishers: ReturnType<typeof createFakePublisher>[] = [];
+    const route = new HostSfuRoute({
+      getStream: () => ({}) as MediaStream,
+      getProfile: () => QUALITY_PROFILES["720p30"],
+      getVideoCodec: () => "vp8",
+      reconcileChildren: () => undefined,
+      send: (message) => {
+        messages.push(message);
+        return true;
+      },
+      createPublisher: () => {
+        const publisher = createFakePublisher([], "publisher-active");
+        publishers.push(publisher);
+        return publisher;
+      },
+    });
+
+    await route.acceptAndWait({
+      revision: 1,
+      phase: "active",
+      assignment: hostAssignment("publication-active"),
+    });
+    await route.acceptConfig(sfuConfig(1));
+    publishers[0]!.replaceStream.mockResolvedValue(false);
+
+    await expect(route.replaceStream({} as MediaStream)).resolves.toBe(false);
+    expect(publishers[0]!.disconnect).toHaveBeenCalledOnce();
+    expect(messages).toContainEqual({ type: "refresh-sfu", revision: 1 });
+  });
+
+  it("retires a failed pending source without invalidating healthy active SFU", async () => {
+    const messages: ClientMessage[] = [];
+    const publishers: ReturnType<typeof createFakePublisher>[] = [];
+    const route = new HostSfuRoute({
+      getStream: () => ({}) as MediaStream,
+      getProfile: () => QUALITY_PROFILES["720p30"],
+      getVideoCodec: () => "vp8",
+      reconcileChildren: () => undefined,
+      send: (message) => {
+        messages.push(message);
+        return true;
+      },
+      createPublisher: () => {
+        const publisher = createFakePublisher(
+          [],
+          `publisher-${publishers.length + 1}`,
+        );
+        publishers.push(publisher);
+        return publisher;
+      },
+    });
+
+    await route.acceptAndWait({
+      revision: 1,
+      phase: "active",
+      assignment: hostAssignment("publication-active"),
+    });
+    await route.acceptConfig(sfuConfig(1));
+    route.accept({
+      revision: 2,
+      phase: "prepare",
+      assignment: hostAssignment("publication-pending"),
+      candidate: candidate(2, "viewer_12345678", "sfu"),
+    });
+    await route.acceptConfig(sfuConfig(2));
+    publishers[1]!.replaceStream.mockResolvedValue(false);
+
+    await expect(route.replaceStream({} as MediaStream)).resolves.toBe(true);
+    expect(publishers[0]!.disconnect).not.toHaveBeenCalled();
+    expect(publishers[1]!.disconnect).toHaveBeenCalledOnce();
+    expect(messages).toContainEqual({
+      type: "route-failed",
+      revision: 2,
+      phase: "prepare",
+      connectionId: null,
+    });
+  });
+
+  it("reports success when a pending SFU promotion supersedes an old failure", async () => {
+    const publishers: ReturnType<typeof createFakePublisher>[] = [];
+    const route = new HostSfuRoute({
+      getStream: () => ({}) as MediaStream,
+      getProfile: () => QUALITY_PROFILES["720p30"],
+      getVideoCodec: () => "vp8",
+      reconcileChildren: () => undefined,
+      send: () => true,
+      createPublisher: () => {
+        const publisher = createFakePublisher(
+          [],
+          `publisher-${publishers.length + 1}`,
+        );
+        publishers.push(publisher);
+        return publisher;
+      },
+    });
+
+    await route.acceptAndWait({
+      revision: 1,
+      phase: "active",
+      assignment: hostAssignment("publication-old"),
+    });
+    await route.acceptConfig(sfuConfig(1));
+    route.accept({
+      revision: 2,
+      phase: "prepare",
+      assignment: hostAssignment("publication-new"),
+      candidate: candidate(2, "viewer_12345678", "sfu"),
+    });
+    await route.acceptConfig(sfuConfig(2));
+
+    let resolveOldReplacement!: (replaced: boolean) => void;
+    publishers[0]!.replaceStream.mockImplementation(
+      () =>
+        new Promise<boolean>((resolve) => {
+          resolveOldReplacement = resolve;
+        }),
+    );
+    const oldReplacementCalls = publishers[0]!.replaceStream.mock.calls.length;
+    const replacing = route.replaceStream({} as MediaStream);
+    await vi.waitFor(() =>
+      expect(publishers[0]!.replaceStream).toHaveBeenCalledTimes(
+        oldReplacementCalls + 1,
+      ),
+    );
+
+    await route.acceptAndWait({
+      revision: 2,
+      phase: "active",
+      assignment: hostAssignment("publication-new"),
+    });
+    resolveOldReplacement(false);
+
+    await expect(replacing).resolves.toBe(true);
+    expect(publishers[1]!.disconnect).not.toHaveBeenCalled();
+  });
+
 
 
   it("reconciles paused Host active truth by exact SFU publication", async () => {
@@ -510,6 +648,147 @@ describe("minimal route transition contracts", () => {
     expect(publishers[1]?.disconnect).toHaveBeenCalled();
   });
 
+  it("ignores a retired publisher failure after same-generation resync", async () => {
+    let releaseOldConnect!: () => void;
+    const oldConnectGate = new Promise<void>((resolve) => {
+      releaseOldConnect = resolve;
+    });
+    const messages: ClientMessage[] = [];
+    const publishers: ReturnType<typeof createFakePublisher>[] = [];
+    const route = new HostSfuRoute({
+      getStream: () => ({}) as MediaStream,
+      getProfile: () => QUALITY_PROFILES["720p30"],
+      getVideoCodec: () => "vp8",
+      reconcileChildren: () => undefined,
+      send: (message) => {
+        messages.push(message);
+        return true;
+      },
+      createPublisher: () => {
+        const publisher = createFakePublisher(
+          [],
+          `publisher-${publishers.length + 1}`,
+        );
+        if (publishers.length === 1) {
+          publisher.connect.mockImplementation(async () => {
+            await oldConnectGate;
+            throw new Error("retired publisher");
+          });
+        }
+        publishers.push(publisher);
+        return publisher;
+      },
+    });
+
+    await route.acceptAndWait({
+      revision: 1,
+      phase: "active",
+      assignment: hostAssignment("publication-old"),
+    });
+    await route.acceptConfig(sfuConfig(1));
+    route.accept({
+      revision: 2,
+      phase: "prepare",
+      assignment: hostAssignment("publication-new"),
+      candidate: candidate(2, "viewer_12345678", "sfu"),
+    });
+    const stale = route.acceptConfig(sfuConfig(2));
+    await vi.waitFor(() => expect(publishers[1]?.connect).toHaveBeenCalledOnce());
+
+    await route.resyncAuthoritative({
+      revision: 1,
+      phase: "active",
+      assignment: hostAssignment("publication-old"),
+    });
+    route.accept({
+      revision: 2,
+      phase: "prepare",
+      assignment: hostAssignment("publication-new"),
+      candidate: candidate(2, "viewer_12345678", "sfu"),
+    });
+    const current = route.acceptConfig(sfuConfig(2));
+    await vi.waitFor(() => expect(publishers).toHaveLength(3));
+    releaseOldConnect();
+    await Promise.all([stale, current]);
+
+    expect(messages.filter((message) => message.type === "route-failed")).toEqual(
+      [],
+    );
+    expect(publishers[2]?.activate).toHaveBeenCalledOnce();
+  });
+
+  it("ignores a retired subscriber failure after same-generation resync", async () => {
+    let releaseOldConnect!: () => void;
+    const oldConnectGate = new Promise<void>((resolve) => {
+      releaseOldConnect = resolve;
+    });
+    const messages: ClientMessage[] = [];
+    const subscribers: ReturnType<typeof createFakeSubscriber>[] = [];
+    const route = new ViewerSfuRoute("viewer_12345678", {
+      activatePeer: () => true,
+      reconcileSfuChildren: () => undefined,
+      onSfuStream: () => undefined,
+      send: (message) => {
+        messages.push(message);
+        return true;
+      },
+      createSubscriber: (events) => {
+        const subscriber = createFakeSubscriber(
+          events,
+          [],
+          `subscriber-${subscribers.length + 1}`,
+        );
+        if (subscribers.length === 1) {
+          subscriber.connect.mockImplementation(async () => {
+            await oldConnectGate;
+            throw new Error("retired subscriber");
+          });
+        }
+        subscribers.push(subscriber);
+        return subscriber;
+      },
+    });
+
+    route.accept({
+      revision: 1,
+      phase: "active",
+      assignment: viewerSfuAssignment([], "publication-old"),
+    });
+    await route.acceptConfig(sfuConfig(1));
+    route.accept({
+      revision: 2,
+      phase: "prepare",
+      assignment: viewerSfuAssignment([], "publication-new"),
+      candidate: candidate(2, "viewer_12345678", "sfu"),
+    });
+    const stale = route.acceptConfig(sfuConfig(2));
+    await vi.waitFor(() => expect(subscribers[1]?.connect).toHaveBeenCalledOnce());
+
+    await route.resyncAuthoritative(
+      {
+        revision: 1,
+        phase: "active",
+        assignment: viewerSfuAssignment([], "publication-old"),
+      },
+      "viewer_12345678",
+    );
+    route.accept({
+      revision: 2,
+      phase: "prepare",
+      assignment: viewerSfuAssignment([], "publication-new"),
+      candidate: candidate(2, "viewer_12345678", "sfu"),
+    });
+    const current = route.acceptConfig(sfuConfig(2));
+    await vi.waitFor(() => expect(subscribers).toHaveLength(3));
+    releaseOldConnect();
+    await Promise.all([stale, current]);
+
+    expect(messages.filter((message) => message.type === "route-failed")).toEqual(
+      [],
+    );
+    expect(subscribers[2]?.activate).toHaveBeenCalledOnce();
+  });
+
 
   it("uses the subscriber first decoded frame as the only SFU prepare ready", async () => {
     const log: string[] = [];

tests/sfu-client.test.ts

@@ -720,6 +720,7 @@ describe("SfuPublisher", () => {
 
     const room = livekit.state.rooms[0];
     expect(room.options).toEqual({
+      disconnectOnPageLeave: false,
       dynacast: true,
       stopLocalTrackOnUnpublish: false,
     });
@@ -1656,6 +1657,73 @@ describe("SfuSubscriber", () => {
     expect(proofs).toEqual([8]);
   });
 
+  it("retains pending frame proof through LiveKit full reconnect order", async () => {
+    vi.useFakeTimers();
+    const proofs = vi.fn(() => true);
+    const subscriber = new SfuSubscriber({
+      onStream: vi.fn(),
+      onFirstDecodedFrame: proofs,
+    });
+    await subscriber.connect(connection);
+    const room = livekit.state.rooms[0];
+    const oldHost = new livekit.FakeRemoteParticipant("host");
+    const oldPublication = new livekit.FakeRemotePublication(
+      "host-video-before-reconnect",
+      Track.Source.ScreenShare,
+    );
+    oldHost.add(oldPublication);
+    room.remoteParticipants.set("host", oldHost);
+    expect(subscriber.activate()).toBe(true);
+    subscriber.armDecodedFrameProof();
+
+    let releaseOldRead!: () => void;
+    const oldRead = new Promise<void>((resolve) => {
+      releaseOldRead = resolve;
+    });
+    room.emit(
+      RoomEvent.TrackSubscribed,
+      remoteTrack(track("video", "video-before-reconnect"), async () => {
+        await oldRead;
+        return statsReport([]);
+      }),
+      oldPublication,
+      oldHost,
+    );
+
+    room.remoteParticipants.delete("host");
+    room.emit(RoomEvent.ParticipantDisconnected, oldHost);
+    room.emit(RoomEvent.Reconnecting);
+
+    const newHost = new livekit.FakeRemoteParticipant("host");
+    const newPublication = new livekit.FakeRemotePublication(
+      "host-video-after-reconnect",
+      Track.Source.ScreenShare,
+    );
+    newHost.add(newPublication);
+    room.remoteParticipants.set("host", newHost);
+    room.emit(RoomEvent.Reconnected);
+    room.emit(
+      RoomEvent.TrackSubscribed,
+      remoteTrack(track("video", "video-after-reconnect"), async () =>
+        statsReport([
+          {
+            id: "video-after-reconnect",
+            type: "inbound-rtp",
+            timestamp: 1,
+            kind: "video",
+            framesDecoded: 1,
+          },
+        ]),
+      ),
+      newPublication,
+      newHost,
+    );
+
+    await vi.waitFor(() => expect(proofs).toHaveBeenCalledOnce());
+    releaseOldRead();
+    expect(subscriber.deactivate()).toBe(true);
+  });
+
   it("reconciles a Host publication announced while connect is pending", async () => {
     const gate = deferred();
     livekit.state.connectGate = gate.promise;