refactor(policy): extract combat selection and context modules

This commit is contained in:
0xrsydn 2026-09-22 12:45:46 +07:00
commit 484551047c
9 changed files with 1247 additions and 1167 deletions

687
policy/selection.py Normal file
View file

@ -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")