Coverage for integrations / social / game_ai.py: 92.2%
103 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2HevolveSocial — Server-side AI move generators for solo game play.
4When a user plays solo against the computer, the AI opponent needs to
5produce a move. For engines where the **server holds the answer key**
6(trivia, word_scramble, word_search, sudoku), generating a move is
7cheap: read the ground truth, optionally inject difficulty-scaled
8mistakes, return a validated move shape.
10For **board games** (the 'boardgame' engine family — tictactoe,
11connect4, checkers, reversi, mancala, nim, dots_and_boxes, battleship)
12the rules live client-side in the boardgame.io Game definitions
13(Nunba: landing-page/src/components/Social/Games/board-games/*;
14RN: components/Social/Games/board-games/*). Re-implementing those
15rules here would be a DRY violation — so this module **explicitly
16rejects** board-game AI requests and defers to the client's
17boardgame.io MCTSBot/RandomBot. See Nunba `utils/gameAI.js`.
19For **phaser** (arcade score-chasing games) there is no move to make —
20the player plays against the scene itself. No server-side AI.
22Difficulty model
23────────────────
24All AI classes accept `difficulty` ∈ {'easy', 'medium', 'hard'}. This
25scales a single `error_rate` parameter that determines how often the AI
26chooses a non-optimal move. Each subclass maps error_rate to its own
27domain (wrong answer, wrong guess, wrong sudoku cell, etc.).
29 difficulty error_rate interpretation
30 easy 0.45 ≈ half wrong (beatable by anyone)
31 medium 0.18 ≈ 1/5 wrong (decent challenge)
32 hard 0.04 ≈ rarely wrong (near-perfect)
34Usage
35─────
36 from integrations.social.game_ai import generate_ai_move
38 move = generate_ai_move(
39 game_state=session.game_state,
40 game_type=session.game_type,
41 ai_user_id='ai',
42 difficulty='medium',
43 )
44 # move is the same shape expected by the corresponding game
45 # type's apply_move() — the caller can feed it straight into
46 # GameService.submit_move().
48The caller (api_games.ai_move endpoint) is responsible for:
49 - Verifying the session is active
50 - Authorizing the request (host only, to prevent spam)
51 - Submitting the returned move via the normal submit_move path
53This module does NOT mutate session state. It is a pure move generator.
55Phase 1 limitation — "host plays both sides"
56─────────────────────────────────────────────
57Until a dedicated AI GameParticipant + GameService.submit_move_as_ai
58path lands, the /ai_move endpoint returns a move dict ready for
59`POST /games/<id>/move`, BUT when that second POST runs, the move is
60authenticated as the host (via g.user_id) and credited to the host's
61score — not to a distinct AI opponent. For solo trivia/word-scramble/
62word-search/sudoku this means "the host plays both sides": the AI's
63answer is submitted under the host's identity and the host's score
64rises regardless of whether the host personally answered correctly.
66Phase 1 use cases where this is acceptable:
67 - "Suggest-a-move" hints during a stuck human game
68 - Automated smoke tests / CI for game session flows
69 - Future cross-device agents previewing AI moves before display
71Phase 1 use cases where this is NOT acceptable (and callers should
72avoid /ai_move for now):
73 - Solo ghost-opponent with a live scoreboard
74 - Multiplayer fill-dropped-seat
76Phase 2 work: add `ai_opponent: True` to create_session config, create
77an AI GameParticipant row, add `GameService.submit_move_as_ai(ai_uid)`
78that bypasses the g.user_id-must-be-participant check. The /ai_move
79endpoint then moves from read-only generator to one-step submit.
80"""
82import logging
83import random
84from typing import Dict, List, Optional, Tuple
86logger = logging.getLogger('hevolve_social')
89# ─── Difficulty → error rate map ────────────────────────────────────
91DIFFICULTY_ERROR_RATE = {
92 'easy': 0.45,
93 'medium': 0.18,
94 'hard': 0.04,
95}
98def _error_rate(difficulty: str) -> float:
99 """Resolve difficulty string to an error rate in [0, 1]."""
100 return DIFFICULTY_ERROR_RATE.get(difficulty, DIFFICULTY_ERROR_RATE['medium'])
103# ─── Base class ─────────────────────────────────────────────────────
105class GameAI:
106 """Base class for server-side AI move generators.
108 Subclasses implement generate_move(game_state, ai_user_id, difficulty)
109 and return a move_data dict matching the schema their corresponding
110 BaseGameType.apply_move() expects.
111 """
113 def generate_move(self, game_state: Dict, ai_user_id: str,
114 difficulty: str = 'medium') -> Dict:
115 raise NotImplementedError
117 def _roll_error(self, difficulty: str) -> bool:
118 """Return True iff the AI should make a mistake this turn."""
119 return random.random() < _error_rate(difficulty)
122# ─── Multiple-choice trivia (shared by TriviaGame + OpenTDBTriviaGame) ──
124class MultipleChoiceTriviaAI(GameAI):
125 """AI for both TriviaGame and OpenTDBTriviaGame.
127 Both engines normalize their questions to the same internal shape
128 before storing in game_state:
129 {'q': str, 'a': str, 'options': [str, ...], ...}
130 (see game_types.TriviaGame.CATEGORIES and
131 game_types_extended.OpenTDBTriviaGame._fetch_questions lines 75-89,
132 which do `html.unescape` + shuffle and emit the same 'a'/'options'
133 keys as TriviaGame).
135 The server stores the correct answer at
136 game_state['questions'][current_question_idx]['a']. On error, the
137 AI picks a different option from the question's `options` list.
138 """
140 def generate_move(self, game_state, ai_user_id, difficulty='medium'):
141 idx = game_state.get('current_question_idx', 0)
142 questions = game_state.get('questions', [])
143 if idx >= len(questions):
144 raise ValueError("No active question to answer")
146 question = questions[idx]
147 correct = question.get('a', '')
148 options = question.get('options', []) or []
150 if self._roll_error(difficulty) and len(options) > 1:
151 wrong = [o for o in options if o != correct]
152 if wrong:
153 return {'answer': random.choice(wrong)}
155 return {'answer': correct}
158# Backwards-compatible aliases so callers can import the engine-specific
159# name if they prefer — both resolve to the same implementation.
160TriviaAI = MultipleChoiceTriviaAI
161OpenTDBTriviaAI = MultipleChoiceTriviaAI
164# ─── Word Scramble ──────────────────────────────────────────────────
166class WordScrambleAI(GameAI):
167 """AI for WordScrambleGame.
169 The server stores the unscrambled word at
170 game_state['rounds'][current_round_idx]['word']. On error, the AI
171 emits a deliberately-wrong guess (permutation of the scrambled
172 letters that is not the correct word).
173 """
175 def generate_move(self, game_state, ai_user_id, difficulty='medium'):
176 idx = game_state.get('current_round_idx', 0)
177 rounds = game_state.get('rounds', [])
178 if idx >= len(rounds):
179 raise ValueError("No active round")
181 round_data = rounds[idx]
182 if round_data.get('solved_by'):
183 raise ValueError("Round already solved")
185 correct = round_data.get('word', '')
186 scrambled = round_data.get('scrambled', correct)
188 move = {'word': correct, 'time_ms': self._response_time_ms(difficulty)}
190 if self._roll_error(difficulty) and len(scrambled) > 1:
191 letters = list(scrambled)
192 for _ in range(10):
193 random.shuffle(letters)
194 candidate = ''.join(letters)
195 if candidate.lower() != correct.lower():
196 move['word'] = candidate
197 break
199 return move
201 def _response_time_ms(self, difficulty: str) -> int:
202 """Simulate AI 'thinking time'. Harder AIs respond faster."""
203 return {
204 'easy': random.randint(8000, 15000),
205 'medium': random.randint(4000, 9000),
206 'hard': random.randint(1500, 5000),
207 }.get(difficulty, 5000)
210# ─── Word Search ────────────────────────────────────────────────────
212class WordSearchAI(GameAI):
213 """AI for WordSearchGame.
215 All hidden words are in game_state['words_to_find']. The AI picks
216 an as-yet-unfound word. On error (easy mode) it occasionally picks
217 an already-found word, which the engine treats as a no-op / score 0.
218 """
220 def generate_move(self, game_state, ai_user_id, difficulty='medium'):
221 words_to_find = game_state.get('words_to_find', []) or []
222 found_words = game_state.get('found_words', {}) or {}
224 unfound = [w for w in words_to_find if w.lower() not in found_words]
225 if not unfound:
226 raise ValueError("No unfound words left")
228 # On error, occasionally emit a word that has already been found
229 # (engine returns 0 score — AI effectively wastes a turn).
230 if self._roll_error(difficulty) and found_words:
231 return {'word': random.choice(list(found_words.keys()))}
233 return {'word': random.choice(unfound)}
236# ─── Sudoku ─────────────────────────────────────────────────────────
238class SudokuAI(GameAI):
239 """AI for SudokuGame.
241 The server stores the solved grid at game_state['solution']. The
242 AI picks an empty cell (puzzle[r][c] == 0) and fills it with the
243 correct value. On error it fills with a deliberately-wrong digit.
244 """
246 def generate_move(self, game_state, ai_user_id, difficulty='medium'):
247 puzzle = game_state.get('puzzle', [])
248 solution = game_state.get('solution', [])
249 if not puzzle or not solution:
250 raise ValueError("Sudoku state missing puzzle or solution")
252 empty_cells: List[Tuple[int, int]] = [
253 (r, c)
254 for r in range(len(puzzle))
255 for c in range(len(puzzle[r]))
256 if puzzle[r][c] == 0
257 ]
258 if not empty_cells:
259 raise ValueError("No empty cells remaining")
261 r, c = random.choice(empty_cells)
262 correct = solution[r][c]
264 value = correct
265 if self._roll_error(difficulty):
266 wrong_values = [v for v in range(1, 10) if v != correct]
267 if wrong_values:
268 value = random.choice(wrong_values)
270 return {'row': r, 'col': c, 'value': value}
273# ─── Registry ───────────────────────────────────────────────────────
275# Single shared instance for trivia-family engines — both TriviaGame
276# and OpenTDBTriviaGame normalize to the same question shape, so the
277# same handler works for both.
278_TRIVIA_AI = MultipleChoiceTriviaAI()
280_AI_REGISTRY: Dict[str, GameAI] = {
281 'trivia': _TRIVIA_AI,
282 'quick_match': _TRIVIA_AI, # quick_match defaults to trivia
283 'opentdb_trivia': _TRIVIA_AI, # same normalized shape as TriviaGame
284 'word_scramble': WordScrambleAI(),
285 'word_search': WordSearchAI(),
286 'sudoku': SudokuAI(),
287 # boardgame, phaser intentionally absent — see module docstring.
288 # word_chain, collab_puzzle, compute_challenge: no AI — caller
289 # gets "No server-side AI registered" (distinct from
290 # client-authoritative rejection).
291}
294# Explicit "client-authoritative" set so the API endpoint can return a
295# clear error pointing callers at the client-side path instead of a
296# generic 400. Matches the engines with client-side Game definitions.
297CLIENT_AUTHORITATIVE_ENGINES = frozenset({
298 'boardgame', # use boardgame.io MCTSBot on client
299 'phaser', # arcade — no AI move concept
300})
303def _resolve_catalog_engine(game_type: str) -> Optional[str]:
304 """Resolve a catalog ID to its underlying engine name, or None.
306 Narrow exception catching: only ImportError is tolerated (optional
307 dependency). Other exceptions (malformed catalog entries, type
308 errors) propagate so mis-routing is visible rather than hidden
309 behind a False fallback.
310 """
311 try:
312 from .game_catalog import get_engine_for_catalog_entry
313 except ImportError:
314 return None
315 return get_engine_for_catalog_entry(game_type)
318def get_game_ai(game_type: str) -> Optional[GameAI]:
319 """Resolve a game_type (engine name or catalog id) to an AI handler.
321 Returns None when no server-side AI exists. Callers should check
322 for None and either emit an error or fall back to the client-side
323 path.
324 """
325 # Direct engine-name match
326 if game_type in _AI_REGISTRY:
327 return _AI_REGISTRY[game_type]
329 # Catalog-id lookup — resolve to the underlying engine name
330 engine = _resolve_catalog_engine(game_type)
331 if engine and engine in _AI_REGISTRY:
332 return _AI_REGISTRY[engine]
334 return None
337def is_client_authoritative(game_type: str) -> bool:
338 """True if this engine's AI must run client-side (not here)."""
339 if game_type in CLIENT_AUTHORITATIVE_ENGINES:
340 return True
341 engine = _resolve_catalog_engine(game_type)
342 return bool(engine and engine in CLIENT_AUTHORITATIVE_ENGINES)
345def generate_ai_move(game_state: Dict, game_type: str,
346 ai_user_id: str = 'ai',
347 difficulty: str = 'medium') -> Dict:
348 """Generate an AI move for the given session state.
350 Args:
351 game_state: The session's current game_state dict (from
352 GameSession.game_state). Not mutated.
353 game_type: The session's game_type (engine name or catalog id).
354 ai_user_id: The user_id string the AI will play as. Passed
355 through to the game's move (handlers need it for
356 turn-ownership checks).
357 difficulty: 'easy' | 'medium' | 'hard'.
359 Returns:
360 A move_data dict ready to feed into GameService.submit_move()
361 or the matching game type's apply_move().
363 Raises:
364 ValueError: No server-side AI exists for this game type (the
365 caller should route to the client-side path), OR the game
366 state is in a position where no move is possible.
367 """
368 if is_client_authoritative(game_type):
369 raise ValueError(
370 f"Game type '{game_type}' is client-authoritative — "
371 "use the client-side AI (boardgame.io MCTSBot on Nunba / RN)."
372 )
374 ai = get_game_ai(game_type)
375 if ai is None:
376 raise ValueError(f"No server-side AI registered for game type '{game_type}'")
378 return ai.generate_move(game_state, ai_user_id, difficulty)