"""SQLite storage foundation with versioned schema and explicit retention.""" from __future__ import annotations import sqlite3 from collections.abc import Iterator from contextlib import contextmanager from datetime import UTC, datetime from pathlib import Path _MIGRATIONS: dict[int, tuple[str, ...]] = { 1: ( """ CREATE TABLE tasks ( task_id TEXT PRIMARY KEY, handler TEXT NOT NULL, payload_json TEXT NOT NULL, status TEXT NOT NULL, result_json TEXT, error TEXT, attempts INTEGER NOT NULL DEFAULT 0, created_at REAL NOT NULL, started_at REAL, completed_at REAL ) """, "CREATE INDEX tasks_status_created_idx ON tasks(status, created_at)", """ CREATE TABLE approval_requests ( request_id TEXT PRIMARY KEY, agent_json TEXT NOT NULL, action TEXT NOT NULL, resource TEXT, context_json TEXT NOT NULL, risk_score REAL NOT NULL, status TEXT NOT NULL, created_at REAL NOT NULL, reviewed_at REAL, reviewer TEXT, reason TEXT NOT NULL, decision TEXT, metadata_json TEXT NOT NULL ) """, "CREATE INDEX approvals_status_created_idx ON approval_requests(status, created_at)", """ CREATE TABLE approval_records ( request_id TEXT PRIMARY KEY REFERENCES approval_requests(request_id) ON DELETE CASCADE, agent_id TEXT NOT NULL, action TEXT NOT NULL, resource TEXT, decision TEXT NOT NULL, reviewer TEXT NOT NULL, timestamp REAL NOT NULL, reason TEXT NOT NULL ) """, """ CREATE TABLE audit_events ( event_id TEXT PRIMARY KEY, event_type TEXT NOT NULL, timestamp REAL NOT NULL, actor TEXT NOT NULL, agent_id TEXT NOT NULL, task_id TEXT NOT NULL, model TEXT NOT NULL, tool TEXT NOT NULL, action TEXT NOT NULL, resource TEXT NOT NULL, policy_decision TEXT NOT NULL, approval_id TEXT NOT NULL, result TEXT NOT NULL, error TEXT NOT NULL, resource_usage_json TEXT NOT NULL, latency_ms REAL NOT NULL, network_action TEXT NOT NULL, memory_write TEXT NOT NULL, security_alert TEXT NOT NULL, metadata_json TEXT NOT NULL, status TEXT NOT NULL ) """, "CREATE INDEX audit_task_time_idx ON audit_events(task_id, timestamp)", "CREATE INDEX audit_type_time_idx ON audit_events(event_type, timestamp)", """ CREATE TABLE project_memories ( memory_id TEXT PRIMARY KEY, project_id TEXT NOT NULL, content TEXT NOT NULL, trust_level TEXT NOT NULL, provenance_id TEXT, metadata_json TEXT NOT NULL, created_at REAL NOT NULL, expires_at REAL ) """, "CREATE INDEX memories_project_time_idx ON project_memories(project_id, created_at)", "CREATE INDEX memories_trust_idx ON project_memories(trust_level)", """ CREATE TABLE provenance_records ( content_id TEXT PRIMARY KEY, content_json TEXT NOT NULL, sha256 TEXT NOT NULL, metadata_json TEXT NOT NULL, labels_json TEXT NOT NULL, watermark_json TEXT, created_at REAL NOT NULL ) """, """ CREATE TABLE schema_migrations ( version INTEGER PRIMARY KEY, applied_at TEXT NOT NULL ) """, ), 2: ( """ CREATE TABLE knowledge_documents ( document_id TEXT PRIMARY KEY, source_id TEXT NOT NULL, source_uri TEXT NOT NULL, content_sha256 TEXT NOT NULL, media_type TEXT NOT NULL, trust_level TEXT NOT NULL, sensitivity TEXT NOT NULL, status TEXT NOT NULL, metadata_json TEXT NOT NULL, created_at REAL NOT NULL, updated_at REAL NOT NULL, UNIQUE(source_id, content_sha256) ) """, "CREATE INDEX knowledge_documents_source_idx ON knowledge_documents(source_id, updated_at)", "CREATE INDEX knowledge_documents_status_idx ON knowledge_documents(status, updated_at)", ), 3: ( # Hash chain for audit_events. Additive only: existing rows keep NULL in # the three new columns, which is the declared "epoch 0" boundary that # `audit.chain.verify_chain` reports as `unverified_premise` rather than # as a clean pass. Nothing is rewritten and no row is dropped. "ALTER TABLE audit_events ADD COLUMN seq INTEGER", "ALTER TABLE audit_events ADD COLUMN prev_hash TEXT", "ALTER TABLE audit_events ADD COLUMN entry_hash TEXT", "CREATE INDEX audit_chain_idx ON audit_events(seq)", ), } class SQLiteDatabase: """Local SQLite database shared by durable control-plane repositories. The database is created with owner-only file permissions where supported. Retention is explicit: callers invoke :meth:`prune_expired` from a deployment scheduler or maintenance command rather than deleting records on application startup. """ CURRENT_SCHEMA_VERSION = max(_MIGRATIONS) def __init__(self, path: str | Path, *, retention_days: int = 90) -> None: if retention_days <= 0: raise ValueError("retention_days must be positive") path_text = str(path) if path_text == ":memory:" or path_text.startswith("file:"): # ``file:`` covers the URI forms, e.g. ``file::memory:?cache=shared``. # Only the literal ":memory:" was rejected before, so a URI slipped # past this guard and ``resolve()`` turned it into a real file named # after the URI, written next to whatever the working directory was. raise ValueError("SQLiteDatabase requires a persistent file path") self.path = Path(path).expanduser().resolve() self.retention_days = retention_days parent_existed = self.path.parent.exists() self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) if not parent_existed: try: self.path.parent.chmod(0o700) except OSError: pass self._initialize() @contextmanager def connection(self) -> Iterator[sqlite3.Connection]: """Yield a configured connection and always close it afterward.""" connection = sqlite3.connect(self.path, timeout=5.0, isolation_level=None) connection.row_factory = sqlite3.Row connection.execute("PRAGMA foreign_keys = ON") connection.execute("PRAGMA busy_timeout = 5000") connection.execute("PRAGMA synchronous = FULL") try: yield connection finally: connection.close() @contextmanager def transaction(self) -> Iterator[sqlite3.Connection]: """Yield a write transaction with rollback on failure.""" with self.connection() as connection: connection.execute("BEGIN IMMEDIATE") try: yield connection except BaseException: connection.rollback() raise else: connection.commit() self._restrict_file_permissions() def backup(self, destination: str | Path) -> Path: """Create a consistent SQLite backup using the online backup API.""" target = Path(destination).expanduser().resolve() target.parent.mkdir(parents=True, exist_ok=True, mode=0o700) with self.connection() as source: backup = sqlite3.connect(target) try: source.backup(backup) finally: backup.close() try: target.chmod(0o600) except OSError: pass return target def checkpoint(self) -> None: """Checkpoint the WAL so the main database contains committed data.""" with self.connection() as connection: connection.execute("PRAGMA wal_checkpoint(TRUNCATE)") def count_audit_before(self, cutoff: float) -> int: """Count audit records older than ``cutoff``. Split out from :meth:`prune_expired` so the caller can inspect the chain *before* the evidence is removed. Retention is a deadline and must not be refused because the chain is broken — but whether it is broken has to be known first, otherwise the purge cannot record what it removed. """ with self.connection() as connection: row = connection.execute( "SELECT COUNT(*) AS n FROM audit_events WHERE timestamp < ?", (cutoff,), ).fetchone() return int(row["n"]) def prune_expired(self, *, now: float | None = None) -> dict[str, int]: """Apply the configured retention window to terminal records. Pending approval requests and unfinished tasks are never removed. Completed/failed tasks, decided approvals, and audit events older than the cutoff are deleted. This operation is explicit and auditable by the caller; SQLite does not securely erase old pages or WAL fragments. """ if now is None: now = datetime.now(UTC).timestamp() cutoff = now - self.retention_days * 24 * 60 * 60 with self.transaction() as connection: tasks = connection.execute( "DELETE FROM tasks WHERE status IN ('completed', 'failed') " "AND completed_at IS NOT NULL AND completed_at < ?", (cutoff,), ).rowcount approvals = connection.execute( "DELETE FROM approval_requests WHERE status IN " "('approved', 'denied', 'expired', 'cancelled') AND created_at < ?", (cutoff,), ).rowcount audit = connection.execute( "DELETE FROM audit_events WHERE timestamp < ?", (cutoff,) ).rowcount memories = connection.execute( "DELETE FROM project_memories WHERE expires_at IS NOT NULL AND expires_at <= ?", (now,), ).rowcount return { "tasks": tasks, "approvals": approvals, "audit_events": audit, "project_memories": memories, } def _initialize(self) -> None: with self.connection() as connection: connection.execute("PRAGMA journal_mode = WAL") connection.execute("PRAGMA synchronous = FULL") current = int(connection.execute("PRAGMA user_version").fetchone()[0]) if current > self.CURRENT_SCHEMA_VERSION: raise RuntimeError( f"database schema {current} is newer than supported schema " f"{self.CURRENT_SCHEMA_VERSION}" ) for version in range(current + 1, self.CURRENT_SCHEMA_VERSION + 1): statements = _MIGRATIONS.get(version) if statements is None: raise RuntimeError(f"missing SQLite migration {version}") connection.execute("BEGIN EXCLUSIVE") try: for statement in statements: connection.execute(statement) connection.execute( "INSERT INTO schema_migrations(version, applied_at) VALUES (?, ?)", (version, datetime.now(UTC).isoformat()), ) connection.execute(f"PRAGMA user_version = {version}") connection.commit() except BaseException: connection.rollback() raise self._restrict_file_permissions() def _restrict_file_permissions(self) -> None: try: self.path.chmod(0o600) except OSError: pass __all__ = ["SQLiteDatabase"]