""" session.py — GameSession: one running game wrapping the CANON backend engine. Per-turn flow (claude-handoff/01-backend-port.md, implemented exactly): 1. apply staged chaos cards (each re-checked: energy is authoritative) 2. rules.legal_actions(g) 3. dispatcher picks — ONE model call (or the no-op fumble when the model is off; NEVER a scripted fallback policy) 4. simulation.apply + simulation.advance 5. build the RenderState (render_state.py) with this turn's notifications Model wiring: env MLP_MODELS (JSON name->base_url) + MLP_MODEL (default name). The model behind the dispatcher is the ONLY thing that varies by environment. """ from __future__ import annotations import json import os import random import sys _SERVER_DIR = os.path.dirname(os.path.abspath(__file__)) _BACKEND_DIR = os.path.abspath(os.path.join(_SERVER_DIR, "..", "..", "backend")) for _p in (_BACKEND_DIR, _SERVER_DIR): if _p not in sys.path: sys.path.insert(0, _p) import engine import rules import simulation import render from agents import LLMDispatcher from llm import LLM import render_state _DEFAULT_MODELS = '{"qwen3-8b": "http://localhost:8092", "nemotron": "http://localhost:8093"}' MODELS: dict[str, str] = json.loads(os.environ.get("MLP_MODELS", _DEFAULT_MODELS)) # nemotron default per user (2026-06-12): "it handily defeated me" — and it's the <=4B # tiny-model-award hero. Override with MLP_MODEL=qwen3-8b. DEFAULT_MODEL: str = os.environ.get("MLP_MODEL", "nemotron") # Dispatcher backend: "openai_compat" (local llama.cpp, MODELS values are base URLs) or # "transformers" (HF ZeroGPU Space, MODELS values are HF model ids → gpu_llm). On the Space set # MLP_LLM_BACKEND=transformers and MLP_MODELS to nemotron-only (no qwen). LLM_BACKEND: str = os.environ.get("MLP_LLM_BACKEND", "openai_compat") # The dispatcher-offline fumble: no LLM call, zero actions. Cosmetic telemetry only. OFFLINE_AI = {"priority": "(dispatcher offline)", "actions": [], "announcement": "", "noop": True, "confidence": None} class GameSession: """One seeded game + its dispatcher + the chaos cards staged for this round.""" def __init__(self, seed: int | None = None, model_name: str | None = None, model_on: bool = True): self.seed = seed if seed is not None else random.randint(0, 999_999) self.g = engine.new_game(self.seed) self.prompts = render.load_prompts() self.model_on = model_on self.model_name = "" self.llm: LLM | None = None self.disp: LLMDispatcher | None = None self.set_model(model_name or DEFAULT_MODEL) self.staged: list[tuple[str, str]] = [] # (card, location) queued this round self.last_state: dict = render_state.build_state(self.g) # ----------------------------------------------------------------- model switch def set_model(self, name: str) -> None: if name not in MODELS: raise ValueError(f"unknown model {name!r}; have {sorted(MODELS)}") self.model_name = name target = MODELS[name] # base URL (openai_compat) | HF id (transformers) if LLM_BACKEND == "transformers": self.llm = LLM("dispatcher", "transformers", target, max_tokens=512) else: self.llm = LLM("dispatcher", "openai_compat", name, target, max_tokens=512) self.disp = LLMDispatcher(self.llm, self.prompts) # ----------------------------------------------------------------- chaos staging def play_card(self, card: str, loc: str) -> tuple[bool, str]: if self.g.over: return False, "game over" if len(self.staged) >= 1: # one card per round (peak no longer grants 2) return False, "card limit this round" if not rules.card_available(self.g, card): return False, "unavailable" if not rules.station_free(self.g, loc): return False, "station busy" if engine.card_cost(self.g, card) > self.g.energy: return False, "no energy" self.staged.append((card, loc)) return True, "ok" # ----------------------------------------------------------------- the turn def next_turn(self) -> dict: if self.g.over: # terminal: re-serve the final state return self.last_state g = self.g pre_turn_ids = {i.id for i in g.incidents} # 1) apply staged chaos (re-check each — energy/legality are authoritative here) notes: list[dict] = [] applied: list[dict] = [] for card, loc in self.staged: cost = engine.card_cost(g, card) if (cost <= g.energy and rules.card_available(g, card) and rules.station_free(g, loc)): g.energy -= cost rules.apply_chaos(g, card, loc) applied.append({"card": card, "location": loc}) notes.append({"kind": "chaos", "text": f"You played {card} @ {loc}", "station": loc}) self.staged = [] chaos_ids = {i.id for i in g.incidents} - pre_turn_ids # 2) dispatcher picks (ONE model call — or the offline fumble) A = rules.legal_actions(g) if self.model_on: chosen, tel = self.disp.act(g, A) ai = {"priority": tel.get("priority", ""), "announcement": tel.get("announcement", ""), "noop": bool(tel.get("noop", not chosen)), "confidence": tel.get("confidence")} else: chosen, tel = [], None ai = dict(OFFLINE_AI) by = {a["action_id"]: a for a in A} ai["actions"] = [render.describe_action(by[c]) for c in chosen if c in by] # 3) apply + advance announced, police_at = simulation.apply(g, A, chosen) pre_advance = {i.id: (i.itype, i.location) for i in g.incidents} phase_before = g.phase simulation.advance(g, announced, police_at) # 4) notifications for this turn (ordered: chaos, ai, incident, resolve, peak, end) notes.append({"kind": "ai", "text": f"AI: {ai['priority'] or '(no response)'} " f"[{len(ai['actions'])} action(s)]", "station": None}) after_ids = {i.id for i in g.incidents} for i in g.incidents: # born this turn beyond the chaos cards if i.id not in pre_advance and i.id not in chaos_ids: notes.append({"kind": "incident", "text": f"Incident: {i.itype} @ {i.location}", "station": i.location}) for iid, (itype, iloc) in pre_advance.items(): if iid not in after_ids: notes.append({"kind": "resolve", "text": f"Cleared: {itype} @ {iloc}", "station": iloc}) if phase_before != "peak" and g.phase == "peak": notes.append({"kind": "peak", "text": "PEAK RUSH begins — the escalator is on", "station": None}) if g.over: notes.append({"kind": "win" if g.won else "loss", "text": g.reason, "station": None}) self.last_state = render_state.build_state(g, ai=ai, chaos_last=applied, notifications=notes) return self.last_state