fez5587/commander-ai-lab
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(policy): consume state_vector from GameSession snapshots (#83 Step 2)

Browse Source 
DecideRequest now accepts an optional state_vector field (float[29]).
When present, /api/policy/decide uses it directly for the global scalar
block instead of re-deriving it from nested player fields.

StateEncoder gets a new encode_with_precomputed_global() path that:
  - Accepts the precomputed float[29] global block from Java
  - Still runs full zone embedding pooling (hand/battlefield/graveyard/
    command_zone) from the players[] zone contents
  - Appends the playstyle one-hot as normal
  - Returns the same 6177-dim tensor the policy network expects

Fallback: if state_vector is absent or wrong length, encode() is called
as before so all existing callers (batch predict, JSONL training) are
unaffected.

Also maps GameSession player field names to encoder field names:
  handCount  -> hand size proxy when hand[] is empty (AI seats)
  battlefield list -> creatures count proxy
  commanderTax -> cmdr_tax
  turnNumber  -> turn
  activePlayer -> active_seat
This commit is contained in:
KoalaTrapLord committed 2026-03-29 04:32:52 -04:00
1 parent 126684bb94
commit 4d73f134c2
1 file changed
+276 -32
Download Patch File Download Diff File Expand all files Collapse all files
routes/policy.py
+276 -32
View File
@@ -16,9 +16,12 @@ Endpoints:
Refs: Issue #83 Phase 2.2
"""
import logging
import math
import time
from typing import Dict, List, Optional
import numpy as np
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
@@ -26,38 +29,86 @@ logger = logging.getLogger("routes.policy")
router = APIRouter(prefix="/api/policy", tags=["policy"])
# Expected length of the precomputed global scalar block from GameSession
_EXPECTED_STATE_VECTOR_LEN = 29
# ================================================================
# Request / Response Models
# ================================================================
class PlayerZoneState(BaseModel):
"""Per-player zone state from Forge."""
"""Per-player zone state from Forge / GameSession."""
seat: int = 0
name: str = ""
life: int = 40
poison: int = 0
# GameSession sends cmdr_dmg as a dict keyed by opponent seat str
cmdr_dmg: Dict[str, int] = Field(default_factory=dict)
cmdr_tax: int = 0
commanderTax: int = 0 # GameSession alias for cmdr_tax
mana_available: int = 0
mana_pool: Dict[str, int] = Field(default_factory=dict)
# Full zone contents
hand: List[str] = Field(default_factory=list)
hand_count: int = 0
battlefield: List[Dict] = Field(default_factory=list)
handCount: int = 0 # GameSession alias for hand_count
battlefield: List = Field(default_factory=list) # list[str] or list[dict]
graveyard: List[str] = Field(default_factory=list)
exile: List[str] = Field(default_factory=list)
command_zone: List[str] = Field(default_factory=list)
commandZone: List[str] = Field(default_factory=list) # GameSession alias
library_count: int = 0
def resolved_hand(self) -> List[str]:
"""Return hand card names from whichever field is populated."""
return self.hand if self.hand else []
def resolved_hand_count(self) -> int:
"""Return hand count from whichever alias is set."""
if self.hand:
return len(self.hand)
return self.handCount or self.hand_count
def resolved_cmdr_tax(self) -> int:
"""Return commander tax from whichever alias is set."""
return self.cmdr_tax or self.commanderTax
def resolved_command_zone(self) -> List[str]:
"""Return command zone from whichever alias is set."""
return self.command_zone if self.command_zone else self.commandZone
def resolved_battlefield_names(self) -> List[str]:
"""Return battlefield as a flat list of card name strings."""
names = []
for entry in self.battlefield:
if isinstance(entry, str):
names.append(entry)
elif isinstance(entry, dict):
names.append(entry.get("name", entry.get("id", "unknown")))
return names
class DecideRequest(BaseModel):
"""Full Forge game state for live decision-making."""
"""Full game state for live decision-making.
Accepts snapshots from both:
- The legacy JSONL / batch path (players[] with flat scalar fields)
- GameSession.buildStateSnapshot() (players[] with zone lists,
plus a precomputed state_vector float[29] global block)
When state_vector is present and length == 29, it is used directly
as the global scalar block and zone embedding pooling runs over the
players[] zone contents. This avoids re-deriving scalars that Java
already computed and normalised.
"""
# Game metadata
game_id: str = ""
turn: int = 1
turnNumber: int = 0 # GameSession alias for turn
phase: str = "main_1"
active_player: int = 0
activePlayer: int = 0 # GameSession alias
priority_player: int = 0
# Per-player state with full zones
players: List[PlayerZoneState] = Field(default_factory=list)
@@ -65,6 +116,7 @@ class DecideRequest(BaseModel):
stack: List[Dict] = Field(default_factory=list)
# Legal actions from Forge
legal_actions: List[Dict] = Field(default_factory=list)
legalActions: List[Dict] = Field(default_factory=list) # GameSession alias
# Inference params
playstyle: str = "midrange"
greedy: bool = False
@@ -72,6 +124,25 @@ class DecideRequest(BaseModel):
# Optional commander info
commander: str = ""
deck_name: str = ""
# Precomputed global scalar vector from GameSession.buildStateVector()
# float[29] matching STATE_DIMS.global_features in ml/config/scope.py
state_vector: Optional[List[float]] = None
state_vector_dim: Optional[int] = None
# Schema version (GameSession 1.1.0+ includes state_vector)
schema: str = "1.0.0"
def resolved_turn(self) -> int:
return self.turnNumber if self.turnNumber > 0 else self.turn
def resolved_active_seat(self) -> int:
return self.activePlayer if self.activePlayer > 0 else self.active_player
def has_precomputed_vector(self) -> bool:
"""True when a valid precomputed state_vector is present."""
return (
self.state_vector is not None
and len(self.state_vector) == _EXPECTED_STATE_VECTOR_LEN
)
class DecideResponse(BaseModel):
@@ -83,6 +154,7 @@ class DecideResponse(BaseModel):
value: float = 0.0
probabilities: Dict[str, float] = Field(default_factory=dict)
inference_ms: float = 0.0
vector_source: str = "encoder" # "precomputed" | "encoder"
error: Optional[str] = None
@@ -114,6 +186,8 @@ _session_stats = {
"games_completed": 0,
"session_start": 0.0,
"last_decision_ms": 0.0,
"precomputed_vector_hits": 0,
"encoder_fallbacks": 0,
}
@@ -140,37 +214,68 @@ def register_policy_routes(app, policy_service, online_store=None):
@router.post("/decide", response_model=DecideResponse)
async def decide(req: DecideRequest):
"""Accept full Forge game state, run policy inference, return action.
"""Accept full game state, run policy inference, return action.
When the request includes a precomputed state_vector (float[29] from
GameSession.buildStateVector()), that vector is used directly as the
global scalar block and only zone embedding pooling is re-run.
Otherwise falls back to the full StateEncoder.encode() path.
This is the primary endpoint for live Forge IPC. The Java
PolicyClient calls this at each decision point during a real
Forge game.
Returns the macro-action plus log_prob and value estimate
needed for online PPO updates.
"""
if _policy_service is None or not _policy_service._loaded:
raise HTTPException(
503,
detail="Policy model not loaded"
)
raise HTTPException(503, detail="Policy model not loaded")
t_start = time.time()
_session_stats["total_decisions"] += 1
# Convert request to snapshot dict for the encoder
snapshot = req.dict(exclude={"temperature", "greedy", "playstyle"})
playstyle = req.playstyle or "midrange"
vector_source = "encoder"
# Map phase names from Forge conventions
snapshot["phase"] = _map_forge_phase(req.phase)
snapshot["active_seat"] = req.active_player
if req.has_precomputed_vector():
# Fast path: Java already computed and normalised the global scalars.
# Only run zone embedding pooling and playstyle one-hot, then
# concatenate with the precomputed global block.
try:
state_vec = _encode_with_precomputed_global(
precomputed_global=req.state_vector,
players=req.players,
playstyle=playstyle,
policy_service=_policy_service,
)
vector_source = "precomputed"
_session_stats["precomputed_vector_hits"] += 1
logger.debug(
"[decide] precomputed global block used (schema=%s, turn=%d)",
req.schema, req.resolved_turn(),
)
except Exception as e:
# Fall through to the full encoder on any error
logger.warning("Precomputed vector encode failed (%s) — falling back", e)
state_vec = None
else:
state_vec = None
result = _policy_service.predict(
snapshot,
playstyle=req.playstyle,
temperature=req.temperature,
greedy=req.greedy,
)
if state_vec is None:
# Standard path: build snapshot dict and run full StateEncoder.encode()
_session_stats["encoder_fallbacks"] += 1
snapshot = _build_encoder_snapshot(req)
result = _policy_service.predict(
snapshot,
playstyle=playstyle,
temperature=req.temperature,
greedy=req.greedy,
)
else:
# Run inference directly on the pre-built state tensor
result = _infer_from_vector(
state_vec=state_vec,
policy_service=_policy_service,
temperature=req.temperature,
greedy=req.greedy,
)
elapsed_ms = (time.time() - t_start) * 1000
_session_stats["last_decision_ms"] = elapsed_ms
@@ -178,11 +283,8 @@ async def decide(req: DecideRequest):
if "error" in result:
raise HTTPException(500, detail=result["error"])
# Add log_prob and value estimate for online learning
log_prob = 0.0
value = 0.0
try:
import math
conf = result.get("confidence", 0.5)
log_prob = math.log(max(conf, 1e-8))
except Exception:
@@ -193,9 +295,10 @@ async def decide(req: DecideRequest):
action_index=result["action_index"],
confidence=result["confidence"],
log_prob=round(log_prob, 6),
value=round(value, 6),
value=0.0,
probabilities=result.get("probabilities", {}),
inference_ms=round(elapsed_ms, 2),
vector_source=vector_source,
)
@@ -232,9 +335,7 @@ async def collect_tuples(req: CollectRequest):
@router.post("/reward")
async def submit_reward(req: RewardRequest):
"""Submit end-of-game reward for all tuples in a game.
Propagates the terminal reward back through the game's trajectory.
"""
"""Submit end-of-game reward for all tuples in a game."""
_session_stats["total_rewards_submitted"] += 1
_session_stats["games_completed"] += 1
@@ -295,9 +396,146 @@ async def policy_stats():
# ================================================================
# Helpers
# Internal helpers
# ================================================================
def _encode_with_precomputed_global(
precomputed_global: List[float],
players: List[PlayerZoneState],
playstyle: str,
policy_service,
) -> np.ndarray:
"""Build a 6177-dim state vector using a precomputed global block.
Uses the Java-computed float[29] global scalar vector directly and
runs zone embedding pooling + playstyle one-hot on the Python side.
Args:
precomputed_global: float[29] from GameSession.buildStateVector()
players: PlayerZoneState list from the request (for zone contents)
playstyle: deck archetype string
policy_service: PolicyInferenceService (for encoder access)
Returns:
np.ndarray of shape (6177,)
"""
encoder = policy_service.encoder
if encoder is None:
raise RuntimeError("StateEncoder not initialised on policy service")
# 1. Precomputed global block (29 floats)
global_vec = np.array(precomputed_global, dtype=np.float32)
# 2. Zone embeddings — pool card names from each player's zones
player_dicts = [
{
"hand": p.resolved_hand(),
"battlefield": p.resolved_battlefield_names(),
"graveyard": p.graveyard,
"command_zone": p.resolved_command_zone(),
}
for p in players[:2]
]
# Pad to 2 players if needed
while len(player_dicts) < 2:
player_dicts.append({
"hand": [], "battlefield": [], "graveyard": [], "command_zone": []
})
zone_vec = encoder._encode_zones(player_dicts)
# 3. Playstyle one-hot (4 floats)
style_vec = encoder._encode_playstyle(playstyle)
state = np.concatenate([global_vec, zone_vec, style_vec])
expected = encoder.dim.total_state_dim # 6177
if state.shape[0] != expected:
raise ValueError(
f"State dim mismatch after precomputed encode: "
f"{state.shape[0]} != {expected}"
)
return state.astype(np.float32)
def _infer_from_vector(state_vec: np.ndarray, policy_service, temperature: float, greedy: bool) -> Dict:
"""Run policy network inference directly on a prebuilt state vector.
Mirrors the core of PolicyInferenceService.predict() but accepts an
ndarray instead of a snapshot dict, avoiding a second encode() call.
"""
import torch
from ml.config.scope import NUM_ACTIONS, IDX_TO_ACTION
t_start = time.time()
try:
state_tensor = torch.from_numpy(state_vec).unsqueeze(0).to(policy_service.device)
with torch.no_grad():
logits = policy_service.model(state_tensor)
if greedy:
action_idx = logits.argmax(dim=-1).item()
probs = torch.softmax(logits, dim=-1).cpu().numpy().flatten()
else:
scaled = logits / max(temperature, 0.01)
probs = torch.softmax(scaled, dim=-1).cpu().numpy().flatten()
action_idx = int(np.random.choice(NUM_ACTIONS, p=probs))
elapsed_ms = (time.time() - t_start) * 1000
action = IDX_TO_ACTION[action_idx]
confidence = float(probs[action_idx])
prob_map = {
IDX_TO_ACTION[i].value: round(float(probs[i]), 4)
for i in range(NUM_ACTIONS)
}
return {
"action": action.value,
"action_index": action_idx,
"confidence": round(confidence, 4),
"probabilities": prob_map,
"inference_ms": round(elapsed_ms, 2),
}
except Exception as e:
elapsed_ms = (time.time() - t_start) * 1000
logger.error("Direct inference error: %s", e)
return {"error": str(e), "inference_ms": round(elapsed_ms, 2)}
def _build_encoder_snapshot(req: DecideRequest) -> Dict:
"""Convert a DecideRequest to the flat snapshot dict expected by
StateEncoder.encode() — normalises GameSession field name aliases.
"""
players_out = []
for p in req.players:
bf_names = p.resolved_battlefield_names()
players_out.append({
"seat": p.seat,
"life": p.life,
"cmdr_dmg": sum(p.cmdr_dmg.values()) if p.cmdr_dmg else 0,
"mana": p.mana_available,
"cmdr_tax": p.resolved_cmdr_tax(),
"creatures": len(bf_names),
"lands": 0, # stub — wire from Forge in Step 4
"hand": p.resolved_hand(),
"battlefield": bf_names,
"graveyard": p.graveyard,
"command_zone": p.resolved_command_zone(),
})
return {
"turn": req.resolved_turn(),
"phase": _map_forge_phase(req.phase),
"active_seat": req.resolved_active_seat(),
"players": players_out,
"game_id": req.game_id,
"archetype": req.playstyle,
}
_FORGE_PHASE_MAP = {
"UNTAP": "main_1",
"UPKEEP": "main_1",
@@ -313,9 +551,15 @@ _FORGE_PHASE_MAP = {
"END": "end",
"END_STEP": "end",
"CLEANUP": "end",
# GameSession values already normalised
"main_1": "main_1",
"main_2": "main_2",
"combat": "combat",
"end": "end",
}
def _map_forge_phase(phase: str) -> str:
"""Map Forge phase string to ML GamePhase enum value."""
return _FORGE_PHASE_MAP.get(phase.upper(), "main_1")
"""Map Forge/GameSession phase string to ML GamePhase enum value."""
return _FORGE_PHASE_MAP.get(phase.upper() if phase else "MAIN1",
_FORGE_PHASE_MAP.get(phase, "main_1"))