"""A simple governance ledger using SQLite for auditability.""" import sqlite3 import time from typing import List, Dict, Any, cast from .replay import canonical_json class GovernanceLedger: def __init__(self, db_path: str = ":memory:"): self.conn = sqlite3.connect(db_path) self._init_schema() def _init_schema(self) -> None: cur = self.conn.cursor() cur.execute( """ CREATE TABLE IF NOT EXISTS governance_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, action TEXT NOT NULL, actor TEXT NOT NULL, timestamp REAL NOT NULL, status TEXT NOT NULL, details TEXT ) """ ) self.conn.commit() def log(self, action: str, actor: str, status: str, details: Any | None = None) -> int: ts = time.time() normalized_details = details if isinstance(details, (dict, list)): normalized_details = canonical_json(details) cur = self.conn.cursor() cur.execute( "INSERT INTO governance_logs (action, actor, timestamp, status, details) VALUES (?, ?, ?, ?, ?)", (action, actor, ts, status, normalized_details), ) self.conn.commit() return cast(int, cur.lastrowid) def log_plan(self, actor: str, status: str, plan_hash: str, details: Dict[str, Any] | None = None) -> int: payload = {"plan_hash": plan_hash} if details: payload.update(details) return self.log(action="plan", actor=actor, status=status, details=payload) def list_logs(self) -> List[Dict[str, Any]]: cur = self.conn.cursor() cur.execute("SELECT id, action, actor, timestamp, status, details FROM governance_logs ORDER BY timestamp DESC") rows = cur.fetchall() return [ { "id": r[0], "action": r[1], "actor": r[2], "timestamp": r[3], "status": r[4], "details": r[5], } for r in rows ]