ticket #000031 Phase 2: warrant resolver + 18 derivations rows landed

Closes the warrant-promotion data path: claim-pack records now
bind to surface-ingested textbook chunks via Merkle inclusion
proofs in the existing `derivations` table.

What landed
===========
arborist/qa/warrant_resolver.py — four pure-data steps + one DB
write:

1. parse_citation(s) — regex pipeline turning the claim-pack
   `source_reference` string into structured Citation tuples.
   Handles "Title by Author" (single + Oxford-comma multi +
   et-al), semicolon-separated multi-cite ("Knuth §1.2.6;
   Stanley §1.2; Brualdi §3.5"), and compact author-year
   ("Pascal 1654") forms.

2. resolve_chunks(c, shards_dir) — FTS5 search across sibling
   crawl/ dir's textbook-surface shards. Skips the main numbered
   shards (Wikipedia content; would be false positives). Per-
   shard match filter requires BOTH author last name AND a title
   token in the shard's title-haystack — honest "no match" for
   textbooks not yet surface-ingested.

3. compute_proof(shard, doc_root, chunk_id) — reads
   merkle_nodes, walks layer-by-layer to assemble siblings;
   emits deterministic JSON proof_blob compatible with
   arborist/merkle.py verification.

4. write_derivation(...) — INSERT OR IGNORE into the existing
   derivations table with process_id="warrant-resolver-v1".
   Idempotent at the database layer.

CLI surface
===========
- `arborist warrant-status --shards-dir ...` (read-only) —
  emits per-record JSON: parsed citations, FTS5 candidates,
  whether a derivations row exists.
- `arborist warrant-resolve --shards-dir ... [--write]` —
  default dry-run summary; --write actually computes proofs
  and inserts rows.

End-to-end verification
=======================
Real-shard run: `arborist warrant-resolve --shards-dir
~/.arborist/shards --write` →

  records_total: 92
  records_resolved: 18
  derivations_written: 18

All 18 are pillar-IV Hilbert axioms citing "The Foundations of
Geometry by David Hilbert" — the only cited textbook fully
surface-ingested by Phase 1. The remaining 74 records cite
textbooks not in our shard cluster (Mendelson, Enderton,
Jech, Goldstein, Barendregt, Stanley, Brualdi, Knuth, …) and
correctly produce 0 matches; they stay at ANCHOR-WARRANTED
until those textbooks land via future Phase-1 manifest
expansions.

Re-running the writer is a no-op (PK collision on (core_root,
src_root, process_id) = INSERT OR IGNORE).

Drive-by fix
============
arborist/sources/textbook_tex.py — _extract_title now also
parses PG's plain-text `Author:` line and appends "by Author"
to the title, so the warrant resolver's author-last-name match
works against PG-ingested textbooks (Hilbert "The Foundations
of Geometry by David Hilbert" instead of just "The Foundations
of Geometry").

Test suite
==========
tests/test_warrant_resolver.py — 14 unit tests for the citation
parser (no DB / network). Full suite: 1588 passed / 28 skipped.

Phase 3 (verifier wiring)
=========================
NOT in this commit. The data substrate is in place; the
audit_mode upgrade path that lifts answers citing
claim-pack-records-with-derivations from ANCHOR-WARRANTED to
EVIDENCE-WARRANTED requires a verifier change — touches well-
tested code, worth its own ticket so the regression risk is
bounded.

arborist/cli.py

@@ -3332,6 +3332,71 @@ def _cmd_selfmodel_list(args: argparse.Namespace) -> int:
     return 0
 
 
+def _cmd_warrant_status(args: argparse.Namespace) -> int:
+    """Read-only — show what surface chunks the citation resolver
+    finds for each claim-pack record. JSON output. No DB writes.
+    See ``arborist.qa.warrant_resolver`` for ticket #000031 Phase 2.
+    """
+    from arborist.qa.warrant_resolver import warrant_status
+
+    shards_dir = args.global_shards_dir or args.shards_dir
+    if not shards_dir:
+        print("--shards-dir is required for warrant-status", file=sys.stderr)
+        return 2
+    results = warrant_status(shards_dir, limit=args.limit)
+    out = []
+    for r in results:
+        out.append(
+            {
+                "record_root": r.record_root,
+                "record_title": r.record_title,
+                "source_reference": r.source_reference,
+                "citations": [
+                    {
+                        "title": c.title,
+                        "authors": list(c.authors),
+                        "year": c.year,
+                        "section": c.section,
+                    }
+                    for c in r.citations
+                ],
+                "matches": [
+                    {
+                        "shard": m.shard_path,
+                        "document_root": m.document_root,
+                        "document_title": m.document_title,
+                        "chunk_id": m.chunk_id,
+                        "score": m.score,
+                        "snippet": m.snippet,
+                    }
+                    for m in r.matches
+                ],
+                "has_derivation": r.has_derivation,
+            }
+        )
+    print(json.dumps(out, indent=2, ensure_ascii=False))
+    return 0
+
+
+def _cmd_warrant_resolve(args: argparse.Namespace) -> int:
+    """Run the warrant resolver across all claim-pack records.
+    Default dry-run (just summary counts); ``--write`` flag computes
+    Merkle inclusion proofs for top matches and writes
+    ``derivations`` rows. Idempotent at the database layer.
+    """
+    from arborist.qa.warrant_resolver import warrant_resolve
+
+    shards_dir = args.global_shards_dir or args.shards_dir
+    if not shards_dir:
+        print("--shards-dir is required for warrant-resolve", file=sys.stderr)
+        return 2
+    summary = warrant_resolve(
+        shards_dir, write=args.write, limit=args.limit
+    )
+    print(json.dumps(summary, indent=2))
+    return 0
+
+
 def _cmd_mesh_status(args: argparse.Namespace) -> int:
     """Show mesh state: enabled flag, identity, current epoch, roster."""
     from arborist.mesh import current_epoch, is_enabled, load_identity
@@ -5010,6 +5075,43 @@ def build_parser() -> argparse.ArgumentParser:
     sm_list.add_argument("--limit", type=int, default=20)
     sm_list.set_defaults(func=_cmd_selfmodel_list)
 
+    # ----- warrant resolver (#000031 Phase 2) --------------------------------
+    warrant_status_cmd = sub.add_parser(
+        "warrant-status",
+        help="show citation-resolver matches per claim-pack record (read-only)",
+    )
+    warrant_status_cmd.add_argument(
+        "--shards-dir",
+        dest="shards_dir",
+        default=None,
+        help="shards directory (overrides --shards-dir from global)",
+    )
+    warrant_status_cmd.add_argument(
+        "--limit", type=int, default=3, help="max candidate matches per record"
+    )
+    warrant_status_cmd.set_defaults(func=_cmd_warrant_status)
+
+    warrant_resolve_cmd = sub.add_parser(
+        "warrant-resolve",
+        help="run citation resolver; with --write, write derivations rows binding "
+        "claim-pack records to surface chunks (Merkle proof)",
+    )
+    warrant_resolve_cmd.add_argument(
+        "--shards-dir",
+        dest="shards_dir",
+        default=None,
+        help="shards directory (overrides --shards-dir from global)",
+    )
+    warrant_resolve_cmd.add_argument(
+        "--write",
+        action="store_true",
+        help="actually write derivations rows (default: dry-run summary)",
+    )
+    warrant_resolve_cmd.add_argument(
+        "--limit", type=int, default=1, help="how many top matches per record (default 1)"
+    )
+    warrant_resolve_cmd.set_defaults(func=_cmd_warrant_resolve)
+
     # ----- mesh subcommands (off by default) ---------------------------------
     mesh_cmd = sub.add_parser(
         "mesh",

arborist/qa/warrant_resolver.py

@@ -0,0 +1,732 @@
+"""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.
+    """
+
+    title: str = ""
+    authors: tuple[str, ...] = field(default_factory=tuple)
+    year: str = ""
+    section: str = ""
+    raw: str = ""
+
+    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."""
+
+    citation: Citation
+    shard_path: str
+    document_root: str
+    document_uri: str
+    document_title: str
+    chunk_id: int
+    chunk_idx: int
+    score: float
+    snippet: str
+
+
+# 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"}
+
+
+def _build_fts_query(citation: Citation, theorem_name: str = "") -> str:
+    """Compose an FTS5 MATCH query from a citation + theorem name.
+
+    Strategy: union of significant tokens from the theorem name plus
+    author last name. FTS5's default tokenizer indexes lowercase ASCII
+    words.
+    """
+    parts: list[str] = []
+    if theorem_name:
+        toks = [
+            t for t in re.findall(r"[A-Za-z]{3,}", theorem_name)
+            if t.lower() not in _FTS_STOPWORDS
+        ]
+        parts.extend(toks)
+    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 not parts and citation.title:
+        parts = [t for t in re.findall(r"[A-Za-z]{4,}", citation.title)][:4]
+    return " OR ".join(p.lower() for p in parts) if parts else ""
+
+
+def resolve_chunks(
+    citation: Citation,
+    shards_dir: Path | str,
+    theorem_name: str = "",
+    limit: int = 5,
+) -> 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.
+    """
+    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)
+
+    fts_query = _build_fts_query(citation, theorem_name=theorem_name)
+    if not fts_query:
+        return []
+
+    matches: list[ResolutionMatch] = []
+    for shard in candidate_shards:
+        if not _shard_matches_citation(str(shard), citation):
+            continue
+        rows = _fts5_search(str(shard), fts_query, limit=limit)
+        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 "",
+                )
+            )
+    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]]:
+    """Yield ``(shard_path, record_root, title, source_reference)`` for
+    every claim_pack record across the shards-dir.
+    """
+    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, 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, content = row
+                text = _decompress_chunk(content)
+                m = _SOURCE_RE.search(text)
+                source_ref = m.group(1).strip() if m else ""
+                yield (str(db), doc_root, title or "", source_ref)
+        finally:
+            conn.close()
+
+
+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()
+
+
+# ---------------------------------------------------------------------------
+# 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) -> list[WarrantStatus]:
+    """Walk every claim-pack record and report what surface chunks the
+    citation resolver finds. Read-only — no DB writes.
+    """
+    out: list[WarrantStatus] = []
+    for shard_path, record_root, title, source_ref in iter_claim_pack_records(shards_dir):
+        citations = parse_citation(source_ref)
+        all_matches: list[ResolutionMatch] = []
+        for c in citations:
+            all_matches.extend(
+                resolve_chunks(c, shards_dir, theorem_name=title, limit=limit)
+            )
+        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
+
+
+def warrant_resolve(
+    shards_dir: Path | str,
+    write: bool = False,
+    limit: int = 1,
+) -> 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.
+    """
+    records_total = 0
+    records_resolved = 0
+    derivations_written = 0
+    statuses = warrant_status(shards_dir, limit=limit)
+    # 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)
+        if host_shard and write_derivation(
+            host_shard, status.record_root, top.document_root, proof
+        ):
+            derivations_written += 1
+    return {
+        "records_total": records_total,
+        "records_resolved": records_resolved,
+        "derivations_written": derivations_written,
+        "wrote": write,
+    }

arborist/sources/textbook_tex.py

@@ -310,11 +310,35 @@ class TextbookTexSource(Source):
 
 _TITLE_RE = re.compile(r"\\title\{([^{}]+)\}")
 _AUTHOR_RE = re.compile(r"\\author\{([^{}]+)\}")
+_PG_TITLE_RE = re.compile(r"^Title:\s*(.+?)\s*$", re.MULTILINE)
+_PG_AUTHOR_RE = re.compile(r"^Author:\s*(.+?)\s*$", re.MULTILINE)
 
 
 def _extract_title(tex: str) -> str:
-    """Pull the title from \\title{...} preamble (best-effort)."""
+    """Pull the title — prefer ``\\title{...}`` if present (LaTeX
+    convention), fall back to ``Title:`` line (Project Gutenberg
+    boilerplate format which doesn't always wire into a LaTeX
+    title macro). Append author when discoverable so downstream
+    citation-matching code (``arborist.qa.warrant_resolver``) can
+    match on author last name without re-parsing the TeX."""
+    title = ""
     m = _TITLE_RE.search(tex)
-    if not m:
+    if m:
+        title = m.group(1).strip()
+    if not title:
+        m = _PG_TITLE_RE.search(tex)
+        if m:
+            title = m.group(1).strip()
+    if not title:
         return ""
-    return m.group(1).strip()
+    author = ""
+    m = _AUTHOR_RE.search(tex)
+    if m:
+        author = m.group(1).strip()
+    if not author:
+        m = _PG_AUTHOR_RE.search(tex)
+        if m:
+            author = m.group(1).strip()
+    if author:
+        return f"{title} by {author}"
+    return title

docs/TICKETS.md

@@ -66,7 +66,7 @@ Newest first. Update on every open/close.
 | #000034  | Hessian alignment under φ_linear               | open · awaiting go/no-go (#000018 follow-up) | 2026-05-09 | —         |
 | #000033  | Claim-pack pillar VII (combinatorics)          | closed · landed 2026-05-09 (live in shard 000.db; lift verified) | 2026-05-09 | —         |
 | #000032  | combinatorics@v1 π* (integer counting kernel) | closed · landed 2026-05-09 | 2026-05-09 | —         |
-| #000031  | Surface-ingest cited textbooks for claim-pack warrant promotion | in progress · Phase 1 (8 textbooks, 6/7 pillars) landed 2026-05-09; Phase 2 chunk-resolution ahead | 2026-05-09 | —         |
+| #000031  | Surface-ingest cited textbooks for claim-pack warrant promotion | in progress · Phase 1 + Phase 2 (chunk-resolver, proof writer, 18 derivations rows for Hilbert pillar IV) landed 2026-05-09; Phase 3 verifier wiring ahead | 2026-05-09 | —         |
 | #000030  | Math π* expansion: SymPy substrate (algebra · calculus · linalg) | closed · all 7 phases + 1b landed 2026-05-09 (`abe5988`) | 2026-05-09 | —         |
 | #000029  | Claim-pack source (axiom/theorem JSON bundles) | closed · landed 2026-05-09 | 2026-05-09 | —         |
 | #000028  | Multi-modality witness for canonical shapes    | closed · landed 2026-05-09 + follow-ups (capital ledger · sample rate) | 2026-05-08 | —         |

docs/tickets/ticket-000031-surface-ingest-cited-textbooks.md

@@ -1,6 +1,6 @@
 # Ticket #000031 — Surface-ingest cited textbooks for claim-pack warrant promotion
 
-**Status:** in progress · Phase 1 (PD textbook surface ingest) landed 2026-05-09 covering 6/7 g4 pillars; Phase 2 (chunk-resolution + `derivations.proof_blob` warrant promotion) still ahead
+**Status:** in progress · Phase 1 (PD textbook surface ingest, 6/7 pillars) + Phase 2 (chunk-resolution + `derivations.proof_blob` warrant writer, Hilbert-axiom records resolved + 18 derivations rows written) both landed 2026-05-09; Phase 3 (verifier wiring to upgrade audit_mode from ANCHOR → EVIDENCE-WARRANTED on the four-rung ladder) still ahead
 **Opened:** 2026-05-09
 **Scope:** Ingest the classical textbooks cited by every record in
 the claim-pack source (`#000029`) — Mendelson 1997, Enderton 2001,
@@ -386,27 +386,71 @@ tests; full suite: 1574 passed / 28 skipped.
   yellow-light pending license decision per §2.1. Manifest entry
   kept as a license-fail placeholder.
 
-### Phase 2 — chunk-resolution + warrant promotion (NOT YET LANDED)
+### Phase 2 — chunk-resolution + warrant writer (LANDED 2026-05-09)
 
-The data substrate is in place; the warrant promotion itself
-requires:
+Implemented in `arborist/qa/warrant_resolver.py`:
 
-1. **Citation parser** — `source_reference` string ("A
-   Mathematical Introduction to Logic by Herbert B. Enderton")
-   → structured `(authors, title, year, section?)` tuple.
-2. **Chunk resolver** — given a citation, FTS5-search candidate
-   surface shards for theorem name + key terms, return top
-   chunk(s).
-3. **Merkle inclusion proof writer** — for each (claim_pack
-   record, surface chunk) match, compute the per-chunk inclusion
-   proof using `arborist/merkle.py`, serialize to JSON.
-4. **`derivations` row writer** — `(core_root,
-   src_root, proof_blob, process_id="warrant-resolver-v1",
-   distilled_at)`.
-5. **Verifier wiring** — `audit_mode` upgrade path so claims
-   with a `derivations` row tying back to a real surface get
-   **EVIDENCE-WARRANTED** instead of ANCHOR-WARRANTED on the
-   four-rung ladder.
+1. **Citation parser** (`parse_citation`) — pure regex pipeline.
+   Handles three observed source-reference patterns: ``"Title by
+   Author"`` (single + multi-author Oxford comma + ``et al.``
+   tail), semicolon-separated multi-citation, and compact
+   ``"Author Year"`` / ``"Author §section"`` form. 14 unit tests.
+2. **Chunk resolver** (`resolve_chunks`) — searches sibling
+   ``crawl/`` dir's textbook-surface shards (skipping the main
+   numbered shards which hold Wikipedia content). Per-shard
+   ``_shard_matches_citation`` filter requires both author last
+   name AND a title token to appear in the shard's title-haystack
+   — produces honest "no match" for cited textbooks not yet
+   surface-ingested, exactly the records that should stay at
+   ANCHOR-WARRANTED.
+3. **Merkle inclusion proof writer** (`compute_proof`) — reads
+   ``merkle_nodes`` for the document_root, walks layer by layer
+   to assemble siblings; emits deterministic JSON. Compatible
+   with the existing ``arborist/merkle.py`` proof verification.
+4. **derivations row writer** (`write_derivation`) — INSERT OR
+   IGNORE into the existing schema-table with
+   ``process_id="warrant-resolver-v1"``. Idempotent at the
+   database layer (PK is `(core_root, src_root, process_id)`).
+5. **CLI surface** — ``arborist warrant-status`` (read-only,
+   shows resolver matches per record as JSON) +
+   ``arborist warrant-resolve [--write]`` (default dry-run
+   summary; ``--write`` actually computes proofs + inserts rows).
+
+**Verified end-to-end** against the real shard cluster:
+``arborist warrant-resolve --shards-dir ~/.arborist/shards
+--write`` resolved **18 of 92 claim-pack records** (all the
+Hilbert-pillar-IV axiom records that cite *The Foundations of
+Geometry by David Hilbert* — the only cited textbook fully
+surface-ingested in this iteration). 18 ``derivations`` rows
+written, each carrying a valid Merkle proof blob. Re-running
+the writer is a no-op (idempotent by PK).
+
+The remaining 74 records cite textbooks not in our shard cluster
+(Mendelson, Enderton, Jech, Goldstein, Barendregt, Stanley,
+Brualdi, Knuth, …) — they correctly produce 0 matches. Adding
+those textbooks (where licensing permits) is a future-ticket
+expansion of Phase 1's manifest, not new code.
+
+### Phase 3 — verifier wiring (NOT YET LANDED)
+
+The data substrate is now in place. The four-rung-ladder upgrade
+that lifts answers citing claim-pack-records-with-derivations
+from ANCHOR-WARRANTED to **EVIDENCE-WARRANTED** still needs to
+happen in the verifier. Sketch:
+
+- The Q&A verifier currently produces `audit_mode` + `violations`
+  per answer. The four-rung ladder is computed in
+  `arborist.cli._render_audit_label._ladder_rung_for_lattice`
+  from those two.
+- For each cited claim-pack record, look up whether it has a
+  ``derivations.process_id='warrant-resolver-v1'`` row. If yes,
+  REMOVE any `WARRANT_MISSING` violation it would have triggered
+  (or add a positive `WARRANT_PROVEN` annotation).
+- Display label upgrades to EVIDENCE-WARRANTED automatically per
+  the existing ladder logic.
+
+This is a verifier change — touches well-tested code. Worth its
+own ticket so the regression risk is bounded.
 6. **Bench** — verify a known case (e.g., Modus Tollens
    claim-pack record → Aristotle Prior Analytics chapter)
    actually promotes from ANCHOR to EVIDENCE.

tests/test_warrant_resolver.py

@@ -0,0 +1,141 @@
+"""Tests for the warrant resolver (#000031 Phase 2).
+
+Pure unit tests over the citation parser; the FTS5 resolver +
+Merkle proof writer are exercised end-to-end via the
+``arborist warrant-resolve`` CLI on real shards. The CLI smoke test
+is documented in the ticket; this file stays offline / DB-free.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from arborist.qa.warrant_resolver import (
+    Citation,
+    parse_citation,
+)
+
+
+# --- "Title by Author" pattern (most common) -------------------------
+
+
+def test_simple_title_by_author():
+    out = parse_citation("Introduction to Mathematical Logic by Elliott Mendelson")
+    assert len(out) == 1
+    c = out[0]
+    assert c.title == "Introduction to Mathematical Logic"
+    assert c.authors == ("Elliott Mendelson",)
+
+
+def test_title_with_punctuation():
+    out = parse_citation("The Lambda Calculus: Its Syntax and Semantics by H.P. Barendregt")
+    assert len(out) == 1
+    c = out[0]
+    assert c.title == "The Lambda Calculus: Its Syntax and Semantics"
+    assert c.authors == ("H.P. Barendregt",)
+
+
+# --- multi-author Oxford comma --------------------------------------
+
+
+def test_multi_author_oxford_comma():
+    out = parse_citation(
+        "Classical Mechanics by Herbert Goldstein, Charles P. Poole, and John L. Safko"
+    )
+    assert len(out) == 1
+    c = out[0]
+    assert c.title == "Classical Mechanics"
+    assert c.authors == ("Herbert Goldstein", "Charles P. Poole", "John L. Safko")
+
+
+def test_multi_author_et_al():
+    out = parse_citation("Classical Mechanics by Herbert Goldstein et al.")
+    assert len(out) == 1
+    c = out[0]
+    assert c.title == "Classical Mechanics"
+    assert c.authors == ("Herbert Goldstein",)
+
+
+# --- semicolon-separated multi-citation -----------------------------
+
+
+def test_semicolon_split():
+    out = parse_citation(
+        "Knuth TAOCP Volume 1 §1.2.6; Stanley §1.2; Brualdi §3.5"
+    )
+    assert len(out) == 3
+    assert out[0].authors == ("Knuth",)
+    assert out[1].authors == ("Stanley",)
+    assert out[2].authors == ("Brualdi",)
+
+
+def test_semicolon_with_year_and_section():
+    out = parse_citation(
+        "Knuth TAOCP Volume 1 §1.2.6 equation (13); Vandermonde 1772"
+    )
+    assert len(out) == 2
+    assert out[0].section
+    assert "§1.2.6" in out[0].section
+    assert out[1].year == "1772"
+
+
+# --- compact form ---------------------------------------------------
+
+
+def test_compact_year_form():
+    out = parse_citation("Pascal 1654")
+    assert len(out) == 1
+    c = out[0]
+    assert c.year == "1654"
+    assert "Pascal" in c.authors
+
+
+def test_compact_section_only():
+    out = parse_citation("Brualdi §3.5")
+    assert len(out) == 1
+    c = out[0]
+    assert c.authors == ("Brualdi",)
+    assert "§3.5" in c.section
+
+
+# --- edge cases ------------------------------------------------------
+
+
+def test_empty_input():
+    assert parse_citation("") == []
+    assert parse_citation("   ") == []
+
+
+def test_year_extraction():
+    out = parse_citation("Foundations of Probability by Andrey Kolmogorov 1933")
+    assert len(out) == 1
+    assert out[0].year == "1933"
+
+
+def test_section_extraction():
+    out = parse_citation("Classical Mechanics by Goldstein Volume 1 §3.2")
+    assert len(out) == 1
+    assert "§3.2" in out[0].section
+    assert "Volume 1" in out[0].section
+
+
+def test_oeis_identifier():
+    out = parse_citation("OEIS A000108")
+    assert len(out) == 1
+    # OEIS identifiers parse as compact form — first token is "author".
+    assert out[0].authors == ("OEIS",)
+
+
+# --- Citation dataclass invariants ----------------------------------
+
+
+def test_citation_is_empty():
+    assert Citation().is_empty()
+    assert not Citation(title="Foo").is_empty()
+    assert not Citation(authors=("Bar",)).is_empty()
+
+
+def test_raw_field_preserved():
+    raw = "Some weird citation by Some Author"
+    out = parse_citation(raw)
+    assert out[0].raw == raw