From 9a53f59bbd749550ecd34873eaba8169a550db5a Mon Sep 17 00:00:00 2001 From: muhtasham Date: Tue, 5 May 2026 14:01:13 +0200 Subject: [PATCH] Add Bomberland arena --- README.md | 6 +- codeclash/arenas/__init__.py | 2 + .../arenas/bomberland/Bomberland.Dockerfile | 23 + codeclash/arenas/bomberland/__init__.py | 3 + codeclash/arenas/bomberland/bomberland.py | 161 ++++++ .../arenas/bomberland/runtime/.gitignore | 3 + codeclash/arenas/bomberland/runtime/README.md | 38 ++ .../bomberland/runtime/bomberland_agent.py | 5 + .../bomberland/runtime/run_bomberland.py | 483 ++++++++++++++++++ .../examples/Bomberland__dummy__r1__s2.yaml | 36 ++ docs/reference/arenas/bomberland.md | 87 ++++ docs/reference/index.md | 1 + mkdocs.yml | 1 + tests/arenas/test_bomberland.py | 311 +++++++++++ 14 files changed, 1159 insertions(+), 1 deletion(-) create mode 100644 codeclash/arenas/bomberland/Bomberland.Dockerfile create mode 100644 codeclash/arenas/bomberland/__init__.py create mode 100644 codeclash/arenas/bomberland/bomberland.py create mode 100644 codeclash/arenas/bomberland/runtime/.gitignore create mode 100644 codeclash/arenas/bomberland/runtime/README.md create mode 100644 codeclash/arenas/bomberland/runtime/bomberland_agent.py create mode 100644 codeclash/arenas/bomberland/runtime/run_bomberland.py create mode 100644 configs/examples/Bomberland__dummy__r1__s2.yaml create mode 100644 docs/reference/arenas/bomberland.md create mode 100644 tests/arenas/test_bomberland.py diff --git a/README.md b/README.md index 6dda5e5..8094b5d 100644 --- a/README.md +++ b/README.md @@ -98,7 +98,11 @@ The winner is the LM agent who wins the most rounds. ## 🧩 Available Arenas CodeClash includes competitive programming games and simulation-backed arenas, including BattleSnake, -CoreWar, CybORG, Halite, HuskyBench, RoboCode, RobotRumble, and SCML. +Bomberland, CoreWar, CybORG, Halite, HuskyBench, RoboCode, RobotRumble, and SCML. + +Bomberland is a Bomberman-style grid arena based on Coder One's Bomberland competition. Agents edit +a Python `bomberland_agent.py` implementation and compete to maximize average score across seeded +simulations through survival, damage, kills, and destructible-block control. SCML is a supply-chain negotiation arena based on the ANAC Supply Chain Management League OneShot track. Agents edit a Python `scml_agent.py` implementation and compete to maximize average profit diff --git a/codeclash/arenas/__init__.py b/codeclash/arenas/__init__.py index 700f87d..463d1cc 100644 --- a/codeclash/arenas/__init__.py +++ b/codeclash/arenas/__init__.py @@ -3,6 +3,7 @@ from codeclash.arenas.battlecode24.battlecode24 import BattleCode24Arena from codeclash.arenas.battlecode25.battlecode25 import BattleCode25Arena from codeclash.arenas.battlesnake.battlesnake import BattleSnakeArena +from codeclash.arenas.bomberland.bomberland import BomberlandArena from codeclash.arenas.bridge.bridge import BridgeArena from codeclash.arenas.chess.chess import ChessArena from codeclash.arenas.corewar.corewar import CoreWarArena @@ -23,6 +24,7 @@ BattleCode24Arena, BattleCode25Arena, BattleSnakeArena, + BomberlandArena, BridgeArena, ChessArena, CoreWarArena, diff --git a/codeclash/arenas/bomberland/Bomberland.Dockerfile b/codeclash/arenas/bomberland/Bomberland.Dockerfile new file mode 100644 index 0000000..c395d4f --- /dev/null +++ b/codeclash/arenas/bomberland/Bomberland.Dockerfile @@ -0,0 +1,23 @@ +FROM python:3.11-slim-bookworm + +ARG BOMBERLAND_COMMIT=8b6b7a1c013d96feb0a5468a7a59a63a7c59dadc +ENV BOMBERLAND_UPSTREAM_COMMIT=${BOMBERLAND_COMMIT} + +RUN apt-get update \ + && apt-get install -y --no-install-recommends git ca-certificates \ + && rm -rf /var/lib/apt/lists/* + +# Keep a pinned copy of the upstream competition source for provenance and +# agent authors who want to inspect the original starter-kit shape. +RUN git clone https://github.com/CoderOneHQ/bomberland.git /opt/bomberland \ + && cd /opt/bomberland \ + && git checkout ${BOMBERLAND_COMMIT} + +WORKDIR /workspace +COPY codeclash/arenas/bomberland/runtime/ /workspace/ + +RUN git init \ + && git config user.email "arena@codeclash.com" \ + && git config user.name "CodeClash Arena" \ + && git add . \ + && git commit -m "Initialize Bomberland runtime" diff --git a/codeclash/arenas/bomberland/__init__.py b/codeclash/arenas/bomberland/__init__.py new file mode 100644 index 0000000..db578c4 --- /dev/null +++ b/codeclash/arenas/bomberland/__init__.py @@ -0,0 +1,3 @@ +from codeclash.arenas.bomberland.bomberland import BomberlandArena + +__all__ = ["BomberlandArena"] diff --git a/codeclash/arenas/bomberland/bomberland.py b/codeclash/arenas/bomberland/bomberland.py new file mode 100644 index 0000000..e8faba7 --- /dev/null +++ b/codeclash/arenas/bomberland/bomberland.py @@ -0,0 +1,161 @@ +import json +import shlex +import subprocess + +from codeclash.agents.player import Player +from codeclash.arenas.arena import CodeArena, RoundStats +from codeclash.constants import RESULT_TIE +from codeclash.utils.environment import assert_zero_exit_code + +RESULTS_JSON = "bomberland_results.json" +CRASH_SCORE = -1_000_000.0 + + +class BomberlandArena(CodeArena): + name: str = "Bomberland" + submission: str = "bomberland_agent.py" + description: str = """Bomberland is a Bomberman-style multi-agent arena based on Coder One's Bomberland competition. + +Your bot is a Python file named `bomberland_agent.py` that defines a callable named `next_actions`. +The callable receives a game-state dictionary and should return a dictionary mapping unit ids to actions: + + def next_actions(game_state): + return {"unit_0": "up"} + +Valid actions are `up`, `down`, `left`, `right`, `bomb`, and `stay`. Each round runs several +deterministic seeded games. Your units move on a destructible grid, place bombs, destroy blocks, +damage opposing units, and score by survival, damage, kills, and block destruction. +""" + default_args: dict = { + "sims_per_round": 4, + "ticks": 80, + "width": 11, + "height": 11, + "unit_count": 3, + "agent_timeout": 0.25, + "validation_timeout": 5, + "timeout": 180, + } + + def __init__(self, config: dict, **kwargs): + player_count = len(config.get("players", [])) + if player_count != 2: + raise ValueError("Bomberland requires exactly two players") + game_config = config.get("game", {}) + game_args = game_config.get("args", {}) + sims_per_round = int( + game_args.get("sims_per_round", game_config.get("sims_per_round", self.default_args["sims_per_round"])) + ) + if sims_per_round % 2 != 0: + raise ValueError("Bomberland requires an even sims_per_round so both players get paired starting sides") + super().__init__(config, **kwargs) + + def _game_arg(self, key: str): + nested_args = self.game_config.get("args", {}) + return nested_args.get(key, self.game_config.get(key, self.default_args[key])) + + def _sims_per_round(self) -> int: + return int(self._game_arg("sims_per_round")) + + def validate_code(self, agent: Player) -> tuple[bool, str | None]: + quoted_submission = shlex.quote(self.submission) + file_check = agent.environment.execute(f"test -f {quoted_submission} && echo exists") + if "exists" not in file_check["output"]: + return False, f"Submission file `{self.submission}` not found in the workspace root" + + content = agent.environment.execute(f"cat {quoted_submission}")["output"] + if not content.strip(): + return False, f"`{self.submission}` is empty" + + syntax_check = agent.environment.execute(f"python -m py_compile {quoted_submission}") + if syntax_check["returncode"] != 0: + return False, f"Python syntax error in `{self.submission}`:\n{syntax_check['output']}" + + import_check = agent.environment.execute( + "python - <<'PY'\n" + "import importlib.util\n" + f"spec = importlib.util.spec_from_file_location('submission_agent', {self.submission!r})\n" + "module = importlib.util.module_from_spec(spec)\n" + "spec.loader.exec_module(module)\n" + "assert hasattr(module, 'next_actions'), 'next_actions callable not found'\n" + "assert callable(module.next_actions), 'next_actions must be callable'\n" + "state = {\n" + " 'connection': {'agent_id': 'Alice'},\n" + " 'agents': {'Alice': {'unit_ids': ['u0']}},\n" + " 'unit_state': {'u0': {'agent_id': 'Alice', 'hp': 3, 'coordinates': [1, 1]}},\n" + " 'entities': [],\n" + " 'world': {'width': 5, 'height': 5},\n" + " 'tick': 0,\n" + "}\n" + "result = module.next_actions(state)\n" + "assert result is None or isinstance(result, dict), 'next_actions must return a dict or None'\n" + "PY", + timeout=int(self._game_arg("validation_timeout")), + ) + if import_check["returncode"] != 0: + return False, f"Could not import or call `next_actions` from `{self.submission}`:\n{import_check['output']}" + + return True, None + + def execute_round(self, agents: list[Player]) -> None: + agent_args = [] + for agent in agents: + agent_args.extend(["--agent", f"{agent.name}=/{agent.name}/{self.submission}"]) + + cmd = [ + "python", + "run_bomberland.py", + "--sims", + str(self._sims_per_round()), + "--ticks", + str(self._game_arg("ticks")), + "--width", + str(self._game_arg("width")), + "--height", + str(self._game_arg("height")), + "--unit-count", + str(self._game_arg("unit_count")), + "--agent-timeout", + str(self._game_arg("agent_timeout")), + "--output", + str(self.log_env / RESULTS_JSON), + *agent_args, + ] + full_cmd = " ".join(shlex.quote(part) for part in cmd) + self.logger.info(f"Running game: {full_cmd}") + try: + response = self.environment.execute(full_cmd, timeout=int(self._game_arg("timeout"))) + except subprocess.TimeoutExpired as exc: + raise RuntimeError("Bomberland round timed out") from exc + assert_zero_exit_code(response, logger=self.logger) + + def get_results(self, agents: list[Player], round_num: int, stats: RoundStats): + result_file = self.log_round(round_num) / RESULTS_JSON + if not result_file.exists(): + self.logger.error(f"Missing result file: {result_file}") + stats.winner = RESULT_TIE + for agent in agents: + stats.scores[agent.name] = CRASH_SCORE + stats.player_stats[agent.name].score = CRASH_SCORE + return + + with open(result_file) as f: + result = json.load(f) + + scores = {agent.name: CRASH_SCORE for agent in agents} + for player, score in result.get("average_scores", {}).items(): + if player in scores: + scores[player] = float(score) + + stats.scores = scores + stats.details = result.get("details", []) + for player, score in scores.items(): + stats.player_stats[player].score = score + + if not scores: + stats.winner = RESULT_TIE + return + + top_score = max(scores.values()) + winners = [player for player, score in scores.items() if score == top_score] + stats.winner = winners[0] if len(winners) == 1 else RESULT_TIE diff --git a/codeclash/arenas/bomberland/runtime/.gitignore b/codeclash/arenas/bomberland/runtime/.gitignore new file mode 100644 index 0000000..24f2ad6 --- /dev/null +++ b/codeclash/arenas/bomberland/runtime/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +*.pyc +bomberland_results.json diff --git a/codeclash/arenas/bomberland/runtime/README.md b/codeclash/arenas/bomberland/runtime/README.md new file mode 100644 index 0000000..37ef015 --- /dev/null +++ b/codeclash/arenas/bomberland/runtime/README.md @@ -0,0 +1,38 @@ +# Bomberland CodeClash Runtime + +This runtime adapts the Coder One Bomberland competition format into a compact, +deterministic CodeClash arena. The Docker image keeps a pinned checkout of +`CoderOneHQ/bomberland` at `/opt/bomberland` for provenance and starter-kit +reference, while `run_bomberland.py` provides the runtime used by CodeClash. + +Submissions must provide `bomberland_agent.py` with: + +```python +def next_actions(game_state): + return {"unit_0": "up"} +``` + +Valid string actions are `up`, `down`, `left`, `right`, `bomb`, and `stay`. +The game-state dictionary follows the upstream starter-kit shape where possible: +`connection.agent_id` identifies the player, `agents[player].unit_ids` lists the +controlled units, `unit_state` contains unit coordinates and health, and +`entities` contains walls, destructible blocks, bombs, and blast tiles. + +Round simulation counts must be even so each player receives both starting sides. + +Smoke command from the repository root: + +```bash +uv run python main.py configs/examples/Bomberland__dummy__r1__s2.yaml -o /tmp/codeclash-bomberland-smoke +``` + +Expected result shape: + +```json +{ + "average_scores": {"player_a": 330.0, "player_b": 330.0}, + "total_scores": {"player_a": 660.0, "player_b": 660.0}, + "sims": 2, + "details": ["... per-simulation JSON strings ..."] +} +``` diff --git a/codeclash/arenas/bomberland/runtime/bomberland_agent.py b/codeclash/arenas/bomberland/runtime/bomberland_agent.py new file mode 100644 index 0000000..c040048 --- /dev/null +++ b/codeclash/arenas/bomberland/runtime/bomberland_agent.py @@ -0,0 +1,5 @@ +def next_actions(game_state): + agent_id = game_state["connection"]["agent_id"] + unit_ids = game_state["agents"].get(agent_id, {}).get("unit_ids", []) + unit_state = game_state.get("unit_state", {}) + return {unit_id: "stay" for unit_id in unit_ids if unit_state.get(unit_id, {}).get("hp", 0) > 0} diff --git a/codeclash/arenas/bomberland/runtime/run_bomberland.py b/codeclash/arenas/bomberland/runtime/run_bomberland.py new file mode 100644 index 0000000..4cef424 --- /dev/null +++ b/codeclash/arenas/bomberland/runtime/run_bomberland.py @@ -0,0 +1,483 @@ +import argparse +import copy +import importlib.util +import json +import multiprocessing +import queue +import random +import sys +from collections import defaultdict +from pathlib import Path + +ACTIONS = {"up", "down", "left", "right", "bomb", "stay"} +DELTAS = { + "up": (0, -1), + "down": (0, 1), + "left": (-1, 0), + "right": (1, 0), +} +START_HP = 3 +START_BOMBS = 1 +BLAST_RADIUS = 3 +BOMB_TIMER = 6 + + +def load_agent(name, path): + agent_dir = str(Path(path).resolve().parent) + if agent_dir not in sys.path: + sys.path.insert(0, agent_dir) + spec = importlib.util.spec_from_file_location(f"bomberland_agent_{name}", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + if not hasattr(module, "next_actions") or not callable(module.next_actions): + raise ValueError(f"{path} must define a callable next_actions(game_state)") + return module.next_actions + + +def _agent_worker(path, state, result_queue): + try: + callback = load_agent("runtime", path) + result = callback(copy.deepcopy(state)) + result_queue.put({"actions": result if isinstance(result, dict) else {}}) + except BaseException as exc: + if isinstance(exc, (KeyboardInterrupt, SystemExit)): + raise + result_queue.put({"error": type(exc).__name__}) + + +def call_agent(agent_path, state, timeout): + timeout = max(float(timeout), 0.01) + start_method = "fork" if "fork" in multiprocessing.get_all_start_methods() else "spawn" + context = multiprocessing.get_context(start_method) + result_queue = context.Queue(maxsize=1) + process = context.Process(target=_agent_worker, args=(agent_path, copy.deepcopy(state), result_queue)) + process.start() + process.join(timeout) + if process.is_alive(): + process.terminate() + process.join(0.1) + if process.is_alive() and hasattr(process, "kill"): + process.kill() + process.join() + return {"__error__": "Timeout"} + if process.exitcode not in (0, None): + return {"__error__": f"ExitCode{process.exitcode}"} + try: + message = result_queue.get_nowait() + except queue.Empty: + return {"__error__": "NoResult"} + if "error" in message: + return {"__error__": message["error"]} + return message.get("actions", {}) + + +def mirror(pos, width, height): + x, y = pos + return width - 1 - x, height - 1 - y + + +def player_starts(width, height, unit_count): + if not 1 <= unit_count <= 4: + raise ValueError("unit_count must be between 1 and 4") + left = [(1, 1), (1, height - 2), (2, height // 2), (3, 1)] + right = [mirror(pos, width, height) for pos in left] + return left[:unit_count], right[:unit_count] + + +def build_map(width, height, unit_count, rng): + if width < 7 or height < 7: + raise ValueError("Bomberland maps must be at least 7x7") + + metal = set() + for x in range(width): + metal.add((x, 0)) + metal.add((x, height - 1)) + for y in range(height): + metal.add((0, y)) + metal.add((width - 1, y)) + for x in range(2, width - 2, 2): + for y in range(2, height - 2, 2): + metal.add((x, y)) + + starts_left, starts_right = player_starts(width, height, unit_count) + safe = set() + for pos in [*starts_left, *starts_right]: + safe.add(pos) + for dx, dy in DELTAS.values(): + safe.add((pos[0] + dx, pos[1] + dy)) + + wood = set() + for y in range(1, height - 1): + for x in range(1, width - 1): + pos = (x, y) + mirrored = mirror(pos, width, height) + if pos in metal or pos in safe or mirrored in metal or mirrored in safe: + continue + if pos in wood or mirrored in wood: + continue + if rng.random() < 0.28: + wood.add(pos) + wood.add(mirrored) + return metal, wood + + +def make_units(players, width, height, unit_count): + left_starts, right_starts = player_starts(width, height, unit_count) + if len(players) != 2: + raise ValueError("Bomberland currently expects exactly two players") + + starts_by_player = {players[0]: left_starts, players[1]: right_starts} + units = {} + agents = {} + for player in players: + unit_ids = [] + for index, pos in enumerate(starts_by_player[player]): + unit_id = f"{player}_unit_{index}" + unit_ids.append(unit_id) + units[unit_id] = { + "unit_id": unit_id, + "agent_id": player, + "coordinates": list(pos), + "hp": START_HP, + "inventory": {"bombs": START_BOMBS, "blast_diameter": BLAST_RADIUS * 2 - 1}, + } + agents[player] = {"agent_id": player, "unit_ids": unit_ids} + return units, agents + + +def pos_of(unit): + return tuple(unit["coordinates"]) + + +def entities_for_state(metal, wood, bombs, blasts): + entities = [] + for x, y in sorted(metal): + entities.append({"type": "m", "coordinates": [x, y]}) + for x, y in sorted(wood): + entities.append({"type": "w", "coordinates": [x, y]}) + for bomb in bombs: + entities.append( + { + "type": "b", + "coordinates": list(bomb["pos"]), + "owner": bomb["owner"], + "timer": bomb["timer"], + "blast_diameter": bomb["radius"] * 2 - 1, + } + ) + for (x, y), ttl in sorted(blasts.items()): + entities.append({"type": "x", "coordinates": [x, y], "ttl": ttl}) + return entities + + +def make_state(player, agents, units, metal, wood, bombs, blasts, width, height, tick): + return { + "connection": {"agent_id": player}, + "agents": copy.deepcopy(agents), + "unit_state": copy.deepcopy(units), + "entities": entities_for_state(metal, wood, bombs, blasts), + "world": {"width": width, "height": height}, + "tick": tick, + "config": { + "bomb_timer": BOMB_TIMER, + "start_hp": START_HP, + "start_bombs": START_BOMBS, + "blast_radius": BLAST_RADIUS, + }, + } + + +def normalize_action(raw_action): + if isinstance(raw_action, str): + action = raw_action.lower().strip() + if action in ACTIONS: + return action, None + if action.startswith("detonate:"): + try: + x_raw, y_raw = action.split(":", 1)[1].split(",", 1) + return "detonate", (int(x_raw), int(y_raw)) + except ValueError: + return "invalid", None + if isinstance(raw_action, dict): + action_type = str(raw_action.get("type", raw_action.get("action", ""))).lower() + if action_type == "move": + move = str(raw_action.get("move", raw_action.get("direction", ""))).lower() + return (move, None) if move in DELTAS else ("invalid", None) + if action_type in {"bomb", "stay"}: + return action_type, None + if action_type == "detonate": + coordinates = raw_action.get("coordinates", raw_action.get("coordinate")) + if isinstance(coordinates, (list, tuple)) and len(coordinates) == 2: + try: + return "detonate", (int(coordinates[0]), int(coordinates[1])) + except (TypeError, ValueError): + return "invalid", None + return "invalid", None + + +def blast_cells(origin, radius, metal, wood): + cells = [origin] + ox, oy = origin + for dx, dy in DELTAS.values(): + for distance in range(1, radius + 1): + pos = (ox + dx * distance, oy + dy * distance) + if pos in metal: + break + cells.append(pos) + if pos in wood: + break + return cells + + +def explode_bomb(index, bombs, metal, wood, units, stats, blasts): + if index >= len(bombs): + return + bomb = bombs[index] + cells = blast_cells(bomb["pos"], bomb["radius"], metal, wood) + owner = bomb["owner"] + stats[owner]["bombs_returned"] += 1 + unit_id = bomb.get("unit_id") + if unit_id in units: + units[unit_id]["inventory"]["bombs"] += 1 + + for cell in cells: + blasts[cell] = 1 + if cell in wood: + wood.remove(cell) + stats[owner]["wood_destroyed"] += 1 + + for unit in units.values(): + if unit["hp"] <= 0 or pos_of(unit) not in cells: + continue + unit["hp"] -= 1 + if unit["agent_id"] != owner: + stats[owner]["damage_dealt"] += 1 + if unit["hp"] <= 0: + stats[owner]["kills"] += 1 + + bombs.pop(index) + chained = True + while chained: + chained = False + for chained_index, chained_bomb in enumerate(list(bombs)): + if chained_bomb["pos"] in cells: + explode_bomb(chained_index, bombs, metal, wood, units, stats, blasts) + chained = True + break + + +def live_players(players, units): + alive = set() + for unit in units.values(): + if unit["hp"] > 0: + alive.add(unit["agent_id"]) + return [player for player in players if player in alive] + + +def apply_actions(players, callbacks, agents, units, metal, wood, bombs, blasts, stats, width, height, tick, timeout): + bomb_positions = {bomb["pos"] for bomb in bombs} + actions_by_unit = {} + + for player in players: + state = make_state(player, agents, units, metal, wood, bombs, blasts, width, height, tick) + result = call_agent(callbacks[player], state, timeout) + if "__error__" in result: + stats[player]["agent_errors"] += 1 + result = {} + for unit_id in agents[player]["unit_ids"]: + unit = units[unit_id] + if unit["hp"] <= 0: + continue + action, target = normalize_action(result.get(unit_id, "stay")) + if action == "invalid": + stats[player]["invalid_actions"] += 1 + action = "stay" + actions_by_unit[unit_id] = (action, target) + + for unit_id, (action, target) in actions_by_unit.items(): + unit = units[unit_id] + player = unit["agent_id"] + if action == "bomb": + position = pos_of(unit) + if unit["inventory"]["bombs"] > 0 and position not in bomb_positions: + unit["inventory"]["bombs"] -= 1 + bomb_positions.add(position) + bombs.append( + {"owner": player, "unit_id": unit_id, "pos": position, "timer": BOMB_TIMER, "radius": BLAST_RADIUS} + ) + else: + stats[player]["invalid_actions"] += 1 + elif action == "detonate": + for index, bomb in enumerate(list(bombs)): + if bomb["owner"] == player and (target is None or bomb["pos"] == target): + explode_bomb(index, bombs, metal, wood, units, stats, blasts) + break + + proposals = {} + for unit_id, (action, _target) in actions_by_unit.items(): + unit = units[unit_id] + if unit["hp"] <= 0 or action not in DELTAS: + continue + dx, dy = DELTAS[action] + current = pos_of(unit) + proposed = (current[0] + dx, current[1] + dy) + player = unit["agent_id"] + if proposed in metal or proposed in wood or proposed in bomb_positions: + stats[player]["invalid_actions"] += 1 + continue + if not (0 <= proposed[0] < width and 0 <= proposed[1] < height): + stats[player]["invalid_actions"] += 1 + continue + proposals[unit_id] = proposed + + target_counts = defaultdict(int) + for target in proposals.values(): + target_counts[target] += 1 + + occupied = {pos_of(unit): unit_id for unit_id, unit in units.items() if unit["hp"] > 0} + for unit_id, target in proposals.items(): + player = units[unit_id]["agent_id"] + if target_counts[target] > 1: + stats[player]["invalid_actions"] += 1 + continue + occupant = occupied.get(target) + if occupant is not None: + stats[player]["invalid_actions"] += 1 + continue + units[unit_id]["coordinates"] = list(target) + + +def tick_bombs(bombs, metal, wood, units, stats, blasts): + index = 0 + while index < len(bombs): + bombs[index]["timer"] -= 1 + if bombs[index]["timer"] <= 0: + explode_bomb(index, bombs, metal, wood, units, stats, blasts) + else: + index += 1 + + +def player_score(player, agents, units, stats): + alive_units = [units[unit_id] for unit_id in agents[player]["unit_ids"] if units[unit_id]["hp"] > 0] + alive_hp = sum(unit["hp"] for unit in alive_units) + return ( + alive_hp * 30 + + len(alive_units) * 20 + + stats[player]["damage_dealt"] * 120 + + stats[player]["kills"] * 300 + + stats[player]["wood_destroyed"] * 40 + - stats[player]["invalid_actions"] + - stats[player]["agent_errors"] * 10 + ) + + +def run_game(players, callbacks, seed, ticks, width, height, unit_count, agent_timeout): + rng = random.Random(seed) + metal, wood = build_map(width, height, unit_count, rng) + units, agents = make_units(players, width, height, unit_count) + bombs = [] + blasts = {} + stats = { + player: { + "damage_dealt": 0, + "kills": 0, + "wood_destroyed": 0, + "invalid_actions": 0, + "agent_errors": 0, + "bombs_returned": 0, + } + for player in players + } + + final_tick = 0 + for tick in range(ticks): + final_tick = tick + blasts = {pos: ttl - 1 for pos, ttl in blasts.items() if ttl > 1} + apply_actions( + players, callbacks, agents, units, metal, wood, bombs, blasts, stats, width, height, tick, agent_timeout + ) + tick_bombs(bombs, metal, wood, units, stats, blasts) + if len(live_players(players, units)) <= 1: + break + + scores = {player: float(player_score(player, agents, units, stats)) for player in players} + best_score = max(scores.values()) + winners = [player for player, score in scores.items() if score == best_score] + winner = winners[0] if len(winners) == 1 else "TIE" + detail = { + "seed": seed, + "ticks": final_tick + 1, + "winner": winner, + "scores": scores, + "alive_units": { + player: sum(1 for unit_id in agents[player]["unit_ids"] if units[unit_id]["hp"] > 0) for player in players + }, + "alive_hp": { + player: sum(max(units[unit_id]["hp"], 0) for unit_id in agents[player]["unit_ids"]) for player in players + }, + "stats": stats, + } + return scores, detail + + +def parse_agent_arg(raw): + if "=" not in raw: + raise argparse.ArgumentTypeError("--agent must use NAME=/path/to/bomberland_agent.py") + name, path = raw.split("=", 1) + if not name or not path: + raise argparse.ArgumentTypeError("--agent must include both NAME and path") + return name, path + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--sims", type=int, required=True) + parser.add_argument("--ticks", type=int, required=True) + parser.add_argument("--width", type=int, required=True) + parser.add_argument("--height", type=int, required=True) + parser.add_argument("--unit-count", type=int, required=True) + parser.add_argument("--agent-timeout", type=float, default=0.25) + parser.add_argument("--output", required=True) + parser.add_argument("--agent", action="append", type=parse_agent_arg, required=True) + args = parser.parse_args() + + if len(args.agent) != 2: + raise ValueError("Bomberland currently expects exactly two --agent entries") + if args.sims % 2 != 0: + raise ValueError("--sims must be even so both players get paired starting sides") + + players = [name for name, _path in args.agent] + callbacks = {name: path for name, path in args.agent} + totals = {player: 0.0 for player in players} + details = [] + for sim in range(args.sims): + sim_players = players if sim % 2 == 0 else list(reversed(players)) + scores, detail = run_game( + sim_players, + callbacks, + seed=100_000 + sim, + ticks=args.ticks, + width=args.width, + height=args.height, + unit_count=args.unit_count, + agent_timeout=args.agent_timeout, + ) + for player, score in scores.items(): + totals[player] += score + detail["sim"] = sim + detail["player_order"] = sim_players + details.append(json.dumps(detail, sort_keys=True)) + + result = { + "average_scores": {player: totals[player] / args.sims for player in players}, + "total_scores": totals, + "sims": args.sims, + "details": details, + } + output = Path(args.output) + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n") + + +if __name__ == "__main__": + main() diff --git a/configs/examples/Bomberland__dummy__r1__s2.yaml b/configs/examples/Bomberland__dummy__r1__s2.yaml new file mode 100644 index 0000000..a4e646b --- /dev/null +++ b/configs/examples/Bomberland__dummy__r1__s2.yaml @@ -0,0 +1,36 @@ +tournament: + rounds: 1 +game: + name: Bomberland + sims_per_round: 2 + args: + ticks: 40 + width: 11 + height: 11 + unit_count: 3 + agent_timeout: 0.25 + timeout: 180 +players: +- agent: dummy + name: alpha +- agent: dummy + name: beta +prompts: + game_description: |- + You are a software developer ({{player_id}}) competing in CodeClash's Bomberland arena. + + The game is played in {{total_rounds}} rounds. For every round, you and your competitor edit + code that controls a squad in a Bomberman-style grid arena. This is round {{round}}. + + Your task: improve `bomberland_agent.py`, located in {{working_dir}}. + All commands run from {{working_dir}}. + + Your file must define `next_actions(game_state)`. A valid starting point is: + + def next_actions(game_state): + agent_id = game_state["connection"]["agent_id"] + return {unit_id: "stay" for unit_id in game_state["agents"][agent_id]["unit_ids"]} + + Valid actions are "up", "down", "left", "right", "bomb", and "stay". + The arena runs multiple seeded Bomberland simulations. Your objective is to maximize average + score by surviving, damaging enemy units, destroying blocks, and eliminating enemy units. diff --git a/docs/reference/arenas/bomberland.md b/docs/reference/arenas/bomberland.md new file mode 100644 index 0000000..77f5a9e --- /dev/null +++ b/docs/reference/arenas/bomberland.md @@ -0,0 +1,87 @@ +# Bomberland + +Bomberman-style multi-agent arena based on Coder One's Bomberland competition. + +## Overview + +Bomberland is a grid-world arena where agents control several units, move around indestructible and +destructible blocks, place timed bombs, and try to outscore the opponent through survival, damage, +kills, and block destruction. + +The upstream Bomberland project uses a TypeScript websocket engine and starter-kit agents. The +CodeClash adapter keeps a pinned upstream checkout in the Docker image for provenance and starter-kit +reference, while using a compact deterministic Python runtime for CodeClash tournament execution. +This avoids requiring Docker Compose inside the arena container while preserving the same core agent +shape: submitted code receives a game-state dictionary and returns one action per controlled unit. + +## Resources + +- [Bomberland GitHub Repository](https://github.com/CoderOneHQ/bomberland) + +## Implementation + +::: codeclash.arenas.bomberland.bomberland.BomberlandArena + options: + show_root_heading: true + heading_level: 2 + +## Agent Interface + +Your bot must be a Python file named `bomberland_agent.py` that defines `next_actions`. + +```python +def next_actions(game_state): + agent_id = game_state["connection"]["agent_id"] + unit_ids = game_state["agents"][agent_id]["unit_ids"] + return {unit_id: "stay" for unit_id in unit_ids} +``` + +Valid string actions are `up`, `down`, `left`, `right`, `bomb`, and `stay`. The runtime also accepts +dictionary move actions such as `{"type": "move", "move": "up"}` for compatibility with common +starter-kit styles. + +## Configuration Example + +```yaml +tournament: + rounds: 1 +game: + name: Bomberland + sims_per_round: 2 + args: + ticks: 40 + width: 11 + height: 11 + unit_count: 3 +players: + - agent: dummy + name: alpha + - agent: dummy + name: beta +``` + +## Scoring + +The arena runs `sims_per_round` deterministic seeded games. `sims_per_round` must be even so each +player receives both starting sides for paired seeds. Each player receives an average score computed +from surviving health, surviving units, enemy damage, kills, destroyed blocks, invalid actions, and +agent runtime errors. + +Smoke command: + +```bash +uv run python main.py configs/examples/Bomberland__dummy__r1__s2.yaml -o /tmp/codeclash-bomberland-smoke +``` + +The arena writes `bomberland_results.json` with this shape: + +```json +{ + "average_scores": {"alpha": 330.0, "beta": 330.0}, + "total_scores": {"alpha": 660.0, "beta": 660.0}, + "sims": 2, + "details": ["... per-simulation JSON strings ..."] +} +``` + +--8<-- "docs/_footer.md" diff --git a/docs/reference/index.md b/docs/reference/index.md index 9009735..8a31f0a 100644 --- a/docs/reference/index.md +++ b/docs/reference/index.md @@ -17,6 +17,7 @@ Game environments where agents compete. Each arena implements the `CodeGame` abs Available arenas: - [BattleCode](arenas/battlecode.md) - [BattleSnake](arenas/battlesnake.md) +- [Bomberland](arenas/bomberland.md) - [CoreWar](arenas/corewar.md) - [CybORG](arenas/cyborg.md) - [Halite](arenas/halite.md), [Halite II](arenas/halite2.md), [Halite III](arenas/halite3.md) diff --git a/mkdocs.yml b/mkdocs.yml index 1024a13..8278d81 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -72,6 +72,7 @@ nav: - "CodeGame (Abstract)": "reference/arenas/game.md" - "BattleCode": "reference/arenas/battlecode.md" - "BattleSnake": "reference/arenas/battlesnake.md" + - "Bomberland": "reference/arenas/bomberland.md" - "Bridge": "reference/arenas/bridge.md" - "CoreWar": "reference/arenas/corewar.md" - "CybORG": "reference/arenas/cyborg.md" diff --git a/tests/arenas/test_bomberland.py b/tests/arenas/test_bomberland.py new file mode 100644 index 0000000..a6bd209 --- /dev/null +++ b/tests/arenas/test_bomberland.py @@ -0,0 +1,311 @@ +import importlib.util +import json +import time +from pathlib import Path + +import pytest + +from codeclash.arenas import get_arena +from codeclash.arenas.arena import RoundStats +from codeclash.arenas.bomberland.bomberland import CRASH_SCORE, BomberlandArena +from codeclash.constants import RESULT_TIE + +from .conftest import MockEnvironment, MockPlayer + + +def load_runtime_module(): + runtime_path = Path(__file__).parents[2] / "codeclash/arenas/bomberland/runtime/run_bomberland.py" + spec = importlib.util.spec_from_file_location("run_bomberland_test", runtime_path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class TestBomberlandValidation: + def test_valid_agent(self, mock_player_factory): + arena = BomberlandArena.__new__(BomberlandArena) + arena.submission = "bomberland_agent.py" + arena.config = {"game": {"name": "Bomberland", "sims_per_round": 1}} + player = mock_player_factory( + name="Alice", + files={"bomberland_agent.py": "def next_actions(game_state):\n return {}\n"}, + command_outputs={ + "test -f bomberland_agent.py && echo exists": {"output": "exists\n", "returncode": 0}, + "cat bomberland_agent.py": { + "output": "def next_actions(game_state):\n return {}\n", + "returncode": 0, + }, + "python -m py_compile bomberland_agent.py": {"output": "", "returncode": 0}, + "python - <<'PY'": {"output": "", "returncode": 0}, + }, + ) + + valid, error = arena.validate_code(player) + + assert valid is True + assert error is None + + def test_missing_next_actions(self, mock_player_factory): + arena = BomberlandArena.__new__(BomberlandArena) + arena.submission = "bomberland_agent.py" + arena.config = {"game": {"name": "Bomberland", "sims_per_round": 1}} + player = mock_player_factory( + name="Alice", + files={"bomberland_agent.py": "def choose_action(game_state):\n return {}\n"}, + command_outputs={ + "test -f bomberland_agent.py && echo exists": {"output": "exists\n", "returncode": 0}, + "cat bomberland_agent.py": { + "output": "def choose_action(game_state):\n return {}\n", + "returncode": 0, + }, + "python -m py_compile bomberland_agent.py": {"output": "", "returncode": 0}, + "python - <<'PY'": {"output": "next_actions callable not found", "returncode": 1}, + }, + ) + + valid, error = arena.validate_code(player) + + assert valid is False + assert "Could not import or call" in error + + def test_next_actions_wrong_return_type(self, mock_player_factory): + arena = BomberlandArena.__new__(BomberlandArena) + arena.submission = "bomberland_agent.py" + arena.config = {"game": {"name": "Bomberland", "sims_per_round": 1}} + player = mock_player_factory( + name="Alice", + files={"bomberland_agent.py": "def next_actions(game_state):\n return []\n"}, + command_outputs={ + "test -f bomberland_agent.py && echo exists": {"output": "exists\n", "returncode": 0}, + "cat bomberland_agent.py": { + "output": "def next_actions(game_state):\n return []\n", + "returncode": 0, + }, + "python -m py_compile bomberland_agent.py": {"output": "", "returncode": 0}, + "python - <<'PY'": {"output": "next_actions must return a dict or None", "returncode": 1}, + }, + ) + + valid, error = arena.validate_code(player) + + assert valid is False + assert "Could not import or call" in error + + def test_import_probe_uses_validation_timeout(self, mock_player_factory): + arena = BomberlandArena.__new__(BomberlandArena) + arena.submission = "bomberland_agent.py" + arena.config = {"game": {"name": "Bomberland", "sims_per_round": 1, "args": {"validation_timeout": 9}}} + + class CapturingEnvironment(MockEnvironment): + def __init__(self): + super().__init__( + files={"bomberland_agent.py": "def next_actions(game_state):\n return {}\n"}, + command_outputs={ + "python -m py_compile bomberland_agent.py": {"output": "", "returncode": 0}, + "python - <<'PY'": {"output": "", "returncode": 0}, + }, + ) + self.timeouts = [] + + def execute(self, cmd, cwd=None, timeout=None): + self.timeouts.append(timeout) + return super().execute(cmd, cwd=cwd, timeout=timeout) + + player = MockPlayer("Alice", CapturingEnvironment()) + + valid, error = arena.validate_code(player) + + assert valid is True + assert error is None + assert player.environment.timeouts[-1] == 9 + + +class TestBomberlandResults: + def test_parse_winner(self, tmp_log_dir): + arena = BomberlandArena.__new__(BomberlandArena) + arena.log_local = tmp_log_dir + arena.logger = type("Logger", (), {"error": lambda self, msg: None})() + round_dir = tmp_log_dir / "rounds" / "1" + round_dir.mkdir(parents=True) + (round_dir / "bomberland_results.json").write_text( + json.dumps( + { + "average_scores": {"Alice": 325.5, "Bob": 300.0}, + "details": ['{"sim": 0, "winner": "Alice"}'], + } + ) + ) + + agents = [MockPlayer("Alice"), MockPlayer("Bob")] + stats = RoundStats(round_num=1, agents=agents) + + arena.get_results(agents, 1, stats) + + assert stats.winner == "Alice" + assert stats.scores == {"Alice": 325.5, "Bob": 300.0} + assert stats.player_stats["Alice"].score == 325.5 + assert stats.details == ['{"sim": 0, "winner": "Alice"}'] + + def test_parse_tie(self, tmp_log_dir): + arena = BomberlandArena.__new__(BomberlandArena) + arena.log_local = tmp_log_dir + arena.logger = type("Logger", (), {"error": lambda self, msg: None})() + round_dir = tmp_log_dir / "rounds" / "1" + round_dir.mkdir(parents=True) + (round_dir / "bomberland_results.json").write_text(json.dumps({"average_scores": {"Alice": 10, "Bob": 10}})) + + agents = [MockPlayer("Alice"), MockPlayer("Bob")] + stats = RoundStats(round_num=1, agents=agents) + + arena.get_results(agents, 1, stats) + + assert stats.winner == RESULT_TIE + assert stats.scores == {"Alice": 10.0, "Bob": 10.0} + + def test_missing_player_uses_crash_score(self, tmp_log_dir): + arena = BomberlandArena.__new__(BomberlandArena) + arena.log_local = tmp_log_dir + arena.logger = type("Logger", (), {"error": lambda self, msg: None})() + round_dir = tmp_log_dir / "rounds" / "1" + round_dir.mkdir(parents=True) + (round_dir / "bomberland_results.json").write_text(json.dumps({"average_scores": {"Alice": -5}})) + + agents = [MockPlayer("Alice"), MockPlayer("Bob")] + stats = RoundStats(round_num=1, agents=agents) + + arena.get_results(agents, 1, stats) + + assert stats.winner == "Alice" + assert stats.scores == {"Alice": -5.0, "Bob": CRASH_SCORE} + + +class TestBomberlandExecution: + def test_execute_round_uses_nested_game_args(self): + arena = BomberlandArena.__new__(BomberlandArena) + arena.submission = "bomberland_agent.py" + arena.config = { + "game": { + "sims_per_round": 5, + "args": { + "ticks": 11, + "width": 13, + "height": 15, + "unit_count": 2, + "agent_timeout": 0.1, + "timeout": 17, + }, + } + } + arena.log_env = Path("/logs") + arena.logger = type("Logger", (), {"info": lambda self, msg: None, "error": lambda self, msg: None})() + + class CapturingEnvironment(MockEnvironment): + def __init__(self): + super().__init__() + self.timeout = None + + def execute(self, cmd, cwd=None, timeout=None): + self._executed_commands.append(cmd) + self.timeout = timeout + return {"output": "", "returncode": 0} + + arena.environment = CapturingEnvironment() + + arena.execute_round([MockPlayer("Alice"), MockPlayer("Bob")]) + + cmd = arena.environment._executed_commands[0] + assert "--sims 5" in cmd + assert "--ticks 11" in cmd + assert "--width 13" in cmd + assert "--height 15" in cmd + assert "--unit-count 2" in cmd + assert "--agent-timeout 0.1" in cmd + assert "--output /logs/bomberland_results.json" in cmd + assert "--agent Alice=/Alice/bomberland_agent.py" in cmd + assert "--agent Bob=/Bob/bomberland_agent.py" in cmd + assert arena.environment.timeout == 17 + + +class TestBomberlandRuntime: + def test_call_agent_times_out_submitted_code(self, tmp_path): + runtime = load_runtime_module() + agent_path = tmp_path / "bomberland_agent.py" + agent_path.write_text( + "def next_actions(game_state):\n" + " try:\n" + " while True:\n" + " pass\n" + " except BaseException:\n" + " while True:\n" + " pass\n" + ) + + start = time.perf_counter() + result = runtime.call_agent(str(agent_path), {}, 0.05) + elapsed = time.perf_counter() - start + + assert result == {"__error__": "Timeout"} + assert elapsed < 2 + + def test_call_agent_supports_sibling_imports(self, tmp_path): + runtime = load_runtime_module() + (tmp_path / "helper.py").write_text("ACTION = 'stay'\n") + agent_path = tmp_path / "bomberland_agent.py" + agent_path.write_text("from helper import ACTION\n\ndef next_actions(game_state):\n return {'u0': ACTION}\n") + + result = runtime.call_agent(str(agent_path), {}, 1) + + assert result == {"u0": "stay"} + + def test_malformed_dict_detonate_is_invalid(self): + runtime = load_runtime_module() + + assert runtime.normalize_action({"type": "detonate", "coordinates": ["bad", 0]}) == ("invalid", None) + + +def test_bomberland_registered(monkeypatch, minimal_config, tmp_log_dir): + config = { + **minimal_config, + "game": { + "name": "Bomberland", + "sims_per_round": 2, + }, + } + + monkeypatch.setattr(BomberlandArena, "build_image", lambda self: None) + monkeypatch.setattr(BomberlandArena, "get_environment", lambda self: MockEnvironment()) + + arena = get_arena(config, tournament_id="test", local_output_dir=tmp_log_dir) + + assert isinstance(arena, BomberlandArena) + + +def test_bomberland_rejects_non_two_player_configs(minimal_config, tmp_log_dir): + config = { + **minimal_config, + "game": { + "name": "Bomberland", + "sims_per_round": 1, + }, + "players": [ + {"name": "p1", "agent": "dummy"}, + {"name": "p2", "agent": "dummy"}, + {"name": "p3", "agent": "dummy"}, + ], + } + + with pytest.raises(ValueError, match="exactly two players"): + BomberlandArena(config, tournament_id="test", local_output_dir=tmp_log_dir) + + +def test_bomberland_rejects_odd_sim_counts(minimal_config, tmp_log_dir): + config = { + **minimal_config, + "game": { + "name": "Bomberland", + "sims_per_round": 3, + }, + } + + with pytest.raises(ValueError, match="even sims_per_round"): + BomberlandArena(config, tournament_id="test", local_output_dir=tmp_log_dir)