""" Empirical Challenge & Stress-Test Harness for LevelProgressionService (Milestone M2). Author: challenger_m2_progression_1 Target: server/world/level_progression_service.py & server/world/level_progression_types.py """ from __future__ import annotations import math import sys import os from typing import List, Dict, Any, Tuple # Ensure project root is in sys.path PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../")) if PROJECT_ROOT not in sys.path: sys.path.insert(0, PROJECT_ROOT) from server.world.level_progression_service import LevelProgressionService from server.world.level_progression_types import ( PlayerProgressionState, ExpAwardResult, DeathPenaltyResult, LevelUpEvent, ) class EmpiricalProgressionChallenger: def __init__(self) -> None: self.service = LevelProgressionService() self.passed_tests = 0 self.failed_tests = 0 self.test_log: List[str] = [] def log(self, msg: str) -> None: self.test_log.append(msg) print(msg) def assert_true(self, condition: bool, description: str) -> None: if condition: self.passed_tests += 1 self.log(f" [PASS] {description}") else: self.failed_tests += 1 self.log(f" [FAIL] {description}") # ========================================================================= # CHALLENGE 1: LEVEL GAP EXPONENTIAL DECAY # ========================================================================= def challenge_level_gap_decay(self) -> None: self.log("\n=== CHALLENGE 1: Level Gap Exponential Decay (Delta 0 to 50) ===") # Reference anchor: player_level = 80 # Delta = player_level - monster_level (monster is Delta levels lower) p_lvl = 80 for delta in range(0, 51): m_lvl = p_lvl - delta mult = self.service.calculate_level_gap_multiplier(p_lvl, m_lvl) if delta <= 5: # Delta <= 5 is exactly 1.0 self.assert_true( mult == 1.0, f"Delta={delta} (player {p_lvl}, monster {m_lvl}): expected 1.0, got {mult}", ) else: # Delta > 5 strictly follows exp(-0.60 * (delta - 5)) with floor 0.01 expected_unclamped = math.exp(-0.60 * (delta - 5)) expected = max(0.01, expected_unclamped) diff = abs(mult - expected) self.assert_true( diff < 1e-9, f"Delta={delta}: expected {expected:.6f}, got {mult:.6f} (diff {diff:.2e})", ) # Requirement: Delta >= 10 yields <= 0.05 if delta >= 10: self.assert_true( mult <= 0.05, f"Delta={delta} >= 10: expected <= 0.05, got {mult:.6f}", ) # Monotonicity test: multiplier must be non-increasing as delta goes from 0 to 50 multipliers = [ self.service.calculate_level_gap_multiplier(p_lvl, p_lvl - d) for d in range(51) ] is_monotonic = all(multipliers[i] <= multipliers[i - 1] for i in range(1, len(multipliers))) self.assert_true( is_monotonic, "Level gap decay multiplier is strictly monotonically non-increasing across Delta 0..50", ) # Reverse gap (underleveled player vs high-level monster: anti-boosting) self.log("--- Reverse Level Gap (Underleveled Anti-Boosting) ---") for delta_rev in range(1, 30): # player lower than monster by delta_rev p_low = 50 m_high = p_low + delta_rev mult_rev = self.service.calculate_level_gap_multiplier(p_low, m_high) if delta_rev <= 5: self.assert_true(mult_rev == 1.0, f"Underleveled delta={delta_rev} <= 5 gives 1.0") else: expected_rev = max(0.05, math.exp(-0.40 * (delta_rev - 5))) self.assert_true( abs(mult_rev - expected_rev) < 1e-9, f"Underleveled delta={delta_rev}: expected {expected_rev:.6f}, got {mult_rev:.6f}", ) # ========================================================================= # CHALLENGE 2: LEVEL TRANSITION & LEVEL CAP # ========================================================================= def challenge_level_transitions_and_cap(self) -> None: self.log("\n=== CHALLENGE 2: Level Transitions & Cap (1 EXP, 1k EXP, 10M EXP, Lv 99->100, Lv 100 Cap) ===") # Test 2.1: Gaining 1 EXP at Level 1 p_1 = "test_player_1exp" res_1 = self.service.award_monster_exp(p_1, monster_level=1, base_exp=1, player_level=1) self.assert_true( res_1.exp_awarded == 1 and res_1.current_exp == 1 and res_1.new_level == 1 and not res_1.leveled_up, f"1 EXP gain: level={res_1.new_level} (exp 1), leveled_up={res_1.leveled_up}", ) # Test 2.2: Gaining 1,000 EXP at Level 1 (Lv 1 delta = 600) p_1k = "test_player_1k" res_1k = self.service.award_monster_exp(p_1k, monster_level=1, base_exp=1000, player_level=1) # 1000 - 600 = 400 rollover exp at Level 2 d2 = self.service.get_delta_exp(2) self.assert_true( res_1k.new_level == 2 and res_1k.current_exp == 400 and res_1k.leveled_up and res_1k.levels_gained == 1 and res_1k.exp_to_next_level == d2, f"1,000 EXP gain: new_level={res_1k.new_level}, current_exp={res_1k.current_exp}, rollover=400", ) # Test 2.3: Gaining 10,000,000 EXP at Level 1 (Simulate massive cascade) p_10m = "test_player_10m" # Pre-calculate ground truth level for 10M EXP starting from Lv 1 rem = 10_000_000 gt_lvl = 1 while gt_lvl < 100 and rem >= self.service.get_delta_exp(gt_lvl): rem -= self.service.get_delta_exp(gt_lvl) gt_lvl += 1 res_10m = self.service.award_monster_exp(p_10m, monster_level=1, base_exp=10_000_000, player_level=1) # Note: monster_level=1 vs player_level=1 at the instant of award, so gap=0, mult=1.0 self.assert_true( res_10m.new_level == gt_lvl and res_10m.current_exp == rem and res_10m.levels_gained == gt_lvl - 1, f"10,000,000 EXP jump: advanced from Lv 1 to Lv {res_10m.new_level} with {res_10m.current_exp} rollover (expected Lv {gt_lvl} + {rem})", ) # Test 2.4: Multi-level transition: Level 1 jumps exactly to Level 5 p_jump = "test_player_jump_lv5" # Total needed to reach Lv 5 with 123 remainder exp_to_lv5 = sum(self.service.get_delta_exp(lvl) for lvl in range(1, 5)) + 123 res_jump = self.service.award_monster_exp(p_jump, monster_level=1, base_exp=exp_to_lv5, player_level=1) state_jump = self.service.get_level_info(p_jump) self.assert_true( res_jump.new_level == 5 and res_jump.current_exp == 123 and res_jump.levels_gained == 4 and state_jump.unspent_talent_points == 4 and state_jump.total_talent_points == 4, f"Multi-level jump 1->5: new_level={res_jump.new_level}, current_exp={res_jump.current_exp}, talents={state_jump.unspent_talent_points}", ) # Test 2.5: Level 99 to 100 transition p_99 = "test_player_lv99" d99 = self.service.get_delta_exp(99) self.service.set_player_state(p_99, level=99, current_exp=d99 - 50) # Award 50 EXP to trigger exactly Lv 100 res_99_exact = self.service.award_monster_exp(p_99, monster_level=99, base_exp=50) self.assert_true( res_99_exact.new_level == 100 and res_99_exact.current_exp == 0 and res_99_exact.exp_to_next_level == 0 and res_99_exact.leveled_up and res_99_exact.levels_gained == 1, f"Level 99->100 exact transition: new_level={res_99_exact.new_level}, current_exp={res_99_exact.current_exp}", ) # Test 2.6: Level 99 to 100 with massive excess EXP (no overflow) p_99_over = "test_player_lv99_overflow" self.service.set_player_state(p_99_over, level=99, current_exp=d99 - 100) res_99_over = self.service.award_monster_exp(p_99_over, monster_level=99, base_exp=100 + 50_000_000) self.assert_true( res_99_over.new_level == 100 and res_99_over.current_exp == 0 and res_99_over.exp_to_next_level == 0 and res_99_over.levels_gained == 1, f"Level 99->100 with massive surplus EXP: new_level={res_99_over.new_level}, current_exp={res_99_over.current_exp} (properly clamped to 0)", ) # Test 2.7: Kills at Level 100 award exactly 0 EXP p_100 = "test_player_lv100_cap" self.service.set_player_state(p_100, level=100, current_exp=0) for base in [1, 1_000, 1_000_000, 100_000_000]: res_100 = self.service.award_monster_exp(p_100, monster_level=100, base_exp=base) self.assert_true( res_100.exp_awarded == 0 and res_100.effective_exp == 0 and not res_100.leveled_up and res_100.levels_gained == 0 and res_100.new_level == 100 and res_100.current_exp == 0, f"Kills at Lv 100 (base={base}): exp_awarded={res_100.exp_awarded}, new_level={res_100.new_level}", ) # ========================================================================= # CHALLENGE 3: TIERED DEATH PENALTY MATRIX (ALL 100 LEVELS) # ========================================================================= def challenge_tiered_death_penalty_matrix(self) -> None: self.log("\n=== CHALLENGE 3: Tiered Death Penalty Matrix Across All 100 Levels ===") mismatches = 0 for lvl in range(1, 101): ratio = self.service.get_death_penalty_ratio(lvl) if 1 <= lvl <= 60: expected_ratio = 0.0 elif 61 <= lvl <= 80: expected_ratio = 0.05 elif 81 <= lvl <= 89: expected_ratio = 0.10 elif 90 <= lvl <= 98: expected_ratio = 0.15 elif lvl == 99: expected_ratio = 0.25 elif lvl == 100: expected_ratio = 0.0 else: expected_ratio = 0.0 if abs(ratio - expected_ratio) > 1e-6: mismatches += 1 self.log(f" [MISMATCH] Level {lvl}: expected ratio {expected_ratio}, got {ratio}") # Simulate actual death with 50% EXP at this level pid = f"death_test_lv{lvl}" d_lvl = self.service.get_delta_exp(lvl) half_exp = d_lvl // 2 self.service.set_player_state(pid, level=lvl, current_exp=half_exp) death_res = self.service.apply_death_penalty(pid) nominal_loss = int(math.floor(d_lvl * expected_ratio)) expected_loss = min(half_exp, nominal_loss) expected_new_exp = max(0, half_exp - nominal_loss) if lvl == 100: # Lv 100 always has 0 current_exp and 0 penalty expected_loss = 0 expected_new_exp = 0 self.assert_true( abs(death_res.penalty_ratio - expected_ratio) < 1e-6 and death_res.exp_lost == expected_loss and death_res.current_exp_after == expected_new_exp and not death_res.de_leveled and death_res.level == lvl, f"Level {lvl} death penalty: ratio={death_res.penalty_ratio} (expected {expected_ratio}), lost={death_res.exp_lost} (expected {expected_loss}), after={death_res.current_exp_after}", ) self.assert_true( mismatches == 0, f"100% of levels 1-100 match exact death penalty tier specification (0 mismatches out of 100)", ) # ========================================================================= # CHALLENGE 4: SAFE FLOOR INVARIANT (NO DE-LEVELING & CLAMPING) # ========================================================================= def challenge_safe_floor_invariant(self) -> None: self.log("\n=== CHALLENGE 4: Safe Floor Invariant (Zero EXP & Clamping Tests) ===") # Test 4.1: Dying at 0% EXP at every penalty tier test_levels = [10, 65, 85, 95, 99, 100] for lvl in test_levels: pid = f"zero_exp_lv{lvl}" self.service.set_player_state(pid, level=lvl, current_exp=0) res = self.service.apply_death_penalty(pid) self.assert_true( res.exp_lost == 0 and res.current_exp_after == 0 and not res.de_leveled and res.level == lvl and self.service.get_level_info(pid).level == lvl, f"Dying at 0% EXP at Level {lvl}: lost=0, new_exp=0, de_leveled=False, level={lvl}", ) # Test 4.2: Dying at 5% EXP when penalty is 25% (Level 99) pid_99 = "clamp_lv99_5pct" d99 = self.service.get_delta_exp(99) five_pct_exp = int(math.floor(d99 * 0.05)) self.service.set_player_state(pid_99, level=99, current_exp=five_pct_exp) res_99 = self.service.apply_death_penalty(pid_99) self.assert_true( res_99.exp_lost == five_pct_exp and res_99.current_exp_after == 0 and not res_99.de_leveled and res_99.level == 99, f"Level 99 at 5% EXP (25% penalty): lost exactly {res_99.exp_lost} (5%), clamped to 0, de_leveled=False", ) # Test 4.3: Dying at 2% EXP when penalty is 15% (Level 95) pid_95 = "clamp_lv95_2pct" d95 = self.service.get_delta_exp(95) two_pct_exp = int(math.floor(d95 * 0.02)) self.service.set_player_state(pid_95, level=95, current_exp=two_pct_exp) res_95 = self.service.apply_death_penalty(pid_95) self.assert_true( res_95.exp_lost == two_pct_exp and res_95.current_exp_after == 0 and not res_95.de_leveled and res_95.level == 95, f"Level 95 at 2% EXP (15% penalty): lost exactly {res_95.exp_lost} (2%), clamped to 0, de_leveled=False", ) # Test 4.4: 100 consecutive deaths streak at Level 99 pid_streak = "death_streak_lv99" self.service.set_player_state(pid_streak, level=99, current_exp=int(d99 * 0.60)) for i in range(1, 101): res_s = self.service.apply_death_penalty(pid_streak) self.assert_true( not res_s.de_leveled and res_s.level == 99 and res_s.current_exp_after >= 0, f"Death #{i} in streak: level={res_s.level}, exp={res_s.current_exp_after}, de_leveled={res_s.de_leveled}", ) final_state = self.service.get_level_info(pid_streak) self.assert_true( final_state.level == 99 and final_state.current_exp == 0 and final_state.deaths_count == 100, f"After 100 consecutive deaths: level={final_state.level} (expected 99), exp={final_state.current_exp} (expected 0), deaths={final_state.deaths_count}", ) # Test 4.5: Cumulative EXP floor preservation bench_99 = self.service.get_benchmark(99) base_cum_99 = bench_99.cumulative_exp if bench_99 else 0 self.assert_true( final_state.cumulative_exp >= base_cum_99, f"Cumulative EXP after 100 deaths ({final_state.cumulative_exp}) is >= Level 99 benchmark cumulative floor ({base_cum_99})", ) # ========================================================================= # CHALLENGE 5: POLYMORPHIC PARAMETER CALLS & DTO ALIASES # ========================================================================= def challenge_polymorphic_calls_and_aliases(self) -> None: self.log("\n=== CHALLENGE 5: Polymorphic Parameter Calls & DTO Aliases ===") # Signature 1: 4 positional args: (player_id, monster_level, zone_level, base_exp) res_sig1 = self.service.award_monster_exp("poly_p1", 80, 80, 1000) self.assert_true( res_sig1.exp_awarded == 1000 and res_sig1.level_gap == 0 and res_sig1.gap_multiplier == 1.0, f"Signature 1 (pos 4): (player_id, monster_level, zone_level, base_exp) -> awarded={res_sig1.exp_awarded}", ) # Signature 2: 3 positional args: (player_id, monster_level, base_exp) self.service.set_player_state("poly_p2", level=80, current_exp=0) res_sig2 = self.service.award_monster_exp("poly_p2", 80, 1500) self.assert_true( res_sig2.exp_awarded == 1500 and res_sig2.gap_multiplier == 1.0, f"Signature 2 (pos 3): (player_id, monster_level, base_exp) -> awarded={res_sig2.exp_awarded}", ) # Signature 3: 2 positional args: (player_id, monster_level) [uses benchmark default] self.service.set_player_state("poly_p3", level=80, current_exp=0) bench_80 = self.service.get_benchmark(80) expected_bench_exp = bench_80.monster_benchmark_exp if bench_80 else 25 res_sig3 = self.service.award_monster_exp("poly_p3", 80) self.assert_true( res_sig3.exp_awarded == expected_bench_exp, f"Signature 3 (pos 2): (player_id, monster_level) -> default bench exp={res_sig3.exp_awarded} (expected {expected_bench_exp})", ) # Signature 4: All keyword args self.service.set_player_state("poly_p4", level=80, current_exp=0) res_sig4 = self.service.award_monster_exp( player_id="poly_p4", monster_level=80, zone_level=80, base_exp=2000, ) self.assert_true( res_sig4.exp_awarded == 2000 and res_sig4.effective_exp == 2000, f"Signature 4 (all kw): awarded={res_sig4.exp_awarded}, effective={res_sig4.effective_exp}", ) # Signature 5: Mixed positional and keyword args self.service.set_player_state("poly_p5", level=80, current_exp=0) res_sig5 = self.service.award_monster_exp( "poly_p5", 80, base_exp=2500, zone_level=80, player_level=80 ) self.assert_true( res_sig5.exp_awarded == 2500, f"Signature 5 (mixed pos/kw): awarded={res_sig5.exp_awarded}", ) # Signature 6: Kwargs with extra fields (**kwargs) self.service.set_player_state("poly_p6", level=80, current_exp=0) res_sig6 = self.service.award_monster_exp( "poly_p6", 80, base_exp=3000, arbitrary_tag="bonus_event", extra_data=123 ) self.assert_true( res_sig6.exp_awarded == 3000, f"Signature 6 (**kwargs tolerant): awarded={res_sig6.exp_awarded}", ) # DTO property aliases verification self.assert_true( res_sig1.effective_exp == res_sig1.exp_awarded and res_sig1.level_up_occurred == res_sig1.leveled_up and res_sig1.new_exp == res_sig1.current_exp, "ExpAwardResult aliases (effective_exp, level_up_occurred, new_exp) match canonical fields", ) death_res = self.service.apply_death_penalty("poly_p1") self.assert_true( death_res.penalty_exp_lost == death_res.exp_lost and death_res.penalty_percentage == death_res.penalty_ratio and death_res.new_exp == death_res.current_exp_after, "DeathPenaltyResult aliases (penalty_exp_lost, penalty_percentage, new_exp) match canonical fields", ) # ========================================================================= # CHALLENGE 6: HIGH-LOAD STRESS & EVENT SUBSCRIPTION INTEGRITY # ========================================================================= def challenge_stress_and_listeners(self) -> None: self.log("\n=== CHALLENGE 6: High-Load Stress & Event Subscription Integrity ===") level_up_events: List[LevelUpEvent] = [] death_events: List[DeathPenaltyResult] = [] service = LevelProgressionService() service.add_level_up_listener(lambda ev: level_up_events.append(ev)) service.add_death_penalty_listener(lambda ev: death_events.append(ev)) # Rapidly level 50 players from 1 to 20 num_players = 50 for i in range(num_players): pid = f"stress_p_{i}" # Give enough EXP for level 20 exp_20 = sum(service.get_delta_exp(l) for l in range(1, 20)) res = service.award_monster_exp(pid, 1, 1, exp_20, player_level=1) self.assert_true( res.new_level == 20, f"Stress player {pid} reached Lv 20 (gained {res.levels_gained} levels)", ) self.assert_true( len(level_up_events) == num_players, f"Event listeners captured exactly {len(level_up_events)} level-up events for {num_players} players", ) # Trigger deaths on all 50 players for i in range(num_players): pid = f"stress_p_{i}" service.apply_death_penalty(pid) self.assert_true( len(death_events) == num_players, f"Death penalty listeners captured exactly {len(death_events)} death events", ) def run_all(self) -> bool: self.log("Starting Empirical Progression Challenge Suite...\n") self.challenge_level_gap_decay() self.challenge_level_transitions_and_cap() self.challenge_tiered_death_penalty_matrix() self.challenge_safe_floor_invariant() self.challenge_polymorphic_calls_and_aliases() self.challenge_stress_and_listeners() self.log("\n=======================================================") self.log(f"CHALLENGE SUMMARY: {self.passed_tests} PASSED, {self.failed_tests} FAILED") self.log("=======================================================") return self.failed_tests == 0 if __name__ == "__main__": challenger = EmpiricalProgressionChallenger() success = challenger.run_all() sys.exit(0 if success else 1)