quic: expose flow-control snapshot for prod cliff investigation

Adds a read-only diagnostic surface that lets a test (or any caller)
read the peer's transport parameters, the live connection-level send
credit / consumed counters, the current peer-granted MAX_STREAMS_*
values, and the total bytes sitting in stream send buffers but not
yet handed to STREAM frames.

Goal: pin which budget runs out at the production "stream cliff"
described in nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md.
The plan flagged three candidates — connection-level MAX_DATA, per-
stream MAX_STREAM_DATA, or the relay's MAX_STREAMS_UNI extension
policy. The snapshot makes it possible to attribute the stall by
reading the diff between the pre-pump, post-pump, and post-grace
snapshots from the test's stdout.

QuicConnection:
  - flowControlSnapshot(): suspend, lock-protected, returns
    QuicFlowControlSnapshot (new data class) — peer TPs + live
    accounting + sum of enqueued-not-sent bytes across streams.

QuicWebTransportSession (jvmAndroid adapter):
  - quicFlowControlSnapshot() passthrough so the test can downcast
    its WebTransportSession and read the underlying connection's
    state without poking through the common transport interface.

SendTraceScenario:
  - Optional flowControlSnapshot lambda parameter; when supplied,
    logs three checkpoints — fc-pre, fc-post-pump, fc-post-grace —
    each on a single line with the full snapshot.

NostrnestsProdAudioTransmissionTest + NostrNestsSustainedSendOutcomesInteropTest:
  - withProdSpeakerAndListeners / withHarnessSpeakerAndListeners now
    yield a snapshot lambda to the scenario block. Every sweep test
    automatically dumps fc-* lines to the JUnit XML system-out.

FlowControlSnapshotTest:
  - Pre-handshake: peer TP fields are null, counters zero.
  - Post-handshake: every TP field reflects what the in-process TLS
    server advertised; sendConnectionFlowCredit equals
    initial_max_data; consumed = 0.
  - Post-allocate-and-enqueue: nextLocalUni/BidiIndex advance,
    totalEnqueuedNotSentBytes sums the buffered chunks, and
    streamsWithPendingBytes counts only streams with > 0 pending.

Reading the production sweep output after this commit:
  - fc-pre dumps what the relay grants on handshake (initial_max_data,
    initial_max_stream_data_uni, initial_max_streams_uni).
  - fc-post-pump shows whether sendConnectionFlowConsumed has
    plateaued at the cap (peer didn't extend MAX_DATA) or whether
    bytes are stuck in stream send buffers.
  - The diff between fc-post-pump and fc-post-grace tells us
    whether the relay's eventual MAX_DATA / MAX_STREAMS update did
    or didn't arrive during the 30-60 s grace window.

nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsSustainedSendOutcomesInteropTest.kt

@@ -386,7 +386,7 @@ class NostrNestsSustainedSendOutcomesInteropTest {
     ) = runBlocking {
         NostrNestsHarness.assumeNestsInterop()
         val harness = harnessOrNull ?: return@runBlocking
-        withHarnessSpeakerAndListeners(scope, harness, scenario.parallelSubscriptions) { publisher, listeners, hostPub, pumpScope ->
+        withHarnessSpeakerAndListeners(scope, harness, scenario.parallelSubscriptions) { publisher, listeners, hostPub, pumpScope, flowControlSnapshot ->
             val result =
                 SendTraceScenario.run(
                     scope = scope,
@@ -395,6 +395,7 @@ class NostrNestsSustainedSendOutcomesInteropTest {
                     speakerPubkeyHex = hostPub,
                     scenario = scenario,
                     pumpScope = pumpScope,
+                    flowControlSnapshot = flowControlSnapshot,
                 )
             SendTraceScenario.reportAndAssert(scope, result, expectAllReceived)
         }
@@ -410,6 +411,7 @@ class NostrNestsSustainedSendOutcomesInteropTest {
             listeners: List<com.vitorpamplona.nestsclient.NestsListener>,
             speakerPubkeyHex: String,
             pumpScope: CoroutineScope,
+            flowControlSnapshot: suspend () -> com.vitorpamplona.quic.connection.QuicFlowControlSnapshot,
         ) -> Unit,
     ) {
         val hostSigner = NostrSignerInternal(KeyPair())
@@ -468,7 +470,17 @@ class NostrNestsSustainedSendOutcomesInteropTest {
                 InteropDebug.assertListenerReached(scope, "Connected", listener.state.value)
                 listeners += listener
             }
-            block(publisher, listeners, hostSigner.pubKey, pumpScope)
+            // Diagnostics passthrough mirrors withProdSpeakerAndListeners
+            // — see that helper's comment for rationale.
+            val quicSpeakerWt =
+                speakerWt as com.vitorpamplona.nestsclient.transport.QuicWebTransportSession
+            block(
+                publisher,
+                listeners,
+                hostSigner.pubKey,
+                pumpScope,
+                { quicSpeakerWt.quicFlowControlSnapshot() },
+            )
         } finally {
             for (listener in listeners) {
                 runCatching { listener.close() }

nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt

@@ -944,7 +944,7 @@ class NostrnestsProdAudioTransmissionTest {
         expectAllReceived: Boolean = false,
     ) = runBlocking {
         assumeProd()
-        withProdSpeakerAndListeners(scope, scenario.parallelSubscriptions) { publisher, listeners, hostPub, pumpScope ->
+        withProdSpeakerAndListeners(scope, scenario.parallelSubscriptions) { publisher, listeners, hostPub, pumpScope, flowControlSnapshot ->
             val result =
                 SendTraceScenario.run(
                     scope = scope,
@@ -953,6 +953,7 @@ class NostrnestsProdAudioTransmissionTest {
                     speakerPubkeyHex = hostPub,
                     scenario = scenario,
                     pumpScope = pumpScope,
+                    flowControlSnapshot = flowControlSnapshot,
                 )
             SendTraceScenario.reportAndAssert(scope, result, expectAllReceived)
         }
@@ -976,6 +977,7 @@ class NostrnestsProdAudioTransmissionTest {
             listeners: List<com.vitorpamplona.nestsclient.NestsListener>,
             speakerPubkeyHex: String,
             pumpScope: CoroutineScope,
+            flowControlSnapshot: suspend () -> com.vitorpamplona.quic.connection.QuicFlowControlSnapshot,
         ) -> Unit,
     ) {
         val hostSigner = NostrSignerInternal(KeyPair())
@@ -1030,7 +1032,19 @@ class NostrnestsProdAudioTransmissionTest {
                 listeners += listener
             }
 
-            block(publisher, listeners, hostSigner.pubKey, pumpScope)
+            // Diagnostics passthrough: cast the speaker WT to the
+            // concrete QUIC adapter so the scenario can read the
+            // underlying connection's flow-control snapshot. See
+            // `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`.
+            val quicSpeakerWt =
+                speakerWt as com.vitorpamplona.nestsclient.transport.QuicWebTransportSession
+            block(
+                publisher,
+                listeners,
+                hostSigner.pubKey,
+                pumpScope,
+                { quicSpeakerWt.quicFlowControlSnapshot() },
+            )
         } finally {
             for (listener in listeners) {
                 runCatching { listener.close() }

nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/SendTraceScenario.kt

@@ -177,6 +177,16 @@ data class ScenarioResult(
  *   - Tear down all of those after [run] returns.
  */
 object SendTraceScenario {
+    /**
+     * @param flowControlSnapshot optional supplier that returns the
+     *   speaker-side QUIC connection's flow-control state at the
+     *   moment of the call. When provided, the scenario captures and
+     *   logs three snapshots — before the pump, immediately after the
+     *   pump completes, and after the receive-grace window — so a
+     *   failure mode like "data piled up but conn-level credit ran
+     *   out" can be diagnosed from the test report alone. See
+     *   [com.vitorpamplona.quic.connection.QuicConnection.flowControlSnapshot].
+     */
     suspend fun run(
         scope: String,
         publisher: MoqLitePublisherHandle,
@@ -184,11 +194,25 @@ object SendTraceScenario {
         speakerPubkeyHex: String,
         scenario: Scenario,
         pumpScope: CoroutineScope,
+        flowControlSnapshot: (suspend () -> com.vitorpamplona.quic.connection.QuicFlowControlSnapshot)? = null,
     ): ScenarioResult {
         require(listeners.size == scenario.parallelSubscriptions) {
             "expected ${scenario.parallelSubscriptions} listener(s), got ${listeners.size}"
         }
         InteropDebug.checkpoint(scope, "scenario=$scenario speaker=${speakerPubkeyHex.take(8)}…")
+        flowControlSnapshot?.invoke()?.let { snap ->
+            InteropDebug.checkpoint(
+                scope,
+                "fc-pre: peerInitMaxData=${snap.peerInitialMaxData} " +
+                    "peerInitMaxStreamDataUni=${snap.peerInitialMaxStreamDataUni} " +
+                    "peerInitMaxStreamsUni=${snap.peerInitialMaxStreamsUni} " +
+                    "sendCredit=${snap.sendConnectionFlowCredit} consumed=${snap.sendConnectionFlowConsumed} " +
+                    "peerMaxStreamsUniNow=${snap.peerMaxStreamsUniCurrent} " +
+                    "nextLocalUniIdx=${snap.nextLocalUniIndex} " +
+                    "pendingBytes=${snap.totalEnqueuedNotSentBytes} " +
+                    "pendingStreams=${snap.streamsWithPendingBytes}/${snap.totalStreamsTracked}",
+            )
+        }
 
         val sendOutcomes = BooleanArray(scenario.frameCount)
         val sendDurationsMicros = LongArray(scenario.frameCount)
@@ -283,6 +307,18 @@ object SendTraceScenario {
                 "(target=${scenario.frameCount * scenario.cadenceMs}ms) " +
                 "sendTrue=${sendOutcomes.count { it }}/${scenario.frameCount}",
         )
+        flowControlSnapshot?.invoke()?.let { snap ->
+            InteropDebug.checkpoint(
+                scope,
+                "fc-post-pump: sendCredit=${snap.sendConnectionFlowCredit} " +
+                    "consumed=${snap.sendConnectionFlowConsumed} " +
+                    "(remaining=${snap.sendConnectionFlowCredit - snap.sendConnectionFlowConsumed}) " +
+                    "peerMaxStreamsUniNow=${snap.peerMaxStreamsUniCurrent} " +
+                    "nextLocalUniIdx=${snap.nextLocalUniIndex} " +
+                    "pendingBytes=${snap.totalEnqueuedNotSentBytes} " +
+                    "pendingStreams=${snap.streamsWithPendingBytes}/${snap.totalStreamsTracked}",
+            )
+        }
 
         // Wait for collectors. If they hit `take(N)` they exit naturally;
         // otherwise the per-collector withTimeoutOrNull cancels them.
@@ -292,6 +328,18 @@ object SendTraceScenario {
             }
             if (job.isActive) job.cancelAndJoin()
         }
+        flowControlSnapshot?.invoke()?.let { snap ->
+            InteropDebug.checkpoint(
+                scope,
+                "fc-post-grace: sendCredit=${snap.sendConnectionFlowCredit} " +
+                    "consumed=${snap.sendConnectionFlowConsumed} " +
+                    "(remaining=${snap.sendConnectionFlowCredit - snap.sendConnectionFlowConsumed}) " +
+                    "peerMaxStreamsUniNow=${snap.peerMaxStreamsUniCurrent} " +
+                    "nextLocalUniIdx=${snap.nextLocalUniIndex} " +
+                    "pendingBytes=${snap.totalEnqueuedNotSentBytes} " +
+                    "pendingStreams=${snap.streamsWithPendingBytes}/${snap.totalStreamsTracked}",
+            )
+        }
 
         return ScenarioResult(
             scenario = scenario,