Add replay functionality to corewar. Bug fix robocode

Author: john-b-yang

codeclash/arenas/corewar/corewar.py

@@ -5,36 +5,55 @@
 
 from codeclash.agents.player import Player
 from codeclash.arenas.arena import CodeArena, RoundStats
+from codeclash.arenas.corewar.trace import distill_trace
 from codeclash.constants import RESULT_TIE
 
 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."""
+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)
-        self.run_cmd_round: str = "./src/pmars"
+        # 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]
-        cmd = (
-            f"{self.run_cmd_round} {shlex.join(args)} "
-            f"-r {self.game_config['sims_per_round']} "
-            f"> {self.log_env / COREWAR_LOG.format(idx=idx)};"
-        )
+        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
@@ -45,28 +64,51 @@ def execute_round(self, agents: list[Player]):
             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
+        if not raw.exists():
+            return
+        try:
+            battles = distill_trace(raw, self.log_round(round_num), getattr(self, "_trace_agent_names", []))
+            if battles:
+                dropped = sum(b.cells_dropped for b in battles)
+                note = f" ({dropped} cell-changes sampled out for size)" if dropped else ""
+                self.logger.info(f"CoreWar trace distilled: {len(battles)} replay(s) for round {round_num}{note}")
+            else:
+                self.logger.warning("CoreWar trace was empty; no replay for this round")
+        except Exception as e:
+            self.logger.warning(f"Failed to distill CoreWar trace: {e}")
+        finally:
+            raw.unlink(missing_ok=True)
+
     def get_results(self, agents: list[Player], round_num: int, stats: RoundStats):
         scores, wins = defaultdict(int), defaultdict(int)
+        score_pat = re.compile(r".*\sby\s.*\sscores\s(\d+)")
         for idx in range(len(agents)):
-            shift = agents[idx:] + agents[:idx]  # Shift agents by idx to match simulation order
+            shift = agents[idx:] + agents[:idx]  # Match the command-line warrior order in _run_single_simulation
             with open(self.log_round(round_num) / COREWAR_LOG.format(idx=idx)) as f:
                 result_output = f.read()
 
-            # Get the last n lines which contain the scores (closer to original)
-            lines = result_output.strip().split("\n")
-            relevant_lines = lines[-len(shift) * 2 :] if len(lines) >= len(shift) * 2 else lines
-            relevant_lines = [l for l in relevant_lines if len(l.strip()) > 0]
-
-            # Go through each line; score position is correlated with agent index
-            for i, line in enumerate(relevant_lines):
-                match = re.search(r".*\sby\s.*\sscores\s(\d+)", line)
-                if match:
-                    scores[shift[i].name] += int(match.group(1))
-
-            # Last line corresponds to absolute number of wins
-            last = relevant_lines[-1][len("Results:") :].strip()
-            for i, w in enumerate(last.split()[:-1]):  # NOTE: Omitting ties (last entry)
-                wins[shift[i].name] += int(w)
+            lines = result_output.splitlines()
+            # Sum across every "…scores…" + "Results:" block (the record shift writes two).
+            pending: list[int] = []
+            saw_results = False
+            for line in lines:
+                m = score_pat.search(line)
+                if m:
+                    pending.append(int(m.group(1)))
+                elif line.strip().startswith("Results:"):
+                    saw_results = True
+                    for i, v in enumerate(pending[-len(shift) :]):  # the N score lines of this block
+                        scores[shift[i].name] += v
+                    for i, w in enumerate(line.strip()[len("Results:") :].split()[:-1]):  # omit ties
+                        if i < len(shift):
+                            wins[shift[i].name] += int(w)
+                    pending = []
+            if not saw_results:
+                self.logger.error(f"No 'Results:' line in {COREWAR_LOG.format(idx=idx)} for round {round_num}")
 
         if len(wins) != len(agents):
             # Should not happen

codeclash/arenas/corewar/replay.py

@@ -0,0 +1,121 @@
+"""CoreWar replay renderer: the MARS core as a grid, each cell coloured by its owning warrior.
+
+Parses ``sim_*.jsonl`` from ``trace.py`` and folds the per-frame ``c`` deltas into a cumulative
+owner-map; recent cells flash bright, PCs are white. No cell holds Redcode contents, so replays
+show behaviour, not source (conventions per the corewar-docs visualisation page).
+"""
+
+from __future__ import annotations
+
+import json
+
+from codeclash.replay.base import ReplayData, ReplayRenderer
+
+PALETTE = ["#3B78FF", "#E5484D", "#30A46C", "#F5A623", "#8E4EC6", "#12A594", "#E93D82", "#F76B15"]
+
+DRAW_JS = """
+const ARENA = (function(){
+  let CORE, GW, GH, CELL, COL, NAMES, cw, ch;
+  // Cumulative owner-map folded from frame deltas, with a cursor so forward playback is O(delta).
+  let owner, builtTo;
+  function setup(cv, G){
+    CORE = G.core; GW = G.w; GH = G.h; COL = G.colors; NAMES = G.names;
+    CELL = Math.max(2, Math.floor(Math.min(880 / GW, 620 / GH)));
+    cw = cv.width = GW * CELL; ch = cv.height = GH * CELL;
+    owner = new Int16Array(CORE).fill(-1);
+    builtTo = -1;
+  }
+  function ensure(F, i){
+    if(builtTo > i){ owner.fill(-1); builtTo = -1; }        // scrubbed backward: rebuild
+    for(let k = builtTo + 1; k <= i; k++){
+      const cs = F[k].c;
+      for(let j = 0; j < cs.length; j++) owner[cs[j][0]] = cs[j][1];
+    }
+    builtTo = i;
+  }
+  const cellX = (a) => (a % GW) * CELL, cellY = (a) => Math.floor(a / GW) * CELL;
+  function rect(ctx, a, color){ ctx.fillStyle = color; ctx.fillRect(cellX(a), cellY(a), CELL, CELL); }
+  // Dim the owner colour for settled cells; recent activity is drawn at full strength.
+  function dim(hex){
+    const n = parseInt(hex.slice(1), 16);
+    const r = (n>>16)&255, g = (n>>8)&255, b = n&255;
+    return `rgb(${(r*0.55)|0},${(g*0.55)|0},${(b*0.55)|0})`;
+  }
+  const DIM = {};
+  function draw(ctx, cv, G, i){
+    const F = G.frames, f = F[i];
+    ensure(F, i);
+    ctx.fillStyle = '#2b2f36';                               // grey = untouched core (initial DAT)
+    ctx.fillRect(0, 0, cw, ch);
+    for(let a = 0; a < CORE; a++){
+      const o = owner[a];
+      if(o >= 0){ const c = COL[o]; rect(ctx, a, (DIM[c] || (DIM[c] = dim(c)))); }
+    }
+    (f.c || []).forEach(([a, o]) => rect(ctx, a, COL[o] || '#fff'));   // recent activity, bright
+    (f.p || []).forEach((addr, w) => {                                 // program counters, white
+      if(f.d && f.d[w]) rect(ctx, addr, '#ffffff');
+    });
+  }
+  function side(G, i){
+    const f = G.frames[i], ids = Object.keys(NAMES);
+    return ids.map(id => {
+      const w = +id, dead = f.d ? !f.d[w] : false, procs = f.n ? f.n[w] : 0;
+      return `<div class="sn ${dead?'dead':''}"><span class="sw" style="background:${COL[w]}"></span>
+        <span style="min-width:120px">${NAMES[id]}</span>
+        <span>${dead ? '\\u2620 dead' : procs + ' proc' + (procs===1?'':'s')}</span></div>`;
+    }).join('') + '<div class="row muted" style="font-size:12px">cell colour = owning warrior \\u00b7 '
+      + 'bright = just now \\u00b7 white = program counter</div>';
+  }
+  return {setup, draw, side};
+})();
+"""
+
+
+class CoreWarReplayer(ReplayRenderer):
+    arena = "CoreWar"
+    sim_glob = "sim_*.jsonl"
+    DRAW_JS = DRAW_JS
+
+    def parse(self, raw: bytes, players=None) -> ReplayData:
+        rows = [json.loads(line) for line in raw.decode().splitlines() if line.strip()]
+        if not rows:
+            return ReplayData(w=1, h=1, frames=[])
+        header = rows[0]
+        result = rows[-1] if isinstance(rows[-1], dict) and "winner" in rows[-1] else {}
+        warriors = {str(k): v for k, v in header.get("warriors", {}).items()}
+        colors = {int(k): PALETTE[i % len(PALETTE)] for i, k in enumerate(sorted(warriors, key=int))}
+
+        frames = []
+        for r in rows:
+            if isinstance(r, dict) and "t" in r:
+                frames.append(
+                    {"turn": r["t"], "c": r.get("c", []), "p": r.get("p", []), "n": r.get("n", []), "d": r.get("d", [])}
+                )
+
+        return ReplayData(
+            w=header.get("w", 1),
+            h=header.get("h", 1),
+            frames=frames,
+            winner=result.get("winner"),
+            draw=result.get("draw", False),
+            extra={
+                "core": header.get("core", header.get("w", 1) * header.get("h", 1)),
+                "colors": colors,
+                "names": warriors,
+                "starts": header.get("starts", []),
+            },
+        )
+
+    def peek_winner(self, raw: bytes, players=None) -> tuple[str | None, bool] | None:
+        for line in reversed(raw.decode().splitlines()):
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                obj = json.loads(line)
+            except ValueError:
+                return None
+            if isinstance(obj, dict) and "winner" in obj:
+                return obj.get("winner"), obj.get("draw", False)
+            return None
+        return None