feat(audio-rooms): announce-driven speaker discovery (T4 #17b)

Surface moq-lite's ANNOUNCE flow on NestsListener so the audio
room can render an authoritative "actively broadcasting"
indicator independent of kind-10312 presence's `publishing`
flag. Same channel nostrnests' web client uses for its live
badges (`useRoomAnnouncements` in the JS reference).

Wire-up:
  * RoomAnnouncement(pubkey, active) data class — one update
    per publisher transition (Active → broadcast came up,
    inactive → broadcast went down).
  * NestsListener.announces(): Flow<RoomAnnouncement> with a
    default body that throws UnsupportedOperationException on
    the IETF reference path.
  * MoqLiteNestsListener implements via session.announce("")
    against the room's namespace; the suffix carries the
    speaker pubkey hex straight through.
  * ReconnectingNestsListener routes via collectLatest so the
    consumer-facing flow restarts against each new session
    after a refresh / reconnect (no SubscribeHandle re-issuance
    pump needed — announces is a cold per-collect stream).
  * AudioRoomViewModel.announcedSpeakers: StateFlow<Set<String>>
    populated by observeAnnounces. Active emissions add the
    pubkey, inactive emissions remove it. Cleared on teardown.
    IETF listeners leave the set empty; UI falls back to
    presence's publishingNow flag.

Tests:
  * NestsListenerCatalogTest — adds the announces() default-
    throws-on-collect case.

Closes the audit's Tier-4 #17b gap. UI integration (e.g. a green
"live" dot driven by announcedSpeakers) is a follow-up — the data
flow is in place and downstream consumers can opt in.

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

@@ -171,6 +171,24 @@ class AudioRoomViewModel(
     private val _speakerCatalogs = MutableStateFlow<Map<String, RoomSpeakerCatalog>>(emptyMap())
     val speakerCatalogs: StateFlow<Map<String, RoomSpeakerCatalog>> = _speakerCatalogs.asStateFlow()
 
+    /**
+     * Pubkeys the moq-lite relay has announced as actively
+     * broadcasting in this room. Populated by [observeAnnounces]
+     * once the listener is Connected; an Active announce adds the
+     * pubkey, an Ended announce removes it. Independent of the
+     * kind-10312 `publishing` flag — announces come straight from
+     * the relay's view of who has an open broadcast track, while
+     * presence is the speaker's self-reported state. The two
+     * usually agree but the announce flow is the more authoritative
+     * "is this broadcast really live" signal.
+     *
+     * Empty when the listener doesn't expose announces (IETF
+     * reference path); the UI falls back to `publishingNow` from
+     * presence in that case.
+     */
+    private val _announcedSpeakers = MutableStateFlow<Set<String>>(emptySet())
+    val announcedSpeakers: StateFlow<Set<String>> = _announcedSpeakers.asStateFlow()
+
     /**
      * `true` once the local user has been kicked (#5) — the platform
      * layer flips this on a valid kind-4312 from a host/moderator and
@@ -198,6 +216,7 @@ class AudioRoomViewModel(
     private var listener: NestsListener? = null
     private var connectJob: Job? = null
     private var stateObserverJob: Job? = null
+    private var announcesJob: Job? = null
 
     // Last in-flight `listener.close()` launched by teardown(). A subsequent
     // connect() awaits this before opening a fresh transport so two QUIC
@@ -505,6 +524,31 @@ class AudioRoomViewModel(
         _uiState.update { it.copy(broadcast = targetState) }
     }
 
+    /**
+     * Drain the moq-lite ANNOUNCE flow into [announcedSpeakers].
+     * Each Active emission adds the speaker pubkey; an inactive
+     * emission removes it. Best-effort — listeners that don't
+     * support announces (IETF reference path) throw on the first
+     * `collect` and we silently leave the set empty (the UI falls
+     * back to presence's `publishingNow` flag).
+     */
+    private fun observeAnnounces(l: NestsListener) {
+        announcesJob?.cancel()
+        announcesJob =
+            viewModelScope.launch {
+                runCatching {
+                    l.announces().collect { ann ->
+                        if (closed) return@collect
+                        if (ann.active) {
+                            _announcedSpeakers.update { it + ann.pubkey }
+                        } else {
+                            _announcedSpeakers.update { it - ann.pubkey }
+                        }
+                    }
+                }
+            }
+    }
+
     private fun observeListenerState(l: NestsListener) {
         stateObserverJob?.cancel()
         stateObserverJob =
@@ -583,6 +627,7 @@ class AudioRoomViewModel(
                     }
                     listener = l
                     observeListenerState(l)
+                    observeAnnounces(l)
                 } catch (ce: CancellationException) {
                     throw ce
                 } catch (t: Throwable) {
@@ -746,6 +791,11 @@ class AudioRoomViewModel(
         connectJob = null
         stateObserverJob?.cancel()
         stateObserverJob = null
+        announcesJob?.cancel()
+        announcesJob = null
+        if (_announcedSpeakers.value.isNotEmpty()) {
+            _announcedSpeakers.value = emptySet()
+        }
         // Detach + suspend-stop each player on the cleanup scope. The
         // listener.close() below tears down the MoQ session and drops
         // every active subscription, so we don't need to call

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

@@ -23,11 +23,14 @@ package com.vitorpamplona.nestsclient
 import com.vitorpamplona.nestsclient.moq.MoqObject
 import com.vitorpamplona.nestsclient.moq.SubscribeHandle
 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 kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.map
 import java.util.concurrent.atomic.AtomicLong
 
@@ -108,6 +111,30 @@ class MoqLiteNestsListener internal constructor(
         )
     }
 
+    override fun announces(): Flow<RoomAnnouncement> =
+        flow {
+            check(state.value is NestsListenerState.Connected) {
+                "NestsListener.announces requires Connected state, was ${state.value}"
+            }
+            // Empty prefix → the relay sends every active broadcast
+            // under our session's namespace (one per speaker). The
+            // suffix is the broadcast's path component within the
+            // room — for nests this is the speaker pubkey hex.
+            val handle = session.announce(prefix = "")
+            try {
+                handle.updates.collect { announce ->
+                    emit(
+                        RoomAnnouncement(
+                            pubkey = announce.suffix,
+                            active = announce.status == MoqLiteAnnounceStatus.Active,
+                        ),
+                    )
+                }
+            } finally {
+                runCatching { handle.close() }
+            }
+        }
+
     override suspend fun close() {
         if (state.value is NestsListenerState.Closed) return
         runCatching { session.close() }

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

@@ -26,9 +26,11 @@ import com.vitorpamplona.nestsclient.moq.SubscribeFilter
 import com.vitorpamplona.nestsclient.moq.SubscribeHandle
 import com.vitorpamplona.nestsclient.moq.TrackNamespace
 import com.vitorpamplona.nestsclient.transport.WebTransportSession
+import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.flow.flow
 
 /**
  * High-level listener handle for an audio-room. Hides the layered HTTP +
@@ -72,10 +74,49 @@ interface NestsListener {
             "subscribeCatalog is moq-lite-only; IETF listener has no catalog channel.",
         )
 
+    /**
+     * Cold flow of moq-lite ANNOUNCE updates for the room's
+     * namespace. Each emission represents a publisher transitioning
+     * Active (broadcast started) or inactive (broadcast ended).
+     * Lets clients render an "actively broadcasting" indicator
+     * independent of the kind-10312 `publishing` presence tag —
+     * the JS reference's `useRoomAnnouncements` consumes this.
+     *
+     * Default body throws [UnsupportedOperationException] on the
+     * IETF reference path (`MoqSession` doesn't define an
+     * announce-prefix subscription that fits this shape). Callers
+     * mixing the two should `runCatching` the collect.
+     *
+     * @throws UnsupportedOperationException on the IETF listener.
+     * @throws IllegalStateException if the listener is not in [NestsListenerState.Connected].
+     */
+    fun announces(): Flow<RoomAnnouncement> =
+        flow {
+            throw UnsupportedOperationException(
+                "announces() is moq-lite-only; IETF listener has no announce-prefix flow.",
+            )
+        }
+
     /** Tear down the MoQ session + underlying transport. Idempotent. */
     suspend fun close()
 }
 
+/**
+ * One moq-lite ANNOUNCE update. The JS reference uses these to
+ * drive the "live now" badge on each speaker's avatar.
+ *
+ * @param pubkey the suffix of the announce path — for nests
+ *   audio rooms this is the speaker's pubkey hex (the broadcast
+ *   path `<roomId>/<pubkey>` minus the `<roomId>` prefix the
+ *   listener already established).
+ * @param active true on the `Active` status (broadcast came up);
+ *   false on `Ended` (broadcast went down).
+ */
+data class RoomAnnouncement(
+    val pubkey: String,
+    val active: Boolean,
+)
+
 /**
  * Lifecycle states of a [NestsListener].
  *

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.delay
+import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
@@ -35,6 +36,7 @@ import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withTimeoutOrNull
@@ -205,6 +207,32 @@ private class ReconnectingHandle(
 
     override suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle = reissuingSubscribe { listener -> listener.subscribeCatalog(speakerPubkeyHex) }
 
+    /**
+     * The announce flow is a cold per-collect stream rather than a
+     * SubscribeHandle, so it doesn't go through [reissuingSubscribe].
+     * Instead, restart the inner collect on every activeListener
+     * change — `collectLatest` cancels the prior collection and
+     * re-runs against the fresh session. Best-effort: an inner
+     * listener that throws (IETF reference path) is logged-and-
+     * swallowed so a future moq-lite session keeps emitting.
+     */
+    override fun announces(): Flow<RoomAnnouncement> =
+        flow {
+            activeListener.collectLatest { listener ->
+                if (listener == null) return@collectLatest
+                val terminalOrConnected =
+                    listener.state.first { state ->
+                        state is NestsListenerState.Connected ||
+                            state is NestsListenerState.Closed ||
+                            state is NestsListenerState.Failed
+                    }
+                if (terminalOrConnected !is NestsListenerState.Connected) return@collectLatest
+                runCatching {
+                    listener.announces().collect { emit(it) }
+                }
+            }
+        }
+
     /**
      * Open a SubscribeHandle whose `objects` flow survives session
      * swaps. Each time the wrapper opens a fresh inner listener, the

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

@@ -53,4 +53,19 @@ class NestsListenerCatalogTest {
                 listener.subscribeCatalog("speakerPubkey")
             }
         }
+
+    @Test
+    fun ietfDefaultListenerAnnouncesFlowThrowsOnCollect() =
+        runTest {
+            // The announce-prefix channel is moq-lite only as well —
+            // the IETF reference path's announces() default body
+            // throws UnsupportedOperationException at collect time.
+            // Callers mixing the two protocols can `runCatching` the
+            // collect to no-op cleanly.
+            val listener = IetfStyleListener()
+            val flow = listener.announces()
+            assertFailsWith<UnsupportedOperationException> {
+                flow.collect { /* unreachable */ }
+            }
+        }
 }