Spaces:
Running
Running
| """ | |
| State validators that run after every step() call to catch integrity bugs. | |
| """ | |
| from typing import TYPE_CHECKING, List | |
| import numpy as np | |
| if TYPE_CHECKING: | |
| from engine.game.game_state import GameState | |
| from engine.game.enums import Phase | |
| class StateValidator: | |
| """Validates game state integrity after each step.""" | |
| def validate_post_step(gs: "GameState") -> List[str]: | |
| """ | |
| Run comprehensive validation after every step() call. | |
| Returns list of violation messages (empty = valid). | |
| """ | |
| errors = [] | |
| # V1: No looked_cards without pending choices | |
| if gs.looked_cards and not gs.pending_choices: | |
| errors.append( | |
| f"STALE_LOOKED_CARDS: {len(gs.looked_cards)} cards in looked_cards " | |
| f"with no pending choice (phase={gs.phase.name})" | |
| ) | |
| # V2: PlayerState has no dynamic attributes outside __slots__ | |
| for p in gs.players: | |
| if hasattr(p, "__dict__") and p.__dict__: | |
| errors.append( | |
| f"DYNAMIC_ATTRS: Player {p.player_id} has attrs outside __slots__: {list(p.__dict__.keys())}" | |
| ) | |
| # V3: Phase-specific action validation | |
| legal = gs.get_legal_actions() | |
| phase_errors = StateValidator._validate_phase_actions(gs.phase, legal) | |
| errors.extend(phase_errors) | |
| # V4: Game state consistency | |
| consistency_errors = StateValidator._validate_consistency(gs) | |
| errors.extend(consistency_errors) | |
| return errors | |
| def _validate_phase_actions(phase: Phase, legal: np.ndarray) -> List[str]: | |
| """Verify only appropriate actions are legal for each phase.""" | |
| errors = [] | |
| legal_indices = np.where(legal)[0] | |
| if phase == Phase.LIVE_SET: | |
| # Only action 0 (pass) or 400-459 (set live) should be valid | |
| for i in legal_indices: | |
| if i != 0 and not (400 <= i <= 459): | |
| errors.append(f"ILLEGAL_PHASE_ACTION: Action {i} valid in LIVE_SET") | |
| elif phase == Phase.MULLIGAN_P1 or phase == Phase.MULLIGAN_P2: | |
| # Only action 0 (confirm) or 300-359 (toggle mulligan) valid | |
| for i in legal_indices: | |
| if i != 0 and not (300 <= i <= 359): | |
| errors.append(f"ILLEGAL_PHASE_ACTION: Action {i} valid in MULLIGAN") | |
| elif phase == Phase.ACTIVE or phase == Phase.ENERGY or phase == Phase.DRAW: | |
| # Auto-phases should only have action 0 | |
| for i in legal_indices: | |
| if i != 0: | |
| errors.append(f"ILLEGAL_PHASE_ACTION: Action {i} valid in {phase.name}") | |
| return errors | |
| def _validate_consistency(gs: "GameState") -> List[str]: | |
| """Validate general state consistency.""" | |
| errors = [] | |
| for p in gs.players: | |
| # Hand and hand_added_turn should be same length | |
| if len(p.hand) != len(p.hand_added_turn): | |
| errors.append( | |
| f"HAND_DESYNC: Player {p.player_id} hand={len(p.hand)} vs hand_added_turn={len(p.hand_added_turn)}" | |
| ) | |
| # live_zone and live_zone_revealed should be same length | |
| if len(p.live_zone) != len(p.live_zone_revealed): | |
| errors.append( | |
| f"LIVE_ZONE_DESYNC: Player {p.player_id} live_zone={len(p.live_zone)} " | |
| f"vs live_zone_revealed={len(p.live_zone_revealed)}" | |
| ) | |
| # Stage energy should have 3 slots | |
| if p.stage_energy_count.shape[0] != 3: | |
| errors.append(f"STAGE_ENERGY_SIZE: Player {p.player_id} has {p.stage_energy_count.shape[0]} slots") | |
| # Triggered abilities shouldn't persist if no pending choices | |
| if gs.triggered_abilities and not gs.pending_choices and not gs.pending_effects: | |
| errors.append(f"STALE_TRIGGERED: {len(gs.triggered_abilities)} triggered abilities with no pending work") | |
| return errors | |
| def assert_valid(gs: "GameState") -> None: | |
| """Raise AssertionError if state is invalid.""" | |
| errors = StateValidator.validate_post_step(gs) | |
| if errors: | |
| raise AssertionError("State validation failed:\n" + "\n".join(errors)) | |