""" Level progression service for FreeExile. Authoritative engine managing experience distribution, piecewise level transitions, level gap efficiency decay, and tiered hybrid death penalties (PoE2 2026 Standards). """ from __future__ import annotations import math from typing import Any, Callable, Dict, List, Optional, TYPE_CHECKING if TYPE_CHECKING: from server.world.game_design_matrix_service import GameDesignMatrixService from server.world.level_progression_curve import ( LevelExpBenchmark, calculate_piecewise_exp_curve, ) from server.world.level_progression_types import ( DeathPenaltyResult, ExpAwardResult, LevelUpEvent, PlayerProgressionState, ) class LevelProgressionService: """Authoritative service for level progression and death penalties.""" def __init__( self, matrix_service: Optional[GameDesignMatrixService] = None, db_path: Optional[str] = None, ) -> None: self._matrix_service = matrix_service self._db_path = db_path self._benchmarks: Dict[int, LevelExpBenchmark] = calculate_piecewise_exp_curve() self._players: Dict[str, PlayerProgressionState] = {} self._level_up_listeners: List[Callable[[LevelUpEvent], None]] = [] self._death_listeners: List[Callable[[DeathPenaltyResult], None]] = [] def get_benchmark(self, level: int) -> Optional[LevelExpBenchmark]: """Retrieves mathematical benchmark for a specific level.""" return self._benchmarks.get(level) def get_canonical_benchmark(self, level: int) -> LevelExpBenchmark: """Retrieves canonical mathematical benchmark for level (1-100).""" return self._benchmarks.get(level) or self._benchmarks[100 if level >= 100 else 1] def get_delta_exp(self, level: int) -> int: """Returns experience needed to advance from level to level + 1.""" if level < 1 or level >= 100: return 0 bench = self._benchmarks.get(level) return bench.exp_to_next_level if bench else 0 def get_death_penalty_ratio(self, level: int) -> float: """Returns tiered percentage death penalty ratio for level.""" if level <= 60 or level >= 100: return 0.0 if level <= 80: return 0.05 if level <= 89: return 0.10 if level <= 98: return 0.15 return 0.25 if level == 99 else 0.0 def calculate_level_gap_multiplier(self, player_level: int, monster_level: int) -> float: """Calculates experience yield multiplier based on character-monster level gap.""" gap = player_level - monster_level if abs(gap) <= 5: return 1.0 if gap > 5: delta = gap - 5 return max(0.01, math.exp(-0.60 * delta)) delta = (-gap) - 5 return max(0.05, math.exp(-0.40 * delta)) def get_player_state(self, player_id: str, default_level: int = 1) -> PlayerProgressionState: """Retrieves existing progression state or registers new player at default level.""" if player_id in self._players: return self._players[player_id] level = max(1, min(100, default_level)) bench = self.get_canonical_benchmark(level) state = PlayerProgressionState( player_id=player_id, level=level, current_exp=0, exp_to_next_level=0 if level >= 100 else self.get_delta_exp(level), cumulative_exp=bench.cumulative_exp, lifetime_exp=bench.cumulative_exp, unspent_talent_points=0, total_talent_points=0, deaths_count=0, ) self._players[player_id] = state return state def set_player_state( self, player_id: str, level: int = 1, current_exp: int = 0, unspent_talent_points: int = 0, total_talent_points: int = 0, deaths_count: int = 0, ) -> PlayerProgressionState: """Explicitly sets or updates player state (for test setup and state restoration).""" valid_level = max(1, min(100, level)) bench = self.get_canonical_benchmark(valid_level) cum_exp = bench.cumulative_exp + current_exp exp_next = self.get_delta_exp(valid_level) if valid_level >= 100: cum_exp = min(bench.cumulative_exp, cum_exp) exp_next, current_exp = 0, 0 state = PlayerProgressionState( player_id=player_id, level=valid_level, current_exp=current_exp, exp_to_next_level=exp_next, cumulative_exp=cum_exp, lifetime_exp=cum_exp, unspent_talent_points=unspent_talent_points, total_talent_points=total_talent_points, deaths_count=deaths_count, ) self._players[player_id] = state return state def get_level_info(self, player_id: str) -> PlayerProgressionState: """Returns authoritative progression snapshot for player.""" return self.get_player_state(player_id) def add_level_up_listener(self, callback: Callable[[LevelUpEvent], None]) -> None: """Registers listener invoked when player character levels up.""" self._level_up_listeners.append(callback) def add_death_penalty_listener(self, callback: Callable[[DeathPenaltyResult], None]) -> None: """Registers listener invoked when death penalty is calculated.""" self._death_listeners.append(callback) def award_monster_exp( self, player_id: str, monster_level: int, arg3: Optional[int] = None, arg4: Optional[int] = None, *, base_exp: Optional[int] = None, zone_level: Optional[int] = None, player_level: Optional[int] = None, **kwargs: Any, ) -> ExpAwardResult: """Calculates and awards monster defeat experience with gap decay and level transitions.""" if base_exp is not None and base_exp < 0: base_exp = 0 raw_exp, z_lvl = self._resolve_award_args(arg3, arg4, base_exp, zone_level, monster_level) raw_exp = max(0, raw_exp) player = self._ensure_player_for_award(player_id, player_level, z_lvl, monster_level) if player.level >= 100: return self._build_level_100_award_result(player, raw_exp, monster_level) gap_mult = self.calculate_level_gap_multiplier(player.level, monster_level) awarded = max(0, int(math.floor(raw_exp * gap_mult))) return self._apply_exp_gain(player, awarded, raw_exp, gap_mult, monster_level) def _resolve_award_args( self, arg3: Optional[int], arg4: Optional[int], base_exp: Optional[int], zone_level: Optional[int], monster_level: int, ) -> tuple[int, Optional[int]]: """Resolves polymorphic positional and keyword parameters for award_monster_exp.""" resolved_base, resolved_zone = base_exp, zone_level if resolved_base is None and arg4 is not None: resolved_base = arg4 if resolved_zone is None: resolved_zone = arg3 elif resolved_base is None and arg3 is not None: resolved_base = arg3 if resolved_base is None: bench = self.get_benchmark(monster_level) resolved_base = bench.monster_benchmark_exp if bench else 25 return max(0, resolved_base), resolved_zone def _ensure_player_for_award( self, player_id: str, player_level: Optional[int], zone_level: Optional[int], monster_level: int, ) -> PlayerProgressionState: """Retrieves or initializes player progression state for award operation.""" if player_id not in self._players: init_level = player_level if player_level is not None else ( zone_level if (zone_level is not None and zone_level > 1 and zone_level == monster_level) else 1 ) return self.get_player_state(player_id, default_level=init_level) return self._players[player_id] def _build_level_100_award_result( self, player: PlayerProgressionState, raw_exp: int, monster_level: int ) -> ExpAwardResult: """Constructs zero-exp award result for characters at max cap level 100.""" return ExpAwardResult( exp_awarded=0, raw_exp=raw_exp, gap_multiplier=self.calculate_level_gap_multiplier(100, monster_level), level_gap=100 - monster_level, previous_level=100, new_level=100, leveled_up=False, levels_gained=0, current_exp=0, exp_to_next_level=0, ) def _compute_level_advancement( self, start_level: int, initial_exp: int ) -> tuple[int, int, int, int]: """Resolves level-up thresholds and returns (level, exp, exp_next, gained).""" lvl, exp, gained = start_level, initial_exp, 0 exp_next = self.get_delta_exp(lvl) while lvl < 100 and exp >= exp_next: exp -= exp_next lvl += 1 gained += 1 exp_next = self.get_delta_exp(lvl) if lvl >= 100: lvl, exp, exp_next = 100, 0, 0 return lvl, exp, exp_next, gained def _apply_exp_gain( self, player: PlayerProgressionState, exp_awarded: int, raw_exp: int, gap_multiplier: float, monster_level: int, ) -> ExpAwardResult: """Applies awarded experience, processes single/multi level-ups, and updates player.""" prev_level = player.level curr_lvl, curr_exp, exp_next, gained = self._compute_level_advancement( prev_level, player.current_exp + exp_awarded ) cum_exp = player.cumulative_exp + exp_awarded life_exp = player.lifetime_exp + exp_awarded if curr_lvl >= 100: max_cum = self.get_canonical_benchmark(100).cumulative_exp cum_exp, life_exp = min(max_cum, cum_exp), min(max_cum, life_exp) exp_next, curr_exp = 0, 0 self._players[player.player_id] = PlayerProgressionState( player_id=player.player_id, level=curr_lvl, current_exp=curr_exp, exp_to_next_level=exp_next, cumulative_exp=cum_exp, lifetime_exp=life_exp, unspent_talent_points=player.unspent_talent_points + gained, total_talent_points=player.total_talent_points + gained, deaths_count=player.deaths_count, ) if gained > 0: self._dispatch_level_up(player.player_id, prev_level, curr_lvl, gained) return ExpAwardResult( exp_awarded=exp_awarded, raw_exp=raw_exp, gap_multiplier=gap_multiplier, level_gap=abs(prev_level - monster_level), previous_level=prev_level, new_level=curr_lvl, leveled_up=gained > 0, levels_gained=gained, current_exp=curr_exp, exp_to_next_level=exp_next, ) def _dispatch_level_up( self, player_id: str, old_level: int, new_level: int, levels_gained: int ) -> None: """Dispatches level-up notification to all registered listeners.""" stats = { "stat_points": 5.0 * levels_gained, "passive_points": 1.0 * levels_gained, "hp": 28.0 * levels_gained, } event = LevelUpEvent( player_id=player_id, old_level=old_level, new_level=new_level, talent_points_awarded=levels_gained, stats_gained=stats, ) for listener in self._level_up_listeners: listener(event) def apply_death_penalty(self, player_id: str) -> DeathPenaltyResult: """Calculates and applies tiered death penalty with zero-floor safe rule.""" player = self.get_player_state(player_id) ratio = self.get_death_penalty_ratio(player.level) delta = self.get_delta_exp(player.level) nominal_loss = int(math.floor(delta * ratio)) exp_lost = min(player.current_exp, nominal_loss) new_exp = max(0, player.current_exp - nominal_loss) self._players[player_id] = PlayerProgressionState( player_id=player.player_id, level=player.level, current_exp=new_exp, exp_to_next_level=player.exp_to_next_level, cumulative_exp=player.cumulative_exp - exp_lost, lifetime_exp=player.lifetime_exp, unspent_talent_points=player.unspent_talent_points, total_talent_points=player.total_talent_points, deaths_count=player.deaths_count + 1, ) result = self._build_death_result(player, exp_lost, ratio, new_exp) for listener in self._death_listeners: listener(result) return result def _build_death_result( self, player: PlayerProgressionState, exp_lost: int, ratio: float, new_exp: int ) -> DeathPenaltyResult: """Constructs DeathPenaltyResult with safe floor and no de-leveling.""" return DeathPenaltyResult( player_id=player.player_id, level=player.level, exp_lost=exp_lost, penalty_ratio=ratio, current_exp_before=player.current_exp, current_exp_after=new_exp, de_leveled=False, )