"""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 []