#!/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 # no action POSTs; reads/model/logs still run 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 math import os import pathlib import sys import time import brain import facts as F import sts2 from jev import JevClient, JevError, answer_record from run_state import RunContext from recording import SessionRecorder # Where the game writes its own run records. This is the OUTCOME source: it # carries win/loss, the killer, the seed and the final deck. Override with # STS2_HISTORY_DIR when the steam id differs. HISTORY_DIR = pathlib.Path(os.environ.get( "STS2_HISTORY_DIR", "~/Library/Application Support/SlayTheSpire2/steam/" "76561198141226155/modded/profile1/saves/history")).expanduser() class RecordingClient: """ Wraps a JevClient so the last call's questions and answers travel with the decision row they produced. This makes the log ATTRIBUTABLE. `JEV_TRACE` alone cannot be joined to a decision -- it records a time to the second and the answers, with no run, step or action -- so nothing in it can be traced back to a run. What the join supports is arm-level analysis: this decision belongs to this session, and the session's outcome is the run file, which is what an A/B or a delayed-reward model needs. It does NOT say whether an individual answer was correct. One run result attached to one step cannot label that step; per-decision accuracy needs replay, expert judgement, or a ground truth the code can verify on its own (lethal, legality, affordability). """ def __init__(self, inner: JevClient) -> None: self._inner = inner self.last: dict | None = None self.last_request: dict | None = None @property def model(self) -> str: return self._inner.model def __repr__(self) -> str: # keep the key redacted return f"RecordingClient({self._inner!r})" def ask(self, state, questions, model=None): self.last_request = {"state": state, "questions": questions, "model": model or self.model} response = self._inner.ask(state, questions, model) self.last = { "state": state, "model": response.model, "latency_s": round(response.latency_s, 3), "questions": questions, "answers": {qid: answer_record(a) for qid, a in response.answers.items()}, "answer_gate_scope": "default_helper_not_policy_gate", } return response def history_snapshot() -> set[str]: """Run-record filenames present right now. Diffed to attribute a session.""" try: return {p.name for p in HISTORY_DIR.iterdir() if p.suffix == ".run"} except OSError: return set() def run_outcome(name: str) -> dict: """The outcome fields of one game run record, or a stub if it is unreadable.""" try: data = json.loads((HISTORY_DIR / name).read_text()) except (OSError, json.JSONDecodeError): return {"file": name} players = data.get("players") or [{}] points = data.get("map_point_history") or [] return { "file": name, "win": data.get("win"), "killed_by": str(data.get("killed_by_encounter") or "").replace( "ENCOUNTER.", ""), "seed": data.get("seed"), "deck_size": len(players[0].get("deck") or []), "map_points": sum(len(a) for a in points if isinstance(a, list)), "run_time_s": data.get("run_time"), } def decision_record(step: int, state_type: str, run_state, decision, error: str | None, client, *, session_id: str) -> dict: """ Shared payload for a proposal and its compatible `decide` feed row. `jev` is None for a decision that did not ask the model (code paths and fallbacks), so an absent entry is a fact about the decision, not a gap. """ return { "session": session_id, "step": step, "state_type": state_type, "run": run_state, "event": "decide", "source": decision.source, "action": decision.action, "params": decision.params, "reason": decision.reason, "confidence": decision.confidence, "scoring": decision.scoring, "error": error, "jev": getattr(client, "last", None), } def observe() -> dict: return sts2.state() def preflight(obs: dict, *, dry_run: bool = False, act=None) -> str | None: """ Detect states the bot cannot proceed from, so a session does not silently burn its whole step budget doing nothing. Two that actually bit: * A pending Timeline epoch leaves the main menu with only settings/quit, and the mod REFUSES to automate the reveal. A whole A/B arm ran with 0 decisions before this was noticed. * A parked `game_over` screen blocks every later session. Dismiss it. """ if obs.get("state_type") == "game_over": if dry_run: return None try: result = (act or 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. time.sleep(1.5) print("dismissed a parked game-over screen") except sts2.Sts2Error as exc: return f"BLOCKED: parked on game_over and could not dismiss it: {exc}" return None 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", help="send no action POSTs; " "state/identity 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") 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("--max-jev-errors", type=int, default=8, help="abort the session after this many consecutive model " "failures, so a network outage does not silently " "produce heuristic-only data") ap.add_argument("--stop-on-run-end", action="store_true", help="end the session when the run ends instead of starting " "a fresh one. REQUIRED for A/B work: without it a " "session can contain several runs and the results " "cannot be attributed to an arm.") ap.add_argument("--capture-dir", default="capture") ap.add_argument("--card-skip-policy", choices=("jev", "combined", "composite"), 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") try: with SessionRecorder(args.capture_dir, dry_run=args.dry_run) as recording: history_before = history_snapshot() try: return _run_session(args, recording) finally: try: recording.summary["runs"] = [run_outcome(name) for name in sorted(history_snapshot() - history_before)] except Exception as exc: # Outcome attribution must not mask the session's actual exit. recording.summary.update(runs=[], outcome_error=f"{type(exc).__name__}: {str(exc)[:200]}") except KeyboardInterrupt: print("interrupted; session recording finalized") return 130 except OSError as exc: print(f"recording failed; stopping without another action: {exc}") return 1 def _run_session(args, recording: SessionRecorder) -> int: stats = recording.sources started = time.monotonic() step = waits = 0 run_context = RunContext() # Grepped by ab_card_skip.sh for attribution. New for every main() call. print(f"session: {recording.session_id}") def trace(record: dict) -> None: record.setdefault("observation_id", recording.observation_id) recording.write(record) def finish(code: int, reason: str) -> int: elapsed = time.monotonic() - started print(f"steps={step} waits={waits} elapsed={elapsed:.1f}s sources={stats} " f"stop={reason} exit={code}") return recording.finish(code, reason) def read_observation(phase: str = "loop") -> dict: observation = observe() recording.observe(observation, step=step, phase=phase) return observation def propose(observation, decision, error=None, client=None): record = decision_record(step, observation.get("state_type"), observation.get("run"), decision, error, client, session_id=recording.session_id) record.update(run_id=run_context.run_id, run_seed=run_context.seed, deck_provenance=(run_context.deck or {}).get("provenance"), observation_id=recording.observation_id) proposal_id = recording.propose(record) record["proposal_id"] = proposal_id return record, proposal_id def system_action(observation, action, params, reason): decision = brain.Decision(action, params, reason, "code") record, proposal_id = propose(observation, decision) trace(record) stats["code"] += 1 if args.dry_run: recording.disposition(proposal_id, "dry_run") return None return recording.execute(proposal_id, decision, sts2.act) if not sts2.is_up(): print(f"game not reachable at {sts2.BASE}") return finish(1, "startup_error") # Fail fast on a state the bot cannot leave, instead of burning the whole # step budget on rejected actions. try: obs = read_observation("preflight") if args.dry_run and obs.get("state_type") == "game_over": system_action(obs, "menu_select", {"option": "main_menu"}, "preflight game-over dismissal") blocker = preflight(obs, dry_run=args.dry_run, act=lambda action, **params: system_action(obs, action, params, "preflight game-over dismissal")) except sts2.Sts2Error as exc: trace({"event": "observation_error", "step": step, "phase": "preflight", "error": str(exc)[:160]}) print(f"cannot read state: {exc}") return finish(1, "startup_error") if blocker: print(blocker) return finish(2, "preflight_blocked") 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 = RecordingClient(JevClient()) print(f"jev ready: {client!r}") except JevError as exc: print(f"jev unavailable: {exc}; use --no-jev for an intentional heuristic session") return finish(3, "model_initialization") 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 jev_errors = 0 saw_a_run = False pending_action = None pending_since = time.monotonic() for step in range(1, args.steps + 1): recording.step = step try: obs = read_observation() except sts2.Sts2Error as exc: trace({"event": "observation_error", "step": step, "phase": "loop", "error": str(exc)[:160]}) print(f"[{step:03d}] state read failed: {str(exc)[:160]}") return finish(1, "state_error") st = obs.get("state_type") previous_id = run_context.run_id previous_known_id = run_context.last_known_id identity = None if st not in ("menu", "game_over", "unknown", "overlay"): try: identity = sts2.compendium().get("current_run") except sts2.Sts2Error as exc: trace({"step": step, "event": "run_identity_error", "error": str(exc)[:160]}) run_context.observe(obs, identity) if run_context.run_id != previous_id: trace({"step": step, "event": "run_identity", "run_id": run_context.run_id, "seed": run_context.seed, "source": "compendium.current_run"}) run_changed = (previous_known_id is not None and run_context.run_id is not None and previous_known_id != run_context.run_id) if run_changed: if args.stop_on_run_end: return finish(0, "run_changed") last_sig = last_action_key = None same_state = rejected = duplicate_waits = 0 same_state_since = time.monotonic() last_ok = True policy_context = run_context.policy # One session = one run, when asked. Otherwise the bot dies, returns to # the menu and starts a fresh run inside the same session, so a single # session yields several run records and nothing can be attributed to # an experimental arm. # # CRITICAL: dismiss the game-over screen BEFORE stopping. Breaking # 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") system_action(obs, "menu_select", {"option": "main_menu"}, "run-end dismissal") return finish(0, "run_end_preview") print(f"[{step:03d}] run ended; dismissing game-over, then stopping") try: result = system_action(obs, "menu_select", {"option": "main_menu"}, "run-end dismissal") 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) 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", "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") 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 # 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]) return finish(1, "stuck") # Acceptance does not prove completion. The policy context handles # pending selections and purchases. Rejections allow another proposal # 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) print(f"[{step:03d}] COMBAT {f.describe().splitlines()[0]}") for line in f.describe().splitlines()[1:]: print(f" {line}") run_context.capture_combat(obs, f, step) decision = None decide_error = None if isinstance(client, RecordingClient): # Cleared first: a decision that asks nothing (code paths, fallbacks) # must not inherit the previous step's answers in the log. client.last = None client.last_request = None try: decision = brain.decide(obs, client, run_context.deck, context=policy_context) # 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, "request": getattr(client, "last_request", None)}) 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 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. try: decision = brain.decide(obs, None, run_context.deck, context=policy_context) 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 # 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]}") 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]) return finish(1, "no_decision") record, proposal_id = propose(obs, decision, decide_error, client) # Bound unresolved actions even if unrelated observation fields change. if policy_context.pending is not pending_action: pending_action = policy_context.pending pending_since = time.monotonic() if pending_action is not None and time.monotonic() - pending_since > args.stuck_seconds: recording.disposition(proposal_id, "session_stopped") return finish(1, "pending_action_timeout") # Between turns there is nothing to do but look again. if decision.action == "__wait__": recording.disposition(proposal_id, "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 here, but cannot bypass a pending policy action above. # 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: recording.disposition(proposal_id, "suppressed") 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(record) if args.dry_run: recording.disposition(proposal_id, "dry_run") continue last_action_key = action_key try: result = recording.execute(proposal_id, decision, sts2.act) 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]}) return finish(1, "action_error") run_context.record_result(obs, decision, accepted=result.ok) 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]) 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. time.sleep(args.pause * 3) continue else: rejected = 0 last_ok = True time.sleep(args.pause) 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__": raise SystemExit(main())