Run the differential over all five carrier combinations

Both implementations behaved identically on every case, over tcp/tcp, tcp/tcp
with Mux, udp/udp, udp/tcp and tcp/udp. That is what row 25 asked for, and it is
where this project has twice found a protocol fact that had grown a second shape
the moment a second carrier landed.

The differential is a JVM test and QUIC lived only in an Android library, so
build-host-quic.sh compiles the same JNI sources the app ships against the build
machine. lwIP stays out of it: it is built NO_SYS against an Android port and has
no business running on a host, while the bridge depends on nothing Android at
all. The QUIC rows are skipped, loudly, when no bridge was built — two pairs
reported as four would be worse than two reported as two.

Split flows needed something the duplex ones did not, and it is the most
interesting thing here: **the framing of a UDP flow belongs to a direction, not
to a flow.** With up=udp&down=tcp the uplink sends DATAGRAMs while the downlink
is a TLS stream carrying length-prefixed packets, and with up=tcp&down=udp it is
the other way round. Framing both halves the same way is a flow that opens and
then carries nothing, which is exactly how the two split uot_round_trip cases
diverged before this.

So section 9's carriage moved out of QuicCarrier into DatagramLane, which both
carriers use. Writing it twice is how a protocol fact grows a second shape, and
the case that would have caught it is the one that found it.

One more defect the differential found on its own terms: its UDP case always
framed for a stream, so quic_uot_round_trip diverged with an
IllegalStateException while the oracle echoed 512 bytes. The case now asks the
flow which framing it wants — the same question UdpRelay asks, asked the same
way, because a case that framed for one carrier and ran on the other reports a
divergence that is only its own assumption.

The script's carrier-pairs line stopped saying "tcp/tcp only. up=udp and down=udp
select QUIC, which L1 does not implement." It names what was actually run, and
still says so when a host has no bridge.

L3: twenty-four covered, one partly, two blocked. The two left are rows 26 and
27, and they wait on the same thing — a physical device. Nothing else in L3 is
blocked on code.

Gates green; device suite 49 cases; differential 27 cases across five carriers.

Author: AlbertoDeckow

app/build.gradle.kts

@@ -590,6 +590,18 @@ val nativeLibraryChecks =
 
 tasks.named("check") { dependsOn(nativeLibraryChecks) }
 
+// The host QUIC bridge, when one was built.
+//
+// `conformance/scripts/build-host-quic.sh` compiles the same JNI sources the
+// app ships against this machine, so the oracle differential can compare the
+// two implementations over QUIC without an emulator in the loop. Absent is the
+// ordinary case and the QUIC rows are then skipped, loudly.
+tasks.withType<Test>().configureEach {
+    System.getenv("SOMEWHERE_QUIC_LIBRARY")?.takeIf { it.isNotBlank() }?.let {
+        systemProperty("somewhere.quic.library", it)
+    }
+}
+
 val classpathConsistencyTasks =
     listOf("Debug" to "debug", "Release" to "release").map { (capitalised, lowercase) ->
         tasks.register<ClasspathConsistencyTask>("checkClasspathConsistency$capitalised") {

app/src/main/kotlin/eu/nodepass/somewhere/protocol/session/DatagramLane.kt

@@ -0,0 +1,150 @@
+// SPDX-License-Identifier: GPL-3.0-only
+// Copyright (C) 2026 The Somewhere Authors
+
+package eu.nodepass.somewhere.protocol.session
+
+import eu.nodepass.somewhere.protocol.DecodeResult
+import eu.nodepass.somewhere.protocol.quic.DatagramFrame
+import eu.nodepass.somewhere.protocol.quic.DatagramReassembler
+import eu.nodepass.somewhere.protocol.quic.PacketIds
+import eu.nodepass.somewhere.protocol.quic.QuicDatagram
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.LinkedBlockingQueue
+
+/**
+ * Section 9's UDP carriage, shared by the carriers that can use it.
+ *
+ * ## Why this is not inside one carrier
+ *
+ * A duplex QUIC flow sends and receives packets as DATAGRAMs. A **split** flow
+ * may do it in one direction only: with `up=udp&down=tcp` the uplink is QUIC and
+ * the downlink is a TLS stream carrying length-prefixed packets, and with
+ * `up=tcp&down=udp` it is the other way round. So the framing belongs to a
+ * *direction*, not to a flow, and putting it inside either carrier would mean
+ * writing it twice — which is how a protocol fact grows a second shape.
+ *
+ * ## One demultiplexer
+ *
+ * One connection's datagrams carry every flow's traffic, so something has to
+ * route them and it has to be one thing: two readers pulling from the
+ * connection would each discard what belonged to the other.
+ */
+internal class DatagramLane(
+    private val datagrams: QuicCarrier.Datagrams,
+) {
+    private val inbound = ConcurrentHashMap<UInt, LinkedBlockingQueue<ByteArray>>()
+    private val packetIds = ConcurrentHashMap<UInt, PacketIds>()
+    private val reassembler = DatagramReassembler()
+    private val demultiplex = Any()
+
+    /** Row 15: payload is refused until a Portal has answered a flow. */
+    fun markReady(flowId: UInt) {
+        inbound.putIfAbsent(flowId, LinkedBlockingQueue())
+        packetIds.putIfAbsent(flowId, PacketIds())
+        reassembler.markReady()
+    }
+
+    fun forget(flowId: UInt) {
+        inbound.remove(flowId)
+        packetIds.remove(flowId)
+    }
+
+    /** The largest packet one datagram can carry, or 0 when unsupported. */
+    fun maxDatagram(): Int = datagrams.maxDatagram()
+
+    /**
+     * Sends one packet, fragmenting when the path cannot carry it whole.
+     *
+     * Lossy by design: a packet that cannot be planned is dropped, exactly as a
+     * UDP socket drops one it cannot send. Telling the caller would be adding a
+     * guarantee the transport underneath does not have.
+     */
+    fun send(
+        flowId: UInt,
+        payload: ByteArray,
+    ) {
+        val maxDatagram = datagrams.maxDatagram()
+        when (val planned = QuicDatagram.plan(payload.size, maxDatagram)) {
+            is DecodeResult.Invalid -> Unit
+            is DecodeResult.Ok ->
+                if (planned.value == 1) {
+                    datagrams.send(QuicDatagram.Data(flowId, payload).encode())
+                } else {
+                    // A fresh packet id per packet, and per replan: a peer
+                    // reassembling two layouts under one id would produce a
+                    // packet that is neither.
+                    val packetId = (packetIds[flowId] ?: PacketIds().also { packetIds[flowId] = it }).allocate()
+                    val perFragment = maxDatagram - QuicDatagram.FRAGMENT_HEADER_SIZE
+                    for (index in 0 until planned.value) {
+                        val from = index * perFragment
+                        val to = minOf(from + perFragment, payload.size)
+                        datagrams.send(
+                            QuicDatagram
+                                .Fragment(
+                                    flowId = flowId,
+                                    packetId = packetId,
+                                    index = index,
+                                    count = planned.value,
+                                    totalLength = payload.size,
+                                    payload = payload.copyOfRange(from, to),
+                                ).encode(),
+                        )
+                    }
+                }
+        }
+    }
+
+    /** Waits up to [timeoutMillis] for one packet on [flowId]. */
+    fun receive(
+        flowId: UInt,
+        timeoutMillis: Long,
+    ): ByteArray? {
+        val queue = inbound[flowId] ?: return null
+        val deadline = System.nanoTime() + timeoutMillis * NANOS_PER_MILLI
+        while (true) {
+            queue.poll()?.let { return if (it === CLOSED) null else it }
+            if (System.nanoTime() >= deadline) return null
+            pump(PUMP_MILLIS)
+        }
+    }
+
+    fun close(flowId: UInt) {
+        runCatching { datagrams.send(QuicDatagram.Close(flowId).encode()) }
+        forget(flowId)
+    }
+
+    /**
+     * Takes one datagram and gives it to whoever it is for.
+     *
+     * Under a lock across a single receive-and-route, so a flow waiting for its
+     * own packet does not hold up one whose packet has already arrived.
+     */
+    private fun pump(timeoutMillis: Long) {
+        synchronized(demultiplex) {
+            val raw = datagrams.receive(timeoutMillis) ?: return
+            val frame =
+                when (val decoded = DatagramFrame.decode(raw)) {
+                    is DecodeResult.Ok -> decoded.value
+                    // One unreliable packet. Tearing down every flow on the
+                    // connection because of it would turn a loss into an outage.
+                    is DecodeResult.Invalid -> return
+                }
+            when (val accepted = reassembler.offer(frame, System.currentTimeMillis())) {
+                is DecodeResult.Ok ->
+                    when (val value = accepted.value) {
+                        is DatagramReassembler.Accepted.Payload -> inbound[value.flowId]?.offer(value.bytes)
+                        is DatagramReassembler.Accepted.Closed -> inbound[value.flowId]?.offer(CLOSED)
+                        DatagramReassembler.Accepted.Pending -> Unit
+                    }
+                is DecodeResult.Invalid -> Unit
+            }
+        }
+    }
+
+    private companion object {
+        /** "the far end closed this flow", not a payload. */
+        val CLOSED = ByteArray(0)
+        const val PUMP_MILLIS = 25L
+        const val NANOS_PER_MILLI = 1_000_000L
+    }
+}

app/src/main/kotlin/eu/nodepass/somewhere/protocol/session/NowhereSession.kt

@@ -105,7 +105,7 @@ class NowhereSession(
      */
     private val split: SplitCarrier? =
         if (splitUplink != null && splitDownlink != null) {
-            SplitCarrier(splitUplink, splitDownlink, sharedKey, id)
+            SplitCarrier(splitUplink, splitDownlink, sharedKey, id, quicDatagrams)
         } else {
             null
         }