mnemo_core.py

"""
Mnemo v7.0 Core Engine — Hybrid SQLite + FAISS + NumPy Architecture

ARCHITECTURE:
  SQLite (WAL mode)  = Source of truth. ACID, indexed, FTS5 full-text search.
  FAISS (disposable) = Fast ANN pre-filter. Rebuilt from SQLite on startup.
                       Never persisted separately. No fragile remove_ids().
  NumPy              = Exact cosine reranking on FAISS candidates.
                       Fixes the accuracy gap from IndexIDMap drift.
  Embeddings         = Stored as raw BLOB in SQLite (not base64 JSON).
                       25% smaller, 10x faster load, zero encode/decode.

SEARCH PIPELINE (graph_search):
  Phase 1: Entity graph lookup    (SQLite indices)      ~1ms
  Phase 2: Full-text search       (FTS5 + BM25)         ~3ms
  Phase 3: Semantic pre-filter    (FAISS ANN)           ~5ms
  Phase 4: Exact reranking        (NumPy cosine)        ~1ms
  Phase 5: Score fusion           (graph + FTS + sem)   ~0ms
  Total:                                                ~10ms

PERSISTENCE:
  SQLite .db file uploaded directly to HF Datasets (no serialize/deserialize).
  WAL checkpoint before upload. Download .db on startup.

BACKWARD COMPATIBILITY:
  All public method signatures identical to v6.5.
  Gradio app.py endpoints require zero changes.
  Streamlit mnemo_client.py requires zero changes.

MIGRATION:
  On first startup, if legacy mnemo_db.json exists, imports it into SQLite.
  After migration, all operations use SQLite exclusively.

v7.0 changes from v6.5:
- REPLACED: Dict storage → SQLite with WAL mode + connection pooling
- REPLACED: EntityIndex class → SQLite COLLATE NOCASE indices
- REPLACED: ThreadIndex class → SQLite join tables (thread_points, knot_threads)
- REPLACED: FAISS IndexIDMap → disposable IndexFlatIP rebuilt from DB
- REPLACED: base64 embedding JSON → raw BLOB in SQLite
- REPLACED: serialize()/deserialize() → direct .db file upload
- ADDED: FTS5 full-text search with Porter stemming + BM25 ranking
- ADDED: NumPy exact reranking (Stage 2 of two-stage retrieval)
- ADDED: SQLite WAL mode for concurrent reads during writes
- ADDED: JSON→SQLite migration on first startup
- KEPT: All public method signatures, dataclass definitions, EmbeddingCache
"""

import os
import re
import json
import time
import queue
import sqlite3
import hashlib
import logging
import threading
import numpy as np
from typing import Dict, List, Optional, Tuple, Set
from dataclasses import dataclass, field
from collections import defaultdict, OrderedDict
from contextlib import contextmanager
from enum import Enum

import requests as http_requests  # renamed to avoid clash with other 'requests'

# SentenceTransformer is optional — Cloudflare Workers AI is the primary encoder
# on Streamlit Cloud (saves ~310MB RAM). SentenceTransformer is used on HF Space.
try:
    from sentence_transformers import SentenceTransformer
    HAS_SENTENCE_TRANSFORMERS = True
except ImportError:
    HAS_SENTENCE_TRANSFORMERS = False

try:
    import faiss
    HAS_FAISS = True
except ImportError:
    HAS_FAISS = False

log = logging.getLogger("mnemo")


# =============================================================================
# EMBEDDING ENCODERS (Cloudflare Workers AI primary, SentenceTransformer fallback)
# =============================================================================

class CloudflareEncoder:
    """Embedding encoder via Cloudflare Workers AI REST API.

    Free tier: 10,000 neurons/day (~10,000 embedding calls/day).
    Model: bge-small-en-v1.5 → 384-dim (same as all-MiniLM-L6-v2).
    Latency: ~200ms per call (edge GPU), cached locally after first call.
    RAM: 0MB (no PyTorch, no model weights).

    Provides same .encode() interface as SentenceTransformer.
    """

    # Available Cloudflare embedding models and their dimensions
    MODELS = {
        "@cf/baai/bge-small-en-v1.5": 384,
        "@cf/baai/bge-base-en-v1.5": 768,
        "@cf/baai/bge-large-en-v1.5": 1024,
    }

    def __init__(self, account_id: str = None, api_token: str = None,
                 model: str = "@cf/baai/bge-small-en-v1.5",
                 timeout: int = 15):
        self.account_id = account_id or os.environ.get("CLOUDFLARE_ACCOUNT_ID", "")
        self.api_token = api_token or os.environ.get("CLOUDFLARE_API_TOKEN", "")
        self.model = model
        self.timeout = timeout
        self._dim = self.MODELS.get(model, 384)
        self._url = (
            f"https://api.cloudflare.com/client/v4/accounts/"
            f"{self.account_id}/ai/run/{self.model}"
        )
        self._available = bool(self.account_id and self.api_token)

    @property
    def available(self) -> bool:
        return self._available

    def get_sentence_embedding_dimension(self) -> int:
        return self._dim

    def encode(self, text, **kwargs) -> np.ndarray:
        """Encode text(s) to embeddings. Accepts str or list of str.

        Returns:
            np.ndarray: (dim,) for single text, (N, dim) for list
        """
        if isinstance(text, str):
            return self._encode_single(text)
        elif isinstance(text, (list, tuple)):
            return self._encode_batch(list(text))
        else:
            return self._encode_single(str(text))

    def _encode_single(self, text: str) -> np.ndarray:
        """Encode a single text string."""
        try:
            resp = http_requests.post(
                self._url,
                headers={
                    "Authorization": f"Bearer {self.api_token}",
                    "Content-Type": "application/json",
                },
                json={"text": [text]},
                timeout=self.timeout,
            )
            if resp.status_code != 200:
                raise RuntimeError(f"Cloudflare API error {resp.status_code}: {resp.text[:200]}")

            data = resp.json()
            vectors = data.get("result", {}).get("data", [])
            if not vectors:
                raise RuntimeError(f"No embeddings in response: {data}")

            return np.array(vectors[0], dtype=np.float32)

        except Exception as e:
            log.error(f"CloudflareEncoder error: {e}")
            raise

    def _encode_batch(self, texts: List[str]) -> np.ndarray:
        """Encode a batch of texts. Cloudflare supports up to 100 per call."""
        all_embeddings = []
        # Cloudflare batch limit is 100
        for i in range(0, len(texts), 100):
            batch = texts[i:i + 100]
            try:
                resp = http_requests.post(
                    self._url,
                    headers={
                        "Authorization": f"Bearer {self.api_token}",
                        "Content-Type": "application/json",
                    },
                    json={"text": batch},
                    timeout=self.timeout * 2,  # Longer timeout for batches
                )
                if resp.status_code != 200:
                    raise RuntimeError(f"Cloudflare API error {resp.status_code}")

                data = resp.json()
                vectors = data.get("result", {}).get("data", [])
                if len(vectors) != len(batch):
                    raise RuntimeError(
                        f"Expected {len(batch)} embeddings, got {len(vectors)}")

                all_embeddings.extend(vectors)

            except Exception as e:
                log.error(f"CloudflareEncoder batch error: {e}")
                raise

        return np.array(all_embeddings, dtype=np.float32)


def create_encoder(config: 'MnemoConfig'):
    """Create the best available encoder. Priority:
    1. Cloudflare Workers AI (free, 0MB RAM, ~200ms, needs API key)
    2. SentenceTransformer (local, ~310MB RAM, ~50ms, needs torch)
    3. Raises error if neither available

    Returns (encoder, embedding_dim, encoder_name)
    """
    # Try Cloudflare first (if credentials exist)
    cf_account = os.environ.get("CLOUDFLARE_ACCOUNT_ID", "")
    cf_token = os.environ.get("CLOUDFLARE_API_TOKEN", "")

    if cf_account and cf_token:
        try:
            encoder = CloudflareEncoder(
                account_id=cf_account, api_token=cf_token,
                model=config.cloudflare_model,
            )
            # Quick validation — encode a test string
            test_emb = encoder.encode("test")
            if test_emb is not None and len(test_emb) > 0:
                dim = encoder.get_sentence_embedding_dimension()
                print(f"[ENCODER] ✅ Cloudflare Workers AI ({config.cloudflare_model}, dim={dim})")
                return encoder, dim, f"cloudflare:{config.cloudflare_model}"
        except Exception as e:
            print(f"[ENCODER] ⚠️  Cloudflare failed: {e}")

    # Fall back to SentenceTransformer
    if HAS_SENTENCE_TRANSFORMERS:
        try:
            print(f"[ENCODER] Loading SentenceTransformer: {config.model_name}")
            encoder = SentenceTransformer(config.model_name)
            dim = encoder.get_sentence_embedding_dimension()
            print(f"[ENCODER] ✅ SentenceTransformer ({config.model_name}, dim={dim})")
            return encoder, dim, f"local:{config.model_name}"
        except Exception as e:
            print(f"[ENCODER] ⚠️  SentenceTransformer failed: {e}")

    raise RuntimeError(
        "No embedding encoder available. Either:\n"
        "  1. Set CLOUDFLARE_ACCOUNT_ID + CLOUDFLARE_API_TOKEN env vars, or\n"
        "  2. pip install sentence-transformers torch"
    )


# =============================================================================
# ENUMS & CONSTANTS (unchanged from v6.5)
# =============================================================================

class MemoryTier(Enum):
    WORKING = "working"
    SEMANTIC = "semantic"
    ARCHIVE = "archive"


class LinkType(Enum):
    DIRECT_REFERENCE = "direct_reference"
    SEMANTIC_SIMILARITY = "semantic_similarity"
    CO_OCCURRENCE = "co_occurrence"
    HIERARCHICAL = "hierarchical"
    TEMPORAL = "temporal"
    CAUSAL = "causal"
    CROSS_DOMAIN = "cross_domain"
    ASSOCIATIVE = "associative"


LINK_PROPERTIES = {
    LinkType.DIRECT_REFERENCE:    {"threshold": 0.85, "base_strength": 0.90, "decay_per_day": 0.005},
    LinkType.SEMANTIC_SIMILARITY: {"threshold": 0.50, "base_strength": 0.75, "decay_per_day": 0.010},
    LinkType.CO_OCCURRENCE:       {"threshold": 0.60, "base_strength": 0.70, "decay_per_day": 0.015},
    LinkType.HIERARCHICAL:        {"threshold": 0.80, "base_strength": 0.85, "decay_per_day": 0.003},
    LinkType.TEMPORAL:            {"threshold": 0.55, "base_strength": 0.65, "decay_per_day": 0.020},
    LinkType.CAUSAL:              {"threshold": 0.75, "base_strength": 0.80, "decay_per_day": 0.005},
    LinkType.CROSS_DOMAIN:        {"threshold": 0.70, "base_strength": 0.65, "decay_per_day": 0.008},
    LinkType.ASSOCIATIVE:         {"threshold": 0.45, "base_strength": 0.60, "decay_per_day": 0.025},
}

HIGH_VALUE_MARKERS = [
    "allergic", "allergy", "prefers", "hates", "loves", "needs",
    "birthday", "deadline", "password", "never", "always", "emergency",
    "name is", "lives in", "works at", "born", "married", "diagnosed",
]

# NER stop words for graph_search entity extraction
_NER_STOP = frozenset({
    'the', 'this', 'that', 'these', 'those', 'what', 'which', 'who', 'whom',
    'where', 'when', 'how', 'why', 'will', 'would', 'could', 'should', 'can',
    'may', 'might', 'shall', 'must', 'about', 'with', 'from', 'into', 'through',
    'during', 'before', 'after', 'between', 'under', 'above', 'does', 'have',
    'has', 'had', 'was', 'were', 'been', 'being', 'are', 'not', 'but', 'and',
    'for', 'nor', 'yet', 'also', 'just', 'very', 'too', 'some', 'any', 'all',
    'each', 'every', 'both', 'few', 'more', 'most', 'other', 'only', 'own',
    'than', 'then', 'now', 'here', 'there', 'tell', 'show', 'give', 'get',
    'find', 'know', 'remember', 'recall', 'write', 'create', 'describe',
    'make', 'help', 'please', 'scene', 'chapter', 'story', 'book',
    'character', 'plot', 'setting', 'like', 'want', 'need', 'think',
    'said', 'says', 'going', 'come', 'came', 'take', 'took', 'keep',
})


# =============================================================================
# DATACLASSES (kept for API compat — used as return types, not storage)
# =============================================================================

@dataclass
class SearchResult:
    id: str
    content: str
    score: float
    tier: str
    semantic_score: float = 0.0
    link_score: float = 0.0
    metadata: Dict = field(default_factory=dict)

    def to_dict(self):
        return {
            "id": self.id, "content": self.content,
            "score": round(self.score, 3), "tier": self.tier,
            "semantic_score": round(self.semantic_score, 3),
            "link_score": round(self.link_score, 3), "metadata": self.metadata,
        }


# =============================================================================
# SQLITE DATABASE LAYER
# =============================================================================

SCHEMA_SQL = """
-- Connection Points (primary structured memory)
CREATE TABLE IF NOT EXISTS connection_points (
    id          TEXT PRIMARY KEY,
    entity      TEXT NOT NULL,
    point_type  TEXT NOT NULL,
    value       TEXT NOT NULL DEFAULT '',
    connects_to TEXT DEFAULT '',
    reason      TEXT DEFAULT '',
    weight      REAL DEFAULT 0.5,
    category    TEXT DEFAULT 'fact',
    session_id  TEXT DEFAULT '',
    source      TEXT DEFAULT 'auto_extract',
    thread_id   TEXT DEFAULT '',
    position    INTEGER DEFAULT -1,
    namespace   TEXT DEFAULT 'default',
    created_at  REAL NOT NULL,
    embedding   BLOB
);

CREATE INDEX IF NOT EXISTS idx_cp_entity      ON connection_points(entity COLLATE NOCASE);
CREATE INDEX IF NOT EXISTS idx_cp_connects    ON connection_points(connects_to COLLATE NOCASE);
CREATE INDEX IF NOT EXISTS idx_cp_category    ON connection_points(category);
CREATE INDEX IF NOT EXISTS idx_cp_point_type  ON connection_points(point_type);
CREATE INDEX IF NOT EXISTS idx_cp_session     ON connection_points(session_id);
CREATE INDEX IF NOT EXISTS idx_cp_source      ON connection_points(source);
CREATE INDEX IF NOT EXISTS idx_cp_ent_type    ON connection_points(entity COLLATE NOCASE, point_type);
CREATE INDEX IF NOT EXISTS idx_cp_ent_cat     ON connection_points(entity COLLATE NOCASE, category);

-- FTS5 full-text index on CPs
CREATE VIRTUAL TABLE IF NOT EXISTS cp_fts USING fts5(
    entity, value, connects_to, reason,
    content='connection_points', content_rowid='rowid',
    tokenize='porter unicode61'
);

-- FTS sync triggers
CREATE TRIGGER IF NOT EXISTS cp_fts_ai AFTER INSERT ON connection_points BEGIN
    INSERT INTO cp_fts(rowid, entity, value, connects_to, reason)
    VALUES (new.rowid, new.entity, new.value, new.connects_to, new.reason);
END;
CREATE TRIGGER IF NOT EXISTS cp_fts_ad AFTER DELETE ON connection_points BEGIN
    INSERT INTO cp_fts(cp_fts, rowid, entity, value, connects_to, reason)
    VALUES ('delete', old.rowid, old.entity, old.value, old.connects_to, old.reason);
END;
CREATE TRIGGER IF NOT EXISTS cp_fts_au AFTER UPDATE ON connection_points BEGIN
    INSERT INTO cp_fts(cp_fts, rowid, entity, value, connects_to, reason)
    VALUES ('delete', old.rowid, old.entity, old.value, old.connects_to, old.reason);
    INSERT INTO cp_fts(rowid, entity, value, connects_to, reason)
    VALUES (new.rowid, new.entity, new.value, new.connects_to, new.reason);
END;

-- Threads
CREATE TABLE IF NOT EXISTS threads (
    id              TEXT PRIMARY KEY,
    name            TEXT NOT NULL,
    thread_type     TEXT DEFAULT 'plot_line',
    entity          TEXT DEFAULT '',
    status          TEXT DEFAULT 'active',
    tension_level   REAL DEFAULT 0.5,
    tone_trajectory TEXT DEFAULT '[]',
    current_position INTEGER DEFAULT 0,
    session_id      TEXT DEFAULT '',
    created_at      REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_thread_entity ON threads(entity COLLATE NOCASE);
CREATE INDEX IF NOT EXISTS idx_thread_status ON threads(status);
CREATE INDEX IF NOT EXISTS idx_thread_session ON threads(session_id);

-- Thread ↔ CP ordered membership
CREATE TABLE IF NOT EXISTS thread_points (
    thread_id TEXT NOT NULL,
    cp_id     TEXT NOT NULL,
    position  INTEGER NOT NULL,
    PRIMARY KEY (thread_id, cp_id)
);
CREATE INDEX IF NOT EXISTS idx_tp_thread ON thread_points(thread_id, position);
CREATE INDEX IF NOT EXISTS idx_tp_cp     ON thread_points(cp_id);

-- Knots
CREATE TABLE IF NOT EXISTS knots (
    id               TEXT PRIMARY KEY,
    name             TEXT NOT NULL,
    pivot_type       TEXT DEFAULT 'collision',
    narrative_weight REAL DEFAULT 0.5,
    tension_before   REAL DEFAULT 0.5,
    tension_after    REAL DEFAULT 0.5,
    tone_shift       TEXT DEFAULT '',
    reason           TEXT DEFAULT '',
    unresolved       TEXT DEFAULT '[]',
    active_points    TEXT DEFAULT '{}',
    session_id       TEXT DEFAULT '',
    created_at       REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_knot_session ON knots(session_id);

-- Knot ↔ Thread membership
CREATE TABLE IF NOT EXISTS knot_threads (
    knot_id   TEXT NOT NULL,
    thread_id TEXT NOT NULL,
    PRIMARY KEY (knot_id, thread_id)
);

-- Legacy blob memories
CREATE TABLE IF NOT EXISTS memories (
    id             TEXT PRIMARY KEY,
    content        TEXT NOT NULL,
    tier           TEXT DEFAULT 'semantic',
    namespace      TEXT DEFAULT 'default',
    quality_score  REAL DEFAULT 0.5,
    access_count   INTEGER DEFAULT 0,
    priority       REAL DEFAULT 1.0,
    session_id     TEXT DEFAULT '',
    source         TEXT DEFAULT 'auto_extract',
    created_at     REAL NOT NULL,
    last_accessed  REAL NOT NULL,
    metadata       TEXT DEFAULT '{}',
    embedding      BLOB
);
CREATE INDEX IF NOT EXISTS idx_mem_ns ON memories(namespace);
CREATE INDEX IF NOT EXISTS idx_mem_tier ON memories(tier);
-- NOTE: idx_mem_session and idx_mem_source are created by _migrate_memories_columns()
-- to avoid errors on existing databases that lack those columns during schema init.

-- FTS for blob memories
CREATE VIRTUAL TABLE IF NOT EXISTS memory_fts USING fts5(
    content,
    content='memories', content_rowid='rowid',
    tokenize='porter unicode61'
);
CREATE TRIGGER IF NOT EXISTS mem_fts_ai AFTER INSERT ON memories BEGIN
    INSERT INTO memory_fts(rowid, content) VALUES (new.rowid, new.content);
END;
CREATE TRIGGER IF NOT EXISTS mem_fts_ad AFTER DELETE ON memories BEGIN
    INSERT INTO memory_fts(memory_fts, rowid, content) VALUES ('delete', old.rowid, old.content);
END;

-- Neural links
CREATE TABLE IF NOT EXISTS neural_links (
    id              TEXT PRIMARY KEY,
    source_id       TEXT NOT NULL,
    target_id       TEXT NOT NULL,
    link_type       TEXT NOT NULL,
    strength        REAL NOT NULL,
    created_at      REAL NOT NULL,
    last_maintained REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_link_source ON neural_links(source_id);
CREATE INDEX IF NOT EXISTS idx_link_target ON neural_links(target_id);

-- Stats
CREATE TABLE IF NOT EXISTS engine_stats (
    key   TEXT PRIMARY KEY,
    value INTEGER DEFAULT 0
);
"""


class MnemoDB:
    """SQLite database with WAL mode and read connection pooling."""

    def __init__(self, db_path: str):
        self.db_path = db_path
        os.makedirs(os.path.dirname(db_path) if os.path.dirname(db_path) else ".", exist_ok=True)

        self._write_conn = self._create_connection(readonly=False)
        self._write_conn.executescript(SCHEMA_SQL)
        self._write_conn.commit()

        # Schema migration: add session_id/source columns to memories table
        # (for existing .db files created before v7.1)
        self._migrate_memories_columns()

        self._read_pool: queue.Queue = queue.Queue(maxsize=4)
        for _ in range(4):
            self._read_pool.put(self._create_connection(readonly=True))

    def _migrate_memories_columns(self):
        """Add session_id and source columns to memories if missing (v7.0→v7.1).

        Also backfills existing rows by extracting values from the metadata JSON blob.
        """
        cursor = self._write_conn.execute("PRAGMA table_info(memories)")
        existing_cols = {row[1] for row in cursor.fetchall()}

        if "session_id" not in existing_cols:
            log.info("Migrating memories table: adding session_id column")
            print("[MIGRATE] Adding session_id column to memories table")
            self._write_conn.execute("ALTER TABLE memories ADD COLUMN session_id TEXT DEFAULT ''")
            # Backfill from metadata JSON
            self._write_conn.execute("""
                UPDATE memories SET session_id = COALESCE(json_extract(metadata, '$.session_id'), '')
                WHERE metadata LIKE '%session_id%'
            """)

        if "source" not in existing_cols:
            log.info("Migrating memories table: adding source column")
            print("[MIGRATE] Adding source column to memories table")
            self._write_conn.execute("ALTER TABLE memories ADD COLUMN source TEXT DEFAULT 'auto_extract'")
            # Backfill from metadata JSON
            self._write_conn.execute("""
                UPDATE memories SET source = COALESCE(json_extract(metadata, '$.source'), 'auto_extract')
                WHERE metadata LIKE '%source%'
            """)

        # Create indexes if they don't exist (idempotent)
        self._write_conn.execute("CREATE INDEX IF NOT EXISTS idx_mem_session ON memories(session_id)")
        self._write_conn.execute("CREATE INDEX IF NOT EXISTS idx_mem_source ON memories(source)")
        self._write_conn.commit()

    def _create_connection(self, readonly: bool = False) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path, check_same_thread=False, timeout=30)
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute("PRAGMA synchronous=NORMAL")
        conn.execute("PRAGMA cache_size=-32000")     # 32MB cache
        conn.execute("PRAGMA mmap_size=134217728")    # 128MB mmap
        conn.execute("PRAGMA temp_store=MEMORY")
        conn.row_factory = sqlite3.Row
        if readonly:
            conn.execute("PRAGMA query_only=ON")
        return conn

    @contextmanager
    def read(self):
        """Get a read-only connection from the pool."""
        conn = self._read_pool.get(timeout=10)
        try:
            yield conn
        finally:
            self._read_pool.put(conn)

    @contextmanager
    def write(self):
        """Get the write connection. Auto-commits on success, rolls back on error."""
        try:
            yield self._write_conn
            self._write_conn.commit()
        except Exception:
            self._write_conn.rollback()
            raise

    def checkpoint(self):
        """WAL checkpoint — call before uploading .db file."""
        self._write_conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")

    def close(self):
        self._write_conn.close()
        while not self._read_pool.empty():
            try:
                self._read_pool.get_nowait().close()
            except queue.Empty:
                break


# =============================================================================
# DISPOSABLE FAISS INDEX (rebuilt from SQLite, never persisted)
# =============================================================================

class FAISSIndex:
    """Fast ANN pre-filter. Rebuilt from SQLite on startup.

    No IndexIDMap — uses positional id_map list instead.
    Deletions mark dirty for eventual rebuild (no fragile remove_ids).
    """

    def __init__(self, dim: int = 384):
        self.dim = dim
        self._index: Optional[object] = None
        self._id_map: List[str] = []        # position → cp_id
        self._id_to_pos: Dict[str, int] = {}
        self._lock = threading.RLock()
        self._dirty = False

    def rebuild_from_db(self, db: MnemoDB):
        """Full rebuild from SQLite. ~200ms for 10K vectors."""
        with db.read() as conn:
            rows = conn.execute(
                "SELECT id, embedding FROM connection_points WHERE embedding IS NOT NULL"
            ).fetchall()

        if not rows or not HAS_FAISS:
            with self._lock:
                if HAS_FAISS:
                    self._index = faiss.IndexFlatIP(self.dim)
                self._id_map = []
                self._id_to_pos = {}
                self._dirty = False
            return

        ids = [r["id"] for r in rows]
        vecs = np.stack([np.frombuffer(r["embedding"], dtype=np.float32) for r in rows])

        norms = np.linalg.norm(vecs, axis=1, keepdims=True)
        norms[norms == 0] = 1e-10
        vecs_normed = (vecs / norms).astype(np.float32)

        index = faiss.IndexFlatIP(self.dim)
        index.add(vecs_normed)

        with self._lock:
            self._index = index
            self._id_map = ids
            self._id_to_pos = {cid: i for i, cid in enumerate(ids)}
            self._dirty = False
        log.info(f"FAISS rebuilt: {len(ids)} vectors indexed")

    def add(self, cp_id: str, embedding: np.ndarray):
        """Add single vector between rebuilds."""
        if not HAS_FAISS:
            return
        vec = embedding.astype(np.float32)
        norm = np.linalg.norm(vec)
        if norm > 0:
            vec = vec / norm
        with self._lock:
            if self._index is None:
                self._index = faiss.IndexFlatIP(self.dim)
            self._index.add(vec.reshape(1, -1))
            pos = len(self._id_map)
            self._id_map.append(cp_id)
            self._id_to_pos[cp_id] = pos

    def search(self, query_emb: np.ndarray, top_k: int = 200) -> List[Tuple[str, float]]:
        """Fast ANN candidate retrieval. Returns (cp_id, score) pairs."""
        with self._lock:
            if not HAS_FAISS or self._index is None or self._index.ntotal == 0:
                return []
            vec = query_emb.astype(np.float32)
            norm = np.linalg.norm(vec)
            if norm > 0:
                vec = vec / norm
            k = min(top_k, self._index.ntotal)
            scores, indices = self._index.search(vec.reshape(1, -1), k)
            results = []
            for score, idx in zip(scores[0], indices[0]):
                if 0 <= idx < len(self._id_map):
                    results.append((self._id_map[idx], float(score)))
            return results

    def mark_dirty(self):
        self._dirty = True

    @property
    def needs_rebuild(self) -> bool:
        return self._dirty

    @property
    def total(self) -> int:
        with self._lock:
            return self._index.ntotal if (HAS_FAISS and self._index) else 0


class FAISSBlobIndex:
    """Separate FAISS index for legacy blob memories."""

    def __init__(self, dim: int = 384):
        self.dim = dim
        self._index: Optional[object] = None
        self._id_map: List[str] = []
        self._lock = threading.RLock()

    def rebuild_from_db(self, db: MnemoDB):
        with db.read() as conn:
            rows = conn.execute(
                "SELECT id, embedding FROM memories WHERE embedding IS NOT NULL"
            ).fetchall()
        if not rows or not HAS_FAISS:
            with self._lock:
                if HAS_FAISS:
                    self._index = faiss.IndexFlatIP(self.dim)
                self._id_map = []
            return
        ids = [r["id"] for r in rows]
        vecs = np.stack([np.frombuffer(r["embedding"], dtype=np.float32) for r in rows])
        norms = np.linalg.norm(vecs, axis=1, keepdims=True)
        norms[norms == 0] = 1e-10
        vecs_normed = (vecs / norms).astype(np.float32)
        index = faiss.IndexFlatIP(self.dim)
        index.add(vecs_normed)
        with self._lock:
            self._index = index
            self._id_map = ids

    def search(self, query_emb: np.ndarray, top_k: int = 50) -> Dict[str, float]:
        with self._lock:
            if not HAS_FAISS or self._index is None or self._index.ntotal == 0:
                return {}
            vec = query_emb.astype(np.float32)
            norm = np.linalg.norm(vec)
            if norm > 0:
                vec = vec / norm
            k = min(top_k, self._index.ntotal)
            scores, indices = self._index.search(vec.reshape(1, -1), k)
            result = {}
            for score, idx in zip(scores[0], indices[0]):
                if 0 <= idx < len(self._id_map):
                    result[self._id_map[idx]] = float(score)
            return result

    def add(self, mem_id: str, embedding: np.ndarray):
        if not HAS_FAISS:
            return
        vec = embedding.astype(np.float32)
        norm = np.linalg.norm(vec)
        if norm > 0:
            vec = vec / norm
        with self._lock:
            if self._index is None:
                self._index = faiss.IndexFlatIP(self.dim)
            self._index.add(vec.reshape(1, -1))
            self._id_map.append(mem_id)

    def mark_dirty(self):
        pass  # Blob index doesn't auto-rebuild; only on startup


# =============================================================================
# NUMPY RERANKER (exact cosine similarity on FAISS candidates)
# =============================================================================

class NumpyReranker:
    """Exact cosine reranking for FAISS candidates.

    FAISS pre-filters 10K → 200 candidates. NumPy re-ranks with perfect
    accuracy. This fixes the precision gap from IndexIDMap fragmentation.
    """

    @staticmethod
    def rerank(query_emb: np.ndarray, candidate_ids: List[str],
               candidate_embs: np.ndarray, threshold: float = 0.15
               ) -> List[Tuple[str, float]]:
        """Exact cosine similarity on candidate set.

        Args:
            query_emb: (dim,) query vector
            candidate_ids: list of cp_ids matching candidate_embs rows
            candidate_embs: (N, dim) embedding matrix
            threshold: minimum similarity to include

        Returns:
            List of (cp_id, exact_cosine_score), sorted descending
        """
        if len(candidate_ids) == 0:
            return []
        q_norm = np.linalg.norm(query_emb)
        if q_norm == 0:
            return []
        e_norms = np.linalg.norm(candidate_embs, axis=1)
        dots = candidate_embs @ query_emb
        valid = e_norms > 0
        scores = np.zeros(len(candidate_ids))
        scores[valid] = dots[valid] / (q_norm * e_norms[valid])

        ranked = np.argsort(scores)[::-1]
        results = []
        for idx in ranked:
            s = float(scores[idx])
            if s < threshold:
                break
            results.append((candidate_ids[idx], s))
        return results


# =============================================================================
# EMBEDDING CACHE (Thread-Safe, kept from v6.5)
# =============================================================================

class EmbeddingCache:
    """Thread-safe LRU cache for sentence embeddings."""

    def __init__(self, encoder, max_size: int = 500):
        self.encoder = encoder
        self._cache: OrderedDict = OrderedDict()
        self._max_size = max_size
        self._hits = 0
        self._misses = 0
        self._lock = threading.Lock()

    def get_embedding(self, text: str) -> np.ndarray:
        key = text[:200].lower().strip()
        with self._lock:
            if key in self._cache:
                self._cache.move_to_end(key)
                self._hits += 1
                return self._cache[key]
            emb = self.encoder.encode(text)
            self._cache[key] = emb
            if len(self._cache) > self._max_size:
                self._cache.popitem(last=False)
            self._misses += 1
        return emb

    def encode_batch(self, texts: List[str]) -> np.ndarray:
        """Batch encode (for bulk imports). Returns (N, dim) matrix."""
        with self._lock:
            return self.encoder.encode(texts)

    def get_stats(self) -> dict:
        with self._lock:
            total = self._hits + self._misses
            return {
                "hits": self._hits, "misses": self._misses,
                "hit_rate": round(self._hits / max(total, 1), 3),
                "size": len(self._cache), "max_size": self._max_size,
            }

    def clear(self):
        with self._lock:
            self._cache.clear()


# =============================================================================
# CONFIG
# =============================================================================

@dataclass
class MnemoConfig:
    similarity_threshold: float = 0.25
    quality_threshold: float = 0.35
    decay_rate_per_day: float = 0.01
    prune_quality_floor: float = 0.15
    prune_age_days: float = 30.0
    promote_to_working_accesses: int = 10
    demote_to_archive_days: float = 14.0
    model_name: str = "all-MiniLM-L6-v2"
    cloudflare_model: str = "@cf/baai/bge-small-en-v1.5"
    use_faiss: bool = True
    db_path: str = "/app/data/mnemo.db"


# =============================================================================
# MNEMO ENGINE (SQLite-backed, same public API as v6.5)
# =============================================================================

class MnemoEngine:
    """Core memory engine. All state in SQLite. FAISS is disposable cache."""

    def __init__(self, config: MnemoConfig = None):
        self.config = config or MnemoConfig()

        # SQLite
        self.db = MnemoDB(self.config.db_path)

        # Encoder (Cloudflare → SentenceTransformer → error)
        self.encoder, self._embedding_dim, self._encoder_name = create_encoder(self.config)
        self._emb_cache = EmbeddingCache(self.encoder, max_size=500)

        # FAISS indices (disposable, rebuilt from DB)
        self._cp_faiss = FAISSIndex(dim=self._embedding_dim)
        self._blob_faiss = FAISSBlobIndex(dim=self._embedding_dim)
        if self.config.use_faiss and HAS_FAISS:
            self._cp_faiss.rebuild_from_db(self.db)
            self._blob_faiss.rebuild_from_db(self.db)
            print("FAISS indices rebuilt from SQLite.")
        elif self.config.use_faiss and not HAS_FAISS:
            print("FAISS not installed; using numpy fallback.")

        # Stats cache (loaded from DB, flushed periodically)
        self._stats = self._load_stats()
        self._dirty = False
        self._lock = threading.RLock()  # Only for stats + dirty flag now

    # =========================================================================
    # INTERNAL HELPERS
    # =========================================================================

    def _get_embedding(self, text: str) -> np.ndarray:
        return self._emb_cache.get_embedding(text)

    def _generate_id(self, content: str, namespace: str = "default") -> str:
        return "mem_" + hashlib.sha256((content + namespace).encode()).hexdigest()[:16]

    def _generate_cp_id(self, entity: str, point_type: str, value: str) -> str:
        raw = f"{entity}:{point_type}:{value}".lower()
        return "cp_" + hashlib.sha256(raw.encode()).hexdigest()[:12]

    def _cp_to_searchable(self, entity: str, point_type: str, value: str,
                          connects_to: str, reason: str, category: str) -> str:
        parts = [f"[{category.upper()}]", entity]
        if connects_to:
            parts.append(f"{point_type} {connects_to}")
        else:
            parts.append(point_type)
        if value:
            parts.append(value)
        if reason:
            parts.append(reason)
        return " | ".join(parts)

    def _row_to_cp_dict(self, row) -> dict:
        return {
            "id": row["id"], "entity": row["entity"], "point_type": row["point_type"],
            "connects_to": row["connects_to"], "value": row["value"],
            "reason": row["reason"], "weight": round(row["weight"], 3),
            "category": row["category"], "session_id": row["session_id"],
            "source": row["source"], "thread_id": row["thread_id"],
            "position": row["position"], "namespace": row["namespace"],
            "created_at": row["created_at"],
        }

    def _load_stats(self) -> dict:
        defaults = {
            "adds": 0, "adds_rejected": 0, "searches": 0,
            "links_created": 0, "links_decayed": 0, "links_pruned": 0,
            "inject_recommended": 0, "skip_recommended": 0,
            "decayed": 0, "pruned": 0,
            "tier_promotions": 0, "tier_demotions": 0,
            "points_added": 0, "graph_searches": 0,
            "threads_created": 0, "knots_created": 0,
            "temporal_links_skipped": 0,
        }
        try:
            with self.db.read() as conn:
                rows = conn.execute("SELECT key, value FROM engine_stats").fetchall()
                for r in rows:
                    if r["key"] in defaults:
                        defaults[r["key"]] = r["value"]
        except Exception:
            pass
        return defaults

    def _flush_stats(self):
        try:
            with self.db.write() as conn:
                for key, value in self._stats.items():
                    conn.execute(
                        "INSERT OR REPLACE INTO engine_stats(key, value) VALUES (?, ?)",
                        (key, value))
        except Exception as e:
            log.warning(f"Stats flush failed: {e}")

    def _incr_stat(self, key: str, delta: int = 1):
        with self._lock:
            self._stats[key] = self._stats.get(key, 0) + delta

    def _extract_entities(self, query: str) -> Tuple[Set[str], Set[str]]:
        """Three-signal entity extraction (from v6.4).

        Returns (single_word_entities, multi_word_entities).
        """
        # Signal 1: Title-case words
        candidates = set(re.findall(r'\b[A-Z][a-z]{2,}\b', query))

        # Signal 2: Known entity matching (case-insensitive)
        query_words_lower = {w.lower() for w in re.findall(r"\b\w{3,}\b", query)} - _NER_STOP
        with self.db.read() as conn:
            known_entities = {r["entity"].lower() for r in
                              conn.execute("SELECT DISTINCT entity FROM connection_points").fetchall()}
        for qw in query_words_lower:
            if qw in known_entities:
                candidates.add(qw.title())

        # Signal 3: Multi-word entity detection
        query_lower = query.lower()
        multi_word = set()
        for ent in known_entities:
            if ' ' in ent and ent in query_lower:
                multi_word.add(ent)

        # Consecutive capitalized words
        words = query.split()
        i = 0
        while i < len(words):
            if re.match(r'^[A-Z][a-z]{2,}$', words[i]):
                parts = [words[i]]
                j = i + 1
                while j < len(words) and re.match(r'^[A-Z][a-z]{2,}$', words[j]):
                    parts.append(words[j])
                    j += 1
                if len(parts) >= 2:
                    multi_word.add(" ".join(parts).lower())
                i = j
            else:
                i += 1

        return candidates, multi_word

    # =========================================================================
    # CONNECTION POINT OPERATIONS
    # =========================================================================

    def add_point(self, entity: str, point_type: str, value: str = "",
                  connects_to: str = "", reason: str = "", weight: float = 0.5,
                  category: str = "fact", session_id: str = "",
                  source: str = "auto_extract",
                  thread_id: str = "", position: int = -1,
                  namespace: str = "default") -> Optional[str]:
        cp_id = self._generate_cp_id(entity, point_type, value)

        # Upsert check
        with self.db.read() as conn:
            existing = conn.execute("SELECT id, weight, reason FROM connection_points WHERE id = ?",
                                    (cp_id,)).fetchone()
        if existing:
            with self.db.write() as conn:
                conn.execute(
                    "UPDATE connection_points SET weight = MAX(weight, ?), reason = COALESCE(NULLIF(reason,''), ?) WHERE id = ?",
                    (weight, reason, cp_id))
            return cp_id

        # Compute embedding outside DB lock
        searchable = self._cp_to_searchable(entity, point_type, value, connects_to, reason, category)
        embedding = self._get_embedding(searchable)
        emb_blob = embedding.astype(np.float32).tobytes()

        with self.db.write() as conn:
            conn.execute("""
                INSERT INTO connection_points
                (id, entity, point_type, value, connects_to, reason, weight,
                 category, session_id, source, thread_id, position, namespace, created_at, embedding)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (cp_id, entity, point_type, value, connects_to, reason, weight,
                  category, session_id, source, thread_id, position, namespace,
                  time.time(), emb_blob))
        if thread_id:
            conn.execute(
                "INSERT OR REPLACE INTO thread_points (thread_id, cp_id, position) VALUES (?, ?, ?)",
                (thread_id, cp_id, position))

        # Update FAISS
        self._cp_faiss.add(cp_id, embedding)
        self._incr_stat("points_added")
        self._dirty = True
        return cp_id

    def add_points_batch(self, points: List[dict]) -> List[Optional[str]]:
        return [self.add_point(**p) for p in points]

    def get_point(self, cp_id: str) -> Optional[dict]:
        with self.db.read() as conn:
            row = conn.execute(
                "SELECT * FROM connection_points WHERE id = ?", (cp_id,)
            ).fetchone()
        if row:
            return self._row_to_cp_dict(row)
        return None

    def delete_point(self, cp_id: str) -> bool:
        with self.db.read() as conn:
            row = conn.execute("SELECT id FROM connection_points WHERE id = ?", (cp_id,)).fetchone()
        if not row:
            return False
        with self.db.write() as conn:
            conn.execute("DELETE FROM thread_points WHERE cp_id = ?", (cp_id,))
            conn.execute("DELETE FROM connection_points WHERE id = ?", (cp_id,))
        self._cp_faiss.mark_dirty()
        self._dirty = True
        return True

    def update_point(self, cp_id: str, entity: str = None, value: str = None,
                     connects_to: str = None, reason: str = None,
                     weight: float = None, category: str = None,
                     point_type: str = None) -> Optional[dict]:
        """Update fields on an existing ConnectionPoint. Returns updated CP dict.

        Only non-None fields are updated. Re-embeds if any text field changes.
        """
        with self.db.read() as conn:
            row = conn.execute("SELECT * FROM connection_points WHERE id = ?", (cp_id,)).fetchone()
        if not row:
            return None

        # Determine which fields changed
        updates = {}
        if entity is not None and entity != row["entity"]:
            updates["entity"] = entity
        if value is not None and value != row["value"]:
            updates["value"] = value
        if connects_to is not None and connects_to != row["connects_to"]:
            updates["connects_to"] = connects_to
        if reason is not None and reason != row["reason"]:
            updates["reason"] = reason
        if weight is not None and abs(weight - row["weight"]) > 0.001:
            updates["weight"] = weight
        if category is not None and category != row["category"]:
            updates["category"] = category
        if point_type is not None and point_type != row["point_type"]:
            updates["point_type"] = point_type

        if not updates:
            return self._row_to_cp_dict(row)  # Nothing changed

        # Build SET clause
        set_parts = []
        params = []
        for col, val in updates.items():
            set_parts.append(f"{col} = ?")
            params.append(val)

        # Re-embed if any text field changed
        text_fields = {"entity", "value", "connects_to", "reason", "category", "point_type"}
        if updates.keys() & text_fields:
            new_entity = updates.get("entity", row["entity"])
            new_pt = updates.get("point_type", row["point_type"])
            new_val = updates.get("value", row["value"])
            new_conn = updates.get("connects_to", row["connects_to"])
            new_reason = updates.get("reason", row["reason"])
            new_cat = updates.get("category", row["category"])

            searchable = self._cp_to_searchable(new_entity, new_pt, new_val, new_conn, new_reason, new_cat)
            embedding = self._get_embedding(searchable)
            emb_blob = embedding.astype(np.float32).tobytes()
            set_parts.append("embedding = ?")
            params.append(emb_blob)
            re_embedded = True
        else:
            re_embedded = False

        params.append(cp_id)
        with self.db.write() as conn:
            conn.execute(
                f"UPDATE connection_points SET {', '.join(set_parts)} WHERE id = ?",
                params)

        if re_embedded:
            self._cp_faiss.mark_dirty()  # Will rebuild with new embedding

        self._dirty = True
        return self.get_point(cp_id)

    def delete_session_points(self, session_id: str) -> int:
        """Delete all non-protected CPs and blobs for a session."""
        PROTECTED = ("file_upload", "manual_correction", "consolidation", "manual")
        placeholders = ",".join("?" * len(PROTECTED))

        with self.db.write() as conn:
            # Delete CPs
            conn.execute(f"""
                DELETE FROM thread_points WHERE cp_id IN (
                    SELECT id FROM connection_points
                    WHERE session_id = ? AND source NOT IN ({placeholders})
                )""", (session_id, *PROTECTED))
            cur = conn.execute(f"""
                DELETE FROM connection_points
                WHERE session_id = ? AND source NOT IN ({placeholders})
            """, (session_id, *PROTECTED))
            cp_deleted = cur.rowcount

            # Delete blobs (using proper columns now, not json_extract)
            cur2 = conn.execute(f"""
                DELETE FROM memories
                WHERE session_id = ? AND source NOT IN ({placeholders})
            """, (session_id, *PROTECTED))
            blob_deleted = cur2.rowcount

        total = cp_deleted + blob_deleted
        if total > 0:
            self._cp_faiss.mark_dirty()
            self._dirty = True
        return total

    def list_points(self, limit: int = 200) -> List[dict]:
        """List all connection points (for UI display)."""
        with self.db.read() as conn:
            rows = conn.execute(
                "SELECT * FROM connection_points ORDER BY created_at DESC LIMIT ?",
                (limit,)).fetchall()
        return [self._row_to_cp_dict(r) for r in rows]

    def entity_lookup(self, entity: str) -> List[dict]:
        """Look up all CPs for an entity."""
        with self.db.read() as conn:
            rows = conn.execute(
                """SELECT * FROM connection_points
                   WHERE entity = ? COLLATE NOCASE OR connects_to = ? COLLATE NOCASE""",
                (entity, entity)).fetchall()
        return [self._row_to_cp_dict(r) for r in rows]

    # =========================================================================
    # GRAPH SEARCH (hybrid: SQLite + FTS5 + FAISS + NumPy)
    # =========================================================================

    def graph_search(self, query: str, top_k: int = 15,
                     active_sessions: Optional[List[str]] = None) -> List[dict]:
        """v7.0 hybrid search pipeline.

        Phase 1: Entity graph (SQLite indices)    ~1ms
        Phase 2: Full-text search (FTS5 BM25)     ~3ms
        Phase 3: Semantic pre-filter (FAISS)       ~5ms
        Phase 4: Exact reranking (NumPy)           ~1ms
        Phase 5: Score fusion                      ~0ms
        """
        self._incr_stat("graph_searches")

        # Compute query embedding (cached: ~0ms, uncached: ~50ms)
        query_emb = self._get_embedding(query)

        # --- PHASE 1: Entity graph lookup via SQLite ---
        entities, multi_word = self._extract_entities(query)
        graph_scores: Dict[str, float] = {}

        session_filter = ""
        session_params: list = []
        if active_sessions:
            placeholders = ",".join("?" * len(active_sessions))
            session_filter = f" AND (session_id = '' OR session_id IN ({placeholders}))"
            session_params = list(active_sessions)

        with self.db.read() as conn:
            # Direct entity hits
            for ent in entities:
                rows = conn.execute(
                    f"SELECT id FROM connection_points WHERE entity = ? COLLATE NOCASE{session_filter}",
                    [ent] + session_params).fetchall()
                for r in rows:
                    graph_scores[r["id"]] = graph_scores.get(r["id"], 0) + 0.5

                # 1-hop connection expansion
                rows = conn.execute(
                    f"SELECT id FROM connection_points WHERE connects_to = ? COLLATE NOCASE{session_filter}",
                    [ent] + session_params).fetchall()
                for r in rows:
                    graph_scores[r["id"]] = graph_scores.get(r["id"], 0) + 0.4

            # Multi-word entity hits
            for mw in multi_word:
                rows = conn.execute(
                    f"SELECT id FROM connection_points WHERE entity = ? COLLATE NOCASE{session_filter}",
                    [mw] + session_params).fetchall()
                for r in rows:
                    graph_scores[r["id"]] = graph_scores.get(r["id"], 0) + 0.5
                rows = conn.execute(
                    f"SELECT id FROM connection_points WHERE connects_to = ? COLLATE NOCASE{session_filter}",
                    [mw] + session_params).fetchall()
                for r in rows:
                    graph_scores[r["id"]] = graph_scores.get(r["id"], 0) + 0.4

            # Type-keyword boosting
            type_keywords = {
                "relationship": ["relationship", "brother", "sister", "friend", "rival",
                                 "married", "mentor", "captor", "ally", "between"],
                "fears": ["fear", "fears", "afraid", "terrified", "dread"],
                "tone": ["tone", "mood", "atmosphere", "register", "feeling"],
                "secret": ["secret", "hidden", "private", "unknown"],
                "motivation": ["motivation", "drive", "wants", "desires", "goal"],
                "plot": ["plot", "event", "arc", "storyline", "happened", "when", "timeline", "book"],
            }
            ql = query.lower()
            for pt, keywords in type_keywords.items():
                if any(kw in ql for kw in keywords):
                    all_ents = list(entities) + list(multi_word)
                    for ent in all_ents:
                        rows = conn.execute(
                            f"SELECT id FROM connection_points WHERE entity = ? COLLATE NOCASE AND point_type = ?{session_filter}",
                            [ent, pt] + session_params).fetchall()
                        for r in rows:
                            graph_scores[r["id"]] = graph_scores.get(r["id"], 0) + 0.2

        # --- PHASE 2: FTS5 full-text search ---
        fts_scores: Dict[str, float] = {}
        fts_terms = [w for w in re.findall(r'\b\w{3,}\b', query) if w.lower() not in _NER_STOP]
        if fts_terms:
            fts_query = " OR ".join(f'"{t}"' for t in fts_terms[:10])
            try:
                with self.db.read() as conn:
                    rows = conn.execute("""
                        SELECT cp.id, bm25(cp_fts, 5.0, 3.0, 2.0, 1.0) as rank
                        FROM cp_fts
                        JOIN connection_points cp ON cp.rowid = cp_fts.rowid
                        WHERE cp_fts MATCH ?
                        ORDER BY rank LIMIT 100
                    """, (fts_query,)).fetchall()
                if rows:
                    min_r = min(r["rank"] for r in rows)
                    max_r = max(r["rank"] for r in rows)
                    spread = max_r - min_r if max_r != min_r else 1.0
                    for r in rows:
                        fts_scores[r["id"]] = 1.0 - (r["rank"] - min_r) / spread
            except Exception as e:
                log.warning(f"FTS5 search error: {e}")

        # --- PHASE 3: FAISS semantic pre-filter ---
        faiss_candidates = self._cp_faiss.search(query_emb, top_k=min(top_k * 5, 200))

        # --- PHASE 4: Merge candidate set + load embeddings + NumPy rerank ---
        all_candidate_ids = set(graph_scores.keys()) | set(fts_scores.keys())
        all_candidate_ids.update(cid for cid, _ in faiss_candidates)

        if not all_candidate_ids:
            return []

        # Batch-load embeddings + metadata from SQLite
        id_list = list(all_candidate_ids)
        placeholders = ",".join("?" * len(id_list))
        with self.db.read() as conn:
            rows = conn.execute(f"""
                SELECT id, entity, point_type, value, connects_to, reason,
                       weight, category, session_id, source, thread_id,
                       position, namespace, created_at, embedding
                FROM connection_points
                WHERE id IN ({placeholders}) AND embedding IS NOT NULL
            """, id_list).fetchall()

        # Apply session filter + build numpy arrays
        valid_ids = []
        valid_embs = []
        meta_by_id: Dict[str, dict] = {}
        for row in rows:
            if active_sessions and row["session_id"] and row["session_id"] not in active_sessions:
                continue
            valid_ids.append(row["id"])
            valid_embs.append(np.frombuffer(row["embedding"], dtype=np.float32))
            meta_by_id[row["id"]] = self._row_to_cp_dict(row)

        if not valid_ids:
            return []

        emb_matrix = np.stack(valid_embs)

        # NumPy exact reranking
        reranked = NumpyReranker.rerank(query_emb, valid_ids, emb_matrix, threshold=0.15)

        # --- PHASE 5: Score fusion ---
        results = []
        for cp_id, sem_score in reranked:
            meta = meta_by_id.get(cp_id, {})
            if not meta:
                continue
            graph = min(graph_scores.get(cp_id, 0.0), 1.0)
            fts = min(fts_scores.get(cp_id, 0.0), 1.0)

            # Weighted fusion: semantic primary, graph + FTS boost
            final = sem_score * 0.55 + graph * 0.30 + fts * 0.15
            meta["score"] = round(min(final, 1.0), 3)
            meta["graph_score"] = round(graph, 3)
            meta["semantic_score"] = round(sem_score, 3)
            meta["fts_score"] = round(fts, 3)
            results.append(meta)

        results.sort(key=lambda x: x["score"], reverse=True)
        return results[:top_k]

    # =========================================================================
    # THREAD OPERATIONS
    # =========================================================================

    def add_thread(self, thread_id: str, name: str, entity: str = "",
                   thread_type: str = "plot_line", session_id: str = "",
                   point_ids: Optional[List[str]] = None) -> str:
        with self.db.read() as conn:
            existing = conn.execute("SELECT id FROM threads WHERE id = ?", (thread_id,)).fetchone()

        if existing:
            if point_ids:
                with self.db.write() as conn:
                    # Get current max position
                    row = conn.execute(
                        "SELECT COALESCE(MAX(position), -1) as maxp FROM thread_points WHERE thread_id = ?",
                        (thread_id,)).fetchone()
                    pos = row["maxp"] + 1
                    for pid in point_ids:
                        conn.execute(
                            "INSERT OR IGNORE INTO thread_points(thread_id, cp_id, position) VALUES (?, ?, ?)",
                            (thread_id, pid, pos))
                        pos += 1
            self._dirty = True
            return thread_id

        with self.db.write() as conn:
            conn.execute("""
                INSERT INTO threads (id, name, thread_type, entity, status,
                    tension_level, tone_trajectory, current_position, session_id, created_at)
                VALUES (?, ?, ?, ?, 'active', 0.5, '[]', 0, ?, ?)
            """, (thread_id, name, thread_type, entity, session_id, time.time()))
            if point_ids:
                for pos, pid in enumerate(point_ids):
                    conn.execute(
                        "INSERT OR IGNORE INTO thread_points(thread_id, cp_id, position) VALUES (?, ?, ?)",
                        (thread_id, pid, pos))
        self._incr_stat("threads_created")
        self._dirty = True
        return thread_id

    def advance_thread(self, thread_id: str, new_position: int = -1) -> bool:
        with self.db.read() as conn:
            row = conn.execute("SELECT current_position FROM threads WHERE id = ?",
                               (thread_id,)).fetchone()
        if not row:
            return False
        if new_position >= 0:
            pos = new_position
        else:
            with self.db.read() as conn:
                total = conn.execute(
                    "SELECT COUNT(*) as cnt FROM thread_points WHERE thread_id = ?",
                    (thread_id,)).fetchone()["cnt"]
            pos = min(row["current_position"] + 1, max(total - 1, 0))
        with self.db.write() as conn:
            conn.execute("UPDATE threads SET current_position = ? WHERE id = ?",
                         (pos, thread_id))
        self._dirty = True
        return True

    def trace_thread(self, thread_id: str, from_position: int = -1,
                     direction: str = "back", steps: int = 5) -> List[dict]:
        with self.db.read() as conn:
            thread = conn.execute("SELECT * FROM threads WHERE id = ?",
                                  (thread_id,)).fetchone()
            if not thread:
                return []
            pos = from_position if from_position >= 0 else thread["current_position"]

            if direction == "back":
                rows = conn.execute("""
                    SELECT cp.* FROM thread_points tp
                    JOIN connection_points cp ON cp.id = tp.cp_id
                    WHERE tp.thread_id = ? AND tp.position >= ? AND tp.position < ?
                    ORDER BY tp.position
                """, (thread_id, max(0, pos - steps), pos)).fetchall()
            else:
                rows = conn.execute("""
                    SELECT cp.* FROM thread_points tp
                    JOIN connection_points cp ON cp.id = tp.cp_id
                    WHERE tp.thread_id = ? AND tp.position > ? AND tp.position <= ?
                    ORDER BY tp.position
                """, (thread_id, pos, pos + steps)).fetchall()

        return [self._row_to_cp_dict(r) for r in rows]

    def get_active_threads(self) -> List[dict]:
        with self.db.read() as conn:
            threads = conn.execute("SELECT * FROM threads WHERE status = 'active'").fetchall()
            result = []
            for t in threads:
                points = conn.execute(
                    "SELECT cp_id FROM thread_points WHERE thread_id = ? ORDER BY position",
                    (t["id"],)).fetchall()
                knots = conn.execute(
                    "SELECT knot_id FROM knot_threads WHERE thread_id = ?",
                    (t["id"],)).fetchall()
                result.append({
                    "id": t["id"], "name": t["name"], "thread_type": t["thread_type"],
                    "entity": t["entity"], "status": t["status"],
                    "tension_level": round(t["tension_level"], 3),
                    "tone_trajectory": json.loads(t["tone_trajectory"] or "[]"),
                    "current_position": t["current_position"],
                    "session_id": t["session_id"], "created_at": t["created_at"],
                    "points": [r["cp_id"] for r in points],
                    "knots": [r["knot_id"] for r in knots],
                })
        return result

    def get_thread(self, thread_id: str) -> Optional[dict]:
        with self.db.read() as conn:
            t = conn.execute("SELECT * FROM threads WHERE id = ?", (thread_id,)).fetchone()
            if not t:
                return None
            points = conn.execute(
                "SELECT cp_id FROM thread_points WHERE thread_id = ? ORDER BY position",
                (thread_id,)).fetchall()
            knots = conn.execute(
                "SELECT knot_id FROM knot_threads WHERE thread_id = ?",
                (thread_id,)).fetchall()
        return {
            "id": t["id"], "name": t["name"], "thread_type": t["thread_type"],
            "entity": t["entity"], "status": t["status"],
            "tension_level": round(t["tension_level"], 3),
            "tone_trajectory": json.loads(t["tone_trajectory"] or "[]"),
            "current_position": t["current_position"],
            "session_id": t["session_id"], "created_at": t["created_at"],
            "points": [r["cp_id"] for r in points],
            "knots": [r["knot_id"] for r in knots],
        }

    def delete_thread(self, thread_id: str) -> bool:
        with self.db.write() as conn:
            cur = conn.execute("DELETE FROM threads WHERE id = ?", (thread_id,))
            if cur.rowcount == 0:
                return False
            conn.execute("DELETE FROM thread_points WHERE thread_id = ?", (thread_id,))
            conn.execute("DELETE FROM knot_threads WHERE thread_id = ?", (thread_id,))
        self._dirty = True
        return True

    # =========================================================================
    # KNOT OPERATIONS
    # =========================================================================

    def add_knot(self, knot_id: str, name: str, thread_ids: List[str],
                 pivot_type: str = "collision", reason: str = "",
                 session_id: str = "",
                 active_points: Optional[Dict[str, List[str]]] = None) -> str:
        with self.db.write() as conn:
            conn.execute("""
                INSERT OR REPLACE INTO knots
                (id, name, pivot_type, reason, active_points, session_id, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            """, (knot_id, name, pivot_type, reason,
                  json.dumps(active_points or {}), session_id, time.time()))
            for tid in thread_ids:
                conn.execute(
                    "INSERT OR IGNORE INTO knot_threads(knot_id, thread_id) VALUES (?, ?)",
                    (knot_id, tid))
        self._incr_stat("knots_created")
        self._dirty = True
        return knot_id

    def get_knot_context(self, knot_id: str) -> Optional[dict]:
        with self.db.read() as conn:
            knot = conn.execute("SELECT * FROM knots WHERE id = ?", (knot_id,)).fetchone()
            if not knot:
                return None

            thread_ids = [r["thread_id"] for r in
                          conn.execute("SELECT thread_id FROM knot_threads WHERE knot_id = ?",
                                       (knot_id,)).fetchall()]
            active_pts = json.loads(knot["active_points"] or "{}")

            context = {
                "id": knot["id"], "name": knot["name"],
                "threads": thread_ids, "pivot_type": knot["pivot_type"],
                "narrative_weight": round(knot["narrative_weight"], 3),
                "tension_before": round(knot["tension_before"], 3),
                "tension_after": round(knot["tension_after"], 3),
                "tone_shift": knot["tone_shift"], "reason": knot["reason"],
                "unresolved": json.loads(knot["unresolved"] or "[]"),
                "active_points": active_pts,
                "session_id": knot["session_id"], "created_at": knot["created_at"],
                "thread_context": {},
            }

            for tid in thread_ids:
                t = conn.execute("SELECT * FROM threads WHERE id = ?", (tid,)).fetchone()
                if not t:
                    continue
                cp_ids = active_pts.get(tid, [])
                active_cps = []
                for pid in cp_ids:
                    cp_row = conn.execute("SELECT * FROM connection_points WHERE id = ?",
                                          (pid,)).fetchone()
                    if cp_row:
                        active_cps.append(self._row_to_cp_dict(cp_row))

                buildup = self.trace_thread(tid, t["current_position"], "back", 2)

                context["thread_context"][tid] = {
                    "thread_name": t["name"],
                    "entity": t["entity"],
                    "tension": t["tension_level"],
                    "tone": json.loads(t["tone_trajectory"] or "[]")[-1:]
                           and json.loads(t["tone_trajectory"] or "[]")[-1] or "",
                    "active_points": active_cps,
                    "buildup": buildup,
                }
        return context

    def list_knots(self) -> List[dict]:
        with self.db.read() as conn:
            knots = conn.execute("SELECT * FROM knots").fetchall()
            result = []
            for k in knots:
                thread_ids = [r["thread_id"] for r in
                              conn.execute("SELECT thread_id FROM knot_threads WHERE knot_id = ?",
                                           (k["id"],)).fetchall()]
                result.append({
                    "id": k["id"], "name": k["name"], "threads": thread_ids,
                    "pivot_type": k["pivot_type"], "reason": k["reason"],
                    "session_id": k["session_id"], "created_at": k["created_at"],
                    "narrative_weight": round(k["narrative_weight"], 3),
                    "tension_before": round(k["tension_before"], 3),
                    "tension_after": round(k["tension_after"], 3),
                    "tone_shift": k["tone_shift"],
                    "unresolved": json.loads(k["unresolved"] or "[]"),
                    "active_points": json.loads(k["active_points"] or "{}"),
                })
        return result

    def delete_knot(self, knot_id: str) -> bool:
        with self.db.write() as conn:
            cur = conn.execute("DELETE FROM knots WHERE id = ?", (knot_id,))
            if cur.rowcount == 0:
                return False
            conn.execute("DELETE FROM knot_threads WHERE knot_id = ?", (knot_id,))
        self._dirty = True
        return True

    def delete_session_threads_and_knots(self, session_id: str) -> dict:
        with self.db.write() as conn:
            # Get thread/knot IDs for this session
            t_ids = [r["id"] for r in conn.execute(
                "SELECT id FROM threads WHERE session_id = ?", (session_id,)).fetchall()]
            k_ids = [r["id"] for r in conn.execute(
                "SELECT id FROM knots WHERE session_id = ?", (session_id,)).fetchall()]

            for tid in t_ids:
                conn.execute("DELETE FROM thread_points WHERE thread_id = ?", (tid,))
                conn.execute("DELETE FROM knot_threads WHERE thread_id = ?", (tid,))
            if t_ids:
                placeholders = ",".join("?" * len(t_ids))
                conn.execute(f"DELETE FROM threads WHERE id IN ({placeholders})", t_ids)

            for kid in k_ids:
                conn.execute("DELETE FROM knot_threads WHERE knot_id = ?", (kid,))
            if k_ids:
                placeholders = ",".join("?" * len(k_ids))
                conn.execute(f"DELETE FROM knots WHERE id IN ({placeholders})", k_ids)

        self._dirty = True
        return {"deleted_threads": len(t_ids), "deleted_knots": len(k_ids)}

    # =========================================================================
    # LEGACY BLOB MEMORY OPERATIONS (backward compat)
    # =========================================================================

    def add(self, content: str, namespace: str = "default",
            metadata: dict = None, priority: float = 1.0) -> Optional[str]:
        memory_id = self._generate_id(content, namespace)
        meta = metadata or {}

        # Extract session_id and source into proper columns (not buried in JSON)
        session_id = meta.pop("session_id", "")
        source = meta.pop("source", "auto_extract")
        meta_json = json.dumps(meta)  # Remaining metadata only

        with self.db.read() as conn:
            existing = conn.execute("SELECT id FROM memories WHERE id = ?", (memory_id,)).fetchone()
        if existing:
            with self.db.write() as conn:
                conn.execute("""
                    UPDATE memories SET last_accessed = ?, access_count = access_count + 1,
                    priority = MAX(priority, ?) WHERE id = ?
                """, (time.time(), priority, memory_id))
            self._dirty = True
            return memory_id

        embedding = self._get_embedding(content)
        quality = self._estimate_quality(content, embedding)
        if quality < self.config.quality_threshold:
            self._incr_stat("adds_rejected")
            return None

        emb_blob = embedding.astype(np.float32).tobytes()
        with self.db.write() as conn:
            conn.execute("""
                INSERT INTO memories
                (id, content, tier, namespace, quality_score, access_count, priority,
                 session_id, source, created_at, last_accessed, metadata, embedding)
                VALUES (?, ?, 'semantic', ?, ?, 0, ?, ?, ?, ?, ?, ?, ?)
            """, (memory_id, content, namespace, quality, priority,
                  session_id, source, time.time(), time.time(), meta_json, emb_blob))

        self._blob_faiss.add(memory_id, embedding)
        self._create_links(memory_id, embedding, namespace, content)
        self._incr_stat("adds")
        self._dirty = True
        return memory_id

    def search(self, query: str, top_k: int = 15, namespace: str = None) -> list:
        """Search blob memories. Returns list of SearchResult objects."""
        self._incr_stat("searches")
        query_emb = self._get_embedding(query)

        # FAISS pre-filter
        faiss_results = self._blob_faiss.search(query_emb, top_k=top_k * 3)

        # If no FAISS results, try numpy fallback
        if not faiss_results:
            with self.db.read() as conn:
                ns_clause = " AND namespace = ?" if namespace else ""
                params = [namespace] if namespace else []
                rows = conn.execute(
                    f"SELECT id, embedding FROM memories WHERE embedding IS NOT NULL{ns_clause}",
                    params).fetchall()
            if rows:
                ids = [r["id"] for r in rows]
                embs = np.stack([np.frombuffer(r["embedding"], dtype=np.float32) for r in rows])
                reranked = NumpyReranker.rerank(query_emb, ids, embs, threshold=0.20)
                faiss_results = {cid: score for cid, score in reranked}

        # Load blob metadata for candidates
        candidate_ids = list(faiss_results.keys() if isinstance(faiss_results, dict) else
                             {cid for cid, _ in faiss_results})
        if not candidate_ids:
            return []

        placeholders_str = ",".join("?" * len(candidate_ids))
        with self.db.read() as conn:
            rows = conn.execute(
                f"SELECT * FROM memories WHERE id IN ({placeholders_str})", candidate_ids
            ).fetchall()

        # Link propagation
        link_scores: Dict[str, float] = {}
        sem_scores = faiss_results if isinstance(faiss_results, dict) else dict(faiss_results)
        top_seeds = sorted(sem_scores.items(), key=lambda x: x[1], reverse=True)[:15]

        with self.db.read() as conn:
            for mem_id, base_score in top_seeds:
                if base_score < 0.20:
                    continue
                outgoing = conn.execute(
                    "SELECT target_id, strength FROM neural_links WHERE source_id = ?",
                    (mem_id,)).fetchall()
                for link in outgoing:
                    boost = base_score * link["strength"] * 0.60
                    link_scores[link["target_id"]] = link_scores.get(link["target_id"], 0) + boost
                incoming = conn.execute(
                    "SELECT source_id, strength FROM neural_links WHERE target_id = ?",
                    (mem_id,)).fetchall()
                for link in incoming:
                    boost = base_score * link["strength"] * 0.40
                    link_scores[link["source_id"]] = link_scores.get(link["source_id"], 0) + boost

        results = []
        now = time.time()
        update_ids = []
        for row in rows:
            if namespace and row["namespace"] != namespace:
                continue
            sem = sem_scores.get(row["id"], 0)
            lnk = link_scores.get(row["id"], 0)
            combined = sem * 0.7 + min(lnk, 0.5) * 0.3
            if combined >= self.config.similarity_threshold:
                update_ids.append(row["id"])
                results.append(SearchResult(
                    id=row["id"], content=row["content"], score=combined,
                    tier=row["tier"], semantic_score=sem, link_score=lnk,
                    metadata=json.loads(row["metadata"] or "{}"),
                ))

        if update_ids:
            with self.db.write() as conn:
                placeholders_str = ",".join("?" * len(update_ids))
                conn.execute(
                    f"UPDATE memories SET access_count = access_count + 1, last_accessed = ? WHERE id IN ({placeholders_str})",
                    [now] + update_ids)

        results.sort(key=lambda x: x.score, reverse=True)
        return results[:top_k]

    def should_inject(self, query: str, context: str = "",
                      conversation_history: str = "") -> Tuple[bool, str, float]:
        combined = (query + " " + context).lower()
        skip_signals = ["this is a new", "new topic", "start fresh", "forget everything"]
        for signal in skip_signals:
            if signal in combined:
                self._incr_stat("skip_recommended")
                return False, "skip:" + signal, 0.0

        best_score = 0.0
        source = "none"

        blob_results = self.search(query, top_k=3)
        if blob_results:
            best_score = blob_results[0].score
            source = "blob"

        if best_score < 0.40:
            cp_results = self.graph_search(query, top_k=3)
            if cp_results:
                cp_best = cp_results[0].get("score", 0)
                if cp_best > best_score:
                    best_score = cp_best
                    source = "cp"

        if best_score == 0.0:
            self._incr_stat("skip_recommended")
            return False, "no_relevant_memories", 0.0

        inject_signals = [
            "previous", "earlier", "before", "you said", "you mentioned",
            "based on", "using your", "your analysis", "your framework",
            "compare", "contrast", "synthesize", "combine",
            "apply your", "you previously", "your earlier",
            "tell me everything", "remember",
        ]
        keyword_boost = 0.0
        matched_signal = None
        for signal in inject_signals:
            if signal in combined:
                keyword_boost = 0.15
                matched_signal = signal
                break

        confidence = min(1.0, best_score + keyword_boost)

        if conversation_history and len(conversation_history.split()) > 500:
            query_words = set(query.lower().split())
            if query_words:
                history_words = set(conversation_history.lower().split())
                overlap = len(query_words & history_words)
                if overlap > len(query_words) * 0.6:
                    confidence *= 0.5

        if confidence >= 0.40:
            self._incr_stat("inject_recommended")
            reason = f"inject:confidence={round(confidence, 2)}"
            if matched_signal:
                reason += f",signal={matched_signal}"
            return True, reason, confidence
        else:
            self._incr_stat("skip_recommended")
            return False, f"low_confidence:{round(confidence, 2)}", confidence

    def get_context(self, query: str, top_k: int = 15) -> str:
        results = self.search(query, top_k=top_k)
        if not results:
            return ""
        parts = ["[RELEVANT CONTEXT FROM MEMORY]"]
        for r in results:
            parts.append(f"- [{r.tier.upper()}] {r.content}")
        parts.append("[END CONTEXT]")
        return "\n".join(parts)

    def get(self, memory_id: str) -> Optional[dict]:
        with self.db.read() as conn:
            cp = conn.execute("SELECT * FROM connection_points WHERE id = ?",
                              (memory_id,)).fetchone()
            if cp:
                return self._row_to_cp_dict(cp)
            mem = conn.execute("SELECT * FROM memories WHERE id = ?",
                               (memory_id,)).fetchone()
            if mem:
                return {
                    "id": mem["id"], "content": mem["content"],
                    "tier": mem["tier"], "namespace": mem["namespace"],
                    "quality_score": round(mem["quality_score"], 3),
                    "access_count": mem["access_count"], "priority": mem["priority"],
                    "session_id": mem["session_id"], "source": mem["source"],
                    "created_at": mem["created_at"], "last_accessed": mem["last_accessed"],
                    "metadata": json.loads(mem["metadata"] or "{}"),
                }
        return None

    def delete(self, memory_id: str) -> bool:
        with self.db.write() as conn:
            # Try CP first
            cur = conn.execute("DELETE FROM connection_points WHERE id = ?", (memory_id,))
            if cur.rowcount > 0:
                conn.execute("DELETE FROM thread_points WHERE cp_id = ?", (memory_id,))
                self._cp_faiss.mark_dirty()
                self._dirty = True
                return True
            # Try blob
            cur = conn.execute("DELETE FROM neural_links WHERE source_id = ? OR target_id = ?",
                               (memory_id, memory_id))
            cur2 = conn.execute("DELETE FROM memories WHERE id = ?", (memory_id,))
            if cur2.rowcount > 0:
                self._dirty = True
                return True
        return False

    def list_all(self) -> List[dict]:
        with self.db.read() as conn:
            rows = conn.execute("SELECT * FROM memories").fetchall()
        return [{
            "id": r["id"], "content": r["content"], "tier": r["tier"],
            "namespace": r["namespace"], "quality_score": round(r["quality_score"], 3),
            "access_count": r["access_count"], "priority": r["priority"],
            "session_id": r["session_id"], "source": r["source"],
            "created_at": r["created_at"], "last_accessed": r["last_accessed"],
            "metadata": json.loads(r["metadata"] or "{}"),
        } for r in rows]

    def list_memories(self, namespace: str = None) -> List[dict]:
        """Alias for list_all with optional namespace filter."""
        return self.list_all()

    # =========================================================================
    # LINK CREATION (for blob memories)
    # =========================================================================

    def _estimate_quality(self, content: str, embedding: np.ndarray) -> float:
        score = 0.5
        words = len(content.split())
        lower = content.lower()
        is_high_value = any(m in lower for m in HIGH_VALUE_MARKERS)
        if is_high_value:
            score += 0.2
            if words > 20:
                score += 0.1
        else:
            if words < 5:
                score -= 0.3
            elif words > 20:
                score += 0.1
        # Check duplicate via FAISS
        top_match = self._blob_faiss.search(embedding, top_k=1)
        if top_match:
            best_sim = list(top_match.values())[0] if isinstance(top_match, dict) else (top_match[0][1] if top_match else 0)
            if best_sim > 0.95:
                score -= 0.30
            elif best_sim > 0.85:
                score -= 0.10
            elif best_sim < 0.30:
                score += 0.15
        return max(0.0, min(1.0, score))

    def _create_links(self, memory_id: str, embedding: np.ndarray,
                      namespace: str, content: str):
        """Create neural links for a new blob memory."""
        candidates = self._blob_faiss.search(embedding, top_k=50)
        if isinstance(candidates, dict):
            candidate_items = [(k, v) for k, v in candidates.items() if k != memory_id]
        else:
            candidate_items = [(k, v) for k, v in candidates if k != memory_id]

        my_terms = set(re.findall(r'\b[A-Z][a-z]+\b', content))
        my_causal = any(m in content.lower() for m in [
            "because", "therefore", "led to", "caused", "resulted in",
            "due to", "consequently", "thus", "triggered", "forced"])

        with self.db.read() as conn:
            for other_id, sim in candidate_items:
                other_row = conn.execute("SELECT content, namespace, metadata, created_at FROM memories WHERE id = ?",
                                         (other_id,)).fetchone()
                if not other_row:
                    continue
                other_content = other_row["content"]
                other_terms = set(re.findall(r'\b[A-Z][a-z]+\b', other_content))
                other_meta = json.loads(other_row["metadata"] or "{}")

                link_type = None
                strength = 0.0

                if sim >= 0.85 and my_terms & other_terms:
                    link_type = "direct_reference"
                    strength = 0.90
                elif sim >= 0.80 and my_terms and other_terms and (my_terms <= other_terms or other_terms <= my_terms):
                    link_type = "hierarchical"
                    strength = 0.85
                elif sim >= 0.75 and my_causal:
                    link_type = "causal"
                    strength = 0.80
                elif sim >= 0.50:
                    link_type = "semantic_similarity"
                    strength = 0.75
                elif sim >= 0.45:
                    link_type = "associative"
                    strength = 0.60

                if link_type:
                    self._add_link(memory_id, other_id, link_type, strength)

                # Temporal links (skip batch uploads)
                source = other_meta.get("source", "")
                is_batch = source == "file_upload"
                time_gap = abs(time.time() - other_row["created_at"])
                if time_gap < 300 and not is_batch and link_type != "direct_reference":
                    self._add_link(memory_id, other_id, "temporal", 0.65)

    def _add_link(self, source_id: str, target_id: str,
                  link_type: str, strength: float):
        fwd_id = f"{source_id}:{target_id}:{link_type}"
        rev_id = f"{target_id}:{source_id}:{link_type}"
        now = time.time()
        with self.db.write() as conn:
            conn.execute("""
                INSERT OR IGNORE INTO neural_links (id, source_id, target_id, link_type, strength, created_at, last_maintained)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            """, (fwd_id, source_id, target_id, link_type, strength, now, now))
            conn.execute("""
                INSERT OR IGNORE INTO neural_links (id, source_id, target_id, link_type, strength, created_at, last_maintained)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            """, (rev_id, target_id, source_id, link_type, strength, now, now))
        self._incr_stat("links_created", 2)

    # =========================================================================
    # MAINTENANCE
    # =========================================================================

    def maintenance(self) -> dict:
        now = time.time()
        results = {"decayed": 0, "pruned": 0, "links_decayed": 0,
                   "links_pruned": 0, "tier_promotions": 0, "tier_demotions": 0}

        with self.db.write() as conn:
            # Decay blob memories
            rows = conn.execute("SELECT id, quality_score, priority, last_accessed, access_count, tier, created_at FROM memories").fetchall()
            prune_ids = []
            for r in rows:
                days = (now - r["last_accessed"]) / 86400
                if days > 1:
                    eff_decay = self.config.decay_rate_per_day * days * max(0.1, 1.0 - (r["priority"] - 1.0) * 0.5)
                    new_q = max(0.0, r["quality_score"] - eff_decay)
                    conn.execute("UPDATE memories SET quality_score = ? WHERE id = ?", (new_q, r["id"]))
                    results["decayed"] += 1
                # Promotions/demotions
                if r["tier"] != "working" and r["access_count"] >= self.config.promote_to_working_accesses:
                    conn.execute("UPDATE memories SET tier = 'working' WHERE id = ?", (r["id"],))
                    results["tier_promotions"] += 1
                if r["tier"] != "archive" and days > self.config.demote_to_archive_days and r["access_count"] < self.config.promote_to_working_accesses:
                    conn.execute("UPDATE memories SET tier = 'archive' WHERE id = ?", (r["id"],))
                    results["tier_demotions"] += 1
                days_created = (now - r["created_at"]) / 86400
                if r["quality_score"] < self.config.prune_quality_floor and days_created > self.config.prune_age_days:
                    prune_ids.append(r["id"])

            # Prune low-quality
            for mid in prune_ids:
                conn.execute("DELETE FROM neural_links WHERE source_id = ? OR target_id = ?", (mid, mid))
                conn.execute("DELETE FROM memories WHERE id = ?", (mid,))
                results["pruned"] += 1

            # Decay/prune links
            links = conn.execute("SELECT id, link_type, strength, last_maintained FROM neural_links").fetchall()
            prune_link_ids = []
            for link in links:
                props = LINK_PROPERTIES.get(LinkType(link["link_type"]) if link["link_type"] in [lt.value for lt in LinkType] else LinkType.ASSOCIATIVE, {})
                decay = props.get("decay_per_day", 0.01)
                days = (now - link["last_maintained"]) / 86400
                if days > 0.5:
                    new_str = max(0.0, link["strength"] - decay * days)
                    conn.execute("UPDATE neural_links SET strength = ?, last_maintained = ? WHERE id = ?",
                                 (new_str, now, link["id"]))
                    results["links_decayed"] += 1
                if link["strength"] <= 0.01:
                    prune_link_ids.append(link["id"])
            if prune_link_ids:
                placeholders = ",".join("?" * len(prune_link_ids))
                conn.execute(f"DELETE FROM neural_links WHERE id IN ({placeholders})", prune_link_ids)
                results["links_pruned"] = len(prune_link_ids)

        for key, val in results.items():
            self._incr_stat(key, val)
        self._dirty = True
        return results

    # =========================================================================
    # STATS & CLEAR
    # =========================================================================

    def get_stats(self) -> dict:
        with self.db.read() as conn:
            n_mem = conn.execute("SELECT COUNT(*) as c FROM memories").fetchone()["c"]
            n_links = conn.execute("SELECT COUNT(*) as c FROM neural_links").fetchone()["c"]
            n_cp = conn.execute("SELECT COUNT(*) as c FROM connection_points").fetchone()["c"]
            n_threads = conn.execute("SELECT COUNT(*) as c FROM threads").fetchone()["c"]
            n_knots = conn.execute("SELECT COUNT(*) as c FROM knots").fetchone()["c"]
            n_active = conn.execute("SELECT COUNT(*) as c FROM threads WHERE status='active'").fetchone()["c"]

            link_counts = {}
            for r in conn.execute("SELECT link_type, COUNT(*) as c FROM neural_links GROUP BY link_type").fetchall():
                link_counts[r["link_type"]] = r["c"]

            tier_counts = {}
            for r in conn.execute("SELECT tier, COUNT(*) as c FROM memories GROUP BY tier").fetchall():
                tier_counts[r["tier"]] = r["c"]

            cp_cats = {}
            for r in conn.execute("SELECT category, COUNT(*) as c FROM connection_points GROUP BY category").fetchall():
                cp_cats[r["category"]] = r["c"]

            n_entities = conn.execute("SELECT COUNT(DISTINCT entity) as c FROM connection_points").fetchone()["c"]

        return {
            "total_memories": n_mem,
            "total_links": n_links,
            "links_by_type": link_counts,
            "memories_by_tier": tier_counts,
            "faiss_enabled": HAS_FAISS,
            "cp_faiss_enabled": HAS_FAISS,
            "cp_faiss_indexed": self._cp_faiss.total,
            "total_connection_points": n_cp,
            "total_threads": n_threads,
            "total_knots": n_knots,
            "cp_by_category": cp_cats,
            "active_threads": n_active,
            "entities_indexed": n_entities,
            "embedding_cache": self._emb_cache.get_stats(),
            "storage_backend": "sqlite_v7",
            **self._stats,
        }

    def clear(self):
        with self.db.write() as conn:
            for table in ("connection_points", "threads", "thread_points",
                          "knots", "knot_threads", "memories", "neural_links",
                          "engine_stats"):
                conn.execute(f"DELETE FROM {table}")
            # Rebuild FTS
            conn.execute("INSERT INTO cp_fts(cp_fts) VALUES('rebuild')")
            conn.execute("INSERT INTO memory_fts(memory_fts) VALUES('rebuild')")
        self._emb_cache.clear()
        self._stats = self._load_stats()
        self._cp_faiss.rebuild_from_db(self.db)
        self._blob_faiss.rebuild_from_db(self.db)
        self._dirty = True

    def __len__(self):
        with self.db.read() as conn:
            return conn.execute("SELECT COUNT(*) as c FROM memories").fetchone()["c"]

    @property
    def is_dirty(self):
        return self._dirty

    def mark_clean(self):
        self._flush_stats()
        self._dirty = False

        # Periodic FAISS rebuild if needed
        if self._cp_faiss.needs_rebuild:
            self._cp_faiss.rebuild_from_db(self.db)

    # =========================================================================
    # SESSION DELETE (combined CP + thread + knot cascade)
    # =========================================================================

    def delete_session(self, session_id: str) -> dict:
        """Delete all data for a session (CPs, threads, knots, blobs)."""
        cp_deleted = self.delete_session_points(session_id)
        tk_result = self.delete_session_threads_and_knots(session_id)
        return {
            "points_deleted": cp_deleted,
            **tk_result,
        }

    # =========================================================================
    # JSON → SQLite MIGRATION (runs once on first startup)
    # =========================================================================

    def migrate_from_json(self, json_path: str) -> int:
        """Import legacy mnemo_db.json into SQLite. Returns items imported."""
        if not os.path.exists(json_path):
            return 0

        print(f"Migrating legacy JSON database: {json_path}")
        with open(json_path, "r") as f:
            data = json.load(f)

        imported = 0

        # Memories
        for mid, mdata in data.get("memories", {}).items():
            emb = None
            if "embedding_b64" in mdata:
                try:
                    import base64
                    raw = base64.b64decode(mdata["embedding_b64"])
                    emb = np.frombuffer(raw, dtype=np.dtype(mdata.get("embedding_dtype", "float32"))).copy()
                except Exception:
                    pass
            if emb is None:
                emb = self._get_embedding(mdata.get("content", ""))
            emb_blob = emb.astype(np.float32).tobytes()

            # Extract session_id/source from old metadata blob into proper columns
            old_meta = mdata.get("metadata", {})
            if isinstance(old_meta, str):
                try:
                    old_meta = json.loads(old_meta)
                except Exception:
                    old_meta = {}
            session_id = old_meta.pop("session_id", mdata.get("session_id", ""))
            source = old_meta.pop("source", mdata.get("source", "auto_extract"))
            meta_json = json.dumps(old_meta)  # Remaining metadata only

            with self.db.write() as conn:
                conn.execute("""
                    INSERT OR IGNORE INTO memories
                    (id, content, tier, namespace, quality_score, access_count, priority,
                     session_id, source, created_at, last_accessed, metadata, embedding)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """, (mid, mdata["content"], mdata.get("tier", "semantic"),
                      mdata.get("namespace", "default"), mdata.get("quality_score", 0.5),
                      mdata.get("access_count", 0), mdata.get("priority", 1.0),
                      session_id, source,
                      mdata.get("created_at", time.time()), mdata.get("last_accessed", time.time()),
                      meta_json, emb_blob))
            imported += 1

        # Links
        for lid, ldata in data.get("links", {}).items():
            with self.db.write() as conn:
                conn.execute("""
                    INSERT OR IGNORE INTO neural_links
                    (id, source_id, target_id, link_type, strength, created_at, last_maintained)
                    VALUES (?, ?, ?, ?, ?, ?, ?)
                """, (lid, ldata["source_id"], ldata["target_id"], ldata["link_type"],
                      ldata["strength"], ldata.get("created_at", time.time()),
                      ldata.get("last_maintained", time.time())))

        # Connection Points
        for cpid, cpd in data.get("connection_points", {}).items():
            emb = None
            if "embedding_b64" in cpd:
                try:
                    import base64
                    raw = base64.b64decode(cpd["embedding_b64"])
                    emb = np.frombuffer(raw, dtype=np.dtype(cpd.get("embedding_dtype", "float32"))).copy()
                except Exception:
                    pass
            if emb is None:
                searchable = self._cp_to_searchable(
                    cpd.get("entity", ""), cpd.get("point_type", ""),
                    cpd.get("value", ""), cpd.get("connects_to", ""),
                    cpd.get("reason", ""), cpd.get("category", "fact"))
                emb = self._get_embedding(searchable)
            emb_blob = emb.astype(np.float32).tobytes()

            with self.db.write() as conn:
                conn.execute("""
                    INSERT OR IGNORE INTO connection_points
                    (id, entity, point_type, value, connects_to, reason, weight,
                     category, session_id, source, thread_id, position, namespace, created_at, embedding)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """, (cpid, cpd.get("entity", ""), cpd.get("point_type", ""),
                      cpd.get("value", ""), cpd.get("connects_to", ""),
                      cpd.get("reason", ""), cpd.get("weight", 0.5),
                      cpd.get("category", "fact"), cpd.get("session_id", ""),
                      cpd.get("source", "auto_extract"), cpd.get("thread_id", ""),
                      cpd.get("position", -1), cpd.get("namespace", "default"),
                      cpd.get("created_at", time.time()), emb_blob))
            imported += 1

        # Threads
        for tid, td in data.get("threads", {}).items():
            with self.db.write() as conn:
                conn.execute("""
                    INSERT OR IGNORE INTO threads
                    (id, name, thread_type, entity, status, tension_level,
                     tone_trajectory, current_position, session_id, created_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """, (tid, td.get("name", ""), td.get("thread_type", "plot_line"),
                      td.get("entity", ""), td.get("status", "active"),
                      td.get("tension_level", 0.5),
                      json.dumps(td.get("tone_trajectory", [])),
                      td.get("current_position", 0),
                      td.get("session_id", ""), td.get("created_at", time.time())))
                for pos, pid in enumerate(td.get("points", [])):
                    conn.execute(
                        "INSERT OR IGNORE INTO thread_points(thread_id, cp_id, position) VALUES(?, ?, ?)",
                        (tid, pid, pos))

        # Knots
        for kid, kd in data.get("knots", {}).items():
            with self.db.write() as conn:
                conn.execute("""
                    INSERT OR IGNORE INTO knots
                    (id, name, pivot_type, narrative_weight, tension_before, tension_after,
                     tone_shift, reason, unresolved, active_points, session_id, created_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """, (kid, kd.get("name", ""), kd.get("pivot_type", "collision"),
                      kd.get("narrative_weight", 0.5), kd.get("tension_before", 0.5),
                      kd.get("tension_after", 0.5), kd.get("tone_shift", ""),
                      kd.get("reason", ""), json.dumps(kd.get("unresolved", [])),
                      json.dumps(kd.get("active_points", {})),
                      kd.get("session_id", ""), kd.get("created_at", time.time())))
                for t in kd.get("threads", []):
                    conn.execute(
                        "INSERT OR IGNORE INTO knot_threads(knot_id, thread_id) VALUES(?, ?)",
                        (kid, t))

        # Rebuild FAISS from migrated data
        self._cp_faiss.rebuild_from_db(self.db)
        self._blob_faiss.rebuild_from_db(self.db)

        print(f"Migration complete: {imported} items imported into SQLite.")
        return imported


# =============================================================================
# PERSISTENT WRAPPER (Uses SyncEngine for R2 / HF Datasets)
# =============================================================================

class PersistentMnemo:
    """Wraps MnemoEngine with SyncEngine (R2 / HuggingFace Datasets)."""

    def __init__(self, db_path: str = None, enable_hf_sync: bool = True):
        config = MnemoConfig(db_path=db_path or "/app/data/mnemo.db")
        self._db_path = config.db_path

        # 1. Initialize the new SyncEngine
        from sync_engine import SyncEngine
        self.sync = SyncEngine(db_path=self._db_path)

        # 2. Download .db from cloud on startup
        if enable_hf_sync:
            self.sync.download()

        # 3. Check for legacy JSON and migrate if needed
        legacy_json = self.sync.get_legacy_json_path() or self._db_path.replace(".db", ".json").replace("mnemo.db", "mnemo_db.json")
        if not os.path.exists(self._db_path) or os.path.getsize(self._db_path) == 0:
            for candidate in [legacy_json, "/app/data/mnemo_db.json"]:
                if os.path.exists(candidate):
                    legacy_json = candidate
                    break

        # 4. Create engine (creates SQLite DB if not exists)
        self.engine = MnemoEngine(config)

        # 5. Migrate legacy JSON if SQLite is empty
        with self.engine.db.read() as conn:
            n_cp = conn.execute("SELECT COUNT(*) as c FROM connection_points").fetchone()["c"]
            n_mem = conn.execute("SELECT COUNT(*) as c FROM memories").fetchone()["c"]
        if n_cp == 0 and n_mem == 0 and os.path.exists(legacy_json):
            self.engine.migrate_from_json(legacy_json)

        # 6. Start background sync
        if enable_hf_sync and self.sync.has_credentials:
            t = threading.Thread(target=self._background_sync, daemon=True)
            t.start()
        elif enable_hf_sync:
            print("WARNING: No sync credentials. Memory will be ephemeral.")

    def _background_sync(self):
        """Monitors engine dirty state and triggers SyncEngine uploads."""
        import time
        while True:
            time.sleep(30)
            if self.engine.is_dirty:
                self.engine.mark_clean()
                self.sync.upload()

    # Delegate all engine methods (same signatures as v6.5)
    def add(self, *a, **kw): return self.engine.add(*a, **kw)
    def search(self, *a, **kw): return self.engine.search(*a, **kw)
    def should_inject(self, *a, **kw): return self.engine.should_inject(*a, **kw)
    def get_context(self, *a, **kw): return self.engine.get_context(*a, **kw)
    def get(self, *a, **kw): return self.engine.get(*a, **kw)
    def delete(self, *a, **kw): return self.engine.delete(*a, **kw)
    def maintenance(self): return self.engine.maintenance()
    def list_all(self): return self.engine.list_all()
    def list_memories(self, *a, **kw): return self.engine.list_memories(*a, **kw)
    def get_stats(self):
        stats = self.engine.get_stats()
        stats["sync_engine"] = self.sync.get_stats() # Expose new sync stats
        return stats
    def clear(self): return self.engine.clear()
    def __len__(self): return len(self.engine)
    def __getattr__(self, name): return getattr(self.engine, name)
    def add_point(self, *a, **kw): return self.engine.add_point(*a, **kw)
    def add_points_batch(self, *a, **kw): return self.engine.add_points_batch(*a, **kw)
    def get_point(self, *a, **kw): return self.engine.get_point(*a, **kw)
    def delete_point(self, *a, **kw): return self.engine.delete_point(*a, **kw)
    def update_point(self, *a, **kw): return self.engine.update_point(*a, **kw)
    def delete_session_points(self, *a, **kw): return self.engine.delete_session_points(*a, **kw)
    def graph_search(self, *a, **kw): return self.engine.graph_search(*a, **kw)
    def add_thread(self, *a, **kw): return self.engine.add_thread(*a, **kw)
    def advance_thread(self, *a, **kw): return self.engine.advance_thread(*a, **kw)
    def trace_thread(self, *a, **kw): return self.engine.trace_thread(*a, **kw)
    def get_active_threads(self): return self.engine.get_active_threads()
    def get_thread(self, *a, **kw): return self.engine.get_thread(*a, **kw)
    def delete_thread(self, *a, **kw): return self.engine.delete_thread(*a, **kw)
    def delete_knot(self, *a, **kw): return self.engine.delete_knot(*a, **kw)
    def delete_session_threads_and_knots(self, *a, **kw): return self.engine.delete_session_threads_and_knots(*a, **kw)
    def add_knot(self, *a, **kw): return self.engine.add_knot(*a, **kw)
    def get_knot_context(self, *a, **kw): return self.engine.get_knot_context(*a, **kw)
    def list_knots(self): return self.engine.list_knots()
    def list_points(self, *a, **kw): return self.engine.list_points(*a, **kw)
    def entity_lookup(self, *a, **kw): return self.engine.entity_lookup(*a, **kw)
    def delete_session(self, *a, **kw): return self.engine.delete_session(*a, **kw)