"""MCP server skeleton — fill in PRODUCT_NAME and the tool bodies.

This file is the template's structural anchor. The phases described in
PLAN.md add or extend pieces of this file:

  Phase 3  — search_docs, get_page, list_versions stubs (you are here)
  Phase 6  — reranker integration in search_docs
  Phase 8  — BM25 + hybrid retrieval (HYBRID_SEARCH env gate, _rrf_fuse)
  Phase 9  — diff_versions, list_cluster, bundle_changelog
  Phase 10 — TimedCall wiring (already imported below)
  Phase 11 — <product>_api_lessons tool
  Phase 12 — find_doc_inconsistencies
  Phase 13 — weekly_digest + _digest_history reader

Every stub below has a docstring + `raise NotImplementedError`. Replace
the body when you reach the corresponding phase. Keep the signatures
stable across products — clients depend on them.
"""
from __future__ import annotations

import datetime as _dt
import difflib
import json
import logging
import os
import re
from pathlib import Path
from typing import Annotated

from mcp.server.fastmcp import FastMCP
from pydantic import Field

from .usage import TimedCall

log = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Product-specific configuration. Set these for each new build.
# ---------------------------------------------------------------------------
PRODUCT_NAME = os.environ.get("PRODUCT_NAME", "hvm")
PRODUCT_DOCS_URL = os.environ.get(
    "PRODUCT_DOCS_URL",
    "https://support.hpe.com/hpesc/public/docDisplay?docId=sd00007735en_us",
)
COLLECTION = f"{PRODUCT_NAME}_docs"

# Paths inside the deployed container (and matching layout locally for dev).
ROOT = Path(__file__).resolve().parent.parent
CORPUS = ROOT / "corpus"
CHROMA_DIR = ROOT / "chroma"
BM25_DB = Path(os.environ.get("BM25_DB", str(ROOT / "bm25" / f"{PRODUCT_NAME}_docs.db")))
BUNDLES_JSON = ROOT / "bundles.json"
DIGEST_HISTORY_PATH = CORPUS / ".digest" / "history.jsonl"
API_LESSONS_MD = Path(__file__).resolve().parent / "api_lessons.md"

# ---------------------------------------------------------------------------
# Feature flags (Phase 6 / 8 / 12 enable these as you ship each phase).
# ---------------------------------------------------------------------------
RERANK_URL = os.environ.get("RERANK_URL", "").rstrip("/") or None
RERANK_POOL = int(os.environ.get("RERANK_POOL", "50"))
RERANK_TIMEOUT = float(os.environ.get("RERANK_TIMEOUT", "30"))

HYBRID_SEARCH = os.environ.get("HYBRID_SEARCH", "").lower() in ("true", "1", "yes", "on")
RRF_K = int(os.environ.get("RRF_K", "60"))


# ---------------------------------------------------------------------------
# FastMCP setup.
#
# stateless_http=True — every request creates an ephemeral session and
# discards it on return. Critical for production: clients don't get
# 404 storms when the container is recreated by Watchtower.
# ---------------------------------------------------------------------------
mcp = FastMCP(f"{PRODUCT_NAME}-docs", stateless_http=True)


# ---------------------------------------------------------------------------
# Lazy helpers — instantiate expensive things only when actually needed,
# so the server still starts when (e.g.) Ollama is briefly unreachable.
# ---------------------------------------------------------------------------

def _bundles() -> dict[str, dict]:
    """Cached load of bundles.json into a {slug: bundle_dict} mapping.

    bundles.json is the product-specific catalog written by the Phase 1
    scraper. See PLAN.md Phase 1 for the schema.
    """
    if not BUNDLES_JSON.exists():
        return {}
    cat = json.loads(BUNDLES_JSON.read_text())
    return {b["slug"]: b for b in cat}


def _build_where(version: str | None, platform: str | None, bundle_id: str | None) -> dict | None:
    """Translate filter args into a Chroma `where` clause."""
    conds: list[dict] = []
    if version:
        conds.append({"version": version})
    if platform:
        conds.append({"platform": platform})
    if bundle_id:
        conds.append({"bundle_id": bundle_id})
    if not conds:
        return None
    if len(conds) == 1:
        return conds[0]
    return {"$and": conds}


def _where_for_bm25(version: str | None, platform: str | None, bundle_id: str | None) -> dict | None:
    """BM25Index.query takes a flat dict of equality filters."""
    w: dict[str, str] = {}
    if version: w["version"] = version
    if platform: w["platform"] = platform
    if bundle_id: w["bundle_id"] = bundle_id
    return w or None


def _read_page(bundle_id: str, page_id: str) -> tuple[str, dict] | None:
    """Read a corpus page off disk. Returns (markdown_body, metadata_dict)."""
    md_path = CORPUS / bundle_id / (page_id + ".md")
    json_path = CORPUS / bundle_id / (page_id + ".json")
    if not md_path.exists() or not json_path.exists():
        return None
    return md_path.read_text(), json.loads(json_path.read_text())


_CHROMA = None
_BM25 = None


def _collection():
    """Lazy Chroma collection handle. Cached after first call."""
    global _CHROMA
    if _CHROMA is None:
        import chromadb
        from chromadb.config import Settings
        from rag.embeddings import embedding_function

        client = chromadb.PersistentClient(
            path=str(CHROMA_DIR),
            settings=Settings(anonymized_telemetry=False),
        )
        _CHROMA = client.get_collection(COLLECTION, embedding_function=embedding_function())
    return _CHROMA


def _bm25():
    """Lazy BM25Index handle. None if the FTS5 db isn't built."""
    global _BM25
    if _BM25 is None:
        if not BM25_DB.exists():
            return None
        try:
            from rag.bm25 import BM25Index
            _BM25 = BM25Index(str(BM25_DB))
        except Exception as e:  # defensive: hybrid must never block dense
            log.warning("BM25 unavailable, falling back to dense-only: %s", e)
            return None
    return _BM25


def _enrich_from_chroma(col, chunk_ids: list[str], fused: list | None) -> tuple[list[str], list[dict], list[float]]:
    """Fetch document text + metadata for a list of chunk ids from Chroma, in order."""
    if not chunk_ids:
        return [], [], []
    g = col.get(ids=chunk_ids, include=["documents", "metadatas"])
    by_id = {i: (d, m) for i, d, m in zip(g["ids"], g["documents"], g["metadatas"])}
    docs = [by_id[i][0] for i in chunk_ids if i in by_id]
    metas = [by_id[i][1] for i in chunk_ids if i in by_id]
    if fused is not None:
        dists = [1.0 - score for _id, score, _src in fused[:len(docs)]]
    else:
        dists = [0.0] * len(docs)
    return docs, metas, dists


def _rerank(query: str, candidates: list[tuple[str, str]]) -> list[tuple[str, str]] | None:
    """POST to RERANK_URL /v1/rerank, return candidates re-ordered by relevance.

    `candidates` is `[(chunk_id, text), ...]`. Texts are truncated to ~2000 chars
    before sending so we never blow past jina-reranker's 1024-token per-pair
    cap (which 400s the entire batch). The full untruncated text still goes
    back to the user from Chroma; truncation is reranking-only.

    Returns None on any failure — caller treats that as "skip reranking,
    keep retrieval-order candidates."
    """
    if not RERANK_URL or not candidates:
        return None
    try:
        import httpx
        payload = {
            "query": query,
            "documents": [(text or "")[:2000] for _cid, text in candidates],
            "top_n": len(candidates),
        }
        with httpx.Client(timeout=RERANK_TIMEOUT) as c:
            r = c.post(f"{RERANK_URL}/v1/rerank", json=payload)
            r.raise_for_status()
            results = r.json().get("results") or []
        order = [candidates[item["index"]] for item in results
                 if isinstance(item.get("index"), int) and 0 <= item["index"] < len(candidates)]
        return order or None
    except Exception as e:
        log.warning("rerank failed, keeping retrieval order: %s", e)
        return None


def _rrf_fuse(*ranked_lists: list[str], k: int = RRF_K) -> list[tuple[str, float, dict]]:
    """Reciprocal Rank Fusion. Each ranked list is a sequence of ids in
    descending relevance. Returns [(id, fused_score, per_retriever_contrib), ...]
    sorted by score desc."""
    scores: dict[str, float] = {}
    sources: dict[str, dict] = {}
    names = ("dense", "bm25", "extra")
    for idx, lst in enumerate(ranked_lists):
        src = names[idx] if idx < len(names) else f"r{idx}"
        for rank, ident in enumerate(lst, start=1):
            scores[ident] = scores.get(ident, 0.0) + 1.0 / (k + rank)
            sources.setdefault(ident, {})[src] = rank
    ranked = sorted(scores.items(), key=lambda kv: -kv[1])
    return [(ident, score, sources[ident]) for ident, score in ranked]


def _source_url(bundle_id: str, page_id: str) -> str:
    """Build the canonical docs portal URL for a (bundle, page) pair."""
    b = _bundles().get(bundle_id)
    if not b:
        return ""
    doc_id = b.get("doc_id", "")
    if page_id.startswith("GUID-"):
        return f"https://support.hpe.com/hpesc/public/docDisplay?docId={doc_id}&page={page_id}.html"
    return f"https://support.hpe.com/hpesc/public/docDisplay?docId={doc_id}"


# ===========================================================================
# Tools
# ===========================================================================

@mcp.tool()
def search_docs(
    query: Annotated[str, Field(description=f"Natural-language query about {PRODUCT_NAME}.")],
    version: Annotated[
        str | None,
        Field(description="OPTIONAL version filter — restrict to one product version."),
    ] = None,
    platform: Annotated[
        str | None,
        Field(description="OPTIONAL platform filter. Set to one of the platforms listed by list_versions(); omit for all platforms."),
    ] = None,
    bundle_id: Annotated[
        str | None,
        Field(description="OPTIONAL bundle filter — pin to a specific doc bundle slug."),
    ] = None,
    k: Annotated[int, Field(description="Number of results to return.", ge=1, le=50)] = 10,
) -> str:
    """Search the HPE Morpheus VM Essentials (HVM) docs corpus.

    Returns the top-k most relevant chunks (with full source page URLs)
    given a natural-language query. Optional filters narrow the search
    to one version, one platform, or one bundle. Use list_versions()
    first if you need to discover the available facet values.

    Call this tool whenever the user asks anything that should be
    answerable from the official product documentation — install,
    upgrade, configuration, backups, networking, HVM clusters, the
    Morpheus UI, or any 8.1.x release-notes question.
    """
    with TimedCall("search_docs", {
        "query": query, "version": version, "platform": platform,
        "bundle_id": bundle_id, "k": k,
    }) as _call:
        try:
            col = _collection()
        except Exception as e:
            log.exception("chroma collection unavailable")
            _call.set(hits_returned=0, error=str(e))
            return f"_(search backend unavailable: {e})_"

        where = _build_where(version, platform, bundle_id)
        bm25_where = _where_for_bm25(version, platform, bundle_id)
        pool = max(k * 5, 50)

        # Retrieval mode selection. Eval on this corpus (2026-05-22, 22 golden
        # queries) showed BM25 MRR=0.88 vs dense MRR=0.54 vs hybrid MRR=0.69 —
        # HPE structured docs use controlled vocabulary, so lexical match wins.
        # Dense is kept as fallback when BM25 has no tokens to chew on (e.g.
        # purely stopword queries). HYBRID_SEARCH=true forces RRF fusion.
        bm = _bm25()
        docs: list[str] = []
        metas: list[dict] = []
        dists: list[float] = []
        retrieval_mode = "dense"
        top1_source = "dense_only"

        if HYBRID_SEARCH and bm is not None:
            try:
                dense_res = col.query(query_texts=[query], n_results=pool, where=where)
                dense_ids = (dense_res.get("ids") or [[]])[0]
                bm_hits = bm.query(query, n=pool, where=bm25_where)
                bm_ids = [cid for cid, _s in bm_hits]
                fused = _rrf_fuse(dense_ids, bm_ids)
                docs, metas, dists = _enrich_from_chroma(col, [c for c, _, _ in fused[:k]], fused)
                if fused:
                    src0 = fused[0][2]
                    top1_source = ("both" if {"dense", "bm25"} <= set(src0)
                                   else "bm25_only" if "bm25" in src0
                                   else "dense_only")
                retrieval_mode = "hybrid"
            except Exception as e:
                log.warning("hybrid failed, falling back to BM25→dense: %s", e)

        if not docs and bm is not None:
            try:
                bm_hits = bm.query(query, n=k, where=bm25_where)
                if bm_hits:
                    ids = [cid for cid, _s in bm_hits[:k]]
                    docs, metas, _ = _enrich_from_chroma(col, ids, None)
                    # FTS5 returns negative scores (lower=better). Map onto a
                    # similarity-ish [0..1] just for display.
                    dists = [max(0.0, min(1.0, 1.0 - abs(s) / 20.0)) for _id, s in bm_hits[:k]]
                    retrieval_mode = "bm25"
                    top1_source = "bm25_only"
            except Exception as e:
                log.warning("BM25 retrieval failed, falling back to dense: %s", e)

        if not docs:
            res = col.query(query_texts=[query], n_results=k, where=where)
            docs = (res.get("documents") or [[]])[0]
            metas = (res.get("metadatas") or [[]])[0]
            dists = (res.get("distances") or [[]])[0]

        reranker_fired = False
        if RERANK_URL and docs:
            # Pull a deeper pool to give the reranker something to chew on.
            # We over-fetch up to RERANK_POOL chunks from whichever retriever
            # already won, then ask the reranker to pick the final top-k.
            pool_size = max(k, RERANK_POOL)
            if len(docs) < pool_size:
                if retrieval_mode == "bm25":
                    extra = bm.query(query, n=pool_size, where=bm25_where) if bm else []
                    extra_ids = [cid for cid, _s in extra]
                else:
                    extra_res = col.query(query_texts=[query], n_results=pool_size, where=where)
                    extra_ids = (extra_res.get("ids") or [[]])[0]
                if extra_ids:
                    d2, m2, _ = _enrich_from_chroma(col, extra_ids, None)
                    docs, metas = d2, m2
                    dists = [0.0] * len(docs)
            # Reranker scores chunk_ids — collapse to (id, text) tuples
            pairs = list(zip(
                [f"{m.get('bundle_id','')}::{m.get('page_id','')}::{m.get('ordinal',0)}" for m in metas],
                docs,
            ))
            reranked = _rerank(query, pairs)
            if reranked is not None:
                # Re-sort docs/metas to match. Recompute distances as descending
                # ordinal ranks so display still shows a useful score.
                by_cid = {p[0]: i for i, p in enumerate(pairs)}
                order = [by_cid[cid] for cid, _t in reranked if cid in by_cid]
                docs = [docs[i] for i in order][:k]
                metas = [metas[i] for i in order][:k]
                dists = [1.0 - (rank / len(reranked)) for rank, _ in enumerate(reranked)][:len(docs)]
                reranker_fired = True
            else:
                docs, metas, dists = docs[:k], metas[:k], dists[:k]

        _call.set(hits_returned=len(docs), retrieval_mode=retrieval_mode,
                  top1_source=top1_source, reranker_fired=reranker_fired)
        if not docs:
            return f"_No matches for `{query}`._"

        out = [f"# {len(docs)} result(s) for `{query}`", ""]
        for doc, meta, dist in zip(docs, metas, dists):
            bid = meta.get("bundle_id", "")
            pid = meta.get("page_id", "")
            title = meta.get("title") or pid
            ver = meta.get("version") or ""
            url = _source_url(bid, pid)
            header = f"## {title}"
            if ver:
                header += f"  _(v{ver})_"
            out.append(header)
            out.append(f"[{bid}/{pid}]({url}) · score={1 - dist:.3f}")
            out.append("")
            out.append(doc.strip())
            out.append("")
        return "\n".join(out)


@mcp.tool()
def get_page(
    bundle_id: Annotated[str, Field(description="Bundle slug.")],
    page_id: Annotated[str, Field(description="Page filename within the bundle.")],
) -> str:
    """Return the full markdown for one page, plus a metadata header.

    Use after search_docs surfaces a relevant page and the user (or you)
    want the complete text — not just the matched chunks.
    """
    with TimedCall("get_page", {"bundle_id": bundle_id, "page_id": page_id}) as _call:
        data = _read_page(bundle_id, page_id)
        if data is None:
            _call.set(found=False)
            return f"Page not found: {bundle_id}/{page_id}"
        md, meta = data
        _call.set(found=True, page_chars=len(md))
        title = meta.get("title") or page_id
        ver = meta.get("version")
        parent = meta.get("parent_title")
        url = _source_url(bundle_id, page_id)
        header = [f"# {title}"]
        ctx = []
        if ver:
            ctx.append(f"version **{ver}**")
        if parent:
            ctx.append(f"in **{parent}**")
        if ctx:
            header.append("_" + " · ".join(ctx) + "_")
        header.append(f"[source]({url})")
        header.append("")
        return "\n".join(header) + "\n" + md


@mcp.tool()
def list_versions() -> str:
    """List the available version/platform facets across all bundles.

    Use this to discover valid filter values for search_docs.
    """
    with TimedCall("list_versions", {}) as _call:
        cat = _bundles()
        if not cat:
            return "_(no bundles indexed yet — run the scraper + indexer)_"
        versions = sorted({b.get("version") for b in cat.values() if b.get("version")})
        platforms = sorted({b.get("platform") for b in cat.values() if b.get("platform")})
        _call.set(versions=len(versions), platforms=len(platforms))
        products = sorted({b.get("product") for b in cat.values() if b.get("product")})
        lines = [f"# Facets across {len(cat)} bundle(s)", ""]
        if versions:
            lines += ["## Versions", ""] + [f"- `{v}`" for v in versions] + [""]
        if platforms:
            lines += ["## Platforms", ""] + [f"- `{p}`" for p in platforms] + [""]
        if products:
            lines += ["## Product / doc types", ""] + [f"- {p}" for p in products] + [""]
        lines += ["## Bundles", ""]
        for slug in sorted(cat):
            b = cat[slug]
            kind = b.get("product") or ""
            ver = b.get("version")
            pages = b.get("page_count", "?")
            label = f"{kind} {ver}".strip() if ver else kind
            lines.append(f"- `{slug}` — {label} ({pages} pages)")
        return "\n".join(lines)


# ===========================================================================
# Phase 9 — cross-version tools
# ===========================================================================

def _bundle_pages(bundle_id: str) -> set[str]:
    """Page IDs (= GUID-XXXX) on disk in a bundle. Mirrors rag.index's md_path.stem."""
    bd = CORPUS / bundle_id
    if not bd.is_dir():
        return set()
    return {p.stem for p in bd.glob("*.md")}


def _diff_churn(a: str, b: str) -> tuple[int, int]:
    """Cheap (added, removed) line counts for a pair of markdown bodies."""
    diff = difflib.unified_diff(a.splitlines(keepends=False),
                                b.splitlines(keepends=False), n=0)
    added = removed = 0
    for line in diff:
        if line.startswith(("+++", "---", "@@")):
            continue
        if line.startswith("+"):
            added += 1
        elif line.startswith("-"):
            removed += 1
    return added, removed


@mcp.tool()
def list_cluster(
    bundle_id: Annotated[str, Field(description="Bundle slug of the source topic.")],
    page_id: Annotated[str, Field(description="Page id (GUID-XXXX) of the source topic.")],
) -> str:
    """List cross-version peers of a topic in the HVM docs.

    HPE re-mints the docId per product version but keeps page GUIDs stable,
    so the scrape pipeline synthesizes `topic_cluster.clustered_topics`
    from same-GUID overlap (374/376/376 pages overlap across 8.1.0/.1/.2).
    """
    with TimedCall("list_cluster", {"bundle_id": bundle_id, "page_id": page_id}) as _call:
        out = _read_page(bundle_id, page_id)
        if out is None:
            _call.set(found=False)
            return f"Not found: {bundle_id}/{page_id}"
        _, side = out
        cluster = side.get("topic_cluster") or {}
        peers = cluster.get("clustered_topics") or []
        _call.set(hits_returned=len(peers))
        src_label = cluster.get("clustering_title") or side.get("title") or page_id
        lines = [f"# Cluster for {bundle_id}/{page_id}  ({src_label})", ""]
        if not peers:
            lines.append("_No peer topics in cluster._")
            return "\n".join(lines)
        for p in peers:
            lines.append(f"- `{p['bundle_id']}/{p['page_id']}`  —  {p.get('clustering_title') or ''}")
        return "\n".join(lines)


@mcp.tool()
def diff_versions(
    bundle_id: Annotated[str, Field(description="Bundle slug of the source topic (the 'new' side).")],
    page_id: Annotated[str, Field(description="Page id of the source topic.")],
    against_bundle_id: Annotated[str, Field(description="Bundle slug to diff against. Must be in the source's cluster, or share the same page_id.")],
    context: Annotated[int, Field(description="Lines of context around each hunk.", ge=0, le=10)] = 3,
) -> str:
    """Unified diff of one topic between two bundles (typically two HVM versions).

    Two matching strategies, tried in order:

      1. `topic_cluster` peer (synthesized from same-GUID overlap by the scraper).
      2. Same `page_id` fallback (works because GUIDs are stable across HVM versions).
    """
    with TimedCall("diff_versions", {
        "bundle_id": bundle_id, "page_id": page_id,
        "against_bundle_id": against_bundle_id, "context": context,
    }) as _call:
        src = _read_page(bundle_id, page_id)
        if src is None:
            _call.set(matched_via=None, reason="source_not_found")
            return f"Source not found: {bundle_id}/{page_id}"
        src_md, side = src
        cluster = side.get("topic_cluster") or {}
        peers = {p["bundle_id"]: p for p in (cluster.get("clustered_topics") or [])}

        peer = peers.get(against_bundle_id)
        if peer is not None:
            peer_page_id = peer["page_id"]
            matched_via = "topic_cluster"
        elif _read_page(against_bundle_id, page_id) is not None:
            peer_page_id = page_id
            matched_via = "filename"
        else:
            _call.set(matched_via=None, reason="no_peer")
            valid = list(peers) or ["(no peers)"]
            return (f"No match for {bundle_id}/{page_id} in {against_bundle_id}.\n"
                    f"- No cluster peer. Available peers: {valid}\n"
                    f"- No page {page_id!r} in {against_bundle_id} either.")

        _call.set(matched_via=matched_via)
        peer_data = _read_page(against_bundle_id, peer_page_id)
        if peer_data is None:
            return f"Peer not found in corpus: {against_bundle_id}/{peer_page_id}"
        peer_md, _ = peer_data
        diff = difflib.unified_diff(peer_md.splitlines(keepends=True),
                                     src_md.splitlines(keepends=True),
                                     fromfile=f"{against_bundle_id}/{peer_page_id}",
                                     tofile=f"{bundle_id}/{page_id}",
                                     n=context)
        body = "".join(diff)
        header = f"_matched via {matched_via}_\n\n"
        if not body.strip():
            return header + f"No differences between {bundle_id}/{page_id} and {against_bundle_id}/{peer_page_id}."
        return header + f"```diff\n{body}```"


@mcp.tool()
def bundle_changelog(
    bundle_id_new: Annotated[str, Field(description="New-side bundle slug, e.g. 'hvm_user_manual_8_1_2'.")],
    bundle_id_old: Annotated[str, Field(description="Old-side bundle slug, e.g. 'hvm_user_manual_8_1_1'.")],
    min_churn: Annotated[int, Field(description="Min (added + removed) lines to flag a page as changed.", ge=1, le=1000)] = 5,
    max_changed: Annotated[int, Field(description="Max changed pages to list (sorted by churn desc).", ge=1, le=500)] = 50,
) -> str:
    """High-level diff between two HVM bundles.

    Lists pages added, removed, and changed between an old bundle and a
    new one. Match is by page_id (which is the stable GUID — same GUID
    across versions = same topic). Use after `list_versions` to discover
    valid bundle slugs.
    """
    with TimedCall("bundle_changelog", {
        "bundle_id_new": bundle_id_new, "bundle_id_old": bundle_id_old,
        "min_churn": min_churn, "max_changed": max_changed,
    }) as _call:
        new_pages = _bundle_pages(bundle_id_new)
        old_pages = _bundle_pages(bundle_id_old)
        if not new_pages and not old_pages:
            _call.set(reason="both_empty")
            return f"Neither bundle has pages on disk: {bundle_id_new}, {bundle_id_old}"
        if not new_pages:
            return f"Bundle not found or empty: {bundle_id_new}"
        if not old_pages:
            return f"Bundle not found or empty: {bundle_id_old}"

        added = sorted(new_pages - old_pages)
        removed = sorted(old_pages - new_pages)
        common = sorted(new_pages & old_pages)

        changed: list[tuple[str, int, int]] = []
        for pid in common:
            n = _read_page(bundle_id_new, pid)
            o = _read_page(bundle_id_old, pid)
            if n is None or o is None:
                continue
            a_lines, r_lines = _diff_churn(o[0], n[0])
            if a_lines + r_lines >= min_churn:
                changed.append((pid, a_lines, r_lines))
        changed.sort(key=lambda t: -(t[1] + t[2]))
        _call.set(added=len(added), removed=len(removed),
                  changed=len(changed), unchanged=len(common) - len(changed))

        lines = [
            f"# Bundle changelog: {bundle_id_new} vs {bundle_id_old}", "",
            f"- pages in new: **{len(new_pages)}**",
            f"- pages in old: **{len(old_pages)}**",
            f"- common: **{len(common)}**",
            f"- **added** (in new only): {len(added)}",
            f"- **removed** (in old only): {len(removed)}",
            f"- **changed** (≥{min_churn} lines): {len(changed)} of {len(common)} common",
            f"- unchanged: {len(common) - len(changed)}", "",
        ]
        if added:
            lines += [f"## Added pages ({len(added)})", *(f"- `{p}`" for p in added), ""]
        if removed:
            lines += [f"## Removed pages ({len(removed)})", *(f"- `{p}`" for p in removed), ""]
        if changed:
            shown = changed[:max_changed]
            lines += [
                f"## Changed pages — top {len(shown)} of {len(changed)} by churn", "",
                "| page | +lines | -lines | total |", "|---|---|---|---|",
            ]
            for p, a, r in shown:
                lines.append(f"| `{p}` | +{a} | -{r} | {a + r} |")
            if len(changed) > max_changed:
                lines.append(f"\n_({len(changed) - max_changed} more changed pages omitted; raise `max_changed` to see them.)_")
            lines.append("\nInspect a specific page: `diff_versions(bundle_id_new, page_id, bundle_id_old)`.")
        return "\n".join(lines)


# ===========================================================================
# Phase 13 — weekly digest from corpus/.digest/history.jsonl (built in CI)
# ===========================================================================

_digest_cache: list[dict] | None = None


def _digest_history() -> list[dict]:
    """Lazy load of the digest history JSONL written by scrape.changelog at CI time."""
    global _digest_cache
    if _digest_cache is not None:
        return _digest_cache
    if not DIGEST_HISTORY_PATH.exists():
        log.warning("digest history not found at %s — weekly_digest will return empty.",
                    DIGEST_HISTORY_PATH)
        _digest_cache = []
        return _digest_cache
    records: list[dict] = []
    try:
        with open(DIGEST_HISTORY_PATH) as fh:
            for ln, line in enumerate(fh, start=1):
                line = line.strip()
                if not line:
                    continue
                try:
                    records.append(json.loads(line))
                except json.JSONDecodeError as e:
                    log.warning("digest history: skipping malformed line %d: %s", ln, e)
    except OSError as e:
        log.warning("digest history read failed: %s", e)
    _digest_cache = records
    return _digest_cache


@mcp.tool()
def weekly_digest(
    days: Annotated[int, Field(description="How far back to summarize. 7=last week, 30=last month. Horizon ~120 days.", ge=1, le=120)] = 7,
    version: Annotated[str | None, Field(description="OPTIONAL version filter, e.g. '8.1.2'.")] = None,
    platform: Annotated[str | None, Field(description="OPTIONAL platform filter (HVM bundles don't set platform — leave None).")] = None,
    max_bundles: Annotated[int, Field(description="Cap on per-bundle detail blocks.", ge=1, le=100)] = 25,
    max_pages_per_bundle: Annotated[int, Field(description="Pages to list per bundle.", ge=1, le=50)] = 10,
) -> str:
    """Summarize what changed in the HVM docs over the past N days.

    Call when the user asks *"what's new in HVM docs this week?"*,
    *"what changed in 8.1.2?"*, or *"is there anything new since the
    last release?"*. Reads the pre-baked digest history JSONL written
    by CI from git log over corpus-touching commits.
    """
    with TimedCall("weekly_digest", {
        "days": days, "version": version, "platform": platform,
        "max_bundles": max_bundles, "max_pages_per_bundle": max_pages_per_bundle,
    }) as _call:
        records = _digest_history()
        if not records:
            _call.set(returned="empty_no_history", record_count=0)
            return ("# Weekly digest\n\n"
                    f"_No digest history on this image. `{DIGEST_HISTORY_PATH}` is "
                    "missing — it's populated by the weekly refresh workflow._")

        now = _dt.datetime.now(_dt.timezone.utc)
        cutoff = now - _dt.timedelta(days=days)
        filtered: list[dict] = []
        for r in records:
            try:
                ts = _dt.datetime.fromisoformat(r["timestamp"])
            except (KeyError, ValueError):
                continue
            if ts.tzinfo is None:
                ts = ts.replace(tzinfo=_dt.timezone.utc)
            if ts >= cutoff:
                filtered.append({**r, "_ts": ts})

        if not filtered:
            _call.set(returned="empty_window", record_count=0)
            covers = ""
            if records:
                oldest = min(records, key=lambda r: r.get("timestamp", ""))
                newest = max(records, key=lambda r: r.get("timestamp", ""))
                covers = (f"\n\n_(History on this image covers "
                          f"{oldest.get('timestamp','?')[:10]} through "
                          f"{newest.get('timestamp','?')[:10]}.)_")
            return (f"# Weekly digest — last {days} day{'s' if days != 1 else ''}\n\n"
                    f"_No corpus changes recorded in this window._" + covers)

        cat = _bundles()
        def _passes(bid: str) -> bool:
            if not (version or platform):
                return True
            b = cat.get(bid)
            if b is None:
                return False
            if version and b.get("version") != version:
                return False
            if platform and b.get("platform") != platform:
                return False
            return True

        filtered.sort(key=lambda r: r["_ts"], reverse=True)
        per_bundle_pages: dict[str, list[str]] = {}
        new_bundles_set: set[str] = set()
        drift_bundles_set: set[str] = set()
        commits_in_window = 0
        for r in filtered:
            commits_in_window += 1
            for bid in r.get("new_bundles", []):
                if _passes(bid):
                    new_bundles_set.add(bid)
            for bid in r.get("json_only_bundles", []):
                if _passes(bid):
                    drift_bundles_set.add(bid)
            for bid, pages in (r.get("content_bundles") or {}).items():
                if not _passes(bid):
                    continue
                seen = set(per_bundle_pages.get(bid, []))
                fresh = [p for p in pages if p not in seen]
                if fresh:
                    per_bundle_pages.setdefault(bid, []).extend(fresh)

        total_md = sum(len(p) for p in per_bundle_pages.values())
        bundles_ranked = sorted(per_bundle_pages.items(), key=lambda kv: (-len(kv[1]), kv[0]))
        _call.set(returned="ok", record_count=commits_in_window,
                  bundles_changed=len(per_bundle_pages),
                  new_bundles=len(new_bundles_set))

        ts_oldest = filtered[-1]["_ts"].date().isoformat()
        ts_newest = filtered[0]["_ts"].date().isoformat()
        lines = [
            f"# HVM docs digest — last {days} day{'s' if days != 1 else ''}", "",
            f"_Window: {ts_oldest} → {ts_newest}_  •  _Filters: version={version}, platform={platform}_", "",
            "## Headline", "",
            f"- **{total_md}** page change(s) across **{len(per_bundle_pages)}** bundle(s)",
            f"- **{commits_in_window}** corpus-touching commit(s) in this window",
            f"- **{len(new_bundles_set)}** bundle(s) newly added",
            f"- **{len(drift_bundles_set)}** bundle(s) with sidecar-only drift", "",
        ]
        if not per_bundle_pages and not new_bundles_set:
            lines.append(f"_No bundle changes matched the filter in this window._")
            return "\n".join(lines)
        if new_bundles_set:
            lines += ["## New bundles added", ""]
            for bid in sorted(new_bundles_set):
                b = cat.get(bid, {})
                t = b.get("title") or ""
                tag = f"  *({b.get('version') or '?'})*" if b.get("version") else ""
                lines.append(f"- `{bid}`{tag} {t}")
            lines.append("")
        if bundles_ranked:
            top = bundles_ranked[:max_bundles]
            remainder = len(bundles_ranked) - len(top)
            lines += [f"## Bundles with content changes — top {len(top)}" +
                      (f" of {len(bundles_ranked)}" if remainder else ""), ""]
            for bid, pages in top:
                b = cat.get(bid, {})
                tag = f"  *({b.get('version') or ''})*" if b.get("version") else ""
                lines.append(f"### `{bid}`{tag}")
                if b.get("title"):
                    lines.append(f"_{b['title']}_")
                lines.append(f"{len(pages)} page change(s).")
                for p in pages[:max_pages_per_bundle]:
                    lines.append(f"- `{p}`")
                if len(pages) > max_pages_per_bundle:
                    lines.append(f"  _(+{len(pages) - max_pages_per_bundle} more)_")
                lines.append("")
        lines.append("\nInspect a specific page: `get_page(bundle_id, page_id)` or `diff_versions(...)`.")
        return "\n".join(lines)


@mcp.tool()
def corpus_status() -> str:
    """Freshness + size of the knowledge base.

    Combines: (1) image build time (bundles.json mtime in container),
    (2) most-recent upstream Published date across bundles, (3) total
    bundles / pages / Chroma chunks.
    """
    lines: list[str] = ["# Corpus status", ""]
    try:
        ts = _dt.datetime.fromtimestamp(BUNDLES_JSON.stat().st_mtime, tz=_dt.timezone.utc).isoformat(timespec="seconds")
        lines.append(f"- This image built at: **{ts}**")
    except OSError:
        lines.append("- This image build time: _unknown_")

    cat = _bundles()
    latest_pub: str | None = None
    per_bundle: list[tuple[str, str]] = []
    for slug, b in cat.items():
        pub = (b.get("dates") or {}).get("Published")
        if pub:
            if latest_pub is None or pub > latest_pub:
                latest_pub = pub
            per_bundle.append((slug, pub))
    if latest_pub:
        lines.append(f"- Most-recent upstream Published date (any bundle): **{latest_pub}**")
    lines.append("")
    try:
        chunk_count = _collection().count()
    except Exception:
        chunk_count = -1
    pages_count = sum(1 for d in (CORPUS.iterdir() if CORPUS.exists() else [])
                       if d.is_dir() for _ in d.glob("*.md"))
    lines += [
        f"- Bundles indexed: **{len(cat)}**",
        f"- Pages in corpus: **{pages_count}**",
        f"- Chunks in Chroma: **{chunk_count}**" if chunk_count >= 0 else "- Chunks in Chroma: _(query failed)_",
        "",
    ]
    if per_bundle:
        per_bundle.sort(key=lambda kv: kv[1], reverse=True)
        lines.append("## Most-recently-edited bundles (by HPE)")
        for slug, when in per_bundle[:5]:
            b = cat.get(slug, {})
            lines.append(f"- `{slug}` — {b.get('title') or slug}  (published {when})")
    return "\n".join(lines)


# ===========================================================================
# Phase 11 — curated knowledge: hvm_api_lessons
# ===========================================================================

def _split_lessons_sections(md: str) -> list[tuple[str, str]]:
    sections: list[tuple[str, str]] = []
    current_title: str | None = None
    current_lines: list[str] = []
    for line in md.splitlines(keepends=True):
        m = re.match(r"^##\s+(.+?)\s*$", line)
        if m:
            if current_lines:
                sections.append((current_title or "(prelude)", "".join(current_lines)))
            current_title = m.group(1).strip()
            current_lines = [line]
        else:
            current_lines.append(line)
    if current_lines:
        sections.append((current_title or "(prelude)", "".join(current_lines)))
    return sections


@mcp.tool()
def hvm_api_lessons(
    topic: Annotated[str | None, Field(description="Optional keyword filter — returns only H2 sections whose heading or body contains this substring. Examples: 'manager', 'agent upgrade', 'plugin api', 'worker', 'console keyboard'. Omit for the full doc.")] = None,
) -> str:
    """Curated lessons about HPE Morpheus VM Essentials — non-obvious bits
    that aren't in the official docs and gotchas learned from real
    integration / operation work.

    **Call this proactively whenever the user asks you to:**
    - install, upgrade, or troubleshoot an HVM cluster or manager
    - integrate with HVM (REST API, automation, scripting)
    - upgrade across versions (8.1.0 → 8.1.1 → 8.1.2)
    - work with HVM Host agents
    - configure backups, networking, or storage
    - elevate to HPE Morpheus Enterprise

    With ``topic=...`` you'll get just the relevant H2 section(s). With
    no argument you'll get the full doc — usually the right call when
    starting on a new task since the TL;DR at the top primes the rest.
    """
    with TimedCall("hvm_api_lessons", {"topic": topic}) as _call:
        try:
            md = API_LESSONS_MD.read_text()
        except OSError as e:
            _call.set(error=str(e))
            return f"Lessons doc not present at {API_LESSONS_MD}: {e}"
        if not topic:
            _call.set(returned="full")
            return md
        needle = topic.lower()
        sections = _split_lessons_sections(md)
        kept: list[str] = []
        for title, body in sections:
            if needle in title.lower() or needle in body.lower():
                kept.append(body)
        if not kept:
            _call.set(returned="empty", topic_matched=False)
            return (f"_No sections matched topic={topic!r}. Returning the full document._\n\n" + md)
        _call.set(returned="filtered", sections_matched=len(kept))
        return f"_Filtered to {len(kept)} section(s) matching topic={topic!r}._\n\n" + "".join(kept)


# ===========================================================================
# Phase 12 — find_doc_inconsistencies
# ===========================================================================

_REDIRECT_PHRASE_RE = re.compile(
    r"\bsee\s+(?:the\s+)?[A-Z`\[][^.!?\n]{2,80}(?:for|topic|section|chapter|guide)\b",
    re.IGNORECASE,
)
_VERSION_SUFFIX_RE = re.compile(r"_(\d+_\d+_\d+)$")


def _bundle_family(bundle_id: str) -> str:
    """Strip a trailing `_X_Y_Z` version suffix from an HVM bundle slug.

    `hvm_user_manual_8_1_0` → `hvm_user_manual`
    `hvm_deployment_guide`  → `hvm_deployment_guide` (no version)

    Same-family bundles are version peers; cross-family pairs (User Manual
    vs Release Notes) are intentionally different content.
    """
    return _VERSION_SUFFIX_RE.sub("", bundle_id)


def _check_cross_version_drift(bundle_id: str, page_id: str, md: str, meta: dict) -> dict | None:
    cluster = (meta.get("topic_cluster") or {}).get("clustered_topics") or []
    if not cluster:
        return None
    src_family = _bundle_family(bundle_id)
    src_lines = max(1, len(md.splitlines()))
    in_band: list[tuple[int, str, str, int]] = []
    out_band: list[tuple[int, str, str, int]] = []
    for peer in cluster:
        peer_bid = peer.get("bundle_id")
        peer_pid = peer.get("page_id")
        if not (peer_bid and peer_pid) or peer_bid == bundle_id:
            continue
        if _bundle_family(peer_bid) != src_family:
            continue
        peer_data = _read_page(peer_bid, peer_pid)
        if peer_data is None:
            continue
        peer_md, _ = peer_data
        added, removed = _diff_churn(md, peer_md)
        churn = added + removed
        peer_lines = max(1, len(peer_md.splitlines()))
        denom = max(src_lines, peer_lines)
        pct = (churn * 100) // denom if denom else 0
        tup = (churn, peer_bid, peer_pid, peer_lines)
        if 10 <= pct <= 60:
            in_band.append(tup)
        elif churn >= 5:
            out_band.append(tup)
    if in_band:
        chosen = min(in_band, key=lambda t: t[0])
        confidence = "high"
    elif out_band:
        chosen = min(out_band, key=lambda t: t[0])
        confidence = "low"
    else:
        return None
    churn, peer_bid, peer_pid, peer_lines = chosen
    denom = max(src_lines, peer_lines)
    churn_pct = min(100, (churn * 100) // denom) if denom else 0
    return {
        "check": "cross_version_drift",
        "bundle_id": bundle_id, "page_id": page_id,
        "page_url": _source_url(bundle_id, page_id),
        "peer_bundle_id": peer_bid, "peer_page_id": peer_pid,
        "churn_lines": churn, "churn_pct_of_file": churn_pct,
        "confidence": confidence,
        "summary": (f"Drifts {churn} lines (~{churn_pct}% of file) vs peer "
                    f"{peer_bid}/{peer_pid}. Inspect with "
                    f"diff_versions({bundle_id!r}, {page_id!r}, {peer_bid!r})."),
    }


def _check_redirect_chain(bundle_id: str, page_id: str, md: str, meta: dict) -> dict | None:
    body = re.sub(r"^#[^\n]*\n", "", md, count=1).strip()
    if "```" in body:
        return None
    text_only = re.sub(r"[`\[\]()*_>#-]", "", body)
    text_only = re.sub(r"\s+", " ", text_only).strip()
    if len(text_only) > 600:
        return None
    redirect_matches = list(_REDIRECT_PHRASE_RE.finditer(body))
    if not redirect_matches:
        return None
    evidence = redirect_matches[0].group(0).strip()
    return {
        "check": "redirect_chain",
        "bundle_id": bundle_id, "page_id": page_id,
        "page_url": _source_url(bundle_id, page_id),
        "body_chars": len(text_only),
        "redirect_phrase": evidence[:200],
        "confidence": "medium",
        "summary": (f"Page is {len(text_only)} chars of body text with a "
                    f'"see ... for ..." redirect: "{evidence[:120]}". '
                    "Inspect with get_page to confirm."),
    }


@mcp.tool()
def find_doc_inconsistencies(
    scope_query: Annotated[str, Field(description="Natural-language scope describing what slice to scan. Used as a search to pick candidate pages. Examples: 'backup configuration', 'HVM cluster setup', 'VME manager installation'.")],
    version: Annotated[str | None, Field(description="OPTIONAL version filter — e.g. '8.1.2'.")] = None,
    platform: Annotated[str | None, Field(description="OPTIONAL platform filter (HVM bundles don't set platform — usually leave None).")] = None,
    bundle_id: Annotated[str | None, Field(description="OPTIONAL specific bundle slug to restrict scanning to.")] = None,
    max_pages: Annotated[int, Field(description="How many candidate pages to inspect.", ge=5, le=200)] = 30,
    checks: Annotated[list[str] | None, Field(description="Which checks to run. Available: 'cross_version_drift', 'redirect_chain'. Defaults to all.")] = None,
) -> str:
    """Scan a scoped set of HVM docs pages for likely documentation bugs.

    Surfaces concrete candidates for human review. Workflow:

      1. Run this against a focused scope.
      2. Review each finding; many will be false positives.
      3. For real bugs, drill in with `get_page` / `diff_versions`.
      4. Draft a bug report and file it through the docs portal's own
         feedback channel — this MCP does not submit upstream.
    """
    with TimedCall("find_doc_inconsistencies", {
        "scope_query": scope_query, "version": version, "platform": platform,
        "bundle_id": bundle_id, "max_pages": max_pages, "checks": checks,
    }) as _call:
        all_checks = {"cross_version_drift", "redirect_chain"}
        requested = all_checks if checks is None else {c for c in checks if c in all_checks}
        if not requested:
            _call.set(error="no_valid_checks")
            return f"No valid checks requested. Available: {sorted(all_checks)}."
        try:
            col = _collection()
        except Exception as e:
            _call.set(error=f"collection: {e}")
            return f"Couldn't open Chroma collection: {e}"
        where = _build_where(version, platform, bundle_id)
        try:
            res = col.query(query_texts=[scope_query], n_results=max_pages * 3,
                            where=where, include=["metadatas"])
        except Exception as e:
            _call.set(error=f"query: {e}")
            return f"Scope query failed: {e}"
        seen: set[tuple[str, str]] = set()
        candidates: list[tuple[str, str]] = []
        for meta in (res.get("metadatas") or [[]])[0]:
            key = (meta.get("bundle_id") or "", meta.get("page_id") or "")
            if not key[0] or not key[1] or key in seen:
                continue
            seen.add(key)
            candidates.append(key)
            if len(candidates) >= max_pages:
                break
        _call.set(pages_inspected=len(candidates), checks=sorted(requested))
        if not candidates:
            return f"No pages matched scope `{scope_query}`."
        findings: dict[str, list[dict]] = {c: [] for c in requested}
        for bid, pid in candidates:
            data = _read_page(bid, pid)
            if data is None:
                continue
            md, meta = data
            if "cross_version_drift" in requested:
                f = _check_cross_version_drift(bid, pid, md, meta)
                if f:
                    findings["cross_version_drift"].append(f)
            if "redirect_chain" in requested:
                f = _check_redirect_chain(bid, pid, md, meta)
                if f:
                    findings["redirect_chain"].append(f)
        findings["cross_version_drift"] = sorted(
            findings.get("cross_version_drift", []),
            key=lambda f: (-(1 if f["confidence"] == "high" else 0), -f["churn_lines"]))
        findings["redirect_chain"] = sorted(
            findings.get("redirect_chain", []), key=lambda f: f["body_chars"])
        total = sum(len(v) for v in findings.values())
        _call.set(findings_total=total,
                  findings_by_check={k: len(v) for k, v in findings.items()})
        lines = [
            f"# Doc inconsistency scan — {len(candidates)} pages inspected", "",
            f"_Scope_: `{scope_query}`  •  _Filters_: version={version}, platform={platform}, bundle_id={bundle_id}  •  _Checks_: {sorted(requested)}", "",
            f"**{total} candidate finding{'' if total == 1 else 's'}.** Review each individually. "
            "For real bugs, follow up with `get_page` / `diff_versions`, draft a report, "
            "and file it via the docs portal's own feedback channel.", "",
        ]
        if not total:
            lines.append("_No findings in this scope._")
            return "\n".join(lines)
        for check in sorted(requested):
            items = findings.get(check, [])
            lines += [f"## {check}  ({len(items)})", ""]
            if not items:
                lines.append("_No findings for this check._\n")
                continue
            for i, f in enumerate(items, 1):
                lines.append(f"### {i}. `{f['bundle_id']}/{f['page_id']}`  *({f['confidence']} confidence)*")
                lines.append(f"- URL: {f['page_url']}")
                lines.append(f"- {f['summary']}")
                if check == "cross_version_drift":
                    lines.append(f"- Peer: `{f['peer_bundle_id']}/{f['peer_page_id']}`  •  churn: {f['churn_lines']} lines ({f['churn_pct_of_file']}% of file)")
                elif check == "redirect_chain":
                    lines.append(f"- Body length: {f['body_chars']} chars  •  Phrase: *\"{f['redirect_phrase']}\"*")
                lines.append("")
        return "\n".join(lines)


# ===========================================================================
# Entry point
# ===========================================================================

def main() -> None:
    import argparse
    p = argparse.ArgumentParser(description=f"{PRODUCT_NAME} docs MCP server")
    p.add_argument("--transport", choices=["stdio", "streamable-http", "sse"],
                   default=os.environ.get("MCP_TRANSPORT", "stdio"))
    p.add_argument("--host", default=os.environ.get("MCP_HOST", "0.0.0.0"))
    p.add_argument("--port", type=int, default=int(os.environ.get("MCP_PORT", "8000")))
    args = p.parse_args()

    if args.transport == "stdio":
        mcp.run()
    else:
        mcp.settings.host = args.host
        mcp.settings.port = args.port
        # DNS-rebinding protection defaults to localhost-only — disable for
        # container-network DNS hostnames. See PLAN.md "Hosting" notes.
        if os.environ.get("MCP_DISABLE_DNS_REBINDING_PROTECTION") in {"1", "true", "yes"}:
            mcp.settings.transport_security.enable_dns_rebinding_protection = False
        mcp.run(transport=args.transport)


if __name__ == "__main__":
    main()