debug(nests): wire NestRx/NestTx logs across listener and speaker paths

Diagnostic instrumentation to localise the listener-stuck-on-spinner bug.
None of these are intended for merge — pair logs with a logcat capture,
diagnose, then revert.

Receiver (tag NestRx):
  - MoqLiteNestsListener: log subscribeSpeaker / subscribeCatalog entry
    and per-update RoomAnnouncement emission to the VM
  - MoqLiteSession.subscribe: log SUBSCRIBE_OK / SUBSCRIBE_DROP outcomes
  - MoqLiteSession.drainOneGroup: log every uni group header decode
    and warn on subscription-lookup miss (frame dropped)
  - MoqLiteSession.pumpAnnounceWatch: log every announce update and
    flag the Ended branch that closes the subscription's frames
  - NestViewModel: log VM.openSubscription wiring and the first-frame
    trigger that clears the connectingSpeakers spinner

Speaker (tag NestTx):
  - MoqLiteSession.publish: log entry suffix
  - MoqLiteSession.handleInboundBidi: log inbound AnnouncePlease and
    SUBSCRIBE dispatches plus FIN cleanup
  - MoqLiteSession PublisherStateImpl: log first inbound subscriber
    attach and every group-stream open
  - NestMoqLiteBroadcaster: log when publisher.send returns false
    (no inbound subscriber on relay) and the subscriber-attached
    transition, plus surface throws that runCatching previously
    swallowed only into onError

Capture with `adb logcat -s NestRx:D NestTx:D`.

https://claude.ai/code/session_01PYYez8a6sjiakyjAxsfCEQ

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

@@ -44,6 +44,7 @@ import com.vitorpamplona.nestsclient.moq.SubscribeHandle
 import com.vitorpamplona.nestsclient.transport.WebTransportFactory
 import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
 import com.vitorpamplona.quartz.nip53LiveActivities.chat.LiveActivitiesChatMessageEvent
+import com.vitorpamplona.quartz.utils.Log
 import kotlinx.collections.immutable.ImmutableSet
 import kotlinx.collections.immutable.persistentSetOf
 import kotlinx.collections.immutable.toPersistentSet
@@ -842,7 +843,9 @@ class NestViewModel(
     ) {
         if (closed || activeSubscriptions[pubkey] !== slot) return
         try {
+            Log.d("NestRx") { "VM.openSubscription -> subscribeSpeaker pubkey=$pubkey" }
             val handle = l.subscribeSpeaker(pubkey)
+            Log.d("NestRx") { "VM.openSubscription <- subscribeSpeaker returned for pubkey=$pubkey" }
             // Re-check after the suspending subscribeSpeaker — the user
             // may have removed this speaker via updateSpeakers / disconnected
             // while the SUBSCRIBE was in flight. If so, abandon the handle
@@ -1004,6 +1007,7 @@ class NestViewModel(
         // First frame for this subscription — clear the buffering
         // overlay. Subsequent frames are no-ops here.
         if (_uiState.value.connectingSpeakers.contains(pubkey)) {
+            Log.d("NestRx") { "VM.onSpeakerActivity FIRST frame for pubkey=$pubkey — clearing spinner" }
             _uiState.update { it.copy(connectingSpeakers = (it.connectingSpeakers - pubkey).toPersistentSet()) }
         }
         if (!_uiState.value.speakingNow.contains(pubkey)) {

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

@@ -26,6 +26,7 @@ import com.vitorpamplona.nestsclient.moq.SubscribeOk
 import com.vitorpamplona.nestsclient.moq.lite.MoqLiteAnnounceStatus
 import com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession
 import com.vitorpamplona.nestsclient.moq.lite.MoqLiteSubscribeException
+import com.vitorpamplona.quartz.utils.Log
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
@@ -78,9 +79,15 @@ class MoqLiteNestsListener internal constructor(
     override suspend fun subscribeSpeaker(
         speakerPubkeyHex: String,
         maxLatencyMs: Long,
-    ): SubscribeHandle = wrapSubscription(broadcast = speakerPubkeyHex, track = AUDIO_TRACK, maxLatencyMs = maxLatencyMs)
+    ): SubscribeHandle {
+        Log.d("NestRx") { "subscribeSpeaker pubkey=$speakerPubkeyHex maxLatencyMs=$maxLatencyMs" }
+        return wrapSubscription(broadcast = speakerPubkeyHex, track = AUDIO_TRACK, maxLatencyMs = maxLatencyMs)
+    }
 
-    override suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle = wrapSubscription(broadcast = speakerPubkeyHex, track = CATALOG_TRACK, maxLatencyMs = 0L)
+    override suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle {
+        Log.d("NestRx") { "subscribeCatalog pubkey=$speakerPubkeyHex" }
+        return wrapSubscription(broadcast = speakerPubkeyHex, track = CATALOG_TRACK, maxLatencyMs = 0L)
+    }
 
     private suspend fun wrapSubscription(
         broadcast: String,
@@ -138,10 +145,14 @@ class MoqLiteNestsListener internal constructor(
             val handle = session.announce(prefix = "")
             try {
                 handle.updates.collect { announce ->
+                    val active = announce.status == MoqLiteAnnounceStatus.Active
+                    Log.d("NestRx") {
+                        "RoomAnnouncement pubkey=${announce.suffix} active=$active status=${announce.status} hops=${announce.hops}"
+                    }
                     emit(
                         RoomAnnouncement(
                             pubkey = announce.suffix,
-                            active = announce.status == MoqLiteAnnounceStatus.Active,
+                            active = active,
                         ),
                     )
                 }

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

@@ -21,6 +21,7 @@
 package com.vitorpamplona.nestsclient.audio
 
 import com.vitorpamplona.nestsclient.moq.lite.MoqLitePublisherHandle
+import com.vitorpamplona.quartz.utils.Log
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
@@ -161,6 +162,8 @@ class NestMoqLiteBroadcaster(
                 // we bail. publisher.send returning `false` (no inbound
                 // subscriber) is NOT counted — empty rooms are normal.
                 var consecutiveSendErrors = 0
+                var sendNoSubLogged = false
+                var lastSendHadSub = false
                 try {
                     while (true) {
                         val pcm = capture.readFrame() ?: break
@@ -187,7 +190,21 @@ class NestMoqLiteBroadcaster(
                         // for the production cliff this works around.
                         val sendOutcome =
                             runCatching {
-                                publisher.send(opus)
+                                val accepted = publisher.send(opus)
+                                if (accepted) {
+                                    if (!lastSendHadSub) {
+                                        Log.d("NestTx") { "broadcaster: send accepted (subscriber attached)" }
+                                        lastSendHadSub = true
+                                    }
+                                } else {
+                                    if (!sendNoSubLogged) {
+                                        Log.w("NestTx") {
+                                            "broadcaster: publisher.send returned false (no inbound subscriber on relay)"
+                                        }
+                                        sendNoSubLogged = true
+                                    }
+                                    lastSendHadSub = false
+                                }
                                 framesInCurrentGroup += 1
                                 if (framesInCurrentGroup >= framesPerGroup) {
                                     publisher.endGroup()
@@ -206,6 +223,9 @@ class NestMoqLiteBroadcaster(
                             }.onFailure { t ->
                                 if (t is CancellationException) throw t
                                 consecutiveSendErrors += 1
+                                Log.w("NestTx") {
+                                    "broadcaster: publisher.send threw (#$consecutiveSendErrors): ${t::class.simpleName}: ${t.message}"
+                                }
                                 onError(
                                     AudioException(
                                         AudioException.Kind.PlaybackFailed,

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

@@ -23,6 +23,7 @@ package com.vitorpamplona.nestsclient.moq.lite
 import com.vitorpamplona.nestsclient.moq.MoqCodecException
 import com.vitorpamplona.nestsclient.moq.MoqWriter
 import com.vitorpamplona.nestsclient.transport.WebTransportSession
+import com.vitorpamplona.quartz.utils.Log
 import com.vitorpamplona.quic.Varint
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CompletableDeferred
@@ -298,6 +299,10 @@ class MoqLiteSession internal constructor(
             }
         when (resp) {
             is MoqLiteCodec.SubscribeResponse.Dropped -> {
+                Log.w("NestRx") {
+                    "SUBSCRIBE_DROP id=$id broadcast='$broadcast' track='$track' " +
+                        "errCode=${resp.drop.errorCode} reason='${resp.drop.reasonPhrase}'"
+                }
                 state.withLock { subscriptionsBySubscribeId.remove(id) }
                 frames.close()
                 runCatching { bidi.finish() }
@@ -308,6 +313,7 @@ class MoqLiteSession internal constructor(
             }
 
             is MoqLiteCodec.SubscribeResponse.Ok -> {
+                Log.d("NestRx") { "SUBSCRIBE_OK id=$id broadcast='$broadcast' track='$track'" }
                 return MoqLiteSubscribeHandle(
                     id = id,
                     ok = resp.ok,
@@ -383,6 +389,9 @@ class MoqLiteSession internal constructor(
     private suspend fun pumpAnnounceWatch(handle: MoqLiteAnnouncesHandle) {
         try {
             handle.updates.collect { update ->
+                Log.d("NestRx") {
+                    "announce update status=${update.status} suffix='${update.suffix}' hops=${update.hops}"
+                }
                 if (update.status != MoqLiteAnnounceStatus.Ended) return@collect
                 val targets =
                     state.withLock {
@@ -391,6 +400,9 @@ class MoqLiteSession internal constructor(
                             .toList()
                     }
                 for (sub in targets) {
+                    Log.w("NestRx") {
+                        "announce ENDED closes sub id=${sub.id} broadcast='${sub.request.broadcast}'"
+                    }
                     // Just close the frames channel — the
                     // wrapper-level collect of `frames.consumeAsFlow()`
                     // ends naturally and the wrapper pump re-issues.
@@ -466,6 +478,7 @@ class MoqLiteSession internal constructor(
                     subscribeId = hdr.subscribeId
                     groupSequence = hdr.sequence
                     headerRead = true
+                    Log.d("NestRx") { "uni grpHdr id=$subscribeId seq=$groupSequence" }
                 }
                 while (true) {
                     val frame = buffer.readSizePrefixed() ?: break
@@ -477,6 +490,10 @@ class MoqLiteSession internal constructor(
                                 payload = frame,
                             ),
                         )
+                    } else {
+                        Log.w("NestRx") {
+                            "uni frame drop: no live sub for id=$subscribeId seq=$groupSequence size=${frame.size}"
+                        }
                     }
                     // If the subscription has been closed already we
                     // silently drop the frame — the publisher hasn't
@@ -527,6 +544,7 @@ class MoqLiteSession internal constructor(
     suspend fun publish(broadcastSuffix: String): MoqLitePublisherHandle {
         ensureOpen()
         val normalised = MoqLitePath.normalize(broadcastSuffix)
+        Log.d("NestTx") { "publish suffix='$normalised'" }
         val publisher: PublisherStateImpl
         state.withLock {
             check(!closed) { "session is closed" }
@@ -611,6 +629,9 @@ class MoqLiteSession internal constructor(
                             val please = MoqLiteCodec.decodeAnnouncePlease(pleasePayload)
                             val emittedSuffix =
                                 MoqLitePath.stripPrefix(please.prefix, publisher.suffix) ?: publisher.suffix
+                            Log.d("NestTx") {
+                                "inbound AnnouncePlease prefix='${please.prefix}' → reply Active suffix='$emittedSuffix'"
+                            }
                             bidi.write(
                                 MoqLiteCodec.encodeAnnounce(
                                     MoqLiteAnnounce(
@@ -627,6 +648,10 @@ class MoqLiteSession internal constructor(
                         MoqLiteControlType.Subscribe -> {
                             val subPayload = buffer.readSizePrefixed() ?: return@collect
                             val sub = MoqLiteCodec.decodeSubscribe(subPayload)
+                            Log.d("NestTx") {
+                                "inbound SUBSCRIBE id=${sub.id} broadcast='${sub.broadcast}' track='${sub.track}' " +
+                                    "priority=${sub.priority} maxLatencyMs=${sub.maxLatencyMillis}"
+                            }
                             // Register the subscription BEFORE sending Ok so the
                             // peer's observation of Ok is a happens-after of
                             // `inboundSubs += sub`. Otherwise on dispatchers that
@@ -675,7 +700,12 @@ class MoqLiteSession internal constructor(
         // 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) }
+        inboundSub?.let {
+            Log.d("NestTx") {
+                "inbound SUBSCRIBE FIN'd: removing id=${it.id} broadcast='${it.broadcast}' track='${it.track}'"
+            }
+            publisher.removeInboundSubscription(it)
+        }
     }
 
     /**
@@ -760,7 +790,13 @@ class MoqLiteSession internal constructor(
         suspend fun registerInboundSubscription(sub: MoqLiteSubscribe) {
             gate.withLock {
                 if (publisherClosed) return
+                val wasEmpty = inboundSubs.isEmpty()
                 inboundSubs += sub
+                if (wasEmpty) {
+                    Log.d("NestTx") {
+                        "first inbound subscriber attached id=${sub.id} broadcast='${sub.broadcast}' track='${sub.track}'"
+                    }
+                }
             }
         }
 
@@ -877,6 +913,10 @@ class MoqLiteSession internal constructor(
             val sub = inboundSubs.first()
             val sequence = nextSequence++
             val uni = openGroupStream(subscribeId = sub.id, sequence = sequence)
+            Log.d("NestTx") {
+                "openGroup seq=$sequence keyedOnSubId=${sub.id} broadcast='${sub.broadcast}' track='${sub.track}' " +
+                    "inboundSubsCount=${inboundSubs.size}"
+            }
             return GroupOutbound(sequence = sequence, uni = uni)
         }
     }