diff --git a/.gitignore b/.gitignore index 472a172..44b08ac 100644 --- a/.gitignore +++ b/.gitignore @@ -8,4 +8,6 @@ deck.json # Run artifacts written by run.py / JEV_TRACE during live sessions capture/live_*.json capture/decisions.jsonl +capture/sessions.jsonl +capture/sessions/ capture/jev*.jsonl diff --git a/AGENTS.md b/AGENTS.md index 6614ecb..5674b36 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -5,7 +5,8 @@ - `brain.py`: policy entry point; dispatch, navigation, shops, and minigames. - `policy/combat.py` and `policy/selection.py`: combat and selection proposals. - `policy/context.py`: shared `Decision`, pending actions, and session-owned `PolicyContext`. -- `run.py`: observe–decide–act loop, captures, and session attribution. +- `run.py`: observe–decide–act loop and session attribution. +- `recording.py`: linked session journals; schema and limits in `docs/RECORDING.md`. - `run_state.py`: reported run identity and combat-pile provenance; see `docs/RUN_STATE.md`. - `sts2.py`: local game HTTP client. `jev.py`: TypeSafe model client and gates. - `migrate.py`: dataset migration and integrity checks. @@ -31,6 +32,7 @@ Prefer integration/end-to-end checks; keep only essential regression tests and u - Keep policy memory session-owned. Proposals do not record execution; reconcile accepted requests with fresh observations. - Compute arithmetic and legality in code, not in Jev. Confidence does not prove correctness. - Keep deterministic fallbacks usable with `client=None`. Test model paths with stubs. +- Preserve observation/proposal/attempt links. Flush attempt intent before POST; acceptance is not proof of a game effect. - Preserve session/step attribution. Run outcomes are not per-decision correctness labels. - Split evaluation data by run, not by decision row. diff --git a/CONTEXT.md b/CONTEXT.md index b29b427..547b930 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -25,6 +25,12 @@ _Avoid_: Guaranteed result, proven lethal **Decision**: A proposed next action. A decision does not establish that the game received or applied the action. +**Action attempt**: +One intended submission of a proposed game action. Its record alone does not prove that the game received the request. + +**Subsequent observation**: +The first successfully read game observation after an action attempt. It does not prove that the attempt caused a change. + **Action result**: The game's response to an attempted action. Acceptance alone does not establish that the expected change occurred. diff --git a/docs/POLICY.md b/docs/POLICY.md index 96bfaa8..1bad445 100644 --- a/docs/POLICY.md +++ b/docs/POLICY.md @@ -3,7 +3,8 @@ ## Current boundaries ```text -run.py observe, execute, report results, record the session +run.py observe, execute, report results +recording.py linked observations, proposals, attempts, and results run_state.py reported run identity and scoped combat-pile evidence brain.py entry point, dispatch, navigation, shops, minigames policy/ @@ -51,6 +52,7 @@ shops, and minigames stay together until a further split helps development. The extraction does not change game decisions, confidence gates, fallbacks, state reconciliation, or recording. The subsequent [run-state pass](RUN_STATE.md) adds reported identity and card-evidence provenance. +[Linked recordings](RECORDING.md) provide the corresponding session journal. ## Extraction verification diff --git a/docs/RECORDING.md b/docs/RECORDING.md new file mode 100644 index 0000000..89611a7 --- /dev/null +++ b/docs/RECORDING.md @@ -0,0 +1,154 @@ +# Linked session recordings + +## Files and ownership + +`recording.SessionRecorder` owns one recording per `run.main()` invocation. +It has no game or model client. It records and invokes an action callback supplied +by the runner. Policy execution remains in the runner. + +```text +capture/ + decisions.jsonl compatibility feed + sessions.jsonl compatibility session summaries + sessions// + events.jsonl authoritative, ordered journal + session.json finalized session summary + observations/000001.json full parsed observation + observations/000002.json + ... +``` + +The session ID combines a timestamp with a random UUID. It is generated inside +each invocation, not once per Python process. Session directories and observation +files are created exclusively. Later sessions cannot overwrite earlier captures. +The default session artifacts are ignored by version control. + +All successful state reads are captured, including preflight, menus, overlays, +selection screens, rewards, and terminal screens. Identical observations remain +separate reads. The old `live__combat.json` files are no longer written. + +Observation files contain the unmodified parsed state dictionary. They are not +copies of HTTP headers or the original wire encoding. Each observation event +contains the relative path and SHA-256 hash of the stored bytes. Paths are +relative to the configured capture root. + +## Journal schema, version 1 + +Each journal row includes `schema_version`, `session`, `sequence`, `recorded_at` +(UTC), and the legacy local-time `ts` display field. Sequence orders events +within one session. It is not an ordering across processes. + +| Event | Purpose and links | +| --- | --- | +| `session_start` | Establish the session before preflight or model initialization. | +| `observation` | Full state reference: `observation_id`, `path`, `sha256`, step, phase, and state type. | +| `proposal` | Policy/system decision, linked by `proposal_id` and `observation_id`. Includes action, parameters, reason, source, and available model evidence. | +| `proposal_status` | Disposition of a proposal: `wait`, `suppressed`, `dry_run`, `execute`, or `session_stopped`. | +| `action_attempt` | Intent to submit one request, linked to its proposal and input observation. Has a distinct `attempt_id`. | +| `action_result` | Response to an attempt: `accepted`, `rejected`, or `unknown`. Response/message or transport error is retained. | +| `session_end` | Final exit code, reason, outcome-file candidates, and any attempt still lacking a subsequent observation. | + +Existing diagnostic events, such as `model_error`, `action_error`, and identity +changes, also appear in the journal. `observation_error` records a failed state +read without inventing an observation. Its optional observation reference is the +last successful read, not a payload for the failed request. + +### Join direction + +```text +observation.observation_id + <- proposal.observation_id + <- action_attempt.proposal_id + <- action_result.attempt_id + <- subsequent observation.after_attempt_id +``` + +The first successful read after an attempt carries `after_attempt_id`. Later +reads do not reuse that field until another attempt occurs. A terminal dismissal +may have no subsequent read. The session summary explicitly preserves that +unresolved link in `awaiting_observation_after_attempt`. + +The subsequent observation is temporal evidence, not proof of causation or +completion. A transition overlay can be that first read while policy memory +continues waiting. The journal does not invent a semantic success label. + +Rejected requests are attempts too. Waits, suppressed duplicates, and dry-run +previews have proposals and dispositions, but no attempt or action result. +Dry-run proposals do not update execution-based duplicate suppression. + +Preflight dismissal and stop-on-run-end dismissal use the same recording path +as policy actions. Their observations, proposals, attempts, and results are linked. + +## Model evidence + +Successful model calls retain the request state, questions, and parsed answers +with the proposal. Failed model calls retain the request with `model_error`. +This is not a raw HTTP transcript or a ledger of the model client's internal retries. + +`answer_gate_scope: default_helper_not_policy_gate` labels the existing +per-answer `gated` field. That value does not establish the actual gate used by +a particular policy handler. Actual policy-gate metadata remains separate work. +No correctness label is inferred from confidence, action acceptance, or a run outcome. + +Reported run IDs and card provenance remain subject to the limits in +[run-state documentation](RUN_STATE.md). The recorder does not strengthen save-derived +identity into an atomic observation identity. + +## Durability and failure behavior + +Observation files are flushed before their references are used. The journal's +action intent is flushed before the runner calls the game transport. A recording +failure before that point prevents the POST. + +A transport failure has an `unknown` result and is not retried automatically. +If recording fails after a POST, its effect can remain unknown. A missing result +must never be interpreted as rejection or permission to retry. + +Finalization runs synchronously when the session scope exits. It covers normal +returns, startup failures, handled runtime failures, Python exceptions, and +keyboard interruption. There is no process-global session ID or `atexit` callback. + +SIGTERM, SIGKILL, power failure, and storage failure can still leave an incomplete +journal or no summary. An intent can exist even if the process died before the +POST began. A file flush is not a transactional guarantee across the journal, +observation files, and game service. Readers must treat incomplete records and +truncated final lines as incomplete evidence, not as successful execution. + +## Compatibility and dataset boundaries + +The root `decisions.jsonl` retains `event: decide` rows for unsuppressed actions +and previews. These carry the same IDs as their canonical proposals. Wait and +suppressed proposals appear only in the session journal. Do not count both a +`proposal` and its compatibility `decide` row as separate policy decisions. + +The root session feed retains the outcome-file list and points to its journal +through `recording_path`. The historical outcome association remains a filename +set difference; it is not proof that every new game history file belongs to this bot. +An attribution failure is recorded as `outcome_error` without replacing the session's exit status. + +Compatibility feeds are not transactional or coordinated across writers. Use +per-session journals for reliable joins when multiple processes share a capture root. +Full observations and file flushes add storage and I/O cost intentionally. + +Existing captures and `dataset/` are unchanged. The current dataset migrator +still reads historical inputs and flat captures; it does not import these new +journals automatically. A journal importer and semantic reconciliation labels +can be added separately without rebuilding data during development. + +## Validation + +- The backend-only recording bookmark passed 294 assertions: 68 facts, 110 policy, + and 116 runner assertions. With the separate upgrade-selection fix applied, + the integrated workspace passed 296 assertions (118 runner assertions). +- Runner flows verify hashes, unchanged observation contents, all join keys, + preflight/terminal actions, rejection, unknown transport outcomes, waits, + duplicate suppression, repeated invocations, and keyboard interruption. +- Twenty-five distinct temporary whole-process scenarios passed with real HTTP + clients and local fixture servers. The server checked that each action intent + existed on disk before it received the action POST. +- Injected journal failures before and after POST verify conservative stopping. +- All prior semantic code bookmarks passed their three Python test scripts. +- Dataset integrity, shell syntax, and offline snapshot replay passed. + +No live game actions, paid model calls, new dependencies, or permanent standalone +unit-test files were needed. diff --git a/docs/RUN_STATE.md b/docs/RUN_STATE.md index 837f57d..f15ab21 100644 --- a/docs/RUN_STATE.md +++ b/docs/RUN_STATE.md @@ -97,8 +97,9 @@ Decision rows now include reported `run_id`, `run_seed`, and available it does not mean every handler used it. Combat policy uses the combat observation directly. Identity changes and identity transport failures have separate trace events. -Full observation/action linkage, unique capture names, session identity, and -session finalization remain recording work. +[Linked session recordings](RECORDING.md) now provide observation/action linkage, +unique captures, per-invocation session identity, and synchronous finalization. +The save-derived identity and historical outcome-association limits still apply. ## Validation diff --git a/docs/research/13-policy-state.md b/docs/research/13-policy-state.md index d417056..b965cd1 100644 --- a/docs/research/13-policy-state.md +++ b/docs/research/13-policy-state.md @@ -108,6 +108,6 @@ behavior. See [the current policy structure](../POLICY.md). No plugin framework or class hierarchy was added. [Run identity and card-evidence provenance](../RUN_STATE.md) have since been added. -Recording should next link observations, proposals, action attempts, results, and reconciliation. -Session finalization, capture-name collisions, and actual policy-gate metadata -remain separate work. +[Linked recordings](../RECORDING.md) now connect observations, proposals, attempts, +results, and subsequent reads, with unique captures and synchronous finalization. +Semantic reconciliation labels and actual policy-gate metadata remain separate work. diff --git a/recording.py b/recording.py new file mode 100644 index 0000000..2a9bf17 --- /dev/null +++ b/recording.py @@ -0,0 +1,138 @@ +"""Session-local journal for observations and runner-supplied action calls.""" +from __future__ import annotations + +import hashlib +import json +import os +from pathlib import Path +import time +from datetime import datetime, timezone +import uuid + + +def timestamp() -> str: + return datetime.now(timezone.utc).isoformat() + + +class SessionRecorder: + """A journal is authoritative; root JSONL files are compatibility feeds. + + An attempt records intent before the POST. Neither acceptance nor a later + observation proves that the expected game effect occurred. + """ + + def __init__(self, root: str | Path, *, dry_run: bool = False): + self.root = Path(root) + self.session_id = f"{time.strftime('%Y%m%dT%H%M%S')}-{uuid.uuid4().hex}" + self.directory = self.root / "sessions" / self.session_id + self.directory.mkdir(parents=True, exist_ok=False) + (self.directory / "observations").mkdir() + self.events = self.directory / "events.jsonl" + self.started = timestamp() + self.dry_run = dry_run + self.step = 0 + self.sources = {"code": 0, "jev": 0, "fallback": 0} + self.exit_code = 1 + self.stop_reason = "interrupted" + self.summary: dict = {} + self.sequence = self.observations = self.proposals = self.attempts = 0 + self.observation_id: str | None = None + self.awaiting_observation: str | None = None + + @staticmethod + def _append(path: Path, row: dict, *, durable: bool = False) -> None: + line = json.dumps(row, sort_keys=True, allow_nan=False) + "\n" + with path.open("a", encoding="utf-8") as stream: + stream.write(line) + if durable: + stream.flush() + os.fsync(stream.fileno()) + + def write(self, record: dict, *, mirror: bool = True, durable: bool = False) -> dict: + self.sequence += 1 + row = {**record, "schema_version": 1, "session": self.session_id, + "sequence": self.sequence, "recorded_at": timestamp(), + "ts": time.strftime("%H:%M:%S")} + self._append(self.events, row, durable=durable) + if mirror: + self._append(self.root / "decisions.jsonl", row) + return row + + def observe(self, observation: dict, *, step: int, phase: str) -> str: + self.step = step + self.observations += 1 + observation_id = f"{self.session_id}:o{self.observations:06d}" + path = self.directory / "observations" / f"{self.observations:06d}.json" + data = (json.dumps(observation, sort_keys=True, allow_nan=False) + "\n").encode("utf-8") + with path.open("xb") as stream: + stream.write(data) + stream.flush() + os.fsync(stream.fileno()) + self.observation_id = observation_id + self.write({"event": "observation", "observation_id": observation_id, + "step": step, "phase": phase, "state_type": observation.get("state_type"), + "path": str(path.relative_to(self.root)), "sha256": hashlib.sha256(data).hexdigest(), + "after_attempt_id": self.awaiting_observation}, mirror=False) + self.awaiting_observation = None + return observation_id + + def propose(self, record: dict) -> str: + self.proposals += 1 + proposal_id = f"{self.session_id}:p{self.proposals:06d}" + self.write({**record, "event": "proposal", "proposal_id": proposal_id, + "observation_id": self.observation_id}, mirror=False) + return proposal_id + + def disposition(self, proposal_id: str, status: str) -> None: + self.write({"event": "proposal_status", "proposal_id": proposal_id, + "status": status, "step": self.step}, mirror=False) + + def execute(self, proposal_id: str, decision, act): + """Flush intent before invoking the transport. Never retry here.""" + if self.dry_run: + raise RuntimeError("cannot execute actions in a dry-run recording") + self.attempts += 1 + attempt_id = f"{self.session_id}:a{self.attempts:06d}" + self.disposition(proposal_id, "execute") + self.write({"event": "action_attempt", "attempt_id": attempt_id, + "proposal_id": proposal_id, "observation_id": self.observation_id, + "step": self.step, "action": decision.action, "params": decision.params}, + mirror=False, durable=True) + self.awaiting_observation = attempt_id + try: + result = act(decision.action, **decision.params) + except BaseException as exc: + self.write({"event": "action_result", "attempt_id": attempt_id, + "step": self.step, "outcome": "unknown", + "error": f"{type(exc).__name__}: {str(exc)[:200]}"}, mirror=False) + raise + self.write({"event": "action_result", "attempt_id": attempt_id, "step": self.step, + "outcome": "accepted" if result.ok else "rejected", + "message": result.message, "response": getattr(result, "raw", None)}, mirror=False) + return result + + def finish(self, code: int, reason: str) -> int: + self.exit_code, self.stop_reason = code, reason + return code + + def __enter__(self): + self.write({"event": "session_start", "dry_run": self.dry_run}, mirror=False) + return self + + def __exit__(self, exc_type, exc, traceback): + if exc is not None: + self.exit_code = 130 if isinstance(exc, KeyboardInterrupt) else 1 + self.stop_reason = "interrupted" if isinstance(exc, KeyboardInterrupt) else "exception" + self.summary["error"] = f"{type(exc).__name__}: {str(exc)[:200]}" + row = {**self.summary, "schema_version": 1, "session": self.session_id, + "started": self.started, "ended": timestamp(), "steps": self.step, + "sources": self.sources, "dry_run": self.dry_run, + "exit_code": self.exit_code, "stop_reason": self.stop_reason, + "recording_path": str(self.events.relative_to(self.root)), + "awaiting_observation_after_attempt": self.awaiting_observation} + self.write({**row, "event": "session_end", "step": self.step, "reason": self.stop_reason}, durable=True) + with (self.directory / "session.json").open("x", encoding="utf-8") as stream: + json.dump(row, stream, sort_keys=True, allow_nan=False) + stream.write("\n") + self._append(self.root / "sessions.jsonl", row) + return False diff --git a/run.py b/run.py index c3c0684..6312ec3 100644 --- a/run.py +++ b/run.py @@ -15,20 +15,19 @@ usage: from __future__ import annotations import argparse -import atexit import json import math import os import pathlib import sys import time -import uuid 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 @@ -38,14 +37,6 @@ HISTORY_DIR = pathlib.Path(os.environ.get( "~/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 @@ -66,6 +57,7 @@ class RecordingClient: 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: @@ -75,13 +67,16 @@ class RecordingClient: 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 @@ -114,40 +109,16 @@ def run_outcome(name: str) -> dict: } -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: + error: str | None, client, *, session_id: str) -> dict: """ - One JSONL row per decided action, carrying the model's answers with it. + 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, + "session": session_id, "step": step, "state_type": state_type, "run": run_state, @@ -166,7 +137,7 @@ def observe() -> dict: return sts2.state() -def preflight(obs: dict, *, dry_run: bool = False) -> str | None: +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. @@ -181,7 +152,7 @@ def preflight(obs: dict, *, dry_run: bool = False) -> str | None: if dry_run: return None try: - result = sts2.act("menu_select", option="main_menu") + 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 @@ -250,26 +221,90 @@ def main() -> int: 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 1 + 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: - blocker = preflight(observe(), dry_run=args.dry_run) + 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 1 + return finish(1, "startup_error") if blocker: print(blocker) - return 2 + 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}") - # 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: @@ -278,58 +313,8 @@ def main() -> int: print(f"jev ready: {client!r}") except JevError as exc: print(f"jev unavailable: {exc}; use --no-jev for an intentional heuristic session") - return 3 + return finish(3, "model_initialization") - capdir = pathlib.Path(args.capture_dir) - capdir.mkdir(parents=True, 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") - 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") - - stats = {"code": 0, "jev": 0, "fallback": 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 - exit_code = 1 - stop_reason = "interrupted" - 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) - row.update(exit_code=exit_code, stop_reason=stop_reason, dry_run=args.dry_run) - 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 - - def finish(code: int, reason: str) -> int: - nonlocal exit_code, stop_reason - exit_code, stop_reason = code, reason - trace({"event": "session_end", "step": step, "exit_code": code, "reason": reason}) - elapsed = time.monotonic() - started - print(f"steps={step} waits={waits} elapsed={elapsed:.1f}s sources={stats} " - f"stop={reason} exit={code}") - return code - - atexit.register(write_session_row) - - # Never load the old unscoped deck.json. Visible combat piles are limited - # evidence, not a persistent deck, and remain local to this session/run. - run_context = RunContext() - waits = 0 rejected = 0 last_sig: str | None = None same_state = 0 @@ -343,9 +328,11 @@ def main() -> int: pending_since = time.monotonic() for step in range(1, args.steps + 1): + recording.step = step try: - obs = observe() + 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") @@ -384,10 +371,11 @@ def main() -> int: 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 = sts2.act("menu_select", option="main_menu") + 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") @@ -437,7 +425,6 @@ def main() -> int: 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}") @@ -449,6 +436,7 @@ def main() -> int: # 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 @@ -460,7 +448,7 @@ def main() -> int: 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}) + "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 " @@ -491,15 +479,19 @@ def main() -> int: 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) @@ -518,6 +510,7 @@ def main() -> int: 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) @@ -528,17 +521,15 @@ def main() -> int: stats[decision.source] = stats.get(decision.source, 0) + 1 print(f"[{step:03d}] DECIDE {decision}") - record = decision_record(step, st, obs.get("run"), decision, decide_error, client) - record.update(run_id=run_context.run_id, run_seed=run_context.seed, - deck_provenance=(run_context.deck or {}).get("provenance")) trace(record) - last_action_key = action_key if args.dry_run: + recording.disposition(proposal_id, "dry_run") continue + last_action_key = action_key try: - result = sts2.act(decision.action, **decision.params) + 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", diff --git a/test_run.py b/test_run.py index 620aaa4..a1900a9 100755 --- a/test_run.py +++ b/test_run.py @@ -20,6 +20,7 @@ from __future__ import annotations import contextlib import io +import hashlib import itertools import json import os @@ -125,10 +126,11 @@ class FakeSts2: return types.SimpleNamespace(ok=ok, message="rejected" if not ok else "") -def invoke(fake, *flags, client=None, clock=None, decide=None): +def invoke(fake, *flags, client=None, clock=None, decide=None, directory=None): """Run the real loop with isolated files, no delays, and no live services.""" - callbacks = [] - with tempfile.TemporaryDirectory() as directory, contextlib.ExitStack() as stack: + with contextlib.ExitStack() as stack: + if directory is None: + directory = stack.enter_context(tempfile.TemporaryDirectory()) capdir = pathlib.Path(directory) / "capture" stack.enter_context(patch.object(run, "sts2", fake)) stack.enter_context(patch.object(run, "JevClient", return_value=client or StubClient())) @@ -138,7 +140,6 @@ def invoke(fake, *flags, client=None, clock=None, decide=None): 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", "--capture-dir", str(capdir), *flags])) @@ -148,13 +149,15 @@ def invoke(fake, *flags, client=None, clock=None, decide=None): stack.enter_context(patch.object(brain, "decide", side_effect=decide)) output = stack.enter_context(contextlib.redirect_stdout(io.StringIO())) rc = run.main() - for callback in callbacks: - callback() def rows(name): 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=int(not legacy.exists() or legacy.read_text() != legacy_text), + sessions = rows("sessions.jsonl") + events = [event for session in sessions for event in rows(session["recording_path"])] + blobs = {event["observation_id"]: (capdir / event["path"]).read_bytes() + for event in events if event["event"] == "observation"} + return types.SimpleNamespace(rc=rc, rows=rows("decisions.jsonl"), events=events, blobs=blobs, + sessions=sessions, saved=int(not legacy.exists() or legacy.read_text() != legacy_text), output=output.getvalue()) @@ -173,9 +176,9 @@ 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) + None, stub, session_id="fixture-session") check("jev is null", row["jev"], None) -check("the session id travels with the row", row["session"], run.SESSION_ID) +check("the session id travels with the row", row["session"], "fixture-session") check("the action is still recorded", row["action"], "end_turn") print() @@ -212,7 +215,7 @@ check("...with its value and gate outcome, not just a yes/no", 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}) +check("every row carries the session", {r["session"] for r in rows}, {result.sessions[0]["session"]}) print() print("=== 4. the session row maps to an outcome (join half) ===") @@ -457,6 +460,60 @@ for label, states, identities in [ 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("=== 10. recordings link observations, proposals, attempts, and results ===") +states = [game_over, reward_here, reward_here, {"state_type": "overlay"}, game_over] +client = StubClient(.1) +fake = FakeSts2(states, action_ok=[True, False, True, True]) +result = invoke(fake, "--steps", "4", "--stop-on-run-end", client=client) +observations = {e["observation_id"]: e for e in result.events if e["event"] == "observation"} +proposals = {e["proposal_id"]: e for e in result.events if e["event"] == "proposal"} +attempts = {e["attempt_id"]: e for e in result.events if e["event"] == "action_attempt"} +results = [e for e in result.events if e["event"] == "action_result"] +check("full preflight and loop observations survive unchanged", + [json.loads(result.blobs[key]) for key in observations], states) +check("every stored observation matches its SHA-256 reference", + all(hashlib.sha256(result.blobs[key]).hexdigest() == event["sha256"] + for key, event in observations.items()), True) +check("every proposal and attempt references its exact input observation", + all(p["observation_id"] in observations for p in proposals.values()) and + all(a["proposal_id"] in proposals and a["observation_id"] == proposals[a["proposal_id"]]["observation_id"] + for a in attempts.values()), True) +check("results join one-to-one with actual attempts, including both system dismissals", + ([e["attempt_id"] for e in results], [e["outcome"] for e in results]), + (list(attempts), ["accepted", "rejected", "accepted", "accepted"])) +check("next observations link attempts without asserting their effects", + [e["after_attempt_id"] for e in observations.values()], [None, *list(attempts)[:3], None]) +check("terminal dismissal honestly has no subsequent observation", + result.sessions[0]["awaiting_observation_after_attempt"], list(attempts)[-1]) +check("wait decisions are recorded without action attempts", + ([p["action"] for p in proposals.values()].count("__wait__"), len(attempts)), (1, 4)) +check("successful model records include the exact request state", + [p["jev"]["state"] for p in proposals.values() if p.get("jev")], [s for s, _ in client.requests]) +check("session finalizes exactly once before main returns", + (len(result.sessions), sum(e["event"] == "session_end" for e in result.events)), (1, 1)) + +with tempfile.TemporaryDirectory() as directory: + first = invoke(FakeSts2([menu]), "--steps", "2", "--no-jev", "--dry-run", directory=directory) + second = invoke(FakeSts2([menu]), "--steps", "2", "--no-jev", directory=directory) + check("repeated main calls use distinct session IDs in one capture directory", + len({s["session"] for s in second.sessions}), 2) + check("later sessions never overwrite earlier observations", + {key: second.blobs[key] for key in first.blobs}, first.blobs) + check("dry-run records previews, never attempts or execution-based suppression", + [e["status"] for e in first.events if e["event"] == "proposal_status"], ["dry_run", "dry_run"]) + check("duplicate suppression is visible as a proposal disposition", + [e["status"] for e in second.events if e["event"] == "proposal_status"][-2:], ["execute", "suppressed"]) + +for error, expected_code, reason in [(FakeSts2.Sts2Error("connection closed"), 1, "action_error"), + (KeyboardInterrupt(), 130, "interrupted")]: + fake = FakeSts2([menu], action_error=error) + result = invoke(fake, "--no-jev") + check(f"{reason}: one attempt, unknown outcome, synchronous finalization", + (result.rc, len(fake.actions), [e["outcome"] for e in result.events if e["event"] == "action_result"], + result.sessions[0]["stop_reason"], sum(e["event"] == "session_end" for e in result.events)), + (expected_code, 1, ["unknown"], reason, 1)) + print() print(f"=== {PASS} passed, {FAIL} failed ===") sys.exit(1 if FAIL else 0)