feat(recording): link session observations proposals and action results

Capture all successful state reads with hashes and session-local IDs. Record action intent before POST, retain accepted/rejected/unknown results, and link subsequent observations. Preserve legacy feeds and finalize each invocation synchronously.

.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

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.
 

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.
 

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
 

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/<session-id>/
+        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_<step>_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.

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
+[Linked session recordings](RECORDING.md) now provide observation/action linkage,
-session finalization remain recording work.
+unique captures, per-invocation session identity, and synchronous finalization.
+The save-derived identity and historical outcome-association limits still apply.
 
 ## Validation
 

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.
+[Linked recordings](../RECORDING.md) now connect observations, proposals, attempts,
-Session finalization, capture-name collisions, and actual policy-gate metadata
+results, and subsequent reads, with unique captures and synchronous finalization.
-remain separate work.
+Semantic reconciliation labels and actual policy-gate metadata remain separate work.

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

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",

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 contextlib.ExitStack() as stack:
-    with tempfile.TemporaryDirectory() as directory, 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")
-                                     sessions=rows("sessions.jsonl"), saved=int(not legacy.exists() or legacy.read_text() != legacy_text),
+        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)