Compare commits

...
Author SHA1 Message Date
KoalaTrapLord 968b68947e fix(#176): wire drawGames summary into get_result() response
Calls summarize_draw_games(state.draw_game_snapshots) after the existing
win-type injection and embeds the result under data['drawGames'] before
returning JSONResponse.  The likelyCause field immediately surfaces which
deck most often leads at timeout so the Dictate of Kruphix loop theory
can be confirmed from the dashboard without log-diving.

Part of fix #176 (companion to services/draw_game_analyzer.py).
2026-04-01 14:38:53 -04:00
KoalaTrapLord b6222cbc7c fix(#176): capture life-at-timeout for draw games + add DrawGameAnalyzer
- forge_runner._run_process_blocking: parse [DRAW] PARSE-SUMMARY lines
  and record life totals per seat into state.draw_game_snapshots
- routes/lab.py get_result(): attach draw_game_snapshots to result JSON
  under 'drawGames' so the dashboard and analysis scripts can query them
- New services/draw_game_analyzer.py: report which deck held the highest
  average life at timeout; used by /api/lab/result and scripts/
- Also lowers the default batch --clock from 6000 → 3000 per game
  (30s per-game wall limit for batch mode, down from 60s).
  Interactive games can override via req.clock which defaults stay at 6000.

Closes #176
2026-04-01 14:38:53 -04:00
KoalaTrapLord 5411fd6bf6 fix(#175): alternate win condition parser & winType tagging (#179)
Adds win_condition_parser.py, GameResult.win_type field, and winType stamping on both the Java/Forge and DeepSeek batch paths. Closes #175.
2026-04-01 14:28:10 -04:00
5 changed files with 457 additions and 11 deletions

No files matched your search

+45 -7
View File
@@ -39,6 +39,7 @@ from services.deck_service import parse_dck_file, _load_deck_cards_by_name
from services.forge_runner import (
build_java_command, run_batch_subprocess, _run_deepseek_batch_thread,
)
from services.draw_game_analyzer import summarize_draw_games
from services.logging import log, log_batch
router = APIRouter(tags=["lab"])
@@ -57,6 +58,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 +223,19 @@ 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 #175) ──────────────────────
win_type_override = getattr(state, "win_type_override", "Combat")
_inject_win_type(data, win_type_override)
# ── Inject draw-game summary (fix #176) ─────────────────────────────────
# draw_game_snapshots is populated live by forge_runner._run_process_blocking
# from [DRAW] [PARSE-SUMMARY] lines. summarize_draw_games() aggregates them
# into a single dict with perDeck breakdowns and a likelyCause field so the
# dashboard can immediately show which deck is causing timeout loops.
draw_snapshots = getattr(state, "draw_game_snapshots", [])
data["drawGames"] = summarize_draw_games(draw_snapshots)
data = _sanitize_floats(data)
return JSONResponse(content=data)
@@ -269,31 +309,31 @@ async def list_history():
AI_PROFILES = {
"default": {
"name": "default",
"description": "Balanced \u2014 Forge's default AI behavior",
"description": "Balanced — Forge's default AI behavior",
"aggression": 0.5, "cardAdvantage": 0.5, "removalPriority": 0.5,
"boardPresence": 0.5, "comboPriority": 0.3, "patience": 0.5,
},
"aggro": {
"name": "aggro",
"description": "Aggressive \u2014 attacks early, prioritizes damage",
"description": "Aggressive — attacks early, prioritizes damage",
"aggression": 0.9, "cardAdvantage": 0.3, "removalPriority": 0.3,
"boardPresence": 0.8, "comboPriority": 0.1, "patience": 0.1,
},
"control": {
"name": "control",
"description": "Control \u2014 defensive, removal-heavy, card advantage",
"description": "Control — defensive, removal-heavy, card advantage",
"aggression": 0.2, "cardAdvantage": 0.9, "removalPriority": 0.9,
"boardPresence": 0.3, "comboPriority": 0.4, "patience": 0.9,
},
"combo": {
"name": "combo",
"description": "Combo \u2014 ramps, digs for pieces, assembles combos",
"description": "Combo — ramps, digs for pieces, assembles combos",
"aggression": 0.2, "cardAdvantage": 0.8, "removalPriority": 0.4,
"boardPresence": 0.3, "comboPriority": 0.95, "patience": 0.7,
},
"midrange": {
"name": "midrange",
"description": "Midrange \u2014 flexible, strong board presence, value-oriented",
"description": "Midrange — flexible, strong board presence, value-oriented",
"aggression": 0.5, "cardAdvantage": 0.6, "removalPriority": 0.6,
"boardPresence": 0.7, "comboPriority": 0.3, "patience": 0.5,
},
@@ -469,5 +509,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',
})
+152
View File
@@ -0,0 +1,152 @@
"""
services/draw_game_analyzer.py
==============================
Analyses draw-game snapshots captured by forge_runner._run_process_blocking.
A "draw game" is any game that hit the Forge per-game wall-clock timeout
instead of ending via a win condition. Forge emits a line like:
[DRAW] [PARSE-SUMMARY] game=42 turns=28 time_ms=61200 \
seat0=AdaptiveEnchantment:life=46 seat1=AbzanArmor:life=0 \
seat2=AhoyMateys:life=0 seat3=20WaysToWin:life=8
This module:
1. Parses those lines out of the log into DrawGameSnapshot objects.
2. Aggregates them to identify which deck most often leads at timeout.
3. Exposes a summary dict suitable for embedding in the batch result JSON.
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from typing import Dict, List, Optional
# ─── regex for the Forge DRAW summary line ────────────────────────────────────
# Example:
# [DRAW] [PARSE-SUMMARY] game=42 turns=28 time_ms=61200
# seat0=AdaptiveEnchantment:life=46 seat1=AbzanArmor:life=0
# seat2=AhoyMateys:life=0 seat3=20WaysToWin:life=8
_DRAW_LINE_RE = re.compile(
r'\[DRAW\].*?\[PARSE-SUMMARY\]'
r'.*?game=(\d+)'
r'.*?turns=(\d+)'
r'.*?time_ms=(\d+)'
r'((?:\s+seat\d+=\S+)*)'
, re.IGNORECASE
)
_SEAT_RE = re.compile(r'seat(\d+)=([^:]+):life=([-\d]+)', re.IGNORECASE)
@dataclass
class DrawGameSnapshot:
game_id: int
turns: int
time_ms: int
# seat_index -> {deck_name, life}
seats: Dict[int, dict] = field(default_factory=dict)
@property
def leading_deck(self) -> Optional[str]:
"""Name of the deck with the highest life at timeout (None if tie/empty)."""
if not self.seats:
return None
best = max(self.seats.values(), key=lambda s: s["life"])
return best["deck_name"]
@property
def leading_life(self) -> int:
if not self.seats:
return 0
return max(s["life"] for s in self.seats.values())
def parse_draw_snapshots(log_lines: list[str]) -> list[DrawGameSnapshot]:
"""
Scan raw log lines for [DRAW] [PARSE-SUMMARY] entries and return
a list of DrawGameSnapshot objects, one per draw game.
"""
snapshots: list[DrawGameSnapshot] = []
for line in log_lines:
m = _DRAW_LINE_RE.search(line)
if not m:
continue
game_id = int(m.group(1))
turns = int(m.group(2))
time_ms = int(m.group(3))
seats_str = m.group(4) or ""
seats: Dict[int, dict] = {}
for sm in _SEAT_RE.finditer(seats_str):
seat_idx = int(sm.group(1))
deck_name = sm.group(2).strip()
life = int(sm.group(3))
seats[seat_idx] = {"deck_name": deck_name, "life": life}
snapshots.append(DrawGameSnapshot(game_id=game_id, turns=turns, time_ms=time_ms, seats=seats))
return snapshots
def summarize_draw_games(snapshots: list[DrawGameSnapshot]) -> dict:
"""
Aggregate draw snapshots into a summary dict for embedding in batch JSON.
Returns::
{
"count": 156,
"avgTurns": 27.4,
"avgTimeMs": 61800,
"perDeck": [
{
"deckName": "AdaptiveEnchantment",
"drawLeadCount": 89, # times this deck led at timeout
"drawLeadPct": 57.1,
"avgLifeAtTimeout": 42.3,
},
...
],
"likelyCause": "AdaptiveEnchantment" # deck most often leading at timeout
}
"""
if not snapshots:
return {"count": 0, "avgTurns": 0, "avgTimeMs": 0, "perDeck": [], "likelyCause": None}
total = len(snapshots)
avg_turns = round(sum(s.turns for s in snapshots) / total, 1)
avg_time_ms = round(sum(s.time_ms for s in snapshots) / total)
# Accumulate per-deck stats across all draw games
deck_lead_count: Dict[str, int] = {}
deck_life_sum: Dict[str, float] = {}
deck_life_count: Dict[str, int] = {}
for snap in snapshots:
leader = snap.leading_deck
if leader:
deck_lead_count[leader] = deck_lead_count.get(leader, 0) + 1
for seat in snap.seats.values():
dn = seat["deck_name"]
deck_life_sum[dn] = deck_life_sum.get(dn, 0.0) + seat["life"]
deck_life_count[dn] = deck_life_count.get(dn, 0) + 1
all_decks = set(deck_life_count.keys()) | set(deck_lead_count.keys())
per_deck = []
for dn in sorted(all_decks):
lead_cnt = deck_lead_count.get(dn, 0)
avg_life = round(deck_life_sum.get(dn, 0) / max(deck_life_count.get(dn, 1), 1), 1)
per_deck.append({
"deckName": dn,
"drawLeadCount": lead_cnt,
"drawLeadPct": round(lead_cnt / total * 100, 1),
"avgLifeAtTimeout": avg_life,
})
per_deck.sort(key=lambda x: x["drawLeadCount"], reverse=True)
likely_cause = per_deck[0]["deckName"] if per_deck else None
return {
"count": total,
"avgTurns": avg_turns,
"avgTimeMs": avg_time_ms,
"perDeck": per_deck,
"likelyCause": likely_cause,
}
+99 -4
View File
@@ -210,8 +210,44 @@ def build_java_command(
return cmd
# ── Draw-game life-at-timeout parser ─────────────────────────────────────────
# Forge emits a summary line for each game that hits the wall-clock timeout:
#
# [DRAW] [PARSE-SUMMARY] game=42 turns=28 time_ms=61200
# seat0=AdaptiveEnchantment:life=46 seat1=AbzanArmor:life=0
# seat2=AhoyMateys:life=0 seat3=20WaysToWin:life=8
#
# We capture these live so they're available before the process finishes.
_DRAW_SUMMARY_RE = re.compile(
r'\[DRAW\].*?\[PARSE-SUMMARY\]'
r'.*?game=(\d+)'
r'.*?turns=(\d+)'
r'.*?time_ms=(\d+)'
r'((?:\s+seat\d+=\S+)*)'
, re.IGNORECASE
)
_SEAT_LIFE_RE = re.compile(r'seat(\d+)=([^:]+):life=([-\d]+)', re.IGNORECASE)
def _parse_draw_summary_line(line: str) -> Optional[dict]:
"""Return a draw-game snapshot dict if line matches, else None."""
m = _DRAW_SUMMARY_RE.search(line)
if not m:
return None
seats = {}
for sm in _SEAT_LIFE_RE.finditer(m.group(4) or ""):
seat_idx = int(sm.group(1))
seats[seat_idx] = {"deck_name": sm.group(2).strip(), "life": int(sm.group(3))}
return {
"game_id": int(m.group(1)),
"turns": int(m.group(2)),
"time_ms": int(m.group(3)),
"seats": seats,
}
def _run_process_blocking(state: BatchState, cmd: list):
"""Run Forge subprocess, parse progress, generate deck reports."""
"""Run Forge subprocess, parse progress, collect draw-game snapshots."""
log.info(f"Running: {' '.join(cmd)}")
# Ensure Forge subprocesses use Java 17
@@ -237,10 +273,15 @@ def _run_process_blocking(state: BatchState, cmd: list):
_register_batch(state.batch_id, state, proc)
# ── Live stdout parsing ───────────────────────────────────────────
draw_snapshots: list = []
for line in proc.stdout:
_touch_batch(state.batch_id)
line = line.rstrip()
state.log_lines.append(line)
# Progress counters
gm = re.match(r'\[Game (\d+)/(\d+)\]', line)
if gm:
state.completed_games = int(gm.group(1))
@@ -249,19 +290,46 @@ def _run_process_blocking(state: BatchState, cmd: list):
if pm:
state.sims_per_sec = float(pm.group(1))
# ── Draw-game capture (fix #176) ──────────────────────────────
snap = _parse_draw_summary_line(line)
if snap:
draw_snapshots.append(snap)
proc.wait()
_unregister_batch(state.batch_id)
elapsed = (_datetime.now() - state.start_time).total_seconds() * 1000
state.elapsed_ms = int(elapsed)
# Persist draw snapshots on state for routes/lab.py to embed in JSON
state.draw_game_snapshots = draw_snapshots
if draw_snapshots:
log.info(
f"[DrawGames] Captured {len(draw_snapshots)} draw-game snapshots "
f"for batch {state.batch_id}"
)
if proc.returncode != 0:
state.running = False
# Include last few log lines in error for easier diagnosis
tail = ' | '.join(state.log_lines[-5:]) if state.log_lines else '(no output)'
state.error = f"Exit code {proc.returncode} — last output: {tail}"
else:
state.running = False
state.completed_games = state.total_games
# ── Post-process: detect alternate win conditions in the Java log ──
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,
@@ -269,7 +337,7 @@ async def run_batch_subprocess(
num_games: int,
threads: int,
seed: Optional[int] = None,
clock: int = 6000,
clock: int = 3000,
output_path: str = "results",
use_learned_policy: bool = False,
policy_style: str = "midrange",
@@ -278,7 +346,12 @@ async def run_batch_subprocess(
ai_think_time_ms: int = -1,
max_queue_depth: int = -1,
):
"""Async wrapper that runs Forge subprocess in executor."""
"""Async wrapper that runs Forge subprocess in executor.
clock default lowered from 6000 → 3000 (30s per-game wall limit for batch
mode) to reduce draw-game wall time wasted on timeouts (fix #176).
Interactive games pass an explicit req.clock override so they are unaffected.
"""
try:
policy_server = f"http://localhost:{CFG.port}"
cmd = build_java_command(
@@ -344,6 +417,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 +577,27 @@ def _run_deepseek_batch_thread(
)
gd = result.to_dict()
gd['gameNumber'] = g + 1
# ── Alternate win detection (DeepSeek path) ──────────────
try:
winner_seat = result.winner_seat
if winner_seat >= 0:
winner_bf = getattr(result, '_winner_battlefield', None)
if winner_bf is not None:
win_type = detect_win_type_from_cards(winner_bf)
else:
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:
+6
View File
@@ -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"