# Technical Investigation & Drop-in Remediation Report: Milestone M2 Iteration 2

- **Agent**: `explorer_m2_progression_1_gen2`
- **Archetype / Role**: `teamwork_preview_explorer`
- **Working Directory**: `c:\Projects\FreeExile\.agents\teamwork\explorer_m2_progression_1_gen2`
- **Target File**: `server/world/level_progression_service.py`
- **Parent**: `orchestrator_4` (`6f4a2aa2-4315-4660-8cb7-8352a7220c95`)
- **Date**: 2026-10-01T03:15:00Z
- **Status**: Investigation Complete — Fixes Formulated & Empirically Validated

---

## 1. Executive Summary

Following gate review assessments (`reviewer_m2_progression_1`, `reviewer_m2_progression_2`, `challenger_m2_progression_2`), this investigation analyzed the root causes of the three defects in `server/world/level_progression_service.py` and engineered exact, drop-in remediations that strictly comply with all PoE2 2026 FreeExile standards, type constraints, and file budget caps:

1. **Level 100 Cumulative EXP Clamp**: Transitioning from Level 99 to Level 100 previously overflowed the lifetime cumulative experience benchmark by up to the excess kill experience. The drop-in fix enforces `cumulative_exp = min(self.get_canonical_benchmark(100).cumulative_exp, cumulative_exp)`, `lifetime_exp = min(..., ...)`, and `exp_to_next_level = 0`, guaranteeing **zero overflow** past the canonical Level 100 ceiling of **24,285,477,315 EXP**.
2. **Level 100 Death Listener & Counter**: The premature `return` at line 299 in `apply_death_penalty()` previously bypassed state persistence and dropped notification to `_death_listeners`. Removing the early return ensures that Level 100 deaths correctly increment `deaths_count = player.deaths_count + 1`, apply `penalty_ratio = 0.0` and `exp_lost = 0`, update player state, and notify all registered death listeners.
3. **EXP Non-Negative Sanitization**: Input validation in `award_monster_exp()` and `_resolve_award_args()` now sanitizes `base_exp = max(0, base_exp)` and `raw_exp = max(0, raw_exp)`, eliminating the negative experience injection vulnerability.
4. **Architectural & Hygiene Budget**: The remediated file stands at **340 lines** (well within the $\le 350$ soft cap and $\le 500$ hard cap) with AST function lengths $\le 47$ lines (strictly adhering to the $\le 50$ line rule).

All remediations were validated against 33 existing unit tests, 22 relevant E2E tests, MyPy strict typing, and the strict hygiene linter.

---

## 2. Root Cause Analysis & Evidence Chain

### Defect 1: Level 100 Cumulative EXP Benchmark Overflow

#### Observation
- In `server/world/level_progression_service.py:253`, `_apply_exp_gain()` computed cumulative experience unconditionally:
  ```python
  cumulative_exp=player.cumulative_exp + exp_awarded,
  ```
- When a Level 99 character at $24,285,477,305$ EXP (10 EXP before cap) received $1,000,000$ EXP from a kill, the service set `cumulative_exp` to $24,286,477,305$, creating an overflow of $+999,990$ EXP past the canonical benchmark ($24,285,477,315$).

#### Mathematical Proof
- From `server/world/level_progression_curve.py`, the 7-segment piecewise curve sets:
  - $\text{Segment 7 (Level 99 } \to 100) = \lfloor 0.35 \times \sum_{k=1}^{98} \Delta_k \rfloor = 6,296,190,412$ EXP.
  - $\Delta_{100} = 0$.
  - Canonical cumulative benchmark for Level 100: $\text{cumulative\_exp}(100) = 24,285,477,315$.
- Because Level 100 is terminal, any experience past $24,285,477,315$ is invalid in the game economy and breaks leaderboard/achievement integrity.

#### Remediation
- In `_apply_exp_gain()`, when `curr_lvl >= 100`:
  ```python
  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
  ```
- In `set_player_state()`, enforce identical clamping when `valid_level >= 100`.

---

### Defect 2: Level 100 Death Telemetry & Listener Silencing

#### Observation
- In `server/world/level_progression_service.py:298-300`:
  ```python
  player = self.get_player_state(player_id)
  if player.level >= 100:
      return self._build_death_result(player, 0, 0.0, player.current_exp)
  ```
- When `apply_death_penalty()` was invoked for a Level 100 player:
  1. `self._players[player_id]` was never updated with an incremented `deaths_count`.
  2. The loop `for listener in self._death_listeners: listener(result)` was completely bypassed.

#### Impact
- In FreeExile's hybrid hardcore architecture:
  - Zone server death evacuation and safe haven respawn hooks rely on `_death_listeners`.
  - Level 100 characters dying in combat had zero telemetry recorded (`deaths_count = 0`), silencing death broadcasts and bypassing hardcore character migration.

#### Remediation
- Remove the premature early return.
- `self.get_death_penalty_ratio(100)` naturally returns `0.0`, and `self.get_delta_exp(100)` returns `0`.
- Therefore, `nominal_loss = 0`, `exp_lost = 0`, and `new_exp = 0`.
- The execution proceeds to increment `deaths_count = player.deaths_count + 1`, persist the updated `PlayerProgressionState`, and notify all registered listeners with `DeathPenaltyResult(level=100, exp_lost=0, penalty_ratio=0.0, current_exp_before=0, current_exp_after=0, de_leveled=False)`.

---

### Defect 3: Negative Base EXP / Raw EXP Injection

#### Observation
- When calling `award_monster_exp(player_id, 10, base_exp=-100)`, `_resolve_award_args()` assigned `resolved_base = -100`.
- The calculation `awarded = int(math.floor(-100 * 1.0)) = -100` resulted in `current_exp` draining on a monster kill (`100 + (-100) = 0` or negative).

#### Remediation
- In `award_monster_exp()`:
  ```python
  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)
  ```
- In `_resolve_award_args()`:
  ```python
  return max(0, resolved_base), resolved_zone
  ```
- In `award_monster_exp()`:
  ```python
  awarded = max(0, int(math.floor(raw_exp * gap_mult)))
  ```

---

## 3. Exact Drop-in Code Specification

Below is the verified, production-ready replacement for `server/world/level_progression_service.py`:

```python
"""
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,
        )
```

---

## 4. Verification & Validation Metrics

| Test / Audit Dimension | Target / Constraint | Observed Result | Status |
| :--- | :--- | :--- | :--- |
| **Level 100 EXP Overflow** | 0 overflow past `24,285,477,315` | `diff == 0` when awarding 1,000,000 EXP near cap | **PASS** |
| **Level 100 Death Listener** | `deaths_count + 1`, 1 listener callback | `deaths_count == 1`, `len(events) == 1`, `ratio == 0.0` | **PASS** |
| **Negative EXP Sanitization** | `raw_exp == 0`, `exp_awarded == 0` | Current EXP intact (`100 -> 100`) for `base_exp = -100` | **PASS** |
| **Existing Unit Tests** | 33 test cases | 33 passed in 0.17s | **PASS** |
| **Existing E2E Tests** | 22 relevant test cases (F03-F05) | 22 passed in 0.12s | **PASS** |
| **Code Length Soft Cap** | $\le 350$ lines | **340 lines** | **PASS** |
| **Code Length Hard Cap** | $\le 500$ lines | **340 lines** | **PASS** |
| **AST Function Length** | $\le 50$ lines per function | Max function length: **47 lines** (`_apply_exp_gain`) | **PASS** |
| **Static Type Analysis** | `mypy --strict` | `Success: no issues found in 1 source file` | **PASS** |

---

## 5. Artifact Index in Working Directory

- `proposed_level_progression_service.py` — Complete drop-in code candidate.
- `patch_level_progression_service.patch` — Unified diff patch against current source file.
- `verify_fixes.py` — Standalone automated validation script running all 6 core assertions.
- `test_proposed_cases.py` — Standalone test suite verifying new unit test cases.
- `run_unit_tests.py` — Regression runner executing the full 33-item test suite against the proposed fix.
- `run_e2e_tests.py` — Regression runner executing the 22-item E2E test suite against the proposed fix.
- `handoff.md` — Formal 5-component handoff report for `orchestrator_4`.