"""Warrant resolution: claim-pack record → surface chunk → derivations row.

Implements ticket #000031 Phase 2. Closes the warrant gap left open at
#000029: claim-pack records cap at ANCHOR-WARRANTED on the four-rung
ladder until each citation is bound to a Merkle-proven surface span.
This module turns ``source_reference`` strings into structured
citations, resolves them against the surface-ingested textbooks
(#000031 Phase 1), computes Merkle inclusion proofs over the matching
chunks, and writes ``derivations`` rows that future verifier passes
can use to upgrade the audit_mode of any answer that cites the
record.

This is distinct from ``arborist/qa/warrant.py`` (warrant-LITE
anchor-class check from Module H+ / #000003) — that module verifies
per-claim anchors at Q&A time. This module binds claim-pack records
to their citation surfaces via Merkle proofs at ingest/curation time.

Three pure-data steps + one DB write:

1. ``parse_citation(s) → list[Citation]``  — pure regex, deterministic
2. ``resolve_chunks(c, shards) → [Match]`` — FTS5 across surface shards
3. ``compute_proof(shard, doc_root, chunk_id) → ProofBlob``
4. ``write_derivation(...)``                — INSERT into derivations

CLI surface (in ``arborist/cli.py``):
    arborist warrant-status   --shards-dir ...    [read-only]
    arborist warrant-resolve  --shards-dir ... [--write]

Verifier wiring is intentionally NOT in scope here. Writing the
derivations rows is the data substrate; the four-rung-ladder upgrade
that lifts answers citing these records from ANCHOR-WARRANTED to
EVIDENCE-WARRANTED is a follow-up that reads from the rows this
module writes.

Source: ticket #000031 §8 Phase 2.
"""

from __future__ import annotations

import json
import re
import sqlite3
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Iterator, Optional


# ---------------------------------------------------------------------------
# Citation parsing
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class Citation:
    """A structured representation of one cited work.

    The claim-pack ``source_reference`` field is free-text; this
    dataclass pins what we can recover deterministically. ``title`` is
    the most discriminating field for shard matching; ``authors`` is
    the fallback. ``section`` carries chapter / section / equation
    refs when present (e.g., "§1.2.6 equation (13)") for downstream
    chunk-resolution heuristics.

    ``via_citation_alias`` flags Citations that were appended to a
    record's citation list via `#000041` lookup (rather than parsed
    from the original ``source_reference``). When True, every
    ResolutionMatch produced from this Citation inherits
    ``via_alias=True`` so the derivations `process_id` correctly
    reflects the alias substitution. Defaults False to preserve the
    parse-from-source path.
    """

    title: str = ""
    authors: tuple[str, ...] = field(default_factory=tuple)
    year: str = ""
    section: str = ""
    raw: str = ""
    via_citation_alias: bool = False

    def is_empty(self) -> bool:
        return not (self.title or self.authors)


_BY_RE = re.compile(r"\s+by\s+", re.IGNORECASE)
_YEAR_RE = re.compile(r"\b(1[6-9]\d{2}|20\d{2}|21\d{2})\b")
_SECTION_RE = re.compile(
    r"§[^;,]*|Chapter\s+\d+[^;,]*|Volume\s+\d+[^;,]*|equation\s*\([^)]*\)",
    re.IGNORECASE,
)
_AUTHORS_TAIL_RE = re.compile(r"\s+(?:et\s+al\.?|and\s+others)$", re.IGNORECASE)
_KNOWN_PUBLISHERS_RE = re.compile(
    r"\b(TAOCP|Enumerative Combinatorics|Foundations of [A-Za-z ]+)\b",
    re.IGNORECASE,
)


def parse_citation(source_reference: str) -> list[Citation]:
    """Parse a ``source_reference`` string into one or more Citations.

    Handles three observed patterns:

    1. ``"Title by Author"`` — split on ``" by "``, single citation.
    2. ``"Title by A1, A2, and A3"`` — same, multi-author tail.
    3. Semicolon-separated multi-citation:
       ``"Knuth TAOCP §1.2.6; Stanley §1.2; Brualdi §3.5"`` — split,
       parse each independently.

    Empty / unparseable input yields an empty list. Outputs are best-
    effort: a free-text reference like ``"Pascal 1654"`` returns one
    Citation with ``title=""``, ``authors=("Pascal",)``, ``year="1654"``.
    """
    if not source_reference:
        return []
    text = source_reference.strip()
    if not text:
        return []

    # Pattern 3: semicolon split first.
    if ";" in text:
        out: list[Citation] = []
        for part in text.split(";"):
            out.extend(parse_citation(part.strip()))
        return out

    # Pattern 1+2: "by" split.
    m = _BY_RE.search(text)
    if m:
        title_part = text[: m.start()].strip()
        authors_part = text[m.end() :].strip()
        return [
            Citation(
                title=_clean(title_part),
                authors=_split_authors(authors_part),
                year=_extract_year(text),
                section=_extract_section(text),
                raw=text,
            )
        ]

    # Compact form: "Knuth TAOCP Volume 1 §1.2.6" — heuristic.
    return [
        Citation(
            title=_clean(_strip_metadata(text)),
            authors=_compact_author(text),
            year=_extract_year(text),
            section=_extract_section(text),
            raw=text,
        )
    ]


def _clean(s: str) -> str:
    return re.sub(r"\s+", " ", s).strip(" ,.;:")


def _split_authors(authors_part: str) -> tuple[str, ...]:
    """Split an authors-string on commas + 'and' / 'et al.' patterns.

    Handles Oxford-comma form ("A, B, and C") + simple form ("A and
    B") + comma-only form ("A, B"). The leading "and " on the final
    Oxford-style author is stripped post-split.
    """
    s = _AUTHORS_TAIL_RE.sub("", authors_part).strip()
    s = _strip_metadata(s)
    if not s:
        return ()
    parts = re.split(r",\s*|\s+and\s+", s)
    cleaned: list[str] = []
    for p in parts:
        p = p.strip()
        # Strip a leading "and " left over from Oxford-comma split
        # (e.g., ", and John L. Safko" → "and John L. Safko" → "John L. Safko").
        p = re.sub(r"^and\s+", "", p, flags=re.IGNORECASE).strip()
        if p:
            cleaned.append(p)
    return tuple(cleaned)


def _compact_author(text: str) -> tuple[str, ...]:
    text = _strip_metadata(text)
    m = _KNOWN_PUBLISHERS_RE.search(text)
    if not m:
        first = text.split()
        return (first[0],) if first else ()
    author_part = text[: m.start()].strip()
    return tuple(p.strip() for p in author_part.split(",") if p.strip())


def _extract_year(text: str) -> str:
    m = _YEAR_RE.search(text)
    return m.group(1) if m else ""


def _extract_section(text: str) -> str:
    matches = _SECTION_RE.findall(text)
    return "; ".join(m.strip() for m in matches) if matches else ""


def _strip_metadata(text: str) -> str:
    text = _SECTION_RE.sub("", text)
    text = _YEAR_RE.sub("", text)
    return _clean(text)


# ---------------------------------------------------------------------------
# Chunk resolution
# ---------------------------------------------------------------------------


@dataclass
class ResolutionMatch:
    """One resolved (citation → surface chunk) candidate.

    ``via_alias`` flags matches that REQUIRED a registered citation
    or term alias to be found (#000041 + #000042). The unaliased
    cascade tries first; if it produces no match, the aliased
    cascade is tried, and its matches carry ``via_alias=True``.
    Used by ``warrant_resolve`` to pick the right ``process_id``
    for the derivations row (`warrant-resolver-v1+alias` for
    via_alias matches, `warrant-resolver-v1` otherwise).
    """

    citation: Citation
    shard_path: str
    document_root: str
    document_uri: str
    document_title: str
    chunk_id: int
    chunk_idx: int
    score: float
    snippet: str
    via_alias: bool = False


# Per-shard cached set of (lowercased title tokens). Reading the
# documents table for every (shard, citation) pair is O(N×M); cache
# the lowered title strings once per shard to bring it to O(N+M).
_SHARD_TITLE_CACHE: dict[str, str] = {}


def _shard_title_haystack(shard_path: str) -> str:
    """Concatenate every non-claim_pack surface document's title +
    URI in this shard, lowercased. Cached per shard. Used as the
    haystack for citation-token lookup."""
    if shard_path in _SHARD_TITLE_CACHE:
        return _SHARD_TITLE_CACHE[shard_path]
    try:
        conn = sqlite3.connect(f"file:{shard_path}?mode=ro", uri=True)
    except sqlite3.OperationalError:
        _SHARD_TITLE_CACHE[shard_path] = ""
        return ""
    try:
        rows = conn.execute(
            "SELECT title, document_uri FROM documents "
            "WHERE source_type != 'claim_pack' AND kind = 'surface'"
        ).fetchall()
    finally:
        conn.close()
    parts = []
    for title, uri in rows:
        parts.append((title or "").lower())
        parts.append((uri or "").lower())
    haystack = " ".join(parts)
    _SHARD_TITLE_CACHE[shard_path] = haystack
    return haystack


def _shard_matches_citation(shard_path: str, citation: Citation) -> bool:
    """True if the shard plausibly contains the cited author's work.

    Heuristic: at least one of the citation's author last names
    (length >= 4) AND at least one significant title token appear
    in the haystack of the shard's non-claim_pack surface document
    titles + URIs. Author-only or title-only matches don't count —
    "Wikipedia article ABOUT Mendelson" shouldn't be confused with
    "Mendelson's textbook".

    This produces honest "no match" results for cited textbooks not
    ingested in any shard (Mendelson, Enderton, Jech, Goldstein,
    Barendregt, etc.) — exactly the records that should stay at
    ANCHOR-WARRANTED until those textbooks are surface-ingested.
    """
    if citation.is_empty():
        return False
    haystack = _shard_title_haystack(shard_path)
    if not haystack:
        return False

    author_lasts: list[str] = []
    for a in citation.authors:
        last = a.split()[-1] if a else ""
        # Strip "et al." artifacts
        last = re.sub(r"[^A-Za-z]", "", last)
        if len(last) >= 4 and last[0].isupper():
            author_lasts.append(last.lower())
    if not author_lasts:
        return False
    if not any(a in haystack for a in author_lasts):
        return False

    # Author hit confirmed; require AT LEAST ONE title token (length
    # >= 4) to also appear, ruling out shards that mention the author
    # incidentally (e.g., a biography page).
    title_terms = [
        t.lower() for t in (citation.title or "").split() if len(t) >= 4
    ]
    if not title_terms:
        # Compact citations like "Pascal 1654" — author-only is fine.
        return True
    return any(t in haystack for t in title_terms)


def _fts5_search(
    shard_path: str,
    query: str,
    limit: int = 5,
    surface_only: bool = True,
    exclude_claim_pack: bool = True,
) -> list[tuple]:
    """FTS5 search against a shard. Returns rows of
    ``(chunk_id, idx, document_root, document_uri, title, snippet, rank)``.

    ``exclude_claim_pack`` filters out the very source_type the
    resolver is matching FOR — claim-pack records have title +
    chunk content rich in the cited keywords (they ARE the
    cited material's curriculum-restated form), and an unfiltered
    FTS5 match would just keep returning self-matches. The resolver
    only cares about EXTERNAL surface chunks.
    """
    try:
        conn = sqlite3.connect(f"file:{shard_path}?mode=ro", uri=True)
    except sqlite3.OperationalError:
        return []
    try:
        clauses = []
        if surface_only:
            clauses.append("d.kind = 'surface'")
        if exclude_claim_pack:
            clauses.append("d.source_type != 'claim_pack'")
        where_extra = ("AND " + " AND ".join(clauses)) if clauses else ""
        rows = conn.execute(
            f"""
            SELECT c.chunk_id, c.idx, c.document_root, d.document_uri,
                   d.title, snippet(chunks_fts, 0, '[', ']', '…', 24) AS sn,
                   chunks_fts.rank
            FROM chunks_fts
            JOIN chunks c ON chunks_fts.rowid = c.chunk_id
            JOIN documents d ON c.document_root = d.document_root
            WHERE chunks_fts MATCH ?
              {where_extra}
            ORDER BY chunks_fts.rank
            LIMIT ?
            """,
            (query, limit),
        ).fetchall()
        return list(rows)
    except sqlite3.OperationalError:
        return []
    finally:
        conn.close()


_FTS_STOPWORDS = {"of", "the", "a", "an", "and", "or", "to", "in", "on", "is", "by"}

# Categorical terms that classify a record but don't discriminate
# between records ("Axiom of X", "Theorem of Y" — both share "Axiom"
# / "Theorem"). Drop these from the discriminating-token set so the
# resolver can find DIFFERENT chunks per axiom in the same textbook.
# Without this filter, "Pasch's Axiom" and "Axiom of Betweenness"
# both rank for "axiom" and end up pointing at the same chunk.
_GENERIC_THEOREM_TERMS = {
    "axiom", "axioms", "theorem", "theorems", "lemma", "lemmas",
    "principle", "principles", "rule", "rules", "law", "laws",
    "identity", "identities", "definition", "definitions",
    "property", "properties", "schema",
}


def _discriminating_tokens(theorem_name: str) -> list[str]:
    """Pull the tokens from a theorem name that distinguish it from
    sibling theorems in the same textbook. Drops categorical terms
    like "Axiom" / "Theorem" / "Principle"; keeps proper-noun-ish or
    domain-specific words.

    "Pasch's Axiom"          → ["Pasch"]
    "Axiom of Betweenness"   → ["Betweenness"]
    "Law of Excluded Middle" → ["Excluded", "Middle"]
    "Newton's First Law"     → ["Newton", "First"]
    """
    if not theorem_name:
        return []
    out = []
    for t in re.findall(r"[A-Za-z]{3,}", theorem_name):
        low = t.lower()
        if low in _FTS_STOPWORDS or low in _GENERIC_THEOREM_TERMS:
            continue
        out.append(t)
    return out


def _build_fts_query(citation: Citation, theorem_name: str = "") -> str:
    """Compose an FTS5 MATCH query from a citation + theorem name.

    Strategy: discriminating tokens from the theorem name (proper
    nouns / domain terms, NOT categorical "Axiom"/"Theorem")
    AND'd together — FTS5 will only return chunks that contain ALL
    of them. Falls back to author + title tokens when no
    discriminating signal is available.

    The shift from OR to AND on the discriminating tokens is the
    key fix: per-axiom queries now point at per-axiom chunks
    (Pasch's Axiom → chunk mentioning "Pasch"; Axiom of Betweenness
    → chunk mentioning "Betweenness"), not at one shared chunk
    that mentions "axiom" everywhere.
    """
    discriminating = _discriminating_tokens(theorem_name)
    if discriminating:
        # AND-join on discriminating tokens (FTS5 default operator
        # IS AND — but explicit for readability + safety against
        # quoting weirdness on multi-token records).
        return " AND ".join(t.lower() for t in discriminating)
    # Fall back: author last name + title tokens, OR'd.
    parts: list[str] = []
    for a in citation.authors:
        last = a.split()[-1] if a else ""
        if last and len(last) >= 4 and last not in parts:
            parts.append(last)
    if citation.title:
        for t in re.findall(r"[A-Za-z]{4,}", citation.title)[:4]:
            if t.lower() not in _GENERIC_THEOREM_TERMS:
                parts.append(t)
    return " OR ".join(p.lower() for p in parts) if parts else ""


# ---------------------------------------------------------------------------
# Phase 5 — phrase-first + content-token fallback (#000039)
#
# Problem: discriminating tokens like "Line", "Plane", "Incidence" are
# ubiquitous in Hilbert's geometry text, so the AND-joined query
# returns too many candidates and BM25 can't pick a winner. Layered
# cascade strategy:
#   1. Phrase match on the discriminating multi-word phrase from the
#      theorem name (FTS5 phrase syntax `"line incidence"`).
#   2. Content-token AND-join from the claim-pack record's chunk
#      content (∇verbose prose has more discriminating language than
#      the categorical title).
#   3. Existing discriminating-token AND-join (the current strategy).
#   4. Existing OR-fallback (title + author).
# Each query is tried in order; first non-empty result wins.
# ---------------------------------------------------------------------------

# Common English words to filter out of content-token candidates.
# Hand-picked from the top-100 most frequent English words; small
# enough to keep in source. The full FTS5 stopword list lives in the
# tokenizer config; this is for our DISCRIMINATING-token heuristic.
_COMMON_ENGLISH = {
    "this", "that", "with", "from", "have", "been", "were", "they",
    "their", "there", "which", "would", "could", "should", "these",
    "those", "where", "when", "what", "than", "then", "what", "such",
    "into", "more", "most", "other", "another", "some", "each",
    "every", "many", "much", "only", "even", "also", "still",
    "just", "very", "well", "make", "take", "give", "find",
    "good", "great", "same", "first", "last", "next", "prior",
    "case", "form", "term", "kind", "part", "side", "type",
    "thus", "hence", "must", "shall", "will", "does", "doing",
    "between", "above", "below", "without", "within", "through",
    "shown", "given", "because", "however", "therefore",
    "respectively", "namely", "particular", "particularly",
    "general", "generally", "specific", "specifically",
    "expression", "verbosely", "shorthand", "spelt", "denoted",
    "states", "stating", "stated", "provides", "providing",
    "logically", "logical", "mathematical", "system", "systems",
}


def _phrase_for_axiom(theorem_name: str) -> str:
    """Pick a multi-word discriminating phrase from a theorem name.

    Strips a leading "Axiom of " / "Theorem " / "Principle " prefix,
    then keeps the rest as a phrase if it has at least 2 tokens of
    length >= 3 (otherwise returns "" — single-token theorems use
    the existing AND-join strategy).

    "Axiom of Line Incidence"     → '"line incidence"'
    "Axiom of Plane Incidence"    → '"plane incidence"'
    "Axiom of Side-Angle-Side (SAS)" → '"side angle side"'
    "Pasch's Axiom"               → ""
    "Pythagorean Theorem"         → ""
    """
    if not theorem_name:
        return ""
    # Strip leading "Axiom of " / "Theorem " / "Principle " etc.
    text = re.sub(
        r"^(?:Axiom\s+(?:of\s+|Schema\s+of\s+)?|"
        r"Theorem\s+(?:of\s+)?|"
        r"Principle\s+(?:of\s+)?|"
        r"Law\s+of\s+|"
        r"Rule\s+of\s+|"
        r"Definition\s+of\s+)",
        "",
        theorem_name,
        flags=re.IGNORECASE,
    )
    # Drop trailing parenthetical (e.g., "(SAS)" or "(SQD §14.1)").
    text = re.sub(r"\s*\([^)]*\)\s*$", "", text).strip()
    # Tokenize on non-letter chars (handles hyphens, apostrophes,
    # numbers, etc.). Lower-case for FTS5.
    toks = [
        t.lower() for t in re.findall(r"[A-Za-z]{3,}", text)
        if t.lower() not in _GENERIC_THEOREM_TERMS
        and t.lower() not in _FTS_STOPWORDS
    ]
    if len(toks) < 2:
        return ""
    return '"' + " ".join(toks) + '"'


def _phrase_from_parenthetical(theorem_name: str) -> str:
    """Extract a phrase from a trailing parenthetical when one
    exists, e.g.

      "Kolmogorov's First Axiom (Non-Negativity)"  → '"non negativity"'
      "Newton's First Law (Law of Inertia)"        → '"inertia"' (single tok kept here — paren content is high-signal)
      "Axiom of Side-Angle-Side (SAS)"             → '"sas"' (acronym retained)

    Used as a SECOND-CHANCE phrase when `_phrase_for_axiom` produces
    something like "kolmogorov first" that doesn't appear verbatim
    in the substrate. The parenthetical often carries the actual
    discriminating term the substrate uses.
    """
    if not theorem_name:
        return ""
    m = re.search(r"\(([^)]+)\)", theorem_name)
    if not m:
        return ""
    inner = m.group(1)
    # Drop "Law of " / "Axiom of " etc. INSIDE the parenthetical too.
    inner = re.sub(
        r"^(?:Axiom\s+of\s+|Theorem\s+of\s+|Law\s+of\s+|Principle\s+of\s+)",
        "",
        inner,
        flags=re.IGNORECASE,
    )
    toks = [
        t.lower() for t in re.findall(r"[A-Za-z]{2,}", inner)
        if t.lower() not in _GENERIC_THEOREM_TERMS
        and t.lower() not in _FTS_STOPWORDS
    ]
    if not toks:
        return ""
    return '"' + " ".join(toks) + '"'


def _content_tokens(chunk_content: str, max_n: int = 8) -> list[str]:
    """Extract discriminating tokens from a claim-pack chunk's
    content. Used as the FTS5 query when phrase + title-token
    strategies don't find a match.

    Filters: length >= 5; not in stopword / generic / common-English
    sets; not pure-digits; appears at least twice in the content
    (rare-singleton tokens are likely typos or LaTeX residue).
    Returns top ``max_n`` by length-then-position (longer first;
    ties broken by earlier appearance — body content beats footers).
    """
    if not chunk_content:
        return []
    raw = re.findall(r"[A-Za-z]{5,}", chunk_content)
    counts: dict[str, int] = {}
    first_pos: dict[str, int] = {}
    for i, tok in enumerate(raw):
        low = tok.lower()
        if low in _FTS_STOPWORDS:
            continue
        if low in _GENERIC_THEOREM_TERMS:
            continue
        if low in _COMMON_ENGLISH:
            continue
        counts[low] = counts.get(low, 0) + 1
        if low not in first_pos:
            first_pos[low] = i
    # Keep tokens with count >= 2 (ditch typo / LaTeX-residue
    # singletons); sort by length DESC then first-position ASC.
    candidates = [
        (low, n, first_pos[low])
        for low, n in counts.items()
        if n >= 2
    ]
    candidates.sort(key=lambda x: (-len(x[0]), x[2]))
    return [c[0] for c in candidates[:max_n]]


def _build_record_query_cascade(
    citation: Citation,
    theorem_name: str,
    record_content: str = "",
) -> list[str]:
    """Build a cascade of FTS5 queries to try in order. First
    non-empty query that yields a result wins (caller handles the
    cascade in :func:`resolve_chunks`).

    Order, tightest → loosest:
      1. title-phrase (e.g. '"line incidence"')
      2. parenthetical-phrase (e.g. '"non negativity"' from
         "Kolmogorov's First Axiom (Non-Negativity)")
      3. content-tokens AND-join (top 5)
      4. content-tokens OR-join (top 3) — softer fallback when AND
         is too restrictive
      5. legacy single-token / discriminating-token query
    """
    out: list[str] = []
    seen: set[str] = set()

    def _push(q: str) -> None:
        if q and q not in seen:
            seen.add(q)
            out.append(q)

    _push(_phrase_for_axiom(theorem_name))
    _push(_phrase_from_parenthetical(theorem_name))
    if record_content:
        ctoks = _content_tokens(record_content)
        if ctoks:
            # Tight: top-5 AND-join.
            _push(" AND ".join(ctoks[:5]))
            # Looser: top-3 OR-join — fallback when AND zeroes out.
            if len(ctoks) >= 2:
                _push(" OR ".join(ctoks[:3]))
    # Existing strategy as last resort.
    _push(_build_fts_query(citation, theorem_name=theorem_name))
    return out


def resolve_chunks(
    citation: Citation,
    shards_dir: Path | str,
    theorem_name: str = "",
    record_content: str = "",
    limit: int = 5,
    aliases_db: Path | str | None = None,
    domain: str = "",
) -> list[ResolutionMatch]:
    """Search every surface shard under ``shards_dir`` for chunks that
    match the citation + theorem-name signal. Returns a ranked list.

    Iterates the sibling ``crawl/`` dir's textbook surface shards
    (per the post-#000031-Phase-1 layout). The main numbered shards
    (000.db, 001.db, …) are excluded — they hold Wikipedia content,
    which a textbook-citation resolver shouldn't be matching against;
    Wikipedia's "Mendelson" article is *about* Mendelson, not the
    cited textbook itself, and including the main shards both bloats
    runtime (3.4 M docs per shard) and produces false positives.

    ``record_content`` (Phase 5 / #000039) is the claim-pack chunk's
    body. When provided, the resolver tries a phrase + content-token
    cascade that produces tighter per-axiom matches than the
    title-only AND-join — load-bearing for axioms whose titles use
    ubiquitous-in-the-textbook tokens like "Line" / "Plane" /
    "Incidence".
    """
    if citation.is_empty():
        return []
    shards_dir = Path(shards_dir).expanduser()

    candidate_shards: list[Path] = []
    sibling = shards_dir.parent / "crawl"
    if sibling.is_dir():
        for db in sorted(sibling.glob("*.db")):
            # Skip non-textbook crawl shards (e.g., the ad-hoc
            # `crawl_russell_ballestrini_net.db` blog crawl).
            if db.name.startswith(("crawl_russell_", "qa.", "snapshots.")):
                continue
            candidate_shards.append(db)

    base_queries = _build_record_query_cascade(
        citation, theorem_name=theorem_name, record_content=record_content
    )
    if not base_queries:
        return []

    # Two-pass cascade: try the UNALIASED queries first (matches
    # carry via_alias=False); only fall through to the alias-
    # expanded queries when the unaliased cascade produces nothing
    # for a given shard. Term-alias OR-expansion (#000042) preserves
    # phrase syntax.
    aliased_queries: list[str] = []
    if aliases_db and domain:
        try:
            from arborist.qa.aliases import expand_query_with_term_aliases

            for q in base_queries:
                expanded = expand_query_with_term_aliases(q, domain, aliases_db)
                # Only add to the alias-cascade if the expansion is
                # actually different from the original — otherwise
                # we're just re-running the same query.
                if expanded != q:
                    aliased_queries.append(expanded)
        except Exception:
            aliased_queries = []

    # Citation-alias substitutions (#000041) propagate via_alias to
    # every match they produce — Pass 1 hits from a substitute Citation
    # are still alias-driven, just not via term-alias query expansion
    # (#000042). Both alias mechanisms collapse to the same
    # via_alias=True downstream so process_id correctly attributes
    # the derivation as `warrant-resolver-v1+alias`.
    citation_alias_floor = bool(getattr(citation, "via_citation_alias", False))

    matches: list[ResolutionMatch] = []
    for shard in candidate_shards:
        if not _shard_matches_citation(str(shard), citation):
            continue
        # Pass 1: unaliased cascade.
        hit = False
        for q in base_queries:
            rows = _fts5_search(str(shard), q, limit=limit)
            if rows:
                for chunk_id, idx, doc_root, doc_uri, title, snippet, rank in rows:
                    matches.append(
                        ResolutionMatch(
                            citation=citation,
                            shard_path=str(shard),
                            document_root=doc_root,
                            document_uri=doc_uri,
                            document_title=title or "",
                            chunk_id=chunk_id,
                            chunk_idx=idx,
                            score=-float(rank or 0.0),
                            snippet=snippet or "",
                            via_alias=citation_alias_floor,
                        )
                    )
                hit = True
                break
        if hit:
            continue
        # Pass 2: alias-expanded cascade (only when unaliased found
        # nothing). Matches here carry via_alias=True.
        for q in aliased_queries:
            rows = _fts5_search(str(shard), q, limit=limit)
            if rows:
                for chunk_id, idx, doc_root, doc_uri, title, snippet, rank in rows:
                    matches.append(
                        ResolutionMatch(
                            citation=citation,
                            shard_path=str(shard),
                            document_root=doc_root,
                            document_uri=doc_uri,
                            document_title=title or "",
                            chunk_id=chunk_id,
                            chunk_idx=idx,
                            score=-float(rank or 0.0),
                            snippet=snippet or "",
                            via_alias=True,
                        )
                    )
                break
    matches.sort(key=lambda m: m.score, reverse=True)
    return matches[:limit]


# ---------------------------------------------------------------------------
# Merkle inclusion proof
# ---------------------------------------------------------------------------


@dataclass
class ProofBlob:
    """Serializable Merkle inclusion proof for (document_root,
    chunk_idx). Deterministic JSON serialization."""

    document_root: str
    chunk_idx: int
    chunk_id: int
    leaf_hash: str
    siblings: list[dict]

    def to_json(self) -> str:
        return json.dumps(asdict(self), sort_keys=True)


def compute_proof(
    shard_path: str,
    document_root: str,
    chunk_id: int,
) -> Optional[ProofBlob]:
    """Compute a Merkle inclusion proof for a chunk inside its parent
    document's tree. Returns None if the chunk isn't in the shard.

    Reads ``merkle_nodes`` for the document_root, walks layer by layer
    to assemble siblings. The proof_blob serializes deterministic JSON.
    Compatible with the existing ``MerkleProof`` semantics in
    ``arborist/merkle.py``.
    """
    try:
        conn = sqlite3.connect(f"file:{shard_path}?mode=ro", uri=True)
    except sqlite3.OperationalError:
        return None
    try:
        chunk_row = conn.execute(
            "SELECT idx, leaf_hash FROM chunks "
            "WHERE chunk_id = ? AND document_root = ?",
            (chunk_id, document_root),
        ).fetchone()
        if chunk_row is None:
            return None
        chunk_idx, leaf_hash = chunk_row
        nodes = conn.execute(
            "SELECT layer, idx, hash FROM merkle_nodes "
            "WHERE document_root = ? "
            "ORDER BY layer, idx",
            (document_root,),
        ).fetchall()
        if not nodes:
            # Single-chunk document — no merkle_nodes row, leaf is root.
            return ProofBlob(
                document_root=document_root,
                chunk_idx=chunk_idx,
                chunk_id=chunk_id,
                leaf_hash=leaf_hash,
                siblings=[],
            )
        by_layer: dict[int, dict[int, str]] = {}
        for layer, idx, hash_ in nodes:
            by_layer.setdefault(layer, {})[idx] = hash_
        max_layer = max(by_layer)
        siblings: list[dict] = []
        cur_idx = chunk_idx
        for layer in range(max_layer):
            layer_map = by_layer.get(layer, {})
            if cur_idx % 2 == 0:
                sibling_idx = cur_idx + 1
                is_left = False
            else:
                sibling_idx = cur_idx - 1
                is_left = True
            sibling_hash = layer_map.get(sibling_idx)
            if sibling_hash is None:
                # Odd-element rule: self-duplicate.
                sibling_hash = layer_map.get(cur_idx, "")
                is_left = False
            siblings.append({"hash": sibling_hash, "is_left": is_left})
            cur_idx = cur_idx // 2
        return ProofBlob(
            document_root=document_root,
            chunk_idx=chunk_idx,
            chunk_id=chunk_id,
            leaf_hash=leaf_hash,
            siblings=siblings,
        )
    finally:
        conn.close()


# ---------------------------------------------------------------------------
# Derivations row writer
# ---------------------------------------------------------------------------


WARRANT_PROCESS_ID = "warrant-resolver-v1"


def write_derivation(
    db_path: str,
    record_root: str,
    surface_root: str,
    proof_blob: ProofBlob,
    process_id: str = WARRANT_PROCESS_ID,
    distilled_at: Optional[int] = None,
) -> bool:
    """Insert one derivation row binding a claim-pack record to its
    surface chunk via Merkle proof. Idempotent — same (record_root,
    surface_root, process_id) PK collides on re-insert with
    ``INSERT OR IGNORE``; returns True if the row was inserted, False
    if it already existed.

    Schema reuse note: ``derivations.core_root`` carries the claim-
    pack record root here, even though the record's ``kind`` is
    ``'surface'`` not ``'core'``. The schema FK only requires
    existence in ``documents``, not a specific kind. ``process_id``
    discriminates this use-case (``warrant-resolver-v1``) from the
    existing distillation processes (``tfidf``, ``first_sentence``).
    """
    if distilled_at is None:
        distilled_at = int(time.time())
    conn = sqlite3.connect(db_path)
    try:
        cur = conn.execute(
            "INSERT OR IGNORE INTO derivations "
            "(core_root, src_root, proof_blob, process_id, distilled_at) "
            "VALUES (?, ?, ?, ?, ?)",
            (record_root, surface_root, proof_blob.to_json(), process_id, distilled_at),
        )
        conn.commit()
        return cur.rowcount > 0
    finally:
        conn.close()


# ---------------------------------------------------------------------------
# Claim-pack record iteration
# ---------------------------------------------------------------------------


# `Source:` field extraction. Claim-pack ingest pipeline canonicalizes
# whitespace runs into single spaces (per `arborist.document.canonicalize`),
# so the multi-line layout of the source bundle collapses into a flat
# string at chunk time. Match `Source: <text>` up to the next field-
# marker keyword or end-of-string.
_SOURCE_RE = re.compile(
    r"Source:\s*(.+?)(?=\s+(?:Group|Role|Category|Subfield)\s*:|\s*$)",
    re.IGNORECASE | re.DOTALL,
)


def _decompress_chunk(content) -> str:
    """Decompress a chunks.content blob if zstd-compressed; else
    decode as UTF-8. Mirrors arborist's chunk-loading convention."""
    if isinstance(content, str):
        return content
    if not isinstance(content, (bytes, bytearray)):
        return ""
    if len(content) >= 4 and content[:4] == b"\x28\xb5\x2f\xfd":
        try:
            import zstandard

            return zstandard.ZstdDecompressor().decompress(content).decode("utf-8")
        except Exception:
            return ""
    try:
        return content.decode("utf-8")
    except UnicodeDecodeError:
        return ""


def iter_claim_pack_records(
    shards_dir: Path | str,
) -> Iterator[tuple[str, str, str, str, str, str]]:
    """Yield ``(shard_path, record_root, title, source_reference,
    chunk_content, pillar)`` for every claim_pack record across the
    shards-dir.

    Phase 5 (#000040) added the ``chunk_content`` element so the
    warrant resolver can use the record's ∇verbose body as a
    content-token signal.

    #000042 added ``pillar`` (Roman numeral I-VII / IX) so the
    resolver can pick the right domain when looking up term aliases
    (pillar IV → "geometry"; pillar I → "logic"; etc.). Pillar is
    parsed from the doc URI's `/pillar/<P>/` segment per the
    claim_pack source's URI scheme.
    """
    shards_dir = Path(shards_dir).expanduser()
    for db in sorted(shards_dir.glob("*.db")):
        if db.name.startswith(("qa.", "snapshots.")):
            continue
        try:
            conn = sqlite3.connect(f"file:{db}?mode=ro", uri=True)
        except sqlite3.OperationalError:
            continue
        try:
            for row in conn.execute(
                "SELECT d.document_root, d.title, d.document_uri, c.content "
                "FROM documents d JOIN chunks c ON c.document_root = d.document_root "
                "WHERE d.source_type = 'claim_pack' "
                "ORDER BY d.ingest_ts"
            ):
                doc_root, title, doc_uri, content = row
                text = _decompress_chunk(content)
                m = _SOURCE_RE.search(text)
                source_ref = m.group(1).strip() if m else ""
                pillar = _pillar_from_uri(doc_uri or "")
                yield (
                    str(db), doc_root, title or "", source_ref, text, pillar
                )
        finally:
            conn.close()


_PILLAR_URI_RE = re.compile(r"/pillar/([IVX]+)/", re.IGNORECASE)


def _pillar_from_uri(uri: str) -> str:
    """Pull the pillar Roman numeral from a claim-pack document URI.
    URIs follow the scheme
    `claim-pack://<bundle>/pillar/<P>/{axioms|theorems}/...`."""
    m = _PILLAR_URI_RE.search(uri)
    return m.group(1).upper() if m else ""


def _has_derivation(db_path: str, record_root: str) -> bool:
    try:
        conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
    except sqlite3.OperationalError:
        return False
    try:
        row = conn.execute(
            "SELECT 1 FROM derivations WHERE core_root = ? AND process_id = ? LIMIT 1",
            (record_root, WARRANT_PROCESS_ID),
        ).fetchone()
        return row is not None
    finally:
        conn.close()


def warrant_chains_for_sources(
    sources: list[dict],
    shards_dir: Path | str,
) -> list[dict]:
    """Given a list of cited source dicts (each with ``document_root``),
    return entries for those whose document_root has a warrant-resolver
    derivation row.

    Used at Q&A render time to flip the four-rung-ladder display label
    from ANCHOR-WARRANTED toward EVIDENCE-WARRANTED when the cited
    record has a Merkle-proven binding to a primary-source surface.

    Each returned entry: ``{record_root, surface_root, surface_title,
    surface_uri, chunk_idx, host_shard}``.
    """
    if not sources:
        return []
    shards_dir = Path(shards_dir).expanduser()

    # Walk every shard once to collect the warrant-resolver derivations.
    # The cost is bounded by total number of derivation rows across all
    # shards in the cluster — small relative to a Q&A render.
    chains: list[dict] = []
    for db in sorted(shards_dir.glob("*.db")):
        if db.name.startswith(("qa.", "snapshots.")):
            continue
        try:
            conn = sqlite3.connect(f"file:{db}?mode=ro", uri=True)
        except sqlite3.OperationalError:
            continue
        try:
            for row in conn.execute(
                "SELECT dr.core_root, dr.src_root, dr.proof_blob "
                "FROM derivations dr "
                "WHERE dr.process_id = ?",
                (WARRANT_PROCESS_ID,),
            ):
                core, src, blob = row
                # Match this derivation to one of the answer's cited
                # source document_roots.
                for s in sources:
                    if s.get("document_root") == core:
                        try:
                            pb = json.loads(blob)
                            chunk_idx = pb.get("chunk_idx")
                        except (TypeError, ValueError):
                            chunk_idx = None
                        chains.append({
                            "record_root": core,
                            "surface_root": src,
                            "chunk_idx": chunk_idx,
                            "host_shard": str(db),
                        })
                        break
        finally:
            conn.close()
    # De-duplicate by (record_root, surface_root) — same chain may
    # show up if the same record is cited multiple times.
    seen: set[tuple[str, str]] = set()
    unique: list[dict] = []
    for c in chains:
        k = (c["record_root"], c["surface_root"])
        if k in seen:
            continue
        seen.add(k)
        unique.append(c)
    return unique


# ---------------------------------------------------------------------------
# High-level orchestration
# ---------------------------------------------------------------------------


@dataclass
class WarrantStatus:
    """Per-record summary for the ``warrant-status`` CLI."""

    record_root: str
    record_title: str
    source_reference: str
    citations: list[Citation]
    matches: list[ResolutionMatch]
    has_derivation: bool


def warrant_status(
    shards_dir: Path | str,
    limit: int = 5,
    use_aliases: bool = False,
) -> list[WarrantStatus]:
    """Walk every claim-pack record and report what surface chunks the
    citation resolver finds. Read-only — no DB writes.

    ``use_aliases`` (#000041 + #000042) — when True, the resolver
    looks up registered citation + term aliases stored in the
    shards' main DB and uses them to expand the FTS5 query +
    propose substitute citations. Default False to preserve the
    original strict behavior.
    """
    out: list[WarrantStatus] = []
    aliases_db = (
        str(Path(shards_dir).expanduser() / "000.db") if use_aliases else None
    )
    for shard_path, record_root, title, source_ref, content, pillar in (
        iter_claim_pack_records(shards_dir)
    ):
        citations = parse_citation(source_ref)
        # #000041 — citation-alias substitution.
        if use_aliases and aliases_db and source_ref:
            try:
                from arborist.qa.aliases import lookup_citation_aliases

                for sub in lookup_citation_aliases(aliases_db, source_ref):
                    citations.append(
                        Citation(
                            title=sub.substitute_title,
                            authors=sub.substitute_authors,
                            raw=sub.substitute_ref,
                            via_citation_alias=True,
                        )
                    )
            except Exception:
                pass

        # Map pillar → domain for term-alias lookups.
        domain = ""
        if use_aliases and pillar:
            try:
                from arborist.qa.aliases import domain_for_pillar

                domain = domain_for_pillar(pillar)
            except Exception:
                domain = ""

        all_matches: list[ResolutionMatch] = []
        for c in citations:
            all_matches.extend(
                resolve_chunks(
                    c,
                    shards_dir,
                    theorem_name=title,
                    record_content=content,
                    limit=limit,
                    aliases_db=aliases_db,
                    domain=domain,
                )
            )
        all_matches.sort(key=lambda m: m.score, reverse=True)
        out.append(
            WarrantStatus(
                record_root=record_root,
                record_title=title,
                source_reference=source_ref,
                citations=citations,
                matches=all_matches[:limit],
                has_derivation=_has_derivation(shard_path, record_root),
            )
        )
    return out


WARRANT_PROCESS_ID_ALIASED = "warrant-resolver-v1+alias"


def warrant_resolve(
    shards_dir: Path | str,
    write: bool = False,
    limit: int = 1,
    use_aliases: bool = False,
) -> dict:
    """Run the resolver across every claim-pack record. With
    ``write=False`` (default) returns a dry-run summary. With
    ``write=True``, computes Merkle inclusion proofs for top-scoring
    matches and writes ``derivations`` rows.

    When ``use_aliases=True``, citation + term aliases (#000041 +
    #000042) get applied: query expansion via OR-joined alternate
    terms, and substitute citations alongside originals. Alias-
    resolved derivations carry ``process_id =
    "warrant-resolver-v1+alias"`` so audit can tell substituted
    chains from original ones.
    """
    records_total = 0
    records_resolved = 0
    derivations_written = 0
    statuses = warrant_status(shards_dir, limit=limit, use_aliases=use_aliases)
    # Build a record_root → shard_path map once so we don't iterate
    # again per record.
    record_shards: dict[str, str] = {}
    for shard_path, record_root, _, _, _, _ in iter_claim_pack_records(shards_dir):
        record_shards[record_root] = shard_path

    for status in statuses:
        records_total += 1
        if not status.matches:
            continue
        records_resolved += 1
        if not write:
            continue
        top = status.matches[0]
        proof = compute_proof(top.shard_path, top.document_root, top.chunk_id)
        if proof is None:
            continue
        host_shard = record_shards.get(status.record_root)
        # The match itself carries the alias flag — set when the
        # unaliased cascade missed and the alias-expanded cascade
        # produced the hit. This is honest: alias-resolved chains
        # only get the +alias process_id when the alias was
        # actually load-bearing for the match.
        process_id = (
            WARRANT_PROCESS_ID_ALIASED if top.via_alias
            else WARRANT_PROCESS_ID
        )
        if host_shard and write_derivation(
            host_shard,
            status.record_root,
            top.document_root,
            proof,
            process_id=process_id,
        ):
            derivations_written += 1
    return {
        "records_total": records_total,
        "records_resolved": records_resolved,
        "derivations_written": derivations_written,
        "wrote": write,
        "use_aliases": use_aliases,
    }
