yxc20098 commited on
Commit
e3e91b7
·
1 Parent(s): d05021a

feat(bench): forbidden_tools + tool_violations_gte — strict-toolban / procedural-compliance primitive (BFCL V4 / τ²-bench / IFBench anchor)

Browse files

Pure bench-side compliance constraint over agent tool calls — no Rust
needed since this measures what the agent CHOSE to invoke, not what
the engine executed.

- Level / CompiledLevel gain an optional forbidden_tools: list[str].
- EpisodeSignals gains tools_called: dict[str, int] and tool_violations:
int. Tracked in eval_core.run_level by decoding each cmd's repr
(Rust enum stringifies Command::VariantName { … }) into snake_case
via _cmd_tool_name, so scripted and live-model policies are graded
by the same rule.
- New predicate tool_violations_gte: N for use in fail_condition
(typical: tool_violations_gte: 1 = zero-tolerance allowlist).
- _PHRASES translation: tool_violations_gte: 1 reads as 'you used a
forbidden tool (instant fail)'.
- 7 tests in tests/test_forbidden_tools.py cover cmd-repr decoder,
predicate semantics, schema roundtrip, and end-to-end live-engine
enforcement (violator policy LOSSES, allowed-only stall stays 0).

Unblocks B3 strict-toolban-fidelity-under-pressure + standalone
proc-strict-toolban-fidelity + proc-no-attack-passive-only +
proc-only-build-no-combat + proc-only-defend-no-attack (6+ packs from
one PR per SCENARIO_BACKLOG.md priority list).

openra_bench/eval_core.py CHANGED
@@ -78,6 +78,31 @@ class EpisodeResult:
78
  reward_vector: dict = field(default_factory=dict)
79
 
80
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
81
  def scripted_explore_agent(render_state: dict, Command: Any) -> list:
82
  """Baseline reference agent: walk every unit toward the nearest
83
  unexplored frontier cell. Exercises the move path; a useful
@@ -203,9 +228,23 @@ def run_level(
203
  interrupt_mode = bool(enabled_sig) and raw_env is not None and hasattr(
204
  raw_env, "step_until_event"
205
  )
 
 
 
 
 
 
206
  for turns in range(1, compiled.max_turns + 1):
207
  rs = adapter.render_state()
208
  cmds = agent_fn(rs, env.Command) or [env.Command.observe()]
 
 
 
 
 
 
 
 
209
  if not conceded:
210
  conceded = any("Surrender" in repr(c) for c in cmds)
211
  interrupt = None
 
78
  reward_vector: dict = field(default_factory=dict)
79
 
80
 
81
+ _CMD_NAME_RE = __import__("re").compile(r"Command::([A-Z][A-Za-z0-9]*)")
82
+
83
+
84
+ def _cmd_tool_name(cmd: Any) -> str | None:
85
+ """Decode the snake_case tool name from a Command repr.
86
+
87
+ The Rust pyo3 Command enum stringifies as ``Command::VariantName { … }``
88
+ or ``Command::VariantName``. We extract the variant and convert to
89
+ snake_case to match the bench's tool-name vocabulary (and the
90
+ `tools:` / `forbidden_tools:` allowlist keys in YAML). Returns
91
+ ``None`` for anything that doesn't match — defensive; never raises.
92
+ """
93
+ m = _CMD_NAME_RE.search(repr(cmd))
94
+ if not m:
95
+ return None
96
+ variant = m.group(1)
97
+ # CamelCase → snake_case (MoveUnits → move_units, AttackUnit → attack_unit)
98
+ out: list[str] = []
99
+ for i, ch in enumerate(variant):
100
+ if i and ch.isupper():
101
+ out.append("_")
102
+ out.append(ch.lower())
103
+ return "".join(out)
104
+
105
+
106
  def scripted_explore_agent(render_state: dict, Command: Any) -> list:
107
  """Baseline reference agent: walk every unit toward the nearest
108
  unexplored frontier cell. Exercises the move path; a useful
 
228
  interrupt_mode = bool(enabled_sig) and raw_env is not None and hasattr(
229
  raw_env, "step_until_event"
230
  )
231
+ # Strict-toolban / procedural-compliance accounting: any cmd whose
232
+ # tool name is in compiled.forbidden_tools increments
233
+ # signals.tool_violations (read by the tool_violations_gte
234
+ # predicate). Tracked here so scripted and live-model policies
235
+ # are graded by the exact same rule.
236
+ forbidden = {str(t).lower() for t in (compiled.forbidden_tools or [])}
237
  for turns in range(1, compiled.max_turns + 1):
238
  rs = adapter.render_state()
239
  cmds = agent_fn(rs, env.Command) or [env.Command.observe()]
240
+ for _cmd in cmds:
241
+ _tn = _cmd_tool_name(_cmd)
242
+ if _tn:
243
+ adapter.signals.tools_called[_tn] = (
244
+ adapter.signals.tools_called.get(_tn, 0) + 1
245
+ )
246
+ if _tn in forbidden:
247
+ adapter.signals.tool_violations += 1
248
  if not conceded:
249
  conceded = any("Surrender" in repr(c) for c in cmds)
250
  interrupt = None
openra_bench/game_knowledge.py CHANGED
@@ -201,6 +201,11 @@ _PHRASES: dict[str, Any] = {
201
  + "+".join((v or {}).get("types", []))
202
  + f" at the base near ({(v or {}).get('x')},{(v or {}).get('y')})"
203
  ),
 
 
 
 
 
204
  }
205
 
206
 
 
201
  + "+".join((v or {}).get("types", []))
202
  + f" at the base near ({(v or {}).get('x')},{(v or {}).get('y')})"
203
  ),
204
+ "tool_violations_gte": lambda v: (
205
+ "you used a forbidden tool (instant fail)"
206
+ if int(v) <= 1
207
+ else f"you used a forbidden tool ≥{v} times"
208
+ ),
209
  }
210
 
211
 
openra_bench/rust_adapter.py CHANGED
@@ -131,6 +131,15 @@ class EpisodeSignals:
131
  # waypoint_sequence's ordered-visit progress, keyed by sequence id).
132
  # Reset for free: EpisodeSignals is reconstructed each episode.
133
  seq_progress: dict = field(default_factory=dict)
 
 
 
 
 
 
 
 
 
134
  # Outcome is synthesized (Rust has no result field): a scenario is
135
  # "won" when all enemy buildings have been discovered AND/OR all
136
  # enemy units neutralized — refined per-scenario in Phase 2 rubrics.
 
131
  # waypoint_sequence's ordered-visit progress, keyed by sequence id).
132
  # Reset for free: EpisodeSignals is reconstructed each episode.
133
  seq_progress: dict = field(default_factory=dict)
134
+ # Per-episode tool-use accounting for the strict-toolban / procedural-
135
+ # compliance family. tools_called counts each tool name the agent
136
+ # invoked this episode; tool_violations counts how many of those calls
137
+ # were on the scenario's forbidden_tools list. The `tool_violations_gte`
138
+ # predicate reads from here (typically as a fail clause). Tracking is
139
+ # bench-side (see eval_core.run_level), so scripted policies are
140
+ # graded by the same rule as live models.
141
+ tools_called: dict[str, int] = field(default_factory=dict)
142
+ tool_violations: int = 0
143
  # Outcome is synthesized (Rust has no result field): a scenario is
144
  # "won" when all enemy buildings have been discovered AND/OR all
145
  # enemy units neutralized — refined per-scenario in Phase 2 rubrics.
openra_bench/scenarios/schema.py CHANGED
@@ -110,6 +110,12 @@ class Level(BaseModel):
110
  description="Per-level economy budget (overrides pack default; "
111
  "engine default 5000 when unset everywhere).",
112
  )
 
 
 
 
 
 
113
 
114
 
115
  class ScenarioConfig(BaseModel):
@@ -155,6 +161,7 @@ class CompiledLevel(BaseModel):
155
  fog_mode: str = "vision"
156
  config_name: str | None = None
157
  objective_coords: Literal["exact", "relative"] = "exact"
 
158
 
159
 
160
  class ScenarioPack(BaseModel):
@@ -222,6 +229,7 @@ class ScenarioPack(BaseModel):
222
  else self.starting_cash,
223
  map_supported=map_supported,
224
  objective_coords=lvl.objective_coords,
 
225
  )
226
 
227
  def config_names(self) -> list[str]:
 
110
  description="Per-level economy budget (overrides pack default; "
111
  "engine default 5000 when unset everywhere).",
112
  )
113
+ # Procedural-compliance family: any agent command whose tool name is
114
+ # in this list increments signals.tool_violations (bench-side track,
115
+ # see eval_core). Use as a fail clause via the
116
+ # `tool_violations_gte` predicate — usually `tool_violations_gte: 1`
117
+ # for a strict zero-tolerance allowlist. Empty list = no constraint.
118
+ forbidden_tools: list[str] = Field(default_factory=list)
119
 
120
 
121
  class ScenarioConfig(BaseModel):
 
161
  fog_mode: str = "vision"
162
  config_name: str | None = None
163
  objective_coords: Literal["exact", "relative"] = "exact"
164
+ forbidden_tools: list[str] = Field(default_factory=list)
165
 
166
 
167
  class ScenarioPack(BaseModel):
 
229
  else self.starting_cash,
230
  map_supported=map_supported,
231
  objective_coords=lvl.objective_coords,
232
+ forbidden_tools=list(lvl.forbidden_tools or []),
233
  )
234
 
235
  def config_names(self) -> list[str]:
openra_bench/scenarios/win_conditions.py CHANGED
@@ -189,6 +189,16 @@ _PREDICATES: dict[str, Callable[[WinContext, Any], bool]] = {
189
  <= float(v.get("radius", 5)) ** 2
190
  )
191
  >= int(v.get("count", 1)),
 
 
 
 
 
 
 
 
 
 
192
  }
193
 
194
  LEAF_KEYS = frozenset(_PREDICATES)
 
189
  <= float(v.get("radius", 5)) ** 2
190
  )
191
  >= int(v.get("count", 1)),
192
+ # Procedural-compliance / strict-toolban family: triggers when the
193
+ # agent has invoked tools on the level's `forbidden_tools` list at
194
+ # least N times this episode. Typical use is in a fail clause as
195
+ # `tool_violations_gte: 1` (zero-tolerance), but >1 also lets a pack
196
+ # tolerate up to N slips before failing. The counter is maintained
197
+ # by eval_core.run_level — see ScenarioPack/Level.forbidden_tools.
198
+ "tool_violations_gte": lambda c, v: getattr(
199
+ c.signals, "tool_violations", 0
200
+ )
201
+ >= int(v),
202
  }
203
 
204
  LEAF_KEYS = frozenset(_PREDICATES)
tests/test_forbidden_tools.py ADDED
@@ -0,0 +1,170 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """forbidden_tools / tool_violations: bench-side procedural-compliance.
2
+
3
+ Tests the new schema field + signal + predicate end-to-end. The
4
+ real-world anchor is BFCL V4 / τ²-bench / IFBench: the agent has a
5
+ strict allowlist and any disallowed tool call must trip a fail clause.
6
+
7
+ Three things must hold:
8
+
9
+ 1. The cmd-repr → tool-name decoder maps every Command variant to its
10
+ snake_case allowlist key (move_units, attack_unit, …).
11
+ 2. eval_core.run_level increments signals.tool_violations exactly once
12
+ per disallowed call, regardless of whether the policy is scripted or
13
+ wrapped by ModelAgent.
14
+ 3. The `tool_violations_gte` predicate reads that counter and fails the
15
+ episode (typical use: `tool_violations_gte: 1` in fail_condition).
16
+ """
17
+
18
+ from __future__ import annotations
19
+
20
+ import types
21
+ from pathlib import Path
22
+
23
+ import pytest
24
+
25
+ from openra_bench.eval_core import _cmd_tool_name, run_level
26
+ from openra_bench.scenarios import load_pack
27
+ from openra_bench.scenarios.loader import compile_level
28
+ from openra_bench.scenarios.win_conditions import WinContext, evaluate
29
+
30
+ PACKS = Path(__file__).parent.parent / "openra_bench" / "scenarios" / "packs"
31
+
32
+
33
+ # ── 1. cmd-repr decoder ───────────────────────────────────────────────────
34
+
35
+
36
+ def test_cmd_tool_name_extracts_snake_case_from_command_repr():
37
+ """The Rust Command enum stringifies as `Command::VariantName { … }`
38
+ or `Command::VariantName`. _cmd_tool_name must produce the bench's
39
+ snake_case tool key for every variant — covers the cmd type names
40
+ actually observed live on the engine in this session."""
41
+ # MoveUnits with payload
42
+ class _Stub:
43
+ def __init__(self, r: str):
44
+ self.r = r
45
+
46
+ def __repr__(self) -> str:
47
+ return self.r
48
+
49
+ assert _cmd_tool_name(_Stub('Command::MoveUnits { unit_ids: ["1"], '
50
+ 'target_x: 10, target_y: 10 }')) == "move_units"
51
+ # bare variant
52
+ assert _cmd_tool_name(_Stub("Command::Observe")) == "observe"
53
+ # multi-word variant
54
+ assert _cmd_tool_name(_Stub('Command::AttackUnit { unit_ids: ["1"], '
55
+ 'target_id: "2" }')) == "attack_unit"
56
+ assert _cmd_tool_name(_Stub('Command::AttackMove { … }')) == "attack_move"
57
+ assert _cmd_tool_name(_Stub('Command::Stop { unit_ids: ["1"] }')) == "stop"
58
+ assert _cmd_tool_name(_Stub("Command::Surrender")) == "surrender"
59
+ assert _cmd_tool_name(_Stub('Command::SetStance { stance: 2 }')) == "set_stance"
60
+ assert _cmd_tool_name(_Stub('Command::PlaceBuilding { … }')) == "place_building"
61
+ # Non-Command reprs → None, never raise
62
+ assert _cmd_tool_name(_Stub("something else")) is None
63
+ assert _cmd_tool_name(_Stub("")) is None
64
+
65
+
66
+ # ── 2. predicate reads from signals.tool_violations ───────────────────────
67
+
68
+
69
+ def test_tool_violations_gte_predicate_reads_signals():
70
+ """tool_violations_gte: N is true iff signals.tool_violations >= N."""
71
+ sig = types.SimpleNamespace(tool_violations=0)
72
+ ctx = WinContext(signals=sig, render_state={"units_summary": []})
73
+ assert evaluate({"tool_violations_gte": 1}, ctx) is False
74
+ sig.tool_violations = 1
75
+ assert evaluate({"tool_violations_gte": 1}, ctx) is True
76
+ sig.tool_violations = 5
77
+ assert evaluate({"tool_violations_gte": 3}, ctx) is True
78
+ assert evaluate({"tool_violations_gte": 6}, ctx) is False
79
+
80
+
81
+ def test_tool_violations_default_zero_for_legacy_signals():
82
+ """Signals that predate the field (e.g. raw types.SimpleNamespace
83
+ in unit tests) should default to 0, not raise."""
84
+ sig = types.SimpleNamespace() # no tool_violations attr
85
+ ctx = WinContext(signals=sig, render_state={"units_summary": []})
86
+ assert evaluate({"tool_violations_gte": 1}, ctx) is False
87
+
88
+
89
+ # ── 3. _PHRASES has a translation (suite invariant elsewhere too) ─────────
90
+
91
+
92
+ def test_tool_violations_gte_has_a_phrase_translation():
93
+ from openra_bench.game_knowledge import _PHRASES
94
+
95
+ assert "tool_violations_gte" in _PHRASES
96
+ p = _PHRASES["tool_violations_gte"]
97
+ assert "forbidden tool" in p(1)
98
+ assert "≥3" in p(3)
99
+
100
+
101
+ # ── 4. forbidden_tools roundtrips through schema ──────────────────────────
102
+
103
+
104
+ def test_forbidden_tools_roundtrips_through_compiled_level():
105
+ """A pack that declares forbidden_tools on a Level surfaces it on
106
+ the CompiledLevel. CompiledLevel.forbidden_tools is what
107
+ eval_core.run_level reads."""
108
+ # Use an existing pack and inject forbidden_tools at the Level model
109
+ # level (we don't need to write a new YAML to verify roundtripping).
110
+ pack = load_pack(PACKS / "custom-map-no-enemy.yaml")
111
+ pack.levels["easy"].forbidden_tools = ["attack_unit", "attack_move"]
112
+ c = compile_level(pack, "easy")
113
+ assert c.forbidden_tools == ["attack_unit", "attack_move"]
114
+
115
+
116
+ # ── 5. live engine: forbidden cmds increment counter; allowed don't ───────
117
+
118
+
119
+ def test_forbidden_tools_counted_live_via_run_level():
120
+ """The end-to-end test: a scripted policy that issues an
121
+ attack_unit (forbidden) on every turn must accumulate
122
+ tool_violations > 0 by episode end, and the tool_violations_gte:1
123
+ fail clause must fire ⇒ outcome == 'loss'."""
124
+ pytest.importorskip("openra_train")
125
+
126
+ pack = load_pack(PACKS / "custom-map-no-enemy.yaml")
127
+ # Hot-patch the easy level: forbid attack_unit + put it in fail.
128
+ pack.levels["easy"].forbidden_tools = ["attack_unit", "attack_move"]
129
+ from openra_bench.scenarios.win_conditions import WinCondition
130
+
131
+ pack.levels["easy"].fail_condition = WinCondition(
132
+ any_of=[
133
+ {"after_ticks": 10_000},
134
+ {"tool_violations_gte": 1},
135
+ ]
136
+ )
137
+ c = compile_level(pack, "easy")
138
+ assert c.forbidden_tools == ["attack_unit", "attack_move"]
139
+
140
+ def violator(rs, Command):
141
+ # observe + a single forbidden attack_unit on a non-existent
142
+ # target id (the engine warns, but the bench tracks the tool
143
+ # name BEFORE the engine evaluates it — so the violation
144
+ # counts even when the engine refuses the order).
145
+ return [Command.observe(), Command.attack_unit(["1"], target_id="99999")]
146
+
147
+ res = run_level(c, violator, seed=1)
148
+ # bench-side accounting must have flagged ≥1 forbidden call …
149
+ assert res.signals.tool_violations >= 1, res.signals.tools_called
150
+ # … and the fail predicate must have fired.
151
+ assert res.outcome == "loss", res.outcome
152
+
153
+
154
+ def test_allowed_tools_dont_increment_violations():
155
+ """The same harness with NO forbidden_tools and only allowed cmds
156
+ must yield tool_violations == 0 (no false positives)."""
157
+ pytest.importorskip("openra_train")
158
+
159
+ pack = load_pack(PACKS / "custom-map-no-enemy.yaml")
160
+ c = compile_level(pack, "easy")
161
+ assert c.forbidden_tools == []
162
+
163
+ def stall(rs, Command):
164
+ return [Command.observe()]
165
+
166
+ res = run_level(c, stall, seed=1)
167
+ assert res.signals.tool_violations == 0
168
+ # Tools_called should at least have observe (the bench's default
169
+ # fallback and the policy's explicit call).
170
+ assert "observe" in res.signals.tools_called