data.py

"""Data handling for the HoloViz Documentation MCP server."""

import asyncio
import hashlib
import json
import logging
import os
import re
import shutil
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from pathlib import PurePosixPath
from typing import Any
from typing import Literal
from typing import Optional

import chromadb
import git
from chromadb.api.collection_configuration import CreateCollectionConfiguration
from chromadb.api.shared_system_client import SharedSystemClient
from fastmcp import Context
from nbconvert import MarkdownExporter
from nbformat import read as nbread
from pydantic import HttpUrl

from holoviz_mcp.config.loader import get_config
from holoviz_mcp.config.models import FolderConfig
from holoviz_mcp.config.models import GitRepository
from holoviz_mcp.holoviz_mcp.models import Document

logger = logging.getLogger(__name__)

# Todo: Describe DocumentApp
# Todo: Avoid overflow-x in SearchApp sidebar
# Todo: Add bokeh documentation to README extra config

_CROMA_CONFIGURATION = CreateCollectionConfiguration(
    hnsw={
        "space": "cosine",
        "ef_construction": 200,
        "ef_search": 200,
    }
)


async def log_info(message: str, ctx: Context | None = None):
    """Log an info message to the context or logger."""
    if ctx:
        await ctx.info(message)
    else:
        logger.info(message)


async def log_warning(message: str, ctx: Context | None = None):
    """Log a warning message to the context or logger."""
    if ctx:
        await ctx.warning(message)
    else:
        logger.warning(message)


async def log_exception(message: str, ctx: Context | None = None):
    """Log an error message to the context or logger."""
    if ctx:
        await ctx.error(message)
    else:
        logger.error(message)
        raise Exception(message)


def _normalize_source_path(path: str | Path) -> str:
    """Normalize source paths to POSIX-style separators for cross-platform consistency."""
    return str(path).replace("\\", "/")


def extract_tech_terms(query: str) -> list[str]:
    """Extract technical identifiers from a search query.

    Identifies three categories of terms that benefit from exact substring
    matching rather than pure semantic similarity:

    - **Compound CamelCase** (requires internal case transition):
      ``SelectEditor``, ``ReactiveHTML``, ``TextInput`` — but NOT
      single-word PascalCase like ``Button``, ``Panel``, ``Python``.
    - **snake_case**: ``add_filter``, ``page_size``
    - **Dot-separated qualified names**: ``param.watch``,
      ``pn.widgets.Button`` — excludes common abbreviations like
      ``e.g``, ``i.e`` via a blocklist and minimum length filter.

    Parameters
    ----------
    query : str
        Search query string.

    Returns
    -------
    list[str]
        Deduplicated list of technical terms preserving original case
        and discovery order.  Empty list when no technical terms are found.
    """
    terms: list[str] = []
    seen: set[str] = set()

    # Compound CamelCase: requires an internal lower→upper transition
    # e.g. SelectEditor, ReactiveHTML, TextInput — NOT Button, Panel
    for m in re.finditer(r"\b[A-Z][a-z]+[A-Z][a-zA-Z]*\b", query):
        t = m.group()
        if t not in seen:
            terms.append(t)
            seen.add(t)

    # snake_case identifiers
    for m in re.finditer(r"\b[a-z][a-z0-9]*_[a-z][a-z0-9_]*\b", query):
        t = m.group()
        if t not in seen:
            terms.append(t)
            seen.add(t)

    # Dot-separated qualified names (param.watch, pn.widgets.Button)
    dot_blocklist = {"e.g", "i.e", "vs.", "etc."}
    for m in re.finditer(r"\b[a-z][a-z0-9]*(?:\.[a-zA-Z][a-zA-Z0-9_]*)+\b", query):
        t = m.group()
        if len(t) > 3 and t not in dot_blocklist and t not in seen:
            terms.append(t)
            seen.add(t)

    return terms


_PASCAL_STOPWORDS: set[str] = {
    # Determiners, pronouns, articles
    "The",
    "This",
    "That",
    "These",
    "Those",
    "Each",
    "Every",
    "Some",
    "Any",
    "All",
    "Both",
    "Few",
    "Many",
    "Much",
    "Most",
    "Other",
    "Another",
    "Such",
    "My",
    "Your",
    "His",
    "Her",
    "Its",
    "Our",
    "Their",
    "One",
    "An",
    "No",
    # Short prepositions / conjunctions (sentence-initial)
    "In",
    "At",
    "To",
    "If",
    "Or",
    "So",
    "As",
    "By",
    "Up",
    "On",
    # Interrogatives / relatives
    "Who",
    "What",
    "Which",
    "Where",
    "When",
    "Why",
    "How",
    "Whom",
    "Whose",
    # Personal pronouns
    "He",
    "She",
    "It",
    "We",
    "They",
    # Auxiliary / modal verbs
    "Is",
    "Are",
    "Was",
    "Were",
    "Be",
    "Been",
    "Being",
    "Has",
    "Have",
    "Had",
    "Do",
    "Does",
    "Did",
    "Will",
    "Would",
    "Could",
    "Should",
    "Can",
    "May",
    "Might",
    "Must",
    "Shall",
    # Common verbs (sentence-initial in docs)
    "Get",
    "Set",
    "Let",
    "Make",
    "Take",
    "Give",
    "Put",
    "Run",
    "See",
    "Find",
    "Use",
    "Try",
    "Add",
    "Go",
    "Come",
    "Keep",
    "Show",
    "Tell",
    "Say",
    "Ask",
    "Help",
    "Start",
    "Stop",
    "Open",
    "Close",
    "Read",
    "Write",
    "New",
    "Old",
    "Create",
    "Build",
    "Deploy",
    "Install",
    "Update",
    "Remove",
    "Delete",
    "Enable",
    "Disable",
    "Configure",
    "Define",
    "Return",
    "Check",
    "Pass",
    "Call",
    "Load",
    "Save",
    "Send",
    "Move",
    # Adjectives / adverbs
    "Not",
    "Also",
    "Just",
    "Only",
    "Very",
    "Too",
    "More",
    "Less",
    "First",
    "Last",
    "Next",
    "Best",
    "Good",
    "Bad",
    # Python builtins
    "True",
    "False",
    "None",
    # Conjunctions / prepositions
    "And",
    "But",
    "For",
    "Nor",
    "Yet",
    "With",
    "From",
    "Into",
    "About",
    "After",
    "Before",
    "Between",
    "Through",
    "During",
    "Without",
    "Within",
    "Along",
    "Above",
    "Below",
    # Generic tech words (too broad to be component names)
    "Data",
    "Code",
    "Style",
    "Type",
    "Name",
    "Value",
    "Event",
    "Class",
    "Object",
    "Method",
    "Function",
    "Module",
    "Package",
    "File",
    "Path",
    "String",
    "Number",
    "Integer",
    "Float",
    "List",
    "Dict",
    "Tuple",
    "Array",
    "Index",
    "Key",
    "Error",
    "Warning",
    "Info",
    "Debug",
    "Log",
    # Common sentence starters / discourse markers
    "Here",
    "There",
    "Then",
    "Now",
    "However",
    "Therefore",
    "Furthermore",
    "Moreover",
    "Although",
    "Because",
    "Since",
    "While",
    "Until",
    "Unless",
    "Whether",
    "Though",
    # Documentation filler
    "Note",
    "Tip",
    "Using",
    "Like",
}
# Common English words that can appear PascalCase but are NOT component names.


def extract_pascal_terms(query: str) -> list[str]:
    """Extract single PascalCase words from a query, excluding stopwords.

    Captures words like Scatter, Button, Tabulator that start with an
    uppercase letter followed by at least one lowercase letter.  Compound
    CamelCase words (SelectEditor) are also captured — they overlap with
    :func:`extract_tech_terms` and deduplication happens at the call site.

    Parameters
    ----------
    query : str
        Search query string.

    Returns
    -------
    list[str]
        Deduplicated list of PascalCase terms preserving discovery order.
        Empty list when no terms are found.
    """
    terms: list[str] = []
    seen: set[str] = set()
    for m in re.finditer(r"\b[A-Z][a-z][a-zA-Z]*\b", query):
        t = m.group()
        if t not in seen and t not in _PASCAL_STOPWORDS:
            terms.append(t)
            seen.add(t)
    return terms


def _build_where_document_clause(terms: list[str]) -> dict[str, Any] | None:
    """Build a ChromaDB ``where_document`` clause for keyword pre-filtering.

    Parameters
    ----------
    terms : list[str]
        Technical terms extracted by :func:`extract_tech_terms`.

    Returns
    -------
    dict[str, Any] | None
        ``None`` when *terms* is empty, a single ``{"$contains": term}``
        dict for one term, or ``{"$or": [...]}`` for multiple terms.
    """
    if not terms:
        return None
    if len(terms) == 1:
        return {"$contains": terms[0]}
    return {"$or": [{"$contains": t} for t in terms]}


def _build_stem_boost_clause(pascal_terms: list[str], project: str | None) -> dict[str, Any] | None:
    """Build a ChromaDB ``where`` clause matching ``source_path_stem`` metadata.

    Used to boost results whose filename stem exactly matches a PascalCase
    term from the query (e.g. ``Scatter`` → ``Scatter.ipynb``).

    Parameters
    ----------
    pascal_terms : list[str]
        PascalCase terms extracted by :func:`extract_pascal_terms`.
    project : str | None
        Optional project filter.

    Returns
    -------
    dict[str, Any] | None
        ``None`` when *pascal_terms* is empty, otherwise a ``where``
        clause suitable for ``collection.query(where=...)``.
    """
    if not pascal_terms:
        return None
    filters: list[dict[str, Any]] = []
    if len(pascal_terms) == 1:
        filters.append({"source_path_stem": pascal_terms[0]})
    else:
        filters.append({"$or": [{"source_path_stem": t} for t in pascal_terms]})
    if project:
        filters.append({"project": str(project)})
    return filters[0] if len(filters) == 1 else {"$and": filters}


def extract_keywords(query: str) -> list[str]:
    """Extract meaningful keywords from search query.

    Removes common stopwords and splits into terms.

    Args:
        query: Search query string

    Returns
    -------
        List of meaningful keywords (lowercase)
    """
    stopwords = {
        "the",
        "a",
        "an",
        "and",
        "or",
        "but",
        "in",
        "on",
        "at",
        "to",
        "for",
        "of",
        "with",
        "from",
        "by",
        "about",
        "how",
        "what",
        "where",
        "when",
        "why",
        "which",
        "who",
        "is",
        "are",
        "was",
        "were",
        "be",
        "been",
        "being",
        "have",
        "has",
        "had",
        "do",
        "does",
        "did",
        "will",
        "would",
        "should",
        "could",
        "can",
        "may",
        "might",
        "must",
        "shall",
    }

    # Split and clean
    keywords = query.lower().split()
    # Remove stopwords, keep meaningful terms (> 2 chars)
    keywords = [k for k in keywords if k not in stopwords and len(k) > 2]

    return keywords


def find_keyword_matches(content: str, keywords: list[str]) -> list[tuple[int, int, str]]:
    """Find all positions where keywords appear in content.

    Args:
        content: Document content to search
        keywords: List of keywords to find

    Returns
    -------
        List of (start_pos, end_pos, matched_keyword) tuples, sorted by position
    """
    matches = []
    content_lower = content.lower()

    for keyword in keywords:
        start = 0
        while True:
            pos = content_lower.find(keyword, start)
            if pos == -1:
                break
            matches.append((pos, pos + len(keyword), keyword))
            start = pos + 1

    # Sort by position
    matches.sort(key=lambda x: x[0])
    return matches


def build_excerpts(content: str, matches: list[tuple[int, int, str]], max_chars: int, context_chars: int) -> str:
    """Build excerpt string from matches with context windows.

    Combines nearby matches, adds separators for distant sections.

    Args:
        content: Full document content
        matches: List of (start_pos, end_pos, keyword) tuples
        max_chars: Maximum total characters to return
        context_chars: Characters to include before/after each match

    Returns
    -------
        Excerpt(s) with [...] separators and truncation indicators
    """
    if not matches:
        # Fallback to beginning truncation
        truncated = content[:max_chars]
        last_space = truncated.rfind(" ")
        if last_space > max_chars * 0.8:
            truncated = truncated[:last_space]
        return truncated + "\n\n[... content truncated, use content='full' for complete content ...]"

    # Cluster nearby matches
    clusters = []
    current_cluster = [matches[0]]

    for match in matches[1:]:
        # If within 2x context window, add to current cluster
        if match[0] - current_cluster[-1][1] < 2 * context_chars:
            current_cluster.append(match)
        else:
            clusters.append(current_cluster)
            current_cluster = [match]
    clusters.append(current_cluster)

    # Build excerpts from clusters
    excerpts: list[str] = []
    total_chars = 0

    for cluster in clusters:
        # Get range for this cluster
        start_pos = max(0, cluster[0][0] - context_chars)
        end_pos = min(len(content), cluster[-1][1] + context_chars)

        # Extract excerpt, break at word boundaries
        excerpt = content[start_pos:end_pos]

        # Trim to word boundaries
        if start_pos > 0:
            # Find first space to start at word boundary
            first_space = excerpt.find(" ")
            if first_space != -1 and first_space < context_chars * 0.3:
                excerpt = excerpt[first_space + 1 :]
                start_pos += first_space + 1

        if end_pos < len(content):
            # Find last space to end at word boundary
            last_space = excerpt.rfind(" ")
            if last_space > len(excerpt) * 0.7:
                excerpt = excerpt[:last_space]

        # Check if we have room
        separator = "\n\n[...]\n\n" if excerpts else ""
        if total_chars + len(excerpt) + len(separator) > max_chars:
            # Try to fit partial excerpt
            remaining = max_chars - total_chars - len(separator)
            if remaining > 200:  # Only add if we have reasonable space
                excerpt = excerpt[:remaining]
                last_space = excerpt.rfind(" ")
                if last_space > remaining * 0.7:
                    excerpt = excerpt[:last_space]
                excerpts.append(excerpt)
            break

        excerpts.append(excerpt)
        total_chars += len(excerpt) + len(separator)

    if not excerpts:
        # Fallback if nothing fits
        return build_excerpts(content, [], max_chars, context_chars)

    # Combine excerpts
    result = "\n\n[...]\n\n".join(excerpts)

    # Add indicators at start/end if content was truncated
    first_match_pos = matches[0][0] if matches else 0
    last_match_pos = matches[-1][1] if matches else len(content)

    if first_match_pos > context_chars:
        result = "[...]\n\n" + result
    if last_match_pos < len(content) - context_chars:
        result = result + "\n\n[...]"

    return result


def extract_relevant_excerpt(content: str, query: str, max_chars: int, context_chars: int = 500) -> str:
    """Extract relevant excerpt from content based on query keywords.

    Args:
        content: Full document content
        query: Search query string
        max_chars: Maximum total characters to return
        context_chars: Characters to include before/after each match (default: 500)

    Returns
    -------
        Excerpt(s) centered around query matches, or beginning if no matches
    """
    # Extract keywords from query
    keywords = extract_keywords(query)

    if not keywords:
        # No meaningful keywords, fall back to simple truncation
        return build_excerpts(content, [], max_chars, context_chars)

    # Find keyword matches in content
    matches = find_keyword_matches(content, keywords)

    if not matches:
        # No matches found, fall back to simple truncation
        return build_excerpts(content, [], max_chars, context_chars)

    # Build excerpts from matches
    return build_excerpts(content, matches, max_chars, context_chars)


def truncate_content(content: str | None, max_chars: int | None, query: str | None = None) -> str | None:
    """Truncate content, optionally centering on query matches.

    Args:
        content: The content to truncate
        max_chars: Maximum characters allowed. If None, no truncation is performed.
        query: Optional search query for context-aware truncation

    Returns
    -------
        The original content if under limit, truncated content with ellipsis if over limit,
        or None if content is None.
    """
    if content is None or max_chars is None:
        return content

    if len(content) <= max_chars:
        return content

    # If query provided, use smart excerpt extraction
    if query:
        # Adjust context_chars based on max_chars to ensure keywords fit
        # Use at most 40% of max_chars for context on each side
        context_chars = min(500, int(max_chars * 0.4))
        return extract_relevant_excerpt(content, query, max_chars, context_chars)

    # Otherwise, use simple truncation (existing logic)
    truncated = content[:max_chars]
    last_space = truncated.rfind(" ")

    # Don't cut off too much - if last space is in the last 20% of max_chars, use it
    if last_space > max_chars * 0.8:
        truncated = truncated[:last_space]

    # Add indicator
    return truncated + "\n\n[... content truncated, use content='full' for complete content ...]"


def _find_markdown_header_lines(content: str) -> list[int]:
    """Find line indices of H1/H2 markdown headers that are outside code fences.

    Skips headers inside fenced code blocks (``` ... ```) so that Python
    comments like ``# setup code`` or decorative lines like
    ``# ========`` are never treated as split points.

    Parameters
    ----------
    content : str
        Full markdown content.

    Returns
    -------
    list[int]
        Sorted line indices (0-based) where H1/H2 headers appear.
    """
    header_lines: list[int] = []
    in_code_block = False

    for i, line in enumerate(content.split("\n")):
        if line.strip().startswith("```"):
            in_code_block = not in_code_block
            continue
        if not in_code_block and re.match(r"^#{1,2} ", line):
            header_lines.append(i)

    return header_lines


def chunk_document(doc: dict[str, Any], min_chunk_chars: int = 100) -> list[dict[str, Any]]:
    r"""Split a document into chunks at H1/H2 markdown headers.

    Only headers **outside** fenced code blocks are used as split points,
    so Python comments (``# ...``) and decorative dividers inside code
    blocks are left intact.

    Each chunk stores two content fields:

    - ``content``: the document title prepended to the raw section text
      (``"Title\\n\\n## Section ..."``) so that ChromaDB's embedding model
      associates every chunk with its parent document context.
    - ``raw_content``: the original section text without the title prefix,
      used by ``get_document()`` and ``search_get_reference_guide()`` to
      reconstruct the full document without duplicating the title.

    Parameters
    ----------
    doc : dict[str, Any]
        Document dict with at least 'id', 'title', and 'content' keys,
        plus other metadata fields.
    min_chunk_chars : int
        Minimum character count for a chunk to be kept. Chunks below this
        threshold are discarded (e.g. empty sections). Default: 100.

    Returns
    -------
    list[dict[str, Any]]
        List of chunk dicts. If no H1/H2 headers are found, returns a single
        chunk with chunk_index=0.
    """
    content = doc.get("content", "") or ""
    parent_id = doc["id"]
    title = doc.get("title", "")

    # Find H1/H2 header lines that are outside code fences
    lines = content.split("\n")
    header_indices = _find_markdown_header_lines(content)

    # Build text parts by splitting at header boundaries
    if header_indices:
        parts: list[str] = []
        prev = 0
        for idx in header_indices:
            if idx > prev:
                parts.append("\n".join(lines[prev:idx]))
            prev = idx
        # Last chunk: from last header to end
        parts.append("\n".join(lines[prev:]))
    else:
        parts = [content]

    # Filter out tiny/empty chunks
    parts = [p for p in parts if len(p.strip()) >= min_chunk_chars]

    # If nothing survived filtering, keep the whole content as one chunk
    if not parts:
        parts = [content]

    # Metadata keys to copy from the parent document to each chunk
    metadata_keys = ("title", "url", "project", "source_path", "source_path_stem", "source_url", "description", "is_reference")

    chunks: list[dict[str, Any]] = []
    for idx, part in enumerate(parts):
        chunk: dict[str, Any] = {}
        for key in metadata_keys:
            if key in doc:
                chunk[key] = doc[key]
        chunk["id"] = f"{parent_id}___chunk_{idx}"
        chunk["chunk_index"] = idx
        chunk["parent_id"] = parent_id
        # raw_content: original section text for faithful document reconstruction
        chunk["raw_content"] = part
        # content: context-prefixed text stored in ChromaDB for better embeddings
        context_prefix = _build_context_prefix(doc.get("project", ""), doc.get("source_path", ""), doc.get("is_reference", False))
        if title:
            chunk["content"] = f"{context_prefix}{title}\n\n{part}"
        elif context_prefix:
            chunk["content"] = f"{context_prefix}{part}"
        else:
            chunk["content"] = part
        chunks.append(chunk)

    return chunks


def _strip_title_prefix(
    content: str,
    title: str,
    project: str = "",
    source_path: str = "",
    is_reference: bool = False,
) -> str:
    r"""Remove the title prefix that chunk_document() prepends for embedding.

    Handles both the new format (with context prefix) and the old format
    (title only) for backward compatibility with existing indexes.

    Parameters
    ----------
    content : str
        Chunk content, possibly prefixed with context + title.
    title : str
        The document title to strip.
    project : str
        Project name (used to compute context prefix).
    source_path : str
        Relative source path (used to compute context prefix).
    is_reference : bool
        Whether the document is a reference guide.

    Returns
    -------
    str
        Content with the title prefix removed if present, otherwise unchanged.
    """
    # Try new format: context_prefix + title
    context_prefix = _build_context_prefix(project, source_path, is_reference)
    new_prefix = f"{context_prefix}{title}\n\n"
    if title and content.startswith(new_prefix):
        return content[len(new_prefix) :]
    # Backward compat: old format (title only, no context prefix)
    old_prefix = f"{title}\n\n"
    if title and content.startswith(old_prefix):
        return content[len(old_prefix) :]
    return content


def _extract_reference_category(source_path: str, is_reference: bool) -> str | None:
    """Extract the component category from a reference guide's source path.

    Finds ``"reference"`` in the path parts and returns the next non-filename
    part (i.e. the directory immediately below ``reference/``).

    Parameters
    ----------
    source_path : str
        Relative source path of the document.
    is_reference : bool
        Whether this document is a reference guide.

    Returns
    -------
    str | None
        Category name (e.g. ``"widgets"``, ``"elements"``, ``"panes"``),
        or ``None`` if not a reference doc or no category directory exists.
    """
    if not is_reference:
        return None
    parts = source_path.split("/")
    try:
        ref_idx = parts.index("reference")
    except ValueError:
        return None
    # Category is next part after "reference", if it's not a filename
    if ref_idx + 1 < len(parts):
        candidate = parts[ref_idx + 1]
        if "." not in candidate:  # skip filenames like "guide.md"
            return candidate
    return None


def _build_context_prefix(project: str, source_path: str, is_reference: bool) -> str:
    r"""Build a context line prepended before the title in chunk content.

    The prefix enriches the embedding with project and reference-category
    context so that queries like "HoloViews Scatter" are closer in vector
    space to the Scatter reference guide chunk.

    Parameters
    ----------
    project : str
        Project name (e.g. ``"panel"``, ``"holoviews"``).
    source_path : str
        Relative source path of the document.
    is_reference : bool
        Whether this document is a reference guide.

    Returns
    -------
    str
        A context line ending with ``"\n"`` (e.g. ``"panel widgets\n"``),
        or ``""`` when no context is available.
    """
    parts: list[str] = []
    if project:
        parts.append(project)
    category = _extract_reference_category(source_path, is_reference)
    if category:
        parts.append(category)
    if parts:
        return " ".join(parts) + "\n"
    return ""


def get_skill(name: str) -> str:
    """Get skill for using a project with LLMs.

    This function searches for skill resources in user and default directories,
    with user resources taking precedence over default ones.

    Args:
        name (str): The name of the skill to get.

    Returns
    -------
        str: A string containing the skill in Markdown format.

    Raises
    ------
        FileNotFoundError: If the specified skill is not found in either directory.
    """
    config = get_config()

    # Convert underscored names to hyphenated for file lookup
    skill_filename = name.replace("_", "-") + ".md"

    # Search in user directory first, then default directory
    search_paths = [
        config.skills_dir("user"),
        config.skills_dir("default"),
    ]

    for search_dir in search_paths:
        skills_file = search_dir / skill_filename
        if skills_file.exists():
            return skills_file.read_text(encoding="utf-8")

    # If not found, raise error with helpful message
    available_files = []
    for search_dir in search_paths:
        if search_dir.exists():
            available_files.extend([f.name for f in search_dir.glob("*.md")])

    available_str = ", ".join(set(available_files)) if available_files else "None"
    raise FileNotFoundError(f"Skill file {name} not found. Available skills: {available_str}. Searched in: {[str(p) for p in search_paths]}")


def list_skills() -> list[str]:
    """List all available skills.

    This function discovers available skills from both user and default directories,
    with user resources taking precedence over default ones.

    Returns
    -------
        list[str]: A list of the skills available.
            Names are returned in hyphenated format (e.g., "panel-material-ui").
    """
    config = get_config()

    # Collect available projects from both directories
    available_projects = set()

    search_paths = [
        config.skills_dir("user"),
        config.skills_dir("default"),
    ]

    for search_dir in search_paths:
        if search_dir.exists():
            for md_file in search_dir.glob("*.md"):
                available_projects.add(md_file.stem)

    return sorted(list(available_projects))


def remove_leading_number_sep_from_path(p: Path) -> Path:
    """Remove a leading number + underscore or hyphen from the last path component."""
    new_name = re.sub(r"^\d+[_-]", "", p.name)
    return p.with_name(new_name)


def convert_path_to_url(path: Path, remove_first_part: bool = True, url_transform: Literal["holoviz", "plotly", "datashader"] = "holoviz") -> str:
    """Convert a relative file path to a URL path.

    Converts file paths to web URLs by replacing file extensions with .html
    and optionally removing the first path component for legacy compatibility.

    Args:
        path: The file path to convert
        remove_first_part: Whether to remove the first path component (legacy compatibility)
        url_transform: How to transform the file path into a URL:

            - "holoviz": Replace file extension with .html (default)
            - "plotly": Replace file extension with / (e.g., filename.md -> filename/)
            - "datashader": Remove leading index and replace file extension with .html (e.g., 01_filename.md -> filename.html)

    Returns
    -------
        URL path with .html extension

    Examples
    --------
        >>> convert_path_to_url(Path("doc/getting_started.md"))
        "getting_started.html"
        >>> convert_path_to_url(Path("examples/reference/Button.ipynb"), False)
        "examples/reference/Button.html"
        >>> convert_path_to_url(Path("/doc/python/3d-axes.md"), False, "plotly")
        "/doc/python/3d-axes/"
        >>> convert_path_to_url(Path("/examples/user_guide/10_Performance.ipynb"), False, "datashader")
        "/examples/user_guide/Performance.html"
    """
    if url_transform in ["holoviz", "datashader"]:
        path = remove_leading_number_sep_from_path(path)

    # Convert path to URL format
    parts = list(path.parts)

    # Only remove first part if requested (for legacy compatibility)
    if remove_first_part and parts:
        parts.pop(0)

    # Reconstruct path and convert to string (use PurePosixPath for forward slashes)
    if parts:
        url_path = "/".join(parts)
    else:
        url_path = ""

    # Replace file extensions with suffix
    if url_path:
        path_obj = PurePosixPath(url_path)
        if url_transform == "plotly":
            url_path = str(path_obj.with_suffix(suffix="")) + "/"
            if url_path.endswith("index/"):
                url_path = url_path[: -len("index/")] + "/"
        else:
            url_path = str(path_obj.with_suffix(suffix=".html"))

    return url_path


class DocumentationIndexer:
    """Handles cloning, processing, and indexing of documentation."""

    def __init__(self, *, data_dir: Optional[Path] = None, repos_dir: Optional[Path] = None, vector_dir: Optional[Path] = None):
        """Initialize the DocumentationIndexer.

        Args:
            data_dir: Directory to store index data. Defaults to user config directory.
            repos_dir: Directory to store cloned repositories. Defaults to HOLOVIZ_MCP_REPOS_DIR.
            vector_dir: Directory to store vector database. Defaults to config.vector_dir
        """
        # Use unified config for default paths
        config = self._holoviz_mcp_config = get_config()

        self.data_dir = data_dir or config.user_dir
        self.data_dir.mkdir(parents=True, exist_ok=True)

        # Use configurable repos directory for repository downloads
        self.repos_dir = repos_dir or config.repos_dir
        self.repos_dir.mkdir(parents=True, exist_ok=True)

        # Use configurable directory for vector database path
        self._vector_db_path = vector_dir or config.server.vector_db_path
        self._vector_db_path.parent.mkdir(parents=True, exist_ok=True)

        # Disable ChromaDB telemetry based on config
        if not config.server.anonymized_telemetry:
            os.environ["ANONYMIZED_TELEMETRY"] = "False"

        # Initialize ChromaDB with health check
        try:
            self.chroma_client = chromadb.PersistentClient(path=str(self._vector_db_path))
            self.collection = self.chroma_client.get_or_create_collection("holoviz_docs", configuration=_CROMA_CONFIGURATION)
            self.collection.count()  # Probe read to verify health
        except (KeyboardInterrupt, SystemExit):
            raise
        except BaseException as e:
            logger.error("ChromaDB initialization failed (%s: %s). Wiping and reinitializing.", type(e).__name__, e)
            shutil.rmtree(self._vector_db_path, ignore_errors=True)
            self._vector_db_path.mkdir(parents=True, exist_ok=True)
            SharedSystemClient.clear_system_cache()
            self.chroma_client = chromadb.PersistentClient(path=str(self._vector_db_path))
            self.collection = self.chroma_client.get_or_create_collection("holoviz_docs", configuration=_CROMA_CONFIGURATION)

        # Lazy-initialized async lock for database operations to prevent corruption from concurrent access
        self._db_lock: Optional[asyncio.Lock] = None

        # Thread-local storage for MarkdownExporter (not thread-safe due to Jinja2)
        self._thread_local = threading.local()

        # Load documentation config from the centralized config system
        self.config = get_config().docs

        # Wrap index_documentation to ensure all calls use the async DB lock
        # This prevents race conditions when indexing is called directly (e.g., from
        # update_index tool or run method) and concurrently with search operations
        if not hasattr(self, "_index_documentation_wrapped"):
            original_index_documentation = self.index_documentation
            # Keep a reference to the unwrapped version for use inside ensure_indexed
            # (which already holds db_lock, so calling the wrapped version would deadlock)
            self._index_documentation_unlocked = original_index_documentation

            async def _locked_index_documentation(*args: Any, **kwargs: Any) -> Any:
                async with self.db_lock:
                    return await original_index_documentation(*args, **kwargs)

            self.index_documentation = _locked_index_documentation  # type: ignore[method-assign]
            self._index_documentation_wrapped = True

    def _get_nb_exporter(self) -> MarkdownExporter:
        """Get or create a thread-local MarkdownExporter instance.

        MarkdownExporter uses Jinja2 templates internally which are not
        thread-safe, so each thread gets its own instance.
        """
        if not hasattr(self._thread_local, "nb_exporter"):
            self._thread_local.nb_exporter = MarkdownExporter()
        return self._thread_local.nb_exporter

    @property
    def db_lock(self) -> asyncio.Lock:
        """Lazy-initialize and return the database lock.

        This ensures the lock is created in the correct event loop context.
        """
        if self._db_lock is None:
            self._db_lock = asyncio.Lock()
        return self._db_lock

    @property
    def _backup_path(self) -> Path:
        """Path to the backup copy of the vector database directory."""
        return Path(str(self._vector_db_path) + ".bak")

    @property
    def _hash_file_path(self) -> Path:
        """Path to the hash sidecar file for incremental indexing."""
        return self._vector_db_path.parent / "index_hashes.json"

    async def _restore_from_backup(self, ctx: Context | None = None) -> None:
        """Restore vector database from backup after a write failure."""
        backup_path = self._backup_path
        if not backup_path.exists():
            await log_warning("No backup found to restore from.", ctx)
            return
        try:
            shutil.rmtree(self._vector_db_path, ignore_errors=True)
            shutil.copytree(backup_path, self._vector_db_path, dirs_exist_ok=True)
            SharedSystemClient.clear_system_cache()
            self.chroma_client = chromadb.PersistentClient(path=str(self._vector_db_path))
            self.collection = self.chroma_client.get_or_create_collection("holoviz_docs", configuration=_CROMA_CONFIGURATION)
            # Restore hash file if backup exists
            hash_bak = backup_path.parent / "index_hashes.json.bak"
            if hash_bak.exists():
                shutil.copy2(hash_bak, self._hash_file_path)
            await log_info("Restored vector database from backup.", ctx)
        except (KeyboardInterrupt, SystemExit):
            raise
        except BaseException as e:
            logger.error("Failed to restore from backup (%s: %s). Database may be degraded.", type(e).__name__, e)

    def _load_hashes(self) -> dict[str, str]:
        """Load document hashes from the sidecar file.

        Returns
        -------
        dict[str, str]
            Mapping of doc_id -> SHA-256 hex digest. Empty dict if file
            doesn't exist or is corrupt.
        """
        path = self._hash_file_path
        if not path.exists():
            return {}
        try:
            with open(path, encoding="utf-8") as f:
                data = json.load(f)
            if isinstance(data, dict):
                return data
            return {}
        except (json.JSONDecodeError, OSError):
            logger.warning("Failed to load hash file %s, starting fresh", path)
            return {}

    def _save_hashes(self, hashes: dict[str, str]) -> None:
        """Save document hashes to the sidecar file."""
        path = self._hash_file_path
        with open(path, "w", encoding="utf-8") as f:
            json.dump(hashes, f, indent=2)

    @staticmethod
    def _compute_file_hash(file_path: Path) -> str:
        """Compute SHA-256 hash of a file's contents."""
        return hashlib.sha256(file_path.read_bytes()).hexdigest()

    def _delete_doc_chunks(self, doc_id: str) -> int:
        """Delete all chunks for a document from ChromaDB.

        Parameters
        ----------
        doc_id : str
            The parent document ID (chunks have parent_id metadata matching this).

        Returns
        -------
        int
            Number of chunks deleted.
        """
        results = self.collection.get(
            where={"parent_id": doc_id},
            include=[],  # only need IDs
        )
        if results["ids"]:
            self.collection.delete(ids=results["ids"])
        return len(results["ids"]) if results["ids"] else 0

    def is_indexed(self) -> bool:
        """Check if documentation index exists and is valid."""
        try:
            count = self.collection.count()
            return count > 0
        except (KeyboardInterrupt, SystemExit):
            raise
        except BaseException:
            return False

    async def ensure_indexed(self, ctx: Context | None = None):
        """Ensure documentation is indexed, creating if necessary.

        If the index exists but is missing configured projects (e.g. because the
        default config was merged with a user config after the initial index was
        built, or because ChromaDB lost data), the missing projects are indexed
        incrementally.

        NOTE: This method is called inside db_lock, so it must use
        ``_index_documentation_unlocked`` to avoid deadlock.
        """
        if not self.is_indexed():
            await log_info("Documentation index not found. Creating initial index...", ctx)
            await self._index_documentation_unlocked()
            return

        # Check for configured projects that are not yet in ChromaDB.
        # We check ChromaDB (not just the hash file) because the hash file
        # can be out of sync if ChromaDB lost data after a failed write.
        configured_projects = set(self.config.repositories.keys())
        indexed_projects = self._get_indexed_project_names_from_db()
        missing = sorted(configured_projects - indexed_projects)
        if missing:
            # Remove stale hashes for missing projects so index_documentation
            # re-processes them instead of skipping unchanged files.
            self._remove_hashes_for_projects(missing)
            await log_info(f"Indexing {len(missing)} new project(s): {', '.join(missing)}", ctx)
            await self._index_documentation_unlocked(projects=missing)

    def _get_indexed_project_names_from_db(self) -> set[str]:
        """Return project names that have documents in ChromaDB.

        Queries the vector database for unique project metadata values.
        Used by ``ensure_indexed`` to detect projects that need indexing.
        """
        try:
            results = self.collection.get(include=["metadatas"])
            if not results["metadatas"]:
                return set()
            return {str(m.get("project", "")) for m in results["metadatas"] if m.get("project")}
        except (KeyboardInterrupt, SystemExit):
            raise
        except BaseException:
            return set()

    def _remove_hashes_for_projects(self, projects: list[str]) -> None:
        """Remove hash entries for the given projects so they get re-indexed.

        When ChromaDB is missing projects that the hash file still tracks,
        the incremental indexer would skip them (hash unchanged). Removing
        the stale hashes forces re-extraction and re-embedding.
        """
        hashes = self._load_hashes()
        prefixes = tuple(f"{p}___" for p in projects)
        cleaned = {k: v for k, v in hashes.items() if not k.startswith(prefixes)}
        removed = len(hashes) - len(cleaned)
        if removed > 0:
            logger.info("Removed %d stale hash entries for projects: %s", removed, ", ".join(projects))
            self._save_hashes(cleaned)

    async def clone_or_update_repo(self, repo_name: str, repo_config: "GitRepository", ctx: Context | None = None) -> Optional[Path]:
        """Clone or update a single repository.

        Delegates to the synchronous implementation via run_in_executor.
        """
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(None, self._clone_or_update_repo_sync, repo_name, repo_config)

    def _clone_or_update_repo_sync(self, repo_name: str, repo_config: "GitRepository") -> Optional[Path]:
        """Clone or update a single repository (synchronous, for thread pool).

        Same logic as clone_or_update_repo but uses logger instead of async ctx.
        """
        repo_path = self.repos_dir / repo_name

        try:
            if repo_path.exists():
                repo = git.Repo(repo_path)

                if repo_config.tag:
                    # Tag checkouts leave the repo in detached HEAD; verify via tag→commit mapping.
                    matching_tag = next((t for t in repo.tags if t.name == repo_config.tag), None)
                    if matching_tag and matching_tag.commit == repo.head.commit:
                        # Already on the correct tag — tags are immutable, nothing to pull.
                        return repo_path
                    logger.info(f"Stale checkout for {repo_name}: expected tag '{repo_config.tag}'. Deleting and re-cloning...")
                    shutil.rmtree(repo_path)
                    # Fall through to clone below
                elif repo_config.branch:
                    try:
                        current_branch = repo.active_branch.name
                    except TypeError:
                        current_branch = None  # Detached HEAD

                    if current_branch != repo_config.branch:
                        logger.info(f"Stale checkout for {repo_name}: on '{current_branch}', expected '{repo_config.branch}'. Deleting and re-cloning...")
                        shutil.rmtree(repo_path)
                        # Fall through to clone below
                    else:
                        logger.info(f"Updating {repo_name} repository at {repo_path}...")
                        try:
                            repo.remotes.origin.pull()
                            return repo_path
                        except Exception as pull_err:
                            logger.info(f"Pull failed for {repo_name} ({pull_err}). Deleting and re-cloning...")
                            shutil.rmtree(repo_path)
                            # Fall through to clone below
                else:
                    logger.info(f"Updating {repo_name} repository at {repo_path}...")
                    try:
                        repo.remotes.origin.pull()
                        return repo_path
                    except Exception as pull_err:
                        logger.info(f"Pull failed for {repo_name} ({pull_err}). Deleting and re-cloning...")
                        shutil.rmtree(repo_path)
                        # Fall through to clone below

            # Repo does not exist (fresh or was just deleted above)
            logger.info(f"Cloning {repo_name} repository to {repo_path}...")
            clone_kwargs: dict[str, Any] = {"depth": 1}

            if repo_config.branch:
                clone_kwargs["branch"] = repo_config.branch
            elif repo_config.tag:
                clone_kwargs["branch"] = repo_config.tag
            elif repo_config.commit:
                git.Repo.clone_from(str(repo_config.url), repo_path, **clone_kwargs)
                repo = git.Repo(repo_path)
                repo.git.checkout(repo_config.commit)
                return repo_path

            git.Repo.clone_from(str(repo_config.url), repo_path, **clone_kwargs)
            return repo_path
        except Exception as e:
            logger.warning(f"Failed to clone/update {repo_name}: {e}")
            return None

    def _extract_docs_from_repo_sync(self, repo_path: Path, project: str, old_hashes: dict[str, str] | None = None) -> dict[str, Any]:
        """Extract documentation files from a repository (synchronous, for thread pool).

        When old_hashes is provided, files whose hash matches are skipped (not
        read or converted), making incremental runs much faster.

        Returns
        -------
        dict
            ``{"docs": [...], "skipped_hashes": {doc_id: hash, ...}}``
            where ``docs`` contains only new/changed documents and
            ``skipped_hashes`` maps unchanged doc_ids to their hashes.
        """
        docs = []
        skipped_hashes: dict[str, str] = {}
        repo_config = self.config.repositories[project]

        if isinstance(repo_config.folders, dict):
            folders = repo_config.folders
        else:
            folders = {name: FolderConfig() for name in repo_config.folders}

        files: set = set()
        logger.info(f"Processing {project} documentation files in {','.join(folders.keys())}")

        for folder_name in folders.keys():
            docs_folder: Path = repo_path / folder_name
            if docs_folder.exists():
                for pattern in self.config.index_patterns:
                    files.update(docs_folder.glob(pattern))

        skipped_count = 0
        for file in files:
            if file.exists() and not file.is_dir():
                # Early hash check: skip unchanged files before expensive content extraction
                if old_hashes:
                    relative_path = file.relative_to(repo_path)
                    doc_id = self._generate_doc_id(project, relative_path)
                    file_hash = self._compute_file_hash(file)
                    if old_hashes.get(doc_id) == file_hash:
                        skipped_hashes[doc_id] = file_hash
                        skipped_count += 1
                        continue

                folder_name = ""
                for fname in folders.keys():
                    folder_path = repo_path / fname
                    try:
                        file.relative_to(folder_path)
                        folder_name = fname
                        break
                    except ValueError:
                        continue

                doc_data = self.process_file(file, project, repo_config, folder_name)
                if doc_data:
                    docs.append(doc_data)

        reference_count = sum(1 for doc in docs if doc["is_reference"])
        regular_count = len(docs) - reference_count
        if skipped_count:
            logger.info(f"  {project}: {len(docs)} changed + {skipped_count} unchanged ({regular_count} regular, {reference_count} reference guides)")
        else:
            logger.info(f"  {project}: {len(docs)} total documents ({regular_count} regular, {reference_count} reference guides)")
        return {"docs": docs, "skipped_hashes": skipped_hashes}

    def _clone_and_extract_repo_sync(self, repo_name: str, repo_config: "GitRepository", old_hashes: dict[str, str] | None = None) -> dict[str, Any]:
        """Clone/update a repo and extract its documents (single unit of work for thread pool)."""
        t0 = time.monotonic()
        repo_path = self._clone_or_update_repo_sync(repo_name, repo_config)
        if not repo_path:
            return {"docs": [], "skipped_hashes": {}}
        result = self._extract_docs_from_repo_sync(repo_path, repo_name, old_hashes)
        logger.info(f"Completed {repo_name}: {len(result['docs'])} docs in {time.monotonic() - t0:.1f}s")
        return result

    def _is_reference_document(self, file_path: Path, project: str, folder_name: str = "") -> bool:
        """Check if the document is a reference document using configurable patterns.

        Args:
            file_path: Full path to the file
            project: Project name
            folder_name: Name of the folder this file belongs to

        Returns
        -------
            bool: True if this is a reference document
        """
        repo_config = self.config.repositories[project]
        repo_path = self.repos_dir / project

        try:
            relative_path = file_path.relative_to(repo_path)

            # Check against configured reference patterns
            for pattern in repo_config.reference_patterns:
                if relative_path.match(pattern):
                    return True

            # Fallback to simple "reference" in path check
            return "reference" in relative_path.parts
        except (ValueError, KeyError):
            # If we can't determine relative path or no patterns configured, use simple fallback
            return "reference" in file_path.parts

    def _generate_doc_id(self, project: str, path: Path) -> str:
        """Generate a unique document ID from project and path."""
        readable_path = _normalize_source_path(path).replace("/", "___").replace(".", "_")
        readable_id = f"{project}___{readable_path}"

        return readable_id

    def _generate_doc_url(self, project: str, path: Path, folder_name: str = "") -> str:
        """Generate documentation URL for a file.

        This method creates the final URL where the documentation can be accessed online.
        It handles folder URL mapping to ensure proper URL structure for different documentation layouts.

        Args:
            project: Name of the project/repository (e.g., "panel", "hvplot")
            path: Relative path to the file within the repository
            folder_name: Name of the folder containing the file (e.g., "examples/reference", "doc")
                       Used for URL path mapping when folders have custom URL structures

        Returns
        -------
            Complete URL to the documentation file

        Examples
        --------
            For Panel reference guides:
            - Input: project="panel", path="examples/reference/widgets/Button.ipynb", folder_name="examples/reference"
            - Output: "https://panel.holoviz.org/reference/widgets/Button.html"

            For regular documentation:
            - Input: project="panel", path="doc/getting_started.md", folder_name="doc"
            - Output: "https://panel.holoviz.org/getting_started.html"
        """
        repo_config = self.config.repositories[project]
        base_url = str(repo_config.base_url).rstrip("/")

        # Get the URL path mapping for this folder
        folder_url_path = repo_config.get_folder_url_path(folder_name)

        # If there's a folder URL mapping, we need to adjust the path
        if folder_url_path and folder_name:
            # Remove the folder name from the beginning of the path
            path_str = _normalize_source_path(path)

            # Check if path starts with the folder name
            if path_str.startswith(folder_name + "/"):
                # Remove the folder prefix and leading slash
                remaining_path = path_str[len(folder_name) + 1 :]
                adjusted_path = Path(remaining_path) if remaining_path else Path(".")
            elif path_str == folder_name:
                # The path is exactly the folder name
                adjusted_path = Path(".")
            else:
                # Fallback: try to remove folder parts from the beginning
                path_parts = list(path.parts)
                folder_parts = folder_name.split("/")
                for folder_part in folder_parts:
                    if path_parts and path_parts[0] == folder_part:
                        path_parts = path_parts[1:]
                adjusted_path = Path(*path_parts) if path_parts else Path(".")

            # Don't remove first part since we already adjusted the path
            doc_path = convert_path_to_url(adjusted_path, remove_first_part=False, url_transform=repo_config.url_transform)
        else:
            # Convert file path to URL format normally (remove first part for legacy compatibility)
            doc_path = convert_path_to_url(path, remove_first_part=True, url_transform=repo_config.url_transform)

        # Combine base URL, folder URL path, and document path
        if folder_url_path:
            full_url = f"{base_url}{folder_url_path}/{doc_path}"
        else:
            full_url = f"{base_url}/{doc_path}"

        return full_url.replace("//", "/").replace(":/", "://")  # Fix double slashes

    @staticmethod
    def _to_title(fallback_filename: str = "") -> str:
        """Extract title from a filename or return a default title."""
        title = Path(fallback_filename).stem
        if "_" in title and title.split("_")[0].isdigit():
            title = title.split("_", 1)[-1]
        title = title.replace("_", " ").replace("-", " ").title()
        return title

    @classmethod
    def _extract_title_from_markdown(cls, content: str, fallback_filename: str = "") -> str:
        """Extract title from markdown content, with filename fallback."""
        lines = content.split("\n")
        for line in lines:
            line = line.strip()
            if line.startswith("# "):
                # Return just the title text without the "# " prefix
                return line[2:].strip()
            if line.startswith("##"):
                break

        if fallback_filename:
            return cls._to_title(fallback_filename)

        return "No Title"

    @staticmethod
    def _extract_description_from_markdown(content: str, max_length=200) -> str:
        """Extract description from markdown content."""
        content = content.strip()

        # Plotly documents start with --- ... --- section. Skip the section
        if content.startswith("---"):
            content = content.split("---", 2)[-1].strip()

        lines = content.split("\n")
        clean_lines = []
        in_code_block = False

        for line in lines:
            if line.strip().startswith("```"):
                in_code_block = not in_code_block
                continue

            if in_code_block or line.startswith(("#", "    ", "\t", "---", "___")):
                continue

            clean_lines.append(line)

        # Join lines and clean up
        clean_content = "\n".join(clean_lines).strip()

        # Remove extra whitespace and limit length
        clean_content = " ".join(clean_content.split())

        if len(clean_content) > max_length:
            clean_content = clean_content[:max_length].rsplit(" ", 1)[0]
        if not clean_content.endswith("."):
            clean_content += " ..."

        return clean_content

    def convert_notebook_to_markdown(self, notebook_path: Path) -> str:
        """Convert a Jupyter notebook to markdown."""
        try:
            with open(notebook_path, "r", encoding="utf-8") as f:
                notebook = nbread(f, as_version=4)

            (body, resources) = self._get_nb_exporter().from_notebook_node(notebook)
            return body
        except Exception as e:
            logger.error(f"Failed to convert notebook {notebook_path}: {e}")
            return str(e)

    @staticmethod
    def _to_source_url(file_path: Path, repo_config: GitRepository, raw: bool = False) -> str:
        """Generate source URL for a file based on repository configuration."""
        url = str(repo_config.url)
        branch = repo_config.branch or "main"
        posix_path = _normalize_source_path(file_path)
        if url.startswith("https://github.com") and url.endswith(".git"):
            url = url.replace("https://github.com/", "").replace(".git", "")
            project, repository = url.split("/")
            if raw:
                return f"https://raw.githubusercontent.com/{project}/{repository}/refs/heads/{branch}/{posix_path}"

            return f"https://github.com/{project}/{repository}/blob/{branch}/{posix_path}"
        if "dev.azure.com" in url:
            organisation = url.split("/")[3].split("@")[0]
            project = url.split("/")[-3]
            repo_name = url.split("/")[-1]
            if raw:
                return f"https://dev.azure.com/{organisation}/{project}/_apis/sourceProviders/TfsGit/filecontents?repository={repo_name}&path=/{posix_path}&commitOrBranch={branch}&api-version=7.0"

            return f"https://dev.azure.com/{organisation}/{project}/_git/{repo_name}?path=/{posix_path}&version=GB{branch}"

        raise ValueError(f"Unsupported repository URL format: {url}. Please provide a valid GitHub or Azure DevOps URL.")

    def process_file(self, file_path: Path, project: str, repo_config: GitRepository, folder_name: str = "") -> Optional[dict[str, Any]]:
        """Process a file and extract metadata."""
        try:
            if file_path.suffix == ".ipynb":
                content = self.convert_notebook_to_markdown(file_path)
            elif file_path.suffix in [".md", ".rst", ".txt"]:
                with open(file_path, "r", encoding="utf-8") as f:
                    content = f.read()
            else:
                logger.debug(f"Skipping unsupported file type: {file_path}")
                return None

            title = self._extract_title_from_markdown(content, file_path.name)
            if not title:
                title = file_path.stem.replace("_", " ").title()

            description = self._extract_description_from_markdown(content)

            repo_path = self.repos_dir / project
            relative_path = file_path.relative_to(repo_path)

            doc_id = self._generate_doc_id(project, relative_path)

            is_reference = self._is_reference_document(file_path, project, folder_name)

            source_url = self._to_source_url(relative_path, repo_config)

            return {
                "id": doc_id,
                "title": title,
                "url": self._generate_doc_url(project, relative_path, folder_name),
                "project": project,
                "source_path": _normalize_source_path(relative_path),
                "source_path_stem": file_path.stem,
                "source_url": source_url,
                "description": description,
                "content": content,
                "is_reference": is_reference,
                "file_hash": self._compute_file_hash(file_path),
            }
        except Exception as e:
            logger.error(f"Failed to process file {file_path}: {e}")
            return None

    async def extract_docs_from_repo(self, repo_path: Path, project: str, ctx: Context | None = None) -> dict[str, Any]:
        """Extract documentation files from a repository.

        Delegates to the synchronous implementation via run_in_executor.
        """
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(None, self._extract_docs_from_repo_sync, repo_path, project)

    async def index_documentation(
        self,
        ctx: Context | None = None,
        projects: list[str] | None = None,
        full_rebuild: bool = False,
    ) -> None:
        """Index documentation, incrementally when possible.

        Parameters
        ----------
        ctx : Context | None
            FastMCP context for logging.
        projects : list[str] | None
            Only process these projects. None means all.
        full_rebuild : bool
            Force full rebuild, ignoring cached hashes.
        """
        overall_t0 = time.monotonic()
        await log_info("Starting documentation indexing...", ctx)

        # Validate project names
        if projects:
            available = set(self.config.repositories.keys())
            invalid = [p for p in projects if p not in available]
            if invalid:
                raise ValueError(f"Unknown project(s): {', '.join(invalid)}. Available: {', '.join(sorted(available))}")

        # Determine target repos
        if projects:
            target_repos = {name: cfg for name, cfg in self.config.repositories.items() if name in projects}
        else:
            target_repos = dict(self.config.repositories.items())

        # Load existing hashes BEFORE extraction so workers can skip unchanged files
        old_hashes = self._load_hashes()

        # Integrity check: if the hash cache claims documents exist but the
        # database is empty (e.g. ChromaDB was deleted/corrupted while the
        # hash sidecar file survived), discard stale hashes and force a full
        # rebuild so every document gets re-indexed.
        if old_hashes and self.collection.count() == 0:
            await log_warning(
                "Hash cache has entries but database is empty — forcing full rebuild",
                ctx,
            )
            old_hashes = {}

        is_full_rebuild = full_rebuild or not old_hashes
        worker_hashes = {} if is_full_rebuild else old_hashes

        all_docs: list[dict[str, Any]] = []
        all_skipped_hashes: dict[str, str] = {}

        # Clone/update repositories and extract documentation in parallel
        # Workers receive old_hashes so they can skip unchanged files early
        # (avoiding expensive notebook conversion for files whose hash matches)
        num_repos = len(target_repos)
        if num_repos > 0:
            max_workers = min(4, num_repos)
            await log_info(f"Processing {num_repos} repositories with {max_workers} workers...", ctx)

            loop = asyncio.get_running_loop()
            with ThreadPoolExecutor(max_workers=max_workers) as executor:
                futures = {
                    loop.run_in_executor(executor, self._clone_and_extract_repo_sync, repo_name, repo_config, worker_hashes): repo_name
                    for repo_name, repo_config in target_repos.items()
                }
                gather_results = await asyncio.gather(*futures.keys(), return_exceptions=True)
                for result, repo_name in zip(gather_results, futures.values(), strict=False):
                    if isinstance(result, BaseException):
                        await log_warning(f"Repository {repo_name} failed: {result}", ctx)
                    elif isinstance(result, dict):
                        all_docs.extend(result["docs"])
                        all_skipped_hashes.update(result["skipped_hashes"])

        clone_extract_elapsed = time.monotonic() - overall_t0
        await log_info(
            f"Clone + extract completed in {clone_extract_elapsed:.1f}s ({len(all_docs)} to index, {len(all_skipped_hashes)} unchanged)",
            ctx,
        )

        # Detect deleted docs: in old hashes for target projects but not extracted or skipped
        target_project_names = set(target_repos.keys())
        all_known_ids = {doc["id"] for doc in all_docs} | set(all_skipped_hashes.keys())
        deleted_doc_ids: list[str] = [
            doc_id for doc_id in old_hashes if doc_id not in all_known_ids and any(doc_id.startswith(f"{proj}___") for proj in target_project_names)
        ]

        changed_docs: list[dict[str, Any]] = []

        if is_full_rebuild:
            if not all_docs:
                await log_warning("No documentation found to index", ctx)
                return
            docs_to_index = all_docs
            await log_info(f"Full rebuild: indexing all {len(docs_to_index)} documents", ctx)
        else:
            # Incremental: all_docs contains only new/changed files (skipped were filtered by workers)
            new_docs = [doc for doc in all_docs if doc["id"] not in old_hashes]
            changed_docs = [doc for doc in all_docs if doc["id"] in old_hashes]

            docs_to_index = all_docs  # all are new or changed (unchanged were skipped by workers)
            await log_info(
                f"Incremental: {len(new_docs)} new, {len(changed_docs)} changed, {len(all_skipped_hashes)} unchanged, {len(deleted_doc_ids)} deleted",
                ctx,
            )

            if not docs_to_index and not deleted_doc_ids:
                total_elapsed = time.monotonic() - overall_t0
                await log_info(f"Index up to date, nothing to do ({total_elapsed:.1f}s)", ctx)
                # Still save hashes (preserves skipped + old non-target hashes)
                new_hashes = dict(old_hashes)
                for doc_id in deleted_doc_ids:
                    new_hashes.pop(doc_id, None)
                self._save_hashes(new_hashes)
                return

        if docs_to_index:
            # Validate for duplicate IDs and log details
            await self._validate_unique_ids(docs_to_index)

        # Split documents into chunks at H1/H2 headers for better embedding quality
        all_chunks: list[dict[str, Any]] = []
        for doc in docs_to_index:
            all_chunks.extend(chunk_document(doc))
        await log_info(f"Chunked {len(docs_to_index)} documents into {len(all_chunks)} chunks", ctx)

        # Create pre-write backup of vector database and hash file
        backup_path = self._backup_path
        try:
            if self._vector_db_path.exists():
                t0 = time.monotonic()
                shutil.copytree(self._vector_db_path, backup_path, dirs_exist_ok=True)
                await log_info(f"Pre-write backup created at {backup_path} ({time.monotonic() - t0:.1f}s)", ctx)
            hash_src = self._hash_file_path
            if hash_src.exists():
                shutil.copy2(hash_src, backup_path.parent / "index_hashes.json.bak")
        except Exception as e:
            await log_warning(f"Failed to create pre-write backup: {e}. Continuing without backup.", ctx)

        if is_full_rebuild:
            # Full rebuild: clear existing collection and re-add everything
            await log_info("Clearing existing index...", ctx)
            try:
                count = self.collection.count()
                if count > 0:
                    results = self.collection.get()
                    if results["ids"]:
                        delete_batch_size = self.chroma_client.get_max_batch_size()
                        for i in range(0, len(results["ids"]), delete_batch_size):
                            self.collection.delete(ids=results["ids"][i : i + delete_batch_size])
            except Exception as e:
                logger.warning(f"Failed to clear existing collection: {e}")
                try:
                    self.chroma_client.delete_collection("holoviz_docs")
                    self.collection = self.chroma_client.get_or_create_collection("holoviz_docs", configuration=_CROMA_CONFIGURATION)
                except Exception as e2:
                    await log_exception(f"Failed to recreate collection: {e2}", ctx)
                    raise
        else:
            # Incremental: delete chunks for changed and deleted docs
            for doc_id in deleted_doc_ids:
                deleted_count = self._delete_doc_chunks(doc_id)
                if deleted_count > 0:
                    logger.debug("Deleted %d chunks for removed doc %s", deleted_count, doc_id)
            for doc in changed_docs:
                deleted_count = self._delete_doc_chunks(doc["id"])
                if deleted_count > 0:
                    logger.debug("Deleted %d old chunks for changed doc %s", deleted_count, doc["id"])

        # Add chunks to ChromaDB in batches (ChromaDB enforces a max batch size)
        if all_chunks:
            batch_size = self.chroma_client.get_max_batch_size()
            await log_info(f"Adding {len(all_chunks)} chunks from {len(docs_to_index)} documents to index (batch size {batch_size})...", ctx)

            try:
                for batch_start in range(0, len(all_chunks), batch_size):
                    batch = all_chunks[batch_start : batch_start + batch_size]
                    self.collection.add(
                        documents=[doc["content"] for doc in batch],
                        metadatas=[
                            {
                                "title": doc["title"],
                                "url": doc["url"],
                                "project": doc["project"],
                                "source_path": doc["source_path"],
                                "source_path_stem": doc["source_path_stem"],
                                "source_url": doc["source_url"],
                                "description": doc["description"],
                                "is_reference": doc["is_reference"],
                                "chunk_index": doc["chunk_index"],
                                "parent_id": doc["parent_id"],
                            }
                            for doc in batch
                        ],
                        ids=[doc["id"] for doc in batch],
                    )
            except (KeyboardInterrupt, SystemExit):
                raise
            except BaseException as e:
                logger.error("ChromaDB write failed (%s: %s). Attempting restore from backup.", type(e).__name__, e)
                await self._restore_from_backup(ctx)
                raise

        # Save updated hashes (after successful ChromaDB write)
        new_hashes = dict(old_hashes)  # start from existing (preserves non-target projects)
        # Remove deleted docs
        for doc_id in deleted_doc_ids:
            new_hashes.pop(doc_id, None)
        # Add/update hashes for indexed docs
        for doc in docs_to_index:
            file_hash = doc.get("file_hash")
            if file_hash:
                new_hashes[doc["id"]] = file_hash
        # Include hashes for files that were skipped (unchanged)
        new_hashes.update(all_skipped_hashes)
        self._save_hashes(new_hashes)

        total_elapsed = time.monotonic() - overall_t0
        await log_info(f"Successfully indexed {len(all_chunks)} chunks from {len(docs_to_index)} documents in {total_elapsed:.1f}s", ctx)
        await log_info(f"Vector database stored at: {self._vector_db_path}", ctx)
        await log_info(f"Index contains {self.collection.count()} total documents", ctx)

        # Show detailed summary table
        await self._log_summary_table(ctx)

    async def _validate_unique_ids(self, all_docs: list[dict[str, Any]], ctx: Context | None = None) -> None:
        """Validate that all document IDs are unique and log duplicates."""
        seen_ids: dict = {}
        duplicates = []

        for doc in all_docs:
            doc_id = doc["id"]
            if doc_id in seen_ids:
                duplicates.append(
                    {
                        "id": doc_id,
                        "first_doc": seen_ids[doc_id],
                        "duplicate_doc": {"project": doc["project"], "source_path": doc["source_path"], "title": doc["title"]},
                    }
                )

                await log_warning(f"DUPLICATE ID FOUND: {doc_id}", ctx)
                await log_warning(f"  First document: {seen_ids[doc_id]['project']}/{seen_ids[doc_id]['source_path']} - {seen_ids[doc_id]['title']}", ctx)
                await log_warning(f"  Duplicate document: {doc['project']}/{doc['source_path']} - {doc['title']}", ctx)
            else:
                seen_ids[doc_id] = {"project": doc["project"], "source_path": doc["source_path"], "title": doc["title"]}

        if duplicates:
            error_msg = f"Found {len(duplicates)} duplicate document IDs"
            await log_exception(error_msg, ctx)

            # Log all duplicates for debugging
            for dup in duplicates:
                await log_exception(
                    f"Duplicate ID '{dup['id']}': {dup['first_doc']['project']}/{dup['first_doc']['source_path']} vs {dup['duplicate_doc']['project']}/{dup['duplicate_doc']['source_path']}",  # noqa: D401, E501
                    ctx,
                )

            raise ValueError(f"Document ID collision detected. {len(duplicates)} duplicate IDs found. Check logs for details.")

    async def search_get_reference_guide(
        self,
        component: str,
        project: str | None = None,
        content: bool = True,
        ctx: Context | None = None,
    ) -> list[Document]:
        """Search for reference guides for a specific component."""
        async with self.db_lock:
            await self.ensure_indexed()

            # Build search strategies
            filters: list[dict[str, Any]] = []
            if project:
                filters.append({"project": str(project)})
            filters.append({"source_path_stem": str(component)})
            filters.append({"is_reference": True})
            where_clause: dict[str, Any] = {"$and": filters} if len(filters) > 1 else filters[0]

            filename_results = self.collection.query(query_texts=[component], n_results=1000, where=where_clause)

            # Group chunks by source_path, then merge content per document
            grouped: dict[str, list[tuple[int, str, Any]]] = {}
            if filename_results["ids"] and filename_results["ids"][0]:
                for i, _ in enumerate(filename_results["ids"][0]):
                    if filename_results["metadatas"] and filename_results["metadatas"][0]:
                        metadata = filename_results["metadatas"][0][i]
                        source_path = _normalize_source_path(str(metadata["source_path"]))

                        # Validate filters
                        if project and str(metadata["project"]) != project:
                            await log_exception(f"Project mismatch for component '{component}': expected '{project}', got '{metadata['project']}'", ctx)
                            continue
                        if metadata["source_path_stem"] != component:
                            await log_exception(f"Path stem mismatch for component '{component}': expected '{component}', got '{metadata['source_path_stem']}'", ctx)
                            continue

                        content_text = filename_results["documents"][0][i] if (content and filename_results["documents"]) else None
                        chunk_index = int(metadata.get("chunk_index", 0) or 0)

                        if source_path not in grouped:
                            grouped[source_path] = []
                        grouped[source_path].append((chunk_index, content_text or "", metadata))

            # Build one Document per unique source_path
            all_results: list[Document] = []
            for source_path, chunks in grouped.items():
                chunks.sort(key=lambda c: c[0])
                metadata = chunks[0][2]

                doc_title = str(metadata["title"])

                # Merge content from all chunks if content was requested.
                # Strip the title prefix that chunk_document() prepends for embedding.
                if content:
                    merged_content: str | None = "\n".join(
                        _strip_title_prefix(
                            c[1],
                            doc_title,
                            project=str(metadata["project"]),
                            source_path=source_path,
                            is_reference=bool(metadata["is_reference"]),
                        )
                        for c in chunks
                    )
                else:
                    merged_content = None

                # Safe URL construction
                url_value = metadata.get("url", "https://example.com")
                if not url_value or url_value == "None" or not isinstance(url_value, str):
                    url_value = "https://example.com"

                document = Document(
                    title=doc_title,
                    url=HttpUrl(url_value),
                    project=str(metadata["project"]),
                    source_path=source_path,
                    source_url=HttpUrl(str(metadata.get("source_url", ""))),
                    description=str(metadata["description"]),
                    is_reference=bool(metadata["is_reference"]),
                    content=merged_content,
                    relevance_score=1.0,
                )
                all_results.append(document)

            return all_results

    def _get_example_paths(self, project: str, limit: int = 5) -> list[str]:
        """Get a few example source_path values for a project.

        Must be called while holding db_lock.

        Parameters
        ----------
        project : str
            The project name.
        limit : int
            Maximum number of example paths to return.

        Returns
        -------
        list[str]
            A small sample of valid source_path values.
        """
        try:
            results = self.collection.get(
                where={"project": project},
                include=["metadatas"],
                limit=limit * 3,  # fetch extra to deduplicate chunks
            )
            if not results["metadatas"]:
                return []
            paths = sorted({_normalize_source_path(str(m.get("source_path", ""))) for m in results["metadatas"] if m.get("source_path")})
            return paths[:limit]
        except (KeyboardInterrupt, SystemExit):
            raise
        except BaseException:
            return []

    def _resolve_path_by_suffix(self, path: str, project: str) -> str | None:
        """Find a stored source_path ending with *path* when the exact lookup misses.

        Returns the full path only when exactly one candidate matches; None otherwise.
        """
        try:
            results = self.collection.get(
                where={"project": project},
                include=["metadatas"],
            )
            if not results["metadatas"]:
                return None

            all_paths = {_normalize_source_path(str(m.get("source_path", ""))) for m in results["metadatas"] if m.get("source_path")}

            suffix = "/" + path.lstrip("/")
            matches = [p for p in all_paths if p.endswith(suffix) or p == path]

            if len(matches) == 1:
                return matches[0]
            return None
        except (KeyboardInterrupt, SystemExit):
            raise
        except BaseException:
            return None

    def _reconstruct_document_content(self, source_path: str, project: str) -> str:
        """Reconstruct full document content from its chunks in ChromaDB.

        Uses collection.get() (metadata filter, no embedding computation)
        to fetch all chunks for a document, sorts by chunk_index, strips
        title prefixes, and joins into a single string.

        Must be called while holding db_lock.

        Parameters
        ----------
        source_path : str
            The source path of the document.
        project : str
            The project name.

        Returns
        -------
        str
            The reconstructed document content, or empty string if not found.
        """
        normalized_source_path = _normalize_source_path(source_path)
        results = self.collection.get(
            where={"$and": [{"project": project}, {"source_path": normalized_source_path}]},
            include=["documents", "metadatas"],
        )
        if not results["ids"]:
            windows_source_path = normalized_source_path.replace("/", "\\")
            if windows_source_path != normalized_source_path:
                results = self.collection.get(
                    where={"$and": [{"project": project}, {"source_path": windows_source_path}]},
                    include=["documents", "metadatas"],
                )
        if not results["ids"]:
            return ""

        chunks: list[tuple[int, str, str]] = []
        for i, _ in enumerate(results["ids"]):
            metadata = results["metadatas"][i] if results["metadatas"] else {}
            content_text = results["documents"][i] if results["documents"] else ""
            chunk_index = int(metadata.get("chunk_index", 0) or 0)
            title = str(metadata.get("title", ""))
            chunks.append((chunk_index, content_text or "", title))

        chunks.sort(key=lambda c: c[0])
        title = chunks[0][2]
        is_ref = bool(results["metadatas"][0].get("is_reference", False)) if results["metadatas"] else False
        return "\n".join(_strip_title_prefix(c[1], title, project=project, source_path=normalized_source_path, is_reference=is_ref) for c in chunks)

    def _extract_documents_from_results(
        self,
        results: dict[str, Any],
        documents: list[Document],
        seen_paths: set[str],
        max_results: int,
        content_mode: str | None,
        max_content_chars: int | None,
        query: str,
    ) -> None:
        """Extract :class:`Document` objects from a ChromaDB query result.

        Appends to *documents* in place, deduplicating by ``source_path``
        via the shared *seen_paths* set.  Stops once *documents* reaches
        *max_results*.

        Must be called while holding ``db_lock``.

        Parameters
        ----------
        results : dict[str, Any]
            Raw ChromaDB ``collection.query()`` return value.
        documents : list[Document]
            Accumulator list — new documents are appended here.
        seen_paths : set[str]
            Already-seen ``source_path`` values for deduplication.
        max_results : int
            Stop after this many documents have been collected.
        content_mode : str | None
            One of ``"chunk"``, ``"truncated"``, ``"full"``, or ``None``.
        max_content_chars : int | None
            Passed through to :func:`truncate_content`.
        query : str
            Original search query (used for context-aware truncation).
        """
        if not (results["ids"] and results["ids"][0]):
            return

        for i, _ in enumerate(results["ids"][0]):
            if len(documents) >= max_results:
                break

            if not (results["metadatas"] and results["metadatas"][0]):
                continue

            metadata = results["metadatas"][0][i]

            # Deduplicate by source_path — keep only the best-scoring chunk per document
            source_path = str(metadata["source_path"])
            if source_path in seen_paths:
                continue
            seen_paths.add(source_path)

            # Resolve content based on content mode
            if content_mode is None:
                content_text = None
            elif content_mode == "chunk":
                content_text = results["documents"][0][i] if results["documents"] else None
                if content_text:
                    content_text = _strip_title_prefix(
                        content_text,
                        str(metadata["title"]),
                        project=str(metadata.get("project", "")),
                        source_path=source_path,
                        is_reference=bool(metadata.get("is_reference", False)),
                    )
                content_text = truncate_content(content_text, max_content_chars, query=query)
            else:
                # "truncated", "full", or unknown mode — reconstruct full document
                content_text = self._reconstruct_document_content(source_path, str(metadata["project"]))
                if content_mode != "full":
                    content_text = truncate_content(content_text, max_content_chars, query=query)

            # Safe URL construction
            url_value = metadata.get("url", "https://example.com")
            if not url_value or url_value == "None" or not isinstance(url_value, str):
                url_value = "https://example.com"

            # Safe relevance score calculation
            relevance_score = None
            if (
                results.get("distances")
                and isinstance(results["distances"], list)
                and len(results["distances"]) > 0
                and isinstance(results["distances"][0], list)
                and len(results["distances"][0]) > i
            ):
                try:
                    relevance_score = (2.0 - float(results["distances"][0][i])) / 2.0
                except (ValueError, TypeError):
                    relevance_score = None

            document = Document(
                title=str(metadata["title"]),
                url=HttpUrl(url_value),
                project=str(metadata["project"]),
                source_path=source_path,
                source_url=HttpUrl(str(metadata.get("source_url", ""))),
                description=str(metadata["description"]),
                is_reference=bool(metadata["is_reference"]),
                content=content_text,
                relevance_score=relevance_score,
            )
            documents.append(document)

    def _merge_search_results(
        self,
        metadata_results: dict[str, Any] | None,
        keyword_results: dict[str, Any] | None,
        semantic_results: dict[str, Any],
        max_results: int,
        content_mode: str | None,
        max_content_chars: int | None,
        query: str,
    ) -> list[Document]:
        """Merge metadata-boosted, keyword-filtered and semantic search results.

        Results are merged in priority order: metadata boost (exact stem
        match) > keyword pre-filter (content ``$contains``) > pure semantic
        similarity.  Deduplication by ``source_path`` is maintained across
        all passes.

        Must be called while holding ``db_lock``.

        Parameters
        ----------
        metadata_results : dict[str, Any] | None
            ChromaDB query result with ``source_path_stem`` metadata filter,
            or ``None`` when no PascalCase terms were found.
        keyword_results : dict[str, Any] | None
            ChromaDB query result with ``where_document`` pre-filter, or
            ``None`` when no technical terms were found.
        semantic_results : dict[str, Any]
            ChromaDB query result from pure semantic similarity.
        max_results : int
            Maximum number of documents to return.
        content_mode : str | None
            Content resolution mode.
        max_content_chars : int | None
            Maximum content characters.
        query : str
            Original search query.

        Returns
        -------
        list[Document]
            Merged, deduplicated document list.
        """
        documents: list[Document] = []
        seen_paths: set[str] = set()

        # Pass 0: metadata-boosted results (exact stem match)
        if metadata_results is not None:
            self._extract_documents_from_results(
                metadata_results,
                documents,
                seen_paths,
                max_results,
                content_mode,
                max_content_chars,
                query,
            )

        # Pass 1: keyword-filtered results (content $contains)
        if keyword_results is not None and len(documents) < max_results:
            self._extract_documents_from_results(
                keyword_results,
                documents,
                seen_paths,
                max_results,
                content_mode,
                max_content_chars,
                query,
            )

        # Pass 2: fill remaining slots from semantic results
        if len(documents) < max_results:
            self._extract_documents_from_results(
                semantic_results,
                documents,
                seen_paths,
                max_results,
                content_mode,
                max_content_chars,
                query,
            )

        return documents

    async def search(
        self,
        query: str,
        project: str | None = None,
        content: str | bool = "truncated",
        max_results: int = 5,
        max_content_chars: int | None = 10000,
        ctx: Context | None = None,
    ) -> list[Document]:
        """Search the documentation using semantic similarity."""
        async with self.db_lock:
            await self.ensure_indexed(ctx=ctx)

            # Normalize content parameter for backward compatibility
            if content is True:
                content_mode = "truncated"
            elif content is False or content is None:
                content_mode = None
            else:
                content_mode = str(content)

            # Build where clause for filtering
            where_clause = {"project": str(project)} if project else None

            # Over-query to allow deduplication across chunks of the same document
            n_results = max_results * 3

            # Extract technical terms for keyword pre-filtering
            tech_terms = extract_tech_terms(query)
            # Extract PascalCase terms for metadata boost + content matching
            pascal_terms = extract_pascal_terms(query)

            # When a project filter is active, drop terms that match the
            # project name itself — every document in the project trivially
            # contains the project name, so the pre-filter adds no selectivity
            # and fills merge slots with irrelevant docs.
            if project and (tech_terms or pascal_terms):
                project_lower = project.lower().replace("-", "")
                tech_terms = [t for t in tech_terms if t.lower().replace("-", "") != project_lower]
                pascal_terms = [t for t in pascal_terms if t.lower().replace("-", "") != project_lower]

            # Combine for content pre-filter: tech_terms + pascal_terms (deduplicated)
            all_content_terms = list(dict.fromkeys(tech_terms + pascal_terms))
            where_doc = _build_where_document_clause(all_content_terms)

            # Query 0: metadata boost on source_path_stem
            metadata_results = None
            if pascal_terms:
                stem_clause = _build_stem_boost_clause(pascal_terms, project)
                if stem_clause:
                    metadata_results = self.collection.query(
                        query_texts=[query],
                        n_results=n_results,
                        where=stem_clause,  # type: ignore[arg-type]
                    )

            # Query 1: keyword-filtered (only when content terms are found)
            keyword_results = None
            if where_doc:
                keyword_results = self.collection.query(
                    query_texts=[query],
                    n_results=n_results,
                    where=where_clause,
                    where_document=where_doc,  # type: ignore[arg-type]
                )

            # Query 2: pure semantic similarity (always)
            semantic_results = self.collection.query(query_texts=[query], n_results=n_results, where=where_clause)  # type: ignore[arg-type]

            return self._merge_search_results(
                metadata_results,
                keyword_results,
                semantic_results,
                max_results,
                content_mode,
                max_content_chars,
                query,
            )

    async def get_document(self, path: str, project: str, ctx: Context | None = None) -> Document:
        """Get a specific document, reconstructing from chunks if needed."""
        async with self.db_lock:
            await self.ensure_indexed(ctx=ctx)

            normalized_path = _normalize_source_path(path)

            # Reconstruct full content from chunks
            merged_content = self._reconstruct_document_content(normalized_path, project)

            if not merged_content:
                resolved_path = self._resolve_path_by_suffix(normalized_path, project)
                if resolved_path:
                    normalized_path = resolved_path
                    merged_content = self._reconstruct_document_content(normalized_path, project)

            if not merged_content:
                # Provide example paths from this project to help discoverability
                example_paths = self._get_example_paths(project, limit=5)
                examples_str = ""
                if example_paths:
                    examples_str = f" Example paths for '{project}': {', '.join(example_paths)}"
                raise ValueError(
                    f"No document found for path '{path}' in project '{project}'.{examples_str} "
                    f"Tip: Use the search tool with content=False to discover valid document paths (look at the source_path field)."
                )

            # Get metadata from a single chunk (for Document fields)
            results = self.collection.get(
                where={"$and": [{"project": project}, {"source_path": normalized_path}]},
                include=["metadatas"],
                limit=1,
            )
            if not results["metadatas"]:
                windows_path = normalized_path.replace("/", "\\")
                if windows_path != normalized_path:
                    results = self.collection.get(
                        where={"$and": [{"project": project}, {"source_path": windows_path}]},
                        include=["metadatas"],
                        limit=1,
                    )
            metadata = results["metadatas"][0] if results["metadatas"] else {}

            # Safe URL construction
            url_value = metadata.get("url", "https://example.com")
            if not url_value or url_value == "None" or not isinstance(url_value, str):
                url_value = "https://example.com"

            return Document(
                title=str(metadata.get("title", "")),
                url=HttpUrl(url_value),
                project=str(metadata.get("project", "")),
                source_path=_normalize_source_path(str(metadata.get("source_path", ""))),
                source_url=HttpUrl(str(metadata.get("source_url", ""))),
                description=str(metadata.get("description", "")),
                is_reference=bool(metadata.get("is_reference", False)),
                content=merged_content,
                relevance_score=None,
            )

    async def list_projects(self) -> list[str]:
        """List all available projects with documentation in the index.

        Returns
        -------
        list[str]: A list of project names that have documentation available.
                   Names are returned in hyphenated format (e.g., "panel-material-ui").
        """
        async with self.db_lock:
            await self.ensure_indexed()

            try:
                # Get all documents from the collection to extract unique project names
                results = self.collection.get()

                if not results["metadatas"]:
                    return []

                # Extract unique project names
                projects = set()
                for metadata in results["metadatas"]:
                    project = metadata.get("project")
                    if project:
                        # Convert underscored names to hyphenated format for consistency
                        project_name = str(project).replace("_", "-")
                        projects.add(project_name)

                # Return sorted list
                return sorted(projects)

            except Exception as e:
                logger.error(f"Failed to list projects: {e}")
                return []

    async def list_documents(self, project: str) -> list[dict[str, str | bool]]:
        """List all documents available for a given project.

        Parameters
        ----------
        project : str
            The project name (e.g., "panel", "hvplot").

        Returns
        -------
        list[dict[str, str | bool]]
            Sorted list of dicts with keys: "source_path", "title", "is_reference".

        Raises
        ------
        ValueError
            If no documents are found for the given project.
        """
        async with self.db_lock:
            await self.ensure_indexed()

            # Normalize project name (hyphen to underscore for ChromaDB lookup)
            project_key = project.replace("-", "_")

            try:
                results = self.collection.get(where={"project": project_key}, include=["metadatas"])

                if not results["metadatas"]:
                    msg = f"No documents found for project '{project}'. Use project_list to see available projects."
                    raise ValueError(msg)

                # Deduplicate chunks by source_path
                seen: dict[str, dict[str, str | bool]] = {}
                for metadata in results["metadatas"]:
                    source_path = str(metadata.get("source_path", ""))
                    if source_path in seen:
                        continue
                    seen[source_path] = {
                        "source_path": source_path,
                        "title": str(metadata.get("title", "")),
                        "is_reference": bool(metadata.get("is_reference", False)),
                    }

                return sorted(seen.values(), key=lambda d: d["source_path"])

            except ValueError:
                raise
            except Exception as e:
                logger.error(f"Failed to list documents for project '{project}': {e}")
                return []

    async def _log_summary_table(self, ctx: Context | None = None):
        """Log a summary table showing document counts by repository."""
        try:
            # Get all documents from the collection
            results = self.collection.get()

            if not results["metadatas"]:
                await log_info("No documents found in index", ctx)
                return

            # Count unique documents (not chunks) by project and type
            project_stats: dict[str, dict[str, int]] = {}
            seen_docs: set[str] = set()
            total_chunks = 0

            for metadata in results["metadatas"]:
                total_chunks += 1
                project = str(metadata.get("project", "unknown"))
                source_path = str(metadata.get("source_path", ""))
                is_reference = metadata.get("is_reference", False)

                # Only count each unique document once (not each chunk)
                doc_key = f"{project}::{source_path}"
                if doc_key in seen_docs:
                    continue
                seen_docs.add(doc_key)

                if project not in project_stats:
                    project_stats[project] = {"total": 0, "regular": 0, "reference": 0}

                project_stats[project]["total"] += 1
                if is_reference:
                    project_stats[project]["reference"] += 1
                else:
                    project_stats[project]["regular"] += 1

            # Log summary table
            await log_info("", ctx)
            await log_info("📊 Document Summary by Repository:", ctx)
            await log_info("=" * 60, ctx)
            await log_info(f"{'Repository':<20} {'Total':<8} {'Regular':<8} {'Reference':<10}", ctx)
            await log_info("-" * 60, ctx)

            total_docs = 0
            total_regular = 0
            total_reference = 0

            for project in sorted(project_stats.keys()):
                stats = project_stats[project]
                await log_info(f"{project:<20} {stats['total']:<8} {stats['regular']:<8} {stats['reference']:<10}", ctx)
                total_docs += stats["total"]
                total_regular += stats["regular"]
                total_reference += stats["reference"]

            await log_info("-" * 60, ctx)
            await log_info(f"{'TOTAL':<20} {total_docs:<8} {total_regular:<8} {total_reference:<10}", ctx)
            await log_info("=" * 60, ctx)
            await log_info(f"Total chunks in vector database: {total_chunks}", ctx)

        except Exception as e:
            await log_warning(f"Failed to generate summary table: {e}", ctx)

    def run(self, projects: list[str] | None = None, full_rebuild: bool = False):
        """Update the DocumentationIndexer.

        Parameters
        ----------
        projects : list[str] | None
            Only process these projects. None means all.
        full_rebuild : bool
            Force full rebuild, ignoring cached hashes.
        """
        # Configure logging for the CLI
        logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler()])

        logger.info("HoloViz MCP Documentation Indexer")
        logger.info("=" * 50)

        async def run_indexer(indexer=self):
            logger.info(f"Default config: {indexer._holoviz_mcp_config.config_file_path(location='default')}")
            logger.info(f"User config: {indexer._holoviz_mcp_config.config_file_path(location='user')}")
            logger.info(f"Repository directory: {indexer.repos_dir}")
            logger.info(f"Vector database: {self._vector_db_path}")
            logger.info(f"Hash cache: {self._hash_file_path}")
            logger.info(f"Configured repositories: {len(indexer.config.repositories)}")
            if projects:
                logger.info(f"Target projects: {', '.join(projects)}")
            if full_rebuild:
                logger.info("Mode: full rebuild (ignoring hashes)")
            logger.info("")

            await indexer.index_documentation(projects=projects, full_rebuild=full_rebuild)

            # Final summary
            count = indexer.collection.count()
            logger.info("")
            logger.info("=" * 50)
            logger.info("Indexing completed successfully!")
            logger.info(f"Total documents in database: {count}")
            logger.info("=" * 50)

        asyncio.run(run_indexer())


def main():
    """Run the documentation indexer (full rebuild, called from legacy entry points)."""
    DocumentationIndexer().run()


if __name__ == "__main__":
    main()