Add run loop with decision trace logging
observe -> decide -> act loop: Timeline preflight, stuck detection, duplicate-action suppression, deck snapshot persistence. Appends one JSON line per decision and per rejected action to capture/decisions.jsonl for tail -f.
This commit is contained in:
0xrsydn
parent
bbb91e2f33
commit
41e3ea4a2b
1 changed files with 321 additions and 0 deletions
321
run.py
Normal file
321
run.py
Normal file
|
|
@ -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())
|
||||
Loading…
Add table
Add a link
Reference in a new issue