Add handling for submission validation

Author: john-b-yang

codeclash/games/battlecode/battlecode.py

@@ -1,8 +1,8 @@
 import random
 import re
 from pathlib import Path
-from typing import Any
 
+from codeclash.agents.player import Player
 from codeclash.constants import DIR_WORK, RESULT_TIE
 from codeclash.games.game import CodeGame, RoundStats
 
@@ -23,7 +23,19 @@ def __init__(self, config, *, tournament_id: str, local_output_dir: Path):
             else:
                 self.run_cmd_round += f" --{arg} {val}"
 
-    def get_results(self, agents: list[Any], round_num: int) -> RoundStats:
+    def execute_round(self, agents: list[Player]):
+        for agent in agents:
+            src, dest = f"/{agent.name}/src/mysubmission/", str(DIR_WORK / "src" / agent.name)
+            self.environment.execute(f"cp -r {src} {dest}")
+        random.shuffle(agents)  # Start position matters in BattleCode! Shuffle to be fair.
+        args = [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.logger.info(f"Running game: {cmd}")
+
+        response = self.environment.execute(cmd + f" > {self.log_env / BC_LOG}")
+        assert response["returncode"] == 0, response
+
+    def get_results(self, agents: list[Player], round_num: int, stats: RoundStats):
         winners = []
         with open(self.log_round(round_num) / BC_LOG) as f:
             lines = f.read().strip().split("\n")
@@ -39,19 +51,12 @@ def get_results(self, agents: list[Any], round_num: int) -> RoundStats:
             winners.append(winner)
         else:
             winners.append(RESULT_TIE)
-        return RoundStats(
-            winner=max(set(winners), key=winners.count),
-            scores={agent.name: winners.count(agent.name) for agent in agents},
-        )
 
-    def execute_round(self, agents: list[Any]):
-        for agent in agents:
-            src, dest = f"/{agent.name}/src/mysubmission/", str(DIR_WORK / "src" / agent.name)
-            self.environment.execute(f"cp -r {src} {dest}")
-        random.shuffle(agents)  # Start position matters in BattleCode! Shuffle to be fair.
-        args = [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.logger.info(f"Running game: {cmd}")
+        stats.winner = max(set(winners), key=winners.count)
+        stats.scores = {agent.name: winners.count(agent.name) for agent in agents}
+        for player, score in stats.scores.items():
+            stats.player_stats[player].score = score
 
-        response = self.environment.execute(cmd + f" > {self.log_env / BC_LOG}")
-        assert response["returncode"] == 0, response
+    def validate_code(self, agent: Player) -> tuple[bool, str | None]:
+        # TODO: implement more checks
+        return True, None

codeclash/games/battlesnake/battlesnake.py

@@ -38,18 +38,14 @@ def _wait_for_ports(self, ports: list[int], timeout: float = 3.0) -> None:
 
             time.sleep(0.1)
 
-    def get_results(self, agents: list[Player], round_num: int) -> RoundStats:
-        scores = {}
-        for idx in range(self.game_config["sims_per_round"]):
-            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] = scores.get(winner, 0) + 1
-
-        winner = max(scores, key=scores.get)
-        winner = RESULT_TIE if list(scores.values()).count(scores[winner]) > 1 else winner
-        return RoundStats(winner=winner, scores=scores)
+    def _run_single_simulation(self, cmd: str, idx: int) -> tuple[str, str]:
+        """Run a single battlesnake simulation and return log and result outputs."""
+        assert_zero_exit_code(
+            self.environment.execute(
+                cmd + f" -o {self.log_env / f'sim_{idx}.jsonl'}",
+                cwd=f"{self.environment.config.cwd}/game",
+            )
+        )
 
     def execute_round(self, agents: list[Player]):
         self.logger.debug("Starting game servers")
@@ -86,11 +82,23 @@ def execute_round(self, agents: list[Player]):
             # Kill all python servers when done
             self.environment.execute("pkill -f 'python main.py' || true")
 
-    def _run_single_simulation(self, cmd: str, idx: int) -> tuple[str, str]:
-        """Run a single battlesnake simulation and return log and result outputs."""
-        assert_zero_exit_code(
-            self.environment.execute(
-                cmd + f" -o {self.log_env / f'sim_{idx}.jsonl'}",
-                cwd=f"{self.environment.config.cwd}/game",
-            )
-        )
+    def get_results(self, agents: list[Player], round_num: int, stats: RoundStats):
+        scores = {}
+        for idx in range(self.game_config["sims_per_round"]):
+            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] = scores.get(winner, 0) + 1
+
+        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
+
+    def validate_code(self, agent: Player) -> tuple[bool, str | None]:
+        # TODO: implement more checks
+        return True, None

codeclash/games/corewar/corewar.py

@@ -21,7 +21,18 @@ def __init__(self, config, *, tournament_id: str, local_output_dir: Path):
             else:
                 self.run_cmd_round += f" -{arg} {val}"
 
-    def get_results(self, agents: list[Player], round_num: int) -> RoundStats:
+    def execute_round(self, agents: list[Player]):
+        args = [f"/{agent.name}/warriors/warrior.red" for agent in agents]
+        cmd = (
+            f"{self.run_cmd_round} {shlex.join(args)} "
+            f"-r {self.game_config['sims_per_round']} "
+            f"> {self.log_env / COREWAR_LOG};"
+        )
+        self.logger.info(f"Running game: {cmd}")
+        response = self.environment.execute(cmd)
+        assert response["returncode"] == 0, response
+
+    def get_results(self, agents: list[Player], round_num: int, stats: RoundStats):
         with open(self.log_round(round_num) / COREWAR_LOG) as f:
             result_output = f.read()
         self.logger.debug(f"Determining winner from result output: {result_output}")
@@ -44,22 +55,16 @@ def get_results(self, agents: list[Player], round_num: int) -> RoundStats:
         if scores:
             if len(scores) != len(agents):
                 self.logger.error(f"Have {len(scores)} scores but {len(agents)} agents")
-            return RoundStats(
-                winner=agents[scores.index(max(scores))].name,
-                scores={agent.name: score for agent, score in zip(agents, scores)},
-                details={"stdout": "\n".join(relevant_lines)},
-            )
+            stats.winner = agents[scores.index(max(scores))].name
+            stats.scores = {agent.name: score for agent, score in zip(agents, scores)}
         else:
             self.logger.debug("No scores found, returning unknown")
-            return RoundStats(winner="unknown", scores={agent.name: 0 for agent in agents})
+            stats.winner = "unknown"
+            stats.scores = {agent.name: 0 for agent in agents}
 
-    def execute_round(self, agents: list[Player]):
-        args = [f"/{agent.name}/warriors/warrior.red" for agent in agents]
-        cmd = (
-            f"{self.run_cmd_round} {shlex.join(args)} "
-            f"-r {self.game_config['sims_per_round']} "
-            f"> {self.log_env / COREWAR_LOG};"
-        )
-        self.logger.info(f"Running game: {cmd}")
-        response = self.environment.execute(cmd)
-        assert response["returncode"] == 0, response
+        for player, score in stats.scores.items():
+            stats.player_stats[player].score = score
+
+    def validate_code(self, agent: Player) -> tuple[bool, str | None]:
+        # TODO: implement more checks
+        return True, None

codeclash/games/dummy/dummy_game.py

@@ -10,7 +10,13 @@
 class DummyGame(CodeGame):
     name: str = "DummyGame"
 
-    def get_results(self, agents: list[Player], round_num: int) -> RoundStats:
+    def execute_round(self, agents: list[Player]) -> None:
+        args = [f"/{agent.name}/main.py" for agent in agents]
+        cmd = f"python engine.py {' '.join(args)} -r {self.game_config['sims_per_round']} > {self.log_env / DUMMY_LOG};"
+        self.logger.info(f"Running game: {cmd}")
+        assert_zero_exit_code(self.environment.execute(cmd))
+
+    def get_results(self, agents: list[Player], round_num: int, stats: RoundStats):
         with open(self.log_round(round_num) / DUMMY_LOG) as f:
             round_log = f.read()
         lines = round_log.split("FINAL_RESULTS")[-1].splitlines()
@@ -23,14 +29,11 @@ def get_results(self, agents: list[Player], round_num: int) -> RoundStats:
                 rounds_won = int(match.group(2))
                 scores[agents[int(bot_id) - 1].name] = rounds_won
 
-        return RoundStats(
-            winner=max(scores, key=scores.get) if scores else "unknown",
-            scores=scores,
-            details={"dummy": True},
-        )
+        stats.winner = max(scores, key=scores.get) if scores else "unknown"
+        stats.scores = scores
+        for player, score in scores.items():
+            stats.player_stats[player].score = score
 
-    def execute_round(self, agents: list[Player]) -> None:
-        args = [f"/{agent.name}/main.py" for agent in agents]
-        cmd = f"python engine.py {' '.join(args)} -r {self.game_config['sims_per_round']} > {self.log_env / DUMMY_LOG};"
-        self.logger.info(f"Running game: {cmd}")
-        assert_zero_exit_code(self.environment.execute(cmd))
+    def validate_code(self, agent: Player) -> tuple[bool, str | None]:
+        # TODO: implement more checks
+        return True, None

codeclash/games/game.py

@@ -6,22 +6,55 @@
 from typing import Any
 
 from minisweagent.environments.docker import DockerEnvironment
-from pydantic import BaseModel
 
 from codeclash.agents.player 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_from_container
 from codeclash.utils.log import get_logger
 
 
-class RoundStats(BaseModel):
+class PlayerStats:
+    def __init__(self, name: str):
+        self.name = name
+        self.invalid_reason: str | None = None
+        self.score: float | None = None
+        self.valid_submit = False
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "name": self.name,
+            "invalid_reason": self.invalid_reason,
+            "score": self.score,
+            "valid_submit": self.valid_submit,
+        }
+
+
+class RoundStats:
     winner: str
-    scores: dict[str, float]  # Map of player to game metric (e.g. # of wins, assets accumulated)
-    details: dict[str, Any] | None = None  # Optional, for game-specific info
+    round_num: int
+    scores: dict[str, float]
+    player_stats: dict[str, PlayerStats] | None = None
+
+    def __init__(self, round_num: int, agents: list[Player]):
+        self.winner = None
+        self.round_num = round_num
+        # Map of player to game metric (e.g. # of wins, assets accumulated)
+        self.scores: dict[str, float] = {}
+        self.player_stats: dict[str, PlayerStats] = {agent.name: PlayerStats(name=agent.name) for agent in agents}
 
     def __str__(self) -> str:
         return "\n".join([f"- Winner: {self.winner}", f"- Scores: {self.scores}"])
 
+    def to_dict(self) -> dict[str, Any]:
+        result = {
+            "round_num": self.round_num,
+            "winner": self.winner,
+            "scores": self.scores,
+        }
+        if self.player_stats:
+            result["player_stats"] = {name: stats.to_dict() for name, stats in self.player_stats.items()}
+        return result
+
 
 class CodeGame(ABC):
     name: str
@@ -177,20 +210,32 @@ def run_round(self, agents: list[Player], round_num: int) -> RoundStats:
         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.copy_logs_from_env(round_num)
-        return self.get_results(agents, round_num)
+        stats = RoundStats(round_num, agents)
+        validated = []
+        for agent in agents:
+            is_valid, error = self.validate_code(agent)
+            if not is_valid:
+                self.logger.warning(f"Agent {agent.name} failed verification: {error}")
+                stats.player_stats[agent.name].invalid_reason = error
+                continue
+            stats.player_stats[agent.name].valid_submit = True
+            validated.append(agent)
+
+        run_game = len(validated) > 1
+        if run_game:
+            self._pre_round_setup(validated)
+            self.execute_round(validated)
+            self.copy_logs_from_env(round_num)
+            self.get_results(validated, round_num, stats)
+        return stats
 
     @abstractmethod
-    def get_results(self, agents: list[Player], round_num: int) -> RoundStats:
+    def get_results(self, agents: list[Player], round_num: int, stats: RoundStats):
         """Determine the winner of the game based on the result output.
+        Modifies the stats object in place.
 
         Args:
             agents: List of agents participating in the round
-
-        Returns:
-            RoundStats object
         """
         pass
 
@@ -201,3 +246,16 @@ def execute_round(self, agents: list[Player]):
         includes the pre-round setup, post-round setup, and winner determination.
         """
         pass
+
+    @abstractmethod
+    def validate_code(self, agent: Player) -> tuple[bool, str | None]:
+        """Verify that the given agent can be run by the game.
+
+        Args:
+            agent: The agent to verify
+
+        Returns:
+            Boolean indicating whether the agent passed verification
+            Optional string indicating reason for failure
+        """
+        pass

codeclash/games/huskybench/huskybench.py

@@ -25,7 +25,21 @@ def __init__(self, config, *, tournament_id: str, local_output_dir: Path):
             else:
                 self.run_cmd_round += f" --{arg} {val}"
 
-    def get_results(self, agents: list[Player], round_num: int) -> RoundStats:
+    def execute_round(self, agents: list[Player]):
+        try:
+            cmd = f"{self.run_cmd_round} > {self.log_env / HB_LOG_ENGINE} &"
+            self.logger.debug(f"Starting game engine with command: {cmd}")
+            self.environment.execute(cmd)
+            for agent in agents:
+                cmd = f"python client/main.py --port 8000 > {self.log_env / f'{agent.name}.log'} &"
+                self.logger.debug(f"Adding agent with command: {cmd}")
+                self.environment.execute(cmd, cwd=f"/{agent.name}")
+        finally:
+            # Kill all python servers when done
+            self.environment.execute("pkill -f 'python client/main.py' || true")
+            self.environment.execute("pkill -f 'python engine/main.py' || true")
+
+    def get_results(self, agents: list[Player], round_num: int, stats: RoundStats):
         map_id_to_agent = {}
         for agent in agents:
             with open(self.log_round(round_num) / f"{agent.name}.log") as f:
@@ -41,19 +55,13 @@ def get_results(self, agents: list[Player], round_num: int) -> RoundStats:
             ]
             map_id_to_score = {k: v for k, v in score_updates[-self.num_players :]}
         self.logger.info("Final Scores: " + str(map_id_to_score))
-        agent_to_score = {map_id_to_agent[agent_id]: score for agent_id, score in map_id_to_score.items()}
-        return RoundStats(winner=max(agent_to_score, key=agent_to_score.get), scores=agent_to_score)
+        scores = {map_id_to_agent[agent_id]: score for agent_id, score in map_id_to_score.items()}
 
-    def execute_round(self, agents: list[Player]):
-        try:
-            cmd = f"{self.run_cmd_round} > {self.log_env / HB_LOG_ENGINE} &"
-            self.logger.debug(f"Starting game engine with command: {cmd}")
-            self.environment.execute(cmd)
-            for agent in agents:
-                cmd = f"python client/main.py --port 8000 > {self.log_env / f'{agent.name}.log'} &"
-                self.logger.debug(f"Adding agent with command: {cmd}")
-                self.environment.execute(cmd, cwd=f"/{agent.name}")
-        finally:
-            # Kill all python servers when done
-            self.environment.execute("pkill -f 'python client/main.py' || true")
-            self.environment.execute("pkill -f 'python engine/main.py' || true")
+        stats.winner = max(scores, key=scores.get)
+        stats.scores = scores
+        for player, score in scores.items():
+            stats.player_stats[player].score = score
+
+    def validate_code(self, agent: Player) -> tuple[bool, str | None]:
+        # TODO: implement more checks
+        return True, None