BattleSnake

Multiplayer snake game where snakes compete to be the last survivor.

Overview

BattleSnake is a multi-player version of the classic snake game. Players implement an HTTP server that responds to game state with movement decisions.

Game Rules

Snakes move on a grid
Eat food to grow longer
Avoid walls and other snakes
Last snake standing wins

Submission Format

Players must implement an HTTP server with specific endpoints:

GET /: Return snake metadata
POST /start: Handle game start
POST /move: Return movement decision
POST /end: Handle game end

Configuration Example

game:
  name: BattleSnake
  rounds: 10
  sims_per_round: 5
  timeout: 300

players:
  - name: Snake1
    model: gpt-4
  - name: Snake2
    model: claude-3

Resources

Implementation

codeclash.arenas.battlesnake.battlesnake.BattleSnakeArena

BattleSnakeArena(config, **kwargs)

Bases: CodeArena

Source code in codeclash/arenas/battlesnake/battlesnake.py

def __init__(self, config, **kwargs):
    super().__init__(config, **kwargs)
    self.run_cmd_round: str = "./battlesnake play"
    for arg, val in self.game_config.get("args", self.default_args).items():
        if isinstance(val, bool):
            if val:
                self.run_cmd_round += f" --{arg}"
        else:
            self.run_cmd_round += f" --{arg} {val}"
    self._failed_to_start_player = []

name `class-attribute` `instance-attribute`

name: str = 'BattleSnake'

submission `class-attribute` `instance-attribute`

submission: str = 'main.py'

description `class-attribute` `instance-attribute`

description: str = 'Your bot (`main.py`) controls a snake on a grid-based board.\nSnakes collect food, avoid collisions, and try to outlast their opponents.'

default_args `class-attribute` `instance-attribute`

default_args: dict = {'width': 11, 'height': 11, 'browser': False}

run_cmd_round `instance-attribute`

run_cmd_round: str = './battlesnake play'

execute_round

execute_round(agents: list[Player])

Source code in codeclash/arenas/battlesnake/battlesnake.py

def execute_round(self, agents: list[Player]):
    self._failed_to_start_player = []
    assert len(agents) > 1, "Battlesnake requires at least two players"
    self.logger.debug("Starting game servers")
    player2port = {}
    for idx, agent in enumerate(agents):
        port = 8001 + idx
        player2port[agent.name] = port
        # Start server in background (& ). Submission may be Python (main.py) or any
        # other language via a run.sh launch script.
        self.environment.execute(f"PORT={port} {self._start_cmd(agent)} &", cwd=f"/{agent.name}")

    self.logger.debug(f"Waiting for ports: {player2port}")
    available_ports = self._wait_for_ports(list(player2port.values()))

    if not available_ports:
        raise RuntimeError("All games failed to start")

    if len(available_ports) == 1:
        missing_ports = set(player2port.values()) - set(available_ports)
        missing_player = next(player for player, port in player2port.items() if port in missing_ports)
        self.logger.warning(f"Player {missing_player} failed to start")
        self._failed_to_start_player.append(missing_player)
        return

    if len(available_ports) < len(agents):
        raise RuntimeError(f"Only {len(available_ports)} players started: {available_ports}")

    self.logger.debug("All ports are ready")

    try:
        self.logger.info(f"Running game with players: {list(player2port.keys())}")

        # Use ThreadPoolExecutor for parallel execution. Concurrency is configurable
        # (game.sim_concurrency): the single-threaded bot servers serialize move
        # requests, so total concurrency across all parallel pairs must stay bounded or
        # responses exceed the move timeout and games degenerate. When running many pairs
        # concurrently (ladder --workers), lower this so workers*sim_concurrency stays ~20.
        max_sim_workers = self.game_config.get("sim_concurrency", 20)
        with ThreadPoolExecutor(max_sim_workers) as executor:
            # Submit all simulations to the thread pool
            futures = [
                executor.submit(self._run_single_simulation, player2port, idx)
                for idx in range(self.game_config["sims_per_round"])
            ]

            # Collect results as they complete
            for future in tqdm(as_completed(futures), total=len(futures)):
                future.result()
    finally:
        # Kill all servers started this round (any language) so ports free up for the
        # next round. pkill covers the Python starter; fuser frees each game port for
        # compiled/interpreted servers launched via run.sh.
        self.environment.execute(f"pkill -f 'python {self.submission}' || true")
        for port in player2port.values():
            self.environment.execute(f"fuser -k {port}/tcp 2>/dev/null || true")

get_results

get_results(agents: list[Player], round_num: int, stats: RoundStats)

Source code in codeclash/arenas/battlesnake/battlesnake.py

def get_results(self, agents: list[Player], round_num: int, stats: RoundStats):
    scores = defaultdict(int)
    available_players = [player.name for player in agents if player.name not in self._failed_to_start_player]
    if len(available_players) > 1:
        # We ran the game
        for idx in range(self.game_config["sims_per_round"]):
            try:
                with open(self.log_round(round_num) / f"sim_{idx}.jsonl") as f:
                    lines = f.read().strip().split("\n")
                    results = json.loads(lines[-1])  # Get the last line which contains the game result
                    winner = RESULT_TIE if results["isDraw"] else results["winnerName"]
                    scores[winner] += 1
            except FileNotFoundError:
                self.logger.warning(f"Simulation {idx} not found, skipping")
            except json.JSONDecodeError:
                self.logger.warning(f"Simulation {idx} is not a valid JSON, skipping")
    else:
        self.logger.warning(f"Only one player ({available_players[0]}) started, giving them the win")
        # We didn't run a game, so we just give the one player the win
        available_player = available_players[0]
        scores = {available_player: self.game_config["sims_per_round"]}

    winner = max(scores, key=scores.get)
    winner = RESULT_TIE if list(scores.values()).count(scores[winner]) > 1 else winner
    stats.winner = winner
    stats.scores = scores
    for player, score in scores.items():
        if player != RESULT_TIE:
            stats.player_stats[player].score = score

validate_code

validate_code(agent: Player) -> tuple[bool, str | None]

Source code in codeclash/arenas/battlesnake/battlesnake.py

def validate_code(self, agent: Player) -> tuple[bool, str | None]:
    listing = agent.environment.execute("ls")["output"]
    # Non-Python submissions declare how to launch their server via run.sh (any
    # language). We trust the launch script here; a broken one is caught at runtime
    # by _wait_for_ports (failed-to-start -> forfeit).
    if "run.sh" in listing:
        return True, None
    if self.submission not in listing:
        return False, f"No {self.submission} file found in the root directory"
    # note: no longer calling splitlines
    bot_content = agent.environment.execute(f"cat {self.submission}")["output"]
    error_msg = []
    for func in [
        "def info(",
        "def start(",
        "def end(",
        "def move(",
    ]:
        if func not in bot_content:
            error_msg.append(f"There should be a `{func}` function implemented in `{self.submission}`")
    if len(error_msg) > 0:
        return False, "\n".join(error_msg + ["Don't change the function signatures!"])
    if "__main__" not in bot_content:
        return False, (
            f'`{self.submission}` must keep its `if __name__ == "__main__"` block that starts '
            "the server, or the bot fails to launch."
        )
    return True, None