db.py
Rama main · réplica de lectura del repositorio original.
"""piSynapse Database Layer
SQLite with aiosqlite: conversations, sessions, and long-term memory.
Single persistent connection with WAL mode for Pi-friendly I/O.
"""
import asyncio
import csv
import gzip
import hashlib
import hmac
import json
import logging
import math
import os
import re
import secrets
import time
import aiosqlite
from config import DB_PATH
logger = logging.getLogger("piSynapse")
# DB files hold conversations, memories and audit params (incl. e-mail
# bodies). Restrict the process umask so any file SQLite creates
# (assistant.db / -wal / -shm / -journal) is born 0o600, never world-readable.
os.umask(0o077)
_db: aiosqlite.Connection | None = None
_db_lock = asyncio.Lock()
# Backfill task control (prevents unbounded embedding tasks on legacy memories)
_BACKFILL_SEMAPHORE = asyncio.Semaphore(2)
_BACKFILL_BATCH_SIZE = 10
_backfill_task: asyncio.Task | None = None
_LOCKED_RETRIES = 3
# Serializes grouped multi-statement writes on the single shared connection.
# aiosqlite uses legacy isolation (implicit BEGIN), so an explicit
# BEGIN IMMEDIATE would risk "cannot start a transaction within a
# transaction"; since every writer runs on the same event loop, an
# asyncio.Lock gives the same no-interleaving guarantee with zero
# SQLite-semantics risk. Slow work (embeddings) must stay OUTSIDE the lock.
# The lock is never re-entered: db writers never call each other.
# One lock per event loop: a module-level lock stays bound to the first loop
# that acquires it, which breaks callers that (like the test-suite) drive the
# module from several asyncio.run loops over the process lifetime.
_write_locks: dict = {}
def _write_lock():
"""Return the write-serialization lock for the running loop."""
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = None
lock = _write_locks.get(loop)
if lock is None:
lock = asyncio.Lock()
# Prune locks of dead loops so the registry cannot grow unbounded.
for known in [k for k in _write_locks if k is not None and k.is_closed()]:
_write_locks.pop(known, None)
_write_locks[loop] = lock
return lock
async def _write_with_retry(db: aiosqlite.Connection, sql: str, params: tuple = ()) -> aiosqlite.Cursor:
"""Execute a write, retrying briefly on 'database is locked' / 'busy'."""
import sqlite3
for attempt in range(_LOCKED_RETRIES + 1):
try:
cur = await db.execute(sql, params)
await db.commit()
return cur
except sqlite3.OperationalError as e:
if "locked" not in str(e).lower() and "busy" not in str(e).lower():
raise
if attempt >= _LOCKED_RETRIES:
raise
logger.warning(f"SQLite busy, retrying ({attempt + 1}/{_LOCKED_RETRIES})...")
await asyncio.sleep(0.25 * (attempt + 1))
async def _commit_with_retry(db: aiosqlite.Connection) -> None:
"""Commit a multi-statement write, retrying briefly on locked/busy.
Complementary to _write_with_retry for hand-rolled transaction bodies
(save_message, clear_history, ...) whose statements must stay grouped in
a single transaction. busy_timeout already buffers most contention; this
is a secondary safety net on the critical chat completion path.
"""
import sqlite3
for attempt in range(_LOCKED_RETRIES + 1):
try:
await db.commit()
return
except sqlite3.OperationalError as e:
if "locked" not in str(e).lower() and "busy" not in str(e).lower():
raise
if attempt >= _LOCKED_RETRIES:
raise
logger.warning(f"SQLite busy on commit, retrying ({attempt + 1}/{_LOCKED_RETRIES})...")
await asyncio.sleep(0.25 * (attempt + 1))
async def _secure_db_files() -> None:
"""Force owner-only permissions on the DB and its SQLite sidecars.
The DB is created on first run (not by install.py), so this runs on
every startup as a guarantee for fresh and pre-existing installs.
Covers every *.db* file in the DB directory (not just the configured
path) so stray world-readable copies cannot linger.
"""
paths = [DB_PATH, DB_PATH + "-wal", DB_PATH + "-shm", DB_PATH + "-journal"]
db_dir = os.path.dirname(os.path.abspath(DB_PATH)) or "."
try:
for name in os.listdir(db_dir):
if name.endswith(".db") or ".db-" in name:
paths.append(os.path.join(db_dir, name))
except OSError as e:
logger.warning(f"Could not list DB directory {db_dir}: {e}")
for path in dict.fromkeys(os.path.abspath(p) for p in paths):
try:
if os.path.exists(path):
os.chmod(path, 0o600)
except OSError as e:
logger.warning(f"Could not secure permissions on {path}: {e}")
async def get_db() -> aiosqlite.Connection:
global _db
if _db is not None:
try:
await _db.execute("SELECT 1")
return _db
except Exception:
logger.warning("SQLite connection lost, reconnecting...")
_db = None
async with _db_lock:
if _db is not None:
return _db
conn = await aiosqlite.connect(DB_PATH)
await conn.execute("PRAGMA journal_mode=WAL")
await conn.execute("PRAGMA synchronous=NORMAL")
await conn.execute("PRAGMA cache_size=10000")
await conn.execute("PRAGMA temp_store=MEMORY")
await conn.execute("PRAGMA foreign_keys=ON")
await conn.execute("PRAGMA busy_timeout=10000")
await conn.execute("PRAGMA journal_size_limit=67108864")
await _secure_db_files()
_db = conn
return _db
async def close_db():
global _db
# Stop any in-flight embedding backfill (fire-and-forget: it runs on its
# own connection, so this is housekeeping, not a correctness requirement).
if _backfill_task is not None and not _backfill_task.done():
_backfill_task.cancel()
if _db is not None:
await _db.close()
_db = None
# -- Schema --
# Ordered schema migrations. `user_version` tracks how many are applied.
#
# IMPORTANT — keep this list in sync with the CREATE TABLE definitions in
# init_db() below. The CREATE TABLEs describe the full CURRENT schema (they
# already include every column listed here) and are what brand-new databases
# get. MIGRATIONS only matters for pre-existing databases whose tables were
# created before a column existed. When adding a column:
# 1. add it to the matching CREATE TABLE statement (new DBs), AND
# 2. append a new entry at the END of MIGRATIONS (old DBs).
# Never reorder, edit or remove an existing entry: user_version numbers are
# baked into live databases and renumbering would re-run/skip migrations.
MIGRATIONS: list[tuple[str, str, str]] = [
("conversations", "images", "TEXT"),
("sessions", "name", "TEXT"),
("sessions", "summarized_until", "INTEGER DEFAULT 0"),
("memories", "embedding", "BLOB"),
("conversations", "reasoning", "TEXT"),
("conversations", "embedding", "BLOB"),
("tool_audit_log", "expected_tool", "TEXT"),
("tool_audit_log", "corrected_at", "DATETIME"),
("tool_audit_log", "verification_status", "TEXT"),
("tool_audit_log", "expected_group", "TEXT"),
("tool_audit_log", "confirmed_at", "DATETIME"),
("tool_audit_log", "conversation_id", "INTEGER"),
("conversations", "client_key", "TEXT"),
("conversations", "user_id", "TEXT DEFAULT 'default'"),
("sessions", "user_id", "TEXT DEFAULT 'default'"),
("email_session_map", "user_id", "TEXT DEFAULT 'default'"),
("notes_session_map", "user_id", "TEXT DEFAULT 'default'"),
("tasks_session_map", "user_id", "TEXT DEFAULT 'default'"),
("calendar_session_map", "user_id", "TEXT DEFAULT 'default'"),
("tool_audit_log", "user_id", "TEXT"),
("users", "password_hash", "TEXT"),
("users", "is_approved", "INTEGER DEFAULT 0"),
("sync_items", "origin", "TEXT NOT NULL DEFAULT ''"),
("tool_audit_log", "stats_json", "TEXT"),
]
# One-time data backfills, keyed by the migration index that requires them.
# Unlike the every-boot repairs in init_db, these run exactly once (inside
# _apply_migrations) so later admin edits are never overwritten.
_DATA_BACKFILLS: dict[int, list[str]] = {
# Grandfather pre-approval rows: everyone on board keeps counting.
21: ["UPDATE users SET is_approved = 1"],
}
async def _get_schema_version(db: aiosqlite.Connection) -> int:
cur = await db.execute("PRAGMA user_version")
row = await cur.fetchone()
return int(row[0]) if row else 0
_IDENT_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
_DDL_UNSAFE_RE = re.compile(r";|--|/\*|\*/")
def _safe_ident(name: str) -> str:
"""Quote a schema identifier; rejects anything but plain names."""
if not _IDENT_RE.fullmatch(name or ""):
raise ValueError(f"Unsafe schema identifier: {name!r}")
return f'"{name}"'
def _safe_ddl_definition(definition: str) -> str:
"""Allow column definitions built from types/defaults only.
Blocks statement stacking (;) and comment breakout (--, /* */), the
realistic injection vectors for DDL assembled from parts. (The
``PRAGMA user_version = {i + 1}`` below needs no guard: ``i`` is a
range() int by construction.)
"""
if not definition or _DDL_UNSAFE_RE.search(definition):
raise ValueError(f"Unsafe column definition: {definition!r}")
return definition
async def _apply_migrations(db: aiosqlite.Connection):
version = await _get_schema_version(db)
for i in range(version, len(MIGRATIONS)):
table, column, definition = MIGRATIONS[i]
try:
await _write_with_retry(
db,
f"ALTER TABLE {_safe_ident(table)} ADD COLUMN {_safe_ident(column)} "
f"{_safe_ddl_definition(definition)}",
)
except Exception as e:
if "duplicate column" not in str(e).lower():
logger.warning(f"Migration {table}.{column} failed: {e}")
break
for _stmt in _DATA_BACKFILLS.get(i, []):
await db.execute(_stmt)
await db.execute(f"PRAGMA user_version = {i + 1}")
await _commit_with_retry(db)
async def init_db():
db = await get_db()
await db.execute("""
CREATE TABLE IF NOT EXISTS conversations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
role TEXT NOT NULL,
content TEXT NOT NULL,
images TEXT,
reasoning TEXT,
embedding BLOB,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
client_key TEXT,
user_id TEXT DEFAULT 'default'
)
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS message_feedback (
id INTEGER PRIMARY KEY AUTOINCREMENT,
message_id INTEGER NOT NULL UNIQUE,
value TEXT NOT NULL CHECK (value IN ('up','down')),
note TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
last_active DATETIME DEFAULT CURRENT_TIMESTAMP,
name TEXT,
summary TEXT,
summarized_until INTEGER DEFAULT 0,
user_id TEXT DEFAULT 'default'
)
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS users (
id TEXT PRIMARY KEY,
name TEXT NOT NULL DEFAULT '',
key_hash TEXT NOT NULL UNIQUE,
is_admin INTEGER NOT NULL DEFAULT 0,
password_hash TEXT,
is_approved INTEGER NOT NULL DEFAULT 0,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS user_api_keys (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
name TEXT NOT NULL DEFAULT '',
key_hash TEXT NOT NULL UNIQUE,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
await db.execute("CREATE INDEX IF NOT EXISTS idx_user_api_keys_user ON user_api_keys(user_id)")
await db.execute("""
CREATE TABLE IF NOT EXISTS user_sessions (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
token_hash TEXT NOT NULL UNIQUE,
device TEXT NOT NULL DEFAULT '',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
expires_at DATETIME NOT NULL,
last_seen DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
await db.execute("CREATE INDEX IF NOT EXISTS idx_user_sessions_user ON user_sessions(user_id)")
await db.execute("""
CREATE TABLE IF NOT EXISTS user_settings (
user_id TEXT NOT NULL,
key TEXT NOT NULL,
value TEXT NOT NULL DEFAULT '',
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, key)
)
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS memories (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL DEFAULT 'default',
content TEXT NOT NULL,
category TEXT,
importance INTEGER DEFAULT 5,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
last_accessed DATETIME DEFAULT CURRENT_TIMESTAMP,
access_count INTEGER DEFAULT 0,
embedding BLOB
)
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS tool_audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
tool_name TEXT NOT NULL,
params TEXT,
success INTEGER NOT NULL,
duration_ms REAL,
error TEXT,
is_summary INTEGER NOT NULL DEFAULT 0,
day TEXT,
total_calls INTEGER,
success_count INTEGER,
error_count INTEGER,
tool_breakdown TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
expected_tool TEXT,
corrected_at DATETIME,
expected_group TEXT,
verification_status TEXT,
confirmed_at DATETIME,
conversation_id INTEGER,
user_id TEXT,
stats_json TEXT
)
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS intent_audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
message TEXT NOT NULL,
chosen_group TEXT,
best_sim REAL,
margin REAL,
source TEXT NOT NULL,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS email_session_map (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
seq INTEGER NOT NULL,
message_id TEXT NOT NULL,
subject TEXT DEFAULT '',
sender TEXT DEFAULT '',
preview TEXT DEFAULT '',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
user_id TEXT DEFAULT 'default',
UNIQUE(session_id, seq)
)
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS feedback_votes (
audit_id INTEGER NOT NULL,
user_id TEXT NOT NULL,
signature TEXT NOT NULL DEFAULT '',
proposed_group TEXT NOT NULL DEFAULT '',
signal TEXT NOT NULL DEFAULT 'confirm',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
UNIQUE(audit_id, user_id)
)
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS pattern_approvals (
signature TEXT PRIMARY KEY,
proposed_group TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL DEFAULT 'approved',
decided_by TEXT NOT NULL DEFAULT '',
decided_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS user_credentials (
user_id TEXT NOT NULL,
provider TEXT NOT NULL,
enc_blob TEXT NOT NULL,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, provider)
)
""")
await db.execute("CREATE INDEX IF NOT EXISTS idx_conversations_session ON conversations(session_id, timestamp)")
await db.execute("CREATE INDEX IF NOT EXISTS idx_memories_user ON memories(user_id, importance DESC)")
await db.execute("CREATE INDEX IF NOT EXISTS idx_tool_audit_created ON tool_audit_log(created_at)")
await db.execute("CREATE INDEX IF NOT EXISTS idx_tool_audit_rollup ON tool_audit_log(is_summary, created_at)")
await db.execute("CREATE INDEX IF NOT EXISTS idx_intent_audit_created ON intent_audit_log(created_at)")
await db.execute("CREATE INDEX IF NOT EXISTS idx_intent_audit_source ON intent_audit_log(source, created_at)")
await db.execute("CREATE INDEX IF NOT EXISTS idx_email_session_map_session ON email_session_map(session_id, seq)")
await db.execute("CREATE INDEX IF NOT EXISTS idx_feedback_votes_sig ON feedback_votes(signature, proposed_group)")
# Legacy repairs (dedupe + backfill + unique gate) run after
# _apply_migrations below: they reference migrated columns that may not
# exist yet on old DBs.
await db.execute("CREATE INDEX IF NOT EXISTS idx_memories_created ON memories(created_at)")
await db.execute("CREATE INDEX IF NOT EXISTS idx_tool_audit_conv ON tool_audit_log(conversation_id)")
await db.execute(
"CREATE INDEX IF NOT EXISTS idx_tool_audit_day "
"ON tool_audit_log(substr(created_at, 1, 10))"
)
await db.execute("""
CREATE TABLE IF NOT EXISTS notes_session_map (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
seq INTEGER NOT NULL,
note_id INTEGER NOT NULL,
title TEXT DEFAULT '',
category TEXT DEFAULT '',
preview TEXT DEFAULT '',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
user_id TEXT DEFAULT 'default',
UNIQUE(session_id, seq)
)
""")
await db.execute("CREATE INDEX IF NOT EXISTS idx_notes_session_map_session ON notes_session_map(session_id, seq)")
await db.execute("""
CREATE TABLE IF NOT EXISTS tasks_session_map (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
seq INTEGER NOT NULL,
uid TEXT NOT NULL,
summary TEXT DEFAULT '',
due TEXT DEFAULT '',
priority INTEGER DEFAULT 0,
completed INTEGER DEFAULT 0,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
user_id TEXT DEFAULT 'default',
UNIQUE(session_id, seq)
)
""")
await db.execute("CREATE INDEX IF NOT EXISTS idx_tasks_session_map_session ON tasks_session_map(session_id, seq)")
await db.execute("""
CREATE TABLE IF NOT EXISTS calendar_session_map (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
seq INTEGER NOT NULL,
uid TEXT NOT NULL,
summary TEXT DEFAULT '',
start_time TEXT DEFAULT '',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
user_id TEXT DEFAULT 'default',
UNIQUE(session_id, seq)
)
""")
await db.execute("CREATE INDEX IF NOT EXISTS idx_calendar_session_map_session ON calendar_session_map(session_id, seq)")
# ── Two-way device sync store ──────────────────────────────────────────
# Generic last-write-wins entity table: phones push local items (tasks,
# phone-side notes/memory) and pull other devices' changes. `deleted` is a
# tombstone so removals propagate. `uuid` is the client-generated stable id.
await db.execute("""
CREATE TABLE IF NOT EXISTS sync_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
uuid TEXT NOT NULL,
user_id TEXT NOT NULL DEFAULT 'default',
entity_type TEXT NOT NULL,
data TEXT NOT NULL DEFAULT '{}',
updated_at TEXT NOT NULL,
deleted INTEGER NOT NULL DEFAULT 0,
origin TEXT NOT NULL DEFAULT '',
UNIQUE(user_id, uuid)
)
""")
await db.execute("CREATE INDEX IF NOT EXISTS idx_sync_items_user_type ON sync_items(user_id, entity_type, updated_at)")
# FTS5 full-text index for session message search
# unicode61 tokenizer: proper Turkish/diacritik support
# content= + content_rowid= : external-content table (conversations)
await db.execute("""
CREATE VIRTUAL TABLE IF NOT EXISTS conversations_fts
USING fts5(content, session_id, content='conversations', content_rowid='id',
tokenize='unicode61 remove_diacritics 2')
""")
# Migrate old ascii tokenizer → unicode61 (one-time, only if needed).
# CREATE IF NOT EXISTS won't recreate an existing table with the old
# tokenizer, so we must detect and rebuild the schema explicitly.
try:
async with db.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='conversations_fts'") as cur:
row = await cur.fetchone()
fts_sql = row[0] if row else ""
if fts_sql and "unicode61" not in fts_sql:
logger.info("FTS5 tokenizer migration: ascii → unicode61")
await db.execute("DROP TABLE IF EXISTS conversations_fts")
await db.execute("""
CREATE VIRTUAL TABLE conversations_fts
USING fts5(content, session_id, content='conversations', content_rowid='id',
tokenize='unicode61 remove_diacritics 2')
""")
await db.execute("""
INSERT INTO conversations_fts (rowid, content, session_id)
SELECT id, content, session_id FROM conversations
""")
else:
# Light drift check: rebuild only if row counts diverge
# (e.g. crash mid-write). Avoids O(N) rebuild on every startup.
async with db.execute("SELECT COUNT(*) FROM conversations") as cur:
conv_count = (await cur.fetchone())[0]
async with db.execute("SELECT COUNT(*) FROM conversations_fts") as cur:
fts_count = (await cur.fetchone())[0]
if conv_count != fts_count:
logger.info(f"FTS5 drift detected (conversations={conv_count}, fts={fts_count}) — rebuilding")
await db.execute("INSERT INTO conversations_fts(conversations_fts) VALUES('rebuild')")
except Exception as e:
logger.warning(f"FTS5 migration/drift check failed (non-fatal): {e}")
await _apply_migrations(db)
# Legacy repairs run here (after migrations): they reference migrated
# columns that may not exist yet on old DBs. All idempotent; scans are
# cheap at realistic sizes (a version gate was considered and rejected:
# it risks silently skipping repairs after partial runs, saving <1s).
# Audit ownership first (it joins live owner values), then NULL-norm.
await db.execute(
"UPDATE tool_audit_log SET user_id = "
"(SELECT c.user_id FROM conversations c WHERE c.id = tool_audit_log.conversation_id) "
"WHERE conversation_id IS NOT NULL AND user_id IS NULL "
"AND (SELECT c.user_id FROM conversations c WHERE c.id = tool_audit_log.conversation_id) IS NOT NULL"
)
for _owner_tbl, _owner_col in (("conversations", "user_id"), ("sessions", "user_id"),
("memories", "user_id"), ("tool_audit_log", "user_id")):
await db.execute(f"UPDATE {_owner_tbl} SET {_owner_col} = 'default' WHERE {_owner_col} IS NULL")
# Canonicalize stored sync timestamps (idempotent; see helper doc) so
# LWW and since-comparisons are valid on legacy rows too.
await _normalize_stored_sync_timestamps(db)
doomed_cur = await db.execute(
"SELECT id FROM conversations WHERE client_key IS NOT NULL AND id NOT IN "
"(SELECT MIN(id) FROM conversations WHERE client_key IS NOT NULL "
"GROUP BY user_id, client_key)"
)
doomed = [r[0] for r in await doomed_cur.fetchall()]
if doomed:
logger.warning(
"Removing %d duplicate import rows before enforcing "
"UNIQUE(user_id, client_key)", len(doomed),
)
ph = ",".join("?" * len(doomed))
try:
await db.execute(f"DELETE FROM conversations_fts WHERE rowid IN ({ph})", doomed)
except Exception as fts_e:
# External-content FTS5 quirk: when NONE of the target rowids are
# indexed (past best-effort FTS misses), the DELETE raises
# "malformed" instead of deleting nothing. That same absence means
# there is nothing to orphan, so dropping the FTS leg is safe —
# and init_db must never crash on legacy data (startup SPOF).
logger.warning(f"FTS cleanup skipped during dedupe (non-fatal): {fts_e}")
await db.execute(f"DELETE FROM conversations WHERE id IN ({ph})", doomed)
await db.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS idx_conversations_client_key "
"ON conversations(user_id, client_key) WHERE client_key IS NOT NULL"
)
for _map_tbl in ("email_session_map", "notes_session_map", "tasks_session_map", "calendar_session_map"):
await db.execute(
f"UPDATE {_map_tbl} SET user_id = "
f"(SELECT s.user_id FROM sessions s WHERE s.id = {_map_tbl}.session_id) "
f"WHERE EXISTS (SELECT 1 FROM sessions s WHERE s.id = {_map_tbl}.session_id "
f"AND s.user_id != 'default')"
)
await cleanup_expired_data()
await _commit_with_retry(db)
# Sidecar -wal/-shm now exist after the first write; lock them down too.
await _secure_db_files()
async def _vacuum_if_fragmented(db: aiosqlite.Connection) -> None:
"""Compact the DB file only when it is mostly free pages.
The tool-audit rollup deletes thousands of detail rows every day, leaving
free pages behind; an unconditional daily VACUUM would add pointless I/O
and write-lock contention on every cleanup run. Triggering only above a
20% free-page ratio (and only for a meaningfully large DB) keeps the file
compact in practice with negligible overhead. Never raises.
"""
try:
async with db.execute("PRAGMA freelist_count") as cur:
free = (await cur.fetchone())[0]
async with db.execute("PRAGMA page_count") as cur:
total = (await cur.fetchone())[0]
if total > 256 and free / total > 0.2:
await db.execute("VACUUM")
logger.info(f"VACUUM: reclaimed {free} free pages (file was {total} pages)")
except Exception as e:
logger.warning(f"VACUUM skipped: {e}")
async def _retention_prune_conversations(db: aiosqlite.Connection) -> int:
"""Delete conversations older than retention + purge their orphans.
Runs inside the caller's write lock. Returns the removed row count.
"""
from config import CONVERSATION_RETENTION_DAYS
cur = await _write_with_retry(
db,
"DELETE FROM conversations WHERE timestamp < datetime('now', ?)",
(f"-{CONVERSATION_RETENTION_DAYS} days",),
)
removed = cur.rowcount if cur.rowcount else 0
# Purge matching FTS5 rows
await _write_with_retry(
db,
"DELETE FROM conversations_fts WHERE rowid NOT IN (SELECT id FROM conversations)",
)
await _write_with_retry(
db,
"DELETE FROM sessions WHERE id NOT IN (SELECT DISTINCT session_id FROM conversations)"
)
# Orphan feedback/audits of pruned messages (same pattern as FTS).
await _write_with_retry(
db,
"DELETE FROM message_feedback WHERE message_id NOT IN (SELECT id FROM conversations)",
)
await _write_with_retry(
db,
"DELETE FROM tool_audit_log WHERE conversation_id IS NOT NULL "
"AND conversation_id NOT IN (SELECT id FROM conversations)",
)
# Surviving sessions may point past the pruned tail.
await _repair_summary_boundaries(db)
await _commit_with_retry(db)
return removed
async def db_quick_check() -> str | None:
"""Run PRAGMA quick_check; return None when healthy, else a detail string."""
db = await get_db()
cur = await db.execute("PRAGMA quick_check")
rows = await cur.fetchall()
if len(rows) == 1 and str(rows[0][0]).lower() == "ok":
return None
return "; ".join(str(r[0]) for r in rows[:5])
AUTO_BACKUP_KEEP = 7
AUTO_BACKUP_PREFIX = "piSynapse-auto-"
async def auto_backup_db() -> str | None:
"""Online backup over a dedicated read-only connection, pruning generations.
Uses the sqlite3 online-backup API (chunked + yielding) instead of
VACUUM INTO, which refuses to run while the shared connection has open
statements. Owner-only 0600, atomic tmp+replace, keeps the newest
AUTO_BACKUP_KEEP generations. Returns the backup path or None (never
raises). Runs at startup and from the daily loop.
"""
tmp = None
try:
import datetime as _dt
import sqlite3 as _sqlite3
db_path = os.path.abspath(DB_PATH)
backup_dir = os.path.join(os.path.dirname(db_path) or ".", "backups")
os.makedirs(backup_dir, mode=0o700, exist_ok=True)
stamp = _dt.datetime.now().strftime("%Y%m%d")
dest = os.path.join(backup_dir, f"{AUTO_BACKUP_PREFIX}{stamp}.db")
tmp = dest + ".tmp"
def _copy() -> None:
src = _sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=30)
try:
dst = _sqlite3.connect(tmp, timeout=30)
try:
src.backup(dst, pages=100, sleep=0.05)
finally:
dst.close()
finally:
src.close()
await asyncio.to_thread(_copy)
os.chmod(tmp, 0o600)
os.replace(tmp, dest)
gens = sorted(
f for f in os.listdir(backup_dir)
if f.startswith(AUTO_BACKUP_PREFIX) and f.endswith(".db")
)
for old in gens[:-AUTO_BACKUP_KEEP]:
try:
os.remove(os.path.join(backup_dir, old))
except OSError as e:
logger.warning(f"Could not prune old backup {old}: {e}")
return dest
except Exception as e:
if tmp:
try:
os.remove(tmp)
except OSError:
pass
logger.warning(f"Auto backup failed (non-fatal): {e}")
return None
_AUDIT_EXPORT_SUFFIXES = (".csv", ".ndjson.gz", ".manifest.json")
async def _prune_audit_exports(retention_days: int) -> int:
"""Delete per-day audit export files older than the window. Returns count.
Prunes every artefact for a day together (csv + ndjson.gz + manifest) so a
day is never left half-archived. `retention_days <= 0` keeps everything
(the default: the archive is the only full-detail copy after the rollup,
so deleting it silently loses evidence).
"""
import datetime as _dt
if retention_days <= 0:
return 0
outdir = audit_export_dir()
cutoff = _dt.date.today() - _dt.timedelta(days=retention_days)
removed = 0
try:
names = os.listdir(outdir)
except OSError:
return 0
for name in names:
m = re.fullmatch(r"tool-audit-(\d{4})-(\d{2})-(\d{2})(\.csv|\.ndjson\.gz|\.manifest\.json)", name)
if not m:
continue
try:
day = _dt.date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
except ValueError:
continue
if day < cutoff:
try:
os.remove(os.path.join(outdir, name))
removed += 1
except OSError as e:
logger.warning(f"Could not prune audit export {name}: {e}")
return removed
async def cleanup_expired_data() -> tuple[int, int]:
"""Delete data older than the configured retention (0 = keep forever).
Runs on startup and then periodically (see ``periodic_cleanup_loop``).
Never raises — a failure (DB locked, etc.) is logged and retried on the
next cycle so it cannot crash the service.
Returns (deleted_conversation_rows, deleted_memory_rows).
"""
from config import CONVERSATION_RETENTION_DAYS, MEMORY_RETENTION_DAYS
try:
db = await get_db()
removed_conv = removed_mem = 0
if CONVERSATION_RETENTION_DAYS > 0:
async with _write_lock():
removed_conv = await _retention_prune_conversations(db)
# Archive pruning follows its own policy (default 0 = keep forever):
# the exported files are the only full-detail copy after a rollup.
from config import AUDIT_ARCHIVE_RETENTION_DAYS
removed_csv = await _prune_audit_exports(AUDIT_ARCHIVE_RETENTION_DAYS)
if removed_csv:
logger.info(f"Retention cleanup: {removed_csv} audit export file(s) pruned")
if MEMORY_RETENTION_DAYS > 0:
cur = await _write_with_retry(
db,
"DELETE FROM memories WHERE created_at < datetime('now', ?)",
(f"-{MEMORY_RETENTION_DAYS} days",),
)
removed_mem = cur.rowcount if cur.rowcount else 0
if removed_conv or removed_mem:
logger.info(f"Retention cleanup: {removed_conv} conversations, {removed_mem} memories deleted")
# Free-page compaction (opportunistic, see helper) after the deletes.
await _vacuum_if_fragmented(db)
return removed_conv, removed_mem
except Exception as e:
logger.warning(f"Retention cleanup failed, will retry next cycle: {e}")
return 0, 0
# -- Tool audit log --
# Params keys whose values are user content or secrets; their values are
# replaced with [REDACTED] in the audit row so raw email/note/memory text and
# credentials never hit the DB (the log keeps tool_name/success/duration/error
# for accountability).
_AUDIT_REDACT_KEYS = {
"password", "passwd", "pass", "token", "api_key", "apikey", "api-key",
"secret", "auth", "authorization", "credential", "credentials",
"body", "content", "text", "message", "prompt",
# E-mail fields: recipient addresses, subject and sender are PII and
# must not land in the audit log in plaintext.
"to", "cc", "bcc", "from", "sender", "recipient", "recipients",
"subject", "reply_to", "replyto", "from_address", "to_address",
}
_AUDIT_PARAMS_MAX_CHARS = 2048
def _redact_params(value):
"""Recursively redact sensitive keys and return a log-safe structure."""
if isinstance(value, dict):
return {
k: ("[REDACTED]" if _is_redact_key(k) else _redact_params(v))
for k, v in value.items()
}
if isinstance(value, (list, tuple)):
return [_redact_params(v) for v in value]
return value
def _is_redact_key(key: str) -> bool:
norm = str(key).lower().replace(" ", "_").replace("-", "_")
return norm in _AUDIT_REDACT_KEYS
def _audit_params_json(params) -> str | None:
"""Serialise params with sensitive fields redacted and a size cap."""
if not params:
return None
safe = _redact_params(params)
try:
out = json.dumps(safe, ensure_ascii=False)
except (TypeError, ValueError):
out = json.dumps(_redact_params(str(params)), ensure_ascii=False)
if len(out) > _AUDIT_PARAMS_MAX_CHARS:
out = out[:_AUDIT_PARAMS_MAX_CHARS] + " ...(truncated)"
return out
async def log_tool_call(
tool_name: str,
params: dict | None,
success: bool,
duration_ms: float | None = None,
error: str | None = None,
verification_status: str | None = None,
user_id: str | None = None,
) -> int | None:
"""Append a row to the tool audit log.
``verification_status`` records the outcome of ID-based backend
verification (one of "verified", "verified_by_fallback", "unverified",
"verification_failed", or None when verification is not applicable).
Returns the new row's id (the audit_id the frontend uses to attach a
correction to this exact tool call), or None on failure.
Deliberately swallows every failure (DB down, locked, schema issue) and
only logs a warning — this runs inside the tool-call loop's verification
hook and must never break or stall an actual tool execution.
"""
try:
db = await get_db()
cur = await _write_with_retry(
db,
"INSERT INTO tool_audit_log (tool_name, params, success, duration_ms, error, verification_status, user_id) "
"VALUES (?, ?, ?, ?, ?, ?, ?)",
(tool_name, _audit_params_json(params), 1 if success else 0, duration_ms, error, verification_status,
user_id or "default"),
)
return cur.lastrowid
except Exception as e:
logger.warning(f"Tool audit log write failed for '{tool_name}': {e}")
return None
async def log_intent_audit(message: str, chosen_group: str | None,
best_sim: float | None, margin: float | None,
source: str) -> None:
"""Fire-and-forget intent ambiguity record. Never raises."""
try:
db = await get_db()
async with _write_lock():
await db.execute(
"INSERT INTO intent_audit_log (message, chosen_group, best_sim, margin, source) "
"VALUES (?, ?, ?, ?, ?)",
(message[:500], chosen_group, best_sim, margin, source),
)
await _commit_with_retry(db)
except Exception as e:
logger.warning(f"Intent audit write failed (source={source}): {e}")
async def set_tool_correction(audit_id: int, expected_tool: str | None,
expected_group: str | None = None,
user_id: str | None = None) -> bool:
"""Set a correction on a tool audit log entry.
Updates expected_tool and/or expected_group and sets corrected_at to
current timestamp. expected_tool is a precise positive signal (exact
tool name); expected_group is a coarse signal (domain) used when the
user only picks a group. Either may be NULL. Sets confirmed_at to NULL:
confirmation and correction are mutually exclusive feedback states, so a
row always holds at most one of them. Returns True if a row was updated,
False if not found.
Ownership: pass the caller's user_id to scope the write to their own
rows (multi-user invisibility). None disables scoping — offline
scripts/tests only; HTTP endpoints must always pass current_user.
"""
try:
db = await get_db()
sql = ("UPDATE tool_audit_log SET expected_tool = ?, expected_group = ?, "
"corrected_at = CURRENT_TIMESTAMP, confirmed_at = NULL WHERE id = ?")
args: tuple = (expected_tool, expected_group, audit_id)
if user_id is not None:
sql += " AND user_id = ?"
args = (*args, user_id)
async with _write_lock():
cur = await db.execute(sql, args)
await _commit_with_retry(db)
return cur.rowcount > 0
except Exception as e:
logger.warning(f"Tool correction update failed for audit_id={audit_id}: {e}")
return False
async def get_audit_tool_name(audit_id: int, user_id: str | None = None) -> str | None:
"""Return the tool_name of an audit log row, or None if not found.
Pass user_id to scope the read to the caller's own rows (None = unscoped,
offline use only).
"""
try:
db = await get_db()
sql = "SELECT tool_name FROM tool_audit_log WHERE id = ?"
args: tuple = (audit_id,)
if user_id is not None:
sql += " AND user_id = ?"
args = (*args, user_id)
cur = await db.execute(sql, args)
row = await cur.fetchone()
return row[0] if row else None
except Exception as e:
logger.warning(f"fetch audit tool_name failed for audit_id={audit_id}: {e}")
return None
async def set_tool_confirmation(audit_id: int, user_id: str | None = None) -> bool:
"""Record a positive (confirmation) signal on a tool audit log entry.
Sets confirmed_at to the current timestamp. Confirmation is the opposite
of a correction: any previously stored expected_tool/expected_group and
their corrected_at are cleared so a row never holds two opposing signals.
Returns True if a row was updated, False if not found.
Ownership: pass the caller's user_id (None = unscoped, offline use only).
"""
try:
db = await get_db()
sql = ("UPDATE tool_audit_log SET confirmed_at = CURRENT_TIMESTAMP, "
"expected_tool = NULL, expected_group = NULL, corrected_at = NULL "
"WHERE id = ?")
args: tuple = (audit_id,)
if user_id is not None:
sql += " AND user_id = ?"
args = (*args, user_id)
async with _write_lock():
cur = await db.execute(sql, args)
await _commit_with_retry(db)
return cur.rowcount > 0
except Exception as e:
logger.warning(f"Tool confirmation update failed for audit_id={audit_id}: {e}")
return False
# ── Collective-learning quorum (Phase 3) ──────────────────────────────────────
# Votes mirror feedback writes for quorum accounting. Raw audit rows stay
# per-user and invisible; only anonymized (signature, group) tallies below
# ever influence shared learning. Nothing here raises.
FEEDBACK_DAILY_CAP = 50
QUORUM_MIN_USERS = 2
REPUTATION_FLOOR = 0.5
async def _user_text_for_audit(db, audit_id: int, user_id: str) -> str | None:
"""Resolve the user's trigger sentence behind an audit row (owner-scoped)."""
cur = await db.execute(
"SELECT conversation_id FROM tool_audit_log WHERE id = ? AND user_id = ?",
(audit_id, user_id),
)
row = await cur.fetchone()
if not row or not row[0]:
return None
cur = await db.execute(
"SELECT session_id FROM conversations WHERE id = ? AND user_id = ?",
(row[0], user_id),
)
srow = await cur.fetchone()
if not srow:
return None
cur = await db.execute(
"SELECT content FROM conversations WHERE session_id = ? AND id < ? "
"AND role = 'user' AND user_id = ? ORDER BY id DESC LIMIT 1",
(srow[0], row[0], user_id),
)
urow = await cur.fetchone()
return urow[0] if urow and urow[0] else None
async def record_feedback_vote(audit_id: int, user_id: str, signal: str, group: str | None) -> bool:
"""Mirror a scoped feedback write into quorum accounting. Never raises.
Re-votes overwrite (matches feedback overwrite semantics). Skips quietly
when the trigger text is unresolvable — the feedback itself already applied.
"""
try:
from textnorm import normalize_signature
db = await get_db()
text = await _user_text_for_audit(db, audit_id, user_id)
if not text:
return False
sig = normalize_signature(text)
if not sig:
return False
async with _write_lock():
await db.execute(
"""INSERT INTO feedback_votes (audit_id, user_id, signature, proposed_group, signal)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(audit_id, user_id) DO UPDATE SET
signature = excluded.signature, proposed_group = excluded.proposed_group,
signal = excluded.signal, created_at = CURRENT_TIMESTAMP""",
(audit_id, user_id, sig, group or "", signal),
)
await _commit_with_retry(db)
return True
except Exception as e:
logger.warning(f"Feedback vote record failed for audit_id={audit_id}: {e}")
return False
async def get_pattern_support(signature: str, group: str) -> dict:
"""Distinct-user support/contradiction counts for a (signature, group).
Only approved users count (admins always do) — open registration must
not let sockpuppets manufacture quorum. Deleted users drop out via JOIN.
"""
try:
db = await get_db()
cur = await db.execute(
"SELECT v.proposed_group, COUNT(DISTINCT v.user_id) FROM feedback_votes v "
"JOIN users u ON u.id = v.user_id "
"WHERE v.signature = ? AND (u.is_approved = 1 OR u.is_admin = 1) "
"GROUP BY v.proposed_group",
(signature,),
)
supports = contradicts = 0
for grow, n in await cur.fetchall():
if grow == group:
supports += n
else:
contradicts += n
return {"supports": supports, "contradicts": contradicts}
except Exception:
return {"supports": 0, "contradicts": 0}
async def user_reputation(user_id: str) -> float:
"""Agreement rate with decided patterns (weight, never a ban).
A pattern is decided when ≥2 distinct approved users voted and one group holds a
strict majority of distinct users. Users with no decided votes score 1.0
(neutral). Never raises.
"""
try:
db = await get_db()
cur = await db.execute(
"SELECT v.signature, v.proposed_group, v.user_id FROM feedback_votes v "
"JOIN users u ON u.id = v.user_id "
"WHERE v.signature != '' AND (u.is_approved = 1 OR u.is_admin = 1)"
)
rows = await cur.fetchall()
by_sig: dict[str, list[tuple[str, str]]] = {}
for s, g, u in rows:
by_sig.setdefault(s, []).append((g, u))
decided: dict[str, str] = {}
for s, votes in by_sig.items():
per_group: dict[str, set[str]] = {}
for g, u in votes:
per_group.setdefault(g, set()).add(u)
total = len({u for _, u in votes})
if total < 2:
continue
top = max(len(us) for us in per_group.values())
winners = [g for g, us in per_group.items() if len(us) == top]
if len(winners) == 1 and top > total / 2:
decided[s] = winners[0]
mine = [(s, g) for s, g, u in rows if u == user_id and s in decided]
if not mine:
return 1.0
return sum(1 for s, g in mine if decided[s] == g) / len(mine)
except Exception:
return 1.0
# ── Per-user encrypted credentials (mail/nextcloud) ──────────────────────────
async def save_credential(user_id: str, provider: str, payload: dict) -> bool:
"""Validate + encrypt + upsert a credential. Never raises, never logs secrets."""
try:
from credstore import PROVIDER_FIELDS, VALID_PROVIDERS, encrypt_payload
if provider not in VALID_PROVIDERS:
return False
need = PROVIDER_FIELDS[provider]
if any(not str(payload.get(k) or "").strip() for k in need):
return False
clean = {k: str(payload[k]).strip()[:500] for k in need}
blob = encrypt_payload(clean)
db = await get_db()
async with _write_lock():
await db.execute(
"""INSERT INTO user_credentials (user_id, provider, enc_blob)
VALUES (?, ?, ?)
ON CONFLICT(user_id, provider) DO UPDATE SET
enc_blob = excluded.enc_blob, updated_at = CURRENT_TIMESTAMP""",
(user_id, provider, blob),
)
await _commit_with_retry(db)
return True
except Exception:
return False
async def get_credential(user_id: str | None, provider: str) -> dict | None:
"""Decrypt a credential, or None (missing/unreadable). Never raises."""
if not user_id:
return None
try:
from credstore import decrypt_payload
db = await get_db()
cur = await db.execute(
"SELECT enc_blob FROM user_credentials WHERE user_id = ? AND provider = ?",
(user_id, provider),
)
row = await cur.fetchone()
if not row:
return None
return decrypt_payload(row[0])
except Exception:
return None
async def list_credentials(user_id: str) -> list[dict]:
"""Providers saved by a user with their public account label (no secrets)."""
try:
from credstore import PROVIDER_FIELDS, decrypt_payload
db = await get_db()
cur = await db.execute(
"SELECT provider, enc_blob, updated_at FROM user_credentials WHERE user_id = ? ORDER BY provider",
(user_id,),
)
out = []
for provider, blob, updated in await cur.fetchall():
account = ""
data = decrypt_payload(blob)
if data:
fields = PROVIDER_FIELDS.get(provider, ())
account = str(data.get(fields[0], "")) if fields else ""
out.append({"provider": provider, "account": account, "updated_at": updated})
return out
except Exception:
return []
async def delete_credential(user_id: str, provider: str) -> bool:
"""Remove a credential. Never raises."""
try:
db = await get_db()
async with _write_lock():
cur = await db.execute(
"DELETE FROM user_credentials WHERE user_id = ? AND provider = ?",
(user_id, provider),
)
await _commit_with_retry(db)
return cur.rowcount > 0
except Exception:
return False
async def count_user_votes_today(user_id: str) -> int:
"""Feedback votes cast by a user today (UTC) — flood-control counter."""
try:
db = await get_db()
cur = await db.execute(
"SELECT COUNT(*) FROM feedback_votes WHERE user_id = ? AND date(created_at) = date('now')",
(user_id,),
)
row = await cur.fetchone()
return int(row[0]) if row else 0
except Exception:
return 0
async def record_pattern_decision(signature: str, group: str, status: str, decided_by: str) -> None:
"""Store an admin decision on a routing pattern (approved/rejected).
Approved patterns skip the contradiction freeze; rejected ones are never
auto-added. Never raises.
"""
if status not in ("approved", "rejected") or not signature:
return
try:
db = await get_db()
async with _write_lock():
await db.execute(
"""INSERT INTO pattern_approvals (signature, proposed_group, status, decided_by)
VALUES (?, ?, ?, ?)
ON CONFLICT(signature) DO UPDATE SET
proposed_group = excluded.proposed_group, status = excluded.status,
decided_by = excluded.decided_by, decided_at = CURRENT_TIMESTAMP""",
(signature, group or "", status, decided_by),
)
await _commit_with_retry(db)
except Exception as e:
logger.warning(f"Pattern decision record failed: {e}")
async def get_pattern_decision(signature: str) -> dict | None:
"""Return the admin decision for a signature, or None."""
try:
db = await get_db()
cur = await db.execute(
"SELECT proposed_group, status, decided_by, decided_at FROM pattern_approvals WHERE signature = ?",
(signature,),
)
row = await cur.fetchone()
if not row:
return None
return {"group": row[0], "status": row[1], "decided_by": row[2], "decided_at": row[3]}
except Exception:
return None
async def get_review_queue() -> list[dict]:
"""Frozen candidates for the admin queue: reputable contradiction, no quorum, undecided.
Each item carries supports/contradicts counts plus contradictor ids with
reputations so the admin decides informed. Never raises.
"""
try:
db = await get_db()
cur = await db.execute(
"SELECT DISTINCT v.signature FROM feedback_votes v "
"JOIN users u ON u.id = v.user_id "
"WHERE v.signature != '' AND (u.is_approved = 1 OR u.is_admin = 1)"
)
sigs = [r[0] for r in await cur.fetchall()]
queue = []
for s in sigs:
if await get_pattern_decision(s) is not None:
continue
cur = await db.execute(
"SELECT v.proposed_group, v.user_id FROM feedback_votes v "
"JOIN users u ON u.id = v.user_id "
"WHERE v.signature = ? AND (u.is_approved = 1 OR u.is_admin = 1)",
(s,),
)
per_group: dict[str, set[str]] = {}
for g, u in await cur.fetchall():
per_group.setdefault(g, set()).add(u)
if len(per_group) < 2:
continue
total_supports = 0
groups = []
for g, us in per_group.items():
sub = await get_pattern_support(s, g)
total_supports = max(total_supports, sub["supports"])
groups.append({"group": g, "supports": sub["supports"], "voters": sorted(us)})
if total_supports >= QUORUM_MIN_USERS:
continue
for g in groups:
g["reputations"] = {u: await user_reputation(u) for u in g["voters"]}
queue.append({"signature": s, "groups": groups})
return queue
except Exception as e:
logger.warning(f"Review queue build failed: {e}")
return []
async def upsert_message_feedback(message_id: int, value: str,
note: str | None = None,
user_id: str | None = None) -> bool:
"""Store a user's 👍/👎 verdict for an assistant message that had no tool
call (or whose round failed before any audit existed). One row per message:
pressing the other thumb overwrites the stored verdict; a note can be
attached to a 👎 to capture *why* it was wrong (model dropped the intent,
asked instead of acting, hallucinated, …). Never raises.
Ownership: pass the caller's user_id to accept only their own messages
(None = role check only, offline use only).
"""
if value not in ("up", "down"):
return False
note = (note or "").strip() or None
try:
db = await get_db()
sql = "SELECT role FROM conversations WHERE id = ?"
args: tuple = (message_id,)
if user_id is not None:
sql += " AND user_id = ?"
args = (*args, user_id)
async with db.execute(sql, args) as cur:
row = await cur.fetchone()
if not row or row[0] != "assistant":
return False
async with _write_lock():
await db.execute(
"""INSERT INTO message_feedback (message_id, value, note, created_at, updated_at)
VALUES (?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
ON CONFLICT(message_id) DO UPDATE SET
value = excluded.value, note = excluded.note, updated_at = CURRENT_TIMESTAMP""",
(message_id, value, note),
)
await _commit_with_retry(db)
return True
except Exception as e:
logger.warning(f"Message feedback upsert failed for message_id={message_id}: {e}")
return False
async def link_audits_to_message(conversation_id: int, audit_ids: list[int]):
"""Bind audit rows to the assistant message that produced them.
Called after the assistant message is written, using the audit_ids the
stream collected from its SSE tool events. `IS NULL` guard keeps a row
owned by an earlier message (e.g. a previous regeneration) untouched.
Never raises; returns the number of rows linked.
"""
if not audit_ids or conversation_id is None:
return 0
try:
db = await get_db()
ph = ",".join("?" * len(audit_ids))
async with _write_lock():
cur = await db.execute(
f"UPDATE tool_audit_log SET conversation_id = ? WHERE id IN ({ph}) AND conversation_id IS NULL",
(conversation_id, *audit_ids),
)
await _commit_with_retry(db)
return cur.rowcount
except Exception as e:
logger.warning(f"Audit->message link failed for conversation_id={conversation_id}: {e}")
return 0
async def purge_intent_audit(days: int = 30) -> int:
"""Delete intent audit rows older than `days`. Never raises; returns deleted count."""
try:
db = await get_db()
async with _write_lock():
cur = await db.execute(
"DELETE FROM intent_audit_log WHERE created_at < datetime('now', ?)",
(f"-{days} days",),
)
await _commit_with_retry(db)
return cur.rowcount
except Exception as e:
logger.warning(f"Intent audit purge failed: {e}")
return 0
# -- Audit export before retention rollup --
# The full detail-row shape: every audit column plus the conversation the row
# was linked to (thumbs/fine-tuning context survives the 14-day purge).
AUDIT_EXPORT_COLUMNS = [
"id", "tool_name", "params", "success", "duration_ms", "error", "is_summary",
"day", "total_calls", "success_count", "error_count", "tool_breakdown",
"created_at", "expected_tool", "corrected_at", "verification_status",
"expected_group", "confirmed_at", "conversation_id", "session_id",
]
def audit_export_dir() -> str:
"""Directory for pre-rollup audit exports.
`AUDIT_EXPORT_DIR` env wins; otherwise ``<dirname of DB_PATH>/audit_exports``
— in production that is ``audit_exports/`` next to ``assistant.db``, and in
tests it lands inside the tmp fixture DB's own directory (no repo litter).
"""
env = os.getenv("AUDIT_EXPORT_DIR")
if env:
return env
return os.path.join(os.path.dirname(DB_PATH), "audit_exports")
# Spreadsheet formula-injection guard (OWASP): a cell Excel/Sheets would
# EXECUTE starts with one of these. '-' is deliberately NOT guarded — negative
# numbers and dash-led messages must stay machine-readable, and our params
# are JSON objects (they start with '{'), so the practical attack surface is
# the free-text error column.
_CSV_INJECTION_CHARS = ("=", "+", "@", "\t", "\r")
def _csv_safe_cell(value) -> str:
"""Stringify one CSV cell, neutralising spreadsheet formula injection."""
s = "" if value is None else str(value)
if s and s[0] in _CSV_INJECTION_CHARS:
return "'" + s
return s
def _write_audit_csv(day: str, rows: list[tuple]) -> int:
"""Write `rows` to ``tool-audit-<day>.csv`` atomically (tmp + os.replace).
Runs in a worker thread; overwrites the day's file (idempotent — a crash
after the write but before the rollup commit just re-writes the same file).
Raises on failure so the caller's rollback keeps the rows for the next try.
Rows arrive ORDER BY id (see export_tool_audit_day): same rows in, same
bytes out — a re-export is sha256-identical.
"""
outdir = audit_export_dir()
os.makedirs(outdir, exist_ok=True)
final = os.path.join(outdir, f"tool-audit-{day}.csv")
tmp = final + ".tmp"
with open(tmp, "w", newline="", encoding="utf-8") as f:
w = csv.writer(f)
w.writerow(AUDIT_EXPORT_COLUMNS)
for r in rows:
w.writerow([_csv_safe_cell(v) for v in r])
os.replace(tmp, final)
return len(rows)
def _write_audit_ndjson(day: str, rows: list[tuple]) -> int:
"""Write `rows` to ``tool-audit-<day>.ndjson.gz`` (atomic, deterministic).
One JSON object per line; key order fixed by AUDIT_EXPORT_COLUMNS (dicts
preserve insertion order — no sort_keys). gzip with mtime=0 and an empty
stored filename: the default gzip header embeds the clock, which would
make byte-identical rows produce different files. Deterministic bytes are
the point — the manifest's sha256 is only comparable across runs if two
exports of the same rows hash the same.
"""
outdir = audit_export_dir()
os.makedirs(outdir, exist_ok=True)
final = os.path.join(outdir, f"tool-audit-{day}.ndjson.gz")
tmp = final + ".tmp"
with open(tmp, "wb") as raw:
with gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as gz:
for r in rows:
obj = {c: v for c, v in zip(AUDIT_EXPORT_COLUMNS, r)}
gz.write(json.dumps(obj, ensure_ascii=False).encode("utf-8") + b"\n")
os.replace(tmp, final)
return len(rows)
def _file_sha256(path: str) -> str:
h = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(65536), b""):
h.update(chunk)
return h.hexdigest()
def _write_audit_manifest(day: str, files: dict) -> None:
"""Write ``tool-audit-<day>.manifest.json`` atomically (integrity record).
`files` maps filename -> {"sha256", "bytes"} of the archived artefacts.
Written LAST of the three: a day whose manifest exists is a day whose
archive completed — a crash before it leaves no manifest, which is the
detectable gap (industry pattern: manifest binds the export).
`generated_at` is envelope metadata, deliberately outside the determinism
guarantee that applies to the data files.
"""
import datetime as _dt
outdir = audit_export_dir()
os.makedirs(outdir, exist_ok=True)
final = os.path.join(outdir, f"tool-audit-{day}.manifest.json")
tmp = final + ".tmp"
manifest = {
"format": "pisynapse-audit-export",
"version": 1,
"day": day,
"files": files,
"generated_at": _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="seconds"),
}
with open(tmp, "w", encoding="utf-8") as f:
json.dump(manifest, f, ensure_ascii=False, indent=2, sort_keys=True)
f.write("\n")
os.replace(tmp, final)
def _write_audit_exports(day: str, rows: list[tuple]) -> int:
"""Archive one day: CSV + NDJSON.gz + manifest (worker thread). Returns n.
Any raise propagates so the rollup transaction rolls back — details are
never deleted without a complete archive on disk.
"""
_write_audit_csv(day, rows)
_write_audit_ndjson(day, rows)
outdir = audit_export_dir()
files = {}
for name in (f"tool-audit-{day}.csv", f"tool-audit-{day}.ndjson.gz"):
p = os.path.join(outdir, name)
files[name] = {"sha256": _file_sha256(p), "bytes": os.path.getsize(p)}
_write_audit_manifest(day, files)
return len(rows)
async def export_tool_audit_day(day: str, cutoff_sql: str) -> int:
"""Fetch and archive every detail row of `day`; returns row count.
Called INSIDE the rollup transaction: on success the caller proceeds to
summarize + delete; on exception it rolls back, so nothing is discarded
until its archive exists. ORDER BY t.id fixes the row order — exports are
deterministic (byte-identical for identical rows, sha256-checkable).
"""
cols = ", ".join(
"c.session_id" if c == "session_id" else "t." + c
for c in AUDIT_EXPORT_COLUMNS
)
db = await get_db()
cur = await db.execute(
f"""SELECT {cols}
FROM tool_audit_log t
LEFT JOIN conversations c ON c.id = t.conversation_id
WHERE t.is_summary = 0 AND t.created_at < datetime('now', ?)
AND substr(t.created_at, 1, 10) = ?
ORDER BY t.id""",
(cutoff_sql, day),
)
rows = await cur.fetchall()
if not rows:
return 0
return await asyncio.to_thread(_write_audit_exports, day, rows)
async def verify_audit_export(day: str) -> dict:
"""Re-hash one day's archived files against its manifest. Never raises.
Returns ``{"day", "exists", "ok", "files": {name: {sha256_ok, bytes_ok}}}``.
``exists=False`` means no manifest — either the day was never archived or
it predates the manifest format (legacy CSV-only days). ``ok`` is True
only when every file listed in the manifest is present and matches.
"""
out: dict = {"day": day, "exists": False, "ok": False, "files": {}}
try:
outdir = audit_export_dir()
manifest_path = os.path.join(outdir, f"tool-audit-{day}.manifest.json")
if not os.path.isfile(manifest_path):
return out
with open(manifest_path, "r", encoding="utf-8") as f:
manifest = json.load(f)
out["exists"] = True
all_ok = True
for name, meta in (manifest.get("files") or {}).items():
p = os.path.join(outdir, name)
entry = {"sha256_ok": False, "bytes_ok": False}
if os.path.isfile(p):
entry["bytes_ok"] = os.path.getsize(p) == meta.get("bytes")
entry["sha256_ok"] = _file_sha256(p) == meta.get("sha256")
all_ok = all_ok and entry["sha256_ok"] and entry["bytes_ok"]
out["files"][name] = entry
out["ok"] = all_ok and bool(out["files"])
except Exception as e:
logger.warning(f"verify_audit_export({day}) failed: {e}")
return out
def _build_day_stats(rows: list[tuple]) -> dict:
"""Aggregate detail rows into the per-day stats dict stored in `stats_json`.
`rows` are (tool_name, user_id, success, duration_ms, verification_status)
tuples — the same projection used by the rollup INSERT and by the
dashboard's live-day aggregation, so both paths always agree. NULL
durations are skipped; NULL user_ids count as "unknown"; NULL
verification_status means "not applicable" and is omitted (only
ID-verified tools record a status). p95 is nearest-rank
(ceil(0.95*n)-1). Deterministic and pure — no I/O, no time.
"""
tools: dict[str, int] = {}
users: dict[str, dict[str, int]] = {}
durations: list[float] = []
verify: dict[str, int] = {}
success_total = 0
for tool_name, user_id, success, duration_ms, vstatus in rows:
tools[tool_name] = tools.get(tool_name, 0) + 1
u = user_id or "unknown"
bucket = users.setdefault(u, {"total": 0, "success": 0, "error": 0})
bucket["total"] += 1
if success:
success_total += 1
bucket["success"] += 1
else:
bucket["error"] += 1
if duration_ms is not None:
durations.append(float(duration_ms))
if vstatus:
verify[vstatus] = verify.get(vstatus, 0) + 1
dur: dict = {"n": len(durations)}
if durations:
ordered = sorted(durations)
dur["min"] = round(ordered[0], 1)
dur["max"] = round(ordered[-1], 1)
dur["avg"] = round(sum(ordered) / len(ordered), 1)
dur["p95"] = round(ordered[max(0, math.ceil(0.95 * len(ordered)) - 1)], 1)
return {
"tools": tools,
"users": users,
"dur_ms": dur,
"verify": verify,
"success": success_total,
}
async def rollup_tool_audit(days: int | None = None) -> int:
"""Compress detail rows older than `days` into one daily summary row per day.
Retention policy: detail rows survive for `days` (default: config
``AUDIT_RETENTION_DAYS``, 14), then are exported to ``audit_exports/`` as
per-day CSV files (``tool-audit-YYYY-MM-DD.csv``, one row per audited call,
incl. the linked conversation), aggregated into a per-day summary
(total/success/error counts + per-tool breakdown + `stats_json`:
per-user counts, duration min/avg/max/p95, verification breakdown) and the
detail rows are removed. Each day is processed inside its own ``BEGIN
IMMEDIATE`` transaction so the export completion, the summary INSERT and the
detail DELETE are all-or-nothing: a failed export rolls the day back and
nothing is discarded until a copy exists on disk. Days that already have a
summary row are skipped, so a crash between runs cannot produce duplicate
summaries (or duplicate exports — the day's CSV is overwritten in place).
Idempotent and never raises.
Returns the number of days rolled up.
"""
import sqlite3
from config import AUDIT_RETENTION_DAYS
if days is None:
days = max(1, AUDIT_RETENTION_DAYS)
try:
db = await get_db()
cutoff = f"-{days} days"
cur = await db.execute(
"""SELECT substr(created_at, 1, 10) AS day
FROM tool_audit_log
WHERE is_summary = 0 AND created_at < datetime('now', ?)
AND substr(created_at, 1, 10) NOT IN
(SELECT day FROM tool_audit_log WHERE is_summary = 1)
GROUP BY substr(created_at, 1, 10)""",
(cutoff,),
)
days_found = [row[0] for row in await cur.fetchall()]
days_summarized = 0
for day in days_found:
# Each day is atomic: summary INSERT + detail DELETE in one txn.
# The process-wide write lock is held for the whole transaction:
# every other write path takes it, and without it another
# writer's statements would JOIN this open transaction on the
# shared connection — and roll back WITH it when a busy retry
# discards the day (or BEGIN IMMEDIATE would collide with the
# other path's implicit transaction and the generic rollback
# would kill ITS partial write).
async with _write_lock():
for attempt in range(_LOCKED_RETRIES + 1):
try:
await db.execute("BEGIN IMMEDIATE")
cur2 = await db.execute(
"""SELECT tool_name, user_id, success, duration_ms, verification_status
FROM tool_audit_log
WHERE is_summary = 0 AND created_at < datetime('now', ?)
AND substr(created_at, 1, 10) = ?""",
(cutoff, day),
)
day_rows = await cur2.fetchall()
stats = _build_day_stats(day_rows)
total = len(day_rows)
success_count = stats["success"]
# Archive first, delete second: a failed export raises and
# rolls the day back, so detail rows are never discarded
# without a CSV copy existing on disk.
await export_tool_audit_day(day, cutoff)
await db.execute(
"""INSERT INTO tool_audit_log
(is_summary, day, total_calls, success_count, error_count,
tool_breakdown, tool_name, success, created_at,
expected_tool, corrected_at, stats_json)
VALUES (1, ?, ?, ?, ?, ?, 'rollup', 1, ?, NULL, NULL, ?)""",
(day, total, success_count, total - success_count,
json.dumps(stats["tools"]), day,
json.dumps(stats, separators=(",", ":"))),
)
await db.execute(
"""DELETE FROM tool_audit_log
WHERE is_summary = 0 AND created_at < datetime('now', ?)
AND substr(created_at, 1, 10) = ?""",
(cutoff, day),
)
await _commit_with_retry(db)
days_summarized += 1
break
except sqlite3.OperationalError as e:
try:
await db.rollback()
except Exception:
pass
if "locked" not in str(e).lower() and "busy" not in str(e).lower():
raise
if attempt >= _LOCKED_RETRIES:
raise
logger.warning(f"SQLite busy during rollup of {day}, retrying ({attempt + 1}/{_LOCKED_RETRIES})...")
await asyncio.sleep(0.25 * (attempt + 1))
except Exception:
try:
await db.rollback()
except Exception:
pass
raise
if days_summarized:
logger.info(f"Tool audit rollup: {days_summarized} day(s) summarized")
return days_summarized
except Exception as e:
logger.warning(f"Tool audit rollup failed: {e}")
return 0
ROLLUP_INTERVAL_SECONDS = 24 * 3600 # retention sweep runs once a day
_AUDIT_ALERT_MIN_CALLS = 10 # below this a "100% error rate" is noise, not signal
async def get_audit_overview(days: int | None = None) -> dict:
"""Dashboard payload: per-day tool-audit stats for the last `days` days.
Merges rolled-up summary rows (is_summary=1, older than retention) with a
live aggregation of still-detailed rows, so the trend is continuous across
the retention boundary — a gap there would make the dashboard lie exactly
when the rollup runs. Newest day first. Each entry carries
``source: "summary" | "detail"`` so the UI can label compressed days.
Never raises; returns ``{"days": [], ...}`` on failure.
"""
from config import AUDIT_RETENTION_DAYS
if days is None:
days = max(1, AUDIT_RETENTION_DAYS)
out: dict = {"days": [], "retention_days": days, "window_days": days}
try:
db = await get_db()
cur = await db.execute(
"""SELECT day, total_calls, success_count, error_count,
tool_breakdown, stats_json
FROM tool_audit_log WHERE is_summary = 1
ORDER BY day DESC LIMIT ?""",
(days,),
)
merged: dict[str, dict] = {}
for day, total, succ, err, breakdown, stats_json in await cur.fetchall():
try:
stats = json.loads(stats_json) if stats_json else {}
except (ValueError, TypeError):
stats = {}
try:
tools = json.loads(breakdown) if breakdown else stats.get("tools", {})
except (ValueError, TypeError):
tools = stats.get("tools", {})
total = total or 0
merged[day] = {
"day": day, "total": total, "success": succ or 0,
"errors": err or 0,
"error_rate": round((err or 0) / total, 3) if total else 0.0,
"tools": tools, "users": stats.get("users", {}),
"dur_ms": stats.get("dur_ms", {}), "verify": stats.get("verify", {}),
"source": "summary",
}
# Live days (within the window, not yet rolled up): aggregate details
# with the SAME _build_day_stats the rollup uses, per day.
cur = await db.execute(
"""SELECT substr(created_at, 1, 10) AS day, tool_name, user_id,
success, duration_ms, verification_status
FROM tool_audit_log
WHERE is_summary = 0 AND created_at >= datetime('now', ?)
ORDER BY created_at""",
(f"-{days} days",),
)
by_day: dict[str, list] = {}
for day, tool_name, user_id, success, dur, vstatus in await cur.fetchall():
by_day.setdefault(day, []).append((tool_name, user_id, success, dur, vstatus))
for day, rows in by_day.items():
if day in merged:
continue # already summarized (crash-residue); summary wins
stats = _build_day_stats(rows)
total = len(rows)
succ = stats["success"]
merged[day] = {
"day": day, "total": total, "success": succ,
"errors": total - succ,
"error_rate": round((total - succ) / total, 3) if total else 0.0,
"tools": stats["tools"], "users": stats["users"],
"dur_ms": stats["dur_ms"], "verify": stats["verify"],
"source": "detail",
}
out["days"] = sorted(merged.values(), key=lambda d: d["day"], reverse=True)
except Exception as e:
logger.warning(f"get_audit_overview failed: {e}")
return out
async def audit_error_alert(window_hours: int = 24) -> dict:
"""Rolling error-rate check across detail rows AND summary rows.
Summaries contribute their success_count/error_count: right after a sweep
a details-only window would read empty and the alarm would miss a
catastrophic day that just got rolled up. Whole summaries inside the
window count fully (a deliberate coarseness — an alarm, not an accounting).
Threshold: config ``AUDIT_ERROR_RATE_ALERT`` (0 = off); ignored below
``_AUDIT_ALERT_MIN_CALLS`` total calls. Never raises.
"""
from config import AUDIT_ERROR_RATE_ALERT
out = {
"window_hours": window_hours, "total": 0, "success": 0, "errors": 0,
"error_rate": 0.0, "threshold": AUDIT_ERROR_RATE_ALERT,
"min_calls": _AUDIT_ALERT_MIN_CALLS, "breached": False,
}
try:
db = await get_db()
cur = await db.execute(
"""SELECT COUNT(*), COALESCE(SUM(success), 0)
FROM tool_audit_log
WHERE is_summary = 0 AND created_at >= datetime('now', ?)""",
(f"-{window_hours} hours",),
)
det_total, det_success = await cur.fetchone()
cur = await db.execute(
"""SELECT COALESCE(SUM(total_calls), 0), COALESCE(SUM(success_count), 0)
FROM tool_audit_log
WHERE is_summary = 1 AND day >= date('now', ?)""",
(f"-{window_hours} hours",),
)
sum_total, sum_success = await cur.fetchone()
total = (det_total or 0) + (sum_total or 0)
success = (det_success or 0) + (sum_success or 0)
errors = max(0, total - success)
out.update({
"total": total, "success": success, "errors": errors,
"error_rate": round(errors / total, 3) if total else 0.0,
})
out["breached"] = bool(
AUDIT_ERROR_RATE_ALERT > 0
and total >= _AUDIT_ALERT_MIN_CALLS
and out["error_rate"] > AUDIT_ERROR_RATE_ALERT
)
except Exception as e:
logger.warning(f"audit_error_alert failed: {e}")
return out
def _error_cluster_key(error: str) -> str:
"""Normalize one error message into a cluster key (digits/ids stripped).
"timeout after 12s" and "timeout after 40s" must count as ONE recurring
failure, not two — numbers (durations, ids, ports) are the noise between
copies of the same message. Capped at 80 chars: a cluster key is a label,
not the evidence (the CSV archive keeps the full text).
"""
return re.sub(r"\d+", "#", (error or "").strip().lower())[:80]
async def audit_failure_digest(days: int = 7) -> dict:
"""Clustered summary of the last `days` days of tool failures + signals.
Reads DETAIL rows only (summaries keep counts but not error text — fine,
since the default 7d window sits inside the 14d detail retention). Surfaces
what the weekly flywheel digest needs: which tools fail, which normalized
error recurs, and how many corrections/confirmations/verification
failures the users produced. Never raises.
"""
out: dict = {
"window_days": days, "total_calls": 0, "errors": 0,
"corrections": 0, "confirmations": 0, "verify_failed": 0,
"top_tools": [], "top_errors": [],
}
try:
db = await get_db()
cur = await db.execute(
"""SELECT tool_name, success, error,
expected_group IS NOT NULL,
confirmed_at IS NOT NULL,
verification_status = 'verification_failed'
FROM tool_audit_log
WHERE is_summary = 0 AND created_at >= datetime('now', ?)""",
(f"-{days} days",),
)
tool_stats: dict[str, dict] = {}
err_clusters: dict[str, int] = {}
total = errors = corrections = confirmations = vfail = 0
for tool_name, success, error, corrected, confirmed, v_failed in await cur.fetchall():
total += 1
ts = tool_stats.setdefault(tool_name, {"errors": 0, "total": 0})
ts["total"] += 1
if not success:
errors += 1
ts["errors"] += 1
key = _error_cluster_key(error or "unknown")
err_clusters[key] = err_clusters.get(key, 0) + 1
corrections += 1 if corrected else 0
confirmations += 1 if confirmed else 0
vfail += 1 if v_failed else 0
out.update({
"total_calls": total, "errors": errors,
"corrections": corrections, "confirmations": confirmations,
"verify_failed": vfail,
})
out["top_tools"] = sorted(
({"tool": t, **s, "error_rate": round(s["errors"] / s["total"], 3)}
for t, s in tool_stats.items() if s["errors"]),
key=lambda d: (-d["errors"], d["tool"]),
)[:10]
out["top_errors"] = sorted(
({"error": k, "count": n} for k, n in err_clusters.items()),
key=lambda d: (-d["count"], d["error"]),
)[:10]
except Exception as e:
logger.warning(f"audit_failure_digest failed: {e}")
return out
async def periodic_rollup_loop(interval: float = ROLLUP_INTERVAL_SECONDS, sleep=asyncio.sleep):
"""Run rollup_tool_audit() once per `interval` until cancelled.
Mirrors the "never raise" principle of the verification hook: any failure
(DB lock, disk issue, ...) is logged and retried on the next cycle instead
of crashing the service. Cancellation propagates so the app can shut down
cleanly.
`sleep` is injectable for tests (avoids waiting a real 24 hours).
"""
while True:
await sleep(interval)
try:
await rollup_tool_audit()
await purge_intent_audit()
except asyncio.CancelledError:
raise
except Exception as e:
logger.warning(f"Periodic rollup failed, will retry next cycle: {e}")
# Health check rides the same cadence; never raises. A breach is a
# loud log line — the dashboard renders the same numbers live.
try:
alert = await audit_error_alert()
if alert["breached"]:
logger.warning(
f"Tool error-rate alert: {alert['errors']}/{alert['total']} "
f"calls failed in the last {alert['window_hours']}h "
f"(rate {alert['error_rate']:.0%} > threshold {alert['threshold']:.0%})"
)
except asyncio.CancelledError:
raise
except Exception as e:
logger.warning(f"Periodic audit alert check failed: {e}")
CLEANUP_INTERVAL_SECONDS = 24 * 3600 # retention cleanup runs once a day
async def periodic_cleanup_loop(interval: float = CLEANUP_INTERVAL_SECONDS, sleep=asyncio.sleep):
"""Run cleanup_expired_data() once per `interval` until cancelled.
Mirrors the "never raise" principle of the retention sweep — cleanup
failures are logged and retried on the next cycle instead of crashing the
service. Cancellation propagates so the app can shut down cleanly.
`sleep` is injectable for tests (avoids waiting a real 24 hours).
"""
while True:
await sleep(interval)
try:
await cleanup_expired_data()
except asyncio.CancelledError:
raise
except Exception as e:
logger.warning(f"Periodic retention cleanup failed, will retry next cycle: {e}")
# Daily online backup rides the same cadence (best-effort).
try:
await auto_backup_db()
except asyncio.CancelledError:
raise
except Exception as e:
logger.warning(f"Periodic auto backup failed, will retry next cycle: {e}")
# Keep the WAL file bounded: checkpoint whatever readers no longer need.
# Best-effort; a busy checkpoint simply retries next cycle.
try:
db = await get_db()
await db.execute("PRAGMA wal_checkpoint(PASSIVE)")
except asyncio.CancelledError:
raise
except Exception as e:
logger.debug(f"WAL checkpoint skipped: {e}")
# -- Conversations --
async def save_message(session_id: str, role: str, content: str, images: list[str] | None = None, reasoning: str | None = None, user_id: str = "default"):
import json
images_json = json.dumps(images) if images else None
db = await get_db()
# Retry dedup: same user message as the last row → skip to avoid
# duplicate context when the frontend re-sends after regenerate.
if role == "user":
async with db.execute(
"SELECT role, content FROM conversations "
"WHERE session_id = ? AND user_id = ? ORDER BY id DESC LIMIT 1", (session_id, user_id)
) as cur:
last = await cur.fetchone()
if last and last[0] == "user" and last[1] == content:
return # duplicate — do not insert
# Generate embedding for semantic search (best-effort, never blocks chat on failure)
embedding_blob = None
if content and content.strip():
try:
from embedding import embed_async
embedding_blob = await embed_async(content)
except Exception as e:
logger.warning(f"Embedding failed for save_message (non-fatal): {e}")
async with _write_lock():
# Ownership guard: a session row owned by another user must not
# accept writes. Every read path filters by user_id, so the row
# would be an unreadable orphan that still pollutes FTS, storage
# and any future session-scoped maintenance. Fail-closed with the
# same no-op contract as the duplicate guard above: return None.
async with db.execute(
"SELECT 1 FROM sessions WHERE id = ? AND user_id <> ?",
(session_id, user_id),
) as cur:
if await cur.fetchone() is not None:
return None
cur = await db.execute(
"INSERT INTO conversations (session_id, role, content, images, reasoning, embedding, user_id) VALUES (?, ?, ?, ?, ?, ?, ?)",
(session_id, role, content, images_json, reasoning, embedding_blob, user_id),
)
# Keep FTS5 index in sync (cursor.lastrowid: SELECT last_insert_rowid()
# on the shared connection can return another task's row under concurrency).
rowid = cur.lastrowid
try:
await db.execute(
"INSERT INTO conversations_fts (rowid, content, session_id) VALUES (?, ?, ?)",
(rowid, content, session_id),
)
except Exception as fts_e:
# FTS5 is best-effort; a failure must not eat the message row.
logger.warning(f"save_message FTS insert failed (non-fatal): {fts_e}")
await db.execute(
"""INSERT INTO sessions (id, user_id) VALUES (?, ?)
ON CONFLICT(id) DO UPDATE SET last_active = CURRENT_TIMESTAMP
WHERE sessions.user_id = excluded.user_id""",
(session_id, user_id),
)
if role == "user":
existing = await db.execute("SELECT name FROM sessions WHERE id = ? AND user_id = ?", (session_id, user_id))
row = await existing.fetchone()
if not row or not row[0]:
from title import generate_instant_title
name = generate_instant_title(content)
await db.execute(
"UPDATE sessions SET name = ? WHERE id = ? AND user_id = ?",
(name, session_id, user_id),
)
await _commit_with_retry(db)
return rowid
async def delete_last_assistant(session_id: str, user_id: str = "default") -> bool:
"""Delete the last assistant message in a session.
Returns True if a row was removed, False if no assistant message existed.
Used by the regenerate flow: the old response is purged before the user
message is re-sent so the model never sees the stale reply in context.
"""
db = await get_db()
async with db.execute(
"SELECT id FROM conversations "
"WHERE session_id = ? AND user_id = ? AND role = 'assistant' ORDER BY id DESC LIMIT 1",
(session_id, user_id),
) as cur:
row = await cur.fetchone()
if not row:
return False
async with _write_lock():
await db.execute("DELETE FROM conversations WHERE id = ?", (row[0],))
await db.execute("DELETE FROM conversations_fts WHERE rowid = ?", (row[0],))
await db.execute("DELETE FROM message_feedback WHERE message_id = ?", (row[0],))
await db.execute("DELETE FROM tool_audit_log WHERE conversation_id = ?", (row[0],))
await _repair_summary_boundaries(db)
await _commit_with_retry(db)
return True
async def delete_branch(session_id: str, anchor_id: int, user_id: str = "default") -> list[int]:
"""Truncate a conversation from an anchor message onward (regenerate branch).
Deletes the anchored message and everything saved after it (messages are
appended strictly in id order), so a regenerated reply never sits on a
stale tail — both in the model's context window and in the UI's history.
Returns the list of deleted message ids (FTS rows are removed in lockstep).
Unlike delete_last_assistant this is position-agnostic: the frontend picks
ANY assistant message to regenerate and the whole downstream tail goes away,
exactly like ChatGPT/Claude's branch semantics for old responses.
"""
db = await get_db()
async with db.execute(
"SELECT id FROM conversations WHERE session_id = ? AND user_id = ? AND id >= ? ORDER BY id",
(session_id, user_id, anchor_id),
) as cur:
ids = [r[0] for r in await cur.fetchall()]
async with _write_lock():
if ids:
ph = ",".join("?" * len(ids))
await db.execute(
f"DELETE FROM conversations WHERE session_id = ? AND id IN ({ph})",
(session_id, *ids),
)
await db.execute(f"DELETE FROM conversations_fts WHERE rowid IN ({ph})", ids)
await db.execute(f"DELETE FROM message_feedback WHERE message_id IN ({ph})", ids)
await db.execute(
f"DELETE FROM tool_audit_log WHERE conversation_id IN ({ph})", ids
)
# Keep the session alive + surfaced even though no rows were inserted.
await db.execute(
"INSERT INTO sessions (id, user_id) VALUES (?, ?) "
"ON CONFLICT(id) DO UPDATE SET last_active = CURRENT_TIMESTAMP "
"WHERE sessions.user_id = excluded.user_id",
(session_id, user_id),
)
await _repair_summary_boundaries(db)
await _commit_with_retry(db)
return ids
async def get_history(session_id: str, limit: int = 20, include_reasoning: bool = False,
include_audits: bool = False, user_id: str = "default") -> list[dict]:
import json
db = await get_db()
async with db.execute(
"""SELECT id, role, content, images, timestamp, reasoning FROM conversations
WHERE session_id = ? AND user_id = ? ORDER BY id DESC LIMIT ?""",
(session_id, user_id, limit),
) as cur:
rows = await cur.fetchall()
# Per-assistant-message audit list for the C-7 feedback UI. Only used by
# /chat/history; the LLM context path keeps include_audits=False so audit
# data never pollutes model input.
audits_by_msg: dict[int, list[dict]] = {}
fb_by_msg: dict[int, tuple] = {}
if include_audits and rows:
ids = [r[0] for r in rows]
ph = ",".join("?" * len(ids))
async with db.execute(
f"""SELECT m.id, a.id, a.tool_name, a.confirmed_at, a.corrected_at, a.expected_group
FROM conversations m JOIN tool_audit_log a ON a.conversation_id = m.id
WHERE m.id IN ({ph}) ORDER BY m.id, a.id""",
tuple(ids),
) as cur:
for mid, aid, tool, c_at, cor_at, grp in await cur.fetchall():
audits_by_msg.setdefault(mid, []).append(
{"audit_id": aid, "tool_name": tool,
"confirmed_at": c_at, "corrected_at": cor_at, "expected_group": grp}
)
# Message-level verdict (👍/👎) for messages without tool audits —
# same /chat/history-only gate so context never sees it.
async with db.execute(
f"SELECT message_id, value, note FROM message_feedback WHERE message_id IN ({ph})",
tuple(ids),
) as cur:
for mid, val, note in await cur.fetchall():
fb_by_msg[mid] = (val, note)
result = []
for r in reversed(rows):
item = {"id": r[0], "role": r[1], "content": r[2], "timestamp": r[4]}
if r[3]:
try:
item["images"] = json.loads(r[3])
except Exception:
pass
if include_reasoning and r[5]:
item["reasoning"] = r[5]
if include_audits and r[0] in audits_by_msg:
item["audits"] = audits_by_msg[r[0]]
if include_audits and r[0] in fb_by_msg:
fbval, fbnote = fb_by_msg[r[0]]
item["feedback"] = fbval
if fbnote:
item["feedback_note"] = fbnote
result.append(item)
return result
async def get_all_history(user_id: str = "default", limit: int | None = None, offset: int = 0) -> dict[str, list[dict]]:
"""Return every conversation grouped by session_id, oldest first.
Used by the single-request Android chat sync pull so a phone can
restore its sidebar in one HTTP round-trip instead of one call per
session (which trips the 30 rpm rate limiter). ``limit``/``offset``
paginate that pull; the default (None) keeps the historical unbounded
contract for the sync caller.
"""
import json
db = await get_db()
history: dict[str, list[dict]] = {}
sql = (
"""SELECT session_id, id, role, content, images, timestamp, reasoning
FROM conversations WHERE user_id = ? ORDER BY session_id, id ASC"""
)
params: tuple = (user_id,)
if limit is not None:
sql += " LIMIT ? OFFSET ?"
params = (user_id, limit, offset)
async with db.execute(sql, params) as cur:
rows = await cur.fetchall()
for sid, mid, role, content, images, ts, reasoning in rows:
item = {"id": mid, "role": role, "content": content, "timestamp": ts}
if images:
try:
item["images"] = json.loads(images)
except Exception:
pass
if reasoning:
item["reasoning"] = reasoning
history.setdefault(sid, []).append(item)
return history
async def clear_history(session_id: str, user_id: str = "default"):
db = await get_db()
async with _write_lock():
# Feedback/audit cascade first: their subqueries need the rows present.
await db.execute(
"DELETE FROM message_feedback WHERE message_id IN "
"(SELECT id FROM conversations WHERE session_id = ? AND user_id = ?)",
(session_id, user_id),
)
await db.execute(
"DELETE FROM tool_audit_log WHERE conversation_id IN "
"(SELECT id FROM conversations WHERE session_id = ? AND user_id = ?)",
(session_id, user_id),
)
# FTS first: the rowid subquery needs the conversation rows still present.
await db.execute(
"DELETE FROM conversations_fts WHERE rowid IN "
"(SELECT id FROM conversations WHERE session_id = ? AND user_id = ?)",
(session_id, user_id),
)
await db.execute("DELETE FROM conversations WHERE session_id = ? AND user_id = ?", (session_id, user_id))
await db.execute("DELETE FROM sessions WHERE id = ? AND user_id = ?", (session_id, user_id))
await db.execute("DELETE FROM email_session_map WHERE session_id = ? AND user_id = ?", (session_id, user_id))
await db.execute("DELETE FROM notes_session_map WHERE session_id = ? AND user_id = ?", (session_id, user_id))
await db.execute("DELETE FROM tasks_session_map WHERE session_id = ? AND user_id = ?", (session_id, user_id))
await db.execute("DELETE FROM calendar_session_map WHERE session_id = ? AND user_id = ?", (session_id, user_id))
await _repair_summary_boundaries(db)
await _commit_with_retry(db)
async def import_messages(session_id: str, messages: list[dict], client_key: str, user_id: str = "default") -> int:
"""Idempotently import a batch of chat messages from a paired device.
Each message must carry a stable `client_key` (uuid + index) so re-syncs
never duplicate rows. Session metadata is created on demand and the
conversation title is regenerated from the first user message if missing.
"""
db = await get_db()
inserted = 0
async with _write_lock():
for i, m in enumerate(messages):
role = (m.get("role") or "").strip()
content = (m.get("content") or "").strip()
if role not in ("user", "assistant") or not content:
continue
key = f"{client_key}:{i}"
async with db.execute(
"SELECT 1 FROM conversations WHERE client_key = ? AND user_id = ?", (key, user_id)
) as cur:
if await cur.fetchone():
continue
try:
cur = await db.execute(
"INSERT INTO conversations (session_id, role, content, client_key, user_id) VALUES (?, ?, ?, ?, ?)",
(session_id, role, content, key, user_id),
)
rowid = cur.lastrowid
try:
await db.execute(
"INSERT INTO conversations_fts (rowid, content, session_id) VALUES (?, ?, ?)",
(rowid, content, session_id),
)
except Exception as fts_e:
# FTS5 is best-effort; a failure must not eat the message row.
logger.warning(f"import FTS insert failed (non-fatal): {fts_e}")
inserted += 1
except Exception as e:
# A lost TOCTOU race lands here as a UNIQUE violation: the
# winning task already inserted this key, so skipping is the
# correct idempotent outcome (debug, not warning).
if "UNIQUE constraint failed" in str(e):
logger.debug(f"import deduped by constraint (session={session_id}): {e}")
else:
logger.warning(f"import insert failed (session={session_id}): {e}")
continue
if inserted:
await db.execute(
"INSERT INTO sessions (id, user_id) VALUES (?, ?) "
"ON CONFLICT(id) DO UPDATE SET last_active = CURRENT_TIMESTAMP "
"WHERE sessions.user_id = excluded.user_id",
(session_id, user_id),
)
# Title from first user message if the server has none.
async with db.execute("SELECT name FROM sessions WHERE id = ? AND user_id = ?", (session_id, user_id)) as cur:
row = await cur.fetchone()
if not row or not row[0]:
title_src = next((m.get("content", "").strip() for m in messages if m.get("role") == "user"), "")
if title_src:
from title import generate_instant_title
await db.execute(
"UPDATE sessions SET name = ? WHERE id = ? AND user_id = ?",
(generate_instant_title(title_src), session_id, user_id),
)
await _commit_with_retry(db)
return inserted
async def search_sessions(query: str, limit: int = 20, user_id: str = "default") -> list[dict]:
"""Hybrid search: FTS5 keyword + semantic embedding.
FTS5 gives exact keyword matches (ranked by BM25), semantic gives meaning
matches (e.g. 'hava nasıl' ↔ 'weather outside') via cosine similarity.
Results are merged, deduplicated, and capped at `limit`.
Falls back to LIKE if FTS5 query fails.
"""
import re as _re
db = await get_db()
# Sanitize for FTS5: strip special match characters
safe_q = _re.sub(r'[^\w\s]', '', query).strip()
if not safe_q:
return []
# FTS: try AND first (precise), fallback to OR if no hits
terms = safe_q.split()
fts_query_and = " AND ".join(terms)
fts_query_or = " OR ".join(terms)
rows: list = []
try:
async with db.execute("""
SELECT f.session_id, s.name,
snippet(conversations_fts, 0, '<b>', '</b>', '…', 12) AS snippet
FROM conversations_fts f
JOIN sessions s ON s.id = f.session_id AND s.user_id = ?
WHERE conversations_fts MATCH ?
ORDER BY rank
LIMIT ?
""", (user_id, fts_query_and, limit)) as cur:
rows = await cur.fetchall()
# If AND gave no results and query has multiple terms, try OR for recall
if not rows and len(terms) > 1:
async with db.execute("""
SELECT f.session_id, s.name,
snippet(conversations_fts, 0, '<b>', '</b>', '…', 12) AS snippet
FROM conversations_fts f
JOIN sessions s ON s.id = f.session_id AND s.user_id = ?
WHERE conversations_fts MATCH ?
ORDER BY rank
LIMIT ?
""", (user_id, fts_query_or, limit)) as cur:
rows = await cur.fetchall()
except Exception:
# Fallback: LIKE on raw table (safe_q, escape %/_)
safe_like = safe_q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
async with db.execute("""
SELECT DISTINCT c.session_id, s.name, c.content
FROM conversations c
LEFT JOIN sessions s ON s.id = c.session_id AND s.user_id = ?
WHERE c.content LIKE ? ESCAPE '\\' AND c.user_id = ?
ORDER BY c.timestamp DESC
LIMIT ?
""", (user_id, f"%{safe_like}%", user_id, limit)) as cur:
rows = await cur.fetchall()
seen = set()
results = []
for r in rows:
if r[0] not in seen:
seen.add(r[0])
results.append({"session_id": r[0], "name": r[1] or "", "snippet": r[2] or ""})
# Semantic supplement: only for multi-word queries and when FTS is sparse.
# Single-word exact matches are better served by FTS (semantic gives many
# false positives for short queries: "omlet" vs "teşekkürler" 0.637).
# Threshold 0.50 for higher precision (memories uses 0.35, but search needs stricter).
if len(results) < 5 and len(terms) > 1 and len(results) < limit:
try:
from embedding import cosine_similarity, embed_async
query_emb = await embed_async(safe_q)
async with db.execute("""
SELECT c.session_id, s.name, c.content, c.embedding
FROM conversations c LEFT JOIN sessions s ON s.id = c.session_id AND s.user_id = c.user_id
WHERE c.embedding IS NOT NULL AND c.user_id = ?
ORDER BY c.timestamp DESC LIMIT 200
""", (user_id,)) as cur:
cand_rows = await cur.fetchall()
scored: list[tuple[float, str, str, str]] = []
for sess_id, name, content, blob in cand_rows:
if not blob or sess_id in seen:
continue
try:
sim = cosine_similarity(query_emb, blob)
except Exception:
continue
if sim >= 0.50:
scored.append((sim, sess_id, name or "", content))
scored.sort(key=lambda x: x[0], reverse=True)
for sim, sess_id, name, content in scored[: limit - len(results)]:
if sess_id not in seen:
snippet = content[:80].replace("\n", " ").strip() + "…"
results.append({"session_id": sess_id, "name": name, "snippet": snippet})
seen.add(sess_id)
except Exception as e:
logger.warning(f"Semantic search supplement failed (non-fatal): {e}")
return results
# -- Email list-number → message-ID mapping --
#
# The model sees email lists as "1., 2., ..." and is told never to show raw
# IDs. This table persists the mapping from that list number to the real IMAP
# message ID per session, so a follow-up like "read email 3" keeps resolving
# correctly even after a restart or when a session is resumed later. Rows are
# replaced on every list_emails / search_emails call.
async def save_email_map(session_id: str, emails: list[dict], user_id: str = "default"):
"""Replace the stored email listing for a session with a new one."""
if not session_id or not emails:
return
user_id = user_id or "default"
db = await get_db()
async with _write_lock():
await db.execute(
"DELETE FROM email_session_map WHERE session_id = ? AND user_id = ?",
(session_id, user_id),
)
for seq, m in enumerate(emails, 1):
await db.execute(
"INSERT INTO email_session_map (session_id, seq, message_id, subject, sender, preview, user_id) "
"VALUES (?, ?, ?, ?, ?, ?, ?)",
(
session_id,
seq,
str(m.get("id", "")),
m.get("subject", "")[:200],
m.get("from", "")[:200],
(m.get("body", "") or "")[:200],
user_id,
),
)
await _commit_with_retry(db)
async def get_email_map(session_id: str, user_id: str = "default") -> list[dict]:
"""Return the persisted email listing for a session (list-number order)."""
if not session_id:
return []
user_id = user_id or "default"
db = await get_db()
async with db.execute(
"""SELECT message_id, subject, sender, preview FROM email_session_map
WHERE session_id = ? AND user_id = ? ORDER BY seq ASC""",
(session_id, user_id),
) as cur:
rows = await cur.fetchall()
return [
{"id": r[0], "subject": r[1], "from": r[2], "preview": r[3]}
for r in rows
]
async def clear_email_map(session_id: str):
"""Drop the stored email listing for a session (used on history clear)."""
if not session_id:
return
db = await get_db()
await _write_with_retry(db, "DELETE FROM email_session_map WHERE session_id = ?", (session_id,))
# -- Notes Session Map --
# Same pattern as email_session_map: the model sees numbered note lists
# ("1., 2., ...") and this table maps list positions to real Nextcloud note
# IDs per session, so "read note 3" resolves correctly.
async def save_notes_map(session_id: str, notes: list[dict], user_id: str = "default"):
"""Replace the stored note listing for a session with a new one."""
if not session_id or not notes:
return
user_id = user_id or "default"
db = await get_db()
async with _write_lock():
await db.execute(
"DELETE FROM notes_session_map WHERE session_id = ? AND user_id = ?",
(session_id, user_id),
)
for seq, n in enumerate(notes, 1):
await db.execute(
"INSERT INTO notes_session_map (session_id, seq, note_id, title, category, preview, user_id) "
"VALUES (?, ?, ?, ?, ?, ?, ?)",
(
session_id,
seq,
int(n.get("id", 0)),
n.get("title", "")[:200],
n.get("category", "")[:100],
(n.get("content", "") or "")[:200],
user_id,
),
)
await _commit_with_retry(db)
async def get_notes_map(session_id: str, user_id: str = "default") -> list[dict]:
"""Return the persisted note listing for a session (list-number order)."""
if not session_id:
return []
user_id = user_id or "default"
db = await get_db()
async with db.execute(
"""SELECT note_id, title, category, preview FROM notes_session_map
WHERE session_id = ? AND user_id = ? ORDER BY seq ASC""",
(session_id, user_id,),
) as cur:
rows = await cur.fetchall()
return [
{"id": r[0], "title": r[1], "category": r[2], "preview": r[3]}
for r in rows
]
async def clear_notes_map(session_id: str):
"""Drop the stored note listing for a session (used on history clear)."""
if not session_id:
return
db = await get_db()
await _write_with_retry(db, "DELETE FROM notes_session_map WHERE session_id = ?", (session_id,))
# -- Tasks Session Map --
# Same pattern as email/notes: model sees numbered list, we map to real UIDs.
async def save_tasks_map(session_id: str, tasks: list[dict], user_id: str = "default"):
"""Replace the stored task listing for a session with a new one."""
if not session_id or not tasks:
return
user_id = user_id or "default"
db = await get_db()
async with _write_lock():
await db.execute(
"DELETE FROM tasks_session_map WHERE session_id = ? AND user_id = ?",
(session_id, user_id),
)
for seq, t in enumerate(tasks, 1):
await db.execute(
"INSERT INTO tasks_session_map (session_id, seq, uid, summary, due, priority, completed, user_id) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(
session_id,
seq,
str(t.get("uid", "")),
t.get("summary", "")[:200],
t.get("due", "")[:50],
int(t.get("priority", 0)),
1 if t.get("completed") else 0,
user_id,
),
)
await _commit_with_retry(db)
async def get_tasks_map(session_id: str, user_id: str = "default") -> list[dict]:
"""Return the persisted task listing for a session (list-number order)."""
if not session_id:
return []
user_id = user_id or "default"
db = await get_db()
async with db.execute(
"""SELECT uid, summary, due, priority, completed FROM tasks_session_map
WHERE session_id = ? AND user_id = ? ORDER BY seq ASC""",
(session_id, user_id,),
) as cur:
rows = await cur.fetchall()
return [
{"uid": r[0], "summary": r[1], "due": r[2], "priority": r[3], "completed": bool(r[4])}
for r in rows
]
async def clear_tasks_map(session_id: str):
"""Drop the stored task listing for a session (used on history clear)."""
if not session_id:
return
db = await get_db()
await _write_with_retry(db, "DELETE FROM tasks_session_map WHERE session_id = ?", (session_id,))
# -- Sync Items (two-way device sync) --
def _normalize_sync_ts(raw) -> str | None:
"""Canonical fixed-width ISO-8601 UTC ("YYYY-MM-DDTHH:MM:SS.ffffffZ").
Every comparison path (LWW ``WHERE excluded.updated_at > ...``, the
``since`` pull filter) is a STRING compare — only safe when all values
share one format and width, where lexicographic order == chronological
order. Free-text client timestamps made that a lie (mixed offsets, no
microseconds: "…:00Z" vs "…:00.5Z" compares backwards). Unparseable
input returns None (the caller drops the item / ignores the filter).
"""
if not isinstance(raw, str) or not raw.strip():
return None
s = raw.strip()
if s.endswith(("Z", "z")):
s = s[:-1] + "+00:00"
try:
import datetime as _dt
dt = _dt.datetime.fromisoformat(s)
except ValueError:
return None
if dt.tzinfo is not None:
dt = dt.astimezone(_dt.timezone.utc).replace(tzinfo=None)
return dt.strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z"
async def _normalize_stored_sync_timestamps(db: aiosqlite.Connection) -> None:
"""Every-boot repair: rewrite stored updated_at values to canonical form.
Idempotent — only rows whose value differs from its normalization are
rewritten, so post-migration boots touch nothing. Canonical rows make
the LWW and ``since`` string comparisons valid on legacy data too.
"""
try:
cur = await db.execute("SELECT rowid, updated_at FROM sync_items")
rows = await cur.fetchall()
except Exception:
return # table missing pre-migration — nothing to do
for rowid, ts in rows:
norm = _normalize_sync_ts(ts)
if norm is not None and norm != ts:
await db.execute(
"UPDATE sync_items SET updated_at = ? WHERE rowid = ?",
(norm, rowid),
)
async def upsert_sync_items(user_id: str, items: list[dict]) -> int:
"""Upsert remote sync items for a user. Each item: uuid|entity_type|data|updated_at|deleted|origin.
Last-write-wins: a STRICTLY newer updated_at overwrites the stored row
wholesale; an equal or older write is dropped (a delayed old push can
never clobber newer data). Timestamps are normalized to canonical
fixed-width UTC so the comparison is valid. Returns the number of rows
actually stored.
"""
if not items:
return 0
db = await get_db()
n = 0
async with _write_lock():
for it in items:
uuid = str(it.get("uuid") or "").strip()
etype = str(it.get("entity_type") or "").strip() or "generic"
data = it.get("data") or {}
if not isinstance(data, str):
data = json.dumps(data, ensure_ascii=False, default=str)
ts = _normalize_sync_ts(it.get("updated_at"))
deleted = 1 if it.get("deleted") else 0
origin = str(it.get("origin") or "").strip()
if not uuid or ts is None:
continue # fail-closed: no key or unparseable timestamp
await db.execute(
"""INSERT INTO sync_items (uuid, user_id, entity_type, data, updated_at, deleted, origin)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(user_id, uuid) DO UPDATE SET
data = excluded.data,
entity_type = excluded.entity_type,
updated_at = excluded.updated_at,
deleted = excluded.deleted,
origin = excluded.origin
WHERE excluded.updated_at > sync_items.updated_at""",
(uuid, user_id, etype, data, ts, deleted, origin),
)
# Py3.12: INSERT rowcount is -1 — read SELECT changes() instead.
# Counts 1 for a real insert/update, 0 for an LWW-dropped write.
ccur = await db.execute("SELECT changes()")
n += (await ccur.fetchone())[0]
await _commit_with_retry(db)
return n
async def get_sync_items(user_id: str, entity_type: str | None = None, since: str | None = None) -> list[dict]:
"""Return all sync items (or a subtype) for a user, optionally filtered by
`since` (ISO updated_at cutoff, exclusive — normalized to canonical UTC,
so client offsets are fine). Includes tombstones (deleted=1).
"""
db = await get_db()
since_norm = _normalize_sync_ts(since) if since else None
if since and since_norm is None:
# Unparseable cutoff: return everything rather than a wrong subset —
# the client re-filters from the full set.
since_norm = None
if entity_type and since_norm:
sql = ("SELECT uuid, entity_type, data, updated_at, deleted, origin FROM sync_items "
"WHERE user_id = ? AND entity_type = ? AND updated_at > ?")
params = (user_id, entity_type, since_norm)
elif entity_type:
sql = "SELECT uuid, entity_type, data, updated_at, deleted, origin FROM sync_items WHERE user_id = ? AND entity_type = ?"
params = (user_id, entity_type)
elif since_norm:
sql = "SELECT uuid, entity_type, data, updated_at, deleted, origin FROM sync_items WHERE user_id = ? AND updated_at > ?"
params = (user_id, since_norm)
else:
sql = "SELECT uuid, entity_type, data, updated_at, deleted, origin FROM sync_items WHERE user_id = ?"
params = (user_id,)
async with db.execute(sql, params) as cur:
rows = await cur.fetchall()
out = []
for uuid, etype, data, ts, deleted, origin in rows:
parsed = {}
if data:
try:
parsed = json.loads(data)
except Exception:
parsed = {"_raw": data}
out.append({
"uuid": uuid,
"entity_type": etype,
"data": parsed,
"updated_at": ts,
"origin": origin or "",
"deleted": bool(deleted),
})
return out
# -- Calendar Session Map --
# Same pattern as email/notes/tasks: model sees numbered list, we map to real UIDs.
async def save_calendar_map(session_id: str, events: list[dict], user_id: str = "default"):
"""Replace the stored calendar listing for a session with a new one."""
if not session_id or not events:
return
user_id = user_id or "default"
db = await get_db()
async with _write_lock():
await db.execute(
"DELETE FROM calendar_session_map WHERE session_id = ? AND user_id = ?",
(session_id, user_id),
)
for seq, ev in enumerate(events, 1):
await db.execute(
"INSERT INTO calendar_session_map (session_id, seq, uid, summary, start_time, user_id) "
"VALUES (?, ?, ?, ?, ?, ?)",
(
session_id,
seq,
str(ev.get("uid", "")),
ev.get("summary", "")[:200],
ev.get("start", "")[:50],
user_id,
),
)
await _commit_with_retry(db)
async def get_calendar_map(session_id: str, user_id: str = "default") -> list[dict]:
"""Return the persisted calendar listing for a session (list-number order)."""
if not session_id:
return []
user_id = user_id or "default"
db = await get_db()
async with db.execute(
"""SELECT uid, summary, start_time FROM calendar_session_map
WHERE session_id = ? AND user_id = ? ORDER BY seq ASC""",
(session_id, user_id,),
) as cur:
rows = await cur.fetchall()
return [
{"uid": r[0], "summary": r[1], "start": r[2]}
for r in rows
]
async def clear_calendar_map(session_id: str):
"""Drop the stored calendar listing for a session (used on history clear)."""
if not session_id:
return
db = await get_db()
await _write_with_retry(db, "DELETE FROM calendar_session_map WHERE session_id = ?", (session_id,))
# -- Sessions --
async def update_session_name(session_id: str, name: str, user_id: str = "default"):
db = await get_db()
async with _write_lock():
await db.execute(
"""INSERT INTO sessions (id, name, user_id) VALUES (?, ?, ?)
ON CONFLICT(id) DO UPDATE SET name = excluded.name, last_active = CURRENT_TIMESTAMP
WHERE sessions.user_id = excluded.user_id""",
(session_id, name, user_id),
)
await _commit_with_retry(db)
async def get_all_sessions(user_id: str = "default") -> list[dict]:
db = await get_db()
async with db.execute(
"""SELECT s.id, s.created_at, s.last_active, s.name,
COUNT(c.session_id) as msg_count
FROM sessions s
LEFT JOIN conversations c ON c.session_id = s.id AND c.user_id = ?
WHERE s.user_id = ?
GROUP BY s.id
ORDER BY s.last_active DESC""",
(user_id, user_id),
) as cur:
rows = await cur.fetchall()
return [
{"session_id": r[0], "created_at": r[1], "last_active": r[2],
"name": r[3], "message_count": r[4]}
for r in rows
]
async def get_session_meta(session_id: str, user_id: str = "default") -> dict:
db = await get_db()
async with db.execute(
"SELECT summary, summarized_until FROM sessions WHERE id = ? AND user_id = ?",
(session_id, user_id),
) as cur:
row = await cur.fetchone()
if row is None:
return {"summary": "", "summarized_until": 0}
return {"summary": row[0] or "", "summarized_until": row[1] or 0}
async def session_owned_by(session_id: str, user_id: str) -> bool:
"""True when the session row belongs to user_id.
Used by the abort guard: a foreign session must be indistinguishable
from an idle one (no existence oracle). Sessions are created by
save_message before any generation starts, so a live abort event
always has an owner row to check against.
"""
db = await get_db()
async with db.execute(
"SELECT 1 FROM sessions WHERE id = ? AND user_id = ?",
(session_id, user_id),
) as cur:
return await cur.fetchone() is not None
async def session_foreign(session_id: str, user_id: str) -> bool:
"""True when the session row exists but belongs to a different user.
Distinct from ``not session_owned_by(...)``: a brand-new session id is
NOT foreign (the caller is about to create it); only an existing row
with a different owner is.
"""
db = await get_db()
async with db.execute(
"SELECT 1 FROM sessions WHERE id = ? AND user_id <> ?",
(session_id, user_id),
) as cur:
return await cur.fetchone() is not None
# -- Users (multi-user identity) --
#
# One row per human. API keys are stored as SHA-256 hashes only — the raw key
# exists solely at creation/rotation time (returned once to the caller).
# The pre-multi-user identity keeps id "default": existing rows already point
# there, so admin bootstrap rewrites zero data rows.
DEFAULT_USER_ID = "default"
def generate_api_key() -> str:
"""Generate a new raw API key (returned to the caller exactly once)."""
return secrets.token_urlsafe(32)
def hash_api_key(key: str) -> str:
"""One-way hash for storage/comparison (never persist the raw key)."""
return hashlib.sha256(key.encode("utf-8")).hexdigest()
def _row_to_user(row) -> dict:
return {
"id": row[0],
"name": row[1] or "",
"is_admin": bool(row[2]),
"created_at": row[3],
"is_approved": bool(row[4]) if len(row) > 4 else False,
}
# Display names that would collide with system identities or enable
# impersonation (login is by name; ensure_default_admin owns 'admin').
# Case-insensitive exact match. Homoglyph lookalikes are a known residual.
RESERVED_USERNAMES = frozenset({
"admin", "administrator", "system", "support",
"api", "null", "undefined", "server", "pisynapse",
})
async def count_users() -> int:
db = await get_db()
cur = await db.execute("SELECT COUNT(*) FROM users")
return (await cur.fetchone())[0]
async def get_user(user_id: str) -> dict | None:
db = await get_db()
cur = await db.execute(
"SELECT id, name, is_admin, created_at, password_hash, is_approved FROM users WHERE id = ?", (user_id,)
)
row = await cur.fetchone()
if row is None:
return None
user = _row_to_user((row[0], row[1], row[2], row[3], row[5]))
user["has_password"] = bool(row[4])
return user
async def get_user_by_key_hash(key_hash: str) -> dict | None:
db = await get_db()
cur = await db.execute(
"SELECT id, name, is_admin, created_at, is_approved FROM users WHERE key_hash = ?",
(key_hash,),
)
row = await cur.fetchone()
return _row_to_user(row) if row else None
async def list_users() -> list[dict]:
db = await get_db()
cur = await db.execute("SELECT id, name, is_admin, created_at, is_approved FROM users ORDER BY created_at, id")
return [_row_to_user(r) for r in await cur.fetchall()]
async def create_user(name: str, *, is_admin: bool = False, user_id: str | None = None,
approved: bool = False) -> tuple[dict, str]:
"""Create a user with a fresh API key. Returns (user_dict, raw_key).
The raw key is returned exactly once — only its hash is stored.
Display names are unique case-insensitively (login is by name).
`approved` gates quorum counting (collective learning); admins are
always approved at creation.
"""
import uuid as _uuid
clean = (name or "").strip()
if not clean:
raise ValueError("Name must not be empty")
if clean.casefold() in RESERVED_USERNAMES:
raise ValueError(f"Name is reserved: {clean}")
raw_key = generate_api_key()
uid = user_id or _uuid.uuid4().hex[:12]
db = await get_db()
async with _write_lock():
cur = await db.execute("SELECT 1 FROM users WHERE lower(name) = lower(?)", (clean,))
if await cur.fetchone():
raise ValueError(f"Name already taken: {clean}")
await db.execute(
"INSERT INTO users (id, name, key_hash, is_admin, is_approved) VALUES (?, ?, ?, ?, ?)",
(uid, clean[:100], hash_api_key(raw_key), 1 if is_admin else 0,
1 if (approved or is_admin) else 0),
)
await _commit_with_retry(db)
user = await get_user(uid)
assert user is not None
return user, raw_key
async def rotate_user_key(user_id: str) -> str | None:
"""Replace a user's API key. Returns the new raw key once, else None."""
raw_key = generate_api_key()
db = await get_db()
async with _write_lock():
cur = await db.execute(
"UPDATE users SET key_hash = ? WHERE id = ?", (hash_api_key(raw_key), user_id)
)
await _commit_with_retry(db)
if cur.rowcount == 0:
return None
invalidate_user_cache(user_id)
return raw_key
async def get_user_by_name(name: str) -> dict | None:
"""Case-insensitive lookup for password login. Returns row with hash."""
db = await get_db()
cur = await db.execute(
"SELECT id, name, is_admin, created_at, password_hash, is_approved FROM users "
"WHERE lower(name) = lower(?)",
((name or "").strip(),),
)
row = await cur.fetchone()
if row is None:
return None
user = _row_to_user((row[0], row[1], row[2], row[3], row[5]))
user["has_password"] = bool(row[4])
return user
async def set_password(user_id: str, raw_password: str) -> None:
"""Store a bcrypt hash (CPU work offloaded to a thread)."""
import asyncio as _asyncio
import bcrypt as _bcrypt
hashed = await _asyncio.to_thread(
_bcrypt.hashpw, raw_password.encode("utf-8"), _bcrypt.gensalt()
)
db = await get_db()
async with _write_lock():
cur = await db.execute(
"UPDATE users SET password_hash = ? WHERE id = ?",
(hashed.decode("utf-8"), user_id),
)
await _commit_with_retry(db)
if cur.rowcount == 0:
raise ValueError(f"Unknown user: {user_id}")
async def verify_password(user_id: str, raw_password: str) -> bool:
"""Check a password against the stored bcrypt hash (thread-offloaded)."""
import asyncio as _asyncio
import bcrypt as _bcrypt
db = await get_db()
cur = await db.execute("SELECT password_hash FROM users WHERE id = ?", (user_id,))
row = await cur.fetchone()
if row is None or not row[0]:
return False
try:
return await _asyncio.to_thread(
_bcrypt.checkpw, raw_password.encode("utf-8"), row[0].encode("utf-8")
)
except (ValueError, TypeError):
return False
async def create_device_key(user_id: str, device_name: str = "") -> tuple[dict, str]:
"""Issue an additional per-device API key. Returns (row, raw_key once)."""
import uuid as _uuid
raw_key = generate_api_key()
kid = _uuid.uuid4().hex[:12]
label = (device_name or "").strip()[:100]
db = await get_db()
async with _write_lock():
await db.execute(
"INSERT INTO user_api_keys (id, user_id, name, key_hash) VALUES (?, ?, ?, ?)",
(kid, user_id, label, hash_api_key(raw_key)),
)
await _commit_with_retry(db)
return {"id": kid, "user_id": user_id, "name": label}, raw_key
async def list_device_keys(user_id: str) -> list[dict]:
"""Device keys for a user (never includes hashes)."""
db = await get_db()
cur = await db.execute(
"SELECT id, user_id, name, created_at FROM user_api_keys "
"WHERE user_id = ? ORDER BY created_at, id",
(user_id,),
)
return [
{"id": r[0], "user_id": r[1], "name": r[2], "created_at": r[3]}
for r in await cur.fetchall()
]
async def revoke_device_key(user_id: str, key_id: str) -> bool:
"""Delete one of the user's device keys. Returns True when removed."""
db = await get_db()
async with _write_lock():
cur = await db.execute(
"SELECT key_hash FROM user_api_keys WHERE id = ? AND user_id = ?",
(key_id, user_id),
)
row = await cur.fetchone()
if row is None:
return False
await db.execute(
"DELETE FROM user_api_keys WHERE id = ? AND user_id = ?", (key_id, user_id)
)
await _commit_with_retry(db)
_user_cache.pop(row[0], None)
return True
# ── Browser sessions (cookie auth; API keys stay server-side) ─────────────────
# Opaque tokens: 32 random bytes, SHA-256 stored (like API keys — the raw
# value exists only at creation). autouse-tested hermeticity needs no clock
# mocking: expiry is compared in SQL against CURRENT_TIMESTAMP.
SESSION_COOKIE_NAME = "ps_session"
SESSION_TTL_DAYS = 30
SESSION_TOUCH_DAYS = 15
def _hash_session_token(raw: str) -> str:
import hashlib as _hl
return _hl.sha256(raw.encode("utf-8")).hexdigest()
async def create_session(user_id: str, device: str = "") -> tuple[dict, str]:
"""Open a browser session. Returns (session_row, raw_token) — raw once."""
import secrets as _secrets
import uuid as _uuid
raw = _secrets.token_urlsafe(32)
sid = _uuid.uuid4().hex[:12]
db = await get_db()
async with _write_lock():
await db.execute(
"INSERT INTO user_sessions (id, user_id, token_hash, device, expires_at) "
"VALUES (?, ?, ?, ?, datetime('now', ?))",
(sid, user_id, _hash_session_token(raw), (device or "")[:100], f"+{SESSION_TTL_DAYS} days"),
)
await _commit_with_retry(db)
return {"id": sid, "user_id": user_id}, raw
async def resolve_session(raw_token: str) -> dict | None:
"""Resolve a session cookie to its user dict, or None. Never raises.
Sliding expiry: sessions idle past half their TTL get a fresh window on
use, so daily users never re-login while abandoned ones die on schedule.
"""
from datetime import datetime as _dt
token = (raw_token or "").strip()
if not token:
return None
try:
db = await get_db()
cur = await db.execute(
"SELECT s.id, s.user_id, s.expires_at FROM user_sessions s WHERE s.token_hash = ?",
(_hash_session_token(token),),
)
row = await cur.fetchone()
if not row:
return None
try:
exp = _dt.strptime(row[2][:19], "%Y-%m-%d %H:%M:%S")
except Exception:
return None
now = _dt.utcnow()
if exp <= now:
async with _write_lock():
await db.execute("DELETE FROM user_sessions WHERE id = ?", (row[0],))
await _commit_with_retry(db)
return None
if (exp - now).days < SESSION_TOUCH_DAYS:
async with _write_lock():
await db.execute(
"UPDATE user_sessions SET expires_at = datetime('now', ?), last_seen = CURRENT_TIMESTAMP WHERE id = ?",
(f"+{SESSION_TTL_DAYS} days", row[0]),
)
await _commit_with_retry(db)
return await get_user(row[1])
except Exception:
return None
async def revoke_session(raw_token: str) -> bool:
"""Revoke one session (logout). Always safe to call."""
token = (raw_token or "").strip()
if not token:
return False
try:
db = await get_db()
async with _write_lock():
cur = await db.execute(
"DELETE FROM user_sessions WHERE token_hash = ?", (_hash_session_token(token),)
)
await _commit_with_retry(db)
return cur.rowcount > 0
except Exception:
return False
async def revoke_other_sessions(user_id: str, keep_raw_token: str) -> int:
"""Revoke all of a user's sessions except the current one (password change).
Returns the revoked count. Never raises.
"""
try:
db = await get_db()
async with _write_lock():
cur = await db.execute(
"DELETE FROM user_sessions WHERE user_id = ? AND token_hash != ?",
(user_id, _hash_session_token((keep_raw_token or "").strip())),
)
await _commit_with_retry(db)
return cur.rowcount
except Exception:
return 0
async def delete_user(user_id: str) -> bool:
"""Delete a user and ALL of their data (privacy-complete wipe).
Removes conversations (+FTS rows), sessions, memories, session maps,
tool audits, feedback on their messages, sync items, personal settings,
device keys, browser sessions, quorum votes and the user row itself — in one locked transaction.
Returns False when the user does not exist.
"""
db = await get_db()
async with _write_lock():
cur = await db.execute("SELECT 1 FROM users WHERE id = ?", (user_id,))
if await cur.fetchone() is None:
return False
try:
await db.execute(
"DELETE FROM conversations_fts WHERE rowid IN "
"(SELECT id FROM conversations WHERE user_id = ?)",
(user_id,),
)
except Exception as fts_e:
# Same external-content quirk as the dedupe path: missing index
# entries mean nothing to orphan.
logger.warning(f"FTS cleanup skipped during user delete (non-fatal): {fts_e}")
await db.execute("DELETE FROM conversations WHERE user_id = ?", (user_id,))
await db.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,))
await db.execute("DELETE FROM memories WHERE user_id = ?", (user_id,))
for _map_tbl in ("email_session_map", "notes_session_map", "tasks_session_map",
"calendar_session_map"):
await db.execute(
f"DELETE FROM {_map_tbl} WHERE user_id = ?", (user_id,)
)
await db.execute("DELETE FROM tool_audit_log WHERE user_id = ?", (user_id,))
await db.execute(
"DELETE FROM message_feedback WHERE message_id NOT IN (SELECT id FROM conversations)"
)
await db.execute("DELETE FROM sync_items WHERE user_id = ?", (user_id,))
await db.execute("DELETE FROM user_settings WHERE user_id = ?", (user_id,))
await db.execute("DELETE FROM user_api_keys WHERE user_id = ?", (user_id,))
await db.execute("DELETE FROM user_sessions WHERE user_id = ?", (user_id,))
await db.execute("DELETE FROM feedback_votes WHERE user_id = ?", (user_id,))
await db.execute("DELETE FROM user_credentials WHERE user_id = ?", (user_id,))
await db.execute("DELETE FROM users WHERE id = ?", (user_id,))
await _commit_with_retry(db)
invalidate_user_cache(user_id)
return True
async def ensure_default_admin() -> dict | None:
"""Bootstrap the admin onto the pre-existing 'default' identity.
- Users table non-empty → nothing to do (returns None).
- Empty + `.env` API_KEY set + LEGACY DATA ROWS (conversations/sessions/
memories from a pre-multi-user install) → create admin id 'default'
bound to that key's hash (the single-user owner's key keeps working,
now as admin).
- Empty + no key, or empty + key but a FRESH database (no rows anywhere)
→ nothing (the first registration becomes admin — installer or browser).
This keeps fresh installs free of phantom admins and closes the
first-visitor race: ownership is explicit, never positional.
Existing data rows already point at 'default': zero rewrites, and the
invisibility invariant holds from the first multi-user row on.
"""
db = await get_db()
cur = await db.execute("SELECT COUNT(*) FROM users")
if (await cur.fetchone())[0] > 0:
return None
import os as _os
env_key = (_os.getenv("API_KEY") or "").strip()
if not env_key:
return None
legacy = False
for _tbl in ("conversations", "sessions", "memories"):
try:
cur = await db.execute(f"SELECT 1 FROM {_tbl} LIMIT 1")
if await cur.fetchone() is not None:
legacy = True
break
except Exception:
continue
if not legacy:
return None
async with _write_lock():
await db.execute(
"INSERT OR IGNORE INTO users (id, name, key_hash, is_admin, is_approved) VALUES (?, ?, ?, 1, 1)",
(DEFAULT_USER_ID, "admin", hash_api_key(env_key)),
)
await _commit_with_retry(db)
return await get_user(DEFAULT_USER_ID)
async def set_approved(user_id: str, approved: bool) -> bool:
"""Set a user's quorum-approval flag (admin only via endpoint).
Approval gates collective-learning quorum counting only — chat, tools,
settings and feedback all work identically for unapproved users.
Returns False when the user does not exist.
"""
db = await get_db()
async with _write_lock():
cur = await db.execute(
"UPDATE users SET is_approved = ? WHERE id = ?",
(1 if approved else 0, user_id),
)
await _commit_with_retry(db)
ok = cur.rowcount > 0
if ok:
invalidate_user_cache(user_id)
return ok
_USER_CACHE_TTL = 60.0
_user_cache: dict[str, tuple[dict, float]] = {}
def invalidate_user_cache(user_id: str | None = None) -> None:
"""Drop cached key resolutions (all, or one user's). Call on rotation."""
if user_id is None:
_user_cache.clear()
return
for k in [k for k, (u, _) in _user_cache.items() if u.get("id") == user_id]:
_user_cache.pop(k, None)
async def resolve_user_by_key(api_key: str) -> dict | None:
"""Resolve a raw API key to its user dict, or None. Never raises.
Order: legacy `.env` API_KEY (maps to the default identity, preserving
single-user behavior) → DB `key_hash` lookup with a short TTL cache.
DB errors resolve to None (deny closed; health still reports the outage).
"""
key = (api_key or "").strip()
if not key:
return None
digest = hash_api_key(key)
now = time.monotonic()
hit = _user_cache.get(digest)
if hit is not None:
user, expires = hit
if expires > now:
return user
try:
import os as _os
env_key = (_os.getenv("API_KEY") or "").strip()
user: dict | None = None
if env_key and hmac.compare_digest(key, env_key):
user = await get_user(DEFAULT_USER_ID)
if user is None:
user = {"id": DEFAULT_USER_ID, "name": "", "is_admin": True, "created_at": None}
else:
user = await get_user_by_key_hash(digest)
if user is None:
db = await get_db()
cur = await db.execute(
"SELECT user_id FROM user_api_keys WHERE key_hash = ?", (digest,)
)
row = await cur.fetchone()
if row is not None:
user = await get_user(row[0])
except Exception as e:
logger.warning(f"User resolution failed (denying closed): {e}")
return None
if user is not None:
_user_cache[digest] = (user, now + _USER_CACHE_TTL)
return user
async def get_user_settings(user_id: str) -> dict[str, str]:
"""Return all personal settings for a user as {key: value}."""
db = await get_db()
cur = await db.execute(
"SELECT key, value FROM user_settings WHERE user_id = ?", (user_id,)
)
return {r[0]: r[1] for r in await cur.fetchall()}
async def seed_admin_settings() -> int:
"""One-time migration: copy current `.env` personal values into the
default admin's rows — but only keys the admin has never set (never
overwrite a deliberate user choice). Returns the seeded count.
Idempotent; safe to run on every boot (usually a no-op).
"""
import os as _os
from config import PERSONAL_KEYS
stored = await get_user_settings(DEFAULT_USER_ID)
missing = [k for k in PERSONAL_KEYS if k not in stored and (_os.getenv(k) or "").strip()]
if not missing:
return 0
db = await get_db()
async with _write_lock():
for key in sorted(missing):
await db.execute(
"INSERT OR IGNORE INTO user_settings (user_id, key, value, updated_at) "
"VALUES (?, ?, ?, CURRENT_TIMESTAMP)",
(DEFAULT_USER_ID, key, _os.getenv(key, "").strip()),
)
await _commit_with_retry(db)
return len(missing)
async def set_user_setting(user_id: str, key: str, value: str) -> None:
"""Upsert one personal setting. Raises ValueError for non-personal keys."""
from config import PERSONAL_KEYS
if key not in PERSONAL_KEYS:
raise ValueError(f"Not a personal setting: {key}")
db = await get_db()
async with _write_lock():
await db.execute(
"INSERT INTO user_settings (user_id, key, value, updated_at) "
"VALUES (?, ?, ?, CURRENT_TIMESTAMP) "
"ON CONFLICT(user_id, key) DO UPDATE SET value = excluded.value, "
"updated_at = CURRENT_TIMESTAMP",
(user_id, key, value),
)
await _commit_with_retry(db)
async def effective_setting(user_id: str | None, key: str, default: str = "") -> str:
"""Resolve a setting with user > global precedence.
Personal keys: the user's row wins; otherwise the live global value
(env-derived config attr); otherwise the given default. Unknown or
empty user falls straight to global.
"""
if user_id:
try:
db = await get_db()
cur = await db.execute(
"SELECT value FROM user_settings WHERE user_id = ? AND key = ?",
(user_id, key),
)
row = await cur.fetchone()
if row is not None:
return row[0]
except Exception as e:
logger.warning(f"Personal setting lookup failed ({key}): {e}")
import config as _cfg
val = getattr(_cfg, key, default)
return default if val is None else str(val)
# -- Rolling Summary --
# Bounds for a single fold. Normal sessions fold a handful of messages per
# turn; without a cap, a long-unfolded session (or a lowered history limit)
# would hand the summarizer an unbounded transcript and blow the engine's
# context (observed: 15k input tokens vs a 6656 budget on-device; the server
# default context is 8192). Folding is oldest-first and the boundary advances
# by exactly what was folded, so a huge backlog drains over successive turns.
FOLD_MAX_MESSAGES = 30
FOLD_MAX_CHARS_PER_MESSAGE = 500
async def _repair_summary_boundaries(db: aiosqlite.Connection) -> None:
"""Reconcile summary boundaries with surviving messages. Call inside the
caller's write lock, before its commit, after any delete path (branch /
last / clear / retention).
- Clamps every session's ``summarized_until`` to its surviving MAX(id):
otherwise the fold computes an empty pending span forever (stall) while
the prompt keeps injecting a summary of deleted content.
- Clears summary+until where no messages remain.
Stale facts in a surviving summary are NOT purged here — they dilute as
new folds append (documented residual); re-summarizing from scratch on a
synchronous delete path would stall the API.
"""
await db.execute(
"UPDATE sessions SET summarized_until = MIN(summarized_until, "
"COALESCE((SELECT MAX(c.id) FROM conversations c "
"WHERE c.session_id = sessions.id AND c.user_id = sessions.user_id), 0))"
)
await db.execute(
"UPDATE sessions SET summary = '', summarized_until = 0 "
"WHERE NOT EXISTS (SELECT 1 FROM conversations c "
"WHERE c.session_id = sessions.id AND c.user_id = sessions.user_id)"
)
async def get_messages_to_summarize(
session_id: str,
history_limit: int,
summarized_until: int,
batch_size: int,
early_trigger: int = 0,
user_id: str = "default",
) -> tuple[list[dict], int]:
"""Return the next batch of aged-out messages not yet folded into the summary.
early_trigger: if >0 and this is the first summary (summarized_until == 0),
fire as soon as this many messages are available.
"""
db = await get_db()
async with db.execute(
"SELECT id FROM conversations WHERE session_id = ? AND user_id = ? ORDER BY id ASC",
(session_id, user_id),
) as cur:
ids = [r[0] for r in await cur.fetchall()]
total = len(ids)
if total <= history_limit:
return [], summarized_until
boundary_id = ids[total - history_limit - 1]
pending_ids = [i for i in ids if summarized_until < i <= boundary_id]
pending_count = len(pending_ids)
effective_batch = (
early_trigger
if (early_trigger > 0 and summarized_until == 0 and pending_count >= early_trigger)
else batch_size
)
if pending_count < effective_batch:
return [], summarized_until
# Fold at most FOLD_MAX_MESSAGES, oldest-first; the boundary advances by
# exactly the folded span so the remainder drains on later turns.
take_end = pending_ids[min(len(pending_ids), FOLD_MAX_MESSAGES) - 1]
async with db.execute(
"""SELECT role, content FROM conversations
WHERE session_id = ? AND user_id = ? AND id > ? AND id <= ?
ORDER BY id ASC""",
(session_id, user_id, summarized_until, take_end),
) as cur:
rows = await cur.fetchall()
out = []
for role, content in rows:
c = content or ""
if len(c) > FOLD_MAX_CHARS_PER_MESSAGE:
c = c[:FOLD_MAX_CHARS_PER_MESSAGE] + "…"
if c.strip():
out.append({"role": role, "content": c})
if not out:
# Nothing foldable (all blank): advance past the empty span so the
# fold does not retry it on every turn.
return [], take_end
return out, take_end
async def update_session_summary(session_id: str, summary: str, summarized_until: int, user_id: str = "default",
expected_until: int | None = None) -> bool:
"""Store a folded summary; returns True when applied.
``expected_until`` enables a conditional write for the background fold
race: two concurrent ``_update_summary`` tasks read the same boundary,
the loser must not clobber the winner's wider summary. When given, the
write applies only if the stored boundary still equals it (changes() 0 =
skipped). ``None`` keeps the unconditional legacy behavior (still checks
the owner guard on the conflict path).
"""
db = await get_db()
async with _write_lock():
if expected_until is None:
await db.execute(
"""INSERT INTO sessions (id, summary, summarized_until, user_id) VALUES (?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET summary = excluded.summary, summarized_until = excluded.summarized_until,
last_active = CURRENT_TIMESTAMP
WHERE sessions.user_id = excluded.user_id""",
(session_id, summary, summarized_until, user_id),
)
else:
await db.execute(
"""INSERT INTO sessions (id, summary, summarized_until, user_id) VALUES (?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET summary = excluded.summary, summarized_until = excluded.summarized_until,
last_active = CURRENT_TIMESTAMP
WHERE sessions.user_id = excluded.user_id AND sessions.summarized_until = ?""",
(session_id, summary, summarized_until, user_id, expected_until),
)
# Python >= 3.12 reports cursor.rowcount = -1 for every INSERT (upserts
# included), which made this function return False unconditionally;
# changes() is the portable source of truth for "did the write apply".
applied = (await (await db.execute("SELECT changes()")).fetchone())[0] > 0
await _commit_with_retry(db)
return applied
# -- Long-term Memories --
async def save_memory(content: str, category: str = "general",
importance: int = 5, user_id: str | None = None) -> tuple[str, int]:
from config import DEFAULT_USER, MEMORY_SIMILARITY_THRESHOLD
from embedding import cosine_similarity, embed_async
user_id = user_id or DEFAULT_USER
try:
new_embedding = await embed_async(content)
except Exception as e:
logger.error(f"Embedding failed, saving memory without vector: {e}")
new_embedding = None
db = await get_db()
if new_embedding is not None:
async with db.execute(
"SELECT id, embedding FROM memories WHERE user_id = ? AND embedding IS NOT NULL",
(user_id,),
) as cur:
existing = await cur.fetchall()
for mem_id, existing_embedding in existing:
if cosine_similarity(new_embedding, existing_embedding) >= MEMORY_SIMILARITY_THRESHOLD:
async with _write_lock():
await db.execute(
"""UPDATE memories SET last_accessed = CURRENT_TIMESTAMP,
access_count = access_count + 1 WHERE id = ?""",
(mem_id,),
)
await _commit_with_retry(db)
return "Memory updated (similar content exists).", mem_id
async with _write_lock():
cur = await db.execute(
"INSERT INTO memories (user_id, content, category, importance, embedding) VALUES (?, ?, ?, ?, ?)",
(user_id, content, category, importance, new_embedding),
)
await _commit_with_retry(db)
rowid = cur.lastrowid
return "Memory saved.", rowid
async def search_memories(query: str, user_id: str | None = None, limit: int = 5, query_embedding: bytes | None = None) -> list[dict]:
from config import DEFAULT_USER
from embedding import cosine_similarity, embed_async
user_id = user_id or DEFAULT_USER
try:
if query_embedding is None:
query_embedding = await embed_async(query)
except Exception as e:
logger.error(f"Embedding failed for memory search: {e}")
return []
db = await get_db()
# Phase 1: load only IDs + embeddings for fast similarity scoring
async with db.execute(
"SELECT id, embedding FROM memories WHERE user_id = ?",
(user_id,),
) as cur:
rows = await cur.fetchall()
scored = []
backfill_ids = []
for mem_id, blob in rows:
if blob is None:
backfill_ids.append(mem_id)
continue
sim = cosine_similarity(query_embedding, blob)
scored.append((sim, mem_id))
scored.sort(key=lambda x: x[0], reverse=True)
top = [(sim, mid) for sim, mid in scored[:limit] if sim >= 0.35]
# Phase 2: fetch full content only for top results
result = []
if top:
placeholders = ",".join("?" * len(top))
ids = [m_id for _, m_id in top]
async with db.execute(
f"SELECT id, content, category, importance, created_at FROM memories WHERE id IN ({placeholders})",
ids,
) as cur:
content_map = {r[0]: r[1:] for r in await cur.fetchall()}
for sim, mem_id in top:
c, cat, imp, cat_at = content_map.get(mem_id, ("", "", 0, ""))
result.append({
"id": mem_id, "content": c, "category": cat,
"importance": imp, "created_at": cat_at, "similarity": sim,
})
if backfill_ids:
async def _backfill():
# Own connection: the shared `db` is mid-writing the caller's
# search results, and a backfill commit on it would commit the
# caller's partially-assembled transaction too. WAL allows
# concurrent writers from a separate connection safely.
async with _BACKFILL_SEMAPHORE:
conn = await aiosqlite.connect(DB_PATH)
try:
await conn.execute("PRAGMA busy_timeout=10000")
for i in range(0, len(backfill_ids), _BACKFILL_BATCH_SIZE):
batch = backfill_ids[i : i + _BACKFILL_BATCH_SIZE]
for mid in batch:
try:
row = await conn.execute("SELECT content FROM memories WHERE id = ?", (mid,))
r = await row.fetchone()
if r:
emb = await embed_async(r[0])
await conn.execute("UPDATE memories SET embedding = ? WHERE id = ?", (emb, mid))
except Exception as e:
logger.error("Embedding backfill failed for memory %s: %s", mid, e)
await conn.commit()
# Small yield between batches to avoid starving the event loop
await asyncio.sleep(0.05)
except asyncio.CancelledError:
pass
finally:
await conn.close()
global _backfill_task
# Cancel any previous backfill to avoid duplicate work
if _backfill_task and not _backfill_task.done():
_backfill_task.cancel()
_backfill_task = asyncio.create_task(_backfill())
# Update access stats for retrieved results. The write AND its commit
# must hold the process write lock: an unconditional commit on the
# shared connection would commit another coroutine's partially-
# assembled transaction (D2/D4). Commit only when we actually wrote —
# a no-hit search has nothing of its own to flush.
if result:
ids = [m["id"] for m in result]
placeholders = ",".join("?" * len(ids))
async with _write_lock():
await db.execute(
f"""UPDATE memories SET last_accessed = CURRENT_TIMESTAMP,
access_count = access_count + 1 WHERE id IN ({placeholders})""",
ids,
)
await _commit_with_retry(db)
return result
async def get_memories(user_id: str | None = None, limit: int = 10) -> list[dict]:
from config import DEFAULT_USER
user_id = user_id or DEFAULT_USER
try:
db = await get_db()
async with db.execute(
"""SELECT id, content, category, importance, created_at
FROM memories WHERE user_id = ?
ORDER BY importance DESC, last_accessed DESC LIMIT ?""",
(user_id, limit),
) as cur:
rows = await cur.fetchall()
return [{"id": r[0], "content": r[1], "category": r[2],
"importance": r[3], "created_at": r[4]} for r in rows]
except Exception:
logger.exception("Failed to retrieve memories")
return []
async def delete_memory(user_id: str, memory_id: int):
db = await get_db()
async with _write_lock():
await db.execute("DELETE FROM memories WHERE id = ? AND user_id = ?", (memory_id, user_id))
await _commit_with_retry(db)
async def get_all_memories(user_id: str | None = None) -> list[dict]:
from config import DEFAULT_USER
user_id = user_id or DEFAULT_USER
try:
db = await get_db()
async with db.execute(
"""SELECT id, content, category, importance, created_at
FROM memories WHERE user_id = ? ORDER BY importance DESC""",
(user_id,),
) as cur:
rows = await cur.fetchall()
return [{"id": r[0], "content": r[1], "category": r[2],
"importance": r[3], "created_at": r[4]} for r in rows]
except Exception:
logger.exception("Failed to retrieve all memories")
return []