"""Chunk-level knowledge graph: embeddings, similarity edges, subgraph export."""
from __future__ import annotations

import hashlib
import heapq
import logging
import os
import re
import uuid
from typing import Any

_log = logging.getLogger(__name__)

_CHUNK_NODE_RE = re.compile(r"^knowledge:([a-z0-9][a-z0-9_-]{0,63}):chunk:(\d+)$")
_SIMILARITY_MIN = 0.72
_AUTO_ACTIVE_THRESHOLD = 0.86
_NEIGHBORS_PER_CHUNK = 8
_PROPOSE_PAIR_CAP = 8000
_MAX_NODES_DEFAULT = 120
_ENTITY_SUBGRAPH_MAX_SEEDS = 24
_SAME_SOURCE_PENALTY = 0.05
_ENTITY_FOCUS_RE = re.compile(r"^knowledge:([a-z0-9][a-z0-9_-]{0,63}):entity:(.+)$")


def _chunk_node_id(corpus_slug: str, row_id: int) -> str:
    return f"knowledge:{corpus_slug}:chunk:{int(row_id)}"


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


def _effective_similarity(
    sim: float,
    *,
    same_source: bool,
) -> float:
    if same_source and sim < _AUTO_ACTIVE_THRESHOLD:
        return sim - _SAME_SOURCE_PENALTY
    return sim


def _similarity_status(sim: float) -> str:
    if sim >= _AUTO_ACTIVE_THRESHOLD:
        return "active"
    return "suggested"


def _load_rejected_tombstones(conn: Any, corpus_slug: str) -> set[tuple[str, str, str]]:
    rows = conn.execute(
        """SELECT node_low, node_high, link_kind FROM knowledge_graph_edges
           WHERE corpus_slug = ? AND status = 'rejected'""",
        (corpus_slug,),
    ).fetchall()
    return {(str(low), str(high), str(lk)) for low, high, lk in rows}


def _collect_similarity_candidates(
    nodes: list[dict[str, Any]],
    emap: dict[str, list[float]],
    *,
    cosine_similarity: Any,
) -> tuple[dict[tuple[str, str], float], int, bool]:
    """Per-chunk top-K neighbor picks, unioned and deduped by undirected pair."""
    node_meta = {str(n["node_id"]): n for n in nodes}
    ids = [str(n["node_id"]) for n in nodes if emap.get(str(n["node_id"]))]
    per_node_heaps: dict[str, list[tuple[float, str]]] = {}
    pairs_scanned = 0
    cap_hit = False

    for a in ids:
        va = emap.get(a) or []
        if not va:
            continue
        src_a = str(node_meta[a].get("source_slug") or "")
        heap: list[tuple[float, str]] = []
        for b in ids:
            if b == a:
                continue
            pairs_scanned += 1
            if pairs_scanned > _PROPOSE_PAIR_CAP:
                cap_hit = True
                break
            vb = emap.get(b) or []
            if not vb:
                continue
            sim = float(cosine_similarity(va, vb))
            if sim < _SIMILARITY_MIN:
                continue
            src_b = str(node_meta[b].get("source_slug") or "")
            rank_sim = _effective_similarity(sim, same_source=(src_a == src_b and src_a != ""))
            if len(heap) < _NEIGHBORS_PER_CHUNK:
                heapq.heappush(heap, (rank_sim, b))
            elif rank_sim > heap[0][0]:
                heapq.heapreplace(heap, (rank_sim, b))
        if heap:
            per_node_heaps[a] = heap
        if cap_hit:
            break

    pair_best: dict[tuple[str, str], float] = {}
    for a, heap in per_node_heaps.items():
        va = emap.get(a) or []
        for _rank, b in heap:
            vb = emap.get(b) or []
            if not va or not vb:
                continue
            sim = float(cosine_similarity(va, vb))
            if sim < _SIMILARITY_MIN:
                continue
            low, high = _canonical_pair(a, b)
            pair_best[(low, high)] = max(pair_best.get((low, high), 0.0), sim)
    return pair_best, pairs_scanned, cap_hit


def collect_corpus_chunk_nodes(
    slug: str,
    label: str,
    synthesis_text: str,
    sources: list[dict[str, Any]],
    raw_texts: dict[str, str] | None = None,
    *,
    ingest_profile: str = "docs",
) -> list[dict[str, Any]]:
    """Build chunk node payloads mirroring ``_index_knowledge_in_search_db`` row layout."""
    del ingest_profile
    from tools.knowledge_tools import _build_knowledge_corpus_index

    _rows, _meta, nodes = _build_knowledge_corpus_index(
        slug, label, synthesis_text, sources, raw_texts
    )
    return nodes


def _purge_active_graph_edges(
    conn: Any,
    corpus_slug: str,
    *,
    preserve_rejected: bool,
) -> None:
    slug = str(corpus_slug or "").strip()
    if not slug:
        return
    if preserve_rejected:
        conn.execute(
            """DELETE FROM knowledge_graph_edges
               WHERE corpus_slug = ? AND status IN ('active', 'suggested')""",
            (slug,),
        )
    else:
        conn.execute("DELETE FROM knowledge_graph_edges WHERE corpus_slug = ?", (slug,))


def _clear_corpus_graph_embeddings_and_edges(slug: str) -> int:
    """Delete all corpus embeddings and active/suggested edges; return rows removed."""
    from chat_storage import _ensure_knowledge_graph_tables, _get_connection

    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        cur = conn.execute(
            "DELETE FROM knowledge_chunk_embeddings WHERE corpus_slug = ?",
            (slug,),
        )
        orphans_deleted = max(0, int(cur.rowcount or 0))
        _purge_active_graph_edges(conn, slug, preserve_rejected=True)
        conn.commit()
        return orphans_deleted
    finally:
        conn.close()


def purge_knowledge_graph_for_corpus(
    corpus_slug: str,
    *,
    preserve_rejected: bool = True,
    purge_embeddings: bool = True,
) -> None:
    from chat_storage import _get_connection, _ensure_knowledge_graph_tables

    slug = str(corpus_slug or "").strip()
    if not slug:
        return
    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        if purge_embeddings:
            conn.execute(
                "DELETE FROM knowledge_chunk_embeddings WHERE corpus_slug = ?",
                (slug,),
            )
        _purge_active_graph_edges(conn, slug, preserve_rejected=preserve_rejected)
        conn.commit()
    finally:
        conn.close()


def _node_content_hash(node: dict[str, Any]) -> str:
    """Match session_search_db vector_chunks.content_hash (full indexed body, errors=ignore)."""
    stored = node.get("content_hash")
    if stored:
        return str(stored)
    return hashlib.sha256(_node_embed_text(node).encode("utf-8", errors="ignore")).hexdigest()


def _node_embed_text(node: dict[str, Any]) -> str:
    """Text passed to the embedder (full indexed body when built by knowledge_tools)."""
    return str(node.get("embed_text") or node.get("text") or "")


def _existing_embedding_fresh(
    row: tuple[Any, ...],
    *,
    content_hash: str,
    embed_identity: str,
    chunker_version: str,
    model_id: str,
    dim: int,
) -> bool:
    """True when a stored graph row needs no re-embed (edges-only path)."""
    if len(row) < 7:
        return False
    _nid, stored_hash, stored_model, stored_identity, stored_chunker, stored_dim, _blob = row[:7]
    try:
        stored_dim_i = int(stored_dim or 0)
    except (TypeError, ValueError):
        stored_dim_i = 0
    return (
        str(stored_hash or "") == content_hash
        and str(stored_model or "") == model_id
        and str(stored_identity or "") == embed_identity
        and str(stored_chunker or "") == chunker_version
        and stored_dim_i == dim
    )


def _load_corpus_embedding_rows(conn: Any, slug: str) -> dict[str, tuple[Any, ...]]:
    rows = conn.execute(
        """SELECT node_id, content_hash, model_id, embed_identity, chunker_version, dim, embedding
           FROM knowledge_chunk_embeddings WHERE corpus_slug = ?""",
        (slug,),
    ).fetchall()
    return {str(r[0]): r for r in rows}


def _delete_corpus_embedding_node_ids(
    conn: Any,
    slug: str,
    node_ids: list[str],
    *,
    batch_size: int = 400,
) -> int:
    """Batch-delete graph embedding rows; returns rows removed."""
    deleted = 0
    for i in range(0, len(node_ids), batch_size):
        batch = [nid for nid in node_ids[i : i + batch_size] if nid]
        if not batch:
            continue
        placeholders = ",".join("?" * len(batch))
        cur = conn.execute(
            f"""DELETE FROM knowledge_chunk_embeddings
                WHERE corpus_slug = ? AND node_id IN ({placeholders})""",
            (slug, *batch),
        )
        deleted += max(0, int(cur.rowcount or 0))
    return deleted


def _insert_chunk_embedding_row(
    conn: Any,
    *,
    node: dict[str, Any],
    slug: str,
    content_hash: str,
    embed_model_id: str,
    embed_identity: str,
    chunker_version: str,
    dim: int,
    vec: list[float],
    now: str,
    encode_vec: Any,
) -> None:
    nid = str(node["node_id"])
    conn.execute(
        """INSERT OR REPLACE INTO knowledge_chunk_embeddings
           (node_id, corpus_slug, row_id, source_slug, filename, label, content_hash,
            rel_path, repo_root, model_id, embed_identity, chunker_version, dim, embedding, updated_at)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        (
            nid,
            slug,
            int(node.get("row_id") or 0),
            str(node.get("source_slug") or ""),
            str(node.get("filename") or ""),
            str(node.get("label") or ""),
            content_hash,
            str(node.get("rel_path") or ""),
            str(node.get("repo_root") or ""),
            embed_model_id,
            embed_identity,
            chunker_version,
            dim,
            encode_vec(vec),
            now,
        ),
    )


def rebuild_knowledge_graph_for_corpus(
    slug: str,
    label: str,
    synthesis_text: str,
    sources: list[dict[str, Any]],
    raw_texts: dict[str, str] | None = None,
) -> dict[str, Any]:
    """Reconcile chunk embeddings (reuse search vectors when fresh) and rebuild edges."""
    try:
        materialize_concept_graph_edges(slug)
    except Exception as exc:
        _log.warning("[knowledge_graph] concept layer failed for %s: %s", slug, exc)
    import session_search_db as ss
    from chat_storage import (
        _cosine_similarity_vault,
        _decode_vault_embedding_vec,
        _encode_vault_embedding_vec,
        _ensure_knowledge_graph_tables,
        _get_connection,
        _looks_secret,
        _now_iso,
        _vault_embed_dim,
        _vault_embed_model_id,
    )
    from tools.knowledge_tools import _load_manifest, normalize_ingest_profile
    from utility_model import utility_embed_texts, utility_embedding_identity

    from utils.embedding_chunking import CHUNKER_VERSION

    manifest = _load_manifest(slug) or {}
    profile = normalize_ingest_profile(
        str(manifest.get("ingest_profile") or "docs"),
        source_path=str(manifest.get("source_path") or ""),
    )
    nodes = collect_corpus_chunk_nodes(
        slug, label, synthesis_text, sources, raw_texts, ingest_profile=profile
    )
    empty_stats = {
        "embedded": 0,
        "edges": 0,
        "edges_active": 0,
        "edges_suggested": 0,
        "edges_sequence": 0,
        "pairs_scanned": 0,
        "cap_hit": False,
        "reused": 0,
        "kept": 0,
        "newly_embedded": 0,
        "orphans_deleted": 0,
    }
    if not nodes:
        orphans_deleted = _clear_corpus_graph_embeddings_and_edges(slug)
        return {"ok": True, **empty_stats, "orphans_deleted": orphans_deleted}

    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        rejected_tombstones = _load_rejected_tombstones(conn, slug)
        existing_rows = _load_corpus_embedding_rows(conn, slug)
    finally:
        conn.close()

    try:
        nodes = [n for n in nodes if not _looks_secret(_node_embed_text(n))]
    except Exception:
        pass
    if not nodes:
        orphans_deleted = _clear_corpus_graph_embeddings_and_edges(slug)
        return {"ok": True, **empty_stats, "orphans_deleted": orphans_deleted}

    thread_id = f"knowledge:{slug}"
    search_vectors = ss.get_thread_chunk_vectors(thread_id)
    embed_identity = utility_embedding_identity()
    chunker_version = CHUNKER_VERSION()
    embed_model_id = _vault_embed_model_id()
    dim = _vault_embed_dim()
    index_dim = ss._vector_dim()
    reuse_dim_ok = index_dim == dim
    if not reuse_dim_ok:
        _log.debug(
            "[knowledge_graph] %s search/graph dim mismatch (%d vs %d); skipping vector reuse",
            slug,
            index_dim,
            dim,
        )

    vec_by_node: dict[str, list[float]] = {}
    kept = 0
    reused = 0
    residual_nodes: list[dict[str, Any]] = []
    reuse_writes: list[tuple[dict[str, Any], str, list[float]]] = []

    for node in nodes:
        nid = str(node["node_id"])
        digest = _node_content_hash(node)
        row = existing_rows.get(nid)
        if row and _existing_embedding_fresh(
            row,
            content_hash=digest,
            embed_identity=embed_identity,
            chunker_version=chunker_version,
            model_id=embed_model_id,
            dim=dim,
        ):
            blob = row[6] if len(row) > 6 else b""
            vec = _decode_vault_embedding_vec(bytes(blob or b""))
            if vec and len(vec) == dim:
                vec_by_node[nid] = vec
                kept += 1
                continue
        row_id = int(node.get("row_id") or 0)
        sv = search_vectors.get(row_id) if reuse_dim_ok else None
        if sv:
            search_hash, search_vec = sv
            if search_hash == digest and len(search_vec) == dim:
                vec_by_node[nid] = list(search_vec)
                reuse_writes.append((node, digest, list(search_vec)))
                reused += 1
                continue
        residual_nodes.append(node)

    newly_embedded = 0
    residual_error = ""
    if residual_nodes:
        texts = [_node_embed_text(n) for n in residual_nodes]
        try:
            vecs = utility_embed_texts(
                texts,
                dimensions=dim,
                activity_operation="knowledge_graph_embedding",
            )
        except Exception as exc:
            _log.warning("[knowledge_graph] embedding unavailable for %s: %s", slug, exc)
            residual_error = str(exc)[:200]
            vecs = []
        else:
            try:
                from utils.embedding_model import is_real
            except Exception:
                is_real = lambda: False  # type: ignore[misc, assignment]
            if not vecs or len(vecs) != len(residual_nodes):
                if is_real():
                    _log.warning(
                        "[knowledge_graph] MLX forward failed for %s; skipping %d residual chunks",
                        slug,
                        len(residual_nodes),
                    )
                    vecs = []
                else:
                    residual_error = "embedding_count_mismatch"
                    vecs = []
        for node, vec in zip(residual_nodes, vecs or []):
            if not vec or len(vec) != dim:
                continue
            nid = str(node["node_id"])
            vec_by_node[nid] = list(vec)
            newly_embedded += 1
        skipped_residual = len(residual_nodes) - newly_embedded
        if skipped_residual and not residual_error:
            _log.warning(
                "[knowledge_graph] %s skipped %d residual chunks (dim/embed mismatch)",
                slug,
                skipped_residual,
            )
        if residual_nodes and newly_embedded == 0 and not residual_error:
            try:
                from utils.embedding_model import is_real as _embed_is_real

                if _embed_is_real():
                    residual_error = "embedding_forward_failed"
                else:
                    residual_error = "embedding_residual_failed"
            except Exception:
                residual_error = "embedding_residual_failed"

    if not vec_by_node:
        return {
            "ok": False,
            "error": residual_error or "no_embeddings_available",
            **empty_stats,
        }

    pair_best, pairs_scanned, cap_hit = _collect_similarity_candidates(
        nodes, vec_by_node, cosine_similarity=_cosine_similarity_vault
    )

    now = _now_iso()
    conn = _get_connection()
    edges_active = 0
    edges_suggested = 0
    edges_sequence = 0
    orphans_deleted = 0
    try:
        _ensure_knowledge_graph_tables(conn)
        _purge_active_graph_edges(conn, slug, preserve_rejected=True)
        for node, digest, vec in reuse_writes:
            _insert_chunk_embedding_row(
                conn,
                node=node,
                slug=slug,
                content_hash=digest,
                embed_model_id=embed_model_id,
                embed_identity=embed_identity,
                chunker_version=chunker_version,
                dim=dim,
                vec=vec,
                now=now,
                encode_vec=_encode_vault_embedding_vec,
            )
        residual_by_id = {str(n["node_id"]): n for n in residual_nodes}
        for nid, vec in vec_by_node.items():
            if nid not in residual_by_id:
                continue
            node = residual_by_id[nid]
            _insert_chunk_embedding_row(
                conn,
                node=node,
                slug=slug,
                content_hash=_node_content_hash(node),
                embed_model_id=embed_model_id,
                embed_identity=embed_identity,
                chunker_version=chunker_version,
                dim=dim,
                vec=vec,
                now=now,
                encode_vec=_encode_vault_embedding_vec,
            )

        current_ids = {str(n["node_id"]) for n in nodes}
        if current_ids:
            stale_rows = conn.execute(
                "SELECT node_id FROM knowledge_chunk_embeddings WHERE corpus_slug = ?",
                (slug,),
            ).fetchall()
            orphan_ids = [
                str(r[0])
                for r in stale_rows
                if str(r[0]) and str(r[0]) not in current_ids
            ]
            orphans_deleted += _delete_corpus_embedding_node_ids(conn, slug, orphan_ids)
            unembedded_ids = [nid for nid in current_ids if nid not in vec_by_node]
            orphans_deleted += _delete_corpus_embedding_node_ids(conn, slug, unembedded_ids)
        else:
            cur = conn.execute(
                "DELETE FROM knowledge_chunk_embeddings WHERE corpus_slug = ?",
                (slug,),
            )
            orphans_deleted = max(0, int(cur.rowcount or 0))

        id_to_node = {str(n["node_id"]): n for n in nodes}
        for (low, high), sim in pair_best.items():
            if (low, high, "chunk_similarity") in rejected_tombstones:
                continue
            a, b = (low, high) if low in id_to_node else (high, low)
            if a not in id_to_node or b not in id_to_node:
                a, b = low, high
            status = _similarity_status(sim)
            eid = str(uuid.uuid4())
            before = conn.total_changes
            conn.execute(
                """INSERT OR IGNORE INTO knowledge_graph_edges
                   (id, node_low, node_high, src_id, dst_id, link_kind, status, confidence,
                    corpus_slug, rationale, created_at, updated_at)
                   VALUES (?, ?, ?, ?, ?, 'chunk_similarity', ?, ?, ?, '', ?, ?)""",
                (eid, low, high, a, b, status, float(sim), slug, now, now),
            )
            if conn.total_changes > before:
                if status == "active":
                    edges_active += 1
                else:
                    edges_suggested += 1

        for i in range(len(nodes) - 1):
            a, b = nodes[i], nodes[i + 1]
            if str(a.get("source_slug")) != str(b.get("source_slug")):
                continue
            if int(b.get("row_id") or 0) - int(a.get("row_id") or 0) != 1:
                continue
            aid, bid = str(a["node_id"]), str(b["node_id"])
            if aid not in vec_by_node or bid not in vec_by_node:
                continue
            low, high = _canonical_pair(aid, bid)
            eid = str(uuid.uuid4())
            before = conn.total_changes
            conn.execute(
                """INSERT OR IGNORE INTO knowledge_graph_edges
                   (id, node_low, node_high, src_id, dst_id, link_kind, status, confidence,
                    corpus_slug, rationale, created_at, updated_at)
                   VALUES (?, ?, ?, ?, ?, 'chunk_sequence', 'active', 1.0, ?, '', ?, ?)""",
                (eid, low, high, aid, bid, slug, now, now),
            )
            if conn.total_changes > before:
                edges_sequence += 1

        conn.commit()
        total_edges = edges_active + edges_suggested + edges_sequence
        embedded = len(vec_by_node)
        _log.info(
            "[knowledge_graph] %s reconcile embedded=%d kept=%d reused=%d newly_embedded=%d orphans_deleted=%d",
            slug,
            embedded,
            kept,
            reused,
            newly_embedded,
            orphans_deleted,
        )
        result: dict[str, Any] = {
            "ok": True,
            "embedded": embedded,
            "edges": total_edges,
            "edges_active": edges_active,
            "edges_suggested": edges_suggested,
            "edges_sequence": edges_sequence,
            "pairs_scanned": pairs_scanned,
            "cap_hit": cap_hit,
            "reused": reused,
            "kept": kept,
            "newly_embedded": newly_embedded,
            "orphans_deleted": orphans_deleted,
        }
        if residual_error:
            result["ok"] = False
            result["error"] = residual_error
        if embedded < len(nodes):
            result["ok"] = False
            result.setdefault("error", residual_error or "partial_graph_embed")
            _log.warning(
                "[knowledge_graph] %s partial reconcile: %d/%d nodes embedded (%s)",
                slug,
                embedded,
                len(nodes),
                result.get("error"),
            )
        return result
    finally:
        conn.close()


def _enrich_edges_with_labels(
    conn: Any,
    edge_rows: list[tuple[Any, ...]],
    *,
    focus_node_id: str | None = None,
) -> list[dict[str, Any]]:
    """Join chunk embeddings for human-readable suggestion/neighbor payloads."""
    out: list[dict[str, Any]] = []
    focus = str(focus_node_id or "").strip()
    for row in edge_rows:
        if len(row) >= 8:
            rid, sid, did, lk, conf, corp, rat, created = row[:8]
        else:
            rid, sid, did, lk, conf = row[0], row[1], row[2], row[3], row[4]
            corp, rat, created = row[5] if len(row) > 5 else "", row[6] if len(row) > 6 else "", row[7] if len(row) > 7 else ""
        sid_s, did_s = str(sid), str(did)
        src_row = conn.execute(
            "SELECT label, filename FROM knowledge_chunk_embeddings WHERE node_id = ?",
            (sid_s,),
        ).fetchone()
        dst_row = conn.execute(
            "SELECT label, filename FROM knowledge_chunk_embeddings WHERE node_id = ?",
            (did_s,),
        ).fetchone()
        if src_row:
            src_label = graph_chunk_display_title(
                str(src_row[0] or src_row[1] or sid_s),
                filename=str(src_row[1] or ""),
            )
        else:
            src_label = sid_s
        if dst_row:
            dst_label = graph_chunk_display_title(
                str(dst_row[0] or dst_row[1] or did_s),
                filename=str(dst_row[1] or ""),
            )
        else:
            dst_label = did_s
        other_node_id = ""
        if focus:
            other_node_id = did_s if sid_s == focus else sid_s if did_s == focus else ""
        else:
            other_node_id = did_s
        payload: dict[str, Any] = {
            "id": str(rid),
            "src_id": sid_s,
            "dst_id": did_s,
            "src_label": src_label,
            "dst_label": dst_label,
            "src_filename": str(src_row[1] or "") if src_row else "",
            "dst_filename": str(dst_row[1] or "") if dst_row else "",
            "link_kind": str(lk),
            "confidence": float(conf) if conf is not None else None,
            "corpus_slug": str(corp or ""),
            "rationale": str(rat or ""),
            "created_at": str(created or ""),
        }
        if other_node_id:
            payload["other_node_id"] = other_node_id
        out.append(payload)
    return out


def validate_focus_node_for_corpus(focus_node_id: str, corpus_slug: str) -> bool:
    focus = str(focus_node_id or "").strip()
    corpus = str(corpus_slug or "").strip()
    if not focus or not corpus:
        return False
    if not _CHUNK_NODE_RE.match(focus):
        return False
    m = _CHUNK_NODE_RE.match(focus)
    if not m or m.group(1) != corpus:
        return False
    from chat_storage import _ensure_knowledge_graph_tables, _get_connection

    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        row = conn.execute(
            "SELECT corpus_slug FROM knowledge_chunk_embeddings WHERE node_id = ?",
            (focus,),
        ).fetchone()
        if not row:
            # Canonical node id for this corpus before first embed/rebuild.
            return True
        return str(row[0]) == corpus
    finally:
        conn.close()


def knowledge_graph_stats(*, corpus_slug: str) -> dict[str, Any]:
    from chat_storage import _ensure_knowledge_graph_tables, _get_connection

    slug = str(corpus_slug or "").strip()
    if not slug:
        return {"ok": False, "error": "corpus required"}
    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        embedded = conn.execute(
            "SELECT COUNT(*) FROM knowledge_chunk_embeddings WHERE corpus_slug = ?",
            (slug,),
        ).fetchone()[0]
        active = conn.execute(
            """SELECT COUNT(*) FROM knowledge_graph_edges
               WHERE corpus_slug = ? AND status = 'active' AND link_kind = 'chunk_similarity'""",
            (slug,),
        ).fetchone()[0]
        suggested = conn.execute(
            "SELECT COUNT(*) FROM knowledge_graph_edges WHERE corpus_slug = ? AND status = 'suggested'",
            (slug,),
        ).fetchone()[0]
        rejected = conn.execute(
            "SELECT COUNT(*) FROM knowledge_graph_edges WHERE corpus_slug = ? AND status = 'rejected'",
            (slug,),
        ).fetchone()[0]
        sequence_active = conn.execute(
            """SELECT COUNT(*) FROM knowledge_graph_edges
               WHERE corpus_slug = ? AND status = 'active' AND link_kind = 'chunk_sequence'""",
            (slug,),
        ).fetchone()[0]
        concept_kinds = ",".join(f"'{k}'" for k in _CONCEPT_LINK_KINDS)
        concept_active = conn.execute(
            f"""SELECT COUNT(*) FROM knowledge_graph_edges
               WHERE corpus_slug = ? AND status = 'active' AND link_kind IN ({concept_kinds})""",
            (slug,),
        ).fetchone()[0]
        return {
            "ok": True,
            "corpus_slug": slug,
            "embedded_chunks": int(embedded or 0),
            "active": int(active or 0),
            "suggested": int(suggested or 0),
            "rejected": int(rejected or 0),
            "sequence_active": int(sequence_active or 0),
            "concept_active": int(concept_active or 0),
        }
    finally:
        conn.close()


_CONCEPT_LINK_KINDS = frozenset({"topic_source", "synthesis_link", "topic_related"})
_CHUNK_LINK_KINDS = frozenset({"chunk_similarity", "chunk_sequence"})


def _topic_node_id(slug: str, term: str) -> str:
    norm = re.sub(r"[^a-z0-9]+", "_", (term or "").strip().lower())[:48].strip("_")
    return f"knowledge:{slug}:topic:{norm or 'general'}"


def _source_node_id(slug: str, source_slug: str) -> str:
    safe = re.sub(r"[^a-z0-9_-]+", "_", (source_slug or "").strip().lower())[:48].strip("_")
    return f"knowledge:{slug}:source:{safe or 'unknown'}"


def _concept_node_label(node_id: str) -> str:
    nid = str(node_id or "")
    if ":topic:" in nid:
        return nid.split(":topic:", 1)[-1].replace("_", " ")
    if ":source:" in nid:
        return nid.split(":source:", 1)[-1].replace("_", " ")
    return nid


_CHUNK_LABEL_RE = re.compile(
    r"^(?P<prefix>.+?)\s+§chunk\s+(?P<idx>\d+)\s*$",
    re.IGNORECASE,
)
_SYNTH_CHUNK_LABEL_RE = re.compile(
    r"^(?P<corpus>.+?)\s*>\s*synthesis\s+§chunk\s+(?P<idx>\d+)\s*$",
    re.IGNORECASE,
)


def _shorten_graph_label(text: str, *, max_len: int = 42) -> str:
    s = " ".join(str(text or "").split())
    if len(s) <= max_len:
        return s
    if max_len < 4:
        return s[:max_len]
    head = max(8, (max_len - 1) // 2)
    tail = max_len - head - 1
    return s[:head] + "…" + s[-tail:]


def graph_chunk_display_title(label: str, *, filename: str = "", max_len: int = 42) -> str:
    """Human-facing graph label from internal ``{file} §chunk {n}`` storage."""
    raw = str(label or "").strip()
    if not raw:
        return _shorten_graph_label(filename or "Passage", max_len=max_len)
    m = _SYNTH_CHUNK_LABEL_RE.match(raw)
    if m:
        return f"Synthesis · part {m.group('idx')}"
    m = _CHUNK_LABEL_RE.match(raw)
    if m:
        prefix = m.group("prefix").strip()
        idx = m.group("idx")
        base = os.path.basename(prefix.replace("\\", "/")) if prefix else ""
        if not base and filename:
            base = os.path.basename(str(filename).replace("\\", "/"))
        short = _shorten_graph_label(base or prefix, max_len=max(18, max_len - 10))
        return f"{short} · part {idx}"
    return _shorten_graph_label(raw, max_len=max_len)


_PASSAGE_MAX_CHARS = 12_000


def _strip_knowledge_index_header(text: str) -> str:
    lines = (text or "").splitlines()
    if lines and lines[0].startswith("[") and "]" in lines[0]:
        body = "\n".join(lines[1:]).strip()
        return body or lines[0].strip()
    return (text or "").strip()


def _read_indexed_knowledge_message(thread_id: str, message_row_id: int) -> str:
    """Full indexed passage for one knowledge FTS row (1-based ``message_row_id``)."""
    from session_search_db import _conn_lock, _content_to_text, _safe_get_connection

    tid = (thread_id or "").strip()
    rid = int(message_row_id)
    if not tid or rid <= 0:
        return ""
    with _conn_lock:
        conn = _safe_get_connection()
        if conn is None:
            return ""
        try:
            row = conn.execute(
                """
                SELECT content FROM messages_fts
                WHERE thread_id = ?
                ORDER BY rowid ASC
                LIMIT 1 OFFSET ?
                """,
                (tid, rid - 1),
            ).fetchone()
            if row and row[0] is not None:
                return str(_content_to_text(row[0]) or "").strip()
            previews = conn.execute(
                """
                SELECT text_preview FROM vector_chunks
                WHERE thread_id = ? AND message_row_id = ?
                ORDER BY chunk_index ASC
                """,
                (tid, rid),
            ).fetchall()
            if previews:
                return "\n".join(
                    str(p[0] or "").strip() for p in previews if str(p[0] or "").strip()
                ).strip()
        except Exception as exc:
            _log.debug(
                "[knowledge_graph] indexed message read failed %s row %s: %s",
                tid,
                rid,
                exc,
            )
        finally:
            conn.close()
    return ""


def get_knowledge_graph_node_passage(node_id: str) -> dict[str, Any]:
    """Passage text for graph node selection (chunk nodes return indexed excerpt)."""
    nid = str(node_id or "").strip()
    if not nid:
        return {"ok": False, "error": "node_id required", "node_id": "", "title": "", "text": ""}
    if ":chunk:" in nid:
        match = _CHUNK_NODE_RE.match(nid)
        if not match:
            return {"ok": False, "error": "invalid chunk node", "node_id": nid, "title": "", "text": ""}
        slug, row_s = match.group(1), match.group(2)
        try:
            row_id = int(row_s)
        except ValueError:
            return {"ok": False, "error": "invalid chunk node", "node_id": nid, "title": "", "text": ""}
        from chat_storage import _ensure_knowledge_graph_tables, _get_connection

        conn = _get_connection()
        try:
            _ensure_knowledge_graph_tables(conn)
            meta = conn.execute(
                """SELECT label, filename, source_slug FROM knowledge_chunk_embeddings
                   WHERE node_id = ?""",
                (nid,),
            ).fetchone()
        finally:
            conn.close()
        raw_label = str(meta[0] or meta[1] or nid) if meta else nid
        filename = str(meta[1] or "") if meta else ""
        source_slug = str(meta[2] or "") if meta else ""
        title = graph_chunk_display_title(raw_label, filename=filename)
        raw_text = _read_indexed_knowledge_message(f"knowledge:{slug}", row_id)
        text = _strip_knowledge_index_header(raw_text)[:_PASSAGE_MAX_CHARS]
        if not text:
            return {
                "ok": False,
                "error": "passage not found",
                "node_id": nid,
                "node_type": "chunk",
                "title": title,
                "text": "",
                "corpus_slug": slug,
                "source_slug": source_slug,
            }
        return {
            "ok": True,
            "node_id": nid,
            "node_type": "chunk",
            "title": title,
            "text": text,
            "corpus_slug": slug,
            "source_slug": source_slug,
        }
    if ":entity:" in nid:
        parsed = _parse_entity_focus(nid, None)
        if not parsed:
            return {"ok": False, "error": "invalid entity node", "node_id": nid, "title": "", "text": ""}
        slug, entity_key = parsed
        conn_data = load_connections_index(slug) or {}
        label = entity_key.replace("_", " ")
        for row in conn_data.get("entities") or []:
            if not isinstance(row, dict):
                continue
            ek = str(row.get("entity_key") or row.get("canonical_key") or "").strip().lower()
            if ek == entity_key:
                label = str(row.get("label") or label).strip() or label
                break
        return {
            "ok": True,
            "node_id": nid,
            "node_type": "entity",
            "title": label,
            "text": (
                f"{label} is a named entity in this library. "
                "Open linked passages in the graph or switch to Related passages to read indexed excerpts."
            ),
            "corpus_slug": slug,
        }
    if ":source:" in nid:
        parts = nid.split(":source:", 1)
        slug = parts[0].replace("knowledge:", "", 1) if parts else ""
        source_slug = parts[1] if len(parts) > 1 else ""
        title = _concept_node_label(nid)
        if slug and source_slug:
            try:
                from tools.knowledge_tools import read_knowledge_source_body

                body = read_knowledge_source_body(slug, source_slug, offset=0, limit=6000)
                text = str(body.get("text") or body.get("content") or "").strip()
                if text:
                    return {
                        "ok": True,
                        "node_id": nid,
                        "node_type": "source",
                        "title": title,
                        "text": text[:_PASSAGE_MAX_CHARS],
                        "corpus_slug": slug,
                        "source_slug": source_slug,
                    }
            except Exception as exc:
                _log.debug("[knowledge_graph] source passage failed %s: %s", nid, exc)
        return {
            "ok": True,
            "node_id": nid,
            "node_type": "source",
            "title": title,
            "text": f"Source “{title}”. Switch to Browse to read the full document.",
            "corpus_slug": slug,
            "source_slug": source_slug,
        }
    if ":topic:" in nid:
        title = _concept_node_label(nid)
        slug = nid.split(":", 2)[1] if nid.startswith("knowledge:") else ""
        return {
            "ok": True,
            "node_id": nid,
            "node_type": "topic",
            "title": title,
            "text": (
                f"Topic “{title}”. Select a linked passage node or switch to Related passages "
                "to read indexed text."
            ),
            "corpus_slug": slug,
        }
    return {"ok": False, "error": "unsupported node", "node_id": nid, "title": "", "text": ""}


def _is_topic_hub_node(node_id: str) -> bool:
    return str(node_id or "").strip().endswith(":topic:hub")


def _concept_topic_hub_id(corpus_slug: str) -> str:
    return f"knowledge:{str(corpus_slug or '').strip()}:topic:hub"


def _concept_topic_seed_nodes(
    conn: Any,
    corpus_slug: str,
    *,
    limit: int = 12,
) -> list[str]:
    """Topic nodes with the most active concept edges (excludes the synthetic hub)."""
    slug = str(corpus_slug or "").strip()
    if not slug:
        return []
    hub = _concept_topic_hub_id(slug)
    kinds = ",".join(f"'{k}'" for k in _CONCEPT_LINK_KINDS)
    degree: dict[str, int] = {}
    sql = (
        f"SELECT src_id, dst_id FROM knowledge_graph_edges "
        f"WHERE corpus_slug = ? AND status = 'active' AND link_kind IN ({kinds})"
    )
    for sid, did in conn.execute(sql, (slug,)).fetchall():
        for x in (str(sid), str(did)):
            if ":topic:" in x and x != hub:
                degree[x] = degree.get(x, 0) + 1
    ranked = sorted(degree.keys(), key=lambda k: (-degree[k], k))
    cap = max(1, min(int(limit or 12), 24))
    return ranked[:cap]


def default_concept_focus_node(corpus_slug: str) -> str | None:
    from utils.knowledge_connections_store import load_connections_index

    data = load_connections_index(corpus_slug)
    if not data:
        return None
    topics = data.get("topics") or []
    if topics and isinstance(topics[0], dict):
        row = topics[0]
        from utils.knowledge_topic_labels import normalize_theme_label

        label = str(row.get("label") or "")
        canonical = str(row.get("canonical_key") or "").strip().lower()
        if not canonical and label:
            _, canonical = normalize_theme_label(label)
        if canonical:
            return _topic_node_id(corpus_slug, canonical)
    return None


def materialize_concept_graph_edges(corpus_slug: str) -> int:
    """Import connections.json into knowledge_graph_edges (concept layer)."""
    from chat_storage import _ensure_knowledge_graph_tables, _get_connection, _now_iso
    from utils.knowledge_connections_store import load_connections_index

    slug = str(corpus_slug or "").strip()
    data = load_connections_index(slug)
    if not data:
        return 0
    now = _now_iso()
    conn = _get_connection()
    written = 0
    try:
        _ensure_knowledge_graph_tables(conn)
        conn.execute(
            """DELETE FROM knowledge_graph_edges
               WHERE corpus_slug = ? AND link_kind IN ('topic_source', 'synthesis_link', 'topic_related')""",
            (slug,),
        )
        topic_to_sources: dict[str, list[str]] = {}
        from utils.knowledge_topic_labels import normalize_theme_label

        for row in data.get("topics") or []:
            if not isinstance(row, dict):
                continue
            label = str(row.get("label") or "").strip()
            canonical = str(row.get("canonical_key") or "").strip().lower()
            if not canonical and label:
                _, canonical = normalize_theme_label(label)
            if not canonical:
                continue
            slugs = [
                str(s or "").strip()
                for s in (row.get("source_slugs") or [])
                if str(s or "").strip() and ".." not in str(s)
            ]
            topic_to_sources[canonical] = slugs
            tid = _topic_node_id(slug, canonical)
            for sid in slugs:
                src_id = _source_node_id(slug, sid)
                low, high = _canonical_pair(tid, src_id)
                conn.execute(
                    """INSERT OR IGNORE INTO knowledge_graph_edges
                       (id, node_low, node_high, src_id, dst_id, link_kind, status, confidence,
                        corpus_slug, rationale, created_at, updated_at)
                       VALUES (?, ?, ?, ?, ?, 'topic_source', 'active', 1.0, ?, ?, ?, ?)""",
                    (str(uuid.uuid4()), low, high, tid, src_id, slug, f"Topic {label or canonical}", now, now),
                )
                written += 1
        co_linked: set[tuple[str, str]] = set()
        for edge in data.get("edges") or []:
            if not isinstance(edge, dict):
                continue
            kind = str(edge.get("kind") or "synthesis_link")
            rationale = str(edge.get("rationale") or "")[:500]
            src_term = str(edge.get("src") or "").strip()
            if kind == "topic_cooccurrence" and src_term:
                src_key = str(edge.get("theme_key") or src_term).strip().lower()
                _, normalized = normalize_theme_label(src_key)
                if normalized:
                    src_key = normalized
                src_slugs = set(topic_to_sources.get(src_key) or [])
                if not src_slugs:
                    continue
                tid = _topic_node_id(slug, src_key)
                for other_term, other_slugs in topic_to_sources.items():
                    if other_term == src_key:
                        continue
                    if not src_slugs & set(other_slugs):
                        continue
                    pair = tuple(sorted([tid, _topic_node_id(slug, other_term)]))
                    if pair in co_linked:
                        continue
                    co_linked.add(pair)
                    oid = _topic_node_id(slug, other_term)
                    low, high = _canonical_pair(tid, oid)
                    conn.execute(
                        """INSERT OR IGNORE INTO knowledge_graph_edges
                           (id, node_low, node_high, src_id, dst_id, link_kind, status, confidence,
                            corpus_slug, rationale, created_at, updated_at)
                           VALUES (?, ?, ?, ?, ?, 'topic_related', 'active', ?, ?, ?, ?, ?)""",
                        (
                            str(edge.get("id") or uuid.uuid4()),
                            low,
                            high,
                            tid,
                            oid,
                            float(edge.get("confidence") or 0.8),
                            slug,
                            rationale,
                            now,
                            now,
                        ),
                    )
                    written += 1
                continue
            if kind not in _CONCEPT_LINK_KINDS:
                continue
            if kind == "synthesis_link":
                hub = f"knowledge:{slug}:topic:hub"
                low, high = _canonical_pair(hub, hub)
                conn.execute(
                    """INSERT OR IGNORE INTO knowledge_graph_edges
                       (id, node_low, node_high, src_id, dst_id, link_kind, status, confidence,
                        corpus_slug, rationale, created_at, updated_at)
                       VALUES (?, ?, ?, ?, ?, 'synthesis_link', 'active', 0.85, ?, ?, ?, ?)""",
                    (str(edge.get("id") or uuid.uuid4()), low, high, hub, hub, slug, rationale, now, now),
                )
                written += 1
        conn.commit()
    finally:
        conn.close()
    return written


def export_knowledge_subgraph(
    focus_node_id: str,
    *,
    corpus_slug: str | None = None,
    depth: int = 2,
    max_nodes: int = _MAX_NODES_DEFAULT,
    include_suggested: bool = True,
    layers: str = "concept",
) -> dict[str, Any]:
    layer = (layers or "concept").strip().lower()
    if layer in ("concept", "concepts"):
        return _export_concept_subgraph(
            focus_node_id,
            corpus_slug=corpus_slug,
            depth=depth,
            max_nodes=max_nodes,
        )
    if layer in ("entity", "entities"):
        return _export_entity_subgraph(
            focus_node_id,
            corpus_slug=corpus_slug,
            depth=depth,
            max_nodes=max_nodes,
            include_suggested=include_suggested,
        )
    if layer == "similarity":
        return _export_chunk_subgraph(
            focus_node_id,
            corpus_slug=corpus_slug,
            depth=depth,
            max_nodes=max_nodes,
            include_suggested=include_suggested,
        )
    if layer != "all":
        return {"ok": False, "error": f"unsupported layers: {layers}", "nodes": [], "edges": []}
    return _export_all_subgraph(
        focus_node_id,
        corpus_slug=corpus_slug,
        depth=depth,
        max_nodes=max_nodes,
        include_suggested=include_suggested,
    )


def _chunk_bfs_union(
    conn: Any,
    seeds: list[str],
    *,
    corpus_slug: str,
    depth: int,
    max_nodes: int,
    include_suggested: bool,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[str]]:
    """Multi-seed chunk BFS; returns (node_payloads, edge_payloads, resolved_seed_ids)."""
    eff_corpus = str(corpus_slug or "").strip()
    depth = max(0, min(int(depth or 2), 4))
    max_nodes = max(1, min(int(max_nodes or _MAX_NODES_DEFAULT), 400))
    visited: set[str] = set()
    frontier: set[str] = set()
    resolved_seeds: list[str] = []
    for seed in seeds:
        sid = str(seed or "").strip()
        if not sid:
            continue
        row = conn.execute(
            "SELECT node_id, corpus_slug FROM knowledge_chunk_embeddings WHERE node_id = ?",
            (sid,),
        ).fetchone()
        if not row:
            continue
        if eff_corpus and str(row[1]) != eff_corpus:
            continue
        nid = str(row[0])
        if nid not in visited:
            resolved_seeds.append(nid)
        visited.add(nid)
        frontier.add(nid)
    status_clause = "status != 'rejected'" if include_suggested else "status = 'active'"
    edge_sql = (
        f"SELECT src_id, dst_id, link_kind, status, confidence FROM knowledge_graph_edges "
        f"WHERE (src_id = ? OR dst_id = ?) AND {status_clause}"
    )
    if eff_corpus:
        edge_sql += " AND corpus_slug = ?"
    for _ in range(depth):
        if len(visited) >= max_nodes:
            break
        nxt: set[str] = set()
        for nid in frontier:
            params: tuple[Any, ...] = (nid, nid, eff_corpus) if eff_corpus else (nid, nid)
            for sid, did, *_ in conn.execute(edge_sql, params).fetchall():
                for x in (str(sid), str(did)):
                    if x not in visited:
                        nxt.add(x)
        frontier = set()
        for x in nxt:
            if len(visited) >= max_nodes:
                break
            visited.add(x)
            frontier.add(x)
        if not frontier:
            break
    node_list = list(visited)[:max_nodes]
    node_payloads: list[dict[str, Any]] = []
    for nid in node_list:
        r = conn.execute(
            "SELECT node_id, label, filename, rel_path, repo_root, corpus_slug FROM knowledge_chunk_embeddings WHERE node_id = ?",
            (nid,),
        ).fetchone()
        if not r:
            continue
        filename = str(r[2] or "")
        raw_label = str(r[1] or r[2] or nid)
        title = graph_chunk_display_title(raw_label, filename=filename)
        node_payloads.append(
            {
                "id": str(r[0]),
                "title": title,
                "filename": str(r[2] or ""),
                "rel_path": str(r[3] or ""),
                "repo_root": str(r[4] or ""),
                "corpus_slug": str(r[5] or ""),
            }
        )
    allowed = {n["id"] for n in node_payloads}
    edge_payloads: list[dict[str, Any]] = []
    seen: set[tuple[str, str, str]] = set()
    qe = f"SELECT src_id, dst_id, link_kind, status, confidence FROM knowledge_graph_edges WHERE {status_clause}"
    params2: list[Any] = []
    if eff_corpus:
        qe += " AND corpus_slug = ?"
        params2.append(eff_corpus)
    for sid, did, lk, st, conf in conn.execute(qe, params2).fetchall():
        sa, sb = str(sid), str(did)
        if sa in allowed and sb in allowed:
            key = (sa, sb, str(lk))
            if key in seen:
                continue
            seen.add(key)
            edge_payloads.append(
                {
                    "source": sa,
                    "target": sb,
                    "kind": str(lk),
                    "status": str(st),
                    "confidence": float(conf) if conf is not None else None,
                    "directed": str(lk) == "chunk_sequence",
                }
            )
    return node_payloads, edge_payloads, resolved_seeds


def _export_chunk_subgraph(
    focus_node_id: str,
    *,
    corpus_slug: str | None = None,
    depth: int = 2,
    max_nodes: int = _MAX_NODES_DEFAULT,
    include_suggested: bool = True,
) -> dict[str, Any]:
    from chat_storage import _get_connection, _ensure_knowledge_graph_tables

    focus = str(focus_node_id or "").strip()
    if not focus:
        return {"ok": False, "error": "focus_node_id required", "nodes": [], "edges": []}
    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        row = conn.execute(
            "SELECT node_id, corpus_slug FROM knowledge_chunk_embeddings WHERE node_id = ?",
            (focus,),
        ).fetchone()
        if not row:
            return {"ok": False, "error": "node not found", "nodes": [], "edges": []}
        eff_corpus = str(corpus_slug or row[1] or "").strip()
        if eff_corpus and str(row[1]) != eff_corpus:
            return {"ok": False, "error": "node not in corpus", "nodes": [], "edges": []}
        node_payloads, edge_payloads, _ = _chunk_bfs_union(
            conn,
            [focus],
            corpus_slug=eff_corpus,
            depth=depth,
            max_nodes=max_nodes,
            include_suggested=include_suggested,
        )
        return {"ok": True, "nodes": node_payloads, "edges": edge_payloads, "focus": focus, "layers": "similarity"}
    finally:
        conn.close()


def _parse_entity_focus(focus_node_id: str, corpus_slug: str | None) -> tuple[str, str] | None:
    focus = str(focus_node_id or "").strip()
    match = _ENTITY_FOCUS_RE.match(focus)
    if match:
        return match.group(1), match.group(2).strip().lower()
    corpus = str(corpus_slug or "").strip()
    if not corpus or not focus:
        return None
    # Bare entity key only — not a chunk/topic/source node id misrouted to entity layer.
    if focus.startswith("knowledge:"):
        return None
    return corpus, focus.strip().lower()


def _export_entity_subgraph(
    focus_node_id: str,
    *,
    corpus_slug: str | None = None,
    depth: int = 2,
    max_nodes: int = _MAX_NODES_DEFAULT,
    include_suggested: bool = True,
) -> dict[str, Any]:
    from chat_storage import _ensure_knowledge_graph_tables, _get_connection
    from utils.knowledge_connections_store import _validate_slug, load_connections_index

    parsed = _parse_entity_focus(focus_node_id, corpus_slug)
    if not parsed:
        return {"ok": False, "error": "invalid entity focus", "nodes": [], "edges": []}
    slug, entity_key = parsed
    try:
        norm = _validate_slug(slug)
    except ValueError:
        return {"ok": False, "error": "invalid corpus", "nodes": [], "edges": []}
    if corpus_slug:
        try:
            param_norm = _validate_slug(str(corpus_slug).strip())
        except ValueError:
            return {"ok": False, "error": "invalid corpus", "nodes": [], "edges": []}
        if param_norm != norm:
            return {"ok": False, "error": "focus not in corpus", "nodes": [], "edges": []}
    focus = f"knowledge:{norm}:entity:{entity_key}"
    conn_data = load_connections_index(norm) or {}
    entity_row: dict[str, Any] | None = None
    label_fallback: dict[str, Any] | None = None
    for row in conn_data.get("entities") or []:
        if not isinstance(row, dict):
            continue
        ek = str(row.get("entity_key") or row.get("canonical_key") or "").strip().lower()
        label_low = str(row.get("label") or "").strip().lower()
        if ek and ek == entity_key:
            entity_row = dict(row)
            break
        if label_low == entity_key and label_fallback is None:
            label_fallback = dict(row)
    if entity_row is None:
        entity_row = label_fallback
    if entity_row is None:
        return {"ok": False, "error": "entity not found", "nodes": [], "edges": []}
    label = str(entity_row.get("label") or entity_key).strip() or entity_key
    raw_seeds = [
        str(x).strip()
        for x in (entity_row.get("chunk_node_ids") or [])
        if str(x).strip()
    ][: _ENTITY_SUBGRAPH_MAX_SEEDS]
    hub = {
        "id": focus,
        "title": label,
        "node_type": "entity",
        "filename": "",
        "rel_path": "",
        "repo_root": "",
        "corpus_slug": norm,
    }
    if not raw_seeds:
        return {
            "ok": True,
            "nodes": [hub],
            "edges": [],
            "focus": focus,
            "layers": "entity",
            "note": "rebuild_graph",
        }
    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        node_payloads, edge_payloads, resolved_seeds = _chunk_bfs_union(
            conn,
            raw_seeds,
            corpus_slug=norm,
            depth=depth,
            max_nodes=max_nodes,
            include_suggested=include_suggested,
        )
        if not resolved_seeds:
            return {
                "ok": True,
                "nodes": [hub],
                "edges": [],
                "focus": focus,
                "layers": "entity",
                "note": "rebuild_graph",
            }
        synthetic_edges = [
            {
                "source": focus,
                "target": seed_id,
                "kind": "entity_chunk",
                "status": "active",
                "confidence": None,
                "directed": True,
            }
            for seed_id in resolved_seeds
        ]
        nodes_out = [hub]
        seen_ids = {focus}
        for node in node_payloads:
            nid = str(node.get("id") or "")
            if nid and nid not in seen_ids:
                nodes_out.append(node)
                seen_ids.add(nid)
        return {
            "ok": True,
            "nodes": nodes_out,
            "edges": synthetic_edges + edge_payloads,
            "focus": focus,
            "layers": "entity",
        }
    finally:
        conn.close()


def _export_concept_subgraph(
    focus_node_id: str,
    *,
    corpus_slug: str | None = None,
    depth: int = 2,
    max_nodes: int = _MAX_NODES_DEFAULT,
) -> dict[str, Any]:
    from chat_storage import _ensure_knowledge_graph_tables, _get_connection

    slug = str(corpus_slug or "").strip()
    focus = str(focus_node_id or "").strip()
    if not slug and focus.startswith("knowledge:"):
        parts = focus.split(":")
        if len(parts) >= 3:
            slug = parts[1].strip()
    if not focus or ":chunk:" in focus:
        alt = default_concept_focus_node(slug) if slug else None
        focus = alt or focus
    depth = max(0, min(int(depth or 2), 4))
    max_nodes = max(1, min(int(max_nodes or _MAX_NODES_DEFAULT), 400))
    kinds = ",".join(f"'{k}'" for k in _CONCEPT_LINK_KINDS)
    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        if _is_topic_hub_node(focus):
            alt = default_concept_focus_node(slug) if slug else None
            if alt and not _is_topic_hub_node(alt):
                focus = alt
        multi_seed: list[str] | None = None
        if not focus or _is_topic_hub_node(focus):
            if slug:
                seeds = _concept_topic_seed_nodes(conn, slug, limit=min(12, max_nodes))
                if seeds:
                    multi_seed = seeds
                    focus = seeds[0]
        if not focus:
            return {"ok": False, "error": "no concept focus", "nodes": [], "edges": [], "layers": "concept"}
        if multi_seed:
            visited = set(multi_seed)
            frontier = set(multi_seed)
        else:
            visited = {focus}
            frontier = {focus}
        for _ in range(depth):
            if len(visited) >= max_nodes:
                break
            nxt: set[str] = set()
            for nid in frontier:
                sql = (
                    f"SELECT src_id, dst_id FROM knowledge_graph_edges "
                    f"WHERE (src_id = ? OR dst_id = ?) AND status = 'active' AND link_kind IN ({kinds})"
                )
                params: list[Any] = [nid, nid]
                if slug:
                    sql += " AND corpus_slug = ?"
                    params.append(slug)
                for sid, did in conn.execute(sql, params).fetchall():
                    for x in (str(sid), str(did)):
                        if x not in visited:
                            nxt.add(x)
            frontier = set()
            for x in nxt:
                if len(visited) >= max_nodes:
                    break
                visited.add(x)
                frontier.add(x)
            if not frontier:
                break
        node_payloads = [
            {
                "id": nid,
                "title": _concept_node_label(nid),
                "node_type": "topic" if ":topic:" in nid else "source",
                "display_label": _concept_node_label(nid),
                "corpus_slug": slug,
            }
            for nid in list(visited)[:max_nodes]
        ]
        allowed = {n["id"] for n in node_payloads}
        edge_payloads: list[dict[str, Any]] = []
        qe = f"SELECT src_id, dst_id, link_kind, confidence FROM knowledge_graph_edges WHERE status = 'active' AND link_kind IN ({kinds})"
        params2: list[Any] = []
        if slug:
            qe += " AND corpus_slug = ?"
            params2.append(slug)
        for sid, did, lk, conf in conn.execute(qe, params2).fetchall():
            sa, sb = str(sid), str(did)
            if sa in allowed and sb in allowed:
                edge_payloads.append(
                    {
                        "source": sa,
                        "target": sb,
                        "kind": str(lk),
                        "status": "active",
                        "confidence": float(conf) if conf is not None else None,
                        "directed": False,
                    }
                )
        return {
            "ok": True,
            "nodes": node_payloads,
            "edges": edge_payloads,
            "focus": focus,
            "layers": "concept",
        }
    finally:
        conn.close()


def _export_all_subgraph(
    focus_node_id: str,
    *,
    corpus_slug: str | None = None,
    depth: int = 2,
    max_nodes: int = _MAX_NODES_DEFAULT,
    include_suggested: bool = True,
) -> dict[str, Any]:
    concept = _export_concept_subgraph(
        focus_node_id, corpus_slug=corpus_slug, depth=depth, max_nodes=max_nodes
    )
    if concept.get("nodes"):
        return {**concept, "layers": "all"}
    return _export_chunk_subgraph(
        focus_node_id,
        corpus_slug=corpus_slug,
        depth=depth,
        max_nodes=max_nodes,
        include_suggested=include_suggested,
    )


def focus_node_for_source(corpus_slug: str, source_slug: str) -> str | None:
    """Best-effort chunk node for a studied source (powers Related graph-first)."""
    slug = str(corpus_slug or "").strip()
    src = str(source_slug or "").strip()
    if not slug or not src:
        return None
    from chat_storage import _ensure_knowledge_graph_tables, _get_connection

    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        row = conn.execute(
            """SELECT node_id FROM knowledge_chunk_embeddings
               WHERE corpus_slug = ? AND source_slug = ?
               ORDER BY row_id LIMIT 1""",
            (slug, src),
        ).fetchone()
        return str(row[0]) if row else None
    finally:
        conn.close()


def list_knowledge_graph_neighbors(
    node_id: str,
    *,
    corpus_slug: str | None = None,
    limit: int = 12,
    include_suggested: bool = False,
) -> list[dict[str, Any]]:
    """One-hop neighbors of a chunk node (active edges by default)."""
    focus = str(node_id or "").strip()
    if not focus:
        return []
    limit = max(1, min(int(limit or 12), 40))
    from chat_storage import _ensure_knowledge_graph_tables, _get_connection

    status_clause = "status != 'rejected'" if include_suggested else "status = 'active'"
    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        params: list[Any] = [focus, focus]
        edge_sql = (
            f"SELECT id, src_id, dst_id, link_kind, status, confidence FROM knowledge_graph_edges "
            f"WHERE (src_id = ? OR dst_id = ?) AND {status_clause}"
        )
        if corpus_slug:
            edge_sql += " AND corpus_slug = ?"
            params.append(str(corpus_slug).strip())
        edge_sql += (
            " ORDER BY CASE link_kind WHEN 'chunk_sequence' THEN 0 ELSE 1 END, "
            "COALESCE(confidence, 0) DESC"
        )
        rows = conn.execute(edge_sql, params).fetchall()
        sequence_cap = min(3, limit)
        out: list[dict[str, Any]] = []
        seen: set[str] = set()

        def _append_neighbor(eid: Any, sid: Any, did: Any, lk: Any, st: Any, conf: Any) -> bool:
            other = str(did) if str(sid) == focus else str(sid)
            if other == focus or other in seen:
                return False
            nrow = conn.execute(
                """SELECT node_id, label, filename, source_slug, rel_path, repo_root, corpus_slug
                   FROM knowledge_chunk_embeddings WHERE node_id = ?""",
                (other,),
            ).fetchone()
            if not nrow:
                return False
            seen.add(other)
            direction = "outbound" if str(sid) == focus else "inbound"
            out.append(
                {
                    "neighbor_id": other,
                    "direction": direction,
                    "link_kind": str(lk),
                    "edge_id": str(eid),
                    "status": str(st),
                    "confidence": float(conf) if conf is not None else None,
                    "title": graph_chunk_display_title(
                        str(nrow[1] or nrow[2] or other),
                        filename=str(nrow[2] or ""),
                    ),
                    "filename": str(nrow[2] or ""),
                    "source_slug": str(nrow[3] or ""),
                    "rel_path": str(nrow[4] or ""),
                    "repo_root": str(nrow[5] or ""),
                    "corpus_slug": str(nrow[6] or ""),
                }
            )
            return True

        for eid, sid, did, lk, st, conf in rows:
            if str(lk) != "chunk_sequence":
                continue
            if len([x for x in out if x["link_kind"] == "chunk_sequence"]) >= sequence_cap:
                continue
            _append_neighbor(eid, sid, did, lk, st, conf)
            if len(out) >= limit:
                return out
        for eid, sid, did, lk, st, conf in rows:
            if str(lk) == "chunk_sequence":
                continue
            _append_neighbor(eid, sid, did, lk, st, conf)
            if len(out) >= limit:
                break
        return out
    finally:
        conn.close()


def list_knowledge_graph_suggestions(
    *,
    corpus_slug: str | None = None,
    focus_node_id: str | None = None,
    limit: int = 50,
    offset: int = 0,
) -> dict[str, Any]:
    from chat_storage import _get_connection, _ensure_knowledge_graph_tables

    limit = max(1, min(int(limit or 50), 100))
    offset = max(0, min(int(offset or 0), 10_000))
    corpus = str(corpus_slug or "").strip() or None
    focus = str(focus_node_id or "").strip() or None
    if offset > 0 and not corpus:
        return {"suggestions": [], "total_count": 0, "schema": "mugi.knowledge.graph.suggestions.v2"}
    if focus and corpus and not validate_focus_node_for_corpus(focus, corpus):
        return {"suggestions": [], "total_count": 0, "error": "focus not in corpus", "schema": "mugi.knowledge.graph.suggestions.v2"}

    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        clauses = ["status = 'suggested'"]
        params: list[Any] = []
        if corpus:
            clauses.append("corpus_slug = ?")
            params.append(corpus)
        if focus:
            clauses.append("(src_id = ? OR dst_id = ?)")
            params.extend([focus, focus])
        where = " AND ".join(clauses)
        total_count = conn.execute(
            f"SELECT COUNT(*) FROM knowledge_graph_edges WHERE {where}",
            params,
        ).fetchone()[0]
        rows = conn.execute(
            f"""SELECT id, src_id, dst_id, link_kind, confidence, corpus_slug, rationale, created_at
                FROM knowledge_graph_edges WHERE {where}
                ORDER BY confidence DESC LIMIT ? OFFSET ?""",
            [*params, limit, offset],
        ).fetchall()
        suggestions = _enrich_edges_with_labels(conn, rows, focus_node_id=focus)
        return {
            "schema": "mugi.knowledge.graph.suggestions.v2",
            "suggestions": suggestions,
            "total_count": int(total_count or 0),
        }
    finally:
        conn.close()


def dismiss_all_knowledge_graph_suggestions(*, corpus_slug: str) -> dict[str, Any]:
    from chat_storage import _ensure_knowledge_graph_tables, _get_connection, _now_iso

    slug = str(corpus_slug or "").strip()
    if not slug:
        return {"ok": False, "error": "corpus required"}
    now = _now_iso()
    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        cur = conn.execute(
            """UPDATE knowledge_graph_edges SET status = 'rejected', updated_at = ?
               WHERE corpus_slug = ? AND status = 'suggested'""",
            (now, slug),
        )
        conn.commit()
        return {"ok": True, "corpus_slug": slug, "dismissed_count": int(cur.rowcount or 0)}
    finally:
        conn.close()


def promote_knowledge_graph_suggestions(
    *,
    corpus_slug: str,
    min_confidence: float | None = None,
    focus_node_id: str | None = None,
) -> dict[str, Any]:
    from chat_storage import _ensure_knowledge_graph_tables, _get_connection, _now_iso

    slug = str(corpus_slug or "").strip()
    if not slug:
        return {"ok": False, "error": "corpus required"}
    floor = float(min_confidence if min_confidence is not None else _AUTO_ACTIVE_THRESHOLD)
    floor = max(_SIMILARITY_MIN, min(1.0, floor))
    focus = str(focus_node_id or "").strip() or None
    if focus and not validate_focus_node_for_corpus(focus, slug):
        return {"ok": False, "error": "focus not in corpus"}
    now = _now_iso()
    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        clauses = ["corpus_slug = ?", "status = 'suggested'", "confidence >= ?"]
        params: list[Any] = [slug, floor]
        if focus:
            clauses.append("(src_id = ? OR dst_id = ?)")
            params.extend([focus, focus])
        where = " AND ".join(clauses)
        cur = conn.execute(
            f"UPDATE knowledge_graph_edges SET status = 'active', updated_at = ? WHERE {where}",
            [now, *params],
        )
        conn.commit()
        return {
            "ok": True,
            "corpus_slug": slug,
            "promoted_count": int(cur.rowcount or 0),
            "min_confidence": floor,
        }
    finally:
        conn.close()


def confirm_knowledge_graph_edge(edge_id: str) -> dict[str, Any]:
    from chat_storage import _get_connection, _ensure_knowledge_graph_tables, _now_iso

    eid = str(edge_id or "").strip()
    if not eid:
        return {"ok": False, "error": "edge_id required"}
    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        now = _now_iso()
        cur = conn.execute(
            "UPDATE knowledge_graph_edges SET status = 'active', updated_at = ? WHERE id = ? AND status = 'suggested'",
            (now, eid),
        )
        conn.commit()
        if int(cur.rowcount or 0) == 0:
            row = conn.execute(
                "SELECT status FROM knowledge_graph_edges WHERE id = ?",
                (eid,),
            ).fetchone()
            if not row:
                return {"ok": False, "error": "not found"}
            if str(row[0]) != "active":
                return {"ok": False, "error": "not suggested"}
        row = conn.execute(
            "SELECT id, src_id, dst_id, link_kind, status, confidence, corpus_slug FROM knowledge_graph_edges WHERE id = ?",
            (eid,),
        ).fetchone()
        if not row:
            return {"ok": False, "error": "not found"}
        return {
            "ok": True,
            "edge": {
                "id": row[0],
                "src_id": row[1],
                "dst_id": row[2],
                "link_kind": row[3],
                "status": row[4],
                "confidence": row[5],
                "corpus_slug": row[6],
            },
        }
    finally:
        conn.close()


def reject_knowledge_graph_edge(edge_id: str) -> dict[str, Any]:
    from chat_storage import _get_connection, _ensure_knowledge_graph_tables, _now_iso

    eid = str(edge_id or "").strip()
    if not eid:
        return {"ok": False, "error": "edge_id required"}
    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        now = _now_iso()
        cur = conn.execute(
            "UPDATE knowledge_graph_edges SET status = 'rejected', updated_at = ? WHERE id = ? AND status IN ('suggested', 'active')",
            (now, eid),
        )
        conn.commit()
        return {"ok": cur.rowcount > 0}
    finally:
        conn.close()


def refresh_knowledge_graph_suggestions(corpus_slug: str) -> dict[str, Any]:
    from tools.knowledge_tools import (
        _hydrate_raw_texts_for_indexing,
        _load_manifest,
        _read_text_file,
        _synthesis_path,
    )

    manifest = _load_manifest(corpus_slug)
    if not manifest:
        return {"ok": False, "error": "corpus not found"}
    label = str(manifest.get("label") or corpus_slug)
    synthesis = _read_text_file(_synthesis_path(corpus_slug)) if os.path.isfile(_synthesis_path(corpus_slug)) else ""
    sources = list(manifest.get("sources") or [])
    raw_texts = _hydrate_raw_texts_for_indexing(corpus_slug, sources, None)
    return rebuild_knowledge_graph_for_corpus(
        corpus_slug, label, synthesis, sources, raw_texts=raw_texts
    )


def gitnexus_query_hint_for_node(node_id: str) -> str | None:
    """Best-effort hint for operators/agents to pivot chunk → GitNexus MCP."""
    m = _CHUNK_NODE_RE.match(str(node_id or ""))
    if not m:
        return None
    from chat_storage import _get_connection, _ensure_knowledge_graph_tables

    conn = _get_connection()
    try:
        _ensure_knowledge_graph_tables(conn)
        row = conn.execute(
            "SELECT rel_path, repo_root, filename FROM knowledge_chunk_embeddings WHERE node_id = ?",
            (node_id,),
        ).fetchone()
        if not row:
            return None
        rel, root, fname = str(row[0] or ""), str(row[1] or ""), str(row[2] or "")
        if rel and root:
            return f"gitnexus_context for file {rel} under {root}"
        if fname:
            return f"gitnexus_query for symbols in {fname}"
        return None
    finally:
        conn.close()
