Add fast_forward mode to CC:Ladder (#120)

* Add fast_forward mode to CC:Ladder (skip rungs the bot already dominates)

* Add `early_clinch` flag for ladder (skips remaining rounds if # of rounds to win reached before last round)

Author: john-b-yang

codeclash/cli/app.py

@@ -57,6 +57,12 @@ def run(
     preprocessed_yaml = resolve_includes(yaml_content, base_dir=CONFIG_DIR)
     config = yaml.safe_load(preprocessed_yaml)
 
+    if "ladder_rules" in config:
+        raise typer.BadParameter(
+            "ladder_rules (min_round_wins, win_last_k, fast_forward, early_clinch) applies only to "
+            "`codeclash ladder run`, not `codeclash run`."
+        )
+
     def get_output_path() -> Path:
         if is_running_in_aws_batch():
             # Offset timestamp by random seconds to avoid collisions

codeclash/tournaments/ladder.py

@@ -83,6 +83,50 @@ def resolve_ladder_rules(ladder_rules: dict, rounds: int) -> tuple[int, int]:
     return min_round_wins, win_last_k
 
 
+def resolve_fast_forward(ladder_rules: dict) -> tuple[bool, float]:
+    """Validate the optional ``ladder_rules.fast_forward`` sub-block and return
+    ``(enabled, min_sim_win_rate)``.
+
+    Fast-forward lets a climber *skip playing* a rung whose carried-over codebase already dominates:
+    if the climber wins at least ``min_sim_win_rate`` of the rung's round-0 simulations (ties count
+    as non-wins), the rung is cleared without spending edit rounds. Absent or ``enabled: false`` ->
+    ``(False, 0.0)`` = today's full-play behavior.
+    """
+    ff = ladder_rules.get("fast_forward")
+    if ff is None:
+        return False, 0.0
+    if "enabled" not in ff:
+        raise ValueError("ladder_rules.fast_forward.enabled is required when a fast_forward block is present.")
+    enabled = ff["enabled"]
+    if not isinstance(enabled, bool):
+        raise ValueError(f"ladder_rules.fast_forward.enabled must be a bool, got {enabled!r}.")
+    if not enabled:
+        return False, 0.0
+    if "min_sim_win_rate" not in ff:
+        raise ValueError("ladder_rules.fast_forward.min_sim_win_rate is required when fast_forward is enabled.")
+    rate = ff["min_sim_win_rate"]
+    if isinstance(rate, bool) or not isinstance(rate, (int, float)):
+        raise ValueError(f"ladder_rules.fast_forward.min_sim_win_rate must be a number, got {rate!r}.")
+    if not 0.5 < rate <= 1.0:
+        raise ValueError(f"ladder_rules.fast_forward.min_sim_win_rate must be in (0.5, 1.0], got {rate}.")
+    return True, float(rate)
+
+
+def resolve_early_clinch(ladder_rules: dict, win_last_k: int) -> bool:
+    """Validate the optional ``ladder_rules.early_clinch`` flag (default ``False``).
+
+    When true, a rung stops as soon as the climber has won ``min_round_wins`` agent rounds rather than
+    always playing all ``rounds``. Requires ``win_last_k == 0``: a trailing-rounds requirement can't be
+    decided before the final round, so early-stopping would be unsound.
+    """
+    ec = ladder_rules.get("early_clinch", False)
+    if not isinstance(ec, bool):
+        raise ValueError(f"ladder_rules.early_clinch must be a bool, got {ec!r}.")
+    if ec and win_last_k != 0:
+        raise ValueError("ladder_rules.early_clinch requires ladder_rules.win_last_k == 0.")
+    return ec
+
+
 def build_ladder(config: dict, workers: int = 1) -> None:
     """Build a ladder: run PvP tournaments across all pairs of players (for ranking).
 
@@ -154,6 +198,8 @@ def __init__(
         self.rounds = config["tournament"]["rounds"]
         self.sims = config["game"]["sims_per_round"]
         self.min_round_wins, self.win_last_k = resolve_ladder_rules(config.get("ladder_rules", {}), self.rounds)
+        self.ff_enabled, self.ff_min_win_rate = resolve_fast_forward(config.get("ladder_rules", {}))
+        self.early_clinch = resolve_early_clinch(config.get("ladder_rules", {}), self.win_last_k)
 
         del config["player"]
         del config["ladder"]
@@ -254,9 +300,41 @@ def _scan_resume(self, resume_dir: Path) -> tuple[int, str | 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)
+            # A fast-forwarded rung made no code changes (no round tags), so it isn't a new carry-over
+            # point — keep the tag from the last *played* rung.
+            if not advancement.get("fast_forwarded"):
+                resume_tag = self._climber_final_tag(meta)
         raise ValueError("Run already cleared the entire ladder. Nothing to resume.")
 
+    def _fast_forward_probe(self, rung_config: dict, rung_dir: Path) -> float:
+        """Run round 0 only in a throwaway probe dir and return the climber's share of the round-0
+        simulations (ties count as non-wins). Used to decide whether to skip playing the rung."""
+        probe_config = copy.deepcopy(rung_config)
+        probe_config["tournament"] = {**probe_config["tournament"], "rounds": 0}
+        probe_dir = rung_dir.parent / f".ff-probe.{rung_dir.name}"
+        if probe_dir.exists():
+            shutil.rmtree(probe_dir)
+        try:
+            probe = PvpTournament(
+                probe_config, output_dir=probe_dir, cleanup=self.cleanup, keep_containers=self.keep_containers
+            )
+            probe.run()
+            meta = json.loads((probe_dir / "metadata.json").read_text())
+            round_stats = meta.get("round_stats", {})
+            r0 = round_stats.get("0") or round_stats.get(0) or {}
+            wins = (r0.get("scores") or {}).get(self.player["name"], 0)
+            return wins / self.sims if self.sims else 0.0
+        finally:
+            if probe_dir.exists():
+                shutil.rmtree(probe_dir)
+
+    def _record_fast_forward(self, rung_dir: Path, win_rate: float) -> None:
+        """Write a minimal rung metadata.json marking it cleared-by-fast-forward (no rounds played),
+        so resume and the ladder summary can account for it like any other cleared rung."""
+        rung_dir.mkdir(parents=True, exist_ok=True)
+        meta = {"ladder_advancement": {"cleared": True, "fast_forwarded": True, "round0_win_rate": round(win_rate, 4)}}
+        (rung_dir / "metadata.json").write_text(json.dumps(meta, indent=2))
+
     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.
 
@@ -274,6 +352,7 @@ def run(self) -> dict:
         logger.info(advancement_rule)
 
         advanced = False
+        fast_forwarded = 0
         opponent: dict = {}
         opponent_rank = 0
         rung = 0
@@ -307,11 +386,35 @@ def run(self) -> dict:
             # 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)
+
+            # Fast-forward gate: if enabled and the carried-over bot already wins round 0 by a large
+            # enough margin, clear this rung without playing the edit rounds (see resolve_fast_forward).
+            if self.ff_enabled:
+                ff_rate = self._fast_forward_probe(c, tournament_dir)
+                if ff_rate >= self.ff_min_win_rate:
+                    self._record_fast_forward(tournament_dir, ff_rate)
+                    advanced = True
+                    fast_forwarded += 1
+                    print("=" * 10)
+                    print(
+                        f"{self.player['name']} fast-forwarded rung {rung}/{total} ({opponent['name']}, "
+                        f"elo #{opponent_rank}) — won {ff_rate:.0%} of round-0 sims "
+                        f"(>= {self.ff_min_win_rate:.0%}).\nLadder challenge continuing"
+                    )
+                    print("=" * 10)
+                    continue
+
+            # When early_clinch is on (win_last_k == 0 enforced), stop the rung once the climber has
+            # locked in `min_round_wins` rather than playing out the remaining rounds.
+            early_stop = None
+            if self.early_clinch:
+                early_stop = lambda winners: self._evaluate_advancement(winners, self.player["name"])[2]  # noqa: E731
             tournament = PvpTournament(
                 c,
                 output_dir=tournament_dir,
                 cleanup=self.cleanup,
                 keep_containers=self.keep_containers,
+                early_stop=early_stop,
             )
             tournament.run()
 
@@ -366,6 +469,7 @@ def run(self) -> dict:
             "win_last_k": self.win_last_k,
             "ladder_size": len(self.ladder),
             "rungs_cleared": rungs_cleared,
+            "rungs_fast_forwarded": fast_forwarded,
             "final_opponent": opponent["name"],
             "final_opponent_rank": opponent_rank,
             "cleared_ladder": rungs_cleared == len(self.ladder),

codeclash/tournaments/pvp.py

@@ -5,6 +5,7 @@
 import json
 import shutil
 import subprocess
+from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
@@ -28,13 +29,16 @@ def __init__(
         output_dir: Path,
         cleanup: bool = False,
         keep_containers: bool = False,
+        early_stop: Callable[[list[str]], bool] | None = None,
     ):
         metadata_file = output_dir / "metadata.json"
         if metadata_file.exists():
             raise FileExistsError(f"Metadata file already exists: {metadata_file}")
 
         super().__init__(config, name="PvpTournament", output_dir=output_dir)
         self.cleanup_on_end = cleanup
+        # Given the agent-round winners so far (rounds 1..n), return True to stop before `rounds`.
+        self.early_stop = early_stop
         self.game: CodeArena = get_arena(
             self.config,
             tournament_id=self.tournament_id,
@@ -91,12 +95,18 @@ def run(self) -> None:
         """Main execution function that runs all rounds."""
         try:
             self.run_competition_phase(0)  # Initial round with identical codebases
+            last_round = self.rounds
             for round_num in range(1, self.rounds + 1):
                 self.run_edit_phase(round_num)
                 self.run_competition_phase(round_num)
-            # Need to separately compress the last round, because
+                if self.early_stop is not None:
+                    winners = [self._metadata["round_stats"][r]["winner"] for r in range(1, round_num + 1)]
+                    if self.early_stop(winners):
+                        last_round = round_num
+                        break
+            # Need to separately compress the last round played, because
             # in run_edit_phase we always only compress the previous round
-            self._compress_round_folder(self.rounds)
+            self._compress_round_folder(last_round)
         finally:
             self.end()
 

configs/ladder/README.md

@@ -54,7 +54,37 @@ ladder_rules:
   win_last_k: 0       # ...and must win the last K rounds (1 == just the final round, 0 == disabled)
 ```
 
+These keys are read only by `codeclash ladder run`; a plain PvP `codeclash run` rejects any config that carries a `ladder_rules` block (so `early_clinch`/`fast_forward` can't silently no-op there).
+
 Both keys are **required** — there are no defaults; a config that omits either one errors out. `min_round_wins` is a whole number: the player advances when `player_wins >= min_round_wins`. Round 0 (before any edits against the opponent — identical codebases at the first rung, carried-over at later rungs) is excluded, so with `tournament.rounds: 5` there are 5 scored agent rounds (rounds 1–5). Validation:
 
 - `min_round_wins` must be an integer with `1 <= min_round_wins <= tournament.rounds`.
 - `win_last_k` must be an integer with `0 <= win_last_k <= min_round_wins`. `0` **disables** the trailing-rounds requirement; `1` means "just win the final round".
+
+### Fast-forward (optional)
+
+Since the north-star metric is *the highest rung a bot can reach*, you can skip the (usually redundant) grind of playing edit rounds against opponents your carried-over bot already crushes:
+
+```yaml
+ladder_rules:
+  min_round_wins: 2
+  win_last_k: 0
+  fast_forward:
+    enabled: true
+    min_sim_win_rate: 0.9   # skip a rung if the bot wins >= 90% of that rung's round-0 sims
+```
+
+At each rung, a **round-0-only probe** runs the carried-over bot against the opponent. If the climber wins at least `min_sim_win_rate` of the simulations (ties count as non-wins), the rung is **cleared without playing edit rounds** and recorded with `ladder_advancement.fast_forwarded: true`. Any rung *not* cleanly won still plays in full under the normal `ladder_rules`, so this is safe under non-transitive matchups and never skips a rung it would actually lose. Rung 1 is identical-code (~coin flip at round 0), so it never fast-forwards — the climber always has to earn the first rung. Omit the block (or `enabled: false`) for today's full-play behavior. Validation: `enabled` is a bool (required if the block is present); `min_sim_win_rate` is a number in `(0.5, 1.0]` (required when enabled).
+
+### Early clinch (optional)
+
+Where fast-forward skips a rung entirely, `early_clinch` stops a rung mid-way once the outcome is decided — the climber has already won `min_round_wins` rounds, so the remaining rounds can't change whether it advances:
+
+```yaml
+ladder_rules:
+  min_round_wins: 2
+  win_last_k: 0
+  early_clinch: true
+```
+
+The rung plays rounds normally and breaks the moment `player_wins >= min_round_wins`; advancement is recorded from the rounds actually played. Requires `win_last_k == 0` — a trailing-rounds requirement can't be satisfied before the final round, so the two can't be combined (validation errors otherwise). Composes with `fast_forward` (probe first, then early-clinch the rounds that do get played). Off by default.

configs/ladder/ladder_rules.yaml

@@ -1,2 +1,10 @@
 min_round_wins: 2
 win_last_k: 0
+# Optional: stop a rung as soon as the climber has won `min_round_wins` rounds instead of playing
+# them all. Requires win_last_k == 0. Off by default. Uncomment to enable:
+# early_clinch: true
+# Optional: skip playing a rung whose carried-over bot already dominates round 0 (faster eval,
+# same "highest rung reached" metric). Off by default. Uncomment to enable:
+# fast_forward:
+#   enabled: true
+#   min_sim_win_rate: 0.9   # clear the rung without editing if the bot wins >= 90% of round-0 sims

docs/usage/ladder.md

@@ -0,0 +1,105 @@
+# Ladder Tournament
+
+A ladder is a ranked list of opponents, weakest first.
+One model (the "climber") plays them one rung at a time and keeps going until it fails to advance.
+Its score is how high it climbs.
+
+## Two commands
+
+Build the ladder once — ranks the human bots by round-robin:
+
+```bash
+uv run codeclash ladder make configs/ladder/make_battlesnake.yaml
+```
+
+Then send a model up it:
+
+```bash
+uv run codeclash ladder run configs/ladder/battlesnake__opus_4_8.yaml
+```
+
+Resume an interrupted climb from its log dir (needs `push: True`):
+
+```bash
+uv run codeclash ladder run configs/ladder/battlesnake__opus_4_8.yaml -r logs/<user>/LadderTournament...
+```
+
+Options: `-c/--cleanup`, `-o/--output-dir`, `-s/--suffix`, `-k/--keep-containers`, `-r/--resume`.
+
+## Minimal run config
+
+```yaml
+tournament:
+  rounds: 5
+ladder_rules: !include ladder/ladder_rules.yaml
+game:
+  name: RobotRumble
+  sims_per_round: 250
+player:
+  agent: mini
+  name: claude-sonnet-4-5
+  branch_init: human/anton/anton3000
+  config:
+    agent: !include mini/default.yaml
+    model:
+      model_name: anthropic/claude-sonnet-4-5-20250929
+  push: True
+prompts: !include ladder/ladder_prompt.yaml
+ladder: !include ladder/rungs/robotrumble.yaml
+```
+
+`ladder` is the ranked opponent list; `player` is the climber.
+
+## Advancement settings (`ladder_rules`)
+
+These live in `configs/ladder/ladder_rules.yaml` and decide what it takes to clear a rung.
+
+**`min_round_wins`** (required)
+How many rounds the climber must win to move up a rung.
+Round 0, before any edits, doesn't count.
+
+**`win_last_k`** (required)
+Also require winning the last `k` rounds — `1` means just the final round.
+Set to `0` to turn this off.
+
+**`early_clinch`** (optional)
+Stop a rung early the moment the climber has already won `min_round_wins` rounds.
+Saves time, and the outcome is identical to playing every round.
+Only allowed when `win_last_k` is `0`.
+
+**`fast_forward`** (optional)
+Skip a rung entirely if the climber already crushes it before making any edits.
+It plays only round 0; if the climber wins at least `min_sim_win_rate` of the sims, the rung clears.
+
+## Copy-paste examples
+
+Win 2+ of `n` rounds:
+
+```yaml
+min_round_wins: 2
+win_last_k: 0
+```
+
+Win 2+ rounds, and the final round must be one of them:
+
+```yaml
+min_round_wins: 2
+win_last_k: 1
+```
+
+Fastest eval:
+
+```yaml
+min_round_wins: 2
+win_last_k: 0
+early_clinch: true
+fast_forward:
+  enabled: true
+  min_sim_win_rate: 0.9
+```
+
+* Win 2+ of `n` rounds
+* `early_clinch`: If the model wins 2 rounds before playing all `n`, skip remaining rounds
+* `fast_forward`: If the model's current solution beats the next opponent, skip playing that opponent all together.
+
+--8<-- "docs/_footer.md"