Remote operator control channel + THE teardown crash root-caused and fixed

REMOTE OPERATORS (dev-channel ask): the relay grows a control TCP
listener (console port + 7) speaking a line protocol -- AUTH <secret>
(auto-generated content\operator_secret.txt), then launch / stop /
ping / get mission / set key=value;...  The relay's timestamped log
stream tees to the authenticated operator (bounded per-conn buffers --
a stalled operator can NEVER stall the relay; overflow = drop), with a
400-line history replay on connect so mid-session operators see
current state.  One operator at a time; a new AUTH displaces the old.
stdin commands unchanged (rigs/GUI local mode untouched).

The operator GUI gains a 'Relay host' field: blank = today's local
flow byte-for-byte (child relay dies with the window); a hostname =
drive that machine's relay remotely (launch/stop/apply-settings over
the wire, roster rebuilt from the teed log, local-instance joins point
at the remote relay).  Only the relay host needs port forwarding.

Scripted protocol test: auth reject/accept, takeover, get/set, ping,
history replay, and a FULL 2-node mission launched and stopped
entirely over the socket.

CRASH FIX (the layout-shifting heisenbug -- and almost certainly issue
#35): the new crash self-report finally captured its 24-frame stack:
InterestManager::OrphanInterestOrigin -> RemoveUninterestingEntity ->
DestroyEntityAudioObjects -> {Static,Dynamic}3DPatchSource::
IsAudioSourceClipped, AV reading NULL+0x38 -- the unguarded chain
GetMissionPlayer()->GetPlayerVehicle()->GetSimulationState() (the
authentic burning-mech death-silence check) hit a NULL vehicle during
MP teardown windows.  Both sites now null-guard; burning semantics
preserved whenever the vehicle exists.  Also live-confirms that
interest-based entity teardown RUNS in MP (the #38 paint mechanism).
Regression: full combat round + control-channel drive, zero crashes.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

engine/MUNGA_L4/L4AUDIO.cpp

@@ -1189,13 +1189,31 @@ Dynamic3DPatchSource::Dynamic3DPatchSource(
 Logical
 Dynamic3DPatchSource::IsAudioSourceClipped(AudioHead *audio_head)
 {
-	if (AudioSource::IsAudioSourceClipped(audio_head) || l4_application->GetMissionPlayer()->GetPlayerVehicle()->GetSimulationState() == VTV::BurningState)
+	if (AudioSource::IsAudioSourceClipped(audio_head))
 	{
 		return true;
-	} else
-	{
-		return false;
 	}
+	//
+	// NULL-GUARD (2026-07-24, the layout-shifting teardown crash finally
+	// stack-captured: InterestManager::OrphanInterestOrigin ->
+	// RemoveUninterestingEntity -> DestroyEntityAudioObjects -> here, AV
+	// reading NULL+0x38).  The binary could assume the mission player
+	// always has a vehicle (in a pod you always do); the port's MP join/
+	// teardown/respawn windows briefly leave it NULL.  The burning check
+	// (death-silence: your burning mech clips world audio) keeps its
+	// authentic behavior whenever the vehicle exists.
+	//
+	{
+		Player *mission_player = l4_application->GetMissionPlayer();
+		if (mission_player != 0
+			&& mission_player->GetPlayerVehicle() != 0
+			&& mission_player->GetPlayerVehicle()->GetSimulationState()
+				== VTV::BurningState)
+		{
+			return true;
+		}
+	}
+	return false;
 }
 
 //
@@ -1593,13 +1611,31 @@ Static3DPatchSource::Static3DPatchSource(
 
 Logical	Static3DPatchSource::IsAudioSourceClipped(AudioHead *audio_head)
 {
-	if (AudioSource::IsAudioSourceClipped(audio_head) || l4_application->GetMissionPlayer()->GetPlayerVehicle()->GetSimulationState() == VTV::BurningState)
+	if (AudioSource::IsAudioSourceClipped(audio_head))
 	{
 		return true;
-	} else
-	{
-		return false;
 	}
+	//
+	// NULL-GUARD (2026-07-24, the layout-shifting teardown crash finally
+	// stack-captured: InterestManager::OrphanInterestOrigin ->
+	// RemoveUninterestingEntity -> DestroyEntityAudioObjects -> here, AV
+	// reading NULL+0x38).  The binary could assume the mission player
+	// always has a vehicle (in a pod you always do); the port's MP join/
+	// teardown/respawn windows briefly leave it NULL.  The burning check
+	// (death-silence: your burning mech clips world audio) keeps its
+	// authentic behavior whenever the vehicle exists.
+	//
+	{
+		Player *mission_player = l4_application->GetMissionPlayer();
+		if (mission_player != 0
+			&& mission_player->GetPlayerVehicle() != 0
+			&& mission_player->GetPlayerVehicle()->GetSimulationState()
+				== VTV::BurningState)
+		{
+			return true;
+		}
+	}
+	return false;
 }
 
 //

tools/btconsole.py

@@ -58,6 +58,7 @@ BT_SELF=<their [pilots] entry>.
     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
@@ -241,6 +242,63 @@ ABORT_SETTLE_SECONDS = 8.0     # issue #33: egg-release hold after a round abort
 # '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.
+CONTROL_PORT_OFFSET = 7            # control port = console port + 7
+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
@@ -307,6 +365,43 @@ class RelayConsoleConn:
         # 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):
@@ -380,6 +475,8 @@ class Relay:
         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.ctl_secret = control_secret()
 
     # ---------------- lifecycle ----------------
 
@@ -394,6 +491,18 @@ class Relay:
         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:
@@ -427,6 +536,11 @@ class Relay:
                     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._tick_launch()
             self._tick_stop()
             self._tick_settle()
@@ -1281,6 +1395,162 @@ class Relay:
 
     # ---------------- 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:
+                    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()}", flush=True)
+            self.launch_requested = True
+        elif cmd == "stop":
+            print(f"[ctl] STOP from {conn.name()}", flush=True)
+            self.stop_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
@@ -1380,7 +1650,12 @@ class _StampedOut:
             out.append(ch)
             if ch == "\n":
                 self._at_line_start = True
-        self._inner.write("".join(out))
+        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()

tools/btoperator.py

@@ -20,9 +20,11 @@ with the CLI and the test harnesses.
 """
 import os
 import re
+import socket
 import sys
+import threading
 
-from PySide6.QtCore import Qt, QProcess, QProcessEnvironment
+from PySide6.QtCore import Qt, QProcess, QProcessEnvironment, QThread, Signal
 from PySide6.QtGui import QColor, QFont, QPalette, QAction
 from PySide6.QtWidgets import (
     QApplication, QCheckBox, QComboBox, QFileDialog, QFormLayout, QGroupBox,
@@ -195,6 +197,69 @@ class SessionMonitor:
 # Main window
 # ---------------------------------------------------------------------------
 
+
+class RemoteRelayLink(QThread):
+    """Operator control-channel client (2026-07-24): drives a RELAY RUNNING
+    ON ANOTHER MACHINE over the btconsole control protocol (AUTH, then the
+    relay's log stream in; launch/stop/set commands out).  The reader runs
+    in this thread; send() is called from the GUI thread -- socket sendall
+    on small lines is safe cross-thread, guarded anyway."""
+
+    lineReceived = Signal(str)
+    closed = Signal(str)
+
+    def __init__(self, host, port, secret):
+        super().__init__()
+        self.host = host
+        self.port = port
+        self.secret = secret
+        self.sock = None
+        self._send_lock = threading.Lock()
+        self._stopping = False
+
+    def run(self):
+        try:
+            self.sock = socket.create_connection((self.host, self.port),
+                                                 timeout=8)
+            self.sock.sendall(("AUTH %s\n" % self.secret).encode())
+            self.sock.settimeout(None)
+            buf = b""
+            while not self._stopping:
+                data = self.sock.recv(4096)
+                if not data:
+                    break
+                buf += data
+                while b"\n" in buf:
+                    line, buf = buf.split(b"\n", 1)
+                    text = line.decode("utf-8", "replace").rstrip()
+                    if text:
+                        self.lineReceived.emit(text)
+            self.closed.emit("connection closed by relay"
+                             if not self._stopping else "disconnected")
+        except Exception as e:
+            self.closed.emit("connect/read failed: %r" % e)
+
+    def send(self, text):
+        with self._send_lock:
+            if self.sock is None:
+                return False
+            try:
+                self.sock.sendall((text.rstrip() + "\n").encode())
+                return True
+            except OSError:
+                return False
+
+    def stop(self):
+        self._stopping = True
+        with self._send_lock:
+            if self.sock is not None:
+                try:
+                    self.sock.close()
+                except OSError:
+                    pass
+                self.sock = None
+
+
 class Operator(QMainWindow):
     COLS = ["Identity / Address", "Callsign", "Vehicle", "Color", "Badge",
             "Patch", "Experience", "Local"]
@@ -207,6 +272,7 @@ class Operator(QMainWindow):
         self.egg = eggmodel.new_from_template()
         self.egg_path = DEFAULT_EGG
         self.console_proc = None
+        self.remote_link = None
         self.game_procs = []
         self.monitor = None
         self._build_ui()
@@ -226,6 +292,29 @@ class Operator(QMainWindow):
         self.mode.addItems(["Relay (internet)", "Mesh (LAN legacy)"])
         self.mode.currentIndexChanged.connect(self._mode_changed)
         top.addWidget(self.mode)
+        # REMOTE OPERATOR (2026-07-24): leave blank to run the relay HERE
+        # (unchanged local flow -- relay dies with this window).  Enter a
+        # host to DRIVE a relay running on that machine instead (needs its
+        # content\operator_secret.txt value; only the relay host forwards
+        # ports).
+        top.addSpacing(12)
+        top.addWidget(QLabel("Relay host:"))
+        self.f_relayhost = QLineEdit()
+        self.f_relayhost.setPlaceholderText("blank = run relay here")
+        self.f_relayhost.setFixedWidth(150)
+        self.f_relayhost.setToolTip(
+            "Blank/localhost: start the relay on this machine (it stops when "
+            "you close this window).\nA hostname/IP: connect to a relay "
+            "already running there and operate it remotely.")
+        top.addWidget(self.f_relayhost)
+        top.addWidget(QLabel("Secret:"))
+        self.f_secret = QLineEdit()
+        self.f_secret.setPlaceholderText("remote only")
+        self.f_secret.setFixedWidth(120)
+        self.f_secret.setToolTip(
+            "The relay host's content\\operator_secret.txt value "
+            "(auto-created there on first relay start).")
+        top.addWidget(self.f_secret)
         top.addSpacing(24)
         for label, slot in (("New", self._egg_new), ("Open…", self._egg_open),
                             ("Save", self._egg_save),
@@ -547,6 +636,18 @@ class Operator(QMainWindow):
         """Write the mission-parameter fields into the running session's egg
         file; the relay re-reads it at the next round's egg release, so the
         seated players get the new arena/params WITHOUT rejoining."""
+        if getattr(self, "remote_link", None):
+            spec = ("set map=%s;time=%s;weather=%s;scenario=%s;"
+                    "temperature=%s;length=%s"
+                    % (self.f_map.currentText(), self.f_time.currentText(),
+                       self.f_weather.currentText(),
+                       self.f_scenario.currentText(),
+                       self.f_temp.text().strip(),
+                       self.f_length.text().strip()))
+            self.remote_link.send(spec)
+            self.log.appendPlainText(">> mission settings sent to the remote "
+                                     "relay -- takes effect next round")
+            return
         try:
             doc = eggmodel.EggDoc.load(self.egg_path)
             doc.set_mission(
@@ -656,6 +757,11 @@ class Operator(QMainWindow):
 
     def _start_session(self):
         self.end_sent = False
+        relay_host = self.f_relayhost.text().strip()
+        if (self.mode.currentIndex() == 0 and relay_host
+                and relay_host.lower() not in ("localhost", "127.0.0.1")):
+            self._start_remote_session(relay_host)
+            return
         problems = self._collect_egg()
         if problems:
             QMessageBox.warning(self, "Cannot start", "\n".join(problems))
@@ -703,7 +809,51 @@ class Operator(QMainWindow):
                                   and self.f_manual.isChecked() else ""))
         self._refresh_pod_status()
 
+    def _start_remote_session(self, host):
+        secret = self.f_secret.text().strip()
+        if not secret:
+            QMessageBox.warning(self, "Remote relay",
+                                "Enter the relay host's secret\n"
+                                "(content\\operator_secret.txt on that "
+                                "machine).")
+            return
+        ctl_port = self.f_port.value() + 7          # CONTROL_PORT_OFFSET
+        self.monitor = SessionMonitor(True, [])
+        self.remote_link = RemoteRelayLink(host, ctl_port, secret)
+        self.remote_link.lineReceived.connect(self._ingest_line)
+        self.remote_link.closed.connect(self._remote_closed)
+        self.remote_link.start()
+        self.remote_link.send("get mission")
+        self.start_btn.setEnabled(False)
+        self.stop_btn.setEnabled(True)
+        self.restart_btn.setEnabled(True)
+        self.launch_local_btn.setEnabled(True)
+        self.launch_btn.setEnabled(False)
+        self.apply_btn.setEnabled(True)
+        self.log.appendPlainText("== REMOTE session: driving relay at "
+                                 "%s:%d (the relay stays up when you "
+                                 "disconnect) ==" % (host, ctl_port))
+        self._refresh_pod_status()
+
+    def _remote_closed(self, reason):
+        self.log.appendPlainText("== remote relay link closed: %s ==" % reason)
+        self.remote_link = None
+        self.start_btn.setEnabled(True)
+        self.stop_btn.setEnabled(False)
+        self.restart_btn.setEnabled(False)
+        self.launch_btn.setEnabled(False)
+        self.end_btn.setEnabled(False)
+        self.apply_btn.setEnabled(False)
+        self.pod_status.setText("remote link closed")
+
     def _stop_session(self):
+        if getattr(self, "remote_link", None):
+            link = self.remote_link
+            self.remote_link = None            # closed-signal becomes a no-op
+            link.closed.disconnect(self._remote_closed)
+            link.stop()
+            self.log.appendPlainText("== disconnected from remote relay "
+                                     "(the relay keeps running) ==")
         if self.console_proc:
             self.console_proc.kill()
             self.console_proc = None
@@ -727,7 +877,12 @@ class Operator(QMainWindow):
 
     def _launch_mission(self):
         """Manual launch gate: send the relay its 'launch' command."""
-        if self.console_proc:
+        if getattr(self, "remote_link", None):
+            self.remote_link.send("launch")
+            self.launch_btn.setEnabled(False)
+            self.end_btn.setEnabled(True)
+            self.log.appendPlainText(">> operator LAUNCH sent (remote)")
+        elif self.console_proc:
             self.console_proc.write(b"launch\n")
             self.launch_btn.setEnabled(False)
             self.end_btn.setEnabled(True)
@@ -736,7 +891,12 @@ class Operator(QMainWindow):
     def _end_mission(self):
         """End the running mission now (the relay also auto-stops at the
         egg's [mission] length -- the authentic console mission clock)."""
-        if self.console_proc:
+        if getattr(self, "remote_link", None):
+            self.remote_link.send("stop")
+            self.end_sent = True
+            self.end_btn.setEnabled(False)
+            self.log.appendPlainText(">> operator END MISSION sent (remote)")
+        elif self.console_proc:
             self.console_proc.write(b"stop\n")
             self.end_sent = True
             self.end_btn.setEnabled(False)
@@ -762,10 +922,43 @@ class Operator(QMainWindow):
         except OSError:
             pass
         for line in data.splitlines():
-            if line.strip():
-                self.log.appendPlainText(line)
-                if self.monitor and self.monitor.feed(line):
-                    self._refresh_pod_status()
+            self._ingest_line(line)
+
+    def _ingest_line(self, line):
+        """One relay/console log line -> log pane + tee file + monitor.
+        Shared by the local child-process path and the remote link."""
+        if not line.strip():
+            return
+        if getattr(self, "remote_link", None):
+            # remote lines aren't captured by the child-process tee above
+            try:
+                with open(os.path.join(CONTENT, "operator_relay.log"), "a",
+                          encoding="utf-8") as f:
+                    f.write(line + "\n")
+            except OSError:
+                pass
+            self._maybe_fill_mission_fields(line)
+        self.log.appendPlainText(line)
+        if self.monitor and self.monitor.feed(line):
+            self._refresh_pod_status()
+
+    def _maybe_fill_mission_fields(self, line):
+        """Prefill the mission form from the relay's 'get mission' reply."""
+        m = re.search(r"\[ctl\] mission map=(\S*) time=(\S*) "
+                      r"weather=(\S*) scenario=(\S*) temperature=(\S*) "
+                      r"length=(\S*)", line)
+        if not m:
+            return
+        for combo, value in ((self.f_map, m.group(1)),
+                             (self.f_time, m.group(2)),
+                             (self.f_weather, m.group(3)),
+                             (self.f_scenario, m.group(4))):
+            if value:
+                combo.setCurrentText(value)
+        if m.group(5):
+            self.f_temp.setText(m.group(5))
+        if m.group(6):
+            self.f_length.setText(m.group(6))
 
     def _refresh_pod_status(self):
         if not self.monitor:
@@ -886,7 +1079,11 @@ class Operator(QMainWindow):
                 # relaunches itself with -egg OPERATOR.EGG -net <BT_FE_JOINPORT>
                 # (distinct per instance -- two -net listeners can't share a
                 # port on one box).
-                env.insert("BT_RELAY", "127.0.0.1:%d" % self.f_port.value())
+                relay_addr = "127.0.0.1"
+                if getattr(self, "remote_link", None):
+                    relay_addr = self.remote_link.host
+                env.insert("BT_RELAY", "%s:%d" % (relay_addr,
+                                                  self.f_port.value()))
                 env.insert("BT_FE_JOIN", "1")
                 env.insert("BT_FE_JOINPORT", str(1501 + 100 * r))
                 args = []