""" Empirical Challenge Harness: Combat Engine & Level Progression Integration. Adversarial stress-testing of fatal damage dispatch, progression hooks, death penalties, monster EXP distribution, multi-player state isolation, and latency SLAs. """ from __future__ import annotations import math import os import random import sys import time from typing import Any, Dict, List, Optional, Tuple # Ensure project root is on 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.combat_engine import CombatActor, CombatEngine, DamageEventResult from server.world.level_progression_service import LevelProgressionService from server.world.level_progression_types import ( DeathPenaltyResult, ExpAwardResult, PlayerProgressionState, ) from server.world.martial_matrix import FiveElements class EmpiricalChallengeRunner: """Orchestrates empirical challenge suites and collects quantitative verification data.""" def __init__(self) -> None: self.results: Dict[str, bool] = {} self.metrics: Dict[str, Any] = {} self.findings: List[str] = [] def log(self, section: str, message: str) -> None: print(f"[{section}] {message}") # ========================================================================= # Challenge Suite 1: Fatal Damage Dispatch # ========================================================================= def challenge_fatal_damage_dispatch(self) -> bool: self.log("CHALLENGE_1", "Testing Fatal Damage Dispatch mechanics...") engine = CombatEngine() # 1A: Non-fatal strike attacker = CombatActor(1, "Attacker", FiveElements.HOA, base_attack=50.0) defender = CombatActor(2, "Defender", FiveElements.HOA, current_hp=500.0, max_hp=500.0) engine.register_actor(attacker) engine.register_actor(defender) res_non_fatal = engine.calculate_damage(1, 2, 50.0, FiveElements.HOA, current_timestamp_ms=1000) assert res_non_fatal.is_fatal is False, "Non-fatal damage flagged as fatal!" assert defender.current_hp == 450.0, f"Expected 450.0 HP, got {defender.current_hp}" # 1B: Exact lethal strike (HP drops exactly to 0.0) defender.current_hp = 50.0 res_exact = engine.calculate_damage(1, 2, 50.0, FiveElements.HOA, current_timestamp_ms=1050) assert res_exact.is_fatal is True, "Exact lethal damage not flagged as fatal!" assert defender.current_hp == 0.0, f"Expected 0.0 HP, got {defender.current_hp}" # 1C: Overkill strike (Damage >> Current HP) defender.current_hp = 20.0 res_overkill = engine.calculate_damage(1, 2, 9999.0, FiveElements.HOA, current_timestamp_ms=1100) assert res_overkill.is_fatal is True, "Overkill damage not flagged as fatal!" assert defender.current_hp == 0.0, f"HP not clamped to 0.0 on overkill: {defender.current_hp}" # 1D: Evasion i-frame active during lethal attack defender.current_hp = 10.0 engine.trigger_phantom_evasion(2, timestamp_ms=2000) res_evaded = engine.calculate_damage(1, 2, 9999.0, FiveElements.HOA, current_timestamp_ms=2100) assert res_evaded.is_evaded is True, "Attack was not evaded during i-frame!" assert res_evaded.final_damage == 0.0, "Evaded attack dealt non-zero damage!" assert res_evaded.is_fatal is False, "Evaded attack triggered fatal flag!" assert defender.current_hp == 10.0, "Defender HP changed despite evasion!" # 1E: Missing actor handling res_missing = engine.calculate_damage(999, 2, 100.0, FiveElements.HOA, current_timestamp_ms=3000) assert res_missing.is_fatal is False, "Missing attacker caused fatal damage!" assert res_missing.final_damage == 0.0, "Missing attacker dealt damage!" # 1F: Adversarial edge case: repeated hit on already dead defender (corpse attack) defender.current_hp = 0.0 res_corpse = engine.calculate_damage(1, 2, 100.0, FiveElements.HOA, current_timestamp_ms=4000) # Note: defender was already at 0.0 HP. Does it re-trigger is_fatal? if res_corpse.is_fatal: self.findings.append( "ADVERSARIAL_OBSERVATION: Striking an already-dead actor (current_hp == 0.0) " "re-evaluates is_fatal=True because current_hp <= 0.0 condition is met again." ) self.log("CHALLENGE_1", "PASS: Fatal Damage Dispatch verified.") self.results["fatal_damage_dispatch"] = True return True # ========================================================================= # Challenge Suite 2: Callback Invocation & Payload Integrity # ========================================================================= def challenge_callback_invocation(self) -> bool: self.log("CHALLENGE_2", "Testing Callback Invocation and Payload Integrity...") engine = CombatEngine() callback_calls: List[Tuple[DamageEventResult, CombatActor, CombatActor]] = [] def fatal_hook(res: DamageEventResult, atk: CombatActor, dfn: CombatActor) -> None: callback_calls.append((res, atk, dfn)) engine.set_fatal_damage_hook(fatal_hook) attacker = CombatActor(10, "Assassin", FiveElements.KIM, base_attack=200.0, is_player=True, player_id="p_assassin") defender = CombatActor(20, "Target", FiveElements.MOC, current_hp=100.0, max_hp=100.0) engine.register_actor(attacker) engine.register_actor(defender) # 2A: Non-fatal hit should NOT trigger callback defender.current_hp = 1000.0 engine.calculate_damage(10, 20, 50.0, FiveElements.KIM, current_timestamp_ms=1000) assert len(callback_calls) == 0, f"Callback invoked on non-fatal hit! Calls: {len(callback_calls)}" # 2B: Fatal hit MUST trigger callback with accurate payload defender.current_hp = 50.0 res = engine.calculate_damage(10, 20, 200.0, FiveElements.KIM, current_timestamp_ms=1050) assert res.is_fatal is True assert len(callback_calls) == 1, f"Expected 1 callback call, got {len(callback_calls)}" cb_res, cb_atk, cb_dfn = callback_calls[0] assert cb_res.attacker_id == 10 assert cb_res.defender_id == 20 assert cb_res.is_fatal is True assert cb_atk.actor_id == 10 assert cb_atk.name == "Assassin" assert cb_atk.player_id == "p_assassin" assert cb_dfn.actor_id == 20 assert cb_dfn.name == "Target" assert cb_dfn.current_hp == 0.0 # 2C: Clearing callback (None) should safely disable hook engine.set_fatal_damage_hook(None) defender.current_hp = 10.0 engine.calculate_damage(10, 20, 200.0, FiveElements.KIM, current_timestamp_ms=1100) assert len(callback_calls) == 1, "Callback invoked after being cleared!" self.log("CHALLENGE_2", "PASS: Callback Invocation verified.") self.results["callback_invocation"] = True return True # ========================================================================= # Challenge Suite 3: Progression Service Hook (`attach_progression_service`) # ========================================================================= def challenge_progression_service_hook(self) -> bool: self.log("CHALLENGE_3", "Testing Progression Service Hook Integration...") service = LevelProgressionService() engine = CombatEngine() engine.attach_progression_service(service) # 3A: Monster kills Player -> Death Penalty across tiers tiers_to_test = [ (25, 0.0), # Grace tier (1-60): 0% penalty (70, 0.05), # Mid tier (61-80): 5% penalty (85, 0.10), # High tier (81-89): 10% penalty (95, 0.15), # Endgame tier (90-98): 15% penalty (99, 0.25), # Soft-wall tier (99): 25% penalty (100, 0.0), # Cap tier (100): 0% penalty ] for lvl, expected_ratio in tiers_to_test: pid = f"player_tier_{lvl}" d_exp = service.get_delta_exp(lvl) initial_exp = int(d_exp * 0.50) if lvl < 100 else 0 service.set_player_state(pid, level=lvl, current_exp=initial_exp) boss = CombatActor(100 + lvl, f"Boss_{lvl}", FiveElements.THO, base_attack=9999.0, is_player=False) player_actor = CombatActor( 200 + lvl, f"P_{lvl}", FiveElements.THUY, current_hp=100.0, max_hp=100.0, is_player=True, level=lvl, player_id=pid ) engine.register_actor(boss) engine.register_actor(player_actor) # Boss delivers fatal blow dmg_res = engine.calculate_damage(boss.actor_id, player_actor.actor_id, 9999.0, FiveElements.THO, current_timestamp_ms=1000) assert dmg_res.is_fatal is True assert player_actor.current_hp == 0.0 state = service.get_level_info(pid) if lvl < 100: assert state.deaths_count == 1, f"Expected deaths_count=1, got {state.deaths_count} for level {lvl}" expected_loss = int(math.floor(d_exp * expected_ratio)) expected_exp = max(0, initial_exp - expected_loss) assert state.current_exp == expected_exp, ( f"Level {lvl} EXP mismatch: expected {expected_exp}, got {state.current_exp} " f"(loss={expected_loss}, ratio={expected_ratio})" ) else: # Level 100 empirical finding check if state.deaths_count == 0: self.findings.append( "BUG_FINDING: LevelProgressionService.apply_death_penalty returns early at line 299 " "for player.level >= 100 without incrementing deaths_count or dispatching _death_listeners. " f"Expected deaths_count=1, got {state.deaths_count}." ) assert state.current_exp == 0 # 3B: Safe Floor Test: Current EXP is less than nominal death penalty pid_floor = "player_floor_test" lvl_floor = 95 d95 = service.get_delta_exp(lvl_floor) nominal_penalty = int(math.floor(d95 * 0.15)) tiny_exp = int(nominal_penalty * 0.3) # less than penalty service.set_player_state(pid_floor, level=lvl_floor, current_exp=tiny_exp) boss_floor = CombatActor(301, "FloorBoss", FiveElements.HOA, base_attack=5000.0, is_player=False) player_floor = CombatActor(302, "FloorPlayer", FiveElements.KIM, current_hp=50.0, max_hp=50.0, is_player=True, player_id=pid_floor) engine.register_actor(boss_floor) engine.register_actor(player_floor) engine.calculate_damage(301, 302, 5000.0, FiveElements.HOA, current_timestamp_ms=2000) state_floor = service.get_level_info(pid_floor) assert state_floor.current_exp == 0, f"Expected safe floor at 0, got {state_floor.current_exp}" assert state_floor.level == lvl_floor, "Player de-leveled when hitting 0 exp floor!" # 3C: Player kills Monster -> Monster EXP Awarded pid_slayer = "player_slayer" service.set_player_state(pid_slayer, level=70, current_exp=0) slayer_actor = CombatActor(401, "Slayer", FiveElements.KIM, base_attack=500.0, is_player=True, level=70, player_id=pid_slayer) mob_actor = CombatActor(402, "DungeonMob", FiveElements.MOC, current_hp=100.0, max_hp=100.0, is_player=False, level=70) engine.register_actor(slayer_actor) engine.register_actor(mob_actor) # Fatal strike against monster engine.calculate_damage(401, 402, 500.0, FiveElements.KIM, current_timestamp_ms=3000) state_slayer = service.get_level_info(pid_slayer) assert state_slayer.current_exp > 0, "No EXP awarded to player on monster kill!" assert mob_actor.current_hp == 0.0 # 3D: Player kills Monster with level-up cascade pid_levelup = "player_levelupper" d1 = service.get_delta_exp(1) service.set_player_state(pid_levelup, level=1, current_exp=d1 - 10) # 10 exp away from level up actor_lup = CombatActor(501, "LevelUpper", FiveElements.HOA, base_attack=500.0, is_player=True, level=1, player_id=pid_levelup) mob_high = CombatActor(502, "HighMob", FiveElements.KIM, current_hp=50.0, max_hp=50.0, is_player=False, level=1) engine.register_actor(actor_lup) engine.register_actor(mob_high) engine.calculate_damage(501, 502, 500.0, FiveElements.HOA, current_timestamp_ms=4000) state_lup = service.get_level_info(pid_levelup) assert state_lup.level >= 2, f"Player did not level up! Level is {state_lup.level}" assert state_lup.unspent_talent_points >= 1, "Talent points not awarded on level up!" # 3E: Missing player_id falls back to str(actor.actor_id) actor_nopid = CombatActor(601, "NoPidPlayer", FiveElements.THO, base_attack=500.0, is_player=True, level=1, player_id=None) mob_nopid = CombatActor(602, "MobNoPid", FiveElements.THUY, current_hp=50.0, max_hp=50.0, is_player=False, level=1) engine.register_actor(actor_nopid) engine.register_actor(mob_nopid) engine.calculate_damage(601, 602, 500.0, FiveElements.THO, current_timestamp_ms=5000) state_fallback = service.get_level_info("601") assert state_fallback.current_exp > 0, "Fallback player_id=str(actor_id) failed to award EXP!" # 3F: Monster kills Monster (NPC vs NPC) -> No progression calls mob_npc1 = CombatActor(701, "WolfA", FiveElements.MOC, base_attack=100.0, is_player=False) mob_npc2 = CombatActor(702, "WolfB", FiveElements.THO, current_hp=50.0, max_hp=50.0, is_player=False) engine.register_actor(mob_npc1) engine.register_actor(mob_npc2) res_npc = engine.calculate_damage(701, 702, 100.0, FiveElements.MOC, current_timestamp_ms=6000) assert res_npc.is_fatal is True self.log("CHALLENGE_3", "PASS: Progression Service Hook verified.") self.results["progression_hook"] = True return True # ========================================================================= # Challenge Suite 4: Latency Performance & SLA Benchmark # ========================================================================= def challenge_latency_performance_sla(self, iterations: int = 1000) -> bool: self.log("CHALLENGE_4", f"Benchmarking {iterations} fatal combat calculations with progression hooks...") service = LevelProgressionService() engine = CombatEngine() engine.attach_progression_service(service) # Pre-seed player pid = "perf_test_player" service.set_player_state(pid, level=80, current_exp=10000) # Setup actors player = CombatActor(1001, "PerfPlayer", FiveElements.KIM, base_attack=200.0, is_player=True, level=80, player_id=pid) engine.register_actor(player) latencies_us: List[float] = [] # Run 1,000 iterations: Player kills Monster (fatal + level gap + exp award + potential level check) for i in range(iterations): mob_id = 2000 + i mob = CombatActor(mob_id, f"Mob_{i}", FiveElements.MOC, current_hp=10.0, max_hp=10.0, is_player=False, level=80) engine.register_actor(mob) t_start = time.perf_counter_ns() res = engine.calculate_damage(1001, mob_id, 200.0, FiveElements.KIM, current_timestamp_ms=10000 + i) t_end = time.perf_counter_ns() assert res.is_fatal is True latencies_us.append((t_end - t_start) / 1000.0) # microseconds # Also benchmark 500 fatal damage events where Monster kills Player (death penalty) death_latencies_us: List[float] = [] for i in range(500): boss_id = 4000 + i boss = CombatActor(boss_id, f"Boss_{i}", FiveElements.HOA, base_attack=9999.0, is_player=False, level=85) engine.register_actor(boss) player.current_hp = 50.0 # reset HP for fatal hit t_start = time.perf_counter_ns() res = engine.calculate_damage(boss_id, 1001, 9999.0, FiveElements.HOA, current_timestamp_ms=20000 + i) t_end = time.perf_counter_ns() assert res.is_fatal is True death_latencies_us.append((t_end - t_start) / 1000.0) # Statistical Analysis latencies_ms = [us / 1000.0 for us in latencies_us] death_latencies_ms = [us / 1000.0 for us in death_latencies_us] mean_ms = sum(latencies_ms) / len(latencies_ms) p50_ms = sorted(latencies_ms)[int(len(latencies_ms) * 0.50)] p95_ms = sorted(latencies_ms)[int(len(latencies_ms) * 0.95)] p99_ms = sorted(latencies_ms)[int(len(latencies_ms) * 0.99)] max_ms = max(latencies_ms) death_mean_ms = sum(death_latencies_ms) / len(death_latencies_ms) death_p99_ms = sorted(death_latencies_ms)[int(len(death_latencies_ms) * 0.99)] self.metrics["combat_exp_award_latencies"] = { "iterations": iterations, "mean_ms": mean_ms, "p50_ms": p50_ms, "p95_ms": p95_ms, "p99_ms": p99_ms, "max_ms": max_ms, } self.metrics["combat_death_penalty_latencies"] = { "iterations": 500, "mean_ms": death_mean_ms, "p99_ms": death_p99_ms, } self.log( "CHALLENGE_4", f"EXP Award Latency: mean={mean_ms:.4f}ms, p50={p50_ms:.4f}ms, " f"p95={p95_ms:.4f}ms, p99={p99_ms:.4f}ms, max={max_ms:.4f}ms" ) self.log( "CHALLENGE_4", f"Death Penalty Latency: mean={death_mean_ms:.4f}ms, p99={death_p99_ms:.4f}ms" ) # Assert SLA budget compliance: SLA < 25ms, Target < 5ms assert mean_ms < 5.0, f"Mean latency {mean_ms:.4f}ms exceeds target 5.0ms!" assert p99_ms < 25.0, f"p99 latency {p99_ms:.4f}ms violates 25ms SLA budget!" assert death_mean_ms < 5.0, f"Death penalty mean latency {death_mean_ms:.4f}ms exceeds 5.0ms!" self.log("CHALLENGE_4", "PASS: Latency Performance & SLA verified.") self.results["latency_sla"] = True return True # ========================================================================= # Challenge Suite 5: Multi-Player Isolation & Zero Crosstalk # ========================================================================= def challenge_multi_player_isolation(self) -> bool: self.log("CHALLENGE_5", "Simulating 10 concurrent player IDs for state isolation...") service = LevelProgressionService() engine = CombatEngine() engine.attach_progression_service(service) num_players = 10 player_configs: Dict[str, Dict[str, Any]] = {} # Set up 10 distinct players with varied initial levels initial_levels = [10, 25, 45, 65, 75, 82, 88, 92, 95, 99] for idx, lvl in enumerate(initial_levels): pid = f"player_{idx + 1:02d}" d_exp = service.get_delta_exp(lvl) start_exp = int(d_exp * 0.40) service.set_player_state(pid, level=lvl, current_exp=start_exp) player_configs[pid] = { "actor_id": 1000 + idx, "initial_level": lvl, "initial_exp": start_exp, "kills": 0, "deaths": 0, "expected_exp": start_exp, "expected_level": lvl, } actor = CombatActor( 1000 + idx, f"Actor_{pid}", FiveElements.KIM, current_hp=500.0, max_hp=500.0, base_attack=300.0, is_player=True, level=lvl, player_id=pid ) engine.register_actor(actor) # Perform 500 randomized, interleaved events across the 10 players random.seed(42) event_count = 500 for ev in range(event_count): p_idx = random.randint(0, num_players - 1) pid = f"player_{p_idx + 1:02d}" cfg = player_configs[pid] p_actor_id = cfg["actor_id"] p_actor = engine.actors[p_actor_id] event_type = random.choice(["kill_mob", "die_to_boss", "non_fatal"]) if event_type == "kill_mob": mob_id = 50000 + ev mob_lvl = cfg["expected_level"] mob = CombatActor(mob_id, f"Mob_{ev}", FiveElements.MOC, current_hp=10.0, max_hp=10.0, is_player=False, level=mob_lvl) engine.register_actor(mob) # Record expected outcome mult = service.calculate_level_gap_multiplier(cfg["expected_level"], mob_lvl) bench = service.get_benchmark(mob_lvl) base = bench.monster_benchmark_exp if bench else 25 gain = int(math.floor(base * mult)) engine.calculate_damage(p_actor_id, mob_id, 300.0, FiveElements.KIM, current_timestamp_ms=30000 + ev) cfg["kills"] += 1 elif event_type == "die_to_boss": boss_id = 60000 + ev boss = CombatActor(boss_id, f"Boss_{ev}", FiveElements.HOA, base_attack=9999.0, is_player=False, level=cfg["expected_level"]) engine.register_actor(boss) p_actor.current_hp = 50.0 # make vulnerable to fatal blow engine.calculate_damage(boss_id, p_actor_id, 9999.0, FiveElements.HOA, current_timestamp_ms=30000 + ev) cfg["deaths"] += 1 p_actor.current_hp = 500.0 # respawn elif event_type == "non_fatal": boss_id = 70000 + ev boss = CombatActor(boss_id, f"ScratchBoss_{ev}", FiveElements.HOA, base_attack=10.0, is_player=False) engine.register_actor(boss) p_actor.current_hp = 500.0 engine.calculate_damage(boss_id, p_actor_id, 10.0, FiveElements.HOA, current_timestamp_ms=30000 + ev) # Validate complete isolation and state integrity for pid, cfg in player_configs.items(): state = service.get_level_info(pid) assert state.deaths_count == cfg["deaths"], ( f"Player {pid} death count mismatch! Expected {cfg['deaths']}, got {state.deaths_count}" ) # Ensure no player state is None or corrupt assert state.player_id == pid assert state.level >= cfg["initial_level"], f"Player {pid} de-leveled!" assert state.current_exp >= 0, f"Player {pid} has negative current_exp: {state.current_exp}" # Check immutability: PlayerProgressionState is frozen try: state.level = 999 # type: ignore[misc] raise AssertionError(f"PlayerProgressionState for {pid} is mutable!") except (AttributeError, TypeError): pass # expected: frozen dataclass cannot be assigned to self.log("CHALLENGE_5", f"PASS: Multi-player isolation verified across 10 players and {event_count} interleaved events.") self.results["multi_player_isolation"] = True return True # ========================================================================= # Challenge Suite 6: Adversarial Stress & Vulnerability Exposure # ========================================================================= def challenge_adversarial_vulnerabilities(self) -> bool: self.log("CHALLENGE_6", "Testing Adversarial Edge Cases and Failure Modes...") service = LevelProgressionService() engine = CombatEngine() engine.attach_progression_service(service) # 6A: Multi-hit on Dead Monster Corpse -> EXP duplication vulnerability pid_exploit = "player_corpse_exploit" service.set_player_state(pid_exploit, level=50, current_exp=0) hero = CombatActor(801, "Exploiter", FiveElements.KIM, base_attack=200.0, is_player=True, player_id=pid_exploit) goblin = CombatActor(802, "WeakGoblin", FiveElements.MOC, current_hp=20.0, max_hp=20.0, is_player=False, level=50) engine.register_actor(hero) engine.register_actor(goblin) # Strike 1: kills goblin (drops HP 20 -> 0) res1 = engine.calculate_damage(801, 802, 200.0, FiveElements.KIM, current_timestamp_ms=1000) exp_after_kill = service.get_level_info(pid_exploit).current_exp # Strike 2: hits dead goblin corpse (HP already 0) res2 = engine.calculate_damage(801, 802, 200.0, FiveElements.KIM, current_timestamp_ms=1010) exp_after_corpse_hit = service.get_level_info(pid_exploit).current_exp if exp_after_corpse_hit > exp_after_kill: self.findings.append( f"VULNERABILITY [HIGH/CRITICAL]: Corpse Multi-Hit EXP Duplication! " f"Striking dead goblin (HP=0) awarded additional {exp_after_corpse_hit - exp_after_kill} EXP " f"(Total: {exp_after_kill} -> {exp_after_corpse_hit}). " "Root cause: CombatEngine evaluates is_fatal=(defender.current_hp <= 0.0) without checking " "if defender was already dead (was_alive = current_hp > 0.0)." ) # 6B: Channeled Multi-Hit on Dead Player -> Death Penalty Multi-Drain pid_victim = "player_channeled_victim" d95 = service.get_delta_exp(95) service.set_player_state(pid_victim, level=95, current_exp=d95) boss = CombatActor(901, "VolleyBoss", FiveElements.HOA, base_attack=9999.0, is_player=False, level=95) victim = CombatActor(902, "VictimPlayer", FiveElements.KIM, current_hp=100.0, max_hp=100.0, is_player=True, player_id=pid_victim) engine.register_actor(boss) engine.register_actor(victim) # 3 rapid hits from channeled spell (e.g. fire barrage) for h in range(3): engine.calculate_damage(901, 902, 9999.0, FiveElements.HOA, current_timestamp_ms=2000 + h * 10) victim_state = service.get_level_info(pid_victim) if victim_state.deaths_count > 1: self.findings.append( f"VULNERABILITY [HIGH]: Multi-Hit Channeled Death Penalty Drain! " f"Player struck 3 times by same barrage suffered {victim_state.deaths_count} deaths, " f"losing {victim_state.deaths_count * 15}% EXP instead of 15% for a single death sequence." ) # 6C: Level 100 Death Listener & Telemetry Silencing pid_100 = "player_lv100_telemetry" service.set_player_state(pid_100, level=100, current_exp=0) boss100 = CombatActor(911, "Titan", FiveElements.THO, base_attack=9999.0, is_player=False) p100_actor = CombatActor(912, "GodhoodHero", FiveElements.THUY, current_hp=100.0, max_hp=100.0, is_player=True, player_id=pid_100) engine.register_actor(boss100) engine.register_actor(p100_actor) listener_called = False def death_listener(res: DeathPenaltyResult) -> None: nonlocal listener_called listener_called = True service.add_death_penalty_listener(death_listener) engine.calculate_damage(911, 912, 9999.0, FiveElements.THO, current_timestamp_ms=3000) st100 = service.get_level_info(pid_100) if st100.deaths_count == 0 or not listener_called: self.findings.append( "DEFECT [MEDIUM/HIGH]: Level 100 Player Death Silenced! " f"deaths_count={st100.deaths_count} (expected 1), listener_called={listener_called} (expected True). " "Root cause: LevelProgressionService.apply_death_penalty lines 298-299 returns early before " "updating player state or dispatching listeners." ) self.results["adversarial_vulnerabilities_checked"] = True return True # ========================================================================= # Run All Challenges # ========================================================================= def run_all(self) -> bool: print("=" * 70) print("EMPIRICAL CHALLENGER: Milestone M2 Combat Progression Integration") print("=" * 70) c1 = self.challenge_fatal_damage_dispatch() c2 = self.challenge_callback_invocation() c3 = self.challenge_progression_service_hook() c4 = self.challenge_latency_performance_sla(iterations=1000) c5 = self.challenge_multi_player_isolation() c6 = self.challenge_adversarial_vulnerabilities() all_passed = c1 and c2 and c3 and c4 and c5 and c6 has_vulnerabilities = len(self.findings) > 0 print("=" * 70) print("CHALLENGE SUITE RESULTS:") for name, passed in self.results.items(): print(f" - {name}: {'PASSED' if passed else 'FAILED'}") if self.findings: print("\n" + "!" * 70) print("EMPIRICAL VULNERABILITIES & DEFECTS DETECTED:") print("!" * 70) for f in self.findings: print(f" * {f}\n") print("=" * 70) verdict = "REQUEST_CHANGES" if has_vulnerabilities else "APPROVE" print(f"CHALLENGER VERDICT: {verdict}") print("=" * 70) return not has_vulnerabilities if __name__ == "__main__": runner = EmpiricalChallengeRunner() success = runner.run_all() sys.exit(0 if success else 1)