Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3a94de4c1f | ||
|
|
c52323dd1d |
No files matched your search
+35
-2
@@ -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',
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|
||||||
@@ -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:
|
||||||
|
|||||||
@@ -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"
|
||||||
Reference in new issue
Block a user