diff --git a/run.py b/run.py new file mode 100644 index 0000000..786545b --- /dev/null +++ b/run.py @@ -0,0 +1,321 @@ +#!/usr/bin/env python3 +""" +run.py -- the observe -> decide -> act loop. + +The loop is strictly closed. Playing a card removes it from hand and shifts +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 --steps 40 # actually play + python3 run.py --steps 40 --no-jev # heuristics only +""" + +from __future__ import annotations + +import argparse +import json +import pathlib +import sys +import time + +import brain +import facts as F +import sts2 +from jev import JevClient, JevError + +DECK_FILE = pathlib.Path("deck.json") + +# Known starting decks, used only until the first combat exposes the real one. +# Source: the character_select state, which lists starting_deck per character. +STARTING_DECKS: dict[str, dict[str, int]] = { + "The Ironclad": {"Strike": 5, "Defend": 4, "Bash": 1}, + "The Silent": {"Strike": 5, "Defend": 5, "Neutralize": 1, "Survivor": 1}, + "The Regent": {"Strike": 4, "Defend": 4, "Falling Star": 1, "Venerate": 1}, +} + + +def load_deck() -> dict | None: + if DECK_FILE.exists(): + try: + return json.loads(DECK_FILE.read_text()) + except (json.JSONDecodeError, OSError): + return None + return None + + +def save_deck(deck: dict) -> None: + DECK_FILE.write_text(json.dumps(deck, indent=2, sort_keys=True)) + + +def observe() -> dict: + return sts2.state() + + +def preflight(obs: dict) -> str | None: + """ + Detect states the bot cannot proceed from, so a session does not silently + burn its whole step budget doing nothing. + + The one that actually bit: after a run ends with a pending Timeline epoch, + the main menu offers only `settings` and `quit`, and the mod REFUSES to + automate the reveal. A whole A/B arm ran with 0 decisions before this was + noticed. + """ + if obs.get("state_type") != "menu" or obs.get("menu_screen") != "main": + return None + + options = obs.get("options") or [] + names = {o if isinstance(o, str) else o.get("name") for o in options} + if names & {"singleplayer", "continue"}: + return None + + blocked = obs.get("blocked_options") or [] + for entry in blocked: + if isinstance(entry, dict) and entry.get("reason") == "manual_epoch_reveal_required": + pending = ", ".join(entry.get("pending_epoch_ids") or []) + return ( + "BLOCKED: the Timeline has unrevealed epochs (" + pending + ").\n" + "The mod refuses to automate this by design, and the main menu\n" + "offers only settings/quit until it is done.\n" + " -> Open the Timeline IN GAME and reveal the epoch by hand,\n" + " then re-run." + ) + + return f"BLOCKED: main menu offers only {sorted(n for n in names if n)}; cannot start a run." + + +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("--no-jev", action="store_true") + ap.add_argument("--pause", type=float, default=0.6, + help="seconds to wait after each action") + ap.add_argument("--stuck-seconds", type=float, default=25.0, + help="give up if the state does not change for this long") + ap.add_argument("--max-duplicate-waits", type=int, default=3, + help="how many times to suppress an identical repeated action " + "on an unchanged state before trying it again") + ap.add_argument("--capture-dir", default="capture") + 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 not sts2.is_up(): + print(f"game not reachable at {sts2.BASE}") + return 1 + + # Fail fast on a state the bot cannot leave, instead of burning the whole + # step budget on rejected actions. + try: + blocker = preflight(observe()) + except sts2.Sts2Error as exc: + print(f"cannot read state: {exc}") + return 1 + if blocker: + print(blocker) + return 2 + + if args.card_skip_policy: + brain.CARD_SKIP_POLICY = args.card_skip_policy + print(f"card skip policy: {brain.CARD_SKIP_POLICY}") + + client = None + if not args.no_jev: + try: + client = JevClient() + print(f"jev ready: {client!r}") + except JevError as exc: + print(f"jev unavailable, using heuristics only: {exc}") + + capdir = pathlib.Path(args.capture_dir) + capdir.mkdir(exist_ok=True) + trace_path = capdir / "decisions.jsonl" + + def trace(record: dict) -> None: + # One JSON line per decided action. Tail it while the bot plays: + # tail -f capture/decisions.jsonl | jq + record["ts"] = time.strftime("%H:%M:%S") + with trace_path.open("a", encoding="utf-8") as fh: + 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 card_reward state does not expose the deck, but combat states expose + # all four piles. Snapshot composition during combat and persist it so it + # survives a restart. + deck_snapshot: dict | None = load_deck() + if deck_snapshot: + print(f"deck snapshot loaded: {deck_snapshot}") + waits = 0 + rejected = 0 + last_sig: str | None = None + same_state = 0 + same_state_since = time.monotonic() + last_ok = True + last_action_key: tuple | None = None + duplicate_waits = 0 + + for step in range(1, args.steps + 1): + try: + obs = observe() + except sts2.Sts2Error as exc: + print(f"[{step:03d}] state read failed: {str(exc)[:160]}") + return 1 + + st = obs.get("state_type") + + # 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 + # yet; acting again queues duplicates (e.g. three map moves in a row). + sig = json.dumps(obs, sort_keys=True) + if sig == last_sig: + same_state += 1 + else: + same_state = 0 + last_sig = sig + same_state_since = time.monotonic() + + # Time-based, not count-based: a boss death animation plus the rewards + # transition can easily exceed a fixed number of reads. + unchanged_for = time.monotonic() - same_state_since + 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 + + # Only wait when our own action actually landed and the game is still + # animating. If the action was rejected, fall through and pick a + # different one instead of waiting out the stuck counter. + # + # NOTE: this check now happens AFTER deciding, and only suppresses a + # REPEATED action. Waiting on "state unchanged" alone blocked + # legitimate sequences -- character select needs select-then-embark, and + # the screen does not change between them, so the bot stalled forever. + + if st in ("monster", "elite", "boss"): + f = F.combat_facts(obs) + (capdir / f"live_{step:03d}_combat.json").write_text(json.dumps(obs, indent=2)) + print(f"[{step:03d}] COMBAT {f.describe().splitlines()[0]}") + for line in f.describe().splitlines()[1:]: + print(f" {line}") + if f.deck_counts and f.deck_counts != deck_snapshot: + deck_snapshot = f.deck_counts + save_deck(deck_snapshot) + + decision = None + decide_error = None + try: + decision = brain.decide(obs, client, deck_snapshot) + except JevError as exc: + print(f"[{step:03d}] jev error: {str(exc)[:160]}") + decide_error = f"JevError: {str(exc)[:160]}" + decision = brain.simple_decision(obs, client, deck_snapshot) + except Exception as exc: # noqa: BLE001 + # A network blip or an unexpected shape must not end the run. Fall + # back to the deterministic handlers and keep playing. + 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, client, deck_snapshot) + except Exception as inner: # noqa: BLE001 + print(f"[{step:03d}] fallback also failed: {inner}") + decision = None + + 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 + + # Between turns there is nothing to do but look again. + if decision.action == "__wait__": + waits += 1 + print(f"[{step:03d}] WAIT {decision.reason}") + time.sleep(args.pause) + continue + + # Suppress DUPLICATE actions on an unchanged state, but only for a + # bounded number of reads. Proposing a DIFFERENT action is always + # allowed, which is what makes select-then-embark and multi-purchase + # shops work. The bound matters because some actions legitimately need + # repeating (multi-line Ancient dialogue) and some transitions are just + # slow -- waiting forever on those stalls the run. + action_key = (decision.action, json.dumps(decision.params, sort_keys=True)) + if same_state == 0: + duplicate_waits = 0 + + if same_state > 0 and last_ok and action_key == last_action_key: + duplicate_waits += 1 + if duplicate_waits <= args.max_duplicate_waits: + waits += 1 + print(f"[{step:03d}] WAIT same action on unchanged state ({same_state})") + time.sleep(args.pause) + continue + print(f"[{step:03d}] RETRY repeating {decision.action} after " + f"{duplicate_waits} waits on an unchanged state") + duplicate_waits = 0 + + stats[decision.source] = stats.get(decision.source, 0) + 1 + print(f"[{step:03d}] DECIDE {decision}") + trace({ + "step": step, + "state_type": st, + "run": obs.get("run"), + "event": "decide", + "source": decision.source, + "action": decision.action, + "params": decision.params, + "reason": decision.reason, + "confidence": decision.confidence, + "error": decide_error, + }) + last_action_key = action_key + + if args.dry_run: + continue + + try: + result = sts2.act(decision.action, **decision.params) + except sts2.Sts2Error as exc: + print(f"[{step:03d}] action failed: {str(exc)[:200]}") + trace({"step": step, "event": "action_error", + "action": decision.action, "error": str(exc)[:200]}) + break + + if not result.ok: + print(f"[{step:03d}] action rejected: {result.message}") + trace({"step": step, "event": "action_rejected", + "action": decision.action, "message": result.message}) + last_ok = False + rejected += 1 + if rejected >= 6: + print(f"[{step:03d}] STUCK: {rejected} consecutive rejections -- stopping") + print(json.dumps(obs, indent=2)[:900]) + break + # 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. + time.sleep(args.pause * 3) + continue + else: + rejected = 0 + last_ok = True + + time.sleep(args.pause) + + elapsed = time.monotonic() - started + print() + print(f"steps={step} waits={waits} elapsed={elapsed:.1f}s sources={stats}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())