"""Game validation and repair module.""" from typing import Tuple # Hard safety constraints derived from dataset patterns FORBIDDEN_BEHAVIORS = [ "building", "shop", "roof", "rooftop", "enter building", "private", "private courtyard", "fenced", "fence", "gate", "locked", "river edge", "canal edge", "water edge", "edge of water", "traffic", "road crossing", "railroad", "rail line", "fountain", "climbing", "climb", "purchase", "buy", "stranger", "staff", "permission needed" ] WATER_HAZARDS = [ "river", "canal", "pond", "lake", "fountain", "water", "drowning", "swimming" ] ROAD_HAZARDS = [ "traffic", "road", "street crossing", "highway", "busy road", "intersection", "rail", "railroad", "railway" ] BUILDING_RESTRICTIONS = [ "enter", "inside", "building", "shop", "store", "house", "apartment", "private", "restricted", "no entry" ] def check_forbidden_behaviors(game: dict) -> Tuple[bool, list[str]]: """Check that task descriptions don't describe forbidden behaviors. Uses action-context analysis: "building" in "photograph the building" is safe, but "enter the building" or "go inside the shop" is forbidden. Args: game: Game to check Returns: Tuple of (is_valid, list of violations) """ # Forbidden action patterns — require a verb + noun combination # The verb implies the player must DO something unsafe FORBIDDEN_ACTION_PATTERNS = [ # Entering buildings/shops ('enter', ['building', 'shop', 'store', 'house', 'courtyard']), ('go inside', ['building', 'shop', 'store', 'house']), ('go into', ['building', 'shop', 'store', 'house']), ('step inside', ['building', 'shop']), ('walk inside', ['building', 'shop']), # Climbing ('climb', ['wall', 'fence', 'roof', 'structure', 'gate', 'railing']), ('climb onto', ['roof', 'wall', 'structure']), # Interacting with strangers ('ask', ['stranger', 'staff', 'employee']), ('talk to', ['stranger', 'staff']), ('approach', ['stranger', 'staff']), # Purchasing ('buy', ['item', 'ticket', 'food', 'drink', 'something']), ('purchase', ['item', 'ticket']), # Water proximity ('enter', ['water', 'river', 'canal', 'pond', 'lake', 'fountain']), ('step into', ['water', 'river', 'canal']), ('go near', ['river edge', 'canal edge', 'water edge']), ] violations = [] tasks = game.get("tasks", []) for task in tasks: desc = (task.get('description', '') + ' ' + task.get('location_hint', '')).lower() # Skip if description already forbids the action itself if any(sk in desc for sk in ['forbidden', 'do not', "don't", 'avoid', 'stay away', 'no entering', 'no climbing', 'no hiding']): continue for verb, nouns in FORBIDDEN_ACTION_PATTERNS: for noun in nouns: pattern = f"{verb} {noun}" if pattern in desc: violations.append(f"Task {task.get('task_id')}: describes forbidden behavior '{pattern}'") return len(violations) == 0, violations def check_water_hazards(game: dict) -> Tuple[bool, list[str]]: """Check that tasks mentioning water proximity have explicit safety restrictions. Args: game: Game to check Returns: Tuple of (is_valid, list of violations) """ violations = [] tasks = game.get("tasks", []) WATER_ACTION_WORDS = [ 'cross the river', 'cross the canal', 'go near water', 'enter water', 'enter river', 'enter canal', 'enter lake', 'enter pond', 'step into water', 'swim', 'boat', 'kayak', 'paddle', ] for task in tasks: desc = f"{task.get('description', '')} {task.get('location_hint', '')}".lower() safety_note = task.get('safety_note', '').lower() # Only flag if the water proximity is an ACTION, not a location reference is_water_action = any(aw in desc for aw in WATER_ACTION_WORDS) if is_water_action: has_restriction = any(r in safety_note for r in ["no", "forbidden", "stay away", "do not", "don't", "exclude"]) if not has_restriction: violations.append(f"Task {task.get('task_id')}: mentions water activity but lacks explicit safety restriction") return len(violations) == 0, violations def check_road_hazards(game: dict) -> Tuple[bool, list[str]]: """Check that tasks involving road crossing have explicit safety guidance. Args: game: Game to check Returns: Tuple of (is_valid, list of violations) """ violations = [] tasks = game.get("tasks", []) ROAD_ACTION_WORDS = [ 'cross the road', 'cross the street', 'cross traffic', 'run across', 'dash across', 'jump across', ] for task in tasks: desc = f"{task.get('description', '')} {task.get('location_hint', '')}".lower() safety_note = task.get('safety_note', '').lower() # Only flag road-crossing actions, not passive mentions like "street corner" is_road_action = any(rw in desc for rw in ROAD_ACTION_WORDS) if is_road_action: has_restriction = any(r in safety_note for r in ["safe", "safely", "careful", "caution", "avoid", "crosswalk", "zebra"]) if not has_restriction: violations.append(f"Task {task.get('task_id')}: mentions road crossing but lacks safety guidance") return len(violations) == 0, violations def check_building_access(game: dict) -> Tuple[bool, list[str]]: """Check that tasks don't require entering private buildings. "Find the building" or "photograph the facade" are safe. "Enter the building" or "go inside" are not. Args: game: Game to check Returns: Tuple of (is_valid, list of violations) """ violations = [] tasks = game.get("tasks", []) ENTRY_PHRASES = [ 'enter the building', 'enter building', 'enter the shop', 'enter shop', 'enter the store', 'enter store', 'go inside the', 'step inside', 'walk inside', 'go into the building', 'go into the shop', ] for task in tasks: desc = f"{task.get('description', '')} {task.get('location_hint', '')}".lower() for phrase in ENTRY_PHRASES: if phrase in desc: violations.append(f"Task {task.get('task_id')}: '{phrase}' requires entering a building") return len(violations) == 0, violations def check_task_requirements(game: dict) -> Tuple[bool, list[str]]: """Check that all tasks have required fields with valid values. Args: game: Game to check Returns: Tuple of (is_valid, list of violations) """ violations = [] tasks = game.get("tasks", []) required_fields = { 'task_id': str, 'title': str, 'description': str, 'location_hint': str, 'points': int, 'proof_type': str, 'hint': str, 'safety_note': str } for task in tasks: # Check required fields exist for field, field_type in required_fields.items(): if field not in task: violations.append(f"Task {task.get('task_id')}: missing required field '{field}'") elif task[field] is None or (isinstance(task[field], str) and len(task[field].strip()) == 0): violations.append(f"Task {task.get('task_id')}: '{field}' is empty") # Validate proof_type enum if task.get('proof_type') not in ['photo', 'observation', 'text']: violations.append(f"Task {task.get('task_id')}: invalid proof_type '{task.get('proof_type')}'") # Validate points are positive if task.get('points', 0) <= 0: violations.append(f"Task {task.get('task_id')}: points must be positive, got {task.get('points')}") # Check hint is meaningful if len(task.get('hint', '').split()) < 2: violations.append(f"Task {task.get('task_id')}: hint too short or missing") # Check safety_note is meaningful if len(task.get('safety_note', '').split()) < 3: violations.append(f"Task {task.get('task_id')}: safety_note too brief") return len(violations) == 0, violations def check_task_count_realism(game: dict, config: dict) -> Tuple[bool, list[str]]: """Check that task count is realistic for the duration. Args: game: Game to check config: Original configuration Returns: Tuple of (is_valid, list of violations) """ violations = [] tasks = game.get("tasks", []) duration = config.get('duration_minutes', 45) # Estimate: 10-15 minutes per task (including travel, decision time) min_tasks = max(1, duration // 20) max_tasks = duration // 8 task_count = len(tasks) if task_count < min_tasks: violations.append(f"Too few tasks ({task_count}) for {duration} min duration (recommend {min_tasks}-{max_tasks})") elif task_count > max_tasks: violations.append(f"Too many tasks ({task_count}) for {duration} min duration (recommend {min_tasks}-{max_tasks})") return len(violations) == 0, violations def check_age_appropriate_supervision(game: dict, config: dict) -> Tuple[bool, list[str]]: """Check that kids and mixed-age games require adult supervision. Args: game: Game to check config: Original configuration Returns: Tuple of (is_valid, list of violations) """ violations = [] age_group = config.get('age_group', 'adults') adult_supervision = game.get('safety', {}).get('adult_supervision', False) if age_group in ['kids', 'mixed'] and not adult_supervision: violations.append(f"Age group '{age_group}' requires adult_supervision=true") # Check rules mention supervision for kids if age_group in ['kids', 'mixed']: rules_text = ' '.join(game.get('rules', [])).lower() if 'adult' not in rules_text and 'supervision' not in rules_text: violations.append(f"Age group '{age_group}' should mention adult supervision in rules") return len(violations) == 0, violations def check_win_condition(game: dict) -> Tuple[bool, list[str]]: """Check that game has a clear win condition. Args: game: Game to check Returns: Tuple of (is_valid, list of violations) """ violations = [] # Check tie_breaker exists (indicates clear win condition) tie_breaker = game.get('tie_breaker', '').strip() if not tie_breaker: violations.append("Game must have a clear tie_breaker/win condition") # Check score_rules exist score_rules = game.get('score_rules', []) if not score_rules or len(score_rules) == 0: violations.append("Game must have scoring rules") return len(violations) == 0, violations def check_safety_structure(game: dict) -> Tuple[bool, list[str]]: """Check that safety object is complete and sensible. Args: game: Game to check Returns: Tuple of (is_valid, list of violations) """ violations = [] safety = game.get('safety', {}) required_fields = ['allowed_zone', 'forbidden_behaviors', 'adult_supervision', 'stop_conditions'] for field in required_fields: if field not in safety: violations.append(f"Safety missing required field: '{field}'") # Check forbidden_behaviors is non-empty forbidden = safety.get('forbidden_behaviors', []) if not forbidden or len(forbidden) < 2: violations.append("Safety.forbidden_behaviors should have at least 2 items") # Check stop_conditions is non-empty stops = safety.get('stop_conditions', []) if not stops or len(stops) < 2: violations.append("Safety.stop_conditions should have at least 2 items") # Check allowed_zone is meaningful zone = safety.get('allowed_zone', '').strip() if not zone or len(zone) < 5: violations.append("Safety.allowed_zone should be clearly defined") return len(violations) == 0, violations def validate_game(game: dict, config: dict) -> Tuple[bool, list[str]]: """Validate generated game against hard rules derived from dataset patterns. Checks: - Basic structure (required fields, correct types) - Task requirements (proofs, hints, safety notes) - Safety constraints (no buildings, water/road hazards) - Age appropriateness (adult supervision) - Realism (task count vs duration) - Win conditions and scoring Args: game: Generated game JSON config: Original game configuration Returns: Tuple of (is_valid, list of failure messages) """ all_failures = [] # Run all validation checks checks = [ ("Structure", lambda: check_task_requirements(game)), ("Safety structure", lambda: check_safety_structure(game)), ("Forbidden behaviors", lambda: check_forbidden_behaviors(game)), ("Water hazards", lambda: check_water_hazards(game)), ("Road hazards", lambda: check_road_hazards(game)), ("Building access", lambda: check_building_access(game)), ("Task realism", lambda: check_task_count_realism(game, config)), ("Age appropriateness", lambda: check_age_appropriate_supervision(game, config)), ("Win condition", lambda: check_win_condition(game)), ] for check_name, check_fn in checks: try: is_valid, violations = check_fn() if not is_valid: all_failures.extend([f"[{check_name}] {v}" for v in violations]) except Exception as e: all_failures.append(f"[{check_name}] Error: {str(e)}") return len(all_failures) == 0, all_failures def repair_game(game: dict, failures: list[str], config: dict) -> dict: """Repair a game that failed validation with minimal modifications. The repair strategy applies targeted fixes only to failing fields, preserving all valid content. Uses progressive repair: 1. Structural fixes (missing objects, fields) 2. Content fixes (hints, safety notes, forbidden content) 3. Config-alignment fixes (duration, age group, difficulty) 4. Smart fallback generation if config mismatch. Args: game: Failed game JSON failures: List of validation failures config: Original game configuration Returns: Repaired game JSON (only failing fields modified) """ import uuid # Deep copy to avoid mutating original repaired = __import__('json').loads(__import__('json').dumps(game)) # Categorize failures for targeted repair failure_categories = { 'structure': [], 'safety': [], 'task_field': [], 'task_content': [], 'rules': [], 'forbidden': [], 'age': [], 'win_condition': [], 'realism': [], } for failure in failures: if '[Structure]' in failure or 'missing required field' in failure.lower() or 'empty' in failure.lower(): failure_categories['structure'].append(failure) elif '[Safety' in failure or 'forbidden_behaviors' in failure or 'stop_conditions' in failure or 'allowed_zone' in failure: failure_categories['safety'].append(failure) elif 'adult_supervision' in failure or 'age group' in failure.lower(): failure_categories['age'].append(failure) elif '[Forbidden' in failure or 'forbidden behavior' in failure.lower(): failure_categories['forbidden'].append(failure) elif '[Task realism' in failure or 'realism' in failure.lower() or 'task count' in failure.lower() or 'duration' in failure.lower(): failure_categories['realism'].append(failure) elif '[Task' in failure and ('missing' in failure.lower() or 'empty' in failure.lower()): failure_categories['task_field'].append(failure) elif '[Task' in failure: failure_categories['task_content'].append(failure) elif '[Rules' in failure or 'rules' in failure.lower(): failure_categories['rules'].append(failure) elif '[Water' in failure or '[Road' in failure or '[Building' in failure: failure_categories['forbidden'].append(failure) elif 'win' in failure.lower() or 'tie_breaker' in failure.lower() or 'scoring' in failure.lower(): failure_categories['win_condition'].append(failure) elif 'realism' in failure.lower() or 'task count' in failure.lower() or 'duration' in failure.lower(): failure_categories['realism'].append(failure) else: failure_categories['structure'].append(failure) # ---- 1. STRUCTURAL REPAIRS ---- # Ensure game has an ID if not repaired.get('game_id'): repaired['game_id'] = f"game-{uuid.uuid4().hex[:8]}" # Ensure title uses config if missing if not repaired.get('title'): area = config.get('area', 'City') game_type = config.get('game_type', 'game').replace('_', ' ').title() repaired['title'] = f"{area} {game_type}" # Ensure theme from config if not repaired.get('theme') or len(repaired.get('theme', '').strip()) < 3: if 'scavenger' in config.get('game_type', ''): repaired['theme'] = 'urban discovery' elif 'hide' in config.get('game_type', ''): repaired['theme'] = 'hide and seek' elif 'tag' in config.get('game_type', ''): repaired['theme'] = 'fast-paced chase' else: repaired['theme'] = 'exploration and fun' # Ensure setup uses config values if 'setup' not in repaired or not isinstance(repaired.get('setup'), dict): repaired['setup'] = {} if not repaired['setup'].get('city'): repaired['setup']['city'] = config.get('city', 'Paris') if not repaired['setup'].get('area'): repaired['setup']['area'] = config.get('area', 'Downtown') if not repaired['setup'].get('meeting_point'): repaired['setup']['meeting_point'] = f"Central point of {repaired['setup'].get('area', 'the area')}" if not repaired['setup'].get('duration_minutes'): repaired['setup']['duration_minutes'] = config.get('duration_minutes', 45) if not repaired['setup'].get('num_players'): repaired['setup']['num_players'] = config.get('num_players', 4) # Ensure story_seed if 'story_seed' not in repaired: repaired['story_seed'] = { "tone": "playful", "motifs": ["exploration", "teamwork"], "recap_style": "episode_recap" } if not repaired['story_seed'].get('tone'): repaired['story_seed']['tone'] = 'playful' if not repaired['story_seed'].get('motifs'): repaired['story_seed']['motifs'] = ["exploration", "teamwork"] if not repaired['story_seed'].get('recap_style'): repaired['story_seed']['recap_style'] = "episode_recap" # ---- 2. SAFETY REPAIRS ---- if 'safety' not in repaired or not isinstance(repaired.get('safety'), dict): repaired['safety'] = {} safety = repaired['safety'] # Fix forbidden_behaviors if 'forbidden_behaviors' not in safety or not isinstance(safety.get('forbidden_behaviors'), list) or len(safety.get('forbidden_behaviors', [])) < 2: safety['forbidden_behaviors'] = [ "Entering private buildings or restricted areas", "Crossing busy roads unsafely", "Climbing on structures", "Interacting with strangers", "Photographing people without consent" ] # Fix stop_conditions if 'stop_conditions' not in safety or not isinstance(safety.get('stop_conditions'), list) or len(safety.get('stop_conditions', [])) < 2: safety['stop_conditions'] = [ "Player injury or medical emergency", "Severe weather conditions", "Any safety concern raised by participant" ] # Fix allowed_zone if not safety.get('allowed_zone') or len(str(safety.get('allowed_zone', '')).strip()) < 5: area_name = config.get('area', repaired['setup'].get('area', 'the area')) safety['allowed_zone'] = f"Public streets, parks, and pedestrian areas within {area_name}" # Fix adult_supervision - enforce for kids/mixed regardless of current value age_group = config.get('age_group', 'adults') safety['adult_supervision'] = age_group in ['kids', 'mixed'] # ---- 3. TASK REPAIRS ---- tasks = repaired.get('tasks', []) # Fix task count for duration if needed duration = config.get('duration_minutes', repaired['setup'].get('duration_minutes', 45)) # Recalculate appropriate task count min_tasks = max(1, duration // 20) max_tasks = duration // 8 realism_failures = [f for f in failures if 'Too few tasks' in f or 'Too many tasks' in f] if realism_failures or len(tasks) < min_tasks: # Add tasks to reach minimum proof_types = ['photo', 'observation', 'text'] difficulty = config.get('difficulty', 'medium') points_map = {'easy': [10, 15], 'medium': [15, 20, 25], 'hard': [20, 25, 30]} start_idx = len(tasks) while len(tasks) < min_tasks: idx = len(tasks) pts = points_map.get(difficulty, [15, 20, 25])[idx % 3] time_limit = (duration - 5) // min_tasks if idx < min_tasks - 1 else None tasks.append({ "task_id": f"t{idx + 1}", "title": f"Task {idx + 1}", "description": f"Explore and discover something interesting in the area", "location_hint": f"Look around the central area", "points": pts, "time_limit_minutes": time_limit, "proof_type": proof_types[idx % 3], "hint": "Check visible landmarks and signs for clues", "safety_note": "Stay on public paths and respect local rules" }) elif len(tasks) > max_tasks: # Remove excess tasks (keep first max_tasks) tasks = tasks[:max_tasks] # Fix individual task fields for idx, task in enumerate(tasks): # Add task_id if missing if 'task_id' not in task or not task.get('task_id'): task['task_id'] = f"t{tasks.index(task) + 1}" # Add title if missing if 'title' not in task or not task.get('title'): task['title'] = f"Task {tasks.index(task) + 1}" # Add description if missing if 'description' not in task or not task.get('description'): area = config.get('area', 'the area') task['description'] = f"Find and document a notable feature in {area}" # Add location_hint if missing if 'location_hint' not in task or not task.get('location_hint'): task['location_hint'] = f"Search the public areas" # Fix proof_type if 'proof_type' not in task or task.get('proof_type') not in ['photo', 'observation', 'text']: proof_options = ['photo', 'observation', 'text'] task['proof_type'] = proof_options[tasks.index(task) % 3] # Fix points if 'points' not in task or not isinstance(task.get('points'), int) or task.get('points', 0) <= 0: difficulty = config.get('difficulty', 'medium') points_map = {'easy': [10, 15], 'medium': [15, 20, 25], 'hard': [20, 25, 30]} idx = tasks.index(task) % 3 task['points'] = points_map.get(difficulty, [15, 20, 25])[idx] # Fix time_limit_minutes if None or missing if 'time_limit_minutes' not in task: task['time_limit_minutes'] = max(5, duration // len(tasks)) elif task['time_limit_minutes'] is None: task['time_limit_minutes'] = None # Valid null for last task # Fix hint if 'hint' not in task or not task.get('hint') or len(str(task.get('hint', '')).strip().split()) < 2: task['hint'] = "Look for visible landmarks or signs in the area" # Fix safety_note if 'safety_note' not in task or not task.get('safety_note') or len(str(task.get('safety_note', '')).strip().split()) < 3: if age_group in ['kids', 'mixed']: task['safety_note'] = "Stay with your adult supervisor at all times" else: task['safety_note'] = "Stay in public areas and follow local regulations" repaired['tasks'] = tasks # ---- 4. RULES REPAIRS ---- rules = repaired.get('rules', []) if not rules or len(rules) < 3: area = config.get('area', repaired['setup'].get('area', 'the area')) game_type = config.get('game_type', 'scavenger_hunt') base_rules = [ f"Stay within the {area} boundaries defined at game start", "Respect private property — no entering buildings", "No running on narrow streets or busy roads", "Players must stay together as a team", "Hints may be requested but cost points" ] if game_type == 'hide_and_seek': base_rules = [ f"Stay within {area} boundaries at all times", "No hiding in locked or restricted areas", "Seeker counts to 30 at the meeting point", "Adults must maintain line-of-sight of children", "Game ends when all players are found or time elapses" ] elif game_type == 'tag': base_rules = [ f"Play zone is restricted to {area}", "No running on wet or hazardous surfaces", "Tagged players sit out briefly before rejoining", "No physical contact beyond a light touch", "Game runs in timed rounds" ] # Keep existing good rules, fill gaps with base rules existing_rules = [r for r in rules if len(r.strip()) > 10] # Merge: keep existing valid rules, add from base until we have at least 3 merged_rules = existing_rules[:] for br in base_rules: if br not in merged_rules and len(merged_rules) < 5: merged_rules.append(br) rules = merged_rules[:5] # Max 5 rules repaired['rules'] = rules # ---- 5. WIN CONDITION REPAIRS ---- if not repaired.get('tie_breaker') or len(str(repaired.get('tie_breaker', '')).strip()) < 5: repaired['tie_breaker'] = "Team with the most tasks completed wins. If tied, team that finished first wins." score_rules = repaired.get('score_rules', []) if not score_rules or len(score_rules) < 2: repaired['score_rules'] = [ "Complete each task within the time limit for full points", "Each hint used costs 5 points", "Bonus: +20 points for completing all tasks before time expires" ] global_hints = repaired.get('global_hints', []) if not global_hints or len(global_hints) == 0: repaired['global_hints'] = [ "Work together as a team to cover more ground", "Use the location hints to narrow your search", "Keep track of time for each task" ] # ---- 6. FORBIDDEN CONTENT REPAIRS ---- # Rewrite task descriptions/location_hints that describe forbidden activities. if failure_categories['forbidden']: for task in repaired.get('tasks', []): desc = task.get('description', '') hint = task.get('location_hint', '') safety_note = task.get('safety_note', '') # Rewrite: "enter/inside [place]" -> "observe [place] from outside" desc = desc.replace('enter the building', 'observe the building from outside') desc = desc.replace('enter building', 'observe the building from outside') desc = desc.replace('enter the shop', 'observe the shop from outside') desc = desc.replace('go inside the', 'approach the') desc = desc.replace('go inside', 'observe') desc = desc.replace('inside the building', 'outside the building') desc = desc.replace('inside the shop', 'outside the shop') # Rewrite: "ask staff/strangers" -> "observe" desc = desc.replace('ask staff', 'look for public signs') desc = desc.replace('ask the staff', 'look for public signs') desc = desc.replace('ask a stranger', 'observe') desc = desc.replace('ask strangers', 'observe') desc = desc.replace('talk to staff', 'look for posted information') # Rewrite location_hint hint = hint.replace('inside the building', 'near the building') hint = hint.replace('inside building', 'near the building') hint = hint.replace('inside the shop', 'near the shop') hint = hint.replace('inside the entrance', 'near the entrance') hint = hint.replace('in the building', 'near the building') task['description'] = desc task['location_hint'] = hint # After rewriting, if any forbidden keywords remain, note it in safety_note remaining_forbidden = [kw for kw in FORBIDDEN_BEHAVIORS if kw in f"{desc} {hint}".lower()] if remaining_forbidden and 'forbidden' not in safety_note.lower(): forbid_str = ', '.join(remaining_forbidden[:3]) task['safety_note'] = f"Forbidden: entering {forbid_str}. {safety_note}" if safety_note else \ f"Forbidden: entering {forbid_str}. Stay in public areas only." return repaired