| """ |
| 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 <gguf> --port <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() |
|
|
