From 17cb02a1479f88d52e4c375b06bddaf5ff33099c Mon Sep 17 00:00:00 2001 From: 0xrsydn Date: Tue, 22 Sep 2026 06:06:06 +0700 Subject: [PATCH] feat(run): log every decision with the answers and the run outcome `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. Wrap the client in `RecordingClient` so the last call's questions and answers travel with the decision row they produced, and give each process a `SESSION_ID` so rows from two sessions in the same second stay apart. Diff the history directory before and after a session to attribute the run records it produced. What this supports is arm-level analysis: this decision belongs to this session, and the session's outcome is the run file. 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 ground truth the code can verify on its own (lethal, legality, affordability). A decision that asked nothing must carry no answers, so `client.last` is cleared before each decide(); otherwise rows silently inherit the previous step's answers. --- run.py | 262 +++++++++++++++++++++++++++++++++++++++++++++++----- test_run.py | 183 ++++++++++++++++++++++++++++++++++++ 2 files changed, 420 insertions(+), 25 deletions(-) create mode 100755 test_run.py diff --git a/run.py b/run.py index 786545b..7eee869 100644 --- a/run.py +++ b/run.py @@ -15,18 +15,153 @@ usage: from __future__ import annotations import argparse +import atexit import json +import os import pathlib import sys import time +import uuid import brain import facts as F import sts2 -from jev import JevClient, JevError +from jev import JevClient, JevError, answer_record DECK_FILE = pathlib.Path("deck.json") +# 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() + +# One id per process, written into every trace row. A decision row and the model +# answers that produced it must be joinable, and a bare wall-clock time is not +# an identifier: JEV_TRACE timestamps only to the second, so several calls share +# a timestamp and nothing in it names the run, step or action. The random suffix +# keeps two processes started in the same second apart. +SESSION_ID = f"{time.strftime('%Y%m%dT%H%M%S')}-{uuid.uuid4().hex[:6]}" + + +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 + + @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): + response = self._inner.ask(state, questions, model) + self.last = { + "model": response.model, + "latency_s": round(response.latency_s, 3), + "questions": questions, + "answers": {qid: answer_record(a) + for qid, a in response.answers.items()}, + } + 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 session_record(session_id: str, started_at: str, ended_at: str, steps: int, + sources: dict, produced) -> dict: + """ + session id -> the run record(s) that session produced. + + Nothing else writes this mapping: the game's history file does not know our + session id, and the A/B harness only maps a policy to a filename. Without it + a decision cannot be attributed to the run it belongs to, so no arm-level + outcome analysis (A/B, delayed reward) is possible. + + It is NOT a per-decision accuracy label. One run result attached to a step + says nothing about whether that step's answer was correct; calibrating a + gate needs labels for the individual answers. + """ + return { + "session": session_id, + "started": started_at, + "ended": ended_at, + "steps": steps, + "sources": sources, + "runs": [run_outcome(name) for name in sorted(produced)], + } + + +def decision_record(step: int, state_type: str, run_state, decision, + error: str | None, client) -> dict: + """ + One JSONL row per decided action, carrying the model's answers with it. + + `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, + "error": error, + "jev": getattr(client, "last", None), + } + + # 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]] = { @@ -58,11 +193,24 @@ 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. + 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": + try: + sts2.act("menu_select", option="main_menu") + # 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 @@ -98,6 +246,15 @@ def main() -> int: 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"), default=None, help="how card-reward skips are decided (default: brain's own)") @@ -121,11 +278,13 @@ def main() -> int: if args.card_skip_policy: brain.CARD_SKIP_POLICY = args.card_skip_policy print(f"card skip policy: {brain.CARD_SKIP_POLICY}") + # Grepped by ab_card_skip.sh to stamp its attribution rows with the same id. + print(f"session: {SESSION_ID}") client = None if not args.no_jev: try: - client = JevClient() + client = RecordingClient(JevClient()) print(f"jev ready: {client!r}") except JevError as exc: print(f"jev unavailable, using heuristics only: {exc}") @@ -138,6 +297,7 @@ def main() -> int: # 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") + record.setdefault("session", SESSION_ID) with trace_path.open("a", encoding="utf-8") as fh: fh.write(json.dumps(record, sort_keys=True, default=str) + "\n") @@ -146,6 +306,25 @@ def main() -> int: jev_tokens = 0 started = time.monotonic() + # The outcome half of the join: which run record(s) this session produced. + # Registered with atexit so EVERY exit path writes it -- the normal end, the + # step cap, a stuck screen, a model-failure abort, or an exception. + history_before = history_snapshot() + step = 0 + started_at = time.strftime("%Y-%m-%dT%H:%M:%S") + + def write_session_row() -> None: + try: + row = session_record(SESSION_ID, started_at, + time.strftime("%Y-%m-%dT%H:%M:%S"), step, stats, + history_snapshot() - history_before) + 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 + + atexit.register(write_session_row) + # 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. @@ -160,6 +339,8 @@ def main() -> int: last_ok = True last_action_key: tuple | None = None duplicate_waits = 0 + jev_errors = 0 + saw_a_run = False for step in range(1, args.steps + 1): try: @@ -170,6 +351,32 @@ def main() -> int: st = obs.get("state_type") + # 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": + print(f"[{step:03d}] run ended; dismissing game-over, then stopping") + try: + sts2.act("menu_select", option="main_menu") + except sts2.Sts2Error as exc: + print(f"[{step:03d}] could not dismiss game-over: {exc}") + time.sleep(args.pause) + break + 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"): + 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 + # 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). @@ -204,26 +411,42 @@ def main() -> int: 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 + # The snapshot carries the name->count map (card identities, which + # are the synergy signal) plus stable all-pile aggregates. Only the + # 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) 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 try: decision = brain.decide(obs, client, deck_snapshot) + jev_errors = 0 except JevError as exc: - print(f"[{step:03d}] jev error: {str(exc)[:160]}") + jev_errors += 1 + print(f"[{step:03d}] jev error ({jev_errors}): {str(exc)[:140]}") decide_error = f"JevError: {str(exc)[:160]}" - decision = brain.simple_decision(obs, client, deck_snapshot) + 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 + # 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) 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. + # A network blip or an unexpected shape must not end the run. 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) + decision = brain.simple_decision(obs, None, deck_snapshot) except Exception as inner: # noqa: BLE001 print(f"[{step:03d}] fallback also failed: {inner}") decision = None @@ -265,18 +488,7 @@ def main() -> int: 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, - }) + trace(decision_record(step, st, obs.get("run"), decision, decide_error, client)) last_action_key = action_key if args.dry_run: diff --git a/test_run.py b/test_run.py new file mode 100755 index 0000000..6b41e1f --- /dev/null +++ b/test_run.py @@ -0,0 +1,183 @@ +#!/usr/bin/env python3 +""" +test_run.py -- regression tests for the decision log in run.py. + +The log is the only record of what the model was asked and what it answered, so +two things must hold or every later analysis is wrong: + + * a model decision's row carries that decision's answers; + * a decision that asked NOTHING carries none -- no inheritance from the + previous step, which is what `client.last = None` before each decide() is + for, and which fails silently if it is ever dropped. + +Runs offline: a stub client stands in for JevClient and a fake sts2 module +stands in for the game, so no network and no running game are needed. + +Run: python3 test_run.py +""" + +from __future__ import annotations + +import json +import pathlib +import sys +import tempfile +import types + +import brain +import jev +import run +from jev import ChoiceAnswer, JevResponse, NoulAnswer, ScoreAnswer + +PASS = 0 +FAIL = 0 + + +def check(label: str, got, want) -> None: + global PASS, FAIL + if got == want: + PASS += 1 + print(f" ok {label}: {got!r}") + else: + FAIL += 1 + print(f" FAIL {label}: got {got!r}, want {want!r}") + + +class StubClient: + """Answers every question locally, so the log can be tested without a model.""" + + model = "stub" + + def __init__(self, noul: float = 0.9): + self.noul = noul + self.calls = 0 + + def __repr__(self) -> str: + return "StubClient()" + + def ask(self, state, questions, model=None): + self.calls += 1 + answers = {} + for qid, q in questions.items(): + if q.get("type") == "choice": + options = list((q.get("criteria") or {}).keys()) + answers[qid] = ChoiceAnswer( + choice=options[0] if options else "", + probabilities={o: (0.9 if o == options[0] else 0.05) + for o in options}, + confidence=0.9, + ) + elif q.get("type") == "noul": + answers[qid] = NoulAnswer(noul=self.noul) + else: + answers[qid] = ScoreAnswer(score=1.0, confidence=0.9) + return JevResponse(answers=answers, model="stub", input_tokens=0, + output_tokens=0, latency_s=0.0) + + +class FakeSts2: + """A fixed sequence of states, then the last one forever.""" + + class Sts2Error(RuntimeError): + pass + + def __init__(self, states): + self.states = states + self.i = 0 + + def is_up(self) -> bool: + return True + + def state(self) -> dict: + state = self.states[min(self.i, len(self.states) - 1)] + self.i += 1 + return state + + def act(self, *a, **k): + return types.SimpleNamespace(ok=True, message="") + + +print("=== 1. answer_record shapes (what the log stores) ===") +check("noul", jev.answer_record(NoulAnswer(0.72)), + {"kind": "noul", "noul": 0.72, "yes": True, "gated": True}) +check("noul gate outcome is recorded", jev.answer_record(NoulAnswer(0.52))["gated"], False) +choice = jev.answer_record(ChoiceAnswer("a", {"a": 0.7, "b": 0.2}, 0.5)) +check("choice margin", choice["margin"], 0.5) +check("choice gated", choice["gated"], True) +check("choice probabilities survive", choice["probabilities"], {"a": 0.7, "b": 0.2}) +check("score", jev.answer_record(ScoreAnswer(1.4, {}, {}, 0.9))["kind"], "score") + +print() +print("=== 2. a decision that asked nothing logs jev: null ===") +stub = run.RecordingClient(StubClient()) +code_decision = brain.Decision("end_turn", {}, "no playable cards", "code") +row = run.decision_record(1, "monster", {"act": 1, "floor": 1}, code_decision, + None, stub) +check("jev is null", row["jev"], None) +check("the session id travels with the row", row["session"], run.SESSION_ID) +check("the action is still recorded", row["action"], "end_turn") + +print() +print("=== 3. end to end: a model row keeps its answers, the next row does not ===") +# Step 1 is a card_reward, which always asks the model. Step 2 is the main menu, +# which never does. If the reset is dropped, row 2 inherits row 1's answers -- +# the failure this test exists to catch. +card_reward = { + "state_type": "card_reward", + "card_reward": {"can_skip": True, "cards": [ + {"id": "A", "name": "Stomp", "type": "Attack", "cost": "3", + "description": "Deal 12 damage.", "rarity": "Uncommon", + "is_upgraded": False, "index": 0}, + {"id": "B", "name": "Inflame", "type": "Power", "cost": "1", + "description": "Gain 2 Strength.", "rarity": "Uncommon", + "is_upgraded": False, "index": 1}, + ]}, + "run": {"act": 1, "floor": 2, "ascension": 0}, + "player": {"character": "The Ironclad", "hp": 70, "max_hp": 80, "gold": 99}, +} +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()] +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"]), + ["good_card0", "good_card1", "skip_all"]) +check("...with its value and gate outcome, not just a yes/no", + (rows[0]["jev"]["answers"]["good_card0"]["noul"], + rows[0]["jev"]["answers"]["good_card0"]["gated"]), (0.9, True)) +check("row 2 asked nothing", rows[1]["jev"], None) +check("...and is a code decision", rows[1]["source"], "code") +check("every row carries the session", {r["session"] for r in rows}, {run.SESSION_ID}) + +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)") + +print() +print(f"=== {PASS} passed, {FAIL} failed ===") +sys.exit(1 if FAIL else 0)