Server scaffold: protocol + config + transport abstraction + tick loop skeleton

- Protocol: PacketType, TLV Body types, QUIC varint, header, input_broadcast
  and config_update builders, 2-bit body bitpacking helper
- Config/model: live-config ServerConfig, basic GameState/Snake/Session
- Transport: InMemoryTransport placeholder and QUIC server stub
- Server: asyncio tick loop, periodic config_update broadcast, immediate
  input_broadcast relay; main entry and run.py

run.py

@@ -0,0 +1,10 @@
+from server.server import main
+
+if __name__ == "__main__":
+    import asyncio
+
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass
+

server/__init__.py

@@ -0,0 +1,2 @@
+__all__ = []
+

server/config.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class ServerConfig:
+    width: int = 60
+    height: int = 40
+    tick_rate: int = 10  # TPS, server-controlled, live-configurable
+    wrap_edges: bool = False  # default off
+    apples_per_snake: int = 1  # min 1, max 12
+    apples_cap: int = 255  # absolute cap
+    compression_mode: str = "none"  # "none" | "deflate" (handshake-only)
+    players_max: int = 32
+
+    def validate_runtime(self) -> None:
+        if not (3 <= self.width <= 255 and 3 <= self.height <= 255):
+            raise ValueError("field size must be within 3..255")
+        if not (5 <= self.tick_rate <= 30):
+            raise ValueError("tick_rate must be 5..30 TPS")
+        if not (1 <= self.apples_per_snake <= 12):
+            raise ValueError("apples_per_snake must be 1..12")
+        if not (0 <= self.apples_cap <= 255):
+            raise ValueError("apples_cap must be 0..255")
+        if self.compression_mode not in ("none", "deflate"):
+            raise ValueError("compression_mode must be 'none' or 'deflate'")
+

server/model.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Deque, Dict, List, Optional, Tuple
+from collections import deque
+
+from .protocol import Direction
+
+
+Coord = Tuple[int, int]
+
+
+@dataclass
+class Snake:
+    snake_id: int
+    head: Coord
+    direction: Direction
+    body: Deque[Coord] = field(default_factory=deque)  # includes head at index 0
+
+    @property
+    def length(self) -> int:
+        return len(self.body)
+
+
+@dataclass
+class PlayerSession:
+    player_id: int
+    name: str
+    color_id: int
+    peer: object  # transport-specific handle
+    input_seq: int = 0
+
+
+@dataclass
+class GameState:
+    width: int
+    height: int
+    snakes: Dict[int, Snake] = field(default_factory=dict)
+    apples: List[Coord] = field(default_factory=list)
+    tick: int = 0
+

server/protocol.py

@@ -0,0 +1,193 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import IntEnum
+from typing import Iterable, List, Sequence, Tuple
+
+
+class PacketType(IntEnum):
+    JOIN = 0
+    JOIN_ACK = 1
+    JOIN_DENY = 2
+    INPUT = 3
+    INPUT_BROADCAST = 4
+    STATE_DELTA = 5
+    STATE_FULL = 6
+    PART = 7
+    CONFIG_UPDATE = 8
+    PING = 9
+    PONG = 10
+    ERROR = 11
+
+
+class BodyTLV(IntEnum):
+    BODY_2BIT = 0x00
+    BODY_RLE = 0x01
+    BODY_2BIT_CHUNK = 0x10
+    BODY_RLE_CHUNK = 0x11
+
+
+class Direction(IntEnum):
+    UP = 0
+    RIGHT = 1
+    DOWN = 2
+    LEFT = 3
+
+
+def quic_varint_encode(value: int) -> bytes:
+    """Encode QUIC varint (RFC 9000)."""
+    if value < 0:
+        raise ValueError("varint must be non-negative")
+    if value <= 0x3F:  # 6 bits, 1 byte, 00xx xxxx
+        return bytes([value & 0x3F])
+    if value <= 0x3FFF:  # 14 bits, 2 bytes, 01xx xxxx
+        v = 0x4000 | value
+        return v.to_bytes(2, "big")
+    if value <= 0x3FFFFFFF:  # 30 bits, 4 bytes, 10xx xxxx
+        v = 0x80000000 | value
+        return v.to_bytes(4, "big")
+    if value <= 0x3FFFFFFFFFFFFFFF:  # 62 bits, 8 bytes, 11xx xxxx
+        v = 0xC000000000000000 | value
+        return v.to_bytes(8, "big")
+    raise ValueError("varint too large")
+
+
+def quic_varint_decode(buf: bytes, offset: int = 0) -> Tuple[int, int]:
+    """Decode QUIC varint starting at offset. Returns (value, next_offset)."""
+    first = buf[offset]
+    prefix = first >> 6
+    if prefix == 0:
+        return (first & 0x3F, offset + 1)
+    if prefix == 1:
+        v = int.from_bytes(buf[offset : offset + 2], "big") & 0x3FFF
+        return (v, offset + 2)
+    if prefix == 2:
+        v = int.from_bytes(buf[offset : offset + 4], "big") & 0x3FFFFFFF
+        return (v, offset + 4)
+    v = int.from_bytes(buf[offset : offset + 8], "big") & 0x3FFFFFFFFFFFFFFF
+    return (v, offset + 8)
+
+
+def pack_header(version: int, ptype: PacketType, flags: int, seq: int, tick: int | None) -> bytes:
+    """Pack common header fields.
+
+    Layout:
+    - ver: u8
+    - type: u8
+    - flags: u8
+    - seq: u16 (network order)
+    - tick: optional u16
+    """
+    if not (0 <= version <= 255):
+        raise ValueError("version out of range")
+    if not (0 <= flags <= 255):
+        raise ValueError("flags out of range")
+    if not (0 <= seq <= 0xFFFF):
+        raise ValueError("seq out of range")
+    parts = bytearray()
+    parts.append(version & 0xFF)
+    parts.append(int(ptype) & 0xFF)
+    parts.append(flags & 0xFF)
+    parts.extend(seq.to_bytes(2, "big"))
+    if tick is not None:
+        if not (0 <= tick <= 0xFFFF):
+            raise ValueError("tick out of range")
+        parts.extend(tick.to_bytes(2, "big"))
+    return bytes(parts)
+
+
+def unpack_header(buf: bytes, expect_tick: bool) -> Tuple[int, PacketType, int, int, int | None, int]:
+    """Unpack header; returns (ver, type, flags, seq, tick, next_offset)."""
+    if len(buf) < 5:
+        raise ValueError("buffer too small for header")
+    ver = buf[0]
+    ptype = PacketType(buf[1])
+    flags = buf[2]
+    seq = int.from_bytes(buf[3:5], "big")
+    off = 5
+    tick = None
+    if expect_tick:
+        if len(buf) < 7:
+            raise ValueError("buffer too small for tick")
+        tick = int.from_bytes(buf[5:7], "big")
+        off = 7
+    return ver, ptype, flags, seq, tick, off
+
+
+# Message builders (subset/skeleton)
+
+def build_config_update(
+    *,
+    version: int,
+    seq: int,
+    tick: int,
+    tick_rate: int,
+    wrap_edges: bool,
+    apples_per_snake: int,
+    apples_cap: int,
+) -> bytes:
+    header = pack_header(version, PacketType.CONFIG_UPDATE, 0, seq, tick)
+    body = bytearray()
+    body.append(tick_rate & 0xFF)  # u8
+    body.append(1 if wrap_edges else 0)  # bool u8
+    body.append(apples_per_snake & 0xFF)  # u8
+    body.append(apples_cap & 0xFF)  # u8
+    return header + bytes(body)
+
+
+@dataclass
+class InputEvent:
+    rel_tick_offset: int  # relative to base_tick
+    direction: Direction
+
+
+def build_input_broadcast(
+    *,
+    version: int,
+    seq: int,
+    tick: int,
+    player_id: int,
+    input_seq: int,
+    base_tick: int,
+    events: Sequence[InputEvent],
+    apply_at_tick: int | None = None,
+) -> bytes:
+    header = pack_header(version, PacketType.INPUT_BROADCAST, 0, seq, tick)
+    body = bytearray()
+    body.append(player_id & 0xFF)
+    body.extend(int(input_seq & 0xFFFF).to_bytes(2, "big"))
+    body.extend(int(base_tick & 0xFFFF).to_bytes(2, "big"))
+    # number of events as QUIC varint
+    body.extend(quic_varint_encode(len(events)))
+    for ev in events:
+        # rel offset as QUIC varint, direction as u8 (low 2 bits)
+        body.extend(quic_varint_encode(int(ev.rel_tick_offset)))
+        body.append(int(ev.direction) & 0x03)
+    # Optional absolute apply_at_tick presence flag + value
+    if apply_at_tick is None:
+        body.append(0)
+    else:
+        body.append(1)
+        body.extend(int(apply_at_tick & 0xFFFF).to_bytes(2, "big"))
+    return header + bytes(body)
+
+
+def pack_body_tlv(t: BodyTLV, payload: bytes) -> bytes:
+    return quic_varint_encode(int(t)) + quic_varint_encode(len(payload)) + payload
+
+
+def bitpack_2bit_directions(directions: Iterable[Direction]) -> bytes:
+    out = bytearray()
+    acc = 0
+    bits = 0
+    for d in directions:
+        acc |= (int(d) & 0x03) << bits
+        bits += 2
+        if bits >= 8:
+            out.append(acc & 0xFF)
+            acc = acc >> 8
+            bits -= 8
+    if bits:
+        out.append(acc & 0xFF)  # zero-padded high bits
+    return bytes(out)
+

server/server.py

@@ -0,0 +1,115 @@
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass
+from typing import Dict, List
+
+from .config import ServerConfig
+from .model import GameState, PlayerSession
+from .protocol import (
+    Direction,
+    InputEvent,
+    PacketType,
+    build_config_update,
+    build_input_broadcast,
+    pack_header,
+)
+from .transport import DatagramServerTransport, TransportPeer
+
+
+@dataclass
+class ServerRuntime:
+    config: ServerConfig
+    state: GameState
+    seq: int = 0
+    version: int = 1
+
+    def next_seq(self) -> int:
+        self.seq = (self.seq + 1) & 0xFFFF
+        return self.seq
+
+
+class GameServer:
+    def __init__(self, transport: DatagramServerTransport, config: ServerConfig):
+        self.transport = transport
+        self.runtime = ServerRuntime(config=config, state=GameState(config.width, config.height))
+        self.sessions: Dict[int, PlayerSession] = {}
+        self._config_update_interval_ticks = 50  # periodic resend
+
+    async def on_datagram(self, data: bytes, peer: TransportPeer) -> None:
+        # Skeleton: just ignore inbound for now; real impl would parse and dispatch
+        return
+
+    async def broadcast_config_update(self) -> None:
+        r = self.runtime
+        payload = build_config_update(
+            version=r.version,
+            seq=r.next_seq(),
+            tick=r.state.tick & 0xFFFF,
+            tick_rate=r.config.tick_rate,
+            wrap_edges=r.config.wrap_edges,
+            apples_per_snake=r.config.apples_per_snake,
+            apples_cap=r.config.apples_cap,
+        )
+        # Broadcast to all sessions (requires real peer handles)
+        for session in list(self.sessions.values()):
+            await self.transport.send(payload, TransportPeer(session.peer))
+
+    async def relay_input_broadcast(
+        self,
+        *,
+        from_player_id: int,
+        input_seq: int,
+        base_tick: int,
+        events: List[InputEvent],
+        apply_at_tick: int | None,
+    ) -> None:
+        r = self.runtime
+        payload = build_input_broadcast(
+            version=r.version,
+            seq=r.next_seq(),
+            tick=r.state.tick & 0xFFFF,
+            player_id=from_player_id,
+            input_seq=input_seq,
+            base_tick=base_tick & 0xFFFF,
+            events=events,
+            apply_at_tick=apply_at_tick,
+        )
+        for session in list(self.sessions.values()):
+            if session.player_id == from_player_id:
+                continue
+            await self.transport.send(payload, TransportPeer(session.peer))
+
+    async def tick_loop(self) -> None:
+        r = self.runtime
+        tick_duration = 1.0 / max(1, r.config.tick_rate)
+        next_cfg_resend = self._config_update_interval_ticks
+        while True:
+            start = asyncio.get_event_loop().time()
+            # TODO: process inputs, update snakes, collisions, apples, deltas
+            r.state.tick = (r.state.tick + 1) & 0xFFFFFFFF
+
+            next_cfg_resend -= 1
+            if next_cfg_resend <= 0:
+                await self.broadcast_config_update()
+                next_cfg_resend = self._config_update_interval_ticks
+
+            elapsed = asyncio.get_event_loop().time() - start
+            await asyncio.sleep(max(0.0, tick_duration - elapsed))
+
+
+async def main() -> None:
+    from .transport import InMemoryTransport
+
+    cfg = ServerConfig()
+    server = GameServer(transport=InMemoryTransport(lambda d, p: server.on_datagram(d, p)), config=cfg)
+    # In-memory transport never returns; run tick loop in parallel
+    await asyncio.gather(server.transport.run(), server.tick_loop())
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass
+

server/transport.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass
+from typing import Awaitable, Callable, Optional, Tuple
+
+
+OnDatagram = Callable[[bytes, object], Awaitable[None]]
+
+
+@dataclass
+class TransportPeer:
+    addr: object  # opaque peer handle (e.g., QUIC session)
+
+
+class DatagramServerTransport:
+    async def send(self, data: bytes, peer: TransportPeer) -> None:
+        raise NotImplementedError
+
+    async def run(self) -> None:
+        raise NotImplementedError
+
+
+class InMemoryTransport(DatagramServerTransport):
+    """A test transport that loops datagrams back to registered peers."""
+
+    def __init__(self, on_datagram: OnDatagram):
+        self._on_datagram = on_datagram
+        self._peers: list[TransportPeer] = []
+
+    def register_peer(self, peer: TransportPeer) -> None:
+        self._peers.append(peer)
+
+    async def send(self, data: bytes, peer: TransportPeer) -> None:
+        # In-memory: deliver to all except sender to simulate broadcast domain
+        for p in self._peers:
+            if p is peer:
+                continue
+            await self._on_datagram(data, p)
+
+    async def run(self) -> None:
+        # Nothing to do in in-memory test transport
+        await asyncio.Future()
+
+
+class QuicWebTransportServer(DatagramServerTransport):
+    """Placeholder for a real WebTransport (HTTP/3) datagram server.
+
+    Integrate with aioquic or another QUIC library and invoke the provided
+    on_datagram callback when a datagram arrives.
+    """
+
+    def __init__(self, on_datagram: OnDatagram):
+        self._on_datagram = on_datagram
+
+    async def send(self, data: bytes, peer: TransportPeer) -> None:
+        raise NotImplementedError("QUIC server not implemented in skeleton")
+
+    async def run(self) -> None:
+        raise NotImplementedError("QUIC server not implemented in skeleton")
+

server/utils.py

@@ -0,0 +1,10 @@
+from __future__ import annotations
+
+
+def is_newer(a: int, b: int) -> bool:
+    """Return True if 16-bit sequence number a is newer than b (wrap-aware).
+
+    Uses half-range window on unsigned 16-bit arithmetic.
+    """
+    return ((a - b) & 0xFFFF) < 0x8000
+