Merge pull request #2621 from vitorpamplona/claude/review-nostr-nests-compliance-hKBnS

feat(nests): proactive JWT refresh + reconnect for speaker path

nestsClient/plans/2026-04-28-listener-survives-publisher-recycle.md

@@ -0,0 +1,88 @@
+# Listener-survives-publisher-recycle: resolution log
+
+**Status:** ✅ All three reconnecting-listener interop scenarios
+pass against the real moq-rs relay: happy-path, session-swap,
+and listener-survives-publisher-recycle. Two further fixes landed
+on top of `851045c6` to close the session-swap gap; see "Round 2"
+below.
+
+## What was broken (recap)
+
+When a remote speaker JWT-refreshed (every 9 min via
+`connectReconnectingNestsSpeaker`), any listener with a vanilla
+`SubscribeHandle` open against that speaker's broadcast went silent.
+The listener's session stayed Connected; the listener's
+`MoqLiteSubscribeHandle.frames` `Channel.consumeAsFlow()` just sat
+open waiting for frames that never arrived. moq-lite Lite-03 has no
+explicit "publisher gone" message on the subscribe bidi (the relay
+keeps that bidi open across publisher cycles in case a fresh
+publisher takes over the suffix), so the announce stream's
+`Announce(Ended)` event is the only reliable signal.
+
+## What landed (commit `851045c6`)
+
+**Session layer** (`MoqLiteSession.kt`):
+
+- Lazy single shared announce-watch pump per session, opened on
+  first subscribe.
+- On `Announce(Ended)` for a broadcast suffix, close the matching
+  `ListenerSubscription`'s frames `Channel` and remove it from the
+  map.
+
+**Wrapper layer** (`ReconnectingNestsListener.kt`):
+
+- Inner `while (currentCoroutineContext().isActive)` loop in
+  `reissuingSubscribe`. When the underlying frames flow completes
+  (signalled by the session layer), re-issue subscribe against the
+  same listener with a 100 ms backoff. moq-lite supports
+  subscribe-before-announce so the new subscribe attaches cleanly
+  when the next publisher comes up under the same suffix.
+
+**Verified against the real moq-rs relay** (host build, external
+mode): the new
+`NostrNestsReconnectingListenerInteropTest.subscribe_handle_survives_publisher_recycle`
+test passes — single SubscribeHandle keeps emitting frames across
+multiple speaker JWT-refresh cycles. Speaker reconnect tests still
+pass too.
+
+## Round 2 — closing the session-swap gap (commit `d8ab4fd9`)
+
+The `reconnecting_wrapper_keeps_handle_alive_across_session_swap`
+test exposed two compounding issues that round 1 didn't address:
+
+1. **Orchestrator break-on-Closed** — `if (terminal is Closed) break`
+   in `ReconnectingNestsListener.kt` exited whenever the inner
+   listener went Closed for ANY reason, including the test's own
+   `firstListener.close()`. Removed the break: user-driven
+   `reconnecting.close()` already cancels the orchestrator
+   coroutine separately, so any other Closed (peer transport drop,
+   recycle) is now a reconnect trigger.
+
+2. **Publisher single-group architecture** —
+   `NestMoqLiteBroadcaster` only ever called `publisher.send(opus)`,
+   never `endGroup()`. The entire broadcast was one giant moq-lite
+   group; a subscriber that joined mid-broadcast got nothing
+   because `from-latest` subscribe semantics give the NEXT group's
+   frames, and the publisher was in a never-ending group. Fixed by
+   adding `publisher.endGroup()` after each send — one Opus frame
+   per moq-lite group, mirroring the kixelated reference's audio
+   publish path.
+
+Three companion changes in `MoqLiteSession.kt` were needed to make
+those work cleanly:
+
+- `ensureAnnounceWatchStarted()` runs synchronously before the
+  first subscribe, so the relay sees us as an audience member
+  before we ask to subscribe (otherwise it returns "not found").
+- `handleInboundBidi` refactored to a single long-running
+  collector with the varint `typeCode` hoisted outside `collect`
+  (an earlier draft re-read the body bytes as the type code on the
+  second collect pass).
+- `removeInboundSubscription(sub)` FINs the publisher's
+  `currentGroup` when an inbound subscribe bidi closes, so the next
+  send opens a fresh uni stream keyed off a live subscriber rather
+  than the dead one that was first in the inboundSubs set.
+
+The round-trip interop test's groupId assertion was updated from
+`groupId == 0` to `groupId == idx` to match the new
+one-group-per-frame contract.

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

@@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
@@ -39,6 +40,7 @@ import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withTimeoutOrNull
 import java.util.concurrent.atomic.AtomicReference
@@ -169,9 +171,20 @@ suspend fun connectReconnectingNestsListener(
                     // never enters Reconnecting.
                     continue
                 }
-                val terminal = state.value
-                if (terminal is NestsListenerState.Closed) break
+                // Note: we do NOT break on terminal=Closed. The
+                // user-driven stop path goes through
+                // [ReconnectingHandle.close], which calls
+                // `orchestrator.cancel()` BEFORE closing the inner
+                // listener; cancellation propagates through the
+                // next suspending call (typically `delay` below or
+                // `openOnce` on the next loop iteration) and the
+                // orchestrator exits cleanly. Any *other* path that
+                // produces a Closed inner listener — peer-driven
+                // transport close, half-broken session that was
+                // closed by some internal cleanup — should be
+                // treated as an unexpected drop and reconnected.
                 if (policy.isExhausted(attempt + 1)) break
+                val terminal = state.value
                 val delayMs =
                     if (terminal is NestsListenerState.Reconnecting) {
                         terminal.delayMs
@@ -263,9 +276,38 @@ private class ReconnectingHandle(
             )
         val liveHandleRef = AtomicReference<SubscribeHandle?>(null)
 
-        // Re-subscribe pump: every time activeListener changes, drop
-        // the prior subscription (collectLatest cancels the inner
-        // body) and open a new one against the fresh session.
+        // Re-subscribe pump. Two re-issue triggers, layered:
+        //
+        //   1. Listener session swap (outer collectLatest) — fires
+        //      when the orchestrator opens a fresh listener after
+        //      the 540 s JWT-refresh window or a transport-loss
+        //      reconnect. collectLatest cancels the prior pump
+        //      iteration so the next iteration runs against the
+        //      new listener.
+        //
+        //   2. Publisher session swap (inner while loop) — fires
+        //      when the underlying SubscribeHandle.objects flow
+        //      completes mid-stream because the *publisher*
+        //      cycled. The moq-lite session layer detects publisher
+        //      disconnect via the announce stream's Ended event
+        //      and closes the underlying frames channel; that
+        //      naturally ends `handle.objects.collect` here. We
+        //      then loop into a fresh subscribe — moq-lite supports
+        //      subscribe-before-announce, so the new subscribe
+        //      attaches cleanly to whichever publisher serves the
+        //      suffix next, including one that comes up AFTER us.
+        //
+        // Without the inner loop, a remote speaker's JWT refresh
+        // (every 9 min on the speaker side via
+        // [connectReconnectingNestsSpeaker]) would silently kill
+        // every listener's audio — the listener's own JWT refresh
+        // fires on a different cadence and can't be relied on to
+        // coincide.
+        //
+        // Bounded by:
+        //   - listener swap → outer collectLatest cancels us
+        //   - unsubscribeAction → pumpJob.cancel()
+        //   - opener-throws → break + wait for next swap
         val pumpJob =
             scope.launch {
                 activeListener.collectLatest { listener ->
@@ -277,14 +319,23 @@ private class ReconnectingHandle(
                                 state is NestsListenerState.Failed
                         }
                     if (terminalOrConnected !is NestsListenerState.Connected) return@collectLatest
-                    val handle =
-                        runCatching { opener(listener) }
-                            .getOrNull() ?: return@collectLatest
-                    liveHandleRef.set(handle)
-                    try {
-                        handle.objects.collect { frames.emit(it) }
-                    } finally {
-                        if (liveHandleRef.get() === handle) liveHandleRef.set(null)
+
+                    while (currentCoroutineContext().isActive) {
+                        val handle =
+                            runCatching { opener(listener) }
+                                .getOrNull() ?: break
+                        liveHandleRef.set(handle)
+                        try {
+                            handle.objects.collect { frames.emit(it) }
+                        } finally {
+                            if (liveHandleRef.get() === handle) liveHandleRef.set(null)
+                        }
+                        // Brief backoff so a permanently-gone
+                        // publisher doesn't tight-loop the relay
+                        // with re-subscribes. 100 ms stays well
+                        // under the SUBSCRIBE_BUFFER's 1.3 s of
+                        // audio headroom.
+                        delay(RESUBSCRIBE_BACKOFF_MS)
                     }
                 }
             }
@@ -318,6 +369,12 @@ private class ReconnectingHandle(
         // grow the queue unbounded.
         private const val SUBSCRIBE_BUFFER = 64
 
+        // Inner-pump backoff between publisher-cycle re-subscribes.
+        // Short enough to stay well under the SUBSCRIBE_BUFFER's
+        // ~1.3 s of audio headroom; long enough that a permanently-
+        // gone publisher doesn't spin the relay with re-subscribes.
+        private const val RESUBSCRIBE_BACKOFF_MS = 100L
+
         private val SYNTH_OK =
             SubscribeOk(
                 subscribeId = -1L,

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

@@ -0,0 +1,405 @@
+/*
+ * 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.nestsclient
+
+import com.vitorpamplona.nestsclient.audio.AudioCapture
+import com.vitorpamplona.nestsclient.audio.OpusEncoder
+import com.vitorpamplona.nestsclient.transport.WebTransportFactory
+import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.awaitCancellation
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.flow.collectLatest
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withTimeoutOrNull
+import java.util.concurrent.atomic.AtomicReference
+
+/**
+ * `connectNestsSpeaker` plus a transport-loss reconnect loop with
+ * exponential backoff and proactive JWT refresh. Mirror of
+ * [connectReconnectingNestsListener] for the publish side.
+ *
+ * The returned [NestsSpeaker]'s state surfaces the underlying speaker's
+ * state directly while a session is alive, but flips to
+ * [NestsSpeakerState.Reconnecting] between attempts. The speaker is
+ * auto-redirected to the freshly-opened session under the hood —
+ * `startBroadcasting()` returns a stable [BroadcastHandle] whose
+ * `setMuted` and `close` survive every refresh.
+ *
+ * **Why this exists** — moq-auth issues 600 s bearer tokens
+ * (`moq-auth/src/index.ts`). Without proactive refresh, any room a
+ * user keeps the stage in for >10 min hits an authorisation failure
+ * the moment the relay tears down the session, the publish stream
+ * goes silent, and the user has to manually re-tap "Talk". The
+ * proactive recycle keeps the WebTransport session young so the
+ * relay never sees an expired token.
+ *
+ * **Broadcast-handle re-issuance** — the caller-owned
+ * [BroadcastHandle] survives a refresh / reconnect. Internally the
+ * wrapper opens a fresh underlying `BroadcastHandle` against each
+ * new session, replays the user's mute intent on it, and forwards
+ * `setMuted` calls to whichever live handle exists at the time.
+ * `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.
+ *
+ * Cancellation: cancelling [scope] (typically the room screen's VM
+ * scope) cancels the reconnect loop and closes both the active
+ * session and the active broadcast. [NestsSpeaker.close] is
+ * idempotent.
+ */
+suspend fun connectReconnectingNestsSpeaker(
+    httpClient: NestsClient,
+    transport: WebTransportFactory,
+    scope: CoroutineScope,
+    room: NestsRoomConfig,
+    signer: NostrSigner,
+    speakerPubkeyHex: String,
+    captureFactory: () -> AudioCapture,
+    encoderFactory: () -> OpusEncoder,
+    policy: NestsReconnectPolicy = NestsReconnectPolicy(),
+    /**
+     * Proactive JWT refresh window. moq-auth issues bearer tokens
+     * with a 600 s lifetime; once the token expires the relay
+     * tears down the WebTransport session and we'd otherwise
+     * recover via the regular reconnect path with a brief audible
+     * dropout AND a permanent broadcast loss until the user taps
+     * Talk again. By recycling the session a minute before expiry
+     * we stay ahead of the relay's tear-down: the new session
+     * opens, the broadcast pump reopens publishing on it (carrying
+     * the user's mute intent), and the wrapper's outward state
+     * never enters the user-visible Reconnecting state.
+     *
+     * Set to 0 or negative to disable.
+     */
+    tokenRefreshAfterMs: Long = 540_000L,
+    /**
+     * Test seam — defaults to the production [connectNestsSpeaker].
+     * Tests pass a fake that returns a scripted [NestsSpeaker] so
+     * the reconnect state machine can be exercised without a real
+     * WebTransport stack.
+     */
+    connector: suspend () -> NestsSpeaker = {
+        connectNestsSpeaker(
+            httpClient = httpClient,
+            transport = transport,
+            scope = scope,
+            room = room,
+            signer = signer,
+            speakerPubkeyHex = speakerPubkeyHex,
+            captureFactory = captureFactory,
+            encoderFactory = encoderFactory,
+        )
+    },
+): NestsSpeaker {
+    val state = MutableStateFlow<NestsSpeakerState>(NestsSpeakerState.Idle)
+    val activeSpeaker = MutableStateFlow<NestsSpeaker?>(null)
+
+    suspend fun openOnce(): NestsSpeaker {
+        val speaker = connector()
+        activeSpeaker.value = speaker
+        state.value = speaker.state.value
+        return speaker
+    }
+
+    val orchestrator =
+        scope.launch {
+            var attempt = 0
+            while (true) {
+                val speaker =
+                    runCatching { openOnce() }.getOrElse {
+                        state.value = NestsSpeakerState.Failed("connect failed: ${it.message}", it)
+                        null
+                    }
+                var refreshTriggered = false
+                if (speaker != null) {
+                    // Wait for either a terminal state OR the proactive
+                    // JWT-refresh deadline. withTimeoutOrNull returns
+                    // null when the timer fires first; we then close
+                    // the (still-healthy) speaker and loop to mint a
+                    // fresh JWT via openOnce(). The broadcast-pump
+                    // re-issues publishing onto the new session
+                    // without the wrapper's outward state ever
+                    // showing Reconnecting.
+                    //
+                    // The `onEach { mirror } + first` pattern (rather
+                    // than `state.collect { mirror; if terminal break }`)
+                    // is what lets `withTimeoutOrNull` cancel cleanly
+                    // mid-mirror — once cancelled, the underlying
+                    // speaker's subsequent state changes (e.g. the
+                    // Closed we trigger on the next line) don't leak
+                    // out to the wrapper's state.
+                    val terminalAwait: suspend () -> NestsSpeakerState = {
+                        speaker.state
+                            .onEach { state.value = it }
+                            .first { s ->
+                                s is NestsSpeakerState.Failed || s is NestsSpeakerState.Closed
+                            }
+                    }
+                    val terminal =
+                        if (tokenRefreshAfterMs > 0L) {
+                            withTimeoutOrNull(tokenRefreshAfterMs) { terminalAwait() }
+                        } else {
+                            terminalAwait()
+                        }
+                    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.
+                        runCatching { speaker.close() }
+                        attempt = 0
+                        refreshTriggered = true
+                    } else if (terminal is NestsSpeakerState.Failed && !isUserCancelledSpeaker(terminal)) {
+                        // Transport-side failure → schedule a reconnect.
+                        attempt++
+                        if (!policy.isExhausted(attempt)) {
+                            val delayMs = policy.delayForAttempt(attempt)
+                            state.value = NestsSpeakerState.Reconnecting(attempt, delayMs)
+                        }
+                    }
+                }
+                if (refreshTriggered) {
+                    // Skip the reconnect-schedule path entirely — a
+                    // refresh is a planned cutover, not a backoff event.
+                    continue
+                }
+                val terminal = state.value
+                if (terminal is NestsSpeakerState.Closed) break
+                if (policy.isExhausted(attempt + 1)) break
+                val delayMs =
+                    if (terminal is NestsSpeakerState.Reconnecting) {
+                        terminal.delayMs
+                    } else {
+                        policy.delayForAttempt(++attempt)
+                    }
+                state.value = NestsSpeakerState.Reconnecting(attempt.coerceAtLeast(1), delayMs)
+                delay(delayMs)
+            }
+        }
+
+    // Match the existing [connectNestsSpeaker] semantics: suspend
+    // until the first session is up (or hard-fails) so the VM's
+    // call site `val s = speakerConnector.connect(...); s.startBroadcasting()`
+    // keeps working without changes. This is a deliberate departure
+    // from the listener wrapper, which returns immediately and
+    // expects the VM to gate `subscribeSpeaker` on the Connected
+    // state — the speaker side has a tighter `startBroadcasting`
+    // contract that requires a live session at call time.
+    val firstReady =
+        state.first { s ->
+            s is NestsSpeakerState.Connected ||
+                s is NestsSpeakerState.Broadcasting ||
+                s is NestsSpeakerState.Failed
+        }
+    if (firstReady is NestsSpeakerState.Failed) {
+        // Unwind: cancel the orchestrator + close any speaker that
+        // managed to open before the failure surfaced.
+        orchestrator.cancel()
+        runCatching { activeSpeaker.value?.close() }
+        throw NestsException(firstReady.reason, firstReady.cause)
+    }
+
+    return ReconnectingSpeakerHandle(state, activeSpeaker, orchestrator, scope)
+}
+
+private fun isUserCancelledSpeaker(state: NestsSpeakerState.Failed): Boolean {
+    val msg = state.reason
+    // Forward-compat seam — same shape as [isUserCancelled] on the
+    // listener side. User-driven close goes through Closed today;
+    // anything else surfaced as Failed is a transport / handshake
+    // error worth retrying.
+    return msg.contains("user cancelled", ignoreCase = true)
+}
+
+private class ReconnectingSpeakerHandle(
+    private val mutableState: MutableStateFlow<NestsSpeakerState>,
+    private val activeSpeaker: MutableStateFlow<NestsSpeaker?>,
+    private val orchestrator: Job,
+    private val scope: CoroutineScope,
+) : NestsSpeaker {
+    override val state: StateFlow<NestsSpeakerState> = mutableState.asStateFlow()
+
+    private val gate = Mutex()
+
+    @Volatile private var activeBroadcast: ReissuingBroadcastHandle? = null
+
+    override suspend fun startBroadcasting(): BroadcastHandle =
+        gate.withLock {
+            check(state.value !is NestsSpeakerState.Closed) {
+                "startBroadcasting on a closed speaker"
+            }
+            check(activeBroadcast == null) {
+                "speaker is already broadcasting"
+            }
+            // Require a live (or just-connected) session — matches
+            // the listener wrapper's `subscribeSpeaker` contract.
+            // The wrapper's own `connect()` already suspended until
+            // the first session was up, so this check almost never
+            // fails in practice; it guards the second-call-after-
+            // close case.
+            activeSpeaker.value
+                ?: error("no live session — wait for state == Connected before startBroadcasting")
+
+            val handle =
+                ReissuingBroadcastHandle(activeSpeaker, scope) { closed ->
+                    if (activeBroadcast === closed) activeBroadcast = null
+                }
+            handle.start()
+            activeBroadcast = handle
+            handle
+        }
+
+    override suspend fun close() {
+        orchestrator.cancel()
+        runCatching { activeBroadcast?.close() }
+        runCatching { activeSpeaker.value?.close() }
+        if (mutableState.value !is NestsSpeakerState.Closed) {
+            mutableState.value = NestsSpeakerState.Closed
+        }
+    }
+}
+
+/**
+ * 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.
+ *
+ * `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.
+ */
+private class ReissuingBroadcastHandle(
+    private val activeSpeaker: StateFlow<NestsSpeaker?>,
+    private val scope: CoroutineScope,
+    private val onClose: (ReissuingBroadcastHandle) -> Unit,
+) : BroadcastHandle {
+    @Volatile private var desiredMuted: Boolean = false
+
+    @Volatile private var closed: Boolean = false
+    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.
+        pumpJob =
+            scope.launch {
+                activeSpeaker.collectLatest { sp ->
+                    if (sp == null || closed) return@collectLatest
+                    // Wait until the underlying speaker is ready to
+                    // broadcast (or has gone terminal). For a fresh
+                    // session this resolves immediately because the
+                    // wrapper's openOnce already saw Connected.
+                    val ready =
+                        sp.state.first { st ->
+                            st is NestsSpeakerState.Connected ||
+                                st is NestsSpeakerState.Broadcasting ||
+                                st is NestsSpeakerState.Closed ||
+                                st is NestsSpeakerState.Failed
+                        }
+                    if (ready !is NestsSpeakerState.Connected && ready !is NestsSpeakerState.Broadcasting) {
+                        return@collectLatest
+                    }
+                    if (closed) return@collectLatest
+                    val handle =
+                        runCatching { sp.startBroadcasting() }
+                            .getOrNull() ?: 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() }
+                    }
+                }
+            }
+    }
+
+    override suspend fun setMuted(muted: Boolean) {
+        if (closed) return
+        desiredMuted = muted
+        liveHandle.get()?.let { runCatching { it.setMuted(muted) } }
+    }
+
+    override suspend fun close() {
+        if (closed) return
+        closed = true
+        pumpJob?.cancel()
+        liveHandle.getAndSet(null)?.let { runCatching { it.close() } }
+        onClose(this)
+    }
+}

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

@@ -87,17 +87,33 @@ class NestMoqLiteBroadcaster(
                             }
                         if (opus.isEmpty()) continue
                         if (muted) continue
-                        runCatching { publisher.send(opus) }
-                            .onFailure { t ->
-                                if (t is CancellationException) throw t
-                                onError(
-                                    AudioException(
-                                        AudioException.Kind.PlaybackFailed,
-                                        "publisher.send failed",
-                                        t,
-                                    ),
-                                )
-                            }
+                        // One Opus frame per moq-lite group — mirrors the
+                        // nests JS reference's audio publish path, and is
+                        // load-bearing for the listener-survives-publisher-
+                        // recycle invariant: a brand-new subscriber that
+                        // attaches mid-broadcast (e.g. listener wrapper
+                        // re-subscribing after a publisher cycle) gets the
+                        // NEXT group's frames per moq-lite "from-latest"
+                        // semantics. Without endGroup, the entire broadcast
+                        // is one giant group and new subscribers wait
+                        // indefinitely. The 20 ms cadence here means at
+                        // most one frame of audio missed for any new
+                        // subscriber. See
+                        // `nestsClient/plans/2026-04-26-moq-lite-gap.md`'s
+                        // "Group size: 1 frame per group" line.
+                        runCatching {
+                            publisher.send(opus)
+                            publisher.endGroup()
+                        }.onFailure { t ->
+                            if (t is CancellationException) throw t
+                            onError(
+                                AudioException(
+                                    AudioException.Kind.PlaybackFailed,
+                                    "publisher.send failed",
+                                    t,
+                                ),
+                            )
+                        }
                     }
                 } catch (ce: CancellationException) {
                     throw ce

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

@@ -33,7 +33,6 @@ import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.consumeAsFlow
-import kotlinx.coroutines.flow.firstOrNull
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
@@ -77,6 +76,17 @@ class MoqLiteSession internal constructor(
     /** Lazily-launched relay→us inbound bidi pump; only runs while a publisher is active. */
     private var bidiPump: Job? = null
 
+    /**
+     * Single shared announce-watch pump that runs while we have any
+     * listener-side subscription. Closes the frames channel of any
+     * subscription whose broadcast suffix goes Ended on the relay's
+     * announce stream — see [pumpAnnounceWatch] for why this is the
+     * only reliable signal of publisher disconnect under moq-lite
+     * Lite-03. Lazily launched on first subscribe; lives until the
+     * session scope is cancelled.
+     */
+    private var announceWatchJob: Job? = null
+
     /** Single active publisher per session (moq-lite doesn't model multi-broadcast publishers). */
     private var activePublisher: PublisherStateImpl? = null
 
@@ -152,6 +162,18 @@ class MoqLiteSession internal constructor(
         endGroup: Long? = null,
     ): MoqLiteSubscribeHandle {
         ensureOpen()
+        // Open the announce-watch bidi BEFORE the subscribe goes
+        // out. moq-rs uses the announce stream to propagate
+        // broadcast availability into the subscriber session — a
+        // subscribe that arrives before our session has any
+        // announce bidi open is rejected with "not found", even
+        // when the publisher's session is alive on the relay. The
+        // bidi must be on the wire before subscribe sends; lazy-
+        // launching after subscribe (the obvious-but-wrong shape)
+        // races the relay's discovery and produces flaky misses,
+        // especially for fresh listener sessions opened after a
+        // wrapper-driven reconnect.
+        ensureAnnounceWatchStarted()
         val id =
             state.withLock {
                 check(!closed) { "session is closed" }
@@ -173,6 +195,27 @@ class MoqLiteSession internal constructor(
         bidi.write(Varint.encode(MoqLiteControlType.Subscribe.code))
         bidi.write(MoqLiteCodec.encodeSubscribe(request))
 
+        // Single long-running collector pump for the bidi's response
+        // side. Reads the SubscribeResponse, then keeps collecting
+        // until the peer FINs (or scope is cancelled, or
+        // handle.unsubscribe() FINs our side and the relay echoes).
+        // The flow completion IS the moq-lite-03 signal that the
+        // publisher has disconnected mid-broadcast — Lite-03 has no
+        // explicit "publisher gone" message; bidi close is it.
+        // Without this watch, the frames Channel below would never
+        // close on remote disconnect, and any consumer collecting
+        // from the wrapper-level [MoqLiteSubscribeHandle.frames]
+        // flow would sit silent indefinitely after a publisher
+        // cycle even though the relay is happy to serve a fresh
+        // subscribe under the same broadcast suffix.
+        //
+        // Why a single pump (vs separate response read + death
+        // watch): the underlying QUIC stream's `incoming` is
+        // backed by `Channel<ByteArray>.consumeAsFlow()` — which
+        // CANCELS the channel when the first collect ends. A second
+        // collect on a fresh `bidi.incoming()` Flow would see an
+        // already-cancelled channel and fire prematurely. Keeping
+        // one collect alive sidesteps that entirely.
         // moq-lite's subscribe-response is a single size-prefixed
         // message on the response side of the bidi. Read incoming
         // chunks into a buffer until the buffer holds a full payload,
@@ -213,6 +256,102 @@ class MoqLiteSession internal constructor(
         }
     }
 
+    /**
+     * Lock that serializes lazy-launch of the announce-watch.
+     * Distinct from [state] so the synchronous `announce(prefix="")`
+     * inside [ensureAnnounceWatchStarted] can suspend without
+     * blocking other state-mutating operations.
+     */
+    private val announceWatchLock = Mutex()
+
+    /**
+     * Open the shared announce-watch bidi *synchronously* (and
+     * launch its collector coroutine) if it isn't already running.
+     * Idempotent. Called from [subscribe] before the subscribe
+     * message goes on the wire so moq-rs has a chance to propagate
+     * broadcast availability into our session before the subscribe
+     * arrives — see the comment in [subscribe].
+     */
+    private suspend fun ensureAnnounceWatchStarted() {
+        announceWatchLock.withLock {
+            if (announceWatchJob != null) return
+            val handle =
+                try {
+                    announce(prefix = "")
+                } catch (ce: kotlinx.coroutines.CancellationException) {
+                    throw ce
+                } catch (_: Throwable) {
+                    // Couldn't open the announce bidi — best effort,
+                    // bail. Subscriptions still work; we just lose
+                    // automatic cycle detection (the wrapper still
+                    // re-issues on listener swap / explicit failure).
+                    return
+                }
+            announceWatchJob =
+                scope.launch {
+                    try {
+                        pumpAnnounceWatch(handle)
+                    } finally {
+                        announceWatchLock.withLock { announceWatchJob = null }
+                    }
+                }
+        }
+    }
+
+    /**
+     * Single shared announce-watch pump for ALL subscriptions on
+     * this session. Driven by the bidi opened in
+     * [ensureAnnounceWatchStarted]. For each
+     * [MoqLiteAnnounceStatus.Ended] update, iterates the
+     * subscription map and closes the frames channel of any
+     * subscription whose `broadcast` matches the announce suffix.
+     * The closed channel ends the consumer-facing
+     * `frames.consumeAsFlow()` flow naturally — same shape as a
+     * user-driven `handle.unsubscribe()` from the consumer's POV —
+     * which lets the wrapper's re-issuance pump drive a fresh
+     * subscribe against the same broadcast path. moq-lite supports
+     * subscribe-before-announce, so a subscribe issued during the
+     * gap (between Ended and the next Active under the same suffix)
+     * attaches cleanly when the new publisher comes up.
+     *
+     * This pump survives announce-bidi errors via best-effort
+     * silence — the session itself recovers via its own reconnect
+     * path. Cancelled when [scope] is cancelled (session close).
+     */
+    private suspend fun pumpAnnounceWatch(handle: MoqLiteAnnouncesHandle) {
+        try {
+            handle.updates.collect { update ->
+                if (update.status != MoqLiteAnnounceStatus.Ended) return@collect
+                val targets =
+                    state.withLock {
+                        subscriptionsBySubscribeId.values
+                            .filter { it.request.broadcast == update.suffix }
+                            .toList()
+                    }
+                for (sub in targets) {
+                    // Just close the frames channel — the
+                    // wrapper-level collect of `frames.consumeAsFlow()`
+                    // ends naturally and the wrapper pump re-issues.
+                    // Don't fire `unsubscribe(id)` here: that'd FIN
+                    // OUR side of the (still-alive) subscribe bidi,
+                    // and the wrapper's re-issue would have to open
+                    // a fresh bidi anyway. Keeping the subscribe
+                    // bidi open lets a future subscribe-before-
+                    // announce land cleanly.
+                    sub.frames.close()
+                    state.withLock { subscriptionsBySubscribeId.remove(sub.id) }
+                    runCatching { sub.bidi.finish() }
+                }
+            }
+        } catch (ce: kotlinx.coroutines.CancellationException) {
+            throw ce
+        } catch (_: Throwable) {
+            // Announce bidi died — same best-effort fallback.
+        } finally {
+            runCatching { handle.close() }
+        }
+    }
+
     /**
      * Drain inbound uni streams and route each one's group frames to
      * the matching subscription. The relay opens a fresh uni stream
@@ -355,86 +494,106 @@ class MoqLiteSession internal constructor(
     }
 
     private suspend fun handleInboundBidi(bidi: com.vitorpamplona.nestsclient.transport.WebTransportBidiStream) {
-        val buffer = MoqLiteFrameBuffer()
         val publisher = state.withLock { activePublisher } ?: return
+
+        // Single long-running collector for the bidi's full lifetime.
+        // Pre-fix this dispatch was split into a `firstOrNull()` to
+        // peek the control byte + a `readSizePrefixedFromBidiInto`
+        // to read the body — but `bidi.incoming()` is backed by
+        // `Channel<ByteArray>.consumeAsFlow(consume=true)`, which
+        // CANCELS the channel when the first collect ends. Any
+        // attempt to re-collect from the bidi (e.g. to watch for
+        // subscriber-disconnect FIN) saw an immediately-empty
+        // closed flow, firing the cleanup right after registration
+        // and starving the publisher's send path. With one collector
+        // for the bidi's whole life, the dispatch reads the message,
+        // the collector continues silently until peer FIN, and the
+        // post-collect cleanup runs exactly once — same shape the
+        // moq-lite session's [announce] pump already uses.
+        val buffer = MoqLiteFrameBuffer()
+        // typeCode is hoisted outside the collect lambda so it
+        // survives across invocations — `buffer.readVarint()`
+        // advances `pos`, so calling it again on the next collect
+        // tick would read body bytes as if they were the control
+        // varint and tear the dispatch state apart.
+        var typeCode: Long? = null
+        var dispatched = false
+        var inboundSub: MoqLiteSubscribe? = null
         try {
-            // Read the leading ControlType varint from the first chunk.
-            val first =
-                bidi.incoming().firstOrNull() ?: return
-            buffer.push(first)
-            val controlCode = buffer.readVarint() ?: return
-            val controlType = MoqLiteControlType.fromCode(controlCode) ?: return
-            when (controlType) {
-                MoqLiteControlType.Announce -> {
-                    handleAnnounceRequest(bidi, buffer, publisher)
-                }
+            bidi.incoming().collect { chunk ->
+                buffer.push(chunk)
+                if (!dispatched) {
+                    if (typeCode == null) typeCode = buffer.readVarint()
+                    val tc = typeCode ?: return@collect
+                    val controlType =
+                        MoqLiteControlType.fromCode(tc) ?: run {
+                            dispatched = true
+                            runCatching { bidi.finish() }
+                            return@collect
+                        }
+                    when (controlType) {
+                        MoqLiteControlType.Announce -> {
+                            val pleasePayload = buffer.readSizePrefixed() ?: return@collect
+                            val please = MoqLiteCodec.decodeAnnouncePlease(pleasePayload)
+                            val emittedSuffix =
+                                MoqLitePath.stripPrefix(please.prefix, publisher.suffix) ?: publisher.suffix
+                            bidi.write(
+                                MoqLiteCodec.encodeAnnounce(
+                                    MoqLiteAnnounce(
+                                        status = MoqLiteAnnounceStatus.Active,
+                                        suffix = emittedSuffix,
+                                        hops = 0L,
+                                    ),
+                                ),
+                            )
+                            publisher.registerAnnounceBidi(bidi, emittedSuffix)
+                            dispatched = true
+                        }
 
-                MoqLiteControlType.Subscribe -> {
-                    handleSubscribeRequest(bidi, buffer, publisher)
-                }
+                        MoqLiteControlType.Subscribe -> {
+                            val subPayload = buffer.readSizePrefixed() ?: return@collect
+                            val sub = MoqLiteCodec.decodeSubscribe(subPayload)
+                            bidi.write(
+                                MoqLiteCodec.encodeSubscribeOk(
+                                    MoqLiteSubscribeOk(
+                                        priority = sub.priority,
+                                        ordered = sub.ordered,
+                                        maxLatencyMillis = sub.maxLatencyMillis,
+                                        startGroup = null,
+                                        endGroup = null,
+                                    ),
+                                ),
+                            )
+                            publisher.registerInboundSubscription(sub)
+                            inboundSub = sub
+                            dispatched = true
+                        }
 
-                else -> {
-                    // Lite-03 treats Session/Fetch/Probe as separate flows;
-                    // we don't implement them here. Drop the bidi.
-                    runCatching { bidi.finish() }
+                        else -> {
+                            // Lite-03 treats Session/Fetch/Probe as
+                            // separate flows; we don't implement them.
+                            runCatching { bidi.finish() }
+                            dispatched = true
+                        }
+                    }
                 }
+                // Post-dispatch chunks are silently discarded —
+                // Lite-03's announce / subscribe bidis are idle
+                // after the response. The signal we care about is
+                // the flow's natural completion (peer FIN =
+                // subscriber-disconnect, or transport drop).
             }
         } catch (ce: CancellationException) {
             throw ce
         } catch (_: Throwable) {
-            runCatching { bidi.finish() }
+            // Bidi errored — fall through to the same cleanup.
         }
-    }
-
-    private suspend fun handleAnnounceRequest(
-        bidi: com.vitorpamplona.nestsclient.transport.WebTransportBidiStream,
-        seedBuffer: MoqLiteFrameBuffer,
-        publisher: PublisherStateImpl,
-    ) {
-        val pleasePayload = readSizePrefixedFromBidiInto(bidi.incoming(), seedBuffer)
-        val please = MoqLiteCodec.decodeAnnouncePlease(pleasePayload)
-        // The relay sets the prefix to the namespace it expects us to
-        // publish under (typically `claims.root`). Our broadcast path
-        // (after stripping the prefix) is `publisher.suffix`. moq-lite
-        // requires the suffix on the wire to be the *remaining* part
-        // after `please.prefix` — so strip it.
-        val emittedSuffix = MoqLitePath.stripPrefix(please.prefix, publisher.suffix) ?: publisher.suffix
-        bidi.write(
-            MoqLiteCodec.encodeAnnounce(
-                MoqLiteAnnounce(
-                    status = MoqLiteAnnounceStatus.Active,
-                    suffix = emittedSuffix,
-                    hops = 0L,
-                ),
-            ),
-        )
-        // Hold the bidi open until the publisher closes; if/when the
-        // application stops broadcasting, send `Ended`.
-        publisher.registerAnnounceBidi(bidi, emittedSuffix)
-    }
-
-    private suspend fun handleSubscribeRequest(
-        bidi: com.vitorpamplona.nestsclient.transport.WebTransportBidiStream,
-        seedBuffer: MoqLiteFrameBuffer,
-        publisher: PublisherStateImpl,
-    ) {
-        val subPayload = readSizePrefixedFromBidiInto(bidi.incoming(), seedBuffer)
-        val sub = MoqLiteCodec.decodeSubscribe(subPayload)
-        // Reply Ok right away — moq-lite is permissive on the publisher
-        // side; the relay decides whether the subscriber is allowed to
-        // see this broadcast.
-        bidi.write(
-            MoqLiteCodec.encodeSubscribeOk(
-                MoqLiteSubscribeOk(
-                    priority = sub.priority,
-                    ordered = sub.ordered,
-                    maxLatencyMillis = sub.maxLatencyMillis,
-                    startGroup = null,
-                    endGroup = null,
-                ),
-            ),
-        )
-        publisher.registerInboundSubscription(sub)
+        // Flow ended (peer FIN or error). Remove the inbound
+        // subscribe so the publisher's send path stops keying new
+        // groups off this dead subscriber. Announce bidis are
+        // owned by the publisher state for sending Ended on
+        // publisher-close — we don't remove them here.
+        inboundSub?.let { publisher.removeInboundSubscription(it) }
     }
 
     /**
@@ -622,6 +781,24 @@ class MoqLiteSession internal constructor(
             }
         }
 
+        /**
+         * Remove an inbound subscription whose bidi was FIN'd by the
+         * relay (subscriber disconnected). FINs the current group
+         * defensively because [openNextGroupLocked] keys each uni
+         * stream off `inboundSubs.first()`'s id; if the dropped sub
+         * was first, the current uni stream is dead-routed and the
+         * next send must open a fresh group keyed off whatever
+         * live sub is now first.
+         */
+        suspend fun removeInboundSubscription(sub: MoqLiteSubscribe) {
+            gate.withLock {
+                if (publisherClosed) return
+                if (!inboundSubs.remove(sub)) return
+                runCatching { currentGroup?.uni?.finish() }
+                currentGroup = null
+            }
+        }
+
         override suspend fun startGroup() {
             gate.withLock {
                 if (publisherClosed) return

nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeakerTest.kt

@@ -0,0 +1,474 @@
+/*
+ * 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.nestsclient
+
+import com.vitorpamplona.nestsclient.transport.WebTransportFactory
+import com.vitorpamplona.nestsclient.transport.WebTransportSession
+import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
+import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
+import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+class ReconnectingNestsSpeakerTest {
+    private val room =
+        NestsRoomConfig(
+            authBaseUrl = "https://relay.example.com/api/v1/nests",
+            endpoint = "https://relay.example.com/moq",
+            hostPubkey = "0".repeat(64),
+            roomId = "abc",
+        )
+    private val signer = NostrSignerInternal(KeyPair())
+
+    // The custom `connector` lambda below short-circuits the real
+    // mintToken / WebTransport handshake, so these two sentinels
+    // only satisfy the function signature — they're never invoked.
+    private val httpClient =
+        object : NestsClient {
+            override suspend fun mintToken(
+                room: NestsRoomConfig,
+                publish: Boolean,
+                signer: NostrSigner,
+            ): String = error("not invoked — connector overrides")
+        }
+    private val transport =
+        object : WebTransportFactory {
+            override suspend fun connect(
+                authority: String,
+                path: String,
+                bearerToken: String?,
+            ): WebTransportSession = error("not invoked — connector overrides")
+        }
+
+    /**
+     * Scripted [BroadcastHandle] paired with a [ScriptedSpeaker].
+     * Tracks `setMuted` / `close` calls and exposes the latest
+     * desired-mute value so tests can assert that the wrapper
+     * replayed mute intent on a fresh handle after a recycle.
+     */
+    private class ScriptedBroadcastHandle : BroadcastHandle {
+        // Atomic plumbing — the broadcast pump on Dispatchers.Default
+        // races against the runBlocking-thread assertions, same
+        // reasoning as ReconnectingNestsListenerTest's ScriptedListener.
+        // Backing-property names match the public projection so
+        // ktlint's `standard:backing-property-naming` rule passes.
+        private val muted = AtomicBoolean(false)
+        private val closed = AtomicBoolean(false)
+        private val setMutedCount = AtomicInteger(0)
+        private val closeCount = AtomicInteger(0)
+
+        override val isMuted: Boolean get() = muted.get()
+        val isClosed: Boolean get() = closed.get()
+        val setMutedCalls: Int get() = setMutedCount.get()
+        val closeCalls: Int get() = closeCount.get()
+
+        override suspend fun setMuted(muted: Boolean) {
+            this.muted.set(muted)
+            setMutedCount.incrementAndGet()
+        }
+
+        override suspend fun close() {
+            closed.set(true)
+            closeCount.incrementAndGet()
+        }
+    }
+
+    /**
+     * Scripted [NestsSpeaker] that opens in
+     * [NestsSpeakerState.Connected] by default. Each call to
+     * [startBroadcasting] returns a fresh [ScriptedBroadcastHandle]
+     * the test can introspect.
+     */
+    private class ScriptedSpeaker(
+        connectedRoom: NestsRoomConfig,
+        moqVersion: Long = 1,
+    ) : NestsSpeaker {
+        private val mutableState =
+            MutableStateFlow<NestsSpeakerState>(
+                NestsSpeakerState.Connected(connectedRoom, moqVersion),
+            )
+        override val state: StateFlow<NestsSpeakerState> = mutableState.asStateFlow()
+
+        private val _startCount = AtomicInteger(0)
+        val startCount: Int get() = _startCount.get()
+        val handles = mutableListOf<ScriptedBroadcastHandle>()
+
+        override suspend fun startBroadcasting(): BroadcastHandle {
+            _startCount.incrementAndGet()
+            val handle = ScriptedBroadcastHandle()
+            handles += handle
+            // Mirror the production speaker's contract: transition
+            // Connected → Broadcasting on startBroadcasting.
+            val current = mutableState.value
+            if (current is NestsSpeakerState.Connected) {
+                mutableState.value =
+                    NestsSpeakerState.Broadcasting(
+                        room = current.room,
+                        negotiatedMoqVersion = current.negotiatedMoqVersion,
+                        isMuted = false,
+                    )
+            }
+            return handle
+        }
+
+        override suspend fun close() {
+            mutableState.value = NestsSpeakerState.Closed
+        }
+
+        fun fail(reason: String) {
+            mutableState.value = NestsSpeakerState.Failed(reason)
+        }
+    }
+
+    /**
+     * Pre-fail the connector for the first N attempts so the
+     * orchestrator's failure path is exercised.
+     */
+    private class ScriptedFailure : NestsSpeaker {
+        override val state: StateFlow<NestsSpeakerState> =
+            MutableStateFlow(NestsSpeakerState.Failed("scripted-fail")).asStateFlow()
+
+        override suspend fun startBroadcasting(): BroadcastHandle = error("never connected")
+
+        override suspend fun close() {}
+    }
+
+    @Test
+    fun startBroadcasting_returns_handle_against_first_session() =
+        runBlocking {
+            val scope = CoroutineScope(SupervisorJob())
+            try {
+                val first = ScriptedSpeaker(room)
+                val reconnecting =
+                    connectReconnectingNestsSpeaker(
+                        httpClient = httpClient,
+                        transport = transport,
+                        scope = scope,
+                        room = room,
+                        signer = signer,
+                        speakerPubkeyHex = "speaker-pubkey",
+                        captureFactory = { error("not invoked — connector overrides") },
+                        encoderFactory = { error("not invoked — connector overrides") },
+                        policy = NestsReconnectPolicy(initialDelayMs = 1L),
+                        connector = { first },
+                    )
+
+                val handle = reconnecting.startBroadcasting()
+
+                // Wait for the pump to actually open the underlying
+                // broadcast against the first session.
+                withTimeout(5_000L) {
+                    while (first.startCount == 0) delay(5)
+                }
+
+                assertEquals(1, first.startCount)
+                assertEquals(1, first.handles.size)
+                assertFalse(handle.isMuted, "fresh handle should not be muted")
+
+                reconnecting.close()
+            } finally {
+                scope.cancel()
+            }
+        }
+
+    @Test
+    fun proactive_token_refresh_recycles_speaker_without_failure_state() =
+        runBlocking {
+            val scope = CoroutineScope(SupervisorJob())
+            try {
+                val first = ScriptedSpeaker(room)
+                val second = ScriptedSpeaker(room)
+                val third = ScriptedSpeaker(room)
+                val speakersInOrder = mutableListOf(first, second, third)
+
+                // Synchronized list — the watcher coroutine appends
+                // concurrently with the assertion's `.any { ... }`,
+                // and a plain ArrayList raises CME on iteration. The
+                // listener test gets away with a plain list because
+                // its watcher is cancelled by `reconnecting.close()`
+                // before the assertion runs in some timing windows;
+                // we don't rely on that here.
+                val seenStates: MutableList<NestsSpeakerState> =
+                    java.util.Collections.synchronizedList(mutableListOf())
+
+                val reconnecting =
+                    connectReconnectingNestsSpeaker(
+                        httpClient = httpClient,
+                        transport = transport,
+                        scope = scope,
+                        room = room,
+                        signer = signer,
+                        speakerPubkeyHex = "speaker-pubkey",
+                        captureFactory = { error("unused") },
+                        encoderFactory = { error("unused") },
+                        // 50 ms refresh — small enough to fire
+                        // quickly; large enough that openOnce
+                        // resolves Connected first.
+                        tokenRefreshAfterMs = 50L,
+                        connector = { speakersInOrder.removeAt(0) },
+                    )
+
+                val watcher =
+                    scope.launch {
+                        reconnecting.state.collect { seenStates += it }
+                    }
+
+                val handle = reconnecting.startBroadcasting()
+
+                // Wait for at least 2 underlying speakers to have
+                // been consumed (initial + first refresh).
+                withTimeout(5_000L) {
+                    while (speakersInOrder.size > 1) delay(10)
+                }
+
+                // Wait for the broadcast pump to have started a
+                // broadcast on the second session.
+                withTimeout(5_000L) {
+                    while (second.startCount == 0) delay(5)
+                }
+
+                // Critical postcondition: the wrapper's outward
+                // state must NEVER show Reconnecting or Failed
+                // during a clean refresh — the user-visible UI
+                // stays Broadcasting (or briefly Connected during
+                // the cutover) throughout.
+                //
+                // Cancel the watcher BEFORE the assertion so we can
+                // iterate seenStates without racing against further
+                // appends. (synchronizedList only guards individual
+                // ops, not iteration.)
+                watcher.cancel()
+                watcher.join()
+                val snapshot: List<NestsSpeakerState> = synchronized(seenStates) { seenStates.toList() }
+                val sawReconnecting = snapshot.any { it is NestsSpeakerState.Reconnecting }
+                val sawFailed = snapshot.any { it is NestsSpeakerState.Failed }
+                assertTrue(
+                    !sawReconnecting && !sawFailed,
+                    "proactive refresh must not surface Reconnecting/Failed; saw=$snapshot",
+                )
+
+                // Both speakers should have seen exactly one
+                // startBroadcasting — the pump re-issues against
+                // the new session, doesn't double-broadcast on
+                // the old one.
+                assertEquals(1, first.startCount)
+                assertEquals(1, second.startCount)
+
+                handle.close()
+                reconnecting.close()
+            } finally {
+                scope.cancel()
+            }
+        }
+
+    @Test
+    fun setMuted_intent_replays_on_new_session_after_refresh() =
+        runBlocking {
+            val scope = CoroutineScope(SupervisorJob())
+            try {
+                val first = ScriptedSpeaker(room)
+                val second = ScriptedSpeaker(room)
+                val speakersInOrder = mutableListOf(first, second)
+
+                val reconnecting =
+                    connectReconnectingNestsSpeaker(
+                        httpClient = httpClient,
+                        transport = transport,
+                        scope = scope,
+                        room = room,
+                        signer = signer,
+                        speakerPubkeyHex = "speaker-pubkey",
+                        captureFactory = { error("unused") },
+                        encoderFactory = { error("unused") },
+                        tokenRefreshAfterMs = 80L,
+                        connector = { speakersInOrder.removeAt(0) },
+                    )
+
+                val handle = reconnecting.startBroadcasting()
+
+                withTimeout(5_000L) {
+                    while (first.startCount == 0) delay(5)
+                }
+
+                // User mutes mid-broadcast on the first session.
+                handle.setMuted(true)
+
+                withTimeout(5_000L) {
+                    while (first.handles[0].setMutedCalls == 0) delay(5)
+                }
+                assertTrue(first.handles[0].isMuted, "first handle should be muted by user toggle")
+
+                // Wait for refresh to swap to the second session.
+                withTimeout(5_000L) {
+                    while (second.startCount == 0) delay(5)
+                }
+
+                // Critical postcondition: the new underlying handle
+                // must have inherited the muted=true intent without
+                // the user calling setMuted again.
+                assertTrue(
+                    second.handles[0].isMuted,
+                    "post-refresh handle must inherit muted=true; setMutedCalls=${second.handles[0].setMutedCalls}",
+                )
+
+                handle.close()
+                reconnecting.close()
+            } finally {
+                scope.cancel()
+            }
+        }
+
+    @Test
+    fun close_cancels_pump_and_closes_live_handle() =
+        runBlocking {
+            val scope = CoroutineScope(SupervisorJob())
+            try {
+                val only = ScriptedSpeaker(room)
+                val reconnecting =
+                    connectReconnectingNestsSpeaker(
+                        httpClient = httpClient,
+                        transport = transport,
+                        scope = scope,
+                        room = room,
+                        signer = signer,
+                        speakerPubkeyHex = "speaker-pubkey",
+                        captureFactory = { error("unused") },
+                        encoderFactory = { error("unused") },
+                        connector = { only },
+                    )
+
+                val handle = reconnecting.startBroadcasting()
+
+                withTimeout(5_000L) {
+                    while (only.startCount == 0) delay(5)
+                }
+                val live = only.handles[0]
+
+                handle.close()
+
+                // Live handle should be closed exactly once after
+                // close(). The pump-side close+the user-side close
+                // race; the handle's own idempotence guarantees
+                // exactly one effective close.
+                withTimeout(5_000L) {
+                    while (!live.isClosed) delay(5)
+                }
+                assertTrue(live.isClosed)
+
+                reconnecting.close()
+            } finally {
+                scope.cancel()
+            }
+        }
+
+    @Test
+    fun first_attempt_failure_throws_from_connect() =
+        runBlocking {
+            val scope = CoroutineScope(SupervisorJob())
+            try {
+                // Connector returns a speaker already in Failed
+                // state — connectReconnectingNestsSpeaker should
+                // surface this as a thrown NestsException rather
+                // than returning a wrapper in Failed state, so the
+                // VM's `try/catch` around the connect call lights
+                // up.
+                var threw: Throwable? = null
+                try {
+                    connectReconnectingNestsSpeaker(
+                        httpClient = httpClient,
+                        transport = transport,
+                        scope = scope,
+                        room = room,
+                        signer = signer,
+                        speakerPubkeyHex = "speaker-pubkey",
+                        captureFactory = { error("unused") },
+                        encoderFactory = { error("unused") },
+                        policy = NestsReconnectPolicy(maxAttempts = 1),
+                        connector = { ScriptedFailure() },
+                    )
+                } catch (t: Throwable) {
+                    threw = t
+                }
+
+                assertTrue(
+                    threw is NestsException,
+                    "first-attempt failure should throw NestsException; got $threw",
+                )
+                assertTrue(
+                    threw.message?.contains("scripted-fail") == true,
+                    "exception should preserve underlying reason; got ${threw.message}",
+                )
+            } finally {
+                scope.cancel()
+            }
+        }
+
+    @Test
+    fun startBroadcasting_after_close_throws() =
+        runBlocking {
+            val scope = CoroutineScope(SupervisorJob())
+            try {
+                val only = ScriptedSpeaker(room)
+                val reconnecting =
+                    connectReconnectingNestsSpeaker(
+                        httpClient = httpClient,
+                        transport = transport,
+                        scope = scope,
+                        room = room,
+                        signer = signer,
+                        speakerPubkeyHex = "speaker-pubkey",
+                        captureFactory = { error("unused") },
+                        encoderFactory = { error("unused") },
+                        connector = { only },
+                    )
+
+                reconnecting.close()
+                withTimeout(5_000L) {
+                    reconnecting.state.first { it is NestsSpeakerState.Closed }
+                }
+
+                var threw: Throwable? = null
+                try {
+                    reconnecting.startBroadcasting()
+                } catch (t: Throwable) {
+                    threw = t
+                }
+                assertTrue(threw is IllegalStateException, "must reject startBroadcasting on closed wrapper; got $threw")
+            } finally {
+                scope.cancel()
+            }
+        }
+}

nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt

@@ -28,8 +28,10 @@ import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.async
 import kotlinx.coroutines.cancelAndJoin
 import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.firstOrNull
 import kotlinx.coroutines.flow.take
 import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.flow.transformWhile
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withTimeout
 import kotlin.test.AfterTest
@@ -77,8 +79,7 @@ class MoqLiteSessionTest {
 
             val peerHandlesSubscribe =
                 async {
-                    val bidi = serverSide.peerOpenedBidiStreams().first()
-                    val req = readSubscribeRequest(bidi)
+                    val (bidi, req) = nextSubscribeBidi(serverSide)
                     assertEquals("speakerPubkey", req.broadcast)
                     assertEquals("audio/data", req.track)
                     assertEquals(MoqLiteSession.DEFAULT_PRIORITY, req.priority)
@@ -128,8 +129,7 @@ class MoqLiteSessionTest {
 
             val peer =
                 async {
-                    val bidi = serverSide.peerOpenedBidiStreams().first()
-                    readSubscribeRequest(bidi)
+                    val (bidi, _) = nextSubscribeBidi(serverSide)
                     bidi.write(
                         MoqLiteCodec.encodeSubscribeDrop(
                             MoqLiteSubscribeDrop(errorCode = 4L, reasonPhrase = "no such broadcast"),
@@ -203,10 +203,14 @@ class MoqLiteSessionTest {
 
             // Set up two parallel subscriptions — peer accepts both, replies
             // with Ok, then pushes one group per subscription out of order.
+            // Use [nextSubscribeBidi] (not raw `peerOpenedBidiStreams`)
+            // because the session lazy-opens an announce-watch bidi on
+            // first subscribe (publisher-disconnect detection); raw
+            // `.first()` would race with that and occasionally pick up
+            // the announce bidi instead of the subscribe one.
             val subAck =
                 async {
-                    val bidiA = serverSide.peerOpenedBidiStreams().first()
-                    readSubscribeRequest(bidiA)
+                    val (bidiA, _) = nextSubscribeBidi(serverSide)
                     bidiA.write(MoqLiteCodec.encodeSubscribeOk(okFor(0L)))
                     bidiA
                 }
@@ -215,8 +219,7 @@ class MoqLiteSessionTest {
 
             val subAck2 =
                 async {
-                    val bidiB = serverSide.peerOpenedBidiStreams().first()
-                    readSubscribeRequest(bidiB)
+                    val (bidiB, _) = nextSubscribeBidi(serverSide)
                     bidiB.write(MoqLiteCodec.encodeSubscribeOk(okFor(1L)))
                     bidiB
                 }
@@ -391,8 +394,7 @@ class MoqLiteSessionTest {
             var peerBidi: FakeBidiStream? = null
             val peer =
                 async {
-                    val bidi = serverSide.peerOpenedBidiStreams().first()
-                    readSubscribeRequest(bidi)
+                    val (bidi, _) = nextSubscribeBidi(serverSide)
                     bidi.write(MoqLiteCodec.encodeSubscribeOk(okFor(0L)))
                     peerBidi = bidi
                     // Drain whatever the listener writes after Ok — moq-lite
@@ -429,6 +431,47 @@ class MoqLiteSessionTest {
         return MoqLiteCodec.decodeSubscribe(payload)
     }
 
+    /**
+     * Pull the next Subscribe bidi the peer's side has accepted,
+     * skipping any housekeeping bidis (e.g. the announce-watch
+     * bidi that [MoqLiteSession.subscribe] lazy-launches once per
+     * session to detect publisher disconnect via `Announce(Ended)`
+     * — see `pumpAnnounceWatch`). Each candidate bidi is peeked
+     * by reading its first chunk; if the control varint is
+     * Subscribe, the bidi + decoded body are returned. Other bidis
+     * are simply abandoned — their pump-side `bidi.incoming()`
+     * collect just sits idle until the test ends.
+     */
+    private suspend fun nextSubscribeBidi(serverSide: FakeWebTransport): Pair<FakeBidiStream, MoqLiteSubscribe> =
+        serverSide
+            .peerOpenedBidiStreams()
+            .transformWhile { bidi ->
+                val firstChunk = bidi.incoming().firstOrNull() ?: return@transformWhile true
+                val code = MoqLiteFrameBuffer().apply { push(firstChunk) }.readVarint()
+                if (code != MoqLiteControlType.Subscribe.code) {
+                    // Housekeeping bidi (announce watch, etc.) —
+                    // drop it on the floor; the session-side
+                    // collector will idle indefinitely, which is
+                    // fine for unit tests under runBlocking +
+                    // pumpScope cleanup.
+                    return@transformWhile true
+                }
+                val bodyChunk =
+                    bidi.incoming().firstOrNull()
+                        ?: error("subscribe stream FIN before body")
+                val payload =
+                    MoqLiteFrameBuffer().apply { push(bodyChunk) }.readSizePrefixed()
+                        ?: error("subscribe body chunk did not contain a complete size-prefixed payload")
+                emit(bidi to MoqLiteCodec.decodeSubscribe(payload))
+                // Terminate upstream collection — without this the
+                // helper would block waiting for a NEXT bidi that
+                // may never come, since `takeWhile` only re-checks
+                // its predicate when the next value emits. Tests
+                // that open exactly two subscribes (no third bidi
+                // to nudge the flow forward) used to hang here.
+                false
+            }.firstOrNull() ?: error("flow ended without a Subscribe bidi")
+
     private fun framePayload(bytes: ByteArray): ByteArray = Varint.encode(bytes.size.toLong()) + bytes
 
     private fun okFor(

nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/OkHttpNestsClient.kt

@@ -22,6 +22,7 @@ package com.vitorpamplona.nestsclient
 
 import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.withContext
 import okhttp3.MediaType.Companion.toMediaType
 import okhttp3.OkHttpClient
@@ -31,6 +32,10 @@ import okhttp3.Response
 import java.io.EOFException
 import java.io.IOException
 import java.net.SocketException
+import java.time.ZoneId
+import java.time.ZonedDateTime
+import java.time.format.DateTimeFormatter
+import kotlin.math.min
 
 /**
  * OkHttp-backed [NestsClient] used on JVM + Android. A shared [OkHttpClient]
@@ -54,17 +59,20 @@ class OkHttpNestsClient(
                 append('}')
             }
         val bodyBytes = bodyJson.encodeToByteArray()
-        // NIP-98 binds the signed event to (url, method, body-hash) so the
-        // server can reject a token replayed against a different request.
-        val authHeader =
-            NestsAuth.header(
-                signer = signer,
-                url = url,
-                method = "POST",
-                payload = bodyBytes,
-            )
 
-        val request =
+        // NIP-98 events embed `created_at`; the moq-auth reference
+        // accepts a 60 s validity window. A retry that waits longer
+        // than that (e.g. exponential backoff after 429) MUST re-sign
+        // or the server returns 401 "Event too old". So we build the
+        // request lazily on every attempt instead of once up front.
+        val buildRequest: suspend () -> Request = {
+            val authHeader =
+                NestsAuth.header(
+                    signer = signer,
+                    url = url,
+                    method = "POST",
+                    payload = bodyBytes,
+                )
             Request
                 .Builder()
                 .url(url)
@@ -72,9 +80,10 @@ class OkHttpNestsClient(
                 .header("Authorization", authHeader)
                 .header("Accept", "application/json")
                 .build()
+        }
 
         return withContext(Dispatchers.IO) {
-            executeWithTransportRetry(request, url)
+            executeWithRetry(buildRequest, url)
                 .use { response ->
                     val body = response.body.string()
                     if (!response.isSuccessful) {
@@ -97,44 +106,124 @@ class OkHttpNestsClient(
     }
 
     /**
-     * Send [request] and tolerate one transport-layer hiccup. OkHttp's
-     * built-in `retryOnConnectionFailure` does NOT retry POSTs once any
-     * byte of the request body has been written — but a stale pooled
-     * connection can RST or EOF *exactly* in that window, especially on
-     * mobile networks (and during interop test runs after an idle gap
-     * between test classes). One retry on `SocketException` /
-     * `EOFException` / `IOException` recovers cleanly because
-     * `Request` builders are immutable; OkHttp opens a fresh
-     * connection on the second try.
+     * Send [request] and tolerate two recoverable failure modes:
      *
-     * Anything that's not a transient transport failure (HTTP 4xx /
-     * 5xx, malformed response) is left to the caller as before.
+     *   1. Transport hiccup. OkHttp's built-in `retryOnConnectionFailure`
+     *      does NOT retry POSTs once any byte of the request body has
+     *      been written — but a stale pooled connection can RST or EOF
+     *      *exactly* in that window, especially on mobile networks (and
+     *      during interop test runs after an idle gap between test
+     *      classes). One retry on `SocketException` / `EOFException` /
+     *      `IOException` recovers cleanly because `Request` builders
+     *      are immutable; OkHttp opens a fresh connection on the
+     *      second try.
+     *
+     *   2. HTTP 429 (Too Many Requests). The nostrnests reference
+     *      `moq-auth` sidecar rate-limits 20/min/IP; production
+     *      back-ends may be stricter. We respect a `Retry-After`
+     *      header (delta-seconds OR HTTP-date) when present and fall
+     *      back to capped exponential backoff when absent, retrying
+     *      up to [MAX_RATE_LIMIT_RETRIES] times. Cancellable: the
+     *      backoff suspends with `delay`, so a coroutine cancellation
+     *      tears the retry loop down at the next sleep boundary.
+     *
+     * Anything that's not a transient transport failure or a 429 (HTTP
+     * 4xx other than 429, 5xx, malformed response) is left to the
+     * caller as before.
      */
-    private fun executeWithTransportRetry(
-        request: Request,
+    private suspend fun executeWithRetry(
+        buildRequest: suspend () -> Request,
         url: String,
     ): Response {
-        var lastError: Throwable? = null
-        repeat(2) { attempt ->
-            try {
-                return http.newCall(request).execute()
-            } catch (e: SocketException) {
-                lastError = e
-            } catch (e: EOFException) {
-                lastError = e
-            } catch (e: IOException) {
-                // OkHttp wraps a wide variety of transport faults
-                // (StreamResetException, ConnectionShutdownException,
-                // …) under IOException. Retry once; second pass either
-                // succeeds against a fresh connection or surfaces the
-                // real error.
-                lastError = e
+        var transportError: Throwable? = null
+        var transportAttempts = 0
+        var rateLimitAttempts = 0
+        while (true) {
+            val request = buildRequest()
+            val response: Response =
+                try {
+                    http.newCall(request).execute()
+                } catch (e: SocketException) {
+                    transportError = e
+                    if (++transportAttempts >= MAX_TRANSPORT_RETRIES) throw NestsException("Failed to reach $url", e)
+                    continue
+                } catch (e: EOFException) {
+                    transportError = e
+                    if (++transportAttempts >= MAX_TRANSPORT_RETRIES) throw NestsException("Failed to reach $url", e)
+                    continue
+                } catch (e: IOException) {
+                    // OkHttp wraps a wide variety of transport faults
+                    // (StreamResetException, ConnectionShutdownException,
+                    // …) under IOException. Retry once; second pass
+                    // either succeeds against a fresh connection or
+                    // surfaces the real error.
+                    transportError = e
+                    if (++transportAttempts >= MAX_TRANSPORT_RETRIES) throw NestsException("Failed to reach $url", e)
+                    continue
+                }
+
+            if (response.code != 429 || rateLimitAttempts >= MAX_RATE_LIMIT_RETRIES) {
+                return response
             }
+
+            val retryAfter = response.header("Retry-After")
+            // Drain + close so the connection returns to the pool;
+            // the Response we hand back to the caller is the one from
+            // the next iteration.
+            response.close()
+            val delayMs = computeRateLimitBackoffMs(retryAfter, rateLimitAttempts)
+            rateLimitAttempts++
+            delay(delayMs)
         }
-        throw NestsException("Failed to reach $url", lastError)
+        // Unreachable — every path either returns or throws.
+        @Suppress("UNREACHABLE_CODE")
+        throw NestsException("Failed to reach $url", transportError)
     }
 
     private companion object {
         private val JSON_MEDIA_TYPE = "application/json".toMediaType()
+
+        private const val MAX_TRANSPORT_RETRIES = 2
     }
 }
+
+// Worst-case total wait at INITIAL_BACKOFF_MS=1s, MAX_BACKOFF_MS=16s,
+// 7 retries: 1+2+4+8+16+16+16 = 63 s — just over the moq-auth
+// reference 60 s/IP rate-limit window, so a clustered burst of mints
+// (typical in interop test runs) will outlast the bucket reset
+// instead of cascading.
+internal const val MAX_RATE_LIMIT_RETRIES = 7
+
+internal const val INITIAL_BACKOFF_MS = 1_000L
+
+internal const val MAX_BACKOFF_MS = 16_000L
+
+/**
+ * Translate a `Retry-After` header (RFC 7231 §7.1.3 — either
+ * delta-seconds or HTTP-date) into millis to sleep. Falls back to
+ * capped exponential backoff (1s, 2s, 4s, 8s, …) when the header is
+ * absent or unparseable.
+ *
+ * `nowMs` is parameterised so date-driven cases are testable without
+ * `Thread.sleep`-style time travel.
+ */
+internal fun computeRateLimitBackoffMs(
+    retryAfterHeader: String?,
+    attempt: Int,
+    nowMs: Long = System.currentTimeMillis(),
+): Long {
+    if (retryAfterHeader != null) {
+        retryAfterHeader.trim().toLongOrNull()?.let { seconds ->
+            return seconds.coerceAtLeast(0L) * 1_000L
+        }
+        try {
+            val target = ZonedDateTime.parse(retryAfterHeader, DateTimeFormatter.RFC_1123_DATE_TIME)
+            val targetMs = target.withZoneSameInstant(ZoneId.of("UTC")).toInstant().toEpochMilli()
+            if (targetMs > nowMs) return targetMs - nowMs
+        } catch (_: Throwable) {
+            // Fall through to exponential backoff.
+        }
+    }
+    val base = INITIAL_BACKOFF_MS shl attempt
+    return min(base, MAX_BACKOFF_MS)
+}

nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/OkHttpNestsClientRateLimitTest.kt

@@ -0,0 +1,304 @@
+/*
+ * 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.nestsclient
+
+import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
+import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
+import kotlinx.coroutines.runBlocking
+import okhttp3.OkHttpClient
+import org.junit.Test
+import java.io.BufferedReader
+import java.io.IOException
+import java.io.InputStreamReader
+import java.io.OutputStream
+import java.net.ServerSocket
+import java.net.Socket
+import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.thread
+import kotlin.concurrent.withLock
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+import kotlin.test.assertTrue
+
+/**
+ * Backoff math + retry-loop coverage for [OkHttpNestsClient]. The
+ * production case is the nostrnests `moq-auth` sidecar's 20/min/IP
+ * limiter — when the all-tests-together interop run blasts more than
+ * 20 mintToken calls in <60 s, every test class after the 20th hit
+ * cascades on `429`. The retry-with-backoff added here lets the
+ * client wait out the 60 s window instead of bubbling up a hard
+ * failure.
+ */
+class OkHttpNestsClientRateLimitTest {
+    @Test
+    fun computes_backoff_from_retry_after_seconds_header() {
+        // Plain integer = delta-seconds per RFC 7231 §7.1.3.
+        assertEquals(2_000L, computeRateLimitBackoffMs("2", attempt = 0))
+        assertEquals(2_000L, computeRateLimitBackoffMs(" 2 ", attempt = 0))
+        // Negative → 0 (don't sleep backwards).
+        assertEquals(0L, computeRateLimitBackoffMs("-5", attempt = 0))
+    }
+
+    @Test
+    fun computes_backoff_from_retry_after_http_date_header() {
+        // 30 s from `now`. RFC_1123 format = the HTTP-date wire format.
+        // 1_700_000_000_000 ms = 2023-11-14T22:13:20Z (Tuesday).
+        val now = 1_700_000_000_000L
+        val target = "Tue, 14 Nov 2023 22:13:50 GMT"
+        assertEquals(
+            30_000L,
+            computeRateLimitBackoffMs(target, attempt = 0, nowMs = now),
+        )
+    }
+
+    @Test
+    fun falls_back_to_exponential_backoff_when_header_absent() {
+        assertEquals(1_000L, computeRateLimitBackoffMs(null, attempt = 0))
+        assertEquals(2_000L, computeRateLimitBackoffMs(null, attempt = 1))
+        assertEquals(4_000L, computeRateLimitBackoffMs(null, attempt = 2))
+        assertEquals(8_000L, computeRateLimitBackoffMs(null, attempt = 3))
+        assertEquals(16_000L, computeRateLimitBackoffMs(null, attempt = 4))
+        // Capped — further attempts stay at MAX_BACKOFF_MS.
+        assertEquals(16_000L, computeRateLimitBackoffMs(null, attempt = 5))
+        assertEquals(16_000L, computeRateLimitBackoffMs(null, attempt = 10))
+    }
+
+    @Test
+    fun unparseable_retry_after_falls_back_to_exponential_backoff() {
+        // Garbage header → ignore + use exponential.
+        assertEquals(1_000L, computeRateLimitBackoffMs("garbage", attempt = 0))
+        assertEquals(2_000L, computeRateLimitBackoffMs("garbage", attempt = 1))
+    }
+
+    @Test
+    fun mint_token_retries_through_429_and_returns_token() =
+        runBlocking {
+            val server = TinyHttpServer()
+            // First 2 responses = 429 with Retry-After: 0, 3rd = 200 with token.
+            server.enqueue(rateLimited(retryAfterSeconds = 0))
+            server.enqueue(rateLimited(retryAfterSeconds = 0))
+            server.enqueue(success(token = "T1"))
+
+            try {
+                val client = OkHttpNestsClient(http = OkHttpClient())
+                val token =
+                    client.mintToken(
+                        room = roomConfig(server.baseUrl),
+                        publish = false,
+                        signer = NostrSignerInternal(KeyPair()),
+                    )
+                assertEquals("T1", token)
+                assertEquals(3, server.requestCount(), "expected 2 retries before the 200")
+            } finally {
+                server.close()
+            }
+        }
+
+    @Test
+    fun mint_token_gives_up_after_max_rate_limit_retries() =
+        runBlocking {
+            val server = TinyHttpServer()
+            // 6 = MAX_RATE_LIMIT_RETRIES + 1 → expect retries to exhaust on
+            // the (MAX_RATE_LIMIT_RETRIES+1)-th 429 and surface a NestsException.
+            repeat(MAX_RATE_LIMIT_RETRIES + 2) { server.enqueue(rateLimited(retryAfterSeconds = 0)) }
+
+            try {
+                val client = OkHttpNestsClient(http = OkHttpClient())
+                val ex =
+                    assertFailsWith<NestsException> {
+                        client.mintToken(
+                            room = roomConfig(server.baseUrl),
+                            publish = false,
+                            signer = NostrSignerInternal(KeyPair()),
+                        )
+                    }
+                assertEquals(429, ex.status)
+                assertEquals(
+                    MAX_RATE_LIMIT_RETRIES + 1,
+                    server.requestCount(),
+                    "expected one initial attempt + MAX_RATE_LIMIT_RETRIES retries",
+                )
+            } finally {
+                server.close()
+            }
+        }
+
+    @Test
+    fun non_429_4xx_is_not_retried() =
+        runBlocking {
+            val server = TinyHttpServer()
+            server.enqueue(StubResponse(401, "Unauthorized", body = "{\"error\":\"bad sig\"}"))
+
+            try {
+                val client = OkHttpNestsClient(http = OkHttpClient())
+                val ex =
+                    assertFailsWith<NestsException> {
+                        client.mintToken(
+                            room = roomConfig(server.baseUrl),
+                            publish = false,
+                            signer = NostrSignerInternal(KeyPair()),
+                        )
+                    }
+                assertEquals(401, ex.status)
+                assertEquals(1, server.requestCount(), "401 must not trigger retry")
+                assertTrue(ex.message?.contains("bad sig") == true)
+            } finally {
+                server.close()
+            }
+        }
+
+    private fun roomConfig(baseUrl: String): NestsRoomConfig =
+        NestsRoomConfig(
+            authBaseUrl = baseUrl,
+            endpoint = "https://example.invalid:4443/",
+            hostPubkey = "0".repeat(64),
+            roomId = "ratelimit-${System.nanoTime()}",
+        )
+
+    private fun rateLimited(retryAfterSeconds: Int): StubResponse =
+        StubResponse(
+            status = 429,
+            statusText = "Too Many Requests",
+            body = "{\"error\":\"Too many requests, try again later\"}",
+            headers = mapOf("Retry-After" to retryAfterSeconds.toString()),
+        )
+
+    private fun success(token: String): StubResponse =
+        StubResponse(
+            status = 200,
+            statusText = "OK",
+            body = "{\"token\":\"$token\",\"url\":\"https://example.invalid:4443/\"}",
+        )
+}
+
+/**
+ * Single canned HTTP response — the server pops one of these per
+ * incoming connection.
+ */
+private data class StubResponse(
+    val status: Int,
+    val statusText: String,
+    val body: String,
+    val headers: Map<String, String> = emptyMap(),
+)
+
+/**
+ * Tiny single-threaded HTTP/1.1 server backed by [ServerSocket]. Pops
+ * one [StubResponse] per request from the FIFO queue. Drains the
+ * request body so OkHttp's pool sees a clean keepalive frame; closes
+ * the connection after each response so the next request reads from
+ * a clean Socket. Sufficient for unit testing OkHttp's retry path
+ * without pulling MockWebServer in as a dep.
+ */
+private class TinyHttpServer : AutoCloseable {
+    private val socket = ServerSocket(0)
+    private val responses = ArrayDeque<StubResponse>()
+    private val lock = ReentrantLock()
+    private val seen = AtomicInteger(0)
+    private val running =
+        java.util.concurrent.atomic
+            .AtomicBoolean(true)
+    private val acceptor =
+        thread(name = "TinyHttpServer-acceptor", isDaemon = true) {
+            while (running.get()) {
+                val client =
+                    try {
+                        socket.accept()
+                    } catch (_: IOException) {
+                        return@thread
+                    }
+                handle(client)
+            }
+        }
+
+    val baseUrl: String = "http://127.0.0.1:${socket.localPort}"
+
+    fun enqueue(response: StubResponse) {
+        lock.withLock { responses.addLast(response) }
+    }
+
+    fun requestCount(): Int = seen.get()
+
+    private fun handle(client: Socket) {
+        client.use { sock ->
+            val input = BufferedReader(InputStreamReader(sock.getInputStream(), Charsets.ISO_8859_1))
+            // Request line.
+            input.readLine() ?: return
+            // Drain headers, capture Content-Length.
+            var contentLength = 0
+            while (true) {
+                val line = input.readLine() ?: return
+                if (line.isEmpty()) break
+                val parts = line.split(":", limit = 2)
+                if (parts.size == 2 && parts[0].equals("Content-Length", ignoreCase = true)) {
+                    contentLength = parts[1].trim().toIntOrNull() ?: 0
+                }
+            }
+            // Drain body so the request is fully consumed before we reply.
+            if (contentLength > 0) {
+                val drained = CharArray(contentLength)
+                var read = 0
+                while (read < contentLength) {
+                    val n = input.read(drained, read, contentLength - read)
+                    if (n < 0) break
+                    read += n
+                }
+            }
+            seen.incrementAndGet()
+
+            val response =
+                lock.withLock { responses.removeFirstOrNull() }
+                    ?: StubResponse(500, "Internal Server Error", "no response queued")
+
+            writeResponse(sock.getOutputStream(), response)
+        }
+    }
+
+    private fun writeResponse(
+        out: OutputStream,
+        response: StubResponse,
+    ) {
+        val bodyBytes = response.body.toByteArray(Charsets.UTF_8)
+        val headerLines =
+            buildString {
+                append("HTTP/1.1 ")
+                    .append(response.status)
+                    .append(' ')
+                    .append(response.statusText)
+                    .append("\r\n")
+                append("Content-Type: application/json\r\n")
+                append("Content-Length: ").append(bodyBytes.size).append("\r\n")
+                append("Connection: close\r\n")
+                response.headers.forEach { (k, v) -> append(k).append(": ").append(v).append("\r\n") }
+                append("\r\n")
+            }
+        out.write(headerLines.toByteArray(Charsets.ISO_8859_1))
+        out.write(bodyBytes)
+        out.flush()
+    }
+
+    override fun close() {
+        running.set(false)
+        runCatching { socket.close() }
+        runCatching { acceptor.join(1_000) }
+    }
+}

nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsReconnectingListenerInteropTest.kt

@@ -24,12 +24,14 @@ import com.vitorpamplona.nestsclient.NestsListener
 import com.vitorpamplona.nestsclient.NestsListenerState
 import com.vitorpamplona.nestsclient.NestsReconnectPolicy
 import com.vitorpamplona.nestsclient.NestsRoomConfig
+import com.vitorpamplona.nestsclient.NestsSpeakerState
 import com.vitorpamplona.nestsclient.OkHttpNestsClient
 import com.vitorpamplona.nestsclient.audio.AudioCapture
 import com.vitorpamplona.nestsclient.audio.OpusEncoder
 import com.vitorpamplona.nestsclient.connectNestsListener
 import com.vitorpamplona.nestsclient.connectNestsSpeaker
 import com.vitorpamplona.nestsclient.connectReconnectingNestsListener
+import com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker
 import com.vitorpamplona.nestsclient.transport.QuicWebTransportFactory
 import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
 import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
@@ -408,4 +410,176 @@ class NostrNestsReconnectingListenerInteropTest {
             harnessOrNull = null
         }
     }
+
+    /**
+     * Captures the listener-survives-publisher-recycle invariant —
+     * the gap discovered while validating
+     * [connectReconnectingNestsSpeaker]. The setup:
+     *
+     *   - SUT (listener): a [connectReconnectingNestsListener]-backed
+     *     handle, vanilla refresh disabled so the listener's own
+     *     session never recycles during the test.
+     *   - Driver (speaker): a [connectReconnectingNestsSpeaker] with
+     *     a small `tokenRefreshAfterMs`, forcing the publisher's
+     *     session to recycle mid-stream.
+     *
+     * The single [SubscribeHandle] returned from
+     * `subscribeSpeaker(pubkey)` MUST keep emitting frames after the
+     * publisher cycles. The session-layer death-watch in
+     * [com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession.subscribe]
+     * detects the publisher's bidi-FIN, closes the frames channel,
+     * the wrapper-level `reissuingSubscribe` pump's collect ends
+     * naturally, and the pump re-issues a fresh subscribe via the
+     * outer collectLatest's loop semantics.
+     *
+     * Skipped by default — set `-DnestsInterop=true` to enable.
+     */
+    @Test
+    fun subscribe_handle_survives_publisher_recycle() =
+        runBlocking {
+            NostrNestsHarness.assumeNestsInterop()
+            val harness = harnessOrNull ?: return@runBlocking
+
+            val signer = NostrSignerInternal(KeyPair())
+            val pubkey = signer.pubKey
+            val room =
+                NestsRoomConfig(
+                    authBaseUrl = harness.authBaseUrl,
+                    endpoint = harness.moqEndpoint,
+                    hostPubkey = pubkey,
+                    roomId = "lst-pub-cycle-${System.currentTimeMillis()}",
+                )
+
+            val httpClient = OkHttpNestsClient()
+            val transport =
+                QuicWebTransportFactory(
+                    certificateValidator = PermissiveCertificateValidator(),
+                )
+            val supervisor = SupervisorJob()
+            val pumpScope = CoroutineScope(supervisor + Dispatchers.IO)
+
+            val capturesLock = Any()
+            val captures = mutableListOf<DriverCapture>()
+            val captureFactory: () -> AudioCapture = {
+                val c = DriverCapture()
+                synchronized(capturesLock) { captures += c }
+                c
+            }
+            val encoder = StubEncoder(prefix = "LSP-".encodeToByteArray())
+
+            val speakerOpenCount = AtomicInteger(0)
+
+            val scope = "listener-survives-publisher-recycle"
+            try {
+                val speaker =
+                    InteropDebug.stepSuspending(scope, "connectReconnectingNestsSpeaker (refresh=${PUBCYCLE_REFRESH_MS}ms)") {
+                        connectReconnectingNestsSpeaker(
+                            httpClient = httpClient,
+                            transport = transport,
+                            scope = pumpScope,
+                            room = room,
+                            signer = signer,
+                            speakerPubkeyHex = pubkey,
+                            captureFactory = captureFactory,
+                            encoderFactory = { encoder },
+                            policy = NestsReconnectPolicy(initialDelayMs = 250L),
+                            tokenRefreshAfterMs = PUBCYCLE_REFRESH_MS,
+                            connector = {
+                                speakerOpenCount.incrementAndGet()
+                                connectNestsSpeaker(
+                                    httpClient = httpClient,
+                                    transport = transport,
+                                    scope = pumpScope,
+                                    room = room,
+                                    signer = signer,
+                                    speakerPubkeyHex = pubkey,
+                                    captureFactory = captureFactory,
+                                    encoderFactory = { encoder },
+                                )
+                            },
+                        )
+                    }
+                val broadcast = speaker.startBroadcasting()
+                withTimeoutOrNull(BROADCAST_READY_MS) {
+                    speaker.state.first { it is NestsSpeakerState.Broadcasting }
+                } ?: fail("[$scope] speaker never reached initial Broadcasting")
+
+                // SUT: reconnecting listener with refresh disabled.
+                val listener =
+                    InteropDebug.stepSuspending(scope, "connectReconnectingNestsListener (refresh disabled)") {
+                        connectReconnectingNestsListener(
+                            httpClient = httpClient,
+                            transport = transport,
+                            scope = pumpScope,
+                            room = room,
+                            signer = signer,
+                            policy = NestsReconnectPolicy(initialDelayMs = 250L),
+                            tokenRefreshAfterMs = 0L,
+                        )
+                    }
+                withTimeoutOrNull(CONNECT_TIMEOUT_MS) {
+                    listener.state.first { it is NestsListenerState.Connected }
+                } ?: fail("[$scope] listener never reached Connected")
+
+                // The single subscription that MUST survive every
+                // publisher recycle.
+                val subscription = listener.subscribeSpeaker(pubkey)
+                val received =
+                    async(pumpScope.coroutineContext) {
+                        withTimeoutOrNull(LISTENER_SURVIVAL_TIMEOUT_MS) {
+                            subscription.objects.take(N_FRAMES_CYCLE).toList()
+                        }
+                    }
+                kotlinx.coroutines.delay(SUBSCRIBE_SETTLE_MS)
+
+                for (i in 0 until N_FRAMES_CYCLE) {
+                    val current =
+                        synchronized(capturesLock) { captures.lastOrNull() }
+                            ?: error("captureFactory was never invoked")
+                    current.push(shortArrayOf(i.toShort()))
+                    kotlinx.coroutines.delay(CYCLE_FRAME_SPACING_MS)
+                    if (i == N_FRAMES_CYCLE / 2) {
+                        InteropDebug.checkpoint(scope, "midpoint — waiting for speaker recycle")
+                        withTimeoutOrNull(SWAP_TIMEOUT_MS) {
+                            while (speakerOpenCount.get() < 2) kotlinx.coroutines.delay(50)
+                            speaker.state.first { it is NestsSpeakerState.Broadcasting }
+                        } ?: fail("[$scope] speaker did not recycle — openCount=${speakerOpenCount.get()}")
+                        kotlinx.coroutines.delay(POST_RECYCLE_SETTLE_MS)
+                    }
+                }
+
+                val datagrams = received.await()
+                if (datagrams == null) {
+                    fail(
+                        "[$scope] listener subscription went silent across publisher recycle — " +
+                            "speakerOpenCount=${speakerOpenCount.get()}, " +
+                            "listener=${InteropDebug.describe(listener.state.value)}",
+                    )
+                }
+                assertEquals(N_FRAMES_CYCLE, datagrams.size, "all frames must arrive on the SAME subscribe handle across the publisher recycle")
+                val payloads = datagrams.map { it.payload.last().toInt() and 0xFF }.toSet()
+                assertEquals((0 until N_FRAMES_CYCLE).toSet(), payloads, "all frames pre- AND post-recycle must round-trip")
+                assertTrue(
+                    speakerOpenCount.get() >= 2,
+                    "expected ≥2 underlying speaker sessions across the burst (one before, one after recycle); got ${speakerOpenCount.get()}",
+                )
+
+                runCatching { subscription.unsubscribe() }
+                runCatching { listener.close() }
+                runCatching { broadcast.close() }
+                runCatching { speaker.close() }
+            } finally {
+                synchronized(capturesLock) { captures.forEach { runCatching { it.stop() } } }
+                supervisor.cancelAndJoin()
+            }
+            Unit
+        }
 }
+
+private const val PUBCYCLE_REFRESH_MS = 4_000L
+private const val BROADCAST_READY_MS = 15_000L
+private const val LISTENER_SURVIVAL_TIMEOUT_MS = 60_000L
+private const val POST_RECYCLE_SETTLE_MS = 1_500L
+private const val CYCLE_FRAME_SPACING_MS = 80L
+private const val N_FRAMES_CYCLE = 10
+private const val SWAP_TIMEOUT_MS = 60_000L

nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsReconnectingSpeakerInteropTest.kt

@@ -0,0 +1,514 @@
+/*
+ * 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.nestsclient.interop
+
+import com.vitorpamplona.nestsclient.NestsListenerState
+import com.vitorpamplona.nestsclient.NestsReconnectPolicy
+import com.vitorpamplona.nestsclient.NestsRoomConfig
+import com.vitorpamplona.nestsclient.NestsSpeakerState
+import com.vitorpamplona.nestsclient.OkHttpNestsClient
+import com.vitorpamplona.nestsclient.audio.AudioCapture
+import com.vitorpamplona.nestsclient.audio.OpusEncoder
+import com.vitorpamplona.nestsclient.connectNestsListener
+import com.vitorpamplona.nestsclient.connectNestsSpeaker
+import com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker
+import com.vitorpamplona.nestsclient.transport.QuicWebTransportFactory
+import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
+import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
+import com.vitorpamplona.quic.tls.PermissiveCertificateValidator
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.async
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.take
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeoutOrNull
+import org.junit.AfterClass
+import org.junit.BeforeClass
+import org.junit.Test
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+import kotlin.test.fail
+
+/**
+ * Interop test for [connectReconnectingNestsSpeaker] against the real
+ * nostrnests stack. The speaker side mirrors the listener's
+ * [NostrNestsReconnectingListenerInteropTest] but exercises the
+ * publish path:
+ *
+ *   1. **Happy path** — the wrapper drives a single real session,
+ *      Opus frames flow listener-side. Sanity-check that the
+ *      broadcast pump didn't break the round-trip vs the bare
+ *      [connectNestsSpeaker] path.
+ *
+ *   2. **Forced JWT refresh** — small `tokenRefreshAfterMs` forces
+ *      the orchestrator to recycle the underlying speaker mid-stream.
+ *      Frames before the recycle and after the recycle MUST both
+ *      land on the same listener-side [SubscribeHandle] — that's
+ *      the load-bearing postcondition for the production
+ *      540 s ↔ 600 s JWT-TTL relationship. Also verifies that the
+ *      wrapper's outward state never dips into Reconnecting /
+ *      Failed during a clean refresh.
+ *
+ * Skipped by default — set `-DnestsInterop=true` to enable.
+ */
+class NostrNestsReconnectingSpeakerInteropTest {
+    @Test
+    fun reconnecting_speaker_round_trips_frames_via_real_relay() =
+        runBlocking {
+            NostrNestsHarness.assumeNestsInterop()
+            val harness = harnessOrNull ?: return@runBlocking
+
+            val signer = NostrSignerInternal(KeyPair())
+            val pubkey = signer.pubKey
+            val room =
+                NestsRoomConfig(
+                    authBaseUrl = harness.authBaseUrl,
+                    endpoint = harness.moqEndpoint,
+                    hostPubkey = pubkey,
+                    roomId = "spk-rec-${System.currentTimeMillis()}",
+                )
+
+            val httpClient = OkHttpNestsClient()
+            val transport =
+                QuicWebTransportFactory(
+                    certificateValidator = PermissiveCertificateValidator(),
+                )
+            val supervisor = SupervisorJob()
+            val pumpScope = CoroutineScope(supervisor + Dispatchers.IO)
+
+            // captureFactory is invoked once per underlying session
+            // (broadcaster.start opens a fresh capture) — we keep
+            // every instance in `captures` so the test can push
+            // frames into whichever one is currently live.
+            val capturesLock = Any()
+            val captures = mutableListOf<DriverCapture>()
+            val captureFactory: () -> AudioCapture = {
+                val c = DriverCapture()
+                synchronized(capturesLock) { captures += c }
+                c
+            }
+            val encoder = StubEncoder(prefix = "SREC-".encodeToByteArray())
+
+            val scope = "reconnecting-speaker-happy-path"
+            try {
+                val reconnecting =
+                    InteropDebug.stepSuspending(scope, "connectReconnectingNestsSpeaker") {
+                        connectReconnectingNestsSpeaker(
+                            httpClient = httpClient,
+                            transport = transport,
+                            scope = pumpScope,
+                            room = room,
+                            signer = signer,
+                            speakerPubkeyHex = pubkey,
+                            captureFactory = captureFactory,
+                            encoderFactory = { encoder },
+                            policy = NestsReconnectPolicy(initialDelayMs = 250L),
+                            // Disable proactive refresh — happy path
+                            // exercises a single session only.
+                            tokenRefreshAfterMs = 0L,
+                        )
+                    }
+                InteropDebug.assertSpeakerReached(scope, "Connected", reconnecting.state.value)
+
+                val broadcast =
+                    InteropDebug.stepSuspending(scope, "reconnecting.startBroadcasting") {
+                        reconnecting.startBroadcasting()
+                    }
+
+                // Wait for the wrapper to land Broadcasting before we
+                // start pushing — the broadcast pump's underlying
+                // startBroadcasting needs to run first for the relay
+                // to serve the announce.
+                withTimeoutOrNull(BROADCAST_TIMEOUT_MS) {
+                    reconnecting.state.first { it is NestsSpeakerState.Broadcasting }
+                } ?: fail("[$scope] reconnecting wrapper never reached Broadcasting")
+
+                // Vanilla listener — just consumes our published frames.
+                val listener =
+                    InteropDebug.stepSuspending(scope, "connectNestsListener") {
+                        connectNestsListener(
+                            httpClient = httpClient,
+                            transport = transport,
+                            scope = pumpScope,
+                            room = room,
+                            signer = signer,
+                        )
+                    }
+                InteropDebug.assertListenerReached(scope, "Connected", listener.state.value)
+
+                val subscription =
+                    InteropDebug.stepSuspending(scope, "listener.subscribeSpeaker(pubkey)") {
+                        listener.subscribeSpeaker(pubkey)
+                    }
+
+                val received =
+                    async(pumpScope.coroutineContext) {
+                        withTimeoutOrNull(RECEIVE_TIMEOUT_MS) {
+                            subscription.objects.take(N_FRAMES).toList()
+                        }
+                    }
+
+                delay(SUBSCRIBE_SETTLE_MS)
+
+                for (i in 0 until N_FRAMES) {
+                    pushTo(captures, capturesLock, shortArrayOf(i.toShort()))
+                    delay(FRAME_SPACING_MS)
+                }
+
+                val datagrams = received.await()
+                if (datagrams == null) {
+                    fail(
+                        "[$scope] did not receive $N_FRAMES frames within ${RECEIVE_TIMEOUT_MS}ms — " +
+                            "wrapper=${InteropDebug.describe(reconnecting.state.value)}, " +
+                            "listener=${InteropDebug.describe(listener.state.value)}",
+                    )
+                }
+                assertEquals(N_FRAMES, datagrams.size, "expected exactly $N_FRAMES frames")
+                val payloads = datagrams.map { it.payload.last().toInt() and 0xFF }.toSet()
+                assertEquals((0 until N_FRAMES).toSet(), payloads, "all unique frame indices round-tripped")
+
+                runCatching { subscription.unsubscribe() }
+                runCatching { listener.close() }
+                runCatching { broadcast.close() }
+                runCatching { reconnecting.close() }
+            } finally {
+                synchronized(capturesLock) { captures.forEach { runCatching { it.stop() } } }
+                supervisor.cancelAndJoin()
+            }
+            Unit
+        }
+
+    @Test
+    fun reconnecting_speaker_recycles_session_on_jwt_refresh_without_dropping_frames() =
+        runBlocking {
+            NostrNestsHarness.assumeNestsInterop()
+            val harness = harnessOrNull ?: return@runBlocking
+
+            val signer = NostrSignerInternal(KeyPair())
+            val pubkey = signer.pubKey
+            val room =
+                NestsRoomConfig(
+                    authBaseUrl = harness.authBaseUrl,
+                    endpoint = harness.moqEndpoint,
+                    hostPubkey = pubkey,
+                    roomId = "spk-refr-${System.currentTimeMillis()}",
+                )
+
+            val httpClient = OkHttpNestsClient()
+            val transport =
+                QuicWebTransportFactory(
+                    certificateValidator = PermissiveCertificateValidator(),
+                )
+            val supervisor = SupervisorJob()
+            val pumpScope = CoroutineScope(supervisor + Dispatchers.IO)
+
+            val capturesLock = Any()
+            val captures = mutableListOf<DriverCapture>()
+            val captureFactory: () -> AudioCapture = {
+                val c = DriverCapture()
+                synchronized(capturesLock) { captures += c }
+                c
+            }
+            val encoder = StubEncoder(prefix = "REFR-".encodeToByteArray())
+
+            // Track every state the wrapper surfaces. The
+            // load-bearing JWT-refresh invariant: outward state
+            // must NEVER show Reconnecting / Failed during a
+            // clean recycle. The list is snapshotted under a
+            // lock before the assertion to dodge CME from the
+            // concurrent `state.collect` writer.
+            val seenStatesLock = Any()
+            val seenStates: MutableList<NestsSpeakerState> = mutableListOf()
+
+            // Connector counts how many real sessions the
+            // orchestrator opened. ≥2 is the marker that the
+            // proactive refresh path actually fired.
+            val openCount = AtomicInteger(0)
+
+            val scope = "reconnecting-speaker-jwt-refresh"
+            try {
+                val reconnecting =
+                    InteropDebug.stepSuspending(scope, "connectReconnectingNestsSpeaker (refresh=${REFRESH_WINDOW_MS}ms)") {
+                        connectReconnectingNestsSpeaker(
+                            httpClient = httpClient,
+                            transport = transport,
+                            scope = pumpScope,
+                            room = room,
+                            signer = signer,
+                            speakerPubkeyHex = pubkey,
+                            captureFactory = captureFactory,
+                            encoderFactory = { encoder },
+                            policy = NestsReconnectPolicy(initialDelayMs = 250L),
+                            tokenRefreshAfterMs = REFRESH_WINDOW_MS,
+                            connector = {
+                                openCount.incrementAndGet()
+                                connectNestsSpeaker(
+                                    httpClient = httpClient,
+                                    transport = transport,
+                                    scope = pumpScope,
+                                    room = room,
+                                    signer = signer,
+                                    speakerPubkeyHex = pubkey,
+                                    captureFactory = captureFactory,
+                                    encoderFactory = { encoder },
+                                )
+                            },
+                        )
+                    }
+
+                // Watch wrapper state for the no-Reconnecting/Failed
+                // assertion below.
+                val watcher =
+                    pumpScope.launch {
+                        reconnecting.state.collect { st ->
+                            synchronized(seenStatesLock) { seenStates += st }
+                        }
+                    }
+
+                val broadcast =
+                    InteropDebug.stepSuspending(scope, "reconnecting.startBroadcasting") {
+                        reconnecting.startBroadcasting()
+                    }
+
+                withTimeoutOrNull(BROADCAST_TIMEOUT_MS) {
+                    reconnecting.state.first { it is NestsSpeakerState.Broadcasting }
+                } ?: fail("[$scope] never reached initial Broadcasting")
+
+                // Phase 1: validate frames flow on the FIRST
+                // session. Open a vanilla listener, subscribe, push
+                // a small batch, confirm round-trip. This proves
+                // the wrapper's first-session publish path is sound
+                // before we induce the refresh.
+                val firstListener =
+                    InteropDebug.stepSuspending(scope, "connectNestsListener (pre-refresh)") {
+                        connectNestsListener(
+                            httpClient = httpClient,
+                            transport = transport,
+                            scope = pumpScope,
+                            room = room,
+                            signer = signer,
+                        )
+                    }
+                withTimeoutOrNull(CONNECT_TIMEOUT_MS) {
+                    firstListener.state.first { it is NestsListenerState.Connected }
+                } ?: fail("[$scope] first listener never reached Connected")
+                val firstSub = firstListener.subscribeSpeaker(pubkey)
+                val preFrames =
+                    async(pumpScope.coroutineContext) {
+                        withTimeoutOrNull(RECEIVE_TIMEOUT_MS) {
+                            firstSub.objects.take(HALF_FRAMES).toList()
+                        }
+                    }
+                delay(SUBSCRIBE_SETTLE_MS)
+                for (i in 0 until HALF_FRAMES) {
+                    pushTo(captures, capturesLock, shortArrayOf(i.toShort()))
+                    delay(FRAME_SPACING_MS)
+                }
+                val pre =
+                    preFrames.await() ?: fail(
+                        "[$scope] pre-refresh frames did not arrive — wrapper=${InteropDebug.describe(reconnecting.state.value)}",
+                    )
+                assertEquals(HALF_FRAMES, pre.size, "all pre-refresh frames must round-trip on the first session")
+                runCatching { firstSub.unsubscribe() }
+                runCatching { firstListener.close() }
+
+                // Phase 2: wait for the proactive refresh to fire.
+                // The orchestrator's withTimeoutOrNull fires at
+                // REFRESH_WINDOW_MS, closes the underlying speaker,
+                // reopens via the connector (openCount→2).
+                InteropDebug.checkpoint(scope, "waiting for proactive refresh")
+                withTimeoutOrNull(SWAP_TIMEOUT_MS) {
+                    while (openCount.get() < 2) delay(50)
+                    // Wrapper outward state may briefly dip to
+                    // Connected during cutover; wait for the
+                    // second Broadcasting so the new session is
+                    // serving the announce.
+                    reconnecting.state.first { it is NestsSpeakerState.Broadcasting }
+                } ?: fail(
+                    "[$scope] speaker did not recycle — openCount=${openCount.get()}, " +
+                        "wrapper=${InteropDebug.describe(reconnecting.state.value)}",
+                )
+
+                delay(POST_SWAP_SETTLE_MS)
+
+                // Phase 3: validate frames flow on the SECOND
+                // (post-refresh) session. Open a FRESH listener +
+                // subscription so we exercise the new publisher
+                // directly. (A pre-recycle subscription would be
+                // bound to the dead session — that listener-survival-
+                // across-publisher-recycle is a separate concern, not
+                // a speaker-reconnect bug.)
+                val secondListener =
+                    InteropDebug.stepSuspending(scope, "connectNestsListener (post-refresh)") {
+                        connectNestsListener(
+                            httpClient = httpClient,
+                            transport = transport,
+                            scope = pumpScope,
+                            room = room,
+                            signer = signer,
+                        )
+                    }
+                withTimeoutOrNull(CONNECT_TIMEOUT_MS) {
+                    secondListener.state.first { it is NestsListenerState.Connected }
+                } ?: fail("[$scope] second listener never reached Connected")
+                val secondSub = secondListener.subscribeSpeaker(pubkey)
+                val postFrames =
+                    async(pumpScope.coroutineContext) {
+                        withTimeoutOrNull(RECEIVE_TIMEOUT_MS) {
+                            secondSub.objects.take(HALF_FRAMES).toList()
+                        }
+                    }
+                delay(SUBSCRIBE_SETTLE_MS)
+                for (i in HALF_FRAMES until N_FRAMES_SWAP) {
+                    pushTo(captures, capturesLock, shortArrayOf(i.toShort()))
+                    delay(FRAME_SPACING_MS)
+                }
+                val post =
+                    postFrames.await() ?: fail(
+                        "[$scope] post-refresh frames did not arrive on the new session — " +
+                            "openCount=${openCount.get()}, wrapper=${InteropDebug.describe(reconnecting.state.value)}",
+                    )
+                assertEquals(HALF_FRAMES, post.size, "all post-refresh frames must round-trip on the new session")
+
+                // Wrapper-side invariant: outward state must NEVER
+                // have surfaced Reconnecting/Failed during the
+                // refresh — that's the load-bearing user-visible
+                // promise of proactive JWT recycle.
+                watcher.cancel()
+                watcher.join()
+                val snapshot = synchronized(seenStatesLock) { seenStates.toList() }
+                val sawReconnecting = snapshot.any { it is NestsSpeakerState.Reconnecting }
+                val sawFailed = snapshot.any { it is NestsSpeakerState.Failed }
+                assertTrue(
+                    !sawReconnecting && !sawFailed,
+                    "[$scope] proactive refresh must not surface Reconnecting/Failed; saw=$snapshot",
+                )
+
+                assertTrue(
+                    openCount.get() >= 2,
+                    "expected ≥2 underlying speaker sessions (one before refresh, one after); got ${openCount.get()}",
+                )
+
+                runCatching { broadcast.close() }
+                runCatching { reconnecting.close() }
+            } finally {
+                synchronized(capturesLock) { captures.forEach { runCatching { it.stop() } } }
+                supervisor.cancelAndJoin()
+            }
+            Unit
+        }
+
+    /**
+     * Push a PCM frame into whichever live capture the broadcast
+     * pump is currently using. The factory hands out a fresh
+     * [DriverCapture] per session (the production speaker calls
+     * `captureFactory()` from inside `startBroadcasting`), so the
+     * "current" capture is always the most recently created one
+     * — older captures correspond to recycled sessions whose
+     * channels have been closed by [DriverCapture.stop].
+     */
+    private fun pushTo(
+        captures: MutableList<DriverCapture>,
+        lock: Any,
+        pcm: ShortArray,
+    ) {
+        val current =
+            synchronized(lock) { captures.lastOrNull() }
+                ?: error("captureFactory was never invoked — no live capture to push to")
+        current.push(pcm)
+    }
+
+    /** Channel-driven capture seam — same shape the round-trip test uses. */
+    private class DriverCapture : AudioCapture {
+        private val frames = Channel<ShortArray>(capacity = Channel.UNLIMITED)
+
+        @Volatile private var started: Boolean = false
+
+        override fun start() {
+            started = true
+        }
+
+        override suspend fun readFrame(): ShortArray? {
+            if (!started) return null
+            return frames.receiveCatching().getOrNull()
+        }
+
+        override fun stop() {
+            frames.close()
+        }
+
+        fun push(pcm: ShortArray) {
+            frames.trySend(pcm)
+        }
+    }
+
+    private class StubEncoder(
+        private val prefix: ByteArray,
+    ) : OpusEncoder {
+        override fun encode(pcm: ShortArray): ByteArray = prefix + byteArrayOf(pcm.first().toByte())
+
+        override fun release() = Unit
+    }
+
+    companion object {
+        private const val N_FRAMES = 6
+        private const val N_FRAMES_SWAP = 8
+        private const val HALF_FRAMES = 3
+        private const val SUBSCRIBE_SETTLE_MS = 500L
+        private const val POST_SWAP_SETTLE_MS = 1_500L
+        private const val FRAME_SPACING_MS = 50L
+
+        // Refresh window — small enough that the orchestrator's
+        // proactive recycle fires after the first half-batch but
+        // before the second; large enough to leave headroom for
+        // the WebTransport handshake + first batch of frame
+        // pushes (~450 ms on a warm Docker stack).
+        private const val REFRESH_WINDOW_MS = 4_000L
+        private const val CONNECT_TIMEOUT_MS = 10_000L
+        private const val BROADCAST_TIMEOUT_MS = 15_000L
+        private const val RECEIVE_TIMEOUT_MS = 15_000L
+        private const val SWAP_TIMEOUT_MS = 60_000L
+
+        private var harnessOrNull: NostrNestsHarness? = null
+
+        @BeforeClass
+        @JvmStatic
+        fun setUpHarness() {
+            if (NostrNestsHarness.isEnabled()) {
+                harnessOrNull = NostrNestsHarness.shared()
+            }
+        }
+
+        @AfterClass
+        @JvmStatic
+        fun tearDownHarness() {
+            harnessOrNull = null
+        }
+    }
+}

nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsRoundTripInteropTest.kt

@@ -200,9 +200,13 @@ class NostrNestsRoundTripInteropTest {
                     assertEquals(
                         idx.toLong(),
                         obj.objectId,
-                        "object id at index $idx — MoQ requires monotonic ids per group",
+                        "object id at index $idx — MoqLiteNestsListener uses a session-level counter, monotonic across all groups",
+                    )
+                    assertEquals(
+                        idx.toLong(),
+                        obj.groupId,
+                        "group id at index $idx — broadcaster emits one moq-lite group per Opus frame (audio-rooms NIP draft) so groupId increments 1:1 with the frame index",
                     )
-                    assertEquals(0L, obj.groupId, "single-group track per audio-rooms NIP draft")
                     assertContentEquals(
                         "FRAME-".encodeToByteArray() + byteArrayOf(idx.toByte()),
                         obj.payload,