Added logging

Author: klieret

codeclash/agents/__init__.py

@@ -16,6 +16,7 @@ def get_agent(config: dict, game: CodeGame) -> Player:
         "game_name": game.name,
         "game_id": game.game_id,
         "rounds": game.rounds,
+        "round": 1,
         "player_id": config["name"],
         "game_description": game.config.get("description", ""),
     }

codeclash/agents/abstract.py

@@ -4,8 +4,9 @@
 from dotenv import load_dotenv
 from minisweagent import Environment
 
-from codeclash.constants import GH_ORG
+from codeclash.constants import DIR_LOGS, GH_ORG
 from codeclash.utils.environment import assert_zero_exit_code
+from codeclash.utils.log import get_logger
 
 load_dotenv()
 
@@ -21,6 +22,11 @@ def __init__(
         self.name = f"{template_vars['game_id']}_{config['name']}"
         self.environment = environment
         self.template_vars = template_vars
+        self.logger = get_logger(
+            self.name,
+            log_path=DIR_LOGS / template_vars["game_id"] / f"{self.name}.log",
+            emoji="👤",
+        )
 
     def commit(self):
         """Commit changes to the agent's codebase."""
@@ -30,11 +36,14 @@ def commit(self):
             f"git commit --allow-empty -m 'Round {self.round}/{rounds} Update'",
         ]:
             assert_zero_exit_code(self.environment.execute(cmd))
-        print(f"Committed changes for {self.name} for round {self.round}/{rounds}")
+        self.logger.info(
+            f"Committed changes for {self.name} for round {self.round}/{rounds}"
+        )
 
     def on_round_update(self, new_round: int):
         """Update the agent's round to match the game round."""
         self.round = new_round
+        self.template_vars["round"] = new_round
 
     def push(self):
         """Push codebase to a branch on the game's remote repository."""
@@ -43,7 +52,7 @@ def push(self):
             raise ValueError("GITHUB_TOKEN environment variable is required")
 
         for cmd in [
-            f"git remote remove origin",
+            "git remote remove origin",
             f"git remote add origin https://x-access-token:{token}@github.com/{GH_ORG}/{self.template_vars['game_name']}.git",
             f"git push origin {self.name}",
         ]:

codeclash/agents/minisweagent.py

@@ -1,3 +1,4 @@
+import logging
 import os
 import platform
 import traceback
@@ -31,19 +32,22 @@ def __init__(
         name: str,
         template_vars: dict,
         *,
+        logger: logging.Logger,
         config_class: Callable = AgentConfig,
         **kwargs,
     ):
         super().__init__(model, env, config_class=config_class, **kwargs)
         self.name = name
         self.template_vars = template_vars
         self.console = Console()
+        self.logger = logger
 
     def add_message(self, role: str, content: str, **kwargs):
         super().add_message(role, content, **kwargs)
+        self.logger.debug(f"[{role}] {content}", extra={"highlighter": None})
         if role == "assistant":
-            self.console.print(
-                f"[{self.name}] Step taken (step {self.model.n_calls}, cost {self.model.cost:.2f})"
+            self.logger.info(
+                f"Step taken (step {self.model.n_calls}, cost {self.model.cost:.2f})"
             )
 
     def render_template(self, template: str, **kwargs) -> str:
@@ -58,8 +62,7 @@ def render_template(self, template: str, **kwargs) -> str:
 
     def run(self) -> tuple[str, str]:
         """Run step() until agent is finished. Return exit status & message"""
-        with self.console.status(f"[bold green]{self.name} updating codebase..."):
-            return super().run(task="")
+        return super().run(task="")
 
 
 class MiniSWEAgent(Player):
@@ -75,6 +78,7 @@ def __init__(self, config: dict, environment: Environment, template_vars: dict):
             self.environment,
             self.name,
             template_vars,
+            logger=self.logger,
             **yaml.safe_load(Path(config["config"]).read_text())["agent"],
         )
 
@@ -93,7 +97,7 @@ def run(self):
             save_traj(
                 self.agent,  # type: ignore
                 DIR_LOGS
-                / f"{self.template_vars['game_id']}/{player_id}_r{self.round}.traj.json",
+                / f"{self.template_vars['game_id']}/{player_id}_r{self.template_vars['round']}.traj.json",
                 exit_status=exit_status,
                 result=result,
             )

codeclash/games/abstract.py

@@ -13,6 +13,7 @@
 from codeclash.constants import DIR_LOGS, DIR_WORK, GH_ORG
 from codeclash.games.utils import copy_between_containers, copy_file_from_container
 from codeclash.utils.environment import assert_zero_exit_code
+from codeclash.utils.log import get_logger
 
 
 class CodeGame(ABC):
@@ -27,6 +28,9 @@ def __init__(self, config: dict):
         self.round: int = 0
         self.game_id: str = f"{self.name}{uuid4().hex[:6]}"
         self.log_path: Path = (DIR_WORK / DIR_LOGS / self.game_id).resolve()
+        self.logger = get_logger(
+            self.name, log_path=DIR_LOGS / self.game_id / "game.log", emoji="🏓"
+        )
         self.environment: DockerEnvironment = self.get_environment()
         assert len(config["players"]) >= 2, "At least two players are required"
 
@@ -57,18 +61,20 @@ def build_image(self):
             text=True,
         )
         if result.returncode == 0:
-            print(f"✅ Built Docker image {self.image_name}")
+            self.logger.info(f"✅ Built Docker image {self.image_name}")
         else:
-            print(f"❌ Failed to build Docker image: {result.stderr}")
+            self.logger.error(
+                f"❌ Failed to build Docker image: {result.stderr}\n{result.stdout}{result.stderr}"
+            )
             raise
 
     def end(self, cleanup: bool = False):
-        print(Counter([x[1] for x in self.scoreboard]))
+        self.logger.info(Counter([x[1] for x in self.scoreboard]))
         if cleanup:
             for artifact in self.artifacts:
                 if artifact.exists():
                     subprocess.run(f"rm -rf {artifact}", shell=True)
-            print(f"🧼 Cleaned up {self.name} game")
+            self.logger.info(f"🧼 Cleaned up {self.name} game")
 
     def get_environment(self, branch_name: str | None = None) -> DockerEnvironment:
         """Get docker container ID with the game code installed."""
@@ -87,7 +93,7 @@ def get_environment(self, branch_name: str | None = None) -> DockerEnvironment:
             'git config --global user.name "Player"',
             "git config --global commit.gpgsign false",
         ]:
-            assert_zero_exit_code(environment.execute(cmd))
+            assert_zero_exit_code(environment.execute(cmd), logger=self.logger)
         return environment
 
     def _pre_round_setup(self, agents: list[Player]):
@@ -97,10 +103,11 @@ def _pre_round_setup(self, agents: list[Player]):
         for agent in agents:
             if hasattr(agent, "on_round_update"):
                 agent.on_round_update(self.round)
-        print(f"▶️ Running {self.name} round {self.round}...")
+        self.logger.info(f"▶️ Running {self.name} round {self.round}...")
 
         # Copy agent codebases into game's container
         for agent in agents:
+            self.logger.debug(f"Copying {agent.name}'s codebase")
             copy_between_containers(
                 src_container=agent.environment,
                 dest_container=self.environment,
@@ -109,8 +116,12 @@ def _pre_round_setup(self, agents: list[Player]):
             )
 
         # Ensure the log path + file exists
-        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}"))
+        assert_zero_exit_code(
+            self.environment.execute(f"mkdir -p {self.log_path}"), 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:
@@ -132,11 +143,11 @@ def _post_round_setup(self, agents: list[Player]):
                     f"{agent.environment.config.cwd}/logs/round_{self.round}.log",
                 )
             except Exception:
-                print(
+                self.logger.error(
                     f"Error copying round log to {agent.name}'s container: {traceback.format_exc()}"
                 )
             else:
-                print(f"Copied round log to {agent.name}'s container.")
+                self.logger.info(f"Copied round log to {agent.name}'s container.")
 
             try:
                 copy_file_from_container(
@@ -145,14 +156,14 @@ def _post_round_setup(self, agents: list[Player]):
                     DIR_LOGS / f"{self.game_id}/round_{self.round}.log",
                 )
             except Exception:
-                print(
+                self.logger.error(
                     f"Error copying round log to {agent.name}'s container: {traceback.format_exc()}"
                 )
             else:
-                print(
+                self.logger.info(
                     f"Copied round log from {agent.name}'s container to local log dir."
                 )
-        print(f"Round {self.round} completed.")
+        self.logger.info(f"Round {self.round} completed.")
 
     def run_round(self, agents: list[Player]):
         """
@@ -164,7 +175,7 @@ def run_round(self, agents: list[Player]):
         self.execute_round(agents)
         self.determine_winner(agents)
         last_winner = self.scoreboard[-1][1]
-        print(f"Round {self.round} winner: {last_winner}")
+        self.logger.info(f"Round {self.round} winner: {last_winner}")
         self._post_round_setup(agents)
 
     @property

codeclash/games/battlesnake/main.py

@@ -40,7 +40,7 @@ def execute_round(self, agents: list[Player]):
 
         cmd.append(f"-o {self.round_log_path}")
         cmd = " ".join(cmd)
-        print(f"Running command: {cmd}")
+        self.logger.info(f"Running command: {cmd}")
 
         # todo: should probably keep output somewhere?
         try:

codeclash/utils/environment.py

@@ -1,5 +1,12 @@
-def assert_zero_exit_code(result: dict) -> dict:
+import logging
+
+
+def assert_zero_exit_code(
+    result: dict, *, logger: logging.Logger | None = None
+) -> dict:
     if result.get("returncode", 0) != 0:
         msg = f"Command failed with exit code {result.get('returncode')}:\n{result.get('output')}"
+        if logger is not None:
+            logger.error(msg)
         raise RuntimeError(msg)
     return result

codeclash/utils/log.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+from rich.console import Console
+from rich.text import Text
+
+_STREAM_LEVEL = logging.DEBUG
+_FILE_LEVEL = logging.DEBUG
+
+# logging.getLogger().setLevel(logging.DEBUG)
+
+
+class RichFormatter(logging.Formatter):
+    """Custom formatter that uses Rich for colorized output."""
+
+    def __init__(self, console: Console | None = None, emoji: str = ""):
+        super().__init__()
+        self.console = console or Console()
+        self.emoji = emoji + " " if emoji and not emoji.endswith(" ") else emoji
+        self.level_colors = {
+            logging.DEBUG: "dim cyan",
+            logging.INFO: "green",
+            logging.WARNING: "yellow",
+            logging.ERROR: "red",
+            logging.CRITICAL: "bold red",
+        }
+
+    def format(self, record: logging.LogRecord) -> str:
+        level_color = self.level_colors.get(record.levelno, "white")
+        level_name = record.levelname.replace("WARNING", "WARN")
+
+        # Calculate the prefix length for indentation
+        prefix = f"{self.emoji}{level_name} [{record.name}] "
+        indent = " " * len(prefix)
+
+        text = Text()
+        text.append(self.emoji, style="white")
+        text.append(f"{level_name} ", style=level_color)
+        text.append(f"[{record.name}] ", style="dim")
+
+        # Handle multiline messages by indenting continuation lines
+        message = record.getMessage()
+        lines = message.split("\n")
+        text.append(lines[0], style="white")
+
+        for line in lines[1:]:
+            text.append("\n")
+            text.append(indent, style="dim")
+            text.append(line, style="white")
+
+        if record.exc_info:
+            text.append("\n")
+            exception_lines = self.formatException(record.exc_info).split("\n")
+            for i, exc_line in enumerate(exception_lines):
+                if i > 0:
+                    text.append("\n")
+                    text.append(indent, style="dim")
+                text.append(exc_line, style="red")
+
+        with self.console.capture() as capture:
+            self.console.print(text)
+        return capture.get().rstrip()
+
+
+def get_logger(
+    name: str, *, emoji: str = "", log_path: Path | None = None
+) -> logging.Logger:
+    """Get logger. Use this instead of `logging.getLogger` to ensure
+    that the logger is set up with the correct handlers.
+    """
+    logger = logging.getLogger(name)
+    if logger.handlers:
+        # Already set up
+        return logger
+
+    console = Console()
+    handler = logging.StreamHandler()
+    formatter = RichFormatter(console, emoji=emoji)
+
+    handler.setFormatter(formatter)
+    handler.setLevel(_STREAM_LEVEL)
+
+    # Set to lowest level and only use stream handlers to adjust levels
+    logger.setLevel(1)
+    logger.addHandler(handler)
+    logger.propagate = True
+
+    if log_path is not None:
+        log_path.parent.mkdir(parents=True, exist_ok=True)
+        file_handler = logging.FileHandler(log_path)
+        file_handler.setLevel(_FILE_LEVEL)
+
+        # Use a standard formatter for file logs with time, name, and level
+        file_formatter = logging.Formatter(
+            fmt="%(asctime)s [%(name)s] %(levelname)s %(message)s",
+            datefmt="%Y-%m-%d %H:%M:%S",
+        )
+        file_handler.setFormatter(file_formatter)
+
+        logger.addHandler(file_handler)
+
+    return logger