"""SQLite-backed append-only ledger — the persistent backend for long-running scenarios. Replaces the in-memory Ledger for production use. The in-memory Ledger remains valid for tests and short demo runs where persistence is not required. Design decisions: - UNIQUE constraint on id enforces idempotency at the DB layer. - Serial OFFSET column gives a deterministic ordering guarantee even if clock skew produces non-monotonic created_at timestamps. - snapshot_to() uses SQLite's native .backup() API — atomic, zero-copy. - from_file() rehydrates the in-memory cache from disk, so hot reads hit the cache and the DB is only consulted for replay after a crash. - reset() is deliberately destructive: it clears the current run, not all runs. Multi-run persistence (keeping history across resets) is a Phase 3 milestone. """ from __future__ import annotations import json import sqlite3 from datetime import datetime, timezone from pathlib import Path from src.core.events import Event from src.core.ledger import Ledger class SQLiteLedger(Ledger): """Persistent append-only ledger backed by SQLite. Drop-in replacement for the in-memory Ledger. The same API, the same idempotency guarantee, plus durable storage and snapshot/restore. """ def __init__(self, path: str | Path = ":memory:") -> None: self._path = str(path) self._conn = sqlite3.connect(self._path, check_same_thread=False) self._conn.execute("PRAGMA journal_mode=WAL") self._conn.execute("PRAGMA synchronous=NORMAL") self._create_schema() self._cache: list[Event] = [] self._seen_ids: set[str] = set() if self._path != ":memory:": self._load_cache() # ── schema ──────────────────────────────────────────────────────────────── def _create_schema(self) -> None: self._conn.executescript(""" CREATE TABLE IF NOT EXISTS events ( offset INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT UNIQUE NOT NULL, run_id TEXT NOT NULL, turn INTEGER NOT NULL, kind TEXT NOT NULL, actor TEXT NOT NULL, payload TEXT NOT NULL, created_at TEXT NOT NULL, schema_version INTEGER NOT NULL DEFAULT 1 ); CREATE INDEX IF NOT EXISTS idx_run_id ON events(run_id); CREATE INDEX IF NOT EXISTS idx_kind ON events(kind); CREATE INDEX IF NOT EXISTS idx_actor ON events(actor); """) self._conn.commit() # ── Ledger API ──────────────────────────────────────────────────────────── def append(self, event: Event) -> Event: if event.id in self._seen_ids: return event try: self._conn.execute( "INSERT INTO events " "(id, run_id, turn, kind, actor, payload, created_at, schema_version) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", ( event.id, event.run_id, event.turn, event.kind, event.actor, json.dumps(event.payload), event.created_at.isoformat(), event.schema_version, ), ) self._conn.commit() self._cache.append(event) self._seen_ids.add(event.id) except sqlite3.IntegrityError: # Duplicate id inserted concurrently — safe to ignore. pass return event @property def events(self) -> tuple[Event, ...]: return tuple(self._cache) def reset(self) -> None: self._conn.execute("DELETE FROM events") self._conn.commit() self._cache.clear() self._seen_ids.clear() # ── persistence helpers ─────────────────────────────────────────────────── def snapshot_to(self, dest: str | Path) -> None: """Copy the database to *dest* atomically using SQLite's backup API.""" backup = sqlite3.connect(str(dest)) try: self._conn.backup(backup) finally: backup.close() @classmethod def from_file(cls, path: str | Path) -> "SQLiteLedger": """Open an existing database and rehydrate the in-memory cache.""" ledger = cls(path) return ledger def _load_cache(self) -> None: rows = self._conn.execute( "SELECT id, run_id, turn, kind, actor, payload, created_at, schema_version " "FROM events ORDER BY offset" ).fetchall() for row in rows: try: created_at = datetime.fromisoformat(row[6]) if created_at.tzinfo is None: created_at = created_at.replace(tzinfo=timezone.utc) except (ValueError, TypeError): created_at = datetime.now(timezone.utc) e = Event( id=row[0], run_id=row[1], turn=row[2], kind=row[3], # type: ignore[arg-type] actor=row[4], payload=json.loads(row[5]), created_at=created_at, schema_version=row[7], ) self._cache.append(e) self._seen_ids.add(e.id) def tail(self, from_offset: int = 0) -> tuple[Event, ...]: """Return events with offset > from_offset (for crash-recovery replay).""" rows = self._conn.execute( "SELECT id, run_id, turn, kind, actor, payload, created_at, schema_version " "FROM events WHERE offset > ? ORDER BY offset", (from_offset,), ).fetchall() events = [] for row in rows: e = Event( id=row[0], run_id=row[1], turn=row[2], kind=row[3], # type: ignore[arg-type] actor=row[4], payload=json.loads(row[5]), schema_version=row[7], ) events.append(e) return tuple(events) def latest_offset(self) -> int: row = self._conn.execute("SELECT MAX(offset) FROM events").fetchone() return row[0] or 0 def close(self) -> None: self._conn.close()