scripts/database_source_contract.py

"""Evidence-aware provenance and deterministic public record links."""

from __future__ import annotations

import json
import re
from datetime import datetime, timezone
from functools import lru_cache
from pathlib import Path
from typing import Any, Iterable
from urllib.parse import parse_qsl, quote, unquote, urlencode, urlsplit, urlunsplit

REGISTRY_PATH = Path(__file__).resolve().parents[1] / "references" / "source-links.json"
REDACTED = "REDACTED"

_SECRET_QUERY_KEYS = frozenset(
    {
        "access_token",
        "api_key",
        "apikey",
        "auth",
        "authorization",
        "bearer",
        "client_secret",
        "code",
        "credential",
        "credentials",
        "email",
        "jwt",
        "key",
        "password",
        "phpsessid",
        "private_key",
        "refresh_token",
        "secret",
        "session",
        "sessionid",
        "sig",
        "signature",
        "sid",
        "sas",
        "token",
        "web_env",
        "webenv",
    }
)
_PRIVATE_QUERY_KEYS = frozenset(
    {
        "criteria",
        "diagnosis",
        "expr",
        "expression",
        "filter",
        "input",
        "keyword",
        "keywords",
        "patient",
        "prompt",
        "q",
        "query",
        "search",
        "sequence",
        "subject",
        "term",
        "terms",
        "text",
        "variables",
        "where",
    }
)
_NON_EVIDENCE_MODES = frozenset(
    {
        "check",
        "connectivity",
        "config",
        "empty",
        "egquery",
        "einfo",
        "enums",
        "field_sizes",
        "field_values",
        "fields",
        "health",
        "healthz",
        "info",
        "introspection",
        "espell",
        "metadata",
        "ping",
        "routing",
        "schema",
        "search_areas",
        "stats_size",
        "status",
        "service_info",
    }
)
_METADATA_FIELDS = frozenset(
    {
        "available",
        "apiversion",
        "build",
        "capabilities",
        "config",
        "configuration",
        "cost",
        "count",
        "cursor",
        "description",
        "duration",
        "endpoint",
        "exclude",
        "extensions",
        "fields",
        "health",
        "healthcheck",
        "head",
        "limit",
        "message",
        "meta",
        "metadata",
        "name",
        "next",
        "offset",
        "options",
        "ok",
        "operations",
        "page",
        "pages",
        "pagesize",
        "pagination",
        "queryoptions",
        "reachable",
        "schema",
        "schemaversion",
        "service",
        "servicestatus",
        "skip",
        "status",
        "timestamp",
        "title",
        "tracing",
        "total",
        "totalcount",
        "typename",
        "updated",
        "uptime",
        "url",
        "vars",
        "variables",
        "version",
        "validation",
    }
)
_ERROR_FIELDS = frozenset({"error", "errors"})
_DIAGNOSTIC_FIELDS = _ERROR_FIELDS | frozenset({"message", "messages", "warning", "warnings"})
_STRUCTURAL_METADATA_FIELDS = frozenset(
    {
        "cursor",
        "extensions",
        "meta",
        "metadata",
        "schema",
        "tracing",
        "type",
        "typename",
    }
)
_EMPTY_COUNT_FIELDS = (
    "record_count_returned",
    "association_count",
    "eqtl_count",
    "result_count",
    "returned_rows",
    "hit_count",
    "matches_count",
)
_RECORD_COLLECTION_FIELDS = (
    "records",
    "rows",
    "associations",
    "eqtls",
    "results",
    "items",
    "studies",
    "variants",
    "genes",
    "matches",
    "hits",
)
_GENERAL_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,255}$")
_DOI = re.compile(r"^10[.][0-9]{4,9}/[-._;()/:A-Z0-9]+$", re.IGNORECASE)
_UNIPROT_ACCESSION = re.compile(
    r"(?:[OPQ][0-9][A-Z0-9]{3}[0-9]|[A-NR-Z][0-9][A-Z0-9]{3}[0-9](?:[A-Z0-9]{4})?)",
    re.IGNORECASE,
)
_ENSEMBL_GENE = re.compile(r"ENS[A-Z]*G[0-9]{6,}(?:[.][0-9]+)?", re.IGNORECASE)
_VARIANT_COORDINATE = re.compile(
    r"(?:chr)?(?:[0-9]{1,2}|X|Y|MT)[-_:][0-9]+[-_:][ACGTN]+[-_:][ACGTN]+",
    re.IGNORECASE,
)


@lru_cache(maxsize=1)
def _registry() -> dict[str, Any]:
    return json.loads(REGISTRY_PATH.read_text(encoding="utf-8"))["skills"]


def _authority_has_port_delimiter(netloc: str) -> bool:
    """Return whether an authority explicitly delimits a port, including an empty one."""
    authority = netloc.rsplit("@", 1)[-1]
    if authority.startswith("["):
        closing_bracket = authority.find("]")
        return closing_bracket >= 0 and authority[closing_bracket + 1 :].startswith(":")
    return ":" in authority


def _normalized_hostname(url: str) -> tuple[str, int] | None:
    """Return a credential-free HTTPS hostname and effective port."""
    if any(character.isspace() or ord(character) < 32 or character == "\\" for character in url):
        return None
    try:
        parts = urlsplit(url)
        port = parts.port
    except (TypeError, ValueError):
        return None
    if (
        parts.scheme.casefold() != "https"
        or not parts.netloc
        or parts.hostname is None
        or parts.username is not None
        or parts.password is not None
        or _authority_has_port_delimiter(parts.netloc)
        or port is not None
    ):
        return None
    hostname = parts.hostname
    if not hostname or hostname.endswith(".") or not hostname.isascii():
        return None
    try:
        normalized = hostname.encode("idna").decode("ascii").casefold()
    except UnicodeError:
        return None
    return normalized, 443


def _normalized_source_scope(url: str) -> tuple[str, int, str] | None:
    """Return an exact HTTPS host, port, and normalized path scope."""
    normalized_host = _normalized_hostname(url)
    if normalized_host is None:
        return None
    try:
        path = urlsplit(url).path
    except ValueError:
        return None
    for _ in range(len(path) + 1):
        decoded = unquote(path)
        if decoded == path:
            break
        path = decoded
    else:
        return None
    if any(character.isspace() or ord(character) < 32 or character == "\\" for character in path):
        return None
    segments = [segment for segment in path.split("/") if segment]
    if any(segment.split(";", 1)[0] in {".", ".."} for segment in segments):
        return None
    normalized_path = "/" + "/".join(segments) if segments else "/"
    hostname, port = normalized_host
    return hostname, port, normalized_path


@lru_cache(maxsize=None)
def _trusted_source_prefixes(skill_name: str) -> tuple[tuple[str, int, str], ...]:
    """Load exact per-skill request URL scopes from the source registry."""
    entry = _registry().get(skill_name)
    if not isinstance(entry, dict):
        return ()
    prefixes: list[tuple[str, int, str]] = []
    for url in entry.get("request_url_prefixes", []):
        if not isinstance(url, str):
            continue
        normalized = _normalized_source_scope(url)
        if normalized is None or normalized in prefixes:
            continue
        prefixes.append(normalized)
    return tuple(prefixes)


@lru_cache(maxsize=None)
def _trusted_source_base_urls(skill_name: str) -> tuple[tuple[str, int, str], ...]:
    """Load exact exceptional base URLs that sit above a request scope."""
    entry = _registry().get(skill_name)
    if not isinstance(entry, dict):
        return ()
    base_urls: list[tuple[str, int, str]] = []
    for url in entry.get("request_base_urls", []):
        if not isinstance(url, str):
            continue
        try:
            parts = urlsplit(url)
        except ValueError:
            continue
        if parts.query or parts.fragment:
            continue
        normalized = _normalized_source_scope(url)
        if normalized is None or normalized in base_urls:
            continue
        base_urls.append(normalized)
    return tuple(base_urls)


def is_registered_source_origin(skill_name: str, url: str | None) -> bool:
    """Return whether a URL uses an exact registered request origin."""
    if not isinstance(url, str) or not url:
        return False
    normalized = _normalized_hostname(url)
    if normalized is None:
        return False
    return any(normalized == prefix[:2] for prefix in _trusted_source_prefixes(skill_name))


def is_registered_source_base_url(skill_name: str, url: str | None) -> bool:
    """Return whether a base URL is independently registered for this skill.

    A base normally has to be inside the same host-and-path request scope as the
    final URL. A small exact allowlist supports documented API roots that sit
    above a narrower request scope without broadening that request scope.
    """
    if not isinstance(url, str) or not url:
        return False
    try:
        parts = urlsplit(url)
    except ValueError:
        return False
    if parts.query or parts.fragment:
        return False
    normalized = _normalized_source_scope(url)
    if normalized is None:
        return False
    return is_registered_source_url(skill_name, url) or normalized in _trusted_source_base_urls(
        skill_name
    )


def is_registered_source_url(skill_name: str, url: str | None) -> bool:
    """Return whether an HTTPS URL is within an exact per-skill request scope."""
    if not isinstance(url, str) or not url:
        return False
    normalized = _normalized_source_scope(url)
    if normalized is None:
        return False
    hostname, port, path = normalized
    return any(
        hostname == prefix_host
        and port == prefix_port
        and (
            prefix_path == "/"
            or path == prefix_path
            or path.startswith(prefix_path.rstrip("/") + "/")
        )
        for prefix_host, prefix_port, prefix_path in _trusted_source_prefixes(skill_name)
    )


def sanitize_request_url(url: str | None) -> str | None:
    """Remove credentials, fragments, secrets, and user-supplied search text."""
    if not isinstance(url, str) or not url.strip():
        return None
    parts = urlsplit(url)
    if parts.scheme not in {"http", "https"} or not parts.hostname:
        return None
    sanitized_query: list[tuple[str, str]] = []
    for key, value in parse_qsl(parts.query, keep_blank_values=True):
        separated = re.sub(r"([A-Z]+)([A-Z][a-z])", r"\1_\2", key)
        separated = re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", separated)
        normalized = re.sub(r"[^a-z0-9]+", "_", separated.casefold()).strip("_")
        tokens = frozenset(normalized.split("_"))
        private = (
            normalized in _SECRET_QUERY_KEYS
            or normalized in _PRIVATE_QUERY_KEYS
            or bool(tokens & (_SECRET_QUERY_KEYS | _PRIVATE_QUERY_KEYS))
            or re.fullmatch(r"q[0-9]+", normalized) is not None
            or normalized.startswith(("query_", "search_", "filter_"))
            or normalized.endswith(
                (
                    "_credential",
                    "_jwt",
                    "_key",
                    "_password",
                    "_query",
                    "_secret",
                    "_sequence",
                    "_signature",
                    "_term",
                    "_token",
                )
            )
        )
        sanitized_query.append((key, REDACTED if private else value))
    netloc = parts.netloc.rsplit("@", 1)[-1]
    return urlunsplit((parts.scheme, netloc, parts.path, urlencode(sanitized_query), ""))


def _normalized_key(key: str) -> str:
    return re.sub(r"[^a-z0-9]+", "", key.casefold())


def _find_record_value(record: Any, field: str, depth: int = 0) -> Any:
    if depth > 5:
        return None
    wanted = _normalized_key(field)
    if isinstance(record, dict):
        for key, value in record.items():
            normalized = _normalized_key(str(key))
            if normalized in _DIAGNOSTIC_FIELDS:
                continue
            if normalized == wanted and value not in (None, ""):
                return value
        for key, value in record.items():
            if _normalized_key(str(key)) in _DIAGNOSTIC_FIELDS:
                continue
            if isinstance(value, (dict, list)):
                found = _find_record_value(value, field, depth + 1)
                if found is not None:
                    return found
    elif isinstance(record, list):
        for item in record[:20]:
            found = _find_record_value(item, field, depth + 1)
            if found is not None:
                return found
    return None


def _normalize_identifier(
    identifier_type: str,
    value: Any,
    transform: str | None,
    *,
    skill_name: str,
) -> str | None:
    if isinstance(value, bool) or not isinstance(value, (int, str)):
        return None
    identifier = str(value).strip()
    if not identifier or len(identifier) > 256:
        return None
    kind = identifier_type.casefold()
    if transform == "strip_clinvar_vcv":
        matched = re.fullmatch(r"(?:VCV)?0*([0-9]+)(?:[.][0-9]+)?", identifier, re.IGNORECASE)
        return matched.group(1) if matched else None
    if transform == "strip_rhea_prefix":
        matched = re.fullmatch(r"(?:RHEA:)?([0-9]+)", identifier, re.IGNORECASE)
        return matched.group(1) if matched else None
    if kind == "doi":
        if not _DOI.fullmatch(identifier):
            return None
        if any(segment in {".", ".."} for segment in identifier.split("/")):
            return None
        return identifier
    if not _GENERAL_IDENTIFIER.fullmatch(identifier):
        return None
    if kind in {"uniprot accession", "uniprotkb accession"}:
        return identifier.upper() if _UNIPROT_ACCESSION.fullmatch(identifier) else None
    if kind == "variant":
        if skill_name == "gwas-catalog-skill":
            return (
                identifier.lower() if re.fullmatch(r"rs[0-9]+", identifier, re.IGNORECASE) else None
            )
        return identifier if _VARIANT_COORDINATE.fullmatch(identifier) else None
    if kind == "variant id":
        if skill_name == "civic-skill":
            return identifier if identifier.isdigit() else None
        if skill_name == "pharmgkb-skill":
            return (
                identifier.upper() if re.fullmatch(r"PA[0-9]+", identifier, re.IGNORECASE) else None
            )
        return None
    if kind == "collection id":
        return (
            identifier.lower()
            if re.fullmatch(
                r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}",
                identifier,
                re.IGNORECASE,
            )
            else None
        )
    if kind == "compound id":
        return (
            identifier.upper() if re.fullmatch(r"CHEMBL[0-9]+", identifier, re.IGNORECASE) else None
        )
    if kind == "target id":
        if skill_name == "chembl-skill":
            return (
                identifier.upper()
                if re.fullmatch(r"CHEMBL[0-9]+", identifier, re.IGNORECASE)
                else None
            )
        if skill_name == "opentargets-skill":
            return identifier if _ENSEMBL_GENE.fullmatch(identifier) else None
        return None
    if kind == "gene id":
        if skill_name == "pharmgkb-skill":
            return (
                identifier.upper() if re.fullmatch(r"PA[0-9]+", identifier, re.IGNORECASE) else None
            )
        if skill_name in {"gnomad-graphql-skill", "gtex-eqtl-skill"}:
            return identifier if _ENSEMBL_GENE.fullmatch(identifier) else None
        return None
    if kind == "study accession":
        patterns = {
            "biostudies-arrayexpress-skill": r"(?:E-[A-Z]+-[0-9]+|S-[A-Z0-9-]+|GSE[0-9]+)",
            "gwas-catalog-skill": r"GCST[0-9]+",
            "mgnify-skill": r"MGYS[0-9]+",
        }
        pattern = patterns.get(skill_name)
        return (
            identifier.upper()
            if pattern and re.fullmatch(pattern, identifier, re.IGNORECASE)
            else None
        )
    if kind == "encode accession":
        return (
            identifier.upper()
            if re.fullmatch(r"ENC[A-Z0-9]{6,}", identifier, re.IGNORECASE)
            else None
        )
    if kind == "association accession":
        return identifier if identifier.isdigit() else None
    if kind == "phenotype":
        return identifier if re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,63}", identifier) else None
    if kind == "gene symbol or ensembl gene id":
        if _ENSEMBL_GENE.fullmatch(identifier):
            return identifier
        return (
            identifier
            if re.fullmatch(r"[A-Z][A-Z0-9]{1,14}(?:-[A-Z0-9]{1,10})?", identifier)
            else None
        )
    if kind in {
        "pmid",
        "ncbi gene id",
        "compound cid",
        "substance sid",
        "assay aid",
        "evidence id",
    }:
        return identifier if identifier.isdigit() else None
    if kind == "nct id":
        return (
            identifier.upper() if re.fullmatch(r"NCT[0-9]{8}", identifier, re.IGNORECASE) else None
        )
    if kind == "refsnp id":
        return identifier.lower() if re.fullmatch(r"rs[0-9]+", identifier, re.IGNORECASE) else None
    if kind == "geo accession":
        return (
            identifier.upper()
            if re.fullmatch(r"(?:GSE|GSM|GPL|GDS)[0-9]+", identifier, re.IGNORECASE)
            else None
        )
    if kind == "pmcid":
        return (
            identifier.upper()
            if re.fullmatch(r"PMC[0-9]+(?:[.][0-9]+)?", identifier, re.IGNORECASE)
            else None
        )
    if kind == "pdb id":
        return identifier.upper() if re.fullmatch(r"[A-Za-z0-9]{4}", identifier) else None
    if kind == "reactome stable id":
        return identifier.upper() if re.fullmatch(r"R-[A-Za-z]{3}-[0-9]+", identifier) else None
    if kind == "go term":
        return (
            identifier.upper() if re.fullmatch(r"GO:[0-9]{7}", identifier, re.IGNORECASE) else None
        )
    if kind == "chebi id":
        return (
            identifier.upper() if re.fullmatch(r"CHEBI:[0-9]+", identifier, re.IGNORECASE) else None
        )
    if kind == "metabolights study accession":
        return (
            identifier.upper() if re.fullmatch(r"MTBLS[0-9]+", identifier, re.IGNORECASE) else None
        )
    if kind == "analysis accession":
        return (
            identifier.upper() if re.fullmatch(r"MGYA[0-9]+", identifier, re.IGNORECASE) else None
        )
    if kind == "sample accession":
        return (
            identifier.upper()
            if re.fullmatch(r"(?:SAMN|SAMD|SAMEA|ERS|SRS)[0-9]+", identifier, re.IGNORECASE)
            else None
        )
    if kind == "genome accession":
        return (
            identifier.upper()
            if re.fullmatch(r"(?:GCF|GCA)_[0-9]{9}(?:[.][0-9]+)?", identifier, re.IGNORECASE)
            else None
        )
    if kind == "disease id":
        return (
            identifier
            if re.fullmatch(
                r"(?:EFO|MONDO|HP|DOID|NCIT|OTAR|Orphanet)[_:][0-9]+",
                identifier,
                re.IGNORECASE,
            )
            else None
        )
    if kind == "chemical id":
        return identifier.upper() if re.fullmatch(r"PA[0-9]+", identifier, re.IGNORECASE) else None
    if kind == "pride project accession" or kind == "proteomexchange accession":
        return identifier.upper() if re.fullmatch(r"PXD[0-9]+", identifier, re.IGNORECASE) else None
    if kind == "uniref cluster":
        return identifier if re.fullmatch(r"UniRef(?:50|90|100)_[A-Z0-9_]+", identifier) else None
    if kind == "uniparc accession":
        return (
            identifier.upper()
            if re.fullmatch(r"UPI[A-Z0-9]{7,}", identifier, re.IGNORECASE)
            else None
        )
    if kind == "stable ensembl id":
        return (
            identifier
            if re.fullmatch(r"ENS[A-Z]*[GTPER][0-9]{6,}(?:[.][0-9]+)?", identifier, re.IGNORECASE)
            else None
        )
    if kind == "rnacentral id":
        return (
            identifier.upper()
            if re.fullmatch(r"URS[0-9A-F]{10,}", identifier, re.IGNORECASE)
            else None
        )
    if kind == "string protein identifier":
        return identifier if re.fullmatch(r"[0-9]+[.][A-Za-z0-9_.-]+", identifier) else None
    return None


def _canonical_record_urls(
    skill_name: str,
    record: dict[str, Any],
    *,
    database: str | None = None,
    identifier_type: str | None = None,
) -> list[str]:
    """Construct every public record page backed by an explicit, safe mapping."""
    if not isinstance(record, dict):
        return []
    entry = _registry().get(skill_name)
    if not isinstance(entry, dict):
        return []
    database_name = str(database or record.get("database") or record.get("db") or "").casefold()
    canonical_urls: list[str] = []
    for mapping in entry.get("record_url_templates", []):
        kind = mapping.get("identifier_type")
        if not isinstance(kind, str) or (identifier_type is not None and kind != identifier_type):
            continue
        fields = list(mapping.get("identifier_fields", []))
        if (
            skill_name == "civic-skill"
            and identifier_type in {"variant ID", "evidence ID"}
            and kind == identifier_type
        ):
            fields.extend(("id", "uid"))
        if skill_name == "ncbi-entrez-skill":
            if kind == "PMID" and database_name == "pubmed":
                fields.extend(("id", "uid"))
            elif kind == "PMCID" and database_name == "pmc":
                fields.extend(("id", "uid"))
            elif kind == "NCBI Gene ID" and database_name == "gene":
                fields.extend(("id", "uid"))
            elif kind == "PMID" and database_name and database_name != "pubmed":
                continue
            elif kind == "PMCID" and database_name and database_name != "pmc":
                continue
            elif kind == "NCBI Gene ID" and database_name and database_name != "gene":
                continue
        for field in fields:
            value = _find_record_value(record, field)
            if (
                kind == "PMCID"
                and database_name == "pmc"
                and isinstance(value, (int, str))
                and not isinstance(value, bool)
                and str(value).isdigit()
            ):
                value = f"PMC{value}"
            identifier = _normalize_identifier(
                kind, value, mapping.get("transform"), skill_name=skill_name
            )
            if identifier is None:
                continue
            template = mapping.get("template")
            if (
                not isinstance(template, str)
                or not template.startswith("https://")
                or "{id}" not in template
            ):
                continue
            encoded = quote(identifier, safe="/" if kind.casefold() == "doi" else "")
            canonical = template.replace("{id}", encoded)
            if canonical not in canonical_urls:
                canonical_urls.append(canonical)
            break
    return canonical_urls


def canonical_record_url(
    skill_name: str,
    record: dict[str, Any],
    *,
    database: str | None = None,
    identifier_type: str | None = None,
) -> str | None:
    """Construct the first deterministic public page backed by a safe mapping."""
    canonical_urls = _canonical_record_urls(
        skill_name,
        record,
        database=database,
        identifier_type=identifier_type,
    )
    return canonical_urls[0] if canonical_urls else None


def _has_substantive_value(value: Any) -> bool:
    """Find a nonempty scalar without depth limits or cycle risk."""
    pending = [value]
    seen: set[int] = set()
    while pending:
        current = pending.pop()
        if current is None or current is False or current == "":
            continue
        if isinstance(current, (dict, list)):
            if id(current) in seen:
                continue
            seen.add(id(current))
            pending.extend(current.values() if isinstance(current, dict) else current)
            continue
        return True
    return False


def _is_zero_count(value: Any) -> bool:
    """Recognize numeric zero counts without coercing arbitrary strings."""
    if isinstance(value, bool):
        return False
    if isinstance(value, int):
        return value == 0
    if not isinstance(value, str):
        return False
    return (
        re.fullmatch(
            r"[+-]?(?:0+(?:[.]0*)?|[.]0+)(?:[eE][+-]?[0-9]+)?",
            value.strip(),
        )
        is not None
    )


def _has_collection_value(value: Any) -> bool:
    """Find a record value while ignoring structural metadata wrappers."""
    if not isinstance(value, (dict, list)):
        return False
    pending = [value]
    seen: set[int] = set()
    while pending:
        current = pending.pop()
        if isinstance(current, (dict, list)):
            if id(current) in seen:
                continue
            seen.add(id(current))
            if isinstance(current, dict):
                pending.extend(
                    child
                    for key, child in current.items()
                    if _normalized_key(str(key))
                    not in _DIAGNOSTIC_FIELDS | _STRUCTURAL_METADATA_FIELDS
                )
            else:
                pending.extend(current)
        elif not isinstance(current, bool) and current is not None and current != "":
            return True
    return False


def _has_identifier_collection_value(value: Any) -> bool:
    """Accept containers or a single positive numeric Entrez identifier."""
    if isinstance(value, (dict, list)):
        return _has_collection_value(value)
    if isinstance(value, bool):
        return False
    if isinstance(value, int):
        return value > 0
    return isinstance(value, str) and value.strip().isdigit() and int(value.strip()) > 0


def _summary_mode(summary: Any) -> str:
    """Classify nested response summaries with metadata-wrapper suppression."""
    if not _has_substantive_value(summary):
        return "empty"
    if not isinstance(summary, dict):
        return "evidence"
    normalized_collections = {_normalized_key(name) for name in _RECORD_COLLECTION_FIELDS}
    identifier_collections = {"idlist", "ids", "uids"}
    normalized_counts = {
        "associationcount",
        "count",
        "eqtlcount",
        "hitcount",
        "matchescount",
        "numresults",
        "numtotalresults",
        "recordcount",
        "recordcountreturned",
        "returnedrows",
        "resultcount",
        "total",
        "totalcount",
    }
    saw_empty = False
    saw_evidence = False
    saw_failure = False
    pending: list[tuple[dict[str, Any] | list[Any], bool]] = [(summary, False)]
    seen: set[int] = set()
    while pending:
        current, suppress_leaves = pending.pop()
        if id(current) in seen:
            continue
        seen.add(id(current))
        if isinstance(current, list):
            for item in current:
                if isinstance(item, (dict, list)):
                    pending.append((item, suppress_leaves))
                elif (
                    not suppress_leaves
                    and not isinstance(item, bool)
                    and _has_substantive_value(item)
                ):
                    saw_evidence = True
            continue
        for key, value in current.items():
            normalized = _normalized_key(str(key))
            if normalized in _STRUCTURAL_METADATA_FIELDS:
                continue
            if normalized in _DIAGNOSTIC_FIELDS:
                if normalized in _ERROR_FIELDS and _has_substantive_value(value):
                    saw_failure = True
                continue
            if normalized in normalized_collections:
                if _has_collection_value(value):
                    saw_evidence = True
                else:
                    saw_empty = True
            elif normalized in identifier_collections:
                if _has_identifier_collection_value(value):
                    saw_evidence = True
                else:
                    saw_empty = True
            elif normalized in normalized_counts:
                if _is_zero_count(value):
                    saw_empty = True
            elif normalized == "data" and not _has_substantive_value(value):
                saw_empty = True
            elif isinstance(value, (dict, list)):
                pending.append((value, suppress_leaves or normalized in _METADATA_FIELDS))
            elif (
                not suppress_leaves
                and normalized not in _METADATA_FIELDS
                and not isinstance(value, bool)
                and _has_substantive_value(value)
            ):
                saw_evidence = True
    if saw_evidence:
        return "evidence"
    if saw_failure:
        return "failure"
    if saw_empty:
        return "empty"
    return "metadata"


def _strip_reserved_canonical_fields(value: Any) -> None:
    """Remove untrusted upstream annotations before deciding whether evidence exists."""
    pending = [value]
    seen: set[int] = set()
    while pending:
        current = pending.pop()
        if not isinstance(current, (dict, list)) or id(current) in seen:
            continue
        seen.add(id(current))
        if isinstance(current, list):
            pending.extend(current)
            continue
        current.pop("canonical_url", None)
        current.pop("canonical_urls", None)
        pending.extend(child for child in current.values() if isinstance(child, (dict, list)))


def _evidence_mode(output: dict[str, Any], requested_mode: str | None) -> str:
    mode = str(requested_mode or output.get("mode") or output.get("action") or "").casefold()
    if mode in _NON_EVIDENCE_MODES:
        return mode
    if output.get("endpoint") in {
        "einfo",
        "egquery",
        "espell",
        "schema",
        "status",
        "health",
    }:
        return "metadata"
    path = str(output.get("path") or "").casefold().split("?", 1)[0].rstrip("/")
    if "meta" in path.split("/"):
        return "metadata"
    if path.rsplit("/", 1)[-1] in {
        "capabilities",
        "config",
        "configuration",
        "fields",
        "health",
        "healthz",
        "info",
        "introspection",
        "liveness",
        "metadata",
        "openapi",
        "openapi.json",
        "ping",
        "readiness",
        "ready",
        "schema",
        "service-info",
        "service_info",
        "status",
    }:
        return "metadata"
    summary = output.get("summary")
    summary_mode = _summary_mode(summary) if "summary" in output else None
    if summary_mode == "evidence":
        return "evidence"
    saw_record_collection = False
    for field in _RECORD_COLLECTION_FIELDS:
        if field in output:
            saw_record_collection = True
            if _has_collection_value(output[field]):
                return "evidence"
    if summary_mode == "failure":
        return "failure"
    if saw_record_collection:
        return "empty"
    for field in _EMPTY_COUNT_FIELDS:
        if field in output:
            value = output[field]
            if (
                isinstance(value, int) and not isinstance(value, bool) and value <= 0
            ) or _is_zero_count(value):
                return "empty"
    if summary_mode is not None:
        return summary_mode
    if output.get("text_head") or output.get("raw_output_path"):
        return "evidence"
    return "metadata"


def _annotate_records(
    value: Any,
    skill_name: str,
    urls: list[str],
    database: str | None,
    depth: int = 0,
    identifier_type: str | None = None,
) -> None:
    if depth > 5:
        return
    if isinstance(value, list):
        for item in value[:100]:
            _annotate_records(item, skill_name, urls, database, depth + 1, identifier_type)
        return
    if not isinstance(value, dict):
        if isinstance(value, (int, str)) and not isinstance(value, bool):
            canonical = canonical_record_url(skill_name, {"id": value}, database=database)
            if canonical is not None and canonical not in urls:
                urls.append(canonical)
        return
    record = value
    if skill_name == "pharmgkb-skill":
        object_class = value.get("objCls")
        identifier = value.get("id")
        if isinstance(object_class, str) and isinstance(identifier, (int, str)):
            identifier_field = {
                "chemical": "chemical_id",
                "gene": "gene_id",
                "variant": "variant_id",
            }.get(object_class.casefold())
            if identifier_field is not None:
                record = {**value, identifier_field: identifier}
    elif skill_name == "rhea-skill":
        accession = value.get("accession")
        if isinstance(accession, dict) and isinstance(accession.get("value"), str):
            record = {**value, "rhea_id": accession["value"]}
    record_urls = _canonical_record_urls(
        skill_name, record, database=database, identifier_type=identifier_type
    )
    if record_urls:
        value["canonical_url"] = record_urls[0]
        if len(record_urls) > 1:
            value["canonical_urls"] = record_urls
        else:
            value.pop("canonical_urls", None)
        for canonical in record_urls:
            if canonical not in urls:
                urls.append(canonical)
    else:
        value.pop("canonical_url", None)
        value.pop("canonical_urls", None)
    for key, child in list(value.items()):
        if (
            _normalized_key(str(key)) not in _DIAGNOSTIC_FIELDS
            and key
            not in {
                "sources",
                "checked_sources",
                "canonical_url",
                "canonical_urls",
            }
            and isinstance(child, (dict, list))
        ):
            context_type = None
            if skill_name == "civic-skill":
                normalized = _normalized_key(str(key))
                if normalized in {"variant", "variants"}:
                    context_type = "variant ID"
                elif normalized in {"evidence", "evidenceitem", "evidenceitems"}:
                    context_type = "evidence ID"
            _annotate_records(child, skill_name, urls, database, depth + 1, context_type)


def apply_source_contract(
    output: dict[str, Any],
    skill_name: str,
    request_url: str | None = None,
    *,
    mode: str | None = None,
) -> dict[str, Any]:
    """Attach evidence-only sources; retain empty/metadata checks separately."""
    if not isinstance(output, dict) or output.get("ok") is not True:
        return output
    entry = _registry().get(skill_name)
    if not isinstance(entry, dict):
        return output
    _strip_reserved_canonical_fields(output)
    sanitized_url = sanitize_request_url(request_url)
    kind = _evidence_mode(output, mode)
    canonical_urls: list[str] = []
    database = output.get("database") or output.get("db")
    if kind == "evidence":
        for field in (
            *_RECORD_COLLECTION_FIELDS,
            "summary",
            "matrix",
            "variant",
            "query_variant",
        ):
            if field in output:
                _annotate_records(
                    output[field],
                    skill_name,
                    canonical_urls,
                    str(database) if database else None,
                )
    source: dict[str, Any] = {
        "name": entry["source_name"],
        "url": (
            canonical_urls[0]
            if len(canonical_urls) == 1
            else sanitized_url or entry.get("homepage_url")
        ),
        "retrieved_at": datetime.now(timezone.utc).isoformat(),
        "supports_claim": kind == "evidence",
        "kind": "evidence" if kind == "evidence" else "checked",
    }
    if sanitized_url is not None:
        source["request_url"] = sanitized_url
    if canonical_urls:
        source["canonical_url"] = canonical_urls[0]
    if len(canonical_urls) > 1:
        source["canonical_urls"] = canonical_urls
    if kind == "evidence":
        existing = [
            item
            for item in output.get("sources", [])
            if isinstance(item, dict) and item.get("supports_claim") is True
        ]
        output["sources"] = existing + [source]
    else:
        source["reason"] = kind
        output.pop("sources", None)
        output.setdefault("checked_sources", []).append(source)
    return output


def evidence_sources(outputs: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
    """Propagate only actual evidence; never promote checked-but-empty sources."""
    collected: list[dict[str, Any]] = []
    seen: set[tuple[str, str]] = set()
    for output in outputs:
        if not isinstance(output, dict) or output.get("ok") is not True:
            continue
        for source in output.get("sources", []):
            if not isinstance(source, dict) or source.get("supports_claim") is not True:
                continue
            if source.get("kind") != "evidence":
                continue
            key = (
                str(source.get("name", "")),
                str(source.get("canonical_url") or source.get("url", "")),
            )
            if key not in seen:
                seen.add(key)
                collected.append(source)
    return collected