diff --git a/AGENTS.md b/AGENTS.md index b0513a5..427aaef 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -2,11 +2,13 @@ ## Code map - `facts.py`: pure game-state parsing and arithmetic. -- `brain.py`: decision policy and session-owned `PolicyContext`; proposes one action per observation. +- `brain.py`: policy entry point; dispatch, navigation, shops, and minigames. +- `policy/combat.py` and `policy/selection.py`: combat and selection proposals. +- `policy/context.py`: shared `Decision`, pending actions, and session-owned `PolicyContext`. - `run.py`: observe–decide–act loop, captures, and session attribution. - `sts2.py`: local game HTTP client. `jev.py`: TypeSafe model client and gates. - `migrate.py`: dataset migration and integrity checks. -- `CONTEXT.md`: domain terms. `docs/DATASET.md`: dataset schema and limits. +- `CONTEXT.md`: domain terms. `docs/POLICY.md`: policy boundaries. `docs/DATASET.md`: dataset schema and limits. - `docs/research/11-prototype-hardening.md`: audited findings and the proposed development order. - Treat historical claims in `docs/DESIGN.md` as context; verify against current code. diff --git a/brain.py b/brain.py index 242f91e..72602f8 100644 --- a/brain.py +++ b/brain.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 """ -brain.py -- the decision layer. +brain.py -- policy dispatch and the remaining navigation, shop, and minigame flows. Precedence, and why it is ordered this way: @@ -19,13 +19,10 @@ so the loop must re-observe after each action. from __future__ import annotations -import json -import re -from dataclasses import dataclass, field -from typing import Any - import facts as F from jev import JevClient, ChoiceAnswer, NoulAnswer, choice, gate, noul +from policy import combat, selection +from policy.context import Decision, PolicyContext # 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. @@ -55,543 +52,6 @@ def event_safety_rank(option: dict) -> int: 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 _lethal_line(playable: list[dict], energy: int, player_status: list, - enemy: F.EnemyFact) -> list[dict] | None: - """Use the same supported damage calculation as the fact layer.""" - return F.lethal_line(playable, energy, player_status, enemy) - - -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 in a SHORT fight, front-loading damage is better: - # HP is a resource. (STS2MCP strategy notes: "HP is a resource, not a - # score", "Front-load damage".) - # - # The old test was `threat in (HEAVY, SEVERE, LETHAL) or hp in (WOUNDED, - # CRITICAL)`, which ignored any hit of <=12 at 80 max HP. That is how the - # bot lost 9 runs to THE_KIN_BOSS. `block_urgent` adds the fight-length - # view: a small hit that repeats for 10 turns is not a small hit. - blockers = F.best_block_line(playable, f.energy) - if blockers and f.block_urgent: - best = blockers[0] - 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 _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 _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 {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 - }, - ) - - # NOTE: there used to be a `should_defend` Noul here. It was asked on every - # combat turn and never read -- `grep -rn should_defend` returned only the - # line that created it -- so it cost latency and did nothing. The defense - # decision is now made in code before Jev is consulted (see - # `combat_decision`), and the fight-length facts it needs are in - # `combat.fight_is_grinding` / `combat.this_turn_is_dangerous`. - - 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. A supported direct-damage line against one enemy. - for enemy in f.enemies: - line = _lethal_line(f.playable, f.energy, f.player_status, enemy) - if line: - card = line[0] - params = _target_params(card, f, force_target=enemy.entity_id) - return Decision( - "play_card", params, - f"direct-damage line on {enemy.entity_id} ({len(line)} cards)", "code", - ) - - # 2. Defense, decided in CODE and taken before Jev is asked. - # - # Measured, this was the single biggest hole. `THE_KIN_BOSS` ended 9 of 37 - # runs; those fights lasted 5-10 turns and cost 44-80 HP, ~10-13 a turn, - # with block cards in hand the whole time. Both decision paths preferred - # damage: the fallback classed a 12-damage hit at 80 max HP as "chip", and - # Jev, asked "which play best advances winning this fight?", chose Bash at - # 0.42 confidence. - # - # The fight-duration projection is a heuristic, not a known future. - # Use it only while there is a current block deficit. Start a maximum-block - # plan instead of greedily choosing the largest individual block card. - if f.block_urgent: - blockers = F.best_block_line(f.playable, f.energy) - if blockers: - best = blockers[0] - return Decision( - "play_card", - _target_params(best, f), - f"defense forced: {f.projected_incoming} projected vs " - f"{f.affordable_loss} affordable ({f.turns_to_kill} turns)", - "code", - ) - - # 3. Jev for preference, 4. heuristic if it is unsure. - if client is not None: - # The runner owns the failure budget and the model-free retry. - return _jev_combat(f, client) - 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": F.deck_context(deck), - "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) - ): - 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: - 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: - 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) - - # The questions are keyed `good_relicN`, so the ranking MUST read the - # prefixed ids. Reading `relicN` found nothing, `best_by_noul` returned - # (None, 0.0) for every state, and this path could only ever take the - # rarest relic -- every answer Jev gave was silently dropped. - best_key, best_noul = best_by_noul(response, [f"good_{k}" for k in 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"good_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.", @@ -687,214 +147,6 @@ def map_decision(obs: dict, client: JevClient | None, deck: dict | None) -> Deci f"jev chose {node.get('type')}", "jev", pick.confidence) -# 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, add. - - Three traps, all 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"/"simple_select" while the - PROMPT says what is actually happening. Measured: "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. - * "Choose 2 Common Cards to Add to Your Deck." is neither: it is ADDING. - - 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" - if "add to your deck" in text or "add to your deck" in text: - return "add" - if "add" in text and "deck" in text: - return "add" - # 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) - if kind == "add": - # Adding: take the rarest card on offer. - return max(cards, key=lambda c: RARITY_RANK.get(str(c.get("rarity")), 1)) - return min(cards, key=upgrade_rank) - - -def _card_select_pick(index: int, reason: str, source: str, - confidence: float | None = None) -> Decision: - """Propose a toggle. Only an accepted request enters policy memory.""" - 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. - - The pattern must not assume the word "cards" directly follows the number: - "Choose 2 Common Cards to Add to Your Deck." has "Common" in between, and a - stricter pattern silently returned 1 and stalled the screen. - """ - match = re.search(r"choose\s+(\d+)", 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, - context: PolicyContext) -> 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 can make `confirm_selection` report ok without a visible - change. The context waits for the screen to close instead of repeating - or cancelling an action whose outcome is still unknown. - """ - cs = obs.get("card_select") or {} - chosen = context.accepted_card_indices - - 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 cs.get("can_confirm") and (len(chosen) >= need or cs.get("preview_showing")): - return Decision("confirm_selection", {}, "confirm the selection", "code") - if len(chosen) >= need: - # Enough chosen but the game has not enabled confirm yet. Selecting more - # would overshoot, so wait. - return Decision("__wait__", {}, - f"{len(chosen)}/{need} accepted toggles; 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. Do not repeat an accepted request for this grid. - remaining = [c for c in cards if c.get("index") not in chosen] - if not remaining: - if cs.get("can_confirm"): - return Decision("confirm_selection", {}, "all selectable cards chosen", "code") - return Decision("__wait__", {}, - f"{len(chosen)}/{need} accepted toggles; 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 exclude indices with accepted toggle requests. - 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?" - ) - elif kind == "add": - questions[f"good_{key}"] = noul( - f"Would adding `cards.{key}` to this deck make it 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": F.deck_context(deck), - "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, @@ -1017,9 +269,6 @@ def rest_site_decision(obs: dict) -> Decision: # 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 @@ -1208,251 +457,6 @@ def treasure_decision(obs: dict, client: JevClient | None, # 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) - - -def rewards_decision(obs: dict, context: PolicyContext) -> 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 context.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") - - -def bundle_select_decision(obs: dict, client: JevClient | None, - deck: dict | None) -> Decision: - """ - Bundle choice: pick one of several 3-card bundles. - - `select_bundle` errors when a preview is already open. Confirm the preview - instead. The context waits for evidence after either accepted request; - an unchanged read does not prove that confirmation failed. - """ - bs = obs.get("bundle_select") or {} - - if bs.get("preview_showing") and bs.get("can_confirm"): - 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": F.deck_context(deck), - "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") - - def crystal_sphere_decision(obs: dict, context: PolicyContext) -> Decision: """ Crystal Sphere minigame. @@ -1540,13 +544,13 @@ def simple_decision(obs: dict, client: JevClient | None = None, return Decision("menu_select", {"option": "main_menu"}, "run ended", "code") if st == "rewards": - return rewards_decision(obs, context) + return selection.rewards_decision(obs, context) if st == "card_reward": - return card_reward_decision(obs, client, deck) + return selection.card_reward_decision(obs, client, deck, skip_policy=CARD_SKIP_POLICY) if st == "relic_select": - return relic_select_decision(obs, client) + return selection.relic_select_decision(obs, client) if st == "map": return map_decision(obs, client, deck) @@ -1564,13 +568,13 @@ def simple_decision(obs: dict, client: JevClient | None = None, return shop_decision(obs, client, deck) if st == "hand_select": - return hand_select_decision(obs, client, deck) + return selection.hand_select_decision(obs, client, deck) if st == "card_select": - return card_select_decision(obs, client, deck, context) + return selection.card_select_decision(obs, client, deck, context) if st == "bundle_select": - return bundle_select_decision(obs, client, deck) + return selection.bundle_select_decision(obs, client, deck) if st == "crystal_sphere": return crystal_sphere_decision(obs, context) @@ -1583,136 +587,6 @@ def simple_decision(obs: dict, client: JevClient | None = None, return None -# 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) - - -@dataclass -class PendingAction: - """An accepted request, not proof that the game completed it.""" - - decision: Decision - screen: dict - - -@dataclass -class PolicyContext: - """Session-owned memory. Proposals never record execution. - - Grid toggles and character selection lack observable selection fields in - the mod. Their accepted requests are tracked explicitly, not called facts. - Other guarded actions wait for relevant screen evidence before continuing. - """ - - screen_group: str | None = None - accepted_card_indices: set[int] = field(default_factory=set) - accepted_card_grid: list[dict] | None = None - accepted_crystal_cells: set[tuple[int, int]] = field(default_factory=set) - rewards_skipped_card: bool = False - character_selected: bool = False - pending: PendingAction | None = None - - def observe(self, obs: dict) -> None: - st = obs.get("state_type") - if st in ("unknown", "overlay"): - return # A transient overlay is not evidence of completion. - group = _screen_group(st, obs.get("menu_screen")) - if group != self.screen_group: - self.screen_group = group - self.accepted_card_indices.clear() - self.accepted_card_grid = None - self.accepted_crystal_cells.clear() - self.rewards_skipped_card = False - self.character_selected = False - self.pending = None - return - pending = self.pending - if pending is None: - return - action, params = pending.decision.action, pending.decision.params - screen = obs.get(st) or {} - if action == "select_card" and pending.screen.get("screen_type") != "choose": - # The API exposes no selected indices. Reserve accepted toggles - # after a fresh read, even when their effect is not visible yet. - if screen.get("cards") != pending.screen.get("cards"): - return # Cannot map a toggle safely onto a changed grid. - self.accepted_card_indices.add(params["index"]) - self.accepted_card_grid = pending.screen.get("cards") - elif action == "menu_select" and params.get("option") != "embark": - self.character_selected = True - elif action == "skip_card_reward": - if st != "rewards": - return - self.rewards_skipped_card = True - elif action in ("confirm_selection", "cancel_selection", - "confirm_bundle_selection", "cancel_bundle_selection", - "combat_confirm_selection"): - return # Wait for the screen to close; do not cancel a slow confirm. - elif action == "crystal_sphere_click_cell": - cell = (params["x"], params["y"]) - clickable = {(c.get("x"), c.get("y")) - for c in screen.get("clickable_cells", [])} - if cell in clickable and not screen.get("can_proceed"): - return - self.accepted_crystal_cells.add(cell) - elif action == "shop_purchase": - # Gold alone can change for unrelated reasons. Require this item - # to change/disappear, or a screen transition (e.g. card removal). - before = pending.screen.get("shop", pending.screen) - after = screen.get("shop", screen) - index = params["index"] - old = next((i for i in before.get("items", []) if i.get("index") == index), None) - new = next((i for i in after.get("items", []) if i.get("index") == index), None) - def inventory_item(item): - return {k: v for k, v in item.items() if k not in ("can_afford", "price")} if item else None - if inventory_item(old) == inventory_item(new): - return - elif action == "combat_select_card": - if len(screen.get("selected_cards") or []) <= len(pending.screen.get("selected_cards") or []): - return - elif action == "select_bundle": - if not screen.get("preview_showing") or not screen.get("can_confirm"): - return - else: - return # Direct choices and embark require a screen transition. - self.pending = None - - def record_result(self, obs: dict, decision: Decision, *, accepted: bool) -> None: - """Called only after a real action result. Transport errors stop the runner.""" - if not accepted: - self.pending = None - if decision.action == "menu_select" and decision.params.get("option") == "embark": - self.character_selected = False - return - st = obs.get("state_type") - guarded = decision.action in { - "select_card", "confirm_selection", "cancel_selection", - "select_bundle", "confirm_bundle_selection", "cancel_bundle_selection", - "shop_purchase", "skip_card_reward", "crystal_sphere_click_cell", - "combat_select_card", "combat_confirm_selection", - } or (st == "menu" and obs.get("menu_screen") == "character_select") - if guarded: - # Detach from mutable fixtures/callers. This is a small screen node, - # not another copy of the entire combat observation. - screen = json.loads(json.dumps(obs.get(st) or {})) - self.pending = PendingAction(decision, screen) - - def decide(obs: dict, client: JevClient | None, deck: dict | None = None, *, context: PolicyContext | None = None) -> Decision | None: """Reconcile a fresh observation, then propose one action. @@ -1730,5 +604,5 @@ def decide(obs: dict, client: JevClient | None, deck: dict | None = None, and (obs.get("card_select") or {}).get("cards") != context.accepted_card_grid): return Decision("__wait__", {}, "card grid changed; accepted indices cannot be mapped safely", "code") if st in ("monster", "elite", "boss"): - return combat_decision(F.combat_facts(obs), client) + return combat.combat_decision(F.combat_facts(obs), client) return simple_decision(obs, client, deck, context=context) diff --git a/docs/POLICY.md b/docs/POLICY.md new file mode 100644 index 0000000..63d6139 --- /dev/null +++ b/docs/POLICY.md @@ -0,0 +1,66 @@ +# Policy structure + +## Current boundaries + +```text +run.py observe, execute, report results, record the session +brain.py entry point, dispatch, navigation, shops, minigames +policy/ + __init__.py package marker; no registration or initialization + context.py Decision, PendingAction, PolicyContext + combat.py combat proposals and model questions + selection.py card, relic, bundle, hand, and reward proposals +facts.py observation parsing and combat calculations +``` + +`brain.decide()` remains the policy entry point. The runner owns one +`PolicyContext` per session and reports action results to it. +`brain.Decision` and `brain.PolicyContext` remain available as direct imports +of the shared types, so the runner interface does not change. + +Policy modules may ask Jev for preferences. They must not call the game API. +`context.py` does not call either service. Combat arithmetic remains in +`facts.py`, not duplicated inside combat policy. + +The dependency direction is simple: + +- The runner uses `brain`. +- `brain` uses the combat, selection, and context modules. +- Combat and selection use context, facts, and the model client. +- Context uses only the Python standard library. +- No policy module imports `brain` or the runner. + +## State and configuration + +A proposal does not update execution memory. Accepted requests are reconciled +with fresh observations. See [policy-state behavior and limits](research/13-policy-state.md) +for the evidence required by each guarded action. + +The existing card-skip setting stays in `brain.CARD_SKIP_POLICY`. The dispatcher +passes its value explicitly to `selection.card_reward_decision(skip_policy=...)`. +This avoids a circular import and preserves the existing command-line behavior. +It is configuration, not per-screen action memory. + +## Why stop here + +These are plain modules, not a plugin framework or class hierarchy. Navigation, +shops, and minigames stay together until a further split helps development. + +The extraction does not change game decisions, confidence gates, fallbacks, +state reconciliation, or recording. Run identity and persistent-deck provenance +remain separate work. + +## Extraction verification + +- 262 existing assertions passed. No test files or assertions were added for the extraction. +- Thirteen isolated whole-process scenarios passed with local HTTP fixtures. +- All 346 stored observations produced identical decisions before and after extraction, + in both model-free mode and local-model-stub mode: 692 comparisons. +- The local model requests also matched, including their state and questions. +- Thirty-six function/class definitions retained identical abstract syntax trees. + Only the two dispatch functions and explicit card-skip parameter changed. +- Dataset integrity, Python compilation, and shell syntax checks passed. + +The differential and whole-process probes were temporary. They made no live +game or paid model calls. These checks establish refactoring equivalence for +the tested inputs, not correctness for every possible game state. diff --git a/docs/research/13-policy-state.md b/docs/research/13-policy-state.md index 14460d7..6742596 100644 --- a/docs/research/13-policy-state.md +++ b/docs/research/13-policy-state.md @@ -6,9 +6,9 @@ This pass replaces per-screen globals with explicit session memory. It follows [the correctness pass](12-correctness-pass.md). It does not redesign recording, identify runs, or establish persistent-deck provenance. -`brain.py` remains one file. Moving policy code and changing its state behavior -at the same time would make failures harder to diagnose. A later extraction -can preserve the interface and the runner-level tests. +This state pass kept `brain.py` as one file. Moving policy code and changing its +state behavior together would make failures harder to diagnose. The subsequent +[policy extraction](../POLICY.md) preserves the interface and runner-level tests. ## Lifecycle @@ -103,9 +103,9 @@ use the actual runner and policy together, with isolated game/model boundaries. ## Next boundaries -The state interface now gives a stable boundary for extracting combat and -selection policy without changing execution behavior. Keep a small dispatcher; -no plugin framework or class hierarchy is needed. +Combat and selection policy have since been extracted without changing execution +behavior. See [the current policy structure](../POLICY.md). No plugin framework +or class hierarchy was added. The next state work is run identity and deck provenance. Recording should then link observations, proposals, action attempts, results, and reconciliation. diff --git a/policy/__init__.py b/policy/__init__.py new file mode 100644 index 0000000..bc2478c --- /dev/null +++ b/policy/__init__.py @@ -0,0 +1 @@ +"""Policy components. Use brain.decide as the observe-to-proposal entry point.""" diff --git a/policy/combat.py b/policy/combat.py new file mode 100644 index 0000000..4f1a125 --- /dev/null +++ b/policy/combat.py @@ -0,0 +1,300 @@ +"""Combat proposals. Arithmetic and supported damage searches remain in facts.py.""" +from __future__ import annotations + +from typing import Any + +import facts as F +from jev import JevClient, NoulAnswer, choice, gate, noul +from .context import Decision + +# Confidence below this escalates instead of acting. +CONFIDENCE_FLOOR = 0.55 + + +def _lethal_line(playable: list[dict], energy: int, player_status: list, + enemy: F.EnemyFact) -> list[dict] | None: + """Use the same supported damage calculation as the fact layer.""" + return F.lethal_line(playable, energy, player_status, enemy) + + +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 in a SHORT fight, front-loading damage is better: + # HP is a resource. (STS2MCP strategy notes: "HP is a resource, not a + # score", "Front-load damage".) + # + # The old test was `threat in (HEAVY, SEVERE, LETHAL) or hp in (WOUNDED, + # CRITICAL)`, which ignored any hit of <=12 at 80 max HP. That is how the + # bot lost 9 runs to THE_KIN_BOSS. `block_urgent` adds the fight-length + # view: a small hit that repeats for 10 turns is not a small hit. + blockers = F.best_block_line(playable, f.energy) + if blockers and f.block_urgent: + best = blockers[0] + 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 _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 _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 {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 + }, + ) + + # NOTE: there used to be a `should_defend` Noul here. It was asked on every + # combat turn and never read -- `grep -rn should_defend` returned only the + # line that created it -- so it cost latency and did nothing. The defense + # decision is now made in code before Jev is consulted (see + # `combat_decision`), and the fight-length facts it needs are in + # `combat.fight_is_grinding` / `combat.this_turn_is_dangerous`. + + 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. A supported direct-damage line against one enemy. + for enemy in f.enemies: + line = _lethal_line(f.playable, f.energy, f.player_status, enemy) + if line: + card = line[0] + params = _target_params(card, f, force_target=enemy.entity_id) + return Decision( + "play_card", params, + f"direct-damage line on {enemy.entity_id} ({len(line)} cards)", "code", + ) + + # 2. Defense, decided in CODE and taken before Jev is asked. + # + # Measured, this was the single biggest hole. `THE_KIN_BOSS` ended 9 of 37 + # runs; those fights lasted 5-10 turns and cost 44-80 HP, ~10-13 a turn, + # with block cards in hand the whole time. Both decision paths preferred + # damage: the fallback classed a 12-damage hit at 80 max HP as "chip", and + # Jev, asked "which play best advances winning this fight?", chose Bash at + # 0.42 confidence. + # + # The fight-duration projection is a heuristic, not a known future. + # Use it only while there is a current block deficit. Start a maximum-block + # plan instead of greedily choosing the largest individual block card. + if f.block_urgent: + blockers = F.best_block_line(f.playable, f.energy) + if blockers: + best = blockers[0] + return Decision( + "play_card", + _target_params(best, f), + f"defense forced: {f.projected_incoming} projected vs " + f"{f.affordable_loss} affordable ({f.turns_to_kill} turns)", + "code", + ) + + # 3. Jev for preference, 4. heuristic if it is unsure. + if client is not None: + # The runner owns the failure budget and the model-free retry. + return _jev_combat(f, client) + return _fallback_combat(f) diff --git a/policy/context.py b/policy/context.py new file mode 100644 index 0000000..f144801 --- /dev/null +++ b/policy/context.py @@ -0,0 +1,149 @@ +"""Shared proposal types and session-owned policy memory. No game or model calls.""" +from __future__ import annotations + +import json +from dataclasses import dataclass, field + + +@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}" + + +# 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) + + +@dataclass +class PendingAction: + """An accepted request, not proof that the game completed it.""" + + decision: Decision + screen: dict + + +@dataclass +class PolicyContext: + """Session-owned memory. Proposals never record execution. + + Grid toggles and character selection lack observable selection fields in + the mod. Their accepted requests are tracked explicitly, not called facts. + Other guarded actions wait for relevant screen evidence before continuing. + """ + + screen_group: str | None = None + accepted_card_indices: set[int] = field(default_factory=set) + accepted_card_grid: list[dict] | None = None + accepted_crystal_cells: set[tuple[int, int]] = field(default_factory=set) + rewards_skipped_card: bool = False + character_selected: bool = False + pending: PendingAction | None = None + + def observe(self, obs: dict) -> None: + st = obs.get("state_type") + if st in ("unknown", "overlay"): + return # A transient overlay is not evidence of completion. + group = _screen_group(st, obs.get("menu_screen")) + if group != self.screen_group: + self.screen_group = group + self.accepted_card_indices.clear() + self.accepted_card_grid = None + self.accepted_crystal_cells.clear() + self.rewards_skipped_card = False + self.character_selected = False + self.pending = None + return + pending = self.pending + if pending is None: + return + action, params = pending.decision.action, pending.decision.params + screen = obs.get(st) or {} + if action == "select_card" and pending.screen.get("screen_type") != "choose": + # The API exposes no selected indices. Reserve accepted toggles + # after a fresh read, even when their effect is not visible yet. + if screen.get("cards") != pending.screen.get("cards"): + return # Cannot map a toggle safely onto a changed grid. + self.accepted_card_indices.add(params["index"]) + self.accepted_card_grid = pending.screen.get("cards") + elif action == "menu_select" and params.get("option") != "embark": + self.character_selected = True + elif action == "skip_card_reward": + if st != "rewards": + return + self.rewards_skipped_card = True + elif action in ("confirm_selection", "cancel_selection", + "confirm_bundle_selection", "cancel_bundle_selection", + "combat_confirm_selection"): + return # Wait for the screen to close; do not cancel a slow confirm. + elif action == "crystal_sphere_click_cell": + cell = (params["x"], params["y"]) + clickable = {(c.get("x"), c.get("y")) + for c in screen.get("clickable_cells", [])} + if cell in clickable and not screen.get("can_proceed"): + return + self.accepted_crystal_cells.add(cell) + elif action == "shop_purchase": + # Gold alone can change for unrelated reasons. Require this item + # to change/disappear, or a screen transition (e.g. card removal). + before = pending.screen.get("shop", pending.screen) + after = screen.get("shop", screen) + index = params["index"] + old = next((i for i in before.get("items", []) if i.get("index") == index), None) + new = next((i for i in after.get("items", []) if i.get("index") == index), None) + def inventory_item(item): + return {k: v for k, v in item.items() if k not in ("can_afford", "price")} if item else None + if inventory_item(old) == inventory_item(new): + return + elif action == "combat_select_card": + if len(screen.get("selected_cards") or []) <= len(pending.screen.get("selected_cards") or []): + return + elif action == "select_bundle": + if not screen.get("preview_showing") or not screen.get("can_confirm"): + return + else: + return # Direct choices and embark require a screen transition. + self.pending = None + + def record_result(self, obs: dict, decision: Decision, *, accepted: bool) -> None: + """Called only after a real action result. Transport errors stop the runner.""" + if not accepted: + self.pending = None + if decision.action == "menu_select" and decision.params.get("option") == "embark": + self.character_selected = False + return + st = obs.get("state_type") + guarded = decision.action in { + "select_card", "confirm_selection", "cancel_selection", + "select_bundle", "confirm_bundle_selection", "cancel_bundle_selection", + "shop_purchase", "skip_card_reward", "crystal_sphere_click_cell", + "combat_select_card", "combat_confirm_selection", + } or (st == "menu" and obs.get("menu_screen") == "character_select") + if guarded: + # Detach from mutable fixtures/callers. This is a small screen node, + # not another copy of the entire combat observation. + screen = json.loads(json.dumps(obs.get(st) or {})) + self.pending = PendingAction(decision, screen) diff --git a/policy/selection.py b/policy/selection.py new file mode 100644 index 0000000..e599fa6 --- /dev/null +++ b/policy/selection.py @@ -0,0 +1,687 @@ +"""Card, relic, bundle, and reward proposals. Execution belongs to the runner.""" +from __future__ import annotations + +import re + +import facts as F +from jev import JevClient, NoulAnswer, choice, gate, noul +from .context import Decision, PolicyContext + +# Minimum absolute Noul before adding an offered card to the deck. +CARD_PICK_THRESHOLD = 0.60 + + +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, + *, skip_policy: str = "jev") -> 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": F.deck_context(deck), + "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) + ): + 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 skip_policy == "combined" and can_skip and best_noul < CARD_PICK_THRESHOLD: + 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: + 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) + + # The questions are keyed `good_relicN`, so the ranking MUST read the + # prefixed ids. Reading `relicN` found nothing, `best_by_noul` returned + # (None, 0.0) for every state, and this path could only ever take the + # rarest relic -- every answer Jev gave was silently dropped. + best_key, best_noul = best_by_noul(response, [f"good_{k}" for k in 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"good_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) + + +# 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, add. + + Three traps, all 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"/"simple_select" while the + PROMPT says what is actually happening. Measured: "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. + * "Choose 2 Common Cards to Add to Your Deck." is neither: it is ADDING. + + 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" + if "add to your deck" in text or "add to your deck" in text: + return "add" + if "add" in text and "deck" in text: + return "add" + # 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) + if kind == "add": + # Adding: take the rarest card on offer. + return max(cards, key=lambda c: RARITY_RANK.get(str(c.get("rarity")), 1)) + return min(cards, key=upgrade_rank) + + +def _card_select_pick(index: int, reason: str, source: str, + confidence: float | None = None) -> Decision: + """Propose a toggle. Only an accepted request enters policy memory.""" + 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. + + The pattern must not assume the word "cards" directly follows the number: + "Choose 2 Common Cards to Add to Your Deck." has "Common" in between, and a + stricter pattern silently returned 1 and stalled the screen. + """ + match = re.search(r"choose\s+(\d+)", 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, + context: PolicyContext) -> 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 can make `confirm_selection` report ok without a visible + change. The context waits for the screen to close instead of repeating + or cancelling an action whose outcome is still unknown. + """ + cs = obs.get("card_select") or {} + chosen = context.accepted_card_indices + + 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 cs.get("can_confirm") and (len(chosen) >= need or cs.get("preview_showing")): + return Decision("confirm_selection", {}, "confirm the selection", "code") + if len(chosen) >= need: + # Enough chosen but the game has not enabled confirm yet. Selecting more + # would overshoot, so wait. + return Decision("__wait__", {}, + f"{len(chosen)}/{need} accepted toggles; 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. Do not repeat an accepted request for this grid. + remaining = [c for c in cards if c.get("index") not in chosen] + if not remaining: + if cs.get("can_confirm"): + return Decision("confirm_selection", {}, "all selectable cards chosen", "code") + return Decision("__wait__", {}, + f"{len(chosen)}/{need} accepted toggles; 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 exclude indices with accepted toggle requests. + 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?" + ) + elif kind == "add": + questions[f"good_{key}"] = noul( + f"Would adding `cards.{key}` to this deck make it 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": F.deck_context(deck), + "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) + + +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) + + +def rewards_decision(obs: dict, context: PolicyContext) -> 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 context.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") + + +def bundle_select_decision(obs: dict, client: JevClient | None, + deck: dict | None) -> Decision: + """ + Bundle choice: pick one of several 3-card bundles. + + `select_bundle` errors when a preview is already open. Confirm the preview + instead. The context waits for evidence after either accepted request; + an unchanged read does not prove that confirmation failed. + """ + bs = obs.get("bundle_select") or {} + + if bs.get("preview_showing") and bs.get("can_confirm"): + 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": F.deck_context(deck), + "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") diff --git a/test_brain.py b/test_brain.py index 528d17a..09672dd 100644 --- a/test_brain.py +++ b/test_brain.py @@ -17,6 +17,7 @@ from __future__ import annotations import sys import brain +from policy import combat as combat_policy, selection import facts import sts2 @@ -306,26 +307,26 @@ 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( +from_brain = selection.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( +from_brain = selection.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( +from_brain = selection.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( +from_brain = selection.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") @@ -658,27 +659,27 @@ check("even with jev, no enemies means wait", d.action, "__wait__") # 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") + selection.screen_kind("NDeckEnchantSelectScreen"), "upgrade") check("friendly names still work", - (brain.screen_kind("remove"), brain.screen_kind("transform")), ("remove", "transform")) + (selection.screen_kind("remove"), selection.screen_kind("transform")), ("remove", "transform")) # 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) + selection.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) + selection.card_select_need("Choose a card to Upgrade."), 1) # The number is not always immediately followed by "cards": # "Choose 2 Common Cards to Add to Your Deck." has "Common" in between. A # stricter pattern returned 1 and stalled the screen waiting for a confirm. check("the count survives an adjective", - brain.card_select_need("Choose 2 Common Cards to Add to Your Deck."), 2) + selection.card_select_need("Choose 2 Common Cards to Add to Your Deck."), 2) check("an ADD screen is not an upgrade screen", - brain.screen_kind("simple_select", "Choose 2 Common Cards to Add to Your Deck."), + selection.screen_kind("simple_select", "Choose 2 Common Cards to Add to Your Deck."), "add") check("a REMOVE prompt beats a generic screen_type", - brain.screen_kind("select", "Choose 5 cards to Remove."), "remove") + selection.screen_kind("select", "Choose 5 cards to Remove."), "remove") # Crystal Sphere: can_proceed is FALSE until tiles are revealed. The old # unconditional crystal_sphere_proceed was rejected and stalled the run. @@ -703,9 +704,9 @@ check("crystal sphere proceeds when unlocked", d3.action, "crystal_sphere_procee # "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) +check("a singular prompt needs 1", selection.hand_select_need("Choose a card to Exhaust."), 1) +check("a counted prompt needs N", selection.hand_select_need("Choose 2 cards to discard."), 2) +check("\"any number\" is unbounded", selection.hand_select_need("Choose any number of cards to replace."), None) d = brain.decide( {"state_type": "hand_select", @@ -805,7 +806,7 @@ kin_hand = [ def blk(obs): """The block value of whatever combat_decision picks, or None.""" - d = brain.combat_decision(facts.combat_facts(obs), None) + d = combat_policy.combat_decision(facts.combat_facts(obs), None) if d.action != "play_card": return d.action return facts.block_value(facts.combat_facts(obs).hand[d.params["card_index"]]) @@ -826,16 +827,16 @@ check("nothing incoming does not force a block", blk(quiet), 0) # Defense must never pre-empt lethal. lethal = combat(hand=kin_hand + [card("Bludgeon", index=4, cost="3", desc="Deal 32 damage.")], enemies=[enemy("KIN_0", hp=20, intent=12)], hp=74) -d = brain.combat_decision(facts.combat_facts(lethal), None) +d = combat_policy.combat_decision(facts.combat_facts(lethal), None) check("lethal still beats blocking", facts.combat_facts(lethal).hand[d.params["card_index"]]["name"], "Bludgeon") # The dead question must stay dead. Assert on the QUESTION being built, not # on the string: the replacement comment legitimately names it. check("the should_defend question is no longer asked", - 'questions["should_defend"]' in open("brain.py").read(), False) + 'questions["should_defend"]' in open("policy/combat.py").read(), False) check("...and nothing reads it either", - "response.get(\"should_defend\")" in open("brain.py").read(), False) + "response.get(\"should_defend\")" in open("policy/combat.py").read(), False) print() print("=== relic_select reads the ids it asked for (regression) ===") @@ -859,7 +860,7 @@ relic_obs = { } relic_stub = StubClient(noul=0.30, noul_override={"good_relic0": 0.80, "good_relic1": 0.40}) -d = brain.relic_select_decision(relic_obs, relic_stub) +d = selection.relic_select_decision(relic_obs, relic_stub) check("relic_select acts on Jev's answer", d.source, "jev") check("...and takes the relic Jev rated highest, not the rarest", d.params.get("index"), 0) @@ -875,7 +876,7 @@ blocked = combat(hand=[card(index=n) for n in range(3)], bf = facts.combat_facts(blocked) check("18 raw vs 10 hp + 5 block is lethal", bf.lethal_available, True) check("...and the executor finds the line", - bool(brain._lethal_line(bf.playable, bf.energy, bf.player_status, bf.enemies[0])), + bool(combat_policy._lethal_line(bf.playable, bf.energy, bf.player_status, bf.enemies[0])), True) strong = combat(hand=[card(index=n, desc="Deal 12 damage.") for n in range(2)], @@ -883,11 +884,11 @@ strong = combat(hand=[card(index=n, desc="Deal 12 damage.") for n in range(2)], status=[{"name": "Strength", "amount": 6, "type": "Buff"}]) sf = facts.combat_facts(strong) check("the executor uses Strength-adjusted hand text", - [c["name"] for c in (brain._lethal_line(sf.playable, sf.energy, + [c["name"] for c in (combat_policy._lethal_line(sf.playable, sf.energy, sf.player_status, sf.enemies[0]) or [])], ["Strike", "Strike"]) check("...and does not need to add Strength again", - bool(brain._lethal_line(sf.playable, sf.energy, [], sf.enemies[0])), True) + bool(combat_policy._lethal_line(sf.playable, sf.energy, [], sf.enemies[0])), True) print() print("=== correctness: shared damage and block plans ===") @@ -897,7 +898,7 @@ adjusted = combat(hand=[card(index=5, desc="Deal 8 damage.")], status=[{"name": "Strength", "amount": 2}]) af = facts.combat_facts(adjusted) check("executor does not invent a Strength-based lethal line", - brain._lethal_line(af.playable, af.energy, af.player_status, af.enemies[0]), None) + combat_policy._lethal_line(af.playable, af.energy, af.player_status, af.enemies[0]), None) defenders = [card("Big Block", index=5, cost="2", desc="Gain 9 Block.", ctype="Skill", target="Self"), card("Small Block", index=8, desc="Gain 6 Block.", ctype="Skill", target="Self"),