Protocol/Server: hybrid body TLV (2-bit vs RLE), state_full body builder, and partitioned snapshot on join

- Protocol: RLE packing, TLV chooser, body_2bit_chunk helper, state_full_body builder
- Server: on join, build snapshot body and partition across PART packets if over MTU; include apples only in first part; chunk single oversized snake with 2-bit chunking

server/protocol.py

@@ -192,6 +192,37 @@ def bitpack_2bit_directions(directions: Iterable[Direction]) -> bytes:
     return bytes(out)
 
 
+def rle_pack_directions(directions: Sequence[Direction]) -> bytes:
+    """Pack directions as runs: dir (u8) + count (QUIC varint >=1)."""
+    if not directions:
+        return b""
+    out = bytearray()
+    run_dir = directions[0]
+    run_len = 1
+    for d in directions[1:]:
+        if d == run_dir:
+            run_len += 1
+        else:
+            out.append(int(run_dir) & 0xFF)
+            out.extend(quic_varint_encode(run_len))
+            run_dir = d
+            run_len = 1
+    out.append(int(run_dir) & 0xFF)
+    out.extend(quic_varint_encode(run_len))
+    return bytes(out)
+
+
+def build_body_tlv_for_dirs(directions: Sequence[Direction]) -> bytes:
+    """Choose the more compact of 2-bit stream vs RLE for given directions and return its TLV."""
+    # 2-bit body size
+    two_bit_payload = bitpack_2bit_directions(directions)
+    two_bit = pack_body_tlv(BodyTLV.BODY_2BIT, two_bit_payload)
+    # RLE
+    rle_payload = rle_pack_directions(directions)
+    rle = pack_body_tlv(BodyTLV.BODY_RLE, rle_payload)
+    return two_bit if len(two_bit) <= len(rle) else rle
+
+
 # Join / Ack / Deny
 
 def build_join(name_utf8: bytes, preferred_color_id: int | None = None) -> bytes:
@@ -280,6 +311,13 @@ def build_state_full(
     apples: sequence of (x, y)
     """
     header = pack_header(version, PacketType.STATE_FULL, 0, seq, tick & 0xFFFF)
+    body = build_state_full_body(snakes=snakes, apples=apples)
+    return header + body
+
+
+def build_state_full_body(
+    *, snakes: Sequence[Tuple[int, int, int, int, Sequence[Direction]]], apples: Sequence[Tuple[int, int]]
+) -> bytes:
     body = bytearray()
     # snakes count
     body.extend(quic_varint_encode(len(snakes)))
@@ -288,15 +326,29 @@ def build_state_full(
         body.extend(int(slen & 0xFFFF).to_bytes(2, "big"))
         body.append(hx & 0xFF)
         body.append(hy & 0xFF)
-        payload = bitpack_2bit_directions(dirs)
+        tlv = build_body_tlv_for_dirs(dirs)
-        tlv = pack_body_tlv(BodyTLV.BODY_2BIT, payload)
         body.extend(tlv)
     # apples
     body.extend(quic_varint_encode(len(apples)))
     for ax, ay in apples:
         body.append(ax & 0xFF)
         body.append(ay & 0xFF)
-    return header + bytes(body)
+    return bytes(body)
+
+
+def build_body_2bit_chunk(directions: Sequence[Direction], start_index: int, dirs_in_chunk: int) -> bytes:
+    """Build chunk TLV for a sub-range of directions (2-bit)."""
+    sub = directions[start_index : start_index + dirs_in_chunk]
+    payload = bytearray()
+    payload.extend(int(start_index & 0xFFFF).to_bytes(2, "big"))
+    payload.extend(int(dirs_in_chunk & 0xFFFF).to_bytes(2, "big"))
+    payload.extend(bitpack_2bit_directions(sub))
+    return pack_body_tlv(BodyTLV.BODY_2BIT_CHUNK, bytes(payload))
+
+
+def estimate_body_2bit_bytes(snake_len: int) -> int:
+    used_bits = max(0, (snake_len - 1) * 2)
+    return (used_bits + 7) // 8
 
 
 # --- State Delta ---

server/server.py

@@ -16,6 +16,9 @@ from .protocol import (
     build_config_update,
     build_input_broadcast,
     build_state_full,
+    build_state_full_body,
+    build_body_tlv_for_dirs,
+    build_body_2bit_chunk,
     build_state_delta,
     build_state_delta_body,
     build_part,
@@ -416,7 +419,7 @@ class GameServer:
             compression_mode=0 if cfg.compression_mode == "none" else 1,
         )
         await self.transport.send(ack, peer)
-        # Send initial state_full
+        # Send initial state_full (partitioned if needed)
         snakes_dirs: List[Tuple[int, int, int, int, List[Direction]]] = []
         for s in self.runtime.state.snakes.values():
             dirs: List[Direction] = []
@@ -435,14 +438,117 @@ class GameServer:
                     dirs.append(Direction.LEFT)
             hx, hy = coords[0]
             snakes_dirs.append((s.snake_id, s.length, hx, hy, dirs))
-        full = build_state_full(
+        # Build body and partition across parts if needed
-            version=self.runtime.version,
+        body = build_state_full_body(snakes=snakes_dirs, apples=self.runtime.state.apples)
-            seq=self.runtime.next_seq(),
+        MTU = 1200
-            tick=self.runtime.state.tick,
+        if len(body) <= MTU:
-            snakes=snakes_dirs,
+            full = pack_header(
-            apples=self.runtime.state.apples,
+                self.runtime.version, PacketType.STATE_FULL, 0, self.runtime.next_seq(), self.runtime.state.tick & 0xFFFF
-        )
+            ) + body
-        await self.transport.send(full, peer)
+            await self.transport.send(full, peer)
+        else:
+            # Partition by packing whole snakes first, apples only in first part; chunk a single oversized snake using 2-bit chunks
+            update_id = self.runtime.next_update_id()
+            parts: List[bytes] = []
+            # Prepare apples buffer for first part
+            def encode_apples(apples: List[Coord]) -> bytes:
+                from .protocol import quic_varint_encode
+
+                b = bytearray()
+                b.extend(quic_varint_encode(len(apples)))
+                for ax, ay in apples:
+                    b.append(ax & 0xFF)
+                    b.append(ay & 0xFF)
+                return bytes(b)
+
+            apples_encoded = encode_apples(self.runtime.state.apples)
+            # Cursor over snakes, building per part bodies
+            remaining = list(snakes_dirs)
+            first = True
+            while remaining:
+                # Start part body with snakes count placeholder; we'll rebuild as we pack
+                part_snakes: List[bytes] = []
+                packed_snakes = 0
+                budget = MTU
+                # Reserve apples size on first part
+                apples_this = apples_encoded if first else encode_apples([])
+                budget -= len(apples_this)
+                # Greedily add snakes
+                i = 0
+                while i < len(remaining):
+                    sid, slen, hx, hy, dirs = remaining[i]
+                    # Build this snake record with chosen TLV
+                    tlv = build_body_tlv_for_dirs(dirs)
+                    record = bytearray()
+                    # one snake: id u8, len u16, head x y, TLV
+                    record.append(sid & 0xFF)
+                    record.extend(int(slen & 0xFFFF).to_bytes(2, "big"))
+                    record.append(hx & 0xFF)
+                    record.append(hy & 0xFF)
+                    record.extend(tlv)
+                    if len(record) <= budget:
+                        part_snakes.append(bytes(record))
+                        budget -= len(record)
+                        packed_snakes += 1
+                        i += 1
+                        continue
+                    # If single snake doesn't fit and no snakes packed yet, chunk this snake
+                    if packed_snakes == 0:
+                        # Use 2-bit chunking; compute directions chunk size to fit budget minus fixed headers (id/len/head + TLV type/len + chunk header)
+                        # Fixed snake header = 1+2+1+1 = 5 bytes; TLV type/len (worst-case 2 bytes for small values) + chunk header 4 bytes
+                        overhead = 5 + 2 + 4
+                        dir_capacity_bytes = max(0, budget - overhead)
+                        if dir_capacity_bytes <= 0:
+                            break
+                        # Convert capacity bytes to number of directions (each 2 bits → 4 dirs per byte)
+                        dir_capacity = dir_capacity_bytes * 4
+                        if dir_capacity <= 0:
+                            break
+                        # Emit at least one direction
+                        dirs_in_chunk = max(1, min(dir_capacity, len(dirs)))
+                        tlv_chunk = build_body_2bit_chunk(dirs, start_index=0, dirs_in_chunk=dirs_in_chunk)
+                        record = bytearray()
+                        record.append(sid & 0xFF)
+                        record.extend(int(slen & 0xFFFF).to_bytes(2, "big"))
+                        record.append(hx & 0xFF)
+                        record.append(hy & 0xFF)
+                        record.extend(tlv_chunk)
+                        if len(record) <= budget:
+                            part_snakes.append(bytes(record))
+                            budget -= len(record)
+                            # Replace remaining snake with truncated version (advance dirs)
+                            remaining[i] = (sid, slen, hx, hy, list(dirs)[dirs_in_chunk:])
+                            packed_snakes += 1
+                        else:
+                            break
+                    else:
+                        break
+                # Build part body: snakes_count + snakes + apples
+                body_part = bytearray()
+                from .protocol import quic_varint_encode
+
+                body_part.extend(quic_varint_encode(len(part_snakes)))
+                for rec in part_snakes:
+                    body_part.extend(rec)
+                body_part.extend(apples_this)
+                parts.append(bytes(body_part))
+                # Advance remaining list
+                remaining = remaining[packed_snakes:]
+                first = False
+            # Emit part packets
+            total = len(parts)
+            for i, chunk_body in enumerate(parts):
+                pkt = build_part(
+                    version=self.runtime.version,
+                    seq=self.runtime.next_seq(),
+                    tick=self.runtime.state.tick & 0xFFFF,
+                    update_id=update_id,
+                    part_index=i,
+                    parts_total=total,
+                    inner_type=PacketType.STATE_FULL,
+                    chunk_payload=chunk_body,
+                )
+                await self.transport.send(pkt, peer)
 
     async def _handle_input(self, buf: bytes, off: int, peer: TransportPeer) -> None:
         try: