feat(quic): priority-aware stream scheduling for moq-lite groups

Bias the QUIC connection writer's drain loop toward higher-priority
streams so moq-lite group streams with newer (higher) sequence numbers
drain ahead of older ones under congestion. Implements the T11.3
follow-up flagged in nestsClient/plans/2026-05-06-stream-priority-
followup.md (now removed).

QuicStream gets a `@Volatile var priority: Int = 0`. The writer's
streamsView iteration is replaced by a stable sortedByDescending pass
so same-priority streams keep their existing rotating-start round
robin while higher-priority tiers always drain first.

WebTransportWriteStream gains a `setPriority(Int)` hook; the QUIC-
backed adapter forwards to the underlying QuicStream, while the
in-memory test fakes treat it as a no-op. MoqLiteSession.openGroupStream
calls `uni.setPriority(sequence)` (saturating to Int.MAX_VALUE) to
mirror moq-rs's `Publisher::serve_group`.

Tests: a new InMemoryQuicPipe.decryptClientApplicationFrames helper
walks past coalesced long-header packets to surface 1-RTT frames,
which lets QuicConnectionWriterTest assert that the higher-priority
stream's StreamFrame lands first inside a single drained packet.

https://claude.ai/code/session_01KWdr4RjVvyYZfEuPVaQfUa

quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt

@@ -410,10 +410,22 @@ private fun buildApplicationPacket(
     // insertion-ordered and stays in sync with the streams map.
     val streamsView = conn.streamsListLocked()
     if (streamsView.isNotEmpty()) {
-        val start = conn.streamRoundRobinStart % streamsView.size
-        for (i in streamsView.indices) {
+        // Priority-then-round-robin: stable sortedByDescending preserves
+        // insertion order within a tier, so same-priority streams keep
+        // the rotating start-index round-robin behaviour. Higher-priority
+        // streams (e.g. moq-lite newer-sequence group streams) drain
+        // first under congestion. Default priority is 0; if every stream
+        // is at the default, iteration order matches pre-priority code.
+        // Cost: O(N log N) per drain pass and one transient list
+        // allocation. N is small (1–10 in the moq-lite audio path); if
+        // it ever grows enough to matter, switch to an indirect index
+        // sort or maintain an incrementally-sorted view on setPriority.
+        val sorted =
+            if (streamsView.size > 1) streamsView.sortedByDescending { it.priority } else streamsView
+        val start = conn.streamRoundRobinStart % sorted.size
+        for (i in sorted.indices) {
             if (packetBudget <= 64) break
-            val stream = streamsView[(start + i) % streamsView.size]
+            val stream = sorted[(start + i) % sorted.size]
             val streamRemaining = (stream.sendCredit - stream.send.sentOffset).coerceAtLeast(0L)
             // Skip if both stream and connection have no credit; FIN-only
             // (zero-byte) chunks may still go through because they don't

quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt

@@ -47,6 +47,27 @@ class QuicStream(
     val send = SendBuffer(bestEffort = bestEffort)
     val receive = ReceiveBuffer()
 
+    /**
+     * Send-side scheduling priority. The connection writer's drain loop
+     * iterates streams by descending priority; same-priority streams keep
+     * their existing round-robin order. Higher value = drains first under
+     * congestion. Default 0 matches pre-priority round-robin behaviour for
+     * every existing call site.
+     *
+     * Used by moq-lite group streams: the publisher assigns each new group
+     * a priority equal to its sequence number so that newer groups
+     * (fresher audio) drain ahead of older ones when retransmits queue up
+     * on a lossy link. Mirrors `Publisher::serve_group` in
+     * `rs/moq-lite/src/lite/publisher.rs` (`stream.set_priority`).
+     *
+     * `@Volatile` because callers (e.g. moq-lite's openGroupStream)
+     * assign from arbitrary coroutines while the writer reads it under
+     * [com.vitorpamplona.quic.connection.QuicConnection.lock] during a
+     * drain pass.
+     */
+    @Volatile
+    var priority: Int = 0
+
     /**
      * Bytes received and confirmed contiguous, exposed as a flow to the consumer.
      *

quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt

@@ -327,6 +327,46 @@ class InMemoryQuicPipe(
      */
     fun buildServerApplicationPacket(frames: List<com.vitorpamplona.quic.frame.Frame>): ByteArray? = buildServerApplicationDatagram(frames)
 
+    /**
+     * Decrypt the application-level (1-RTT) packet inside a client-emitted
+     * datagram and return its frames in wire order. Test-only helper for
+     * assertions that depend on per-frame ordering inside a packet (e.g.
+     * stream priority scheduling). Walks past any coalesced long-header
+     * packets (Initial / Handshake) at the front of the datagram, since the
+     * client may still flush ACKs at those levels post-handshake. Returns
+     * null if no short-header packet is present or decryption fails.
+     */
+    fun decryptClientApplicationFrames(datagram: ByteArray): List<com.vitorpamplona.quic.frame.Frame>? {
+        if (datagram.isEmpty()) return null
+        var offset = 0
+        while (offset < datagram.size) {
+            val first = datagram[offset].toInt() and 0xFF
+            if ((first and 0x80) == 0) {
+                val proto = serverApplicationRx ?: return null
+                val parsed =
+                    ShortHeaderPacket.parseAndDecrypt(
+                        bytes = datagram,
+                        offset = offset,
+                        dcidLen = serverScid.length,
+                        aead = proto.aead,
+                        key = proto.key,
+                        iv = proto.iv,
+                        hp = proto.hp,
+                        hpKey = proto.hpKey,
+                        largestReceivedInSpace = applicationPnSpace.largestReceived,
+                    ) ?: return null
+                applicationPnSpace.observeInbound(parsed.packet.packetNumber, 0L)
+                return decodeFrames(parsed.packet.payload)
+            }
+            // Long header — skip past it using the encoded length field so
+            // we can inspect any short-header packet that was coalesced
+            // after it.
+            val peeked = LongHeaderPacket.peekHeader(datagram, offset) ?: return null
+            offset += peeked.totalLength
+        }
+        return null
+    }
+
     private fun buildServerInitialPacket(crypto: ByteArray): ByteArray {
         val proto =
             PacketProtection(

quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriterTest.kt

@@ -110,6 +110,50 @@ class QuicConnectionWriterTest {
         }
     }
 
+    @Test
+    fun writer_drains_higher_priority_streams_before_lower_priority() {
+        // T11.3 follow-up: the writer's drain loop must iterate streams
+        // in descending priority order so moq-lite group streams with a
+        // higher sequence number drain ahead of older ones under
+        // congestion. Pre-fix this test, the writer iterated in stable
+        // round-robin order regardless of priority — so a backlog of
+        // retransmits on an older group could starve the listener of
+        // fresh frames. We pin the load-bearing invariant by inspecting
+        // StreamFrame order in a single emitted packet: low-priority
+        // stream is opened FIRST (so insertion order would normally win
+        // round-robin), but the high-priority stream's bytes must land
+        // earlier in the packet.
+        runBlocking {
+            val (client, pipe) = connectedClient()
+            val low = client.openBidiStream()
+            val high = client.openBidiStream()
+            low.priority = 0
+            high.priority = 10
+            // Distinct payloads small enough to coexist in one packet.
+            val lowPayload = ByteArray(200) { 0xAA.toByte() }
+            val highPayload = ByteArray(200) { 0xBB.toByte() }
+            low.send.enqueue(lowPayload)
+            high.send.enqueue(highPayload)
+
+            val datagram = drainOutbound(client, nowMillis = 0L)
+            assertNotNull(datagram, "drain must emit a packet")
+            val frames = pipe.decryptClientApplicationFrames(datagram)
+            assertNotNull(frames, "decrypt must succeed at the application level")
+            val streamFrames = frames.filterIsInstance<StreamFrame>()
+            assertEquals(
+                2,
+                streamFrames.size,
+                "expected one StreamFrame per stream in this drain, got $streamFrames",
+            )
+            assertEquals(
+                high.streamId,
+                streamFrames[0].streamId,
+                "higher-priority stream must drain first; saw ${streamFrames.map { it.streamId }}",
+            )
+            assertEquals(low.streamId, streamFrames[1].streamId)
+        }
+    }
+
     @Test
     fun writer_respects_connection_level_send_credit_cap() {
         // Audit-4 #9: pre-fix the writer ignored sendConnectionFlowCredit