yxc20098 commited on
Commit
4a5b0dd
·
1 Parent(s): 680e8db

Add perception ablation grid (observation channel × fog of war)

Browse files

`fog_mode` now spans a 3×2 grid — three observation channels, each
fogged or `-clear` (no fog):

- structured — text briefing + a text 'unexplored regions' block.
- vision — text briefing + PNG minimap (the text already lists
units/enemies, so the image is a supplement).
- image — image-PRIMARY: the text briefing is redacted of every
coordinate; a labelled minimap (per-unit handles,
greedy collision avoidance + leader lines) is the sole
spatial source. The clean image-reading probe. The
model references units by the on-map handle (tank-1,
enemy-2); the agent maps it back to the engine id.

The `-clear` variants disable fog via the engine `reveal_map` flag —
perfect-information controls that isolate the perception cost; the
no-cheat bar applies only to the fogged cells.

- schema.py: PERCEPTION_MODES + CompiledLevel.reveal_map/obs_channel.
- eval_core.py: emit `reveal_map: true` into the engine scenario YAML
for `-clear` cells.
- prompt_v2.py: perception_labels (stable per-actor handles) +
briefing_image_primary (position-redacted briefing).
- minimap.py: render_tactical_minimap unit_labels — legible per-unit
labels with collision avoidance and leader lines.
- agent.py: image-primary channel dispatch, handle↔engine-id
translation, string-typed tool schemas for the image channel.
- run_eval.py: --perception-sweep expands every pack:level into the
six modality cells (pack:level:<mode>).

Pinned by tests/test_perception_ablation.py.

CLAUDE.md CHANGED
@@ -195,6 +195,39 @@ A scenario is defective if any of the following hold:
195
  `stance:3` AGENT combat unit auto-hunts the whole map; for a
196
  perception pack keep the agent's combat arm `stance:0` so a
197
  stall policy can't win for free by self-delivering the army.
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
198
  - **`pbox` costs 600** (not the 400 some old specs assumed);
199
  defense and infantry are SEPARATE production queues so an
200
  efficient policy queues `build('pbox')` and `build('e1')` in
 
195
  `stance:3` AGENT combat unit auto-hunts the whole map; for a
196
  perception pack keep the agent's combat arm `stance:0` so a
197
  stall policy can't win for free by self-delivering the army.
198
+ - **`reveal_map:` — the no-fog perception cell** (pinned by
199
+ `OpenRA-Rust/openra-data/tests/test_reveal_map.rs` +
200
+ `tests/test_perception_ablation.py`). A scenario may declare a
201
+ top-level `reveal_map: true`; the agent player then observes the
202
+ whole map with NO fog of war — every enemy actor is reported
203
+ regardless of shroud and `explored_percent` is 100. Parsed by
204
+ `oramap.rs::parse_scenario_yaml` (mirrors `spawn_mcvs`), applied
205
+ in `env.rs` (`is_visible_to` short-circuits true for the agent;
206
+ `refresh_explored_cells` fills the playable rectangle). This is
207
+ the no-fog half of the **perception ablation grid**. The bench
208
+ `fog_mode` spans 3 observation channels × 2 fog states = 6 cells
209
+ (`openra_bench/scenarios/schema.py::PERCEPTION_MODES`):
210
+ - `structured` — text briefing + a text 'Unexplored regions'
211
+ block; no image.
212
+ - `vision` — text briefing + PNG minimap. NOTE the briefing
213
+ already enumerates units/enemies with coordinates, so the image
214
+ is a redundant SUPPLEMENT — `vision` does NOT isolate
215
+ image-reading.
216
+ - `image` — image-PRIMARY: `prompt_v2.briefing_image_primary`
217
+ redacts every coordinate from the text; the labelled minimap
218
+ (`render_tactical_minimap(..., unit_labels=...)`) is the sole
219
+ spatial source. The model references units by the on-map
220
+ handle (`tank-1`, `enemy-2`); `agent._to_commands` maps the
221
+ handle back to the engine id. The clean "can the model read a
222
+ minimap" probe.
223
+ Each channel has a fogged form and a `-clear` (no-fog) form. A
224
+ `-clear` cell is a perfect-information CONTROL that isolates the
225
+ perception cost — a stall/observe policy WINS a `-clear`
226
+ perception pack (perception removed), so the no-cheat bar
227
+ applies ONLY to the fogged cells, never the `-clear` ones. Run
228
+ the full grid with `run_eval --perception-sweep` (expands every
229
+ `pack:level` into `pack:level:<mode>`); the human Play tab stays
230
+ on the canonical `vision` (fogged) modality.
231
  - **`pbox` costs 600** (not the 400 some old specs assumed);
232
  defense and infantry are SEPARATE production queues so an
233
  efficient policy queues `build('pbox')` and `build('e1')` in
openra_bench/agent.py CHANGED
@@ -375,39 +375,50 @@ def _render_minimap_b64(
375
  return None
376
 
377
 
378
- def _to_commands(tool_calls: list[dict], Command: Any) -> list:
 
 
 
 
 
 
 
 
 
 
 
379
  cmds = []
380
  for call in tool_calls:
381
  name = _TOOL_ALIASES.get(call.get("name", ""), call.get("name", ""))
382
  args = call.get("arguments") or {}
383
  try:
384
  if name == "move_units":
385
- ids = [str(i) for i in args["unit_ids"]]
386
  cmds.append(
387
  Command.move_units(ids, int(args["target_x"]), int(args["target_y"]))
388
  )
389
  elif name == "attack_unit":
390
- ids = [str(i) for i in args["unit_ids"]]
391
- cmds.append(Command.attack_unit(ids, str(args["target_id"])))
392
  elif name == "guard":
393
- ids = [str(i) for i in args["unit_ids"]]
394
- cmds.append(Command.guard(ids, str(args["target_id"])))
395
  elif name == "enter_transport":
396
- ids = [str(i) for i in args["unit_ids"]]
397
  cmds.append(
398
- Command.enter_transport(ids, str(args["target_id"]))
399
  )
400
  elif name == "observe":
401
  cmds.append(Command.observe())
402
  elif name == "surrender":
403
  cmds.append(Command.surrender())
404
  elif name == "set_stance":
405
- ids = [str(i) for i in args["unit_ids"]]
406
  cmds.append(Command.set_stance(ids, int(args["stance"])))
407
  elif name == "patrol":
408
- cmds.append(Command.patrol([str(i) for i in args["unit_ids"]]))
409
  elif name in ("attack_move", "harvest", "set_rally_point"):
410
- ids = [str(i) for i in args["unit_ids"]]
411
  fn = getattr(Command, name)
412
  cmds.append(fn(ids, int(args["target_x"]), int(args["target_y"])))
413
  elif name in (
@@ -419,7 +430,7 @@ def _to_commands(tool_calls: list[dict], Command: Any) -> list:
419
  "set_primary",
420
  "unload",
421
  ):
422
- ids = [str(i) for i in args["unit_ids"]]
423
  cmds.append(getattr(Command, name)(ids))
424
  elif name in ("build", "cancel_production"):
425
  cmds.append(getattr(Command, name)(str(args["item"])))
@@ -434,6 +445,35 @@ def _to_commands(tool_calls: list[dict], Command: Any) -> list:
434
  return cmds
435
 
436
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
437
  class ModelAgent:
438
  """One instance per episode (keeps bounded chat history).
439
 
@@ -462,6 +502,15 @@ class ModelAgent:
462
  self._level = level
463
  # Scenario config wins over the model-side cfg default.
464
  self._fog_mode = fog_mode or getattr(cfg, "fog_mode", "vision")
 
 
 
 
 
 
 
 
 
465
  # Real terrain (map.png from the .oramap) for the vendored
466
  # training bitmap minimap; persistent fog history across turns.
467
  self._terrain: bytes | None = None
@@ -484,6 +533,16 @@ class ModelAgent:
484
  sys_content = SYSTEM_PROMPT + (
485
  f"\n\n{system_extra}" if system_extra else ""
486
  )
 
 
 
 
 
 
 
 
 
 
487
  self.history: list[dict] = [{"role": "system", "content": sys_content}]
488
  self.stats = {"turns": 0, "tool_calls": 0, "empty_replies": 0}
489
  # Controller contract (openra_bench/controller.py): a ModelAgent
@@ -492,7 +551,60 @@ class ModelAgent:
492
  # all drive it interchangeably with any other policy backend.
493
  self.name = getattr(cfg, "model", None) or "model"
494
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
495
  def _user_message(self, render_state: dict) -> dict:
 
 
 
 
496
  # Briefing = vendored training briefing_v2 (one unit/line,
497
  # "moving to (x,y)", Idle list). Objective is in the system
498
  # prompt now, so it's NOT repeated here.
@@ -502,10 +614,11 @@ class ModelAgent:
502
  text = _v2_brief(render_state)
503
  except Exception: # noqa: BLE001 — never break a turn
504
  text = build_briefing(render_state, self.objective)
505
- # Structured-fog mode: NO image — append the text "Unexplored
506
- # regions" block instead (text-vs-vision A/B; pair with the
507
- # easy/medium level of the setup).
508
- if self._fog_mode == "structured":
 
509
  try:
510
  from .prompt_v2 import structured_fog as _v2_fog
511
 
@@ -608,7 +721,7 @@ class ModelAgent:
608
  ],
609
  }
610
  )
611
- cmds = _to_commands(reply.tool_calls, Command)
612
  self.stats["tool_calls"] += len(cmds)
613
  if not cmds:
614
  self.stats["empty_replies"] += 1
 
375
  return None
376
 
377
 
378
+ def _to_commands(
379
+ tool_calls: list[dict], Command: Any, label_to_id: dict | None = None
380
+ ) -> list:
381
+ # In the image-primary channel the model references units by the
382
+ # legible handle shown on the minimap (`tank-1`); map it back to the
383
+ # engine actor id. Numeric ids (every other channel) pass straight
384
+ # through — the lookup simply misses.
385
+ label_to_id = label_to_id or {}
386
+
387
+ def _rid(x: Any) -> str:
388
+ return label_to_id.get(str(x), str(x))
389
+
390
  cmds = []
391
  for call in tool_calls:
392
  name = _TOOL_ALIASES.get(call.get("name", ""), call.get("name", ""))
393
  args = call.get("arguments") or {}
394
  try:
395
  if name == "move_units":
396
+ ids = [_rid(i) for i in args["unit_ids"]]
397
  cmds.append(
398
  Command.move_units(ids, int(args["target_x"]), int(args["target_y"]))
399
  )
400
  elif name == "attack_unit":
401
+ ids = [_rid(i) for i in args["unit_ids"]]
402
+ cmds.append(Command.attack_unit(ids, _rid(args["target_id"])))
403
  elif name == "guard":
404
+ ids = [_rid(i) for i in args["unit_ids"]]
405
+ cmds.append(Command.guard(ids, _rid(args["target_id"])))
406
  elif name == "enter_transport":
407
+ ids = [_rid(i) for i in args["unit_ids"]]
408
  cmds.append(
409
+ Command.enter_transport(ids, _rid(args["target_id"]))
410
  )
411
  elif name == "observe":
412
  cmds.append(Command.observe())
413
  elif name == "surrender":
414
  cmds.append(Command.surrender())
415
  elif name == "set_stance":
416
+ ids = [_rid(i) for i in args["unit_ids"]]
417
  cmds.append(Command.set_stance(ids, int(args["stance"])))
418
  elif name == "patrol":
419
+ cmds.append(Command.patrol([_rid(i) for i in args["unit_ids"]]))
420
  elif name in ("attack_move", "harvest", "set_rally_point"):
421
+ ids = [_rid(i) for i in args["unit_ids"]]
422
  fn = getattr(Command, name)
423
  cmds.append(fn(ids, int(args["target_x"]), int(args["target_y"])))
424
  elif name in (
 
430
  "set_primary",
431
  "unload",
432
  ):
433
+ ids = [_rid(i) for i in args["unit_ids"]]
434
  cmds.append(getattr(Command, name)(ids))
435
  elif name in ("build", "cancel_production"):
436
  cmds.append(getattr(Command, name)(str(args["item"])))
 
445
  return cmds
446
 
447
 
448
+ def _image_primary_tools(tools: list[dict]) -> list[dict]:
449
+ """Re-type unit/target handles as strings for the image-primary
450
+ channel: the model references actors by the legible label drawn on
451
+ the minimap (`tank-1`, `enemy-2`), not numeric engine ids. The
452
+ `_to_commands` `label_to_id` map turns them back into engine ids."""
453
+ import copy
454
+
455
+ out = copy.deepcopy(tools)
456
+ for t in out:
457
+ props = (
458
+ t.get("function", {}).get("parameters", {}).get("properties", {})
459
+ )
460
+ ui = props.get("unit_ids")
461
+ if isinstance(ui, dict) and ui.get("type") == "array":
462
+ ui["items"] = {"type": "string"}
463
+ ui["description"] = (
464
+ 'unit handles EXACTLY as labelled on the minimap, '
465
+ 'e.g. ["tank-1","jeep-2"]'
466
+ )
467
+ tid = props.get("target_id")
468
+ if isinstance(tid, dict):
469
+ tid["type"] = "string"
470
+ tid["description"] = (
471
+ 'the target actor\'s handle as labelled on the minimap, '
472
+ 'e.g. "enemy-1"'
473
+ )
474
+ return out
475
+
476
+
477
  class ModelAgent:
478
  """One instance per episode (keeps bounded chat history).
479
 
 
502
  self._level = level
503
  # Scenario config wins over the model-side cfg default.
504
  self._fog_mode = fog_mode or getattr(cfg, "fog_mode", "vision")
505
+ # Image-primary channel: the text briefing carries no positions —
506
+ # the labelled minimap is the sole spatial source, and the model
507
+ # references units by those labels. Re-type the tool handles to
508
+ # strings; `_labels` / `_label_to_id` are rebuilt each turn.
509
+ self._image_primary = self._fog_mode.startswith("image")
510
+ if self._image_primary:
511
+ self.tools = _image_primary_tools(self.tools)
512
+ self._labels: dict[str, str] = {}
513
+ self._label_to_id: dict[str, str] = {}
514
  # Real terrain (map.png from the .oramap) for the vendored
515
  # training bitmap minimap; persistent fog history across turns.
516
  self._terrain: bytes | None = None
 
533
  sys_content = SYSTEM_PROMPT + (
534
  f"\n\n{system_extra}" if system_extra else ""
535
  )
536
+ if self._image_primary:
537
+ sys_content += (
538
+ "\n\nPERCEPTION MODE — IMAGE-PRIMARY. The text briefing "
539
+ "lists WHAT units exist but never where anything is. "
540
+ "Every position — your units AND the enemy — is shown "
541
+ "ONLY on the minimap image. Each marker is tagged with a "
542
+ "legible label (tank-1, jeep-2, enemy-1). Read the image "
543
+ "to locate units and threats; pass those exact labels as "
544
+ "the ids in your tool calls (e.g. unit_ids=[\"tank-1\"])."
545
+ )
546
  self.history: list[dict] = [{"role": "system", "content": sys_content}]
547
  self.stats = {"turns": 0, "tool_calls": 0, "empty_replies": 0}
548
  # Controller contract (openra_bench/controller.py): a ModelAgent
 
551
  # all drive it interchangeably with any other policy backend.
552
  self.name = getattr(cfg, "model", None) or "model"
553
 
554
+ def _image_primary_message(self, render_state: dict) -> dict:
555
+ """Image-primary turn message: a position-redacted text briefing
556
+ plus a labelled minimap — the minimap is the ONLY place the
557
+ model learns where its units and the enemy are."""
558
+ from .prompt_v2 import briefing_image_primary, perception_labels
559
+
560
+ # Carry last turn's map forward so a label stays pinned to its
561
+ # actor for the whole episode (stable handles across turns).
562
+ self._labels = perception_labels(render_state, self._labels)
563
+ self._label_to_id = {v: k for k, v in self._labels.items()}
564
+ text = briefing_image_primary(render_state, self._labels)
565
+ b64 = None
566
+ try:
567
+ import base64
568
+ import io
569
+
570
+ from .minimap import render_tactical_minimap
571
+
572
+ # Keep the PNG ≤ ~1560px wide so the vision API does not
573
+ # downscale it (which would shrink the unit labels below
574
+ # legibility); the 6px base cell × scale sets the width.
575
+ rows = [
576
+ r for r in (render_state.get("minimap") or "").split("\n")
577
+ if r
578
+ ]
579
+ w = max((len(r) for r in rows), default=64)
580
+ scale = max(2, min(6, 1560 // max(1, w * 6)))
581
+ img = render_tactical_minimap(
582
+ render_state, scale=scale, unit_labels=self._labels,
583
+ )
584
+ if img is not None:
585
+ buf = io.BytesIO()
586
+ img.save(buf, "PNG")
587
+ b64 = base64.b64encode(buf.getvalue()).decode("ascii")
588
+ except Exception: # noqa: BLE001 — degrade to text-only on render fail
589
+ b64 = None
590
+ if b64:
591
+ return {
592
+ "role": "user",
593
+ "content": [
594
+ {"type": "text", "text": text},
595
+ {
596
+ "type": "image_url",
597
+ "image_url": {"url": f"data:image/png;base64,{b64}"},
598
+ },
599
+ ],
600
+ }
601
+ return {"role": "user", "content": text}
602
+
603
  def _user_message(self, render_state: dict) -> dict:
604
+ # Image-primary channel builds its own (position-redacted)
605
+ # briefing + labelled minimap — dispatch before the text path.
606
+ if self._image_primary:
607
+ return self._image_primary_message(render_state)
608
  # Briefing = vendored training briefing_v2 (one unit/line,
609
  # "moving to (x,y)", Idle list). Objective is in the system
610
  # prompt now, so it's NOT repeated here.
 
614
  text = _v2_brief(render_state)
615
  except Exception: # noqa: BLE001 — never break a turn
616
  text = build_briefing(render_state, self.objective)
617
+ # Structured channel: NO image — append the text "Unexplored
618
+ # regions" block instead (text-vs-vision A/B). Covers both
619
+ # `structured` (fogged) and `structured-clear` (no fog — under
620
+ # reveal_map the block reports zero unexplored regions).
621
+ if self._fog_mode.startswith("structured"):
622
  try:
623
  from .prompt_v2 import structured_fog as _v2_fog
624
 
 
721
  ],
722
  }
723
  )
724
+ cmds = _to_commands(reply.tool_calls, Command, self._label_to_id)
725
  self.stats["tool_calls"] += len(cmds)
726
  if not cmds:
727
  self.stats["empty_replies"] += 1
openra_bench/eval_core.py CHANGED
@@ -67,6 +67,11 @@ def _scenario_to_tmp_yaml(compiled: CompiledLevel) -> str:
67
  sched = getattr(compiled, "scheduled_events", None) or []
68
  if sched:
69
  data["scheduled_events"] = sched
 
 
 
 
 
70
  fd = tempfile.NamedTemporaryFile(
71
  "w", suffix=f"_{compiled.pack_id}_{compiled.level}.yaml", delete=False
72
  )
 
67
  sched = getattr(compiled, "scheduled_events", None) or []
68
  if sched:
69
  data["scheduled_events"] = sched
70
+ # No-fog perception cells (fog_mode ends with "-clear") flip the
71
+ # engine's `reveal_map` flag: the agent observes the whole map with
72
+ # no fog of war — the clear half of the perception ablation grid.
73
+ if getattr(compiled, "reveal_map", False):
74
+ data["reveal_map"] = True
75
  fd = tempfile.NamedTemporaryFile(
76
  "w", suffix=f"_{compiled.pack_id}_{compiled.level}.yaml", delete=False
77
  )
openra_bench/minimap.py CHANGED
@@ -284,6 +284,7 @@ def render_tactical_minimap(
284
  legend: bool = True,
285
  selected=None,
286
  arrows=None,
 
287
  ):
288
  """A legible tactical minimap as a PIL RGB image:
289
 
@@ -295,7 +296,13 @@ def render_tactical_minimap(
295
  strip beneath the map.
296
 
297
  `scale` multiplies the 6px base cell. Returns None if Pillow is
298
- missing or there is nothing to draw."""
 
 
 
 
 
 
299
  try:
300
  from PIL import Image, ImageDraw
301
  except Exception: # noqa: BLE001
@@ -341,7 +348,9 @@ def render_tactical_minimap(
341
  is_b = force_building or bool(it.get("is_building"))
342
  atype = (it.get("actor_type") or it.get("type") or "?")
343
  shape = _unit_shape(atype, is_b)
344
- by_cell.setdefault((cx, cy), []).append((side, shape))
 
 
345
  if side == "own" and atype != "?":
346
  own_types.setdefault(str(atype).lower(), shape)
347
 
@@ -362,11 +371,14 @@ def render_tactical_minimap(
362
  badge_font = _minimap_font(max(9, int(cp * 0.62)))
363
  for (cx, cy), occ in by_cell.items():
364
  # Dominant occupant decides the shape; prefer a building.
365
- side, shape = next(
366
  (o for o in occ if _is_bld(o[1])), occ[0]
367
  )
368
  _draw_unit_shape(draw, cx, cy, cp, shape, _color(side, shape))
369
- if len(occ) > 1:
 
 
 
370
  tx, ty = (cx + 1) * cp - cp * 0.42, cy * cp + 1
371
  draw.text(
372
  (tx, ty), str(len(occ)), fill=(255, 255, 255),
@@ -426,6 +438,65 @@ def render_tactical_minimap(
426
  stroke_width=3, stroke_fill=(0, 0, 0),
427
  )
428
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
429
  # Legend strip — the unit TYPES actually present, each with its
430
  # own shape, so the player can read 1tnk vs 2tnk vs e3 etc.
431
  if legend:
@@ -449,10 +520,16 @@ def render_tactical_minimap(
449
  except Exception: # noqa: BLE001
450
  tw = len(tname) * cp * 0.5
451
  x = int(tx + tw + cp * 0.7)
452
- draw.text(
453
- (int(cp * 0.4), ly + int(cp * 1.05)),
 
 
 
454
  "green = you red/orange = enemy number = units stacked "
455
- " white box = selected arrow = move/attack order",
 
 
 
456
  fill=(200, 202, 212), font=lfont,
457
  )
458
 
 
284
  legend: bool = True,
285
  selected=None,
286
  arrows=None,
287
+ unit_labels=None,
288
  ):
289
  """A legible tactical minimap as a PIL RGB image:
290
 
 
296
  strip beneath the map.
297
 
298
  `scale` multiplies the 6px base cell. Returns None if Pillow is
299
+ missing or there is nothing to draw.
300
+
301
+ `unit_labels` (id-str → handle, e.g. `{"1004": "tank-1"}`) drives
302
+ the image-primary perception channel: every actor is tagged with
303
+ its legible handle so the model can identify and command units
304
+ from the picture alone (the text briefing carries no positions).
305
+ When set, the per-cell count badge is replaced by the labels."""
306
  try:
307
  from PIL import Image, ImageDraw
308
  except Exception: # noqa: BLE001
 
348
  is_b = force_building or bool(it.get("is_building"))
349
  atype = (it.get("actor_type") or it.get("type") or "?")
350
  shape = _unit_shape(atype, is_b)
351
+ by_cell.setdefault((cx, cy), []).append(
352
+ (side, shape, it.get("id"))
353
+ )
354
  if side == "own" and atype != "?":
355
  own_types.setdefault(str(atype).lower(), shape)
356
 
 
371
  badge_font = _minimap_font(max(9, int(cp * 0.62)))
372
  for (cx, cy), occ in by_cell.items():
373
  # Dominant occupant decides the shape; prefer a building.
374
+ side, shape, _aid = next(
375
  (o for o in occ if _is_bld(o[1])), occ[0]
376
  )
377
  _draw_unit_shape(draw, cx, cy, cp, shape, _color(side, shape))
378
+ # The count badge and per-unit labels are mutually exclusive —
379
+ # `unit_labels` (image-primary) names each occupant individually,
380
+ # which already disambiguates a stack.
381
+ if len(occ) > 1 and not unit_labels:
382
  tx, ty = (cx + 1) * cp - cp * 0.42, cy * cp + 1
383
  draw.text(
384
  (tx, ty), str(len(occ)), fill=(255, 255, 255),
 
438
  stroke_width=3, stroke_fill=(0, 0, 0),
439
  )
440
 
441
+ # Per-unit ID labels — the image-primary channel. Each actor's
442
+ # legible handle (`tank-1`, `enemy-2`) is placed near its marker
443
+ # with greedy collision avoidance (labels nudge clear of one
444
+ # another into free space) and a leader line when a label drifts
445
+ # off its cell — so the model can identify every unit from the
446
+ # picture alone. Drawn last so nothing occludes the text.
447
+ if unit_labels:
448
+ lab_font = _minimap_font(max(15, int(cp * 0.95)))
449
+ img_w, img_h = w * cp, h * cp
450
+ # One entry per actor id — an actor can appear in both the unit
451
+ # and building lists; dedup so its label is drawn once.
452
+ seen: set = set()
453
+ actors = [] # (cx, cy, label, side)
454
+ for (cx, cy), occ in by_cell.items():
455
+ for side, _shape, aid in occ:
456
+ key = str(aid)
457
+ if aid is None or key in seen or key not in unit_labels:
458
+ continue
459
+ seen.add(key)
460
+ actors.append((cx, cy, unit_labels[key], side))
461
+ placed: list = [] # occupied label rects (x0, y0, x1, y1)
462
+
463
+ def _free(x0, y0, x1, y1):
464
+ return all(
465
+ x1 < r[0] or x0 > r[2] or y1 < r[1] or y0 > r[3]
466
+ for r in placed
467
+ )
468
+
469
+ for cx, cy, lab, side in sorted(actors, key=lambda a: (a[1], a[0])):
470
+ try:
471
+ bb = draw.textbbox((0, 0), lab, font=lab_font, stroke_width=3)
472
+ tw, th = bb[2] - bb[0], bb[3] - bb[1]
473
+ except Exception: # noqa: BLE001
474
+ tw, th = len(lab) * 9, 16
475
+ mx, my = cx * cp + cp // 2, cy * cp + cp // 2
476
+ lx = min(max(0, mx + 4), img_w - tw - 1)
477
+ ly = my - cp - th
478
+ step = th + 3
479
+ for _ in range(60): # nudge upward into free space
480
+ cand = min(max(1, ly), img_h - th - 1)
481
+ if _free(lx, cand, lx + tw, cand + th):
482
+ ly = cand
483
+ break
484
+ ly -= step
485
+ else:
486
+ ly = min(max(1, my + cp), img_h - th - 1)
487
+ placed.append((lx, ly, lx + tw, ly + th))
488
+ # Leader line when the label sits away from its marker.
489
+ if abs(lx + tw // 2 - mx) > cp or abs(ly + th // 2 - my) > cp:
490
+ draw.line(
491
+ [(mx, my), (lx + 2, ly + th // 2)],
492
+ fill=(165, 167, 178), width=1,
493
+ )
494
+ col = (175, 255, 175) if side == "own" else (255, 190, 172)
495
+ draw.text(
496
+ (lx, ly), lab, fill=col, font=lab_font,
497
+ stroke_width=3, stroke_fill=(0, 0, 0),
498
+ )
499
+
500
  # Legend strip — the unit TYPES actually present, each with its
501
  # own shape, so the player can read 1tnk vs 2tnk vs e3 etc.
502
  if legend:
 
520
  except Exception: # noqa: BLE001
521
  tw = len(tname) * cp * 0.5
522
  x = int(tx + tw + cp * 0.7)
523
+ help_txt = (
524
+ "green = you red/orange = enemy "
525
+ "label = unit id (pass it to your tools) "
526
+ "scout the dark area to reveal more"
527
+ if unit_labels else
528
  "green = you red/orange = enemy number = units stacked "
529
+ " white box = selected arrow = move/attack order"
530
+ )
531
+ draw.text(
532
+ (int(cp * 0.4), ly + int(cp * 1.05)), help_txt,
533
  fill=(200, 202, 212), font=lfont,
534
  )
535
 
openra_bench/prompt_v2.py CHANGED
@@ -191,6 +191,121 @@ def structured_fog(render_state: dict) -> str:
191
  return format_structured_fog(obs, bounds)
192
 
193
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
194
  def minimap_b64(
195
  render_state: dict, terrain_png: bytes | None,
196
  explored_history: set | None,
 
191
  return format_structured_fog(obs, bounds)
192
 
193
 
194
+ # ── Image-primary channel ────────────────────────────────────────────
195
+ # The `image` perception modality: the text briefing is redacted of
196
+ # every coordinate, so the PNG minimap (with per-unit labels) is the
197
+ # ONLY source of spatial state. Units carry a short legible handle
198
+ # (`tank-1`, `enemy-2`) shown identically in the briefing roster, on
199
+ # the minimap, and as the id the model passes to its tools.
200
+ _FRIENDLY_TYPE = {
201
+ "1tnk": "tank", "2tnk": "tank", "3tnk": "tank", "4tnk": "tank",
202
+ "jeep": "jeep", "apc": "apc", "arty": "arty", "harv": "harvester",
203
+ "mcv": "mcv", "e1": "rifle", "e2": "grenadier", "e3": "rocket",
204
+ "e6": "engineer",
205
+ }
206
+
207
+
208
+ def _friendly_word(actor_type: object) -> str:
209
+ """A short legible word for an actor type — `1tnk` → `tank`. Falls
210
+ back to the raw type (so buildings read `proc-1`, `pbox-1`)."""
211
+ t = str(actor_type or "").lower()
212
+ return _FRIENDLY_TYPE.get(t, t or "unit")
213
+
214
+
215
+ def perception_labels(
216
+ render_state: dict, prior: dict[str, str] | None = None
217
+ ) -> dict[str, str]:
218
+ """Per-actor handle map for the image-primary channel: engine id
219
+ (str) → a short legible label. Own actors get a type-word handle
220
+ (`tank-1`, `jeep-2`, `proc-1`); enemy actors get `enemy-N`.
221
+
222
+ `prior` (the previous turn's map) is carried forward so a label
223
+ stays pinned to its actor for the actor's whole lifetime — an
224
+ enemy revealed mid-episode keeps the same handle on every later
225
+ turn instead of renumbering when a lower-id foe is later scouted.
226
+ New actors are assigned in engine-id order, continuing the count."""
227
+ labels: dict[str, str] = dict(prior or {})
228
+ # Highest index already used per prefix — so new handles continue
229
+ # the sequence rather than colliding with carried-over ones.
230
+ used: dict[str, int] = {}
231
+ for lab in labels.values():
232
+ pre, _, num = lab.rpartition("-")
233
+ if pre and num.isdigit():
234
+ used[pre] = max(used.get(pre, 0), int(num))
235
+
236
+ def _assign(items, prefix_fn):
237
+ for a in sorted(items, key=lambda a: int(a.get("id", 0) or 0)):
238
+ aid = a.get("id")
239
+ if aid is None or str(aid) in labels:
240
+ continue
241
+ pre = prefix_fn(a)
242
+ used[pre] = used.get(pre, 0) + 1
243
+ labels[str(aid)] = f"{pre}-{used[pre]}"
244
+
245
+ _assign(
246
+ render_state.get("units_summary", []) or [],
247
+ lambda a: _friendly_word(a.get("type")),
248
+ )
249
+ _assign(
250
+ render_state.get("own_buildings", []) or [],
251
+ lambda a: _friendly_word(a.get("type")),
252
+ )
253
+ enemies = list(render_state.get("enemy_summary", []) or [])
254
+ enemies += list(render_state.get("enemy_buildings_summary", []) or [])
255
+ _assign(enemies, lambda a: "enemy")
256
+ return labels
257
+
258
+
259
+ def briefing_image_primary(render_state: dict, labels: dict[str, str]) -> str:
260
+ """Image-primary briefing: every coordinate redacted. The text
261
+ keeps the non-spatial scaffolding (funds, power, the unit roster as
262
+ label+type handles, production) so the model knows WHAT it has —
263
+ the minimap image is the only place it learns WHERE anything is."""
264
+ st = state_from_render(render_state)
265
+ econ = st["economy"]
266
+ funds = econ["cash"] + econ["ore"]
267
+ out = [
268
+ f"--- TURN BRIEFING (tick {st['tick']}) ---",
269
+ f"Funds: ${funds} (cash=${econ['cash']} + ore=${econ['ore']}) | "
270
+ f"Power: {st['power_balance']:+d} | "
271
+ f"Harvesters: {econ['harvester_count']} | "
272
+ f"Explored: {st['explored_percent']}%",
273
+ ]
274
+ units = sorted(
275
+ st["units_summary"], key=lambda u: int(u.get("id", 0) or 0)
276
+ )
277
+ out.append(f"Your units ({len(units)}) — find each on the minimap:")
278
+ idle = []
279
+ for u in units:
280
+ lab = labels.get(str(u.get("id")), str(u.get("id")))
281
+ out.append(f" {lab} ({u.get('type') or '?'})")
282
+ if str(u.get("activity", "")).lower() in ("", "idle"):
283
+ idle.append(lab)
284
+ if idle:
285
+ out.append(f"Idle: {', '.join(idle)}")
286
+ bsum = st["buildings_summary"]
287
+ if bsum:
288
+ out.append(f"Your buildings ({len(bsum)}) — find each on the minimap:")
289
+ for b in bsum:
290
+ lab = labels.get(str(b.get("id")), str(b.get("id")))
291
+ out.append(f" {lab} ({b.get('type') or '?'})")
292
+ else:
293
+ out.append("Buildings: none (you command mobile units only).")
294
+ out.append(
295
+ "Enemies: NOT listed here — read the minimap. Enemy markers are "
296
+ "labelled enemy-1, enemy-2, … (only those in your units' sight; "
297
+ "scout the fog to reveal more)."
298
+ )
299
+ prod = st["production_items"]
300
+ out.append(f"Production: {', '.join(prod) if prod else 'IDLE'}")
301
+ out.append(
302
+ "ALL unit and enemy POSITIONS are on the minimap image only — "
303
+ "each marker is labelled with the id shown above; pass that id "
304
+ "to your tools."
305
+ )
306
+ return "\n".join(out)
307
+
308
+
309
  def minimap_b64(
310
  render_state: dict, terrain_png: bytes | None,
311
  explored_history: set | None,
openra_bench/run_eval.py CHANGED
@@ -117,12 +117,18 @@ def evaluate(
117
  dry_run: bool = False,
118
  report_path: str | Path | None = None,
119
  progress=None,
 
120
  ) -> dict:
121
  """Run packs×levels×seeds. If `held_out_seeds` is given, those are
122
  run too and tagged split='held_out'; the report adds
123
  `overall_held_out` and `generalization_gap` (public composite −
124
  held-out composite) — the anti-memorization metric the
125
  generalization literature (Procgen/SMACv2/lmgame-Bench) requires.
 
 
 
 
 
126
  """
127
  from .resilience import (
128
  BudgetExceeded,
@@ -201,9 +207,22 @@ def evaluate(
201
  f"{pack.meta.quarantine_reason or 'excluded from default set'})"
202
  )
203
  continue
 
 
 
 
 
 
 
 
 
 
 
 
 
204
  # Declared configs (pack:config_name, each pins level+fog_mode)
205
  # supersede the raw 3-level enumeration when present.
206
- if pack.configs:
207
  from .scenarios.loader import is_map_supported
208
 
209
  ms = is_map_supported(pack.base_map)
@@ -577,6 +596,10 @@ def main(argv: list[str]) -> int:
577
  ap.add_argument("--fog-mode", default="vision",
578
  choices=["vision", "structured"],
579
  help="spatial channel: PNG minimap vs text fog")
 
 
 
 
580
  a = ap.parse_args(argv[1:])
581
 
582
  cfg = None
@@ -618,6 +641,7 @@ def main(argv: list[str]) -> int:
618
  smoke=a.smoke,
619
  dry_run=a.dry_run,
620
  report_path=a.out,
 
621
  progress=lambda d, n, rec, c: print(
622
  f"[{d}/{n}] {rec['cell']}:{rec['split']}#{rec['seed']} "
623
  f"{rec['outcome']} comp={rec['composite']} "
 
117
  dry_run: bool = False,
118
  report_path: str | Path | None = None,
119
  progress=None,
120
+ perception_sweep: bool = False,
121
  ) -> dict:
122
  """Run packs×levels×seeds. If `held_out_seeds` is given, those are
123
  run too and tagged split='held_out'; the report adds
124
  `overall_held_out` and `generalization_gap` (public composite −
125
  held-out composite) — the anti-memorization metric the
126
  generalization literature (Procgen/SMACv2/lmgame-Bench) requires.
127
+
128
+ `perception_sweep` expands every pack×level into the 4 perception
129
+ ablation cells (`pack:level:<mode>` for mode in PERCEPTION_MODES —
130
+ vision/structured × fog/no-fog) instead of the raw 3 levels, so one
131
+ run yields the full channel-cost / fog-cost decomposition.
132
  """
133
  from .resilience import (
134
  BudgetExceeded,
 
207
  f"{pack.meta.quarantine_reason or 'excluded from default set'})"
208
  )
209
  continue
210
+ # Perception sweep: every level × the 4 modality cells
211
+ # (pack:level:<mode>). Overrides both declared configs and the
212
+ # raw enumeration — it is an explicit ablation request.
213
+ if perception_sweep:
214
+ from .scenarios.schema import PERCEPTION_MODES
215
+
216
+ unit_iter = []
217
+ for lv in levels:
218
+ for mode in PERCEPTION_MODES:
219
+ cl = compile_level(pack, lv)
220
+ cl.fog_mode = mode
221
+ cl.config_name = f"{lv}:{mode}"
222
+ unit_iter.append((cl, f"{pack.meta.id}:{lv}:{mode}"))
223
  # Declared configs (pack:config_name, each pins level+fog_mode)
224
  # supersede the raw 3-level enumeration when present.
225
+ elif pack.configs:
226
  from .scenarios.loader import is_map_supported
227
 
228
  ms = is_map_supported(pack.base_map)
 
596
  ap.add_argument("--fog-mode", default="vision",
597
  choices=["vision", "structured"],
598
  help="spatial channel: PNG minimap vs text fog")
599
+ ap.add_argument("--perception-sweep", action="store_true",
600
+ help="run the 2x2 perception ablation: every "
601
+ "pack:level expanded into vision/structured x "
602
+ "fog/no-fog (pack:level:<mode>)")
603
  a = ap.parse_args(argv[1:])
604
 
605
  cfg = None
 
641
  smoke=a.smoke,
642
  dry_run=a.dry_run,
643
  report_path=a.out,
644
+ perception_sweep=a.perception_sweep,
645
  progress=lambda d, n, rec, c: print(
646
  f"[{d}/{n}] {rec['cell']}:{rec['split']}#{rec['seed']} "
647
  f"{rec['outcome']} comp={rec['composite']} "
openra_bench/scenarios/schema.py CHANGED
@@ -118,16 +118,47 @@ class Level(BaseModel):
118
  forbidden_tools: list[str] = Field(default_factory=list)
119
 
120
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
121
  class ScenarioConfig(BaseModel):
122
  """A named runnable configuration of one pack: pins a difficulty
123
- `level` and the observation `fog_mode`. The same setup at
124
- `fog_mode: structured` (text 'Unexplored regions') vs `vision`
125
- (PNG minimap) becomes two distinct cells, so text-vs-vision is a
126
- first-class comparison the YAML declares (not just a CLI flag)."""
 
127
 
128
  name: str = Field(..., description="cell suffix, e.g. easy-structured")
129
  level: LevelName
130
- fog_mode: Literal["vision", "structured"] = "vision"
131
  # Optional override of the level's objective_coords for this cell.
132
  objective_coords: Literal["exact", "relative"] | None = None
133
 
@@ -155,9 +186,9 @@ class CompiledLevel(BaseModel):
155
  map_supported: bool = Field(
156
  ..., description="False => Rust lacks this map (Phase 3 gate)"
157
  )
158
- # Observation channel + the cell label. config_name is None for
159
- # legacy level cells (pack:level); set for declared configs
160
- # (pack:config_name).
161
  fog_mode: str = "vision"
162
  config_name: str | None = None
163
  objective_coords: Literal["exact", "relative"] = "exact"
@@ -171,6 +202,24 @@ class CompiledLevel(BaseModel):
171
  # scenario, and re-attached at YAML-write time.
172
  scheduled_events: list[dict[str, Any]] = Field(default_factory=list)
173
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
174
 
175
  class ScenarioPack(BaseModel):
176
  """The contributor-authored unit. One file = one decision problem."""
 
118
  forbidden_tools: list[str] = Field(default_factory=list)
119
 
120
 
121
+ # The perception ablation grid: observation channel × fog of war.
122
+ #
123
+ # Three channels:
124
+ # structured — text briefing + a text 'Unexplored regions' block;
125
+ # NO image. The text-only condition.
126
+ # vision — text briefing + PNG minimap. The multimodal "both
127
+ # available" condition — note the text briefing already
128
+ # enumerates units/enemies, so the image is a SUPPLEMENT.
129
+ # image — image-PRIMARY: the text briefing is redacted of every
130
+ # coordinate (and the enemy line dropped); the PNG, with
131
+ # legible per-unit labels, is the ONLY source of spatial
132
+ # state. The clean "can the model read a minimap" probe.
133
+ #
134
+ # Fog axis: bare name ⇒ fog ON (canonical scoring modality); the
135
+ # `-clear` variant reveals the whole map (engine `reveal_map: true` ⇒
136
+ # every enemy observed, `explored_percent` 100). A `-clear` cell is a
137
+ # perfect-information CONTROL for measuring the perception cost — the
138
+ # no-cheat bar applies only to the fogged cells.
139
+ PERCEPTION_MODES = (
140
+ "structured", "structured-clear",
141
+ "vision", "vision-clear",
142
+ "image", "image-clear",
143
+ )
144
+ FogMode = Literal[
145
+ "structured", "structured-clear",
146
+ "vision", "vision-clear",
147
+ "image", "image-clear",
148
+ ]
149
+
150
+
151
  class ScenarioConfig(BaseModel):
152
  """A named runnable configuration of one pack: pins a difficulty
153
+ `level` and the observation `fog_mode`. `fog_mode` spans the 2×2
154
+ perception grid — channel (`vision` PNG minimap vs `structured`
155
+ text 'Unexplored regions') × fog (on, or `-clear` ⇒ no fog) — so
156
+ text-vs-vision AND fogged-vs-clear are first-class comparisons the
157
+ YAML declares, not just CLI flags."""
158
 
159
  name: str = Field(..., description="cell suffix, e.g. easy-structured")
160
  level: LevelName
161
+ fog_mode: FogMode = "vision"
162
  # Optional override of the level's objective_coords for this cell.
163
  objective_coords: Literal["exact", "relative"] | None = None
164
 
 
186
  map_supported: bool = Field(
187
  ..., description="False => Rust lacks this map (Phase 3 gate)"
188
  )
189
+ # Observation modality (channel × fog — see PERCEPTION_MODES) + the
190
+ # cell label. config_name is None for legacy level cells
191
+ # (pack:level); set for declared configs / sweep cells.
192
  fog_mode: str = "vision"
193
  config_name: str | None = None
194
  objective_coords: Literal["exact", "relative"] = "exact"
 
202
  # scenario, and re-attached at YAML-write time.
203
  scheduled_events: list[dict[str, Any]] = Field(default_factory=list)
204
 
205
+ @property
206
+ def reveal_map(self) -> bool:
207
+ """No-fog cell? The `-clear` perception modes disable fog of
208
+ war — `_scenario_to_tmp_yaml` emits `reveal_map: true` so the
209
+ engine reveals the whole map to the agent."""
210
+ return self.fog_mode.endswith("-clear")
211
+
212
+ @property
213
+ def obs_channel(self) -> str:
214
+ """The observation channel — `structured` (text only), `image`
215
+ (image-primary: text redacted, PNG is the sole spatial source),
216
+ or `vision` (text + PNG) — independent of the fog axis."""
217
+ if self.fog_mode.startswith("structured"):
218
+ return "structured"
219
+ if self.fog_mode.startswith("image"):
220
+ return "image"
221
+ return "vision"
222
+
223
 
224
  class ScenarioPack(BaseModel):
225
  """The contributor-authored unit. One file = one decision problem."""
tests/test_perception_ablation.py ADDED
@@ -0,0 +1,205 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """The perception ablation grid: observation channel × fog of war.
2
+
3
+ `fog_mode` spans 3 channels × 2 fog states = 6 cells:
4
+ structured / structured-clear — text only, no image
5
+ vision / vision-clear — text briefing + PNG minimap
6
+ image / image-clear — image-PRIMARY: text redacted of all
7
+ positions; the labelled minimap is
8
+ the sole spatial source.
9
+ The `-clear` cells disable fog (engine `reveal_map: true`) — they are
10
+ perfect-information CONTROLS that isolate the perception cost; the
11
+ fogged cells keep the no-cheat bar.
12
+
13
+ Decomposition the grid enables:
14
+ score(vision-clear) − score(vision) = partial-observability cost
15
+ score(vision) − score(image) = text-scaffolding crutch
16
+ score(vision) − score(structured) = visual-channel advantage
17
+ """
18
+
19
+ from __future__ import annotations
20
+
21
+ from pathlib import Path
22
+
23
+ import pytest
24
+ import yaml
25
+
26
+ pytest.importorskip("openra_rl_training", reason="Rust env wheel not installed")
27
+
28
+ from openra_bench.eval_core import _scenario_to_tmp_yaml, run_level
29
+ from openra_bench.scenarios import load_pack
30
+ from openra_bench.scenarios.loader import compile_level
31
+ from openra_bench.scenarios.schema import PERCEPTION_MODES, ScenarioConfig
32
+
33
+ PACKS = Path(__file__).parent.parent / "openra_bench" / "scenarios" / "packs"
34
+ _PACK = "perception-count-the-threat.yaml"
35
+
36
+
37
+ def test_perception_modes_are_the_3x2_grid():
38
+ assert set(PERCEPTION_MODES) == {
39
+ "structured", "structured-clear",
40
+ "vision", "vision-clear",
41
+ "image", "image-clear",
42
+ }
43
+
44
+
45
+ @pytest.mark.parametrize("mode", PERCEPTION_MODES)
46
+ def test_scenario_config_accepts_every_mode(mode):
47
+ cfg = ScenarioConfig(name="cell", level="easy", fog_mode=mode)
48
+ assert cfg.fog_mode == mode
49
+
50
+
51
+ def test_compiled_level_reveal_map_and_channel_properties():
52
+ """`-clear` ⇒ reveal_map True; channel is independent of fog."""
53
+ c = compile_level(load_pack(PACKS / _PACK), "easy")
54
+ cases = {
55
+ "structured": (False, "structured"),
56
+ "structured-clear": (True, "structured"),
57
+ "vision": (False, "vision"),
58
+ "vision-clear": (True, "vision"),
59
+ "image": (False, "image"),
60
+ "image-clear": (True, "image"),
61
+ }
62
+ for mode, (reveal, channel) in cases.items():
63
+ c.fog_mode = mode
64
+ assert c.reveal_map is reveal, mode
65
+ assert c.obs_channel == channel, mode
66
+
67
+
68
+ def test_tmp_yaml_emits_reveal_map_only_for_clear_modes():
69
+ """The no-fog cells must write `reveal_map: true` into the engine
70
+ scenario YAML; the fogged cells must not mention it at all."""
71
+ c = compile_level(load_pack(PACKS / _PACK), "easy")
72
+ for mode in PERCEPTION_MODES:
73
+ c.fog_mode = mode
74
+ data = yaml.safe_load(Path(_scenario_to_tmp_yaml(c)).read_text())
75
+ if mode.endswith("-clear"):
76
+ assert data.get("reveal_map") is True, mode
77
+ else:
78
+ assert "reveal_map" not in data, mode
79
+
80
+
81
+ def test_no_fog_cell_reveals_what_the_fogged_cell_hides():
82
+ """End-to-end: an identical perception scenario, observe-only.
83
+ Under fog a stall sees nothing; under `reveal_map` every enemy is
84
+ observed and the whole map is explored — the perception control."""
85
+ pack = load_pack(PACKS / _PACK)
86
+ stall = lambda rs, Command: [Command.observe()] # noqa: E731
87
+
88
+ fogged = compile_level(pack, "medium")
89
+ fogged.fog_mode = "vision"
90
+ f = run_level(fogged, stall, seed=1)
91
+
92
+ clear = compile_level(pack, "medium")
93
+ clear.fog_mode = "vision-clear"
94
+ c = run_level(clear, stall, seed=1)
95
+
96
+ # Fog hides the enemies from an observe-only policy.
97
+ assert len(f.signals.enemies_seen_ids) == 0
98
+ assert f.signals.explored_percent < 50.0
99
+ # No-fog reveals every enemy and the entire map.
100
+ assert len(c.signals.enemies_seen_ids) > 0
101
+ assert c.signals.explored_percent > 99.0
102
+
103
+
104
+ def test_perception_sweep_expands_to_six_cells_per_level():
105
+ from openra_bench.run_eval import evaluate
106
+
107
+ out = evaluate(
108
+ [PACKS / _PACK], levels=["easy"], seeds=[1],
109
+ perception_sweep=True, dry_run=True,
110
+ )
111
+ expected = {
112
+ f"perception-count-the-threat:easy:{m}" for m in PERCEPTION_MODES
113
+ }
114
+ assert set(out["cells"]) == expected, out["cells"]
115
+
116
+
117
+ # ── Image-primary channel ────────────────────────────────────────────
118
+
119
+ def _render_state(mode: str) -> dict:
120
+ """A live render_state for the count-the-threat pack under `mode`."""
121
+ from openra_bench.rust_adapter import RustObsAdapter
122
+ from openra_rl_training.training.rust_env_pool import RustEnvPool
123
+
124
+ c = compile_level(load_pack(PACKS / _PACK), "medium")
125
+ c.fog_mode = mode
126
+ pool = RustEnvPool(size=1, scenario_path=_scenario_to_tmp_yaml(c))
127
+ ad = RustObsAdapter()
128
+ ad.observe(pool.acquire().reset(seed=1))
129
+ return ad.render_state()
130
+
131
+
132
+ def test_perception_labels_are_friendly_and_unique():
133
+ from openra_bench.prompt_v2 import perception_labels
134
+
135
+ labels = perception_labels(_render_state("image-clear"))
136
+ assert labels, "expected labels for own + enemy actors"
137
+ # Each engine id maps to a distinct legible handle.
138
+ assert len(set(labels.values())) == len(labels)
139
+ assert any(v.startswith("tank-") for v in labels.values())
140
+ assert any(v.startswith("enemy-") for v in labels.values())
141
+
142
+
143
+ def test_image_primary_briefing_has_no_coordinates():
144
+ """The redacted briefing must carry WHAT, never WHERE."""
145
+ import re
146
+
147
+ from openra_bench.prompt_v2 import (briefing_image_primary,
148
+ perception_labels)
149
+
150
+ rs = _render_state("image-clear")
151
+ text = briefing_image_primary(rs, perception_labels(rs))
152
+ # No `@(x,y)` coordinate tokens, and the enemy line is not enumerated.
153
+ assert not re.search(r"@\(\d+,\s*\d+\)", text), text
154
+ assert "8xe3" not in text and "1xbarr" not in text
155
+ assert "minimap" in text.lower()
156
+
157
+
158
+ def test_image_primary_minimap_labels_every_unit():
159
+ from openra_bench.minimap import render_tactical_minimap
160
+ from openra_bench.prompt_v2 import perception_labels
161
+
162
+ rs = _render_state("image-clear")
163
+ labels = perception_labels(rs)
164
+ img = render_tactical_minimap(rs, scale=3, unit_labels=labels)
165
+ assert img is not None and img.size[0] > 0
166
+
167
+
168
+ def test_to_commands_translates_labels_to_engine_ids():
169
+ """In the image channel the model passes minimap handles; the agent
170
+ maps them back to engine ids. Numeric ids pass straight through."""
171
+ import openra_train
172
+
173
+ from openra_bench.agent import _to_commands
174
+
175
+ Command = openra_train.Command
176
+ label_to_id = {"tank-1": "1004", "enemy-1": "1006"}
177
+ cmds = _to_commands(
178
+ [{"name": "attack_unit",
179
+ "arguments": {"unit_ids": ["tank-1"], "target_id": "enemy-1"}}],
180
+ Command, label_to_id,
181
+ )
182
+ assert '"1004"' in repr(cmds[0]) and '"1006"' in repr(cmds[0])
183
+ # Empty map (every other channel) — numeric ids untouched.
184
+ passthru = _to_commands(
185
+ [{"name": "stop", "arguments": {"unit_ids": [1004]}}], Command, {},
186
+ )
187
+ assert '"1004"' in repr(passthru[0])
188
+
189
+
190
+ def test_image_primary_tools_retype_handles_as_strings():
191
+ from openra_bench.agent import _image_primary_tools, _tool_schemas
192
+
193
+ base = _tool_schemas(["move_units", "attack_unit"])
194
+ img = _image_primary_tools(base)
195
+ for t in img:
196
+ props = t["function"]["parameters"]["properties"]
197
+ if "unit_ids" in props:
198
+ assert props["unit_ids"]["items"]["type"] == "string"
199
+ if "target_id" in props:
200
+ assert props["target_id"]["type"] == "string"
201
+ # The base schemas are deep-copied, not mutated.
202
+ assert (
203
+ base[0]["function"]["parameters"]["properties"]
204
+ ["unit_ids"]["items"]["type"] == "integer"
205
+ )