scripts/source_retrieval.py

#!/usr/bin/env python3
"""Build and query a compact official-source retrieval database."""

from __future__ import annotations

import argparse
import hashlib
import html
import importlib
import math
import os
from contextlib import contextmanager
from html.parser import HTMLParser
import json
import re
import sqlite3
import sys
from dataclasses import asdict, dataclass
from pathlib import Path
from types import SimpleNamespace
from typing import Any
from urllib.parse import urljoin, urlparse
from urllib.request import Request, urlopen

DEFAULT_SOURCES_PATH = Path("benchmarks/sources.json")
DEFAULT_NOTES_PATH = Path("retrieval/source_passage_notes.jsonl")
DEFAULT_DB_PATH = Path("outputs/source_retrieval/official_sources.harrier.sqlite3")
DEFAULT_DOC_CACHE_DIR = Path("outputs/source_retrieval/document_cache")
DEFAULT_NOLEGRAPH_SRC = Path("/Users/hasse_h/nolegraph/src")
DEFAULT_EMBEDDING_MODEL = "microsoft/harrier-oss-v1-0.6b"
QUERY_INSTRUCTION = "Represent this query for retrieving relevant passages: "
MAX_SEARCH_TERMS = 64
MAX_CRAWL_LINKS = 500
MIN_PARAGRAPH_CHARS = 35
EMBED_BATCH_SIZE = 8
EMBED_TEXT_MAX_CHARS = 1800
USER_AGENT = "Synderesis-SourceIndexer/0.1 (+local research cache)"

WORD_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9']+")
EMBED_TOKEN_RE = re.compile(r"[\w-]+", re.UNICODE)
OFFICIAL_LOCATION_RE = re.compile(r"^\s*(?:no\.\s*)?(\d{1,4})(?:[\.\s:,-]|$)", re.IGNORECASE)
STOPWORDS = {
    "about",
    "after",
    "also",
    "and",
    "are",
    "background",
    "based",
    "because",
    "been",
    "being",
    "but",
    "can",
    "could",
    "does",
    "document",
    "documents",
    "for",
    "from",
    "go",
    "goes",
    "has",
    "have",
    "how",
    "into",
    "its",
    "like",
    "make",
    "means",
    "not",
    "one",
    "other",
    "out",
    "post",
    "reach",
    "reflects",
    "regardless",
    "says",
    "should",
    "that",
    "the",
    "their",
    "there",
    "this",
    "through",
    "together",
    "togheter",
    "what",
    "when",
    "where",
    "while",
    "with",
    "word",
}
SYNONYMS = {
    "catholic": ["catholic", "universal", "church"],
    "church": ["church", "catholic", "mission"],
    "convert": ["convert", "conversion", "proselytism", "evangelization", "mission", "witness"],
    "conversion": ["convert", "conversion", "proselytism", "evangelization", "mission", "witness"],
    "evangelize": ["evangelization", "mission", "witness", "proclaim"],
    "jew": ["jew", "jews", "jewish", "judaism", "israel"],
    "jews": ["jew", "jews", "jewish", "judaism", "israel"],
    "jewish": ["jew", "jews", "jewish", "judaism", "israel"],
    "mission": ["mission", "evangelization", "witness", "proclaim", "universal"],
    "universal": ["universal", "catholic", "whole", "human", "mission"],
    "vatican": ["vatican", "council", "conciliar"],
}


class SourceRetrievalError(Exception):
    """Raised when the source retrieval database cannot be built or queried."""


@dataclass(frozen=True)
class SourceChunk:
    """One compact searchable source passage note."""

    source_id: str
    chunk_kind: str
    title: str
    url: str
    publisher: str
    source_type: str
    location: str
    paragraph_index: int
    topics: list[str]
    summary: str
    text: str
    keywords: list[str]


@dataclass(frozen=True)
class SearchResult:
    """One retrieved source passage."""

    source_id: str
    chunk_kind: str
    title: str
    url: str
    publisher: str
    source_type: str
    location: str
    paragraph_index: int
    topics: list[str]
    summary: str
    text: str
    keywords: list[str]
    score: float

    def to_dict(self) -> dict[str, Any]:
        """Return an API-safe dictionary representation."""
        return asdict(self)


def embed_tokenize(text: str) -> list[str]:
    """Tokenize text for the dependency-free hash embedder."""
    return [token.casefold() for token in EMBED_TOKEN_RE.findall(text) if len(token) > 1]


def l2_normalize(vector: list[float]) -> list[float]:
    """Return an L2-normalized vector."""
    norm = math.sqrt(sum(value * value for value in vector))
    if norm == 0.0:
        return vector
    return [value / norm for value in vector]


class HashEmbedder:
    """Deterministic, dependency-free feature-hashing embedder."""

    def __init__(self, dim: int = 256) -> None:
        if dim <= 0:
            raise ValueError("dim must be positive")
        self._dim = dim

    @property
    def dim(self) -> int:
        return self._dim

    @property
    def signature(self) -> str:
        return f"hash-{self._dim}"

    def encode(self, texts: list[str], *, is_query: bool = False) -> list[list[float]]:
        """Encode text using stable token hashing."""
        vectors: list[list[float]] = []
        for text in texts:
            source = (QUERY_INSTRUCTION + text) if is_query else text
            vector = [0.0] * self._dim
            for token in embed_tokenize(source):
                digest = hashlib.sha256(token.encode("utf-8")).digest()
                bucket = int.from_bytes(digest[:8], "big") % self._dim
                vector[bucket] += 1.0
            vectors.append(l2_normalize(vector))
        return vectors


class HarrierEmbedder:
    """sentence-transformers-backed Harrier embedder compatible with nolegraph."""

    def __init__(self, model_name: str = DEFAULT_EMBEDDING_MODEL) -> None:
        module = importlib.import_module("sentence_transformers")
        transformer_cls: Any = module.SentenceTransformer
        self._model_name = model_name
        try:
            self._model: Any = transformer_cls(model_name, local_files_only=True)
        except Exception:
            self._model = transformer_cls(model_name)
        get_dim = getattr(self._model, "get_embedding_dimension", self._model.get_sentence_embedding_dimension)
        dimension = get_dim()
        self._dim = int(dimension) if dimension is not None else 1024

    @property
    def dim(self) -> int:
        return self._dim

    @property
    def signature(self) -> str:
        return f"st:{self._model_name}:{self._dim}"

    def encode(self, texts: list[str], *, is_query: bool = False) -> list[list[float]]:
        """Encode text using Harrier and normalized embeddings."""
        prepared = [QUERY_INSTRUCTION + text for text in texts] if is_query else list(texts)
        raw = self._model.encode(prepared, normalize_embeddings=True)
        if hasattr(raw, "tolist"):
            raw = raw.tolist()
        return [[float(value) for value in row] for row in raw]


def load_default_embedder() -> Any | None:
    """Load the default Harrier embedder only when explicitly enabled."""
    flag = os.environ.get("NOLEGRAPH_AUTOLOAD_EMBEDDER", "").strip().lower()
    if flag not in {"1", "true", "yes", "on"}:
        return None
    model_name = os.environ.get("NOLEGRAPH_EMBEDDING_MODEL", DEFAULT_EMBEDDING_MODEL)
    try:
        return HarrierEmbedder(model_name)
    except Exception as error:
        print(f"source_retrieval: embedder {model_name!r} failed to load, falling back to lexical search: {error}", file=sys.stderr)
        return None


def load_sources(path: Path) -> dict[str, dict[str, Any]]:
    """Load the official source registry keyed by source id."""
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError as exc:
        raise SourceRetrievalError(f"source registry not found: {path}") from exc
    except json.JSONDecodeError as exc:
        raise SourceRetrievalError(f"source registry is invalid JSON: {path}") from exc
    sources = payload.get("sources")
    if not isinstance(sources, list):
        raise SourceRetrievalError("source registry must contain a sources list")
    indexed: dict[str, dict[str, Any]] = {}
    for source in sources:
        if not isinstance(source, dict) or not isinstance(source.get("id"), str):
            raise SourceRetrievalError("each source registry entry must include an id")
        indexed[source["id"]] = source
    return indexed


def load_notes(path: Path, sources: dict[str, dict[str, Any]]) -> dict[tuple[str, str], dict[str, Any]]:
    """Load curated passage notes keyed by source id and location."""
    notes: dict[tuple[str, str], dict[str, Any]] = {}
    if not path.exists():
        return notes
    for line_number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
        line = raw_line.strip()
        if not line:
            continue
        try:
            note = json.loads(line)
        except json.JSONDecodeError as exc:
            raise SourceRetrievalError(f"invalid JSONL in {path}:{line_number}") from exc
        source_id = note.get("source_id")
        location = note.get("location")
        summary = note.get("summary")
        if not isinstance(source_id, str) or source_id not in sources:
            raise SourceRetrievalError(f"unknown source_id in {path}:{line_number}")
        if not isinstance(location, str) or not location.strip():
            raise SourceRetrievalError(f"missing location in {path}:{line_number}")
        if not isinstance(summary, str) or not summary.strip():
            raise SourceRetrievalError(f"missing summary in {path}:{line_number}")
        keywords = note.get("keywords", [])
        if isinstance(keywords, str):
            keywords = [keywords]
        if not isinstance(keywords, list) or not all(isinstance(item, str) for item in keywords):
            raise SourceRetrievalError(f"keywords must be strings in {path}:{line_number}")
        notes[(source_id, location)] = {"summary": summary.strip(), "keywords": [item.strip() for item in keywords if item.strip()]}
    return notes


def registry_summary(source: dict[str, Any], location: str) -> str:
    """Create a fallback searchable summary from registry metadata."""
    topics = ", ".join(str(topic) for topic in source.get("topics", []))
    return f"{source.get('title', source['id'])} {location}. Official {source.get('type', 'source')} covering {topics}."


def registry_keywords(source: dict[str, Any], location: str) -> list[str]:
    """Create fallback retrieval keywords from registry metadata."""
    values = [source.get("title", ""), source.get("official_id", ""), source.get("type", ""), location]
    values.extend(str(topic) for topic in source.get("topics", []))
    return [value for value in values if value]


def clean_text(value: str) -> str:
    """Normalize HTML-derived whitespace."""
    return re.sub(r"\s+", " ", html.unescape(value)).strip()


class VaticanHtmlParser(HTMLParser):
    """Extract links and block-level text from official source HTML."""

    block_tags = {"p", "li", "h1", "h2", "h3", "h4", "h5", "h6", "blockquote"}
    skip_tags = {"script", "style", "noscript"}

    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self.links: list[str] = []
        self.paragraphs: list[str] = []
        self._skip_depth = 0
        self._block_depth = 0
        self._buffer: list[str] = []

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        tag = tag.lower()
        if tag in self.skip_tags:
            self._skip_depth += 1
            return
        if tag == "a":
            for key, value in attrs:
                if key.lower() == "href" and value:
                    self.links.append(value)
        if tag in self.block_tags:
            if self._block_depth == 0:
                self._buffer = []
            self._block_depth += 1

    def handle_endtag(self, tag: str) -> None:
        tag = tag.lower()
        if tag in self.skip_tags and self._skip_depth:
            self._skip_depth -= 1
            return
        if tag in self.block_tags and self._block_depth:
            self._block_depth -= 1
            if self._block_depth == 0:
                text = clean_text(" ".join(self._buffer))
                if len(text) >= MIN_PARAGRAPH_CHARS:
                    self.paragraphs.append(text)
                self._buffer = []

    def handle_data(self, data: str) -> None:
        if self._skip_depth or not self._block_depth:
            return
        value = clean_text(data)
        if value:
            self._buffer.append(value)


def noisy_paragraph(text: str) -> bool:
    """Filter obvious navigation and language-switcher fragments."""
    normalized = clean_text(text)
    if len(normalized) < MIN_PARAGRAPH_CHARS:
        return True
    if normalized.startswith("[") and normalized.endswith("]") and len(normalized) < 220:
        return True
    lowered = normalized.lower()
    noisy_exact = {
        "the holy see",
        "copyright",
        "vatican.va",
    }
    if lowered in noisy_exact:
        return True
    if lowered.count(" - ") >= 5 and len(normalized) < 260:
        return True
    return False


def parse_with_bs4(html_text: str) -> tuple[list[str], list[str]] | None:
    """Extract links and paragraphs with BeautifulSoup when available."""
    try:
        from bs4 import BeautifulSoup
    except Exception:
        return None
    soup = BeautifulSoup(html_text, "html.parser")
    for tag in soup(["script", "style", "noscript", "nav", "header", "footer"]):
        tag.decompose()
    links = [str(tag.get("href")) for tag in soup.find_all("a") if tag.get("href")]
    paragraphs: list[str] = []
    seen: set[str] = set()
    for tag in soup.find_all(["h1", "h2", "h3", "h4", "h5", "h6", "p", "li", "blockquote"]):
        text = clean_text(tag.get_text(" ", strip=True))
        if noisy_paragraph(text) or text in seen:
            continue
        seen.add(text)
        paragraphs.append(text)

    return links, paragraphs


def cache_path_for_url(cache_dir: Path, source_id: str, url: str) -> Path:
    """Return the local HTML cache path for a source URL."""
    digest = hashlib.sha256(url.encode("utf-8")).hexdigest()[:16]
    return cache_dir / source_id / f"{digest}.html"


def fetch_url(url: str, cache_dir: Path, source_id: str, force_fetch: bool = False) -> str:
    """Fetch URL with a generated local cache under outputs."""
    cache_path = cache_path_for_url(cache_dir, source_id, url)
    if cache_path.exists() and not force_fetch:
        return cache_path.read_text(encoding="utf-8", errors="replace")
    cache_path.parent.mkdir(parents=True, exist_ok=True)
    request = Request(url, headers={"User-Agent": USER_AGENT})
    with urlopen(request, timeout=30) as response:
        raw = response.read()
        charset = response.headers.get_content_charset() or "utf-8"
    text = raw.decode(charset, errors="replace")
    cache_path.write_text(text, encoding="utf-8")
    return text


def parse_html_document(html_text: str) -> VaticanHtmlParser:
    """Parse one HTML document into paragraphs and links."""
    bs4_result = parse_with_bs4(html_text)
    if bs4_result is not None:
        links, paragraphs = bs4_result
        parser = VaticanHtmlParser()
        parser.links = links
        parser.paragraphs = paragraphs
        return parser
    parser = VaticanHtmlParser()
    parser.feed(html_text)
    parser.close()
    parser.paragraphs = [paragraph for paragraph in parser.paragraphs if not noisy_paragraph(paragraph)]
    return parser


def same_document_link(base_url: str, candidate_url: str) -> bool:
    """Return whether a link should be crawled as part of the same source."""
    base = urlparse(base_url)
    candidate = urlparse(candidate_url)
    if candidate.scheme not in {"http", "https"}:
        return False
    if candidate.netloc != base.netloc:
        return False
    base_dir = base.path.rsplit("/", 1)[0] + "/"
    if not candidate.path.startswith(base_dir):
        return False
    return candidate.path.lower().endswith((".htm", ".html"))


def document_urls_for_source(source: dict[str, Any], cache_dir: Path, force_fetch: bool) -> list[str]:
    """Return URLs to fetch for one source, crawling Catechism-style index pages."""
    source_id = str(source["id"])
    base_url = str(source.get("url", ""))
    if not base_url:
        return []
    urls = [base_url]
    if base_url.upper().endswith("_INDEX.HTM"):
        parser = parse_html_document(fetch_url(base_url, cache_dir, source_id, force_fetch))
        discovered: list[str] = []
        for href in parser.links:
            absolute = urljoin(base_url, href.split("#", 1)[0])
            if same_document_link(base_url, absolute) and absolute not in urls and absolute not in discovered:
                discovered.append(absolute)
            if len(discovered) >= MAX_CRAWL_LINKS:
                break
        urls.extend(discovered)
    return urls


def official_location(text: str, paragraph_index: int, used: set[str]) -> str:
    """Infer an official paragraph number when present, otherwise use a local paragraph id."""
    match = OFFICIAL_LOCATION_RE.match(text)
    if match:
        candidate = match.group(1)
        if candidate not in used:
            used.add(candidate)
            return candidate
    fallback = f"p{paragraph_index:04d}"
    used.add(fallback)
    return fallback


def paragraph_summary(text: str, max_chars: int = 700) -> str:
    """Return a bounded paragraph summary/excerpt."""
    if len(text) <= max_chars:
        return text
    clipped = text[:max_chars].rsplit(" ", 1)[0].rstrip()
    return f"{clipped}..."


def document_chunks(
    sources: dict[str, dict[str, Any]],
    cache_dir: Path,
    force_fetch: bool = False,
) -> list[SourceChunk]:
    """Fetch official documents and return paragraph-level chunks."""
    chunks: list[SourceChunk] = []
    for source_id, source in sources.items():
        topics = [str(topic) for topic in source.get("topics", [])]
        used_locations: set[str] = set()
        paragraph_index = 0
        for url in document_urls_for_source(source, cache_dir, force_fetch):
            parser = parse_html_document(fetch_url(url, cache_dir, source_id, force_fetch))
            for paragraph in parser.paragraphs:
                text = clean_text(paragraph)
                if len(text) < MIN_PARAGRAPH_CHARS:
                    continue
                paragraph_index += 1
                location = official_location(text, paragraph_index, used_locations)
                chunks.append(
                    SourceChunk(
                        source_id=source_id,
                        chunk_kind="document",
                        title=str(source.get("title", source_id)),
                        url=url,
                        publisher=str(source.get("publisher", "")),
                        source_type=str(source.get("type", "")),
                        location=location,
                        paragraph_index=paragraph_index,
                        topics=topics,
                        summary=paragraph_summary(text),
                        text=text,
                        keywords=registry_keywords(source, location),
                    )
                )
    return chunks


def load_chunks(
    sources_path: Path,
    notes_path: Path,
    *,
    include_documents: bool = False,
    cache_dir: Path = DEFAULT_DOC_CACHE_DIR,
    force_fetch: bool = False,
) -> list[SourceChunk]:
    """Build source chunks from registry key refs plus curated notes."""
    sources = load_sources(sources_path)
    notes = load_notes(notes_path, sources)
    chunks: dict[tuple[str, str, str, int], SourceChunk] = {}
    for source_id, source in sources.items():
        key_refs = source.get("key_refs", [])
        if not isinstance(key_refs, list):
            raise SourceRetrievalError(f"{source_id} key_refs must be a list")
        topics = [str(topic) for topic in source.get("topics", [])]
        for raw_location in key_refs:
            location = str(raw_location)
            note = notes.get((source_id, location), {})
            summary = str(note.get("summary") or registry_summary(source, location))
            chunk_kind = "note" if note else "registry"
            chunks[(source_id, location, chunk_kind, 0)] = SourceChunk(
                source_id=source_id,
                chunk_kind=chunk_kind,
                title=str(source.get("title", source_id)),
                url=str(source.get("url", "")),
                publisher=str(source.get("publisher", "")),
                source_type=str(source.get("type", "")),
                location=location,
                paragraph_index=0,
                topics=topics,
                summary=summary,
                text=summary,
                keywords=list(note.get("keywords") or registry_keywords(source, location)),
            )
    for (source_id, location), note in notes.items():
        if (source_id, location, "note", 0) in chunks:
            continue
        source = sources[source_id]
        topics = [str(topic) for topic in source.get("topics", [])]
        chunks[(source_id, location, "note", 0)] = SourceChunk(
            source_id=source_id,
            chunk_kind="note",
            title=str(source.get("title", source_id)),
            url=str(source.get("url", "")),
            publisher=str(source.get("publisher", "")),
            source_type=str(source.get("type", "")),
            location=location,
            paragraph_index=0,
            topics=topics,
            summary=str(note["summary"]),
            text=str(note["summary"]),
            keywords=list(note.get("keywords", [])),
        )
    if include_documents:
        for chunk in document_chunks(sources, cache_dir, force_fetch):
            chunks[(chunk.source_id, chunk.location, chunk.chunk_kind, chunk.paragraph_index)] = chunk
    return sorted(chunks.values(), key=lambda chunk: (chunk.source_id, chunk.chunk_kind, chunk.paragraph_index, chunk.location))


def connect(db_path: Path) -> sqlite3.Connection:
    """Open a SQLite connection for source retrieval."""
    connection = sqlite3.connect(db_path)
    connection.row_factory = sqlite3.Row
    return connection


@contextmanager
def db_connection(db_path: Path) -> Any:
    """Open a source-retrieval SQLite connection and always close it."""
    connection = connect(db_path)
    try:
        yield connection
        connection.commit()
    finally:
        connection.close()


def init_source_db(connection: sqlite3.Connection) -> None:
    """Create source retrieval tables."""
    connection.executescript(
        """
        CREATE TABLE IF NOT EXISTS source_chunks (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            source_id TEXT NOT NULL,
            chunk_kind TEXT NOT NULL,
            title TEXT NOT NULL,
            url TEXT NOT NULL,
            publisher TEXT NOT NULL,
            source_type TEXT NOT NULL,
            location TEXT NOT NULL,
            paragraph_index INTEGER NOT NULL,
            topics TEXT NOT NULL,
            summary TEXT NOT NULL,
            text TEXT NOT NULL,
            keywords TEXT NOT NULL,
            UNIQUE(source_id, location, chunk_kind, paragraph_index)
        );

        CREATE TABLE IF NOT EXISTS source_chunk_embeddings (
            chunk_id INTEGER PRIMARY KEY,
            vector TEXT NOT NULL,
            content_hash TEXT NOT NULL,
            embedder_signature TEXT NOT NULL,
            FOREIGN KEY(chunk_id) REFERENCES source_chunks(id) ON DELETE CASCADE
        );

        CREATE TABLE IF NOT EXISTS meta (
            key TEXT PRIMARY KEY,
            value TEXT NOT NULL
        );

        CREATE VIRTUAL TABLE IF NOT EXISTS source_chunks_fts USING fts5(
            source_id,
            chunk_kind,
            title,
            location,
            topics,
            summary,
            text,
            keywords,
            content='source_chunks',
            content_rowid='id',
            tokenize='unicode61 remove_diacritics 2'
        );
        """
    )


def insert_chunk(connection: sqlite3.Connection, chunk: SourceChunk) -> None:
    """Insert one source chunk and its FTS row."""
    cursor = connection.execute(
        """
        INSERT INTO source_chunks (
            source_id, chunk_kind, title, url, publisher, source_type, location,
            paragraph_index, topics, summary, text, keywords
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            chunk.source_id,
            chunk.chunk_kind,
            chunk.title,
            chunk.url,
            chunk.publisher,
            chunk.source_type,
            chunk.location,
            chunk.paragraph_index,
            json.dumps(chunk.topics, ensure_ascii=False),
            chunk.summary,
            chunk.text,
            json.dumps(chunk.keywords, ensure_ascii=False),
        ),
    )
    rowid = cursor.lastrowid
    connection.execute(
        """
        INSERT INTO source_chunks_fts (
            rowid, source_id, chunk_kind, title, location, topics, summary, text, keywords
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            rowid,
            chunk.source_id,
            chunk.chunk_kind,
            chunk.title,
            chunk.location,
            " ".join(chunk.topics),
            chunk.summary,
            chunk.text,
            " ".join(chunk.keywords),
        ),
    )


def import_nolegraph_embed() -> Any:
    """Import nolegraph.embed from an installed package or the local repo checkout."""
    if str(DEFAULT_NOLEGRAPH_SRC) not in sys.path and DEFAULT_NOLEGRAPH_SRC.exists():
        sys.path.insert(0, str(DEFAULT_NOLEGRAPH_SRC))
    try:
        from nolegraph import embed as nolegraph_embed
    except Exception as exc:
        if os.environ.get("SYNDERESIS_REQUIRE_NOLEGRAPH_EMBED", "").strip().lower() in {"1", "true", "yes", "on"}:
            raise SourceRetrievalError("could not import nolegraph.embed") from exc
        return SimpleNamespace(
            HashEmbedder=HashEmbedder,
            HarrierEmbedder=HarrierEmbedder,
            load_default_embedder=load_default_embedder,
        )
    return nolegraph_embed


def resolve_embedder(mode: str) -> Any | None:
    """Resolve an optional embedder using nolegraph's embedding interfaces."""
    mode = mode.strip().lower()
    if mode in {"", "none", "off"}:
        return None
    nolegraph_embed = import_nolegraph_embed()
    if mode == "hash":
        return nolegraph_embed.HashEmbedder(dim=512)
    if mode == "harrier":
        try:
            return nolegraph_embed.HarrierEmbedder()
        except Exception as exc:
            raise SourceRetrievalError(
                "Harrier embedder failed to load; install sentence-transformers in this Python environment "
                "or run from the nolegraph environment that has it"
            ) from exc
    if mode == "auto":
        return nolegraph_embed.load_default_embedder()
    raise SourceRetrievalError("embedder must be one of: none, hash, harrier, auto")


def content_hash(text: str) -> str:
    """Return a stable content hash."""
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def embedding_text(row: sqlite3.Row) -> str:
    """Return bounded text for the embedding model while preserving full DB text."""
    value = f"{row['title']} {row['source_id']}:{row['location']} {row['chunk_kind']}\n{row['summary']}\n{row['text']}"
    if len(value) <= EMBED_TEXT_MAX_CHARS:
        return value
    return value[:EMBED_TEXT_MAX_CHARS].rsplit(" ", 1)[0].rstrip()


def write_embeddings(connection: sqlite3.Connection, embedder: Any | None) -> None:
    """Encode and persist source chunk vectors when an embedder is configured."""
    if embedder is None:
        return
    rows = connection.execute(
        "SELECT id, source_id, location, chunk_kind, title, summary, text FROM source_chunks ORDER BY id"
    ).fetchall()
    if not rows:
        return
    connection.execute("DELETE FROM source_chunk_embeddings")
    signature = str(embedder.signature)
    for offset in range(0, len(rows), EMBED_BATCH_SIZE):
        batch = rows[offset : offset + EMBED_BATCH_SIZE]
        texts = [embedding_text(row) for row in batch]
        vectors = embedder.encode(texts, is_query=False)
        for row, vector in zip(batch, vectors, strict=True):
            text = f"{row['summary']}\n{row['text']}"
            connection.execute(
                """
                INSERT INTO source_chunk_embeddings(chunk_id, vector, content_hash, embedder_signature)
                VALUES (?, ?, ?, ?)
                """,
                (row["id"], json.dumps(vector), content_hash(text), signature),
            )
    connection.execute(
        "INSERT OR REPLACE INTO meta(key, value) VALUES ('embedder_signature', ?)",
        (signature,),
    )


def build_source_db(
    db_path: Path,
    sources_path: Path = DEFAULT_SOURCES_PATH,
    notes_path: Path = DEFAULT_NOTES_PATH,
    *,
    include_documents: bool = False,
    cache_dir: Path = DEFAULT_DOC_CACHE_DIR,
    force_fetch: bool = False,
    embedder: str = "none",
) -> int:
    """Rebuild the source retrieval SQLite database and return the chunk count."""
    chunks = load_chunks(
        sources_path,
        notes_path,
        include_documents=include_documents,
        cache_dir=cache_dir,
        force_fetch=force_fetch,
    )
    resolved_embedder = resolve_embedder(embedder)
    db_path.parent.mkdir(parents=True, exist_ok=True)
    if db_path.exists():
        db_path.unlink()
    with db_connection(db_path) as connection:
        connection.execute("PRAGMA foreign_keys=ON")
        init_source_db(connection)
        for chunk in chunks:
            insert_chunk(connection, chunk)
        write_embeddings(connection, resolved_embedder)
    return len(chunks)


def query_terms(query: str) -> list[str]:
    """Extract and expand FTS-safe search terms."""
    terms: list[str] = []
    seen: set[str] = set()
    for raw_term in WORD_RE.findall(query.lower()):
        term = raw_term.strip("'")
        if len(term) < 2 or term in STOPWORDS:
            continue
        expanded = SYNONYMS.get(term, [term])
        for candidate in expanded:
            normalized = re.sub(r"[^a-z0-9]", "", candidate.lower())
            if len(normalized) < 2 or normalized in STOPWORDS or normalized in seen:
                continue
            terms.append(normalized)
            seen.add(normalized)
            if len(terms) >= MAX_SEARCH_TERMS:
                return terms
    return terms


def fts_query(terms: list[str]) -> str:
    """Build a safe FTS5 OR query from normalized terms."""
    return " OR ".join(f"{term}*" for term in terms)


def parse_json_list(value: str) -> list[str]:
    """Parse a JSON list from the database."""
    try:
        parsed = json.loads(value)
    except json.JSONDecodeError:
        return []
    if isinstance(parsed, list):
        return [str(item) for item in parsed]
    return []


def vector_dot(left: list[float], right: list[float]) -> float:
    """Dot product for L2-normalized embedding vectors."""
    return sum(a * b for a, b in zip(left, right, strict=False))


def stored_embedder_signature(db_path: Path) -> str | None:
    """Return the embedder signature stored in the source DB, if present."""
    if not db_path.exists():
        return None
    with db_connection(db_path) as connection:
        try:
            row = connection.execute("SELECT value FROM meta WHERE key = 'embedder_signature'").fetchone()
        except sqlite3.OperationalError:
            return None
    return None if row is None else str(row["value"])


def resolve_search_embedder(db_path: Path) -> Any | None:
    """Resolve an embedder for query vectors when the runtime allows it."""
    signature = stored_embedder_signature(db_path)
    if signature is None:
        return None
    requested = os.environ.get("SYNDERESIS_SOURCE_EMBEDDER", "").strip().lower()
    if not requested:
        requested = "auto" if os.environ.get("NOLEGRAPH_AUTOLOAD_EMBEDDER", "").strip().lower() in {"1", "true", "yes", "on"} else "none"
    if requested in {"", "none", "off"}:
        return None
    cache_key = (
        signature,
        requested,
        os.environ.get("NOLEGRAPH_EMBEDDING_MODEL", DEFAULT_EMBEDDING_MODEL),
        os.environ.get("NOLEGRAPH_AUTOLOAD_EMBEDDER", ""),
    )
    cache = getattr(resolve_search_embedder, "_cache", {})
    if cache_key in cache:
        return cache[cache_key]
    try:
        embedder = resolve_embedder(requested)
    except SourceRetrievalError as error:
        print(f"source_retrieval: search embedder failed, falling back to lexical search: {error}", file=sys.stderr)
        cache[cache_key] = None
        setattr(resolve_search_embedder, "_cache", cache)
        return None
    if embedder is None or str(embedder.signature) != signature:
        cache[cache_key] = None
        setattr(resolve_search_embedder, "_cache", cache)
        return None
    cache[cache_key] = embedder
    setattr(resolve_search_embedder, "_cache", cache)
    return embedder


def semantic_search(db_path: Path, query: str, limit: int) -> list[tuple[int, float]]:
    """Return chunk ids and semantic scores using stored embeddings."""
    embedder = resolve_search_embedder(db_path)
    if embedder is None:
        return []
    query_vector = embedder.encode([query], is_query=True)[0]
    with db_connection(db_path) as connection:
        try:
            rows = connection.execute("SELECT chunk_id, vector FROM source_chunk_embeddings").fetchall()
        except sqlite3.OperationalError:
            return []
    scored: list[tuple[int, float]] = []
    for row in rows:
        try:
            vector = [float(value) for value in json.loads(row["vector"])]
        except (TypeError, ValueError, json.JSONDecodeError):
            continue
        if len(vector) != len(query_vector):
            continue
        scored.append((int(row["chunk_id"]), vector_dot(query_vector, vector)))
    scored.sort(key=lambda item: item[1], reverse=True)
    return scored[: max(limit * 4, 24)]


def token_set(*values: str) -> set[str]:
    """Return lowercase tokens from one or more strings."""
    return {token.strip("'") for value in values for token in WORD_RE.findall(value.lower())}


def overlap_score(terms: list[str], row: sqlite3.Row) -> float:
    """Compute a small rerank score to prefer semantically dense matches."""
    topics = " ".join(parse_json_list(row["topics"]))
    keywords = " ".join(parse_json_list(row["keywords"]))
    haystack = token_set(row["source_id"], row["title"], row["location"], topics, row["summary"], keywords)
    keyword_tokens = token_set(keywords)
    score = 0.0
    for term in terms:
        matched = any(token.startswith(term) or term.startswith(token) for token in haystack)
        if matched:
            score += 1.0
        keyword_matched = any(token.startswith(term) or term.startswith(token) for token in keyword_tokens)
        if keyword_matched:
            score += 0.75
    return score


def normalize_search_filters(filters: dict[str, Any] | None) -> dict[str, set[str]]:
    """Normalize optional API search filters."""
    if not filters:
        return {}
    normalized: dict[str, set[str]] = {}
    for key in ("source_ids", "source_types", "publishers", "chunk_kinds", "topics"):
        raw_values = filters.get(key, [])
        if isinstance(raw_values, str):
            raw_values = [raw_values]
        if not isinstance(raw_values, list):
            continue
        values = {str(value).strip().casefold() for value in raw_values if str(value).strip()}
        if values:
            normalized[key] = values
    return normalized


def row_matches_search_filters(row: sqlite3.Row, filters: dict[str, set[str]]) -> bool:
    """Return whether a source row satisfies normalized filters."""
    if not filters:
        return True
    if "source_ids" in filters and str(row["source_id"]).casefold() not in filters["source_ids"]:
        return False
    if "source_types" in filters and str(row["source_type"]).casefold() not in filters["source_types"]:
        return False
    if "publishers" in filters and str(row["publisher"]).casefold() not in filters["publishers"]:
        return False
    if "chunk_kinds" in filters and str(row["chunk_kind"]).casefold() not in filters["chunk_kinds"]:
        return False
    if "topics" in filters:
        row_topics = {topic.casefold() for topic in parse_json_list(row["topics"])}
        if row_topics.isdisjoint(filters["topics"]):
            return False
    return True


def result_from_row(row: sqlite3.Row, score: float) -> SearchResult:
    """Build a SearchResult from a source_chunks row."""
    return SearchResult(
        source_id=row["source_id"],
        chunk_kind=row["chunk_kind"],
        title=row["title"],
        url=row["url"],
        publisher=row["publisher"],
        source_type=row["source_type"],
        location=row["location"],
        paragraph_index=int(row["paragraph_index"]),
        topics=parse_json_list(row["topics"]),
        summary=row["summary"],
        text=row["text"],
        keywords=parse_json_list(row["keywords"]),
        score=round(score, 6),
    )


def search_source_chunks(db_path: Path, query: str, limit: int = 6, filters: dict[str, Any] | None = None) -> list[SearchResult]:
    """Search source chunks by query content."""
    if limit <= 0:
        return []
    if not db_path.exists():
        raise SourceRetrievalError(f"source retrieval database not found: {db_path}")
    terms = query_terms(query)
    if not terms:
        return []
    expression = fts_query(terms)
    normalized_filters = normalize_search_filters(filters)
    candidate_limit = max(limit * 40, 200) if normalized_filters else max(limit * 8, 24)
    semantic_scores = dict(semantic_search(db_path, query, candidate_limit if normalized_filters else limit))
    with db_connection(db_path) as connection:
        rows = connection.execute(
            """
            SELECT
                c.id,
                c.source_id,
                c.chunk_kind,
                c.title,
                c.url,
                c.publisher,
                c.source_type,
                c.location,
                c.paragraph_index,
                c.topics,
                c.summary,
                c.text,
                c.keywords,
                bm25(source_chunks_fts) AS bm25_rank
            FROM source_chunks_fts
            JOIN source_chunks c ON c.id = source_chunks_fts.rowid
            WHERE source_chunks_fts MATCH ?
            ORDER BY bm25_rank
            LIMIT ?
            """,
            (expression, candidate_limit),
        ).fetchall()
        semantic_rows: list[sqlite3.Row] = []
        if semantic_scores:
            placeholders = ",".join("?" for _ in semantic_scores)
            semantic_rows = connection.execute(
                f"""
                SELECT
                    id, source_id, chunk_kind, title, url, publisher, source_type,
                    location, paragraph_index, topics, summary, text, keywords
                FROM source_chunks
                WHERE id IN ({placeholders})
                """,
                tuple(semantic_scores),
            ).fetchall()
    results: list[SearchResult] = []
    seen_ids: set[int] = set()
    for row in rows:
        if not row_matches_search_filters(row, normalized_filters):
            continue
        chunk_id = int(row["id"])
        seen_ids.add(chunk_id)
        semantic_score = overlap_score(terms, row)
        bm25_rank = float(row["bm25_rank"])
        score = semantic_score + max(0.0, -bm25_rank) + semantic_scores.get(chunk_id, 0.0) * 20.0
        results.append(result_from_row(row, score))
    for row in semantic_rows:
        if not row_matches_search_filters(row, normalized_filters):
            continue
        chunk_id = int(row["id"])
        if chunk_id in seen_ids:
            continue
        score = semantic_scores.get(chunk_id, 0.0) * 20.0
        results.append(result_from_row(row, score))
    results.sort(key=lambda item: (-item.score, item.source_id, item.location))
    max_per_source = max(3, limit // 2)
    selected: list[SearchResult] = []
    selected_refs: set[tuple[str, str]] = set()
    source_counts: dict[str, int] = {}
    overflow: list[SearchResult] = []
    for result in results:
        ref_key = (result.source_id, result.location)
        if ref_key in selected_refs:
            continue
        if source_counts.get(result.source_id, 0) >= max_per_source:
            overflow.append(result)
            continue
        selected.append(result)
        selected_refs.add(ref_key)
        source_counts[result.source_id] = source_counts.get(result.source_id, 0) + 1
        if len(selected) >= limit:
            return selected
    for result in overflow:
        ref_key = (result.source_id, result.location)
        if ref_key in selected_refs:
            continue
        selected.append(result)
        selected_refs.add(ref_key)
        if len(selected) >= limit:
            break
    return selected


def parse_args() -> argparse.Namespace:
    """Parse CLI arguments."""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--db-path", default=str(DEFAULT_DB_PATH))
    parser.add_argument("--sources-path", default=str(DEFAULT_SOURCES_PATH))
    parser.add_argument("--notes-path", default=str(DEFAULT_NOTES_PATH))
    parser.add_argument("--cache-dir", default=str(DEFAULT_DOC_CACHE_DIR))
    parser.add_argument("--build", action="store_true", help="Build or rebuild the SQLite retrieval database")
    parser.add_argument("--include-documents", action="store_true", help="Fetch official source pages and add paragraph-level chunks")
    parser.add_argument("--force-fetch", action="store_true", help="Refresh cached HTML documents")
    parser.add_argument("--embedder", choices=["none", "hash", "harrier", "auto"], default="harrier", help="Optional vector embedder for chunk embeddings")
    parser.add_argument("--search", default="", help="Search query to run after opening the database")
    parser.add_argument("--limit", type=int, default=6)
    parser.add_argument("--json", action="store_true", help="Print search results as JSON")
    return parser.parse_args()


def main() -> int:
    """Build or search the source retrieval database."""
    args = parse_args()
    db_path = Path(args.db_path)
    if args.build:
        count = build_source_db(
            db_path,
            Path(args.sources_path),
            Path(args.notes_path),
            include_documents=args.include_documents,
            cache_dir=Path(args.cache_dir),
            force_fetch=args.force_fetch,
            embedder=args.embedder,
        )
        print(json.dumps({"db_path": str(db_path), "chunk_count": count}, indent=2))
    if args.search:
        results = search_source_chunks(db_path, args.search, args.limit)
        if args.json:
            print(json.dumps([result.to_dict() for result in results], indent=2, ensure_ascii=False))
        else:
            for result in results:
                print(f"{result.source_id}:{result.location}\t{result.score:.3f}\t{result.title}\t{result.summary}")
    if not args.build and not args.search:
        raise SystemExit("provide --build and/or --search")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())