"""Run the post-generation-4 curriculum from the promoted Stage-3 policy. This is intentionally separate from curriculum.py/curriculum_state.json: generation 4 is a completed lineage and its final checkpoint is generation 5's fixed foundation. Stages 4-6 add one difficulty at a time: 4 handling -- upright, nose-led low-altitude movement 5 intercepts -- useful moving-ball aerial interceptions 6 league -- robustness against a pool of frozen historical styles Each stage resumes from its passing predecessor, exports through the normal run_training.sh parity check, records tail telemetry, and runs a paired 100-episode regression evaluation. State is restart-safe in generation5_state.json. """ from __future__ import annotations import argparse import json import pathlib import subprocess import sys from datetime import datetime from tensorboard.backend.event_processing.event_accumulator import EventAccumulator TRAINING_DIR = pathlib.Path(__file__).resolve().parent REPO_ROOT = TRAINING_DIR.parent STATE_PATH = TRAINING_DIR / "generation5_state.json" EVAL_HISTORY_PATH = TRAINING_DIR / "eval_history.json" FOUNDATION_EXPERIMENT = "20260806-1939-curric-s3-gauntlet" FOUNDATION_CHECKPOINT = TRAINING_DIR / "checkpoints" / FOUNDATION_EXPERIMENT / "final.zip" FOUNDATION_EXPORT = REPO_ROOT / "Game" / "bots" / f"{FOUNDATION_EXPERIMENT}.json" PROMOTED_EASY = REPO_ROOT / "Game" / "bots" / "promoted" / "easy.json" MAX_RETRIES = 2 EVAL_EPISODES = 100 REGRESSION_MARGIN = 0.15 STANDING_ARGS = ["--ent-coef", "0.01", "--entropy-floor"] # Scoring/ball-direction shaping inherited from generation 4. Handling # replaces half the orientation-agnostic closing reward and all generic speed # reward with nose-led ground approach, while keeping global tilt pressure # small enough for flight. The first three Stage-4 attempts (2026-08-08/09) # plateaued with upright_fraction/forward_motion_fraction flat at ~0.22-0.26 # against 0.45/0.25 floors for 120M cumulative timesteps: ground_tilt_penalty # at 0.003 only cost a fully-sideways episode ~2.7 reward, trivial next to a # goal (80) or a touch (0.7). ground_tilt_penalty is raised ~17x to 0.05 (a # full sideways episode now costs ~45, comparable to a goal) and # non_forward_penalty is a new term (ship_ai_controller.gd) directly costing # sideways/reverse planar velocity near the floor, independent of the ball, # since nothing previously penalized that at all. Both are floor-proximity # penalties only, with nothing equivalent above GROUND_HANDLING_HEIGHT — on # its own that risks teaching "avoid the floor" instead of "handle well on # it", worsening Stage 3's already-airborne-heavy baseline. grounded_upright_ # reward is the positive counterpart: a bonus for genuine floor contact # (not just low altitude) while upright, so grounding well is the locally # profitable choice rather than merely the least-punished one. HANDLING_REWARD_FLAGS = [ "--velocity-to-ball-weight", "0.04", "--forward-velocity-to-ball-weight", "0.06", "--ball-distance-penalty", "0.01", "--ball-touch-reward", "0.7", "--ball-velocity-to-goal-weight", "0.06", "--goal-reward", "80", "--speed-reward-weight", "0.0", "--tilt-penalty", "0.0002", "--ground-tilt-penalty", "0.05", "--non-forward-penalty", "0.04", "--grounded-upright-reward", "0.015", ] STAGES = [ { "number": 4, "name": "handling", "timesteps": 40_000_000, "flags": [ "--opponent-mode", "self_play", "--kickoff-chance", "0.15", "--near-goal-chance", "0.25", "--air-drill-chance", "0.20", "--air-intercept-chance", "0.0", *HANDLING_REWARD_FLAGS, ], # Conservative catastrophe floors, not claims of mastery. Tail values # are recorded in state so later thresholds can be based on evidence. "telemetry_floors": { "rollout/goal_rate": 0.80, "rollout/upright_fraction": 0.45, "rollout/forward_motion_fraction": 0.25, }, # At least 80% of the paired candidate-vs-Stage-3 episodes must end # in a goal. This is separate from win-rate regression: a draw-heavy # handling policy must not advance merely because neither bot won. "evaluation_goal_rate_floor": 0.80, # The paired side swap also measures physical spawn/team bias. This # catches a broken team-frame action mapping even when model A's # aggregate result looks balanced because it plays both sides. "physical_side_imbalance_ceiling": 0.20, }, { "number": 5, "name": "intercepts", "timesteps": 60_000_000, "flags": [ "--opponent-mode", "self_play", "--kickoff-chance", "0.10", "--near-goal-chance", "0.20", "--air-drill-chance", "0.10", "--air-intercept-chance", "0.45", *HANDLING_REWARD_FLAGS, ], "telemetry_floors": { "rollout/goal_rate": 0.75, "rollout/upright_fraction": 0.40, "rollout/forward_motion_fraction": 0.20, "rollout/productive_air_touch_fraction": 0.005, }, "evaluation_goal_rate_floor": 0.75, "physical_side_imbalance_ceiling": 0.20, }, { "number": 6, "name": "league", "timesteps": 100_000_000, "flags": [ "--opponent-mode", "league", "--kickoff-chance", "0.15", "--near-goal-chance", "0.25", "--air-drill-chance", "0.15", "--air-intercept-chance", "0.25", *HANDLING_REWARD_FLAGS, ], "telemetry_floors": { "rollout/goal_rate": 0.70, "rollout/upright_fraction": 0.35, "rollout/forward_motion_fraction": 0.18, "rollout/productive_air_touch_fraction": 0.003, }, "evaluation_goal_rate_floor": 0.70, "physical_side_imbalance_ceiling": 0.20, "league_pool": True, }, ] def fresh_state() -> dict: return {"stage_index": 0, "attempt": 0, "status": "in_progress", "log": []} def load_state() -> dict: return json.loads(STATE_PATH.read_text()) if STATE_PATH.exists() else fresh_state() def save_state(state: dict) -> None: STATE_PATH.write_text(json.dumps(state, indent=2) + "\n") def passing_entry(state: dict, stage_index: int) -> dict: for entry in state["log"]: if entry["stage_index"] == stage_index and entry["decision"] == "pass": return entry raise RuntimeError(f"No passing generation-5 stage index {stage_index}") def previous_attempt_entry(state: dict, stage_index: int, attempt: int) -> dict: for entry in reversed(state["log"]): if entry["stage_index"] == stage_index and entry["attempt"] == attempt - 1: return entry raise RuntimeError(f"No previous attempt for stage index {stage_index}, attempt {attempt}") def resume_checkpoint(state: dict, stage_index: int, attempt: int, foundation: pathlib.Path) -> pathlib.Path: if attempt > 0: exp = previous_attempt_entry(state, stage_index, attempt)["experiment"] return TRAINING_DIR / "checkpoints" / exp / "final.zip" if stage_index == 0: return foundation exp = passing_entry(state, stage_index - 1)["experiment"] return TRAINING_DIR / "checkpoints" / exp / "final.zip" def reference_export(state: dict, stage_index: int) -> pathlib.Path: if stage_index == 0: return PROMOTED_EASY exp = passing_entry(state, stage_index - 1)["experiment"] return REPO_ROOT / "Game" / "bots" / f"{exp}.json" def league_pool(state: dict) -> list[pathlib.Path]: stage4 = passing_entry(state, 0)["experiment"] stage5 = passing_entry(state, 1)["experiment"] return [ FOUNDATION_EXPORT, REPO_ROOT / "Game" / "bots" / f"{stage4}.json", REPO_ROOT / "Game" / "bots" / f"{stage5}.json", ] def telemetry_tail(experiment: str, count: int = 500) -> dict[str, float]: log_dirs = sorted((TRAINING_DIR / "logs").glob(f"{experiment}_*")) if not log_dirs: return {} event_files = sorted(log_dirs[-1].glob("events.out.tfevents.*")) if not event_files: return {} accumulator = EventAccumulator(str(event_files[-1]), size_guidance={"scalars": 0}) accumulator.Reload() result = {} for tag in accumulator.Tags().get("scalars", []): if not tag.startswith("rollout/"): continue values = [point.value for point in accumulator.Scalars(tag)[-count:]] if values: result[tag] = sum(values) / len(values) return result def telemetry_passes(stage: dict, telemetry: dict[str, float]) -> tuple[bool, list[str]]: failures = [] for metric, floor in stage.get("telemetry_floors", {}).items(): value = telemetry.get(metric) if value is None: failures.append(f"{metric} missing") elif value < floor: failures.append(f"{metric}={value:.4f} < {floor:.4f}") return not failures, failures def run_training(state: dict, stage_index: int, attempt: int, args) -> str: stage = STAGES[stage_index] suffix = "" if attempt == 0 else f"-retry{attempt}" experiment = f"{datetime.now().strftime('%Y%m%d-%H%M')}-gen5-s{stage['number']}-{stage['name']}{suffix}" resume = resume_checkpoint(state, stage_index, attempt, pathlib.Path(args.foundation_checkpoint)) if not resume.exists(): raise FileNotFoundError(f"Resume checkpoint not found: {resume}") cmd = [ "./run_training.sh", experiment, "--timesteps", str(stage["timesteps"]), "--n-parallel", str(args.n_parallel), "--speedup", str(args.speedup), "--resume", str(resume), *STANDING_ARGS, *stage["flags"], ] if stage.get("league_pool"): pool = league_pool(state) missing = [str(path) for path in pool if not path.exists()] if missing: raise FileNotFoundError(f"League pool models missing: {missing}") cmd += ["--opponent-pool", ",".join(str(path) for path in pool)] print(f"\n=== Generation 5 Stage {stage['number']} {stage['name']} attempt {attempt + 1} ===") print(" ".join(cmd)) if args.dry_run: return experiment subprocess.run(cmd, cwd=TRAINING_DIR, check=True) return experiment def evaluate(experiment: str, reference: pathlib.Path, args) -> dict: candidate = REPO_ROOT / "Game" / "bots" / f"{experiment}.json" cmd = [ ".venv/bin/python", "evaluate.py", str(candidate), str(reference), "--episodes", str(EVAL_EPISODES), "--speedup", str(args.speedup), ] if args.godot_bin: cmd += ["--godot_bin", args.godot_bin] subprocess.run(cmd, cwd=TRAINING_DIR, check=True) return json.loads(EVAL_HISTORY_PATH.read_text())[-1] def match_passes(record: dict) -> bool: candidate = record["wins_a"] / record["episodes"] reference = record["wins_b"] / record["episodes"] return reference - candidate < REGRESSION_MARGIN def evaluation_goal_rate(record: dict) -> float: """Fraction of paired evaluation episodes that ended in either bot scoring.""" return (record["wins_a"] + record["wins_b"]) / record["episodes"] def physical_side_imbalance(record: dict) -> float: """Absolute physical-team win margin as a fraction of all episodes.""" physical = record["physical_team_wins"] return abs(physical["team_0"] - physical["team_1"]) / record["episodes"] def commit_progress(experiment: str) -> None: subprocess.run(["git", "add", STATE_PATH.name, EVAL_HISTORY_PATH.name], cwd=TRAINING_DIR, check=True) if subprocess.run(["git", "diff", "--cached", "--quiet"], cwd=TRAINING_DIR).returncode == 0: return subprocess.run( ["git", "commit", "-m", f"chore(training): generation 5 progress after {experiment}"], cwd=TRAINING_DIR, check=True, ) subprocess.run(["git", "push"], cwd=TRAINING_DIR, check=True) def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--n-parallel", type=int, default=14) parser.add_argument("--speedup", type=int, default=16) parser.add_argument("--godot-bin", default=None, help="Godot binary for post-stage evaluation") parser.add_argument("--foundation-checkpoint", default=str(FOUNDATION_CHECKPOINT)) parser.add_argument("--force-retry", action="store_true") parser.add_argument("--skip-to-next-stage", action="store_true") parser.add_argument("--dry-run", action="store_true", help="Print the next run command without executing it") args = parser.parse_args() state = load_state() if state["status"] == "done": print("Generation 5 is already complete.") return if state["status"] == "blocked": if args.force_retry: state["attempt"] += 1 state["status"] = "in_progress" save_state(state) elif args.skip_to_next_stage: state["stage_index"] += 1 state["attempt"] = 0 state["status"] = "in_progress" save_state(state) else: stage = STAGES[state["stage_index"]] print(f"BLOCKED at Stage {stage['number']} {stage['name']}; inspect {STATE_PATH.name}.") print("Use --force-retry after adjustment or --skip-to-next-stage after human review.") sys.exit(1) while state["stage_index"] < len(STAGES): stage_index = state["stage_index"] attempt = state["attempt"] stage = STAGES[stage_index] experiment = run_training(state, stage_index, attempt, args) if args.dry_run: return telemetry = telemetry_tail(experiment) telemetry_ok, telemetry_failures = telemetry_passes(stage, telemetry) references = [reference_export(state, stage_index)] if stage.get("league_pool"): references.extend(league_pool(state)) # Preserve order while avoiding a duplicate Stage-5 evaluation in # the league stage (its predecessor is also in the pool). references = list(dict.fromkeys(references)) records = [evaluate(experiment, reference, args) for reference in references] match_ok = all(match_passes(record) for record in records) evaluation_goal_floor = stage.get("evaluation_goal_rate_floor", 0.0) evaluation_goal_failures = [ f"{pathlib.Path(record['model_b']).name}: goal_rate={evaluation_goal_rate(record):.3f} " f"< {evaluation_goal_floor:.3f}" for record in records if evaluation_goal_rate(record) < evaluation_goal_floor ] scoring_ok = not evaluation_goal_failures side_imbalance_ceiling = stage.get("physical_side_imbalance_ceiling", 1.0) side_balance_failures = [ f"{pathlib.Path(record['model_b']).name}: physical_side_imbalance=" f"{physical_side_imbalance(record):.3f} > {side_imbalance_ceiling:.3f}" for record in records if physical_side_imbalance(record) > side_imbalance_ceiling ] side_balance_ok = not side_balance_failures decision = "pass" if match_ok and telemetry_ok else "fail" if not scoring_ok or not side_balance_ok: decision = "fail" entry = { "stage_index": stage_index, "stage_number": stage["number"], "stage_name": stage["name"], "experiment": experiment, "attempt": attempt, "telemetry_tail": telemetry, "telemetry_failures": telemetry_failures, "evaluation_goal_failures": evaluation_goal_failures, "side_balance_failures": side_balance_failures, "eval": records[0], "evals": records, "decision": decision, } state["log"].append(entry) print( f"{experiment}: match={'pass' if match_ok else 'fail'}, " f"scoring={'pass' if scoring_ok else 'fail'}, " f"side_balance={'pass' if side_balance_ok else 'fail'}, " f"telemetry={'pass' if telemetry_ok else 'fail'} -> {decision}" ) for failure in telemetry_failures: print(f" {failure}") for failure in evaluation_goal_failures: print(f" {failure}") for failure in side_balance_failures: print(f" {failure}") if decision == "pass": state["stage_index"] += 1 state["attempt"] = 0 save_state(state) commit_progress(experiment) continue if attempt >= MAX_RETRIES: state["status"] = "blocked" save_state(state) commit_progress(experiment) print(f"BLOCKED after {MAX_RETRIES + 1} attempts at Stage {stage['number']}.") sys.exit(1) state["attempt"] += 1 save_state(state) commit_progress(experiment) state["status"] = "done" save_state(state) commit_progress(state["log"][-1]["experiment"]) print("Generation 5 complete: handling, intercepts, and league stages passed.") if __name__ == "__main__": main()