1"""Wire protocol TITAN-X1: length-prefixed frames, HMAC-SHA256 authentication of every frame with a pre-shared token.23Frame = magic(4) | kind(1) | seq(4) | header_len(4) | payload_len(8) | header(JSON) | payload(bytes) | mac(32)4mac = HMAC(token, magic..payload). A worker only answers frames whose MAC verifies; a bad MAC closes the connection.5Kinds: 0 hello, 1 hello-ack, 2 exec request, 3 exec reply, 4 load request (install experts), 5 load reply, 6 ping, 7 pong, 9 error.6Activations are sent as raw float16 rows; results come back as raw float16 rows (one per (row, k) slot), so the coordinator7performs the weighted sum in exactly the local order (bit-for-bit identical aggregation).8"""9from __future__ import annotations10import hashlib, hmac, json, socket, struct1112MAGIC = b"TX1\0"13HDR = struct.Struct("<4sBIIQ")14KIND = {"hello": 0, "hello_ack": 1, "exec": 2, "exec_reply": 3, "load": 4, "load_reply": 5, "ping": 6, "pong": 7, "error": 9}15KIND_NAME = {v: k for k, v in KIND.items()}161718def _recv_exact(sock: socket.socket, n: int) -> bytes:19 buf = bytearray(n); view = memoryview(buf); got = 020 while got < n:21 r = sock.recv_into(view[got:], n - got)22 if r == 0:23 raise ConnectionError("peer closed")24 got += r25 return bytes(buf)262728class Framer:29 def __init__(self, sock: socket.socket, token: bytes):30 self.sock = sock31 self.token = token32 self.seq = 033 self.bytes_sent = 034 self.bytes_recv = 03536 def send(self, kind: str, header: dict, payload: bytes = b""):37 h = json.dumps(header, separators=(",", ":")).encode()38 head = HDR.pack(MAGIC, KIND[kind], self.seq, len(h), len(payload))39 mac = hmac.new(self.token, head + h + payload, hashlib.sha256).digest()40 self.sock.sendall(b"".join([head, h, payload, mac]))41 self.seq += 142 self.bytes_sent += HDR.size + len(h) + len(payload) + 324344 def recv(self):45 head = _recv_exact(self.sock, HDR.size)46 magic, kind, seq, hl, pl = HDR.unpack(head)47 if magic != MAGIC:48 raise ConnectionError("bad magic")49 h = _recv_exact(self.sock, hl)50 payload = _recv_exact(self.sock, pl)51 mac = _recv_exact(self.sock, 32)52 if not hmac.compare_digest(mac, hmac.new(self.token, head + h + payload, hashlib.sha256).digest()):53 raise ConnectionError("bad MAC")54 self.bytes_recv += HDR.size + hl + pl + 3255 return KIND_NAME[kind], json.loads(h), payload56