feat(audio-rooms): MoqLiteSession listener-side (phase 5c-listener)

Listener-side moq-lite (Lite-03) session that wraps a WebTransportSession and exposes: - announce(prefix) — opens an Announce bidi with ControlType=Announce + AnnouncePlease(prefix), surfaces a flow of relay Announce updates (Active/Ended) - subscribe(broadcast, track, ...) — opens a Subscribe bidi with ControlType=Subscribe + body, reads the size-prefixed response, returns a MoqLiteSubscribeHandle whose frames flow yields each frame the publisher pushes - close() — cancels pumps, FINs every active subscribe bidi (moq-lite UNSUBSCRIBE is "FIN the bidi"), closes the transport Group demux: a single inbound-uni-stream pump per session reads the DataType byte (Group=0) + size-prefixed group header, then routes each subsequent (size, payload) frame to the matching subscriptionsBySubscribeId entry's frame channel. Speaker side stops at the announce/subscribe interface — moq-lite publisher flows depend on accept_bi() / server-initiated bidi streams (see clarifying agent summary in plans/2026-04-26-moq-lite-gap.md). That's a follow-up; the WebTransportSession interface needs an acceptBidiStream() method first. FakeWebTransport.openPeerUniStream() — new helper so tests can inject group data from the peer side without going through the session's outbound uni stream API. Tests in MoqLiteSessionTest — runBlocking-based (not runTest) so real-time channel coordination works without virtual-time artifacts: - subscribe_writes_request_and_returns_handle_on_ok - subscribe_throws_on_drop_response - announce_streams_relay_updates - groups_are_demuxed_by_subscribeId (no cross-talk) - unsubscribe_FINs_the_subscribe_bidi
2026-04-26 17:24:17 +00:00
parent fb47a4cf75
commit 4e136cacf4
4 changed files with 890 additions and 0 deletions
@@ -0,0 +1,137 @@
+/*
+ * 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.moq.lite
+
+import com.vitorpamplona.nestsclient.moq.MoqCodecException
+import com.vitorpamplona.quic.Varint
+import kotlinx.coroutines.CancellationException
+
+/**
+ * Streaming frame buffer for moq-lite. Every wire datum on a control
+ * stream is a varint length followed by `length` bytes; every uni
+ * stream starts with a one-varint type byte then a size-prefixed
+ * group header then a sequence of size-prefixed frames until QUIC FIN.
+ *
+ * Network reads arrive as opaque chunks, so this class holds a rolling
+ * buffer and exposes:
+ *   - [readVarint] — pull one varint when enough bytes have arrived
+ *   - [readSizePrefixed] — pull one size-prefixed payload (varint
+ *     length + payload bytes) when complete
+ *
+ * Both return null when more bytes are needed; the caller buffers
+ * additional chunks via [push] and retries.
+ *
+ * Not thread-safe — used from a single coroutine per stream.
+ */
+class MoqLiteFrameBuffer {
+    private var buf: ByteArray = ByteArray(0)
+    private var pos: Int = 0
+
+    /** Append [chunk] to the buffer, growing/compacting as needed. */
+    fun push(chunk: ByteArray) {
+        if (chunk.isEmpty()) return
+        compact()
+        val needed = buf.size + chunk.size
+        // Power-of-two doubling matches MoqWriter's growth policy and
+        // avoids quadratic copies under bursty arrivals.
+        var newCap = if (buf.size == 0) 64 else buf.size
+        while (newCap < needed) newCap *= 2
+        val grown = ByteArray(newCap)
+        buf.copyInto(grown, 0, 0, buf.size)
+        chunk.copyInto(grown, buf.size, 0, chunk.size)
+        buf = grown.copyOf(needed)
+    }
+
+    /** Try to read one varint. Returns null if not enough bytes. */
+    fun readVarint(): Long? {
+        val dec = Varint.decode(buf, pos) ?: return null
+        pos += dec.bytesConsumed
+        return dec.value
+    }
+
+    /**
+     * Try to read one size-prefixed payload. Returns the payload bytes
+     * (no length prefix) or null if either the length varint or the
+     * payload itself isn't fully buffered yet. On invalid lengths
+     * (negative, > Int.MAX_VALUE) throws.
+     */
+    fun readSizePrefixed(): ByteArray? {
+        val savedPos = pos
+        val len = readVarint() ?: return null
+        if (len < 0 || len > Int.MAX_VALUE) {
+            throw MoqCodecException("absurd moq-lite size prefix: $len")
+        }
+        if (pos + len.toInt() > buf.size) {
+            // Roll the cursor back so the next call sees the same
+            // varint and only commits when the whole payload arrives.
+            pos = savedPos
+            return null
+        }
+        val payload = buf.copyOfRange(pos, pos + len.toInt())
+        pos += len.toInt()
+        return payload
+    }
+
+    /** Bytes still buffered after [pos] — exposed for diagnostic / EOF detection. */
+    val remaining: Int get() = buf.size - pos
+
+    /** Drop the consumed prefix when half the buffer is dead weight. */
+    private fun compact() {
+        if (pos == 0) return
+        if (pos < buf.size / 2) return
+        val live = buf.size - pos
+        val newBuf = ByteArray(live)
+        buf.copyInto(newBuf, 0, pos, buf.size)
+        buf = newBuf
+        pos = 0
+    }
+}
+
+/**
+ * Pull *one* size-prefixed message off a chunk channel. Drains chunks
+ * one at a time into [buffer] until [MoqLiteFrameBuffer.readSizePrefixed]
+ * returns a full payload. Returns null if [chunks] closes before a
+ * complete message arrives.
+ *
+ * Channel-based rather than Flow-based — `collect { throw EarlyExit }`
+ * is fragile under virtual-time test dispatchers and doesn't compose
+ * well with multiple sequential reads on the same channel. The caller
+ * owns the channel and is responsible for keeping it pumped (typically
+ * a single launched collector that forwards `incoming` chunks into the
+ * channel).
+ */
+internal suspend fun readOneSizePrefixedFrom(
+    chunks: kotlinx.coroutines.channels.ReceiveChannel<ByteArray>,
+    buffer: MoqLiteFrameBuffer,
+): ByteArray? {
+    while (true) {
+        buffer.readSizePrefixed()?.let { return it }
+        val next =
+            try {
+                chunks.receive()
+            } catch (ce: CancellationException) {
+                throw ce
+            } catch (_: kotlinx.coroutines.channels.ClosedReceiveChannelException) {
+                return null
+            }
+        buffer.push(next)
+    }
+}
@@ -0,0 +1,410 @@
+/*
+ * 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.moq.lite
+
+import com.vitorpamplona.nestsclient.moq.MoqCodecException
+import com.vitorpamplona.nestsclient.transport.WebTransportSession
+import com.vitorpamplona.quic.Varint
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.channels.BufferOverflow
+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
+
+/**
+ * Listener-side moq-lite (Lite-03) session. Wraps a connected
+ * [WebTransportSession] and exposes:
+ *
+ *   - [announce] — open an Announce bidi with a prefix, observe live
+ *     `Active` / `Ended` updates from the relay.
+ *   - [subscribe] — open a Subscribe bidi for a `(broadcast, track)`
+ *     pair, await SubscribeOk, return a [MoqLiteSubscribeHandle] whose
+ *     [MoqLiteSubscribeHandle.frames] yields each frame the publisher
+ *     pushes.
+ *
+ * Speaker-side is **not yet implemented** — the publisher direction
+ * needs server-initiated bidi acceptance (relay → us) which the
+ * current `WebTransportSession` interface does not surface. Tracked in
+ * `nestsClient/plans/2026-04-26-moq-lite-gap.md` phase-5c-speaker.
+ *
+ * Wire-protocol scope: Lite-03 — no SETUP, no datagrams, one fresh
+ * client-initiated bidi per request, group data on uni streams. See
+ * the gap doc for the byte-level layout.
+ */
+class MoqLiteSession internal constructor(
+    private val transport: WebTransportSession,
+    private val scope: CoroutineScope,
+) {
+    private val state = Mutex()
+    private val subscriptionsBySubscribeId: MutableMap<Long, ListenerSubscription> = HashMap()
+    private var nextSubscribeId: Long = 0L
+    private var groupPump: Job? = null
+
+    @Volatile private var closed: Boolean = false
+
+    val isClosed: Boolean get() = closed
+
+    /**
+     * Open an Announce bidi against the relay. Sends
+     * `ControlType=Announce` + `AnnouncePlease(prefix)`, then surfaces
+     * a flow of [MoqLiteAnnounce] updates the relay streams back.
+     *
+     * The returned [MoqLiteAnnouncesHandle] keeps the bidi open until
+     * [MoqLiteAnnouncesHandle.close] is called. FINing the bidi is
+     * how moq-lite signals "no longer interested" — there is no
+     * ANNOUNCE_CANCEL message.
+     */
+    suspend fun announce(prefix: String): MoqLiteAnnouncesHandle {
+        ensureOpen()
+        val bidi = transport.openBidiStream()
+        bidi.write(Varint.encode(MoqLiteControlType.Announce.code))
+        bidi.write(MoqLiteCodec.encodeAnnouncePlease(MoqLiteAnnouncePlease(prefix)))
+
+        val updates = MutableSharedFlow<MoqLiteAnnounce>(replay = 0, extraBufferCapacity = 64)
+        val pump =
+            scope.launch {
+                val buffer = MoqLiteFrameBuffer()
+                try {
+                    bidi.incoming().collect { chunk ->
+                        buffer.push(chunk)
+                        while (true) {
+                            val payload = buffer.readSizePrefixed() ?: break
+                            updates.emit(MoqLiteCodec.decodeAnnounce(payload))
+                        }
+                    }
+                } catch (ce: CancellationException) {
+                    throw ce
+                } catch (_: Throwable) {
+                    // Flow terminated (peer FIN or transport close).
+                    // The Announce stream's emit-side just stops; consumers
+                    // see an end-of-flow.
+                }
+            }
+        return MoqLiteAnnouncesHandle(
+            updates = updates,
+            close = {
+                runCatching { bidi.finish() }
+                pump.cancelAndJoin()
+            },
+        )
+    }
+
+    /**
+     * Open a Subscribe bidi for `(broadcast, track)`. Sends
+     * `ControlType=Subscribe` + the [MoqLiteSubscribe] body, awaits
+     * the [MoqLiteCodec.SubscribeResponse] reply, and on Ok returns a
+     * [MoqLiteSubscribeHandle] whose `frames` flow yields each frame
+     * the publisher pushes through the relay (one [MoqLiteFrame] per
+     * payload, bundled into [MoqLiteFrame] structs that carry the
+     * group sequence).
+     *
+     * Throws [MoqLiteSubscribeException] if the publisher / relay
+     * replies with Drop, or if the response stream tears down before
+     * any reply arrives.
+     */
+    suspend fun subscribe(
+        broadcast: String,
+        track: String,
+        priority: Int = DEFAULT_PRIORITY,
+        ordered: Boolean = true,
+        maxLatencyMillis: Long = 0L,
+        startGroup: Long? = null,
+        endGroup: Long? = null,
+    ): MoqLiteSubscribeHandle {
+        ensureOpen()
+        val id =
+            state.withLock {
+                check(!closed) { "session is closed" }
+                val next = nextSubscribeId++
+                next
+            }
+        val request =
+            MoqLiteSubscribe(
+                id = id,
+                broadcast = broadcast,
+                track = track,
+                priority = priority,
+                ordered = ordered,
+                maxLatencyMillis = maxLatencyMillis,
+                startGroup = startGroup,
+                endGroup = endGroup,
+            )
+        val bidi = transport.openBidiStream()
+        bidi.write(Varint.encode(MoqLiteControlType.Subscribe.code))
+        bidi.write(MoqLiteCodec.encodeSubscribe(request))
+
+        // 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,
+        // then stop. We don't need a separate collector pump because
+        // post-Ok the bidi is idle (group data flows on its own uni
+        // streams).
+        val responseBuffer = MoqLiteFrameBuffer()
+        val responsePayload = readSubscribeResponseFromBidi(bidi.incoming(), responseBuffer, id)
+        when (val resp = MoqLiteCodec.decodeSubscribeResponse(responsePayload)) {
+            is MoqLiteCodec.SubscribeResponse.Dropped -> {
+                throw MoqLiteSubscribeException(
+                    "publisher rejected subscribe id=$id: errorCode=${resp.drop.errorCode} " +
+                        "reason='${resp.drop.reasonPhrase}'",
+                )
+            }
+
+            is MoqLiteCodec.SubscribeResponse.Ok -> {
+                val frames = Channel<MoqLiteFrame>(capacity = DEFAULT_FRAME_BUFFER, onBufferOverflow = BufferOverflow.DROP_OLDEST)
+                val sub =
+                    ListenerSubscription(
+                        id = id,
+                        request = request,
+                        ok = resp.ok,
+                        bidi = bidi,
+                        frames = frames,
+                    )
+                state.withLock {
+                    subscriptionsBySubscribeId[id] = sub
+                    if (groupPump == null) groupPump = scope.launch { pumpUniStreams() }
+                }
+                return MoqLiteSubscribeHandle(
+                    id = id,
+                    ok = resp.ok,
+                    frames = frames.consumeAsFlow(),
+                    unsubscribeAction = { unsubscribe(id) },
+                )
+            }
+        }
+    }
+
+    /**
+     * Drain inbound uni streams and route each one's group frames to
+     * the matching subscription. The relay opens a fresh uni stream
+     * per (subscribe id, group sequence); the first byte is
+     * [MoqLiteDataType] (0 = Group), followed by a size-prefixed
+     * [MoqLiteGroupHeader], followed by `varint(size) + payload`
+     * frames until QUIC FIN.
+     *
+     * One pump per session — started lazily on the first subscribe.
+     */
+    private suspend fun pumpUniStreams() {
+        try {
+            transport.incomingUniStreams().collect { stream ->
+                scope.launch { drainOneGroup(stream) }
+            }
+        } catch (ce: CancellationException) {
+            throw ce
+        } catch (_: Throwable) {
+            // Transport closed — subscriptions will surface end-of-flow
+            // via their own bidi pumps as well.
+        }
+    }
+
+    private suspend fun drainOneGroup(stream: com.vitorpamplona.nestsclient.transport.WebTransportReadStream) {
+        val buffer = MoqLiteFrameBuffer()
+        var typeRead = false
+        var subscribeId: Long = -1L
+        var groupSequence: Long = -1L
+        var headerRead = false
+        try {
+            stream.incoming().collect { chunk ->
+                buffer.push(chunk)
+                if (!typeRead) {
+                    val type = buffer.readVarint() ?: return@collect
+                    if (type != MoqLiteDataType.Group.code) {
+                        throw MoqCodecException("unknown moq-lite uni-stream type code: $type")
+                    }
+                    typeRead = true
+                }
+                if (!headerRead) {
+                    val payload = buffer.readSizePrefixed() ?: return@collect
+                    val hdr = MoqLiteCodec.decodeGroupHeader(payload)
+                    subscribeId = hdr.subscribeId
+                    groupSequence = hdr.sequence
+                    headerRead = true
+                }
+                while (true) {
+                    val frame = buffer.readSizePrefixed() ?: break
+                    val sub = state.withLock { subscriptionsBySubscribeId[subscribeId] }
+                    if (sub != null) {
+                        sub.frames.trySend(
+                            MoqLiteFrame(
+                                groupSequence = groupSequence,
+                                payload = frame,
+                            ),
+                        )
+                    }
+                    // If the subscription has been closed already we
+                    // silently drop the frame — the publisher hasn't
+                    // observed the unsubscribe yet (its uni streams
+                    // are independent of our bidi FIN).
+                }
+            }
+        } catch (ce: CancellationException) {
+            throw ce
+        } catch (_: Throwable) {
+            // Stream errored / FIN'd. Nothing to do — the next group
+            // arrives on a fresh uni stream.
+        }
+    }
+
+    private suspend fun unsubscribe(id: Long) {
+        val sub =
+            state.withLock { subscriptionsBySubscribeId.remove(id) } ?: return
+        runCatching { sub.bidi.finish() }
+        sub.frames.close()
+    }
+
+    suspend fun close() {
+        if (closed) return
+        closed = true
+        val toClose: List<ListenerSubscription>
+        state.withLock {
+            toClose = subscriptionsBySubscribeId.values.toList()
+            subscriptionsBySubscribeId.clear()
+        }
+        for (sub in toClose) {
+            runCatching { sub.bidi.finish() }
+            sub.frames.close()
+        }
+        groupPump?.cancelAndJoin()
+        runCatching { transport.close() }
+    }
+
+    private fun ensureOpen() {
+        check(!closed) { "MoqLiteSession is closed" }
+    }
+
+    /**
+     * Read the single size-prefixed subscribe response off a fresh
+     * subscribe bidi. moq-lite's subscribe-response is one wire write,
+     * so the first chunk we receive contains the entire payload (matches
+     * what the IETF [com.vitorpamplona.nestsclient.moq.MoqSession] does
+     * during its SETUP read).
+     *
+     * Throws [MoqLiteSubscribeException] if the bidi closes before any
+     * chunk arrives.
+     */
+    private suspend fun readSubscribeResponseFromBidi(
+        incoming: kotlinx.coroutines.flow.Flow<ByteArray>,
+        buffer: MoqLiteFrameBuffer,
+        id: Long,
+    ): ByteArray {
+        val chunk =
+            incoming.firstOrNull()
+                ?: throw MoqLiteSubscribeException("subscribe stream FIN before reply for id=$id")
+        buffer.push(chunk)
+        return buffer.readSizePrefixed()
+            ?: throw MoqLiteSubscribeException(
+                "first chunk on subscribe response did not carry a complete size-prefixed payload " +
+                    "(id=$id, chunkSize=${chunk.size})",
+            )
+    }
+
+    private class ListenerSubscription(
+        val id: Long,
+        val request: MoqLiteSubscribe,
+        val ok: MoqLiteSubscribeOk,
+        val bidi: com.vitorpamplona.nestsclient.transport.WebTransportBidiStream,
+        val frames: Channel<MoqLiteFrame>,
+    )
+
+    companion object {
+        /** moq-lite priority byte midpoint — neutral default. */
+        const val DEFAULT_PRIORITY: Int = 0x80
+
+        /**
+         * Per-subscription channel buffer for inbound frames. 128 audio
+         * frames at Opus 20 ms ≈ 2.5 s of backlog before DROP_OLDEST,
+         * matching a real-time listener's tolerance.
+         */
+        const val DEFAULT_FRAME_BUFFER: Int = 128
+
+        /**
+         * Attach to a [WebTransportSession] in the client role. Lite-03
+         * has no SETUP — the WT handshake itself is the handshake — so
+         * this returns a ready-to-use session immediately.
+         *
+         * @param pumpScope where the inbound-uni-stream + per-bidi pumps
+         *   live. Production code passes the owning ViewModel scope.
+         */
+        fun client(
+            transport: WebTransportSession,
+            pumpScope: CoroutineScope,
+        ): MoqLiteSession = MoqLiteSession(transport, pumpScope)
+    }
+}
+
+/**
+ * One frame received from a subscription. moq-lite's wire format
+ * carries no per-frame envelope beyond the size; [groupSequence] is
+ * pulled from the group header so consumers can detect group rollover
+ * (e.g. for keyframe boundaries).
+ */
+data class MoqLiteFrame(
+    val groupSequence: Long,
+    val payload: ByteArray,
+) {
+    override fun equals(other: Any?): Boolean {
+        if (this === other) return true
+        if (other !is MoqLiteFrame) return false
+        return groupSequence == other.groupSequence && payload.contentEquals(other.payload)
+    }
+
+    override fun hashCode(): Int = 31 * groupSequence.hashCode() + payload.contentHashCode()
+}
+
+/**
+ * Active subscription handle returned by [MoqLiteSession.subscribe].
+ * [frames] emits every frame the publisher pushes; [unsubscribe]
+ * FINs the bidi to signal "no longer interested" (moq-lite has no
+ * UNSUBSCRIBE message — FIN is the protocol).
+ */
+class MoqLiteSubscribeHandle internal constructor(
+    val id: Long,
+    val ok: MoqLiteSubscribeOk,
+    val frames: Flow<MoqLiteFrame>,
+    private val unsubscribeAction: suspend () -> Unit,
+) {
+    suspend fun unsubscribe() = unsubscribeAction()
+}
+
+/**
+ * Active announce-discovery handle returned by [MoqLiteSession.announce].
+ * [updates] emits every [MoqLiteAnnounce] update the relay streams
+ * back; [close] FINs the bidi to stop receiving updates.
+ */
+class MoqLiteAnnouncesHandle internal constructor(
+    val updates: Flow<MoqLiteAnnounce>,
+    private val close: suspend () -> Unit,
+) {
+    suspend fun close() = close.invoke()
+}
+
+/** Thrown when subscribe is rejected (Drop) or the response stream dies. */
+class MoqLiteSubscribeException(
+    message: String,
+    cause: Throwable? = null,
+) : RuntimeException(message, cause)
@@ -46,6 +46,7 @@ class FakeWebTransport private constructor(
    private val outboundBidiStreams: Channel<FakeBidiStream>,
    private val inboundBidiStreams: Channel<FakeBidiStream>,
    private val inboundUniStreams: Channel<WebTransportReadStream>,
+    private val outboundUniStreams: Channel<WebTransportReadStream>,
 ) : WebTransportSession {
    private val stateLock = Mutex()
    private var open = true
@@ -86,6 +87,7 @@ class FakeWebTransport private constructor(
        outboundBidiStreams.close()
        inboundBidiStreams.close()
        inboundUniStreams.close()
+        outboundUniStreams.close()
    }

    /**
@@ -94,6 +96,19 @@ class FakeWebTransport private constructor(
     */
    fun peerOpenedBidiStreams(): Flow<FakeBidiStream> = inboundBidiStreams.receiveAsFlow()

+    /**
+     * Open a uni stream from this side toward the paired peer. Tests use this
+     * to simulate a publisher pushing a moq-lite group uni stream — write the
+     * full sequence of chunks via [WebTransportWriteStream.write] then call
+     * [WebTransportWriteStream.finish] to FIN.
+     */
+    suspend fun openPeerUniStream(): WebTransportWriteStream {
+        stateLock.withLock { check(open) { "session closed" } }
+        val pipe = Channel<ByteArray>(Channel.BUFFERED)
+        outboundUniStreams.send(FakeReadStream(pipe))
+        return ChannelWriteStream(pipe)
+    }
+
    companion object {
        /**
         * Create two linked fakes that act as "client" and "server" endpoints of
@@ -115,6 +130,7 @@ class FakeWebTransport private constructor(
                    outboundBidiStreams = aToBBidi,
                    inboundBidiStreams = bToABidi,
                    inboundUniStreams = bToAUni,
+                    outboundUniStreams = aToBUni,
                )
            val b =
                FakeWebTransport(
@@ -123,6 +139,7 @@ class FakeWebTransport private constructor(
                    outboundBidiStreams = bToABidi,
                    inboundBidiStreams = aToBBidi,
                    inboundUniStreams = aToBUni,
+                    outboundUniStreams = bToAUni,
                )
            return a to b
        }
@@ -149,3 +166,20 @@ class FakeReadStream internal constructor(
 ) : WebTransportReadStream {
    override fun incoming(): Flow<ByteArray> = read.receiveAsFlow()
 }
+
+/**
+ * Write-only adapter over a [Channel]. Used by [FakeWebTransport.openPeerUniStream]
+ * so a test can drive a peer-initiated uni stream by writing chunks then
+ * FIN'ing via [finish].
+ */
+private class ChannelWriteStream(
+    private val channel: Channel<ByteArray>,
+) : WebTransportWriteStream {
+    override suspend fun write(chunk: ByteArray) {
+        channel.send(chunk)
+    }
+
+    override suspend fun finish() {
+        channel.close()
+    }
+}
@@ -0,0 +1,309 @@
+/*
+ * 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.moq.lite
+
+import com.vitorpamplona.nestsclient.transport.FakeBidiStream
+import com.vitorpamplona.nestsclient.transport.FakeWebTransport
+import com.vitorpamplona.quic.Varint
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.async
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.take
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import kotlin.test.AfterTest
+import kotlin.test.Test
+import kotlin.test.assertContentEquals
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+
+/**
+ * Drives a [MoqLiteSession] from the listener side against a fake peer
+ * that plays the relay role. Validates:
+ *
+ *   - subscribe() writes the right ControlType + body, blocks until the
+ *     peer's SubscribeOk arrives, then yields a SubscribeHandle whose
+ *     frames flow surfaces every incoming group.
+ *   - subscribe() throws MoqLiteSubscribeException on a Drop response.
+ *   - announce() writes the AnnouncePlease and surfaces every Announce
+ *     update the peer streams back.
+ *   - Group uni streams are demuxed by subscribeId — frames for sub A
+ *     never leak into sub B.
+ *
+ * No transport mock magic — these tests use the production codec to
+ * encode/decode bytes, so a wire-shape regression on either side fails
+ * here.
+ */
+class MoqLiteSessionTest {
+    /** Real-time scope for session pumps. Cancelled in [tearDown] so each
+     * test ends with a clean coroutine ledger.
+     */
+    private val supervisor = SupervisorJob()
+    private val pumpScope = CoroutineScope(supervisor)
+
+    @AfterTest
+    fun tearDown() =
+        runBlocking {
+            supervisor.cancelAndJoin()
+        }
+
+    @Test
+    fun subscribe_writes_request_and_returns_handle_on_ok() =
+        runBlocking {
+            val (clientSide, serverSide) = FakeWebTransport.pair()
+            val session = MoqLiteSession.client(clientSide, pumpScope)
+
+            val peerHandlesSubscribe =
+                async {
+                    val bidi = serverSide.peerOpenedBidiStreams().first()
+                    val req = readSubscribeRequest(bidi)
+                    assertEquals("speakerPubkey", req.broadcast)
+                    assertEquals("audio/data", req.track)
+                    assertEquals(MoqLiteSession.DEFAULT_PRIORITY, req.priority)
+                    val ok =
+                        MoqLiteSubscribeOk(
+                            priority = req.priority,
+                            ordered = req.ordered,
+                            maxLatencyMillis = req.maxLatencyMillis,
+                            startGroup = null,
+                            endGroup = null,
+                        )
+                    bidi.write(MoqLiteCodec.encodeSubscribeOk(ok))
+                    req
+                }
+
+            val handle = session.subscribe("speakerPubkey", "audio/data")
+            val request = peerHandlesSubscribe.await()
+            assertEquals(request.id, handle.id)
+            assertEquals(0L, handle.id, "first subscribe id is 0")
+
+            // Now push one group with two frames from the server side.
+            val uni = serverSide.openPeerUniStream()
+            uni.write(Varint.encode(MoqLiteDataType.Group.code))
+            uni.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId = handle.id, sequence = 7L)))
+            uni.write(framePayload(byteArrayOf(0x10, 0x11)))
+            uni.write(framePayload(byteArrayOf(0x20, 0x21)))
+            uni.finish()
+
+            val frames =
+                withTimeout(2_000) {
+                    handle.frames.take(2).toList()
+                }
+            assertEquals(2, frames.size)
+            assertEquals(7L, frames[0].groupSequence)
+            assertEquals(7L, frames[1].groupSequence)
+            assertContentEquals(byteArrayOf(0x10, 0x11), frames[0].payload)
+            assertContentEquals(byteArrayOf(0x20, 0x21), frames[1].payload)
+
+            session.close()
+        }
+
+    @Test
+    fun subscribe_throws_on_drop_response() =
+        runBlocking {
+            val (clientSide, serverSide) = FakeWebTransport.pair()
+            val session = MoqLiteSession.client(clientSide, pumpScope)
+
+            val peer =
+                async {
+                    val bidi = serverSide.peerOpenedBidiStreams().first()
+                    readSubscribeRequest(bidi)
+                    bidi.write(
+                        MoqLiteCodec.encodeSubscribeDrop(
+                            MoqLiteSubscribeDrop(errorCode = 4L, reasonPhrase = "no such broadcast"),
+                        ),
+                    )
+                }
+
+            assertFailsWith<MoqLiteSubscribeException> {
+                session.subscribe("nope", "audio/data")
+            }
+            peer.await()
+            session.close()
+        }
+
+    @Test
+    fun announce_streams_relay_updates() =
+        runBlocking {
+            val (clientSide, serverSide) = FakeWebTransport.pair()
+            val session = MoqLiteSession.client(clientSide, pumpScope)
+
+            val peer =
+                async {
+                    val bidi = serverSide.peerOpenedBidiStreams().first()
+                    val chunks = bidi.incoming().take(2).toList()
+                    val controlByte = MoqLiteFrameBuffer().apply { push(chunks[0]) }.readVarint()
+                    assertEquals(MoqLiteControlType.Announce.code, controlByte)
+                    val plea =
+                        MoqLiteFrameBuffer().apply { push(chunks[1]) }.readSizePrefixed()
+                            ?: error("AnnouncePlease chunk did not contain a complete size-prefixed payload")
+                    assertEquals("nests/30312:abc:room", MoqLiteCodec.decodeAnnouncePlease(plea).prefix)
+
+                    // Send two Announce updates back.
+                    bidi.write(
+                        MoqLiteCodec.encodeAnnounce(
+                            MoqLiteAnnounce(
+                                MoqLiteAnnounceStatus.Active,
+                                "speakerOne",
+                                hops = 1L,
+                            ),
+                        ),
+                    )
+                    bidi.write(
+                        MoqLiteCodec.encodeAnnounce(
+                            MoqLiteAnnounce(
+                                MoqLiteAnnounceStatus.Active,
+                                "speakerTwo",
+                                hops = 1L,
+                            ),
+                        ),
+                    )
+                }
+
+            val announces = session.announce("nests/30312:abc:room")
+            val updates = withTimeout(2_000) { announces.updates.take(2).toList() }
+            peer.await()
+
+            assertEquals(2, updates.size)
+            assertEquals("speakerOne", updates[0].suffix)
+            assertEquals(MoqLiteAnnounceStatus.Active, updates[0].status)
+            assertEquals("speakerTwo", updates[1].suffix)
+
+            announces.close()
+            session.close()
+        }
+
+    @Test
+    fun groups_are_demuxed_by_subscribeId() =
+        runBlocking {
+            val (clientSide, serverSide) = FakeWebTransport.pair()
+            val session = MoqLiteSession.client(clientSide, pumpScope)
+
+            // Set up two parallel subscriptions — peer accepts both, replies
+            // with Ok, then pushes one group per subscription out of order.
+            val subAck =
+                async {
+                    val bidiA = serverSide.peerOpenedBidiStreams().first()
+                    readSubscribeRequest(bidiA)
+                    bidiA.write(MoqLiteCodec.encodeSubscribeOk(okFor(0L)))
+                    bidiA
+                }
+            val handleA = session.subscribe("speakerA", "audio/data")
+            subAck.await()
+
+            val subAck2 =
+                async {
+                    val bidiB = serverSide.peerOpenedBidiStreams().first()
+                    readSubscribeRequest(bidiB)
+                    bidiB.write(MoqLiteCodec.encodeSubscribeOk(okFor(1L)))
+                    bidiB
+                }
+            val handleB = session.subscribe("speakerB", "audio/data")
+            subAck2.await()
+
+            assertEquals(0L, handleA.id)
+            assertEquals(1L, handleB.id)
+
+            // Push one group for A with payload "a", one for B with payload "b".
+            val uniB = serverSide.openPeerUniStream()
+            uniB.write(Varint.encode(MoqLiteDataType.Group.code))
+            uniB.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId = handleB.id, sequence = 0L)))
+            uniB.write(framePayload("b".encodeToByteArray()))
+            uniB.finish()
+
+            val uniA = serverSide.openPeerUniStream()
+            uniA.write(Varint.encode(MoqLiteDataType.Group.code))
+            uniA.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId = handleA.id, sequence = 0L)))
+            uniA.write(framePayload("a".encodeToByteArray()))
+            uniA.finish()
+
+            val fromA = withTimeout(2_000) { handleA.frames.first() }
+            val fromB = withTimeout(2_000) { handleB.frames.first() }
+            assertContentEquals("a".encodeToByteArray(), fromA.payload)
+            assertContentEquals("b".encodeToByteArray(), fromB.payload)
+
+            handleA.unsubscribe()
+            handleB.unsubscribe()
+            session.close()
+        }
+
+    @Test
+    fun unsubscribe_FINs_the_subscribe_bidi() =
+        runBlocking {
+            val (clientSide, serverSide) = FakeWebTransport.pair()
+            val session = MoqLiteSession.client(clientSide, pumpScope)
+
+            var peerBidi: FakeBidiStream? = null
+            val peer =
+                async {
+                    val bidi = serverSide.peerOpenedBidiStreams().first()
+                    readSubscribeRequest(bidi)
+                    bidi.write(MoqLiteCodec.encodeSubscribeOk(okFor(0L)))
+                    peerBidi = bidi
+                    // Drain whatever the listener writes after Ok — moq-lite
+                    // unsubscribe is a FIN, which surfaces here as the
+                    // incoming flow completing.
+                    bidi.incoming().toList()
+                }
+
+            val handle = session.subscribe("speakerX", "audio/data")
+            handle.unsubscribe()
+            withTimeout(2_000) { peer.await() }
+            assertEquals(true, peerBidi != null, "peer received the bidi")
+
+            session.close()
+        }
+
+    // ---------- helpers ----------
+
+    /**
+     * Read a Subscribe request from the peer side of the bidi. Each
+     * `bidi.write` on the session side becomes one channel send (so
+     * one chunk on this end), and the session writes ControlType then
+     * the encoded body — exactly two chunks. Pull both with a single
+     * [take]+[toList] and parse them, no streaming buffer needed.
+     */
+    private suspend fun readSubscribeRequest(bidi: FakeBidiStream): MoqLiteSubscribe {
+        val chunks = bidi.incoming().take(2).toList()
+        check(chunks.size == 2) { "expected 2 chunks (control byte + body), got ${chunks.size}" }
+        val controlByte = MoqLiteFrameBuffer().apply { push(chunks[0]) }.readVarint()
+        assertEquals(MoqLiteControlType.Subscribe.code, controlByte)
+        val payload =
+            MoqLiteFrameBuffer().apply { push(chunks[1]) }.readSizePrefixed()
+                ?: error("subscribe body chunk did not contain a complete size-prefixed payload")
+        return MoqLiteCodec.decodeSubscribe(payload)
+    }
+
+    private fun framePayload(bytes: ByteArray): ByteArray = Varint.encode(bytes.size.toLong()) + bytes
+
+    private fun okFor(
+        @Suppress("UNUSED_PARAMETER") id: Long,
+    ) = MoqLiteSubscribeOk(
+        priority = MoqLiteSession.DEFAULT_PRIORITY,
+        ordered = true,
+        maxLatencyMillis = 0L,
+        startGroup = null,
+        endGroup = null,
+    )
+}