Merge pull request #2733 from vitorpamplona/claude/fix-nests-audio-dropout-EGlUZ

Eliminate audio gaps during JWT refresh and improve network resilience

amethyst/src/main/AndroidManifest.xml

@@ -37,6 +37,11 @@
     <!-- To know receive notifications when the app connects/disconnects from the web -->
     <uses-permission android:name="android.permission.CHANGE_NETWORK_STATE" />
 
+    <!-- For ConnectivityManager.registerDefaultNetworkCallback used by NestForegroundService
+         to detect Wi-Fi ↔ cellular handover and recycle the active QUIC session promptly
+         (otherwise the wrapper waits ~30s for QUIC PTO before reconnecting). -->
+    <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
+
     <!-- Audio/Video Playback -->
     <uses-permission android:name="android.permission.FOREGROUND_SERVICE" />
     <uses-permission android:name="android.permission.FOREGROUND_SERVICE_MEDIA_PLAYBACK" />

amethyst/src/main/java/com/vitorpamplona/amethyst/service/nests/NestForegroundService.kt

@@ -28,13 +28,23 @@ import android.app.Service
 import android.content.Context
 import android.content.Intent
 import android.content.pm.ServiceInfo
+import android.media.AudioDeviceCallback
+import android.media.AudioDeviceInfo
+import android.media.AudioFocusRequest
+import android.media.AudioManager
+import android.net.ConnectivityManager
+import android.net.Network
 import android.os.Build
 import android.os.IBinder
 import android.os.PowerManager
 import androidx.core.app.NotificationCompat
 import androidx.core.content.ContextCompat
 import com.vitorpamplona.amethyst.R
+import com.vitorpamplona.amethyst.commons.viewmodels.NestAudioFocusBus
+import com.vitorpamplona.amethyst.commons.viewmodels.NestAudioFocusState
+import com.vitorpamplona.amethyst.commons.viewmodels.NestNetworkChangeBus
 import com.vitorpamplona.amethyst.ui.MainActivity
+import com.vitorpamplona.quartz.utils.Log
 
 /**
  * Process-anchor for an active audio-room session. Holds a partial wake-lock
@@ -57,7 +67,42 @@ import com.vitorpamplona.amethyst.ui.MainActivity
 class NestForegroundService : Service() {
     private var wakeLock: PowerManager.WakeLock? = null
     private var promoted = false
-    private var audioFocusRequest: android.media.AudioFocusRequest? = null
+    private var audioFocusRequest: AudioFocusRequest? = null
+
+    /**
+     * Logs route changes (Bluetooth headset attach/detach, wired
+     * headset, USB audio, speakerphone) so field reports of
+     * "audio cut out when I plugged in headphones" can be correlated
+     * with a concrete device-add / device-remove event.
+     *
+     * Doesn't drive playback decisions — Android's AudioTrack +
+     * AudioRecord auto-route to whichever device the OS treats as
+     * active, so the brief silence on a route swap is unavoidable
+     * without going through the (intrusive) `MODE_IN_COMMUNICATION` +
+     * `setCommunicationDevice` flow that this service deliberately
+     * avoids. v1 ships observability only; future work could pause
+     * playback briefly across a route change to mask the swap.
+     */
+    private val deviceCallback =
+        object : AudioDeviceCallback() {
+            override fun onAudioDevicesAdded(addedDevices: Array<out AudioDeviceInfo>?) {
+                addedDevices?.forEach { dev ->
+                    Log.i("NestAudio") {
+                        "audio device added: type=${dev.type} name='${dev.productName}' " +
+                            "isSink=${dev.isSink} isSource=${dev.isSource}"
+                    }
+                }
+            }
+
+            override fun onAudioDevicesRemoved(removedDevices: Array<out AudioDeviceInfo>?) {
+                removedDevices?.forEach { dev ->
+                    Log.i("NestAudio") {
+                        "audio device removed: type=${dev.type} name='${dev.productName}' " +
+                            "isSink=${dev.isSink} isSource=${dev.isSource}"
+                    }
+                }
+            }
+        }
 
     override fun onCreate() {
         super.onCreate()
@@ -67,13 +112,122 @@ class NestForegroundService : Service() {
                 .newWakeLock(PowerManager.PARTIAL_WAKE_LOCK, "amethyst:audio-room")
                 .apply { setReferenceCounted(false) }
         requestAudioFocus()
+        registerAudioDeviceCallback()
+        registerNetworkCallback()
+    }
+
+    private fun registerAudioDeviceCallback() {
+        runCatching {
+            val mgr = getSystemService(Context.AUDIO_SERVICE) as AudioManager
+            // null Handler → callback runs on the main looper, which is
+            // the cheapest path for "log a line" handlers like this.
+            mgr.registerAudioDeviceCallback(deviceCallback, null)
+        }
+    }
+
+    private fun unregisterAudioDeviceCallback() {
+        runCatching {
+            val mgr = getSystemService(Context.AUDIO_SERVICE) as AudioManager
+            mgr.unregisterAudioDeviceCallback(deviceCallback)
+        }
     }
 
     /**
-     * Request transient-may-duck audio focus so an inbound phone
-     * call lowers the room volume cleanly instead of mixing two
-     * voices on top of each other. Acquired once per service
-     * lifetime; released in [onDestroy].
+     * Track the current default network so the NetworkCallback can
+     * distinguish "first onAvailable after register" (no-op) from "the
+     * default network just changed under us" (publish). Volatile because
+     * the callback fires on a binder thread; the publish path is
+     * lock-free via [NestNetworkChangeBus].
+     *
+     * `seenInitialNetwork` is the registration-time suppression flag —
+     * set true on the first onAvailable and never cleared. The earlier
+     * shape used `previous != null` as the suppression check, which
+     * also suppressed the WiFi-loss-then-cellular-available case
+     * (`onLost` clears [currentDefaultNetwork] back to null, then the
+     * follow-up `onAvailable` looks identical to the registration
+     * callback). That broke the most important scenario this whole
+     * code path exists for — a clean Wi-Fi ↔ cellular handover.
+     */
+    @Volatile private var currentDefaultNetwork: Network? = null
+
+    @Volatile private var seenInitialNetwork: Boolean = false
+
+    /**
+     * Listens for the device's default-network changing (Wi-Fi ↔
+     * cellular handover, plane mode toggle, hotspot swap) and signals
+     * every active [com.vitorpamplona.amethyst.commons.viewmodels.NestViewModel]
+     * to recycle its QUIC session. Without this nudge the QUIC
+     * connection sitting on the now-dead socket would have to wait
+     * for its PTO (~30 s) before the wrapper notices — a long
+     * audible silence on every handover.
+     *
+     * The callback also fires once right after registration with the
+     * current default network — we suppress that emission so the VM
+     * doesn't recycle on every service start.
+     */
+    private val networkCallback =
+        object : ConnectivityManager.NetworkCallback() {
+            override fun onAvailable(network: Network) {
+                val previous = currentDefaultNetwork
+                currentDefaultNetwork = network
+                if (!seenInitialNetwork) {
+                    // Registration-time callback — fires once with the
+                    // currently-active default network. Suppress so the
+                    // VM doesn't recycle on every service start.
+                    seenInitialNetwork = true
+                    return
+                }
+                if (previous != network) {
+                    Log.i("NestNet") {
+                        "default network changed ($previous → $network), recycling QUIC sessions"
+                    }
+                    NestNetworkChangeBus.publish()
+                }
+            }
+
+            override fun onLost(network: Network) {
+                if (currentDefaultNetwork == network) {
+                    // We've lost the current default. Don't publish here —
+                    // the next onAvailable (with the replacement network)
+                    // will, and recycling NOW means the wrapper would try
+                    // to handshake on no network at all and fail-then-
+                    // backoff. Just clear so the next onAvailable is
+                    // recognised as a change.
+                    currentDefaultNetwork = null
+                }
+            }
+        }
+
+    private fun registerNetworkCallback() {
+        runCatching {
+            val mgr = getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
+            mgr.registerDefaultNetworkCallback(networkCallback)
+        }
+    }
+
+    private fun unregisterNetworkCallback() {
+        runCatching {
+            val mgr = getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
+            mgr.unregisterNetworkCallback(networkCallback)
+        }
+        currentDefaultNetwork = null
+        seenInitialNetwork = false
+    }
+
+    /**
+     * Request audio focus for the duration of the audio-room session
+     * and route the system's focus-change events into [NestAudioFocusBus]
+     * so every active [com.vitorpamplona.amethyst.commons.viewmodels.NestViewModel]
+     * can react.
+     *
+     * Why we actually handle the focus change (vs the previous no-op
+     * listener): the platform only auto-ducks streams that opted into
+     * auto-ducking (CONTENT_TYPE_MUSIC, etc.) — a `CONTENT_TYPE_SPEECH`
+     * stream is left alone, so without a real listener an inbound phone
+     * call would mix on top of the room audio and a Maps voice prompt
+     * would be inaudible against an active speaker. The bus carries
+     * the translated state to the VM, which silences the listener
+     * playback and the broadcast mic for the duration of the loss.
      *
      * Matches the playback `AudioAttributes` we set on `AudioTrack`
      * in `AudioTrackPlayer` (USAGE_MEDIA + CONTENT_TYPE_SPEECH) so
@@ -82,7 +236,7 @@ class NestForegroundService : Service() {
      */
     private fun requestAudioFocus() {
         if (audioFocusRequest != null) return
-        val mgr = getSystemService(Context.AUDIO_SERVICE) as android.media.AudioManager
+        val mgr = getSystemService(Context.AUDIO_SERVICE) as AudioManager
         val attrs =
             android.media.AudioAttributes
                 .Builder()
@@ -90,26 +244,62 @@ class NestForegroundService : Service() {
                 .setContentType(android.media.AudioAttributes.CONTENT_TYPE_SPEECH)
                 .build()
         val request =
-            android.media.AudioFocusRequest
-                .Builder(android.media.AudioManager.AUDIOFOCUS_GAIN)
+            AudioFocusRequest
+                .Builder(AudioManager.AUDIOFOCUS_GAIN)
                 .setAudioAttributes(attrs)
                 .setAcceptsDelayedFocusGain(false)
-                .setOnAudioFocusChangeListener({ /* no-op — duck-on-loss handled by the OS */ })
-                .build()
-        // Best-effort: a refused request just means the OS will
-        // duck/pause us based on its own policy. Don't fail the
-        // service start over a focus denial.
-        runCatching { mgr.requestAudioFocus(request) }
+                .setOnAudioFocusChangeListener { focusChange ->
+                    NestAudioFocusBus.publish(translateFocusChange(focusChange))
+                }.build()
+        // Strict GRANTED check: anything else — FAILED (active call
+        // blocking us), DELAYED (we set [setAcceptsDelayedFocusGain]
+        // false so this shouldn't happen but treat it as not-granted
+        // defensively), or a swallowed exception from runCatching —
+        // means we don't actually own playback yet, so the VM must
+        // start muted. The earlier shape used `!= FAILED` which fell
+        // through to Granted on `null` (exception path) and on
+        // DELAYED (= 2), playing audio over a call that the OS hadn't
+        // released to us.
+        val result = runCatching { mgr.requestAudioFocus(request) }.getOrNull()
+        if (result == AudioManager.AUDIOFOCUS_REQUEST_GRANTED) {
+            NestAudioFocusBus.publish(NestAudioFocusState.Granted)
+        } else {
+            NestAudioFocusBus.publish(NestAudioFocusState.TransientLoss)
+        }
         audioFocusRequest = request
     }
 
     private fun abandonAudioFocus() {
         val request = audioFocusRequest ?: return
-        val mgr = getSystemService(Context.AUDIO_SERVICE) as android.media.AudioManager
+        val mgr = getSystemService(Context.AUDIO_SERVICE) as AudioManager
         runCatching { mgr.abandonAudioFocusRequest(request) }
         audioFocusRequest = null
+        // Reset to Granted so a future foreground-service start that
+        // happens before the next focus request lands doesn't inherit
+        // a stale "muted because we lost focus" state.
+        NestAudioFocusBus.publish(NestAudioFocusState.Granted)
     }
 
+    private fun translateFocusChange(focusChange: Int): NestAudioFocusState =
+        when (focusChange) {
+            AudioManager.AUDIOFOCUS_GAIN,
+            AudioManager.AUDIOFOCUS_GAIN_TRANSIENT,
+            AudioManager.AUDIOFOCUS_GAIN_TRANSIENT_MAY_DUCK,
+            AudioManager.AUDIOFOCUS_GAIN_TRANSIENT_EXCLUSIVE,
+            -> NestAudioFocusState.Granted
+
+            AudioManager.AUDIOFOCUS_LOSS_TRANSIENT,
+            AudioManager.AUDIOFOCUS_LOSS_TRANSIENT_CAN_DUCK,
+            -> NestAudioFocusState.TransientLoss
+
+            AudioManager.AUDIOFOCUS_LOSS -> NestAudioFocusState.Loss
+
+            // Unknown future codes: be defensive and treat as Granted
+            // so a vendor-specific extension can't silently mute the
+            // room forever.
+            else -> NestAudioFocusState.Granted
+        }
+
     override fun onStartCommand(
         intent: Intent?,
         flags: Int,
@@ -217,6 +407,8 @@ class NestForegroundService : Service() {
         wakeLock?.takeIf { it.isHeld }?.release()
         wakeLock = null
         abandonAudioFocus()
+        unregisterAudioDeviceCallback()
+        unregisterNetworkCallback()
         super.onDestroy()
     }
 

commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestAudioFocusBus.kt

@@ -0,0 +1,92 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.commons.viewmodels
+
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
+
+/**
+ * Process-wide audio-focus signal published by the platform-side audio
+ * focus listener (Android: `NestForegroundService`'s
+ * `OnAudioFocusChangeListener`) and consumed by every active
+ * [NestViewModel].
+ *
+ * Decoupled from the platform audio APIs so commons can stay free of
+ * `android.media.AudioManager` references — the platform layer
+ * translates focus-change codes into the small enum below before
+ * publishing.
+ *
+ * Why a singleton: there's at most one foreground audio-room session
+ * per process (the foreground service is unique), and every ViewModel
+ * wants to observe the same focus signal, so keying by some other
+ * dimension would just complicate the wiring. The bus survives
+ * activity / VM rotations naturally.
+ */
+object NestAudioFocusBus {
+    private val _state = MutableStateFlow(NestAudioFocusState.Granted)
+
+    /**
+     * The latest audio-focus state. Defaults to [NestAudioFocusState.Granted]
+     * — i.e. "we own playback" — so consumers don't have to special-case
+     * the boot-up window before the foreground service has a chance to
+     * register its listener.
+     */
+    val state: StateFlow<NestAudioFocusState> = _state.asStateFlow()
+
+    /**
+     * Publish a new focus state. Only the platform-side listener calls
+     * this; consumers observe via [state].
+     */
+    fun publish(newState: NestAudioFocusState) {
+        _state.value = newState
+    }
+}
+
+/**
+ * Audio-focus state translated from the platform's `AUDIOFOCUS_*` codes.
+ * Maps them onto the three actions an audio-room app actually cares
+ * about: keep playing, pause-because-something-else-is-playing, and
+ * stop-because-something-else-is-now-the-primary-audio-source.
+ */
+enum class NestAudioFocusState {
+    /** We own playback. Normal operation. */
+    Granted,
+
+    /**
+     * We've lost focus temporarily — typical triggers are an inbound
+     * phone call, a maps voice prompt, a system alarm. The
+     * [NestViewModel] reacts by silencing playback + the broadcast
+     * mic (the user-visible "muted" state stays unchanged so it
+     * restores on regain). Audio pipeline keeps running so resume
+     * is sample-accurate.
+     */
+    TransientLoss,
+
+    /**
+     * Permanent focus loss — another app has taken over as the
+     * primary audio source for the foreseeable future. Same effective
+     * action as [TransientLoss] in v1 (silence both directions), but
+     * the distinction is preserved so future enhancements can tear
+     * down the audio device entirely on long-form loss.
+     */
+    Loss,
+}

commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestNetworkChangeBus.kt

@@ -0,0 +1,80 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.commons.viewmodels
+
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.SharedFlow
+import kotlinx.coroutines.flow.asSharedFlow
+
+/**
+ * Process-wide signal published by the platform-side connectivity
+ * listener (Android: `NestForegroundService`'s
+ * `ConnectivityManager.NetworkCallback`) and consumed by every active
+ * [NestViewModel] to recycle its [com.vitorpamplona.nestsclient.NestsListener]
+ * / [com.vitorpamplona.nestsclient.NestsSpeaker] sessions.
+ *
+ * Why this exists: when a phone hands over from Wi-Fi to cellular (or
+ * the other way), the local socket binding's source IP changes. The
+ * QUIC connection sitting on the previous socket isn't notified of
+ * the change — it'll keep retransmitting into the void until its PTO
+ * fires (`~30 s` for an idle connection). Without an external nudge
+ * the user hears 30 seconds of silence before the wrapper's reconnect
+ * loop notices the failure. With this bus, the wrapper recycles the
+ * QUIC session the moment the platform sees the network change,
+ * shrinking the audible gap to a single re-handshake (≈ 1 s on
+ * typical mobile networks).
+ *
+ * Decoupled from `android.net.ConnectivityManager` so commons stays
+ * platform-free — the platform layer translates `onAvailable` /
+ * `onLost` callbacks into a single `publish()` event before the
+ * VM observes it.
+ */
+object NestNetworkChangeBus {
+    /**
+     * `extraBufferCapacity = 1, DROP_OLDEST` — the consumer only cares
+     * that *a* network change happened, not how many. A burst of
+     * onLost / onAvailable calls during a Wi-Fi flap collapses to a
+     * single recycle event, which is exactly what we want (one
+     * handshake instead of N).
+     */
+    private val _events =
+        MutableSharedFlow<Unit>(
+            replay = 0,
+            extraBufferCapacity = 1,
+            onBufferOverflow = BufferOverflow.DROP_OLDEST,
+        )
+
+    /**
+     * Hot flow of network-change events. Consumers ([NestViewModel])
+     * collect this and call `recycleSession()` on their listener +
+     * speaker.
+     */
+    val events: SharedFlow<Unit> = _events.asSharedFlow()
+
+    /**
+     * Publish a network-change event. Only the platform-side listener
+     * calls this; consumers observe via [events].
+     */
+    fun publish() {
+        _events.tryEmit(Unit)
+    }
+}

commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt

@@ -34,6 +34,7 @@ import com.vitorpamplona.nestsclient.NestsRoomConfig
 import com.vitorpamplona.nestsclient.NestsSpeaker
 import com.vitorpamplona.nestsclient.NestsSpeakerState
 import com.vitorpamplona.nestsclient.audio.AudioCapture
+import com.vitorpamplona.nestsclient.audio.AudioException
 import com.vitorpamplona.nestsclient.audio.AudioPlayer
 import com.vitorpamplona.nestsclient.audio.NestPlayer
 import com.vitorpamplona.nestsclient.audio.OpusDecoder
@@ -282,6 +283,34 @@ class NestViewModel(
     private var speakerStateJob: Job? = null
     private var speakerConnectJob: Job? = null
 
+    /**
+     * Mirror of [NestAudioFocusBus.state] flipped to a boolean: `true`
+     * while another app holds audio focus (inbound phone call, Maps
+     * voice prompt, system alarm). Composed with the user-visible
+     * mute states (`NestUiState.isMuted` for listeners,
+     * `BroadcastUiState.Broadcasting.isMuted` for the speaker mic) via
+     * [effectiveListenMuted] / [effectiveMicMuted] so the user-chosen
+     * state is preserved across a focus loss + regain cycle.
+     *
+     * Maintained by [audioFocusObserverJob], started in [launchConnect]
+     * (so a never-connected VM doesn't subscribe needlessly).
+     */
+    @Volatile private var focusMuted: Boolean = false
+    private var audioFocusObserverJob: Job? = null
+
+    /**
+     * Subscribed in [ensureNetworkChangeObserverStarted]; calls
+     * [com.vitorpamplona.nestsclient.NestsListener.recycleSession]
+     * (and the speaker counterpart) when the platform reports a
+     * default-network change. Without this, a Wi-Fi → cellular
+     * handover leaves the QUIC connection sitting on the now-dead
+     * socket until its PTO fires (~30 s) — the wrapper's reconnect
+     * loop only learns about the failure at PTO. With the bus
+     * nudge, recycle happens immediately and the audible gap
+     * shrinks to a single re-handshake.
+     */
+    private var networkChangeObserverJob: Job? = null
+
     /**
      * Push the latest known speaker set from the room event. The user's
      * own pubkey (when broadcasting) is filtered out so we don't subscribe
@@ -317,13 +346,97 @@ class NestViewModel(
             teardown(targetState = ConnectionUiState.Idle, finalCleanup = false)
         }
 
+        ensureAudioFocusObserverStarted()
+        ensureNetworkChangeObserverStarted()
         launchConnect()
     }
 
+    /**
+     * Subscribe to [NestAudioFocusBus] for the lifetime of this VM (or
+     * until [teardown] cancels). On loss, mute every active listener
+     * + the broadcast mic without touching the user-visible mute
+     * states; on regain, restore the user's choice. Idempotent —
+     * a second call while the job is alive is a no-op.
+     */
+    private fun ensureAudioFocusObserverStarted() {
+        if (audioFocusObserverJob?.isActive == true) return
+        audioFocusObserverJob =
+            viewModelScope.launch {
+                NestAudioFocusBus.state.collect { focusState ->
+                    if (closed) return@collect
+                    val newFocusMuted = focusState != NestAudioFocusState.Granted
+                    if (newFocusMuted == focusMuted) return@collect
+                    focusMuted = newFocusMuted
+                    // Re-apply both directions. effectiveListenMuted()
+                    // and the per-broadcast effective rebuild below
+                    // both consume the just-updated [focusMuted] field.
+                    applyEffectiveListenMute()
+                    applyEffectiveMicMute()
+                }
+            }
+    }
+
+    /**
+     * Subscribe to [NestNetworkChangeBus] for the lifetime of this VM
+     * (or until [teardown]). On a default-network change, ask both
+     * the listener and the speaker wrappers to recycle their
+     * underlying sessions — much faster than waiting for the QUIC
+     * PTO on a dead socket. Idempotent.
+     */
+    private fun ensureNetworkChangeObserverStarted() {
+        if (networkChangeObserverJob?.isActive == true) return
+        networkChangeObserverJob =
+            viewModelScope.launch {
+                NestNetworkChangeBus.events.collect {
+                    if (closed) return@collect
+                    // Best-effort. The wrapper-level recycleSession is a
+                    // no-op on raw / non-reconnecting implementations
+                    // (test fakes, IETF reference) so a network change
+                    // there is silently ignored — the production paths
+                    // wrap with [connectReconnectingNestsListener] /
+                    // [connectReconnectingNestsSpeaker] which both
+                    // override the hook to close the inner session.
+                    listener?.runCatching { recycleSession() }
+                    speaker?.runCatching { recycleSession() }
+                }
+            }
+    }
+
+    /**
+     * Push the effective mic mute (`user choice OR focus-loss`) onto
+     * the live broadcast handle, if any. The user-visible
+     * [BroadcastUiState.Broadcasting.isMuted] stays the user's choice
+     * — UI only reflects that, never the focus-driven temporary
+     * silencing — so a focus regain is invisible to the user.
+     */
+    private fun applyEffectiveMicMute() {
+        val handle = broadcastHandle ?: return
+        val ui = _uiState.value.broadcast
+        val userMuted = (ui as? BroadcastUiState.Broadcasting)?.isMuted ?: false
+        val effective = userMuted || focusMuted
+        viewModelScope.launch {
+            runCatching { handle.setMuted(effective) }
+        }
+    }
+
     fun setMuted(muted: Boolean) {
         if (closed) return
         _uiState.update { it.copy(isMuted = muted) }
-        activeSubscriptions.values.forEach { it.player?.setMutedSafe(muted) }
+        applyEffectiveListenMute()
+    }
+
+    /**
+     * Effective listener mute = user-chosen mute (from the talk-bar)
+     * OR audio-focus loss (phone call / nav prompt). Applied to every
+     * active player. The user-visible state (`NestUiState.isMuted`)
+     * stays the user's choice, so a focus regain restores it
+     * automatically.
+     */
+    private fun effectiveListenMuted(): Boolean = _uiState.value.isMuted || focusMuted
+
+    private fun applyEffectiveListenMute() {
+        val effective = effectiveListenMuted()
+        activeSubscriptions.values.forEach { it.player?.setMutedSafe(effective) }
     }
 
     /**
@@ -523,7 +636,13 @@ class NestViewModel(
         if (closed) return
         val handle = broadcastHandle ?: return
         viewModelScope.launch {
-            val result = handle.runCatching { setMuted(muted) }
+            // Apply (user-mute OR audio-focus-loss) to the wire so the
+            // mic stays silent through a phone call without us losing
+            // the user's intent. UI then reflects the user choice (not
+            // the focus-effective state) so a focus-driven mute doesn't
+            // visually flip the talk button.
+            val effective = muted || focusMuted
+            val result = handle.runCatching { setMuted(effective) }
             if (closed) return@launch
             _uiState.update {
                 val current = it.broadcast
@@ -866,9 +985,26 @@ class NestViewModel(
                     throw t
                 }
             try {
-                val isMuted = _uiState.value.isMuted
+                // Apply (user-mute OR audio-focus-loss) so a speaker
+                // that comes on stage during a phone call attaches
+                // already silenced. The focus observer re-runs
+                // applyEffectiveListenMute() on regain, restoring the
+                // user's intent.
+                val isMuted = effectiveListenMuted()
                 val isHushed = pubkey in _uiState.value.locallyHushed
-                val roomPlayer = NestPlayer(decoder, player, viewModelScope)
+                val roomPlayer =
+                    NestPlayer(
+                        decoder = decoder,
+                        player = player,
+                        scope = viewModelScope,
+                        // ~100 ms of audio buffered before the AudioTrack
+                        // starts consuming. Long enough to hide a typical
+                        // Main-thread hiccup (Compose recomposition / GC)
+                        // without making the join feel laggy. Empirically
+                        // tuned in the audio-rooms audit; see NestPlayer
+                        // kdoc for details.
+                        prerollFrames = ROOM_PLAYER_PREROLL_FRAMES,
+                    )
                 // Apply current mute + per-speaker hush state before play()
                 // opens the device so the first frame respects them.
                 player.setMutedSafe(isMuted)
@@ -880,7 +1016,32 @@ class NestViewModel(
                 val instrumented = handle.objects.onEach { onSpeakerActivity(pubkey) }
                 roomPlayer.play(
                     instrumented,
-                    onError = { /* swallow per-packet decoder errors */ },
+                    onError = { err ->
+                        // Per-packet decoder errors are noise — Opus PLC
+                        // papers over a single bad frame. Swallow those.
+                        // PlaybackFailed (or DeviceUnavailable from a
+                        // deferred [AudioPlayer.beginPlayback], or the
+                        // audio-priority dispatcher dying mid-stream) is
+                        // terminal: the device is wedged for THIS
+                        // subscription. Roll the slot back so a future
+                        // reconcile can retry rather than leave a
+                        // perma-spinning speaker tile.
+                        when (err.kind) {
+                            AudioException.Kind.DecoderError,
+                            AudioException.Kind.EncoderError,
+                            -> {
+                                Unit
+                            }
+
+                            AudioException.Kind.PlaybackFailed,
+                            AudioException.Kind.DeviceUnavailable,
+                            -> {
+                                if (activeSubscriptions[pubkey] === slot) {
+                                    activeSubscriptions.remove(pubkey)?.let { closeSubscription(it) }
+                                }
+                            }
+                        }
+                    },
                     onLevel = { onAudioLevel(pubkey, it) },
                 )
                 slot.attach(handle, roomPlayer, player)
@@ -928,6 +1089,20 @@ class NestViewModel(
         announcesJob = null
         levelEmitterJob?.cancel()
         levelEmitterJob = null
+        // Audio-focus observation only ends on the final teardown —
+        // a transient disconnect+reconnect (user retry, room swap)
+        // keeps the bus subscription alive so a focus loss that
+        // happens during the gap is still picked up. The observer
+        // is idempotent under [ensureAudioFocusObserverStarted], so
+        // the next [connect] call is safe whether or not we
+        // cancelled here.
+        if (finalCleanup) {
+            audioFocusObserverJob?.cancel()
+            audioFocusObserverJob = null
+            focusMuted = false
+            networkChangeObserverJob?.cancel()
+            networkChangeObserverJob = null
+        }
         rawAudioLevels.clear()
         if (_audioLevels.value.isNotEmpty()) {
             _audioLevels.value = emptyMap()
@@ -1246,6 +1421,17 @@ sealed class BroadcastUiState {
  */
 const val SPEAKING_TIMEOUT_MS: Long = 250L
 
+/**
+ * Per-speaker pre-roll: number of decoded PCM frames buffered before the
+ * underlying [com.vitorpamplona.nestsclient.audio.AudioPlayer] starts
+ * consuming. 5 × 20 ms ≈ 100 ms of audio — long enough to mask a typical
+ * Main-thread stall (Compose recomposition / GC) without adding perceptible
+ * join latency. Combines with [com.vitorpamplona.nestsclient.audio.AudioTrackPlayer]'s
+ * ~250 ms AudioTrack buffer for ~350 ms of total slack between the
+ * arrival-of-frame and the underrun horizon.
+ */
+const val ROOM_PLAYER_PREROLL_FRAMES: Int = 5
+
 /**
  * Coalescing interval for [NestViewModel.audioLevels]. The decode loop
  * pushes a fresh peak every ~20 ms (one per Opus frame); we publish to

nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRecordCapture.kt

@@ -22,6 +22,10 @@ package com.vitorpamplona.nestsclient.audio
 
 import android.media.AudioRecord
 import android.media.MediaRecorder
+import android.media.audiofx.AcousticEchoCanceler
+import android.media.audiofx.AutomaticGainControl
+import android.media.audiofx.NoiseSuppressor
+import com.vitorpamplona.quartz.utils.Log
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.withContext
 import android.media.AudioFormat as AndroidAudioFormat
@@ -32,6 +36,19 @@ import android.media.AudioFormat as AndroidAudioFormat
  * voice-chat libraries use, so it gets the platform's echo-cancellation and
  * noise-suppression filters when available.
  *
+ * **Audio effects (AEC / NS / AGC).** On most modern Android devices the
+ * `VOICE_COMMUNICATION` source is enough to engage the platform's echo
+ * canceller automatically. On a small set of older / OEM-customised
+ * devices it isn't — the AEC engine only attaches when the device is in
+ * `MODE_IN_COMMUNICATION`, which an audio-room app deliberately avoids
+ * driving (it reroutes everything through the call audio path and shows a
+ * "phone call" notification icon). To cover those devices without
+ * touching `AudioManager.mode` we explicitly attach the standalone
+ * [AcousticEchoCanceler] / [NoiseSuppressor] / [AutomaticGainControl]
+ * effects to the AudioRecord's session id. On devices where the source
+ * already engages them, attaching a second effect is a no-op (the
+ * platform deduplicates by session id) — so this is purely additive.
+ *
  * **Permission:** the caller is responsible for holding `RECORD_AUDIO` before
  * calling [start]; this class will throw [AudioException.Kind.DeviceUnavailable]
  * if the OS denies the resource.
@@ -40,6 +57,9 @@ class AudioRecordCapture(
     private val source: Int = MediaRecorder.AudioSource.VOICE_COMMUNICATION,
 ) : AudioCapture {
     private var record: AudioRecord? = null
+    private var aec: AcousticEchoCanceler? = null
+    private var ns: NoiseSuppressor? = null
+    private var agc: AutomaticGainControl? = null
     private var stopped = false
 
     override fun start() {
@@ -97,6 +117,34 @@ class AudioRecordCapture(
             )
         }
         record = rec
+        // Attach standalone audio effects to the session. This covers
+        // devices where the VOICE_COMMUNICATION source alone doesn't
+        // engage AEC (typically because they only attach AEC under
+        // MODE_IN_COMMUNICATION). All three are best-effort — a device
+        // that doesn't support an effect leaves it null without
+        // affecting capture.
+        attachAudioEffects(rec.audioSessionId)
+    }
+
+    private fun attachAudioEffects(sessionId: Int) {
+        if (AcousticEchoCanceler.isAvailable()) {
+            aec =
+                runCatching { AcousticEchoCanceler.create(sessionId)?.apply { enabled = true } }
+                    .onFailure { Log.w("NestTx") { "AcousticEchoCanceler.create failed: ${it.message}" } }
+                    .getOrNull()
+        }
+        if (NoiseSuppressor.isAvailable()) {
+            ns =
+                runCatching { NoiseSuppressor.create(sessionId)?.apply { enabled = true } }
+                    .onFailure { Log.w("NestTx") { "NoiseSuppressor.create failed: ${it.message}" } }
+                    .getOrNull()
+        }
+        if (AutomaticGainControl.isAvailable()) {
+            agc =
+                runCatching { AutomaticGainControl.create(sessionId)?.apply { enabled = true } }
+                    .onFailure { Log.w("NestTx") { "AutomaticGainControl.create failed: ${it.message}" } }
+                    .getOrNull()
+        }
     }
 
     override suspend fun readFrame(): ShortArray? {
@@ -128,6 +176,16 @@ class AudioRecordCapture(
     override fun stop() {
         if (stopped) return
         stopped = true
+        // Release the audio effects BEFORE the AudioRecord — they hold
+        // a session-id reference and the platform expects effect
+        // teardown to precede the AudioRecord.release() that frees
+        // the session.
+        runCatching { aec?.release() }
+        runCatching { ns?.release() }
+        runCatching { agc?.release() }
+        aec = null
+        ns = null
+        agc = null
         val rec = record ?: return
         record = null
         runCatching { rec.stop() }

nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioTrackPlayer.kt

@@ -22,8 +22,12 @@ package com.vitorpamplona.nestsclient.audio
 
 import android.media.AudioAttributes
 import android.media.AudioTrack
-import kotlinx.coroutines.Dispatchers
+import android.os.Process
+import kotlinx.coroutines.ExecutorCoroutineDispatcher
+import kotlinx.coroutines.asCoroutineDispatcher
 import kotlinx.coroutines.withContext
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
 import android.media.AudioFormat as AndroidAudioFormat
 
 /**
@@ -45,9 +49,30 @@ import android.media.AudioFormat as AndroidAudioFormat
  * `MediaRecorder.AudioSource.VOICE_COMMUNICATION` regardless of the
  * playback usage.
  *
- * Buffer sizing: 4× minimum so the producer can fall behind by ~80 ms before
- * dropouts, which roughly matches the jitter the WebTransport datagram path
- * introduces over typical mobile networks.
+ * Buffer sizing: target ~250 ms of slack, computed as
+ * `max(minBuffer * 16, 250 ms-equivalent)`. The previous 4× minimum (~80 ms
+ * by the device-reported floor) underran on devices with very small
+ * `getMinBufferSize` returns and on any handset whose decode loop got
+ * stalled by Compose recomposition / GC on Main. 250 ms matches the
+ * jitter-buffer depth Spaces / Clubhouse use for hands-free audio rooms,
+ * and combined with the per-subscription pre-roll in [NestPlayer] keeps
+ * the AudioTrack from underruning across typical mobile network jitter.
+ *
+ * Threading: writes go through a per-instance audio-priority single-thread
+ * dispatcher (`HandlerThread` + [Process.THREAD_PRIORITY_AUDIO]) instead of
+ * `Dispatchers.IO`. The previous shape did one IO dispatcher hop per Opus
+ * frame (~50 hops/sec/speaker) and contended with whatever else `Dispatchers.IO`
+ * was running; an audio-priority dedicated thread gets reliable scheduling
+ * and removes the contention.
+ *
+ * Two-phase startup: [start] allocates the AudioTrack (in stopped state)
+ * + spins up the writer thread. [beginPlayback] calls `AudioTrack.play()`
+ * to flip the device into the playing state. Splitting the two lets
+ * [NestPlayer] pre-roll several decoded frames into the AudioTrack's
+ * internal buffer BEFORE the hardware starts pulling samples, so playback
+ * begins with ~100 ms of buffered audio instead of underrunning on the
+ * first frame. AudioTrack in `MODE_STREAM` explicitly supports `write()`
+ * before `play()` per the platform docs — this is the intended pattern.
  */
 class AudioTrackPlayer(
     private val usage: Int = AudioAttributes.USAGE_MEDIA,
@@ -57,6 +82,18 @@ class AudioTrackPlayer(
     private var muted: Boolean = false
     private var volume: Float = 1f
 
+    /**
+     * Dedicated audio-priority single-thread executor for AudioTrack writes.
+     * Lazily created on [start] and shut down on [stop] so a never-started
+     * player doesn't leak a thread. `THREAD_PRIORITY_AUDIO` is the standard
+     * Linux nice level for VoIP / WebRTC playback paths on Android — it sits
+     * above background but below true audio-callback priority, so the OS
+     * scheduler keeps it running through GC / Compose recomposition without
+     * starving other threads.
+     */
+    private var audioExecutor: ExecutorService? = null
+    private var audioDispatcher: ExecutorCoroutineDispatcher? = null
+
     override fun start() {
         if (track != null) return
 
@@ -79,7 +116,14 @@ class AudioTrackPlayer(
                 "AudioTrack.getMinBufferSize returned $minBuffer for ${AudioFormat.SAMPLE_RATE_HZ} Hz",
             )
         }
-        val bufferBytes = minBuffer * 4
+        // Target ~250 ms of audio: enough headroom so the decode loop can
+        // miss its 20 ms cadence by an order of magnitude before the device
+        // underruns. Take the larger of `minBuffer * 16` and an explicit
+        // 250 ms-equivalent so devices that report a small minBuffer still
+        // get the same wall-clock slack.
+        val targetBytes250Ms =
+            (AudioFormat.SAMPLE_RATE_HZ / 4) * AudioFormat.BYTES_PER_SAMPLE * AudioFormat.CHANNELS
+        val bufferBytes = maxOf(minBuffer * 16, targetBytes250Ms)
 
         val newTrack =
             try {
@@ -109,25 +153,66 @@ class AudioTrackPlayer(
                 )
             }
 
+        // NOTE: deliberately NOT calling `newTrack.play()` here — that's
+        // [beginPlayback]'s job. AudioTrack.write is legal in the not-yet-
+        // playing state for `MODE_STREAM`, which is the contract that lets
+        // [NestPlayer] pre-roll decoded frames into the buffer before the
+        // hardware starts consuming.
+        applyMuteVolume(newTrack)
+        // Spin up the audio-priority writer thread. The executor is private
+        // to this player instance so per-speaker NestPlayer pumps don't
+        // contend on a shared queue. Priority is set inside the thread's
+        // Runnable because Linux thread priority is per-OS-thread, not
+        // per-Java Thread; `Thread.setPriority` does NOT translate to a
+        // Linux nice level on Android. `Process.setThreadPriority` does.
+        val executor =
+            Executors.newSingleThreadExecutor { r ->
+                Thread(
+                    {
+                        runCatching { Process.setThreadPriority(Process.THREAD_PRIORITY_AUDIO) }
+                        r.run()
+                    },
+                    "nest-audio-writer",
+                )
+            }
+        audioExecutor = executor
+        audioDispatcher = executor.asCoroutineDispatcher()
+        track = newTrack
+    }
+
+    override fun beginPlayback() {
+        // Idempotent: AudioTrack.play() on an already-playing track is a
+        // no-op, and we gate on the track itself being non-null so a
+        // beginPlayback before start (or after stop) silently no-ops
+        // rather than blowing up — matches the rest of the player's
+        // tolerant-of-misordered-calls posture.
+        val t = track ?: return
         try {
-            newTrack.play()
-        } catch (t: Throwable) {
-            runCatching { newTrack.release() }
+            t.play()
+        } catch (e: Throwable) {
+            // PLAY-on-uninitialized AudioTrack is the only realistic
+            // failure here, and we already guard against an unallocated
+            // track above. Throw so [NestPlayer]'s outer catch surfaces
+            // it via `onError(AudioException.PlaybackFailed)` — same path
+            // that handles every other mid-stream device failure.
             throw AudioException(
                 AudioException.Kind.DeviceUnavailable,
                 "AudioTrack.play() rejected start",
-                t,
+                e,
             )
         }
-        applyMuteVolume(newTrack)
-        track = newTrack
     }
 
     override suspend fun enqueue(pcm: ShortArray) {
         val t = track ?: throw AudioException(AudioException.Kind.PlaybackFailed, "player not started")
-        // AudioTrack.write blocks if the internal buffer is full. Run on IO so
-        // we don't stall a coroutine dispatcher backed by a small thread pool.
-        withContext(Dispatchers.IO) {
+        val dispatcher =
+            audioDispatcher
+                ?: throw AudioException(AudioException.Kind.PlaybackFailed, "audio dispatcher not initialized")
+        // AudioTrack.write blocks if the internal buffer is full. Run on the
+        // per-instance audio-priority writer thread so the WRITE_BLOCKING
+        // suspension is on a thread the OS schedules tightly, and so we
+        // don't compete with whatever else `Dispatchers.IO` is running.
+        withContext(dispatcher) {
             val written = t.write(pcm, 0, pcm.size, AudioTrack.WRITE_BLOCKING)
             if (written < 0) {
                 throw AudioException(
@@ -155,6 +240,16 @@ class AudioTrackPlayer(
         runCatching { t.flush() }
         runCatching { t.stop() }
         runCatching { t.release() }
+        // Tear down the audio-priority writer. close() on the
+        // ExecutorCoroutineDispatcher shuts down the executor (and the
+        // underlying single thread) — pending writes are abandoned rather
+        // than blocked on, since the AudioTrack itself has already been
+        // stopped + released two lines up so any further `write` would
+        // throw IllegalStateException anyway.
+        audioDispatcher?.close()
+        audioDispatcher = null
+        audioExecutor?.let { runCatching { it.shutdownNow() } }
+        audioExecutor = null
     }
 
     private fun applyMuteVolume(track: AudioTrack) {

nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusDecoder.kt

@@ -57,8 +57,33 @@ class MediaCodecOpusDecoder : OpusDecoder {
     override fun decode(opusPacket: ByteArray): ShortArray {
         check(!released) { "decoder released" }
 
-        val inputIndex = codec.dequeueInputBuffer(DEQUEUE_TIMEOUT_US)
-        if (inputIndex < 0) return ShortArray(0)
+        // Pre-sized output. Write decoded samples directly into the array
+        // via ShortBuffer.get(dst, off, len) — the previous shape went
+        // through ArrayList<Short>, which boxed every PCM sample
+        // (~48 000 alloc/sec/speaker on the audio hot path).
+        val out = ShortArray(AudioFormat.FRAME_SIZE_SAMPLES)
+        var outPos = 0
+
+        // 1. Acquire an input slot. On rare back-pressure (output buffers
+        //    haven't been drained fast enough → all input slots are tied
+        //    up), drain whatever output IS ready first to free slots,
+        //    then retry input dequeue with a longer timeout. The previous
+        //    shape returned `ShortArray(0)` immediately when the first
+        //    10 ms dequeue missed, turning every transient stall (thermal
+        //    throttle, GC pause) into a 20 ms audio gap.
+        var inputIndex = codec.dequeueInputBuffer(DEQUEUE_TIMEOUT_US)
+        if (inputIndex < 0) {
+            outPos = drainAvailableOutput(out, outPos)
+            inputIndex = codec.dequeueInputBuffer(DEQUEUE_RETRY_TIMEOUT_US)
+            if (inputIndex < 0) {
+                // Genuine input starvation past the retry window. Drop
+                // this packet (presentationTimeUs is NOT advanced — Opus
+                // PLC will paper over the gap on the listener) and
+                // return whatever we drained from output so the player
+                // doesn't underrun on the back of one tight cycle.
+                return if (outPos > 0) out.copyOf(outPos) else ShortArray(0)
+            }
+        }
         val inputBuffer =
             codec.getInputBuffer(inputIndex)
                 ?: throw AudioException(
@@ -71,17 +96,25 @@ class MediaCodecOpusDecoder : OpusDecoder {
         // Advance presentation time by one 20 ms frame.
         presentationTimeUs += FRAME_DURATION_US
 
-        // Drain whatever output is ready right now. A single Opus packet
-        // typically yields exactly one output buffer, but on some devices the
-        // first call returns INFO_OUTPUT_FORMAT_CHANGED before the PCM frame.
-        //
-        // Write the decoded samples directly into a pre-sized ShortArray
-        // via ShortBuffer.get(dst, off, len) — the previous shape went via
-        // ArrayList<Short>, which boxed every PCM sample (one heap object
-        // per Short × 960 samples × 50 fps × N speakers ≈ 48 000
-        // allocations/sec/speaker on the audio hot path).
-        val out = ShortArray(AudioFormat.FRAME_SIZE_SAMPLES)
-        var outPos = 0
+        // 2. Drain whatever output is ready right now. A single Opus
+        //    packet typically yields exactly one output buffer, but
+        //    [drainAvailableOutput] tolerates an INFO_OUTPUT_FORMAT_CHANGED
+        //    that some devices emit before the first PCM frame.
+        outPos = drainAvailableOutput(out, outPos)
+        return if (outPos == out.size) out else out.copyOf(outPos)
+    }
+
+    /**
+     * Drain any output buffers MediaCodec has ready, copying samples into
+     * [out] starting at [startPos]. Returns the new write position.
+     * Stops on the first empty / EOS / TRY_AGAIN_LATER signal — never
+     * blocks past the [DEQUEUE_TIMEOUT_US] dequeue probe.
+     */
+    private fun drainAvailableOutput(
+        out: ShortArray,
+        startPos: Int,
+    ): Int {
+        var outPos = startPos
         var formatChangeAbsorbed = false
         drain@ while (true) {
             val outputIndex = codec.dequeueOutputBuffer(bufferInfo, DEQUEUE_TIMEOUT_US)
@@ -126,7 +159,7 @@ class MediaCodecOpusDecoder : OpusDecoder {
                 }
             }
         }
-        return if (outPos == out.size) out else out.copyOf(outPos)
+        return outPos
     }
 
     override fun release() {
@@ -138,6 +171,20 @@ class MediaCodecOpusDecoder : OpusDecoder {
 
     companion object {
         private const val DEQUEUE_TIMEOUT_US = 10_000L // 10 ms
+
+        /**
+         * Fallback timeout after the first input dequeue misses + we
+         * drain pending output to free slots. Deliberately well under
+         * the 20 ms frame cadence so a tight back-pressure window is
+         * absorbed without falling off the per-frame budget. The
+         * previous code took a 0 ms retry (i.e. dropped the frame
+         * outright on the first miss), which turned every transient
+         * stall into ~20 ms of silence. 5 ms gives MediaCodec time to
+         * free a slot after the output drain without blowing the
+         * frame budget.
+         */
+        private const val DEQUEUE_RETRY_TIMEOUT_US = 5_000L // 5 ms
+
         private const val FRAME_DURATION_US = 20_000L // 20 ms
 
         private fun buildFormat(): MediaFormat {

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt

@@ -59,7 +59,8 @@ class MoqLiteNestsSpeaker internal constructor(
      * Defaults to [NestMoqLiteBroadcaster.DEFAULT_FRAMES_PER_GROUP].
      */
     private val framesPerGroup: Int = NestMoqLiteBroadcaster.DEFAULT_FRAMES_PER_GROUP,
-) : NestsSpeaker {
+) : NestsSpeaker,
+    HotSwappablePublisherSource {
     override val state: StateFlow<NestsSpeakerState> = mutableState.asStateFlow()
 
     private val gate = Mutex()
@@ -100,7 +101,7 @@ class MoqLiteNestsSpeaker internal constructor(
                     NestMoqLiteBroadcaster(
                         capture = captureFactory(),
                         encoder = encoderFactory(),
-                        publisher = publisher,
+                        initialPublisher = publisher,
                         scope = scope,
                         framesPerGroup = framesPerGroup,
                     ).also {
@@ -113,7 +114,7 @@ class MoqLiteNestsSpeaker internal constructor(
                                 // the session — without this signal the
                                 // outward state stays on Broadcasting
                                 // and the room is silently mute.
-                                onBroadcastTerminalFailure()
+                                reportBroadcastTerminalFailure()
                             },
                             onLevel = onLevel,
                         )
@@ -139,6 +140,15 @@ class MoqLiteNestsSpeaker internal constructor(
         }
     }
 
+    /**
+     * [HotSwappablePublisherSource] implementation. See the interface
+     * kdoc — this method mints a fresh publisher on the session
+     * WITHOUT spinning up a broadcaster on top of it. Used by the
+     * reconnect wrapper's hot-swap path; not called from the
+     * non-reconnecting path which goes through [startBroadcasting].
+     */
+    override suspend fun openPublisherForHotSwap(track: String): MoqLitePublisherHandle = session.publish(broadcastSuffix = speakerPubkeyHex, track = track)
+
     /**
      * Compare-and-clear that runs from inside [close] (already holds
      * [gate]) and from [MoqLiteBroadcastHandle.close] (doesn't).
@@ -160,8 +170,13 @@ class MoqLiteNestsSpeaker internal constructor(
      * the reconnect orchestrator (`ReconnectingNestsSpeaker`) observes
      * a terminal state and recycles the session. No-op if the speaker
      * is already in a terminal state.
+     *
+     * Also exposed via [HotSwappablePublisherSource.reportBroadcastTerminalFailure]
+     * so the hot-swap pump (which owns its own long-lived broadcaster)
+     * can drive the same orchestrator-reconnect path the legacy
+     * `startBroadcasting` flow does.
      */
-    private fun onBroadcastTerminalFailure() {
+    override fun reportBroadcastTerminalFailure() {
         val current = mutableState.value
         if (current is NestsSpeakerState.Failed || current is NestsSpeakerState.Closed) return
         mutableState.value =
@@ -257,3 +272,40 @@ internal class MoqLiteBroadcastHandle(
         parent.broadcastClosed(this)
     }
 }
+
+/**
+ * Internal hot-swap seam: speakers that expose this interface let the
+ * reconnect wrapper retarget a long-lived
+ * [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster] onto a
+ * freshly-opened moq-lite session's publisher without restarting the
+ * AudioRecord / Opus encoder pipeline. Implemented by
+ * [MoqLiteNestsSpeaker]; not implemented by the IETF reference
+ * [DefaultNestsSpeaker], which falls back to the close-then-restart path
+ * inside [com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker].
+ *
+ * The wrapper uses an `as?` cast to detect support so this interface
+ * can stay package-internal — protocol consumers never see it.
+ */
+internal interface HotSwappablePublisherSource {
+    /**
+     * Open a fresh [MoqLitePublisherHandle] on the underlying moq-lite
+     * session. Caller owns the returned handle's lifetime (typically
+     * via [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster.swapPublisher]'s
+     * close-the-old contract).
+     */
+    suspend fun openPublisherForHotSwap(track: String): MoqLitePublisherHandle
+
+    /**
+     * Surface a broadcast-pipeline terminal failure (e.g. sustained
+     * `publisher.send` errors past
+     * [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster.MAX_CONSECUTIVE_SEND_ERRORS])
+     * by flipping the speaker's state to [NestsSpeakerState.Failed].
+     * Called by the hot-swap pump when the long-lived broadcaster's
+     * `onTerminalFailure` fires; lets the reconnect orchestrator
+     * observe the terminal state and recycle the session, matching
+     * the legacy
+     * [MoqLiteNestsSpeaker.startBroadcasting] path's failure
+     * propagation.
+     */
+    fun reportBroadcastTerminalFailure()
+}

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsListener.kt

@@ -116,6 +116,27 @@ interface NestsListener {
             "announces() is moq-lite-only; IETF listener has no announce-prefix flow.",
         )
 
+    /**
+     * Force the underlying transport / MoQ session to be torn down and
+     * a fresh one opened in its place — without permanently closing
+     * this listener. Used by the platform layer on a network change
+     * (Wi-Fi ↔ cellular handover) to skip the ~30 s QUIC PTO that
+     * would otherwise have to fire before the wrapper notices the
+     * old socket is dead.
+     *
+     * Default no-op for non-reconnecting implementations (raw
+     * [DefaultNestsListener] / [MoqLiteNestsListener]) — there's no
+     * orchestrator to drive a reconnect, so a force recycle would
+     * just close the listener. The reconnecting wrapper
+     * (`ReconnectingHandle` from [connectReconnectingNestsListener])
+     * overrides to close the active inner listener so its
+     * orchestrator opens a fresh session.
+     */
+    suspend fun recycleSession() {
+        // no-op — only the reconnecting wrapper has anywhere to
+        // recycle to.
+    }
+
     /** Tear down the MoQ session + underlying transport. Idempotent. */
     suspend fun close()
 }

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsSpeaker.kt

@@ -66,6 +66,27 @@ interface NestsSpeaker {
      */
     suspend fun startBroadcasting(onLevel: (Float) -> Unit = { /* no-op */ }): BroadcastHandle
 
+    /**
+     * Force the underlying transport / MoQ session to be torn down and
+     * a fresh one opened in its place — without permanently closing
+     * this speaker. Mirror of [NestsListener.recycleSession]; same
+     * use case (network handover) and same default no-op for non-
+     * reconnecting implementations.
+     *
+     * The reconnecting wrapper from
+     * [connectReconnectingNestsSpeaker] overrides to close the
+     * inner speaker so its orchestrator opens a fresh session.
+     * For moq-lite speakers, the long-lived
+     * [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster]
+     * keeps capturing through the recycle and the
+     * [BroadcastHandle.swapPublisher]-based pump retargets onto
+     * the new session's publisher with no audible gap.
+     */
+    suspend fun recycleSession() {
+        // no-op — only the reconnecting wrapper has anywhere to
+        // recycle to.
+    }
+
     /** Tear down the MoQ session + transport. Idempotent. */
     suspend fun close()
 }

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt

@@ -340,6 +340,14 @@ private class ReconnectingHandle(
                         }
                     if (terminalOrConnected !is NestsListenerState.Connected) return@collectLatest
 
+                    // Backoff for the "opener threw" retry path — exponential
+                    // 250 → 500 → 1 000 ms with reset on success. The previous
+                    // shape used a flat 1 000 ms which, combined with the
+                    // 64-frame (~1.3 s) wrapper buffer, just barely hid the
+                    // first-retry gap; a quick retry usually succeeds because
+                    // moq-rs propagates announces in < 200 ms.
+                    var subscribeRetryDelayMs = SUBSCRIBE_RETRY_BACKOFF_INITIAL_MS
+
                     while (currentCoroutineContext().isActive) {
                         val handle =
                             try {
@@ -368,14 +376,21 @@ private class ReconnectingHandle(
                                 // backoff used between publisher cycles.
                                 Log.w("NestRx") {
                                     "ReconnectingHandle.opener threw ${t::class.simpleName}: ${t.message} " +
-                                        "— retrying after ${SUBSCRIBE_RETRY_BACKOFF_MS}ms"
+                                        "— retrying after ${subscribeRetryDelayMs}ms"
                                 }
                                 null
                             }
                         if (handle == null) {
-                            delay(SUBSCRIBE_RETRY_BACKOFF_MS)
+                            delay(subscribeRetryDelayMs)
+                            subscribeRetryDelayMs =
+                                (subscribeRetryDelayMs * 2)
+                                    .coerceAtMost(SUBSCRIBE_RETRY_BACKOFF_MAX_MS)
                             continue
                         }
+                        // Success: reset the backoff so the next publisher-
+                        // cycle gap starts at the floor again (rather than
+                        // inheriting the last failure window's saturation).
+                        subscribeRetryDelayMs = SUBSCRIBE_RETRY_BACKOFF_INITIAL_MS
                         liveHandleRef.set(handle)
                         try {
                             handle.objects.collect { frames.emit(it) }
@@ -405,6 +420,29 @@ private class ReconnectingHandle(
         )
     }
 
+    /**
+     * Force-close the active inner listener so the orchestrator's
+     * `terminalAwait` returns Closed and the next loop iteration opens
+     * a fresh session. Used by the platform layer on a network change
+     * (Wi-Fi ↔ cellular) — without this, the wrapper would have to
+     * wait for the QUIC PTO to fire on the now-dead old socket
+     * (~30 s of silence) before noticing.
+     *
+     * The SubscribeHandle re-issuance pump cuts existing subs over to
+     * the new session as soon as it's Connected — same path the
+     * JWT-refresh recycle uses — so the consumer-facing
+     * [SubscribeHandle.objects] flow keeps emitting once the new
+     * session lands.
+     */
+    override suspend fun recycleSession() {
+        val current = activeListener.value ?: return
+        // Best-effort. If the close races a concurrent reconnect path
+        // (extremely rare — we only call this on deliberate user /
+        // platform signals), the orchestrator absorbs both via its
+        // existing terminal-state handling.
+        runCatching { current.close() }
+    }
+
     override suspend fun close() {
         orchestrator.cancel()
         runCatching { activeListener.value?.close() }
@@ -427,14 +465,24 @@ private class ReconnectingHandle(
         // gone publisher doesn't spin the relay with re-subscribes.
         private const val RESUBSCRIBE_BACKOFF_MS = 100L
 
-        // Backoff after an opener throws (typically: subscribe-before-
-        // announce arrives at the relay before the publisher exists,
-        // and the relay FINs the bidi without a SubscribeOk/Drop). One
-        // second is well over moq-rs's typical announce-propagation
-        // latency (< 200 ms in production traces), so the next retry
-        // usually succeeds; long enough that a never-arriving publisher
-        // doesn't hammer the relay either.
-        private const val SUBSCRIBE_RETRY_BACKOFF_MS = 1_000L
+        // Exponential backoff for the opener-throws retry path.
+        // Typical case: subscribe-before-announce arrives at the relay
+        // before the publisher exists, and the relay FINs the bidi
+        // without a SubscribeOk/Drop reply.
+        //
+        // - INITIAL = 250 ms: under moq-rs's typical announce-propagation
+        //   latency (< 200 ms in production traces), so the very first
+        //   retry usually succeeds and the listener-side buffer hides
+        //   the gap.
+        // - Doubles each miss → 500 ms → 1 000 ms.
+        // - MAX = 1 000 ms: matches the previous flat constant; long
+        //   enough that a never-arriving publisher doesn't hammer the
+        //   relay, short enough to stay under the wrapper SharedFlow's
+        //   ~1.3 s buffer once playback is rolling on the next attempt.
+        // - Reset on first successful subscribe so a subsequent
+        //   publisher-cycle gap doesn't inherit the previous saturation.
+        private const val SUBSCRIBE_RETRY_BACKOFF_INITIAL_MS = 250L
+        private const val SUBSCRIBE_RETRY_BACKOFF_MAX_MS = 1_000L
 
         private val SYNTH_OK =
             SubscribeOk(

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt

@@ -21,6 +21,7 @@
 package com.vitorpamplona.nestsclient
 
 import com.vitorpamplona.nestsclient.audio.AudioCapture
+import com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster
 import com.vitorpamplona.nestsclient.audio.OpusEncoder
 import com.vitorpamplona.nestsclient.transport.WebTransportFactory
 import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
@@ -68,15 +69,23 @@ import java.util.concurrent.atomic.AtomicReference
  * `close` cancels the re-issue pump and best-effort closes the
  * latest live handle.
  *
- * **Audio gap during refresh** — the wrapper closes the current
- * underlying speaker (which stops the mic capture + Opus encoder +
- * publisher) before opening the next, so the listener side will
- * hear ~50–150 ms of silence at each recycle boundary. That's the
- * trade-off we pay for a clean session swap; the alternative
- * (carrying the mic capture across sessions) would require deeper
- * plumbing into the audio pipeline. Acceptable for v1 —
- * 9-min-spaced 150 ms gaps are well below the noise floor of a
- * voice call.
+ * **Audio gap during refresh — eliminated for moq-lite** — when the
+ * underlying speaker implements [HotSwappablePublisherSource]
+ * (which [MoqLiteNestsSpeaker] does), the wrapper keeps a single
+ * long-lived [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster]
+ * alive across session recycles and only swaps the
+ * [com.vitorpamplona.nestsclient.moq.lite.MoqLitePublisherHandle]
+ * underneath it. The mic + encoder run continuously through the
+ * boundary and the listener hears no silence. The old session
+ * close runs in parallel with the new openOnce so the WebTransport
+ * teardown doesn't block the swap.
+ *
+ * **Audio gap during refresh — legacy path** — for speakers that
+ * don't implement [HotSwappablePublisherSource] (the IETF reference
+ * `DefaultNestsSpeaker` and test fakes), the wrapper falls back to
+ * close-then-restart: the listener hears ~50–150 ms of silence per
+ * recycle. Acceptable because the IETF path is reference-only —
+ * production runs the moq-lite hot-swap path.
  *
  * Cancellation: cancelling [scope] (typically the room screen's VM
  * scope) cancels the reconnect loop and closes both the active
@@ -186,16 +195,41 @@ suspend fun connectReconnectingNestsSpeaker(
                         }
                     if (terminal == null) {
                         // Refresh deadline hit before any terminal state —
-                        // planned recycle, not a failure. Close the old
-                        // speaker; don't bump `attempt` (it's not a
-                        // backoff event) so the next openOnce() runs
-                        // immediately.
-                        try {
-                            speaker.close()
-                        } catch (ce: kotlinx.coroutines.CancellationException) {
-                            throw ce
-                        } catch (_: Throwable) {
-                            // Best-effort.
+                        // planned recycle, not a failure. Don't bump
+                        // `attempt` (it's not a backoff event) so the next
+                        // openOnce() runs immediately.
+                        //
+                        // Hand the old speaker's close to a separate
+                        // launch so it runs IN PARALLEL with the next
+                        // iteration's openOnce(). Sequence on the wire:
+                        //
+                        //   1. (concurrent) old speaker close — closes
+                        //      old session's publisher + WebTransport.
+                        //   2. (concurrent) new openOnce() → sets
+                        //      activeSpeaker = newSpeaker.
+                        //   3. Wrapper's hot-swap pump observes the
+                        //      activeSpeaker change, opens a publisher
+                        //      on the new session, swaps it into the
+                        //      long-lived broadcaster, closes the old
+                        //      publisher.
+                        //
+                        // The previous shape closed the old speaker
+                        // synchronously BEFORE openOnce, which forced the
+                        // mic + encoder to stop and re-open across the
+                        // boundary (50–150 ms audible silence at the
+                        // listener). With the close hoisted onto a
+                        // sibling job, the broadcaster keeps capturing
+                        // through the recycle and the listener hears
+                        // continuous audio.
+                        val toClose = speaker
+                        scope.launch {
+                            try {
+                                toClose.close()
+                            } catch (ce: kotlinx.coroutines.CancellationException) {
+                                throw ce
+                            } catch (_: Throwable) {
+                                // Best-effort.
+                            }
                         }
                         attempt = 0
                         refreshTriggered = true
@@ -249,7 +283,14 @@ suspend fun connectReconnectingNestsSpeaker(
         throw NestsException(firstReady.reason, firstReady.cause)
     }
 
-    return ReconnectingSpeakerHandle(state, activeSpeaker, orchestrator, scope)
+    return ReconnectingSpeakerHandle(
+        mutableState = state,
+        activeSpeaker = activeSpeaker,
+        orchestrator = orchestrator,
+        scope = scope,
+        captureFactory = captureFactory,
+        encoderFactory = encoderFactory,
+    )
 }
 
 private fun isUserCancelledSpeaker(state: NestsSpeakerState.Failed): Boolean {
@@ -266,6 +307,17 @@ private class ReconnectingSpeakerHandle(
     private val activeSpeaker: MutableStateFlow<NestsSpeaker?>,
     private val orchestrator: Job,
     private val scope: CoroutineScope,
+    /**
+     * Audio-pipeline factories propagated from
+     * [connectReconnectingNestsSpeaker]. Used by the hot-swap
+     * [ReissuingBroadcastHandle] to construct ONE long-lived
+     * [NestMoqLiteBroadcaster] that survives session recycles —
+     * without these the broadcaster would have to be re-built
+     * (which restarts the mic + encoder + adds a 50–150 ms
+     * audible gap) on every JWT refresh.
+     */
+    private val captureFactory: () -> AudioCapture,
+    private val encoderFactory: () -> OpusEncoder,
 ) : NestsSpeaker {
     override val state: StateFlow<NestsSpeakerState> = mutableState.asStateFlow()
 
@@ -291,14 +343,32 @@ private class ReconnectingSpeakerHandle(
                 ?: error("no live session — wait for state == Connected before startBroadcasting")
 
             val handle =
-                ReissuingBroadcastHandle(activeSpeaker, scope, onLevel) { closed ->
-                    if (activeBroadcast === closed) activeBroadcast = null
-                }
+                ReissuingBroadcastHandle(
+                    activeSpeaker = activeSpeaker,
+                    scope = scope,
+                    captureFactory = captureFactory,
+                    encoderFactory = encoderFactory,
+                    onLevel = onLevel,
+                    onClose = { closed ->
+                        if (activeBroadcast === closed) activeBroadcast = null
+                    },
+                )
             handle.start()
             activeBroadcast = handle
             handle
         }
 
+    /**
+     * Force-close the active inner speaker so the orchestrator opens
+     * a fresh session against the (presumably new) network. Used by
+     * the platform layer on a network change. Mirror of the listener
+     * wrapper's [com.vitorpamplona.nestsclient.NestsListener.recycleSession].
+     */
+    override suspend fun recycleSession() {
+        val current = activeSpeaker.value ?: return
+        runCatching { current.close() }
+    }
+
     override suspend fun close() {
         orchestrator.cancel()
         runCatching { activeBroadcast?.close() }
@@ -310,26 +380,42 @@ private class ReconnectingSpeakerHandle(
 }
 
 /**
- * Stable [BroadcastHandle] backed by a re-issuing pump. Each time
- * the wrapper opens a fresh session the pump cancels its prior
- * iteration, calls [NestsSpeaker.startBroadcasting] on the new
- * session, replays the cached mute intent on the resulting
- * underlying handle, and parks until the next session swap.
+ * Stable [BroadcastHandle] backed by a re-issuing pump. Two paths,
+ * picked per-iteration based on whether the active speaker exposes
+ * the [HotSwappablePublisherSource] hook:
  *
- * `setMuted` updates the cached intent unconditionally and forwards
- * to whichever live underlying handle exists at the time. If no
- * underlying handle is up (e.g. a brief gap during recycle), the
- * intent is replayed on the next handle the pump opens, so the
- * user-observed mute state is monotonic across recycles.
+ *  - **Hot-swap path** (moq-lite): the pump constructs ONE
+ *    long-lived [NestMoqLiteBroadcaster] on the first session
+ *    and, on every subsequent session swap, opens a fresh
+ *    [com.vitorpamplona.nestsclient.moq.lite.MoqLitePublisherHandle]
+ *    on the new session and atomically swaps it into the
+ *    broadcaster via
+ *    [NestMoqLiteBroadcaster.swapPublisher]. The mic +
+ *    encoder + capture loop keep running through the swap, so
+ *    listeners hear ZERO gap at the JWT-refresh boundary
+ *    (vs the legacy ~50–150 ms close-then-restart silence).
+ *
+ *  - **Legacy path** (IETF reference / test fakes): each session
+ *    swap closes the prior `BroadcastHandle` from
+ *    [NestsSpeaker.startBroadcasting] and opens a fresh one. Same
+ *    behaviour as before this hot-swap was introduced.
+ *
+ * `setMuted` updates the cached intent unconditionally; the pump
+ * applies the latest intent to whichever underlying broadcaster /
+ * handle is currently live, so user-observed mute is monotonic
+ * across recycles.
  */
 private class ReissuingBroadcastHandle(
     private val activeSpeaker: StateFlow<NestsSpeaker?>,
     private val scope: CoroutineScope,
+    private val captureFactory: () -> AudioCapture,
+    private val encoderFactory: () -> OpusEncoder,
     /**
-     * Forwarded to every underlying [NestsSpeaker.startBroadcasting]
-     * the pump opens, so the local-speaking ring keeps animating
-     * across session recycles. Mirrors how [desiredMuted] is
-     * replayed — the user-observed signal is monotonic.
+     * Forwarded to the underlying broadcaster (hot-swap path) or
+     * `sp.startBroadcasting` (legacy path) so the local-speaking ring
+     * keeps animating across session recycles. Mirrors how
+     * [desiredMuted] is replayed — the user-observed signal is
+     * monotonic.
      */
     private val onLevel: (Float) -> Unit,
     private val onClose: (ReissuingBroadcastHandle) -> Unit,
@@ -337,17 +423,22 @@ private class ReissuingBroadcastHandle(
     @Volatile private var desiredMuted: Boolean = false
 
     @Volatile private var closed: Boolean = false
+
+    /** Hot-swap path's long-lived broadcaster. Null until the first session arrives. */
+    @Volatile private var hotSwapBroadcaster: NestMoqLiteBroadcaster? = null
+
+    /** Legacy path's per-session handle. Cleared when the session swaps. */
     private val liveHandle = AtomicReference<BroadcastHandle?>(null)
     private var pumpJob: Job? = null
 
     override val isMuted: Boolean get() = desiredMuted
 
     fun start() {
-        // Re-broadcast pump: every time activeSpeaker changes, drop
-        // the prior broadcast (collectLatest cancels the inner
-        // body via awaitCancellation) and open a new one against
-        // the fresh session. The pattern mirrors the listener's
-        // SubscribeHandle re-issuance pump.
+        // Per-iteration pump: every time activeSpeaker changes, decide
+        // whether the new speaker supports hot swap. If yes, retarget
+        // the long-lived broadcaster onto its publisher; if no, fall
+        // back to the close-then-restart path the previous version
+        // used uniformly.
         pumpJob =
             scope.launch {
                 activeSpeaker.collectLatest { sp ->
@@ -367,60 +458,151 @@ private class ReissuingBroadcastHandle(
                         return@collectLatest
                     }
                     if (closed) return@collectLatest
-                    val handle =
-                        try {
-                            sp.startBroadcasting(onLevel)
-                        } catch (ce: kotlinx.coroutines.CancellationException) {
-                            // Don't `return@collectLatest` on cancel —
-                            // propagate so the launched pumpJob actually
-                            // dies on close/scope cancellation. The old
-                            // `runCatching` shape ate the cancel.
-                            throw ce
-                        } catch (_: Throwable) {
-                            null
-                        } ?: return@collectLatest
-                    if (closed) {
-                        runCatching { handle.close() }
-                        return@collectLatest
-                    }
-                    // Apply current mute intent BEFORE storing the
-                    // handle so a setMuted that races us applies
-                    // exactly once: either (a) we set intent →
-                    // apply intent → store, and the racing setMuted
-                    // sees the live handle and applies again (no-op
-                    // on the broadcaster); or (b) the racing
-                    // setMuted updates intent → we read intent →
-                    // apply. Order doesn't matter; idempotent.
-                    if (desiredMuted) {
-                        runCatching { handle.setMuted(true) }
-                    }
-                    liveHandle.set(handle)
-                    try {
-                        // Park until activeSpeaker emits a new value
-                        // (collectLatest cancels us) or close() runs
-                        // (pumpJob.cancel).
-                        awaitCancellation()
-                    } finally {
-                        // Clear our slot only if we still own it —
-                        // close() may have already swapped in null.
-                        if (liveHandle.get() === handle) liveHandle.set(null)
-                        // Best-effort close on the way out: the user
-                        // may have called wrapper.close (closed=true,
-                        // pump cancelling), or activeSpeaker swapped
-                        // (the prior speaker is about to be closed
-                        // by the orchestrator anyway, but defensively
-                        // closing here releases the broadcaster +
-                        // publisher promptly rather than waiting for
-                        // the speaker.close()).
-                        runCatching { handle.close() }
+
+                    val hotSwap = sp as? HotSwappablePublisherSource
+                    if (hotSwap != null) {
+                        runHotSwapIteration(hotSwap)
+                    } else {
+                        runLegacyIteration(sp)
                     }
                 }
             }
     }
 
+    /**
+     * Hot-swap iteration body. On the first session: open a publisher,
+     * construct the long-lived broadcaster, start it. On subsequent
+     * sessions: open a publisher on the new session, swap into the
+     * existing broadcaster, close the OLD publisher (FINs its
+     * announce + group streams on the about-to-die session).
+     *
+     * Parks via [awaitCancellation] so [collectLatest] can cancel us
+     * cleanly when the next session arrives. The broadcaster is NOT
+     * stopped here — it lives across iterations and is only stopped
+     * when the wrapper itself closes.
+     */
+    private suspend fun runHotSwapIteration(hotSwap: HotSwappablePublisherSource) {
+        val newPublisher =
+            try {
+                hotSwap.openPublisherForHotSwap(MoqLiteNestsListener.AUDIO_TRACK)
+            } catch (ce: kotlinx.coroutines.CancellationException) {
+                throw ce
+            } catch (_: Throwable) {
+                // Couldn't mint a publisher on this session (rare —
+                // the session is already past Connected). Bail; the
+                // next session swap will retry.
+                return
+            }
+        if (closed) {
+            runCatching { newPublisher.close() }
+            return
+        }
+
+        val existing = hotSwapBroadcaster
+        if (existing == null) {
+            // First-ever iteration: build the broadcaster and start it.
+            val broadcaster =
+                NestMoqLiteBroadcaster(
+                    capture = captureFactory(),
+                    encoder = encoderFactory(),
+                    initialPublisher = newPublisher,
+                    scope = scope,
+                    framesPerGroup = NestMoqLiteBroadcaster.DEFAULT_FRAMES_PER_GROUP,
+                )
+            try {
+                broadcaster.start(
+                    onTerminalFailure = {
+                        // Bubble up via the speaker's state machine so
+                        // the orchestrator's normal Failed-handling
+                        // path recycles the session. Same shape the
+                        // legacy `startBroadcasting`-internal path uses.
+                        runCatching { hotSwap.reportBroadcastTerminalFailure() }
+                    },
+                    onLevel = onLevel,
+                )
+            } catch (t: Throwable) {
+                runCatching { newPublisher.close() }
+                throw t
+            }
+            // Apply current mute intent (a setMuted that arrived before
+            // the broadcaster existed has already updated desiredMuted).
+            if (desiredMuted) broadcaster.setMuted(true)
+            hotSwapBroadcaster = broadcaster
+        } else {
+            // Subsequent iteration: hot swap. The capture / encoder
+            // pipeline is already running and feeding [existing.publisher]
+            // (the soon-to-be-old one). Install the new publisher,
+            // grab the old, close it. The capture loop's next snapshot
+            // picks up the new publisher and resets its group counter.
+            val old = existing.swapPublisher(newPublisher)
+            // Re-apply mute on the broadcaster — it survives swap, but
+            // a setMuted that arrived during the gap between the last
+            // session's terminal state and this swap may have flipped
+            // [desiredMuted] without finding a live broadcaster (no, it
+            // would have seen [hotSwapBroadcaster] since that's
+            // long-lived; but the no-op on equality keeps this safe).
+            existing.setMuted(desiredMuted)
+            // Close the old publisher AFTER the broadcaster has the
+            // new one. This is the order that matters: if we closed
+            // first, the capture loop's next send would race the swap
+            // and might see the closed publisher.
+            if (old != null) runCatching { old.close() }
+        }
+
+        try {
+            // Park until [collectLatest] cancels us on the next session
+            // swap, OR [close] cancels [pumpJob]. The broadcaster keeps
+            // running through the cancellation; only close() stops it.
+            awaitCancellation()
+        } finally {
+            // Intentionally do NOT close the broadcaster here.
+            // collectLatest cancels this iteration on every session
+            // swap; closing the broadcaster would create the exact
+            // 50–150 ms gap this whole path exists to avoid.
+        }
+    }
+
+    /**
+     * Legacy iteration body — used for IETF reference speakers and any
+     * future [NestsSpeaker] that doesn't implement
+     * [HotSwappablePublisherSource]. Identical to the pre-hot-swap
+     * behaviour: open a fresh handle on each session, close it on
+     * cancellation.
+     */
+    private suspend fun runLegacyIteration(sp: NestsSpeaker) {
+        val handle =
+            try {
+                sp.startBroadcasting(onLevel)
+            } catch (ce: kotlinx.coroutines.CancellationException) {
+                throw ce
+            } catch (_: Throwable) {
+                null
+            } ?: return
+        if (closed) {
+            runCatching { handle.close() }
+            return
+        }
+        if (desiredMuted) {
+            runCatching { handle.setMuted(true) }
+        }
+        liveHandle.set(handle)
+        try {
+            awaitCancellation()
+        } finally {
+            if (liveHandle.get() === handle) liveHandle.set(null)
+            runCatching { handle.close() }
+        }
+    }
+
     override suspend fun setMuted(muted: Boolean) {
         if (closed) return
         desiredMuted = muted
+        // Apply to whichever live underlying exists. Hot-swap and
+        // legacy paths are mutually exclusive in steady state — a
+        // moq-lite session's wrapper never has a live legacy handle,
+        // and vice versa — but both reads are cheap and harmless if
+        // the unused one is null.
+        hotSwapBroadcaster?.setMuted(muted)
         liveHandle.get()?.let { runCatching { it.setMuted(muted) } }
     }
 
@@ -428,6 +610,13 @@ private class ReissuingBroadcastHandle(
         if (closed) return
         closed = true
         pumpJob?.cancel()
+        // Tear down whichever path was active. For hot-swap, stopping
+        // the broadcaster releases the mic + encoder AND closes the
+        // current publisher (broadcaster.stop calls publisher.close
+        // internally). For legacy, the per-session handle's close
+        // releases its own broadcaster.
+        hotSwapBroadcaster?.let { runCatching { it.stop() } }
+        hotSwapBroadcaster = null
         liveHandle.getAndSet(null)?.let { runCatching { it.close() } }
         onClose(this)
     }

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/Audio.kt

@@ -99,11 +99,37 @@ interface AudioCapture {
 /**
  * Sink for PCM audio playback. Implementations buffer internally — [enqueue]
  * may suspend if the device's playback buffer is full.
+ *
+ * **Two-phase startup.** [start] allocates the underlying device + per-instance
+ * resources but does NOT begin consuming samples; [beginPlayback] flips the
+ * device into the playing state. Splitting the two lets [com.vitorpamplona.nestsclient.audio.NestPlayer]
+ * pre-roll a few decoded frames into the device's buffer before playback
+ * starts, so the AudioTrack has slack the moment the hardware begins pulling
+ * samples. Calling [enqueue] between [start] and [beginPlayback] is allowed
+ * and is the intended pattern. Implementations that don't need the
+ * distinction (test fakes, software-only sinks) can leave the default
+ * [beginPlayback] no-op alone.
  */
 interface AudioPlayer {
-    /** Allocate underlying audio resources and begin playback. */
+    /**
+     * Allocate underlying audio resources. The returned device is in a
+     * "ready, not playing" state — [enqueue] is allowed but the hardware
+     * doesn't consume samples until [beginPlayback] is called. Throws on
+     * device-unavailable so callers (typically
+     * [com.vitorpamplona.nestsclient.audio.NestPlayer.play]) get a
+     * synchronous failure they can roll back the subscription on.
+     */
     fun start()
 
+    /**
+     * Transition the allocated device into the playing state. The
+     * hardware begins pulling samples from whatever's already been
+     * [enqueue]d. Default no-op so test fakes / software-only sinks
+     * don't have to grow a method they'll never use; production
+     * Android implementations override to call `AudioTrack.play()`.
+     */
+    fun beginPlayback() {}
+
     /**
      * Feed one PCM frame (any length, but typically [AudioFormat.FRAME_SIZE_SAMPLES]
      * samples) into the playback queue.

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt

@@ -39,7 +39,7 @@ import kotlinx.coroutines.launch
 class NestMoqLiteBroadcaster(
     private val capture: AudioCapture,
     private val encoder: OpusEncoder,
-    private val publisher: MoqLitePublisherHandle,
+    initialPublisher: MoqLitePublisherHandle,
     private val scope: CoroutineScope,
     /**
      * Number of consecutive Opus frames to pack into a single moq-lite
@@ -98,6 +98,26 @@ class NestMoqLiteBroadcaster(
 
     @Volatile private var muted: Boolean = false
 
+    /**
+     * Active publisher the capture loop pushes frames into. Mutable +
+     * `@Volatile` so [swapPublisher] can atomically retarget the loop
+     * onto a fresh moq-lite session's publisher when the
+     * [com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker]
+     * orchestrator recycles the WebTransport session for a JWT
+     * refresh — without restarting capture or the encoder. The
+     * capture loop snapshots this reference once per frame, so a swap
+     * takes effect on the very next frame and any in-flight send on
+     * the previous publisher is allowed to complete (or fail
+     * gracefully, caught by `runCatching`).
+     *
+     * Per-instance group sequence: each publisher restarts at sequence
+     * 0, since the publisher state lives inside [MoqLitePublisherHandle]
+     * (one per moq-lite session). That's fine — listeners use group
+     * sequence for ordering within a single subscription's uni-stream-
+     * per-group flow, not across publisher cycles.
+     */
+    @Volatile private var publisher: MoqLitePublisherHandle = initialPublisher
+
     /**
      * Start capturing + encoding + publishing in the background.
      * Returns immediately. Calling twice is an error. If
@@ -161,6 +181,16 @@ class NestMoqLiteBroadcaster(
                 // we bail. publisher.send returning `false` (no inbound
                 // subscriber) is NOT counted — empty rooms are normal.
                 var consecutiveSendErrors = 0
+                // Track which publisher we last sent to. On swapPublisher
+                // (JWT-refresh hot swap), the snapshot below picks up the
+                // new reference; we reset framesInCurrentGroup so the
+                // first frame on the new publisher starts a fresh group
+                // (the new publisher's group sequence is 0 anyway). Without
+                // this, a mid-group swap would feed the new publisher
+                // frames as if they continued the old publisher's group,
+                // and the relay would see two unrelated uni streams under
+                // the same logical group.
+                var lastPublisher: MoqLitePublisherHandle = publisher
                 try {
                     while (true) {
                         val pcm = capture.readFrame() ?: break
@@ -181,16 +211,28 @@ class NestMoqLiteBroadcaster(
                             }
                         if (opus.isEmpty()) continue
                         if (muted) continue
+                        // Snapshot the publisher reference once per frame.
+                        // If [swapPublisher] mid-loop installed a new
+                        // reference, pick it up here and reset the group
+                        // counter so the new publisher's first frame
+                        // starts a clean group. Frames already in flight
+                        // on the previous publisher are allowed to
+                        // complete (or fail gracefully on `runCatching`).
+                        val current = publisher
+                        if (current !== lastPublisher) {
+                            framesInCurrentGroup = 0
+                            lastPublisher = current
+                        }
                         // Pack [framesPerGroup] consecutive Opus frames
                         // into the same moq-lite group / QUIC uni stream
                         // before FINning. See the [framesPerGroup] kdoc
                         // for the production cliff this works around.
                         val sendOutcome =
                             runCatching {
-                                publisher.send(opus)
+                                current.send(opus)
                                 framesInCurrentGroup += 1
                                 if (framesInCurrentGroup >= framesPerGroup) {
-                                    publisher.endGroup()
+                                    current.endGroup()
                                     framesInCurrentGroup = 0
                                 }
                             }
@@ -235,11 +277,14 @@ class NestMoqLiteBroadcaster(
                             }
                     }
                     // EOF on the capture side. Flush whatever's in the
-                    // open group so the relay sees its FIN and the
-                    // listener doesn't sit on a half-delivered group
-                    // buffer. Mirrors the per-group endGroup above.
+                    // open group on the most-recently-used publisher so
+                    // the relay sees its FIN and the listener doesn't
+                    // sit on a half-delivered group buffer. Use
+                    // [lastPublisher] (not the live [publisher] field)
+                    // because a swap-then-EOF would otherwise FIN a
+                    // group on a publisher that didn't open it.
                     if (framesInCurrentGroup > 0) {
-                        runCatching { publisher.endGroup() }
+                        runCatching { lastPublisher.endGroup() }
                     }
                 } catch (ce: CancellationException) {
                     throw ce
@@ -264,6 +309,38 @@ class NestMoqLiteBroadcaster(
         this.muted = muted
     }
 
+    /**
+     * Hot-swap the underlying publisher. Used by
+     * [com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker]'s
+     * orchestrator to retarget the broadcaster onto a fresh moq-lite
+     * session's publisher when the JWT-refresh window fires — without
+     * tearing down the AudioRecord / Opus encoder. Returns the previous
+     * publisher reference so the caller can close it after the new one
+     * is wired up.
+     *
+     * **Closing the returned old publisher** is the caller's
+     * responsibility — this method only swaps the reference so the
+     * capture loop's next snapshot picks up the new publisher. The
+     * caller typically does:
+     *
+     *   1. Open a new moq-lite session, mint a new publisher on it.
+     *   2. Call [swapPublisher] with the new publisher; cache the old.
+     *   3. Close the old publisher (FINs the announce bidi + current
+     *      group's uni stream on the old session).
+     *   4. Close the old moq-lite session (drops the WebTransport).
+     *
+     * No-op if the broadcaster is already [stop]ped (returns null) — the
+     * capture loop is already gone, so swapping in a new publisher would
+     * just leak it.
+     */
+    fun swapPublisher(newPublisher: MoqLitePublisherHandle): MoqLitePublisherHandle? {
+        if (stopped) return null
+        val old = publisher
+        if (old === newPublisher) return null
+        publisher = newPublisher
+        return old
+    }
+
     /**
      * Stop the loop, release the mic, release the encoder, close the
      * moq-lite publisher (which sends `Announce(Ended)` on every active

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayer.kt

@@ -45,7 +45,27 @@ class NestPlayer(
     private val decoder: OpusDecoder,
     private val player: AudioPlayer,
     private val scope: CoroutineScope,
+    /**
+     * Number of decoded PCM frames to buffer before starting the underlying
+     * [AudioPlayer]. Without pre-roll, the device begins consuming audio the
+     * instant the first frame arrives and underruns at the first decode that
+     * misses its 20 ms cadence — the most common cause of perceptible
+     * dropouts on devices where Compose recomposition or GC briefly stalls
+     * the decode loop.
+     *
+     * Production callers typically pass `5` (≈ 100 ms) — long enough to mask
+     * a typical Main-thread hiccup, short enough that listeners don't notice
+     * the join latency. Tests pass `0` to preserve the synchronous test
+     * scheduler behaviour the existing assertions rely on.
+     *
+     * Default is `0` so existing tests stand without modification.
+     */
+    private val prerollFrames: Int = 0,
 ) {
+    init {
+        require(prerollFrames >= 0) { "prerollFrames must be >= 0, got $prerollFrames" }
+    }
+
     private var job: Job? = null
     private var stopped = false
 
@@ -73,9 +93,56 @@ class NestPlayer(
         check(!stopped) { "NestPlayer already stopped" }
         check(job == null) { "NestPlayer.play already called" }
 
+        // Allocate the device synchronously so a [AudioException.DeviceUnavailable]
+        // (audio policy denial, AudioTrack rejected, etc.) propagates to
+        // the caller — typically `NestViewModel.openSubscription`, which
+        // catches and rolls back the freshly-reserved subscription slot.
+        // Routing this failure through `onError` instead would attach the
+        // slot first and leave a permanent "Connecting…" spinner on the
+        // speaker tile when the device fails to allocate.
+        //
+        // Two-phase startup: [AudioPlayer.start] allocates without
+        // beginning playback; [AudioPlayer.beginPlayback] flips the
+        // device into the playing state. We delay [beginPlayback] until
+        // the pre-roll buffer is full so the first frames already
+        // populate the device's internal buffer when the hardware
+        // starts pulling samples.
         player.start()
+
+        // Pre-roll buffer holds decoded PCM until either [prerollFrames]
+        // frames have arrived or the upstream flow ends. Once the
+        // threshold is met (or the flow ends with anything queued), we
+        // call [AudioPlayer.beginPlayback] and flush the buffer in a
+        // tight loop so the device starts playback with a populated
+        // buffer. A flow that never produces PCM (decoder always empty,
+        // no audio in the room) never calls [beginPlayback] — the
+        // allocated device sits in the "ready, not playing" state until
+        // [stop] tears it down.
         job =
             scope.launch {
+                val preroll = ArrayDeque<ShortArray>(prerollFrames.coerceAtLeast(1))
+                var playbackBegun = false
+
+                suspend fun beginAndFlushIfNeeded() {
+                    if (playbackBegun) return
+                    if (preroll.isEmpty()) return
+                    // Flush BEFORE [beginPlayback] so the device starts
+                    // playing against an already-populated buffer rather
+                    // than emitting silence for the microseconds it
+                    // takes the flush loop to fill. AudioTrack
+                    // MODE_STREAM explicitly supports write() before
+                    // play() per the Android docs — that's the
+                    // textbook pre-roll pattern. Order matters
+                    // because [beginPlayback] is the moment the
+                    // hardware starts pulling samples; getting
+                    // [enqueue] in first means the very first sample
+                    // pulled is from our pre-rolled audio, not silence.
+                    while (preroll.isNotEmpty()) {
+                        player.enqueue(preroll.removeFirst())
+                    }
+                    player.beginPlayback()
+                    playbackBegun = true
+                }
                 try {
                     objects.collect { obj ->
                         val pcm =
@@ -95,9 +162,21 @@ class NestPlayer(
                             }
                         if (pcm.isNotEmpty()) {
                             onLevel(peakAmplitude(pcm))
-                            player.enqueue(pcm)
+                            if (playbackBegun) {
+                                player.enqueue(pcm)
+                            } else {
+                                preroll.addLast(pcm)
+                                if (preroll.size >= prerollFrames) {
+                                    beginAndFlushIfNeeded()
+                                }
+                            }
                         }
                     }
+                    // Flow ended without enough frames to fill the pre-roll
+                    // (e.g. the publisher cycled before pre-roll was full,
+                    // or the room ended). Flush whatever's queued so any
+                    // already-decoded audio still reaches the device.
+                    beginAndFlushIfNeeded()
                 } catch (ce: CancellationException) {
                     throw ce
                 } catch (t: Throwable) {

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt

@@ -202,8 +202,6 @@ class MoqLiteSession internal constructor(
                 endGroup = endGroup,
             )
         val bidi = transport.openBidiStream()
-        bidi.write(Varint.encode(MoqLiteControlType.Subscribe.code))
-        bidi.write(MoqLiteCodec.encodeSubscribe(request))
 
         // Single long-running collector for the bidi's whole lifetime.
         // The collector parses the SubscribeResponse inline, signals it
@@ -228,16 +226,31 @@ class MoqLiteSession internal constructor(
         val frames = Channel<MoqLiteFrame>(capacity = DEFAULT_FRAME_BUFFER, onBufferOverflow = BufferOverflow.DROP_OLDEST)
         val responseDeferred = CompletableDeferred<MoqLiteCodec.SubscribeResponse>()
 
-        // Pre-register the subscription BEFORE launching the collector.
-        // Otherwise: if the publisher FINs the bidi immediately after
-        // sending Ok, the collector's exit-cleanup races subscribe()'s
-        // post-await registration — collector exits, runs `remove(id)`
-        // against an empty map, no-ops; subscribe() then inserts; the
-        // frames channel is now in the map with no live collector to
-        // close it on transport tear-down. Consumer hangs forever.
-        // Pre-registering means the collector's idempotent
-        // remove+close cleanup always finds and closes the frames
-        // channel, regardless of timing.
+        // Pre-register the subscription BEFORE launching the collector
+        // AND before the SUBSCRIBE goes on the wire. The order matters
+        // for two reasons:
+        //
+        //   1. Collector race (original reason): if the publisher FINs
+        //      the bidi immediately after sending Ok, the collector's
+        //      exit-cleanup races subscribe()'s post-await registration
+        //      — collector exits, runs `remove(id)` against an empty
+        //      map, no-ops; subscribe() then inserts; the frames
+        //      channel is now in the map with no live collector to
+        //      close it on transport tear-down. Consumer hangs
+        //      forever. Pre-registering means the collector's
+        //      idempotent remove+close cleanup always finds and
+        //      closes the frames channel, regardless of timing.
+        //
+        //   2. First-group race (audit fix #3): the relay can open
+        //      the first group's uni stream BEFORE our subscribe()
+        //      continuation re-enters [state] to register `id`. If
+        //      the SUBSCRIBE bytes hit the wire before the map entry
+        //      lands, [drainOneGroup] looks the id up against an
+        //      empty map and silently drops the frame. Registering
+        //      the id before writing the SUBSCRIBE closes the
+        //      window — by the time the relay can have parsed the
+        //      message and opened a uni stream in response, the map
+        //      already has our entry.
         val sub =
             ListenerSubscription(
                 id = id,
@@ -249,18 +262,44 @@ class MoqLiteSession internal constructor(
             subscriptionsBySubscribeId[id] = sub
             if (groupPump == null) groupPump = scope.launch { pumpUniStreams() }
         }
+        // Now that the subscription is registered, push the SUBSCRIBE
+        // bytes. If `bidi.write` throws (transport torn down, peer
+        // reset) we'd otherwise leave an orphaned map entry whose
+        // frames channel never closes — the response collector hasn't
+        // been launched yet so its idempotent cleanup wouldn't run.
+        // Roll back explicitly on throw.
+        try {
+            bidi.write(Varint.encode(MoqLiteControlType.Subscribe.code))
+            bidi.write(MoqLiteCodec.encodeSubscribe(request))
+        } catch (t: Throwable) {
+            state.withLock { subscriptionsBySubscribeId.remove(id) }
+            frames.close()
+            runCatching { bidi.finish() }
+            throw t
+        }
 
         scope.launch {
             val responseBuffer = MoqLiteFrameBuffer()
             var responseParsed = false
+            // typeCode hoisted outside the collect lambda — `readVarint`
+            // advances `pos` permanently while `readSizePrefixed` rolls
+            // its own length-varint back on incomplete-body but does NOT
+            // roll back the type-code that's already been consumed. If
+            // the response chunks split between type and body (possible
+            // under MTU pressure / fragmentation), a per-tick `val
+            // typeCode = readVarint()` would read the body's size
+            // prefix as the type code on the next chunk and misframe
+            // the response. Mirrors the same fix in [handleInboundBidi].
+            var typeCode: Long? = null
             try {
                 bidi.incoming().collect { chunk ->
                     if (!responseParsed) {
                         responseBuffer.push(chunk)
-                        val typeCode = responseBuffer.readVarint() ?: return@collect
+                        if (typeCode == null) typeCode = responseBuffer.readVarint()
+                        val tc = typeCode ?: return@collect
                         val body = responseBuffer.readSizePrefixed() ?: return@collect
                         val out = MoqWriter()
-                        out.writeVarint(typeCode)
+                        out.writeVarint(tc)
                         out.writeVarint(body.size.toLong())
                         out.writeBytes(body)
                         val resp = MoqLiteCodec.decodeSubscribeResponse(out.toByteArray())

nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayerTest.kt

@@ -216,6 +216,130 @@ class NestPlayerTest {
             sut.stop()
         }
 
+    /**
+     * Pre-roll: with `prerollFrames=3`, [AudioPlayer.beginPlayback]
+     * must NOT fire until the third decoded frame has arrived, and
+     * once it fires the AudioPlayer's queue must already contain
+     * the buffered pre-roll (i.e. flushed atomically with playback
+     * start, not lazily on the next enqueue).
+     */
+    @Test
+    fun preroll_defers_beginPlayback_until_threshold_is_met() =
+        runTest {
+            val channel = Channel<MoqObject>(capacity = 8)
+            val decoder = FakeOpusDecoder { byteToShorts(it) }
+            val player = FakeAudioPlayer()
+
+            val sut =
+                NestPlayer(
+                    decoder = decoder,
+                    player = player,
+                    scope = this,
+                    prerollFrames = 3,
+                )
+            sut.play(channel.receiveAsFlow())
+            testScheduler.runCurrent()
+
+            // Frames 1 and 2: pre-roll buffer fills, beginPlayback NOT
+            // called yet, AudioPlayer hasn't seen a single enqueue.
+            channel.send(moqObject(byteArrayOf(0x01)))
+            channel.send(moqObject(byteArrayOf(0x02)))
+            testScheduler.advanceUntilIdle()
+            assertEquals(0, player.beginPlaybackCount, "beginPlayback before threshold")
+            assertEquals(0, player.queued.size, "no enqueue before threshold")
+
+            // Frame 3 trips the threshold: beginPlayback fires exactly
+            // once, and the pre-rolled frames must ALREADY be sitting
+            // in the AudioPlayer queue at that moment (otherwise the
+            // device starts playback against an empty buffer and
+            // pre-roll's whole point is defeated). [NestPlayer.play]
+            // implements this by flushing-then-beginPlayback —
+            // AudioTrack MODE_STREAM explicitly supports write()
+            // before play().
+            channel.send(moqObject(byteArrayOf(0x03)))
+            testScheduler.advanceUntilIdle()
+            assertEquals(1, player.beginPlaybackCount)
+            // The fake records the queue size at the moment
+            // beginPlayback fires. With flush-then-begin ordering,
+            // all 3 pre-rolled frames are already in the queue.
+            assertEquals(3, player.queuedAtBeginPlayback, "pre-roll flushed before beginPlayback")
+            assertEquals(3, player.queued.size, "buffer populated when device starts")
+
+            // Subsequent frames bypass the buffer and go directly
+            // through enqueue.
+            channel.send(moqObject(byteArrayOf(0x04)))
+            testScheduler.advanceUntilIdle()
+            assertEquals(1, player.beginPlaybackCount, "beginPlayback only fires once")
+            assertEquals(4, player.queued.size)
+
+            sut.stop()
+        }
+
+    /**
+     * Pre-roll: a flow that ends BEFORE the pre-roll threshold fires
+     * must still flush its partial buffer to the AudioPlayer.
+     * Otherwise a fast-cycling publisher could leave already-decoded
+     * frames stranded forever.
+     */
+    @Test
+    fun preroll_flushes_partial_buffer_when_flow_ends_early() =
+        runTest {
+            val decoder = FakeOpusDecoder { byteToShorts(it) }
+            val player = FakeAudioPlayer()
+
+            val sut =
+                NestPlayer(
+                    decoder = decoder,
+                    player = player,
+                    scope = this,
+                    prerollFrames = 5,
+                )
+            // Only 2 frames — pre-roll never reaches its 5-frame floor,
+            // but the upstream Flow ends so the loop's flush hook must
+            // begin playback and drain whatever's queued.
+            sut.play(
+                flowOf(
+                    moqObject(byteArrayOf(0x01)),
+                    moqObject(byteArrayOf(0x02)),
+                ),
+            )
+            testScheduler.advanceUntilIdle()
+
+            assertEquals(1, player.beginPlaybackCount)
+            assertEquals(2, player.queued.size)
+            assertContentEquals(byteToShorts(byteArrayOf(0x01)), player.queued[0])
+            assertContentEquals(byteToShorts(byteArrayOf(0x02)), player.queued[1])
+
+            sut.stop()
+        }
+
+    /**
+     * Pre-roll edge case: an empty flow shouldn't start playback at
+     * all. The AudioTrack stays in its allocated-but-not-playing
+     * state until [stop] tears it down.
+     */
+    @Test
+    fun preroll_does_not_begin_playback_when_flow_emits_no_pcm() =
+        runTest {
+            val decoder = FakeOpusDecoder { ShortArray(0) }
+            val player = FakeAudioPlayer()
+
+            val sut =
+                NestPlayer(
+                    decoder = decoder,
+                    player = player,
+                    scope = this,
+                    prerollFrames = 3,
+                )
+            sut.play(flowOf(moqObject(byteArrayOf(0x01)), moqObject(byteArrayOf(0x02))))
+            testScheduler.advanceUntilIdle()
+
+            assertEquals(0, player.beginPlaybackCount, "no PCM → no playback")
+            assertEquals(0, player.queued.size)
+
+            sut.stop()
+        }
+
     // -- helpers -----------------------------------------------------------
 
     private fun moqObject(payload: ByteArray): MoqObject =
@@ -246,6 +370,26 @@ class NestPlayerTest {
     private class FakeAudioPlayer : AudioPlayer {
         var started = false
             private set
+
+        /**
+         * Counts the [AudioPlayer.beginPlayback] invocations. Used by
+         * the pre-roll regression tests to assert that playback only
+         * begins AFTER `prerollFrames` decoded frames have arrived (or
+         * after the upstream flow ends with a partial buffer). The
+         * default no-op `beginPlayback` in the interface lets fakes
+         * skip overriding when they don't care; we override here so
+         * the tests can verify the pre-roll wiring.
+         *
+         * Also tracks the size of `queued` at the moment beginPlayback
+         * fired — pre-roll's contract is that the buffer is flushed
+         * IN A TIGHT LOOP after beginPlayback, so we can read the
+         * snapshot to verify the flush ordering.
+         */
+        var beginPlaybackCount = 0
+            private set
+        var queuedAtBeginPlayback: Int = -1
+            private set
+
         var stopCount = 0
             private set
         val stopped: Boolean get() = stopCount > 0
@@ -257,6 +401,11 @@ class NestPlayerTest {
             started = true
         }
 
+        override fun beginPlayback() {
+            beginPlaybackCount++
+            queuedAtBeginPlayback = queued.size
+        }
+
         override suspend fun enqueue(pcm: ShortArray) {
             queued.add(pcm)
         }