Training-parity minimap (real terrain + legend) + viewer (system/thinking/debrief)

Model-facing fix — 'is the minimap correctly sent?': it now is, with
an interpretable legend.
- minimap.terrain_png_for(): extract map.png bytes from the resolved
.oramap (zip), lru-cached
- minimap.render_b64(): prefer the training render_minimap (real
terrain + embedded legend the model can read) when terrain is
available; bench-native PNG only as fallback
- ModelAgent(base_map=…) resolves terrain once; _user_message +
eval_core playback both render via the training path so the saved
PNGs match exactly what the model saw; run_eval factories pass
base_map

Viewer parity with training view_v7_pipeline.py — show everything:
- battle_viewer.episode_view exposes system_prompt, debrief (the
briefing/scoped re-prompt handed to the model), tool_result
- app.py Battle Viewer: collapsible system prompt (char count),
open DEBRIEF/briefing (full, interrupt-tagged), reasoning(thinking),
tool calls, tool result, signals, goal
- Streamlit scripts/view_playback.py: system-prompt expander + model/
run caption (same format)

tests: test_minimap.py +2 (terrain extract + training renderer used,
agent terrain wiring). 15 touched-module tests green.

app.py

@@ -247,22 +247,38 @@ def _bv_turn_md(v: dict, heading: str) -> str:
             "reward vector: "
             + " ".join(f"`{k}={float(x):.2f}`" for k, x in rv.items()),
         ]
+    # System prompt (the deterministic scenario knowledge the model
+    # was given) — collapsible, shown with char count like the
+    # training pipeline viewer.
+    sp = str(v.get("system_prompt") or "")
+    if sp:
+        lines += [
+            "", f"<details><summary>🧠 system prompt ({len(sp)} chars)"
+            "</summary>\n\n```\n" + sp[:6000] + "\n```\n</details>"
+        ]
+    # DEBRIEF / briefing actually handed to the model this turn
+    # (post-interrupt this is the scoped re-prompt).
+    deb = str(v.get("debrief") or v.get("briefing") or "")
+    if deb:
+        tag = "⚡ DEBRIEF (interrupt)" if v.get("interrupt") else "briefing"
+        lines += [
+            "", f"<details open><summary>{tag}</summary>\n\n```\n"
+            + deb[:8000] + "\n```\n</details>"
+        ]
     if v.get("reasoning"):
-        lines += ["", "**reasoning**", "> " + str(v["reasoning"]).replace(
-            "\n", "\n> ")]
+        lines += ["", "**🤔 reasoning (thinking)**",
+                  "> " + str(v["reasoning"]).replace("\n", "\n> ")]
     if v.get("assistant_text"):
         lines += ["", "**model said**", str(v["assistant_text"])]
     cmds = v.get("commands", [])
-    lines += ["", "**commands**", "```\n" + (
+    lines += ["", "**tool calls**", "```\n" + (
         "\n".join(cmds) if cmds else "(none)") + "\n```"]
+    if v.get("tool_result"):
+        lines += [f"tool result: `{v['tool_result']}`"]
     sig = v.get("signals", {})
     if sig:
         lines += ["signals: " + " ".join(
             f"`{k}={sig[k]}`" for k in sig)]
-    if v.get("briefing"):
-        b = str(v["briefing"])
-        lines += ["", "<details><summary>briefing</summary>\n\n```\n"
-                  + b[:2000] + "\n```\n</details>"]
     return "\n\n".join(s for s in lines if s != "")
 
 

openra_bench/agent.py

@@ -374,13 +374,16 @@ def build_briefing(render_state: dict, objective: str = "") -> str:
     return "\n".join(lines)
 
 
-def _render_minimap_b64(render_state: dict) -> str | None:
-    """Best-effort minimap PNG. Returns None (text-only fallback) when
-    terrain isn't resolvable — vision degrades gracefully in Phase 0."""
+def _render_minimap_b64(
+    render_state: dict, terrain_png: bytes | None = None
+) -> str | None:
+    """Best-effort minimap PNG. With `terrain_png` uses the training
+    renderer (real terrain + an embedded legend the model can read);
+    else the bench fallback. None ⇒ graceful text-only."""
     try:
-        from .minimap import render_png_b64
+        from .minimap import render_b64
 
-        return render_png_b64(render_state)
+        return render_b64(render_state, terrain_png)
     except Exception as e:  # noqa: BLE001 — vision is optional
         logger.debug("minimap render skipped: %s", e)
         return None
@@ -466,11 +469,22 @@ class ModelAgent:
         objective: str = "",
         provider: ChatProvider | None = None,
         system_extra: str = "",
+        base_map: str = "",
     ):
         self.cfg = cfg
         self.objective = objective
         self.tools = _tool_schemas(allowed_tools)
         self.provider = provider or make_provider(cfg)
+        # Real terrain (map.png from the .oramap) so the model gets the
+        # training renderer's terrain+legend minimap, not colour dots.
+        self._terrain: bytes | None = None
+        if base_map:
+            try:
+                from .minimap import terrain_png_for
+
+                self._terrain = terrain_png_for(base_map)
+            except Exception:  # noqa: BLE001
+                self._terrain = None
         sys_content = SYSTEM_PROMPT
         if system_extra:
             # Deterministic scenario-scoped game knowledge (glossary,
@@ -483,7 +497,7 @@ class ModelAgent:
     def _user_message(self, render_state: dict) -> dict:
         text = build_briefing(render_state, self.objective)
         if self.cfg.vision:
-            b64 = _render_minimap_b64(render_state)
+            b64 = _render_minimap_b64(render_state, self._terrain)
             if b64:
                 return {
                     "role": "user",

openra_bench/battle_viewer.py

@@ -145,9 +145,20 @@ def episode_view(ep_dir: str | Path, turn_idx: int) -> dict:
         return {"n_turns": 0, "manifest": ep["manifest"]}
     i = max(0, min(turn_idx, n - 1))
     t = turns[i]
-    users = _role_turns(ep["messages"], "user")
-    assts = _role_turns(ep["messages"], "assistant")
+    msgs = ep["messages"]
+    sys_prompt = next(
+        (m.get("content", "") for m in msgs if m.get("role") == "system"), ""
+    )
+    if isinstance(sys_prompt, list):
+        sys_prompt = _text(sys_prompt)
+    users = _role_turns(msgs, "user")
+    assts = _role_turns(msgs, "assistant")
+    tools = _role_turns(msgs, "tool")
     g = t.get("goal") or {}
+    # The model's "debrief" for a turn = the briefing it was handed
+    # (post-interrupt this is the scoped re-prompt). Surface it
+    # explicitly + the interrupt reason, matching the training viewer.
+    briefing = _text(users[i]["content"]) if i < len(users) else ""
     return {
         "n_turns": n,
         "turn_idx": i,
@@ -156,11 +167,15 @@ def episode_view(ep_dir: str | Path, turn_idx: int) -> dict:
         "interrupt": t.get("interrupt"),
         "minimap_png": t.get("minimap_png"),
         "ascii_minimap": t.get("ascii_minimap", ""),
-        "briefing": _text(users[i]["content"]) if i < len(users) else "",
+        "system_prompt": sys_prompt,
+        "briefing": briefing,
+        "debrief": briefing,  # alias: what the model was (re-)briefed with
         "reasoning": (assts[i].get("reasoning", "")
                       if i < len(assts) else ""),
         "assistant_text": (_text(assts[i].get("content"))
                            if i < len(assts) else ""),
+        "tool_result": (_text(tools[i].get("content"))
+                        if i < len(tools) else ""),
         "commands": t.get("commands", []),
         "signals": t.get("signals", {}),
         "goal": g,

openra_bench/eval_core.py

@@ -219,8 +219,11 @@ def run_level(
                 _png = None
                 try:
                     from .agent import _render_minimap_b64
+                    from .minimap import terrain_png_for
 
-                    _png = _render_minimap_b64(rs)
+                    _png = _render_minimap_b64(
+                        rs, terrain_png_for(compiled.scenario.base_map)
+                    )
                 except Exception:  # noqa: BLE001 — playback never breaks a run
                     pass
                 from .goal_tracker import turn_goal

openra_bench/minimap.py

@@ -12,9 +12,59 @@ from __future__ import annotations
 import base64
 import io
 import logging
+import zipfile
+from functools import lru_cache
 
 logger = logging.getLogger(__name__)
 
+
+@lru_cache(maxsize=32)
+def terrain_png_for(base_map: str) -> bytes | None:
+    """Raw `map.png` bytes from the resolved `.oramap` (a zip), cached.
+    None when the map can't be resolved — caller falls back."""
+    if not base_map:
+        return None
+    try:
+        from .scenarios.loader import resolve_map_path
+
+        p = resolve_map_path(base_map)
+        if not p:
+            return None
+        with zipfile.ZipFile(p, "r") as zf:
+            if "map.png" in zf.namelist():
+                return zf.read("map.png")
+    except Exception as e:  # noqa: BLE001
+        logger.debug("terrain extract failed for %s: %s", base_map, e)
+    return None
+
+
+def render_b64(render_state: dict, terrain_png: bytes | None = None) -> str | None:
+    """Preferred minimap: the training renderer (real terrain + an
+    embedded legend the model can read) when terrain is available;
+    otherwise the self-contained bench fallback. Either way a *valid*
+    base64 PNG, or None for graceful text-only."""
+    if terrain_png:
+        try:
+            from openra_rl_training.training.minimap_renderer import (
+                render_minimap,
+            )
+
+            b64 = render_minimap(
+                terrain_png=terrain_png,
+                map_width=int(render_state.get("map_width", 64) or 64),
+                map_height=int(render_state.get("map_height", 64) or 64),
+                bounds_x=int(render_state.get("bounds_x", 0) or 0),
+                bounds_y=int(render_state.get("bounds_y", 0) or 0),
+                own_units=render_state.get("units_summary", []) or [],
+                enemy_units=render_state.get("enemy_summary", []) or [],
+                ascii_minimap=render_state.get("minimap", "") or "",
+            )
+            if b64:
+                return b64
+        except Exception as e:  # noqa: BLE001 — fall back below
+            logger.debug("training minimap failed, using fallback: %s", e)
+    return render_png_b64(render_state)
+
 CELL = 6  # px per map cell (≈768×240 for a 128×40 map — legible)
 
 # (R,G,B)

openra_bench/playback_view.py

@@ -72,10 +72,27 @@ def render_streamlit(root: str) -> None:  # pragma: no cover - UI glue
     m = ep["manifest"]
     st.title(f"{m.get('scenario', pick)} — {m.get('outcome', '?')}")
     st.caption(
+        f"{m.get('model','?')} · run {m.get('run_id','?')} · "
         f"turns {m.get('turns')}/{m.get('max_turns')} · "
         f"capability {m.get('capability')} · seed {m.get('seed')}"
     )
 
+    # System prompt once at top (the deterministic scenario knowledge
+    # the model was given) — same as the training pipeline viewer.
+    sysp = next(
+        (x.get("content", "") for x in ep["messages"]
+         if x.get("role") == "system"), ""
+    )
+    if isinstance(sysp, list):
+        sysp = " ".join(
+            p.get("text", "") for p in sysp if isinstance(p, dict)
+        )
+    if sysp:
+        with st.expander(f"🧠 System prompt ({len(sysp)} chars)",
+                         expanded=False):
+            st.code(sysp, language="text")
+
+    users = [x for x in ep["messages"] if x.get("role") == "user"]
     asst = _assistant_turns(ep["messages"])
     for i, t in enumerate(ep["turns"]):
         with st.expander(

openra_bench/run_eval.py

@@ -53,6 +53,7 @@ def _default_agent_factory(provider_cfg) -> AgentFactory:
                 compiled.max_turns,
             ),
             system_extra=scenario_primer(compiled),
+            base_map=compiled.scenario.base_map,
         )
         return agent.agent_fn
 
@@ -145,6 +146,7 @@ def evaluate(
                 ),
                 provider=shared,
                 system_extra=scenario_primer(compiled),
+                base_map=compiled.scenario.base_map,
             ).agent_fn
 
     # Run/model identity so a single playback root can hold many runs

tests/test_minimap.py

@@ -77,3 +77,37 @@ def test_agent_attaches_image_when_vision_on():
     assert url.startswith("data:image/png;base64,")
     # the attached payload is itself a valid PNG
     Image.open(io.BytesIO(base64.b64decode(url.split(",", 1)[1]))).verify()
+
+
+# ── training renderer (real terrain + embedded legend) ─────────────────────
+
+
+def test_terrain_png_extracted_and_training_renderer_used():
+    pytest.importorskip("matplotlib")
+    from openra_bench.minimap import render_b64, terrain_png_for
+
+    t = terrain_png_for("rush-hour-arena")
+    assert t and t[:4] == b"\x89PNG"            # real map.png from .oramap
+    rs = {
+        "minimap": "\n".join("#" * 128 for _ in range(40)),
+        "map_width": 128, "map_height": 40, "bounds_x": 0, "bounds_y": 0,
+        "units_summary": [{"id": "1", "cell_x": 6, "cell_y": 8}],
+        "enemy_summary": [],
+    }
+    with_terrain = render_b64(rs, t)
+    fallback = render_b64(rs, None)
+    im = Image.open(io.BytesIO(base64.b64decode(with_terrain)))
+    im.verify()
+    assert im.format == "PNG"
+    assert with_terrain != fallback             # training path actually used
+    assert terrain_png_for("no-such-map") is None  # graceful
+
+
+def test_agent_uses_terrain_when_base_map_given():
+    from openra_bench.agent import ModelAgent
+    from openra_bench.providers import ProviderConfig
+
+    a = ModelAgent(ProviderConfig(vision=True), allowed_tools=["observe"],
+                   provider=type("P", (), {"complete": lambda *x, **k: None})(),
+                   base_map="rush-hour-arena")
+    assert a._terrain and a._terrain[:4] == b"\x89PNG"