Compare commits

...
Author SHA1 Message Date
KoalaTrapLord 3a94de4c1f fix(#180): wire win_type_override into Java batch JSON output in get_result()
After loading the result JSON from disk, stamp winType on every game dict
that doesn't already have one (setdefault preserves DeepSeek per-game values).
Also stamps metadata.winType so callers have a batch-level summary field.

Closes #180
2026-04-01 13:44:20 -04:00
KoalaTrapLord c52323dd1d fix(#175): add alternate win condition parser
- New win_condition_parser.py with ALTERNATE_WIN_CARDS dict and detect_win_type()
- GameResult.win_type field added to models.py, emitted in to_dict()
- forge_runner.py Java path: scan log_lines post-run for 'you win the game'
- forge_runner.py DeepSeek path: stamp winType on each game result dict

Closes #175
2026-04-01 13:32:32 -04:00
4 changed files with 239 additions and 2 deletions

No files matched your search

+35 -2
View File
@@ -57,6 +57,32 @@ def _sanitize_floats(obj):
return 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) @router.post("/api/lab/start", response_model=StartResponse)
async def start_batch(req: StartRequest, background_tasks: BackgroundTasks): async def start_batch(req: StartRequest, background_tasks: BackgroundTasks):
"""Start a new batch simulation run.""" """Start a new batch simulation run."""
@@ -196,6 +222,15 @@ async def get_result(batchId: Optional[str] = None):
data = json.load(f) data = json.load(f)
except (json.JSONDecodeError, OSError) as e: except (json.JSONDecodeError, OSError) as e:
raise HTTPException(500, f"Failed to read result file: {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) data = _sanitize_floats(data)
return JSONResponse(content=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', 'message': f'Running {num_games} games across {len(decks)} decks with DeepSeek AI',
'engine': 'deepseek', 'engine': 'deepseek',
}) })
+43
View File
@@ -262,6 +262,23 @@ def _run_process_blocking(state: BatchState, cmd: list):
state.running = False state.running = False
state.completed_games = state.total_games 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( async def run_batch_subprocess(
state: BatchState, state: BatchState,
@@ -344,6 +361,7 @@ def _run_deepseek_batch_thread(
from commander_ai_lab.sim.models import Card from commander_ai_lab.sim.models import Card
from commander_ai_lab.sim.deepseek_engine import DeepSeekGameEngine from commander_ai_lab.sim.deepseek_engine import DeepSeekGameEngine
from commander_ai_lab.sim.rules import enrich_card 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...') state.log_lines.append('[DeepSeek Batch] Initializing DeepSeek brain...')
brain = _get_deepseek_brain() brain = _get_deepseek_brain()
@@ -503,6 +521,31 @@ def _run_deepseek_batch_thread(
) )
gd = result.to_dict() gd = result.to_dict()
gd['gameNumber'] = g + 1 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) deck_stats['games'].append(gd)
if result.winner == 0: if result.winner == 0:
+6
View File
@@ -354,11 +354,16 @@ class GameResult:
- players: list of PlayerResult, one per seat - players: list of PlayerResult, one per seat
- winner_seat: index of the winning player (-1 = draw) - 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] - Backward-compat properties player_a_stats / player_b_stats delegate to players[0] / players[1]
""" """
winner_seat: int = -1 winner_seat: int = -1
turns: int = 0 turns: int = 0
win_type: str = "Combat" # populated by win_condition_parser
players: list[PlayerResult] = field(default_factory=list) players: list[PlayerResult] = field(default_factory=list)
game_log: list = field(default_factory=list) game_log: list = field(default_factory=list)
@@ -428,6 +433,7 @@ class GameResult:
# Legacy keys (UI expects these) # Legacy keys (UI expects these)
"winner": self.winner_seat, "winner": self.winner_seat,
"turns": self.turns, "turns": self.turns,
"winType": self.win_type,
# New N-player keys (Java BatchResult schema) # New N-player keys (Java BatchResult schema)
"winningSeat": self.winner_seat, "winningSeat": self.winner_seat,
"playerResults": [p.to_dict() for p in self.players], "playerResults": [p.to_dict() for p in self.players],
@@ -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: <CardName> ..." or
# "Player won via <CardName>".
_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"