Class CoreWarArena

codeclash/arenas/corewar/corewar.py:13–107

COREWAR_LOG = "sim_{idx}.log"
TRACE_RAW = "trace_raw.txt"


class CoreWarArena(CodeArena):
    name: str = "CoreWar"
    description: str = """CoreWar is a programming battle where you write "warriors" in an assembly-like language called Redcode to compete within a virtual machine (MARS), aiming to eliminate your rivals by making their code self-terminate.
Victory comes from crafting clever tactics—replicators, scanners, bombers—that exploit memory layout and instruction timing to control the core.

Reading the logs: each round's score is computed over many simulated battles, but only a sample of them are saved as replay traces (`sim_*.jsonl`) in `/logs/` due to storage limits. The score reflects every battle played, not just the replayed subset, so treat the replays as representative examples rather than the full basis of the result."""
    submission: str = "warrior.red"

    def __init__(self, config, **kwargs):
        super().__init__(config, **kwargs)
        # Always run in brief mode (`-b`). Without it, pmars prints disassembled listing of
        # every warrior, which leaks opponent
        self.run_cmd_round: str = "./src/pmars -b"
        for arg, val in self.game_config.get("args", self.default_args).items():
            if isinstance(val, bool):
                if val:
                    self.run_cmd_round += f" -{arg}"
            else:
                self.run_cmd_round += f" -{arg} {val}"

    def _record_count(self) -> int:
        # Bounded subset of scored battles to record for replay (recording is the costly part).
        sims = self.game_config["sims_per_round"]
        return max(1, min(sims, self.game_config.get("record_battles", 100)))

    def _run_single_simulation(self, agents: list[Player], idx: int):
        # Shift agents by idx to vary starting positions
        agents = agents[idx:] + agents[:idx]
        args = [f"/{agent.name}/{self.submission}" for agent in agents]
        n = self.game_config["sims_per_round"]
        log = self.log_env / COREWAR_LOG.format(idx=idx)
        if idx == 0:
            # Record R scored battles with -T + the rest plain; both score blocks append to one
            # log (get_results sums them), so replays are genuine scored battles. i == agents[i].
            r = self._record_count()
            self._trace_agent_names = [agent.name for agent in agents]
            parts = [f"{self.run_cmd_round} {shlex.join(args)} -r {r} -T {self.log_env / TRACE_RAW} >> {log}"]
            if n - r > 0:
                parts.append(f"{self.run_cmd_round} {shlex.join(args)} -r {n - r} >> {log}")
            cmd = f"rm -f {log}; " + "; ".join(parts) + ";"
        else:
            cmd = f"{self.run_cmd_round} {shlex.join(args)} -r {n} > {log};"
        self.logger.info(f"Running game: {cmd}")
        response = self.environment.execute(cmd)
        assert response["returncode"] == 0, response

    def execute_round(self, agents: list[Player]):
        with ThreadPoolExecutor(self.game_config.get("sim_concurrency", 4)) as executor:
            futures = [executor.submit(self._run_single_simulation, agents, idx) for idx in range(len(agents))]
            for future in as_completed(futures):
                future.result()

    def copy_logs_from_env(self, round_num: int) -> None:
        # Distill the -T battles into sim_{i}.jsonl + trace.md on the host, then drop the raw stream.
        super().copy_logs_from_env(round_num)
        raw = self.log_round(round_num) / TRACE_RAW