From f05067bbfed16c65f7f8b120f0fe880e00b7d3a1 Mon Sep 17 00:00:00 2001 From: KoalaTrapLord Date: Sat, 14 Mar 2026 00:13:50 -0400 Subject: [PATCH] feat(brain): pipe GameResult winner_seat + per-player life totals into flush_log() for RL training --- src/commander_ai_lab/sim/deepseek_brain.py | 256 ++++++++++++--------- 1 file changed, 152 insertions(+), 104 deletions(-) diff --git a/src/commander_ai_lab/sim/deepseek_brain.py b/src/commander_ai_lab/sim/deepseek_brain.py index 5680873..0d1dd10 100644 --- a/src/commander_ai_lab/sim/deepseek_brain.py +++ b/src/commander_ai_lab/sim/deepseek_brain.py @@ -18,6 +18,8 @@ Performance features: - Response caching for identical game states - Structured JSON output with validation - Decision logging (JSONL) for future RL training data + Each JSONL entry includes game_result (winner_seat, winner_name, + player_lives, reward) once flush_log() is called with a GameResult. """ from __future__ import annotations @@ -115,7 +117,7 @@ def build_game_state_snapshot( opp = sim_state.players[1 - player_index] ctx = deck_context or {} - # ── Mana by Color (Enhancement #8) ────────────────────────── + # ── Mana by Color (Enhancement #8) ──────────────────────────── mana_total = 0 mana_by_color = {"W": 0, "U": 0, "B": 0, "R": 0, "G": 0, "C": 0} my_bf = sim_state.get_battlefield(player_index) @@ -127,7 +129,6 @@ def build_game_state_snapshot( if land_name in _LAND_COLOR_MAP: mana_by_color[_LAND_COLOR_MAP[land_name]] += 1 elif c.oracle_text: - # Parse oracle text for mana production oracle_lower = c.oracle_text.lower() if "any color" in oracle_lower: for clr in "WUBRG": @@ -138,14 +139,13 @@ def build_game_state_snapshot( if sym in oracle_lower or f"add {clr.lower()}" in oracle_lower: mana_by_color[clr] += 1 if not any(mana_by_color[k] for k in "WUBRG"): - mana_by_color["C"] += 1 # colorless + mana_by_color["C"] += 1 else: mana_by_color["C"] += 1 - # Only include colors that have sources mana_colors = {k: v for k, v in mana_by_color.items() if v > 0} - # ── My Hand — Full Oracle Text (Enhancement #5) ──────────── + # ── My Hand — Full Oracle Text (Enhancement #5) ──────────────── my_hand = [] for c in me.hand: card_info = { @@ -157,7 +157,6 @@ def build_game_state_snapshot( card_info["pt"] = c.pt if c.keywords: card_info["keywords"] = c.keywords[:5] - # Role detection roles = [] if c.is_removal: roles.append("removal") @@ -169,22 +168,17 @@ def build_game_state_snapshot( roles.append("commander") if roles: card_info["roles"] = roles - # Full oracle text — no truncation (Enhancement #5) if c.oracle_text: card_info["oracle"] = c.oracle_text - # Castability check against available mana card_info["castable"] = (c.cmc or 0) <= mana_total and not c.is_land() my_hand.append(card_info) - # ── My Board — with keywords and oracle for synergy ─────── + # ── My Board ── my_board = [] for c in my_bf: if c.is_land(): continue - entry = { - "name": c.name, - "type": _short_type(c.type_line), - } + entry = {"name": c.name, "type": _short_type(c.type_line)} if c.pt: entry["pt"] = c.pt if c.tapped: @@ -195,7 +189,7 @@ def build_game_state_snapshot( entry["is_commander"] = True my_board.append(entry) - # ── Opponent's Board — with Threat Assessment (Enhancement #9) + # ── Opponent's Board — with Threat Assessment (Enhancement #9) ── opp_board = [] opp_total_power = 0 opp_evasion_count = 0 @@ -203,10 +197,7 @@ def build_game_state_snapshot( for c in opp_bf: if c.is_land(): continue - entry = { - "name": c.name, - "type": _short_type(c.type_line), - } + entry = {"name": c.name, "type": _short_type(c.type_line)} if c.pt: entry["pt"] = c.pt opp_total_power += c.get_power() @@ -221,7 +212,6 @@ def build_game_state_snapshot( opp_protected_count += 1 opp_board.append(entry) - # Threat level summary if opp_total_power >= me.life: threat_level = "LETHAL" elif opp_total_power >= me.life * 0.5: @@ -233,7 +223,7 @@ def build_game_state_snapshot( else: threat_level = "NONE" - # ── Graveyard Contents (Enhancement #6) ──────────────────── + # ── Graveyard Contents (Enhancement #6) ────────────────────── my_gy = [{"name": c.name, "type": _short_type(c.type_line)} for c in me.graveyard[:15]] opp_gy = [{"name": c.name, "type": _short_type(c.type_line)} for c in opp.graveyard[:10]] @@ -243,7 +233,6 @@ def build_game_state_snapshot( cards_remaining = len(me.library) pct_drawn = round(cards_seen / max(total_deck_size, 1) * 100) - # Key cards still in library (if we know the decklist) key_cards_in_library = [] if ctx.get("key_cards"): library_names = {c.name.lower() for c in me.library} @@ -251,7 +240,7 @@ def build_game_state_snapshot( if kc.lower() in library_names: key_cards_in_library.append(kc) - # ── Build the snapshot ───────────────────────────────────── + # ── Build the snapshot ──────────────────────────────────── snapshot = { "turn": turn + 1, "my_life": me.life, @@ -274,11 +263,9 @@ def build_game_state_snapshot( "deck_drawn_pct": pct_drawn, } - # ── Commander Identity (Enhancement #2) ──────────────────── if ctx.get("commander_name"): - # Determine commander zone cmd_name_lower = ctx["commander_name"].lower() - cmd_zone = "library" # default + cmd_zone = "library" for c in me.hand: if c.name.lower() == cmd_name_lower: cmd_zone = "hand" @@ -291,26 +278,18 @@ def build_game_state_snapshot( if c.name.lower() == cmd_name_lower: cmd_zone = "graveyard" break - snapshot["commander"] = { "name": ctx["commander_name"], "zone": cmd_zone, "color_identity": ctx.get("color_identity", []), } - # ── Full Decklist Awareness (Enhancement #1) ─────────────── if ctx.get("deck_summary"): snapshot["deck_summary"] = ctx["deck_summary"] - - # ── Win Conditions & Combos (Enhancement #3) ─────────────── if ctx.get("combo_pieces"): snapshot["combo_pieces"] = ctx["combo_pieces"] - - # ── Key Cards in Library (Enhancement #7) ────────────────── if key_cards_in_library: snapshot["key_cards_in_library"] = key_cards_in_library[:10] - - # ── Historical Win Rate (Enhancement #10) ────────────────── if ctx.get("win_rate") is not None: snapshot["historical_win_rate"] = ctx["win_rate"] @@ -340,7 +319,7 @@ def _short_type(type_line: str) -> str: def build_deck_context( - full_deck: list, # list[Card] — the full original deck + full_deck: list, commander_name: str = "", color_identity: list[str] | None = None, archetype: str = "midrange", @@ -360,7 +339,6 @@ def build_deck_context( if win_rate is not None: ctx["win_rate"] = win_rate - # ── Deck Composition Summary (Enhancement #1) ────────────── type_counts: dict[str, int] = {} roles: dict[str, list[str]] = {"removal": [], "ramp": [], "board_wipe": [], "draw": []} key_cards: list[str] = [] @@ -371,32 +349,25 @@ def build_deck_context( for card in full_deck: ctype = _short_type(card.type_line) if card.type_line else "?" type_counts[ctype] = type_counts.get(ctype, 0) + 1 - if not card.is_land(): total_cmc += card.cmc or 0 non_land_count += 1 - - # Track roles if card.is_removal: roles["removal"].append(card.name) if card.is_board_wipe: roles["board_wipe"].append(card.name) if card.is_ramp: roles["ramp"].append(card.name) - # Detect draw engines oracle = (card.oracle_text or "").lower() if re.search(r"draw .* cards?", oracle): roles["draw"].append(card.name) - - # ── Win Conditions & Combos (Enhancement #3) ────────── for pattern, label in _COMBO_PATTERNS: if re.search(pattern, oracle): combo_pieces.append({"name": card.name, "role": label}) key_cards.append(card.name) - break # one label per card + break avg_cmc = round(total_cmc / max(non_land_count, 1), 1) - ctx["deck_summary"] = { "card_types": type_counts, "avg_cmc": avg_cmc, @@ -406,10 +377,8 @@ def build_deck_context( "draw_count": len(roles["draw"]), "archetype": archetype, } - - ctx["combo_pieces"] = combo_pieces[:15] # cap to save tokens + ctx["combo_pieces"] = combo_pieces[:15] ctx["key_cards"] = list(set(key_cards))[:15] - return ctx @@ -417,20 +386,18 @@ def build_deck_context( # Action Schema (Step 3) # ══════════════════════════════════════════════════════════════ -# Valid actions the LLM can return VALID_ACTIONS = [ - "play_land", # Play a land from hand - "cast_creature", # Cast a creature spell - "cast_removal", # Cast targeted removal - "cast_board_wipe", # Cast a board wipe - "cast_ramp", # Cast a ramp spell - "cast_spell", # Cast any other non-creature spell - "attack_all", # Attack with all eligible creatures - "attack_safe", # Attack only with evasive/large creatures - "hold", # Pass / don't act (save mana, bluff) + "play_land", + "cast_creature", + "cast_removal", + "cast_board_wipe", + "cast_ramp", + "cast_spell", + "attack_all", + "attack_safe", + "hold", ] -# Action schema for the LLM's JSON response ACTION_SCHEMA = { "action": "one of: " + ", ".join(VALID_ACTIONS), "target_card": "(optional) name of card to cast or target", @@ -529,14 +496,19 @@ class DeepSeekBrain: """ LLM-powered decision engine for the AI opponent. - Sends game state to a local DeepSeek model and parses the action response. + Sends game state to a local DeepSeek/GPT-OSS model and parses the action response. Falls back to heuristic on timeout/error. + + Decision log entries include game outcome once flush_log() is called + with a GameResult, enabling downstream RL weight learning. """ def __init__(self, config: DeepSeekConfig | None = None): self.config = config or DeepSeekConfig() self._cache: OrderedDict[str, dict] = OrderedDict() self._decision_log: list[dict] = [] + self._current_game_id: str = self._new_game_id() + self._ai_player_index: int | None = None # set by caller if known self._total_calls = 0 self._total_fallbacks = 0 self._total_cache_hits = 0 @@ -544,6 +516,21 @@ class DeepSeekBrain: self._connected = False self._log_file = None + @staticmethod + def _new_game_id() -> str: + """Generate a short unique game ID for grouping decisions in the log.""" + return hashlib.md5(str(time.time()).encode()).hexdigest()[:12] + + def new_game(self, ai_player_index: int | None = None) -> None: + """ + Signal the start of a new game. + + Call this before each game so that log entries are grouped by game_id + and the ai_player_index is recorded for reward calculation. + """ + self._current_game_id = self._new_game_id() + self._ai_player_index = ai_player_index + def check_connection(self) -> bool: """Test if the LLM endpoint is reachable.""" try: @@ -555,7 +542,6 @@ class DeepSeekBrain: self._connected = True models = data.get("data", []) if models: - # Auto-detect model name from LM Studio model_id = models[0].get("id", self.config.model) if model_id: self.config.model = model_id @@ -569,20 +555,20 @@ class DeepSeekBrain: def choose_action( self, - sim_state, # SimState - player_index: int, # Which player is the AI + sim_state, + player_index: int, turn: int, available_mana: int | None = None, - deck_context: dict | None = None, # Full deck intelligence from build_deck_context() + deck_context: dict | None = None, ) -> dict: """ Choose an action for the AI player. Returns: { - "action": str, # One of VALID_ACTIONS - "target_card": str, # Optional card name - "reasoning": str, # Brief explanation + "action": str, + "target_card": str, + "reasoning": str, "source": "deepseek" | "heuristic" | "cache", "latency_ms": float, } @@ -590,12 +576,10 @@ class DeepSeekBrain: self._total_calls += 1 t_start = time.time() - # Build game state snapshot with deck intelligence snapshot = build_game_state_snapshot(sim_state, player_index, turn, deck_context=deck_context) if available_mana is not None: snapshot["my_mana_available"] = available_mana - # Check cache if self.config.cache_enabled: cache_key = self._cache_key(snapshot) if cache_key in self._cache: @@ -605,7 +589,6 @@ class DeepSeekBrain: cached["latency_ms"] = round((time.time() - t_start) * 1000, 1) return cached - # Try LLM if not self._connected: return self._fallback_action(snapshot, t_start, "not_connected") @@ -614,13 +597,11 @@ class DeepSeekBrain: result["source"] = "deepseek" result["latency_ms"] = round((time.time() - t_start) * 1000, 1) - # Cache the result if self.config.cache_enabled: self._put_cache(cache_key, result) - # Log decision if self.config.log_decisions: - self._log_decision(snapshot, result) + self._log_decision(snapshot, result, player_index) self._total_latency_ms += result["latency_ms"] return result @@ -633,11 +614,9 @@ class DeepSeekBrain: def _call_llm(self, snapshot: dict) -> dict: """Send game state to LLM and parse response.""" - # Build the prompt game_state_json = json.dumps(snapshot, indent=2) user_prompt = USER_PROMPT_TEMPLATE.format(game_state_json=game_state_json) - # Build OpenAI-compatible request payload = { "model": self.config.model, "messages": [ @@ -658,7 +637,6 @@ class DeepSeekBrain: with urlopen(req, timeout=self.config.request_timeout) as resp: resp_data = json.loads(resp.read()) - # Extract the response text choices = resp_data.get("choices", []) if not choices: raise ValueError("Empty response from LLM") @@ -672,18 +650,12 @@ class DeepSeekBrain: Handles common LLM quirks (markdown fences, extra text, etc.) """ text = raw_text.strip() - - # Strip ... tags (DeepSeek-R1 reasoning) - # Handle both closed tags and unclosed tags (truncated by max_tokens) text = re.sub(r'.*?', '', text, flags=re.DOTALL).strip() text = re.sub(r'.*', '', text, flags=re.DOTALL).strip() - - # Strip markdown code fences text = re.sub(r'^```(?:json)?\s*', '', text) text = re.sub(r'\s*```$', '', text) text = text.strip() - # Try to find a JSON object in the text json_match = re.search(r'\{[^{}]*\}', text, re.DOTALL) if json_match: text = json_match.group(0) @@ -691,7 +663,6 @@ class DeepSeekBrain: try: parsed = json.loads(text) except json.JSONDecodeError: - # Last resort: try to extract action keyword for action in VALID_ACTIONS: if action in raw_text.lower(): return { @@ -701,16 +672,14 @@ class DeepSeekBrain: } raise ValueError(f"Could not parse LLM response as JSON: {raw_text[:200]}") - # Validate action action = parsed.get("action", "hold") if action not in VALID_ACTIONS: - # Fuzzy match for valid in VALID_ACTIONS: if valid in action.lower() or action.lower() in valid: action = valid break else: - action = "hold" # safe default + action = "hold" return { "action": action, @@ -728,18 +697,15 @@ class DeepSeekBrain: opp_board = snapshot.get("opp_board", []) turn = snapshot.get("turn", 1) - # Priority: land > ramp (early) > removal (if outnumbered) > creature > attack > hold action = "hold" target = "" reasoning = f"Fallback ({reason})" - # 1. Play a land if available lands_in_hand = [c for c in hand if c.get("type") == "land"] if lands_in_hand: action = "play_land" target = lands_in_hand[0]["name"] reasoning = "Play land first" - # 2. Early game: prioritize ramp elif turn <= 4: ramp_cards = [c for c in hand if c.get("role") == "ramp" and c.get("cmc", 99) <= mana] if ramp_cards: @@ -748,38 +714,31 @@ class DeepSeekBrain: target = ramp_cards[0]["name"] reasoning = "Early ramp" else: - # Play cheapest creature creatures = [c for c in hand if c.get("type") == "creature" and c.get("cmc", 99) <= mana] if creatures: creatures.sort(key=lambda c: c.get("cmc", 0)) action = "cast_creature" target = creatures[0]["name"] reasoning = "Cheap creature development" - # 3. If opponent has many creatures and we have board wipe elif len(opp_board) >= 3 and len(my_board) <= 1: wipes = [c for c in hand if c.get("role") == "board_wipe" and c.get("cmc", 99) <= mana] if wipes: action = "cast_board_wipe" target = wipes[0]["name"] reasoning = "Opponent has board advantage, wiping" - # 4. If opponent has a big threat, use removal elif opp_board: removal = [c for c in hand if c.get("role") == "removal" and c.get("cmc", 99) <= mana] if removal: action = "cast_removal" target = removal[0]["name"] reasoning = "Remove opponent threat" - # 5. Play best creature if action == "hold": castable = [c for c in hand if c.get("type") == "creature" and c.get("cmc", 99) <= mana] if castable: - # Pick the one with highest power castable.sort(key=lambda c: -int(c.get("pt", "0/0").split("/")[0]) if c.get("pt") else 0) action = "cast_creature" target = castable[0]["name"] reasoning = "Play threat" - - # 6. Attack if we have creatures on board if action == "hold" and my_board: creatures_on_board = [c for c in my_board if c.get("type") == "creature" and not c.get("tapped")] if creatures_on_board: @@ -802,13 +761,10 @@ class DeepSeekBrain: # ── Cache helpers ── def _cache_key(self, snapshot: dict) -> str: - """Create a deterministic cache key from game state.""" - # Normalize — exclude things that don't affect decisions key_data = json.dumps(snapshot, sort_keys=True) return hashlib.md5(key_data.encode()).hexdigest() def _put_cache(self, key: str, value: dict): - """LRU cache insert.""" if key in self._cache: self._cache.move_to_end(key) self._cache[key] = value @@ -817,10 +773,17 @@ class DeepSeekBrain: # ── Decision logging ── - def _log_decision(self, snapshot: dict, result: dict): - """Log decision for future training data.""" + def _log_decision(self, snapshot: dict, result: dict, player_index: int | None = None) -> None: + """ + Append a decision entry to the in-memory log. + + The entry is tagged with game_id so flush_log() can backfill + game_result fields once the game is complete. + """ entry = { "timestamp": time.time(), + "game_id": self._current_game_id, + "player_index": player_index if player_index is not None else self._ai_player_index, "game_state": snapshot, "decision": { "action": result.get("action"), @@ -829,14 +792,96 @@ class DeepSeekBrain: "source": result.get("source"), "latency_ms": result.get("latency_ms"), }, + # game_result is None until flush_log() is called with a GameResult + "game_result": None, } self._decision_log.append(entry) - def flush_log(self, filepath: str | None = None): - """Write accumulated decision log to JSONL file.""" - if not self._decision_log: - return + def flush_log( + self, + filepath: str | None = None, + game_result=None, # Optional[GameResult] from engine.py + ) -> str | None: + """ + Write accumulated decision log to a JSONL file. + Parameters + ---------- + filepath : str, optional + Destination file path. Defaults to logs/decisions/.jsonl + next to the package root. + game_result : GameResult, optional + Result from GameEngine.run() / run_n(). When provided, every + buffered entry is backfilled with: + + ``game_result`` dict containing: + - ``winner_seat`` : int — seat index of the winner + - ``winner_name`` : str — display name of the winning player + - ``turns`` : int — total game length in turns + - ``player_lives`` : dict — {seat_index: final_life} for all seats + - ``reward`` : float — normalised reward in [-1.0, 1.0] + for the AI player recorded in each entry + (winner_life - loser_life) / 40.0 + + Returns the absolute path written, or None if the log was empty. + """ + if not self._decision_log: + return None + + # ── Build game_result payload from GameResult dataclass ── + result_payload: dict | None = None + if game_result is not None: + try: + winner_seat: int = game_result.winner_seat + winner_name: str = game_result.players[winner_seat].name + turns: int = game_result.turns + + # Per-seat final life totals + player_lives: dict[int, int] = { + pr.seat_index: pr.life + for pr in game_result.players + } + + result_payload = { + "winner_seat": winner_seat, + "winner_name": winner_name, + "turns": turns, + "player_lives": player_lives, + } + except Exception as exc: + logger.warning("Could not extract game_result fields: %s", exc) + + # ── Backfill each entry with game_result + per-entry reward ── + for entry in self._decision_log: + if result_payload is None: + entry["game_result"] = None + continue + + # Determine reward for the player who made this decision + pi = entry.get("player_index") + winner_seat = result_payload["winner_seat"] + player_lives = result_payload["player_lives"] + + if pi is not None and len(player_lives) >= 2: + my_life = player_lives.get(pi, 0) + # Sum of all opponents' final life totals + opp_life_sum = sum( + v for k, v in player_lives.items() if k != pi + ) + opp_count = max(len(player_lives) - 1, 1) + avg_opp_life = opp_life_sum / opp_count + raw_reward = (my_life - avg_opp_life) / 40.0 + reward = round(max(-1.0, min(1.0, raw_reward)), 4) + else: + # Fallback: binary win/loss + reward = 1.0 if (pi is not None and pi == winner_seat) else -1.0 + + entry["game_result"] = { + **result_payload, + "reward": reward, + } + + # ── Write to JSONL ── if filepath is None: log_dir = self.config.log_dir or os.path.join( os.path.dirname(__file__), "..", "..", "..", "logs", "decisions" @@ -859,7 +904,9 @@ class DeepSeekBrain: def get_stats(self) -> dict: """Return performance statistics.""" avg_latency = ( - self._total_latency_ms / max(self._total_calls - self._total_fallbacks - self._total_cache_hits, 1) + self._total_latency_ms / max( + self._total_calls - self._total_fallbacks - self._total_cache_hits, 1 + ) ) return { "total_calls": self._total_calls, @@ -871,4 +918,5 @@ class DeepSeekBrain: "model": self.config.model, "api_base": self.config.api_base, "pending_log_entries": len(self._decision_log), + "current_game_id": self._current_game_id, }