magicciv/tools/p1-survival-score.py

#!/usr/bin/env python3
"""p1-survival-score.py — score a 10-seed autoplay batch for objective p1-29d.

p1-29d carries two contradictory gate definitions (see the objective file):

  SURVIVAL  (body Acceptance bullet): >=7/10 *alive-aware* seeds where the
            trailing AI (P1, slot 1) ends alive with tier_peak >= 2.
  CONVERGENCE (title + dispatch): the trailing AI is eliminated OR stalled
            *before T100* in 10/10 seeds (stable late-game pacing).

Rather than pick, this scorer prints the raw per-seed table and BOTH gate
verdicts as derived views, so a human/orchestrator decides which is canonical.

The convergence gate needs P1's *elimination turn*, not the game-end turn —
these are multi-clan domination games, so P1 can die well before the winner
achieves domination. We therefore scan the full turn_stats stream for the
first turn P1's city count hits 0 (and stays 0), not just the final line.

Usage:
    tools/p1-survival-score.py <results_dir>   # dir containing game_*_seedN/

stdlib only.
"""
from __future__ import annotations

import json
import re
import sys
from pathlib import Path

TRAILING_SLOT = "1"  # P1 = slot 1 is the designated trailing AI per p1-29d
LEADER_SLOT = "0"
T100 = 100


def _player(stats: dict, slot: str) -> dict | None:
    """player_stats is an object keyed by player-index string."""
    if not isinstance(stats, dict):
        return None
    return stats.get(slot)


def scan_seed(game_dir: Path) -> dict | None:
    ts = game_dir / "turn_stats.jsonl"
    if not ts.exists():
        return None
    lines = [ln for ln in ts.read_text().splitlines() if ln.strip()]
    if not lines:
        return None

    m = re.search(r"seed(\d+)", game_dir.name)
    seed = int(m.group(1)) if m else -1

    last = json.loads(lines[-1])
    pstats = last.get("player_stats", {})
    p1 = _player(pstats, TRAILING_SLOT) or {}
    p0 = _player(pstats, LEADER_SLOT) or {}

    final_turn = last.get("turn")
    outcome = last.get("outcome")
    victory_type = last.get("victory_type")
    winner_index = last.get("winner_index")

    p1_tp = p1.get("tier_peak", 0) or 0
    p0_tp = p0.get("tier_peak", 0) or 0
    p1_cities_end = p1.get("cities", 0) or 0
    p1_lost = p1.get("cities_lost", 0) or 0
    p1_alive = p1_cities_end >= 1

    # P1 elimination turn: first turn its cities hit 0 and never recover.
    p1_elim_turn = None
    for ln in lines:
        try:
            rec = json.loads(ln)
        except json.JSONDecodeError:
            continue
        p1t = _player(rec.get("player_stats", {}), TRAILING_SLOT)
        if p1t is None:
            continue
        turn = rec.get("turn")
        cities = p1t.get("cities", 0) or 0
        if cities == 0 and not p1t.get("cities_founded_pending"):
            # candidate elimination; require it to be terminal (stays 0)
            if p1_elim_turn is None:
                p1_elim_turn = turn
        else:
            p1_elim_turn = None  # recovered → reset
    # If alive at end, no elimination
    if p1_alive:
        p1_elim_turn = None

    return {
        "seed": seed,
        "final_turn": final_turn,
        "outcome": outcome,
        "victory_type": victory_type,
        "winner_index": winner_index,
        "p0_tp": p0_tp,
        "p1_tp": p1_tp,
        "p1_cities_end": p1_cities_end,
        "p1_cities_lost": p1_lost,
        "p1_alive": p1_alive,
        "p1_elim_turn": p1_elim_turn,
        "p1_mil": p1.get("mil", 0),
        "p1_kills": p1.get("kills", 0),
        "p1_units_lost": p1.get("units_lost", 0),
        "p1_pop": p1.get("pop", 0),
    }


def main(argv: list[str]) -> int:
    if len(argv) != 2:
        print(__doc__)
        return 2
    root = Path(argv[1])
    if not root.exists():
        print(f"no such dir: {root}", file=sys.stderr)
        return 2

    def _seed_of(d: Path) -> int:
        m = re.search(r"seed(\d+)", d.name)
        return int(m.group(1)) if m else -1

    game_dirs = sorted(
        {p.parent for p in root.rglob("turn_stats.jsonl")}, key=_seed_of)
    rows = [r for d in game_dirs if (r := scan_seed(d))]
    if not rows:
        print(f"no parseable seeds under {root}", file=sys.stderr)
        return 1
    rows.sort(key=lambda r: r["seed"])

    # ── Raw per-seed table ────────────────────────────────────────────
    print(f"\n=== p1-29d batch scoring: {root} ===")
    print(f"{len(rows)} seeds\n")
    hdr = ("seed", "endT", "outcome", "P0tp", "P1tp", "P1cit", "P1lost",
           "P1alive", "P1elimT", "P1mil", "P1kills")
    print("{:>4} {:>5} {:>11} {:>4} {:>4} {:>5} {:>6} {:>7} {:>7} {:>5} {:>6}".format(*hdr))
    for r in rows:
        print("{:>4} {:>5} {:>11} {:>4} {:>4} {:>5} {:>6} {:>7} {:>7} {:>5} {:>6}".format(
            r["seed"], str(r["final_turn"]), str(r["outcome"])[:11],
            r["p0_tp"], r["p1_tp"], r["p1_cities_end"], r["p1_cities_lost"],
            "yes" if r["p1_alive"] else "no",
            str(r["p1_elim_turn"]) if r["p1_elim_turn"] is not None else "-",
            r["p1_mil"], r["p1_kills"]))

    n = len(rows)

    # ── Gate A: SURVIVAL (file Acceptance) ────────────────────────────
    # alive-aware seed with P1 tier_peak >= 2 (P1 ends alive AND developed)
    survival_pass = [r for r in rows if r["p1_alive"] and r["p1_tp"] >= 2]
    # also report the stricter both-developed alive-aware count (p1-29a def)
    both_dev = [r for r in rows if r["p0_tp"] >= 2 and r["p1_tp"] >= 2 and r["p1_alive"]]
    print(f"\n--- GATE A (SURVIVAL, file): P1 alive AND tier_peak>=2 ---")
    print(f"  {len(survival_pass)}/{n} seeds  (need >=7)  -> "
          f"{'PASS' if len(survival_pass) >= 7 else 'FAIL'}")
    print(f"  (stricter both-developed alive-aware: {len(both_dev)}/{n})")
    print(f"  seeds passing: {[r['seed'] for r in survival_pass]}")

    # ── Gate B: CONVERGENCE (title/dispatch) ──────────────────────────
    # P1 eliminated before T100 OR stalled (alive but never developed past tp1)
    conv_rows = []
    for r in rows:
        elim_before_t100 = (not r["p1_alive"]) and (
            r["p1_elim_turn"] is not None and r["p1_elim_turn"] <= T100)
        # fallback: if elim turn unknown but dead and game ended by T100
        if (not r["p1_alive"]) and r["p1_elim_turn"] is None \
                and isinstance(r["final_turn"], int) and r["final_turn"] <= T100:
            elim_before_t100 = True
        stalled = r["p1_alive"] and r["p1_tp"] <= 1
        converged = elim_before_t100 or stalled
        reason = ("elim<=T100" if elim_before_t100
                  else "stalled(alive,tp<=1)" if stalled
                  else ("elim>T100" if not r["p1_alive"] else "alive,developing"))
        conv_rows.append((r["seed"], converged, reason))
    conv_pass = [c for c in conv_rows if c[1]]
    print(f"\n--- GATE B (CONVERGENCE, title/dispatch): "
          f"P1 eliminated<=T100 OR stalled, 10/10 ---")
    print(f"  {len(conv_pass)}/{n} seeds  (need {n}/{n})  -> "
          f"{'PASS' if len(conv_pass) == n else 'FAIL'}")
    for seed, ok, reason in conv_rows:
        print(f"    s{seed}: {'OK ' if ok else 'NO '} ({reason})")

    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))