13e924a87c
Replaced awkward "lookback" pattern with clean "prepare → execute → save" orchestration that captures state snapshots BEFORE each play. Key improvements: - Added per-team batter indices (away_team_batter_idx, home_team_batter_idx) - Added play snapshot fields (current_batter/pitcher/catcher_lineup_id) - Added on_base_code bit field for efficient base situation queries - Created _prepare_next_play() method for snapshot preparation - Refactored start_game() with hard lineup validation requirement - Refactored resolve_play() with explicit 6-step orchestration - Updated _save_play_to_db() to use snapshots (no DB lookbacks) - Enhanced state recovery to rebuild from last play (single query) - Added defensive lineup position validator Benefits: - No special cases for first play - Single source of truth in GameState - Saves 18+ database queries per game - Fast state recovery without replay - Complete runner tracking (before/after positions) - Explicit orchestration (easy to debug) Testing: - Added 3 new test functions (lineup validation, snapshot tracking, batting order) - All 5 test suites passing (100%) - Type checking cleaned up with targeted suppressions for SQLAlchemy Documentation: - Added comprehensive "Type Checking & Common False Positives" section to CLAUDE.md - Created type-checking-guide.md and type-checking-summary.md - Added mypy.ini configuration for SQLAlchemy/Pydantic 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
522 lines
19 KiB
Python
522 lines
19 KiB
Python
"""
|
|
Game Engine - Main game orchestration engine.
|
|
|
|
Coordinates game flow, validates actions, resolves plays, and persists state.
|
|
Integrates DiceSystem for roll tracking with context and batch saving.
|
|
|
|
Author: Claude
|
|
Date: 2025-10-24
|
|
"""
|
|
import logging
|
|
from uuid import UUID
|
|
from typing import Optional, List
|
|
|
|
from app.core.state_manager import state_manager
|
|
from app.core.play_resolver import play_resolver, PlayResult
|
|
from app.core.validators import game_validator, ValidationError
|
|
from app.core.dice import dice_system
|
|
from app.database.operations import DatabaseOperations
|
|
from app.models.game_models import (
|
|
GameState, RunnerState, DefensiveDecision, OffensiveDecision
|
|
)
|
|
|
|
logger = logging.getLogger(f'{__name__}.GameEngine')
|
|
|
|
|
|
class GameEngine:
|
|
"""Main game orchestration engine"""
|
|
|
|
def __init__(self):
|
|
self.db_ops = DatabaseOperations()
|
|
# Track rolls per inning for batch saving
|
|
self._rolls_this_inning: dict[UUID, List] = {}
|
|
|
|
async def start_game(self, game_id: UUID) -> GameState:
|
|
"""
|
|
Start a game
|
|
|
|
Transitions from 'pending' to 'active'.
|
|
Validates that both teams have complete lineups (minimum 9 players each).
|
|
Prepares the first play snapshot.
|
|
|
|
Raises:
|
|
ValidationError: If game already started or lineups incomplete
|
|
"""
|
|
state = state_manager.get_state(game_id)
|
|
if not state:
|
|
raise ValueError(f"Game {game_id} not found in state manager")
|
|
|
|
if state.status != "pending":
|
|
raise ValidationError(f"Game already started (status: {state.status})")
|
|
|
|
# HARD REQUIREMENT: Validate both lineups are complete
|
|
# At game start, we validate BOTH teams (exception to the "defensive only" rule)
|
|
home_lineup = await self.db_ops.get_active_lineup(state.game_id, state.home_team_id)
|
|
away_lineup = await self.db_ops.get_active_lineup(state.game_id, state.away_team_id)
|
|
|
|
# Check minimum 9 players per team
|
|
if not home_lineup or len(home_lineup) < 9:
|
|
raise ValidationError(
|
|
f"Home team lineup incomplete: {len(home_lineup) if home_lineup else 0} players "
|
|
f"(minimum 9 required)"
|
|
)
|
|
|
|
if not away_lineup or len(away_lineup) < 9:
|
|
raise ValidationError(
|
|
f"Away team lineup incomplete: {len(away_lineup) if away_lineup else 0} players "
|
|
f"(minimum 9 required)"
|
|
)
|
|
|
|
# Validate defensive positions - at game start, check BOTH teams
|
|
try:
|
|
game_validator.validate_defensive_lineup_positions(home_lineup)
|
|
except ValidationError as e:
|
|
raise ValidationError(f"Home team: {e}")
|
|
|
|
try:
|
|
game_validator.validate_defensive_lineup_positions(away_lineup)
|
|
except ValidationError as e:
|
|
raise ValidationError(f"Away team: {e}")
|
|
|
|
# Mark as active
|
|
state.status = "active"
|
|
state.inning = 1
|
|
state.half = "top"
|
|
state.outs = 0
|
|
|
|
# Initialize roll tracking for this game
|
|
self._rolls_this_inning[game_id] = []
|
|
|
|
# Prepare first play snapshot
|
|
await self._prepare_next_play(state)
|
|
|
|
# Update state
|
|
state_manager.update_state(game_id, state)
|
|
|
|
# Persist to DB
|
|
await self.db_ops.update_game_state(
|
|
game_id=game_id,
|
|
inning=1,
|
|
half="top",
|
|
home_score=0,
|
|
away_score=0,
|
|
status="active"
|
|
)
|
|
|
|
logger.info(
|
|
f"Started game {game_id} - First batter: lineup_id={state.current_batter_lineup_id}"
|
|
)
|
|
return state
|
|
|
|
async def submit_defensive_decision(
|
|
self,
|
|
game_id: UUID,
|
|
decision: DefensiveDecision
|
|
) -> GameState:
|
|
"""Submit defensive team decision"""
|
|
state = state_manager.get_state(game_id)
|
|
if not state:
|
|
raise ValueError(f"Game {game_id} not found")
|
|
|
|
game_validator.validate_game_active(state)
|
|
game_validator.validate_defensive_decision(decision, state)
|
|
|
|
# Store decision
|
|
state.decisions_this_play['defensive'] = decision.model_dump()
|
|
state.pending_decision = "offensive"
|
|
|
|
state_manager.update_state(game_id, state)
|
|
logger.info(f"Defensive decision submitted for game {game_id}")
|
|
|
|
return state
|
|
|
|
async def submit_offensive_decision(
|
|
self,
|
|
game_id: UUID,
|
|
decision: OffensiveDecision
|
|
) -> GameState:
|
|
"""Submit offensive team decision"""
|
|
state = state_manager.get_state(game_id)
|
|
if not state:
|
|
raise ValueError(f"Game {game_id} not found")
|
|
|
|
game_validator.validate_game_active(state)
|
|
game_validator.validate_offensive_decision(decision, state)
|
|
|
|
# Store decision
|
|
state.decisions_this_play['offensive'] = decision.model_dump()
|
|
state.pending_decision = "resolution"
|
|
|
|
state_manager.update_state(game_id, state)
|
|
logger.info(f"Offensive decision submitted for game {game_id}")
|
|
|
|
return state
|
|
|
|
async def resolve_play(self, game_id: UUID) -> PlayResult:
|
|
"""
|
|
Resolve the current play with dice roll
|
|
|
|
Explicit orchestration sequence:
|
|
1. Resolve play with dice rolls
|
|
2. Save play to DB (uses snapshot from GameState)
|
|
3. Apply result to state (outs, score, runners)
|
|
4. Update game state in DB
|
|
5. Check for inning change (outs >= 3)
|
|
6. Prepare next play (always last step)
|
|
"""
|
|
state = state_manager.get_state(game_id)
|
|
if not state:
|
|
raise ValueError(f"Game {game_id} not found")
|
|
|
|
game_validator.validate_game_active(state)
|
|
|
|
# Get decisions
|
|
defensive_decision = DefensiveDecision(**state.decisions_this_play.get('defensive', {}))
|
|
offensive_decision = OffensiveDecision(**state.decisions_this_play.get('offensive', {}))
|
|
|
|
# STEP 1: Resolve play (this internally calls dice_system.roll_ab)
|
|
result = play_resolver.resolve_play(state, defensive_decision, offensive_decision)
|
|
|
|
# Track roll for batch saving at end of inning
|
|
if game_id not in self._rolls_this_inning:
|
|
self._rolls_this_inning[game_id] = []
|
|
self._rolls_this_inning[game_id].append(result.ab_roll)
|
|
|
|
# STEP 2: Save play to DB (uses snapshot from GameState)
|
|
await self._save_play_to_db(state, result)
|
|
|
|
# STEP 3: Apply result to state (outs, score, runners)
|
|
self._apply_play_result(state, result)
|
|
|
|
# STEP 4: Update game state in DB
|
|
await self.db_ops.update_game_state(
|
|
game_id=state.game_id,
|
|
inning=state.inning,
|
|
half=state.half,
|
|
home_score=state.home_score,
|
|
away_score=state.away_score,
|
|
status=state.status
|
|
)
|
|
|
|
# STEP 5: Check for inning change
|
|
if state.outs >= 3:
|
|
await self._advance_inning(state, game_id)
|
|
# Update DB again after inning change
|
|
await self.db_ops.update_game_state(
|
|
game_id=state.game_id,
|
|
inning=state.inning,
|
|
half=state.half,
|
|
home_score=state.home_score,
|
|
away_score=state.away_score,
|
|
status=state.status
|
|
)
|
|
# Batch save rolls at half-inning boundary
|
|
await self._batch_save_inning_rolls(game_id)
|
|
|
|
# STEP 6: Prepare next play (always last step)
|
|
if state.status == "active": # Only prepare if game is still active
|
|
await self._prepare_next_play(state)
|
|
|
|
# Clear decisions for next play
|
|
state.decisions_this_play = {}
|
|
state.pending_decision = "defensive"
|
|
|
|
# Update in-memory state
|
|
state_manager.update_state(game_id, state)
|
|
|
|
logger.info(f"Resolved play {state.play_count} for game {game_id}: {result.description}")
|
|
return result
|
|
|
|
def _apply_play_result(self, state: GameState, result: PlayResult) -> None:
|
|
"""
|
|
Apply play result to in-memory game state.
|
|
|
|
Only updates state - NO database writes (handled by orchestration layer).
|
|
"""
|
|
# Update outs
|
|
state.outs += result.outs_recorded
|
|
|
|
# Update runners
|
|
new_runners = []
|
|
|
|
# Advance existing runners
|
|
for runner in state.runners:
|
|
advanced = False
|
|
for from_base, to_base in result.runners_advanced:
|
|
if runner.on_base == from_base:
|
|
if to_base < 4: # Not scored
|
|
runner.on_base = to_base
|
|
new_runners.append(runner)
|
|
advanced = True
|
|
break
|
|
|
|
# Runner not in advancement list - stays put
|
|
if not advanced:
|
|
new_runners.append(runner)
|
|
|
|
# Add batter if reached base
|
|
if result.batter_result and result.batter_result < 4:
|
|
# Use current batter from snapshot
|
|
new_runners.append(RunnerState(
|
|
lineup_id=state.current_batter_lineup_id or 0,
|
|
card_id=0, # Will be populated from lineup in future
|
|
on_base=result.batter_result
|
|
))
|
|
|
|
state.runners = new_runners
|
|
|
|
# Update score
|
|
if state.half == "top":
|
|
state.away_score += result.runs_scored
|
|
else:
|
|
state.home_score += result.runs_scored
|
|
|
|
# Increment play count
|
|
state.play_count += 1
|
|
state.last_play_result = result.description
|
|
|
|
logger.debug(
|
|
f"Applied play result: outs={state.outs}, "
|
|
f"score={state.away_score}-{state.home_score}, "
|
|
f"runners={len(state.runners)}"
|
|
)
|
|
|
|
async def _advance_inning(self, state: GameState, game_id: UUID) -> None:
|
|
"""
|
|
Advance to next half inning.
|
|
|
|
Only handles inning transition - NO database writes, NO prepare_next_play.
|
|
Those are handled by the orchestration layer.
|
|
|
|
Validates defensive team lineup positions at start of each half inning.
|
|
"""
|
|
if state.half == "top":
|
|
state.half = "bottom"
|
|
else:
|
|
state.half = "top"
|
|
state.inning += 1
|
|
|
|
# Clear bases and reset outs
|
|
state.outs = 0
|
|
state.runners = []
|
|
|
|
# Validate defensive team lineup positions
|
|
# Top of inning: home team is defending
|
|
# Bottom of inning: away team is defending
|
|
defensive_team = state.home_team_id if state.half == "top" else state.away_team_id
|
|
defensive_lineup = await self.db_ops.get_active_lineup(state.game_id, defensive_team)
|
|
|
|
if not defensive_lineup:
|
|
raise ValidationError(f"No lineup found for defensive team {defensive_team}")
|
|
|
|
game_validator.validate_defensive_lineup_positions(defensive_lineup)
|
|
|
|
logger.info(f"Advanced to inning {state.inning} {state.half}")
|
|
|
|
# Check if game is over
|
|
if game_validator.is_game_over(state):
|
|
state.status = "completed"
|
|
logger.info(f"Game {state.game_id} completed - Final: Away {state.away_score}, Home {state.home_score}")
|
|
|
|
async def _prepare_next_play(self, state: GameState) -> None:
|
|
"""
|
|
Prepare snapshot for the next play.
|
|
|
|
This method:
|
|
1. Determines current batter based on batting order index
|
|
2. Advances the appropriate team's batter index (with wraparound)
|
|
3. Fetches active lineups from database
|
|
4. Sets snapshot fields: current_batter/pitcher/catcher_lineup_id
|
|
5. Calculates on_base_code from current runners
|
|
|
|
This snapshot is used when saving the Play record to DB.
|
|
"""
|
|
# Determine which team is batting
|
|
if state.half == "top":
|
|
# Away team batting
|
|
current_idx = state.away_team_batter_idx
|
|
state.away_team_batter_idx = (current_idx + 1) % 9
|
|
batting_team = state.away_team_id
|
|
fielding_team = state.home_team_id
|
|
else:
|
|
# Home team batting
|
|
current_idx = state.home_team_batter_idx
|
|
state.home_team_batter_idx = (current_idx + 1) % 9
|
|
batting_team = state.home_team_id
|
|
fielding_team = state.away_team_id
|
|
|
|
# Fetch active lineups from database
|
|
batting_lineup = await self.db_ops.get_active_lineup(state.game_id, batting_team)
|
|
fielding_lineup = await self.db_ops.get_active_lineup(state.game_id, fielding_team)
|
|
|
|
# Set current player snapshot
|
|
# Batter: use the batting order index to find the player
|
|
if batting_lineup and current_idx < len(batting_lineup):
|
|
# Get batting order sorted list
|
|
batting_order = sorted(
|
|
[p for p in batting_lineup if p.batting_order is not None],
|
|
key=lambda x: x.batting_order
|
|
)
|
|
if current_idx < len(batting_order):
|
|
# SQLAlchemy model .id is int at runtime, but typed as Column[int]
|
|
state.current_batter_lineup_id = batting_order[current_idx].id # type: ignore[assignment]
|
|
else:
|
|
state.current_batter_lineup_id = None
|
|
logger.warning(f"Batter index {current_idx} out of range for batting order")
|
|
else:
|
|
state.current_batter_lineup_id = None
|
|
logger.warning(f"No batting lineup found for team {batting_team}")
|
|
|
|
# Pitcher and catcher: find by position
|
|
# SQLAlchemy model .id is int at runtime, but typed as Column[int]
|
|
pitcher = next((p for p in fielding_lineup if p.position == "P"), None) if fielding_lineup else None # type: ignore
|
|
state.current_pitcher_lineup_id = pitcher.id if pitcher else None # type: ignore[assignment]
|
|
|
|
catcher = next((p for p in fielding_lineup if p.position == "C"), None) if fielding_lineup else None # type: ignore
|
|
state.current_catcher_lineup_id = catcher.id if catcher else None # type: ignore[assignment]
|
|
|
|
# Calculate on_base_code from current runners (bit field)
|
|
state.current_on_base_code = 0
|
|
for runner in state.runners:
|
|
if runner.on_base == 1:
|
|
state.current_on_base_code |= 1 # Bit 0: first base
|
|
elif runner.on_base == 2:
|
|
state.current_on_base_code |= 2 # Bit 1: second base
|
|
elif runner.on_base == 3:
|
|
state.current_on_base_code |= 4 # Bit 2: third base
|
|
|
|
logger.debug(
|
|
f"Prepared next play: batter={state.current_batter_lineup_id}, "
|
|
f"pitcher={state.current_pitcher_lineup_id}, "
|
|
f"catcher={state.current_catcher_lineup_id}, "
|
|
f"on_base_code={state.current_on_base_code}"
|
|
)
|
|
|
|
async def _batch_save_inning_rolls(self, game_id: UUID) -> None:
|
|
"""
|
|
Batch save all rolls from the inning
|
|
|
|
This is called at end of each half-inning to persist
|
|
all dice rolls with their context to the database.
|
|
"""
|
|
if game_id not in self._rolls_this_inning:
|
|
logger.debug(f"No rolls to save for game {game_id}")
|
|
return
|
|
|
|
rolls = self._rolls_this_inning[game_id]
|
|
if not rolls:
|
|
logger.debug(f"Empty roll list for game {game_id}")
|
|
return
|
|
|
|
try:
|
|
await self.db_ops.save_rolls_batch(rolls)
|
|
logger.info(f"Batch saved {len(rolls)} rolls for game {game_id}")
|
|
|
|
# Clear rolls for this inning
|
|
self._rolls_this_inning[game_id] = []
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to batch save rolls for game {game_id}: {e}")
|
|
# Don't fail the game - rolls are still in dice_system history
|
|
# We can recover them later if needed
|
|
|
|
async def _save_play_to_db(self, state: GameState, result: PlayResult) -> None:
|
|
"""
|
|
Save play to database using snapshot from GameState.
|
|
|
|
Uses the pre-calculated snapshot fields (no database lookbacks).
|
|
"""
|
|
# Use snapshot from GameState (set by _prepare_next_play)
|
|
batter_id = state.current_batter_lineup_id
|
|
pitcher_id = state.current_pitcher_lineup_id
|
|
catcher_id = state.current_catcher_lineup_id
|
|
on_base_code = state.current_on_base_code
|
|
|
|
# Runners on base BEFORE play (from state.runners)
|
|
on_first_id = next((r.lineup_id for r in state.runners if r.on_base == 1), None)
|
|
on_second_id = next((r.lineup_id for r in state.runners if r.on_base == 2), None)
|
|
on_third_id = next((r.lineup_id for r in state.runners if r.on_base == 3), None)
|
|
|
|
# Runners AFTER play (from result.runners_advanced)
|
|
# Build dict of from_base -> to_base for quick lookup
|
|
finals = {from_base: to_base for from_base, to_base in result.runners_advanced}
|
|
on_first_final = finals.get(1) # None if out/scored, 1-4 if advanced
|
|
on_second_final = finals.get(2) # None if out/scored, 1-4 if advanced
|
|
on_third_final = finals.get(3) # None if out/scored, 1-4 if advanced
|
|
|
|
# Batter result (None=out, 1-4=base reached)
|
|
batter_final = result.batter_result
|
|
|
|
play_data = {
|
|
"game_id": state.game_id,
|
|
"play_number": state.play_count,
|
|
"inning": state.inning,
|
|
"half": state.half,
|
|
"outs_before": state.outs - result.outs_recorded,
|
|
"outs_recorded": result.outs_recorded,
|
|
# Player IDs from snapshot
|
|
"batter_id": batter_id,
|
|
"pitcher_id": pitcher_id,
|
|
"catcher_id": catcher_id,
|
|
# Base situation snapshot
|
|
"on_base_code": on_base_code,
|
|
"on_first_id": on_first_id,
|
|
"on_second_id": on_second_id,
|
|
"on_third_id": on_third_id,
|
|
# Final positions
|
|
"on_first_final": on_first_final,
|
|
"on_second_final": on_second_final,
|
|
"on_third_final": on_third_final,
|
|
"batter_final": batter_final,
|
|
# Play outcome
|
|
"dice_roll": str(result.ab_roll),
|
|
"hit_type": result.outcome.value,
|
|
"result_description": result.description,
|
|
"runs_scored": result.runs_scored,
|
|
"away_score": state.away_score,
|
|
"home_score": state.home_score,
|
|
"complete": True,
|
|
# Strategic decisions
|
|
"defensive_choices": state.decisions_this_play.get('defensive', {}),
|
|
"offensive_choices": state.decisions_this_play.get('offensive', {})
|
|
}
|
|
|
|
await self.db_ops.save_play(play_data)
|
|
logger.debug(f"Saved play {state.play_count}: batter={batter_id}, on_base={on_base_code}")
|
|
|
|
async def get_game_state(self, game_id: UUID) -> Optional[GameState]:
|
|
"""Get current game state"""
|
|
return state_manager.get_state(game_id)
|
|
|
|
async def end_game(self, game_id: UUID) -> GameState:
|
|
"""
|
|
Manually end a game
|
|
|
|
For forfeit, abandonment, etc.
|
|
"""
|
|
state = state_manager.get_state(game_id)
|
|
if not state:
|
|
raise ValueError(f"Game {game_id} not found")
|
|
|
|
# Batch save any remaining rolls
|
|
await self._batch_save_inning_rolls(game_id)
|
|
|
|
state.status = "completed"
|
|
state_manager.update_state(game_id, state)
|
|
|
|
await self.db_ops.update_game_state(
|
|
game_id=game_id,
|
|
inning=state.inning,
|
|
half=state.half,
|
|
home_score=state.home_score,
|
|
away_score=state.away_score,
|
|
status="completed"
|
|
)
|
|
|
|
logger.info(f"Game {game_id} ended manually")
|
|
return state
|
|
|
|
|
|
# Singleton instance
|
|
game_engine = GameEngine()
|