Eval resilience layer for real OpenRouter runs

resilience.py (pure, thread-safe, fully unit-tested):
- RetryPolicy + retry_call: bounded exponential backoff, honors
Retry-After, transient(429/5xx/timeout) vs FatalProviderError
- RateLimiter: global min-interval throttle across the pool
- CostMeter: token/USD accumulation + hard cap → BudgetExceeded
- RunJournal: append-only completed-episode journal, torn-line tolerant

providers: ProviderConfig gains max_retries/retry_*/qps/price_*/
max_history_turns; ChatReply.usage; OpenAICompatibleProvider wraps the
POST in retry+ratelimit, captures usage, meters cost, raises on budget;
make_provider injects a shared limiter+meter

agent: ModelAgent._window — sliding wire-history window (system + last
N user groups, pairing intact); self.history stays full for playback

run_eval.evaluate: shared meter/limiter + one shared provider;
checkpoint journal + --resume (lossless skip+refold via _finalize/
_ScoreShim); --max-spend graceful truncate-and-finalize; --smoke;
--dry-run; incremental report flush + progress heartbeat. CLI flags
wired. Determinism contract excludes the wall-clock run_id label.

tests: test_resilience.py (8) + pinned run_id in the concurrency
determinism test — 345 passed, 1 skipped

SCENARIO_QUALITY.md

@@ -0,0 +1,375 @@
+# OpenRA-Bench — Scenario Design-Quality Audit & Scoring
+
+Analysis only. No scenario files changed. Prepared 2026-05-17.
+Scope: the 48 `status: active` packs (every pack without
+`status: quarantine`); TEMPLATE and all `cat-c*-02..05` quarantine
+packs excluded. Grounded against `SCENARIO_AUDIT.md`,
+`win_conditions.py`, `scoring.py`, `goal_tracker.py`,
+`eval_core.run_level`, the schema, and `openra_env.game_data`
+(verified unit/building costs, prerequisites, tech tree).
+
+---
+
+## 1. Executive Summary
+
+The capability taxonomy and predicate grammar are sound, and the
+hand-authored families (perception-*, reasoning-*, action-*,
+artofwar-*, strict-*) are mostly genuine, well-grounded decision
+problems. But the active set has four structural quality problems that
+materially weaken it as a discriminating benchmark:
+
+1. **Loss is impossible in ~70% of active packs.** Only a
+   `fail_condition` can produce `loss`; `run_level` leaves the outcome
+   at `draw` when the win predicate is unmet (eval_core.py L174, L243).
+   Every perception-*, reasoning-*, action-*, economy-*, custom-map,
+   strict-sequence, building-and-planning, and most `cat-*` pack has
+   **no real fail_condition** (the `cat-*` ones carry
+   `units_lost_lte: -1`, which is *never* satisfiable because
+   `units_lost = max(0, …) ≥ 0` — a dead no-op). Consequence: a
+   do-nothing agent and a near-miss agent both score `draw` (0.5
+   outcome); the binary outcome only separates win vs. not-win.
+   Partial credit via `objective_progress` rescues *some*
+   discrimination, but the headline outcome axis is degenerate on the
+   majority of the suite.
+
+2. **The `cat-c*` `-00`/`-01` kept representatives are near-exact
+   duplicates, not distinct decisions.** `diff cat-c1-00 cat-c1-01`
+   differs only in id/title and ±1 `explored_pct` / ±200 ticks.
+   c5↔c6, c7↔c8, c9↔c11, c4↔c3 are the same decision with relabelled
+   meta. The two kept variants per category add **zero** distinct
+   capability — keep one, cut the second.
+
+3. **`always-movement-wins` degeneracy in the navigation/perception
+   `cat-*` and several hand-authored packs.** Where the only win
+   clauses are `explored_pct_gte` + `within_ticks` with no fail and a
+   generous clock, the *baseline scripted frontier agent*
+   (`scripted_explore_agent`) already wins — these do not separate
+   strong from weak models, only "moves" from "doesn't move."
+
+4. **A genuine solvability defect in `strict-production-bom` hard**:
+   the budget cannot cover the required tech path (see §2). Several
+   "spend/build" packs are solvable but only test "spend the obvious
+   thing," not an allocation trade-off, because the engine has no
+   harvest income (documented S0/S1) so there is never scarcity
+   pressure beyond a single fixed `starting_cash`.
+
+**Bottom line:** ~9 packs should be CUT now (8 redundant `cat-*`
+duplicates + 1 unsolvable level needs a budget fix or cut), ~16 need a
+FIX (almost all: "add a real fail_condition so loss≠draw"), and the
+hand-authored perception/reasoning/artofwar/strict-bom core is KEEP.
+The single highest-leverage repo-wide fix is adding a real
+`fail_condition` (timeout-loss or attrition-loss) to every active pack
+so the outcome axis stops collapsing loss into draw.
+
+---
+
+## 2. Solvability spot-checks (verified against game_data)
+
+Verified building/unit costs & prereqs: `powr` 300 (no prereq, +100
+pwr), `tent`/`barr` 500 (prereq `powr`, −20 pwr), `proc` 1400 (prereq
+`powr`), `pbox` 600 (prereq `tent`), `gun` 800 (prereq `weap`),
+`tsla` 1200 (prereq `weap`, −75 pwr), `weap` War Factory (~2000),
+`fact` 2000, e1 100 / e3 300 (prereq `barr|tent`).
+
+- **strict-production-bom HARD — UNSOLVABLE on budget.** Spec:
+  `tsla` + 3 e1 + 2 e3, `starting_cash: 4200`, agent faction soviet.
+  `tsla` requires `weap` (≈2000) which is **not pre-placed and not in
+  the actor list**. Minimum spend: barr 500 + weap 2000 + tsla 1200 +
+  3·e1 300 + 2·e3 600 = **4600 > 4200**. Also tsla is −75 power with
+  only one `powr` (+100) minus barr/weap drain → likely power-starved.
+  → **FIX (raise hard `starting_cash` to ~5200 and pre-place or
+  budget `weap`) or CUT the hard level.** Easy/medium are solvable
+  (barr 500 + 3 e1; barr + 3 e1 + 2 e3 within 2200/3000) and a good
+  BFCL-style fidelity test.
+- **building-and-planning** easy (build powr+tent, cash 6000),
+  medium (2 pbox in region, tent pre-placed, cash 5000: 2·600+powr),
+  hard (MCV→deploy fact, 2 buildings near (60,20), cash 4000):
+  all solvable; hard is tight but feasible. KEEP.
+- **economy-investment / economy-time-box / economy-force-buildup**:
+  all levels solvable on the given `starting_cash` (2nd proc 1400 +
+  2nd powr 300 ≤ budgets; e1 at 100 makes `own_units_gte` trivially
+  affordable). Solvable but **decision-thin** — no harvest income
+  means "wide vs deep" is not a real trade (both fit the budget); the
+  test reduces to "issue the build commands." FIX (see §4).
+- **adversarial-* / artofwar-* / reasoning-risk-route /
+  reasoning-frontier-commit / perception-***: policies exist
+  (kite/focus-fire; bait then run; take the safe detour; commit to
+  the correct fog region). Solvable; these are the strongest packs.
+- **cat-c3/c4** (`building_total_gte` + `power_surplus_gte: 0`):
+  solvable (powr is +100, free of prereq; spam powr). But
+  `power_surplus_gte: 0` + `building_total_gte` is trivially gamed by
+  building only power plants — see Rigor notes.
+- **rush-hour / strategy-gauntlet / strategy-twobody / strategy-
+  dilemma**: 14 scattered enemy infantry vs ~32 agent combat units,
+  `units_killed_gte` 3–9; solvable, but with stance-3 auto-fighting
+  units the agent barely has to act — see Discrimination.
+
+---
+
+## 3. Scored Table (1–5; 5 = best)
+
+solv = solvability · disc = discrimination · grnd = grounding ·
+rigor = non-gameable + principled scaling · ovr = overall
+
+| pack id | capability | solv | disc | grnd | rigor | ovr | verdict | one-line note |
+|---|---|---|---|---|---|---|---|---|
+| action-multiunit-coordination | action | 5 | 4 | 5 | 4 | **4** | KEEP | Real parallel-control test; add fail_condition so non-win≠draw. |
+| action-sequenced-execution | action | 5 | 4 | 5 | 4 | **4** | KEEP | after_ticks gates encode ordering well; add timeout fail_condition. |
+| adversarial-duel | adversarial | 4 | 4 | 5 | 4 | **4** | KEEP | Has real fail (own_units_gte:1); reactive enemy = unique RTS value. |
+| adversarial-siege | adversarial | 4 | 4 | 5 | 4 | **4** | KEEP | Same model as duel; entrenched defender variant — distinct enough. |
+| adversarial-skirmish | adversarial | 4 | 4 | 5 | 4 | **4** | KEEP | Outnumbered/maneuver variant; genuine asymmetric tactic. |
+| artofwar-decoy-sacrifice | reasoning | 4 | 5 | 5 | 5 | **5** | KEEP | Loss cap forces "spend bait, save army" — true credit assignment. |
+| artofwar-indirect-approach | reasoning | 4 | 5 | 5 | 5 | **5** | KEEP | units_lost_lte:0 + hazard short path = clean long-horizon test. |
+| artofwar-lure-the-tiger | reasoning | 3 | 4 | 5 | 4 | **4** | FIX | Two-phase intent good, but win has no clause proving the lure happened — reaching region by any survivable route also wins; add an ordering/after_ticks gate. |
+| artofwar-sequenced-citadel | reasoning | 5 | 4 | 5 | 4 | **4** | KEEP | after_ticks+within_ticks band genuinely grades order; solid. |
+| building-and-planning | reasoning | 4 | 4 | 5 | 4 | **4** | KEEP | 3 genuinely different decisions (tech dep / placement / relocate); add fail_condition. |
+| custom-map-no-enemy | perception | 5 | 2 | 4 | 2 | **2** | FIX | Pure nav, no enemy, no fail → baseline move-agent wins easy/medium; only hard (all_units_in_region, tight clock) discriminates. Keep hard, cut/merge easy+medium or add fail. |
+| economy-force-buildup | reasoning | 5 | 2 | 3 | 2 | **2** | FIX | e1@100 makes own_units_gte trivial; no scarcity (no harvest). Tests "click build," not allocation. Make win require a building mix + fail on idle. |
+| economy-investment | reasoning | 5 | 3 | 4 | 3 | **3** | FIX | Best of the economy packs but "wide vs deep" isn't a real trade w/o income; both paths fit budget. Tighten cash so paths are mutually exclusive; add fail. |
+| economy-time-box | reasoning | 5 | 2 | 3 | 2 | **2** | FIX | Near-duplicate of force-buildup + a building_total bar; same triviality. Merge into economy-investment or harden. |
+| perception-frontier-reading | perception | 5 | 3 | 5 | 3 | **3** | FIX | Strong concept; but easy/medium = explored_pct only, no fail → baseline frontier agent already wins. Hard (decoys + units_lost_lte:1) is the real test. Add fail to easy/medium. |
+| perception-target-vs-fog | perception | 5 | 4 | 5 | 4 | **4** | KEEP | buildings_discovered (not coverage) defeats the blind-sweep agent; medium/hard force inference. Add timeout fail. |
+| reasoning-frontier-commit | reasoning | 5 | 4 | 5 | 4 | **4** | KEEP | Single scout = un-hedgeable commitment; decoy buildings vs units is a clean discriminator. |
+| reasoning-risk-route | reasoning | 5 | 5 | 5 | 5 | **5** | KEEP | units_lost_lte:0 + lethal short path + clock = textbook non-gameable risk/route. Exemplary. |
+| rush-hour | action | 5 | 2 | 3 | 2 | **2** | FIX | ~32 stance-3 auto-fighting agent units vs 14 scattered enemy; kill 3/6/9 in 40 turns is near-automatic. Low discrimination; not load-bearing. Cut to a real sweep (fewer units, coverage+kill, fail on timeout). |
+| strategy-dilemma | reasoning | 4 | 2 | 3 | 2 | **2** | FIX | Win = buildings_discovered_gte:1 only → any scout that bumps the enemy base wins; "risk dilemma" framing not enforced by the predicate. Replace win with reach_region + units_lost_lte. |
+| strategy-gauntlet | reasoning | 4 | 2 | 3 | 2 | **2** | FIX | Same defect: buildings_discovered_gte:1 with a huge friendly force; corridor never has to be solved. Re-spec to reach_region + attrition. |
+| strategy-twobody | action | 4 | 2 | 3 | 2 | **2** | FIX | Two pre-split friendly blobs already near the enemy base; discover-1-building wins with no real coordination. Re-spec to all_units_in_region (true rendezvous). |
+| strict-production-bom | action | 3 | 5 | 5 | 5 | **4** | FIX | Excellent BFCL-style exact-spec test (unit_type_count_eq + overproduction fail). HARD level unsolvable on 4200 cash (needs weap+tsla≈4600) — raise cash or cut hard. |
+| strict-sequence | action | 5 | 4 | 5 | 4 | **4** | KEEP | Tool allowlist + after_ticks band genuinely tests API-fidelity ordering; add timeout fail so a stalled agent loses, not draws. |
+| cat-c1-00 (Frontier Scouting) | perception | 5 | 3 | 4 | 3 | **3** | KEEP | Representative of C1; explored_pct+clock, hard adds units_lost_lte:0. OK as the single C1 kept level. |
+| cat-c1-01 | perception | 5 | 2 | 4 | 2 | **2** | CUT | ±1 explored_pct / ±200 ticks vs cat-c1-00. Pure duplicate. |
+| cat-c2-00 (Threat Enumeration) | perception | 5 | 3 | 4 | 3 | **3** | KEEP | enemies_discovered + units_lost_lte is a real perception-under-decoy test; keep one. |
+| cat-c2-01 | perception | 5 | 2 | 4 | 2 | **2** | CUT | Duplicate of cat-c2-00 (only tick deltas). |
+| cat-c3-00 (Tech Critical Path) | reasoning | 4 | 2 | 4 | 2 | **2** | FIX | building_total_gte + power_surplus_gte:0 is gamed by spamming powr (each +100, no prereq) — doesn't test the tech *path*. Require has_building tent/proc to force the dependency. |
+| cat-c3-01 | reasoning | 4 | 2 | 4 | 2 | **2** | CUT | Duplicate of cat-c3-00. |
+| cat-c4-00 (Power-Budget Online) | reasoning | 4 | 2 | 4 | 2 | **2** | FIX | Same gameability as c3 (powr-spam satisfies both clauses); near-identical to c3. Merge c3+c4 into one fixed pack. |
+| cat-c4-01 | reasoning | 4 | 2 | 4 | 2 | **2** | CUT | Duplicate of cat-c4-00 (and of c3). |
+| cat-c5-00 (Budget Allocation) | reasoning | 5 | 2 | 3 | 2 | **2** | FIX | powr n:2 + building_total bar, no income → "build 5 cheap buildings"; not an indivisible-budget trade. Same as c6. |
+| cat-c5-01 | reasoning | 5 | 2 | 3 | 2 | **2** | CUT | Duplicate of cat-c5-00. |
+| cat-c6-00 (Time-Boxed Deploy) | reasoning | 5 | 2 | 3 | 2 | **2** | FIX | own_units_gte+building_total; e1@100 trivial; duplicate decision of c5/economy-time-box. Merge. |
+| cat-c6-01 | reasoning | 5 | 2 | 3 | 2 | **2** | CUT | Duplicate of cat-c6-00. |
+| cat-c7-00 (Defensive-Direction) | perception | 4 | 3 | 4 | 3 | **3** | KEEP | building_in_region for pbox forces a directional commit; pbox needs tent (pre-placed) — solvable. Keep one. |
+| cat-c7-01 | perception | 4 | 2 | 4 | 2 | **2** | CUT | Duplicate of cat-c7-00. |
+| cat-c8-00 (Base-Placement) | perception | 4 | 3 | 4 | 3 | **3** | FIX | Same family as c7 (place building in inferred region). Distinct enough to keep one but merge c7/c8 conceptually; ensure region is reachable/buildable. |
+| cat-c8-01 | perception | 4 | 2 | 4 | 2 | **2** | CUT | Duplicate of cat-c8-00. |
+| cat-c9-00 (Commit-vs-Retreat) | reasoning | 4 | 3 | 4 | 3 | **3** | KEEP | units_killed + units_lost_lte is a real risk call; no real fail (dead −1) — FIX-lite: make fail = units_lost_gte cap. |
+| cat-c9-01 | reasoning | 4 | 2 | 4 | 2 | **2** | CUT | Duplicate of cat-c9-00. |
+| cat-c10-00 (Force Coordination) | action | 5 | 3 | 4 | 3 | **3** | KEEP | all_units_in_region + units_lost_lte = genuine rendezvous; better-spec'd than strategy-twobody. Keep one. |
+| cat-c10-01 | action | 5 | 2 | 4 | 2 | **2** | CUT | Duplicate of cat-c10-00. |
+| cat-c11-00 (Tempo Window) | reasoning | 4 | 3 | 4 | 3 | **3** | KEEP | after_ticks + units_killed + within_ticks genuinely encodes a window; only tempo coverage — keep one. |
+| cat-c11-01 | reasoning | 4 | 2 | 4 | 2 | **2** | CUT | Duplicate of cat-c11-00 (and decision-twin of c9). |
+| cat-c12-00 (Error Recovery) | reasoning | 4 | 2 | 3 | 2 | **2** | FIX | "Replan after setback" but nothing is destroyed at start — it's just a timed build with after_ticks:1200. Not a recovery test; either script a mid-episode loss or CUT. |
+| cat-c12-01 | reasoning | 4 | 2 | 3 | 2 | **2** | CUT | Duplicate of cat-c12-00. |
+
+Aggregate: KEEP 19 · FIX 18 · CUT 11. (TEMPLATE excluded.)
+
+---
+
+## 4. CUT list (concrete — safe to delete now)
+
+These add no distinct capability and/or are non-discriminating:
+
+1. **`cat-c1-01`** — byte-near duplicate of cat-c1-00 (±1 pct/±200 tk).
+2. **`cat-c2-01`** — duplicate of cat-c2-00.
+3. **`cat-c3-01`** — duplicate of cat-c3-00.
+4. **`cat-c4-01`** — duplicate of cat-c4-00 (and decision-twin of c3).
+5. **`cat-c5-01`** — duplicate of cat-c5-00.
+6. **`cat-c6-01`** — duplicate of cat-c6-00 (decision-twin of c5).
+7. **`cat-c7-01`** — duplicate of cat-c7-00.
+8. **`cat-c8-01`** — duplicate of cat-c8-00.
+9. **`cat-c9-01`** — duplicate of cat-c9-00.
+10. **`cat-c10-01`** — duplicate of cat-c10-00.
+11. **`cat-c11-01`** — duplicate of cat-c11-00.
+12. **`cat-c12-01`** — duplicate of cat-c12-00.
+
+Net: keep exactly one representative per `cat-c*` category (the `-00`),
+deleting all 12 `-01` actives — removes 12 files with zero capability
+loss. (Category-merge candidates, can be done after: fold c4→c3, c6→c5,
+c8→c7, c11→c9 since each pair is the same predicate decision.)
+
+Additionally **CUT or hard-FIX**: `cat-c12-00` (not actually an
+error-recovery test — nothing fails mid-episode; it is a delayed timed
+build), and `strict-production-bom` HARD level (unsolvable on 4200
+cash; either raise to ~5200 + pre-place `weap`, or delete the hard
+level and keep easy/medium).
+
+---
+
+## 5. The repo-wide FIX (do this first, highest leverage)
+
+**Add a real `fail_condition` to every active pack that lacks one.**
+Today `run_level` only emits `loss` if a `fail_condition` evaluates
+true; otherwise an unmet win stays `draw` (eval_core L174/L243). With
+no fail, a do-nothing agent draws and a 95%-there agent draws — the
+outcome axis cannot rank effort. The `cat-*` `units_lost_lte: -1`
+"fail" is a dead no-op (units_lost ≥ 0 always).
+
+Mechanical, non-gameable fixes using the *existing* vocabulary:
+
+- Timed tasks (perception-*, reasoning-*, action-*, economy-*,
+  strict-*, building-and-planning, cat-c1/c3/c4/c5/c6/c7/c8/c12):
+  `fail_condition: {not: {within_ticks: <max_ticks>}}` — i.e. running
+  out the clock without meeting the win is a *loss*, not a draw.
+- Attrition tasks (anything with a `units_lost_lte: N` win clause):
+  add `fail_condition: {not: {units_lost_lte: N}}` so exceeding the
+  cap loses immediately (also prunes wasted ticks).
+- `cat-*`: replace the dead `units_lost_lte: -1` with the timeout-loss
+  above (one-line generator change; regenerate the kept `-00` files).
+
+This single change restores a 3-way outcome (win/draw/loss) across the
+suite and makes `objective_progress` partial credit meaningful instead
+of every non-win collapsing to 0.5.
+
+Secondary FIXes (per pack, already noted in the table):
+
+- **strategy-dilemma / strategy-gauntlet / strategy-twobody**: win is
+  `buildings_discovered_gte: 1` — satisfied by *seeing* the pre-placed
+  enemy base, which a huge friendly stance-3 force does incidentally.
+  Re-spec to `reach_region` (the objective) + `units_lost_lte`
+  (enforce the risk/coordination the title claims).
+- **cat-c3/c4**: add `has_building: tent` (or `proc`) to the win so
+  `building_total_gte` can't be met by powr-spam; this restores the
+  tech-*dependency* the pack claims to test.
+- **economy-investment / -time-box / -force-buildup**: tighten
+  `starting_cash` so the wide and deep paths are mutually exclusive
+  (currently both fit), and add the timeout fail. Until engine S0/S1
+  (ore income) lands these can only ever be "spend the obvious thing."
+- **artofwar-lure-the-tiger**: add an `after_ticks` gate or a
+  staging `reach_region` on the lure so the win actually requires the
+  two-phase plan, not just any survivable route to the objective.
+- **rush-hour**: drop it from the "strategy" claim; reduce the
+  friendly force and require coverage (`explored_pct_gte`) + kills +
+  timeout-fail so it tests coordinated sweep, not auto-battle.
+
+---
+
+## 6. ADD list (top 10 new scenarios to author)
+
+Gaps the active set still has, with win/fail sketches in the existing
+predicate grammar and the external benchmark each parallels.
+
+1. **adversarial-counterstrategy-read** (capability: adversarial).
+   Opponent commits one of two observable openings (rush near-lane vs
+   tech far-corner); agent must scout then commit the dominant
+   counter. Win: `all_of[{buildings_discovered_gte:1},{units_killed_gte:N},{within_ticks:T}]`;
+   fail: `{not:{own_units_gte:1}}` or timeout. Parallel:
+   StarCraft-II-Arena, game-theoretic LLM evals. (Fills the
+   opponent-modeling gap — the unique RTS value prop.)
+
+2. **longhorizon-opening-to-assault** (reasoning). One episode,
+   40k+ ticks: scout (explored_pct) → tech (`has_building: proc`) →
+   army (`own_units_gte`) → strike (`reach_region` enemy base) — all
+   four as `all_of`, terminal-only credit, timeout fail. Parallel:
+   Terminal-Bench 2.0 / OSWorld 50-step. (Fills long-horizon credit
+   assignment — currently thin.)
+
+3. **strict-toolban-fidelity** (action). Achieve `reach_region` but
+   the allowlist excludes `attack*`; win additionally requires
+   `units_lost_lte: 0` AND a `not{after_ticks:T0}`-then-`reach`
+   ordering. Any disallowed-tool call → fail. Parallel: BFCL V4
+   relevance / τ²-bench. (Isolates API fidelity as the objective.)
+
+4. **economy-throughput-real** (reasoning) — *gated on engine S0/S1*.
+   Once ore income exists: `economy_value_gte: V within_ticks:T`,
+   wide (2 proc) vs deep (1 proc + harvesters) genuinely mutually
+   exclusive. Parallel: PlanBench cost-optimal. (Replaces the 5
+   non-discriminating economy/cat-c5/c6 packs with one real test.)
+
+5. **perception-count-the-threat** (perception). Win:
+   `enemies_discovered_gte: K` where K = exact hidden count and
+   over/under-scouting wastes the clock; fail timeout +
+   `units_lost_lte`. Parallel: ERQA state-estimation. (Strengthens
+   the validated transfer target with a counting variant.)
+
+6. **reasoning-replan-after-loss** (reasoning) — a *true* error-
+   recovery pack (what cat-c12 only pretends to be). Pre-place the
+   agent base in the path of a scripted stance-3 enemy push that
+   destroys 1–2 buildings early; win =
+   `all_of[{building_total_gte:N},{has_building:tent},{after_ticks:t_loss},{within_ticks:T}]`.
+   Parallel: PlanBench replanning. (Requires interrupt/scripted-loss
+   support — see §7.)
+
+7. **adversarial-feint-handling** (adversarial). Opponent shows a
+   decoy force on one axis; win keyed to committing against the
+   *real* axis (`building_in_region`/`units_killed_gte` at the true
+   threat), losing if you commit to the decoy (timeout). Parallel:
+   opponent modeling / adversarial robustness.
+
+8. **coordination-staggered-window** (action). Two squads must each
+   be in *different* regions within *different* tick bands
+   (`all_of` of two `reach_region` + paired `after_ticks`/`within_ticks`),
+   forcing true parallel scheduling, not a single column. Parallel:
+   Watch-And-Help / multi-robot dispatch. (Upgrades the thin
+   strategy-twobody/cat-c10 coordination bucket.)
+
+9. **tempo-double-window** (reasoning). Two separate strike windows
+   in one episode (`after_ticks:t1`+`units_killed_gte:a` then a
+   forced lull then `after_ticks:t2`+`units_killed_gte:b`), so acting
+   continuously fails. Parallel: TextStarCraft II tempo. (Tempo is
+   currently a single cat-c11 category.)
+
+10. **navigation-confined-hard-only** (perception). Replace the
+    trivial custom-map easy/medium with one hard map-read:
+    `all_units_in_region` in a far bounded corner, tight clock,
+    `fail_condition` on timeout — a real ARC-/ERQA-style spatial
+    commit with no degenerate easy tier.
+
+---
+
+## 7. Predicate-vocabulary additions needed
+
+The current leaf set is adequate for §6 #1,2,5,7,8,9,10. To fully
+author the others, add:
+
+- **`tool_called`/`tool_not_called: <name>`** (or a scenario-level
+  `forbidden_tools` enforced as an automatic `fail`) — needed for
+  #3 strict-toolban-fidelity; today the only tool-fidelity signal is
+  the cross-cutting `actions_warned/issued` score, not a win/fail
+  predicate. This is the most leaderboard-relevant missing primitive
+  (BFCL/τ²-bench relevance detection).
+- **`building_destroyed_gte` / `own_building_lost_gte`** — needed for
+  #6 (detect the scripted setback) and to give offensive packs a real
+  "took the base" predicate instead of proxying with
+  `buildings_discovered`/`reach_region`.
+- **An ordering composite (e.g. `then: [A, B]` meaning A must have
+  held true at some tick strictly before B)** — `after_ticks` only
+  approximates sequencing by wall-clock; a true happened-before would
+  let artofwar-lure-the-tiger / sequenced-citadel grade *order*
+  rather than *timing*, and make #2/#6 non-gameable.
+- A scripted-adversary / mid-episode-event hook (engine side, already
+  flagged in SCENARIO_AUDIT as Phase-1) is the prerequisite for #1,
+  #6, #7 to be more than scripted-stance encounters.
+
+---
+
+## 8. Prioritized action summary
+
+**Cut now (12 files, zero capability loss):** all `cat-c*-01` actives
+(c1–c12 `-01`). Optionally also retire `cat-c12-00` (mislabelled) and
+the `strict-production-bom` hard level (unsolvable) unless budgeted.
+
+**Fix now (one mechanical pass):** add a real `fail_condition`
+(timeout-loss; attrition-loss where a `units_lost_lte` win clause
+exists) to all ~36 active packs lacking one, and replace the dead
+`units_lost_lte:-1` in the kept `cat-*` files. This is the single
+change that most improves benchmark discrimination.
+
+**Fix next (per-pack, ~6 packs):** strategy-dilemma/gauntlet/twobody
+(win predicate doesn't enforce the claimed decision), cat-c3/c4
+(powr-spam gameable — add `has_building`), economy-* (tighten cash so
+allocation is a real trade), artofwar-lure-the-tiger (add ordering
+gate), rush-hour (re-spec or stop counting it as strategy),
+strict-production-bom hard (raise budget or cut).
+
+**Top 10 to add (ranked):** #1 adversarial-counterstrategy-read,
+#2 longhorizon-opening-to-assault, #3 strict-toolban-fidelity,
+#6 reasoning-replan-after-loss, #7 adversarial-feint-handling,
+#8 coordination-staggered-window, #5 perception-count-the-threat,
+#9 tempo-double-window, #4 economy-throughput-real (after S0/S1),
+#10 navigation-confined-hard-only.
+
+**Predicate additions (in priority order):** (1) tool-call
+fidelity predicate / `forbidden_tools` auto-fail; (2)
+`building_destroyed_gte`; (3) a happened-before ordering composite;
+(4) scripted mid-episode adversary hook (engine, Phase-1).

openra_bench/agent.py

@@ -476,6 +476,28 @@ class ModelAgent:
                 }
         return {"role": "user", "content": text}
 
+    @staticmethod
+    def _window(history: list[dict], max_turns: int) -> list[dict]:
+        """Wire-history sliding window: keep all leading system
+        messages + the last `max_turns` user-led groups. Slicing on a
+        user boundary keeps every assistant↔tool pairing intact (only
+        whole older groups are dropped, so no dangling tool replies).
+        `self.history` itself is untouched — playback keeps the full
+        transcript; only what's POSTED is bounded."""
+        if max_turns <= 0:
+            return history
+        lead = 0
+        while lead < len(history) and history[lead].get("role") == "system":
+            lead += 1
+        user_idx = [
+            i for i in range(lead, len(history))
+            if history[i].get("role") == "user"
+        ]
+        if len(user_idx) <= max_turns:
+            return history
+        cut = user_idx[-max_turns]
+        return history[:lead] + history[cut:]
+
     @staticmethod
     def _strip_old_images(history: list[dict]) -> None:
         """Keep only the latest image to bound ViT token cost (mirrors
@@ -495,7 +517,10 @@ class ModelAgent:
         self.stats["turns"] += 1
         self.history.append(self._user_message(render_state))
         self._strip_old_images(self.history)
-        reply = self.provider.complete(self.history, self.tools)
+        wire = self._window(
+            self.history, getattr(self.cfg, "max_history_turns", 16)
+        )
+        reply = self.provider.complete(wire, self.tools)
         self.history.append(
             {
                 "role": "assistant",

openra_bench/providers.py

@@ -52,6 +52,14 @@ class ProviderConfig:
     timeout_s: float = 120.0
     vision: bool = True
     extra_headers: dict[str, str] = field(default_factory=dict)
+    # Resilience (real OpenRouter runs): bounded retry, throttle, price.
+    max_retries: int = 5
+    retry_base_s: float = 1.0
+    retry_cap_s: float = 30.0
+    qps: float = 0.0  # 0 = unthrottled; shared limiter set by evaluate
+    max_history_turns: int = 16  # sliding wire-history window (0=unbounded)
+    price_in_per_m: float = 0.0   # USD / 1M prompt tokens
+    price_out_per_m: float = 0.0  # USD / 1M completion tokens
 
     def resolved_base_url(self) -> str:
         if self.base_url:
@@ -82,6 +90,7 @@ class ChatReply:
     text: str
     tool_calls: list[dict]  # [{"name": str, "arguments": dict}]
     reasoning: str = ""  # chain-of-thought, when the model/provider emits it
+    usage: dict = field(default_factory=dict)  # prompt/completion tokens
     raw: dict = field(default_factory=dict)
 
 
@@ -93,11 +102,58 @@ class ChatProvider:
 class OpenAICompatibleProvider(ChatProvider):
     """OpenAI /chat/completions with `tools`. vLLM + OpenRouter + OpenAI."""
 
-    def __init__(self, cfg: ProviderConfig):
+    def __init__(self, cfg: ProviderConfig, *, rate_limiter=None,
+                 cost_meter=None):
         self.cfg = cfg
         self._client = httpx.Client(timeout=cfg.timeout_s)
+        from .resilience import CostMeter, RateLimiter, RetryPolicy
+
+        self._rl = rate_limiter or RateLimiter(cfg.qps)
+        self._cost = cost_meter or CostMeter(
+            cfg.price_in_per_m, cfg.price_out_per_m
+        )
+        self._policy = RetryPolicy(
+            max_attempts=max(1, cfg.max_retries),
+            base=cfg.retry_base_s,
+            cap=cfg.retry_cap_s,
+        )
+
+    @property
+    def cost_meter(self):
+        return self._cost
+
+    def _post_once(self, url, headers, body):
+        from .resilience import FatalProviderError
+
+        try:
+            resp = self._client.post(url, headers=headers, json=body)
+        except httpx.TimeoutException as e:
+            e.transient = True  # type: ignore[attr-defined]
+            e.retry_after = None  # type: ignore[attr-defined]
+            raise
+        except httpx.TransportError as e:
+            e.transient = True  # type: ignore[attr-defined]
+            e.retry_after = None  # type: ignore[attr-defined]
+            raise
+        if resp.status_code >= 400:
+            ra = resp.headers.get("retry-after")
+            try:
+                retry_after = float(ra) if ra is not None else None
+            except ValueError:
+                retry_after = None
+            transient = self._policy.is_transient_status(resp.status_code)
+            cls = RuntimeError if transient else FatalProviderError
+            exc = cls(
+                f"{resp.status_code} from provider: {resp.text[:200]}"
+            )
+            exc.transient = transient  # type: ignore[attr-defined]
+            exc.retry_after = retry_after  # type: ignore[attr-defined]
+            raise exc
+        return resp
 
     def complete(self, messages: list[dict], tools: list[dict]) -> ChatReply:
+        from .resilience import retry_call
+
         cfg = self.cfg
         headers = {
             "Authorization": f"Bearer {cfg.resolved_api_key()}",
@@ -113,13 +169,17 @@ class OpenAICompatibleProvider(ChatProvider):
         if tools:
             body["tools"] = tools
             body["tool_choice"] = "auto"
-        resp = self._client.post(
-            f"{cfg.resolved_base_url()}/chat/completions",
-            headers=headers,
-            json=body,
+        url = f"{cfg.resolved_base_url()}/chat/completions"
+
+        self._rl.acquire()
+        resp = retry_call(
+            lambda: self._post_once(url, headers, body), self._policy
         )
-        resp.raise_for_status()
-        return self._reply_from_data(resp.json())
+        reply = self._reply_from_data(resp.json())
+        u = reply.usage or {}
+        self._cost.add(u.get("prompt_tokens", 0), u.get("completion_tokens", 0))
+        self._cost.check()  # raises BudgetExceeded → evaluate finalizes
+        return reply
 
     # Keys the OpenAI Chat Completions wire format accepts per message.
     # `history` carries extra playback-only keys (notably "reasoning");
@@ -157,10 +217,15 @@ class OpenAICompatibleProvider(ChatProvider):
             rc = "".join(
                 p.get("text", "") if isinstance(p, dict) else str(p) for p in rc
             )
+        usage = data.get("usage") or {}
         return ChatReply(
             text=msg.get("content") or "",
             tool_calls=calls,
             reasoning=str(rc),
+            usage={
+                "prompt_tokens": usage.get("prompt_tokens", 0),
+                "completion_tokens": usage.get("completion_tokens", 0),
+            },
             raw=data,
         )
 
@@ -182,9 +247,12 @@ class BedrockProvider(ChatProvider):
         )
 
 
-def make_provider(cfg: ProviderConfig) -> ChatProvider:
+def make_provider(cfg: ProviderConfig, *, rate_limiter=None,
+                  cost_meter=None) -> ChatProvider:
     if cfg.provider == "bedrock":
         return BedrockProvider(cfg)
     if cfg.provider in ("openai", "vllm", "openrouter"):
-        return OpenAICompatibleProvider(cfg)
+        return OpenAICompatibleProvider(
+            cfg, rate_limiter=rate_limiter, cost_meter=cost_meter
+        )
     raise ValueError(f"unknown provider {cfg.provider!r}")

openra_bench/resilience.py

@@ -0,0 +1,205 @@
+"""Resilience primitives for real (OpenRouter) end-to-end runs.
+
+A long sweep is tens of thousands of API calls over hours; transient
+429/5xx/timeouts, credit exhaustion, and process death are *expected*,
+not exceptional. These primitives are pure and thread-safe so the
+evaluator can retry, throttle, cap spend, and resume losslessly.
+
+Nothing here imports the engine or a provider — fully unit-testable.
+"""
+
+from __future__ import annotations
+
+import json
+import threading
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+
+
+class BudgetExceeded(RuntimeError):
+    """Raised when the cost meter passes the hard cap. The evaluator
+    catches it, finalizes from the journal, and marks the run truncated
+    (a partial result is always better than a lost 6-hour run)."""
+
+
+class FatalProviderError(RuntimeError):
+    """A non-retryable provider failure (4xx other than 429)."""
+
+
+# ── retry / backoff ────────────────────────────────────────────────────────
+
+_TRANSIENT_STATUS = frozenset({408, 409, 425, 429, 500, 502, 503, 504})
+
+
+@dataclass
+class RetryPolicy:
+    max_attempts: int = 5
+    base: float = 1.0       # seconds; exponential: base * 2**(attempt-1)
+    cap: float = 30.0       # max single sleep
+    jitter: float = 0.1     # fraction of delay added deterministically*0
+
+    def is_transient_status(self, status: int) -> bool:
+        return status in _TRANSIENT_STATUS
+
+    def delay(self, attempt: int, retry_after: float | None = None) -> float:
+        """Sleep before retry `attempt` (1-based). Honors a server
+        Retry-After when present and larger than our backoff."""
+        backoff = min(self.cap, self.base * (2 ** max(0, attempt - 1)))
+        if retry_after is not None and retry_after > 0:
+            return min(self.cap, max(backoff, retry_after))
+        return backoff
+
+
+def retry_call(fn, policy: RetryPolicy, *, on_retry=None, sleep=time.sleep):
+    """Call `fn()` with bounded exponential backoff.
+
+    `fn` raises to signal failure; it may attach `.transient` (bool)
+    and `.retry_after` (float|None) to the exception to steer policy.
+    Non-transient → re-raised immediately. Exhausted attempts →
+    last exception re-raised.
+    """
+    last: Exception | None = None
+    for attempt in range(1, policy.max_attempts + 1):
+        try:
+            return fn()
+        except Exception as exc:  # noqa: BLE001 — policy decides
+            last = exc
+            transient = getattr(exc, "transient", True)
+            if not transient or attempt >= policy.max_attempts:
+                raise
+            d = policy.delay(attempt, getattr(exc, "retry_after", None))
+            if on_retry is not None:
+                on_retry(attempt, exc, d)
+            sleep(d)
+    assert last is not None
+    raise last
+
+
+# ── rate limiting ──────────────────────────────────────────────────────────
+
+
+class RateLimiter:
+    """Thread-safe minimum-interval limiter (≈ qps cap) shared across
+    the concurrency pool. qps<=0 disables it."""
+
+    def __init__(self, qps: float = 0.0):
+        self._interval = 1.0 / qps if qps and qps > 0 else 0.0
+        self._lock = threading.Lock()
+        self._next = 0.0
+
+    def acquire(self, *, now=time.monotonic, sleep=time.sleep) -> float:
+        if self._interval <= 0:
+            return 0.0
+        with self._lock:
+            t = now()
+            wait = max(0.0, self._next - t)
+            self._next = max(t, self._next) + self._interval
+        if wait > 0:
+            sleep(wait)
+        return wait
+
+
+# ── cost / token metering ──────────────────────────────────────────────────
+
+
+@dataclass
+class CostMeter:
+    """Thread-safe token + USD accumulator with a hard cap.
+
+    Pricing is per 1M tokens (OpenRouter-style). `max_usd<=0` disables
+    the cap; metering still runs so the report carries spend."""
+
+    price_in_per_m: float = 0.0
+    price_out_per_m: float = 0.0
+    max_usd: float = 0.0
+    prompt_tokens: int = 0
+    completion_tokens: int = 0
+    calls: int = 0
+    _lock: threading.Lock = field(default_factory=threading.Lock, repr=False)
+
+    @property
+    def usd(self) -> float:
+        return round(
+            self.prompt_tokens / 1e6 * self.price_in_per_m
+            + self.completion_tokens / 1e6 * self.price_out_per_m,
+            6,
+        )
+
+    def add(self, prompt: int, completion: int) -> None:
+        with self._lock:
+            self.prompt_tokens += int(prompt or 0)
+            self.completion_tokens += int(completion or 0)
+            self.calls += 1
+
+    def check(self) -> None:
+        if self.max_usd and self.max_usd > 0 and self.usd >= self.max_usd:
+            raise BudgetExceeded(
+                f"spend ${self.usd:.4f} ≥ cap ${self.max_usd:.2f} "
+                f"({self.calls} calls, {self.prompt_tokens}+"
+                f"{self.completion_tokens} tok)"
+            )
+
+    def snapshot(self) -> dict:
+        return {
+            "calls": self.calls,
+            "prompt_tokens": self.prompt_tokens,
+            "completion_tokens": self.completion_tokens,
+            "usd": self.usd,
+            "max_usd": self.max_usd,
+        }
+
+
+# ── checkpoint / resume journal ────────────────────────────────────────────
+
+
+def episode_key(pack: str, level: str, split: str, seed: int) -> str:
+    return f"{pack}|{level}|{split}|{seed}"
+
+
+class RunJournal:
+    """Append-only JSONL of completed episodes. Resume = skip keys
+    already present; the aggregate is rebuilt from the journal so a
+    killed run continues losslessly."""
+
+    def __init__(self, path: str | Path):
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._lock = threading.Lock()
+
+    def done_keys(self) -> set[str]:
+        if not self.path.exists():
+            return set()
+        keys: set[str] = set()
+        for line in self.path.read_text().splitlines():
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                keys.add(json.loads(line)["_key"])
+            except Exception:  # noqa: BLE001 — tolerate a torn last line
+                continue
+        return keys
+
+    def append(self, key: str, record: dict) -> None:
+        row = dict(record)
+        row["_key"] = key
+        line = json.dumps(row)
+        with self._lock:
+            with open(self.path, "a") as f:
+                f.write(line + "\n")
+                f.flush()
+
+    def records(self) -> list[dict]:
+        if not self.path.exists():
+            return []
+        out: list[dict] = []
+        for line in self.path.read_text().splitlines():
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                out.append(json.loads(line))
+            except Exception:  # noqa: BLE001
+                continue
+        return out

openra_bench/run_eval.py

@@ -21,6 +21,7 @@ import statistics
 import sys
 import time
 from collections import Counter
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Callable
 
@@ -80,6 +81,13 @@ def evaluate(
     concurrency: int = 1,
     run_id: str | None = None,
     model: str | None = None,
+    journal_path: str | Path | None = None,
+    resume: bool = False,
+    max_spend_usd: float = 0.0,
+    smoke: bool = False,
+    dry_run: bool = False,
+    report_path: str | Path | None = None,
+    progress=None,
 ) -> dict:
     """Run packs×levels×seeds. If `held_out_seeds` is given, those are
     run too and tagged split='held_out'; the report adds
@@ -87,7 +95,42 @@ def evaluate(
     held-out composite) — the anti-memorization metric the
     generalization literature (Procgen/SMACv2/lmgame-Bench) requires.
     """
-    factory = agent_factory or _default_agent_factory(provider_cfg)
+    from .resilience import (
+        BudgetExceeded,
+        CostMeter,
+        RateLimiter,
+        RunJournal,
+        episode_key,
+    )
+
+    # One shared cost meter + rate limiter across the whole sweep, so
+    # the budget cap and throttle apply globally (not per episode).
+    meter = CostMeter(
+        getattr(provider_cfg, "price_in_per_m", 0.0),
+        getattr(provider_cfg, "price_out_per_m", 0.0),
+        max_usd=max_spend_usd,
+    )
+    limiter = RateLimiter(getattr(provider_cfg, "qps", 0.0) or 0.0)
+    if agent_factory is not None:
+        factory = agent_factory
+    elif provider_cfg is None:
+        factory = lambda _c: scripted_explore_agent  # noqa: E731
+    else:
+        from .agent import ModelAgent
+        from .providers import make_provider
+
+        shared = make_provider(
+            provider_cfg, rate_limiter=limiter, cost_meter=meter
+        )
+
+        def factory(compiled: CompiledLevel):
+            return ModelAgent(
+                provider_cfg,
+                allowed_tools=compiled.scenario.tools,
+                objective=compiled.scenario.description,
+                provider=shared,
+            ).agent_fn
+
     # Run/model identity so a single playback root can hold many runs
     # and the viewer can filter run → model → scenario.
     run_id = run_id or time.strftime("%Y%m%d-%H%M%S", time.gmtime())
@@ -169,48 +212,166 @@ def evaluate(
             "_sc": sc,
         }
 
-    if concurrency > 1 and len(tasks) > 1:
-        from concurrent.futures import ThreadPoolExecutor
+    # Pre-flight: dry-run validates compile/selection without engine or
+    # API spend; smoke runs exactly one episode.
+    if dry_run:
+        return {
+            "dry_run": True,
+            "run_id": run_id,
+            "model": model,
+            "tasks": len(tasks),
+            "skipped": skipped,
+            "cells": sorted({t[1] for t in tasks}),
+        }
+    if smoke:
+        tasks = tasks[:1]
+
+    # Checkpoint/resume: a journal of completed episodes. On resume we
+    # skip done (pack|level|split|seed) and fold prior records back in,
+    # so a killed multi-hour run continues losslessly.
+    jp = journal_path
+    if jp is None and playback_root is not None:
+        jp = Path(playback_root) / f"{run_id}__{_safe_model}" / "_journal.jsonl"
+    journal = RunJournal(jp) if jp is not None else None
+    prior: list[dict] = []
+    if journal is not None and resume:
+        done = journal.done_keys()
+        prior = journal.records()
+        tasks = [
+            t for t in tasks
+            if episode_key(t[0].meta.id, t[0].level, t[2], t[3]) not in done
+        ]
 
-        with ThreadPoolExecutor(max_workers=concurrency) as ex:
-            results = list(ex.map(_run_one, tasks))
-    else:
-        results = [_run_one(t) for t in tasks]
+    def _persist(rec: dict) -> None:
+        if journal is None:
+            return
+        slim = {k: v for k, v in rec.items() if k != "_sc"}
+        journal.append(
+            episode_key(
+                rec["cell"].rsplit(":", 1)[0],
+                rec["cell"].rsplit(":", 1)[1],
+                rec["split"],
+                rec["seed"],
+            ),
+            slim,
+        )
 
-    # Deterministic aggregation: sort so the report is identical
-    # regardless of worker scheduling.
-    results.sort(key=lambda r: (r["cell"], r["split"], r["seed"]))
+    new_results: list[dict] = []
+    truncated = False
+    done_n = 0
+
+    def _record(rec: dict) -> None:
+        nonlocal done_n
+        _persist(rec)
+        new_results.append(rec)
+        done_n += 1
+        if progress is not None:
+            progress(done_n, len(tasks), rec, meter.snapshot())
+        if report_path is not None:
+            # Incremental flush so a long run is always inspectable.
+            try:
+                write_report(
+                    _finalize(prior, new_results, skipped, run_id, model,
+                              meter, truncated=False),
+                    report_path,
+                )
+            except Exception:  # noqa: BLE001 — flush must never abort a run
+                pass
+
+    try:
+        if concurrency > 1 and len(tasks) > 1:
+            from concurrent.futures import ThreadPoolExecutor
+
+            with ThreadPoolExecutor(max_workers=concurrency) as ex:
+                futs = {ex.submit(_run_one, t): t for t in tasks}
+                from concurrent.futures import as_completed
+
+                for fu in as_completed(futs):
+                    _record(fu.result())
+        else:
+            for t in tasks:
+                _record(_run_one(t))
+    except BudgetExceeded as e:
+        truncated = True
+        skipped.append(f"BUDGET STOP: {e}")
+
+    out = _finalize(prior, new_results, skipped, run_id, model, meter,
+                    truncated=truncated)
+    if report_path is not None:
+        write_report(out, report_path)
+    return out
+
+
+@dataclass
+class _ScoreShim:
+    """Reconstruct the fields `_agg` needs from a journaled episode
+    dict, so resume aggregates prior + new identically to a fresh run."""
+
+    composite: float
+    outcome: str
+    perception: float
+    reasoning: float
+    action: float
+    weakest_link: str
+    dimensions: dict
+
+
+def _shim(r: dict):
+    sc = r.get("_sc")
+    if sc is not None:
+        return sc
+    return _ScoreShim(
+        composite=r.get("composite", 0.0),
+        outcome=r.get("outcome", "draw"),
+        perception=r.get("perception", 0.0),
+        reasoning=r.get("reasoning", 0.0),
+        action=r.get("action", 0.0),
+        weakest_link=r.get("weakest_link", "n/a"),
+        dimensions={"objective": r.get("objective_progress", 0.0)},
+    )
+
+
+def _finalize(prior: list[dict], new: list[dict], skipped: list[str],
+              run_id, model, meter, *, truncated: bool) -> dict:
+    rows = list(prior) + list(new)
+    rows.sort(key=lambda r: (r.get("cell", ""), r.get("split", ""),
+                             r.get("seed", 0)))
     by_cell: dict[str, list] = {}
     public_scores: list = []
     held_scores: list = []
     episodes: list[dict] = []
-    for r in results:
-        sc = r.pop("_sc")
-        if r["split"] == "public":
+    for r in rows:
+        sc = _shim(r)
+        slim = {k: v for k, v in r.items() if k != "_sc"}
+        if r.get("split") == "public":
             by_cell.setdefault(r["cell"], []).append(sc)
             public_scores.append(sc)
         else:
             held_scores.append(sc)
-        episodes.append(r)
+        episodes.append(slim)
 
-    # Mean cumulative reward vector across public episodes — the
-    # scenario-agnostic progress signature, comparable across runs.
-    pub = [r for r in episodes if r["split"] == "public" and r.get("reward_vector")]
+    pub = [r for r in episodes
+           if r.get("split") == "public" and r.get("reward_vector")]
     rv_mean: dict = {}
     if pub:
         for k in pub[0]["reward_vector"]:
             rv_mean[k] = round(
-                statistics.fmean(r["reward_vector"].get(k, 0.0) for r in pub), 4
+                statistics.fmean(r["reward_vector"].get(k, 0.0) for r in pub),
+                4,
             )
 
     out = {
-        "summary": {cell: _agg(scs) for cell, scs in by_cell.items()},
+        "run_id": run_id,
+        "model": model,
+        "truncated": truncated,
+        "resumed": len(prior),
+        "cost": meter.snapshot() if meter is not None else {},
+        "summary": {c: _agg(s) for c, s in by_cell.items()},
         "overall": _agg(public_scores),
         "reward_vector_mean": rv_mean,
         "episodes": episodes,
         "skipped": skipped,
     }
-    # Adversarial spotlight: per-pack ladder ratings + headline mean.
     from .adversarial import adversarial_summary
 
     adv = adversarial_summary(out)
@@ -278,6 +439,19 @@ def main(argv: list[str]) -> int:
         help="publish this run to the leaderboard store (optional path; "
         "default data/leaderboard.jsonl)",
     )
+    # Resilience flags for real OpenRouter runs.
+    ap.add_argument("--resume", action="store_true",
+                    help="skip episodes already in the run journal")
+    ap.add_argument("--journal", default=None,
+                    help="checkpoint journal path (default: under --playback)")
+    ap.add_argument("--max-spend", type=float, default=0.0,
+                    help="hard USD cap; the run finalizes when hit")
+    ap.add_argument("--qps", type=float, default=0.0,
+                    help="global request/sec throttle (0 = unthrottled)")
+    ap.add_argument("--smoke", action="store_true",
+                    help="run exactly one episode (live preflight)")
+    ap.add_argument("--dry-run", action="store_true",
+                    help="validate/compile + list tasks, no engine/API")
     a = ap.parse_args(argv[1:])
 
     cfg = None
@@ -289,6 +463,7 @@ def main(argv: list[str]) -> int:
             model=a.model,
             base_url=a.base_url,
             vision=not a.no_vision,
+            qps=a.qps,
         )
 
     stats = evaluate(
@@ -299,7 +474,23 @@ def main(argv: list[str]) -> int:
         held_out_seeds=[int(s) for s in a.held_out_seeds.split(",") if s.strip()],
         playback_root=a.playback,
         concurrency=a.concurrency,
+        model=a.model if a.provider else None,
+        journal_path=a.journal,
+        resume=a.resume,
+        max_spend_usd=a.max_spend,
+        smoke=a.smoke,
+        dry_run=a.dry_run,
+        report_path=a.out,
+        progress=lambda d, n, rec, c: print(
+            f"[{d}/{n}] {rec['cell']}:{rec['split']}#{rec['seed']} "
+            f"{rec['outcome']} comp={rec['composite']} "
+            f"${c['usd']:.4f}", flush=True
+        ),
     )
+    if stats.get("dry_run"):
+        print(f"dry-run: {stats['tasks']} tasks over "
+              f"{len(stats['cells'])} cells; skipped {len(stats['skipped'])}")
+        return 0
     write_report(stats, a.out)
     o = stats["overall"]
     print(f"\nwrote {a.out}")

tests/test_resilience.py

@@ -0,0 +1,168 @@
+"""Resilience layer: retry/backoff, throttle, cost cap, journal/resume,
+bounded history, dry-run/smoke. All deterministic — fake clocks/sleeps,
+no network, scripted agent for the evaluate-integration paths.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from openra_bench.resilience import (
+    BudgetExceeded,
+    CostMeter,
+    FatalProviderError,
+    RateLimiter,
+    RetryPolicy,
+    RunJournal,
+    episode_key,
+    retry_call,
+)
+
+
+# ── retry / backoff ────────────────────────────────────────────────────────
+
+
+def test_retry_policy_delay_exponential_capped_and_retry_after():
+    p = RetryPolicy(base=1.0, cap=30.0, max_attempts=6)
+    assert [p.delay(a) for a in (1, 2, 3, 4, 10)] == [1.0, 2.0, 4.0, 8.0, 30.0]
+    # server Retry-After wins when larger; still capped
+    assert p.delay(1, retry_after=5.0) == 5.0
+    assert p.delay(1, retry_after=999) == 30.0
+    assert p.is_transient_status(429) and not p.is_transient_status(400)
+
+
+def test_retry_call_succeeds_after_transient_then_stops_on_fatal():
+    slept = []
+    n = {"i": 0}
+
+    def flaky():
+        n["i"] += 1
+        if n["i"] < 3:
+            e = RuntimeError("503"); e.transient = True
+            raise e
+        return "ok"
+
+    assert retry_call(flaky, RetryPolicy(max_attempts=5),
+                       sleep=slept.append) == "ok"
+    assert len(slept) == 2  # two backoffs before the 3rd attempt
+
+    def fatal():
+        e = FatalProviderError("400"); e.transient = False
+        raise e
+
+    with pytest.raises(FatalProviderError):
+        retry_call(fatal, RetryPolicy(max_attempts=5), sleep=slept.append)
+
+    def always():
+        e = RuntimeError("503"); e.transient = True
+        raise e
+
+    with pytest.raises(RuntimeError):
+        retry_call(always, RetryPolicy(max_attempts=3), sleep=lambda *_: None)
+
+
+# ── rate limiter ───────────────────────────────────────────────────────────
+
+
+def test_rate_limiter_enforces_min_interval():
+    clk = {"t": 0.0}
+    slept = []
+    rl = RateLimiter(qps=2.0)  # 0.5s spacing
+
+    def now():
+        return clk["t"]
+
+    def slp(s):
+        slept.append(s)
+        clk["t"] += s
+
+    assert rl.acquire(now=now, sleep=slp) == 0.0       # first is free
+    w = rl.acquire(now=now, sleep=slp)                  # immediate 2nd waits
+    assert w == pytest.approx(0.5)
+    assert RateLimiter(0.0).acquire() == 0.0            # disabled
+
+
+# ── cost meter ─────────────────────────────────────────────────────────────
+
+
+def test_cost_meter_accumulates_and_caps():
+    m = CostMeter(price_in_per_m=1.0, price_out_per_m=2.0, max_usd=0.01)
+    m.add(1000, 1000)  # 0.001 + 0.002 = 0.003
+    m.check()  # under cap
+    assert m.usd == pytest.approx(0.003)
+    m.add(2_000_000, 2_000_000)  # blows the cap
+    with pytest.raises(BudgetExceeded):
+        m.check()
+    assert m.snapshot()["calls"] == 2
+
+
+# ── journal / resume ───────────────────────────────────────────────────────
+
+
+def test_journal_roundtrip_and_torn_line(tmp_path):
+    j = RunJournal(tmp_path / "j.jsonl")
+    assert j.done_keys() == set()
+    k = episode_key("p", "easy", "public", 1)
+    j.append(k, {"cell": "p:easy", "composite": 0.4})
+    j.append(episode_key("p", "hard", "public", 2), {"cell": "p:hard"})
+    assert k in j.done_keys() and len(j.done_keys()) == 2
+    with open(tmp_path / "j.jsonl", "a") as f:
+        f.write('{"_key": "torn"')  # no newline / invalid
+    assert len(j.records()) == 2  # torn tail tolerated
+
+
+# ── bounded chat history ───────────────────────────────────────────────────
+
+
+def test_model_agent_window_keeps_system_and_last_turns():
+    from openra_bench.agent import ModelAgent
+
+    h = [{"role": "system", "content": "S"}]
+    for t in range(5):
+        h.append({"role": "user", "content": f"u{t}"})
+        h.append({"role": "assistant", "content": f"a{t}",
+                  "tool_calls": [{"id": f"c{t}"}]})
+        h.append({"role": "tool", "tool_call_id": f"c{t}", "content": "ok"})
+
+    w = ModelAgent._window(h, 2)
+    assert w[0]["role"] == "system"
+    users = [m for m in w if m["role"] == "user"]
+    assert [m["content"] for m in users] == ["u3", "u4"]  # last 2 groups
+    # pairing intact: first non-system is a user (no dangling tool reply)
+    assert w[1]["role"] == "user"
+    # no trimming when within budget; passthrough when disabled
+    assert ModelAgent._window(h, 99) is h
+    assert ModelAgent._window(h, 0) is h
+
+
+# ── evaluate: dry-run / smoke / journal+resume (scripted, no API) ───────────
+
+
+def test_evaluate_dry_run_lists_without_running():
+    from openra_bench.run_eval import evaluate
+
+    PACK = Path("openra_bench/scenarios/packs/perception-frontier-reading.yaml")
+    out = evaluate([PACK], ["easy", "medium"], [1], dry_run=True)
+    assert out["dry_run"] and out["tasks"] == 2
+    assert "episodes" not in out  # nothing executed
+
+
+def test_evaluate_journal_resume_is_lossless(tmp_path):
+    pytest.importorskip("openra_train")
+    from openra_bench.run_eval import evaluate
+
+    PACK = Path("openra_bench/scenarios/packs/perception-frontier-reading.yaml")
+    jp = tmp_path / "j.jsonl"
+    a = evaluate([PACK], ["easy"], [1, 2], journal_path=jp)
+    assert a["overall"]["n"] == 2 and a["resumed"] == 0
+    n_lines = len(jp.read_text().splitlines())
+    assert n_lines == 2
+
+    # resume: both episodes already journaled → 0 new, same aggregate
+    b = evaluate([PACK], ["easy"], [1, 2], journal_path=jp, resume=True)
+    assert b["resumed"] == 2 and b["overall"]["n"] == 2
+    assert len(jp.read_text().splitlines()) == 2  # nothing re-appended
+    assert "cost" in b and "truncated" in b

tests/test_run_eval.py

@@ -92,8 +92,10 @@ def test_concurrency_is_deterministic_and_isolated():
         PACKS / "perception-frontier-reading.yaml",
         PACKS / "reasoning-frontier-commit.yaml",
     ]
-    seq = evaluate(packs, ["easy"], [1, 2, 3], concurrency=1)
-    par = evaluate(packs, ["easy"], [1, 2, 3], concurrency=4)
+    # Pin run_id: it's a wall-clock label by design, not part of the
+    # determinism contract (scores/episodes/aggregates are).
+    seq = evaluate(packs, ["easy"], [1, 2, 3], concurrency=1, run_id="t")
+    par = evaluate(packs, ["easy"], [1, 2, 3], concurrency=4, run_id="t")
     # Same report regardless of worker scheduling (episodes sorted,
     # aggregates order-independent) — episodes ran in isolation.
     assert json.dumps(seq, sort_keys=True) == json.dumps(par, sort_keys=True)