Add RUDP fragmentation across runtimes

docs/cultnet-transport-parity.md

@@ -180,16 +180,18 @@ Current progress:
   dropped data packet in both directions between TypeScript and C#/Kotlin/Python/Rust.
   The drops happen below the transport connection through a UDP bridge, so the
   recovery is RUDP resend/ack behavior, not application-level retry.
-- TypeScript and Python now have the first RUDP fragmentation implementation:
-  the session can split oversized reliable ordered payloads into fragment
-  packets, reassemble them before delivery, and the socket transport can carry
-  fragmented `schema` frames when `maxFragmentBytes` is configured. C#, Rust,
-  and Kotlin still need the same implementation before fragmentation can claim
-  cross-runtime parity.
+- TypeScript, C#, Rust, Python, and Kotlin now share the first RUDP
+  fragmentation implementation: the session can split oversized reliable
+  ordered payloads into fragment packets, reassemble them before delivery, and
+  the socket transport can carry fragmented `schema` frames when
+  `maxFragmentBytes` is configured.
+- The TypeScript interop harness now proves fragmented reliable ordered
+  `schema` frames in both request and response direction between TypeScript and
+  C#/Kotlin/Python/Rust one-shot UDP peers.
 - The remaining parity work is to deepen Python's server-side use of its port
   and move each runtime's existing TCP/LiteNetLib/WebSocket bodies behind those
   ports, then broaden RUDP interop into the full loss, reordering,
-  fragmentation, reconnect, and schema-message matrix.
+  reconnect, backpressure, and schema-message matrix.
 
 ## RUDP Packet Contract V0
 

packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt

@@ -171,18 +171,28 @@ data class CultNetRudpSendOptions(
 )
 
 private data class PendingReliablePacket(var packet: CultNetRudpPacket, var lastSentAtMs: Long)
+private data class PendingOrderedFrame(val frame: CultNetRudpDeliveredFrame, val nextSequence: Long)
+private data class FragmentBuffer(
+    val channelId: String,
+    val ordered: Boolean,
+    val fragmentCount: Int,
+    val payloads: MutableMap<Int, ByteArray> = linkedMapOf(),
+    val sequences: MutableMap<Int, Long> = linkedMapOf(),
+)
 
 class CultNetRudpSession(options: CultNetRudpSessionOptions) {
     val connectionId: Long = uint32(options.connectionId, "connectionId")
     val resendDelayMs: Long = options.resendDelayMs
     private var nextSequence = uint32(options.initialSequence, "initialSequence")
+    private var nextFragmentId = 1
     var connected: Boolean = false
         private set
     private var highestReceivedSequence: Long? = null
     private val receivedSequences = linkedSetOf<Long>()
     private val pendingReliable = linkedMapOf<Long, PendingReliablePacket>()
     private val orderedNextSequenceByChannel = linkedMapOf<String, Long>()
-    private val orderedBuffers = linkedMapOf<String, TreeMap<Long, CultNetRudpDeliveredFrame>>()
+    private val orderedBuffers = linkedMapOf<String, TreeMap<Long, PendingOrderedFrame>>()
+    private val fragmentBuffers = linkedMapOf<Pair<String, Int>, FragmentBuffer>()
 
     val pendingReliableSequences: List<Long>
         get() = pendingReliable.keys.sorted()
@@ -204,10 +214,41 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) {
     }
 
     fun send(channelId: String, payload: ByteArray, options: CultNetRudpSendOptions = CultNetRudpSendOptions()): CultNetRudpPacket {
+        return sendMany(channelId, payload, options).first()
+    }
+
+    fun sendMany(
+        channelId: String,
+        payload: ByteArray,
+        options: CultNetRudpSendOptions = CultNetRudpSendOptions(),
+        maxFragmentBytes: Int? = null,
+    ): List<CultNetRudpPacket> {
         if (!connected) throw IOException("Cannot send RUDP data before the session is connected")
+        if (maxFragmentBytes != null && maxFragmentBytes <= 0) throw IOException("RUDP maxFragmentBytes must be greater than zero")
+        if (maxFragmentBytes != null && payload.size > maxFragmentBytes) {
+            val fragmentCount = (payload.size + maxFragmentBytes - 1) / maxFragmentBytes
+            if (fragmentCount > 0xffff) throw IOException("RUDP payload requires more than 65535 fragments")
+            val fragmentId = allocateFragmentId()
+            return (0 until fragmentCount).map { index ->
+                val start = index * maxFragmentBytes
+                val packet = createPacket(
+                    channelId = channelId,
+                    packetType = CultNetRudpPacketType.Data,
+                    payload = payload.copyOfRange(start, minOf(start + maxFragmentBytes, payload.size)),
+                    reliable = options.reliable,
+                    ordered = options.ordered,
+                    sequenced = options.sequenced,
+                    fragmentId = fragmentId,
+                    fragmentIndex = index,
+                    fragmentCount = fragmentCount,
+                )
+                if (packet.reliable) trackReliable(packet, options.nowMs)
+                packet
+            }
+        }
         val packet = createPacket(channelId = channelId, packetType = CultNetRudpPacketType.Data, payload = payload, reliable = options.reliable, ordered = options.ordered, sequenced = options.sequenced)
         if (packet.reliable) trackReliable(packet, options.nowMs)
-        return packet
+        return listOf(packet)
     }
 
     fun receive(packet: CultNetRudpPacket, nowMs: Long = 0): CultNetRudpReceiveResult {
@@ -236,8 +277,8 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) {
         val duplicate = receivedSequences.contains(packet.sequence)
         rememberReceived(packet.sequence)
         if (duplicate) return CultNetRudpReceiveResult()
-        val frame = CultNetRudpDeliveredFrame(packet.channelId, packet.payload.copyOf(), packet.sequence)
-        return CultNetRudpReceiveResult(delivered = if (packet.ordered) deliverOrdered(frame) else listOf(frame))
+        val reassembled = reassemble(packet) ?: return CultNetRudpReceiveResult()
+        return CultNetRudpReceiveResult(delivered = if (reassembled.ordered) deliverOrdered(reassembled.frame, reassembled.nextSequence) else listOf(reassembled.frame))
     }
 
     fun createAck(): CultNetRudpPacket = createPacket(CultNetRudpPacketType.Ack, "control", ByteArray(0))
@@ -256,11 +297,14 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) {
         reliable: Boolean = false,
         ordered: Boolean = false,
         sequenced: Boolean = false,
+        fragmentId: Int = 0,
+        fragmentIndex: Int = 0,
+        fragmentCount: Int = 0,
     ): CultNetRudpPacket {
         val sequence = nextSequence
         nextSequence = uint32(nextSequence + 1, "sequence")
         val (ack, ackMask) = ackState()
-        return CultNetRudpPacket(packetType, connectionId, sequence, ack, ackMask, channelId, reliable, ordered, sequenced, payload = payload.copyOf())
+        return CultNetRudpPacket(packetType, connectionId, sequence, ack, ackMask, channelId, reliable, ordered, sequenced, fragmentId, fragmentIndex, fragmentCount, payload.copyOf())
     }
 
     private fun trackReliable(packet: CultNetRudpPacket, nowMs: Long) {
@@ -290,18 +334,48 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) {
         return ack to ackMask
     }
 
-    private fun deliverOrdered(frame: CultNetRudpDeliveredFrame): List<CultNetRudpDeliveredFrame> {
+    private data class ReassembledFrame(val frame: CultNetRudpDeliveredFrame, val ordered: Boolean, val nextSequence: Long)
+
+    private fun reassemble(packet: CultNetRudpPacket): ReassembledFrame? {
+        if (packet.fragmentCount == 0) {
+            return ReassembledFrame(CultNetRudpDeliveredFrame(packet.channelId, packet.payload.copyOf(), packet.sequence), packet.ordered, packet.sequence + 1)
+        }
+        if (packet.fragmentId == 0) throw IOException("RUDP fragmented packet must have a non-zero fragment id")
+        if (packet.fragmentIndex >= packet.fragmentCount) throw IOException("RUDP fragment index must be lower than fragment count")
+        val key = packet.channelId to packet.fragmentId
+        val buffer = fragmentBuffers.getOrPut(key) {
+            FragmentBuffer(packet.channelId, packet.ordered, packet.fragmentCount)
+        }
+        if (buffer.fragmentCount != packet.fragmentCount || buffer.ordered != packet.ordered) {
+            throw IOException("RUDP fragment metadata changed within a fragment set")
+        }
+        buffer.payloads[packet.fragmentIndex] = packet.payload.copyOf()
+        buffer.sequences[packet.fragmentIndex] = packet.sequence
+        if (buffer.payloads.size < packet.fragmentCount) return null
+        val payload = ByteArray(buffer.payloads.values.sumOf { it.size })
+        var offset = 0
+        for (index in 0 until packet.fragmentCount) {
+            val chunk = buffer.payloads[index] ?: return null
+            chunk.copyInto(payload, offset)
+            offset += chunk.size
+        }
+        val sequences = buffer.sequences.values
+        fragmentBuffers.remove(key)
+        return ReassembledFrame(CultNetRudpDeliveredFrame(buffer.channelId, payload, sequences.minOrNull() ?: packet.sequence), buffer.ordered, (sequences.maxOrNull() ?: packet.sequence) + 1)
+    }
+
+    private fun deliverOrdered(frame: CultNetRudpDeliveredFrame, nextSequenceAfterFrame: Long): List<CultNetRudpDeliveredFrame> {
         val next = orderedNextSequenceByChannel[frame.channelId]
         if (next == null) {
-            orderedNextSequenceByChannel[frame.channelId] = frame.sequence + 1
+            orderedNextSequenceByChannel[frame.channelId] = nextSequenceAfterFrame
             return listOf(frame) + drainOrdered(frame.channelId)
         }
         if (frame.sequence < next) return emptyList()
         if (frame.sequence > next) {
-            orderedBuffers.getOrPut(frame.channelId) { TreeMap() }[frame.sequence] = frame
+            orderedBuffers.getOrPut(frame.channelId) { TreeMap() }[frame.sequence] = PendingOrderedFrame(frame, nextSequenceAfterFrame)
             return emptyList()
         }
-        orderedNextSequenceByChannel[frame.channelId] = next + 1
+        orderedNextSequenceByChannel[frame.channelId] = nextSequenceAfterFrame
         return listOf(frame) + drainOrdered(frame.channelId)
     }
 
@@ -310,13 +384,20 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) {
         val buffer = orderedBuffers[channelId] ?: return delivered
         while (true) {
             val next = orderedNextSequenceByChannel[channelId] ?: break
-            val frame = buffer.remove(next) ?: break
-            delivered.add(frame)
-            orderedNextSequenceByChannel[channelId] = next + 1
+            val pending = buffer.remove(next) ?: break
+            delivered.add(pending.frame)
+            orderedNextSequenceByChannel[channelId] = pending.nextSequence
         }
         return delivered
     }
 
+    private fun allocateFragmentId(): Int {
+        val fragmentId = nextFragmentId
+        nextFragmentId += 1
+        if (nextFragmentId > 0xffff) nextFragmentId = 1
+        return fragmentId
+    }
+
     private fun requireConnection(packet: CultNetRudpPacket) {
         if (packet.connectionId != connectionId) throw IOException("RUDP packet connection id ${packet.connectionId} does not match $connectionId")
     }
@@ -332,6 +413,7 @@ class CultNetRudpSocketTransportConnection(
     remoteAddress: InetSocketAddress? = null,
     initialSequence: Long = 1,
     resendDelayMs: Long = 250,
+    private val maxFragmentBytes: Int? = null,
 ) : AutoCloseable {
     private val session = CultNetRudpSession(CultNetRudpSessionOptions(connectionId, initialSequence, resendDelayMs))
     private var remote = remoteAddress
@@ -345,6 +427,7 @@ class CultNetRudpSocketTransportConnection(
         runtimeId = runtimeId,
         host = socket.localAddress.hostAddress,
         port = socket.localPort,
+        maxFragmentBytes = maxFragmentBytes,
     )
     val connected: Boolean get() = session.connected
     val stats: CultNetTransportStats get() = CultNetTransportStats(bytesReceived, bytesSent, framesReceived, framesSent)
@@ -355,7 +438,7 @@ class CultNetRudpSocketTransportConnection(
     }
 
     fun send(channelId: String, payload: ByteArray) {
-        sendPacket(session.send(channelId, payload, channelSendOptions(channelId, nowMs())))
+        session.sendMany(channelId, payload, channelSendOptions(channelId, nowMs()), maxFragmentBytes).forEach { sendPacket(it) }
         framesSent += 1
     }
 
@@ -494,7 +577,9 @@ private fun nowMs(): Long = Instant.now().toEpochMilli()
 fun main(args: Array<String>) {
     if (args.isEmpty()) {
         rudpPacketCodecUsesDeterministicReliableOrderedFixture()
+        rudpSessionFragmentsAndReassemblesReliableOrderedPayloads()
         rudpSocketTransportHandshakesAndCarriesReliableOrderedSchemaFrames()
+        rudpSocketTransportCarriesFragmentedReliableOrderedSchemaFrames()
         return
     }
 
@@ -509,6 +594,9 @@ fun main(args: Array<String>) {
 private fun rudpServeOnce(options: Map<String, String>) {
     val bindHost = options["bind-host"] ?: "127.0.0.1"
     val bindPort = options["bind-port"]?.toInt() ?: 0
+    val expectedClientPayload = (options["client-payload"] ?: "ts-kotlin-client-state").toByteArray(StandardCharsets.UTF_8)
+    val serverPayload = (options["server-payload"] ?: "kotlin-server-state").toByteArray(StandardCharsets.UTF_8)
+    val maxFragmentBytes = options["max-fragment-bytes"]?.toInt()
     val socket = DatagramSocket(bindPort, InetAddress.getByName(bindHost)).also { it.soTimeout = 20 }
     CultNetRudpSocketTransportConnection(
         socket = socket,
@@ -517,14 +605,15 @@ private fun rudpServeOnce(options: Map<String, String>) {
         connectionId = 0x446688aaL,
         initialSequence = 100,
         resendDelayMs = 25,
+        maxFragmentBytes = maxFragmentBytes,
     ).use { transport ->
         println("""{"status":"ready","port":${socket.localPort}}""")
         val deadline = System.nanoTime() + 5_000_000_000L
         while (System.nanoTime() < deadline) {
             val frame = transport.receiveOnce()
             if (frame != null) {
-                requireRudpFrame(frame, "schema", "ts-kotlin-client-state")
-                transport.send("schema", "kotlin-server-state".toByteArray(StandardCharsets.UTF_8))
+                requireRudpFrame(frame, "schema", expectedClientPayload)
+                transport.send("schema", serverPayload)
                 pollRudpAfterSend(transport, 250)
                 println("""{"status":"ok"}""")
                 return
@@ -581,8 +670,11 @@ private fun pollRudpAfterSend(transport: CultNetRudpSocketTransportConnection, d
 }
 
 private fun requireRudpFrame(frame: CultNetTransportFrame, expectedChannelId: String, expectedPayload: String) {
-    val expectedBytes = expectedPayload.toByteArray(StandardCharsets.UTF_8)
-    if (frame.channelId != expectedChannelId || !frame.payload.contentEquals(expectedBytes)) {
+    requireRudpFrame(frame, expectedChannelId, expectedPayload.toByteArray(StandardCharsets.UTF_8))
+}
+
+private fun requireRudpFrame(frame: CultNetTransportFrame, expectedChannelId: String, expectedPayload: ByteArray) {
+    if (frame.channelId != expectedChannelId || !frame.payload.contentEquals(expectedPayload)) {
         error("Unexpected RUDP frame: channel=${frame.channelId}, payload=${String(frame.payload, StandardCharsets.UTF_8)}")
     }
 }
@@ -637,6 +729,31 @@ private fun rudpPacketCodecUsesDeterministicReliableOrderedFixture() {
     check(String(decoded.payload, StandardCharsets.UTF_8) == "hello")
 }
 
+private fun rudpSessionFragmentsAndReassemblesReliableOrderedPayloads() {
+    val sender = CultNetRudpSession(CultNetRudpSessionOptions(connectionId = 456, initialSequence = 1))
+    val receiver = CultNetRudpSession(CultNetRudpSessionOptions(connectionId = 456, initialSequence = 100))
+    sender.receive(CultNetRudpPacket(CultNetRudpPacketType.Accept, 456, 90, 0, 0, "control"))
+    receiver.receive(CultNetRudpPacket(CultNetRudpPacketType.Accept, 456, 91, 0, 0, "control"))
+
+    val packets = sender.sendMany(
+        "schema",
+        "fragment-me-please".toByteArray(StandardCharsets.UTF_8),
+        CultNetRudpSendOptions(reliable = true, ordered = true, nowMs = 10),
+        maxFragmentBytes = 5,
+    )
+    check(packets.size == 4)
+    check(packets.map { it.fragmentCount } == listOf(4, 4, 4, 4))
+    check(packets.map { it.fragmentIndex } == listOf(0, 1, 2, 3))
+    check(packets.all { it.fragmentId == packets.first().fragmentId })
+    check(receiver.receive(packets[0]).delivered.isEmpty())
+    check(receiver.receive(packets[1]).delivered.isEmpty())
+    check(receiver.receive(packets[2]).delivered.isEmpty())
+    val delivered = receiver.receive(packets[3]).delivered
+    check(delivered.size == 1)
+    check(String(delivered.first().payload, StandardCharsets.UTF_8) == "fragment-me-please")
+    check(delivered.first().sequence == packets.first().sequence)
+}
+
 private fun rudpSocketTransportHandshakesAndCarriesReliableOrderedSchemaFrames() {
     val loopback = InetAddress.getByName("127.0.0.1")
     val serverSocket = DatagramSocket(0, loopback).also { it.soTimeout = 20 }
@@ -680,6 +797,43 @@ private fun rudpSocketTransportHandshakesAndCarriesReliableOrderedSchemaFrames()
     }
 }
 
+private fun rudpSocketTransportCarriesFragmentedReliableOrderedSchemaFrames() {
+    val loopback = InetAddress.getByName("127.0.0.1")
+    val serverSocket = DatagramSocket(0, loopback).also { it.soTimeout = 20 }
+    val clientSocket = DatagramSocket(0, loopback).also { it.soTimeout = 20 }
+    val connectionId = 0x10203041L
+    CultNetRudpSocketTransportConnection(
+        socket = serverSocket,
+        mode = CultNetRudpSocketMode.Server,
+        runtimeId = "kotlin-rudp-fragment-server",
+        connectionId = connectionId,
+        initialSequence = 100,
+        resendDelayMs = 25,
+        maxFragmentBytes = 8,
+    ).use { server ->
+        CultNetRudpSocketTransportConnection(
+            socket = clientSocket,
+            mode = CultNetRudpSocketMode.Client,
+            runtimeId = "kotlin-rudp-fragment-client",
+            connectionId = connectionId,
+            remoteAddress = InetSocketAddress(loopback, serverSocket.localPort),
+            initialSequence = 1,
+            resendDelayMs = 25,
+            maxFragmentBytes = 8,
+        ).use { client ->
+            val payload = "this-schema-frame-is-larger-than-one-rudp-fragment".toByteArray(StandardCharsets.UTF_8)
+            client.connect("join".toByteArray(StandardCharsets.UTF_8))
+            pumpRudpHandshake(client, server)
+            client.send("schema", payload)
+            val serverFrame = receiveRudpFrame(server)
+            check(serverFrame.channelId == "schema")
+            check(serverFrame.payload.contentEquals(payload))
+            check(client.stats.framesSent == 1L)
+            check(server.stats.framesReceived == 1L)
+        }
+    }
+}
+
 private fun pumpRudpHandshake(
     client: CultNetRudpSocketTransportConnection,
     server: CultNetRudpSocketTransportConnection,

packages/cultnet-rs/examples/cultnet_interop_peer.rs

@@ -458,6 +458,19 @@ fn rudp_serve_once(options: &BTreeMap<String, String>) -> Result<()> {
         .transpose()
         .with_context(|| "argument --bind-port must be a u16")?
         .unwrap_or(0);
+    let expected_client_payload = options
+        .get("client-payload")
+        .map(|value| value.as_bytes().to_vec())
+        .unwrap_or_else(|| b"ts-rust-client-state".to_vec());
+    let server_payload = options
+        .get("server-payload")
+        .map(|value| value.as_bytes().to_vec())
+        .unwrap_or_else(|| b"rust-server-state".to_vec());
+    let max_fragment_bytes = options
+        .get("max-fragment-bytes")
+        .map(|value| value.parse::<u32>())
+        .transpose()
+        .with_context(|| "argument --max-fragment-bytes must be a u32")?;
     let socket = UdpSocket::bind((bind_host.as_str(), bind_port))
         .with_context(|| format!("failed to bind RUDP server on {bind_host}:{bind_port}"))?;
     socket.set_read_timeout(Some(Duration::from_millis(20)))?;
@@ -466,6 +479,7 @@ fn rudp_serve_once(options: &BTreeMap<String, String>) -> Result<()> {
         CultNetRudpSocketTransportOptions::server("rust-rudp-interop", socket, 0x22446688);
     transport_options.initial_sequence = 100;
     transport_options.resend_delay_ms = 25;
+    transport_options.max_fragment_bytes = max_fragment_bytes;
     let mut transport = CultNetRudpSocketTransportConnection::new(transport_options)?;
 
     print_json(&serde_json::json!({
@@ -476,14 +490,14 @@ fn rudp_serve_once(options: &BTreeMap<String, String>) -> Result<()> {
     let deadline = Instant::now() + Duration::from_secs(5);
     while Instant::now() < deadline {
         if let Some(frame) = transport.receive_once()? {
-            if frame.channel_id != "schema" || frame.payload != b"ts-rust-client-state" {
+            if frame.channel_id != "schema" || frame.payload != expected_client_payload {
                 return Err(anyhow!(
                     "unexpected RUDP frame: channel={} payload={:?}",
                     frame.channel_id,
                     frame.payload
                 ));
             }
-            transport.send("schema", b"rust-server-state".to_vec())?;
+            transport.send("schema", server_payload.clone())?;
             poll_rudp_after_send(&mut transport, Duration::from_millis(250))?;
             print_json(&serde_json::json!({ "status": "ok" }))?;
             return Ok(());

packages/cultnet-rs/src/rudp.rs

@@ -94,16 +94,33 @@ struct PendingReliablePacket {
     last_sent_at_ms: u64,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+struct PendingOrderedFrame {
+    frame: CultNetRudpDeliveredFrame,
+    next_sequence: u32,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+struct FragmentBuffer {
+    channel_id: String,
+    ordered: bool,
+    fragment_count: u16,
+    payloads: BTreeMap<u16, Vec<u8>>,
+    sequences: BTreeMap<u16, u32>,
+}
+
 pub struct CultNetRudpSession {
     connection_id: u32,
     resend_delay_ms: u64,
     next_sequence: u32,
+    next_fragment_id: u16,
     connected: bool,
     highest_received_sequence: Option<u32>,
     received_sequences: BTreeSet<u32>,
     pending_reliable: BTreeMap<u32, PendingReliablePacket>,
     ordered_next_sequence_by_channel: BTreeMap<String, u32>,
-    ordered_buffers: BTreeMap<String, BTreeMap<u32, CultNetRudpDeliveredFrame>>,
+    ordered_buffers: BTreeMap<String, BTreeMap<u32, PendingOrderedFrame>>,
+    fragment_buffers: BTreeMap<(String, u16), FragmentBuffer>,
 }
 
 impl CultNetRudpSession {
@@ -112,12 +129,14 @@ impl CultNetRudpSession {
             connection_id: options.connection_id,
             resend_delay_ms: options.resend_delay_ms,
             next_sequence: options.initial_sequence,
+            next_fragment_id: 1,
             connected: false,
             highest_received_sequence: None,
             received_sequences: BTreeSet::new(),
             pending_reliable: BTreeMap::new(),
             ordered_next_sequence_by_channel: BTreeMap::new(),
             ordered_buffers: BTreeMap::new(),
+            fragment_buffers: BTreeMap::new(),
         }
     }
 
@@ -184,12 +203,59 @@ impl CultNetRudpSession {
         payload: Vec<u8>,
         options: CultNetRudpSendOptions,
     ) -> Result<CultNetRudpPacket> {
+        self.send_many(channel_id, payload, options, None)?
+            .into_iter()
+            .next()
+            .ok_or_else(|| anyhow!("RUDP send produced no packets"))
+    }
+
+    pub fn send_many(
+        &mut self,
+        channel_id: &str,
+        payload: Vec<u8>,
+        options: CultNetRudpSendOptions,
+        max_fragment_bytes: Option<usize>,
+    ) -> Result<Vec<CultNetRudpPacket>> {
         if !self.connected {
             return Err(anyhow!(
                 "Cannot send RUDP data before the session is connected"
             ));
         }
 
+        if let Some(max_fragment_bytes) = max_fragment_bytes {
+            if max_fragment_bytes == 0 {
+                return Err(anyhow!("RUDP max_fragment_bytes must be greater than zero"));
+            }
+            if payload.len() > max_fragment_bytes {
+                let fragment_count = payload.len().div_ceil(max_fragment_bytes);
+                if fragment_count > u16::MAX as usize {
+                    return Err(anyhow!("RUDP payload requires more than 65535 fragments"));
+                }
+                let fragment_id = self.allocate_fragment_id();
+                let mut packets = Vec::new();
+                for index in 0..fragment_count {
+                    let start = index * max_fragment_bytes;
+                    let end = (start + max_fragment_bytes).min(payload.len());
+                    let packet = self.create_packet_with_fragments(
+                        CultNetRudpPacketType::Data,
+                        channel_id,
+                        payload[start..end].to_vec(),
+                        options.reliable,
+                        options.ordered,
+                        options.sequenced,
+                        fragment_id,
+                        index as u16,
+                        fragment_count as u16,
+                    );
+                    if packet.reliable {
+                        self.track_reliable(packet.clone(), options.now_ms);
+                    }
+                    packets.push(packet);
+                }
+                return Ok(packets);
+            }
+        }
+
         let packet = self.create_packet(
             CultNetRudpPacketType::Data,
             channel_id,
@@ -201,7 +267,7 @@ impl CultNetRudpSession {
         if packet.reliable {
             self.track_reliable(packet.clone(), options.now_ms);
         }
-        Ok(packet)
+        Ok(vec![packet])
     }
 
     pub fn receive(
@@ -262,13 +328,14 @@ impl CultNetRudpSession {
             });
         }
 
-        let frame = CultNetRudpDeliveredFrame {
-            channel_id: packet.channel_id.clone(),
-            payload: packet.payload.clone(),
-            sequence: packet.sequence,
+        let Some((frame, ordered, next_sequence)) = self.reassemble(packet)? else {
+            return Ok(CultNetRudpReceiveResult {
+                delivered: Vec::new(),
+                reply: None,
+            });
         };
-        let delivered = if packet.ordered {
-            self.deliver_ordered(frame)
+        let delivered = if ordered {
+            self.deliver_ordered(frame, next_sequence)
         } else {
             vec![frame]
         };
@@ -309,6 +376,32 @@ impl CultNetRudpSession {
         reliable: bool,
         ordered: bool,
         sequenced: bool,
+    ) -> CultNetRudpPacket {
+        self.create_packet_with_fragments(
+            packet_type,
+            channel_id,
+            payload,
+            reliable,
+            ordered,
+            sequenced,
+            0,
+            0,
+            0,
+        )
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    fn create_packet_with_fragments(
+        &mut self,
+        packet_type: CultNetRudpPacketType,
+        channel_id: &str,
+        payload: Vec<u8>,
+        reliable: bool,
+        ordered: bool,
+        sequenced: bool,
+        fragment_id: u16,
+        fragment_index: u16,
+        fragment_count: u16,
     ) -> CultNetRudpPacket {
         let sequence = self.next_sequence;
         self.next_sequence = self
@@ -326,9 +419,9 @@ impl CultNetRudpSession {
             reliable,
             ordered,
             sequenced,
-            fragment_id: 0,
-            fragment_index: 0,
-            fragment_count: 0,
+            fragment_id,
+            fragment_index,
+            fragment_count,
             payload,
         }
     }
@@ -373,9 +466,88 @@ impl CultNetRudpSession {
         (ack, ack_mask)
     }
 
+    fn reassemble(
+        &mut self,
+        packet: &CultNetRudpPacket,
+    ) -> Result<Option<(CultNetRudpDeliveredFrame, bool, u32)>> {
+        if packet.fragment_count == 0 {
+            return Ok(Some((
+                CultNetRudpDeliveredFrame {
+                    channel_id: packet.channel_id.clone(),
+                    payload: packet.payload.clone(),
+                    sequence: packet.sequence,
+                },
+                packet.ordered,
+                packet.sequence + 1,
+            )));
+        }
+        if packet.fragment_id == 0 {
+            return Err(anyhow!(
+                "RUDP fragmented packet must have a non-zero fragment id"
+            ));
+        }
+        if packet.fragment_index >= packet.fragment_count {
+            return Err(anyhow!(
+                "RUDP fragment index must be lower than fragment count"
+            ));
+        }
+
+        let key = (packet.channel_id.clone(), packet.fragment_id);
+        let buffer = self
+            .fragment_buffers
+            .entry(key.clone())
+            .or_insert_with(|| FragmentBuffer {
+                channel_id: packet.channel_id.clone(),
+                ordered: packet.ordered,
+                fragment_count: packet.fragment_count,
+                payloads: BTreeMap::new(),
+                sequences: BTreeMap::new(),
+            });
+        if buffer.fragment_count != packet.fragment_count || buffer.ordered != packet.ordered {
+            return Err(anyhow!(
+                "RUDP fragment metadata changed within a fragment set"
+            ));
+        }
+        buffer
+            .payloads
+            .insert(packet.fragment_index, packet.payload.clone());
+        buffer
+            .sequences
+            .insert(packet.fragment_index, packet.sequence);
+        if buffer.payloads.len() < packet.fragment_count as usize {
+            return Ok(None);
+        }
+
+        let mut payload = Vec::new();
+        let mut sequences = Vec::new();
+        for index in 0..packet.fragment_count {
+            let Some(chunk) = buffer.payloads.get(&index) else {
+                return Ok(None);
+            };
+            let Some(sequence) = buffer.sequences.get(&index) else {
+                return Ok(None);
+            };
+            payload.extend_from_slice(chunk);
+            sequences.push(*sequence);
+        }
+        let channel_id = buffer.channel_id.clone();
+        let ordered = buffer.ordered;
+        self.fragment_buffers.remove(&key);
+        Ok(Some((
+            CultNetRudpDeliveredFrame {
+                channel_id,
+                payload,
+                sequence: *sequences.iter().min().unwrap(),
+            },
+            ordered,
+            sequences.iter().max().unwrap() + 1,
+        )))
+    }
+
     fn deliver_ordered(
         &mut self,
         frame: CultNetRudpDeliveredFrame,
+        next_sequence_after_frame: u32,
     ) -> Vec<CultNetRudpDeliveredFrame> {
         let channel_id = frame.channel_id.clone();
         let Some(next) = self
@@ -384,7 +556,7 @@ impl CultNetRudpSession {
             .copied()
         else {
             self.ordered_next_sequence_by_channel
-                .insert(channel_id.clone(), frame.sequence + 1);
+                .insert(channel_id.clone(), next_sequence_after_frame);
             let mut delivered = vec![frame];
             delivered.extend(self.drain_ordered(&channel_id));
             return delivered;
@@ -395,15 +567,18 @@ impl CultNetRudpSession {
         }
 
         if frame.sequence > next {
-            self.ordered_buffers
-                .entry(channel_id)
-                .or_default()
-                .insert(frame.sequence, frame);
+            self.ordered_buffers.entry(channel_id).or_default().insert(
+                frame.sequence,
+                PendingOrderedFrame {
+                    frame,
+                    next_sequence: next_sequence_after_frame,
+                },
+            );
             return Vec::new();
         }
 
         self.ordered_next_sequence_by_channel
-            .insert(channel_id.clone(), next + 1);
+            .insert(channel_id.clone(), next_sequence_after_frame);
         let mut delivered = vec![frame];
         delivered.extend(self.drain_ordered(&channel_id));
         delivered
@@ -422,16 +597,25 @@ impl CultNetRudpSession {
             let Some(buffer) = self.ordered_buffers.get_mut(channel_id) else {
                 break;
             };
-            let Some(frame) = buffer.remove(&next) else {
+            let Some(pending) = buffer.remove(&next) else {
                 break;
             };
-            delivered.push(frame);
+            delivered.push(pending.frame);
             self.ordered_next_sequence_by_channel
-                .insert(channel_id.to_string(), next + 1);
+                .insert(channel_id.to_string(), pending.next_sequence);
         }
         delivered
     }
 
+    fn allocate_fragment_id(&mut self) -> u16 {
+        let fragment_id = self.next_fragment_id;
+        self.next_fragment_id = self.next_fragment_id.saturating_add(1);
+        if self.next_fragment_id == 0 {
+            self.next_fragment_id = 1;
+        }
+        fragment_id
+    }
+
     fn require_connection(&self, packet: &CultNetRudpPacket) -> Result<()> {
         if packet.connection_id != self.connection_id {
             return Err(anyhow!(
@@ -508,6 +692,7 @@ pub struct CultNetRudpSocketTransportConnection {
     pub profile: CultNetTransportProfile,
     stats: CultNetTransportStats,
     delivered_frames: VecDeque<CultNetTransportFrame>,
+    max_fragment_bytes: Option<usize>,
 }
 
 impl CultNetRudpSocketTransportConnection {
@@ -535,6 +720,7 @@ impl CultNetRudpSocketTransportConnection {
             profile,
             stats: CultNetTransportStats::default(),
             delivered_frames: VecDeque::new(),
+            max_fragment_bytes: options.max_fragment_bytes.map(|value| value as usize),
         })
     }
 
@@ -558,8 +744,12 @@ impl CultNetRudpSocketTransportConnection {
 
     pub fn send(&mut self, channel_id: &str, payload: Vec<u8>) -> Result<()> {
         let options = channel_send_options(channel_id, now_ms());
-        let packet = self.session.send(channel_id, payload, options)?;
-        self.send_packet(&packet)?;
+        let packets =
+            self.session
+                .send_many(channel_id, payload, options, self.max_fragment_bytes)?;
+        for packet in packets {
+            self.send_packet(&packet)?;
+        }
         self.stats.frames_sent += 1;
         Ok(())
     }

packages/cultnet-rs/tests/cultnet.rs

@@ -571,6 +571,97 @@ fn rudp_session_suppresses_duplicates_and_delivers_reliable_ordered_payloads_in_
     Ok(())
 }
 
+#[test]
+fn rudp_session_fragments_and_reassembles_reliable_ordered_payloads() -> Result<()> {
+    let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions {
+        connection_id: 456,
+        initial_sequence: 1,
+        ..CultNetRudpSessionOptions::default()
+    });
+    let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions {
+        connection_id: 456,
+        initial_sequence: 100,
+        ..CultNetRudpSessionOptions::default()
+    });
+    sender.receive(
+        &CultNetRudpPacket {
+            packet_type: CultNetRudpPacketType::Accept,
+            connection_id: 456,
+            sequence: 90,
+            ack: 0,
+            ack_mask: 0,
+            channel_id: "control".to_string(),
+            reliable: false,
+            ordered: false,
+            sequenced: false,
+            fragment_id: 0,
+            fragment_index: 0,
+            fragment_count: 0,
+            payload: Vec::new(),
+        },
+        0,
+    )?;
+    receiver.receive(
+        &CultNetRudpPacket {
+            packet_type: CultNetRudpPacketType::Accept,
+            connection_id: 456,
+            sequence: 91,
+            ack: 0,
+            ack_mask: 0,
+            channel_id: "control".to_string(),
+            reliable: false,
+            ordered: false,
+            sequenced: false,
+            fragment_id: 0,
+            fragment_index: 0,
+            fragment_count: 0,
+            payload: Vec::new(),
+        },
+        0,
+    )?;
+
+    let packets = sender.send_many(
+        "schema",
+        b"fragment-me-please".to_vec(),
+        CultNetRudpSendOptions {
+            reliable: true,
+            ordered: true,
+            now_ms: 10,
+            ..CultNetRudpSendOptions::default()
+        },
+        Some(5),
+    )?;
+    assert_eq!(packets.len(), 4);
+    assert_eq!(
+        packets
+            .iter()
+            .map(|packet| packet.fragment_count)
+            .collect::<Vec<_>>(),
+        vec![4, 4, 4, 4]
+    );
+    assert_eq!(
+        packets
+            .iter()
+            .map(|packet| packet.fragment_index)
+            .collect::<Vec<_>>(),
+        vec![0, 1, 2, 3]
+    );
+    assert!(
+        packets
+            .iter()
+            .all(|packet| packet.fragment_id == packets[0].fragment_id)
+    );
+
+    assert!(receiver.receive(&packets[0], 0)?.delivered.is_empty());
+    assert!(receiver.receive(&packets[1], 0)?.delivered.is_empty());
+    assert!(receiver.receive(&packets[2], 0)?.delivered.is_empty());
+    let delivered = receiver.receive(&packets[3], 0)?.delivered;
+    assert_eq!(delivered.len(), 1);
+    assert_eq!(delivered[0].payload, b"fragment-me-please");
+    assert_eq!(delivered[0].sequence, packets[0].sequence);
+    Ok(())
+}
+
 #[test]
 fn rudp_socket_transport_handshakes_and_carries_reliable_ordered_schema_frames() -> Result<()> {
     let server_socket = bind_udp_socket()?;
@@ -628,6 +719,52 @@ fn rudp_socket_transport_handshakes_and_carries_reliable_ordered_schema_frames()
     Ok(())
 }
 
+#[test]
+fn rudp_socket_transport_carries_fragmented_reliable_ordered_schema_frames() -> Result<()> {
+    let server_socket = bind_udp_socket()?;
+    let client_socket = bind_udp_socket()?;
+    let server_addr = server_socket.local_addr()?;
+    let connection_id = 0x1020_3041;
+    let mut server =
+        CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions {
+            runtime_id: "rust-rudp-fragment-server".to_string(),
+            socket: server_socket,
+            mode: CultNetRudpSocketMode::Server,
+            remote_addr: None,
+            connection_id,
+            initial_sequence: 100,
+            resend_delay_ms: 25,
+            transport_id: None,
+            max_payload_bytes: None,
+            max_fragment_bytes: Some(8),
+        })?;
+    let mut client =
+        CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions {
+            runtime_id: "rust-rudp-fragment-client".to_string(),
+            socket: client_socket,
+            mode: CultNetRudpSocketMode::Client,
+            remote_addr: Some(server_addr),
+            connection_id,
+            initial_sequence: 1,
+            resend_delay_ms: 25,
+            transport_id: None,
+            max_payload_bytes: None,
+            max_fragment_bytes: Some(8),
+        })?;
+
+    client.connect(b"join".to_vec())?;
+    pump_rudp_handshake(&mut client, &mut server)?;
+
+    let payload = b"this-schema-frame-is-larger-than-one-rudp-fragment".to_vec();
+    client.send("schema", payload.clone())?;
+    let server_frame = receive_rudp_frame(&mut server)?;
+    assert_eq!(server_frame.channel_id, "schema");
+    assert_eq!(server_frame.payload, payload);
+    assert_eq!(client.stats().frames_sent, 1);
+    assert_eq!(server.stats().frames_received, 1);
+    Ok(())
+}
+
 #[test]
 fn legacy_gamecult_networking_contract_round_trips_login_union() -> Result<()> {
     let message = CultNetMessage::Login {

packages/cultnet-ts/test/interop/cultnet-interop.test.ts

@@ -72,12 +72,18 @@ interface RudpServerInteropPeerSpec {
   name: string;
   connectionId: number;
   build: () => Promise<void>;
-  spawn: () => RunningPythonRudpPeer;
+  spawn: (options?: RudpServerSpawnOptions) => RunningPythonRudpPeer;
   joinPayload: string;
   clientPayload: string;
   expectedPayload: string;
 }
 
+interface RudpServerSpawnOptions {
+  clientPayload?: string;
+  serverPayload?: string;
+  maxFragmentBytes?: number;
+}
+
 function rudpServerInteropPeers(): RudpServerInteropPeerSpec[] {
   return [
     {
@@ -120,6 +126,7 @@ function rudpServerInteropPeers(): RudpServerInteropPeerSpec[] {
 }
 
 const pythonRudpPeerScript = String.raw`
+import os
 import socket
 import time
 
@@ -129,6 +136,11 @@ from cultnet_py import (
     CultNetRudpSocketTransportOptions,
 )
 
+expected_client_payload = os.environ.get("RUDP_CLIENT_PAYLOAD", "client-state").encode()
+server_payload = os.environ.get("RUDP_SERVER_PAYLOAD", "python-state").encode()
+max_fragment_bytes_raw = os.environ.get("RUDP_MAX_FRAGMENT_BYTES")
+max_fragment_bytes = int(max_fragment_bytes_raw) if max_fragment_bytes_raw else None
+
 sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
 sock.bind(("127.0.0.1", 0))
 sock.settimeout(0.02)
@@ -139,6 +151,7 @@ transport = CultNetRudpSocketTransportConnection(CultNetRudpSocketTransportOptio
     connection_id=0x55667788,
     initial_sequence=100,
     resend_delay_ms=25,
+    max_fragment_bytes=max_fragment_bytes,
 ))
 print(f"PORT {sock.getsockname()[1]}", flush=True)
 
@@ -149,10 +162,10 @@ while time.time() < deadline:
     if frame is None:
         time.sleep(0.005)
         continue
-    if frame.channel_id != "schema" or frame.payload != b"client-state":
+    if frame.channel_id != "schema" or frame.payload != expected_client_payload:
         transport.close()
         raise SystemExit(f"unexpected RUDP frame: {frame.channel_id!r} {frame.payload!r}")
-    transport.send("schema", b"python-state")
+    transport.send("schema", server_payload)
     print("OK", flush=True)
     ack_deadline = time.time() + 0.25
     while time.time() < ack_deadline:
@@ -1153,6 +1166,58 @@ test("CultNet RUDP reliable schema response survives one dropped server data pac
   }
 });
 
+test("CultNet RUDP fragmented schema frames cross runtime boundaries", async () => {
+  for (const peer of rudpServerInteropPeers()) {
+    await peer.build();
+    const clientPayload = `${peer.clientPayload}:fragmented:${"x".repeat(96)}`;
+    const serverPayload = `${peer.expectedPayload}:fragmented:${"y".repeat(96)}`;
+    const serverPeer = peer.spawn({
+      clientPayload,
+      serverPayload,
+      maxFragmentBytes: 16,
+    });
+    const clientSocket = await bindUdpSocket();
+    let client: CultNetRudpSocketTransportConnection | undefined;
+
+    try {
+      client = new CultNetRudpSocketTransportConnection({
+        runtimeId: `ts-fragmented-${peer.name.replace(/\W/g, "").toLowerCase()}-rudp-interop`,
+        socket: clientSocket,
+        mode: "client",
+        remoteHost: "127.0.0.1",
+        remotePort: await serverPeer.ready,
+        connectionId: peer.connectionId,
+        initialSequence: 1,
+        resendDelayMs: 25,
+        resendPollMs: 5,
+        maxFragmentBytes: 16,
+      });
+
+      const receivedFrame = once(client, "frame");
+      client.connect(Buffer.from(peer.joinPayload));
+      await waitFor(() => client?.connected === true, `TypeScript RUDP client to complete ${peer.name} fragmented handshake`);
+      client.send("schema", Buffer.from(clientPayload));
+
+      const [frame] = await withTimeout(receivedFrame, 2_000, `${peer.name} fragmented RUDP response frame`);
+      assert.equal(frame.channelId, "schema");
+      assert.deepEqual(Buffer.from(frame.payload), Buffer.from(serverPayload));
+
+      const [exitCode] = await withTimeout(once(serverPeer.child, "exit"), 2_000, `${peer.name} RUDP peer exit after fragmented exchange`);
+      assert.equal(exitCode, 0, serverPeer.stderr.join(""));
+    } finally {
+      if (client) {
+        client.close();
+      } else {
+        clientSocket.close();
+      }
+      if (serverPeer.child.exitCode === null && !serverPeer.child.killed) {
+        serverPeer.child.kill("SIGTERM");
+        await once(serverPeer.child, "exit").catch(() => undefined);
+      }
+    }
+  }
+});
+
 async function buildInteropPeers(): Promise<void> {
   await buildRustInteropPeer();
   await buildCSharpInteropPeer();
@@ -1305,10 +1370,16 @@ interface RunningPythonRudpPeer {
   stderr: string[];
 }
 
-function spawnPythonRudpPeer(): RunningPythonRudpPeer {
+function spawnPythonRudpPeer(options: RudpServerSpawnOptions = {}): RunningPythonRudpPeer {
   const child = spawn(pythonCommand, ["-c", pythonRudpPeerScript], {
     cwd: cultcachePyRoot,
-    env: { ...process.env, PYTHONPATH: cultcachePySrc },
+    env: {
+      ...process.env,
+      PYTHONPATH: cultcachePySrc,
+      ...(options.clientPayload ? { RUDP_CLIENT_PAYLOAD: options.clientPayload } : {}),
+      ...(options.serverPayload ? { RUDP_SERVER_PAYLOAD: options.serverPayload } : {}),
+      ...(options.maxFragmentBytes ? { RUDP_MAX_FRAGMENT_BYTES: String(options.maxFragmentBytes) } : {}),
+    },
     stdio: ["ignore", "pipe", "pipe"],
   });
   const stderr: string[] = [];
@@ -1353,8 +1424,14 @@ function spawnPythonRudpPeer(): RunningPythonRudpPeer {
   return { child, ready, stderr };
 }
 
-function spawnRustRudpServer(): RunningPythonRudpPeer {
-  const child = spawn(rustBinaryPath, ["rudp-serve-once", "--bind-host", "127.0.0.1"], {
+function spawnRustRudpServer(options: RudpServerSpawnOptions = {}): RunningPythonRudpPeer {
+  const child = spawn(rustBinaryPath, [
+    "rudp-serve-once",
+    "--bind-host", "127.0.0.1",
+    ...(options.clientPayload ? ["--client-payload", options.clientPayload] : []),
+    ...(options.serverPayload ? ["--server-payload", options.serverPayload] : []),
+    ...(options.maxFragmentBytes ? ["--max-fragment-bytes", String(options.maxFragmentBytes)] : []),
+  ], {
     cwd: cultnetRsRoot,
     env: process.env,
     stdio: ["ignore", "pipe", "pipe"],
@@ -1403,8 +1480,15 @@ function spawnRustRudpServer(): RunningPythonRudpPeer {
   return { child, ready, stderr };
 }
 
-function spawnCSharpRudpServer(): RunningPythonRudpPeer {
-  const child = spawn(dotnetCommand, [csharpDllPath, "rudp-serve-once", "--bind-host", "127.0.0.1"], {
+function spawnCSharpRudpServer(options: RudpServerSpawnOptions = {}): RunningPythonRudpPeer {
+  const child = spawn(dotnetCommand, [
+    csharpDllPath,
+    "rudp-serve-once",
+    "--bind-host", "127.0.0.1",
+    ...(options.clientPayload ? ["--client-payload", options.clientPayload] : []),
+    ...(options.serverPayload ? ["--server-payload", options.serverPayload] : []),
+    ...(options.maxFragmentBytes ? ["--max-fragment-bytes", String(options.maxFragmentBytes)] : []),
+  ], {
     cwd: cultLibRoot,
     env: process.env,
     stdio: ["ignore", "pipe", "pipe"],
@@ -1453,12 +1537,15 @@ function spawnCSharpRudpServer(): RunningPythonRudpPeer {
   return { child, ready, stderr };
 }
 
-function spawnKotlinRudpServer(): RunningPythonRudpPeer {
+function spawnKotlinRudpServer(options: RudpServerSpawnOptions = {}): RunningPythonRudpPeer {
   const child = spawn(kotlinJavaCommand, [
     "-cp", kotlinClasspath,
     "org.gamecult.cultmesh.CultMeshKt",
     "rudp-serve-once",
     "--bind-host", "127.0.0.1",
+    ...(options.clientPayload ? ["--client-payload", options.clientPayload] : []),
+    ...(options.serverPayload ? ["--server-payload", options.serverPayload] : []),
+    ...(options.maxFragmentBytes ? ["--max-fragment-bytes", String(options.maxFragmentBytes)] : []),
   ], {
     cwd: cultLibRoot,
     env: process.env,

src/GameCult.Networking/CultNetTransport.cs

@@ -561,14 +561,31 @@ namespace GameCult.Networking
             public long LastSentAtMs { get; set; }
         }
 
+        private sealed class PendingOrderedFrame
+        {
+            public CultNetRudpDeliveredFrame Frame { get; set; } = new CultNetRudpDeliveredFrame();
+            public uint NextSequence { get; set; }
+        }
+
+        private sealed class FragmentBuffer
+        {
+            public string ChannelId { get; set; } = string.Empty;
+            public bool Ordered { get; set; }
+            public ushort FragmentCount { get; set; }
+            public Dictionary<ushort, byte[]> Payloads { get; } = new Dictionary<ushort, byte[]>();
+            public Dictionary<ushort, uint> Sequences { get; } = new Dictionary<ushort, uint>();
+        }
+
         private uint _nextSequence;
+        private ushort _nextFragmentId = 1;
         private bool _connected;
         private uint? _highestReceivedSequence;
         private readonly HashSet<uint> _receivedSequences = new HashSet<uint>();
         private readonly Dictionary<uint, PendingReliablePacket> _pendingReliable = new Dictionary<uint, PendingReliablePacket>();
         private readonly Dictionary<string, uint> _orderedNextSequenceByChannel = new Dictionary<string, uint>(StringComparer.Ordinal);
-        private readonly Dictionary<string, SortedDictionary<uint, CultNetRudpDeliveredFrame>> _orderedBuffers =
-            new Dictionary<string, SortedDictionary<uint, CultNetRudpDeliveredFrame>>(StringComparer.Ordinal);
+        private readonly Dictionary<string, SortedDictionary<uint, PendingOrderedFrame>> _orderedBuffers =
+            new Dictionary<string, SortedDictionary<uint, PendingOrderedFrame>>(StringComparer.Ordinal);
+        private readonly Dictionary<string, FragmentBuffer> _fragmentBuffers = new Dictionary<string, FragmentBuffer>(StringComparer.Ordinal);
 
         /// <summary>
         /// Initializes a RUDP reliability session.
@@ -630,6 +647,14 @@ namespace GameCult.Networking
         /// Creates a data packet.
         /// </summary>
         public CultNetRudpPacket Send(string channelId, byte[] payload, CultNetRudpSendOptions? options = null)
+        {
+            return SendMany(channelId, payload, options).First();
+        }
+
+        /// <summary>
+        /// Creates one or more data packets, fragmenting when requested.
+        /// </summary>
+        public IReadOnlyList<CultNetRudpPacket> SendMany(string channelId, byte[] payload, CultNetRudpSendOptions? options = null, int? maxFragmentBytes = null)
         {
             if (!_connected)
             {
@@ -637,6 +662,47 @@ namespace GameCult.Networking
             }
 
             options ??= new CultNetRudpSendOptions();
+            payload ??= Array.Empty<byte>();
+            if (maxFragmentBytes.HasValue && maxFragmentBytes.Value <= 0)
+            {
+                throw new ArgumentOutOfRangeException(nameof(maxFragmentBytes), "RUDP maxFragmentBytes must be greater than zero.");
+            }
+
+            if (maxFragmentBytes.HasValue && payload.Length > maxFragmentBytes.Value)
+            {
+                var fragmentCount = (payload.Length + maxFragmentBytes.Value - 1) / maxFragmentBytes.Value;
+                if (fragmentCount > ushort.MaxValue)
+                {
+                    throw new InvalidOperationException("RUDP payload requires more than 65535 fragments.");
+                }
+
+                var fragmentId = AllocateFragmentId();
+                var packets = new List<CultNetRudpPacket>();
+                for (var index = 0; index < fragmentCount; index++)
+                {
+                    var start = index * maxFragmentBytes.Value;
+                    var length = Math.Min(maxFragmentBytes.Value, payload.Length - start);
+                    var chunk = new byte[length];
+                    Array.Copy(payload, start, chunk, 0, length);
+                    var fragmentPacket = CreatePacket(
+                        CultNetRudpPacketType.Data,
+                        channelId,
+                        chunk,
+                        options.Reliable,
+                        options.Ordered,
+                        options.Sequenced,
+                        fragmentId,
+                        (ushort)index,
+                        (ushort)fragmentCount);
+                    if (fragmentPacket.Reliable)
+                    {
+                        TrackReliable(fragmentPacket, options.NowMs);
+                    }
+                    packets.Add(fragmentPacket);
+                }
+                return packets;
+            }
+
             var packet = CreatePacket(
                 CultNetRudpPacketType.Data,
                 channelId,
@@ -648,7 +714,7 @@ namespace GameCult.Networking
             {
                 TrackReliable(packet, options.NowMs);
             }
-            return packet;
+            return new[] { packet };
         }
 
         /// <summary>
@@ -699,16 +765,17 @@ namespace GameCult.Networking
                 return new CultNetRudpReceiveResult();
             }
 
-            var frame = new CultNetRudpDeliveredFrame
+            var reassembled = Reassemble(packet);
+            if (reassembled == null)
             {
-                ChannelId = packet.ChannelId,
-                Payload = packet.Payload ?? Array.Empty<byte>(),
-                Sequence = packet.Sequence
-            };
+                return new CultNetRudpReceiveResult();
+            }
 
             return new CultNetRudpReceiveResult
             {
-                Delivered = packet.Ordered ? DeliverOrdered(frame) : new[] { frame }
+                Delivered = reassembled.Ordered
+                    ? DeliverOrdered(reassembled.Frame, reassembled.NextSequence)
+                    : new[] { reassembled.Frame }
             };
         }
 
@@ -745,6 +812,20 @@ namespace GameCult.Networking
             bool reliable,
             bool ordered,
             bool sequenced)
+        {
+            return CreatePacket(packetType, channelId, payload, reliable, ordered, sequenced, 0, 0, 0);
+        }
+
+        private CultNetRudpPacket CreatePacket(
+            CultNetRudpPacketType packetType,
+            string channelId,
+            byte[] payload,
+            bool reliable,
+            bool ordered,
+            bool sequenced,
+            ushort fragmentId,
+            ushort fragmentIndex,
+            ushort fragmentCount)
         {
             var sequence = _nextSequence;
             _nextSequence = checked(_nextSequence + 1);
@@ -760,6 +841,9 @@ namespace GameCult.Networking
                 Reliable = reliable,
                 Ordered = ordered,
                 Sequenced = sequenced,
+                FragmentId = fragmentId,
+                FragmentIndex = fragmentIndex,
+                FragmentCount = fragmentCount,
                 Payload = payload ?? Array.Empty<byte>()
             };
         }
@@ -809,11 +893,90 @@ namespace GameCult.Networking
             return (ack, ackMask);
         }
 
-        private IReadOnlyList<CultNetRudpDeliveredFrame> DeliverOrdered(CultNetRudpDeliveredFrame frame)
+        private sealed class ReassembledFrame
+        {
+            public CultNetRudpDeliveredFrame Frame { get; set; } = new CultNetRudpDeliveredFrame();
+            public bool Ordered { get; set; }
+            public uint NextSequence { get; set; }
+        }
+
+        private ReassembledFrame? Reassemble(CultNetRudpPacket packet)
+        {
+            if (packet.FragmentCount == 0)
+            {
+                return new ReassembledFrame
+                {
+                    Frame = new CultNetRudpDeliveredFrame
+                    {
+                        ChannelId = packet.ChannelId,
+                        Payload = packet.Payload ?? Array.Empty<byte>(),
+                        Sequence = packet.Sequence
+                    },
+                    Ordered = packet.Ordered,
+                    NextSequence = packet.Sequence + 1
+                };
+            }
+            if (packet.FragmentId == 0)
+            {
+                throw new InvalidOperationException("RUDP fragmented packet must have a non-zero fragment id.");
+            }
+            if (packet.FragmentIndex >= packet.FragmentCount)
+            {
+                throw new InvalidOperationException("RUDP fragment index must be lower than fragment count.");
+            }
+
+            var key = $"{packet.ChannelId}\0{packet.FragmentId}";
+            if (!_fragmentBuffers.TryGetValue(key, out var buffer))
+            {
+                buffer = new FragmentBuffer
+                {
+                    ChannelId = packet.ChannelId,
+                    Ordered = packet.Ordered,
+                    FragmentCount = packet.FragmentCount
+                };
+                _fragmentBuffers[key] = buffer;
+            }
+            if (buffer.FragmentCount != packet.FragmentCount || buffer.Ordered != packet.Ordered)
+            {
+                throw new InvalidOperationException("RUDP fragment metadata changed within a fragment set.");
+            }
+
+            buffer.Payloads[packet.FragmentIndex] = packet.Payload?.ToArray() ?? Array.Empty<byte>();
+            buffer.Sequences[packet.FragmentIndex] = packet.Sequence;
+            if (buffer.Payloads.Count < packet.FragmentCount)
+            {
+                return null;
+            }
+
+            var payloadLength = buffer.Payloads.Values.Sum(chunk => chunk.Length);
+            var payload = new byte[payloadLength];
+            var offset = 0;
+            for (ushort index = 0; index < packet.FragmentCount; index++)
+            {
+                var chunk = buffer.Payloads[index];
+                Array.Copy(chunk, 0, payload, offset, chunk.Length);
+                offset += chunk.Length;
+            }
+            var sequences = buffer.Sequences.Values.ToArray();
+            _fragmentBuffers.Remove(key);
+            return new ReassembledFrame
+            {
+                Frame = new CultNetRudpDeliveredFrame
+                {
+                    ChannelId = buffer.ChannelId,
+                    Payload = payload,
+                    Sequence = sequences.Min()
+                },
+                Ordered = buffer.Ordered,
+                NextSequence = sequences.Max() + 1
+            };
+        }
+
+        private IReadOnlyList<CultNetRudpDeliveredFrame> DeliverOrdered(CultNetRudpDeliveredFrame frame, uint nextSequenceAfterFrame)
         {
             if (!_orderedNextSequenceByChannel.TryGetValue(frame.ChannelId, out var next))
             {
-                _orderedNextSequenceByChannel[frame.ChannelId] = frame.Sequence + 1;
+                _orderedNextSequenceByChannel[frame.ChannelId] = nextSequenceAfterFrame;
                 return new[] { frame }.Concat(DrainOrdered(frame.ChannelId)).ToArray();
             }
 
@@ -826,15 +989,15 @@ namespace GameCult.Networking
             {
                 if (!_orderedBuffers.TryGetValue(frame.ChannelId, out var buffer))
                 {
-                    buffer = new SortedDictionary<uint, CultNetRudpDeliveredFrame>();
+                    buffer = new SortedDictionary<uint, PendingOrderedFrame>();
                     _orderedBuffers[frame.ChannelId] = buffer;
                 }
 
-                buffer[frame.Sequence] = frame;
+                buffer[frame.Sequence] = new PendingOrderedFrame { Frame = frame, NextSequence = nextSequenceAfterFrame };
                 return Array.Empty<CultNetRudpDeliveredFrame>();
             }
 
-            _orderedNextSequenceByChannel[frame.ChannelId] = next + 1;
+            _orderedNextSequenceByChannel[frame.ChannelId] = nextSequenceAfterFrame;
             return new[] { frame }.Concat(DrainOrdered(frame.ChannelId)).ToArray();
         }
 
@@ -846,16 +1009,27 @@ namespace GameCult.Networking
                 return delivered;
             }
 
-            while (_orderedNextSequenceByChannel.TryGetValue(channelId, out var next) && buffer.TryGetValue(next, out var frame))
+            while (_orderedNextSequenceByChannel.TryGetValue(channelId, out var next) && buffer.TryGetValue(next, out var pending))
             {
                 buffer.Remove(next);
-                delivered.Add(frame);
-                _orderedNextSequenceByChannel[channelId] = next + 1;
+                delivered.Add(pending.Frame);
+                _orderedNextSequenceByChannel[channelId] = pending.NextSequence;
             }
 
             return delivered;
         }
 
+        private ushort AllocateFragmentId()
+        {
+            var fragmentId = _nextFragmentId;
+            _nextFragmentId++;
+            if (_nextFragmentId == 0)
+            {
+                _nextFragmentId = 1;
+            }
+            return fragmentId;
+        }
+
         private void RequireConnection(CultNetRudpPacket packet)
         {
             if (packet.ConnectionId != ConnectionId)
@@ -893,6 +1067,7 @@ namespace GameCult.Networking
         private readonly Socket _socket;
         private readonly CultNetRudpSession _session;
         private readonly CultNetRudpSocketMode _mode;
+        private readonly int? _maxFragmentBytes;
         private readonly CultNetTransportStats _stats = new CultNetTransportStats();
         private readonly Queue<CultNetTransportFrame> _deliveredFrames = new Queue<CultNetTransportFrame>();
         private EndPoint? _remoteEndPoint;
@@ -908,6 +1083,7 @@ namespace GameCult.Networking
             _socket = options.Socket ?? throw new ArgumentNullException(nameof(options.Socket));
             _mode = options.Mode;
             _remoteEndPoint = options.RemoteEndPoint;
+            _maxFragmentBytes = options.MaxFragmentBytes;
             _session = new CultNetRudpSession(new CultNetRudpSessionOptions
             {
                 ConnectionId = options.ConnectionId,
@@ -961,8 +1137,10 @@ namespace GameCult.Networking
         /// </summary>
         public void Send(string channelId, byte[] payload)
         {
-            var packet = _session.Send(channelId, payload, ChannelSendOptions(channelId));
-            SendPacket(packet);
+            foreach (var packet in _session.SendMany(channelId, payload, ChannelSendOptions(channelId), _maxFragmentBytes))
+            {
+                SendPacket(packet);
+            }
             _stats.FramesSent++;
         }
 

tests/GameCult.Networking.InteropPeer/Program.cs

@@ -583,6 +583,15 @@ static void RudpServeOnce(Dictionary<string, string> options)
     var bindPort = options.TryGetValue("bind-port", out var configuredBindPort)
         ? int.Parse(configuredBindPort, CultureInfo.InvariantCulture)
         : 0;
+    var expectedClientPayload = Ascii(options.TryGetValue("client-payload", out var configuredClientPayload)
+        ? configuredClientPayload
+        : "ts-csharp-client-state");
+    var serverPayload = Ascii(options.TryGetValue("server-payload", out var configuredServerPayload)
+        ? configuredServerPayload
+        : "csharp-server-state");
+    var maxFragmentBytes = options.TryGetValue("max-fragment-bytes", out var configuredMaxFragmentBytes)
+        ? int.Parse(configuredMaxFragmentBytes, CultureInfo.InvariantCulture)
+        : (int?)null;
 
     using var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
     socket.Bind(new IPEndPoint(IPAddress.Parse(bindHost), bindPort));
@@ -595,7 +604,8 @@ static void RudpServeOnce(Dictionary<string, string> options)
         Mode = CultNetRudpSocketMode.Server,
         ConnectionId = 0x33557799,
         InitialSequence = 100,
-        ResendDelayMs = 25
+        ResendDelayMs = 25,
+        MaxFragmentBytes = maxFragmentBytes
     });
 
     WriteJsonLine(new
@@ -610,8 +620,8 @@ static void RudpServeOnce(Dictionary<string, string> options)
         var frame = transport.ReceiveOnce();
         if (frame != null)
         {
-            RequireRudpFrame(frame, "schema", "ts-csharp-client-state");
-            transport.Send("schema", Ascii("csharp-server-state"));
+            RequireRudpFrameBytes(frame, "schema", expectedClientPayload);
+            transport.Send("schema", serverPayload);
             PollRudpAfterSend(transport, TimeSpan.FromMilliseconds(250));
             WriteJsonLine(new { status = "ok" });
             return;
@@ -680,9 +690,14 @@ static void PollRudpAfterSend(CultNetRudpSocketTransportConnection transport, Ti
 }
 
 static void RequireRudpFrame(CultNetTransportFrame frame, string expectedChannelId, string expectedPayload)
+{
+    RequireRudpFrameBytes(frame, expectedChannelId, Ascii(expectedPayload));
+}
+
+static void RequireRudpFrameBytes(CultNetTransportFrame frame, string expectedChannelId, byte[] expectedPayload)
 {
     if (!string.Equals(frame.ChannelId, expectedChannelId, StringComparison.Ordinal)
-        || !frame.Payload.SequenceEqual(Ascii(expectedPayload)))
+        || !frame.Payload.SequenceEqual(expectedPayload))
     {
         throw new InvalidOperationException(
             $"Unexpected RUDP frame: channel={frame.ChannelId}, payload={Convert.ToHexString(frame.Payload)}.");

tests/GameCult.Networking.Tests/NetworkingTests.cs

@@ -494,6 +494,42 @@ namespace GameCult.Networking.Tests
                 Is.EqualTo(new[] { "second", "third" }));
         }
 
+        [Test]
+        public void RudpSession_FragmentsAndReassemblesReliableOrderedPayloads()
+        {
+            var sender = new CultNetRudpSession(new CultNetRudpSessionOptions
+            {
+                ConnectionId = 456,
+                InitialSequence = 1
+            });
+            var receiver = new CultNetRudpSession(new CultNetRudpSessionOptions
+            {
+                ConnectionId = 456,
+                InitialSequence = 100
+            });
+            sender.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = 456, Sequence = 90, ChannelId = "control" });
+            receiver.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = 456, Sequence = 91, ChannelId = "control" });
+
+            var packets = sender.SendMany(
+                "schema",
+                Encoding.UTF8.GetBytes("fragment-me-please"),
+                new CultNetRudpSendOptions { Reliable = true, Ordered = true, NowMs = 10 },
+                maxFragmentBytes: 5).ToArray();
+
+            Assert.That(packets, Has.Length.EqualTo(4));
+            Assert.That(packets.Select(packet => packet.FragmentCount).ToArray(), Is.EqualTo(new ushort[] { 4, 4, 4, 4 }));
+            Assert.That(packets.Select(packet => packet.FragmentIndex).ToArray(), Is.EqualTo(new ushort[] { 0, 1, 2, 3 }));
+            Assert.That(packets.All(packet => packet.FragmentId == packets[0].FragmentId), Is.True);
+
+            Assert.That(receiver.Receive(packets[0]).Delivered, Is.Empty);
+            Assert.That(receiver.Receive(packets[1]).Delivered, Is.Empty);
+            Assert.That(receiver.Receive(packets[2]).Delivered, Is.Empty);
+            var delivered = receiver.Receive(packets[3]).Delivered.ToArray();
+            Assert.That(delivered, Has.Length.EqualTo(1));
+            Assert.That(Encoding.UTF8.GetString(delivered[0].Payload), Is.EqualTo("fragment-me-please"));
+            Assert.That(delivered[0].Sequence, Is.EqualTo(packets[0].Sequence));
+        }
+
         [Test]
         public void RudpSocketTransport_HandshakesAndCarriesReliableOrderedSchemaFrames()
         {
@@ -540,6 +576,46 @@ namespace GameCult.Networking.Tests
             Assert.That(server.Stats.FramesReceived, Is.EqualTo(1));
         }
 
+        [Test]
+        public void RudpSocketTransport_CarriesFragmentedReliableOrderedSchemaFrames()
+        {
+            using var serverSocket = BindUdpSocket();
+            using var clientSocket = BindUdpSocket();
+            var serverEndPoint = serverSocket.LocalEndPoint!;
+            const uint connectionId = 0x10203041;
+            using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
+            {
+                RuntimeId = "csharp-rudp-fragment-server",
+                Socket = serverSocket,
+                Mode = CultNetRudpSocketMode.Server,
+                ConnectionId = connectionId,
+                InitialSequence = 100,
+                ResendDelayMs = 25,
+                MaxFragmentBytes = 8
+            });
+            using var client = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
+            {
+                RuntimeId = "csharp-rudp-fragment-client",
+                Socket = clientSocket,
+                Mode = CultNetRudpSocketMode.Client,
+                RemoteEndPoint = serverEndPoint,
+                ConnectionId = connectionId,
+                InitialSequence = 1,
+                ResendDelayMs = 25,
+                MaxFragmentBytes = 8
+            });
+
+            var payload = Encoding.UTF8.GetBytes("this-schema-frame-is-larger-than-one-rudp-fragment");
+            client.Connect(Encoding.UTF8.GetBytes("join"));
+            PumpRudpHandshake(client, server);
+            client.Send("schema", payload);
+            var serverFrame = ReceiveRudpFrame(server);
+            Assert.That(serverFrame.ChannelId, Is.EqualTo("schema"));
+            Assert.That(serverFrame.Payload, Is.EqualTo(payload));
+            Assert.That(client.Stats.FramesSent, Is.EqualTo(1));
+            Assert.That(server.Stats.FramesReceived, Is.EqualTo(1));
+        }
+
         [Test]
         public void CultNetSchemaMessageSerialization_RoundTrips_RawSnapshotResponse()
         {