"""
wivexx_memory_store.py — WIVEXX long-term memory store  (v1.10)
═══════════════════════════════════════════════════════════════════════
The access layer for wivexx_memory.db, shared by wivexx_api (consolidation +
retrieval) and wivexx_gui (the 🧠 Memory manager).

Techniques borrowed from the current best memory systems:

  • RELEVANCE-SCORED RETRIEVAL (Mem0 / Zep): instead of injecting the newest
    N facts into every prompt, each query retrieves the memories that match
    it — scored by lexical overlap + importance + recency + reinforcement.
  • IMPORTANCE + RECENCY + REINFORCEMENT scoring (Stanford Generative
    Agents): every fact has an importance (1-10); retrieved facts get their
    use_count bumped, so memories that keep proving useful rank higher.
  • ADD / UPDATE / DELETE operations (Mem0): the consolidator decides whether
    new information adds a fact, corrects an existing one, or removes one —
    contradictions get fixed instead of duplicated.
  • IDENTITY CORE (MemGPT/Letta "core memory"): the highest-importance facts
    (≥8 — name, who the user is, explicit "remember this" items) are ALWAYS
    in context, like Letta's always-resident core block.
  • EXPLICIT USER CONTROL (ChatGPT memory): full view/edit/delete via the GUI
    manager and "remember/forget" voice commands.

Schema (migrated in place from the v1.3 tables):
    core_profile(id, key UNIQUE, value, updated_at,
                 importance INT, source TEXT, last_used REAL, use_count INT)
    topic_summaries(id, topic UNIQUE, summary, updated_at)
    episodic_log(id, ts, user_text, assistant_text)
"""

import math
import os
import re
import sqlite3
import threading
import time
import unicodedata
from typing import Optional

_DB_PATH = os.path.join(
    os.path.dirname(os.path.abspath(__file__)), "wivexx_memory.db"
)
_LOCK = threading.Lock()
_logger = None

# Categories that are never real memories (junk the extractor might emit)
KEY_BLACKLIST = {
    "question", "query", "request", "command", "weather", "time", "date",
    "search", "topic", "conversation", "user", "assistant", "jarvis", "wivexx",
    "current discussion", "discussion",
    # v1.34: structural keys from the consolidator's own JSON that a confused
    # local model once emitted as "facts" ("topic_updates: MrBeast, real name…")
    "topic_updates", "memory_ops", "profile_updates", "topics", "summary",
    "device", "info", "fact", "facts",
}
VALUE_REJECT_RE = re.compile(
    r"\b(?:asked\s+(?:about|for)|wants\s+to\s+know|requested|searched\s+for"
    r"|wondering\s+about|question\s+about|no\s+(?:facts|information)"
    # v1.34: a user QUESTION stored verbatim as a "fact"
    r"|could\s+you\s+tell\s+me|can\s+you\s+tell\s+me|tell\s+me\s+(?:what|how|about)"
    r"|what\s+(?:is|are)\s+the)\b",
    re.IGNORECASE,
)

# v1.34: facts describing the ASSISTANT are never facts about the USER — the
# engine once stored "name: Wivexx" (importance 8!) and "occupation: AI Virtual
# Assistant" as the user's identity.
_ASSISTANT_VALUE_RE = re.compile(
    r"^(?:wivexx|jarvis)$"
    r"|^(?:an?\s+)?(?:personal|ai|virtual|voice)?[\s/]*(?:ai\s+)?(?:virtual\s+)?"
    r"(?:voice\s+)?assistant\b",
    re.IGNORECASE,
)


def _fact_ok(key: str, value: str) -> bool:
    """Shared quality gate for EVERY write path (LLM consolidator, rule
    extractor, legacy). Rejects structural keys, sentence-length keys,
    question-like values and assistant-identity values."""
    if not key or key in KEY_BLACKLIST:
        return False
    if len(key.split()) > 4:                     # a whole sentence is not a key
        return False
    if not value or value.endswith("?") or VALUE_REJECT_RE.search(value):
        return False
    if _ASSISTANT_VALUE_RE.match(value.strip()):
        return False
    if value.strip().startswith("{") or value.strip().startswith("["):
        return False                             # leaked JSON is not a fact
    return True


def filter_ops_to_user_text(ops: list, user_text: str) -> list:
    """v1.34: facts must come from the USER'S OWN WORDS. The local LLM kept
    storing what the ASSISTANT said about topics the user merely asked about
    ("mrbeast: Jimmy Donaldson"). An add/update op survives only if its value
    shares at least one significant token with the user's utterance (deletes
    pass through — they reference existing keys, not new content)."""
    ut = _tokens(user_text or "")
    out = []
    for op in ops if isinstance(ops, list) else []:
        if not isinstance(op, dict):
            continue
        if str(op.get("op", "")).lower() == "delete":
            out.append(op)
            continue
        vt = _tokens(str(op.get("value", "")))
        if vt and ut and not (vt & ut):
            continue                              # assistant-derived — drop
        out.append(op)
    return out

_STOPWORDS = {
    "the", "a", "an", "is", "are", "was", "were", "i", "my", "me", "you",
    "your", "it", "its", "of", "to", "and", "or", "in", "on", "at", "for",
    "with", "about", "that", "this", "do", "does", "what", "who", "how",
    "have", "has", "be", "am", "sir", "jarvis", "wivexx",
}


def set_logger(logger) -> None:
    global _logger
    _logger = logger


# ── Semantic recall (v1.16): pluggable embedder + cosine similarity ────────────
# The store stays dependency-light: wivexx_api injects a text→vector function
# (local Ollama embeddings).  When it's set, retrieval BLENDS semantic similarity
# with the existing lexical/recency/importance score; when it's None or the model
# is unavailable, behaviour is exactly the old keyword scoring (graceful).
import array as _array
try:
    import numpy as _np
except Exception:
    _np = None

_embedder = None
_SEM_WEIGHT = 2.5         # weight of cosine similarity added to the lexical score
_SEM_THRESHOLD = 0.55     # min cosine for a NO-keyword fact to count as relevant


def set_embedder(fn) -> None:
    """Inject a text→list[float] embedder (or None to disable semantic recall)."""
    global _embedder
    _embedder = fn


def semantic_enabled() -> bool:
    return _embedder is not None


def configure_semantic(weight=None, threshold=None) -> None:
    global _SEM_WEIGHT, _SEM_THRESHOLD
    if weight is not None:
        _SEM_WEIGHT = float(weight)
    if threshold is not None:
        _SEM_THRESHOLD = float(threshold)


def _embed(text):
    if _embedder is None:
        return None
    try:
        v = _embedder(text)
        return v if v else None
    except Exception:
        return None


def _vec_to_blob(vec):
    try:
        return _array.array("f", vec).tobytes() if vec else None
    except Exception:
        return None


def _blob_to_vec(blob):
    try:
        a = _array.array("f")
        a.frombytes(blob)
        return a
    except Exception:
        return None


def _cosine(a, b) -> float:
    try:
        if _np is not None:
            av = _np.asarray(a, dtype="float32")
            bv = _np.asarray(b, dtype="float32")
            na = float(_np.linalg.norm(av))
            nb = float(_np.linalg.norm(bv))
            return float(av.dot(bv) / (na * nb)) if na and nb else 0.0
        dot = sum(x * y for x, y in zip(a, b))
        na = math.sqrt(sum(x * x for x in a))
        nb = math.sqrt(sum(y * y for y in b))
        return dot / (na * nb) if na and nb else 0.0
    except Exception:
        return 0.0


def _semantic_scores(query: str, facts: list) -> dict:
    """{fact_id: cosine_similarity} for the query vs each fact's stored vector.
    Empty dict when semantic recall is off or the query can't be embedded."""
    if _embedder is None:
        return {}
    qv = _embed(query)
    if not qv:
        return {}
    out = {}
    for f in facts:
        blob = f.get("embedding")
        if not blob:
            continue
        v = _blob_to_vec(blob)
        if v is not None:
            out[f["id"]] = _cosine(qv, v)
    return out


_ensured_for_path = ""    # self-healing: migrate once per DB path on first use


def _connect() -> sqlite3.Connection:
    global _ensured_for_path
    if _ensured_for_path != _DB_PATH:
        _ensured_for_path = _DB_PATH
        try:
            _migrate_unlocked()
        except Exception:
            _ensured_for_path = ""
    conn = sqlite3.connect(_DB_PATH, timeout=10)
    conn.execute("PRAGMA journal_mode=WAL;")
    conn.row_factory = sqlite3.Row
    return conn


def _migrate_unlocked() -> None:
    """Create tables / add v1.10 columns. Called lazily by _connect()."""
    conn = sqlite3.connect(_DB_PATH, timeout=10)
    try:
        conn.executescript("""
            CREATE TABLE IF NOT EXISTS core_profile (
                id         INTEGER PRIMARY KEY AUTOINCREMENT,
                key        TEXT UNIQUE NOT NULL,
                value      TEXT NOT NULL,
                updated_at REAL NOT NULL
            );
            CREATE TABLE IF NOT EXISTS topic_summaries (
                id         INTEGER PRIMARY KEY AUTOINCREMENT,
                topic      TEXT UNIQUE NOT NULL,
                summary    TEXT NOT NULL,
                updated_at REAL NOT NULL
            );
            CREATE TABLE IF NOT EXISTS episodic_log (
                id             INTEGER PRIMARY KEY AUTOINCREMENT,
                ts             REAL NOT NULL,
                user_text      TEXT,
                assistant_text TEXT
            );
        """)
        cols = {r[1] for r in conn.execute("PRAGMA table_info(core_profile)")}
        if "importance" not in cols:
            conn.execute("ALTER TABLE core_profile "
                         "ADD COLUMN importance INTEGER DEFAULT 5")
        if "source" not in cols:
            conn.execute("ALTER TABLE core_profile "
                         "ADD COLUMN source TEXT DEFAULT ''")
        if "last_used" not in cols:
            conn.execute("ALTER TABLE core_profile "
                         "ADD COLUMN last_used REAL DEFAULT 0")
        if "use_count" not in cols:
            conn.execute("ALTER TABLE core_profile "
                         "ADD COLUMN use_count INTEGER DEFAULT 0")
        if "embedding" not in cols:                # v1.16: semantic recall vector
            conn.execute("ALTER TABLE core_profile ADD COLUMN embedding BLOB")
        conn.commit()
    finally:
        conn.close()


# ── Schema / migration ─────────────────────────────────────────────────────────
def ensure_schema() -> None:
    """Create tables and migrate older DBs in place (adds the v1.10 columns).
    Also runs lazily on first use via _connect(), so the store is always
    safe to import directly."""
    global _ensured_for_path
    with _LOCK:
        _migrate_unlocked()
        _ensured_for_path = _DB_PATH


# ── CRUD (used by the GUI manager and voice commands) ─────────────────────────
def list_facts(search: str = "") -> list:
    with _LOCK:
        conn = _connect()
        try:
            rows = conn.execute(
                "SELECT * FROM core_profile ORDER BY importance DESC, "
                "updated_at DESC").fetchall()
        finally:
            conn.close()
    facts = [dict(r) for r in rows]
    s = (search or "").lower().strip()
    if s:
        facts = [f for f in facts
                 if s in f["key"].lower() or s in f["value"].lower()]
    return facts


def add_fact(key: str, value: str, importance: int = 5,
             source: str = "manual") -> bool:
    key = (key or "").strip().lower()[:50]
    value = (value or "").strip()[:300]
    if not _fact_ok(key, value):
        return False
    importance = max(1, min(10, int(importance)))
    # embed BEFORE taking the lock (the embedder may do local network I/O)
    emb = _vec_to_blob(_embed(f"{key}: {value}")) if _embedder is not None else None
    with _LOCK:
        conn = _connect()
        try:
            if emb is not None:
                conn.execute(
                    "INSERT INTO core_profile (key, value, updated_at, importance,"
                    " source, last_used, use_count, embedding) "
                    "VALUES (?, ?, ?, ?, ?, 0, 0, ?) "
                    "ON CONFLICT(key) DO UPDATE SET value=excluded.value, "
                    "updated_at=excluded.updated_at, importance=excluded.importance,"
                    " source=excluded.source, embedding=excluded.embedding",
                    (key, value, time.time(), importance, source, emb))
            else:
                conn.execute(
                    "INSERT INTO core_profile (key, value, updated_at, importance, "
                    "source, last_used, use_count) VALUES (?, ?, ?, ?, ?, 0, 0) "
                    "ON CONFLICT(key) DO UPDATE SET value=excluded.value, "
                    "updated_at=excluded.updated_at, importance=excluded.importance, "
                    "source=excluded.source",
                    (key, value, time.time(), importance, source))
            conn.commit()
            return True
        finally:
            conn.close()


def update_fact(fact_id: int, key: Optional[str] = None,
                value: Optional[str] = None,
                importance: Optional[int] = None) -> bool:
    sets, args = [], []
    if key is not None and key.strip():
        sets.append("key = ?")
        args.append(key.strip().lower()[:50])
    if value is not None and value.strip():
        v = value.strip()[:300]
        sets.append("value = ?")
        args.append(v)
        emb = _vec_to_blob(_embed(((key + ": ") if key else "") + v)) \
            if _embedder is not None else None
        if emb is not None:                       # keep the vector in sync (v1.16)
            sets.append("embedding = ?")
            args.append(emb)
    if importance is not None:
        sets.append("importance = ?")
        args.append(max(1, min(10, int(importance))))
    if not sets:
        return False
    sets.append("updated_at = ?")
    args.append(time.time())
    args.append(int(fact_id))
    with _LOCK:
        conn = _connect()
        try:
            cur = conn.execute(
                f"UPDATE core_profile SET {', '.join(sets)} WHERE id = ?", args)
            conn.commit()
            return cur.rowcount > 0
        except sqlite3.IntegrityError:
            return False    # key collision with another fact
        finally:
            conn.close()


def delete_fact(fact_id: int) -> bool:
    with _LOCK:
        conn = _connect()
        try:
            cur = conn.execute("DELETE FROM core_profile WHERE id = ?",
                               (int(fact_id),))
            conn.commit()
            return cur.rowcount > 0
        finally:
            conn.close()


def list_topics() -> list:
    with _LOCK:
        conn = _connect()
        try:
            rows = conn.execute(
                "SELECT * FROM topic_summaries ORDER BY updated_at DESC"
            ).fetchall()
        finally:
            conn.close()
    return [dict(r) for r in rows]


def delete_topic(topic_id: int) -> bool:
    with _LOCK:
        conn = _connect()
        try:
            cur = conn.execute("DELETE FROM topic_summaries WHERE id = ?",
                               (int(topic_id),))
            conn.commit()
            return cur.rowcount > 0
        finally:
            conn.close()


# ── Relevance scoring + retrieval ──────────────────────────────────────────────
def _strip_accents(s: str) -> str:
    # fold á→a, ñ→n so accented Spanish words tokenize as ONE whole word (the old
    # ASCII-only tokenizer split "cumpleaños"→"cumplea"+"os"). No-op for English.
    return "".join(c for c in unicodedata.normalize("NFD", s)
                   if unicodedata.category(c) != "Mn")


def _tokens(text: str) -> set:
    return {t for t in re.findall(r"[a-z0-9]+", _strip_accents((text or "").lower()))
            if t not in _STOPWORDS and len(t) > 2}


# v1.33: recall queries rarely use the words facts were stored under ("where do
# I live" vs key "location"). Without an embedder there was no bridge at all, so
# common query words expand to the fact keys they usually mean. Cheap, offline,
# and only ADDS candidate tokens — never removes a direct keyword match.
_QUERY_ALIASES = {
    "live": {"location", "home", "city"}, "home": {"location"},
    "city": {"location"}, "from": {"location"}, "address": {"location"},
    "called": {"name"}, "job": {"work", "profession"},
    "work": {"profession", "job"}, "profession": {"work"},
    "born": {"birthday"}, "old": {"birthday", "age"}, "age": {"birthday"},
    "camera": {"gear"}, "equipment": {"gear"}, "setup": {"gear", "pc"},
    "shoot": {"gear", "camera", "work"},
    "family": {"sister", "brother", "mom", "dad", "mother", "father",
               "wife", "husband", "son", "daughter"},
    "pet": {"dog", "cat"}, "pets": {"dog", "cat"},
    "studying": {"studies"}, "studied": {"studies"}, "study": {"studies"},
    # v1.46: Spanish recall words → keys (ES facts store with English keys for
    # name/location/work/birthday/studies, Spanish keys for relations/pets).
    # NOTE: alias keys/values are ACCENT-FOLDED to match `_tokens` (mamá→mama).
    "nombre": {"name"}, "llamo": {"name"}, "llamas": {"name"},
    "vivo": {"location"}, "vives": {"location"}, "ciudad": {"location"},
    "trabajo": {"work", "profession"}, "trabajas": {"work", "profession"},
    "chamba": {"work"}, "cumpleanos": {"birthday"},
    "estudio": {"studies"}, "estudias": {"studies"}, "estudie": {"studies"},
    "mascota": {"perro", "gato", "dog", "cat"}, "mascotas": {"perro", "gato", "dog", "cat"},
    "perro": {"perro", "dog"}, "perros": {"perro", "dog"},
    "gato": {"gato", "cat"}, "gatos": {"gato", "cat"},
    "familia": {"hermano", "hermana", "mama", "papa", "madre", "padre",
                "esposa", "esposo", "hijo", "hija", "novia", "novio",
                "sister", "brother", "mom", "dad", "mother", "father"},
}


def _expand_query_tokens(toks: set) -> set:
    out = set(toks)
    for t in toks:
        out |= _QUERY_ALIASES.get(t, set())
    return out


def _score(fact: dict, q_tokens: set, now: float) -> float:
    """Generative-Agents style: relevance + importance + recency + use."""
    f_tokens = _tokens(fact["key"] + " " + fact["value"])
    overlap = len(q_tokens & f_tokens) / max(1, len(q_tokens)) if q_tokens else 0.0
    age_days = max(0.0, (now - (fact.get("updated_at") or 0)) / 86400)
    recency = math.exp(-age_days / 120)                # slow half-life
    importance = (fact.get("importance") or 5) / 10
    reinforcement = min(int(fact.get("use_count") or 0), 8) * 0.03
    return overlap * 3.0 + importance + recency * 0.4 + reinforcement


def find_facts(query: str, limit: int = 6, min_overlap: bool = True) -> list:
    """Facts most relevant to *query*, best first.  Blends semantic similarity
    (when an embedder is set) with the lexical/recency score, and lets a strong
    semantic match through even with no shared keywords."""
    facts = list_facts()
    q_tokens = _expand_query_tokens(_tokens(query))
    now = time.time()
    sem = _semantic_scores(query, facts)
    scored = []
    for f in facts:
        sim = sem.get(f["id"], 0.0)
        s = _score(f, q_tokens, now) + _SEM_WEIGHT * sim
        has_overlap = bool(q_tokens & _tokens(f["key"] + " " + f["value"]))
        if min_overlap and q_tokens and not has_overlap and sim < _SEM_THRESHOLD:
            continue
        scored.append((s, f))
    scored.sort(key=lambda t: t[0], reverse=True)
    return [f for _, f in scored[:limit]]


def touch_facts(fact_ids: list) -> None:
    """Reinforcement: retrieved memories get used → they rank higher later."""
    if not fact_ids:
        return
    with _LOCK:
        conn = _connect()
        try:
            conn.executemany(
                "UPDATE core_profile SET last_used = ?, use_count = use_count + 1 "
                "WHERE id = ?",
                [(time.time(), int(i)) for i in fact_ids])
            conn.commit()
        finally:
            conn.close()


def embed_missing(limit: int = 1000) -> int:
    """Backfill vectors for facts created before semantic recall (or while the
    embedder was offline).  Safe no-op when no embedder is set.  Returns count."""
    if _embedder is None:
        return 0
    done = 0
    for f in list_facts():
        if f.get("embedding"):
            continue
        emb = _vec_to_blob(_embed(f"{f['key']}: {f['value']}"))
        if emb is None:
            continue
        with _LOCK:
            conn = _connect()
            try:
                conn.execute("UPDATE core_profile SET embedding = ? WHERE id = ?",
                             (emb, f["id"]))
                conn.commit()
            finally:
                conn.close()
        done += 1
        if done >= limit:
            break
    if _logger and done:
        _logger.info("memory: backfilled %d fact embedding(s)", done)
    return done


def relevant_context(question: str, max_facts: int = 8,
                     max_topics: int = 3) -> str:
    """Build the memory block for the system prompt:
    identity core (importance ≥ 8, always) + facts relevant to the question
    + matching topic summaries.  Returns '' when there's nothing useful."""
    try:
        all_facts = list_facts()
        if not all_facts and not list_topics():
            return ""
        q_tokens = _expand_query_tokens(_tokens(question))
        now = time.time()

        sem = _semantic_scores(question, all_facts)
        core = [f for f in all_facts if (f.get("importance") or 5) >= 8][:4]
        core_ids = {f["id"] for f in core}

        def _blended(f):
            return _score(f, q_tokens, now) + _SEM_WEIGHT * sem.get(f["id"], 0.0)

        scored = sorted(
            ((_blended(f), f) for f in all_facts if f["id"] not in core_ids),
            key=lambda t: t[0], reverse=True)
        # relevant = shares a keyword OR is a strong semantic match (v1.16);
        # when nothing matches (greetings etc.) a few recent high-scorers fill in
        relevant = [f for _, f in scored
                    if (q_tokens and (q_tokens & _tokens(f["key"] + " " + f["value"])))
                    or sem.get(f["id"], 0.0) >= _SEM_THRESHOLD]
        filler = [f for _, f in scored if f not in relevant]
        chosen = core + relevant[:max_facts - len(core)]
        filler_budget = 2 if relevant else max(0, max_facts - len(chosen))
        chosen += filler[:min(filler_budget, max(0, max_facts - len(chosen)))]
        chosen = chosen[:max_facts]

        parts: list = []
        if chosen:
            lines = "\n".join(f"  • {f['key']}: {f['value']}" for f in chosen)
            parts.append(f"[Persistent facts about the user]\n{lines}")

        topics = list_topics()
        rel_topics = [t for t in topics
                      if q_tokens & _tokens(t["topic"] + " " + t["summary"])]
        # v1.33: when NO topic matches the question, inject only the single most
        # recent one (continuity) — pushing 3 unrelated summaries into every
        # prompt wasted tokens and polluted the model's context.
        rel_topics = rel_topics[:max_topics] if rel_topics else topics[:1]
        if rel_topics:
            lines = "\n".join(f"  • {t['topic']}: {t['summary']}"
                              for t in rel_topics)
            parts.append(f"[Relevant prior topics]\n{lines}")

        touch_facts([f["id"] for f in chosen])
        if not parts:
            return ""
        # The notes above are mostly English regardless of what language the user
        # speaks. Without this rule the block used to DRAG replies into English
        # ("¿recuerdas algo?" → an English answer) — the reply must follow the
        # conversation, not the notes.
        return ("\n\n=== Wivexx LONG-TERM MEMORY ===\n"
                + "\n\n".join(parts)
                + "\n(These are internal notes; their language is incidental. "
                  "Always reply in the language the user is speaking right now, "
                  "translating any note you use naturally — proper names stay "
                  "as written.)"
                + "\n=== END MEMORY ===\n")
    except Exception as e:
        if _logger:
            _logger.debug("memory context failed: %s", e)
        return ""


# ── Rule-based fact extractor (v1.33) ──────────────────────────────────────────
# The LLM consolidator needs a local Ollama — which most public installs (and
# GPU-starved machines) don't have, so automatic memory silently did NOTHING for
# them. This extractor covers the clearest self-statement patterns with plain
# regexes: high PRECISION over recall (a missed fact is fine, a junk fact is
# not). Output feeds apply_ops(), so dedup/reconciliation rules still apply.

_FEELING_WORDS = re.compile(
    r"\b(?:tired|bored|happy|sad|sorry|sure|okay|fine|busy|ready|little|bit|"
    r"hurry|mood|afraid|scared|glad|curious|confused|lost|done|back|here)\b",
    re.IGNORECASE)

_NAME_TOKEN = r"([A-Za-z][a-zA-Z'\-]{1,29})"
_FREE_VALUE = r"([\w][\w\s'\-\.]{2,50}?)"
_END = r"(?=[.,;!?]|$|\s+(?:and|but|so|because|which|though)\b)"
# Spanish variants: name token allows accents/ñ; clause end uses Spanish conjunctions.
_ES_NAME = r"([A-Za-zÁÉÍÓÚÜÑáéíóúüñ][a-zA-ZÁÉÍÓÚÜÑáéíóúüñ'\-]{1,29})"
_ES_END = r"(?=[.,;!?]|$|\s+(?:y|pero|porque|que|aunque|cuando|aparte)\b)"

# (regex, key_template, importance) — {N} in key_template = capture group N
_RULE_PATTERNS = [
    (re.compile(r"\bmy\s+name\s+is\s+" + _NAME_TOKEN, re.I), "name", 9),
    (re.compile(r"\b(?:please\s+)?call\s+me\s+" + _NAME_TOKEN + r"\b(?!\s+(?:back|later|when|if|at|an?\b))", re.I), "name", 9),
    (re.compile(r"\bmy\s+(?:favorite|favourite)\s+([a-z][a-z\s]{2,20}?)\s+is\s+" + _FREE_VALUE + _END, re.I), "favorite {1}", 5),
    (re.compile(r"\bi\s+live\s+in\s+" + _FREE_VALUE + _END, re.I), "location", 7),
    (re.compile(r"\bi\s+work\s+(?:as\s+an?\s+|at\s+|for\s+)" + _FREE_VALUE + _END, re.I), "work", 7),
    (re.compile(r"\bmy\s+birthday\s+is\s+(?:on\s+)?" + _FREE_VALUE + _END, re.I), "birthday", 8),
    (re.compile(r"\bmy\s+(wife|husband|girlfriend|boyfriend|fianc[ée]e?|sister|brother|"
                r"mom|mother|dad|father|son|daughter|dog|cat)(?:'s\s+name)?\s+is\s+"
                r"(?:called\s+|named\s+)?" + _NAME_TOKEN, re.I), "{1}", 7),
    (re.compile(r"\bi\s+(?:shoot\s+with|drive|ride)\s+an?\s+" + _FREE_VALUE + _END, re.I), "gear", 6),
    (re.compile(r"\bi\s+(?:study|studied)\s+" + _FREE_VALUE + _END, re.I), "studies", 6),
    (re.compile(r"\bi\s*(?:'m|\s+am)\s+an?\s+" + _FREE_VALUE + _END, re.I), "profession", 6),
    # ── Spanish fact patterns (v1.45 — offline ES memory parity, no LLM) ──
    (re.compile(r"\bme\s+llamo\s+(?!(?:la|el|un|una|lo|los|las)\b)" + _ES_NAME, re.I), "name", 9),
    (re.compile(r"\bmi\s+nombre\s+es\s+" + _ES_NAME, re.I), "name", 9),
    (re.compile(r"\bvivo\s+en\s+" + _FREE_VALUE + _ES_END, re.I), "location", 7),
    (re.compile(r"\btrabajo\s+(?:como\s+(?:un[ao]?\s+)?(?!loco|burro|esclavo|mula|bestia|animal|enfermo|condenad[oa])|de\s+|en\s+)" + _FREE_VALUE + _ES_END, re.I), "work", 7),
    (re.compile(r"\bmi\s+cumplea[ñn]os\s+es\s+(?:el\s+)?" + _FREE_VALUE + _ES_END, re.I), "birthday", 8),
    (re.compile(r"\bmi\s+(esposa|esposo|novia|novio|hermana|hermano|mam[aá]|pap[aá]|madre|padre|"
                r"hijo|hija|perro|perra|gato|gata)\s+se\s+llama\s+" + _ES_NAME, re.I), "{1}", 7),
    (re.compile(r"\b(?:estudio|estudi[eé])\s+(?!para\b|mucho\b|poco\b|un\s+poco\b|toda\b|todo\b|en\b)" + _FREE_VALUE + _ES_END, re.I), "studies", 6),
]


def extract_facts_rules(text: str) -> list:
    """High-precision, LLM-free fact extraction from ONE user utterance.
    Returns apply_ops-shaped [{op, key, value, importance}] — usually empty."""
    t = (text or "").strip()
    if not t or len(t) > 400 or t.endswith("?"):
        return []
    ops, seen = [], set()
    for rx, key_tpl, imp in _RULE_PATTERNS:
        for m in rx.finditer(t):
            groups = m.groups()
            key = key_tpl
            for i, g in enumerate(groups, 1):
                key = key.replace("{%d}" % i, (g or "").strip().lower())
            value = (groups[-1] or "").strip(" .,!'\"")
            key = key.strip().lower()[:50]
            if not key or not value or key in KEY_BLACKLIST or key in seen:
                continue
            # precision filters: no feelings ("i'm a little tired"), no
            # pronoun/filler heads, nothing that reads like a clause
            if _FEELING_WORDS.search(value):
                continue
            if re.match(r"^(?:not|just|so|too|very|going|gonna|about|kind|sort)\b",
                        value, re.I):
                continue
            if key == "profession" and len(value.split()) > 4:
                continue
            seen.add(key)
            ops.append({"op": "add", "key": key, "value": value[:120],
                        "importance": imp})
    return ops


# ── Episodic log retention (v1.33) ──────────────────────────────────────────────
def prune_episodic(max_rows: int = 2000) -> int:
    """Keep only the newest *max_rows* raw turns — the episodic log was
    append-only and grew without bound. Returns rows dropped."""
    with _LOCK:
        conn = _connect()
        try:
            cur = conn.execute(
                "DELETE FROM episodic_log WHERE id NOT IN ("
                "  SELECT id FROM episodic_log ORDER BY ts DESC LIMIT ?)",
                (int(max_rows),))
            conn.commit()
            return cur.rowcount
        finally:
            conn.close()


# ── Consolidator operations (Mem0-style ADD / UPDATE / DELETE) ─────────────────
def apply_ops(ops: list, source: str = "auto") -> dict:
    """Apply [{op, key, value, importance}] decided by the consolidator.
    Quality filters identical for every path. Returns counters."""
    counts = {"add": 0, "update": 0, "delete": 0, "rejected": 0}
    for op in ops if isinstance(ops, list) else []:
        if not isinstance(op, dict):
            continue
        kind = str(op.get("op", "")).lower()
        key = str(op.get("key", "")).strip().lower()[:50]
        value = str(op.get("value", "")).strip()[:300]
        try:
            imp = max(1, min(10, int(op.get("importance", 5))))
        except Exception:
            imp = 5
        if not key or key in KEY_BLACKLIST:
            counts["rejected"] += 1
            continue
        if kind == "delete":
            with _LOCK:
                conn = _connect()
                try:
                    cur = conn.execute(
                        "DELETE FROM core_profile WHERE key = ?", (key,))
                    conn.commit()
                finally:
                    conn.close()
            counts["delete"] += 1 if cur.rowcount else 0
            continue
        if kind not in ("add", "update"):
            continue
        if not _fact_ok(key, value):
            counts["rejected"] += 1
            continue
        existing = next((f for f in list_facts() if f["key"] == key), None)
        if existing:
            import difflib
            if difflib.SequenceMatcher(
                    None, existing["value"].lower(), value.lower()).ratio() > 0.92:
                continue                      # nothing new
            # never let an automatic update clobber a manual/explicit memory
            # with something less important
            if (existing.get("source") in ("manual", "explicit")
                    and source == "auto"
                    and imp < (existing.get("importance") or 5)):
                counts["rejected"] += 1
                continue
        add_fact(key, value, importance=imp, source=source)
        counts["update" if existing else "add"] += 1
    return counts


def prune(max_facts: int = 60) -> int:
    """Forget-by-decay: when over capacity, drop the lowest-scored stale,
    low-importance, never-used facts (manual/explicit ones are safe)."""
    facts = list_facts()
    if len(facts) <= max_facts:
        return 0
    now = time.time()
    candidates = [f for f in facts
                  if f.get("source") not in ("manual", "explicit")
                  and (f.get("importance") or 5) <= 4]
    candidates.sort(key=lambda f: _score(f, set(), now))
    to_drop = candidates[:len(facts) - max_facts]
    for f in to_drop:
        delete_fact(f["id"])
    if _logger and to_drop:
        _logger.info("memory: pruned %d low-value facts", len(to_drop))
    return len(to_drop)
