"""
Encrypted chat and scratchpad persistence: single SQLite DB, key from macOS Keychain.
- messages: thread_id, role, content (encrypted)
- scratchpads: id, tab_id, title (encrypted), body (encrypted), position, created_at, updated_at
"""
from __future__ import annotations

import json
import logging
import math
import os
import sqlite3
import struct
import unicodedata
import uuid
import hashlib
import re
import threading
from datetime import datetime, timedelta, timezone
from typing import Any

try:
    from persistence import MUGI_DIR
except ImportError:
    MUGI_DIR = os.path.expanduser("~/.mugi")

CHATS_DB_FILENAME = "chats.db"
DB_PATH = os.path.join(MUGI_DIR, CHATS_DB_FILENAME)

_KEYRING_SERVICE = "Mugi"
_KEYRING_ACCOUNT = "chat_encryption_key"

_fernet = None
_key_just_created = False


def _set_keychain_item_accessible_when_unlocked(service, account):
    """Set the Keychain item to kSecAttrAccessibleWhenUnlockedThisDeviceOnly so it's not synced and only readable when the device is unlocked."""
    try:
        from Security import (
            SecItemUpdate,
            kSecClass, kSecClassGenericPassword,
            kSecAttrService, kSecAttrAccount,
            kSecAttrAccessible, kSecAttrAccessibleWhenUnlockedThisDeviceOnly,
        )
        query = {kSecClass: kSecClassGenericPassword, kSecAttrService: service, kSecAttrAccount: account}
        update = {kSecAttrAccessible: kSecAttrAccessibleWhenUnlockedThisDeviceOnly}
        SecItemUpdate(query, update)
    except Exception:
        pass


def _get_fernet():
    """Lazy-init Fernet with key from macOS Keychain; create and store key if missing.

    Test-only escape hatch: when ``MUGI_SKIP_BACKEND_BOOTSTRAP=1`` is set AND
    ``MUGI_TEST_FERNET_KEY`` is provided, that key is used instead of touching the
    macOS Keychain. This lets ``tests/migrations/`` and ``tests/smoke/`` exercise
    the encrypted persistence layer without polluting the user's Keychain or
    needing host permissions. The guard is doubly gated (skip-bootstrap AND
    test-key) so it cannot accidentally activate in a real backend run.
    """
    global _fernet, _key_just_created
    if _fernet is not None:
        return _fernet
    try:
        from cryptography.fernet import Fernet
    except ImportError as e:
        raise RuntimeError(
            "chat_storage requires cryptography. "
            "Install with: pip install cryptography"
        ) from e

    test_key = os.environ.get("MUGI_TEST_FERNET_KEY", "").strip()
    if test_key and os.environ.get("MUGI_SKIP_BACKEND_BOOTSTRAP") == "1":
        _fernet = Fernet(test_key.encode("ascii"))
        _key_just_created = True
        return _fernet

    try:
        import keyring
    except ImportError as e:
        raise RuntimeError(
            "chat_storage requires keyring and cryptography. "
            "Install with: pip install keyring cryptography"
        ) from e
    # Force macOS Keychain backend so py2app bundles it (keyring discovers backends via entry points otherwise missing in frozen app).
    try:
        import keyring.backends.macOS  # noqa: F401
    except ImportError:
        pass

    key_b64 = keyring.get_password(_KEYRING_SERVICE, _KEYRING_ACCOUNT)
    if not key_b64:
        key = Fernet.generate_key()
        key_b64 = key.decode("ascii")
        try:
            keyring.set_password(_KEYRING_SERVICE, _KEYRING_ACCOUNT, key_b64)
            _set_keychain_item_accessible_when_unlocked(_KEYRING_SERVICE, _KEYRING_ACCOUNT)
        except Exception as e:
            raise RuntimeError(
                "Could not store encryption key in Keychain. "
                "On macOS, ensure keyring is installed (pip install keyring)."
            ) from e
        _key_just_created = True
    _fernet = Fernet(key_b64.encode("ascii"))
    return _fernet


def probe_encryption_ready() -> bool:
    """True when a Fernet key already exists; does not create Keychain material (DEF-034)."""
    global _fernet
    if _fernet is not None:
        return True
    test_key = os.environ.get("MUGI_TEST_FERNET_KEY", "").strip()
    if test_key and os.environ.get("MUGI_SKIP_BACKEND_BOOTSTRAP") == "1":
        return True
    try:
        import keyring
    except ImportError:
        return False
    try:
        import keyring.backends.macOS  # noqa: F401
    except ImportError:
        pass
    try:
        return bool(keyring.get_password(_KEYRING_SERVICE, _KEYRING_ACCOUNT))
    except Exception:
        return False


def ensure_encryption_ready() -> bool:
    """Ensure the encryption key and DB are ready. Returns True if the key was just created this run."""
    _get_fernet()
    global _key_just_created
    try:
        return _key_just_created
    finally:
        _key_just_created = False


def _encrypt(content: str) -> bytes:
    if not content:
        return _get_fernet().encrypt(b"")
    return _get_fernet().encrypt(content.encode("utf-8"))


def _decrypt(data: bytes) -> str:
    if not data:
        return ""
    return _get_fernet().decrypt(data).decode("utf-8")


def _encrypt_bytes(data: bytes) -> bytes:
    return _get_fernet().encrypt(data or b"")


def _decrypt_bytes(data: bytes) -> bytes:
    if not data:
        return b""
    return _get_fernet().decrypt(data)


def _get_connection() -> sqlite3.Connection:
    os.makedirs(os.path.dirname(DB_PATH), exist_ok=True, mode=0o700)
    # Capture newness BEFORE connect: sqlite3.connect() creates the file
    # immediately, so checking existence afterward would always report False
    # and skip the 0o600 lockdown for a freshly-created DB.
    _db_is_new = not os.path.exists(DB_PATH)
    conn = sqlite3.connect(DB_PATH)
    # Wait up to 5s on SQLITE_BUSY instead of failing immediately under load.
    conn.execute("PRAGMA busy_timeout=5000")
    path_key = os.path.abspath(DB_PATH)
    if path_key not in _schema_initialized_paths:
        with _schema_init_lock:
            if path_key not in _schema_initialized_paths:
                _bootstrap_schema(conn, path_key, is_new=_db_is_new)
                _schema_initialized_paths.add(path_key)
    _run_migrations(conn)
    return conn


def _bootstrap_schema(conn: sqlite3.Connection, path_key: str, *, is_new: bool) -> None:
    """One-time WAL + CREATE DDL for a DB path (June 2026 storm amplifier fix).

    ``journal_mode=WAL`` is persistent in the file; re-running it on every
    connection was the #1 hot path in the freeze diag (Pragma → pwrite).
    ``is_new`` must be sampled by the caller before ``sqlite3.connect`` opens
    (and thereby creates) the file.
    """
    conn.execute("PRAGMA journal_mode=WAL")
    if is_new:
        try:
            os.chmod(path_key, 0o600)
        except OSError:
            pass
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS messages (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            thread_id TEXT NOT NULL,
            role TEXT NOT NULL,
            content BLOB NOT NULL,
            created_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_messages_thread_id ON messages(thread_id)"
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS scratchpads (
            id TEXT PRIMARY KEY,
            tab_id TEXT NOT NULL,
            title_encrypted BLOB NOT NULL,
            body_encrypted BLOB NOT NULL,
            position INTEGER NOT NULL DEFAULT 0,
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_scratchpads_tab_id ON scratchpads(tab_id)"
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS thread_branches (
            branch_thread_id TEXT PRIMARY KEY,
            parent_thread_id TEXT NOT NULL,
            branch_at_message_index INTEGER NOT NULL
        )
        """
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_thread_branches_parent ON thread_branches(parent_thread_id)"
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS scheduled_jobs (
            id TEXT PRIMARY KEY,
            prompt TEXT NOT NULL,
            schedule TEXT NOT NULL,
            delivery TEXT NOT NULL DEFAULT 'origin',
            origin_thread_id TEXT,
            repeat_count INTEGER DEFAULT 1,
            runs_completed INTEGER DEFAULT 0,
            next_run_at TEXT,
            created_at TEXT NOT NULL,
            last_run_at TEXT,
            status TEXT DEFAULT 'active'
        )
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS thread_context_anchors (
            thread_id TEXT PRIMARY KEY,
            through_message_id INTEGER NOT NULL,
            summary_ciphertext BLOB NOT NULL,
            updated_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS thread_task_plans (
            thread_id TEXT PRIMARY KEY,
            plan_ciphertext BLOB NOT NULL,
            updated_at TEXT NOT NULL
        )
        """
    )
    # Kanban tables — durable, encrypted task board shared across agents.
    # See kanban_storage.py for the public API and architecture.md for the lifecycle.
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS kanban_tasks (
            id TEXT PRIMARY KEY,
            title_ciphertext BLOB NOT NULL,
            body_ciphertext BLOB NOT NULL,
            status TEXT NOT NULL DEFAULT 'todo',
            assignee TEXT NOT NULL DEFAULT 'general',
            tenant TEXT NOT NULL DEFAULT 'default',
            priority INTEGER NOT NULL DEFAULT 3,
            parent_id TEXT,
            origin_thread_id TEXT NOT NULL DEFAULT '',
            tags_ciphertext BLOB NOT NULL,
            metadata_ciphertext BLOB NOT NULL,
            failure_count INTEGER NOT NULL DEFAULT 0,
            last_attempt INTEGER NOT NULL DEFAULT 0,
            claim_handle TEXT NOT NULL DEFAULT '',
            claim_token TEXT NOT NULL DEFAULT '',
            claim_expires_at TEXT NOT NULL DEFAULT '',
            deadline_at TEXT NOT NULL DEFAULT '',
            idempotency_key TEXT NOT NULL DEFAULT '',
            model_override TEXT NOT NULL DEFAULT '',
            max_retries_override INTEGER NOT NULL DEFAULT -1,
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL,
            completed_at TEXT NOT NULL DEFAULT '',
            schema_version INTEGER NOT NULL DEFAULT 1
        )
        """
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_kanban_tasks_status_assignee ON kanban_tasks(status, assignee)"
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_kanban_tasks_parent ON kanban_tasks(parent_id)"
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_kanban_tasks_tenant ON kanban_tasks(tenant)"
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_kanban_tasks_claim_expires ON kanban_tasks(status, claim_expires_at)"
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_kanban_tasks_origin_thread ON kanban_tasks(origin_thread_id)"
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS thread_coordinator_state (
            thread_id TEXT PRIMARY KEY,
            batch_id TEXT NOT NULL,
            activated_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS thread_council_state (
            thread_id TEXT PRIMARY KEY,
            batch_id TEXT NOT NULL,
            activated_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS thread_council_pending_offer (
            thread_id TEXT PRIMARY KEY,
            offer_id TEXT NOT NULL,
            council_frame TEXT NOT NULL,
            frame_hash TEXT NOT NULL,
            rationale TEXT NOT NULL DEFAULT '',
            created_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        "CREATE UNIQUE INDEX IF NOT EXISTS idx_kanban_tasks_idem "
        "ON kanban_tasks(tenant, idempotency_key) WHERE idempotency_key != ''"
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS kanban_task_runs (
            id TEXT PRIMARY KEY,
            task_id TEXT NOT NULL,
            attempt INTEGER NOT NULL,
            outcome TEXT NOT NULL,
            assignee TEXT NOT NULL,
            handle_id TEXT NOT NULL DEFAULT '',
            claim_token TEXT NOT NULL DEFAULT '',
            summary_ciphertext BLOB NOT NULL,
            metadata_ciphertext BLOB NOT NULL,
            error_ciphertext BLOB NOT NULL,
            started_at TEXT NOT NULL,
            ended_at TEXT NOT NULL DEFAULT ''
        )
        """
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_kanban_runs_task ON kanban_task_runs(task_id)"
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_kanban_runs_outcome ON kanban_task_runs(outcome)"
    )
    # Plan B: many-to-many parent edges. The fast-path single-parent column
    # `kanban_tasks.parent_id` stays as the canonical "primary parent" for
    # legacy reads; this join table holds the full parent set used by the
    # cycle walker, AND-all-parents-done promotion, and cascade cancel.
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS kanban_task_parents (
            task_id TEXT NOT NULL,
            parent_id TEXT NOT NULL,
            ordinal INTEGER NOT NULL DEFAULT 0,
            created_at TEXT NOT NULL,
            PRIMARY KEY (task_id, parent_id)
        )
        """
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_kanban_task_parents_task ON kanban_task_parents(task_id)"
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_kanban_task_parents_parent ON kanban_task_parents(parent_id)"
    )
    # Plan B: subscriptions for terminal-state notifications. Target is
    # encrypted because thread ids / webhook URLs / push targets are
    # operator-sensitive.
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS kanban_subscriptions (
            id TEXT PRIMARY KEY,
            task_id TEXT NOT NULL,
            channel TEXT NOT NULL,
            target_ciphertext BLOB NOT NULL,
            fire_on TEXT NOT NULL DEFAULT 'gave_up,done',
            created_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_kanban_subscriptions_task ON kanban_subscriptions(task_id)"
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS capture_projects (
            id TEXT PRIMARY KEY,
            name_ciphertext BLOB NOT NULL,
            description_ciphertext BLOB NOT NULL,
            name_key TEXT NOT NULL UNIQUE,
            sort_order INTEGER NOT NULL DEFAULT 0,
            metadata_json TEXT NOT NULL DEFAULT '{}',
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL,
            archived_at TEXT NOT NULL DEFAULT ''
        )
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS captures (
            id TEXT PRIMARY KEY,
            body_ciphertext BLOB NOT NULL,
            title_ciphertext BLOB NOT NULL,
            reason_ciphertext BLOB NOT NULL,
            normalized_ciphertext BLOB NOT NULL DEFAULT X'',
            body_hash TEXT NOT NULL,
            source TEXT NOT NULL DEFAULT 'quick_panel',
            thread_id TEXT NOT NULL DEFAULT '',
            status TEXT NOT NULL DEFAULT 'inbox',
            kind TEXT NOT NULL DEFAULT 'raw',
            tags_json TEXT NOT NULL DEFAULT '[]',
            attachment_refs_json TEXT NOT NULL DEFAULT '[]',
            metadata_json TEXT NOT NULL DEFAULT '{}',
            project_id TEXT NOT NULL DEFAULT '',
            section TEXT NOT NULL DEFAULT '',
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL,
            deleted_at TEXT NOT NULL DEFAULT ''
        )
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS capture_connections (
            id TEXT PRIMARY KEY,
            summary_ciphertext BLOB NOT NULL,
            why_ciphertext BLOB NOT NULL,
            source_ids_json TEXT NOT NULL DEFAULT '[]',
            status TEXT NOT NULL DEFAULT 'suggested',
            scope TEXT NOT NULL DEFAULT 'theme',
            project_id TEXT NOT NULL DEFAULT '',
            feedback TEXT NOT NULL DEFAULT '',
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS action_cards (
            id TEXT PRIMARY KEY,
            title_ciphertext BLOB NOT NULL,
            body_ciphertext BLOB NOT NULL,
            action_type TEXT NOT NULL DEFAULT 'explain',
            trust_level TEXT NOT NULL DEFAULT 'informational',
            state TEXT NOT NULL DEFAULT 'suggested',
            source_ids_json TEXT NOT NULL DEFAULT '[]',
            project_id TEXT NOT NULL DEFAULT '',
            failed_reason TEXT NOT NULL DEFAULT '',
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS vault_attachments (
            id TEXT PRIMARY KEY,
            note_id TEXT NOT NULL DEFAULT '',
            project_id TEXT NOT NULL DEFAULT '',
            mime_type TEXT NOT NULL,
            byte_size INTEGER NOT NULL,
            storage_path TEXT NOT NULL,
            content_hash TEXT NOT NULL,
            metadata_json TEXT NOT NULL DEFAULT '{}',
            created_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS note_links (
            src_id TEXT NOT NULL,
            dst_id TEXT NOT NULL,
            link_kind TEXT NOT NULL DEFAULT 'wikilink',
            context TEXT NOT NULL DEFAULT '',
            created_at TEXT NOT NULL,
            PRIMARY KEY (src_id, dst_id, link_kind)
        )
        """
    )
    _ensure_vault_schema(conn)
    conn.execute("CREATE INDEX IF NOT EXISTS idx_capture_projects_sort ON capture_projects(archived_at, sort_order, created_at)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_captures_project ON captures(project_id, deleted_at, updated_at)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_captures_status ON captures(status, deleted_at, updated_at)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_captures_hash ON captures(body_hash, deleted_at)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_capture_connections_project ON capture_connections(project_id, status, created_at)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_action_cards_project ON action_cards(project_id, state, created_at)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_vault_attachments_note ON vault_attachments(note_id, created_at)")
    conn.commit()


# ---------- Schema migrations (idempotent, gated by PRAGMA user_version) ------

_TARGET_SCHEMA_VERSION = 10
_migrations_done = False
_schema_initialized_paths: set[str] = set()
_schema_init_lock = threading.Lock()

_SCHEDULE_REWRITE_WORD_BLOCKLIST = frozenset(
    {
        "test",
        "demo",
        "plan",
        "task",
        "note",
        "flow",
        "go",
    }
)


def _table_columns(conn: sqlite3.Connection, table_name: str) -> set[str]:
    try:
        return {str(row[1]) for row in conn.execute(f"PRAGMA table_info({table_name})").fetchall()}
    except Exception:
        return set()


def _add_column_if_missing(conn: sqlite3.Connection, table_name: str, column_name: str, definition: str) -> None:
    if column_name in _table_columns(conn, table_name):
        return
    conn.execute(f"ALTER TABLE {table_name} ADD COLUMN {column_name} {definition}")


def _rewrite_scheduled_prompt_text(prompt: str, mapping: dict[str, str | None]) -> tuple[str, bool]:
    text = str(prompt or "")
    if not text or not isinstance(mapping, dict) or not mapping:
        return text, False

    changed = False
    archived_refs: list[str] = []

    for old_raw, new_raw in mapping.items():
        old = str(old_raw or "").strip().lower()
        if not old:
            continue
        new = None if new_raw is None else str(new_raw).strip().lower()

        escaped_old = re.escape(old)
        explicit_patterns = [
            # skills/<slug>/... or skills/<slug>
            (re.compile(rf"(?i)(\bskills/){escaped_old}(?=/|\b)"), lambda m: f"{m.group(1)}{new}" if new else m.group(0)),
            # `skill:<slug>`
            (re.compile(rf"(?i)(`skill:)({escaped_old})(`)"), lambda m: f"{m.group(1)}{new}{m.group(3)}" if new else m.group(0)),
            # skill_name=<slug>
            (re.compile(rf"(?i)(\bskill_name\s*=\s*)({escaped_old})(\b)"), lambda m: f"{m.group(1)}{new}{m.group(3)}" if new else m.group(0)),
        ]

        local_changed = False
        for pattern, repl in explicit_patterns:
            new_text, count = pattern.subn(repl, text)
            if count > 0:
                text = new_text
                changed = True
                local_changed = True

        allow_whole_word = len(old) >= 8 and old not in _SCHEDULE_REWRITE_WORD_BLOCKLIST
        if allow_whole_word:
            whole_word = re.compile(rf"(?i)\b{escaped_old}\b")
            if new:
                text, count = whole_word.subn(new, text)
                if count > 0:
                    changed = True
                    local_changed = True
            else:
                if whole_word.search(text):
                    local_changed = True

        if (new is None) and local_changed:
            archived_refs.append(old)

    if archived_refs:
        archived_unique = sorted(set(archived_refs))
        note = (
            f"[Note: workflow reference(s) archived: {', '.join(archived_unique)}. "
            "Use skills_list to pick an active replacement.]"
        )
        if note not in text:
            text = (text.rstrip() + "\n\n" + note).strip()
            changed = True

    return text, changed


def rewrite_scheduled_job_prompts(mapping: dict[str, str | None], *, active_only: bool = True) -> dict[str, Any]:
    """Best-effort scheduled job prompt rewrite after skill rename/archive changes."""
    if not isinstance(mapping, dict) or not mapping:
        return {"ok": True, "updated": 0, "scanned": 0}

    conn = _get_connection()
    scanned = 0
    updated = 0
    samples: list[dict[str, str]] = []
    try:
        if active_only:
            rows = conn.execute(
                "SELECT id, prompt FROM scheduled_jobs WHERE status = 'active'"
            ).fetchall()
        else:
            rows = conn.execute("SELECT id, prompt FROM scheduled_jobs").fetchall()
        scanned = len(rows)
        for row in rows:
            try:
                job_id = str(row[0] or "")
                prompt = str(row[1] or "")
                new_prompt, changed = _rewrite_scheduled_prompt_text(prompt, mapping)
                if not changed or new_prompt == prompt:
                    continue
                conn.execute("UPDATE scheduled_jobs SET prompt = ? WHERE id = ?", (new_prompt, job_id))
                updated += 1
                if len(samples) < 5:
                    samples.append(
                        {
                            "id": job_id,
                            "before": prompt[:160],
                            "after": new_prompt[:160],
                        }
                    )
            except Exception:
                continue
        conn.commit()
        if updated:
            logging.getLogger(__name__).info(
                "rewrite_scheduled_job_prompts updated=%d scanned=%d samples=%s",
                updated,
                scanned,
                samples,
            )
        return {"ok": True, "updated": updated, "scanned": scanned, "samples": samples}
    except Exception as exc:
        return {"ok": False, "error": str(exc), "updated": updated, "scanned": scanned, "samples": samples}
    finally:
        conn.close()


def _ensure_vault_schema(conn: sqlite3.Connection) -> None:
    """Repair partial Vault tables from development builds.

    `CREATE TABLE IF NOT EXISTS` does not add columns to an existing table, and
    early Pattern Inbox snapshots may already have created `captures` without
    the newer Vault columns. Keep this additive so current local DBs boot.
    """
    _add_column_if_missing(conn, "capture_projects", "name_ciphertext", "BLOB NOT NULL DEFAULT X''")
    _add_column_if_missing(conn, "capture_projects", "description_ciphertext", "BLOB NOT NULL DEFAULT X''")
    _add_column_if_missing(conn, "capture_projects", "name_key", "TEXT NOT NULL DEFAULT ''")
    _add_column_if_missing(conn, "capture_projects", "sort_order", "INTEGER NOT NULL DEFAULT 0")
    _add_column_if_missing(conn, "capture_projects", "metadata_json", "TEXT NOT NULL DEFAULT '{}'")
    _add_column_if_missing(conn, "capture_projects", "archived_at", "TEXT NOT NULL DEFAULT ''")
    _add_column_if_missing(conn, "capture_projects", "created_at", "TEXT NOT NULL DEFAULT ''")
    _add_column_if_missing(conn, "capture_projects", "updated_at", "TEXT NOT NULL DEFAULT ''")

    _add_column_if_missing(conn, "captures", "body_ciphertext", "BLOB NOT NULL DEFAULT X''")
    _add_column_if_missing(conn, "captures", "title_ciphertext", "BLOB NOT NULL DEFAULT X''")
    _add_column_if_missing(conn, "captures", "reason_ciphertext", "BLOB NOT NULL DEFAULT X''")
    _add_column_if_missing(conn, "captures", "body_hash", "TEXT NOT NULL DEFAULT ''")
    _add_column_if_missing(conn, "captures", "source", "TEXT NOT NULL DEFAULT 'quick_panel'")
    _add_column_if_missing(conn, "captures", "thread_id", "TEXT NOT NULL DEFAULT ''")
    _add_column_if_missing(conn, "captures", "status", "TEXT NOT NULL DEFAULT 'inbox'")
    _add_column_if_missing(conn, "captures", "kind", "TEXT NOT NULL DEFAULT 'raw'")
    _add_column_if_missing(conn, "captures", "tags_json", "TEXT NOT NULL DEFAULT '[]'")
    _add_column_if_missing(conn, "captures", "attachment_refs_json", "TEXT NOT NULL DEFAULT '[]'")
    _add_column_if_missing(conn, "captures", "metadata_json", "TEXT NOT NULL DEFAULT '{}'")
    _add_column_if_missing(conn, "captures", "project_id", "TEXT NOT NULL DEFAULT ''")
    _add_column_if_missing(conn, "captures", "section", "TEXT NOT NULL DEFAULT ''")
    _add_column_if_missing(conn, "captures", "created_at", "TEXT NOT NULL DEFAULT ''")
    _add_column_if_missing(conn, "captures", "updated_at", "TEXT NOT NULL DEFAULT ''")
    _add_column_if_missing(conn, "captures", "deleted_at", "TEXT NOT NULL DEFAULT ''")
    _add_column_if_missing(conn, "captures", "normalized_ciphertext", "BLOB NOT NULL DEFAULT X''")

    _add_column_if_missing(conn, "capture_connections", "feedback", "TEXT NOT NULL DEFAULT ''")
    _add_column_if_missing(conn, "action_cards", "failed_reason", "TEXT NOT NULL DEFAULT ''")
    _add_column_if_missing(conn, "vault_attachments", "note_id", "TEXT NOT NULL DEFAULT ''")
    _add_column_if_missing(conn, "vault_attachments", "project_id", "TEXT NOT NULL DEFAULT ''")

    for project_id, name_blob in conn.execute("SELECT id, name_ciphertext FROM capture_projects WHERE name_key = ''").fetchall():
        try:
            name = _decrypt(name_blob) if name_blob else ""
        except Exception:
            name = ""
        conn.execute(
            "UPDATE capture_projects SET name_key = ? WHERE id = ?",
            (_normalize_project_key(name) or str(project_id), project_id),
        )

    for row_id, body_blob in conn.execute("SELECT id, body_ciphertext FROM captures WHERE body_hash = ''").fetchall():
        try:
            body = _decrypt(body_blob) if body_blob else ""
        except Exception:
            body = ""
        digest = hashlib.sha256(body.encode("utf-8")).hexdigest()
        conn.execute("UPDATE captures SET body_hash = ? WHERE id = ?", (digest, row_id))

    _ensure_vault_graph_tables(conn)
    _ensure_knowledge_graph_tables(conn)


def _ensure_knowledge_graph_tables(conn: sqlite3.Connection) -> None:
    """Chunk embeddings + studied-material graph edges (KB Phase 2)."""
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS knowledge_chunk_embeddings (
            node_id TEXT PRIMARY KEY,
            corpus_slug TEXT NOT NULL,
            row_id INTEGER NOT NULL,
            source_slug TEXT NOT NULL DEFAULT '',
            filename TEXT NOT NULL DEFAULT '',
            label TEXT NOT NULL DEFAULT '',
            content_hash TEXT NOT NULL DEFAULT '',
            rel_path TEXT NOT NULL DEFAULT '',
            repo_root TEXT NOT NULL DEFAULT '',
            model_id TEXT NOT NULL DEFAULT '',
            dim INTEGER NOT NULL,
            embedding BLOB NOT NULL,
            updated_at TEXT NOT NULL
        )
        """
    )
    _add_column_if_missing(conn, "knowledge_chunk_embeddings", "model_id", "TEXT NOT NULL DEFAULT ''")
    _add_column_if_missing(conn, "knowledge_chunk_embeddings", "embed_identity", "TEXT NOT NULL DEFAULT ''")
    _add_column_if_missing(conn, "knowledge_chunk_embeddings", "chunker_version", "TEXT NOT NULL DEFAULT ''")
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_knowledge_chunk_embeddings_corpus ON knowledge_chunk_embeddings(corpus_slug)"
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS knowledge_graph_edges (
            id TEXT PRIMARY KEY,
            node_low TEXT NOT NULL,
            node_high TEXT NOT NULL,
            src_id TEXT NOT NULL,
            dst_id TEXT NOT NULL,
            link_kind TEXT NOT NULL,
            status TEXT NOT NULL DEFAULT 'suggested',
            confidence REAL,
            corpus_slug TEXT NOT NULL DEFAULT '',
            rationale TEXT NOT NULL DEFAULT '',
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        "CREATE UNIQUE INDEX IF NOT EXISTS idx_knowledge_graph_edges_canonical ON knowledge_graph_edges(node_low, node_high, link_kind)"
    )
    conn.execute("CREATE INDEX IF NOT EXISTS idx_knowledge_graph_edges_src ON knowledge_graph_edges(src_id)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_knowledge_graph_edges_dst ON knowledge_graph_edges(dst_id)")
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_knowledge_graph_edges_status ON knowledge_graph_edges(status, corpus_slug)"
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS knowledge_entities (
            entity_key TEXT NOT NULL,
            corpus_slug TEXT NOT NULL,
            canonical_key TEXT NOT NULL DEFAULT '',
            label TEXT NOT NULL DEFAULT '',
            kind TEXT NOT NULL DEFAULT 'term',
            salience REAL NOT NULL DEFAULT 0,
            mention_count INTEGER NOT NULL DEFAULT 0,
            source_slugs TEXT NOT NULL DEFAULT '[]',
            chunk_node_ids TEXT NOT NULL DEFAULT '[]',
            keywords TEXT NOT NULL DEFAULT '[]',
            algo_version TEXT NOT NULL DEFAULT '',
            updated_at TEXT NOT NULL,
            PRIMARY KEY (corpus_slug, entity_key)
        )
        """
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_knowledge_entities_corpus ON knowledge_entities(corpus_slug)"
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS knowledge_claims (
            id TEXT PRIMARY KEY,
            corpus_slug TEXT NOT NULL,
            claim_text TEXT NOT NULL DEFAULT '',
            entity_keys TEXT NOT NULL DEFAULT '[]',
            entity_labels TEXT NOT NULL DEFAULT '[]',
            chunk_node_ids TEXT NOT NULL DEFAULT '[]',
            source_slug TEXT NOT NULL DEFAULT '',
            confidence REAL NOT NULL DEFAULT 0,
            llm_phrased INTEGER NOT NULL DEFAULT 0,
            algo_version TEXT NOT NULL DEFAULT '',
            updated_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_knowledge_claims_corpus ON knowledge_claims(corpus_slug)"
    )


def _json_list_field(value: Any) -> str:
    import json

    if isinstance(value, str):
        return value
    try:
        return json.dumps(list(value) if value is not None else [], ensure_ascii=False)
    except (TypeError, ValueError):
        return "[]"


def _decode_json_list_field(raw: Any) -> list[Any]:
    import json

    if raw is None:
        return []
    if isinstance(raw, list):
        return list(raw)
    text = str(raw or "").strip()
    if not text:
        return []
    try:
        parsed = json.loads(text)
        return list(parsed) if isinstance(parsed, list) else []
    except (json.JSONDecodeError, TypeError, ValueError):
        return []


def replace_corpus_entities(corpus_slug: str, rows: list[dict[str, Any]]) -> int:
    """Replace entity rows; keep existing claims (single transaction with claims)."""
    entities_written, _claims_written = replace_corpus_entities_and_claims(
        corpus_slug,
        rows,
        load_corpus_claims(corpus_slug),
    )
    return entities_written


def replace_corpus_claims(corpus_slug: str, rows: list[dict[str, Any]]) -> int:
    """Replace claim rows; keep existing entities (single transaction with entities)."""
    _entities_written, claims_written = replace_corpus_entities_and_claims(
        corpus_slug,
        load_corpus_entities(corpus_slug),
        rows,
    )
    return claims_written


def purge_corpus_entities_and_claims(corpus_slug: str) -> tuple[int, int]:
    """Delete all entity and claim rows for a corpus (e.g. on corpus delete)."""
    from utils.knowledge_connections_store import _validate_slug

    try:
        slug = _validate_slug(corpus_slug)
    except ValueError:
        return 0, 0
    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        ent_row = conn.execute(
            "SELECT COUNT(*) FROM knowledge_entities WHERE corpus_slug = ?",
            (slug,),
        ).fetchone()
        claim_row = conn.execute(
            "SELECT COUNT(*) FROM knowledge_claims WHERE corpus_slug = ?",
            (slug,),
        ).fetchone()
        conn.execute("DELETE FROM knowledge_entities WHERE corpus_slug = ?", (slug,))
        conn.execute("DELETE FROM knowledge_claims WHERE corpus_slug = ?", (slug,))
        conn.commit()
        return int(ent_row[0] or 0) if ent_row else 0, int(claim_row[0] or 0) if claim_row else 0
    finally:
        conn.close()


def replace_corpus_entities_and_claims(
    corpus_slug: str,
    entity_rows: list[dict[str, Any]],
    claim_rows: list[dict[str, Any]],
) -> tuple[int, int]:
    """Replace entity + claim rows for a corpus in one transaction."""
    from utils.knowledge_connections_store import _validate_slug

    slug = _validate_slug(corpus_slug)
    now = _now_iso()
    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        conn.execute("BEGIN")
        conn.execute("DELETE FROM knowledge_entities WHERE corpus_slug = ?", (slug,))
        conn.execute("DELETE FROM knowledge_claims WHERE corpus_slug = ?", (slug,))
        entities_written = 0
        for row in entity_rows:
            if not isinstance(row, dict):
                continue
            entity_key = str(row.get("entity_key") or row.get("canonical_key") or "").strip()
            if not entity_key:
                continue
            conn.execute(
                """INSERT INTO knowledge_entities
                   (entity_key, corpus_slug, canonical_key, label, kind, salience, mention_count,
                    source_slugs, chunk_node_ids, keywords, algo_version, updated_at)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                (
                    entity_key,
                    slug,
                    str(row.get("canonical_key") or entity_key),
                    str(row.get("label") or ""),
                    str(row.get("kind") or "term"),
                    float(row.get("salience") or 0),
                    int(row.get("mention_count") or 0),
                    _json_list_field(row.get("source_slugs")),
                    _json_list_field(row.get("chunk_node_ids")),
                    _json_list_field(row.get("keywords")),
                    str(row.get("algo_version") or ""),
                    now,
                ),
            )
            entities_written += 1
        claims_written = 0
        seen_ids: set[str] = set()
        for row in claim_rows:
            if not isinstance(row, dict):
                continue
            claim_id = str(row.get("id") or "").strip()
            if not claim_id or claim_id in seen_ids:
                continue
            seen_ids.add(claim_id)
            conn.execute(
                """INSERT INTO knowledge_claims
                   (id, corpus_slug, claim_text, entity_keys, entity_labels, chunk_node_ids,
                    source_slug, confidence, llm_phrased, algo_version, updated_at)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                (
                    claim_id,
                    slug,
                    str(row.get("claim_text") or ""),
                    _json_list_field(row.get("entity_keys")),
                    _json_list_field(row.get("entity_labels")),
                    _json_list_field(row.get("chunk_node_ids")),
                    str(row.get("source_slug") or ""),
                    float(row.get("confidence") or 0),
                    1 if row.get("llm_phrased") else 0,
                    str(row.get("algo_version") or ""),
                    now,
                ),
            )
            claims_written += 1
        conn.commit()
        return entities_written, claims_written
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def load_corpus_entities(corpus_slug: str) -> list[dict[str, Any]]:
    from utils.knowledge_connections_store import _validate_slug

    slug = _validate_slug(corpus_slug)
    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        rows = conn.execute(
            """SELECT entity_key, canonical_key, label, kind, salience, mention_count,
                      source_slugs, chunk_node_ids, keywords, algo_version
               FROM knowledge_entities WHERE corpus_slug = ? ORDER BY salience DESC, entity_key""",
            (slug,),
        ).fetchall()
        out: list[dict[str, Any]] = []
        for row in rows:
            out.append(
                {
                    "entity_key": str(row[0] or ""),
                    "canonical_key": str(row[1] or ""),
                    "label": str(row[2] or ""),
                    "kind": str(row[3] or "term"),
                    "salience": float(row[4] or 0),
                    "mention_count": int(row[5] or 0),
                    "source_slugs": _decode_json_list_field(row[6]),
                    "chunk_node_ids": _decode_json_list_field(row[7]),
                    "keywords": _decode_json_list_field(row[8]),
                    "algo_version": str(row[9] or ""),
                }
            )
        return out
    finally:
        conn.close()


def load_corpus_claims(corpus_slug: str) -> list[dict[str, Any]]:
    from utils.knowledge_connections_store import _validate_slug

    slug = _validate_slug(corpus_slug)
    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        rows = conn.execute(
            """SELECT id, claim_text, entity_keys, entity_labels, chunk_node_ids,
                      source_slug, confidence, llm_phrased, algo_version
               FROM knowledge_claims WHERE corpus_slug = ? ORDER BY confidence DESC, id""",
            (slug,),
        ).fetchall()
        out: list[dict[str, Any]] = []
        for row in rows:
            out.append(
                {
                    "id": str(row[0] or ""),
                    "claim_text": str(row[1] or ""),
                    "entity_keys": _decode_json_list_field(row[2]),
                    "entity_labels": _decode_json_list_field(row[3]),
                    "chunk_node_ids": _decode_json_list_field(row[4]),
                    "source_slug": str(row[5] or ""),
                    "confidence": float(row[6] or 0),
                    "llm_phrased": bool(row[7]),
                    "algo_version": str(row[8] or ""),
                }
            )
        return out
    finally:
        conn.close()


def _ensure_vault_graph_tables(conn: sqlite3.Connection) -> None:
    """Knowledge-graph edges + per-note embeddings (schema v6).

    Kept out of the versioned migration chain's exclusive path so DBs that hit
    ``user_version >= target`` without these tables (failed migration, tooling,
    or restores) self-repair on every open. Statements are IF NOT EXISTS.
    """
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS vault_graph_edges (
            id TEXT PRIMARY KEY,
            note_low TEXT NOT NULL,
            note_high TEXT NOT NULL,
            src_id TEXT NOT NULL,
            dst_id TEXT NOT NULL,
            link_kind TEXT NOT NULL,
            status TEXT NOT NULL DEFAULT 'suggested',
            confidence REAL,
            source TEXT NOT NULL DEFAULT 'system',
            rationale TEXT NOT NULL DEFAULT '',
            project_id TEXT NOT NULL DEFAULT '',
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        "CREATE UNIQUE INDEX IF NOT EXISTS idx_vault_graph_edges_canonical ON vault_graph_edges(note_low, note_high, link_kind)"
    )
    conn.execute("CREATE INDEX IF NOT EXISTS idx_vault_graph_edges_src ON vault_graph_edges(src_id)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_vault_graph_edges_dst ON vault_graph_edges(dst_id)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_vault_graph_edges_status ON vault_graph_edges(status, project_id)")
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS capture_embeddings (
            note_id TEXT PRIMARY KEY,
            content_hash TEXT NOT NULL,
            model_id TEXT NOT NULL,
            dim INTEGER NOT NULL,
            embedding BLOB NOT NULL,
            updated_at TEXT NOT NULL
        )
        """
    )
    _ensure_knowledge_graph_tables(conn)


def _run_migrations(conn: sqlite3.Connection) -> None:
    global _migrations_done
    if _migrations_done:
        _ensure_vault_schema(conn)
        conn.commit()
        return
    version = conn.execute("PRAGMA user_version").fetchone()[0]
    if version >= _TARGET_SCHEMA_VERSION:
        _ensure_vault_schema(conn)
        conn.commit()
        _migrations_done = True
        return
    migrations = [
        # version 1: cron scheduler enhancements
        [
            "ALTER TABLE scheduled_jobs ADD COLUMN timezone TEXT DEFAULT ''",
            "ALTER TABLE scheduled_jobs ADD COLUMN cron_expr TEXT DEFAULT ''",
            "ALTER TABLE scheduled_jobs ADD COLUMN webhook_url TEXT DEFAULT ''",
            "ALTER TABLE scheduled_jobs ADD COLUMN retry_count INTEGER DEFAULT 0",
            "ALTER TABLE scheduled_jobs ADD COLUMN max_retries INTEGER DEFAULT 2",
            "ALTER TABLE scheduled_jobs ADD COLUMN last_error TEXT DEFAULT ''",
            "ALTER TABLE scheduled_jobs ADD COLUMN model_override TEXT DEFAULT ''",
        ],
        # version 2: thread context anchors (model-only compressed context)
        [
            """CREATE TABLE IF NOT EXISTS thread_context_anchors (
                thread_id TEXT PRIMARY KEY,
                through_message_id INTEGER NOT NULL,
                summary_ciphertext BLOB NOT NULL,
                updated_at TEXT NOT NULL
            )""",
        ],
        # version 3: idle tick metadata on messages
        [
            "ALTER TABLE messages ADD COLUMN is_idle_tick INTEGER DEFAULT 0",
        ],
        # version 4: structured task plans (one JSON blob per thread, encrypted)
        [
            """CREATE TABLE IF NOT EXISTS thread_task_plans (
                thread_id TEXT PRIMARY KEY,
                plan_ciphertext BLOB NOT NULL,
                updated_at TEXT NOT NULL
            )""",
        ],
        # version 5: Pattern Inbox / Vault notes, projects, attachments, and overlays
        [
            """CREATE TABLE IF NOT EXISTS capture_projects (
                id TEXT PRIMARY KEY,
                name_ciphertext BLOB NOT NULL,
                description_ciphertext BLOB NOT NULL,
                name_key TEXT NOT NULL UNIQUE,
                sort_order INTEGER NOT NULL DEFAULT 0,
                metadata_json TEXT NOT NULL DEFAULT '{}',
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                archived_at TEXT NOT NULL DEFAULT ''
            )""",
            """CREATE TABLE IF NOT EXISTS captures (
                id TEXT PRIMARY KEY,
                body_ciphertext BLOB NOT NULL,
                title_ciphertext BLOB NOT NULL,
                reason_ciphertext BLOB NOT NULL,
                normalized_ciphertext BLOB NOT NULL DEFAULT X'',
                body_hash TEXT NOT NULL,
                source TEXT NOT NULL DEFAULT 'quick_panel',
                thread_id TEXT NOT NULL DEFAULT '',
                status TEXT NOT NULL DEFAULT 'inbox',
                kind TEXT NOT NULL DEFAULT 'raw',
                tags_json TEXT NOT NULL DEFAULT '[]',
                attachment_refs_json TEXT NOT NULL DEFAULT '[]',
                metadata_json TEXT NOT NULL DEFAULT '{}',
                project_id TEXT NOT NULL DEFAULT '',
                section TEXT NOT NULL DEFAULT '',
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                deleted_at TEXT NOT NULL DEFAULT ''
            )""",
            """CREATE TABLE IF NOT EXISTS capture_connections (
                id TEXT PRIMARY KEY,
                summary_ciphertext BLOB NOT NULL,
                why_ciphertext BLOB NOT NULL,
                source_ids_json TEXT NOT NULL DEFAULT '[]',
                status TEXT NOT NULL DEFAULT 'suggested',
                scope TEXT NOT NULL DEFAULT 'theme',
                project_id TEXT NOT NULL DEFAULT '',
                feedback TEXT NOT NULL DEFAULT '',
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            )""",
            """CREATE TABLE IF NOT EXISTS action_cards (
                id TEXT PRIMARY KEY,
                title_ciphertext BLOB NOT NULL,
                body_ciphertext BLOB NOT NULL,
                action_type TEXT NOT NULL DEFAULT 'explain',
                trust_level TEXT NOT NULL DEFAULT 'informational',
                state TEXT NOT NULL DEFAULT 'suggested',
                source_ids_json TEXT NOT NULL DEFAULT '[]',
                project_id TEXT NOT NULL DEFAULT '',
                failed_reason TEXT NOT NULL DEFAULT '',
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            )""",
            """CREATE TABLE IF NOT EXISTS vault_attachments (
                id TEXT PRIMARY KEY,
                note_id TEXT NOT NULL DEFAULT '',
                project_id TEXT NOT NULL DEFAULT '',
                mime_type TEXT NOT NULL,
                byte_size INTEGER NOT NULL,
                storage_path TEXT NOT NULL,
                content_hash TEXT NOT NULL,
                metadata_json TEXT NOT NULL DEFAULT '{}',
                created_at TEXT NOT NULL
            )""",
            """CREATE TABLE IF NOT EXISTS note_links (
                src_id TEXT NOT NULL,
                dst_id TEXT NOT NULL,
                link_kind TEXT NOT NULL DEFAULT 'wikilink',
                context TEXT NOT NULL DEFAULT '',
                created_at TEXT NOT NULL,
                PRIMARY KEY (src_id, dst_id, link_kind)
            )""",
            "CREATE INDEX IF NOT EXISTS idx_capture_projects_sort ON capture_projects(archived_at, sort_order, created_at)",
            "CREATE INDEX IF NOT EXISTS idx_captures_project ON captures(project_id, deleted_at, updated_at)",
            "CREATE INDEX IF NOT EXISTS idx_captures_status ON captures(status, deleted_at, updated_at)",
            "CREATE INDEX IF NOT EXISTS idx_captures_hash ON captures(body_hash, deleted_at)",
            "CREATE INDEX IF NOT EXISTS idx_capture_connections_project ON capture_connections(project_id, status, created_at)",
            "CREATE INDEX IF NOT EXISTS idx_action_cards_project ON action_cards(project_id, state, created_at)",
            "CREATE INDEX IF NOT EXISTS idx_vault_attachments_note ON vault_attachments(note_id, created_at)",
        ],
        # version 6: vault knowledge graph (suggested edges + per-note embeddings)
        [
            """CREATE TABLE IF NOT EXISTS vault_graph_edges (
                id TEXT PRIMARY KEY,
                note_low TEXT NOT NULL,
                note_high TEXT NOT NULL,
                src_id TEXT NOT NULL,
                dst_id TEXT NOT NULL,
                link_kind TEXT NOT NULL,
                status TEXT NOT NULL DEFAULT 'suggested',
                confidence REAL,
                source TEXT NOT NULL DEFAULT 'system',
                rationale TEXT NOT NULL DEFAULT '',
                project_id TEXT NOT NULL DEFAULT '',
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            )""",
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_vault_graph_edges_canonical ON vault_graph_edges(note_low, note_high, link_kind)",
            "CREATE INDEX IF NOT EXISTS idx_vault_graph_edges_src ON vault_graph_edges(src_id)",
            "CREATE INDEX IF NOT EXISTS idx_vault_graph_edges_dst ON vault_graph_edges(dst_id)",
            "CREATE INDEX IF NOT EXISTS idx_vault_graph_edges_status ON vault_graph_edges(status, project_id)",
            """CREATE TABLE IF NOT EXISTS capture_embeddings (
                note_id TEXT PRIMARY KEY,
                content_hash TEXT NOT NULL,
                model_id TEXT NOT NULL,
                dim INTEGER NOT NULL,
                embedding BLOB NOT NULL,
                updated_at TEXT NOT NULL
            )""",
        ],
        # version 7: cron -> kanban hook (Plan B). Optional fields on
        # scheduled_jobs that re-route the cron firing into a kanban task
        # instead of an inline run_subagent. Backward compatible: jobs
        # without these fields keep the legacy inline path.
        [
            "ALTER TABLE scheduled_jobs ADD COLUMN kanban_assignee TEXT NOT NULL DEFAULT ''",
            "ALTER TABLE scheduled_jobs ADD COLUMN kanban_tenant TEXT NOT NULL DEFAULT 'default'",
            "ALTER TABLE scheduled_jobs ADD COLUMN kanban_metadata_json TEXT NOT NULL DEFAULT '{}'",
        ],
        # version 8: kanban coordinator mode (origin_thread index + active batch pointer)
        [
            "CREATE INDEX IF NOT EXISTS idx_kanban_tasks_origin_thread ON kanban_tasks(origin_thread_id)",
            """CREATE TABLE IF NOT EXISTS thread_coordinator_state (
                thread_id TEXT PRIMARY KEY,
                batch_id TEXT NOT NULL,
                activated_at TEXT NOT NULL
            )""",
        ],
        # version 9: council mode active batch pointer
        [
            """CREATE TABLE IF NOT EXISTS thread_council_state (
                thread_id TEXT PRIMARY KEY,
                batch_id TEXT NOT NULL,
                activated_at TEXT NOT NULL
            )""",
        ],
        # version 10: council pending Approve/Skip offer (Slice 2)
        [
            """CREATE TABLE IF NOT EXISTS thread_council_pending_offer (
                thread_id TEXT PRIMARY KEY,
                offer_id TEXT NOT NULL,
                council_frame TEXT NOT NULL,
                frame_hash TEXT NOT NULL,
                rationale TEXT NOT NULL DEFAULT '',
                created_at TEXT NOT NULL
            )""",
        ],
    ]
    for i, stmts in enumerate(migrations):
        if version <= i:
            step_failed = False
            for stmt in stmts:
                try:
                    conn.execute(stmt)
                except sqlite3.OperationalError as op_exc:
                    msg = str(op_exc).lower()
                    if "duplicate column" in msg or "already exists" in msg:
                        continue
                    logging.getLogger(__name__).error(
                        "chat_storage migration %d failed on statement %r: %s",
                        i + 1,
                        stmt,
                        op_exc,
                        exc_info=True,
                    )
                    step_failed = True
                    break
                except Exception as exc:
                    logging.getLogger(__name__).error(
                        "chat_storage migration %d raised on statement %r: %s",
                        i + 1,
                        stmt,
                        exc,
                        exc_info=True,
                    )
                    step_failed = True
                    break
            if step_failed:
                try:
                    conn.rollback()
                except Exception:
                    pass
                return
            conn.execute(f"PRAGMA user_version = {i + 1}")
            conn.commit()
    _migrations_done = True


def _messages_has_idle_tick(conn: sqlite3.Connection) -> bool:
    try:
        cols = conn.execute("PRAGMA table_info(messages)").fetchall()
        return any((c[1] if len(c) > 1 else "") == "is_idle_tick" for c in cols)
    except Exception:
        return False


def _plaintext_messages_for_search(
    messages: list[dict[str, Any]],
    *,
    row_ids: list[int] | None = None,
    created_ats: list[str] | None = None,
) -> list[dict[str, Any]]:
    """Normalize persisted messages into plaintext payloads for search indexers."""
    out: list[dict[str, Any]] = []
    row_ids = row_ids or []
    created_ats = created_ats or []
    for i, msg in enumerate(messages or []):
        role = (msg.get("role") or "user").strip() or "user"
        content = msg.get("content")
        if content is None:
            content = ""
        elif isinstance(content, list):
            text_parts = [
                (p.get("text", "") if isinstance(p, dict) and p.get("type") == "text" else (p if isinstance(p, str) else ""))
                for p in content
            ]
            content = " ".join(text_parts)
        else:
            content = str(content)
        payload = {"role": role, "content": str(content) if content else ""}
        if i < len(row_ids):
            payload["_row_id"] = row_ids[i]
        if i < len(created_ats):
            payload["created_at"] = created_ats[i]
        out.append(payload)
    return out


# ---------- Messages (chats) ----------


def _format_display_timestamp(created_at: str | None) -> str:
    """Format created_at (ISO UTC) as display string in local time.

    Today's messages: '2:30 PM'
    This year's messages: 'Mar 25, 2:30 PM'
    Older messages: 'Mar 25 2025, 2:30 PM'
    """
    if not created_at or not isinstance(created_at, str):
        return ""
    try:
        s = created_at.replace("Z", "+00:00")
        dt = datetime.fromisoformat(s)
        from datetime import timezone

        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        local = dt.astimezone(tz=None)
        now_local = datetime.now().astimezone(tz=None)
        time_str = local.strftime("%I:%M %p").lstrip("0")
        if local.date() == now_local.date():
            return time_str
        if local.year == now_local.year:
            return local.strftime("%b %-d, ") + time_str
        return local.strftime("%b %-d %Y, ") + time_str
    except (ValueError, TypeError):
        return ""


def get_messages(thread_id: str, *, include_row_id: bool = False) -> list[dict[str, Any]]:
    """Load messages for a thread.

    Returns list of {"role", "content", "timestamp"}.
    When *include_row_id* is True, each dict also has "_row_id" (messages.id) for
    anchor computation — callers that build model context need this.
    """
    if not thread_id or not isinstance(thread_id, str):
        thread_id = "default"
    conn = _get_connection()
    try:
        has_idle_tick = _messages_has_idle_tick(conn)
        if include_row_id:
            if has_idle_tick:
                rows = conn.execute(
                    "SELECT id, role, content, created_at, is_idle_tick FROM messages WHERE thread_id = ? ORDER BY id ASC",
                    (thread_id,),
                ).fetchall()
            else:
                rows = conn.execute(
                    "SELECT id, role, content, created_at FROM messages WHERE thread_id = ? ORDER BY id ASC",
                    (thread_id,),
                ).fetchall()
            out = []
            for row in rows:
                if has_idle_tick:
                    row_id, role, content_blob, created_at, idle_flag = row
                else:
                    row_id, role, content_blob, created_at = row
                    idle_flag = 0
                raw = _decrypt(content_blob) if content_blob else ""
                if isinstance(raw, str) and raw.strip().startswith("["):
                    try:
                        content = json.loads(raw)
                        if not isinstance(content, list):
                            content = raw
                    except (json.JSONDecodeError, TypeError):
                        content = raw
                else:
                    content = raw
                msg = {
                    "role": role,
                    "content": content,
                    "timestamp": _format_display_timestamp(created_at),
                    "_row_id": row_id,
                }
                if idle_flag:
                    msg["_idle_tick"] = True
                out.append(msg)
            return out
        else:
            if has_idle_tick:
                rows = conn.execute(
                    "SELECT role, content, created_at, is_idle_tick FROM messages WHERE thread_id = ? ORDER BY id ASC",
                    (thread_id,),
                ).fetchall()
            else:
                rows = conn.execute(
                    "SELECT role, content, created_at FROM messages WHERE thread_id = ? ORDER BY id ASC",
                    (thread_id,),
                ).fetchall()
            out = []
            for row in rows:
                if has_idle_tick:
                    role, content_blob, created_at, idle_flag = row
                else:
                    role, content_blob, created_at = row
                    idle_flag = 0
                raw = _decrypt(content_blob) if content_blob else ""
                if isinstance(raw, str) and raw.strip().startswith("["):
                    try:
                        content = json.loads(raw)
                        if not isinstance(content, list):
                            content = raw
                    except (json.JSONDecodeError, TypeError):
                        content = raw
                else:
                    content = raw
                msg = {
                    "role": role,
                    "content": content,
                    "timestamp": _format_display_timestamp(created_at),
                }
                if idle_flag:
                    msg["_idle_tick"] = True
                out.append(msg)
            return out
    finally:
        conn.close()


def get_last_real_user_message_age_seconds(thread_id: str) -> float | None:
    """Seconds since the most recent **user** row that is not an idle-tick injection.

    Used to suppress idle continuity ticks while a human-driven thread is still "hot".
    Returns ``None`` when there is no qualifying row or timestamps cannot be parsed.
    """
    if not thread_id or not isinstance(thread_id, str):
        return None
    conn = _get_connection()
    try:
        has_idle_tick = _messages_has_idle_tick(conn)
        if has_idle_tick:
            row = conn.execute(
                """
                SELECT created_at FROM messages
                WHERE thread_id = ? AND LOWER(role) = 'user' AND COALESCE(is_idle_tick, 0) = 0
                ORDER BY id DESC LIMIT 1
                """,
                (thread_id,),
            ).fetchone()
        else:
            row = conn.execute(
                """
                SELECT created_at FROM messages
                WHERE thread_id = ? AND LOWER(role) = 'user'
                ORDER BY id DESC LIMIT 1
                """,
                (thread_id,),
            ).fetchone()
    finally:
        conn.close()
    if not row or not row[0]:
        return None
    created_at = row[0]
    if not isinstance(created_at, str) or not created_at.strip():
        return None
    try:
        s = created_at.replace("Z", "+00:00")
        dt = datetime.fromisoformat(s)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        delta = datetime.now(timezone.utc) - dt
        return max(0.0, delta.total_seconds())
    except (ValueError, TypeError, OSError):
        return None


def append_messages(thread_id: str, new_messages: list[dict[str, Any]]) -> int | None:
    """Append messages to a thread. Each item: {"role": "user"|"assistant", "content": str}.

    Returns the ``messages.id`` of the last inserted row, or None if nothing was inserted.

    ``btw_`` prefixed threads are ephemeral side-questions (see utils/btw.py) and
    are intentionally NOT persisted to the chat store. This is a defensive guard;
    callers should not be sending btw turns through append_messages in the first
    place.
    """
    if not thread_id or not isinstance(thread_id, str):
        thread_id = "default"
    if isinstance(thread_id, str) and thread_id.startswith("btw_"):
        return None
    if not new_messages:
        return None
    last_id: int | None = None
    inserted_row_ids: list[int] = []
    inserted_created_ats: list[str] = []
    conn = _get_connection()
    try:
        has_idle_tick = _messages_has_idle_tick(conn)
        now = datetime.utcnow().isoformat() + "Z"
        for msg in new_messages:
            role = (msg.get("role") or "user").strip() or "user"
            content = msg.get("content")
            if content is None:
                content = ""
            if isinstance(content, list):
                # Store multipart as JSON. Keep http(s) and data: (base64) image URLs.
                # Base64 survives app restart; localhost URLs do not (temp files are purged).
                storable = []
                for b in content:
                    if isinstance(b, dict):
                        if b.get("type") == "image_url" or "image_url" in b:
                            url = (b.get("image_url") or {}).get("url") or ""
                            if isinstance(url, str) and url.strip():
                                lower = url.strip().lower()
                                if lower.startswith(("http://", "https://")) or lower.startswith("data:"):
                                    storable.append({"type": "image_url", "image_url": {"url": url}})
                                else:
                                    storable.append({"type": "image_url", "image_url": {"url": ""}})
                            else:
                                storable.append({"type": "image_url", "image_url": {"url": ""}})
                        else:
                            storable.append(dict(b))
                    else:
                        storable.append(b)
                content = json.dumps(storable)
            else:
                content = str(content) if content is not None else ""
            encrypted = _encrypt(content)
            is_idle = 1 if msg.get("_idle_tick") else 0
            if has_idle_tick:
                cursor = conn.execute(
                    "INSERT INTO messages (thread_id, role, content, created_at, is_idle_tick) VALUES (?, ?, ?, ?, ?)",
                    (thread_id, role, encrypted, now, is_idle),
                )
            else:
                cursor = conn.execute(
                    "INSERT INTO messages (thread_id, role, content, created_at) VALUES (?, ?, ?, ?)",
                    (thread_id, role, encrypted, now),
                )
            last_id = cursor.lastrowid
            if last_id is not None:
                inserted_row_ids.append(int(last_id))
                inserted_created_ats.append(now)
        conn.commit()
    finally:
        conn.close()

    # Index into FTS5 search database (best-effort; non-critical path).
    # Note: this indexes the exact assistant/user content passed to append_messages,
    # so FTS/search alignment depends on callers persisting already-finalized text.
    try:
        from session_search_db import index_messages as _index_search
        _plaintext_msgs = _plaintext_messages_for_search(
            new_messages,
            row_ids=inserted_row_ids,
            created_ats=inserted_created_ats,
        )
        _index_search(thread_id, _plaintext_msgs)
    except Exception:
        pass

    return last_id


def replace_messages(thread_id: str, new_messages: list[dict]) -> None:
    """Atomically replace all messages in a thread (used by compaction).

    Uses BEGIN IMMEDIATE for exclusive write lock, ensuring the DELETE + INSERT
    sequence is atomic.  Also clears any thread context anchor (row ids become
    meaningless after a full replace).  FTS entries are cleared and re-indexed afterwards.
    """
    if not thread_id:
        thread_id = "default"
    conn = _get_connection()
    inserted_row_ids: list[int] = []
    inserted_created_ats: list[str] = []
    try:
        conn.execute("BEGIN IMMEDIATE")
        conn.execute("DELETE FROM messages WHERE thread_id = ?", (thread_id,))
        conn.execute("DELETE FROM thread_context_anchors WHERE thread_id = ?", (thread_id,))
        now = datetime.utcnow().isoformat() + "Z"
        for msg in new_messages:
            role = (msg.get("role") or "user").strip() or "user"
            content = msg.get("content")
            if content is None:
                content = ""
            elif isinstance(content, list):
                content = json.dumps(content)
            else:
                content = str(content)
            encrypted = _encrypt(content)
            cursor = conn.execute(
                "INSERT INTO messages (thread_id, role, content, created_at) VALUES (?, ?, ?, ?)",
                (thread_id, role, encrypted, now),
            )
            row_id = cursor.lastrowid
            if row_id is not None:
                inserted_row_ids.append(int(row_id))
                inserted_created_ats.append(now)
        conn.execute("COMMIT")
    except Exception:
        try:
            conn.execute("ROLLBACK")
        except Exception:
            pass
        raise
    finally:
        conn.close()

    # Re-index search indexes (FTS + vectors): clear old entries first.
    try:
        from session_search_db import clear_thread_index, index_messages as _index_search

        clear_thread_index(thread_id, clear_fts=True, clear_vectors=True)
        plaintext_msgs = _plaintext_messages_for_search(
            new_messages,
            row_ids=inserted_row_ids,
            created_ats=inserted_created_ats,
        )
        _index_search(thread_id, plaintext_msgs)
    except Exception:
        pass


def remove_last_message(thread_id: str) -> bool:
    """Remove the most recent message in the thread. Returns True if a row was deleted.

    Also clears any thread context anchor for safety — the anchor's
    ``through_message_id`` may reference a row that no longer exists.
    """
    if not thread_id or not isinstance(thread_id, str):
        thread_id = "default"
    conn = _get_connection()
    try:
        cursor = conn.execute(
            "DELETE FROM messages WHERE thread_id = ? AND id = (SELECT MAX(id) FROM messages WHERE thread_id = ?)",
            (thread_id, thread_id),
        )
        if cursor.rowcount > 0:
            conn.execute(
                "DELETE FROM thread_context_anchors WHERE thread_id = ?",
                (thread_id,),
            )
        conn.commit()
        return cursor.rowcount > 0
    finally:
        conn.close()


# ---------- Thread context anchors ----------


import logging as _logging

_anchor_log = _logging.getLogger(__name__)


def get_anchor(thread_id: str) -> dict[str, Any] | None:
    """Return the anchor for *thread_id*, or None.

    Result: ``{"through_message_id": int, "summary": str, "updated_at": str}``.
    If the stored blob cannot be decrypted (key mismatch / corruption), the row
    is deleted and None is returned so callers fall back to full literal history.
    """
    if not thread_id:
        return None
    conn = _get_connection()
    try:
        row = conn.execute(
            "SELECT through_message_id, summary_ciphertext, updated_at "
            "FROM thread_context_anchors WHERE thread_id = ?",
            (thread_id,),
        ).fetchone()
        if not row:
            return None
        through_id, blob, updated_at = row
        try:
            summary = _decrypt(blob)
        except Exception:
            _anchor_log.warning(
                "anchor decrypt failed for thread %s – deleting corrupt row",
                thread_id,
            )
            conn.execute(
                "DELETE FROM thread_context_anchors WHERE thread_id = ?",
                (thread_id,),
            )
            conn.commit()
            return None
        return {
            "through_message_id": through_id,
            "summary": summary,
            "updated_at": updated_at,
        }
    finally:
        conn.close()


def upsert_anchor(
    thread_id: str, through_message_id: int, summary_plaintext: str
) -> None:
    """Insert or update the context anchor for *thread_id*."""
    if not thread_id:
        return
    conn = _get_connection()
    try:
        now = datetime.utcnow().isoformat() + "Z"
        blob = _encrypt(summary_plaintext)
        conn.execute(
            "INSERT INTO thread_context_anchors "
            "(thread_id, through_message_id, summary_ciphertext, updated_at) "
            "VALUES (?, ?, ?, ?) "
            "ON CONFLICT(thread_id) DO UPDATE SET "
            "through_message_id = excluded.through_message_id, "
            "summary_ciphertext = excluded.summary_ciphertext, "
            "updated_at = excluded.updated_at",
            (thread_id, through_message_id, blob, now),
        )
        conn.commit()
        _anchor_log.debug(
            "upsert anchor thread=%s through_id=%d len=%d",
            thread_id, through_message_id, len(summary_plaintext),
        )
    finally:
        conn.close()


def delete_anchor(thread_id: str) -> None:
    """Remove the context anchor for *thread_id* (idempotent)."""
    if not thread_id:
        return
    conn = _get_connection()
    try:
        conn.execute(
            "DELETE FROM thread_context_anchors WHERE thread_id = ?",
            (thread_id,),
        )
        conn.commit()
    finally:
        conn.close()


# ---------- Thread branches ----------


def _content_to_str(content: Any) -> str:
    """Extract displayable text from message content (str or list of parts). Matches Swift filtering."""
    if content is None:
        return ""
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = []
        for b in content:
            if isinstance(b, dict):
                if b.get("type") == "text":
                    parts.append((b.get("text") or "").strip())
                elif b.get("type") == "image_url" or "image_url" in b:
                    parts.append("[Image]")
                else:
                    parts.append((b.get("text", str(b)) or "").strip())
            else:
                parts.append(str(b))
        return " ".join(parts)
    return str(content)


def _is_displayable_message(msg: dict[str, Any]) -> bool:
    """Match Swift ChatViewModel filtering: role user|assistant, non-empty, not init."""
    role = (msg.get("role") or "").strip()
    if role not in ("user", "assistant"):
        return False
    content_str = _content_to_str(msg.get("content")).strip()
    if not content_str:
        return False
    if content_str.strip().lower() == "init":
        return False
    return True


def create_branch(parent_thread_id: str, branch_at_message_index: int) -> str:
    """
    Create a new branch thread by copying messages from parent up to (and including)
    the message at branch_at_message_index. Uses the same message filtering as Swift
    (role user|assistant, exclude empty/init). The message at the index must be assistant.
    Returns the new branch_thread_id. Raises ValueError for invalid args.
    """
    if not parent_thread_id or not isinstance(parent_thread_id, str):
        raise ValueError("parent_thread_id required")
    if not isinstance(branch_at_message_index, int) or branch_at_message_index < 0:
        raise ValueError("branch_at_message_index must be non-negative integer")

    raw_messages = get_messages(parent_thread_id) or []
    displayable: list[dict[str, Any]] = []
    for m in raw_messages:
        if _is_displayable_message(m):
            displayable.append(m)

    if branch_at_message_index >= len(displayable):
        raise ValueError(
            f"branch_at_message_index {branch_at_message_index} out of range (0..{len(displayable) - 1})"
        )
    if (displayable[branch_at_message_index].get("role") or "").strip() != "assistant":
        raise ValueError("Branch point must be an assistant message")

    to_copy = displayable[: branch_at_message_index + 1]
    branch_thread_id = str(uuid.uuid4())

    append_messages(branch_thread_id, to_copy)

    conn = _get_connection()
    try:
        conn.execute(
            "INSERT INTO thread_branches (branch_thread_id, parent_thread_id, branch_at_message_index) VALUES (?, ?, ?)",
            (branch_thread_id, parent_thread_id, branch_at_message_index),
        )
        conn.commit()
    finally:
        conn.close()

    return branch_thread_id


def get_branches_of(parent_thread_id: str) -> list[dict[str, Any]]:
    """Return branches of a root thread. Each dict has branch_thread_id, branch_at_message_index."""
    if not parent_thread_id or not isinstance(parent_thread_id, str):
        return []
    conn = _get_connection()
    try:
        rows = conn.execute(
            "SELECT branch_thread_id, branch_at_message_index FROM thread_branches WHERE parent_thread_id = ? ORDER BY branch_at_message_index ASC",
            (parent_thread_id,),
        ).fetchall()
        return [
            {"branch_thread_id": bid, "branch_at_message_index": idx}
            for bid, idx in rows
        ]
    finally:
        conn.close()


def _repair_min_iso(existing: str | None, candidate: str | None) -> str | None:
    """Return the earlier of two ISO-ish timestamps (lexicographic on normalized Z suffices)."""
    if not candidate or not isinstance(candidate, str):
        return existing
    c = candidate.strip()
    if not c:
        return existing
    if not existing or not isinstance(existing, str) or not existing.strip():
        return c
    e = existing.strip()
    ec = e.replace("Z", "+00:00")
    cc = c.replace("Z", "+00:00")
    try:
        return c if datetime.fromisoformat(cc) < datetime.fromisoformat(ec) else e
    except (ValueError, TypeError):
        return c if c < e else e


def repair_conversation_meta_from_message_threads(
    conversation_meta: dict[str, Any],
) -> tuple[int, int, list[str]]:
    """Ensure every non-internal thread_id with stored messages has a ``conversation_meta`` row.

    Sidebar rows are persisted in ``chats_db.json`` while message bodies live in SQLite. If
    ``conversation_meta`` was truncated or wiped while ``messages`` rows remain, the Swift
    client shows an empty sidebar. This repopulates minimal metadata (``created_at`` from the
    earliest message row). Titles are set to a neutral ``Untitled`` placeholder; the backend
    is expected to regenerate real titles via the utility model (see
    ``HeadlessDelegate._schedule_batch_conversation_title_utility_refresh``).

    Only **main sidebar** threads are considered (mesh/subagent/btw/branch sessions are
    skipped via ``utils.conversation_sidebar`` + ``thread_branches``).

    Returns ``(inserted_count, retitled_recovered_chat_count, affected_thread_ids)``.
    """
    if not isinstance(conversation_meta, dict):
        return (0, 0, [])

    from utils.conversation_sidebar import thread_id_allowed_in_main_conversation_sidebar

    conn = _get_connection()
    try:
        branch_rows = conn.execute("SELECT branch_thread_id FROM thread_branches").fetchall()
        branch_ids = {str(r[0]).strip() for r in branch_rows if r and r[0]}
        has_idle_tick = _messages_has_idle_tick(conn)
        if has_idle_tick:
            rows = conn.execute(
                """
                SELECT thread_id, role, content, created_at, COALESCE(is_idle_tick, 0)
                FROM messages
                WHERE LENGTH(COALESCE(thread_id, '')) > 0
                ORDER BY thread_id ASC, id ASC
                """
            ).fetchall()
        else:
            rows = conn.execute(
                """
                SELECT thread_id, role, content, created_at, 0
                FROM messages
                WHERE LENGTH(COALESCE(thread_id, '')) > 0
                ORDER BY thread_id ASC, id ASC
                """
            ).fetchall()
    finally:
        conn.close()

    per_tid_earliest: dict[str, str | None] = {}
    branch_skipped = 0
    sidebar_skipped = 0
    for row in rows:
        tid_raw, _role, _blob, created_at, _idle_flag = row[0], row[1], row[2], row[3], row[4]
        tid = str(tid_raw or "").strip()
        if not tid:
            continue
        if tid in branch_ids:
            branch_skipped += 1
            continue
        if not thread_id_allowed_in_main_conversation_sidebar(tid):
            sidebar_skipped += 1
            continue
        ca = created_at if isinstance(created_at, str) else None
        per_tid_earliest[tid] = _repair_min_iso(per_tid_earliest.get(tid), ca)

    placeholder = "Untitled"
    inserted = 0
    retitled = 0
    affected: list[str] = []

    for tid, earliest in per_tid_earliest.items():
        if tid not in conversation_meta:
            conversation_meta[tid] = {
                "title": placeholder,
                "created_at": earliest or datetime.now(timezone.utc).isoformat(),
                "saved_path": None,
                "folder_id": None,
                "pinned": False,
                "archived": False,
                "trashed": False,
                "sort_order": 0,
            }
            inserted += 1
            affected.append(tid)
            continue

        meta = conversation_meta[tid]
        prev_title = (meta.get("title") or "").strip()
        if prev_title == "Recovered chat":
            meta["title"] = placeholder
            meta.pop("title_utility_milestone", None)
            meta["created_at"] = (
                _repair_min_iso(meta.get("created_at") if isinstance(meta.get("created_at"), str) else None, earliest)
                or earliest
                or meta.get("created_at")
            )
            retitled += 1
            affected.append(tid)

    logging.getLogger(__name__).debug(
        "Meta repair: threads=%d inserted=%d retitled=%d affected=%d",
        len(per_tid_earliest), inserted, retitled, len(affected),
    )

    return inserted, retitled, affected


def is_branch(thread_id: str) -> bool:
    """Return True if thread_id is a branch (exists in thread_branches)."""
    if not thread_id or not isinstance(thread_id, str):
        return False
    conn = _get_connection()
    try:
        row = conn.execute(
            "SELECT 1 FROM thread_branches WHERE branch_thread_id = ? LIMIT 1",
            (thread_id,),
        ).fetchone()
        return row is not None
    finally:
        conn.close()


def delete_branch_entry(branch_thread_id: str) -> None:
    """Remove branch metadata for a branch thread. Call when deleting a branch."""
    if not branch_thread_id:
        return
    conn = _get_connection()
    try:
        conn.execute("DELETE FROM thread_branches WHERE branch_thread_id = ?", (branch_thread_id,))
        conn.commit()
    finally:
        conn.close()


def purge_thread_messages(thread_id: str) -> int:
    """Delete every message + anchor + branch row + search index entry for *thread_id*.

    Single-statement bulk delete inside one transaction, so emptying a thread with
    hundreds of messages is O(1) round-trips instead of the O(n^2) loop you get
    from calling :func:`remove_last_message` until ``get_messages`` returns empty.
    Returns the number of message rows deleted.
    """
    if not thread_id or not isinstance(thread_id, str):
        return 0
    conn = _get_connection()
    deleted = 0
    try:
        conn.execute("BEGIN IMMEDIATE")
        cursor = conn.execute("DELETE FROM messages WHERE thread_id = ?", (thread_id,))
        deleted = int(cursor.rowcount or 0)
        conn.execute("DELETE FROM thread_context_anchors WHERE thread_id = ?", (thread_id,))
        conn.execute("DELETE FROM thread_branches WHERE branch_thread_id = ?", (thread_id,))
        conn.execute("COMMIT")
    except Exception:
        try:
            conn.execute("ROLLBACK")
        except Exception:
            pass
        raise
    finally:
        conn.close()
    try:
        from session_search_db import clear_thread_index
        clear_thread_index(thread_id, clear_fts=True, clear_vectors=True)
    except Exception:
        pass
    return deleted


def get_parent_of_branch(branch_thread_id: str) -> str | None:
    """Return parent_thread_id for a branch, or None if not a branch."""
    if not branch_thread_id:
        return None
    conn = _get_connection()
    try:
        row = conn.execute(
            "SELECT parent_thread_id FROM thread_branches WHERE branch_thread_id = ? LIMIT 1",
            (branch_thread_id,),
        ).fetchone()
        return row[0] if row else None
    finally:
        conn.close()


# ---------- Pattern Inbox / Vault notes ----------


_CAPTURE_TITLE_MAX = 96
_CAPTURE_SECTION_MAX = 160
_VAULT_GRAPH_EMBED_MODEL_ID = "utility_embed_v1"


def _vault_embed_dim() -> int:
    try:
        from utils.embedding_model import embedding_dim

        return max(8, embedding_dim())
    except Exception:
        return 256


def _vault_embed_model_id() -> str:
    try:
        from utility_model import utility_embedding_identity

        return utility_embedding_identity()
    except Exception:
        return _VAULT_GRAPH_EMBED_MODEL_ID
_VAULT_EMBED_TEXT_MAX = 12_000
_VAULT_SIMILARITY_MIN_DEFAULT = 0.68
_VAULT_NEIGHBOR_CAP_DEFAULT = 80
_VAULT_GRAPH_SCAN_CAP = 500
_VAULT_GRAPH_PROPOSE_PAIR_CAP = 400
_VAULT_FOLDER_MAX_SEGMENTS = 3
_VAULT_FOLDER_SEGMENT_MAX = 48
_ATTACHMENT_MAX_BYTES = 50 * 1024 * 1024
_ALLOWED_ATTACHMENT_MIME_PREFIXES = ("image/", "audio/", "video/", "text/")
_ALLOWED_ATTACHMENT_MIME_TYPES = {
    "application/pdf",
    "application/json",
    "application/zip",
    "application/octet-stream",
}


def _now_iso() -> str:
    return datetime.utcnow().isoformat() + "Z"


def _json_loads(value: str | None, default: Any) -> Any:
    if not value:
        return default
    try:
        parsed = json.loads(value)
        return parsed if parsed is not None else default
    except Exception:
        return default


def _json_dumps(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, separators=(",", ":"))


def _normalize_project_key(name: str) -> str:
    return " ".join(str(name or "").strip().lower().split())


def _normalize_tags(tags: Any) -> list[str]:
    if isinstance(tags, str):
        tags = [tags]
    if not isinstance(tags, list):
        return []
    out: list[str] = []
    for value in tags:
        tag = str(value or "").strip().lstrip("#")[:48]
        if tag and tag not in out:
            out.append(tag)
        if len(out) >= 24:
            break
    return out


def _normalize_attachment_refs(refs: Any) -> list[str]:
    if isinstance(refs, str):
        refs = [refs]
    if not isinstance(refs, list):
        return []
    out: list[str] = []
    for value in refs:
        ref = str(value or "").strip()
        if ref and ref not in out:
            out.append(ref)
        if len(out) >= 100:
            break
    return out


def _looks_like_uuid(value: str) -> bool:
    try:
        uuid.UUID(str(value).strip())
        return True
    except (ValueError, TypeError, AttributeError):
        return False


def _is_temp_paste_filename(name: str) -> bool:
    base = os.path.basename(str(name or "").strip())
    lower = base.lower()
    return lower.startswith("mugi_quick_paste_") or lower.startswith("mugi_paste_")


def _is_quick_panel_staging_filename(name: str) -> bool:
    """Auto-generated Quick Panel temp names (not user-chosen filenames)."""
    base = os.path.basename(str(name or "").strip())
    lower = base.lower()
    return _is_temp_paste_filename(name) or lower.startswith("mugi-quick-audio-") or lower.startswith("mugi-quick-")


def _vault_alt_and_default_title_for_mime(mime: str, original_filename: str) -> tuple[str, str]:
    """Markdown image alt text (Vault codec) + short capture title when body is inline vault markdown."""
    m = (mime or "").lower().split(";", 1)[0].strip()
    fn = os.path.basename(str(original_filename or "").strip())
    nice_fn = fn[:_CAPTURE_TITLE_MAX].rstrip() if fn else ""
    if m.startswith("image/"):
        if nice_fn and not _is_quick_panel_staging_filename(fn):
            return "Vault image", nice_fn
        return "Vault image", "Pasted image"
    if m.startswith("audio/"):
        if nice_fn and not _is_quick_panel_staging_filename(fn):
            return "Vault audio", nice_fn
        return "Vault audio", "Voice memo"
    if m.startswith("video/"):
        if nice_fn and not _is_quick_panel_staging_filename(fn):
            return "Vault video", nice_fn
        return "Vault video", "Video clip"
    if nice_fn and not _is_quick_panel_staging_filename(fn):
        return "Vault file", nice_fn
    return "Vault file", "Attachment"


def _vault_markdown_body_and_title_for_attachment_refs(refs: list[str]) -> tuple[str, str] | None:
    """One markdown line per attachment + title for attachment-only (placeholder) captures."""
    if not refs:
        return None
    lines: list[str] = []
    titles: list[str] = []
    for raw_id in refs:
        aid = str(raw_id).strip()
        if not aid:
            continue
        att = get_vault_attachment(aid)
        if not att:
            return None
        meta = att.get("metadata") if isinstance(att.get("metadata"), dict) else {}
        fn = str(meta.get("original_filename") or "").strip()
        mime = str(att.get("mime_type") or "")
        alt, t = _vault_alt_and_default_title_for_mime(mime, fn)
        lines.append(f"![{alt}](mugi-vault://{att['id']})")
        titles.append(t)
    if not lines:
        return None
    if len(lines) == 1:
        return lines[0], titles[0]
    return "\n\n".join(lines), f"Pasted files ({len(lines)})"


def _ingest_path_attachment_items(refs: Any) -> list[tuple[str, str]]:
    """Read Swift/client `{path, name, type}` items from disk into `vault_attachments`.

    Returns `(attachment_id, display_name)` per successful ingest. Skips missing paths
    and unsupported / oversize files.
    """
    if not isinstance(refs, list):
        return []
    out: list[tuple[str, str]] = []
    for item in refs:
        if not isinstance(item, dict):
            continue
        path = str(item.get("path") or "").strip()
        display = str(item.get("name") or "").strip() or (os.path.basename(path) if path else "")
        mime = str(item.get("type") or item.get("mime_type") or "").strip() or "application/octet-stream"
        if not path or not os.path.isfile(path):
            continue
        try:
            size = os.path.getsize(path)
        except OSError:
            continue
        if size > _ATTACHMENT_MAX_BYTES:
            continue
        try:
            with open(path, "rb") as f:
                data = f.read()
        except OSError:
            continue
        if not data:
            continue
        try:
            att = create_vault_attachment(
                data,
                mime_type=mime,
                filename=display or path,
                note_id="",
                project_id="",
                metadata={"source": "capture_path_ingest"},
            )
            aid = str(att.get("id") or "").strip()
            if aid:
                out.append((aid, display or os.path.basename(path) or "attachment"))
        except ValueError:
            continue
        if len(out) >= 100:
            break
    return out


def _derive_capture_title(body: str) -> str:
    text = " ".join(str(body or "").strip().split())
    if not text:
        return "Untitled"
    return text[:_CAPTURE_TITLE_MAX].rstrip()


def _normalize_capture_plaintext(text: str) -> str:
    """NFC-normalized body for `normalized_ciphertext` (stable form for storage constraints / future search)."""
    return unicodedata.normalize("NFC", str(text or "").strip())


def _extract_wikilinks(body: str) -> list[str]:
    links: list[str] = []
    for match in re.finditer(r"\[\[([^\]\n]{1,160})\]\]", body or ""):
        value = match.group(1).split("|", 1)[0].strip()
        if value and value not in links:
            links.append(value)
    return links[:200]


def _resolve_note_link(conn: sqlite3.Connection, raw_link: str) -> str | None:
    link = str(raw_link or "").strip()
    if not link:
        return None
    row = conn.execute("SELECT id FROM captures WHERE id = ? AND deleted_at = ''", (link,)).fetchone()
    if row:
        return str(row[0])
    rows = conn.execute(_capture_select_sql() + " WHERE deleted_at = '' ORDER BY updated_at DESC LIMIT 500").fetchall()
    wanted = " ".join(link.lower().split())
    for row in rows:
        cap = _capture_from_row(conn, row)
        if " ".join(str(cap.get("title") or "").lower().split()) == wanted:
            return str(cap.get("id") or "")
    return None


def _refresh_wikilinks(conn: sqlite3.Connection, note_id: str, body: str) -> None:
    now = _now_iso()
    conn.execute("DELETE FROM note_links WHERE src_id = ?", (note_id,))
    for raw in _extract_wikilinks(body):
        dst = _resolve_note_link(conn, raw)
        if not dst or dst == note_id:
            continue
        conn.execute(
            "INSERT OR IGNORE INTO note_links (src_id, dst_id, link_kind, context, created_at) VALUES (?, ?, 'wikilink', ?, ?)",
            (note_id, dst, raw, now),
        )


def _project_lookup(conn: sqlite3.Connection, project_id: str) -> dict[str, Any] | None:
    if not project_id:
        return None
    row = conn.execute(
        "SELECT id, name_ciphertext, description_ciphertext, sort_order, metadata_json, created_at, updated_at, archived_at FROM capture_projects WHERE id = ?",
        (project_id,),
    ).fetchone()
    if not row:
        return None
    return _project_from_row(row)


def _project_name(conn: sqlite3.Connection, project_id: str) -> str:
    project = _project_lookup(conn, project_id)
    return str(project.get("name") or "") if project else ""


def _project_from_row(row: tuple[Any, ...]) -> dict[str, Any]:
    pid, name_blob, desc_blob, sort_order, metadata_json, created_at, updated_at, archived_at = row
    return {
        "id": pid,
        "name": _decrypt(name_blob) if name_blob else "",
        "description": _decrypt(desc_blob) if desc_blob else "",
        "sort_order": int(sort_order or 0),
        "metadata": _json_loads(metadata_json, {}),
        "created_at": created_at or "",
        "updated_at": updated_at or "",
        "archived_at": archived_at or "",
    }


def _capture_from_row(conn: sqlite3.Connection, row: tuple[Any, ...]) -> dict[str, Any]:
    (
        cid,
        body_blob,
        title_blob,
        reason_blob,
        _body_hash,
        source,
        thread_id,
        status,
        kind,
        tags_json,
        attachment_refs_json,
        metadata_json,
        project_id,
        section,
        created_at,
        updated_at,
        deleted_at,
    ) = row
    metadata = _json_loads(metadata_json, {})
    effective_status = status or "inbox"
    if effective_status in {"processed", "promoted"}:
        metadata = dict(metadata)
        metadata.setdefault("processing_state", "processed" if effective_status == "processed" else "used_in_pattern")
        effective_status = "inbox"
    body = _decrypt(body_blob) if body_blob else ""
    title = _decrypt(title_blob) if title_blob else _derive_capture_title(body)
    return {
        "id": cid,
        "title": title,
        "body": body,
        "reason": _decrypt(reason_blob) if reason_blob else "",
        "source": source or "",
        "thread_id": thread_id or "",
        "status": effective_status,
        "kind": kind or "raw",
        "tags": _json_loads(tags_json, []),
        "attachment_refs": _json_loads(attachment_refs_json, []),
        "metadata": metadata,
        "project_id": project_id or "",
        "project_name": _project_name(conn, project_id or ""),
        "section": section or "",
        "created_at": created_at or "",
        "updated_at": updated_at or "",
        "deleted_at": deleted_at or "",
    }


def _capture_select_sql() -> str:
    return (
        "SELECT id, body_ciphertext, title_ciphertext, reason_ciphertext, body_hash, source, thread_id, "
        "status, kind, tags_json, attachment_refs_json, metadata_json, project_id, section, created_at, updated_at, deleted_at "
        "FROM captures"
    )


def create_capture_project(name: str, *, description: str = "", metadata: dict[str, Any] | None = None) -> dict[str, Any]:
    key = _normalize_project_key(name)
    if not key:
        raise ValueError("project name required")
    clean_name = " ".join(str(name or "").strip().split())[:120]
    description = str(description or "")[:4000]
    now = _now_iso()
    conn = _get_connection()
    try:
        dup = conn.execute(
            "SELECT id, name_ciphertext, description_ciphertext, sort_order, metadata_json, created_at, updated_at, archived_at FROM capture_projects WHERE name_key = ?",
            (key,),
        ).fetchone()
        if dup:
            project = _project_from_row(dup)
            project["duplicate"] = True
            return project
        next_sort = (conn.execute("SELECT COALESCE(MAX(sort_order), 0) + 1 FROM capture_projects").fetchone() or (1,))[0]
        pid = str(uuid.uuid4())
        conn.execute(
            """INSERT INTO capture_projects
               (id, name_ciphertext, description_ciphertext, name_key, sort_order, metadata_json, created_at, updated_at, archived_at)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, '')""",
            (pid, _encrypt(clean_name), _encrypt(description), key, int(next_sort or 1), _json_dumps(metadata or {}), now, now),
        )
        conn.commit()
        project = _project_lookup(conn, pid) or {}
        project["duplicate"] = False
        return project
    finally:
        conn.close()


def list_capture_projects(*, include_archived: bool = False, limit: int = 200) -> list[dict[str, Any]]:
    limit = max(1, min(int(limit or 200), 500))
    conn = _get_connection()
    try:
        where = "" if include_archived else "WHERE archived_at = ''"
        rows = conn.execute(
            f"SELECT id, name_ciphertext, description_ciphertext, sort_order, metadata_json, created_at, updated_at, archived_at FROM capture_projects {where} ORDER BY sort_order ASC, created_at DESC LIMIT ?",
            (limit,),
        ).fetchall()
        return [_project_from_row(row) for row in rows]
    finally:
        conn.close()


def reorder_capture_projects(project_ids: list[str]) -> None:
    """Assign ``sort_order`` from index for all **non-archived** projects.

    ``project_ids`` must list each active (``archived_at == ''``) project id **exactly once**.
    """
    ids = [str(x).strip() for x in (project_ids or []) if str(x).strip()]
    if not ids:
        raise ValueError("project_ids required")
    if len(ids) != len(set(ids)):
        raise ValueError("duplicate project id")
    conn = _get_connection()
    try:
        active = [
            str(r[0])
            for r in conn.execute(
                "SELECT id FROM capture_projects WHERE archived_at = '' ORDER BY sort_order ASC, created_at DESC"
            ).fetchall()
        ]
        if sorted(active) != sorted(ids):
            raise ValueError("project_ids must list each active project exactly once")
        now = _now_iso()
        for idx, pid in enumerate(ids):
            conn.execute(
                "UPDATE capture_projects SET sort_order = ?, updated_at = ? WHERE id = ? AND archived_at = ''",
                (idx, now, pid),
            )
        conn.commit()
    finally:
        conn.close()


def _normalize_vault_folder_segment(name: str) -> str:
    s = " ".join(str(name or "").strip().split())
    if not s:
        raise ValueError("folder name required")
    if "/" in s or "\x1e" in s or "\n" in s or "\r" in s:
        raise ValueError("invalid folder name")
    return s[:_VAULT_FOLDER_SEGMENT_MAX]


def add_capture_project_vault_folder(project_id: str, *, parent_path: str = "", name: str) -> dict[str, Any] | None:
    """Register an empty folder path under a capture project (metadata only).

    Folder paths are stored in project metadata under ``vault_folder_paths`` and
    mirrored into capture ``section`` when filing notes. Max depth: 3 segments
    (e.g. ``a/b/c``), matching ``_CAPTURE_SECTION_MAX``.
    """
    if not project_id:
        return None
    seg = _normalize_vault_folder_segment(name)
    parent = "/".join(x for x in str(parent_path or "").split("/") if x).strip("/")
    if parent:
        parent_depth = len(parent.split("/"))
        if parent_depth >= _VAULT_FOLDER_MAX_SEGMENTS:
            raise ValueError("folder depth limit is 3 levels")
        new_path = f"{parent}/{seg}"
    else:
        new_path = seg
    depth = len(new_path.split("/"))
    if depth > _VAULT_FOLDER_MAX_SEGMENTS:
        raise ValueError("folder depth limit is 3 levels")
    if len(new_path) > _CAPTURE_SECTION_MAX:
        raise ValueError("folder path too long")
    conn = _get_connection()
    try:
        project = _project_lookup(conn, project_id)
        if not project:
            return None
        meta = dict(project.get("metadata") or {})
        paths = list(meta.get("vault_folder_paths") or [])
        if not isinstance(paths, list):
            paths = []
        clean = [str(p).strip().strip("/") for p in paths if str(p).strip()]
        if new_path not in clean:
            clean.append(new_path)
            clean.sort()
        meta["vault_folder_paths"] = clean
        conn.execute(
            "UPDATE capture_projects SET metadata_json = ?, updated_at = ? WHERE id = ?",
            (_json_dumps(meta), _now_iso(), project_id),
        )
        conn.commit()
        return _project_lookup(conn, project_id)
    finally:
        conn.close()


def move_capture_project_vault_folder(
    project_id: str, *, from_path: str, to_parent_path: str = ""
) -> dict[str, Any] | None:
    """Move a vault folder (and descendants in metadata + capture sections) under a new parent within the same project.

    ``from_path`` and ``to_parent_path`` use slash segments (no leading/trailing slashes stored).
    Dropping on project root uses ``to_parent_path == ""``; the folder keeps its leaf segment name.
    """
    if not project_id:
        return None
    fp = "/".join(x for x in str(from_path or "").split("/") if x).strip("/")
    tp = "/".join(x for x in str(to_parent_path or "").split("/") if x).strip("/")
    if not fp:
        raise ValueError("from_path required")
    if tp == fp or (tp.startswith(fp + "/")):
        raise ValueError("cannot move folder into itself")
    parts = fp.split("/")
    leaf = parts[-1]
    if tp:
        new_path = f"{tp}/{leaf}"
    else:
        new_path = leaf
    if new_path == fp:
        conn = _get_connection()
        try:
            return _project_lookup(conn, project_id)
        finally:
            conn.close()

    def _depth(p: str) -> int:
        return len([x for x in p.split("/") if x])

    if _depth(new_path) > _VAULT_FOLDER_MAX_SEGMENTS:
        raise ValueError("folder depth limit is 3 levels")

    def _remap_path(p: str) -> str:
        if p == fp:
            return new_path
        if p.startswith(fp + "/"):
            return new_path + p[len(fp) :]
        return p

    conn = _get_connection()
    try:
        project = _project_lookup(conn, project_id)
        if not project:
            return None
        meta = dict(project.get("metadata") or {})
        paths = list(meta.get("vault_folder_paths") or [])
        if not isinstance(paths, list):
            paths = []
        clean = [str(p).strip().strip("/") for p in paths if str(p).strip()]
        affected = [p for p in clean if p == fp or p.startswith(fp + "/")]
        if not affected:
            raise ValueError("folder not found")
        affected_set = set(affected)
        unchanged = [p for p in clean if p not in affected_set]
        remapped = [_remap_path(p) for p in affected]
        for p in remapped:
            if _depth(p) > _VAULT_FOLDER_MAX_SEGMENTS:
                raise ValueError("folder depth limit is 3 levels")
        if len(set(remapped)) != len(remapped):
            raise ValueError("folder path conflict")
        for p in remapped:
            if p in unchanged:
                raise ValueError("folder path conflict")
        final_paths = sorted(set(unchanged) | set(remapped))
        meta["vault_folder_paths"] = final_paths
        conn.execute(
            "UPDATE capture_projects SET metadata_json = ?, updated_at = ? WHERE id = ?",
            (_json_dumps(meta), _now_iso(), project_id),
        )
        rows = conn.execute(
            "SELECT id, section FROM captures WHERE project_id = ? AND deleted_at = ''",
            (project_id,),
        ).fetchall()
        now = _now_iso()
        for cap_id, sec_raw in rows:
            sec = str(sec_raw or "").strip().strip("/")
            if not sec:
                continue
            new_sec: str | None = None
            if sec == fp:
                new_sec = new_path
            elif sec.startswith(fp + "/"):
                new_sec = (new_path + sec[len(fp) :]).strip("/")
            if new_sec is not None:
                new_sec = str(new_sec)[:_CAPTURE_SECTION_MAX]
                conn.execute(
                    "UPDATE captures SET section = ?, updated_at = ? WHERE id = ?",
                    (new_sec, now, cap_id),
                )
        conn.commit()
        return _project_lookup(conn, project_id)
    finally:
        conn.close()


def delete_capture_project_vault_folder(project_id: str, *, folder_path: str) -> dict[str, Any] | None:
    """Remove a vault folder path (and descendants) from project metadata; move captures to the parent folder.

    Captures with ``section`` equal to the deleted path, or under it, are promoted one level:
    ``Alpha/Beta/gamma`` after deleting ``Alpha/Beta`` becomes ``Alpha/gamma``; ``Testing 2/note``
    after deleting ``Testing 2`` at project root becomes ``note``.
    """
    if not project_id:
        return None
    fp = "/".join(x for x in str(folder_path or "").split("/") if x).strip("/")
    if not fp:
        raise ValueError("folder_path required")
    parts = fp.split("/")
    parent_prefix = "/".join(parts[:-1]) if len(parts) > 1 else ""

    conn = _get_connection()
    try:
        project = _project_lookup(conn, project_id)
        if not project:
            return None
        meta = dict(project.get("metadata") or {})
        paths = list(meta.get("vault_folder_paths") or [])
        if not isinstance(paths, list):
            paths = []
        clean = [str(p).strip().strip("/") for p in paths if str(p).strip()]
        to_remove = {p for p in clean if p == fp or p.startswith(fp + "/")}

        rows = conn.execute(
            "SELECT id, section FROM captures WHERE project_id = ? AND deleted_at = ''",
            (project_id,),
        ).fetchall()
        cap_touches_folder = False
        for _cap_id, sec_raw in rows:
            sec = str(sec_raw or "").strip().strip("/")
            if sec == fp or (sec.startswith(fp + "/")):
                cap_touches_folder = True
                break
        if not to_remove and not cap_touches_folder:
            raise ValueError("folder not found")

        remaining_paths = sorted({p for p in clean if p != fp and not p.startswith(fp + "/")})
        meta["vault_folder_paths"] = remaining_paths
        conn.execute(
            "UPDATE capture_projects SET metadata_json = ?, updated_at = ? WHERE id = ?",
            (_json_dumps(meta), _now_iso(), project_id),
        )
        now = _now_iso()
        for cap_id, sec_raw in rows:
            sec = str(sec_raw or "").strip().strip("/")
            new_sec: str | None = None
            if sec == fp:
                new_sec = parent_prefix
            elif sec.startswith(fp + "/"):
                tail = sec[len(fp) + 1 :]
                new_sec = f"{parent_prefix}/{tail}" if parent_prefix else tail
            if new_sec is None:
                continue
            new_sec = str(new_sec).strip().strip("/")[:_CAPTURE_SECTION_MAX]
            conn.execute(
                "UPDATE captures SET section = ?, updated_at = ? WHERE id = ?",
                (new_sec, now, cap_id),
            )
        conn.commit()
        return _project_lookup(conn, project_id)
    finally:
        conn.close()


def update_capture_project(project_id: str, **updates: Any) -> dict[str, Any] | None:
    if not project_id:
        return None
    conn = _get_connection()
    try:
        if not _project_lookup(conn, project_id):
            return None
        sets: list[str] = []
        params: list[Any] = []
        if "name" in updates:
            name = " ".join(str(updates.get("name") or "").strip().split())[:120]
            if not name:
                raise ValueError("project name required")
            sets.extend(["name_ciphertext = ?", "name_key = ?"])
            params.extend([_encrypt(name), _normalize_project_key(name)])
        if "description" in updates:
            sets.append("description_ciphertext = ?")
            params.append(_encrypt(str(updates.get("description") or "")[:4000]))
        if "sort_order" in updates:
            sets.append("sort_order = ?")
            params.append(int(updates.get("sort_order") or 0))
        if "metadata" in updates and isinstance(updates.get("metadata"), dict):
            sets.append("metadata_json = ?")
            params.append(_json_dumps(updates.get("metadata") or {}))
        if not sets:
            return _project_lookup(conn, project_id)
        sets.append("updated_at = ?")
        params.append(_now_iso())
        params.append(project_id)
        conn.execute(f"UPDATE capture_projects SET {', '.join(sets)} WHERE id = ?", params)
        conn.commit()
        return _project_lookup(conn, project_id)
    finally:
        conn.close()


def archive_capture_project(project_id: str, *, archived: bool = True) -> dict[str, Any] | None:
    if not project_id:
        return None
    conn = _get_connection()
    try:
        if not _project_lookup(conn, project_id):
            return None
        now = _now_iso()
        conn.execute(
            "UPDATE capture_projects SET archived_at = ?, updated_at = ? WHERE id = ?",
            (now if archived else "", now, project_id),
        )
        conn.commit()
        return _project_lookup(conn, project_id)
    finally:
        conn.close()


def delete_capture_project(project_id: str) -> dict[str, Any] | None:
    """Remove a capture project permanently.

    Active captures in the project are unfiled (``project_id`` and ``section`` cleared); their
    ``vault_attachments.project_id`` is cleared. Rows in ``capture_connections``, ``action_cards``,
    and ``vault_graph_edges`` scoped to this project are deleted. The ``capture_projects`` row is removed.
    """
    if not project_id:
        return None
    conn = _get_connection()
    try:
        project = _project_lookup(conn, project_id)
        if not project:
            return None
        now = _now_iso()
        cur = conn.execute(
            "UPDATE captures SET project_id = '', section = '', updated_at = ? WHERE project_id = ? AND deleted_at = ''",
            (now, project_id),
        )
        unfiled = int(cur.rowcount or 0)
        conn.execute("UPDATE vault_attachments SET project_id = '' WHERE project_id = ?", (project_id,))
        conn.execute("DELETE FROM capture_connections WHERE project_id = ?", (project_id,))
        conn.execute("DELETE FROM action_cards WHERE project_id = ?", (project_id,))
        conn.execute("DELETE FROM vault_graph_edges WHERE project_id = ?", (project_id,))
        conn.execute("DELETE FROM capture_projects WHERE id = ?", (project_id,))
        conn.commit()
        return {
            "id": project_id,
            "name": str(project.get("name") or ""),
            "unfiled_count": unfiled,
        }
    finally:
        conn.close()


def create_capture(
    text: str,
    *,
    source: str = "quick_panel",
    thread_id: str = "",
    tags: list[str] | None = None,
    attachment_refs: Any | None = None,
    metadata: dict[str, Any] | None = None,
    project_id: str = "",
    section: str = "",
    title: str = "",
) -> dict[str, Any]:
    ingested = _ingest_path_attachment_items(attachment_refs or [])
    refs = [pair[0] for pair in ingested]
    if isinstance(attachment_refs, list):
        for item in attachment_refs:
            if isinstance(item, str):
                s = item.strip()
                if s and _looks_like_uuid(s) and s not in refs:
                    refs.append(s)
    body = str(text or "").strip()
    bl = body.lower()
    placeholder = bl in ("[attached files]", "[attached file]") or body == ""
    placeholder_title = ""
    if placeholder and refs:
        vm = _vault_markdown_body_and_title_for_attachment_refs(refs)
        if vm:
            body, placeholder_title = vm
        elif ingested:
            if len(ingested) == 1:
                disp = ingested[0][1]
                body = "Pasted image" if _is_temp_paste_filename(disp) else f"Pasted image: {disp}"
            else:
                clean = [n for _, n in ingested if not _is_temp_paste_filename(n)]
                if not clean:
                    body = f"Pasted files ({len(ingested)})"
                elif len(clean) == len(ingested):
                    names = ", ".join(n for _, n in ingested[:5])
                    tail = f" (+{len(ingested) - 5} more)" if len(ingested) > 5 else ""
                    body = f"Pasted files ({len(ingested)}): {names}{tail}"
                else:
                    names = ", ".join(clean[:5])
                    tail = f" (+{len(clean) - 5} more)" if len(clean) > 5 else ""
                    body = f"Pasted files ({len(ingested)}): {names}{tail}"
        else:
            # Pre-uploaded UUIDs but vault rows missing (should not happen); keep a safe fallback.
            body = "Pasted image" if len(refs) == 1 else f"Pasted files ({len(refs)})"
    elif bl in ("[attached files]", "[attached file]") and not refs:
        body = "Image capture"
    if not body:
        raise ValueError("text required")
    metadata = dict(metadata or {})
    metadata.setdefault("triage_state", "inbox")
    metadata.setdefault("processing_state", "pending")
    cid = str(uuid.uuid4())
    now = _now_iso()
    # De-dupe must consider vault attachments: many captures share the same placeholder body
    # (e.g. "Pasted image") but are different notes with different bytes.
    if refs:
        _dup_key = body + "\0" + ",".join(sorted(refs))
        body_hash = hashlib.sha256(_dup_key.encode("utf-8")).hexdigest()
    else:
        body_hash = hashlib.sha256(body.encode("utf-8")).hexdigest()
    title = str(title or "").strip()
    if not title and placeholder_title:
        title = placeholder_title
    if not title:
        title = _derive_capture_title(body)
    title = title[:_CAPTURE_TITLE_MAX]
    tag_list = _normalize_tags(tags or [])
    conn = _get_connection()
    try:
        if project_id and not _project_lookup(conn, project_id):
            raise ValueError("project not found")
        allow_duplicate = bool(metadata.get("allow_duplicate"))
        if not allow_duplicate:
            dup = conn.execute(_capture_select_sql() + " WHERE body_hash = ? AND deleted_at = '' ORDER BY created_at DESC LIMIT 1", (body_hash,)).fetchone()
            if dup:
                capture = _capture_from_row(conn, dup)
                capture["duplicate"] = True
                return capture
        conn.execute(
            """INSERT INTO captures
               (id, body_ciphertext, title_ciphertext, reason_ciphertext, normalized_ciphertext, body_hash, source, thread_id, status, kind,
                tags_json, attachment_refs_json, metadata_json, project_id, section, created_at, updated_at, deleted_at)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'inbox', 'raw', ?, ?, ?, ?, ?, ?, ?, '')""",
            (
                cid,
                _encrypt(body),
                _encrypt(title),
                _encrypt(""),
                _encrypt(_normalize_capture_plaintext(body)),
                body_hash,
                str(source or "quick_panel")[:80],
                str(thread_id or "")[:128],
                _json_dumps(tag_list),
                _json_dumps(refs),
                _json_dumps(metadata),
                str(project_id or ""),
                str(section or "")[:_CAPTURE_SECTION_MAX],
                now,
                now,
            ),
        )
        if refs:
            conn.execute(
                f"UPDATE vault_attachments SET note_id = ?, project_id = ? WHERE id IN ({','.join('?' for _ in refs)})",
                [cid, str(project_id or ""), *refs],
            )
        _refresh_wikilinks(conn, cid, body)
        conn.commit()
        try:
            maybe_refresh_capture_embedding(cid)
        except Exception:
            pass
        capture = get_capture(cid) or {}
        capture["duplicate"] = False
        return capture
    finally:
        conn.close()


def get_capture(capture_id: str) -> dict[str, Any] | None:
    if not capture_id:
        return None
    conn = _get_connection()
    try:
        row = conn.execute(_capture_select_sql() + " WHERE id = ?", (capture_id,)).fetchone()
        return _capture_from_row(conn, row) if row else None
    finally:
        conn.close()


_CAPTURE_LIST_MAX_LIMIT = 500
_CAPTURE_LIST_MAX_OFFSET = 100_000


def list_captures(
    status: str | None = "inbox",
    *,
    include_deleted: bool = False,
    limit: int = 100,
    offset: int = 0,
    project_id: str | None = None,
    tag: str | None = None,
    unfiled: bool = False,
    processing_state: str | None = None,
    triage_state: str | None = None,
) -> list[dict[str, Any]]:
    limit = max(1, min(int(limit or 100), _CAPTURE_LIST_MAX_LIMIT + 1))
    offset = max(0, min(int(offset or 0), _CAPTURE_LIST_MAX_OFFSET))
    clauses: list[str] = []
    params: list[Any] = []
    if not include_deleted:
        clauses.append("deleted_at = ''")
    if status:
        clauses.append("(status = ? OR (? = 'inbox' AND status IN ('processed', 'promoted')))")
        params.extend([status, status])
    if project_id is not None:
        clauses.append("project_id = ?")
        params.append(project_id)
    if unfiled:
        clauses.append("project_id = ''")
    where = " WHERE " + " AND ".join(clauses) if clauses else ""
    conn = _get_connection()
    try:
        rows = conn.execute(
            _capture_select_sql() + where + " ORDER BY updated_at DESC, created_at DESC LIMIT ? OFFSET ?",
            [*params, limit, offset],
        ).fetchall()
        captures = [_capture_from_row(conn, row) for row in rows]
    finally:
        conn.close()
    if tag:
        captures = [c for c in captures if tag in (c.get("tags") or [])]
    if processing_state:
        captures = [c for c in captures if (c.get("metadata") or {}).get("processing_state") == processing_state]
    if triage_state:
        captures = [c for c in captures if (c.get("metadata") or {}).get("triage_state") == triage_state]
    return captures[:limit]


def list_recent_vault_notes(
    *,
    limit: int = 30,
    project_id: str | None = None,
    since_hours: float | None = None,
) -> dict[str, Any]:
    """List recently updated Vault notes (captures) with metadata only — no body text.

    Ordered by ``updated_at`` descending. Use ``get_note`` for full content.
    """
    limit = max(1, min(int(limit or 30), 200))
    clauses = ["deleted_at = ''"]
    params: list[Any] = []
    if project_id is not None and str(project_id).strip():
        clauses.append("project_id = ?")
        params.append(str(project_id).strip())
    if since_hours is not None and float(since_hours) > 0:
        cutoff = datetime.utcnow() - timedelta(hours=float(since_hours))
        clauses.append("updated_at >= ?")
        params.append(cutoff.isoformat() + "Z")
    where = " WHERE " + " AND ".join(clauses)
    conn = _get_connection()
    notes: list[dict[str, Any]] = []
    try:
        rows = conn.execute(
            "SELECT id, title_ciphertext, project_id, status, kind, updated_at, created_at FROM captures"
            + where
            + " ORDER BY updated_at DESC, created_at DESC LIMIT ?",
            [*params, limit],
        ).fetchall()
        for row in rows:
            cid, title_ct, pid, status, kind, updated_at, created_at = row
            try:
                title = _decrypt(title_ct) if title_ct else ""
            except Exception:
                title = ""
            pid_s = pid or ""
            notes.append(
                {
                    "id": cid,
                    "title": title or "(untitled)",
                    "project_id": pid_s,
                    "project_name": _project_name(conn, pid_s),
                    "status": status or "",
                    "kind": kind or "",
                    "updated_at": updated_at or "",
                    "created_at": created_at or "",
                }
            )
    finally:
        conn.close()
    return {"ok": True, "notes": notes}


def list_vault_notes_discussed_since(
    *,
    since_hours: float = 168,
    limit: int = 24,
) -> dict[str, Any]:
    """List vault notes whose ``metadata.discussed_at`` falls within ``since_hours``.

    Includes notes even when ``updated_at`` is older than the window (e.g. user
    discussed a stale capture via Discuss-this).
    """
    limit = max(1, min(int(limit or 24), 100))
    since_hours = max(0.0, float(since_hours or 0))
    if since_hours <= 0:
        return {"ok": True, "notes": []}
    cutoff = datetime.utcnow() - timedelta(hours=since_hours)

    conn = _get_connection()
    candidates: list[tuple[datetime, dict[str, Any]]] = []
    try:
        rows = conn.execute(
            "SELECT id, title_ciphertext, project_id, status, kind, updated_at, created_at, metadata_json "
            "FROM captures WHERE deleted_at = '' AND metadata_json LIKE ?",
            ('%"discussed_at"%',),
        ).fetchall()
        for row in rows:
            cid, title_ct, pid, status, kind, updated_at, created_at, metadata_json = row
            metadata = _json_loads(metadata_json, {})
            discussed_raw = str((metadata or {}).get("discussed_at") or "").strip()
            if not discussed_raw:
                continue
            discussed_dt = _parse_capture_iso_ts(discussed_raw)
            if discussed_dt is None or discussed_dt < cutoff:
                continue
            try:
                title = _decrypt(title_ct) if title_ct else ""
            except Exception:
                title = ""
            candidates.append(
                (
                    discussed_dt,
                    {
                        "id": cid,
                        "title": title or "(untitled)",
                        "project_id": pid or "",
                        "project_name": _project_name(conn, pid or ""),
                        "status": status or "",
                        "kind": kind or "",
                        "updated_at": updated_at or "",
                        "created_at": created_at or "",
                        "discussed_at": discussed_raw,
                    },
                )
            )
    finally:
        conn.close()

    candidates.sort(key=lambda item: item[0], reverse=True)
    notes = [note for _dt, note in candidates[:limit]]
    return {"ok": True, "notes": notes}


def _parse_capture_iso_ts(raw: str) -> datetime | None:
    text = (raw or "").strip()
    if not text:
        return None
    try:
        if text.endswith("Z"):
            text = text[:-1] + "+00:00"
        dt = datetime.fromisoformat(text)
        if dt.tzinfo is not None:
            dt = dt.replace(tzinfo=None)
        return dt
    except Exception:
        return None


def list_recent_vault_graph_pairs(
    *,
    since_hours: float = 168,
    limit: int = 12,
) -> dict[str, Any]:
    """List recent Vault graph edges as title pairs (wikilink + active/suggested edges).

    Includes an edge when at least one endpoint note was updated within ``since_hours``.
    """
    limit = max(1, min(int(limit or 12), 50))
    since_hours = max(0.0, float(since_hours or 0))
    cutoff = ""
    if since_hours > 0:
        cutoff = (datetime.utcnow() - timedelta(hours=since_hours)).isoformat() + "Z"

    conn = _get_connection()
    pairs: list[dict[str, str]] = []
    seen: set[tuple[str, str, str]] = set()
    try:
        def _title_for(note_id: str) -> str:
            row = conn.execute(
                "SELECT title_ciphertext FROM captures WHERE id = ? AND deleted_at = ''",
                (note_id,),
            ).fetchone()
            if not row or not row[0]:
                return "(untitled)"
            try:
                return _decrypt(row[0]) or "(untitled)"
            except Exception:
                return "(untitled)"

        def _add_pair(title_a: str, title_b: str, kind: str) -> None:
            a = (title_a or "(untitled)").strip()
            b = (title_b or "(untitled)").strip()
            if a.lower() > b.lower():
                a, b = b, a
            key = (a, b, kind)
            if key in seen:
                return
            seen.add(key)
            pairs.append({"title_a": a, "title_b": b, "kind": kind})

        wikilink_sql = """
            SELECT nl.src_id, nl.dst_id
            FROM note_links nl
            INNER JOIN captures cs ON cs.id = nl.src_id AND cs.deleted_at = ''
            INNER JOIN captures cd ON cd.id = nl.dst_id AND cd.deleted_at = ''
            WHERE nl.link_kind = 'wikilink'
        """
        wikilink_params: list[Any] = []
        if cutoff:
            wikilink_sql += " AND (cs.updated_at >= ? OR cd.updated_at >= ?)"
            wikilink_params.extend([cutoff, cutoff])
        wikilink_sql += " ORDER BY cs.updated_at DESC, cd.updated_at DESC LIMIT ?"
        wikilink_params.append(limit * 3)
        for src_id, dst_id in conn.execute(wikilink_sql, wikilink_params).fetchall():
            if len(pairs) >= limit:
                break
            _add_pair(_title_for(str(src_id)), _title_for(str(dst_id)), "wikilink")

        if len(pairs) < limit:
            edge_sql = """
                SELECT e.src_id, e.dst_id, e.status, e.link_kind
                FROM vault_graph_edges e
                INNER JOIN captures cs ON cs.id = e.src_id AND cs.deleted_at = ''
                INNER JOIN captures cd ON cd.id = e.dst_id AND cd.deleted_at = ''
                WHERE e.status != 'rejected'
            """
            edge_params: list[Any] = []
            if cutoff:
                edge_sql += " AND (cs.updated_at >= ? OR cd.updated_at >= ?)"
                edge_params.extend([cutoff, cutoff])
            edge_sql += " ORDER BY e.updated_at DESC LIMIT ?"
            edge_params.append(limit * 3)
            for src_id, dst_id, status, link_kind in conn.execute(edge_sql, edge_params).fetchall():
                if len(pairs) >= limit:
                    break
                kind = str(status or link_kind or "linked").strip() or "linked"
                _add_pair(_title_for(str(src_id)), _title_for(str(dst_id)), kind)
    finally:
        conn.close()
    return {"ok": True, "pairs": pairs[:limit]}


def update_capture(capture_id: str, **updates: Any) -> dict[str, Any] | None:
    current = get_capture(capture_id)
    if not current:
        return None
    metadata = dict(current.get("metadata") or {})
    sets: list[str] = []
    params: list[Any] = []
    if "body" in updates or "text" in updates:
        body = str(updates.get("body", updates.get("text", "")) or "")
        sets.extend(["body_ciphertext = ?", "body_hash = ?", "normalized_ciphertext = ?"])
        params.extend(
            [
                _encrypt(body),
                hashlib.sha256(body.encode("utf-8")).hexdigest(),
                _encrypt(_normalize_capture_plaintext(body)),
            ]
        )
        if "title" not in updates:
            sets.append("title_ciphertext = ?")
            params.append(_encrypt(_derive_capture_title(body)))
    if "title" in updates:
        sets.append("title_ciphertext = ?")
        params.append(_encrypt(str(updates.get("title") or "")[:_CAPTURE_TITLE_MAX]))
    if "reason" in updates:
        sets.append("reason_ciphertext = ?")
        params.append(_encrypt(str(updates.get("reason") or "")))
    for key in ("source", "thread_id", "status", "kind"):
        if key in updates:
            sets.append(f"{key} = ?")
            params.append(str(updates.get(key) or ""))
    if "section" in updates:
        sets.append("section = ?")
        params.append(str(updates.get("section") or "")[:_CAPTURE_SECTION_MAX])
    if "project_id" in updates:
        project_id = str(updates.get("project_id") or "")
        if project_id:
            conn_check = _get_connection()
            try:
                if not _project_lookup(conn_check, project_id):
                    raise ValueError("project not found")
            finally:
                conn_check.close()
        sets.append("project_id = ?")
        params.append(project_id)
    if "tags" in updates:
        sets.append("tags_json = ?")
        params.append(_json_dumps(_normalize_tags(updates.get("tags"))))
    if "attachment_refs" in updates:
        sets.append("attachment_refs_json = ?")
        params.append(_json_dumps(_normalize_attachment_refs(updates.get("attachment_refs"))))
    if "metadata" in updates and isinstance(updates.get("metadata"), dict):
        metadata.update(updates.get("metadata") or {})
    sets.append("metadata_json = ?")
    params.append(_json_dumps(metadata))
    sets.append("updated_at = ?")
    params.append(_now_iso())
    params.append(capture_id)
    new_body_for_links = str(updates.get("body", updates.get("text", current.get("body") or "")) or "")
    conn = _get_connection()
    try:
        conn.execute(f"UPDATE captures SET {', '.join(sets)} WHERE id = ?", params)
        if "project_id" in updates:
            conn.execute("UPDATE vault_attachments SET project_id = ? WHERE note_id = ?", (str(updates.get("project_id") or ""), capture_id))
        if "body" in updates or "text" in updates:
            _refresh_wikilinks(conn, capture_id, new_body_for_links)
        conn.commit()
        if "body" in updates or "text" in updates:
            try:
                maybe_refresh_capture_embedding(capture_id)
            except Exception:
                pass
    finally:
        conn.close()
    return get_capture(capture_id)


def file_capture(capture_id: str, *, project_id: str, section: str = "", tags: list[str] | None = None) -> dict[str, Any] | None:
    if not project_id:
        raise ValueError("project_id required")
    current = get_capture(capture_id)
    if not current:
        return None
    merged_tags = list(current.get("tags") or [])
    for tag in _normalize_tags(tags or []):
        if tag not in merged_tags:
            merged_tags.append(tag)
    return update_capture(capture_id, project_id=project_id, section=section, tags=merged_tags)


def unfile_capture(capture_id: str) -> dict[str, Any] | None:
    return update_capture(capture_id, project_id="", section="")


def tag_capture(capture_id: str, *, tags: list[str], mode: str = "add") -> dict[str, Any] | None:
    current = get_capture(capture_id)
    if not current:
        return None
    existing = list(current.get("tags") or [])
    normalized = _normalize_tags(tags)
    if mode == "replace":
        updated = normalized
    elif mode == "remove":
        updated = [tag for tag in existing if tag not in normalized]
    else:
        updated = existing[:]
        for tag in normalized:
            if tag not in updated:
                updated.append(tag)
    return update_capture(capture_id, tags=updated)


def soft_delete_capture(capture_id: str) -> bool:
    if not capture_id:
        return False
    now = _now_iso()
    conn = _get_connection()
    try:
        cur = conn.execute("UPDATE captures SET deleted_at = ?, updated_at = ? WHERE id = ? AND deleted_at = ''", (now, now, capture_id))
        conn.commit()
        return cur.rowcount > 0
    finally:
        conn.close()


def restore_capture(capture_id: str) -> dict[str, Any] | None:
    if not capture_id:
        return None
    conn = _get_connection()
    try:
        conn.execute("UPDATE captures SET deleted_at = '', updated_at = ? WHERE id = ?", (_now_iso(), capture_id))
        conn.commit()
    finally:
        conn.close()
    return get_capture(capture_id)


def _looks_secret(text: str) -> bool:
    lower = (text or "").lower()
    if any(marker in lower for marker in ("api_key", "apikey", "secret", "password", "authorization:", "bearer ")):
        return True
    return "sk-" in text and len(text) > 24


def _shared_terms(captures: list[dict[str, Any]]) -> list[str]:
    stop = {"the", "and", "this", "that", "with", "from", "into", "another", "recurring"}
    counts: dict[str, int] = {}
    for cap in captures:
        words = {w.strip(".,:;!?()[]{}").lower() for w in (cap.get("body") or "").split()}
        for word in words:
            if len(word) >= 4 and word not in stop:
                counts[word] = counts.get(word, 0) + 1
    return [word for word, count in sorted(counts.items(), key=lambda kv: (-kv[1], kv[0])) if count >= 2][:5]


def process_captures(ids: list[str] | None = None, *, limit: int = 25, project_id: str | None = None) -> dict[str, Any]:
    if ids:
        captures = [c for cid in ids if (c := get_capture(str(cid)))]
    else:
        captures = list_captures("inbox", limit=limit, project_id=project_id, processing_state="pending")
    if project_id:
        captures = [c for c in captures if c.get("project_id") == project_id]
    processed: list[dict[str, Any]] = []
    skipped: list[dict[str, Any]] = []
    for cap in captures[: max(1, min(int(limit or 25), 100))]:
        meta = dict(cap.get("metadata") or {})
        if _looks_secret(cap.get("body") or ""):
            meta["processing_state"] = "skipped_secret"
            updated = update_capture(cap["id"], metadata=meta)
            if updated:
                skipped.append(updated)
            continue
        meta["processing_state"] = "processed"
        updated = update_capture(cap["id"], metadata=meta)
        if updated:
            processed.append(updated)
    connections = generate_capture_connections(project_id=project_id, captures=processed)
    return {"processed": processed, "skipped": skipped, "connections": connections}


def generate_capture_connections(project_id: str | None = None, captures: list[dict[str, Any]] | None = None) -> list[dict[str, Any]]:
    captures = captures if captures is not None else list_captures("inbox", limit=100, project_id=project_id, include_deleted=False)
    captures = [c for c in captures if not project_id or c.get("project_id") == project_id]
    if len(captures) < 2:
        return []
    terms = _shared_terms(captures)
    if not terms and len(captures) < 2:
        return []
    source_ids = [c["id"] for c in captures[:8]]
    summary = "Recurring theme: " + (", ".join(terms[:3]) if terms else "related captures")
    why = "Multiple notes mention similar words or context. Review this as a suggestion; source notes are unchanged."
    conn = _get_connection()
    try:
        existing = conn.execute(
            "SELECT id, summary_ciphertext, why_ciphertext, source_ids_json, status, scope, project_id, feedback, created_at, updated_at FROM capture_connections WHERE source_ids_json = ? AND project_id = ? LIMIT 1",
            (_json_dumps(source_ids), project_id or ""),
        ).fetchone()
        if existing:
            return [_connection_from_row(conn, existing)]
        cid = str(uuid.uuid4())
        now = _now_iso()
        conn.execute(
            """INSERT INTO capture_connections
               (id, summary_ciphertext, why_ciphertext, source_ids_json, status, scope, project_id, feedback, created_at, updated_at)
               VALUES (?, ?, ?, ?, 'suggested', 'theme', ?, '', ?, ?)""",
            (cid, _encrypt(summary), _encrypt(why), _json_dumps(source_ids), project_id or "", now, now),
        )
        conn.commit()
    finally:
        conn.close()
    for cap in captures:
        meta = dict(cap.get("metadata") or {})
        if meta.get("processing_state") == "processed":
            meta["processing_state"] = "used_in_pattern"
            update_capture(cap["id"], metadata=meta)
    card = _create_action_card(
        title=summary,
        body=why,
        source_ids=source_ids,
        project_id=project_id or "",
        action_type="review_pattern",
    )
    connection = get_connection_card(cid)
    return [connection] if connection else []


def _connection_from_row(conn: sqlite3.Connection, row: tuple[Any, ...]) -> dict[str, Any]:
    cid, summary_blob, why_blob, source_ids_json, status, scope, project_id, feedback, created_at, updated_at = row
    return {
        "id": cid,
        "summary": _decrypt(summary_blob) if summary_blob else "",
        "why": _decrypt(why_blob) if why_blob else "",
        "source_ids": _json_loads(source_ids_json, []),
        "status": status or "suggested",
        "scope": scope or "theme",
        "project_id": project_id or "",
        "project_name": _project_name(conn, project_id or ""),
        "feedback": feedback or "",
        "created_at": created_at or "",
        "updated_at": updated_at or "",
    }


def get_connection_card(connection_id: str) -> dict[str, Any] | None:
    conn = _get_connection()
    try:
        row = conn.execute(
            "SELECT id, summary_ciphertext, why_ciphertext, source_ids_json, status, scope, project_id, feedback, created_at, updated_at FROM capture_connections WHERE id = ?",
            (connection_id,),
        ).fetchone()
        return _connection_from_row(conn, row) if row else None
    finally:
        conn.close()


def list_connections(status: str | None = "suggested", *, limit: int = 100, project_id: str | None = None) -> list[dict[str, Any]]:
    clauses: list[str] = []
    params: list[Any] = []
    if status:
        clauses.append("status = ?")
        params.append(status)
    if project_id is not None:
        clauses.append("project_id = ?")
        params.append(project_id)
    where = " WHERE " + " AND ".join(clauses) if clauses else ""
    conn = _get_connection()
    try:
        rows = conn.execute(
            "SELECT id, summary_ciphertext, why_ciphertext, source_ids_json, status, scope, project_id, feedback, created_at, updated_at FROM capture_connections"
            + where
            + " ORDER BY created_at DESC LIMIT ?",
            [*params, max(1, min(int(limit or 100), 500))],
        ).fetchall()
        return [_connection_from_row(conn, row) for row in rows]
    finally:
        conn.close()


def update_connection_card(connection_id: str, *, status: str = "", feedback: str = "") -> dict[str, Any] | None:
    if not connection_id:
        return None
    sets: list[str] = []
    params: list[Any] = []
    if status:
        sets.append("status = ?")
        params.append(status)
    if feedback:
        sets.append("feedback = ?")
        params.append(feedback[:2000])
    if not sets:
        return get_connection_card(connection_id)
    sets.append("updated_at = ?")
    params.append(_now_iso())
    params.append(connection_id)
    conn = _get_connection()
    try:
        cur = conn.execute(f"UPDATE capture_connections SET {', '.join(sets)} WHERE id = ?", params)
        conn.commit()
        if cur.rowcount <= 0:
            return None
    finally:
        conn.close()
    return get_connection_card(connection_id)


def _action_card_from_row(conn: sqlite3.Connection, row: tuple[Any, ...]) -> dict[str, Any]:
    aid, title_blob, body_blob, action_type, trust_level, state, source_ids_json, project_id, failed_reason, created_at, updated_at = row
    return {
        "id": aid,
        "title": _decrypt(title_blob) if title_blob else "",
        "body": _decrypt(body_blob) if body_blob else "",
        "action_type": action_type or "explain",
        "trust_level": trust_level or "informational",
        "state": state or "suggested",
        "source_ids": _json_loads(source_ids_json, []),
        "project_id": project_id or "",
        "project_name": _project_name(conn, project_id or ""),
        "failed_reason": failed_reason or "",
        "created_at": created_at or "",
        "updated_at": updated_at or "",
    }


def _create_action_card(*, title: str, body: str, source_ids: list[str], project_id: str = "", action_type: str = "explain") -> dict[str, Any]:
    conn = _get_connection()
    try:
        existing = conn.execute(
            "SELECT id, title_ciphertext, body_ciphertext, action_type, trust_level, state, source_ids_json, project_id, failed_reason, created_at, updated_at FROM action_cards WHERE source_ids_json = ? AND project_id = ? LIMIT 1",
            (_json_dumps(source_ids), project_id),
        ).fetchone()
        if existing:
            return _action_card_from_row(conn, existing)
        aid = str(uuid.uuid4())
        now = _now_iso()
        conn.execute(
            """INSERT INTO action_cards
               (id, title_ciphertext, body_ciphertext, action_type, trust_level, state, source_ids_json, project_id, failed_reason, created_at, updated_at)
               VALUES (?, ?, ?, ?, 'informational', 'suggested', ?, ?, '', ?, ?)""",
            (aid, _encrypt(title[:140]), _encrypt(body), action_type, _json_dumps(source_ids), project_id, now, now),
        )
        conn.commit()
        card = conn.execute(
            "SELECT id, title_ciphertext, body_ciphertext, action_type, trust_level, state, source_ids_json, project_id, failed_reason, created_at, updated_at FROM action_cards WHERE id = ?",
            (aid,),
        ).fetchone()
        return _action_card_from_row(conn, card)
    finally:
        conn.close()


def get_action_card(action_id: str) -> dict[str, Any] | None:
    conn = _get_connection()
    try:
        row = conn.execute(
            "SELECT id, title_ciphertext, body_ciphertext, action_type, trust_level, state, source_ids_json, project_id, failed_reason, created_at, updated_at FROM action_cards WHERE id = ?",
            (action_id,),
        ).fetchone()
        return _action_card_from_row(conn, row) if row else None
    finally:
        conn.close()


def list_action_cards(state: str | None = "suggested", *, limit: int = 100, project_id: str | None = None) -> list[dict[str, Any]]:
    clauses: list[str] = []
    params: list[Any] = []
    if state:
        clauses.append("state = ?")
        params.append(state)
    if project_id is not None:
        clauses.append("project_id = ?")
        params.append(project_id)
    where = " WHERE " + " AND ".join(clauses) if clauses else ""
    conn = _get_connection()
    try:
        rows = conn.execute(
            "SELECT id, title_ciphertext, body_ciphertext, action_type, trust_level, state, source_ids_json, project_id, failed_reason, created_at, updated_at FROM action_cards"
            + where
            + " ORDER BY created_at DESC LIMIT ?",
            [*params, max(1, min(int(limit or 100), 500))],
        ).fetchall()
        return [_action_card_from_row(conn, row) for row in rows]
    finally:
        conn.close()


def update_action_card(action_id: str, *, state: str = "", failed_reason: str = "") -> dict[str, Any] | None:
    if not action_id:
        return None
    sets: list[str] = []
    params: list[Any] = []
    if state:
        sets.append("state = ?")
        params.append(state)
    if failed_reason:
        sets.append("failed_reason = ?")
        params.append(failed_reason[:2000])
    if not sets:
        return get_action_card(action_id)
    sets.append("updated_at = ?")
    params.append(_now_iso())
    params.append(action_id)
    conn = _get_connection()
    try:
        cur = conn.execute(f"UPDATE action_cards SET {', '.join(sets)} WHERE id = ?", params)
        conn.commit()
        if cur.rowcount <= 0:
            return None
    finally:
        conn.close()
    return get_action_card(action_id)


def get_capture_project_summary(project_id: str) -> dict[str, Any] | None:
    conn = _get_connection()
    try:
        project = _project_lookup(conn, project_id)
    finally:
        conn.close()
    if not project:
        return None
    capture_page = list_captures(None, project_id=project_id, limit=_CAPTURE_LIST_MAX_LIMIT + 1)
    captures_has_more = len(capture_page) > _CAPTURE_LIST_MAX_LIMIT
    captures = capture_page[:_CAPTURE_LIST_MAX_LIMIT]
    connections = list_connections(None, project_id=project_id, limit=100)
    action_cards = list_action_cards(None, project_id=project_id, limit=100)
    attachments = list_vault_attachments(project_id=project_id, limit=200)
    by_status: dict[str, int] = {}
    processed_count = 0
    for cap in captures:
        status = cap.get("status") or "inbox"
        by_status[status] = by_status.get(status, 0) + 1
        if (cap.get("metadata") or {}).get("processing_state") in {"processed", "used_in_pattern"}:
            processed_count += 1
    return {
        "project": project,
        "counts": {
            "captures": len(captures),
            "connections": len(connections),
            "action_cards": len(action_cards),
            "attachments": len(attachments),
            "processed_captures": processed_count,
            "by_status": by_status,
        },
        "captures": captures,
        "captures_has_more": captures_has_more,
        "recent_captures": captures[:20],
        "recent_notes": captures[:20],
        "attachments": attachments[:50],
        "connections": connections,
        "action_cards": action_cards,
    }


def _attachments_dir() -> str:
    base = os.path.join(os.path.dirname(DB_PATH), "vault_attachments")
    os.makedirs(base, exist_ok=True, mode=0o700)
    return base


def _attachment_path(attachment_id: str) -> str:
    return os.path.join(_attachments_dir(), f"{attachment_id}.bin")


def _validate_attachment_mime(mime_type: str) -> str:
    mime = str(mime_type or "application/octet-stream").split(";")[0].strip().lower()
    if not mime:
        mime = "application/octet-stream"
    if mime in _ALLOWED_ATTACHMENT_MIME_TYPES or any(mime.startswith(prefix) for prefix in _ALLOWED_ATTACHMENT_MIME_PREFIXES):
        return mime
    raise ValueError("unsupported attachment type")


def create_vault_attachment(
    data: bytes,
    *,
    mime_type: str,
    filename: str = "",
    note_id: str = "",
    project_id: str = "",
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    raw = data or b""
    if not raw:
        raise ValueError("attachment is empty")
    if len(raw) > _ATTACHMENT_MAX_BYTES:
        raise ValueError("attachment too large")
    mime = _validate_attachment_mime(mime_type)
    aid = str(uuid.uuid4())
    content_hash = hashlib.sha256(raw).hexdigest()
    meta = dict(metadata or {})
    if filename:
        meta["original_filename"] = os.path.basename(filename)[:240]
    path = _attachment_path(aid)
    with open(path, "wb") as f:
        f.write(_encrypt_bytes(raw))
    try:
        os.chmod(path, 0o600)
    except OSError:
        pass
    now = _now_iso()
    conn = _get_connection()
    try:
        conn.execute(
            """INSERT INTO vault_attachments
               (id, note_id, project_id, mime_type, byte_size, storage_path, content_hash, metadata_json, created_at)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (aid, str(note_id or "").strip(), project_id or "", mime, len(raw), path, content_hash, _json_dumps(meta), now),
        )
        conn.commit()
    finally:
        conn.close()
    return get_vault_attachment(aid) or {}


def get_vault_attachment(attachment_id: str, *, include_data: bool = False) -> dict[str, Any] | None:
    if not attachment_id:
        return None
    conn = _get_connection()
    try:
        row = conn.execute(
            "SELECT id, note_id, project_id, mime_type, byte_size, storage_path, content_hash, metadata_json, created_at FROM vault_attachments WHERE id = ?",
            (attachment_id,),
        ).fetchone()
    finally:
        conn.close()
    if not row:
        return None
    aid, note_id, project_id, mime_type, byte_size, storage_path, content_hash, metadata_json, created_at = row
    out = {
        "id": aid,
        "note_id": note_id or "",
        "project_id": project_id or "",
        "mime_type": mime_type or "application/octet-stream",
        "byte_size": int(byte_size or 0),
        "content_hash": content_hash or "",
        "metadata": _json_loads(metadata_json, {}),
        "created_at": created_at or "",
    }
    if include_data:
        abs_path = os.path.abspath(storage_path or "")
        base = os.path.abspath(_attachments_dir())
        if not abs_path.startswith(base + os.sep):
            raise ValueError("invalid attachment path")
        with open(abs_path, "rb") as f:
            out["data"] = _decrypt_bytes(f.read())
    return out


def link_vault_attachment(attachment_id: str, *, note_id: str, project_id: str = "") -> dict[str, Any] | None:
    if not attachment_id or not note_id:
        return None
    cap = get_capture(note_id)
    if not cap:
        return None
    project_id = project_id or cap.get("project_id") or ""
    conn = _get_connection()
    try:
        cur = conn.execute(
            "UPDATE vault_attachments SET note_id = ?, project_id = ? WHERE id = ?",
            (note_id, project_id, attachment_id),
        )
        conn.commit()
        if cur.rowcount <= 0:
            return None
    finally:
        conn.close()
    refs = _normalize_attachment_refs([*(cap.get("attachment_refs") or []), attachment_id])
    update_capture(note_id, attachment_refs=refs)
    return get_vault_attachment(attachment_id)


def list_vault_attachments(note_id: str = "", *, project_id: str = "", limit: int = 100) -> list[dict[str, Any]]:
    clauses: list[str] = []
    params: list[Any] = []
    if note_id:
        clauses.append("note_id = ?")
        params.append(note_id)
    if project_id:
        clauses.append("project_id = ?")
        params.append(project_id)
    where = " WHERE " + " AND ".join(clauses) if clauses else ""
    conn = _get_connection()
    try:
        rows = conn.execute(
            "SELECT id FROM vault_attachments" + where + " ORDER BY created_at DESC LIMIT ?",
            [*params, max(1, min(int(limit or 100), 500))],
        ).fetchall()
    finally:
        conn.close()
    return [att for row in rows if (att := get_vault_attachment(row[0]))]


def delete_vault_attachment(attachment_id: str) -> dict[str, Any] | None:
    """Delete a vault attachment row and its encrypted file; drop its id from the parent capture's refs."""
    if not attachment_id:
        return None
    att = get_vault_attachment(attachment_id)
    if not att:
        return None
    note_id = str(att.get("note_id") or "").strip()
    candidate = _attachment_path(attachment_id)
    abs_path = os.path.realpath(candidate)
    base = os.path.realpath(_attachments_dir())
    if not abs_path.startswith(base + os.sep):
        raise ValueError("invalid attachment storage path")
    conn = _get_connection()
    try:
        conn.execute("DELETE FROM vault_attachments WHERE id = ?", (attachment_id,))
        conn.commit()
    finally:
        conn.close()
    try:
        if abs_path and os.path.isfile(abs_path):
            os.remove(abs_path)
    except OSError:
        pass
    if note_id:
        cap = get_capture(note_id)
        if cap:
            old_refs = list(cap.get("attachment_refs") or [])
            new_refs = [r for r in old_refs if str(r) != attachment_id]
            if new_refs != old_refs:
                update_capture(note_id, attachment_refs=new_refs)
    return {"ok": True, "id": attachment_id}


def _orphan_attachment_kind(note_id: str, capture_id: str | None, deleted_at: str) -> str | None:
    nid = str(note_id or "").strip()
    if not nid:
        return "unlinked"
    if not capture_id:
        return "missing_note"
    if str(deleted_at or "").strip():
        return "deleted_note"
    return None


def _orphan_attachment_note_id_expr() -> str:
    return "TRIM(COALESCE(va.note_id, ''))"


def _orphan_attachment_where_sql(*, kinds: tuple[str, ...] | None = None) -> str:
    allowed = {"unlinked", "missing_note", "deleted_note"}
    selected = [k for k in (kinds or ("unlinked", "missing_note", "deleted_note")) if k in allowed]
    if not selected:
        selected = ["unlinked"]
    note_id = _orphan_attachment_note_id_expr()
    parts: list[str] = []
    if "unlinked" in selected:
        parts.append(f"{note_id} = ''")
    if "missing_note" in selected:
        parts.append(f"({note_id} != '' AND c.id IS NULL)")
    if "deleted_note" in selected:
        parts.append(f"({note_id} != '' AND c.deleted_at != '')")
    return "(" + " OR ".join(parts) + ")"


def orphan_vault_attachment_report(*, limit: int = 200, offset: int = 0) -> dict[str, Any]:
    """List vault attachment rows with no live parent note (operator hygiene report)."""
    limit = max(1, min(int(limit or 200), 500))
    offset = max(0, min(int(offset or 0), _CAPTURE_LIST_MAX_OFFSET))
    where = _orphan_attachment_where_sql()
    conn = _get_connection()
    try:
        counts_by_kind: dict[str, int] = {}
        for kind in ("unlinked", "missing_note", "deleted_note"):
            kind_where = _orphan_attachment_where_sql(kinds=(kind,))
            counts_by_kind[kind] = int(
                conn.execute(
                    f"""SELECT COUNT(*)
                        FROM vault_attachments va
                        LEFT JOIN captures c ON c.id = va.note_id
                        WHERE {kind_where}""",
                ).fetchone()[0]
                or 0
            )
        total = sum(counts_by_kind.values())
        rows = conn.execute(
            f"""SELECT va.id, va.note_id, va.project_id, va.mime_type, va.byte_size, va.created_at,
                       c.id AS capture_id, c.deleted_at
                FROM vault_attachments va
                LEFT JOIN captures c ON c.id = va.note_id
                WHERE {where}
                ORDER BY va.created_at DESC
                LIMIT ? OFFSET ?""",
            (limit, offset),
        ).fetchall()
    finally:
        conn.close()
    orphans: list[dict[str, Any]] = []
    for aid, note_id, project_id, mime_type, byte_size, created_at, capture_id, deleted_at in rows:
        kind = _orphan_attachment_kind(note_id, capture_id, deleted_at)
        if not kind:
            continue
        orphans.append(
            {
                "id": aid,
                "kind": kind,
                "note_id": str(note_id or "").strip(),
                "project_id": project_id or "",
                "mime_type": mime_type or "application/octet-stream",
                "byte_size": int(byte_size or 0),
                "created_at": created_at or "",
            }
        )
    page_rows = len(rows)
    return {
        "ok": True,
        "total": total,
        "limit": limit,
        "offset": offset,
        "has_more": offset + page_rows < total,
        "counts_by_kind": counts_by_kind,
        "orphans": orphans,
    }


def gc_orphan_vault_attachments(
    *,
    dry_run: bool = True,
    max_delete: int = 50,
    include_dangling: bool = False,
    include_deleted_note: bool = False,
) -> dict[str, Any]:
    """Delete orphan attachment rows (+ `.bin` files). Default: pre-upload ``unlinked`` only."""
    max_delete = max(1, min(int(max_delete or 50), 200))
    kinds: list[str] = ["unlinked"]
    if include_dangling:
        kinds.append("missing_note")
    if include_deleted_note:
        kinds.append("deleted_note")
    where = _orphan_attachment_where_sql(kinds=tuple(kinds))
    conn = _get_connection()
    try:
        rows = conn.execute(
            f"""SELECT va.id, va.note_id, va.project_id, va.mime_type, va.byte_size, va.created_at,
                       c.id AS capture_id, c.deleted_at
                FROM vault_attachments va
                LEFT JOIN captures c ON c.id = va.note_id
                WHERE {where}
                ORDER BY va.created_at ASC
                LIMIT ?""",
            (max_delete,),
        ).fetchall()
    finally:
        conn.close()
    actions: list[dict[str, Any]] = []
    errors: list[dict[str, Any]] = []
    bytes_reclaimed = 0
    candidate_count = 0
    for aid, note_id, _project_id, mime_type, byte_size, created_at, capture_id, deleted_at in rows:
        kind = _orphan_attachment_kind(note_id, capture_id, deleted_at)
        if not kind or kind not in kinds:
            continue
        candidate_count += 1
        entry = {
            "id": aid,
            "kind": kind,
            "mime_type": mime_type or "",
            "byte_size": int(byte_size or 0),
            "created_at": created_at or "",
        }
        if dry_run:
            entry["dry_run"] = True
            actions.append(entry)
            bytes_reclaimed += int(byte_size or 0)
            continue
        try:
            result = delete_vault_attachment(aid)
            if not result:
                errors.append({"id": aid, "error": "not found"})
                continue
            entry["deleted"] = True
            actions.append(entry)
            bytes_reclaimed += int(byte_size or 0)
        except Exception as exc:
            errors.append({"id": aid, "error": str(exc)})
    deleted_count = sum(1 for action in actions if action.get("deleted"))
    return {
        "ok": len(errors) == 0,
        "dry_run": bool(dry_run),
        "kinds": kinds,
        "candidate_count": candidate_count,
        "selected": len(actions),
        "deleted_count": deleted_count,
        "error_count": len(errors),
        "bytes_reclaimed": bytes_reclaimed,
        "actions": actions,
        "errors": errors,
    }


def update_note_links(note_id: str, links: list[str]) -> None:
    if not note_id:
        return
    now = _now_iso()
    clean_links = []
    for link in links or []:
        value = str(link or "").strip()
        if value and value != note_id and value not in clean_links:
            clean_links.append(value)
    conn = _get_connection()
    try:
        conn.execute("DELETE FROM note_links WHERE src_id = ?", (note_id,))
        for dst in clean_links[:200]:
            conn.execute(
                "INSERT OR IGNORE INTO note_links (src_id, dst_id, link_kind, context, created_at) VALUES (?, ?, 'wikilink', '', ?)",
                (note_id, dst, now),
            )
        conn.commit()
    finally:
        conn.close()


def list_note_backlinks(note_id: str) -> list[dict[str, Any]]:
    if not note_id:
        return []
    conn = _get_connection()
    try:
        rows = conn.execute(
            "SELECT src_id, link_kind, context, created_at FROM note_links WHERE dst_id = ? ORDER BY created_at DESC",
            (note_id,),
        ).fetchall()
    finally:
        conn.close()
    return [
        {"src_id": src_id, "link_kind": kind, "context": context, "created_at": created_at}
        for src_id, kind, context, created_at in rows
    ]


# ---------- Scratchpads ----------


def get_scratchpads(tab_id: str) -> list[dict[str, Any]]:
    """Load scratchpads for a tab. Returns list of {"id": str, "title": str, "body": str} ordered by position, created_at."""
    if not tab_id or not isinstance(tab_id, str):
        return []
    conn = _get_connection()
    try:
        rows = conn.execute(
            """SELECT id, title_encrypted, body_encrypted FROM scratchpads
               WHERE tab_id = ? ORDER BY position ASC, created_at ASC""",
            (tab_id,),
        ).fetchall()
        out = []
        for row_id, title_blob, body_blob in rows:
            title = _decrypt(title_blob) if title_blob else ""
            body = _decrypt(body_blob) if body_blob else ""
            out.append({"id": row_id, "title": title, "body": body})
        return out
    finally:
        conn.close()


def save_scratchpad(sp_id: str, tab_id: str, title: str, body: str) -> None:
    """Insert or replace a scratchpad (upsert)."""
    if not sp_id or not tab_id:
        return
    now = datetime.utcnow().isoformat() + "Z"
    title = str(title) if title is not None else ""
    body = str(body) if body is not None else ""
    conn = _get_connection()
    try:
        conn.execute(
            """INSERT INTO scratchpads (id, tab_id, title_encrypted, body_encrypted, position, created_at, updated_at)
               VALUES (?, ?, ?, ?, COALESCE((SELECT MAX(position) FROM scratchpads WHERE tab_id = ?), 0) + 1, ?, ?)
               ON CONFLICT(id) DO UPDATE SET
                 title_encrypted = excluded.title_encrypted,
                 body_encrypted = excluded.body_encrypted,
                 updated_at = excluded.updated_at""",
            (sp_id, tab_id, _encrypt(title), _encrypt(body), tab_id, now, now),
        )
        conn.commit()
    finally:
        conn.close()


def update_scratchpad(sp_id: str, tab_id: str, title: str, body: str) -> bool:
    """Update an existing scratchpad. Returns True if updated, False if not found."""
    if not sp_id or not tab_id:
        return False
    now = datetime.utcnow().isoformat() + "Z"
    title = str(title) if title is not None else ""
    body = str(body) if body is not None else ""
    conn = _get_connection()
    try:
        cursor = conn.execute(
            """UPDATE scratchpads
               SET title_encrypted = ?, body_encrypted = ?, updated_at = ?
               WHERE id = ? AND tab_id = ?""",
            (_encrypt(title), _encrypt(body), now, sp_id, tab_id),
        )
        conn.commit()
        return cursor.rowcount > 0
    finally:
        conn.close()


def create_scratchpad(
    tab_id: str, title: str = "Untitled", body: str = "", sp_id: str | None = None
) -> str:
    """Create a new scratchpad. Returns its id. If sp_id is provided, use it; else generate uuid."""
    if not tab_id:
        raise ValueError("tab_id required")
    sid = sp_id or str(uuid.uuid4())
    now = datetime.utcnow().isoformat() + "Z"
    title = str(title) if title is not None else "Untitled"
    body = str(body) if body is not None else ""
    conn = _get_connection()
    try:
        next_pos = (conn.execute(
            "SELECT COALESCE(MAX(position), 0) + 1 FROM scratchpads WHERE tab_id = ?",
            (tab_id,),
        ).fetchone() or (1,))[0]
        conn.execute(
            """INSERT INTO scratchpads (id, tab_id, title_encrypted, body_encrypted, position, created_at, updated_at)
               VALUES (?, ?, ?, ?, ?, ?, ?)""",
            (sid, tab_id, _encrypt(title), _encrypt(body), next_pos, now, now),
        )
        conn.commit()
    finally:
        conn.close()
    return sid


def delete_scratchpad(sp_id: str) -> None:
    """Delete a scratchpad by id."""
    if not sp_id:
        return
    conn = _get_connection()
    try:
        conn.execute("DELETE FROM scratchpads WHERE id = ?", (sp_id,))
        conn.commit()
    finally:
        conn.close()


# ---------- Task plans (one structured plan per conversation thread) ----------


def get_task_plan_with_meta(thread_id: str) -> tuple[dict[str, Any] | None, str | None]:
    """Return ``(plan_dict, updated_at_iso)`` with a single decrypt pass.

    ``updated_at_iso`` is the stored row timestamp (UTC ISO with ``Z`` suffix
    from :func:`set_task_plan`). Either element may be ``None`` if missing or
    malformed.
    """
    if not thread_id or not isinstance(thread_id, str):
        return None, None
    conn = _get_connection()
    try:
        row = conn.execute(
            "SELECT plan_ciphertext, updated_at FROM thread_task_plans WHERE thread_id = ?",
            (thread_id,),
        ).fetchone()
    finally:
        conn.close()
    if not row:
        return None, None
    updated_at = row[1] if isinstance(row[1], str) else None
    try:
        raw = _decrypt(row[0])
        obj = json.loads(raw)
        if not isinstance(obj, dict):
            return None, updated_at
        return obj, updated_at
    except Exception:
        return None, updated_at


def get_task_plan(thread_id: str) -> dict[str, Any] | None:
    """Return the structured task plan for *thread_id*, or ``None`` if missing.

    Returns the canonical shape produced by :func:`utils.task_plan.validate_and_normalize`.
    Malformed rows (decrypt or JSON failures) are treated as "missing" so callers
    never see partial state.
    """
    plan, _ = get_task_plan_with_meta(thread_id)
    return plan


def get_thread_coordinator_batch(thread_id: str) -> str:
    """Return active coordinator batch id for *thread_id*, or empty string."""
    if not thread_id or not isinstance(thread_id, str):
        return ""
    conn = _get_connection()
    try:
        row = conn.execute(
            "SELECT batch_id FROM thread_coordinator_state WHERE thread_id=?",
            (thread_id,),
        ).fetchone()
    finally:
        conn.close()
    if not row:
        return ""
    bid = row[0] if isinstance(row, tuple) else row["batch_id"]
    return str(bid or "").strip()


def set_thread_coordinator_batch(thread_id: str, batch_id: str) -> None:
    if not thread_id or not batch_id:
        raise ValueError("thread_id and batch_id are required")
    now = datetime.utcnow().isoformat() + "Z"
    conn = _get_connection()
    try:
        conn.execute(
            """
            INSERT INTO thread_coordinator_state (thread_id, batch_id, activated_at)
            VALUES (?, ?, ?)
            ON CONFLICT(thread_id) DO UPDATE SET
                batch_id=excluded.batch_id,
                activated_at=excluded.activated_at
            """,
            (thread_id, batch_id, now),
        )
        conn.commit()
    finally:
        conn.close()


def clear_thread_coordinator_batch(thread_id: str) -> None:
    if not thread_id:
        return
    conn = _get_connection()
    try:
        conn.execute("DELETE FROM thread_coordinator_state WHERE thread_id=?", (thread_id,))
        conn.commit()
    finally:
        conn.close()


def list_coordinator_thread_ids() -> list[str]:
    """Return thread ids with an active coordinator batch pointer."""
    conn = _get_connection()
    try:
        rows = conn.execute("SELECT thread_id FROM thread_coordinator_state").fetchall()
    finally:
        conn.close()
    out: list[str] = []
    for row in rows:
        tid = row[0] if isinstance(row, tuple) else row["thread_id"]
        tid_s = str(tid or "").strip()
        if tid_s:
            out.append(tid_s)
    return out


def get_thread_council_batch(thread_id: str) -> str:
    """Return active council batch id for *thread_id*, or empty string."""
    if not thread_id or not isinstance(thread_id, str):
        return ""
    conn = _get_connection()
    try:
        row = conn.execute(
            "SELECT batch_id FROM thread_council_state WHERE thread_id=?",
            (thread_id,),
        ).fetchone()
    finally:
        conn.close()
    if not row:
        return ""
    bid = row[0] if isinstance(row, tuple) else row["batch_id"]
    return str(bid or "").strip()


def get_thread_council_activated_at(thread_id: str) -> str:
    """Return ISO timestamp when the council batch pointer was set, or empty."""
    if not thread_id or not isinstance(thread_id, str):
        return ""
    conn = _get_connection()
    try:
        row = conn.execute(
            "SELECT activated_at FROM thread_council_state WHERE thread_id=?",
            (thread_id,),
        ).fetchone()
    finally:
        conn.close()
    if not row:
        return ""
    ts = row[0] if isinstance(row, tuple) else row["activated_at"]
    return str(ts or "").strip()


def set_thread_council_batch(thread_id: str, batch_id: str) -> None:
    if not thread_id or not batch_id:
        raise ValueError("thread_id and batch_id are required")
    now = datetime.utcnow().isoformat() + "Z"
    conn = _get_connection()
    try:
        conn.execute(
            """
            INSERT INTO thread_council_state (thread_id, batch_id, activated_at)
            VALUES (?, ?, ?)
            ON CONFLICT(thread_id) DO UPDATE SET
                batch_id=excluded.batch_id,
                activated_at=excluded.activated_at
            """,
            (thread_id, batch_id, now),
        )
        conn.commit()
    finally:
        conn.close()


def list_thread_ids_with_council_state() -> list[str]:
    """Return thread ids with an active council pointer (for deadline reconcile)."""
    conn = _get_connection()
    try:
        rows = conn.execute(
            "SELECT thread_id FROM thread_council_state ORDER BY activated_at ASC"
        ).fetchall()
    finally:
        conn.close()
    out: list[str] = []
    for row in rows or []:
        tid = row[0] if isinstance(row, tuple) else row["thread_id"]
        tid_s = str(tid or "").strip()
        if tid_s:
            out.append(tid_s)
    return out


def clear_thread_council_batch(thread_id: str) -> None:
    if not thread_id:
        return
    conn = _get_connection()
    try:
        conn.execute("DELETE FROM thread_council_state WHERE thread_id=?", (thread_id,))
        conn.commit()
    finally:
        conn.close()


def set_thread_council_pending_offer(
    thread_id: str,
    *,
    offer_id: str,
    council_frame: str,
    frame_hash: str,
    rationale: str = "",
) -> None:
    if not thread_id or not offer_id or not council_frame:
        raise ValueError("thread_id, offer_id, and council_frame are required")
    now = datetime.utcnow().isoformat() + "Z"
    conn = _get_connection()
    try:
        conn.execute(
            """
            INSERT INTO thread_council_pending_offer (
                thread_id, offer_id, council_frame, frame_hash, rationale, created_at
            ) VALUES (?, ?, ?, ?, ?, ?)
            ON CONFLICT(thread_id) DO UPDATE SET
                offer_id=excluded.offer_id,
                council_frame=excluded.council_frame,
                frame_hash=excluded.frame_hash,
                rationale=excluded.rationale,
                created_at=excluded.created_at
            """,
            (
                thread_id,
                offer_id,
                council_frame,
                frame_hash,
                (rationale or "").strip(),
                now,
            ),
        )
        conn.commit()
    finally:
        conn.close()


def get_thread_council_pending_offer(thread_id: str) -> dict | None:
    if not thread_id:
        return None
    conn = _get_connection()
    try:
        row = conn.execute(
            """
            SELECT offer_id, council_frame, frame_hash, rationale, created_at
            FROM thread_council_pending_offer WHERE thread_id=?
            """,
            (thread_id,),
        ).fetchone()
    finally:
        conn.close()
    if not row:
        return None
    if isinstance(row, tuple):
        return {
            "offer_id": str(row[0] or ""),
            "council_frame": str(row[1] or ""),
            "frame_hash": str(row[2] or ""),
            "rationale": str(row[3] or ""),
            "created_at": str(row[4] or ""),
        }
    return {
        "offer_id": str(row["offer_id"] or ""),
        "council_frame": str(row["council_frame"] or ""),
        "frame_hash": str(row["frame_hash"] or ""),
        "rationale": str(row["rationale"] or ""),
        "created_at": str(row["created_at"] or ""),
    }


def clear_thread_council_pending_offer(thread_id: str) -> None:
    if not thread_id:
        return
    conn = _get_connection()
    try:
        conn.execute(
            "DELETE FROM thread_council_pending_offer WHERE thread_id=?",
            (thread_id,),
        )
        conn.commit()
    finally:
        conn.close()


def council_offer_age_seconds(created_at: str) -> float:
    """Seconds since offer created_at ISO string; large value if unparsable."""
    raw = (created_at or "").strip()
    if not raw:
        return 1e9
    try:
        from datetime import datetime, timezone

        if raw.endswith("Z"):
            raw = raw[:-1] + "+00:00"
        dt = datetime.fromisoformat(raw)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        now = datetime.now(timezone.utc)
        return max(0.0, (now - dt).total_seconds())
    except Exception:
        return 1e9


def get_thread_message_last_created_ats(thread_id: str) -> tuple[str | None, str | None]:
    """Return ``(last_any_created_at, last_assistant_created_at)`` without decrypting content.

    ISO strings match the ``messages.created_at`` column. Both may be ``None``.
    """
    if not thread_id or not isinstance(thread_id, str):
        return None, None
    conn = _get_connection()
    try:
        row_any = conn.execute(
            "SELECT MAX(created_at) FROM messages WHERE thread_id = ?",
            (thread_id,),
        ).fetchone()
        row_asst = conn.execute(
            "SELECT MAX(created_at) FROM messages WHERE thread_id = ? AND role = ?",
            (thread_id, "assistant"),
        ).fetchone()
    finally:
        conn.close()
    last_any = row_any[0] if row_any and row_any[0] else None
    last_asst = row_asst[0] if row_asst and row_asst[0] else None
    return (
        last_any if isinstance(last_any, str) else None,
        last_asst if isinstance(last_asst, str) else None,
    )


def set_task_plan(thread_id: str, plan: dict[str, Any]) -> None:
    """Upsert the task plan for *thread_id*. Caller must pass a normalized dict.

    Validation (limits + allowed statuses + size) belongs in the tool / HTTP
    layer via :func:`utils.task_plan.validate_and_normalize`. This helper does a
    defensive JSON round-trip and size check so malformed callers cannot bloat
    the DB silently.
    """
    if not thread_id or not isinstance(thread_id, str):
        raise ValueError("thread_id is required")
    if not isinstance(plan, dict):
        raise ValueError("plan must be a dict")
    from utils.task_plan import MAX_JSON_BYTES
    payload = json.dumps(plan, separators=(",", ":"))
    if len(payload.encode("utf-8")) > MAX_JSON_BYTES:
        raise ValueError(f"plan exceeds the {MAX_JSON_BYTES}-byte limit")
    now = datetime.utcnow().isoformat() + "Z"
    conn = _get_connection()
    try:
        conn.execute(
            """INSERT INTO thread_task_plans (thread_id, plan_ciphertext, updated_at)
               VALUES (?, ?, ?)
               ON CONFLICT(thread_id) DO UPDATE SET
                 plan_ciphertext = excluded.plan_ciphertext,
                 updated_at = excluded.updated_at""",
            (thread_id, _encrypt(payload), now),
        )
        conn.commit()
    finally:
        conn.close()


def delete_task_plan(thread_id: str) -> bool:
    """Delete the task plan row for *thread_id*. Returns True if a row was removed."""
    if not thread_id or not isinstance(thread_id, str):
        return False
    conn = _get_connection()
    try:
        cur = conn.execute(
            "DELETE FROM thread_task_plans WHERE thread_id = ?", (thread_id,)
        )
        conn.commit()
        return cur.rowcount > 0
    finally:
        conn.close()


# ---------- Backup / Restore / Key ----------


def verify_backup_readable(backup_db_path: str) -> bool:
    """Return True if *backup_db_path* opens and at least one non-empty row decrypts.

    Empty backups (no ciphertext) return False so callers cannot treat them as verified.
    """
    if not backup_db_path or not os.path.isfile(backup_db_path):
        return False
    try:
        conn = sqlite3.connect(backup_db_path)
        try:
            row = conn.execute(
                "SELECT content FROM messages WHERE content IS NOT NULL AND length(content) > 0 LIMIT 1"
            ).fetchone()
            if row and row[0]:
                _decrypt(row[0])
                return True
            row = conn.execute(
                "SELECT body_encrypted FROM scratchpads WHERE body_encrypted IS NOT NULL AND length(body_encrypted) > 0 LIMIT 1"
            ).fetchone()
            if row and row[0]:
                _decrypt(row[0])
                return True
            return False
        finally:
            conn.close()
    except Exception:
        return False


def export_encryption_key() -> str | None:
    """Return the encryption key as base64 string for backup. None if no key in Keychain.

    SECURITY: This function returns the raw encryption key. Only call it for explicit
    user-initiated backup operations. Access is logged for audit purposes.
    """
    import logging as _logging
    _logging.getLogger(__name__).warning(
        "export_encryption_key() called — encryption key is being exported"
    )
    try:
        import keyring
        key_b64 = keyring.get_password(_KEYRING_SERVICE, _KEYRING_ACCOUNT)
        return key_b64 if key_b64 else None
    except Exception:
        return None


def import_encryption_key(key_b64: str) -> None:
    """Store the given key in Keychain and clear cached Fernet so next use picks it up."""
    global _fernet
    if not key_b64 or not isinstance(key_b64, str):
        raise ValueError("Key must be a non-empty string")
    key_b64 = key_b64.strip()
    if not key_b64:
        raise ValueError("Key must be a non-empty string")
    try:
        import keyring
        from cryptography.fernet import Fernet
        Fernet(key_b64.encode("ascii"))
    except Exception as e:
        raise ValueError(f"Invalid key: {e}") from e
    keyring.set_password(_KEYRING_SERVICE, _KEYRING_ACCOUNT, key_b64)
    _fernet = None


# ---------- Vault knowledge graph (embeddings + suggested edges) ----------


def _encode_vault_embedding_vec(vec: list[float]) -> bytes:
    if not vec:
        return b""
    return struct.pack("<" + "f" * len(vec), *[float(x) for x in vec])


def _decode_vault_embedding_vec(blob: bytes) -> list[float]:
    if not blob:
        return []
    n = len(blob) // 4
    if n <= 0:
        return []
    return list(struct.unpack("<" + "f" * n, blob[: n * 4]))


def _cosine_similarity_vault(a: list[float], b: list[float]) -> float:
    if not a or not b:
        return 0.0
    n = min(len(a), len(b))
    dot = 0.0
    na = 0.0
    nb = 0.0
    for i in range(n):
        av = float(a[i])
        bv = float(b[i])
        dot += av * bv
        na += av * av
        nb += bv * bv
    if na <= 0.0 or nb <= 0.0:
        return 0.0
    return dot / (math.sqrt(na) * math.sqrt(nb))


def _canonical_note_pair(a: str, b: str) -> tuple[str, str]:
    x, y = str(a), str(b)
    return (x, y) if x <= y else (y, x)


def _vault_text_for_embedding(title: str, body: str) -> str:
    t = " ".join(str(title or "").strip().split())
    b = str(body or "").strip()
    chunk = (t + "\n\n" + b) if t and b else (t or b)
    if len(chunk) > _VAULT_EMBED_TEXT_MAX:
        chunk = chunk[:_VAULT_EMBED_TEXT_MAX]
    return chunk


def maybe_refresh_capture_embedding(note_id: str) -> dict[str, Any]:
    """Best-effort embedding for vault similarity. Skips secret-looking text and unavailable embedders."""
    note_id = str(note_id or "").strip()
    out: dict[str, Any] = {"ok": False, "note_id": note_id, "skipped": "missing"}
    if not note_id:
        return out
    cap = get_capture(note_id)
    if not cap or cap.get("deleted_at"):
        return {**out, "skipped": "not_found"}
    blob = _vault_text_for_embedding(str(cap.get("title") or ""), str(cap.get("body") or ""))
    if not blob.strip():
        return {**out, "skipped": "empty"}
    if _looks_secret(blob):
        return {**out, "skipped": "secret"}
    digest = hashlib.sha256(blob.encode("utf-8")).hexdigest()
    active_model = _vault_embed_model_id()
    conn = _get_connection()
    try:
        row = conn.execute(
            "SELECT content_hash, model_id, dim FROM capture_embeddings WHERE note_id = ?",
            (note_id,),
        ).fetchone()
        if row and row[0] == digest and str(row[1] or "") == active_model and int(row[2] or 0) == _vault_embed_dim():
            return {"ok": True, "note_id": note_id, "skipped": "unchanged"}
    finally:
        conn.close()
    try:
        from utility_model import utility_embed_texts
    except Exception:
        return {**out, "skipped": "no_utility"}
    vecs = utility_embed_texts(
        [blob],
        dimensions=_vault_embed_dim(),
        activity_operation="session_search_embedding",
        activity_thread_id=f"vault:{note_id}",
    )
    if not vecs or len(vecs) != 1 or not vecs[0]:
        return {**out, "skipped": "embed_failed"}
    vec = vecs[0]
    expected_dim = _vault_embed_dim()
    if len(vec) != expected_dim:
        return {**out, "skipped": "dim_mismatch", "expected": expected_dim, "got": len(vec)}
    emb_blob = _encode_vault_embedding_vec(vec)
    now = _now_iso()
    conn = _get_connection()
    try:
        conn.execute(
            """INSERT INTO capture_embeddings (note_id, content_hash, model_id, dim, embedding, updated_at)
               VALUES (?, ?, ?, ?, ?, ?)
               ON CONFLICT(note_id) DO UPDATE SET
                 content_hash = excluded.content_hash,
                 model_id = excluded.model_id,
                 dim = excluded.dim,
                 embedding = excluded.embedding,
                 updated_at = excluded.updated_at""",
            (note_id, digest, active_model, len(vec), emb_blob, now),
        )
        conn.commit()
    finally:
        conn.close()
    return {"ok": True, "note_id": note_id, "dim": len(vec)}


def _get_embedding_row(conn: sqlite3.Connection, note_id: str) -> tuple[bytes, int, str] | None:
    row = conn.execute(
        "SELECT embedding, dim, model_id FROM capture_embeddings WHERE note_id = ?", (note_id,)
    ).fetchone()
    if not row or not row[0]:
        return None
    return row[0], int(row[1] or 0), str(row[2] or "")


def find_similar_capture_notes(
    note_id: str,
    *,
    project_id: str | None = None,
    top_k: int = 12,
    min_similarity: float = _VAULT_SIMILARITY_MIN_DEFAULT,
    candidate_limit: int = _VAULT_GRAPH_SCAN_CAP,
) -> list[dict[str, Any]]:
    note_id = str(note_id or "").strip()
    if not note_id:
        return []
    conn = _get_connection()
    try:
        row = _get_embedding_row(conn, note_id)
        if not row:
            return []
        blob, dim, query_model = row
        qv = _decode_vault_embedding_vec(blob)
        if not qv or len(qv) != dim:
            return []
        clauses = ["c.deleted_at = ''", "c.id != ?"]
        params: list[Any] = [note_id]
        if project_id is not None and str(project_id).strip():
            clauses.append("c.project_id = ?")
            params.append(str(project_id).strip())
        where = " AND ".join(clauses)
        rows = conn.execute(
            f"""SELECT c.id, e.embedding, e.dim, e.model_id
                FROM captures c
                JOIN capture_embeddings e ON e.note_id = c.id
                WHERE {where}
                ORDER BY c.updated_at DESC
                LIMIT ?""",
            [*params, max(1, min(int(candidate_limit or _VAULT_GRAPH_SCAN_CAP), _VAULT_GRAPH_SCAN_CAP))],
        ).fetchall()
        scored: list[tuple[float, str]] = []
        for cid, eb, cand_dim, cand_model in rows:
            if int(cand_dim or 0) != dim or str(cand_model or "") != query_model:
                continue
            ov = _decode_vault_embedding_vec(eb)
            if not ov or len(ov) != dim:
                continue
            sim = _cosine_similarity_vault(qv, ov)
            if sim >= float(min_similarity):
                scored.append((sim, str(cid)))
        scored.sort(key=lambda t: (-t[0], t[1]))
        out: list[dict[str, Any]] = []
        for sim, cid in scored[: max(1, int(top_k or 12))]:
            cap = _capture_from_row(
                conn,
                conn.execute(_capture_select_sql() + " WHERE id = ?", (cid,)).fetchone(),
            )
            out.append(
                {
                    "note_id": cid,
                    "similarity": round(sim, 4),
                    "title": cap.get("title") or "",
                    "project_id": cap.get("project_id") or "",
                }
            )
        return out
    finally:
        conn.close()


def list_graph_neighbors(
    note_id: str,
    *,
    kinds: list[str] | None = None,
    limit: int = _VAULT_NEIGHBOR_CAP_DEFAULT,
    project_id: str | None = None,
) -> list[dict[str, Any]]:
    note_id = str(note_id or "").strip()
    if not note_id:
        return []
    limit = max(1, min(int(limit or _VAULT_NEIGHBOR_CAP_DEFAULT), 200))
    want = set(str(k) for k in (kinds or []) if k)
    conn = _get_connection()
    try:
        cap_row = conn.execute(_capture_select_sql() + " WHERE id = ? AND deleted_at = ''", (note_id,)).fetchone()
        if not cap_row:
            return []
        cap = _capture_from_row(conn, cap_row)
        eff_project = str(project_id).strip() if project_id is not None else str(cap.get("project_id") or "")
        rows_out: list[dict[str, Any]] = []

        def _want_kind(k: str) -> bool:
            return not want or k in want

        if _want_kind("wikilink"):
            for dst_id, kind, ctx, created_at in conn.execute(
                "SELECT dst_id, link_kind, context, created_at FROM note_links WHERE src_id = ? AND link_kind = 'wikilink'",
                (note_id,),
            ).fetchall():
                nrow = conn.execute(
                    _capture_select_sql() + " WHERE id = ? AND deleted_at = ''", (dst_id,)
                ).fetchone()
                if not nrow:
                    continue
                oc = _capture_from_row(conn, nrow)
                if eff_project and str(oc.get("project_id") or "") != eff_project:
                    continue
                rows_out.append(
                    {
                        "neighbor_id": str(dst_id),
                        "direction": "outbound",
                        "link_kind": "wikilink",
                        "edge_id": None,
                        "status": "active",
                        "confidence": None,
                        "title": oc.get("title") or "",
                        "context": str(ctx or ""),
                        "created_at": str(created_at or ""),
                    }
                )
            for src_id, kind, ctx, created_at in conn.execute(
                "SELECT src_id, link_kind, context, created_at FROM note_links WHERE dst_id = ? AND link_kind = 'wikilink'",
                (note_id,),
            ).fetchall():
                nrow = conn.execute(
                    _capture_select_sql() + " WHERE id = ? AND deleted_at = ''", (src_id,)
                ).fetchone()
                if not nrow:
                    continue
                oc = _capture_from_row(conn, nrow)
                if eff_project and str(oc.get("project_id") or "") != eff_project:
                    continue
                rows_out.append(
                    {
                        "neighbor_id": str(src_id),
                        "direction": "inbound",
                        "link_kind": "wikilink",
                        "edge_id": None,
                        "status": "active",
                        "confidence": None,
                        "title": oc.get("title") or "",
                        "context": str(ctx or ""),
                        "created_at": str(created_at or ""),
                    }
                )

        edge_kinds = ("suggested_similarity", "suggested_theme")
        for ek in edge_kinds:
            if not _want_kind(ek):
                continue
            for eid, sid, did, lk, st, conf, rat, pid, cat in conn.execute(
                """SELECT id, src_id, dst_id, link_kind, status, confidence, rationale, project_id, created_at
                   FROM vault_graph_edges
                   WHERE (src_id = ? OR dst_id = ?) AND status != 'rejected' AND link_kind = ?""",
                (note_id, note_id, ek),
            ).fetchall():
                other = str(did) if str(sid) == note_id else str(sid)
                nrow = conn.execute(
                    _capture_select_sql() + " WHERE id = ? AND deleted_at = ''", (other,)
                ).fetchone()
                if not nrow:
                    continue
                oc = _capture_from_row(conn, nrow)
                if eff_project and str(oc.get("project_id") or "") != eff_project:
                    continue
                rows_out.append(
                    {
                        "neighbor_id": other,
                        "direction": "outbound" if str(sid) == note_id else "inbound",
                        "link_kind": str(lk),
                        "edge_id": str(eid),
                        "status": str(st),
                        "confidence": float(conf) if conf is not None else None,
                        "title": oc.get("title") or "",
                        "rationale": str(rat or "")[:500],
                        "created_at": str(cat or ""),
                    }
                )

        dedup: dict[tuple[str, str, str], dict[str, Any]] = {}
        for r in rows_out:
            key = (r["neighbor_id"], r["link_kind"], r["direction"])
            if key not in dedup:
                dedup[key] = r
        out = list(dedup.values())
        out.sort(key=lambda x: (x["link_kind"], x["title"] or ""))
        return out[:limit]
    finally:
        conn.close()


def propose_vault_similarity_edges(
    *,
    project_id: str | None = None,
    limit_notes: int = 60,
    min_similarity: float = _VAULT_SIMILARITY_MIN_DEFAULT,
    max_new_edges: int = 200,
) -> dict[str, Any]:
    limit_notes = max(2, min(int(limit_notes or 60), 120))
    max_new_edges = max(1, min(int(max_new_edges or 200), 500))
    conn = _get_connection()
    try:
        clauses = ["c.deleted_at = ''"]
        params: list[Any] = []
        if project_id is not None and str(project_id).strip():
            clauses.append("c.project_id = ?")
            params.append(str(project_id).strip())
        where = " AND ".join(clauses)
        active_model = _vault_embed_model_id()
        active_dim = _vault_embed_dim()
        rows = conn.execute(
            f"""SELECT c.id, e.embedding, e.dim, e.model_id, c.project_id
                FROM captures c
                JOIN capture_embeddings e ON e.note_id = c.id
                WHERE {where}
                ORDER BY c.updated_at DESC
                LIMIT ?""",
            [*params, limit_notes],
        ).fetchall()
        if len(rows) < 2:
            return {"ok": True, "inserted": 0, "skipped": len(rows), "reason": "not_enough_embedded_notes"}
        ids = [str(r[0]) for r in rows]
        emap: dict[str, list[float]] = {}
        pmap: dict[str, str] = {}
        for rid, eb, dim, model_id, pid in rows:
            if int(dim or 0) != active_dim or str(model_id or "") != active_model:
                continue
            vec = _decode_vault_embedding_vec(eb)
            if not vec or len(vec) != active_dim:
                continue
            emap[str(rid)] = vec
            pmap[str(rid)] = str(pid or "")
        ids = list(emap.keys())
        if len(ids) < 2:
            return {"ok": True, "inserted": 0, "skipped": 0, "reason": "not_enough_homogeneous_embeddings"}
        inserted = 0
        now = _now_iso()
        pairs = 0
        for i, a in enumerate(ids):
            va = emap.get(a) or []
            if not va:
                continue
            for b in ids[i + 1 :]:
                pairs += 1
                if pairs > _VAULT_GRAPH_PROPOSE_PAIR_CAP:
                    break
                vb = emap.get(b) or []
                if not vb:
                    continue
                sim = _cosine_similarity_vault(va, vb)
                if sim < float(min_similarity):
                    continue
                low, high = _canonical_note_pair(a, b)
                eid = str(uuid.uuid4())
                proj = pmap.get(a) or pmap.get(b) or ""
                try:
                    before = conn.total_changes
                    conn.execute(
                        """INSERT OR IGNORE INTO vault_graph_edges
                           (id, note_low, note_high, src_id, dst_id, link_kind, status, confidence, source, rationale, project_id, created_at, updated_at)
                           VALUES (?, ?, ?, ?, ?, 'suggested_similarity', 'suggested', ?, 'system', '', ?, ?, ?)""",
                        (eid, low, high, a, b, float(sim), proj, now, now),
                    )
                    if conn.total_changes > before:
                        inserted += 1
                except sqlite3.IntegrityError:
                    pass
                if inserted >= max_new_edges:
                    break
            if inserted >= max_new_edges:
                break
        conn.commit()
        return {"ok": True, "inserted": inserted, "pairs_scanned": pairs, "notes": len(ids)}
    finally:
        conn.close()


def list_vault_graph_suggestions(
    *,
    project_id: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> list[dict[str, Any]]:
    limit = max(1, min(int(limit or 100), 500))
    offset = max(0, int(offset or 0))
    conn = _get_connection()
    try:
        clauses = ["status = 'suggested'"]
        params: list[Any] = []
        if project_id is not None and str(project_id).strip():
            clauses.append("project_id = ?")
            params.append(str(project_id).strip())
        where = " AND ".join(clauses)
        rows = conn.execute(
            f"""SELECT id, src_id, dst_id, link_kind, confidence, rationale, project_id, created_at
                FROM vault_graph_edges
                WHERE {where}
                ORDER BY updated_at DESC
                LIMIT ? OFFSET ?""",
            [*params, limit, offset],
        ).fetchall()
        out: list[dict[str, Any]] = []
        for rid, sid, did, lk, conf, rat, pid, cat in rows:
            out.append(
                {
                    "edge_id": str(rid),
                    "src_id": str(sid),
                    "dst_id": str(did),
                    "link_kind": str(lk),
                    "confidence": float(conf) if conf is not None else None,
                    "rationale": str(rat or ""),
                    "project_id": str(pid or ""),
                    "created_at": str(cat or ""),
                }
            )
        return out
    finally:
        conn.close()


def confirm_vault_graph_edge(edge_id: str) -> dict[str, Any] | None:
    edge_id = str(edge_id or "").strip()
    if not edge_id:
        return None
    now = _now_iso()
    conn = _get_connection()
    try:
        cur = conn.execute(
            "UPDATE vault_graph_edges SET status = 'active', updated_at = ? WHERE id = ? AND status = 'suggested'",
            (now, edge_id),
        )
        conn.commit()
        if cur.rowcount <= 0:
            return None
        row = conn.execute(
            "SELECT id, src_id, dst_id, link_kind, status, confidence, project_id, created_at, updated_at FROM vault_graph_edges WHERE id = ?",
            (edge_id,),
        ).fetchone()
        if not row:
            return None
        return {
            "edge_id": str(row[0]),
            "src_id": str(row[1]),
            "dst_id": str(row[2]),
            "link_kind": str(row[3]),
            "status": str(row[4]),
            "confidence": float(row[5]) if row[5] is not None else None,
            "project_id": str(row[6] or ""),
            "created_at": str(row[7] or ""),
            "updated_at": str(row[8] or ""),
        }
    finally:
        conn.close()


def reject_vault_graph_edge(edge_id: str) -> dict[str, Any] | None:
    edge_id = str(edge_id or "").strip()
    if not edge_id:
        return None
    now = _now_iso()
    conn = _get_connection()
    try:
        cur = conn.execute(
            "UPDATE vault_graph_edges SET status = 'rejected', updated_at = ? WHERE id = ? AND status IN ('suggested', 'active')",
            (now, edge_id),
        )
        conn.commit()
        if cur.rowcount <= 0:
            return None
        return {"edge_id": edge_id, "status": "rejected"}
    finally:
        conn.close()


def list_vault_graph_mentions(
    target_note_id: str,
    *,
    limit: int = 40,
    scan_limit: int = 400,
    include_unlinked: bool = False,
) -> dict[str, Any]:
    target_note_id = str(target_note_id or "").strip()
    if not target_note_id:
        return {"ok": False, "error": "note_id required", "mentions": []}
    target = get_capture(target_note_id)
    if not target or target.get("deleted_at"):
        return {"ok": False, "error": "note not found", "mentions": []}
    title = str(target.get("title") or "").strip()
    if len(title) < 2:
        return {"ok": True, "mentions": [], "target_title": title}
    wikire = re.compile(r"\[\[" + re.escape(title) + r"(?:\|[^\]\n]+)?\]\]", re.IGNORECASE)
    if len(title) >= 3:
        plainre = re.compile(r"\b" + re.escape(title) + r"\b", re.IGNORECASE)
    else:
        plainre = re.compile(re.escape(title), re.IGNORECASE)
    proj = str(target.get("project_id") or "")
    conn = _get_connection()
    try:
        clauses = ["deleted_at = ''", "id != ?"]
        params: list[Any] = [target_note_id]
        if proj:
            clauses.append("project_id = ?")
            params.append(proj)
        where = " AND ".join(clauses)
        rows = conn.execute(
            _capture_select_sql() + f" WHERE {where} ORDER BY updated_at DESC LIMIT ?",
            [*params, max(1, min(int(scan_limit or 400), 800))],
        ).fetchall()
        mentions: list[dict[str, Any]] = []
        for row in rows:
            cap = _capture_from_row(conn, row)
            sid = str(cap.get("id") or "")
            body = str(cap.get("body") or "")
            if not body:
                continue
            wiki_ranges: list[tuple[int, int]] = []
            for m in wikire.finditer(body):
                start, end = m.start(), m.end()
                wiki_ranges.append((start, end))
                mentions.append(
                    {
                        "source_note_id": sid,
                        "match_kind": "wikilink",
                        "start": start,
                        "end": end,
                        "snippet": _mention_snippet(body, start, end),
                        "snippet_segments": _mention_snippet_segments(body, start, end),
                        "is_wikilink": True,
                        "is_unlinked": False,
                        "_source_updated_at": str(cap.get("updated_at") or ""),
                    }
                )

            def _inside_wikilink(pos: int) -> bool:
                for a, b in wiki_ranges:
                    if a <= pos < b:
                        return True
                return False

            for m in plainre.finditer(body):
                if _inside_wikilink(m.start()):
                    continue
                has_link = conn.execute(
                    "SELECT 1 FROM note_links WHERE src_id = ? AND dst_id = ? AND link_kind = 'wikilink' LIMIT 1",
                    (sid, target_note_id),
                ).fetchone()
                is_unlinked = not bool(has_link)
                if not include_unlinked and is_unlinked:
                    continue
                start, end = m.start(), m.end()
                mentions.append(
                    {
                        "source_note_id": sid,
                        "match_kind": "plain",
                        "start": start,
                        "end": end,
                        "snippet": _mention_snippet(body, start, end),
                        "snippet_segments": _mention_snippet_segments(body, start, end),
                        "is_wikilink": False,
                        "is_unlinked": is_unlinked,
                        "_source_updated_at": str(cap.get("updated_at") or ""),
                    }
                )
            if len(mentions) >= limit * 3:
                break
        mentions.sort(
            key=lambda x: (x.get("_source_updated_at") or "", x["source_note_id"], x["start"]),
            reverse=True,
        )
        out_mentions: list[dict[str, Any]] = []
        for m in mentions[:limit]:
            m2 = {k: v for k, v in m.items() if k != "_source_updated_at"}
            out_mentions.append(m2)
        return {"ok": True, "target_title": title, "mentions": out_mentions}
    finally:
        conn.close()


def _mention_snippet(body: str, start: int, end: int, radius: int = 72) -> str:
    lo = max(0, start - radius)
    hi = min(len(body), end + radius)
    return body[lo:hi].replace("\n", " ").strip()


def _mention_snippet_segments(
    body: str, start: int, end: int, radius: int = 72
) -> list[dict[str, str]]:
    """Build context/match/context segments for clients (Swift) without index translation."""
    lo = max(0, start - radius)
    hi = min(len(body), end + radius)
    chunk = body[lo:hi]
    if not chunk:
        return []
    ms = max(0, min(start - lo, len(chunk)))
    me = max(ms, min(end - lo, len(chunk)))

    def _norm(s: str) -> str:
        return s.replace("\n", " ")

    segs: list[dict[str, str]] = []
    if ms > 0:
        segs.append({"role": "context", "text": _norm(chunk[:ms])})
    segs.append({"role": "match", "text": _norm(chunk[ms:me])})
    if me < len(chunk):
        segs.append({"role": "context", "text": _norm(chunk[me:])})
    if segs:
        segs[0]["text"] = segs[0]["text"].lstrip()
        segs[-1]["text"] = segs[-1]["text"].rstrip()
        return [s for s in segs if s.get("text")]


def link_vault_unlinked_mention(
    source_note_id: str,
    target_note_id: str,
    start: int,
    end: int,
) -> dict[str, Any]:
    """Replace body[start:end] with [[target title]] when it matches an unlinked plain mention of target."""
    source_note_id = str(source_note_id or "").strip()
    target_note_id = str(target_note_id or "").strip()
    if not source_note_id or not target_note_id or source_note_id == target_note_id:
        return {"ok": False, "error": "invalid note ids"}
    src = get_capture(source_note_id)
    tgt = get_capture(target_note_id)
    if not src or src.get("deleted_at") or not tgt or tgt.get("deleted_at"):
        return {"ok": False, "error": "note not found"}
    body = str(src.get("body") or "")
    n = len(body)
    start_i = int(start)
    end_i = int(end)
    if start_i < 0 or end_i > n or start_i >= end_i:
        return {"ok": False, "error": "invalid span"}
    title = str(tgt.get("title") or "").strip()
    if len(title) < 2:
        return {"ok": False, "error": "target title too short"}
    slice_ = body[start_i:end_i]
    norm_slice = " ".join(slice_.split()).lower()
    norm_title = " ".join(title.split()).lower()
    if norm_slice != norm_title:
        return {"ok": False, "error": "span does not match target title"}
    conn = _get_connection()
    try:
        has_link = conn.execute(
            "SELECT 1 FROM note_links WHERE src_id = ? AND dst_id = ? AND link_kind = 'wikilink' LIMIT 1",
            (source_note_id, target_note_id),
        ).fetchone()
    finally:
        conn.close()
    if has_link:
        return {"ok": False, "error": "wikilink already exists"}
    safe_title = title.replace("]]", "").replace("[[", "")
    wikilink = "[[" + safe_title + "]]"
    new_body = body[:start_i] + wikilink + body[end_i:]
    updated = update_capture(source_note_id, body=new_body)
    if not updated:
        return {"ok": False, "error": "update failed"}
    return {"ok": True, "capture": updated}


def export_vault_subgraph(
    focus_note_id: str,
    *,
    depth: int = 2,
    project_id: str | None = None,
    include_suggested: bool = True,
    include_wikilinks: bool = True,
    include_similarity_edges: bool = True,
    max_nodes: int = 150,
) -> dict[str, Any]:
    focus_note_id = str(focus_note_id or "").strip()
    if not focus_note_id:
        return {"ok": False, "error": "focus_note_id required", "nodes": [], "edges": []}
    depth = max(0, min(int(depth or 2), 4))
    max_nodes = max(1, min(int(max_nodes or 150), 400))
    conn = _get_connection()
    try:
        seed = conn.execute(
            _capture_select_sql() + " WHERE id = ? AND deleted_at = ''", (focus_note_id,)
        ).fetchone()
        if not seed:
            return {"ok": False, "error": "note not found", "nodes": [], "edges": []}
        focus_cap = _capture_from_row(conn, seed)
        eff_proj = str(project_id).strip() if project_id is not None else str(focus_cap.get("project_id") or "")

        visited: set[str] = {focus_note_id}
        frontier: set[str] = {focus_note_id}
        edge_sql = (
            "SELECT src_id, dst_id FROM vault_graph_edges WHERE (src_id = ? OR dst_id = ?) AND status != 'rejected'"
            if include_suggested
            else "SELECT src_id, dst_id FROM vault_graph_edges WHERE (src_id = ? OR dst_id = ?) AND status = 'active'"
        )
        for _ in range(depth):
            if len(visited) >= max_nodes:
                break
            nxt: set[str] = set()
            for nid in frontier:
                for dst_id, in conn.execute(
                    "SELECT dst_id FROM note_links WHERE src_id = ? AND link_kind = 'wikilink'", (nid,)
                ).fetchall():
                    d = str(dst_id)
                    if d not in visited:
                        nxt.add(d)
                for src_id, in conn.execute(
                    "SELECT src_id FROM note_links WHERE dst_id = ? AND link_kind = 'wikilink'", (nid,)
                ).fetchall():
                    s = str(src_id)
                    if s not in visited:
                        nxt.add(s)
                for sid, did in conn.execute(edge_sql, (nid, nid)).fetchall():
                    for x in (str(sid), str(did)):
                        if x not in visited:
                            nxt.add(x)
            frontier = set()
            for x in nxt:
                if len(visited) >= max_nodes:
                    break
                if x not in visited:
                    visited.add(x)
                    frontier.add(x)
            if not frontier:
                break

        node_list = list(visited)[:max_nodes]
        node_payloads: list[dict[str, Any]] = []
        for nid in node_list:
            row = conn.execute(_capture_select_sql() + " WHERE id = ?", (nid,)).fetchone()
            if not row:
                continue
            c = _capture_from_row(conn, row)
            # Keep the focus note visible even when the UI sends a project filter that
            # does not match (e.g. note is unfiled while a project is selected).
            if eff_proj and str(c.get("project_id") or "") != eff_proj and nid != focus_note_id:
                continue
            if c.get("deleted_at"):
                continue
            node_payloads.append({"id": nid, "title": c.get("title") or "", "project_id": c.get("project_id") or ""})
        allowed = {n["id"] for n in node_payloads}
        edge_payloads: list[dict[str, Any]] = []
        seen: set[tuple[str, str, str]] = set()
        if include_wikilinks:
            for a, b in conn.execute(
                "SELECT src_id, dst_id FROM note_links WHERE link_kind = 'wikilink'"
            ).fetchall():
                sa, sb = str(a), str(b)
                if sa in allowed and sb in allowed:
                    key = tuple(sorted((sa, sb))) + ("wikilink",)
                    if key not in seen:
                        seen.add(key)
                        edge_payloads.append({"source": sa, "target": sb, "kind": "wikilink", "directed": True})
        if include_similarity_edges:
            qe = "SELECT src_id, dst_id, link_kind, status, confidence FROM vault_graph_edges WHERE status != 'rejected'"
            if not include_suggested:
                qe = qe.replace("status != 'rejected'", "status = 'active'")
            for sid, did, lk, st, conf in conn.execute(qe).fetchall():
                sa, sb = str(sid), str(did)
                if sa in allowed and sb in allowed:
                    key = (sa, sb, str(lk))
                    if key not in seen:
                        seen.add(key)
                        edge_payloads.append(
                            {
                                "source": sa,
                                "target": sb,
                                "kind": str(lk),
                                "status": str(st),
                                "confidence": float(conf) if conf is not None else None,
                                "directed": False,
                            }
                        )
        return {"ok": True, "nodes": node_payloads, "edges": edge_payloads}
    finally:
        conn.close()


def vault_graph_topic_mix(
    focus_note_id: str,
    *,
    depth: int = 2,
    project_id: str | None = None,
    max_nodes: int = 120,
    k_topics: int = 5,
    include_wikilinks: bool = True,
    include_similarity_edges: bool = True,
    include_suggested: bool = True,
) -> dict[str, Any]:
    """Cluster embedded notes in the focus subgraph by cosine similarity to high-degree seeds."""
    k_topics = max(2, min(int(k_topics or 5), 8))
    max_nodes = max(1, min(int(max_nodes or 120), 400))
    sg = export_vault_subgraph(
        focus_note_id,
        depth=depth,
        project_id=project_id,
        include_suggested=include_suggested,
        include_wikilinks=include_wikilinks,
        include_similarity_edges=include_similarity_edges,
        max_nodes=max_nodes,
    )
    if not sg.get("ok"):
        return sg
    nodes = sg.get("nodes") or []
    edges = sg.get("edges") or []
    nid_list = [str(n["id"]) for n in nodes if n.get("id")]
    degree: dict[str, int] = {nid: 0 for nid in nid_list}
    for e in edges:
        s, t = str(e.get("source")), str(e.get("target"))
        if s in degree:
            degree[s] += 1
        if t in degree:
            degree[t] += 1
    id_to_title = {str(n.get("id")): str(n.get("title") or "")[:120] for n in nodes}
    active_model = _vault_embed_model_id()
    active_dim = _vault_embed_dim()
    conn = _get_connection()
    try:
        emap: dict[str, list[float]] = {}
        for nid in nid_list:
            row = conn.execute(
                "SELECT embedding, dim, model_id FROM capture_embeddings WHERE note_id = ?",
                (nid,),
            ).fetchone()
            if not row or not row[0]:
                continue
            if int(row[1] or 0) != active_dim or str(row[2] or "") != active_model:
                continue
            vec = _decode_vault_embedding_vec(row[0])
            if vec and len(vec) == active_dim:
                emap[nid] = vec
    finally:
        conn.close()
    embedded_ranked = sorted(((nid, degree.get(nid, 0)) for nid in emap), key=lambda x: -x[1])
    n_embed = len(embedded_ranked)
    n_sub = len(nid_list)
    if n_embed < 2:
        return {
            "ok": True,
            "mode": "topic_mix",
            "topics": [],
            "notes_with_embeddings": n_embed,
            "notes_in_subgraph": n_sub,
            "degraded": True,
        }
    n_seeds = min(k_topics, n_embed)
    seeds = [embedded_ranked[i][0] for i in range(n_seeds)]
    clusters: dict[str, list[str]] = {s: [] for s in seeds}
    for nid, _deg in embedded_ranked:
        vec = emap[nid]
        best = seeds[0]
        best_sim = -1.0
        for s in seeds:
            sim = _cosine_similarity_vault(vec, emap[s])
            if sim > best_sim:
                best_sim = sim
                best = s
        clusters[best].append(nid)
    topics: list[dict[str, Any]] = []
    for s in seeds:
        cnt = len(clusters[s])
        frac = cnt / n_embed if n_embed else 0.0
        topics.append(
            {
                "seed_id": s,
                "label": id_to_title.get(s, s[:8]),
                "fraction": round(frac, 4),
                "count": cnt,
            }
        )
    topics.sort(key=lambda x: (-float(x.get("fraction") or 0), str(x.get("label") or "")))
    degraded = n_embed < max(3, n_sub // 4) if n_sub else True
    return {
        "ok": True,
        "mode": "topic_mix",
        "topics": topics,
        "notes_with_embeddings": n_embed,
        "notes_in_subgraph": n_sub,
        "degraded": degraded,
    }


def search_vault_captures(
    query: str,
    *,
    project_id: str | None = None,
    limit_notes: int = 40,
    hits_per_note: int = 4,
    scan_limit: int = 500,
) -> dict[str, Any]:
    """Lightweight vault search: substring match with snippet segments (for navigator parity)."""
    q = " ".join(str(query or "").strip().split())
    if len(q) < 1:
        return {"ok": True, "query": "", "results": [], "scan_truncated": False}
    q_low = q.lower()
    scan_limit = max(10, min(int(scan_limit or 500), 800))
    limit_notes = max(1, min(int(limit_notes or 40), 80))
    hits_per_note = max(1, min(int(hits_per_note or 4), 12))
    conn = _get_connection()
    results: list[dict[str, Any]] = []
    try:
        clauses = ["deleted_at = ''"]
        params: list[Any] = []
        if project_id is not None and str(project_id).strip():
            clauses.append("project_id = ?")
            params.append(str(project_id).strip())
        where = " AND ".join(clauses)
        rows = conn.execute(
            _capture_select_sql() + f" WHERE {where} ORDER BY updated_at DESC LIMIT ?",
            [*params, scan_limit + 1],
        ).fetchall()
        scan_truncated = len(rows) > scan_limit
        rows = rows[:scan_limit]
        for row in rows:
            if len(results) >= limit_notes:
                break
            cap = _capture_from_row(conn, row)
            body = str(cap.get("body") or "")
            title = str(cap.get("title") or "")
            if q_low not in body.lower() and q_low not in title.lower():
                continue
            hits: list[dict[str, Any]] = []
            hay = body
            hay_low = hay.lower()
            pos = 0
            while len(hits) < hits_per_note:
                j = hay_low.find(q_low, pos)
                if j < 0:
                    break
                e = j + len(q)
                hits.append(
                    {
                        "start": j,
                        "end": e,
                        "snippet": _mention_snippet(hay, j, e),
                        "snippet_segments": _mention_snippet_segments(hay, j, e),
                    }
                )
                pos = j + max(1, len(q))
            if not hits and q_low in title.lower():
                tlo = title.lower()
                j = tlo.find(q_low)
                e = j + len(q) if j >= 0 else len(title)
                if j >= 0:
                    hits.append(
                        {
                            "start": j,
                            "end": e,
                            "snippet": _mention_snippet(title, j, e, radius=min(72, len(title))),
                            "snippet_segments": _mention_snippet_segments(title, j, e, radius=min(72, len(title))),
                        }
                    )
            if hits:
                results.append(
                    {
                        "note_id": cap["id"],
                        "project_id": cap.get("project_id") or "",
                        "title": title,
                        "updated_at": cap.get("updated_at") or "",
                        "hits": hits,
                    }
                )
    finally:
        conn.close()
    return {"ok": True, "query": q, "results": results, "scan_truncated": scan_truncated}


def vault_graph_insights(
    focus_note_id: str,
    *,
    mode: str = "gaps",
    depth: int = 2,
    project_id: str | None = None,
    max_nodes: int = 120,
    with_focus_context: bool = False,
    include_wikilinks: bool = True,
    include_similarity_edges: bool = True,
) -> dict[str, Any]:
    sg = export_vault_subgraph(
        focus_note_id,
        depth=depth,
        project_id=project_id,
        include_suggested=True,
        include_wikilinks=include_wikilinks,
        include_similarity_edges=include_similarity_edges,
        max_nodes=max_nodes,
    )
    if not sg.get("ok"):
        return sg
    nodes = sg.get("nodes") or []
    edges = sg.get("edges") or []
    focus_ctx: dict[str, Any] | None = None
    if with_focus_context:
        fc = get_capture(focus_note_id)
        if fc:
            fb = str(fc.get("body") or "")
            excerpt = fb.replace("\n", " ").strip()
            if len(excerpt) > 420:
                excerpt = excerpt[:417].rstrip() + "…"
            focus_ctx = {
                "note_id": focus_note_id,
                "title": str(fc.get("title") or ""),
                "excerpt": excerpt,
            }
    degree: dict[str, int] = {str(n.get("id")): 0 for n in nodes if n.get("id")}
    for e in edges:
        s, t = str(e.get("source")), str(e.get("target"))
        if s in degree:
            degree[s] += 1
        if t in degree:
            degree[t] += 1
    mode = (mode or "gaps").lower().strip()
    if mode == "topic_mix":
        mix = vault_graph_topic_mix(
            focus_note_id,
            depth=depth,
            project_id=project_id,
            max_nodes=max_nodes,
            include_wikilinks=include_wikilinks,
            include_similarity_edges=include_similarity_edges,
        )
        if focus_ctx:
            mix["focus_context"] = focus_ctx
        return mix
    if mode == "gaps":
        orphans = sorted([nid for nid, d in degree.items() if d <= 0], key=lambda x: x)[:50]
        low = sorted([(nid, d) for nid, d in degree.items() if 0 < d <= 1], key=lambda x: x[1])[:50]
        out: dict[str, Any] = {
            "ok": True,
            "mode": mode,
            "orphan_note_ids": orphans,
            "low_degree": [{"note_id": nid, "degree": d} for nid, d in low],
        }
        if focus_ctx:
            out["focus_context"] = focus_ctx
        return out
    if mode == "trends" or mode == "topics":
        top = sorted(degree.items(), key=lambda x: -x[1])[:15]
        labels = []
        id_to_title = {str(n.get("id")): str(n.get("title") or "") for n in nodes}
        for nid, d in top:
            labels.append({"note_id": nid, "degree": d, "title": id_to_title.get(nid, "")})
        out = {"ok": True, "mode": mode, "top_connected": labels}
        if focus_ctx:
            out["focus_context"] = focus_ctx
        return out
    if mode == "concepts":
        out = {
            "ok": True,
            "mode": mode,
            "concepts": vault_graph_insights(
                focus_note_id,
                mode="topics",
                depth=depth,
                project_id=project_id,
                max_nodes=max_nodes,
                with_focus_context=False,
                include_wikilinks=include_wikilinks,
                include_similarity_edges=include_similarity_edges,
            ).get("top_connected", []),
        }
        if focus_ctx:
            out["focus_context"] = focus_ctx
        return out
    out = {"ok": True, "mode": mode, "detail": "unknown_mode"}
    if focus_ctx:
        out["focus_context"] = focus_ctx
    return out


def reindex_capture_embeddings_batch(
    *,
    limit: int = 80,
    project_id: str | None = None,
) -> dict[str, Any]:
    limit = max(1, min(int(limit or 80), 500))
    conn = _get_connection()
    try:
        clauses = ["deleted_at = ''"]
        params: list[Any] = []
        if project_id is not None and str(project_id).strip():
            clauses.append("project_id = ?")
            params.append(str(project_id).strip())
        where = " AND ".join(clauses)
        rows = list(
            conn.execute(
                _capture_select_sql() + f" WHERE {where} ORDER BY updated_at DESC LIMIT ?",
                [*params, limit],
            ).fetchall()
        )
    finally:
        conn.close()
    done = 0
    errors = 0
    for row in rows:
        cid = row[0]
        try:
            r = maybe_refresh_capture_embedding(str(cid))
            if r.get("ok") or r.get("skipped") in ("unchanged",):
                done += 1
            else:
                errors += 1
        except Exception:
            errors += 1
    return {"ok": True, "refreshed": done, "errors": errors, "candidates": len(rows)}


__all__ = [
    "get_messages",
    "append_messages",
    "remove_last_message",
    "get_anchor",
    "upsert_anchor",
    "delete_anchor",
    "create_branch",
    "get_branches_of",
    "is_branch",
    "delete_branch_entry",
    "get_parent_of_branch",
    "get_scratchpads",
    "save_scratchpad",
    "update_scratchpad",
    "create_scratchpad",
    "delete_scratchpad",
    "create_capture",
    "get_capture",
    "list_captures",
    "update_capture",
    "soft_delete_capture",
    "restore_capture",
    "file_capture",
    "unfile_capture",
    "tag_capture",
    "process_captures",
    "create_capture_project",
    "list_capture_projects",
    "reorder_capture_projects",
    "update_capture_project",
    "archive_capture_project",
    "delete_capture_project",
    "get_capture_project_summary",
    "generate_capture_connections",
    "list_connections",
    "get_connection_card",
    "update_connection_card",
    "list_action_cards",
    "get_action_card",
    "update_action_card",
    "create_vault_attachment",
    "get_vault_attachment",
    "link_vault_attachment",
    "list_vault_attachments",
    "delete_vault_attachment",
    "orphan_vault_attachment_report",
    "gc_orphan_vault_attachments",
    "update_note_links",
    "list_note_backlinks",
    "maybe_refresh_capture_embedding",
    "find_similar_capture_notes",
    "list_graph_neighbors",
    "propose_vault_similarity_edges",
    "list_vault_graph_suggestions",
    "confirm_vault_graph_edge",
    "reject_vault_graph_edge",
    "list_vault_graph_mentions",
    "link_vault_unlinked_mention",
    "export_vault_subgraph",
    "vault_graph_topic_mix",
    "search_vault_captures",
    "list_recent_vault_notes",
    "list_vault_notes_discussed_since",
    "list_recent_vault_graph_pairs",
    "vault_graph_insights",
    "reindex_capture_embeddings_batch",
    "DB_PATH",
    "verify_backup_readable",
    "import_encryption_key",
]
