#!/usr/bin/env python
"""btconsole.py -- minimal Tesla/WinTesla CONSOLE emulator (the operator station).

The pod-side engine (MUNGA_L4/L4NET.CPP) boots `-net <port>` into ConsoleOnly state:
it listens on TCP <port> and waits for a console to connect and stream the mission
egg.  The real console software is absent from every archive, so this tool speaks
the console's wire protocol:

  packet := NetworkPacketHeader + NetworkManager::ReceiveEggFileMessage
    off  0  int32  clientID      = 0    (NetworkClient::NetworkManagerClientID)
    off  4  int32  gameID        = 0    (NETWORK.cpp:121 -- always 0)
    off  8  int32  fromHost      = 1    (FirstLegalHostID == the console)
    off 12  int32  timeStamp     = 0    (Time; unused by the egg handler)
    off 16  uint32 messageLength = 1024 (sizeof(ReceiveEggFileMessage))
    off 20  int32  messageID     = 3    (NetworkManager::ReceiveEggFileMessageID)
    off 24  uint32 messageFlags  = 1    (Receiver::Message::ReliableFlag)
    off 28  int32  sequenceNumber      (0..n-1; -1 is the solo-local path only)
    off 32  int32  notationFileLength  (total egg bytes)
    off 36  int32  thisMessageLength   (bytes used in this chunk, <= 1000)
    off 40  char   notationData[1000]
    total 1040 bytes per chunk
  (constants verified from the port build via `btl4.exe` BT_NET_PROBE=1)

The pod reassembles chunks sequentially (L4NET.CPP:773-816: seq 0 allocates,
append until notationFileLength reached), sets NormalState, CreateMission ->
StartConnecting (the egg's [pilots] list forms the pod mesh), and sends an
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 (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 collections
import os
import random
import selectors
# player callsign/mech seat requests mutate the session egg via eggmodel;
# import guarded so a stripped install still relays with roster defaults.
try:
    import eggmodel
except ImportError:
    eggmodel = None
import socket
import struct
import sys
import threading
import time

CHUNK = 1000

# LIVENESS (2026-07-26, operator report "I have to reboot the console").  Nothing
# on the relay ever noticed a pod that vanished WITHOUT closing its TCP -- a
# sleeping laptop, a dropped Wi-Fi link, a killed process, a NAT idle-timeout on
# the internet relay.  There was no SO_KEEPALIVE and no idle deadline anywhere,
# so the dead connection stayed in console_conns with acked=True forever, which
# (a) permanently blocked the round reset (it requires NO conn acked) and (b) held
# a phantom seat in the launch count.  Only restarting the relay cleared it.
KEEPALIVE_IDLE_SECONDS = 30    # start probing after this much silence
KEEPALIVE_INTERVAL_SECONDS = 10
DEAD_PEER_SECONDS = 180.0      # app-level deadline: no bytes at all for this long
SEND_TIMEOUT_SECONDS = 10.0    # a blocking sendall may not stall the whole relay


def _enable_keepalive(sock):
    """Turn on TCP keepalive so the OS reaps half-open peers for us.  Best
    effort: the tunables are platform-specific and a plain SO_KEEPALIVE is still
    a large improvement over nothing."""
    try:
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
    except OSError:
        return
    try:                                    # Windows: one ioctl sets both
        sock.ioctl(socket.SIO_KEEPALIVE_VALS,
                   (1, KEEPALIVE_IDLE_SECONDS * 1000,
                    KEEPALIVE_INTERVAL_SECONDS * 1000))
    except (AttributeError, OSError, ValueError):
        for opt, val in (("TCP_KEEPIDLE", KEEPALIVE_IDLE_SECONDS),
                         ("TCP_KEEPINTVL", KEEPALIVE_INTERVAL_SECONDS),
                         ("TCP_KEEPCNT", 4)):
            code = getattr(socket, opt, None)
            if code is not None:
                try:
                    sock.setsockopt(socket.IPPROTO_TCP, code, val)
                except OSError:
                    pass


def _send_all_guarded(sock, payload, what=""):
    """Blocking sendall with a TIMEOUT.  The relay is single-threaded: every
    pod-facing send used to flip the socket to blocking with no timeout, so one
    wedged peer could freeze the entire relay -- no launch tick, no UDP fan-out,
    no operator control channel -- for as long as the OS kept retransmitting.
    Raises OSError on failure/timeout; callers already handle that by dropping.
    """
    sock.settimeout(SEND_TIMEOUT_SECONDS)
    try:
        sock.sendall(payload)
    finally:
        try:
            sock.setblocking(False)
        except OSError:
            pass


PKT_FMT_HDR = "<iiii"          # clientID, gameID, fromHost, timeStamp
PKT_FMT_MSG = "<IiIiii"        # messageLength, messageID, messageFlags, seq, fileLen, thisLen
MESSAGE_LENGTH = 1024          # sizeof(ReceiveEggFileMessage)
EGG_MESSAGE_ID = 3             # NetworkManager::ReceiveEggFileMessageID
RELIABLE_FLAG = 1
CONSOLE_HOST_ID = 1            # FirstLegalHostID

# The LAUNCH: pods load to WaitingForLaunch and sit until the console sends
# Application::RunMissionMessage (the operator's launch button).  The handler
# ladder needs it twice: WaitingForLaunch->LaunchingMission (dispatches the
# player MissionStarting/translocation), then LaunchingMission->RunningMission.
# Sending it early (state==CreatingMission) hits the handler's Fail() -- hence
# the settle delay.  Constants from BT_NET_PROBE=1.
APPLICATION_CLIENT_ID = 4      # NetworkClient::ApplicationClientID
RUN_MISSION_MESSAGE_ID = 5     # Application::RunMissionMessageID
STOP_MISSION_MESSAGE_ID = 6    # Application::StopMissionMessageID (APP.h:383)
LAUNCH_SETTLE_SECONDS = 20.0   # egg -> first launch (pods must reach WaitingForLaunch)
LAUNCH_STEP_SECONDS = 4.0      # first launch -> second (Launching -> Running)
STOP_GRACE_SECONDS = 2.0       # past the pods' own zero so the ranking shows


def run_mission_packet():
    pkt = struct.pack(PKT_FMT_HDR, APPLICATION_CLIENT_ID, 0, CONSOLE_HOST_ID, 0)
    pkt += struct.pack("<IiI", 12, RUN_MISSION_MESSAGE_ID, RELIABLE_FLAG)
    assert len(pkt) == 16 + 12
    return pkt


def stop_mission_packet():
    # Application::StopMissionMessage {exitCode=NullExitCodeID} (id 6,
    # APPMSG.cpp:21).  The 1995 console ENDED the timed mission -- the pods
    # only display the countdown (secondsRemainingInGame, APP.cpp:652) and
    # pop the ranking window for the final 30s (DIRECTOR.cpp:113); the stop
    # itself was always console-side, which is us.
    pkt = struct.pack(PKT_FMT_HDR, APPLICATION_CLIENT_ID, 0, CONSOLE_HOST_ID, 0)
    pkt += struct.pack("<IiIi", 16, STOP_MISSION_MESSAGE_ID, RELIABLE_FLAG, 0)
    assert len(pkt) == 16 + 16
    return pkt


def parse_egg_mission_length(egg_path):
    """[mission] length= in seconds (0 = untimed)."""
    section = None
    try:
        for raw in open(egg_path, "r", encoding="latin-1"):
            line = raw.strip()
            if line.startswith("[") and line.endswith("]"):
                section = line[1:-1].lower()
            elif section == "mission" and "=" in line:
                key, _, value = line.partition("=")
                if key.strip().lower() == "length":
                    return float(value.strip())
    except (OSError, ValueError):
        pass
    return 0.0


def egg_wire_bytes(egg_text_bytes):
    """Convert raw egg TEXT to the wire form NotationFile::ReadText expects:
    NUL-separated lines (NOTATION.cpp:1043 walks `strchr(buffer,'\\0')+1` per
    line).  The real 1996 console pre-processed the egg the same way; sending
    the raw file text makes the whole egg parse as one garbage line ("ERROR:
    no map in egg!")."""
    lines = egg_text_bytes.replace(b"\r\n", b"\n").split(b"\n")
    return b"".join(line + b"\0" for line in lines)


def egg_packets(egg_bytes):
    total = len(egg_bytes)
    seq = 0
    off = 0
    while off < total:
        chunk = egg_bytes[off:off + CHUNK]
        pkt = struct.pack(PKT_FMT_HDR, 0, 0, CONSOLE_HOST_ID, 0)
        pkt += struct.pack(PKT_FMT_MSG, MESSAGE_LENGTH, EGG_MESSAGE_ID, RELIABLE_FLAG,
                           seq, total, len(chunk))
        pkt += chunk.ljust(CHUNK, b"\0")
        assert len(pkt) == 16 + MESSAGE_LENGTH
        yield pkt
        off += len(chunk)
        seq += 1


def serve_pod(hostport, egg_bytes):
    host, port = hostport.rsplit(":", 1)
    port = int(port)
    name = f"{host}:{port}"
    # The pod may not be listening yet -- retry like the pods do to each other.
    while True:
        try:
            s = socket.create_connection((host, port), timeout=5)
            break
        except OSError as e:
            print(f"[{name}] waiting for pod ({e})")
            time.sleep(1)
    print(f"[{name}] connected; streaming egg ({len(egg_bytes)} bytes)")
    n = 0
    for pkt in egg_packets(egg_bytes):
        s.sendall(pkt)
        n += 1
    print(f"[{name}] egg sent in {n} chunk(s); launch in {LAUNCH_SETTLE_SECONDS}s")
    s.settimeout(2.0)
    launch_at = time.time() + LAUNCH_SETTLE_SECONDS
    launches_sent = 0
    while True:
        if launches_sent < 2 and time.time() >= launch_at:
            s.sendall(run_mission_packet())
            launches_sent += 1
            launch_at = time.time() + LAUNCH_STEP_SECONDS
            print(f"[{name}] RunMission #{launches_sent} sent")
        try:
            data = s.recv(4096)
            if not data:
                print(f"[{name}] pod closed the console socket", flush=True)
                return
            # Expect the AcknowledgeEggFile (and possibly other console traffic).
            if len(data) >= 24:
                (clid, gid, fh, ts) = struct.unpack_from("<iiii", data, 0)
                (mlen, mid) = struct.unpack_from("<Ii", data, 16)
                print(f"[{name}] pod->console packet: clientID={clid} fromHost={fh} "
                      f"msgID={mid} len={mlen} ({len(data)} bytes)")
            else:
                print(f"[{name}] pod->console {len(data)} bytes")
        except socket.timeout:
            continue
        except OSError as e:
            # Previously UNHANDLED: a reset/abort here killed the thread silently
            # (buffered stdout lost) and the console process exited -- leaving the
            # pods holding dead console sockets.  Log it and keep the thread alive
            # (idle) so the process + diagnosis survive.
            print(f"[{name}] console socket error: {e!r} -- idling", flush=True)
            while True:
                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
ROUTE_SEAT_REQUEST = -6   # client->relay: assign me a free roster seat
ROUTE_SEAT_ASSIGN = -7    # relay->client: int32 hostID + NUL-terminated tag
ROUTE_SEAT_FULL = -8      # relay->client: no free seats
ROUTE_MATCHLOG = -9       # client->relay: matchlog upload (filename NUL + bytes)
ROUTE_READY = -10         # client->relay: mission load complete (READY light)
ROUTE_REJOIN = -11        # relay->client: abandon the round, rejoin (peer died mid-load)
# the 18 certified mech tags (mirror of the FE kVehicles catalog)
VEHICLE_TAGS = {"blkhawk", "loki", "bhk1", "madcat", "thor", "owens",
                "own1", "thr1", "lok1", "mad1", "avatar", "ava1",
                "sunder", "snd1", "vulture", "vul1", "lok2", "mad2"}
MATCHLOG_CAP = 8 * 1024 * 1024   # matchlog upload frame cap (client caps at 8MB)
SEAT_RESERVE_SECONDS = 60.0
ABORT_SETTLE_SECONDS = 8.0     # issue #33: egg-release hold after a round abort

# issue #38: the vehicle table's CAMO colors (a color= outside this set
# paints the mech silent-gray -- 'Red' is a PATCH color, the camo is
# 'Crimson'; caught live when a test egg authored color=Red).
VALID_CAMO_COLORS = {"Black", "Brown", "Crimson", "Green", "Grey", "Tan",
                     "White"}

# REMOTE OPERATOR CONTROL (2026-07-24): a TCP line protocol so an operator
# GUI on ANOTHER machine can drive this relay (only the relay host needs
# port forwarding).  Additive: stdin commands keep working unchanged.
#   client -> relay:  AUTH <secret> | launch | stop | ping |
#                     get mission | set key=value;key=value
#   relay  -> client: "OK ..." on auth, then the relay's own timestamped
#                     log stream (the GUI already parses those lines).
# One authenticated operator at a time -- a new AUTH displaces the old.
# A slow/stalled operator NEVER blocks the relay: writes are buffered per
# connection and the connection is dropped if the buffer overflows.
# Console-protocol reassembly bounds (see _console_read).  A pod message is a
# 16-byte header + messageLength bytes; the egg chunk is the biggest thing the
# protocol carries at 16+1024, so anything wildly past that means the stream is
# no longer frame-aligned.
CONSOLE_MSG_MAX = 65536
CONSOLE_INBUF_MAX = 1 << 20

CONTROL_PORT_OFFSET = 7            # control port = console port + 7
SEAT_LEFT_GRACE_SECONDS = 15.0    # a beacon dying younger than this is
                                   # the pod's designed post-assign re-dial
                                   # (NOT a leave -- 2026-07-26 identity swap)
RESTART_EXIT_CODE = 42             # relay_main's exit code for `restart`
                                   # (btrelay_park.py relaunches on it)
CONTROL_AUTH_TIMEOUT = 10.0        # unauthenticated sockets die after this
CONTROL_OUTBUF_MAX = 262144        # slow-reader cutoff (bytes)
CONTROL_SET_KEYS = {"map", "time", "weather", "scenario", "temperature",
                    "length"}      # whitelist for the 'set' command

def control_secret():
    """Shared secret for control-channel auth, persisted next to the eggs.
    Auto-generated on first use; the operator shares it out-of-band."""
    import secrets as _secrets
    # The relay always runs with cwd = content\ (operator GUI and all rigs
    # set it); the secret lives beside the eggs it guards.
    path = "operator_secret.txt"
    try:
        with open(path, "r", encoding="ascii") as f:
            value = f.read().strip()
        if value:
            return value
    except OSError:
        pass
    value = _secrets.token_hex(16)
    try:
        with open(path, "w", encoding="ascii") as f:
            f.write(value + "\n")
    except OSError:
        pass
    return value


# Module-level sink list consulted by _StampedOut (the wrapper is installed
# before any Relay exists).  Guarded by a lock: the stdin thread prints too.
# _CONTROL_HISTORY keeps the last N log lines so an operator who connects
# MID-SESSION gets recent state replayed (their GUI rebuilds the roster from
# parsed lines -- without replay it would start blank).
_CONTROL_SINKS = []
_CONTROL_LOCK = threading.Lock()
_CONTROL_HISTORY = collections.deque(maxlen=400)


def _control_tee(text):
    """Best-effort fan-out of log text to authenticated control conns."""
    data = text.encode("utf-8", "replace")
    with _CONTROL_LOCK:
        _CONTROL_HISTORY.append(data)
        for conn in list(_CONTROL_SINKS):
            conn.queue_out(data)
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

# LAN auto-discovery (BT_RELAY=auto): pods broadcast DISC_PROBE on this UDP
# port; the relay answers DISC_REPLY + <H consolePort>.  The pod combines the
# replier's source IP with the advertised port.  Broadcast never crosses
# routers, so this is LAN-only by construction; internet players keep using
# the explicit host:port.
DISCOVERY_PORT = 15999
DISC_PROBE = b"BTR1DISC"
DISC_REPLY = b"BTR1HERE"


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"))
    # issue #38: an unknown camo color paints the mech SILENT GRAY (the
    # engine tolerates the vehicle-table miss by design) -- warn the
    # operator at load instead of letting a hand-edited egg ship it.
    for raw in open(egg_path, "rb").read().replace(b"\r\n", b"\n").split(b"\n"):
        line = raw.strip()
        if line.lower().startswith(b"color="):
            color = line.split(b"=", 1)[1].decode("ascii", "replace").strip()
            if color and color not in VALID_CAMO_COLORS:
                print(f"[relay] WARNING: egg color={color!r} is not a camo "
                      f"color (valid: {sorted(VALID_CAMO_COLORS)}) -- that "
                      f"mech will paint GRAY", flush=True)
    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
        self.last_seen = time.time()   # liveness deadline (see DEAD_PEER_SECONDS)

    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
        self.buf = b""          # inbound reassembly (see _console_read)
        self.acked = False      # pod sent AcknowledgeEggFile -- a REAL pod.
        # LIVENESS (2026-07-26): a pod whose machine sleeps, whose Wi-Fi drops,
        # or which is killed without a FIN leaves a HALF-OPEN socket.  With no
        # keepalive and no deadline it used to sit here forever holding
        # acked=True, which permanently blocked _maybe_reset_round (it requires
        # NO conn acked) and held a phantom seat in the launch count.  Only a
        # console restart cleared it -- the operator's "reboot the console".
        self.last_seen = time.time()
        # Internet hardening: the console port is exposed, and random scanners
        # connect and would otherwise count as pods.  Only ACKED connections
        # count toward the launch gate; scanners never speak the protocol.


class RelayControlConn:
    """One operator control connection.  All sends are buffered/non-blocking
    so a stalled remote operator can never stall the relay."""

    def __init__(self, sock, addr):
        self.sock = sock
        self.addr = addr
        self.buf = b""
        self.outbuf = b""
        self.authed = False
        self.connected_at = time.time()
        self.dead = False

    def name(self):
        return "%s:%d" % (self.addr[0], self.addr[1])

    def queue_out(self, data):
        if self.dead or not self.authed:
            return
        self.outbuf += data
        if len(self.outbuf) > CONTROL_OUTBUF_MAX:
            self.dead = True                 # slow reader: cut it loose
            return
        self.flush_out()

    def flush_out(self):
        if self.dead or not self.outbuf:
            return
        try:
            n = self.sock.send(self.outbuf)
            self.outbuf = self.outbuf[n:]
        except (BlockingIOError, InterruptedError):
            pass
        except OSError:
            self.dead = True


class Relay:
    def __init__(self, console_port, egg_path, bind_addr, udp_drop_pct,
                 manual_launch=False, reserved_tags=None):
        self.console_port = console_port
        self.game_port = console_port + 1
        self.bind_addr = bind_addr
        self.udp_drop_pct = udp_drop_pct
        self.manual_launch = manual_launch
        self._reserved_tags = set(reserved_tags or ())
        self.launch_requested = False            # set by the operator (stdin)
        self._launch_blocked_warned = False      # warned once about empty seats
        self.stop_requested = False              # operator 'stop' (End Mission)
        self.mission_length = parse_egg_mission_length(egg_path)
        self.mission_started_at = None           # set when RunMission #2 fires
        self.stop_sent = False
        self.eggs_done_at = None
        self.egg_bytes = egg_wire_bytes(open(egg_path, "rb").read())
        self.egg_path = egg_path
        # WALK-UP SEAT PREFS (2026-07-22): callsign/mech requested in the
        # SEAT_REQUEST payload, keyed by assigned host id.  Eggs are HELD
        # until the roster completes so every pod's egg carries every
        # player's callsign (an early pod's egg would otherwise miss late
        # joiners' labels); pods animate in WAITING FOR MISSION ASSIGNMENT
        # meanwhile.
        self.seat_prefs = {}
        self.eggs_released = False
        # SEAT RECLAIM (2026-07-23): tag -> expiry; set when a seated player
        # drops (round end / crash).  A rejoin presenting that tag gets the
        # SAME seat back (static ordering, real-pod style); other joiners
        # skip reclaim-held seats until the grace expires.
        self.seat_reclaim = {}
        # ROUND RESET originals (2026-07-22): trim/prefs FINALIZE the egg for
        # one round; when every pod leaves, the round resets to these so the
        # NEXT round's joins hold/rewrite a FRESH egg.  (Field report: a
        # rejoin after a crash got the stale finalized egg -- the new mech
        # request was recorded but never applied.)
        self.orig_egg_path = egg_path
        self.orig_egg_bytes = self.egg_bytes
        # PRESENCE BEACONS (2026-07-22, "launch with whoever connects"): the
        # pod KEEPS its seat-request TCP connection open until it exits; a
        # live beacon = a seated player.  A beacon dropping before the seat
        # was claimed (game-side HELLO) FREES the seat.  The operator's
        # LAUNCH trims the roster to the seats present.
        self.seat_beacons = {}
        self.roster = parse_egg_roster(egg_path)
        self.orig_roster = list(self.roster)    # round-reset restore point
        self.expected_ids = set(range(FIRST_GAME_HOST_ID,
                                      FIRST_GAME_HOST_ID + len(self.roster)))
        # OPERATOR-RESERVED SEATS: roster seats the operator flies LOCALLY (with
        # an explicit BT_SELF).  Those instances take ~15s to boot, so a remote
        # join.bat (no BT_SELF -> SEAT_REQUEST) would otherwise be assigned the
        # operator's seat first, and the operator's later HELLO would collide
        # ("already registered" -> dropped -> game never launches).  Seat
        # ASSIGNMENT skips these host_ids; the explicit HELLO still claims them.
        self.reserved_host_ids = set()
        for i, tag in enumerate(self.roster):
            if tag in self._reserved_tags:
                self.reserved_host_ids.add(FIRST_GAME_HOST_ID + i)
        if self.reserved_host_ids:
            print(f"[relay] operator-reserved seats (no auto-assign): "
                  f"{sorted(self.reserved_host_ids)}", flush=True)
        self.sel = selectors.DefaultSelector()
        self.console_conns = []
        self.game_conns = []                     # RelayGameConn (incl. unregistered)
        self.by_host = {}                        # hostID -> RelayGameConn
        self.seat_reservations = {}              # hostID -> expiry (auto-seats)
        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
        self.ctl_conns = []                    # RelayControlConn list
        self.restart_requested = False         # remote `restart` (parked relay)
        self.ctl_secret = control_secret()

    # ---------------- 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))
        # Remote-operator control channel (best-effort; the game relay must
        # come up even if this port is taken).
        self.ctl_port = self.console_port + CONTROL_PORT_OFFSET
        try:
            ctl_l = self._listener(self.ctl_port)
            self.sel.register(ctl_l, selectors.EVENT_READ, ("ctl_listen", None))
            print(f"[ctl] operator control on port {self.ctl_port} "
                  f"(secret in content\\operator_secret.txt)", flush=True)
        except OSError as e:
            print(f"[ctl] control port {self.ctl_port} unavailable ({e!r}) "
                  f"-- remote operators disabled; stdin still works",
                  flush=True)
        # LAN auto-discovery responder (best-effort: another relay may own it).
        self.disc_sock = None
        try:
            self.disc_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            self.disc_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            self.disc_sock.bind((self.bind_addr, DISCOVERY_PORT))
            self.disc_sock.setblocking(False)
            self.sel.register(self.disc_sock, selectors.EVENT_READ,
                              ("disc", None))
            print(f"[relay] LAN discovery answering on udp/{DISCOVERY_PORT} "
                  f"(pods may use BT_RELAY=auto)", flush=True)
        except OSError as e:
            print(f"[relay] LAN discovery unavailable ({e!r}) -- "
                  f"pods must use the explicit address", flush=True)
            self.disc_sock = 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()
                elif kind == "disc":
                    self._disc_read()
                elif kind == "ctl_listen":
                    self._ctl_accept(key.fileobj)
                elif kind == "ctl":
                    self._ctl_read(obj)
            self._tick_control()
            self._reap_dead_peers()
            self._tick_launch()
            self._tick_stop()
            self._tick_settle()
            self._tick_stats()
            if self.restart_requested:
                # Remote `restart` (parked-relay deployment): leave the loop so
                # relay_main can exit with RESTART_EXIT_CODE and the supervisor
                # relaunches us with a fresh egg read.  This is the ONLY way a
                # travelling operator can resize the roster or recover a wedged
                # relay -- the control channel cannot do either in-process.
                print("[relay] RESTART requested by the operator -- closing "
                      "for the supervisor to relaunch", flush=True)
                return

    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()
        _enable_keepalive(sock)
        sock.setblocking(False)
        conn = RelayConsoleConn(sock, addr)
        self.console_conns.append(conn)
        self.sel.register(sock, selectors.EVENT_READ, ("console", conn))
        # Egg delivery: HELD until every pod's console pad is connected
        # (walk-up seat prefs must be in every pod's egg copy).  NOT gated on
        # game-side registration -- a pod HELLOs only after it parses the
        # egg's roster, so that gate would deadlock.  Seat requests precede
        # the console pad, so by the Nth pad every pref is already in.
        if self.eggs_released:
            self._send_egg(conn)
        elif len(self.console_conns) >= len(self.roster):
            self._release_eggs()
        else:
            print(f"[relay] console conn {addr[0]}:{addr[1]}: egg HELD "
                  f"({len(self.console_conns)}/{len(self.roster)} pods "
                  f"present)", flush=True)

    def _send_egg(self, conn):
        addr = conn.addr        # NOT getpeername(): it raises on a dead peer
        try:
            n = 0
            for pkt in egg_packets(self.egg_bytes):
                _send_all_guarded(conn.sock, pkt, "egg chunk")
                n += 1
            conn.egg_sent = True
            conn.egg_sent_at = time.time()      # the ACK-debt clock starts NOW
            print(f"[relay] console conn {addr[0]}:{addr[1]}: egg sent "
                  f"({n} chunks); awaiting pod ACK", 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:
                conn.sock.setblocking(False)
            except OSError:
                pass

    def _reload_egg_file(self):
        """BETWEEN-ROUNDS MISSION EDITS (2026-07-23): re-read the session egg
        FILE so operator changes (arena/time/weather/length via the console's
        Apply button) take effect next round without a session restart.  The
        roster SHAPE stays fixed mid-session (seat ids are positional): a
        row-count change is refused loudly."""
        try:
            new_roster = parse_egg_roster(self.orig_egg_path)
            if len(new_roster) != len(self.orig_roster):
                print(f"[relay] egg file roster changed "
                      f"{len(self.orig_roster)}->{len(new_roster)} seats -- "
                      f"IGNORED (restart the session to resize)", flush=True)
                return
            self.orig_egg_bytes = egg_wire_bytes(
                open(self.orig_egg_path, "rb").read())
            self.egg_bytes = self.orig_egg_bytes
            self.egg_path = self.orig_egg_path
            new_length = parse_egg_mission_length(self.orig_egg_path)
            if new_length != self.mission_length:
                print(f"[relay] mission length now {new_length:.0f}s",
                      flush=True)
                self.mission_length = new_length
        except OSError as e:
            print(f"[relay] egg reload failed ({e!r}) -- keeping the "
                  f"previous mission", flush=True)

    def _release_eggs(self, present=None):
        if self.eggs_released:
            return
        # issue #33: after a ROUND ABORT, hold the next release until the
        # settle window passes -- rejoiners re-ACK within a second and an
        # instant re-release hands the flapping pod a fresh round to abort.
        if time.time() < getattr(self, "round_hold_until", 0):
            if not getattr(self, "_hold_noted", False):
                self._hold_noted = True
                print(f"[relay] egg release HELD "
                      f"({self.round_hold_until - time.time():.0f}s settle "
                      f"after the round abort)", flush=True)
            return
        self._hold_noted = False
        self._reload_egg_file()             # pick up between-rounds edits
        self.eggs_released = True
        if present is not None and len(present) < len(self.roster):
            self._trim_roster(present)
        if self.seat_prefs:
            self._apply_seat_prefs()
        for conn in list(self.console_conns):
            if not conn.egg_sent:
                self._send_egg(conn)

    def _trim_roster(self, present):
        """LAUNCH WITH WHOEVER CONNECTS: shrink the session to the seats
        with live beacons (or claims).  Positions shift, so seat ids REMAP --
        a pod re-derives its host id from its TAG's position in the received
        egg, so prefs/beacons re-key by position and every pod stays
        consistent.  Runs pre-release only (no pod has HELLO'd yet)."""
        keep_idx = [h - FIRST_GAME_HOST_ID for h in present]
        keep_tags = [self.roster[i] for i in keep_idx if 0 <= i < len(self.roster)]
        old_ids = list(present)
        print(f"[relay] LAUNCH trim: {len(keep_tags)}/{len(self.roster)} "
              f"seats present -- roster shrinks to {keep_tags}", flush=True)
        if eggmodel is not None:
            try:
                doc = eggmodel.EggDoc.load(self.egg_path)
                for i, pl in enumerate(doc.pilots()):
                    if i not in keep_idx:
                        doc.remove_section(pl["address"])
                doc.replace_section("pilots",
                                    ["pilot=%s" % t for t in keep_tags])
                trimmed = os.path.join(os.path.dirname(self.egg_path) or ".",
                                       "_relay_trimmed.egg")
                doc.save(trimmed)
                self.egg_path = trimmed          # prefs rewrite loads THIS
                self.egg_bytes = egg_wire_bytes(
                    doc.emit().encode("latin-1", "replace"))
            except Exception as e:
                print(f"[relay] roster trim FAILED ({e!r}) -- launching the "
                      f"FULL roster (empty seats will stall!)", flush=True)
                return
        # re-key everything by the new positions
        remap = {old: FIRST_GAME_HOST_ID + n for n, old in enumerate(old_ids)}
        self.seat_prefs = {remap[h]: v for h, v in self.seat_prefs.items()
                           if h in remap}
        new_beacons = {}
        for old, conn in self.seat_beacons.items():
            if old in remap:
                conn.seat_host = remap[old]
                new_beacons[remap[old]] = conn
        self.seat_beacons = new_beacons
        self.roster = keep_tags
        self.expected_ids = set(range(FIRST_GAME_HOST_ID,
                                      FIRST_GAME_HOST_ID + len(keep_tags)))

    def _apply_seat_prefs(self):
        # Rewrite vehicle= + the callsign system from the walk-up prefs.
        # eggmodel's rasterizer needs PySide6 (present when launched from the
        # operator app); without it the vehicle still applies and the
        # callsign falls back to the text name= (labels keep the roster
        # default bitmaps).
        if eggmodel is None:
            print("[relay] eggmodel unavailable -- seat prefs NOT applied",
                  flush=True)
            return
        try:
            doc = eggmodel.EggDoc.load(self.egg_path)
            pilots = doc.pilots()
            names = [p.get("name") or p["address"] for p in pilots]
            for host_id, (callsign, mech) in sorted(self.seat_prefs.items()):
                i = host_id - FIRST_GAME_HOST_ID
                if not (0 <= i < len(pilots)):
                    continue
                if mech:
                    doc.set_kv(pilots[i]["address"], "vehicle", mech)
                if callsign:
                    names[i] = callsign
            try:
                doc.set_callsigns(eggmodel.make_callsigns(names))
            except Exception as e:
                print(f"[relay] callsign rasterizer unavailable ({e!r}) -- "
                      f"vehicles applied, callsign bitmaps unchanged",
                      flush=True)
            self.egg_bytes = egg_wire_bytes(
                doc.emit().encode("latin-1", "replace"))
            for host_id, (callsign, mech) in sorted(self.seat_prefs.items()):
                i = host_id - FIRST_GAME_HOST_ID
                if 0 <= i < len(pilots):
                    print(f"[relay] player {host_id - FIRST_GAME_HOST_ID + 1} "
                          f"egg: vehicle={mech or pilots[i].get('vehicle')} "
                          f"callsign={callsign or names[i]!r}", flush=True)
        except Exception as e:
            print(f"[relay] seat-pref egg rewrite FAILED ({e!r}) -- "
                  f"serving the original egg", flush=True)

    def _rekey_seats_to_roster(self):
        """Re-key live beacons/prefs by each TAG's position in the CURRENT
        roster.  Needed whenever the roster changes shape (round reset, re-arm):
        seat ids are positional, so a beacon keyed by a TRIMMED round's host id
        points at the wrong seat once the template roster is restored -- a
        departed player's tag was being minted into the next round's egg while a
        present player's was trimmed out (review 2026-07-26).  Beacons whose tag
        is no longer in the roster are dropped from the maps (their seats free
        up); reservations are cleared wholesale, matching the round-reset
        semantics."""
        old_beacons = self.seat_beacons
        old_prefs = self.seat_prefs
        self.seat_beacons = {}
        self.seat_prefs = {}
        self.seat_reservations = {}
        for old_id, conn in old_beacons.items():
            tag = getattr(conn, "seat_tag", None)
            if tag in self.roster:
                new_id = FIRST_GAME_HOST_ID + self.roster.index(tag)
                conn.seat_host = new_id
                self.seat_beacons[new_id] = conn
                if old_id in old_prefs:
                    self.seat_prefs[new_id] = old_prefs[old_id]

    def _maybe_reset_round(self):
        # All pods gone from a FINALIZED round -> restore the template so the
        # next round's joins are honored (held egg, fresh prefs/trim).  Live
        # beacons (players already waiting for the next round) are re-keyed
        # by their TAG's position in the restored roster.
        if not self.eggs_released:
            return
        if self.by_host or any(c.acked for c in self.console_conns):
            return
        self.eggs_released = False
        self.egg_path = self.orig_egg_path
        self._reload_egg_file()
        self.egg_bytes = self.orig_egg_bytes
        self.roster = list(self.orig_roster)
        self.expected_ids = set(range(FIRST_GAME_HOST_ID,
                                      FIRST_GAME_HOST_ID + len(self.roster)))
        self.launches_sent = 0
        self.launch_at = None
        self.launch_requested = False
        self.eggs_done_at = None
        self._launch_blocked_warned = False
        self.stop_sent = False
        self.stop_requested = False
        self.mission_started_at = None
        self._rekey_seats_to_roster()
        print(f"[relay] round RESET -- roster/egg restored "
              f"({len(self.seat_beacons)} player(s) already waiting)",
              flush=True)

    def _reap_dead_peers(self):
        """Age out a peer that OWES a protocol response and has gone silent.

        THE DOCTRINE (rewritten twice in review, 2026-07-26 -- keep it): an
        app-level deadline is only valid where silence is ABNORMAL, i.e. while a
        response is owed.  Everything else is TCP keepalive's job (set on every
        accept; ~130s detection via SIO_KEEPALIVE_VALS), which kills the socket
        and lands in the normal recv/drop path.

        What a healthy peer's silence looks like, so nobody re-widens this:
          * a console pad sends exactly ONE message in its life -- the egg ACK.
            After ACKing it is silent FOREVER.  The first cut reaped ACKed pads
            180s into any staging hold (a manual-launch operator waiting on a
            straggler), force-relaunching every healthy client.  [blocker]
          * a REGISTERED game conn is loading the mission: quiet on TCP until
            READY, and its 15s UDP HELLOs never arrive if the pod runs
            BT_RELAY_TCP_ONLY=1 or UDP is blocked (the documented TCP fallback).
            Reaping it _abort_round()s the whole night, every ~3 minutes,
            blaming a healthy player.  [major]
          * between rounds every pod is byte-silent by design (the seat beacon
            is write-only: L4NET.CPP "the relay ignores its silence").

        So the ONLY thing this reaps is a console pad that was sent the egg and
        never ACKed within DEAD_PEER_SECONDS -- a wedged or dead client holding
        the eggs_done gate open.  Debt paid (acked) = exempt.  No egg = exempt.
        Game conns are never reaped here at all.
        """
        if self.launches_sent >= 2 or not self.eggs_released:
            return
        now = time.time()
        for conn in list(self.console_conns):
            if not conn.egg_sent or conn.acked:
                continue                # nothing owed / debt already paid
            # Measure the debt from EGG SEND, not from the last byte: a pad that
            # connected early and sat through a long held-egg wait would
            # otherwise be over the deadline the instant the egg went out.
            owed_since = max(getattr(conn, "last_seen", now),
                             getattr(conn, "egg_sent_at", now))
            if now - owed_since > DEAD_PEER_SECONDS:
                print(f"[relay] console conn {conn.addr[0]}:{conn.addr[1]} "
                      f"REAPED -- egg sent {now - owed_since:.0f}s ago and "
                      f"never ACKed; it can no longer hold the round open",
                      flush=True)
                self._drop_console(conn)

    def _tag_of(self, host_id):
        """The LIVE roster tag for a host id ('?' if out of range)."""
        i = host_id - FIRST_GAME_HOST_ID
        return self.roster[i] if 0 <= i < len(self.roster) else "?"

    def _pods_ready(self):
        """REAL pods = console connections that ACKED the egg (scanners never
        speak the protocol, so they can't hold a roster slot)."""
        return sum(1 for c in self.console_conns if c.acked)

    def _check_launch_gate(self):
        # BACK-TO-BACK MISSIONS (2026-07-18): a previous mission finished
        # (launches_sent==2) and now all seats have RE-ACKED -- i.e. every pod
        # exited, re-ran join.bat, and reconnected fresh.  Reset the launch
        # state so the relay (and the operator's LAUNCH button) re-arm for a
        # new round -- no Stop/Start needed.  The binary needs nothing: a
        # rejoined pod is a brand-new process at WaitingForLaunch.  (Only ever
        # reached on a pod ACK, which never happens mid-mission.)
        if self.launches_sent >= 2 and self._pods_ready() >= len(self.roster):
            print("[relay] all seats rejoined after the last mission -- "
                  "re-arming for a NEW mission", flush=True)
            self.launches_sent = 0
            self.stop_sent = False
            self.stop_requested = False
            self.mission_started_at = None
            self.launch_at = None
            self.launch_requested = False
            self._launch_blocked_warned = False
            self.eggs_done_at = None
            # fall through -> the normal gate re-prints WAITING FOR OPERATOR

        # GLOBAL launch: arm the timer when the LAST pod has ACKED its egg
        # (auto mode), or wait for the operator's 'launch' command (manual
        # mode -- the operator app's Launch button writes it to our stdin).
        if self._pods_ready() >= len(self.roster) and self.launches_sent == 0 \
                and self.eggs_done_at is None:
            self.eggs_done_at = time.time()
            if self.manual_launch:
                print("[relay] all pods ACKED the egg; WAITING FOR OPERATOR "
                      "LAUNCH", flush=True)
            else:
                self.launch_at = self.eggs_done_at + LAUNCH_SETTLE_SECONDS
                print(f"[relay] all pods ACKED the egg; LAUNCH pair in "
                      f"{LAUNCH_SETTLE_SECONDS}s", flush=True)

    def _console_read(self, conn):
        try:
            data = conn.sock.recv(4096)
        except OSError:
            data = b""
        if data:
            conn.last_seen = time.time()
        if not data:
            print(f"[relay] console conn {conn.addr[0]}:{conn.addr[1]} closed",
                  flush=True)
            self._drop_console(conn)
            return
        #
        # REASSEMBLE, then walk EVERY complete message.  This used to parse one
        # recv() at a fixed offset 0 and look at nothing else, which had two
        # failure modes on a real network (loopback hid both):
        #   * the pod's 28-byte ACK arriving SPLIT (say 16 bytes then 12) was
        #     dropped by the `len(data) >= 24` test and never seen again -- and a
        #     missed ACK means that seat never counts toward the launch gate, so
        #     the round can never be released.  A silent, unrecoverable wedge.
        #   * two messages coalescing into one recv() -- only the first was read.
        # A frame is a 16-byte header + `messageLength` bytes (so the egg's 1040
        # = 16 + 1024, and the ACK's 28 = 16 + 12).
        #
        conn.buf += data
        if len(conn.buf) > CONSOLE_INBUF_MAX:
            print(f"[relay] console conn {conn.addr[0]}:{conn.addr[1]} dropped: "
                  f"inbound buffer over {CONSOLE_INBUF_MAX} bytes (desynced)",
                  flush=True)
            self._drop_console(conn)
            return
        while len(conn.buf) >= 24:
            (clid, _gid, fh, _ts) = struct.unpack_from("<iiii", conn.buf, 0)
            (mlen, mid) = struct.unpack_from("<Ii", conn.buf, 16)
            #
            # A stream protocol cannot be resynced by guessing, so an impossible
            # length means we are no longer frame-aligned: say so and drop, rather
            # than silently mis-reading this pod for the rest of the night.
            #
            if mlen < 8 or mlen > CONSOLE_MSG_MAX:
                print(f"[relay] console conn {conn.addr[0]}:{conn.addr[1]} "
                      f"dropped: implausible messageLength {mlen} "
                      f"(clientID={clid} msgID={mid}) -- stream desynced",
                      flush=True)
                self._drop_console(conn)
                return
            frame = 16 + mlen
            if len(conn.buf) < frame:
                break                       # partial: wait for the rest
            conn.buf = conn.buf[frame:]
            print(f"[relay] pod->console clientID={clid} fromHost={fh} "
                  f"msgID={mid} len={mlen}", flush=True)
            # AcknowledgeEggFile (clientID=0 NetworkManager, msgID=4): this
            # connection is a REAL pod -- count it toward the launch gate.
            if clid == 0 and mid == 4 and not conn.acked:
                conn.acked = True
                print(f"[relay] pod ACK from {conn.addr[0]}:{conn.addr[1]} "
                      f"({self._pods_ready()}/{len(self.roster)} ready)",
                      flush=True)
                self._check_launch_gate()

    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)
        self._maybe_reset_round()

    def _log_launch_readiness(self):
        # PRE-LAUNCH READINESS (2026-07-18): make the "launching short" footgun
        # VISIBLE.  A roster seat with no registered pod means the mission will
        # STALL -- every pod waits on that empty seat's connection and the
        # All-connections-completed gate never fires.  Log filled vs empty
        # seats so the operator sees it BEFORE the confusion.
        filled, empty = [], []
        for i, tag in enumerate(self.roster):
            host_id = FIRST_GAME_HOST_ID + i
            (filled if host_id in self.by_host else empty).append(
                f"seat{i + 1}({tag})")
        print(f"[relay] LAUNCH readiness: {len(filled)}/{len(self.roster)} "
              f"seats filled -- {', '.join(filled) or 'NONE'}", flush=True)
        if empty:
            print(f"[relay] *** WARNING: {len(empty)} EMPTY seat(s): "
                  f"{', '.join(empty)} -- the mission will STALL (every pod "
                  f"waits on the empty seat).  Stop, reduce the roster to the "
                  f"players present, and re-launch.", flush=True)

    def _rearm_for_new_round(self, why):
        """OPERATOR-DRIVEN RE-ARM (fixes the "LAUNCH does nothing after a round"
        wedge, operator report 2026-07-25).

        The two automatic re-arm paths are both ALL-OR-NOTHING:
        `_check_launch_gate` needs EVERY seat of the last round to re-ACK, and
        `_maybe_reset_round` needs EVERY pod to be gone.  A games night lives
        between those: most pods rejoin, one player closes their window or
        crashes.  In that state `launches_sent` stayed latched at 2, the manual
        LAUNCH branch (guarded on `launches_sent == 0`) ignored every press, and
        because the "not all seats filled" diagnostic sits inside that same
        guard, NOTHING was printed -- the operator saw a dead button and had to
        restart the session.  (The operator confirmed the mirror image: with a
        stable group, where everyone does rejoin, relaunching worked fine.)

        So: an explicit operator LAUNCH is now itself sufficient authority to
        start a new round.  Reset only the ROUND state -- seats, beacons, prefs
        and connections are left alone, so whoever is here stays here.
        """
        #
        # GUARDED ON EVERY CHANNEL (review 2026-07-26).  The `launch_at is None`
        # guard lived only on the LAUNCH-press path, but the ctl `rearm` command,
        # the stdin reader and the always-lit GUI button all reached here
        # unconditionally.  Mid-mission (launches_sent==2) a re-arm zeroes the
        # counters, which permanently kills the mission clock AND makes End
        # Mission print "no mission is running" -- the round becomes
        # unstoppable except by Stop Session, the exact outcome this feature
        # exists to eliminate.  Mid-pair (launch_at set) it destroys the
        # RunMission pair and re-releases eggs to mid-launch pods -- the re-arm
        # storm.  Refuse both, loudly.
        #
        # "Running" is launches_sent >= 2 AND no StopMission issued yet.  NOT
        # launches_sent alone: after a FINISHED round (clock expiry or End
        # Mission -> stop_sent True) launches_sent still reads 2 until the
        # re-arm itself clears it -- refusing on that would resurrect the very
        # wedge this feature exists to fix (caught by the regression suite when
        # the review's own proposed guard was applied verbatim).
        if self.launches_sent >= 2 and not self.stop_sent:
            print("[relay] RE-ARM refused -- a mission is RUNNING; press "
                  "End Mission first", flush=True)
            self.launch_requested = False       # do not let the press linger
            return
        if self.launch_at is not None:
            print(f"[relay] RE-ARM refused -- a launch sequence is in flight "
                  f"(fires in {max(0.0, self.launch_at - time.time()):.0f}s)",
                  flush=True)
            self.launch_requested = False
            return
        print(f"[relay] RE-ARM for a new round ({why}); "
              f"previous round state cleared", flush=True)
        self.launches_sent = 0
        self.launch_at = None
        self.stop_sent = False
        # A stop the operator asked for during the LAST round must not carry
        # into this one -- see _tick_stop, where a stale latch StopMissioned
        # the next mission in the tick it launched.
        self.stop_requested = False
        self.mission_started_at = None
        self.eggs_done_at = None
        self._launch_blocked_warned = False
        # Re-open the round for joins/edits: restore the untrimmed roster and the
        # template egg, so round 2+ does not serve last round's trimmed egg (new
        # joiners got ROSTER FULL) and between-round mission edits are picked up.
        #
        # UNCONDITIONAL, and that matters.  An earlier cut gated this on
        # `eggs_released`, which made it useless in the one state that most needs
        # it: _abort_round clears eggs_released BEFORE the survivors' sockets
        # close, so _maybe_reset_round's own `if not self.eggs_released: return`
        # then skips the template restore forever.  The roster stays trimmed to
        # last round's attendance, the release gates (which all compare against
        # len(self.roster)) can never be met, walk-ups get ROSTER FULL, and the
        # relay sits dead -- live-proven on 2026-07-25, nearly 5 minutes of a
        # games night lost.  Re-arm has to be able to dig out of exactly that.
        self.eggs_released = False
        self.egg_path = self.orig_egg_path
        self.egg_bytes = self.orig_egg_bytes
        self.roster = list(self.orig_roster)
        self.expected_ids = set(range(FIRST_GAME_HOST_ID,
                                      FIRST_GAME_HOST_ID + len(self.roster)))
        self.round_hold_until = 0       # do not inherit an abort's settle window
        self._reload_egg_file()
        # Re-key beacons/prefs to the RESTORED roster (review 2026-07-26): they
        # were keyed by the previous round's TRIMMED host ids, and the launch
        # fall-through would reinterpret those against the full roster -- a
        # departed player's tag minted into round 2's egg, a present player's
        # trimmed out, and every callsign/mech shifted a slot.
        self._rekey_seats_to_roster()
        for c in self.console_conns:
            c.acked = False             # last round's ACK must not count
            c.egg_sent = False          # every pod needs THIS round's egg

    def _tick_launch(self):
        # RE-ARM ON DEMAND: an operator LAUNCH while a previous round is still
        # latched (launches_sent >= 1 -- 2 after a normal round, 1 if a round
        # was aborted between the RunMission pair) is a deliberate new-round
        # request, not a no-op.  This must run BEFORE the gate below, whose
        # `launches_sent == 0` guard is exactly what used to swallow the press.
        # `launch_at is not None` means a launch SEQUENCE IS IN FLIGHT: the
        # RunMission pair is timed, and `launch_requested` deliberately stays set
        # across it (it is only cleared once #2 has gone out).  Re-arming on that
        # state resets launches_sent between #1 and #2, so the pair never
        # completes and the relay re-releases eggs every tick -- a re-arm storm
        # that kicks every pod into an identity-resync loop.  Caught on the rig,
        # 2026-07-26.  So: only re-arm when nothing is in flight.
        if (self.manual_launch and self.launch_requested
                and self.launch_at is None and self.launches_sent >= 1):
            self._rearm_for_new_round("operator LAUNCH after a finished round")
            # fall through: the request stands and is served by the gate below.

        # Manual mode: the operator's 'launch' arms the timer, still honouring
        # the settle window (pods must reach WaitingForLaunch or the handler
        # Fail()s -- the same reason the auto timer waits).
        if (self.manual_launch and self.launch_requested
                and self.launch_at is None and self.launches_sent == 0):
            if not self.eggs_released:
                # LAUNCH WITH WHOEVER CONNECTS: the operator fired before the
                # roster filled -- shrink the session to the seated players
                # and release the eggs; the launch pair fires once they ACK.
                present = sorted(set(self.seat_beacons) | set(self.by_host))
                if not present:
                    if not self._launch_blocked_warned:
                        self._launch_blocked_warned = True
                        print("[relay] LAUNCH pressed but NO players are "
                              "seated yet -- waiting for joins", flush=True)
                    return
                self._launch_blocked_warned = False
                self._release_eggs(present)
                return                      # gate proceeds as their ACKs land
            if self.eggs_done_at is not None:
                self.launch_at = max(time.time(),
                                     self.eggs_done_at + LAUNCH_SETTLE_SECONDS)
                wait = max(0.0, self.launch_at - time.time())
                print(f"[relay] operator LAUNCH received; firing in "
                      f"{wait:.0f}s", flush=True)
                self._log_launch_readiness()
            elif not self._launch_blocked_warned:
                # Operator pressed LAUNCH but the gate is NOT ready (not every
                # roster seat has ACKed its egg) -- previously this did NOTHING
                # visible ("I pressed launch and nothing happened").  Say WHY.
                self._launch_blocked_warned = True
                print("[relay] LAUNCH pressed but NOT all seats have ACKed "
                      "their egg yet -- holding:", flush=True)
                self._log_launch_readiness()
                print("[relay] if a listed seat is never coming back, press "
                      "RE-ARM to re-open the round and launch with whoever is "
                      "here (no session restart needed)", flush=True)
        if self.launch_at is None or self.launches_sent >= 2:
            return
        if time.time() < self.launch_at:
            return
        # ALL-READY GATE (2026-07-22, live finding round 3): firing the
        # mission while a pod is still LOADING drowns that pod's load in the
        # running mission's update streams (same event queue) -- the operator,
        # who always joins last, loaded 5+ minutes or crashed while testers
        # took ~20s.  Hold the RunMission pair until every REGISTERED pod has
        # sent READY (route -10): everyone enters together, 1995-style.
        # Escape hatches: a pod that never readies is announced every 10s
        # (the operator sees who), and a 3-minute cap force-fires so one
        # wedged pod cannot hold the night hostage.
        if self.launches_sent == 0:
            not_ready = [h for h, c in sorted(self.by_host.items())
                         if not getattr(c, "ready", False)]
            if not_ready and time.time() < self.launch_at + 180.0:
                now = time.time()
                if now - getattr(self, "_hold_notice_at", 0) >= 10.0:
                    self._hold_notice_at = now
                    names = ", ".join(
                        "PLAYER %d" % (h - FIRST_GAME_HOST_ID + 1)
                        for h in not_ready)
                    print(f"[relay] launch HELD -- still loading: {names} "
                          f"({len(self.by_host) - len(not_ready)}/"
                          f"{len(self.by_host)} ready)", flush=True)
                return
            if not_ready:
                print(f"[relay] launch FORCED after 180s hold -- "
                      f"{len(not_ready)} pod(s) never readied", flush=True)
        pkt = run_mission_packet()
        for conn in list(self.console_conns):
            try:
                _send_all_guarded(conn.sock, pkt, "launch")
            except OSError as e:
                print(f"[relay] launch send failed to {conn.addr}: {e!r} "
                      f"-- dropping that pod", flush=True)
                self._drop_console(conn)
        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)
        if self.launches_sent == 2:
            self.mission_started_at = time.time()
            # CONSUME the launch: clear the timer + the request so the NEXT
            # LAUNCH press is a fresh, deliberate new-mission request (else the
            # back-to-back block would re-fire while this mission is running).
            self.launch_at = None
            self.launch_requested = False
            if self.mission_length > 0:
                print(f"[relay] mission clock: {self.mission_length:.0f}s "
                      "(StopMission at expiry)", flush=True)

    def _tick_stop(self):
        # THE MISSION CLOCK (console-side, as in 1995): send StopMission when
        # the egg's [mission] length expires, or on the operator's 'stop'
        # command (the End Mission button).
        if self.launches_sent < 2:
            # NO MISSION IS RUNNING.  A stop asked for now (End Mission pressed
            # between rounds, which is a natural operator reflex) used to sit
            # latched and then StopMission the NEXT mission in the very tick it
            # launched -- indistinguishable, from the operator's seat, from
            # "launch did nothing".  Consume and announce it instead.
            if self.stop_requested:
                self.stop_requested = False
                print("[relay] END MISSION ignored -- no mission is running "
                      "(the latch was cleared, so it cannot kill the next "
                      "launch)", flush=True)
            return
        if self.stop_sent:
            return
        expired = (self.mission_length > 0
                   and self.mission_started_at is not None
                   and time.time() >= self.mission_started_at
                       + self.mission_length + STOP_GRACE_SECONDS)
        if not (expired or self.stop_requested):
            return
        pkt = stop_mission_packet()
        for conn in list(self.console_conns):
            try:
                _send_all_guarded(conn.sock, pkt, "stop")
            except OSError as e:
                print(f"[relay] stop send failed to {conn.addr}: {e!r}",
                      flush=True)
        self.stop_sent = True
        why = "operator END MISSION" if self.stop_requested else "time expired"
        print(f"[relay] StopMission sent to {len(self.console_conns)} pod(s) "
              f"({why})", flush=True)

    # ---------------- game TCP side ----------------

    def _accept_game(self, listener):
        sock, addr = listener.accept()
        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        _enable_keepalive(sock)
        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 data:
            conn.last_seen = time.time()
        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)
            # game frames are small; the matchlog upload (route -9) is a whole
            # file -- allow it its own generous cap (matches the client's 8MB)
            cap = MATCHLOG_CAP if route == ROUTE_MATCHLOG else FRAME_CAP
            if length > 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 and route == ROUTE_MATCHLOG:
            # MATCHLOG AUTO-UPLOAD: a pod's post-mission throwaway dial hands
            # us its match forensic log (payload: filename NUL + file bytes).
            # Saved under matchlogs/ prefixed with the sender address so all
            # peers' files land side by side for tools/matchcheck.py.
            nul = payload.find(b"\0")
            if nul <= 0 or nul > 150:
                self._drop_game(conn, "malformed matchlog upload")
                return
            name = os.path.basename(payload[:nul].decode("ascii", "replace"))
            if not (name.startswith("matchlog_") and name.endswith(".txt")):
                self._drop_game(conn, f"suspicious matchlog name {name!r}")
                return
            data = payload[nul + 1:]
            os.makedirs("matchlogs", exist_ok=True)
            peer = conn.sock.getpeername()[0].replace(":", "_")
            out_path = os.path.join("matchlogs", f"{peer}_{name}")
            with open(out_path, "wb") as f:
                f.write(data)
            print(f"[relay] matchlog saved: {out_path} ({len(data)} bytes)",
                  flush=True)
            return
        if conn.host_id is None and route == ROUTE_SEAT_REQUEST:
            # SERVER-ASSIGNED SEATS: hand out the lowest roster seat that is
            # neither claimed (by_host) nor reserved; reserve it briefly so
            # two simultaneous joiners can't race onto the same seat.  The
            # requesting connection closes after the reply; the pod re-dials
            # and claims the seat with a normal HELLO.
            now = time.time()
            self.seat_reservations = {h: t for h, t in
                                      self.seat_reservations.items()
                                      if t > now or h in self.seat_beacons}
            self.seat_reclaim = {t: e for t, e in self.seat_reclaim.items()
                                 if e > now}
            claim_tag = ""
            if payload:
                fields = payload.split(b"\0")
                if len(fields) > 2:
                    claim_tag = fields[2].decode("ascii", "replace").strip()
            assigned = None
            if claim_tag and claim_tag in self.roster:
                host_id = FIRST_GAME_HOST_ID + self.roster.index(claim_tag)
                if host_id not in self.by_host and host_id not in self.seat_beacons:
                    assigned = (host_id, claim_tag)
                    self.seat_reclaim.pop(claim_tag, None)
                    print(f"[relay] seat RECLAIMED by returning player "
                          f"(tag '{claim_tag}')", flush=True)
            # STATIC SEATS (issue #34, night-2 capture): a manually restarted
            # client carries NO claim and used to mint a NEW seat while its
            # old one sat reserved with the same callsign -- duplicate roster
            # entries, and exactly what pods must never do.  Fallback: match
            # the requester's (IP + callsign) against remembered seat
            # identities and RECLAIM that seat -- displacing a zombie beacon
            # if one is still squatting on it.  (Pods are fixed machines:
            # same box + same name = same seat, always.)
            if assigned is None and payload:
                req_ip = conn.sock.getpeername()[0]
                req_callsign = (payload.split(b"\0")[0]
                                .decode("ascii", "replace").strip()[:15])
                if not hasattr(self, "seat_identity"):
                    self.seat_identity = {}
                if req_callsign:
                    for i, tag in enumerate(self.roster):
                        if self.seat_identity.get(tag) != (req_ip, req_callsign):
                            continue
                        host_id = FIRST_GAME_HOST_ID + i
                        if host_id in self.by_host:
                            # A half-open REGISTERED conn (machine slept, power
                            # cut) squats here until keepalive reaps it (~2 min),
                            # and its own player got ROSTER FULL meanwhile.  No
                            # mission running + same source IP = this player's
                            # own ghost: displace it, exactly like a zombie
                            # beacon (review 2026-07-26).
                            ghost = self.by_host[host_id]
                            if (self.launches_sent < 2
                                    and ghost.addr[0] == req_ip):
                                self._drop_game(
                                    ghost, "displaced by its returning player")
                            else:
                                break           # actively PLAYING: not ours
                        old = self.seat_beacons.get(host_id)
                        if old is not None:     # zombie duplicate: displace it
                            print(f"[relay] displacing stale beacon on seat "
                                  f"'{tag}' (same player rejoining)",
                                  flush=True)
                            old.seat_host = None
                            self.seat_beacons.pop(host_id, None)
                        assigned = (host_id, tag)
                        self.seat_reclaim.pop(tag, None)
                        print(f"[relay] seat RECLAIMED by identity "
                              f"({req_callsign}@{req_ip} -> '{tag}')",
                              flush=True)
                        break
            if assigned is None:
                for i, tag in enumerate(self.roster):
                    host_id = FIRST_GAME_HOST_ID + i
                    if (host_id in self.by_host or host_id in self.seat_reservations
                            or host_id in self.reserved_host_ids
                            or self.seat_reclaim.get(tag, 0) > now):
                        continue            # claimed / racing / held for return
                    assigned = (host_id, tag)
                    break
            if assigned is None:
                self._send_raw(conn, struct.pack(ENV_FMT_TCP,
                                                 ROUTE_SEAT_FULL, 0))
                print(f"[relay] {conn.name()} seat request: ROSTER FULL",
                      flush=True)
                return
            host_id, tag = assigned
            self.seat_reservations[host_id] = now + SEAT_RESERVE_SECONDS
            # WALK-UP PREFS: optional payload {callsign NUL mech NUL}
            pref_note = ""
            if payload:
                parts = payload.split(b"\0")
                callsign = parts[0].decode("ascii", "replace").strip()[:15]
                mech = (parts[1].decode("ascii", "replace").strip().lower()
                        if len(parts) > 1 else "")
                callsign = "".join(c for c in callsign
                                   if c.isalnum() or c in " -_.")
                if mech and mech not in VEHICLE_TAGS:
                    print(f"[relay] {conn.name()} requested unknown mech "
                          f"{mech!r} -- keeping the roster default", flush=True)
                    mech = ""
                if callsign or mech:
                    self.seat_prefs[host_id] = (callsign, mech)
                    pref_note = f" callsign={callsign!r} mech={mech or '(default)'}"
            payload_out = struct.pack("<i", host_id) + tag.encode() + b"\0"
            self._send_raw(conn, struct.pack(ENV_FMT_TCP, ROUTE_SEAT_ASSIGN,
                                             len(payload_out)) + payload_out)
            conn.seat_host = host_id            # this conn IS the beacon now
            conn.seat_tag = tag                 # tag survives a round reset
            conn.seated_at = now                # for the LEFT-grace window
            self.seat_beacons[host_id] = conn
            # issue #34: remember who sat here (IP + callsign) so the same
            # player's claim-less restart reclaims THIS seat, never a new one
            if payload:
                _cs = (payload.split(b"\0")[0]
                       .decode("ascii", "replace").strip()[:15])
                if _cs:
                    if not hasattr(self, "seat_identity"):
                        self.seat_identity = {}
                    self.seat_identity[tag] = (conn.sock.getpeername()[0], _cs)
            print(f"[relay] {conn.name()} seat request -> assigned host "
                  f"{host_id} '{tag}' (reserved {SEAT_RESERVE_SECONDS:.0f}s)"
                  f"{pref_note}", flush=True)
            cs, mech = self.seat_prefs.get(host_id, ("", ""))
            print(f"[relay] PLAYER {host_id - FIRST_GAME_HOST_ID + 1} SEATED "
                  f"(host {host_id}) tag='{tag}' "
                  f"callsign='{cs}' mech='{mech}' "
                  f"({len(self.seat_beacons)} seated)", flush=True)
            return
        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:
                # issue #33: a plain drop bounced the client straight into a
                # relaunch loop (stale identity -> reject -> relaunch -> same
                # stale identity, ~1/s).  Tell it to REJOIN instead (the
                # rejoin path re-requests a seat = identity resync), rate-
                # limited per IP so a truly broken client can't spin fast.
                ip = conn.sock.getpeername()[0] if conn.sock else "?"
                if not hasattr(self, "hello_reject_at"):
                    self.hello_reject_at = {}
                now = time.time()
                if now - self.hello_reject_at.get(ip, 0) > 5.0:
                    self.hello_reject_at[ip] = now
                    try:
                        self._send_raw(conn, struct.pack(ENV_FMT_TCP,
                                                         ROUTE_REJOIN, 0))
                    except OSError:
                        pass
                self._drop_game(conn, f"HELLO hostID {host_id} not in roster "
                                      f"{sorted(self.expected_ids)} -- "
                                      f"REJOIN sent (identity resync)")
                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
            self.seat_reservations.pop(host_id, None)   # claim clears reserve
            # tag='...' rides along for the GUI: its positional hostID->tag map
            # is computed against the UNTRIMMED egg roster, so after a LAUNCH
            # trim remaps seat ids the GUI would light (and, since the roster
            # display fix, clear) the WRONG row.  The relay's roster here is the
            # live -- possibly trimmed -- one, so this tag is the true identity.
            # Stash the tag AT REGISTRATION: _drop_game resolves against the
            # LIVE roster, and a trimmed-round conn dying after a re-arm restored
            # the full roster would otherwise be blamed on the wrong tag -- and
            # the tag-trusting GUI would clear the wrong seat (review 2026-07-26).
            conn.reg_tag = self._tag_of(host_id)
            print(f"[relay] {conn.name()} REGISTERED "
                  f"({len(self.by_host)}/{len(self.roster)}) "
                  f"tag='{conn.reg_tag}'", flush=True)
            if len(self.by_host) >= len(self.roster):
                self._release_eggs()        # fallback; normally already fired
            # 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 list(self.by_host.values()):   # snapshot: a failed
                # send drops that peer INSIDE the loop (mutates by_host)
                if other is not conn:
                    self._send_control(other, ROUTE_PEER_UP, host_id)
            return
        if route == ROUTE_READY:
            conn.ready = True
            ready_count = sum(1 for c in self.by_host.values()
                              if getattr(c, "ready", False))
            i = conn.host_id - FIRST_GAME_HOST_ID
            tag = self.roster[i] if 0 <= i < len(self.roster) else "?"
            print(f"[relay] PLAYER {conn.host_id - FIRST_GAME_HOST_ID + 1} "
                  f"READY (host {conn.host_id}) tag='{tag}' "
                  f"({ready_count}/{len(self.by_host)} loaded)", flush=True)
            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:
            _send_all_guarded(conn.sock, data)
            self.stats["tcp_tx"] += 1
        except OSError as e:
            self._drop_game(conn, f"send failed {e!r}")

    def _abort_round(self, why):
        """ROUND ABORT (connection audit 2026-07-23): a registered pod died
        AFTER egg release but BEFORE launch -- the engine's all-connections
        gate would leave every survivor waiting on it forever.  Bounce all
        remaining pods back to their seats (REJOIN) and reset the round; the
        reclaim window puts everyone back in the same seats in seconds.

        STORM HARDENING (issue #33, night-2 capture): the first abort's
        rejoiners reconnected + re-ACKed within a second, RE-releasing eggs
        while a flapping pod was still bouncing -- its next drop aborted the
        NEW round, cascading (3 aborts in 8 s, every client relaunch-churning
        into the 'Could not create a Direct3D device' 20 s deadline).  Two
        dampers: stale console-conn ACK flags are cleared (they counted
        toward the next release), and egg release is HELD for a settle
        window after any abort so the round re-forms only once the flapping
        stops."""
        if not self.eggs_released or self.launches_sent >= 2:
            return
        print(f"[relay] ROUND ABORT ({why}) -- sending every pod back to "
              f"the join wait for a fresh round "
              f"(egg release held {ABORT_SETTLE_SECONDS:.0f}s)", flush=True)
        # reset FIRST so the cascade of resulting drops can't re-trigger
        self.eggs_released = False
        self.launch_at = None
        self.launch_requested = False
        self.eggs_done_at = None
        self._launch_blocked_warned = False
        self.round_hold_until = time.time() + ABORT_SETTLE_SECONDS
        for c in self.console_conns:
            c.acked = False         # an aborted round's ACK must not count
            c.egg_sent = False      # toward re-releasing the next one
        survivors = list(self.by_host.values())
        for conn in survivors:
            try:
                self._send_raw(conn, struct.pack(ENV_FMT_TCP, ROUTE_REJOIN, 0))
            except OSError:
                pass
        # the full state restore happens in _maybe_reset_round as they drop

    def _drop_game(self, conn, why):
        # tag='...' for registered conns, same reason as the REGISTERED print:
        # the GUI must not resolve a remapped post-trim hostID positionally.
        drop_tag = ""
        if conn.host_id is not None:
            _dt = getattr(conn, "reg_tag", None) or self._tag_of(conn.host_id)
            drop_tag = " tag='%s'" % _dt
        print(f"[relay] {conn.name()} dropped: {why}{drop_tag}", 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)
        seat = getattr(conn, "seat_host", None)
        if seat is not None and self.seat_beacons.get(seat) is conn:
            del self.seat_beacons[seat]
            tag_for_seat = getattr(conn, "seat_tag", None)
            if tag_for_seat:                    # hold for the returning player
                self.seat_reclaim[tag_for_seat] = time.time() + 90.0
            in_grace = (time.time() - getattr(conn, "seated_at", 0.0)
                        < SEAT_LEFT_GRACE_SECONDS)
            if seat in self.by_host or in_grace:
                # Claimed seat, or the DESIGNED post-assignment close (the
                # 2026-07-26 identity swap: SAURON re-seated at round end, his
                # request conn closed to re-dial for the HELLO, and this
                # branch called that a LEAVE -- popping the pref written
                # milliseconds earlier.  The egg release then resurrected the
                # DISK egg's stale callsign, and he played the round as
                # 'Draco'.)  A pod's post-assign re-dial lands within seconds;
                # a real join-menu leaver has held the seat far longer.  Keep
                # the pref + the reservation inside the grace and print
                # nothing -- the GUI would blink a false LEFT.
                pass
            else:                               # never claimed: player left
                self.seat_reservations.pop(seat, None)
                self.seat_prefs.pop(seat, None)
                i = seat - FIRST_GAME_HOST_ID
                tag = self.roster[i] if 0 <= i < len(self.roster) else "?"
                print(f"[relay] PLAYER {seat - FIRST_GAME_HOST_ID + 1} LEFT "
                      f"(host {seat}) tag='{tag}' "
                      f"({len(self.seat_beacons)} seated)", flush=True)
        was_registered = (conn.host_id is not None
                          and self.by_host.get(conn.host_id) is 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 list(self.by_host.values()):   # snapshot: a failed
                # send drops that peer INSIDE the loop (mutates by_host)
                self._send_control(other, ROUTE_PEER_DOWN, conn.host_id)
        if was_registered:
            self._abort_round(f"host {conn.host_id} dropped mid-load")
        self._maybe_reset_round()

    # ---------------- 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
            live = self.by_host.get(from_host)
            #
            # ANTI-HIJACK (2026-07-26).  `from_host` is the sender's OWN claim
            # about who it is, and the next line rewrites that host's downstream
            # endpoint.  Unchecked, ANY datagram claiming host N silently steals
            # host N's traffic -- a zombie pod process from a previous round, a
            # stale NAT mapping, or a third party who guessed a host id.  The
            # victim then just stops receiving, on a channel that looks healthy.
            #
            # Bind the claim to the identity we actually authenticated: the IP of
            # that host's live TCP game connection.  The PORT is deliberately not
            # checked -- it moves on a NAT rebind, which is the whole reason the
            # endpoint map is refreshed per datagram.  A genuine IP change cannot
            # happen without the TCP connection breaking and being re-registered
            # with the new address, so this never rejects a legitimate pod.
            #
            if live is not None and endpoint[0] != live.addr[0]:
                self.stats["udp_spoofed"] = self.stats.get("udp_spoofed", 0) + 1
                if not hasattr(self, "_udp_spoof_warned"):
                    self._udp_spoof_warned = set()
                key = (from_host, endpoint[0])
                if len(self._udp_spoof_warned) > 256:
                    self._udp_spoof_warned.clear()      # bounded under a flood
                if key not in self._udp_spoof_warned:
                    self._udp_spoof_warned.add(key)
                    print(f"[relay] UDP from {endpoint[0]}:{endpoint[1]} CLAIMS "
                          f"host {from_host}, whose TCP is {live.addr[0]} -- "
                          f"DROPPED (it would have stolen that pod's downstream)",
                          flush=True)
                continue
            self.stats["udp_rx"] += 1
            # NAT-rebind tolerant: every datagram refreshes the endpoint map.
            self.udp_endpoint[from_host] = endpoint
            # ... and counts as LIVENESS.  Without this, a pod streaming updates
            # over UDP while its TCP sits idle looks "silent" to the reaper.
            if live is not None:
                live.last_seen = time.time()
            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

    # ---------------- LAN auto-discovery ----------------

    def _disc_read(self):
        while True:
            try:
                data, endpoint = self.disc_sock.recvfrom(64)
            except (BlockingIOError, InterruptedError, OSError):
                return
            if data == DISC_PROBE:
                try:
                    self.disc_sock.sendto(
                        DISC_REPLY + struct.pack("<H", self.console_port),
                        endpoint)
                    print(f"[relay] discovery probe from "
                          f"{endpoint[0]}:{endpoint[1]} -> answered "
                          f"(console port {self.console_port})", flush=True)
                except OSError:
                    pass

    # ---------------- stats ----------------

    # ---------------- remote operator control ----------------

    def _ctl_accept(self, listener):
        try:
            sock, addr = listener.accept()
            sock.setblocking(False)
            conn = RelayControlConn(sock, addr)
            self.ctl_conns.append(conn)
            self.sel.register(sock, selectors.EVENT_READ, ("ctl", conn))
            print(f"[ctl] operator connection from {conn.name()} "
                  f"(awaiting AUTH)", flush=True)
        except OSError:
            pass

    def _ctl_drop(self, conn, why):
        try:
            self.sel.unregister(conn.sock)
        except (KeyError, ValueError):
            pass
        try:
            conn.sock.close()
        except OSError:
            pass
        with _CONTROL_LOCK:
            if conn in _CONTROL_SINKS:
                _CONTROL_SINKS.remove(conn)
        if conn in self.ctl_conns:
            self.ctl_conns.remove(conn)
        if conn.authed or why != "auth timeout":
            print(f"[ctl] {conn.name()} dropped: {why}", flush=True)

    def _ctl_read(self, conn):
        try:
            data = conn.sock.recv(4096)
        except (BlockingIOError, InterruptedError):
            return
        except OSError:
            self._ctl_drop(conn, "closed")
            return
        if not data:
            self._ctl_drop(conn, "closed")
            return
        conn.buf += data
        if len(conn.buf) > 8192:
            self._ctl_drop(conn, "oversize command")
            return
        while b"\n" in conn.buf:
            line, conn.buf = conn.buf.split(b"\n", 1)
            try:
                self._ctl_command(conn, line.decode("utf-8", "replace").strip())
            except Exception as e:               # a control bug must NEVER
                print(f"[ctl] command error: {e!r}", flush=True)   # kill games

    def _ctl_command(self, conn, line):
        if not line:
            return
        if not conn.authed:
            parts = line.split(None, 1)
            if len(parts) == 2 and parts[0] == "AUTH" \
                    and parts[1].strip() == self.ctl_secret:
                # one operator at a time: the newest displaces the rest
                for other in list(self.ctl_conns):
                    if other is not conn and other.authed:
                        other.queue_out(b"[ctl] displaced by "
                                        + conn.name().encode() + b"\n")
                        other.flush_out()
                        self._ctl_drop(other, "displaced by new operator")
                conn.authed = True
                try:
                    conn.sock.send(("OK bt-relay-control 1 console=%d\n"
                                    % self.console_port).encode())
                except OSError:
                    conn.dead = True
                # replay recent history (bounded), THEN subscribe live --
                # under the lock so no line is lost or duplicated between
                # the replay and the live stream
                with _CONTROL_LOCK:
                    # The GUI adopts its roster from the startup "roster:" line,
                    # which is printed ONCE and ages out of the 400-line history
                    # in ~33 minutes of stats chatter -- an operator AUTHing
                    # late into the night got a permanently blank roster and a
                    # dead LAUNCH button (review 2026-07-26).  Re-issue the LIVE
                    # roster in the same adoptable format before the replay.
                    conn.queue_out(
                        (f"[relay] roster: {len(self.roster)} pilot(s) -> "
                         f"hostIDs {sorted(self.expected_ids)}: "
                         f"{self.roster}\n").encode("utf-8", "replace"))
                    conn.queue_out(b"[ctl] --- session history replay ---\n")
                    for past in _CONTROL_HISTORY:
                        conn.queue_out(past)
                    conn.queue_out(b"[ctl] --- live ---\n")
                    _CONTROL_SINKS.append(conn)
                print(f"[ctl] operator AUTHENTICATED: {conn.name()} "
                      f"(driving this relay)", flush=True)
            else:
                self._ctl_drop(conn, "bad auth")
            return
        cmd = line.split(None, 1)[0].lower()
        if cmd == "launch":
            print(f"[ctl] LAUNCH from {conn.name()} "
                  f"(launches_sent={self.launches_sent} "
                  f"eggs_released={self.eggs_released} "
                  f"pods_acked={self._pods_ready()}/{len(self.roster)})",
                  flush=True)
            self.launch_requested = True
        elif cmd == "stop":
            print(f"[ctl] STOP from {conn.name()}", flush=True)
            self.stop_requested = True
        elif cmd in ("rearm", "newround"):
            # Explicit escape hatch: recover the launcher without restarting the
            # session.  Before this existed, killing the relay was the only way
            # out of a latched round (operator report 2026-07-25).
            print(f"[ctl] RE-ARM from {conn.name()}", flush=True)
            self._rearm_for_new_round("operator re-arm command")
        elif cmd == "restart":
            # Parked-relay deployment: the operator is not at the machine, so
            # this is their only route to a fresh relay (roster resize, or a
            # wedge nothing else clears).  Refused mid-mission -- it would
            # drop every pod out of a running round.
            if self.launches_sent >= 2 and not self.stop_sent:
                conn.queue_out(b"[ctl] restart REFUSED -- a mission is "
                               b"running; press End Mission first\n")
                print("[ctl] restart refused (mission running)", flush=True)
            else:
                conn.queue_out(b"[ctl] restarting -- reconnect in a few "
                               b"seconds\n")
                conn.flush_out()
                print(f"[ctl] RESTART from {conn.name()}", flush=True)
                self.restart_requested = True
        elif cmd == "ping":
            conn.queue_out(b"[ctl] pong\n")
        elif cmd == "get":
            self._ctl_get_mission(conn)
        elif cmd == "set":
            self._ctl_set_mission(conn, line[3:].strip())
        else:
            conn.queue_out(("[ctl] unknown command: %s\n" % cmd).encode())

    def _ctl_get_mission(self, conn):
        try:
            doc = eggmodel.EggDoc.load(self.orig_egg_path)
            m = doc.mission()
            print("[ctl] mission map=%s time=%s weather=%s scenario=%s "
                  "temperature=%s length=%s seats=%d"
                  % (m.get("map", ""), m.get("time", ""),
                     m.get("weather", ""), m.get("scenario", ""),
                     m.get("temperature", ""), m.get("length", ""),
                     len(self.roster)), flush=True)
        except Exception as e:
            print(f"[ctl] get mission failed: {e!r}", flush=True)

    def _ctl_set_mission(self, conn, spec):
        if eggmodel is None:
            print("[ctl] set rejected: eggmodel unavailable", flush=True)
            return
        kw = {}
        for pair in spec.split(";"):
            pair = pair.strip()
            if not pair or "=" not in pair:
                continue
            k, v = pair.split("=", 1)
            k = k.strip().lower()
            v = v.strip()
            if k in CONTROL_SET_KEYS and v:
                kw[k] = v
        if not kw:
            print("[ctl] set: nothing applied (allowed keys: %s)"
                  % ",".join(sorted(CONTROL_SET_KEYS)), flush=True)
            return
        try:
            doc = eggmodel.EggDoc.load(self.orig_egg_path)
            doc.set_mission(**kw)
            doc.save(self.orig_egg_path)
            print("[ctl] mission settings applied by %s: %s -- takes effect "
                  "next round" % (conn.name(),
                                  " ".join("%s=%s" % kv for kv in
                                           sorted(kw.items()))), flush=True)
        except Exception as e:
            print(f"[ctl] set failed: {e!r}", flush=True)

    def _tick_control(self):
        now = time.time()
        for conn in list(self.ctl_conns):
            if conn.dead:
                self._ctl_drop(conn, "send buffer overflow/socket error")
            elif (not conn.authed
                    and now - conn.connected_at > CONTROL_AUTH_TIMEOUT):
                self._ctl_drop(conn, "auth timeout")
            else:
                conn.flush_out()

    def _tick_settle(self):
        # issue #33: the post-abort settle hold blocks egg release while the
        # rejoin flapping calms; once it expires, re-run the release check
        # (the ACKs that arrived DURING the hold produced no later event to
        # trigger it).
        hold = getattr(self, "round_hold_until", 0)
        if hold and time.time() >= hold and not self.eggs_released:
            self.round_hold_until = 0
            waiting = [c for c in self.console_conns if not c.egg_sent]
            if waiting and len(self.console_conns) >= len(self.roster):
                print("[relay] abort settle window over -- releasing eggs to "
                      f"{len(waiting)} waiting pod(s)", flush=True)
                self._release_eggs()

    def _tick_stats(self):
        if time.time() < self.stats_at:
            return
        self.stats_at = time.time() + STATS_PERIOD
        s = self.stats
        # spoofed rides along only when non-zero: the anti-hijack counter was
        # invisible (never printed anywhere), so an actual hijack attempt left
        # no operator-visible trace beyond the one-shot warning line.
        spoofed = s.get("udp_spoofed", 0)
        spoof_part = f", spoofed {spoofed}" if spoofed else ""
        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"{spoof_part}) | "
              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]
    [--manual-launch]"""
    args = argv[:]
    bind_addr = "0.0.0.0"
    udp_drop = 0.0
    manual = False
    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 "--manual-launch" in args:
        manual = True
        args.remove("--manual-launch")
    reserved_tags = ()
    if "--reserve" in args:
        i = args.index("--reserve")
        reserved_tags = tuple(t for t in args[i + 1].split(",") if t)
        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, manual,
                  reserved_tags=reserved_tags)
    if not relay.roster:
        print(f"[relay] ERROR: no [pilots] entries in {egg_path}")
        return 2
    # Operator command channel: a daemon thread reads stdin.  'launch' fires
    # the mission (manual mode only); 'stop' ends it early (End Mission).
    def stdin_reader():
        if sys.stdin is None:               # pythonw / broken shim pipeline
            print("[relay] no stdin -- operator commands unavailable",
                  flush=True)
            return
        for line in sys.stdin:
            command = line.strip().lower()
            if command == "launch":
                # ALWAYS record the press at RECEIPT.  Every wedge in the launch
                # state machine used to be silent, so the operator could not
                # tell "the command never arrived" from "the relay ignored it".
                print(f"[relay] operator LAUNCH command received "
                      f"(launches_sent={relay.launches_sent} "
                      f"eggs_released={relay.eggs_released})", flush=True)
                relay.launch_requested = True
            elif command == "stop":
                print("[relay] operator STOP command received", flush=True)
                relay.stop_requested = True
            elif command in ("rearm", "newround"):
                print("[relay] operator RE-ARM command received", flush=True)
                relay._rearm_for_new_round("operator re-arm command")
            elif command == "restart":
                print("[relay] operator RESTART command received", flush=True)
                relay.restart_requested = True
            elif command:
                print(f"[relay] unknown operator command: {command!r}",
                      flush=True)
    threading.Thread(target=stdin_reader, daemon=True).start()
    try:
        relay.run()
    except KeyboardInterrupt:
        print("[relay] shutting down")
        return 0
    if relay.restart_requested:
        return RESTART_EXIT_CODE       # the supervisor's cue to relaunch
    return 0


class _StampedOut:
    """Line-timestamping stdout wrapper: every log line gets a wall-clock
    prefix so post-mortems can measure delays instead of guessing (operator
    request 2026-07-22).  The GUI's monitor regexes all use .search(), so
    the prefix is transparent to them."""

    def __init__(self, inner):
        self._inner = inner
        self._at_line_start = True

    def write(self, text):
        out = []
        for ch in text:
            if self._at_line_start and ch != "\n":
                out.append(time.strftime("%H:%M:%S "))
                self._at_line_start = False
            out.append(ch)
            if ch == "\n":
                self._at_line_start = True
        joined = "".join(out)
        self._inner.write(joined)
        try:
            _control_tee(joined)
        except Exception:
            pass                                # the tee must never kill logs

    def flush(self):
        self._inner.flush()

    def __getattr__(self, name):
        return getattr(self._inner, name)


def main():
    sys.stdout = _StampedOut(sys.stdout)
    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
    egg_bytes = egg_wire_bytes(open(sys.argv[1], "rb").read())
    pods = sys.argv[2:]
    threads = [threading.Thread(target=serve_pod, args=(p, egg_bytes), daemon=True)
               for p in pods]
    for t in threads:
        t.start()
    try:
        while any(t.is_alive() for t in threads):
            time.sleep(0.5)
    except KeyboardInterrupt:
        print("console shutting down")
    return 0


if __name__ == "__main__":
    sys.exit(main())