test(nests): skip housekeeping bidi in groups_are_demuxed_by_subscribeId

The session-layer fix in 851045c6 lazy-launches a single shared
announce-watch bidi on first subscribe (publisher-disconnect
detection). The unit test groups_are_demuxed_by_subscribeId
assumed each session.subscribe() opened exactly one peer-side bidi
and grabbed them by raw `peerOpenedBidiStreams().first()`, which
races with the announce-watch bidi.

New helper nextSubscribeBidi(serverSide) iterates accepted bidis,
peeks the control-byte varint, and skips any that aren't Subscribe.
The race window (announce-watch bidi between subscribe #1 and
subscribe #2) is now handled gracefully — the test reliably
finds both subscribe bidis regardless of the announce-watch's
launch ordering.

Also extends the listener-survives-publisher-recycle plan doc
with the full diagnosis of why
reconnecting_wrapper_keeps_handle_alive_across_session_swap
remains a separate, narrower failure (publisher single-group
architecture: NestMoqLiteBroadcaster never rotates groups, so
mid-stream listener resubscribes have nothing to attach to).

https://claude.ai/code/session_01HXf3zG3F2ev2ASeQju7Y5S

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

@@ -28,7 +28,9 @@ 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.takeWhile
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withTimeout
@@ -203,10 +205,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 +221,7 @@ class MoqLiteSessionTest {
 
             val subAck2 =
                 async {
-                    val bidiB = serverSide.peerOpenedBidiStreams().first()
-                    readSubscribeRequest(bidiB)
+                    val (bidiB, _) = nextSubscribeBidi(serverSide)
                     bidiB.write(MoqLiteCodec.encodeSubscribeOk(okFor(1L)))
                     bidiB
                 }
@@ -429,6 +434,44 @@ 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> {
+        val bidiFlow = serverSide.peerOpenedBidiStreams()
+        var found: Pair<FakeBidiStream, MoqLiteSubscribe>? = null
+        bidiFlow
+            .takeWhile { found == null }
+            .collect { bidi ->
+                val firstChunk = bidi.incoming().firstOrNull() ?: return@collect
+                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@collect
+                }
+                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")
+                found = bidi to MoqLiteCodec.decodeSubscribe(payload)
+            }
+        return found ?: error("flow ended without a Subscribe bidi")
+    }
+
     private fun framePayload(bytes: ByteArray): ByteArray = Varint.encode(bytes.size.toLong()) + bytes
 
     private fun okFor(