"""SQLite-backed project memory and local provenance provider.""" from __future__ import annotations import hashlib import json import re import time from dataclasses import dataclass, field from enum import Enum from typing import Any from uuid import uuid4 from cyberbrain26.data_plane.sqlite_store import SQLiteDatabase from cyberbrain26.extension_layer.provenance_provider import ProvenanceProvider from cyberbrain26.extension_layer.retrieval_provider import RetrievalProvider from cyberbrain26.security_plane.audit import AuditEventType, AuditLog from cyberbrain26.security_plane.privacy_gateway import PrivacyGateway class MemoryTrustLevel(str, Enum): """Origin/review label, not a claim that memory content is safe to execute.""" UNTRUSTED = "untrusted" EXTERNAL = "external" REVIEWED = "reviewed" TRUSTED = "trusted" @dataclass class ProjectMemoryRecord: memory_id: str project_id: str content: str trust_level: MemoryTrustLevel provenance_id: str | None = None metadata: dict[str, Any] = field(default_factory=dict) created_at: float = 0.0 expires_at: float | None = None class SQLiteMemoryProvider(RetrievalProvider, ProvenanceProvider): """Local keyword retrieval, trust-labeled project memory, and provenance. Provenance hashes detect accidental content changes but are not signatures or protection against a database owner who can rewrite both content and hash. The provider records labels/watermark metadata; it does not embed watermarks into binary media. """ _TOKEN_RE = re.compile(r"[\w-]+", re.UNICODE) def __init__( self, database: SQLiteDatabase, *, audit_log: AuditLog | None = None, privacy_gateway: PrivacyGateway | None = None, redact_on_write: bool = True, ) -> None: self.database = database self.audit_log = audit_log self.privacy_gateway = privacy_gateway or PrivacyGateway(audit_log=audit_log) self.redact_on_write = redact_on_write @property def strategy_name(self) -> str: return "sqlite-keyword" @property def provider_name(self) -> str: return "sqlite-local-provenance" def add_memory( self, project_id: str, content: str, *, trust_level: MemoryTrustLevel = MemoryTrustLevel.EXTERNAL, provenance_id: str | None = None, metadata: dict[str, Any] | None = None, expires_at: float | None = None, memory_id: str | None = None, ) -> ProjectMemoryRecord: """Store a project-scoped memory with an explicit source trust label.""" if not project_id.strip(): raise ValueError("project_id must not be empty") if not isinstance(content, str) or not content.strip(): raise ValueError("memory content must be a non-empty string") redaction_findings: dict[str, int] = {} if self.redact_on_write: privacy_result = self.privacy_gateway.redact(content, context="project_memory") content = privacy_result.text redaction_findings = { finding.category: finding.count for finding in privacy_result.findings } record = ProjectMemoryRecord( memory_id=memory_id or f"memory_{uuid4().hex}", project_id=project_id, content=content, trust_level=MemoryTrustLevel(trust_level), provenance_id=provenance_id, metadata=dict(metadata or {}), created_at=time.time(), expires_at=expires_at, ) metadata_json = json.dumps(record.metadata, separators=(",", ":"), allow_nan=False) with self.database.transaction() as connection: connection.execute( """ INSERT INTO project_memories( memory_id, project_id, content, trust_level, provenance_id, metadata_json, created_at, expires_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( record.memory_id, record.project_id, record.content, record.trust_level.value, record.provenance_id, metadata_json, record.created_at, record.expires_at, ), ) if self.audit_log is not None: self.audit_log.record( AuditEventType.MEMORY_UPDATED, action="project_memory_write", resource=project_id, memory_write=record.memory_id, metadata={ "trust_level": record.trust_level.value, "provenance_id": record.provenance_id, "redactions": redaction_findings, }, ) return record def get_memory(self, memory_id: str) -> ProjectMemoryRecord | None: with self.database.connection() as connection: row = connection.execute( "SELECT * FROM project_memories WHERE memory_id = ?", (memory_id,) ).fetchone() return self._row_to_memory(row) if row is not None else None async def retrieve( self, query: str, *, limit: int = 10, filters: dict[str, Any] | None = None, ) -> list[dict[str, Any]]: """Retrieve keyword matches, excluding untrusted memories by default.""" if limit <= 0: return [] filters = filters or {} project_id = filters.get("project_id") raw_levels = filters.get("trust_levels") if raw_levels is None: allowed = { MemoryTrustLevel.EXTERNAL.value, MemoryTrustLevel.REVIEWED.value, MemoryTrustLevel.TRUSTED.value, } else: if isinstance(raw_levels, (str, MemoryTrustLevel)): raw_levels = [raw_levels] allowed = { value.value if isinstance(value, MemoryTrustLevel) else str(value) for value in raw_levels } allowed &= {level.value for level in MemoryTrustLevel} placeholders = ",".join("?" for _ in allowed) if not placeholders: return [] sql = f"SELECT * FROM project_memories WHERE trust_level IN ({placeholders})" parameters: list[Any] = sorted(allowed) if project_id is not None: sql += " AND project_id = ?" parameters.append(str(project_id)) sql += " AND (expires_at IS NULL OR expires_at > ?)" parameters.append(time.time()) sql += " ORDER BY created_at DESC" with self.database.connection() as connection: rows = connection.execute(sql, parameters).fetchall() query_tokens = {token.casefold() for token in self._TOKEN_RE.findall(query)} ranked: list[tuple[int, float, ProjectMemoryRecord]] = [] for row in rows: record = self._row_to_memory(row) content_tokens = {token.casefold() for token in self._TOKEN_RE.findall(record.content)} score = len(query_tokens & content_tokens) if not query_tokens or score: ranked.append((score, record.created_at, record)) ranked.sort(key=lambda item: (item[0], item[1]), reverse=True) return [ { "memory_id": record.memory_id, "project_id": record.project_id, "content": record.content, "trust_level": record.trust_level.value, "provenance_id": record.provenance_id, "metadata": record.metadata, "created_at": record.created_at, "score": score, } for score, _, record in ranked[:limit] ] async def attest( self, content_id: str, content: dict[str, Any], *, metadata: dict[str, Any] | None = None, ) -> dict[str, Any]: """Store canonical JSON and a SHA-256 integrity digest locally.""" if not content_id.strip(): raise ValueError("content_id must not be empty") content_json = json.dumps( content, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False ) content_hash = hashlib.sha256(content_json.encode("utf-8")).hexdigest() metadata_json = json.dumps(metadata or {}, separators=(",", ":"), allow_nan=False) created_at = time.time() with self.database.transaction() as connection: connection.execute( """ INSERT INTO provenance_records( content_id, content_json, sha256, metadata_json, labels_json, watermark_json, created_at ) VALUES (?, ?, ?, ?, '[]', NULL, ?) """, (content_id, content_json, content_hash, metadata_json, created_at), ) return { "content_id": content_id, "sha256": content_hash, "provider": self.provider_name, "created_at": created_at, "integrity_only": True, "signature_verified": False, } async def verify(self, content_id: str) -> dict[str, Any] | None: with self.database.connection() as connection: row = connection.execute( "SELECT * FROM provenance_records WHERE content_id = ?", (content_id,) ).fetchone() if row is None: return None computed_hash = hashlib.sha256(row["content_json"].encode("utf-8")).hexdigest() return { "content_id": content_id, "sha256": row["sha256"], "integrity_valid": computed_hash == row["sha256"], "metadata": json.loads(row["metadata_json"]), "labels": json.loads(row["labels_json"]), "watermark": json.loads(row["watermark_json"]) if row["watermark_json"] else None, "created_at": row["created_at"], "integrity_only": True, "signature_verified": False, } async def watermark( self, content_id: str, label: str, *, visible: bool = False, ) -> dict[str, Any]: watermark = { "label": label, "visible": visible, "recorded_at": time.time(), "applied_to_content": False, } with self.database.transaction() as connection: updated = connection.execute( "UPDATE provenance_records SET watermark_json = ? WHERE content_id = ?", (json.dumps(watermark, separators=(",", ":")), content_id), ).rowcount if updated: return watermark return { "content_id": content_id, "label": label, "visible": visible, "applied_to_content": False, "error": "content not found", } async def label(self, content_id: str, labels: list[str]) -> bool: with self.database.transaction() as connection: row = connection.execute( "SELECT labels_json FROM provenance_records WHERE content_id = ?", (content_id,), ).fetchone() if row is None: return False merged = sorted(set(json.loads(row["labels_json"])) | set(labels)) connection.execute( "UPDATE provenance_records SET labels_json = ? WHERE content_id = ?", (json.dumps(merged, separators=(",", ":")), content_id), ) return True @staticmethod def _row_to_memory(row: Any) -> ProjectMemoryRecord: return ProjectMemoryRecord( memory_id=row["memory_id"], project_id=row["project_id"], content=row["content"], trust_level=MemoryTrustLevel(row["trust_level"]), provenance_id=row["provenance_id"], metadata=json.loads(row["metadata_json"]), created_at=row["created_at"], expires_at=row["expires_at"], ) __all__ = ["MemoryTrustLevel", "ProjectMemoryRecord", "SQLiteMemoryProvider"]