magicciv/tools/p1-settle-analysis.py

#!/usr/bin/env python3
"""
p1-settle-analysis.py — Convergence gate analyzer for objective p1-29d.

Gate (dispatch + objective title): the trailing AI (P1) must be EITHER
eliminated OR stalled BEFORE T100 in 10/10 autoplay seeds. This is a
late-game *pacing* gate, not a survival/thrive gate — it asks whether P1's
structural fate has converged (stopped changing) by turn 100, so Wave-1
balance work runs against a stable regime.

"Settled" definition (rigorous, structural):
  P1 is settled at turn T_s = the LAST turn at which P1's structural state
  (city_count, tier_peak) changed from the previous turn. From T_s to game
  end, neither cities nor tier_peak move. Two settled regimes:
    - ELIMINATED: P1's city_count reaches 0 after having founded — the last
      structural change is the capital loss. settle_turn = that turn.
    - STALLED:    P1 alive with a frozen footprint (cities + tier_peak flat)
      through game end. settle_turn = last tier_peak/cities change.
  Within-city pop growth is intentionally NOT a destabilizer — a frozen
  single city whose pop ticks up does not change the strategic/pacing
  picture (no new cities, no new unit tier). We track pop/mil for context
  only.

A seed PASSES the convergence gate iff settle_turn <= T_GATE (default 100).

Reads game_<stamp>_seed<N>/turn_stats.jsonl under a results dir (the smoke/
subdir of a batch). stdlib only.

Usage:
    tools/p1-settle-analysis.py <results_dir> [--gate 100] [--player 1] [--json]
"""
from __future__ import annotations

import json
import re
import sys
from pathlib import Path

T_GATE_DEFAULT = 100
SEED_RE = re.compile(r"_seed(\d+)$")


def find_game_dirs(results_dir: Path) -> list[tuple[int, Path]]:
    by_seed: dict[int, Path] = {}
    for d in sorted(results_dir.iterdir()):
        if not d.is_dir():
            continue
        m = SEED_RE.search(d.name)
        if not m:
            continue
        if not (d / "turn_stats.jsonl").exists():
            continue
        seed = int(m.group(1))
        # Most recent stamp wins (lexicographic on dir name).
        if seed not in by_seed or d.name > by_seed[seed].name:
            by_seed[seed] = d
    return sorted(by_seed.items())


def analyze_seed(path: Path, pid: str) -> dict:
    lines = [json.loads(l) for l in (path / "turn_stats.jsonl").open() if l.strip()]
    if not lines:
        return {"error": "empty turn_stats"}
    last = lines[-1]
    end_turn = last["turn"]

    # Build P1 structural series: (turn, cities, tier_peak).
    series = []
    for d in lines:
        p = d["player_stats"].get(pid, {})
        series.append((d["turn"], p.get("cities", 0), p.get("tier_peak", 0),
                       p.get("pop", 0), p.get("mil", 0), p.get("cities_lost", 0)))

    founded = any(c > 0 for _, c, *_ in series)
    end_cities = series[-1][1]
    end_tp = series[-1][2]
    end_lost = series[-1][5]

    # Settle turn = last turn (cities, tier_peak) changed vs prior turn.
    settle_turn = series[0][0]
    prev = (series[0][1], series[0][2])
    for turn, cities, tp, *_ in series[1:]:
        if (cities, tp) != prev:
            settle_turn = turn
            prev = (cities, tp)

    # Classify regime.
    if founded and end_cities == 0:
        regime = "eliminated"
    elif end_cities >= 1:
        regime = "stalled" if end_cities == 1 else "alive-multi"
    else:
        regime = "never-founded"

    return {
        "end_turn": end_turn,
        "settle_turn": settle_turn,
        "regime": regime,
        "end_cities": end_cities,
        "end_tier_peak": end_tp,
        "end_cities_lost": end_lost,
        "end_pop": series[-1][3],
        "outcome": last.get("outcome"),
        "victory_type": last.get("victory_type"),
        "winner_index": last.get("winner_index"),
    }


def main(argv: list[str]) -> int:
    if len(argv) < 2:
        print(__doc__, file=sys.stderr)
        return 2
    results_dir = Path(argv[1])
    gate = T_GATE_DEFAULT
    pid = "1"
    as_json = False
    rest = argv[2:]
    i = 0
    while i < len(rest):
        a = rest[i]
        if a == "--gate":
            gate = int(rest[i + 1]); i += 2
        elif a == "--player":
            pid = rest[i + 1]; i += 2
        elif a == "--json":
            as_json = True; i += 1
        else:
            print(f"unknown arg: {a}", file=sys.stderr); return 2
    if not results_dir.is_dir():
        print(f"not a directory: {results_dir}", file=sys.stderr)
        return 2

    games = find_game_dirs(results_dir)
    if not games:
        print(f"no game_*_seed* dirs with turn_stats.jsonl under {results_dir}", file=sys.stderr)
        return 1

    rows = []
    for seed, path in games:
        r = analyze_seed(path, pid)
        r["seed"] = seed
        r["pass"] = ("error" not in r) and (r["settle_turn"] <= gate)
        rows.append(r)

    n_pass = sum(1 for r in rows if r.get("pass"))
    n = len(rows)

    if as_json:
        print(json.dumps({"gate": gate, "player": pid, "n_pass": n_pass,
                          "n": n, "rows": rows}, indent=2))
        return 0 if n_pass == n else 1

    print(f"P1 convergence gate (player {pid}): settled (eliminated OR stalled) by T{gate}")
    print(f"results: {results_dir}")
    print()
    print(f"{'seed':>4} {'settle':>6} {'regime':>13} {'end_t':>5} {'cities':>6} "
          f"{'tp':>3} {'lost':>4} {'pop':>4} {'verdict':>7}")
    for r in sorted(rows, key=lambda x: x["seed"]):
        if "error" in r:
            print(f"{r['seed']:>4}  ERROR: {r['error']}")
            continue
        verdict = "PASS" if r["pass"] else "FAIL"
        print(f"{r['seed']:>4} {r['settle_turn']:>6} {r['regime']:>13} "
              f"{r['end_turn']:>5} {r['end_cities']:>6} {r['end_tier_peak']:>3} "
              f"{r['end_cities_lost']:>4} {r['end_pop']:>4} {verdict:>7}")
    print()
    print(f"GATE {n_pass}/{n} seeds settled by T{gate}  "
          f"-> {'PASS (10/10 required)' if n_pass == n and n >= 10 else 'FAIL'}")
    return 0 if (n_pass == n and n >= 10) else 1


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