Refactor server into reusable core and GUI

Split the monolithic test server into a reusable server core and a thin terminal launcher, and add a PySide6 developer GUI. Added server_core.py (FalloutTogetherServer) with lifecycle control, thread-safe client snapshots, stats and log listener support; added client_session.py for per-client state; added dev_server_app.py GUI and requirements.txt. Updated server.py to use the new server core, and revised server/README.md and docs/dev-log.md to document the new structure, usage, and validation steps. Protocol behavior (welcome/transform/disconnect handling and newline-separated JSON) remains unchanged.

docs/dev-log.md

@@ -882,6 +882,102 @@ Player position: X=2048.00, Y=2048.00, Z=0.00, AngleZ=0.00
 
 ---
 
+## 2026-06-02 - Reusable Server Core Refactor
+
+### What Changed
+
+- Split the Python relay server into a reusable server core and a thin terminal
+  launcher.
+- Added `server/client_session.py` to track connected client state, packet
+  counters, last packet time, and last transform snapshots.
+- Added `server/server_core.py` with `FalloutTogetherServer` lifecycle methods,
+  log listeners, client snapshots, server stats, and thread-safe shared state.
+- Updated `server/README.md` with the new server structure and validation
+  commands.
+
+### What Worked
+
+- The terminal command remains `cd server` followed by `python server.py`.
+- The server still defaults to `127.0.0.1:7777`.
+- Newline-separated JSON handling, welcome packets, transform mutation,
+  transform relay, disconnect relay, and unknown transform field preservation
+  remain protocol-compatible.
+- The refactor prepares the server for a future desktop dev-server UI without
+  adding PySide6, fake-player management, movement scripts, or gameplay
+  features.
+
+### What Broke
+
+- Nothing recorded during implementation.
+
+### Notes
+
+- The future UI can use `FalloutTogetherServer.start()`, `stop()`,
+  `is_running()`, `get_clients()`, `get_stats()`, and log listener callbacks
+  without scraping terminal output.
+- Fallout 4 plugin code and network protocol fields were left unchanged.
+
+### Next Steps
+
+- Run compile checks for the server files.
+- Validate server startup, fake-client welcome handling, transform relay, and
+  disconnect relay with fake clients and Fallout 4 plugin clients.
+
+---
+
+## 2026-06-02 - Dev Server GUI
+
+### What Changed
+
+- Added a PySide6 desktop developer GUI for the reusable Python server core.
+- Added a Server Console tab with start/stop controls, live server logs, live
+  stats, and a connected-client table.
+- Added a Fake Clients tab placeholder for future fake-client management and
+  movement-script controls.
+- Added a server-local requirements file for the GUI dependency.
+- Updated server documentation with GUI install and run instructions while
+  preserving the terminal server flow.
+
+### What Worked
+
+- The GUI uses `FalloutTogetherServer` instead of duplicating networking logic.
+- Server logs flow through `add_log_listener(...)` and a Qt signal bridge rather
+  than stdout scraping.
+- Stats and connected-client snapshots are refreshed through `get_stats()` and
+  `get_clients()` on a `QTimer`.
+- The terminal launcher, fake client, network protocol, and plugin code were
+  left unchanged.
+
+### What Broke
+
+- Nothing recorded during implementation.
+
+### Notes
+
+- Fake client spawning and movement scripts remain intentionally unimplemented.
+- Full GUI interaction and Fallout 4/F4SE validation still need to be run
+  manually on the desktop.
+
+### Testing
+
+- Ran `python -m py_compile server.py server_core.py client_session.py fake_client.py dev_server_app.py`
+  from `server`; it succeeded.
+- Checked IDE lints for the new GUI and edited docs; no diagnostics were
+  reported.
+- Checked the current Python environment for `PySide6`; it is not installed, so
+  `pip install -r requirements.txt` is required before launching the GUI.
+- Did not start a duplicate terminal server because an existing `python
+  server.py` session and `python fake_client.py` session were already present.
+
+### Next Steps
+
+- Run `python dev_server_app.py`, click `Start Server`, and connect
+  `fake_client.py`.
+- Launch Fallout 4 through F4SE and confirm the plugin appears in the GUI client
+  table with transform stats updating.
+
+---
+
 ## Entry Template
 
 Use this format for future updates:

server/README.md

@@ -2,7 +2,7 @@
 
 This folder is for the external multiplayer test server.
 
-The current server is a tiny local-only TCP test server. It listens on
+The current server is a local-only TCP test server. It listens on
 `127.0.0.1:7777`, accepts one or more clients, reads newline-separated JSON
 packets, assigns incrementing `playerId` values, sends `welcome` packets, adds
 `playerId` and `serverTime` to transform packets, broadcasts transform packets
@@ -19,11 +19,21 @@ playable yet.
 
 ## Run
 
+Terminal server:
+
 ```bash
 cd server
 python server.py
 ```
 
+Developer GUI server:
+
+```bash
+cd server
+pip install -r requirements.txt
+python dev_server_app.py
+```
+
 Then launch Fallout 4 through F4SE and move the player. The plugin should send
 local player transform packets, and this server should print matching
 `transform` packets.
@@ -31,6 +41,35 @@ local player transform packets, and this server should print matching
 If the server is not running, the plugin should log a warning and Fallout 4
 should continue launching normally.
 
+## Structure
+
+- `server.py`: thin terminal launcher for the local test server.
+- `server_core.py`: reusable `FalloutTogetherServer` core for networking,
+  lifecycle control, packet relay, stats, client snapshots, and log callbacks.
+- `client_session.py`: `ClientSession` state model for connected clients.
+- `dev_server_app.py`: PySide6 developer GUI that starts/stops the reusable
+  server core, displays logs, shows live stats, and lists connected clients.
+- `fake_client.py`: lightweight receiver for testing welcome, transform, and
+  disconnect packets without launching a second Fallout 4 instance.
+- `requirements.txt`: optional Python dependencies for the desktop developer
+  GUI.
+
+The reusable server core exposes methods intended for a future desktop dev
+server UI:
+
+- `start()`
+- `serve_forever()`
+- `stop()`
+- `is_running()`
+- `get_clients()`
+- `get_stats()`
+- `add_log_listener()`
+- `remove_log_listener()`
+
+`get_clients()` and `get_stats()` return snapshot dictionaries, not live
+internal server state. Log listeners receive the same useful console messages
+that the terminal launcher prints today.
+
 ## Current Test Flow
 
 Terminal 1:
@@ -49,6 +88,17 @@ python fake_client.py
 
 Then launch Fallout 4 through F4SE and move the player.
 
+Alternative GUI flow:
+
+```bash
+cd server
+pip install -r requirements.txt
+python dev_server_app.py
+```
+
+Click `Start Server`, then run `python fake_client.py` or launch Fallout 4
+through F4SE.
+
 Expected behavior:
 
 - The server assigns a `playerId` to each connected client and sends each client
@@ -70,6 +120,39 @@ The fake client exists so broadcast behavior can be tested before coordinating a
 second Fallout 4/F4SE instance. It stores remote player transform state only; it
 does not spawn remote actors or send movement.
 
+## Validation
+
+From inside this folder:
+
+```bash
+python -m py_compile server.py server_core.py client_session.py fake_client.py dev_server_app.py
+```
+
+Then run the terminal server:
+
+```bash
+python server.py
+```
+
+In another terminal, run:
+
+```bash
+python fake_client.py
+```
+
+To validate the developer GUI instead of the terminal launcher:
+
+```bash
+pip install -r requirements.txt
+python dev_server_app.py
+```
+
+Click `Start Server`, then connect `fake_client.py` or the Fallout 4 plugin.
+
+For in-game validation, launch Fallout 4 through F4SE and confirm the plugin
+connects, receives a `welcome` packet, sends transforms, and receives relayed
+remote transforms or disconnect packets when another client is connected.
+
 ## Implemented
 
 - Start local server
@@ -82,6 +165,10 @@ does not spawn remote actors or send movement.
 - Broadcast transforms to other connected clients
 - Broadcast disconnect packets
 - Handle disconnect cleanup
+- Reusable server core with thread-safe client/stat snapshots
+- Log callbacks for future server console UI
+- PySide6 developer server GUI with live logs, stats, connected-client table,
+  and a placeholder Fake Clients tab
 
 ## Working In Fake Client Only
 
@@ -90,13 +177,10 @@ does not spawn remote actors or send movement.
 - Storing remote player state
 - Removing disconnected remote players
 
-The Fallout 4 plugin does not yet receive remote transforms from the server.
-
 ## Planned Later
 
-- Fallout 4 plugin receive loop
+- Fake client management
-- Plugin-side remote player state storage
+- Fake client movement scripts
-- Remote actor spawning
 - Gameplay synchronization
 - Authentication
 - Public matchmaking

server/client_session.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import socket
+import threading
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class ClientSession:
+    connection: socket.socket
+    address: tuple[str, int]
+    player_id: int
+    connected_at: float
+    last_packet_at: float | None = None
+    last_transform: dict[str, Any] | None = None
+    packets_received: int = 0
+    packets_sent: int = 0
+    packets_broadcast: int = 0
+    _lock: Any = field(repr=False, compare=False, default_factory=threading.RLock)
+
+    @property
+    def label(self) -> str:
+        return f"{self.address[0]}:{self.address[1]}"
+
+    def record_received(self, received_at: float) -> None:
+        with self._lock:
+            self.last_packet_at = received_at
+            self.packets_received += 1
+
+    def record_transform(self, packet: dict[str, Any]) -> None:
+        with self._lock:
+            self.last_transform = dict(packet)
+
+    def record_sent(self, *, broadcast: bool = False) -> None:
+        with self._lock:
+            self.packets_sent += 1
+            if broadcast:
+                self.packets_broadcast += 1
+
+    def to_snapshot(self) -> dict[str, Any]:
+        with self._lock:
+            return {
+                "playerId": self.player_id,
+                "address": self.address[0],
+                "port": self.address[1],
+                "connectedAt": self.connected_at,
+                "lastPacketAt": self.last_packet_at,
+                "lastTransform": dict(self.last_transform) if self.last_transform is not None else None,
+                "packetsReceived": self.packets_received,
+                "packetsSent": self.packets_sent,
+                "packetsBroadcast": self.packets_broadcast,
+            }

server/dev_server_app.py

@@ -0,0 +1,328 @@
+from __future__ import annotations
+
+import sys
+import time
+from typing import Any
+
+from PySide6.QtCore import QObject, QTimer, Signal
+from PySide6.QtGui import QCloseEvent
+from PySide6.QtWidgets import (
+    QApplication,
+    QAbstractItemView,
+    QGridLayout,
+    QHBoxLayout,
+    QHeaderView,
+    QLabel,
+    QMainWindow,
+    QMessageBox,
+    QPlainTextEdit,
+    QPushButton,
+    QTableWidget,
+    QTableWidgetItem,
+    QTabWidget,
+    QVBoxLayout,
+    QWidget,
+)
+
+from server_core import FalloutTogetherServer
+
+
+REFRESH_INTERVAL_MS = 500
+
+
+class ServerLogBridge(QObject):
+    message_received = Signal(str)
+
+
+class DevServerWindow(QMainWindow):
+    def __init__(self) -> None:
+        super().__init__()
+
+        self.server = FalloutTogetherServer()
+        self.log_bridge = ServerLogBridge()
+        self.log_bridge.message_received.connect(self._append_log)
+        self.server.add_log_listener(self._handle_server_log)
+
+        self.stat_labels: dict[str, QLabel] = {}
+
+        self.setWindowTitle("Fallout 4 Together Dev Server")
+        self.resize(1100, 760)
+        self.setCentralWidget(self._create_tabs())
+
+        self.refresh_timer = QTimer(self)
+        self.refresh_timer.timeout.connect(self._refresh_server_state)
+        self.refresh_timer.start(REFRESH_INTERVAL_MS)
+
+        self._refresh_server_state()
+
+    def closeEvent(self, event: QCloseEvent) -> None:
+        self.refresh_timer.stop()
+        self.server.remove_log_listener(self._handle_server_log)
+        self.server.stop()
+        event.accept()
+
+    def _create_tabs(self) -> QTabWidget:
+        tabs = QTabWidget()
+        tabs.addTab(self._create_server_console_tab(), "Server Console")
+        tabs.addTab(self._create_fake_clients_tab(), "Fake Clients")
+        return tabs
+
+    def _create_server_console_tab(self) -> QWidget:
+        tab = QWidget()
+        layout = QVBoxLayout(tab)
+
+        layout.addLayout(self._create_stats_layout())
+
+        layout.addWidget(QLabel("Console"))
+        self.console = QPlainTextEdit()
+        self.console.setReadOnly(True)
+        layout.addLayout(self._create_server_button_layout())
+        layout.addWidget(self.console, stretch=2)
+
+        layout.addWidget(QLabel("Connected Clients"))
+        self.clients_table = QTableWidget()
+        self.clients_table.setColumnCount(9)
+        self.clients_table.setHorizontalHeaderLabels(
+            [
+                "Player ID",
+                "Address",
+                "Port",
+                "Connected for / connected time",
+                "Last packet time",
+                "Packets received",
+                "Packets broadcast",
+                "Last cellId",
+                "Last position",
+            ]
+        )
+        self.clients_table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
+        self.clients_table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
+        self.clients_table.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)
+        self.clients_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch)
+        layout.addWidget(self.clients_table, stretch=1)
+
+        return tab
+
+    def _create_stats_layout(self) -> QGridLayout:
+        layout = QGridLayout()
+        stats = [
+            ("status", "Status"),
+            ("host", "Host"),
+            ("port", "Port"),
+            ("uptime", "Uptime"),
+            ("connectedClients", "Connected clients"),
+            ("clientsConnected", "Total clients connected"),
+            ("clientsDisconnected", "Total clients disconnected"),
+            ("packetsReceived", "Packets received"),
+            ("packetsBroadcast", "Packets broadcast"),
+        ]
+
+        for index, (key, label_text) in enumerate(stats):
+            row = index // 3
+            column = (index % 3) * 2
+            label = QLabel(f"{label_text}:")
+            value = QLabel("-")
+            self.stat_labels[key] = value
+            layout.addWidget(label, row, column)
+            layout.addWidget(value, row, column + 1)
+
+        return layout
+
+    def _create_server_button_layout(self) -> QHBoxLayout:
+        layout = QHBoxLayout()
+
+        self.start_button = QPushButton("Start Server")
+        self.start_button.clicked.connect(self._start_server)
+        layout.addWidget(self.start_button)
+
+        self.stop_button = QPushButton("Stop Server")
+        self.stop_button.clicked.connect(self._stop_server)
+        layout.addWidget(self.stop_button)
+
+        self.clear_console_button = QPushButton("Clear Console")
+        self.clear_console_button.clicked.connect(self.console.clear)
+        layout.addWidget(self.clear_console_button)
+
+        self.copy_console_button = QPushButton("Copy Console")
+        self.copy_console_button.clicked.connect(self._copy_console)
+        layout.addWidget(self.copy_console_button)
+
+        layout.addStretch()
+        return layout
+
+    def _create_fake_clients_tab(self) -> QWidget:
+        tab = QWidget()
+        layout = QVBoxLayout(tab)
+
+        button_layout = QHBoxLayout()
+
+        self.add_fake_client_button = QPushButton("+ Add Fake Client")
+        self.add_fake_client_button.clicked.connect(self._show_fake_clients_not_implemented)
+        button_layout.addWidget(self.add_fake_client_button)
+
+        self.remove_fake_client_button = QPushButton("- Remove Selected")
+        self.remove_fake_client_button.clicked.connect(self._show_fake_clients_not_implemented)
+        button_layout.addWidget(self.remove_fake_client_button)
+
+        button_layout.addStretch()
+        layout.addLayout(button_layout)
+
+        self.fake_clients_table = QTableWidget()
+        self.fake_clients_table.setColumnCount(6)
+        self.fake_clients_table.setHorizontalHeaderLabels(
+            ["Name", "Player ID", "Status", "Script", "Cell ID", "Position"]
+        )
+        self.fake_clients_table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
+        self.fake_clients_table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
+        self.fake_clients_table.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)
+        self.fake_clients_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch)
+        layout.addWidget(self.fake_clients_table)
+
+        placeholder = [
+            "Fake Client 1",
+            "-",
+            "Not implemented yet",
+            "None",
+            "-",
+            "-",
+        ]
+        self.fake_clients_table.setRowCount(1)
+        for column, value in enumerate(placeholder):
+            self.fake_clients_table.setItem(0, column, self._create_table_item(value))
+
+        # TODO: Add a fake client manager that can own multiple test clients.
+        # TODO: Add movement scripts once fake clients can be spawned safely.
+        # TODO: Wire add/remove fake client controls without duplicating networking logic.
+        # TODO: Add script controls for selecting and starting/stopping movement scripts.
+
+        return tab
+
+    def _start_server(self) -> None:
+        try:
+            self.server.start()
+        except OSError as error:
+            self._append_log(f"Failed to start server on {self.server.host}:{self.server.port}: {error}")
+        except RuntimeError as error:
+            self._append_log(f"Failed to start server: {error}")
+
+        self._refresh_server_state()
+
+    def _stop_server(self) -> None:
+        if self.server.is_running():
+            self.server.stop()
+            self._append_log("Server stopped.")
+
+        self._refresh_server_state()
+
+    def _copy_console(self) -> None:
+        QApplication.clipboard().setText(self.console.toPlainText())
+
+    def _show_fake_clients_not_implemented(self) -> None:
+        QMessageBox.information(
+            self,
+            "Fake Clients",
+            "Fake client spawning is not implemented yet.",
+        )
+
+    def _handle_server_log(self, message: str) -> None:
+        self.log_bridge.message_received.emit(message)
+
+    def _append_log(self, message: str) -> None:
+        timestamp = time.strftime("%H:%M:%S", time.localtime())
+        self.console.appendPlainText(f"[{timestamp}] {message}")
+
+    def _refresh_server_state(self) -> None:
+        stats = self.server.get_stats()
+        clients = self.server.get_clients()
+        is_running = bool(stats.get("isRunning", False))
+
+        self.stat_labels["status"].setText("Running" if is_running else "Stopped")
+        self.stat_labels["host"].setText(str(stats.get("host", "-")))
+        self.stat_labels["port"].setText(str(stats.get("port", "-")))
+        self.stat_labels["uptime"].setText(self._format_duration(float(stats.get("uptimeSeconds", 0.0))))
+        self.stat_labels["connectedClients"].setText(str(stats.get("connectedClients", 0)))
+        self.stat_labels["clientsConnected"].setText(str(stats.get("clientsConnected", 0)))
+        self.stat_labels["clientsDisconnected"].setText(str(stats.get("clientsDisconnected", 0)))
+        self.stat_labels["packetsReceived"].setText(str(stats.get("packetsReceived", 0)))
+        self.stat_labels["packetsBroadcast"].setText(str(stats.get("packetsBroadcast", 0)))
+
+        self.start_button.setEnabled(not is_running)
+        self.stop_button.setEnabled(is_running)
+
+        self._refresh_clients_table(clients)
+
+    def _refresh_clients_table(self, clients: list[dict[str, Any]]) -> None:
+        self.clients_table.setRowCount(len(clients))
+        now = time.time()
+
+        for row, client in enumerate(clients):
+            connected_at = self._get_optional_float(client.get("connectedAt"))
+            last_packet_at = self._get_optional_float(client.get("lastPacketAt"))
+            last_transform = client.get("lastTransform")
+            if not isinstance(last_transform, dict):
+                last_transform = {}
+
+            connected_text = "-"
+            if connected_at is not None:
+                connected_text = f"{self._format_duration(now - connected_at)} / {self._format_timestamp(connected_at)}"
+
+            values = [
+                client.get("playerId", "-"),
+                client.get("address", "-"),
+                client.get("port", "-"),
+                connected_text,
+                self._format_timestamp(last_packet_at) if last_packet_at is not None else "-",
+                client.get("packetsReceived", 0),
+                client.get("packetsBroadcast", 0),
+                last_transform.get("cellId", "-"),
+                self._format_position(last_transform),
+            ]
+
+            for column, value in enumerate(values):
+                self.clients_table.setItem(row, column, self._create_table_item(str(value)))
+
+    @staticmethod
+    def _create_table_item(value: str) -> QTableWidgetItem:
+        return QTableWidgetItem(value)
+
+    @staticmethod
+    def _get_optional_float(value: Any) -> float | None:
+        try:
+            return float(value)
+        except (TypeError, ValueError):
+            return None
+
+    @staticmethod
+    def _format_duration(seconds: float) -> str:
+        total_seconds = max(0, int(seconds))
+        hours, remainder = divmod(total_seconds, 3600)
+        minutes, secs = divmod(remainder, 60)
+        if hours:
+            return f"{hours:d}:{minutes:02d}:{secs:02d}"
+        return f"{minutes:d}:{secs:02d}"
+
+    @staticmethod
+    def _format_timestamp(timestamp: float) -> str:
+        return time.strftime("%H:%M:%S", time.localtime(timestamp))
+
+    @staticmethod
+    def _format_position(transform: dict[str, Any]) -> str:
+        try:
+            x = float(transform["x"])
+            y = float(transform["y"])
+            z = float(transform["z"])
+        except (KeyError, TypeError, ValueError):
+            return "-"
+
+        return f"{x:.2f}, {y:.2f}, {z:.2f}"
+
+
+def main() -> int:
+    app = QApplication(sys.argv)
+    window = DevServerWindow()
+    window.show()
+    return app.exec()
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

server/requirements.txt

@@ -0,0 +1 @@
+PySide6

server/server.py

@@ -1,227 +1,20 @@
 from __future__ import annotations
 
-import json
+from server_core import FalloutTogetherServer
-import socket
-import threading
-import time
-from dataclasses import dataclass
-from typing import Any
-
-
-HOST = "127.0.0.1"
-PORT = 7777
-ACCEPT_TIMEOUT_SECONDS = 0.5
-
-clients_lock = threading.Lock()
-clients: dict[socket.socket, "Client"] = {}
-next_player_id = 1
-
-
-@dataclass
-class Client:
-    connection: socket.socket
-    address: tuple[str, int]
-    player_id: int
-
-    @property
-    def label(self) -> str:
-        return f"{self.address[0]}:{self.address[1]}"
 
 
 def log(message: str) -> None:
     print(message, flush=True)
 
 
-def assign_client(connection: socket.socket, address: tuple[str, int]) -> Client:
-    global next_player_id
-
-    with clients_lock:
-        # The server owns player IDs so early clients do not need to coordinate
-        # identity with each other or know anything about remote players yet.
-        client = Client(connection=connection, address=address, player_id=next_player_id)
-        next_player_id += 1
-        clients[connection] = client
-
-    return client
-
-
-def remove_client(client: Client) -> bool:
-    with clients_lock:
-        return clients.pop(client.connection, None) is not None
-
-
-def send_packet(connection: socket.socket, packet: dict[str, Any]) -> None:
-    encoded = json.dumps(packet, separators=(",", ":")).encode("utf-8") + b"\n"
-    connection.sendall(encoded)
-
-
-def disconnect_client(client: Client) -> None:
-    if not remove_client(client):
-        return
-
-    try:
-        client.connection.close()
-    except OSError:
-        pass
-
-    log(f"Client disconnected: {client.label} (player {client.player_id})")
-    broadcast_disconnect(client)
-
-
-def handle_client(connection: socket.socket, address: tuple[str, int]) -> None:
-    client = assign_client(connection, address)
-    log(f"Client connected: {client.label} (player {client.player_id})")
-
-    with connection:
-        buffer = ""
-
-        try:
-            send_packet(
-                connection,
-                {
-                    "type": "welcome",
-                    "playerId": client.player_id,
-                    "serverTime": time.time(),
-                },
-            )
-
-            while True:
-                chunk = connection.recv(4096)
-                if not chunk:
-                    break
-
-                buffer += chunk.decode("utf-8", errors="replace")
-                while "\n" in buffer:
-                    line, buffer = buffer.split("\n", 1)
-                    handle_line(client, line.strip())
-        except ConnectionResetError:
-            log(f"Client disconnected unexpectedly: {client.label} (player {client.player_id})")
-        except OSError as error:
-            log(f"Client connection error: {client.label} (player {client.player_id}): {error}")
-        finally:
-            disconnect_client(client)
-
-
-def handle_line(client: Client, line: str) -> None:
-    if not line:
-        return
-
-    try:
-        packet = json.loads(line)
-    except json.JSONDecodeError as error:
-        log(f"Invalid JSON from {client.label}: {error}: {line}")
-        return
-
-    if not isinstance(packet, dict):
-        log(f"Invalid packet from {client.label}: expected JSON object: {packet}")
-        return
-
-    if packet.get("type") == "transform":
-        packet["playerId"] = client.player_id
-        packet["serverTime"] = time.time()
-
-    print_packet(client, packet)
-
-    if packet.get("type") == "transform":
-        broadcast_transform(client, packet)
-
-
-def broadcast_transform(sender: Client, packet: dict[str, Any]) -> None:
-    with clients_lock:
-        recipients = [client for client in clients.values() if client.connection != sender.connection]
-
-    failed_recipients: list[Client] = []
-    for recipient in recipients:
-        try:
-            # Transforms go to every other client only. The sender already knows
-            # its own movement; echoing it back would create duplicate local state.
-            send_packet(recipient.connection, packet)
-        except OSError:
-            failed_recipients.append(recipient)
-
-    for recipient in failed_recipients:
-        disconnect_client(recipient)
-
-    log(f"Broadcast transform from player {sender.player_id} to {len(recipients) - len(failed_recipients)} other client(s)")
-
-
-def broadcast_disconnect(disconnected_client: Client) -> None:
-    packet = {
-        "type": "disconnect",
-        "playerId": disconnected_client.player_id,
-        "serverTime": time.time(),
-    }
-
-    with clients_lock:
-        recipients = list(clients.values())
-
-    failed_recipients: list[Client] = []
-    for recipient in recipients:
-        try:
-            send_packet(recipient.connection, packet)
-        except OSError:
-            failed_recipients.append(recipient)
-
-    for recipient in failed_recipients:
-        disconnect_client(recipient)
-
-    log(
-        f"Broadcast disconnect for player {disconnected_client.player_id} "
-        f"to {len(recipients) - len(failed_recipients)} other client(s)"
-    )
-
-
-def print_packet(client: Client, packet: dict[str, Any]) -> None:
-    if packet.get("type") != "transform":
-        log(f"Packet from {client.label}: {packet}")
-        return
-
-    try:
-        x = float(packet["x"])
-        y = float(packet["y"])
-        z = float(packet["z"])
-        angle_z = float(packet["angleZ"])
-    except (KeyError, TypeError, ValueError):
-        log(f"Malformed transform packet from {client.label}: {packet}")
-        return
-
-    movement_type = packet.get("movementType", "normal")
-    extra_fields = []
-    for field_name in ("playerId", "cellId", "worldspaceId", "clientTime", "serverTime"):
-        if field_name in packet:
-            extra_fields.append(f"{field_name}={packet[field_name]}")
-
-    extra_details = ""
-    if extra_fields:
-        extra_details = ", " + ", ".join(extra_fields)
-
-    log(
-        "Transform from "
-        f"{client.label}: x={x:.2f}, y={y:.2f}, z={z:.2f}, angleZ={angle_z:.2f}, "
-        f"movementType={movement_type}{extra_details}"
-    )
-
-
 def main() -> None:
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_socket:
+    server = FalloutTogetherServer()
-        if hasattr(socket, "SO_EXCLUSIVEADDRUSE"):
+    server.add_log_listener(log)
-            server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
-        server_socket.bind((HOST, PORT))
-        server_socket.listen()
-        server_socket.settimeout(ACCEPT_TIMEOUT_SECONDS)
 
-        log(f"Fallout 4 Together local test server listening on {HOST}:{PORT}")
-        log("Waiting for newline-separated JSON transform packets...")
-
-        while True:
     try:
-                connection, address = server_socket.accept()
+        server.serve_forever()
-            except socket.timeout:
+    finally:
-                # Wake periodically so Ctrl+C is handled promptly in Windows terminals.
+        server.stop()
-                continue
-
-            thread = threading.Thread(target=handle_client, args=(connection, address), daemon=True)
-            thread.start()
 
 
 if __name__ == "__main__":

server/server_core.py

@@ -0,0 +1,381 @@
+from __future__ import annotations
+
+import json
+import socket
+import threading
+import time
+from collections.abc import Callable
+from typing import Any
+
+from client_session import ClientSession
+
+
+HOST = "127.0.0.1"
+PORT = 7777
+ACCEPT_TIMEOUT_SECONDS = 0.5
+
+
+class FalloutTogetherServer:
+    def __init__(self, host: str = HOST, port: int = PORT) -> None:
+        self.host = host
+        self.port = port
+
+        self._lock = threading.RLock()
+        self._log_lock = threading.RLock()
+        self._clients: dict[socket.socket, ClientSession] = {}
+        self._log_listeners: list[Callable[[str], None]] = []
+        self._server_socket: socket.socket | None = None
+        self._accept_thread: threading.Thread | None = None
+        self._running = False
+        self._started_at: float | None = None
+        self._next_player_id = 1
+
+        self._stats: dict[str, int] = {
+            "clientsConnected": 0,
+            "clientsDisconnected": 0,
+            "packetsReceived": 0,
+            "packetsSent": 0,
+            "packetsBroadcast": 0,
+            "transformPacketsReceived": 0,
+            "transformPacketsBroadcast": 0,
+            "disconnectPacketsBroadcast": 0,
+        }
+
+    def start(self) -> None:
+        with self._lock:
+            if self._running:
+                return
+
+            self._prepare_server_socket()
+            self._accept_thread = threading.Thread(target=self._accept_loop, daemon=True)
+            self._accept_thread.start()
+
+    def serve_forever(self) -> None:
+        with self._lock:
+            if self._running:
+                raise RuntimeError("Server is already running.")
+
+            self._prepare_server_socket()
+
+        self._accept_loop()
+
+    def stop(self) -> None:
+        clients: list[ClientSession]
+        server_socket: socket.socket | None
+
+        with self._lock:
+            self._running = False
+            server_socket = self._server_socket
+            self._server_socket = None
+            clients = list(self._clients.values())
+
+        if server_socket is not None:
+            try:
+                server_socket.close()
+            except OSError:
+                pass
+
+        for client in clients:
+            self._close_client_socket(client)
+
+        accept_thread = self._accept_thread
+        if accept_thread is not None and accept_thread is not threading.current_thread():
+            accept_thread.join(timeout=1.0)
+
+    def is_running(self) -> bool:
+        with self._lock:
+            return self._running
+
+    def get_clients(self) -> list[dict[str, Any]]:
+        with self._lock:
+            clients = list(self._clients.values())
+
+        return [client.to_snapshot() for client in clients]
+
+    def get_stats(self) -> dict[str, Any]:
+        with self._lock:
+            connected_clients = len(self._clients)
+            started_at = self._started_at
+            stats = dict(self._stats)
+            stats.update(
+                {
+                    "host": self.host,
+                    "port": self.port,
+                    "isRunning": self._running,
+                    "startedAt": started_at,
+                    "uptimeSeconds": time.time() - started_at if started_at is not None else 0.0,
+                    "connectedClients": connected_clients,
+                    "nextPlayerId": self._next_player_id,
+                }
+            )
+
+        return stats
+
+    def add_log_listener(self, callback: Callable[[str], None]) -> None:
+        with self._log_lock:
+            if callback not in self._log_listeners:
+                self._log_listeners.append(callback)
+
+    def remove_log_listener(self, callback: Callable[[str], None]) -> None:
+        with self._log_lock:
+            if callback in self._log_listeners:
+                self._log_listeners.remove(callback)
+
+    def _prepare_server_socket(self) -> None:
+        server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        try:
+            if hasattr(socket, "SO_EXCLUSIVEADDRUSE"):
+                server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
+            server_socket.bind((self.host, self.port))
+            server_socket.listen()
+            server_socket.settimeout(ACCEPT_TIMEOUT_SECONDS)
+        except OSError:
+            server_socket.close()
+            raise
+
+        self._server_socket = server_socket
+        self._running = True
+        self._started_at = time.time()
+
+        self._log(f"Fallout 4 Together local test server listening on {self.host}:{self.port}")
+        self._log("Waiting for newline-separated JSON transform packets...")
+
+    def _accept_loop(self) -> None:
+        try:
+            while self.is_running():
+                with self._lock:
+                    server_socket = self._server_socket
+
+                if server_socket is None:
+                    break
+
+                try:
+                    connection, address = server_socket.accept()
+                except socket.timeout:
+                    # Wake periodically so Ctrl+C is handled promptly in Windows terminals.
+                    continue
+                except OSError:
+                    if self.is_running():
+                        self._log("Server socket closed unexpectedly.")
+                    break
+
+                thread = threading.Thread(target=self._handle_client, args=(connection, address), daemon=True)
+                thread.start()
+        finally:
+            with self._lock:
+                server_socket = self._server_socket
+                self._running = False
+                self._server_socket = None
+
+            if server_socket is not None:
+                try:
+                    server_socket.close()
+                except OSError:
+                    pass
+
+    def _assign_client(self, connection: socket.socket, address: tuple[str, int]) -> ClientSession:
+        with self._lock:
+            # The server owns player IDs so early clients do not need to coordinate
+            # identity with each other or know anything about remote players yet.
+            client = ClientSession(
+                connection=connection,
+                address=address,
+                player_id=self._next_player_id,
+                connected_at=time.time(),
+            )
+            self._next_player_id += 1
+            self._clients[connection] = client
+            self._stats["clientsConnected"] += 1
+
+        return client
+
+    def _remove_client(self, client: ClientSession) -> bool:
+        with self._lock:
+            removed = self._clients.pop(client.connection, None) is not None
+            if removed:
+                self._stats["clientsDisconnected"] += 1
+            return removed
+
+    def _handle_client(self, connection: socket.socket, address: tuple[str, int]) -> None:
+        client = self._assign_client(connection, address)
+        self._log(f"Client connected: {client.label} (player {client.player_id})")
+
+        with connection:
+            buffer = ""
+
+            try:
+                self._send_packet(
+                    client,
+                    {
+                        "type": "welcome",
+                        "playerId": client.player_id,
+                        "serverTime": time.time(),
+                    },
+                )
+
+                while self.is_running():
+                    chunk = connection.recv(4096)
+                    if not chunk:
+                        break
+
+                    buffer += chunk.decode("utf-8", errors="replace")
+                    while "\n" in buffer:
+                        line, buffer = buffer.split("\n", 1)
+                        self._handle_line(client, line.strip())
+            except ConnectionResetError:
+                self._log(f"Client disconnected unexpectedly: {client.label} (player {client.player_id})")
+            except OSError as error:
+                if self.is_running():
+                    self._log(f"Client connection error: {client.label} (player {client.player_id}): {error}")
+            finally:
+                self._disconnect_client(client)
+
+    def _handle_line(self, client: ClientSession, line: str) -> None:
+        if not line:
+            return
+
+        received_at = time.time()
+        client.record_received(received_at)
+
+        with self._lock:
+            self._stats["packetsReceived"] += 1
+
+        try:
+            packet = json.loads(line)
+        except json.JSONDecodeError as error:
+            self._log(f"Invalid JSON from {client.label}: {error}: {line}")
+            return
+
+        if not isinstance(packet, dict):
+            self._log(f"Invalid packet from {client.label}: expected JSON object: {packet}")
+            return
+
+        if packet.get("type") == "transform":
+            packet["playerId"] = client.player_id
+            packet["serverTime"] = time.time()
+            client.record_transform(packet)
+            with self._lock:
+                self._stats["transformPacketsReceived"] += 1
+
+        self._print_packet(client, packet)
+
+        if packet.get("type") == "transform":
+            self._broadcast_transform(client, packet)
+
+    def _send_packet(self, client: ClientSession, packet: dict[str, Any], *, broadcast: bool = False) -> None:
+        encoded = json.dumps(packet, separators=(",", ":")).encode("utf-8") + b"\n"
+        client.connection.sendall(encoded)
+        client.record_sent(broadcast=broadcast)
+
+        with self._lock:
+            self._stats["packetsSent"] += 1
+            if broadcast:
+                self._stats["packetsBroadcast"] += 1
+
+    def _disconnect_client(self, client: ClientSession) -> None:
+        if not self._remove_client(client):
+            return
+
+        self._close_client_socket(client)
+        self._log(f"Client disconnected: {client.label} (player {client.player_id})")
+        self._broadcast_disconnect(client)
+
+    def _close_client_socket(self, client: ClientSession) -> None:
+        try:
+            client.connection.close()
+        except OSError:
+            pass
+
+    def _broadcast_transform(self, sender: ClientSession, packet: dict[str, Any]) -> None:
+        with self._lock:
+            recipients = [client for client in self._clients.values() if client.connection != sender.connection]
+
+        failed_recipients: list[ClientSession] = []
+        successful_sends = 0
+        for recipient in recipients:
+            try:
+                # Transforms go to every other client only. The sender already knows
+                # its own movement; echoing it back would create duplicate local state.
+                self._send_packet(recipient, packet, broadcast=True)
+                successful_sends += 1
+            except OSError:
+                failed_recipients.append(recipient)
+
+        with self._lock:
+            self._stats["transformPacketsBroadcast"] += successful_sends
+
+        for recipient in failed_recipients:
+            self._disconnect_client(recipient)
+
+        self._log(f"Broadcast transform from player {sender.player_id} to {successful_sends} other client(s)")
+
+    def _broadcast_disconnect(self, disconnected_client: ClientSession) -> None:
+        packet = {
+            "type": "disconnect",
+            "playerId": disconnected_client.player_id,
+            "serverTime": time.time(),
+        }
+
+        with self._lock:
+            recipients = list(self._clients.values())
+
+        failed_recipients: list[ClientSession] = []
+        successful_sends = 0
+        for recipient in recipients:
+            try:
+                self._send_packet(recipient, packet, broadcast=True)
+                successful_sends += 1
+            except OSError:
+                failed_recipients.append(recipient)
+
+        with self._lock:
+            self._stats["disconnectPacketsBroadcast"] += successful_sends
+
+        for recipient in failed_recipients:
+            self._disconnect_client(recipient)
+
+        self._log(
+            f"Broadcast disconnect for player {disconnected_client.player_id} "
+            f"to {successful_sends} other client(s)"
+        )
+
+    def _print_packet(self, client: ClientSession, packet: dict[str, Any]) -> None:
+        if packet.get("type") != "transform":
+            self._log(f"Packet from {client.label}: {packet}")
+            return
+
+        try:
+            x = float(packet["x"])
+            y = float(packet["y"])
+            z = float(packet["z"])
+            angle_z = float(packet["angleZ"])
+        except (KeyError, TypeError, ValueError):
+            self._log(f"Malformed transform packet from {client.label}: {packet}")
+            return
+
+        movement_type = packet.get("movementType", "normal")
+        extra_fields = []
+        for field_name in ("playerId", "cellId", "worldspaceId", "clientTime", "serverTime"):
+            if field_name in packet:
+                extra_fields.append(f"{field_name}={packet[field_name]}")
+
+        extra_details = ""
+        if extra_fields:
+            extra_details = ", " + ", ".join(extra_fields)
+
+        self._log(
+            "Transform from "
+            f"{client.label}: x={x:.2f}, y={y:.2f}, z={z:.2f}, angleZ={angle_z:.2f}, "
+            f"movementType={movement_type}{extra_details}"
+        )
+
+    def _log(self, message: str) -> None:
+        with self._log_lock:
+            listeners = list(self._log_listeners)
+
+        for listener in listeners:
+            try:
+                listener(message)
+            except Exception:
+                # Future UI listeners must not be able to break server networking.
+                pass