fix(quic): exempt retired stream ids from client-initiated squatting guard

The audit-4 #5 guard ran before the existing phantom-stream check, so an
msquic-style aggressive STREAM retransmit on a stream we'd opened and
retired (peer's loss-detector refire racing our FIN-ACK) closed the
connection with STREAM_STATE_ERROR. Observed in the parallel `transfer`
interop test where retransmits on retired streams 0/4 truncated whichever
URL was still mid-receive (5 MB → 2.2 MB).

Add `!isStreamIdRetiredLocked` to the guard so legitimate retransmits
fall through to the existing silent-drop branch. Genuine squatting on
never-opened CLIENT_* ids still closes — the existing FrameRoutingTest
case stays green because id 0 is never put into the retired ring.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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

@@ -554,8 +554,18 @@ private fun dispatchFrames(
                 // 9000 §19.8, only the side that owns the parity may open;
                 // a server squatting on a CLIENT_* id is a protocol violation
                 // and could otherwise inject phantom streams into newPeerStreams.
+                //
+                // Exempt retired ids: msquic-style aggressive loss recovery
+                // can retransmit a STREAM frame on a stream we've fully
+                // retired before the peer sees our FIN-ACK. That's a
+                // legitimate retransmit on a stream WE opened, not squatting
+                // — the next guard drops it silently. Without this exemption
+                // the close fires on the parallel `transfer` interop test
+                // (3 GETs on streams 0/4/8) and truncates whichever stream
+                // is still mid-receive when the close lands.
                 if (StreamId.isClientInitiated(frame.streamId) &&
-                    conn.streamByIdLocked(frame.streamId) == null
+                    conn.streamByIdLocked(frame.streamId) == null &&
+                    !conn.isStreamIdRetiredLocked(frame.streamId)
                 ) {
                     conn.markClosedExternally(
                         "peer opened stream ${frame.streamId} on client-initiated id space (STREAM_STATE_ERROR)",

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

@@ -270,6 +270,88 @@ class StreamRetirementSoakTest {
             )
         }
 
+    @Test
+    fun phantomGuardDropsRetransmitOnRetiredClientBidiStream() =
+        runBlocking {
+            // Same shape as [phantomGuardDropsRetransmitOnRetiredPeerStream]
+            // but for a CLIENT_BIDI id we opened ourselves. The audit-4 #5
+            // guard rejects STREAM frames on client-initiated ids without a
+            // live local stream (server squatting). Pre-fix that guard fired
+            // BEFORE the retired-id check, so msquic-style aggressive
+            // retransmits on a stream we'd already retired (peer's
+            // loss-detector refire racing our FIN-ACK) closed the
+            // connection with STREAM_STATE_ERROR. Observed in the parallel
+            // `transfer` interop test where retransmits on retired
+            // streams 0/4 truncated whichever stream was still mid-receive.
+            val (client, pipe) = newConnectedClient()
+            val stream = client.openBidiStream()
+            val streamId = stream.streamId
+            stream.send.enqueue("hi".encodeToByteArray())
+            stream.send.finish()
+            val anyOutput = drainAll(client, pipe)
+            assertTrue(anyOutput, "writer must emit our request + FIN")
+
+            // Peer ACKs everything sent so far → finAcked latches.
+            val largestSent = client.application.pnSpace.nextPacketNumber - 1L
+            val ackPacket =
+                pipe.buildServerApplicationDatagram(
+                    listOf(
+                        AckFrame(
+                            largestAcknowledged = largestSent,
+                            ackDelay = 0L,
+                            firstAckRange = largestSent,
+                        ),
+                    ),
+                )!!
+            feedDatagram(client, ackPacket, nowMillis = 0L)
+            assertTrue(stream.send.finAcked, "send-side FIN must be ACKed")
+
+            // Peer responds and FINs the bidi stream.
+            val responseFrame =
+                StreamFrame(
+                    streamId = streamId,
+                    offset = 0L,
+                    data = "hello".encodeToByteArray(),
+                    fin = true,
+                )
+            feedDatagram(
+                client,
+                pipe.buildServerApplicationDatagram(listOf(responseFrame))!!,
+                nowMillis = 0L,
+            )
+            // Drain the application-side flow → isFullyRead transitions true.
+            client.streamById(streamId)?.incoming?.toList()
+            // Trigger retirement at the top of the next buildApplicationPacket.
+            drainAll(client, pipe)
+            assertTrue(
+                client.isStreamIdRetiredLocked(streamId),
+                "fully-settled CLIENT_BIDI stream must enter retired-id ring",
+            )
+
+            // The regression: replay the same STREAM+FIN frame as a peer
+            // retransmit. Pre-fix the audit-4 #5 guard fired with reason
+            // "peer opened stream X on client-initiated id space
+            // (STREAM_STATE_ERROR)" because it ran BEFORE the retired-id
+            // check. Post-fix the guard exempts retired ids and the
+            // existing phantom-stream branch silently drops the frame.
+            feedDatagram(
+                client,
+                pipe.buildServerApplicationDatagram(listOf(responseFrame))!!,
+                nowMillis = 0L,
+            )
+
+            assertEquals(
+                QuicConnection.Status.CONNECTED,
+                client.status,
+                "retransmit on retired CLIENT_BIDI id must NOT close the connection",
+            )
+            assertEquals(
+                0,
+                client.streamsListLocked().size,
+                "retransmit must not mint a phantom stream entry",
+            )
+        }
+
     @Test
     fun retirementPreservesConnectionLevelMaxDataAccounting() =
         runBlocking {