sts2-bot/recording.py

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