WIP: Factored out tournament class from game class

Author: klieret

codeclash/agents/abstract.py

@@ -3,7 +3,7 @@
 from abc import ABC, abstractmethod
 
 from dotenv import load_dotenv
-from minisweagent import Environment
+from minisweagent.environments.docker import DockerEnvironment
 
 from codeclash.agents.utils import GameContext
 from codeclash.constants import GH_ORG
@@ -17,7 +17,7 @@ class Player(ABC):
     def __init__(
         self,
         config: dict,
-        environment: Environment,
+        environment: DockerEnvironment,
         game_context: GameContext,
     ) -> None:
         self.config = config

codeclash/games/__init__.py

@@ -1,3 +1,5 @@
+from pathlib import Path
+
 from codeclash.games.abstract import CodeGame
 from codeclash.games.battlecode.main import BattleCodeGame
 from codeclash.games.battlesnake.main import BattleSnakeGame
@@ -7,16 +9,17 @@
 
 
 # might consider postponing imports to avoid loading things we don't need
-def get_game(config: dict) -> CodeGame:
+def get_game(config: dict, *, tournament_id: str, local_output_dir: Path) -> CodeGame:
     game = {
-        x.name: x for x in [
+        x.name: x
+        for x in [
             BattleCodeGame,
             BattleSnakeGame,
             CoreWarGame,
             RoboCodeGame,
-            RobotRumbleGame
+            RobotRumbleGame,
         ]
     }.get(config["game"]["name"])
     if game is None:
         raise ValueError(f"Unknown game: {config['game']['name']}")
-    return game(config)
+    return game(config, tournament_id=tournament_id, local_output_dir=local_output_dir)

codeclash/games/abstract.py

@@ -1,47 +1,52 @@
-import getpass
 import json
 import os
 import subprocess
-import time
-import traceback
 from abc import ABC, abstractmethod
 from collections import Counter
 from pathlib import Path
-from typing import Any
 
 from minisweagent.environments.docker import DockerEnvironment
 
 from codeclash.agents.abstract import Player
 from codeclash.constants import DIR_LOGS, DIR_WORK, GH_ORG
-from codeclash.utils.environment import (
-    assert_zero_exit_code,
-    copy_between_containers,
-    copy_file_from_container,
-)
+from codeclash.utils.environment import assert_zero_exit_code, copy_between_containers
 from codeclash.utils.log import get_logger
 
 
 class CodeGame(ABC):
     name: str
 
-    def __init__(self, config: dict):
+    def __init__(self, config: dict, *, tournament_id: str, local_output_dir: Path):
+        """The CodeGame class is responsible for running games, i.e., taking a list of code
+        from different agents/players and running them against each other.
+        It also provides the environments for the game and agents to run in.
+
+        The central method is `run_round`, which takes a list of agents and returns the winner of the round.
+
+        At the end of the the tournament, run the `end` method to clean up the game and agents and write the metadata.
+        """
         self.url_gh: str = f"git@github.com:{GH_ORG}/{self.name}.git"
         self.artifacts: list[Path] = []
         """Artifact objects that we might want to clean up after the game."""
         self.scoreboard: list[tuple[int, str]] = []
         """List of (round number, winner (player id))"""
         self.game_config: dict = config["game"]
         self.config: dict = config
-        self.rounds: int = self.game_config.get("rounds", 1)
-        self.round: int = 0
-        self.game_id: str = f"{self.name}{time.strftime('%y%m%d%H%M%S')}"
+        self.game_id: str = tournament_id
         self.log_env: Path = (DIR_WORK / DIR_LOGS / self.game_id).resolve()
-        self.log_local: Path = (DIR_LOGS / getpass.getuser() / self.game_id).resolve()
+        self.log_local: Path = local_output_dir
         self.logger = get_logger(
             self.name, log_path=self.log_local / "game.log", emoji="🏓"
         )
         self.environment: DockerEnvironment = self.get_environment()
+        """The running docker environment for executing the game"""
         # assert len(config["players"]) >= 2, "At least two players are required"
+        """Total number of rounds to play"""
+        self._metadata: dict = {
+            "name": self.name,
+            "config": self.config,
+            "game_id": self.game_id,
+        }
 
     @property
     def image_name(self) -> str:
@@ -84,12 +89,7 @@ def get_metadata(self) -> dict:
         """This is what we write to metadata.json.
         You can subclass extend this to add more details for specific games.
         """
-        return {
-            "name": self.name,
-            "scoreboard": self.scoreboard,
-            "config": self.config,
-            "game_id": self.game_id,
-        }
+        return self._metadata
 
     def end(self, cleanup: bool = False):
         self.logger.info("Overall score: %s", Counter([x[1] for x in self.scoreboard]))
@@ -121,11 +121,7 @@ def get_environment(self, branch_name: str | None = None) -> DockerEnvironment:
         return environment
 
     def _pre_round_setup(self, agents: list[Player]):
-        """Copy agent codebases into game's container and make round log file"""
-        self.round += 1
-        # Notify agents of round update
-        self.logger.info(f"▶️ Running {self.name} round {self.round}...")
-
+        """Copy agent codebases into game's container"""
         # Copy agent codebases into game's container
         for agent in agents:
             self.logger.debug(f"Copying {agent.name}'s codebase")
@@ -136,78 +132,55 @@ def _pre_round_setup(self, agents: list[Player]):
                 dest_path=f"/{agent.name}",
             )
 
-        # Ensure the log path + file exists
+        # Ensure the log directory exists
         assert_zero_exit_code(
             self.environment.execute(f"mkdir -p {self.log_env}"),
             logger=self.logger,
         )
-        assert_zero_exit_code(
-            self.environment.execute(f"touch {self.round_log_path}"), logger=self.logger
-        )
 
     @abstractmethod
-    def determine_winner(self, agents: list[Player]) -> Any:
-        """Determine the winner of the game based on the round results,
-        Should update self.scoreboard
+    def determine_winner(
+        self, result_output: str, agents: list[Player]
+    ) -> dict[str, str]:
+        """Determine the winner of the game based on the result output.
+
+        Args:
+            result_output: The specific output containing winning information
+            agents: List of agents participating in the round
+
+        Returns:
+            Dictionary with key "winner" containing the winner's name
         """
         pass
 
     @abstractmethod
-    def execute_round(self, agents: list[Player]):
-        """Subclasses implement their game-specific logic here, must write results to round_log_path.
+    def execute_round(self, agents: list[Player]) -> dict[str, str]:
+        """Subclasses implement their game-specific logic here.
         This is the low level implementation, you probably want to use run_round instead, which
         includes the pre-round setup, post-round setup, and winner determination.
+
+        Returns:
+            Dictionary with keys "log_output" and "result_output"
         """
         pass
 
-    def _post_round_setup(self, agents: list[Player]):
-        for agent in agents:
-            try:
-                copy_between_containers(
-                    self.environment,
-                    agent.environment,
-                    self.round_log_path,
-                    f"{agent.environment.config.cwd}/logs/round_{self.round}.log",
-                )
-            except Exception:
-                self.logger.error(
-                    f"Error copying round log to {agent.name}'s container: {traceback.format_exc()}"
-                )
-            else:
-                self.logger.info(f"Copied round log to {agent.name}'s container.")
-
-            try:
-                copy_file_from_container(
-                    self.environment,
-                    self.round_log_path,
-                    self.log_local / self.round_log_path.name,
-                )
-            except Exception:
-                self.logger.error(
-                    f"Error copying round log to {agent.name}'s container: {traceback.format_exc()}"
-                )
-            else:
-                self.logger.info(
-                    f"Copied round log from {agent.name}'s container to local log dir."
-                )
-        self.logger.info(f"Round {self.round} completed.")
-
-    def run_round(self, agents: list[Player]):
+    def run_round(self, agents: list[Player]) -> dict[str, str]:
         """
         Run a single round of the game with the given agents.
 
-        Writes to directory containing logs and results of the round(s).
+        Returns the log output, result output, and winner name. All bookkeeping should be
+        handled by the tournament class.
         """
         self._pre_round_setup(agents)
-        self.execute_round(agents)
-        self.determine_winner(agents)
-        last_winner = self.scoreboard[-1][1]
-        self.logger.info(f"Round {self.round} winner: {last_winner}")
-        self._post_round_setup(agents)
+        result = self.execute_round(agents)
+        log_output = result["log_output"]
+        result_output = result["result_output"]
 
-    @property
-    def round_log_path(self) -> Path:
-        """
-        Get the path to the current round's log file.
-        """
-        return self.log_env / f"round_{self.round}.log"
+        winner_result = self.determine_winner(result_output, agents)
+        winner_name = winner_result["winner"]
+
+        return {
+            "log_output": log_output,
+            "result_output": result_output,
+            "winner": winner_name,
+        }

codeclash/games/battlecode/main.py

@@ -1,4 +1,5 @@
 import re
+from pathlib import Path
 from typing import Any
 
 from codeclash.constants import DIR_WORK, RESULT_TIE
@@ -8,8 +9,10 @@
 class BattleCodeGame(CodeGame):
     name: str = "BattleCode"
 
-    def __init__(self, config):
-        super().__init__(config)
+    def __init__(self, config, *, tournament_id: str, local_output_dir: Path):
+        super().__init__(
+            config, tournament_id=tournament_id, local_output_dir=local_output_dir
+        )
         assert len(config["players"]) == 2, "BattleCode is a two-player game"
         self.run_cmd_round: str = "python run.py run"
         for arg, val in self.game_config.get("args", {}).items():
@@ -19,13 +22,27 @@ def __init__(self, config):
             else:
                 self.run_cmd_round += f" --{arg} {val}"
 
-    def determine_winner(self, agents: list[Any]):
-        response = self.environment.execute(f"tail -3 {self.round_log_path} | head -1")
-        winner = re.search(r"\s\((.*)\)\swins\s\(", response["output"]).group(1)
-        winner = {"A": agents[0].name, "B": agents[1].name}.get(winner, RESULT_TIE)
-        self.scoreboard.append((self.round, winner))
+    def determine_winner(self, result_output: str, agents: list[Any]) -> dict[str, str]:
+        self.logger.debug(f"Determining winner from result output: {result_output}")
+        lines = result_output.strip().split("\n")
+        # Get the third-to-last line which contains the winner info
+        winner_line = lines[-3] if len(lines) >= 3 else ""
+        self.logger.debug(f"Winner line: {winner_line}")
+        match = re.search(r"\s\((.*)\)\swins\s\(", winner_line)
+        if match:
+            winner_key = match.group(1)
+            self.logger.debug(f"Winner key from match: {winner_key}")
+            # Map A/B to actual agent names (much closer to original code)
+            winner = {"A": agents[0].name, "B": agents[1].name}.get(
+                winner_key, RESULT_TIE
+            )
+            self.logger.debug(f"Concluding winner: {winner}")
+            return {"winner": winner}
+        else:
+            self.logger.debug("No winner match found, returning tie")
+            return {"winner": RESULT_TIE}
 
-    def execute_round(self, agents: list[Any]):
+    def execute_round(self, agents: list[Any]) -> dict[str, str]:
         for agent in agents:
             src, dest = f"/{agent.name}/src/mysubmission/", str(
                 DIR_WORK / "src" / agent.name
@@ -35,7 +52,10 @@ def execute_round(self, agents: list[Any]):
             f"--p{idx+1}-dir src --p{idx+1} {agent.name}"
             for idx, agent in enumerate(agents)
         ]
-        cmd = f"{self.run_cmd_round} {' '.join(args)} > {self.round_log_path}"
+        cmd = f"{self.run_cmd_round} {' '.join(args)}"
         self.logger.info(f"Running command: {cmd}")
         response = self.environment.execute(cmd)
         assert response["returncode"] == 0, response
+        # For BattleCode, log_output and result_output are the same
+        output = response["output"]
+        return {"log_output": output, "result_output": output}

codeclash/games/battlesnake/main.py

@@ -1,5 +1,6 @@
 import json
 import time
+from pathlib import Path
 
 from codeclash.agents.abstract import Player
 from codeclash.games.abstract import CodeGame
@@ -9,8 +10,10 @@
 class BattleSnakeGame(CodeGame):
     name: str = "BattleSnake"
 
-    def __init__(self, config):
-        super().__init__(config)
+    def __init__(self, config, *, tournament_id: str, local_output_dir: Path):
+        super().__init__(
+            config, tournament_id=tournament_id, local_output_dir=local_output_dir
+        )
         self.run_cmd_round: str = "./battlesnake play"
         for arg, val in self.game_config.get("args", {}).items():
             if isinstance(val, bool):
@@ -19,14 +22,19 @@ def __init__(self, config):
             else:
                 self.run_cmd_round += f" --{arg} {val}"
 
-    def determine_winner(self, agents: list[Player]):
-        response = assert_zero_exit_code(
-            self.environment.execute(f"tail -1 {self.round_log_path}")
-        )
-        winner = json.loads(response["output"].strip("\n"))["winnerName"]
-        self.scoreboard.append((self.round, winner))
+    def determine_winner(
+        self, result_output: str, agents: list[Player]
+    ) -> dict[str, str]:
+        self.logger.debug(f"Determining winner from result output: {result_output}")
+        lines = result_output.strip().split("\n")
+        # Get the last line which contains the game result
+        last_line = lines[-1] if lines else ""
+        self.logger.debug(f"Last line: {last_line}")
+        winner = json.loads(last_line)["winnerName"]
+        self.logger.debug(f"Concluding winner: {winner}")
+        return {"winner": winner}
 
-    def execute_round(self, agents: list[Player]):
+    def execute_round(self, agents: list[Player]) -> dict[str, str]:
         cmd = []
         for idx, agent in enumerate(agents):
             port = 8001 + idx
@@ -38,18 +46,27 @@ def execute_round(self, agents: list[Player]):
 
         time.sleep(3)  # Give servers time to start
 
-        cmd.append(f"-o {self.round_log_path}")
-        cmd = " ".join(cmd)
-        self.logger.info(f"Running command: {cmd}")
+        # Create temporary output file for results
+        output_file = f"battlesnake_output_{int(time.time())}.json"
+        cmd_str = " ".join(cmd) + f" -o {output_file}"
+        self.logger.info(f"Running command: {self.run_cmd_round} {cmd_str}")
 
-        # todo: should probably keep output somewhere?
         try:
-            assert_zero_exit_code(
+            response = assert_zero_exit_code(
                 self.environment.execute(
-                    f"{self.run_cmd_round} {cmd}",
+                    f"{self.run_cmd_round} {cmd_str}",
                     cwd=f"{self.environment.config.cwd}/game",
                 )
             )
+
+            # Read the output file for result information
+            result_response = self.environment.execute(f"cat game/{output_file}")
+            result_output = result_response["output"]
+
+            # Clean up the output file
+            self.environment.execute(f"rm -f game/{output_file}")
+
+            return {"log_output": response["output"], "result_output": result_output}
         finally:
             # Kill all python servers when done
             self.environment.execute("pkill -f 'python main.py' || true")