rag_module.py

import json
import logging
import math
import os
import re
import sqlite3
import sys
import urllib.parse
import urllib.request
from collections import Counter

from tracernet.config import apply_env_defaults

apply_env_defaults()
try:
    import posthog
    posthog.capture = lambda *args, **kwargs: None
except Exception:
    pass

try:
    from langchain_chroma import Chroma
    from langchain_huggingface import HuggingFaceEmbeddings
    LANGCHAIN_AVAILABLE = True
except Exception:
    Chroma = None
    HuggingFaceEmbeddings = None
    LANGCHAIN_AVAILABLE = False

from rag_query_generator import TracePathQueryGenerator
from chemical_formatter import (
    display_condition_label as _display_condition_casing,
    format_chemical_formula,
    plain_formula,
)
from tracernet.crn.repository import CRNRepository, clean_cell
from tracernet.services.pathway_evidence import (
    build_default_matcher,
    canonical_formula,
    formula_phase,
    normalize_text,
    parse_formula,
    parse_path,
    path_conditions,
    path_species,
)
from tracernet.services.bibliography import (
    BibliographicRecord,
    BibliographyRegistry,
    extract_dois,
    first_doi,
    format_gbt7714,
    normalize_doi,
)
try:
    from tracernet.services.doi_metadata import resolve_crossref_metadata
except ImportError:
    # Keep the RAG service operational when an older deployment omits the
    # optional helper module. This fallback deliberately uses the same public
    # Crossref API and returns only normalized bibliographic metadata.
    def resolve_crossref_metadata(doi, timeout=8.0):
        normalized = normalize_doi(doi)
        if not normalized:
            return None
        url = "https://api.crossref.org/works/" + urllib.parse.quote(
            normalized, safe=""
        )
        request = urllib.request.Request(
            url,
            headers={
                "Accept": "application/json",
                "User-Agent": "TRACER-Net/1.0 (bibliographic verification)",
            },
        )
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                message = json.loads(response.read().decode("utf-8")).get(
                    "message", {}
                )
        except Exception as exc:
            logger = logging.getLogger(__name__)
            logger.warning("Crossref DOI lookup failed for %s: %s", normalized, exc)
            return None

        def first(value):
            if isinstance(value, list):
                return str(value[0]).strip() if value else ""
            return str(value or "").strip()

        date_parts = (
            message.get("published-print", {}).get("date-parts")
            or message.get("published-online", {}).get("date-parts")
            or message.get("issued", {}).get("date-parts")
            or []
        )
        year = date_parts[0][0] if date_parts and date_parts[0] else None
        authors = []
        for author in message.get("author") or []:
            name = " ".join(
                part for part in (author.get("given", ""), author.get("family", ""))
                if part
            ).strip()
            if name:
                authors.append(name)
        return {
            "title": first(message.get("title")),
            "authors": authors,
            "journal": first(message.get("container-title")),
            "year": year,
            "volume": first(message.get("volume")),
            "issue": first(message.get("issue")),
            "pages": first(message.get("page")),
            "article_number": first(message.get("article-number")),
            "doi": normalize_doi(message.get("DOI") or normalized),
            "url": first(message.get("URL")),
        }
from tracernet.services.report_text import (
    compose_graph_only_note,
    is_bad_generation_text,
)
from tracernet.utils.assets import ensure_gzip_asset, find_asset

try:
    from rag_evaluation import evaluate_rag_report
except Exception:
    evaluate_rag_report = None

logger = logging.getLogger(__name__)


def _resolve_db_path():
    override = os.getenv("RAG_DB_PATH", "").strip()
    if override:
        candidate = os.path.abspath(os.path.expanduser(override))
        if os.path.isfile(candidate):
            return os.path.dirname(candidate)
        if os.path.isfile(os.path.join(candidate, "chroma.sqlite3")):
            return candidate
    search_roots = (
        os.path.dirname(os.path.abspath(__file__)),
        os.getcwd(),
    )
    database = find_asset("chroma.sqlite3", search_roots, max_depth=2)
    compressed = find_asset("chroma.sqlite3.gz", search_roots, max_depth=2)
    if compressed is not None:
        extracted = ensure_gzip_asset(compressed, compressed.with_suffix(""))
        if extracted is not None:
            database = extracted
    return str(database.parent) if database is not None else ""


def _patch_chroma_schema(db_dir: str) -> None:
    if not db_dir:
        return
    db_path = db_dir
    if os.path.isdir(db_dir):
        db_path = os.path.join(db_dir, "chroma.sqlite3")
    if not os.path.exists(db_path):
        return
    try:
        # Ensure the file is writable before attempting UPDATE.
        # Common failure: extracted from .gz with read-only permissions.
        if not os.access(db_path, os.W_OK):
            try:
                import stat
                current = os.stat(db_path).st_mode
                os.chmod(db_path, current | stat.S_IWUSR | stat.S_IWGRP)
            except Exception:
                return
        con = sqlite3.connect(db_path)
        default_config = {
            "_type": "CollectionConfigurationInternal",
            "hnsw_configuration": {
                "_type": "HNSWConfigurationInternal",
                "space": "l2",
                "ef_construction": 100,
                "ef_search": 100,
                "num_threads": os.cpu_count() or 4,
                "M": 16,
                "resize_factor": 1.2,
                "batch_size": 100,
                "sync_threshold": 1000,
            },
        }
        cur = con.execute("SELECT id, schema_str, config_json_str FROM collections")
        rows = cur.fetchall()
        changed = False
        for cid, schema_str, config_str in rows:
            if not schema_str:
                schema = None
            else:
                try:
                    schema = json.loads(schema_str)
                except Exception:
                    schema = None
            if schema is not None:
                updated = False
                def _fix(obj):
                    nonlocal updated
                    if isinstance(obj, dict):
                        if "embedding_function" in obj and isinstance(obj["embedding_function"], dict):
                            ef = obj["embedding_function"]
                            if "_type" not in ef and "type" in ef:
                                ef["_type"] = ef["type"]
                                updated = True
                        for v in obj.values():
                            _fix(v)
                    elif isinstance(obj, list):
                        for v in obj:
                            _fix(v)
                _fix(schema)
                if updated:
                    con.execute(
                        "UPDATE collections SET schema_str = ? WHERE id = ?",
                        (json.dumps(schema), cid),
                    )
                    changed = True
            config_obj = None
            if config_str:
                try:
                    config_obj = json.loads(config_str)
                except Exception:
                    config_obj = None
            if not config_obj or "_type" not in config_obj:
                con.execute(
                    "UPDATE collections SET config_json_str = ? WHERE id = ?",
                    (json.dumps(default_config), cid),
                )
                changed = True
        if changed:
            con.commit()
        con.close()
    except Exception:
        return

DB_PATH = _resolve_db_path()
_default_embed = "BAAI/bge-m3"
EMBEDDING_MODEL = os.getenv("RAG_EMBEDDING_MODEL", _default_embed)
RAG_COLLECTION = os.getenv("RAG_COLLECTION")
def _load_key_from_file() -> str:
    """Fallback: read the DeepSeek key from a local file when the env var is
    unset. Looks next to this module and in the current working directory so
    the key can simply live in ``DEEPSEEK_API_KEY.txt`` without exporting it."""
    candidates = [
        os.path.join(os.path.dirname(os.path.abspath(__file__)), "DEEPSEEK_API_KEY.txt"),
        os.path.join(os.getcwd(), "DEEPSEEK_API_KEY.txt"),
    ]
    for path in candidates:
        try:
            if os.path.isfile(path):
                with open(path, "r", encoding="utf-8") as fh:
                    value = fh.read().strip()
                if value:
                    return value
        except OSError:
            continue
    return ""

_raw_deepseek_key = os.getenv("DEEPSEEK_API_KEY", "").strip() or _load_key_from_file()
_raw_deepseek_base_url = os.getenv("DEEPSEEK_BASE_URL", "").strip()
_raw_deepseek_model = os.getenv("DEEPSEEK_MODEL", "").strip()

DEEPSEEK_API_KEY = _raw_deepseek_key
DEEPSEEK_BASE_URL = _raw_deepseek_base_url or "https://api.deepseek.com"
DEEPSEEK_MODEL = _raw_deepseek_model or "deepseek-chat"

# Boot-time confirmation of LLM configuration (never logs the key value).
logger.info(
    "[RAG][LLM] config model=%s base=%s key_set=%s",
    DEEPSEEK_MODEL,
    DEEPSEEK_BASE_URL,
    "yes" if DEEPSEEK_API_KEY else "no",
)

# Recent transformers refuses to call torch.load on a .bin checkpoint unless
# torch >= 2.6 (CVE-2025-32434); the pinned torch is 2.2.2. Patching only
# transformers.utils.import_utils is NOT enough: transformers.modeling_utils
# does `from .utils.import_utils import check_torch_load_is_safe` at import
# time and therefore holds its own reference. Both must be replaced or the
# embedding model fails to load and the whole RAG service is unavailable.
def _allow_local_torch_load(*_args, **_kwargs):
    return None


try:
    import transformers.utils.import_utils

    # Patching import_utils is enough *provided it happens before
    # transformers.modeling_utils is first imported*: modeling_utils binds this
    # name at its own import time and so picks up the patched function.
    transformers.utils.import_utils.check_torch_load_is_safe = _allow_local_torch_load
    # Safety net for the reverse order (modeling_utils already loaded, e.g. in
    # an offline build script). Never import it here — that pulls in the whole
    # generation stack and stalls application startup.
    _loaded_modeling_utils = sys.modules.get("transformers.modeling_utils")
    if _loaded_modeling_utils is not None and hasattr(
        _loaded_modeling_utils, "check_torch_load_is_safe"
    ):
        _loaded_modeling_utils.check_torch_load_is_safe = _allow_local_torch_load
except ImportError:
    pass


# A reference-list entry reads as evidence to the matcher: "Douglass DL, Shing
# CC, Wang G (1992) The light-induced alteration of realgar to pararealgar"
# names both species and a conversion, so it was being classified "direct" and
# printed as a supporting passage. It states nothing -- it is a pointer to a
# paper that is not in the corpus -- and two such entries were inflating one
# edge's evidence count from five real passages to seven.
_BIBLIO_AUTHOR_YEAR_RE = re.compile(
    r"[A-Z][A-Za-z'’-]+\s+[A-Z]{1,3}[,;]?\s*(?:\(\d{4}[a-z]?\)|\d{4}[a-z]?\b)"
)
_BIBLIO_LOCATOR_RE = re.compile(
    r"\b(?:doi:|https?://|\d+\s*:\s*\d+\s*[-–]\s*\d+|\bpp?\.\s*\d+)", re.IGNORECASE
)
# "[33] Douglass DL" -- a citation marker in front of a name with initials.
# In prose the marker trails the claim it supports; only a reference list puts
# one before an author.
_BIBLIO_MARKER_AUTHOR_RE = re.compile(
    r"\[\d{1,3}\]\s*[A-Z][A-Za-z'’-]{2,}\s+[A-Z]{1,3}\b"
)


def _is_bibliography_text(text: object) -> bool:
    """True when the passage is a reference-list entry rather than prose.

    The decisive shape is a citation marker followed immediately by an author
    name and initials -- "[33] Douglass DL, Shing CC". Prose also carries [n],
    but at the end of the sentence it supports, never in front of a name, so
    this separates the two without needing a locator: the entries seen in this
    corpus often carry no DOI or page range at all.

    A verb test cannot stand in for it. "Can Mineral 18:525-527 [33] Douglass
    DL ..." opens with a journal abbreviation that any modal-verb pattern reads
    as the word "can".
    """
    s = re.sub(r"\s+", " ", str(text or "")).strip()
    if not s:
        return False
    if _BIBLIO_MARKER_AUTHOR_RE.search(s):
        return True
    # A full entry without a bracketed marker: several author-initial groups
    # and a locator, and short enough to be one entry rather than a paragraph
    # that happens to cite several works.
    authors = len(_BIBLIO_AUTHOR_YEAR_RE.findall(s))
    if authors >= 2 and _BIBLIO_LOCATOR_RE.search(s) and len(s) < 400:
        return True
    return authors >= 1 and _BIBLIO_LOCATOR_RE.search(s) is not None and len(s) < 220


def _compose_direct_evidence_intro(match, ref_index):
    """Canonical evidence-anchored introduction sentence.

    A single generic template shared by every direct-evidence fallback so the
    Introduction reads identically regardless of which code path produced it
    (generation fallback or evidence-guard synthesis). Driven only by the
    matched reactant/product/condition and the pathway-vs-edge scope; contains
    no pigment- or example-specific wording.
    """
    match = match if isinstance(match, dict) else {}

    def _label(name, formula):
        name = str(name or "").strip()
        formula = str(formula or "").strip()
        if formula and formula.casefold() != name.casefold():
            return f"{name} ({formula})"
        return name or formula

    reactant_label = _label(
        match.get("reactant_span") or match.get("reactant") or "the source material",
        match.get("reactant"),
    )
    product_label = _label(
        match.get("product_span") or match.get("product") or "the product",
        match.get("product"),
    )
    condition = re.sub(
        r"\s*\+\s*",
        " + ",
        _display_condition_casing(str(match.get("condition") or "").strip()),
    )
    scope_text = (
        "overall conversion"
        if match.get("evidence_scope") == "pathway_endpoint"
        else "conversion"
    )
    if condition:
        return (
            f"Under {condition} conditions, {reactant_label} is documented to "
            f"undergo the {scope_text} to {product_label}, as directly supported "
            f"by the retrieved literature [{ref_index}]."
        )
    lead = reactant_label[:1].upper() + reactant_label[1:]
    return (
        f"{lead} is documented to undergo the {scope_text} to {product_label}, "
        f"as directly supported by the retrieved literature [{ref_index}]."
    )


# How strongly a passage supports a pathway step, weakest to strongest. Used
# both to pick the best of several matches and to describe, in the report, how
# far a source got when it did not reach "direct".
VERDICT_PRIORITY = {
    "direct": 8,
    "qualified": 7,
    "inferred": 6,
    "related": 5,
    "condition_mismatch": 4,
    "contradicted": 3,
    "phase_conflict": 2,
    "unsupported": 1,
}


class _DenseDoc:
    """Minimal stand-in for a langchain Document.

    The dense route can now query the Chroma collection directly with a
    pre-batched query vector (see _embed_queries_batch), which returns plain
    strings/dicts rather than Document objects. This wrapper gives that raw
    result the ``.page_content`` / ``.metadata`` shape the downstream code
    already expects, so nothing after retrieval had to change.
    """

    __slots__ = ("page_content", "metadata")

    def __init__(self, page_content, metadata):
        self.page_content = page_content
        self.metadata = metadata


class RAGService:
    def __init__(self, crn_repository: CRNRepository | None = None):
        self.llm_available = bool(DEEPSEEK_API_KEY)
        self.llm_error = (
            ""
            if self.llm_available
            else "DeepSeek is not configured. Please set DEEPSEEK_API_KEY."
        )
        logging.info("[RAG 1/4] Loading CRN and bibliography metadata")
        self.query_generator = TracePathQueryGenerator()
        self.pathway_reference_records = []
        self.crn_repository = (
            crn_repository
            if crn_repository is not None
            else CRNRepository.discover(
                (
                    os.path.dirname(os.path.abspath(__file__)),
                    os.getcwd(),
                ),
                max_depth=1,
            )
        )
        self.bibliography = BibliographyRegistry.load(
            os.path.dirname(os.path.abspath(__file__))
        )
        self.doi_metadata_resolver = resolve_crossref_metadata
        self._doi_metadata_cache = {}
        self.evidence_matcher = build_default_matcher(
            os.path.dirname(os.path.abspath(__file__))
        )

        self.vector_db = None
        self.chroma_collection = None
        self.embedder = None
        self.collection_count = None
        self.retrieval_mode = "none"
        self.retrieval_error = ""
        self.last_query_error = ""
        db_dir = DB_PATH
        if not db_dir:
            self.retrieval_error = "Literature vector database was not found."
            self._load_pathway_references()
            return
        if os.path.isfile(db_dir):
            db_dir = os.path.dirname(db_dir)

        # The on-disk Chroma format is version sensitive: an index written by a
        # newer chromadb than the runtime can fail to load its collection config
        # ("'dict' object has no attribute 'dimensionality'"). Log the versions
        # so a rebuilt index can be produced with a matching one.
        try:
            import chromadb as _chromadb

            _chroma_version = getattr(_chromadb, "__version__", "?")
        except Exception:
            _chroma_version = "unavailable"
        try:
            from importlib.metadata import version as _pkg_version

            _lc_chroma_version = _pkg_version("langchain-chroma")
        except Exception:
            _lc_chroma_version = "?"
        logging.info(
            "[RAG][CHROMA] chromadb=%s langchain-chroma=%s",
            _chroma_version,
            _lc_chroma_version,
        )
        logging.info("[RAG 2/4] Opening Chroma database at %s", db_dir)
        available_collections = []
        if os.path.exists(os.path.join(db_dir, "chroma.sqlite3")):
            _patch_chroma_schema(db_dir)
            try:
                import chromadb
                client_probe = chromadb.PersistentClient(path=db_dir)
                available_collections = [c.name for c in client_probe.list_collections()]
            except Exception:
                available_collections = []
            try:
                if not LANGCHAIN_AVAILABLE:
                    raise RuntimeError("langchain_chroma unavailable")
                collection_name = RAG_COLLECTION
                if not collection_name:
                    if "literature" in available_collections:
                        collection_name = "literature"
                    elif available_collections:
                        collection_name = available_collections[0]
                    else:
                        collection_name = "langchain"
                os.environ["RAG_COLLECTION"] = collection_name
                logging.info(
                    "[RAG 3/4] Loading embedding model %s on CPU",
                    EMBEDDING_MODEL,
                )
                embeddings = HuggingFaceEmbeddings(
                    model_name=EMBEDDING_MODEL,
                    model_kwargs={"device": "cpu"},
                    encode_kwargs={"normalize_embeddings": True},
                )
                self.vector_db = Chroma(
                    persist_directory=db_dir,
                    embedding_function=embeddings,
                    collection_name=collection_name,
                )
                try:
                    self.collection_count = self.vector_db._collection.count() if self.vector_db else None
                except Exception:
                    self.collection_count = None
                if self.collection_count == 0:
                    try:
                        client = getattr(self.vector_db, "_client", None)
                        if client and hasattr(client, "list_collections"):
                            for col in client.list_collections():
                                try:
                                    if col.count() > 0 and col.name != self.vector_db._collection.name:
                                        self.vector_db = Chroma(
                                            persist_directory=db_dir,
                                            embedding_function=embeddings,
                                            collection_name=col.name,
                                        )
                                        self.collection_count = self.vector_db._collection.count()
                                        os.environ["RAG_COLLECTION"] = col.name
                                        break
                                except Exception:
                                    continue
                    except Exception:
                        pass
                self.retrieval_mode = "langchain" if self.vector_db else "none"
                if self.retrieval_mode != "none":
                    self.retrieval_error = ""
                logging.info(
                    "[RAG 4/4] LangChain vector store ready: collection=%s, documents=%s",
                    collection_name,
                    self.collection_count,
                )
            except Exception as exc:
                logging.exception(
                    "LangChain RAG initialization failed; trying direct Chroma fallback: %s",
                    exc,
                )
                self.vector_db = None
                self.collection_count = None
                try:
                    import chromadb
                    from chromadb.utils import embedding_functions
                    client = chromadb.PersistentClient(path=db_dir)
                    collection_name = os.getenv("RAG_COLLECTION")
                    if not collection_name:
                        if "literature" in available_collections:
                            collection_name = "literature"
                        elif available_collections:
                            collection_name = available_collections[0]
                        else:
                            collection_name = "langchain"
                    try:
                        self.chroma_collection = client.get_collection(name=collection_name)
                    except Exception:
                        cols = client.list_collections()
                        self.chroma_collection = client.get_collection(name=cols[0].name) if cols else None
                    self.collection_count = (
                        self.chroma_collection.count()
                        if self.chroma_collection
                        else None
                    )
                    logging.info(
                        "[RAG 3/4] Loading fallback embedding model %s",
                        EMBEDDING_MODEL,
                    )
                    try:
                        self.embedder = (
                            embedding_functions.SentenceTransformerEmbeddingFunction(
                                model_name=EMBEDDING_MODEL
                            )
                        )
                        self.retrieval_mode = (
                            "chroma" if self.chroma_collection else "none"
                        )
                    except Exception as embedding_exc:
                        # A readable Chroma collection is still useful without
                        # the query embedding model: CRN DOI lookups and the
                        # strict entity/direction/condition scan below operate
                        # directly on stored documents. Keep that deterministic
                        # evidence path available instead of disabling RAG
                        # completely.
                        self.embedder = None
                        self.retrieval_mode = (
                            "chroma_lexical"
                            if self.chroma_collection
                            else "none"
                        )
                        self.retrieval_error = (
                            "Dense embedding unavailable; using CRN provenance "
                            "and strict collection scan only. "
                            f"{type(embedding_exc).__name__}: "
                            f"{str(embedding_exc)[:180]}"
                        )
                        logging.warning(
                            "[RAG 4/4] Dense embedding unavailable; direct "
                            "collection evidence fallback enabled: %s",
                            embedding_exc,
                        )
                    if self.retrieval_mode != "none":
                        if self.retrieval_mode != "chroma_lexical":
                            self.retrieval_error = ""
                    logging.info(
                        "[RAG 4/4] Direct Chroma store ready: mode=%s, "
                        "collection=%s, documents=%s",
                        self.retrieval_mode,
                        collection_name,
                        self.collection_count,
                    )
                except Exception as fallback_exc:
                    logging.exception(
                        "Direct Chroma fallback initialization failed: %s",
                        fallback_exc,
                    )
                    self.chroma_collection = None
                    self.embedder = None
                    self.collection_count = None
                    self.retrieval_error = (
                        f"{type(fallback_exc).__name__}: "
                        f"{str(fallback_exc)[:240]}"
                    )
        if self.retrieval_mode == "none" and not self.retrieval_error:
            self.retrieval_error = "No readable literature vector collection was initialized."
        self._load_pathway_references()

    def _smart_truncate(self, text, max_chars=1500):
        if len(text) <= max_chars:
            return text
        return text[:max_chars] + "\n...[Refs truncated]..."

    def _dedupe_results(self, results):
        if not results:
            return []
        seen = {}
        for item in results:
            key = (item.get("content", "").strip(), item.get("source", "").strip())
            score = item.get("score")
            if key not in seen:
                seen[key] = item
                continue
            prev = seen[key]
            if score is not None and (prev.get("score") is None or score > prev.get("score")):
                seen[key] = item
        return list(seen.values())

    def _compact_query(self, query: str, max_len: int = 180) -> str:
        text = re.sub(r"\s+", " ", str(query or "")).strip()
        if len(text) > max_len:
            return text[:max_len - 3] + "..."
        return text

    def _log_retrieval(self, query: str, mode: str, scores: list, kept: int):
        q = self._compact_query(query)
        top_score = max(scores) if scores else None
        top_str = f"{top_score:.4f}" if isinstance(top_score, (int, float)) else "n/a"
        logger.info("[RAG][RETRIEVE] query=%s", q)
        logger.info("[RAG][RETRIEVE] mode=%s top_score=%s kept=%d", mode, top_str, kept)

    def _embed_query_text(self, text: str):
        if not self.embedder:
            return None
        if hasattr(self.embedder, "embed_query"):
            return self.embedder.embed_query(text)
        if callable(self.embedder):
            try:
                return self.embedder([text])[0]
            except Exception:
                return None
        return None

    def _embed_queries_batch(self, queries):
        """Embed every path query in one batched forward pass.

        Returns ``{query_text: vector}``. On CPU, encoding the ~20 query
        variants of a path together is several times faster than embedding them
        one at a time inside the retrieval loop, which is the dominant cost of
        dense retrieval. The vectors are identical to per-query embedding, so
        retrieval results are unchanged -- only faster. Returns ``{}`` on any
        failure so callers transparently fall back to per-query embedding.
        """
        texts = [
            text for text in dict.fromkeys(queries)
            if str(text or "").strip()
        ]
        if not texts:
            return {}
        embed_documents = None
        langchain_embed = getattr(self.vector_db, "_embedding_function", None)
        if langchain_embed is not None and hasattr(
            langchain_embed, "embed_documents"
        ):
            embed_documents = langchain_embed.embed_documents
        elif hasattr(self.embedder, "embed_documents"):
            embed_documents = self.embedder.embed_documents
        try:
            if embed_documents is not None:
                vectors = embed_documents(texts)
            elif callable(self.embedder):
                vectors = self.embedder(texts)
            else:
                return {}
        except Exception:
            logger.exception(
                "[RAG][RETRIEVE] batch query embedding failed; falling back "
                "to per-query embedding"
            )
            return {}
        if vectors is None or len(vectors) != len(texts):
            return {}
        return {text: vector for text, vector in zip(texts, vectors)}

    def _load_pathway_references(self):
        self.pathway_reference_records = [
            {
                "reactant": row.reactant,
                "condition": row.condition,
                "product": row.final_product,
                "source": row.source,
                "doi": first_doi(row.source),
            }
            for row in self.crn_repository.rows
            if row.reactant and row.source
        ]

    def _formula_key(self, value):
        text = normalize_text(value).strip()
        if not text:
            return ""
        formula = canonical_formula(text)
        if formula:
            return formula.casefold()
        return re.sub(r"[^\w]+", "", text, flags=re.UNICODE).casefold()

    def _text_key(self, value):
        return re.sub(
            r"\s+",
            " ",
            normalize_text(value).replace("_", " ").strip(),
        ).casefold()

    def _rank_pathway_reference_records(self, root_material, path_str):
        species = path_species(path_str)
        conditions = path_conditions(path_str)
        species_keys = {self._formula_key(item) for item in ([root_material] + species) if item}
        condition_keys = {self._text_key(item) for item in conditions if item}
        step_pairs = set()
        for idx in range(max(0, len(species) - 1)):
            step_pairs.add((self._formula_key(species[idx]), self._formula_key(species[idx + 1])))
        path_start_key = self._formula_key(species[0]) if species else self._formula_key(root_material)
        path_end_key = self._formula_key(species[-1]) if species else ""

        ranked = []
        for record in self.pathway_reference_records:
            source = record.get("source") or ""
            if not source:
                continue
            reactant_key = self._formula_key(record.get("reactant"))
            product_key = self._formula_key(record.get("product"))
            condition_key = self._text_key(record.get("condition"))
            score = 0
            if reactant_key and product_key and reactant_key == path_start_key and product_key == path_end_key:
                score += 8
            if (reactant_key, product_key) in step_pairs:
                score += 5
            if reactant_key and reactant_key in species_keys:
                score += 2
            if product_key and product_key in species_keys:
                score += 2
            if condition_key and condition_key in condition_keys:
                score += 2
            if score >= 3:
                ranked.append((score, record))
        ranked.sort(key=lambda item: item[0], reverse=True)
        return ranked

    def _get_pathway_provenance_hints(self, root_material, path_str):
        ranked = self._rank_pathway_reference_records(root_material, path_str)
        if not ranked:
            return [], []

        top_score = ranked[0][0]
        sources = []
        dois = []
        for score, record in ranked:
            if score < top_score:
                break
            source = clean_cell(record.get("source"))
            if source and source not in sources:
                sources.append(source)
            doi = record.get("doi") or first_doi(source)
            if doi and doi not in dois:
                dois.append(doi)
        return sources[:5], dois[:5]


    def answer(self, prompt: str, max_tokens: int = 1200) -> str:
        try:
            content = self._chat_completion(
                messages=[
                    {"role": "system", "content": "You are a professional heritage conservation expert."},
                    {"role": "user", "content": prompt},
                ],
                temperature=0.3,
                max_tokens=max_tokens,
            )
            return content or ""
        except Exception as e:
            logger.warning("Language-model answer failed: %s", e.__class__.__name__)
            return "Model service is temporarily unavailable. Please try again later."

    def _chat_completion(self, messages, temperature=0.3, max_tokens=1200, purpose="llm"):
        url = DEEPSEEK_BASE_URL.rstrip("/") + "/chat/completions"
        payload = {
            "model": DEEPSEEK_MODEL,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
        }
        headers = {
            "Content-Type": "application/json",
        }
        if DEEPSEEK_API_KEY:
            headers["Authorization"] = f"Bearer {DEEPSEEK_API_KEY}"
        # Diagnostic: shows in the backend whether DeepSeek is actually invoked.
        # Never logs the key value, only whether one is configured.
        logger.info(
            "[RAG][LLM] call purpose=%s model=%s base=%s key_set=%s",
            purpose,
            DEEPSEEK_MODEL,
            DEEPSEEK_BASE_URL,
            "yes" if DEEPSEEK_API_KEY else "no",
        )
        req = urllib.request.Request(
            url,
            data=json.dumps(payload).encode("utf-8"),
            headers=headers,
            method="POST",
        )
        try:
            with urllib.request.urlopen(req, timeout=60) as resp:
                status = getattr(resp, "status", None) or resp.getcode()
                data = json.loads(resp.read().decode("utf-8"))
        except Exception as exc:
            # Surface HTTP body for auth/quota errors when available.
            detail = ""
            body = getattr(exc, "read", None)
            if callable(body):
                try:
                    detail = body().decode("utf-8", "ignore")[:300]
                except Exception:
                    detail = ""
            logger.warning(
                "[RAG][LLM] call FAILED purpose=%s error=%s %s",
                purpose,
                exc.__class__.__name__,
                detail or str(exc),
            )
            raise
        content = ""
        message = {}
        finish_reason = ""
        try:
            choice = data["choices"][0]
            message = choice.get("message") or {}
            content = (message.get("content") or "").strip()
            finish_reason = choice.get("finish_reason") or ""
        except Exception:
            content = ""
        logger.info(
            "[RAG][LLM] response purpose=%s http=%s chars=%d finish=%s",
            purpose,
            status,
            len(content),
            finish_reason or "?",
        )
        if not content:
            # Diagnose an empty answer: wrong model name, content filter, or the
            # whole token budget consumed by reasoning show up here.
            try:
                raw = json.dumps(data)[:500]
            except Exception:
                raw = str(data)[:500]
            logger.warning(
                "[RAG][LLM] EMPTY content purpose=%s finish=%s reasoning_chars=%d usage=%s msg_keys=%s raw=%s",
                purpose,
                finish_reason or "?",
                len(str((message or {}).get("reasoning_content") or "")),
                data.get("usage") if isinstance(data, dict) else None,
                sorted(message.keys()) if isinstance(message, dict) else None,
                raw,
            )
        return content

    def query(self, query: str, k: int = 3, threshold: float = 0.2):
        self.last_query_error = ""
        if not self.vector_db and not self.chroma_collection:
            self.last_query_error = self.retrieval_error or "Vector database is unavailable."
            return []
        try:
            score_threshold = float(os.getenv("RAG_QUERY_THRESHOLD", str(threshold)))
            output = []
            vector_ready = self.vector_db and (self.collection_count is None or self.collection_count > 0)
            chroma_ready = self.chroma_collection and self.embedder and (self.collection_count is None or self.collection_count > 0)
            if vector_ready:
                results = self.vector_db.similarity_search_with_score(query, k=k)
                raw_scores = []
                for doc, dist in results:
                    score = 1.0 / (1.0 + dist) if isinstance(dist, (int, float)) and dist >= 0 else 0.0
                    raw_scores.append(score)
                    if score < score_threshold:
                        continue
                    metadata = doc.metadata or {}
                    source = self._format_reference_source(metadata)
                    output.append({
                        "content": (doc.page_content or "")[:500],
                        "source": source,
                        "score": score,
                        "metadata": metadata,
                    })
                self._log_retrieval(query, "langchain", raw_scores, len(output))
                return self._dedupe_results(output)
            if chroma_ready:
                vec = self._embed_query_text(query)
                if vec is None:
                    self._log_retrieval(query, "chroma", [], 0)
                    return []
                results = self.chroma_collection.query(
                    query_embeddings=[vec],
                    n_results=k,
                    include=["documents", "metadatas", "distances"],
                )
                docs = results.get("documents", [[]])[0]
                metas = results.get("metadatas", [[]])[0]
                dists = results.get("distances", [[]])[0]
                raw_scores = []
                for doc, meta, dist in zip(docs, metas, dists):
                    score = 1.0 / (1.0 + dist) if isinstance(dist, (int, float)) and dist >= 0 else 0.0
                    raw_scores.append(score)
                    if score < score_threshold:
                        continue
                    metadata = meta or {}
                    source = self._format_reference_source(metadata)
                    output.append({
                        "content": (doc or "")[:500],
                        "source": source,
                        "score": score,
                        "metadata": metadata,
                    })
                self._log_retrieval(query, "chroma", raw_scores, len(output))
                return self._dedupe_results(output)
        except Exception as exc:
            self.last_query_error = f"{type(exc).__name__}: {str(exc)[:240]}"
            logger.exception("[RAG][RETRIEVE] query failed: %s", query[:120])
            return []

    def _latexify_inline_species(self, text):
        if not text:
            return ""
        plain = plain_formula(text)
        leading_coefficient = ""
        coefficient_match = re.match(r"^(\d+)(?=[A-Z])", plain)
        if coefficient_match:
            leading_coefficient = coefficient_match.group(1)
            plain = plain[coefficient_match.end():]
        tokens = re.findall(r'[A-Z][a-z]?|[a-z]+|\d+|[^A-Za-z0-9]+', plain)
        latex = leading_coefficient
        buf = ""
        def _flush_buf():
            nonlocal latex, buf
            if buf:
                latex += f"\\text{{{buf}}}"
                buf = ""
        for tok in tokens:
            if tok.isdigit():
                _flush_buf()
                latex += f"_{{{tok}}}"
                continue
            if re.match(r'^[A-Z][a-z]?$', tok):
                _flush_buf()
                latex += f"\\mathrm{{{tok}}}"
                continue
            if re.match(r'^[a-z]+$', tok):
                buf += tok
                continue
            buf += tok
        _flush_buf()
        return f"${latex}$"

    def _build_loose_formula_pattern(self, plain):
        if not plain:
            return ""
        chars = [re.escape(ch) for ch in plain]
        return r'\b' + r'\s*'.join(chars) + r'\b'

    def _extract_graph_refs(self, path_str, full_path=None):
        node_ids = []
        edge_ids = []
        if full_path and isinstance(full_path, (list, tuple)):
            for idx, node in enumerate(full_path):
                if idx % 2 == 0:
                    if node and node not in node_ids:
                        node_ids.append(node)
                else:
                    if node and node not in edge_ids:
                        edge_ids.append(node)
        if not node_ids:
            species = path_species(path_str)
            for sp in species:
                if sp not in node_ids:
                    node_ids.append(sp)
        return {"node_ids": node_ids, "edge_ids": edge_ids}



    def _build_evidence_blocks(self, ref_list, graph_refs):
        blocks = []
        for ref in ref_list or []:
            dois = extract_dois(ref)
            blocks.append({
                "source": ref,
                "doi": dois[0] if dois else None,
                "snippet": None,
                "node_ids": graph_refs.get("node_ids", []),
                "edge_ids": graph_refs.get("edge_ids", []),
            })
        return blocks






    def _get_evidence_matcher(self):
        matcher = getattr(self, "evidence_matcher", None)
        if matcher is None:
            matcher = build_default_matcher(os.path.dirname(os.path.abspath(__file__)))
            self.evidence_matcher = matcher
        return matcher

    def _authored_step_types(self):
        """Steps the network author marked as something other than literature.

        Keyed by (reactant, condition, product). A step marked "extended" is one
        the author derived chemically and knows the source does not state, so a
        report should say the gap is deliberate rather than implying the
        evidence search came up short.
        """
        cache = getattr(self, "_authored_step_type_cache", None)
        if cache is not None:
            return cache
        marked = {}
        try:
            from tracernet.crn.repository import CRNRepository

            repo = CRNRepository.from_xlsx(
                os.path.join(os.path.dirname(os.path.abspath(__file__)), "CRNs.xlsx")
            )
            for row in repo.rows:
                species = [row.reactant, *row.products]
                for step in range(1, len(row.products) + 1):
                    kind = row.evidence_type_for(step)
                    if kind and kind != "literature":
                        marked[(species[step - 1], row.condition, species[step])] = kind
        except Exception:
            marked = {}
        self._authored_step_type_cache = marked
        return marked

    def _document_text(self, doi):
        """Every chunk of one source, joined, for document-scoped conditions.

        A paper states its conditions once in the methods section and does not
        repeat them beside each result, so the condition for an edge is usually
        outside the passage that describes the conversion. Read straight from
        the metadata table rather than through the vector store: this needs no
        embeddings, so it still works when the vector backend is unavailable.
        """
        key = str(doi or "").strip().lower()
        if not key:
            return ""
        cache = getattr(self, "_document_text_cache", None)
        if cache is None:
            cache = self._document_text_cache = {}
        if key in cache:
            return cache[key]
        text = ""
        try:
            import sqlite3

            db_root = DB_PATH or os.path.dirname(os.path.abspath(__file__))
            db = os.path.join(db_root, "chroma.sqlite3")
            with sqlite3.connect(f"file:{db}?mode=ro", uri=True) as conn:
                rows = conn.execute(
                    "select d.string_value from embedding_metadata d "
                    "join embedding_metadata m on m.id = d.id "
                    "where d.key='chroma:document' and m.key='doi' "
                    "and lower(trim(m.string_value)) = ?",
                    (key,),
                ).fetchall()
            text = " ".join(r[0] for r in rows if r and r[0])
        except Exception:
            text = ""
        cache[key] = text
        return text

    def _adjacent_document_text(self, doi, chunk_index, current_text):
        """Join only neighbouring chunks from one DOI for composite evidence.

        Papers commonly separate the experimental setup from the sentence that
        reports the product.  A bounded +/-1 chunk window recovers that local
        context without treating unrelated experiments elsewhere in the same
        paper as one claim.
        """
        key = str(doi or "").strip().lower()
        try:
            center = int(chunk_index)
        except (TypeError, ValueError):
            return str(current_text or "")
        if not key:
            return str(current_text or "")
        cache = getattr(self, "_adjacent_document_text_cache", None)
        if cache is None:
            cache = self._adjacent_document_text_cache = {}
        cache_key = (key, center)
        if cache_key in cache:
            return cache[cache_key]
        combined = str(current_text or "")
        try:
            import sqlite3

            db_root = DB_PATH or os.path.dirname(os.path.abspath(__file__))
            db = os.path.join(db_root, "chroma.sqlite3")
            with sqlite3.connect(f"file:{db}?mode=ro", uri=True) as conn:
                rows = conn.execute(
                    "select ci.int_value, d.string_value "
                    "from embedding_metadata d "
                    "join embedding_metadata m on m.id=d.id "
                    "join embedding_metadata ci on ci.id=d.id "
                    "where d.key='chroma:document' and m.key='doi' "
                    "and ci.key='chunk_index' "
                    "and lower(trim(m.string_value))=? "
                    "and ci.int_value between ? and ? "
                    "order by ci.int_value",
                    (key, center - 1, center + 1),
                ).fetchall()
            texts = [str(row[1] or "").strip() for row in rows if row and row[1]]
            if texts:
                combined = "\n".join(dict.fromkeys(texts))
        except Exception:
            combined = str(current_text or "")
        cache[cache_key] = combined
        return combined

    def _path_evidence_matches(
        self, content, path_str, provenance_level="unspecified", doi=None
    ):
        if not content or not path_str:
            return []
        # One path analysis issues the same query in several spellings, and the
        # variants retrieve overlapping documents, so the same chunk is matched
        # against the same path many times over -- eight times for some sources
        # in a single run. Matching is a pure function of these four inputs, so
        # the repeats can be served from memory. Scoped to the service instance,
        # which lives for one process.
        key = (
            str(path_str),
            str(provenance_level),
            str(doi or ""),
            len(content),
            hash(content),
        )
        cache = getattr(self, "_edge_match_cache", None)
        if cache is None:
            cache = self._edge_match_cache = {}
        if key not in cache:
            cache[key] = self._get_evidence_matcher().match_path(
                path_str,
                content,
                provenance_level=provenance_level,
                condition_context=self._document_text(doi),
            )
        return cache[key]

    def _path_endpoint_evidence_match(
        self,
        content,
        path_str,
        provenance_level="unspecified",
    ):
        if not content or not path_str:
            return None
        matcher = self._get_evidence_matcher()
        endpoint_matcher = getattr(matcher, "best_endpoint_match", None)
        if not callable(endpoint_matcher):
            return None
        # Same rationale as _path_evidence_matches: a single path analysis issues
        # the query in several spellings, and each retrieved chunk is offered to
        # both the resolver and the lexical route, so this endpoint match was
        # being recomputed many times over for identical inputs. It is a pure
        # function of these three inputs, so serve the repeats from memory.
        key = (
            str(path_str),
            str(provenance_level),
            len(content),
            hash(content),
        )
        cache = getattr(self, "_endpoint_match_cache", None)
        if cache is None:
            cache = self._endpoint_match_cache = {}
        if key not in cache:
            cache[key] = endpoint_matcher(
                path_str,
                content,
                provenance_level=provenance_level,
            )
        return cache[key]

    def _evidence_corpus(self, collection, batch_size, max_documents):
        """Read the local corpus once and cache text, folded form and BM25 stats.

        The four-route scan reads every chunk from Chroma, normalises each one
        several times (resolver prefilter, lexical prefilter, BM25 tokeniser)
        and recomputes BM25 document frequencies -- all of it identical from
        path to path because the corpus is static. Doing it once per process
        turns a multi-second-per-path cost into a one-off. Only raw text and
        metadata are cached; per-path metadata enrichment still happens at the
        use sites, so provenance stays path-specific.
        """
        try:
            signature = collection.count()
        except Exception:
            signature = None
        cache = getattr(self, "_evidence_corpus_cache", None)
        if (
            cache is not None
            and cache.get("signature") == signature
            and cache.get("max_documents") == max_documents
        ):
            return cache
        docs = []
        offset = 0
        while offset < max_documents:
            payload = collection.get(
                limit=min(batch_size, max_documents - offset),
                offset=offset,
                include=["documents", "metadatas"],
            )
            documents = list(payload.get("documents") or [])
            metadatas = list(payload.get("metadatas") or [])
            if not documents:
                break
            if len(metadatas) < len(documents):
                metadatas.extend(
                    {} for _ in range(len(documents) - len(metadatas))
                )
            for document, metadata in zip(documents, metadatas):
                docs.append((document, metadata or {}))
            offset += len(documents)
            if len(documents) < batch_size:
                break
        # ``folded`` is exactly what resolver_may_match/document_may_match used
        # to compute per document per route; ``tokenized`` matches bm25_tokens
        # applied to the same folded text, so scoring is byte-for-byte identical.
        folded = [normalize_text(document).casefold() for document, _ in docs]
        tokenized = [
            re.findall(r"[a-z0-9]+(?:[-_][a-z0-9]+)*", text) for text in folded
        ]
        doc_freq = Counter()
        for tokens in tokenized:
            doc_freq.update(set(tokens))
        corpus_size = len(tokenized)
        average_length = (
            sum(len(tokens) for tokens in tokenized) / corpus_size
            if corpus_size else 0.0
        )
        cache = {
            "signature": signature,
            "max_documents": max_documents,
            "docs": docs,
            "folded": folded,
            "tokenized": tokenized,
            "doc_freq": doc_freq,
            "corpus_size": corpus_size,
            "average_length": average_length,
        }
        self._evidence_corpus_cache = cache
        return cache

    def _get_bibliography(self):
        registry = getattr(self, "bibliography", None)
        if registry is None:
            registry = BibliographyRegistry.load(
                os.path.dirname(os.path.abspath(__file__))
            )
            self.bibliography = registry
        return registry

    def _format_doi_reference(self, doi, metadata=None):
        normalized_doi = normalize_doi(doi)
        values = dict(self._resolve_doi_metadata(normalized_doi) or {})
        values.update(dict(metadata or {}))
        values["doi"] = normalized_doi
        record = self._get_bibliography().resolve(values)
        if record is None:
            record = BibliographicRecord.from_mapping(values)
        return format_gbt7714(record)

    def _resolve_doi_metadata(self, doi):
        """Resolve and cache generic DOI metadata without paper-specific rules."""
        normalized_doi = normalize_doi(doi)
        if not normalized_doi:
            return None
        cache = getattr(self, "_doi_metadata_cache", None)
        if cache is None:
            cache = {}
            self._doi_metadata_cache = cache
        if normalized_doi in cache:
            return cache[normalized_doi]

        resolver = getattr(self, "doi_metadata_resolver", None)
        if not callable(resolver):
            cache[normalized_doi] = None
            return None
        try:
            resolved = resolver(normalized_doi)
        except Exception:
            resolved = None
        if not isinstance(resolved, dict):
            resolved = None
        elif normalize_doi(resolved.get("doi")) != normalized_doi:
            resolved = None
        cache[normalized_doi] = resolved
        return resolved

    @staticmethod
    def _reference_title_key(value):
        """Return a conservative key for matching a title or document filename."""
        text = re.sub(r"\s+", " ", str(value or "")).strip()
        if not text:
            return ""
        text = text.replace("\\", "/").rsplit("/", 1)[-1]
        text = re.sub(r"\.(?:md|txt|pdf|docx?|html?)$", "", text, flags=re.I)
        return re.sub(r"[^a-z0-9]+", " ", text.casefold()).strip()

    def _normalize_reference_metadata(self, metadata):
        """Normalize citation metadata without paper-specific aliases."""
        values = dict(metadata or {}) if isinstance(metadata, dict) else {}
        doi = normalize_doi(
            values.get("doi") or values.get("DOI") or values.get("doi_id")
        )
        if not doi:
            for key in (
                "citation", "reference", "full_citation", "formatted_reference",
                "source", "source_file", "filename", "file_name", "url", "URL",
            ):
                candidates = extract_dois(values.get(key))
                if candidates:
                    doi = candidates[0]
                    break
        if doi:
            values["doi"] = doi

        title = (
            values.get("title")
            or values.get("paper_title")
            or values.get("document_title")
        )
        if not str(title or "").strip():
            source = (
                values.get("source_file")
                or values.get("filename")
                or values.get("file_name")
            )
            if isinstance(source, str) and source.strip():
                source = source.replace("\\", "/").rsplit("/", 1)[-1]
                title = re.sub(
                    r"\.(?:md|txt|pdf|docx?|html?)$", "", source, flags=re.I
                ).strip()
        if str(title or "").strip():
            values["title"] = re.sub(r"\s+", " ", str(title)).strip()
        return values

    def _provenance_metadata_records(self, path_sources, path_dois):
        """Build generic CRN provenance records for DOI/title reconciliation."""
        records = []
        seen = set()
        registry = self._get_bibliography()
        for raw_source in list(path_sources or []) + list(path_dois or []):
            source = clean_cell(raw_source)
            doi = first_doi(source)
            record = registry.resolve({"doi": doi}) if doi else None
            resolved_metadata = self._resolve_doi_metadata(doi) if doi else None
            if record is None and resolved_metadata:
                record = BibliographicRecord.from_mapping(resolved_metadata)
            title = record.title if record is not None else ""
            if not title and source:
                title = re.sub(
                    r"(?:https?://(?:dx\.)?doi\.org/|\bdoi\s*:\s*)?"
                    r"10\.\d{4,9}/[^\s,;]+",
                    "",
                    source,
                    flags=re.I,
                )
                title = re.sub(
                    r"\.(?:md|txt|pdf|docx?|html?)$", "",
                    title.replace("\\", "/").rsplit("/", 1)[-1],
                    flags=re.I,
                ).strip(" .;,:-")
            item = {
                "source": source,
                "doi": normalize_doi(doi),
                "title": title,
                "metadata": dict(resolved_metadata or {}),
            }
            key = (
                item["doi"], self._reference_title_key(item["title"]),
                source.casefold(),
            )
            if key not in seen and any(item.values()):
                seen.add(key)
                records.append(item)
        return records

    def _enrich_reference_metadata(self, metadata, provenance_records):
        """Backfill DOI/title only when generic identifiers match exactly."""
        values = self._normalize_reference_metadata(metadata)
        doi = normalize_doi(values.get("doi"))
        title_key = self._reference_title_key(values.get("title"))
        for candidate in provenance_records or []:
            candidate_doi = normalize_doi(candidate.get("doi"))
            candidate_title = str(candidate.get("title") or "").strip()
            candidate_title_key = self._reference_title_key(candidate_title)
            doi_match = bool(doi and candidate_doi and doi == candidate_doi)
            title_match = bool(
                title_key and candidate_title_key and title_key == candidate_title_key
            )
            if not (doi_match or title_match):
                continue
            if not doi and candidate_doi:
                values["doi"] = candidate_doi
                doi = candidate_doi
            if not title_key and candidate_title:
                values["title"] = candidate_title
                title_key = candidate_title_key
            for key, value in (candidate.get("metadata") or {}).items():
                if value not in (None, "", [], {}) and not values.get(key):
                    values[key] = value
            break
        return values

    def _format_reference_source(self, metadata):
        if not isinstance(metadata, dict):
            return "Unknown"
        metadata = self._normalize_reference_metadata(metadata)
        record = self._get_bibliography().resolve(metadata)
        if record is not None:
            return format_gbt7714(record)

        inline = BibliographicRecord.from_mapping(metadata)
        has_publication_metadata = bool(
            inline.authors
            or inline.journal
            or inline.year
            or inline.volume
            or inline.issue
            or inline.pages
            or inline.article_number
        )
        if inline.title and inline.doi and not has_publication_metadata:
            return f"{inline.title}. DOI: {inline.doi}."
        has_structured_metadata = bool(
            has_publication_metadata
            or inline.doi
            or inline.url
        )
        if inline.title and not has_structured_metadata:
            return inline.title
        if inline.title or has_structured_metadata:
            return format_gbt7714(inline, mark_incomplete=True)

        for key in ("citation", "reference", "full_citation", "formatted_reference"):
            value = metadata.get(key)
            if isinstance(value, str) and value.strip():
                return re.sub(r"\s+", " ", value).strip()

        source = (
            metadata.get("source_file")
            or metadata.get("source")
            or metadata.get("filename")
            or metadata.get("file_name")
        )
        if isinstance(source, str) and source.strip():
            source = source.replace("\\", "/").rsplit("/", 1)[-1]
            return re.sub(r"\.[A-Za-z0-9]+$", "", source).strip()

        doi = metadata.get("doi") or metadata.get("DOI") or metadata.get("doi_id")
        return self._format_doi_reference(doi) if doi else "Unknown"





    def _normalize_inline_formulas(self, content, path_str):
        if not content:
            return content
        preserved = []

        def _capture(match):
            preserved.append(match.group(0))
            return f"__MATHBLOCKX{len(preserved) - 1}__"

        content = re.sub(r'\$\$.*?\$\$', _capture, content, flags=re.DOTALL)
        content = re.sub(r'\\\[.*?\\\]', _capture, content, flags=re.DOTALL)
        
        content = re.sub(r'\$(?!\$)(.*?)\$', r'\1', content)
        content = re.sub(r'\\\((.*?)\\\)', r'\1', content)
        
        # Strip only genuine LaTeX "\text" artifacts (real backslash required).
        # The previous optional-backslash / case-insensitive [A-Z] patterns also
        # deleted the substrings text/ext/tex from ordinary words
        # (contextual->conual, external->ernal, texture->ture).
        content = re.sub(r'(?i)\\\s*t\s*e\s*x\s*t\s*\{([^}]*)\}', r'\1', content)
        content = re.sub(r'(?i)\\\s*t\s*e\s*x\s*t\b\s*', '', content)
        for _ in range(3):
            normalized_wrappers = re.sub(
                r"(?i)\\(?:mathrm|mathbf|mathit|operatorname)\s*\{([^{}]*)\}",
                r"\1",
                content,
            )
            if normalized_wrappers == content:
                break
            content = normalized_wrappers
        content = re.sub(
            r"(?i)(?<![A-Za-z])(?:mathrm|mathbf|mathit|operatorname)"
            r"(?=[A-Z][a-z]?)",
            "",
            content,
        )
        content = re.sub(r'\\alpha', '\u03b1', content)
        content = re.sub(r'\\beta', '\u03b2', content)
        content = re.sub(r'\\gamma', '\u03b3', content)
        content = re.sub(r'_\{?(\d+)\}?', r'\1', content)
        # Language models occasionally wrap a chemical token in bare TeX
        # grouping braces without a command, e.g. `{CuS}` or `{H₂S}`. These
        # braces have no semantic value in prose and render visibly in Gradio.
        content = re.sub(
            r"\{\s*([A-Z][A-Za-z0-9₀₁₂₃₄₅₆₇₈₉⁰¹²³⁴⁵⁶⁷⁸⁹()\[\]·.\-+^]*)\s*\}",
            r"\1",
            content,
        )
        species = path_species(path_str)

        # Rebuild the subject of common generated transition phrases from the
        # selected graph endpoints. This catches severely malformed model
        # output such as missing hydrate dots/parentheses, where ordinary
        # formula-token normalization cannot safely recover the identity.
        if len(species) >= 2:
            matcher = self._get_evidence_matcher()

            def _endpoint_label(entity, generated_segment):
                canonical = canonical_formula(entity)
                entity_phase = formula_phase(entity)
                # Restrict to names registered for this exact polymorph. Without
                # the phase filter, realgar and pararealgar (both As4S4) would be
                # treated as labels for the same node.
                aliases = {
                    record.name
                    for (phase, formula), records in matcher.resolver.by_formula.items()
                    if formula == canonical and phase == entity_phase
                    for record in records
                    if str(record.name or "").strip()
                }
                matched_aliases = [
                    alias for alias in aliases
                    if re.search(
                        rf"(?<!\w){re.escape(alias)}(?!\w)",
                        generated_segment,
                        flags=re.IGNORECASE,
                    )
                ]
                alias = max(matched_aliases, key=len) if matched_aliases else ""
                formula = plain_formula(entity)
                return f"{formula} ({alias})" if alias else formula

            transition_subject = re.compile(
                r"(?P<prefix>\b(?:reaction\s+pathway|conversion|transformation)\s+"
                r"from\s+)(?P<source>[^.;\n]{1,140}?)(?P<connector>\s+to\s+)"
                r"(?P<product>[^.;\n]{1,100}?)(?=\s+(?:is|was|occurs?|occurred|"
                r"takes?|proceeds?|under|upon|when|where|through)\b)",
                flags=re.IGNORECASE,
            )

            def _repair_transition_subject(match):
                return (
                    match.group("prefix")
                    + _endpoint_label(species[0], match.group("source"))
                    + match.group("connector")
                    + _endpoint_label(species[-1], match.group("product"))
                )

            content = transition_subject.sub(_repair_transition_subject, content)

        # When generated prose names a database alias and follows it with a
        # chemical-looking parenthetical formula, enforce the exact formula
        # from the selected graph path. Ordinary descriptive parentheticals
        # remain unchanged.
        matcher = self._get_evidence_matcher()
        for graph_species in species:
            canonical = canonical_formula(graph_species)
            if not canonical:
                continue
            species_phase = formula_phase(graph_species)
            # Only names registered for this exact polymorph may have their
            # parenthetical formula rewritten to this node; otherwise
            # "realgar (alpha-As4S4)" would be rewritten to the pararealgar
            # (p-As4S4) node, which shares the same formula.
            aliases = {
                record.name
                for (phase, formula), records in matcher.resolver.by_formula.items()
                if formula == canonical and phase == species_phase
                for record in records
                if str(record.name or "").strip()
            }
            for alias in sorted(aliases, key=len, reverse=True):
                # Only for a genuinely unclosed parenthetical. Excluding "()"
                # from the inner class stops this from running past a closing
                # paren and deleting following prose. A comma must NOT terminate
                # it either: "arsenolite (As2O3, with As in the +3 state)" is a
                # well-formed parenthetical and was being turned into
                # "arsenolite (As2O3), with As in the +3 state)".
                tolerant_alias_pattern = re.compile(
                    rf"(?<!\w)({re.escape(alias)})(\s*)\("
                    r"([^().;\n]{1,40}?)"
                    r"(?=\s+(?:to|under|upon|when|after|before|was|were|is|are|"
                    r"occurs?|converts?|converted|degrades?|changes?|forms?)\b|[.;])",
                    flags=re.IGNORECASE,
                )
                alias_pattern = re.compile(
                    rf"(?<!\w)({re.escape(alias)})(\s*)"
                    r"\(((?:[^()]|\([^()]*\))*)\)",
                    flags=re.IGNORECASE,
                )

                def _correct_alias_formula(match, expected=graph_species):
                    inner = match.group(3)
                    chemical_like = (
                        bool(parse_formula(inner))
                        or (
                            bool(re.search(r"\d", inner))
                            and bool(
                                re.fullmatch(
                                    r"[\sA-Za-z0-9()\[\]{}·.\-+]+",
                                    inner,
                                )
                            )
                        )
                    )
                    if not chemical_like:
                        return match.group(0)
                    return f"{match.group(1)} ({plain_formula(expected)})"

                content = tolerant_alias_pattern.sub(_correct_alias_formula, content)
                content = alias_pattern.sub(_correct_alias_formula, content)

        targets = []
        for sp in species:
            if not re.search(r'\d', sp):
                continue
            plain = plain_formula(sp)
            if plain:
                targets.append(plain)
        if not targets:
            for i, block in enumerate(preserved):
                content = content.replace(f"__MATHBLOCKX{i}__", block)
            return self._repair_math_artifacts(content)

        replacements = []
        for plain in sorted(set(targets), key=len, reverse=True):
            pattern = self._build_loose_formula_pattern(plain)
            replacements.append((pattern, self._latexify_inline_species(plain)))

        parts = re.split(r'(__MATHBLOCKX\d+__)', content)
        for idx, part in enumerate(parts):
            if not part or part.startswith("__MATHBLOCKX"):
                continue
            for pattern, dst in replacements:
                if not pattern:
                    continue
                part = re.sub(pattern, lambda m, d=dst: d, part)
            parts[idx] = part

        content = "".join(parts)
        for i, block in enumerate(preserved):
            content = content.replace(f"__MATHBLOCKX{i}__", block)
        return self._repair_math_artifacts(content)


    def _repair_math_artifacts(self, text: str) -> str:
        if not text:
            return text
        
        preserved_blocks = []
        def _capture_block(match):
            preserved_blocks.append(match.group(0))
            return f"__LATEXBLOCK{len(preserved_blocks) - 1}__"
        inline_blocks = []
        def _capture_inline(match):
            inline_blocks.append(match.group(0))
            return f"__INLINEMATHBLOCK{len(inline_blocks) - 1}__"
        
        text = re.sub(r'\$\$.*?\$\$', _capture_block, text, flags=re.DOTALL)
        text = re.sub(r'\\\[.*?\\\]', _capture_block, text, flags=re.DOTALL)
        text = re.sub(r'\$[^$]+\$', _capture_inline, text)
        
        text = re.sub(r'\\mathrm\{\s*([^}]+?)\s*\)', r'\\mathrm{\1}', text)
        def _fix_unclosed(match):
            inner = match.group(1)
            tail = match.group(2)
            if "}" in inner:
                return match.group(0)
            return f"\\mathrm{{{inner}}}{tail}"
        text = re.sub(r'\\mathrm\{([^}\n]{1,50})([)\].,;:])', _fix_unclosed, text)
        # Strip only genuine LaTeX artifacts (the "\text" command with a real
        # backslash). The previous patterns used an optional backslash and a
        # case-insensitive [A-Z], so they also deleted the substrings "text"/
        # "ext" from ordinary words (contextual->conual, external->ernal).
        text = re.sub(r'(?i)(?<!\\xrightarrow\{)\\\s*t\s*e\s*x\s*t\s*', '', text)
        text = re.sub(r'(?<!\\xrightarrow\{)\\ext(?=[A-Za-z])', '', text)
        text = re.sub(
            r'\b(?:[A-Za-z]\s+){2,}[A-Za-z]\b',
            lambda match: match.group(0).replace(" ", ""),
            text,
        )
        
        for i, block in enumerate(preserved_blocks):
            text = text.replace(f"__LATEXBLOCK{i}__", block)
        for i, block in enumerate(inline_blocks):
            text = text.replace(f"__INLINEMATHBLOCK{i}__", block)
        
        return text

    def _build_latex_path(self, path_str):
        edges = parse_path(path_str)
        if not edges:
            return ""

        def _latex_species(text):
            t = str(text).strip()
            if not t:
                return ""
            plain = plain_formula(t)
            return self._latexify_inline_species(plain).strip("$")

        latex = _latex_species(edges[0].reactant)
        for edge in edges:
            cond = _display_condition_casing(edge.condition)
            if cond:
                arrow = f"\\xrightarrow{{{cond}}}"
            else:
                arrow = "\\rightarrow"
            latex += f" {arrow} {_latex_species(edge.product)}"
        return latex.strip()


    def _generate_unverified_path_report(
        self,
        root_material,
        path_str,
        retrieved_context=False,
    ):
        species = path_species(path_str)
        conditions = path_conditions(path_str)
        conditions = list(dict.fromkeys(conditions))
        start_material = species[0] if species else root_material
        final_product = species[-1] if len(species) >= 2 else "the mapped product"
        condition_text = (
            ", ".join(
                _display_condition_casing(format_chemical_formula(item))
                for item in conditions
            )
            if conditions
            else "unspecified graph conditions"
        )
        relation = " -> ".join(
            format_chemical_formula(item) for item in species
        ) or (
            f"{format_chemical_formula(start_material)} -> "
            f"{format_chemical_formula(final_product)}"
        )
        retrieval_available = bool(
            getattr(self, "vector_db", None)
            or getattr(self, "chroma_collection", None)
        )
        return compose_graph_only_note(
            relation,
            condition_text,
            retrieved_context=bool(retrieved_context),
            retrieval_available=retrieval_available,
        )

    def _authored_steps_for_path(self, path_str):
        """The author's notes on the steps of one path, as (step, kind) pairs."""
        marked = self._authored_step_types()
        out = []
        for edge in parse_path(path_str):
            kind = marked.get((edge.reactant, edge.condition, edge.product))
            if kind:
                out.append((f"{edge.reactant} -> {edge.product}", kind))
        return out

    def _generate_contextual_narrative(
        self,
        root_material,
        path_str,
        safe_context,
        related_context=None,
        authored_steps=None,
    ):
        """Synthesize a grounded context paragraph when no edge is verified.

        When retrieval returns literature that is topically relevant but not
        classified as direct evidence for a specific graph edge, a bare "no
        evidence" statement wastes the retrieved content. Instead, summarize
        what the studies actually report (samples, conditions, characterization,
        observed substances, aging behavior) as *related context*, explicitly
        without claiming the graph pathway or its steps are verified.
        """
        # Number the sources the way the evidence-backed branch does. Passing
        # "(Source: <full citation>)" left the model with no numbers to use, so
        # it fell back to author-year while every other report cites [n], and
        # the renderer had no numbering to build a reference list from -- which
        # is why those citations resolved to nothing.
        used = []
        lines = []
        for item in (related_context or []):
            if not isinstance(item, dict) or not str(item.get("snippet") or "").strip():
                continue
            source = str(item.get("source") or "").strip()
            if source and source not in used:
                used.append(source)
            index = (used.index(source) + 1) if source in used else len(used) + 1
            lines.append(f"[{index}] {item.get('snippet')}")
            if len(lines) >= 4:
                break
        related_text = "\n\n".join(lines)
        self._contextual_sources = tuple(used)
        primary = (safe_context or "").strip()
        # Never feed the "nothing retrieved" placeholder to the model: it makes
        # the LLM echo "no literature was retrieved" and ignore the related
        # paragraphs that actually are available.
        if primary.startswith("No relevant literature evidence was retrieved"):
            primary = ""
        parts = [chunk for chunk in (primary, related_text) if chunk]
        if not parts:
            return ""
        context = "\n\nAdditional related paragraphs:\n".join(parts) if len(parts) == 2 else parts[0]

        system_prompt = (
            "You are a heritage conservation scientist. The reaction graph proposes "
            "a degradation pathway, but the retrieved literature does not contain a "
            "snippet that directly verifies its specific steps. Using ONLY the "
            "retrieved literature provided, write one cohesive paragraph of RELATED "
            "CONTEXT that helps a reader interpret the proposed pathway. Describe, "
            "where the sources state them: the samples or experiments studied, the "
            "environmental or experimental conditions, the characterization "
            "techniques used, the substances or phases observed, and how these "
            "changed with aging or exposure over time. Attribute observations to "
            "their citation [n]. Never claim the literature proves the graph "
            "pathway, its intermediates, or its elementary steps; frame it as "
            "supporting context. If the retrieved text does not actually concern "
            "this pathway, say so plainly instead of inventing a connection."
        )
        # This branch runs whenever entity matching found no direct evidence,
        # which is not the same as the source being silent: it also fires when
        # the paper writes a species differently from the network. Without this,
        # the prompt's blanket "the literature does not verify its steps" was
        # asserted in the prose about steps the paper does state.
        ionic = [step for step, kind in (authored_steps or []) if kind == "ionic_form"]
        extended = [step for step, kind in (authored_steps or []) if kind == "extended"]
        if ionic:
            system_prompt += (
                " The cited source DOES state these steps, written as ionic "
                f"half-reactions rather than as the salts the graph names: "
                f"{'; '.join(ionic)}. Describe them as reported by the source. "
                "Do not say the literature fails to verify them."
            )
        if extended:
            system_prompt += (
                " These steps are the network author's own chemical "
                f"extrapolation and the source does not state them: "
                f"{'; '.join(extended)}. Say so plainly."
            )
        # The closing sentence is what the reader takes away, so it must not
        # deny steps the source states. Where the author has recorded that the
        # source gives them in ionic form, the caveat is scoped to the rest.
        if ionic:
            closing = (
                "End with one sentence distinguishing the steps the source "
                "states in ionic form from the remaining steps, which these "
                "observations do not by themselves verify."
            )
            heading = "Proposed graph pathway:"
        else:
            closing = (
                "End with one sentence making clear these are contextual "
                "observations that do not by themselves verify the proposed "
                "pathway steps."
            )
            heading = "Proposed graph pathway (NOT yet verified):"
        user_prompt = f"""
{heading} {path_str}
Starting material: {root_material}

Retrieved literature (with evidence classification labels):
{context}

Write one paragraph (about 4-7 sentences) of related context following the
rules. Do not add a title, headings, equations, or a references list. Do not
include a "Validation note"; it is added automatically outside your paragraph.
{closing}
"""
        try:
            content = self._chat_completion(
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_prompt},
                ],
                temperature=0.3,
                max_tokens=1000,
                purpose="contextual_narrative",
            ) or ""
        except Exception:
            content = ""
        content = (content or "").strip()
        if is_bad_generation_text(content):
            content = ""
        # Reject a degenerate "nothing was retrieved" echo: real related context
        # was supplied, so an answer that *opens* by denying it is a generation
        # failure and should defer to the terse note instead. (A narrative that
        # mentions "no direct evidence" mid-sentence is fine and kept.)
        if content and re.match(
            r"\s*no\s+(?:relevant\s+)?(?:literature|evidence|context|"
            r"contextual|retrieved|snippet|pathway-matching)\b",
            content,
            flags=re.IGNORECASE,
        ):
            content = ""
        if content and len(content) < 40:
            content = ""
        logger.info(
            "[RAG][LLM] contextual_narrative %s",
            f"used LLM output ({len(content)} chars)" if content
            else "empty/rejected -> terse note fallback",
        )
        return content

    def _build_report_payload(
        self,
        content,
        root_material,
        path_str,
        ref_list,
        ref_snippets,
        full_path=None,
        evidence_is_weak=False,
        retrieval_trace=None,
        related_context=None,
        contextual_narrative=False,
    ):
        latex_path = self._build_latex_path(path_str)
        latex_block = latex_path.strip() if latex_path else ""
        if latex_block and not (
            latex_block.startswith("$$") or latex_block.startswith("\\[")
        ):
            latex_block = f"$$\n{latex_block}\n$$"

        species = path_species(path_str)
        start_material = species[0] if species else root_material
        final_product = species[-1] if len(species) >= 2 else "the mapped product"
        header = (
            "Provenance Analysis Report: Pathway from "
            f"{format_chemical_formula(start_material)} to "
            f"{format_chemical_formula(final_product)} in Mural Pigments"
        )
        body = (content or "").strip()
        report = (
            body
            if body.lower().startswith("provenance analysis report:")
            else f"{header}\n\n{body}\n\n{latex_block}"
        )
        report = self._normalize_inline_formulas(report, path_str)

        graph_refs = self._extract_graph_refs(path_str, full_path=full_path)
        path_terms = (
            path_species(path_str)
            + path_conditions(path_str)
        )
        evidence_blocks = self._build_evidence_blocks(ref_list, graph_refs)
        evaluation_blocks = []
        verdict_priority = VERDICT_PRIORITY

        for ref_offset, evidence in enumerate(evidence_blocks):
            bundle = (
                ref_snippets[ref_offset]
                if ref_offset < len(ref_snippets)
                else {}
            )
            if not isinstance(bundle, dict):
                bundle = {"text": str(bundle or ""), "snippets": []}

            snippet_items = [
                dict(item)
                for item in bundle.get("snippets", [])
                if isinstance(item, dict) and str(item.get("text") or "").strip()
            ]
            if not snippet_items and str(bundle.get("text") or "").strip():
                snippet_items = [{
                    "text": str(bundle.get("text") or "").strip(),
                    "score": bundle.get("score"),
                    "edge_matches": [],
                }]

            all_matches = []
            snippet_statuses = []
            for item in snippet_items:
                matches = [
                    dict(match)
                    for match in item.get("edge_matches", [])
                    if isinstance(match, dict)
                ]
                all_matches.extend(matches)
                direct = any(
                    match.get("verdict") == "direct" for match in matches
                )
                if matches:
                    status = max(
                        (str(match.get("verdict") or "unsupported") for match in matches),
                        key=lambda value: verdict_priority.get(value, 0),
                    )
                else:
                    status = "unclassified"
                snippet_statuses.append(status)
                evaluation_blocks.append({
                    "reference_index": ref_offset + 1,
                    "source": evidence.get("source") or (
                        ref_list[ref_offset]
                        if ref_offset < len(ref_list)
                        else ""
                    ),
                    "snippet": str(item.get("text") or "").strip(),
                    "score": item.get("score"),
                    "retrieval_origin": item.get("retrieval_origin", "unknown"),
                    "edge_matches": matches,
                    "evidence_status": "direct" if direct else status,
                })

            has_direct = any(
                match.get("verdict") == "direct" for match in all_matches
            )
            has_non_direct = any(
                status not in {"direct", "unclassified"}
                for status in snippet_statuses
            )
            if has_direct and has_non_direct:
                aggregate_status = "mixed"
            elif has_direct:
                aggregate_status = "direct"
            elif snippet_statuses:
                aggregate_status = max(
                    snippet_statuses,
                    key=lambda value: verdict_priority.get(value, 0),
                )
            elif str(bundle.get("text") or "").strip():
                aggregate_status = "unclassified"
            else:
                aggregate_status = "metadata_only"

            evidence.update({
                "snippet": str(bundle.get("text") or "").strip(),
                "score": bundle.get("score"),
                "snippets": snippet_items,
                "provenance_only": bool(bundle.get("provenance_only")),
                "best_verdict": bundle.get("best_verdict"),
                "edge_matches": all_matches,
                "evidence_status": aggregate_status,
                "has_direct_evidence": has_direct,
            })

        all_edge_matches = [
            match
            for block in evaluation_blocks
            for match in block.get("edge_matches", [])
        ]
        best_by_edge = {}
        for match in all_edge_matches:
            edge_index = match.get("edge_index")
            if (
                str(match.get("evidence_scope") or "edge") != "edge"
                or not isinstance(edge_index, int)
                or edge_index < 1
            ):
                continue
            current = best_by_edge.get(edge_index)
            if current is None or verdict_priority.get(
                match.get("verdict"), 0
            ) > verdict_priority.get(current.get("verdict"), 0):
                best_by_edge[edge_index] = match

        edge_count = len(parse_path(path_str))
        direct_edges = sum(
            match.get("verdict") == "direct"
            for match in best_by_edge.values()
        )
        pathway_endpoint_matches = [
            match
            for match in all_edge_matches
            if (
                match.get("verdict") == "direct"
                and str(match.get("evidence_scope") or "") == "pathway_endpoint"
            )
        ]
        # For a one-edge path the directly supported edge is itself the
        # start-to-final conversion. Multi-edge paths remain stricter: endpoint
        # evidence never implies support for unmentioned intermediate edges.
        single_edge_endpoint_supported = edge_count == 1 and direct_edges == 1
        endpoint_conversion_supported = bool(
            pathway_endpoint_matches or single_edge_endpoint_supported
        )
        pathway_endpoint_snippet_count = sum(
            any(
                match.get("verdict") == "direct"
                and str(match.get("evidence_scope") or "") == "pathway_endpoint"
                for match in block.get("edge_matches", [])
            )
            for block in evaluation_blocks
        )
        if single_edge_endpoint_supported:
            pathway_endpoint_snippet_count = sum(
                any(
                    match.get("verdict") == "direct"
                    and str(match.get("evidence_scope") or "edge") == "edge"
                    for match in block.get("edge_matches", [])
                )
                for block in evaluation_blocks
            )
        metrics = {
            "ref_count": len(ref_list),
            "retrieval_available": bool(
                getattr(self, "vector_db", None)
                or getattr(self, "chroma_collection", None)
            ),
            "retrieval_mode": getattr(self, "retrieval_mode", "none"),
            "retrieval_error": getattr(self, "retrieval_error", ""),
            "retrieval_trace": dict(retrieval_trace or {}),
            "retrieval_strategy": str(
                (retrieval_trace or {}).get("strategy") or "dense vector retrieval"
            ),
            "edge_evidence_coverage": (
                round(direct_edges / edge_count, 3) if edge_count else 0.0
            ),
            "edge_evidence": [
                best_by_edge.get(
                    index,
                    {"edge_index": index, "verdict": "unsupported"},
                )
                for index in range(1, edge_count + 1)
            ],
            "endpoint_conversion_supported": endpoint_conversion_supported,
            "pathway_endpoint_evidence_coverage": (
                1.0 if endpoint_conversion_supported else 0.0
            ),
            "pathway_endpoint_snippet_count": pathway_endpoint_snippet_count,
            "related_context_count": len(related_context or []),
        }

        if endpoint_conversion_supported and direct_edges < edge_count:
            scope_note = (
                "Validation note: the cited pathway-level evidence supports "
                "the overall start-to-final conversion only; "
                f"{edge_count - direct_edges} of {edge_count} individual graph "
                "edges remain graph-derived and unverified."
            )
            if scope_note.casefold() not in report.casefold():
                if latex_block and report.rstrip().endswith(latex_block):
                    prefix = report.rstrip()[:-len(latex_block)].rstrip()
                    report = f"{prefix}\n\n{scope_note}\n\n{latex_block}"
                else:
                    report = f"{report.rstrip()}\n\n{scope_note}"

        rag_evaluation = None
        if evaluate_rag_report:
            try:
                rag_evaluation = evaluate_rag_report(
                    report=report,
                    retrieved_snippets=evaluation_blocks,
                    pathway_graph={
                        "path_str": path_str,
                        "path_terms": path_terms,
                        "graph_refs": graph_refs,
                    },
                    references=ref_list,
                )
            except Exception as exc:
                logger.warning(
                    "RAG report evaluation failed: %s",
                    exc,
                    exc_info=True,
                )

        if isinstance(rag_evaluation, dict) and not evidence_is_weak:
            evaluated_rows = [
                item
                for item in rag_evaluation.get("sentence_level_results", [])
                if isinstance(item, dict)
            ]
            # Keep two kinds of sentence, both requiring a verified citation:
            #   * edge-level direct evidence ("A converts to B"), and
            #   * a close restatement of a cited snippet (mechanism, conditions,
            #     instrumentation, observed phases).
            # Without the second kind the Introduction collapsed to a single
            # synthesized line, discarding detail that came straight from the
            # retrieved literature shown to the reader below it.
            accepted_statements = [
                str(item.get("statement") or "").strip()
                for item in evaluated_rows
                if (
                    item.get("direct_evidence_status")
                    or item.get("restated_evidence_status")
                )
                and item.get("citation_status") is True
            ]
            rejected_count = max(
                0,
                len(evaluated_rows) - len(accepted_statements),
            )
            if not accepted_statements:
                for block in evaluation_blocks:
                    direct_match = next((
                        match
                        for match in block.get("edge_matches", [])
                        if isinstance(match, dict)
                        and match.get("verdict") == "direct"
                    ), None)
                    if not direct_match:
                        continue
                    ref_number = int(block.get("reference_index") or 1)
                    accepted_statements.append(
                        _compose_direct_evidence_intro(direct_match, ref_number)
                    )
                    break
            if rejected_count:
                if edge_count > direct_edges:
                    endpoint_note = (
                        "The overall start-to-final conversion has direct "
                        "pathway-level evidence; "
                        if endpoint_conversion_supported
                        else ""
                    )
                    # Name the steps rather than counting them. "1 of 2 edges"
                    # makes the reader work out which half of the pathway is
                    # unsupported, and the answer is only inferable from the
                    # "edge 1: direct" labels further down the report.
                    authored = self._authored_step_types()
                    unestablished, extrapolated, ionic = [], [], []
                    for edge in parse_path(path_str):
                        if (best_by_edge.get(edge.index) or {}).get("verdict") == "direct":
                            continue
                        name = f"{edge.reactant} -> {edge.product}"
                        key = (edge.reactant, edge.condition, edge.product)
                        kind = authored.get(key)
                        if kind == "extended":
                            extrapolated.append(name)
                        elif kind == "ionic_form":
                            ionic.append(name)
                        else:
                            unestablished.append(name)

                    def _phrase(names, verb_single, verb_plural):
                        if len(names) == 1:
                            return f"the step {names[0]} {verb_single}"
                        return (
                            "the steps "
                            + ", ".join(names[:-1])
                            + f" and {names[-1]} {verb_plural}"
                        )

                    parts = []
                    if unestablished:
                        parts.append(
                            f"{_phrase(unestablished, 'is', 'are')} not established "
                            "by direct literature snippets; generated claims for "
                            "those gaps were omitted"
                        )
                    # An author-marked extrapolation is a different statement
                    # from a failed search, and reporting them alike understates
                    # one and overstates the other.
                    if extrapolated:
                        parts.append(
                            f"{_phrase(extrapolated, 'is', 'are')} marked by the "
                            "network author as a chemical extrapolation rather "
                            "than a literature-stated conversion"
                        )
                    # The source does state this conversion; it writes the
                    # species as an aqueous ion where the network writes the
                    # salt, and the two have different compositions, so entity
                    # matching cannot join them. Reporting it as unestablished
                    # would understate evidence the paper actually gives.
                    if ionic:
                        parts.append(
                            f"{_phrase(ionic, 'is', 'are')} stated in the cited "
                            "source as an ionic half-reaction, which this "
                            "network represents as the corresponding salt"
                        )
                    if not parts:
                        parts.append(
                            f"{edge_count - direct_edges} of {edge_count} individual "
                            "graph edges are not established by direct literature "
                            "snippets"
                        )
                    validation_note = (
                        f"Validation note: {endpoint_note}" + "; ".join(parts) + "."
                    )
                else:
                    # Every edge is supported and only individual sentences were
                    # dropped. Saying so on every report states the gating
                    # policy, not a result of this analysis, so it is left to
                    # the exported evaluation detail. A note appears here only
                    # when part of the pathway is genuinely unestablished.
                    validation_note = ""
                guarded_body = " ".join(
                    statement
                    for statement in accepted_statements + [validation_note]
                    if statement
                ).strip()
                report = self._normalize_inline_formulas(
                    f"{header}\n\n{guarded_body}\n\n{latex_block}",
                    path_str,
                )
                try:
                    rag_evaluation = evaluate_rag_report(
                        report=report,
                        retrieved_snippets=evaluation_blocks,
                        pathway_graph={
                            "path_str": path_str,
                            "path_terms": path_terms,
                            "graph_refs": graph_refs,
                        },
                        references=ref_list,
                    )
                    rag_evaluation[
                        "generation_guard_removed_statements"
                    ] = rejected_count
                except Exception as exc:
                    logger.warning(
                        "Guarded RAG report evaluation failed: %s",
                        exc,
                        exc_info=True,
                    )

        sentence_traces = []
        if isinstance(rag_evaluation, dict):
            if evidence_is_weak or direct_edges < edge_count:
                rag_evaluation["manual_verification"] = True
                rag_evaluation["manual_verification_required"] = True
                rag_evaluation["manual_verification_status"] = "required"
                if direct_edges < edge_count:
                    gap_reason = (
                        f"{edge_count - direct_edges} of {edge_count} graph "
                        "edges lack direct snippet evidence."
                    )
                    if endpoint_conversion_supported:
                        gap_reason += (
                            " Endpoint conversion evidence does not verify the "
                            "intermediate graph steps."
                        )
                    current_reason = str(
                        rag_evaluation.get("manual_verification_reason") or ""
                    ).strip()
                    if gap_reason not in current_reason:
                        rag_evaluation["manual_verification_reason"] = (
                            f"{current_reason} {gap_reason}".strip()
                        )
            metrics.update({
                "total_sentences": rag_evaluation.get("total_statements", 0),
                "supported_sentences": rag_evaluation.get(
                    "supported_statements", 0
                ),
                "total_statements": rag_evaluation.get("total_statements", 0),
                "supported_statements": rag_evaluation.get(
                    "supported_statements", 0
                ),
                "direct_evidence_statements": rag_evaluation.get(
                    "direct_evidence_statements", 0
                ),
                "restated_evidence_statements": rag_evaluation.get(
                    "restated_evidence_statements", 0
                ),
                "grounded_statements": rag_evaluation.get(
                    "grounded_statements", 0
                ),
                "unsupported_statements": rag_evaluation.get(
                    "unsupported_statements", 0
                ),
                "statements_without_direct_evidence": rag_evaluation.get(
                    "statements_without_direct_evidence", 0
                ),
                "unsupported_by_both_statements": rag_evaluation.get(
                    "unsupported_by_both_statements", 0
                ),
                "any_supported_statements": rag_evaluation.get(
                    "any_supported_statements", 0
                ),
                "evidence_coverage": rag_evaluation.get(
                    "evidence_coverage", 0.0
                ),
                "direct_evidence_coverage": rag_evaluation.get(
                    "direct_evidence_coverage", 0.0
                ),
                "statement_evidence_coverage": rag_evaluation.get(
                    "statement_evidence_coverage",
                    rag_evaluation.get("evidence_coverage", 0.0),
                ),
                "citation_accuracy": rag_evaluation.get("citation_accuracy"),
                "citation_verification_status": rag_evaluation.get(
                    "citation_verification_status", "not_applicable"
                ),
                "citation_checked_statements": rag_evaluation.get(
                    "citation_checked_statements", 0
                ),
                "citation_unverifiable_statements": rag_evaluation.get(
                    "citation_unverifiable_statements", 0
                ),
                "graph_alignment": rag_evaluation.get(
                    "graph_alignment", 0.0
                ),
                "graph_alignment_applicable_statements": rag_evaluation.get(
                    "graph_alignment_applicable_statements", 0
                ),
                "graph_alignment_status": rag_evaluation.get(
                    "graph_alignment_status", "not_applicable"
                ),
                "graph_derived_statements": rag_evaluation.get(
                    "graph_derived_statements", 0
                ),
                "inference_only_statements": rag_evaluation.get(
                    "inference_only_statements", 0
                ),
                "retrieved_snippet_count": rag_evaluation.get(
                    "retrieved_snippet_count", 0
                ),
                "source_quality_score": rag_evaluation.get(
                    "source_quality_score", 0.0
                ),
                "source_quality_level": rag_evaluation.get(
                    "source_quality_level", "unavailable"
                ),
                "source_quality_details": rag_evaluation.get(
                    "source_quality_details", []
                ),
                "source_quality_basis": rag_evaluation.get(
                    "source_quality_basis", ""
                ),
                "overall_reliability_score": rag_evaluation.get(
                    "overall_reliability_score", 0.0
                ),
                "overall_reliability_level": rag_evaluation.get(
                    "overall_reliability_level", "not_evaluable"
                ),
                "overall_reliability_summary": rag_evaluation.get(
                    "overall_reliability_summary", ""
                ),
                "overall_reliability_reasons": rag_evaluation.get(
                    "overall_reliability_reasons", []
                ),
                "manual_verification": rag_evaluation.get(
                    "manual_verification"
                ),
                "manual_verification_reason": rag_evaluation.get(
                    "manual_verification_reason"
                ),
                "generation_guard_removed_statements": rag_evaluation.get(
                    "generation_guard_removed_statements",
                    0,
                ),
            })
            for item in rag_evaluation.get("sentence_level_results", []):
                if not isinstance(item, dict):
                    continue
                supporting = item.get("supporting_evidence")
                sentence_traces.append({
                    "sentence_id": item.get("sentence_id"),
                    "text": item.get("statement") or "",
                    "support": (
                        "evidence"
                        if item.get("direct_evidence_status")
                        else "inference"
                    ),
                    "relevance": (
                        supporting.get("match_score")
                        if isinstance(supporting, dict)
                        else None
                    ),
                    "evidence": [supporting] if supporting else [],
                    "matched_evidence": supporting,
                    "direct_evidence_status": item.get(
                        "direct_evidence_status"
                    ),
                    "restated_evidence_status": item.get(
                        "restated_evidence_status"
                    ),
                    "grounded_evidence_status": item.get(
                        "grounded_evidence_status"
                    ),
                    "support_status": item.get("support_status"),
                    "support_basis": item.get("support_basis"),
                    "citation_status": item.get("citation_status"),
                    "graph_alignment": item.get("graph_alignment"),
                    "graph_alignment_applicable": item.get(
                        "graph_alignment_applicable"
                    ),
                    "graph_alignment_status": item.get(
                        "graph_alignment_status"
                    ),
                    "statement_type": item.get("statement_type"),
                    "cited_references": item.get("cited_references", []),
                    "cited_dois": item.get("cited_dois", []),
                })

        if evidence_is_weak:
            metrics["evidence_is_weak"] = True
            metrics["manual_verification"] = True
        context_sources = ()
        if contextual_narrative:
            # Marks report bodies that are a grounded synthesis of retrieved
            # literature (framed as related context), so the display layer may
            # keep them instead of substituting the terse "no evidence" text.
            metrics["has_contextual_narrative"] = True
            # The [n] the narrative cites are only resolvable if the numbering
            # travels with the report. Kept separate from "references", which
            # means "sources that establish an edge" and must stay empty here.
            context_sources = tuple(getattr(self, "_contextual_sources", ()) or ())

        return {
            "report": report,
            "root_material": root_material,
            "path_str": path_str,
            "references": ref_list,
            "evidence": evidence_blocks,
            "evaluation_evidence": evaluation_blocks,
            "reference_snippets": ref_snippets,
            "graph_refs": graph_refs,
            "context_sources": list(context_sources),
            "metrics": metrics,
            "sentence_traces": sentence_traces,
            "rag_evaluation": rag_evaluation,
            "related_context": list(related_context or []),
            "rag_debug": [
                f"retrieval_mode: {self.retrieval_mode}",
                (
                    "collection_count: "
                    f"{self.collection_count if self.collection_count is not None else 'n/a'}"
                ),
                f"ref_snippet_count: {len(evaluation_blocks)}",
                f"deepseek_base_url: {DEEPSEEK_BASE_URL}",
                f"deepseek_model: {DEEPSEEK_MODEL}",
                (
                    "deepseek_api_key_set: "
                    f"{'yes' if DEEPSEEK_API_KEY else 'no'}"
                ),
                f"evidence_is_weak: {'yes' if evidence_is_weak else 'no'}",
            ],
        }

    @staticmethod
    def _dense_depth_config():
        """Resolve the dense-retrieval depths (ANN_FETCH_K, RAG_OUTPUT_TOP_K).

        RAG_OUTPUT_TOP_K is how many distance-sorted candidates per query survive
        into the threshold filter and the cross-query union. RAG_ANN_FETCH_K is
        the HNSW over-fetch depth: Chroma's effective search_ef = max(index
        search_ef, n_results), and the persisted index ships search_ef=10, so we
        over-fetch to raise the effective ef before truncating. Depth 200 is the
        pre-registered ANN-calibration choice (eval_spec.md 20). ANN_FETCH_K is
        never allowed below RAG_OUTPUT_TOP_K.
        """
        top_k = int(os.getenv("RAG_TOP_K", "6"))
        output_top_k = max(top_k, int(os.getenv("RAG_FALLBACK_TOP_K", "20")))
        ann_fetch_k = max(output_top_k, int(os.getenv("RAG_ANN_FETCH_K", "200")))
        return ann_fetch_k, output_top_k

    @staticmethod
    def _limit_dense_candidates(candidates, output_top_k):
        """Keep only the nearest ``output_top_k`` over-fetched ANN candidates.

        Candidates arrive distance-sorted (ascending) from the HNSW query; this
        truncation is what enforces the per-query cap so at most RAG_OUTPUT_TOP_K
        items reach the threshold filter and the union, regardless of how deep we
        over-fetched to raise the effective search_ef.
        """
        return list(candidates)[:max(0, int(output_top_k))]

    def _report_cache_dir(self):
        return os.getenv("RAG_REPORT_CACHE_DIR") or os.path.join(
            os.path.dirname(os.path.abspath(__file__)), "rag_report_cache"
        )

    def _report_cache_signature(self):
        """Corpus fingerprint, so a rebuilt knowledge base invalidates the cache.

        A cached report is only valid for the corpus it was computed from. The
        document count changes whenever papers are added or removed, which is
        exactly when the stored evidence stops being reproducible.

        Note: this sees the corpus, not the gating rules. If retrieval or
        evidence gating is ever changed, the cache must be cleared by hand --
        the stored evidence was computed under the previous rules.
        """
        count = self.collection_count
        if count is None:
            try:
                collection = getattr(self.vector_db, "_collection", None) or self.chroma_collection
                count = collection.count() if collection is not None else -1
            except Exception:
                count = -1
        return f"docs={count}"

    def _report_cache_key(self, root_material, path_str):
        import hashlib

        payload = "|".join([
            str(root_material or ""),
            str(path_str or ""),
            self._report_cache_signature(),
        ])
        return hashlib.sha1(payload.encode("utf-8")).hexdigest()

    def generate_report_for_path(self, root_material, path_str, full_path=None):
        """Return the pathway report, reusing cached retrieval evidence.

        Retrieval and evidence gating are deterministic in (path, corpus) yet cost
        minutes of CPU, so recomputing them per request spends that time
        reproducing an identical answer; they are cached on disk, keyed by path
        and corpus fingerprint, so a changed knowledge base recomputes rather
        than serving stale evidence. The narrative is deliberately NOT cached --
        it is regenerated on every call so the wording is never frozen, which
        costs seconds rather than minutes. Set RAG_REPORT_CACHE=0 to bypass.
        """
        cache_enabled = os.getenv("RAG_REPORT_CACHE", "1").strip().lower() not in (
            "0", "false", "no"
        )
        evidence = self.cached_path_evidence(root_material, path_str, full_path)
        return self._narrate_path_report(
            evidence, root_material, path_str, full_path
        )

    def cached_path_evidence(self, root_material, path_str, full_path=None):
        """Retrieval evidence for one path, computed once and cached on disk.

        Exposed separately so the offline warm-up can fill the cache without
        generating a narrative: narration is a per-request concern and, with an
        API key configured, warming 88 paths through the full report path would
        spend 88 LLM calls on text that is thrown away.
        """
        cache_enabled = os.getenv("RAG_REPORT_CACHE", "1").strip().lower() not in (
            "0", "false", "no"
        )
        cache_file = None
        evidence = None
        if cache_enabled:
            cache_file = os.path.join(
                self._report_cache_dir(),
                f"{self._report_cache_key(root_material, path_str)}.json",
            )
            try:
                if os.path.isfile(cache_file):
                    with open(cache_file, "r", encoding="utf-8") as handle:
                        cached = json.load(handle)
                    stored = cached.get("evidence")
                    # Every field the narrative half reads must be present, or a
                    # cache written by an older layout would fail mid-report.
                    if isinstance(stored, dict) and all(
                        key in stored
                        for key in (
                            "context_str", "ref_list", "ref_snippets",
                            "related_context", "retrieval_trace",
                        )
                    ):
                        evidence = stored
                        logger.info(
                            "[RAG][CACHE] evidence hit path=%s", str(path_str)[:120]
                        )
            except (OSError, ValueError):
                logger.warning(
                    "[RAG][CACHE] unreadable entry ignored: %s", cache_file
                )

        if evidence is None:
            evidence = self._retrieve_path_evidence(
                root_material, path_str, full_path
            )
            if cache_enabled and cache_file:
                try:
                    os.makedirs(os.path.dirname(cache_file), exist_ok=True)
                    # Write to a temporary file first so a crash mid-write cannot
                    # leave a truncated entry that later reads as valid evidence.
                    temporary = cache_file + ".tmp"
                    with open(temporary, "w", encoding="utf-8") as handle:
                        json.dump(
                            {
                                "root_material": root_material,
                                "path_str": path_str,
                                "signature": self._report_cache_signature(),
                                "evidence": evidence,
                            },
                            handle,
                            ensure_ascii=False,
                            default=str,
                        )
                    os.replace(temporary, cache_file)
                    logger.info(
                        "[RAG][CACHE] stored evidence path=%s", str(path_str)[:120]
                    )
                except (OSError, TypeError, ValueError):
                    logger.warning("[RAG][CACHE] could not store %s", cache_file)

        return evidence

    def _retrieve_path_evidence(self, root_material, path_str, full_path=None):
        edges = parse_path(path_str)
        mini_report = {root_material: [path_str]}
        generated_queries = self.query_generator.generate_queries(mini_report)
        endpoint_query = ""
        elemental_endpoint_query = ""
        if len(edges) >= 2:
            endpoint_conditions = list(dict.fromkeys(
                edge.condition
                for edge in edges
                if str(edge.condition or "").strip()
            ))
            endpoint_query = " ".join(
                [edges[0].reactant]
                + endpoint_conditions
                + [edges[-1].product]
            )
        if edges:
            endpoint_composition = parse_formula(edges[-1].product)
            if (
                endpoint_composition
                and len(endpoint_composition) == 1
                and endpoint_composition[0][1] == 1
                and canonical_formula(edges[-1].product) == endpoint_composition[0][0]
            ):
                conditions = list(dict.fromkeys(
                    edge.condition for edge in edges if edge.condition
                ))
                elemental_endpoint_query = " ".join(
                    [edges[0].reactant]
                    + conditions
                    + [
                        "metallic", edges[-1].product,
                        "elemental", edges[-1].product,
                        "zero-valent",
                    ]
                )
        queries = list(dict.fromkeys(
            [edge.query for edge in edges if edge.query]
            + ([endpoint_query] if endpoint_query else [])
            + ([elemental_endpoint_query] if elemental_endpoint_query else [])
            + [query for query in generated_queries if str(query or "").strip()]
        ))

        ref_list = []
        ref_snippets = []
        reference_index = {}
        seen_snippets = set()
        related_context = []
        seen_related = set()
        retrieval_trace = {
            "query_count": len(queries),
            "dense_candidates": 0,
            "dense_kept": 0,
            "resolver_scanned": 0,
            "resolver_candidates": 0,
            "resolver_kept": 0,
            "bm25_scanned": 0,
            "bm25_candidates": 0,
            "bm25_kept": 0,
            "lexical_scanned": 0,
            "lexical_candidates": 0,
            "lexical_kept": 0,
            "strategy": (
                "four-route retrieval (resolver + BM25 + lexical + BGE-M3) "
                "with chemical evidence gating"
            ),
        }
        provenance_only = []
        # Best verdict any passage of a source reached, including passages the
        # strict gate rejected. A source that matched at "related" is not the
        # same as one nothing matched in, and the report must not say it is.
        provenance_best_verdict: dict[str, str] = {}
        path_sources, path_dois = self._get_pathway_provenance_hints(
            root_material,
            path_str,
        )
        provenance_records = self._provenance_metadata_records(
            path_sources,
            path_dois,
        )

        score_threshold = float(os.getenv("RAG_SCORE_THRESHOLD", "2.0"))
        ann_fetch_k, output_top_k = self._dense_depth_config()

        normalized_path_dois = {
            normalize_doi(value) for value in path_dois if normalize_doi(value)
        }
        provenance_title_keys = {
            self._reference_title_key(record.get("title"))
            for record in provenance_records
            if self._reference_title_key(record.get("title"))
        }

        def _is_path_provenance_source(source, metadata):
            values = metadata or {}
            doi = normalize_doi(
                values.get("doi") or values.get("DOI") or values.get("doi_id")
                or first_doi(source)
            )
            if doi and doi in normalized_path_dois:
                return True
            title_candidates = [
                values.get("title"), values.get("source_file"), source,
            ]
            return any(
                self._reference_title_key(value) in provenance_title_keys
                for value in title_candidates
                if self._reference_title_key(value)
            )

        def _match_dict(match, evidence_scope="edge"):
            return {
                "edge_index": match.edge.index,
                "reactant": match.edge.reactant,
                "product": match.edge.product,
                "condition": match.edge.condition,
                "evidence_scope": evidence_scope,
                "verdict": match.verdict,
                "score": match.score,
                "window": match.window,
                "reactant_match": match.reactant.match_type,
                "product_match": match.product.match_type,
                "reactant_span": match.reactant.span,
                "product_span": match.product.span,
                "relation_basis": match.relation_basis,
                "condition_status": match.condition_status,
                "provenance_level": match.provenance_level,
                "reasons": list(match.reasons),
            }

        def _record_best_verdict(source, edge_matches, endpoint_match):
            """Remember the strongest verdict a source reached, kept or rejected.

            The strict gate publishes only "direct" matches, so without this the
            report cannot tell a source whose passages matched the species but
            not the recorded condition from one that matched nothing at all.
            """
            key = str(source or "").strip().casefold()
            if not key:
                return
            candidates = list(edge_matches or [])
            if endpoint_match is not None:
                candidates.append(endpoint_match)
            if not candidates:
                return
            best = max(
                (str(match.verdict or "unsupported") for match in candidates),
                key=lambda verdict: VERDICT_PRIORITY.get(verdict, 0),
            )
            current = provenance_best_verdict.get(key)
            if current is None or VERDICT_PRIORITY.get(
                best, 0
            ) > VERDICT_PRIORITY.get(current, 0):
                provenance_best_verdict[key] = best

        def _add_evidence(
            source,
            content,
            score=None,
            metadata=None,
            provenance_level="unspecified",
            diagnose=False,
            retrieval_origin="dense_vector",
        ):
            raw_content = str(content or "")
            composite_used = False
            # Preserve line breaks while matching so Markdown section headings
            # can be removed by split_evidence_windows. Collapse only the
            # final published snippet.
            matching_content = re.sub(r"[ \t]+", " ", raw_content).strip()
            clean = re.sub(r"\s+", " ", raw_content).strip()[:1600]
            if not clean:
                return False
            low_quality_related_context = bool(
                re.search(r"\[MISSING_PAGE_FAIL", clean, flags=re.IGNORECASE)
                or re.search(r"\\begin\{(?:table|tabular)\}", clean, flags=re.IGNORECASE)
            )
            if _is_bibliography_text(clean):
                return False
            edge_matches = self._path_evidence_matches(
                matching_content,
                path_str,
                provenance_level=provenance_level,
                doi=(metadata or {}).get("doi")
                or (metadata or {}).get("DOI")
                or (metadata or {}).get("doi_id"),
            )
            relevant = [
                (match, "edge")
                for match in edge_matches
                if match.verdict == "direct"
            ]
            endpoint_match = self._path_endpoint_evidence_match(
                matching_content,
                path_str,
                provenance_level=provenance_level,
            )
            if endpoint_match is not None and endpoint_match.verdict == "direct":
                relevant.append((endpoint_match, "pathway_endpoint"))
            if not relevant:
                values = metadata or {}
                doi_value = (
                    values.get("doi") or values.get("DOI")
                    or values.get("doi_id") or first_doi(source)
                )
                composite_content = self._adjacent_document_text(
                    doi_value,
                    values.get("chunk_index"),
                    raw_content,
                )
                if composite_content.strip() and composite_content != raw_content:
                    composite_matches = self._path_evidence_matches(
                        composite_content,
                        path_str,
                        provenance_level=provenance_level,
                        doi=doi_value,
                    )
                    composite_endpoint = self._path_endpoint_evidence_match(
                        composite_content,
                        path_str,
                        provenance_level=provenance_level,
                    )
                    composite_relevant = [
                        (match, "edge")
                        for match in composite_matches
                        if match.verdict == "direct"
                    ]
                    if (
                        composite_endpoint is not None
                        and composite_endpoint.verdict == "direct"
                    ):
                        composite_relevant.append(
                            (composite_endpoint, "pathway_endpoint")
                        )
                    _record_best_verdict(
                        source, composite_matches, composite_endpoint
                    )
                    if composite_relevant:
                        matching_content = composite_content
                        clean = re.sub(
                            r"\s+", " ", composite_content
                        ).strip()[:1600]
                        edge_matches = composite_matches
                        endpoint_match = composite_endpoint
                        relevant = composite_relevant
                        composite_used = True
            _record_best_verdict(source, edge_matches, endpoint_match)
            if not relevant:
                # Preserve a small, explicitly non-evidential audit trail for
                # chemically related candidates rejected by the strict gate.
                candidates = list(edge_matches)
                if endpoint_match is not None:
                    candidates.append(endpoint_match)
                candidates.sort(key=lambda item: item.score, reverse=True)
                best = candidates[0] if candidates else None
                provenance_source = _is_path_provenance_source(
                    source,
                    metadata,
                )
                materially_related = bool(
                    best
                    # Endpoint-only matches from another starting material are
                    # out of scope. Preserve a missing-entity paragraph only
                    # when it comes from this path's own provenance source.
                    and (best.reactant.matched or provenance_source)
                    and best.relation_basis != "cooccurrence"
                )
                related_key = (
                    str(source or "").strip().casefold(),
                    str(best.window if best else clean).strip().casefold(),
                )
                if (
                    materially_related
                    and not low_quality_related_context
                    and related_key not in seen_related
                ):
                    seen_related.add(related_key)
                    related_context.append({
                        "source": str(source or "").strip()
                        or self._format_reference_source(metadata or {}),
                        # Related context is not sent to claim generation, so
                        # retain the surrounding paragraph for human review.
                        "snippet": clean,
                        "retrieval_origin": retrieval_origin,
                        "match": _match_dict(best),
                        "reason": (
                            "Retrieved as chemically related context, but it "
                            "does not directly establish this graph edge."
                        ),
                    })
                    related_context.sort(
                        key=lambda item: float(
                            (item.get("match") or {}).get("score") or 0
                        ),
                        reverse=True,
                    )
                    del related_context[5:]
                if diagnose:
                    decisions = [
                        {
                            "edge": match.edge.index,
                            "verdict": match.verdict,
                            "reactant": match.reactant.match_type,
                            "product": match.product.match_type,
                            "relation": match.relation_basis,
                            "condition": match.condition_status,
                        }
                        for match in edge_matches
                    ]
                    logger.info(
                        "[RAG][FILTER] source=%s rejected=%s",
                        str(source or "Unknown")[:180],
                        decisions,
                    )
                return False
            source = str(source or "").strip() or self._format_reference_source(metadata or {})
            if not source or source == "Unknown":
                return False

            # Publish only the sentence window that actually matched. Keeping
            # the whole retrieval chunk would leak unrelated nearby examples
            # into the prompt, report, and PDF.
            window_matches = {}
            for match, evidence_scope in relevant:
                window = re.sub(r"\s+", " ", str(match.window or "")).strip()[:1600]
                if not window:
                    continue
                window_matches.setdefault(window, []).append(
                    _match_dict(match, evidence_scope=evidence_scope)
                )
            if not window_matches:
                return False

            if source not in reference_index:
                reference_index[source] = len(ref_list)
                ref_list.append(source)
                ref_snippets.append({"text": "", "score": score, "snippets": []})
            ref_index = reference_index[source]
            bundle = ref_snippets[ref_index]
            added = False
            for window, match_records in window_matches.items():
                snippet_key = (source.casefold(), window.casefold())
                if snippet_key in seen_snippets:
                    # The same evidence window may be recalled by several
                    # routes. Preserve all contributing routes instead of
                    # silently attributing it only to the first route.
                    for existing_bundle in ref_snippets:
                        for existing in existing_bundle.get("snippets", []):
                            existing_key = (
                                str(source).casefold(),
                                str(existing.get("text") or "").casefold(),
                            )
                            if existing_key == snippet_key:
                                routes = existing.setdefault(
                                    "retrieval_routes",
                                    [existing.get("retrieval_origin", "unknown")],
                                )
                                if retrieval_origin not in routes:
                                    routes.append(retrieval_origin)
                                break
                    continue
                seen_snippets.add(snippet_key)
                evidence_level = (
                    "edge_direct"
                    if any(item.get("evidence_scope") == "edge" for item in match_records)
                    else "pathway_endpoint"
                )
                bundle["snippets"].append({
                    "text": window,
                    "score": score,
                    "metadata": dict(metadata or {}),
                    "evidence_level": evidence_level,
                    "retrieval_origin": retrieval_origin,
                    "retrieval_routes": [retrieval_origin],
                    "evidence_aggregation": (
                        "adjacent_chunks_same_doi"
                        if composite_used else "single_fragment"
                    ),
                    "edge_matches": match_records,
                })
                added = True
            if not added:
                if not bundle.get("snippets"):
                    ref_list.pop()
                    ref_snippets.pop()
                    reference_index.pop(source, None)
                return False
            bundle["text"] = "\n".join(
                snippet["text"] for snippet in bundle["snippets"]
            )
            numeric_scores = [
                snippet.get("score") for snippet in bundle["snippets"]
                if isinstance(snippet.get("score"), (int, float))
            ]
            bundle["score"] = min(numeric_scores) if numeric_scores else None
            return True

        for source in path_sources:
            doi = first_doi(source)
            if doi:
                continue
            provenance_only.append(f"CRN source: {source}")

        for doi in path_dois:
            normalized_doi = normalize_doi(doi)
            provenance_only.append(self._format_doi_reference(normalized_doi))

        # Chroma keeps vectors in a separate on-disk HNSW segment. When that
        # segment is missing or stale every similarity search returns nothing,
        # and embedding each of ~25 path queries just to discard the result
        # costs seconds per path. Detect it once and fall straight through to
        # the lexical evidence scan.
        dense_probe_queries = max(1, int(os.getenv("RAG_DENSE_PROBE_QUERIES", "3")))
        dense_disabled = False
        dense_attempts = 0

        # Embed every path query once, batched, instead of embedding each variant
        # separately inside the loop below -- on CPU that per-query embedding is
        # the dominant cost of dense retrieval. The batched vectors are identical
        # to per-query embedding, so results are unchanged. Querying the Chroma
        # collection directly by vector then reuses these instead of re-embedding.
        query_vectors = self._embed_queries_batch(queries)
        dense_collection = (
            getattr(self.vector_db, "_collection", None) if self.vector_db else None
        )

        for query in queries:
            if not self.vector_db and not self.chroma_collection:
                break
            vector_ready = self.vector_db and not dense_disabled and (
                self.collection_count is None or self.collection_count > 0
            )
            chroma_ready = (
                self.chroma_collection and self.embedder and not dense_disabled and (
                    self.collection_count is None or self.collection_count > 0
                )
            )
            raw_scores = []
            before = len(seen_snippets)
            if vector_ready:
                qv = query_vectors.get(query)
                if qv is not None and dense_collection is not None:
                    # Reuse the pre-batched vector: query the collection directly
                    # instead of letting similarity_search_with_score embed the
                    # query again. Distances are the same cosine distances that
                    # method returns, so downstream scoring is unchanged.
                    raw = dense_collection.query(
                        query_embeddings=[qv],
                        n_results=ann_fetch_k,
                        include=["documents", "metadatas", "distances"],
                    )
                    r_docs = (raw.get("documents") or [[]])[0]
                    r_metas = (raw.get("metadatas") or [[]])[0]
                    r_dists = (raw.get("distances") or [[]])[0]
                    results = [
                        (_DenseDoc(doc, meta or {}), dist)
                        for doc, meta, dist in zip(r_docs, r_metas, r_dists)
                    ]
                else:
                    results = self.vector_db.similarity_search_with_score(
                        query,
                        k=ann_fetch_k,
                    )
                # ANN candidates come back distance-sorted (ascending); keep the
                # nearest output_top_k before threshold filtering and union.
                results = self._limit_dense_candidates(results, output_top_k)
                retrieval_trace["dense_candidates"] += len(results)
                dense_attempts += 1
                if (
                    not retrieval_trace["dense_candidates"]
                    and dense_attempts >= dense_probe_queries
                ):
                    dense_disabled = True
                    retrieval_trace["dense_index_unusable"] = True
                    logger.warning(
                        "[RAG][RETRIEVE] dense index returned no candidates in %d "
                        "queries; skipping dense retrieval for this path. Rebuild "
                        "the Chroma HNSW segment to restore semantic search.",
                        dense_attempts,
                    )
                for rank, (document, score) in enumerate(results):
                    similarity = (
                        1.0 / (1.0 + score)
                        if isinstance(score, (int, float)) and score >= 0
                        else 0.0
                    )
                    raw_scores.append(similarity)
                    if not isinstance(score, (int, float)) or score > score_threshold:
                        continue
                    metadata = self._enrich_reference_metadata(
                        document.metadata or {},
                        provenance_records,
                    )
                    reference = self._format_reference_source(metadata)
                    doi = metadata.get("doi") or metadata.get("DOI") or metadata.get("doi_id")
                    if doi and "doi" not in str(reference).casefold():
                        reference = f"{reference}. DOI: {normalize_doi(doi)}"
                    added = _add_evidence(
                        reference,
                        document.page_content,
                        score=score,
                        metadata=metadata,
                        provenance_level=str(
                            metadata.get("provenance_level") or "unspecified"
                        ),
                        diagnose=rank < 3,
                        retrieval_origin="bge_m3",
                    )
                    if added:
                        retrieval_trace["dense_kept"] += 1
                self._log_retrieval(
                    query,
                    "langchain_path",
                    raw_scores,
                    len(seen_snippets) - before,
                )
            elif chroma_ready:
                vector = query_vectors.get(query)
                if vector is None:
                    vector = self._embed_query_text(query)
                if vector is None:
                    self._log_retrieval(query, "chroma_path", [], 0)
                    continue
                results = self.chroma_collection.query(
                    query_embeddings=[vector],
                    n_results=ann_fetch_k,
                    include=["documents", "metadatas", "distances"],
                )
                # over-fetched, distance-sorted; truncate to output_top_k before
                # threshold filtering and union.
                documents = self._limit_dense_candidates(
                    results.get("documents", [[]])[0], output_top_k)
                metadatas = self._limit_dense_candidates(
                    results.get("metadatas", [[]])[0], output_top_k)
                distances = self._limit_dense_candidates(
                    results.get("distances", [[]])[0], output_top_k)
                retrieval_trace["dense_candidates"] += len(documents)
                for rank, (document, metadata, score) in enumerate(
                    zip(documents, metadatas, distances)
                ):
                    similarity = (
                        1.0 / (1.0 + score)
                        if isinstance(score, (int, float)) and score >= 0
                        else 0.0
                    )
                    raw_scores.append(similarity)
                    if not isinstance(score, (int, float)) or score > score_threshold:
                        continue
                    metadata = self._enrich_reference_metadata(
                        metadata or {},
                        provenance_records,
                    )
                    reference = self._format_reference_source(metadata)
                    doi = metadata.get("doi") or metadata.get("DOI") or metadata.get("doi_id")
                    if doi and "doi" not in str(reference).casefold():
                        reference = f"{reference}. DOI: {normalize_doi(doi)}"
                    added = _add_evidence(
                        reference,
                        document,
                        score=score,
                        metadata=metadata,
                        provenance_level=str(
                            metadata.get("provenance_level") or "unspecified"
                        ),
                        diagnose=rank < 3,
                        retrieval_origin="bge_m3",
                    )
                    if added:
                        retrieval_trace["dense_kept"] += 1
                self._log_retrieval(
                    query,
                    "chroma_path",
                    raw_scores,
                    len(seen_snippets) - before,
                )

        # CRN-provenance targeted retrieval. When the reaction network records
        # the DOI an edge was derived from, that paper is the intended source of
        # truth for this pathway, yet general similarity search may rank it
        # below a topically adjacent paper about a different pigment. Look for
        # evidence inside that paper directly instead of hoping it surfaces.
        # The same strict entity/direction/condition gate still applies, so a
        # cited paper that does not actually state the conversion is not
        # promoted to evidence.
        if normalized_path_dois:
            provenance_collection = getattr(
                getattr(self, "vector_db", None), "_collection", None
            )
            if provenance_collection is None:
                provenance_collection = getattr(self, "chroma_collection", None)
            if provenance_collection is not None and hasattr(
                provenance_collection, "get"
            ):
                for provenance_doi in sorted(normalized_path_dois):
                    try:
                        payload = provenance_collection.get(
                            where={"doi": provenance_doi},
                            include=["documents", "metadatas"],
                        )
                    except Exception:
                        logger.exception(
                            "[RAG][PROVENANCE] lookup failed for doi=%s",
                            provenance_doi,
                        )
                        continue
                    documents = list(payload.get("documents") or [])
                    metadatas = list(payload.get("metadatas") or [])
                    if len(metadatas) < len(documents):
                        metadatas.extend(
                            {} for _ in range(len(documents) - len(metadatas))
                        )
                    kept_before = len(seen_snippets)
                    for document, metadata in zip(documents, metadatas):
                        metadata = self._enrich_reference_metadata(
                            metadata or {}, provenance_records
                        )
                        reference = self._format_reference_source(metadata)
                        doi_value = (
                            metadata.get("doi")
                            or metadata.get("DOI")
                            or metadata.get("doi_id")
                        )
                        if doi_value and "doi" not in str(reference).casefold():
                            reference = f"{reference}. DOI: {normalize_doi(doi_value)}"
                        _add_evidence(
                            reference,
                            document,
                            metadata=metadata,
                            provenance_level=str(
                                metadata.get("provenance_level") or "unspecified"
                            ),
                            retrieval_origin="crn_provenance",
                        )
                    logger.info(
                        "[RAG][PROVENANCE] doi=%s chunks=%d kept=%d",
                        provenance_doi,
                        len(documents),
                        len(seen_snippets) - kept_before,
                    )

        # Build the remaining three retrieval routes over the local corpus on
        # every run. Dense BGE-M3 retrieval above is one candidate generator;
        # resolver aliases, BM25 and lexical matching add complementary
        # candidates. All four routes still pass through _add_evidence, so
        # co-occurrence or a wrong direction/condition never becomes evidence.
        if True:
            collection = getattr(getattr(self, "vector_db", None), "_collection", None)
            if collection is None:
                collection = getattr(self, "chroma_collection", None)
            if collection is not None and hasattr(collection, "get"):
                scan_before = len(seen_snippets)
                scanned = 0
                prefiltered = 0
                batch_size = max(100, int(os.getenv("RAG_EVIDENCE_SCAN_BATCH", "500")))
                max_documents = max(
                    batch_size,
                    int(os.getenv("RAG_EVIDENCE_SCAN_MAX_DOCUMENTS", "10000")),
                )
                try:
                    matcher = self._get_evidence_matcher()

                    def entity_terms(entity):
                        terms = {normalize_text(entity).casefold()}
                        formula = canonical_formula(entity)
                        if formula:
                            terms.add(normalize_text(formula).casefold())
                            for (_phase, known_formula), records in (
                                matcher.resolver.by_formula.items()
                            ):
                                if known_formula == formula:
                                    terms.update(
                                        normalize_text(record.name).casefold()
                                        for record in records
                                        if str(record.name or "").strip()
                                    )
                        return {term for term in terms if len(term) >= 2}

                    def condition_term_groups(condition):
                        groups = []
                        for component in re.split(
                            r"[+;/]",
                            normalize_text(condition),
                        ):
                            key = component.strip().casefold()
                            if not key:
                                continue
                            aliases = {key}
                            aliases.update(
                                normalize_text(alias).casefold()
                                for alias in matcher.condition_resolver.aliases.get(
                                    key, set()
                                )
                            )
                            groups.append({alias for alias in aliases if len(alias) >= 2})
                        return groups

                    scan_specs = [
                        (
                            entity_terms(edge.reactant),
                            entity_terms(edge.product),
                            condition_term_groups(edge.condition),
                        )
                        for edge in edges
                    ]

                    def resolver_may_match(folded):
                        """Entity-resolver route: require both path endpoints.

                        This route deliberately uses the resolver-expanded
                        names/formulae rather than free-text relevance. The
                        strict matcher later decides direction and conditions.
                        ``folded`` is the pre-normalised, case-folded document
                        text supplied by the corpus cache, so normalisation is
                        never repeated per document per route.
                        """
                        for reactants, products, _condition_groups in scan_specs:
                            if (
                                any(term in folded for term in reactants)
                                and any(term in folded for term in products)
                            ):
                                return True
                        return False

                    def document_may_match(folded):
                        relation_hit = bool(re.search(
                            r"\b(?:convert|conversion|transform|transition|degrad|"
                            r"oxid|react|form|produc|yield|result)\w*\b",
                            folded,
                            flags=re.IGNORECASE,
                        ))
                        for reactants, products, condition_groups in scan_specs:
                            reactant_hit = bool(
                                reactants and any(term in folded for term in reactants)
                            )
                            product_hit = bool(
                                products and any(term in folded for term in products)
                            )
                            condition_hit = any(
                                any(term in folded for term in group)
                                for group in condition_groups if group
                            )
                            # Require chemical-path scope, not a condition word
                            # alone. Endpoint-only recall is allowed only when
                            # both a condition and transformation relation occur.
                            if reactant_hit and (
                                product_hit or condition_hit or relation_hit
                            ):
                                return True
                            if product_hit and condition_hit and relation_hit:
                                return True
                        return False

                    def bm25_tokens(value):
                        # Keep formula-like tokens (Pb3O4, beta-PbO2) intact
                        # while also indexing ordinary words.
                        return re.findall(
                            r"[a-z0-9]+(?:[-_][a-z0-9]+)*",
                            normalize_text(value).casefold(),
                        )

                    def route_kept_count(route):
                        return sum(
                            1
                            for bundle in ref_snippets
                            for snippet in bundle.get("snippets", [])
                            if route in snippet.get(
                                "retrieval_routes",
                                [snippet.get("retrieval_origin")],
                            )
                        )

                    # The corpus is static, so its text, folded form and BM25
                    # statistics are read and computed once per process and
                    # reused across every path rather than rebuilt each call.
                    corpus_cache = self._evidence_corpus(
                        collection, batch_size, max_documents
                    )
                    corpus = corpus_cache["docs"]
                    folded_texts = corpus_cache["folded"]
                    for (document, raw_metadata), folded in zip(corpus, folded_texts):
                        scanned += 1
                        hit_resolver = resolver_may_match(folded)
                        hit_lexical = document_may_match(folded)
                        if not (hit_resolver or hit_lexical):
                            continue
                        # Metadata enrichment and reference formatting are only
                        # needed for candidates that clear a prefilter, so they
                        # no longer run for every document in the corpus.
                        metadata = self._enrich_reference_metadata(
                            raw_metadata or {},
                            provenance_records,
                        )
                        reference = self._format_reference_source(metadata)
                        doi = (
                            metadata.get("doi")
                            or metadata.get("DOI")
                            or metadata.get("doi_id")
                        )
                        if doi and "doi" not in str(reference).casefold():
                            reference = f"{reference}. DOI: {normalize_doi(doi)}"
                        provenance_level = str(
                            metadata.get("provenance_level") or "unspecified"
                        )
                        if hit_resolver:
                            retrieval_trace["resolver_candidates"] += 1
                            _add_evidence(
                                reference,
                                document,
                                metadata=metadata,
                                provenance_level=provenance_level,
                                retrieval_origin="resolver",
                            )
                        if hit_lexical:
                            prefiltered += 1
                            _add_evidence(
                                reference,
                                document,
                                metadata=metadata,
                                provenance_level=provenance_level,
                                retrieval_origin="lexical",
                            )

                    # Dependency-free Okapi BM25 keeps the HF package small and
                    # makes the scoring definition explicit and reproducible.
                    # Corpus tokenisation and document frequencies come from the
                    # cache; only the query side is recomputed per path.
                    tokenized = corpus_cache["tokenized"]
                    doc_freq = corpus_cache["doc_freq"]
                    corpus_size = corpus_cache["corpus_size"]
                    average_length = corpus_cache["average_length"]
                    query_tokens = set(
                        token
                        for query in queries
                        for token in bm25_tokens(query)
                    )
                    k1 = float(os.getenv("RAG_BM25_K1", "1.5"))
                    b = float(os.getenv("RAG_BM25_B", "0.75"))
                    bm25_top_k = max(
                        1,
                        int(os.getenv("RAG_BM25_TOP_K", "100")),
                    )
                    ranked = []
                    for index, tokens in enumerate(tokenized):
                        if not tokens or not query_tokens:
                            continue
                        frequencies = Counter(tokens)
                        length_norm = (
                            len(tokens) / average_length
                            if average_length
                            else 1.0
                        )
                        score = 0.0
                        for token in query_tokens:
                            frequency = frequencies.get(token, 0)
                            if not frequency:
                                continue
                            frequency_docs = doc_freq.get(token, 0)
                            inverse_document_frequency = math.log(
                                1.0
                                + (
                                    corpus_size - frequency_docs + 0.5
                                ) / (frequency_docs + 0.5)
                            )
                            score += inverse_document_frequency * (
                                frequency * (k1 + 1.0)
                            ) / (
                                frequency
                                + k1 * (1.0 - b + b * length_norm)
                            )
                        if score > 0:
                            ranked.append((score, index))
                    ranked.sort(reverse=True)
                    for bm25_score, index in ranked[:bm25_top_k]:
                        document, raw_metadata = corpus[index]
                        metadata = self._enrich_reference_metadata(
                            raw_metadata or {},
                            provenance_records,
                        )
                        reference = self._format_reference_source(metadata)
                        doi = (
                            metadata.get("doi")
                            or metadata.get("DOI")
                            or metadata.get("doi_id")
                        )
                        if doi and "doi" not in str(reference).casefold():
                            reference = (
                                f"{reference}. DOI: {normalize_doi(doi)}"
                            )
                        _add_evidence(
                            reference,
                            document,
                            # BM25 scores are not on the same scale as Chroma
                            # distances. Keep the route score as metadata rather
                            # than mixing it into the generic distance field.
                            metadata={
                                **metadata,
                                "bm25_score": round(bm25_score, 6),
                            },
                            provenance_level=str(
                                metadata.get("provenance_level") or "unspecified"
                            ),
                            retrieval_origin="bm25",
                        )
                    logger.info(
                        "[RAG][FOUR_ROUTE] path=%s scanned=%d "
                        "resolver=%d bm25=%d lexical=%d",
                        path_str,
                        scanned,
                        retrieval_trace["resolver_candidates"],
                        min(len(ranked), bm25_top_k),
                        prefiltered,
                    )
                    retrieval_trace["resolver_scanned"] = scanned
                    retrieval_trace["resolver_kept"] = route_kept_count(
                        "resolver"
                    )
                    retrieval_trace["bm25_scanned"] = scanned
                    retrieval_trace["bm25_candidates"] = min(
                        len(ranked), bm25_top_k
                    )
                    retrieval_trace["bm25_kept"] = route_kept_count("bm25")
                    retrieval_trace["lexical_scanned"] = scanned
                    retrieval_trace["lexical_candidates"] = prefiltered
                    retrieval_trace["lexical_kept"] = route_kept_count(
                        "lexical"
                    )
                except Exception:
                    logger.exception(
                        "[RAG][FOUR_ROUTE] collection scan failed for path=%s",
                        path_str,
                    )

        # Related context is an audit aid for evidence gaps, not an appendix
        # to a successful answer. When direct evidence exists, suppress it to
        # avoid repeating the accepted paper and surfacing weaker distractors.
        # Otherwise retain the strongest paragraph from each source so a weak
        # pathway can still be explained from the retrieved literature.
        _has_direct_match = any(
            isinstance(match, dict) and match.get("verdict") == "direct"
            for bundle in ref_snippets
            if isinstance(bundle, dict)
            for snippet in bundle.get("snippets", [])
            for match in snippet.get("edge_matches", [])
        )
        if _has_direct_match:
            related_context.clear()
        elif related_context:
            strongest_by_source = {}
            for item in related_context:
                source_key = re.sub(
                    r"\s+", " ", str(item.get("source") or "Unknown source")
                ).strip().casefold()
                current = strongest_by_source.get(source_key)
                item_score = float((item.get("match") or {}).get("score") or 0)
                current_score = float(
                    ((current or {}).get("match") or {}).get("score") or 0
                )
                if current is None or item_score > current_score:
                    strongest_by_source[source_key] = item
            related_context[:] = sorted(
                strongest_by_source.values(),
                key=lambda item: float(
                    (item.get("match") or {}).get("score") or 0
                ),
                reverse=True,
            )[:3]

        for reference in dict.fromkeys(provenance_only):
            if reference in reference_index:
                continue
            reference_index[reference] = len(ref_list)
            ref_list.append(reference)
            ref_snippets.append({
                "text": "",
                "score": None,
                "snippets": [],
                "provenance_only": True,
                "best_verdict": provenance_best_verdict.get(
                    str(reference or "").strip().casefold()
                ),
            })

        context_records = []
        for ref_index, bundle in enumerate(ref_snippets, 1):
            for snippet in bundle.get("snippets", []):
                matches = snippet.get("edge_matches", [])
                labels = ", ".join(
                    (
                        "pathway endpoints "
                        f"{item.get('reactant', '')} -> {item.get('product', '')}: "
                        f"{item.get('verdict', 'unsupported')}"
                        if item.get("evidence_scope") == "pathway_endpoint"
                        else f"edge {item['edge_index']}: {item['verdict']}"
                    )
                    for item in matches
                )
                direct_priority = int(any(
                    item.get("verdict") == "direct"
                    for item in matches
                    if isinstance(item, dict)
                ))
                context_records.append((
                    direct_priority,
                    ref_index,
                    f"[{ref_index}] Evidence classification: {labels}\n"
                    f"{snippet.get('text', '')}",
                ))
        context_records.sort(key=lambda item: (-item[0], item[1]))
        context_parts = [item[2] for item in context_records]
        context_str = "\n\n".join(context_parts)
        if not context_str:
            context_str = "No relevant literature evidence was retrieved from the vector database."

        return {
            "context_str": context_str,
            "ref_list": ref_list,
            "ref_snippets": ref_snippets,
            "related_context": related_context,
            "retrieval_trace": retrieval_trace,
        }

    def _narrate_path_report(
        self, evidence, root_material, path_str, full_path=None
    ):
        """Write the report narrative from already-retrieved evidence.

        Split from retrieval so the expensive, deterministic half can be cached
        while this half runs fresh on every request. Retrieval and evidence
        gating depend only on (path, corpus) and cost minutes; the narrative is
        the one part that should not be frozen in a cache, and it costs seconds.
        """
        context_str = evidence["context_str"]
        ref_list = evidence["ref_list"]
        ref_snippets = evidence["ref_snippets"]
        related_context = evidence["related_context"]
        retrieval_trace = evidence["retrieval_trace"]

        safe_context = self._smart_truncate(
            context_str,
            max_chars=int(os.getenv("RAG_CONTEXT_MAX_CHARS", "8000")),
        )
        has_snippet_evidence = any(
            isinstance(item, dict) and str(item.get("text") or "").strip()
            for item in ref_snippets
        )
        has_direct_evidence = any(
            match.get("verdict") == "direct"
            for bundle in ref_snippets
            if isinstance(bundle, dict)
            for snippet in bundle.get("snippets", [])
            for match in snippet.get("edge_matches", [])
        )
        if not has_direct_evidence:
            content = ""
            is_contextual = False
            # If topically relevant literature was retrieved (even without an
            # edge-verified match), synthesize a grounded related-context
            # paragraph instead of a bare "no evidence" statement.
            if has_snippet_evidence or related_context:
                content = self._generate_contextual_narrative(
                    root_material=root_material,
                    path_str=path_str,
                    safe_context=safe_context,
                    related_context=related_context,
                    authored_steps=self._authored_steps_for_path(path_str),
                )
                is_contextual = bool(content)
            if not content:
                content = self._generate_unverified_path_report(
                    root_material=root_material,
                    path_str=path_str,
                    retrieved_context=bool(has_snippet_evidence or related_context),
                )
            return self._build_report_payload(
                content=content,
                root_material=root_material,
                path_str=path_str,
                ref_list=ref_list,
                ref_snippets=ref_snippets,
                full_path=full_path,
                evidence_is_weak=True,
                retrieval_trace=retrieval_trace,
                related_context=related_context,
                contextual_narrative=is_contextual,
            )

        system_prompt = (
            "You are a heritage conservation scientist writing an evidence-grounded "
            "reaction-mechanism introduction for a provenance report. Open by "
            "identifying the starting pigment (common name and formula) in one "
            "clause, then explain the pathway in reaction order: starting material, "
            "operative condition or reagent, transformation/mechanism, and product, "
            "naming the observed products when the snippets report them. Fold the "
            "pigment identification into the first evidence-cited sentence so it is "
            "not a free-standing uncited claim. Treat the reaction graph and "
            "literature snippets as different evidence types. Never turn graph "
            "connectivity into a documented molecular mechanism or observation."
        )

        user_prompt = f"""
Write one concise English Introduction paragraph explaining the reaction
pathway and the evidence-supported level of mechanism.

Pigment: {root_material}
Graph pathway: {path_str}

Numbered literature snippets:
{safe_context}

Rules:
- Open with a concise identification of the starting pigment (common name and chemical formula), folded into the first evidence-cited sentence; do not add a generic cultural-heritage or degradation preamble, and do not make it a separate uncited sentence.
- Describe supported transformations in graph order, stating reactant, experimental condition or reagent, transformation, and product; when a snippet reports the reaction mechanism (e.g. photo-oxidation) or names the observed products, state them.
- Use "reaction pathway" or "observed conversion" when the snippets establish only conversion. Use "molecular mechanism" or "elementary step" only when the snippet explicitly reports that level of detail.
- Do not include a "Validation note" or scope caveat; that note is added automatically outside your paragraph.
- End with one concise scope sentence when the observed conversion is supported but molecular intermediates, by-products, or elementary steps are not given by the retrieved evidence.
- A claim that a graph edge is established may use only a snippet classified as direct for that edge, and must end with its matching citation [n].
- A snippet classified as direct for "pathway endpoints" supports only the overall start-to-final conversion and facts explicitly stated in that snippet. It does not establish any intermediate, individual graph edge, elementary mechanism, or complete step sequence.
- Do not discuss, cite, or list retrieval candidates that are not classified as direct evidence for either a current graph edge or the current pathway endpoints.
- Pathway-endpoint evidence may be cited for its scoped overall conversion even when individual graph edges remain unverified.
- Do not cite a reference entry that has no snippet text.
- Write the paragraph in your own words as a connected explanation. Do NOT stitch quoted fragments together; synthesise the snippets into flowing scientific prose.
- Ground the substance, not the wording: every chemical species, environmental condition and characterisation technique you name must appear in a snippet, but you are expected to rephrase, condense and connect them.
- Never introduce a species, reagent, technique or mechanism that no snippet mentions, and never negate or reverse what a snippet reports.
- End each sentence with the citation [n] of the snippet it draws on. A sentence citing the wrong snippet is discarded, so cite the source the content actually came from.
- Report the specific experimental reagent or condition stated in the snippet; never generalize it to every member of a broader graph condition label.
- Do not turn "compared with/to" into "matched", "identical", or "equivalent" unless the snippet explicitly uses a matching or equivalence statement.
- Distinguish an observed conversion/product identified by the paper from an elementary reaction mechanism. Do not say that the conversion is unconfirmed when the snippet reports that it occurred.
- If the paper reports multiple products or phases, name the relevant co-products and do not present the selected graph product as the sole product.
- For every graph step with NO snippet classified as direct for that edge, explicitly call it "graph-derived and unverified"; do not fill the gap by inference. This applies only to steps with no direct snippet of their own: a step whose evidence label reads "edge N: direct" is established, and calling it unverified contradicts the evidence list printed beside the report. A pathway-endpoint snippet does not make a step unverified either -- it simply adds nothing to it, so judge each step by its own edge label alone.
- NEVER write that a step is unverified, unsupported, graph-derived, or lacking direct evidence when that step's evidence label reads "edge N: direct". The report prints the labels next to your paragraph, so such a sentence contradicts the evidence in the same document. For a step that IS labelled direct but whose snippets only report that the conversion occurred, the correct statement is that the transformation is supported while the molecular intermediates, by-products and elementary steps are not resolved by the retrieved evidence -- not that the step is unverified.
- Do not add a closing sentence about unresolved mechanism or evidence scope: a separate "Evidence Limitation" section is generated from the evidence records and appended automatically. Writing your own version risks stating a different boundary than the one the records support.
- Do not invent color or appearance changes, prevention advice, treatment advice, kinetics, or causal explanations.
- Preserve the supplied citation numbering.
- Use inline LaTeX ($...$) for chemical formulas.
- Output only the paragraph, without a title, headings, equation, or references list.
"""

        try:
            content = self._chat_completion(
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_prompt},
                ],
                temperature=0.3,
                max_tokens=1200,
                purpose="path_intro",
            ) or ""
        except Exception as e:
            content = f"Error: {str(e)}"

        if is_bad_generation_text(content):
            logger.info(
                "[RAG][LLM] path_intro rejected (is_bad_generation_text); "
                "using template fallback. preview=%r",
                content[:120],
            )
            content = ""
        elif content:
            logger.info("[RAG][LLM] path_intro used LLM output (%d chars)", len(content))
        if not content:
            # Unified fallback: reuse the single canonical intro template so this
            # path reads identically to the evidence-guard synthesis. The scope /
            # validation caveat is appended downstream, not duplicated here.
            fallback = None
            for ref_index, bundle in enumerate(ref_snippets, 1):
                for snippet in bundle.get("snippets", []):
                    for match in snippet.get("edge_matches", []):
                        if match.get("verdict") != "direct":
                            continue
                        fallback = _compose_direct_evidence_intro(match, ref_index)
                        break
                    if fallback:
                        break
                if fallback:
                    break
            content = fallback or (
                f"The graph records {path_str}; a direct-evidence narrative could not "
                "be generated, so unsupported graph steps remain graph-derived and "
                "unverified."
            )

        return self._build_report_payload(
            content=content,
            root_material=root_material,
            path_str=path_str,
            ref_list=ref_list,
            ref_snippets=ref_snippets,
            full_path=full_path,
            evidence_is_weak=not has_direct_evidence,
            retrieval_trace=retrieval_trace,
            related_context=related_context,
        )