D1 phase 1: relay TCP core in btconsole.py (--relay mode)

The console emulator gains a relay/dispatcher mode for internet play (D1): pods dial OUT to this process (NAT-friendly) instead of the console dialing into each pod. Legacy dial-out mode is untouched (verified: an unmodified 2-node mesh session still forms end-to-end through it). --relay <consolePort> <egg> [--bind ADDR] [--udp-drop PCT], single selectors-based event loop: - console listener (consolePort): streams the NUL-line egg to each pod on connect; GLOBAL launch timer -- RunMission x2 to ALL pods at +20s/+4s after the LAST pod has its egg (fixes the per-pod skew of the legacy tool). - game TCP listener (consolePort+1): envelope router {route i32, length u32}. First frame must be HELLO ('BTR1' + hostID validated against the egg [pilots] roster); PEER_UP/PEER_DOWN exchange; route>=2 unicast (route rewritten to sender), -1 broadcast-except-sender (the client sends each broadcast ONCE -- kills the N-1x upload duplication). Frame cap 1600 (NETWORKMANAGER_BUFFER_SIZE); protocol violations drop the connection. - game UDP socket (consolePort+1/udp): endpoint learned per-datagram (NAT-rebind tolerant), HELLO->HELLO-ACK, verbatim forward by route, TCP-wrap fallback when the target's UDP endpoint is unknown; --udp-drop test hook for the robustness phase. Verified: scratchpad/relay_stub_test.py (self-contained; spawns the relay + stub clients) -- 24/24 checks pass: egg delivery, registration, PEER events, unicast route-rewrite, broadcast fan-out (sender excluded), UDP ack/forward/ fallback, and all reject paths (bad magic, dup id, out-of-roster, oversize). Plan: ~/.claude/plans/partitioned-snuggling-piglet.md (D1 relay + UDP). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-18 08:44:01 -05:00
parent e2c21c4db2
commit bf8c2ffaae
2 changed files with 637 additions and 1 deletions
@@ -28,11 +28,38 @@ AcknowledgeEggFile back on this socket.  The console must STAY CONNECTED (a
 disconnect trips the pod's console-loss path, which also closes the game
 listener -- an engine bug).

-Usage:
+Usage (legacy dial-out console -- the pod-authentic direction):
  python btconsole.py <egg-file> <host:port> [<host:port> ...]
 Example (two instances on one box, -net 1501 / -net 1601):
  python btconsole.py MP.EGG 127.0.0.1:1501 127.0.0.1:1601
+
+RELAY MODE (D1 -- internet play; pods dial OUT to this process):
+  python btconsole.py --relay <consolePort> <egg-file> [--bind ADDR] [--udp-drop PCT]
+Example:
+  python btconsole.py --relay 1500 MP_RELAY.EGG
+The relay listens on <consolePort>/tcp (console protocol: streams the egg to each
+pod that connects, sends the LAUNCH pair once ALL pods have the egg), on
+<consolePort>+1/tcp (the GAME relay: envelope-framed frame router, see
+RelayEnvelope below), and on <consolePort>+1/udp (the unreliable channel
+forwarder).  Pods run with BT_RELAY=<host>:<consolePort> and a unique
+BT_SELF=<their [pilots] entry>.
+
+  RelayEnvelope (game TCP): { int32 route; uint32 length; } + length payload bytes
+    route >= 2 : client->relay: unicast to that hostID
+                 relay->client: the sender's hostID (frames also self-identify
+                 via NetworkPacketHeader.fromHost)
+    route == -1: broadcast to every registered pod except the sender
+    route == -2: HELLO   (payload: int32 magic 'BTR1', int32 myHostID)
+    route == -3: PEER_UP   (payload: int32 hostID)  relay->client only
+    route == -4: PEER_DOWN (payload: int32 hostID)  relay->client only
+  UDP envelope: { int32 route; int32 fromHost; uint32 seq; } + frame
+    route as above; -2 = HELLO/keepalive (no payload) -> relay answers -5
+    (HELLO-ACK).  The relay learns each pod's public endpoint from every
+    inbound datagram and forwards verbatim; unknown target endpoint falls
+    back to wrapping the frame onto the target's game TCP connection.
 """
+import random
+import selectors
 import socket
 import struct
 import sys
@@ -144,7 +171,415 @@ def serve_pod(hostport, egg_bytes):
                time.sleep(60)


+#############################################################################
+# RELAY MODE (D1)
+#############################################################################
+
+ENV_FMT_TCP = "<iI"            # route, length
+ENV_FMT_UDP = "<iiI"           # route, fromHost, seq
+ENV_TCP_SIZE = 8
+ENV_UDP_SIZE = 12
+ROUTE_BCAST = -1
+ROUTE_HELLO = -2
+ROUTE_PEER_UP = -3
+ROUTE_PEER_DOWN = -4
+ROUTE_UDP_ACK = -5
+HELLO_MAGIC = 0x31525442       # 'BTR1' little-endian
+FRAME_CAP = 1600               # NETWORKMANAGER_BUFFER_SIZE (NETWORK.h:89)
+FIRST_GAME_HOST_ID = 2         # FirstLegalHostID(1) == the console; pods 2..N+1
+STATS_PERIOD = 10.0
+
+
+def parse_egg_roster(egg_path):
+    """Count the [pilots] pilot= entries -> expected hostIDs 2..N+1 in egg order.
+    The relay's ONLY game knowledge; it never parses game frames."""
+    entries = []
+    in_pilots = False
+    for raw in open(egg_path, "rb").read().replace(b"\r\n", b"\n").split(b"\n"):
+        line = raw.strip()
+        if line.startswith(b"["):
+            in_pilots = (line.lower() == b"[pilots]")
+            continue
+        if in_pilots and line.lower().startswith(b"pilot="):
+            entries.append(line.split(b"=", 1)[1].decode("ascii", "replace"))
+    return entries
+
+
+class RelayGameConn:
+    """One accepted game-TCP connection: ACCEPTED -> (HELLO) REGISTERED -> GONE."""
+    def __init__(self, sock, addr):
+        self.sock = sock
+        self.addr = addr
+        self.buf = b""
+        self.host_id = None            # None until HELLO registers it
+
+    def name(self):
+        hid = self.host_id if self.host_id is not None else "?"
+        return f"game[{self.addr[0]}:{self.addr[1]} host={hid}]"
+
+
+class RelayConsoleConn:
+    """One accepted console connection (legacy raw protocol, relay-terminated)."""
+    def __init__(self, sock, addr):
+        self.sock = sock
+        self.addr = addr
+        self.egg_sent = False
+
+
+class Relay:
+    def __init__(self, console_port, egg_path, bind_addr, udp_drop_pct):
+        self.console_port = console_port
+        self.game_port = console_port + 1
+        self.bind_addr = bind_addr
+        self.udp_drop_pct = udp_drop_pct
+        self.egg_bytes = egg_wire_bytes(open(egg_path, "rb").read())
+        self.roster = parse_egg_roster(egg_path)
+        self.expected_ids = set(range(FIRST_GAME_HOST_ID,
+                                      FIRST_GAME_HOST_ID + len(self.roster)))
+        self.sel = selectors.DefaultSelector()
+        self.console_conns = []
+        self.game_conns = []                     # RelayGameConn (incl. unregistered)
+        self.by_host = {}                        # hostID -> RelayGameConn
+        self.udp_endpoint = {}                   # hostID -> (addr, port)
+        self.udp_sock = None
+        self.launch_at = None
+        self.launches_sent = 0
+        self.stats = {"tcp_rx": 0, "tcp_tx": 0, "udp_rx": 0, "udp_tx": 0,
+                      "udp_dropped": 0, "udp_tcp_fallback": 0}
+        self.stats_at = time.time() + STATS_PERIOD
+
+    # ---------------- lifecycle ----------------
+
+    def run(self):
+        print(f"[relay] roster: {len(self.roster)} pilot(s) -> hostIDs "
+              f"{sorted(self.expected_ids)}: {self.roster}", flush=True)
+        con_l = self._listener(self.console_port)
+        game_l = self._listener(self.game_port)
+        self.udp_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        self.udp_sock.bind((self.bind_addr, self.game_port))
+        self.udp_sock.setblocking(False)
+        self.sel.register(con_l, selectors.EVENT_READ, ("con_listen", None))
+        self.sel.register(game_l, selectors.EVENT_READ, ("game_listen", None))
+        self.sel.register(self.udp_sock, selectors.EVENT_READ, ("udp", None))
+        print(f"[relay] console {self.bind_addr}:{self.console_port} | game tcp/udp "
+              f"{self.bind_addr}:{self.game_port} | udp-drop {self.udp_drop_pct}%",
+              flush=True)
+        while True:
+            for key, _ in self.sel.select(timeout=0.5):
+                kind, obj = key.data
+                if kind == "con_listen":
+                    self._accept_console(key.fileobj)
+                elif kind == "game_listen":
+                    self._accept_game(key.fileobj)
+                elif kind == "console":
+                    self._console_read(obj)
+                elif kind == "game":
+                    self._game_read(obj)
+                elif kind == "udp":
+                    self._udp_read()
+            self._tick_launch()
+            self._tick_stats()
+
+    def _listener(self, port):
+        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        s.bind((self.bind_addr, port))
+        s.listen(16)
+        s.setblocking(False)
+        return s
+
+    # ---------------- console side (legacy protocol, relay-terminated) ----------------
+
+    def _accept_console(self, listener):
+        sock, addr = listener.accept()
+        sock.setblocking(False)
+        conn = RelayConsoleConn(sock, addr)
+        self.console_conns.append(conn)
+        self.sel.register(sock, selectors.EVENT_READ, ("console", conn))
+        # Stream the egg immediately (blocking sendall is fine: the pod's console
+        # pad drains fast and the egg is a few KB).
+        sock.setblocking(True)
+        try:
+            n = 0
+            for pkt in egg_packets(self.egg_bytes):
+                sock.sendall(pkt)
+                n += 1
+            conn.egg_sent = True
+            print(f"[relay] console conn {addr[0]}:{addr[1]}: egg sent "
+                  f"({n} chunks); {self._eggs_out()}/{len(self.roster)} pods have it",
+                  flush=True)
+        except OSError as e:
+            print(f"[relay] console conn {addr[0]}:{addr[1]}: egg send failed {e!r}",
+                  flush=True)
+            self._drop_console(conn)
+            return
+        finally:
+            try:
+                sock.setblocking(False)
+            except OSError:
+                pass
+        # GLOBAL launch timer: arm/rearm when the LAST pod has its egg.
+        if self._eggs_out() >= len(self.roster) and self.launches_sent == 0:
+            self.launch_at = time.time() + LAUNCH_SETTLE_SECONDS
+            print(f"[relay] all pods have the egg; LAUNCH pair in "
+                  f"{LAUNCH_SETTLE_SECONDS}s", flush=True)
+
+    def _eggs_out(self):
+        return sum(1 for c in self.console_conns if c.egg_sent)
+
+    def _console_read(self, conn):
+        try:
+            data = conn.sock.recv(4096)
+        except OSError:
+            data = b""
+        if not data:
+            print(f"[relay] console conn {conn.addr[0]}:{conn.addr[1]} closed",
+                  flush=True)
+            self._drop_console(conn)
+            return
+        if len(data) >= 24:
+            (clid, _gid, fh, _ts) = struct.unpack_from("<iiii", data, 0)
+            (mlen, mid) = struct.unpack_from("<Ii", data, 16)
+            print(f"[relay] pod->console clientID={clid} fromHost={fh} msgID={mid} "
+                  f"len={mlen}", flush=True)
+
+    def _drop_console(self, conn):
+        try:
+            self.sel.unregister(conn.sock)
+        except (KeyError, ValueError):
+            pass
+        try:
+            conn.sock.close()
+        except OSError:
+            pass
+        if conn in self.console_conns:
+            self.console_conns.remove(conn)
+
+    def _tick_launch(self):
+        if self.launch_at is None or self.launches_sent >= 2:
+            return
+        if time.time() < self.launch_at:
+            return
+        pkt = run_mission_packet()
+        for conn in list(self.console_conns):
+            try:
+                conn.sock.setblocking(True)
+                conn.sock.sendall(pkt)
+                conn.sock.setblocking(False)
+            except OSError as e:
+                print(f"[relay] launch send failed to {conn.addr}: {e!r}", flush=True)
+        self.launches_sent += 1
+        self.launch_at = time.time() + LAUNCH_STEP_SECONDS
+        print(f"[relay] RunMission #{self.launches_sent} sent to "
+              f"{len(self.console_conns)} pod(s)", flush=True)
+
+    # ---------------- game TCP side ----------------
+
+    def _accept_game(self, listener):
+        sock, addr = listener.accept()
+        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+        sock.setblocking(False)
+        conn = RelayGameConn(sock, addr)
+        self.game_conns.append(conn)
+        self.sel.register(sock, selectors.EVENT_READ, ("game", conn))
+        print(f"[relay] {conn.name()} accepted (awaiting HELLO)", flush=True)
+
+    def _game_read(self, conn):
+        try:
+            data = conn.sock.recv(8192)
+        except OSError:
+            data = b""
+        if not data:
+            self._drop_game(conn, "closed")
+            return
+        conn.buf += data
+        while len(conn.buf) >= ENV_TCP_SIZE:
+            route, length = struct.unpack_from(ENV_FMT_TCP, conn.buf, 0)
+            if length > FRAME_CAP:
+                self._drop_game(conn, f"oversize frame ({length})")
+                return
+            if len(conn.buf) < ENV_TCP_SIZE + length:
+                break                                   # partial; wait for more
+            payload = conn.buf[ENV_TCP_SIZE:ENV_TCP_SIZE + length]
+            conn.buf = conn.buf[ENV_TCP_SIZE + length:]
+            self.stats["tcp_rx"] += 1
+            self._handle_game_frame(conn, route, payload)
+            if conn.sock.fileno() < 0:                  # dropped mid-loop
+                return
+
+    def _handle_game_frame(self, conn, route, payload):
+        if conn.host_id is None:
+            # First frame MUST be a valid HELLO.
+            if route != ROUTE_HELLO or len(payload) < 8:
+                self._drop_game(conn, f"first frame not HELLO (route={route})")
+                return
+            magic, host_id = struct.unpack_from("<iI", payload, 0)
+            host_id = int(host_id)
+            if magic != HELLO_MAGIC:
+                self._drop_game(conn, f"bad HELLO magic 0x{magic:x}")
+                return
+            if host_id not in self.expected_ids:
+                self._drop_game(conn, f"HELLO hostID {host_id} not in roster "
+                                      f"{sorted(self.expected_ids)}")
+                return
+            if host_id in self.by_host:
+                self._drop_game(conn, f"HELLO hostID {host_id} already registered")
+                return
+            conn.host_id = host_id
+            self.by_host[host_id] = conn
+            print(f"[relay] {conn.name()} REGISTERED "
+                  f"({len(self.by_host)}/{len(self.roster)})", flush=True)
+            # PEER_UP exchange: newcomer learns everyone; everyone learns newcomer.
+            for other_id in sorted(self.by_host):
+                if other_id != host_id:
+                    self._send_control(conn, ROUTE_PEER_UP, other_id)
+            for other in self.by_host.values():
+                if other is not conn:
+                    self._send_control(other, ROUTE_PEER_UP, host_id)
+            return
+        if route == ROUTE_BCAST:
+            env = struct.pack(ENV_FMT_TCP, conn.host_id, len(payload))
+            for other in list(self.by_host.values()):
+                if other is not conn:
+                    self._send_raw(other, env + payload)
+        elif route >= FIRST_GAME_HOST_ID:
+            target = self.by_host.get(route)
+            if target is None:
+                return                                  # peer gone; drop silently
+            env = struct.pack(ENV_FMT_TCP, conn.host_id, len(payload))
+            self._send_raw(target, env + payload)
+        else:
+            print(f"[relay] {conn.name()} bad route {route} -- ignored", flush=True)
+
+    def _send_control(self, conn, route, host_id):
+        self._send_raw(conn, struct.pack(ENV_FMT_TCP, route, 4)
+                       + struct.pack("<i", host_id))
+
+    def _send_raw(self, conn, data):
+        try:
+            conn.sock.setblocking(True)
+            conn.sock.sendall(data)
+            conn.sock.setblocking(False)
+            self.stats["tcp_tx"] += 1
+        except OSError as e:
+            self._drop_game(conn, f"send failed {e!r}")
+
+    def _drop_game(self, conn, why):
+        print(f"[relay] {conn.name()} dropped: {why}", flush=True)
+        try:
+            self.sel.unregister(conn.sock)
+        except (KeyError, ValueError):
+            pass
+        try:
+            conn.sock.close()
+        except OSError:
+            pass
+        if conn in self.game_conns:
+            self.game_conns.remove(conn)
+        if conn.host_id is not None and self.by_host.get(conn.host_id) is conn:
+            del self.by_host[conn.host_id]
+            self.udp_endpoint.pop(conn.host_id, None)
+            for other in self.by_host.values():
+                self._send_control(other, ROUTE_PEER_DOWN, conn.host_id)
+
+    # ---------------- game UDP side ----------------
+
+    def _udp_read(self):
+        while True:
+            try:
+                data, endpoint = self.udp_sock.recvfrom(4096)
+            except (BlockingIOError, InterruptedError):
+                return
+            except OSError:
+                return
+            if len(data) < ENV_UDP_SIZE:
+                continue
+            route, from_host, _seq = struct.unpack_from(ENV_FMT_UDP, data, 0)
+            if from_host not in self.by_host:
+                continue                                # not TCP-registered: drop
+            self.stats["udp_rx"] += 1
+            # NAT-rebind tolerant: every datagram refreshes the endpoint map.
+            self.udp_endpoint[from_host] = endpoint
+            if route == ROUTE_HELLO:
+                ack = struct.pack(ENV_FMT_UDP, ROUTE_UDP_ACK, CONSOLE_HOST_ID, 0)
+                try:
+                    self.udp_sock.sendto(ack, endpoint)
+                except OSError:
+                    pass
+                continue
+            if route == ROUTE_BCAST:
+                targets = [h for h in self.by_host if h != from_host]
+            elif route >= FIRST_GAME_HOST_ID and route in self.by_host:
+                targets = [route]
+            else:
+                continue
+            for host_id in targets:
+                if self.udp_drop_pct and random.random() * 100.0 < self.udp_drop_pct:
+                    self.stats["udp_dropped"] += 1
+                    continue
+                ep = self.udp_endpoint.get(host_id)
+                if ep is not None:
+                    try:
+                        self.udp_sock.sendto(data, ep)      # forward VERBATIM
+                        self.stats["udp_tx"] += 1
+                        continue
+                    except OSError:
+                        pass
+                # Target's UDP endpoint unknown/broken: wrap the frame onto its
+                # game TCP connection (asymmetric-UDP-blockage fallback).
+                frame = data[ENV_UDP_SIZE:]
+                target = self.by_host.get(host_id)
+                if target is not None and frame:
+                    env = struct.pack(ENV_FMT_TCP, from_host, len(frame))
+                    self._send_raw(target, env + frame)
+                    self.stats["udp_tcp_fallback"] += 1
+
+    # ---------------- stats ----------------
+
+    def _tick_stats(self):
+        if time.time() < self.stats_at:
+            return
+        self.stats_at = time.time() + STATS_PERIOD
+        s = self.stats
+        print(f"[relay-stats] tcp rx/tx {s['tcp_rx']}/{s['tcp_tx']} | "
+              f"udp rx/tx {s['udp_rx']}/{s['udp_tx']} "
+              f"(dropped {s['udp_dropped']}, tcp-fallback {s['udp_tcp_fallback']}) | "
+              f"registered {sorted(self.by_host)} udp-known {sorted(self.udp_endpoint)}",
+              flush=True)
+
+
+def relay_main(argv):
+    """argv: --relay <consolePort> <egg-file> [--bind ADDR] [--udp-drop PCT]"""
+    args = argv[:]
+    bind_addr = "0.0.0.0"
+    udp_drop = 0.0
+    if "--bind" in args:
+        i = args.index("--bind")
+        bind_addr = args[i + 1]
+        del args[i:i + 2]
+    if "--udp-drop" in args:
+        i = args.index("--udp-drop")
+        udp_drop = float(args[i + 1])
+        del args[i:i + 2]
+    if len(args) != 2:
+        print(__doc__)
+        return 2
+    console_port = int(args[0])
+    egg_path = args[1]
+    relay = Relay(console_port, egg_path, bind_addr, udp_drop)
+    if not relay.roster:
+        print(f"[relay] ERROR: no [pilots] entries in {egg_path}")
+        return 2
+    try:
+        relay.run()
+    except KeyboardInterrupt:
+        print("[relay] shutting down")
+    return 0
+
+
 def main():
+    if len(sys.argv) >= 2 and sys.argv[1] == "--relay":
+        return relay_main(sys.argv[2:])
    if len(sys.argv) < 3:
        print(__doc__)
        return 2