From 3f243eaeeed8d5148b64e4e7b74628f87d85342b Mon Sep 17 00:00:00 2001 From: 0xrsydn Date: Tue, 22 Sep 2026 12:45:46 +0700 Subject: [PATCH] fix(bot): correct combat estimates and enforce client and runner failures --- brain.py | 87 ++-------- docs/research/12-correctness-pass.md | 131 +++++++++++++++ facts.py | 237 +++++++++++++++------------ jev.py | 78 +++++++-- run.py | 108 ++++++++---- sts2.py | 30 +++- test_brain.py | 29 +++- test_facts.py | 81 +++++++-- test_run.py | 169 +++++++++++++++---- utils/audit_prototype.py | 182 ++++++++++++++++++++ 10 files changed, 859 insertions(+), 273 deletions(-) create mode 100644 docs/research/12-correctness-pass.md create mode 100644 utils/audit_prototype.py diff --git a/brain.py b/brain.py index 4ff0ade..6ea03a1 100644 --- a/brain.py +++ b/brain.py @@ -4,9 +4,9 @@ brain.py -- the decision layer. Precedence, and why it is ordered this way: - 1. CODE decides lethal. facts.py proves lethal is reachable; a deterministic - greedy then executes it. Jev is never asked "can I kill this", because it - answered that question wrongly at 0.79 confidence. + 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 @@ -19,14 +19,13 @@ so the loop must re-observe after each action. from __future__ import annotations -import itertools import json import re from dataclasses import dataclass, field from typing import Any import facts as F -from jev import JevClient, JevError, ChoiceAnswer, NoulAnswer, choice, gate, noul +from jev import JevClient, ChoiceAnswer, NoulAnswer, choice, gate, noul # Events are high-stakes and often irreversible, so the bar is higher than for # combat. Measured: Jev picked a run-ending option at confidence 0.49. @@ -78,49 +77,10 @@ class Decision: # Combat # -------------------------------------------------------------------------- -def _affordable(hand: list[dict], energy: int) -> list[dict]: - return [ - c for c in hand - if c.get("can_play") and F._as_int(c.get("cost")) <= energy - ] - def _lethal_line(playable: list[dict], energy: int, player_status: list, enemy: F.EnemyFact) -> list[dict] | None: - """ - Smallest energy-cost subset of playable cards that kills `enemy`. - - Returns the cards to play, highest damage first. None when lethal is - unreachable. Deterministic, so Jev is never involved. - """ - best: list[dict] | None = None - n = min(len(playable), 12) - for r in range(1, n + 1): - for combo in itertools.combinations(range(n), r): - cards = [playable[i] for i in combo] - cost = sum(F._as_int(c.get("cost")) for c in cards) - if cost > energy: - continue - total = 0 - for c in cards: - dmg = F.parse_card_damage(c) - if dmg.raw_total == 0: - continue - per_hit = dmg.base + F.power_amount(player_status, "Strength") - if F.power_amount(player_status, "Weak"): - per_hit = int(per_hit * 0.75) - card_total = per_hit * dmg.hits - if F.power_amount(F.enemy_status_names(enemy), "Vulnerable"): - card_total = int(card_total * 1.5) - total += card_total - # `total` is raw damage; `effective_hp` is hp + block, so the block - # is subtracted exactly once. Subtracting it here as well required - # hp + 2*block and discarded real lethal lines. - if total >= enemy.effective_hp: - if best is None or len(cards) < len(best): - best = cards - if best: - best.sort(key=lambda c: -F.parse_card_damage(c).raw_total) - return best + """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: @@ -165,9 +125,9 @@ def _fallback_combat(f: F.CombatFacts) -> Decision: # 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 = [c for c in playable if _block_value(c) > 0] + blockers = F.best_block_line(playable, f.energy) if blockers and f.block_urgent: - best = max(blockers, key=lambda c: _block_value(c)) + best = blockers[0] return Decision("play_card", _target_params(best, f), f"threat={f.threat} hp={f.hp_bucket}, take block", "fallback") @@ -187,10 +147,6 @@ def _fallback_combat(f: F.CombatFacts) -> Decision: return Decision("end_turn", {}, "nothing playable", "fallback") -def _block_value(card: dict) -> int: - return F.block_value(card) - - def _target_params(card: dict, f: F.CombatFacts, force_target: str | None = None) -> dict: params: dict[str, Any] = {"card_index": card.get("index")} if card.get("target_type") == "AnyEnemy": @@ -206,10 +162,6 @@ def _target_params(card: dict, f: F.CombatFacts, force_target: str | None = None return params -def _needs_enemy_target(card: dict) -> bool: - return card.get("target_type") == "AnyEnemy" - - def _usable_potions(f: F.CombatFacts) -> list[dict]: """Potions the game says can be used right now.""" return [p for p in f.potions if p.get("can_use_in_combat", True)] @@ -239,7 +191,7 @@ def _jev_combat(f: F.CombatFacts, client: JevClient) -> Decision: key = f"card{c['index']}" by_key[key] = c options[key] = ( - f"{c['name']} (cost {F._as_int(c.get('cost'))}): {c.get('description')}" + f"{c['name']} (cost {c.get('cost')}): {c.get('description')}" ) potion_options: dict[str, str] = {} @@ -372,7 +324,7 @@ def combat_decision(f: F.CombatFacts, client: JevClient | None) -> Decision: if not f.enemies: return Decision("__wait__", {}, "no enemies yet; re-observe", "code") - # 1. Deterministic lethal. + # 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: @@ -380,7 +332,7 @@ def combat_decision(f: F.CombatFacts, client: JevClient | None) -> Decision: params = _target_params(card, f, force_target=enemy.entity_id) return Decision( "play_card", params, - f"lethal line on {enemy.entity_id} ({len(line)} cards)", "code", + f"direct-damage line on {enemy.entity_id} ({len(line)} cards)", "code", ) # 2. Defense, decided in CODE and taken before Jev is asked. @@ -392,14 +344,13 @@ def combat_decision(f: F.CombatFacts, client: JevClient | None) -> Decision: # Jev, asked "which play best advances winning this fight?", chose Bash at # 0.42 confidence. # - # Blocking is a fact about arithmetic -- total incoming over the remaining - # fight versus the HP we can spare -- so it belongs here, not in a - # preference judgement. Jev is not asked to make it. + # 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 = [c for c in f.playable if F.block_value(c) > 0] + blockers = F.best_block_line(f.playable, f.energy) if blockers: - # Block hardest first; a single Defend is still better than a Bash. - best = max(blockers, key=lambda c: F.block_value(c)) + best = blockers[0] return Decision( "play_card", _target_params(best, f), @@ -410,10 +361,8 @@ def combat_decision(f: F.CombatFacts, client: JevClient | None) -> Decision: # 3. Jev for preference, 4. heuristic if it is unsure. if client is not None: - try: - return _jev_combat(f, client) - except JevError as exc: - print(f" [jev unavailable: {str(exc)[:90]}]") + # The runner owns the failure budget and the model-free retry. + return _jev_combat(f, client) return _fallback_combat(f) diff --git a/docs/research/12-correctness-pass.md b/docs/research/12-correctness-pass.md new file mode 100644 index 0000000..55d0bbe --- /dev/null +++ b/docs/research/12-correctness-pass.md @@ -0,0 +1,131 @@ +# 12 — First correctness pass + +Status: implemented; live gameplay quality is not yet measured. + +This pass addresses the reproducible defects from [the prototype audit](11-prototype-hardening.md). +It keeps the current Python modules and synchronous loop. No runtime dependency or test framework was added. + +## Combat changes + +- Hand damage already includes attacker Strength and Weak. The damage calculation no longer applies them again. +- The fact layer and policy share one direct-damage subset search. +- Vulnerable rounding happens per hit. Block is absorbed once when comparing a line with enemy HP plus block. +- The direct-damage search accepts plain attacks with fixed energy costs. + Compound/conditional descriptions, unknown/X costs, star spending, and unsupported powers are excluded. +- Individual killability remains separate from whole-combat lethal. + `lethal_available` is `None` for multiple enemies; no joint resource-allocation search exists yet. +- Unknown/X costs remain visible in model context rather than becoming zero. +- Block planning maximizes displayed immediate block under the fixed energy budget. + The policy starts that plan instead of selecting the largest block card independently. +- Fully covered incoming damage no longer forces additional defense. +- Estimated survival permits nonlethal HP loss; reaching zero HP does not count as survival. + +These are limited mechanics calculations, not a full simulator. +Relic hooks, draw outcomes, card side effects, and dynamic sequences remain outside the model. +A false result means no supported line was found, not that every possible game line was ruled out. +Fight duration and projected future damage remain heuristics. + +The stricter search can decline lines that the old code accepted, including compound attacks such as Bash. +Those cards remain available to the ordinary policy; they are not removed from playable actions. +This deliberately favors an explicit limitation over an unsupported lethal claim. + +## Runner changes + +`--dry-run` now sends no action POSTs, including both game-over dismissal paths. +It does not persist `deck.json`. It still reads state, can call Jev, and writes capture logs. +Some mod GET handlers have automatic UI behavior, so dry-run is not a game-state sandbox. + +Combat model errors reach the runner. Below the failure limit, the runner uses `brain.decide(..., client=None)` +so combat has a real heuristic fallback. Only a successful model response resets the failure counter; +a code-only action or animation wait does not demonstrate recovery. + +Missing model credentials no longer silently select heuristic-only play. Use `--no-jev` intentionally. +Unexpected policy errors stop the session instead of hiding programming defects behind another decision. +Game-over dismissals must return an accepted action result. + +### Exit codes + +| Code | Meaning | +|---|---| +| `0` | Requested bounded session or preview completed, or run end was observed | +| `1` | State/action/policy failure, repeated rejection, or unchanged-state timeout | +| `2` | Preflight blocker or invalid command-line arguments | +| `3` | Model initialization failed or consecutive model failures reached the limit | +| `4` | `--stop-on-run-end` reached the step limit without observing run end | + +A rejected final action cannot become success merely because the step limit was reached. +Each initialized session records `exit_code`, `stop_reason`, and `dry_run` in its session row. +The trace also records `session_end` and model failures, including the failure that triggers an abort. +These are small additions for failure visibility, not the complete recording redesign. + +## Client boundaries + +The game client wraps connection failures, timeouts, and read failures as `Sts2Error`. +JSON responses must be objects. Markdown requests use the same transport error handling. +The client never retries an action POST: after an ambiguous failure, the action may already have reached the game. + +The Jev parser rejects missing Noul values instead of converting them to a confident no. +It validates required answer fields, finite numeric ranges, Choice membership, Score levels, and usage counts. +Responses must match the requested question IDs, primitive types, and option/level sets. +Protocol failures raise `JevError` so the runner can apply its failure policy. + +## Verification approach + +Follow the project testing preference: integration/end-to-end checks first. +Keep only essential persistent regressions. Use temporary isolated probes for low-level edge cases. +The existing script tests remain; this pass does not migrate the legacy suite or add a permanent client unit-test suite. + +Permanent checks cover the corrected hand-description contract, shared-energy ambiguity, +block planning, and the real runner's dry-run/fallback/failure paths. +Runner checks isolate files and external boundaries, but execute the actual policy loop. +History checks now use a temporary run record instead of reading a personal game history directory. + +Temporary whole-process checks ran the real runner, facts, policy, and HTTP clients against local fixture endpoints. +Each process used a temporary working directory and capture/history paths. +Model credentials were dummy values, and the request wrapper rejected nonlocal URLs. +No real game or paid model was contacted. + +| Whole-process scenario | Result | +|---|---| +| Model decision, action, then game-over dismissal | Exit 0; two game action POSTs; one model request | +| Dry-run on a parked game-over screen | Exit 0; no POSTs | +| Dry-run combat with a model response | Exit 0; one model request; no game action POST or deck write | +| Model outage, heuristic fallback, wait, second outage | Exit 3; one fallback action; two model requests | +| Malformed model answer | Exit 3; no game action | +| Connection closes during an action POST | Exit 1; exactly one action attempt | +| Action rejected at the step limit | Exit 1 | +| Run still active at the step limit | Exit 4 | + +A Nix shell is optional for dependency isolation. It does not isolate game saves or prohibit network access. +The checks used the existing Python interpreter and standard library; temporary directories isolated their outputs. + +Run the permanent checks: + +```sh +python3 test_facts.py && python3 test_brain.py && python3 test_run.py +python3 utils/audit_prototype.py +python3 migrate.py --check-only +bash -n eval_batch.sh ab_card_skip.sh +``` + +Results: **272 assertions passed** across the three scripts. The dataset integrity and shell syntax checks passed. +The audit replayed **346 observations without policy exceptions**. Both captured Strength examples now calculate 8 damage, +and no captured state triggers forced defense after incoming damage is covered. +A temporary cross-check matched the block planner against exhaustive enumeration on 500 generated hands. + +The audit remains diagnostic, not a pass/fail suite. It also reports known issues that belong to the next step. + +## Next: policy state and recording + +Still open: + +1. Selection/shop/minigame memory can change when an action is proposed, before execution succeeds. +2. Captures cover only combat and reuse filenames across sessions. +3. Decisions lack an exact observation reference and a separate action-attempt identity. +4. Logged default gates can disagree with the gate actually used by a policy handler. +5. Combat-pile snapshots are still used as deck context without run identity or persistent-deck provenance. +6. Session identity is process-global, and session finalization still depends on `atexit`. + +The next change should introduce explicit policy memory and reconcile attempts with action results and observations. +Then link every observation, proposal, attempt, and result without expanding the module structure unnecessarily. +Do not infer improved win rate from these correctness checks. diff --git a/facts.py b/facts.py index 43dc783..6c2e760 100644 --- a/facts.py +++ b/facts.py @@ -14,24 +14,19 @@ Conclusion: every comparison, sum, and threshold is computed here. Jev is handed conclusions ("lethal_available": true) and is only ever asked about preference, priority, and semantics. -Damage model and its honest limits ----------------------------------- -Handled: - * base damage parsed from the card description - * multi-hit ("Deal 3 damage 2 times.") - * area of effect ("... to ALL enemies.") - * Strength on the attacker (+ per hit) - * Vulnerable on the target (x1.5) - * Weak on the attacker (x0.75) - * enemy Block -Not handled (documented, not guessed): - * ordering effects (Bash applying Vulnerable before a later attack) - * relics that modify damage - * enemy powers that reduce incoming damage - * X-cost cards -So `lethal_available` is a LOWER bound: true means lethal is genuinely -reachable. False may still be reachable in game. That is the safe direction -for a bot -- we never claim lethal we cannot deliver. +Damage model and its limits +--------------------------- +Hand descriptions from STS2MCP already include attacker modifiers such as +Strength and Weak. Never apply those modifiers to the displayed damage again. +Pile descriptions have a different display context and are not damage inputs. + +The direct-damage search supports fixed energy costs and plain damage text. +It excludes compound/conditional cards, star costs, and unsupported powers. +Target Vulnerable is applied per hit; enemy block is absorbed once per line. +The search does not simulate relic hooks, card ordering effects, or shared +resources across enemies. `lethal_available` is None for multi-enemy combat +and unsupported power contexts. Other results describe this limited model, +not a full game simulation. Fight duration and future damage are estimates. """ from __future__ import annotations @@ -89,7 +84,7 @@ def power_amount(statuses: list | None, name: str) -> int: @dataclass(frozen=True) class CardDamage: - base: int + base: int # displayed damage per hit, not the card's unmodified base value hits: int is_aoe: bool @@ -99,8 +94,8 @@ class CardDamage: def block_value(card: dict) -> int: - """Block a card grants, parsed from its description.""" - match = BLOCK_RE.search(card.get("description") or "") + """Displayed immediate block; do not treat conditional triggers as block now.""" + match = BLOCK_RE.match((card.get("description") or "").strip()) return int(match.group(1)) if match else 0 @@ -123,22 +118,17 @@ def damage_to_target( target_status: list | None, target_block: int = 0, ) -> int: - """Damage one card deals to one target after buffs, debuffs, and block.""" + """Estimate damage from HAND text, with target Vulnerable and block. + + `attacker_status` remains accepted for callers, but its modifiers are + already included in the displayed value. This is not a base-card API. + Other target powers and relic hooks are outside this calculation. + """ dmg = parse_card_damage(card) - if dmg.raw_total == 0: - return 0 - - strength = power_amount(attacker_status, "Strength") - per_hit = dmg.base + strength - if power_amount(attacker_status, "Weak"): - per_hit = int(per_hit * 0.75) - - total = per_hit * dmg.hits - + per_hit = dmg.base if power_amount(target_status, "Vulnerable"): - total = int(total * 1.5) - - return max(0, total - max(0, target_block)) + per_hit = per_hit * 3 // 2 + return max(0, per_hit * dmg.hits - max(0, target_block)) # -------------------------------------------------------------------------- @@ -250,9 +240,82 @@ def enemy_facts(enemy: dict) -> EnemyFact: # -------------------------------------------------------------------------- -# Lethal search +# Small-hand calculations. Unknown costs never become free cards. # -------------------------------------------------------------------------- +def fixed_energy_cost(card: dict) -> int | None: + """A supported fixed cost, or None for unknown/X costs and star spending.""" + if card.get("star_cost") not in (None, 0, "0"): + return None + text = str(card.get("cost", "")).strip() + return int(text) if text.isdecimal() else None + + +def best_block_line(playable: list[dict], energy: int) -> list[dict]: + """Maximize displayed immediate block with fixed costs, using each card once. + + This ignores draw outcomes and other side effects. Ties use less energy, + then fewer cards. The returned list is a plan, not a queued action list. + """ + plans: dict[int, tuple[int, list[dict]]] = {0: (0, [])} + for card in playable: + cost = fixed_energy_cost(card) + value = block_value(card) + if cost is None or cost > energy or value <= 0: + continue + for spent, (total, line) in list(plans.items()): + new_spent = spent + cost + if new_spent > energy: + continue + candidate = (total + value, line + [card]) + old = plans.get(new_spent) + if old is None or (candidate[0], -len(candidate[1])) > (old[0], -len(old[1])): + plans[new_spent] = candidate + _, (_, line) = max(plans.items(), key=lambda item: (item[1][0], -item[0], -len(item[1][1]))) + return line + + +def damage_context_supported(attacker_status: list, enemy: EnemyFact) -> bool: + """Reject power contexts outside the limited direct-damage calculation.""" + known = {"strength", "weak", "vulnerable", "dexterity", "frail"} + names = [str(s.get("name", "")).lower() for s in attacker_status if isinstance(s, dict)] + return all(n in known for n in names + [n.lower() for n in enemy.statuses]) + + +def damage_subsets(playable: list[dict], energy: int, attacker_status: list, enemy: EnemyFact): + """Yield (cards, raw damage) for supported subsets, shortest first. + + Only plain direct attacks are modeled. A matching phrase inside a + conditional or a card with additional effects is not a supported attack. + """ + if not damage_context_supported(attacker_status, enemy): + return + candidates = [ + c for c in playable + if c.get("type") == "Attack" + and c.get("target_type") in ("AnyEnemy", "AllEnemies") + and fixed_energy_cost(c) is not None + and DAMAGE_RE.fullmatch((c.get("description") or "").strip().rstrip(".")) + ][:12] + for count in range(1, len(candidates) + 1): + for subset in itertools.combinations(candidates, count): + if sum(fixed_energy_cost(c) for c in subset) > energy: + continue + total = sum(damage_to_target(c, attacker_status, enemy_status_names(enemy)) for c in subset) + yield list(subset), total + + +def lethal_line(playable: list[dict], energy: int, attacker_status: list, + enemy: EnemyFact) -> list[dict] | None: + """Shortest supported direct-damage line against one enemy, if found.""" + if enemy.hp <= 0: + return None + for cards, total in damage_subsets(playable, energy, attacker_status, enemy): + if total >= enemy.effective_hp: + return sorted(cards, key=lambda c: -parse_card_damage(c).raw_total) + return None + + # -------------------------------------------------------------------------- # Combat facts # -------------------------------------------------------------------------- @@ -273,7 +336,7 @@ class CombatFacts: playable: list[dict] potions: list[dict] player_status: list - lethal_available: bool + lethal_available: bool | None killable: list[str] deck_counts: dict deck_summary: dict @@ -312,26 +375,17 @@ class CombatFacts: @property def max_block_available(self) -> int: - """Block obtainable from playable cards within the energy budget.""" - energy = self.energy - total = 0 - for card in sorted(self.playable, key=block_value, reverse=True): - value = block_value(card) - if value <= 0: - continue - cost = _as_int(card.get("cost")) - if cost <= energy: - total += value - energy -= cost - return total + """Maximum displayed immediate block in the fixed-cost model.""" + return sum(block_value(c) for c in best_block_line(self.playable, self.energy)) @property def survives_with_cards(self) -> bool: """ - True when existing block plus reachable card block covers the incoming - damage. Used to decide whether a potion is the ONLY way to survive. + Estimate survival after known incoming attacks and the block plan. + Survival can include HP loss. Card side effects are not simulated. """ - return self.block + self.max_block_available >= self.incoming_damage + remaining = max(0, self.incoming_damage - self.block - self.max_block_available) + return remaining < self.hp # --------------------------------------------------------------------- # Fight-length projection. @@ -346,15 +400,15 @@ class CombatFacts: @property def attack_power(self) -> int: - """Best single-turn damage the hand can produce, ignoring targets.""" + """Greedy damage estimate, ignoring targets, ordering, and unknown costs.""" energy = self.energy total = 0 for card in sorted(self.playable, key=lambda x: -parse_card_damage(x).raw_total): dmg = parse_card_damage(card).raw_total if dmg <= 0: continue - cost = _as_int(card.get("cost")) - if cost <= energy: + cost = fixed_energy_cost(card) + if cost is not None and cost <= energy: total += dmg energy -= cost return total @@ -362,11 +416,10 @@ class CombatFacts: @property def turns_to_kill(self) -> int: """ - Turns this fight still needs, from this turn's reachable damage. + Estimate remaining turns by assuming later hands resemble this hand. - Deliberately pessimistic: it assumes every later turn looks like this - one. If the hand is all block, the estimate is huge, which correctly - pushes toward "this fight will grind me down". + This ignores future draws and effects. A block-only hand produces a + large estimate; it does not establish the actual fight duration. """ power = self.attack_power if power <= 0: @@ -391,14 +444,12 @@ class CombatFacts: @property def must_block(self) -> bool: """ - True when the rest of this fight will cost more HP than we can spare. + Request current block when estimated future damage exceeds the HP budget. - This is the decision the bot was missing. It fires on the Kin turn - (74/80 HP, 12 incoming, ~12 turns to kill -> 144 projected vs 50 - affordable) and stays silent on a trivial fight (15 HP enemy, 6 - incoming, 2 turns -> 12 projected vs 50 affordable). + This is a heuristic. It must stop requesting block once the current + incoming damage is covered; ordinary block does not cover future turns. """ - if self.incoming_damage <= 0: + if self.unblocked_damage <= 0: return False return self.projected_incoming > self.affordable_loss @@ -409,7 +460,7 @@ class CombatFacts: Separate from `must_block` because it also covers the case where the current hit is lethal-ish regardless of how long the fight lasts. """ - if self.incoming_damage <= 0: + if self.unblocked_damage <= 0: return False return ( self.threat in (THREAT_SEVERE, THREAT_LETHAL) @@ -421,8 +472,8 @@ class CombatFacts: """ The compact, semantic state handed to Jev. - Deliberately contains NO raw numbers that Jev would have to compare. - Buckets and booleans only, so arithmetic never crosses the boundary. + Include computed comparisons so Jev need not perform arithmetic. + Preserve displayed costs, including X. Lethal can be unknown (None). """ return { "combat": { @@ -450,7 +501,7 @@ class CombatFacts: { "index": c.get("index"), "name": c.get("name"), - "cost": _as_int(c.get("cost")), + "cost": c.get("cost"), "type": c.get("type"), "targets": c.get("target_type"), "text": c.get("description"), @@ -510,15 +561,20 @@ def combat_facts(observation: dict) -> CombatFacts: playable = [c for c in hand if c.get("can_play")] # A card is playable only if we can also afford it. - affordable = [c for c in playable if _as_int(c.get("cost")) <= energy] + affordable = [c for c in playable + if fixed_energy_cost(c) is None or fixed_energy_cost(c) <= energy] killable: list[str] = [] for enemy in enemies: best = _subset_damage(affordable, energy, player_status, enemy) - if enemy.effective_hp > 0 and best >= enemy.effective_hp: + if enemy.hp > 0 and best >= enemy.effective_hp: killable.append(enemy.entity_id) - lethal = bool(enemies) and len(killable) == len(enemies) + # Individually killable enemies may require the SAME cards and energy. + # Do not claim joint lethal without a shared-resource search. + lethal = bool(killable) if len(enemies) == 1 else None + if len(enemies) == 1 and not damage_context_supported(player_status, enemies[0]): + lethal = None # draw/discard/exhaust piles expose only {name, cost, star_cost, description} # with no `type`, so composition is counted by card name for every pile. @@ -554,42 +610,13 @@ def combat_facts(observation: dict) -> CombatFacts: def _subset_damage(playable: list[dict], energy: int, attacker_status: list, enemy: EnemyFact) -> int: - """ - Exact max RAW damage over subsets, honouring energy and attacker statuses. - - Raw means before enemy block: compare the result against `enemy.effective_hp` - (hp + block) so the block is subtracted exactly once. Subtracting it here and - comparing against `effective_hp` required hp + 2*block, which silently - discarded real lethal lines whenever an enemy played Defend. - """ - best = 0 - n = min(len(playable), 12) - for r in range(1, n + 1): - for combo in itertools.combinations(range(n), r): - cost = sum(_as_int(playable[i].get("cost")) for i in combo) - if cost > energy: - continue - total = 0 - for i in combo: - card = playable[i] - dmg = parse_card_damage(card) - if dmg.raw_total == 0: - continue - per_hit = dmg.base + power_amount(attacker_status, "Strength") - if power_amount(attacker_status, "Weak"): - per_hit = int(per_hit * 0.75) - card_total = per_hit * dmg.hits - if power_amount(enemy_status_names(enemy), "Vulnerable"): - card_total = int(card_total * 1.5) - total += card_total - if total > best: - best = total - return best + """Maximum supported raw damage; compare with HP + block exactly once.""" + return max((total for _, total in damage_subsets(playable, energy, attacker_status, enemy)), default=0) def enemy_status_names(enemy: EnemyFact) -> list: - """Adapt the name list back into the shape power_amount expects.""" - return [{"name": n, "amount": 1} for n in enemy.statuses] + """Preserve observed amounts; accept legacy name-only EnemyFact values.""" + return enemy.status_amounts or [{"name": n, "amount": 1} for n in enemy.statuses] def deck_summary(player: dict) -> dict: diff --git a/jev.py b/jev.py index 7f6ff84..ec5bb6d 100644 --- a/jev.py +++ b/jev.py @@ -44,6 +44,7 @@ from __future__ import annotations import http.client import json +import math import os import time import urllib.error @@ -381,10 +382,20 @@ class JevClient: elapsed = time.monotonic() - started try: data = json.loads(raw) - except json.JSONDecodeError as exc: + except (json.JSONDecodeError, UnicodeDecodeError) as exc: raise JevError(f"non-JSON response: {raw[:200]!r}") from exc response = self._parse(data, elapsed) + if set(response.answers) != set(questions): + raise JevError("response question IDs do not match the request") + for qid, answer in response.answers.items(): + question = questions[qid] + if answer.kind != question.get("type"): + raise JevError(f"answer type does not match question {qid!r}") + if isinstance(answer, ChoiceAnswer) and set(answer.probabilities) != set(question["criteria"]): + raise JevError(f"answer options do not match question {qid!r}") + if isinstance(answer, ScoreAnswer) and len(answer.legend) != len(question["criteria"]): + raise JevError(f"answer levels do not match question {qid!r}") _trace({ "model": response.model, "latency_s": round(elapsed, 3), @@ -399,34 +410,69 @@ class JevClient: @staticmethod def _parse(data: dict, elapsed: float) -> JevResponse: + def number(value, field: str, *, probability: bool = False) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise JevError(f"missing or non-numeric {field}") + if not math.isfinite(value) or (probability and not 0 <= value <= 1): + raise JevError(f"invalid {field}: expected a finite {'probability' if probability else 'number'}") + return float(value) + + def probabilities(item: dict) -> dict[str, float]: + values = item.get("probabilities") + if not isinstance(values, dict) or not values: + raise JevError("missing or invalid probability distribution") + if not all(isinstance(k, str) for k in values): + raise JevError("probability keys must be strings") + return {k: number(v, f"probabilities[{k!r}]", probability=True) for k, v in values.items()} + + if not isinstance(data, dict): + raise JevError("expected a response object") + raw_answers = data.get("answers") + if not isinstance(raw_answers, dict) or not raw_answers: + raise JevError("missing or invalid answers object") answers: dict[str, Answer] = {} - for qid, item in (data.get("answers") or {}).items(): + for qid, item in raw_answers.items(): + if not isinstance(qid, str) or not isinstance(item, dict): + raise JevError("invalid question ID or answer object") kind = item.get("type") if kind == "noul": - answers[qid] = NoulAnswer(noul=float(item.get("noul", 0.0))) + answers[qid] = NoulAnswer(number(item.get("noul"), "noul", probability=True)) elif kind == "choice": + values = probabilities(item) + chosen = item.get("choice") + if not isinstance(chosen, str) or chosen not in values: + raise JevError(f"invalid choice for question {qid!r}") answers[qid] = ChoiceAnswer( - choice=item.get("choice"), - probabilities=item.get("probabilities") or {}, - confidence=float(item.get("confidence", 0.0)), + choice=chosen, probabilities=values, + confidence=number(item.get("confidence"), "confidence", probability=True), ) elif kind == "score": + values = probabilities(item) + legend = item.get("legend") + if not isinstance(legend, dict) or not 2 <= len(legend) <= 10: + raise JevError(f"invalid score legend for question {qid!r}") + levels = {str(i) for i in range(len(legend))} + if set(legend) != levels or set(values) != levels: + raise JevError(f"invalid score levels for question {qid!r}") + value = number(item.get("score"), "score") + if not 0 <= value <= len(legend) - 1: + raise JevError(f"score outside its levels for question {qid!r}") answers[qid] = ScoreAnswer( - score=float(item.get("score", 0.0)), - legend=item.get("legend") or {}, - probabilities=item.get("probabilities") or {}, - confidence=float(item.get("confidence", 0.0)), + score=value, legend=legend, probabilities=values, + confidence=number(item.get("confidence"), "confidence", probability=True), ) else: raise JevError(f"unknown answer type {kind!r} for question {qid!r}") - usage = data.get("usage") or {} + usage = data.get("usage", {}) + if not isinstance(usage, dict): + raise JevError("invalid token usage object") + counts = [usage.get(key, 0) for key in ("input_tokens", "output_tokens")] + if any(isinstance(n, bool) or not isinstance(n, int) or n < 0 for n in counts): + raise JevError("invalid token usage counts") return JevResponse( - answers=answers, - model=data.get("model", "?"), - input_tokens=int(usage.get("input_tokens", 0)), - output_tokens=int(usage.get("output_tokens", 0)), - latency_s=elapsed, + answers=answers, model=data.get("model", "?"), + input_tokens=counts[0], output_tokens=counts[1], latency_s=elapsed, ) diff --git a/run.py b/run.py index 7eee869..1bd3302 100644 --- a/run.py +++ b/run.py @@ -7,7 +7,7 @@ every later index, so we re-read the state after every single action. We never precompute an action list. usage: - python3 run.py --dry-run --steps 5 # show decisions, touch nothing + python3 run.py --dry-run --steps 5 # no action POSTs; reads/model/logs still run python3 run.py --steps 40 # actually play python3 run.py --steps 40 --no-jev # heuristics only """ @@ -17,6 +17,7 @@ from __future__ import annotations import argparse import atexit import json +import math import os import pathlib import sys @@ -188,7 +189,7 @@ def observe() -> dict: return sts2.state() -def preflight(obs: dict) -> str | None: +def preflight(obs: dict, *, dry_run: bool = False) -> str | None: """ Detect states the bot cannot proceed from, so a session does not silently burn its whole step budget doing nothing. @@ -200,8 +201,12 @@ def preflight(obs: dict) -> str | None: * A parked `game_over` screen blocks every later session. Dismiss it. """ if obs.get("state_type") == "game_over": + if dry_run: + return None try: - sts2.act("menu_select", option="main_menu") + result = sts2.act("menu_select", option="main_menu") + if not result.ok: + return f"BLOCKED: game-over dismissal rejected: {result.message}" # The dismissal is not instant. Without this wait the loop reads the # state again, still sees game_over, and stops the session at step 1 # -- measured, one whole session made 0 decisions. @@ -237,7 +242,9 @@ def preflight(obs: dict) -> str | None: def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--steps", type=int, default=20) - ap.add_argument("--dry-run", action="store_true") + ap.add_argument("--dry-run", action="store_true", + help="send no action POSTs and do not update deck.json; " + "state reads, model calls, and capture logs still run") ap.add_argument("--no-jev", action="store_true") ap.add_argument("--pause", type=float, default=0.6, help="seconds to wait after each action") @@ -259,6 +266,12 @@ def main() -> int: ap.add_argument("--card-skip-policy", choices=("jev", "combined"), default=None, help="how card-reward skips are decided (default: brain's own)") args = ap.parse_args() + if args.steps < 1 or args.max_jev_errors < 1 or args.max_duplicate_waits < 0: + ap.error("steps and max-jev-errors must be positive; max-duplicate-waits must be nonnegative") + if not math.isfinite(args.pause) or args.pause < 0: + ap.error("pause must be finite and nonnegative") + if not math.isfinite(args.stuck_seconds) or args.stuck_seconds <= 0: + ap.error("stuck-seconds must be finite and positive") if not sts2.is_up(): print(f"game not reachable at {sts2.BASE}") @@ -267,7 +280,7 @@ def main() -> int: # Fail fast on a state the bot cannot leave, instead of burning the whole # step budget on rejected actions. try: - blocker = preflight(observe()) + blocker = preflight(observe(), dry_run=args.dry_run) except sts2.Sts2Error as exc: print(f"cannot read state: {exc}") return 1 @@ -287,10 +300,11 @@ def main() -> int: client = RecordingClient(JevClient()) print(f"jev ready: {client!r}") except JevError as exc: - print(f"jev unavailable, using heuristics only: {exc}") + print(f"jev unavailable: {exc}; use --no-jev for an intentional heuristic session") + return 3 capdir = pathlib.Path(args.capture_dir) - capdir.mkdir(exist_ok=True) + capdir.mkdir(parents=True, exist_ok=True) trace_path = capdir / "decisions.jsonl" def trace(record: dict) -> None: @@ -302,8 +316,6 @@ def main() -> int: fh.write(json.dumps(record, sort_keys=True, default=str) + "\n") stats = {"code": 0, "jev": 0, "fallback": 0} - jev_calls = 0 - jev_tokens = 0 started = time.monotonic() # The outcome half of the join: which run record(s) this session produced. @@ -311,6 +323,8 @@ def main() -> int: # step cap, a stuck screen, a model-failure abort, or an exception. history_before = history_snapshot() step = 0 + exit_code = 1 + stop_reason = "interrupted" started_at = time.strftime("%Y-%m-%dT%H:%M:%S") def write_session_row() -> None: @@ -318,11 +332,21 @@ def main() -> int: row = session_record(SESSION_ID, started_at, time.strftime("%Y-%m-%dT%H:%M:%S"), step, stats, history_snapshot() - history_before) + row.update(exit_code=exit_code, stop_reason=stop_reason, dry_run=args.dry_run) with (capdir / "sessions.jsonl").open("a", encoding="utf-8") as fh: fh.write(json.dumps(row, sort_keys=True, default=str) + "\n") except OSError: pass # bookkeeping must never break the exit + def finish(code: int, reason: str) -> int: + nonlocal exit_code, stop_reason + exit_code, stop_reason = code, reason + trace({"event": "session_end", "step": step, "exit_code": code, "reason": reason}) + elapsed = time.monotonic() - started + print(f"steps={step} waits={waits} elapsed={elapsed:.1f}s sources={stats} " + f"stop={reason} exit={code}") + return code + atexit.register(write_session_row) # The card_reward state does not expose the deck, but combat states expose @@ -347,7 +371,7 @@ def main() -> int: obs = observe() except sts2.Sts2Error as exc: print(f"[{step:03d}] state read failed: {str(exc)[:160]}") - return 1 + return finish(1, "state_error") st = obs.get("state_type") @@ -360,22 +384,30 @@ def main() -> int: # first left the game parked on `game_over`, so every later session saw # it at step 1 and stopped instantly -- the whole A/B produced nothing. if args.stop_on_run_end and st == "game_over": + if args.dry_run: + print(f"[{step:03d}] dry run: would dismiss game-over") + return finish(0, "run_end_preview") print(f"[{step:03d}] run ended; dismissing game-over, then stopping") try: - sts2.act("menu_select", option="main_menu") + result = sts2.act("menu_select", option="main_menu") except sts2.Sts2Error as exc: print(f"[{step:03d}] could not dismiss game-over: {exc}") + return finish(1, "dismiss_error") + if not result.ok: + print(f"[{step:03d}] game-over dismissal rejected: {result.message}") + return finish(1, "dismiss_rejected") time.sleep(args.pause) - break + return finish(0, "run_ended") if args.stop_on_run_end and st in ("monster", "elite", "boss", "map", "rewards", "card_reward", "event", "rest_site", "shop", "treasure", - "card_select", "hand_select"): + "card_select", "hand_select", "bundle_select", + "relic_select", "crystal_sphere", "fake_merchant"): saw_a_run = True if args.stop_on_run_end and saw_a_run and st == "menu" and \ obs.get("menu_screen") == "main": print(f"[{step:03d}] back at the main menu; run is over, stopping session") - break + return finish(0, "run_ended") # Guard against re-acting while the game is still animating a transition. # An identical state after our own action means the action has not landed @@ -394,7 +426,7 @@ def main() -> int: if unchanged_for > args.stuck_seconds: print(f"[{step:03d}] STUCK: state unchanged for {unchanged_for:.0f}s -- stopping") print(json.dumps(obs, indent=2)[:900]) - break + return finish(1, "stuck") # Only wait when our own action actually landed and the game is still # animating. If the action was rejected, fall through and pick a @@ -416,7 +448,8 @@ def main() -> int: # counts decide whether the deck actually changed. if f.deck_counts and f.deck_counts != (deck_snapshot or {}).get("counts"): deck_snapshot = {"counts": f.deck_counts, "summary": f.deck_summary} - save_deck(deck_snapshot) + if not args.dry_run: + save_deck(deck_snapshot) decision = None decide_error = None @@ -426,37 +459,45 @@ def main() -> int: client.last = None try: decision = brain.decide(obs, client, deck_snapshot) - jev_errors = 0 + # A procedural action or animation wait does not establish model + # recovery. Only a successful model response resets the budget. + if client is not None and client.last is not None: + jev_errors = 0 except JevError as exc: jev_errors += 1 print(f"[{step:03d}] jev error ({jev_errors}): {str(exc)[:140]}") decide_error = f"JevError: {str(exc)[:160]}" + trace({"step": step, "event": "model_error", "error": decide_error, + "consecutive_failures": jev_errors}) if jev_errors >= args.max_jev_errors: print(f"[{step:03d}] ABORT: {jev_errors} consecutive model failures. " f"The run would continue on heuristics alone, which is not " f"the data we want. Retry this session.") - return 3 + return finish(3, "model_failures") # Fall back WITHOUT the model. Passing the client again just retries # the same failing request -- measured, a DNS blip re-raised out of # the "fallback" and killed the session. - decision = brain.simple_decision(obs, None, deck_snapshot) + try: + decision = brain.decide(obs, None, deck_snapshot) + except Exception as inner: # noqa: BLE001 + trace({"step": step, "event": "fallback_error", "error": str(inner)[:160]}) + print(f"[{step:03d}] fallback also failed: {inner}") + return finish(1, "fallback_error") except Exception as exc: # noqa: BLE001 - # A network blip or an unexpected shape must not end the run. + # Unexpected policy failures are bugs, not evidence that a blind + # fallback is safe. Stop and retain the error for diagnosis. print(f"[{step:03d}] unexpected error in decide(): " f"{type(exc).__name__}: {str(exc)[:160]}") - decide_error = f"{type(exc).__name__}: {str(exc)[:160]}" - try: - decision = brain.simple_decision(obs, None, deck_snapshot) - except Exception as inner: # noqa: BLE001 - print(f"[{step:03d}] fallback also failed: {inner}") - decision = None + trace({"step": step, "event": "policy_error", + "error": f"{type(exc).__name__}: {str(exc)[:160]}"}) + return finish(1, "policy_error") if decision is None: trace({"step": step, "state_type": st, "event": "no_decision", "error": decide_error}) print(f"[{step:03d}] no decision for state_type={st!r} -- stopping") print(json.dumps(obs, indent=2)[:800]) - break + return finish(1, "no_decision") # Between turns there is nothing to do but look again. if decision.action == "__wait__": @@ -500,7 +541,7 @@ def main() -> int: print(f"[{step:03d}] action failed: {str(exc)[:200]}") trace({"step": step, "event": "action_error", "action": decision.action, "error": str(exc)[:200]}) - break + return finish(1, "action_error") if not result.ok: print(f"[{step:03d}] action rejected: {result.message}") @@ -511,7 +552,7 @@ def main() -> int: if rejected >= 6: print(f"[{step:03d}] STUCK: {rejected} consecutive rejections -- stopping") print(json.dumps(obs, indent=2)[:900]) - break + return finish(1, "action_rejections") # Transient rejections while the game animates are normal -- a # rest-site `proceed` right after a heal is rejected for a moment # and then succeeds. Back off longer than the usual pause. @@ -523,10 +564,11 @@ def main() -> int: time.sleep(args.pause) - elapsed = time.monotonic() - started - print() - print(f"steps={step} waits={waits} elapsed={elapsed:.1f}s sources={stats}") - return 0 + if rejected: + return finish(1, "action_rejections") + if args.stop_on_run_end and not args.dry_run: + return finish(4, "step_limit_before_run_end") + return finish(0, "step_limit") if __name__ == "__main__": diff --git a/sts2.py b/sts2.py index c290eff..ba0bfcd 100644 --- a/sts2.py +++ b/sts2.py @@ -18,6 +18,7 @@ Never precompute an action list. from __future__ import annotations +import http.client import json import urllib.error import urllib.request @@ -78,7 +79,7 @@ class ActionResult: return self.status == "ok" -def _request(path: str, payload: dict | None = None, timeout: float = TIMEOUT) -> dict: +def _read(path: str, payload: dict | None = None, timeout: float = TIMEOUT) -> bytes: url = BASE + path if payload is None: request = urllib.request.Request(url, method="GET") @@ -93,17 +94,30 @@ def _request(path: str, payload: dict | None = None, timeout: float = TIMEOUT) - with urllib.request.urlopen(request, timeout=timeout) as response: body = response.read() except urllib.error.HTTPError as exc: - detail = exc.read().decode("utf-8", "replace") + try: + detail = exc.read().decode("utf-8", "replace") + except (OSError, http.client.HTTPException): + detail = "could not read the error body" raise Sts2Error(f"HTTP {exc.code} on {path}: {detail[:300]}") from exc except urllib.error.URLError as exc: raise Sts2Error( f"cannot reach the game on {BASE}. Is STS2 running with the mod loaded? ({exc})" ) from exc + except (OSError, http.client.HTTPException) as exc: + # A failed POST may already have reached the game. Never retry it here. + raise Sts2Error(f"transport failure on {path}: {type(exc).__name__}: {exc}") from exc + return body + +def _request(path: str, payload: dict | None = None, timeout: float = TIMEOUT) -> dict: + body = _read(path, payload, timeout) try: - return json.loads(body) - except json.JSONDecodeError as exc: + data = json.loads(body) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: raise Sts2Error(f"non-JSON response from {path}: {body[:200]!r}") from exc + if not isinstance(data, dict): + raise Sts2Error(f"expected a JSON object from {path}, got {type(data).__name__}") + return data # -------------------------------------------------------------------------- @@ -121,9 +135,11 @@ def is_up() -> bool: def state(fmt: str = "json") -> dict: """Current game state. fmt: 'json' or 'markdown' (markdown returns text).""" if fmt == "markdown": - req = urllib.request.Request(f"{BASE}/api/v1/singleplayer?format=markdown") - with urllib.request.urlopen(req, timeout=TIMEOUT) as response: - return {"markdown": response.read().decode("utf-8")} + body = _read("/api/v1/singleplayer?format=markdown") + try: + return {"markdown": body.decode("utf-8")} + except UnicodeDecodeError as exc: + raise Sts2Error("invalid UTF-8 markdown response") from exc return _request("/api/v1/singleplayer?format=json") diff --git a/test_brain.py b/test_brain.py index b6e59e4..289818a 100644 --- a/test_brain.py +++ b/test_brain.py @@ -1075,16 +1075,37 @@ check("...and the executor finds the line", bool(brain._lethal_line(bf.playable, bf.energy, bf.player_status, bf.enemies[0])), True) -strong = combat(hand=[card(index=n) for n in range(2)], +strong = combat(hand=[card(index=n, desc="Deal 12 damage.") for n in range(2)], enemies=[enemy("E_0", hp=18, intent=0)], status=[{"name": "Strength", "amount": 6, "type": "Buff"}]) sf = facts.combat_facts(strong) -check("the executor sees the player's Strength", +check("the executor uses Strength-adjusted hand text", [c["name"] for c in (brain._lethal_line(sf.playable, sf.energy, sf.player_status, sf.enemies[0]) or [])], ["Strike", "Strike"]) -check("...where an empty status list finds nothing", - brain._lethal_line(sf.playable, sf.energy, [], sf.enemies[0]), None) +check("...and does not need to add Strength again", + bool(brain._lethal_line(sf.playable, sf.energy, [], sf.enemies[0])), True) + +print() +print("=== correctness: shared damage and block plans ===") + +adjusted = combat(hand=[card(index=5, desc="Deal 8 damage.")], + enemies=[enemy("E_0", hp=9, intent=0)], + 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) + +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"), + card("Small Block", index=11, desc="Gain 6 Block.", ctype="Skill", target="Self")] +blocking = combat(hand=defenders, enemies=[enemy("E_0", hp=100, intent=12)], hp=30) +blocking["player"]["energy"] = 2 +bd = brain.decide(blocking, None) +check("forced defense starts a maximum-block plan", bd.params.get("card_index") in (8, 11), True) +covered = combat(hand=[card("Defend", index=5, desc="Gain 5 Block.", ctype="Skill", target="Self"), + card(index=8)], enemies=[enemy("E_0", hp=100, intent=12)], hp=30, block=12) +check("fully blocked incoming does not force Defend", brain.decide(covered, None).params.get("card_index"), 8) print() print(f"=== {PASS} passed, {FAIL} failed ===") diff --git a/test_facts.py b/test_facts.py index a904ab3..8df5c02 100644 --- a/test_facts.py +++ b/test_facts.py @@ -11,6 +11,7 @@ Run: python3 test_facts.py from __future__ import annotations +import gzip import json import pathlib import sys @@ -133,12 +134,18 @@ plain = card("Defend", 1, "Gain 5 Block.", ctype="Skill", target="Self") check("non-damage card", facts.parse_card_damage(plain).raw_total, 0) print() -print("=== 3. modifiers ===") +print("=== 3. hand descriptions include attacker modifiers ===") strike = card("Strike", 1, "Deal 6 damage.") check("bare strike", facts.damage_to_target(strike, [], [], 0), 6) -check("+2 Strength", facts.damage_to_target(strike, [{"name": "Strength", "amount": 2}], [], 0), 8) +strong_strike = card("Strike", 1, "Deal 8 damage.") +check("Strength is already in hand text", facts.damage_to_target( + strong_strike, [{"name": "Strength", "amount": 2}], [], 0), 8) check("target Vulnerable (x1.5)", facts.damage_to_target(strike, [], [{"name": "Vulnerable", "amount": 2}], 0), 9) -check("attacker Weak (x0.75)", facts.damage_to_target(strike, [{"name": "Weak", "amount": 1}], [], 0), 4) +weak_strike = card("Strike", 1, "Deal 4 damage.") +check("Weak is already in hand text", facts.damage_to_target( + weak_strike, [{"name": "Weak", "amount": 1}], [], 0), 4) +check("Vulnerable rounds each hit, not the combined total", facts.damage_to_target( + multi, [], [{"name": "Vulnerable", "amount": 1}], 0), 8) check("enemy block absorbed", facts.damage_to_target(strike, [], [], 4), 2) check("block exceeds damage", facts.damage_to_target(strike, [], [], 10), 0) @@ -230,21 +237,73 @@ check("18 raw vs 20 hp + 5 block is not lethal", s.lethal_available, False) check("...and nothing is listed killable", s.killable, []) print() -print("=== 9. player statuses reach the lethal search (regression) ===") -# The executor used to call the search with `[]`, so facts could report -# `lethal_available: true` while the code meant to execute the kill found -# nothing -- Strength and Weak were invisible to it. -strong_hand = [card("Strike", 1, "Deal 6 damage.", index=n) for n in range(2)] +print("=== 9. lethal calculations use displayed damage once ===") +# Hand text already includes Strength; statuses remain available as context. +strong_hand = [card("Strike", 1, "Deal 12 damage.", index=n) for n in range(2)] strong = observation(strong_hand, [enemy("E_0", 18)], status=[{"name": "Strength", "amount": 6, "type": "Buff"}]) st = facts.combat_facts(strong) check("Strength is carried on the facts", st.player_status[0]["name"], "Strength") check("12+12 raw damage vs 18 hp is lethal", st.lethal_available, True) -check("ignoring the statuses misses it", - facts._subset_damage(st.playable, st.energy, [], st.enemies[0]), 12) -check("using them finds it", +check("displayed damage works without reapplying statuses", + facts._subset_damage(st.playable, st.energy, [], st.enemies[0]), 24) +check("passing statuses must not apply Strength twice", facts._subset_damage(st.playable, st.energy, st.player_status, st.enemies[0]), 24) +print() +print("=== 10. captured hand text is not base damage ===") +fixture = pathlib.Path("dataset/states/06/0618279b1a19a34c32469d654d00386fa904e76fc1cf8ee0943d47469d257a49.json.gz") +captured = json.loads(gzip.decompress(fixture.read_bytes())) +cp = captured["player"] +cs = next(c for c in cp["hand"] if c["name"] == "Strike") +check("captured Strength", facts.power_amount(cp["status"], "Strength"), 2) +check("captured Strike text", cs["description"], "Deal 8 damage.") +check("captured damage without target modifiers", + facts.damage_to_target(cs, cp["status"], [], 0), 8) +cf = facts.combat_facts(observation([cs], [enemy("E_0", 9)], energy=1, status=cp["status"])) +check("8 displayed damage cannot kill 9 HP", cf.lethal_available, False) + +print() +print("=== 11. individual killability does not prove joint lethal ===") +two = facts.combat_facts(observation([strike], [enemy("E_0", 6), enemy("E_1", 6)], energy=1)) +check("each target is individually killable", two.killable, ["E_0", "E_1"]) +check("joint lethal is unknown until a shared-resource search exists", two.lethal_available, None) +check("the model receives unknown, not a false proof", two.to_state()["combat"]["lethal_available"], None) + +print() +print("=== 12. fixed-cost block optimization and covered damage ===") +block_hand = [card("Big Block", 2, "Gain 9 Block.", ctype="Skill", target="Self", index=0), + card("Small Block", 1, "Gain 6 Block.", ctype="Skill", target="Self", index=1), + card("Small Block", 1, "Gain 6 Block.", ctype="Skill", target="Self", index=2)] +bf = facts.combat_facts(observation(block_hand, [enemy("E_0", 100, intent_damage=12)], energy=2)) +check("two small blocks beat the largest card", bf.max_block_available, 12) +free = card("Free Block", 0, "Gain 3 Block.", ctype="Skill", target="Self", index=3) +ff = facts.combat_facts(observation([free] + block_hand, [enemy("E_0", 100)], energy=2)) +check("zero-cost block is counted once", ff.max_block_available, 15) +covered = facts.combat_facts(observation(block_hand, [enemy("E_0", 100, intent_damage=12)], + hp=30, block=12)) +check("covered damage does not force block", covered.block_urgent, False) +check("covered damage does not demand more current block", covered.must_block, False) +partial = facts.combat_facts(observation([plain], [enemy("E_0", 100, intent_damage=12)], hp=10)) +check("survival allows nonlethal HP loss", partial.survives_with_cards, True) +exact = facts.combat_facts(observation([plain], [enemy("E_0", 100, intent_damage=15)], hp=10)) +check("zero HP is not survival", exact.survives_with_cards, False) + +print() +print("=== 13. unsupported mechanics do not yield deterministic lethal lines ===") +x_card = card("Variable Attack", "X", "Deal 20 damage.") +xf = facts.combat_facts(observation([x_card], [enemy("E_0", 10)], energy=1)) +check("X cost is not a free card in damage search", + facts._subset_damage(xf.playable, xf.energy, [], xf.enemies[0]), 0) +check("X cost remains visible to the model", xf.to_state()["combat"]["hand"][0]["cost"], "X") +conditional = card("Conditional", 1, "If the enemy is Vulnerable, deal 20 damage.") +cc = facts.combat_facts(observation([conditional], [enemy("E_0", 10)])) +check("conditional text is not unconditional damage", + facts._subset_damage(cc.playable, cc.energy, [], cc.enemies[0]), 0) +armored = facts.combat_facts(observation([strike], [enemy("E_0", 6, + status=[{"name": "Intangible", "amount": 1}])])) +check("unsupported target powers do not produce a killable target", armored.killable, []) + print() print(f"=== {PASS} passed, {FAIL} failed ===") sys.exit(1 if FAIL else 0) diff --git a/test_run.py b/test_run.py index 6b41e1f..0dbb496 100755 --- a/test_run.py +++ b/test_run.py @@ -18,11 +18,15 @@ Run: python3 test_run.py from __future__ import annotations +import contextlib +import io +import itertools import json import pathlib import sys import tempfile import types +from unittest.mock import patch import brain import jev @@ -81,9 +85,14 @@ class FakeSts2: class Sts2Error(RuntimeError): pass - def __init__(self, states): + BASE = "offline://game" + + def __init__(self, states, *, action_ok=True, action_error=None): self.states = states self.i = 0 + self.actions = [] + self.action_ok = action_ok + self.action_error = action_error def is_up(self) -> bool: return True @@ -91,10 +100,46 @@ class FakeSts2: def state(self) -> dict: state = self.states[min(self.i, len(self.states) - 1)] self.i += 1 + if isinstance(state, Exception): + raise state return state def act(self, *a, **k): - return types.SimpleNamespace(ok=True, message="") + self.actions.append((a, k)) + if self.action_error: + raise self.action_error + return types.SimpleNamespace(ok=self.action_ok, message="rejected" if not self.action_ok else "") + + +def invoke(fake, *flags, client=None, clock=None, decide=None): + """Run the real loop with isolated files, no delays, and no live services.""" + callbacks = [] + brain._reset_screen_guards("test-run-reset") + with tempfile.TemporaryDirectory() as directory, contextlib.ExitStack() as stack: + capdir = pathlib.Path(directory) / "capture" + stack.enter_context(patch.object(run, "sts2", fake)) + stack.enter_context(patch.object(run, "JevClient", return_value=client or StubClient())) + stack.enter_context(patch.object(run, "history_snapshot", return_value=set())) + stack.enter_context(patch.object(run, "load_deck", return_value=None)) + save = stack.enter_context(patch.object(run, "save_deck")) + stack.enter_context(patch.object(run.atexit, "register", side_effect=callbacks.append)) + stack.enter_context(patch.object(run.time, "sleep")) + stack.enter_context(patch.object(sys, "argv", ["run.py", "--steps", "10", "--pause", "0", + "--capture-dir", str(capdir), *flags])) + if clock is not None: + stack.enter_context(patch.object(run.time, "monotonic", side_effect=clock)) + if decide is not None: + stack.enter_context(patch.object(brain, "decide", side_effect=decide)) + output = stack.enter_context(contextlib.redirect_stdout(io.StringIO())) + rc = run.main() + for callback in callbacks: + callback() + def rows(name): + path = capdir / name + return [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else [] + return types.SimpleNamespace(rc=rc, rows=rows("decisions.jsonl"), + sessions=rows("sessions.jsonl"), saved=save.call_count, + output=output.getvalue()) print("=== 1. answer_record shapes (what the log stores) ===") @@ -138,21 +183,10 @@ card_reward = { menu = {"state_type": "menu", "menu_screen": "main", "options": ["singleplayer", "quit"], "run": None} -tmp = pathlib.Path(tempfile.mkdtemp()) -real_sts2, real_client = run.sts2, run.JevClient -# preflight() reads one state before the loop starts, so the sequence leads with -# a menu: preflight consumes that, step 1 sees the card_reward. -run.sts2 = FakeSts2([menu, card_reward, menu, menu]) -run.JevClient = StubClient -argv = sys.argv -sys.argv = ["run.py", "--steps", "2", "--dry-run", "--capture-dir", str(tmp)] -try: - rc = run.main() -finally: - run.sts2, run.JevClient, sys.argv = real_sts2, real_client, argv - -check("main() completed", rc, 0) -rows = [json.loads(line) for line in (tmp / "decisions.jsonl").read_text().splitlines()] +# Preflight consumes the menu; the two loop steps see the reward and menu. +result = invoke(FakeSts2([menu, card_reward, menu]), "--steps", "2", "--dry-run") +check("main() completed", result.rc, 0) +rows = [r for r in result.rows if r["event"] == "decide"] check("one row per decided action", len(rows), 2) check("row 1 came from the model", rows[0]["source"], "jev") check("...and carries its answers", sorted(rows[0]["jev"]["answers"]), @@ -166,17 +200,96 @@ check("every row carries the session", {r["session"] for r in rows}, {run.SESSIO print() print("=== 4. the session row maps to an outcome (join half) ===") -# Real run files if this machine has them; skipped otherwise. -if run.HISTORY_DIR.exists(): - names = sorted(run.history_snapshot()) - check("history records visible", len(names) > 0, True) - if names: - outcome = run.run_outcome(names[-1]) - check("outcome names the file", outcome["file"], names[-1]) - check("outcome has the killer", "killed_by" in outcome, True) - check("outcome has the deck size", isinstance(outcome["deck_size"], int), True) -else: - print(" (no history directory on this machine, skipped)") +with tempfile.TemporaryDirectory() as directory, patch.object(run, "HISTORY_DIR", pathlib.Path(directory)): + record = {"win": False, "killed_by_encounter": "ENCOUNTER.TEST", "seed": "fixture", + "players": [{"deck": ["Strike"]}], "map_point_history": [[{}, {}]], "run_time": 42} + (run.HISTORY_DIR / "fixture.run").write_text(json.dumps(record)) + check("history records visible", run.history_snapshot(), {"fixture.run"}) + outcome = run.run_outcome("fixture.run") + check("outcome names the file", outcome["file"], "fixture.run") + check("outcome has the killer", outcome["killed_by"], "TEST") + check("outcome has the deck size", outcome["deck_size"], 1) + +print() +print("=== 5. dry-run never sends game actions or updates the deck cache ===") +game_over = {"state_type": "game_over"} +combat = { + "state_type": "monster", "run": {"act": 1, "floor": 1}, + "battle": {"round": 1, "turn": "player", "is_play_phase": True, + "enemies": [{"entity_id": "E_0", "name": "E", "hp": 100, "max_hp": 100, + "block": 0, "status": [], "intents": []}]}, + "player": {"hp": 80, "max_hp": 80, "energy": 1, "block": 0, "status": [], "potions": [], + "hand": [{"index": 0, "name": "Strike", "type": "Attack", "cost": "1", + "description": "Deal 6 damage.", "target_type": "AnyEnemy", "can_play": True}]}, +} +for states, extra, label in [ + ([game_over, menu], [], "parked game-over preflight"), + ([menu, game_over], ["--stop-on-run-end"], "run-end dismissal"), + ([menu, game_over], [], "ordinary game-over decision"), + ([menu, combat], [], "combat action and deck snapshot"), +]: + fake = FakeSts2(states) + result = invoke(fake, "--dry-run", "--no-jev", "--steps", "1", *extra) + check(label + ": no POST", fake.actions, []) + check(label + ": completed preview", result.rc, 0) + check(label + ": no deck write", result.saved, 0) + +print() +print("=== 6. failed sessions do not return success ===") +for fake, flags, expected, label in [ + (FakeSts2([game_over], action_ok=False), [], 2, "preflight rejection"), + (FakeSts2([menu, game_over], action_ok=False), ["--stop-on-run-end"], 1, "run-end rejection"), + (FakeSts2([menu, game_over], action_error=FakeSts2.Sts2Error("timeout")), ["--stop-on-run-end"], 1, "run-end timeout"), + (FakeSts2([menu, {"state_type": "not-supported"}]), [], 1, "no decision"), + (FakeSts2([menu, menu], action_error=FakeSts2.Sts2Error("timeout")), [], 1, "action transport failure"), + (FakeSts2([menu, menu], action_ok=False), ["--steps", "6"], 1, "rejection budget"), + (FakeSts2([menu, menu], action_ok=False), ["--steps", "1"], 1, "last action rejected at step limit"), + (FakeSts2([menu, FakeSts2.Sts2Error("read timeout")]), [], 1, "state transport failure"), + (FakeSts2([menu, combat]), ["--stop-on-run-end", "--steps", "1"], 4, "unfinished run at step limit"), +]: + result = invoke(fake, "--no-jev", *flags) + check(label, result.rc, expected) + if result.sessions: + check(label + ": session carries status", result.sessions[0].get("exit_code"), expected) + check(label + ": session has a reason", bool(result.sessions[0].get("stop_reason")), True) + +result = invoke(FakeSts2([menu, {"state_type": "overlay"}]), "--no-jev", "--stuck-seconds", "1.5", + clock=itertools.count().__next__) +check("unchanged-state timeout is a failure", result.rc, 1) +check("unchanged-state timeout is recorded", result.sessions[0].get("stop_reason"), "stuck") +result = invoke(FakeSts2([menu, menu]), "--no-jev", decide=ValueError("bad observation")) +check("unexpected policy error stops instead of hiding the bug", result.rc, 1) + +print() +print("=== 7. combat outages use the runner's fallback and failure budget ===") +class OutageClient(StubClient): + def ask(self, *args, **kwargs): + self.calls += 1 + raise jev.JevError("offline injected outage") + +client = OutageClient() +fake = FakeSts2([menu, combat, {"state_type": "overlay"}, combat]) +result = invoke(fake, "--max-jev-errors", "2", client=client) +check("abort after two model failures, even with a code-only wait between", result.rc, 3) +check("the model was tried twice", client.calls, 2) +check("first failed call gets one heuristic combat action", len(fake.actions), 1) +decisions = [r for r in result.rows if r["event"] == "decide"] +check("combat fallback is recorded", decisions[0]["source"] if decisions else None, "fallback") +check("combat fallback preserves its error", "JevError" in (decisions[0]["error"] or "") if decisions else False, True) +check("terminal model failure is recorded", len([r for r in result.rows if r["event"] == "model_error"]), 2) +check("model abort carries a session stop reason", result.sessions[0].get("stop_reason"), "model_failures") + +class RecoveringClient(StubClient): + def ask(self, *args, **kwargs): + if self.calls == 1: + return super().ask(*args, **kwargs) + self.calls += 1 + raise jev.JevError("offline injected outage") + +client = RecoveringClient() +result = invoke(FakeSts2([menu, combat]), "--max-jev-errors", "2", client=client) +check("successful model call resets the failure budget", client.calls, 4) +check("two failures after recovery abort", result.rc, 3) print() print(f"=== {PASS} passed, {FAIL} failed ===") diff --git a/utils/audit_prototype.py b/utils/audit_prototype.py new file mode 100644 index 0000000..ff87527 --- /dev/null +++ b/utils/audit_prototype.py @@ -0,0 +1,182 @@ +#!/usr/bin/env python3 +"""Read-only, offline diagnostics for the prototype; not a pass/fail test suite. + +Run: python3 utils/audit_prototype.py +Prints synthetic probes and a snapshot replay of dataset/. No game, model, +credentials, or game history are accessed. Findings describe current behavior; +no action-quality or win-rate claim follows from this replay. +""" +from __future__ import annotations + +import contextlib +import gzip +import io +import json +import pathlib +import sys +from collections import Counter + +ROOT = pathlib.Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +import brain +import facts as F +import jev + + +def card(index: int, *, damage=6, block=0, cost=1) -> dict: + return { + "index": index, "id": f"C{index}", + "name": "Defend" if block else "Strike", + "type": "Skill" if block else "Attack", "cost": str(cost), + "description": f"Gain {block} Block." if block else f"Deal {damage} damage.", + "target_type": "Self" if block else "AnyEnemy", "can_play": True, + "rarity": "Basic", "is_upgraded": False, + } + + +def enemy(name="E0", hp=100, incoming=12) -> dict: + return { + "entity_id": name, "name": name, "hp": hp, "max_hp": hp, + "block": 0, "status": [], + "intents": [{"type": "Attack", "label": str(incoming)}], + } + + +def observation(cards: list, *, enemies=None, energy=1, hp=80, block=0) -> dict: + return { + "state_type": "monster", + "battle": {"round": 1, "turn": "player", "is_play_phase": True, + "enemies": [enemy()] if enemies is None else enemies}, + "player": {"hp": hp, "max_hp": 80, "block": block, "energy": energy, + "max_energy": 3, "hand": cards, "status": [], "potions": []}, + } + + +def reset_policy() -> None: + # Each stored observation is independent. These are not ordered trajectories. + brain._reset_screen_guards("audit-reset") + + +def synthetic_probes() -> dict: + report = {} + obs = observation([card(0)], enemies=[enemy("E0", 6, 0), enemy("E1", 6, 0)]) + facts = F.combat_facts(obs) + report["shared_energy_lethal"] = { + "reported": facts.lethal_available, "killable": facts.killable, + "expected_report": None, "actual_can_kill_all": False, + "case": "One single-target Strike, one energy, two enemies with six HP each; joint search is not implemented.", + } + + obs = observation([card(0, block=5), card(1)], hp=30, block=12) + facts = F.combat_facts(obs) + decision = brain.decide(obs, None) + report["already_covered_defense"] = { + "unblocked": facts.unblocked_damage, "decision": vars(decision), + "case": "Incoming damage is fully blocked; Defend has no other effect.", + } + + obs = observation([card(0, block=9, cost=2), card(1, block=6), card(2, block=6)], energy=2) + report["greedy_block_not_maximum"] = { + "reported": F.combat_facts(obs).max_block_available, "reachable": 12, + } + + class BrokenClient: + def ask(self, *args, **kwargs): + raise jev.JevError("offline injected outage") + + obs = observation([card(0)], enemies=[enemy(incoming=0)]) + with contextlib.redirect_stdout(io.StringIO()) as output: + try: + decision = brain.decide(obs, BrokenClient()) + result = {"propagated": False, "decision": vars(decision)} + except jev.JevError: + result = {"propagated": True} + report["combat_model_error"] = {**result, "diagnostic": output.getvalue().strip()} + + reset_policy() + obs = { + "state_type": "card_select", + "card_select": {"screen_type": "upgrade", "prompt": "Choose a card to Upgrade.", + "cards": [card(5)], "can_confirm": False, + "can_cancel": True, "preview_showing": False}, + } + first = brain.decide(obs, None) + second = brain.decide(obs, None) + report["selection_without_execution"] = { + "first": first.action, "second": second.action, "reason": second.reason, + "actual_game_actions": 0, + } + reset_policy() + + try: + answer = jev.JevClient._parse({"answers": {"test": {"type": "noul"}}}, 0)["test"] + result = {"rejected": False, "value": answer.noul, "gate": jev.gate(answer)} + except jev.JevError: + result = {"rejected": True} + report["missing_noul"] = result + answer = jev.ChoiceAnswer("a", {"a": .55, "b": .25, "c": .20}, .325) + report["gate_logging"] = { + "logged": jev.answer_record(answer)["gated"], + "event_numeric_gate": jev.gate_choice(answer, brain.EVENT_TOP_MIN, brain.EVENT_MARGIN_MIN), + } + return report + + +def corpus_audit() -> dict: + dataset = ROOT / "dataset" + index = dataset / "states_index.jsonl" + if not index.exists(): + return {"skipped": "dataset/states_index.jsonl is absent"} + rows = [json.loads(line) for line in index.read_text().splitlines() if line.strip()] + types, phases, actions = Counter(), Counter(), Counter() + exceptions, covered, strength = [], [], [] + status_in_deck = 0 + for row in rows: + path = (dataset / row["path"]).resolve() + if not path.is_relative_to(dataset.resolve()): + raise ValueError("State path leaves dataset directory") + obs = json.loads(gzip.decompress(path.read_bytes())) + types[obs.get("state_type")] += 1 + reset_policy() + try: + decision = brain.decide(obs, None) + actions[decision.action if decision else "no_decision"] += 1 + if obs.get("state_type") not in ("monster", "elite", "boss"): + continue + facts = F.combat_facts(obs) + player = obs.get("player") or {} + phases["play_phase" if facts.in_play_phase else "not_play_phase"] += 1 + if facts.in_play_phase and facts.enemies: + phases["play_phase_with_enemies"] += 1 + if decision and decision.reason.startswith("defense forced") and facts.unblocked_damage == 0: + covered.append(row["path"]) + for item in player.get("hand") or []: + if item.get("name") == "Strike" and F.power_amount(player.get("status"), "Strength") == 2: + strength.append({"path": row["path"], "text": item["description"], + "strength": 2, + "calculated_damage": F.damage_to_target(item, player["status"], [], 0)}) + status_names = {item.get("name") for item in player.get("hand") or [] + if item.get("type") == "Status"} + if status_names & facts.deck_counts.keys(): + status_in_deck += 1 + except Exception as exc: + exceptions.append({"path": row["path"], "error": f"{type(exc).__name__}: {exc}"}) + reset_policy() + return { + "state_types": dict(types), "combat_phases": dict(phases), "actions": dict(actions), + "exceptions": exceptions, "forced_block_when_covered": covered, + "strength_examples": strength, "states_with_combat_status_in_deck_counts": status_in_deck, + } + + +def main() -> int: + reset_policy() + report = {"synthetic": synthetic_probes(), "corpus": corpus_audit()} + print(json.dumps(report, indent=2, sort_keys=True)) + # Successful audit execution is not a clean bill of health. Read the report. + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())