feat(quic,nestsclient): plumb RESET_STREAM + STOP_SENDING through WebTransport

Extends the WebTransport interface and its three implementations
(QUIC-backed, in-memory fake, peer-stripped demux) with typed
error-code cancel paths so moq-lite Lite-03's M2/M3 audit items can
land.

QUIC layer (`:quic`):
  - `StrippedWtStream` gains optional `reset` and `stopSending`
    closures, wired by `WtPeerStreamDemux.emitStripped` through
    `QuicStream.resetStream(errorCode)` /
    `QuicStream.stopSending(errorCode)` + `driver.wakeup()` so the
    frames actually leave the connection. `reset` is null on
    peer-initiated uni streams (no send half on our side);
    `stopSending` is unconditional (every surfaced stream has a
    read side).

WebTransport interface (`:nestsClient`):
  - `WebTransportReadStream.stopSending(errorCode: Long)` —
    suspending; first-call-wins.
  - `WebTransportWriteStream.reset(errorCode: Long)` —
    suspending; first-call-wins. Distinct from the existing
    graceful `finish()` / FIN.

Adapters:
  - `QuicBidiStreamAdapter`, `QuicReadStreamAdapter`,
    `QuicUniWriteStreamAdapter` route directly to the underlying
    `QuicStream` API.
  - `StrippedWtReadStreamAdapter` /
    `StrippedWtBidiStreamAdapter` route through the demux's
    closures; defensive `?:` fallbacks defend against a future
    demux bug that forgets to wire a closure.
  - `FakeBidiStream`, `FakeReadStream`, `ChannelWriteStream` (the
    in-memory test transport) record reset / stopSending codes
    in shared `AtomicLong` cells so tests can assert "the peer
    reset with code X" / "the listener stopSending with code Y"
    via new `lastResetCode` / `lastStopSendingCode` /
    `lastPeerResetCode` / `lastPeerStopSendingCode` properties.
    Sentinel `NO_CODE = Long.MIN_VALUE` distinguishes "not called"
    from a legitimate code 0.

Pure plumbing — no moq-lite call sites use the new methods yet.
Subsequent commits wire M3 (publisher Drop reply uses
reset(errorCode)) and M2 (listener group-cancel uses
stopSending(errorCode)).

Audit doc: nestsClient/plans/2026-05-09-moq-lite-rfc-compliance.md (M2 + M3).

https://claude.ai/code/session_012TGfo99Ugz7fcCPv85a8Tq

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt

@@ -57,8 +57,40 @@ class FakeWebTransport private constructor(
         stateLock.withLock { check(open) { "session closed" } }
         val localToPeer = Channel<ByteArray>(Channel.BUFFERED)
         val peerToLocal = Channel<ByteArray>(Channel.BUFFERED)
-        val local = FakeBidiStream(write = localToPeer, read = peerToLocal)
-        val peer = FakeBidiStream(write = peerToLocal, read = localToPeer)
+        // Shared error-code cells: each side records its own reset()
+        // and stopSending() codes here so the peer can introspect.
+        // `localResetCell` records resets on `local.reset(code)` (=
+        // peer's `lastPeerResetCode`), and vice-versa.
+        val localResetCell =
+            java.util.concurrent.atomic
+                .AtomicLong(NO_CODE)
+        val peerResetCell =
+            java.util.concurrent.atomic
+                .AtomicLong(NO_CODE)
+        val localStopSendingCell =
+            java.util.concurrent.atomic
+                .AtomicLong(NO_CODE)
+        val peerStopSendingCell =
+            java.util.concurrent.atomic
+                .AtomicLong(NO_CODE)
+        val local =
+            FakeBidiStream(
+                write = localToPeer,
+                read = peerToLocal,
+                myResetCell = localResetCell,
+                myStopSendingCell = localStopSendingCell,
+                peerResetCell = peerResetCell,
+                peerStopSendingCell = peerStopSendingCell,
+            )
+        val peer =
+            FakeBidiStream(
+                write = peerToLocal,
+                read = localToPeer,
+                myResetCell = peerResetCell,
+                myStopSendingCell = peerStopSendingCell,
+                peerResetCell = localResetCell,
+                peerStopSendingCell = localStopSendingCell,
+            )
         // Send *peer* half to the other side's inbound queue.
         outboundBidiStreams.send(peer)
         return local
@@ -87,8 +119,20 @@ class FakeWebTransport private constructor(
         val priorityCell =
             java.util.concurrent.atomic
                 .AtomicInteger(0)
-        outboundUniStreams.send(FakeReadStream(pipe, priorityCell))
-        return ChannelWriteStream(pipe, priorityCell)
+        // Shared reset / stopSending cells. The writer (us) records its
+        // `reset(code)` call here; the peer-side reader exposes it via
+        // `FakeReadStream.lastResetCode`. Symmetric: the reader's
+        // `stopSending(code)` call lands in `stopSendingCell` so the
+        // writer (= moq-lite publisher tests) can assert the listener
+        // canceled with the expected code.
+        val resetCell =
+            java.util.concurrent.atomic
+                .AtomicLong(NO_CODE)
+        val stopSendingCell =
+            java.util.concurrent.atomic
+                .AtomicLong(NO_CODE)
+        outboundUniStreams.send(FakeReadStream(pipe, priorityCell, resetCell, stopSendingCell))
+        return ChannelWriteStream(pipe, priorityCell, resetCell, stopSendingCell)
     }
 
     override suspend fun sendDatagram(payload: ByteArray): Boolean {
@@ -161,6 +205,20 @@ class FakeWebTransport private constructor(
 class FakeBidiStream internal constructor(
     private val write: Channel<ByteArray>,
     private val read: Channel<ByteArray>,
+    /**
+     * Cell that records this side's `reset(code)` calls. Read by
+     * [lastResetCode] from the same side (rare — usually it's the
+     * peer's [peerResetCell] you want).
+     */
+    private val myResetCell: java.util.concurrent.atomic.AtomicLong? = null,
+    private val myStopSendingCell: java.util.concurrent.atomic.AtomicLong? = null,
+    /**
+     * Cell that records the *peer's* `reset(code)` calls (i.e.
+     * mirror of the peer's [myResetCell]). [lastPeerResetCode] reads
+     * this so tests can assert "the peer reset the bidi with code X."
+     */
+    private val peerResetCell: java.util.concurrent.atomic.AtomicLong? = null,
+    private val peerStopSendingCell: java.util.concurrent.atomic.AtomicLong? = null,
 ) : WebTransportBidiStream {
     override fun incoming(): Flow<ByteArray> = read.receiveAsFlow()
 
@@ -172,6 +230,40 @@ class FakeBidiStream internal constructor(
         write.close()
     }
 
+    override suspend fun reset(errorCode: Long) {
+        // First call wins, mirroring `:quic`'s `QuicStream.resetStream`
+        // contract. Subsequent reset/finish/write are silently
+        // dropped to keep the wire frame stable in the eyes of any
+        // test that reads `lastResetCode`.
+        myResetCell?.compareAndSet(NO_CODE, errorCode)
+        write.close()
+    }
+
+    override suspend fun stopSending(errorCode: Long) {
+        myStopSendingCell?.compareAndSet(NO_CODE, errorCode)
+        // The peer's send half is our receive half; close it so a
+        // collector sees end-of-flow.
+        read.close()
+    }
+
+    /**
+     * Code our side passed to [reset], or `null` if reset wasn't
+     * called. Mostly useful when a test holds the peer's reference;
+     * [lastPeerResetCode] is the more common assertion.
+     */
+    val lastResetCode: Long?
+        get() = myResetCell?.get()?.takeIf { it != NO_CODE }
+
+    /** Code the peer side passed to [reset], or `null` if not called. */
+    val lastPeerResetCode: Long?
+        get() = peerResetCell?.get()?.takeIf { it != NO_CODE }
+
+    val lastStopSendingCode: Long?
+        get() = myStopSendingCell?.get()?.takeIf { it != NO_CODE }
+
+    val lastPeerStopSendingCode: Long?
+        get() = peerStopSendingCell?.get()?.takeIf { it != NO_CODE }
+
     override fun setPriority(priority: Int) = Unit
 }
 
@@ -184,9 +276,33 @@ class FakeReadStream internal constructor(
      * uni-stream writer (e.g. tests that hand-roll a [Channel]).
      */
     private val priorityCell: java.util.concurrent.atomic.AtomicInteger? = null,
+    /**
+     * Optional shared cell that records the writer's `reset(code)`
+     * call. Exposed via [lastResetCode] so tests can assert the
+     * publisher reset its uni stream with the expected typed error.
+     */
+    private val resetCell: java.util.concurrent.atomic.AtomicLong? = null,
+    /**
+     * Optional shared cell that records this side's own
+     * `stopSending(code)` call. Exposed via [lastStopSendingCode]
+     * so the test can assert the listener actually fired the cancel.
+     */
+    private val stopSendingCell: java.util.concurrent.atomic.AtomicLong? = null,
 ) : WebTransportReadStream {
     override fun incoming(): Flow<ByteArray> = read.receiveAsFlow()
 
+    override suspend fun stopSending(errorCode: Long) {
+        // First-call-wins like the production path. Closing `read`
+        // here would race the production path's "publisher closes
+        // its write end" semantic — moq-lite calls `stopSending`
+        // when it's done caring about the bytes, but the publisher
+        // is still expected to finish what it was sending. We
+        // record the code without closing so a test that asserts
+        // `incoming().toList()` post-stopSending still sees any
+        // bytes the publisher had already enqueued.
+        stopSendingCell?.compareAndSet(NO_CODE, errorCode)
+    }
+
     /**
      * Last priority value the paired write side applied via
      * [WebTransportWriteStream.setPriority], or `0` if no priority was
@@ -195,6 +311,17 @@ class FakeReadStream internal constructor(
      */
     val lastSetPriority: Int?
         get() = priorityCell?.get()
+
+    /**
+     * Code the peer-side writer passed to [WebTransportWriteStream.reset],
+     * or `null` if reset wasn't called.
+     */
+    val lastResetCode: Long?
+        get() = resetCell?.get()?.takeIf { it != NO_CODE }
+
+    /** Code we passed to [stopSending], or `null` if not called. */
+    val lastStopSendingCode: Long?
+        get() = stopSendingCell?.get()?.takeIf { it != NO_CODE }
 }
 
 /**
@@ -210,6 +337,8 @@ private class ChannelWriteStream(
      * stores each [setPriority] call so the peer can introspect.
      */
     private val priorityCell: java.util.concurrent.atomic.AtomicInteger? = null,
+    private val resetCell: java.util.concurrent.atomic.AtomicLong? = null,
+    private val stopSendingCell: java.util.concurrent.atomic.AtomicLong? = null,
 ) : WebTransportWriteStream {
     override suspend fun write(chunk: ByteArray) {
         channel.send(chunk)
@@ -219,7 +348,28 @@ private class ChannelWriteStream(
         channel.close()
     }
 
+    override suspend fun reset(errorCode: Long) {
+        // First-call-wins. We close the channel so the peer's
+        // receive flow ends — same shape as `finish` but with a
+        // typed error code recorded for the peer to introspect.
+        resetCell?.compareAndSet(NO_CODE, errorCode)
+        channel.close()
+    }
+
     override fun setPriority(priority: Int) {
         priorityCell?.set(priority)
     }
+
+    @Suppress("unused")
+    private val unusedStopSendingCellTether: java.util.concurrent.atomic.AtomicLong? = stopSendingCell
 }
+
+/**
+ * Sentinel for "no reset / stopSending code recorded yet" in the
+ * [java.util.concurrent.atomic.AtomicLong] cells used by [FakeBidiStream]
+ * and [FakeReadStream] / [ChannelWriteStream]. Real moq-lite codes are
+ * non-negative varints (RFC 9000 application error codes are u32
+ * unsigned-fitting-in-Long); `Long.MIN_VALUE` is safely outside that
+ * range.
+ */
+internal const val NO_CODE: Long = Long.MIN_VALUE

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/WebTransportSession.kt

@@ -104,6 +104,30 @@ interface WebTransportBidiStream :
 interface WebTransportReadStream {
     /** Flow of chunks as they arrive. Completes when the peer closes its write side. */
     fun incoming(): Flow<ByteArray>
+
+    /**
+     * RFC 9000 §3.5: ask the peer to stop sending on this stream.
+     * Causes a `STOP_SENDING(applicationErrorCode)` frame to land at
+     * the peer, which typically responds with `RESET_STREAM` carrying
+     * the same code. Call this when the application no longer needs
+     * the stream's bytes — it lets the peer abandon any pending
+     * retransmits instead of wasting bandwidth on data we'd discard.
+     *
+     * First call wins per `:quic`'s `QuicStream.stopSending`
+     * lock-free first-call-wins gate; subsequent calls (including
+     * with a different code) are silently ignored to keep the wire
+     * frame stable.
+     *
+     * Used by moq-lite Lite-03's group-cancel path: a listener that
+     * decides a specific group is too stale can `stopSending` its
+     * uni stream so the publisher abandons any in-flight retransmits
+     * instead of wasting bandwidth on bytes the listener will discard.
+     *
+     * Implementations that don't model receive-side cancellation
+     * (e.g. the in-memory fake when no test asserts on it) MAY
+     * treat this as a no-op.
+     */
+    suspend fun stopSending(errorCode: Long)
 }
 
 /** Write-only WebTransport stream. */
@@ -113,6 +137,31 @@ interface WebTransportWriteStream {
     /** Half-close the write side (FIN). No further writes after this call. */
     suspend fun finish()
 
+    /**
+     * RFC 9000 §3.5: send `RESET_STREAM(applicationErrorCode)` on
+     * this stream's send half, abandoning any pending bytes. Distinct
+     * from [finish], which is a graceful FIN — `reset` carries a
+     * typed error code the peer can act on.
+     *
+     * First call wins per `:quic`'s `QuicStream.resetStream`
+     * lock-free first-call-wins gate; subsequent calls (and any
+     * subsequent [write] / [finish]) are silently ignored to keep
+     * the wire frame stable.
+     *
+     * Used by moq-lite Lite-03's typed cancel paths: a publisher
+     * that rejects a Subscribe (e.g. broadcast / track does not
+     * exist) follows the SubscribeDrop body with a
+     * `RESET_STREAM(errorCode)` so the subscriber sees a typed
+     * application reason rather than the ambiguous "publisher FINed
+     * the bidi" signal that overlaps with a graceful publisher
+     * shutdown.
+     *
+     * Implementations that don't model sender-driven reset (e.g.
+     * the in-memory fake when no test asserts on it) MAY treat this
+     * as a [finish].
+     */
+    suspend fun reset(errorCode: Long)
+
     /**
      * Hint to the transport about this stream's drain priority relative
      * to other streams on the same session. Higher value drains first

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

@@ -366,6 +366,16 @@ private class QuicBidiStreamAdapter(
         driver.wakeup()
     }
 
+    override suspend fun reset(errorCode: Long) {
+        stream.resetStream(errorCode)
+        driver.wakeup()
+    }
+
+    override suspend fun stopSending(errorCode: Long) {
+        stream.stopSending(errorCode)
+        driver.wakeup()
+    }
+
     override fun setPriority(priority: Int) {
         stream.priority = priority
     }
@@ -373,8 +383,14 @@ private class QuicBidiStreamAdapter(
 
 private class QuicReadStreamAdapter(
     private val stream: QuicStream,
+    private val driver: com.vitorpamplona.quic.connection.QuicConnectionDriver,
 ) : WebTransportReadStream {
     override fun incoming(): Flow<ByteArray> = stream.incoming
+
+    override suspend fun stopSending(errorCode: Long) {
+        stream.stopSending(errorCode)
+        driver.wakeup()
+    }
 }
 
 /**
@@ -397,6 +413,11 @@ private class QuicUniWriteStreamAdapter(
         driver.wakeup()
     }
 
+    override suspend fun reset(errorCode: Long) {
+        stream.resetStream(errorCode)
+        driver.wakeup()
+    }
+
     override fun setPriority(priority: Int) {
         stream.priority = priority
     }
@@ -407,13 +428,24 @@ private class StrippedWtReadStreamAdapter(
     private val stripped: com.vitorpamplona.quic.webtransport.StrippedWtStream,
 ) : WebTransportReadStream {
     override fun incoming(): Flow<ByteArray> = stripped.data
+
+    override suspend fun stopSending(errorCode: Long) {
+        // Demux unconditionally wires `stopSending` for every surfaced
+        // stream — the contract in `StrippedWtStream` is "non-null on
+        // streams with a read side", which is true here. The `?:`
+        // fallback exists only to defend against a future demux bug
+        // that forgets to wire it; Unit is a safe no-op for tests
+        // that mock a stripped stream without a real receiver.
+        val ss = stripped.stopSending ?: return
+        ss(errorCode)
+    }
 }
 
 /**
  * Adapter for a peer-initiated bidi WT stream whose WT_BIDI_STREAM prefix
- * has been stripped. Routes [write] / [finish] through the demux's
- * driver-aware closures so application bytes actually leave the
- * connection.
+ * has been stripped. Routes [write] / [finish] / [reset] / [stopSending]
+ * through the demux's driver-aware closures so application bytes actually
+ * leave the connection.
  */
 private class StrippedWtBidiStreamAdapter(
     private val stripped: com.vitorpamplona.quic.webtransport.StrippedWtStream,
@@ -440,13 +472,28 @@ private class StrippedWtBidiStreamAdapter(
         finish()
     }
 
+    override suspend fun reset(errorCode: Long) {
+        // Bidi streams have a send side we own, so `reset` is wired by
+        // the demux. Same defensive `?:` shape as in `write` / `finish`.
+        val r =
+            stripped.reset
+                ?: error("peer-initiated bidi stream has no reset — demux didn't wire one")
+        r(errorCode)
+    }
+
+    override suspend fun stopSending(errorCode: Long) {
+        val ss = stripped.stopSending ?: return
+        ss(errorCode)
+    }
+
     /**
      * No-op: peer-initiated bidi streams arrive through the demux as a
      * [com.vitorpamplona.quic.webtransport.StrippedWtStream] which exposes
-     * only `send`/`finish` closures, not the underlying [QuicStream]. The
-     * moq-lite priority use case targets locally-opened uni group streams
-     * only, so this path doesn't need to model priority — see the
-     * [WebTransportWriteStream.setPriority] contract.
+     * only `send`/`finish`/`reset`/`stopSending` closures, not the
+     * underlying [QuicStream]. The moq-lite priority use case targets
+     * locally-opened uni group streams only, so this path doesn't need
+     * to model priority — see the [WebTransportWriteStream.setPriority]
+     * contract.
      */
     override fun setPriority(priority: Int) = Unit
 }