debug(quic): periodic 5s flow-control snapshot from QuicWebTransportSession

Listener cliffs at uni stream #61 even though MAX_STREAMS_UNI(150)
was emitted at count=50 — and the cliff number is variable across
runs (124 in one trace, 61 in another), which strongly suggests the
limiter isn't the stream-id cap any more. Need visibility into what
QUIC flow-control state looks like at the moment streams stop.

QuicWebTransportSession now optionally takes a parentScope and, when
provided, launches a daemon coroutine that calls
QuicConnection.flowControlSnapshot() every 5 s. The snapshot dumps
the fields the cliff-investigation plan
(nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md)
identified as smoking-gun candidates:

  - peerInitiatedUniCount + advertisedMaxStreamsUni (stream-id cap)
  - peerInitMaxStreamsUni (peer's view of our cap, from handshake)
  - sendCredit + consumed (connection-level data flow control)
  - pendingBytes / pendingStreams (anything stuck in send buffers)
  - udpRecvDatagrams + udpRecvBytes (raw socket-level reception)

The factory passes its parentScope into the session so the snapshot
job dies cleanly with the same supervisor that owns the driver.
QuicWebTransportFactory tests and SendTraceScenario don't pass a
scope and won't see the periodic logger.

Filter `adb logcat -s NestQuic:D` and look for "snapshot peerInitiatedUni=…".
At cliff time the snapshot will print whichever counter is wedged.

https://claude.ai/code/session_01PYYez8a6sjiakyjAxsfCEQ

nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt

@@ -20,6 +20,7 @@
  */
 package com.vitorpamplona.nestsclient.transport
 
+import com.vitorpamplona.quartz.utils.Log
 import com.vitorpamplona.quic.connection.QuicConnection
 import com.vitorpamplona.quic.connection.QuicConnectionConfig
 import com.vitorpamplona.quic.connection.QuicConnectionDriver
@@ -37,9 +38,12 @@ import com.vitorpamplona.quic.webtransport.buildExtendedConnectHeaders
 import com.vitorpamplona.quic.webtransport.encodeHeadersFrame
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.launch
 
 /**
  * Pure-Kotlin WebTransport over QUIC v1, sitting on top of every layer in
@@ -181,7 +185,7 @@ class QuicWebTransportFactory(
             }
 
             val state = QuicWebTransportSessionState(conn, driver, requestStream.streamId)
-            return QuicWebTransportSession(state)
+            return QuicWebTransportSession(state, parentScope)
         } catch (we: WebTransportException) {
             throw we
         } catch (ce: kotlinx.coroutines.CancellationException) {
@@ -259,9 +263,43 @@ class QuicWebTransportFactory(
 /** Adapter that wraps the :quic [QuicWebTransportSessionState] in the nestsClient interface. */
 class QuicWebTransportSession(
     private val state: QuicWebTransportSessionState,
+    parentScope: CoroutineScope? = null,
 ) : WebTransportSession {
     override val isOpen: Boolean get() = state.isOpen
 
+    /**
+     * Periodic flow-control snapshot logger. Wakes every
+     * [SNAPSHOT_INTERVAL_MS] and dumps the QUIC layer's view of cap +
+     * count + send credit so we can correlate "audio cliff at uni
+     * stream #N" against "what did flow control look like at that
+     * moment". Cancelled when [close] tears down the session.
+     *
+     * Lazy-launched (via [parentScope]) only when the caller has
+     * provided a scope to host it — tests construct the session
+     * without one and don't need the logger.
+     */
+    private val snapshotJob: Job? =
+        parentScope?.launch {
+            while (true) {
+                delay(SNAPSHOT_INTERVAL_MS)
+                if (!state.isOpen) break
+                runCatching {
+                    val snap = state.connection.flowControlSnapshot()
+                    Log.d("NestQuic") {
+                        "snapshot peerInitiatedUni=${snap.peerInitiatedUniCount} " +
+                            "advertisedMaxStreamsUni=${snap.advertisedMaxStreamsUni} " +
+                            "peerInitMaxStreamsUni=${snap.peerInitialMaxStreamsUni} " +
+                            "sendCredit=${snap.sendConnectionFlowCredit} " +
+                            "consumed=${snap.sendConnectionFlowConsumed} " +
+                            "pendingBytes=${snap.totalEnqueuedNotSentBytes} " +
+                            "pendingStreams=${snap.streamsWithPendingBytes}/${snap.totalStreamsTracked} " +
+                            "udpRecvDatagrams=${snap.udp?.receivedDatagrams ?: -1L} " +
+                            "udpRecvBytes=${snap.udp?.receivedBytes ?: -1L}"
+                    }
+                }
+            }
+        }
+
     /**
      * Diagnostic-only passthrough to
      * [com.vitorpamplona.quic.connection.QuicConnection.flowControlSnapshot].
@@ -326,8 +364,19 @@ class QuicWebTransportSession(
         code: Int,
         reason: String,
     ) {
+        snapshotJob?.cancel()
         state.close(code, reason)
     }
+
+    private companion object {
+        /**
+         * Cadence for the periodic flow-control snapshot. 5 s is short
+         * enough to catch the audio-cliff transition (which lands within
+         * 5–15 s of subscribe) and long enough that the snapshot itself
+         * isn't a meaningful CPU cost.
+         */
+        const val SNAPSHOT_INTERVAL_MS = 5_000L
+    }
 }
 
 private class QuicBidiStreamAdapter(