""" eval.py — headless evaluation harness (the balance tool). Plays seeded games with a swappable dispatcher vs SmartChaos (the benchmark adversary), captures per-turn telemetry, measures doctrine-adherence vs the oracle, and reports win-rate + the 4-track stats. Calibrate difficulty here against the fixed-skill bot — NOT against an expert human, who learns across replays while the model does not. python3 eval.py --mode baseline # the oracle (no model) — sanity ceiling python3 eval.py --mode random --games 5 # random picker — loss should be reachable python3 eval.py --mode llm_dispatch --games 8 \ --dispatch-name qwen3-8b --dispatch-url http://localhost:8092 """ from __future__ import annotations import argparse, json, os, random, statistics, sys, time import engine import rules import simulation import render from doctrine import doctrine from agents import LLMDispatcher from chaos_players import ChaosPlayer, SmartChaos from llm import LLM, ping RUNS_DIR = os.path.join(os.path.dirname(__file__), "runs") os.makedirs(RUNS_DIR, exist_ok=True) def jaccard(a, b): a, b = set(a), set(b) if not a and not b: return 1.0 u = a | b return len(a & b) / len(u) if u else 1.0 def _tel(source, chosen, **extra): tel = {"source": source, "json_valid": True, "repair_used": False, "noop": len(chosen) == 0, "illegal_ids_dropped": 0, "budget_violation": False, "n_actions": len(chosen), "latency_ms": 0.0, "priority": "", "announcement": "", "confidence": None} tel.update(extra) return tel class TrackOracle: name = "doctrine" def reset(self, seed): pass def act(self, g, A): chosen, prio = doctrine(g, A) return chosen, _tel("doctrine", chosen, priority=prio) class RandomDispatcher: name = "random" def __init__(self): self.r = random.Random(0) def reset(self, seed): self.r = random.Random(seed * 13 + 5) def act(self, g, A): from agents import enforce_budget by = {a["action_id"]: a for a in A} ids = list(by); self.r.shuffle(ids) chosen, violated = enforce_budget(ids, by) return chosen, _tel("random", chosen, budget_violation=violated) def make_chaos(mode): return SmartChaos() if mode == "smart" else ChaosPlayer(mode=mode) def run_game(seed, dispatcher, chaos, oracle, max_turns=None): g = engine.new_game(seed) dispatcher.reset(seed); chaos.reset(seed) turns = [] while not g.over: lc = rules.legal_cards(g) plays, ctel = chaos.play(g, lc) for card, loc in plays: if engine.CARDS[card] <= g.energy and rules.card_available(g, card) and rules.station_free(g, loc): g.energy -= engine.CARDS[card] rules.apply_chaos(g, card, loc) A = rules.legal_actions(g) chosen, dtel = dispatcher.act(g, A) adh = jaccard(chosen, oracle.act(g, A)[0]) if oracle is not None else None announced, police_at = simulation.apply(g, A, chosen) col0, xo0 = g.collisions_avoided, g.crossover_blocks simulation.advance(g, announced, police_at) turns.append({ "t": g.turn, "phase": g.phase, "anger": round(g.anger, 1), "safety": round(g.safety, 1), "pressure": round(g.pressure, 1), "adherence": adh, "stuck": sum(1 for t in g.trains if t.stuck_turns > 0), "col": g.collisions_avoided - col0, "xover": g.crossover_blocks - xo0, "disp": {k: dtel.get(k) for k in ("json_valid", "repair_used", "noop", "illegal_ids_dropped", "budget_violation", "n_actions", "latency_ms")}, "chaos": {k: ctel.get(k) for k in ("n_plays", "energy_spent")}, }) if max_turns and g.turn >= max_turns and not g.over: g.over, g.won, g.reason = True, True, f"capped at {max_turns}"; break return g, turns def summarize(label, records): n = len(records) wins = sum(r["won"] for r in records) surv = [r["survival_turn"] for r in records] causes = {} for r in records: if not r["won"]: causes[r["loss_cause"]] = causes.get(r["loss_cause"], 0) + 1 flat = [t for r in records for t in r["turns"]] adh = [t["adherence"] for t in flat if t["adherence"] is not None] lat = [t["disp"]["latency_ms"] for t in flat if t["disp"]["latency_ms"]] print(f"\n===== {label} ({n} games) =====") print(f" win-rate : {100*wins/n:.0f}% ({wins}/{n})") print(f" median survival : {int(statistics.median(surv))}t (range {min(surv)}–{max(surv)})") print(f" loss causes : {causes or '—'}") print(f" collisions/game : {sum(r['collisions_avoided'] for r in records)/n:.1f} " f"crossover blocks/game: {sum(r['crossover_blocks'] for r in records)/n:.1f}") print(f" adherence vs oracle: {statistics.mean(adh):.2f}" if adh else " adherence: —") print(f" json-valid / noop : {100*statistics.mean([t['disp']['json_valid'] for t in flat]):.0f}%" f" / {100*statistics.mean([t['disp']['noop'] for t in flat]):.0f}%") print(f" mean actions/turn : {statistics.mean([t['disp']['n_actions'] for t in flat]):.2f}") if lat: print(f" mean latency/turn : {statistics.mean(lat)/1000:.1f}s") def main(): ap = argparse.ArgumentParser() ap.add_argument("--mode", choices=["baseline", "random", "llm_dispatch"], default="baseline") ap.add_argument("--games", type=int, default=3) ap.add_argument("--seed-start", type=int, default=0) ap.add_argument("--max-turns", type=int, default=0) ap.add_argument("--chaos-mode", default="smart", choices=["random", "adversarial", "smart"]) ap.add_argument("--temperature", type=float, default=0.4) ap.add_argument("--max-tokens", type=int, default=512) ap.add_argument("--turn-timeout", type=int, default=120) ap.add_argument("--grammar", action="store_true") ap.add_argument("--tag", default="") ap.add_argument("--dispatch-name", default="dispatch") ap.add_argument("--dispatch-model", default="local") ap.add_argument("--dispatch-url", default="http://localhost:8092") ap.add_argument("--dispatch-backend", default="openai_compat") args = ap.parse_args() args.max_turns = args.max_turns or None oracle = TrackOracle() chaos = make_chaos(args.chaos_mode) if args.mode == "baseline": dispatcher, label, oracle = TrackOracle(), "doctrine (oracle)", None elif args.mode == "random": dispatcher, label = RandomDispatcher(), "random picker" else: schema = None if args.grammar: with open(os.path.join(os.path.dirname(__file__), "prompts", "output_schema.json"), encoding="utf-8") as f: schema = json.load(f) url = args.dispatch_url if not args.dispatch_url.startswith(":") else "http://localhost" + args.dispatch_url llm = LLM(args.dispatch_name, args.dispatch_backend, args.dispatch_model, url, api_key="sk-local", temperature=args.temperature, max_tokens=args.max_tokens, timeout=args.turn_timeout, json_schema=schema) if args.dispatch_backend == "openai_compat" and not ping(url): sys.exit(f"No server at {url}. Start: llama-server -m --port -ngl 99 --reasoning off --jinja") dispatcher = LLMDispatcher(llm, render.load_prompts()) label = f"{args.dispatch_name}{' +grammar' if args.grammar else ''}" tag = args.tag or args.mode out_path = os.path.join(RUNS_DIR, f"eval__{label.split()[0].replace('/', '-')}__{tag}.jsonl") seeds = list(range(args.seed_start, args.seed_start + args.games)) print(f"[4-TRACK eval] dispatcher={label} chaos={args.chaos_mode} " f"trains={engine.B['n_trains']} games={seeds}") records, t0 = [], time.perf_counter() with open(out_path, "w", encoding="utf-8") as f: for seed in seeds: g, turns = run_game(seed, dispatcher, chaos, oracle, args.max_turns) rec = {"seed": seed, "won": g.won, "survival_turn": g.turn, "loss_cause": None if g.won else g.reason, "final": {"anger": round(g.anger, 1), "safety": round(g.safety, 1), "score": round(g.score, 1)}, "collisions_avoided": g.collisions_avoided, "crossover_blocks": g.crossover_blocks, "turns": turns} records.append(rec) f.write(json.dumps(rec) + "\n"); f.flush() print(f" seed {seed:>2}: {'WIN ' if g.won else 'LOSS'} @T{g.turn:<2} ({(g.reason or '')[:24]:<24}) " f"anger={g.anger:3.0f} safety={g.safety:3.0f} col={g.collisions_avoided:>3}") summarize(label, records) print(f"\n ({time.perf_counter()-t0:.0f}s) -> {out_path}") if __name__ == "__main__": main()