"""HTTP routes for Knowledge Library — search, browse, connections (operator UI)."""
from __future__ import annotations

import asyncio
import logging
import os
import re
import time
from collections import defaultdict
from typing import Any

from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse, PlainTextResponse

_log = logging.getLogger(__name__)

_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
_MAX_SEARCH_RESULTS = 20
_SEARCH_RATE_LIMIT = 30
_ASK_RATE_LIMIT = 20
_CONNECTIONS_REBUILD_RATE_LIMIT = 10
_SEARCH_WINDOW_SEC = 60.0
_search_hits: dict[str, list[float]] = defaultdict(list)
_kg_graph_refresh_until = 0.0


_MAX_ASK_QUESTION_CHARS = 8000


def _parse_ask_max_sources(raw: Any, default: int = 8) -> int:
    try:
        val = int(raw if raw is not None else default)
    except (TypeError, ValueError):
        val = default
    return max(1, min(val, 12))


def _ask_payload_sync(q: str, corpus: str | None, max_sources: int) -> dict[str, Any]:
    from tools.knowledge_tools import ask_knowledge_corpus

    return ask_knowledge_corpus(q, corpus=corpus, max_sources=max_sources)


def _validate_ask_question(q: str) -> str:
    text = (q or "").strip()
    if not text:
        raise HTTPException(status_code=400, detail="q required")
    if len(text) > _MAX_ASK_QUESTION_CHARS:
        raise HTTPException(
            status_code=400,
            detail=f"q exceeds {_MAX_ASK_QUESTION_CHARS} characters",
        )
    return text


def _allow_search(bucket: str = "search") -> bool:
    now = time.time()
    entries = _search_hits[bucket]
    entries[:] = [ts for ts in entries if (now - ts) < _SEARCH_WINDOW_SEC]
    if bucket == "ask":
        limit = _ASK_RATE_LIMIT
    elif bucket == "connections_rebuild":
        limit = _CONNECTIONS_REBUILD_RATE_LIMIT
    else:
        limit = _SEARCH_RATE_LIMIT
    if len(entries) >= limit:
        return False
    entries.append(now)
    return True


_MAX_CONNECTIONS_PARAM_CHARS = 200


def _bounded_connections_param(raw: str | None, name: str) -> str | None:
    text = (raw or "").strip()
    if not text:
        return None
    if len(text) > _MAX_CONNECTIONS_PARAM_CHARS:
        raise HTTPException(
            status_code=400,
            detail=f"{name} exceeds {_MAX_CONNECTIONS_PARAM_CHARS} characters",
        )
    return text


def _slug_from_param(slug: str) -> str:
    raw = (slug or "").strip().lower()
    if not raw or not _SLUG_RE.match(raw):
        raise HTTPException(status_code=400, detail="Invalid corpus slug")
    return raw


def _search_payload_sync(
    q: str,
    corpus: str | None,
    limit: int,
    semantic: bool,
) -> dict[str, Any]:
    from tools.knowledge_tools import format_knowledge_hits_as_text, search_knowledge_hits

    hits = search_knowledge_hits(q, corpus=corpus, max_results=limit, semantic=semantic)
    return {
        "schema": "mugi.knowledge.search.v1",
        "query": q,
        "corpus": corpus,
        "semantic": semantic,
        "hits": hits,
        "text": format_knowledge_hits_as_text(hits, include_tip=False),
    }


def _parse_unified_scopes(raw: str | None) -> set[str] | None:
    text = (raw or "").strip().lower()
    if not text or text == "all":
        return None
    parts = {p.strip() for p in text.split(",") if p.strip()}
    return parts or None


def _unified_search_payload_sync(
    q: str,
    scopes: set[str] | None,
    semantic: bool,
    max_knowledge: int,
    max_memory: int,
    max_vault: int,
) -> dict[str, Any]:
    from tools.knowledge_tools import unified_library_search

    return unified_library_search(
        q,
        scopes=scopes,
        max_knowledge=max_knowledge,
        max_memory=max_memory,
        max_vault=max_vault,
        semantic=semantic,
    )


def _topics_payload_sync() -> dict[str, Any]:
    from tools.knowledge_tools import get_knowledge_topic_index

    index = get_knowledge_topic_index()
    topics = [
        {"topic": term, "corpora": entries}
        for term, entries in sorted(index.items())
        if len(entries) >= 1
    ]
    return {"schema": "mugi.knowledge.topics.v1", "topics": topics}


def _connections_payload_sync(
    topic: str | None,
    query: str | None,
    max_corpora: int,
    corpus: str | None = None,
    *,
    related: bool = False,
    theme_detail: bool = False,
) -> dict[str, Any]:
    from tools.knowledge_tools import build_knowledge_connections_payload

    payload = build_knowledge_connections_payload(
        topic=topic,
        query=query,
        corpus=corpus,
        max_corpora=max_corpora,
        related=related,
        theme_detail=theme_detail,
    )
    return {"schema": "mugi.knowledge.connections.v1", **payload}


def _connections_rebuild_sync(slug: str) -> dict[str, Any]:
    from tools.knowledge_tools import rebuild_connections_index

    try:
        return rebuild_connections_index(slug)
    except ValueError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except RuntimeError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc


def _gaps_payload_sync(limit: int) -> dict[str, Any]:
    from tools.knowledge_tools import read_knowledge_gaps

    gaps = read_knowledge_gaps(limit=limit)
    return {"schema": "mugi.knowledge.gaps.v1", "gaps": gaps}


def _catalog_payload_sync(norm: str) -> dict[str, Any] | None:
    from tools.knowledge_tools import _load_manifest, _synthesis_path, list_knowledge_corpora

    manifest = _load_manifest(norm)
    if not manifest:
        return None
    sources_out: list[dict[str, Any]] = []
    for src in manifest.get("sources") or []:
        if not isinstance(src, dict):
            continue
        sources_out.append(
            {
                "slug": src.get("slug"),
                "filename": src.get("filename"),
                "path": src.get("path") or "",
                "rel_path": src.get("rel_path") or "",
                "repo_root": src.get("repo_root") or "",
                "status": src.get("status"),
                "topics": src.get("topics") or [],
                "last_error": src.get("last_error") or "",
            }
        )
    row = next((c for c in list_knowledge_corpora() if c.get("slug") == norm), {})
    return {
        "schema": "mugi.knowledge.catalog.v1",
        "slug": norm,
        "label": manifest.get("label") or norm,
        "source_path": manifest.get("source_path") or "",
        "digest_status": manifest.get("digest_status") or row.get("digest_status"),
        "topics": manifest.get("topics") or [],
        "broad_topics": manifest.get("broad_topics") or [],
        "sources": sources_out,
        "source_count": len(sources_out),
        "failed_source_count": sum(
            1 for s in sources_out if str(s.get("status") or "").lower() == "failed"
        ),
        "has_synthesis": os.path.isfile(_synthesis_path(norm)),
        "ingest_profile": manifest.get("ingest_profile") or "docs",
    }


def _synthesis_text_sync(norm: str) -> str | None:
    from tools.knowledge_tools import _synthesis_path, polish_knowledge_summary_body
    from utils.text_io import read_text_file_with_encoding_fallback

    path = _synthesis_path(norm)
    if not os.path.isfile(path):
        return None
    raw = read_text_file_with_encoding_fallback(path)
    if not raw:
        return raw
    if raw.lstrip().startswith("#"):
        lines = raw.splitlines()
        head = lines[0] if lines else ""
        body = "\n".join(lines[1:]).strip()
        polished = polish_knowledge_summary_body(body)
        return f"{head}\n\n{polished}".strip() if polished else head
    return polish_knowledge_summary_body(raw)


def _source_read_payload_sync(
    norm: str,
    source_id: str,
    offset: int,
    limit: int,
    notes: bool,
    summary_only: bool,
) -> dict[str, Any]:
    if notes:
        from tools.knowledge_tools import read_knowledge_source_notes_body

        body = read_knowledge_source_notes_body(
            norm, source_id, offset=offset, limit=limit, summary_only=summary_only
        )
    else:
        from tools.knowledge_tools import read_knowledge_source_body

        body = read_knowledge_source_body(norm, source_id, offset=offset, limit=limit)
    return {"schema": "mugi.knowledge.source.v1", "slug": norm, **body}


def _related_payload_sync(
    norm: str,
    query: str | None,
    source_id: str | None,
    limit: int,
) -> dict[str, Any]:
    from tools.knowledge_tools import search_knowledge_related_with_graph

    hits = search_knowledge_related_with_graph(
        norm, query=query, source_id=source_id, max_results=limit
    )
    return {"schema": "mugi.knowledge.related.v1", "slug": norm, "hits": hits}


def _allow_graph_refresh() -> bool:
    global _kg_graph_refresh_until
    now = time.time()
    if now < _kg_graph_refresh_until:
        return False
    _kg_graph_refresh_until = now + 45.0
    return True


class _KnowledgeReadFailed(Exception):
    """Sync read handler failed; route should return JSON 500."""


async def _read_value_from_thread(fn, *args, **kwargs):
    try:
        return await asyncio.to_thread(fn, *args, **kwargs)
    except HTTPException:
        raise
    except Exception as exc:
        _log.exception("knowledge read route %s failed", getattr(fn, "__name__", "handler"))
        raise _KnowledgeReadFailed(str(exc)) from exc


def _read_error_json(exc: _KnowledgeReadFailed) -> JSONResponse:
    return JSONResponse(content={"ok": False, "error": exc.args[0]}, status_code=500)


async def _read_json_from_thread(fn, *args, **kwargs) -> JSONResponse:
    try:
        content = await _read_value_from_thread(fn, *args, **kwargs)
    except _KnowledgeReadFailed as exc:
        return _read_error_json(exc)
    return JSONResponse(content=content)


def _knowledge_graph_mutation_response(content: dict) -> JSONResponse:
    if content.get("ok"):
        return JSONResponse(content=content)
    err = str(content.get("error") or "").lower()
    if not err:
        return JSONResponse(content=content, status_code=404)
    if "not found" in err:
        return JSONResponse(content=content, status_code=404)
    if "required" in err or "invalid" in err:
        return JSONResponse(content=content, status_code=400)
    if "embedding" in err or "no_embeddings" in err:
        return JSONResponse(content=content, status_code=503)
    return JSONResponse(content=content, status_code=409)


def _knowledge_graph_edge_response(content: dict) -> JSONResponse:
    return _knowledge_graph_mutation_response(content)


def register_knowledge_http_routes(fastapi_app: Any) -> None:
    """Mount read/search knowledge routes (auth via global middleware)."""

    @fastapi_app.get("/knowledge/search")
    async def knowledge_search(request: Request):
        if not _allow_search():
            raise HTTPException(status_code=429, detail="Knowledge search rate limited")
        q = (request.query_params.get("q") or "").strip()
        if not q:
            raise HTTPException(status_code=400, detail="q required")
        corpus = (request.query_params.get("corpus") or "").strip() or None
        semantic_raw = (request.query_params.get("semantic") or "true").lower()
        semantic = semantic_raw in ("1", "true", "yes")
        try:
            limit = int(request.query_params.get("limit") or _MAX_SEARCH_RESULTS)
        except ValueError:
            limit = _MAX_SEARCH_RESULTS
        limit = max(1, min(limit, _MAX_SEARCH_RESULTS))

        return await _read_json_from_thread(_search_payload_sync, q, corpus, limit, semantic)

    @fastapi_app.get("/knowledge/unified_search")
    async def knowledge_unified_search(request: Request):
        if not _allow_search():
            raise HTTPException(status_code=429, detail="Knowledge search rate limited")
        q = (request.query_params.get("q") or "").strip()
        if not q:
            raise HTTPException(status_code=400, detail="q required")
        semantic_raw = (request.query_params.get("semantic") or "true").lower()
        semantic = semantic_raw in ("1", "true", "yes")
        scopes = _parse_unified_scopes(request.query_params.get("scopes"))

        def _lim(name: str, default: int, ceiling: int) -> int:
            try:
                val = int(request.query_params.get(name) or default)
            except ValueError:
                val = default
            return max(0, min(val, ceiling))

        max_knowledge = _lim("max_knowledge", 12, _MAX_SEARCH_RESULTS)
        max_memory = _lim("max_memory", 8, _MAX_SEARCH_RESULTS)
        max_vault = _lim("max_vault", 10, 40)

        return await _read_json_from_thread(
            _unified_search_payload_sync,
            q,
            scopes,
            semantic,
            max_knowledge,
            max_memory,
            max_vault,
        )

    @fastapi_app.get("/knowledge/ask")
    async def knowledge_ask_get(request: Request):
        if not _allow_search("ask"):
            raise HTTPException(status_code=429, detail="Knowledge ask rate limited")
        q = _validate_ask_question(request.query_params.get("q") or "")
        corpus = (request.query_params.get("corpus") or "").strip() or None
        max_sources = _parse_ask_max_sources(request.query_params.get("max_sources"))
        return await _read_json_from_thread(_ask_payload_sync, q, corpus, max_sources)

    @fastapi_app.post("/knowledge/ask")
    async def knowledge_ask_post(request: Request):
        if not _allow_search("ask"):
            raise HTTPException(status_code=429, detail="Knowledge ask rate limited")
        try:
            body = await request.json()
        except Exception:
            body = {}
        if not isinstance(body, dict):
            body = {}
        q = _validate_ask_question(str(body.get("q") or body.get("question") or ""))
        corpus = str(body.get("corpus") or "").strip() or None
        max_sources = _parse_ask_max_sources(body.get("max_sources"))
        return await _read_json_from_thread(_ask_payload_sync, q, corpus, max_sources)

    @fastapi_app.get("/knowledge/topics")
    async def knowledge_topics():
        return await _read_json_from_thread(_topics_payload_sync)

    @fastapi_app.get("/knowledge/connections")
    async def knowledge_connections(request: Request):
        topic = _bounded_connections_param(request.query_params.get("topic"), "topic")
        q = _bounded_connections_param(request.query_params.get("q"), "q")
        corpus_raw = (request.query_params.get("corpus") or "").strip()
        corpus = _slug_from_param(corpus_raw) if corpus_raw else None
        related = (request.query_params.get("related") or "").strip() == "1"
        theme_detail = (request.query_params.get("theme_detail") or "").strip() == "1"
        if related and not (q or topic):
            raise HTTPException(status_code=400, detail="q or topic required when related=1")
        if related and not _allow_search("search"):
            raise HTTPException(status_code=429, detail="Connections search rate limit exceeded")
        try:
            max_corpora = int(request.query_params.get("max_corpora") or 10)
        except ValueError:
            max_corpora = 10
        max_corpora = max(1, min(max_corpora, 20))

        return await _read_json_from_thread(
            _connections_payload_sync,
            topic,
            q,
            max_corpora,
            corpus,
            related=related,
            theme_detail=theme_detail or bool(topic),
        )

    @fastapi_app.post("/knowledge/{slug}/connections/rebuild")
    async def knowledge_connections_rebuild(slug: str):
        norm = _slug_from_param(slug)
        if not _allow_search("connections_rebuild"):
            raise HTTPException(status_code=429, detail="Connections rebuild rate limit exceeded")
        try:
            content = await _read_value_from_thread(_connections_rebuild_sync, norm)
        except _KnowledgeReadFailed as exc:
            return _read_error_json(exc)
        return JSONResponse(content=content)

    @fastapi_app.get("/knowledge/{slug}/entities")
    async def knowledge_corpus_entities(slug: str, request: Request):
        norm = _slug_from_param(slug)
        if not _allow_search("search"):
            raise HTTPException(status_code=429, detail="Knowledge entities rate limit exceeded")
        q = _bounded_connections_param(request.query_params.get("q"), "q")
        try:
            limit = int(request.query_params.get("limit") or 40)
        except ValueError:
            limit = 40
        limit = max(1, min(limit, 40))
        from tools.knowledge_tools import list_corpus_entities

        try:
            content = await _read_value_from_thread(list_corpus_entities, norm, q=q, limit=limit)
        except _KnowledgeReadFailed as exc:
            return _read_error_json(exc)
        return JSONResponse(content={**content, "schema": "mugi.knowledge.entities.v1"})

    @fastapi_app.get("/knowledge/{slug}/entity")
    async def knowledge_corpus_entity_detail(slug: str, request: Request):
        norm = _slug_from_param(slug)
        if not _allow_search("search"):
            raise HTTPException(status_code=429, detail="Knowledge entity detail rate limit exceeded")
        entity_key = _bounded_connections_param(request.query_params.get("key"), "key")
        if not entity_key:
            raise HTTPException(status_code=400, detail="key required")
        from tools.knowledge_tools import get_corpus_entity_detail

        try:
            content = await _read_value_from_thread(get_corpus_entity_detail, norm, entity_key)
        except _KnowledgeReadFailed as exc:
            return _read_error_json(exc)
        return JSONResponse(content={**content, "schema": "mugi.knowledge.entity.v1"})

    @fastapi_app.get("/knowledge/{slug}/contradictions")
    async def knowledge_corpus_contradictions(slug: str, request: Request):
        norm = _slug_from_param(slug)
        if not _allow_search("search"):
            raise HTTPException(status_code=429, detail="Knowledge contradictions rate limit exceeded")
        try:
            limit = int(request.query_params.get("limit") or 20)
        except ValueError:
            limit = 20
        limit = max(1, min(limit, 40))
        from tools.knowledge_tools import list_corpus_contradictions

        try:
            content = await _read_value_from_thread(list_corpus_contradictions, norm, limit=limit)
        except _KnowledgeReadFailed as exc:
            return _read_error_json(exc)
        return JSONResponse(content={**content, "schema": "mugi.knowledge.contradictions.v1"})

    @fastapi_app.post("/knowledge/gaps/resolve")
    async def knowledge_gaps_resolve(request: Request):
        try:
            body = await request.json()
        except Exception:
            body = {}
        if not isinstance(body, dict):
            body = {}
        gap_id = str(body.get("id") or "").strip()
        if not gap_id:
            raise HTTPException(status_code=400, detail="id required")
        from tools.knowledge_tools import resolve_knowledge_gap

        ok = await asyncio.to_thread(resolve_knowledge_gap, gap_id)
        if not ok:
            raise HTTPException(status_code=404, detail="Gap not found")
        return JSONResponse(content={"schema": "mugi.knowledge.gaps.v1", "ok": True, "id": gap_id})

    @fastapi_app.get("/knowledge/gaps")
    async def knowledge_gaps(request: Request):
        try:
            limit = int(request.query_params.get("limit") or 30)
        except ValueError:
            limit = 30
        return await _read_json_from_thread(_gaps_payload_sync, limit)

    @fastapi_app.get("/knowledge/{slug}/catalog")
    async def knowledge_catalog(slug: str):
        norm = _slug_from_param(slug)
        try:
            content = await _read_value_from_thread(_catalog_payload_sync, norm)
        except _KnowledgeReadFailed as exc:
            return _read_error_json(exc)
        if content is None:
            raise HTTPException(status_code=404, detail="Corpus not found")
        return JSONResponse(content=content)

    @fastapi_app.get("/knowledge/{slug}/synthesis")
    async def knowledge_synthesis(slug: str):
        norm = _slug_from_param(slug)
        try:
            text = await _read_value_from_thread(_synthesis_text_sync, norm)
        except _KnowledgeReadFailed as exc:
            return _read_error_json(exc)
        if text is None:
            raise HTTPException(status_code=404, detail="Synthesis not found")
        return PlainTextResponse(text)

    @fastapi_app.get("/knowledge/{slug}/source/{source_id}")
    async def knowledge_source_read(slug: str, source_id: str, request: Request):
        norm = _slug_from_param(slug)
        try:
            offset = int(request.query_params.get("offset") or 0)
        except ValueError:
            offset = 0
        try:
            limit = int(request.query_params.get("limit") or 5000)
        except ValueError:
            limit = 5000
        notes = str(request.query_params.get("notes") or "").lower() in ("1", "true", "yes")
        summary_only = str(request.query_params.get("summary_only") or "").lower() in ("1", "true", "yes")
        try:
            body = await _read_value_from_thread(
                _source_read_payload_sync,
                norm,
                source_id,
                offset,
                limit,
                notes,
                summary_only,
            )
        except _KnowledgeReadFailed as exc:
            return _read_error_json(exc)
        err_prefixes = ("No corpus", "corpus is required", "No source matching")
        if str(body.get("text") or "").startswith(err_prefixes):
            raise HTTPException(status_code=404, detail=str(body.get("text") or "")[:500])
        return JSONResponse(content=body)

    @fastapi_app.get("/knowledge/{slug}/related")
    async def knowledge_related(slug: str, request: Request):
        if not _allow_search("related"):
            raise HTTPException(status_code=429, detail="Knowledge related rate limited")
        norm = _slug_from_param(slug)
        q = (request.query_params.get("q") or "").strip() or None
        source_id = (request.query_params.get("source_id") or "").strip() or None
        if not q and not source_id:
            raise HTTPException(status_code=400, detail="q or source_id required")
        try:
            limit = int(request.query_params.get("limit") or 6)
        except ValueError:
            limit = 6
        return await _read_json_from_thread(_related_payload_sync, norm, q, source_id, limit)

    @fastapi_app.get("/knowledge/graph/subgraph")
    async def knowledge_graph_subgraph(request: Request):
        if not _allow_search("search"):
            raise HTTPException(status_code=429, detail="Knowledge graph subgraph rate limit exceeded")
        focus = _bounded_connections_param(request.query_params.get("focus"), "focus")
        if not focus:
            raise HTTPException(status_code=400, detail="focus required")
        corpus = (request.query_params.get("corpus") or "").strip() or None
        try:
            depth = int(request.query_params.get("depth") or 2)
        except ValueError:
            depth = 2
        try:
            max_nodes = int(request.query_params.get("max_nodes") or 120)
        except ValueError:
            max_nodes = 120
        layers = (request.query_params.get("layers") or "concept").strip().lower()
        from utils.knowledge_graph import export_knowledge_subgraph

        try:
            content = await _read_value_from_thread(
                export_knowledge_subgraph,
                focus,
                corpus_slug=corpus,
                depth=depth,
                max_nodes=max_nodes,
                layers=layers,
            )
        except _KnowledgeReadFailed as exc:
            return _read_error_json(exc)
        if not content.get("ok"):
            return _knowledge_graph_mutation_response(content)
        return JSONResponse(content=content)

    @fastapi_app.get("/knowledge/graph/suggestions")
    async def knowledge_graph_suggestions(request: Request):
        corpus_raw = (request.query_params.get("corpus") or "").strip()
        corpus = _slug_from_param(corpus_raw) if corpus_raw else None
        focus = (request.query_params.get("focus") or "").strip() or None
        try:
            limit = int(request.query_params.get("limit") or 50)
        except ValueError:
            limit = 50
        try:
            offset = int(request.query_params.get("offset") or 0)
        except ValueError:
            offset = 0
        from utils.knowledge_graph import list_knowledge_graph_suggestions

        try:
            content = await _read_value_from_thread(
                list_knowledge_graph_suggestions,
                corpus_slug=corpus,
                focus_node_id=focus,
                limit=limit,
                offset=offset,
            )
        except _KnowledgeReadFailed as exc:
            return _read_error_json(exc)
        if content.get("error") == "focus not in corpus":
            raise HTTPException(status_code=400, detail="focus not in corpus")
        return JSONResponse(content=content)

    @fastapi_app.get("/knowledge/graph/focus")
    async def knowledge_graph_focus(request: Request):
        corpus_raw = (request.query_params.get("corpus") or "").strip()
        if not corpus_raw:
            raise HTTPException(status_code=400, detail="corpus required")
        corpus = _slug_from_param(corpus_raw)
        source_id = (request.query_params.get("source_id") or "").strip() or None
        from utils.knowledge_graph import focus_node_for_source

        focus = None
        if source_id:
            try:
                focus = await _read_value_from_thread(focus_node_for_source, corpus, source_id)
            except _KnowledgeReadFailed as exc:
                return _read_error_json(exc)
        layers = (request.query_params.get("layers") or "concept").strip().lower()
        if not focus:
            if layers in ("concept", "concepts"):
                from utils.knowledge_graph import default_concept_focus_node

                try:
                    focus = await _read_value_from_thread(default_concept_focus_node, corpus)
                except _KnowledgeReadFailed as exc:
                    return _read_error_json(exc)
                focus = focus or f"knowledge:{corpus}:topic:hub"
            else:
                try:
                    focus = await _read_value_from_thread(focus_node_for_source, corpus, "synthesis")
                except _KnowledgeReadFailed as exc:
                    return _read_error_json(exc)
                focus = focus or f"knowledge:{corpus}:chunk:1"
        return JSONResponse(
            content={"schema": "mugi.knowledge.graph.focus.v1", "corpus": corpus, "focus": focus, "layers": layers}
        )

    @fastapi_app.get("/knowledge/graph/stats")
    async def knowledge_graph_stats(request: Request):
        corpus_raw = (request.query_params.get("corpus") or "").strip()
        if not corpus_raw:
            raise HTTPException(status_code=400, detail="corpus required")
        corpus = _slug_from_param(corpus_raw)
        from utils.knowledge_graph import knowledge_graph_stats as _kg_stats

        try:
            content = await _read_value_from_thread(_kg_stats, corpus_slug=corpus)
        except _KnowledgeReadFailed as exc:
            return _read_error_json(exc)
        if not content.get("ok"):
            raise HTTPException(status_code=400, detail=str(content.get("error") or "stats failed"))
        return JSONResponse(content={**content, "schema": "mugi.knowledge.graph.stats.v1"})

    @fastapi_app.get("/knowledge/graph/neighbors")
    async def knowledge_graph_neighbors(request: Request):
        node_id = (request.query_params.get("node_id") or "").strip()
        if not node_id:
            raise HTTPException(status_code=400, detail="node_id required")
        corpus = (request.query_params.get("corpus") or "").strip() or None
        try:
            limit = int(request.query_params.get("limit") or 12)
        except ValueError:
            limit = 12
        include_suggested = str(request.query_params.get("include_suggested") or "").lower() in (
            "1",
            "true",
            "yes",
        )
        from utils.knowledge_graph import list_knowledge_graph_neighbors

        try:
            rows = await _read_value_from_thread(
                list_knowledge_graph_neighbors,
                node_id,
                corpus_slug=corpus,
                limit=limit,
                include_suggested=include_suggested,
            )
        except _KnowledgeReadFailed as exc:
            return _read_error_json(exc)
        return JSONResponse(
            content={"schema": "mugi.knowledge.graph.neighbors.v1", "node_id": node_id, "neighbors": rows}
        )

    @fastapi_app.post("/knowledge/graph/reindex")
    async def knowledge_graph_reindex(request: Request):
        if not _allow_graph_refresh():
            raise HTTPException(status_code=429, detail="Knowledge graph reindex rate limited")
        try:
            body = await request.json()
        except Exception:
            body = {}
        slug_raw = str(
            body.get("corpus")
            or body.get("slug")
            or request.query_params.get("corpus")
            or ""
        ).strip()
        if not slug_raw:
            raise HTTPException(status_code=400, detail="corpus required")
        slug = _slug_from_param(slug_raw)
        from utils.knowledge_graph import refresh_knowledge_graph_suggestions

        content = await asyncio.to_thread(refresh_knowledge_graph_suggestions, slug)
        return _knowledge_graph_mutation_response(content)

    @fastapi_app.post("/knowledge/graph/suggestions/refresh")
    async def knowledge_graph_suggestions_refresh(request: Request):
        if not _allow_graph_refresh():
            raise HTTPException(status_code=429, detail="Knowledge graph refresh rate limited")
        try:
            body = await request.json()
        except Exception:
            body = {}
        slug_raw = str(
            body.get("corpus")
            or body.get("slug")
            or request.query_params.get("corpus")
            or ""
        ).strip()
        if not slug_raw:
            raise HTTPException(status_code=400, detail="corpus required")
        slug = _slug_from_param(slug_raw)
        from utils.knowledge_graph import refresh_knowledge_graph_suggestions

        content = await asyncio.to_thread(refresh_knowledge_graph_suggestions, slug)
        return _knowledge_graph_mutation_response(content)

    @fastapi_app.post("/knowledge/graph/suggestions/dismiss_all")
    async def knowledge_graph_suggestions_dismiss_all(request: Request):
        if not _allow_graph_refresh():
            raise HTTPException(status_code=429, detail="Knowledge graph bulk dismiss rate limited")
        try:
            body = await request.json()
        except Exception:
            body = {}
        slug_raw = str(body.get("corpus") or body.get("slug") or "").strip()
        if not slug_raw:
            raise HTTPException(status_code=400, detail="corpus required")
        slug = _slug_from_param(slug_raw)
        from utils.knowledge_graph import dismiss_all_knowledge_graph_suggestions

        content = await asyncio.to_thread(dismiss_all_knowledge_graph_suggestions, corpus_slug=slug)
        resp = _knowledge_graph_mutation_response(content)
        if resp.status_code == 200:
            body = content if isinstance(content, dict) else {}
            return JSONResponse(
                content={**body, "schema": "mugi.knowledge.graph.dismiss_all.v1"},
                status_code=200,
            )
        return resp

    @fastapi_app.post("/knowledge/graph/suggestions/promote")
    async def knowledge_graph_suggestions_promote(request: Request):
        if not _allow_graph_refresh():
            raise HTTPException(status_code=429, detail="Knowledge graph bulk promote rate limited")
        try:
            body = await request.json()
        except Exception:
            body = {}
        slug_raw = str(body.get("corpus") or body.get("slug") or "").strip()
        if not slug_raw:
            raise HTTPException(status_code=400, detail="corpus required")
        slug = _slug_from_param(slug_raw)
        focus = str(body.get("focus") or body.get("focus_node_id") or "").strip() or None
        min_conf = body.get("min_confidence")
        try:
            min_confidence = float(min_conf) if min_conf is not None else None
        except (TypeError, ValueError):
            min_confidence = None
        from utils.knowledge_graph import promote_knowledge_graph_suggestions

        content = await asyncio.to_thread(
            promote_knowledge_graph_suggestions,
            corpus_slug=slug,
            min_confidence=min_confidence,
            focus_node_id=focus,
        )
        if not content.get("ok"):
            raise HTTPException(status_code=400, detail=str(content.get("error") or "promote failed"))
        return JSONResponse(content={**content, "schema": "mugi.knowledge.graph.promote.v1"})

    @fastapi_app.post("/knowledge/graph/edges/{edge_id}/confirm")
    async def knowledge_graph_edge_confirm(edge_id: str):
        from utils.knowledge_graph import confirm_knowledge_graph_edge

        content = await asyncio.to_thread(confirm_knowledge_graph_edge, edge_id)
        return _knowledge_graph_edge_response(content)

    @fastapi_app.post("/knowledge/graph/edges/{edge_id}/reject")
    async def knowledge_graph_edge_reject(edge_id: str):
        from utils.knowledge_graph import reject_knowledge_graph_edge

        content = await asyncio.to_thread(reject_knowledge_graph_edge, edge_id)
        return _knowledge_graph_edge_response(content)

    @fastapi_app.get("/knowledge/graph/passage")
    async def knowledge_graph_passage(request: Request):
        if not _allow_search("search"):
            raise HTTPException(status_code=429, detail="Knowledge graph passage rate limit exceeded")
        node_id = _bounded_connections_param(request.query_params.get("node_id"), "node_id")
        if not node_id:
            raise HTTPException(status_code=400, detail="node_id required")
        from utils.knowledge_graph import get_knowledge_graph_node_passage

        try:
            content = await _read_value_from_thread(get_knowledge_graph_node_passage, node_id)
        except _KnowledgeReadFailed as exc:
            return _read_error_json(exc)
        if not content.get("ok"):
            return JSONResponse(
                status_code=404,
                content={**content, "schema": "mugi.knowledge.graph.passage.v1"},
            )
        return JSONResponse(content={**content, "schema": "mugi.knowledge.graph.passage.v1"})

    @fastapi_app.get("/knowledge/graph/gitnexus_hint")
    async def knowledge_graph_gitnexus_hint(request: Request):
        node_id = (request.query_params.get("node_id") or "").strip()
        from utils.knowledge_graph import gitnexus_query_hint_for_node

        try:
            hint = await _read_value_from_thread(gitnexus_query_hint_for_node, node_id)
        except _KnowledgeReadFailed as exc:
            return _read_error_json(exc)
        return JSONResponse(content={"node_id": node_id, "hint": hint})

    @fastapi_app.get("/knowledge/embedding_reindex/status")
    async def knowledge_embedding_reindex_status():
        from utils.embedding_model import get_embedding_reindex_status

        try:
            content = await _read_value_from_thread(get_embedding_reindex_status)
        except _KnowledgeReadFailed as exc:
            return _read_error_json(exc)
        return JSONResponse(content=content)

    @fastapi_app.post("/knowledge/embeddings/reindex")
    async def knowledge_embeddings_reindex(request: Request):
        if not _allow_search(bucket="connections_rebuild"):
            raise HTTPException(status_code=429, detail="Embedding reindex rate limited")
        try:
            body = await request.json()
        except Exception:
            body = {}
        slug_raw = str(
            body.get("corpus")
            or body.get("slug")
            or request.query_params.get("corpus")
            or ""
        ).strip()
        from utils.embedding_model import request_embedding_reindex, startup_embedding_identity_sweep

        if slug_raw:
            slug = _slug_from_param(slug_raw)
            content = await asyncio.to_thread(request_embedding_reindex, slug)
            if not content.get("ok"):
                return JSONResponse(content=content, status_code=400)
            return JSONResponse(content=content)
        force_raw = body.get("force") or request.query_params.get("force") or ""
        force = str(force_raw).strip().lower() in ("1", "true", "yes")
        await asyncio.to_thread(startup_embedding_identity_sweep, force=force)
        return JSONResponse(content={"ok": True, "scheduled": "startup_sweep", "force": force})
