"""Jarvis memory service — the shared brain (deploy on Railway).

DUMB ON PURPOSE. There is NO model on this server. The intelligence lives in
whatever model is already running on a device (Jarvis's Claude on Mac/Windows
now, a local model on the phone later, future screenshot/transcribe daemons) —
each one decides WHAT to remember and WHEN to recall, and structures the fact
itself. This service only:
  • stores each memory as a node with a LOCAL embedding (FastEmbed, CPU — no
    LLM, no API), and
  • answers searches by MEANING (vector similarity) + returns CONNECTED facts
    (1 graph hop), so a vague query pulls back the linked context.

Backed by FalkorDB (a graph DB with native vectors). Every device/daemon is a
thin client that POSTs facts and GETs searches over HTTPS.

Endpoints
  GET  /health                 — liveness + memory/entity count + embed model + reflection
  POST /remember               — {text, kind?, when?, importance?, raw?, source?,
                                  group?, links?, entities?} → {id, op, auto_linked, entities}
                                  (server-side dedup/supersede WRITE PIPELINE)
  GET  /search?q=&k=&group=&hops=&as_of=&when_from=&when_to= — connected facts by meaning
  GET  /memories?group=&limit=&offset= — newest-first list (HUD viewer)
  GET  /graph?group=           — whole graph {nodes,edges} (Memory+Entity+REL)
  POST /link                   — {from, to, type}
  POST /unlink                 — {from_id, to} cut the edge(s) between two nodes
  POST /backfill_links         — {dry_run?,threshold?} auto-link all existing memories
  GET  /entity?name=          — everything mentioning an entity (facts + relations)
  GET  /profile?name=&group=  — fast RAW one-entity aggregate for a HUD card: entity + summary + read +
                                quick contacts + relations + timeline + facts_by_kind + tasks + counts +
                                has_composed (fuzzy resolver: difflib ratio ≥0.85 beats a substring hit)
  POST /dossier                — {name, group?, force?} an LLM READS the /profile aggregate and COMPOSES
                                an organized, adaptive card {header, sections[text|kv|chips|list]};
                                cached on the entity (24h / fact-count) → {composed, cached}
  POST /entity/update         — {key, group?, etype?, name?} fix an entity's type / display name
  POST /summary                — {name, text} upsert an entity's current-state summary
  POST /retire                 — {id} soft-retire a memory (stale but kept, for dedup)
  POST /attach_entities        — {memory_id, entities} attach entities (backfill hook)
  POST /relation               — {from, to, rtype} entity↔entity edge (resolved ends)
  POST /entities/merge         — {keep_key, merge_keys} fold duplicate entities into one
  POST /entities/dedup_backfill — {dry_run} scan/auto-merge duplicate entities
  POST /reembed                — {batch} re-embed every node with the current model
  GET  /export                 — complete dump (feeds nightly backups)
  POST /reflection_ran         — {status, note} record the reflection engine's last run
  GET  /core_block · POST /core — Ahmed's core context block (singleton)
  POST /insight                — {text,title,itype} reflection finding · GET /insights

RECALL is SALIENCE-ranked: relevance × recency × confidence × reinforcement ×
importance (latest+firm+often-recalled facts surface first; nothing is deleted).
Facts carry a `confidence` (0..1) and an `importance` (1..10, default 5).

WRITE PIPELINE (Mem0-style, server-side): /remember embeds the text, pulls the
top similar valid memories, and — if DEEPSEEK_API_KEY is set — asks ONE cheap
DeepSeek call whether to ADD / UPDATE (supersede a specific memory) / NOOP (a
duplicate), cleans a raw utterance into an atomic third-person fact, extracts the
event date + entities + relations, and marks single-valued facts (srel) so a new
value auto-retires the old. NO key ⇒ graceful degrade (cosine≥0.95 ⇒ noop, else
plain add). An LLM failure NEVER loses a memory — it falls back to a plain add.

BI-TEMPORAL: every memory carries valid_from (event/creation time) and valid_to
(null while current, set on retire/supersede) alongside the legacy `valid` bool +
invalid_at. /search?as_of=<ISO> answers "what was true then".

AUTO-LINKING (Layer 1): every write draws edges to the most similar existing
memories (embedding + shared words) — no LLM. AUTOLINK=0 off.
ENTITIES (Layer 2): entities resolve by exact key / alias / fuzzy (embedding +
difflib) / DeepSeek adjudication into shared :Entity nodes, wired by
(:Memory)-[:MENTIONS]->(:Entity) and (:Entity)-[:REL]->(:Entity).
  POST /supersede              — {old_id, text, ...} mark old fact stale + add new
  POST /forget                 — {id} permanently delete a memory + its links

Env: FALKOR_HOST/FALKOR_PORT, MEMORY_API_KEY (bearer the devices send),
     MEMORY_GROUP (default namespace), EMBED_MODEL, EMBED_DIM,
     DEEPSEEK_API_KEY (optional — enables the LLM write pipeline), WRITE_MODEL,
     DEEPSEEK_URL.
"""
from __future__ import annotations

import difflib
import json
import math
import os
import re
import urllib.request
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, Field

from falkordb import FalkorDB
from fastembed import TextEmbedding

import mcp_server  # streamable-HTTP MCP endpoint (mounted below; purely additive)

API_KEY = os.environ.get("MEMORY_API_KEY", "")
GROUP_DEFAULT = os.environ.get("MEMORY_GROUP", "ahmed")
GRAPH_NAME = os.environ.get("MEMORY_GRAPH", "jarvis")
# Multilingual by default (Ahmed speaks Arabic). paraphrase-multilingual-MiniLM
# is 384d — same dimension as the old English-only bge-small — so switching the
# model needs only a /reembed (no vector-index rebuild). Override both together.
EMBED_MODEL = os.environ.get(
    "EMBED_MODEL", "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2")
EMBED_DIM = int(os.environ.get("EMBED_DIM", "384"))

# WRITE PIPELINE — one cheap DeepSeek call decides add/update/noop + cleans the
# fact + extracts date/entities/relations. Off (graceful degrade) with no key.
DEEPSEEK_KEY = os.environ.get("DEEPSEEK_API_KEY", "")
DS_URL = os.environ.get("DEEPSEEK_URL", "https://api.deepseek.com").rstrip("/")
WRITE_MODEL = os.environ.get("WRITE_MODEL", "deepseek-v4-flash")
# DOSSIER — a richer model READS the raw /profile aggregate and COMPOSES the
# organized, adaptive profile card. One call per /dossier, then cached on the
# entity node so repeat opens are instant (see POST /dossier).
# v4 models spend tokens on REASONING before the JSON content — the budget
# must cover both or the content arrives empty/truncated. flash default: the
# card should organize in seconds, not tens of seconds (env-override to -pro).
DOSSIER_MODEL = os.environ.get("DOSSIER_MODEL", "deepseek-v4-flash")

# Layer 1 — AUTO-LINKING. Every save draws edges to the most-similar existing
# memories (embedding similarity + shared significant words), so the graph wires
# itself with no LLM and no manual work. Tunable via env; AUTOLINK=0 disables.
AUTOLINK = os.environ.get("AUTOLINK", "1") != "0"
def _sim(distance) -> float:
    """FalkorDB's db.idx.vector.queryNodes YIELDs cosine DISTANCE (0 = identical,
    ~1 = unrelated), NOT similarity. Convert at every consumption site so all
    thresholds/scores in this file read as plain cosine similarity."""
    return 1.0 - float(distance)


AUTOLINK_THRESHOLD = float(os.environ.get("AUTOLINK_THRESHOLD", "0.60"))
# 0.60 is calibrated for TRUE cosine (paraphrase-multilingual-MiniLM): related
# facts land ~0.55-0.75, unrelated ~0.0-0.3. The old 0.85 default was tuned to
# bge-small's inflated similarity scale and would stop meaning-links entirely.
INSIGHT_DEDUP = float(os.environ.get("INSIGHT_DEDUP", "0.78"))
AUTOLINK_MAX = int(os.environ.get("AUTOLINK_MAX", "5"))

_graph = None
_embedder: TextEmbedding | None = None


def _embed(text: str) -> list[float]:
    return [float(x) for x in next(_embedder.embed([text]))]


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def _today() -> str:
    return datetime.now(timezone.utc).date().isoformat()


def _deepseek_json(prompt: str, max_tokens: int = 800, model: str | None = None,
                   temperature: float = 0.1) -> dict | None:
    """ONE DeepSeek call that must return a JSON OBJECT (response_format json).

    Stdlib urllib only (same dependency-light pattern as reflection-service). Any
    failure — no key, network, non-JSON — RAISES so callers fall back to a plain
    add and never lose a memory. Returns the parsed dict. `model`/`temperature`
    are overridable (the dossier composer uses a richer model at temp 0.3); the
    defaults keep every existing write-pipeline call unchanged."""
    if not DEEPSEEK_KEY:
        raise RuntimeError("no DEEPSEEK_API_KEY")
    body = json.dumps({
        "model": model or WRITE_MODEL, "temperature": temperature,
        "stream": False, "max_tokens": max_tokens,
        "response_format": {"type": "json_object"},
        "messages": [{"role": "user", "content": prompt}],
    }).encode()
    req = urllib.request.Request(
        DS_URL + "/chat/completions", data=body,
        headers={"Content-Type": "application/json",
                 "Authorization": f"Bearer {DEEPSEEK_KEY}"})
    with urllib.request.urlopen(req, timeout=60) as r:
        o = json.loads(r.read())
    content = (((o.get("choices") or [{}])[0].get("message") or {})
               .get("content", "") or "").strip()
    return json.loads(content)


def _cosine(a, b) -> float:
    try:
        a = [float(x) for x in a]
        b = [float(x) for x in b]
    except Exception:  # noqa: BLE001
        return 0.0
    if not a or not b or len(a) != len(b):
        return 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))
    if na == 0.0 or nb == 0.0:
        return 0.0
    return dot / (na * nb)


def _init_graph() -> None:
    global _graph, _embedder, EMBED_DIM
    _embedder = TextEmbedding(EMBED_MODEL)
    # EMBED_DIM is DERIVED from the model (the env value is only a hint/fallback),
    # so overriding EMBED_MODEL alone keeps the vector indexes dimensionally
    # correct without a second env var to remember.
    try:
        real_dim = len(_embed("dimension probe"))
        if real_dim and real_dim != EMBED_DIM:
            print(f"[startup] EMBED_DIM {EMBED_DIM} -> {real_dim} (from model)",
                  flush=True)
            EMBED_DIM = real_dim
    except Exception as e:  # noqa: BLE001 — fall back to the env/default dim
        print(f"[startup] dim-probe note: {e}", flush=True)
    db = FalkorDB(host=os.environ.get("FALKOR_HOST", "localhost"),
                  port=int(os.environ.get("FALKOR_PORT", "6379")),
                  password=os.environ.get("FALKOR_PASSWORD") or None)
    _graph = db.select_graph(GRAPH_NAME)
    # a vector index over memory embeddings (idempotent — ignore "already exists")
    try:
        _graph.query(
            "CREATE VECTOR INDEX FOR (m:Memory) ON (m.embedding) "
            f"OPTIONS {{dimension: {EMBED_DIM}, similarityFunction: 'cosine'}}")
    except Exception as e:  # noqa: BLE001
        if "already" not in str(e).lower():
            print(f"[startup] vector index note: {e}", flush=True)
    try:  # Layer 2 — fast entity MERGE by canonical key
        _graph.query("CREATE INDEX FOR (e:Entity) ON (e.key)")
    except Exception as e:  # noqa: BLE001
        if "already" not in str(e).lower():
            print(f"[startup] entity index note: {e}", flush=True)
    try:  # reflection insights — vector index for dedup
        _graph.query(
            "CREATE VECTOR INDEX FOR (i:Insight) ON (i.embedding) "
            f"OPTIONS {{dimension: {EMBED_DIM}, similarityFunction: 'cosine'}}")
    except Exception as e:  # noqa: BLE001
        if "already" not in str(e).lower():
            print(f"[startup] insight index note: {e}", flush=True)
    try:  # entity-name vector index — powers fuzzy entity resolution
        _graph.query(
            "CREATE VECTOR INDEX FOR (e:Entity) ON (e.embedding) "
            f"OPTIONS {{dimension: {EMBED_DIM}, similarityFunction: 'cosine'}}")
    except Exception as e:  # noqa: BLE001
        if "already" not in str(e).lower():
            print(f"[startup] entity vector index note: {e}", flush=True)
    for _stmt in ("CREATE INDEX FOR (m:Memory) ON (m.group)",
                  "CREATE INDEX FOR (m:Memory) ON (m.created_at)"):
        try:  # plain/range indexes so group + newest-first scans stay fast
            _graph.query(_stmt)
        except Exception as e:  # noqa: BLE001
            if "already" not in str(e).lower():
                print(f"[startup] range index note: {e}", flush=True)
    print(f"memory service ready ({EMBED_MODEL}, dim {EMBED_DIM})", flush=True)


@asynccontextmanager
async def _lifespan(app: FastAPI):
    # Same startup work as before (graph + embedder + indexes), now via lifespan
    # so the MCP session manager's task group can run alongside it.
    _init_graph()
    async with mcp_server.lifespan():
        yield


app = FastAPI(title="Jarvis Memory", lifespan=_lifespan)
mcp_server.mount(app)  # attach /mcp and /mcp/<token> (streamable-HTTP MCP doors)


def _auth(authorization: str | None) -> None:
    if API_KEY and authorization != f"Bearer {API_KEY}":
        raise HTTPException(status_code=401, detail="bad or missing API key")


class Remember(BaseModel):
    text: str
    kind: str | None = "fact"          # fact/event/preference/observation/…
    when: str | None = None            # ISO event date the memory REFERS to
    importance: int | None = 5         # 1..10 how much it should weigh in recall
    raw: bool | None = False           # True = a raw utterance needing cleanup
    source: str | None = "jarvis"      # which brain wrote it
    group: str | None = None           # namespace (default MEMORY_GROUP)
    links: list[dict] | None = None    # [{"to": <id>, "type": "for"}]
    entities: list[dict] | None = None  # LAYER 2: [{"name":..,"type":..}]
    confidence: float | None = None    # 0..1 how firm the fact is (recall weight)
    tone: str | None = None            # vocal delivery it was said with ("slow, flat")
    historical: bool | None = False    # IMPORT mode: an OLD fact being backfilled
    #   — it may be deduped (noop) but may NEVER retire/supersede anything
    #   already stored: current memory always outranks an old import.


class Link(BaseModel):
    from_id: str
    to: str
    type: str | None = "related"


class Supersede(BaseModel):
    old_id: str
    text: str
    kind: str | None = "fact"
    when: str | None = None
    source: str | None = "jarvis"
    group: str | None = None
    entities: list[dict] | None = None
    confidence: float | None = None
    tone: str | None = None


@app.get("/health")
def health():
    """Liveness — unauthenticated (same as before). Reports memory/entity counts,
    the reflection engine's last run, and the active embedding model + dim."""
    try:
        n = _graph.query("MATCH (m:Memory) RETURN count(m)").result_set[0][0]
    except Exception:  # noqa: BLE001
        n = -1
    try:
        ec = _graph.query("MATCH (e:Entity) RETURN count(e)").result_set[0][0]
    except Exception:  # noqa: BLE001
        ec = -1
    last = ""
    try:
        r = _graph.query(
            "MATCH (x:Meta {id:'reflection', group:$g}) RETURN x.last_run",
            {"g": GROUP_DEFAULT}).result_set
        last = (r[0][0] or "") if r else ""
    except Exception:  # noqa: BLE001
        pass
    return {"ok": True, "memory_count": n, "entity_count": ec,
            "reflection_last_run": last, "embed_model": EMBED_MODEL,
            "dim": EMBED_DIM, "memories": n}  # `memories` kept for old callers


def _create(text: str, kind: str, when: str | None, source: str,
            group: str, emb: list[float] | None = None,
            confidence: float = 0.75, tone: str | None = None,
            importance: int = 5, srel: str | None = None) -> str:
    """Create a :Memory node. Bi-temporal: valid_from = the event date (`when`)
    if given, else creation time; valid_to is left UNSET (= null = current) and
    only set on retire/supersede. `srel` (subject|relation, lowercased) marks a
    single-valued fact so a newer value can deterministically supersede it."""
    mid = uuid.uuid4().hex
    if emb is None:
        emb = _embed(text)
    now = _now_iso()
    _graph.query(
        "CREATE (m:Memory {id:$id, text:$text, kind:$kind, source:$source, "
        "group:$grp, created_at:$now, refers_to:$when, valid:true, "
        "confidence:$conf, tone:$tone, importance:$imp, valid_from:$vf, "
        "srel:$srel}) SET m.embedding = vecf32($emb)",
        {"id": mid, "text": text, "kind": kind, "source": source,
         "grp": group, "now": now, "when": when or "", "emb": emb,
         "conf": float(confidence), "tone": tone or "",
         "imp": int(importance) if importance is not None else 5,
         "vf": (when or now), "srel": srel or ""})
    return mid


def _retire_nodes(ids: list[str], superseded_by: str | None = None) -> None:
    """Bi-temporal retire: set valid=false + invalid_at + valid_to=now (and
    superseded_by when a replacement exists). Idempotent; skips already-stale."""
    if not ids:
        return
    now = _now_iso()
    _graph.query(
        "UNWIND $ids AS rid MATCH (m:Memory {id:rid}) "
        "SET m.valid=false, m.invalid_at=$now, m.valid_to=$now",
        {"ids": ids, "now": now})
    if superseded_by:
        _graph.query(
            "UNWIND $ids AS rid MATCH (m:Memory {id:rid}) SET m.superseded_by=$sb",
            {"ids": ids, "sb": superseded_by})


# ---------------------------------------------------------------------------
# WRITE PIPELINE (Mem0-style, server-side dedup/supersede). A fact comes in; we
# embed it, pull the top-8 similar valid memories, and let ONE cheap DeepSeek
# call decide add / update / noop, clean a raw utterance into an atomic fact,
# and extract the event date + entities + relations. With no key we degrade
# gracefully (cosine≥0.95 ⇒ noop). An LLM failure never loses a memory.
# ---------------------------------------------------------------------------
_WRITE_PROMPT = """You are the write layer of Ahmed's personal memory graph. \
Decide how to store one new incoming memory relative to what is already stored.

Today's date is {today}.

NEW INPUT (raw_utterance={raw}): {text}

EXISTING SIMILAR MEMORIES (most similar first), "id: text":
{similar}

Choose ONE operation:
- "add": genuinely new information -> store it.
- "update": REPLACES/UPDATES a specific existing memory above (a changed fact or \
a new current value) -> put that memory's id in target_id.
- "noop": an existing memory above ALREADY states this (a duplicate) -> put its \
id in target_id; nothing new is stored.

Return ONLY a JSON object with EXACTLY these keys:
{{"op":"add|update|noop",
"target_id":"<existing id for update/noop, else null>",
"text":"<the fact as ONE clean third-person atomic statement about Ahmed's \
world; MANDATORY rewrite when raw_utterance=true: keep meaning, drop filler and \
first-person; for noop echo the existing text>",
"when":"<ISO date YYYY-MM-DD the fact REFERS TO if it implies an event date \
(resolve relative dates like 'yesterday' against today above), else null>",
"sro":{{"subject":"","relation":"","object":"","single_valued":true_or_false}} \
or null (single_valued=true only when the subject can hold ONE current value for \
this relation, e.g. lives-in, is-CEO-of, phone-number, so a new value supersedes),
"entities":[{{"name":"","type":"person|company|project|place|product|amount|thing|concept"}}],
"relations":[{{"from":"","to":"","rtype":""}}]}}

JSON:"""


def _similar(emb: list[float], group: str, n: int = 8) -> list[dict]:
    """Top-n similar VALID same-group memories by embedding (empty on any error,
    e.g. before the vector index has any rows)."""
    try:
        rows = _graph.query(
            "CALL db.idx.vector.queryNodes('Memory','embedding',$k,vecf32($emb)) "
            "YIELD node, score WHERE node.group=$grp AND node.valid=true "
            "RETURN node.id, node.text, score",
            {"k": max(n, 1) * 3 + 10, "emb": emb,
             "grp": group}).result_set[:max(n, 1)]
    except Exception:  # noqa: BLE001
        return []
    return [{"id": r[0], "text": r[1], "score": _sim(r[2])} for r in rows]


def _norm_plan(p: dict, orig_text: str, orig_when: str | None,
               similar_ids: set) -> dict:
    """Validate/normalize the model's plan — never trust its shape or a
    hallucinated target_id; degrade update/noop-without-a-real-target to add."""
    op = str(p.get("op", "add")).strip().lower()
    if op not in ("add", "update", "noop"):
        op = "add"
    tid = p.get("target_id")
    tid = str(tid).strip() if tid not in (None, "") else ""
    if tid and tid not in similar_ids:
        tid = ""                      # model invented an id — ignore it
    if op in ("update", "noop") and not tid:
        op = "add"                    # no real target ⇒ it's just an add
    text = str(p.get("text") or "").strip() or orig_text
    when = p.get("when")
    when = str(when).strip() if when not in (None, "") else (orig_when or None)
    sro = p.get("sro") if isinstance(p.get("sro"), dict) else None
    ents = [e for e in (p.get("entities") or [])
            if isinstance(e, dict) and str(e.get("name", "")).strip()]
    rels = [r for r in (p.get("relations") or [])
            if isinstance(r, dict) and str(r.get("from", "")).strip()
            and str(r.get("to", "")).strip()]
    return {"op": op, "target_id": tid or None, "text": text, "when": when,
            "sro": sro, "entities": ents, "relations": rels}


def _plan_memory(text: str, kind: str, when: str | None, group: str,
                 raw: bool) -> tuple[dict, list[float]]:
    """Build the write PLAN. DeepSeek when keyed (fall back to plain add on any
    error — never lose a memory); cosine≥0.95 noop-dedup when not keyed."""
    emb = _embed(text)
    similar = _similar(emb, group, 8)
    plain_add = {"op": "add", "target_id": None, "text": text, "when": when,
                 "sro": None, "entities": [], "relations": []}
    if DEEPSEEK_KEY:
        try:
            sim_txt = "\n".join(f"{s['id']}: {s['text']}" for s in similar) \
                or "(none)"
            got = _deepseek_json(_WRITE_PROMPT.format(
                today=_today(), raw=str(bool(raw)).lower(), text=text,
                similar=sim_txt))
            if not isinstance(got, dict):
                raise ValueError("plan not an object")
            plan = _norm_plan(got, text, when, {s["id"] for s in similar})
        except Exception:  # noqa: BLE001 — any failure ⇒ plain add
            plan = plain_add
    elif similar and similar[0]["score"] >= 0.95:
        plan = {"op": "noop", "target_id": similar[0]["id"], "text": text,
                "when": when, "sro": None, "entities": [], "relations": []}
    else:
        plan = plain_add
    return plan, emb


def _apply_relations(relations, group: str) -> None:
    for r in relations or []:
        fn = str((r or {}).get("from", "")).strip()
        tn = str((r or {}).get("to", "")).strip()
        rt = str((r or {}).get("rtype", "related")).strip() or "related"
        if fn and tn:
            try:
                _add_relation(fn, tn, rt, group)
            except Exception:  # noqa: BLE001
                pass


def _apply_plan(plan: dict, orig_text: str, orig_emb: list[float], kind: str,
                source: str, group: str, confidence: float, tone: str | None,
                importance: int, body_entities, body_links,
                no_retire: bool = False) -> dict:
    """Execute a write plan: noop / add / update(+supersede) with bi-temporal
    retirement, single-valued (srel) supersession, entities and relations.
    `no_retire` (historical imports): keep the srel marker on the new node but
    NEVER retire existing memories — old data cannot supersede current truth."""
    all_entities = list(plan.get("entities") or []) + list(body_entities or [])
    op = plan["op"]
    if op == "noop" and plan.get("target_id"):
        # Don't store a duplicate, but still attach any explicit/extracted
        # entities+relations to the memory that already covers it.
        ents = _attach_entities(plan["target_id"], all_entities, group)
        _apply_relations(plan.get("relations"), group)
        return {"id": plan["target_id"], "op": "noop", "auto_linked": 0,
                "entities": ents}

    text = plan["text"]
    when = plan.get("when")
    emb = orig_emb if text == orig_text else _embed(text)
    sro = plan.get("sro")

    # single-valued (srel) supersession — deterministic, no LLM: a new value for
    # the same subject|relation retires every prior valid one.
    srel = None
    retired: list[str] = []
    if isinstance(sro, dict) and sro.get("single_valued"):
        subj = str(sro.get("subject", "")).strip().lower()
        rel = str(sro.get("relation", "")).strip().lower()
        if subj and rel:
            srel = subj + "|" + rel
            rows = _graph.query(
                "MATCH (m:Memory) WHERE m.group=$grp AND m.valid=true "
                "AND m.srel=$srel RETURN m.id",
                {"grp": group, "srel": srel}).result_set
            retired += [r[0] for r in rows]
    if op == "update" and plan.get("target_id") \
            and plan["target_id"] not in retired:
        retired.append(plan["target_id"])
    if no_retire:
        retired = []             # historical import: srel kept, nothing retired

    _retire_nodes(retired)   # bi-temporal retire BEFORE creating the new node
    mid = _create(text, kind, when, source, group, emb=emb,
                  confidence=confidence, tone=tone, importance=importance,
                  srel=srel)
    if retired:
        _graph.query(
            "UNWIND $ids AS rid MATCH (m:Memory {id:rid}) SET m.superseded_by=$sb",
            {"ids": retired, "sb": mid})
        _graph.query(
            "UNWIND $ids AS rid MATCH (a:Memory {id:$mid}),(b:Memory {id:rid}) "
            "CREATE (a)-[:LINK {type:'replaces'}]->(b)",
            {"mid": mid, "ids": retired})

    for lk in (body_links or []):
        to = str((lk or {}).get("to", "")).strip()
        if to:
            _graph.query(
                "MATCH (a:Memory {id:$a}),(b:Memory {id:$b}) "
                "CREATE (a)-[:LINK {type:$t}]->(b)",
                {"a": mid, "b": to, "t": str((lk or {}).get("type", "related"))})
    auto = _auto_link(mid, text, emb, group) if AUTOLINK else []
    ents = _attach_entities(mid, all_entities, group)
    _apply_relations(plan.get("relations"), group)
    op_out = "update" if retired else "add"
    return {"id": mid, "op": op_out, "auto_linked": len(auto), "entities": ents}


@app.post("/remember")
def remember(body: Remember, authorization: str | None = Header(default=None)):
    """Save one memory THROUGH the write pipeline (dedup/supersede/cleanup).
    Backward compatible: an old client POSTing just {text} still works — with no
    DEEPSEEK_API_KEY it's a plain add (plus cosine≥0.95 duplicate suppression)."""
    _auth(authorization)
    if not body.text.strip():
        raise HTTPException(status_code=400, detail="empty text")
    grp = body.group or GROUP_DEFAULT
    text = body.text.strip()
    conf = body.confidence if body.confidence is not None else 0.75
    imp = 5 if body.importance is None else max(1, min(10, int(body.importance)))
    plan, emb = _plan_memory(text, body.kind or "fact", body.when, grp,
                             bool(body.raw))
    if body.historical and plan.get("op") != "noop":
        # Historical backfill can only ADD (with its old valid_from) or noop —
        # an outdated fact must never supersede the current truth. A FUTURE
        # current-fact write may still retire the historical one (srel kept),
        # which is the correct direction of time.
        plan["op"] = "add"
        plan["target_id"] = None
    return _apply_plan(plan, text, emb, kind=body.kind or "fact",
                       source=body.source or "jarvis", group=grp,
                       confidence=conf, tone=body.tone, importance=imp,
                       body_entities=body.entities, body_links=body.links,
                       no_retire=bool(body.historical))


@app.post("/link")
def link(body: Link, authorization: str | None = Header(default=None)):
    _auth(authorization)
    _graph.query(
        "MATCH (a:Memory {id:$a}),(b:Memory {id:$b}) "
        "CREATE (a)-[:LINK {type:$t}]->(b)",
        {"a": body.from_id, "b": body.to, "t": body.type or "related"})
    return {"ok": True}


@app.post("/supersede")
def supersede(body: Supersede, authorization: str | None = Header(default=None)):
    """A fact changed: mark the old one stale (kept for history — this is the
    'temporal' bit) and add the new current one, linked to what it replaced."""
    _auth(authorization)
    grp = body.group or GROUP_DEFAULT
    text = body.text.strip()
    emb = _embed(text)
    mid = _create(text, body.kind or "fact", body.when,
                  body.source or "jarvis", grp, emb=emb,
                  confidence=body.confidence if body.confidence is not None else 0.75,
                  tone=body.tone)
    _retire_nodes([body.old_id], superseded_by=mid)  # bi-temporal retire
    _graph.query("MATCH (a:Memory {id:$a}),(b:Memory {id:$b}) "
                 "CREATE (a)-[:LINK {type:'replaces'}]->(b)",
                 {"a": mid, "b": body.old_id})
    if AUTOLINK:
        _auto_link(mid, text, emb, grp)
    ents = _attach_entities(mid, body.entities, grp)
    return {"id": mid, "entities": ents}


_STOP = {"what", "when", "who", "where", "which", "about", "with", "the",
         "for", "and", "that", "this", "have", "has", "did", "does", "was",
         "were", "are", "you", "your", "his", "her", "she", "him", "they",
         "them", "from", "into", "tell", "know", "remember", "ahmed", "sir",
         "jarvis", "me", "my", "i", "a", "an", "of", "to", "is", "it", "do"}


def _significant_words(text: str) -> set:
    """Content words (names/terms) of a fact — used to spot shared entities."""
    import re as _re
    return {w for w in _re.findall(r"[a-z0-9]{3,}", (text or "").lower())
            if w not in _STOP}


def _edge_exists(a: str, b: str) -> bool:
    n = _graph.query(
        "MATCH (a:Memory {id:$a})-[l:LINK]-(b:Memory {id:$b}) RETURN count(l)",
        {"a": a, "b": b}).result_set[0][0]
    return int(n) > 0


def _auto_link(new_id: str, text: str, emb: list[float], group: str,
               threshold: float = None, max_links: int = None,
               dry_run: bool = False) -> list[dict]:
    """LAYER 1 — connect a fact to the memories most like it, automatically.

    Candidates come from BOTH vector similarity (meaning) and shared significant
    words (so two notes that both name "Altimi" always link, even when the
    embeddings are only lukewarm). Best `max_links` above `threshold` get a
    :LINK edge of type 'similar' (skipping any pair already connected). No LLM."""
    threshold = AUTOLINK_THRESHOLD if threshold is None else threshold
    max_links = AUTOLINK_MAX if max_links is None else max_links
    my_words = _significant_words(text)
    cand: dict = {}   # candidate id -> (best combined score, its text)

    # meaning pass — nearest neighbours by embedding (shared words nudge it up)
    vrows = _graph.query(
        "CALL db.idx.vector.queryNodes('Memory','embedding',$k,vecf32($emb)) "
        "YIELD node, score WHERE node.group=$grp AND node.valid=true "
        "AND node.id <> $nid RETURN node.id, node.text, score",
        {"k": max_links * 3 + 24, "emb": emb, "grp": group,
         "nid": new_id}).result_set
    for cid, ctext, score in vrows:
        shared = len(my_words & _significant_words(ctext or ""))
        # Cap the shared-word nudge: long boilerplate texts (project checkpoints)
        # share many generic words and an uncapped bonus was linking unrelated
        # facts. Meaning must carry the link; words only nudge.
        combined = _sim(score) + 0.08 * min(shared, 3)
        if cid not in cand or combined > cand[cid][0]:
            cand[cid] = (combined, ctext)

    # entity pass — facts sharing >=2 significant words (catches names/numbers
    # the embedding might miss)
    words = list(my_words)[:8]
    if words:
        kwq = ("MATCH (m:Memory) WHERE m.group=$grp AND m.valid=true "
               "AND m.id <> $nid AND (" +
               " OR ".join(f"toLower(m.text) CONTAINS $w{i}"
                           for i in range(len(words))) +
               ") RETURN m.id, m.text")
        params = {"grp": group, "nid": new_id,
                  **{f"w{i}": w for i, w in enumerate(words)}}
        for cid, ctext in _graph.query(kwq, params).result_set:
            shared = len(my_words & _significant_words(ctext or ""))
            # Base 0.35 keeps a pure word-overlap link below the 0.60 threshold
            # until ~4 shared significant words — generic checkpoint boilerplate
            # ("project", "jarvis", dates) shares 2-3 and must NOT link; a real
            # name/number overlap shares more.
            if shared >= 2:
                combined = 0.35 + 0.08 * min(shared, 4)
                if cid not in cand or combined > cand[cid][0]:
                    cand[cid] = (combined, ctext)

    ranked = sorted(cand.items(), key=lambda kv: kv[1][0], reverse=True)
    linked = []
    for cid, (sc, ctext) in ranked:
        if sc < threshold:
            break            # ranked high→low, nothing below will qualify
        if len(linked) >= max_links:
            break
        if _edge_exists(new_id, cid):
            continue
        if not dry_run:
            _graph.query(
                "MATCH (a:Memory {id:$a}),(b:Memory {id:$b}) "
                "CREATE (a)-[:LINK {type:'similar'}]->(b)",
                {"a": new_id, "b": cid})
        linked.append({"id": cid, "text": ctext, "score": round(sc, 3)})
    return linked


# ---------------------------------------------------------------------------
# LAYER 2 — ENTITY RESOLUTION. The device's local model pulls the people /
# companies / amounts out of each fact (no API cost); this service turns them
# into shared :Entity nodes and (:Memory)-[:MENTIONS]->(:Entity) edges, so
# "pull up Al Temimi" returns everything hanging off that one node.
# ---------------------------------------------------------------------------

def _norm_entity(name: str) -> str:
    """Canonical key for an entity so the same name reuses one node."""
    return " ".join((name or "").lower().split())


def _sorted_tokens(s: str) -> str:
    """Order-independent token string — 'al temimi' and 'temimi al' compare equal
    under difflib, so a reversed name still resolves to the same entity."""
    return " ".join(sorted((s or "").lower().split()))


def _append_alias(entity_key: str, alias: str, group: str) -> None:
    """Add `alias` (a canonical key) to an entity's alias list if new. Read-modify
    -write in Python (FalkorDB has no APOC list helpers); None-tolerant."""
    if not alias or alias == entity_key:
        return
    try:
        rows = _graph.query(
            "MATCH (e:Entity {key:$k, group:$grp}) RETURN e.aliases",
            {"k": entity_key, "grp": group}).result_set
        cur = (rows[0][0] if rows else None) or []
        if alias in cur:
            return
        _graph.query(
            "MATCH (e:Entity {key:$k, group:$grp}) SET e.aliases=$al",
            {"k": entity_key, "grp": group, "al": list(cur) + [alias]})
    except Exception:  # noqa: BLE001 — aliasing is best-effort, never fatal
        pass


def _resolve_entity(name: str, etype: str, group: str) -> str | None:
    """ENTITY RESOLUTION — map an extracted name to ONE canonical :Entity, so
    'Al Temimi' / 'Al-Tamimi' / a reversed spelling all land on the same node.

    Flow: (1) exact canonical-key OR alias match (any type) → reuse it. (2) else
    fuzzy over SAME-etype same-group entities, scored max(name-embedding cosine,
    difflib token ratio over name+aliases): ≥0.92 → attach + record the incoming
    spelling as an alias; 0.80–0.92 AND a DeepSeek key → ONE yes/no adjudication
    → attach+alias or create; (3) else create a fresh entity (with a name
    embedding, so the next lookalike can find it). Returns the canonical key."""
    key = _norm_entity(name)
    if not key:
        return None
    etype = (etype or "thing").strip().lower() or "thing"
    # (1) exact key or alias — type-agnostic (a name is a name)
    rows = _graph.query(
        "MATCH (e:Entity {group:$grp}) "
        "WHERE e.key=$key OR $key IN coalesce(e.aliases,[]) "
        "RETURN e.key LIMIT 1", {"grp": group, "key": key}).result_set
    if rows:
        return rows[0][0]
    # (2) fuzzy over same-etype same-group entities
    emb = _embed(name)
    emb_sim: dict = {}
    try:  # nearest by name-embedding (index may be empty on the first entities)
        for r in _graph.query(
            "CALL db.idx.vector.queryNodes('Entity','embedding',$k,vecf32($emb)) "
            "YIELD node, score WHERE node.group=$grp AND node.etype=$et "
            "RETURN node.key, score",
            {"k": 10, "emb": emb, "grp": group, "et": etype}).result_set:
            emb_sim[r[0]] = _sim(r[1])
    except Exception:  # noqa: BLE001
        pass
    cands = _graph.query(
        "MATCH (e:Entity {group:$grp, etype:$et}) "
        "RETURN e.key, e.name, e.aliases", {"grp": group, "et": etype}).result_set
    qtok = _sorted_tokens(name)
    best_key, best_score, best_tok = None, 0.0, 0.0
    for ckey, cname, caliases in cands:
        tok = difflib.SequenceMatcher(None, qtok, _sorted_tokens(cname or ckey)
                                      ).ratio()
        for a in (caliases or []):
            tok = max(tok, difflib.SequenceMatcher(
                None, qtok, _sorted_tokens(a)).ratio())
        sc = max(emb_sim.get(ckey, 0.0), tok)
        if sc > best_score:
            best_key, best_score, best_tok = ckey, sc, tok
    # Same hard rule as dedup: embedding alone never auto-attaches an alias —
    # the strings must also broadly agree, else it goes to adjudication.
    if best_key and best_score >= 0.92 and best_tok >= 0.6:
        _append_alias(best_key, key, group)
        return best_key
    if best_key and best_score >= 0.80 and DEEPSEEK_KEY:
        try:
            prompt = ('Are these the same ' + etype + '? '
                      'A: "' + name + '"  B: "' + best_key + '". '
                      'Reply ONLY JSON {"same": true} or {"same": false}.')
            got = _deepseek_json(prompt, max_tokens=20)
            if isinstance(got, dict) and bool(got.get("same")):
                _append_alias(best_key, key, group)
                return best_key
        except Exception:  # noqa: BLE001 — adjudication failure ⇒ create new
            pass
    # (3) create a new entity (with a name embedding for future resolution)
    _graph.query(
        "MERGE (e:Entity {key:$key, group:$grp}) "
        "ON CREATE SET e.name=$name, e.etype=$et, e.created_at=$now, "
        "e.aliases=[] SET e.embedding = vecf32($emb)",
        {"key": key, "grp": group, "name": name, "et": etype,
         "now": _now_iso(), "emb": emb})
    return key


def _attach_entities(mem_id: str, entities: list | None, group: str) -> int:
    """Resolve each extracted entity to a canonical node and wire
    (:Memory)-[:MENTIONS]->(:Entity). Returns how many were attached."""
    n = 0
    for e in entities or []:
        name = str((e or {}).get("name", "")).strip()
        if len(name) < 2:
            continue
        etype = (str((e or {}).get("type", "thing")).strip().lower()
                 or "thing")
        key = _resolve_entity(name, etype, group)
        if not key:
            continue
        _graph.query(
            "MATCH (m:Memory {id:$mid}),(e:Entity {key:$key, group:$grp}) "
            "MERGE (m)-[:MENTIONS]->(e)",
            {"mid": mem_id, "key": key, "grp": group})
        n += 1
    return n


# The LLM extracts relation labels as free text — "is at", "located at",
# "located_at" all became DIFFERENT edge types and the dossier showed a mess of
# near-duplicate groups. Canonicalize every rtype at write AND read.
_RTYPE_CANON = {
    "located_at": "located_in", "located_in": "located_in", "based_in":
    "located_in", "based_at": "located_in", "is_at": "located_in", "is_in":
    "located_in", "at": "located_in", "in": "located_in", "location":
    "located_in", "lives_in": "located_in",
    "works_at": "works_at", "works_for": "works_at", "employed_at":
    "works_at", "employed_by": "works_at", "employee_of": "works_at",
    "works_in": "works_at",
    "owns": "owns", "owner_of": "owns", "founded": "owns", "founder_of":
    "owns", "runs": "owns", "ceo_of": "owns",
    "client_of": "client_of", "customer_of": "client_of",
    "partner_of": "partner_of", "partners_with": "partner_of",
    "works_with": "works_with", "collaborates_with": "works_with",
    "colleague_of": "works_with",
    "part_of": "part_of", "member_of": "part_of", "belongs_to": "part_of",
    "related": "related", "related_to": "related",
}


def _norm_rtype(rt) -> str:
    key = re.sub(r"[\s\-]+", "_", str(rt or "").strip().lower()).strip("_")
    return _RTYPE_CANON.get(key, key or "related")


def _add_relation(from_name: str, to_name: str, rtype: str, group: str) -> None:
    """(:Entity)-[:REL {rtype, created_at}]->(:Entity). BOTH ends go through the
    resolver first, so relations attach to the same canonical nodes as mentions;
    rtype is canonicalized so phrasing variants share one edge type.
    Self-relations (both names resolving to one entity) are skipped."""
    fk = _resolve_entity(from_name, "thing", group)
    tk = _resolve_entity(to_name, "thing", group)
    if not fk or not tk or fk == tk:
        return
    _graph.query(
        "MATCH (a:Entity {key:$fk, group:$grp}),(b:Entity {key:$tk, group:$grp}) "
        "MERGE (a)-[r:REL {rtype:$rt}]->(b) ON CREATE SET r.created_at=$now",
        {"fk": fk, "tk": tk, "grp": group, "rt": _norm_rtype(rtype),
         "now": _now_iso()})


def _salience(f: dict) -> float:
    """Rank facts by relevance + keyword hits + recency + confidence +
    reinforcement + importance — the Generative-Agents recency×importance×
    relevance idea, plus a use-based boost. Recency anchors on the EVENT date
    (refers_to) when present, else creation time. Nothing is ever dropped."""
    from datetime import datetime as _dt, timezone as _tz
    base = f.get("score", 0.0) + 0.15 * f.get("keyword_hits", 0)
    rec = 0.0
    anchor = (f.get("refers_to") or "").strip() or (f.get("created_at") or "")
    if anchor:
        try:
            age_days = (_dt.now(_tz.utc) - _dt.fromisoformat(anchor)
                        ).total_seconds() / 86400.0
            rec = 0.25 * (2.0 ** (-max(age_days, 0.0) / 120.0))  # half-life 120d
        except Exception:  # noqa: BLE001
            pass
    conf = 0.2 * (f.get("confidence", 0.75) - 0.5)
    ac = f.get("access_count") or 0
    reinforce = 0.1 * math.log(1 + max(int(ac), 0))
    imp = f.get("importance")
    imp = 5 if imp is None else imp
    imp_term = 0.05 * (int(imp) - 5)
    return base + rec + conf + reinforce + imp_term


def _search_filter(alias: str, include_stale: bool, as_of: str | None,
                   when_from: str | None, when_to: str | None) -> str:
    """WHERE-fragment (leading ' AND …') scoping a node alias by validity + the
    temporal window. `as_of` selects bi-temporally (what was true then, tolerant
    of old nodes missing valid_from/valid_to via coalesce); otherwise the legacy
    `valid` bool. when_from/when_to filter the EVENT date (refers_to else
    created_at). All values are parameters — never interpolated."""
    parts = []
    if as_of:
        parts.append(f"coalesce({alias}.valid_from,{alias}.created_at)<=$as_of")
        parts.append(
            f"(coalesce({alias}.valid_to,{alias}.invalid_at) IS NULL "
            f"OR coalesce({alias}.valid_to,{alias}.invalid_at)>$as_of)")
    elif not include_stale:
        parts.append(f"{alias}.valid=true")
    if when_from or when_to:
        anchor = (f"(CASE WHEN {alias}.refers_to<>'' THEN {alias}.refers_to "
                  f"ELSE {alias}.created_at END)")
        if when_from:
            parts.append(f"{anchor}>=$when_from")
        if when_to:
            parts.append(f"{anchor}<=$when_to")
    return (" AND " + " AND ".join(parts) + " ") if parts else " "


@app.get("/search")
def search(q: str, k: int = 10, group: str | None = None, hops: int = 1,
          include_stale: bool = False, as_of: str | None = None,
          when_from: str | None = None, when_to: str | None = None,
          authorization: str | None = Header(default=None)):
    """HYBRID recall: vector similarity (meaning) UNIONed with keyword matches
    (names/terms). A long detailed memory has a blurry embedding and can lose a
    short vague one on pure vector search, so we also surface any fact whose
    text literally contains the query's significant words — that's how "Ziyad"
    or "Journey Joy" reliably comes back.

    TEMPORAL: `as_of=<ISO>` answers "what was true THEN" (bi-temporal — valid_from
    <= as_of AND (valid_to null OR > as_of)); `when_from`/`when_to` window the EVENT
    date (refers_to else created_at). Old clients omit all three and get today's
    valid facts, exactly as before. Primary hits are REINFORCED (access_count++,
    last_accessed=now) and their 1-hop linked context is folded in, scored at
    0.6× the parent's salience, capped at k+5 total."""
    import re as _re
    _auth(authorization)
    grp = group or GROUP_DEFAULT
    emb = _embed(q)
    fnode = _search_filter("node", include_stale, as_of, when_from, when_to)
    fm = _search_filter("m", include_stale, as_of, when_from, when_to)
    tparams: dict = {}
    if as_of:
        tparams["as_of"] = as_of
    if when_from:
        tparams["when_from"] = when_from
    if when_to:
        tparams["when_to"] = when_to
    # Over-fetch: KNN yields exactly $k candidates BEFORE the WHERE filter, so
    # retired/superseded neighbours can starve a small k down to zero results.
    rows = _graph.query(
        "CALL db.idx.vector.queryNodes('Memory','embedding',$k,vecf32($emb)) "
        "YIELD node, score WHERE node.group=$grp" + fnode +
        "RETURN node.id, node.text, node.kind, node.refers_to, "
        "node.created_at, score, node.confidence, node.importance, "
        "node.access_count",
        {"k": max(k, 1) * 3 + 10, "emb": emb, "grp": grp,
         **tparams}).result_set[:max(k, 1)]
    facts = []
    seen = set()
    for r in rows:
        seen.add(r[0])
        facts.append({"id": r[0], "fact": r[1], "kind": r[2],
                      "refers_to": r[3], "created_at": r[4],
                      "score": round(_sim(r[5]), 3),
                      "confidence": float(r[6]) if r[6] is not None else 0.75,
                      "importance": r[7] if r[7] is not None else 5,
                      "access_count": r[8] or 0, "linked": False})
    # keyword pass: pull any fact literally mentioning a significant query word,
    # ranked by how many distinct query words it contains (names win)
    words = [w for w in _re.findall(r"[a-z0-9]{3,}", q.lower())
             if w not in _STOP][:8]
    if words:
        kw = "MATCH (m:Memory) WHERE m.group=$grp" + fm \
             + "AND (" + " OR ".join(
                 f"toLower(m.text) CONTAINS $w{i}" for i in range(len(words))) \
             + ") RETURN m.id, m.text, m.kind, m.refers_to, m.created_at, " \
             + "m.confidence, m.importance, m.access_count"
        params = {"grp": grp, **tparams,
                  **{f"w{i}": w for i, w in enumerate(words)}}
        for r in _graph.query(kw, params).result_set:
            if r[0] in seen:
                continue
            hits = sum(1 for w in words if w in (r[1] or "").lower())
            facts.append({"id": r[0], "fact": r[1], "kind": r[2],
                          "refers_to": r[3], "created_at": r[4],
                          "score": round(0.5 + 0.1 * hits, 3),
                          "confidence": float(r[5]) if r[5] is not None else 0.75,
                          "importance": r[6] if r[6] is not None else 5,
                          "access_count": r[7] or 0,
                          "keyword_hits": hits, "linked": False})
            seen.add(r[0])
    # SALIENCE rank — relevance + keyword hits + recency (newer floats up) +
    # confidence (firm facts beat hedged ones) + reinforcement + importance.
    # Latest/important/often-recalled surface first; nothing is ever deleted.
    for f in facts:
        f["_sal"] = _salience(f)
    facts.sort(key=lambda f: f["_sal"], reverse=True)
    # REINFORCEMENT — the facts we actually surfaced get "used": one batched
    # write bumps access_count + last_accessed so recalled facts rise next time.
    # coalesce() tolerates old nodes that predate these fields.
    prim_ids = [f["id"] for f in facts]
    if prim_ids:
        try:
            _graph.query(
                "UNWIND $ids AS rid MATCH (m:Memory {id:rid}) "
                "SET m.access_count = coalesce(m.access_count,0)+1, "
                "m.last_accessed=$now", {"ids": prim_ids, "now": _now_iso()})
        except Exception:  # noqa: BLE001 — reinforcement is best-effort
            pass
    # LINKED CONTEXT — 1-hop [:LINK] neighbours, scored at 0.6× the parent's
    # salience (max over parents), merged into the ranked list, then cap k+5.
    if hops > 0 and prim_ids:
        sal_by_id = {f["id"]: f["_sal"] for f in facts}
        fn = _search_filter("n", include_stale, as_of, when_from, when_to)
        neigh = _graph.query(
            "MATCH (m:Memory)-[l:LINK]-(n:Memory) WHERE m.id IN $ids" + fn +
            "RETURN m.id, n.id, n.text, n.kind, n.refers_to, n.created_at, "
            "n.confidence, l.type", {"ids": prim_ids, **tparams}).result_set
        best: dict = {}   # neighbour id -> (score, row)
        for r in neigh:
            parent, nid = r[0], r[1]
            if nid in seen:
                continue
            sc = 0.6 * sal_by_id.get(parent, 0.0)
            if nid not in best or sc > best[nid][0]:
                best[nid] = (sc, r)
        for nid, (sc, r) in best.items():
            seen.add(nid)
            facts.append({"id": nid, "fact": r[2], "kind": r[3],
                          "refers_to": r[4], "created_at": r[5],
                          "confidence": float(r[6]) if r[6] is not None else 0.75,
                          "link": r[7], "linked": True,
                          "score": round(sc, 3), "_sal": sc})
        facts.sort(key=lambda f: f["_sal"], reverse=True)
    for f in facts:
        f.pop("_sal", None)
    return {"facts": facts[:max(k, 1) + 5]}


@app.get("/memories")
def memories(group: str | None = None, limit: int = 40, offset: int = 0,
             include_stale: bool = False, kind: str | None = None,
             authorization: str | None = Header(default=None)):
    """List memories newest-first — powers the HUD memory VIEWER (browse &
    curate, not search). `kind` filters (e.g. feedback/preference) so the
    reflection engine can pull just Ahmed's stated preferences to respect."""
    _auth(authorization)
    grp = group or GROUP_DEFAULT
    valid = "" if include_stale else "AND m.valid=true "
    kf = "AND m.kind=$kind " if kind else ""
    params = {"grp": grp, "off": max(offset, 0),
              "lim": max(1, min(limit, 200))}
    if kind:
        params["kind"] = kind
    rows = _graph.query(
        "MATCH (m:Memory) WHERE m.group=$grp " + valid + kf +
        "RETURN m.id, m.text, m.kind, m.refers_to, m.created_at, m.source, m.tone "
        "ORDER BY m.created_at DESC SKIP $off LIMIT $lim",
        params).result_set
    total = _graph.query(
        "MATCH (m:Memory) WHERE m.group=$grp " + valid + "RETURN count(m)",
        {"grp": grp}).result_set[0][0]
    return {"total": total, "memories": [
        {"id": r[0], "fact": r[1], "kind": r[2], "refers_to": r[3],
         "created_at": r[4], "source": r[5], "tone": r[6] or ""} for r in rows]}


class Forget(BaseModel):
    id: str
    group: str | None = None


@app.post("/forget")
def forget(body: Forget, authorization: str | None = Header(default=None)):
    """Permanently delete a memory (and its links) — Ahmed removed a WRONG fact
    from the HUD viewer. Unlike /supersede (which keeps the old node as history),
    this is gone for good; use it only for genuinely bad/incorrect memories."""
    _auth(authorization)
    if not body.id.strip():
        raise HTTPException(status_code=400, detail="no id")
    _graph.query("MATCH (m:Memory {id:$id}) DETACH DELETE m",
                 {"id": body.id.strip()})
    return {"ok": True}


@app.get("/graph")
def graph(group: str | None = None, include_stale: bool = False,
          authorization: str | None = Header(default=None)):
    """The WHOLE memory graph for a group — :Memory nodes + :LINK edges, PLUS
    Layer-2 :Entity nodes and (:Memory)-[:MENTIONS]->(:Entity) edges — for the
    HUD's graph view. Ahmed walks/zooms this to see what's connected to what."""
    _auth(authorization)
    grp = group or GROUP_DEFAULT
    valid = "" if include_stale else "AND m.valid=true "
    nrows = _graph.query(
        "MATCH (m:Memory) WHERE m.group=$grp " + valid +
        "RETURN m.id, m.text, m.kind, m.created_at", {"grp": grp}).result_set
    erows = _graph.query(
        "MATCH (a:Memory)-[l:LINK]->(b:Memory) "
        "WHERE a.group=$grp AND b.group=$grp "
        + ("" if include_stale else "AND a.valid=true AND b.valid=true ") +
        "RETURN a.id, b.id, l.type", {"grp": grp}).result_set
    # entity nodes + the facts that mention them
    entrows = _graph.query(
        "MATCH (e:Entity) WHERE e.group=$grp RETURN e.key, e.name, e.etype",
        {"grp": grp}).result_set
    mentrows = _graph.query(
        "MATCH (m:Memory)-[:MENTIONS]->(e:Entity) "
        "WHERE m.group=$grp AND e.group=$grp " +
        ("" if include_stale else "AND m.valid=true ") +
        "RETURN m.id, e.key", {"grp": grp}).result_set
    # entity↔entity relations (Layer-2 REL edges) for the graph view
    relrows = _graph.query(
        "MATCH (a:Entity)-[r:REL]->(b:Entity) "
        "WHERE a.group=$grp AND b.group=$grp "
        "RETURN a.key, b.key, r.rtype", {"grp": grp}).result_set
    nodes = [{"id": r[0], "text": r[1], "kind": r[2], "created_at": r[3]}
             for r in nrows]
    nodes += [{"id": "ent:" + r[0], "text": r[1],
               "kind": "entity:" + (r[2] or "thing"), "created_at": ""}
              for r in entrows]
    edges = [{"from": r[0], "to": r[1], "type": r[2] or "related"}
             for r in erows]
    edges += [{"from": r[0], "to": "ent:" + r[1], "type": "mentions"}
              for r in mentrows]
    edges += [{"from": "ent:" + r[0], "to": "ent:" + r[1],
               "type": "rel:" + (r[2] or "related")} for r in relrows]
    return {"nodes": nodes, "edges": edges}


class Unlink(BaseModel):
    from_id: str
    to: str
    group: str | None = None


@app.post("/unlink")
def unlink(body: Unlink, authorization: str | None = Header(default=None)):
    """Cut the connection(s) between two memories (either direction) — Ahmed
    disconnected two nodes in the graph view. The nodes stay; only the edge goes."""
    _auth(authorization)
    _graph.query(
        "MATCH (a:Memory {id:$a})-[l:LINK]-(b:Memory {id:$b}) DELETE l",
        {"a": body.from_id, "b": body.to})
    return {"ok": True}


@app.get("/entity")
def entity(name: str, group: str | None = None,
           authorization: str | None = Header(default=None)):
    """Everything hanging off one entity — the 'pull up Al Temimi' view. Matches
    by canonical key with light fuzziness (substring either way)."""
    _auth(authorization)
    grp = group or GROUP_DEFAULT
    key = _norm_entity(name)
    if not key:
        return {"entities": []}
    rows = _graph.query(
        "MATCH (e:Entity)<-[:MENTIONS]-(m:Memory) "
        "WHERE e.group=$grp AND m.valid=true AND "
        "(e.key=$key OR e.key CONTAINS $key OR $key CONTAINS e.key "
        "OR $key IN coalesce(e.aliases,[])) "
        "RETURN e.key, e.name, e.etype, e.summary, e.read, m.id, m.text, "
        "m.created_at ORDER BY e.name, m.created_at",
        {"grp": grp, "key": key}).result_set
    ents: dict = {}
    for ekey, ename, etype, summary, eread, mid, mtext, mcreated in rows:
        e = ents.setdefault(ekey, {"key": ekey, "name": ename, "type": etype,
                                   "summary": summary or "", "read": eread or "",
                                   "facts": [], "relations": []})
        e["facts"].append({"id": mid, "fact": mtext, "created_at": mcreated})
    # attach entity↔entity relations (both directions) to each matched entity
    for ekey, e in ents.items():
        rels = []
        for direction, arrow in (("out", "-[r:REL]->"), ("in", "<-[r:REL]-")):
            for rr in _graph.query(
                "MATCH (a:Entity {key:$k, group:$grp})" + arrow + "(b:Entity) "
                "RETURN b.name, b.etype, r.rtype",
                    {"k": ekey, "grp": grp}).result_set:
                rels.append({"other": rr[0], "otype": rr[1] or "thing",
                             "rtype": _norm_rtype(rr[2]),
                             "direction": direction})
        e["relations"] = rels
    return {"entities": list(ents.values())}


# ---------------------------------------------------------------------------
# ENTITY PROFILE (dossier) — pull EVERYTHING known about one entity into a
# fast-reading card for the HUD. Resolution mirrors /entity (exact key → alias →
# substring); on several matches the most-mentioned wins and the rest land in
# `also_matched`. Empty sections are omitted so the shape degrades for any etype.
# ---------------------------------------------------------------------------

# Canonical entity types (used to validate a manual /entity/update). `concept`
# and product/amount are first-class alongside the classics.
ETYPES = {"person", "company", "place", "product", "amount", "thing", "concept"}

# Durable fact kinds surfaced as their own section on the card.
_DURABLE_KINDS = ("preference", "plan", "pattern", "synthesis", "convention")

# quick-contact harvesting (regex over the entity's facts' texts) --------------
# Saudi mobile: 05XXXXXXXX / 9665XXXXXXXX / +9665XXXXXXXX (all → 05XXXXXXXX).
_PHONE_RE = re.compile(r"(?<!\d)(?:\+?966|0)5\d{8}(?!\d)")
_EMAIL_RE = re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}")
_URL_RE = re.compile(r"(?:https?://|www\.)[^\s<>\"')]+", re.I)
_AMOUNT_RE = re.compile(
    r"(?:(?:SAR|SR|USD|AED|\$|€|£|﷼|ريال|ر\.?\s?س)\s?\d[\d,]*(?:\.\d+)?)"
    r"|(?:\d[\d,]*(?:\.\d+)?\s?(?:SAR|SR|USD|AED|riyals?|ريال|درهم|ر\.?\s?س))",
    re.I)
_SENT_SPLIT = re.compile(r"(?<=[.!?؟])\s")


def _first_sentence(text: str) -> str:
    """First sentence of a summary (for a relation's other_summary). "" if empty."""
    t = (text or "").strip()
    return _SENT_SPLIT.split(t, maxsplit=1)[0].strip() if t else ""


def _norm_phone(p: str) -> str:
    """Normalize a matched Saudi mobile to the local 05XXXXXXXX form for dedupe."""
    d = (p or "").lstrip("+")
    if d.startswith("966"):
        d = "0" + d[3:]
    return d


def _dedupe(seq) -> list:
    """Order-preserving dedupe."""
    return list(dict.fromkeys(x for x in seq if x))


def _harvest(texts: list, near_terms: list | None = None) -> dict:
    """Regex-harvest contacts/amounts from an entity's fact texts (None-tolerant).

    `near_terms` (the entity's name + aliases, lowercased): when given, a PHONE
    or EMAIL only counts if it appears within ~100 chars of one of the terms in
    that fact — a shared fact mentioning two people must not donate one person's
    email to the other's card (Elie's email showed up on Modern Intelligent
    Solution's dossier). urls/amounts stay loose — they're rarely personal."""
    phones, emails, urls, amounts = [], [], [], []
    terms = [t for t in (near_terms or []) if t and len(t) >= 2]

    def _near(s_low: str, start: int, end: int) -> bool:
        if not terms:
            return True
        lo, hi = max(0, start - 100), end + 100
        win = s_low[lo:hi]
        return any(t in win for t in terms)

    for t in texts:
        s = t or ""
        if not s:
            continue
        s_low = s.lower()
        for m in _PHONE_RE.finditer(s):
            if _near(s_low, m.start(), m.end()):
                phones.append(_norm_phone(m.group(0)))
        for m in _EMAIL_RE.finditer(s):
            if _near(s_low, m.start(), m.end()):
                emails.append(m.group(0).lower())
        urls += [m.rstrip(".,;:)]}'\"") for m in _URL_RE.findall(s)]
        amounts += [" ".join(m.split()) for m in _AMOUNT_RE.findall(s)]
    return {"phones": _dedupe(phones), "emails": _dedupe(emails),
            "urls": _dedupe(urls), "amounts": _dedupe(amounts)}


# --- COMPOSED DOSSIER (POST /dossier) — an LLM reads the raw aggregate below
# and returns an organized, adaptive card. The Swift renderer knows EXACTLY four
# widget "type"s and five chip "kind"s; we validate hard and DROP anything else.
_WIDGET_TYPES = {"text", "kv", "chips", "list"}
_CHIP_KINDS = {"phone", "email", "link", "map", "whatsapp"}
_LIST_ICONS = {"person", "company", "event", "task", "money", "note"}
_DOSSIER_MAX_AGE_S = 24 * 3600     # a cached dossier older than this is stale


def _sv(x) -> str:
    """Coerce any model-supplied scalar to a trimmed string ('' for None)."""
    return str(x).strip() if x is not None else ""


def _dossier_fresh(cache: dict, nfacts_now: int) -> bool:
    """A stored dossier is usable when it exists, was built for the SAME fact
    count (so a new fact invalidates it), and is < 24h old. None-tolerant."""
    cache = cache or {}
    if not cache.get("json"):
        return False
    n = cache.get("nfacts")
    if n is None or int(n) != int(nfacts_now):
        return False
    at = _sv(cache.get("at"))
    if not at:
        return False
    try:
        age = (datetime.now(timezone.utc)
               - datetime.fromisoformat(at)).total_seconds()
    except Exception:  # noqa: BLE001 — unparseable timestamp ⇒ recompose
        return False
    return 0 <= age < _DOSSIER_MAX_AGE_S


_DOSSIER_PROMPT = """You are the dossier composer for Ahmed's personal memory. \
Ahmed Al-Rajeh runs Reval Studio, a Dammam digital agency; he also works at \
Alrugaib Furniture. Today is {today}.

You are given EVERYTHING the memory graph knows about ONE entity. READ all of it \
and COMPOSE a clean, organized profile card for Ahmed. NEVER paste raw memory \
text and NEVER dump checkpoint blobs — REWRITE every line into short, scannable \
prose or key/value rows. Arabic fragments may stay in Arabic. Invent nothing.

ENTITY: "{name}"  (stored memory type: {etype})

FACTS (newest first) — "date · kind · text":
{facts}

RELATIONS (this entity → other, and how they connect):
{relations}

QUICK CONTACTS harvested from the facts:
{quick}

OPEN TASKS naming this entity:
{tasks}

STEP 1 — Decide what this entity IS TO AHMED: client / outreach partner / \
employee-intern / colleague / lead / supplier / friend / his own product / \
project / company / concept — and WHICH context connects them (Reval Studio, \
Alrugaib, a named campaign, etc.).

STEP 2 — Choose 4-7 sections that FIT THAT relationship. Do NOT force one \
template:
  - a client -> the deal, what Reval does for them, THEIR clients/leads, money owed/paid;
  - an employee/intern -> role, what they're working on now, performance notes;
  - an outreach partner/lead -> status, last touch, next follow-up, money discussed;
  - a concept/project -> what it is, current status, next steps;
  - a supplier -> what they supply, terms, how to reach them.
Surface MONEY, DATES, PROMISES and FOLLOW-UPS prominently. OMIT any section with \
nothing real to say. Max 4-7 sections.

STEP 3 — Write a one-line role_line for the header (e.g. "Your outreach client \
— Modern Intelligent Solution, pays Reval 3,500 SAR") and 1-3 short lowercase \
badges (e.g. ["client","paying","active"]).

OUTPUT — return ONLY this JSON object, nothing else. Use EXACTLY these four \
section "type" values; any other type is dropped:
{{"header": {{"role_line": "<one line>", "badges": ["<=3 short tags>"]}},
 "sections": [
   {{"title": "...", "type": "text",  "body": "<clean prose>"}},
   {{"title": "...", "type": "kv",    "rows": [{{"k": "Deal", "v": "3,500 SAR/mo"}}]}},
   {{"title": "...", "type": "chips", "chips": [{{"kind": "phone|email|link|map|whatsapp", "label": "...", "value": "..."}}]}},
   {{"title": "...", "type": "list",  "rows": [{{"icon": "person|company|event|task|money|note", "title": "...", "sub": "<or null>", "date": "<ISO or null>", "entity": "<other entity name to drill into, or null>"}}]}}
 ]}}

CHIP values MUST be actionable:
  phone    -> "tel:+9665XXXXXXXX"
  whatsapp -> "https://wa.me/9665XXXXXXXX"
  email    -> "mailto:name@example.com"
  map      -> "https://maps.apple.com/?q=<url-escaped address or city>"
  link     -> the raw URL
A list row with "entity" set becomes a tap-through to that entity's card — set it \
for the people/companies connected to this one.

EXAMPLE (shape only — copy the STRUCTURE, never this content):
{{"header": {{"role_line": "Your outreach client — Modern Intelligent Solution, pays Reval 3,500 SAR", "badges": ["client","paying","active"]}},
 "sections": [
   {{"title": "The deal", "type": "kv", "rows": [{{"k": "Pays Reval", "v": "3,500 SAR/mo"}}, {{"k": "Service", "v": "Lead-gen + ads"}}, {{"k": "Since", "v": "2026-05"}}]}},
   {{"title": "Contact", "type": "chips", "chips": [{{"kind": "whatsapp", "label": "WhatsApp Faisal", "value": "https://wa.me/966501234567"}}, {{"kind": "map", "label": "Riyadh office", "value": "https://maps.apple.com/?q=Riyadh"}}]}},
   {{"title": "Their leads", "type": "list", "rows": [{{"icon": "company", "title": "Bin Dawood", "sub": "warm lead handed off", "date": "2026-06-20", "entity": "Bin Dawood"}}]}},
   {{"title": "Follow-ups", "type": "text", "body": "Promised a campaign report by 2026-07-18; Ahmed still owes them the June numbers."}}
 ]}}

JSON:"""


def _build_dossier_prompt(ctx: dict) -> str:
    """Render the composition prompt from a /profile ctx (facts, relations, quick
    contacts, tasks). Includes a compact JSON example (in _DOSSIER_PROMPT)."""
    chosen = ctx.get("chosen") or {}
    facts = ctx.get("facts") or []
    relations = ctx.get("relations") or []
    quick = ctx.get("quick") or {}
    tasks = ctx.get("tasks") or []
    facts_txt = "\n".join(
        f"- {(f.get('date') or '?')} · {f.get('kind') or 'fact'} · "
        f"{f.get('fact') or ''}" for f in facts[:80]) or "(none)"
    rel_txt = "\n".join(
        f"- {r.get('rtype') or 'related'} "
        f"{'->' if r.get('direction') == 'out' else '<-'} "
        f"{r.get('other') or ''} ({r.get('otype') or 'thing'})"
        + (f": {r['other_summary']}" if r.get("other_summary") else "")
        for r in relations) or "(none)"
    qparts = [f"{lbl}: " + ", ".join(quick.get(lbl) or [])
              for lbl in ("phones", "emails", "urls", "amounts")
              if quick.get(lbl)]
    quick_txt = "; ".join(qparts) or "(none)"
    tasks_txt = "\n".join(
        f"- {t.get('text') or ''}"
        + (f" (due {t['due']})" if t.get("due") else "")
        for t in tasks) or "(none)"
    return _DOSSIER_PROMPT.format(
        today=_today(), name=chosen.get("name") or "",
        etype=chosen.get("etype") or "thing", facts=facts_txt,
        relations=rel_txt, quick=quick_txt, tasks=tasks_txt)


def _normalize_dossier(obj) -> dict | None:
    """Validate/normalize the model's output to the strict widget schema the
    Swift renderer knows. Drops unknown section types + chip kinds, coerces
    icons, caps sections=8 / rows=12 / chips=8. Returns None when nothing
    usable survives (so the caller falls back to the raw /profile card)."""
    if not isinstance(obj, dict):
        return None
    hdr = obj.get("header") if isinstance(obj.get("header"), dict) else {}
    badges = [_sv(b) for b in (hdr.get("badges") or []) if _sv(b)][:3]
    header = {"role_line": _sv(hdr.get("role_line")), "badges": badges}
    out: list = []
    for sec in (obj.get("sections") or []):
        if len(out) >= 8:
            break
        if not isinstance(sec, dict):
            continue
        typ = _sv(sec.get("type")).lower()
        if typ not in _WIDGET_TYPES:
            continue                       # unknown widget — never trust it
        title = _sv(sec.get("title"))
        if typ == "text":
            body = _sv(sec.get("body"))
            if body:
                out.append({"title": title, "type": "text", "body": body})
        elif typ == "kv":
            rows = []
            for r in (sec.get("rows") or []):
                if len(rows) >= 12:
                    break
                if not isinstance(r, dict):
                    continue
                k, v = _sv(r.get("k")), _sv(r.get("v"))
                if k or v:
                    rows.append({"k": k, "v": v})
            if rows:
                out.append({"title": title, "type": "kv", "rows": rows})
        elif typ == "chips":
            chips = []
            for c in (sec.get("chips") or []):
                if len(chips) >= 8:
                    break
                if not isinstance(c, dict):
                    continue
                kind = _sv(c.get("kind")).lower()
                value = _sv(c.get("value"))
                if kind not in _CHIP_KINDS or not value:
                    continue               # unknown chip kind / empty — drop
                chips.append({"kind": kind, "value": value,
                              "label": _sv(c.get("label")) or value})
            if chips:
                out.append({"title": title, "type": "chips", "chips": chips})
        elif typ == "list":
            rows = []
            for r in (sec.get("rows") or []):
                if len(rows) >= 12:
                    break
                if not isinstance(r, dict):
                    continue
                rtitle = _sv(r.get("title"))
                if not rtitle:
                    continue
                icon = _sv(r.get("icon")).lower()
                rows.append({"icon": icon if icon in _LIST_ICONS else "note",
                             "title": rtitle,
                             "sub": _sv(r.get("sub")) or None,
                             "date": _sv(r.get("date")) or None,
                             "entity": _sv(r.get("entity")) or None})
            if rows:
                out.append({"title": title, "type": "list", "rows": rows})
    if not out:
        return None                        # no usable section ⇒ fall back
    return {"header": header, "sections": out}


def _gather_profile(name: str, grp: str) -> tuple[dict, dict]:
    """Shared aggregation behind BOTH /profile and /dossier. Resolves the entity
    (fuzzy-tolerant), then gathers timeline, durable facts, quick contacts,
    relations, co-mentions and open tasks into the /profile response dict. Also
    returns a `ctx` carrying the chosen key, the UNCAPPED facts, and the entity's
    dossier cache props — so /dossier can compose + cache without re-querying.
    entity=None (and ctx["ekey"]=None) when nothing resolves."""
    key = _norm_entity(name)
    empty = {"entity": None, "summary": "", "read": "",
             "quick": {"phones": [], "emails": [], "urls": [], "amounts": []},
             "counts": {"facts": 0, "relations": 0, "last_touch": ""}}
    empty_ctx = {"ekey": None, "facts_total": 0, "cache": {}}
    if not key:
        return empty, empty_ctx

    # (A) resolve — FUZZY-TOLERANT. Substring matching ALONE mis-resolves a
    # voice-mangled full name ("Saud Altamimi" heard as "Saad Altamimi") to a
    # SHORT contained name ("saad", the intern). So score EVERY same-group entity
    # by max(tier score, difflib token-sort ratio over its name+aliases): a
    # full-name ratio (~0.92) BEATS a bare substring hit (0.55). Candidates are
    # the exact/alias/substring matches PLUS any entity whose ratio ≥ the fuzzy
    # floor; the top score wins, the rest go to also_matched.
    cand_rows = _graph.query(
        "MATCH (e:Entity) WHERE e.group=$grp "
        "OPTIONAL MATCH (e)<-[:MENTIONS]-(m:Memory {valid:true}) "
        "RETURN e.key, e.name, e.etype, e.aliases, e.created_at, e.summary, "
        "e.read, count(m) AS mentions", {"grp": grp}).result_set
    if not cand_rows:
        return empty, empty_ctx

    def _tier(ck, aliases) -> int:
        if ck == key:
            return 0                       # exact canonical key
        if key in (aliases or []):
            return 1                       # alias hit
        if key in ck or ck in key:
            return 2                       # substring fuzzy
        return 3                           # ratio-only (no substring overlap)

    _TIER_SCORE = {0: 1.0, 1: 1.0, 2: 0.55, 3: 0.0}
    RATIO_FLOOR = 0.85     # a ratio-only entity qualifies as a candidate here
    qtok = _sorted_tokens(name)
    cands = []
    for r in cand_rows:
        ck, aliases = r[0], (r[3] or [])
        tier = _tier(ck, aliases)
        ratio = difflib.SequenceMatcher(
            None, qtok, _sorted_tokens(r[1] or ck)).ratio()
        for a in aliases:
            ratio = max(ratio, difflib.SequenceMatcher(
                None, qtok, _sorted_tokens(a)).ratio())
        if tier == 3 and ratio < RATIO_FLOOR:
            continue                       # unrelated name — not a candidate
        cands.append({"key": ck, "name": r[1] or ck, "etype": r[2] or "thing",
                      "aliases": aliases, "created_at": r[4] or "",
                      "summary": r[5] or "", "read": r[6] or "",
                      "mentions": int(r[7] or 0),
                      "score": max(_TIER_SCORE[tier], ratio)})
    if not cands:
        return empty, empty_ctx
    # rank by max(tier-score, ratio); mentions then name break ties. A full-name
    # ratio ≥0.85 outranks a short exact-substring hit (0.55).
    cands.sort(key=lambda c: (-c["score"], -c["mentions"], c["name"].lower()))
    chosen = cands[0]
    ekey = chosen["key"]
    also_matched = [c["name"] for c in cands[1:]]

    # Fact-merge set = the CHOSEN entity + its true name-VARIANTS only, NEVER a
    # different person who merely shared the query ("saad" vs "saud altamimi").
    # A variant is substring-related, alias-linked, or ≥0.85-ratio to the chosen
    # name — so "AIN" + "AIN Smartlead outreach" still merge, but the intern's
    # card can never inherit the client's facts.
    chosen_tok = _sorted_tokens(chosen["name"])
    chosen_aliases = set(chosen["aliases"])

    def _variant(c) -> bool:
        ck = c["key"]
        if ck == ekey or ck in ekey or ekey in ck:
            return True
        if ck in chosen_aliases or ekey in set(c["aliases"]):
            return True
        return difflib.SequenceMatcher(
            None, chosen_tok, _sorted_tokens(c["name"])).ratio() >= RATIO_FLOOR

    variants = [c for c in cands if _variant(c)]
    all_keys = [c["key"] for c in variants]

    # (B) timeline + durable kinds + counts + quick, merged across (1) the
    # variant entities' MENTIONS and (2) never-linked facts that literally name
    # the topic (pre-pipeline writes).
    frows = _graph.query(
        "MATCH (m:Memory)-[:MENTIONS]->(e:Entity) "
        "WHERE e.group=$grp AND e.key IN $keys AND m.valid=true "
        "RETURN DISTINCT m.id, m.text, m.kind, m.refers_to, m.created_at, m.tone",
        {"keys": all_keys, "grp": grp}).result_set
    seen_ids = {r[0] for r in frows}
    # text sweep for never-linked facts — chosen name + aliases + variant names
    sweep_terms = _dedupe(
        [chosen["name"].lower()] + [a.lower() for a in chosen["aliases"]]
        + [c["name"].lower() for c in variants if c["key"] != ekey])
    sweep_terms = [t for t in sweep_terms if len(t) >= 3][:8]
    if sweep_terms:
        cond = " OR ".join(f"toLower(m.text) CONTAINS $s{i}"
                           for i in range(len(sweep_terms)))
        sparams = {"grp": grp, **{f"s{i}": t for i, t in enumerate(sweep_terms)}}
        # CONTAINS is only a cheap prefilter — for a short name like "ain" it
        # matches "again"/"maintain". Require a real word boundary in Python,
        # forbidding '@'/'.' neighbours too, so a name inside an email
        # local-part ("saad@tryain.com") is NOT read as a person mention.
        bound = [re.compile(r"(?<![a-z0-9@.])" + re.escape(t) + r"(?![a-z0-9@])")
                 for t in sweep_terms]
        for r in _graph.query(
            "MATCH (m:Memory) WHERE m.group=$grp AND m.valid=true AND ("
            + cond + ") RETURN m.id, m.text, m.kind, m.refers_to, "
            "m.created_at, m.tone", sparams).result_set:
            low = (r[1] or "").lower()
            if r[0] not in seen_ids and any(b.search(low) for b in bound):
                seen_ids.add(r[0])
                frows.append(r)
    timeline, event_dates = [], []
    facts_by_kind: dict = {}
    for mid, mtext, mkind, refers_to, created_at, tone in frows:
        date = (refers_to or "").strip() or (created_at or "")
        kind = mkind or "fact"
        timeline.append({"id": mid, "date": date, "kind": kind,
                         "fact": mtext or "", "tone": tone or ""})
        if kind in _DURABLE_KINDS:
            facts_by_kind.setdefault(kind, []).append(
                {"id": mid, "fact": mtext or "", "date": date})
        if kind in ("event", "checkpoint") and date:
            event_dates.append(date)
    timeline.sort(key=lambda f: f["date"], reverse=True)
    for lst in facts_by_kind.values():
        lst.sort(key=lambda f: f["date"], reverse=True)
    quick = _harvest([f["fact"] for f in timeline],
                     near_terms=[chosen["name"].lower()]
                     + [a.lower() for a in chosen["aliases"]])
    last_touch = max(event_dates) if event_dates else ""
    facts_total = len(frows)
    facts_all = list(timeline)             # UNCAPPED, for the dossier composer
    timeline = timeline[:60]

    # (C) relations both directions — ONE query (startNode picks direction);
    # guarded fallback to two directed queries if startNode is unsupported.
    def _rel(nm, ot, rt, summ, direction):
        return {"other": nm, "otype": ot or "thing",
                "rtype": _norm_rtype(rt),   # legacy free-text labels regroup
                "direction": direction, "other_summary": _first_sentence(summ)}
    try:
        rrows = _graph.query(
            "MATCH (e:Entity {key:$ekey, group:$grp})-[r:REL]-(b:Entity) "
            "RETURN b.name, b.etype, r.rtype, b.summary, "
            "CASE WHEN startNode(r).key=$ekey THEN 'out' ELSE 'in' END "
            "ORDER BY b.name",
            {"ekey": ekey, "grp": grp}).result_set
        relations = [_rel(r[0], r[1], r[2], r[3], r[4]) for r in rrows]
    except Exception:  # noqa: BLE001 — startNode unsupported: two directed queries
        relations = []
        for direction, arrow in (("out", "-[r:REL]->"), ("in", "<-[r:REL]-")):
            for r in _graph.query(
                "MATCH (a:Entity {key:$ekey, group:$grp})" + arrow + "(b:Entity) "
                "RETURN b.name, b.etype, r.rtype, b.summary",
                    {"ekey": ekey, "grp": grp}).result_set:
                relations.append(_rel(r[0], r[1], r[2], r[3], direction))

    # (C2) co-mentioned entities — the cast that appears in the SAME facts
    # (clients, CEOs, partners). Cap to entities with >=2 co-mentioning facts OR
    # a real summary, so one-off generic 'task'/'thing' nodes don't clutter it.
    have_rel = {r["other"].lower() for r in relations}
    try:
        for r in _graph.query(
            "MATCH (e:Entity)<-[:MENTIONS]-(m:Memory {valid:true})"
            "-[:MENTIONS]->(o:Entity) "
            "WHERE e.group=$grp AND e.key IN $keys AND NOT o.key IN $keys "
            "AND o.group=$grp AND o.etype <> 'amount' "
            "AND NOT (e)-[:NOT_RELATED]-(o) "
            "WITH o, count(DISTINCT m) AS n "
            "WHERE n >= 2 OR (o.summary IS NOT NULL AND o.summary <> '') "
            "RETURN o.name, o.etype, o.summary, n "
            "ORDER BY n DESC LIMIT 12",
                {"keys": all_keys, "grp": grp}).result_set:
            if (r[0] or "").lower() not in have_rel:
                relations.append(_rel(r[0], r[1], "mentioned-with", r[2], "out"))
    except Exception:  # noqa: BLE001 — co-mentions are best-effort
        pass

    # (D) open tasks whose text names the entity or an alias — ONE query
    terms = _dedupe([chosen["name"].lower()] + list(chosen["aliases"]))
    terms = [t for t in terms if len(t) >= 2]
    tasks = []
    if terms:
        cond = " OR ".join(f"toLower(t.text) CONTAINS $t{i}"
                           for i in range(len(terms)))
        tparams = {"grp": grp, **{f"t{i}": t for i, t in enumerate(terms)}}
        tasks = [{"id": r[0], "text": r[1], "due": r[2] or None}
                 for r in _graph.query(
                     "MATCH (t:Task) WHERE t.grp=$grp AND t.done=false AND ("
                     + cond + ") RETURN t.id, t.text, t.due",
                     tparams).result_set]

    # (E) the chosen entity's dossier cache props — powers /dossier freshness +
    # /profile's has_composed, with no second round trip.
    cache: dict = {}
    try:
        crow = _graph.query(
            "MATCH (e:Entity {key:$k, group:$grp}) "
            "RETURN e.dossier_json, e.dossier_at, e.dossier_nfacts",
            {"k": ekey, "grp": grp}).result_set
        if crow:
            cache = {"json": crow[0][0], "at": crow[0][1],
                     "nfacts": crow[0][2]}
    except Exception:  # noqa: BLE001 — cache read is best-effort
        pass

    # assemble — always-present core; empty optional sections omitted
    resp = {
        "entity": {"key": ekey, "name": chosen["name"],
                   "etype": chosen["etype"], "aliases": chosen["aliases"],
                   "created_at": chosen["created_at"]},
        "summary": chosen["summary"], "read": chosen["read"], "quick": quick,
        "counts": {"facts": facts_total, "relations": len(relations),
                   "last_touch": last_touch},
    }
    if relations:
        resp["relations"] = relations
    if timeline:
        resp["timeline"] = timeline
    if facts_by_kind:
        resp["facts_by_kind"] = facts_by_kind
    if tasks:
        resp["tasks"] = tasks
    if also_matched:
        resp["also_matched"] = also_matched
    ctx = {"ekey": ekey, "chosen": chosen, "facts": facts_all,
           "facts_total": facts_total, "relations": relations, "quick": quick,
           "tasks": tasks, "cache": cache}
    return resp, ctx


@app.get("/profile")
def profile(name: str, group: str | None = None,
            authorization: str | None = Header(default=None)):
    """FULL dossier of one entity for a HUD card — the FAST quick-render the HUD
    depends on, so this response SHAPE is stable. Resolves the name fuzzily
    (exact key → alias → substring → difflib token-sort ratio; a full-name ratio
    ≥0.85 beats a short substring hit), picks the best match (rest → also_matched),
    and returns {entity, summary, read, quick, relations, timeline, facts_by_kind,
    tasks, counts, also_matched}. Every section is None-tolerant and empty optional
    sections are omitted, so any etype renders gracefully. `has_composed` (NEW,
    additive) is true when a fresh LLM-composed dossier is cached and instantly
    available from POST /dossier."""
    _auth(authorization)
    grp = group or GROUP_DEFAULT
    resp, ctx = _gather_profile(name, grp)
    resp["has_composed"] = _dossier_fresh(ctx.get("cache") or {},
                                          ctx.get("facts_total", 0))
    return resp


class Dossier(BaseModel):
    name: str
    group: str | None = None
    force: bool | None = False


@app.post("/dossier")
def dossier(body: Dossier, authorization: str | None = Header(default=None)):
    """COMPOSED dossier — an LLM READS the raw /profile aggregate and returns an
    organized, ADAPTIVE profile card {header{role_line,badges}, sections[]} whose
    sections fit what the entity IS TO AHMED (client vs intern vs concept …). The
    four widget types are text / kv / chips / list (validated hard in code).

    CACHE: the composed JSON is stored on the entity node (dossier_json/at/nfacts).
    A cache hit — SAME fact count, < 24h old, and `force` not set — returns
    instantly {composed, cached:true}. Otherwise ONE DeepSeek call (DOSSIER_MODEL,
    temp 0.3) composes it; ANY LLM failure (or no key) returns {composed:null} so
    the caller falls back to the raw /profile card. On success the result is
    validated, cached, and returned {composed, cached:false}."""
    _auth(authorization)
    grp = body.group or GROUP_DEFAULT
    _resp, ctx = _gather_profile(body.name, grp)
    ekey = ctx.get("ekey")
    if not ekey:
        return {"composed": None, "cached": False}
    nfacts = int(ctx.get("facts_total", 0))
    cache = ctx.get("cache") or {}
    # CACHE hit — same fact count, fresh, not forced ⇒ return the stored dossier
    if not body.force and _dossier_fresh(cache, nfacts):
        try:
            return {"composed": json.loads(cache["json"]), "cached": True}
        except Exception:  # noqa: BLE001 — corrupt cache ⇒ recompose below
            pass
    # compose — ONE DeepSeek call; any failure ⇒ {composed:null} (graceful)
    try:
        got = _deepseek_json(_build_dossier_prompt(ctx), max_tokens=5000,
                             model=DOSSIER_MODEL, temperature=0.3)
        composed = _normalize_dossier(got)
    except Exception as e:  # noqa: BLE001 — LLM failure ⇒ fall back to /profile
        print(f"dossier compose FAILED for {ekey!r}: "
              f"{type(e).__name__}: {str(e)[:200]}", flush=True)
        return {"composed": None, "cached": False}
    if composed is None:
        return {"composed": None, "cached": False}
    # store on the entity node (best-effort — a cache-write failure isn't fatal)
    try:
        _graph.query(
            "MATCH (e:Entity {key:$k, group:$grp}) "
            "SET e.dossier_json=$j, e.dossier_at=$at, e.dossier_nfacts=$n",
            {"k": ekey, "grp": grp, "j": json.dumps(composed),
             "at": _now_iso(), "n": nfacts})
    except Exception:  # noqa: BLE001
        pass
    return {"composed": composed, "cached": False}


class EntityUpdate(BaseModel):
    key: str
    group: str | None = None
    etype: str | None = None
    name: str | None = None


@app.post("/entity/update")
def entity_update(body: EntityUpdate,
                  authorization: str | None = Header(default=None)):
    """Correct one entity's TYPE or DISPLAY NAME (e.g. a person mis-typed as a
    thing). Aliases + canonical key are untouched, so all its facts/relations stay
    attached. `etype` must be one of person/company/place/product/amount/thing/
    concept. Nothing else is mutable here."""
    _auth(authorization)
    grp = body.group or GROUP_DEFAULT
    key = _norm_entity(body.key)
    if not key:
        raise HTTPException(status_code=400, detail="key required")
    rows = _graph.query(
        "MATCH (e:Entity {key:$k, group:$grp}) RETURN e.name, e.etype",
        {"k": key, "grp": grp}).result_set
    if not rows:
        raise HTTPException(status_code=404, detail="entity not found")
    sets, params = [], {"k": key, "grp": grp}
    if body.etype is not None:
        et = (body.etype or "").strip().lower()
        if et not in ETYPES:
            raise HTTPException(
                status_code=400,
                detail="etype must be one of " + ", ".join(sorted(ETYPES)))
        sets.append("e.etype=$et")
        params["et"] = et
    if body.name is not None and body.name.strip():
        sets.append("e.name=$nm")
        params["nm"] = body.name.strip()
    if sets:
        _graph.query(
            "MATCH (e:Entity {key:$k, group:$grp}) SET " + ", ".join(sets),
            params)
    out = _graph.query(
        "MATCH (e:Entity {key:$k, group:$grp}) RETURN e.name, e.etype",
        {"k": key, "grp": grp}).result_set
    return {"ok": True, "key": key,
            "name": (out[0][0] if out else ""),
            "etype": (out[0][1] if out else "")}


class Summary(BaseModel):
    name: str
    text: str
    group: str | None = None


@app.post("/summary")
def summary_set(body: Summary, authorization: str | None = Header(default=None)):
    """Upsert a current-state SUMMARY onto an entity — the reflection engine's
    rolling 'profile' so 'pull up Al Temimi' returns the gist, not just raw facts."""
    _auth(authorization)
    grp = body.group or GROUP_DEFAULT
    key = _norm_entity(body.name)
    if not key or not body.text.strip():
        raise HTTPException(status_code=400, detail="name and text required")
    _graph.query(
        "MERGE (e:Entity {key:$key, group:$grp}) "
        "ON CREATE SET e.name=$name, e.etype='thing', e.created_at=$now "
        "SET e.summary=$text, e.summary_at=$now",
        {"key": key, "grp": grp, "name": body.name.strip(),
         "text": body.text.strip(), "now": _now_iso()})
    return {"ok": True}


@app.post("/read")
def read_set(body: Summary, authorization: str | None = Header(default=None)):
    """THEORY OF MIND — upsert a psychological 'read' onto a person/company:
    what they seem to want, how they decide, the state of the relationship, and
    how Ahmed should approach them. Models THEM, for Ahmed. (Reuses the Summary
    shape: {name, text}.)"""
    _auth(authorization)
    grp = body.group or GROUP_DEFAULT
    key = _norm_entity(body.name)
    if not key or not body.text.strip():
        raise HTTPException(status_code=400, detail="name and text required")
    _graph.query(
        "MERGE (e:Entity {key:$key, group:$grp}) "
        "ON CREATE SET e.name=$name, e.etype='thing', e.created_at=$now "
        "SET e.read=$text, e.read_at=$now",
        {"key": key, "grp": grp, "name": body.name.strip(),
         "text": body.text.strip(), "now": _now_iso()})
    return {"ok": True}


class Values(BaseModel):
    text: str
    group: str | None = None


@app.post("/values")
def values_set(body: Values, authorization: str | None = Header(default=None)):
    """VALUE MODEL — upsert Ahmed's north-star: his priorities, what he chases,
    what he avoids. A singleton profile the reflection engine and the brain
    consult so advice fits HIM, not generic best-practice."""
    _auth(authorization)
    grp = body.group or GROUP_DEFAULT
    if not body.text.strip():
        raise HTTPException(status_code=400, detail="text required")
    _graph.query(
        "MERGE (p:Profile {id:'values', group:$grp}) "
        "SET p.text=$text, p.updated_at=$now",
        {"grp": grp, "text": body.text.strip(), "now": _now_iso()})
    return {"ok": True}


@app.get("/values")
def values_get(group: str | None = None,
               authorization: str | None = Header(default=None)):
    _auth(authorization)
    grp = group or GROUP_DEFAULT
    rows = _graph.query(
        "MATCH (p:Profile {id:'values', group:$grp}) RETURN p.text, p.updated_at",
        {"grp": grp}).result_set
    if not rows:
        return {"text": "", "updated_at": ""}
    return {"text": rows[0][0] or "", "updated_at": rows[0][1] or ""}


class Retire(BaseModel):
    id: str
    group: str | None = None


@app.post("/retire")
def retire(body: Retire, authorization: str | None = Header(default=None)):
    """Soft-retire a memory — mark it stale (hidden from recall) but KEEP it as
    history, unlike /forget which deletes. Used by the consolidation pass to fold
    away near-duplicate facts without losing anything."""
    _auth(authorization)
    _graph.query("MATCH (m:Memory {id:$id}) SET m.valid=false, m.invalid_at=$now",
                 {"id": body.id.strip(), "now": _now_iso()})
    return {"ok": True}


class AttachEntities(BaseModel):
    memory_id: str
    entities: list[dict]
    group: str | None = None


@app.post("/attach_entities")
def attach_entities_ep(body: AttachEntities,
                       authorization: str | None = Header(default=None)):
    """Backfill hook — attach extracted entities to an EXISTING memory (used by
    the engine-side entity backfill over the memories saved before Layer 2)."""
    _auth(authorization)
    grp = body.group or GROUP_DEFAULT
    n = _attach_entities(body.memory_id, body.entities, grp)
    return {"attached": n}


class Relation(BaseModel):
    from_e: str = Field(alias="from")   # 'from' is a Python keyword → aliased
    to: str
    rtype: str | None = "related"
    group: str | None = None


@app.post("/relation")
def relation_add(body: Relation,
                 authorization: str | None = Header(default=None)):
    """Wire an entity↔entity edge (:Entity)-[:REL {rtype}]->(:Entity). Both the
    `from` and `to` names run through the resolver, so the edge attaches to the
    same canonical nodes that mentions do."""
    _auth(authorization)
    grp = body.group or GROUP_DEFAULT
    fn = (body.from_e or "").strip()
    tn = (body.to or "").strip()
    if not fn or not tn:
        raise HTTPException(status_code=400, detail="from and to required")
    _add_relation(fn, tn, body.rtype or "related", grp)
    return {"ok": True}


def _merge_one(keep: str, mk: str, group: str) -> bool:
    """Fold entity `mk` into `keep`: re-point MENTIONS + REL edges, union aliases,
    keep the richer (longer) summary/read, then delete `mk`. No APOC — REL edges
    are collected in Python and recreated so their rtype/created_at survive."""
    if not mk or mk == keep:
        return False
    krow = _graph.query(
        "MATCH (e:Entity {key:$k, group:$g}) "
        "RETURN e.aliases, e.summary, e.read", {"k": keep, "g": group}).result_set
    mrow = _graph.query(
        "MATCH (e:Entity {key:$k, group:$g}) "
        "RETURN e.aliases, e.summary, e.read", {"k": mk, "g": group}).result_set
    if not krow or not mrow:
        return False
    # move MENTIONS (property-less → safe to re-point inline)
    _graph.query(
        "MATCH (m:Memory)-[r:MENTIONS]->(:Entity {key:$mk, group:$g}) "
        "MATCH (k:Entity {key:$keep, group:$g}) "
        "MERGE (m)-[:MENTIONS]->(k) DELETE r",
        {"mk": mk, "keep": keep, "g": group})
    # move REL out (mk -> b) and in (b -> mk), preserving rtype/created_at
    now = _now_iso()
    outs = _graph.query(
        "MATCH (:Entity {key:$mk, group:$g})-[r:REL]->(b:Entity) "
        "RETURN b.key, r.rtype, r.created_at", {"mk": mk, "g": group}).result_set
    for bkey, rt, cat in outs:
        if bkey in (keep, mk):
            continue
        _graph.query(
            "MATCH (k:Entity {key:$keep, group:$g}),(b:Entity {key:$bk, group:$g}) "
            "MERGE (k)-[nr:REL {rtype:$rt}]->(b) ON CREATE SET nr.created_at=$cat",
            {"keep": keep, "bk": bkey, "g": group, "rt": rt or "related",
             "cat": cat or now})
    ins = _graph.query(
        "MATCH (b:Entity)-[r:REL]->(:Entity {key:$mk, group:$g}) "
        "RETURN b.key, r.rtype, r.created_at", {"mk": mk, "g": group}).result_set
    for bkey, rt, cat in ins:
        if bkey in (keep, mk):
            continue
        _graph.query(
            "MATCH (b:Entity {key:$bk, group:$g}),(k:Entity {key:$keep, group:$g}) "
            "MERGE (b)-[nr:REL {rtype:$rt}]->(k) ON CREATE SET nr.created_at=$cat",
            {"keep": keep, "bk": bkey, "g": group, "rt": rt or "related",
             "cat": cat or now})
    # union aliases (+ the merged key), keep the richer summary/read
    kal = krow[0][0] or []
    mal = mrow[0][0] or []
    union = [a for a in dict.fromkeys(list(kal) + list(mal) + [mk]) if a != keep]
    ksum, kread = krow[0][1] or "", krow[0][2] or ""
    msum, mread = mrow[0][1] or "", mrow[0][2] or ""
    summary = ksum if len(ksum) >= len(msum) else msum
    read = kread if len(kread) >= len(mread) else mread
    _graph.query(
        "MATCH (k:Entity {key:$keep, group:$g}) "
        "SET k.aliases=$al, k.summary=$sum, k.read=$rd",
        {"keep": keep, "g": group, "al": union, "sum": summary, "rd": read})
    _graph.query("MATCH (e:Entity {key:$mk, group:$g}) DETACH DELETE e",
                 {"mk": mk, "g": group})
    return True


class EntityUnrelate(BaseModel):
    a: str
    b: str
    group: str | None = None


class EntityAliases(BaseModel):
    key: str
    aliases: list
    group: str | None = None


@app.post("/entity/aliases")
def entity_aliases(body: EntityAliases,
                   authorization: str | None = Header(default=None)):
    """REPAIR/curation: REPLACE an entity's alias list outright (dedup, lowered,
    self-name removed). Used to strip pollution after the corrupt-vector alias
    auto-attach incident; aliases regrow organically through resolution."""
    _auth(authorization)
    grp = body.group or GROUP_DEFAULT
    k = _norm_entity(body.key)
    clean = _dedupe([_norm_entity(a) for a in (body.aliases or [])
                     if _norm_entity(a) and _norm_entity(a) != k])
    res = _graph.query(
        "MATCH (e:Entity {key:$k, group:$grp}) SET e.aliases=$al "
        "RETURN e.key", {"k": k, "grp": grp, "al": clean}).result_set
    if not res:
        raise HTTPException(status_code=404, detail="entity not found")
    return {"ok": True, "key": k, "aliases": clean}


class EntityDetach(BaseModel):
    key: str
    group: str | None = None


@app.post("/entity/detach")
def entity_detach(body: EntityDetach,
                  authorization: str | None = Header(default=None)):
    """REPAIR: remove MENTIONS edges from an entity to memories whose text does
    NOT actually contain the entity's name or an alias (word-boundary). Used to
    undo bad merges that hijacked another entity's facts. Returns the count."""
    _auth(authorization)
    grp = body.group or GROUP_DEFAULT
    k = _norm_entity(body.key)
    rows = _graph.query(
        "MATCH (e:Entity {key:$k, group:$grp})<-[r:MENTIONS]-(m:Memory) "
        "RETURN m.id, m.text, e.name, e.aliases", {"k": k, "grp": grp}).result_set
    if not rows:
        return {"ok": True, "detached": 0}
    name = (rows[0][2] or k).lower()
    aliases = [a.lower() for a in (rows[0][3] or [])]
    terms = [t for t in _dedupe([name] + aliases) if len(t) >= 2]
    bound = [re.compile(r"(?<![a-z0-9@.])" + re.escape(t) + r"(?![a-z0-9@])")
             for t in terms]
    bad = [r[0] for r in rows
           if not any(b.search((r[1] or "").lower()) for b in bound)]
    if bad:
        _graph.query(
            "UNWIND $ids AS mid MATCH (e:Entity {key:$k, group:$grp})"
            "<-[r:MENTIONS]-(m:Memory {id:mid}) DELETE r",
            {"ids": bad, "k": k, "grp": grp})
    return {"ok": True, "detached": len(bad)}


@app.post("/entity/unrelate")
def entity_unrelate(body: EntityUnrelate,
                    authorization: str | None = Header(default=None)):
    """HUMAN-CONFIRMED edge removal (the dossier card's 'Not related' action).
    Deletes every REL edge between the two entities and marks the pair
    NOT_RELATED so co-mention grouping never reconnects them. Deliberately a
    human-only action — no LLM is allowed to cut edges autonomously."""
    _auth(authorization)
    grp = body.group or GROUP_DEFAULT
    ka, kb = _norm_entity(body.a), _norm_entity(body.b)
    if not ka or not kb or ka == kb:
        raise HTTPException(status_code=400, detail="two distinct names needed")
    # resolve fuzzily like /entity does (exact → alias → substring)
    def _resolve(k):
        rows = _graph.query(
            "MATCH (e:Entity) WHERE e.group=$grp AND "
            "(e.key=$k OR $k IN coalesce(e.aliases,[]) OR e.key CONTAINS $k) "
            "RETURN e.key ORDER BY size(e.key) LIMIT 1",
            {"grp": grp, "k": k}).result_set
        return rows[0][0] if rows else None
    ra, rb = _resolve(ka), _resolve(kb)
    if not ra or not rb or ra == rb:
        raise HTTPException(status_code=404, detail="entity not found")
    _graph.query(
        "MATCH (a:Entity {key:$a, group:$grp})-[r:REL]-(b:Entity {key:$b, "
        "group:$grp}) DELETE r", {"a": ra, "b": rb, "grp": grp})
    _graph.query(
        "MATCH (a:Entity {key:$a, group:$grp}),(b:Entity {key:$b, group:$grp}) "
        "MERGE (a)-[x:NOT_RELATED]->(b) ON CREATE SET x.created_at=$now",
        {"a": ra, "b": rb, "grp": grp, "now": _now_iso()})
    return {"ok": True, "a": ra, "b": rb}


class EntityMerge(BaseModel):
    keep_key: str
    merge_keys: list[str]
    group: str | None = None


@app.post("/entities/merge")
def entities_merge(body: EntityMerge,
                   authorization: str | None = Header(default=None)):
    """Manually fold one or more duplicate entities into a canonical one — moves
    every MENTIONS + REL edge, unions aliases, and keeps the richer profile."""
    _auth(authorization)
    grp = body.group or GROUP_DEFAULT
    keep = _norm_entity(body.keep_key)
    if not keep:
        raise HTTPException(status_code=400, detail="keep_key required")
    merged = []
    for raw in (body.merge_keys or []):
        mk = _norm_entity(raw)
        if mk and mk != keep and _merge_one(keep, mk, grp):
            merged.append(mk)
    return {"ok": True, "keep": keep, "merged": merged}


class DedupBackfill(BaseModel):
    dry_run: bool = True
    group: str | None = None


@app.post("/entities/dedup_backfill")
def entities_dedup_backfill(body: DedupBackfill,
                            authorization: str | None = Header(default=None)):
    """Scan same-type entities for lookalike pairs (max of name-embedding cosine
    and difflib token ratio). `dry_run` returns every candidate pair (≥0.80) with
    its score; a real run AUTO-MERGES the confident ones (≥0.95, keeping whichever
    has more mentions) and returns the 0.80–0.95 band as `ambiguous` for review."""
    _auth(authorization)
    grp = body.group or GROUP_DEFAULT
    rows = _graph.query(
        "MATCH (e:Entity {group:$grp}) RETURN e.key, e.name, e.etype, e.aliases",
        {"grp": grp}).result_set
    ents = [{"key": r[0], "name": r[1] or r[0], "etype": (r[2] or "thing"),
             "aliases": r[3] or []} for r in rows]
    embs = {e["key"]: _embed(e["name"]) for e in ents}
    # mention counts pick the survivor on an auto-merge (richer node wins)
    mc: dict = {}
    for r in _graph.query(
        "MATCH (e:Entity {group:$grp}) OPTIONAL MATCH (e)<-[r:MENTIONS]-() "
            "RETURN e.key, count(r)", {"grp": grp}).result_set:
        mc[r[0]] = int(r[1] or 0)
    by_type: dict = {}
    for e in ents:
        by_type.setdefault(e["etype"], []).append(e)
    pairs = []
    for etype, group_ents in by_type.items():
        for i in range(len(group_ents)):
            for j in range(i + 1, len(group_ents)):
                a, b = group_ents[i], group_ents[j]
                emb_sim = _cosine(embs[a["key"]], embs[b["key"]])
                atok = _sorted_tokens(a["name"])
                btok = _sorted_tokens(b["name"])
                tok = difflib.SequenceMatcher(None, atok, btok).ratio()
                for al in list(a["aliases"]) + list(b["aliases"]):
                    tok = max(tok, difflib.SequenceMatcher(
                        None, atok, _sorted_tokens(al)).ratio(),
                        difflib.SequenceMatcher(
                            None, btok, _sorted_tokens(al)).ratio())
                sc = max(emb_sim, tok)
                if sc >= 0.80:
                    pairs.append((a["key"], b["key"], round(float(sc), 3),
                                  round(float(tok), 3)))
    pairs.sort(key=lambda p: p[2], reverse=True)
    if body.dry_run:
        return {"dry_run": True, "candidates": len(pairs),
                "pairs": [{"a": a, "b": b, "score": s, "tok": t}
                          for a, b, s, t in pairs]}
    # real run — auto-merge the confident ones, surface the rest as ambiguous.
    # HARD RULE learned 2026-07-16: embedding similarity ALONE can never
    # auto-merge — corrupted/degenerate vectors once scored unrelated names at
    # 1.0 and folded "smart price tags" into "ahmed". Auto-merge needs the
    # STRINGS to agree too (tok >= 0.7); embedding-only pairs go to the
    # ambiguous band for LLM/human adjudication.
    gone: set = set()
    merged, ambiguous = [], []
    for a, b, s, t in pairs:
        if a in gone or b in gone:
            continue
        if s >= 0.95 and t >= 0.7:
            keep, drop = (a, b) if mc.get(a, 0) >= mc.get(b, 0) else (b, a)
            if _merge_one(keep, drop, grp):
                gone.add(drop)
                merged.append({"keep": keep, "merged": drop, "score": s})
        else:
            ambiguous.append({"a": a, "b": b, "score": s, "tok": t})
    return {"dry_run": False, "merged": merged, "ambiguous": ambiguous}


# ---------------------------------------------------------------------------
# REFLECTION — the nightly engine (DeepSeek, on a Railway cron) writes its
# findings here as :Insight nodes; devices pull the unsurfaced ones and Jarvis
# raises them proactively, then marks them seen.
# ---------------------------------------------------------------------------
class Insight(BaseModel):
    text: str
    title: str | None = ""
    itype: str | None = "observation"   # opportunity | observation | nudge
    group: str | None = None


@app.post("/insight")
def insight_add(body: Insight, authorization: str | None = Header(default=None)):
    _auth(authorization)
    text = body.text.strip()
    if not text:
        raise HTTPException(status_code=400, detail="empty insight")
    grp = body.group or GROUP_DEFAULT
    emb = _embed(text)
    # dedup — skip if a near-identical insight already exists (don't repeat the
    # same thought every night)
    try:
        dup = _graph.query(
            "CALL db.idx.vector.queryNodes('Insight','embedding',3,vecf32($e)) "
            "YIELD node, score WHERE node.group=$grp AND score<$thr "
            "RETURN count(node)", {"e": emb, "grp": grp,
                                   "thr": 1.0 - INSIGHT_DEDUP}).result_set
        if dup and int(dup[0][0]) > 0:
            return {"created": False, "reason": "duplicate"}
    except Exception:  # noqa: BLE001 — index may be empty/first run
        pass
    iid = uuid.uuid4().hex
    _graph.query(
        "CREATE (i:Insight {id:$id, text:$text, title:$title, itype:$itype, "
        "group:$grp, created_at:$now, surfaced:false}) "
        "SET i.embedding = vecf32($emb)",
        {"id": iid, "text": text, "title": body.title or "",
         "itype": (body.itype or "observation"), "grp": grp,
         "now": _now_iso(), "emb": emb})
    return {"id": iid, "created": True}


@app.get("/insights")
def insights_list(group: str | None = None, unsurfaced_only: bool = True,
                  limit: int = 20,
                  authorization: str | None = Header(default=None)):
    _auth(authorization)
    grp = group or GROUP_DEFAULT
    cond = "AND i.surfaced=false " if unsurfaced_only else ""
    rows = _graph.query(
        "MATCH (i:Insight) WHERE i.group=$grp " + cond +
        "RETURN i.id, i.text, i.title, i.itype, i.created_at "
        "ORDER BY i.created_at DESC LIMIT $lim",
        {"grp": grp, "lim": max(1, min(limit, 100))}).result_set
    return {"insights": [{"id": r[0], "text": r[1], "title": r[2],
                          "itype": r[3], "created_at": r[4]} for r in rows]}


@app.get("/insights/search")
def insights_search(q: str, group: str | None = None, k: int = 3,
                    authorization: str | None = Header(default=None)):
    """Unsurfaced insights RELEVANT to a query — used to raise a reflection
    contextually (only when Ahmed asks about something related), never on a timer."""
    import re as _re
    _auth(authorization)
    grp = group or GROUP_DEFAULT
    emb = _embed(q)
    seen: set = set()
    out: list = []
    # meaning pass
    try:
        for r in _graph.query(
            "CALL db.idx.vector.queryNodes('Insight','embedding',$k,vecf32($emb)) "
            "YIELD node, score WHERE node.group=$grp AND node.surfaced=false "
            "AND score < 0.5 RETURN node.id, node.text, node.title, node.itype, "
                "score", {"k": max(k, 1) * 3 + 6, "emb": emb,
                          "grp": grp}).result_set:
            if r[0] not in seen:
                seen.add(r[0])
                out.append({"id": r[0], "text": r[1], "title": r[2],
                            "itype": r[3], "score": round(_sim(r[4]), 3)})
    except Exception:  # noqa: BLE001 — index empty / no insights yet
        pass
    # keyword/entity pass — an insight naming what he asked about (e.g. a client)
    words = [w for w in _re.findall(r"[a-z0-9]{3,}", q.lower())
             if w not in _STOP][:8]
    if words:
        kwq = ("MATCH (i:Insight) WHERE i.group=$grp AND i.surfaced=false AND (" +
               " OR ".join(f"toLower(i.text) CONTAINS $w{n}"
                           for n in range(len(words))) +
               ") RETURN i.id, i.text, i.title, i.itype")
        params = {"grp": grp, **{f"w{n}": w for n, w in enumerate(words)}}
        try:
            for r in _graph.query(kwq, params).result_set:
                if r[0] not in seen:
                    seen.add(r[0])
                    out.append({"id": r[0], "text": r[1], "title": r[2],
                                "itype": r[3], "score": 0.6})
        except Exception:  # noqa: BLE001
            pass
    return {"insights": out[:max(k, 1)]}


class InsightSeen(BaseModel):
    id: str
    group: str | None = None


@app.post("/insight/seen")
def insight_seen(body: InsightSeen,
                 authorization: str | None = Header(default=None)):
    _auth(authorization)
    _graph.query("MATCH (i:Insight {id:$id}) SET i.surfaced=true",
                 {"id": body.id})
    return {"ok": True}


class Backfill(BaseModel):
    threshold: float | None = None
    max_links: int | None = None
    dry_run: bool = False
    group: str | None = None


@app.post("/backfill_links")
def backfill_links(body: Backfill,
                   authorization: str | None = Header(default=None)):
    """LAYER 1 backfill — run auto-linking over EVERY existing memory once, so
    today's loose cloud gets wired up. `dry_run` reports what it WOULD link (with
    samples) without touching the graph, so the threshold can be tuned first."""
    _auth(authorization)
    grp = body.group or GROUP_DEFAULT
    thr = body.threshold if body.threshold is not None else AUTOLINK_THRESHOLD
    mx = body.max_links or AUTOLINK_MAX
    nodes = _graph.query(
        "MATCH (m:Memory) WHERE m.group=$grp AND m.valid=true "
        "RETURN m.id, m.text", {"grp": grp}).result_set
    pairs: set = set()
    samples: list = []
    for nid, ntext in nodes:
        emb = _embed(ntext or "")
        for lk in _auto_link(nid, ntext or "", emb, grp, thr, mx,
                             dry_run=body.dry_run):
            pair = frozenset((nid, lk["id"]))
            if pair in pairs:
                continue
            pairs.add(pair)
            if len(samples) < 30:
                samples.append({"a": (ntext or "")[:60],
                                "b": (lk["text"] or "")[:60],
                                "score": lk["score"]})
    return {"nodes": len(nodes), "linked_pairs": len(pairs),
            "threshold": thr, "dry_run": body.dry_run, "samples": samples}


# ---------------------------------------------------------------------------
# Re-embedding, export, reflection heartbeat, and Ahmed's core context block.
# ---------------------------------------------------------------------------
class Reembed(BaseModel):
    batch: int | None = 200


@app.post("/reembed")
def reembed(body: Reembed, authorization: str | None = Header(default=None)):
    """Re-embed EVERY Memory + Insight + Entity with the CURRENT model and
    recreate the vector indexes at the current dim (drop+create, so a dim change
    from an EMBED_MODEL swap is handled). Idempotent — re-running just recomputes
    the same vectors. Run once on deploy day after switching the embed model."""
    _auth(authorization)
    labels = (("Memory", "embedding"), ("Insight", "embedding"),
              ("Entity", "embedding"))
    # DROP the vector indexes FIRST so a dimension CHANGE can't fail: with no
    # index present, re-embedding is free to write new-dim vectors, then we
    # recreate the indexes at EMBED_DIM afterwards. (Same-dim runs: harmless.)
    for label, prop in labels:
        try:
            _graph.query(f"DROP VECTOR INDEX FOR (n:{label}) (n.{prop})")
        except Exception:  # noqa: BLE001 — not present / unsupported drop syntax
            pass
    batch = max(1, min(int(body.batch or 200), 2000))
    counts: dict = {}
    # Memory + Insight are keyed by id
    for label in ("Memory", "Insight"):
        rows = _graph.query(
            f"MATCH (n:{label}) RETURN n.id, n.text").result_set
        c = 0
        for i in range(0, len(rows), batch):
            for rid, txt in rows[i:i + batch]:
                if not rid:
                    continue
                _graph.query(
                    f"MATCH (n:{label} {{id:$id}}) SET n.embedding=vecf32($emb)",
                    {"id": rid, "emb": _embed(txt or "")})
                c += 1
        counts[label] = c
    # Entity is keyed by (key, group); embed its name
    erows = _graph.query(
        "MATCH (e:Entity) RETURN e.key, e.group, e.name").result_set
    c = 0
    for ekey, eg, ename in erows:
        if not ekey:
            continue
        _graph.query(
            "MATCH (e:Entity {key:$k, group:$g}) SET e.embedding=vecf32($emb)",
            {"k": ekey, "g": eg, "emb": _embed(ename or ekey)})
        c += 1
    counts["Entity"] = c
    # recreate the vector indexes at the current EMBED_DIM (now that every node
    # holds a current-dim vector). A same-dim run just no-ops ("already exists").
    for label, prop in labels:
        try:
            _graph.query(
                f"CREATE VECTOR INDEX FOR (n:{label}) ON (n.{prop}) "
                f"OPTIONS {{dimension: {EMBED_DIM}, similarityFunction: 'cosine'}}")
        except Exception as e:  # noqa: BLE001
            if "already" not in str(e).lower():
                print(f"[reembed] {label} index note: {e}", flush=True)
    return {"ok": True, "model": EMBED_MODEL, "dim": EMBED_DIM,
            "reembedded": counts}


@app.get("/export")
def export(group: str | None = None,
           authorization: str | None = Header(default=None)):
    """COMPLETE dump of a group for nightly backups — memories (INCLUDING stale),
    entities, links, mentions, relations, insights, tasks, profile, core, meta."""
    _auth(authorization)
    grp = group or GROUP_DEFAULT
    memories = [{"id": r[0], "text": r[1], "kind": r[2], "source": r[3],
                 "created_at": r[4], "refers_to": r[5], "valid": r[6],
                 "valid_from": r[7], "valid_to": r[8], "invalid_at": r[9],
                 "superseded_by": r[10], "confidence": r[11], "importance": r[12],
                 "srel": r[13], "tone": r[14], "access_count": r[15] or 0}
                for r in _graph.query(
        "MATCH (m:Memory {group:$grp}) RETURN m.id, m.text, m.kind, m.source, "
        "m.created_at, m.refers_to, m.valid, m.valid_from, m.valid_to, "
        "m.invalid_at, m.superseded_by, m.confidence, m.importance, m.srel, "
        "m.tone, m.access_count", {"grp": grp}).result_set]
    entities = [{"key": r[0], "name": r[1], "etype": r[2], "aliases": r[3] or [],
                 "summary": r[4] or "", "read": r[5] or "", "created_at": r[6]}
                for r in _graph.query(
        "MATCH (e:Entity {group:$grp}) RETURN e.key, e.name, e.etype, e.aliases, "
        "e.summary, e.read, e.created_at", {"grp": grp}).result_set]
    links = [{"from": r[0], "to": r[1], "type": r[2]} for r in _graph.query(
        "MATCH (a:Memory {group:$grp})-[l:LINK]->(b:Memory {group:$grp}) "
        "RETURN a.id, b.id, l.type", {"grp": grp}).result_set]
    mentions = [{"memory": r[0], "entity": r[1]} for r in _graph.query(
        "MATCH (m:Memory {group:$grp})-[:MENTIONS]->(e:Entity {group:$grp}) "
        "RETURN m.id, e.key", {"grp": grp}).result_set]
    relations = [{"from": r[0], "to": r[1], "rtype": r[2], "created_at": r[3]}
                 for r in _graph.query(
        "MATCH (a:Entity {group:$grp})-[r:REL]->(b:Entity {group:$grp}) "
        "RETURN a.key, b.key, r.rtype, r.created_at", {"grp": grp}).result_set]
    insights = [{"id": r[0], "text": r[1], "title": r[2], "itype": r[3],
                 "created_at": r[4], "surfaced": r[5]} for r in _graph.query(
        "MATCH (i:Insight {group:$grp}) RETURN i.id, i.text, i.title, i.itype, "
        "i.created_at, i.surfaced", {"grp": grp}).result_set]
    tasks = [{"id": r[0], "text": r[1], "due": r[2], "done": r[3],
              "created_at": r[4], "done_at": r[5]} for r in _graph.query(
        "MATCH (t:Task {grp:$grp}) RETURN t.id, t.text, t.due, t.done, "
        "t.created_at, t.done_at", {"grp": grp}).result_set]
    profile = [{"id": r[0], "text": r[1], "updated_at": r[2]}
               for r in _graph.query(
        "MATCH (p:Profile {group:$grp}) RETURN p.id, p.text, p.updated_at",
        {"grp": grp}).result_set]
    core = [{"text": r[0], "updated_at": r[1]} for r in _graph.query(
        "MATCH (c:Core {group:$grp}) RETURN c.text, c.updated_at",
        {"grp": grp}).result_set]
    meta = [{"id": r[0], "last_run": r[1], "status": r[2], "note": r[3]}
            for r in _graph.query(
        "MATCH (x:Meta {group:$grp}) RETURN x.id, x.last_run, x.status, x.note",
        {"grp": grp}).result_set]
    return {"group": grp, "exported_at": _now_iso(), "memories": memories,
            "entities": entities, "links": links, "mentions": mentions,
            "relations": relations, "insights": insights, "tasks": tasks,
            "profile": profile, "core": core, "meta": meta}


class ReflectionRan(BaseModel):
    status: str | None = "ok"
    note: str | None = ""
    group: str | None = None


@app.post("/reflection_ran")
def reflection_ran(body: ReflectionRan,
                   authorization: str | None = Header(default=None)):
    """The reflection engine calls this after each run so /health can report when
    it last executed (stored on a :Meta {id:'reflection'} singleton per group)."""
    _auth(authorization)
    grp = body.group or GROUP_DEFAULT
    now = _now_iso()
    _graph.query(
        "MERGE (x:Meta {id:'reflection', group:$grp}) "
        "SET x.last_run=$now, x.status=$st, x.note=$note",
        {"grp": grp, "now": now, "st": body.status or "ok",
         "note": body.note or ""})
    return {"ok": True, "last_run": now}


class Core(BaseModel):
    text: str
    group: str | None = None


@app.get("/core_block")
def core_block(group: str | None = None,
               authorization: str | None = Header(default=None)):
    """Ahmed's always-on core context block (a :Core singleton per group) — the
    identity/preamble every brain loads before recalling anything else."""
    _auth(authorization)
    grp = group or GROUP_DEFAULT
    rows = _graph.query(
        "MATCH (c:Core {id:'core', group:$grp}) RETURN c.text, c.updated_at",
        {"grp": grp}).result_set
    if not rows:
        return {"text": "", "updated_at": ""}
    return {"text": rows[0][0] or "", "updated_at": rows[0][1] or ""}


@app.post("/core")
def core_set(body: Core, authorization: str | None = Header(default=None)):
    """Upsert Ahmed's core context block (singleton)."""
    _auth(authorization)
    grp = body.group or GROUP_DEFAULT
    _graph.query(
        "MERGE (c:Core {id:'core', group:$grp}) "
        "SET c.text=$text, c.updated_at=$now",
        {"grp": grp, "text": body.text or "", "now": _now_iso()})
    return {"ok": True}


# ---------------------------------------------------------------------------
# Tasks & reminders — stored as :Task nodes so they SYNC across every device.
# A reminder is just a task with a `due` time (ISO). `done` toggles completion.
# ---------------------------------------------------------------------------
class Task(BaseModel):
    text: str
    due: str | None = None            # ISO time for a reminder; null = plain task
    group: str | None = None
    source: str | None = "jarvis"


class TaskId(BaseModel):
    id: str
    group: str | None = None


@app.post("/task")
def task_add(body: Task, authorization: str | None = Header(default=None)):
    _auth(authorization)
    if not body.text.strip():
        raise HTTPException(status_code=400, detail="empty task")
    tid = uuid.uuid4().hex
    _graph.query(
        "CREATE (t:Task {id:$id, text:$text, due:$due, done:false, "
        "source:$src, grp:$grp, created_at:$now, done_at:''})",
        {"id": tid, "text": body.text.strip(), "due": body.due or "",
         "src": body.source or "jarvis", "grp": body.group or GROUP_DEFAULT,
         "now": _now_iso()})
    return {"id": tid}


@app.get("/tasks")
def task_list(group: str | None = None, include_done: bool = False,
              authorization: str | None = Header(default=None)):
    _auth(authorization)
    grp = group or GROUP_DEFAULT
    rows = _graph.query(
        "MATCH (t:Task) WHERE t.grp=$grp "
        + ("" if include_done else "AND t.done=false ") +
        "RETURN t.id, t.text, t.due, t.done, t.created_at, t.done_at "
        "ORDER BY t.done, t.due, t.created_at",
        {"grp": grp}).result_set
    return {"tasks": [{"id": r[0], "text": r[1], "due": r[2] or None,
                       "done": bool(r[3]), "created_at": r[4],
                       "done_at": r[5] or None} for r in rows]}


@app.post("/task/done")
def task_done(body: TaskId, authorization: str | None = Header(default=None)):
    _auth(authorization)
    _graph.query("MATCH (t:Task {id:$id}) SET t.done=true, t.done_at=$now",
                 {"id": body.id, "now": _now_iso()})
    return {"ok": True}


@app.post("/task/reopen")
def task_reopen(body: TaskId, authorization: str | None = Header(default=None)):
    _auth(authorization)
    _graph.query("MATCH (t:Task {id:$id}) SET t.done=false, t.done_at=''",
                 {"id": body.id})
    return {"ok": True}


@app.post("/task/delete")
def task_delete(body: TaskId, authorization: str | None = Header(default=None)):
    _auth(authorization)
    _graph.query("MATCH (t:Task {id:$id}) DELETE t", {"id": body.id})
    return {"ok": True}


@app.get("/reminders/due")
def reminders_due(group: str | None = None,
                  authorization: str | None = Header(default=None)):
    """Tasks whose reminder time has passed and aren't done/fired — the
    scheduler polls this to know what to announce."""
    _auth(authorization)
    grp = group or GROUP_DEFAULT
    now = _now_iso()
    rows = _graph.query(
        "MATCH (t:Task) WHERE t.grp=$grp AND t.done=false AND t.due<>'' "
        "AND t.due<=$now AND coalesce(t.fired,false)=false "
        "RETURN t.id, t.text, t.due", {"grp": grp, "now": now}).result_set
    # mark fired so the scheduler announces each due reminder only once
    for r in rows:
        _graph.query("MATCH (t:Task {id:$id}) SET t.fired=true", {"id": r[0]})
    return {"due": [{"id": r[0], "text": r[1], "due": r[2]} for r in rows]}