feat(quic): wire STREAM data retransmit — token emission + ACK/loss dispatch

Closes commit C of the deferred-follow-ups pass. With the
SendBuffer rewrite in commit B (retain-until-ACK with markAcked /
markLost), the connection now wires STREAM frames into the same
RFC 9002 retransmit path that already handles flow-control
extensions.

# Writer side

QuicConnectionWriter.buildApplicationPacket records a
RecoveryToken.Stream(streamId, offset, length, fin) for every
STREAM frame it emits. The token captures the on-wire byte range
plus the FIN bit so retransmit can reproduce the same StreamFrame
on next drain.

# ACK side

New QuicConnection.onTokensAcked() mirrors onTokensLost. The
parser's AckFrame handler iterates the drained packets and routes
each to onTokensAcked, which:
  - For Stream tokens: calls SendBuffer.markAcked(offset, length).
    The buffer removes the range from in-flight; if the contiguous
    low end is now fully ACK'd, flushedFloor advances and storage
    shifts forward.
  - For Crypto tokens: same shape, applied to the per-level
    cryptoSend buffer (commit E will exercise this path for
    handshake reliability — Crypto retransmit is wired now but the
    writer's CRYPTO emission path doesn't yet record Crypto tokens;
    that's commit E).
  - For control-frame and Ack tokens: ACK-no-op. The frame already
    did its job by reaching the peer; no per-buffer state to
    release.

# Loss side

onTokensLost (commit A) already routes Stream tokens to
SendBuffer.markLost. With commit B's real implementation (was a
no-op stub), this now actually re-queues the byte range for
retransmit. The next writer drain pulls from the retransmit queue
before any fresh sends, with the original offset preserved (RFC
9000 §13.3 idempotent retransmit).

# Tests added (3, all pass)

  - streamFrame_carriesStreamToken_inSentPacket: writer emits a
    Stream token whose fields match the StreamFrame on the wire
  - streamData_lostAndRetransmittedOnNextDrain: simulate loss via
    direct dispatch, observe re-emit at the same offset in a fresh
    SentPacket (different PN)
  - streamData_ackedReleasesBuffer: ACK via onTokensAcked,
    enqueue more bytes, observe the next send picks up at the
    post-ACK offset (proves bytes were released and floor advanced)

Full :quic test suite, nestsClient moq-lite tests, amethyst Android
compile all pass.

Net result: lost STREAM data (e.g. nestsClient bidi control-stream
bytes — moq-lite Subscribe/Announce control messages travel on
QUIC bidi streams) is now recovered automatically. Audio rooms
benefit indirectly: the relay's announce/subscribe path is more
resilient to packet loss.

https://claude.ai/code/session_01PYYez8a6sjiakyjAxsfCEQ

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

@@ -766,6 +766,58 @@ class QuicConnection(
     /** Caller must hold [lock]. */
     internal fun streamByIdLocked(id: Long): QuicStream? = streams[id]
 
+    /**
+     * ACK-path counterpart to [onTokensLost]. Called by the parser
+     * after [com.vitorpamplona.quic.connection.recovery.drainAckedSentPackets]
+     * removes the carrying packet from the sent map. Tokens whose
+     * underlying byte ranges live in a [com.vitorpamplona.quic.stream.SendBuffer]
+     * (Stream / Crypto) trigger a [com.vitorpamplona.quic.stream.SendBuffer.markAcked]
+     * call so the buffer can advance its flushedFloor and release
+     * memory. Other token types are ACK-no-ops (the peer's
+     * acknowledgment of a control frame doesn't require any local
+     * action — the frame already did its job by reaching the peer).
+     *
+     * Caller must hold [lock].
+     */
+    internal fun onTokensAcked(tokens: List<com.vitorpamplona.quic.connection.recovery.RecoveryToken>) {
+        for (token in tokens) {
+            when (token) {
+                com.vitorpamplona.quic.connection.recovery.RecoveryToken.Ack -> {
+                    // ACK-of-ACK is a no-op.
+                }
+
+                is com.vitorpamplona.quic.connection.recovery.RecoveryToken.MaxStreamsUni,
+                is com.vitorpamplona.quic.connection.recovery.RecoveryToken.MaxStreamsBidi,
+                is com.vitorpamplona.quic.connection.recovery.RecoveryToken.MaxData,
+                is com.vitorpamplona.quic.connection.recovery.RecoveryToken.MaxStreamData,
+                is com.vitorpamplona.quic.connection.recovery.RecoveryToken.ResetStream,
+                is com.vitorpamplona.quic.connection.recovery.RecoveryToken.StopSending,
+                is com.vitorpamplona.quic.connection.recovery.RecoveryToken.NewConnectionId,
+                -> {
+                    // Control frames have no per-buffer state to release on
+                    // ACK. (Pending* maps are populated only on loss; an ACK
+                    // for a frame that never lost is naturally absent.)
+                }
+
+                is com.vitorpamplona.quic.connection.recovery.RecoveryToken.Stream -> {
+                    val stream = streamByIdLocked(token.streamId) ?: continue
+                    stream.send.markAcked(offset = token.offset, length = token.length)
+                    if (token.length == 0L && token.fin) {
+                        // FIN-only ACK: treat as zero-length ACK at offset.
+                        // markAcked already handles length==0 ⇒ FIN match.
+                    }
+                }
+
+                is com.vitorpamplona.quic.connection.recovery.RecoveryToken.Crypto -> {
+                    levelState(token.level).cryptoSend.markAcked(
+                        offset = token.offset,
+                        length = token.length,
+                    )
+                }
+            }
+        }
+    }
+
     /**
      * Step 6 of `quic/plans/2026-05-04-control-frame-retransmit.md`:
      * dispatch the tokens of declared-lost packets to the matching

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

@@ -180,6 +180,13 @@ private fun dispatchFrames(
                 // returned list on the floor.
                 val largestSentTime = state.sentPackets[frame.largestAcknowledged]?.sentAtMillis
                 val drained = drainAckedSentPackets(state.sentPackets, frame)
+                // Step C of the deferred-follow-ups pass: dispatch ACK
+                // to per-buffer markAcked for Stream / Crypto tokens.
+                // Releases SendBuffer memory and advances its
+                // flushedFloor as low-end ACKs accumulate.
+                for (drainedPacket in drained) {
+                    conn.onTokensAcked(drainedPacket.tokens)
+                }
                 val advancedLargest =
                     state.largestAckedPn?.let { it < frame.largestAcknowledged } ?: true
                 if (advancedLargest) {

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

@@ -325,6 +325,18 @@ private fun buildApplicationPacket(
                         fin = chunk.fin,
                         explicitLength = true,
                     )
+                // Step C of the deferred-follow-ups pass: track this
+                // STREAM emission so RFC 9002 retransmit can re-queue
+                // the byte range on loss. SendBuffer.markLost (commit B)
+                // moves the range from in-flight back to the retransmit
+                // queue, and the next takeChunk replays it.
+                tokens +=
+                    RecoveryToken.Stream(
+                        streamId = stream.streamId,
+                        offset = chunk.offset,
+                        length = chunk.data.size.toLong(),
+                        fin = chunk.fin,
+                    )
                 packetBudget -= chunk.data.size + 32
                 connBudget -= chunk.data.size
                 conn.sendConnectionFlowConsumed += chunk.data.size

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

@@ -0,0 +1,200 @@
+/*
+ * 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.quic.connection
+
+import com.vitorpamplona.quic.connection.recovery.RecoveryToken
+import com.vitorpamplona.quic.tls.InProcessTlsServer
+import kotlinx.coroutines.runBlocking
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
+
+/**
+ * Step C of the deferred-follow-ups pass: STREAM data retransmit
+ * end-to-end. Verifies the writer emits a Stream token per
+ * STREAM frame, the parser dispatches it to markAcked / markLost,
+ * and the buffer either releases or re-queues the bytes
+ * accordingly.
+ */
+class StreamRetransmitTest {
+    @Test
+    fun streamFrame_carriesStreamToken_inSentPacket() =
+        runBlocking {
+            val client = handshakedClient()
+            val stream = client.openUniStream()
+            stream.send.enqueue("hello".encodeToByteArray())
+
+            val sizeBefore = client.application.sentPackets.size
+            runCatching { drainOutbound(client, nowMillis = 1L) }
+            val newEntries =
+                client.application.sentPackets.entries
+                    .sortedBy { it.key }
+                    .drop(sizeBefore)
+            val streamPacket =
+                newEntries.firstOrNull { entry ->
+                    entry.value.tokens.any { it is RecoveryToken.Stream }
+                }
+            assertNotNull(streamPacket, "writer must record a Stream token in the SentPacket")
+            val streamToken =
+                streamPacket.value.tokens
+                    .filterIsInstance<RecoveryToken.Stream>()
+                    .single()
+            assertEquals(stream.streamId, streamToken.streamId)
+            assertEquals(0L, streamToken.offset)
+            assertEquals(5L, streamToken.length)
+            assertEquals(false, streamToken.fin)
+        }
+
+    @Test
+    fun streamData_lostAndRetransmittedOnNextDrain() =
+        runBlocking {
+            val client = handshakedClient()
+            val stream = client.openUniStream()
+            stream.send.enqueue("hello".encodeToByteArray())
+            // First drain — emits StreamFrame, records SentPacket.
+            runCatching { drainOutbound(client, nowMillis = 1L) }
+
+            // Locate the Stream token's carrying packet.
+            val firstPacketEntry =
+                client.application.sentPackets.entries.firstOrNull { entry ->
+                    entry.value.tokens.any { it is RecoveryToken.Stream }
+                }
+            assertNotNull(firstPacketEntry, "expected a SentPacket with Stream token after first drain")
+            val firstPn = firstPacketEntry.key
+
+            // Simulate loss via direct dispatch.
+            client.lock.lock()
+            try {
+                val streamToken =
+                    firstPacketEntry.value.tokens
+                        .filterIsInstance<RecoveryToken.Stream>()
+                        .single()
+                client.onTokensLost(listOf(streamToken))
+                // Remove the lost packet from the sent map (the loss
+                // detector would have done this).
+                client.application.sentPackets.remove(firstPn)
+            } finally {
+                client.lock.unlock()
+            }
+
+            // SendBuffer should have re-queued the bytes for retransmit.
+            assertEquals(5, stream.send.readableBytes, "lost 5 bytes should be back in the queue")
+
+            // Second drain — emits the retransmit at the same offset.
+            val sizeBeforeReplay = client.application.sentPackets.size
+            runCatching { drainOutbound(client, nowMillis = 2L) }
+            val replayEntries =
+                client.application.sentPackets.entries
+                    .sortedBy { it.key }
+                    .drop(sizeBeforeReplay)
+            val replayPacket =
+                replayEntries.firstOrNull { entry ->
+                    entry.value.tokens.any { it is RecoveryToken.Stream }
+                }
+            assertNotNull(replayPacket, "retransmit must produce a fresh SentPacket carrying the Stream token")
+            val replayToken =
+                replayPacket.value.tokens
+                    .filterIsInstance<RecoveryToken.Stream>()
+                    .single()
+            assertEquals(0L, replayToken.offset, "retransmit must replay original offset (RFC 9000 §13.3 idempotent)")
+            assertEquals(5L, replayToken.length)
+            assertTrue(replayPacket.key != firstPn, "retransmit uses a fresh PN")
+        }
+
+    @Test
+    fun streamData_ackedReleasesBuffer() =
+        runBlocking {
+            val client = handshakedClient()
+            val stream = client.openUniStream()
+            stream.send.enqueue("hello".encodeToByteArray())
+            runCatching { drainOutbound(client, nowMillis = 1L) }
+
+            // Bytes are in-flight. SendBuffer holds them.
+            // Now ACK via direct dispatch.
+            val packet =
+                client.application.sentPackets.entries
+                    .first { it.value.tokens.any { t -> t is RecoveryToken.Stream } }
+            client.lock.lock()
+            try {
+                client.onTokensAcked(packet.value.tokens)
+            } finally {
+                client.lock.unlock()
+            }
+
+            // After ACK: enqueue more, observe that the buffer
+            // continues from offset 5 (proves the prior bytes were
+            // released, sentOffset advanced).
+            stream.send.enqueue("world".encodeToByteArray())
+            val sizeBefore = client.application.sentPackets.size
+            runCatching { drainOutbound(client, nowMillis = 2L) }
+            val newEntries =
+                client.application.sentPackets.entries
+                    .sortedBy { it.key }
+                    .drop(sizeBefore)
+            val nextStreamToken =
+                newEntries
+                    .flatMap { it.value.tokens }
+                    .filterIsInstance<RecoveryToken.Stream>()
+                    .firstOrNull()
+            assertNotNull(nextStreamToken)
+            assertEquals(5L, nextStreamToken.offset, "after ACK, next send picks up at offset 5")
+            assertEquals(5L, nextStreamToken.length)
+        }
+
+    private fun handshakedClient(): QuicConnection =
+        runBlocking {
+            val client =
+                QuicConnection(
+                    serverName = "example.test",
+                    config = QuicConnectionConfig(),
+                    tlsCertificateValidator =
+                        com.vitorpamplona.quic.tls
+                            .PermissiveCertificateValidator(),
+                )
+            val serverScid = ConnectionId.random(8)
+            val tlsServer =
+                InProcessTlsServer(
+                    transportParameters =
+                        TransportParameters(
+                            initialMaxData = 1_000_000,
+                            initialMaxStreamDataBidiLocal = 100_000,
+                            initialMaxStreamDataBidiRemote = 100_000,
+                            initialMaxStreamDataUni = 100_000,
+                            initialMaxStreamsBidi = 100,
+                            initialMaxStreamsUni = 100,
+                            initialSourceConnectionId = serverScid.bytes,
+                            originalDestinationConnectionId = client.destinationConnectionId.bytes,
+                        ).encode(),
+                )
+            val pipe =
+                InMemoryQuicPipe(
+                    client = client,
+                    initialDcid = client.destinationConnectionId.bytes,
+                    serverScid = serverScid,
+                    tlsServer = tlsServer,
+                )
+            client.start()
+            pipe.drive(maxRounds = 16)
+            assertEquals(QuicConnection.Status.CONNECTED, client.status)
+            client
+        }
+}