173 lines
7 KiB
Python
173 lines
7 KiB
Python
#!/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 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()}
|
|
|
|
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},
|
|
}
|
|
context = brain.PolicyContext()
|
|
first = brain.decide(obs, None, context=context)
|
|
second = brain.decide(obs, None, context=context)
|
|
report["selection_without_execution"] = {
|
|
"first": first.action, "second": second.action, "reason": second.reason,
|
|
"actual_game_actions": 0,
|
|
}
|
|
|
|
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
|
|
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}"})
|
|
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:
|
|
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())
|