Merge branch 'main' of github.com:emagedoc/CodeClash

Author: john-b-yang

.gitignore

@@ -1,3 +1,9 @@
+# Project specific
+
+*.traj.json
+
+# -------------
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[codz]

codeclash/agents/abstract.py

@@ -5,6 +5,7 @@
 from minisweagent import Environment
 
 from codeclash.constants import GH_ORG
+from codeclash.utils.environment import assert_zero_exit_code
 
 load_dotenv()
 
@@ -29,7 +30,7 @@ def commit(self):
             "git add -A",
             f"git commit --allow-empty -m 'Round {self.round}/{rounds} Update'",
         ]:
-            self.environment.execute(cmd)
+            assert_zero_exit_code(self.environment.execute(cmd))
         print(f"Committed changes for {self.name} for round {self.round}/{rounds}")
 
     def push(self):
@@ -42,7 +43,7 @@ def push(self):
             f"git remote add origin https://x-access-token:{token}@github.com/{GH_ORG}/{self.name}.git",
             "git push -u origin main",
         ]:
-            self.environment.execute(cmd)
+            assert_zero_exit_code(self.environment.execute(cmd))
 
     @abstractmethod
     def run(self):

codeclash/games/abstract.py

@@ -8,8 +8,10 @@
 
 from minisweagent.environments.docker import DockerEnvironment
 
+from codeclash.agents.abstract import Player
 from codeclash.constants import DIR_LOGS, DIR_WORK, GH_ORG
 from codeclash.games.utils import copy_between_containers
+from codeclash.utils.environment import assert_zero_exit_code
 
 
 class CodeGame(ABC):
@@ -84,13 +86,10 @@ def get_environment(self, branch_name: str | None = None) -> DockerEnvironment:
             'git config --global user.name "Player"',
             "git config --global commit.gpgsign false",
         ]:
-            out = environment.execute(cmd)
-            if out.get("returncode", 0) != 0:
-                msg = f"Failed to execute command: {cmd}. Output so far:\n{out.get('output')}"
-                raise RuntimeError(msg)
+            assert_zero_exit_code(environment.execute(cmd))
         return environment
 
-    def _pre_round_setup(self, agents: list[Any]):
+    def _pre_round_setup(self, agents: list[Player]):
         """Copy agent codebases into game's container and make round log file"""
         self.round += 1
         print(f"▶️ Running {self.name} round {self.round}...")
@@ -105,20 +104,20 @@ def _pre_round_setup(self, agents: list[Any]):
             )
 
         # Ensure the log path + file exists
-        self.environment.execute(f"mkdir -p {self.log_path}")
-        self.environment.execute(f"touch {self.round_log_path}")
+        assert_zero_exit_code(self.environment.execute(f"mkdir -p {self.log_path}"))
+        assert_zero_exit_code(self.environment.execute(f"touch {self.round_log_path}"))
 
     @abstractmethod
-    def determine_winner(self, agents: list[Any]) -> Any:
+    def determine_winner(self, agents: list[Player]) -> Any:
         """Determine the winner of the game based on the round results, updates scoreboard"""
         pass
 
     @abstractmethod
-    def execute_round(self, agents: list[Any]):
+    def execute_round(self, agents: list[Player]):
         """Subclasses implement their game-specific logic here, must write results to round_log_path"""
         pass
 
-    def _post_round_setup(self, agents: list[Any]):
+    def _post_round_setup(self, agents: list[Player]):
         for agent in agents:
             copy_between_containers(
                 self.environment,
@@ -129,7 +128,7 @@ def _post_round_setup(self, agents: list[Any]):
             print(f"Copied round log to {agent.name}'s container.")
         print(f"Round {self.round} completed.")
 
-    def run_round(self, agents: list[Any]):
+    def run_round(self, agents: list[Player]):
         """
         Run a single round of the game with the given agents.
 

codeclash/games/battlesnake/main.py

@@ -1,8 +1,9 @@
 import json
 import time
-from typing import Any
 
+from codeclash.agents.abstract import Player
 from codeclash.games.abstract import CodeGame
+from codeclash.utils.environment import assert_zero_exit_code
 
 
 class BattleSnakeGame(CodeGame):
@@ -18,12 +19,14 @@ 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 -1 {self.round_log_path}")
+    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 execute_round(self, agents: list[Any]):
+    def execute_round(self, agents: list[Player]):
         cmd = []
         for idx, agent in enumerate(agents):
             port = 8001 + idx
@@ -39,12 +42,14 @@ def execute_round(self, agents: list[Any]):
         cmd = " ".join(cmd)
         print(f"Running command: {cmd}")
 
+        # todo: should probably keep output somewhere?
         try:
-            response = self.environment.execute(
-                f"{self.run_cmd_round} {cmd}",
-                cwd=f"{self.environment.config.cwd}/game",
+            assert_zero_exit_code(
+                self.environment.execute(
+                    f"{self.run_cmd_round} {cmd}",
+                    cwd=f"{self.environment.config.cwd}/game",
+                )
             )
-            assert response["returncode"] == 0, response
         finally:
             # Kill all python servers when done
             self.environment.execute("pkill -f 'python main.py' || true")

codeclash/games/corewar/main.py

@@ -1,6 +1,6 @@
 import re
-from typing import Any
 
+from codeclash.agents.abstract import Player
 from codeclash.games.abstract import CodeGame
 
 
@@ -17,7 +17,7 @@ def __init__(self, config):
             else:
                 self.run_cmd_round += f" -{arg} {val}"
 
-    def determine_winner(self, agents: list[Any]):
+    def determine_winner(self, agents: list[Player]):
         scores = []
         n = len(agents) * 2
         response = self.environment.execute(f"tail -{n} {self.round_log_path}")
@@ -28,7 +28,7 @@ def determine_winner(self, agents: list[Any]):
         winner = agents[scores.index(max(scores))].name
         self.scoreboard.append((self.round, winner))
 
-    def execute_round(self, agents: list[Any]):
+    def execute_round(self, agents: list[Player]):
         args = [f"/{agent.name}/warriors/warrior.red" for agent in agents]
         cmd = f"{self.run_cmd_round} {' '.join(args)} > {self.round_log_path}"
         print(f"Running command: {cmd}")

codeclash/games/robocode/main.py

@@ -1,6 +1,6 @@
 import subprocess
-from typing import Any
 
+from codeclash.agents.abstract import Player
 from codeclash.games.abstract import CodeGame
 from codeclash.games.utils import copy_file_to_container
 
@@ -51,12 +51,12 @@ def dict_to_lines(d, prefix=""):
         dict_to_lines(default_battle_config)
         return "\n".join(battle_lines)
 
-    def determine_winner(self, agents: list[Any]):
+    def determine_winner(self, agents: list[Player]):
         response = self.environment.execute(f"head -3 {self.round_log_path} | tail -1")
         winner = response["output"].split()[1].rsplit(".", 1)[0]
         self.scoreboard.append((self.round, winner))
 
-    def execute_round(self, agents: list[Any]):
+    def execute_round(self, agents: list[Player]):
         for agent in agents:
             # Copy the agent codebase into the game codebase and compile it
             for cmd in [

codeclash/games/robotrumble/main.py

@@ -1,5 +1,4 @@
-from typing import Any
-
+from codeclash.agents.abstract import Player
 from codeclash.constants import RESULT_TIE
 from codeclash.games.abstract import CodeGame
 
@@ -12,7 +11,7 @@ def __init__(self, config):
         assert len(config["players"]) == 2, "RobotRumble is a two-player game"
         self.run_cmd_round: str = "./rumblebot run term"
 
-    def determine_winner(self, agents: list[Any]):
+    def determine_winner(self, agents: list[Player]):
         response = self.environment.execute(f"tail -2 {self.round_log_path}")
         if "Blue won" in response["output"]:
             self.scoreboard.append((self.round, agents[0].name))
@@ -21,7 +20,7 @@ def determine_winner(self, agents: list[Any]):
         elif "it was a tie" in response["output"]:
             self.scoreboard.append((self.round, RESULT_TIE))
 
-    def execute_round(self, agents: list[Any]):
+    def execute_round(self, agents: list[Player]):
         args = [f"/{agent.name}/robot.py" for agent in agents]
         cmd = f"{self.run_cmd_round} {' '.join(args)} > {self.round_log_path}"
         print(f"Running command: {cmd}")

codeclash/games/utils.py

@@ -4,6 +4,8 @@
 
 from minisweagent.environments.docker import DockerEnvironment
 
+from codeclash.utils.environment import assert_zero_exit_code
+
 
 def copy_between_containers(
     src_container: DockerEnvironment,
@@ -25,15 +27,17 @@ def copy_between_containers(
             str(temp_path),
         ]
         result_src = subprocess.run(
-            cmd_src, check=True, stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL
+            cmd_src, check=False, capture_output=True, text=True
         )
         if result_src.returncode != 0:
             raise RuntimeError(
-                f"Failed to copy from {src_container.container_id} to local temp"
+                f"Failed to copy from {src_container.container_id} to local temp: {result_src.stdout}{result_src.stderr}"
             )
 
         # Ensure destination folder exists
-        dest_container.execute(f"mkdir -p {Path(dest_path).parent}")
+        assert_zero_exit_code(
+            dest_container.execute(f"mkdir -p {Path(dest_path).parent}")
+        )
 
         # Copy from temporary local directory to destination container
         cmd_dest = [
@@ -43,11 +47,11 @@ def copy_between_containers(
             f"{dest_container.container_id}:{dest_path}",
         ]
         result_dest = subprocess.run(
-            cmd_dest, check=True, stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL
+            cmd_dest, check=False, capture_output=True, text=True
         )
         if result_dest.returncode != 0:
             raise RuntimeError(
-                f"Failed to copy from local temp to {dest_container.container_id}"
+                f"Failed to copy from local temp to {dest_container.container_id}: {result_dest.stdout}{result_dest.stderr}"
             )
 
 
@@ -67,10 +71,8 @@ def copy_file_to_container(
         str(src_path),
         f"{container.container_id}:{dest_path}",
     ]
-    result = subprocess.run(
-        cmd, check=True, stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL
-    )
+    result = subprocess.run(cmd, check=False, capture_output=True, text=True)
     if result.returncode != 0:
         raise RuntimeError(
-            f"Failed to copy {src_path} to {container.container_id}:{dest_path}"
+            f"Failed to copy {src_path} to {container.container_id}:{dest_path}: {result.stdout}{result.stderr}"
         )

codeclash/utils/__init__.py

@@ -0,0 +1,5 @@
+def assert_zero_exit_code(result: dict) -> dict:
+    if result.get("returncode", 0) != 0:
+        msg = f"Command failed with exit code {result.get('returncode')}:\n{result.get('output')}"
+        raise RuntimeError(msg)
+    return result