cal/strat-gameplay-webapp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

CLAUDE: Implement Week 7 Task 1 - Strategic Decision Integration

Browse Source 
Enhanced game engine with async decision workflow and AI opponent integration:

GameState Model Enhancements:
- Added pending_defensive_decision and pending_offensive_decision fields
- Added decision_phase tracking (idle, awaiting_defensive, awaiting_offensive, resolving, completed)
- Added decision_deadline field for timeout handling
- Added is_batting_team_ai() and is_fielding_team_ai() helper methods
- Added validator for decision_phase

StateManager Enhancements:
- Added _pending_decisions dict for asyncio.Future-based decision queue
- Added set_pending_decision() to create decision futures
- Added await_decision() to wait for decision submission
- Added submit_decision() to resolve pending futures
- Added cancel_pending_decision() for cleanup

GameEngine Enhancements:
- Added await_defensive_decision() with AI/human branching and timeout
- Added await_offensive_decision() with AI/human branching and timeout
- Enhanced submit_defensive_decision() to resolve pending futures
- Enhanced submit_offensive_decision() to resolve pending futures
- Added DECISION_TIMEOUT constant (30 seconds)

AI Opponent (Stub):
- Created ai_opponent.py with stub implementations
- generate_defensive_decision() returns default "normal" positioning
- generate_offensive_decision() returns default "normal" approach
- TODO markers for Week 9 full AI logic implementation

Integration:
- Backward compatible with existing terminal client workflow
- New async methods ready for WebSocket integration (Week 7 Task 4)
- AI teams get instant decisions, human teams wait with timeout
- Default decisions applied on timeout (no game blocking)

Testing:
- Config tests: 58/58 passing ✅
- Terminal client: Working perfectly ✅
- Existing workflows: Fully compatible ✅

Week 7 Task 1: Complete
Next: Task 2 - Decision Validators

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Cal Corum 2025-10-29 21:38:11 -05:00
parent d7caa75310
commit 95d8703f56
4 changed files with 497 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

159
backend/app/core/ai_opponent.py Normal file
View File

@ -0,0 +1,159 @@
"""
AI Opponent - Automated decision-making for AI-controlled teams.
Provides strategic decision generation for AI opponents in single-player
and AI vs AI game modes.
This is a stub implementation for Week 7. Full AI logic will be developed
in Week 9 as part of Phase 3 AI opponent integration.
Author: Claude
Date: 2025-10-29
"""
import logging
from typing import Optional
from app.models.game_models import GameState, DefensiveDecision, OffensiveDecision
logger = logging.getLogger(f'{__name__}.AIOpponent')
class AIOpponent:
"""
AI opponent decision-making engine.
Generates defensive and offensive decisions for AI-controlled teams.
Current implementation uses simple heuristics. Full AI logic will be
implemented in Week 9.
"""
def __init__(self, difficulty: str = "balanced"):
"""
Initialize AI opponent.
Args:
difficulty: AI difficulty level
- "balanced": Standard decision-making
- "yolo": Aggressive playstyle (more risks)
- "safe": Conservative playstyle (fewer risks)
"""
self.difficulty = difficulty
logger.info(f"AIOpponent initialized with difficulty: {difficulty}")
async def generate_defensive_decision(
self,
state: GameState
) -> DefensiveDecision:
"""
Generate defensive decision for AI-controlled fielding team.
Week 7 stub: Returns default "normal" positioning.
Week 9: Implement full decision logic based on game situation.
Args:
state: Current game state
Returns:
DefensiveDecision with AI-generated strategy
TODO (Week 9):
- Analyze batter tendencies (pull/opposite field)
- Consider runner speed for hold decisions
- Evaluate double play opportunities
- Adjust positioning based on inning/score
"""
logger.debug(f"Generating defensive decision for game {state.game_id}")
# Week 7 stub: Simple default decision
decision = DefensiveDecision(
alignment="normal",
infield_depth="normal",
outfield_depth="normal",
hold_runners=[]
)
# TODO Week 9: Add actual AI logic
# if state.outs < 2 and state.is_runner_on_first():
# decision.infield_depth = "double_play"
# if state.is_runner_on_third() and state.outs < 2:
# decision.infield_depth = "in"
logger.info(f"AI defensive decision: {decision.alignment}, IF: {decision.infield_depth}")
return decision
async def generate_offensive_decision(
self,
state: GameState
) -> OffensiveDecision:
"""
Generate offensive decision for AI-controlled batting team.
Week 7 stub: Returns default "normal" approach.
Week 9: Implement full decision logic based on game situation.
Args:
state: Current game state
Returns:
OffensiveDecision with AI-generated strategy
TODO (Week 9):
- Evaluate stealing opportunities (runner speed, pitcher, catcher)
- Consider bunting in appropriate situations
- Adjust batting approach based on score/inning
- Implement hit-and-run logic
"""
logger.debug(f"Generating offensive decision for game {state.game_id}")
# Week 7 stub: Simple default decision
decision = OffensiveDecision(
approach="normal",
steal_attempts=[],
hit_and_run=False,
bunt_attempt=False
)
# TODO Week 9: Add actual AI logic
# if state.is_runner_on_first() and not state.is_runner_on_second():
# # Consider stealing second
# if self._should_attempt_steal(state):
# decision.steal_attempts = [2]
logger.info(f"AI offensive decision: {decision.approach}")
return decision
def _should_attempt_steal(self, state: GameState) -> bool:
"""
Determine if AI should attempt a steal (Week 9).
TODO: Implement steal decision logic
- Evaluate runner speed
- Evaluate pitcher hold rating
- Evaluate catcher arm
- Consider game situation (score, inning, outs)
Returns:
True if steal should be attempted
"""
# Week 7 stub: Never steal
return False
def _should_attempt_bunt(self, state: GameState) -> bool:
"""
Determine if AI should attempt a bunt (Week 9).
TODO: Implement bunt decision logic
- Consider sacrifice bunt situations
- Evaluate batter bunting ability
- Assess defensive positioning
Returns:
True if bunt should be attempted
"""
# Week 7 stub: Never bunt
return False
# Singleton instance for global access
ai_opponent = AIOpponent()

179
backend/app/core/game_engine.py
View File

@ -4,18 +4,23 @@ 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.
Phase 3: Enhanced with async decision workflow and AI opponent integration.
Author: Claude
Date: 2025-10-24
"""
import asyncio
import logging
from uuid import UUID
from typing import Optional, List
import pendulum
from app.core.state_manager import state_manager
from app.core.play_resolver import play_resolver, PlayResult
from app.config import PlayOutcome
from app.core.validators import game_validator, ValidationError
from app.core.dice import dice_system
from app.core.ai_opponent import ai_opponent
from app.database.operations import DatabaseOperations
from app.models.game_models import (
GameState, DefensiveDecision, OffensiveDecision
@ -27,6 +32,9 @@ logger = logging.getLogger(f'{__name__}.GameEngine')
class GameEngine:
"""Main game orchestration engine"""
# Phase 3: Decision timeout in seconds
DECISION_TIMEOUT = 30
def __init__(self):
self.db_ops = DatabaseOperations()
# Track rolls per inning for batch saving
@ -114,7 +122,11 @@ class GameEngine:
game_id: UUID,
decision: DefensiveDecision
) -> GameState:
"""Submit defensive team decision"""
"""
Submit defensive team decision.
Phase 3: Now integrates with decision queue to resolve pending futures.
"""
state = state_manager.get_state(game_id)
if not state:
raise ValueError(f"Game {game_id} not found")
@ -122,9 +134,19 @@ class GameEngine:
game_validator.validate_game_active(state)
game_validator.validate_defensive_decision(decision, state)
# Store decision
# Store decision in state (for backward compatibility)
state.decisions_this_play['defensive'] = decision.model_dump()
state.pending_decision = "offensive"
state.pending_defensive_decision = decision
# Phase 3: Resolve pending future if exists
fielding_team_id = state.get_fielding_team_id()
try:
state_manager.submit_decision(game_id, fielding_team_id, decision)
logger.info(f"Resolved pending defensive decision future for game {game_id}")
except ValueError:
# No pending future - that's okay (direct submission without await)
logger.debug(f"No pending defensive decision for game {game_id}")
state_manager.update_state(game_id, state)
logger.info(f"Defensive decision submitted for game {game_id}")
@ -136,7 +158,11 @@ class GameEngine:
game_id: UUID,
decision: OffensiveDecision
) -> GameState:
"""Submit offensive team decision"""
"""
Submit offensive team decision.
Phase 3: Now integrates with decision queue to resolve pending futures.
"""
state = state_manager.get_state(game_id)
if not state:
raise ValueError(f"Game {game_id} not found")
@ -144,15 +170,160 @@ class GameEngine:
game_validator.validate_game_active(state)
game_validator.validate_offensive_decision(decision, state)
# Store decision
# Store decision in state (for backward compatibility)
state.decisions_this_play['offensive'] = decision.model_dump()
state.pending_decision = "resolution"
state.pending_offensive_decision = decision
# Phase 3: Resolve pending future if exists
batting_team_id = state.get_batting_team_id()
try:
state_manager.submit_decision(game_id, batting_team_id, decision)
logger.info(f"Resolved pending offensive decision future for game {game_id}")
except ValueError:
# No pending future - that's okay (direct submission without await)
logger.debug(f"No pending offensive decision for game {game_id}")
state_manager.update_state(game_id, state)
logger.info(f"Offensive decision submitted for game {game_id}")
return state
# ============================================================================
# PHASE 3: ENHANCED DECISION WORKFLOW
# ============================================================================
async def await_defensive_decision(
self,
state: GameState,
timeout: int = None
) -> DefensiveDecision:
"""
Wait for defensive team to submit decision.
For AI teams: Generate decision immediately
For human teams: Wait for WebSocket submission (with timeout)
Args:
state: Current game state
timeout: Seconds to wait before using default decision (default: class constant)
Returns:
DefensiveDecision (validated)
Raises:
asyncio.TimeoutError: If timeout exceeded (async games only)
"""
if timeout is None:
timeout = self.DECISION_TIMEOUT
fielding_team_id = state.get_fielding_team_id()
# Check if fielding team is AI
if state.is_fielding_team_ai():
logger.info(f"Generating AI defensive decision for game {state.game_id}")
return await ai_opponent.generate_defensive_decision(state)
# Human team: wait for decision via WebSocket
logger.info(f"Awaiting human defensive decision for game {state.game_id}, team {fielding_team_id}")
# Set pending decision in state manager
state_manager.set_pending_decision(
game_id=state.game_id,
team_id=fielding_team_id,
decision_type="defensive"
)
# Update state with decision phase
state.decision_phase = "awaiting_defensive"
state.decision_deadline = pendulum.now('UTC').add(seconds=timeout).to_iso8601_string()
state_manager.update_state(state.game_id, state)
# TODO Week 7 Task 4: Emit WebSocket event to notify frontend
# await self.connection_manager.emit_decision_required(
# game_id=state.game_id,
# team_id=fielding_team_id,
# decision_type="defensive",
# timeout=timeout,
# game_situation=state.to_situation_summary()
# )
try:
# Wait for decision with timeout
decision = await asyncio.wait_for(
state_manager.await_decision(state.game_id, fielding_team_id, "defensive"),
timeout=timeout
)
logger.info(f"Received defensive decision for game {state.game_id}")
return decision
except asyncio.TimeoutError:
# Use default decision on timeout
logger.warning(f"Defensive decision timeout for game {state.game_id}, using default")
return DefensiveDecision() # All defaults
async def await_offensive_decision(
self,
state: GameState,
timeout: int = None
) -> OffensiveDecision:
"""
Wait for offensive team to submit decision.
Similar to await_defensive_decision but for batting team.
Args:
state: Current game state
timeout: Seconds to wait before using default decision
Returns:
OffensiveDecision (validated)
Raises:
asyncio.TimeoutError: If timeout exceeded (async games only)
"""
if timeout is None:
timeout = self.DECISION_TIMEOUT
batting_team_id = state.get_batting_team_id()
# Check if batting team is AI
if state.is_batting_team_ai():
logger.info(f"Generating AI offensive decision for game {state.game_id}")
return await ai_opponent.generate_offensive_decision(state)
# Human team: wait for decision via WebSocket
logger.info(f"Awaiting human offensive decision for game {state.game_id}, team {batting_team_id}")
# Set pending decision in state manager
state_manager.set_pending_decision(
game_id=state.game_id,
team_id=batting_team_id,
decision_type="offensive"
)
# Update state with decision phase
state.decision_phase = "awaiting_offensive"
state.decision_deadline = pendulum.now('UTC').add(seconds=timeout).to_iso8601_string()
state_manager.update_state(state.game_id, state)
# TODO Week 7 Task 4: Emit WebSocket event to notify frontend
# await self.connection_manager.emit_decision_required(...)
try:
# Wait for decision with timeout
decision = await asyncio.wait_for(
state_manager.await_decision(state.game_id, batting_team_id, "offensive"),
timeout=timeout
)
logger.info(f"Received offensive decision for game {state.game_id}")
return decision
except asyncio.TimeoutError:
# Use default decision on timeout
logger.warning(f"Offensive decision timeout for game {state.game_id}, using default")
return OffensiveDecision() # All defaults
async def resolve_play(self, game_id: UUID) -> PlayResult:
"""
Resolve the current play with dice roll

142
backend/app/core/state_manager.py
View File

@ -10,12 +10,13 @@ Author: Claude
Date: 2025-10-22
"""
import asyncio
import logging
from typing import Dict, Optional
from typing import Dict, Optional, Union
from uuid import UUID
import pendulum
from app.models.game_models import GameState, TeamLineupState
from app.models.game_models import GameState, TeamLineupState, DefensiveDecision, OffensiveDecision
from app.database.operations import DatabaseOperations
logger = logging.getLogger(f'{__name__}.StateManager')
@ -39,6 +40,11 @@ class StateManager:
self._states: Dict[UUID, GameState] = {}
self._lineups: Dict[UUID, Dict[int, TeamLineupState]] = {} # game_id -> {team_id: lineup}
self._last_access: Dict[UUID, pendulum.DateTime] = {}
# Phase 3: Decision queue for async decision awaiting
# Key: (game_id, team_id, decision_type)
self._pending_decisions: Dict[tuple[UUID, int, str], asyncio.Future] = {}
self.db_ops = DatabaseOperations()
logger.info("StateManager initialized")
@ -392,6 +398,138 @@ class StateManager:
"""
return list(self._states.keys())
# ============================================================================
# PHASE 3: DECISION QUEUE MANAGEMENT
# ============================================================================
def set_pending_decision(
self,
game_id: UUID,
team_id: int,
decision_type: str
) -> None:
"""
Mark that a decision is required and create a future for it.
Args:
game_id: Game identifier
team_id: Team that needs to make the decision
decision_type: Type of decision ('defensive' or 'offensive')
"""
key = (game_id, team_id, decision_type)
# Create a new future for this decision
self._pending_decisions[key] = asyncio.Future()
logger.debug(f"Set pending {decision_type} decision for game {game_id}, team {team_id}")
async def await_decision(
self,
game_id: UUID,
team_id: int,
decision_type: str
) -> Union[DefensiveDecision, OffensiveDecision]:
"""
Wait for a decision to be submitted.
This coroutine will block until submit_decision() is called
with matching parameters.
Args:
game_id: Game identifier
team_id: Team making the decision
decision_type: Type of decision expected
Returns:
The submitted decision (DefensiveDecision or OffensiveDecision)
Raises:
ValueError: If no pending decision exists for these parameters
asyncio.TimeoutError: If decision not received within timeout (handled by caller)
"""
key = (game_id, team_id, decision_type)
if key not in self._pending_decisions:
raise ValueError(
f"No pending {decision_type} decision for game {game_id}, team {team_id}"
)
# Await the future (will be resolved by submit_decision)
decision = await self._pending_decisions[key]
logger.debug(f"Received {decision_type} decision for game {game_id}, team {team_id}")
return decision
def submit_decision(
self,
game_id: UUID,
team_id: int,
decision: Union[DefensiveDecision, OffensiveDecision]
) -> None:
"""
Submit a decision (called by WebSocket handler or AI opponent).
This resolves the pending future created by set_pending_decision().
Args:
game_id: Game identifier
team_id: Team making the decision
decision: The decision being submitted
Raises:
ValueError: If no pending decision exists
"""
# Determine decision type from the decision object
from app.models.game_models import DefensiveDecision
decision_type = "defensive" if isinstance(decision, DefensiveDecision) else "offensive"
key = (game_id, team_id, decision_type)
if key not in self._pending_decisions:
raise ValueError(
f"No pending {decision_type} decision for game {game_id}, team {team_id}"
)
future = self._pending_decisions[key]
# Check if already resolved (should not happen)
if future.done():
logger.warning(f"Decision already submitted for {key}")
return
# Resolve the future with the decision
future.set_result(decision)
# Clean up the future
del self._pending_decisions[key]
logger.info(f"Submitted {decision_type} decision for game {game_id}, team {team_id}")
def cancel_pending_decision(
self,
game_id: UUID,
team_id: int,
decision_type: str
) -> None:
"""
Cancel a pending decision (e.g., on timeout or game abort).
Args:
game_id: Game identifier
team_id: Team that was expected to decide
decision_type: Type of decision
"""
key = (game_id, team_id, decision_type)
if key in self._pending_decisions:
future = self._pending_decisions[key]
if not future.done():
future.cancel()
del self._pending_decisions[key]
logger.debug(f"Cancelled pending {decision_type} decision for game {game_id}, team {team_id}")
# Singleton instance for global access
state_manager = StateManager()

23
backend/app/models/game_models.py
View File

@ -264,6 +264,12 @@ class GameState(BaseModel):
pending_decision: Optional[str] = None # 'defensive', 'offensive', 'result_selection'
decisions_this_play: Dict[str, Any] = Field(default_factory=dict)
# Phase 3: Enhanced decision workflow
pending_defensive_decision: Optional[DefensiveDecision] = None
pending_offensive_decision: Optional[OffensiveDecision] = None
decision_phase: str = "idle" # idle, awaiting_defensive, awaiting_offensive, resolving, completed
decision_deadline: Optional[str] = None # ISO8601 timestamp for timeout
# Play tracking
play_count: int = Field(default=0, ge=0)
last_play_result: Optional[str] = None
@ -304,6 +310,15 @@ class GameState(BaseModel):
raise ValueError(f"pending_decision must be one of {valid}")
return v
@field_validator('decision_phase')
@classmethod
def validate_decision_phase(cls, v: str) -> str:
"""Ensure decision_phase is valid"""
valid = ['idle', 'awaiting_defensive', 'awaiting_offensive', 'resolving', 'completed']
if v not in valid:
raise ValueError(f"decision_phase must be one of {valid}")
return v
# Helper methods
def get_batting_team_id(self) -> int:
@ -314,6 +329,14 @@ class GameState(BaseModel):
"""Get the ID of the team currently fielding"""
return self.home_team_id if self.half == "top" else self.away_team_id
def is_batting_team_ai(self) -> bool:
"""Check if the currently batting team is AI-controlled"""
return self.away_team_is_ai if self.half == "top" else self.home_team_is_ai
def is_fielding_team_ai(self) -> bool:
"""Check if the currently fielding team is AI-controlled"""
return self.home_team_is_ai if self.half == "top" else self.away_team_is_ai
def is_runner_on_first(self) -> bool:
"""Check if there's a runner on first base"""
return self.on_first is not None