Add resumable CC:Ladder runs (ladder run --resume)

Author: john-b-yang

codeclash/agents/player.py

@@ -30,6 +30,9 @@ def __init__(
         self.environment = environment
         self.game_context = game_context
         self.push = config.get("push", False)
+        # Force the branch push (used on ladder resume, to overwrite an interrupted rung's partial
+        # pushes). Safe here because a run's push branch has a single writer. Uses --force-with-lease.
+        self._force_push = config.get("force_push", False)
         self.logger = get_logger(
             self.name,
             log_path=self.game_context.log_local / "players" / self.name / "player.log",
@@ -140,8 +143,9 @@ def post_run_hook(self, *, round: int) -> None:
         self._write_changes_to_file(round=round)
 
         if self.push:
+            force = " --force-with-lease" if self._force_push else ""
             for cmd in [
-                f"git push -u origin {self._branch_name}",
+                f"git push{force} -u origin {self._branch_name}",
                 "git push origin --tags",
             ]:
                 assert_zero_exit_code(self.environment.execute(cmd), logger=self.logger)

codeclash/cli/ladder.py

@@ -49,10 +49,18 @@ def run(
     keep_containers: bool = typer.Option(
         False, "--keep-containers", "-k", help="Do not remove containers after games/agent finish."
     ),
+    resume_from: Path | None = typer.Option(
+        None,
+        "--resume",
+        "-r",
+        help="Resume an interrupted run: pass its log dir. Skips cleared rungs and seeds from the "
+        "last cleared rung's pushed codebase. Requires push: True and the same config.",
+    ),
 ):
     """Send a model up a ranked ladder, rung by rung, until it loses.
 
     [dim]• codeclash ladder run path/to/ladder_config.yaml -c  # clean up after each rung[/dim]
+    [dim]• codeclash ladder run path/to/ladder_config.yaml -r logs/<user>/LadderTournament...  # resume[/dim]
     """
     config = _load_config(config_path)
     try:
@@ -62,6 +70,7 @@ def run(
             suffix=suffix,
             cleanup=cleanup,
             keep_containers=keep_containers,
+            resume_from=resume_from,
         )
     except ValueError as e:
         typer.echo(str(e))

codeclash/tournaments/ladder.py

@@ -14,6 +14,7 @@
 import copy
 import getpass
 import json
+import shutil
 import time
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from pathlib import Path
@@ -143,6 +144,7 @@ def __init__(
         suffix: str = "",
         cleanup: bool = False,
         keep_containers: bool = False,
+        resume_from: Path | None = None,
     ):
         # Extract ladder-specific keys and strip them from the config that gets handed to each
         # per-rung PvpTournament (which only understands `players`).
@@ -161,13 +163,45 @@ def __init__(
         self.cleanup = cleanup
         self.keep_containers = keep_containers
         self.output_dir = output_dir
+        self._resuming = resume_from is not None
+        self._start_idx = 0
 
-        timestamp = time.strftime("%y%m%d%H%M%S")
-        game_name = config["game"]["name"]
-        ladder_folder = f"LadderTournament.{game_name}.r{self.rounds}.s{self.sims}.{self.player['name']}.{timestamp}"
-        self.player["branch"] = ladder_folder
         base = base_dir if base_dir is not None else LOCAL_LOG_DIR / getpass.getuser()
-        self.parent_dir = base / ladder_folder
+
+        if not self._resuming:
+            timestamp = time.strftime("%y%m%d%H%M%S")
+            game_name = config["game"]["name"]
+            ladder_folder = (
+                f"LadderTournament.{game_name}.r{self.rounds}.s{self.sims}.{self.player['name']}.{timestamp}"
+            )
+            self.player["branch"] = ladder_folder
+            self.parent_dir = base / ladder_folder
+        else:
+            # Continue the interrupted run IN PLACE: reuse its log dir and its push branch (the dir
+            # name IS the branch name). The top-level metadata.json is written only after the climb
+            # loop finishes (win or lose), so its presence means there is nothing to resume.
+            self.parent_dir = resume_from
+            self.player["branch"] = resume_from.name
+            if not self.player.get("push"):
+                raise ValueError(
+                    "--resume requires push: True — the pushed branch and round tags are the codebase store."
+                )
+            if (resume_from / "metadata.json").exists():
+                raise ValueError(
+                    f"{resume_from.name} already finished (top-level metadata.json present); nothing to resume."
+                )
+            self._start_idx, resume_tag = self._scan_resume(resume_from)
+            if resume_tag is not None:
+                self.player["branch_init"] = resume_tag
+            # The first resumed rung force-resets the branch to `resume_tag`, discarding any partial
+            # rounds the interrupted rung pushed; later rungs then fast-forward normally.
+            self.player["force_push"] = True
+            msg = (
+                f"Resuming {resume_from.name}: {self._start_idx} rung(s) already cleared; "
+                f"seeding codebase from {resume_tag or self.player.get('branch_init')}."
+            )
+            print(msg)
+            logger.info(msg)
 
     def _advancement_rule_str(self) -> str:
         last_k_rule = "disabled" if self.win_last_k == 0 else f"win the last {self.win_last_k} round(s)"
@@ -176,15 +210,53 @@ def _advancement_rule_str(self) -> str:
             f"(baseline round 0 excluded) and {last_k_rule}."
         )
 
-    def _rung_dir(self, players: list[str]) -> Path:
+    def _rung_folder_name(self, players: list[str]) -> str:
         p_num = len(players)
         p_list = ".".join(players)
         suffix_part = f".{self.suffix}" if self.suffix else ""
-        folder_name = (
-            f"PvpTournament.{self.config['game']['name']}.r{self.rounds}.s{self.sims}.p{p_num}.{p_list}{suffix_part}"
-        )
+        return f"PvpTournament.{self.config['game']['name']}.r{self.rounds}.s{self.sims}.p{p_num}.{p_list}{suffix_part}"
+
+    def _rung_dir(self, players: list[str]) -> Path:
+        folder_name = self._rung_folder_name(players)
         return self.parent_dir / folder_name if self.output_dir is None else self.output_dir / folder_name
 
+    def _climber_final_tag(self, rung_metadata: dict) -> str:
+        """The climber's git tag for the last round of a (cleared) rung — the codebase to carry over."""
+        for agent in rung_metadata.get("agents", []):
+            if agent.get("name") == self.player["name"]:
+                tags = agent.get("round_tags") or {}
+                if not tags:
+                    raise ValueError("A cleared rung has no round tags to resume from (was push enabled?).")
+                return tags[max(tags, key=lambda k: int(k))]
+        raise ValueError(f"Climber {self.player['name']!r} not found in the resumed rung's metadata.")
+
+    def _scan_resume(self, resume_dir: Path) -> tuple[int, str | None]:
+        """Inspect an interrupted run and return ``(start_idx, resume_tag)``: the first rung not yet
+        cleared, and the climber's codebase tag at the end of the last cleared rung (``None`` if the
+        run never cleared a rung). Reads only artifacts a normal run already writes — per-rung
+        ``metadata.json`` (``ladder_advancement.cleared``) and the pushed ``round_tags``.
+        """
+        if not resume_dir.exists():
+            raise ValueError(f"--resume directory does not exist: {resume_dir}")
+        resume_tag: str | None = None
+        for idx, opponent in enumerate(self.ladder):
+            players = [self.player["name"], _player_slug(opponent["branch_init"])]
+            meta_path = resume_dir / self._rung_folder_name(players) / "metadata.json"
+            if not meta_path.exists():
+                if idx == 0 and not any(resume_dir.glob("PvpTournament.*")):
+                    return 0, None  # nothing was completed → equivalent to a fresh run
+                if idx == 0:
+                    raise ValueError(f"--resume dir has no rung matching this config: {resume_dir}")
+                return idx, resume_tag  # reached the interrupted rung
+            meta = json.loads(meta_path.read_text())
+            advancement = meta.get("ladder_advancement")
+            if advancement is None:
+                return idx, resume_tag  # rung ran but never finished → resume here
+            if not advancement.get("cleared"):
+                raise ValueError(f"Run already ended: climber lost at rung {idx + 1}. Nothing to resume.")
+            resume_tag = self._climber_final_tag(meta)
+        raise ValueError("Run already cleared the entire ladder. Nothing to resume.")
+
     def _evaluate_advancement(self, round_winners: list[str], player_name: str) -> tuple[int, bool, bool]:
         """Apply the advancement rule to a rung's round winners.
 
@@ -207,6 +279,8 @@ def run(self) -> dict:
         rung = 0
         total = len(self.ladder)
         for idx, opponent in enumerate(self.ladder):
+            if idx < self._start_idx:
+                continue  # already cleared in the run we're resuming
             # `rung` counts the climb from the bottom (1 = weakest opponent faced first, `total` =
             # strongest); `opponent_rank` is the opponent's Elo standing (1 = strongest overall).
             rung = idx + 1
@@ -215,10 +289,10 @@ def run(self) -> dict:
             # Prefix the climber's commit/tag messages with rung context (a prefix carrying its own
             # trailing separator; see Player._round_message).
             self.player["commit_label"] = f"Rung {rung}/{total} ({opponent['name']}, elo #{opponent_rank}) — "
-            if "branch_init" in self.player and idx > 0:
-                # After first opponent, remove branch_init so the player continues from the
-                # previous tournament's codebase.
-                del self.player["branch_init"]
+            if idx > self._start_idx:
+                # After the first executed rung, drop branch_init so the player continues from the
+                # previous rung's pushed codebase (carry-over).
+                self.player.pop("branch_init", None)
             c = {
                 **self.config,
                 "players": [
@@ -229,6 +303,10 @@ def run(self) -> dict:
 
             players = [p["name"] for p in c["players"]]
             tournament_dir = self._rung_dir(players)
+            # When resuming in place, the interrupted rung left a partial dir; clear it so
+            # PvpTournament (which refuses a pre-existing metadata.json) can re-run it fresh.
+            if self._resuming and idx == self._start_idx and tournament_dir.exists():
+                shutil.rmtree(tournament_dir)
             tournament = PvpTournament(
                 c,
                 output_dir=tournament_dir,