0xrsydn/sts2-bot
1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(state): scope card evidence and policy memory by reported run identity

Browse source
This commit is contained in:
0xrsydn 2026-09-22 12:45:46 +07:00
commit ba519a850e
9 changed files with 343 additions and 82 deletions
Download patch file Download diff file Expand all files Collapse all files

75
test_run.py
View file

@ -22,6 +22,7 @@ import contextlib
import io
import itertools
import json
import os
import pathlib
import sys
import tempfile
@ -55,12 +56,14 @@ class StubClient:
def __init__(self, noul: float = 0.9):
self.noul = noul
self.calls = 0
self.requests = []
def __repr__(self) -> str:
return "StubClient()"
def ask(self, state, questions, model=None):
self.calls += 1
self.requests.append((state, questions))
answers = {}
for qid, q in questions.items():
if q.get("type") == "choice":
@ -87,12 +90,14 @@ class FakeSts2:
BASE = "offline://game"
def __init__(self, states, *, action_ok=True, action_error=None):
def __init__(self, states, *, action_ok=True, action_error=None, identities=None):
self.states = states
self.i = 0
self.actions = []
self.action_ok = action_ok
self.action_error = action_error
self.identities = identities or [{"is_in_progress": True, "run_id": "fixture:A", "seed": "same-seed"}]
self.identity_reads = 0
def is_up(self) -> bool:
return True
@ -104,6 +109,13 @@ class FakeSts2:
raise state
return state
def compendium(self):
identity = self.identities[min(self.identity_reads, len(self.identities) - 1)]
self.identity_reads += 1
if isinstance(identity, Exception):
raise identity
return {"current_run": identity}
def act(self, *a, **k):
self.actions.append((a, k))
if self.action_error:
@ -121,8 +133,11 @@ def invoke(fake, *flags, client=None, clock=None, decide=None):
stack.enter_context(patch.object(run, "sts2", fake))
stack.enter_context(patch.object(run, "JevClient", return_value=client or StubClient()))
stack.enter_context(patch.object(run, "history_snapshot", return_value=set()))
stack.enter_context(patch.object(run, "load_deck", return_value=None))
save = stack.enter_context(patch.object(run, "save_deck"))
stack.callback(os.chdir, os.getcwd())
os.chdir(directory)
legacy = pathlib.Path(directory) / "deck.json"
legacy_text = '{"counts": {"STALE CARD": 999}}'
legacy.write_text(legacy_text)
stack.enter_context(patch.object(run.atexit, "register", side_effect=callbacks.append))
stack.enter_context(patch.object(run.time, "sleep"))
stack.enter_context(patch.object(sys, "argv", ["run.py", "--steps", "10", "--pause", "0",
@ -139,7 +154,7 @@ def invoke(fake, *flags, client=None, clock=None, decide=None):
path = capdir / name
return [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else []
return types.SimpleNamespace(rc=rc, rows=rows("decisions.jsonl"),
sessions=rows("sessions.jsonl"), saved=save.call_count,
sessions=rows("sessions.jsonl"), saved=int(not legacy.exists() or legacy.read_text() != legacy_text),
output=output.getvalue())
@ -390,6 +405,58 @@ check("pending purchase times out despite unrelated observation changes",
(result.rc, result.sessions[0]["stop_reason"], len(fake.actions)),
(1, "pending_action_timeout", 1))
print()
print("=== 9. run identity and card-evidence provenance ===")
run_a = {"is_in_progress": True, "run_id": "fixture:A", "seed": "same-seed"}
run_b = dict(run_a, run_id="fixture:B")
fake = FakeSts2([menu, grid], identities=[run_a, run_b])
result = invoke(fake, "--no-jev", "--steps", "2")
check("different run IDs reset selection and duplicate guards despite identical seeds/screens",
fake.actions, [(("select_card",), {"index": 0})] * 2)
check("decision rows carry reported run identity",
[r["run_id"] for r in result.rows if r["event"] == "decide"], ["fixture:A", "fixture:B"])
fake = FakeSts2([menu, grid], identities=[run_a, run_b])
result = invoke(fake, "--no-jev", "--steps", "2", "--stop-on-run-end")
check("single-run mode never acts in the replacement run",
(result.rc, result.sessions[0]["stop_reason"], len(fake.actions)), (0, "run_changed", 1))
fake = FakeSts2([menu, grid, grid, confirmable],
identities=[run_a, FakeSts2.Sts2Error("metadata unavailable"), run_a])
result = invoke(fake, "--no-jev", "--steps", "3")
check("metadata failure preserves accepted toggles instead of selecting them again",
fake.actions, [(("select_card",), {"index": 0}), (("select_card",), {"index": 1}),
(("confirm_selection",), {})])
check("metadata failure is traced without inventing identity",
[r["run_id"] for r in result.rows if r["event"] == "decide"], ["fixture:A", None, "fixture:A"])
reward_here = dict(card_reward, run=combat["run"])
with_status = dict(combat, player=dict(combat["player"], hand=combat["player"]["hand"] + [
{"index": 1, "name": "Wound", "type": "Status", "cost": "1", "description": "Unplayable.", "can_play": False}]))
client = StubClient(.1) # Declines skipping; selects a card, then invalidates old evidence.
result = invoke(FakeSts2([menu, with_status, reward_here, reward_here]), "--steps", "3", client=client)
evidence = [state["deck_composition"] for state, _ in client.requests if "deck_composition" in state]
check("temporary Status cards remain labeled combat evidence, not a persistent deck",
(evidence[0]["cards"].get("Wound"), evidence[0]["provenance"]["persistent_deck"],
evidence[0]["provenance"]["source"]), (1, False, "combat_piles"))
check("carried card evidence records run, step, age, and exposed piles",
{k: evidence[0]["provenance"][k] for k in ("run_id", "observed_step", "freshness", "pile_lists_present")},
{"run_id": "fixture:A", "observed_step": 1, "freshness": "historical_combat_piles", "pile_lists_present": ["hand"]})
check("accepted card addition invalidates old composition", evidence[1], "unknown")
check("legacy deck.json stays untouched even during an executing session", result.saved, 0)
for label, states, identities in [
("another run", [menu, combat, reward_here], [run_a, run_b]),
("missing identity", [menu, combat, reward_here], [None]),
("identity read failure", [menu, combat, reward_here], [run_a, FakeSts2.Sts2Error("unavailable")]),
("another room", [menu, combat, card_reward], [run_a]),
("unknown room", [menu, combat, dict(reward_here, run=None)], [run_a]),
("fresh session with a legacy cache", [menu, reward_here], [run_a]),
]:
client = StubClient()
result = invoke(FakeSts2(states, identities=identities), "--steps", str(len(states) - 1), client=client)
evidence = [state["deck_composition"] for state, _ in client.requests if "deck_composition" in state]
check(label + " receives unknown deck context", evidence[-1], "unknown")
print()
print(f"=== {PASS} passed, {FAIL} failed ===")
sys.exit(1 if FAIL else 0)