server/server_core.py

@@ -13,6 +13,7 @@ from typing import Any
 from ban_store import BanStore
 from client_session import ClientSession
 from lan_discovery import DISCOVERY_PORT, LanDiscoveryResponder
+from npc_authority import NpcAuthorityManager, ScopeKey, scope_from_transform
 from world_state_presets import normalize_fw_console_arg, relay_weather_form_id
 
 HOST = "0.0.0.0"
@@ -152,12 +153,6 @@ def _scope_from_state(state: dict[str, Any] | None) -> tuple[str, str, float, fl
 
 
 def states_share_interest(a: dict[str, Any] | None, b: dict[str, Any] | None) -> bool:
-    """Return True when two states should see each other.
-
-    Missing scope data intentionally falls back to True for compatibility with older clients.
-    Interiors are matched by exact cell. Exteriors can also see nearby peers in the same
-    worldspace so cell-border crossings do not make players pop in and out abruptly.
-    """
     a_scope = _scope_from_state(a)
     b_scope = _scope_from_state(b)
     if a_scope is None or b_scope is None:
@@ -226,6 +221,8 @@ class FalloutTogetherServer:
         self._world_state_host_player_id: int | None = None
         self._server_world_state: dict[str, str] = {}
         self._last_npc_state: dict[str, Any] | None = None
+        self._last_npc_state_by_scope: dict[ScopeKey, dict[str, Any]] = {}
+        self._npc_authority = NpcAuthorityManager()
         self._connect_attempts: dict[str, deque[float]] = {}
 
         self._stats: dict[str, int] = {
@@ -246,6 +243,8 @@ class FalloutTogetherServer:
             "worldStatePacketsBroadcast": 0,
             "npcStatePacketsReceived": 0,
             "npcStatePacketsBroadcast": 0,
+            "npcAuthorityChanges": 0,
+            "npcAuthorityRejects": 0,
             "combatHitsReceived": 0,
             "combatHitsRouted": 0,
             "worldStateHostPacketsBroadcast": 0,
@@ -335,6 +334,7 @@ class FalloutTogetherServer:
                     "pendingConnections": pending,
                     "nextPlayerId": self._next_player_id,
                     "protocolVersion": PROTOCOL_VERSION,
+                    "npcAuthorityScopes": len(self._npc_authority.assignments()),
                 }
             )
         return stats
@@ -442,6 +442,8 @@ class FalloutTogetherServer:
         self._next_player_id = 1
         self._world_state_host_player_id = None
         self._last_npc_state = None
+        self._last_npc_state_by_scope.clear()
+        self._npc_authority.clear()
         self._connect_attempts.clear()
 
         self._log(f"Commonwealth Online server listening on {self.host}:{self.port}")
@@ -603,7 +605,14 @@ class FalloutTogetherServer:
                         "serverName": self.server_name,
                         "serverDescription": self.server_description,
                         "protocolVersion": PROTOCOL_VERSION,
-                        "capabilities": ["interest-v1", "hello-v2", "bounded-framing", "rate-limit-v1", "movement-correction-v1"],
+                        "capabilities": [
+                            "interest-v1",
+                            "hello-v2",
+                            "bounded-framing",
+                            "rate-limit-v1",
+                            "movement-correction-v1",
+                            "npc-authority-epoch-v1",
+                        ],
                     },
                 )
                 buffer = b""
@@ -720,23 +729,24 @@ class FalloutTogetherServer:
                 self._reject_packet(client, "Malformed transform")
                 return False
             accepted_monotonic = time.monotonic()
-            movement_valid, movement_reason = self._validate_transform_movement(
+            movement_valid, movement_reason = self._validate_transform_movement(client, normalized, accepted_monotonic)
-                client,
-                normalized,
-                accepted_monotonic,
-            )
             if not movement_valid:
                 with self._lock:
                     self._stats["movementPacketsRejected"] += 1
                 self._reject_packet(client, movement_reason)
                 self._send_position_correction(client, movement_reason)
                 return False
+            previous_scope = scope_from_transform(client.last_transform)
             normalized["playerId"] = client.player_id
             normalized["serverTime"] = time.time()
             client.record_transform(normalized, accepted_monotonic)
             with self._lock:
                 self._stats["transformPacketsReceived"] += 1
             self._broadcast_transform(client, normalized)
+            self._reconcile_npc_authority()
+            current_scope = scope_from_transform(normalized)
+            if current_scope is not None and current_scope != previous_scope:
+                self._send_npc_authority_for_client(client, current_scope)
             return True
 
         if packet_type == "worldState":
@@ -755,18 +765,47 @@ class FalloutTogetherServer:
             return True
 
         if packet_type == "npcState":
-            if not self._is_world_host(client):
-                self._reject_packet(client, "npcState from non-authority client", warning=False)
-                return False
             normalized = self._normalize_npc_state_packet(packet)
             if normalized is None:
                 self._reject_packet(client, "Malformed npcState")
                 return False
+
+            if client.protocol_version >= PROTOCOL_VERSION:
+                scope = self._npc_scope_from_packet(normalized)
+                epoch = normalized.get("authorityEpoch")
+                if scope is None or not _is_int(epoch, 1):
+                    with self._lock:
+                        self._stats["npcAuthorityRejects"] += 1
+                    self._reject_packet(client, "Protocol V2 npcState missing valid authority scope/epoch")
+                    return False
+                if not self._npc_authority.authorize(client.player_id, scope, epoch):
+                    with self._lock:
+                        self._stats["npcAuthorityRejects"] += 1
+                    self._reject_packet(client, "Stale or unauthorized npcState authority epoch")
+                    return False
+                if any(
+                    npc.get("cellId") != scope.cell_id or npc.get("worldspaceId", "") != scope.worldspace_id
+                    for npc in normalized.get("npcs", [])
+                ):
+                    with self._lock:
+                        self._stats["npcAuthorityRejects"] += 1
+                    self._reject_packet(client, "npcState contains NPCs outside declared authority scope")
+                    return False
+                normalized["authorityCellId"] = scope.cell_id
+                normalized["authorityWorldspaceId"] = scope.worldspace_id
+                with self._lock:
+                    self._last_npc_state_by_scope[scope] = dict(normalized)
+            else:
+                if not self._is_world_host(client):
+                    self._reject_packet(client, "Legacy npcState from non-authority client", warning=False)
+                    return False
+                with self._lock:
+                    self._last_npc_state = dict(normalized)
+
             normalized["playerId"] = client.player_id
             normalized["serverTime"] = time.time()
             normalized["fullReplace"] = True
             with self._lock:
-                self._last_npc_state = dict(normalized)
                 self._stats["npcStatePacketsReceived"] += 1
             self._broadcast_npc_state(client, normalized)
             return True
@@ -794,11 +833,7 @@ class FalloutTogetherServer:
     def _handle_hello(self, client: ClientSession, packet: dict[str, Any]) -> bool:
         version = packet.get("protocolVersion")
         if not _is_int(version, 1, 0xFFFF) or version != PROTOCOL_VERSION:
-            self._end_client_session(
+            self._end_client_session(client, code=SESSION_ENDED_PROTOCOL, reason=f"Server requires protocol {PROTOCOL_VERSION}.")
-                client,
-                code=SESSION_ENDED_PROTOCOL,
-                reason=f"Server requires protocol {PROTOCOL_VERSION}.",
-            )
             return False
         if not self._activate_client(client, version):
             self._end_client_session(client, code=SESSION_ENDED_FULL, reason="Server is full.")
@@ -854,42 +889,30 @@ class FalloutTogetherServer:
             normalized["actionEvents"] = _normalize_action_events(packet["actionEvents"])
         return normalized
 
-    def _validate_transform_movement(
+    def _validate_transform_movement(self, client: ClientSession, packet: dict[str, Any], now_monotonic: float) -> tuple[bool, str]:
-        self,
-        client: ClientSession,
-        packet: dict[str, Any],
-        now_monotonic: float,
-    ) -> tuple[bool, str]:
         previous, previous_monotonic = client.get_last_transform_anchor()
         if previous is None or previous_monotonic is None:
             return True, "first transform"
-
         movement_type = packet.get("movementType", "normal")
         if movement_type in MOVEMENT_TRANSITION_TYPES:
             return True, f"explicit {movement_type} transition"
-
         previous_cell = previous.get("cellId", "")
         current_cell = packet.get("cellId", "")
         previous_world = previous.get("worldspaceId", "")
         current_world = packet.get("worldspaceId", "")
         if previous_cell != current_cell or previous_world != current_world:
             return False, "scope changed without an explicit movement transition"
-
         elapsed = now_monotonic - previous_monotonic
         if not math.isfinite(elapsed) or elapsed < 0.0:
             elapsed = 0.0
         elapsed = min(elapsed, MAX_MOVEMENT_VALIDATION_ELAPSED_SECONDS)
-
         dx = packet["x"] - previous["x"]
         dy = packet["y"] - previous["y"]
         dz = packet["z"] - previous["z"]
         distance = math.sqrt((dx * dx) + (dy * dy) + (dz * dz))
         allowed_distance = MOVEMENT_GRACE_DISTANCE + (MAX_NORMAL_MOVEMENT_SPEED * elapsed)
         if not math.isfinite(distance) or distance > allowed_distance:
-            return (
+            return False, f"normal movement exceeded server envelope: distance={distance:.1f}, allowed={allowed_distance:.1f}, elapsed={elapsed:.3f}s"
-                False,
-                f"normal movement exceeded server envelope: distance={distance:.1f}, allowed={allowed_distance:.1f}, elapsed={elapsed:.3f}s",
-            )
         return True, "normal movement accepted"
 
     def _send_position_correction(self, client: ClientSession, reason: str) -> None:
@@ -959,8 +982,30 @@ class FalloutTogetherServer:
             clean_npcs.append(clean)
         normalized = dict(packet)
         normalized["npcs"] = clean_npcs
+        if "authorityEpoch" in packet:
+            if not _is_int(packet["authorityEpoch"], 1):
+                return None
+            normalized["authorityEpoch"] = int(packet["authorityEpoch"])
+        if "authorityCellId" in packet:
+            if not _is_hex_form_id(packet["authorityCellId"], allow_zero=False):
+                return None
+            normalized["authorityCellId"] = _normalize_hex_form_id(packet["authorityCellId"])
+        if "authorityWorldspaceId" in packet:
+            world = packet["authorityWorldspaceId"]
+            if not _is_hex_form_id(world, allow_empty=True, allow_zero=True):
+                return None
+            normalized["authorityWorldspaceId"] = _normalize_hex_form_id(world) if world else ""
         return normalized
 
+    def _npc_scope_from_packet(self, packet: dict[str, Any]) -> ScopeKey | None:
+        cell = packet.get("authorityCellId")
+        world = packet.get("authorityWorldspaceId", "")
+        if not _is_hex_form_id(cell, allow_zero=False):
+            return None
+        if not _is_hex_form_id(world, allow_empty=True, allow_zero=True):
+            return None
+        return ScopeKey(_normalize_hex_form_id(cell), _normalize_hex_form_id(world) if world else "")
+
     def _normalize_combat_hit_packet(self, packet: dict[str, Any]) -> dict[str, Any] | None:
         target = packet.get("targetPlayerId")
         sequence = packet.get("sequence")
@@ -988,6 +1033,52 @@ class FalloutTogetherServer:
         with self._lock:
             return self._world_state_host_player_id == client.player_id
 
+    def _authority_packet(self, assignment_player_id: int, epoch: int, scope: ScopeKey) -> dict[str, Any]:
+        return {
+            "type": "npcAuthority",
+            "authorityPlayerId": assignment_player_id,
+            "authorityEpoch": epoch,
+            "authorityCellId": scope.cell_id,
+            "authorityWorldspaceId": scope.worldspace_id,
+            "serverTime": time.time(),
+        }
+
+    def _reconcile_npc_authority(self) -> None:
+        with self._lock:
+            players = []
+            for client in self._active_clients_locked():
+                if client.protocol_version < PROTOCOL_VERSION:
+                    continue
+                scope = scope_from_transform(client.last_transform)
+                if scope is not None:
+                    players.append((client.player_id, scope))
+            changes = self._npc_authority.reconcile(players)
+            for change in changes:
+                self._last_npc_state_by_scope.pop(change.scope, None)
+        for change in changes:
+            packet = self._authority_packet(change.player_id, change.epoch, change.scope)
+            successful, failed = self._broadcast_to_active(packet, v2_only=True)
+            with self._lock:
+                self._stats["npcAuthorityChanges"] += 1
+            for client in failed:
+                self._disconnect_client(client)
+            self._log(
+                f"NPC authority scope {change.scope.cell_id}/{change.scope.worldspace_id or '<interior>'}: "
+                f"{change.previous_player_id} -> {change.player_id}, epoch {change.epoch} "
+                f"(notified {successful} client(s))."
+            )
+
+    def _send_npc_authority_for_client(self, client: ClientSession, scope: ScopeKey) -> None:
+        if client.protocol_version < PROTOCOL_VERSION:
+            return
+        assignment = self._npc_authority.get(scope)
+        if assignment is None:
+            return
+        try:
+            self._send_packet(client, self._authority_packet(assignment.player_id, assignment.epoch, scope))
+        except (OSError, ValueError):
+            self._disconnect_client(client)
+
     def _send_packet(self, client: ClientSession, packet: dict[str, Any], *, broadcast: bool = False) -> None:
         encoded = json.dumps(packet, separators=(",", ":"), allow_nan=False).encode("utf-8") + b"\n"
         if len(encoded) > MAX_PACKET_CHARS:
@@ -1001,10 +1092,7 @@ class FalloutTogetherServer:
 
     def _send_existing_transforms_to_client(self, new_client: ClientSession) -> None:
         with self._lock:
-            peers = [
+            peers = [client for client in self._active_clients_locked() if client.connection != new_client.connection and client.last_transform is not None]
-                client for client in self._active_clients_locked()
-                if client.connection != new_client.connection and client.last_transform is not None
-            ]
         successful = 0
         for peer in peers:
             if not states_share_interest(peer.last_transform, new_client.last_transform):
@@ -1073,8 +1161,13 @@ class FalloutTogetherServer:
             self._disconnect_client(client)
 
     def _send_existing_npc_state_to_client(self, client: ClientSession) -> None:
-        with self._lock:
+        if client.protocol_version >= PROTOCOL_VERSION:
-            snapshot = dict(self._last_npc_state) if self._last_npc_state is not None else None
+            scope = scope_from_transform(client.last_transform)
+            with self._lock:
+                snapshot = dict(self._last_npc_state_by_scope[scope]) if scope in self._last_npc_state_by_scope else None
+        else:
+            with self._lock:
+                snapshot = dict(self._last_npc_state) if self._last_npc_state is not None else None
         if snapshot is None or snapshot.get("playerId") == client.player_id:
             return
         scoped = self._npc_packet_for_recipient(snapshot, client)
@@ -1113,11 +1206,7 @@ class FalloutTogetherServer:
         if snapshot.get("timeHHmm"):
             packets.append({"type": "serverWorldState", "timeHHmm": snapshot["timeHHmm"], "serverTime": time.time()})
         if snapshot.get("weatherConsoleArg"):
-            weather_packet: dict[str, Any] = {
+            weather_packet: dict[str, Any] = {"type": "serverWorldState", "weatherConsoleArg": snapshot["weatherConsoleArg"], "serverTime": time.time()}
-                "type": "serverWorldState",
-                "weatherConsoleArg": snapshot["weatherConsoleArg"],
-                "serverTime": time.time(),
-            }
             if snapshot.get("weatherFormId"):
                 weather_packet["weatherFormId"] = snapshot["weatherFormId"]
             packets.append(weather_packet)
@@ -1135,11 +1224,19 @@ class FalloutTogetherServer:
         for client in {failed_client.connection: failed_client for failed_client in failed}.values():
             self._disconnect_client(client)
 
-    def _broadcast_to_active(self, packet: dict[str, Any], *, exclude: ClientSession | None = None) -> tuple[int, list[ClientSession]]:
+    def _broadcast_to_active(
+        self,
+        packet: dict[str, Any],
+        *,
+        exclude: ClientSession | None = None,
+        v2_only: bool = False,
+    ) -> tuple[int, list[ClientSession]]:
         with self._lock:
             recipients = [
-                client for client in self._active_clients_locked()
+                client
-                if exclude is None or client.connection != exclude.connection
+                for client in self._active_clients_locked()
+                if (exclude is None or client.connection != exclude.connection)
+                and (not v2_only or client.protocol_version >= PROTOCOL_VERSION)
             ]
         successful = 0
         failed: list[ClientSession] = []
@@ -1199,6 +1296,7 @@ class FalloutTogetherServer:
             return
         self._log(f"Client disconnected: {client.label} (player {client.player_id})")
         self._broadcast_disconnect(client)
+        self._reconcile_npc_authority()
         if was_host:
             self._reassign_world_state_host()
 
@@ -1253,10 +1351,7 @@ class FalloutTogetherServer:
     def _reject_packet(self, client: ClientSession, reason: str, *, warning: bool = True) -> None:
         with self._lock:
             self._stats["packetsRejected"] += 1
-        self._log(
+        self._log(f"Rejected packet from {client.label} (player {client.player_id}): {reason}", level="warning" if warning else "debug")
-            f"Rejected packet from {client.label} (player {client.player_id}): {reason}",
-            level="warning" if warning else "debug",
-        )
 
     def _should_log(self, level: str) -> bool:
         configured = _LOG_LEVELS.get(self.log_verbosity, _LOG_LEVELS[DEFAULT_LOG_VERBOSITY])