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 "")