"""Can a 554 client still play against tonight's build? (2026-07-26) The operator's testers are on BT411_4.11.554. Tonight's build carries Cyd's cockpit-refit merge, so before telling anyone "you don't need to update", prove it: one pod from the OLD zip and one from the NEW build, on the same relay, through a real launch. Pass = both REGISTER, the relay launches, and NEITHER pod drops during the round. Fail = a mismatch that would strand half the players tonight. Usage: python scratchpad/test_mixed_version.py [path-to-old-554-root] Spawns its own relay + pods; kills only that tree. """ import os import re import shutil import subprocess import sys import time REPO = r"C:\git\bt411" CONTENT = os.path.join(REPO, "content") SCRATCH = os.environ.get("TEMP", ".") OLD_ROOT = (sys.argv[1] if len(sys.argv) > 1 else os.path.join(SCRATCH, "claude", "C--git-bt411", "89ab96e9-6cf0-4555-939d-05bf3708745a", "scratchpad", "old554", "BT411_4.11.554")) PORT = 1500 SRC_EGG = "FOGDAY.EGG" EGG = "_mixed_test.EGG" RELAY_LOG = os.path.join(SCRATCH, "mixed_relay.log") fails = [] procs = [] def check(label, ok, extra=""): print(" %-52s %s %s" % (label, "OK" if ok else "*** FAIL ***", extra)) if not ok: fails.append(label) def relay_text(): try: with open(RELAY_LOG, encoding="utf-8", errors="replace") as f: return f.read() except OSError: return "" def wait_for(pattern, timeout): end = time.time() + timeout while time.time() < end: if re.search(pattern, relay_text()): return True time.sleep(0.25) return False def kill_tree(): ours = {p.pid for p in procs} try: out = subprocess.run(["wmic", "process", "get", "ProcessId,ParentProcessId"], capture_output=True, text=True).stdout pairs = [] for line in out.splitlines(): b = line.split() if len(b) == 2 and b[0].isdigit() and b[1].isdigit(): pairs.append((int(b[0]), int(b[1]))) for _ in range(4): for parent, pid in pairs: if parent in ours: ours.add(pid) except Exception: pass for p in procs: try: if p.poll() is None: p.kill() except Exception: pass for pid in ours: subprocess.run(["taskkill", "/PID", str(pid), "/F", "/T"], capture_output=True) try: old_exe = os.path.join(OLD_ROOT, "build", "Release", "btl4.exe") new_exe = os.path.join(REPO, "build", "Release", "btl4.exe") old_content = os.path.join(OLD_ROOT, "content") for p in (old_exe, new_exe, old_content): if not os.path.exists(p): sys.exit("missing: %s" % p) if "btl4.exe" in subprocess.run( ["tasklist", "/FI", "IMAGENAME eq btl4.exe", "/NH"], capture_output=True, text=True).stdout: sys.exit("ABORT: btl4 already running") print("OLD pod: %s" % old_exe) print("NEW pod: %s" % new_exe) # the SAME egg for both sides (copy into each content dir) shutil.copyfile(os.path.join(CONTENT, SRC_EGG), os.path.join(CONTENT, EGG)) shutil.copyfile(os.path.join(CONTENT, SRC_EGG), os.path.join(old_content, EGG)) log = open(RELAY_LOG, "w", encoding="utf-8") procs.append(subprocess.Popen( [sys.executable, "-u", os.path.join(REPO, "tools", "btconsole.py"), "--relay", str(PORT), os.path.join(CONTENT, EGG), "--bind", "127.0.0.1", "--manual-launch"], cwd=CONTENT, stdout=log, stderr=subprocess.STDOUT)) check("relay up", wait_for(r"console 127\.0\.0\.1:%d" % PORT, 20)) base = {"BT_START_INSIDE": "1", "BT_DEV_GAUGES": "1", "BT_MATCHLOG": "0", "BT_RELAY": "127.0.0.1:%d" % PORT} # seat 1 = the OLD 554 pod, seat 2 = tonight's build for label, exe, cwd, netport in ( ("554(old)", old_exe, old_content, 1502), ("584(new)", new_exe, CONTENT, 1602)): e = dict(os.environ) e.update(base) e["BT_LOG"] = "mixed_%s.log" % netport e["BT_SELF"] = "127.0.0.1:%d" % netport procs.append(subprocess.Popen( [exe, "-egg", EGG, "-net", str(netport)], cwd=cwd, env=e)) print(" started %s on :%d" % (label, netport)) time.sleep(3) check("BOTH pods staged (old + new on one relay)", wait_for(r"WAITING FOR OPERATOR", 120)) regs = len(re.findall(r"REGISTERED", relay_text())) check("both REGISTERED", regs >= 2, "registered=%d" % regs) # launch a real round import socket with open(os.path.join(CONTENT, "operator_secret.txt")) as f: secret = f.read().strip() s = socket.create_connection(("127.0.0.1", PORT + 7), timeout=10) s.sendall(("AUTH %s\n" % secret).encode()) time.sleep(0.6) s.sendall(b"launch\n") time.sleep(1) s.close() check("relay launched the mixed round", wait_for(r"RunMission #2 sent to 2 pod", 60)) # both must SURVIVE the round -- a wire mismatch shows up as a drop print(" holding the round for 30s to watch for drops...") time.sleep(30) # Match REAL pod drops only. A bare "dropped" also matches the stats # line's own counter -- "(dropped 0, tcp-fallback 0)" -- and the operator's # control socket closing normally, which cost one false failure here. txt = relay_text() drops = re.findall(r"game\[[^\]]*\] dropped|PLAYER \d+ LEFT", txt) check("neither pod dropped during the round", len(drops) == 0, "drops=%r" % (drops[:3],)) last = txt.rstrip().splitlines()[-1] if txt.strip() else "" m = re.search(r"registered \[([^\]]*)\] udp-known \[([^\]]*)\]", txt[::-1][:0] + txt) reg = re.findall(r"registered \[([^\]]*)\] udp-known \[([^\]]*)\]", txt) check("both hosts still registered AND on UDP at the end", bool(reg) and reg[-1][0].count(",") == 1 and reg[-1][1].count(",") == 1, "last stats: registered [%s] udp-known [%s]" % (reg[-1] if reg else ("?", "?"))) alive = sum(1 for p in procs[1:] if p.poll() is None) check("both pod processes still alive", alive == 2, "alive=%d/2" % alive) finally: kill_tree() for junk in (os.path.join(CONTENT, EGG), os.path.join(old_content, EGG)): try: os.remove(junk) except OSError: pass print() print("WIRE-COMPATIBLE with 554" if not fails else "FAILURES: %s" % fails) sys.exit(1 if fails else 0)