fix(nests): listener subscriptions survive publisher recycle

Two layered fixes for the gap captured in
nestsClient/plans/2026-04-28-listener-survives-publisher-recycle.md.

Session layer (MoqLiteSession.kt):
  - Lazy single shared announce-watch pump per session, opened on
    first subscribe. 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 Ended event
    is the only reliable signal.
  - On Announce(Ended) for a broadcast suffix, close the
    matching ListenerSubscription's frames Channel and remove it
    from the map. The wrapper-level frames.consumeAsFlow() flow
    ends naturally — same shape as a user-driven
    handle.unsubscribe() — so the wrapper pump's collect-
    completion path drives the re-issue.

Wrapper layer (ReconnectingNestsListener.kt):
  - Inner re-subscribe `while (currentCoroutineContext().isActive)`
    loop in reissuingSubscribe. When the underlying frames flow
    completes (publisher cycled, signalled by the session layer
    above), re-issue subscribe against the same listener with a
    100 ms backoff. moq-lite supports subscribe-before-announce so
    a re-subscribe issued during the gap 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.

In production, this closes the audio dropout that would have
fired every 9 minutes per speaker JWT refresh on long Nest calls
(see the plan doc for the original diagnosis).

https://claude.ai/code/session_01HXf3zG3F2ev2ASeQju7Y5S

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
@@ -263,9 +265,38 @@ private class ReconnectingHandle(
             )
         val liveHandleRef = AtomicReference<SubscribeHandle?>(null)
 
-        // Re-subscribe pump: every time activeListener changes, drop
+        // Re-subscribe pump. Two re-issue triggers, layered:
-        // the prior subscription (collectLatest cancels the inner
+        //
-        // body) and open a new one against the fresh session.
+        //   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 +308,23 @@ private class ReconnectingHandle(
                                 state is NestsListenerState.Failed
                         }
                     if (terminalOrConnected !is NestsListenerState.Connected) return@collectLatest
-                    val handle =
+
-                        runCatching { opener(listener) }
+                    while (currentCoroutineContext().isActive) {
-                            .getOrNull() ?: return@collectLatest
+                        val handle =
-                    liveHandleRef.set(handle)
+                            runCatching { opener(listener) }
-                    try {
+                                .getOrNull() ?: break
-                        handle.objects.collect { frames.emit(it) }
+                        liveHandleRef.set(handle)
-                    } finally {
+                        try {
-                        if (liveHandleRef.get() === handle) liveHandleRef.set(null)
+                            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 +358,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/moq/lite/MoqLiteSession.kt

@@ -77,6 +77,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
 
@@ -173,6 +184,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,
@@ -202,6 +234,29 @@ class MoqLiteSession internal constructor(
                 state.withLock {
                     subscriptionsBySubscribeId[id] = sub
                     if (groupPump == null) groupPump = scope.launch { pumpUniStreams() }
+                    // Lazy-launch the publisher-disconnect watcher
+                    // — one shared announce bidi per session whose
+                    // sole job is to close the frames channel on any
+                    // ListenerSubscription whose broadcast path goes
+                    // Ended. 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 IS the only signal.
+                    //
+                    // Without this, a wrapper-layer consumer
+                    // collecting from [MoqLiteSubscribeHandle.frames]
+                    // would sit silent indefinitely after a
+                    // publisher cycle even though the relay is happy
+                    // to serve a fresh subscribe under the same
+                    // suffix. The wrapper-level
+                    // [com.vitorpamplona.nestsclient.ReconnectingNestsListener]
+                    // pump's natural collect-completion path then
+                    // re-issues the subscribe.
+                    if (announceWatchJob == null) {
+                        announceWatchJob = scope.launch { pumpAnnounceWatch() }
+                    }
                 }
                 return MoqLiteSubscribeHandle(
                     id = id,
@@ -213,6 +268,71 @@ class MoqLiteSession internal constructor(
         }
     }
 
+    /**
+     * Single shared announce-watch pump for ALL subscriptions on
+     * this session. Opens one announce bidi (prefix="") on first
+     * subscribe, then 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() {
+        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.
+                return
+            }
+        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() }
+            state.withLock { announceWatchJob = null }
+        }
+    }
+
     /**
      * Drain inbound uni streams and route each one's group frames to
      * the matching subscription. The relay opens a fresh uni stream

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