608 lines
25 KiB
Python
608 lines
25 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
brain.py -- policy dispatch and the remaining navigation, shop, and minigame flows.
|
|
|
|
Precedence, and why it is ordered this way:
|
|
|
|
1. CODE searches supported direct-damage lines in facts.py. The policy plays
|
|
one card, then observes again. This is a limited model, not a complete
|
|
combat simulation. Jev is never asked to calculate damage.
|
|
2. CODE decides the fallback. When Jev is unsure, or unavailable, a documented
|
|
heuristic (adapted from the STS2MCP AGENTS.md strategy notes) acts instead.
|
|
3. JEV decides preference. Only when lethal is not available and the fallback
|
|
is not forced do we ask Jev which play is best. That is a judgement about
|
|
semantics, which is what Jev is actually good at.
|
|
|
|
Every decision returns exactly ONE action. Card indices shift on every play,
|
|
so the loop must re-observe after each action.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import 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.
|
|
EVENT_TOP_MIN = 0.60
|
|
EVENT_MARGIN_MIN = 0.30
|
|
|
|
# Deterministic safety net for uncertain event choices.
|
|
#
|
|
# A "does this risk losing the run?" Noul was tried and REMOVED: measured, it
|
|
# ranked "Lose Everything" (0.46) as LESS risky than "Keep Deciphering" (0.52)
|
|
# and lower than "Stop" would suggest. It is actively misleading, so danger is
|
|
# detected by keywords instead. Lower score is safer.
|
|
EVENT_RISK_WORDS = (
|
|
"everything", "keep", "continue", "deeper", "again", "more", "all of",
|
|
"gamble", "risk", "sacrifice", "lose", "push", "betray", "accept",
|
|
)
|
|
EVENT_STOP_WORDS = (
|
|
"stop", "leave", "proceed", "back", "refuse", "decline", "end",
|
|
"take what", "walk away", "done", "nothing",
|
|
)
|
|
|
|
|
|
def event_safety_rank(option: dict) -> int:
|
|
"""Lower is safer. Deterministic; never consults the model."""
|
|
text = f"{option.get('title', '')} {option.get('description', '')}".lower()
|
|
score = sum(1 for w in EVENT_RISK_WORDS if w in text)
|
|
score -= sum(1 for w in EVENT_STOP_WORDS if w in text)
|
|
return score
|
|
|
|
MAP_NODE_MEANINGS = {
|
|
"Monster": "A normal fight. Costs some health, pays a card reward and gold.",
|
|
"Elite": "A hard fight. Pays a relic. Dangerous at low health.",
|
|
"Rest": "A campfire. Heal, or upgrade a card.",
|
|
"Shop": "Spend gold on cards, relics, potions, or removing a card.",
|
|
"Treasure": "A free relic with no fight.",
|
|
"Unknown": "Unknown. Could be a fight, an event, a shop, or treasure.",
|
|
"Boss": "The act boss. Ends the act.",
|
|
}
|
|
|
|
|
|
def _fallback_map(opts: list[dict], hp_pct: float, gold: int) -> Decision:
|
|
"""Documented pathing heuristic from the STS2MCP strategy notes."""
|
|
kinds = {str(o.get("type")): o for o in opts}
|
|
|
|
if hp_pct < 0.5 and "Rest" in kinds:
|
|
return Decision("choose_map_node", {"index": kinds["Rest"]["index"]},
|
|
f"hp {hp_pct:.0%} is low, take the rest site", "fallback")
|
|
if hp_pct > 0.7 and "Elite" in kinds:
|
|
return Decision("choose_map_node", {"index": kinds["Elite"]["index"]},
|
|
f"hp {hp_pct:.0%} is healthy, take the elite for a relic", "fallback")
|
|
if "Treasure" in kinds:
|
|
return Decision("choose_map_node", {"index": kinds["Treasure"]["index"]},
|
|
"free relic", "fallback")
|
|
if gold >= 100 and "Shop" in kinds:
|
|
return Decision("choose_map_node", {"index": kinds["Shop"]["index"]},
|
|
f"{gold} gold, visit the shop", "fallback")
|
|
if "Unknown" in kinds and hp_pct > 0.6:
|
|
return Decision("choose_map_node", {"index": kinds["Unknown"]["index"]},
|
|
"healthy enough to gamble on unknown", "fallback")
|
|
return Decision("choose_map_node", {"index": opts[0]["index"]},
|
|
"default to the first option", "fallback")
|
|
|
|
|
|
def map_decision(obs: dict, client: JevClient | None, deck: dict | None) -> Decision:
|
|
"""
|
|
Map pathing. A macro decision: elite fights pay relics but can end a run.
|
|
The choice depends on health, gold, and deck quality, so Jev is asked.
|
|
"""
|
|
m = obs.get("map") or {}
|
|
opts = [o for o in (m.get("next_options") or []) if isinstance(o, dict)]
|
|
|
|
if not opts:
|
|
return Decision("__wait__", {}, "map has no next options yet; re-observe", "code")
|
|
if len(opts) == 1:
|
|
return Decision("choose_map_node", {"index": opts[0]["index"]},
|
|
f"only option: {opts[0].get('type')}", "code")
|
|
|
|
player = obs.get("player") or {}
|
|
run = obs.get("run") or {}
|
|
hp = player.get("hp") or 0
|
|
max_hp = player.get("max_hp") or 1
|
|
hp_pct = hp / max_hp if max_hp else 0.0
|
|
gold = player.get("gold") or 0
|
|
|
|
if client is None:
|
|
return _fallback_map(opts, hp_pct, gold)
|
|
|
|
options = {
|
|
f"node{o['index']}": (
|
|
f"{o.get('type')} (column {o.get('col')}, row {o.get('row')}). "
|
|
f"{MAP_NODE_MEANINGS.get(str(o.get('type')), '')}"
|
|
)
|
|
for o in opts
|
|
}
|
|
|
|
boss = m.get("boss") or {}
|
|
state = {
|
|
"run": {"act": run.get("act"), "floor": run.get("floor")},
|
|
"character": player.get("character"),
|
|
"health": F._hp_bucket(hp_pct, hp),
|
|
"gold": gold,
|
|
"deck_composition": F.deck_context(deck),
|
|
"act_boss": boss.get("name") if isinstance(boss, dict) else None,
|
|
}
|
|
|
|
response = client.ask(state, {
|
|
"next_node": choice(
|
|
"Which node should the player travel to next? Health is a resource, "
|
|
"but an elite fight at low health can end the run.",
|
|
options,
|
|
)
|
|
})
|
|
|
|
pick = response["next_node"]
|
|
if not gate(pick):
|
|
return _fallback_map(opts, hp_pct, gold)
|
|
|
|
node = next((o for o in opts if f"node{o['index']}" == pick.choice), None)
|
|
if node is None:
|
|
return _fallback_map(opts, hp_pct, gold)
|
|
return Decision("choose_map_node", {"index": node["index"]},
|
|
f"jev chose {node.get('type')}", "jev", pick.confidence)
|
|
|
|
|
|
def event_decision(obs: dict, client: JevClient | None) -> Decision:
|
|
"""
|
|
Events. Option 0 is often locked, and Ancient events start in dialogue,
|
|
which is why a blind choose_event_option(index=0) gets rejected.
|
|
"""
|
|
ev = obs.get("event") or {}
|
|
|
|
if ev.get("in_dialogue"):
|
|
return Decision("advance_dialogue", {}, "click through dialogue", "code")
|
|
|
|
opts = [o for o in (ev.get("options") or []) if isinstance(o, dict)]
|
|
usable = [o for o in opts if not o.get("is_locked") and not o.get("was_chosen")]
|
|
if not usable:
|
|
usable = [o for o in opts if not o.get("is_locked")]
|
|
if not usable:
|
|
return Decision("__wait__", {}, "no usable event options; re-observe", "code")
|
|
|
|
pool = [o for o in usable if not o.get("is_proceed")] or usable
|
|
|
|
if len(pool) == 1:
|
|
return Decision("choose_event_option", {"index": pool[0]["index"]},
|
|
f"only option: {pool[0].get('title')}", "code")
|
|
|
|
if client is None:
|
|
return Decision("choose_event_option", {"index": pool[0]["index"]},
|
|
"first usable option", "fallback")
|
|
|
|
player = obs.get("player") or {}
|
|
options = {
|
|
f"opt{o['index']}": f"{o.get('title')}: {o.get('description')}"
|
|
for o in pool
|
|
}
|
|
|
|
questions: dict[str, dict] = {
|
|
"best_option": choice(
|
|
"Which event option is the best choice for the player?", options)
|
|
}
|
|
|
|
response = client.ask(
|
|
{
|
|
"event": ev.get("body") or "",
|
|
"character": player.get("character"),
|
|
"health": F._hp_bucket(
|
|
(player.get("hp") or 0) / (player.get("max_hp") or 1),
|
|
player.get("hp"),
|
|
),
|
|
"gold": player.get("gold"),
|
|
},
|
|
questions,
|
|
)
|
|
|
|
pick = response["best_option"]
|
|
|
|
if isinstance(pick, ChoiceAnswer):
|
|
top = pick.probabilities.get(pick.choice, 0.0)
|
|
runner = max(
|
|
(v for k, v in pick.probabilities.items() if k != pick.choice),
|
|
default=0.0,
|
|
)
|
|
else:
|
|
top = runner = 0.0
|
|
|
|
chosen = next((o for o in pool if f"opt{o['index']}" == getattr(pick, "choice", None)), None)
|
|
|
|
# Act on the model only when it is BOTH confident and not obviously risky.
|
|
if (
|
|
chosen is not None
|
|
and top >= EVENT_TOP_MIN
|
|
and (top - runner) >= EVENT_MARGIN_MIN
|
|
and event_safety_rank(chosen) <= 0
|
|
):
|
|
return Decision("choose_event_option", {"index": chosen["index"]},
|
|
f"jev chose {chosen.get('title')} ({top:.2f})", "jev",
|
|
getattr(pick, "confidence", None))
|
|
|
|
safest = min(pool, key=event_safety_rank)
|
|
if chosen is not None and event_safety_rank(chosen) > 0:
|
|
reason = f"jev picked risky '{chosen.get('title')}'; took safest option"
|
|
else:
|
|
reason = f"uncertain ({top:.2f}); took safest option"
|
|
return Decision("choose_event_option", {"index": safest["index"]}, reason,
|
|
"fallback", top or None)
|
|
|
|
|
|
def rest_site_decision(obs: dict) -> Decision:
|
|
"""
|
|
Heal when hurt, otherwise upgrade. (STS2MCP notes: rest before boss.)
|
|
|
|
Rest options expose `name` and `id`, NOT `title`, plus `is_enabled`.
|
|
"""
|
|
rs = obs.get("rest_site") or {}
|
|
opts = [
|
|
o for o in (rs.get("options") or [])
|
|
if isinstance(o, dict) and o.get("is_enabled", True)
|
|
]
|
|
player = obs.get("player") or {}
|
|
hp_pct = (player.get("hp") or 0) / (player.get("max_hp") or 1)
|
|
|
|
if not opts:
|
|
# As with the shop, do not gate the exit on `can_proceed`.
|
|
return Decision("proceed", {}, "nothing to do here; proceed", "code")
|
|
|
|
def label(o: dict) -> str:
|
|
return f"{o.get('name', '')} {o.get('id', '')} {o.get('description', '')}".lower()
|
|
|
|
rest = next((o for o in opts if "rest" in label(o) or "heal" in label(o)), None)
|
|
smith = next((o for o in opts if "smith" in label(o) or "upgrade" in label(o)), None)
|
|
|
|
if hp_pct < 0.6 and rest is not None:
|
|
return Decision("choose_rest_option", {"index": rest["index"]},
|
|
f"hp {hp_pct:.0%}, heal", "fallback")
|
|
if smith is not None:
|
|
return Decision("choose_rest_option", {"index": smith["index"]},
|
|
f"hp {hp_pct:.0%}, upgrade a card", "fallback")
|
|
return Decision("choose_rest_option", {"index": opts[0]["index"]},
|
|
"first rest option", "fallback")
|
|
|
|
|
|
# Minimum absolute Noul before buying anything in a shop. Absolute judgements
|
|
# can legitimately be low for every candidate, so this is a floor, not a rank.
|
|
SHOP_BUY_THRESHOLD = 0.60
|
|
|
|
# How the card-reward skip is decided.
|
|
# "jev" : Jev decides via `skip_all`. Measured, Jev almost never says
|
|
# skip (13 takes / 0 skips over three sessions), so the deck
|
|
# grows by roughly +3.3 cards versus the threshold policy.
|
|
# "combined" : skip when Jev says skip OR when the best card is clearly weak.
|
|
# Keeps Jev in charge of the ranking while putting a floor under
|
|
# the deck size.
|
|
CARD_SKIP_POLICY = "jev"
|
|
|
|
|
|
def shop_item_text(item: dict) -> tuple[str, str]:
|
|
"""
|
|
Resolve a shop item's display name and description.
|
|
|
|
The shop uses category-specific field names, verified live:
|
|
card -> card_name / card_description
|
|
relic -> relic_name / relic_description
|
|
potion -> potion_name / potion_description
|
|
card_removal -> neither
|
|
Reading `name`/`description` yields None for every category.
|
|
"""
|
|
category = str(item.get("category") or "")
|
|
if category == "card_removal":
|
|
return "Card Removal", "Remove one card from your deck permanently."
|
|
for prefix in ("card", "relic", "potion"):
|
|
name = item.get(f"{prefix}_name")
|
|
if name:
|
|
return str(name), str(item.get(f"{prefix}_description") or "")
|
|
return str(item.get("name") or "?"), str(item.get("description") or "")
|
|
|
|
|
|
def shop_decision(obs: dict, client: JevClient | None, deck: dict | None) -> Decision:
|
|
"""
|
|
Shop. Run 001 finished with 279 unspent gold because this always skipped.
|
|
|
|
The state carries `price`, `is_stocked` and `can_afford` per item, so no
|
|
affordability arithmetic needs to reach the model.
|
|
"""
|
|
# `fake_merchant` nests its inventory one level deeper: fake_merchant.shop.
|
|
# Reading only obs["shop"] made every fake-merchant shop look empty, so the
|
|
# bot always left immediately without buying.
|
|
node = obs.get("shop")
|
|
if not isinstance(node, dict):
|
|
fm = obs.get("fake_merchant")
|
|
if isinstance(fm, dict):
|
|
node = fm.get("shop") if isinstance(fm.get("shop"), dict) else fm
|
|
node = node if isinstance(node, dict) else {}
|
|
items = [i for i in (node.get("items") or []) if isinstance(i, dict)]
|
|
affordable = [
|
|
i for i in items
|
|
if i.get("is_stocked", True) and i.get("can_afford", True)
|
|
]
|
|
|
|
if not affordable:
|
|
# `can_proceed` is UNRELIABLE here: measured False while proceed()
|
|
# worked and moved the game to the map. Never wait on it indefinitely.
|
|
return Decision("proceed", {}, "nothing affordable; leave", "code")
|
|
|
|
if client is None:
|
|
return Decision("proceed", {}, "no jev; skip shop", "fallback")
|
|
|
|
player = obs.get("player") or {}
|
|
run = obs.get("run") or {}
|
|
gold = player.get("gold") or 0
|
|
|
|
# A shop can offer 14+ affordable items. A single Choice over all of them
|
|
# dilutes the probability mass: measured, the top option scored only 0.26
|
|
# (runner 0.17, margin 0.09) and the margin gate correctly rejected it, so
|
|
# the bot would always leave. Use the documented re-ranking pattern instead:
|
|
# one ABSOLUTE Noul per candidate, then take the argmax in code. Absolute
|
|
# judgements do not dilute as the candidate count grows.
|
|
item_keys: dict[str, dict] = {f"item{i['index']}": i for i in affordable}
|
|
|
|
state = {
|
|
"run": {"act": run.get("act"), "floor": run.get("floor")},
|
|
"character": player.get("character"),
|
|
"health": F._hp_bucket(
|
|
(player.get("hp") or 0) / (player.get("max_hp") or 1),
|
|
player.get("hp"),
|
|
),
|
|
"gold": gold,
|
|
"deck_composition": F.deck_context(deck),
|
|
"items": {
|
|
key: {
|
|
"name": shop_item_text(item)[0],
|
|
"category": item.get("category"),
|
|
"price": item.get("price"),
|
|
"text": shop_item_text(item)[1],
|
|
}
|
|
for key, item in item_keys.items()
|
|
},
|
|
}
|
|
|
|
# Ask about DECK FIT ONLY. Measured on this exact shop: putting the price
|
|
# into the question collapsed the spread across candidates from 0.48 to
|
|
# 0.28 and pulled the top item down from 0.67 to 0.51. Weighing value
|
|
# against a number is exactly what Jev is documented to fail at.
|
|
# Affordability is already filtered in code, so the price stays in the
|
|
# state for context but stays OUT of the question.
|
|
questions: dict[str, dict] = {}
|
|
for key in item_keys:
|
|
questions[f"worth_{key}"] = noul(
|
|
f"Would `items.{key}` make this deck stronger?"
|
|
)
|
|
|
|
response = client.ask(state, questions)
|
|
|
|
ranked = [
|
|
(ans.noul, key)
|
|
for key in item_keys
|
|
if isinstance((ans := response.get(f"worth_{key}")), NoulAnswer)
|
|
]
|
|
if not ranked:
|
|
return Decision("proceed", {}, "no usable answers; leave", "fallback")
|
|
|
|
best_noul, best_key = max(ranked)
|
|
if best_noul < SHOP_BUY_THRESHOLD:
|
|
return Decision("proceed", {},
|
|
f"best item only noul={best_noul:.2f}; leave", "jev", best_noul)
|
|
|
|
item = item_keys[best_key]
|
|
return Decision("shop_purchase", {"index": item["index"]},
|
|
f"jev bought {shop_item_text(item)[0]} (noul={best_noul:.2f})",
|
|
"jev", best_noul)
|
|
|
|
|
|
def treasure_decision(obs: dict, client: JevClient | None,
|
|
deck: dict | None = None) -> Decision:
|
|
"""
|
|
The chest auto-opens. While it opens, `relics` is absent and only
|
|
`can_proceed` is set, so claiming then just gets rejected.
|
|
|
|
A chest usually offers one relic, in which case there is nothing to judge.
|
|
Jev is only consulted when there is a real choice.
|
|
"""
|
|
tr = obs.get("treasure") or {}
|
|
relics = [r for r in (tr.get("relics") or []) if isinstance(r, dict)]
|
|
|
|
if not relics:
|
|
if tr.get("can_proceed"):
|
|
return Decision("proceed", {}, "treasure done; proceed", "code")
|
|
return Decision("__wait__", {}, "chest opening; re-observe", "code")
|
|
|
|
if len(relics) == 1 or client is None:
|
|
return Decision("claim_treasure_relic", {"index": relics[0].get("index", 0)},
|
|
f"claim {relics[0].get('name')}", "code")
|
|
|
|
options = {
|
|
f"relic{r.get('index', 0)}": (
|
|
f"{r.get('name')} ({r.get('rarity')}): {r.get('description')}"
|
|
)
|
|
for r in relics
|
|
}
|
|
player = obs.get("player") or {}
|
|
|
|
response = client.ask(
|
|
{
|
|
"character": player.get("character"),
|
|
"health": F._hp_bucket(
|
|
(player.get("hp") or 0) / (player.get("max_hp") or 1),
|
|
player.get("hp"),
|
|
),
|
|
"deck_composition": F.deck_context(deck),
|
|
"relics": {
|
|
f"relic{r.get('index', 0)}": {
|
|
"name": r.get("name"),
|
|
"text": r.get("description"),
|
|
}
|
|
for r in relics
|
|
},
|
|
},
|
|
{"best_relic": choice("Which relic is strongest for this run?", options)},
|
|
)
|
|
|
|
pick = response["best_relic"]
|
|
if not gate(pick):
|
|
return Decision("claim_treasure_relic", {"index": relics[0].get("index", 0)},
|
|
"low confidence; take the first", "fallback")
|
|
chosen = next((r for r in relics if f"relic{r.get('index', 0)}" == pick.choice), None)
|
|
if chosen is None:
|
|
return Decision("claim_treasure_relic", {"index": relics[0].get("index", 0)},
|
|
"unmapped choice", "fallback")
|
|
return Decision("claim_treasure_relic", {"index": chosen.get("index", 0)},
|
|
f"jev chose {chosen.get('name')}", "jev", pick.confidence)
|
|
|
|
|
|
# Rough potion value, used only to choose which potion to drop when every slot
|
|
# is full. Lower is discarded first. Unknown potions rank in the middle.
|
|
def crystal_sphere_decision(obs: dict, context: PolicyContext) -> Decision:
|
|
"""
|
|
Crystal Sphere minigame.
|
|
|
|
Measured: `can_proceed` is FALSE until tiles are revealed, so the old
|
|
unconditional crystal_sphere_proceed was rejected and the run stalled.
|
|
Reveal clickable cells until the proceed button unlocks.
|
|
"""
|
|
cs = obs.get("crystal_sphere") or {}
|
|
|
|
if cs.get("can_proceed"):
|
|
return Decision("crystal_sphere_proceed", {}, "minigame done; proceed", "code")
|
|
|
|
clickable = [c for c in (cs.get("clickable_cells") or []) if isinstance(c, dict)]
|
|
|
|
# Never re-click a cell: that wastes a divination and can loop.
|
|
fresh = [
|
|
c for c in clickable
|
|
if (c.get("x"), c.get("y")) not in context.accepted_crystal_cells
|
|
]
|
|
|
|
if not fresh:
|
|
if clickable:
|
|
return Decision("crystal_sphere_proceed", {},
|
|
"nothing new to reveal; try to proceed", "fallback")
|
|
return Decision("__wait__", {}, "no clickable cells; re-observe", "code")
|
|
|
|
# Prefer the cell closest to the centre, which is where items tend to sit.
|
|
width = cs.get("grid_width") or 0
|
|
height = cs.get("grid_height") or 0
|
|
cx, cy = (width - 1) / 2, (height - 1) / 2
|
|
cell = min(fresh, key=lambda c: abs(c.get("x", 0) - cx) + abs(c.get("y", 0) - cy))
|
|
|
|
return Decision("crystal_sphere_click_cell",
|
|
{"x": cell.get("x"), "y": cell.get("y")},
|
|
f"reveal ({cell.get('x')},{cell.get('y')})", "code")
|
|
|
|
|
|
def simple_decision(obs: dict, client: JevClient | None = None,
|
|
deck: dict | None = None, *, context: PolicyContext | None = None) -> Decision | None:
|
|
"""Mechanical screens. Most need no model -- they are pure procedure."""
|
|
context = context if context is not None else PolicyContext()
|
|
st = obs.get("state_type")
|
|
|
|
if st == "menu":
|
|
screen = obs.get("menu_screen")
|
|
if screen == "main":
|
|
opts = obs.get("options") or []
|
|
names = [o if isinstance(o, str) else o.get("name") for o in opts]
|
|
return Decision("menu_select",
|
|
{"option": "continue" if "continue" in names else "singleplayer"},
|
|
"main menu", "code")
|
|
if screen == "tutorial_prompt":
|
|
return Decision("menu_select", {"option": "no"}, "disable tutorials", "code")
|
|
# Mode select, which appears after the first epoch unlock. `embark` is
|
|
# rejected here; only standard/daily/custom/back are valid.
|
|
if screen == "singleplayer":
|
|
return Decision("menu_select", {"option": "standard"},
|
|
"choose standard mode", "code")
|
|
# `embark` is REJECTED with "Embark button not available - select a
|
|
# character first" unless a character is chosen, and the state carries
|
|
# NO "selected" indicator: the mod hardcodes `message` to
|
|
# "Select a character." regardless. Verified in AddCharacterSelectMenuState.
|
|
#
|
|
# Embarking immediately after selecting is also FLAKY -- measured, three
|
|
# consecutive "select a character first" rejections -- because the
|
|
# selection has not registered yet. Select after a rejected embark,
|
|
# but wait after an accepted embark. Proposals alone do not advance
|
|
# the sequence.
|
|
if screen == "character_select":
|
|
options = obs.get("options") or []
|
|
names = [o if isinstance(o, str) else o.get("name") for o in options]
|
|
pick = next((c for c in ("IRONCLAD", "SILENT") if c in names), None)
|
|
|
|
if not context.character_selected and pick is not None:
|
|
return Decision("menu_select", {"option": pick},
|
|
f"select {pick}", "code")
|
|
|
|
return Decision("menu_select", {"option": "embark"},
|
|
"embark (a rejected embark is followed by a re-select)",
|
|
"code")
|
|
return None
|
|
|
|
if st == "game_over":
|
|
return Decision("menu_select", {"option": "main_menu"}, "run ended", "code")
|
|
|
|
if st == "rewards":
|
|
return selection.rewards_decision(obs, context)
|
|
|
|
if st == "card_reward":
|
|
return selection.card_reward_decision(obs, client, deck, skip_policy=CARD_SKIP_POLICY)
|
|
|
|
if st == "relic_select":
|
|
return selection.relic_select_decision(obs, client)
|
|
|
|
if st == "map":
|
|
return map_decision(obs, client, deck)
|
|
|
|
if st == "rest_site":
|
|
return rest_site_decision(obs)
|
|
|
|
if st == "treasure":
|
|
return treasure_decision(obs, client, deck)
|
|
|
|
if st == "event":
|
|
return event_decision(obs, client)
|
|
|
|
if st in ("shop", "fake_merchant"):
|
|
return shop_decision(obs, client, deck)
|
|
|
|
if st == "hand_select":
|
|
return selection.hand_select_decision(obs, client, deck)
|
|
|
|
if st == "card_select":
|
|
return selection.card_select_decision(obs, client, deck, context)
|
|
|
|
if st == "bundle_select":
|
|
return selection.bundle_select_decision(obs, client, deck)
|
|
|
|
if st == "crystal_sphere":
|
|
return crystal_sphere_decision(obs, context)
|
|
|
|
# Transitions and unhandled overlays are not dead ends. Wait and look again;
|
|
# run.py's unchanged-state guard bounds this so a real dead end still stops.
|
|
if st in ("unknown", "overlay"):
|
|
return Decision("__wait__", {}, f"{st} state; re-observe", "code")
|
|
|
|
return None
|
|
|
|
|
|
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.
|
|
|
|
Omit context for independent snapshot analysis. Runners must retain one
|
|
context and report action results; repeated proposals alone advance nothing.
|
|
"""
|
|
context = context if context is not None else PolicyContext()
|
|
context.observe(obs)
|
|
st = obs.get("state_type")
|
|
if context.pending is not None:
|
|
return Decision("__wait__", {},
|
|
f"awaiting screen evidence after {context.pending.decision.action}", "code")
|
|
if (st == "card_select" and context.accepted_card_grid is not 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.combat_decision(F.combat_facts(obs), client)
|
|
return simple_decision(obs, client, deck, context=context)
|