server/npc_authority.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Iterable
+
+
+@dataclass(frozen=True, order=True)
+class ScopeKey:
+    cell_id: str
+    worldspace_id: str = ""
+
+
+@dataclass(frozen=True)
+class AuthorityAssignment:
+    scope: ScopeKey
+    player_id: int
+    epoch: int
+
+
+@dataclass(frozen=True)
+class AuthorityChange:
+    scope: ScopeKey
+    previous_player_id: int
+    player_id: int
+    epoch: int
+
+
+class NpcAuthorityManager:
+    """Server-owned NPC simulation authority keyed by exact cell/worldspace scope."""
+
+    def __init__(self) -> None:
+        self._assignments: dict[ScopeKey, AuthorityAssignment] = {}
+        self._last_epoch: dict[ScopeKey, int] = {}
+
+    def clear(self) -> None:
+        self._assignments.clear()
+        self._last_epoch.clear()
+
+    def get(self, scope: ScopeKey) -> AuthorityAssignment | None:
+        return self._assignments.get(scope)
+
+    def assignments(self) -> tuple[AuthorityAssignment, ...]:
+        return tuple(self._assignments[key] for key in sorted(self._assignments))
+
+    def authorize(self, player_id: int, scope: ScopeKey, epoch: int) -> bool:
+        assignment = self._assignments.get(scope)
+        return (
+            assignment is not None
+            and assignment.player_id == int(player_id)
+            and assignment.epoch == int(epoch)
+        )
+
+    def reconcile(self, players: Iterable[tuple[int, ScopeKey]]) -> tuple[AuthorityChange, ...]:
+        candidates: dict[ScopeKey, list[int]] = {}
+        for player_id, scope in players:
+            if int(player_id) <= 0:
+                continue
+            candidates.setdefault(scope, []).append(int(player_id))
+
+        desired: dict[ScopeKey, int] = {
+            scope: min(player_ids)
+            for scope, player_ids in candidates.items()
+            if player_ids
+        }
+
+        changes: list[AuthorityChange] = []
+        all_scopes = set(self._assignments) | set(desired)
+        for scope in sorted(all_scopes):
+            previous = self._assignments.get(scope)
+            previous_player_id = previous.player_id if previous is not None else 0
+            next_player_id = desired.get(scope, 0)
+            if previous_player_id == next_player_id:
+                continue
+
+            epoch = self._last_epoch.get(scope, 0) + 1
+            self._last_epoch[scope] = epoch
+            if next_player_id == 0:
+                self._assignments.pop(scope, None)
+            else:
+                self._assignments[scope] = AuthorityAssignment(scope, next_player_id, epoch)
+            changes.append(
+                AuthorityChange(
+                    scope=scope,
+                    previous_player_id=previous_player_id,
+                    player_id=next_player_id,
+                    epoch=epoch,
+                )
+            )
+
+        return tuple(changes)
+
+
+def scope_from_transform(state: dict[str, Any] | None) -> ScopeKey | None:
+    if not isinstance(state, dict):
+        return None
+    cell_id = state.get("cellId")
+    worldspace_id = state.get("worldspaceId", "")
+    if not isinstance(cell_id, str) or not cell_id:
+        return None
+    if not isinstance(worldspace_id, str):
+        return None
+    return ScopeKey(cell_id.upper().zfill(8), worldspace_id.upper().zfill(8) if worldspace_id else "")