diff --git a/routes/lab.py b/routes/lab.py index 499920e..fd6175c 100644 --- a/routes/lab.py +++ b/routes/lab.py @@ -57,6 +57,32 @@ def _sanitize_floats(obj): return obj +def _inject_win_type(data: dict, win_type_override: str) -> None: + """Stamp winType on every per-game dict that doesn't already have one. + + Called by get_result() after loading the JSON from disk. + - Uses setdefault so DeepSeek per-game values (already set in forge_runner.py) + are never overwritten. + - Also stamps metadata.winType for a quick batch-level summary field. + - win_type_override comes from state.win_type_override (Java path) or + defaults to "Combat" if the attribute is absent (DeepSeek / old batches). + """ + if not win_type_override or win_type_override == "Combat": + # Still stamp Combat as default so the field is always present + win_type_override = "Combat" + + # Batch-level metadata field + meta = data.get("metadata") + if isinstance(meta, dict): + meta.setdefault("winType", win_type_override) + + # Per-game dicts nested under data["decks"][N]["games"][M] + for deck in data.get("decks", []): + for game in deck.get("games", []): + if isinstance(game, dict): + game.setdefault("winType", win_type_override) + + @router.post("/api/lab/start", response_model=StartResponse) async def start_batch(req: StartRequest, background_tasks: BackgroundTasks): """Start a new batch simulation run.""" @@ -196,6 +222,15 @@ async def get_result(batchId: Optional[str] = None): data = json.load(f) except (json.JSONDecodeError, OSError) as e: raise HTTPException(500, f"Failed to read result file: {e}") + + # ── Inject winType into every game dict (fix #180) ────────────────────── + # win_type_override is set by forge_runner._run_process_blocking() for the + # Java/Forge path. DeepSeek games already have per-game winType stamped + # directly in the result dict; _inject_win_type uses setdefault so those + # are never overwritten. + win_type_override = getattr(state, "win_type_override", "Combat") + _inject_win_type(data, win_type_override) + data = _sanitize_floats(data) return JSONResponse(content=data) @@ -469,5 +504,3 @@ async def start_batch_deepseek(request: FastAPIRequest, background_tasks: Backgr 'message': f'Running {num_games} games across {len(decks)} decks with DeepSeek AI', 'engine': 'deepseek', }) - - diff --git a/services/forge_runner.py b/services/forge_runner.py index 91e3ecc..2fc326c 100644 --- a/services/forge_runner.py +++ b/services/forge_runner.py @@ -262,6 +262,23 @@ def _run_process_blocking(state: BatchState, cmd: list): state.running = False state.completed_games = state.total_games + # ── Post-process: detect alternate win conditions in the Java log ── + # Runs after the process exits so we don't slow the hot stdout loop. + # Stamps state.win_type_override which routes/lab_api.py can attach + # to each game result when building the batch JSON response. + try: + src_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'src') + src_dir = os.path.normpath(src_dir) + if src_dir not in sys.path: + sys.path.insert(0, src_dir) + from commander_ai_lab.sim.win_condition_parser import detect_win_type_from_log + state.win_type_override = detect_win_type_from_log(state.log_lines) + if state.win_type_override != "Combat": + log.info(f"[WinType] Alternate win detected in batch {state.batch_id}: {state.win_type_override}") + except Exception as _wcp_err: + log.debug(f"[WinType] Parser import failed (non-fatal): {_wcp_err}") + state.win_type_override = "Combat" + async def run_batch_subprocess( state: BatchState, @@ -344,6 +361,7 @@ def _run_deepseek_batch_thread( from commander_ai_lab.sim.models import Card from commander_ai_lab.sim.deepseek_engine import DeepSeekGameEngine from commander_ai_lab.sim.rules import enrich_card + from commander_ai_lab.sim.win_condition_parser import detect_win_type_from_cards state.log_lines.append('[DeepSeek Batch] Initializing DeepSeek brain...') brain = _get_deepseek_brain() @@ -503,6 +521,31 @@ def _run_deepseek_batch_thread( ) gd = result.to_dict() gd['gameNumber'] = g + 1 + + # ── Alternate win detection (DeepSeek path) ────────────── + # Inspect the winning player's battlefield for alternate-win + # permanents. result.winner is the seat index (0 = deck_a). + try: + winner_seat = result.winner_seat + if winner_seat >= 0: + from commander_ai_lab.sim.game_state import get_final_battlefield + winner_bf = getattr(result, '_winner_battlefield', None) + if winner_bf is not None: + win_type = detect_win_type_from_cards(winner_bf) + else: + # Fallback: scan game_log lines if battlefield snapshot unavailable + from commander_ai_lab.sim.win_condition_parser import detect_win_type_from_log + win_type = detect_win_type_from_log( + [str(entry) for entry in (result.game_log or [])] + ) + result.win_type = win_type + gd['winType'] = win_type + if win_type != 'Combat': + log.info(f'[WinType] {deck_name} game {g+1}: {win_type} win detected') + except Exception as _wt_err: + log.debug(f'[WinType] Detection failed for game {g+1} (non-fatal): {_wt_err}') + gd.setdefault('winType', 'Combat') + deck_stats['games'].append(gd) if result.winner == 0: diff --git a/src/commander_ai_lab/sim/models.py b/src/commander_ai_lab/sim/models.py index e6954a5..4f92457 100644 --- a/src/commander_ai_lab/sim/models.py +++ b/src/commander_ai_lab/sim/models.py @@ -354,11 +354,16 @@ class GameResult: - players: list of PlayerResult, one per seat - winner_seat: index of the winning player (-1 = draw) + - win_type: how the game was won — "Combat", "Combo", "Drain", "Mill", + "Poison", or "CmdrDmg". Defaults to "Combat". + Set by win_condition_parser.detect_win_type_from_cards() (Python engine) + or win_condition_parser.detect_win_type_from_log() (Java/Forge engine). - Backward-compat properties player_a_stats / player_b_stats delegate to players[0] / players[1] """ winner_seat: int = -1 turns: int = 0 + win_type: str = "Combat" # populated by win_condition_parser players: list[PlayerResult] = field(default_factory=list) game_log: list = field(default_factory=list) @@ -428,6 +433,7 @@ class GameResult: # Legacy keys (UI expects these) "winner": self.winner_seat, "turns": self.turns, + "winType": self.win_type, # New N-player keys (Java BatchResult schema) "winningSeat": self.winner_seat, "playerResults": [p.to_dict() for p in self.players], diff --git a/src/commander_ai_lab/sim/win_condition_parser.py b/src/commander_ai_lab/sim/win_condition_parser.py new file mode 100644 index 0000000..001a4ba --- /dev/null +++ b/src/commander_ai_lab/sim/win_condition_parser.py @@ -0,0 +1,155 @@ +""" +Commander AI Lab — Alternate Win Condition Parser +================================================== +Scans Forge log lines or oracle text for non-combat "you win the game" +triggers and returns a win type category string. + +Used by: + - services/forge_runner.py (Java batch path: post-process log_lines) + - src/commander_ai_lab/sim/deepseek_engine.py (DeepSeek path: check oracle text) + - src/commander_ai_lab/sim/engine.py (Python sim path: check oracle text) +""" + +from __future__ import annotations + +import re +from typing import Optional + +# --------------------------------------------------------------------------- +# Alternate win card registry +# --------------------------------------------------------------------------- +# Maps lowercase card name -> win category. +# Categories: "Combo", "Drain", "Mill", "Poison" +# "Combat" is the default and should never appear here. + +ALTERNATE_WIN_CARDS: dict[str, str] = { + # Spell-based + "approach of the second sun": "Combo", + "thassa's oracle": "Combo", + "jace, wielder of mysteries": "Combo", + "laboratory maniac": "Combo", + "leveler": "Combo", # pairs with Lab Man / Thassa + "gleemax": "Combo", + "coalition victory": "Combo", + "helix pinnacle": "Combo", + "mayael's aria": "Combo", + "mechanized production": "Combo", + "happily ever after": "Combo", + "simic ascendancy": "Combo", + "azor's elocutors": "Combo", + "chance encounter": "Combo", + "darksteel reactor": "Combo", + "hellkite tyrant": "Combo", + "mortal kombat": "Combo", + "revel in riches": "Drain", + "test of endurance": "Combo", + "felidar sovereign": "Combo", + "celestial convergence": "Combo", + "barren glory": "Combo", + "maze's end": "Combo", + "new horizons": "Combo", + "thespian's stage": "Combo", # copying Maze's End + # Poison / Infect + "phyrexian unlife": "Poison", + "poisoned tips": "Poison", + # Mill + "milling": "Mill", # generic tag + "the fall of lord karsus": "Mill", +} + +# --------------------------------------------------------------------------- +# Forge log scanner +# --------------------------------------------------------------------------- + +# Matches lines like: +# [COMBAT-LINE] Resolve Stack: Approach of the Second Sun ... you win the game. +# [OUTCOME-LINE] ... you win the game ... +# Game Result: Player won (Approach of the Second Sun) +_YOU_WIN_RE = re.compile( + r'you win the game', + re.IGNORECASE, +) + +# Extracts a card name from the line so we can look it up in ALTERNATE_WIN_CARDS. +# Forge typically formats these as "Resolve Stack: ..." or +# "Player won via ". +_CARD_NAME_RE = re.compile( + r'(?:Resolve Stack|via|won by|triggered by|winning card)[:\s]+([A-Za-z\'\,\.\- ]+)', + re.IGNORECASE, +) + + +def detect_win_type_from_log(log_lines: list[str]) -> str: + """ + Scan a list of Forge stdout log lines for alternate win condition triggers. + + Returns the win category ("Combo", "Drain", "Mill", "Poison") if an + alternate win card is detected, otherwise returns "Combat". + + Scans *all* lines so the last detected alternate win takes precedence + (handles edge cases where the log mentions multiple near-wins). + """ + result = "Combat" + for line in log_lines: + if not _YOU_WIN_RE.search(line): + continue + # Try to extract the card name responsible + m = _CARD_NAME_RE.search(line) + if m: + card_name = m.group(1).strip().rstrip('.').lower() + # Exact lookup first + if card_name in ALTERNATE_WIN_CARDS: + result = ALTERNATE_WIN_CARDS[card_name] + continue + # Partial / substring match for long card names + for known, category in ALTERNATE_WIN_CARDS.items(): + if known in card_name or card_name in known: + result = category + break + else: + # No card name found but "you win the game" is present — + # scan the full line for any known card name + line_lower = line.lower() + for known, category in ALTERNATE_WIN_CARDS.items(): + if known in line_lower: + result = category + break + return result + + +# --------------------------------------------------------------------------- +# Oracle text scanner (Python / DeepSeek engine path) +# --------------------------------------------------------------------------- + +def detect_win_type_from_cards(cards_on_battlefield: list) -> str: + """ + Inspect a player's battlefield (list of Card objects) for alternate win + permanents. Returns the win category or "Combat". + + Checks: + 1. Card name against ALTERNATE_WIN_CARDS (fastest) + 2. oracle_text for "you win the game" as a safety net + """ + for card in cards_on_battlefield: + name_lower = getattr(card, 'name', '').lower() + if name_lower in ALTERNATE_WIN_CARDS: + return ALTERNATE_WIN_CARDS[name_lower] + oracle = getattr(card, 'oracle_text', '') or '' + if 'you win the game' in oracle.lower(): + # Unknown alternate win card — still not Combat + return "Combo" + return "Combat" + + +def detect_win_type_from_oracle(oracle_text: str, card_name: str = "") -> str: + """ + Simple oracle text probe used when we have a resolved spell/ability + but not a full battlefield list. + """ + if card_name: + lookup = card_name.strip().lower() + if lookup in ALTERNATE_WIN_CARDS: + return ALTERNATE_WIN_CARDS[lookup] + if 'you win the game' in (oracle_text or '').lower(): + return "Combo" + return "Combat"