Protocol/Server: implement STATE_DELTA + PART partitioning and per-tick minimal changes

- Protocol: SnakeDelta structure; build_state_delta(_body); build_part
- Server: compute per-snake changes (move/grow/blocked-shrink), apples diffs
- Partition large deltas by snake changes with apples in first part; use update_id
- Send STATE_DELTA when under MTU; else PART packets referencing STATE_DELTA

server/protocol.py

@@ -297,3 +297,80 @@ def build_state_full(
         body.append(ax & 0xFF)
         body.append(ay & 0xFF)
     return header + bytes(body)
+
+
+# --- State Delta ---
+
+@dataclass
+class SnakeDelta:
+    snake_id: int
+    head_moved: bool
+    tail_removed: bool
+    grew: bool
+    blocked: bool
+    new_head_x: int = 0
+    new_head_y: int = 0
+    direction: Direction = Direction.RIGHT
+
+
+def build_state_delta_body(
+    *, update_id: int, changes: Sequence[SnakeDelta], apples_added: Sequence[Tuple[int, int]], apples_removed: Sequence[Tuple[int, int]]
+) -> bytes:
+    body = bytearray()
+    body.extend(int(update_id & 0xFFFF).to_bytes(2, "big"))
+    # snakes
+    body.extend(quic_varint_encode(len(changes)))
+    for ch in changes:
+        body.append(ch.snake_id & 0xFF)
+        flags = (
+            (1 if ch.head_moved else 0)
+            | ((1 if ch.tail_removed else 0) << 1)
+            | ((1 if ch.grew else 0) << 2)
+            | ((1 if ch.blocked else 0) << 3)
+        )
+        body.append(flags & 0xFF)
+        body.append(int(ch.direction) & 0x03)
+        if ch.head_moved:
+            body.append(ch.new_head_x & 0xFF)
+            body.append(ch.new_head_y & 0xFF)
+    # apples added
+    body.extend(quic_varint_encode(len(apples_added)))
+    for ax, ay in apples_added:
+        body.append(ax & 0xFF)
+        body.append(ay & 0xFF)
+    # apples removed
+    body.extend(quic_varint_encode(len(apples_removed)))
+    for rx, ry in apples_removed:
+        body.append(rx & 0xFF)
+        body.append(ry & 0xFF)
+    return bytes(body)
+
+
+def build_state_delta(
+    *, version: int, seq: int, tick: int, update_id: int, changes: Sequence[SnakeDelta], apples_added: Sequence[Tuple[int, int]], apples_removed: Sequence[Tuple[int, int]]
+) -> bytes:
+    header = pack_header(version, PacketType.STATE_DELTA, 0, seq, tick & 0xFFFF)
+    body = build_state_delta_body(update_id=update_id, changes=changes, apples_added=apples_added, apples_removed=apples_removed)
+    return header + body
+
+
+def build_part(
+    *,
+    version: int,
+    seq: int,
+    tick: int,
+    update_id: int,
+    part_index: int,
+    parts_total: int,
+    inner_type: PacketType,
+    chunk_payload: bytes,
+) -> bytes:
+    header = pack_header(version, PacketType.PART, 0, seq, tick & 0xFFFF)
+    body = bytearray()
+    body.extend(int(update_id & 0xFFFF).to_bytes(2, "big"))
+    body.append(part_index & 0xFF)
+    body.append(parts_total & 0xFF)
+    body.append(int(inner_type) & 0xFF)
+    # include the chunk payload bytes
+    body.extend(chunk_payload)
+    return header + bytes(body)

server/server.py

@@ -16,9 +16,10 @@ from .protocol import (
     build_config_update,
     build_input_broadcast,
     build_state_full,
-    BodyTLV,
+    build_state_delta,
-    build_state_full,
+    build_state_delta_body,
-    quic_varint_encode,
+    build_part,
+    SnakeDelta,
     pack_header,
     parse_input,
     parse_join,
@@ -32,11 +33,16 @@ class ServerRuntime:
     state: GameState
     seq: int = 0
     version: int = 1
+    update_id: int = 0
 
     def next_seq(self) -> int:
         self.seq = (self.seq + 1) & 0xFFFF
         return self.seq
 
+    def next_update_id(self) -> int:
+        self.update_id = (self.update_id + 1) & 0xFFFF
+        return self.update_id
+
 
 class GameServer:
     def __init__(self, transport: DatagramServerTransport, config: ServerConfig):
@@ -157,6 +163,8 @@ class GameServer:
         st = self.runtime.state
         cfg = self.runtime.config
         apples_eaten: List[Coord] = []
+        apples_before = set(st.apples)
+        changes: List[SnakeDelta] = []
         # Prepare snapshot of tails to allow moving into own tail when it vacates
         tails: Dict[int, Coord] = {}
         for sid, s in st.snakes.items():
@@ -187,6 +195,18 @@ class GameServer:
                 if s.length > 1:
                     tx, ty = s.body.pop()
                     st.occupancy.pop((tx, ty), None)
+                changes.append(
+                    SnakeDelta(
+                        snake_id=sid,
+                        head_moved=False,
+                        tail_removed=(s.length > 1),
+                        grew=False,
+                        blocked=True,
+                        new_head_x=hx,
+                        new_head_y=hy,
+                        direction=s.direction,
+                    )
+                )
                 continue
             # Move or grow
             s.blocked = False
@@ -198,40 +218,98 @@ class GameServer:
                 # eat apple; no tail removal this tick
                 st.apples.remove((nx, ny))
                 apples_eaten.append((nx, ny))
+                changes.append(
+                    SnakeDelta(
+                        snake_id=sid,
+                        head_moved=True,
+                        tail_removed=False,
+                        grew=True,
+                        blocked=False,
+                        new_head_x=nx,
+                        new_head_y=ny,
+                        direction=s.direction,
+                    )
+                )
             else:
                 # normal move: remove tail (unless length==0)
                 tx, ty = s.body.pop()
                 if (tx, ty) in st.occupancy:
                     st.occupancy.pop((tx, ty), None)
+                changes.append(
+                    SnakeDelta(
+                        snake_id=sid,
+                        head_moved=True,
+                        tail_removed=True,
+                        grew=False,
+                        blocked=False,
+                        new_head_x=nx,
+                        new_head_y=ny,
+                        direction=s.direction,
+                    )
+                )
         # Replenish apples to target
         self._ensure_apples()
+        apples_after = set(st.apples)
+        apples_added = sorted(list(apples_after - apples_before))
+        apples_removed = sorted(list(apples_before - apples_after))
         # Broadcast a basic delta (currently full snakes + apples as delta)
-        snakes_dirs: List[Tuple[int, int, int, int, List[Direction]]] = []
+        update_id = self.runtime.next_update_id()
-        for s in st.snakes.values():
+        # Serialize full delta body once, then partition if large
-            dirs: List[Direction] = []
+        full_body = build_state_delta_body(
-            coords = list(s.body)
+            update_id=update_id,
-            for i in range(len(coords) - 1):
+            changes=changes,
-                x0, y0 = coords[i]
+            apples_added=apples_added,
-                x1, y1 = coords[i + 1]
+            apples_removed=apples_removed,
-                if x1 == x0 and y1 == y0 - 1:
-                    dirs.append(Direction.UP)
-                elif x1 == x0 + 1 and y1 == y0:
-                    dirs.append(Direction.RIGHT)
-                elif x1 == x0 and y1 == y0 + 1:
-                    dirs.append(Direction.DOWN)
-                elif x1 == x0 - 1 and y1 == y0:
-                    dirs.append(Direction.LEFT)
-            hx, hy = coords[0]
-            snakes_dirs.append((s.snake_id, s.length, hx, hy, dirs))
-        delta = build_state_full(
-            version=self.runtime.version,
-            seq=self.runtime.next_seq(),
-            tick=st.tick,
-            snakes=snakes_dirs,
-            apples=st.apples,
         )
-        for session in list(self.sessions.values()):
+        MTU = 1200  # soft limit for payload to avoid fragmentation
-            await self.transport.send(delta, TransportPeer(session.peer))
+        if len(full_body) <= MTU:
+            packet = pack_header(self.runtime.version, PacketType.STATE_DELTA, 0, self.runtime.next_seq(), st.tick & 0xFFFF) + full_body
+            for session in list(self.sessions.values()):
+                await self.transport.send(packet, TransportPeer(session.peer))
+        else:
+            # Partition by splitting snake changes across parts; include apples only in the first part
+            parts: List[bytes] = []
+            remaining = list(changes)
+            idx = 0
+            part_index = 0
+            while remaining:
+                chunk: List[SnakeDelta] = []
+                # greedy pack until size would exceed MTU
+                # start with apples only for first part
+                add_ap = apples_added if part_index == 0 else []
+                rem_ap = apples_removed if part_index == 0 else []
+                # Try adding changes one by one
+                for i, ch in enumerate(remaining):
+                    tmp_body = build_state_delta_body(update_id=update_id, changes=chunk + [ch], apples_added=add_ap, apples_removed=rem_ap)
+                    if len(tmp_body) > MTU and chunk:
+                        break
+                    if len(tmp_body) > MTU:
+                        # even a single change + apples doesn't fit; force single
+                        chunk.append(ch)
+                        i += 1
+                        break
+                    chunk.append(ch)
+                # Remove chunked items
+                remaining = remaining[len(chunk) :]
+                # Build chunk body
+                chunk_body = build_state_delta_body(update_id=update_id, changes=chunk, apples_added=add_ap, apples_removed=rem_ap)
+                parts.append(chunk_body)
+                part_index += 1
+            # Emit PART packets
+            total = len(parts)
+            for i, body in enumerate(parts):
+                pkt = build_part(
+                    version=self.runtime.version,
+                    seq=self.runtime.next_seq(),
+                    tick=st.tick & 0xFFFF,
+                    update_id=update_id,
+                    part_index=i,
+                    parts_total=total,
+                    inner_type=PacketType.STATE_DELTA,
+                    chunk_payload=body,
+                )
+                for session in list(self.sessions.values()):
+                    await self.transport.send(pkt, TransportPeer(session.peer))
 
     # --- Join / Spawn ---