Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3a94de4c1f | ||
|
|
c52323dd1d |
No files matched your search
+35
-2
@@ -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',
|
||||
})
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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],
|
||||
|
||||
@@ -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