"""Deterministic scoring module. Computes team scores entirely in Python — never delegated to an LLM. Score breakdown is transparent and stored in ``scoring_explanation`` so it can be displayed in the UI and included in story packets. """ from datetime import datetime, timezone from typing import Optional # ── Constants ─────────────────────────────────────────────────────────────── HINT_PENALTY = 5 # Points deducted per hint used FAST_FINISH_BONUS = 20 # Bonus for completing all tasks before time expires COMPLETION_RATIO_BONUS = 10 # Bonus for completing >75 % of tasks MAX_TIME_BONUS = 15 # Max bonus points for finishing early # ── Internal helpers ──────────────────────────────────────────────────────── def _build_task_map(game: dict) -> dict[str, dict]: """Index tasks by ``task_id`` for O(1) lookups.""" return {t["task_id"]: t for t in game.get("tasks", [])} def _parse_ts(ts_str: str) -> Optional[datetime]: """Parse an ISO-8601 timestamp string, returning None on failure.""" try: # Handle both Z suffix and +00:00 ts = ts_str.replace("Z", "+00:00") return datetime.fromisoformat(ts) except (ValueError, TypeError): return None def _seconds_between(start: str, end: str) -> Optional[float]: """Return seconds between two ISO-8601 timestamps, or None.""" dt_start = _parse_ts(start) dt_end = _parse_ts(end) if dt_start and dt_end: return max(0, (dt_end - dt_start).total_seconds()) return None # ── Public API ────────────────────────────────────────────────────────────── def compute_scores(events: list[dict], game: dict) -> dict: """Compute final scores from gameplay events. Scoring logic: • Each ``task_completed`` awards the task's base ``points``. • Each ``hint_used`` deducts ``HINT_PENALTY`` points. • ``task_skipped`` awards 0 points for that task. • ``fast_finish`` bonus: +``FAST_FINISH_BONUS`` if all tasks are completed before the game's duration elapses. • ``completion_ratio`` bonus: +``COMPLETION_RATIO_BONUS`` if the team completed more than 75 % of tasks. • ``time_bonus``: up to ``MAX_TIME_TIME_BONUS`` points for finishing early, scaled linearly by how much time remains. Args: events: List of gameplay event dicts. game: The original game definition (with tasks, setup, etc.). Returns: Scoring output dict matching the ``event_schema`` score contract: ``{"team_scores": [...], "winner": str, "scoring_explanation": [...]}`` """ task_map = _build_task_map(game) duration_seconds = game.get("setup", {}).get("duration_minutes", 45) * 60 # ── Collect per-team data ─────────────────────────────────────────── team_data: dict[str, dict] = {} for ev in events: team_id = ev.get("team_id", "team-a") if team_id not in team_data: team_data[team_id] = { "completed_tasks": [], "skipped_tasks": [], "hints_used": 0, "photos_uploaded": 0, "journals": [], "first_event_ts": ev.get("timestamp"), "last_event_ts": ev.get("timestamp"), } td = team_data[team_id] ev_type = ev.get("event_type") payload = ev.get("payload", {}) ts = ev.get("timestamp", "") # Track time window if ts < td["first_event_ts"]: td["first_event_ts"] = ts if ts > td["last_event_ts"]: td["last_event_ts"] = ts if ev_type == "task_completed": task_id = payload.get("task_id") if task_id and task_id not in td["completed_tasks"]: td["completed_tasks"].append(task_id) elif ev_type == "task_skipped": task_id = payload.get("task_id") if task_id and task_id not in td["skipped_tasks"]: td["skipped_tasks"].append(task_id) elif ev_type == "hint_used": td["hints_used"] += 1 elif ev_type == "photo_uploaded": td["photos_uploaded"] += 1 elif ev_type == "journal_recorded": td["journals"].append(payload) # ── Score each team ───────────────────────────────────────────────── all_tasks = game.get("tasks", []) total_possible_tasks = len(all_tasks) team_scores: list[dict] = [] explanations: list[str] = [] for team_id, td in team_data.items(): base_points = 0 for task_id in td["completed_tasks"]: task = task_map.get(task_id, {}) pts = task.get("points", 0) base_points += pts # Hint penalty hint_penalty = td["hints_used"] * HINT_PENALTY # Completion ratio completed_count = len(td["completed_tasks"]) completion_ratio = completed_count / total_possible_tasks if total_possible_tasks else 0 # Time bonus elapsed = _seconds_between(td["first_event_ts"], td["last_event_ts"]) time_bonus = 0 if elapsed is not None and elapsed < duration_seconds: remaining_ratio = 1 - (elapsed / duration_seconds) time_bonus = min(MAX_TIME_BONUS, round(remaining_ratio * MAX_TIME_BONUS)) # Fast-finish bonus fast_finish = FAST_FINISH_BONUS if completed_count >= total_possible_tasks and elapsed is not None and elapsed < duration_seconds else 0 # Completion-ratio bonus completion_bonus = COMPLETION_RATIO_BONUS if completion_ratio > 0.75 else 0 total_points = max(0, base_points - hint_penalty + time_bonus + fast_finish + completion_bonus) # Build per-team explanation team_explanation = [ f"Base points from {completed_count} completed tasks: +{base_points}", f"Hints used: {td['hints_used']} × {HINT_PENALTY} penalty = -{hint_penalty}", ] if time_bonus > 0: team_explanation.append(f"Time bonus for finishing early: +{time_bonus}") if fast_finish > 0: team_explanation.append(f"Fast-finish bonus (all tasks): +{fast_finish}") if completion_bonus > 0: team_explanation.append(f"Completion ratio bonus (>75%): +{completion_bonus}") team_explanation.append(f"Final total: {total_points}") bonuses = [] if fast_finish > 0: bonuses.append("fast_finish") if completion_bonus > 0: bonuses.append("completion_ratio") team_scores.append({ "team_id": team_id, "points": total_points, "base_points": base_points, "hint_penalty": hint_penalty, "time_bonus": time_bonus, "fast_finish_bonus": fast_finish, "completion_bonus": completion_bonus, "completed_tasks": completed_count, "total_tasks": total_possible_tasks, "hints_used": td["hints_used"], "bonuses": bonuses, "scoring_breakdown": team_explanation, }) explanations.append(f"Team {team_id}: {total_points} pts ({completed_count}/{total_possible_tasks} tasks)") # ── Determine winner ──────────────────────────────────────────────── winner: Optional[str] = None if team_scores: # Sort descending by points; tie-break by fewer hints ranked = sorted( team_scores, key=lambda t: (t["points"], -t["hints_used"]), reverse=True, ) winner = ranked[0]["team_id"] if len(ranked) > 1 and ranked[0]["points"] == ranked[1]["points"]: explanations.append( f"⚠ Tie between teams with {ranked[0]['points']} pts — " "tie-breaker: fewer hints used wins." ) # Re-rank with tie-breaker ranked_tied = sorted( ranked, key=lambda t: t["hints_used"], ) winner = ranked_tied[0]["team_id"] explanations.append(f"Tie-break winner: {winner} (fewer hints)") return { "team_scores": team_scores, "winner": winner, "scoring_explanation": explanations, }