From bbb91e2f339156d5fa4dd3be9744bfab8f7aebef Mon Sep 17 00:00:00 2001 From: 0xrsydn Date: Mon, 21 Sep 2026 17:09:43 +0700 Subject: [PATCH] Add decision brain with tests Per-state decision logic: Jev-first with deterministic fallbacks for combat, card rewards, relic select, map, card select, events, rest sites, shops, treasures, bundles, and hand select. 113 tests. --- brain.py | 1718 +++++++++++++++++++++++++++++++++++++++++++++++++ test_brain.py | 960 +++++++++++++++++++++++++++ 2 files changed, 2678 insertions(+) create mode 100644 brain.py create mode 100644 test_brain.py diff --git a/brain.py b/brain.py new file mode 100644 index 0000000..fc743ec --- /dev/null +++ b/brain.py @@ -0,0 +1,1718 @@ +#!/usr/bin/env python3 +""" +brain.py -- the decision layer. + +Precedence, and why it is ordered this way: + + 1. CODE decides lethal. facts.py proves lethal is reachable; a deterministic + greedy then executes it. Jev is never asked "can I kill this", because it + answered that question wrongly at 0.79 confidence. + 2. CODE decides the fallback. When Jev is unsure, or unavailable, a documented + heuristic (adapted from the STS2MCP AGENTS.md strategy notes) acts instead. + 3. JEV decides preference. Only when lethal is not available and the fallback + is not forced do we ask Jev which play is best. That is a judgement about + semantics, which is what Jev is actually good at. + +Every decision returns exactly ONE action. Card indices shift on every play, +so the loop must re-observe after each action. +""" + +from __future__ import annotations + +import itertools +import json +import re +from dataclasses import dataclass, field +from typing import Any + +import facts as F +from jev import JevClient, JevError, ChoiceAnswer, NoulAnswer, choice, gate, noul + +# Events are high-stakes and often irreversible, so the bar is higher than for +# combat. Measured: Jev picked a run-ending option at confidence 0.49. +EVENT_TOP_MIN = 0.60 +EVENT_MARGIN_MIN = 0.30 + +# Deterministic safety net for uncertain event choices. +# +# A "does this risk losing the run?" Noul was tried and REMOVED: measured, it +# ranked "Lose Everything" (0.46) as LESS risky than "Keep Deciphering" (0.52) +# and lower than "Stop" would suggest. It is actively misleading, so danger is +# detected by keywords instead. Lower score is safer. +EVENT_RISK_WORDS = ( + "everything", "keep", "continue", "deeper", "again", "more", "all of", + "gamble", "risk", "sacrifice", "lose", "push", "betray", "accept", +) +EVENT_STOP_WORDS = ( + "stop", "leave", "proceed", "back", "refuse", "decline", "end", + "take what", "walk away", "done", "nothing", +) + + +def event_safety_rank(option: dict) -> int: + """Lower is safer. Deterministic; never consults the model.""" + text = f"{option.get('title', '')} {option.get('description', '')}".lower() + score = sum(1 for w in EVENT_RISK_WORDS if w in text) + score -= sum(1 for w in EVENT_STOP_WORDS if w in text) + return score + +# Confidence below this escalates instead of acting. +CONFIDENCE_FLOOR = 0.55 + + +@dataclass +class Decision: + action: str + params: dict = field(default_factory=dict) + reason: str = "" + source: str = "code" # code | jev | fallback + confidence: float | None = None + + def __str__(self) -> str: + params = ", ".join(f"{k}={v!r}" for k, v in self.params.items()) + conf = f" conf={self.confidence:.2f}" if self.confidence is not None else "" + return f"[{self.source}{conf}] {self.action}({params}) # {self.reason}" + + +# -------------------------------------------------------------------------- +# Combat +# -------------------------------------------------------------------------- + +def _affordable(hand: list[dict], energy: int) -> list[dict]: + return [ + c for c in hand + if c.get("can_play") and F._as_int(c.get("cost")) <= energy + ] + +def _lethal_line(playable: list[dict], energy: int, player_status: list, + enemy: F.EnemyFact) -> list[dict] | None: + """ + Smallest energy-cost subset of playable cards that kills `enemy`. + + Returns the cards to play, highest damage first. None when lethal is + unreachable. Deterministic, so Jev is never involved. + """ + best: list[dict] | None = None + n = min(len(playable), 12) + for r in range(1, n + 1): + for combo in itertools.combinations(range(n), r): + cards = [playable[i] for i in combo] + cost = sum(F._as_int(c.get("cost")) for c in cards) + if cost > energy: + continue + total = 0 + for c in cards: + dmg = F.parse_card_damage(c) + if dmg.raw_total == 0: + continue + per_hit = dmg.base + F.power_amount(player_status, "Strength") + if F.power_amount(player_status, "Weak"): + per_hit = int(per_hit * 0.75) + card_total = per_hit * dmg.hits + if F.power_amount(F.enemy_status_names(enemy), "Vulnerable"): + card_total = int(card_total * 1.5) + total += card_total + if total - max(0, enemy.block) >= enemy.effective_hp: + if best is None or len(cards) < len(best): + best = cards + if best: + best.sort(key=lambda c: -F.parse_card_damage(c).raw_total) + return best + + +def _fallback_combat(f: F.CombatFacts) -> Decision: + """ + Documented heuristic, used when Jev is unsure or unavailable. + + Order follows the STS2MCP strategy notes: never block a sleeping or + buffing enemy, and do not waste energy on block when nothing is incoming. + """ + playable = f.playable + + # Don't die holding potions. If the incoming hit is lethal and a potion is + # available, spend one -- "Dying with full potions is the worst outcome." + if f.threat == F.THREAT_LETHAL: + usable = _usable_potions(f) + if usable: + return Decision("use_potion", _potion_params(usable[0], f), + "incoming damage is lethal; spend a potion", "fallback") + + # Nothing incoming: spend everything on damage. + if f.unblocked_damage <= 0: + attacks = [c for c in playable if F.parse_card_damage(c).raw_total > 0] + if attacks: + best = max(attacks, key=lambda c: F.parse_card_damage(c).raw_total) + return Decision( + "play_card", + _target_params(best, f), + "no incoming damage, take the biggest attack", + "fallback", + ) + if playable: + return Decision("play_card", _target_params(playable[0], f), + "nothing incoming, dump a card", "fallback") + return Decision("end_turn", {}, "no playable cards", "fallback") + + # Only spend energy on block when the hit actually matters. At full health + # against a small hit, front-loading damage is better: HP is a resource. + # (STS2MCP strategy notes: "HP is a resource, not a score", + # "Front-load damage", "Don't waste energy on block when enemies aren't attacking".) + must_respect = ( + f.threat in (F.THREAT_HEAVY, F.THREAT_SEVERE, F.THREAT_LETHAL) + or f.hp_bucket in (F.HP_WOUNDED, F.HP_CRITICAL) + ) + blockers = [c for c in playable if _block_value(c) > 0] + if blockers and must_respect: + best = max(blockers, key=lambda c: _block_value(c)) + return Decision("play_card", _target_params(best, f), + f"threat={f.threat} hp={f.hp_bucket}, take block", + "fallback") + + # Otherwise hit the enemy closest to death. + attacks = [c for c in playable if F.parse_card_damage(c).raw_total > 0] + if attacks and f.enemies: + weakest = min(f.enemies, key=lambda e: e.effective_hp) + best = max(attacks, key=lambda c: F.parse_card_damage(c).raw_total) + params = _target_params(best, f, force_target=weakest.entity_id) + return Decision("play_card", params, + f"no block available, focus {weakest.entity_id}", "fallback") + + if playable: + return Decision("play_card", _target_params(playable[0], f), + "no better option", "fallback") + return Decision("end_turn", {}, "nothing playable", "fallback") + + +def _block_value(card: dict) -> int: + return F.block_value(card) + + +def _target_params(card: dict, f: F.CombatFacts, force_target: str | None = None) -> dict: + params: dict[str, Any] = {"card_index": card.get("index")} + if card.get("target_type") == "AnyEnemy": + if force_target: + params["target"] = force_target + elif len(f.enemies) == 1: + params["target"] = f.enemies[0].entity_id + elif f.enemies: + params["target"] = min(f.enemies, key=lambda e: e.effective_hp).entity_id + # With no enemies there is no target to give. The caller must not play + # this card at all -- the game rejects it with + # "Card requires a target. Provide 'target' with an entity_id." + return params + + +def _needs_enemy_target(card: dict) -> bool: + return card.get("target_type") == "AnyEnemy" + + +def _usable_potions(f: F.CombatFacts) -> list[dict]: + """Potions the game says can be used right now.""" + return [p for p in f.potions if p.get("can_use_in_combat", True)] + + +def _potion_params(potion: dict, f: F.CombatFacts) -> dict: + """`slot` is the potion slot index, not the list position.""" + params: dict[str, Any] = {"slot": potion.get("slot", 0)} + if potion.get("target_type") == "AnyEnemy" and f.enemies: + params["target"] = min(f.enemies, key=lambda e: e.effective_hp).entity_id + return params + + +def _jev_combat(f: F.CombatFacts, client: JevClient) -> Decision: + """ + Ask Jev for this turn's decision. ONE batched call for everything. + + Potions are included in the same call. They cost no energy, so they are a + parallel option rather than a separate step, and adding the questions costs + no measurable latency (1 question 0.73s, 3 questions over a full state + 0.90s). Run 001 died to the Act 1 boss holding all three potions, so this + is the highest-value gap to close. + """ + options: dict[str, str] = {} + by_key: dict[str, dict] = {} + for c in f.playable: + key = f"card{c['index']}" + by_key[key] = c + options[key] = ( + f"{c['name']} (cost {F._as_int(c.get('cost'))}): {c.get('description')}" + ) + + potion_options: dict[str, str] = {} + potion_by_key: dict[str, dict] = {} + for p in _usable_potions(f): + key = f"potion{p.get('slot')}" + potion_by_key[key] = p + potion_options[key] = f"{p.get('name')}: {p.get('description')}" + + if not options and not potion_options: + return Decision("end_turn", {}, "nothing playable and no usable potions", "code") + + questions: dict[str, dict] = {} + + if options: + questions["best_play"] = choice( + "Which single play best advances winning this fight?", options) + + if potion_options: + questions["use_potion"] = noul( + "Given `combat.incoming_threat`, `combat.your_health` and `combat.round`, " + "is spending a potion this turn clearly better than saving it?" + ) + questions["which_potion"] = choice( + "If a potion is used now, which one?", potion_options) + + if len(f.enemies) > 1: + questions["target"] = choice( + "If a single-target attack is played, which enemy should it hit first?", + { + e.entity_id: ( + f"{e.name}, health {F._hp_bucket(e.hp_pct, e.hp)}, " + f"incoming {e.incoming_damage}, " + f"{', '.join(e.intent_text) or 'intent unknown'}" + ) + for e in f.enemies + }, + ) + + questions["should_defend"] = noul( + "Given `combat.incoming_threat` and `combat.your_health`, " + "is preventing damage more valuable than dealing damage this turn?" + ) + + response = client.ask(f.to_state(), questions) + + # Potions first: buffs and debuffs should land before the cards they boost. + if potion_options: + wants = response.get("use_potion") + pick_potion = response.get("which_potion") + + # CODE decides THAT a potion must be used when the incoming hit is + # lethal and no card play can prevent it. Jev only decides WHICH. + # Measured: on this exact state Jev answered use_potion=0.61, just + # under the 0.65 Noul floor, and the bot would have played a Strike + # and died. Saving a potion is not a choice when the alternative is + # losing the run. + hard_need = f.threat == F.THREAT_LETHAL and not f.survives_with_cards + + soft_ok = isinstance(wants, NoulAnswer) and wants.yes and gate(wants) + + if hard_need: + potion = None + if pick_potion is not None and gate(pick_potion): + potion = potion_by_key.get(pick_potion.choice) + if potion is None: + usable = _usable_potions(f) + potion = usable[0] if usable else None + if potion is not None: + return Decision( + "use_potion", + _potion_params(potion, f), + "lethal hit and no card can prevent it; spend a potion", + "code", + ) + + if soft_ok and pick_potion is not None and gate(pick_potion): + potion = potion_by_key.get(pick_potion.choice) + if potion is not None: + return Decision( + "use_potion", + _potion_params(potion, f), + f"jev used {potion.get('name')}", + "jev", + pick_potion.confidence, + ) + + if not options: + return Decision("end_turn", {}, "no playable cards", "code") + + pick = response["best_play"] + if not gate(pick): + return _fallback_combat(f) + + card = by_key.get(pick.choice) + if card is None: + return _fallback_combat(f) + + params = _target_params(card, f) + + target_answer = response.get("target") + if ( + params.get("card_index") is not None + and card.get("target_type") == "AnyEnemy" + and target_answer is not None + and gate(target_answer, CONFIDENCE_FLOOR) + ): + params["target"] = target_answer.choice + + return Decision( + "play_card", + params, + f"jev chose {card['name']}", + "jev", + confidence=pick.confidence, + ) + + +def combat_decision(f: F.CombatFacts, client: JevClient | None) -> Decision: + # Between turns (enemy acting, animations, post-combat) there is nothing to + # do but re-observe. Acting here just produces rejected actions. + if not f.in_play_phase: + return Decision("__wait__", {}, "not our play phase; re-observe", "code") + + # Enemies gone but combat still in progress: the game is spawning the next + # wave (Phrog Parasite does this). Measured, playing here emits a targetless + # attack and the game rejects it with "Card requires a target". + if not f.enemies: + return Decision("__wait__", {}, "no enemies yet; re-observe", "code") + + # 1. Deterministic lethal. + for enemy in f.enemies: + line = _lethal_line(f.playable, f.energy, [], enemy) + if line: + card = line[0] + params = _target_params(card, f, force_target=enemy.entity_id) + return Decision( + "play_card", params, + f"lethal line on {enemy.entity_id} ({len(line)} cards)", "code", + ) + + # 2. Jev for preference, 3. heuristic if it is unsure. + if client is not None: + try: + return _jev_combat(f, client) + except JevError as exc: + print(f" [jev unavailable: {str(exc)[:90]}]") + return _fallback_combat(f) + + +# -------------------------------------------------------------------------- +# Non-combat states +# -------------------------------------------------------------------------- + +RARITY_RANK = {"Basic": 0, "Common": 1, "Uncommon": 2, "Rare": 3, "Special": 4} + + +def best_by_noul(response, keys, threshold: float) -> tuple[str | None, float]: + """ + Pick the highest absolute Noul among `keys`, or (None, best) below threshold. + + Absolute Nouls do not dilute as the candidate count grows, unlike a single + Choice over many options. This is the documented re-ranking pattern. + """ + ranked = [ + (ans.noul, key) + for key in keys + if isinstance((ans := response.get(key)), NoulAnswer) + ] + if not ranked: + return None, 0.0 + best_noul, best_key = max(ranked) + return (best_key if best_noul >= threshold else None), best_noul + + +def best_by_rarity(cards: list[dict]) -> dict | None: + """Deterministic fallback: prefer the rarest card, never blindly index 0.""" + if not cards: + return None + return max(cards, key=lambda c: RARITY_RANK.get(str(c.get("rarity")), 1)) + + +def card_reward_decision(obs: dict, client: JevClient | None, deck: dict | None) -> Decision: + """ + Deck building -- the macro decision that actually decides a run. + + The reward state does NOT expose the deck, so `deck` is a composition + snapshot taken from a combat state (which exposes all four piles). + """ + reward = obs.get("card_reward") or {} + cards = reward.get("cards") or [] + if not cards: + return Decision("skip_card_reward", {}, "no cards offered", "code") + + if client is None: + fallback_card = best_by_rarity(cards) + return Decision("select_card_reward", {"card_index": fallback_card["index"]}, + "no jev; took the rarest", "fallback") + + run = obs.get("run") or {} + player = obs.get("player") or {} + + state = { + "run": { + "act": run.get("act"), + "floor": run.get("floor"), + "ascension": run.get("ascension"), + }, + "character": player.get("character"), + "health": F._hp_bucket( + (player.get("hp") or 0) / (player.get("max_hp") or 1), + player.get("hp"), + ), + "deck_composition": deck or "unknown", + "offered": { + f"card{c['index']}": { + "name": c["name"], + "type": c["type"], + "text": c["description"], + } + for c in cards + }, + } + + # ONE question decides skip, and one per card decides WHICH. + # + # Previously a hardcoded CARD_PICK_THRESHOLD of 0.60 made the skip call, and + # Jev only ranked. Measured, Jev's ratings form a smooth continuum from 0.45 + # to 0.68, so that threshold was cutting the distribution in half -- the + # skip was the threshold's decision, not the model's. Now Jev decides skip + # explicitly and the per-card ratings only choose which card to take. + keys = [f"card{c['index']}" for c in cards] + questions: dict[str, dict] = { + "skip_all": noul( + "Should this deck skip EVERY card in `offered` and stay lean? " + "A lean deck draws its key cards more often, so skipping is often " + "correct." + ) + } + for c in cards: + questions[f"good_card{c['index']}"] = noul( + f"Would `offered.card{c['index']}` make this deck stronger?" + ) + + response = client.ask(state, questions) + + can_skip = reward.get("can_skip", True) + skip_answer = response.get("skip_all") + + # JEV decides the skip. Only a confident yes skips. + if ( + can_skip + and isinstance(skip_answer, NoulAnswer) + and skip_answer.yes + and gate(skip_answer) + ): + global _rewards_skipped_card + _rewards_skipped_card = True + return Decision("skip_card_reward", {}, + f"jev: skip all (noul={skip_answer.noul:.2f}); keep the deck lean", + "jev", skip_answer.noul) + + # Otherwise take the card Jev rated highest. + best_key, best_noul = best_by_noul(response, [f"good_{k}" for k in keys], 0.0) + + # "combined": Jev declined to skip, but if nothing it rated clears the + # floor the deck is better off lean. This is the safety net that stops the + # deck bloating when Jev is indifferent. + if CARD_SKIP_POLICY == "combined" and can_skip and best_noul < CARD_PICK_THRESHOLD: + _rewards_skipped_card = True + return Decision("skip_card_reward", {}, + f"combined: jev said don't skip but best={best_noul:.2f} " + f"< {CARD_PICK_THRESHOLD}; keep the deck lean", + "fallback", best_noul) + + if best_key is None: + if can_skip: + _rewards_skipped_card = True + return Decision("skip_card_reward", {}, + "no usable ratings; keeping the deck lean", "fallback") + fallback_card = best_by_rarity(cards) + return Decision("select_card_reward", {"card_index": fallback_card["index"]}, + "no ratings and cannot skip; took the rarest", "fallback") + + card = next((c for c in cards if f"good_card{c['index']}" == best_key), None) + if card is None: + fallback_card = best_by_rarity(cards) + return Decision("select_card_reward", + {"card_index": fallback_card["index"]}, + "unmapped choice", "fallback") + return Decision("select_card_reward", {"card_index": card["index"]}, + f"jev took {card['name']} (noul={best_noul:.2f})", + "jev", best_noul) + + +def relic_select_decision(obs: dict, client: JevClient | None) -> Decision: + """ + Boss/elite relic choice. Verified shape: `relic_select.relics[]` with + index/id/name/description/rarity, plus `can_skip`. + """ + node = obs.get("relic_select") or obs.get("relics") or {} + if not isinstance(node, dict): + node = {} + relics = [r for r in (node.get("relics") or node.get("options") or []) + if isinstance(r, dict)] + + if not relics: + return Decision("skip_relic_selection", {}, "no relics parsed", "code") + + # A single offered relic is not a choice. + if len(relics) == 1: + return Decision("select_relic", {"index": relics[0].get("index", 0)}, + f"only relic: {relics[0].get('name')}", "code") + + def rarity_pick() -> dict: + return max(relics, key=lambda r: RARITY_RANK.get(str(r.get("rarity")), 1)) + + if client is None: + chosen = rarity_pick() + return Decision("select_relic", {"index": chosen.get("index", 0)}, + f"no jev; took the rarest ({chosen.get('name')})", "fallback") + + keys = [f"relic{r.get('index', 0)}" for r in relics] + questions = { + f"good_{k}": noul( + f"Would taking `relics.{k}` make this run stronger than the other " + "relics offered?" + ) + for k in keys + } + if node.get("can_skip"): + questions["want_any"] = noul( + "Is any relic in `relics` worth taking over skipping?" + ) + + response = client.ask( + { + "character": (obs.get("player") or {}).get("character"), + "relics": { + f"relic{r.get('index', 0)}": { + "name": r.get("name"), + "rarity": r.get("rarity"), + "text": r.get("description"), + } + for r in relics + }, + }, + questions, + ) + + wants = response.get("want_any") + if isinstance(wants, NoulAnswer) and (not wants.yes or not gate(wants)): + if node.get("can_skip"): + return Decision("skip_relic_selection", {}, + f"jev: skip (noul={wants.noul:.2f})", "jev", wants.noul) + + best_key, best_noul = best_by_noul(response, keys, CARD_PICK_THRESHOLD) + if best_key is None: + chosen = rarity_pick() + return Decision("select_relic", {"index": chosen.get("index", 0)}, + f"nothing cleared {CARD_PICK_THRESHOLD} (best {best_noul:.2f}); " + f"took the rarest ({chosen.get('name')})", + "fallback", best_noul or None) + + chosen = next((r for r in relics if f"relic{r.get('index', 0)}" == best_key), None) + if chosen is None: + chosen = rarity_pick() + return Decision("select_relic", {"index": chosen.get("index", 0)}, + "unmapped choice", "fallback") + return Decision("select_relic", {"index": chosen.get("index", 0)}, + f"jev chose {chosen.get('name')} (noul={best_noul:.2f})", + "jev", best_noul) + + +MAP_NODE_MEANINGS = { + "Monster": "A normal fight. Costs some health, pays a card reward and gold.", + "Elite": "A hard fight. Pays a relic. Dangerous at low health.", + "Rest": "A campfire. Heal, or upgrade a card.", + "Shop": "Spend gold on cards, relics, potions, or removing a card.", + "Treasure": "A free relic with no fight.", + "Unknown": "Unknown. Could be a fight, an event, a shop, or treasure.", + "Boss": "The act boss. Ends the act.", +} + + +def _fallback_map(opts: list[dict], hp_pct: float, gold: int) -> Decision: + """Documented pathing heuristic from the STS2MCP strategy notes.""" + kinds = {str(o.get("type")): o for o in opts} + + if hp_pct < 0.5 and "Rest" in kinds: + return Decision("choose_map_node", {"index": kinds["Rest"]["index"]}, + f"hp {hp_pct:.0%} is low, take the rest site", "fallback") + if hp_pct > 0.7 and "Elite" in kinds: + return Decision("choose_map_node", {"index": kinds["Elite"]["index"]}, + f"hp {hp_pct:.0%} is healthy, take the elite for a relic", "fallback") + if "Treasure" in kinds: + return Decision("choose_map_node", {"index": kinds["Treasure"]["index"]}, + "free relic", "fallback") + if gold >= 100 and "Shop" in kinds: + return Decision("choose_map_node", {"index": kinds["Shop"]["index"]}, + f"{gold} gold, visit the shop", "fallback") + if "Unknown" in kinds and hp_pct > 0.6: + return Decision("choose_map_node", {"index": kinds["Unknown"]["index"]}, + "healthy enough to gamble on unknown", "fallback") + return Decision("choose_map_node", {"index": opts[0]["index"]}, + "default to the first option", "fallback") + + +def map_decision(obs: dict, client: JevClient | None, deck: dict | None) -> Decision: + """ + Map pathing. A macro decision: elite fights pay relics but can end a run. + The choice depends on health, gold, and deck quality, so Jev is asked. + """ + m = obs.get("map") or {} + opts = [o for o in (m.get("next_options") or []) if isinstance(o, dict)] + + if not opts: + return Decision("__wait__", {}, "map has no next options yet; re-observe", "code") + if len(opts) == 1: + return Decision("choose_map_node", {"index": opts[0]["index"]}, + f"only option: {opts[0].get('type')}", "code") + + player = obs.get("player") or {} + run = obs.get("run") or {} + hp = player.get("hp") or 0 + max_hp = player.get("max_hp") or 1 + hp_pct = hp / max_hp if max_hp else 0.0 + gold = player.get("gold") or 0 + + if client is None: + return _fallback_map(opts, hp_pct, gold) + + options = { + f"node{o['index']}": ( + f"{o.get('type')} (column {o.get('col')}, row {o.get('row')}). " + f"{MAP_NODE_MEANINGS.get(str(o.get('type')), '')}" + ) + for o in opts + } + + boss = m.get("boss") or {} + state = { + "run": {"act": run.get("act"), "floor": run.get("floor")}, + "character": player.get("character"), + "health": F._hp_bucket(hp_pct, hp), + "gold": gold, + "deck_composition": deck or "unknown", + "act_boss": boss.get("name") if isinstance(boss, dict) else None, + } + + response = client.ask(state, { + "next_node": choice( + "Which node should the player travel to next? Health is a resource, " + "but an elite fight at low health can end the run.", + options, + ) + }) + + pick = response["next_node"] + if not gate(pick): + return _fallback_map(opts, hp_pct, gold) + + node = next((o for o in opts if f"node{o['index']}" == pick.choice), None) + if node is None: + return _fallback_map(opts, hp_pct, gold) + return Decision("choose_map_node", {"index": node["index"]}, + f"jev chose {node.get('type')}", "jev", pick.confidence) + + +_last_card_select_sig: str | None = None +_card_select_picked = False +_card_select_confirmed = False +_card_select_chosen: list[int] = [] + +# Deterministic fallbacks for a grid selection screen, used only when Jev does +# not clear the threshold. Lower rank wins. +# +# The old fallback was `cards[0]`, which on an upgrade screen is always a basic +# Strike -- which is why the bot only ever upgraded Strikes. +def upgrade_rank(card: dict) -> int: + name = str(card.get("name") or "") + if card.get("is_upgraded"): + return 9 + if name.startswith("Bash"): + return 0 + if not (name.startswith("Strike") or name.startswith("Defend")): + return 1 + return 5 if name.startswith("Strike") else 6 + + +def removal_rank(card: dict) -> int: + """For a removal screen the ordering is inverted: shed basics first.""" + # An already-upgraded card is worth keeping, so check that BEFORE the name: + # otherwise "Strike+" matches the Strike rule and becomes the top target. + if card.get("is_upgraded"): + return 9 + name = str(card.get("name") or "") + if name.startswith("Strike"): + return 0 + if name.startswith("Defend"): + return 1 + return 3 + + +def screen_kind(screen: str, prompt: str = "") -> str: + """ + Normalise a card_select screen to one of: upgrade, remove, transform. + + Two traps, both measured: + * The mod maps only four screens to friendly names and falls through to + the RAW C# CLASS NAME for everything else -- e.g. + "NDeckEnchantSelectScreen". + * `screen_type` can be the generic "select" while the PROMPT says what is + actually happening. Measured: screen_type "select" with prompt + "Choose 5 cards to Remove." was treated as an upgrade, so Jev was asked + "would upgrading this make the deck stronger?" on a REMOVAL screen and + offered to remove Bash. + + So the prompt is consulted too. + """ + text = f"{screen or ''} {prompt or ''}".lower() + if "remove" in text: + return "remove" + if "transform" in text: + return "transform" + # upgrade, smith, enchant: pick the card that benefits most. + return "upgrade" + + +def card_select_fallback(cards: list[dict], screen: str, + prompt: str = "") -> dict: + """Deterministic pick when the model does not clear the threshold.""" + kind = screen_kind(screen, prompt) + if kind in ("remove", "transform"): + return min(cards, key=removal_rank) + return min(cards, key=upgrade_rank) + + +def _card_select_pick(index: int, reason: str, source: str, + confidence: float | None = None) -> Decision: + """Emit a select_card and remember that this index has been toggled on.""" + global _card_select_picked + _card_select_picked = True + if index not in _card_select_chosen: + _card_select_chosen.append(index) + return Decision("select_card", {"index": index}, reason, source, confidence) + + +def card_select_need(prompt: str) -> int: + """ + How many cards this screen wants. + + Measured: "Choose 5 cards to Remove." with can_confirm FALSE until all five + are picked, and `card_select` exposes NO `selected_cards` field. So the + count is parsed from the prompt and tracked by us. + """ + match = re.search(r"choose\s+(\d+)\s+cards?", str(prompt or ""), re.IGNORECASE) + return int(match.group(1)) if match else 1 + + +def card_select_decision(obs: dict, client: JevClient | None, deck: dict | None) -> Decision: + """ + Grid selection overlay: upgrade, transform, remove, choose-a-card. + + Two traps, both observed live: + * `select_card` TOGGLES on grid screens. Calling it twice on one index + deselects and freezes the screen. + * A preview left over from a desynced state makes `confirm_selection` + report ok while changing nothing. If the same screen reappears + unchanged, reset with cancel_selection instead of confirming again. + """ + global _last_card_select_sig, _card_select_picked, _card_select_confirmed + + cs = obs.get("card_select") or {} + sig = json.dumps(cs, sort_keys=True) + repeated = sig == _last_card_select_sig + _last_card_select_sig = sig + + prompt = str(cs.get("prompt") or "Choose a card.") + screen = str(cs.get("screen_type") or "") + need = card_select_need(prompt) + + # Confirm once enough cards have been picked. Do NOT rely on + # `preview_showing`: screen_type "NDeckEnchantSelectScreen" toggles a + # selection with NO preview at all. And do NOT confirm early on a + # multi-select: "Choose 5 cards to Remove" keeps can_confirm FALSE until all + # five are chosen. + if len(_card_select_chosen) >= need: + if cs.get("can_confirm"): + if _card_select_confirmed: + return Decision("cancel_selection", {}, + "confirm did not apply; reset the screen", "code") + _card_select_confirmed = True + return Decision("confirm_selection", {}, + f"confirm {len(_card_select_chosen)}/{need} selected", "code") + # Enough chosen but the game has not enabled confirm yet. Selecting more + # would overshoot, so wait. + return Decision("__wait__", {}, + f"{len(_card_select_chosen)}/{need} chosen; waiting for confirm", + "code") + + cards = [c for c in (cs.get("cards") or []) if isinstance(c, dict)] + if not cards: + return Decision("cancel_selection", {}, "no cards to select", "code") + + # `select_card` TOGGLES, so never re-select an index we already toggled on. + remaining = [c for c in cards if c.get("index") not in _card_select_chosen] + if not remaining: + if cs.get("can_confirm"): + _card_select_confirmed = True + return Decision("confirm_selection", {}, "all selectable cards chosen", "code") + return Decision("__wait__", {}, + f"{len(_card_select_chosen)}/{need} chosen; nothing new to select", + "code") + + # One ABSOLUTE Noul per candidate, argmax in code. A single Choice over a + # 13+ card deck diluted badly and the fallback then always took index 0. + # Candidates are `remaining` -- never an index we already toggled on. + keys = [f"card{c['index']}" for c in remaining] + kind = screen_kind(screen, prompt) + questions: dict[str, dict] = {} + for c in remaining: + key = f"card{c['index']}" + if kind == "remove": + questions[f"good_{key}"] = noul( + f"Would removing `cards.{key}` make this deck stronger?" + ) + elif kind == "transform": + questions[f"good_{key}"] = noul( + f"Would transforming `cards.{key}` into a random card make " + "this deck stronger?" + ) + else: + questions[f"good_{key}"] = noul( + f"Would upgrading `cards.{key}` make this deck stronger?" + ) + + if client is None or len(remaining) == 1: + fallback_card = card_select_fallback(remaining, screen, prompt) + return _card_select_pick(fallback_card["index"], + f"no jev; {screen or 'select'} heuristic", "fallback") + + response = client.ask( + { + "prompt": prompt, + "screen_type": screen, + "deck_composition": deck or "unknown", + "cards": { + f"card{c['index']}": { + "name": c["name"], + "text": c["description"], + "already_upgraded": c.get("is_upgraded"), + } + for c in remaining + }, + }, + questions, + ) + + best_key, best_noul = best_by_noul( + response, [f"good_{k}" for k in keys], CARD_PICK_THRESHOLD + ) + if best_key is None: + fallback_card = card_select_fallback(remaining, screen, prompt) + return _card_select_pick( + fallback_card["index"], + f"no card cleared {CARD_PICK_THRESHOLD} (best {best_noul:.2f}); " + f"used the {screen or 'select'} heuristic", + "fallback", best_noul or None) + + card = next((c for c in remaining if f"good_card{c['index']}" == best_key), None) + if card is None: + fallback_card = card_select_fallback(remaining, screen, prompt) + return _card_select_pick(fallback_card["index"], "unmapped choice", "fallback") + return _card_select_pick(card["index"], + f"jev chose {card['name']} (noul={best_noul:.2f})", + "jev", best_noul) + + +def event_decision(obs: dict, client: JevClient | None) -> Decision: + """ + Events. Option 0 is often locked, and Ancient events start in dialogue, + which is why a blind choose_event_option(index=0) gets rejected. + """ + ev = obs.get("event") or {} + + if ev.get("in_dialogue"): + return Decision("advance_dialogue", {}, "click through dialogue", "code") + + opts = [o for o in (ev.get("options") or []) if isinstance(o, dict)] + usable = [o for o in opts if not o.get("is_locked") and not o.get("was_chosen")] + if not usable: + usable = [o for o in opts if not o.get("is_locked")] + if not usable: + return Decision("__wait__", {}, "no usable event options; re-observe", "code") + + pool = [o for o in usable if not o.get("is_proceed")] or usable + + if len(pool) == 1: + return Decision("choose_event_option", {"index": pool[0]["index"]}, + f"only option: {pool[0].get('title')}", "code") + + if client is None: + return Decision("choose_event_option", {"index": pool[0]["index"]}, + "first usable option", "fallback") + + player = obs.get("player") or {} + options = { + f"opt{o['index']}": f"{o.get('title')}: {o.get('description')}" + for o in pool + } + + questions: dict[str, dict] = { + "best_option": choice( + "Which event option is the best choice for the player?", options) + } + + response = client.ask( + { + "event": ev.get("body") or "", + "character": player.get("character"), + "health": F._hp_bucket( + (player.get("hp") or 0) / (player.get("max_hp") or 1), + player.get("hp"), + ), + "gold": player.get("gold"), + }, + questions, + ) + + pick = response["best_option"] + + if isinstance(pick, ChoiceAnswer): + top = pick.probabilities.get(pick.choice, 0.0) + runner = max( + (v for k, v in pick.probabilities.items() if k != pick.choice), + default=0.0, + ) + else: + top = runner = 0.0 + + chosen = next((o for o in pool if f"opt{o['index']}" == getattr(pick, "choice", None)), None) + + # Act on the model only when it is BOTH confident and not obviously risky. + if ( + chosen is not None + and top >= EVENT_TOP_MIN + and (top - runner) >= EVENT_MARGIN_MIN + and event_safety_rank(chosen) <= 0 + ): + return Decision("choose_event_option", {"index": chosen["index"]}, + f"jev chose {chosen.get('title')} ({top:.2f})", "jev", + getattr(pick, "confidence", None)) + + safest = min(pool, key=event_safety_rank) + if chosen is not None and event_safety_rank(chosen) > 0: + reason = f"jev picked risky '{chosen.get('title')}'; took safest option" + else: + reason = f"uncertain ({top:.2f}); took safest option" + return Decision("choose_event_option", {"index": safest["index"]}, reason, + "fallback", top or None) + + +def rest_site_decision(obs: dict) -> Decision: + """ + Heal when hurt, otherwise upgrade. (STS2MCP notes: rest before boss.) + + Rest options expose `name` and `id`, NOT `title`, plus `is_enabled`. + """ + rs = obs.get("rest_site") or {} + opts = [ + o for o in (rs.get("options") or []) + if isinstance(o, dict) and o.get("is_enabled", True) + ] + player = obs.get("player") or {} + hp_pct = (player.get("hp") or 0) / (player.get("max_hp") or 1) + + if not opts: + # As with the shop, do not gate the exit on `can_proceed`. + return Decision("proceed", {}, "nothing to do here; proceed", "code") + + def label(o: dict) -> str: + return f"{o.get('name', '')} {o.get('id', '')} {o.get('description', '')}".lower() + + rest = next((o for o in opts if "rest" in label(o) or "heal" in label(o)), None) + smith = next((o for o in opts if "smith" in label(o) or "upgrade" in label(o)), None) + + if hp_pct < 0.6 and rest is not None: + return Decision("choose_rest_option", {"index": rest["index"]}, + f"hp {hp_pct:.0%}, heal", "fallback") + if smith is not None: + return Decision("choose_rest_option", {"index": smith["index"]}, + f"hp {hp_pct:.0%}, upgrade a card", "fallback") + return Decision("choose_rest_option", {"index": opts[0]["index"]}, + "first rest option", "fallback") + + +_last_shop_sig: str | None = None +_last_shop_purchased = False + +# Minimum absolute Noul before buying anything in a shop. Absolute judgements +# can legitimately be low for every candidate, so this is a floor, not a rank. +SHOP_BUY_THRESHOLD = 0.60 + +# Minimum absolute Noul before adding an offered card to the deck. +CARD_PICK_THRESHOLD = 0.60 + +# How the card-reward skip is decided. +# "jev" : Jev decides via `skip_all`. Measured, Jev almost never says +# skip (13 takes / 0 skips over three sessions), so the deck +# grows by roughly +3.3 cards versus the threshold policy. +# "combined" : skip when Jev says skip OR when the best card is clearly weak. +# Keeps Jev in charge of the ranking while putting a floor under +# the deck size. +CARD_SKIP_POLICY = "jev" + + +def shop_item_text(item: dict) -> tuple[str, str]: + """ + Resolve a shop item's display name and description. + + The shop uses category-specific field names, verified live: + card -> card_name / card_description + relic -> relic_name / relic_description + potion -> potion_name / potion_description + card_removal -> neither + Reading `name`/`description` yields None for every category. + """ + category = str(item.get("category") or "") + if category == "card_removal": + return "Card Removal", "Remove one card from your deck permanently." + for prefix in ("card", "relic", "potion"): + name = item.get(f"{prefix}_name") + if name: + return str(name), str(item.get(f"{prefix}_description") or "") + return str(item.get("name") or "?"), str(item.get("description") or "") + + +def shop_decision(obs: dict, client: JevClient | None, deck: dict | None) -> Decision: + """ + Shop. Run 001 finished with 279 unspent gold because this always skipped. + + The state carries `price`, `is_stocked` and `can_afford` per item, so no + affordability arithmetic needs to reach the model. + """ + global _last_shop_sig, _last_shop_purchased + + # `fake_merchant` nests its inventory one level deeper: fake_merchant.shop. + # Reading only obs["shop"] made every fake-merchant shop look empty, so the + # bot always left immediately without buying. + node = obs.get("shop") + if not isinstance(node, dict): + fm = obs.get("fake_merchant") + if isinstance(fm, dict): + node = fm.get("shop") if isinstance(fm.get("shop"), dict) else fm + node = node if isinstance(node, dict) else {} + items = [i for i in (node.get("items") or []) if isinstance(i, dict)] + affordable = [ + i for i in items + if i.get("is_stocked", True) and i.get("can_afford", True) + ] + + if not affordable: + # `can_proceed` is UNRELIABLE here: measured False while proceed() + # worked and moved the game to the map. Never wait on it indefinitely. + return Decision("proceed", {}, "nothing affordable; leave", "code") + + # Only stop buying when we purchased from this EXACT shop state and it did + # not change. Keying on the signature alone made a fresh shop look stalled + # because module-level state leaked in from an earlier screen. + sig = json.dumps(node, sort_keys=True) + if sig != _last_shop_sig: + _last_shop_sig = sig + _last_shop_purchased = False + elif _last_shop_purchased: + return Decision("proceed", {}, "shop unchanged after a purchase; leave", "code") + + if client is None: + return Decision("proceed", {}, "no jev; skip shop", "fallback") + + player = obs.get("player") or {} + run = obs.get("run") or {} + gold = player.get("gold") or 0 + + # A shop can offer 14+ affordable items. A single Choice over all of them + # dilutes the probability mass: measured, the top option scored only 0.26 + # (runner 0.17, margin 0.09) and the margin gate correctly rejected it, so + # the bot would always leave. Use the documented re-ranking pattern instead: + # one ABSOLUTE Noul per candidate, then take the argmax in code. Absolute + # judgements do not dilute as the candidate count grows. + item_keys: dict[str, dict] = {f"item{i['index']}": i for i in affordable} + + state = { + "run": {"act": run.get("act"), "floor": run.get("floor")}, + "character": player.get("character"), + "health": F._hp_bucket( + (player.get("hp") or 0) / (player.get("max_hp") or 1), + player.get("hp"), + ), + "gold": gold, + "deck_composition": deck or "unknown", + "items": { + key: { + "name": shop_item_text(item)[0], + "category": item.get("category"), + "price": item.get("price"), + "text": shop_item_text(item)[1], + } + for key, item in item_keys.items() + }, + } + + # Ask about DECK FIT ONLY. Measured on this exact shop: putting the price + # into the question collapsed the spread across candidates from 0.48 to + # 0.28 and pulled the top item down from 0.67 to 0.51. Weighing value + # against a number is exactly what Jev is documented to fail at. + # Affordability is already filtered in code, so the price stays in the + # state for context but stays OUT of the question. + questions: dict[str, dict] = {} + for key in item_keys: + questions[f"worth_{key}"] = noul( + f"Would `items.{key}` make this deck stronger?" + ) + + response = client.ask(state, questions) + + ranked = [ + (ans.noul, key) + for key in item_keys + if isinstance((ans := response.get(f"worth_{key}")), NoulAnswer) + ] + if not ranked: + return Decision("proceed", {}, "no usable answers; leave", "fallback") + + best_noul, best_key = max(ranked) + if best_noul < SHOP_BUY_THRESHOLD: + return Decision("proceed", {}, + f"best item only noul={best_noul:.2f}; leave", "jev", best_noul) + + item = item_keys[best_key] + _last_shop_purchased = True + return Decision("shop_purchase", {"index": item["index"]}, + f"jev bought {shop_item_text(item)[0]} (noul={best_noul:.2f})", + "jev", best_noul) + + +def treasure_decision(obs: dict, client: JevClient | None, + deck: dict | None = None) -> Decision: + """ + The chest auto-opens. While it opens, `relics` is absent and only + `can_proceed` is set, so claiming then just gets rejected. + + A chest usually offers one relic, in which case there is nothing to judge. + Jev is only consulted when there is a real choice. + """ + tr = obs.get("treasure") or {} + relics = [r for r in (tr.get("relics") or []) if isinstance(r, dict)] + + if not relics: + if tr.get("can_proceed"): + return Decision("proceed", {}, "treasure done; proceed", "code") + return Decision("__wait__", {}, "chest opening; re-observe", "code") + + if len(relics) == 1 or client is None: + return Decision("claim_treasure_relic", {"index": relics[0].get("index", 0)}, + f"claim {relics[0].get('name')}", "code") + + options = { + f"relic{r.get('index', 0)}": ( + f"{r.get('name')} ({r.get('rarity')}): {r.get('description')}" + ) + for r in relics + } + player = obs.get("player") or {} + + response = client.ask( + { + "character": player.get("character"), + "health": F._hp_bucket( + (player.get("hp") or 0) / (player.get("max_hp") or 1), + player.get("hp"), + ), + "deck_composition": deck or "unknown", + "relics": { + f"relic{r.get('index', 0)}": { + "name": r.get("name"), + "text": r.get("description"), + } + for r in relics + }, + }, + {"best_relic": choice("Which relic is strongest for this run?", options)}, + ) + + pick = response["best_relic"] + if not gate(pick): + return Decision("claim_treasure_relic", {"index": relics[0].get("index", 0)}, + "low confidence; take the first", "fallback") + chosen = next((r for r in relics if f"relic{r.get('index', 0)}" == pick.choice), None) + if chosen is None: + return Decision("claim_treasure_relic", {"index": relics[0].get("index", 0)}, + "unmapped choice", "fallback") + return Decision("claim_treasure_relic", {"index": chosen.get("index", 0)}, + f"jev chose {chosen.get('name')}", "jev", pick.confidence) + + +# Rough potion value, used only to choose which potion to drop when every slot +# is full. Lower is discarded first. Unknown potions rank in the middle. +POTION_VALUE = { + "FRUIT_JUICE": 3, + "BLOCK_POTION": 4, + "BLOOD_POTION": 4, + "WEAK_POTION": 4, + "FIRE_POTION": 5, + "SWIFT_POTION": 5, + "FEAR_POTION": 5, + "STRENGTH_POTION": 6, + "DEXTERITY_POTION": 6, + "EXPLOSIVE_POTION": 6, + "ENERGY_POTION": 7, +} +DEFAULT_POTION_VALUE = 5 + + +def _potions_full(obs: dict) -> bool: + p = obs.get("player") or {} + return len(p.get("potions") or []) >= (p.get("max_potion_slots") or 0) + + +def _weakest_potion_slot(obs: dict) -> int | None: + potions = [p for p in ((obs.get("player") or {}).get("potions") or []) + if isinstance(p, dict)] + if not potions: + return None + weakest = min( + potions, + key=lambda p: POTION_VALUE.get(str(p.get("id")), DEFAULT_POTION_VALUE), + ) + return weakest.get("slot", 0) + + +_last_rewards_sig: str | None = None +_rewards_skipped_card = False + + +def rewards_decision(obs: dict) -> Decision: + """ + Reward screen. + + Two traps, both observed live: + * Claim right-to-left. Each claim rebuilds the item list and shifts + every later index, so claiming from the front loops forever. + * A potion reward with every slot full reports ok and is silently + dropped, leaving the item in place forever. Free a slot first. + * A SKIPPED CARD REWARD IS NOT CONSUMED. The card stays in the list, so + claim -> card screen -> skip -> claim again loops forever. Track the + skip and ignore card rewards afterwards. + """ + node = obs.get("rewards") or {} + items = [i for i in (node.get("items") or []) if isinstance(i, dict)] + + if _rewards_skipped_card: + items = [i for i in items if i.get("type") != "card"] + + if not items: + return Decision("proceed", {}, "no claimable rewards left; proceed", "code") + + last = items[-1] + idx = last.get("index", len(items) - 1) + + if last.get("type") == "potion" and _potions_full(obs): + slot = _weakest_potion_slot(obs) + if slot is not None: + return Decision( + "discard_potion", {"slot": slot}, + f"potion slots full; drop slot {slot} to take {last.get('potion_name')}", + "fallback", + ) + + return Decision("claim_reward", {"index": idx}, + f"claim reward {idx} (right-to-left)", "code") + + +_last_bundle_sig: str | None = None + + +def bundle_select_decision(obs: dict, client: JevClient | None, + deck: dict | None) -> Decision: + """ + Bundle choice: pick one of several 3-card bundles. + + Same trap as `card_select`: `select_bundle` errors with "A bundle preview + is already open - confirm or cancel it first" once a preview is showing. + Confirm instead, and reset with cancel if a repeated state proves the + confirm did not apply. + """ + global _last_bundle_sig + + bs = obs.get("bundle_select") or {} + sig = json.dumps(bs, sort_keys=True) + repeated = sig == _last_bundle_sig + _last_bundle_sig = sig + + if bs.get("preview_showing") and bs.get("can_confirm"): + if repeated: + return Decision("cancel_bundle_selection", {}, + "bundle preview did not apply; reset", "code") + return Decision("confirm_bundle_selection", {}, "confirm the bundle", "code") + + bundles = [b for b in (bs.get("bundles") or []) if isinstance(b, dict)] + if not bundles: + return Decision("cancel_bundle_selection", {}, "no bundles; cancel", "code") + + if client is None or len(bundles) == 1: + return Decision("select_bundle", {"index": bundles[0].get("index", 0)}, + "select the first bundle", "fallback") + + options = { + f"bundle{b.get('index', 0)}": "; ".join( + f"{c.get('name')} ({c.get('description')})" + for c in (b.get("cards") or []) + ) + for b in bundles + } + + response = client.ask( + { + "prompt": bs.get("prompt"), + "deck_composition": deck or "unknown", + "bundles": { + f"bundle{b.get('index', 0)}": { + "cards": [c.get("name") for c in (b.get("cards") or [])] + } + for b in bundles + }, + }, + {"best_bundle": choice("Which bundle best improves this deck?", options)}, + ) + + pick = response["best_bundle"] + if not gate(pick): + return Decision("select_bundle", {"index": bundles[0].get("index", 0)}, + "low confidence; select the first", "fallback") + chosen = next((b for b in bundles if f"bundle{b.get('index', 0)}" == pick.choice), None) + if chosen is None: + return Decision("select_bundle", {"index": bundles[0].get("index", 0)}, + "unmapped choice", "fallback") + return Decision("select_bundle", {"index": chosen.get("index", 0)}, + f"jev chose bundle {chosen.get('index')}", "jev", pick.confidence) + + +def hand_select_need(prompt: str) -> int | None: + """ + How many cards a hand-select prompt wants, or None for "any number". + + Measured: mode was `simple_select` with prompt "Choose a card to Exhaust." + and can_confirm TRUE. The handler only confirmed for `upgrade_select`, so it + kept toggling between two cards forever. + """ + text = str(prompt or "").lower() + if "any number" in text: + return None + match = re.search(r"choose\s+(\d+)\s+cards?", text) + if match: + return int(match.group(1)) + return 1 # "Choose a card to ..." + + +def hand_select_decision(obs: dict, client: JevClient | None, + deck: dict | None) -> Decision: + """ + In-combat hand selection: exhaust, discard, or replace cards. + + `cards` lists what is still selectable and `selected_cards` lists what is + already chosen. When `cards` is empty the only useful action is to confirm; + blindly sending combat_select_card(index=0) fails with + "Card index 0 out of range (0 selectable cards)". + + Which cards to give up is a real deck decision, but the prompt is usually + "replace/exhaust any number", so a deterministic preference for basic + Strikes and Defends is both safe and close to optimal. + """ + hs = obs.get("hand_select") or {} + cards = [c for c in (hs.get("cards") or []) if isinstance(c, dict)] + selected = hs.get("selected_cards") or [] + mode = str(hs.get("mode") or "") + + # `upgrade_select` pre-selects the card and only needs confirming. Measured: + # mode=upgrade_select, prompt="Confirm Card to Upgrade", can_confirm=true, + # cards=[Defend]. Sending combat_select_card is a silent no-op there, so the + # loop spun 18 times with no progress. combat_confirm_selection closes it. + if mode == "upgrade_select" and hs.get("can_confirm"): + return Decision("combat_confirm_selection", {}, + "upgrade target already chosen; confirm", "code") + + if not cards: + if hs.get("can_confirm"): + return Decision("combat_confirm_selection", {}, + f"nothing left to select ({len(selected)} chosen); confirm", + "code") + return Decision("__wait__", {}, "hand select not ready; re-observe", "code") + + # `combat_select_card` TOGGLES. Re-selecting a card we already chose + # DESELECTS it, so the state never changes and the loop stalls -- measured, + # 15 consecutive "give up Defend" with no progress. `selected_cards` lists + # what is already chosen, so exclude those by NAME (the two lists use + # different index spaces, so names are the only reliable match). + already: dict[str, int] = {} + for entry in selected: + if isinstance(entry, dict): + name = str(entry.get("name") or "") + already[name] = already.get(name, 0) + 1 + + remaining: list[dict] = [] + for c in cards: + name = str(c.get("name") or "") + if already.get(name, 0) > 0: + already[name] -= 1 + continue + remaining.append(c) + + if not remaining: + if hs.get("can_confirm"): + return Decision("combat_confirm_selection", {}, + f"everything selectable is already chosen " + f"({len(selected)}); confirm", "code") + return Decision("__wait__", {}, "nothing new to select; re-observe", "code") + + # Confirm as soon as the prompt's requirement is met. A singular prompt + # ("Choose a card to Exhaust.") needs exactly one; "any number" needs us to + # decide when to stop. + need = hand_select_need(hs.get("prompt")) + if need is not None and len(selected) >= need and hs.get("can_confirm"): + return Decision("combat_confirm_selection", {}, + f"{len(selected)}/{need} chosen; confirm", "code") + + def give_up_priority(card: dict) -> int | None: + """Only basic Strikes and Defends are worth giving up. None means keep.""" + name = str(card.get("name") or "") + if name.startswith("Strike"): + return 0 + if name.startswith("Defend"): + return 1 + return None + + candidates = [ + (give_up_priority(c), c) + for c in remaining + if give_up_priority(c) is not None + ] + + if candidates: + _, target = min(candidates, key=lambda pair: pair[0]) + return Decision("combat_select_card", {"card_index": target.get("index", 0)}, + f"give up {target.get('name')}", "fallback") + + # Nothing disposable left. Do NOT feed good cards to a "choose any number" + # prompt: measured, the previous rule gave up Uppercut, Stomp and Bash. + if hs.get("can_confirm"): + return Decision("combat_confirm_selection", {}, + f"only good cards left; keep them ({len(selected)} chosen)", + "code") + + # The prompt requires a selection, so give up the first thing available. + target = remaining[0] + return Decision("combat_select_card", {"card_index": target.get("index", 0)}, + f"forced selection; gave up {target.get('name')}", "fallback") + + +_crystal_clicked: set[tuple[int, int]] = set() + + +def crystal_sphere_decision(obs: dict) -> Decision: + """ + Crystal Sphere minigame. + + Measured: `can_proceed` is FALSE until tiles are revealed, so the old + unconditional crystal_sphere_proceed was rejected and the run stalled. + Reveal clickable cells until the proceed button unlocks. + """ + cs = obs.get("crystal_sphere") or {} + + if cs.get("can_proceed"): + return Decision("crystal_sphere_proceed", {}, "minigame done; proceed", "code") + + clickable = [c for c in (cs.get("clickable_cells") or []) if isinstance(c, dict)] + + # Never re-click a cell: that wastes a divination and can loop. + fresh = [ + c for c in clickable + if (c.get("x"), c.get("y")) not in _crystal_clicked + ] + + if not fresh: + if clickable: + return Decision("crystal_sphere_proceed", {}, + "nothing new to reveal; try to proceed", "fallback") + return Decision("__wait__", {}, "no clickable cells; re-observe", "code") + + # Prefer the cell closest to the centre, which is where items tend to sit. + width = cs.get("grid_width") or 0 + height = cs.get("grid_height") or 0 + cx, cy = (width - 1) / 2, (height - 1) / 2 + cell = min(fresh, key=lambda c: abs(c.get("x", 0) - cx) + abs(c.get("y", 0) - cy)) + + _crystal_clicked.add((cell.get("x"), cell.get("y"))) + return Decision("crystal_sphere_click_cell", + {"x": cell.get("x"), "y": cell.get("y")}, + f"reveal ({cell.get('x')},{cell.get('y')})", "code") + + +def simple_decision(obs: dict, client: JevClient | None = None, + deck: dict | None = None) -> Decision | None: + """Mechanical screens. Most need no model -- they are pure procedure.""" + st = obs.get("state_type") + + if st == "menu": + screen = obs.get("menu_screen") + if screen == "main": + opts = obs.get("options") or [] + names = [o if isinstance(o, str) else o.get("name") for o in opts] + return Decision("menu_select", + {"option": "continue" if "continue" in names else "singleplayer"}, + "main menu", "code") + if screen == "tutorial_prompt": + return Decision("menu_select", {"option": "no"}, "disable tutorials", "code") + # Mode select, which appears after the first epoch unlock. `embark` is + # rejected here; only standard/daily/custom/back are valid. + if screen == "singleplayer": + return Decision("menu_select", {"option": "standard"}, + "choose standard mode", "code") + # `embark` is REJECTED with "Embark button not available - select a + # character first" unless a character is chosen, and the state carries + # NO "selected" indicator: the mod hardcodes `message` to + # "Select a character." regardless. Verified in AddCharacterSelectMenuState. + # + # Embarking immediately after selecting is also FLAKY -- measured, three + # consecutive "select a character first" rejections -- because the + # selection has not registered yet. So ALTERNATE: select, embark, + # select, embark. A rejected embark is always followed by a fresh + # select, which makes the sequence self-correcting whatever the timing. + if screen == "character_select": + global _charselect_phase + + options = obs.get("options") or [] + names = [o if isinstance(o, str) else o.get("name") for o in options] + pick = next((c for c in ("IRONCLAD", "SILENT") if c in names), None) + + if _charselect_phase == 0 and pick is not None: + _charselect_phase = 1 + return Decision("menu_select", {"option": pick}, + f"select {pick}", "code") + + _charselect_phase = 0 + return Decision("menu_select", {"option": "embark"}, + "embark (a rejected embark is followed by a re-select)", + "code") + return None + + if st == "game_over": + return Decision("menu_select", {"option": "main_menu"}, "run ended", "code") + + if st == "rewards": + return rewards_decision(obs) + + if st == "card_reward": + return card_reward_decision(obs, client, deck) + + if st == "relic_select": + return relic_select_decision(obs, client) + + if st == "map": + return map_decision(obs, client, deck) + + if st == "rest_site": + return rest_site_decision(obs) + + if st == "treasure": + return treasure_decision(obs, client, deck) + + if st == "event": + return event_decision(obs, client) + + if st in ("shop", "fake_merchant"): + return shop_decision(obs, client, deck) + + if st == "hand_select": + return hand_select_decision(obs, client, deck) + + if st == "card_select": + return card_select_decision(obs, client, deck) + + if st == "bundle_select": + return bundle_select_decision(obs, client, deck) + + if st == "crystal_sphere": + return crystal_sphere_decision(obs) + + # Transitions and unhandled overlays are not dead ends. Wait and look again; + # run.py's unchanged-state guard bounds this so a real dead end still stops. + if st in ("unknown", "overlay"): + return Decision("__wait__", {}, f"{st} state; re-observe", "code") + + return None + + +_last_state_type: str | None = None + + +_last_charselect_sig: str | None = None +_charselect_seen = False +_charselect_phase = 0 + +# rewards and card_reward are two views of one flow: claiming a card reward +# opens the card screen, and skipping returns to the rewards screen. Treat them +# as ONE screen group so per-flow state is not cleared on every hop. +SCREEN_GROUPS = {"rewards": "rewards_flow", "card_reward": "rewards_flow"} + + +def _screen_group(state_type: str | None, menu_screen: str | None = None) -> str | None: + """ + A stable key for "which screen am I on". + + The `menu` state_type covers the main menu, mode select, character select + and the tutorial prompt. Those are different screens with different valid + actions, so the menu_screen is part of the key. + """ + if state_type == "menu": + return f"menu:{menu_screen}" + return SCREEN_GROUPS.get(state_type, state_type) + + +def _reset_screen_guards(state_type: str | None, menu_screen: str | None = None) -> None: + """ + Clear per-screen module state when the screen changes. + + These guards detect "the same screen reappeared unchanged", which is only + meaningful within a single screen. Left alone they leak across screens and + runs, so a fresh shop or card grid gets mistaken for a stalled one. Found + by test_brain.py: a fresh fake-merchant shop was reported as + "unchanged after a purchase" because a signature from an earlier case was + still set. + + `unknown` and `overlay` are TRANSITIONS, not screens. Resetting on them + wipes the state mid-flow -- during embark the state flickers through + `unknown`, which used to clear the character-select guard and restart the + select/embark cycle. + """ + global _last_state_type, _last_card_select_sig, _last_bundle_sig + global _last_shop_sig, _last_shop_purchased, _rewards_skipped_card + global _charselect_seen, _charselect_phase + global _card_select_picked, _card_select_confirmed, _card_select_chosen + global _crystal_clicked + + if state_type in ("unknown", "overlay"): + return + + group = _screen_group(state_type, menu_screen) + if group == _last_state_type: + return + _last_state_type = group + _last_card_select_sig = None + _last_bundle_sig = None + _last_shop_sig = None + _last_shop_purchased = False + _rewards_skipped_card = False + _charselect_seen = False + _charselect_phase = 0 + _card_select_picked = False + _card_select_confirmed = False + _card_select_chosen.clear() + _crystal_clicked.clear() + + +def decide(obs: dict, client: JevClient | None, deck: dict | None = None) -> Decision | None: + _reset_screen_guards(obs.get("state_type"), obs.get("menu_screen")) + st = obs.get("state_type") + if st in ("monster", "elite", "boss"): + return combat_decision(F.combat_facts(obs), client) + return simple_decision(obs, client, deck) diff --git a/test_brain.py b/test_brain.py new file mode 100644 index 0000000..357fa25 --- /dev/null +++ b/test_brain.py @@ -0,0 +1,960 @@ +#!/usr/bin/env python3 +""" +test_brain.py -- structural regression tests for the decision layer. + +These catch PROGRAMMATIC bugs, not decision quality: wrong action names, +actions that are not legal for the current state_type, handlers that crash on +a shape, and fallbacks that ignore their own inputs. + +Run with no model: every case uses client=None so the deterministic paths are +exercised. Fast, offline, and it fails loudly. + + python3 test_brain.py +""" + +from __future__ import annotations + +import sys + +import brain +import sts2 + +PASS = 0 +FAIL = 0 + +# `__wait__` is our internal "re-observe" marker, not a game action. +INTERNAL = {"__wait__"} + + +def check(label: str, got, want) -> None: + global PASS, FAIL + if got == want: + PASS += 1 + print(f" ok {label}: {got!r}") + else: + FAIL += 1 + print(f" FAIL {label}: got {got!r}, want {want!r}") + + +def legal(obs: dict, decision: brain.Decision | None) -> bool: + """A decision is valid when its action is legal for this state_type.""" + if decision is None: + return False + if decision.action in INTERNAL: + return True + allowed = sts2.LEGAL_ACTIONS.get(obs.get("state_type"), ()) + return decision.action in allowed + + +def expect_legal(label: str, obs: dict) -> brain.Decision | None: + global PASS, FAIL + try: + decision = brain.decide(obs, None) + except Exception as exc: # noqa: BLE001 - the point is to catch anything + FAIL += 1 + print(f" FAIL {label}: raised {type(exc).__name__}: {exc}") + return None + if decision is None: + FAIL += 1 + print(f" FAIL {label}: returned None (no handler)") + return None + if not legal(obs, decision): + FAIL += 1 + allowed = sts2.LEGAL_ACTIONS.get(obs.get("state_type"), ()) + print(f" FAIL {label}: {decision.action!r} not legal for " + f"{obs.get('state_type')!r} (allowed: {allowed})") + return decision + PASS += 1 + print(f" ok {label}: {decision}") + return decision + + +def player(**kw) -> dict: + base = { + "character": "The Ironclad", "hp": 60, "max_hp": 80, "block": 0, + "gold": 150, "status": [], "relics": [], "potions": [], + "max_potion_slots": 3, + } + base.update(kw) + return base + + +def card(name="Strike", index=0, cost="1", desc="Deal 6 damage.", + ctype="Attack", target="AnyEnemy", can_play=True, rarity="Common", + upgraded=False) -> dict: + return { + "id": name.upper(), "name": name, "type": ctype, "cost": cost, + "star_cost": None, "description": desc, "rarity": rarity, + "is_upgraded": upgraded, "keywords": [], "index": index, + "target_type": target, "can_play": can_play, "unplayable_reason": None, + } + + +def enemy(entity_id="NIBBIT_0", hp=44, block=0, intent=12) -> dict: + intents = [] + if intent: + intents = [{"type": "Attack", "label": str(intent), "title": "Aggressive", + "description": f"This enemy intends to Attack for {intent} damage."}] + return {"entity_id": entity_id, "combat_id": 1, "name": entity_id.split("_")[0], + "hp": hp, "max_hp": hp, "block": block, "status": [], "intents": intents} + + +def combat(state_type="monster", hand=None, enemies=None, **pkw) -> dict: + return { + "state_type": state_type, + "battle": {"round": 1, "turn": "player", "is_play_phase": True, + "enemies": enemies if enemies is not None else [enemy()]}, + "run": {"act": 1, "floor": 5, "ascension": 0}, + "player": player(energy=3, max_energy=3, + hand=hand if hand is not None else [card()], + draw_pile=[], draw_pile_count=0, + discard_pile=[], discard_pile_count=0, + exhaust_pile=[], exhaust_pile_count=0, **pkw), + } + + +_reset_counter = 0 + + +def force_reset() -> None: + """Force a screen-group change so every per-screen guard is cleared. + + `_reset_screen_guards` only resets on a GROUP CHANGE, so calling it with + the same group twice is a deliberate no-op. Tests need a guaranteed reset. + """ + global _reset_counter + _reset_counter += 1 + brain._reset_screen_guards(f"__test{_reset_counter}__") + + +print("=== 1. every state_type produces a LEGAL action ===") + +cases: list[tuple[str, dict]] = [ + ("menu:main", {"state_type": "menu", "menu_screen": "main", + "options": ["singleplayer", "settings", "quit"]}), + ("menu:main+continue", {"state_type": "menu", "menu_screen": "main", + "options": ["continue", "abandon_run", "quit"]}), + ("menu:mode-select", {"state_type": "menu", "menu_screen": "singleplayer", + "options": [{"name": "standard", "enabled": True}]}), + ("menu:character-select", {"state_type": "menu", "menu_screen": "character_select", + "message": "Select a character.", + "options": [{"name": "IRONCLAD", "enabled": True}, + {"name": "embark", "enabled": True}]}), + ("menu:character-selected", {"state_type": "menu", "menu_screen": "character_select", + "message": "Selected The Ironclad. Use 'confirm' to embark.", + "options": [{"name": "IRONCLAD", "enabled": True}, + {"name": "embark", "enabled": True}]}), + ("menu:tutorial", {"state_type": "menu", "menu_screen": "tutorial_prompt", + "options": [{"name": "no", "enabled": True}]}), + ("game_over", {"state_type": "game_over", + "game_over": {"message": "Run ended.", "options": ["main_menu"]}}), + ("monster", combat("monster")), + ("elite", combat("elite")), + ("boss", combat("boss")), + ("combat:not-play-phase", + {**combat(), "battle": {"round": 1, "turn": "enemy", "is_play_phase": False, + "enemies": [enemy()]}}), + ("combat:no-playable", + combat(hand=[card(can_play=False)])), + ("hand_select:empty", {"state_type": "hand_select", + "hand_select": {"mode": "simple_select", + "prompt": "Choose any number of cards to replace.", + "cards": [], + "selected_cards": [{"index": 0, "name": "Defend"}], + "can_confirm": True}}), + ("hand_select:with-cards", {"state_type": "hand_select", + "hand_select": {"mode": "simple_select", + "prompt": "Choose cards to exhaust.", + "cards": [card("Strike", 0), card("Bash", 1)], + "selected_cards": [], + "can_confirm": False}}), + ("rewards:items", {"state_type": "rewards", + "rewards": {"items": [{"index": 0, "type": "gold", + "description": "19 Gold", "gold_amount": 19}], + "can_proceed": True}}), + ("rewards:empty", {"state_type": "rewards", + "rewards": {"items": [], "can_proceed": True}}), + ("card_reward", {"state_type": "card_reward", + "card_reward": {"cards": [card("Stomp", 0), card("Inflame", 1, rarity="Uncommon")], + "can_skip": True}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("map", {"state_type": "map", + "map": {"visited": [], "next_options": [ + {"index": 0, "col": 0, "row": 1, "type": "Monster"}, + {"index": 1, "col": 2, "row": 1, "type": "Rest"}], + "nodes": []}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("map:no-options", {"state_type": "map", "map": {"next_options": []}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("event", {"state_type": "event", + "event": {"in_dialogue": False, "body": "…", + "options": [{"index": 0, "title": "Stop", + "description": "Take what you have and leave.", + "is_locked": False, "is_proceed": False, + "was_chosen": False}]}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("event:dialogue", {"state_type": "event", + "event": {"in_dialogue": True, "body": "…", "options": []}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("event:all-locked", {"state_type": "event", + "event": {"in_dialogue": False, "body": "…", + "options": [{"index": 0, "title": "Locked", + "description": "", "is_locked": True, + "is_proceed": False, "was_chosen": False}]}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("rest_site", {"state_type": "rest_site", + "rest_site": {"options": [ + {"index": 0, "id": "REST", "name": "Rest", + "description": "Heal 24 HP.", "is_enabled": True}, + {"index": 1, "id": "SMITH", "name": "Smith", + "description": "Upgrade a card.", "is_enabled": True}], + "can_proceed": False}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("rest_site:empty", {"state_type": "rest_site", + "rest_site": {"options": [], "can_proceed": True}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("shop", {"state_type": "shop", + "shop": {"items": [{"index": 0, "category": "relic", "price": 150, + "is_stocked": True, "can_afford": True, + "relic_name": "Vajra", + "relic_description": "Start each combat with 1 Strength."}], + "can_proceed": False}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("shop:nothing-affordable", + {"state_type": "shop", + "shop": {"items": [{"index": 0, "category": "relic", "price": 900, + "is_stocked": True, "can_afford": False, + "relic_name": "X", "relic_description": "y"}], + "can_proceed": False}, + "run": {"act": 1, "floor": 3}, "player": player()}), + # fake_merchant nests its shop one level deeper. This case exists because + # reading only obs["shop"] made the bot leave every fake merchant instantly. + ("fake_merchant", {"state_type": "fake_merchant", + "fake_merchant": {"event_id": "FAKE_MERCHANT", + "started_fight": False, + "shop": {"items": [ + {"index": 0, "category": "potion", + "price": 60, "is_stocked": True, + "can_afford": True, + "potion_name": "Fire Potion", + "potion_description": "Deal 20 damage."}], + "can_proceed": False}}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("treasure:opening", {"state_type": "treasure", + "treasure": {"message": "Opening chest..."}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("treasure:relics", {"state_type": "treasure", + "treasure": {"relics": [{"index": 0, "id": "NUNCHAKU", + "name": "Nunchaku", "rarity": "Common", + "description": "..."}], + "can_proceed": True}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("card_select:grid", {"state_type": "card_select", + "card_select": {"screen_type": "upgrade", + "prompt": "Choose a card to Upgrade.", + "cards": [card("Strike", 0), card("Bash", 1)], + "preview_showing": False, "can_confirm": False}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("card_select:preview", {"state_type": "card_select", + "card_select": {"screen_type": "upgrade", + "prompt": "Choose a card to Upgrade.", + "cards": [card("Strike", 0)], + "preview_showing": True, "can_confirm": True}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("card_select:empty", {"state_type": "card_select", + "card_select": {"screen_type": "upgrade", "cards": [], + "preview_showing": False, "can_confirm": False}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("bundle_select", {"state_type": "bundle_select", + "bundle_select": {"screen_type": "bundle", "prompt": "Choose a bundle.", + "bundles": [ + {"index": 0, "card_count": 3, + "cards": [card("Anger", 0)]}, + {"index": 1, "card_count": 3, + "cards": [card("Pommel Strike", 0)]}], + "preview_showing": False, "can_confirm": False}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("bundle_select:preview", {"state_type": "bundle_select", + "bundle_select": {"screen_type": "bundle", + "bundles": [{"index": 0, "cards": []}], + "preview_showing": True, "can_confirm": True}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("relic_select", {"state_type": "relic_select", + "relic_select": {"prompt": "Choose a relic.", + "relics": [{"index": 0, "id": "A", "name": "Vajra", + "rarity": "Common", "description": "x"}, + {"index": 1, "id": "B", "name": "Lantern", + "rarity": "Uncommon", "description": "y"}], + "can_skip": True}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("crystal_sphere", {"state_type": "crystal_sphere", + "crystal_sphere": {"grid_width": 3, "grid_height": 3, + "instructions_title": "Divine"}, + "run": {"act": 1, "floor": 3}, "player": player()}), + ("unknown", {"state_type": "unknown", "room_type": None}), + ("overlay", {"state_type": "overlay"}), +] + +for label, obs in cases: + expect_legal(label, obs) + +print() +print("=== 2. structural invariants ===") + +# Every documented state_type must have a handler. +handled = {c[1]["state_type"] for c in cases} +unhandled = [st for st in sts2.LEGAL_ACTIONS if st not in handled] +check("every state_type covered by a test case", unhandled, []) + +# Every action the decision layer can emit must exist in LEGAL_ACTIONS. +emitted = set() +for _, obs in cases: + d = brain.decide(obs, None) + if d: + emitted.add(d.action) +declared = {a for actions in sts2.LEGAL_ACTIONS.values() for a in actions} +check("every emitted action is declared somewhere", sorted(emitted - declared - INTERNAL), []) + +print() +print("=== 3. fallbacks must respect their inputs ===") + +# Removal must not target an upgraded card while a plain Strike exists. +from_brain = brain.card_select_fallback( + [card("Strike+", 0, upgraded=True), card("Strike", 1), card("Bash", 2)], + "remove", +) +check("remove prefers an un-upgraded Strike", from_brain["name"], "Strike") + +# Upgrade must not target a basic card while a real card exists. +from_brain = brain.card_select_fallback( + [card("Strike", 0), card("Demon Form", 1, rarity="Rare")], "upgrade" +) +check("upgrade prefers a non-basic card", from_brain["name"], "Demon Form") + +# Upgrade must never target an already-upgraded card. +from_brain = brain.card_select_fallback( + [card("Strike", 0, upgraded=True), card("Defend", 1)], "upgrade" +) +check("upgrade skips an already-upgraded card", from_brain["name"], "Defend") + +# Card reward fallback must prefer rarity, not index 0. +from_brain = brain.best_by_rarity( + [card("Common A", 0), card("Rare B", 1, rarity="Rare"), card("Uncommon C", 2, rarity="Uncommon")] +) +check("card reward fallback takes the rarest", from_brain["name"], "Rare B") + +# hand_select must not give up a good card when a Strike is available. +d = brain.decide( + {"state_type": "hand_select", + "hand_select": {"mode": "simple_select", "prompt": "Choose cards to exhaust.", + "cards": [card("Bash", 0), card("Strike", 1)], + "selected_cards": [], "can_confirm": False}}, + None, +) +check("hand_select gives up the Strike, not Bash", d.params.get("card_index"), 1) + +# ...and must stop rather than exhaust the whole deck. +d = brain.decide( + {"state_type": "hand_select", + "hand_select": {"mode": "simple_select", "prompt": "Choose any number.", + "cards": [card("Bash", 0), card("Uppercut", 1)], + "selected_cards": [{"index": 0, "name": "Strike"}], + "can_confirm": True}}, + None, +) +check("hand_select keeps good cards and confirms", d.action, "combat_confirm_selection") + +# fake_merchant nests its inventory at fake_merchant.shop. Prove it is found: +# with no Jev, the reason differs depending on whether an affordable item parsed. +def fake_merchant(can_afford: bool) -> dict: + return { + "state_type": "fake_merchant", + "fake_merchant": { + "event_id": "FAKE_MERCHANT", "started_fight": False, + "shop": {"items": [{"index": 0, "category": "potion", "price": 60, + "is_stocked": True, "can_afford": can_afford, + "potion_name": "Fire Potion", + "potion_description": "Deal 20 damage."}], + "can_proceed": False}, + }, + "run": {"act": 1, "floor": 3}, "player": player(), + } + + +d = brain.decide(fake_merchant(True), None) +check("fake_merchant finds its nested shop", "no jev" in d.reason, True) + +d = brain.decide(fake_merchant(False), None) +check("fake_merchant reports nothing affordable", + "nothing affordable" in d.reason, True) + +# A skipped card reward is NOT consumed by the game. Without the guard the bot +# loops claim_reward -> card screen -> skip -> claim_reward forever. +force_reset() +rewards_with_card = { + "state_type": "rewards", + "rewards": {"items": [{"index": 0, "type": "gold", "description": "18 Gold", + "gold_amount": 18}, + {"index": 1, "type": "card", + "description": "Add a card to your deck."}], + "can_proceed": True}, + "run": {"act": 1, "floor": 2}, "player": player(), +} + +d = brain.decide(rewards_with_card, None) +check("rewards claims the last item first", d.params.get("index"), 1) + +brain._rewards_skipped_card = True +d = brain.decide(rewards_with_card, None) +check("a skipped card reward is not re-claimed", d.action, "claim_reward") +check("...and the gold is taken instead", d.params.get("index"), 0) + +brain._rewards_skipped_card = True +d = brain.decide( + {"state_type": "rewards", + "rewards": {"items": [{"index": 0, "type": "card", + "description": "Add a card to your deck."}], + "can_proceed": True}, + "run": {"act": 1, "floor": 2}, "player": player()}, + None, +) +check("rewards with only a skipped card proceeds", d.action, "proceed") + +# rewards -> card_reward must stay in ONE screen group, or the skip flag is +# cleared on every hop and the loop returns. +brain._reset_screen_guards("rewards") +brain._rewards_skipped_card = True +brain._reset_screen_guards("card_reward") +check("skip flag survives rewards <-> card_reward", brain._rewards_skipped_card, True) +brain._reset_screen_guards("monster") +check("skip flag clears when leaving the flow", brain._rewards_skipped_card, False) + +class StubClient: + """ + Deterministic stand-in for JevClient so the model paths can be tested + offline. Returns a high Noul for every ranking question and the first + option for every Choice, unless overridden. + """ + + def __init__(self, noul: float = 0.90, choice_override: dict | None = None, + noul_override: dict | None = None): + self.model = "stub" + self._noul = noul + self._choice = choice_override or {} + self._noul_override = noul_override or {} + + def __repr__(self) -> str: + return "StubClient()" + + def ask(self, state, questions, model=None): + from jev import ChoiceAnswer, NoulAnswer, JevResponse + + answers = {} + for qid, q in questions.items(): + if q.get("type") == "choice": + options = list((q.get("criteria") or {}).keys()) + picked = self._choice.get(qid, options[0] if options else "") + answers[qid] = ChoiceAnswer( + choice=picked, + probabilities={o: (0.9 if o == picked else 0.02) for o in options}, + confidence=0.9, + ) + elif q.get("type") == "noul": + answers[qid] = NoulAnswer( + noul=self._noul_override.get(qid, self._noul) + ) + else: + from jev import ScoreAnswer + answers[qid] = ScoreAnswer(score=1.0, confidence=0.9) + return JevResponse(answers=answers, model="stub", input_tokens=0, + output_tokens=0, latency_s=0.0) + + +print() +print("=== 4. indices must come from the data, never from list position ===") + +# Every indexed action is given cards whose `index` field does NOT match their +# list position. Any code that selects by array position will emit 0/1/2 and +# fail here. This is the regression test for the bug that made the bot upgrade +# only Strikes: it chose cards[0] rather than reading the card's identity. +OFFSET = 5 + +def offset_card(name, pos, **kw) -> dict: + return card(name, index=OFFSET + pos, **kw) + +valid_indices = {OFFSET, OFFSET + 1, OFFSET + 2} +stub = StubClient() + +# card_select (upgrade) +force_reset() +d = brain.decide( + {"state_type": "card_select", + "card_select": {"screen_type": "upgrade", "prompt": "Choose a card to Upgrade.", + "cards": [offset_card("Strike", 0), offset_card("Defend", 1), + offset_card("Demon Form", 2, rarity="Rare")], + "preview_showing": False, "can_confirm": False}, + "run": {"act": 1, "floor": 3}, "player": player()}, + stub, +) +check("card_select emits the card's own index", d.params.get("index") in valid_indices, True) +check("...and picks by identity, not position", + d.params.get("index"), OFFSET + 2) + +# card_reward with the stub. `skip_all` is answered NO so the take path is +# exercised; otherwise a stub that says yes to everything always skips. +take_stub = StubClient(noul_override={"skip_all": 0.10}) +force_reset() +d = brain.decide( + {"state_type": "card_reward", + "card_reward": {"cards": [offset_card("Common A", 0), + offset_card("Rare B", 1, rarity="Rare")], + "can_skip": True}, + "run": {"act": 1, "floor": 3}, "player": player()}, + take_stub, +) +check("card_reward emits the card's own index", d.params.get("card_index") in valid_indices, True) + +# And with `skip_all` answered YES, Jev's skip is honoured. +skip_stub = StubClient(noul_override={"skip_all": 0.95}) +force_reset() +d = brain.decide( + {"state_type": "card_reward", + "card_reward": {"cards": [offset_card("Common A", 0)], "can_skip": True}, + "run": {"act": 1, "floor": 3}, "player": player()}, + skip_stub, +) +check("jev can decide to skip", d.action, "skip_card_reward") +check("...and the reason cites jev", "jev:" in d.reason, True) + +# An UNCERTAIN skip answer must not skip -- take the best card instead. +unsure_stub = StubClient(noul_override={"skip_all": 0.55}) +force_reset() +d = brain.decide( + {"state_type": "card_reward", + "card_reward": {"cards": [offset_card("Common A", 0)], "can_skip": True}, + "run": {"act": 1, "floor": 3}, "player": player()}, + unsure_stub, +) +check("an uncertain skip answer takes a card", d.action, "select_card_reward") + +# `can_skip: false` must take a card even if Jev would skip. +force_reset() +d = brain.decide( + {"state_type": "card_reward", + "card_reward": {"cards": [offset_card("Common A", 0)], "can_skip": False}, + "run": {"act": 1, "floor": 3}, "player": player()}, + skip_stub, +) +check("cannot skip -> takes a card anyway", d.action, "select_card_reward") + +# hand_select (in-combat selectable subset) +force_reset() +d = brain.decide( + {"state_type": "hand_select", + "hand_select": {"mode": "simple_select", "prompt": "Choose cards to exhaust.", + "cards": [offset_card("Bash", 0), offset_card("Strike", 1)], + "selected_cards": [], "can_confirm": False}, + "run": {"act": 1, "floor": 3}, "player": player()}, + None, +) +check("hand_select emits the selectable card's own index", + d.params.get("card_index"), OFFSET + 1) + +# rewards +force_reset() +d = brain.decide( + {"state_type": "rewards", + "rewards": {"items": [{"index": OFFSET, "type": "gold", "description": "1 Gold"}, + {"index": OFFSET + 1, "type": "gold", "description": "2 Gold"}], + "can_proceed": True}, + "run": {"act": 1, "floor": 3}, "player": player()}, + None, +) +check("rewards emits the item's own index", d.params.get("index"), OFFSET + 1) + +# shop +force_reset() +d = brain.decide( + {"state_type": "shop", + "shop": {"items": [{"index": OFFSET, "category": "relic", "price": 10, + "is_stocked": True, "can_afford": True, + "relic_name": "Vajra", "relic_description": "x"}, + {"index": OFFSET + 1, "category": "relic", "price": 10, + "is_stocked": True, "can_afford": True, + "relic_name": "Lantern", "relic_description": "y"}], + "can_proceed": False}, + "run": {"act": 1, "floor": 3}, "player": player()}, + stub, +) +check("shop emits the item's own index", d.params.get("index") in valid_indices, True) + +# map +force_reset() +d = brain.decide( + {"state_type": "map", + "map": {"next_options": [{"index": OFFSET, "col": 0, "row": 1, "type": "Monster"}, + {"index": OFFSET + 1, "col": 2, "row": 1, "type": "Elite"}], + "nodes": []}, + "run": {"act": 1, "floor": 3}, "player": player()}, + stub, +) +check("map emits the node's own index", d.params.get("index") in valid_indices, True) + +# bundle_select +force_reset() +d = brain.decide( + {"state_type": "bundle_select", + "bundle_select": {"screen_type": "bundle", "prompt": "Choose a bundle.", + "bundles": [{"index": OFFSET, "cards": [card("A", 0)]}, + {"index": OFFSET + 1, "cards": [card("B", 0)]}], + "preview_showing": False, "can_confirm": False}, + "run": {"act": 1, "floor": 3}, "player": player()}, + stub, +) +check("bundle_select emits the bundle's own index", d.params.get("index") in valid_indices, True) + +# relic_select +force_reset() +d = brain.decide( + {"state_type": "relic_select", + "relic_select": {"prompt": "Choose a relic.", + "relics": [{"index": OFFSET, "name": "Vajra", "rarity": "Common", + "description": "x"}, + {"index": OFFSET + 1, "name": "Lantern", + "rarity": "Uncommon", "description": "y"}], + "can_skip": True}, + "run": {"act": 1, "floor": 3}, "player": player()}, + stub, +) +check("relic_select emits the relic's own index", d.params.get("index") in valid_indices, True) + +# A state whose list positions are ALL zero-ish while the real indices are high: +# any position-based code returns 0, which is not a valid index here. +force_reset() +d = brain.decide( + {"state_type": "card_select", + "card_select": {"screen_type": "remove", "prompt": "Choose a card to Remove.", + "cards": [offset_card("Bash", 0), offset_card("Strike", 1)], + "preview_showing": False, "can_confirm": False}, + "run": {"act": 1, "floor": 3}, "player": player()}, + None, +) +check("a position-based pick of 0 would be invalid here", + d.params.get("index") != 0, True) + +print() +print("=== 5. character select needs a select-then-embark sequence ===") + +# The state carries NO "selected" indicator -- the mod hardcodes `message` to +# "Select a character." whatever is chosen. So the handler selects once and +# embarks when the screen comes back unchanged. Without this the bot selects +# the already-selected character forever and stalls. +char_select_state = { + "state_type": "menu", "menu_screen": "character_select", + "message": "Select a character.", + "options": [{"name": "IRONCLAD", "enabled": True}, + {"name": "SILENT", "enabled": True}, + {"name": "confirm", "enabled": True}, + {"name": "embark", "enabled": True}, + {"name": "back", "enabled": True}], +} + +force_reset() +d1 = brain.decide(char_select_state, None) +check("character_select first selects a character", d1.params.get("option"), "IRONCLAD") + +# Same screen unchanged -> a character is already selected -> embark. +d2 = brain.decide(char_select_state, None) +check("character_select then embarks", d2.params.get("option"), "embark") + +# The sequence ALTERNATES, which is what makes it self-correcting: a rejected +# embark ("select a character first") is always followed by a fresh select. +d2b = brain.decide(char_select_state, None) +check("a third call re-selects", d2b.params.get("option"), "IRONCLAD") +d2c = brain.decide(char_select_state, None) +check("and then embarks again", d2c.params.get("option"), "embark") + +# A changing signature must not derail the alternation. +changed = dict(char_select_state) +changed["characters"] = [{"name": "The Ironclad", "id": "IRONCLAD", "locked": False}] +d2d = brain.decide(changed, None) +check("a changing signature does not derail the cycle", + d2d.params.get("option"), "IRONCLAD") + +# Both actions are legal for this screen. +check("select is legal", legal(char_select_state, d1), True) +check("embark is legal", legal(char_select_state, d2), True) + +# Leaving the screen must clear the guard, or the next run embarks immediately +# without selecting anything. +brain._reset_screen_guards("menu", "main") +d3 = brain.decide(char_select_state, None) +check("guard resets when leaving character select", d3.params.get("option"), "IRONCLAD") + +# The alternation must also reset, or a fresh run embarks without selecting. +brain._reset_screen_guards("menu", "main") +d3b = brain.decide(char_select_state, None) +check("alternation resets too", d3b.params.get("option"), "IRONCLAD") + +# menu_screen must be part of the screen key: main menu and character select are +# both state_type "menu" but have different valid actions. +brain._reset_screen_guards("menu", "main") +check("main menu and character select are different screens", + brain._screen_group("menu", "main") != brain._screen_group("menu", "character_select"), + True) + +# combat_select_card TOGGLES: re-selecting an already-chosen card deselects it, +# so the state never changes and the loop stalls. `selected_cards` is the +# authoritative list of what is already chosen. +d = brain.decide( + {"state_type": "hand_select", + "hand_select": {"mode": "simple_select", "prompt": "Choose any number.", + "cards": [card("Defend", 0), card("Defend", 1), card("Strike", 2)], + "selected_cards": [{"index": 0, "name": "Defend"}], + "can_confirm": True}}, + None, +) +check("hand_select skips an already-chosen card", d.params.get("card_index"), 2) + +# Every selectable card already chosen -> confirm rather than toggle forever. +d = brain.decide( + {"state_type": "hand_select", + "hand_select": {"mode": "simple_select", "prompt": "Choose any number.", + "cards": [card("Defend", 0)], + "selected_cards": [{"index": 0, "name": "Defend"}], + "can_confirm": True}}, + None, +) +check("hand_select confirms when all are chosen", d.action, "combat_confirm_selection") + +# Duplicate names: two Defends, one already chosen, one still selectable. +d = brain.decide( + {"state_type": "hand_select", + "hand_select": {"mode": "simple_select", "prompt": "Choose any number.", + "cards": [card("Defend", 0), card("Defend", 1)], + "selected_cards": [{"index": 0, "name": "Defend"}], + "can_confirm": True}}, + None, +) +check("hand_select still offers the second Defend", d.params.get("card_index"), 1) + +# `upgrade_select` pre-selects the card; selecting is a no-op and only the +# confirm makes progress. Measured live: cards=[Defend], can_confirm=true, and +# combat_select_card changed nothing across 18 attempts. +d = brain.decide( + {"state_type": "hand_select", + "hand_select": {"mode": "upgrade_select", "prompt": "Confirm Card to Upgrade", + "cards": [card("Defend", 0)], "selected_cards": [], + "can_confirm": True}}, + None, +) +check("upgrade_select confirms instead of selecting", d.action, "combat_confirm_selection") + +# No enemies but combat still in progress (Phrog Parasite spawning the next +# wave). Measured live: the bot emitted play_card(card_index=1) with no target +# and the game rejected it with "Card requires a target". +force_reset() +obs_no_enemies = combat(enemies=[]) +d = brain.decide(obs_no_enemies, None) +check("no enemies -> wait, do not play", d.action, "__wait__") + +# A target-requiring card must never be emitted without a target. +force_reset() +obs_one_enemy = combat(hand=[card("Strike", 0)], enemies=[enemy()]) +d = brain.decide(obs_one_enemy, None) +check("a single enemy is auto-targeted", d.params.get("target"), "NIBBIT_0") + +force_reset() +d = brain.decide(obs_no_enemies, StubClient()) +check("even with jev, no enemies means wait", d.action, "__wait__") + +# The enchant screen: screen_type is the RAW C# class name, select_card toggles +# with NO preview, and can_confirm is true from the start. Measured live: the +# handler re-selected index 10 forever because preview_showing stayed false. +check("raw class name normalises to upgrade", + brain.screen_kind("NDeckEnchantSelectScreen"), "upgrade") +check("friendly names still work", + (brain.screen_kind("remove"), brain.screen_kind("transform")), ("remove", "transform")) + +enchant_state = { + "state_type": "card_select", + "card_select": {"screen_type": "NDeckEnchantSelectScreen", + "prompt": "Choose a card to Enchant.", + "cards": [offset_card("Strike", 0), offset_card("Perfected Strike", 1)], + "preview_showing": False, + "can_cancel": False, "can_confirm": True}, + "run": {"act": 1, "floor": 13}, "player": player(), +} + +force_reset() +d1 = brain.decide(enchant_state, None) +check("enchant screen selects first", d1.action, "select_card") + +d2 = brain.decide(enchant_state, None) +check("enchant screen then confirms", d2.action, "confirm_selection") + +# The full flow must terminate: select, confirm, done. +force_reset() +seq = [brain.decide(enchant_state, None).action, brain.decide(enchant_state, None).action] +check("enchant flow terminates", seq, ["select_card", "confirm_selection"]) + +# MULTI-select: "Choose 5 cards to Remove" keeps can_confirm FALSE until all +# five are picked, and card_select has no `selected_cards` field. Measured live: +# the handler re-selected the same index forever and stalled. +check("the required count is parsed from the prompt", + brain.card_select_need("Choose 5 cards to Remove."), 5) +check("single-select defaults to 1", + brain.card_select_need("Choose a card to Upgrade."), 1) + +multi_state = { + "state_type": "card_select", + "card_select": {"screen_type": "select", "prompt": "Choose 5 cards to Remove.", + "cards": [offset_card(f"Card{i}", i) for i in range(8)], + "preview_showing": False, + "can_cancel": False, "can_confirm": False}, + "run": {"act": 2, "floor": 18}, "player": player(), +} + +force_reset() +seq = [brain.decide(multi_state, None).action for _ in range(6)] +check("multi-select picks 5 then waits", + seq, ["select_card"] * 5 + ["__wait__"]) + +# Every pick must be a DIFFERENT index, or select_card toggles it back off. +force_reset() +idxs = [brain.decide(multi_state, None).params.get("index") for _ in range(5)] +check("each pick targets a different index", len(set(idxs)), 5) + +# Once can_confirm turns true, it confirms. +force_reset() +for _ in range(5): + brain.decide(multi_state, None) +confirmable = dict(multi_state) +confirmable["card_select"] = dict(multi_state["card_select"], can_confirm=True) +d = brain.decide(confirmable, None) +check("multi-select confirms once the count is met", d.action, "confirm_selection") + +# Crystal Sphere: can_proceed is FALSE until tiles are revealed. The old +# unconditional crystal_sphere_proceed was rejected and stalled the run. +force_reset() +cs_state = { + "state_type": "crystal_sphere", + "crystal_sphere": {"grid_width": 11, "grid_height": 11, + "cells": [], "revealed_items": [], "tool": "big", + "clickable_cells": [{"x": 5, "y": 5}, {"x": 0, "y": 0}], + "can_proceed": False}, + "run": {"act": 2, "floor": 25}, "player": player(), +} +d = brain.decide(cs_state, None) +check("crystal sphere reveals a cell first", d.action, "crystal_sphere_click_cell") +check("...starting from the centre", (d.params.get("x"), d.params.get("y")), (5, 5)) + +# It must not re-click the same cell. +d2 = brain.decide(cs_state, None) +check("crystal sphere does not repeat a cell", + (d2.params.get("x"), d2.params.get("y")) != (d.params.get("x"), d.params.get("y")), True) + +# Once can_proceed unlocks, it proceeds. +force_reset() +cs_done = dict(cs_state) +cs_done["crystal_sphere"] = dict(cs_state["crystal_sphere"], can_proceed=True) +d3 = brain.decide(cs_done, None) +check("crystal sphere proceeds when unlocked", d3.action, "crystal_sphere_proceed") + +# "Choose a card to Exhaust." is mode simple_select with can_confirm TRUE and +# one card already chosen. Measured live: the handler only confirmed for +# upgrade_select, so it toggled between two cards forever. +check("a singular prompt needs 1", brain.hand_select_need("Choose a card to Exhaust."), 1) +check("a counted prompt needs N", brain.hand_select_need("Choose 2 cards to discard."), 2) +check("\"any number\" is unbounded", brain.hand_select_need("Choose any number of cards to replace."), None) + +force_reset() +d = brain.decide( + {"state_type": "hand_select", + "hand_select": {"mode": "simple_select", "prompt": "Choose a card to Exhaust.", + "cards": [card("Defend", 0)], + "selected_cards": [{"index": 0, "name": "Strike"}], + "can_confirm": True}}, + None, +) +check("a satisfied singular prompt confirms", d.action, "combat_confirm_selection") + +# An unsatisfied counted prompt keeps selecting. +force_reset() +d = brain.decide( + {"state_type": "hand_select", + "hand_select": {"mode": "simple_select", "prompt": "Choose 2 cards to discard.", + "cards": [card("Strike", 0), card("Defend", 1)], + "selected_cards": [], + "can_confirm": True}}, + None, +) +check("an unsatisfied counted prompt selects", d.action, "combat_select_card") + +# The `combined` policy: Jev declines to skip, but if nothing it rated clears +# the floor, stay lean anyway. This is the safety net against deck bloat. +_saved_policy = brain.CARD_SKIP_POLICY +try: + weak_stub = StubClient(noul=0.50, noul_override={"skip_all": 0.10}) + + brain.CARD_SKIP_POLICY = "jev" + force_reset() + d = brain.decide( + {"state_type": "card_reward", + "card_reward": {"cards": [offset_card("Common A", 0)], "can_skip": True}, + "run": {"act": 1, "floor": 3}, "player": player()}, + weak_stub, + ) + check("jev policy takes a weak card", d.action, "select_card_reward") + + brain.CARD_SKIP_POLICY = "combined" + force_reset() + d = brain.decide( + {"state_type": "card_reward", + "card_reward": {"cards": [offset_card("Common A", 0)], "can_skip": True}, + "run": {"act": 1, "floor": 3}, "player": player()}, + weak_stub, + ) + check("combined policy skips a weak card", d.action, "skip_card_reward") + check("...and says why", "combined:" in d.reason, True) + + # A strong card is taken under BOTH policies. + strong_stub = StubClient(noul=0.80, noul_override={"skip_all": 0.10}) + for policy in ("jev", "combined"): + brain.CARD_SKIP_POLICY = policy + force_reset() + d = brain.decide( + {"state_type": "card_reward", + "card_reward": {"cards": [offset_card("Strong", 0)], "can_skip": True}, + "run": {"act": 1, "floor": 3}, "player": player()}, + strong_stub, + ) + check(f"{policy} takes a strong card", d.action, "select_card_reward") +finally: + brain.CARD_SKIP_POLICY = _saved_policy + +# Preflight must catch states the bot cannot leave, so a whole session does not +# silently burn its step budget. This actually happened: an A/B arm ran with 0 +# decisions because the Timeline epoch was pending. +import run as run_module + +check("preflight passes a normal main menu", + run_module.preflight({"state_type": "menu", "menu_screen": "main", + "options": ["singleplayer", "settings", "quit"]}), None) + +blocked = run_module.preflight({ + "state_type": "menu", "menu_screen": "main", + "options": ["settings", "quit"], + "blocked_options": [{"name": "timeline", "enabled": False, + "reason": "manual_epoch_reveal_required", + "pending_epoch_ids": ["IRONCLAD5_EPOCH"]}], +}) +check("preflight catches the timeline blocker", bool(blocked), True) +check("...and names the pending epoch", "IRONCLAD5_EPOCH" in (blocked or ""), True) + +check("preflight ignores non-menu states", + run_module.preflight({"state_type": "monster"}), None) + +print() +print(f"=== {PASS} passed, {FAIL} failed ===") +sys.exit(1 if FAIL else 0)