fix(quic): RFC 9002 §6.2.4 — emit two ack-eliciting packets per PTO probe

Single-packet probes need 6 PTO doublings (~19s) to land one datagram
through the `amplificationlimit` interop scenario's 6-drop window.
quic-go and msquic kill the connection at ~10s of silence regardless
of our handshake-timeout budget, so we never recovered against them
(diagnosed in the parent investigation; the 10s→30s timeout bump in
0a892b0d4b only fixed picoquic).

RFC 9002 §6.2.4 allows up to 2 ack-eliciting packets per PTO. Adding
the second probe halves recovery to ~3 PTO rounds (~5s) and lands
within strict server tolerances.

Wiring:
- New `QuicConnection.pendingProbePackets`, set to 2 by handlePtoFired.
- Extracted `requeueInflightForProbe` from handlePtoFired so the send
  loop can re-requeue inflight CRYPTO / STREAM bytes between probes.
- Send loop decrements the budget after each probe-bearing send; if
  the budget is still positive, re-requeues AND re-arms `pendingPing`
  so the no-data fallback (post-handshake idle) still emits a second
  PING. Without the `pendingPing` re-arm, only the first probe fires
  when CRYPTO is fully ACK'd — `pendingPing` is one-shot in
  collectHandshakeLevelFrames.

Verified end-to-end:
- amplificationlimit: ✕→✓ vs quic-go (35s→7s); ✓ no-regression vs
  picoquic (19s→7s) and quinn (15s→7s); msquic now reports server-
  side UNSUPPORTED (was failing). Recovery times across the board
  drop ~3x because handshake-loss recovery is ~3 PTO rounds instead
  of ~6.
- handshake / transfer / multiplexing / handshakeloss all green vs
  quic-go, quinn, picoquic, msquic — no regression on the core matrix.

Tests:
- New `ptoEmitsTwoProbePacketsPerRfc9002` in PtoCryptoRetransmitTest
  invokes the EXACT helpers the send loop uses (handlePtoFired then
  requeueInflightForProbe between drains) and asserts two distinct
  Initial datagrams with the same CRYPTO bytes at offset 0 on
  distinct PNs. Verified the test fails when budget is reverted to 1.
- Existing PTO + recovery tests stay green.

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

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

@@ -573,6 +573,32 @@ class QuicConnection(
     @Volatile
     internal var consecutivePtoCount: Int = 0
 
+    /**
+     * RFC 9002 §6.2.4 probe budget. Set to 2 each time the PTO timer
+     * fires (`handlePtoFired`); decremented by the driver's send loop
+     * after each probe-bearing datagram leaves the socket. Between
+     * sends the loop re-requeues any inflight CRYPTO / STREAM bytes
+     * via [requeueInflightForProbe] so the next [drainOutbound] has
+     * payload to probe with.
+     *
+     * Why two and not one: under heavy consecutive packet loss
+     * (`amplificationlimit` interop scenario drops 6 client→server
+     * packets in a row, `handshakeloss` / `handshakecorruption` drop
+     * ~30% with burst=3), single-packet probes need 6 PTO doublings
+     * (~19s) to land one datagram. Strict server-side
+     * handshake-progress watchdogs (quic-go, msquic) tear the
+     * connection down at ~10s of silence regardless of our
+     * `max_idle_timeout`. Two probes per PTO halves recovery to
+     * ~3 PTO rounds (~5s) and keeps these servers from giving up.
+     *
+     * `@Volatile`: the send loop reads/writes from a background
+     * coroutine that may be on a different IO worker thread than
+     * the PTO timer's setter. JLS allows long-tearing on 32-bit
+     * JVMs without volatile.
+     */
+    @Volatile
+    internal var pendingProbePackets: Int = 0
+
     /**
      * Optional supplier of underlying UDP-socket counters. Wired by the
      * platform-specific driver since `UdpSocket`'s counters are

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

@@ -190,7 +190,54 @@ class QuicConnectionDriver(
                         drainOutbound(connection, nowMillis())
                     } ?: break
                 socket.send(out)
+                // RFC 9002 §6.2.4 probe budget: when the PTO timer
+                // fired we set `pendingProbePackets = 2`; consume one
+                // slot per probe-bearing datagram. If budget remains,
+                // re-requeue inflight CRYPTO / STREAM bytes so the
+                // next drainOutbound iteration has payload to probe
+                // with. The loop's own break-on-null then terminates
+                // naturally if there's nothing left to requeue (e.g.
+                // a connection with no inflight data — the first
+                // probe is a bare PING, the second slot harmlessly
+                // expires with no follow-up datagram).
+                //
+                // Decrementing AFTER socket.send (not before) ensures
+                // a non-PTO drain — wakeup-driven, no probe budget
+                // outstanding — leaves the field at 0 untouched. The
+                // post-decrement check uses `> 0` (not `>= 0`) so
+                // we only requeue while another probe is still owed.
+                if (connection.pendingProbePackets > 0) {
+                    connection.pendingProbePackets--
+                    if (connection.pendingProbePackets > 0) {
+                        // Two cooperating signals for the next probe:
+                        //   (a) requeue inflight CRYPTO / STREAM data so a
+                        //       payload-bearing probe can fire (preferred —
+                        //       a CRYPTO retransmit also covers the
+                        //       ack-eliciting requirement);
+                        //   (b) re-arm `pendingPing` for the no-data fallback
+                        //       (CRYPTO already ACK'd or never in flight, e.g.
+                        //       handshake-confirmed connections idling
+                        //       between transfers). Without (b), the writer
+                        //       suppresses the second probe entirely because
+                        //       `pendingPing` is one-shot — the first probe
+                        //       cleared it. The writer's own
+                        //       `collectHandshakeLevelFrames` still
+                        //       suppresses the PING when a CRYPTO frame
+                        //       lands in the same packet, so this never
+                        //       wastes a frame on top of a retransmit.
+                        requeueInflightForProbe(connection)
+                        connection.pendingPing = true
+                    }
+                }
             }
+            // Defence-in-depth: if the inner loop exited with budget
+            // still unconsumed (e.g. drainOutbound returned null on
+            // the very first iteration because nothing was inflight
+            // — only `pendingPing` would have been set, and it can
+            // be cleared by the writer without an outbound emission
+            // when no level has send keys), zero it out so the next
+            // wakeup-driven drain doesn't accidentally re-requeue.
+            connection.pendingProbePackets = 0
             // Use the loss-detection's PTO calculation in BOTH the pre- and
             // post-first-RTT-sample regimes. Pre-sample, smoothed_rtt =
             // INITIAL_RTT_MS so ptoBaseMs returns
@@ -344,6 +391,35 @@ class QuicConnectionDriver(
  */
 internal suspend fun handlePtoFired(conn: QuicConnection) {
     conn.pendingPing = true
+    requeueInflightForProbe(conn)
+    // RFC 9002 §6.2.4: the spec allows up to 2 ack-eliciting packets
+    // per PTO. The first probe falls out of the next drain naturally
+    // (we just requeued); the send loop watches [pendingProbePackets]
+    // and re-requeues for the second probe. Set AFTER the requeue so
+    // a concurrent reader of the field doesn't see "budget set,
+    // nothing requeued yet" — the requeue is what makes the budget
+    // meaningful.
+    conn.pendingProbePackets = 2
+    conn.consecutivePtoCount = (conn.consecutivePtoCount + 1).coerceAtMost(6)
+}
+
+/**
+ * Move any sent-but-not-ACK'd CRYPTO / STREAM bytes back onto the
+ * retransmit queue at every active encryption level so the next
+ * [drainOutbound] re-emits them. Used by both [handlePtoFired] (the
+ * initial PTO requeue) and the send loop's RFC 9002 §6.2.4 probe
+ * budget (the re-requeue between the first and second probe
+ * datagram). Idempotent on levels with nothing inflight, so calling
+ * it for already-ACKed or already-discarded levels is harmless.
+ *
+ * The level walk mirrors [handlePtoFired]'s original inline
+ * sequence: Handshake first (most-recent CRYPTO), then Initial
+ * (oldest), then Application (1-RTT STREAM data + post-handshake
+ * CRYPTO). The application branch holds [QuicConnection.streamsLock]
+ * because [QuicConnection.requeueAllInflightStreamData] iterates
+ * `streamsList`, which `openBidiStream` mutates under the same lock.
+ */
+internal suspend fun requeueInflightForProbe(conn: QuicConnection) {
     if (conn.handshake.sendProtection != null && !conn.handshake.keysDiscarded) {
         conn.requeueAllInflightCrypto(EncryptionLevel.HANDSHAKE)
     }

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

@@ -167,6 +167,120 @@ class PtoCryptoRetransmitTest {
             )
         }
 
+    @Test
+    fun ptoEmitsTwoProbePacketsPerRfc9002() =
+        runBlocking {
+            // RFC 9002 §6.2.4: a PTO probe SHOULD send up to 2
+            // ack-eliciting packets at the encryption level with
+            // unacknowledged data. Halves recovery time across
+            // consecutive drops — the `amplificationlimit` interop
+            // scenario drops 6 client→server packets in a row, so
+            // two probes per PTO collapse the recovery window from
+            // ~19s (six PTO doublings, single packet each) to ~5s
+            // (three rounds, two packets each). Strict server-side
+            // handshake-progress watchdogs (quic-go, msquic) tear
+            // the connection down at ~10s of silence regardless of
+            // our budget, so the half-time matters for landing in
+            // their tolerance.
+            //
+            // This test drives the EXACT helpers the send loop calls
+            // — [handlePtoFired] (initial PTO requeue + budget=2)
+            // followed by [requeueInflightForProbe] (the re-requeue
+            // step the loop performs between probe-bearing drains).
+            // If anyone unwires the budget or the re-requeue, this
+            // test breaks.
+            val client = newClientWithStartedHandshake()
+
+            val firstDrain = drainOutbound(client, nowMillis = 1L)
+            assertNotNull(firstDrain, "first drain emits the ClientHello datagram")
+
+            val firstSent =
+                client.initial.sentPackets.entries.firstOrNull { entry ->
+                    entry.value.tokens.any { it is RecoveryToken.Crypto }
+                }
+            assertNotNull(firstSent, "Initial-level SentPacket must carry a Crypto token")
+            val firstCrypto =
+                firstSent.value.tokens
+                    .filterIsInstance<RecoveryToken.Crypto>()
+                    .single()
+
+            // PTO fires.
+            handlePtoFired(client)
+            assertEquals(
+                2,
+                client.pendingProbePackets,
+                "handlePtoFired must seed [pendingProbePackets]=2 per RFC 9002 §6.2.4",
+            )
+
+            // Probe 1: drain consumes the requeued CRYPTO. Send loop
+            // decrements the budget and re-requeues for probe 2.
+            val probe1 = drainOutbound(client, nowMillis = 2L)
+            assertNotNull(probe1, "probe 1 must produce a datagram")
+            assertTrue(
+                probe1.size >= 1200,
+                "probe 1 carries Initial → padded to ≥ 1200 (got ${probe1.size})",
+            )
+            client.pendingProbePackets--
+            requeueInflightForProbe(client)
+
+            // Probe 2: a SECOND independent datagram with the same
+            // CRYPTO bytes at the same offset, allocated a fresh PN.
+            val probe2 = drainOutbound(client, nowMillis = 3L)
+            assertNotNull(probe2, "probe 2 must produce a SECOND datagram (RFC 9002 §6.2.4)")
+            assertTrue(
+                probe2.size >= 1200,
+                "probe 2 also Initial-bearing → padded to ≥ 1200 (got ${probe2.size})",
+            )
+            client.pendingProbePackets--
+
+            // Both probes must contain the original ClientHello CRYPTO
+            // at offset 0. Their wire bytes will differ (different PN
+            // → different AEAD nonce → different ciphertext) but the
+            // decrypted CRYPTO payload matches.
+            val firstFrames = decodeInitialFrames(firstDrain, client)
+            val probe1Frames = decodeInitialFrames(probe1, client)
+            val probe2Frames = decodeInitialFrames(probe2, client)
+            val firstHello =
+                firstFrames.filterIsInstance<CryptoFrame>().single { it.offset == firstCrypto.offset }
+            val probe1Hello =
+                probe1Frames.filterIsInstance<CryptoFrame>().firstOrNull { it.offset == firstCrypto.offset }
+            val probe2Hello =
+                probe2Frames.filterIsInstance<CryptoFrame>().firstOrNull { it.offset == firstCrypto.offset }
+            assertNotNull(probe1Hello, "probe 1 must carry CRYPTO at offset 0 — bare PING is not enough")
+            assertNotNull(probe2Hello, "probe 2 must ALSO carry CRYPTO at offset 0 (RFC 9002 §6.2.4)")
+            assertTrue(
+                probe1Hello.data.contentEquals(firstHello.data),
+                "probe 1 replays the original ClientHello bytes",
+            )
+            assertTrue(
+                probe2Hello.data.contentEquals(firstHello.data),
+                "probe 2 replays the SAME original ClientHello bytes (re-requeue preserves them)",
+            )
+
+            // The two probes must use DISTINCT packet numbers — same
+            // PN twice in the same encryption-level space would be
+            // RFC 9000 §17.2.5.1 protocol violation, and was the
+            // root cause of the picoquic-retry diagnostic earlier
+            // this session.
+            val sentPNs =
+                client.initial.sentPackets.keys
+                    .filter { it != firstSent.key }
+                    .toSet()
+            assertEquals(
+                2,
+                sentPNs.size,
+                "two probes must consume two distinct PNs (got $sentPNs)",
+            )
+
+            // After both probes, no more inflight CRYPTO to re-requeue
+            // → next drain returns null. The send loop's break path.
+            val tail = drainOutbound(client, nowMillis = 4L)
+            assertTrue(
+                tail == null || tail.isEmpty(),
+                "no third probe — budget exhausted (got ${tail?.size} bytes)",
+            )
+        }
+
     /**
      * Decode an Initial-level long-header packet from a freshly-drained
      * datagram and return its frames. We open with the client's own